a2a

package
v1.10.0 Latest Latest
Warning

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

 Go to latest Published: Oct 30, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package a2a gives cortex agents direct, addressable communication: FIPA-ACL messages, durable conversations, mailboxes, and an ask that suspends the sender's run until a peer answers.

It is a leaf package. It depends only on cortex and id, and reaches host capability (running an agent, resuming a paused run, persistence, delivery) through injected interfaces, never by importing the engine.

Index

Constants

View Source 
const (
	ProtocolContractNet = "fipa-contract-net"
	ProtocolRequest     = "fipa-request"
	ProtocolQuery       = "fipa-query"
)

The FIPA interaction protocol names, as they travel on Envelope.Protocol.

Stamping one lets a reader of a stored conversation tell a tender from an ordinary exchange, and gives a remote peer the name the FIPA specification uses rather than one cortex made up.

View Source 
const (
	// StatusOpen means messages may still be delivered on it.
	StatusOpen = "open"
	// StatusClosed means a cancel closed it, or it ran to completion.
	StatusClosed = "closed"
	// StatusExpired means its deadline passed with an ask still waiting.
	StatusExpired = "expired"
)

Conversation statuses.

View Source 
const (
	// DeliveryQueued means the bus accepted it and nobody has carried it yet.
	DeliveryQueued = "queued"
	// DeliveryDelivering means a worker claimed it and is carrying it now.
	DeliveryDelivering = "delivering"
	// DeliveryDelivered means it reached the recipient: an inbox row for an
	// informative, a started run for a directive.
	DeliveryDelivered = "delivered"
	// DeliveryFailed means delivery was attempted and could not complete.
	DeliveryFailed = "failed"
)

Delivery states. The state is what the dispatcher redrives from after a restart, which is why it is kept apart from the read state below it.

View Source 
const (
	DefaultHopCeiling    = 8
	DefaultWorkers       = 4
	DefaultReplyBy       = 5 * time.Minute
	DefaultSweepInterval = 30 * time.Second
	// DefaultDeliveryClaimTTL is how long a claimed delivery may sit
	// before it is assumed abandoned.
	//
	// It is generous on purpose. A remote delivery legitimately holds its
	// claim while the peer is polled, which the client bounds at two
	// minutes, and reclaiming a delivery somebody is still carrying would
	// deliver the same message twice. Fifteen minutes is well past any
	// honest delivery and well short of a human noticing.
	DefaultDeliveryClaimTTL = 15 * time.Minute
)

Defaults for Options. The hop ceiling is the containment budget: eight derived messages is generous for a real delegation chain and short of anything that reads as a runaway.

View Source 
const (
	RecipientWaiting   = "waiting"
	RecipientSucceeded = "succeeded"
	RecipientFailed    = "failed"
)

Recipient observation states describe retained execution, not result assurance.

View Source 
const HandoffLanguage = "application/vnd.cortex.task-handoff+json;version=1"

HandoffLanguage carries the versioned JSON body through existing text parts.

View Source 
const HandoffProtocol = "cortex-task-handoff/1"

HandoffProtocol identifies assignment proposals, not ordinary agent prompts.

View Source 
const HeldAskMetadata = "cortex.held_ask"

HeldAskMetadata marks questions that must use versioned durable release. Older questions without this marker cannot be adopted as unreleased work.

Variables

View Source 
var (
	// ErrNoStore means NewBus was called without persistence.
	ErrNoStore = errors.New("cortex: a2a: no store configured")
	// ErrNoRunner means NewBus was called without an agent runner.
	ErrNoRunner = errors.New("cortex: a2a: no agent runner configured")
	// ErrConversationClosed means the conversation no longer accepts messages.
	ErrConversationClosed = errors.New("cortex: a2a: conversation is closed")
	// ErrHopCeiling means the conversation used up its containment budget.
	ErrHopCeiling = errors.New("cortex: a2a: conversation hop ceiling exceeded")
	// ErrUnroutable means no transport handles the receiver's address.
	ErrUnroutable = errors.New("cortex: a2a: no transport handles that address")
	// ErrUnknownReceiver means the address is routable in shape but names
	// no agent the sender can reach.
	ErrUnknownReceiver = errors.New("cortex: a2a: receiver does not resolve")
)

