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}