coordination

package
v0.11.0-beta.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 23, 2026 License: Apache-2.0, MIT Imports: 36 Imported by: 0

Documentation

Overview

Package coordination provides durable, project-scoped correspondence between agents. Credentials identify senders; message payloads never choose identity.

Index

Constants

View Source 
const (
	MaxRequestDepth         = 8
	MaxConversationRequests = 32
	MaxPeerQueue            = 128
	MaxProjectQueue         = 512
	MaxExecutionAttempts    = 3
)

These protocol limits bound autonomous work, not provider token or cost use.

View Source 
const (
	MinPeerCredentialTTL     = time.Minute
	MaxPeerCredentialTTL     = 30 * 24 * time.Hour
	DefaultPeerCredentialTTL = 24 * time.Hour
)
View Source 
const MaxDependencies = 32
View Source 
const RuntimePresenceTTL = 20 * time.Second

Variables

View Source 
var (
	ErrQueueFull     = errors.New("coordination: queue admission blocked")
	ErrRequestBudget = errors.New("coordination: request budget exhausted")
)
View Source 
var (
	ErrOwnershipConflict  = errors.New("coordination: path owned by another contract")
	ErrDependencyCycle    = errors.New("coordination: contract dependency cycle")
	ErrContractValidation = errors.New("coordination: invalid handoff contract")
)
View Source 
var (
	ErrRequestContracts = errors.New("coordination: required contract agreement unavailable")
	ErrUnauthorized     = errors.New("coordination: unauthorized peer")
	ErrNotFound         = errors.New("coordination: not found")
	ErrConflict         = errors.New("coordination: revision conflict")
	ErrIdempotency      = errors.New("coordination: idempotency key reused for a different request")
	ErrBusy             = errors.New("coordination: peer has active work")
	ErrLeaseLost        = errors.New("coordination: worker lease lost")
	ErrRecoveryRequired = errors.New("coordination: interrupted work needs explicit recovery")
)
View Source 
var ErrPeerCredentialUnconfirmed = errors.New("coordination: credential commit unconfirmed; retain private credential file for inspection")

ErrPeerCredentialUnconfirmed preserves the delivered private file when commit acknowledgement is uncertain. Operators can inspect/revoke without losing it.

View Source 
var ErrUsageBudget = errors.New("coordination: task usage requires review")

Functions

func CallbackSignature 

func CallbackSignature(secret []byte, id string, sequence int64, timestamp string, body []byte) string

CallbackSignature signs version, stable event ID, sequence, attempt timestamp and exact body bytes. The secret is hex encoded in the one-time setup result.

func LocalProjectID 

func LocalProjectID(workspace string) string

LocalProjectID groups explicitly registered local peers in one canonical workspace. Callers cannot choose a foreign project through model metadata.

func NewPeerHTTPClient 

func NewPeerHTTPClient(endpoint string) (*http.Client, error)

NewPeerHTTPClient admits explicit loopback HTTP for local agents. HTTPS uses the callback transport's public-IP validation, DNS pinning and TLS policy. Advertised discovery endpoints are never dialled automatically.

func NormalizeOwnedPath 

func NormalizeOwnedPath(p string) (string, error)

NormalizeOwnedPath returns the portable project-relative reservation key. Conservative Unicode folding also reserves case-distinct Linux spellings; this prevents macOS case/normalization aliases from obtaining two owners.

func ValidateCallbackDestinations 

func ValidateCallbackDestinations(destinations map[string]string) error

ValidateCallbackDestinations validates operator configuration, never model input.

func ValidatePeerEndpoint 

func ValidatePeerEndpoint(endpoint string) error

ValidatePeerEndpoint validates advertised ingress metadata. Discovery never dials this value; an outbound caller must independently apply URL policy.

func WithTransportExecution 

func WithTransportExecution(ctx context.Context, peerID string, proof TransportExecution) (context.Context, error)

Types

type Binding 

type Binding struct {
	Peer      Peer
	SessionID string
	// contains filtered or unexported fields
}

type CallbackDelivery 

type CallbackDelivery struct {
	Sequence    int64     `json:"sequence"`
	State       string    `json:"state"`
	Attempts    int       `json:"attempts"`
	NextAttempt time.Time `json:"next_attempt"`
	StatusCode  int       `json:"status_code"`
	Failure     string    `json:"failure"`
}

type CallbackSubscription 

type CallbackSubscription struct {
	ID          string `json:"id"`
	Destination string `json:"destination"`
	Cursor      int64  `json:"cursor"`
	Active      bool   `json:"active"`
}

type CallbackWorker 

type CallbackWorker struct {
	Store        *Store
	Destinations map[string]string
	// Transport is an explicit application/test injection. Production leaves nil
	// to enforce public pinned DNS, HTTPS verification and no proxy/redirects.
	Transport      http.RoundTripper
	PollInterval   time.Duration
	AttemptTimeout time.Duration
	MaxAttempts    int
	OnError        func(error)
}

