Documentation
¶
Overview ¶
Package eventlog is the durable, append-only audit log that backs agent.Agent's session.Service. Each event the ADK runner appends to a session is persisted to the underlying database (SQLite, MySQL, or Postgres via GORM) and assigned a monotonic seq number. Subscribers can replay history with Since(fromSeq) or live-tail with Watch(fromSeq).
The package layers on top of ADK's session/database service: ADK owns the events / sessions / state tables, and we add a thin agent_eventlog overlay table whose rows reference ADK's events by id and add the seq column. Two GORM connections (ADK's and ours) share the same database file/DSN — atomic-across-tables writes are not provided in v1; the AppendEvent path writes ADK first, then the overlay, and surfaces overlay-write errors so callers can retry (event_id is unique-indexed for safe idempotency).
See docs/eventlog-plan.md and docs/eventlog-decisions.md for the design rationale and milestone breakdown.
Index ¶
- Constants
- Variables
- func IsSessionNotFound(err error) bool
- type BootRecord
- type BranchLister
- type Entry
- type Handle
- func (h *Handle) AcquireLock(ctx context.Context, app, user, session string) (*SessionLock, error)
- func (h *Handle) Close() error
- func (h *Handle) Ping(ctx context.Context) error
- func (h *Handle) RecentBoots(ctx context.Context, since time.Time) ([]BootRecord, error)
- func (h *Handle) RecordBoot(ctx context.Context, at time.Time, attempted []string) (uint, error)
- func (h *Handle) UpdateBootAttempted(ctx context.Context, id uint, attempted []string) error
- type MetadataExtractor
- type Option
- type QueryOption
- func ForSession(appName, userID, sessionID string) QueryOption
- func WithAnyBranchPrefix(prefixes ...string) QueryOption
- func WithAuthor(name string) QueryOption
- func WithAuthorSuffix(suffix string) QueryOption
- func WithBranchPrefix(prefix string) QueryOption
- func WithLimit(n int) QueryOption
- func WithSessionTree(appName, userID, parentSessionID string) QueryOption
- type SessionLock
- type Stream
Constants ¶
const ( // MetadataKeyCaller is the effective Caller.Identity that // originated the turn this event belongs to. Empty when no auth // context was available (legacy / single-user / out-of-band code // paths). MetadataKeyCaller = "caller" // MetadataKeyProxyBy is set when the effective Caller was // asserted via the proxy path (X-Asserted-Caller header): records // the proxying identity (e.g., "sa:slack-bot"). Empty for direct // authentication. MetadataKeyProxyBy = "proxy_by" )
MetadataKey* are the well-known keys agent.Agent uses when wiring the per-request caller context into the eventlog metadata sidecar. They're exported so audit consumers can read what's there without guessing at conventions.
const FinishReasonMetadataKey = "finish_reason"
FinishReasonMetadataKey is the CustomMetadata key under which AppendEvent records a model turn's genai FinishReason before persist. ADK's storage row has no FinishReason column (createEventFromStorageEvent drops it), but it DOES round-trip CustomMetadata — so stamping the reason here is what lets a reloaded event distinguish a MAX_TOKENS truncation (resume-able) from a normal STOP completion. Read by pkg/agent's auto-continue classifier (#582); mirrors the CompactionMetadataKey piggy-back the classifier already consults.
Variables ¶
var ErrClosed = errors.New("eventlog: stream is closed")
ErrClosed is returned by Stream methods invoked after Close.
var ErrSessionLocked = errors.New("eventlog: session is locked by another process")
ErrSessionLocked is returned by AcquireLock when another live process already holds the lease. The error message includes the holder identifier so operators can diagnose contention.
Functions ¶
func IsSessionNotFound ¶ added in v2.9.0
IsSessionNotFound reports whether err means "that session is not in the log", as opposed to a real failure to read it.
ADK's database session service wraps GORM's ErrRecordNotFound with %w behind the text "database error while fetching session", which reads like a database fault and is not one — a session that has never been written is the ordinary state of every session before its first event. Callers that treat any non-nil error as a problem log a scary line on a healthy cold boot; this is the check that tells the two apart without matching on message text.
It lives here rather than in the callers because the GORM coupling is this package's to own.
Types ¶
type BootRecord ¶ added in v2.8.0
BootRecord is the public view of one boot-scan run.
type BranchLister ¶ added in v2.9.0
type BranchLister interface {
Branches(ctx context.Context, opts ...QueryOption) ([]string, error)
}
BranchLister is an OPTIONAL Stream extension: report the distinct branch labels that exist under a query's filters, without hydrating a single event.
It answers "what ran here?", which the Since/Watch pair can only answer by reading every row and looking at the branch column. The caller that needs it is the attach layer's subagent-events endpoint: a subagent declared as "cluster" writes its turns under the branch the runner minted for the instance ("bg.cluster-1"), so resolving a name to its branches means asking the log which branches are there (go-steer/core-agent#694).
Not part of Stream itself — adding a method to a published interface breaks every implementation outside this repo. Type-assert for it and degrade when it's absent:
if bl, ok := stream.(eventlog.BranchLister); ok { ... }
Honors the same QueryOptions as Since, including WithLimit (a cap on the number of DISTINCT labels returned, not on rows scanned). Labels come back sorted, and the empty branch — the parent session's own turns — is omitted.
type Entry ¶
Entry is one row from the event log: the assigned seq plus the underlying ADK session.Event (loaded via the paired session.Service).
Metadata is an optional sidecar map populated by a MetadataExtractor (see WithMetadataExtractor). The eventlog package itself is agnostic to the keys — agent.Agent wires an extractor that pulls auth.Caller.Identity (key "caller") and proxy attribution (key "proxy_by") from the request context. Rows persisted before the sidecar column shipped read back as a nil Metadata map.
type Handle ¶
type Handle struct {
// Stream is the seq + replay + watch primitive.
Stream Stream
// Service is the session.Service backed by the same database.
// Pass to agent.WithSessionService (or use the
// agent.WithEventLog convenience that does both at once).
Service session.Service
// DB exposes the overlay-table connection so adjacent
// substrates (e.g., pkg/attach.SessionACLStore) can share the
// same database without re-opening it. Read-only access from
// outside pkg/eventlog — mutations happen via the typed
// stores. Nil before Open returns and after Close.
DB *gorm.DB
// contains filtered or unexported fields
}
Handle bundles the Stream with the session.Service that writes to the same database. agent.WithEventLog(handle) wires both into an agent.Agent in one call.
func Open ¶
Open constructs a Handle backed by the supplied GORM dialector. Pass any standard dialector (sqlite.Open, postgres.Open, mysql.Open).
Open does several things:
- Constructs ADK's database.SessionService against the dialector and runs its AutoMigrate so the events / sessions / state tables exist.
- Opens a second GORM connection for our overlay table and AutoMigrates agent_eventlog.
- For SQLite (detected via the dialector's Name()), enables WAL journal mode so concurrent readers can run alongside the writer. Disable with WithSkipWAL.
- Wraps the ADK service so AppendEvent writes to both layers.
func (*Handle) AcquireLock ¶
AcquireLock takes an exclusive lease on (app, user, session) for the lifetime of the returned *SessionLock. Returns ErrSessionLocked if another process holds a fresh lease (heartbeat within staleAfter); steals the lease if the existing holder's heartbeat is older than staleAfter (indicating a crashed process).
The lease is heartbeated automatically until SessionLock.Release is called; Release is idempotent and safe to defer.
func (*Handle) Close ¶
Close releases all resources held by the Handle (Stream + the underlying database connection). Safe to call multiple times.
func (*Handle) Ping ¶ added in v2.9.0
Ping reports whether the event log is still readable, for use by a readiness probe (#946).
It issues a real bounded read against the overlay table rather than pinging the connection pool. On SQLite a pool ping proves almost nothing: the pool hands back a live *sql.DB whose file may since have been deleted, replaced, or locked by another writer, and database/sql only discovers that on a statement. One indexed row read is the cheapest query that actually touches the file.
A closed Handle reports ErrClosed rather than panicking, so a probe racing shutdown fails cleanly instead of taking the process with it.
func (*Handle) RecentBoots ¶ added in v2.8.0
RecentBoots returns boot-scan records with BootAt >= since, oldest first. Rows with unparsable Attempted blobs are returned with a nil slice rather than dropped — the breaker counts boots, and losing a row to corruption would weaken the guard exactly when things are already going wrong.
func (*Handle) RecordBoot ¶ added in v2.8.0
RecordBoot appends a boot-scan record and returns the new row's ID. attempted lists the session IDs the scan triggered continuations for (empty slice = scan ran, found nothing). The returned ID lets a caller later narrow the row to the sessions that actually got a fair shot via UpdateBootAttempted — outcome-aware accounting for the write-ahead intent record (#575). AutoMigrate is idempotent and cheap at once-per-boot call frequency.
func (*Handle) UpdateBootAttempted ¶ added in v2.8.0
UpdateBootAttempted rewrites the attempted-session list of an existing boot-log row (identified by the ID from RecordBoot). It exists to narrow a pessimistic write-ahead record down to the sessions a scan actually attempted, after synchronous skips (a run lock held by another daemon, a raced-clean tail) have been resolved — so those skips don't burn the per-session cumulative cap on daemons that never got a fair shot (#575, the fleet cap-burn). The write-ahead property is preserved: this only ever runs AFTER the intent record exists, so a daemon killed mid-scan leaves the full pessimistic list in place.
type MetadataExtractor ¶
MetadataExtractor pulls the per-event sidecar metadata from the context at Append time. Return nil (or empty map) to skip — empty maps round-trip as nil on the read side. Callers wire an extractor via WithMetadataExtractor; the default is no-op (preserves the pre-sidecar shape on disk).
The function is called inside Append's request context, so it can safely fetch request-scoped values without spawning goroutines.
type Option ¶
type Option func(*openOpts)
Option configures Open.
func WithGORMConfig ¶
WithGORMConfig overrides the gorm.Config used for the overlay connection. Useful for silencing the default logger in tests (gorm.Config{Logger: logger.Default.LogMode(logger.Silent)}).
func WithMetadataExtractor ¶
func WithMetadataExtractor(fn MetadataExtractor) Option
WithMetadataExtractor wires a function that produces sidecar metadata for each appended event. The map (JSON-encoded) is stored in the overlay row's metadata column and surfaces on Entry.Metadata at read time. Nil disables the sidecar (default).
agent.New wires an extractor that pulls the per-request caller + proxy attribution from the context so multi-session audit logs carry who triggered each event without coupling pkg/eventlog to pkg/auth.
func WithSkipWAL ¶
func WithSkipWAL() Option
WithSkipWAL disables the automatic PRAGMA journal_mode=WAL set on SQLite databases at Open time. WAL is on by default because it permits concurrent readers alongside a writer; turn it off for in-memory databases or read-only setups where WAL adds no value.
func WithWatchInterval ¶
WithWatchInterval sets the polling interval Watch uses to check for new rows. Default is 200ms. Smaller values reduce subscriber latency at the cost of database load; larger values do the opposite.
type QueryOption ¶
type QueryOption func(*queryOpts)
QueryOption filters Since/Watch results.
func ForSession ¶
func ForSession(appName, userID, sessionID string) QueryOption
ForSession restricts results to one session triple. Without it, queries scan across every session in the database — useful for audit dashboards, dangerous for high-volume reads.
func WithAnyBranchPrefix ¶ added in v2.9.0
func WithAnyBranchPrefix(prefixes ...string) QueryOption
WithAnyBranchPrefix matches events whose Branch begins with ANY of the supplied prefixes (an OR group, AND'd with the other filters).
One subagent is reachable under several branch spellings depending on how it was launched — a sync subagent tool tags its events with the bare name, the background runner with "bg.<name>", RunSubtask with "sub.<name>", the remote runner with "remote.<name>" — so an operator asking "what did subagent X do?" needs the union, not a single prefix (#638).
Empty prefixes are dropped: a caller building the list from user input can't accidentally widen the query to everything by passing "". LIKE metacharacters ('%', '_') in a prefix are escaped, so a prefix matches literally and only the intended subtree.
func WithAuthor ¶
func WithAuthor(name string) QueryOption
WithAuthor matches events emitted by a specific author. The autonomous driver uses Author="<binary>/autonomous" for checkpoint events; consumer-supplied authors work the same way.
func WithAuthorSuffix ¶
func WithAuthorSuffix(suffix string) QueryOption
WithAuthorSuffix matches events whose Author ends with the supplied suffix. Used by ResumeAutonomous to find checkpoint events regardless of which binary emitted them — checkpoints land with Author="<binary>/autonomous", so suffix "/autonomous" matches checkpoints from any core-agent-family process. Empty suffix is a no-op (matches everything).
func WithBranchPrefix ¶
func WithBranchPrefix(prefix string) QueryOption
WithBranchPrefix matches events whose Branch field begins with prefix. Use to scope queries to a subagent subtree once Phase 4 of the eventlog plan ships subagent runners that set Branch.
Repeated calls accumulate and are OR'd together — see WithAnyBranchPrefix, which this delegates to.
func WithLimit ¶
func WithLimit(n int) QueryOption
WithLimit caps the number of entries returned. Zero or negative is treated as no limit.
func WithSessionTree ¶
func WithSessionTree(appName, userID, parentSessionID string) QueryOption
WithSessionTree restricts results to the parent session ID and any derived sub-session IDs. The subagent runner names its session "<parent>:sub:<branch>" by convention; this option's underlying SQL matches the parent + every "<parent>:sub:%" descendant in one query so an audit can pull the whole tree without a follow-up join.
When set, takes precedence over the (App, User, Session) triple from ForSession — the two are mutually exclusive in practice because WithSessionTree implies the (app, user) pair already. Mutually composable with the other QueryOptions (WithBranchPrefix, WithAuthor, WithAuthorSuffix, WithLimit).
type SessionLock ¶
type SessionLock struct {
// contains filtered or unexported fields
}
SessionLock is the lease returned by Handle.AcquireLock. It runs a background goroutine that refreshes heartbeat_at every heartbeatInterval until Release is called. Safe to call Release multiple times.
func (*SessionLock) Holder ¶
func (l *SessionLock) Holder() string
Holder returns the identifier we registered when acquiring the lock. Useful for diagnostics + for tests that need to assert the row content.
func (*SessionLock) Lost ¶ added in v2.8.0
func (l *SessionLock) Lost() <-chan struct{}
Lost returns a channel that is closed if the lease is stolen out from under us while it is held — the heartbeat's conditional UPDATE matched zero rows, meaning another process reclaimed the lease after our heartbeat lapsed past the staleness window (a >staleAfter GC pause, sleep, or DB stall). A caller running work under the lock — e.g. the autonomous run loop — must select on this channel and abort promptly, otherwise both processes run against the same session: the exact split-brain the lock exists to prevent. The channel is never closed for a lock that is cleanly Released while still held.
func (*SessionLock) Release ¶
func (l *SessionLock) Release() error
Release ends the lease and stops the heartbeat goroutine. Idempotent; safe to defer.
type Stream ¶
type Stream interface {
// Append writes ev to the log under sess. Returns the assigned
// seq number. The event itself is also expected to be persisted
// via the paired session.Service.AppendEvent — Stream.Append
// only writes the overlay row that carries the seq.
//
// Most callers don't invoke this directly; agent.Run drives the
// session.Service which in turn calls Append internally.
Append(ctx context.Context, sess session.Session, ev *session.Event) (seq int64, err error)
// Since returns events with seq > fromSeq, in seq order. Bounded
// by current end-of-log; returns when caught up. Apply filters
// via QueryOption (ForSession, WithBranchPrefix, WithAuthor,
// WithLimit).
Since(ctx context.Context, fromSeq int64, opts ...QueryOption) iter.Seq2[Entry, error]
// Watch returns events with seq > fromSeq, in seq order, blocking
// for new events as they're appended. Cancel ctx to stop. Same
// QueryOptions as Since.
//
// The default poll interval is 200ms, configurable via Open's
// WithWatchInterval option.
Watch(ctx context.Context, fromSeq int64, opts ...QueryOption) iter.Seq2[Entry, error]
// Close releases resources held by the Stream (typically the
// underlying gorm.DB connection pool). Safe to call multiple
// times.
Close() error
}
Stream is the append-only event log primitive. Implementations are expected to be safe for concurrent use.
Source Files