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
	// Traces lists traces — weft and non-weft alike, one row per trace
	// id the spans table holds — newest first (TraceQuery, TracePage):
	// the row is computed from the trace's spans alone, so a trace
	// whose root never arrived is listed with Root nil.
	Traces(ctx context.Context, q TraceQuery) (TracePage, error)

	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)

	// Deletion and retention (plan A7, §13.13). Every one touches the
	// observability tables alone — never another module's tables
	// sharing the file (thread's sessions in one SQLite file stay
	// byte-identical) and never the experiments table (its definitions
	// name runs, they do not hold them).
	//
	// DeleteRun removes a run and its subagent children — every run
	// whose ParentRunID chain reaches id — with everything stored under
	// them: records (deltas kept by KeepDeltas included), the search
	// index rows, the spans named by DeletedSpans and the app log
	// records emitted under those spans. ErrNotFound for an unknown id.
	// A run still being written is deleted as it stands: a later write
	// for it starts a new run row from what that write carries.
	DeleteRun(ctx context.Context, id string) error
	// DeleteSession is DeleteRun for every run carrying the session id
	// (weft.session.id), children included. The app's own conversation
	// store is not obsdb's and is never touched. ErrNotFound when no
	// run carries the id.
	DeleteSession(ctx context.Context, id string) error
	// Sweep deletes, as DeleteRun does, the runs whose LastSeen is
	// before the cutoff and whose status is not running (succeeded,
	// failed or interrupted — a run silent past InterruptedAfter is
	// interrupted, so it counts as not running and is swept), with
	// their children that are not running either; a running child keeps
	// its own subtree and is swept on its own once it ends. Until then
	// that child of a swept parent is listed only under RunQuery's
	// ParentRunID "*" (the API's all=1 / parent=*), never among the
	// top-level runs. It works in passes of at most SweepBatch roots
	// (each with its subtree), so a large backlog never holds the
	// writer for the whole sweep; ctx is checked between passes. It
	// returns how many runs it deleted, children counted, over every
	// pass — on an error, the passes before it stay deleted.
	Sweep(ctx context.Context, before time.Time) (int, error)
	// Vacuum gives the space deletions freed back: SQLite runs VACUUM
	// (the whole file is rewritten, other modules' tables with their
	// contents unchanged); ClickHouse runs OPTIMIZE TABLE … FINAL on
	// the tables deletions mutate, which merges every part — an
	// operator's command, heavy on a large database.
	Vacuum(ctx context.Context) 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
	// ExperimentCommand is the rest of the §5.1 command the matrix
	// sent, so a saved experiment reopens as the form that ran it.
	// Zero on an experiment saved before obsdb kept it.
	ExperimentCommand
}

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 ExperimentCommand ¶ added in v0.14.0

type ExperimentCommand struct {
	Engine          string          `json:"engine,omitempty"`           // live | scripted
	SideEffects     string          `json:"side_effects,omitempty"`     // substitute | park | allow
	Thread          string          `json:"thread,omitempty"`           // ephemeral | fork
	FromStep        *int            `json:"from_step,omitempty"`        // the step every cell continues from
	SourceRunID     string          `json:"source_run_id,omitempty"`    // the run the cells replay from
	TranscriptEdits json.RawMessage `json:"transcript_edits,omitempty"` // the list as sent
}

ExperimentCommand is the experiment-wide part of the §5.1 run command, kept as sent: the modes, the step the cells continue from, the source run and the transcript edits. Every field is optional — zero is "not recorded" (an older row) or the command's default, and a reader does not tell the two apart. obsdb stores it as one JSON document beside the definition and does not validate it (Studio does, on the way in).

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"`
	Engine      string          `json:"engine,omitempty"`
	SideEffects string          `json:"side_effects,omitempty"`
	Thread      string          `json:"thread,omitempty"`
}

ExperimentVariant is one column of the matrix: a key, the overrides object it applies (verbatim — instructions, tools_enabled, model, thinking, options, params, tool_choice, park_on, only_tools, and any key a later command adds), and the modes it runs under when they differ from the experiment's (empty: the experiment's own).

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"
	// CauseRawOff: the attempt has no raw record (ADR 0028 §12) — wire
	// bodies are recorded only under otel.RawBodies(true) with
	// WEFT_ENV=dev, by an adapter that reports them (not_recorded: the
	// requests row's raw, the wire tab).
	CauseRawOff HoleCause = "raw_off"
)

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 MessageText ¶ added in v0.14.0

type MessageText struct {
	Role, Text string
	ToolCall   bool
}

MessageText is one message of a messages record body reduced to what search reads: its role and the text of its text parts, joined by "\n" (every part of type "text", an empty one included, in order), and whether it carries a tool call (a part of type "tool_call").

func MessagesText(body []byte) []MessageText

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 RawRecord ¶ added in v0.14.0

