a2a

type Address struct {
	Agent string `json:"agent"`
	Node  string `json:"node,omitempty"`
}

type AskParams struct {
	SendParams
	AskerRunID id.AgentRunID
	ToolCallID string

	// Hold writes the message and the pending ask but routes nothing.
	// The question is not queued, and no ask it answers is resumed, until
	// Release is called with the result.
	//
	// A caller that suspends its run after Ask returns needs this. Routed
	// at once, the question can be carried, answered and replied to
	// before the asker has paused, and a reply that finds no paused run
	// has nothing to resume: the asker then waits on an answer that has
	// already come and gone.
	Hold bool
	// DurableHold requires a native versioned release after the run is paused.
	// It supports engine-origin questions, not replies to another pending ask.
	DurableHold bool
}

type AskReply struct {
	Replies []AskReplyItem `json:"replies"`
	// Complete says every recipient answered. False means the deadline
	// arrived first and Replies holds whoever did, which is an ordinary
	// outcome for a tender rather than a failure.
	Complete bool `json:"complete"`
}

type AskReplyItem struct {
	Performative   string `json:"performative"`
	Sender         string `json:"sender"`
	Content        string `json:"content"`
	ConversationID string `json:"conversation_id"`
}

type AskResolution struct {
	SchemaVersion int        `json:"schema_version"`
	ResolutionID  string     `json:"resolution_id"`
	Ask           PendingAsk `json:"ask"`
	Payload       string     `json:"payload"`
	BindingDigest string     `json:"binding_digest"`
	OutcomeDigest string     `json:"outcome_digest"`
	SealedAt      time.Time  `json:"sealed_at"`
	Fresh         bool       `json:"-"`
}

type AskResolutionPage struct {
	Resolutions []AskResolution
	Next        string
}

type AskResolutionStore interface {
	ReadPendingAsk(context.Context, string) (*PendingAsk, error)
	SealAskResolution(context.Context, *PendingAsk, AskReply) (AskResolution, error)
	GetAskResolution(context.Context, string) (AskResolution, error)
	ListAskResolutions(context.Context, string, int) (AskResolutionPage, error)
}

type AskResult struct {
	ToolCallID     string            `json:"tool_call_id,omitempty"`
	MessageID      id.MessageID      `json:"message_id"`
	ConversationID id.ConversationID `json:"conversation_id"`
	ReplyWith      string            `json:"reply_with"`
	// contains filtered or unexported fields
}

type AtomicMessageStore interface {
	RecordMessage(context.Context, *Envelope) (*Envelope, error)
}

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

type BusConfig struct {
	// RecipientRecovery enables bounded automatic observation of deferred replies.
	RecipientRecovery *RecipientRecoveryConfig
	// Roster restricts a non-global bus to captured local participant identities.
	Roster *Roster
	// ReservedRunner replaces Runner for partitioned execution with durable admission.
	ReservedRunner cortex.ReservedAgentRunner
	// RecipientObserver enables guarded completion of previously paused recipients.
	RecipientObserver RecipientObserver
	// OrchestrationID fixes this bus to one routing partition. Zero is global.
	OrchestrationID id.OrchestrationID
	// HandoffHandler is required for structured assignments. No ordinary agent
	// fallback is permitted. The host owns current authority and durable intake.
	HandoffHandler HandoffHandler
	Store          Store
	Runner         cortex.AgentRunner
	Resumer        Resumer
	Resolver       Resolver
	Hooks          HookEmitter
	Clock          Clock
	Transports     []Transport
	Options        Options

	// Metrics records the bus's delivery counter. Nil, the default,
	// records nothing; the engine passes whatever WithMetrics gave it.
	Metrics gometrics.Metrics

	// Synchronous makes the dispatcher deliver only when Drain is called.
	// Tests set it so an assertion can never observe a run mid-flight.
	Synchronous bool
}

type Class string

type Clock interface {
	Now() time.Time
}

type ContractNetParams struct {
	Initiator  Address
	Recipients []Address
	Content    string
	Ontology   string
	ReplyBy    *time.Time
	AskerRunID id.AgentRunID
	ToolCallID string
}

type Conversation struct {
	OrchestrationID string `json:"orchestration_id,omitempty"`
	cortex.Entity
	ID           id.ConversationID `json:"id"`
	Scope        cortex.Scope      `json:"scope"`
	Protocol     string            `json:"protocol,omitempty"`
	Initiator    Address           `json:"initiator"`
	Participants []Address         `json:"participants,omitempty"`
	Status       string            `json:"status"`
	HopCeiling   int               `json:"hop_ceiling"`
	HopsUsed     int               `json:"hops_used"`
	Deadline     *time.Time        `json:"deadline,omitempty"`

	// PeerNode and PeerContext record the remote conversation this one
	// stands in for.
	//
	// A contextId from a peer names a conversation in the peer's own
	// database and means nothing here, so an inbound thread gets a
	// conversation of ours and the pairing is written down. Without it
	// every inbound message would open a new thread, and a new thread is
	// a new hop budget: a peer could talk forever simply by never
	// reusing an id.
	PeerNode    string `json:"peer_node,omitempty"`
	PeerContext string `json:"peer_context,omitempty"`
}

