Struct RunCtx 

pub struct RunCtx { /* private fields */ }

The public durability substrate for one run. See the module docs.

Implementations

impl RunCtx

pub fn new( store: Arc<dyn EventStore>, run_id: RunId, log: Vec<EventEnvelope>, ) -> Result<Self, RuntimeError>

Builds a context over a run’s recorded log (empty for a fresh run), with the default clock (the real UTC clock) and the default random source (operating-system randomness).

Errors

Returns RuntimeError::Replay when the log is not a well-formed run history.

pub fn with_hooks( store: Arc<dyn EventStore>, run_id: RunId, log: Vec<EventEnvelope>, clock: ClockFn, random: RandomFn, ) -> Result<Self, RuntimeError>

Builds a context with an injected clock and random source.

The clock stamps every persisted envelope and answers live now observations; the random source answers live random observations. Injecting deterministic functions makes complete event logs comparable across runs, which is how the kill/resume tests prove byte-identical recovery.

Errors

Returns RuntimeError::Replay when the log is not a well-formed run history.

pub fn with_record_prompts(self, record_prompts: bool) -> Self

Turns on recording of the full model request body into the durable log.

Additive and off by default: the existing new and with_hooks constructors leave it off, so no caller that predates this method changes behavior. Chained builder style keeps those signatures intact, which is why the flag arrives this way rather than as a new constructor argument.

When on, each live model_call records the exact request it sent on the ModelCallRequested event, so the v0.3 dashboard inspector can show the prompt. This is PII-sensitive: the body can hold user data and secrets, which is why the default is off and turning it on is a deliberate per-agent or operator choice. The recorded body lands only in the event log; it never reaches the progress stream or any console output. It does not affect replay: the request hash is computed the same either way, and replay ignores the body.

pub fn with_labels(self, labels: BTreeMap<String, String>) -> Self

Sets the correlation tags to stamp on a genuinely fresh RunStarted.

Additive and unset by default: the existing new and with_hooks constructors leave it unset, so no caller that predates this method changes behavior. Chained builder style, mirroring with_record_prompts.

Labels are checked against the sanity bounds (see crate::validate_labels) only on begin’s live path, the moment a RunStarted is actually about to be created; RuntimeError::InvalidLabels surfaces there, not here, so this setter itself is infallible. A replayed begin never re-checks them: whatever the log already holds is trusted and returned as recorded. Labels never enter agent_def_hash or any request hash; they are a tag on the run, not part of its identity.

pub fn set_resume_input(&mut self, input: Value)

Provides the input a parked run is being resumed with. The next await_resume that reaches live mode records it as the Resumed event and returns it; without one, a live await_resume reports Resumption::Parked.

pub fn staged_resume_input(&self) -> Option<&Value>

The resume input staged by set_resume_input and not yet consumed, without consuming it.

This is the read-only half of the accept edge. A driver that needs to vet a resume input against something only it knows (the graph engine checks a gate’s declared approval_schema) has to see the value BEFORE await_resume turns it into a Resumed event, because after that it is history and refusing it would mean an appended event the run has to live with. Peeking here and refusing leaves the log untouched and the run parked exactly as it was.

None once await_resume has taken the value, or when none was staged.

pub fn run_id(&self) -> RunId

The run this context drives.

pub fn is_replaying(&self) -> bool

Whether recorded history remains to be consumed.

pub fn next_seq(&self) -> SequenceNumber

The log position the next consumed or emitted event occupies.

pub async fn begin( &mut self, agent_def_hash: &str, input: &Value, ) -> Result<Value, RuntimeError>

Starts (or replays the start of) the run.

Live: records RunStarted with input and the labels set through with_labels (if any), and returns input. Replayed: verifies agent_def_hash against the recorded event and returns the recorded input, which always wins; the input argument is only used when the log is empty, exactly like labels.

Errors

RuntimeError::Replay on a definition-hash mismatch or any other divergence; RuntimeError::InvalidLabels when the labels set through with_labels violate the sanity bounds (only checked on the live path; see that method); RuntimeError::Store when persistence fails.

pub async fn begin_graph( &mut self, graph_hash: &str, input: &Value, ) -> Result<Value, RuntimeError>

