store

Documentation

Overview

Package store records weft runs and reads them back.

A RunRecord is what a run did: its identity, its event stream (every weft.Event, Nested inline, in the order they were emitted), and its result — kept on failure too, as the RunError's partial transcript. It is deliberately not a checkpoint: weft records what happened and replays it; resuming is the approval boundary (weft.Approve), not a mid-step rewind. The format, its versioning, and the derivation rules are ADR 0010.

The three moving parts:

• Record is the tap: a weft.Option (composed from weft.Tap and weft.OnRunEnd) that writes every event as it arrives — nothing buffered, so a crash loses nothing that was emitted and a reader can tail a live run — and closes the record at the run's end, with the result and, on failure, the error text.
• Store is the backend interface: Save, Append, Get, List, Delete. Memory is the in-process implementation (tests, examples); the sqlite subpackage is the durable one. Both run the storetest conformance table.
• Query lists runs without their events (a Total comes back, and paging is a Before cursor on Started, never an offset); Get returns everything, and fails loudly on an event type this weft does not know (ErrUnknownEvent) — List still works, so an older reader shows the run and says why it cannot open it.

Status is derived, never free text: rows store running, succeeded, or failed; interrupted is what a reader concludes from a running row whose heartbeat is older than HeartbeatTimeout — a crash leaves evidence, and a live run in another process is not mistaken for a corpse.

Index

• Constants
• Variables
• func MarshalResult(r *weft.RunResult) ([]byte, error)
• func Record(s Store, opts ...RecordOption) weft.Option
• func UnmarshalResult(b []byte) (*weft.RunResult, error)
• type Page
• type Query
• type RecordOption
  • func Tags(kv map[string]string) RecordOption
• type RunRecord
• type Status
  • func DeriveStatus(status Status, heartbeat, now time.Time) Status
• type Store
  • func Memory() Store

Examples

• Record

Constants

const FormatVersion = 1

FormatVersion is the wire version of every document the store writes (ADR 0010 §2.3) — the integer the manifest already uses ("weft": 1). It moves when any stored document changes incompatibly: a renamed or retyped key bumps it; an added optional key never does (readers ignore unknown keys). Backends record it per row and refuse a file whose schema is newer than the binary (sqlite.ErrNewerSchema).

const HeartbeatTimeout = 30 * time.Second

HeartbeatTimeout is how stale a running record's heartbeat may be before a reader concludes the run was interrupted: 30 s by default. The recorder bumps the heartbeat on every event and on a ticker at a third of this timeout while a run is alive but quiet, so a slow tool call never reads as a crash.

Variables

var (
	// ErrNotFound is returned by Get, Append, and Delete for a run id
	// the store does not hold.
	ErrNotFound = errors.New("store: run not found")

	// ErrUnknownEvent wraps the decode failure Get reports for an
	// event whose "type" this build's weft does not know: a recorded
	// stream must replay exactly what was emitted, and an older weft
	// reading a newer recording says so instead of guessing. List —
	// which never reads events — still works, so an older Inspector
	// shows the run and names why it cannot open it.
	ErrUnknownEvent = errors.New("store: event type unknown to this weft")

	// ErrNewerFormat wraps the decode failure UnmarshalResult reports
	// for a result document whose {"weft": N} envelope is ahead of
	// FormatVersion: the document came from a newer weft, and decoding
	// it with this build's tags would misread it (ADR 0010 §2.3 — the
	// number moves only on an incompatible change). The schema
	// migrations enforce the same rule for the database as a whole at
	// Open (sqlite.ErrNewerSchema); this is the document-level half.
	ErrNewerFormat = errors.New("store: document format newer than this weft")
)

Functions

func MarshalResult

func MarshalResult(r *weft.RunResult) ([]byte, error)

MarshalResult encodes a RunResult as the store's result document. A nil result encodes as null bytes ("null"), which UnmarshalResult reads back as nil.

func Record

func Record(s Store, opts ...RecordOption) weft.Option

Record returns a weft.Option that records every run of the agent it is installed on into s — install it on every agent of a fleet, since a subagent's child records itself (see below).