CallbackWorker only delivers durable events. It never reserves or executes agent work. HTTP acknowledgement and a DB commit cannot be atomic: receivers must deduplicate X-Overkill-Event-ID, which is stable through every retry.

func (*CallbackWorker) Run 

func (w *CallbackWorker) Run(ctx context.Context) error

type CompatibilityChange 

type CompatibilityChange struct {
	Kind        string `json:"kind"`
	FromVersion string `json:"from_version"`
	ToVersion   string `json:"to_version"`
	Description string `json:"description"`
}

type Contract 

type Contract struct {
	ID          string       `json:"id"`
	ProjectID   string       `json:"project_id"`
	OwnerPeerID string       `json:"owner_peer_id"`
	Revision    int64        `json:"revision"`
	Spec        ContractSpec `json:"spec"`
	CreatedAt   time.Time    `json:"created_at"`
	UpdatedAt   time.Time    `json:"updated_at"`
}

type ContractAcknowledgement 

type ContractAcknowledgement struct {
	ContractID     string    `json:"contract_id"`
	PeerID         string    `json:"peer_id"`
	Revision       int64     `json:"revision"`
	AcknowledgedAt time.Time `json:"acknowledged_at"`
}

type ContractArtifact 

type ContractArtifact struct {
	Name      string `json:"name"`
	Path      string `json:"path"`
	MediaType string `json:"media_type,omitempty"`
	Digest    string `json:"digest,omitempty"`
}

type ContractBlocker 

type ContractBlocker struct {
	ContractID       string `json:"contract_id"`
	Reason           string `json:"reason"`
	ExpectedRevision int64  `json:"expected_revision,omitempty"`
	ActualRevision   int64  `json:"actual_revision,omitempty"`
}

type ContractDependency 

type ContractDependency struct {
	ContractID string `json:"contract_id"`
	Revision   int64  `json:"revision"`
}

type ContractReadiness 

type ContractReadiness struct {
	ContractID string            `json:"contract_id"`
	Revision   int64             `json:"revision"`
	PeerID     string            `json:"peer_id"`
	Ready      bool              `json:"ready"`
	Blockers   []ContractBlocker `json:"blockers"`
	Source     string            `json:"source"`
}

type ContractSpec 

type ContractSpec struct {
	Name                 string                `json:"name"`
	Endpoint             string                `json:"endpoint"`
	SchemaVersion        string                `json:"schema_version"`
	Schema               json.RawMessage       `json:"schema"`
	AcceptanceCriteria   []string              `json:"acceptance_criteria"`
	Artifacts            []ContractArtifact    `json:"artifacts"`
	Workspace            string                `json:"workspace"`
	Worktree             string                `json:"worktree,omitempty"`
	OwnedPaths           []string              `json:"owned_paths"`
	Dependencies         []ContractDependency  `json:"dependencies"`
	CompatibilityChanges []CompatibilityChange `json:"compatibility_changes"`
	TaskID               string                `json:"task_id,omitempty"`
}

ContractSpec is an interface agreement, not a task completion assertion. TaskID is a reference only: internal/tasks has no transactional dependency graph or PostgreSQL execution state to infer acceptance from.

type CorrespondencePage 

type CorrespondencePage struct {
	Messages []Message `json:"messages"`
	BeforeID string    `json:"before_id,omitempty"`
	HasMore  bool      `json:"has_more"`
}

type CredentialMetadata 

type CredentialMetadata struct {
	ExpiresAt *time.Time `json:"expires_at,omitempty"`
}

type DependencyBlocker 

type DependencyBlocker struct {
	ID     string `json:"id"`
	State  State  `json:"state,omitempty"`
	Reason string `json:"reason"`
}

type DependencyGraph 

type DependencyGraph struct {
	RootID             string           `json:"root_id"`
	Nodes              []DependencyNode `json:"nodes"`
	Truncated          bool             `json:"truncated"`
	HiddenDependencies bool             `json:"hidden_dependencies"`
}

type DependencyNode 

type DependencyNode struct {
	ContractBlockers []ContractBlocker   `json:"contract_blockers"`
	Message          Message             `json:"message"`
	Readiness        string              `json:"readiness"`
	Blockers         []DependencyBlocker `json:"blockers"`
}

type DispatchOptions 

type DispatchOptions struct {
	PollInterval time.Duration
	LeaseTTL     time.Duration
	Concurrency  int
	OnError      func(peerID string, err error)
}

type Dispatcher 

type Dispatcher struct {
	Store    *Store
	Executor Executor
	Options  DispatchOptions
}

Dispatcher owns background executions until Run returns. Delivery remains queued while a session is busy, and interrupted work requires review.

func (*Dispatcher) Run 

func (d *Dispatcher) Run(ctx context.Context) error

type Escalation 

type Escalation struct {
	Sequence        int64     `json:"sequence"`
	ProjectID       string    `json:"project_id"`
	From            string    `json:"from"`
	To              string    `json:"to"`
	ParentMessageID string    `json:"parent_message_id,omitempty"`
	ConversationID  string    `json:"conversation_id,omitempty"`
	Reason          string    `json:"reason"`
	State           State     `json:"state"`
	RequiresReview  bool      `json:"requires_review"`
	CreatedAt       time.Time `json:"created_at"`
}