Starts (or replays the start of) a graph run: the graph-document counterpart of begin.

Live: records salvor_core::Event::GraphRunStarted with input, the labels set through with_labels (if any), and no fork origin, then returns input. Replayed: verifies graph_hash against the recorded head (a changed graph document must not silently resume an old run) and returns the recorded input, which always wins.

A graph run’s log opens with this event rather than RunStarted because a graph coordinates many agent hashes and has none at its head. The graph engine calls this once, then frames each node with node_entered / node_exited and records the single terminal itself after the last node.

Errors

RuntimeError::Replay on a graph-hash mismatch or any other divergence; RuntimeError::InvalidLabels when the labels set through with_labels violate the sanity bounds (only checked on the live path, exactly as begin does); RuntimeError::Store when persistence fails.

pub async fn node_entered(&mut self, node: &str) -> Result<(), RuntimeError>

Records (or replays) entry into a graph node. A graph node’s own events (an agent loop’s model calls, a tool call) are recorded between this and the matching node_exited.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn node_exited(&mut self, node: &str) -> Result<(), RuntimeError>

Records (or replays) exit from a graph node, having produced its output. The counterpart of node_entered.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn node_skipped( &mut self, node: &str, reason: &str, ) -> Result<(), RuntimeError>

Records (or replays) that a graph node was skipped: reached on the walk but deliberately not run (a branch routed past it). Unlike an executed node there is no node_entered/node_exited pair; the skip is the node’s sole marker, which is what lets a projection tell “skipped” apart from “never reached”. reason must be a pure function of the document and recorded values so it reproduces on replay.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn branch_taken( &mut self, node: &str, case: &str, ) -> Result<(), RuntimeError>

Records (or replays) that a branch node routed: the named case fired. Recorded between the branch’s node_entered and node_exited, it is the sole authority for which way the branch went. The chosen case must be a deterministic function of recorded values (a pure expression over the routed value, or a decision recomputed from a replayed model reply) so replay reproduces the route.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn map_fanned_out( &mut self, node: &str, items: &Value, ) -> Result<(), RuntimeError>

Records (or replays) that a map node fanned out over a resolved item list.

Recorded between the map node’s node_entered and its per-iteration markers. The items must be a deterministic function of recorded values (the map’s over reference resolved against the recorded routed value), so replay reproduces the identical fan-out, which is what makes the derived per-iteration child ids reproducible.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn map_iteration_started( &mut self, node: &str, index: u64, child_run: &str, ) -> Result<(), RuntimeError>

Records (or replays) that one iteration of a map fan-out started, as a child run with the derived id child_run. The child_run is derived from the parent run id, the node id, and the index. On replay the RECORDED id wins and the match is on node + index alone, so a fork (which replays the origin’s prefix under a new run id and thus re-derives a different id) still replays its inherited map markers cleanly.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn map_iteration_joined( &mut self, node: &str, index: u64, ) -> Result<(), RuntimeError>

Records (or replays) that one iteration of a map fan-out joined back into the map node’s output. Joins must be recorded in index order, never completion order, so the concurrency of the fan-out never influences the parent log’s byte sequence.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn now(&mut self) -> Result<OffsetDateTime, RuntimeError>

The recorded clock: reads the injected clock once, live, and replays the identical instant forever after.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn random(&mut self) -> Result<u64, RuntimeError>

The recorded random source: draws 64 bits from the injected source once, live, and replays the identical bits forever after. Richer random values must be derived from these bits deterministically.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn model_call( &mut self, client: &Client, request: &MessageRequest, ) -> Result<ModelTurn, RuntimeError>

A recorded model call.

The request is identified by its content hash (sha256: over the canonical serialization; see [crate::hash]). Replayed: the recorded response is decoded and returned; the provider is never contacted. Live: the intent event is persisted, the provider is called through client, and the completion (response plus usage) is persisted. A recorded intent with no completion (a call the process died inside) is re-issued safely: the fresh completion correlates to the recorded intent.

When with_record_prompts is on, the exact request body is recorded alongside the hash on the fresh live intent. It is the same value the hash was computed over, it never feeds into the hash, and replay ignores it, so recording it changes nothing about how the run replays.

Errors

