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

events

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Sep 17, 2026 License: Apache-2.0 Imports: 22 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 DenialResultText = "The user declined this tool call."

DenialResultText is what a refused call is answered with when the client gives no deny_message. Never an empty text block, for the reason InterruptResultText is not: a Messages endpoint rejects one, and the denial is replayed into every later request this session assembles.

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.

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 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 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 LatestOutcomeStartID added in v0.2.0 

func LatestOutcomeStartID(history []domain.Event, outcomeID domain.ID) domain.ID

LatestOutcomeStartID returns the id of the most recent span.outcome_evaluation_start for outcomeID in the history — the start the cycle's end event references. 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 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: an idle thread has a human verdict or a message outstanding, not a tool. 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) and checks the client's explicit session_thread_id against it, 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 — and a claim that names another thread is the client's 400. An interrupt's claim names the one thread it ends: it must be a live thread of this session; 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, per event, whether the client named a thread at all — 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 ToolConfirmationRefs 

func ToolConfirmationRefs(evs []NewEvent) []string

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

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 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, evs []NewEvent, batchInterrupts bool) error

ValidateDefineOutcomes enforces the DB-backed halves of accepting a user.define_outcome, 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.

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.

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 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 events at seq <= the watermark — the brain recording
	// which inbound events its turn consumed. Zero means no stamping.
	MarkProcessedThrough int64
	// 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 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 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.

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

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

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

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

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.

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.

The denial's result shape is an inference: the reference documents the confirmation event, not the result a denial produces (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, raws []json.RawMessage) ([]NewEvent, error)

NormalizeInbound validates one send batch. envKind is the session's environment kind ("cloud" | "self_hosted"), which gates user.tool_result.

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 delivers; waking is WakeOnThreadEnded's, and the endings a client drives pair the two. 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 sessions.status, and returns the events to append — the thread's own first (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), the session's second when the folded value changed or Force — 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.

func WakeOnThreadEnded added in v0.4.0 

func WakeOnThreadEnded(ctx context.Context, tx pgx.Tx, sessionID, child domain.ID) ([]NewEvent, *domain.SessionStatus, bool, error)

WakeOnThreadEnded wakes the primary when the child that just ended was the last thing a parked coordinator could have been waiting for, and returns the events to append, the session status the wake moved to, and whether it woke (the caller enqueues the primary's model_turn then). Run before the ending thread's own transition, so this reads the status it ended from.

Both halves of the test are load-bearing. A wait parks only while some busy child exists, so an ending that removes the last one is the last chance to wake the coordinator: without it the notice sits unread on a session folding idle with no turn coming — the wedge W1 exists to avoid, reached the long way round. And an ending that leaves another busy child behind — or that ends a child nothing was waiting on, an idle one a client tidied away — must NOT wake: the coordinator either has a report still coming, whose own arrival wakes it, or was never parked at all, and a turn it did not ask for costs a model call, blocks the session's own archive behind it, and puts the archive path's idle re-advertisement (decision 4's pick) out of reach.

The primary is woken on the rule a report uses (WakeThread): idle on end_turn only. A running coordinator chains on the notice by seq at its own settle, and one parked on requires_action stays parked until its human answers.

"Parked on a wait" is approximated by "a busy child existed and this was the last of them", because nothing on the log says a turn ended in a wait — the wait's answer is an ordinary tool_result. The approximation errs towards waking: a coordinator that idled without waiting is woken too, and reads the notice a turn earlier than it otherwise would. That costs a model call the client did not ask for; the other direction costs a coordinator parked on a child that no longer exists, which nothing can end.

func WakeThread added in v0.4.0 

func WakeThread(ctx context.Context, tx pgx.Tx, sessionID, threadID domain.ID) (pair []NewEvent, moved *domain.SessionStatus, woke bool, err error)

WakeThread decides the one question a delivered message raises: whether anything else will ever move the target thread. A thread idle on end_turn — or on no reason at all, which a primary row the thread migration backfilled can be — has no turn coming, so it is flipped to running here and the caller enqueues its model_turn; a thread parked on requires_action is idle only until its human answers and replays the message when its turn resumes; a running one reads it at its next settle (plan 35 decision 6). A thread that exhausted its retries is not woken: nothing here can make its turn succeed, and its coordinator was already told it stopped.

It runs in the caller's transaction, under the session row lock every settlement and trigger holds, which is what serializes two children reporting at once: the first finds the parent idle and wakes it, the second finds it running and leaves the chain to the parent's own settlement.

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 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) ([]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 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. Both are empty for the primary thread, whose agent name is absent — null on the wire — because it has a role rather than a roster name.

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.

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.

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.