type Event 

type Event struct {
	ProjectID string          `json:"project_id"`
	Sequence  int64           `json:"sequence"`
	MessageID string          `json:"message_id"`
	Revision  int64           `json:"revision"`
	Type      string          `json:"type"`
	Payload   json.RawMessage `json:"payload"`
	CreatedAt time.Time       `json:"created_at"`
}

type Execution 

type Execution interface {
	Run(context.Context, Message) (json.RawMessage, error)
	Release()
}

type Executor 

type Executor interface {
	Reserve(context.Context, Binding) (Execution, error)
}

Executor reserves the same session admission used by interactive clients. Reservation must not execute work; a busy session leaves messages accepted.

type Lease 

type Lease struct {
	PeerID    string
	OwnerID   string
	ExpiresAt time.Time
	// contains filtered or unexported fields
}

type Message 

type Message struct {
	RequiredContracts []ContractDependency `json:"required_contracts,omitempty"`
	DependsOn         []string             `json:"depends_on,omitempty"`
	ID                string               `json:"id"`
	ProjectID         string               `json:"project_id"`
	From              string               `json:"from"`
	To                string               `json:"to"`
	Kind              string               `json:"kind"`
	ConversationID    string               `json:"conversation_id"`
	TaskID            string               `json:"task_id,omitempty"`
	InReplyTo         string               `json:"in_reply_to,omitempty"`
	Payload           json.RawMessage      `json:"payload"`
	State             State                `json:"state"`
	Revision          int64                `json:"revision"`
	CreatedAt         time.Time            `json:"created_at"`
	UpdatedAt         time.Time            `json:"updated_at"`
	Deadline          time.Time            `json:"deadline"`
	ParentMessageID   string               `json:"parent_message_id,omitempty"`
	RequestDepth      int                  `json:"request_depth"`
	ExecutionAttempts int                  `json:"execution_attempts"`
}

type Peer 

type Peer struct {
	ID           string   `json:"id"`
	ProjectID    string   `json:"project_id"`
	Name         string   `json:"name"`
	Role         string   `json:"role"`
	Workspace    string   `json:"workspace"`
	Endpoint     string   `json:"endpoint"`
	Capabilities []string `json:"capabilities"`
	Enabled      bool     `json:"enabled"`
}

type ReconcileTaskUsageInput 

type ReconcileTaskUsageInput struct {
	MessageID                  string  `json:"message_id"`
	ExpectedMessageRevision    int64   `json:"expected_message_revision"`
	ExpectedUsageRevision      int64   `json:"expected_usage_revision"`
	ExpectedJournalFingerprint string  `json:"expected_journal_fingerprint"`
	ReconciliationID           string  `json:"reconciliation_id"`
	InputTokens                int64   `json:"input_tokens"`
	OutputTokens               int64   `json:"output_tokens"`
	CachedInputTokens          int64   `json:"cached_input_tokens"`
	CostUSD                    float64 `json:"cost_usd"`
	UsageKnown                 bool    `json:"usage_known"`
	PriceKnown                 bool    `json:"price_known"`
	Reason                     string  `json:"reason"`
	Evidence                   string  `json:"evidence"`
	Basis                      string  `json:"basis"`
	AcknowledgeUncertainty     bool    `json:"acknowledge_uncertainty"`
}

type ReconcileTaskUsageResult 

type ReconcileTaskUsageResult struct {
	Usage          TaskUsage                   `json:"usage"`
	Reconciliation UsageReconciliationRecord   `json:"reconciliation"`
	Snapshot       UsageReconciliationSnapshot `json:"snapshot"`
}

type RepositoryInfo 

type RepositoryInfo struct {
	Scope     string `json:"scope"`
	Root      string `json:"root"`
	CommonDir string `json:"common_dir"`
	Head      string `json:"head,omitempty"`
	Branch    string `json:"branch,omitempty"`
}

RepositoryInfo describes a local, explicitly opted-in repository binding. HEAD and Branch are observations, not immutable identity or an ownership claim.

type ReviewRequestInput 

type ReviewRequestInput struct {
	MessageID             string      `json:"message_id"`
	ExpectedRevision      int64       `json:"expected_revision"`
	ExpectedUsageRevision int64       `json:"expected_usage_revision"`
	ReviewID              string      `json:"review_id"`
	Note                  string      `json:"note"`
	Limits                UsageLimits `json:"limits"`
	Requeue               bool        `json:"requeue"`
}

ReviewRequestInput is a trusted application control, never a model tool. Requeue explicitly acknowledges that replay may repeat earlier side effects.

type ReviewRequestResult 

type ReviewRequestResult struct {
	Message Message   `json:"message"`
	Usage   TaskUsage `json:"usage"`
}

type RuntimeObservation 

type RuntimeObservation struct {
	SessionID string `json:"session_id"`
	State     string `json:"state"`
}

type RuntimePresence 

