events

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Aug 16, 2026 License: Apache-2.0 Imports: 20 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 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 = "The user interrupted this tool call before it returned a result."

InterruptResultText is what an abandoned call is answered with. Never an empty text block: a Messages endpoint rejects one, and that request is what every later replay of this session sends.

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 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 HasUnansweredMCPToolUse added in v0.3.0 

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

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

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 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 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 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 written in one place (AppendInTx's SetStatus), but committed in several, so the recording lives at each commit site rather than beside the write.

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 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 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 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 toolset.IsWebTool): 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 {
	// SetStatus flips sessions.status alongside the append. The batch should
	// carry the matching session.status_* event; this option only moves the
	// resource column.
	SetStatus *domain.SessionStatus
	// AddUsage folds one model turn's usage into sessions.usage.
	AddUsage *domain.ModelUsage
	// MarkProcessedThrough stamps processed_at on 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.

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 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 NotifyWorkEnqueued needs, 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
}

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

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
}

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.

That an interrupt answers the calls at all, and in this shape, is an inference: the reference documents the interrupt's stop reason, not what it writes for the calls it abandons (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.

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

type ToolUseRef struct {
	ID   string
	Type domain.EventType
}

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.

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.