Bus errors.

View Source 
var (
	// ErrAskNeedsReceivers means Ask was given nobody to ask.
	ErrAskNeedsReceivers = errors.New("cortex: a2a: ask needs at least one receiver")
	// ErrAskNeedsDirective means the performative does not demand an answer,
	// so nothing would ever resume the asker.
	ErrAskNeedsDirective = errors.New("cortex: a2a: ask needs a directive performative")
)

Ask errors.

View Source 
var (
	// ErrInvalidPerformative means the envelope names a speech act that is
	// not one of the 22 FIPA-ACL performatives.
	ErrInvalidPerformative = errors.New("cortex: a2a: unknown performative")
	// ErrNoReceivers means the envelope addresses nobody, or addresses an
	// entry with an empty agent name.
	ErrNoReceivers = errors.New("cortex: a2a: envelope has no receivers")
	// ErrNoSender means the envelope carries no sender.
	ErrNoSender = errors.New("cortex: a2a: envelope has no sender")
	// ErrSelfAddressed means the sender is among the receivers. Cortex
	// refuses it outright: it is a loop with no use case behind it.
	ErrSelfAddressed = errors.New("cortex: a2a: envelope is self-addressed")
)

Envelope validation errors.

View Source 
var (
	// ErrHeldAskUnavailable means durable held-question release is unsupported.
	ErrHeldAskUnavailable = errors.New("cortex: a2a: durable held question release unavailable")
	// ErrHeldAskConflict means the question or pinned pause does not match.
	ErrHeldAskConflict = errors.New("cortex: a2a: held question binding conflict")
	// ErrHeldAskNotFound means no durable release receipt exists.
	ErrHeldAskNotFound = errors.New("cortex: a2a: held question release not found")
)
View Source 
var (
	// ErrAskNotFound means no pending ask carries that reply-with token.
	ErrAskNotFound = errors.New("cortex: a2a: pending ask not found")
	// ErrAskAlreadyClaimed is the losing side of a race between a reply, a
	// deadline sweep and a cancel. Exactly one of them may resume the run.
	ErrAskAlreadyClaimed = errors.New("cortex: a2a: pending ask already claimed")
)

Pending-ask errors.

View Source 
var (
	// ErrReplyCommitUnavailable means atomic reply publication is unsupported.
	ErrReplyCommitUnavailable = errors.New("cortex: a2a: atomic reply commit unavailable")
	// ErrReplyCommitConflict means a different result already owns this delivery.
	ErrReplyCommitConflict = errors.New("cortex: a2a: reply commit conflict")
	// ErrReplyCommitNotFound means no reply was committed for this delivery.
	ErrReplyCommitNotFound = errors.New("cortex: a2a: reply commit not found")
)
View Source 
var (
	// ErrAskResolutionUnavailable means the store cannot atomically retain an ask outcome.
	ErrAskResolutionUnavailable = errors.New("cortex: a2a: atomic ask resolution unavailable")
	// ErrAskResolutionConflict means the request or retained resolution is inconsistent.
	ErrAskResolutionConflict = errors.New("cortex: a2a: ask resolution conflict")
	// ErrAskResolutionNotFound means no outcome was retained for this ask.
	ErrAskResolutionNotFound = errors.New("cortex: a2a: ask resolution not found")
)
View Source 
var (
	// ErrMessageNotFound means no envelope carries that id in this scope.
	ErrMessageNotFound = errors.New("cortex: a2a: message not found")
	// ErrConversationNotFound means no conversation carries that id in this scope.
	ErrConversationNotFound = errors.New("cortex: a2a: conversation not found")
)

Store lookup errors. They are package sentinels rather than each backend's own error, so a caller matches one thing with errors.Is whichever database is underneath.

View Source 
var ErrDeliveryAlreadyClaimed = errors.New("cortex: a2a: delivery already claimed")

ErrDeliveryAlreadyClaimed is the losing side of two workers reaching for one row. It is what stops a directive running twice.

View Source 
var ErrDeliveryBindingUnavailable = errors.New("cortex: a2a: delivery run binding unavailable")

ErrDeliveryBindingUnavailable means persistence cannot fence run admission.

View Source 
var ErrDeliveryClaimLost = errors.New("cortex: a2a: delivery claim lost")

