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 ¶
- Variables
- func HasUnansweredPlatformToolUse(ctx context.Context, q Querier, sessionID domain.ID, extraRefs []string) (bool, error)
- func HasUnansweredToolUse(ctx context.Context, q Querier, sessionID domain.ID, extraRefs []string) (bool, error)
- func Previewable(t domain.EventType) bool
- func ToolConfirmationRefs(evs []NewEvent) []string
- func ToolResultRefs(evs []NewEvent) []string
- func UnconfirmedAskEvents(ctx context.Context, q Querier, sessionID domain.ID, extraConfirmed []string) ([]string, error)
- func ValidateToolConfirmations(ctx context.Context, q Querier, sessionID domain.ID, evs []NewEvent) error
- func ValidateToolResults(ctx context.Context, q Querier, sessionID domain.ID, evs []NewEvent) error
- type AppendOptions
- type Broker
- type ListQuery
- type Log
- func (l *Log) Append(ctx context.Context, sessionID domain.ID, evs []NewEvent) ([]domain.Event, error)
- func (l *Log) AppendInTx(ctx context.Context, tx pgx.Tx, sessionID domain.ID, evs []NewEvent, ...) ([]domain.Event, error)
- func (l *Log) AppendWith(ctx context.Context, sessionID domain.ID, evs []NewEvent, opts AppendOptions) ([]domain.Event, error)
- func (l *Log) List(ctx context.Context, sessionID domain.ID, q ListQuery) ([]domain.Event, error)
- func (l *Log) PublishEventFrame(ctx context.Context, sessionID domain.ID, event map[string]any) error
- func (l *Log) StartModelRequest(ctx context.Context, sessionID domain.ID) (context.Context, *ModelRequest, error)
- func (l *Log) StartPreview(ctx context.Context, sessionID domain.ID, typ domain.EventType) (*Preview, error)
- type ModelRequest
- type NewEvent
- type Preview
- type Querier
- type Subscription
Constants ¶
This section is empty.
Variables ¶
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 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 Previewable ¶
PreviewableTypes are the only event types the wire allows previews for.
func ToolConfirmationRefs ¶
ToolConfirmationRefs collects the tool-use ids a batch's user.tool_confirmation events resolve, in batch order.
func ToolResultRefs ¶
ToolResultRefs collects the tool-use ids referenced by a batch's inbound tool-result events, in batch order.
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.
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 ¶
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.
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
// 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 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.
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 (*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 ¶
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) (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). The returned context carries the span for downstream propagation.
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).
func (*ModelRequest) Finish ¶
func (m *ModelRequest) Finish(isError bool, commitErr error)
Finish closes the OTel span. 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) 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 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 Preview ¶
type Preview struct {
// contains filtered or unexported fields
}
Preview is one in-flight previewed event.
func (*Preview) Delta ¶
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).
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{}
Source Files