obsdb

type Assembler ¶ added in v0.13.0

type Assembler struct {
	// contains filtered or unexported fields
}

Assembler is AssembleStep over one run's growth batches decoded once: each Step call answers exactly what AssembleStep answers for the same records and step, without re-reading every batch (a reader that walks every step of a run — the wefttest fixture export — would otherwise decode the transcript once per step). The messages Step returns share their parts with the Assembler's decoded batches: read them, or copy before patching. Not safe for concurrent use.

func NewAssembler(batches []TranscriptBatch) *Assembler

func (a *Assembler) Step(requests []RequestRecord, compactions []Compaction, step int) (StepMessages, error)

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 Compaction ¶ added in v0.10.0

type Compaction struct {
	Scope        string
	Hash         string // weft.compaction.hash
	Index        int64
	Step         int
	FromSeq      int64
	ToSeq        int64
	Messages     json.RawMessage
	Reason       string
	Replaced     int
	Entries      int
	TokensBefore int64
	TokensAfter  int64
}

Compaction is one compaction a run's records name (ADR 0028 §8), read by DB.Compactions. Two scopes:

• CompactionRun: a compaction view, the messages record the core emits when a PrepareStep sent a request whose messages are not the run's transcript. The request whose messages_ref names Index saw the growth records up to Index, concatenated, with the transcript range [FromSeq, ToSeq) replaced by Messages. It is never part of the plain transcript (Transcript and TranscriptBatches hold growth records only) and never applies to another request.
• CompactionSession: the informational marker weft/thread emits (a record of kind compaction) when the session compacts, filed under the last run that produced the compacted context (or, for a context no run produced, under the next run). The next run's input record holds the compacted context literally; the marker carries no messages and a reader never applies it. Index and Step are -1, FromSeq and ToSeq 0.

Replaced and Entries are the change in messages: for a view, ToSeq-FromSeq and the length of Messages; for a session marker, the counts thread reported over the session's context. TokensBefore and TokensAfter are thread's estimates (session scope only; 0 for a view, whose tokens the core does not estimate). Reason is thread's compaction reason (manual, threshold, overflow, from_hook, trim); "" for a view.

func CompactionOf(kind string, index int64, step int, attrs map[string]any, body []byte) (c Compaction, ok bool, err error)

