Documentation
¶
Overview ¶
Package broker provides a harness-agnostic run broker for agent communication. It handles run registration, control message queuing, report/finish/reply ingestion, and heartbeat tracking scoped by run ID.
Harness-specific concerns (e.g., MCP sidecar registration, tmux bootstrap, .mcp.json entries) live in the individual harness adapter packages, not here.
Index ¶
- func EnrichStopReason(b *Broker, brokerRunID string, stopReason *string)
- func MakeToken() string
- type AgentMessage
- type Broker
- func (b *Broker) Addr() string
- func (b *Broker) CreateRun(runID string) (string, error)
- func (b *Broker) DeleteRun(runID string)
- func (b *Broker) EnsureRun(runID string) (string, bool)
- func (b *Broker) FinishCount(runID string) int
- func (b *Broker) GetRun(runID string) *RunState
- func (b *Broker) Ingest(runID string, kind string, payload json.RawMessage) error
- func (b *Broker) LastFinishStatus(runID string) string
- func (b *Broker) PollAgentMessages(ctx context.Context, runID string, onMessage func(wrappedContent string))
- func (b *Broker) PushControl(runID string, msg ControlMessage) error
- func (b *Broker) Reset()
- func (b *Broker) RunCount() int
- func (b *Broker) Send(runID string, kind string, payload json.RawMessage, correlationID string) error
- func (b *Broker) SendTo(fromRunID, toRunID, kind string, payload json.RawMessage, correlationID string) error
- func (b *Broker) Start() error
- func (b *Broker) Stop() error
- type ControlMessage
- type Envelope
- type Finish
- type PermissionState
- type Recorder
- type Reply
- type Report
- type RunState
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func EnrichStopReason ¶
EnrichStopReason supplements a stop reason with the last finish status from a broker run, when the stop reason is empty and broker state is available. It is safe to call when broker is nil or brokerRunID is empty.
Types ¶
type AgentMessage ¶
type AgentMessage struct {
From string `json:"from"`
FromRunID string `json:"from_run_id"`
Message string `json:"message"`
Role string `json:"role"`
}
AgentMessage is the payload for agent-to-agent channel messages. It is used as the payload of "agent_message" control messages pushed via SendTo.
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker is a harness-agnostic, in-process HTTP server that coordinates communication between agent harnesses and their orchestrators. It owns run-scoped state, message queuing, and lifecycle event ingestion.
func New ¶
New creates a broker on an ephemeral loopback port. The resulting addr is available after Start.
func (*Broker) DeleteRun ¶
DeleteRun removes a run from the broker. Safe to call even if the run does not exist.
func (*Broker) EnsureRun ¶
EnsureRun registers a new run if one does not already exist for the given runID. Returns the run's token (existing or newly created) and whether the run was newly created.
func (*Broker) FinishCount ¶
func (*Broker) Ingest ¶
Ingest routes an incoming message from an agent into the appropriate RunState bucket based on kind.
Known kinds:
- "done", "failed", "blocked" → stored in Finishes
- everything else → stored in Reports
func (*Broker) LastFinishStatus ¶
func (*Broker) PollAgentMessages ¶
func (b *Broker) PollAgentMessages(ctx context.Context, runID string, onMessage func(wrappedContent string))
PollAgentMessages polls the broker for agent_message control messages destined to the given runID every 2 seconds. Each received message is wrapped with channelwrap and passed to onMessage. The goroutine exits when ctx is done.
func (*Broker) PushControl ¶
func (b *Broker) PushControl(runID string, msg ControlMessage) error
PushControl queues a control message for the given run.
func (*Broker) Send ¶
func (b *Broker) Send(runID string, kind string, payload json.RawMessage, correlationID string) error
Send enqueues a control message for the given run using raw fields. It is a convenience wrapper around PushControl that builds the ControlMessage internally. correlationID is optional; it is used as the ControlMessage ID when non-empty.
func (*Broker) SendTo ¶
func (b *Broker) SendTo(fromRunID, toRunID, kind string, payload json.RawMessage, correlationID string) error
SendTo sends a message from one run to another run's control queue. The fromRunID must belong to a registered run whose token matches the authenticated caller's token.
type ControlMessage ¶
type ControlMessage struct {
ID string `json:"id"`
Type string `json:"type"`
RunID string `json:"run_id"`
FromRunID string `json:"from_run_id,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
ControlMessage is a message pushed from Avenor to the sidecar for delivery to Claude.
func (ControlMessage) ToEnvelope ¶
func (m ControlMessage) ToEnvelope() Envelope
type Envelope ¶
type Envelope struct {
FromRunID string `json:"from_run_id"`
ToRunID string `json:"to_run_id"`
To string `json:"to,omitempty"`
Kind string `json:"kind"`
CorrelationID string `json:"correlation_id,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
Envelope is the internal generic message format used by the broker. It is not exposed on the wire; HTTP endpoints continue to use the existing Report/Finish/Reply/ControlMessage types.
type Finish ¶
type Finish struct {
RunID string `json:"run_id"`
Status string `json:"status"`
Summary string `json:"summary"`
FilesChanged []string `json:"files_changed,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
}
Finish is the payload from an avenor_finish tool call.
func (Finish) ToEnvelope ¶
type PermissionState ¶
type PermissionState struct {
RequestID string
ToolName string
Desc string
Preview string
CreatedAt time.Time
}
PermissionState holds a pending permission relay request.
type Recorder ¶
type Recorder struct {
// contains filtered or unexported fields
}
Recorder translates events.Event values into broker.Ingest calls so that any backend can mirror its event stream into the broker without an agent-side sidecar process.
Usage:
rec := broker.NewRecorder(brk, runID)
for evt := range provider.Events(ctx, sessionID) {
rec.Feed(evt)
subscriberCh <- evt // keep existing fanout
}
func NewRecorder ¶
NewRecorder creates a Recorder for the given broker and run ID.
func (*Recorder) Feed ¶
Feed translates an event into a broker.Ingest call. Events not listed below are silently dropped — the Recorder is a filtered mirror, not a complete event log.
session.end → kind = stop_reason (e.g. "done", "failed") — stored in Finishes agent.status → kind = "status" — stored in Reports agent.message_chunk → kind = "thinking" — stored in Reports permission.request → kind = "permission_requested" — stored in Reports child.question → kind = "child_question" — stored in Reports
Ingest errors (run not found, broker not started) are intentionally suppressed — the Recorder is best-effort and must not block or panic the event pump.
type Reply ¶
type Reply struct {
RunID string `json:"run_id"`
To string `json:"to"`
Payload json.RawMessage `json:"payload,omitempty"`
}
Reply is the payload from an avenor_reply tool call.
func (Reply) ToEnvelope ¶
type Report ¶
type Report struct {
RunID string `json:"run_id"`
State string `json:"state"`
Payload json.RawMessage `json:"payload,omitempty"`
}
Report is the payload from an avenor_report tool call.
func (Report) ToEnvelope ¶
type RunState ¶
type RunState struct {
RunID string
Token string
ControlQueue []*ControlMessage
RegisteredAt time.Time
LastSeen time.Time
Reports []Report
Finishes []Finish
Replies []Reply
PermissionRequests map[string]*PermissionState
PermissionDecisions map[string]string // requestID -> "allow" or "deny"
Mu sync.Mutex
Notify chan struct{}
}
RunState holds all per-run broker state.
Source Files