obsdb

Documentation

Overview

Package obsdb is the observability database: what weft/otel's local sink writes and Weft Studio reads, in one module so neither owns the schema (ADR 0024, S3).

The model is OTLP-shaped — spans and log records, resource included — so a stock OpenTelemetry pipeline's data and weft's own land in the same tables. Weft identity (the run → session → public-id chain) is derived from attributes, never parsed out of ids, and is stored as indexed columns through the Weft struct.

Three moving parts:

• DB is the backend interface: Write (idempotent on (run, record kind, position) and (trace, span)), run/session queries, the event page with its gap detector, the transcript a replay reads, and the spans of a run or trace. Status is derived from the four-row table (DeriveStatus), never stored.
• The hub (Hub, Frame) is the live lane: every Write publishes its frames before returning, so an in-process subscriber — Studio's SSE handler in setup A — sees a record without a network hop.
• obsdbtest is the conformance table every backend runs (the storetest pattern); sqlite is the default backend, clickhouse the hosted one (its own module).

Deltas travel but are not stored (their own counter means their absence never looks like a lost event); heartbeats are never stored, they only move a run's last-seen. Both rules are Write's, shared by every backend.

Index

• Constants
• Variables
• func DedupTranscript(bodies []json.RawMessage) []json.RawMessage
• func LimitOf(n int) int
• func MetaOf(attrs map[string]any) map[string]string
• type Batch
• type DB
• type EventPage
• type Experiment
• type ExperimentInput
• type ExperimentVariant
• type Frame
  • func RecordFrame(r Record) Frame
  • func RunFrame(r RunRow) Frame
• type Hub
  • func NewHub(opts ...HubOption) Hub
• type HubOption
  • func QueueSize(n int) HubOption
• type PosEvent
• type Record
  • func FromOTLPLogs(req *collogspb.ExportLogsServiceRequest) []Record
• type RunDetail
• type RunPage
• type RunQuery
• type RunRow
• type Selector
• type SessionDetail
• type SessionPage
• type SessionQuery
• type SessionRow
• type Span
  • func FromOTLPTraces(req *coltracepb.ExportTraceServiceRequest) []Span
• type SpanEvent
• type Status
  • func DeriveStatus(spanFailed, runFinished bool, lastSeen, now time.Time) Status
• type Weft
  • func DeriveRecord(r Record) Weft
  • func DeriveSpan(s Span) Weft

Constants

const (
	FrameRecord = "record"
	FrameRun    = "run"
)

Frame kinds.

const InterruptedAfter = 30 * time.Second

InterruptedAfter is the "last seen" age that turns a non-terminal run interrupted: three missed heartbeats.

const MaxGaps = 1000

MaxGaps bounds EventPage.Gaps: positions are the sender's numbers, and one stray high position must not turn a page read into a list of every position below it.

const MaxTies = 500

MaxTies bounds how far a page may run past its Limit to finish a group of rows sharing the cursor's time (RunPage, SessionPage): one page never grows unbounded however many rows share one timestamp.

Variables

var ErrClosed = errors.New("obsdb: closed")

ErrClosed is returned by a backend used after Close.

var ErrNotFound = errors.New("obsdb: not found")

ErrNotFound is returned by Run, Session, Events, Transcript, RunSpans, Experiment and ResolvePublicID for an id the database does not hold. Trace answers an unknown trace with no spans instead: a trace is only ever the spans that arrived.

Functions

func DedupTranscript

func DedupTranscript(bodies []json.RawMessage) []json.RawMessage

DedupTranscript makes a run's messages bodies concatenate to the transcript the run actually held. bodies are the messages records' bodies in index order, as a backend's Transcript reads them; every backend's Transcript returns its result through this.

