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