ErrDeliveryClaimLost means a worker's claim no longer permits a mutation.

View Source 
var ErrDeliveryNotFound = errors.New("cortex: a2a: delivery not found")

ErrDeliveryNotFound means no delivery row carries that id.

View Source 
var ErrRecipientDiscoveryUnavailable = errors.New("cortex: a2a: recipient discovery unavailable")

ErrRecipientDiscoveryUnavailable means bounded recipient discovery is absent.

View Source 
var ErrRecipientObservationUnavailable = errors.New("cortex: a2a: recipient observation unavailable")

ErrRecipientObservationUnavailable means no qualified observer was configured.

View Source 
var ErrReplyBinding = errors.New("cortex: a2a: invalid reply binding")

ErrReplyBinding means retained ask identity disagrees with its source message.

View Source 
var ErrRosterDenied = errors.New("cortex: a2a: endpoint outside participant roster")

ErrRosterDenied means the scope or endpoint is outside the captured team.

View Source 
var ErrRosterUnavailable = errors.New("cortex: a2a: participant roster unavailable")

ErrRosterUnavailable means identity lookup has no captured roster to consult.

View Source 
var ErrRoutingUnavailable = errors.New("cortex: a2a: orchestration routing unavailable")

ErrRoutingUnavailable means the store cannot isolate orchestration traffic.

Functions

func IsHandoffProtocol 

func IsHandoffProtocol(protocol, language string) bool

IsHandoffProtocol reserves the namespace, including unsupported versions.

func RenderInput 

func RenderInput(e *Envelope) string

RenderInput turns an envelope into the text a recipient's run receives. It names the sender and the speech act, because an agent that cannot tell a request from a proposal cannot answer either one properly.

func RoutingDiscoveryScope 

func RoutingDiscoveryScope(ctx context.Context) (cortex.Scope, bool, error)

RoutingDiscoveryScope returns the exact scope selected for background work. Without a binding, existing global queue discovery retains its scope policy.

func RoutingOrchestration 

func RoutingOrchestration(ctx context.Context) string

RoutingOrchestration is the partition selected by NewRoutingStore. Backends must apply it to every read, mutation and dispatcher query. Empty selects legacy global traffic, never all partitions.

func WithReplyRoute added in v1.10.0 

func WithReplyRoute(ctx context.Context, route ReplyRoute) context.Context

WithReplyRoute is ctx carrying the route of the directive a run is being started for. Only the runner the bus calls should read it.

Types

type Address 

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

Address identifies one messaging endpoint. Node empty means an agent in this engine; a non-empty Node names a remote peer and is the only field remote delivery needs.

func ParseAddress 

func ParseAddress(s string) Address

ParseAddress reads the textual form agents use: "worker" for a local agent, "worker@peer.example" for one at a remote node.

It splits on the LAST @, because an agent name cannot contain one but a node conceivably can. Without this, an agent naming a remote peer would be looked up as a local agent whose name happens to contain an @, which fails as "agent not found" and says nothing about why.

func (Address) Equal 

func (a Address) Equal(other Address) bool

Equal reports whether two addresses name the same endpoint.

func (Address) IsLocal 

func (a Address) IsLocal() bool

IsLocal reports whether the address resolves inside this engine.

func (Address) IsZero 

func (a Address) IsZero() bool

IsZero reports whether the address names nothing.

func (Address) String 

func (a Address) String() string

String renders the address as agent, or agent@node for a remote peer.

type Admitter added in v1.10.0 

type Admitter interface {
	AdmitRun(ctx context.Context) (context.Context, func(), error)
}

Admitter lets the host decide whether this process may carry work that can start or resume a run. The bus asks before it claims a delivery and before it claims or seals a waiting ask, so a refusal leaves both untouched for another worker. A refusal matching cortex.ErrEngineStopping means "not here, not now": the bus stops carrying and sends no reply.

The returned context must be used for the work and release called when it ends. A context that is already admitted is admitted again at once, which is what lets a resume that follows a consumed claim finish while the host drains.

type AskParams 

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
}

AskParams is one message whose sender suspends until the answer arrives.

type AskReply 

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"`
}

AskReply is what a resumed agent_ask tool call returns to the model.

