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 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. For a re-executed recorded intent the recorded key wins, whatever is passed here must match it (the cursor checks).

Errors

RuntimeError::Replay on divergence or a dangling write intent; RuntimeError::Store when persistence fails. 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.