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