type RawRecord struct {
	Index           int64     `json:"index"`
	Step            int       `json:"step"`
	Attempt         int64     `json:"attempt"`
	Time            time.Time `json:"time"`
	Request         RawSide   `json:"request"`
	Response        *RawSide  `json:"response"`
	RedactedHeaders []string  `json:"redacted_headers"`
	RedactedQuery   []string  `json:"redacted_query,omitempty"`
	Content         Hole      `json:"content"`
	TruncatedBytes  int64     `json:"truncated_bytes"`
}

RawRecord is one stored raw record: the attempt's wire exchange, as the core shaped it — credential headers stripped and named in RedactedHeaders, credential query values replaced and named in RedactedQuery, each body capped at 256 KiB with the cut counted. Index is weft.raw.index, Step the stored weft.step.index, Attempt the body's attempt (the attempt in flight when the adapter reported). TruncatedBytes is weft.content.truncated_bytes, both bodies' cut; Content is HoleTruncated when it is above 0, HoleDerived when the body did not parse (a malformed producer: the sides are zero).

func RawRecordOf(r StoredRecord) RawRecord

type RawSide ¶ added in v0.14.0

type RawSide struct {
	Method         string              `json:"method,omitempty"`
	URL            string              `json:"url,omitempty"`
	Status         int                 `json:"status,omitempty"`
	Headers        map[string][]string `json:"headers"`
	MediaType      string              `json:"media_type,omitempty"`
	Body           string              `json:"body"`
	Encoding       string              `json:"encoding,omitempty"`
	Bytes          int64               `json:"bytes"`
	TruncatedBytes int64               `json:"truncated_bytes,omitempty"`
	// Complete is the response side's: false when the body was not read
	// to its end (closed unread by a transport retry, a timeout, a
	// reader that stopped) — the body is what was read, never the whole;
	// nil on the request side and on records that do not say.
	Complete *bool `json:"complete,omitempty"`
}

RawSide is one side of the exchange. Body is the bytes as text, or base64 when Encoding is "base64" (the bytes were not UTF-8); Bytes is the whole body's size before the cap, TruncatedBytes what the cap removed. Method and URL are the request's, Status the response's (absent on the other side). A nil Response is an exchange that got no response (a transport error).

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
	Wire           *RawRecord
}

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

Wire is the attempt's raw record (ADR 0028 §12: the provider request and response on the wire, recorded only under otel.RawBodies in development), nil when the run holds none for this attempt (AttachRaw).

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
	// LastSeenAfter keeps the runs last seen at or after it (zero = no
	// bound). A running run is always among them once the bound is
	// older than InterruptedAfter: running means seen that recently.
	LastSeenAfter time.Time
	// 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
	// PromptPreview is the run's prompt — the turn's own, PromptText's
	// rule with TurnPrompt's resume clause (Searcher's prompt field) — cut to
	// PromptPreviewRunes runes; "" when the records carried no content
	// (a content-off run stores no messages) or no user text.
	PromptPreview string
}

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 SearchQuery ¶ added in v0.14.0

type SearchQuery struct {
	RunQuery
	Text, Tool, Model, Error string
	Since, Until             time.Time
	MinTokens                int64
	Sort                     SearchSort
	Offset                   int
}