type RuntimePresence struct {
	State      string     `json:"state"`
	Source     string     `json:"source"`
	Observers  int        `json:"observers"`
	ObservedAt *time.Time `json:"observed_at,omitempty"`
	ExpiresAt  *time.Time `json:"expires_at,omitempty"`
}

RuntimePresence describes an observed application runtime, not a client connection, delivery guarantee, or exclusive execution lease.

type SendInput 

type SendInput struct {
	RequiredContracts []ContractDependency `json:"required_contracts,omitempty"`
	DependsOn         []string             `json:"depends_on,omitempty"`
	To                string               `json:"to"`
	Kind              string               `json:"kind"`
	ConversationID    string               `json:"conversation_id,omitempty"`
	TaskID            string               `json:"task_id,omitempty"`
	InReplyTo         string               `json:"in_reply_to,omitempty"`
	Payload           json.RawMessage      `json:"payload"`
	Deadline          time.Time            `json:"deadline,omitempty"`
	IdempotencyKey    string               `json:"idempotency_key"`
}

type SessionPeer 

type SessionPeer struct {
	Presence RuntimePresence `json:"presence"`
	Peer
	SessionID  string          `json:"session_id"`
	Scope      string          `json:"scope"`
	Repository *RepositoryInfo `json:"repository,omitempty"`
}

type SessionTransport 

type SessionTransport struct {
	SessionID      string
	PeerID         string
	ProjectID      string
	Workspace      string
	Endpoint       string
	CredentialFile string `json:"-"`
	Revision       int64
}

SessionTransport is trusted application configuration. CredentialFile is a private local reference and must never be returned in peer/model metadata.

type State 

type State string
const (
	Accepted  State = "accepted"
	Started   State = "started"
	Blocked   State = "blocked"
	Completed State = "completed"
	Failed    State = "failed"
	Cancelled State = "cancelled"
)

func (State) Terminal 

func (s State) Terminal() bool

type Store 

type Store struct {
	// contains filtered or unexported fields
}

func NewStore 

func NewStore(ctx context.Context, database *sql.DB) (*Store, error)

NewStore borrows the pool. Initialization is transactional and fatal on error.

func (*Store) AcknowledgeContract 

func (s *Store) AcknowledgeContract(ctx context.Context, peerID, id string, expectedRevision int64) (ContractAcknowledgement, error)

func (*Store) AcquireLease 

func (s *Store) AcquireLease(ctx context.Context, peerID, ownerID string, ttl time.Duration) (Lease, error)

AcquireLease serializes one peer's executor across processes. Expired work becomes blocked for explicit recovery; acquiring never replays a started run.

func (*Store) Authenticate 

func (s *Store) Authenticate(ctx context.Context, token string) (Peer, error)

func (*Store) AuthenticateScoped 

func (s *Store) AuthenticateScoped(ctx context.Context, token string) (Peer, context.Context, error)

AuthenticateScoped binds only the authenticated actor and credential hash to request context. Token bytes never enter downstream metadata or errors.

func (*Store) Await 

func (s *Store) Await(ctx context.Context, peerID, messageID string, afterRevision int64) (Message, error)

Await waits for a newer durable revision or a terminal result. The caller supplies a bounded context; cancellation never claims or cancels the task.

func (*Store) BeginTaskUsage 

func (s *Store) BeginTaskUsage(ctx context.Context, messageID string) error

func (*Store) BindSession 

func (s *Store) BindSession(ctx context.Context, peerID, sessionID string) error

BindSession is a trusted application operation. Both records must exist and identify the same canonical workspace. A live peer cannot be retargeted to another conversation by changing model-visible metadata.

func (*Store) BindSessionTransport 

func (s *Store) BindSessionTransport(ctx context.Context, sessionID string, remote Peer, endpoint, credentialFile string) (SessionTransport, error)

BindSessionTransport requires an already verified same-hub authenticated peer snapshot. Application caller owns session admission and rebuilds its cache.

func (*Store) CallbackDeliveries 

func (s *Store) CallbackDeliveries(ctx context.Context, peerID, id string) ([]CallbackDelivery, error)

func (*Store) CallbackSubscriptions 

func (s *Store) CallbackSubscriptions(ctx context.Context, peerID string) ([]CallbackSubscription, error)

func (*Store) CancelCallback 

func (s *Store) CancelCallback(ctx context.Context, peerID, id string) error

CancelCallback revokes delivery; it never cancels the underlying agent work.

func (*Store) CheckOwnedPaths 

func (s *Store) CheckOwnedPaths(ctx context.Context, peerID string, paths []string) error

func (*Store) CheckTaskUsage 

func (s *Store) CheckTaskUsage(ctx context.Context, messageID string) error

CheckTaskUsage revalidates the live execution and records review evidence once per exhausted/unknown budget. Limits gate subsequent calls, not provider bills or requests already in flight on another peer.

func (*Store) Claim 

func (s *Store) Claim(ctx context.Context, peerID string) (Message, error)

Claim atomically starts one pending message. Started messages are not automatically reclaimed after a crash: replaying a partially executed tool task could duplicate its side effects. Recovery inspects its durable state.