func ViewOf(rec RequestRecord, compactions []Compaction) (Compaction, bool)

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
	// growth messages record (a compaction view, ADR 0028 §8, is not
	// transcript: see Compactions), read through DedupTranscript — so
	// they concatenate to the transcript the run held.
	Transcript(ctx context.Context, runID string) ([]json.RawMessage, error)
	// TranscriptBatches is Transcript with what each messages record
	// stored beside its body: its index, the step it joined
	// (weft.step.index, -1 when the record carried none) and whether
	// it is the run's input (weft.messages.input). The bodies are
	// Transcript's, through DedupTranscript, one per record.
	TranscriptBatches(ctx context.Context, runID string) ([]TranscriptBatch, error)
	// Compactions returns the compactions a run's records name (ADR
	// 0028 §8): every run-scope view in index order, then the session
	// markers thread filed under the run, in emission order — what a reader needs to draw the compaction and
	// rebuild a request's messages. Transcript and TranscriptBatches
	// never include a view: the plain transcript is growth records
	// only. A messages record with an unknown weft.messages.reason is
	// an error. ErrNotFound for an unknown run; empty, not an error,
	// for a run that never compacted.
	Compactions(ctx context.Context, runID string) ([]Compaction, error)

	// The request record (ADR 0028). Requests pages a run's request
	// records in index order, one per model-call attempt
	// (RequestQuery). Prompt and Tools return the run's prompt or tools
	// record of one hash: ErrNotFound when the run holds none, as a
	// *HoleError when the reason is known (not_recorded, stripped,
	// gap — ExplainMissing). Catalogs returns every tools record of the
	// run, one per hash, in index order; a content-off run has none.
	// All four answer ErrNotFound for an unknown run.
	Requests(ctx context.Context, runID string, q RequestQuery) ([]RequestRecord, error)
	Prompt(ctx context.Context, runID, hash string) (PromptRecord, error)
	Tools(ctx context.Context, runID, hash string) (ToolsRecord, error)
	Catalogs(ctx context.Context, runID string) ([]ToolsRecord, error)
	RunSpans(ctx context.Context, runID string) ([]Span, error)
	// OtherLogs pages a run's app log records — the non-weft records
	// the writers store beside weft's (other_logs on SQLite, otel_logs
	// on ClickHouse), attributed to the run through the spans they were
	// emitted under — in time order (LogQuery, ReadOtherLogs), with
	// what the read could not show (LogPage: partial while running,
	// lines under a never-stored span, the candidate cap).
	// ErrNotFound for an unknown run; empty Logs, not nil, for a run
	// with no app logs; a *HoleError (Kind "logs", HoleNotRecorded) for
	// a finished run with no span to attribute them through.
	OtherLogs(ctx context.Context, runID string, q LogQuery) (LogPage, 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(r Record) Frame

func RunFrame(r RunRow) Frame

type Hole ¶ added in v0.10.0

type Hole string

Hole is one badge of ADR 0028 §11's closed table: the one word a reader shows wherever a record is missing, cut or derived. The vocabulary is closed — a new badge is an amendment to that ADR — and Holes lists it. The zero value means no hole.

const (
	HoleTruncated   Hole = "truncated"    // a destination's cap cut the content (weft.content.truncated_bytes)
	HoleStripped    Hole = "stripped"     // the destination's chain is content-off (weft.content = stripped)
	HoleRedacted    Hole = "redacted"     // the destination's Redact changed the content (reserved: weft's pipeline does not mark a redaction)
	HoleMaxTokens   Hole = "max_tokens"   // the step finished on the output token limit
	HoleInterrupted Hole = "interrupted"  // the run stopped reporting (DeriveStatus)
	HoleGap         Hole = "gap"          // a hole in a contiguous counter, or a record a destination dropped
	HoleNotRecorded Hole = "not_recorded" // the record kind did not exist in the version that wrote the run
	HoleDerived     Hole = "derived"      // the value was computed by the reader, not recorded
	HoleHidden      Hole = "hidden"       // the reader's scope may not see it
	HoleCompacted   Hole = "compacted"    // the model saw a compacted view
)

func Holes() []Hole

type HoleCause ¶ added in v0.10.0

type HoleCause string

HoleCause narrows a hole to the cause a reader established, when the table words that cause apart: CauseDefault is the hole's own note.

const (
	CauseDefault HoleCause = ""
	// CauseNoSpans: the run's records arrived but no span did — it ran
	// without a tracer (not_recorded: attempts, the answering model).
	CauseNoSpans HoleCause = "no_spans"
	// CauseResultCap: the core's own tool-result cap cut the content
	// before any destination saw it (truncated: a tool result the model
	// itself read as a prefix plus the marker, weft.MaxResultBytes).
	CauseResultCap HoleCause = "result_cap"
	// CauseLogCap: the app-log reader scanned its candidate cap
	// (MaxLogCandidates) before attributing lines to the run, so later
	// lines may be missing (truncated: no destination setting changes
	// it).
	CauseLogCap HoleCause = "log_cap"
	// CauseNoPublicID: the session's turns carry no weft.public_id —
	// a thread session created without thread.PublicID, or an app that
	// never stamped one (not_recorded: Studio's
	// GET /api/sessions/{id}/public_id).
	CauseNoPublicID HoleCause = "no_public_id"
	// CauseDevTokenOnly: the route answers the dev (server) token and
	// setup A's open API only; a panel token, scoped to one public id,
	// may not learn another (hidden: the session-to-public-id lookup).
	CauseDevTokenOnly HoleCause = "dev_token_only"
	// CauseNotServed: the Studio asked serves no request record (its
	// requests capability is off, or it predates the record), so the
	// run's request, prompt and tools records were not read — the run
	// may hold them (not_recorded: the panel's Request tab).
	CauseNotServed HoleCause = "not_served"
	// CauseResponseCap: the route bounds how much one response reads
	// (Studio's GET /api/diff compares a bounded number of steps), and
	// the run holds more than it read (truncated: nothing was lost; the
	// rest is a read away).
	CauseResponseCap HoleCause = "response_cap"
)

func HoleCauses() []HoleCause

type HoleError ¶ added in v0.10.0

type HoleError struct {
	Kind string // "prompt", "tools" or "logs"
	Hash string
	Hole Hole
}

HoleError is the ErrNotFound DB.Prompt and DB.Tools return when the reader knows why the record is missing: errors.Is(err, ErrNotFound) holds, and errors.As yields the badge. Hole is one of:

• HoleNotRecorded: the run was written before ADR 0028 (its InstructionsHash is ""), so no such record ever existed;
• HoleStripped: a request naming the hash came through a content-off chain, which drops prompt and tools records;
• HoleGap: a request naming the hash was stored as emitted, yet the record it names is absent — a destination dropped it.

A hash no request of the run names is a plain ErrNotFound.

DB.OtherLogs returns one with Kind "logs", no Hash and HoleNotRecorded for a run that has no span to attribute app logs through (ReadOtherLogs).

func (e *HoleError) Error() string

func (e *HoleError) Is(target error) bool

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(opts ...HubOption) Hub

type HubOption ¶

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

HubOption configures NewHub.

func QueueSize(n int) HubOption

type LogCandidates ¶ added in v0.10.0

type LogCandidates func(ctx context.Context, traceIDs []string, from, to time.Time, limit int) ([]OtherLog, error)

LogCandidates is a backend's half of DB.OtherLogs: the non-weft log records (no weft.run.id) of the given traces whose time is within [from, to]: the first limit of them, returned in time order (ties in a stable order of the backend's). ReadOtherLogs does the rest.

