Documentation
¶
Overview ¶
Package export follows the events table of a Store and hands every committed event, in commit order, to a sink: the JSON Lines writer here, or the OpenTelemetry span exporter in the nested module export/otel.
Events are written in the transaction that makes the state change they explain, so the table is complete and a follower needs no other input (docs/architecture.md, ADR 2). The event's Seq is its cursor: SQLite assigns it from an AUTOINCREMENT key inside the single write transaction that commits the event, so it is strictly increasing in commit order across every run in a database and never reused. A follower that remembers the last Seq it delivered resumes exactly after it.
A Follower only reads. Opened on a Store from agentrt.OpenExisting with readOnly set, it can run as a separate process beside the consumer that owns the database, which needs no code for it. It reads a bounded page at a time through agentrt.Store.ListEventsAfter, with every text column capped at agentrt.MaxPageText, so a huge table, or one crafted row, cannot exhaust the follower's memory.
Index ¶
Examples ¶
Constants ¶
const ( // DefaultInterval is how long a Follower waits between polls when it // has delivered everything committed so far and Interval is zero. DefaultInterval = 250 * time.Millisecond // MaxInterval bounds Interval: a follower never waits longer than this // between polls, so an event is never more than a minute from its sink. MaxInterval = time.Minute // DefaultPageSize is how many events a Follower reads per query when // PageSize is zero. DefaultPageSize = 500 // MaxPageSize bounds PageSize, so one page holds at most this many // payloads of up to agentrt.MaxPageText each. MaxPageSize = 10_000 )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Cursor ¶
type Cursor interface {
Load(ctx context.Context) (int64, error)
Save(ctx context.Context, seq int64) error
}
Cursor persists the Seq of the last event a Follower delivered. Load returns zero when nothing was saved yet.
type FileCursor ¶
type FileCursor struct {
Path string
}
FileCursor keeps the cursor in a file holding the Seq in decimal, written to a temporary file in the same directory and renamed over the old one, so a crash leaves the previous cursor rather than a torn file. A missing file loads as zero.
type Follower ¶
type Follower struct {
// Store is read and never written. agentrt.OpenExisting(path, true)
// opens one that cannot write.
Store *agentrt.Store
// RunID restricts delivery to one run. Empty delivers every run's
// events interleaved in commit order.
RunID string
// Cursor is where the follower resumes from and saves to. Nil starts
// at the first event and remembers nothing beyond this call.
Cursor Cursor
// Interval is the wait between polls once everything committed has
// been delivered: DefaultInterval when zero, never more than
// MaxInterval.
Interval time.Duration
// PageSize is how many events one query reads: DefaultPageSize when
// zero, never more than MaxPageSize.
PageSize int
// Once makes Follow return nil as soon as it has delivered every event
// committed when it last looked, instead of polling for more.
Once bool
}
Follower delivers a Store's events in commit order, each once per cursor position. The zero value is not usable: Store must be set.
Example ¶
A follower delivers a run's events in commit order to a sink; here the JSON Lines sink, into a buffer, and Once returns when the committed events are delivered. An operator's process would open the database with agentrt.OpenExisting(path, true) and leave Once unset.
store, err := agentrt.OpenStore(":memory:")
if err != nil {
panic(err)
}
defer store.Close()
echo := tool{spec: agentrt.ToolSpec{Name: "echo", Description: "echo", InputSchema: []byte(`{"type":"object"}`), SideEffect: agentrt.ReadOnly, Timeout: time.Second},
fn: func(c agentrt.ToolCall) (agentrt.ToolResult, error) {
return agentrt.ToolResult{Content: c.Args, Summary: "echoed"}, nil
}}
driver, err := agentrt.NewDriver(agentrt.Config{Store: store, Agent: &scripted.Agent{Decisions: []agentrt.Decision{
scripted.ToolCall("echo", `{"n":1}`, "look"),
scripted.Complete(`{"ok":true}`),
}}, Policy: agentrt.DefaultPolicy(), Tools: []agentrt.Tool{echo}})
if err != nil {
panic(err)
}
if _, err := driver.Start(context.Background(), "echo once", agentrt.DefaultLimits()); err != nil {
panic(err)
}
var out bytes.Buffer
follower := &export.Follower{Store: store, Cursor: &export.MemCursor{}, Once: true}
if err := follower.Follow(context.Background(), export.JSONL(&out)); err != nil {
panic(err)
}
for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) {
var rec struct {
Seq int64 `json:"seq"`
Type string `json:"type"`
}
if err := json.Unmarshal(line, &rec); err != nil {
panic(err)
}
fmt.Println(rec.Seq, rec.Type)
}
Output: 1 run.created 2 run.started 3 step.started 4 step.decided 5 step.policy 6 step.tool_started 7 step.tool_finished 8 step.started 9 step.decided 10 run.finished
func (*Follower) Follow ¶
Follow reads events after the cursor and calls sink for each, in Seq order, until ctx is done, sink fails, or, with Once, nothing more is committed. Follow returns ctx.Err() when ctx ends it.
The cursor is saved after each page and before Follow returns, so delivery is exactly once across stops that return: a follower stopped by its context, by a sink error, or by Once, and started again on the same cursor, delivers each event once (the event a sink refused is delivered again, since it was never delivered). It is at least once after a hard kill: a process that dies between a sink call and the save that follows it, or whose save fails, starts again from the cursor last saved and delivers again what it delivered since, at most one page of PageSize events. A sink that must not act twice keys on Seq.
func (*Follower) Head ¶ added in v0.5.0
Head returns the Seq the follower has delivered up to, as its Cursor saved it, and the stored hash of that event: the head of the event chain as far as the follower delivered it, cut at 128 characters as agentrt.Store.EventHash reads it. An operator keeps the pair where whoever can write the database cannot reach, beside what the sink wrote, and later checks it with agentrt verify -head SEQ:HASH, which fails unless the walk from the first event reaches Seq with no break and the event at Seq still has that hash. In code the check is the same three parts: agentrt.Store.VerifyEvents from the first event to at least Seq with no Break, a To of at least Seq (a database cut before Seq verifies intact up to where it ends), and the event at Seq's hash, the report's Hash when To is Seq or EventHash otherwise, equal to the one kept. A chain that passes was not rewritten up to Seq. The hash covers every run's events up to Seq, a follower of one run included, because the chain is the whole database's. Head is zero and empty when nothing was delivered, or the Cursor is nil and so remembers nothing; agentrt.ErrNotFound means the event the cursor names is no longer stored. It reads the cursor and one row and changes nothing, and may be called while Follow runs, with a Cursor safe for that, as FileCursor is.
type MemCursor ¶
type MemCursor struct {
Seq int64
}
MemCursor is a Cursor in memory, for a follower that starts over each time it runs or whose caller persists Seq itself.
Source Files