One shape needs it. A resume whose input already holds a partial tool message (some calls of the step answered, the rest parked) records that input as fed — index 0 — and then the tool message it rebuilt over the partial one, at index 1 (the core's attachResults; both records are the contract, ADR 0024 D1). Concatenated naively the two versions of the one message sit side by side; the later record is the authoritative one. So when the second body is a lone tool message and the first holds, directly after its last assistant message with tool calls, a tool message whose results the lone one repeats, the partial message is dropped from the first body. Every other message stays byte-for-byte, the number of bodies never changes (a body's place is still its record's index), and any other shape is returned untouched.

func LimitOf

func LimitOf(n int) int

LimitOf normalizes a query limit: 0 means the default 50, values above 500 clamp to 500, negatives read as the default.

func MetaOf

func MetaOf(attrs map[string]any) map[string]string

MetaOf returns the caller metadata a set of attributes carries: every string value under a key outside the record/span contract (what core.Metadata put there, enduser.id included). Keys under "weft." that are not part of the contract — thread's weft.session.parent, a future module's own — are metadata and are kept.

Types

type Batch

type Batch struct {
	Spans   []Span
	Records []Record
}

Batch is one write: spans and records as ingested, in arrival order. Write is idempotent on (run, record kind, pos) and (trace, span), so a retried transport's duplicate is a no-op.

type DB

type DB interface {
	// Write stores one batch, idempotent on (run, record kind, pos)
	// and (trace, span): a retried transport's duplicate is a no-op.
	// Deltas are counted, never stored; heartbeats are never stored —
	// they only move the run's last-seen (DeriveStatus's clock).
	Write(ctx context.Context, b Batch) error

	Runs(ctx context.Context, q RunQuery) (RunPage, error)
	Run(ctx context.Context, id string) (RunDetail, error) // ErrNotFound
	// Events pages a run's durable events in position order. after is
	// exclusive — pass -1 to read from the start (0 would skip position
	// 0, the run_start) and EventPage.NextAfter to continue. limit: 0 =
	// 100, max 1000.
	Events(ctx context.Context, runID string, after int64, limit int) (EventPage, error)
	// Transcript returns the messages bodies in index order, one per
	// messages record, read through DedupTranscript — so they
	// concatenate to the transcript the run held.
	Transcript(ctx context.Context, runID string) ([]json.RawMessage, error)
	RunSpans(ctx context.Context, runID string) ([]Span, error)
	Trace(ctx context.Context, traceID string) ([]Span, error) // empty, not an error, for an unknown trace

	Sessions(ctx context.Context, q SessionQuery) (SessionPage, error)
	Session(ctx context.Context, id string) (SessionDetail, error)
	ResolvePublicID(ctx context.Context, publicID string) (sessionID string, err error)

	// The playground's saved experiments (WEFT-PLAYGROUND §10.4,
	// PQ4): the definition rows keyed by the weft.experiment.id the
	// runs carry — one small table beside runs.
	SaveExperiment(ctx context.Context, e Experiment) error // upsert by ID
	Experiments(ctx context.Context) ([]Experiment, error)  // newest update first
	Experiment(ctx context.Context, id string) (Experiment, error)

	Close() error
}

DB is the observability database every backend implements (sqlite by default, clickhouse hosted). Implementations must be safe for concurrent use: writers are rare (the local sink's synchronous processors), readers many (Studio).

type EventPage

type EventPage struct {
	Events    []PosEvent
	NextAfter *int64
	Done      bool // the run is terminal and every stored event was returned
	Gaps      []int64
}

EventPage is one page of a run's durable events. Gaps are durable positions missing below the high-water mark: a lost batch, never a delta — deltas are on their own counter, so their absence cannot open a hole here (D3). At most MaxGaps are listed, lowest first.

type Experiment

type Experiment struct {
	ID, Name, Agent  string
	Created, Updated time.Time
	Variants         []ExperimentVariant
	Inputs           []ExperimentInput
}

Experiment is one saved playground group: a name, the variants and inputs that define it, keyed by the id its runs carry as weft.experiment.id (§10.4). Variants and inputs are the wire shapes verbatim (overrides JSON included); the runs themselves stay in the runs table.

type ExperimentInput

type ExperimentInput struct {
	Key         string `json:"key"`
	SourceRunID string `json:"source_run_id,omitempty"`
	Text        string `json:"text,omitempty"`
}

ExperimentInput is one row of the matrix: a key, a source run to take the turn from, or a literal text.

type ExperimentVariant

type ExperimentVariant struct {
	Key       string          `json:"key"`
	Overrides json.RawMessage `json:"overrides,omitempty"`
}

ExperimentVariant is one column of the matrix: a key and the overrides object it applies.

type Frame

type Frame struct {
	Kind   string  // "record" | "run"
	Record *Record // Kind == record; Weft derived
	Weft   Weft
	Run    *RunRow // Kind == run
	Seq    uint64  // hub-wide, monotonic: the SSE id and the resume cursor
}

Frame is one live-lane message: a record as ingested (deltas included), or a run row that changed.

func RecordFrame

func RecordFrame(r Record) Frame

RecordFrame builds the frame a Write publishes for one record: the record as ingested with its derived identity, deltas included.

func RunFrame

func RunFrame(r RunRow) Frame

RunFrame builds the frame a Write publishes for a run row that changed.

type Hub

type Hub interface {
	Publish(ctx context.Context, f Frame)
	Subscribe(ctx context.Context, sel Selector, after uint64) (<-chan Frame, error)
}

Hub is the live lane. Publish never blocks the caller beyond a bounded per-subscriber queue; a slow subscriber is dropped with Overflow.

func NewHub

func NewHub(opts ...HubOption) Hub

NewHub returns the in-process hub: one Seq counter for every frame, one bounded queue per subscriber (QueueSize, default 1000). Hosted Studio swaps in a Redis/NATS implementation of the same interface; this one is what setup A's live lane runs on.

type HubOption

type HubOption interface {
	// contains filtered or unexported methods
}

HubOption configures NewHub.

func QueueSize

func QueueSize(n int) HubOption

QueueSize sets the per-subscriber queue depth. Publish to a full queue drops the subscriber (its channel closes) rather than blocking the writer: the durable lane is the database's job, and a live tail can always resume from it.

type PosEvent

type PosEvent struct {
	Pos   int64
	Time  time.Time
	Event json.RawMessage // the body, verbatim
}

PosEvent is one positioned event of a run: the body verbatim, at its durable position.

type Record

type Record struct {
	Time, Observed  time.Time
	TraceID, SpanID string
	Severity        int
	EventName       string
	Body            string
	Service         string
	Attrs           map[string]any
	Resource        map[string]any
}

Record is one log record, OTLP-shaped.

func FromOTLPLogs

func FromOTLPLogs(req *collogspb.ExportLogsServiceRequest) []Record

FromOTLPLogs decodes an OTLP/HTTP log export request into the model. weft records (weft.run.id present) and non-weft records (slog lines, other services) both pass through; the kind they carry decides where Write puts them.

type RunDetail

type RunDetail struct {
	RunRow
	Children []RunRow
}

RunDetail is one run with its subagent children (runs whose ParentRunID is this run, by Started).

type RunPage

type RunPage struct {
	Runs         []RunRow
	Total        int
	NextBefore   *time.Time
	NextBeforeID string
}

RunPage is one Runs result, newest Started first (ties by ID, descending). NextBefore and NextBeforeID are the next page's RunQuery.Before and BeforeID, nil and "" on the last page. A page does not end inside a group of runs sharing one Started: the rest of the group rides along, so a caller paging on Before alone skips nothing — up to MaxTies rows past Limit. A longer tie (a bulk import at one truncated timestamp) is cut there, and only the (Before, BeforeID) pair resumes inside it. Total counts the whole match, cursor aside.

type RunQuery

type RunQuery struct {
	Agent, SessionID, PublicID, ParentRunID string            // ParentRunID: "" top-level only, "*" all
	Status                                  Status            // "" any
	Playground                              *bool             // nil any
	ExperimentID                            string            // the runs of one experiment
	Meta                                    map[string]string // subset match on metadata
	Before                                  time.Time         // cursor on Started; zero = newest
	// BeforeID, with Before, makes the cursor the pair (Started, ID):
	// the runs after that one in the list order — exact inside a group
	// sharing one Started. Pass RunPage.NextBeforeID.
	BeforeID string
	Limit    int // 0 = 50, max 500
}

RunQuery selects runs for Runs. The zero value lists top-level runs, newest first, 50 at a time.

type RunRow

type RunRow struct {
	ID, ParentRunID, ParentCallID, TraceID                     string
	Agent, Provider, Model, ManifestHash, WeftVersion, Service string
	SessionID, PublicID                                        string
	Turn                                                       int
	Playground                                                 bool
	ExperimentID, ForkedFrom                                   string
	Meta                                                       map[string]string
	Started                                                    time.Time
	Finished                                                   *time.Time
	LastSeen                                                   time.Time
	Status                                                     Status
	Err                                                        string
	Steps, Pending                                             int
	StopReason                                                 string
	Usage                                                      core.Usage
	EventCount, MessageCount                                   int64
	DeltaCount                                                 int64 // deltas are counted, never stored (Q4)
}

RunRow is one run in a list or detail: the identity chain, timing, derived status and the denormalised usage and counts the list view reads without a second query.

type Selector

type Selector struct{ RunID, SessionID, PublicID, Agent string }

Selector scopes a subscription: exactly one field set. A run-scoped subscription follows one run; a session- or public-id-scoped one spans the many runs of one conversation, which is why the cursor is hub-wide rather than per run.

type SessionDetail

type SessionDetail struct {
	SessionRow
	Runs []RunRow
}

SessionDetail is one session with its turns — all of them: top-level runs by Turn, then Started; experiments excluded (they hang off runs via ForkedFrom).

type SessionPage

type SessionPage struct {
	Sessions     []SessionRow
	Total        int
	NextBefore   *time.Time
	NextBeforeID string
}

SessionPage is one Sessions result, paged as RunPage is: newest LastSeen first (ties by ID, descending), NextBefore/NextBeforeID are the next SessionQuery.Before/BeforeID, sessions sharing the last row's LastSeen ride along up to MaxTies, and Total counts the whole match.

type SessionQuery

type SessionQuery struct {
	Agent, PublicID string
	Before          time.Time // cursor on LastSeen; zero = newest
	BeforeID        string    // with Before: the (LastSeen, ID) pair, as RunQuery.BeforeID
	Limit           int       // 0 = 50, max 500
}

SessionQuery selects sessions. The zero value lists them newest activity first, 50 at a time.

type SessionRow

type SessionRow struct {
	ID, PublicID, Agent string
	Turns               int
	FirstSeen, LastSeen time.Time
	Status              Status // the newest turn's
	Usage               core.Usage
}

SessionRow is one session (a GROUP BY over its top-level, non-playground runs).

type Span

type Span struct {
	TraceID, SpanID, ParentSpanID string // lowercase hex
	Name                          string
	Kind                          int // OTLP SpanKind
	Start, End                    time.Time
	StatusCode                    int // 0 unset, 1 ok, 2 error
	StatusMessage                 string
	Service                       string         // resource service.name
	Attrs                         map[string]any // string, bool, int64, float64, []any
	Resource                      map[string]any
	Events                        []SpanEvent // exceptions etc.
}

Span is one finished span, OTLP-shaped.

func FromOTLPTraces

func FromOTLPTraces(req *coltracepb.ExportTraceServiceRequest) []Span

FromOTLPTraces decodes an OTLP/HTTP trace export request into the model. The receiver (studio/ingest) and a stock collector's dump produce the same rows the local sink does; non-weft spans pass through unchanged — a polyglot trace is a first-class citizen.

type SpanEvent

type SpanEvent struct {
	Time  time.Time
	Name  string
	Attrs map[string]any
}

SpanEvent is one event recorded on a span (an exception, a marker).

type Status

type Status string

Status is a run's derived state. running, succeeded and failed are what the data says; interrupted is what a reader concludes from a stale last-seen (DeriveStatus) — never stored, so a crash leaves evidence and a live run in another process is not mistaken for a corpse (the store's rule, moved to last-seen).

const (
	StatusRunning     Status = "running"
	StatusSucceeded   Status = "succeeded"
	StatusFailed      Status = "failed"
	StatusInterrupted Status = "interrupted"
)

func DeriveStatus

func DeriveStatus(spanFailed, runFinished bool, lastSeen, now time.Time) Status

DeriveStatus applies the four-row table every backend reads through:

the invoke_agent span ended with status error  →  failed
a run_finish record exists (and no error span) →  succeeded
neither, and now - LastSeen ≤ InterruptedAfter →  running
neither, and older                             →  interrupted

spanFailed is the invoke_agent span's error status; runFinished is the run_finish record's presence. LastSeen is the newest record or span time for the run, heartbeats included. failed outranks succeeded: a run whose span errored failed, whatever records raced behind it. Pending > 0 on a succeeded run means parked at an approval, not stuck.

type Weft

type Weft struct {
	RunID, ParentRunID, ParentCallID string
	SessionID, PublicID              string
	Turn                             int
	Agent                            string
	Record                           string // event | delta | messages | heartbeat | "" (non-weft)
	EventType                        string
	Pos                              int64 // event pos, messages index, or delta pos
	Step                             int   // -1 when absent
	ToolSeq                          int64 // -1 when absent
	Playground                       bool
	ExperimentID, ForkedFrom         string
}

Weft is what obsdb derives from a span or record's attributes, stored as indexed columns. Zero values mean absent.

func DeriveRecord

func DeriveRecord(r Record) Weft

DeriveRecord returns the weft identity a log record carries. Pos is the event position for event records, the messages index for transcript records, and the delta position for delta records — the value the (run, kind, pos) key and the live lane's dedup need. A record with no weft.run.id is non-weft and lands in other_logs.

func DeriveSpan

func DeriveSpan(s Span) Weft

DeriveSpan returns the weft identity a span carries, from its attributes. A span with no weft.run.id is non-weft (a stock OTel application's span): every field reads absent.