type LogPage ¶ added in v0.10.0

type LogPage struct {
	Logs      []OtherLog
	Partial   bool
	Gap       int
	Truncated bool
}

LogPage is one DB.OtherLogs answer: the page and what the reader knows about what it could not show.

• Partial: the run is running. Spans are exported when they end, so a line under an in-flight span (invoke_agent, an open execute_tool) is not attributable yet: it appears once its span arrives, and it may then take an index below lines already served — indexes are stable only once the run has ended.
• Gap: lines in the run's traces (deduplicated, in its window) that name a span no stored span has — dropped (a lost batch, a process that exited before exporting), or still open (an app span that encloses the run and has not ended). They may belong to this run or to another in the same trace, and are not in Logs. Always 0 while Partial (the span may still be on its way).
• Truncated: the run's traces hold more than MaxLogCandidates log lines in its window — anyone's, the cap applies before attribution — so only lines among the first MaxLogCandidates by time were read: later lines of this run may be missing, and Logs may hold far fewer than MaxLogCandidates.

func ReadOtherLogs(ctx context.Context, db DB, runID string, q LogQuery, candidates LogCandidates) (LogPage, error)

type LogQuery ¶ added in v0.10.0

type LogQuery struct {
	From        int64
	Limit       int
	MinSeverity int
}

LogQuery selects a run's app log records for DB.OtherLogs. The zero value reads every record from index 0, 100 at a time. From is the first index read (inclusive): pass the last returned Index + 1 to continue; a page shorter than PageLimit is the last. MinSeverity, when above 0, keeps the records whose OTLP severity number is at least that (ParseSeverity reads a level name); Index is the record's place among all of the run's app logs, so a filtered page's indexes skip and From continues a filtered walk alike. Limit: 0 = 100, max 1000.

func (q LogQuery) PageLimit() int

type OtherLog ¶ added in v0.10.0

type OtherLog struct {
	Index     int64
	Time      time.Time
	Severity  int
	EventName string
	Body      string
	Attrs     map[string]any
	TraceID   string
	SpanID    string
	Service   string
}

OtherLog is one app log record of a run: a non-weft record (no weft.run.id — the app's own slog or OTel Logs API lines, which the writers keep in other_logs on SQLite and in otel_logs on ClickHouse) attributed to the run through the span it was emitted under (DB.OtherLogs). Index is its place in the run's app logs, in time order; Severity the OTLP severity number (0 unspecified, 1–24; SeverityText names it). Attrs are the record's attributes: typed on SQLite, strings on ClickHouse, whose otel_logs keeps the collector's string map only.

type PosEvent ¶

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

PosEvent is one positioned event of a run: the body verbatim, at its durable position, with the two content attributes the destination's chain stamped on it (ADR 0028 §11): Content is weft.content ("stripped" from a content-off chain, "" as emitted) and TruncatedBytes weft.content.truncated_bytes (what a content-on chain's cap cut, 0 for none). A ClickHouse row written before migration 0004 carries neither.

type PromptRecord ¶ added in v0.10.0

type PromptRecord struct {
	Index          int64
	Time           time.Time
	Hash           string
	Text           string
	Content        Hole
	TruncatedBytes int64
}

PromptRecord is one stored prompt record (ADR 0028 §4): the composed system text of one distinct system hash. Index is weft.prompt.index. Content is HoleTruncated when a destination's cap cut Text (TruncatedBytes > 0), HoleDerived when the body did not parse (a malformed producer; Hash is then the weft.system.hash attribute). Hash is over the text before the cut.