func (*Store) ClaimLease 

func (s *Store) ClaimLease(ctx context.Context, lease Lease) (Message, error)

func (*Store) ClearSessionTransport 

func (s *Store) ClearSessionTransport(ctx context.Context, sessionID string) error

func (*Store) CompleteLease 

func (s *Store) CompleteLease(ctx context.Context, lease Lease, messageID string, payload json.RawMessage) (Message, error)

CompleteLease atomically finishes a claimed run and, for a request, accepts its correlated answer. Failure cannot publish an answer without completion. Notification messages finish without generating an endless reply cycle.

func (*Store) ConfigureSessionPeer 

func (s *Store) ConfigureSessionPeer(ctx context.Context, anchorSession, targetSession, endpoint string, capabilities []string, deliver func(string) error) (Peer, error)

ConfigureSessionPeer is a trusted operator control, never a model tool. Both identities must already have validated persisted session bindings in the same project. Delivery of the new credential must succeed before its hash commits. The callback must keep the credential private and remove its file on error.

func (*Store) ConfigureSessionPeerWithTTL 

func (s *Store) ConfigureSessionPeerWithTTL(ctx context.Context, anchorSession, targetSession, endpoint string, capabilities []string, ttl time.Duration, deliver func(string) error) (Peer, CredentialMetadata, error)

ConfigureSessionPeerWithTTL is trusted provisioning. TTL uses database time; legacy credentials remain NULL-expiry, never silently shortened or deleted.

func (*Store) ContractReadiness 

func (s *Store) ContractReadiness(ctx context.Context, peerID, id string) (ContractReadiness, error)

ContractReadiness reports protocol agreement, never task execution or shipped acceptance. The caller must acknowledge the queried revision; each dependency is pinned and acknowledged by its dependent contract owner, recursively.

func (*Store) ContractRevision 

func (s *Store) ContractRevision(ctx context.Context, peerID, id string, revision int64) (ContractSpec, error)

ContractRevision reads an immutable prior agreement under the same project authorization as the current contract, for explicit compatibility review.

func (*Store) Correspondence 

func (s *Store) Correspondence(ctx context.Context, peerID, beforeID string, limit int) (CorrespondencePage, error)

Correspondence pages all durable states, newest first. The immutable creation-time/UUID cursor prevents status updates from moving items between pages. A cursor outside this peer's correspondence is rejected.

func (*Store) CreateContract 

func (s *Store) CreateContract(ctx context.Context, peerID string, spec ContractSpec) (Contract, error)

func (*Store) DeleteContract 

func (s *Store) DeleteContract(ctx context.Context, peerID, id string, expectedRevision int64) error

func (*Store) DisablePeer 

func (s *Store) DisablePeer(ctx context.Context, peerID string) error

DisablePeer revokes every credential and prevents further reads/changes. This is an application/admin operation, not a model-visible operation.

func (*Store) DispatchBindings 

func (s *Store) DispatchBindings(ctx context.Context, afterID string, limit int) ([]Binding, error)

DispatchBindings pages only peers with outstanding work. The keyset stays stable while another worker finishes messages or creates new peer bindings.

func (*Store) Events 

func (s *Store) Events(ctx context.Context, peerID string, after int64, limit int) ([]Event, error)

Events replays only this peer's correspondence after a project cursor. A cursor remains valid over restart and does not acknowledge execution.

func (*Store) ExternalClaim 

func (s *Store) ExternalClaim(ctx context.Context, peerID string) (Message, error)

ExternalClaim/ExternalTransition are manual ingress controls, not remote worker leases. Any existing worker ownership requires its explicit release or recovery before manual processing. Callers hold admission then peer lock, which also fences AcquireLease; no check-then-mutate race is introduced.

func (*Store) ExternalReply 

func (s *Store) ExternalReply(ctx context.Context, peerID, messageID string, payload json.RawMessage, key string) (Message, error)

func (*Store) ExternalSend 

func (s *Store) ExternalSend(ctx context.Context, peerID string, in SendInput) (Message, error)

func (*Store) ExternalTransition 

func (s *Store) ExternalTransition(ctx context.Context, peerID, messageID string, revision int64, event string, payload json.RawMessage) (Message, error)

func (*Store) FinishTaskUsage 

func (s *Store) FinishTaskUsage(ctx context.Context, messageID string) error

func (*Store) Graph 

func (s *Store) Graph(ctx context.Context, viewerPeerID, messageID string, limit int) (DependencyGraph, error)

Graph returns the root and its prerequisite ancestry from one database snapshot. Payload-free envelopes are participant-only. For a prerequisite named by a visible envelope, its state is intentionally available as blocker metadata; inaccessible envelopes and their ancestry are never traversed or returned. Readiness is dependency readiness, not execution admission or provider health.

func (*Store) Inbox 

func (s *Store) Inbox(ctx context.Context, peerID string, limit int) ([]Message, error)

Inbox includes started/blocked work so a restarted worker can present an interrupted task for recovery instead of silently executing it again.