type ConversationListFilter struct {
	Exact  bool
	Status string
	Limit  int
	Offset int
}

type Delivery struct {
	// ClaimToken identifies the current claim, independently of clock precision.
	ClaimToken      string `json:"claim_token,omitempty"`
	OrchestrationID string `json:"orchestration_id,omitempty"`
	cortex.Entity
	ID        id.DeliveryID `json:"id"`
	Scope     cortex.Scope  `json:"scope"`
	MessageID id.MessageID  `json:"message_id"`
	Receiver  Address       `json:"receiver"`
	State     string        `json:"state"`
	Error     string        `json:"error,omitempty"`
	// ClaimedAt is when a worker took the row. It is what tells a
	// reclaim the difference between a delivery in flight and one a dead
	// process was carrying.
	ClaimedAt   *time.Time    `json:"claimed_at,omitempty"`
	DeliveredAt *time.Time    `json:"delivered_at,omitempty"`
	ReadAt      *time.Time    `json:"read_at,omitempty"`
	RunID       id.AgentRunID `json:"run_id,omitempty"` // the run a directive started
}

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

type DeliveryOutcome struct {
	Receiver Address `json:"receiver"`
	Status   string  `json:"status"`
	Error    string  `json:"error,omitempty"`
	// DeliveryID is the row this delivery became.
	//
	// It exists because a caller often needs a handle to the work before
	// the work exists: at send time nothing has been delivered, so there
	// is no run to name yet, and the delivery is the only durable thing
	// to point at. The remote transport hands it to a peer as a task id.
	DeliveryID id.DeliveryID `json:"delivery_id,omitempty"`
}

type DeliveryRunBinder interface {
	BindDeliveryRun(context.Context, *Delivery, id.AgentRunID) error
}

type Envelope struct {
	OrchestrationID string `json:"orchestration_id,omitempty"`
	cortex.Entity
	ID    id.MessageID `json:"id"`
	Scope cortex.Scope `json:"scope"`

	Performative   Performative      `json:"performative"`
	Sender         Address           `json:"sender"`
	Receivers      []Address         `json:"receivers"`
	ReplyTo        []Address         `json:"reply_to,omitempty"`
	Content        string            `json:"content"`
	Language       string            `json:"language,omitempty"`
	Encoding       string            `json:"encoding,omitempty"`
	Ontology       string            `json:"ontology,omitempty"`
	Protocol       string            `json:"protocol,omitempty"`
	ConversationID id.ConversationID `json:"conversation_id"`
	ReplyWith      string            `json:"reply_with,omitempty"`
	InReplyTo      string            `json:"in_reply_to,omitempty"`
	ReplyBy        *time.Time        `json:"reply_by,omitempty"`

	Hops        int            `json:"hops"`
	OriginRunID id.AgentRunID  `json:"origin_run_id,omitempty"`
	Metadata    map[string]any `json:"metadata,omitempty"`
}

type HandoffHandler interface {
	ValidateHandoff(context.Context, *Envelope) error
	ReceiveHandoff(context.Context, *Envelope) error
}

type HeldAskReceipt struct {
	SchemaVersion   int             `json:"schema_version"`
	Scope           cortex.Scope    `json:"scope"`
	OrchestrationID string          `json:"orchestration_id"`
	Request         HeldAskRequest  `json:"request"`
	MessageID       id.MessageID    `json:"message_id"`
	Deliveries      []id.DeliveryID `json:"deliveries"`
	CommittedAt     time.Time       `json:"committed_at"`
	Fresh           bool            `json:"-"`
}

type HeldAskRequest struct {
	RunID        id.AgentRunID   `json:"run_id"`
	SuspensionID id.SuspensionID `json:"suspension_id"`
	RunRevision  int64           `json:"run_revision"`
	ReplyWith    string          `json:"reply_with"`
}

type HeldAskStore interface {
	ReleaseHeldAsk(context.Context, HeldAskRequest) (HeldAskReceipt, error)
	GetHeldAskRelease(context.Context, string) (HeldAskReceipt, error)
}

type HookEmitter interface {
	MessageSent(ctx context.Context, msgID id.MessageID, from, to, performative string)
	MessageDelivered(ctx context.Context, msgID id.MessageID, to string)
	MessageRefused(ctx context.Context, msgID id.MessageID, to, reason string)
}