func PromptRecordOf(r StoredRecord) PromptRecord

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(req *collogspb.ExportLogsServiceRequest) []Record

type RequestBody ¶ added in v0.10.0

type RequestBody struct {
	Step            int                `json:"step"`
	Attempt         int64              `json:"attempt"`
	SystemHash      string             `json:"system_hash"`
	MessagesRef     RequestMessagesRef `json:"messages_ref"`
	Tools           RequestTools       `json:"tools"`
	ToolChoice      *RequestToolChoice `json:"tool_choice,omitempty"`
	Thinking        *RequestThinking   `json:"thinking,omitempty"`
	SequentialTools bool               `json:"sequential_tools"`
	Params          RequestParams      `json:"params"`
	Model           RequestModel       `json:"model"`
	Stream          bool               `json:"stream"`
}

RequestBody is the request record's body (ADR 0028 §3), keys as the core writes them; an absent field is unset.

type RequestMessagesRef ¶ added in v0.10.0

type RequestMessagesRef struct {
	Index *int64 `json:"index,omitempty"`
	Count int    `json:"count"`
}

RequestMessagesRef says which messages the request carried: the run's view as of messages record Index, Count messages long. Index is nil where no messages record exists (capture off).

type RequestModel ¶ added in v0.10.0

type RequestModel struct {
	Provider string `json:"provider,omitempty"`
	Name     string `json:"name,omitempty"`
}

RequestModel is the model the attempt requested.

type RequestParams ¶ added in v0.10.0

type RequestParams struct {
	Temperature *float64 `json:"temperature,omitempty"`
	TopP        *float64 `json:"top_p,omitempty"`
	MaxTokens   *int     `json:"max_tokens,omitempty"`
	Stop        []string `json:"stop,omitempty"`
	Seed        *int64   `json:"seed,omitempty"`
}

RequestParams are the sampling parameters; nil is the adapter's default.

type RequestQuery ¶ added in v0.10.0

type RequestQuery struct {
	Step  *int
	From  int64
	Limit int
}

RequestQuery selects a run's request records for DB.Requests, in request-index order. The zero value reads every step from index 0, 100 at a time. Step, when set, keeps step *Step's attempts only. From is the first index read (inclusive): pass the last returned Index + 1 to continue; a page shorter than PageLimit is the last. Limit: 0 = 100, max 1000.

func (q RequestQuery) PageLimit() int

type RequestRecord ¶ added in v0.10.0

type RequestRecord struct {
	Index          int64
	Step           int
	Attempt        int64
	Time           time.Time
	SystemHash     string
	CatalogHash    string
	Content        Hole
	TruncatedBytes int64 // informational only; a request record is never capped by weft
	Body           RequestBody
	Raw            json.RawMessage
}

RequestRecord is one stored request record (ADR 0028 §3): one model call attempt. Index is weft.request.index, Step the stored weft.step.index (-1 when absent), Attempt the body's attempt. The record's weft.system.hash and weft.catalog.hash attributes win over the body's system_hash and tools.catalog_hash when set and they disagree; SystemHash and CatalogHash fall back to the body's where the attribute is absent. Raw is the body verbatim.

Content is the record's hole, from ADR 0028 §11's table:

• "": stored as emitted;
• HoleStripped: it came through a content-off chain (weft.content = stripped: its params.stop and messages_ref.index were removed, and that chain dropped the prompt and tools records it names) — whether or not its body parses;
• HoleDerived: a malformed producer — an unstripped record whose body did not parse, so Body is zero, Attempt is 0, and the two hashes come from the record's attributes.

A request record is never capped (only prompt and tools records are), and a destination's Redact over params.stop is not marked on the record: HoleRedacted is reserved for the readers that can tell (ADR 0028 §11, A3).

func RequestRecordOf(r StoredRecord) RequestRecord

type RequestThinking ¶ added in v0.10.0

type RequestThinking struct {
	Level  string `json:"level"`
	Budget int64  `json:"budget,omitempty"`
}

RequestThinking is the thinking configuration in force.

type RequestToolChoice ¶ added in v0.10.0

type RequestToolChoice struct {
	Mode string `json:"mode"`
	Name string `json:"name,omitempty"`
}

RequestToolChoice is the tool choice in force.

type RequestTools ¶ added in v0.10.0

type RequestTools struct {
	CatalogHash string   `json:"catalog_hash"`
	Names       []string `json:"names"`
}