func (*Store) ListContracts 

func (s *Store) ListContracts(ctx context.Context, peerID string) ([]Contract, error)

func (*Store) LocalPeersForSession 

func (s *Store) LocalPeersForSession(ctx context.Context, sessionID string) ([]SessionPeer, error)

LocalPeersForSession lists explicitly configured peers in the bound scope. Only opted-in repository sessions can inventory across linked worktrees; unregistered and default sessions retain the canonical workspace boundary.

func (*Store) MessageEvents 

func (s *Store) MessageEvents(ctx context.Context, peerID, messageID string, afterRevision int64, limit int) ([]Event, error)

func (*Store) PeerForSession 

func (s *Store) PeerForSession(ctx context.Context, sessionID string) (Peer, error)

func (*Store) Peers 

func (s *Store) Peers(ctx context.Context, peerID string) ([]Peer, error)

Peers exposes only enabled peers from the caller's project. Credentials and their hashes never enter this result or model-visible discovery metadata.

func (*Store) PendingCount 

func (s *Store) PendingCount(ctx context.Context, peerID string) (int, error)

PendingCount can drive application wakeups without claiming or replaying work.

func (*Store) PublishRuntimePresence 

func (s *Store) PublishRuntimePresence(ctx context.Context, observer string, observations []RuntimeObservation) error

PublishRuntimePresence is application-only. Models and RPC clients cannot supply observations. The caller serializes snapshots for its unique observer. Snapshot replacement and expiry cleanup commit together using database time.

func (*Store) Read 

func (s *Store) Read(ctx context.Context, peerID, messageID string) (Message, error)

Read deliberately returns not-found for messages outside the authenticated peer's correspondence so arbitrary IDs reveal no other project's activity.

func (*Store) ReadContract 

func (s *Store) ReadContract(ctx context.Context, peerID, id string) (Contract, error)

func (*Store) ReadEscalations 

func (s *Store) ReadEscalations(ctx context.Context, peerID string, after int64, limit int) ([]Escalation, error)

func (*Store) ReconcileExecutionContracts 

func (s *Store) ReconcileExecutionContracts(ctx context.Context, lease Lease, messageID string) error

ReconcileExecutionContracts records agreement invalidation even when a run ended with a tool error, cancellation, or another failure rather than output. Existing blocked/terminal decisions are preserved; repeated finalization does not duplicate their events or review escalations.

func (*Store) ReconcileTaskUsage 

func (s *Store) ReconcileTaskUsage(ctx context.Context, peerID string, in ReconcileTaskUsageInput) (ReconcileTaskUsageResult, error)

ReconcileTaskUsage is a trusted application control. Its values are manual attestations, never provider reports or independently verified invoices.

func (*Store) RecordTaskUsage 

func (s *Store) RecordTaskUsage(ctx context.Context, messageID, key string, sample UsageSample) error

func (*Store) RecoverExternal 

func (s *Store) RecoverExternal(ctx context.Context, peerID string) error

RecoverExternal uses existing fenced recovery: live owners refuse admission, expired started work becomes blocked exactly once and is never replayed. An operator must stop local delivery before switching to manual processing.

func (*Store) RegisterPeer 

func (s *Store) RegisterPeer(ctx context.Context, peer Peer) (Peer, string, error)

func (*Store) RegisterRepositorySessionPeer 

func (s *Store) RegisterRepositorySessionPeer(ctx context.Context, sessionID, name, role string) (Peer, error)

RegisterRepositorySessionPeer explicitly joins peers in linked worktrees of the same canonical Git common directory. Existing bindings never change scope.

func (*Store) RegisterSessionPeer 

func (s *Store) RegisterSessionPeer(ctx context.Context, sessionID, name, role string) (Peer, error)

RegisterSessionPeer is an authenticated application operation. Peer creation and immutable session binding commit together, including concurrent retries. Local tools need no bearer credential; no unused external credential is made.

func (*Store) ReleaseLease 

func (s *Store) ReleaseLease(ctx context.Context, lease Lease) error

func (*Store) RemoveRuntimeObserver 

func (s *Store) RemoveRuntimeObserver(ctx context.Context, observer string) error

func (*Store) RenewLease 

func (s *Store) RenewLease(ctx context.Context, lease Lease, ttl time.Duration) (Lease, error)

func (*Store) Reply 

func (s *Store) Reply(ctx context.Context, peerID, messageID string, payload json.RawMessage, key string) (Message, error)

func (*Store) RepositoryInfoForSession 

func (s *Store) RepositoryInfoForSession(ctx context.Context, sessionID string) (RepositoryInfo, error)

RepositoryInfoForSession returns validated trusted local metadata only for an explicitly repository-scoped binding. Unregistered/workspace chats are absent.

func (*Store) RetryCallback 

func (s *Store) RetryCallback(ctx context.Context, peerID, id string, sequence int64) error

RetryCallback explicitly resumes a dead-letter with the same stable identity.

func (*Store) ReviewRequest 