RuntimeError::Replay on divergence, RuntimeError::Store when persistence fails, RuntimeError::Model when the live provider call fails (the log stays intact and the run is recoverable), RuntimeError::RequestEncode / RuntimeError::RecordedResponseDecode on the JSON edges.

pub async fn model_call_streaming( &mut self, client: &Client, request: &MessageRequest, on_event: impl FnMut(&StreamEvent), ) -> Result<ModelTurn, RuntimeError>

A recorded model call that streams live events to on_event while it runs, recording the identical completion model_call would record.

This is a live-progress affordance layered on top of the durable record, not a different kind of call. The recorded log is byte-for-byte what model_call writes for the same underlying response: the request is hashed the same way (see [crate::hash]), the intent is the same ModelCallRequested, and the completion carries the same response value and usage. A run does not care which path recorded it, and replay is deterministic either way.

Replayed: the recorded response is decoded and returned, exactly as model_call does. The provider is never contacted and on_event never fires, because there are no live tokens to report; the caller gets the final result at once.

Live: the intent event is persisted first (write-ahead, the same ordering model_call uses), then the provider stream is opened through client. Each StreamEvent is handed to on_event for a live ticker (text deltas ride StreamEvent::ContentBlockDelta, token counts ride StreamEvent::MessageDelta) and, in the same pass, applied to a MessageAccumulator. When the stream ends, the assembled MessageResponse is converted with the same response_value and usage logic model_call uses, the completion is persisted, and the ModelTurn is returned.

All persistence lives inside this method, so a caller cannot record a partial or wrong completion: the completion is written only after the stream is fully assembled. A caller that drops the returned future before the stream completes leaves a dangling model intent (the write-ahead intent with no completion), exactly like a live model_call the process died inside. That intent is re-issued safely on resume: the fresh completion correlates to the recorded intent. on_event firing is not part of the durable record, so a ticker that saw partial tokens before the drop has no effect on what replay produces.

When with_record_prompts is on, the exact request body is recorded on the fresh live intent, identically to model_call.

Errors

RuntimeError::Replay on divergence, RuntimeError::Store when persistence fails, RuntimeError::Model when the live stream fails (opening it, an error event or transport fault mid-stream, or a tool-call fragment that does not parse) surfaced as the same error type model_call returns, with the log left intact and the run recoverable, and RuntimeError::RequestEncode / RuntimeError::RecordedResponseDecode on the JSON edges.

pub async fn tool_call( &mut self, tool: &dyn DynTool, input: &Value, idempotency_key: Option<&str>, ) -> Result<ToolCallResult, RuntimeError>

A recorded tool call: one intent/completion pair, whatever happens in between.

Replayed: the recorded completion output is decoded (an output, a failure object, or a suspension sentinel; see [crate::wire]) and the tool is never executed. Live: the intent is persisted before the tool executes (write-ahead), the tool runs with retries per its effect’s RetryPolicy (see MAX_TOOL_ATTEMPTS), and the completion is persisted. A recorded Read/Idempotent intent with no completion re-executes here under its recorded idempotency key; a dangling Write intent fails with ReplayError::NeedsReconciliation before anything runs.

idempotency_key is the key for a fresh call; the built-in loop derives it from random for Idempotent tools so it reproduces on replay. A key the tool declares for itself (DynTool::idempotency_key) takes precedence over the one passed here, because only the tool can say what effect a call is. For a re-executed recorded intent the recorded key wins, and whatever is presented must match it (the cursor checks).

Cross-run deduplication

Within one run, nothing happens twice because a recorded completion is replayed rather than re-executed. Across independent runs there is no log to replay, so something else has to hold the line, and that something is the idempotency key.

Which keys count

Only a key the tool declares for itself, through DynTool::idempotency_key, is an identity to deduplicate on. A key the runtime derives on a tool’s behalf is not, and the difference is not a technicality.

A hand-written tool makes that declaration in Rust. An MCP or wasm tool has no code here to make it in, so its operator does, by naming the input field that identifies a call in the agent file (idempotency_keys); the tool derives the key from that field on every call and answers through the same trait method. Nothing below this distinguishes the two, because there is no distinction to make: both are a statement about what the call does in the world, from someone in a position to know.

