biam

package
v0.22.166 Latest Latest
Warning

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

 Go to latest Published: May 25, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Overview

Package biam — Unix-socket dispatch server. Lets `clawtool send --async` (a separate CLI process from the daemon) hand a prompt off to the daemon's BIAM runner so the dispatch goroutine lives in the daemon process. That guarantees the WatchHub frame broadcasts cross to the orchestrator's socket subscribers — the CLI's own in-process WatchHub never leaves its process.

Without this socket, async CLI dispatches would spawn a short-lived runner inside the CLI process, frames would broadcast only on the CLI's WatchHub, and the orchestrator (subscribed to the daemon's hub) would see zero stream lines even though the task itself transits SQLite via the store hook.

Wire format: JSON-line dispatch request → JSON-line dispatch response. One request per connection, then close.

Permissions: socket file is mode 0600 — same posture as the task-watch socket. XDG_STATE_HOME lives outside config + data, matching the runtime-state convention.

Package biam — TaskEventBuffer is the per-task ring buffer that backs the SSE A2A async-push handler at GET /v1/biam/subscribe.

Why a separate buffer alongside WatchHub:

  • WatchHub is a process-wide live broadcaster with no replay. A subscriber that connects mid-stream sees only events from that point forward.
  • SSE clients reconnect with a Last-Event-ID header and expect the server to replay everything that happened between the drop and the resume. WatchHub's design (drop-on-slow-consumer, no history) can't do that.
  • A2A peers polling /v1/biam/subscribe?task_id=… need both replay (for crash-recovery / resume-after-disconnect) and edge-triggered push (so they don't poll a dead socket).

Design (per ADR-024 §Resolved 2026-05-02):

  • One TaskEventBuffer per process (singleton `Events`).
  • Per-task circular buffer, capacity 256 events. Oldest events drop when the cap is hit.
  • Each event carries a u64 monotonic ID (per-task, starts at 1). The SSE handler emits these as `id:` lines so a reconnecting client's Last-Event-ID header lets us replay only the missed suffix.
  • Append fires a non-blocking notify on every channel registered for that task. SSE handlers select{} on notify + ctx.Done.
  • Bounded-buffer drops are visible: when the requested Last-Event-ID is older than the oldest retained event, the replay returns from the earliest available + the buffer's dropped-prefix marker so the client can detect the gap.

Lifetime: the daemon constructs the singleton at boot. Tests use NewTaskEventBuffer for isolation.

Package biam — bridge between WatchHub broadcasts and the per-task TaskEventBuffer that backs /v1/biam/subscribe SSE replay.

Why a bridge: WatchHub is the live broadcaster every existing in-process consumer (orchestrator, dashboard, watchsocket) reads from; the SSE handler needs a *replayable* per-task history. The bridge subscribes to WatchHub and rewrites every event into the per-task ring so SSE clients get both the snapshot+replay and the edge-trigger fan-out.

Wired once at daemon boot from internal/server.buildMCPServer.

Package biam — Bidirectional Inter-Agent Messaging substrate (ADR-015 Phase 1). identity.go owns the per-instance Ed25519 keypair: every clawtool listener generates one on first launch at ~/.config/clawtool/identity.ed25519 and exchanges public keys with peers via the trust file (peers.toml). Signed envelopes use the private key; receivers verify against the trust map.

The identity file is mode 0600 + 32-byte raw seed; the public key is derived deterministically. We don't ship a CA or PKI — peer trust is operator-managed (one-line `clawtool peer add`).

Package biam — process-internal completion notifier (ADR-024 preview / TaskNotify support). The SQLite-backed task store is the durable record; this is the *edge-triggered* fast path so a TaskNotify caller doesn't have to poll. Lifetime = clawtool serve process. Subscriptions evaporate on restart — completed tasks remain queryable via TaskGet.

Package biam — WatchHub broadcasts task transitions AND live stream frames to in-process subscribers. The Unix-socket server (watchsocket.go) is the out-of-process consumer that lets `clawtool task watch`, `clawtool dashboard`, and `clawtool orchestrator` ditch SQLite polling.

Why a second hub alongside Notifier:

  • Notifier is a one-shot terminal-only push for TaskNotify / `clawtool send --wait`. It clears its subscriber list per task after a single Publish.
  • WatchHub fans EVERY transition (active, message_count++, terminal) AND every line the upstream agent emits as a StreamFrame to long-lived watchers. The orchestrator pane reconstructs a live stdout view from this.

Subscribers receive on a buffered channel (cap 64 for tasks, cap 256 for frames since stream lines are higher cadence). A slow subscriber drops events past the buffer rather than blocking the publisher — losing a transition is preferable to stalling every other watcher.

Package biam — Unix-socket task-watch server. The daemon runs ServeWatchSocket alongside its HTTP gateway; `clawtool task watch` dials the same socket and reads NDJSON Task events as they happen, eliminating the 250ms SQLite poll.

Wire format: one Task JSON per line, newline-terminated. The server emits a snapshot of every existing task on connect (so late joiners catch up without polling), then streams the live hub feed until the client disconnects or the daemon exits.

Permissions: socket file is mode 0600 — same security posture as the listener-token. The XDG_STATE_HOME path keeps it off the user's $HOME root.

Index

Constants

View Source 
const DefaultRingCap = 256

DefaultRingCap is the per-task event retention. 256 events covers the bursty-then-quiet shape of a typical dispatch (boot + dozens of frames + terminal) without making short reconnects walk a dense history.

Variables

View Source 
var ErrNoDispatchSocket = errors.New("biam dispatchsocket: socket not reachable — start `clawtool serve` first")

ErrNoDispatchSocket signals the CLI fallback path: no daemon is running. Callers can either error out with a "start the daemon" hint or fall back to the legacy in-process runner (with the caveat that frames won't reach the orchestrator).

View Source 
var (
	ErrNoWatchSocket = errors.New("biam watchsocket: socket not reachable")
)

Errors exposed for caller branching.

View Source 
var Events = NewTaskEventBuffer(DefaultRingCap)

Events is the process-wide singleton. The daemon installs publish hooks at boot; tests use NewTaskEventBuffer for isolation.

View Source 
var Notifier = &notifier{subs: map[string][]chan Task{}}

Notifier is the process-wide singleton. Tests use ResetForTest.

View Source 
var Watch = &WatchHub{
	subs:   map[*watchSub]struct{}{},
	frames: map[*frameSub]struct{}{},
	system: map[*systemSub]struct{}{},
}

Watch is the process-wide singleton. Tests use ResetWatchForTest.

Functions

func DefaultDispatchSocketPath added in v0.22.6 

func DefaultDispatchSocketPath() string

DefaultDispatchSocketPath sits beside DefaultWatchSocketPath in $XDG_STATE_HOME/clawtool/. Both sockets share the same lifecycle (daemon up = both bound; daemon down = both gone) so a CLI client either uses both or neither.

func DefaultIdentityPath 

func DefaultIdentityPath() string

DefaultIdentityPath honours XDG_CONFIG_HOME, falls back to HOME.

func DefaultStorePath 

func DefaultStorePath() string

DefaultStorePath honours XDG_DATA_HOME, falls back to HOME.

func DefaultWatchSocketPath added in v0.22.0 

func DefaultWatchSocketPath() string

DefaultWatchSocketPath honours XDG_STATE_HOME, falls back to ~/.local/state. Keeps the runtime socket out of $XDG_CONFIG_HOME (config = static) and $XDG_DATA_HOME (data = durable).

func DialWatchSocket added in v0.22.0 

func DialWatchSocket(path string) (net.Conn, error)

DialWatchSocket returns an open net.Conn to the daemon's task- watch socket. CLI-side helper. Empty path uses the default. Caller closes.

func ParsePublicKey 

func ParsePublicKey(s string) (ed25519.PublicKey, error)

ParsePublicKey decodes the `ed25519:<hex>` form back into a key.

func ServeDispatchSocket added in v0.22.6 

func ServeDispatchSocket(ctx context.Context, runner dispatchSubmitter, path string) error

ServeDispatchSocket binds the dispatch socket at `path`, accepting one request per connection until ctx cancels. `runner` is the daemon's process-wide BIAM runner — its goroutine lives in the daemon process, so frames it broadcasts via Watch.BroadcastFrame reach every WatchHub subscriber on the daemon (including orchestrator socket clients). Pass an empty path to use the default.

Auth: socket file mode 0600 + parent dir 0700. No bearer token — any process running as the same user can submit, mirroring the trust model of the watch socket.

func ServeWatchSocket added in v0.22.0 

func ServeWatchSocket(ctx context.Context, store *Store, hub *WatchHub, path string) error

ServeWatchSocket binds the Unix socket at `path`, accepting clients until ctx cancels. Each accepted connection gets:

  1. A backlog snapshot — every current task as a JSONL line, so a late watcher catches up without re-polling SQLite.
  2. A live tail subscribed to `hub` — every Broadcast becomes another JSONL line.

Returns when ctx is done OR the listener accept errors fatally. A nil hub falls back to the package singleton.

func Verify 

func Verify(pub ed25519.PublicKey, message, signature []byte) bool

Verify checks a signature against a peer's known public key.

func WirePushHooks added in v0.22.135 

func WirePushHooks(ctx context.Context, hub *WatchHub, events *TaskEventBuffer) (stop func())

WirePushHooks bridges hub's broadcasts into events. Returns a stop func that unsubscribes both bridges. Idempotent at the call-site (the daemon only calls it once); calling twice creates two bridges and produces duplicate events.

Bridges:

  • hub.Subscribe() → events.Append(taskID, "task" | "terminal", ...)
  • hub.SubscribeFrames() → events.Append(taskID, "frame", ...)

System notifications are NOT bridged — they're not task-scoped (no task_id), so they don't fit the SSE ?task_id=… contract. SSE clients that want daemon-level events poll a future /v1/biam/system endpoint instead.

Types

type Address 

type Address struct {
	HostID     string `json:"host_id"`
	InstanceID string `json:"instance_id"`
}

Address points at one peer instance. Format: `host_id/instance_id`.

func (Address) String 

func (a Address) String() string

type Body 

type Body struct {
	Text   string         `json:"text,omitempty"`
	Extras map[string]any `json:"extras,omitempty"`
}

Body is the per-message payload. `Text` is the agent-readable content; `Extras` carries opt-in structured data without forcing a schema bump.

type DispatchClient added in v0.22.6 

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

DispatchClient is the CLI-side handle for submitting a dispatch request to a running daemon. Single-use — Dial + Submit + Close. Caller is expected to defer Close.

func DialDispatchSocket added in v0.22.6 

func DialDispatchSocket(path string) (*DispatchClient, error)

DialDispatchSocket connects to the daemon's dispatch socket. Empty path uses the default. Returns ErrNoDispatchSocket when the socket is missing — useful for "is the daemon running?" detection in CLI flows that fall back gracefully.

func (*DispatchClient) Close added in v0.22.6 

func (c *DispatchClient) Close() error

Close releases the connection. Idempotent; safe to call after Submit (which already closes).

func (*DispatchClient) Submit added in v0.22.6 

func (c *DispatchClient) Submit(ctx context.Context, instance, prompt string, opts map[string]any) (string, error)

Submit sends one dispatch request and waits for the response. The connection is closed afterwards regardless of outcome.

type DispatchRequest added in v0.22.6 

type DispatchRequest struct {
	Action   string         `json:"action"` // "submit"
	Instance string         `json:"instance,omitempty"`
	Prompt   string         `json:"prompt,omitempty"`
	Opts     map[string]any `json:"opts,omitempty"`
}

DispatchRequest is the JSON-line wire request. `Action` is an enum so the protocol can grow (cancel, list, etc.) without breaking older clients — they ignore unknown actions and fall through to an error response.

type DispatchResponse added in v0.22.6 

type DispatchResponse struct {
	TaskID string `json:"task_id,omitempty"`
	Error  string `json:"error,omitempty"`
}

DispatchResponse is the JSON-line wire response. Exactly one of `TaskID` / `Error` is populated.

type Envelope 

type Envelope struct {
	Version        string       `json:"v"`
	TaskID         string       `json:"task_id"`
	MessageID      string       `json:"message_id"`
	ParentID       string       `json:"parent_id,omitempty"`
	CorrelationID  string       `json:"correlation_id,omitempty"`
	From           Address      `json:"from"`
	To             Address      `json:"to"`
	ReplyTo        Address      `json:"reply_to"`
	Kind           EnvelopeKind `json:"kind"`
	Body           Body         `json:"body"`
	HopCount       int          `json:"hop_count"`
	MaxHops        int          `json:"max_hops"`
	Trace          []string     `json:"trace"`
	CreatedAt      time.Time    `json:"created_at"`
	TTLSeconds     int64        `json:"ttl_seconds"`
	IdempotencyKey string       `json:"idempotency_key"`
	Signature      string       `json:"signature,omitempty"`
}

Envelope is the wire shape every BIAM message takes. Locked at `v: biam-v1` per ADR-015. Field rules in the ADR's "Wire envelope" section.

func NewEnvelope 

func NewEnvelope(from, to Address, taskID string, kind EnvelopeKind, body Body) *Envelope

NewEnvelope stamps the routine fields a fresh envelope needs and leaves the caller to set Body / ParentID / Kind. Trace seeds with the sender's address so cycle detection works on hop 1.

func (*Envelope) HasCycle 

func (e *Envelope) HasCycle(peer Address) bool

HasCycle reports whether `peer` already appears in the envelope's trace — a clean way to detect "this came back to me, drop it."

func (*Envelope) Hop 

func (e *Envelope) Hop(me Address) error

Hop bumps the hop count + appends `me` to the trace. Returns the fresh max-hops error when the cap is exceeded.

func (*Envelope) Sign 

func (e *Envelope) Sign(id *Identity) error

Sign computes the Ed25519 signature over the canonical JSON form (every field except Signature itself) and stores it on the envelope.

func (*Envelope) Verify 

func (e *Envelope) Verify(pub ed25519.PublicKey) error

Verify decodes the envelope's signature and checks it against the sender's known public key. Receivers must call this before trusting any field on the envelope.

type EnvelopeKind 

type EnvelopeKind string

EnvelopeKind enumerates what a message represents in a BIAM thread. 

const (
	KindPrompt        EnvelopeKind = "prompt"
	KindReply         EnvelopeKind = "reply"
	KindClarification EnvelopeKind = "clarification"
	KindResult        EnvelopeKind = "result"
	KindError         EnvelopeKind = "error"
	KindCancel        EnvelopeKind = "cancel"
)

type Identity 

type Identity struct {
	HostID     string
	InstanceID string
	Public     ed25519.PublicKey
	// contains filtered or unexported fields
}

Identity carries the Ed25519 keypair plus the human-friendly host / instance label every signed envelope's `from` field uses.

func LoadOrCreateIdentity 

func LoadOrCreateIdentity(path string) (*Identity, error)

LoadOrCreateIdentity reads the seed file at path; creates a new keypair on first launch. The host_id and instance_id default to the host's hostname + "default" when not set in the seed metadata.

First-launch creation is guarded by a sibling .lock file (flock): two clawtool processes starting in parallel must not race two keypairs into the same path, with the last-write winner stranding every envelope the loser had already signed. The lock is held only over the create-and-publish window — readers on a healthy file never touch it.

func (*Identity) PublicKeyB64 

func (i *Identity) PublicKeyB64() string

PublicKeyB64 returns the public key encoded as `ed25519:<hex>` — the format the peers.toml file stores.

func (*Identity) Sign 

func (i *Identity) Sign(message []byte) []byte

Sign produces the signature for the canonical-JSON envelope.

type Runner 

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

Runner glues the BIAM store to the supervisor's dispatch surface: async submissions land in the store as `prompt` envelopes; a goroutine drains the upstream stream and persists `result` (or `error`) envelopes; tasks transition through pending → active → done|failed.

func NewRunner 

func NewRunner(store *Store, id *Identity, send SendStream) *Runner

NewRunner wires the runner. Identity + store are mandatory; send is the supervisor's dispatch func.

func (*Runner) Cancel added in v0.21.5 

func (r *Runner) Cancel(ctx context.Context, taskID string) error

Cancel propagates a cancellation request to the dispatch goroutine for taskID. Idempotent: returns nil for unknown / already-terminal tasks. The actual upstream process kill happens in streamingProcess.Close on ctx.Done — the runner's responsibility here is just to flip the row and wake the goroutine.

func (*Runner) Stop added in v0.22.30 

func (r *Runner) Stop()

Stop cancels every in-flight dispatch and waits for the spawned goroutines to drain. Idempotent. Caller (daemon shutdown sequence) invokes this BEFORE closing the underlying *Store, so the store's last-second writes from terminating dispatches don't race the store's Close. The goroutines drop terminal envelopes via recordResult on cancel, so the durable state stays consistent.

func (*Runner) Submit 

func (r *Runner) Submit(ctx context.Context, instance, prompt string, opts map[string]any) (string, error)

Submit enqueues an async dispatch. Returns the new task_id immediately; the goroutine streams the response into the store and transitions the task on completion. Cancel via `Cancel(taskID)`.

`opts["from_instance"]` overrides the default `from` address. Cross- host bidi: when codex / gemini / opencode dispatch back to claude through the shared daemon, they pass their own family name so the resulting envelope's `from` reflects the actual sender, not the daemon's own identity. Without this, every BIAM thread looked like it originated from one centralised initiator and downstream reply-tracking ambiguated.

func (*Runner) WaitForTerminal 

func (r *Runner) WaitForTerminal(ctx context.Context, taskID string, poll time.Duration) (*Task, error)

WaitForTerminal proxies to the store with a default poll interval.

type SendStream 

type SendStream func(ctx context.Context, instance, prompt string, opts map[string]any) (io.ReadCloser, error)

SendStream is the function shape the runner expects from Supervisor: invoke `instance` with `prompt` + `opts`, return a streaming io.ReadCloser. Matches Supervisor.Send so we can swap in tests.

type Store 

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

Store wraps the per-instance SQLite file. Methods are safe for concurrent calls — the underlying connection pool serialises writes; readers fan out via WAL.

func OpenStore 

func OpenStore(path string) (*Store, error)

OpenStore opens (creating if absent) the SQLite database at path. WAL mode + busy-timeout makes concurrent writers tolerant.

func (*Store) Close 

func (s *Store) Close() error

Close flushes + closes the underlying database. Idempotent.

`s.db` mutation needs s.mu — every other store method dereferences `s.db` under the same lock (or via sql.DB's own pool concurrency). Without this, a Close racing an in-flight PutEnvelope / GetTask nil-derefs in the middle of teardown.

func (*Store) CreateTask 

func (s *Store) CreateTask(ctx context.Context, taskID, initiatedBy, agent string) error

CreateTask inserts a new task row and returns the row's task_id. Idempotent: an existing task_id returns nil error.

func (*Store) GetTask 

func (s *Store) GetTask(ctx context.Context, taskID string) (*Task, error)

GetTask returns the row for the given task_id, plus the message count via a sub-query so the caller doesn't need a second round trip.

func (*Store) ListTasks 

func (s *Store) ListTasks(ctx context.Context, limit int) ([]Task, error)

ListTasks returns the most-recent tasks (default limit 50, max 1000).

func (*Store) MessagesFor 

func (s *Store) MessagesFor(ctx context.Context, taskID string) ([]Envelope, error)

MessagesFor returns every envelope persisted under task_id, oldest first. Snapshot — does not subscribe.

func (*Store) PutEnvelope 

func (s *Store) PutEnvelope(ctx context.Context, env *Envelope, inbound bool) error

PutEnvelope inserts a message into the messages table. Inbound vs outbound is the caller's call. Dedupe via idempotency_key prevents double-inserts on retry.

func (*Store) ReapStaleTasks added in v0.22.2 

func (s *Store) ReapStaleTasks(ctx context.Context, pendingThreshold, activeThreshold time.Duration) (int, error)

ReapStaleTasks marks pending tasks older than `pendingThreshold` AND active tasks older than `activeThreshold` as `expired`. Returns the number of rows affected.

Why: a daemon crash leaves rows stuck in pending/active forever. Without recovery, `clawtool task list` accumulates ghost rows from every prior daemon process, and TaskNotify subscribers wait for a terminal state that will never come. Running this at daemon startup catches the orphans from the previous boot.

Threshold rationale:

  • pending → active is supposed to flip in milliseconds. A pending row older than ~1 minute is presumed orphan.
  • active rows stay active legitimately for as long as the upstream agent runs (codex deep-research can hit 10+ minutes). Pass 0 (or a very large duration) to skip the active sweep when you can't bound legitimate runtime.

Both thresholds zero = sweep every non-terminal row regardless of age. Not the default — only safe when the caller knows no other daemon shares this DB.

func (*Store) SetTaskCloseHook added in v0.22.107 

func (s *Store) SetTaskCloseHook(fn func(taskID string))

SetTaskCloseHook registers a callback fired EXCLUSIVELY when a task transitions into a terminal status (done / failed / cancelled / expired). Idempotent — pass nil to clear. Distinct from SetTaskHook (which fires on every state change) so one consumer can subscribe to terminal events (e.g. peer-lifecycle auto-close) without competing with another's broader subscription. The hook runs synchronously after the store mutex is released; safe to re-enter the store from inside.

Why a second hook seam instead of overloading taskHook: the daemon already binds taskHook to WatchHub.Broadcast for the live `task watch` stream. Auto-close needs a separate dispatch point because (a) it only cares about terminal transitions, (b) it shouldn't fire on every PutEnvelope churn, and (c) overloading the existing hook would mean a watchsocket replacement clobbers auto-close (last-writer-wins on a single-slot mechanism).

func (*Store) SetTaskHook added in v0.22.0 

func (s *Store) SetTaskHook(fn func(taskID string))

SetTaskHook registers a callback fired after every successful task state mutation (SetTaskStatus + PutEnvelope). Idempotent — pass nil to clear. The hook runs synchronously after the store mutex is released, so it can do its own DB reads without deadlocking. The daemon wires this to WatchHub.Broadcast so cross-process watchers (Unix socket) see live transitions instead of polling.

func (*Store) SetTaskStatus 

func (s *Store) SetTaskStatus(ctx context.Context, taskID string, status TaskStatus, lastMessage string) error

SetTaskStatus updates the task row + (when terminal) closed_at + last_message. Pass empty `lastMessage` to leave it untouched.

func (*Store) WaitForTerminal 

func (s *Store) WaitForTerminal(ctx context.Context, taskID string, poll time.Duration) (*Task, error)

WaitForTerminal polls (cheap) until the task reaches a terminal state or the context is cancelled. The caller usually wraps this in a timeout.

type StreamFrame added in v0.22.1 

type StreamFrame struct {
	TaskID string    `json:"task_id"`
	Agent  string    `json:"agent,omitempty"` // family-only, never instance label
	Line   string    `json:"line"`
	Kind   string    `json:"kind,omitempty"` // "stdout" (default) | "error" | "meta"
	TS     time.Time `json:"ts"`
}

StreamFrame is one line emitted by an upstream agent. The orchestrator pane appends frames to a per-task ringbuffer and renders them as live stdout. Frames carry the `kind` so the renderer can colour `error` or `meta` lines differently from regular output.

type Sub 

type Sub struct {
	Ch <-chan Task
	// contains filtered or unexported fields
}

Sub is a handle to one subscription. Cancel removes the channel from the subscriber list so a goroutine that bails out doesn't leak its slot until the next Publish.

func (*Sub) Cancel 

func (s *Sub) Cancel()

Cancel detaches this subscription. Safe to call after Publish has fired (no-op).

type SystemNotification added in v0.22.9 

type SystemNotification struct {
	Kind       string    `json:"kind"`     // taxonomy: "update_available" | "warning" | "info" | "error"
	Severity   string    `json:"severity"` // "info" (default) | "warning" | "error"
	Title      string    `json:"title"`
	Body       string    `json:"body,omitempty"`
	ActionHint string    `json:"action_hint,omitempty"` // e.g. "run: clawtool upgrade"
	TS         time.Time `json:"ts"`
}

SystemNotification is a daemon-level inline message broadcast to every connected watcher. Distinct from Task / StreamFrame because it isn't tied to a dispatch — examples: "clawtool update available v0.22.5 → v0.23.0", "sandbox-worker disconnected", "telemetry key rotation pending". Severity drives the renderer's colour pill; ActionHint is an optional one-line CLI suggestion the operator can copy-paste.

type Task 

type Task struct {
	TaskID       string     `json:"task_id"`
	Status       TaskStatus `json:"status"`
	InitiatedBy  string     `json:"initiated_by"` // who started it; empty for inbound
	Agent        string     `json:"agent"`        // agent instance the dispatch hit
	CreatedAt    time.Time  `json:"created_at"`
	ClosedAt     *time.Time `json:"closed_at,omitempty"`
	LastMessage  string     `json:"last_message,omitempty"` // tail of the latest result
	MessageCount int        `json:"message_count"`
}

Task is the BIAM-level row for a multi-message thread.

type TaskEvent added in v0.22.135 

type TaskEvent struct {
	ID   uint64    `json:"id"`
	Kind string    `json:"kind"` // "task" | "frame" | "system" | "terminal"
	Data []byte    `json:"data"` // pre-encoded JSON body
	TS   time.Time `json:"ts"`
}

TaskEvent is one entry in a per-task ring buffer. Kind is the SSE `event:` line; Data is the SSE `data:` payload (JSON-encoded by the publisher). ID is per-task monotonic.

type TaskEventBuffer added in v0.22.135 

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

TaskEventBuffer is the per-process registry of per-task rings.

func NewTaskEventBuffer added in v0.22.135 

func NewTaskEventBuffer(cap int) *TaskEventBuffer

NewTaskEventBuffer constructs a buffer with the given per-task capacity. Cap ≤ 0 falls back to DefaultRingCap.

func (*TaskEventBuffer) Append added in v0.22.135 

func (b *TaskEventBuffer) Append(taskID, kind string, data []byte) TaskEvent

Append records an event for taskID. Assigns the next monotonic ID, drops the oldest entry on overflow, fires a non-blocking notify on every subscriber. `kind == "terminal"` flips the ring's terminal flag so HasTerminal() returns true for late-joiners.

func (*TaskEventBuffer) HasTerminal added in v0.22.135 

func (b *TaskEventBuffer) HasTerminal(taskID string) bool

HasTerminal reports whether a terminal event has been appended for taskID. Lets the SSE handler short-circuit a connect to a task that already finished (replay history + close).

func (*TaskEventBuffer) ResetForTest added in v0.22.135 

func (b *TaskEventBuffer) ResetForTest()

ResetForTest wipes every per-task ring. Test-only.

func (*TaskEventBuffer) Since added in v0.22.135 

func (b *TaskEventBuffer) Since(taskID string, sinceID uint64) (events []TaskEvent, dropped bool)

Since returns every event for taskID with ID > sinceID, in chronological order. When sinceID is 0 the full retained ring is returned. dropped reports whether sinceID points at a position older than the oldest retained event (the client missed events past the cap; reconnect-with-snapshot is the only recovery).

func (*TaskEventBuffer) Subscribe added in v0.22.135 

func (b *TaskEventBuffer) Subscribe(taskID string) (<-chan struct{}, func())

Subscribe registers an edge-trigger notify channel for taskID. Returns the receive channel + an unsubscribe func. The channel receives one signal per Append; receivers MUST re-read the ring on wake. Buffer cap = 1 — coalesced; the publisher drops duplicate signals so a slow subscriber sees one wake per batch rather than N pending wakes.

func (*TaskEventBuffer) SubscriberCount added in v0.22.135 

func (b *TaskEventBuffer) SubscriberCount(taskID string) int

SubscriberCount is test-only.

type TaskStatus 

type TaskStatus string

TaskStatus enumerates the per-task lifecycle ADR-015 §"State machine" locks at v1. 

const (
	TaskPending   TaskStatus = "pending"
	TaskActive    TaskStatus = "active"
	TaskDone      TaskStatus = "done"
	TaskFailed    TaskStatus = "failed"
	TaskCancelled TaskStatus = "cancelled"
	TaskExpired   TaskStatus = "expired"
)

func (TaskStatus) IsTerminal 

func (s TaskStatus) IsTerminal() bool

IsTerminal reports whether a status closes the task.

type WatchEnvelope added in v0.22.1 

type WatchEnvelope struct {
	Kind   string              `json:"kind"`             // "task" | "frame" | "system"
	Task   *Task               `json:"task,omitempty"`   // populated when Kind=="task"
	Frame  *StreamFrame        `json:"frame,omitempty"`  // populated when Kind=="frame"
	System *SystemNotification `json:"system,omitempty"` // populated when Kind=="system"
}

WatchEnvelope is the JSONL wire-format wrapping every event the watch socket emits. `Kind` distinguishes "task" snapshots from "frame" stream lines so a single connection can multiplex both. CLI / TUI consumers branch on Kind. Older clients that pre-date the wrapping detect the new shape (top-level `kind` key) and upgrade their parser; nothing breaks if a Task lands in `Task` and `Frame` stays nil.

type WatchHub added in v0.22.0 

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

WatchHub is the multi-subscriber broadcaster. Lifetime = process.

func (*WatchHub) Broadcast added in v0.22.0 

func (h *WatchHub) Broadcast(t Task)

Broadcast fans the task snapshot to every subscriber. Non-blocking: a subscriber whose buffer is full drops this event silently. The store hook calls this after every state mutation.

The select-send runs INSIDE the lock — sounds backwards but is correct: the `default:` arm makes every send bounded-time (a full buffer falls through instantly), so holding the lock for the loop costs nothing, and crucially it closes the broadcast- then-close race. Pre-fix, a concurrent unsubscribe call could `close(sub.ch)` between our snapshot and our send → panic on send-to-closed-channel. Race detector wouldn't catch it (timing- bound). With the lock held, unsub blocks until the broadcast loop finishes, which is at most O(N) bounded operations.

func (*WatchHub) BroadcastFrame added in v0.22.1 

func (h *WatchHub) BroadcastFrame(f StreamFrame)

BroadcastFrame fans one StreamFrame to every frame subscriber. Non-blocking: a subscriber whose 256-cap buffer is full drops the event silently. The runner calls this after every line scanned from the upstream rc. Lock-during-send for the same race-closure reason documented on Broadcast.

func (*WatchHub) BroadcastSystem added in v0.22.9 

func (h *WatchHub) BroadcastSystem(s SystemNotification)

BroadcastSystem fans one SystemNotification to every system subscriber. Non-blocking — a slow watcher drops the event past the 16-cap buffer. The poller / sandbox-worker monitor / etc. call this when daemon-level state changes. Lock-during-send for the same race-closure reason documented on Broadcast.

func (*WatchHub) FrameSubsCount added in v0.22.1 

func (h *WatchHub) FrameSubsCount() int

FrameSubsCount is test-only.

func (*WatchHub) ResetWatchForTest added in v0.22.0 

func (h *WatchHub) ResetWatchForTest()

ResetWatchForTest wipes every subscriber. Test-only.

func (*WatchHub) SubsCount added in v0.22.0 

func (h *WatchHub) SubsCount() int

SubsCount is test-only — exposed so tests assert that unsubscribe actually frees the slot.

func (*WatchHub) Subscribe added in v0.22.0 

func (h *WatchHub) Subscribe() (<-chan Task, func())

Subscribe registers a buffered channel for every Broadcast. Returns the receive channel + an unsubscribe func. Callers MUST call unsubscribe to free the slot — usually via defer.

func (*WatchHub) SubscribeFrames added in v0.22.1 

func (h *WatchHub) SubscribeFrames() (<-chan StreamFrame, func())

SubscribeFrames registers a stream-frame subscriber. Higher buffer (256) than Subscribe — agents emit dozens of lines/second. Caller MUST unsub.

func (*WatchHub) SubscribeSystem added in v0.22.9 

func (h *WatchHub) SubscribeSystem() (<-chan SystemNotification, func())

SubscribeSystem registers a system-notification subscriber. Smaller buffer (16) than tasks/frames — system events are rare (handful per hour at most). Caller MUST unsub.

func (*WatchHub) SystemSubsCount added in v0.22.9 

func (h *WatchHub) SystemSubsCount() int

SystemSubsCount is test-only.

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.