Replies is plural because an ask may have gone to several agents at once, which is what a call for proposals is. A single-recipient ask carries exactly one entry, so the payload never changes shape with the number of recipients: an agent reads the same structure either way.

type AskReplyItem 

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

AskReplyItem is one agent's answer.

type AskResolution 

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:"-"`
}

AskResolution retains the exact tool result selected when an ask was claimed. Fresh belongs only to the commit acknowledgment. Observing a resolution does not authorize another resume or prove that the original resume completed.

type AskResolutionPage 

type AskResolutionPage struct {
	Resolutions []AskResolution
	Next        string
}

AskResolutionPage is bounded history in one exact scope and routing partition. Next is an opaque cursor, empty when the selected history is exhausted.

type AskResolutionStore 

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)
}

AskResolutionStore claims an ask and retains its exact outcome in one commit. Equal retries return the retained observation; a competing outcome returns ErrAskAlreadyClaimed. Changed ask identity fails. Legacy claims without an outcome cannot be adopted. Implementations never resume a run in a transaction.

type AskResult 

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"`
	// ReplyBy is when the answer is due, the ask's own deadline. The ask
	// sweep resolves the ask at this instant, and an engine suspending
	// its run on the ask sets the run's deadline from it.
	ReplyBy time.Time `json:"reply_by"`
	// contains filtered or unexported fields
}

AskResult identifies the message and the token a reply must carry.

type AtomicMessageStore 

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

AtomicMessageStore serializes hop admission with immutable envelope storage. An identical retry returns the original envelope without another hop charge.

type Bus 

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

Bus routes envelopes between agents.

func NewBus 

func NewBus(cfg BusConfig) (*Bus, error)

NewBus builds a Bus from cfg.

func (*Bus) AddTransport 

func (b *Bus) AddTransport(t Transport)

AddTransport registers a transport after the bus has been built.

It exists because construction cannot be circular. A remote transport needs the bus, since a peer's reply is fed back through Send so it can resolve a waiting ask, and the bus needs the transport to carry the question out. The host builds the bus, builds the transport with it, and registers it here.

func (*Bus) Ask 

func (b *Bus) Ask(ctx context.Context, p AskParams) (*AskResult, error)

Ask sends a directive and records the sender's run as waiting on the answer. The caller suspends its run once this returns.

The writes go message, then pending ask, then routing, and both halves of that order are load-bearing. A pending ask with no message behind it is a run nothing could ever resume. A question routed before its pending ask exists can be answered before there is anything for the answer to claim, and the reply then queues as an ordinary message while the asker waits out its deadline.

Everything that can refuse the send happens before the first write.

func (*Bus) Drain 

func (b *Bus) Drain(ctx context.Context) (int, error)

Drain delivers everything currently queued and reports how many rows it carried. Tests call it directly; workers call it in a loop. When the bus's Admitter refuses with cortex.ErrEngineStopping, Drain stops and returns that error, leaving the remaining rows queued.

The count includes replies the directives produced, because a reply is itself a queued delivery and a drain that left them behind would stop halfway through the conversation it just started.

func (*Bus) HeldAskPending 

func (b *Bus) HeldAskPending(ctx context.Context, token string) (bool, error)

HeldAskPending observes a marked question lacking a release receipt. The host must authorize its run before calling. Legacy questions are never adopted.

func (*Bus) Inbox 

func (b *Bus) Inbox(ctx context.Context, agentName string, f InboxFilter) ([]InboxItem, error)

Inbox returns delivered messages for an agent and marks what it returns as read. Reading is the acknowledgement: an agent that already saw a message in a tool result must not be handed it again next turn.

func (*Bus) ParticipantAddress 

func (b *Bus) ParticipantAddress(ctx context.Context, agentID id.AgentID) (Address, error)

ParticipantAddress resolves a captured agent ID without consulting mutable names. The host must authenticate the invoking run before using its agent identity.

func (*Bus) PendingDispatch 

func (b *Bus) PendingDispatch(ctx context.Context) (bool, error)

PendingDispatch reports queued deliveries or expired asks in the captured roster scope. Current discovery authority is checked around the metadata read. It never claims work, publishes a reply or resumes an agent.

func (*Bus) ReclaimStaleDeliveries 

func (b *Bus) ReclaimStaleDeliveries(ctx context.Context) (int, error)