A declared key is a statement about the world: "pay_claim:wreck-9931" means this is the payout for claim 9931, and two calls carrying it are the same payment no matter which run asked for them. A derived key says something much weaker. The built-in loop draws one from recorded randomness so a retry within a run reuses it; the graph engine derives one from a node’s position so a fork re-executing a node reuses it. Both are attempt identifiers, scoped to one run or one lineage, and two unrelated runs can hold the same derived key over completely different arguments. Treating one as an effect identity would let a second run collect the first run’s output for a call it never made, which is a worse failure than the duplicate execution this is meant to stop.

So a derived key keeps doing exactly what it always did, at the provider and inside its own run, and is recorded exactly as before. Cross-run deduplication waits for a tool to say what its calls are.

A declared key is also what a call records, in preference to a derived one, since only the tool can name its own effect.

The mechanism

The decision is made here, live, before the intent is recorded and before the tool runs. For a Effect::Write or Effect::Idempotent call carrying a declared key, this method claims the identity (tool name, idempotency key) in the store (EventStore::claim_call), which is the arbiter:

• Claimed. This run is the one execution. The intent is recorded, the tool runs, and the completion is appended and settles the commitment as one atomic step, so no crash can leave a committed completion the store still calls unfinished.
• Held, and settled. An equal call is already committed. The origin run’s log is read back (through read_log, so its hash chain is verified before a single byte is copied), its recorded input is checked against this call’s, and its output becomes this call’s output. The intent is still recorded, because an intent that resolves as a duplicate is an honest thing to have recorded, and the completion carries a DedupOrigin naming what it copied. The tool does not run.
• Held, and unfinished. Refused with RuntimeError::CallInFlight, before anything is recorded. See that variant for why refusing beats guessing.

A call with no declared key is untouched by any of this: there is no identity to deduplicate on, so a keyless write behaves exactly as it always has, and so does a write carrying only a derived key. So does every Effect::Read, which has no effect worth naming.

Replay never participates. A recorded completion replays from the log, whether it was witnessed or copied, with no store lookup of any kind; the DedupOrigin on it is read by humans and audits, never by the cursor. That is what keeps a recorded log a self-contained description of a run.

The one place resume consults the store is the gap a crash can leave between a deduplicated intent and its copied completion. That intent executed nothing (this run never held the identity, so it never held the right to execute), and the store can prove it, so the call is finished as the duplicate it was rather than parked for a human. Every other dangling write still parks: see recover_deduplicated_intent.

Errors

RuntimeError::Replay on divergence or a dangling write intent; RuntimeError::Store when persistence fails; RuntimeError::CallInFlight when another run holds this call’s identity and has not finished with it; RuntimeError::IdempotencyKeyCollision when one key names two different calls; RuntimeError::CommitmentUnreadable when the store points at a completion its own log does not hold. A failing tool is not an Err: it returns ToolCallResult::Failed, because the failure is a recorded outcome the orchestration must handle deterministically.

pub async fn suspend( &mut self, reason: &str, input_schema: &Value, ) -> Result<(), RuntimeError>

Parks the run: records Suspended { reason, input_schema }. Follow with await_resume.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn await_resume(&mut self) -> Result<Resumption, RuntimeError>

Obtains the input a parked run was resumed with.

Replayed: the recorded Resumed input. Live: when a resume input was provided through set_resume_input, it is recorded and returned; otherwise the run stays parked and Resumption::Parked tells the caller to stop driving.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn budget_exceeded( &mut self, budget: Budget, observed: f64, ) -> Result<(), RuntimeError>

Records a budget crossing. The check that led here must be computed from replayed data (recorded usage, recorded now observations) so it re-fires identically on replay. Follow with await_resume, exactly like a suspension.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.

pub async fn complete_run(&mut self, output: &Value) -> Result<(), RuntimeError>

Completes the run with output. Every request after this is a divergence.

Errors

RuntimeError::Replay on divergence (including an output that does not match the recorded one); RuntimeError::Store when persistence fails.

pub async fn fail_run(&mut self, error: &str) -> Result<(), RuntimeError>

Fails the run with error. Every request after this is a divergence.

Errors

RuntimeError::Replay on divergence; RuntimeError::Store when persistence fails.