func (s *Store) ReviewRequest(ctx context.Context, peerID string, in ReviewRequestInput) (ReviewRequestResult, error)

ReviewRequest changes limits and requeues one blocked request atomically. It never starts execution, resets usage, erases uncertainty, or renews deadlines. The next worker claim still applies dependency, agreement and attempt gates.

func (*Store) RevokePeerCredential 

func (s *Store) RevokePeerCredential(ctx context.Context, anchorSession, targetSession, token string) error

RevokePeerCredential is trusted operator administration, not peer ingress or a model tool. Retiring one key never disables the peer or its replacement.

func (*Store) RotateCallbackKey 

func (s *Store) RotateCallbackKey(ctx context.Context, peerID, id string) (string, error)

RotateCallbackKey recovers a lost setup response without replaying agent work. Generation fencing prevents an old in-flight worker from acknowledging after rotation; retries preserve the exact delivery ID and body. Receivers should install the new secret then explicitly retry any authentication dead-letter.

func (*Store) RuntimePresenceForPeers 

func (s *Store) RuntimePresenceForPeers(ctx context.Context, ids []string) (map[string]RuntimePresence, error)

RuntimePresenceForPeers is a bounded internal read. Callers derive IDs from their authenticated workspace inventory, never an arbitrary client argument.

func (*Store) Send 

func (s *Store) Send(ctx context.Context, fromPeer string, input SendInput) (Message, error)

Send persists acceptance before returning a handle. fromPeer comes from the authenticated application boundary, not a user-supplied payload field.

func (*Store) SendWithBudget 

func (s *Store) SendWithBudget(ctx context.Context, peer string, input SendInput, limits UsageLimits) (Message, error)

SendWithBudget is an application control, deliberately absent from model SendInput. Limits are committed in the same transaction as the first request.

func (*Store) SessionTransport 

func (s *Store) SessionTransport(ctx context.Context, sessionID string) (SessionTransport, error)

func (*Store) SubscribeCallback 

func (s *Store) SubscribeCallback(ctx context.Context, peerID, destination string, after int64) (CallbackSubscription, string, error)

func (*Store) Transition 

func (s *Store) Transition(ctx context.Context, peerID, messageID string, expectedRevision int64, eventType string, payload json.RawMessage) (Message, error)

Transition records status or progress together with its replayable event. Only the recipient reports work; either correspondent may cancel. Revisions prevent a delayed progress report from undoing completion or cancellation.

func (*Store) UpdateContract 

func (s *Store) UpdateContract(ctx context.Context, peerID, id string, expectedRevision int64, spec ContractSpec) (Contract, error)

func (*Store) UsageForMessage 

func (s *Store) UsageForMessage(ctx context.Context, peer, messageID string) (TaskUsage, error)

UsageForMessage is participant-filtered, even when a task spans conversations.

func (*Store) UsageReconciliationCandidates 

func (s *Store) UsageReconciliationCandidates(ctx context.Context, peerID, messageID, afterID string, limit int) (UsageReconciliationCandidatePage, error)

UsageReconciliationCandidates discovers metadata only. A known root message selects the scope; callers cannot supply a project or arbitrary task owner.

func (*Store) UsageReconciliationForMessage 

func (s *Store) UsageReconciliationForMessage(ctx context.Context, peerID, messageID string) (UsageReconciliationSnapshot, error)

func (*Store) WithOwnedPaths 

func (s *Store) WithOwnedPaths(ctx context.Context, peerID string, paths []string, fn func(context.Context) error) error

WithOwnedPaths holds the same project advisory lock as contract mutation until fn returns. Callback side effects are not rolled back by PostgreSQL; the lock prevents claim changes during this synchronous operation. Callbacks must not recursively call contract APIs on another database connection.

func (*Store) WithUnrestrictedWorkspace 

func (s *Store) WithUnrestrictedWorkspace(ctx context.Context, peerID string, fn func(context.Context) error) error

WithUnrestrictedWorkspace permits opaque effectful tools when every foreign project reservation is in a physically disjoint registered workspace. Same or nested roots remain blocked, including disabled owners' reservations. This application-level guard holds the project lock through fn; it does not sandbox arbitrary commands or prevent filesystem changes outside that lock.

func (*Store) WorkerTransition 

func (s *Store) WorkerTransition(ctx context.Context, lease Lease, messageID string, expectedRevision int64, eventType string, payload json.RawMessage) (Message, error)

WorkerTransition fences an expired executor even if it later produces output.

type TaskUsage 

type TaskUsage struct {
	OwnerPeerID           string      `json:"owner_peer_id,omitempty"`
	PendingExecutions     int64       `json:"pending_executions"`
	UnconfirmedExecutions int64       `json:"unconfirmed_executions"`
	Scope                 string      `json:"scope"`
	TaskID                string      `json:"task_id,omitempty"`
	Limits                UsageLimits `json:"limits"`
	InputTokens           int64       `json:"input_tokens"`
	OutputTokens          int64       `json:"output_tokens"`
	CachedInputTokens     int64       `json:"cached_input_tokens"`
	EstimatedCostUSD      float64     `json:"estimated_cost_usd"`
	UnknownUsage          int64       `json:"unknown_usage"`
	UnknownPrice          int64       `json:"unknown_price"`
	Revision              int64       `json:"revision"`
	RequiresReview        bool        `json:"requires_review"`
	Reason                string      `json:"reason,omitempty"`
}