RequestTools is the catalog a request offered: its hash ("" for no tools) and the names in the order offered.

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)
	// ADR 0028 §10: run_start's weft.instructions.hash ("" on a run
	// written before the request record — RequestsHole reads the
	// table), request index 0's weft.catalog.hash ("" when it offered
	// no tools) and the request records' high-water mark (max
	// weft.request.index + 1). The API names them instructions_hash,
	// catalog_hash and request_count.
	InstructionsHash, CatalogHash string
	RequestCount                  int64
}

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.

func (r RunRow) RequestsHole() Hole

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(req *coltracepb.ExportTraceServiceRequest) []Span

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(spanFailed, runFinished bool, lastSeen, now time.Time) Status

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

type StepMessages ¶ added in v0.13.0

type StepMessages struct {
	Step     int
	Messages []core.Message
	View     *Compaction
	Derived  bool
}

StepMessages is the messages one step's model call carried — what the model saw at that step (ADR 0028 §8, ADR 0029): the replay prefix for from_step Step, the "compacted messages" a fixture keys on. Step is the step asked for. View is the run-scope compaction view the step's request named (its messages_ref.index), nil when the request carried the plain transcript; Messages is then the growth records up to the view's index with its range replaced, else the growth records up to the request's own index. Derived is true when no request record placed the messages — a run written before ADR 0028, or a request whose messages_ref a content-off chain stripped — and the from_step cut rule did instead: the transcript before step Step's assistant message (no view can apply there: nothing names one).

func AssembleStep(batches []TranscriptBatch, requests []RequestRecord, compactions []Compaction, step int) (StepMessages, error)

func MessagesAsOf(ctx context.Context, db DB, runID string, step int) (StepMessages, error)

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/weftgo/weft/obsdb"
	"github.com/weftgo/weft/obsdb/sqlite"
)

func main() {
	db, err := sqlite.Open(":memory:")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = db.Close() }()
	ctx := context.Background()
	at := time.Unix(0, 1790845923120000000).UTC()
	text := func(role, t string) string {
		return `{"role":"` + role + `","content":[{"type":"text","text":"` + t + `"}]}`
	}
	msgs := func(index, step int64, body string, extra map[string]any) obsdb.Record {
		attrs := map[string]any{"weft.record": "messages", "weft.run.id": "run_1",
			"weft.messages.index": index, "weft.step.index": step}
		for k, v := range extra {
			attrs[k] = v
		}
		return obsdb.Record{Time: at, EventName: "weft.messages", Body: body, Attrs: attrs}
	}
	request := func(index, step int64, ref int64, count int) obsdb.Record {
		return obsdb.Record{Time: at, EventName: "weft.request",
			Body: fmt.Sprintf(`{"step":%d,"attempt":1,"messages_ref":{"index":%d,"count":%d},"tools":{"catalog_hash":"","names":[]},"params":{},"model":{"name":"m"}}`, step, ref, count),
			Attrs: map[string]any{"weft.record": "request", "weft.run.id": "run_1",
				"weft.request.index": index, "weft.step.index": step, "weft.attempt.index": int64(1)}}
	}
	err = db.Write(ctx, obsdb.Batch{Records: []obsdb.Record{
		{Time: at, EventName: "weft.event", Body: `{"type":"run_start","id":"run_1"}`,
			Attrs: map[string]any{"weft.record": "event", "weft.run.id": "run_1", "weft.event.type": "run_start", "weft.event.pos": int64(0)}},
		msgs(0, 0, "["+text("user", "u1")+"]", map[string]any{"weft.messages.input": true}),
		request(0, 0, 0, 1),
		msgs(1, 0, `[{"role":"assistant","content":[{"type":"tool_call","id":"c1","name":"lookup","args":{}}]}]`, nil),
		msgs(2, 0, `[{"role":"tool","content":[{"type":"tool_result","call_id":"c1","name":"lookup","content":"t1"}]}]`, nil),
		msgs(3, 1, "["+text("user", "s")+"]", map[string]any{"weft.messages.reason": "compacted",
			"weft.messages.from_seq": int64(1), "weft.messages.to_seq": int64(3),
			"weft.compaction.scope": "run", "weft.compaction.hash": "h1"}),
		request(1, 1, 3, 2),
	}})
	if err != nil {
		log.Fatal(err)
	}
	for _, step := range []int{0, 1} {
		sm, err := obsdb.MessagesAsOf(ctx, db, "run_1", step)
		if err != nil {
			log.Fatal(err)
		}
		var roles []string
		for _, m := range sm.Messages {
			roles = append(roles, string(m.Role))
		}
		fmt.Println(step, roles, sm.View != nil)
	}
	_, err = obsdb.MessagesAsOf(ctx, db, "run_1", 5)
	fmt.Println(errors.Is(err, obsdb.ErrNotFound))
}

