Skip to main content

pointlock_runner/
engine.rs

1//! The M2 execution engine: tree-walking sequential execution of the
2//! control-flow step vocabulary (action/call/if/foreach/let/assert) with
3//! the spine §6.1/§6.2 event-order discipline.
4//!
5//! Ledger discipline per action step (verbatim order, spine §6.1 M1 note):
6//! ready (argument snapshot frozen) → `stepEntered` (carries the step's
7//! effect/judge hashes and the resolved-inputs snapshot) → probing
8//! (declared `preflight` only: fresh observe material → `preflightProbed`)
9//! → `actionIntent` (own transaction, fsynced *before* dispatch) →
10//! `provider.execute` → `actionSettled` → error classification (spine §5)
11//! → `observationRecorded` (evidence localized first, file-before-row-
12//! before-log) → `assertionEvaluated` per assertion → `verdictRecorded`
13//! (when a verdict exists) + `ProviderSession::record_verdict` write-back →
14//! `stepExited` (carries the projected output, when one exists).
15//!
16//! Control-flow steps (M2):
17//! - `call` (07 §1): call-by-value in both directions — inputs evaluated in
18//!   the caller scope, snapshotted and schema-gated inbound; the callee
19//!   body runs in a fresh frame (`params` = inputs, `env` read-only
20//!   pass-through, the caller's steps/vars invisible — 07 §1.2 verbatim);
21//!   declared outputs evaluated in the callee scope and schema-gated
22//!   outbound; `callFramePushed`/`callFramePopped` bracket the frame; the
23//!   call step's verdict *is* the callee's flow verdict (spine §6.3).
24//! - `if`: strict-boolean `cond`; the unselected branch's steps each leave
25//!   an `entered(resolvedInputs: null)`/`exited(skipped)` pair (the
26//!   blocked precedent — ledger completeness); containers yield no verdict
27//!   of their own (R4).
28//! - `foreach`: `items` must evaluate to an array; each round runs the body
29//!   under an `iteration` path frame (`[i]`) with `iter.<as>` bound; the
30//!   `stepEntered` snapshot carries `{ items, as }` (the fold's IterState
31//!   carrier and the resume-time position authority — 07 §4.6).
32//! - `let`: pure bindings into the frame's `vars.*` (SSA; rebinding is a
33//!   compiler-refused shape — the runtime check is a defense line).
34//! - `assert`: `observe: "fresh"` captures via `session.observe` and goes
35//!   through the same localization as action observations; `fromStep`
36//!   replays the archived material of a prior action step — zero device
37//!   I/O.
38//!
39//! Evidence-localization degradation (M2): a failure to localize
40//! (`fetch_evidence` unsupported, stream rupture, integrity mismatch,
41//! `ui.snapshot.get` failure) never aborts the run — the observation
42//! record keeps the affected field absent and the dependent verify channel
43//! receives a typed gap, degrading honestly toward `unknown` (principle 4).
44//!
45//! The stop token is honored at step boundaries at any depth
46//! (`runSuspended` → [`RunOutcome::Suspended`]); suspension leaves the
47//! open spans and live frames in place, and resume walks back into the
48//! exact frame position (07 §4.6) by adopting completed step instances
49//! path-by-path.
50
51use std::collections::{BTreeMap, BTreeSet};
52use std::fmt::Write as _;
53use std::sync::Arc;
54use std::time::{SystemTime, UNIX_EPOCH};
55
56use futures_util::StreamExt;
57use futures_util::future::LocalBoxFuture;
58use pointlock_expr::Scope;
59use pointlock_ir::{
60    ActionExecution, ActionOutcome, ActionResult, ActionStepIR, AssertStepIR, AssertionIR,
61    AssetRef, BoundAttempt, CallFrame, CallStepIR, EffectClassAction, ErrorClass, ErrorInfo,
62    EvidenceRef, ExecutionMode, FlowIR, ForeachStepIR, HandlerAction, HandlerBinding, HandlerHook,
63    HumanMode, HumanPending, HumanPurpose, HumanStepIR, IfStepIR, LetStepIR, Observation,
64    ObservationRecord, ObservationSource, ParamDecl, PathFrame, Phase, PredicateIR,
65    ProviderStateSummary, RetryPolicy, RunLogPayload, RunPath, StepBase, StepIR, StepId,
66    StepRecord, StepState, SupervisePolicy, UiSnapshotOmissionReason, Verdict, VerdictStatus,
67    VerifyChannel, render_run_path, to_canonical_json,
68};
69use pointlock_provider_kit::{
70    BoundActionCall, CancellationToken, ObserveRequest, ObserveWant, ProviderSession,
71    SessionOutcome, UiSnapshotOutcome, VERDICT_EVIDENCE_MAX_ENTRIES, VERDICT_SUMMARY_MAX_CHARS,
72    VerdictWrite,
73};
74use pointlock_store::Store;
75use pointlock_vision::VisionVerifier;
76use serde_json::{Map, Value};
77
78use crate::error::{BlockedReason, RunnerError};
79use crate::judge::{
80    FoldedVerdict, eval_expr_assertion, fold_flow_verdict, fold_step_verdict, project_output,
81};
82use crate::load::{LoadedFlow, MAX_CALL_DEPTH};
83use crate::observe_eval::{
84    EvaluatedAssertion, ObserveMaterial, eval_observed_assertion, material_from_observation,
85};
86
87/// Terminal outcome of `Runner::run` / `Runner::resume`.
88#[derive(Debug, Clone, PartialEq)]
89pub enum RunOutcome {
90    /// The run reached `runFinished`. `verdict` is the folded flow verdict;
91    /// absent when no step produced a verdict (all-unverified flows) or
92    /// when the run was aborted by a `cancelled` action terminal.
93    Finished {
94        /// The folded flow verdict, when one exists.
95        verdict: Option<Verdict>,
96    },
97    /// The run reached `runSuspended` (stop token at a step boundary, or a
98    /// provider error left an action without a terminal — resume
99    /// reconciles it).
100    Suspended,
101    /// The run cannot proceed without a human decision: a drifted
102    /// preflight whose `onResumeDrift` ladder is exhausted (or absent), or
103    /// a reconcile adjudication that could not even be requested (defense
104    /// line). The store records a `runSuspended` with the reason.
105    Blocked {
106        /// Why a human is required.
107        reason: BlockedReason,
108    },
109    /// A human interaction (human step or R13 supervision gate) is
110    /// pending: `humanRequested` is fsynced and `runSuspended` recorded —
111    /// the runner never blocks waiting. The attached-TTY inline experience
112    /// lives in the CLI layer (collect a response through the store
113    /// arbitration, then resume in the same process); resume settles the
114    /// paired response, re-awaits an unanswered one, or lazily settles an
115    /// expired deadline to `unknown` (06 §5.3).
116    AwaitingHuman {
117        /// The pending request (also materialized in
118        /// `CheckpointView.humanPending`).
119        pending: HumanPending,
120    },
121}
122
123/// Shared deadline of the two live capture RPCs (`health()` +
124/// `currentCursor()`, 07 §2.2): capture failures degrade the affected
125/// fields — they never block the suspend/exit path.
126const SUMMARY_CAPTURE_BUDGET_MS: u64 = 2_000;
127
128/// Captures the failure/suspension-instant provider profile (07 §2.2,
129/// incorporated 2026-07-18). Pure forensics: nothing downstream may
130/// consume it as a control input. Every failure mode degrades honestly:
131/// a failed `health()` records `{ ok: false, degraded: <class> }`, a
132/// failed `currentCursor()` leaves the cursor absent (never a stale
133/// bind-time value — principle 4).
134pub(crate) async fn capture_provider_state_summary(
135    session: &dyn ProviderSession,
136    known_lineage: &[String],
137    device_id: &str,
138    platform: Option<&str>,
139) -> ProviderStateSummary {
140    let budget = std::time::Duration::from_millis(SUMMARY_CAPTURE_BUDGET_MS);
141    let started = std::time::Instant::now();
142    let health = match tokio::time::timeout(budget, session.health()).await {
143        Ok(Ok(health)) => pointlock_ir::SessionHealthSnapshot {
144            ok: health.ok,
145            degraded: health.degraded,
146        },
147        Ok(Err(error)) => pointlock_ir::SessionHealthSnapshot {
148            ok: false,
149            degraded: Some(
150                serde_json::to_value(error.error_class)
151                    .ok()
152                    .and_then(|value| value.as_str().map(str::to_owned))
153                    .unwrap_or_else(|| "unknown".to_owned()),
154            ),
155        },
156        Err(_) => pointlock_ir::SessionHealthSnapshot {
157            ok: false,
158            degraded: Some("capture_timeout".to_owned()),
159        },
160    };
161    let remaining = budget.saturating_sub(started.elapsed());
162    let event_cursor = match tokio::time::timeout(remaining, session.current_cursor()).await {
163        Ok(Ok(cursor)) => Some(cursor),
164        _ => None,
165    };
166    let attestation = session.attestation();
167    let mut session_lineage = known_lineage.to_vec();
168    if let Some(cursor) = &event_cursor
169        && session_lineage.last() != Some(&cursor.session_id)
170    {
171        session_lineage.push(cursor.session_id.clone());
172    }
173    ProviderStateSummary {
174        session_lineage,
175        event_cursor,
176        attestation: pointlock_ir::AttestationSnapshot {
177            lockfile_digest: attestation.lockfile_digest.clone(),
178            attested_at: attestation.attested_at.clone(),
179        },
180        health,
181        device_id: device_id.to_owned(),
182        platform: platform.map(str::to_owned),
183    }
184}
185
186/// The manifest of a locally minted human-evidence asset (item ③):
187/// localized by construction (put_evidence wrote the bytes before the
188/// verdict cites them).
189/// The cited/manifest pair of an escalate ruling's verdict (06 §6): the
190/// settlement document when the ruling carries one, empty otherwise.
191fn escalate_verdict_material(evidence: &Option<AssetRef>) -> (Vec<AssetRef>, EvidenceManifest) {
192    match evidence {
193        Some(asset) => (vec![asset.clone()], human_manifest(asset)),
194        None => (Vec::new(), EvidenceManifest::default()),
195    }
196}
197
198fn human_manifest(asset: &AssetRef) -> EvidenceManifest {
199    EvidenceManifest {
200        localized: vec![pointlock_ir::EvidenceRef {
201            asset: asset.clone(),
202            sha256: asset.sha256.clone().unwrap_or_default(),
203            local_path: asset.uri.clone(),
204        }],
205        gaps: Vec::new(),
206    }
207}
208
209/// One judgment's settlement-evidence localization outcome (item ③):
210/// what landed and what typed-failed. Rides the `verdictRecorded`
211/// payload; observation assets are excluded (they ride
212/// `observationRecorded`).
213#[derive(Debug, Clone, Default)]
214pub(crate) struct EvidenceManifest {
215    /// Localized copies (evidence table + `localized` payload field).
216    pub localized: Vec<pointlock_ir::EvidenceRef>,
217    /// Typed failures (`localizationGaps` payload field).
218    pub gaps: Vec<pointlock_ir::EvidenceGap>,
219}
220
221/// The act-chain re-entry position of a crash-resume (item ②, 07 §1.4:
222/// resume lands at the precise position, never restarts the chain): the
223/// recorded 1-based `chainIndex` maps to its 0-based enumeration slot;
224/// an out-of-range index is a TYPED refusal (never a guessed position —
225/// unreachable via the shipped resume rules); a pre-incorporation
226/// ledger (no index) falls back to the head — the pre-ruling behavior,
227/// honest under principle 4.
228pub(crate) fn chain_start(
229    chain_index: Option<u32>,
230    step: &ActionStepIR,
231) -> Result<usize, RunnerError> {
232    match chain_index {
233        None => Ok(0),
234        Some(index) if index >= 1 && ((index - 1) as usize) < step.binding.attempts.len() => {
235            Ok((index - 1) as usize)
236        }
237        Some(index) => Err(RunnerError::M0Unsupported {
238            detail: format!(
239                "the pending intent's recorded chainIndex {index} does not exist in the \
240                 resumed step's binding chain (len {}); refusing to guess a re-entry \
241                 position (unreachable through the shipped resume rules — same-IR chains \
242                 cannot shrink and effect-dirty repairs never adopt/replay)",
243                step.binding.attempts.len()
244            ),
245        }),
246    }
247}
248
249/// SPI ingestion quarantine (M3a viewport review): `Viewport.scaleFactor`
250/// is the only f64 in the durable event domain, and serde_json writes a
251/// non-finite f64 as `null` — a value the ledger would never read back
252/// (every later refold/verify/projection of the run fails permanently).
253/// A `succeeded` terminal embedding one is a provider contract violation:
254/// it is recorded as a *final failure* with a precise code instead — the
255/// honest ledger fact ("the provider reported an unpersistable
256/// terminal"), taking the ordinary failure path (handlers may escalate)
257/// rather than poisoning the ledger or falsifying the observation.
258pub(crate) fn quarantine_unpersistable(outcome: ActionOutcome) -> ActionOutcome {
259    let poisoned = match &outcome {
260        ActionOutcome::Succeeded { result } => result
261            .before
262            .iter()
263            .chain(result.after.iter())
264            .find(|observation| !observation.viewport.scale_factor.is_finite()),
265        _ => None,
266    };
267    match poisoned {
268        Some(observation) => ActionOutcome::Failed {
269            error: ErrorInfo {
270                code: "observation_viewport_invalid".to_owned(),
271                message: format!(
272                    "provider contract violation: observation {} carries a non-finite \
273                     viewport scaleFactor, which cannot be persisted",
274                    observation.id
275                ),
276                retryable: false,
277                details: None,
278            },
279        },
280        None => outcome,
281    }
282}
283
284/// Milliseconds since the Unix epoch (informational `atMs` on events).
285pub(crate) fn now_ms() -> u64 {
286    SystemTime::now()
287        .duration_since(UNIX_EPOCH)
288        .map(|duration| duration.as_millis() as u64)
289        .unwrap_or(0)
290}
291
292/// The root run path of a flow (hard rule: flow frames carry the irHash).
293pub(crate) fn root_path(flow: &FlowIR) -> RunPath {
294    vec![PathFrame::Flow {
295        flow_id: flow.flow_id.clone(),
296        ir_hash: flow.ir_hash.clone(),
297    }]
298}
299
300/// Extracts the last attempt number of a run path.
301pub(crate) fn attempt_of(path: &RunPath) -> Option<u64> {
302    path.iter().rev().find_map(|frame| match frame {
303        PathFrame::Attempt { n } => Some(*n),
304        _ => None,
305    })
306}
307
308/// The IR-version-independent identity of a step *instance*: stepId path
309/// plus iteration indexes, with hashes and attempt/phase suffixes
310/// stripped. Stable across a repair (irHash change), unique within a run
311/// (stepIds are flow-unique; iterations disambiguate rounds). Used to key
312/// adoption, open spans, and attempt watermarks.
313/// Whether a `--stop-at` / `--stop-after` breakpoint target names the
314/// step instance at `path` (a step-instance path: prefix + child frame).
315///
316/// Target grammar (CLI-facing, documented on `RunOptions::stop_at`):
317/// - a **bare step id** (no `/`, `[` or `@`) matches the first instance
318///   whose step id equals it — the shorthand;
319/// - otherwise a **run path**: either the canonical rendering
320///   (`flow@hash8/each[1]/tap`) or the same path relative to the root
321///   flow segment (`each[1]/tap`), compared by rendered equality, so an
322///   iteration-qualified path matches exactly that iteration.
323pub(crate) fn breakpoint_matches(target: &str, path: &[PathFrame]) -> bool {
324    let bare_id = !target.contains(['/', '[', '@']);
325    if bare_id {
326        return match path.last() {
327            Some(PathFrame::Step { step_id }) => step_id.as_str() == target,
328            Some(PathFrame::Call {
329                step_id: Some(step_id),
330                ..
331            }) => step_id.as_str() == target,
332            _ => false,
333        };
334    }
335    let rendered = render_run_path(path);
336    rendered == target
337        || rendered
338            .split_once('/')
339            .is_some_and(|(_, rel)| rel == target)
340}
341
342/// The structured `runSuspended.reason` of a breakpoint stop:
343/// `stopped at breakpoint <flag> <canonical instance path>` (the matched
344/// instance, never the shorthand the user typed).
345fn breakpoint_reason(flag: &str, path: &[PathFrame]) -> String {
346    format!("stopped at breakpoint {flag} {}", render_run_path(path))
347}
348
349pub(crate) fn instance_key(path: &[PathFrame]) -> String {
350    // The per-attempt suffix (attempt/phase/assertion frames) is never
351    // part of the instance identity — but only the TRAILING run of such
352    // frames is a suffix. Stripping trailing-only (instead of breaking
353    // at the first attempt frame) is byte-identical for every path shape
354    // the engine produces today, and stops aliasing distinct
355    // interior-attempt instances once the ruled attempt-framed call
356    // re-entry of 07 §1 lands (interior attempts render as `#n`).
357    let trimmed = {
358        let mut end = path.len();
359        while end > 0
360            && matches!(
361                path[end - 1],
362                PathFrame::Attempt { .. } | PathFrame::Phase { .. } | PathFrame::Assertion { .. }
363            )
364        {
365            end -= 1;
366        }
367        &path[..end]
368    };
369    let mut key = String::new();
370    for frame in trimmed {
371        match frame {
372            PathFrame::Flow { .. } => {}
373            PathFrame::Step { step_id } => {
374                let _ = write!(key, "/{step_id}");
375            }
376            PathFrame::Call { step_id, .. } => {
377                let _ = match step_id {
378                    Some(step_id) => write!(key, "/{step_id}"),
379                    None => write!(key, "/hook-call"),
380                };
381            }
382            PathFrame::Iteration { index, key: item } => {
383                let _ = match item {
384                    Some(item) => write!(key, "[{index}:{item}]"),
385                    None => write!(key, "[{index}]"),
386                };
387            }
388            PathFrame::Hook { hook, trigger } => {
389                let _ = write!(key, "/hook:{hook:?}:{trigger}");
390            }
391            PathFrame::Attempt { n } => {
392                let _ = write!(key, "#{n}");
393            }
394            PathFrame::Phase { .. } | PathFrame::Assertion { .. } => {}
395        }
396    }
397    key
398}
399
400/// The path frame a step contributes below its parent prefix: call steps
401/// contribute a `call` frame (one frame, two rendered segments — 07 §2.1),
402/// every other kind a plain `step` frame.
403pub(crate) fn child_frame(step: &StepIR) -> PathFrame {
404    match step {
405        StepIR::Call(call) => PathFrame::Call {
406            step_id: Some(call.base.step_id.clone()),
407            callee_flow_id: call.flow_ref.flow_id.clone(),
408            callee_ir_hash: call.flow_ref.ir_hash.clone(),
409        },
410        other => PathFrame::Step {
411            step_id: other.step_id().clone(),
412        },
413    }
414}
415
416/// Mid-flight work for the frontier step of a resume (07 §4.4 decision
417/// table). All variants imply the step's `stepEntered` span is already
418/// open in the log — the engine must not re-enter it.
419pub(crate) enum FrontierWork {
420    /// `reconcile → completed`: adopt the archived terminal — append the
421    /// `actionSettled` the crash swallowed, then dispose it through the
422    /// exact same settled-outcome path as a live execute (§6.7-B).
423    Adopt {
424        /// The reconciled callId.
425        call_id: String,
426        /// The run path of the original `actionIntent` (the settle anchors
427        /// to the same attempt).
428        intent_path: RunPath,
429        /// The archived terminal outcome, verbatim.
430        outcome: Box<ActionOutcome>,
431        /// The archived `argsSnapshot` (never re-evaluated, spine §6.6).
432        args: Value,
433        /// The intent's recorded 1-based chain position (item ②): the
434        /// act chain re-enters HERE, per 07 §1.4's resume-lands-at-the-
435        /// precise-position rule. Absent on pre-incorporation ledgers
436        /// (falls back to the chain head — the pre-ruling behavior).
437        chain_index: Option<u32>,
438    },
439    /// `reconcile → neverDispatched` (or an authorized uncertain replay):
440    /// dispatch again using the archived args snapshot — never
441    /// re-evaluated (spine §6.6).
442    Replay {
443        /// The archived `argsSnapshot` from the pending intent.
444        args: Value,
445        /// The intent's recorded chain position (see `Adopt.chain_index`):
446        /// the replay re-dispatches THIS attempt — a mid-chain crashed
447        /// intent's args belong to that attempt, not the chain head.
448        chain_index: Option<u32>,
449    },
450    /// A human `adopt` adjudication of an uncertain reconcile (07 §4.4):
451    /// the ruling says the effect stands, so nothing is dispatched; the
452    /// step proceeds straight to the observation-confirmation path
453    /// ([`ActPhase::Unconfirmed`]) and its assertions verify the ruled
454    /// world.
455    ConfirmEffect {
456        /// The adjudication context, human-readable.
457        message: String,
458        /// The archived ready snapshot (the span is open; never
459        /// re-evaluated).
460        args: Value,
461    },
462    /// A human `abort` adjudication of an uncertain reconcile (07 §4.4):
463    /// close the open span `aborted` and abort the run.
464    AbortRuled {
465        /// The archived ready snapshot.
466        args: Value,
467    },
468}
469
470/// A reconciled frontier terminal whose WAL intent and `actionSettled` are
471/// already on record: the act chain consumes it as the first try's settled
472/// outcome instead of dispatching — one disposal code path for live and
473/// adopted terminals.
474struct AdoptedSettle {
475    /// The archived terminal outcome.
476    outcome: ActionOutcome,
477    /// The `seq` of the appended `actionSettled` event (evidence linking).
478    settled_seq: u64,
479    /// The attempt number of the original intent.
480    attempt_n: u64,
481}
482
483/// How a step's execution affects the surrounding body walk.
484enum Ctl {
485    /// Step concluded (pass/unknown/no-verdict); continue with the next.
486    Continue,
487    /// A fail verdict: halt — the remaining steps of the current body are
488    /// recorded `blocked`, and the halt propagates through enclosing
489    /// containers and frames (a callee halt folds into the call step's
490    /// verdict, spine §6.3).
491    HaltFail,
492    /// No terminal could be obtained (transport-class provider error) or a
493    /// stop was requested: suspend the run; open spans and live frames
494    /// stay open for the frame-precise resume (07 §4.6).
495    Suspend(String),
496    /// A `cancelled` terminal: the step is recorded `aborted` and the run
497    /// finishes without a flow verdict (spine §5).
498    Abort,
499    /// The run cannot proceed without a human (a drifted preflight whose
500    /// `onResumeDrift` ladder is exhausted or absent).
501    Blocked(BlockedReason),
502    /// A human request is pending (human step or supervision gate): the
503    /// run suspends (`runSuspended` after the fsynced `humanRequested`)
504    /// and surfaces [`RunOutcome::AwaitingHuman`]. Open spans and live
505    /// frames stay open, exactly like `Suspend`.
506    AwaitHuman(HumanPending),
507}
508
509/// What a handler consultation decided for its host step (spine §3: a
510/// disposition, never data — R10).
511enum Consulted {
512    /// No binding matched, or the trigger budget is exhausted: the
513    /// natural path stands.
514    None,
515    /// Re-enter the failing phase under the handler's retry policy
516    /// (budget independent of `StepBase.retry` — spine §6.5 mount 2).
517    Retry(RetryPolicy),
518    /// Record and release: the verdict stands, downstream is not halted.
519    Continue,
520    /// Abort the run.
521    Abort,
522    /// An escalate human superseded the host outcome with this status.
523    Escalated {
524        status: VerdictStatus,
525        summary: String,
526        /// The canonical settlement evidence document (06 §6) — cited by
527        /// the superseding verdict. Absent only on pre-doc paths.
528        evidence: Option<AssetRef>,
529    },
530    /// An escalate human (repairWorld) declared the world repaired:
531    /// re-enter the failing phase once.
532    Repaired,
533    /// An escalate human is pending: suspend awaiting the response.
534    Pending(HumanPending),
535    /// The repair subflow completed cleanly: re-enter the failing phase.
536    RepairDone,
537    /// The repair subflow failed: the host's natural path stands (the
538    /// repair flow's own verdict records carry the failure detail).
539    RepairFailed,
540    /// The repair subflow hit a control outcome (suspend/awaiting-human/
541    /// blocked): propagate it.
542    Propagate(Ctl),
543}
544
545/// The hook audit frame (`/hook:<name>:<n>`, 07 §2.1).
546fn hook_frame(hook: HandlerHook, trigger: u64) -> PathFrame {
547    PathFrame::Hook { hook, trigger }
548}
549
550/// The run path of an escalate hook human: host + hook frame + step frame.
551fn hook_child_path(
552    step_path: &RunPath,
553    hook: HandlerHook,
554    trigger: u64,
555    human: &HumanStepIR,
556) -> RunPath {
557    let mut path = step_path.clone();
558    path.push(hook_frame(hook, trigger));
559    path.push(PathFrame::Step {
560        step_id: human.base.step_id.clone(),
561    });
562    path
563}
564
565/// Maps an escalate human's arbitrated response to a consultation outcome
566/// (the four-mode table of 06 §2.2, narrowed to the escalate context).
567fn map_escalate_response(human: &HumanStepIR, response: &Value) -> Consulted {
568    let decision = response
569        .get("decision")
570        .and_then(Value::as_str)
571        .unwrap_or_default();
572    match human.mode {
573        HumanMode::RepairWorld => match decision {
574            // 06 §2.1's closed repairWorld vocabulary. In the escalate
575            // ladder `done` re-enters the disposition (re-probe / re-act);
576            // `cannotRepair` is the human's explicit "the world cannot be
577            // brought back" — the run aborts rather than looping on a
578            // declared impossibility. The catch-all is defense: the store
579            // arbitrates the vocabulary before anything reaches here.
580            "done" => Consulted::Repaired,
581            _ => Consulted::Abort,
582        },
583        HumanMode::Confirm => {
584            let first = human
585                .decisions
586                .as_ref()
587                .and_then(|labels| labels.first())
588                .map(String::as_str);
589            let status = if first == Some(decision) {
590                VerdictStatus::Pass
591            } else {
592                VerdictStatus::Fail
593            };
594            Consulted::Escalated {
595                status,
596                summary: format!("escalate confirm decision '{decision}' (position-mapped)"),
597                evidence: None,
598            }
599        }
600        // Judge (provideInput escalates are refused at load).
601        _ => {
602            let status = match response.get("status").and_then(Value::as_str) {
603                Some("pass") => VerdictStatus::Pass,
604                Some("fail") => VerdictStatus::Fail,
605                _ => VerdictStatus::Unknown,
606            };
607            Consulted::Escalated {
608                status,
609                evidence: None,
610                summary: format!(
611                    "escalate judge ruling: {}",
612                    response
613                        .get("status")
614                        .and_then(Value::as_str)
615                        .unwrap_or("unknown")
616                ),
617            }
618        }
619    }
620}
621
622/// The facts of one `humanRequested` on the ledger, harvested for resume
623/// settlement (the log is the truth, I1). A supervision `suspend` answer
624/// is non-final and never fills `final_response` (spine §6.9).
625#[derive(Debug, Clone)]
626pub(crate) struct HumanRequestFact {
627    /// The request id a response must pair with.
628    pub request_id: String,
629    /// The request's anchor path (the awaiting/gated step).
630    pub run_path: RunPath,
631    /// Step vs supervision gate.
632    pub purpose: HumanPurpose,
633    /// Interaction mode (`purpose="step"` only).
634    pub mode: Option<HumanMode>,
635    /// The prompt shown to the human.
636    pub prompt: String,
637    /// The materialized presents snapshot (cited by the evidence doc).
638    pub presents: Value,
639    /// Absolute deadline; absent for supervision requests.
640    pub deadline_at_ms: Option<u64>,
641    /// The paired final response payload, when one was arbitrated.
642    pub final_response: Option<Value>,
643    /// Who gave the final response.
644    pub final_actor: Option<String>,
645}
646
647/// Outcome of the act phase (attempt chain + in-attempt retry).
648enum ActPhase {
649    /// A `succeeded` terminal.
650    Succeeded {
651        result: Box<ActionResult>,
652        /// Whether the provider reported an execution mode outside the
653        /// attempt's whitelist (§6.4 R-degrade).
654        degraded: bool,
655        /// The `seq` of the `actionSettled` event (evidence linking).
656        settled_seq: u64,
657        /// The attempt number the terminal settled on.
658        attempt_n: u64,
659    },
660    /// Step fails (final failure, exhausted retries, invalid arguments).
661    StepFail { class: ErrorClass, message: String },
662    /// Step folds to unknown (timeout without idempotence/retry; session
663    /// degradation).
664    StepUnknown {
665        message: String,
666        /// The error class that produced the unknown, when the class still
667        /// governs handler selection. `session_degraded` is the one the
668        /// spine §5 error table routes to a flow-level `onError` while the
669        /// step itself folds to unknown — dropping the class here would
670        /// silently send it to `onUnknown` instead, so a declared
671        /// `on_error: { error_classes: [session_degraded] }` would never
672        /// fire. A non-idempotent timeout keeps `None`: its row prescribes
673        /// the unknown path and no error hook.
674        error_class: Option<ErrorClass>,
675    },
676    /// The act's fate is sealed or ruled but its EFFECT is unproven, and
677    /// the step declares assertions that can ask the world. Two producers:
678    /// a `timedOut` terminal (spine §5 / 07 §4.3 — a recorded timeout is
679    /// certain and reconcile never upgrades it; the observe half of 「先
680    /// reconcile/observe 确认」 is what remains), and a human `adopt`
681    /// adjudication of an uncertain reconcile (07 §4.4 — the ruling says
682    /// the effect stands; the step's own assertions verify). The
683    /// settlement loop captures a fresh observation and evaluates the
684    /// assertions over it: decisive ones conclude pass/fail, indecisive
685    /// ones fold to unknown — exactly the path the bare uncertainty took
686    /// before.
687    Unconfirmed {
688        /// What made the effect unprovable, human-readable (prefixes the
689        /// verdict summary).
690        message: String,
691    },
692    /// A `cancelled` terminal.
693    Aborted,
694    /// No terminal (transport-class provider error) — suspend.
695    Suspend(String),
696}
697
698/// The localized before/after observation records of an executed action
699/// step — the material source of `observe: { fromStep }` assert steps.
700pub(crate) struct StepObs {
701    /// The localized records, in capture order.
702    pub observations: Vec<ObservationRecord>,
703    /// The before observation's id, when one was captured.
704    pub before_id: Option<String>,
705    /// The after observation's id, when one was captured.
706    pub after_id: Option<String>,
707}
708
709/// A completed step instance carried over by resume (adoption by exact
710/// instance path — 07 §4.6: completed steps enter as records, never
711/// re-execute).
712pub(crate) struct Adopted {
713    /// The archived record (fold output).
714    pub record: StepRecord,
715    /// The before observation's id (harvested from `actionSettled`).
716    pub before_id: Option<String>,
717    /// The after observation's id (harvested from `actionSettled`).
718    pub after_id: Option<String>,
719}
720
721/// Whether a record is execution history (vs a blocked/skipped accounting
722/// pair, which concluded nothing and seeds nothing).
723pub(crate) fn is_history(record: &StepRecord) -> bool {
724    !record.attempts.is_empty()
725        || record.verdict.is_some()
726        || record.output.is_some()
727        || !record.resolved_inputs.is_null()
728}
729
730/// One live execution frame: the root flow or a callee (07 §1.2 — a
731/// frame's full execution semantics are determined by
732/// `(calleeIrHash, inputsSnapshot, env)`). Scope contents never cross the
733/// frame boundary except read-only `env.*`.
734pub(crate) struct FrameState<'a> {
735    /// The flow executing in this frame.
736    pub flow: &'a FlowIR,
737    /// The frame's root path (`[flow]` for the root frame; up to and
738    /// including the `call` frame for a callee).
739    pub base_path: RunPath,
740    /// `params.*`: the run params (root) or the gated inputs snapshot.
741    pub params: Map<String, Value>,
742    /// `vars.*` accumulated by `let` steps (SSA).
743    pub vars: BTreeMap<String, Value>,
744    /// Live `iter.<as>` bindings, innermost last.
745    pub iters: Vec<(String, Value)>,
746    /// `steps.<id>.output` of concluded steps in this frame.
747    pub outputs: BTreeMap<String, Value>,
748    /// `steps.<id>.verdict` of concluded steps in this frame.
749    pub verdicts: BTreeMap<String, (VerdictStatus, bool)>,
750    /// Every step-instance verdict produced in this frame, in execution
751    /// order — the flow-verdict fold input (iteration instances count
752    /// individually; callee-internal verdicts fold through their call
753    /// step, never leak here).
754    pub fold: Vec<(VerdictStatus, bool)>,
755    /// Localized observations per executed action step (assert `fromStep`).
756    pub observed: BTreeMap<String, StepObs>,
757    /// Call depth (root = 1).
758    pub depth: usize,
759}
760
761impl<'a> FrameState<'a> {
762    /// A fresh frame over `flow`.
763    pub fn new(
764        flow: &'a FlowIR,
765        base_path: RunPath,
766        params: Map<String, Value>,
767        depth: usize,
768    ) -> Self {
769        FrameState {
770            flow,
771            base_path,
772            params,
773            vars: BTreeMap::new(),
774            iters: Vec::new(),
775            outputs: BTreeMap::new(),
776            verdicts: BTreeMap::new(),
777            fold: Vec::new(),
778            observed: BTreeMap::new(),
779            depth,
780        }
781    }
782
783    /// Materializes the closed evaluation scope of this frame (spine §7):
784    /// `params.* / env.* / vars.* / iter.<as> / steps.<id>.*`, plus an
785    /// optional self-output binding (raw output for projection, projected
786    /// output for assertions — 02 §4.1.1).
787    pub fn scope(&self, env: &[(String, Value)], self_binding: Option<(&str, &Value)>) -> Scope {
788        let mut scope = Scope::new();
789        for (name, value) in &self.params {
790            scope.set_param(name.clone(), value.clone());
791        }
792        for (name, value) in env {
793            scope.set_env(name.clone(), value.clone());
794        }
795        for (name, value) in &self.vars {
796            scope.set_var(name.clone(), value.clone());
797        }
798        for (name, value) in &self.iters {
799            scope.set_iter(name.clone(), value.clone());
800        }
801        for (step_id, output) in &self.outputs {
802            scope.set_step_output(step_id.clone(), output.clone());
803        }
804        for (step_id, (status, _degraded)) in &self.verdicts {
805            let status = serde_json::to_value(status).expect("VerdictStatus serializes");
806            scope.set_step_verdict(step_id.clone(), status);
807        }
808        if let Some((step_id, value)) = self_binding {
809            scope.set_step_output(step_id.to_owned(), value.clone());
810        }
811        scope
812    }
813
814    fn seed_verdict(&mut self, step_id: &StepId, status: VerdictStatus, degraded: bool) {
815        self.verdicts
816            .insert(step_id.as_str().to_owned(), (status, degraded));
817        self.fold.push((status, degraded));
818    }
819
820    /// Replaces the most recently seeded verdict (an escalate handler's
821    /// superseding judgment for the step it was consulted on — the host
822    /// verdict is by construction the last seeded entry at consultation
823    /// time).
824    fn reseed_last(&mut self, step_id: &StepId, status: VerdictStatus, degraded: bool) {
825        self.verdicts
826            .insert(step_id.as_str().to_owned(), (status, degraded));
827        self.fold.pop();
828        self.fold.push((status, degraded));
829    }
830}
831
832/// The single-run execution engine. Owns the provider session; borrows the
833/// single-writer store (the runner keeps store use single-threaded — async
834/// exists only because the SPI is async).
835pub(crate) struct Execution<'a> {
836    pub flows: &'a LoadedFlow<'a>,
837    pub session: Box<dyn ProviderSession>,
838    pub store: &'a mut Store,
839    pub run_id: String,
840    pub stop: CancellationToken,
841    /// Breakpoint before entry (`--stop-at`): the first step instance
842    /// matching the target suspends the run BEFORE it enters (no
843    /// `stepEntered`); see [`breakpoint_matches`]. Per segment. A span
844    /// re-entered from [`Self::open_spans`] is continued, not entered,
845    /// and never matches.
846    pub stop_at: Option<String>,
847    /// Breakpoint after exit (`--stop-after`): the first matching step
848    /// instance suspends the run right after its own `stepExited`, before
849    /// any following step enters. Per segment.
850    pub stop_after: Option<String>,
851    /// `env.*` bindings (deviceId / runId / platform): run-constant,
852    /// read-only pass-through across every frame (07 §1.2).
853    pub env: Vec<(String, Value)>,
854    /// Highest attempt number already used per step instance (resume
855    /// continues the numbering; empty on a fresh run).
856    pub attempt_base: BTreeMap<String, u64>,
857    /// Step spans left open by a crash/suspension: instance key → the
858    /// archived ready-phase snapshot. Execution re-enters these spans
859    /// without a second `stepEntered`, and containers reuse the archived
860    /// snapshot instead of re-evaluating (spine §6.6).
861    pub open_spans: BTreeMap<String, Value>,
862    /// Call frames already pushed (and not popped) by a previous segment:
863    /// instance key → the callee `irHash` the open frame currently claims.
864    /// Resume must not push them again; when the pin moved under a
865    /// down-drill it re-enters them instead (07 §5.2 case (a)).
866    pub live_frames: BTreeMap<String, pointlock_ir::Hash>,
867    /// Completed step instances to adopt instead of executing, keyed by
868    /// instance path.
869    pub adoptable: BTreeMap<String, Adopted>,
870    /// Reconciled mid-flight work for the frontier step instance.
871    pub frontier: Option<(String, FrontierWork)>,
872    /// Whether this segment is a RESUME. It decides where the honest
873    /// `unprobed` mark belongs (07 §4.2 rule 1): a fresh run never
874    /// re-touches a world it stopped watching, so nothing in it is
875    /// unprobed.
876    pub resumed: bool,
877    /// Step ids released through the 07 §5.4 gate this segment. Step 3 of
878    /// that rule extends the preflight guard to every one of them, so each
879    /// is an `unprobed` site of its own when it declares no probes.
880    pub authorized: BTreeSet<String>,
881    /// Latch: the segment's re-entry step has been reached. Set the first
882    /// time a step gets as far as probing — adopted steps short-circuit
883    /// long before, so the first one that arrives here IS 07 §4.2's
884    /// 「resume 的首个待执行 step」.
885    pub reentry_seen: bool,
886    /// The vision verifier for `vision` verify-chain tails. `None` is
887    /// equivalent to the stub: the vision channel cannot complete and
888    /// reports `"vision verifier not configured"`.
889    pub vision: Option<Arc<dyn VisionVerifier>>,
890    /// Known session generations (checkpoint lineage; a fresh run seeds
891    /// the bind-time session). Best-effort input of the failure-instant
892    /// provider profile (07 §2.2).
893    pub session_lineage: Vec<String>,
894    /// Failure-instant provider profiles captured at verdict time, keyed
895    /// by step-instance key; attached to the span's `stepExited` by
896    /// `append` (intensional gate by construction) and discarded on a
897    /// superseding pass, an aborted follow-up exit, or span re-entry.
898    pub pending_summaries: BTreeMap<String, ProviderStateSummary>,
899    /// This segment's supervision policy (R13, spine §6.9): per segment,
900    /// never inherited. `None` — unsupervised.
901    pub supervise: Option<SupervisePolicy>,
902    /// Human requests on the ledger, keyed by step-instance key (resume
903    /// settlement input; empty on a fresh run).
904    pub human: BTreeMap<String, HumanRequestFact>,
905    /// Settled terminals on the ledger, keyed by step-instance key: the
906    /// re-entry material for open action spans whose act already settled
907    /// before a handler-wave suspension (never re-dispatch, I2).
908    pub settled: BTreeMap<String, crate::align::SettledFact>,
909    /// Recorded verdicts on the ledger, keyed by step-instance key: the
910    /// handler-consultation re-entry point on resume.
911    pub recorded_verdicts: BTreeMap<String, (VerdictStatus, bool)>,
912    /// Handler trigger watermarks ("{instance}|{hook}" → highest trigger
913    /// on the ledger for the instance's current execution life):
914    /// `maxTriggers` is a per-life loop bound — it counts across segments
915    /// (a resume never resets it) and is cleared only when a fresh
916    /// `stepEntered` opens a new life of the instance (`enter_step`;
917    /// the harvest mirrors it, 07 §5.2).
918    pub hook_triggers: BTreeMap<String, u64>,
919    /// Injectable wall clock for deadline computation and lazy timeout
920    /// settlement; `None` uses the system clock. The settlement *result*
921    /// is a pure function of `deadlineAtMs` and response presence — never
922    /// of the settlement instant (06 §5.3).
923    pub clock: Option<Arc<dyn Fn() -> u64 + Send + Sync>>,
924}
925
926impl<'a> Execution<'a> {
927    fn append(&mut self, path: &RunPath, payload: &RunLogPayload) -> Result<u64, RunnerError> {
928        // The 07 §2.2 attach point: a fail/unknown-verdict span exiting
929        // (any exit site — the gate is intensional, not an enumerated
930        // list) carries the verdict-instant provider profile. Aborted
931        // follow-up exits make no semantic claim and discard it; span
932        // re-entry invalidates a stale capture.
933        let enriched;
934        let payload = match payload {
935            RunLogPayload::StepEntered { .. } => {
936                self.pending_summaries.remove(&instance_key(path));
937                payload
938            }
939            RunLogPayload::StepExited {
940                state,
941                output,
942                provider_state_summary: None,
943                localized,
944                localization_gaps,
945            } => match (state, self.pending_summaries.remove(&instance_key(path))) {
946                (StepState::Aborted, _) | (_, None) => payload,
947                (_, Some(summary)) => {
948                    enriched = RunLogPayload::StepExited {
949                        state: *state,
950                        output: output.clone(),
951                        provider_state_summary: Some(summary),
952                        localized: localized.clone(),
953                        localization_gaps: localization_gaps.clone(),
954                    };
955                    &enriched
956                }
957            },
958            _ => payload,
959        };
960        Ok(self
961            .store
962            .append_event(&self.run_id, now_ms(), path, payload)?)
963    }
964
965    /// Pre-stashes resume-generation profiles for crash-opened spans a
966    /// sync `record_pairs` cascade is about to close: a span whose ledger
967    /// verdict is fail/unknown must not exit summary-less just because
968    /// its verdict was recorded by a previous segment (07 §2.2 note 3).
969    async fn stash_open_span_summaries(&mut self) {
970        let keys: Vec<String> = self
971            .open_spans
972            .keys()
973            .filter(|key| {
974                matches!(
975                    self.recorded_verdicts.get(*key),
976                    Some((VerdictStatus::Fail | VerdictStatus::Unknown, _))
977                ) && !self.pending_summaries.contains_key(*key)
978            })
979            .cloned()
980            .collect();
981        if keys.is_empty() {
982            return;
983        }
984        let summary = self.capture_summary().await;
985        for key in keys {
986            self.pending_summaries.insert(key, summary.clone());
987        }
988    }
989
990    /// Captures the provider profile with this run's identity bindings.
991    async fn capture_summary(&self) -> ProviderStateSummary {
992        let env_str = |key: &str| {
993            self.env
994                .iter()
995                .find(|(name, _)| name == key)
996                .and_then(|(_, value)| value.as_str().map(str::to_owned))
997        };
998        capture_provider_state_summary(
999            self.session.as_ref(),
1000            &self.session_lineage,
1001            &env_str("deviceId").unwrap_or_default(),
1002            env_str("platform").as_deref(),
1003        )
1004        .await
1005    }
1006
1007    /// The wall clock the human-deadline machinery reads (injectable for
1008    /// tests; event `atMs` stamps stay on the system clock — they are
1009    /// informational, deadlines are semantics).
1010    fn now(&self) -> u64 {
1011        match &self.clock {
1012            Some(clock) => clock(),
1013            None => now_ms(),
1014        }
1015    }
1016
1017    /// Runs the root body from `start` and settles the run terminal.
1018    pub async fn run(
1019        mut self,
1020        mut root: FrameState<'a>,
1021        start: usize,
1022    ) -> Result<RunOutcome, RunnerError> {
1023        let flows = self.flows;
1024        let body: &'a [StepIR] = &flows.root.body;
1025        let prefix = root.base_path.clone();
1026        let ctl = self
1027            .exec_body(&mut root, prefix.clone(), body, start)
1028            .await?;
1029        match ctl {
1030            Ctl::Continue | Ctl::HaltFail => self.finish(false, &root).await,
1031            Ctl::Abort => self.finish(true, &root).await,
1032            Ctl::Suspend(reason) => {
1033                // Suspension-instant profile (07 §2.2): captured while
1034                // the session is still live, before teardown.
1035                let summary = self.capture_summary().await;
1036                self.append(
1037                    &prefix,
1038                    &RunLogPayload::RunSuspended {
1039                        provider_state_summary: Some(summary),
1040                        reason: Some(reason),
1041                    },
1042                )?;
1043                self.end_session(SessionOutcome::Shutdown).await;
1044                Ok(RunOutcome::Suspended)
1045            }
1046            Ctl::Blocked(reason) => {
1047                let summary = self.capture_summary().await;
1048                self.append(
1049                    &prefix,
1050                    &RunLogPayload::RunSuspended {
1051                        provider_state_summary: Some(summary),
1052                        reason: Some(reason.to_string()),
1053                    },
1054                )?;
1055                self.end_session(SessionOutcome::Shutdown).await;
1056                Ok(RunOutcome::Blocked { reason })
1057            }
1058            Ctl::AwaitHuman(pending) => {
1059                // The unified wait semantics: `humanRequested` is already
1060                // fsynced (its append committed); the segment suspends and
1061                // the process may exit — notification and collection are
1062                // the CLI layer's job (spine §6.8, 06 §5.1/§5.2). The
1063                // session is released while waiting; resume opens a new
1064                // one (session lineage).
1065                let summary = self.capture_summary().await;
1066                self.append(
1067                    &prefix,
1068                    &RunLogPayload::RunSuspended {
1069                        provider_state_summary: Some(summary),
1070                        reason: Some(format!(
1071                            "awaiting human response (requestId {})",
1072                            pending.request_id
1073                        )),
1074                    },
1075                )?;
1076                self.end_session(SessionOutcome::Shutdown).await;
1077                Ok(RunOutcome::AwaitingHuman { pending })
1078            }
1079        }
1080    }
1081
1082    /// Folds the root flow verdict, appends `runFinished`, ends the
1083    /// session.
1084    async fn finish(
1085        mut self,
1086        aborted: bool,
1087        root: &FrameState<'a>,
1088    ) -> Result<RunOutcome, RunnerError> {
1089        let prefix = root.base_path.clone();
1090        let verdict = if aborted {
1091            // An aborted run makes no flow-level semantic claim.
1092            None
1093        } else {
1094            fold_flow_verdict(&root.fold, root.flow.verdict_policy).map(|folded| Verdict {
1095                status: folded.status,
1096                degraded: folded.degraded,
1097                summary: folded.summary,
1098                evidence: Vec::new(),
1099                supersedes: None,
1100            })
1101        };
1102        let remote_archival_error = match &verdict {
1103            // Judgment authority is Pointlock's; the daemon only persists
1104            // (spine §6.3 write-back). Failure is annotation material,
1105            // never a run error (04 §5).
1106            Some(verdict) => self.try_verdict_writeback(verdict).await,
1107            None => None,
1108        };
1109        self.append(
1110            &prefix,
1111            &RunLogPayload::RunFinished {
1112                verdict: verdict.clone(),
1113                remote_archival_error,
1114            },
1115        )?;
1116        let session_outcome = if aborted {
1117            SessionOutcome::Cancelled
1118        } else if verdict
1119            .as_ref()
1120            .is_some_and(|verdict| verdict.status == VerdictStatus::Fail)
1121        {
1122            SessionOutcome::Failed
1123        } else {
1124            SessionOutcome::Completed
1125        };
1126        self.end_session(session_outcome).await;
1127        Ok(RunOutcome::Finished { verdict })
1128    }
1129
1130    /// Executes one body level sequentially. The stop token is honored
1131    /// before every step (step boundaries, any depth); a fail halts the
1132    /// level and records the remaining steps `blocked`.
1133    async fn exec_body(
1134        &mut self,
1135        frame: &mut FrameState<'a>,
1136        prefix: RunPath,
1137        body: &'a [StepIR],
1138        start: usize,
1139    ) -> Result<Ctl, RunnerError> {
1140        for (index, step) in body.iter().enumerate().skip(start) {
1141            if self.stop.is_cancelled() {
1142                return Ok(Ctl::Suspend("stop requested".to_owned()));
1143            }
1144            match self.exec_step(frame, &prefix, step).await? {
1145                Ctl::Continue => {}
1146                Ctl::HaltFail => {
1147                    // Halt-on-fail: remaining steps of this level are
1148                    // explicitly recorded blocked (never silently dropped
1149                    // from the ledger).
1150                    self.stash_open_span_summaries().await;
1151                    self.record_pairs(&prefix, &body[index + 1..], StepState::Blocked)?;
1152                    return Ok(Ctl::HaltFail);
1153                }
1154                other => return Ok(other),
1155            }
1156        }
1157        Ok(Ctl::Continue)
1158    }
1159
1160    /// Executes (or adopts) one step instance. Boxed: the recursion point
1161    /// of the tree walk (containers and calls re-enter `exec_body`).
1162    fn exec_step<'s>(
1163        &'s mut self,
1164        frame: &'s mut FrameState<'a>,
1165        prefix: &'s RunPath,
1166        step: &'a StepIR,
1167    ) -> LocalBoxFuture<'s, Result<Ctl, RunnerError>>
1168    where
1169        'a: 's,
1170    {
1171        Box::pin(async move {
1172            let mut path = prefix.clone();
1173            path.push(child_frame(step));
1174            let key = instance_key(&path);
1175            // Resume adoption (07 §4.6/I2): a concluded instance enters as
1176            // its record and never re-executes.
1177            if self
1178                .adoptable
1179                .get(&key)
1180                .is_some_and(|adopted| is_history(&adopted.record))
1181            {
1182                let adopted = self.adoptable.remove(&key).expect("checked present");
1183                self.adopt_step(frame, step, adopted);
1184                return Ok(Ctl::Continue);
1185            }
1186            // Breakpoint before entry: the instance is about to enter for
1187            // real (adoption above never enters). Suspend with the honest
1188            // reason; nothing of this instance reaches the ledger. A span
1189            // left open by a previous segment (`open_spans`: a container
1190            // suspended mid-way) is CONTINUED, not entered — its
1191            // `stepEntered` is already on the ledger and `enter_step` will
1192            // not write a second — so it is no breakpoint match either.
1193            if let Some(target) = &self.stop_at
1194                && !self.open_spans.contains_key(&key)
1195                && breakpoint_matches(target, &path)
1196            {
1197                return Ok(Ctl::Suspend(breakpoint_reason("--stop-at", &path)));
1198            }
1199            let ctl = match step {
1200                StepIR::Action(s) => self.exec_action(frame, path.clone(), s).await?,
1201                StepIR::Call(s) => self.exec_call(frame, path.clone(), s).await?,
1202                StepIR::If(s) => self.exec_if(frame, path.clone(), s).await?,
1203                StepIR::Foreach(s) => self.exec_foreach(frame, path.clone(), s).await?,
1204                StepIR::Let(s) => self.exec_let(frame, path.clone(), s).await?,
1205                StepIR::Assert(s) => self.exec_assert(frame, path.clone(), s).await?,
1206                StepIR::Human(s) => self.exec_human(frame, path.clone(), s).await?,
1207            };
1208            // Breakpoint after exit: the instance concluded and its
1209            // `stepExited` is on the ledger; suspend before any following
1210            // step enters. Only a concluded step (Continue) is "exited" in
1211            // this sense — a halt, abort, wait or suspension of its own
1212            // already carries the run elsewhere.
1213            if matches!(ctl, Ctl::Continue)
1214                && let Some(target) = &self.stop_after
1215                && breakpoint_matches(target, &path)
1216            {
1217                return Ok(Ctl::Suspend(breakpoint_reason("--stop-after", &path)));
1218            }
1219            Ok(ctl)
1220        })
1221    }
1222
1223    /// Seeds a frame with an adopted record's effects (outputs / verdicts /
1224    /// vars / observation material); containers recursively consume their
1225    /// children's records using the archived control snapshots — never a
1226    /// re-evaluation (I3).
1227    fn adopt_step(&mut self, frame: &mut FrameState<'a>, step: &'a StepIR, adopted: Adopted) {
1228        let id = step.step_id().as_str().to_owned();
1229        let record = adopted.record;
1230        match step {
1231            StepIR::Action(_) => {
1232                if let Some(verdict) = &record.verdict {
1233                    frame.seed_verdict(step.step_id(), verdict.status, verdict.degraded);
1234                }
1235                if let Some(output) = record.output.clone() {
1236                    frame.outputs.insert(id.clone(), output);
1237                }
1238                frame.observed.insert(
1239                    id,
1240                    StepObs {
1241                        observations: record.observations,
1242                        before_id: adopted.before_id,
1243                        after_id: adopted.after_id,
1244                    },
1245                );
1246            }
1247            StepIR::Assert(_) | StepIR::Call(_) | StepIR::Human(_) => {
1248                // A settled human step re-enters as its verdict/output
1249                // (the response was already arbitrated and folded into the
1250                // record) — never re-asked.
1251                if let Some(verdict) = &record.verdict {
1252                    frame.seed_verdict(step.step_id(), verdict.status, verdict.degraded);
1253                }
1254                if let Some(output) = record.output.clone() {
1255                    frame.outputs.insert(id, output);
1256                }
1257            }
1258            StepIR::Let(_) => {
1259                // The archived ready snapshot *is* the bindings product.
1260                if let Value::Object(bindings) = record.resolved_inputs {
1261                    for (name, value) in bindings {
1262                        frame.vars.insert(name, value);
1263                    }
1264                }
1265            }
1266            StepIR::If(s) => {
1267                // Consume both branches: the selected branch's records seed
1268                // effects, the unselected branch's skipped pairs seed
1269                // nothing — both leave the adoption set.
1270                self.adopt_children(frame, &record.run_path, &s.then);
1271                if let Some(otherwise) = &s.r#else {
1272                    self.adopt_children(frame, &record.run_path, otherwise);
1273                }
1274            }
1275            StepIR::Foreach(s) => {
1276                let rounds = record
1277                    .resolved_inputs
1278                    .get("items")
1279                    .and_then(Value::as_array)
1280                    .map(Vec::len)
1281                    .unwrap_or(0);
1282                for index in 0..rounds {
1283                    let mut prefix = record.run_path.clone();
1284                    prefix.push(PathFrame::Iteration {
1285                        index: index as u64,
1286                        key: None,
1287                    });
1288                    self.adopt_children(frame, &prefix, &s.body);
1289                }
1290            }
1291        }
1292    }
1293
1294    fn adopt_children(
1295        &mut self,
1296        frame: &mut FrameState<'a>,
1297        prefix: &RunPath,
1298        steps: &'a [StepIR],
1299    ) {
1300        for step in steps {
1301            let mut path = prefix.clone();
1302            path.push(child_frame(step));
1303            let key = instance_key(&path);
1304            if let Some(adopted) = self.adoptable.remove(&key) {
1305                self.adopt_step(frame, step, adopted);
1306            }
1307        }
1308    }
1309
1310    /// Records `entered(resolvedInputs: null)`/`exited(state)` pairs for a
1311    /// subtree that will not execute (skipped branches, blocked tails) —
1312    /// ledger completeness per the blocked precedent. Children are handled
1313    /// before their container so that crash-opened spans close innermost
1314    /// first (the fold's exit pairing is positional). Instances already on
1315    /// the ledger from a previous segment are kept, not re-emitted.
1316    fn record_pairs(
1317        &mut self,
1318        prefix: &RunPath,
1319        steps: &'a [StepIR],
1320        state: StepState,
1321    ) -> Result<(), RunnerError> {
1322        for step in steps {
1323            let mut path = prefix.clone();
1324            path.push(child_frame(step));
1325            match step {
1326                StepIR::If(s) => {
1327                    self.record_pairs(&path, &s.then, state)?;
1328                    if let Some(otherwise) = &s.r#else {
1329                        self.record_pairs(&path, otherwise, state)?;
1330                    }
1331                }
1332                StepIR::Foreach(s) => self.record_pairs(&path, &s.body, state)?,
1333                _ => {}
1334            }
1335            let key = instance_key(&path);
1336            if self.adoptable.remove(&key).is_some() {
1337                continue;
1338            }
1339            if self.open_spans.remove(&key).is_some() {
1340                // A previous segment opened this span; close it with the
1341                // terminal state instead of double-entering.
1342                self.append(
1343                    &path,
1344                    &RunLogPayload::StepExited {
1345                        provider_state_summary: None,
1346                        state,
1347                        output: None,
1348                        localized: Vec::new(),
1349                        localization_gaps: Vec::new(),
1350                    },
1351                )?;
1352                continue;
1353            }
1354            self.append(
1355                &path,
1356                &RunLogPayload::StepEntered {
1357                    step_id: step.step_id().clone(),
1358                    effect_hash: step.base().effect_hash.clone(),
1359                    judge_hash: step.base().judge_hash.clone(),
1360                    resolved_inputs: Value::Null,
1361                },
1362            )?;
1363            self.append(
1364                &path,
1365                &RunLogPayload::StepExited {
1366                    provider_state_summary: None,
1367                    state,
1368                    output: None,
1369                    localized: Vec::new(),
1370                    localization_gaps: Vec::new(),
1371                },
1372            )?;
1373        }
1374        Ok(())
1375    }
1376
1377    /// The archived ready snapshot of a crash/suspension-opened span, when
1378    /// this instance has one (peek — `enter_step` consumes it).
1379    fn open_span_inputs(&self, key: &str) -> Option<Value> {
1380        self.open_spans.get(key).cloned()
1381    }
1382
1383    /// Appends `stepEntered` unless the instance's span is already open
1384    /// (resume: the log has an unmatched `stepEntered`). The payload
1385    /// carries the step's dual hashes and the frozen ready-phase input
1386    /// snapshot (spine §6.1 M1 note).
1387    ///
1388    /// A fresh `stepEntered` for an instance the ledger already knows is
1389    /// a new execution life (re-execution after positional invalidation
1390    /// or a forced re-execution, 07 §5.2): liveness wins, and the
1391    /// previous life's handler trigger budget is among the facts the new
1392    /// life discards — `maxTriggers` bounds the loop within one life.
1393    fn enter_step(
1394        &mut self,
1395        path: &RunPath,
1396        base: &StepBase,
1397        resolved_inputs: Value,
1398    ) -> Result<(), RunnerError> {
1399        let key = instance_key(path);
1400        if self.open_spans.remove(&key).is_some() {
1401            return Ok(());
1402        }
1403        crate::align::retain_other_instances(&mut self.hook_triggers, &key);
1404        self.append(
1405            path,
1406            &RunLogPayload::StepEntered {
1407                step_id: base.step_id.clone(),
1408                effect_hash: base.effect_hash.clone(),
1409                judge_hash: base.judge_hash.clone(),
1410                resolved_inputs,
1411            },
1412        )?;
1413        Ok(())
1414    }
1415
1416    /// Evaluates one bound attempt's argument expressions against the
1417    /// frame scope (the ready-phase resolution).
1418    fn resolve_args(
1419        &self,
1420        frame: &FrameState<'a>,
1421        attempt: &BoundAttempt,
1422    ) -> Result<Value, String> {
1423        let scope = frame.scope(&self.env, None);
1424        let mut evaluated = serde_json::Map::new();
1425        for (name, expr) in attempt.args.iter() {
1426            match pointlock_expr::eval(expr, &scope) {
1427                Ok(value) => {
1428                    evaluated.insert(name.as_str().to_owned(), value);
1429                }
1430                Err(error) => return Err(format!("argument evaluation failed: {error}")),
1431            }
1432        }
1433        Ok(Value::Object(evaluated))
1434    }
1435
1436    // ─── probing (spine §6.2; 07 §4.2) ──────────────────────────────────────
1437
1438    /// Runs a step's declared `preflight`, or records that there was none
1439    /// to run (07 §4.2 rule 1 / I3).
1440    ///
1441    /// 「该步无声明则跳过并在报告标 `unprobed`(诚实优先于安慰)」. The
1442    /// carrier is a `preflightProbed` with an EMPTY outcome list, which is
1443    /// unambiguous rather than clever: `preflight` is `minItems: 1` in the
1444    /// schema, so a declared probe list can never evaluate to zero
1445    /// outcomes. No new event type, no new payload field, and old ledgers
1446    /// — which never emitted it — refold byte-identically.
1447    ///
1448    /// It is written at exactly the two places the spec names, and nowhere
1449    /// else. A step in the middle of a continuously-executing run is not
1450    /// re-touching a world anyone stopped watching, and marking it would
1451    /// turn an honest signal into noise:
1452    /// - the segment's re-entry step, when the segment is a resume
1453    ///   (§4.2 rule 1);
1454    /// - every step released through the §5.4 gate (step 3: 「本条
1455    ///   preflight 守护对 `positionalReplay`/`orderInvalidated`/
1456    ///   `frontierUnknown` 的步同样强制适用」) — those re-execute onto a
1457    ///   world that carries the earlier effect, which is the whole reason
1458    ///   they had to be authorized by name.
1459    async fn probe_or_note(
1460        &mut self,
1461        frame: &mut FrameState<'a>,
1462        path: &RunPath,
1463        base: &'a StepBase,
1464    ) -> Result<Option<Ctl>, RunnerError> {
1465        let reentry = self.resumed && !self.reentry_seen;
1466        self.reentry_seen = true;
1467        if let Some(probes) = &base.preflight {
1468            return self.probe_preflight(frame, path, base, probes).await;
1469        }
1470        if reentry || self.authorized.contains(base.step_id.as_str()) {
1471            let mut probe_path = path.clone();
1472            probe_path.push(PathFrame::Phase {
1473                phase: Phase::Preflight,
1474            });
1475            self.append(
1476                &probe_path,
1477                &RunLogPayload::PreflightProbed {
1478                    outcomes: Vec::new(),
1479                },
1480            )?;
1481        }
1482        Ok(None)
1483    }
1484
1485    /// Evaluates a step's declared `preflight` probes over fresh observe
1486    /// material (spine §6.7-C operationalized). A probe that does not hold
1487    /// — or cannot be evaluated (exhausted chain) — is drift: the step's
1488    /// `onResumeDrift` ladder is consulted (repair → re-probe, escalate →
1489    /// `repairWorld`); with none left the run blocks (`drifted` →
1490    /// `runSuspended`).
1491    async fn probe_preflight(
1492        &mut self,
1493        frame: &mut FrameState<'a>,
1494        path: &RunPath,
1495        base: &'a StepBase,
1496        probes: &'a [AssertionIR],
1497    ) -> Result<Option<Ctl>, RunnerError> {
1498        let step_id = &base.step_id;
1499        let mut probe_path = path.clone();
1500        probe_path.push(PathFrame::Phase {
1501            phase: Phase::Preflight,
1502        });
1503        let needs = VerifyNeeds::of(probes);
1504        let mut active_retry: Option<(RetryPolicy, u32)> = None;
1505        loop {
1506            let material = self.fresh_material(&needs, &probe_path).await?;
1507            let scope = frame.scope(&self.env, None);
1508            let mut outcomes = Vec::with_capacity(probes.len());
1509            for probe in probes {
1510                let evaluated = match &probe.predicate {
1511                    PredicateIR::Expr { expr } => EvaluatedAssertion {
1512                        record: eval_expr_assertion(probe, expr, &scope),
1513                        degraded_verify: false,
1514                    },
1515                    _ => eval_observed_assertion(probe, &material, self.vision.as_deref()).await,
1516                };
1517                outcomes.push(evaluated.record);
1518            }
1519            self.append(
1520                &probe_path,
1521                &RunLogPayload::PreflightProbed {
1522                    outcomes: outcomes.clone(),
1523                },
1524            )?;
1525            let Some(missed) = outcomes
1526                .iter()
1527                .find(|outcome| outcome.result != VerdictStatus::Pass)
1528            else {
1529                return Ok(None);
1530            };
1531            // Unable-to-confirm is drift too (07 §4.2 rule 2: not being
1532            // able to see the world is not the world being fine —
1533            // principle 4).
1534            let what = match missed.result {
1535                VerdictStatus::Fail => "did not hold",
1536                _ => "could not be evaluated (treated as drift)",
1537            };
1538            let detail = format!("probe '{}' {what}: {}", missed.assert_id, missed.reason);
1539
1540            // In-force drift-handler retry budget: re-probe (readonly).
1541            if let Some((policy, used)) = active_retry.take()
1542                && used < policy.max_attempts
1543            {
1544                self.backoff_policy(&policy, used).await;
1545                active_retry = Some((policy, used + 1));
1546                continue;
1547            }
1548
1549            // A failed probe consults `onResumeDrift` (spine §6.2/§6.7-C:
1550            // probing → drifted → the drift handler; exhausted budgets
1551            // block awaiting a human decision).
1552            match self
1553                .consult_hook(
1554                    frame,
1555                    path,
1556                    base.handlers.as_deref(),
1557                    HandlerHook::OnResumeDrift,
1558                    None,
1559                )
1560                .await?
1561            {
1562                Consulted::None | Consulted::RepairFailed => {
1563                    return Ok(Some(Ctl::Blocked(BlockedReason::Drifted {
1564                        step_id: step_id.as_str().to_owned(),
1565                        detail,
1566                    })));
1567                }
1568                Consulted::Continue => {
1569                    // The author accepts the drifted world: proceed.
1570                    return Ok(None);
1571                }
1572                Consulted::Abort => return Ok(Some(Ctl::Abort)),
1573                Consulted::Escalated { status, .. } => match status {
1574                    // A human judged the world acceptable: proceed.
1575                    VerdictStatus::Pass => return Ok(None),
1576                    _ => {
1577                        return Ok(Some(Ctl::Blocked(BlockedReason::Drifted {
1578                            step_id: step_id.as_str().to_owned(),
1579                            detail: format!("{detail}; escalate ruling: not acceptable"),
1580                        })));
1581                    }
1582                },
1583                Consulted::Retry(policy) => {
1584                    self.backoff_policy(&policy, 0).await;
1585                    active_retry = Some((policy, 1));
1586                }
1587                // A repaired world (declared or via the repair flow):
1588                // re-probe — the probe, not the declaration, readmits.
1589                Consulted::Repaired | Consulted::RepairDone => {}
1590                Consulted::Pending(pending) => return Ok(Some(Ctl::AwaitHuman(pending))),
1591                Consulted::Propagate(ctl) => return Ok(Some(ctl)),
1592            }
1593        }
1594    }
1595
1596    /// Captures a fresh observation (`session.observe`) sized to the
1597    /// declared verify needs, localizes it (`observationRecorded`), and
1598    /// returns the verify-chain material. An observe failure is a typed
1599    /// material gap, never a run abort — the dependent assertions degrade
1600    /// toward unknown.
1601    async fn fresh_material(
1602        &mut self,
1603        needs: &VerifyNeeds,
1604        anchor: &RunPath,
1605    ) -> Result<ObserveMaterial, RunnerError> {
1606        let mut wants = Vec::new();
1607        if needs.ui_tree {
1608            wants.push(ObserveWant::UiSnapshot);
1609        }
1610        if needs.vision {
1611            wants.push(ObserveWant::Screenshot);
1612        }
1613        if wants.is_empty() {
1614            // Expr-only consumers need no observation channel.
1615            return Ok(ObserveMaterial::default());
1616        }
1617        let observation = match self.session.observe(ObserveRequest { wants }, None).await {
1618            Ok(observation) => observation,
1619            Err(error) => {
1620                return Ok(ObserveMaterial::absent(&format!(
1621                    "fresh observation failed: {error}"
1622                )));
1623            }
1624        };
1625        let mut cited = Vec::new();
1626        let mut material = ObserveMaterial::default();
1627        let record = self
1628            .localize_observation(&observation, &mut cited, Some((needs, &mut material)))
1629            .await?;
1630        self.append(
1631            anchor,
1632            &RunLogPayload::ObservationRecorded {
1633                observation: record,
1634            },
1635        )?;
1636        Ok(material)
1637    }
1638
1639    // ─── action steps ───────────────────────────────────────────────────────
1640
1641    async fn exec_action(
1642        &mut self,
1643        frame: &mut FrameState<'a>,
1644        step_path: RunPath,
1645        step: &'a ActionStepIR,
1646    ) -> Result<Ctl, RunnerError> {
1647        let step_id = step.base.step_id.clone();
1648        let key = instance_key(&step_path);
1649        let work = match &self.frontier {
1650            Some((frontier_key, _)) if *frontier_key == key => {
1651                self.frontier.take().map(|(_, work)| work)
1652            }
1653            _ => None,
1654        };
1655
1656        // Ready precedes entered (spine §6.1 M1 note): the first bound
1657        // attempt's argument expressions are resolved once and frozen;
1658        // `stepEntered` carries the snapshot and lands before any
1659        // preflight probe or `actionIntent`. Frontier work reuses the
1660        // archived snapshot verbatim — never re-evaluated (spine §6.6).
1661        let resolved = match &work {
1662            Some(FrontierWork::Adopt { args, .. })
1663            | Some(FrontierWork::Replay { args, .. })
1664            | Some(FrontierWork::ConfirmEffect { args, .. })
1665            | Some(FrontierWork::AbortRuled { args }) => Ok(args.clone()),
1666            None => {
1667                let attempt = step
1668                    .binding
1669                    .attempts
1670                    .first()
1671                    .expect("sealed action steps carry at least one bound attempt");
1672                self.resolve_args(frame, attempt)
1673            }
1674        };
1675        let resolved_inputs = match resolved {
1676            Ok(args) => args,
1677            Err(message) => {
1678                // Inputs never resolved: the span still opens and closes
1679                // (one entered/exited pair per step), with
1680                // `resolvedInputs: null` — a failing argument evaluation
1681                // is a compiler/expression bug signal
1682                // (bind_arguments_invalid discipline): step fails, no
1683                // retry.
1684                self.enter_step(&step_path, &step.base, Value::Null)?;
1685                return self
1686                    .settle_error(
1687                        frame,
1688                        &step_path,
1689                        &step_id,
1690                        VerdictStatus::Fail,
1691                        format!("act phase failed [bind_arguments_invalid]: {message}"),
1692                    )
1693                    .await;
1694            }
1695        };
1696        let had_open_span = self.open_span_inputs(&key).is_some();
1697        self.enter_step(&step_path, &step.base, resolved_inputs.clone())?;
1698
1699        // Handler-wave resume re-entry (I2): an open span whose act
1700        // already settled and whose verdict is on the ledger means a
1701        // previous segment suspended mid-handler (a pending escalate).
1702        // Never re-dispatch — enter the disposition loop directly from
1703        // the recorded ruling.
1704        let resume_ruling = if work.is_none() && had_open_span {
1705            match (self.settled.get(&key), self.recorded_verdicts.get(&key)) {
1706                (Some(_), Some(ruling)) => Some(*ruling),
1707                _ => None,
1708            }
1709        } else {
1710            None
1711        };
1712
1713        // Probing (§6.2): declared preflight evaluates between entered and
1714        // acting. An adopted frontier terminal (or a handler-wave
1715        // re-entry) means the act already left in a previous life —
1716        // probing "is the world ready for the act" after the act is
1717        // meaningless, so it is skipped there.
1718        if !matches!(work, Some(FrontierWork::Adopt { .. }))
1719            && resume_ruling.is_none()
1720            && let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await?
1721        {
1722            return Ok(ctl);
1723        }
1724
1725        // R13 supervision gate (spine §6.9): sits wholly before the
1726        // `actionIntent` WAL — a refused act never had an intent on the
1727        // ledger. Adopt/Replay frontier work is never re-gated: the act
1728        // already happened, or its intent was WAL-authorized by a
1729        // previous segment.
1730        if work.is_none()
1731            && resume_ruling.is_none()
1732            && let Some(ctl) =
1733                self.gate_supervision(&step_path, step, &resolved_inputs, had_open_span)?
1734        {
1735            return Ok(ctl);
1736        }
1737
1738        let mut acted = if resume_ruling.is_some() {
1739            None
1740        } else {
1741            Some(match work {
1742                Some(FrontierWork::Adopt {
1743                    call_id,
1744                    intent_path,
1745                    outcome,
1746                    args,
1747                    chain_index,
1748                }) => {
1749                    // Record the terminal the crash swallowed; this settles
1750                    // the pending intent in the checkpoint fold.
1751                    let attempt_n = attempt_of(&intent_path).unwrap_or(1);
1752                    let outcome = quarantine_unpersistable(*outcome);
1753                    let settled_seq = self.append(
1754                        &intent_path,
1755                        &RunLogPayload::ActionSettled {
1756                            call_id,
1757                            outcome: outcome.clone(),
1758                        },
1759                    )?;
1760                    // From here the adopted terminal takes the exact same
1761                    // settled-outcome path as a live one.
1762                    let adopted = AdoptedSettle {
1763                        outcome,
1764                        settled_seq,
1765                        attempt_n,
1766                    };
1767                    self.act_chain(
1768                        frame,
1769                        step,
1770                        &step_path,
1771                        Some(args),
1772                        Some(adopted),
1773                        chain_start(chain_index, step)?,
1774                    )
1775                    .await?
1776                }
1777                Some(FrontierWork::Replay { args, chain_index }) => {
1778                    self.act_chain(
1779                        frame,
1780                        step,
1781                        &step_path,
1782                        Some(args),
1783                        None,
1784                        chain_start(chain_index, step)?,
1785                    )
1786                    .await?
1787                }
1788                Some(FrontierWork::ConfirmEffect { message, .. }) => {
1789                    // No dispatch: the adjudication already ruled on the
1790                    // act; only the world's testimony is still owed.
1791                    ActPhase::Unconfirmed { message }
1792                }
1793                Some(FrontierWork::AbortRuled { .. }) => {
1794                    // The ruled abort mirrors a `cancelled` terminal's
1795                    // unwind: the open span closes `aborted` and the run
1796                    // makes no further semantic claim.
1797                    ActPhase::Aborted
1798                }
1799                // The fresh path hands the ready snapshot to the first
1800                // chain attempt — resolved exactly once, above.
1801                None => {
1802                    self.act_chain(
1803                        frame,
1804                        step,
1805                        &step_path,
1806                        Some(resolved_inputs.clone()),
1807                        None,
1808                        0,
1809                    )
1810                    .await?
1811                }
1812            })
1813        };
1814
1815        // ── the settlement/disposition loop (M2 W3) ─────────────────────
1816        //
1817        // One round = one settled act (or the resumed recorded ruling)
1818        // judged and, on fail/unknown, consulted against the handlers.
1819        // Retry-class dispositions re-enter the act with the frozen
1820        // snapshot (new callId, new WAL intent); every re-fold records a
1821        // new verdict superseding the previous one (spine §2 concept 12).
1822        let mut last_verdict_seq: Option<u64> = None;
1823        let mut seeded = false;
1824        let mut active_retry: Option<(RetryPolicy, u32)> = None;
1825        let mut ruling = resume_ruling;
1826        loop {
1827            // What this round established: (status, degraded, summary,
1828            // cited evidence, projected output, error-path class).
1829            let (status, degraded, summary, cited, projected, error_class): (
1830                Option<VerdictStatus>,
1831                bool,
1832                String,
1833                Vec<AssetRef>,
1834                Option<Value>,
1835                Option<ErrorClass>,
1836            );
1837            let mut round_manifest = EvidenceManifest::default();
1838            let mut ruled_from_ledger = false;
1839            match (acted.take(), ruling.take()) {
1840                (None, Some((recorded_status, recorded_degraded))) => {
1841                    // Resumed at the recorded verdict: derive the output
1842                    // projection from the archived succeeded terminal so
1843                    // downstream refs keep working (pure re-projection).
1844                    ruled_from_ledger = true;
1845                    let recovered = match self.settled.get(&key).map(|fact| &fact.outcome) {
1846                        Some(ActionOutcome::Succeeded { result }) => {
1847                            let raw_scope =
1848                                frame.scope(&self.env, Some((step_id.as_str(), &result.output)));
1849                            project_output(step, &result.output, &raw_scope).ok()
1850                        }
1851                        _ => None,
1852                    };
1853                    status = Some(recorded_status);
1854                    degraded = recorded_degraded;
1855                    summary = "resumed at the recorded verdict (handler re-entry)".to_owned();
1856                    cited = Vec::new();
1857                    projected = recovered;
1858                    error_class = None;
1859                    // Cross-segment gate (07 §2.2 note 3): the previous
1860                    // segment's verdict is in force but its stash died
1861                    // with the process. Capture the RESUME-generation
1862                    // profile so the eventual exit still carries one —
1863                    // self-describing via its own sessionLineage/cursor;
1864                    // the failure-instant profile rides the prior
1865                    // segment's runSuspended.
1866                    if matches!(
1867                        recorded_status,
1868                        VerdictStatus::Fail | VerdictStatus::Unknown
1869                    ) {
1870                        let captured = self.capture_summary().await;
1871                        self.pending_summaries.insert(key.clone(), captured);
1872                    }
1873                }
1874                (Some(phase), _) => match phase {
1875                    ActPhase::Succeeded {
1876                        result,
1877                        degraded: degraded_execution,
1878                        settled_seq,
1879                        attempt_n,
1880                    } => {
1881                        let (round_cited, material, observed, manifest) = self
1882                            .observing(step, &step_path, attempt_n, &result, settled_seq)
1883                            .await?;
1884                        round_manifest = manifest;
1885                        frame.observed.insert(step_id.as_str().to_owned(), observed);
1886                        // Output projection (self-refs see the raw output).
1887                        let raw_scope =
1888                            frame.scope(&self.env, Some((step_id.as_str(), &result.output)));
1889                        let round_projected = match project_output(step, &result.output, &raw_scope)
1890                        {
1891                            Ok(value) => value,
1892                            Err(error) => {
1893                                // A failing output projection is a
1894                                // compiler/expression bug signal: step
1895                                // fails, no retry, no consultation
1896                                // (bind_arguments_invalid discipline).
1897                                return self
1898                                    .settle_error(
1899                                        frame,
1900                                        &step_path,
1901                                        &step_id,
1902                                        VerdictStatus::Fail,
1903                                        format!("output projection failed: {error}"),
1904                                    )
1905                                    .await;
1906                            }
1907                        };
1908                        // Asserting: pure computation over materialized
1909                        // values (the vision tail is the one declared
1910                        // exception to purity).
1911                        let assert_scope =
1912                            frame.scope(&self.env, Some((step_id.as_str(), &round_projected)));
1913                        let mut outcomes = Vec::with_capacity(step.assertions.len());
1914                        let mut degraded_verify = false;
1915                        for assertion in &step.assertions {
1916                            let evaluated = match &assertion.predicate {
1917                                PredicateIR::Expr { expr } => EvaluatedAssertion {
1918                                    record: eval_expr_assertion(assertion, expr, &assert_scope),
1919                                    degraded_verify: false,
1920                                },
1921                                _ => {
1922                                    eval_observed_assertion(
1923                                        assertion,
1924                                        &material,
1925                                        self.vision.as_deref(),
1926                                    )
1927                                    .await
1928                                }
1929                            };
1930                            degraded_verify |= evaluated.degraded_verify;
1931                            outcomes.push(evaluated.record);
1932                        }
1933                        for outcome in &outcomes {
1934                            let mut path = step_path.clone();
1935                            path.push(PathFrame::Phase {
1936                                phase: Phase::Assert,
1937                            });
1938                            path.push(PathFrame::Assertion {
1939                                assert_id: outcome.assert_id.clone(),
1940                            });
1941                            self.append(
1942                                &path,
1943                                &RunLogPayload::AssertionEvaluated {
1944                                    outcome: outcome.clone(),
1945                                },
1946                            )?;
1947                        }
1948                        if outcomes.is_empty() {
1949                            // No assertions ⇒ no verdict (spine R4):
1950                            // execution status only (`unverified`).
1951                            status = None;
1952                            degraded = false;
1953                            summary = String::new();
1954                        } else {
1955                            let folded = fold_step_verdict(
1956                                &outcomes,
1957                                degraded_execution,
1958                                degraded_verify,
1959                                frame.flow.verdict_policy,
1960                            );
1961                            status = Some(folded.status);
1962                            degraded = folded.degraded;
1963                            summary = folded.summary;
1964                        }
1965                        cited = round_cited;
1966                        projected = Some(round_projected);
1967                        error_class = None;
1968                    }
1969                    ActPhase::Unconfirmed { message } => {
1970                        // The observe half of 「先 reconcile/observe 确认」:
1971                        // a fresh observation, the step's own assertions
1972                        // over it — the same pure evaluation an assert step
1973                        // runs. Expr assertions reference an output the
1974                        // timeout never produced and resolve unknown
1975                        // (`onMissingInput`, principle 4); element and
1976                        // visual predicates read the world and can be
1977                        // decisive in both directions. The fold is the
1978                        // confirmation verdict: pass — the effect is
1979                        // visibly there; fail — visibly not; unknown —
1980                        // unconfirmable, exactly the path the bare timeout
1981                        // always took.
1982                        let needs = VerifyNeeds::of(&step.assertions);
1983                        let material = self.fresh_material(&needs, &step_path).await?;
1984                        let assert_scope = frame.scope(&self.env, None);
1985                        let mut outcomes = Vec::with_capacity(step.assertions.len());
1986                        let mut degraded_verify = false;
1987                        for assertion in &step.assertions {
1988                            let evaluated = match &assertion.predicate {
1989                                PredicateIR::Expr { expr } => EvaluatedAssertion {
1990                                    record: eval_expr_assertion(assertion, expr, &assert_scope),
1991                                    degraded_verify: false,
1992                                },
1993                                _ => {
1994                                    eval_observed_assertion(
1995                                        assertion,
1996                                        &material,
1997                                        self.vision.as_deref(),
1998                                    )
1999                                    .await
2000                                }
2001                            };
2002                            degraded_verify |= evaluated.degraded_verify;
2003                            outcomes.push(evaluated.record);
2004                        }
2005                        for outcome in &outcomes {
2006                            let mut path = step_path.clone();
2007                            path.push(PathFrame::Phase {
2008                                phase: Phase::Assert,
2009                            });
2010                            path.push(PathFrame::Assertion {
2011                                assert_id: outcome.assert_id.clone(),
2012                            });
2013                            self.append(
2014                                &path,
2015                                &RunLogPayload::AssertionEvaluated {
2016                                    outcome: outcome.clone(),
2017                                },
2018                            )?;
2019                        }
2020                        let folded = fold_step_verdict(
2021                            &outcomes,
2022                            false,
2023                            degraded_verify,
2024                            frame.flow.verdict_policy,
2025                        );
2026                        status = Some(folded.status);
2027                        degraded = folded.degraded;
2028                        summary = format!(
2029                            "{message}; assertions over a fresh observation: {}",
2030                            folded.summary
2031                        );
2032                        cited = Vec::new();
2033                        projected = None;
2034                        error_class = None;
2035                    }
2036                    ActPhase::StepFail { class, message } => {
2037                        let wire = serde_json::to_value(class).expect("ErrorClass serializes");
2038                        let wire = wire.as_str().expect("ErrorClass is a string literal");
2039                        status = Some(VerdictStatus::Fail);
2040                        degraded = false;
2041                        summary = format!("act phase failed [{wire}]: {message}");
2042                        cited = Vec::new();
2043                        projected = None;
2044                        error_class = Some(class);
2045                    }
2046                    ActPhase::StepUnknown {
2047                        message,
2048                        error_class: class,
2049                    } => {
2050                        status = Some(VerdictStatus::Unknown);
2051                        degraded = false;
2052                        summary = message;
2053                        cited = Vec::new();
2054                        projected = None;
2055                        // Usually none — an unknown has no error to route.
2056                        // `session_degraded` is the exception the spine §5
2057                        // table names, and it keeps its class so the hook
2058                        // selector below reaches `onError`.
2059                        error_class = class;
2060                    }
2061                    ActPhase::Aborted => {
2062                        self.append(
2063                            &step_path,
2064                            &RunLogPayload::StepExited {
2065                                provider_state_summary: None,
2066                                state: StepState::Aborted,
2067                                output: None,
2068                                localized: Vec::new(),
2069                                localization_gaps: Vec::new(),
2070                            },
2071                        )?;
2072                        return Ok(Ctl::Abort);
2073                    }
2074                    ActPhase::Suspend(reason) => return Ok(Ctl::Suspend(reason)),
2075                },
2076                (None, None) => unreachable!("every round has an act result or a ruling"),
2077            }
2078
2079            // Record this round's verdict (unless it came off the ledger)
2080            // and seed/reseed the frame fold.
2081            let round_status = status;
2082            if let Some(current) = round_status {
2083                if !ruled_from_ledger {
2084                    let folded = FoldedVerdict {
2085                        status: current,
2086                        degraded,
2087                        summary: summary.clone(),
2088                    };
2089                    let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
2090                    let seq = self
2091                        .record_step_verdict(
2092                            &step_path,
2093                            &folded,
2094                            cited,
2095                            supersedes,
2096                            std::mem::take(&mut round_manifest),
2097                        )
2098                        .await?;
2099                    last_verdict_seq = Some(seq);
2100                }
2101                if seeded {
2102                    frame.reseed_last(&step_id, current, degraded);
2103                } else {
2104                    frame.seed_verdict(&step_id, current, degraded);
2105                    seeded = true;
2106                }
2107            }
2108
2109            // Pass / unverified: exit judged and release.
2110            if round_status.is_none() || round_status == Some(VerdictStatus::Pass) {
2111                // An UNVERIFIED exit (R4: no assertions ⇒ no verdict ⇒
2112                // no verdictRecorded carrier) must not drop its
2113                // settlement-evidence manifest — it rides the exit
2114                // instead (item ③ review fix). A pass-verdict exit's
2115                // manifest already rode its verdictRecorded.
2116                let exit_manifest = if round_status.is_none() {
2117                    std::mem::take(&mut round_manifest)
2118                } else {
2119                    EvidenceManifest::default()
2120                };
2121                self.append(
2122                    &step_path,
2123                    &RunLogPayload::StepExited {
2124                        provider_state_summary: None,
2125                        state: StepState::Judged,
2126                        output: projected.clone(),
2127                        localized: exit_manifest.localized,
2128                        localization_gaps: exit_manifest.gaps,
2129                    },
2130                )?;
2131                if let Some(projected) = projected {
2132                    frame.outputs.insert(step_id.as_str().to_owned(), projected);
2133                }
2134                return Ok(Ctl::Continue);
2135            }
2136            let current = round_status.expect("checked above");
2137
2138            // In-force handler retry budget (one consultation grants
2139            // `max_attempts` re-entries with its backoff schedule).
2140            if let Some((policy, used)) = active_retry.take()
2141                && used < policy.max_attempts
2142            {
2143                self.backoff_policy(&policy, used).await;
2144                active_retry = Some((policy, used + 1));
2145                acted = Some(
2146                    // In-force retry policy: re-enter from the chain head
2147                    // (item ② ruling — handler retry restarts the chain).
2148                    self.act_chain(
2149                        frame,
2150                        step,
2151                        &step_path,
2152                        Some(resolved_inputs.clone()),
2153                        None,
2154                        0,
2155                    )
2156                    .await?,
2157                );
2158                continue;
2159            }
2160
2161            // Consult the hook: assertion negatives walk onFail/onUnknown,
2162            // error-path negatives walk onError, error-path unknowns walk
2163            // onUnknown (AssertionFailure is a verdict, not an error —
2164            // spine §5).
2165            let hook = if error_class.is_some() {
2166                HandlerHook::OnError
2167            } else if current == VerdictStatus::Fail {
2168                HandlerHook::OnFail
2169            } else {
2170                HandlerHook::OnUnknown
2171            };
2172            match self
2173                .consult_hook(
2174                    frame,
2175                    &step_path,
2176                    step.base.handlers.as_deref(),
2177                    hook,
2178                    error_class,
2179                )
2180                .await?
2181            {
2182                Consulted::None | Consulted::RepairFailed => {
2183                    self.append(
2184                        &step_path,
2185                        &RunLogPayload::StepExited {
2186                            provider_state_summary: None,
2187                            state: StepState::Judged,
2188                            output: projected.clone(),
2189                            localized: Vec::new(),
2190                            localization_gaps: Vec::new(),
2191                        },
2192                    )?;
2193                    if let Some(projected) = projected {
2194                        frame.outputs.insert(step_id.as_str().to_owned(), projected);
2195                    }
2196                    return Ok(if current == VerdictStatus::Fail {
2197                        Ctl::HaltFail
2198                    } else {
2199                        Ctl::Continue
2200                    });
2201                }
2202                Consulted::Continue => {
2203                    // Record-and-release: the verdict stands, downstream
2204                    // is not halted (spine §3 disposition table).
2205                    self.append(
2206                        &step_path,
2207                        &RunLogPayload::StepExited {
2208                            provider_state_summary: None,
2209                            state: StepState::Judged,
2210                            output: projected.clone(),
2211                            localized: Vec::new(),
2212                            localization_gaps: Vec::new(),
2213                        },
2214                    )?;
2215                    if let Some(projected) = projected {
2216                        frame.outputs.insert(step_id.as_str().to_owned(), projected);
2217                    }
2218                    return Ok(Ctl::Continue);
2219                }
2220                Consulted::Abort => {
2221                    self.append(
2222                        &step_path,
2223                        &RunLogPayload::StepExited {
2224                            provider_state_summary: None,
2225                            state: StepState::Aborted,
2226                            output: None,
2227                            localized: Vec::new(),
2228                            localization_gaps: Vec::new(),
2229                        },
2230                    )?;
2231                    return Ok(Ctl::Abort);
2232                }
2233                Consulted::Escalated {
2234                    status: ruled,
2235                    summary: ruled_summary,
2236                    evidence: ruling_evidence,
2237                } => {
2238                    let folded = FoldedVerdict {
2239                        status: ruled,
2240                        degraded: false,
2241                        summary: ruled_summary,
2242                    };
2243                    let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
2244                    let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
2245                    let ruled_seq = self
2246                        .record_step_verdict(&step_path, &folded, cited, supersedes, manifest)
2247                        .await?;
2248                    self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
2249                    if seeded {
2250                        frame.reseed_last(&step_id, ruled, false);
2251                    } else {
2252                        frame.seed_verdict(&step_id, ruled, false);
2253                    }
2254                    self.append(
2255                        &step_path,
2256                        &RunLogPayload::StepExited {
2257                            provider_state_summary: None,
2258                            state: StepState::Judged,
2259                            output: projected.clone(),
2260                            localized: Vec::new(),
2261                            localization_gaps: Vec::new(),
2262                        },
2263                    )?;
2264                    if let Some(projected) = projected {
2265                        frame.outputs.insert(step_id.as_str().to_owned(), projected);
2266                    }
2267                    return Ok(if ruled == VerdictStatus::Fail {
2268                        Ctl::HaltFail
2269                    } else {
2270                        Ctl::Continue
2271                    });
2272                }
2273                Consulted::Retry(policy) => {
2274                    self.backoff_policy(&policy, 0).await;
2275                    active_retry = Some((policy, 1));
2276                    acted = Some(
2277                        self.act_chain(
2278                            frame,
2279                            step,
2280                            &step_path,
2281                            Some(resolved_inputs.clone()),
2282                            None,
2283                            0,
2284                        )
2285                        .await?,
2286                    );
2287                }
2288                Consulted::Repaired | Consulted::RepairDone => {
2289                    // The world was (declared) fixed: one re-entry; a
2290                    // further negative re-consults (maxTriggers bounds
2291                    // the total).
2292                    acted = Some(
2293                        self.act_chain(
2294                            frame,
2295                            step,
2296                            &step_path,
2297                            Some(resolved_inputs.clone()),
2298                            None,
2299                            0,
2300                        )
2301                        .await?,
2302                    );
2303                }
2304                Consulted::Pending(pending) => {
2305                    // The verdict of this round is on the ledger (recorded
2306                    // above): the resume segment re-enters through the
2307                    // recorded-ruling jump. Span stays open.
2308                    return Ok(Ctl::AwaitHuman(pending));
2309                }
2310                Consulted::Propagate(ctl) => return Ok(ctl),
2311            }
2312        }
2313    }
2314
2315    /// Settles a step whose execution ended in a definite negative (fail)
2316    /// or an unconfirmable state (unknown): verdict, write-back, exit.
2317    async fn settle_error(
2318        &mut self,
2319        frame: &mut FrameState<'a>,
2320        step_path: &RunPath,
2321        step_id: &StepId,
2322        status: VerdictStatus,
2323        summary: String,
2324    ) -> Result<Ctl, RunnerError> {
2325        let folded = FoldedVerdict {
2326            status,
2327            degraded: false,
2328            summary,
2329        };
2330        self.record_step_verdict(
2331            step_path,
2332            &folded,
2333            Vec::new(),
2334            None,
2335            EvidenceManifest::default(),
2336        )
2337        .await?;
2338        frame.seed_verdict(step_id, status, false);
2339        self.append(
2340            step_path,
2341            &RunLogPayload::StepExited {
2342                provider_state_summary: None,
2343                state: StepState::Judged,
2344                // No output projection completed on the error path.
2345                output: None,
2346                localized: Vec::new(),
2347                localization_gaps: Vec::new(),
2348            },
2349        )?;
2350        if status == VerdictStatus::Fail {
2351            Ok(Ctl::HaltFail)
2352        } else {
2353            Ok(Ctl::Continue)
2354        }
2355    }
2356
2357    /// The act phase: the bound attempt chain with in-attempt retry
2358    /// (spine §6.5 mount point 1 — every retry is a new callId and a new
2359    /// WAL intent; the chain advances only on `action_failed_final`).
2360    async fn act_chain(
2361        &mut self,
2362        frame: &FrameState<'a>,
2363        step: &'a ActionStepIR,
2364        step_path: &RunPath,
2365        args_override: Option<Value>,
2366        mut adopted: Option<AdoptedSettle>,
2367        start_position: usize,
2368    ) -> Result<ActPhase, RunnerError> {
2369        let key = instance_key(step_path);
2370        let chain_len = step.binding.attempts.len();
2371        for (position, attempt) in step
2372            .binding
2373            .attempts
2374            .iter()
2375            .enumerate()
2376            .skip(start_position)
2377        {
2378            // The entry attempt consumes the override snapshot (the ready
2379            // snapshot on a fresh run, the archived argsSnapshot on a
2380            // crash re-entry — anchored at the recorded chain position,
2381            // 07 §1.4); a chain advance re-resolves the next attempt's
2382            // own argument expressions.
2383            let args = if position == start_position && args_override.is_some() {
2384                args_override.clone().expect("checked is_some")
2385            } else {
2386                match self.resolve_args(frame, attempt) {
2387                    Ok(args) => args,
2388                    Err(message) => {
2389                        return Ok(ActPhase::StepFail {
2390                            class: ErrorClass::BindArgumentsInvalid,
2391                            message,
2392                        });
2393                    }
2394                }
2395            };
2396
2397            let mut tries: u32 = 0;
2398            loop {
2399                tries += 1;
2400                let (outcome, settled_seq, attempt_n) = match adopted.take() {
2401                    // The reconciled terminal is this try's settled
2402                    // outcome; its WAL intent and `actionSettled` are
2403                    // already on record — no dispatch.
2404                    Some(settle) => (settle.outcome, settle.settled_seq, settle.attempt_n),
2405                    None => {
2406                        let attempt_n = self.next_attempt_n(&key);
2407                        let call_id = uuid::Uuid::new_v4().to_string();
2408                        let mut attempt_path = step_path.clone();
2409                        attempt_path.push(PathFrame::Attempt { n: attempt_n });
2410                        // WAL discipline (spine §6.2): the intent commit
2411                        // *is* the fsync; only after it returns may the
2412                        // dispatch leave.
2413                        self.store.write_action_intent(
2414                            &self.run_id,
2415                            now_ms(),
2416                            &attempt_path,
2417                            &call_id,
2418                            args.clone(),
2419                            Some(pointlock_store::IntentDispatch {
2420                                chain_index: (position + 1) as u32,
2421                                channel: attempt.channel,
2422                                action_name: attempt.action_name.clone(),
2423                            }),
2424                        )?;
2425                        let call = BoundActionCall {
2426                            call_id: call_id.clone(),
2427                            action_name: attempt.action_name.clone(),
2428                            arguments: args.clone(),
2429                            action_timeout_ms: step.base.timeout_ms,
2430                            request_timeout_ms: None,
2431                        };
2432                        let outcome = match self.session.execute(call, None).await {
2433                            Ok(outcome) => outcome,
2434                            Err(error) => {
2435                                // No terminal could be obtained: the intent
2436                                // stays pending; suspend and reconcile on
2437                                // resume.
2438                                return Ok(ActPhase::Suspend(format!(
2439                                    "no terminal for callId {call_id}: {error}"
2440                                )));
2441                            }
2442                        };
2443                        let outcome = quarantine_unpersistable(outcome);
2444                        let settled_seq = self.append(
2445                            &attempt_path,
2446                            &RunLogPayload::ActionSettled {
2447                                call_id: call_id.clone(),
2448                                outcome: outcome.clone(),
2449                            },
2450                        )?;
2451                        (outcome, settled_seq, attempt_n)
2452                    }
2453                };
2454                match outcome {
2455                    ActionOutcome::Succeeded { result } => {
2456                        let degraded = !execution_accepted(attempt, &result.execution);
2457                        return Ok(ActPhase::Succeeded {
2458                            result,
2459                            degraded,
2460                            settled_seq,
2461                            attempt_n,
2462                        });
2463                    }
2464                    other => {
2465                        let class = classify(&other);
2466                        let message = terminal_message(&other);
2467                        if class == ErrorClass::ActionCancelled {
2468                            return Ok(ActPhase::Aborted);
2469                        }
2470                        if retry_allowed(step, class, tries) {
2471                            self.backoff(step, tries).await;
2472                            continue;
2473                        }
2474                        match class {
2475                            // Chain advance: only a final (possibly
2476                            // degradable) failure tries the next attempt.
2477                            ErrorClass::ActionFailedFinal if position + 1 < chain_len => break,
2478                            ErrorClass::ActionTimedOut => {
2479                                // A recorded `timedOut` is a certain fate —
2480                                // reconcile returns it verbatim and adds
2481                                // nothing — so the only confirmation channel
2482                                // left is OBSERVATION (spine §5 / 07 §4.3).
2483                                // With assertions declared, the settlement
2484                                // loop asks the world; without any there is
2485                                // nothing that could confirm, and the step
2486                                // folds to the honest unknown directly.
2487                                if !step.assertions.is_empty() {
2488                                    return Ok(ActPhase::Unconfirmed {
2489                                        message: format!("action timed out ({message})"),
2490                                    });
2491                                }
2492                                return Ok(ActPhase::StepUnknown {
2493                                    message: format!(
2494                                        "action timed out and the outcome could not be \
2495                                         confirmed: {message}"
2496                                    ),
2497                                    error_class: None,
2498                                });
2499                            }
2500                            ErrorClass::SessionDegraded => {
2501                                // spine §5: "当前 step → unknown,触发 flow 级
2502                                // onError handler" — both halves, so the
2503                                // class rides along to the hook selector.
2504                                return Ok(ActPhase::StepUnknown {
2505                                    message: format!("session degraded: {message}"),
2506                                    error_class: Some(ErrorClass::SessionDegraded),
2507                                });
2508                            }
2509                            _ => {
2510                                return Ok(ActPhase::StepFail { class, message });
2511                            }
2512                        }
2513                    }
2514                }
2515            }
2516        }
2517        unreachable!("the act chain always returns from within its last attempt")
2518    }
2519
2520    /// Allocates the next attempt number for a step instance (monotonic
2521    /// across resume segments — the base is harvested from the log).
2522    fn next_attempt_n(&mut self, key: &str) -> u64 {
2523        let counter = self.attempt_base.entry(key.to_owned()).or_insert(0);
2524        *counter += 1;
2525        *counter
2526    }
2527
2528    /// Waits out the retry backoff (spine §6.5 mount 1).
2529    async fn backoff(&self, step: &ActionStepIR, tries: u32) {
2530        let Some(policy) = &step.base.retry else {
2531            return;
2532        };
2533        self.backoff_policy(policy, tries).await;
2534    }
2535
2536    /// Sleeps one backoff period of an explicit policy (the handler-retry
2537    /// disposition carries its own policy, independent of `StepBase.retry`
2538    /// — spine §6.5 mount 2).
2539    async fn backoff_policy(&self, policy: &RetryPolicy, tries: u32) {
2540        let ms = backoff_ms(policy, tries);
2541        if ms > 0 {
2542            tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
2543        }
2544    }
2545
2546    // ─── human steps & supervision gate (06 §2/§5; spine §6.8/§6.9) ─────────
2547
2548    /// The R13 supervision gate of one action step, consulted between the
2549    /// preflight probes and the act chain (i.e. strictly before any
2550    /// `actionIntent`). Returns `None` to let the dispatch proceed.
2551    ///
2552    /// A pending gate request from a previous segment resolves first —
2553    /// regardless of *this* segment's policy (the request survives across
2554    /// segments, spine §6.9): `proceed` falls through to the intent,
2555    /// `abort` exits the step `aborted` and aborts the run without
2556    /// consulting any handler, anything else (unanswered, or the non-final
2557    /// `suspend` ruling) re-awaits. Fresh gating follows this segment's
2558    /// policy: `mutating` gates mutating action steps only, `all` gates
2559    /// every action step.
2560    fn gate_supervision(
2561        &mut self,
2562        step_path: &RunPath,
2563        step: &'a ActionStepIR,
2564        resolved_inputs: &Value,
2565        had_open_span: bool,
2566    ) -> Result<Option<Ctl>, RunnerError> {
2567        let key = instance_key(step_path);
2568        // A gate request is live only while the span it was raised in is
2569        // open: a re-executed instance (fresh span, 07 §5.2) is gated
2570        // afresh, never auto-proceeded on a previous life's ruling.
2571        if had_open_span
2572            && let Some(fact) = self
2573                .human
2574                .get(&key)
2575                .filter(|fact| fact.purpose == HumanPurpose::Supervision)
2576        {
2577            let decision = fact
2578                .final_response
2579                .as_ref()
2580                .and_then(|response| response.get("decision"))
2581                .and_then(Value::as_str);
2582            return match decision {
2583                // humanResponded(proceed) is on the ledger before the
2584                // intent this clears the way for (§6.9 WAL order).
2585                Some("proceed") => Ok(None),
2586                Some("abort") => {
2587                    // The human ruling is final: no handler is consulted
2588                    // (§6.9 R13); the step exits `aborted` and the run
2589                    // takes the existing aborted terminal.
2590                    self.append(
2591                        step_path,
2592                        &RunLogPayload::StepExited {
2593                            provider_state_summary: None,
2594                            state: StepState::Aborted,
2595                            output: None,
2596                            localized: Vec::new(),
2597                            localization_gaps: Vec::new(),
2598                        },
2599                    )?;
2600                    Ok(Some(Ctl::Abort))
2601                }
2602                _ => Ok(Some(Ctl::AwaitHuman(HumanPending {
2603                    run_path: fact.run_path.clone(),
2604                    request_id: fact.request_id.clone(),
2605                    purpose: HumanPurpose::Supervision,
2606                    mode: None,
2607                    prompt: fact.prompt.clone(),
2608                    deadline_at_ms: None,
2609                }))),
2610            };
2611        }
2612        let Some(policy) = self.supervise else {
2613            return Ok(None);
2614        };
2615        let gated = match policy {
2616            SupervisePolicy::All => true,
2617            SupervisePolicy::Mutating => step.effect == EffectClassAction::Mutating,
2618        };
2619        if !gated {
2620            return Ok(None);
2621        }
2622        // Fresh gate: auto-generated description over runPath /
2623        // actionName / resolvedInputs (§6.9). No mode, no decisions
2624        // contract, no deadline — the decision vocabulary is the closed
2625        // proceed | abort | suspend, arbitrated by the store.
2626        let attempt = step
2627            .binding
2628            .attempts
2629            .first()
2630            .expect("sealed action steps carry at least one bound attempt");
2631        let action_name = attempt.action_name.as_str().to_owned();
2632        let rendered = render_run_path(step_path);
2633        let request_id = uuid::Uuid::new_v4().to_string();
2634        let prompt = format!(
2635            "Supervision gate: approve dispatching action '{action_name}' at {rendered}? \
2636             The resolved inputs are presented."
2637        );
2638        let presents = serde_json::json!([
2639            { "kind": "value", "label": "runPath", "value": rendered },
2640            { "kind": "value", "label": "actionName", "value": action_name },
2641            { "kind": "value", "label": "resolvedInputs", "value": resolved_inputs },
2642        ]);
2643        // fsync-before-notify (spine §6.9): the append commit *is* the
2644        // fsync; the runner then suspends — any notification happens in
2645        // the CLI layer, strictly after this returns.
2646        self.append(
2647            step_path,
2648            &RunLogPayload::HumanRequested {
2649                request_id: request_id.clone(),
2650                purpose: HumanPurpose::Supervision,
2651                mode: None,
2652                prompt: prompt.clone(),
2653                presents,
2654                decisions: None,
2655                output_schema: None,
2656                deadline_at_ms: None,
2657            },
2658        )?;
2659        Ok(Some(Ctl::AwaitHuman(HumanPending {
2660            run_path: step_path.clone(),
2661            request_id,
2662            purpose: HumanPurpose::Supervision,
2663            mode: None,
2664            prompt,
2665            deadline_at_ms: None,
2666        })))
2667    }
2668
2669    /// Executes one human step (06 §5.1 pinned order): ready (presents
2670    /// materialized once and frozen) → `stepEntered` → declared preflight
2671    /// → `humanRequested` (fsynced by its commit) → suspend /
2672    /// [`RunOutcome::AwaitingHuman`]. Resume settles a paired response
2673    /// through the four-mode mapping, re-awaits an unanswered request
2674    /// inside its deadline, and lazily settles an expired one to `unknown`
2675    /// — the settlement result depends only on `deadlineAtMs` and response
2676    /// presence, never on the settlement instant (06 §5.3).
2677    async fn exec_human(
2678        &mut self,
2679        frame: &mut FrameState<'a>,
2680        step_path: RunPath,
2681        step: &'a HumanStepIR,
2682    ) -> Result<Ctl, RunnerError> {
2683        let step_id = step.base.step_id.clone();
2684        let key = instance_key(&step_path);
2685
2686        // A request already on the ledger for this instance: settle or
2687        // keep waiting — never a second `humanRequested` while one is
2688        // pending. The request is live only while its span is open: a
2689        // re-executed instance (positional invalidation, 07 §5.2) enters
2690        // a fresh span and must ask again, never settle from a previous
2691        // life's answer.
2692        if self.open_span_inputs(&key).is_some()
2693            && let Some(fact) = self
2694                .human
2695                .get(&key)
2696                .filter(|fact| fact.purpose == HumanPurpose::Step)
2697        {
2698            let fact = fact.clone();
2699            if let Some(response) = fact.final_response.clone() {
2700                // The span is open by construction (settlement closes it
2701                // for good); enter_step only consumes it.
2702                self.enter_step(&step_path, &step.base, Value::Null)?;
2703                return self
2704                    .settle_human_response(frame, &step_path, step, &fact, response)
2705                    .await;
2706            }
2707            match fact.deadline_at_ms {
2708                Some(deadline) if self.now() > deadline => {
2709                    self.enter_step(&step_path, &step.base, Value::Null)?;
2710                    return self
2711                        .settle_human_timeout(frame, &step_path, step, &fact)
2712                        .await;
2713                }
2714                // Unanswered and not expired: re-await the same request
2715                // (no new requestId, no re-notify obligation here).
2716                _ => {
2717                    return Ok(Ctl::AwaitHuman(HumanPending {
2718                        run_path: fact.run_path.clone(),
2719                        request_id: fact.request_id.clone(),
2720                        purpose: HumanPurpose::Step,
2721                        mode: Some(step.mode),
2722                        prompt: fact.prompt.clone(),
2723                        deadline_at_ms: fact.deadline_at_ms,
2724                    }));
2725                }
2726            }
2727        }
2728
2729        // Ready: materialize `presents` once and freeze — the
2730        // `resolvedInputs` snapshot discipline (06 §2.3). A
2731        // crash/suspension-opened span reuses the archived snapshot.
2732        let snapshot = match self.open_span_inputs(&key) {
2733            Some(archived) => archived,
2734            None => {
2735                let scope = frame.scope(&self.env, None);
2736                let mut items = Vec::with_capacity(step.presents.len());
2737                let mut error = None;
2738                for (index, expr) in step.presents.iter().enumerate() {
2739                    match pointlock_expr::eval(expr, &scope) {
2740                        Ok(value) => items.push(value),
2741                        Err(eval_error) => {
2742                            error =
2743                                Some(format!("present #{index} evaluation failed: {eval_error}"));
2744                            break;
2745                        }
2746                    }
2747                }
2748                if let Some(message) = error {
2749                    // A failing presents evaluation is a compiler /
2750                    // expression bug signal (bind_arguments_invalid
2751                    // discipline): step fails, nothing is asked.
2752                    self.enter_step(&step_path, &step.base, Value::Null)?;
2753                    return self
2754                        .settle_error(
2755                            frame,
2756                            &step_path,
2757                            &step_id,
2758                            VerdictStatus::Fail,
2759                            format!("human presents failed [bind_arguments_invalid]: {message}"),
2760                        )
2761                        .await;
2762                }
2763                serde_json::json!({ "presents": items })
2764            }
2765        };
2766        self.enter_step(&step_path, &step.base, snapshot.clone())?;
2767        if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
2768            return Ok(ctl);
2769        }
2770        let presents = snapshot
2771            .get("presents")
2772            .cloned()
2773            .unwrap_or(Value::Array(Vec::new()));
2774        let request_id = uuid::Uuid::new_v4().to_string();
2775        // `timeoutMs` converts to the absolute deadline watermark at
2776        // request creation (06 §5.3) — the sole lazy-settlement input.
2777        let deadline_at_ms = self.now().saturating_add(step.timeout_ms);
2778        // fsync-before-notify (spine §6.8, 06 §5.1): the append commit
2779        // *is* the fsync (WAL + synchronous=FULL); the runner suspends
2780        // and never notifies — channels are the CLI layer's job.
2781        self.append(
2782            &step_path,
2783            &RunLogPayload::HumanRequested {
2784                request_id: request_id.clone(),
2785                purpose: HumanPurpose::Step,
2786                mode: Some(step.mode),
2787                prompt: step.prompt.clone(),
2788                presents,
2789                decisions: step.decisions.clone(),
2790                output_schema: step.output_schema.clone(),
2791                deadline_at_ms: Some(deadline_at_ms),
2792            },
2793        )?;
2794        Ok(Ctl::AwaitHuman(HumanPending {
2795            run_path: step_path,
2796            request_id,
2797            purpose: HumanPurpose::Step,
2798            mode: Some(step.mode),
2799            prompt: step.prompt.clone(),
2800            deadline_at_ms: Some(deadline_at_ms),
2801        }))
2802    }
2803
2804    /// Settles a human step from its arbitrated final response — the
2805    /// four-mode verdict/output mapping (06 §2.2 as adjudicated):
2806    ///
2807    /// | mode | response | verdict | step output |
2808    /// |---|---|---|---|
2809    /// | `confirm` | `decision` = first label | pass | the response object |
2810    /// | `confirm` | `decision` = second label | fail | the response object |
2811    /// | `judge` | `status` pass/fail/unknown | verbatim | the response object |
2812    /// | `provideInput` | `input` (schema-checked by the store) | pass | the input value |
2813    /// | `repairWorld` | `decision: "done"` | pass (testimony, not observation; never degraded) | the response object |
2814    /// | `repairWorld` | `decision: "cannotRepair"` | fail (06 §2.2 — a verdict, catchable by `onFail`; never a run abort) | the response object |
2815    ///
2816    /// Every settlement materializes the response as a canonical JSON
2817    /// evidence document and cites it from the verdict (06 §6).
2818    async fn settle_human_response(
2819        &mut self,
2820        frame: &mut FrameState<'a>,
2821        step_path: &RunPath,
2822        step: &'a HumanStepIR,
2823        fact: &HumanRequestFact,
2824        response: Value,
2825    ) -> Result<Ctl, RunnerError> {
2826        let step_id = step.base.step_id.clone();
2827        let decision = response
2828            .get("decision")
2829            .and_then(Value::as_str)
2830            .unwrap_or_default()
2831            .to_owned();
2832        let (status, output, summary) = match step.mode {
2833            HumanMode::Confirm => {
2834                let labels = step
2835                    .decisions
2836                    .as_ref()
2837                    .expect("load validated confirm decisions");
2838                // Position-mapped double label: first → pass, second →
2839                // fail (membership was the store arbitration's check).
2840                let status = if labels.first().map(String::as_str) == Some(decision.as_str()) {
2841                    VerdictStatus::Pass
2842                } else {
2843                    VerdictStatus::Fail
2844                };
2845                let position = if status == VerdictStatus::Pass {
2846                    "first"
2847                } else {
2848                    "second"
2849                };
2850                (
2851                    status,
2852                    Some(response.clone()),
2853                    format!(
2854                        "human confirm decision '{decision}' is the {position} label \
2855                         (position-mapped verdict)"
2856                    ),
2857                )
2858            }
2859            HumanMode::Judge => {
2860                // The human ruling *is* the verdict (spine §6.3).
2861                let status = match response.get("status").and_then(Value::as_str) {
2862                    Some("pass") => VerdictStatus::Pass,
2863                    Some("fail") => VerdictStatus::Fail,
2864                    _ => VerdictStatus::Unknown,
2865                };
2866                let label = response
2867                    .get("status")
2868                    .and_then(Value::as_str)
2869                    .unwrap_or("unknown");
2870                (
2871                    status,
2872                    Some(response.clone()),
2873                    format!("human judge ruling: {label}"),
2874                )
2875            }
2876            HumanMode::ProvideInput => {
2877                // The store validated `input` against `outputSchema`; the
2878                // established fact is "a human provided schema-valid
2879                // input" — pass, and the input *is* the step output.
2880                let input = response.get("input").cloned().unwrap_or(Value::Null);
2881                (
2882                    VerdictStatus::Pass,
2883                    Some(input),
2884                    "human provided input validated against the outputSchema".to_owned(),
2885                )
2886            }
2887            HumanMode::RepairWorld => {
2888                // 06 §2.2's closed mapping: `done` → pass, `cannotRepair`
2889                // → fail. Testimony, not observation — pass carries no
2890                // degraded flag (machine re-checks belong to follow-up
2891                // assert/preflight steps), and a fail is a verdict the
2892                // step's own `onFail` may catch, never a unilateral run
2893                // abort.
2894                let status = if decision == "done" {
2895                    VerdictStatus::Pass
2896                } else {
2897                    VerdictStatus::Fail
2898                };
2899                let summary = if status == VerdictStatus::Pass {
2900                    "human declared the world repaired (testimony; follow-up machine \
2901                     re-check advised)"
2902                        .to_owned()
2903                } else {
2904                    "human declared the world unrepairable (cannotRepair)".to_owned()
2905                };
2906                (status, Some(response.clone()), summary)
2907            }
2908        };
2909        let asset = self.put_human_evidence(step_path, fact, Some(&response), "response")?;
2910        let folded = FoldedVerdict {
2911            status,
2912            degraded: false,
2913            summary,
2914        };
2915        let verdict_seq = self
2916            .record_step_verdict(
2917                step_path,
2918                &folded,
2919                vec![asset.clone()],
2920                None,
2921                human_manifest(&asset),
2922            )
2923            .await?;
2924        if let Some(sha256) = &asset.sha256 {
2925            self.store
2926                .link_evidence(&self.run_id, verdict_seq, &asset.id, sha256)?;
2927        }
2928        frame.seed_verdict(&step_id, status, false);
2929        self.append(
2930            step_path,
2931            &RunLogPayload::StepExited {
2932                provider_state_summary: None,
2933                state: StepState::Judged,
2934                output: output.clone(),
2935                localized: Vec::new(),
2936                localization_gaps: Vec::new(),
2937            },
2938        )?;
2939        if let Some(output) = output {
2940            frame.outputs.insert(step_id.as_str().to_owned(), output);
2941        }
2942        if status == VerdictStatus::Fail {
2943            Ok(Ctl::HaltFail)
2944        } else {
2945            Ok(Ctl::Continue)
2946        }
2947    }
2948
2949    /// Lazy timeout settlement (06 §5.3): the deadline passed with no
2950    /// arbitrated response — verdict `unknown` (`onTimeout` is fixed),
2951    /// no output (downstream consumers of it block). The judgment inputs
2952    /// are `deadlineAtMs` and response absence only; the settlement
2953    /// instant leaves no trace in the verdict ("no one came" is itself a
2954    /// recorded historical fact, 06 §6).
2955    async fn settle_human_timeout(
2956        &mut self,
2957        frame: &mut FrameState<'a>,
2958        step_path: &RunPath,
2959        step: &'a HumanStepIR,
2960        fact: &HumanRequestFact,
2961    ) -> Result<Ctl, RunnerError> {
2962        let step_id = &step.base.step_id;
2963        let deadline = fact.deadline_at_ms.unwrap_or_default();
2964        let asset = self.put_human_evidence(step_path, fact, None, "timeout")?;
2965        let folded = FoldedVerdict {
2966            status: VerdictStatus::Unknown,
2967            degraded: false,
2968            summary: format!(
2969                "human response deadline (deadlineAtMs {deadline}) passed without a \
2970                 response; onTimeout is fixed to unknown"
2971            ),
2972        };
2973        let verdict_seq = self
2974            .record_step_verdict(
2975                step_path,
2976                &folded,
2977                vec![asset.clone()],
2978                None,
2979                human_manifest(&asset),
2980            )
2981            .await?;
2982        if let Some(sha256) = &asset.sha256 {
2983            self.store
2984                .link_evidence(&self.run_id, verdict_seq, &asset.id, sha256)?;
2985        }
2986        frame.seed_verdict(step_id, VerdictStatus::Unknown, false);
2987
2988        // A timed-out human step is an unknown verdict: the onUnknown
2989        // ladder applies (06's escalation pattern). Re-ask dispositions
2990        // (retry/repair) need fresh-request machinery — typed M2 refusal.
2991        match self
2992            .consult_hook(
2993                frame,
2994                step_path,
2995                step.base.handlers.as_deref(),
2996                HandlerHook::OnUnknown,
2997                None,
2998            )
2999            .await?
3000        {
3001            Consulted::None | Consulted::RepairFailed | Consulted::Continue => {}
3002            Consulted::Abort => {
3003                self.append(
3004                    step_path,
3005                    &RunLogPayload::StepExited {
3006                        provider_state_summary: None,
3007                        state: StepState::Aborted,
3008                        output: None,
3009                        localized: Vec::new(),
3010                        localization_gaps: Vec::new(),
3011                    },
3012                )?;
3013                return Ok(Ctl::Abort);
3014            }
3015            Consulted::Escalated {
3016                status: ruled,
3017                summary: ruled_summary,
3018                evidence: ruling_evidence,
3019            } => {
3020                let folded = FoldedVerdict {
3021                    status: ruled,
3022                    degraded: false,
3023                    summary: ruled_summary,
3024                };
3025                let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
3026                let ruled_seq = self
3027                    .record_step_verdict(
3028                        step_path,
3029                        &folded,
3030                        cited,
3031                        Some(format!("seq:{verdict_seq}")),
3032                        manifest,
3033                    )
3034                    .await?;
3035                self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
3036                frame.reseed_last(step_id, ruled, false);
3037                self.append(
3038                    step_path,
3039                    &RunLogPayload::StepExited {
3040                        provider_state_summary: None,
3041                        state: StepState::Judged,
3042                        output: None,
3043                        localized: Vec::new(),
3044                        localization_gaps: Vec::new(),
3045                    },
3046                )?;
3047                return Ok(if ruled == VerdictStatus::Fail {
3048                    Ctl::HaltFail
3049                } else {
3050                    Ctl::Continue
3051                });
3052            }
3053            Consulted::Retry(_) | Consulted::Repaired | Consulted::RepairDone => {
3054                return Err(RunnerError::M0Unsupported {
3055                    detail: format!(
3056                        "re-ask dispositions on the timed-out human step '{step_id}' need \
3057                         fresh-request machinery — not in the M2 subset \
3058                         (escalate/continue/abort are supported)"
3059                    ),
3060                });
3061            }
3062            Consulted::Pending(pending) => return Ok(Ctl::AwaitHuman(pending)),
3063            Consulted::Propagate(ctl) => return Ok(ctl),
3064        }
3065
3066        self.append(
3067            step_path,
3068            &RunLogPayload::StepExited {
3069                provider_state_summary: None,
3070                state: StepState::Judged,
3071                output: None,
3072                localized: Vec::new(),
3073                localization_gaps: Vec::new(),
3074            },
3075        )?;
3076        Ok(Ctl::Continue)
3077    }
3078
3079    /// Materializes a human settlement as a canonical JSON evidence
3080    /// document in the content-addressed store and mints its citable
3081    /// [`AssetRef`] (06 §6: who / when-relative-to-deadline / what was
3082    /// decided / what was presented). Deliberately clock-free: the
3083    /// document is a pure function of the request and the arbitrated
3084    /// response (or its absence).
3085    /// Links an escalate ruling's settlement document to its superseding
3086    /// verdict row (same file-before-row-before-log join as body humans).
3087    fn link_ruling_evidence(
3088        &mut self,
3089        verdict_seq: u64,
3090        evidence: &Option<AssetRef>,
3091    ) -> Result<(), RunnerError> {
3092        if let Some(asset) = evidence
3093            && let Some(sha256) = &asset.sha256
3094        {
3095            self.store
3096                .link_evidence(&self.run_id, verdict_seq, &asset.id, sha256)?;
3097        }
3098        Ok(())
3099    }
3100
3101    fn put_human_evidence(
3102        &mut self,
3103        step_path: &RunPath,
3104        fact: &HumanRequestFact,
3105        response: Option<&Value>,
3106        settled_as: &str,
3107    ) -> Result<AssetRef, RunnerError> {
3108        let document = serde_json::json!({
3109            "pointlockEvidence": "humanResponse/1",
3110            "requestId": fact.request_id,
3111            "runId": self.run_id,
3112            "runPath": render_run_path(step_path),
3113            "purpose": fact.purpose,
3114            "mode": fact.mode,
3115            "prompt": fact.prompt,
3116            "presented": fact.presents,
3117            "response": response.cloned().unwrap_or(Value::Null),
3118            "actor": match (settled_as, &fact.final_actor) {
3119                // Timeout settlements have no actor (06 §6).
3120                ("timeout", _) => Value::Null,
3121                (_, Some(actor)) => Value::String(actor.clone()),
3122                (_, None) => Value::Null,
3123            },
3124            "deadlineAtMs": fact.deadline_at_ms,
3125            "settledAs": settled_as,
3126        });
3127        let bytes = to_canonical_json(&document).into_bytes();
3128        // file-before-row-before-log: the bytes are durable before the
3129        // verdict event cites them.
3130        let put = self.store.put_evidence(&bytes, "application/json")?;
3131        Ok(AssetRef {
3132            id: format!("humanResponse:{}", fact.request_id),
3133            media_type: "application/json".to_owned(),
3134            // Locally minted evidence: the URI is the content-addressed
3135            // library path (06 §6).
3136            uri: put.local_path,
3137            sha256: Some(put.sha256),
3138        })
3139    }
3140
3141    // ─── handler engine (spine §3/§6.5 mount 2; M2 W3) ──────────────────────
3142    //
3143    // Handlers are explicit policies on four hooks; they yield a
3144    // disposition, never data (R10). Consultation happens *inside* the
3145    // host step's open span, before `stepExited`, so a retry disposition
3146    // re-enters the failing phase within the same entered/exited pair.
3147    // The `handlerTriggered` audit event anchors at the host step path;
3148    // the hook frame (`/hook:<name>:<n>`) anchors the disposition's own
3149    // work (escalate humans, repair frames).
3150
3151    /// Resolves the binding a hook consults: the step-level list first
3152    /// (first match wins), else the flow-level list (spine §3
3153    /// `StepBase.handlers` overrides `FlowIR.handlers`). `onError`
3154    /// bindings additionally filter on `errorClasses`.
3155    fn resolve_binding(
3156        &self,
3157        frame: &FrameState<'a>,
3158        step_handlers: Option<&'a [HandlerBinding]>,
3159        hook: HandlerHook,
3160        error_class: Option<ErrorClass>,
3161    ) -> Option<&'a HandlerBinding> {
3162        let matches = |binding: &&'a HandlerBinding| {
3163            binding.hook == hook
3164                && (hook != HandlerHook::OnError
3165                    || match (&binding.error_classes, error_class) {
3166                        (None, _) => true,
3167                        (Some(filter), Some(class)) => filter.contains(&class),
3168                        (Some(_), None) => false,
3169                    })
3170        };
3171        step_handlers
3172            .and_then(|bindings| bindings.iter().find(matches))
3173            .or_else(|| {
3174                frame
3175                    .flow
3176                    .handlers
3177                    .as_deref()
3178                    .and_then(|bindings| bindings.iter().find(matches))
3179            })
3180    }
3181
3182    /// Consults the matching handler binding for `hook` on the host step.
3183    ///
3184    /// Trigger counting is per host instance per hook per execution life
3185    /// and continues across segments (harvested from the ledger; a new
3186    /// life resets it — see `enter_step`); an exhausted budget
3187    /// returns [`Consulted::None`] — the natural path stands. A pending
3188    /// escalate continuation (a previous segment's unanswered hook human)
3189    /// is settled or re-awaited *without* consuming a new trigger.
3190    async fn consult_hook(
3191        &mut self,
3192        frame: &mut FrameState<'a>,
3193        step_path: &RunPath,
3194        step_handlers: Option<&'a [HandlerBinding]>,
3195        hook: HandlerHook,
3196        error_class: Option<ErrorClass>,
3197    ) -> Result<Consulted, RunnerError> {
3198        let Some(binding) = self.resolve_binding(frame, step_handlers, hook, error_class) else {
3199            return Ok(Consulted::None);
3200        };
3201        let counter_key = crate::align::hook_trigger_key(&instance_key(step_path), hook);
3202        let current = self.hook_triggers.get(&counter_key).copied().unwrap_or(0);
3203
3204        // Escalate continuation: trigger N already on the ledger and its
3205        // hook human still governs — settle or re-await, no new trigger.
3206        if current >= 1
3207            && let HandlerAction::Escalate { human } = &binding.action
3208        {
3209            let human_path = hook_child_path(step_path, hook, current, human);
3210            if self.human.contains_key(&instance_key(&human_path)) {
3211                // A request exists for the current trigger: it fully
3212                // governs (settle its response, its timeout, or re-await).
3213                return self
3214                    .run_hook_human(frame, step_path, hook, current, human)
3215                    .await;
3216            }
3217        }
3218
3219        let trigger = current + 1;
3220        if trigger > u64::from(binding.max_triggers) {
3221            return Ok(Consulted::None);
3222        }
3223        self.hook_triggers.insert(counter_key, trigger);
3224        let disposition = match &binding.action {
3225            HandlerAction::Retry { .. } => "retry",
3226            HandlerAction::Continue => "continue",
3227            HandlerAction::Abort => "abort",
3228            HandlerAction::Escalate { .. } => "escalate",
3229            HandlerAction::Repair { .. } => "repair",
3230        };
3231        self.append(
3232            step_path,
3233            &RunLogPayload::HandlerTriggered {
3234                hook,
3235                trigger,
3236                disposition: Some(disposition.to_owned()),
3237            },
3238        )?;
3239
3240        match &binding.action {
3241            HandlerAction::Retry { policy } => Ok(Consulted::Retry(policy.clone())),
3242            HandlerAction::Continue => Ok(Consulted::Continue),
3243            HandlerAction::Abort => Ok(Consulted::Abort),
3244            HandlerAction::Escalate { human } => {
3245                self.run_hook_human(frame, step_path, hook, trigger, human)
3246                    .await
3247            }
3248            HandlerAction::Repair { flow_ref } => {
3249                self.run_repair(frame, step_path, hook, trigger, flow_ref)
3250                    .await
3251            }
3252        }
3253    }
3254
3255    /// Runs (or settles) an escalate hook human. Hook humans are not body
3256    /// steps: they open no span and seed no frame verdict — their ruling
3257    /// is returned to the consultation site, which supersedes the host
3258    /// verdict and cites the canonical settlement evidence document
3259    /// minted here (06 §6; the request/response ledger pair is the join).
3260    async fn run_hook_human(
3261        &mut self,
3262        frame: &mut FrameState<'a>,
3263        step_path: &RunPath,
3264        hook: HandlerHook,
3265        trigger: u64,
3266        human: &'a HumanStepIR,
3267    ) -> Result<Consulted, RunnerError> {
3268        let human_path = hook_child_path(step_path, hook, trigger, human);
3269        let key = instance_key(&human_path);
3270        if let Some(fact) = self.human.get(&key) {
3271            let fact = fact.clone();
3272            if let Some(response) = fact.final_response.clone() {
3273                // Consume the settlement: a ruling governs exactly once —
3274                // a later consult on the same host walks a *new* trigger
3275                // (or exhausts the budget), never re-reads this answer.
3276                self.human.remove(&key);
3277                // Every settlement materializes the canonical evidence
3278                // document (06 §6); the Escalated superseding verdict
3279                // cites it below. Non-verdict dispositions (Repaired/
3280                // Abort) keep it durable in the library with the
3281                // request/response pair as the join.
3282                let asset =
3283                    self.put_human_evidence(&human_path, &fact, Some(&response), "response")?;
3284                return Ok(match map_escalate_response(human, &response) {
3285                    Consulted::Escalated {
3286                        status, summary, ..
3287                    } => Consulted::Escalated {
3288                        status,
3289                        summary,
3290                        evidence: Some(asset),
3291                    },
3292                    other => other,
3293                });
3294            }
3295            match fact.deadline_at_ms {
3296                Some(deadline) if self.now() > deadline => {
3297                    // Lazy timeout settlement: unknown, fixed (onTimeout);
3298                    // consumed like any settlement — with the canonical
3299                    // evidence document (06 §6, actor null on timeout).
3300                    self.human.remove(&key);
3301                    let asset = self.put_human_evidence(&human_path, &fact, None, "timeout")?;
3302                    return Ok(Consulted::Escalated {
3303                        status: VerdictStatus::Unknown,
3304                        summary: format!(
3305                            "escalate human '{}' timed out (deadline watermark passed): unknown",
3306                            human.base.step_id
3307                        ),
3308                        evidence: Some(asset),
3309                    });
3310                }
3311                _ => {
3312                    return Ok(Consulted::Pending(HumanPending {
3313                        run_path: fact.run_path.clone(),
3314                        request_id: fact.request_id.clone(),
3315                        purpose: HumanPurpose::Step,
3316                        mode: Some(human.mode),
3317                        prompt: fact.prompt.clone(),
3318                        deadline_at_ms: fact.deadline_at_ms,
3319                    }));
3320                }
3321            }
3322        }
3323
3324        // First encounter: materialize presents in the host frame's scope
3325        // and freeze; fsync-before-notify discipline as everywhere.
3326        let scope = frame.scope(&self.env, None);
3327        let mut items = Vec::with_capacity(human.presents.len());
3328        for expr in &human.presents {
3329            match pointlock_expr::eval(expr, &scope) {
3330                Ok(value) => items.push(value),
3331                // A failing presents expression degrades to an empty
3332                // exhibit — the request still goes out (the human can
3333                // rule without exhibits; principle 8 over strictness).
3334                Err(_) => items.push(Value::Null),
3335            }
3336        }
3337        let request_id = uuid::Uuid::new_v4().to_string();
3338        let deadline_at_ms = self.now().saturating_add(human.timeout_ms);
3339        self.append(
3340            &human_path,
3341            &RunLogPayload::HumanRequested {
3342                request_id: request_id.clone(),
3343                purpose: HumanPurpose::Step,
3344                mode: Some(human.mode),
3345                prompt: human.prompt.clone(),
3346                presents: Value::Array(items),
3347                decisions: human.decisions.clone(),
3348                output_schema: human.output_schema.clone(),
3349                deadline_at_ms: Some(deadline_at_ms),
3350            },
3351        )?;
3352        Ok(Consulted::Pending(HumanPending {
3353            run_path: human_path,
3354            request_id,
3355            purpose: HumanPurpose::Step,
3356            mode: Some(human.mode),
3357            prompt: human.prompt.clone(),
3358            deadline_at_ms: Some(deadline_at_ms),
3359        }))
3360    }
3361
3362    /// Runs a repair subflow under the hook frame (a call frame without a
3363    /// host call step — spine §9). Repair flows take no caller inputs
3364    /// (their params materialize from declared defaults, 06 §7.5 binding-
3365    /// flow pattern); they yield no data (R10) — only their flow verdict
3366    /// comes back as the disposition signal.
3367    async fn run_repair(
3368        &mut self,
3369        frame: &mut FrameState<'a>,
3370        step_path: &RunPath,
3371        hook: HandlerHook,
3372        trigger: u64,
3373        flow_ref: &'a pointlock_ir::FlowRef,
3374    ) -> Result<Consulted, RunnerError> {
3375        let callee = self.flows.callee(flow_ref);
3376        let mut repair_path = step_path.clone();
3377        repair_path.push(hook_frame(hook, trigger));
3378        repair_path.push(PathFrame::Call {
3379            step_id: None,
3380            callee_flow_id: callee.flow_id.clone(),
3381            callee_ir_hash: callee.ir_hash.clone(),
3382        });
3383        // Defaults-only inbound materialization.
3384        let params = match call_inputs_gate(callee, Map::new()) {
3385            Ok(params) => params,
3386            // A repair flow whose params cannot materialize from defaults
3387            // is a repair failure; its detail is compiler-diagnosable.
3388            Err(_) => return Ok(Consulted::RepairFailed),
3389        };
3390        self.append(
3391            &repair_path,
3392            &RunLogPayload::CallFramePushed {
3393                frame: CallFrame {
3394                    flow_id: callee.flow_id.clone(),
3395                    ir_hash: callee.ir_hash.clone(),
3396                    call_step_id: None,
3397                    inputs_snapshot: Value::Object(params.clone()),
3398                    vars: BTreeMap::new(),
3399                    iter_stack: Vec::new(),
3400                    next_index: 0,
3401                },
3402                rebase: false,
3403            },
3404        )?;
3405        let mut repair_frame =
3406            FrameState::new(callee, repair_path.clone(), params, frame.depth + 1);
3407        let ctl = self
3408            .exec_body(&mut repair_frame, repair_path.clone(), &callee.body, 0)
3409            .await?;
3410        match ctl {
3411            Ctl::Continue | Ctl::HaltFail => {}
3412            other => {
3413                // Suspension inside a repair leaves its frame live; the
3414                // resume path refuses live hook frames with a typed error
3415                // (registered M2 limitation) — the ledger stays honest.
3416                return Ok(Consulted::Propagate(other));
3417            }
3418        }
3419        self.append(
3420            &repair_path,
3421            &RunLogPayload::CallFramePopped { outputs: None },
3422        )?;
3423        let verdict = fold_flow_verdict(&repair_frame.fold, callee.verdict_policy);
3424        match (ctl, verdict) {
3425            (Ctl::HaltFail, _) => Ok(Consulted::RepairFailed),
3426            (_, Some(folded)) if folded.status != VerdictStatus::Pass => {
3427                Ok(Consulted::RepairFailed)
3428            }
3429            _ => Ok(Consulted::RepairDone),
3430        }
3431    }
3432
3433    // ─── call steps (07 §1) ─────────────────────────────────────────────────
3434
3435    async fn exec_call(
3436        &mut self,
3437        frame: &mut FrameState<'a>,
3438        call_path: RunPath,
3439        step: &'a CallStepIR,
3440    ) -> Result<Ctl, RunnerError> {
3441        let step_id = step.base.step_id.clone();
3442        let key = instance_key(&call_path);
3443        let callee: &'a FlowIR = self.flows.callee(&step.flow_ref);
3444        // Runtime defense line for maxCallDepth (the load check already
3445        // bounds the static closure; this guards the walk itself).
3446        if frame.depth + 1 > MAX_CALL_DEPTH {
3447            return Err(RunnerError::CallDepthExceeded {
3448                depth: frame.depth + 1,
3449                max: MAX_CALL_DEPTH,
3450            });
3451        }
3452
3453        // Ready: call-by-value inputs snapshot. A suspension-opened span
3454        // reuses the archived snapshot (already gated) — never
3455        // re-evaluated (spine §6.6, 07 §5.2 corollary).
3456        let archived = self.open_span_inputs(&key);
3457        let gated = match archived {
3458            Some(Value::Object(map)) => map,
3459            Some(other) => {
3460                // A non-object archived snapshot is a ledger anomaly; the
3461                // honest disposition is a bind-class failure.
3462                self.enter_step(&call_path, &step.base, other)?;
3463                return self
3464                    .settle_error(
3465                        frame,
3466                        &call_path,
3467                        &step_id,
3468                        VerdictStatus::Fail,
3469                        "archived call inputs snapshot is not an object".to_owned(),
3470                    )
3471                    .await;
3472            }
3473            None => {
3474                // Evaluate each input expression in the *caller* scope.
3475                let scope = frame.scope(&self.env, None);
3476                let mut inputs = Map::new();
3477                let mut eval_error = None;
3478                for (name, expr) in step.inputs.iter() {
3479                    match pointlock_expr::eval(expr, &scope) {
3480                        Ok(value) => {
3481                            inputs.insert(name.as_str().to_owned(), value);
3482                        }
3483                        Err(error) => {
3484                            eval_error = Some(format!("input '{name}' evaluation failed: {error}"));
3485                            break;
3486                        }
3487                    }
3488                }
3489                if let Some(message) = eval_error {
3490                    self.enter_step(&call_path, &step.base, Value::Null)?;
3491                    return self
3492                        .settle_error(
3493                            frame,
3494                            &call_path,
3495                            &step_id,
3496                            VerdictStatus::Fail,
3497                            format!("call inputs failed [bind_arguments_invalid]: {message}"),
3498                        )
3499                        .await;
3500                }
3501                // Inbound gate: defaults + per-param schema validation
3502                // (07 §1.1 — runtime re-check; failure classifies
3503                // bind_arguments_invalid, no retry).
3504                match call_inputs_gate(callee, inputs.clone()) {
3505                    Ok(gated) => gated,
3506                    Err(message) => {
3507                        self.enter_step(&call_path, &step.base, Value::Object(inputs))?;
3508                        return self
3509                            .settle_error(
3510                                frame,
3511                                &call_path,
3512                                &step_id,
3513                                VerdictStatus::Fail,
3514                                format!("call inputs failed [bind_arguments_invalid]: {message}"),
3515                            )
3516                            .await;
3517                    }
3518                }
3519            }
3520        };
3521        self.enter_step(&call_path, &step.base, Value::Object(gated.clone()))?;
3522
3523        if let Some(ctl) = self.probe_or_note(frame, &call_path, &step.base).await? {
3524            return Ok(ctl);
3525        }
3526
3527        // Frame push (07 §3.1 frame-transfer materialization point) —
3528        // unless a previous segment already pushed it and we are resuming
3529        // back into the live frame. Re-entering one whose callee pin moved
3530        // is announced as a `rebase`, so `frames` names the callee actually
3531        // executing rather than the one the crashed segment entered
3532        // (07 §5.2 case (a)); the fold updates that frame's `irHash` in
3533        // place and touches nothing else.
3534        let open_pin = self.live_frames.remove(&key);
3535        let rebase = match &open_pin {
3536            None => false,
3537            Some(pin) => *pin != callee.ir_hash,
3538        };
3539        if open_pin.is_none() || rebase {
3540            self.append(
3541                &call_path,
3542                &RunLogPayload::CallFramePushed {
3543                    frame: CallFrame {
3544                        flow_id: callee.flow_id.clone(),
3545                        ir_hash: callee.ir_hash.clone(),
3546                        call_step_id: Some(step_id.clone()),
3547                        inputs_snapshot: Value::Object(gated.clone()),
3548                        vars: BTreeMap::new(),
3549                        iter_stack: Vec::new(),
3550                        next_index: 0,
3551                    },
3552                    rebase,
3553                },
3554            )?;
3555        }
3556
3557        // The callee body runs in a fresh scope: params = inputs, env
3558        // passes through read-only, the caller's steps/vars are invisible
3559        // (07 §1.2 — hard boundary).
3560        let mut callee_frame = FrameState::new(callee, call_path.clone(), gated, frame.depth + 1);
3561        let ctl = self
3562            .exec_body(&mut callee_frame, call_path.clone(), &callee.body, 0)
3563            .await?;
3564        match ctl {
3565            Ctl::Continue | Ctl::HaltFail => {}
3566            Ctl::Abort => {
3567                // Unwind the frame so the ledger stays balanced; an
3568                // aborted run makes no semantic claim.
3569                self.append(
3570                    &call_path,
3571                    &RunLogPayload::CallFramePopped { outputs: None },
3572                )?;
3573                self.append(
3574                    &call_path,
3575                    &RunLogPayload::StepExited {
3576                        provider_state_summary: None,
3577                        state: StepState::Aborted,
3578                        output: None,
3579                        localized: Vec::new(),
3580                        localization_gaps: Vec::new(),
3581                    },
3582                )?;
3583                return Ok(Ctl::Abort);
3584            }
3585            // Suspension/blocking leaves the frame live: resume falls back
3586            // into the exact position (07 §4.6), never restarts the frame.
3587            other => return Ok(other),
3588        }
3589
3590        // Outbound gate: declared outputs evaluated in the *callee* scope,
3591        // schema-validated, snapshotted (07 §1.1). Only a completed body
3592        // has outputs; a halted callee pops without them.
3593        let outputs = if matches!(ctl, Ctl::Continue) {
3594            match self.call_outputs_gate(callee, &callee_frame) {
3595                Ok(outputs) => Some(outputs),
3596                Err(message) => {
3597                    self.append(
3598                        &call_path,
3599                        &RunLogPayload::CallFramePopped { outputs: None },
3600                    )?;
3601                    return self
3602                        .settle_error(
3603                            frame,
3604                            &call_path,
3605                            &step_id,
3606                            VerdictStatus::Fail,
3607                            format!("callee outputs failed the outbound gate: {message}"),
3608                        )
3609                        .await;
3610                }
3611            }
3612        } else {
3613            None
3614        };
3615        self.append(
3616            &call_path,
3617            &RunLogPayload::CallFramePopped {
3618                outputs: outputs.clone(),
3619            },
3620        )?;
3621
3622        // The call step's verdict *is* the callee's flow verdict
3623        // (spine §6.3); `degraded` propagates verbatim and participates in
3624        // the caller's fold.
3625        let callee_verdict = fold_flow_verdict(&callee_frame.fold, callee.verdict_policy);
3626        let mut status = None;
3627        let mut verdict_seq = None;
3628        if let Some(folded) = callee_verdict {
3629            let folded = FoldedVerdict {
3630                status: folded.status,
3631                degraded: folded.degraded,
3632                summary: format!(
3633                    "callee '{}' flow verdict: {}",
3634                    callee.flow_id, folded.summary
3635                ),
3636            };
3637            let seq = self
3638                .record_step_verdict(
3639                    &call_path,
3640                    &folded,
3641                    Vec::new(),
3642                    None,
3643                    EvidenceManifest::default(),
3644                )
3645                .await?;
3646            verdict_seq = Some(seq);
3647            frame.seed_verdict(&step_id, folded.status, folded.degraded);
3648            status = Some(folded.status);
3649        }
3650
3651        // Handler consultation on the call step's own verdict (the callee
3652        // handled its internal failures itself; this hook is the caller's
3653        // policy about the aggregate). Re-invocation dispositions (retry /
3654        // repair-then-re-call) need the 07 §1 attempt-framed full re-call
3655        // — a typed M2 refusal, registered.
3656        if matches!(
3657            status,
3658            Some(VerdictStatus::Fail) | Some(VerdictStatus::Unknown)
3659        ) {
3660            let hook = if status == Some(VerdictStatus::Fail) {
3661                HandlerHook::OnFail
3662            } else {
3663                HandlerHook::OnUnknown
3664            };
3665            match self
3666                .consult_hook(frame, &call_path, step.base.handlers.as_deref(), hook, None)
3667                .await?
3668            {
3669                Consulted::None | Consulted::RepairFailed => {}
3670                Consulted::Continue => {
3671                    self.append(
3672                        &call_path,
3673                        &RunLogPayload::StepExited {
3674                            provider_state_summary: None,
3675                            state: StepState::Judged,
3676                            output: outputs.clone(),
3677                            localized: Vec::new(),
3678                            localization_gaps: Vec::new(),
3679                        },
3680                    )?;
3681                    if let Some(outputs) = outputs {
3682                        frame.outputs.insert(step_id.as_str().to_owned(), outputs);
3683                    }
3684                    return Ok(Ctl::Continue);
3685                }
3686                Consulted::Abort => {
3687                    self.append(
3688                        &call_path,
3689                        &RunLogPayload::StepExited {
3690                            provider_state_summary: None,
3691                            state: StepState::Aborted,
3692                            output: None,
3693                            localized: Vec::new(),
3694                            localization_gaps: Vec::new(),
3695                        },
3696                    )?;
3697                    return Ok(Ctl::Abort);
3698                }
3699                Consulted::Escalated {
3700                    status: ruled,
3701                    summary: ruled_summary,
3702                    evidence: ruling_evidence,
3703                } => {
3704                    let folded = FoldedVerdict {
3705                        status: ruled,
3706                        degraded: false,
3707                        summary: ruled_summary,
3708                    };
3709                    let supersedes = verdict_seq.map(|seq| format!("seq:{seq}"));
3710                    let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
3711                    let ruled_seq = self
3712                        .record_step_verdict(&call_path, &folded, cited, supersedes, manifest)
3713                        .await?;
3714                    self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
3715                    frame.reseed_last(&step_id, ruled, false);
3716                    self.append(
3717                        &call_path,
3718                        &RunLogPayload::StepExited {
3719                            provider_state_summary: None,
3720                            state: StepState::Judged,
3721                            output: outputs.clone(),
3722                            localized: Vec::new(),
3723                            localization_gaps: Vec::new(),
3724                        },
3725                    )?;
3726                    if let Some(outputs) = outputs {
3727                        frame.outputs.insert(step_id.as_str().to_owned(), outputs);
3728                    }
3729                    return Ok(if ruled == VerdictStatus::Fail {
3730                        Ctl::HaltFail
3731                    } else {
3732                        Ctl::Continue
3733                    });
3734                }
3735                Consulted::Retry(_) | Consulted::Repaired | Consulted::RepairDone => {
3736                    return Err(RunnerError::M0Unsupported {
3737                        detail: format!(
3738                            "handler re-invocation dispositions on call step '{step_id}' need the 07 §1 \
3739                             attempt-framed full re-call — not in the M2 subset \
3740                             (escalate/continue/abort are supported)"
3741                        ),
3742                    });
3743                }
3744                Consulted::Pending(pending) => {
3745                    return Ok(Ctl::AwaitHuman(pending));
3746                }
3747                Consulted::Propagate(inner) => return Ok(inner),
3748            }
3749        }
3750
3751        self.append(
3752            &call_path,
3753            &RunLogPayload::StepExited {
3754                provider_state_summary: None,
3755                state: StepState::Judged,
3756                output: outputs.clone(),
3757                localized: Vec::new(),
3758                localization_gaps: Vec::new(),
3759            },
3760        )?;
3761        if let Some(outputs) = outputs {
3762            frame.outputs.insert(step_id.as_str().to_owned(), outputs);
3763        }
3764        if matches!(ctl, Ctl::HaltFail) || status == Some(VerdictStatus::Fail) {
3765            Ok(Ctl::HaltFail)
3766        } else {
3767            Ok(Ctl::Continue)
3768        }
3769    }
3770
3771    /// The outbound gate: callee `outputs` declarations evaluated over the
3772    /// callee frame's scope and validated against their schemas.
3773    fn call_outputs_gate(
3774        &self,
3775        callee: &FlowIR,
3776        callee_frame: &FrameState<'a>,
3777    ) -> Result<Value, String> {
3778        let scope = callee_frame.scope(&self.env, None);
3779        let mut outputs = Map::new();
3780        for decl in &callee.outputs {
3781            let value = pointlock_expr::eval(&decl.from, &scope)
3782                .map_err(|error| format!("output '{}' evaluation failed: {error}", decl.name))?;
3783            jsonschema::validate(decl.schema.as_value(), &value)
3784                .map_err(|error| format!("output '{}' failed its schema: {error}", decl.name))?;
3785            outputs.insert(decl.name.as_str().to_owned(), value);
3786        }
3787        Ok(Value::Object(outputs))
3788    }
3789
3790    // ─── if steps ───────────────────────────────────────────────────────────
3791
3792    async fn exec_if(
3793        &mut self,
3794        frame: &mut FrameState<'a>,
3795        step_path: RunPath,
3796        step: &'a IfStepIR,
3797    ) -> Result<Ctl, RunnerError> {
3798        let step_id = step.base.step_id.clone();
3799        let key = instance_key(&step_path);
3800        // A suspension-opened span reuses the archived branch decision —
3801        // the snapshot rule (spine §6.6) applies to control values too.
3802        let cond_value = match self.open_span_inputs(&key) {
3803            Some(archived) => archived.get("cond").cloned().unwrap_or(Value::Null),
3804            None => {
3805                let scope = frame.scope(&self.env, None);
3806                match pointlock_expr::eval(&step.cond, &scope) {
3807                    Ok(value) => value,
3808                    Err(error) => {
3809                        self.enter_step(&step_path, &step.base, Value::Null)?;
3810                        return self
3811                            .settle_error(
3812                                frame,
3813                                &step_path,
3814                                &step_id,
3815                                VerdictStatus::Fail,
3816                                format!("if cond evaluation failed: {error}"),
3817                            )
3818                            .await;
3819                    }
3820                }
3821            }
3822        };
3823        // Strict boolean (02 §4.5): anything else is a compiler/expression
3824        // bug signal — step fail, no branch is taken.
3825        let Some(cond) = cond_value.as_bool() else {
3826            self.enter_step(
3827                &step_path,
3828                &step.base,
3829                serde_json::json!({ "cond": cond_value }),
3830            )?;
3831            return self
3832                .settle_error(
3833                    frame,
3834                    &step_path,
3835                    &step_id,
3836                    VerdictStatus::Fail,
3837                    format!("if cond evaluated to a non-boolean value: {cond_value}"),
3838                )
3839                .await;
3840        };
3841        self.enter_step(&step_path, &step.base, serde_json::json!({ "cond": cond }))?;
3842
3843        if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
3844            return Ok(ctl);
3845        }
3846
3847        let empty: &'a [StepIR] = &[];
3848        let (selected, unselected): (&'a [StepIR], &'a [StepIR]) = if cond {
3849            (&step.then, step.r#else.as_deref().unwrap_or(empty))
3850        } else {
3851            (step.r#else.as_deref().unwrap_or(empty), &step.then)
3852        };
3853        // The unselected branch's steps each leave an
3854        // entered(null)/exited(skipped) pair — ledger completeness (the
3855        // blocked precedent). Recorded before the selected branch runs so
3856        // a mid-branch suspension leaves a complete account.
3857        self.stash_open_span_summaries().await;
3858        self.record_pairs(&step_path, unselected, StepState::Skipped)?;
3859
3860        let ctl = self
3861            .exec_body(frame, step_path.clone(), selected, 0)
3862            .await?;
3863        match ctl {
3864            Ctl::Continue | Ctl::HaltFail => {
3865                // Containers yield no verdict of their own (R4): the exit
3866                // closes the span; child verdicts already folded.
3867                self.append(
3868                    &step_path,
3869                    &RunLogPayload::StepExited {
3870                        provider_state_summary: None,
3871                        state: StepState::Judged,
3872                        output: None,
3873                        localized: Vec::new(),
3874                        localization_gaps: Vec::new(),
3875                    },
3876                )?;
3877                Ok(ctl)
3878            }
3879            Ctl::Abort => {
3880                self.append(
3881                    &step_path,
3882                    &RunLogPayload::StepExited {
3883                        provider_state_summary: None,
3884                        state: StepState::Aborted,
3885                        output: None,
3886                        localized: Vec::new(),
3887                        localization_gaps: Vec::new(),
3888                    },
3889                )?;
3890                Ok(Ctl::Abort)
3891            }
3892            other => Ok(other),
3893        }
3894    }
3895
3896    // ─── foreach steps ──────────────────────────────────────────────────────
3897
3898    async fn exec_foreach(
3899        &mut self,
3900        frame: &mut FrameState<'a>,
3901        step_path: RunPath,
3902        step: &'a ForeachStepIR,
3903    ) -> Result<Ctl, RunnerError> {
3904        let step_id = step.base.step_id.clone();
3905        let key = instance_key(&step_path);
3906        let items_value = match self.open_span_inputs(&key) {
3907            Some(archived) => archived.get("items").cloned().unwrap_or(Value::Null),
3908            None => {
3909                let scope = frame.scope(&self.env, None);
3910                match pointlock_expr::eval(&step.items, &scope) {
3911                    Ok(value) => value,
3912                    Err(error) => {
3913                        self.enter_step(&step_path, &step.base, Value::Null)?;
3914                        return self
3915                            .settle_error(
3916                                frame,
3917                                &step_path,
3918                                &step_id,
3919                                VerdictStatus::Fail,
3920                                format!("foreach items evaluation failed: {error}"),
3921                            )
3922                            .await;
3923                    }
3924                }
3925            }
3926        };
3927        let Some(items) = items_value.as_array().cloned() else {
3928            self.enter_step(
3929                &step_path,
3930                &step.base,
3931                serde_json::json!({ "items": items_value, "as": step.r#as.as_str() }),
3932            )?;
3933            return self
3934                .settle_error(
3935                    frame,
3936                    &step_path,
3937                    &step_id,
3938                    VerdictStatus::Fail,
3939                    format!("foreach items evaluated to a non-array value: {items_value}"),
3940                )
3941                .await;
3942        };
3943        // The snapshot carries `{ items, as }`: the position authority for
3944        // the positional (index-keyed) resume regime and the fold's
3945        // IterState carrier.
3946        self.enter_step(
3947            &step_path,
3948            &step.base,
3949            serde_json::json!({ "items": items, "as": step.r#as.as_str() }),
3950        )?;
3951
3952        if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
3953            return Ok(ctl);
3954        }
3955
3956        for (index, item) in items.iter().enumerate() {
3957            frame
3958                .iters
3959                .push((step.r#as.as_str().to_owned(), item.clone()));
3960            let mut iter_prefix = step_path.clone();
3961            iter_prefix.push(PathFrame::Iteration {
3962                index: index as u64,
3963                key: None,
3964            });
3965            let ctl = self.exec_body(frame, iter_prefix, &step.body, 0).await?;
3966            frame.iters.pop();
3967            match ctl {
3968                Ctl::Continue => {}
3969                Ctl::HaltFail => {
3970                    // The failing iteration already blocked its own tail;
3971                    // later iterations never materialize as instances.
3972                    self.append(
3973                        &step_path,
3974                        &RunLogPayload::StepExited {
3975                            provider_state_summary: None,
3976                            state: StepState::Judged,
3977                            output: None,
3978                            localized: Vec::new(),
3979                            localization_gaps: Vec::new(),
3980                        },
3981                    )?;
3982                    return Ok(Ctl::HaltFail);
3983                }
3984                Ctl::Abort => {
3985                    self.append(
3986                        &step_path,
3987                        &RunLogPayload::StepExited {
3988                            provider_state_summary: None,
3989                            state: StepState::Aborted,
3990                            output: None,
3991                            localized: Vec::new(),
3992                            localization_gaps: Vec::new(),
3993                        },
3994                    )?;
3995                    return Ok(Ctl::Abort);
3996                }
3997                other => return Ok(other),
3998            }
3999        }
4000        self.append(
4001            &step_path,
4002            &RunLogPayload::StepExited {
4003                provider_state_summary: None,
4004                state: StepState::Judged,
4005                output: None,
4006                localized: Vec::new(),
4007                localization_gaps: Vec::new(),
4008            },
4009        )?;
4010        Ok(Ctl::Continue)
4011    }
4012
4013    // ─── let steps ──────────────────────────────────────────────────────────
4014
4015    async fn exec_let(
4016        &mut self,
4017        frame: &mut FrameState<'a>,
4018        step_path: RunPath,
4019        step: &'a LetStepIR,
4020    ) -> Result<Ctl, RunnerError> {
4021        let step_id = step.base.step_id.clone();
4022        let key = instance_key(&step_path);
4023        let evaluated = match self.open_span_inputs(&key) {
4024            // The archived snapshot *is* the bindings product (pure,
4025            // deterministic) — never re-evaluated on resume.
4026            Some(Value::Object(map)) => map,
4027            Some(_) | None => {
4028                let scope = frame.scope(&self.env, None);
4029                let mut evaluated = Map::new();
4030                let mut error = None;
4031                for (name, expr) in step.bindings.iter() {
4032                    // SSA single assignment: rebinding is refused by the
4033                    // compiler check phase; this is the runtime defense
4034                    // line against hand-built IR.
4035                    if frame.vars.contains_key(name.as_str()) {
4036                        error = Some(format!(
4037                            "binding '{name}' rebinds an existing var (SSA single assignment)"
4038                        ));
4039                        break;
4040                    }
4041                    match pointlock_expr::eval(expr, &scope) {
4042                        Ok(value) => {
4043                            evaluated.insert(name.as_str().to_owned(), value);
4044                        }
4045                        Err(eval_error) => {
4046                            error =
4047                                Some(format!("binding '{name}' evaluation failed: {eval_error}"));
4048                            break;
4049                        }
4050                    }
4051                }
4052                if let Some(message) = error {
4053                    self.enter_step(&step_path, &step.base, Value::Null)?;
4054                    return self
4055                        .settle_error(
4056                            frame,
4057                            &step_path,
4058                            &step_id,
4059                            VerdictStatus::Fail,
4060                            format!("let bindings failed: {message}"),
4061                        )
4062                        .await;
4063                }
4064                evaluated
4065            }
4066        };
4067        // The ready snapshot carries the evaluated bindings — the resume
4068        // walk re-seeds `vars.*` from exactly this carrier.
4069        self.enter_step(&step_path, &step.base, Value::Object(evaluated.clone()))?;
4070        if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
4071            return Ok(ctl);
4072        }
4073        self.append(
4074            &step_path,
4075            &RunLogPayload::StepExited {
4076                provider_state_summary: None,
4077                state: StepState::Judged,
4078                output: None,
4079                localized: Vec::new(),
4080                localization_gaps: Vec::new(),
4081            },
4082        )?;
4083        for (name, value) in evaluated {
4084            frame.vars.insert(name, value);
4085        }
4086        Ok(Ctl::Continue)
4087    }
4088
4089    // ─── assert steps ───────────────────────────────────────────────────────
4090
4091    async fn exec_assert(
4092        &mut self,
4093        frame: &mut FrameState<'a>,
4094        step_path: RunPath,
4095        step: &'a AssertStepIR,
4096    ) -> Result<Ctl, RunnerError> {
4097        let step_id = step.base.step_id.clone();
4098        // An assert step resolves no input expressions; the span still
4099        // opens with an explicitly-null snapshot.
4100        self.enter_step(&step_path, &step.base, Value::Null)?;
4101        if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
4102            return Ok(ctl);
4103        }
4104        let needs = VerifyNeeds::of(&step.assertions);
4105        let mut last_verdict_seq: Option<u64> = None;
4106        let mut seeded = false;
4107        let mut active_retry: Option<(RetryPolicy, u32)> = None;
4108        loop {
4109            let material = match &step.observe {
4110                // Fresh capture: one `session.observe`, localized through the
4111                // same observing pipeline as action observations.
4112                ObservationSource::Fresh(_) => {
4113                    let mut anchor = step_path.clone();
4114                    anchor.push(PathFrame::Phase {
4115                        phase: Phase::Observe,
4116                    });
4117                    self.fresh_material(&needs, &anchor).await?
4118                }
4119                // Archive reuse: the referenced action step's localized
4120                // observation material — zero device I/O, offline
4121                // re-judgeable by construction.
4122                ObservationSource::FromStep(from) => {
4123                    let which = from.which;
4124                    match frame.observed.get(from.from_step.as_str()) {
4125                        None => ObserveMaterial::absent(&format!(
4126                            "step '{}' has no archived observation in this frame",
4127                            from.from_step
4128                        )),
4129                        Some(observed) => {
4130                            let wanted = match which {
4131                                pointlock_ir::ObservationWhich::After => &observed.after_id,
4132                                pointlock_ir::ObservationWhich::Before => &observed.before_id,
4133                            };
4134                            match wanted {
4135                                None => ObserveMaterial::absent(&format!(
4136                                    "step '{}' recorded no {:?} observation",
4137                                    from.from_step, which
4138                                )),
4139                                Some(observation_id) => {
4140                                    match observed
4141                                        .observations
4142                                        .iter()
4143                                        .find(|record| record.observation_id == *observation_id)
4144                                    {
4145                                        None => ObserveMaterial::absent(
4146                                            "the referenced observation was never localized",
4147                                        ),
4148                                        Some(record) => {
4149                                            material_from_observation(self.store, record)
4150                                        }
4151                                    }
4152                                }
4153                            }
4154                        }
4155                    }
4156                }
4157            };
4158            let scope = frame.scope(&self.env, None);
4159            let mut outcomes = Vec::with_capacity(step.assertions.len());
4160            let mut degraded_verify = false;
4161            for assertion in &step.assertions {
4162                let evaluated = match &assertion.predicate {
4163                    PredicateIR::Expr { expr } => EvaluatedAssertion {
4164                        record: eval_expr_assertion(assertion, expr, &scope),
4165                        degraded_verify: false,
4166                    },
4167                    _ => {
4168                        eval_observed_assertion(assertion, &material, self.vision.as_deref()).await
4169                    }
4170                };
4171                degraded_verify |= evaluated.degraded_verify;
4172                outcomes.push(evaluated.record);
4173            }
4174            for outcome in &outcomes {
4175                let mut path = step_path.clone();
4176                path.push(PathFrame::Phase {
4177                    phase: Phase::Assert,
4178                });
4179                path.push(PathFrame::Assertion {
4180                    assert_id: outcome.assert_id.clone(),
4181                });
4182                self.append(
4183                    &path,
4184                    &RunLogPayload::AssertionEvaluated {
4185                        outcome: outcome.clone(),
4186                    },
4187                )?;
4188            }
4189            // Assert steps declare ≥ 1 assertion, so a verdict always folds.
4190            let folded =
4191                fold_step_verdict(&outcomes, false, degraded_verify, frame.flow.verdict_policy);
4192            let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
4193            let seq = self
4194                .record_step_verdict(
4195                    &step_path,
4196                    &folded,
4197                    Vec::new(),
4198                    supersedes,
4199                    EvidenceManifest::default(),
4200                )
4201                .await?;
4202            last_verdict_seq = Some(seq);
4203            if seeded {
4204                frame.reseed_last(&step_id, folded.status, folded.degraded);
4205            } else {
4206                frame.seed_verdict(&step_id, folded.status, folded.degraded);
4207                seeded = true;
4208            }
4209            if folded.status == VerdictStatus::Pass {
4210                self.append(
4211                    &step_path,
4212                    &RunLogPayload::StepExited {
4213                        provider_state_summary: None,
4214                        state: StepState::Judged,
4215                        output: None,
4216                        localized: Vec::new(),
4217                        localization_gaps: Vec::new(),
4218                    },
4219                )?;
4220                return Ok(Ctl::Continue);
4221            }
4222
4223            // In-force handler retry budget (observe + assert re-entry is
4224            // readonly by construction — always replay-safe).
4225            if let Some((policy, used)) = active_retry.take()
4226                && used < policy.max_attempts
4227            {
4228                self.backoff_policy(&policy, used).await;
4229                active_retry = Some((policy, used + 1));
4230                continue;
4231            }
4232
4233            let hook = if folded.status == VerdictStatus::Fail {
4234                HandlerHook::OnFail
4235            } else {
4236                HandlerHook::OnUnknown
4237            };
4238            match self
4239                .consult_hook(frame, &step_path, step.base.handlers.as_deref(), hook, None)
4240                .await?
4241            {
4242                Consulted::None | Consulted::RepairFailed => {
4243                    self.append(
4244                        &step_path,
4245                        &RunLogPayload::StepExited {
4246                            provider_state_summary: None,
4247                            state: StepState::Judged,
4248                            output: None,
4249                            localized: Vec::new(),
4250                            localization_gaps: Vec::new(),
4251                        },
4252                    )?;
4253                    return Ok(if folded.status == VerdictStatus::Fail {
4254                        Ctl::HaltFail
4255                    } else {
4256                        Ctl::Continue
4257                    });
4258                }
4259                Consulted::Continue => {
4260                    self.append(
4261                        &step_path,
4262                        &RunLogPayload::StepExited {
4263                            provider_state_summary: None,
4264                            state: StepState::Judged,
4265                            output: None,
4266                            localized: Vec::new(),
4267                            localization_gaps: Vec::new(),
4268                        },
4269                    )?;
4270                    return Ok(Ctl::Continue);
4271                }
4272                Consulted::Abort => {
4273                    self.append(
4274                        &step_path,
4275                        &RunLogPayload::StepExited {
4276                            provider_state_summary: None,
4277                            state: StepState::Aborted,
4278                            output: None,
4279                            localized: Vec::new(),
4280                            localization_gaps: Vec::new(),
4281                        },
4282                    )?;
4283                    return Ok(Ctl::Abort);
4284                }
4285                Consulted::Escalated {
4286                    status: ruled,
4287                    summary: ruled_summary,
4288                    evidence: ruling_evidence,
4289                } => {
4290                    let folded = FoldedVerdict {
4291                        status: ruled,
4292                        degraded: false,
4293                        summary: ruled_summary,
4294                    };
4295                    let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
4296                    let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
4297                    let ruled_seq = self
4298                        .record_step_verdict(&step_path, &folded, cited, supersedes, manifest)
4299                        .await?;
4300                    self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
4301                    frame.reseed_last(&step_id, ruled, false);
4302                    self.append(
4303                        &step_path,
4304                        &RunLogPayload::StepExited {
4305                            provider_state_summary: None,
4306                            state: StepState::Judged,
4307                            output: None,
4308                            localized: Vec::new(),
4309                            localization_gaps: Vec::new(),
4310                        },
4311                    )?;
4312                    return Ok(if ruled == VerdictStatus::Fail {
4313                        Ctl::HaltFail
4314                    } else {
4315                        Ctl::Continue
4316                    });
4317                }
4318                Consulted::Retry(policy) => {
4319                    self.backoff_policy(&policy, 0).await;
4320                    active_retry = Some((policy, 1));
4321                }
4322                Consulted::Repaired | Consulted::RepairDone => {}
4323                Consulted::Pending(pending) => {
4324                    return Ok(Ctl::AwaitHuman(pending));
4325                }
4326                Consulted::Propagate(ctl) => return Ok(ctl),
4327            }
4328            // Loop: re-observe and re-evaluate (fresh material each round;
4329            // fromStep archives are stable, so a re-round only makes sense
4330            // after a repair — both are honest re-judgements on new/declared
4331            // world state).
4332        }
4333    }
4334
4335    // ─── observing / localization ───────────────────────────────────────────
4336
4337    /// The observing phase: localize evidence (spine §6.6 — provider-side
4338    /// retention is not guaranteed) and record the before/after
4339    /// observations. Returns the provider asset refs cited by the verdict,
4340    /// the [`ObserveMaterial`] of the *after* observation, and the
4341    /// localized records (the `fromStep` archive of this step).
4342    async fn observing(
4343        &mut self,
4344        step: &'a ActionStepIR,
4345        step_path: &RunPath,
4346        attempt_n: u64,
4347        result: &ActionResult,
4348        settled_seq: u64,
4349    ) -> Result<(Vec<AssetRef>, ObserveMaterial, StepObs, EvidenceManifest), RunnerError> {
4350        let mut observe_path = step_path.clone();
4351        observe_path.push(PathFrame::Attempt { n: attempt_n });
4352        observe_path.push(PathFrame::Phase {
4353            phase: Phase::Observe,
4354        });
4355        let needs = VerifyNeeds::of(&step.assertions);
4356        let mut cited = Vec::new();
4357        let mut material =
4358            ObserveMaterial::absent("the action result carries no after observation");
4359        let mut observed = StepObs {
4360            observations: Vec::new(),
4361            before_id: None,
4362            after_id: None,
4363        };
4364        if let Some(observation) = &result.before {
4365            let record = self
4366                .localize_observation(observation, &mut cited, None)
4367                .await?;
4368            // file-before-row-before-log: the bytes are on disk and indexed
4369            // before this event references them.
4370            self.append(
4371                &observe_path,
4372                &RunLogPayload::ObservationRecorded {
4373                    observation: record.clone(),
4374                },
4375            )?;
4376            observed.before_id = Some(record.observation_id.clone());
4377            observed.observations.push(record);
4378        }
4379        if let Some(observation) = &result.after {
4380            material = ObserveMaterial::default();
4381            let record = self
4382                .localize_observation(observation, &mut cited, Some((&needs, &mut material)))
4383                .await?;
4384            self.append(
4385                &observe_path,
4386                &RunLogPayload::ObservationRecorded {
4387                    observation: record.clone(),
4388                },
4389            )?;
4390            observed.after_id = Some(record.observation_id.clone());
4391            observed.observations.push(record);
4392        }
4393        let mut manifest = EvidenceManifest::default();
4394        for asset in &result.evidence {
4395            // Bounded manifest (the cited-list cap's sibling): entries
4396            // beyond the cap are recorded as typed gaps — bounded DTOs
4397            // must not silently truncate (spine §10 bounded-render
4398            // discipline).
4399            if manifest.localized.len() >= VERDICT_EVIDENCE_MAX_ENTRIES {
4400                manifest.gaps.push(pointlock_ir::EvidenceGap {
4401                    asset: asset.clone(),
4402                    reason: format!(
4403                        "evidence cap exceeded ({VERDICT_EVIDENCE_MAX_ENTRIES} max per judgment)"
4404                    ),
4405                });
4406                continue;
4407            }
4408            // Auxiliary settlement evidence (item ③, 2026-07-18): a
4409            // success joins this judgment's localized manifest (and the
4410            // evidence_ref index); a failure is a TYPED gap on the
4411            // verdict record — never a silent omission (principle 4/R4).
4412            match self.try_localize(asset).await? {
4413                Ok((evidence, _bytes)) => {
4414                    self.store.link_evidence(
4415                        &self.run_id,
4416                        settled_seq,
4417                        &asset.id,
4418                        &evidence.sha256,
4419                    )?;
4420                    cited.push(asset.clone());
4421                    manifest.localized.push(evidence);
4422                }
4423                Err(reason) => {
4424                    manifest.gaps.push(pointlock_ir::EvidenceGap {
4425                        asset: asset.clone(),
4426                        reason,
4427                    });
4428                }
4429            }
4430        }
4431        Ok((cited, material, observed, manifest))
4432    }
4433
4434    /// Localizes one observation's evidence and builds its durable record.
4435    /// Omissions are typed data and pass through verbatim; a localization
4436    /// failure leaves the affected field absent and feeds the dependent
4437    /// verify channel a typed gap — the run is never aborted over it (M2
4438    /// degradation rule; principle 4 routes it to `unknown`).
4439    async fn localize_observation(
4440        &mut self,
4441        observation: &Observation,
4442        cited: &mut Vec<AssetRef>,
4443        mut material: Option<(&VerifyNeeds, &mut ObserveMaterial)>,
4444    ) -> Result<ObservationRecord, RunnerError> {
4445        let screenshot = match &observation.screenshot {
4446            Some(asset) => match self.try_localize(asset).await? {
4447                Ok((evidence, bytes)) => {
4448                    cited.push(asset.clone());
4449                    if let Some((needs, material)) = material.as_mut()
4450                        && needs.vision
4451                    {
4452                        material.screenshot = Some((bytes, asset.media_type.clone()));
4453                    }
4454                    Some(evidence)
4455                }
4456                Err(gap) => {
4457                    if let Some((needs, material)) = material.as_mut()
4458                        && needs.vision
4459                    {
4460                        material.screenshot_gap = Some(gap);
4461                    }
4462                    None
4463                }
4464            },
4465            None => {
4466                if let Some((needs, material)) = material.as_mut()
4467                    && needs.vision
4468                {
4469                    material.screenshot_gap = Some(match observation.screenshot_omission {
4470                        Some(reason) => {
4471                            format!("screenshot omitted by the provider ({reason:?})")
4472                        }
4473                        None => "the observation carries no screenshot".to_owned(),
4474                    });
4475                }
4476                None
4477            }
4478        };
4479        let mut ui_snapshot_omission = observation.ui_snapshot_omission;
4480        let ui_snapshot = match &observation.ui_snapshot {
4481            Some(snapshot) => {
4482                let wants_tree = matches!(&material, Some((needs, _)) if needs.ui_tree);
4483                let localized = if wants_tree {
4484                    self.localize_ui_tree(
4485                        observation,
4486                        snapshot,
4487                        &mut material,
4488                        &mut ui_snapshot_omission,
4489                    )
4490                    .await?
4491                } else {
4492                    // No assertion consumes the tree: localize the
4493                    // provider-side evidence object as before (retention is
4494                    // not guaranteed), without the `ui.snapshot.get` pull.
4495                    // A failure is a gap, not an abort.
4496                    self.try_localize(&snapshot.evidence)
4497                        .await?
4498                        .ok()
4499                        .map(|(evidence, _bytes)| evidence)
4500                };
4501                // Cite only what actually landed (item ③ review fix): a
4502                // citation the local library cannot serve would be a
4503                // silent gallery omission. Citation granularity is the
4504                // ASSET (the pointer); the byte-exact truth (sha256 +
4505                // localPath of what was actually stored — the canonical
4506                // tree on the wants_tree path) lives on the record's
4507                // EvidenceRef, which is what consumers resolve.
4508                if localized.is_some() {
4509                    cited.push(snapshot.evidence.clone());
4510                }
4511                localized
4512            }
4513            None => {
4514                if let Some((needs, material)) = material.as_mut()
4515                    && needs.ui_tree
4516                {
4517                    material.ui_tree_gap = Some(match observation.ui_snapshot_omission {
4518                        Some(reason) => {
4519                            format!("uiSnapshot omitted by the provider ({reason:?})")
4520                        }
4521                        None => "the observation carries no uiSnapshot".to_owned(),
4522                    });
4523                }
4524                None
4525            }
4526        };
4527        // Finiteness guard: `scaleFactor` is the only f64 in the durable
4528        // record domain, and serde_json writes a non-finite f64 as `null`
4529        // — which round-trips into a permanent ledger read failure (every
4530        // refold/verify/projection of the run errors). A provider that
4531        // reports a non-finite viewport gets the field honestly absent
4532        // rather than a poisoned ledger (clamping would falsify evidence;
4533        // aborting the run over a cosmetic field would violate the M2
4534        // degradation rule).
4535        let viewport = observation
4536            .viewport
4537            .scale_factor
4538            .is_finite()
4539            .then(|| observation.viewport.clone());
4540        Ok(ObservationRecord {
4541            observation_id: observation.id.clone(),
4542            captured_at_ms: observation.captured_at_ms,
4543            viewport,
4544            screenshot,
4545            screenshot_omission: observation.screenshot_omission,
4546            ui_snapshot,
4547            ui_snapshot_omission,
4548        })
4549    }
4550
4551    /// Pulls the observation's normalized UI tree (`ui.snapshot.get`),
4552    /// localizes the canonical bytes, and feeds the verify-chain material.
4553    /// A typed `Unavailable` — or a provider error on the dereference — is
4554    /// data, not a run abort: it becomes the uiTree channel's gap (unknown
4555    /// propagation), never a fabricated tree.
4556    async fn localize_ui_tree(
4557        &mut self,
4558        observation: &Observation,
4559        snapshot: &pointlock_ir::UiSnapshotRef,
4560        material: &mut Option<(&VerifyNeeds, &mut ObserveMaterial)>,
4561        ui_snapshot_omission: &mut Option<UiSnapshotOmissionReason>,
4562    ) -> Result<Option<EvidenceRef>, RunnerError> {
4563        match self.session.ui_snapshot(&observation.id).await {
4564            Ok(UiSnapshotOutcome::Available { snapshot: tree }) => {
4565                let bytes =
4566                    serde_json::to_vec(&tree).expect("a serde_json::Value always serializes");
4567                let put = self.store.put_evidence(&bytes, "application/json")?;
4568                if let Some((_, material)) = material.as_mut() {
4569                    material.ui_tree = Some(bytes);
4570                }
4571                Ok(Some(EvidenceRef {
4572                    asset: snapshot.evidence.clone(),
4573                    sha256: put.sha256,
4574                    local_path: put.local_path,
4575                }))
4576            }
4577            Ok(UiSnapshotOutcome::Unavailable { reason }) => {
4578                *ui_snapshot_omission = Some(reason);
4579                if let Some((_, material)) = material.as_mut() {
4580                    material.ui_tree_gap =
4581                        Some(format!("uiSnapshot dereference unavailable ({reason:?})"));
4582                }
4583                Ok(None)
4584            }
4585            Err(error) => {
4586                if let Some((_, material)) = material.as_mut() {
4587                    material.ui_tree_gap = Some(format!("uiSnapshot dereference failed: {error}"));
4588                }
4589                Ok(None)
4590            }
4591        }
4592    }
4593
4594    /// Fetches an asset's bytes into the content-addressed evidence area.
4595    /// The outer `Result` is infrastructure (store I/O — still fatal); the
4596    /// inner one is the typed localization gap (fetch unsupported/ruptured,
4597    /// integrity mismatch — the run degrades, never aborts; 04 §4.3 is
4598    /// honored by *not using* mismatched bytes, with the reason on record).
4599    async fn try_localize(
4600        &mut self,
4601        asset: &AssetRef,
4602    ) -> Result<Result<(EvidenceRef, Vec<u8>), String>, RunnerError> {
4603        let mut stream = match self.session.fetch_evidence(asset).await {
4604            Ok(stream) => stream,
4605            Err(error) => {
4606                return Ok(Err(format!(
4607                    "evidence fetch failed for asset {}: {error}",
4608                    asset.id
4609                )));
4610            }
4611        };
4612        let mut bytes = Vec::new();
4613        while let Some(chunk) = stream.next().await {
4614            match chunk {
4615                Ok(part) => bytes.extend(part),
4616                Err(error) => {
4617                    return Ok(Err(format!(
4618                        "evidence stream failed for asset {}: {error}",
4619                        asset.id
4620                    )));
4621                }
4622            }
4623        }
4624        let put = self.store.put_evidence(&bytes, &asset.media_type)?;
4625        if let Some(expected) = &asset.sha256
4626            && expected != &put.sha256
4627        {
4628            return Ok(Err(format!(
4629                "evidence integrity failure for asset {}: sha256 {} != declared {expected}",
4630                asset.id, put.sha256
4631            )));
4632        }
4633        let evidence = EvidenceRef {
4634            asset: asset.clone(),
4635            sha256: put.sha256,
4636            local_path: put.local_path,
4637        };
4638        Ok(Ok((evidence, bytes)))
4639    }
4640
4641    /// Appends `verdictRecorded` and writes the verdict back through the
4642    /// provider (`verdict.record` — the daemon only validates and
4643    /// persists). Returns the `verdictRecorded` seq (evidence linking).
4644    async fn record_step_verdict(
4645        &mut self,
4646        path: &RunPath,
4647        folded: &FoldedVerdict,
4648        cited: Vec<AssetRef>,
4649        supersedes: Option<String>,
4650        manifest: EvidenceManifest,
4651    ) -> Result<u64, RunnerError> {
4652        let verdict = Verdict {
4653            status: folded.status,
4654            degraded: folded.degraded,
4655            // The local ledger keeps the FULL summary — the 16384-char
4656            // cap is a wire hard limit, applied at write-back only
4657            // (04 §5).
4658            summary: folded.summary.clone(),
4659            evidence: cited
4660                .into_iter()
4661                .take(VERDICT_EVIDENCE_MAX_ENTRIES)
4662                .collect(),
4663            supersedes,
4664        };
4665        // Failure-instant capture (07 §2.2): the verdict instant IS the
4666        // failure instant, so the capture runs FIRST — before the
4667        // write-back RPC below can delay it on a degraded daemon. The
4668        // profile rides the span's exit through the `append` attach; a
4669        // superseding pass discards it. Deliberate cost: a retry round
4670        // that will be superseded still pays a capture (bounded by the
4671        // 2s budget) — skipping it would need the consult outcome, which
4672        // is only known after the handler runs, and a wrong skip would
4673        // ship a summary-less fail exit. Correctness over the bounded
4674        // RPC.
4675        let key = instance_key(path);
4676        match verdict.status {
4677            VerdictStatus::Fail | VerdictStatus::Unknown => {
4678                let summary = self.capture_summary().await;
4679                self.pending_summaries.insert(key, summary);
4680            }
4681            VerdictStatus::Pass => {
4682                self.pending_summaries.remove(&key);
4683            }
4684        }
4685        // Remote archival before the append so its outcome can ride the
4686        // event; it is archival of an already-derived verdict, not a
4687        // world effect, so the actionIntent WAL discipline does not
4688        // apply. A failure never changes the local verdict and never
4689        // aborts the run — it is annotated here and surfaced by the
4690        // report (04 §5). Deliberate cost: on a hung daemon the
4691        // `verdictRecorded` `at_ms` trails the fold instant by the
4692        // bounded write-back budget — the price of carrying the outcome
4693        // on the event under the closed §6.1 vocabulary.
4694        let remote_archival_error = self.try_verdict_writeback(&verdict).await;
4695        let seq = self.append(
4696            path,
4697            &RunLogPayload::VerdictRecorded {
4698                verdict: verdict.clone(),
4699                localized: manifest.localized,
4700                localization_gaps: manifest.gaps,
4701                remote_archival_error,
4702            },
4703        )?;
4704        Ok(seq)
4705    }
4706
4707    /// `ProviderSession::record_verdict` write-back with the wire caps
4708    /// applied on the runner side (compaction is the runner's job,
4709    /// 04 §5). Returns the failure rendered for the ledger annotation —
4710    /// never an error: remote archival failure must not change the local
4711    /// verdict or abort the run (04 §5, the RunLog is the sole truth).
4712    async fn try_verdict_writeback(&mut self, verdict: &Verdict) -> Option<String> {
4713        self.session
4714            .record_verdict(VerdictWrite {
4715                status: verdict.status,
4716                summary: cap_wire_summary(verdict),
4717                evidence: verdict
4718                    .evidence
4719                    .iter()
4720                    .take(VERDICT_EVIDENCE_MAX_ENTRIES)
4721                    .cloned()
4722                    .collect(),
4723            })
4724            .await
4725            .err()
4726            .map(|error| format!("remote archival failed: {error}"))
4727    }
4728
4729    /// Best-effort session teardown (04 §2.1: `end` must not block the
4730    /// runner's teardown when the transport is already gone).
4731    async fn end_session(&mut self, outcome: SessionOutcome) {
4732        let _ = self.session.end(outcome, None).await;
4733    }
4734}
4735
4736/// The inbound gate of a call step (07 §1.1): apply the callee's declared
4737/// param defaults, refuse undeclared inputs and missing required params,
4738/// and validate every present value against its `ParamDecl.schema`.
4739fn call_inputs_gate(
4740    callee: &FlowIR,
4741    inputs: Map<String, Value>,
4742) -> Result<Map<String, Value>, String> {
4743    for key in inputs.keys() {
4744        if !callee
4745            .params
4746            .iter()
4747            .any(|decl: &ParamDecl| decl.name.as_str() == key)
4748        {
4749            return Err(format!(
4750                "input '{key}' is not a declared param of callee '{}'",
4751                callee.flow_id
4752            ));
4753        }
4754    }
4755    let gated =
4756        params_with_defaults(callee, Value::Object(inputs)).map_err(|error| error.to_string())?;
4757    for decl in &callee.params {
4758        if let Some(value) = gated.get(decl.name.as_str()) {
4759            jsonschema::validate(decl.schema.as_value(), value)
4760                .map_err(|error| format!("param '{}' failed its schema: {error}", decl.name))?;
4761        }
4762    }
4763    Ok(gated)
4764}
4765
4766/// Applies declared param defaults over the supplied params/inputs;
4767/// missing required params without defaults are refused. Shared by the run
4768/// entry (run params) and the call step's inbound gate (07 §1.1).
4769pub(crate) fn params_with_defaults(
4770    flow: &FlowIR,
4771    params: Value,
4772) -> Result<Map<String, Value>, RunnerError> {
4773    let mut map = match params {
4774        Value::Object(map) => map,
4775        Value::Null => Map::new(),
4776        other => {
4777            return Err(RunnerError::InvalidParams {
4778                reason: format!("params must be a JSON object or null, got {other}"),
4779            });
4780        }
4781    };
4782    for decl in &flow.params {
4783        if map.contains_key(decl.name.as_str()) {
4784            continue;
4785        }
4786        if let Some(default) = &decl.default {
4787            map.insert(decl.name.as_str().to_owned(), default.clone());
4788        } else if decl.required {
4789            return Err(RunnerError::InvalidParams {
4790                reason: format!(
4791                    "required param '{}' is missing and has no default",
4792                    decl.name
4793                ),
4794            });
4795        }
4796    }
4797    Ok(map)
4798}
4799
4800/// Which observation channels a set of assertions' verify chains consume —
4801/// decides what the observing phase must localize into
4802/// [`ObserveMaterial`] and what a fresh observe must want.
4803pub(crate) struct VerifyNeeds {
4804    /// Some assertion's chain contains `uiTree`.
4805    pub ui_tree: bool,
4806    /// Some assertion's chain contains `vision`.
4807    pub vision: bool,
4808}
4809
4810impl VerifyNeeds {
4811    /// Scans the assertions (expr predicates consume no channel).
4812    pub fn of(assertions: &[AssertionIR]) -> Self {
4813        let mut needs = VerifyNeeds {
4814            ui_tree: false,
4815            vision: false,
4816        };
4817        for assertion in assertions {
4818            if matches!(assertion.predicate, PredicateIR::Expr { .. }) {
4819                continue;
4820            }
4821            for channel in &assertion.verify_via {
4822                match channel {
4823                    VerifyChannel::UiTree => needs.ui_tree = true,
4824                    VerifyChannel::Vision => needs.vision = true,
4825                    VerifyChannel::Dom => {}
4826                }
4827            }
4828        }
4829        needs
4830    }
4831}
4832
4833/// Truncates a verdict summary to the provider wire cap (char-aware),
4834/// appending a content-hash pointer to the local full verdict when it
4835/// cuts (04 §5: the RunLog keeps the complete summary; the wire copy
4836/// points back at it).
4837pub(crate) fn cap_wire_summary(verdict: &Verdict) -> String {
4838    if verdict.summary.chars().count() <= VERDICT_SUMMARY_MAX_CHARS {
4839        return verdict.summary.clone();
4840    }
4841    let pointer = format!(
4842        " …[truncated; full local verdict {}]",
4843        pointlock_ir::domain_hash(
4844            "pointlock-runner/1/local-verdict",
4845            &serde_json::to_value(verdict).expect("a Verdict always serializes"),
4846        )
4847    );
4848    let keep = VERDICT_SUMMARY_MAX_CHARS.saturating_sub(pointer.chars().count());
4849    let mut capped: String = verdict.summary.chars().take(keep).collect();
4850    capped.push_str(&pointer);
4851    capped
4852}
4853
4854/// The backoff delay before retry number `tries + 1` (spine §3
4855/// `RetryPolicy.backoffMs`).
4856fn backoff_ms(policy: &pointlock_ir::RetryPolicy, tries: u32) -> u64 {
4857    match &policy.backoff_ms {
4858        pointlock_ir::BackoffMs::Fixed(number) => number.as_f64().unwrap_or(0.0) as u64,
4859        pointlock_ir::BackoffMs::Schedule(schedule) => {
4860            let initial = schedule.initial.as_f64().unwrap_or(0.0);
4861            let factor = schedule.factor.as_f64().unwrap_or(1.0);
4862            let max = schedule.max.as_f64().unwrap_or(f64::MAX);
4863            let exponent = tries.saturating_sub(1);
4864            (initial * factor.powi(exponent as i32)).min(max) as u64
4865        }
4866    }
4867}
4868
4869/// Whether an in-attempt retry is allowed (spine §6.5 mount point 1,
4870/// closed): only `action_failed_retryable`, `target_stale`, and — for
4871/// idempotent steps — `action_timed_out`, and only when the policy lists
4872/// the class and the budget is not exhausted.
4873fn retry_allowed(step: &ActionStepIR, class: ErrorClass, tries: u32) -> bool {
4874    let Some(policy) = &step.base.retry else {
4875        return false;
4876    };
4877    if tries >= policy.max_attempts {
4878        return false;
4879    }
4880    if !policy.retry_on.contains(&class) {
4881        return false;
4882    }
4883    match class {
4884        ErrorClass::ActionFailedRetryable | ErrorClass::TargetStale => true,
4885        ErrorClass::ActionTimedOut => step.idempotent,
4886        _ => false,
4887    }
4888}
4889
4890/// Maps a non-succeeded terminal onto the closed `ErrorClass` taxonomy
4891/// (spine §5). When the wire code spells a class verbatim it is adopted
4892/// (mirrors the store fold's best-effort rule); otherwise `failed` maps by
4893/// the daemon-declared `retryable` flag.
4894pub(crate) fn classify(outcome: &ActionOutcome) -> ErrorClass {
4895    match outcome {
4896        ActionOutcome::Succeeded { .. } => {
4897            unreachable!("classify is only called on non-succeeded terminals")
4898        }
4899        ActionOutcome::Failed { error } => code_spelled_class(&error.code).unwrap_or({
4900            if error.retryable {
4901                ErrorClass::ActionFailedRetryable
4902            } else {
4903                ErrorClass::ActionFailedFinal
4904            }
4905        }),
4906        ActionOutcome::TimedOut { .. } => ErrorClass::ActionTimedOut,
4907        ActionOutcome::Cancelled { .. } => ErrorClass::ActionCancelled,
4908    }
4909}
4910
4911fn code_spelled_class(code: &str) -> Option<ErrorClass> {
4912    serde_json::from_value(Value::String(code.to_owned())).ok()
4913}
4914
4915fn terminal_message(outcome: &ActionOutcome) -> String {
4916    let error: &ErrorInfo = match outcome {
4917        ActionOutcome::Failed { error }
4918        | ActionOutcome::Cancelled { error }
4919        | ActionOutcome::TimedOut { error } => error,
4920        ActionOutcome::Succeeded { .. } => {
4921            unreachable!("terminal_message is only called on non-succeeded terminals")
4922        }
4923    };
4924    format!("{} ({})", error.message, error.code)
4925}
4926
4927/// Whether the provider-reported execution mode is inside the attempt's
4928/// whitelist (§6.4 R-degrade). An absent execution report cannot be
4929/// audited and is accepted (the DeviceRail adapter always reports it).
4930fn execution_accepted(attempt: &BoundAttempt, execution: &Option<ActionExecution>) -> bool {
4931    match execution {
4932        None => true,
4933        Some(execution) => {
4934            let mode = match execution {
4935                ActionExecution::NativeSemantic { .. } => ExecutionMode::NativeSemantic,
4936                ActionExecution::WebSemantic { .. } => ExecutionMode::WebSemantic,
4937                ActionExecution::CoordinateFallback { .. } => ExecutionMode::CoordinateFallback,
4938            };
4939            attempt.accept_execution_modes.contains(&mode)
4940        }
4941    }
4942}
4943
4944/// Whether an action step is effectively mutating for the I2 replay gates
4945/// (mutating and not declared idempotent).
4946pub(crate) fn gated_mutating(step: &ActionStepIR) -> bool {
4947    step.effect == EffectClassAction::Mutating && !step.idempotent
4948}
4949
4950/// Whether an uncertain reconcile branch may replay the step
4951/// (07 §4.4: `idempotent: true` or `effect: "readonly"`).
4952pub(crate) fn replay_permitted(step: &ActionStepIR) -> bool {
4953    step.effect == EffectClassAction::Readonly || step.idempotent
4954}
4955
4956#[cfg(test)]
4957mod chain_start_tests {
4958    use super::*;
4959
4960    fn two_attempt_step() -> ActionStepIR {
4961        serde_json::from_value(serde_json::json!({
4962            "kind": "action",
4963            "stepId": "s1",
4964            "effectHash": format!("sha256:{}", "0".repeat(64)),
4965            "judgeHash": format!("sha256:{}", "0".repeat(64)),
4966            "checkpoint": true,
4967            "effect": "mutating",
4968            "idempotent": true,
4969            "binding": { "attempts": [ {
4970                "channel": "uiTree",
4971                "actionName": "tapElement",
4972                "args": {},
4973                "acceptExecutionModes": ["nativeSemantic"],
4974                "protection": "standard"
4975            }, {
4976                "channel": "uiTree",
4977                "actionName": "setElementValue",
4978                "args": {},
4979                "acceptExecutionModes": ["nativeSemantic"],
4980                "protection": "standard"
4981            } ] },
4982            "assertions": []
4983        }))
4984        .expect("fixture step")
4985    }
4986
4987    #[test]
4988    fn maps_recorded_positions_and_refuses_out_of_range() {
4989        let step = two_attempt_step();
4990        assert_eq!(chain_start(None, &step).expect("head"), 0);
4991        assert_eq!(chain_start(Some(1), &step).expect("first"), 0);
4992        assert_eq!(chain_start(Some(2), &step).expect("second"), 1);
4993        assert!(chain_start(Some(3), &step).is_err());
4994        assert!(chain_start(Some(0), &step).is_err());
4995    }
4996}
4997
4998#[cfg(test)]
4999mod tests {
5000    use super::*;
5001
5002    #[test]
5003    fn wire_summary_truncation_appends_the_local_verdict_pointer() {
5004        let verdict = Verdict {
5005            status: VerdictStatus::Fail,
5006            degraded: false,
5007            summary: "x".repeat(VERDICT_SUMMARY_MAX_CHARS + 100),
5008            evidence: Vec::new(),
5009            supersedes: None,
5010        };
5011        let capped = cap_wire_summary(&verdict);
5012        // Exactly at the wire cap — the fake/devicerail providers reject
5013        // anything above it fail-closed.
5014        assert_eq!(capped.chars().count(), VERDICT_SUMMARY_MAX_CHARS);
5015        // The 04 §5 pointer to the full local verdict rides the tail.
5016        assert!(
5017            capped.ends_with(']') && capped.contains("full local verdict sha256:"),
5018            "pointer missing: …{}",
5019            &capped[capped.len().saturating_sub(90)..]
5020        );
5021        assert!(capped.starts_with("xxx"));
5022
5023        // Negative control: an in-cap summary passes through verbatim,
5024        // pointer-free.
5025        let short = Verdict {
5026            summary: "all assertions passed".to_owned(),
5027            ..verdict
5028        };
5029        assert_eq!(cap_wire_summary(&short), "all assertions passed");
5030    }
5031}