ReclaimStaleDeliveries puts back any delivery a dead process was carrying, and reports how many it moved.

It puts them back in the queue rather than delivering them here, so a reclaimed message takes the ordinary path with the ordinary claim, and a worker that turns out to still be alive loses the race rather than duplicating the delivery.

func (*Bus) RecoverRecipient 

func (b *Bus) RecoverRecipient(ctx context.Context, key id.DeliveryID, authorize func(context.Context, id.OrchestrationID, id.DeliveryID) error) (ReservedReply, error)

RecoverRecipient publishes a terminal result for a bound delivery without reserving or executing its child again. The host authorizes this exact team and delivery before reads and again after observation. Waiting returns an empty receipt. Equal retries return the committed receipt without observing the child again. This API is not a model tool or an unauthenticated route.

func (*Bus) RecoverRecipients 

func (b *Bus) RecoverRecipients(ctx context.Context, after id.DeliveryID) (RecipientRecoveryPage, error)

RecoverRecipients discovers and observes one bounded page. It never reserves, executes or resumes a recipient. Callers of synchronous buses drive pages themselves; asynchronous buses run this pass from Start until Stop joins it.

func (*Bus) Redrive 

func (b *Bus) Redrive(ctx context.Context) (int, error)

Redrive picks up deliveries a previous process queued and never carried. It is the work Drain does; the separate name is for the caller that runs it once at startup.

func (*Bus) Release 

func (b *Bus) Release(ctx context.Context, res *AskResult) error

Release routes an ask that was sent with Hold. Calling it on an ask that was routed when it was sent, or calling it twice, does nothing: queueing the deliveries again would carry the question twice.

The caller releases once the asker has paused. A release that fails leaves the ask waiting with nothing carrying the question, and the deadline sweep resolves it into a failure the asking agent can read.

func (*Bus) ReleasePausedAsk 

func (b *Bus) ReleasePausedAsk(ctx context.Context, req HeldAskRequest) (HeldAskReceipt, error)

ReleasePausedAsk queues a stored question only after its exact pause exists. A lost acknowledgment is recovered by retrying the same request. The host authorizes the current caller; storage validates the immutable binding.

func (*Bus) ReplyForRun added in v1.10.0 

func (b *Bus) ReplyForRun(ctx context.Context, route ReplyRoute, output string, runErr error) error

ReplyForRun sends the answer of a directive whose run the worker that started it handed off, from the worker that finished the run. runErr is why the run did not complete, or nil. A reply that cannot be sent still reaches whoever is waiting on it, as a failure.

func (*Bus) Send 

func (b *Bus) Send(ctx context.Context, p SendParams) (*SendResult, error)

Send validates, persists and queues one message. It returns as soon as the deliveries are queued; nothing waits for the recipients.

func (*Bus) Start 

func (b *Bus) Start(ctx context.Context) error

Start launches the delivery workers. Calling it twice is a no-op, and in synchronous mode it starts nothing at all.

func (*Bus) Stop 

func (b *Bus) Stop()

Stop cancels the workers and WAITS for them. Signalling without waiting would let Stop return while a delivery was still writing.

func (*Bus) SupportsHeldAskRelease 

func (b *Bus) SupportsHeldAskRelease() bool

SupportsHeldAskRelease reports whether the configured backend supports release.

func (*Bus) SweepExpiredAsks 

func (b *Bus) SweepExpiredAsks(ctx context.Context) (int, error)

SweepExpiredAsks resolves every ask whose deadline has passed into a timeout failure and resumes the run waiting on it, returning how many it resolved.

The engine's own suspension sweep FAILS a run nobody answered in time. For an agent-reply pause that is the wrong verb, so this runs first: a peer that did not answer is something the asking agent can react to, and killing the run throws that away. The engine sweep stays as the outer backstop for anything this missed.

When the bus's Admitter refuses with cortex.ErrEngineStopping, the pass ends early without an error and the remaining asks stay unclaimed.

type BusConfig 

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

	// Admitter, when set, is asked before the bus claims anything that can
	// start or resume a run. Nil admits everything.
	Admitter Admitter

	// 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
}

BusConfig builds a Bus. Store and one runner are required. Structured handoffs require HandoffHandler explicitly.

type Class 

type Class string

Class is how cortex routes a performative on arrival. 

