Help us shape the future of Go! Take the 2026 Go Developer Survey

events

package
v0.5.1 Latest Latest
Warning

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

 Go to latest Published: Oct 4, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Overview

Package events implements the append-only session event log — the single source of truth for session state — plus its live fan-out: per-session seq allocation, list queries, a Postgres LISTEN/NOTIFY broker for SSE subscribers, ephemeral event_start/event_delta preview frames, and the span.* events emitted from the same instrumentation point as OTel spans.

Index

Constants

View Source 
const (

	// ChannelReapKick is the executor's, not the broker's: the reaper holds
	// its own LISTEN outside the pool, so the name is exported for it rather
	// than dispatched here (plan 48).
	ChannelReapKick = "map_sandbox_reap"
)

Notification channels shared by every control-plane replica. Payloads carry the subscription key — the session id, or for work items the environment id (and for frames the frame itself); LISTEN is server-wide, so one listening connection per process serves every subscriber.

View Source 
const InterruptResultText = "Tool execution was interrupted before completion. Please retry."

InterruptResultText is what an abandoned call is answered with, copied verbatim from the reference: a 2026-09-02 recording interrupted a live bash call and the reference answered it with this exact sentence (issue #78). Never an empty text block: a Messages endpoint rejects one, and that request is what every later replay of this session sends.

The recording covers the agent.tool_use family only. The custom-tool and MCP families reuse this string because one constant is what keeps the three arms from drifting; that reuse is ours, not observed.

Note what the second sentence does that the wording it replaced did not: it tells the model to retry the very call the user just abandoned, and being on the append-only log it says so again in every later replay of the session. It is kept anyway, because the reference chose it and matching the reference is this constant's whole job — a client that reads the text at all reads the reference's. Whether the model actually retries is a prompt-level question the reference answers for itself and this platform is not free to second-guess here; if it ever becomes a problem, the fix belongs in how the brain frames the interrupted turn, not in diverging on the recorded string.

View Source 
const InterruptedExplanation = "The outcome was interrupted by a user.interrupt before evaluation completed."

InterruptedExplanation is the verdict text an interrupt writes into the outcome entry and its terminal end event (ours, INFERRED — the reference documents the interrupted result, not its explanation).

View Source 
const MetricApprovalWait = "approval.wait.duration"

MetricApprovalWait is the human-in-the-loop approval latency histogram: how long a session sat suspended on requires_action before a confirmation cleared the gate. Exported so the telemetry contract test can assert this exact name reaches an OTLP collector.

View Source 
const MetricCacheTokenUsage = "model.cache.token.usage"

MetricCacheTokenUsage is the platform-native instrument recording prompt cache tokens split by cache operation. gen_ai.client.token.usage folds these into its input reading because the convention's gen_ai.token.type has no cache bucket; a long-horizon agent's cache hit rate is worth seeing on its own, so this metric carries the breakdown the convention cannot. It is exported so the telemetry contract test can assert this exact name reaches an OTLP collector.

View Source 
const MetricSessionStatus = "session.status.transitions"

MetricSessionStatus counts session status transitions, keyed by the status a session moved into (session.status attribute). It is exported so the telemetry contract test can assert this exact name reaches an OTLP collector.

Variables

View Source 
var (
	ErrSessionNotFound = errors.New("session not found")
	ErrSessionArchived = errors.New("session is archived")
)

Sentinel errors the API layer maps onto wire error envelopes.

View Source 
var ErrEnvironmentCredentialRequired = errors.New("`user.tool_result` may only be sent with environment credentials " +
	"(the self-hosted runner's Session-Instance JWT); " +
	"an API key or Console session cannot post this event type")

ErrEnvironmentCredentialRequired refuses a user.tool_result sent under any credential but a worker's; NormalizeInbound wraps it with the event's index, and the API answers it 403 permission_error, as the reference does — in its words, the index prefix included (2026-09-02 batch2 `sessW.send.user.tool_result.console-auth`, #540). The credential it names is the reference's: here a worker's environment key or its sessions token is what posts one.

View Source 
var RequestInputTypes = func() []string {
	out := make([]string, 0, len(domain.ConsumedInputs)+1)
	for _, t := range domain.ConsumedInputs {
		out = append(out, string(t))
	}
	return append(out, string(domain.EventSystemMessage))
}()

RequestInputTypes are the rows a model request's start stamps processed (#793; AppendOptions.Consume): domain.ConsumedInputs, and a system.message. That one extra is the only input a request reads that is not a consumed input: the request folds it into its system prompt wherever it sits, so no window holds it and replay never moves it, but it is read, and the start that reads it stamps it. An answer — a confirmation or a tool result — is stamped where its thread's ordered processor takes it, and an interrupt is no input a request reads.

Functions

func ActiveOutcome added in v0.2.0 

func ActiveOutcome(evals []domain.OutcomeEvaluation) (domain.OutcomeEvaluation, bool)

ActiveOutcome returns the first non-terminal outcome entry, if any. The reference allows one active outcome at a time, so first is only.

func Answered added in v0.4.0 

func Answered(ctx context.Context, q Querier, sessionID domain.ID, useID domain.ID) (bool, error)

Answered reports whether a result already references the tool use — the drivers' answered-means-cancelled check (decision 9): a thread-scoped interrupt answers the thread's calls itself, so a call found answered just before it runs is skipped, and one answered mid-run is cancelled.

func AnsweredSet added in v0.4.0 

func AnsweredSet(ctx context.Context, q Querier, sessionID domain.ID, ids []domain.ID) (map[domain.ID]bool, error)

AnsweredSet is Answered over many ids in one query: the subset of ids that some result already answers.

func BeginOutcomeWork added in v0.5.0 

func BeginOutcomeWork(ctx context.Context, tx pgx.Tx, sessionID domain.ID) error

BeginOutcomeWork flips every pending outcome entry to running, inside the caller's transaction and under the session row lock it holds: the primary's model request start calls it, because that request is the one that reads each pending entry's user.define_outcome — the entry and its event commit together, so every entry pending at the start has its event below it (#793). Entry state only, as no wire event exists for the flip.

One guarded statement rather than the read-modify-write MutateOutcomes runs: it runs on every primary start, and on all but the one that begins an outcome there is nothing pending, so the guard leaves the row — updated_at included — untouched, with nothing decoded.

func BusyChild added in v0.4.0 

func BusyChild(ctx context.Context, q Querier, sessionID, except domain.ID) (bool, error)

BusyChild reports whether the session has such a child. except is left out of the count when non-empty (no id is empty, so "" counts them all).

func CheckWhileAwaiting added in v0.5.0 

func CheckWhileAwaiting(awaited []AwaitedResponse, posted []NewEvent, reachesPrimary func(i int) bool) error

CheckWhileAwaiting refuses a send whose user.message or user.define_outcome the primary thread could not read because it would still await a response once the send's own answers are in. Its caller applies it to a primary resting idle on those responses, the state the reference was recorded refusing in, where this platform used to queue the input behind the call. It returns a *WhileAwaitingError naming the first such event and what the primary would still await (ThreadWaits, less the send's answers). The answers count wherever the send carries them, a message posted ahead of the answer that frees the primary included, since the send is settled as one: a result or a denial answers its call, an allow confirms it — a custom or a worker call then awaiting its result still — and an interrupt reaching the primary, reachesPrimary(i), answers every call. A system.message is never refused itself: it may follow only a message or a result (NormalizeInbound), and a refused message ahead of it refuses the send first.

func ClearedApprovalWait added in v0.5.0 

func ClearedApprovalWait(ctx context.Context, q Querier, sid, tid domain.ID) (*float64, error)

ClearedApprovalWait measures a gate that was pending before the transaction's processing step. The caller records the returned reading only after commit.

func ConsumedInput added in v0.5.0 

func ConsumedInput(t domain.EventType) bool

ConsumedInput reports whether t is an input a model request consumes (domain.ConsumedInputs), and so one a consumption window can hold. Nothing else moves — a system.message goes to the system slot wherever it sits, an interrupt or a confirmation is not conversation, no tool result can land inside a window (its call is not committed until the request ends), and a grader verdict is written between requests.

func ConsumptionOrder added in v0.5.0 

func ConsumptionOrder(dst, history []domain.Event) []domain.Event

ConsumptionOrder appends history, a log in seq order, to dst in consumption order and returns the result. dst may be history[:0], which reorders history in place without a copy: Push never appends a row before it has read it, so the writes stay at or behind the reads. A caller that still reads history by position afterwards (a watermark, an outcome's start) passes a slice of its own instead and keeps history in seq order.

func DeniedRefs added in v0.5.0 

func DeniedRefs(evs []NewEvent) []string

DeniedRefs are the calls the confirmations among evs refuse. A denial answers its call, so an interrupt later in the same send leaves the call to the denial's result rather than synthesizing its own.

func FlipNonTerminalOutcomes added in v0.2.0 

func FlipNonTerminalOutcomes(now time.Time) func([]domain.OutcomeEvaluation) ([]domain.OutcomeEvaluation, error)

FlipNonTerminalOutcomes is the MutateOutcomes half of an interrupt: every non-terminal entry goes terminal as interrupted, completed_at stamped now.

func HasRunnableMCPToolUse added in v0.4.0 

func HasRunnableMCPToolUse(ctx context.Context, q Querier, sessionID domain.ID, extraRefs, extraAllowed []string) (bool, error)

HasRunnableMCPToolUse reports whether any MCP call still lacks a result. Every settlement asks it first, because the answer decides a work kind no other driver can serve: only the platform's MCP driver answers an agent.mcp_tool_use — a client may post neither the call nor its result, and a BYOC worker's contract has no MCP surface at all — so a session left with one outstanding and no mcp_exec queued waits forever. Asked over the runnable set (plan 35 decision 5): a gated MCP call still awaiting its human is nothing for the driver to run. extraAllowed are the ids a batch's validated-but-not-yet-inserted allow confirmations release.

func HasUnansweredPlatformToolUse 

func HasUnansweredPlatformToolUse(ctx context.Context, q Querier, sessionID domain.ID, extraRefs []string) (bool, error)

HasUnansweredPlatformToolUse reports whether any platform-executed built-in tool use (agent.tool_use) still lacks a result. The executor runs only these, so a confirmation resume enqueues a tool_exec only when one is outstanding: a turn whose remaining unanswered tools are all client-executed (custom) has no platform work and waits on the client's result instead — enqueuing a tool_exec there would provision a sandbox for nothing.

func HasUnansweredThreadToolUse added in v0.4.0 

func HasUnansweredThreadToolUse(ctx context.Context, q Querier, sessionID, threadID domain.ID, extraRefs []string) (bool, error)

HasUnansweredThreadToolUse is HasUnansweredToolUse over one thread's own rows (empty threadID = the primary's): whether that thread's turn may resume, which a sibling's outstanding call does not decide.

func HasUnansweredToolUse 

func HasUnansweredToolUse(ctx context.Context, q Querier, sessionID domain.ID, extraRefs []string) (bool, error)

HasUnansweredToolUse reports whether any tool-use event in the session still lacks a matching result. extraRefs are treated as answered: the ids referenced by results that are validated but not yet inserted, so the API trigger can decide its batch before appending it.

func LatestOutcomeStart added in v0.5.0 

func LatestOutcomeStart(evs []domain.Event, outcomeID domain.ID) domain.Event

LatestOutcomeStart returns the most recent span.outcome_evaluation_start for outcomeID among evs, or the zero event when there is none — the start the cycle's end event references, and the row whose seq bounds what the cycle grades. A reclaimed cycle re-grades under a fresh start, so the latest is the live one; earlier dangling starts are the recorded crash-window residue.

func NewOutcomeEntry added in v0.2.0 

func NewOutcomeEntry(d DefineOutcome) domain.OutcomeEvaluation

NewOutcomeEntry is the outcome_evaluations entry a freshly accepted define_outcome writes: born pending ("pending before the agent begins work"), completed_at null until a terminal result (ours, INFERRED).

func NotifyReapKick added in v0.4.0 

func NotifyReapKick(ctx context.Context, db Execer) error

NotifyReapKick wakes whichever executors are listening to sweep their owned sandboxes now rather than at their next interval — the producer half of the session-end kick (#354, plan 48). It carries no payload on purpose. The durable statement of what is owed is the row the ending transaction wrote — the deleted_sessions tombstone for a delete, archived_at for an archive — which every reaper re-reads and re-classifies under its session lock; a session id here would be a fact the consumer must not trust anyway, since the wake reaches executors that own nothing of that session. So this only ever says look again, never what to do, and losing it costs one interval rather than a sandbox.

Like NotifyWorkEnqueued it runs on the caller's db handle and is meant to ride the ending transaction: Postgres delivers a NOTIFY only on commit, so the wake cannot reach a reaper before the row it is owed to, and an ending that rolls back wakes nobody. Riding the commit is also what keeps the kick off the response path — a session's end publishes it as one more statement in a transaction it was already running, rather than as post-commit work with a budget, a failure to report, and a window where the commit lands and the process dies before the wake goes out.

func NotifyWorkEnqueued added in v0.2.0 

func NotifyWorkEnqueued(ctx context.Context, db Execer, envID domain.ID) error

NotifyWorkEnqueued fires the work-items wake for one environment — the producer half of the work API's long poll (#74), which subscribes to the broker keyed by the environment id. It runs on the caller's db handle: inside a transaction Postgres delivers the NOTIFY only on commit, so a subscriber never wakes for a row its re-poll cannot yet see. The payload is just a pointer, like the events channel's — the woken poll re-reads the queue.

func NullableThread added in v0.4.0 

func NullableThread(threadID domain.ID) *string

NullableThread binds a thread id as the thread_id column holds it: NULL for the primary (plan 35 decision 2) — the one convention every package that queries by thread shares.

func PendingApprovalThreads added in v0.5.0 

func PendingApprovalThreads(ctx context.Context, q Querier, sid domain.ID) ([]domain.ID, error)

PendingApprovalThreads snapshots gates before an executor settlement, which can process a confirmation received earlier behind another tool's result.

func PendingThreadApprovals added in v0.5.0 

func PendingThreadApprovals(ctx context.Context, q Querier, sid, tid domain.ID) (bool, error)

PendingThreadApprovals counts approvals not yet processed, independently of an external worker's result wait after authorization.

func PreviewTransition added in v0.4.0 

func PreviewTransition(ctx context.Context, q Querier, sessionID domain.ID, t ThreadTransition) (domain.SessionStatus, *domain.StopReason, error)

PreviewTransition is TransitionThread's fold without its writes: the session status (and idle stop reason) the transition would leave, for the settlement that decides before it moves — the outcome intercept fires only when a thread's end_turn is the session's quiescence (decision 15).

func Previewable 

func Previewable(t domain.EventType) bool

PreviewableTypes are the only event types the wire allows previews for.

func RecordApprovalWait added in v0.2.0 

func RecordApprovalWait(ctx context.Context, seconds float64)

RecordApprovalWait records one approval wait, in seconds. The interval is measured in the database (clock_timestamp() minus the requires_action idle event's created_at) so both ends read the same clock, and recorded here after the resuming transaction commits — a confirmation whose commit rolled back did not resume anything.

It resolves the meter per call rather than caching an instrument that would pin whichever MeterProvider was installed first, and never fails the caller: a telemetry error just drops the reading.

func RecordSessionStatus added in v0.2.0 

func RecordSessionStatus(ctx context.Context, status domain.SessionStatus)

RecordSessionStatus counts one session status transition. Callers invoke it AFTER the transaction that moved sessions.status has committed, never before: a status change that rolled back — a lost lease, an aborted settle — did not happen, and counting the attempt would inflate the metric on exactly the infra churn an operator is trying to read. The status column is set when a session is created and moved in two places (TransitionThread's fold and AppendInTx's SetStatus), and committed in several, so the recording lives at each commit site rather than beside a write. A transaction that moves it more than once is counted by one of two rules: a brain settlement counts at most one move, to the status it leaves (a reclaim's rescheduling-and-running pair on a running session counts nothing); the API counts each move, in the order its transaction made them.

It resolves the meter per call rather than caching an instrument that would pin whichever MeterProvider was installed first, and never fails the caller: a telemetry error just drops the count.

func ResumableThreads added in v0.4.0 

func ResumableThreads(ctx context.Context, q Querier, sessionID domain.ID) ([]domain.ID, error)

ResumableThreads lists the session's running threads whose own calls are all answered — the threads a driver's pass has brought to the point of their next model_turn (decision 5: the executor wakes per thread as each thread's calls become answered). The primary is the empty id, and first. A thread's running status is the gate: settlement wakes an idle external-tool wait only after all its results are processed. A session from before the thread resource, with no rows, is its primary alone at the session's status.

func RouteInbound added in v0.4.0 

func RouteInbound(ctx context.Context, q Querier, sessionID domain.ID, evs []NewEvent) ([]bool, error)

RouteInbound resolves the thread each inbound event addresses (plan 35 decision 9), after the batch has passed ValidateToolResults and ValidateToolConfirmations (so every reference names a tool use in this session). A confirmation or a result is written on the thread of the tool use it answers — cross-posted when the call was, so the answer shows on the same surfaces — whatever thread the client named (answerClaim drops that claim). An interrupt's claim names the one thread it ends: it must be a live thread of this session (a *ThreadNotFoundError when it names none); the primary's own id means the primary. user.message, user.define_outcome and system.message carry no thread and address the primary (ThreadID stays empty). A child-scoped interrupt is cross-posted: the client sent it through the session, so the session view shows it with the thread named, the child's own view with null — as the answers to a cross-posted call are (INFERRED, docs/DIVERGENCES.md). On return each event's ThreadID and CrossPosted are what the append stores; the returned slice marks the interrupts that named a thread — what tells a thread-scoped interrupt (the primary's own id included) from a session-wide one once both carry an empty ThreadID.

func RubricSnapshotKey added in v0.2.0 

func RubricSnapshotKey(outcomeID domain.ID) string

RubricSnapshotKey is the outcome-owned blob key a file rubric's bytes are copied to at acceptance, so deleting the source file mid-outcome cannot break replay or grading.

func RunnablePlatformToolNames added in v0.4.0 

func RunnablePlatformToolNames(ctx context.Context, q Querier, sessionID domain.ID, extraRefs, extraAllowed []string) ([]string, error)

RunnablePlatformToolNames is UnansweredPlatformToolNames over the runnable set — what the exec enqueue sites route on, so an ask-gated sibling call never schedules (or re-arms) a driver that may not run it. extraAllowed as in HasRunnableMCPToolUse.

func StampConfirmations added in v0.5.0 

func StampConfirmations(ctx context.Context, tx pgx.Tx, sid domain.ID, ids []domain.ID) error

StampConfirmations stamps the confirmations a send consumes that its settlement's walk will not reach (AnswerPlan.Denied and AnswerPlan.Moot), as AdvanceThreadTools stamps the ones it does: in the settlement of the commit that processes them.

func StampInterrupts added in v0.5.0 

func StampInterrupts(evs []NewEvent)

StampInterrupts stamps the user.interrupt events among evs processed now: an interrupt is consumed on receipt, by the send that carries it, whatever it reaches, as the reference processes one (2026-09-02/batch2.json sessT.events.after-outcome idx 35-37), and nothing later would stamp it (#793). The send calls it once its interrupt arms have run, so no interrupt is stamped ahead of the results it synthesized.

func StampSupersededAnswers added in v0.5.0 

func StampSupersededAnswers(ctx context.Context, tx pgx.Tx, sid domain.ID, calls []ToolUseRef, at time.Time) error

StampSupersededAnswers stamps the answers still unprocessed for calls an interrupt, or a child's archive, answers (InterruptResults): answers the ordered walk held behind an earlier call of their thread, in an earlier send. The interrupt's results are processed as they are written, and threadCalls drops every call whose result is processed, so no later walk would reach them; the interrupt that supersedes them consumes them, in its own commit (#793). A call with any result on the log is not one an interrupt answers, so in practice what this finds is a confirmation. An answer posted in the interrupt's own send is not on the log yet: AnswerPlan places and stamps that one (Moot).

at is the stamp the calls' results were written with (InterruptResults), so each answer reads as processed with its call's result, on the clock that wrote it, and not on the database's, which can run behind it.

func ThreadWaits added in v0.5.0 

func ThreadWaits(ctx context.Context, q Querier, sid, tid domain.ID, kind string, platformOwned func(string) bool) (ToolFlow, []AwaitedResponse, error)

ThreadWaits reads a thread's calls once for a caller that already holds the session's environment kind, and returns both the flow ThreadToolFlow would (the thread's resting place) and the responses it awaits: ToolFlow.Pending, less what was received already. An answer queued behind an earlier call is not awaited, nor is a call whose received confirmation denies it, the denial being its answer — reachable while the denial waits behind an earlier call for the walk that writes its result.

func ToolConfirmationRefs 

func ToolConfirmationRefs(evs []NewEvent) []string

ToolConfirmationRefs collects the tool-use ids a batch's user.tool_confirmation events resolve, in batch order.

func ToolFlowThreads added in v0.5.0 

func ToolFlowThreads(ctx context.Context, q Querier, sid domain.ID) ([]domain.ID, error)

ToolFlowThreads selects threads whose tool state can advance. Idle end_turn and terminated threads are never revived by a shared executor settlement.

func ToolResultRefs 

func ToolResultRefs(evs []NewEvent) []string

ToolResultRefs collects the tool-use ids referenced by a batch's inbound tool-result events, in batch order.

func ToolWaitIDs added in v0.5.0 

func ToolWaitIDs(batch []NewEvent, kind string, platformOwned func(string) bool) []domain.ID

ToolWaitIDs advertises every external input the just-emitted turn will need, including calls beyond its current processing position.

func UnansweredPlatformToolNames added in v0.2.0 

func UnansweredPlatformToolNames(ctx context.Context, q Querier, sessionID domain.ID, extraRefs []string) ([]string, error)

UnansweredPlatformToolNames lists, in log order, the tool names of the platform built-in calls HasUnansweredPlatformToolUse counts. The name is what routes the work: web_fetch/web_search run in the executor's own process as web_exec while the other built-ins ride tool_exec to a sandbox (docs/plan/15_web-tools.md), so a resume trigger picks its work kind from this list. extraRefs are treated as answered, exactly as above.

func UnconfirmedAskEvents 

func UnconfirmedAskEvents(ctx context.Context, q Querier, sessionID domain.ID, extraConfirmed []string) ([]string, error)

UnconfirmedAskEvents returns, in log order, the ids of the session's ask-gated tool-use events that no user.tool_confirmation has resolved yet — the set a requires_action suspension is still blocked on. extraConfirmed are the ids a validated-but-not-yet-inserted confirmation batch resolves, so the API can decide its resume before appending: an empty result means every ask is answered and the session may run; a non-empty result is the remainder to re-emit on session.status_idle.

A gated call that already carries a result is not blocking either, however it got one. Only a user.interrupt produces that (it abandons everything outstanding without asking anyone), and the gate has to let go of it: an ask left "blocked" forever after its call was answered would wedge the session on the very resume the interrupt exists to restore.

func UnconfirmedThreadAskEvents added in v0.4.0 

func UnconfirmedThreadAskEvents(ctx context.Context, q Querier, sessionID, threadID domain.ID, extraConfirmed []string) ([]string, error)

UnconfirmedThreadAskEvents is UnconfirmedAskEvents over one thread's own rows (empty threadID = the primary's): the gate that holds that thread, which a sibling's ask does not.

func ValidateDefineOutcomes added in v0.2.0 

func ValidateDefineOutcomes(ctx context.Context, tx pgx.Tx, sessionID domain.ID, defs []DefineOutcome, batchInterrupts bool) error

ValidateDefineOutcomes enforces the DB-backed halves of accepting a batch's user.define_outcomes, defs as DefineOutcomes parsed them, under the send transaction's session row lock:

  • one active outcome at a time — the reference documents chaining only "after the terminal span.outcome_evaluation_end event of the previous outcome", so a batch with more than one, or one arriving while a stored entry is non-terminal, is rejected (the 400's shape is ours, INFERRED);
  • a file rubric's file_id must name a stored file within the org scope (v1's single-tenant boundary — the registry itself) whose size fits the rubric cap. The row is taken FOR SHARE so a concurrent DELETE /v1/files cannot remove the row and object between this check and the snapshot — the deleter blocks until this transaction commits. It is one row, a batch with two outcomes being refused first, so it owes no lock order of its own; a session create, which also holds the files it mounts, has taken it with them in id order already (internal/api's lockFileRows).

batchInterrupts reports a user.interrupt in the same batch: the interrupt settles the active outcome as `interrupted` in the same transaction — the documented way to chain outcomes in one send — so it clears the stored-entry half of the check.

func ValidateToolConfirmations 

func ValidateToolConfirmations(ctx context.Context, q Querier, sessionID domain.ID, evs []NewEvent) error

ValidateToolConfirmations rejects an inbound user.tool_confirmation that does not name a tool use still awaiting confirmation: the id must reference an ask-gated tool-use event (evaluated_permission "ask") in this session that no prior confirmation has resolved, and not appear twice in one request. Like a tool result, an accepted bad confirmation cannot be taken back from the append-only log, so a wrong reference is the client's 400. An already confirmed or already answered call keeps ours: neither is recorded.

func ValidateToolResults 

func ValidateToolResults(ctx context.Context, q Querier, sessionID domain.ID, evs []NewEvent, platformOwned func(name string) bool) error

ValidateToolResults rejects an inbound tool result that does not reference an outstanding tool call: the id must name an existing tool-use event of the matching kind with no result yet, in the log or earlier in the same batch. The log is append-only — one accepted bad reference would poison every future replay with a request the model protocol rejects, wedging the session permanently.

platformOwned, when non-nil, names the built-in tools only the platform may answer — the API injects the web tools and the six delegation tools, whose calls the brain answers inside its own settlement: a client result for such a call is rejected even while it is still unanswered, closing the double-answer window between the executor's web scan and its commit (#222, docs/plan/16_one-answer-per-tool-call.md). It must never cover the sandbox six — a self_hosted worker answering those via user.tool_result is the BYOC pull protocol.

Types

type AnswerPlan added in v0.5.0 

type AnswerPlan struct {
	// Pending are the posted answers the settlement leaves unprocessed.
	Pending map[domain.ID]bool
	// Steps are, per thread, the posted answers the settlement processes, in
	// the order it processes them.
	Steps map[domain.ID][]AnswerStep
	// Denied are the confirmations whose denial results Steps carries. The
	// send writes those results itself, so it stamps these itself too: once a
	// call's result is on the log, the settlement's walk no longer reaches it.
	Denied []domain.ID
	// Moot are the posted confirmations for calls an interrupt of the same
	// send answers. Each is consumed on receipt with nothing left to do, so it
	// is stamped where it was received, and the call keeps the one result the
	// interrupt wrote.
	Moot []domain.ID
}

AnswerPlan is what a send's own settlement will do with the send's answers, read before they are appended so the send can lay each out where it is processed (#793). It is built one thread at a time (Settle, Interrupted), as the send reaches each thread.

func NewAnswerPlan added in v0.5.0 

func NewAnswerPlan(posted []NewEvent) AnswerPlan

NewAnswerPlan starts the plan of a send whose events are posted: each of its user.tool_result, user.custom_tool_result and user.tool_confirmation events is pending until a walk of its thread processes it. An answer on a thread the send never walks — one neither settled nor interrupted — stays pending.

func (*AnswerPlan) Interrupted added in v0.5.0 

func (p *AnswerPlan) Interrupted(ctx context.Context, q Querier, sid, tid domain.ID, posted, synthesized []NewEvent) error

Interrupted plans the answers posted to thread tid, which an interrupt of the same send ends, once that interrupt's arm has run: synthesized are the results it wrote, stamped as they are written, so the calls they answer drop out of the walk as threadCalls drops any call whose result is processed, and a posted confirmation for such a call is Moot. The walk goes on over the calls the send answers itself — its posted results, and the denials received ahead of the interrupt. The thread's move is the interrupt's, so no flow is read, and a thread the send posts no answer to is not walked at all.

func (*AnswerPlan) Settle added in v0.5.0 

func (p *AnswerPlan) Settle(ctx context.Context, q Querier, sid, tid domain.ID, posted []NewEvent, platformOwned func(string) bool) (ToolFlow, error)

Settle plans the settlement of thread tid, which the send answers and does not interrupt: its calls are walked as AdvanceThreadTools will walk them, the posted answers read as though already on the log, and the flow the walk leaves the thread in is returned — the flow the settlement reads, and what SettleTransition moves the thread to. platformOwned marks the calls no worker runs.

type AnswerStep added in v0.5.0 

type AnswerStep struct {
	Answer  domain.ID
	Denials []NewEvent
}

AnswerStep is one posted answer the settlement processes, with the denial results its walk writes after it, before it reaches the next posted answer.

type AppendOptions 

type AppendOptions struct {
	// ThreadID is the thread the append settles for — the turn's thread, empty
	// for the primary (plan 35 decision 5): AddUsage folds into its row and
	// MarkProcessedThrough stamps its own events alone, never a sibling's
	// queued input.
	ThreadID domain.ID
	// SetStatus flips sessions.status alongside the append. The batch should
	// carry the matching session.status_* event; this option only moves the
	// resource column. TransitionThread, which derives the status from the
	// threads', writes the column itself — callers hand it the status it
	// moved to, so the self-committing wrappers count the transition.
	SetStatus *domain.SessionStatus
	// AddUsage folds one model turn's usage into sessions.usage and the
	// thread's.
	AddUsage *domain.ModelUsage
	// MarkProcessedThrough stamps processed_at on the thread's
	// still-unprocessed request inputs (RequestInputTypes) at seq <= the
	// watermark. Zero means no stamping. A request's start stamps what it
	// consumes (Consume); this is left for the one failure no request
	// follows, the brain's unrouted model.
	MarkProcessedThrough int64
	// Consume stamps the thread's still-unprocessed request inputs below this
	// batch's first row, the ones the model request that row opens consumes
	// (#793). The row's processed_at is taken from this process's clock once
	// the session row lock is held, and the inputs are stamped 1 µs before it,
	// as the reference stamps them. After the lock, so an input that committed
	// while this append waited on it is never dated before the start was free
	// to run; this process's clock, because the brain stamps the request's
	// agent.thinking, agent.message and span.model_request_end on it too, so
	// the request's inputs and outputs read in order whatever the database's
	// clock says. A stamp another process writes — created_at, from the
	// database, or an answer's processed_at, from whichever process took it —
	// can still skew against these by the difference between the two hosts'
	// clocks, as it could before #793. Only a span.model_request_start
	// leading the batch may set it; AppendInTx refuses any other batch.
	Consume bool
	// MutateOutcomes read-modify-writes sessions.outcome_evaluations under the
	// same row lock (the AddUsage pattern): the projection changes atomically
	// with the events that change it, so log and resource can never disagree.
	MutateOutcomes func([]domain.OutcomeEvaluation) ([]domain.OutcomeEvaluation, error)
	// Then runs inside the same transaction after the insert (work enqueue,
	// counters). An error aborts the whole append.
	Then func(ctx context.Context, tx pgx.Tx) error
}

AppendOptions are same-transaction side effects of an append: the session state machine's invariant is that the sessions row (status, usage) and the event log can never disagree, so both change under the one session row lock the append already holds.

type AwaitedResponse added in v0.5.0 

type AwaitedResponse struct {
	ID           domain.ID
	Confirmation bool
	Result       bool
}

AwaitedResponse is one tool call a thread waits on a client's or a worker's response to: a confirmation, for an ask-gated call nobody has confirmed, a result, for a custom call or a self_hosted worker's built-in nobody has answered, or both.

type Backend added in v0.2.0 

type Backend struct {
	// Provider is the OTel gen_ai.provider.name — "anthropic" or "openai",
	// which are the protocol values the registry routes on.
	Provider string
	// Model is the gen_ai.request.model: what the endpoint was asked for.
	Model string
}

Backend names the model backend one request goes to: the protocol its endpoint speaks and the model id sent upstream. It is telemetry's view of the route the provider registry resolved — never the credential, and never the provider itself, which would drag model backends into the event log.

type Broker 

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

Broker fans Postgres NOTIFY traffic out to in-process stream subscribers. One listening connection per process serves every subscriber (multi-replica control planes each run their own), and it is only held while subscribers exist: the listener starts with the first Subscribe and stops with the last Close, so idle processes and finished tests don't pin a connection.

func NewBroker 

func NewBroker(pool *pgxpool.Pool) *Broker

func (*Broker) Ready 

func (b *Broker) Ready(ctx context.Context) error

Ready blocks until the shared listener holds an active LISTEN. Subscribers that snapshot their starting log position after Ready returns cannot miss a wake for anything committed after the snapshot; any later coverage lapse re-wakes everyone on reconnect.

func (*Broker) Subscribe 

func (b *Broker) Subscribe(sessionID domain.ID) *Subscription

Subscribe registers for one key's live traffic — a session id for SSE streams, an environment id for the work API's long poll (work-items wakes) — starting the shared listener if it isn't running. The two key spaces cannot collide (env_ vs sesn_ prefixes). Callers must Close the subscription.

func (*Broker) SubscribeThread added in v0.4.0 

func (b *Broker) SubscribeThread(sessionID, threadID domain.ID) *Subscription

SubscribeThread is Subscribe for one child thread's stream: the same wake on every append to the session (the subscriber re-reads its own scope), and only that thread's preview frames. An empty threadID is the session stream.

type ConsumptionOrderer added in v0.5.0 

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

ConsumptionOrderer is the streaming form of ConsumptionOrder, for a reader that pages the log rather than holding it (RenderDream). Windows are keyed by the row's own thread, so a session-wide read never lets one thread's request hold another's input. The zero value is ready to use.

func (*ConsumptionOrderer) Flush added in v0.5.0 

func (o *ConsumptionOrderer) Flush(out []domain.Event) []domain.Event

Flush appends every held input to out, all threads merged by seq, and closes every window. At the end of a log it is what a request still in flight held. Called early, to bound memory (HeldBytes), it degrades the windows it closes to seq order: what was held comes out now, and the rest of each window is emitted as it arrives.

func (*ConsumptionOrderer) HeldBytes added in v0.5.0 

func (o *ConsumptionOrderer) HeldBytes() int

HeldBytes estimates what the inputs held back so far weigh: each one's body, and heldRowOverhead for the event it is held as.

func (*ConsumptionOrderer) Push added in v0.5.0 

func (o *ConsumptionOrderer) Push(out []domain.Event, ev domain.Event) []domain.Event

Push takes the next row in seq order and appends the rows to emit now to out, in consumption order, returning the extended slice. It is append's contract, so a caller that drains each result before the next push can hand the same buffer back (buf[:0]) and a row costs no allocation of its own:

  • a span.model_request_start, or the span.outcome_evaluation_start of the grader's call, releases its thread's held inputs ahead of itself, then opens a window. A start that never ended (a crash, a lost lease, an interrupted request) is closed this way: the retried call saw what the dead one held.
  • a consumed input on a thread with a window open is held, and so is one that lands on a thread still holding inputs after its window closed: the same next start consumes both, and they keep receipt order.
  • the end that names the open start (model_request_start_id, or outcome_evaluation_start_id) closes the window. An end naming another start leaves the window open; one whose payload cannot say which start it closes closes the open one, since a thread runs one call at a time. What the window held stays held: the next start releases it, right ahead of itself, which is where that request consumed it and where the reference stamps it, 1 µs before the start — after everything the in-flight request produced, its results included, whatever a delegated settle or an executor wrote around them (2026-09-02 batch2 sessT idx 78). With no next start, Flush releases it at the end.
  • everything else is emitted as it stands.

It never appends a row it has not been pushed, so the rows appended stay at or behind the rows pushed — what lets ConsumptionOrder reorder a log in place.

type Credential added in v0.5.0 

type Credential int

Credential is the class of credential a batch arrives under. It decides who may post a user.tool_result (#662), and nothing else. The zero value is every caller that is not a worker, so a sender nobody classified is refused rather than admitted. 

const (
	// ManagementCredential is an API key or a human: the zero value.
	ManagementCredential Credential = iota
	// EnvironmentCredential is a worker's environment key or its item's
	// sessions token.
	EnvironmentCredential
)

type DefineOutcome added in v0.2.0 

type DefineOutcome struct {
	OutcomeID     domain.ID
	Description   string
	MaxIterations int64
	RubricType    string // "text" | "file"
	RubricContent string // text rubric
	RubricFileID  string // file rubric
}

DefineOutcome is the normalized payload of a user.define_outcome event, as normalizeDefineOutcome stored it.

func DefineOutcomes added in v0.2.0 

func DefineOutcomes(evs []NewEvent) ([]DefineOutcome, error)

DefineOutcomes returns the batch's define_outcome payloads in order, parsed once for every check and write a send or a create makes of them. A payload that does not parse is refused (a *Refusal), as the client's batch, and a caller answers that refusal where it calls ValidateDefineOutcomes, ahead of that function's checks. The normalizer writes every payload it admits, so this is a backstop.

func FindDefineOutcome added in v0.2.0 

func FindDefineOutcome(history []domain.Event, outcomeID domain.ID) (DefineOutcome, bool)

FindDefineOutcome scans a session's history for the user.define_outcome event that minted outcomeID and returns its normalized payload.

func ParseDefineOutcome added in v0.2.0 

func ParseDefineOutcome(payload json.RawMessage) (DefineOutcome, error)

ParseDefineOutcome decodes a normalized user.define_outcome payload.

type Delivery added in v0.5.0 

type Delivery struct {
	// Received is the delivered row, on its target's thread
	// (Received.ThreadID; empty for the primary).
	Received NewEvent
	// Wake is the target's running pair — behind a session.status_running when
	// the session was idle — and empty when nothing woke it.
	Wake []NewEvent
	// Moved is the session status the wake moved the column to, nil when it
	// moved nothing: what the caller's post-commit metric counts.
	Moved *domain.SessionStatus
}

Delivery is one delivered row and the wake it caused.

func DeliverAndWake added in v0.5.0 

func DeliverAndWake(ctx context.Context, tx pgx.Tx, sessionID domain.ID, received NewEvent) (Delivery, error)

DeliverAndWake delivers received to the thread it names and wakes that thread when nothing else will move it (wakeThread's rule).

func DeliverThreadEnded added in v0.5.0 

func DeliverThreadEnded(ctx context.Context, tx pgx.Tx, sessionID, child domain.ID, notice NewEvent) (Delivery, error)

DeliverThreadEnded delivers a ThreadEnded notice about child to the primary and wakes it on wakeOnThreadEnded's rule. Like that rule, it runs before the ending thread's own transition.

func (Delivery) Events added in v0.5.0 

func (d Delivery) Events() []NewEvent

Events is the delivery in processing order: the wake, then the row the woken turn consumes.

func (Delivery) Woke added in v0.5.0 

func (d Delivery) Woke() bool

Woke reports whether the delivery woke its target, which the caller answers by enqueueing the target's turn.

type ExecClass added in v0.4.0 

type ExecClass int

ExecClass is the family of exec work the session's runnable calls still need, in the precedence every settlement shares: MCP first — only the platform's MCP driver answers an agent.mcp_tool_use, and a tool_exec is the one kind a BYOC worker claims, so it must never be handed a log with one outstanding; then web — a worker's toolset has no web tools; then the sandbox built-ins. The queue kind is the caller's to name (the events package does not know the queue). 

const (
	ExecNone ExecClass = iota
	ExecMCP
	ExecWeb
	ExecTool
)

func RunnableExecClass added in v0.4.0 

func RunnableExecClass(ctx context.Context, q Querier, sessionID domain.ID, answered, extraAllowed []string,
	isWebTool, isSettlementTool func(string) bool) (ExecClass, error)

RunnableExecClass reports which exec family the session's runnable calls call for next (plan 35 decision 5). Two predicates split the names, because this package knows none of them: isWebTool tells the web built-ins from the sandbox ones, and isSettlementTool the calls no driver may be armed for at all — the delegation tools the settlement answers in the commit that emits them (the callers inject toolset.IsWebTool and toolset.IsDelegationTool). answered are the ids the caller's batch already answers, extraAllowed the ones its allow confirmations release.

Excluding the settlement's own names here rather than at each call site is what keeps every driver and every re-arm agreeing about what is runnable: an unanswered delegation call — which nothing today can produce, every one being answered in its emitting commit — would otherwise class as ExecTool and arm a tool_exec whose pass filters it out and whose settlement re-arms the same kind, a spin loop rather than a wrong answer.

type Execer added in v0.2.0 

type Execer interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
}

Execer is the single pgx method the NOTIFY producers below need, satisfied by a pool and a transaction alike, so the NOTIFY can join the caller's commit.

type ListQuery 

type ListQuery struct {
	Types                                        []string
	CreatedGT, CreatedGTE, CreatedLT, CreatedLTE *time.Time
	AfterSeq                                     *int64
	// BeforeSeq keeps the rows below it (seq < BeforeSeq) whatever the sort:
	// the log as it stood under a known row — a request's own span start.
	BeforeSeq *int64
	Desc      bool
	Limit     int // 0 = unlimited
	// Scope narrows the rows to one surface (plan 35 decision 2). ScopeAll,
	// the zero value, reads the whole log — what every internal reader does.
	Scope    Scope
	ThreadID domain.ID // the thread ScopeThread reads; empty is the primary
	// ThreadToolCalls widens the session view with the child threads' tool
	// calls and the results answering them — the rows a BYOC worker must run
	// and answer, and the only ones it can ever see (plan 35 decision 13 i).
	// Read only under ScopeSession; the other scopes already carry them.
	ThreadToolCalls bool
	// HideTools names agent.tool_use calls a surface omits, together with the
	// agent.tool_result answering each, under any scope. The key is the call's
	// type and name alone, never its evaluated_permission or the answer's
	// is_error; which names, on which sessions, is the caller's to decide
	// (internal/api's eventsView, #675). The rows stay in the log, and a
	// thread's replay, which never sets this, reads them.
	HideTools []string
}

ListQuery narrows and pages a session's event log. AfterSeq is a keyset position (exclusive) in the direction of the sort; seq order and created_at order agree because appends serialize per session.

type Log 

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

Log is the append-only event store over the shared pool.

func NewLog 

func NewLog(pool *pgxpool.Pool) *Log

func (*Log) AdvanceThreadTools added in v0.5.0 

func (l *Log) AdvanceThreadTools(ctx context.Context, tx pgx.Tx, sid, tid domain.ID, platformOwned func(string) bool) (ToolFlow, error)

AdvanceThreadTools consumes the ready prefix while holding the session row lock. It never executes a tool. A processed allow releases execution, not the result wait; a denial produces its ordinary error result before advancing.

func (*Log) Append 

func (l *Log) Append(ctx context.Context, sessionID domain.ID, evs []NewEvent) ([]domain.Event, error)

Append durably appends events to one session's log in order, allocating the per-session seq under the session row lock (concurrent appends to the same session serialize; different sessions don't contend), and notifies stream subscribers on commit.

func (*Log) AppendInTx 

func (l *Log) AppendInTx(ctx context.Context, tx pgx.Tx, sessionID domain.ID, evs []NewEvent, opts AppendOptions) ([]domain.Event, error)

AppendInTx is AppendWith inside a caller-owned transaction, for callers that must decide the batch under the session row lock (the API's state machine reads the current status FOR UPDATE, builds the batch, and appends — all one commit). The NOTIFY still fires only on the caller's commit.

func (*Log) AppendTransition added in v0.4.0 

func (l *Log) AppendTransition(ctx context.Context, sessionID domain.ID, evs []NewEvent, transitions []ThreadTransition, opts AppendOptions) ([]domain.Event, error)

AppendTransition is the self-committing form for callers with no other decision to make under the lock (the brain's reclaim, test harnesses): one transaction that locks the session, runs the transitions in order, appends evs and what they emitted, with opts' side effects, and — after the commit, never before — counts the session's net move, so a reclaim's rescheduled+running pair on a running session counts nothing.

The two are placed in processing order (#793), keyed on whether a transition actually woke a thread — an unforced move to running of a thread that was not running. When one did, evs are the input the woken turn consumes when it starts, after everything this commit processes, so they follow every pair, as the API's message trigger writes them — which is what the brain's test harnesses mimic. When none did, evs are the facts the pairs report (a reply, then the idle it ends in) and precede them. The reclaim's forced rescheduled+running pair is not a wake (its thread was already running, and it consumes nothing), and it passes no evs, so its batch is the pairs alone either way.

func (*Log) AppendWith 

func (l *Log) AppendWith(ctx context.Context, sessionID domain.ID, evs []NewEvent, opts AppendOptions) ([]domain.Event, error)

AppendWith is Append plus atomic session-state side effects.

func (*Log) List 

func (l *Log) List(ctx context.Context, sessionID domain.ID, q ListQuery) ([]domain.Event, error)

List returns events for one session in seq order. It does not check that the session exists — callers that need 404 semantics check first.

func (*Log) PublishEventFrame 

func (l *Log) PublishEventFrame(ctx context.Context, sessionID domain.ID, event map[string]any) error

PublishEventFrame broadcasts a fully-rendered wire event object (e.g. the session.deleted event, whose row cannot outlive the session) to live stream subscribers without persisting it.

func (*Log) PublishThreadEventFrame added in v0.4.0 

func (l *Log) PublishThreadEventFrame(ctx context.Context, sessionID, threadID domain.ID, event map[string]any) error

PublishThreadEventFrame is PublishEventFrame onto one child thread's own stream (decision 2) — the session stream does not receive it; a caller that wants it on both surfaces publishes twice.

func (*Log) SettleToolFlow added in v0.5.0 

func (l *Log) SettleToolFlow(ctx context.Context, tx pgx.Tx, sid, tid domain.ID, flow ToolFlow) (*domain.SessionStatus, error)

SettleToolFlow emits only a changed status or changed blocker set, in its own append: the settlement of a thread whose tools a worker, the executor or the brain moved. A send settles the threads its answers move in its own batch instead, from the flow its plan read (AnswerPlan.Settle), where the answer is consumed (#793).

func (*Log) StartModelRequest 

func (l *Log) StartModelRequest(ctx context.Context, sessionID domain.ID, backend Backend) (context.Context, *ModelRequest, error)

StartModelRequest emits the span.model_request_start event and opens the matching OTel client span from one instrumentation point, so the wire events and the OTel trace can never drift (CLAUDE.md principle 3). Finish records the turn's metrics from the same point, for the same reason. The returned context carries the span for downstream propagation.

func (*Log) StartModelRequestOn added in v0.4.0 

func (l *Log) StartModelRequestOn(ctx context.Context, sessionID, threadID domain.ID, backend Backend, then func(context.Context, pgx.Tx) error) (context.Context, *ModelRequest, error)

StartModelRequestOn is StartModelRequest for one thread's turn (plan 35): the start and end events are written on that thread's own log; an empty threadID is the primary.

The start's commit also processes what the request consumes (#793): the thread's request inputs below it that no earlier start stamped are stamped 1 µs before the start's own processed_at, as the reference stamps a consumed input (AppendOptions.Consume). then runs in that commit — the brain passes its lease proof, so a claimant that lost its item neither stamps nor calls the model; nil is none.

func (*Log) StartOutcomeEvaluation added in v0.2.0 

func (l *Log) StartOutcomeEvaluation(ctx context.Context, sessionID domain.ID, outcomeID domain.ID, iteration int64, startID domain.ID, backend Backend) (context.Context, *OutcomeEvaluation)

StartOutcomeEvaluation opens the OTel span for one grading cycle from the same instrumentation point that owns its wire events (CLAUDE.md principle 3). Unlike StartModelRequest it appends no start event itself: the span.outcome_evaluation_start committed in the settlement transaction that scheduled this cycle (the two-phase design keeps the model call outside every session lock), so the caller passes that event's id in.

func (*Log) StartPreview 

func (l *Log) StartPreview(ctx context.Context, sessionID domain.ID, typ domain.EventType) (*Preview, error)

StartPreview broadcasts the event_start frame and pre-allocates the id the buffered event must later be appended under — on the primary thread's surface, the session stream.

func (*Log) StartPreviewOn added in v0.4.0 

func (l *Log) StartPreviewOn(ctx context.Context, sessionID, threadID domain.ID, typ domain.EventType) (*Preview, error)

StartPreviewOn is StartPreview on one thread's surface (plan 35 decision 2): a child's previews reach its own stream alone, where the event they preview will land; the session stream previews the primary's turns.

type ModelRequest 

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

ModelRequest is one in-flight model call being traced.

func (*ModelRequest) EndEvent 

func (m *ModelRequest) EndEvent(isError bool, usage domain.ModelUsage) (NewEvent, error)

EndEvent renders the span.model_request_end wire event for the caller to append — the turn's settlement commits it atomically with the rest of the turn's output, so an uncommitted turn leaves no half-told span on the log. Finish then closes the OTel side; both halves live on ModelRequest so the wire event and the OTel span still come from one instrumentation point (CLAUDE.md principle 3).

usage is what the wire event reports; a failing turn passes the zero value, because the schema wants a model_usage object whether or not a model ever produced one. It deliberately does not feed the token metric — that reads the usage ModelDone took from the stream itself. Settlement is the wrong place to learn what the model spent: it renders this event on some paths and not others, so sourcing the metric here would both invent zeroes for turns the model never costed and drop real spend for turns that never settle.

func (*ModelRequest) Finish 

func (m *ModelRequest) Finish(ctx context.Context, isError bool, commitErr error)

Finish closes the OTel span and records the turn's metrics. commitErr is the fate of the transaction that carried the EndEvent (or the reason no end event was attempted): non-nil records the drift explicitly, so the trace never masks an aborted request as a clean one.

func (*ModelRequest) ModelCalling added in v0.5.0 

func (m *ModelRequest) ModelCalling()

ModelCalling marks the start of the call to the model provider, where the model-latency clock starts. The caller invokes it just before the call, not at the span start: the brain reads the request's history and builds it after the start commits (#793), which is platform work, and the span's own start and its wire timestamps stay where the start committed. Repeat calls keep the first mark.

func (*ModelRequest) ModelDone added in v0.2.0 

func (m *ModelRequest) ModelDone(usage *domain.ModelUsage)

ModelDone records what the call to the model provider cost: how long it took, and the usage it reported (nil when it reported none, or when the stream failed before saying). The caller must invoke it as soon as the model's stream ends — before settling the turn.

Both are facts of the model's call, known exactly here, which is why they are taken here rather than at Finish. The span and the wire events deliberately stay open past this point: Finish runs after the settlement transaction so it can record whether the end event actually committed. But settlement is a session-locked Postgres transaction the model had nothing to do with, so measuring the duration to Finish would file database contention under a model-latency instrument — and only on the paths that reach settlement, since the abandon paths finish straight after the stream, leaving the metric inconsistent with itself. Usage sourced from settlement would be worse: a turn that streams a full answer and then loses its lease renders no end event at all, so tokens the model really spent and really billed would go unrecorded on exactly the paths that already cost money for nothing.

Repeat calls keep the first reading. The duration runs from ModelCalling; without that mark there was no call to time, and none is taken.

func (*ModelRequest) SetAttributes added in v0.2.0 

func (m *ModelRequest) SetAttributes(attrs ...attribute.KeyValue)

SetAttributes records extra attributes on the model_request span — per-turn facts known only after the span opens, such as Level-1 skill injection. Kept off the wire event: these are trace-only observability, not conversation state.

func (*ModelRequest) StartEventID 

func (m *ModelRequest) StartEventID() domain.ID

StartEventID is the id of the span.model_request_start event, which the end event references as model_request_start_id.

func (*ModelRequest) StartSeq added in v0.5.0 

func (m *ModelRequest) StartSeq() int64

StartSeq is the start event's seq. The request consumes every input of its thread below it that no earlier request did (#793), so it bounds the history the brain reads for the request once the start has committed.

type NewEvent 

type NewEvent struct {
	ID          domain.ID // optional; generated when empty (previews pre-allocate)
	Type        domain.EventType
	Payload     json.RawMessage
	ProcessedAt *time.Time // nil = queued, awaiting in-order processing
	// ThreadID is the emitting child thread; zero means the primary thread
	// (plan 35 decision 2). CrossPosted marks a child's event the session
	// view — the primary thread's own — also surfaces: a child's status
	// events, an ask-gated call and the answer to it. Stored once; each
	// surface filters (ListQuery.Scope).
	ThreadID    domain.ID
	CrossPosted bool
}

NewEvent is one event to append. Payload holds the normalized type-specific wire fields only — never id/type/processed_at, which live on the envelope.

func DenialResults added in v0.3.0 

func DenialResults(ctx context.Context, q Querier, sessionID domain.ID, evs []NewEvent) ([]NewEvent, []string, error)

DenialResults answers each tool call a batch's user.tool_confirmation events refuse, with an error result carrying the client's deny_message. The model protocol requires every tool_use answered before the turn resumes, so a denied call must have a result or the next replay is a request the model rejects. It also returns the ids it answered, which the caller passes on as already-answered to the queries that decide what work the resume schedules.

The result is written in the family of the call that was refused — the same mapping an interrupt answers under, for the same reason. The family is not in the confirmation, which names its call by tool_use_id whichever kind it is, so it is read from the log: one query for the whole batch, since a batch may deny a built-in and an MCP call at once and each needs its own shape.

Nothing is stamped processed_at here, unlike InterruptResults: every confirmable family is answered by an agent.* event, which the store stamps as it inserts. The white-box table test is what keeps that true.

A denied bash call's result — is_error, and the text denialText composes — is the reference's as recorded; any other tool's text is an inference that it names the call the same way, an MCP call by the name the model was offered it under (docs/DIVERGENCES.md).

func InterruptOutcomes added in v0.2.0 

func InterruptOutcomes(ctx context.Context, tx pgx.Tx, sessionID domain.ID) ([]NewEvent, bool, error)

InterruptOutcomes builds the terminal span.outcome_evaluation_end for every non-terminal outcome entry — the docs: an interrupt marks the result interrupted "even if evaluation hadn't started yet", with outcome_evaluation_start_id an empty string when no start fired. An evaluating entry is the exception: its cycle's start committed with the flip, so the end references it (the latest start is the live one; earlier dangling starts are crash-window residue). Returns the end events and whether any entry needs flipping (the caller composes the matching MutateOutcomes under the same lock).

func InterruptResults added in v0.2.0 

func InterruptResults(uses []ToolUseRef) ([]NewEvent, error)

InterruptResults answers each tool call a user.interrupt abandons — the set UnansweredToolUses returns — with an error result. The turn cannot end without them: the model protocol requires every tool_use answered before the conversation continues, so an abandoned call on the append-only log would make every future replay a request the model rejects — the wedge the interrupt exists to undo.

The results are stamped processed on the spot. The platform wrote them, and the one that lands under an inbound type would otherwise render with a null processed_at, indistinguishable from a client event still queued behind earlier ones, and would look to a settling brain like input to chain a turn on.

It lives here beside DenialResults, rather than beside the control plane's confirmation handling, because what both encode is event-shape knowledge this package already owns — which result type answers which use type, under which reference key — and one definition is what keeps them from drifting apart.

For the agent.tool_use family this is now recorded rather than inferred: the reference answers an abandoned call with an agent.tool_result carrying is_error true, exactly one text block, a tool_use_id naming the tool-use *event* id, and a populated processed_at — all four of which this already wrote. What stays inferred is the custom-tool and MCP arms, which no recording covers (docs/DIVERGENCES.md).

func NewOutcomeStartEvent added in v0.2.0 

func NewOutcomeStartEvent(outcomeID domain.ID, iteration int64) (NewEvent, error)

NewOutcomeStartEvent renders a span.outcome_evaluation_start with a pre-minted id, for the settlement that schedules a grading cycle to commit atomically with the entry flip.

func NormalizeInbound 

func NormalizeInbound(envKind string, cred Credential, raws []json.RawMessage) ([]NewEvent, error)

NormalizeInbound validates one send batch. envKind is the session's environment kind ("cloud" | "self_hosted") and cred the class of credential the batch arrived under; both gate user.tool_result, the credential first.

func NormalizeInitialEvents added in v0.5.0 

func NormalizeInitialEvents(envKind string, cred Credential, raws []json.RawMessage) ([]NewEvent, error)

NormalizeInitialEvents is NormalizeInbound for a session's or deployment's initial_events, refusing in this platform's words throughout (see verbatim).

func ThreadEnded added in v0.4.0 

func ThreadEnded(sessionID, child domain.ID, agentName, text string) (NewEvent, error)

ThreadEnded is the notice a coordinator gets about a child that stopped: the received half of a message nothing sent, on the primary's log (plan 35 decision 7 — an ending condition is delivered as text, like a report). The convention is a bracketed line naming the agent and what happened, then the next action open to the coordinator.

It builds the notice; waking is wakeOnThreadEnded's, and every ending pairs the two through DeliverThreadEnded. A session-wide interrupt appends no notice at all — it ends the primary in the same batch, so there is nobody left to tell.

func ThreadMessage added in v0.4.0 

func ThreadMessage(sessionID domain.ID, from, to ThreadPeer, text string) (sent, received NewEvent, err error)

ThreadMessage renders one message as its two events. A message an agent sent appends both; the platform's own notice about a thread that ended appends received alone, because nothing sent that one.

text is the caller's to sanitize and bound: it reaches jsonb, where a NUL faults the append, and the model supplies it on every path but the notices.

func TransitionThread added in v0.4.0 

func TransitionThread(ctx context.Context, tx pgx.Tx, sessionID domain.ID, t ThreadTransition) ([]NewEvent, *domain.SessionStatus, error)

TransitionThread moves one thread's status and folds the session's over its live threads', in the caller's transaction under the session row lock (the API's trigger, the brain's settlements and the thread archive all hold it). It writes the thread row (status, stop_reason, and transitioned on the thread's first real move) and sessions.status, and returns the events to append — the thread's own (a child's is cross-posted to the session view and names its agent; the primary's is completed with the session's agent name by AppendInTx) and, when the folded value changed or Force, the session's, the two ordered as the pair above — and the status the session column moved to, nil when it did not: what the caller's post-commit metric counts, so a re-idle or a reclaim pair never inflates it.

type NoPendingConfirmationError added in v0.5.0 

type NoPendingConfirmationError struct {
	Index     int
	ToolUseID string
	Reason    string
}

NoPendingConfirmationError is ValidateToolConfirmations' refusal of a confirmation whose tool_use_id names no confirmable tool use in the session, or one that was never gated, in the reference's one sentence for both (2026-09-02 batch2 `sessT.send.tool_confirmation.unknown-id` and `tool_confirmation.not-pending`, #540). The sentence cannot tell the two apart, so Reason carries which fired for the caller's log.

func (*NoPendingConfirmationError) Error added in v0.5.0 

func (e *NoPendingConfirmationError) Error() string

type OutcomeEvaluation added in v0.2.0 

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

OutcomeEvaluation is one in-flight grading cycle being traced.

func (*OutcomeEvaluation) EndEvent added in v0.2.0 

func (o *OutcomeEvaluation) EndEvent(result, explanation string, usage domain.ModelUsage) (NewEvent, error)

EndEvent renders the terminal (or cycle-ending) span.outcome_evaluation_end for the caller's settlement to commit atomically with the entry mutation.

func (*OutcomeEvaluation) Finish added in v0.2.0 

func (o *OutcomeEvaluation) Finish(result string, commitErr error)

Finish closes the OTel span. commitErr is the fate of the transaction that carried the EndEvent, so the trace never masks an aborted cycle as clean.

func (*OutcomeEvaluation) Heartbeat added in v0.2.0 

func (o *OutcomeEvaluation) Heartbeat(ctx context.Context) error

Heartbeat appends one span.outcome_evaluation_ongoing — the liveness signal the docs describe as distinguishing "actively running" from "stuck" while the grader's reasoning stays opaque. The append is fenced on the entry still being `evaluating`, under the same session row lock every settlement takes: a cycle settled underneath us (an interrupt's flip commits with a queue cancel the grader only notices at its next lease renewal) must not put a liveness signal after its terminal end event, so the stale append aborts instead — a skip, not an error. The fence is cycle liveness, not lease ownership: a reclaimed cycle being re-graded is still live, and its old brain's heartbeat stays truthful.

type Preview 

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

Preview is one in-flight previewed event.

func (*Preview) Delta 

func (p *Preview) Delta(ctx context.Context, index int64, text string) error

Delta broadcasts one content_delta fragment for the content-array entry at index. Only agent.message streams deltas; agent.thinking is start-only. Fragments longer than a NOTIFY payload allows are split into several frames at the same index (append semantics make that equivalent).

func (*Preview) EventID 

func (p *Preview) EventID() domain.ID

EventID is the pre-allocated id: append the buffered event under it so subscribers can reconcile deltas.

type Querier 

type Querier interface {
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}

Querier is the slice of pgx shared by pools and transactions, so the checks can run inside a caller's transaction.

type Refusal added in v0.5.0 

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

Refusal is the client's error in a send batch the log-reading checks refuse — RouteInbound, ValidateToolResults, ValidateToolConfirmations and ValidateDefineOutcomes, with DefineOutcomes, which parses for it: the API answers it 400. Every other error they return, a *ThreadNotFoundError and a *NoPendingConfirmationError aside, is a fault reading the log — a database error, a cancelled context — and none of it is the client's to read.

func (*Refusal) Error added in v0.5.0 

func (e *Refusal) Error() string

type RunnableToolUse added in v0.4.0 

type RunnableToolUse struct {
	ID          domain.ID
	Payload     json.RawMessage
	ThreadID    domain.ID
	CrossPosted bool
}

RunnableToolUse is one call an exec driver runs: the use event's id and payload (name and input, or the MCP server and tool), its thread (empty for the primary) and whether it was cross-posted — the result answering it is written on the same thread and cross-posted alike.

func RunnableToolUses added in v0.4.0 

func RunnableToolUses(ctx context.Context, q Querier, sessionID domain.ID, useType domain.EventType, isWebTool func(string) bool) ([]RunnableToolUse, error)

RunnableToolUses lists, in log order, the runnable calls of one tool-use type — the set a driver drains, across every thread, and re-scans before it completes its item so a call committed under the live item is never stranded.

type Scope added in v0.4.0 

type Scope int

Scope is the surface a list serves. Events are stored once: a child thread's rows carry its thread_id, the primary thread's carry none, and a child's row the session view also surfaces is cross_posted. 

const (
	ScopeAll     Scope = iota // the whole log
	ScopeSession              // the session view — also the primary thread's own
	ScopeThread               // one thread's own rows: a child's, or with no ThreadID the primary's (what its turn replays)
)

type Subscription 

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

Subscription delivers two kinds of traffic for one session: Wake, a coalesced "new committed events may exist, re-read the log" signal, and Frames, ephemeral broadcast frames (previews, session.deleted) in arrival order. Frames are best-effort by contract — a subscriber that can't keep up loses frames, never log events.

func (*Subscription) Close 

func (s *Subscription) Close()

func (*Subscription) Frames 

func (s *Subscription) Frames() <-chan json.RawMessage

func (*Subscription) Wake 

func (s *Subscription) Wake() <-chan struct{}

type ThreadNotFoundError added in v0.5.0 

type ThreadNotFoundError struct{ ID domain.ID }

ThreadNotFoundError refuses an interrupt whose well-formed session_thread_id names no thread of this session. The API answers it 404 not_found_error in the reference's words, which carry no "events[i]: " prefix (2026-09-02 batch2 #333 `sessK.send.interrupt.thread-of-other-session`; #841). The recording named another session's thread; an id naming no thread anywhere misses the same lookup and is answered alike (INFERRED, docs/DIVERGENCES.md).

func (*ThreadNotFoundError) Error added in v0.5.0 

func (e *ThreadNotFoundError) Error() string

type ThreadPeer added in v0.4.0 

type ThreadPeer struct {
	ThreadID  domain.ID
	AgentName string
}

ThreadPeer names one end of a message: the thread, and the agent it runs. The primary thread's id is empty. Its agent name is the session agent's, and only a message the primary sends carries it: a message to the primary names no target at all.

type ThreadTransition added in v0.4.0 

type ThreadTransition struct {
	ThreadID domain.ID // the moving thread; empty for the primary
	Status   domain.SessionStatus
	Stop     *domain.StopReason // the idle stop reason; nil for the other statuses
	// Reemit and Force are the two value-independent session emissions every
	// session keeps. Reemit is the payload-only re-idle — the interrupt's
	// idle→idle re-emit, a confirmation that shrinks the ask set: the session
	// event is emitted when the fold is unchanged *and equal to* the thread's
	// new status, so a child's re-idle under a running sibling stays a thread
	// event alone. Force is the reclaim's rescheduled+running pair on a
	// running session: emitted whatever the fold says, carrying the thread's
	// status. Callers set nothing else.
	Reemit, Force bool
}

ThreadTransition is one thread's status move.

func SettleTransition added in v0.5.0 

func SettleTransition(ctx context.Context, q Querier, sid, tid domain.ID, flow ToolFlow) (*ThreadTransition, error)

SettleTransition is the move a settlement makes for flow: to running, or to idle on requires_action naming flow.Pending. Nil when the thread already rests there with the same blockers — in particular, self-hosted approval keeps the same result blocker and makes no artificial running move — or has ended, or is being rescheduled. A platform call may run before a later wait.

type ToolFlow added in v0.5.0 

type ToolFlow struct {
	Pending          []domain.ID
	Unsettled        bool
	PlatformRunnable bool
}

ToolFlow separates receipt (which prevents execution and duplicate answers) from processing (which releases a thread). Calls advance in model/log order; an out-of-order reply remains durable without advancing past an earlier call.

func ThreadToolFlow added in v0.5.0 

func ThreadToolFlow(ctx context.Context, q Querier, sid, tid domain.ID, platformOwned func(string) bool) (ToolFlow, error)

ThreadToolFlow is a read-only view, including accepted but unprocessed input.

func (ToolFlow) Equal added in v0.5.0 

func (f ToolFlow) Equal(g ToolFlow) bool

Equal reports whether two flows leave their thread in the same place, waiting on the same calls in the same order.

func (ToolFlow) Running added in v0.5.0 

func (f ToolFlow) Running() bool

Running reports whether the flow leaves its thread running: every call is settled, or the first one left is the platform's to run. Otherwise the thread waits on Pending (SettleTransition).

type ToolUseRef added in v0.2.0 

type ToolUseRef struct {
	ID          string
	Type        domain.EventType
	ThreadID    domain.ID
	CrossPosted bool
}

ToolUseRef names one outstanding tool call: the tool-use event's id and its type, which together decide the shape of the result that answers it, and the thread it was written on (empty for the primary) with whether it was cross-posted — so its answer lands on the same surfaces.

func UnansweredThreadToolUses added in v0.4.0 

func UnansweredThreadToolUses(ctx context.Context, q Querier, sessionID, threadID domain.ID, extraRefs []string) ([]ToolUseRef, error)

UnansweredThreadToolUses is UnansweredToolUses for one thread's own rows (empty threadID = the primary's) — what a thread-scoped interrupt answers, and what archiving a child answers before it terminates. extraRefs as above.

func UnansweredToolUses added in v0.2.0 

func UnansweredToolUses(ctx context.Context, q Querier, sessionID domain.ID, extraRefs []string) ([]ToolUseRef, error)

UnansweredToolUses lists, in log order, the tool calls HasUnansweredToolUse only counts — the set a user.interrupt has to answer before the session can be resumed, since the model protocol requires every tool_use answered and the log is append-only. extraRefs are treated as answered, exactly as above.

type WhileAwaitingError added in v0.5.0 

type WhileAwaitingError struct {
	Type  domain.EventType
	Index int
	IDs   []domain.ID
}

WhileAwaitingError refuses a user.message or a user.define_outcome posted while the primary thread still waits on responses, in the reference's words (2026-09-12-followups budget-runtime.json #4 `rec.budget.message-over-budget`, a user.message while one agent.custom_tool_use awaited its result). Only that sentence's single-id rendering was recorded; IDs render as Go's %v renders a slice, space-separated, which is the inference for several.

func (*WhileAwaitingError) Error added in v0.5.0 

func (e *WhileAwaitingError) Error() string

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.