type InboxFilter struct {
	UnreadOnly     bool
	ConversationID id.ConversationID
	Limit          int
	Offset         int
}

type InboxItem struct {
	DeliveryID     string `json:"delivery_id"`
	MessageID      string `json:"message_id"`
	ConversationID string `json:"conversation_id"`
	Sender         string `json:"sender"`
	Performative   string `json:"performative"`
	Content        string `json:"content"`
	ReceivedAt     string `json:"received_at,omitempty"`
}

type MessageListFilter struct {
	Exact          bool
	ConversationID id.ConversationID
	Limit          int
	Offset         int
}

type Options struct {
	// HopCeiling caps how many messages one conversation may derive.
	HopCeiling int
	// Workers is the dispatcher's concurrency. It has to cover resumed
	// askers as well as recipients, because a resume runs synchronously on
	// the worker that delivered the reply.
	Workers int
	// DefaultReplyBy is the deadline stamped on an ask that names none.
	DefaultReplyBy time.Duration
	// SweepInterval is how often overdue asks are resolved into failures,
	// and how often abandoned deliveries are reclaimed.
	SweepInterval time.Duration
	// DeliveryClaimTTL is how long a claimed delivery may sit before it
	// is treated as abandoned by a process that died mid-delivery.
	DeliveryClaimTTL time.Duration
}

type PendingAsk struct {
	OrchestrationID string `json:"orchestration_id,omitempty"`
	cortex.Entity
	Scope          cortex.Scope      `json:"scope"`
	ReplyWith      string            `json:"reply_with"`
	ConversationID id.ConversationID `json:"conversation_id"`
	MessageID      id.MessageID      `json:"message_id"`
	AskerRunID     id.AgentRunID     `json:"asker_run_id"`
	AskerAgent     string            `json:"asker_agent"`
	ToolCallID     string            `json:"tool_call_id"`
	Expected       Address           `json:"expected"`
	Deadline       *time.Time        `json:"deadline,omitempty"`
	ClaimedAt      *time.Time        `json:"claimed_at,omitempty"`
}

type Performative string

type Proposal struct {
	From    Address `json:"from"`
	Content string  `json:"content"`
}

type RecipientDiscoveryStore interface {
	ListDeferredRecipients(context.Context, id.DeliveryID, int) ([]id.DeliveryID, error)
}

type RecipientObservation struct {
	RunID     id.AgentRunID
	AgentName string
	State     string
	Output    string
}

type RecipientObserver interface {
	ObserveRecipient(context.Context, RecipientRequest) (RecipientObservation, error)
}

type RecipientRecoveryConfig struct {
	AuthorizeDiscovery func(context.Context, id.OrchestrationID) error
	AuthorizeDelivery  func(context.Context, id.OrchestrationID, id.DeliveryID) error
	// BatchSize defaults to 32 and cannot exceed 256.
	BatchSize int
	// AttemptTimeout bounds each discovery or recovery call; default 5s, max 1m.
	AttemptTimeout time.Duration
	// ReportError receives a failed pass on the worker. It must not block shutdown.
	ReportError func(context.Context, error)
}

type RecipientRecoveryPage struct {
	Next      id.DeliveryID
	Examined  int
	Completed int
	Waiting   int
	Failed    int
}

type RecipientRequest struct {
	OrchestrationID id.OrchestrationID
	RunID           id.AgentRunID
	AgentName       string
}

type ReservedReply struct {
	SchemaVersion   int           `json:"schema_version"`
	Scope           cortex.Scope  `json:"scope"`
	OrchestrationID string        `json:"orchestration_id"`
	DeliveryID      id.DeliveryID `json:"delivery_id"`
	RunID           id.AgentRunID `json:"run_id"`
	ReplyMessageID  id.MessageID  `json:"reply_message_id"`
	ReplyDeliveryID id.DeliveryID `json:"reply_delivery_id"`
	RequestDigest   string        `json:"request_digest"`
	CommittedAt     time.Time     `json:"committed_at"`
	Fresh           bool          `json:"-"`
}

type ReservedReplyStore interface {
	AtomicMessageStore
	CommitReservedReply(context.Context, *Delivery, string, bool) (ReservedReply, error)
	GetReservedReply(context.Context, id.DeliveryID) (ReservedReply, error)
}

type ResolutionResumer interface {
	ResumeAgentResolution(context.Context, AskResolution) error
}

type Resolver interface {
	ResolveAddress(ctx context.Context, addr Address) error
}

type Resumer interface {
	ResumeAgentReply(ctx context.Context, runID id.AgentRunID, callID, result string) error
}

type Roster struct {
	Scope        cortex.Scope
	Participants []RosterParticipant
}

type RosterParticipant struct {
	Name    string
	AgentID id.AgentID
}

type RoutingBackend interface {
	Store
	SupportsOrchestrationRouting() bool
}