const (
	// ClassDirective starts a run for the recipient; its output is the reply.
	ClassDirective Class = "directive"
	// ClassInformative lands in the recipient's inbox and starts nothing.
	ClassInformative Class = "informative"
	// ClassControl is interpreted by the bus itself and reaches no agent.
	ClassControl Class = "control"
)

The three routing classes.

type Clock 

type Clock interface {
	Now() time.Time
}

Clock is the package's only source of time. Everything reads it, so a test can move a deadline without sleeping.

type ContractNetParams 

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

ContractNetParams is one call for proposals.

type Conversation 

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"`
}

Conversation is one thread of messages between agents. It carries the containment budget: every derived message increments HopsUsed, and the bus refuses delivery once HopsUsed would exceed HopCeiling.

func (*Conversation) AddParticipant 

func (c *Conversation) AddParticipant(addr Address)

AddParticipant records addr as a participant, ignoring duplicates.

func (*Conversation) HasParticipant 

func (c *Conversation) HasParticipant(addr Address) bool

HasParticipant reports whether addr already took part.

func (*Conversation) IsOpen 

func (c *Conversation) IsOpen() bool

IsOpen reports whether the conversation still accepts messages.

type ConversationListFilter 

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

ConversationListFilter controls a conversation listing.

type Delivery 

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
}

Delivery is one envelope's arrival at one receiver. A message addressed to five agents is five deliveries, which is what makes "unread for agent B" an answerable question.

type DeliveryAdmissionError 

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

DeliveryAdmissionError retains an attempted child when dispatch is refused. Its identity is evidence for reconciliation, never permission to replay.

func (*DeliveryAdmissionError) DeliveryIdentity 

func (e *DeliveryAdmissionError) DeliveryIdentity() id.DeliveryID

DeliveryIdentity returns the delivery whose admission failed.

func (*DeliveryAdmissionError) Error 

func (e *DeliveryAdmissionError) Error() string

func (*DeliveryAdmissionError) RunIdentity 

func (e *DeliveryAdmissionError) RunIdentity() id.AgentRunID

RunIdentity returns the reserved child's identity, including uncertain writes.

func (*DeliveryAdmissionError) Unwrap 

func (e *DeliveryAdmissionError) Unwrap() error

Unwrap preserves matching of the underlying admission failure.

type DeliveryOutcome 

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"`
}

DeliveryOutcome is one receiver's result from a send. A broadcast reports per receiver, because failing the whole send over one bad name throws away the deliveries that were fine.

type DeliveryRunBinder 

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

DeliveryRunBinder atomically records a reserved run under the current claim. Binding is single-use, including after a lost acknowledgment. A bound delivery cannot be reclaimed by timeout; the host must reconcile its execution first.

type Envelope 

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"`
}

Envelope is one FIPA-ACL message. The first block is the ACL parameter set, verbatim; the second is cortex's own additions, kept apart from it.

func (*Envelope) Validate 

func (e *Envelope) Validate() error

Validate checks everything about an envelope that can be decided without touching the store: the performative is real, somebody sent it, somebody receives it, and the sender is not among the receivers.

type HandoffHandler 

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

HandoffHandler keeps assignment policy outside the messaging leaf package. ValidateHandoff authenticates structure and transport bindings before storage and delivery. ReceiveHandoff admits local work with current host authorization and durable deduplication. Neither method may turn a proposal into task success. Implementations must honor cancellation and retain no mutable envelope aliases.

type HeldAskReceipt 

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:"-"`
}

HeldAskReceipt retains the exact deliveries queued by one release transaction. Reading or retrying it grants no authority to execute a recipient.

type HeldAskRequest 

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"`
}

HeldAskRequest pins one stored question to its original paused run.

type HeldAskStore 

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

HeldAskStore atomically validates a pinned pause and queues its question. An identical retry returns the original receipt, including after resume. All operations require exact scope and routing partition.

type HookEmitter 

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)
}

HookEmitter receives messaging lifecycle events. The engine adapts plugin.Registry to it; tests pass a recorder.

type InboxFilter 

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

InboxFilter controls an inbox listing. Scope arrives on the context.

type InboxItem 

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"`
}

InboxItem is one delivered message as an agent sees it.

type MessageListFilter 

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

