broker

package
v0.20.2 Latest Latest
Warning

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

 Go to latest Published: Jul 17, 2026 License: MIT Imports: 15 Imported by: 0

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

Constants

This section is empty.

Variables

This section is empty.

Functions

func EnrichStopReason 

func EnrichStopReason(b *Broker, brokerRunID string, stopReason *string)

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.

func MakeToken 

func MakeToken() string

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 

func New(globalToken string) *Broker

New creates a broker on an ephemeral loopback port. The resulting addr is available after Start.

func (*Broker) Addr 

func (b *Broker) Addr() string

func (*Broker) CreateRun 

func (b *Broker) CreateRun(runID string) (string, error)

CreateRun registers a new run and returns its token.

func (*Broker) DeleteRun 

func (b *Broker) DeleteRun(runID string)

DeleteRun removes a run from the broker. Safe to call even if the run does not exist.

func (*Broker) EnsureRun 

func (b *Broker) EnsureRun(runID string) (string, bool)

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 (b *Broker) FinishCount(runID string) int

func (*Broker) GetRun 

func (b *Broker) GetRun(runID string) *RunState

GetRun returns the RunState for a given run ID, or nil if not found.

func (*Broker) Ingest 

func (b *Broker) Ingest(runID string, kind string, payload json.RawMessage) error

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 (b *Broker) LastFinishStatus(runID string) string

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

func (b *Broker) Reset()

Reset clears all run state.

func (*Broker) RunCount 

func (b *Broker) RunCount() int

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.

func (*Broker) Start 

func (b *Broker) Start() error

Start binds the broker to an ephemeral loopback port.

func (*Broker) Stop 

func (b *Broker) Stop() error

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 

func (f Finish) ToEnvelope() Envelope

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 

func NewRecorder(b *Broker, runID string) *Recorder

NewRecorder creates a Recorder for the given broker and run ID.

func (*Recorder) Feed 

func (rec *Recorder) Feed(evt events.Event)

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 

func (r Reply) ToEnvelope() Envelope

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 

func (r Report) ToEnvelope() Envelope

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.

func (*RunState) Lock 

func (st *RunState) Lock()

func (*RunState) Unlock 

func (st *RunState) Unlock()

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.