SearchQuery selects runs for Search. Its RunQuery carries the filters /api/runs reads — agent, status, session, public id (the panel token's scope), parent, playground, experiment, metadata — with the same meaning, so a scope applied to one applies to the other; RunQuery.Before and BeforeID are not read (a search pages by Offset) and RunQuery.Limit is the page size (0 = 50, max 500).

The search fields:

• Text: the search terms (ParseSearch's grammar). A run matches when every term occurs, case-insensitively, as a substring of at least one of the run's search fields — its prompt (the turn's own: PromptText), its answer (the text of its last assistant message that carries text), its error, or the name of a tool it called (tool_start events) — each term free to match a different field. A content-off run has no prompt or answer: only its error and tool names can match.
• Tool: the exact name of a tool the run called.
• Model: the exact model name the run row carries.
• Error: a case-insensitive substring of the run's error.
• Since, Until: Started ≥ Since, Started < Until (zero = open).
• MinTokens: input + output tokens ≥ MinTokens (0 = any).
• Sort: the order (SearchSort; "" is SortStarted).
• Offset: rows to skip in that order.

type SearchSort ¶ added in v0.14.0

type SearchSort string

SearchSort is a search's order. Every order breaks ties by Started, then run id, both descending, so a result is total and stable.

const (
	SortStarted  SearchSort = "started"  // newest first (the default)
	SortDuration SearchSort = "duration" // longest first: Finished (LastSeen while unfinished) − Started
	SortTokens   SearchSort = "tokens"   // most input + output tokens first
)

type Searcher ¶ added in v0.14.0

type Searcher interface {
	// Search answers the runs SearchQuery matches, sorted as it asks,
	// with Total counting the whole match. NextBefore and NextBeforeID
	// stay nil and "": a search pages by SearchQuery.Offset.
	Search(ctx context.Context, q SearchQuery) (RunPage, error)
}

Searcher is the optional search capability of a DB (plan A7): runs found by what they said and did, not only by who ran them. Both first-party backends implement it — obsdb/sqlite through an FTS5 trigram index maintained on write, obsdb/clickhouse through case-insensitive substring scans over the stored records — and the obsdbtest conformance table holds them to the same answers. A backend that does not implement it is a database that cannot search: Studio then leaves its search capability off and says why.

package main

import (
	"context"
	"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() }()
	t0 := time.Unix(0, 1790845923120000000).UTC()
	write := func(run, prompt, answer string, at time.Duration) {
		rec := func(idx int64, body string) obsdb.Record {
			return obsdb.Record{Time: t0.Add(at), Body: body, Attrs: map[string]any{
				"weft.record": "messages", "weft.run.id": run, "weft.messages.index": idx}}
		}
		if err := db.Write(context.Background(), obsdb.Batch{Records: []obsdb.Record{
			rec(0, `[{"role":"user","content":[{"type":"text","text":"`+prompt+`"}]}]`),
			rec(1, `[{"role":"assistant","content":[{"type":"text","text":"`+answer+`"}]}]`),
		}}); err != nil {
			log.Fatal(err)
		}
	}
	write("run_1", "Refund order 7", "The refund failed: card declined.", 0)
	write("run_2", "Refund order 9", "Refunded.", time.Second)

	page, err := db.(obsdb.Searcher).Search(context.Background(), obsdb.SearchQuery{Text: "refund failed"})
	if err != nil {
		log.Fatal(err)
	}
	for _, r := range page.Runs {
		fmt.Println(r.ID, "-", r.PromptPreview)
	}
}

Output:
run_1 - Refund order 7

type Selector ¶

type Selector struct {
	RunID, SessionID, PublicID, Agent string
	// All matches every run's frames (subagent children included; a
	// record that names no run is no run's, and is not matched). It is
	// a scope of its own: set with any other field, the selector is
	// over-set and matches nothing.
	All bool
}

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. All is the explicit every-run scope (Studio's Live page): it must be asked for, never implied.

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 SpanKey ¶ added in v0.14.0

type SpanKey struct {
	TraceID, SpanID string
}

SpanKey names one span.

func DeletedSpans(all []SpanLink, deleted map[string]bool) []SpanKey

type SpanLink ¶ added in v0.14.0

type SpanLink struct {
	TraceID, SpanID, ParentSpanID, RunID string
}

SpanLink is one stored span's place in its trace, what DeletedSpans walks: its key, its parent and the weft.run.id it carries ("" for a non-weft span).

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 TracePage ¶ added in v0.14.0

type TracePage struct {
	Traces       []TraceRow
	Total        int
	NextBefore   *time.Time
	NextBeforeID string
}

TracePage is one Traces result, paged as RunPage is: newest Start first (ties by TraceID, descending), NextBefore/NextBeforeID are the next TraceQuery.Before/BeforeID (nil and "" on the last page), traces sharing the last row's Start ride along up to MaxTies, and Total counts the whole match, cursor aside. A trace's Start is its earliest span, so a trace whose earlier span arrives between page reads can be listed again on a later page (never skipped); clients dedupe by TraceID.

type TraceQuery ¶ added in v0.14.0

type TraceQuery struct {
	// Service keeps the traces holding at least one span whose resource
	// service name is exactly this ("" any).
	Service string
	// PublicID keeps the traces every span of which carries
	// weft.public_id = PublicID — the rule a panel token's
	// /api/traces/{id} read applies, so a scoped list offers no trace
	// its detail would refuse (a non-weft span or another public id's
	// run in the trace excludes it).
	PublicID string
	Since    time.Time // Start at or after; zero = no bound
	Until    time.Time // Start before; zero = no bound
	Before   time.Time // cursor on Start; zero = newest
	// BeforeID, with Before, makes the cursor the pair (Start, TraceID),
	// as RunQuery.BeforeID: pass TracePage.NextBeforeID. A late earlier
	// span can re-list a trace on a later page (TracePage).
	BeforeID string
	Limit    int // 0 = 50, max 500
}

TraceQuery selects traces for Traces. The zero value lists every trace, newest first, 50 at a time. A trace's time is its Start: the earliest start of the spans it holds.

type TraceRoot ¶ added in v0.14.0

type TraceRoot struct {
	SpanID, Name string
}

TraceRoot is a trace's root span: its id and name.

type TraceRow ¶ added in v0.14.0

type TraceRow struct {
	TraceID    string
	Root       *TraceRoot
	Service    string
	Start, End time.Time
	SpanCount  int
	Error      bool
	RunIDs     []string
}

TraceRow is one trace in a list, computed from its spans: the time range they cover, how many there are (each (trace, span) once, whatever a retried batch duplicated), whether any ended with status error, the distinct weft.run.id values they carry (sorted: a subagent child shares its parent's trace, so both are here) and the root — the earliest span with no parent span id, nil when none arrived (only child spans were stored: a hole the reader badges). Service is the root's, or the earliest span's when the root is missing.

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. A
	// non-string reason reads as its text (never "": the record is not
	// growth), as ClickHouse's stringified attribute column reads it.
	Reason string
	// Input is a messages record's weft.messages.input: the run's input
	// record (ADR 0028 §8) — what search reads the turn's prompt from.
	Input bool
}

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