MessageListFilter controls a message listing. Scope arrives on the context; Exact narrows to rows stored at precisely that depth instead of everything beneath it.

type Options 

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
}

Options tunes the messaging subsystem.

type PendingAsk 

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"`
}

PendingAsk is one suspended run waiting on a peer's answer. It is keyed by ReplyWith, which is FIPA's own correlation token, so a reply carrying InReplyTo finds exactly one row.

type Performative 

type Performative string

Performative is the speech act a message performs. The 22 constants below are the complete FIPA-ACL set. 

const (
	AcceptProposal  Performative = "accept-proposal"
	Agree           Performative = "agree"
	Cancel          Performative = "cancel"
	CFP             Performative = "cfp"
	Confirm         Performative = "confirm"
	Disconfirm      Performative = "disconfirm"
	Failure         Performative = "failure"
	Inform          Performative = "inform"
	InformIf        Performative = "inform-if"
	InformRef       Performative = "inform-ref"
	NotUnderstood   Performative = "not-understood"
	Propagate       Performative = "propagate"
	Propose         Performative = "propose"
	Proxy           Performative = "proxy"
	QueryIf         Performative = "query-if"
	QueryRef        Performative = "query-ref"
	Refuse          Performative = "refuse"
	RejectProposal  Performative = "reject-proposal"
	Request         Performative = "request"
	RequestWhen     Performative = "request-when"
	RequestWhenever Performative = "request-whenever"
	Subscribe       Performative = "subscribe"
)

The FIPA-ACL performatives.

func AllPerformatives 

func AllPerformatives() []Performative

AllPerformatives returns every recognised performative, for validation and for exhaustive tests.

func (Performative) AnswersAsk 

func (p Performative) AnswersAsk() bool

AnswersAsk reports whether a reply carrying p counts as an answer to a waiting ask.

agree is deliberately excluded. It means the peer accepted the task and is still working on it, so an asker that counted it would resume on a message carrying no answer.

refuse and reject-proposal DO count. Declining is an answer: in a tender, a participant that will not bid has told the initiator what it needed to know about that participant.

func (Performative) Class 

func (p Performative) Class() (Class, bool)

Class returns the routing class for p, and whether p is a performative cortex recognises at all.

func (Performative) Valid 

func (p Performative) Valid() bool

Valid reports whether p is one of the 22 FIPA-ACL performatives.

type Proposal 

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

Proposal is one participant's answer to a call for proposals.

type RecipientDiscoveryStore 

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

RecipientDiscoveryStore lists bound deliveries still awaiting a durable reply. Reads require exact scope and routing partition, return identities only, and order them strictly after the cursor. Limit must be between 1 and 256. Discovery grants no execution or observation authority.

type RecipientObservation 

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

RecipientObservation is a detached, durable child outcome. Successful Output must already be approved for publication. Failed and waiting output is ignored.

type RecipientObserver 

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

RecipientObserver reads a child's durable state without executing or resuming it. Validate its scope, captured participant, prepared provenance and current host authority. A read or authorization error is not a terminal child failure.

type RecipientRecoveryConfig 

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)
}

RecipientRecoveryConfig enables one bounded discovery worker on a roster-bound reserved bus. Both authorization callbacks must consult current host authority. Callbacks must honor cancellation. Configuration is copied during NewBus.

type RecipientRecoveryPage 

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

RecipientRecoveryPage reports one pass without exposing message or child data. Next is the exclusive cursor when the page was full, or zero at the end. Per-delivery errors advance Next so one denied or broken row cannot starve later rows. The next traversal retries those rows from the beginning.

type RecipientRequest 

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

RecipientRequest binds an observation to the original reserved child and team.

type ReplyRoute added in v1.10.0 

type ReplyRoute struct {
	// Sender is the recipient, which answers.
	Sender Address `json:"sender"`
	// Receiver is the asker, which the answer goes to.
	Receiver       Address           `json:"receiver"`
	ConversationID id.ConversationID `json:"conversation_id"`
	InReplyTo      string            `json:"in_reply_to,omitempty"`
	Protocol       string            `json:"protocol,omitempty"`
	Ontology       string            `json:"ontology,omitempty"`
}

