events

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 28, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Overview

Package events is the card event stream of nova-tools #2563 and the SQLite fold that is its record: every card transition XADDs one entry to `cards:done`, and one consumer folds the stream into a SQLite file whose views answer "how many, how much, per model, per route, per bench, per day".

THERE IS ONE STREAM (Rowan's ruling on the bus 2026-09-22, accepted by Johnny 17:16Z). `cards:done` is the stream internal/record and internal/ci already read; this package does not mint a second one beside it, it adds the event fields to the entries of that stream. The SQLite file is a fold of `cards:done` and nothing else: a view that `fold --rebuild` can recompute from the stream at any moment, never a second source of truth. An entry written before the fields were added carries no `event` field; the fold counts it as skipped rather than guessing it into `ok` or `fail`.

The division of labour is Johnny's (reports/redis-for-nova-tools-2026-09-21.md section 8, adopted): Redis holds ids, counts and event ids and NOTHING else -- never a diff, a test, a prompt, a transcript or a disposition -- and every query lives in the fold, over core Redis types only, so Valkey stays a drop-in and the hot store never becomes the database. Validate enforces that literally: a field over maxFieldBytes or carrying a control character is refused at the door rather than trusted to the caller's good manners.

Streams deliver at least once, so the event id is the primary key of every fold table and a redelivery is a no-op. That is what makes Johnny's bar 3 -- a hundred DONEs, kill the fold, restart, the count is a hundred -- a property of the schema rather than of luck.

Index

Constants

View Source 
const (
	Stream = "cards:done"
	Group  = "fold"
	MaxLen = 1_000_000
)

Stream is the one stream every card transition writes to, Group the consumer group the fold reads under, and MaxLen the approximate cap XADD trims to (MAXLEN ~ 1e6): a stream that grows without a bound is a store that fills a disk no one is watching.

View Source 
const (
	DefaultPasswordEnv = "NOVA_REDIS_BENCH_PASSWORD"
	AddrEnv            = "NOVA_REDIS_ADDR"
	HostEnv            = "NOVA_REDIS_HOST"
	PortEnv            = "NOVA_REDIS_PORT"
)

The environment this writer reads. The password variable is the one `nova-secrets exec --only NOVA_REDIS_BENCH_PASSWORD` fills; the address is configuration, never a constant in this repository.

View Source 
const DefaultUser = "bench"

DefaultUser is the fleet's ACL user for the card path: it needs XADD on `cards:done`, the one stream (Stream), and nothing else. The fold reads under its own user.

View Source 
const SchemaVersion = 3

SchemaVersion is the migration version schema.sql installs. Version 2 added the decisions table (#2623), the record the decide_log table held. Version 3 (#3159) rebuilds every view on open and prints $/landed as a lower bound with its coverage.

View Source 
const Tables = "attempts,reads,landings,decisions"

Tables is the fold's tables, as the --init receipt names them.

Variables

View Source 
var Kinds = []Kind{Queued, Leased, Started, Turn, OK, Fail, Asked, Harvested, PullReq, Read, Landed, Jev, Decide}

Kinds is every kind the stream accepts, in banner order.

Functions

func EmitOne 

func EmitOne(in EmitInput) int

EmitOne validates the event, writes it and prints the receipt. Exit 0 wrote one entry, exit 2 could not run.

func Float64 

func Float64(f float64) *float64

func FoldRun 

func FoldRun(in FoldInput) int

FoldRun is the verb. Exit 0 is a pass that ran, exit 2 is could not run.

func Int64 

func Int64(n int64) *int64

Int64 and Float64 are the one-line way to fill a reported number: Event{USD: Float64(0.11)}.

func KindList 

func KindList() string

KindList is the kinds as the refusal and the usage banner spell them.

func SortedIDs 

func SortedIDs(entries []Entry) []string

SortedIDs is the entries' ids in stream order; a helper the tests read.

func Text 

func Text(s string) string

Text makes a free-text value fit for the stream: escaped to one line and capped at the field ceiling with oneline's byte mark, so a long reason arrives cut AND SAYS SO rather than being refused at the door and losing the whole decision. The uncut text is in the JSON lines log the decision was also written to.

Types

type DB 

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

DB is the fold's SQLite file: four tables keyed on the event id and the views over them.

func OpenDB 

func OpenDB(ctx context.Context, path string) (*DB, error)

OpenDB opens (creating if absent) the file at path and applies the schema. The schema is idempotent, so this runs on every start.

func (*DB) Apply 

func (d *DB) Apply(ctx context.Context, entries []Entry) (inserted, skipped int, _ error)

Apply folds one batch of entries in ONE transaction and reports how many rows were new. The insert is ON CONFLICT DO NOTHING on the event id, so a redelivery adds nothing; a batch either lands whole or not at all, and the caller acks only after it lands.

An entry the stream holds that this package cannot read is a skip, not a failure: a writer with a bug must not stop the fold for every other writer. It is counted and named.

func (*DB) Close 

func (d *DB) Close() error

Close closes the file.

func (*DB) Count 

func (d *DB) Count(ctx context.Context) (int, error)

Count is how many rows the four tables hold together: the number Johnny's bar 3 asks for.

func (*DB) CountKind 

func (d *DB) CountKind(ctx context.Context, k Kind) (int, error)

CountKind is how many rows of one kind the fold holds.

func (*DB) Dump 

func (d *DB) Dump(ctx context.Context, w io.Writer) error

Dump writes every row of the three card tables as TSV, table by table and ordered inside each table by event id, then the decisions under a header of their own columns. It is deterministic by construction -- no fold timestamp, no insertion order -- which is what lets a rebuild be compared with an incremental fold byte for byte.

func (*DB) Path 

func (d *DB) Path() string

Path is the file this fold writes.

func (*DB) Report 

func (d *DB) Report(ctx context.Context, w io.Writer, max int) error

Report writes the views as TSV blocks: the numbers a coordinator reads and the numbers `nova-pulse status` will read. max caps the rows of each block; 0 means all.

type Decision 

type Decision struct {
	UnitID              string
	Kind                string
	Files               int
	Packages            int
	Lanes               int
	Lane                string
	RungTried           string
	Height              int
	Confidence          *float64
	Floor               *float64
	SteppedUp           bool
	Escalated           bool
	Designated          bool
	Source              string
	RowanPick           string
	Reason              string
	Wait                string
	AwaitingTermination bool
	Refusal             string
	Outcome             string
	RungSucceeded       string
	Calls               int
	UsageFailed         bool
}

Decision is the decide_log row as stream fields. Every string is an id, a name or a short capped reason; every number is a count, a height or a confidence.

type Dial 

type Dial struct {
	Addr     string
	Username string
	Password string
	Stream   string
}

Dial options. The password never arrives as a flag or an argument: the caller reads it from the environment `nova-secrets exec` put it in, and nothing here ever prints it.

type EmitInput 

type EmitInput struct {
	Addr     string
	Username string
	Password string // read from the environment by the caller; never a flag, never printed
	Stream   string
	Event    Event
	Print    bool // render the entry and write nothing, so a writer can be watched offline
	Now      time.Time
	Timeout  time.Duration
	Stdout   io.Writer
	Stderr   io.Writer
}

EmitInput is `nova-pulse event`: one XADD, one line, for the bash writers -- the launcher, harvest and the lander -- until they are Go.

type Emitter 

type Emitter interface {
	Emit(ctx context.Context, e Event) (id string, err error)
}

Emitter is the write half: one XADD per card transition. It is an interface so the unit tests run against the in-memory FakeStream and the integration test against the fleet store, over the same code.

type Entry 

type Entry struct {
	ID     string
	Fields map[string]string
}

Entry is one delivered stream entry: the id Redis minted and the fields the writer wrote. The id is the fold's primary key, so it is the whole of the idempotency story.

type Event 

type Event struct {
	Label     string
	Attempt   int
	Bench     string
	Model     string
	Route     string
	Kind      Kind
	TokensIn  *int64
	TokensOut *int64
	USD       *float64
	PR        string
	Head      string
	At        time.Time

	// Decision is the decide_log row a `decide` entry carries, and nil on every other kind.
	Decision *Decision
}

Event is one card transition. Every string field is an id or a name; every number is a count or a price. There is no field for prose and that is the point.

TokensIn, TokensOut and USD are pointers because an absent cost is not a zero cost (no evidence is not negative evidence): nil means nobody reported the number, the entry then carries no such field at all, and the fold stores NULL. A zero is a writer saying zero.

func FromFields 

func FromFields(f map[string]string) (Event, error)

FromFields reads one stream entry back. A missing tokens_in, tokens_out or usd stays ABSENT (nil), because a writer that did not know the cost has not reported a zero cost; a number that is present and unreadable is an error, because that is a writer with a bug.

func (Event) Day 

func (e Event) Day() string

Day is the UTC date the event belongs to, or "" when it carries no stamp. It is computed once at fold time so the per-day view is a column and not a string function in a query.

func (Event) Fields 

func (e Event) Fields() map[string]string

Fields is the entry as Redis stores it: every value a string, every name the one #2563 spells. tokens_in, tokens_out and usd are present only when the event carries them.

func (Event) Line 

func (e Event) Line(stream, id string) string

Line is the one-line receipt `nova-pulse event` prints.

func (Event) Stamp 

func (e Event) Stamp(now time.Time) Event

Stamp returns the event with At filled from now when the caller left it zero, so the emitter stamps exactly once and a replay never restamps.

func (Event) Validate 

func (e Event) Validate() error

Validate is the door. It refuses what must never reach Redis rather than trusting the caller: an unknown kind, a negative count, a field long enough to be a payload, and any control character, which would also break the one-line output grammar on the way back.

func (Event) Values 

func (e Event) Values() []any

Values is the flat field/value list XADD writes, in fieldNames order. A number nobody reported is left out, so the entry has no such field rather than a zero.

type FakeStream 

type FakeStream struct {

	// Now is the clock Emit stamps an unstamped event with; a test may pin it.
	Now func() time.Time

	// FailEmit, when set, is the error every Emit returns instead of writing. It is
	// the ONLY way to prove the contract the card-path writers are built on -- that a
	// store which refuses an entry does not fail the card, the harvest or the landing
	// -- so the fake carries a way to be broken on purpose (nova-tools #2563 item 1).
	// It fails Emit alone: a fold reading a stream whose writer is broken is a
	// different test.
	FailEmit error
	// contains filtered or unexported fields
}

FakeStream is the in-memory stream the unit tests run against: the same Store the Redis path implements, with the same group semantics -- at-least-once delivery, a pending list that survives a "crash", and an ack that is the only thing that retires an entry.

It is STRICT LIKE THE REAL TOOL (AGENTS.md): Emit validates exactly as RedisStore.Emit does, and a read under a group nobody created fails with NOGROUP, the way Redis does. A lenient fake here would ship the fold broken on the one store that matters.

func NewFakeStream 

func NewFakeStream() *FakeStream

NewFakeStream returns an empty stream named for the real one.

func (*FakeStream) Ack 

func (f *FakeStream) Ack(ctx context.Context, group string, ids ...string) error

Ack retires the ids from the pending list.

func (*FakeStream) Close 

func (f *FakeStream) Close() error

Close is the Store contract; the fake holds nothing to release.

func (*FakeStream) Emit 

func (f *FakeStream) Emit(ctx context.Context, e Event) (string, error)

Emit appends one entry and returns its id, in Redis's `<ms>-<seq>` shape.

func (*FakeStream) EnsureGroup 

func (f *FakeStream) EnsureGroup(ctx context.Context, group string) error

EnsureGroup creates the group at entry 0. A second call is not an error.

func (*FakeStream) Len 

func (f *FakeStream) Len() int

Len is how many entries the stream holds.

func (*FakeStream) Next 

func (f *FakeStream) Next(ctx context.Context, group, consumer string, count int, block time.Duration) ([]Entry, error)

Next takes up to count entries the group has never been delivered. block is accepted and ignored: the fake never waits, because a test that waits on a clock is a test that fails on a loaded bench (AGENTS.md rule 3).

func (*FakeStream) Pending 

func (f *FakeStream) Pending(ctx context.Context, group, consumer string, count int) ([]Entry, error)

Pending returns the group's unacked entries, oldest first: XAUTOCLAIM with MinIdle 0, so a restarted fold under any consumer name takes them back.

func (*FakeStream) PendingCount 

func (f *FakeStream) PendingCount(group string) int

PendingCount is how many entries the group still owes an ack for.

func (*FakeStream) Range 

func (f *FakeStream) Range(ctx context.Context, start string, count int) ([]Entry, error)

Range walks the stream itself from start, inclusive, touching no group.

func (*FakeStream) StreamName 

func (f *FakeStream) StreamName() string

StreamName is the stream this fake stands for.

type FoldInput 

type FoldInput struct {
	Addr     string
	Username string
	Password string
	Stream   string
	Group    string
	Consumer string
	DB       string
	Interval time.Duration
	Count    int
	Timeout  time.Duration
	Max      int

	Once    bool // one pass, then exit; the CI-shaped form
	Rebuild bool // replay the whole stream into a file that does not exist yet
	Init    bool // apply the schema and exit, with no store
	Report  bool // print the views of an existing file, with no store
	Dump    bool // print every row, deterministically, with no store

	Stdout io.Writer
	Stderr io.Writer
}

FoldInput is `nova-pulse fold`: the stream's one consumer, its replay and its two offline readers (--init and --report).

type Folder 

type Folder struct {
	Reader   Reader
	DB       *DB
	Group    string
	Consumer string
	Count    int           // entries per read
	Block    time.Duration // how long one read may wait for a new entry
}

Folder is one consumer of the stream into one file.

func (*Folder) Once 

func (f *Folder) Once(ctx context.Context) (Stats, error)

Once is one pass: reclaim what a dead predecessor left unacked, then take new entries. The order matters -- pending first -- because otherwise a fold that is killed and restarted walks past its own unfinished work and the count is short.

func (*Folder) Run 

func (f *Folder) Run(ctx context.Context, interval time.Duration, out io.Writer, stream string) (Stats, error)

Run holds the loop: one pass per interval until the context is cancelled. A cancelled context is the shutdown, not a failure, so it returns the running total and no error -- which is what makes "kill it half way and restart" an ordinary thing to do.

func (*Folder) Start 

func (f *Folder) Start(ctx context.Context) error

Start creates the group, so a fold that starts before any writer still reads entry 0.

type Kind 

type Kind string

Kind is the transition one entry records. 

const (
	Queued    Kind = "queued"
	Leased    Kind = "leased"
	Started   Kind = "started"
	Turn      Kind = "turn"
	OK        Kind = "ok"
	Fail      Kind = "fail"
	Asked     Kind = "asked"
	Harvested Kind = "harvested"
	PullReq   Kind = "pr"
	Read      Kind = "read"
	Landed    Kind = "landed"
	Jev       Kind = "jev"
)

The kinds, in the order a card walks them. `read` is a friend's read of a pull request, `landed` the lander's merge, `jev` a Jev decision; the rest are the card's own life. `decide` (decide.go) is a routing decision carrying the decide_log fields: the record that was the decide_log table until #2623. 

const Decide Kind = "decide"

Decide is a routing decision: one entry per verdict, carrying a Decision.

func (Kind) Known 

func (k Kind) Known() bool

Known reports whether k is a kind the stream accepts.

type Reader 

type Reader interface {
	EnsureGroup(ctx context.Context, group string) error
	Pending(ctx context.Context, group, consumer string, count int) ([]Entry, error)
	Next(ctx context.Context, group, consumer string, count int, block time.Duration) ([]Entry, error)
	Ack(ctx context.Context, group string, ids ...string) error
	Range(ctx context.Context, start string, count int) ([]Entry, error)
}

Reader is the read half the fold uses. It is XREADGROUP, XAUTOCLAIM, XACK and XRANGE and nothing else: core types only, no server-side logic (Johnny, section 8).

  • EnsureGroup makes the stream and the group at entry 0, so a fold started after the first writer still reads the whole stream.
  • Pending claims what this group was delivered and never acked, from any consumer, so a fold that is killed and restarted under a new consumer name loses nothing.
  • Next takes new entries.
  • Ack is the commit, and it happens only after the fold's transaction has committed.
  • Range is the replay --rebuild reads, and it touches no group state.

type RedisStore 

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

RedisStore is the live stream on the fleet Redis. go-redis is already this repository's Redis client (internal/redisq, internal/record, internal/ci), so this adds no dependency; the five calls below are XADD, XREADGROUP, XAUTOCLAIM, XACK and XRANGE, which is the whole of what "core types only" allows.

func Open 

func Open(ctx context.Context, d Dial) (*RedisStore, error)

Open names the store and sends nothing (#3277): the first read or write dials, so a bad address fails the caller's first batch instead of every open paying a PING round trip.

func (*RedisStore) Ack 

func (s *RedisStore) Ack(ctx context.Context, group string, ids ...string) error

Ack commits the delivery. It runs after the fold's own transaction, never before.

func (*RedisStore) Close 

func (s *RedisStore) Close() error

Close releases the pool.

func (*RedisStore) Emit 

func (s *RedisStore) Emit(ctx context.Context, e Event) (string, error)

Emit is the one XADD: `XADD <stream> MAXLEN ~ 1000000 * <fields>`. The event is validated first, so nothing but ids and counts can ever reach the store.

func (*RedisStore) EnsureGroup 

func (s *RedisStore) EnsureGroup(ctx context.Context, group string) error

EnsureGroup creates the group at entry 0, making the stream if it is absent. A group that already exists is not an error: every fold start runs this.

func (*RedisStore) Next 

func (s *RedisStore) Next(ctx context.Context, group, consumer string, count int, block time.Duration) ([]Entry, error)

Next takes up to count new entries, blocking for at most block. Zero block is never passed to Redis, which reads BLOCK 0 as "forever".

func (*RedisStore) Pending 

func (s *RedisStore) Pending(ctx context.Context, group, consumer string, count int) ([]Entry, error)

Pending claims every entry this group holds unacked, from any consumer, oldest first. MinIdle is zero on purpose: the fold is a single consumer and a restart must take its own dead predecessor's work back at once, not after an idle timeout.

func (*RedisStore) Range 

func (s *RedisStore) Range(ctx context.Context, start string, count int) ([]Entry, error)

Range reads the stream itself from start, forwards, touching no group. It is the replay `fold --rebuild` walks.

func (*RedisStore) Reach 

func (s *RedisStore) Reach(ctx context.Context) error

Reach sends one PING, for a caller that must refuse before costly work rather than find the store down after it; Open itself sends nothing (#3277).

func (*RedisStore) SetClock 

func (s *RedisStore) SetClock(now func() time.Time)

SetClock installs the clock Emit stamps an unstamped event with.

func (*RedisStore) StreamName 

func (s *RedisStore) StreamName() string

StreamName is the stream this store reads and writes.

type Stats 

type Stats struct {
	Read     int // entries taken from the stream
	Inserted int // rows the fold did not already hold
	Acked    int // deliveries committed back to the group
	Skipped  int // entries this package could not read
}

Stats is what one fold pass did.

func Rebuild 

func Rebuild(ctx context.Context, r Reader, db *DB, batch int) (Stats, error)

Rebuild replays the stream from its first entry into this file, reading the stream itself and never the group: a rebuild must not consume what the running fold has not yet folded.

func (*Stats) Add 

func (s *Stats) Add(o Stats)

Add folds one pass's numbers into a running total.

func (Stats) Line 

func (s Stats) Line(db, stream, group string) string

Line is the one-line receipt a fold pass prints.

type Store 

type Store interface {
	Emitter
	Reader
	Close() error
}

Store is a stream that is both halves. The Redis store and the fake both are one.

type Writer 

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

Writer holds an open store, or nothing at all. A disabled writer is not an error: it is the shape every bench without the store's password runs in.

func OpenWriter 

func OpenWriter(ctx context.Context, opt WriterOptions) *Writer

OpenWriter resolves the configuration and dials, and NEVER returns an error: a writer it could not open is a writer that emits nothing. A dial that failed says so once, here, rather than once per card.

func (*Writer) Close 

func (w *Writer) Close()

Close releases the store. It is safe on a nil or disabled writer.

func (*Writer) Enabled 

func (w *Writer) Enabled() bool

Enabled reports whether an entry would actually be written. It exists for a receipt line, never as a guard a caller has to remember: Send is safe either way.

func (*Writer) Send 

func (w *Writer) Send(ctx context.Context, e Event)

Send writes one entry and returns NOTHING. A refused entry, a store that is down and a context that was cancelled are all one line on the log and a return: the card, the harvest or the landing that called this is already done and is not undone by a measurement. There is no error to ignore, because there is no error to return.

func (*Writer) StreamName 

func (w *Writer) StreamName() string

StreamName is the stream entries land on, for a receipt.

type WriterOptions 

type WriterOptions struct {
	Addr        string // --events-store; "" resolves from the environment
	Username    string // "" is DefaultUser
	PasswordEnv string // "" is DefaultPasswordEnv
	Stream      string // "" is Stream
	Timeout     time.Duration
	Log         io.Writer // where the one line per failure goes; nil discards it

	// Lookup is the environment. nil is os.Getenv, and a test passes a map.
	Lookup func(string) string
	// Dial opens the store. nil is the Redis one, and a test passes a fake -- including
	// a fake that ERRORS, which is the only way to prove the caller survives it.
	Dial func(ctx context.Context, d Dial) (Store, error)
}

WriterOptions is how a verb turns the stream on. Every seam a test needs is here, so no test of a card path opens a connection (AGENTS.md rule 2).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.