Output:
0 [user] false
1 [user user] true
true

type StepMessagesError ¶ added in v0.13.0

type StepMessagesError struct {
	Step int
	Hole Hole
	Msg  string
}

StepMessagesError is MessagesAsOf's and AssembleStep's refusal, with the hole that explains it (ADR 0028 §11's table): HoleGap when a record the step's request names is missing or does not fit — a growth record below its index, the growth record its plain ref names, a view that does not fit or was never written (a rewritten request with no view index), a body that is not messages — and HoleStripped when the run stored no messages at all (content capture off, or a content-off chain). errors.Is(err, ErrStepMessages) holds; a reader badges Hole and never guesses.

func (e *StepMessagesError) Error() string

func (e *StepMessagesError) Is(target error) bool

type StoredRecord ¶ added in v0.10.0

type StoredRecord struct {
	Index          int64
	Step           int
	Time           time.Time
	Body           []byte
	Content        string
	TruncatedBytes int64
	SystemHash     string
	CatalogHash    string
}

StoredRecord is one request, prompt or tools record as a backend stored it: its index, stored step (-1 when absent), time and body, and the four attributes the readers need — weft.content (the mark), weft.content.truncated_bytes, weft.system.hash and weft.catalog.hash. A backend fills it from its columns (or its attribute column) and builds the read type through RequestRecordOf, PromptRecordOf or ToolsRecordOf, so every backend parses alike.

type ToolEntry ¶ added in v0.10.0

type ToolEntry struct {
	Name           string          `json:"name"`
	Description    string          `json:"description"`
	Schema         json.RawMessage `json:"schema"`
	TimeoutMS      int64           `json:"timeout_ms"`
	Approval       bool            `json:"approval"`
	Replay         string          `json:"replay"`
	MaxResultBytes int             `json:"max_result_bytes"`
	Sequential     bool            `json:"sequential"`
	Source         string          `json:"source"`
}

ToolEntry is one tool of a catalog: the model-visible triple and the policy chips (ADR 0028 §5).

type ToolsRecord ¶ added in v0.10.0

type ToolsRecord struct {
	Index          int64
	Time           time.Time
	Hash           string
	Tools          []ToolEntry
	Content        Hole
	TruncatedBytes int64
}

ToolsRecord is one stored tools record (ADR 0028 §5): the catalog of one distinct catalog hash, tools in name order. Content is HoleTruncated when a destination's cap dropped entries from the end (TruncatedBytes > 0), HoleDerived when the body did not parse (a malformed producer; Hash is then the weft.catalog.hash attribute). Hash is over the whole catalog.

func ToolsRecordOf(r StoredRecord) ToolsRecord

type TranscriptBatch ¶ added in v0.10.0

type TranscriptBatch struct {
	Index        int64
	Step         int
	Input        bool
	InputDerived bool // Input was inferred by the backend, not read from the record
	Messages     json.RawMessage
}

TranscriptBatch is one messages record as stored (ADR 0028 §8): readers place a batch by its Step, never by its neighbours. Step is -1 when the record carried no weft.step.index — a ClickHouse row written before migration 0004, or a producer that never stamped it; a reader that places such a batch anyway infers the step and says so (the derived badge, ADR 0028 §11). Input is the record's weft.messages.input: the input record — on a partial resume a prefix of what the run was fed, ending at the last assistant message with tool calls; the tail (the rebuilt tool message and what follows it) is the next batch, at step 0, not flagged. A backend that did not store the flag for a row (ClickHouse rows written before weft_records.Input existed) infers Input on index 0 — true unless the body is a lone assistant message, which is step 0 of a run fed no messages (the core writes the input record first, and none for an empty input) — and sets InputDerived on that batch so readers badge it HoleDerived.

func DedupBatches(batches []TranscriptBatch) []TranscriptBatch

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, delta pos, or request/prompt/tools index
	Step                             int   // -1 when absent
	ToolSeq                          int64 // -1 when absent
	Playground                       bool
	ExperimentID, ForkedFrom         string
	// Reason is a messages record's weft.messages.reason: "" for
	// transcript growth, "compacted" for a compaction view (ADR 0028
	// §8), which is never counted or concatenated as transcript.
	Reason string
}

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

func DeriveRecord(r Record) Weft

func DeriveSpan(s Span) Weft