ReplyRoute is where the answer to a directive goes: back from the recipient to the asker, on the same conversation. The bus hands it to the recipient's run when it starts it, and the run keeps it, so a run that is handed off and finished by another worker can still answer.

func ReplyRouteFrom added in v1.10.0 

func ReplyRouteFrom(ctx context.Context) (ReplyRoute, context.Context, bool)

ReplyRouteFrom returns the route WithReplyRoute put on ctx, and ctx without it, so the run started with it does not hand it on to runs its own tools start.

type ReservedReply 

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:"-"`
}

ReservedReply retains the reply and queue identities committed with a bound delivery. Fresh describes this acknowledgment only and is never persisted. Reading a receipt grants no permission to execute or resume a child.

type ReservedReplyStore 

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

ReservedReplyStore atomically finishes the claimed delivery, records its immutable reply, charges a conversation hop and queues the reply delivery. It never invokes a provider or resumes a run. Equal retries return the stored receipt; changed content or binding fails. All operations require exact scope and routing partition. Success describes the child result, not its assurance.

type ResolutionResumer 

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

ResolutionResumer consumes a retained outcome with its question identity. Native buses prefer it to the legacy run-ID-only resumer. Receiving a receipt does not grant authority to replay a previously attempted resume.

type Resolver 

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

Resolver answers whether an address names an agent that actually exists in the caller's scope. The engine implements it with an agent lookup; a remote transport's node would answer for its own peers.

It is separate from Transport.Handles, and the difference matters: a transport says "I know how to reach addresses of this shape", while a resolver says "this particular agent is there". Without the second question, an ask addressed to a typo suspends the asking run against a recipient that will never answer, and the run sits until its deadline to learn what could have been said immediately.

type Resumer 

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

Resumer un-pauses a run that stopped on an agent_ask. It wraps the engine's internal resume path, not the public Resume, because a host must not be able to forge a peer's reply.

type Roster 

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

Roster fixes the local identities allowed to exchange messages in one scope. NewBus detaches it; callers cannot change a running bus's membership.

type RosterParticipant 

type RosterParticipant struct {
	Name    string
	AgentID id.AgentID
}

RosterParticipant binds a captured name to its original agent identity.

type RoutingBackend 

type RoutingBackend interface {
	Store
	SupportsOrchestrationRouting() bool
}

RoutingBackend provides persistence that partitions every A2A operation.

type ScopedRoutingBackend 

type ScopedRoutingBackend interface {
	RoutingBackend
	SupportsScopedOrchestrationDiscovery() bool
}

ScopedRoutingBackend also restricts queue discovery, reclaim and ask expiry when RoutingDiscoveryScope is present. Roster-bound buses require it.

type SendParams 

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
}

SendParams is one outbound message.

type SendResult 

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

SendResult is what a send produced.

type Store 

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)
}

Store is persistence for the messaging subsystem. It folds into the composite store.Store the same way orchestration's two interfaces do.

func NewRoutingStore 

func NewRoutingStore(s Store, key id.OrchestrationID) (Store, error)

NewRoutingStore binds operations to one orchestration independently of scope. Zero selects global traffic. A nonzero ID requires a qualified backend. Binding does not authorize participants or provision an execution environment.

type Tender 

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"`
}

Tender is the state of one call for proposals.

func CollectTender 

func CollectTender(ctx context.Context, b *Bus, convID id.ConversationID, replyWith string) (*Tender, error)

CollectTender reads what a call for proposals has gathered so far, separating the bids from the declines.

It reads the conversation rather than any state of its own, so it is safe to call at any point and after a restart.

func ContractNet 

func ContractNet(ctx context.Context, b *Bus, p ContractNetParams) (*Tender, error)

ContractNet announces a task to several agents at once.

It sends the call and returns the handle to collect against; it does not wait, and it does not choose. Waiting is what the asking run's own suspension does, and choosing is a judgement this package has no business making. An initiator picks its contractor and then awards with accept-proposal, which is an ordinary directive.

This is a convenience for hosts driving a tender in Go. An agent needs nothing from here: a call for proposals is agent_ask with several recipients and the cfp performative, which is the whole point of building the protocol out of the primitives rather than beside them.

type Transport 

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
}

Transport delivers an envelope to one receiver. The in-process implementation resolves agents in this engine; a remote implementation resolves a Node.

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.