• The row is written at RunStart: running, with the identity ADR 0010 defines — agent name and model from the event, the manifest hash (sha256 of weft.Manifest(agent), "" when unnamed), the core version, and the Tags. A crash leaves this row plus whatever events landed: evidence, not silence.
• Every event is one Append as it arrives, Nested inline — nothing is buffered, so a crash loses nothing that was emitted and a reader can tail a live run. The append runs synchronously on the emitting goroutine (pi's cadence; both surveyed paces work at agent event rates) and bumps the heartbeat; a quiet-but-alive run (a long tool call) is kept alive by a ticker at a third of HeartbeatTimeout, so a slow tool never reads as a crash.
• The closing write goes through OnRunEnd — the failure signal a tap cannot carry, because a failed run emits nothing after its last delivered event (ADR 0004). On success: succeeded, the result, the finish time. On failure: failed, the RunError's partial transcript as the result, and its text in Err.

Child runs: a subagent's child records itself. Its tap sees its own unwrapped events; its OnRunEnd fires inside the parent's tool call and closes its record with its own result; and the parent's tool call rides the child's context, so the child's row learns ParentID and ParentCallID from weft.CallFromContext — no run-id parsing. The parent's own stream keeps the child's events inline as Nested, so the parent's record replays as one stream (ADR 0004's order is the record's order). One writer per run id: the parent never writes the child's rows.

Store errors are logged through the agent's logger and never fail the run — observation is not the loop's business (ADR 0006). After a failed write the record is marked degraded and the closing write carries Err prefixed with the store's error, so a gap is visible in the record, not only in the log. Record never retries a write: an event lost by a failed append stays lost, and the record says so.

Example

Record captures every run into the store — install it on every agent of a fleet; a subagent's child records itself and links back to the parent's call.

package main

import (
	"context"
	"fmt"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/store"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	s := store.Memory()
	agt := weft.New(
		wefttest.Script(
			wefttest.ToolCalls(wefttest.Call{Name: "roll_dice"}),
			wefttest.Say("rolled a 4"),
		),
		weft.Name("dice"),
		store.Record(s, store.Tags(map[string]string{"room": "table-1"})),
		weft.Tool("roll_dice", "Roll a die.",
			func(_ context.Context, _ struct{}) (int, error) { return 4, nil }),
	)
	if _, err := agt.Generate(context.Background(), weft.Prompt("Roll.")); err != nil {
		fmt.Println(err)
		return
	}
	page, err := s.List(context.Background(), store.Query{})
	if err != nil {
		fmt.Println(err)
		return
	}
	rec := page.Runs[0]
	fmt.Printf("agent=%s tags=%v steps=%d tokens=%d status=%s\n",
		rec.Agent, rec.Tags, rec.Steps, rec.Usage.Total(), rec.Status)
	full, _ := s.Get(context.Background(), rec.ID)
	for _, ev := range full.Events {
		switch e := ev.(type) {
		case weft.RunStart:
			fmt.Printf("start agent=%s\n", e.Agent)
		case weft.ToolStart:
			fmt.Printf("tool %s\n", e.Name)
		case weft.RunFinish:
			fmt.Printf("finish steps=%d\n", e.Steps)
		}
	}
}

Output:
agent=dice tags=map[room:table-1] steps=2 tokens=30 status=succeeded
start agent=dice
tool roll_dice
finish steps=2

func UnmarshalResult

func UnmarshalResult(b []byte) (*weft.RunResult, error)

UnmarshalResult decodes a result document written by MarshalResult (or by any writer of the format — the tags are the contract). The envelope's version integer must equal FormatVersion: a document from a newer weft fails with ErrNewerFormat instead of being decoded with this build's tags (ADR 0010 §2.3), and a document with a result but no matching envelope was never written by this format — both loud, never a silent misread.

Types

type Page

type Page struct {
	// Runs is the page, newest first by Started, Events omitted — a
	// list body with full transcripts is the storage bloat every
	// surveyed store had to walk back (Mastra's listTracesLight).
	// Get returns the events.
	Runs []RunRecord
	// Total is the number of rows matching the query, ignoring Before
	// and Limit — the "how many pages are there" number.
	Total int
}

Page is one List result.

type Query

type Query struct {
	Agent  string // exact match; "" = any
	Status Status // "" = any; Interrupted matches by the heartbeat rule, Running excludes it
	// ParentID selects the run tree: "" (the zero value) lists
	// top-level runs only, so children do not flood the list view;
	// "*" lists every run; a concrete id lists that run's children.
	ParentID string
	// Tags matches runs carrying every given pair (a subset match —
	// a run may carry more).
	Tags map[string]string
	// Before is the paging cursor: only runs started strictly before
	// it are returned. Zero means newest first. Offsets are
	// deliberately absent — they drift under concurrent inserts
	// (ADR 0010 §0.1: LangGraph keys, Mastra is moving there).
	Before time.Time
	// Limit caps the page: 0 means 50, values above 500 are clamped.
	Limit int
}

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

type RecordOption

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

RecordOption configures a Record tap.

func Tags

func Tags(kv map[string]string) RecordOption

Tags attaches consumer-supplied pairs to every record this tap writes — hmm: {"cwd": wd}, Weft CI: {"pr": n}. Query.Tags matches a record when every queried pair matches; a record may carry more. Several Tags options merge.

type RunRecord

type RunRecord struct {
	ID           string
	ParentID     string // the parent run when this is a subagent's child record; "" = top-level
	ParentCallID string // the parent's tool call that owns the child run
	Agent        string // weft.Name, via RunStart
	Model        weft.ModelInfo
	ManifestHash string            // sha256 of weft.Manifest(agent) at RunStart; "" when unnamed
	WeftVersion  string            // the core module version the recording process built against
	Started      time.Time         // UTC
	Finished     time.Time         // zero while running
	Heartbeat    time.Time         // last write; readers derive Interrupted from it
	Status       Status            // Running | Succeeded | Failed as stored; DeriveStatus may read Interrupted
	Tags         map[string]string // consumer-supplied (Tags); hmm: cwd, Weft CI: pr
	Events       []weft.Event      // Seq order, Nested inline; empty from List
	Result       *weft.RunResult   // set on success and on failure (the partial); omitted by List
	// Usage and Steps are denormalized onto the record from Result
	// (ADR 0010 §2.6: the Usage struct per run) — the list view's
	// token and step columns without a Get.
	Usage weft.Usage
	Steps int
	// Err is the RunError text when Status == Failed, and the
	// degraded marker when a store write failed mid-run.
	Err string
}

RunRecord is one run: its identity, its event stream, and its result. The event stream is every weft.Event the run emitted in emission order, Nested inline, so a parent's record replays as one stream (ADR 0004's total order is the record's order). The result is kept on failure too — RunError carries the partial transcript, so a failed run's transcript up to the failure is part of the record (ADR 0010 §2.1). The record is not a checkpoint: it says what happened, not how to resume mid-step.

type Status

type Status string

Status is a run record's state. running, succeeded, and failed are stored; interrupted is what a reader concludes (DeriveStatus) — never written, so a crash leaves evidence and a live run in another process is not mistaken for a corpse (ADR 0010 §2.4).

const (
	Running     Status = "running"
	Succeeded   Status = "succeeded"
	Failed      Status = "failed"
	Interrupted Status = "interrupted" // derived on read from the heartbeat; never written
)

func DeriveStatus

func DeriveStatus(status Status, heartbeat, now time.Time) Status

DeriveStatus applies the heartbeat rule (ADR 0010 §2.4): a running record whose heartbeat is older than HeartbeatTimeout reads as Interrupted; every other status reads as stored. Get, List, and Query.Status all see records through it. Exported because any reader of the format — an Inspector over a JSONL export, a future backend — must reach the same conclusion from the same bytes.

type Store

type Store interface {
	// Save upserts the run's row by ID and appends any events in
	// r.Events not yet present (compared by position: the first
	// Save's events are the first rows). Saving the same record twice
	// changes nothing; saving one with more events appends the tail.
	// The recorder saves the row at RunStart (running), per heartbeat
	// tick, and at the run's end.
	Save(ctx context.Context, r RunRecord) error

	// Append adds events to a run's stream in arrival order and bumps
	// the heartbeat — the hot path: one transaction per call, nothing
	// buffered. Events are stored under ordinals assigned atomically
	// per run; a call appending to an unknown run fails with
	// ErrNotFound.
	Append(ctx context.Context, id string, ev ...weft.Event) error

	// Get returns the whole record — identity, events in order,
	// result. An event whose type this weft does not know fails with
	// an error wrapping ErrUnknownEvent (naming the type and the run);
	// loud over silent is the format's rule (ADR 0010 §2.5).
	Get(ctx context.Context, id string) (RunRecord, error)

	// List returns a page of records without events, newest first,
	// with the Total count. See Query for filtering and paging.
	List(ctx context.Context, q Query) (Page, error)

	// Delete removes the run and its events. Children survive with
	// their ParentID cleared — orphaned by design: a child is its own
	// record (ADR 0010 §2.1), and deleting a parent does not pretend
	// its delegations never ran. Retention ("older than N") is a
	// consumer loop over List and Delete, not a store feature.
	Delete(ctx context.Context, id string) error
}

Store is a run-record backend: Memory in process, sqlite on disk, Postgres later — all running the storetest conformance table. Implementations must be safe for concurrent use; the recorder is the only writer in weft dev's process, but readers (the Inspector) may read concurrently (WAL, or a second process with its own handle).

func Memory

func Memory() Store

Memory returns a ready-to-use in-process store: records in a map behind a mutex, gone when the process is. For tests, examples, and as the reference behaviour the durable backends are compared against.