biam

package
v0.21.4 Latest Latest
Warning

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

 Go to latest Published: Apr 27, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

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.

Index

Constants

This section is empty.

Variables

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

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

Functions

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 ParsePublicKey 

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

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

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

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

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.

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.