type TransportExecution 

type TransportExecution struct {
	MessageID  string `json:"message_id"`
	OwnerID    string `json:"owner_id"`
	Capability string `json:"capability"`
}

TransportExecution is a private runtime envelope, never a model tool field or response. The random capability is stored only as a hash beside the live lease.

func ExecutionTransportProof 

func ExecutionTransportProof(ctx context.Context) *TransportExecution

type UsageJournalRun 

type UsageJournalRun struct {
	OwnerID  string `json:"owner_id"`
	Finished bool   `json:"finished"`
	Resolved bool   `json:"resolved"`
}

type UsageLimits 

type UsageLimits struct {
	Tokens  int64   `json:"tokens"`
	CostUSD float64 `json:"cost_usd"`
}

type UsageReconciliationCandidate 

type UsageReconciliationCandidate struct {
	UnknownUsage          int64  `json:"unknown_usage"`
	UnknownPrice          int64  `json:"unknown_price"`
	MessageID             string `json:"message_id"`
	MessageRevision       int64  `json:"message_revision"`
	State                 State  `json:"state"`
	PendingExecutions     int64  `json:"pending_executions"`
	UnconfirmedExecutions int64  `json:"unconfirmed_executions"`
	CanReconcile          bool   `json:"can_reconcile"`
	Reason                string `json:"reason"`
}

type UsageReconciliationCandidatePage 

type UsageReconciliationCandidatePage struct {
	Scope      string                         `json:"scope"`
	Candidates []UsageReconciliationCandidate `json:"candidates"`
	NextCursor string                         `json:"next_cursor"`
	Truncated  bool                           `json:"truncated"`
}

type UsageReconciliationRecord 

type UsageReconciliationRecord struct {
	ReconciliationID string                    `json:"reconciliation_id"`
	MessageID        string                    `json:"message_id"`
	ActorPeerID      string                    `json:"actor_peer_id"`
	Basis            string                    `json:"basis"`
	Reason           string                    `json:"reason"`
	Evidence         string                    `json:"evidence"`
	Provenance       string                    `json:"provenance"`
	Before           UsageReconciliationTotals `json:"before"`
	After            UsageReconciliationTotals `json:"after"`
	CreatedAt        time.Time                 `json:"created_at"`
}

type UsageReconciliationSnapshot 

type UsageReconciliationSnapshot struct {
	MessageID          string                      `json:"message_id"`
	MessageRevision    int64                       `json:"message_revision"`
	UsageRevision      int64                       `json:"usage_revision"`
	Scope              string                      `json:"scope"`
	OwnerPeerID        string                      `json:"owner_peer_id"`
	JournalFingerprint string                      `json:"journal_fingerprint"`
	CanReconcile       bool                        `json:"can_reconcile"`
	Reason             string                      `json:"reason"`
	Runs               []UsageJournalRun           `json:"runs"`
	RunCount           int64                       `json:"run_count"`
	RunsTruncated      bool                        `json:"runs_truncated"`
	Samples            []UsageSampleEvidence       `json:"samples"`
	SampleCount        int64                       `json:"sample_count"`
	SamplesTruncated   bool                        `json:"samples_truncated"`
	Observed           UsageReconciliationTotals   `json:"observed"`
	Contribution       UsageReconciliationTotals   `json:"contribution"`
	KnownCostFloorUSD  float64                     `json:"known_cost_floor_usd"`
	Previous           []UsageReconciliationRecord `json:"previous"`
	PreviousCount      int64                       `json:"previous_count"`
	PreviousTruncated  bool                        `json:"previous_truncated"`
	// contains filtered or unexported fields
}

type UsageReconciliationTotals 

type UsageReconciliationTotals struct {
	InputTokens       int64   `json:"input_tokens"`
	OutputTokens      int64   `json:"output_tokens"`
	CachedInputTokens int64   `json:"cached_input_tokens"`
	EstimatedCostUSD  float64 `json:"estimated_cost_usd"`
	UnknownUsage      int64   `json:"unknown_usage"`
	UnknownPrice      int64   `json:"unknown_price"`
}

type UsageSample 

type UsageSample struct {
	Usage            providers.Usage
	Reported         bool
	EstimatedCostUSD float64
	PriceKnown       bool
}

type UsageSampleEvidence 

type UsageSampleEvidence struct {
	Key               string  `json:"key"`
	InputTokens       int64   `json:"input_tokens"`
	OutputTokens      int64   `json:"output_tokens"`
	CachedInputTokens int64   `json:"cached_input_tokens"`
	EstimatedCostUSD  float64 `json:"estimated_cost_usd"`
	UsageKnown        bool    `json:"usage_known"`
	PriceKnown        bool    `json:"price_known"`
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.