workerkit

type AlertStore ¶

type AlertStore interface {
	SaveAlert(context.Context, Notice) error
	DeleteAlert(context.Context, uint64) error
	ListAlerts(context.Context) ([]Notice, error)
}

AlertStore is the optional durable extension for human-visible notices. Keeping it separate preserves the small StateStore contract for custom embedders, while the reference memory and SQLite implementations provide it so production alerts survive process restart and dismissal is durable.

type Author ¶

type Author string

Author identifies who produced a transcript entry.

const (
	AuthorCaller Author = "caller"
	AuthorHuman  Author = "human"
	AuthorSystem Author = "system"
)

type Change ¶

type Change struct {
	ID          string     `json:"id"`
	WorkspaceID string     `json:"workspace_id"`
	Path        string     `json:"path"`
	Kind        ChangeKind `json:"kind"`
	Diff        string     `json:"diff,omitempty"`
	Warning     string     `json:"warning,omitempty"`
}

Change is one reviewable difference between the human's mirror and the last delivered baseline. Diff is display text; the mirror owns the exact bytes.

type ChangeKind ¶

type ChangeKind string

ChangeKind classifies one reviewed workspace change.

const (
	ChangeCreate ChangeKind = "create"
	ChangeModify ChangeKind = "modify"
	ChangeDelete ChangeKind = "delete"
)

type Config ¶

type Config struct {
	Wire  Wire
	State StateStore
	// Mirror optionally enables the Live Workspace review/delivery surface.
	// Without it, DeliverChanges and DiscardChanges fail with ErrNoMirror.
	Mirror Mirror
	// AutoResponder optionally answers an assignment without human attention:
	// returning (text, true) confirms the assignment and delivers text as the
	// final event, and the item never reaches the inbox. Products use it for
	// caller housekeeping requests (e.g. OpenCode's tool-less title
	// generation). A failed auto-response falls back to the inbox.
	AutoResponder func(llm.WorkerAssignmentDelivery) (string, bool)
	// Clock supplies transcript timestamps. Nil uses time.Now.
	Clock func() time.Time
	// IDs allocates event and delivery identifiers. Nil uses crypto/rand. The
	// returned value must satisfy the transport's stable-key pattern.
	IDs func(kind string) (string, error)
	// Observer receives human-side lifecycle events. Nil is a no-op. Observer
	// failures are isolated by observe.Emit and never change command outcomes.
	Observer observe.Observer
}

Config composes one Worker. Wire and State are required borrowed dependencies: the host keeps ownership and releases them after Shutdown.

type Conversation ¶

type Conversation struct {
	Key         ConversationKey              `json:"key"`
	Phase       Phase                        `json:"phase"`
	Assignment  llm.WorkerAssignmentDelivery `json:"assignment"`
	Transcript  []TranscriptEntry            `json:"transcript"`
	ParkedCalls []llm.ToolCall               `json:"parked_calls,omitempty"`
	// Delivery is the in-flight reviewed batch, settled when every call of the
	// batch returns a successful result.
	Delivery *PendingDelivery `json:"delivery,omitempty"`
	Draft    string           `json:"draft,omitempty"`
	// HumanWorkspace is this conversation's Human-host working directory. It
	// never describes or exposes the Agent user's filesystem.
	HumanWorkspace *HumanWorkspace `json:"human_workspace,omitempty"`
	// CallerGone is an advisory, persisted operator-safety flag. It permits a
	// locally active conversation to be abandoned only after the transport has
	// identified that exact task's caller as detached. Awaiting-caller/results
	// conversations remain abandonable without it.
	CallerGone bool      `json:"caller_gone,omitempty"`
	UpdatedAt  time.Time `json:"updated_at"`
}

Conversation is the persisted record of one accepted task-scoped exchange. Assignment is the current (latest) assignment; events always bind to its identity and lease.

type ConversationKey ¶

type ConversationKey struct {
	Caller llm.CallerID `json:"caller"`
	TaskID llm.TaskID   `json:"task_id"`
}

ConversationKey is the durable identity of one accepted conversation. It is the completion task scope: every assignment of the same caller task resumes the same conversation.

type EntryKind ¶

type EntryKind string

EntryKind classifies one transcript entry.

