biam

package
v0.22.8 Latest Latest
Warning

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

 Go to latest Published: Apr 28, 2026 License: MIT Imports: 22 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 — 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

This section is empty.

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 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{}{},
}

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.

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

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

func DecodeWatchEvent added in v0.22.0 

func DecodeWatchEvent(dec *json.Decoder) (*Task, error)

DecodeWatchEvent reads one JSONL Task line from r. EOF returns io.EOF without wrapping so callers can detect clean disconnect.

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"
	Task  *Task        `json:"task,omitempty"`  // populated when Kind=="task"
	Frame *StreamFrame `json:"frame,omitempty"` // populated when Kind=="frame"
}

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.

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.

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.

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.