type ScopedRoutingBackend interface {
	RoutingBackend
	SupportsScopedOrchestrationDiscovery() bool
}

type SendParams struct {
	Sender         Address
	Receivers      []Address
	Performative   Performative
	Content        string
	ConversationID id.ConversationID
	ReplyTo        []Address
	Language       string
	Encoding       string
	Ontology       string
	Protocol       string
	ReplyWith      string
	InReplyTo      string
	ReplyBy        *time.Time
	OriginRunID    id.AgentRunID
	Metadata       map[string]any

	// PeerNode and PeerContext name a remote thread this message
	// belongs to, for a message arriving from another engine. When they
	// are set and no conversation id is, the send joins the conversation
	// opened for that remote thread, or opens one and records the
	// pairing.
	PeerNode    string
	PeerContext string
}

type SendResult struct {
	MessageID      id.MessageID      `json:"message_id"`
	ConversationID id.ConversationID `json:"conversation_id"`
	Deliveries     []DeliveryOutcome `json:"deliveries"`
}

type Store interface {
	CreateMessage(ctx context.Context, e *Envelope) error
	GetMessage(ctx context.Context, msgID id.MessageID) (*Envelope, error)
	ListMessages(ctx context.Context, filter *MessageListFilter) ([]*Envelope, error)

	CreateConversation(ctx context.Context, c *Conversation) error
	GetConversation(ctx context.Context, convID id.ConversationID) (*Conversation, error)
	UpdateConversation(ctx context.Context, c *Conversation) error
	// GetConversationByPeerContext finds the conversation opened for a
	// remote thread. The node is part of the key because two peers can
	// use the same context id, and joining one peer's thread to
	// another's would leak a conversation across a trust boundary.
	GetConversationByPeerContext(ctx context.Context, node, peerContext string) (*Conversation, error)
	ListConversations(ctx context.Context, filter *ConversationListFilter) ([]*Conversation, error)

	CreateDelivery(ctx context.Context, d *Delivery) error
	// GetDelivery reads one delivery row. A caller that was handed a
	// delivery id at send time uses it to find out what became of the
	// message, including the run it eventually started.
	GetDelivery(ctx context.Context, deliveryID id.DeliveryID) (*Delivery, error)
	UpdateDelivery(ctx context.Context, d *Delivery) error
	// ClaimDelivery takes ownership of a queued delivery and marks it
	// delivering. It returns ErrDeliveryAlreadyClaimed when the row is in
	// any other state, which is what stops two workers running one
	// directive twice.
	ClaimDelivery(ctx context.Context, deliveryID id.DeliveryID) (*Delivery, error)
	ListInbox(ctx context.Context, agentName string, filter InboxFilter) ([]*Delivery, error)
	ListQueuedDeliveries(ctx context.Context, limit int) ([]*Delivery, error)
	MarkDeliveryRead(ctx context.Context, deliveryID id.DeliveryID) error
	// ReclaimStaleDeliveries puts abandoned deliveries back in the queue
	// and reports how many it moved.
	//
	// A delivery is abandoned when it was claimed before olderThan and
	// never finished, which is what a process dying mid-delivery leaves
	// behind. Like the queued read, it deliberately crosses scopes: the
	// dispatcher runs per process rather than per tenant.
	ReclaimStaleDeliveries(ctx context.Context, olderThan time.Time, limit int) (int, error)

	CreatePendingAsk(ctx context.Context, a *PendingAsk) error
	// ClaimPendingAsk takes ownership of the ask carrying replyWith. It
	// returns ErrAskNotFound when no such row exists and
	// ErrAskAlreadyClaimed when another caller got there first. Claiming
	// before resuming is what keeps a run from being resumed twice.
	ClaimPendingAsk(ctx context.Context, replyWith string) (*PendingAsk, error)
	ListExpiredAsks(ctx context.Context, now time.Time, limit int) ([]*PendingAsk, error)
	// ListPendingAsksByConversation returns the unclaimed asks waiting on a
	// conversation, which is what a cancel has to un-pause.
	ListPendingAsksByConversation(ctx context.Context, convID id.ConversationID) ([]*PendingAsk, error)
}

type Tender struct {
	ConversationID id.ConversationID `json:"conversation_id"`
	ReplyWith      string            `json:"reply_with"`
	Proposals      []Proposal        `json:"proposals"`
	Refusals       []Proposal        `json:"refusals"`
	// Complete says every agent the call went to has answered.
	Complete bool `json:"complete"`
}

type Transport interface {
	// Deliver hands the envelope to the receiver. Returning an error marks
	// the delivery failed; it does not fail the sender's run.
	Deliver(ctx context.Context, e *Envelope, receiver Address) error
	// Handles reports whether this transport can reach the address.
	Handles(addr Address) bool
}