const (
	EntryText          EntryKind = "text"
	EntryProgress      EntryKind = "progress"
	EntryClarification EntryKind = "clarification"
	EntryFinal         EntryKind = "final"
	EntryToolCalls     EntryKind = "tool_calls"
	EntryToolResult    EntryKind = "tool_result"
	EntryRejected      EntryKind = "rejected"
)

type HumanWorkspace ¶

type HumanWorkspace struct {
	ID        string `json:"id"`
	Path      string `json:"path"`
	Available bool   `json:"available"`
}

HumanWorkspace is one session's Human-owned working directory. ID is safe to expose in reviews and persists across restarts; Path is meaningful only on the Human host.

type InboxItem ¶

type InboxItem struct {
	Delivery llm.WorkerDeliveryID
	Key      ConversationKey
	Tier     llm.CapabilityTier
	Preview  string
	// ToolCount is the number of caller-declared tools. Zero usually marks an
	// auxiliary request (e.g. OpenCode title/summary generation) rather than
	// the main conversation turn.
	ToolCount  int
	ReceivedAt time.Time
}

InboxItem is one assignment awaiting an accept/reject decision. It is not persisted by workerkit; the transport replays unconfirmed assignments.

type MemoryStateStore ¶

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

MemoryStateStore is the reference in-process StateStore. It is the semantic model for the conformance suite and the default for tests and ephemeral embeddings; it does not survive process restart.

func NewMemoryStateStore() (*MemoryStateStore, framework.ReleaseFunc)

func (store *MemoryStateStore) DeleteAlert(ctx context.Context, seq uint64) error

func (store *MemoryStateStore) DeleteConversation(ctx context.Context, key ConversationKey) error

func (store *MemoryStateStore) ListAlerts(ctx context.Context) ([]Notice, error)

func (store *MemoryStateStore) ListConversations(ctx context.Context) ([]Conversation, error)

func (store *MemoryStateStore) SaveAlert(ctx context.Context, notice Notice) error

func (store *MemoryStateStore) SaveConversation(ctx context.Context, conversation Conversation) error

type Mirror ¶

type Mirror interface {
	Reviews() <-chan Review
	Resolve(context.Context, MirrorResolve) ([]llm.ToolCall, error)
	Cancel(context.Context, []string) error
	Settle(context.Context, MirrorSettlement) error
}

Mirror is the Live Workspace port. The mirror owns the human's working copy, change detection, and the byte-exact content behind each change; workerkit owns when changes are delivered and settled:

