Events, resync and diagnostics#

A program that embeds the engine can receive every state change, budget breach and usage update as it happens, instead of polling. The live stream can drop events, by design, when a subscriber falls far behind or the process crashes, so the engine also lets you read back everything it committed from a saved position. This page covers subscribing, catching up with resync_events, and sending the engine's diagnostic lines to your own logs.

Subscribe#

There are two ways in, one for each style of program:

broadcast is re-exported as hekma_engine::broadcast, so you can name the receiver and its error types without adding tokio yourself. Subscribing takes no lock and is safe at any point in the engine's life.

A subscriber sees only events committed after it subscribed, so subscribe before you start the agents you care about. The quickstart uses the blocking form; this is the async form:

use hekma_engine::broadcast::error::RecvError;
use hekma_engine::EngineEvent;

let mut rx = engine.subscribe();
tokio::spawn(async move {
    loop {
        match rx.recv().await {
            Ok(EngineEvent::Transition(t)) => println!("{}: {} -> {}", t.instance, t.prior_state, t.new_state),
            Ok(EngineEvent::BudgetBreach(b)) => println!("{}: breach at {}", b.instance, b.observed),
            Ok(EngineEvent::UsageUpdate(u)) => println!("{}: usage #{}", u.event.instance, u.event.sequence),
            Err(RecvError::Lagged(n)) => eprintln!("missed {n} events; resync to catch up"),
            Err(RecvError::Closed) => break, // the engine was dropped
        }
    }
});

Event types#

EngineEvent has three variants, one per kind of record the engine keeps:

Each payload is a serde struct with its own schema_version, and serialises with a kind tag (transition, budget_breach, usage_update). The fields, their JSON shapes and the full list of transition causes are on JSON output. The enums inside them, such as TransitionCause, are exhaustive; a new cause arrives in a new minor version of the crate, so add a _ case if you match them.

Events from one engine arrive in the order they were committed, for all agents together, so one agent's events are always in order.

Slow subscribers and Lagged#

The event stream is a ring buffer of EVENT_BUS_CAPACITY events, 1024, shared by every subscriber of one engine. Publishing never waits for a subscriber, so a slow or stuck consumer can never slow down supervision, metering or budget enforcement.

The price is that a subscriber more than 1024 events behind loses the oldest ones. Its next recv returns Err(Lagged(n)), naming how many it missed, and the one after that continues from the oldest event still in the buffer. Lagged is not the end of the stream; keep receiving, and call resync_events to fill the gap. A normal agent produces a few events per model call, so a subscriber that drains promptly does not lag in practice.

The crash window#

The engine writes each record durably first and publishes the event second. A subscriber therefore never sees an event whose record failed to write. The other side of that order: if the process crashes between the write and the publish, that one event is in the durable record but never reaches the stream. Delivery is at most once around a crash.

Nothing is lost from the record itself, and resync_events reads it back.

Catch up with resync_events#

resync_events(name, cursor) reads one agent's committed records and returns them as the same EngineEvent values the live stream carries, starting after a position you supply:

use hekma_engine::ResyncCursor;

let batch = engine.resync_events("worker", saved_cursor.unwrap_or(ResyncCursor::START)).await?;
for event in &batch.events {
    handle(event);
}
if batch.torn_tail_skipped {
    eprintln!("a crash cut the last line of a log; an event may be missing");
}
save(&batch.cursor)?; // ResyncCursor is Serialize + Deserialize

In a run of the async example above, a few seconds after start, the batch held 4 events (two transitions, two usage updates) and the next cursor serialised as {"transitions":2,"breaches":0,"usage":2}.

The rules that make it safe to call repeatedly:

• Idempotent. Passing the returned cursor back never repeats a record. Persist it across your own restarts.
• Ordered per kind. Transitions come in instance.log order, breaches in breaches.log order and usage in ledger order. The batch lists all transitions first, then breaches, then usage, because the records carry no shared sequence across the three kinds. If you need one coherent snapshot across kinds, pause or stop the agent while you read.
• Never guesses. A cursor beyond what a log holds (because a log was deleted or truncated) fails with an EngineError::Log naming the kind and both counts. Reset to ResyncCursor::START yourself if that is what you want.
• Survives a torn write. A crash can cut a log's last line in half. If the last line does not parse and the file does not end in a newline, it is skipped and torn_tail_skipped is set. Any other unreadable line is an error.
• One agent per call. Loop over fleet() for the whole fleet. An unknown name fails with NotFound.

Combine the live stream and resync#

Two orders work, with different trade-offs:

After a Lagged, resync the agents you care about from their saved cursors; the gap is in the durable record.

Diagnostics#

Besides events and errors, the engine writes a few operational notes as single lines starting with [hekma] : an agent adopted from a previous engine, usage it could not record, a breach action it could not carry out, a memory folder the adapter does not pass on, an ACP agent's permission request it denied. By default these go to stderr, which is fine for a terminal program and wrong for a daemon, a GUI or a log pipeline.

A DiagnosticSink sends them to your own writer instead. It is Arc<Mutex<Box<dyn Write + Send>>>, so any writer works and one sink can be shared by several engines:

use std::sync::{Arc, Mutex};
use hekma_engine::{DiagnosticSink, Engine};

let file = std::fs::File::create("/var/log/my-host/hekma.log")?;
let sink: DiagnosticSink = Arc::new(Mutex::new(Box::new(file)));

// Install it at open, so even notes about adopted agents go to the sink.
let engine = Engine::open_with_diagnostics(Some(state_dir), sink.clone())?;

// Or install or replace it later, on an engine that is already open.
engine.with_diagnostics(sink); // also: engine.blocking().with_diagnostics(sink)

What to know about the sink:

• Same bytes. Each note arrives as one complete line, [hekma] prefix and trailing newline included, exactly as it would have appeared on stderr. With a sink installed, the engine writes nothing to stderr.
• Install at open if you can. open_with_diagnostics installs the sink before the engine adopts agents and starts watching them, so no note slips out to stderr in between.
• Replacing is allowed; removing is not. Installing a new sink flushes and replaces the old one. There is no way back to stderr except opening a new engine, or installing a writer that writes to stderr.
• Do not call the engine from the writer. The engine holds its internal lock while it writes a note, so a writer that calls back into the engine deadlocks. Pass the line to a channel or your own logger instead.
• Failures are ignored. A write error, or a panic inside your writer, is caught and dropped, and the same sink receives later notes. Supervision never fails because of a broken sink.

Diagnostics are informational. The durable record of what happened is always the event logs and the ledger.

Overhead#

Subscribing is cheap. Hekma's performance harness measures a running fleet with no subscriber and again with one active subscriber, and fails the build if the subscriber adds more than 0.25 percentage points of CPU or 2 MiB of memory per running agent, or more than 100 ms to the slowest fleet reads (the 99th percentile). Those limits sit on top of the engine's own performance budgets.

Read next#

• JSON output: every field of the three event payloads.
• Embed the engine: opening an engine and the rest of the API.
• Architecture: where the durable records live.