Documentation
¶
Overview ¶
Package events is the run-event bus: a stdlib-only leaf that carries what a run is doing, as it does it, from the packages executing a job to whatever is watching.
It exists because a run's story was only ever told to a terminal. The merkle store answers "what content has succeeded" and nodes.result answers "what did this step produce", but neither answers "what is happening right now, in plan order" — the question a web UI, a progress display, or a live log tail is entirely made of.
Two consumers, one stream. A subscriber (the web UI's SSE handler) sees events as they happen; a sink (the store) persists the same events so the run reads back identically after it ends. That is deliberate: a live view and a post-hoc view built from different sources drift, and the drift is always discovered in the middle of an incident.
The bus travels on the context, like the pipeline's own resume and exec-log state, so publishing costs no signature changes in the packages that execute steps. A context with no bus makes every Publish a no-op, which is what `steps run` in a terminal gets.
Index ¶
- Constants
- func EndRun(ctx context.Context, status, text string)
- func Headless(ctx context.Context) bool
- func Hold(ctx context.Context) (io.Writer, func())
- func LogAttrs(ctx context.Context) []slog.Attr
- func LogHandler(inner slog.Handler) slog.Handler
- func Note(ctx context.Context, level, text string)
- func Publish(ctx context.Context, event Event)
- func Renderer(ctx context.Context) func(Event)
- func RunID(ctx context.Context) string
- func StartRun(ctx context.Context, job string) context.Context
- func Stderr(ctx context.Context) io.Writer
- func Stdout(ctx context.Context) io.Writer
- func StepID(ctx context.Context) int64
- func Tee(ctx context.Context, wrap func(stderr bool, inner io.Writer) io.Writer) context.Context
- func Traceparent(ctx context.Context) string
- func WithBus(ctx context.Context, bus *Bus) context.Context
- func WithLogAttrs(ctx context.Context, args ...any) context.Context
- func WithOutput(ctx context.Context, out Output) context.Context
- func WithRenderer(ctx context.Context, render func(Event)) context.Context
- func WithRunID(ctx context.Context, id string) context.Context
- func WithStepID(ctx context.Context, id int64) context.Context
- type Bus
- type Event
- type Output
- type Tracer
Constants ¶
const ( TypeJobStarted = "job_started" TypeJobFinished = "job_finished" TypeStepStarted = "step_started" TypeStepFinished = "step_finished" TypeStepSkipped = "step_skipped" // TypeStepOutput carries what a step printed. Published once when the // step ends rather than streamed line by line: a transcript wants the // output, and one bounded event per step costs a row instead of // thousands. TypeStepOutput = "step_output" // TypeStepOutputChunk is a piece of what a RUNNING step is printing, Status naming the stream (stdout/stderr). Temporary by design: the step's TypeStepOutput replaces every chunk recorded for it, so a finished build costs what it did before chunks existed, and a build that died mid-step keeps them as the only record. TypeStepOutputChunk = "step_output_chunk" // TypeStepNote is what a step's machinery said about it, not its own output — an image pulled, a worker acquired; Status is the level, Text the message. These were fmt.Printf lines once, told to the terminal alone. TypeStepNote = "step_note" // Agent conversation traffic, mirroring the persisted transcript's own // event vocabulary (see internal/agent/transcript.go) so a live view and // a stored transcript render through the same code path. TypeAgentSystem = "agent_system" TypeAgentUser = "agent_user" TypeAgentText = "agent_text" TypeAgentCall = "agent_call" TypeAgentResult = "agent_result" TypeAgentSubagent = "agent_subagent" // TypeAgentCompaction marks where older turns were replaced by a // summary; Name is the label, Text the summary the model worked from. TypeAgentCompaction = "agent_compaction" )
Event types. A consumer switches on Type, so these are the vocabulary.
const ( NoteInfo = "info" NoteWarn = "warn" )
Note levels, carried in Status of a TypeStepNote event.
Variables ¶
This section is empty.
Functions ¶
func Headless ¶
Headless reports whether ctx's output reaches nobody, so there is no person to prompt.
func Hold ¶
Hold takes the terminal for a prompt: the returned writer reaches the person directly, and nothing else draws until release. With no renderer to pause it is Stdout.
func LogAttrs ¶
LogAttrs is what WithLogAttrs put on ctx, which is also a run's identity (pipeline, revision, run, job) to anything recording it.
func LogHandler ¶
LogHandler wraps inner so every record written with a context carries that context's WithLogAttrs attributes, ahead of the call site's own — identity first, where the eye scans for it. A key the call site passed, or a Logger.With in the same group added, wins over the context's, so no key is printed twice, and wrapping twice writes each attribute once for the same reason. Attributes land inside any group the logger has open; nothing here opens one.
func Note ¶
Note publishes a TypeStepNote under the run and step ctx names, or the job (StepIndex -1) outside a step. With no bus to tell — none installed, or one already closed, which is where a job's deferred worker release speaks — it writes the line to Stdout(ctx) instead, or logs it when ctx is Headless, because a note is often the only record of what it says.
func Publish ¶
Publish publishes to whatever bus ctx carries. It is the form the execution packages use: one call, no nil checks, no-op off the web path.
func Stdout ¶
Stdout is where ctx's human-facing output goes, the process's own stdout when no renderer installed one.
func Tee ¶
Tee installs wrap over each of ctx's output streams: it is handed the writer the stream resolves to today and returns the one to use instead, so a caller can watch a step's bytes on their way to whatever renderer is installed. Only Step is replaced; Hold, Headless and the rest are kept.
func Traceparent ¶
Traceparent is the W3C trace context a command started under ctx should continue, or "" with no tracer installed.
func WithBus ¶
WithBus returns a context carrying bus, so the packages that execute steps can publish without every call site growing a parameter.
func WithLogAttrs ¶
WithLogAttrs returns a context whose log lines carry args — key/value pairs read the way slog reads them — so every package executing a step logs under the pipeline/run/job/step it belongs to without that identity being threaded through its signatures. A key ctx already carries is replaced, not repeated.
It lives beside WithRunID, and for the same reason: which run a line belongs to is known by the package that owns the plan walk, and needed by packages that must not import it. A stdlib-only leaf both already depend on is the one place they can meet — and this stays stdlib-only, since log/slog is stdlib. Only attributes travel, not a *slog.Logger: a logger captured here would pin whatever slog.Default() was at the time, while LogHandler reads them when each line is written, through whichever handler is current.
An empty string value is kept, and means absent: LogHandler never writes it, so adding a key with "" clears what a parent context set — how a nameless nested step stops reporting its parent's step=.
The alternative — a jobName/index parameter pair on every function that might log — makes a diagnostic concern dictate the shape of everything it touches, and spreads exactly as far as the logging does.
func WithOutput ¶
WithOutput installs a renderer's writers, so nothing below it writes to the process's streams behind the renderer's back.
func WithRenderer ¶
WithRenderer names what should draw a run started under ctx: the observer a bus the runner attaches will call, in place of its plain lines.
func WithRunID ¶
WithRunID tags ctx with the run every subsequent event belongs to.
The id lives here rather than in the package that mints it because the packages that need to STAMP it (agent, publishing a conversation turn) must not import the package that owns the plan walk. A stdlib-only leaf both already depend on is the one place they can meet.
func WithStepID ¶
WithStepID tags ctx with the display-tree id of the step now executing, so a package that cannot import the pipeline — internal/agent, publishing a conversation's turns — can still say which step's subtree its events hang under. Without it every turn a nested agent produced had to be matched back to a step by (index, name), which a try: and the agent it wraps share.
Types ¶
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus fans one run's events out to live subscribers and, optionally, to a persistence sink. The zero value is not usable; call New.
func FromContext ¶
FromContext returns the bus carried by ctx, or nil. A nil *Bus is a valid receiver for every method here, so callers need not check.
func New ¶
New returns a bus. When sink is non-nil it is called for every published event, in order, on a single goroutine — the caller does not need its own locking. Close stops that goroutine.
func (*Bus) Close ¶
func (b *Bus) Close()
Close drains and stops the sink goroutine, and makes every later Publish a no-op. Subscribers are left alone — each owns its own cancel.
Idempotent, and safe against concurrent publishers: a run still finishing when the process shuts down would otherwise be sending on the very channel this closes.
func (*Bus) Observe ¶
Observe calls fn for every later event, synchronously and in sequence order — what a terminal needs, where a dropped event is a line never seen. A slow fn slows the run, as the fmt.Printf it replaces did; fn must not Publish or cancel itself.
type Event ¶
type Event struct {
// Seq orders events within a bus. Assigned on publish, monotonic, so a
// subscriber that reconnects can ask for everything after what it has.
Seq int64 `json:"seq"`
At time.Time `json:"at"`
Type string `json:"type"`
RunID string `json:"run_id"`
Job string `json:"job"`
// StepIndex is the plan index the event belongs to, -1 for job-level
// events. A fan-out cell reports its parent's index and distinguishes
// itself by StepName, matching how across: cells are already named.
StepIndex int `json:"step_index"`
StepName string `json:"step_name,omitempty"`
StepKind string `json:"step_kind,omitempty"`
// StepID identifies this OCCURRENCE of a step within its run, and
// ParentStepID the container it ran inside — 0 at the top of a plan.
// Together they are the display tree: what a reader sees folded under
// what.
//
// Deliberately not the merkle chain, which cannot answer this. An
// across: cell hashes under the MATRIX'S predecessor rather than under
// the matrix, so nodes.parent_hash records the three cells and the block
// that owns them as siblings; a when:-skipped step's own hash IS its
// parent's, which would graft it onto itself; and a step's hash is not
// known when it STARTS, because load_var substitution rewrites the step
// inside the dispatch. Minted per run, none of which applies.
//
// Also what tells a try: apart from the step it wraps: those publish the
// same index and the same name, and keying a step by that pair folded the
// two into one row.
StepID int64 `json:"step_id,omitempty"`
ParentStepID int64 `json:"parent_step_id,omitempty"`
Status string `json:"status,omitempty"`
Hash string `json:"hash,omitempty"`
// Text carries model output, a log line, or an error message depending on
// Type; Name and Detail carry a tool call's name and its arguments or
// result. Kept as three generic fields rather than a per-type struct for
// the reason the type itself is flat.
Text string `json:"text,omitempty"`
Name string `json:"name,omitempty"`
Detail string `json:"detail,omitempty"`
DurationMS int64 `json:"duration_ms,omitempty"`
// Worker names where a step's commands ran, as "tag (address)", and is
// EMPTY for a step that ran on this machine. Absence is the signal:
// spelling out "local" would put a worker on every row of every pipeline
// that has none, and the interesting rows are the ones that left.
//
// Recorded because nothing else can answer it after the fact. tags: is
// deliberately outside the merkle hash — placement decides where a step
// runs, not what it produces — so the cache cannot say, and a hash that
// matched a local run is exactly the case where the question gets asked.
//
// The ADDRESS, not the mapping as written: a worker URL carries
// ?identity= and ?hostkey= query parameters that describe how to
// authenticate, and a record rendered in a browser is the wrong place for
// them.
Worker string `json:"worker,omitempty"`
}
Event is one thing that happened during a run.
Flat rather than a type union: it is serialized to sqlite and to SSE, and a flat row keeps both ends free of a discriminated-union decoder for what is, in practice, a handful of optional strings.
type Output ¶
type Output struct {
Stdout io.Writer
Stderr io.Writer
// Step, when set, is the writer for one step's bytes, so a renderer that draws per step knows whose they are. It wins over Stdout/Stderr.
Step func(stepID int64, stderr bool) io.Writer
// Hold, when set, stops the renderer drawing and hands back the raw terminal until release is called — for a prompt a person has to read and answer.
Hold func() (w io.Writer, release func())
// Headless says nobody reads this process's streams — a daemon, whose runs are read from their record — so every writer here is io.Discard and nothing prompts.
Headless bool
}
Output is where a run's human-facing bytes go: a command's streamed output, an agent's answer, a prompt a person must see.
type Tracer ¶
type Tracer interface {
// StartRun opens a run's span the moment the run begins, before it has published anything, so a run refused by validation or preflight is still a trace; work under the returned context is traced beneath it.
StartRun(ctx context.Context, runID, job string) context.Context
// EndRun closes a run's span with the status its job finished on, and anything of it still open.
EndRun(ctx context.Context, runID, status, text string)
// Event is told every event a run publishes, inline and in order, before it reaches the bus.
Event(ctx context.Context, event Event)
// Scope returns ctx under the span for a run's step (0 for the run itself), so work the step does is traced beneath it.
Scope(ctx context.Context, runID string, stepID int64) context.Context
// Traceparent is the W3C trace context ctx is under, or "".
Traceparent(ctx context.Context) string
}
Tracer turns a run's lifecycle events into spans. It is an interface here so every package that runs a step reaches tracing through this leaf, and only the one package that implements it imports a tracing SDK.
Source Files