• Reviews delivers complete replacement reviews (coalescing is the implementation's choice); workerkit keeps only the latest.
• Resolve builds native tool calls for selected changes and freezes their exact bytes for the eventual baseline advance, hiding them from review while in flight. If the delivery is never sent, workerkit calls Cancel to return them to review; a crash before the event is durably owned by the transport leaves them for the mirror to reseed into a later review.
• Cancel returns resolved-but-unsent changes to review without advancing the baseline. It is idempotent and a no-op for unknown ids.
• Settle(MirrorDelivered) is called only after the caller returned successful results for every call of the batch — never at send time — so the baseline advances exactly with the caller's working tree. Settle(MirrorDiscarded) drops changes without delivery.

A change that was resolved but never settled must reappear in a later review after restart; the mirror, not workerkit, is the durability owner of review state.

type MirrorOutcome ¶

type MirrorOutcome string

MirrorOutcome is the terminal fate of a reviewed change set.

const (
	// MirrorDelivered means the caller agent returned successful tool results
	// for every call built from these changes; the mirror advances its baseline.
	MirrorDelivered MirrorOutcome = "delivered"
	// MirrorDiscarded means the human dropped the changes without delivery.
	MirrorDiscarded MirrorOutcome = "discarded"
)

type MirrorResolve ¶

type MirrorResolve struct {
	ChangeIDs []string
	Tools     []llm.Tool
	// Scope must match the scope bound to this Human mirror.
	Scope WorkspaceScope
	// Workspace identifies the Human-side session directory selected for this
	// conversation. Change paths passed to builders are relative to this
	// directory and therefore relative to the Agent user's logical project.
	Workspace HumanWorkspace
	// HarnessID and HarnessVersion select an exact, versioned native tool
	// contract. A stock multi-harness mirror must fail closed when no matching
	// builder is registered; tool names alone are not sufficient authority.
	HarnessID      string
	HarnessVersion string
}

MirrorResolve asks the mirror to project selected changes onto the caller's declared native tools. Tools are the caller-declared tools of the current assignment; the mirror (or its host-supplied builder) must only produce calls for tools in this list.

type MirrorSettlement ¶

type MirrorSettlement struct {
	ChangeIDs []string
	Outcome   MirrorOutcome
}

MirrorSettlement finalizes a resolved change set. Settle must be idempotent for an exact repeat.

type Notice ¶

type Notice struct {
	Seq       uint64       `json:"seq"`
	At        time.Time    `json:"at"`
	Code      string       `json:"code"`
	Message   string       `json:"message"`
	Caller    llm.CallerID `json:"caller,omitempty"`
	TaskID    llm.TaskID   `json:"task_id,omitempty"`
	RequestID string       `json:"request_id,omitempty"`
}

Notice is a human-visible alert (e.g. a quarantined outbox row whose reply will never be delivered). Seq is stable for dismissal.

func (notice Notice) Validate() error

type NoticeSource ¶

type NoticeSource interface {
	Notices() <-chan Notice
}

NoticeSource is an optional Wire extension surfacing transport-level alerts a human must see (for example a quarantined durable outbox row). workerkit consumes it when present and shows the notices in State.Alerts.

type PendingDelivery ¶

type PendingDelivery struct {
	ChangeIDs []string `json:"change_ids"`
	CallIDs   []string `json:"call_ids"`
}

PendingDelivery records one in-flight reviewed batch on a conversation: the tool calls sent and the change IDs to settle when every call succeeds.

type Phase ¶

type Phase string

Phase is the human-side lifecycle of one conversation. It mirrors, but does not replace, the core task state machine: the core remains the correctness authority.

const (
	// PhaseActive accepts Reply/Clarify/Final/SubmitToolCalls commands.
	PhaseActive Phase = "active"
	// PhaseAwaitingResults is parked on submitted tool calls; the matching
	// caller continuation resumes it.
	PhaseAwaitingResults Phase = "awaiting_results"
	// PhaseAwaitingCaller follows a clarification; the caller's next request on
	// the same task resumes it.
	PhaseAwaitingCaller Phase = "awaiting_caller"
	// PhaseTerminal follows a final; the conversation accepts no more events.
	PhaseTerminal Phase = "terminal"
)

type Rejection ¶

type Rejection struct {
	Delivery llm.WorkerEventDelivery
	Receipt  llm.WorkerEventReceipt
}

Rejection is the application-facing NACK: the exact event that was rejected together with the deterministic receipt.

type Review ¶

type Review struct {
	Generation uint64   `json:"generation"`
	Changes    []Change `json:"changes"`
}

Review is one complete, self-consistent view of every pending change. Each review replaces the previous one wholesale; Generation increases with every publication.

type SessionBinding ¶

type SessionBinding struct {
	Scope            WorkspaceScope
	HarnessID        string
	HarnessVersion   string
	HarnessSessionID string
}

SessionBinding is the portable identity from which a Human-side session workspace is prepared. It deliberately contains no Agent filesystem path.

type SessionMirror ¶

type SessionMirror interface {
	Mirror
	// PrepareSession binds the session to preferredPath. An empty path selects
	// the implementation's default beneath its Human base workspace. A
	// non-empty path is an explicit Human choice and must already exist.
	PrepareSession(context.Context, SessionBinding, string) (HumanWorkspace, error)
}

SessionMirror is the optional per-conversation workspace extension. The reference filesystem Mirror implements it; custom Mirrors may omit it when they do not expose a Human working directory.

type State ¶

type State struct {
	Inbox         []InboxItem
	Conversations []Conversation
	// Review is the latest complete Live Workspace review, or nil without a
	// configured Mirror (or before its first publication).
	Review *Review
	// Alerts are undismissed human-visible notices, oldest first.
	Alerts []Notice
}

State is a coherent snapshot for UIs. All values are deep copies; mutating them never affects the worker.

type StateStore ¶

type StateStore interface {
	SaveConversation(context.Context, Conversation) error
	DeleteConversation(context.Context, ConversationKey) error
	// ListConversations returns all conversations ordered by Key (caller, then
	// task) for deterministic recovery.
	ListConversations(context.Context) ([]Conversation, error)
}

StateStore persists accepted conversations across worker restarts. It is deliberately tiny: workerkit serializes all access (single writer), records are whole-value upserts, and the store never interprets their content.

Unaccepted inbox items are intentionally NOT stored here: an unconfirmed assignment is replayed by the transport after restart, so persisting it twice would create a second source of truth.

Implementations own independent copies of every value after a successful SaveConversation and return caller-owned copies from ListConversations. Deleting an absent key is a no-op. Implementations must not retry callbacks and must be safe for use from one goroutine at a time.

type TranscriptEntry ¶

type TranscriptEntry struct {
	At         time.Time      `json:"at"`
	Author     Author         `json:"author"`
	Kind       EntryKind      `json:"kind"`
	Text       string         `json:"text,omitempty"`
	ToolCalls  []llm.ToolCall `json:"tool_calls,omitempty"`
	ToolCallID string         `json:"tool_call_id,omitempty"`
	Code       string         `json:"code,omitempty"`
}

TranscriptEntry is display state, not correctness state: the durable outbox and the core's response events remain the correctness record.

type Wire ¶

type Wire interface {
	Assignments() <-chan llm.WorkerAssignmentDelivery
	Rejections() <-chan Rejection
	SendEvent(context.Context, llm.WorkerEventDelivery) error
	ConfirmAssignment(context.Context, llm.WorkerDeliveryID) error
	ConfirmRejection(context.Context, llm.WorkerDeliveryID) error
	Done() <-chan struct{}
	Err() error
}

Wire is the transport port between workerkit and a HumanLLM deployment.

The transport, not workerkit, owns delivery durability: SendEvent returns only after the event is durably owned by the transport (a remote client's Journal/outbox, or a synchronous in-process commit), and the transport replays unconfirmed assignments and unconfirmed rejections after a process restart. workerkit therefore never persists its own copy of in-flight deliveries; duplicating that state would create a second source of truth.

Assignments and Rejections have workerkit as their only receiver. ACKs are silent; failures surface either as a SendEvent error (nothing was accepted) or as an asynchronous Rejection (a deterministic NACK).

func WrapClient(client *workerws.Client) Wire

func WrapConnection(connection llm.WorkerConnection) Wire

type Worker ¶

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

Worker is the headless human-side domain object. All methods are safe for concurrent use; commands are serialized internally.

func Open(ctx context.Context, config Config) (*Worker, error)

func (worker *Worker) Abandon(ctx context.Context, key ConversationKey) error

func (worker *Worker) Accept(ctx context.Context, delivery llm.WorkerDeliveryID) (ConversationKey, error)

func (worker *Worker) Clarify(ctx context.Context, key ConversationKey, text string) error

func (worker *Worker) DeliverChanges(ctx context.Context, key ConversationKey, changeIDs []string) error

func (worker *Worker) DiscardChanges(ctx context.Context, changeIDs []string) error

func (worker *Worker) DismissAlert(seq uint64)

func (worker *Worker) DismissAlertContext(ctx context.Context, seq uint64) error

func (worker *Worker) Final(ctx context.Context, key ConversationKey, text string) error

func (worker *Worker) Notifications() <-chan struct{}

func (worker *Worker) Reject(ctx context.Context, delivery llm.WorkerDeliveryID, reason string) error

func (worker *Worker) Reply(ctx context.Context, key ConversationKey, text string) error

func (worker *Worker) SaveDraft(ctx context.Context, key ConversationKey, draft string) error

func (worker *Worker) SetHumanWorkspace(
	ctx context.Context,
	key ConversationKey,
	path string,
) (HumanWorkspace, error)

func (worker *Worker) Shutdown(ctx context.Context) error

func (worker *Worker) Snapshot() State

func (worker *Worker) SubmitToolCalls(ctx context.Context, key ConversationKey, calls []llm.ToolCall) error

type WorkspaceScope ¶

type WorkspaceScope struct {
	Caller       llm.CallerID `json:"caller"`
	WorkspaceKey string       `json:"workspace_key"`
}

WorkspaceScope binds one Human-owned mirror to one authenticated Agent-user workspace. It is deliberately opaque: it routes changes but does not imply that the Human machine can resolve or mount the caller's directory.

func (scope WorkspaceScope) Validate() error