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) = self.gate_supervision(&step_path, step, &resolved_inputs)?
1651 {
1652 return Ok(ctl);
1653 }
1654
1655 let mut acted = if resume_ruling.is_some() {
1656 None
1657 } else {
1658 Some(match work {
1659 Some(FrontierWork::Adopt {
1660 call_id,
1661 intent_path,
1662 outcome,
1663 args,
1664 chain_index,
1665 }) => {
1666 // Record the terminal the crash swallowed; this settles
1667 // the pending intent in the checkpoint fold.
1668 let attempt_n = attempt_of(&intent_path).unwrap_or(1);
1669 let outcome = quarantine_unpersistable(*outcome);
1670 let settled_seq = self.append(
1671 &intent_path,
1672 &RunLogPayload::ActionSettled {
1673 call_id,
1674 outcome: outcome.clone(),
1675 },
1676 )?;
1677 // From here the adopted terminal takes the exact same
1678 // settled-outcome path as a live one.
1679 let adopted = AdoptedSettle {
1680 outcome,
1681 settled_seq,
1682 attempt_n,
1683 };
1684 self.act_chain(
1685 frame,
1686 step,
1687 &step_path,
1688 Some(args),
1689 Some(adopted),
1690 chain_start(chain_index, step)?,
1691 )
1692 .await?
1693 }
1694 Some(FrontierWork::Replay { args, chain_index }) => {
1695 self.act_chain(
1696 frame,
1697 step,
1698 &step_path,
1699 Some(args),
1700 None,
1701 chain_start(chain_index, step)?,
1702 )
1703 .await?
1704 }
1705 Some(FrontierWork::ConfirmEffect { message, .. }) => {
1706 // No dispatch: the adjudication already ruled on the
1707 // act; only the world's testimony is still owed.
1708 ActPhase::Unconfirmed { message }
1709 }
1710 Some(FrontierWork::AbortRuled { .. }) => {
1711 // The ruled abort mirrors a `cancelled` terminal's
1712 // unwind: the open span closes `aborted` and the run
1713 // makes no further semantic claim.
1714 ActPhase::Aborted
1715 }
1716 // The fresh path hands the ready snapshot to the first
1717 // chain attempt — resolved exactly once, above.
1718 None => {
1719 self.act_chain(
1720 frame,
1721 step,
1722 &step_path,
1723 Some(resolved_inputs.clone()),
1724 None,
1725 0,
1726 )
1727 .await?
1728 }
1729 })
1730 };
1731
1732 // ── the settlement/disposition loop (M2 W3) ─────────────────────
1733 //
1734 // One round = one settled act (or the resumed recorded ruling)
1735 // judged and, on fail/unknown, consulted against the handlers.
1736 // Retry-class dispositions re-enter the act with the frozen
1737 // snapshot (new callId, new WAL intent); every re-fold records a
1738 // new verdict superseding the previous one (spine §2 concept 12).
1739 let mut last_verdict_seq: Option<u64> = None;
1740 let mut seeded = false;
1741 let mut active_retry: Option<(RetryPolicy, u32)> = None;
1742 let mut ruling = resume_ruling;
1743 loop {
1744 // What this round established: (status, degraded, summary,
1745 // cited evidence, projected output, error-path class).
1746 let (status, degraded, summary, cited, projected, error_class): (
1747 Option<VerdictStatus>,
1748 bool,
1749 String,
1750 Vec<AssetRef>,
1751 Option<Value>,
1752 Option<ErrorClass>,
1753 );
1754 let mut round_manifest = EvidenceManifest::default();
1755 let mut ruled_from_ledger = false;
1756 match (acted.take(), ruling.take()) {
1757 (None, Some((recorded_status, recorded_degraded))) => {
1758 // Resumed at the recorded verdict: derive the output
1759 // projection from the archived succeeded terminal so
1760 // downstream refs keep working (pure re-projection).
1761 ruled_from_ledger = true;
1762 let recovered = match self.settled.get(&key).map(|fact| &fact.outcome) {
1763 Some(ActionOutcome::Succeeded { result }) => {
1764 let raw_scope =
1765 frame.scope(&self.env, Some((step_id.as_str(), &result.output)));
1766 project_output(step, &result.output, &raw_scope).ok()
1767 }
1768 _ => None,
1769 };
1770 status = Some(recorded_status);
1771 degraded = recorded_degraded;
1772 summary = "resumed at the recorded verdict (handler re-entry)".to_owned();
1773 cited = Vec::new();
1774 projected = recovered;
1775 error_class = None;
1776 // Cross-segment gate (07 §2.2 note 3): the previous
1777 // segment's verdict is in force but its stash died
1778 // with the process. Capture the RESUME-generation
1779 // profile so the eventual exit still carries one —
1780 // self-describing via its own sessionLineage/cursor;
1781 // the failure-instant profile rides the prior
1782 // segment's runSuspended.
1783 if matches!(
1784 recorded_status,
1785 VerdictStatus::Fail | VerdictStatus::Unknown
1786 ) {
1787 let captured = self.capture_summary().await;
1788 self.pending_summaries.insert(key.clone(), captured);
1789 }
1790 }
1791 (Some(phase), _) => match phase {
1792 ActPhase::Succeeded {
1793 result,
1794 degraded: degraded_execution,
1795 settled_seq,
1796 attempt_n,
1797 } => {
1798 let (round_cited, material, observed, manifest) = self
1799 .observing(step, &step_path, attempt_n, &result, settled_seq)
1800 .await?;
1801 round_manifest = manifest;
1802 frame.observed.insert(step_id.as_str().to_owned(), observed);
1803 // Output projection (self-refs see the raw output).
1804 let raw_scope =
1805 frame.scope(&self.env, Some((step_id.as_str(), &result.output)));
1806 let round_projected = match project_output(step, &result.output, &raw_scope)
1807 {
1808 Ok(value) => value,
1809 Err(error) => {
1810 // A failing output projection is a
1811 // compiler/expression bug signal: step
1812 // fails, no retry, no consultation
1813 // (bind_arguments_invalid discipline).
1814 return self
1815 .settle_error(
1816 frame,
1817 &step_path,
1818 &step_id,
1819 VerdictStatus::Fail,
1820 format!("output projection failed: {error}"),
1821 )
1822 .await;
1823 }
1824 };
1825 // Asserting: pure computation over materialized
1826 // values (the vision tail is the one declared
1827 // exception to purity).
1828 let assert_scope =
1829 frame.scope(&self.env, Some((step_id.as_str(), &round_projected)));
1830 let mut outcomes = Vec::with_capacity(step.assertions.len());
1831 let mut degraded_verify = false;
1832 for assertion in &step.assertions {
1833 let evaluated = match &assertion.predicate {
1834 PredicateIR::Expr { expr } => EvaluatedAssertion {
1835 record: eval_expr_assertion(assertion, expr, &assert_scope),
1836 degraded_verify: false,
1837 },
1838 _ => {
1839 eval_observed_assertion(
1840 assertion,
1841 &material,
1842 self.vision.as_deref(),
1843 )
1844 .await
1845 }
1846 };
1847 degraded_verify |= evaluated.degraded_verify;
1848 outcomes.push(evaluated.record);
1849 }
1850 for outcome in &outcomes {
1851 let mut path = step_path.clone();
1852 path.push(PathFrame::Phase {
1853 phase: Phase::Assert,
1854 });
1855 path.push(PathFrame::Assertion {
1856 assert_id: outcome.assert_id.clone(),
1857 });
1858 self.append(
1859 &path,
1860 &RunLogPayload::AssertionEvaluated {
1861 outcome: outcome.clone(),
1862 },
1863 )?;
1864 }
1865 if outcomes.is_empty() {
1866 // No assertions ⇒ no verdict (spine R4):
1867 // execution status only (`unverified`).
1868 status = None;
1869 degraded = false;
1870 summary = String::new();
1871 } else {
1872 let folded = fold_step_verdict(
1873 &outcomes,
1874 degraded_execution,
1875 degraded_verify,
1876 frame.flow.verdict_policy,
1877 );
1878 status = Some(folded.status);
1879 degraded = folded.degraded;
1880 summary = folded.summary;
1881 }
1882 cited = round_cited;
1883 projected = Some(round_projected);
1884 error_class = None;
1885 }
1886 ActPhase::Unconfirmed { message } => {
1887 // The observe half of 「先 reconcile/observe 确认」:
1888 // a fresh observation, the step's own assertions
1889 // over it — the same pure evaluation an assert step
1890 // runs. Expr assertions reference an output the
1891 // timeout never produced and resolve unknown
1892 // (`onMissingInput`, principle 4); element and
1893 // visual predicates read the world and can be
1894 // decisive in both directions. The fold is the
1895 // confirmation verdict: pass — the effect is
1896 // visibly there; fail — visibly not; unknown —
1897 // unconfirmable, exactly the path the bare timeout
1898 // always took.
1899 let needs = VerifyNeeds::of(&step.assertions);
1900 let material = self.fresh_material(&needs, &step_path).await?;
1901 let assert_scope = frame.scope(&self.env, None);
1902 let mut outcomes = Vec::with_capacity(step.assertions.len());
1903 let mut degraded_verify = false;
1904 for assertion in &step.assertions {
1905 let evaluated = match &assertion.predicate {
1906 PredicateIR::Expr { expr } => EvaluatedAssertion {
1907 record: eval_expr_assertion(assertion, expr, &assert_scope),
1908 degraded_verify: false,
1909 },
1910 _ => {
1911 eval_observed_assertion(
1912 assertion,
1913 &material,
1914 self.vision.as_deref(),
1915 )
1916 .await
1917 }
1918 };
1919 degraded_verify |= evaluated.degraded_verify;
1920 outcomes.push(evaluated.record);
1921 }
1922 for outcome in &outcomes {
1923 let mut path = step_path.clone();
1924 path.push(PathFrame::Phase {
1925 phase: Phase::Assert,
1926 });
1927 path.push(PathFrame::Assertion {
1928 assert_id: outcome.assert_id.clone(),
1929 });
1930 self.append(
1931 &path,
1932 &RunLogPayload::AssertionEvaluated {
1933 outcome: outcome.clone(),
1934 },
1935 )?;
1936 }
1937 let folded = fold_step_verdict(
1938 &outcomes,
1939 false,
1940 degraded_verify,
1941 frame.flow.verdict_policy,
1942 );
1943 status = Some(folded.status);
1944 degraded = folded.degraded;
1945 summary = format!(
1946 "{message}; assertions over a fresh observation: {}",
1947 folded.summary
1948 );
1949 cited = Vec::new();
1950 projected = None;
1951 error_class = None;
1952 }
1953 ActPhase::StepFail { class, message } => {
1954 let wire = serde_json::to_value(class).expect("ErrorClass serializes");
1955 let wire = wire.as_str().expect("ErrorClass is a string literal");
1956 status = Some(VerdictStatus::Fail);
1957 degraded = false;
1958 summary = format!("act phase failed [{wire}]: {message}");
1959 cited = Vec::new();
1960 projected = None;
1961 error_class = Some(class);
1962 }
1963 ActPhase::StepUnknown {
1964 message,
1965 error_class: class,
1966 } => {
1967 status = Some(VerdictStatus::Unknown);
1968 degraded = false;
1969 summary = message;
1970 cited = Vec::new();
1971 projected = None;
1972 // Usually none — an unknown has no error to route.
1973 // `session_degraded` is the exception the spine §5
1974 // table names, and it keeps its class so the hook
1975 // selector below reaches `onError`.
1976 error_class = class;
1977 }
1978 ActPhase::Aborted => {
1979 self.append(
1980 &step_path,
1981 &RunLogPayload::StepExited {
1982 provider_state_summary: None,
1983 state: StepState::Aborted,
1984 output: None,
1985 localized: Vec::new(),
1986 localization_gaps: Vec::new(),
1987 },
1988 )?;
1989 return Ok(Ctl::Abort);
1990 }
1991 ActPhase::Suspend(reason) => return Ok(Ctl::Suspend(reason)),
1992 },
1993 (None, None) => unreachable!("every round has an act result or a ruling"),
1994 }
1995
1996 // Record this round's verdict (unless it came off the ledger)
1997 // and seed/reseed the frame fold.
1998 let round_status = status;
1999 if let Some(current) = round_status {
2000 if !ruled_from_ledger {
2001 let folded = FoldedVerdict {
2002 status: current,
2003 degraded,
2004 summary: summary.clone(),
2005 };
2006 let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
2007 let seq = self
2008 .record_step_verdict(
2009 &step_path,
2010 &folded,
2011 cited,
2012 supersedes,
2013 std::mem::take(&mut round_manifest),
2014 )
2015 .await?;
2016 last_verdict_seq = Some(seq);
2017 }
2018 if seeded {
2019 frame.reseed_last(&step_id, current, degraded);
2020 } else {
2021 frame.seed_verdict(&step_id, current, degraded);
2022 seeded = true;
2023 }
2024 }
2025
2026 // Pass / unverified: exit judged and release.
2027 if round_status.is_none() || round_status == Some(VerdictStatus::Pass) {
2028 // An UNVERIFIED exit (R4: no assertions ⇒ no verdict ⇒
2029 // no verdictRecorded carrier) must not drop its
2030 // settlement-evidence manifest — it rides the exit
2031 // instead (item ③ review fix). A pass-verdict exit's
2032 // manifest already rode its verdictRecorded.
2033 let exit_manifest = if round_status.is_none() {
2034 std::mem::take(&mut round_manifest)
2035 } else {
2036 EvidenceManifest::default()
2037 };
2038 self.append(
2039 &step_path,
2040 &RunLogPayload::StepExited {
2041 provider_state_summary: None,
2042 state: StepState::Judged,
2043 output: projected.clone(),
2044 localized: exit_manifest.localized,
2045 localization_gaps: exit_manifest.gaps,
2046 },
2047 )?;
2048 if let Some(projected) = projected {
2049 frame.outputs.insert(step_id.as_str().to_owned(), projected);
2050 }
2051 return Ok(Ctl::Continue);
2052 }
2053 let current = round_status.expect("checked above");
2054
2055 // In-force handler retry budget (one consultation grants
2056 // `max_attempts` re-entries with its backoff schedule).
2057 if let Some((policy, used)) = active_retry.take()
2058 && used < policy.max_attempts
2059 {
2060 self.backoff_policy(&policy, used).await;
2061 active_retry = Some((policy, used + 1));
2062 acted = Some(
2063 // In-force retry policy: re-enter from the chain head
2064 // (item ② ruling — handler retry restarts the chain).
2065 self.act_chain(
2066 frame,
2067 step,
2068 &step_path,
2069 Some(resolved_inputs.clone()),
2070 None,
2071 0,
2072 )
2073 .await?,
2074 );
2075 continue;
2076 }
2077
2078 // Consult the hook: assertion negatives walk onFail/onUnknown,
2079 // error-path negatives walk onError, error-path unknowns walk
2080 // onUnknown (AssertionFailure is a verdict, not an error —
2081 // spine §5).
2082 let hook = if error_class.is_some() {
2083 HandlerHook::OnError
2084 } else if current == VerdictStatus::Fail {
2085 HandlerHook::OnFail
2086 } else {
2087 HandlerHook::OnUnknown
2088 };
2089 match self
2090 .consult_hook(
2091 frame,
2092 &step_path,
2093 step.base.handlers.as_deref(),
2094 hook,
2095 error_class,
2096 )
2097 .await?
2098 {
2099 Consulted::None | Consulted::RepairFailed => {
2100 self.append(
2101 &step_path,
2102 &RunLogPayload::StepExited {
2103 provider_state_summary: None,
2104 state: StepState::Judged,
2105 output: projected.clone(),
2106 localized: Vec::new(),
2107 localization_gaps: Vec::new(),
2108 },
2109 )?;
2110 if let Some(projected) = projected {
2111 frame.outputs.insert(step_id.as_str().to_owned(), projected);
2112 }
2113 return Ok(if current == VerdictStatus::Fail {
2114 Ctl::HaltFail
2115 } else {
2116 Ctl::Continue
2117 });
2118 }
2119 Consulted::Continue => {
2120 // Record-and-release: the verdict stands, downstream
2121 // is not halted (spine §3 disposition table).
2122 self.append(
2123 &step_path,
2124 &RunLogPayload::StepExited {
2125 provider_state_summary: None,
2126 state: StepState::Judged,
2127 output: projected.clone(),
2128 localized: Vec::new(),
2129 localization_gaps: Vec::new(),
2130 },
2131 )?;
2132 if let Some(projected) = projected {
2133 frame.outputs.insert(step_id.as_str().to_owned(), projected);
2134 }
2135 return Ok(Ctl::Continue);
2136 }
2137 Consulted::Abort => {
2138 self.append(
2139 &step_path,
2140 &RunLogPayload::StepExited {
2141 provider_state_summary: None,
2142 state: StepState::Aborted,
2143 output: None,
2144 localized: Vec::new(),
2145 localization_gaps: Vec::new(),
2146 },
2147 )?;
2148 return Ok(Ctl::Abort);
2149 }
2150 Consulted::Escalated {
2151 status: ruled,
2152 summary: ruled_summary,
2153 evidence: ruling_evidence,
2154 } => {
2155 let folded = FoldedVerdict {
2156 status: ruled,
2157 degraded: false,
2158 summary: ruled_summary,
2159 };
2160 let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
2161 let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
2162 let ruled_seq = self
2163 .record_step_verdict(&step_path, &folded, cited, supersedes, manifest)
2164 .await?;
2165 self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
2166 if seeded {
2167 frame.reseed_last(&step_id, ruled, false);
2168 } else {
2169 frame.seed_verdict(&step_id, ruled, false);
2170 }
2171 self.append(
2172 &step_path,
2173 &RunLogPayload::StepExited {
2174 provider_state_summary: None,
2175 state: StepState::Judged,
2176 output: projected.clone(),
2177 localized: Vec::new(),
2178 localization_gaps: Vec::new(),
2179 },
2180 )?;
2181 if let Some(projected) = projected {
2182 frame.outputs.insert(step_id.as_str().to_owned(), projected);
2183 }
2184 return Ok(if ruled == VerdictStatus::Fail {
2185 Ctl::HaltFail
2186 } else {
2187 Ctl::Continue
2188 });
2189 }
2190 Consulted::Retry(policy) => {
2191 self.backoff_policy(&policy, 0).await;
2192 active_retry = Some((policy, 1));
2193 acted = Some(
2194 self.act_chain(
2195 frame,
2196 step,
2197 &step_path,
2198 Some(resolved_inputs.clone()),
2199 None,
2200 0,
2201 )
2202 .await?,
2203 );
2204 }
2205 Consulted::Repaired | Consulted::RepairDone => {
2206 // The world was (declared) fixed: one re-entry; a
2207 // further negative re-consults (maxTriggers bounds
2208 // the total).
2209 acted = Some(
2210 self.act_chain(
2211 frame,
2212 step,
2213 &step_path,
2214 Some(resolved_inputs.clone()),
2215 None,
2216 0,
2217 )
2218 .await?,
2219 );
2220 }
2221 Consulted::Pending(pending) => {
2222 // The verdict of this round is on the ledger (recorded
2223 // above): the resume segment re-enters through the
2224 // recorded-ruling jump. Span stays open.
2225 return Ok(Ctl::AwaitHuman(pending));
2226 }
2227 Consulted::Propagate(ctl) => return Ok(ctl),
2228 }
2229 }
2230 }
2231
2232 /// Settles a step whose execution ended in a definite negative (fail)
2233 /// or an unconfirmable state (unknown): verdict, write-back, exit.
2234 async fn settle_error(
2235 &mut self,
2236 frame: &mut FrameState<'a>,
2237 step_path: &RunPath,
2238 step_id: &StepId,
2239 status: VerdictStatus,
2240 summary: String,
2241 ) -> Result<Ctl, RunnerError> {
2242 let folded = FoldedVerdict {
2243 status,
2244 degraded: false,
2245 summary,
2246 };
2247 self.record_step_verdict(
2248 step_path,
2249 &folded,
2250 Vec::new(),
2251 None,
2252 EvidenceManifest::default(),
2253 )
2254 .await?;
2255 frame.seed_verdict(step_id, status, false);
2256 self.append(
2257 step_path,
2258 &RunLogPayload::StepExited {
2259 provider_state_summary: None,
2260 state: StepState::Judged,
2261 // No output projection completed on the error path.
2262 output: None,
2263 localized: Vec::new(),
2264 localization_gaps: Vec::new(),
2265 },
2266 )?;
2267 if status == VerdictStatus::Fail {
2268 Ok(Ctl::HaltFail)
2269 } else {
2270 Ok(Ctl::Continue)
2271 }
2272 }
2273
2274 /// The act phase: the bound attempt chain with in-attempt retry
2275 /// (spine §6.5 mount point 1 — every retry is a new callId and a new
2276 /// WAL intent; the chain advances only on `action_failed_final`).
2277 async fn act_chain(
2278 &mut self,
2279 frame: &FrameState<'a>,
2280 step: &'a ActionStepIR,
2281 step_path: &RunPath,
2282 args_override: Option<Value>,
2283 mut adopted: Option<AdoptedSettle>,
2284 start_position: usize,
2285 ) -> Result<ActPhase, RunnerError> {
2286 let key = instance_key(step_path);
2287 let chain_len = step.binding.attempts.len();
2288 for (position, attempt) in step
2289 .binding
2290 .attempts
2291 .iter()
2292 .enumerate()
2293 .skip(start_position)
2294 {
2295 // The entry attempt consumes the override snapshot (the ready
2296 // snapshot on a fresh run, the archived argsSnapshot on a
2297 // crash re-entry — anchored at the recorded chain position,
2298 // 07 §1.4); a chain advance re-resolves the next attempt's
2299 // own argument expressions.
2300 let args = if position == start_position && args_override.is_some() {
2301 args_override.clone().expect("checked is_some")
2302 } else {
2303 match self.resolve_args(frame, attempt) {
2304 Ok(args) => args,
2305 Err(message) => {
2306 return Ok(ActPhase::StepFail {
2307 class: ErrorClass::BindArgumentsInvalid,
2308 message,
2309 });
2310 }
2311 }
2312 };
2313
2314 let mut tries: u32 = 0;
2315 loop {
2316 tries += 1;
2317 let (outcome, settled_seq, attempt_n) = match adopted.take() {
2318 // The reconciled terminal is this try's settled
2319 // outcome; its WAL intent and `actionSettled` are
2320 // already on record — no dispatch.
2321 Some(settle) => (settle.outcome, settle.settled_seq, settle.attempt_n),
2322 None => {
2323 let attempt_n = self.next_attempt_n(&key);
2324 let call_id = uuid::Uuid::new_v4().to_string();
2325 let mut attempt_path = step_path.clone();
2326 attempt_path.push(PathFrame::Attempt { n: attempt_n });
2327 // WAL discipline (spine §6.2): the intent commit
2328 // *is* the fsync; only after it returns may the
2329 // dispatch leave.
2330 self.store.write_action_intent(
2331 &self.run_id,
2332 now_ms(),
2333 &attempt_path,
2334 &call_id,
2335 args.clone(),
2336 Some(pointlock_store::IntentDispatch {
2337 chain_index: (position + 1) as u32,
2338 channel: attempt.channel,
2339 action_name: attempt.action_name.clone(),
2340 }),
2341 )?;
2342 let call = BoundActionCall {
2343 call_id: call_id.clone(),
2344 action_name: attempt.action_name.clone(),
2345 arguments: args.clone(),
2346 action_timeout_ms: step.base.timeout_ms,
2347 request_timeout_ms: None,
2348 };
2349 let outcome = match self.session.execute(call, None).await {
2350 Ok(outcome) => outcome,
2351 Err(error) => {
2352 // No terminal could be obtained: the intent
2353 // stays pending; suspend and reconcile on
2354 // resume.
2355 return Ok(ActPhase::Suspend(format!(
2356 "no terminal for callId {call_id}: {error}"
2357 )));
2358 }
2359 };
2360 let outcome = quarantine_unpersistable(outcome);
2361 let settled_seq = self.append(
2362 &attempt_path,
2363 &RunLogPayload::ActionSettled {
2364 call_id: call_id.clone(),
2365 outcome: outcome.clone(),
2366 },
2367 )?;
2368 (outcome, settled_seq, attempt_n)
2369 }
2370 };
2371 match outcome {
2372 ActionOutcome::Succeeded { result } => {
2373 let degraded = !execution_accepted(attempt, &result.execution);
2374 return Ok(ActPhase::Succeeded {
2375 result,
2376 degraded,
2377 settled_seq,
2378 attempt_n,
2379 });
2380 }
2381 other => {
2382 let class = classify(&other);
2383 let message = terminal_message(&other);
2384 if class == ErrorClass::ActionCancelled {
2385 return Ok(ActPhase::Aborted);
2386 }
2387 if retry_allowed(step, class, tries) {
2388 self.backoff(step, tries).await;
2389 continue;
2390 }
2391 match class {
2392 // Chain advance: only a final (possibly
2393 // degradable) failure tries the next attempt.
2394 ErrorClass::ActionFailedFinal if position + 1 < chain_len => break,
2395 ErrorClass::ActionTimedOut => {
2396 // A recorded `timedOut` is a certain fate —
2397 // reconcile returns it verbatim and adds
2398 // nothing — so the only confirmation channel
2399 // left is OBSERVATION (spine §5 / 07 §4.3).
2400 // With assertions declared, the settlement
2401 // loop asks the world; without any there is
2402 // nothing that could confirm, and the step
2403 // folds to the honest unknown directly.
2404 if !step.assertions.is_empty() {
2405 return Ok(ActPhase::Unconfirmed {
2406 message: format!("action timed out ({message})"),
2407 });
2408 }
2409 return Ok(ActPhase::StepUnknown {
2410 message: format!(
2411 "action timed out and the outcome could not be \
2412 confirmed: {message}"
2413 ),
2414 error_class: None,
2415 });
2416 }
2417 ErrorClass::SessionDegraded => {
2418 // spine §5: "当前 step → unknown,触发 flow 级
2419 // onError handler" — both halves, so the
2420 // class rides along to the hook selector.
2421 return Ok(ActPhase::StepUnknown {
2422 message: format!("session degraded: {message}"),
2423 error_class: Some(ErrorClass::SessionDegraded),
2424 });
2425 }
2426 _ => {
2427 return Ok(ActPhase::StepFail { class, message });
2428 }
2429 }
2430 }
2431 }
2432 }
2433 }
2434 unreachable!("the act chain always returns from within its last attempt")
2435 }
2436
2437 /// Allocates the next attempt number for a step instance (monotonic
2438 /// across resume segments — the base is harvested from the log).
2439 fn next_attempt_n(&mut self, key: &str) -> u64 {
2440 let counter = self.attempt_base.entry(key.to_owned()).or_insert(0);
2441 *counter += 1;
2442 *counter
2443 }
2444
2445 /// Waits out the retry backoff (spine §6.5 mount 1).
2446 async fn backoff(&self, step: &ActionStepIR, tries: u32) {
2447 let Some(policy) = &step.base.retry else {
2448 return;
2449 };
2450 self.backoff_policy(policy, tries).await;
2451 }
2452
2453 /// Sleeps one backoff period of an explicit policy (the handler-retry
2454 /// disposition carries its own policy, independent of `StepBase.retry`
2455 /// — spine §6.5 mount 2).
2456 async fn backoff_policy(&self, policy: &RetryPolicy, tries: u32) {
2457 let ms = backoff_ms(policy, tries);
2458 if ms > 0 {
2459 tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
2460 }
2461 }
2462
2463 // ─── human steps & supervision gate (06 §2/§5; spine §6.8/§6.9) ─────────
2464
2465 /// The R13 supervision gate of one action step, consulted between the
2466 /// preflight probes and the act chain (i.e. strictly before any
2467 /// `actionIntent`). Returns `None` to let the dispatch proceed.
2468 ///
2469 /// A pending gate request from a previous segment resolves first —
2470 /// regardless of *this* segment's policy (the request survives across
2471 /// segments, spine §6.9): `proceed` falls through to the intent,
2472 /// `abort` exits the step `aborted` and aborts the run without
2473 /// consulting any handler, anything else (unanswered, or the non-final
2474 /// `suspend` ruling) re-awaits. Fresh gating follows this segment's
2475 /// policy: `mutating` gates mutating action steps only, `all` gates
2476 /// every action step.
2477 fn gate_supervision(
2478 &mut self,
2479 step_path: &RunPath,
2480 step: &'a ActionStepIR,
2481 resolved_inputs: &Value,
2482 ) -> Result<Option<Ctl>, RunnerError> {
2483 let key = instance_key(step_path);
2484 if let Some(fact) = self
2485 .human
2486 .get(&key)
2487 .filter(|fact| fact.purpose == HumanPurpose::Supervision)
2488 {
2489 let decision = fact
2490 .final_response
2491 .as_ref()
2492 .and_then(|response| response.get("decision"))
2493 .and_then(Value::as_str);
2494 return match decision {
2495 // humanResponded(proceed) is on the ledger before the
2496 // intent this clears the way for (§6.9 WAL order).
2497 Some("proceed") => Ok(None),
2498 Some("abort") => {
2499 // The human ruling is final: no handler is consulted
2500 // (§6.9 R13); the step exits `aborted` and the run
2501 // takes the existing aborted terminal.
2502 self.append(
2503 step_path,
2504 &RunLogPayload::StepExited {
2505 provider_state_summary: None,
2506 state: StepState::Aborted,
2507 output: None,
2508 localized: Vec::new(),
2509 localization_gaps: Vec::new(),
2510 },
2511 )?;
2512 Ok(Some(Ctl::Abort))
2513 }
2514 _ => Ok(Some(Ctl::AwaitHuman(HumanPending {
2515 run_path: fact.run_path.clone(),
2516 request_id: fact.request_id.clone(),
2517 purpose: HumanPurpose::Supervision,
2518 mode: None,
2519 prompt: fact.prompt.clone(),
2520 deadline_at_ms: None,
2521 }))),
2522 };
2523 }
2524 let Some(policy) = self.supervise else {
2525 return Ok(None);
2526 };
2527 let gated = match policy {
2528 SupervisePolicy::All => true,
2529 SupervisePolicy::Mutating => step.effect == EffectClassAction::Mutating,
2530 };
2531 if !gated {
2532 return Ok(None);
2533 }
2534 // Fresh gate: auto-generated description over runPath /
2535 // actionName / resolvedInputs (§6.9). No mode, no decisions
2536 // contract, no deadline — the decision vocabulary is the closed
2537 // proceed | abort | suspend, arbitrated by the store.
2538 let attempt = step
2539 .binding
2540 .attempts
2541 .first()
2542 .expect("sealed action steps carry at least one bound attempt");
2543 let action_name = attempt.action_name.as_str().to_owned();
2544 let rendered = render_run_path(step_path);
2545 let request_id = uuid::Uuid::new_v4().to_string();
2546 let prompt = format!(
2547 "Supervision gate: approve dispatching action '{action_name}' at {rendered}? \
2548 The resolved inputs are presented."
2549 );
2550 let presents = serde_json::json!([
2551 { "kind": "value", "label": "runPath", "value": rendered },
2552 { "kind": "value", "label": "actionName", "value": action_name },
2553 { "kind": "value", "label": "resolvedInputs", "value": resolved_inputs },
2554 ]);
2555 // fsync-before-notify (spine §6.9): the append commit *is* the
2556 // fsync; the runner then suspends — any notification happens in
2557 // the CLI layer, strictly after this returns.
2558 self.append(
2559 step_path,
2560 &RunLogPayload::HumanRequested {
2561 request_id: request_id.clone(),
2562 purpose: HumanPurpose::Supervision,
2563 mode: None,
2564 prompt: prompt.clone(),
2565 presents,
2566 decisions: None,
2567 output_schema: None,
2568 deadline_at_ms: None,
2569 },
2570 )?;
2571 Ok(Some(Ctl::AwaitHuman(HumanPending {
2572 run_path: step_path.clone(),
2573 request_id,
2574 purpose: HumanPurpose::Supervision,
2575 mode: None,
2576 prompt,
2577 deadline_at_ms: None,
2578 })))
2579 }
2580
2581 /// Executes one human step (06 §5.1 pinned order): ready (presents
2582 /// materialized once and frozen) → `stepEntered` → declared preflight
2583 /// → `humanRequested` (fsynced by its commit) → suspend /
2584 /// [`RunOutcome::AwaitingHuman`]. Resume settles a paired response
2585 /// through the four-mode mapping, re-awaits an unanswered request
2586 /// inside its deadline, and lazily settles an expired one to `unknown`
2587 /// — the settlement result depends only on `deadlineAtMs` and response
2588 /// presence, never on the settlement instant (06 §5.3).
2589 async fn exec_human(
2590 &mut self,
2591 frame: &mut FrameState<'a>,
2592 step_path: RunPath,
2593 step: &'a HumanStepIR,
2594 ) -> Result<Ctl, RunnerError> {
2595 let step_id = step.base.step_id.clone();
2596 let key = instance_key(&step_path);
2597
2598 // A request already on the ledger for this instance: settle or
2599 // keep waiting — never a second `humanRequested` while one is
2600 // pending.
2601 if let Some(fact) = self
2602 .human
2603 .get(&key)
2604 .filter(|fact| fact.purpose == HumanPurpose::Step)
2605 {
2606 let fact = fact.clone();
2607 if let Some(response) = fact.final_response.clone() {
2608 // The span is open by construction (settlement closes it
2609 // for good); enter_step only consumes it.
2610 self.enter_step(&step_path, &step.base, Value::Null)?;
2611 return self
2612 .settle_human_response(frame, &step_path, step, &fact, response)
2613 .await;
2614 }
2615 match fact.deadline_at_ms {
2616 Some(deadline) if self.now() > deadline => {
2617 self.enter_step(&step_path, &step.base, Value::Null)?;
2618 return self
2619 .settle_human_timeout(frame, &step_path, step, &fact)
2620 .await;
2621 }
2622 // Unanswered and not expired: re-await the same request
2623 // (no new requestId, no re-notify obligation here).
2624 _ => {
2625 return Ok(Ctl::AwaitHuman(HumanPending {
2626 run_path: fact.run_path.clone(),
2627 request_id: fact.request_id.clone(),
2628 purpose: HumanPurpose::Step,
2629 mode: Some(step.mode),
2630 prompt: fact.prompt.clone(),
2631 deadline_at_ms: fact.deadline_at_ms,
2632 }));
2633 }
2634 }
2635 }
2636
2637 // Ready: materialize `presents` once and freeze — the
2638 // `resolvedInputs` snapshot discipline (06 §2.3). A
2639 // crash/suspension-opened span reuses the archived snapshot.
2640 let snapshot = match self.open_span_inputs(&key) {
2641 Some(archived) => archived,
2642 None => {
2643 let scope = frame.scope(&self.env, None);
2644 let mut items = Vec::with_capacity(step.presents.len());
2645 let mut error = None;
2646 for (index, expr) in step.presents.iter().enumerate() {
2647 match pointlock_expr::eval(expr, &scope) {
2648 Ok(value) => items.push(value),
2649 Err(eval_error) => {
2650 error =
2651 Some(format!("present #{index} evaluation failed: {eval_error}"));
2652 break;
2653 }
2654 }
2655 }
2656 if let Some(message) = error {
2657 // A failing presents evaluation is a compiler /
2658 // expression bug signal (bind_arguments_invalid
2659 // discipline): step fails, nothing is asked.
2660 self.enter_step(&step_path, &step.base, Value::Null)?;
2661 return self
2662 .settle_error(
2663 frame,
2664 &step_path,
2665 &step_id,
2666 VerdictStatus::Fail,
2667 format!("human presents failed [bind_arguments_invalid]: {message}"),
2668 )
2669 .await;
2670 }
2671 serde_json::json!({ "presents": items })
2672 }
2673 };
2674 self.enter_step(&step_path, &step.base, snapshot.clone())?;
2675 if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
2676 return Ok(ctl);
2677 }
2678 let presents = snapshot
2679 .get("presents")
2680 .cloned()
2681 .unwrap_or(Value::Array(Vec::new()));
2682 let request_id = uuid::Uuid::new_v4().to_string();
2683 // `timeoutMs` converts to the absolute deadline watermark at
2684 // request creation (06 §5.3) — the sole lazy-settlement input.
2685 let deadline_at_ms = self.now().saturating_add(step.timeout_ms);
2686 // fsync-before-notify (spine §6.8, 06 §5.1): the append commit
2687 // *is* the fsync (WAL + synchronous=FULL); the runner suspends
2688 // and never notifies — channels are the CLI layer's job.
2689 self.append(
2690 &step_path,
2691 &RunLogPayload::HumanRequested {
2692 request_id: request_id.clone(),
2693 purpose: HumanPurpose::Step,
2694 mode: Some(step.mode),
2695 prompt: step.prompt.clone(),
2696 presents,
2697 decisions: step.decisions.clone(),
2698 output_schema: step.output_schema.clone(),
2699 deadline_at_ms: Some(deadline_at_ms),
2700 },
2701 )?;
2702 Ok(Ctl::AwaitHuman(HumanPending {
2703 run_path: step_path,
2704 request_id,
2705 purpose: HumanPurpose::Step,
2706 mode: Some(step.mode),
2707 prompt: step.prompt.clone(),
2708 deadline_at_ms: Some(deadline_at_ms),
2709 }))
2710 }
2711
2712 /// Settles a human step from its arbitrated final response — the
2713 /// four-mode verdict/output mapping (06 §2.2 as adjudicated):
2714 ///
2715 /// | mode | response | verdict | step output |
2716 /// |---|---|---|---|
2717 /// | `confirm` | `decision` = first label | pass | the response object |
2718 /// | `confirm` | `decision` = second label | fail | the response object |
2719 /// | `judge` | `status` pass/fail/unknown | verbatim | the response object |
2720 /// | `provideInput` | `input` (schema-checked by the store) | pass | the input value |
2721 /// | `repairWorld` | `decision: "done"` | pass (testimony, not observation; never degraded) | the response object |
2722 /// | `repairWorld` | `decision: "cannotRepair"` | fail (06 §2.2 — a verdict, catchable by `onFail`; never a run abort) | the response object |
2723 ///
2724 /// Every settlement materializes the response as a canonical JSON
2725 /// evidence document and cites it from the verdict (06 §6).
2726 async fn settle_human_response(
2727 &mut self,
2728 frame: &mut FrameState<'a>,
2729 step_path: &RunPath,
2730 step: &'a HumanStepIR,
2731 fact: &HumanRequestFact,
2732 response: Value,
2733 ) -> Result<Ctl, RunnerError> {
2734 let step_id = step.base.step_id.clone();
2735 let decision = response
2736 .get("decision")
2737 .and_then(Value::as_str)
2738 .unwrap_or_default()
2739 .to_owned();
2740 let (status, output, summary) = match step.mode {
2741 HumanMode::Confirm => {
2742 let labels = step
2743 .decisions
2744 .as_ref()
2745 .expect("load validated confirm decisions");
2746 // Position-mapped double label: first → pass, second →
2747 // fail (membership was the store arbitration's check).
2748 let status = if labels.first().map(String::as_str) == Some(decision.as_str()) {
2749 VerdictStatus::Pass
2750 } else {
2751 VerdictStatus::Fail
2752 };
2753 let position = if status == VerdictStatus::Pass {
2754 "first"
2755 } else {
2756 "second"
2757 };
2758 (
2759 status,
2760 Some(response.clone()),
2761 format!(
2762 "human confirm decision '{decision}' is the {position} label \
2763 (position-mapped verdict)"
2764 ),
2765 )
2766 }
2767 HumanMode::Judge => {
2768 // The human ruling *is* the verdict (spine §6.3).
2769 let status = match response.get("status").and_then(Value::as_str) {
2770 Some("pass") => VerdictStatus::Pass,
2771 Some("fail") => VerdictStatus::Fail,
2772 _ => VerdictStatus::Unknown,
2773 };
2774 let label = response
2775 .get("status")
2776 .and_then(Value::as_str)
2777 .unwrap_or("unknown");
2778 (
2779 status,
2780 Some(response.clone()),
2781 format!("human judge ruling: {label}"),
2782 )
2783 }
2784 HumanMode::ProvideInput => {
2785 // The store validated `input` against `outputSchema`; the
2786 // established fact is "a human provided schema-valid
2787 // input" — pass, and the input *is* the step output.
2788 let input = response.get("input").cloned().unwrap_or(Value::Null);
2789 (
2790 VerdictStatus::Pass,
2791 Some(input),
2792 "human provided input validated against the outputSchema".to_owned(),
2793 )
2794 }
2795 HumanMode::RepairWorld => {
2796 // 06 §2.2's closed mapping: `done` → pass, `cannotRepair`
2797 // → fail. Testimony, not observation — pass carries no
2798 // degraded flag (machine re-checks belong to follow-up
2799 // assert/preflight steps), and a fail is a verdict the
2800 // step's own `onFail` may catch, never a unilateral run
2801 // abort.
2802 let status = if decision == "done" {
2803 VerdictStatus::Pass
2804 } else {
2805 VerdictStatus::Fail
2806 };
2807 let summary = if status == VerdictStatus::Pass {
2808 "human declared the world repaired (testimony; follow-up machine \
2809 re-check advised)"
2810 .to_owned()
2811 } else {
2812 "human declared the world unrepairable (cannotRepair)".to_owned()
2813 };
2814 (status, Some(response.clone()), summary)
2815 }
2816 };
2817 let asset = self.put_human_evidence(step_path, fact, Some(&response), "response")?;
2818 let folded = FoldedVerdict {
2819 status,
2820 degraded: false,
2821 summary,
2822 };
2823 let verdict_seq = self
2824 .record_step_verdict(
2825 step_path,
2826 &folded,
2827 vec![asset.clone()],
2828 None,
2829 human_manifest(&asset),
2830 )
2831 .await?;
2832 if let Some(sha256) = &asset.sha256 {
2833 self.store
2834 .link_evidence(&self.run_id, verdict_seq, &asset.id, sha256)?;
2835 }
2836 frame.seed_verdict(&step_id, status, false);
2837 self.append(
2838 step_path,
2839 &RunLogPayload::StepExited {
2840 provider_state_summary: None,
2841 state: StepState::Judged,
2842 output: output.clone(),
2843 localized: Vec::new(),
2844 localization_gaps: Vec::new(),
2845 },
2846 )?;
2847 if let Some(output) = output {
2848 frame.outputs.insert(step_id.as_str().to_owned(), output);
2849 }
2850 if status == VerdictStatus::Fail {
2851 Ok(Ctl::HaltFail)
2852 } else {
2853 Ok(Ctl::Continue)
2854 }
2855 }
2856
2857 /// Lazy timeout settlement (06 §5.3): the deadline passed with no
2858 /// arbitrated response — verdict `unknown` (`onTimeout` is fixed),
2859 /// no output (downstream consumers of it block). The judgment inputs
2860 /// are `deadlineAtMs` and response absence only; the settlement
2861 /// instant leaves no trace in the verdict ("no one came" is itself a
2862 /// recorded historical fact, 06 §6).
2863 async fn settle_human_timeout(
2864 &mut self,
2865 frame: &mut FrameState<'a>,
2866 step_path: &RunPath,
2867 step: &'a HumanStepIR,
2868 fact: &HumanRequestFact,
2869 ) -> Result<Ctl, RunnerError> {
2870 let step_id = &step.base.step_id;
2871 let deadline = fact.deadline_at_ms.unwrap_or_default();
2872 let asset = self.put_human_evidence(step_path, fact, None, "timeout")?;
2873 let folded = FoldedVerdict {
2874 status: VerdictStatus::Unknown,
2875 degraded: false,
2876 summary: format!(
2877 "human response deadline (deadlineAtMs {deadline}) passed without a \
2878 response; onTimeout is fixed to unknown"
2879 ),
2880 };
2881 let verdict_seq = self
2882 .record_step_verdict(
2883 step_path,
2884 &folded,
2885 vec![asset.clone()],
2886 None,
2887 human_manifest(&asset),
2888 )
2889 .await?;
2890 if let Some(sha256) = &asset.sha256 {
2891 self.store
2892 .link_evidence(&self.run_id, verdict_seq, &asset.id, sha256)?;
2893 }
2894 frame.seed_verdict(step_id, VerdictStatus::Unknown, false);
2895
2896 // A timed-out human step is an unknown verdict: the onUnknown
2897 // ladder applies (06's escalation pattern). Re-ask dispositions
2898 // (retry/repair) need fresh-request machinery — typed M2 refusal.
2899 match self
2900 .consult_hook(
2901 frame,
2902 step_path,
2903 step.base.handlers.as_deref(),
2904 HandlerHook::OnUnknown,
2905 None,
2906 )
2907 .await?
2908 {
2909 Consulted::None | Consulted::RepairFailed | Consulted::Continue => {}
2910 Consulted::Abort => {
2911 self.append(
2912 step_path,
2913 &RunLogPayload::StepExited {
2914 provider_state_summary: None,
2915 state: StepState::Aborted,
2916 output: None,
2917 localized: Vec::new(),
2918 localization_gaps: Vec::new(),
2919 },
2920 )?;
2921 return Ok(Ctl::Abort);
2922 }
2923 Consulted::Escalated {
2924 status: ruled,
2925 summary: ruled_summary,
2926 evidence: ruling_evidence,
2927 } => {
2928 let folded = FoldedVerdict {
2929 status: ruled,
2930 degraded: false,
2931 summary: ruled_summary,
2932 };
2933 let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
2934 let ruled_seq = self
2935 .record_step_verdict(
2936 step_path,
2937 &folded,
2938 cited,
2939 Some(format!("seq:{verdict_seq}")),
2940 manifest,
2941 )
2942 .await?;
2943 self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
2944 frame.reseed_last(step_id, ruled, false);
2945 self.append(
2946 step_path,
2947 &RunLogPayload::StepExited {
2948 provider_state_summary: None,
2949 state: StepState::Judged,
2950 output: None,
2951 localized: Vec::new(),
2952 localization_gaps: Vec::new(),
2953 },
2954 )?;
2955 return Ok(if ruled == VerdictStatus::Fail {
2956 Ctl::HaltFail
2957 } else {
2958 Ctl::Continue
2959 });
2960 }
2961 Consulted::Retry(_) | Consulted::Repaired | Consulted::RepairDone => {
2962 return Err(RunnerError::M0Unsupported {
2963 detail: format!(
2964 "re-ask dispositions on the timed-out human step '{step_id}' need \
2965 fresh-request machinery — not in the M2 subset \
2966 (escalate/continue/abort are supported)"
2967 ),
2968 });
2969 }
2970 Consulted::Pending(pending) => return Ok(Ctl::AwaitHuman(pending)),
2971 Consulted::Propagate(ctl) => return Ok(ctl),
2972 }
2973
2974 self.append(
2975 step_path,
2976 &RunLogPayload::StepExited {
2977 provider_state_summary: None,
2978 state: StepState::Judged,
2979 output: None,
2980 localized: Vec::new(),
2981 localization_gaps: Vec::new(),
2982 },
2983 )?;
2984 Ok(Ctl::Continue)
2985 }
2986
2987 /// Materializes a human settlement as a canonical JSON evidence
2988 /// document in the content-addressed store and mints its citable
2989 /// [`AssetRef`] (06 §6: who / when-relative-to-deadline / what was
2990 /// decided / what was presented). Deliberately clock-free: the
2991 /// document is a pure function of the request and the arbitrated
2992 /// response (or its absence).
2993 /// Links an escalate ruling's settlement document to its superseding
2994 /// verdict row (same file-before-row-before-log join as body humans).
2995 fn link_ruling_evidence(
2996 &mut self,
2997 verdict_seq: u64,
2998 evidence: &Option<AssetRef>,
2999 ) -> Result<(), RunnerError> {
3000 if let Some(asset) = evidence
3001 && let Some(sha256) = &asset.sha256
3002 {
3003 self.store
3004 .link_evidence(&self.run_id, verdict_seq, &asset.id, sha256)?;
3005 }
3006 Ok(())
3007 }
3008
3009 fn put_human_evidence(
3010 &mut self,
3011 step_path: &RunPath,
3012 fact: &HumanRequestFact,
3013 response: Option<&Value>,
3014 settled_as: &str,
3015 ) -> Result<AssetRef, RunnerError> {
3016 let document = serde_json::json!({
3017 "pointlockEvidence": "humanResponse/1",
3018 "requestId": fact.request_id,
3019 "runId": self.run_id,
3020 "runPath": render_run_path(step_path),
3021 "purpose": fact.purpose,
3022 "mode": fact.mode,
3023 "prompt": fact.prompt,
3024 "presented": fact.presents,
3025 "response": response.cloned().unwrap_or(Value::Null),
3026 "actor": match (settled_as, &fact.final_actor) {
3027 // Timeout settlements have no actor (06 §6).
3028 ("timeout", _) => Value::Null,
3029 (_, Some(actor)) => Value::String(actor.clone()),
3030 (_, None) => Value::Null,
3031 },
3032 "deadlineAtMs": fact.deadline_at_ms,
3033 "settledAs": settled_as,
3034 });
3035 let bytes = to_canonical_json(&document).into_bytes();
3036 // file-before-row-before-log: the bytes are durable before the
3037 // verdict event cites them.
3038 let put = self.store.put_evidence(&bytes, "application/json")?;
3039 Ok(AssetRef {
3040 id: format!("humanResponse:{}", fact.request_id),
3041 media_type: "application/json".to_owned(),
3042 // Locally minted evidence: the URI is the content-addressed
3043 // library path (06 §6).
3044 uri: put.local_path,
3045 sha256: Some(put.sha256),
3046 })
3047 }
3048
3049 // ─── handler engine (spine §3/§6.5 mount 2; M2 W3) ──────────────────────
3050 //
3051 // Handlers are explicit policies on four hooks; they yield a
3052 // disposition, never data (R10). Consultation happens *inside* the
3053 // host step's open span, before `stepExited`, so a retry disposition
3054 // re-enters the failing phase within the same entered/exited pair.
3055 // The `handlerTriggered` audit event anchors at the host step path;
3056 // the hook frame (`/hook:<name>:<n>`) anchors the disposition's own
3057 // work (escalate humans, repair frames).
3058
3059 /// Resolves the binding a hook consults: the step-level list first
3060 /// (first match wins), else the flow-level list (spine §3
3061 /// `StepBase.handlers` overrides `FlowIR.handlers`). `onError`
3062 /// bindings additionally filter on `errorClasses`.
3063 fn resolve_binding(
3064 &self,
3065 frame: &FrameState<'a>,
3066 step_handlers: Option<&'a [HandlerBinding]>,
3067 hook: HandlerHook,
3068 error_class: Option<ErrorClass>,
3069 ) -> Option<&'a HandlerBinding> {
3070 let matches = |binding: &&'a HandlerBinding| {
3071 binding.hook == hook
3072 && (hook != HandlerHook::OnError
3073 || match (&binding.error_classes, error_class) {
3074 (None, _) => true,
3075 (Some(filter), Some(class)) => filter.contains(&class),
3076 (Some(_), None) => false,
3077 })
3078 };
3079 step_handlers
3080 .and_then(|bindings| bindings.iter().find(matches))
3081 .or_else(|| {
3082 frame
3083 .flow
3084 .handlers
3085 .as_deref()
3086 .and_then(|bindings| bindings.iter().find(matches))
3087 })
3088 }
3089
3090 /// Consults the matching handler binding for `hook` on the host step.
3091 ///
3092 /// Trigger counting is per host instance per hook and continues
3093 /// across segments (harvested from the ledger); an exhausted budget
3094 /// returns [`Consulted::None`] — the natural path stands. A pending
3095 /// escalate continuation (a previous segment's unanswered hook human)
3096 /// is settled or re-awaited *without* consuming a new trigger.
3097 async fn consult_hook(
3098 &mut self,
3099 frame: &mut FrameState<'a>,
3100 step_path: &RunPath,
3101 step_handlers: Option<&'a [HandlerBinding]>,
3102 hook: HandlerHook,
3103 error_class: Option<ErrorClass>,
3104 ) -> Result<Consulted, RunnerError> {
3105 let Some(binding) = self.resolve_binding(frame, step_handlers, hook, error_class) else {
3106 return Ok(Consulted::None);
3107 };
3108 let counter_key = crate::align::hook_trigger_key(&instance_key(step_path), hook);
3109 let current = self.hook_triggers.get(&counter_key).copied().unwrap_or(0);
3110
3111 // Escalate continuation: trigger N already on the ledger and its
3112 // hook human still governs — settle or re-await, no new trigger.
3113 if current >= 1
3114 && let HandlerAction::Escalate { human } = &binding.action
3115 {
3116 let human_path = hook_child_path(step_path, hook, current, human);
3117 if self.human.contains_key(&instance_key(&human_path)) {
3118 // A request exists for the current trigger: it fully
3119 // governs (settle its response, its timeout, or re-await).
3120 return self
3121 .run_hook_human(frame, step_path, hook, current, human)
3122 .await;
3123 }
3124 }
3125
3126 let trigger = current + 1;
3127 if trigger > u64::from(binding.max_triggers) {
3128 return Ok(Consulted::None);
3129 }
3130 self.hook_triggers.insert(counter_key, trigger);
3131 let disposition = match &binding.action {
3132 HandlerAction::Retry { .. } => "retry",
3133 HandlerAction::Continue => "continue",
3134 HandlerAction::Abort => "abort",
3135 HandlerAction::Escalate { .. } => "escalate",
3136 HandlerAction::Repair { .. } => "repair",
3137 };
3138 self.append(
3139 step_path,
3140 &RunLogPayload::HandlerTriggered {
3141 hook,
3142 trigger,
3143 disposition: Some(disposition.to_owned()),
3144 },
3145 )?;
3146
3147 match &binding.action {
3148 HandlerAction::Retry { policy } => Ok(Consulted::Retry(policy.clone())),
3149 HandlerAction::Continue => Ok(Consulted::Continue),
3150 HandlerAction::Abort => Ok(Consulted::Abort),
3151 HandlerAction::Escalate { human } => {
3152 self.run_hook_human(frame, step_path, hook, trigger, human)
3153 .await
3154 }
3155 HandlerAction::Repair { flow_ref } => {
3156 self.run_repair(frame, step_path, hook, trigger, flow_ref)
3157 .await
3158 }
3159 }
3160 }
3161
3162 /// Runs (or settles) an escalate hook human. Hook humans are not body
3163 /// steps: they open no span and seed no frame verdict — their ruling
3164 /// is returned to the consultation site, which supersedes the host
3165 /// verdict and cites the canonical settlement evidence document
3166 /// minted here (06 §6; the request/response ledger pair is the join).
3167 async fn run_hook_human(
3168 &mut self,
3169 frame: &mut FrameState<'a>,
3170 step_path: &RunPath,
3171 hook: HandlerHook,
3172 trigger: u64,
3173 human: &'a HumanStepIR,
3174 ) -> Result<Consulted, RunnerError> {
3175 let human_path = hook_child_path(step_path, hook, trigger, human);
3176 let key = instance_key(&human_path);
3177 if let Some(fact) = self.human.get(&key) {
3178 let fact = fact.clone();
3179 if let Some(response) = fact.final_response.clone() {
3180 // Consume the settlement: a ruling governs exactly once —
3181 // a later consult on the same host walks a *new* trigger
3182 // (or exhausts the budget), never re-reads this answer.
3183 self.human.remove(&key);
3184 // Every settlement materializes the canonical evidence
3185 // document (06 §6); the Escalated superseding verdict
3186 // cites it below. Non-verdict dispositions (Repaired/
3187 // Abort) keep it durable in the library with the
3188 // request/response pair as the join.
3189 let asset =
3190 self.put_human_evidence(&human_path, &fact, Some(&response), "response")?;
3191 return Ok(match map_escalate_response(human, &response) {
3192 Consulted::Escalated {
3193 status, summary, ..
3194 } => Consulted::Escalated {
3195 status,
3196 summary,
3197 evidence: Some(asset),
3198 },
3199 other => other,
3200 });
3201 }
3202 match fact.deadline_at_ms {
3203 Some(deadline) if self.now() > deadline => {
3204 // Lazy timeout settlement: unknown, fixed (onTimeout);
3205 // consumed like any settlement — with the canonical
3206 // evidence document (06 §6, actor null on timeout).
3207 self.human.remove(&key);
3208 let asset = self.put_human_evidence(&human_path, &fact, None, "timeout")?;
3209 return Ok(Consulted::Escalated {
3210 status: VerdictStatus::Unknown,
3211 summary: format!(
3212 "escalate human '{}' timed out (deadline watermark passed): unknown",
3213 human.base.step_id
3214 ),
3215 evidence: Some(asset),
3216 });
3217 }
3218 _ => {
3219 return Ok(Consulted::Pending(HumanPending {
3220 run_path: fact.run_path.clone(),
3221 request_id: fact.request_id.clone(),
3222 purpose: HumanPurpose::Step,
3223 mode: Some(human.mode),
3224 prompt: fact.prompt.clone(),
3225 deadline_at_ms: fact.deadline_at_ms,
3226 }));
3227 }
3228 }
3229 }
3230
3231 // First encounter: materialize presents in the host frame's scope
3232 // and freeze; fsync-before-notify discipline as everywhere.
3233 let scope = frame.scope(&self.env, None);
3234 let mut items = Vec::with_capacity(human.presents.len());
3235 for expr in &human.presents {
3236 match pointlock_expr::eval(expr, &scope) {
3237 Ok(value) => items.push(value),
3238 // A failing presents expression degrades to an empty
3239 // exhibit — the request still goes out (the human can
3240 // rule without exhibits; principle 8 over strictness).
3241 Err(_) => items.push(Value::Null),
3242 }
3243 }
3244 let request_id = uuid::Uuid::new_v4().to_string();
3245 let deadline_at_ms = self.now().saturating_add(human.timeout_ms);
3246 self.append(
3247 &human_path,
3248 &RunLogPayload::HumanRequested {
3249 request_id: request_id.clone(),
3250 purpose: HumanPurpose::Step,
3251 mode: Some(human.mode),
3252 prompt: human.prompt.clone(),
3253 presents: Value::Array(items),
3254 decisions: human.decisions.clone(),
3255 output_schema: human.output_schema.clone(),
3256 deadline_at_ms: Some(deadline_at_ms),
3257 },
3258 )?;
3259 Ok(Consulted::Pending(HumanPending {
3260 run_path: human_path,
3261 request_id,
3262 purpose: HumanPurpose::Step,
3263 mode: Some(human.mode),
3264 prompt: human.prompt.clone(),
3265 deadline_at_ms: Some(deadline_at_ms),
3266 }))
3267 }
3268
3269 /// Runs a repair subflow under the hook frame (a call frame without a
3270 /// host call step — spine §9). Repair flows take no caller inputs
3271 /// (their params materialize from declared defaults, 06 §7.5 binding-
3272 /// flow pattern); they yield no data (R10) — only their flow verdict
3273 /// comes back as the disposition signal.
3274 async fn run_repair(
3275 &mut self,
3276 frame: &mut FrameState<'a>,
3277 step_path: &RunPath,
3278 hook: HandlerHook,
3279 trigger: u64,
3280 flow_ref: &'a pointlock_ir::FlowRef,
3281 ) -> Result<Consulted, RunnerError> {
3282 let callee = self.flows.callee(flow_ref);
3283 let mut repair_path = step_path.clone();
3284 repair_path.push(hook_frame(hook, trigger));
3285 repair_path.push(PathFrame::Call {
3286 step_id: None,
3287 callee_flow_id: callee.flow_id.clone(),
3288 callee_ir_hash: callee.ir_hash.clone(),
3289 });
3290 // Defaults-only inbound materialization.
3291 let params = match call_inputs_gate(callee, Map::new()) {
3292 Ok(params) => params,
3293 // A repair flow whose params cannot materialize from defaults
3294 // is a repair failure; its detail is compiler-diagnosable.
3295 Err(_) => return Ok(Consulted::RepairFailed),
3296 };
3297 self.append(
3298 &repair_path,
3299 &RunLogPayload::CallFramePushed {
3300 frame: CallFrame {
3301 flow_id: callee.flow_id.clone(),
3302 ir_hash: callee.ir_hash.clone(),
3303 call_step_id: None,
3304 inputs_snapshot: Value::Object(params.clone()),
3305 vars: BTreeMap::new(),
3306 iter_stack: Vec::new(),
3307 next_index: 0,
3308 },
3309 rebase: false,
3310 },
3311 )?;
3312 let mut repair_frame =
3313 FrameState::new(callee, repair_path.clone(), params, frame.depth + 1);
3314 let ctl = self
3315 .exec_body(&mut repair_frame, repair_path.clone(), &callee.body, 0)
3316 .await?;
3317 match ctl {
3318 Ctl::Continue | Ctl::HaltFail => {}
3319 other => {
3320 // Suspension inside a repair leaves its frame live; the
3321 // resume path refuses live hook frames with a typed error
3322 // (registered M2 limitation) — the ledger stays honest.
3323 return Ok(Consulted::Propagate(other));
3324 }
3325 }
3326 self.append(
3327 &repair_path,
3328 &RunLogPayload::CallFramePopped { outputs: None },
3329 )?;
3330 let verdict = fold_flow_verdict(&repair_frame.fold, callee.verdict_policy);
3331 match (ctl, verdict) {
3332 (Ctl::HaltFail, _) => Ok(Consulted::RepairFailed),
3333 (_, Some(folded)) if folded.status != VerdictStatus::Pass => {
3334 Ok(Consulted::RepairFailed)
3335 }
3336 _ => Ok(Consulted::RepairDone),
3337 }
3338 }
3339
3340 // ─── call steps (07 §1) ─────────────────────────────────────────────────
3341
3342 async fn exec_call(
3343 &mut self,
3344 frame: &mut FrameState<'a>,
3345 call_path: RunPath,
3346 step: &'a CallStepIR,
3347 ) -> Result<Ctl, RunnerError> {
3348 let step_id = step.base.step_id.clone();
3349 let key = instance_key(&call_path);
3350 let callee: &'a FlowIR = self.flows.callee(&step.flow_ref);
3351 // Runtime defense line for maxCallDepth (the load check already
3352 // bounds the static closure; this guards the walk itself).
3353 if frame.depth + 1 > MAX_CALL_DEPTH {
3354 return Err(RunnerError::CallDepthExceeded {
3355 depth: frame.depth + 1,
3356 max: MAX_CALL_DEPTH,
3357 });
3358 }
3359
3360 // Ready: call-by-value inputs snapshot. A suspension-opened span
3361 // reuses the archived snapshot (already gated) — never
3362 // re-evaluated (spine §6.6, 07 §5.2 corollary).
3363 let archived = self.open_span_inputs(&key);
3364 let gated = match archived {
3365 Some(Value::Object(map)) => map,
3366 Some(other) => {
3367 // A non-object archived snapshot is a ledger anomaly; the
3368 // honest disposition is a bind-class failure.
3369 self.enter_step(&call_path, &step.base, other)?;
3370 return self
3371 .settle_error(
3372 frame,
3373 &call_path,
3374 &step_id,
3375 VerdictStatus::Fail,
3376 "archived call inputs snapshot is not an object".to_owned(),
3377 )
3378 .await;
3379 }
3380 None => {
3381 // Evaluate each input expression in the *caller* scope.
3382 let scope = frame.scope(&self.env, None);
3383 let mut inputs = Map::new();
3384 let mut eval_error = None;
3385 for (name, expr) in step.inputs.iter() {
3386 match pointlock_expr::eval(expr, &scope) {
3387 Ok(value) => {
3388 inputs.insert(name.as_str().to_owned(), value);
3389 }
3390 Err(error) => {
3391 eval_error = Some(format!("input '{name}' evaluation failed: {error}"));
3392 break;
3393 }
3394 }
3395 }
3396 if let Some(message) = eval_error {
3397 self.enter_step(&call_path, &step.base, Value::Null)?;
3398 return self
3399 .settle_error(
3400 frame,
3401 &call_path,
3402 &step_id,
3403 VerdictStatus::Fail,
3404 format!("call inputs failed [bind_arguments_invalid]: {message}"),
3405 )
3406 .await;
3407 }
3408 // Inbound gate: defaults + per-param schema validation
3409 // (07 §1.1 — runtime re-check; failure classifies
3410 // bind_arguments_invalid, no retry).
3411 match call_inputs_gate(callee, inputs.clone()) {
3412 Ok(gated) => gated,
3413 Err(message) => {
3414 self.enter_step(&call_path, &step.base, Value::Object(inputs))?;
3415 return self
3416 .settle_error(
3417 frame,
3418 &call_path,
3419 &step_id,
3420 VerdictStatus::Fail,
3421 format!("call inputs failed [bind_arguments_invalid]: {message}"),
3422 )
3423 .await;
3424 }
3425 }
3426 }
3427 };
3428 self.enter_step(&call_path, &step.base, Value::Object(gated.clone()))?;
3429
3430 if let Some(ctl) = self.probe_or_note(frame, &call_path, &step.base).await? {
3431 return Ok(ctl);
3432 }
3433
3434 // Frame push (07 §3.1 frame-transfer materialization point) —
3435 // unless a previous segment already pushed it and we are resuming
3436 // back into the live frame. Re-entering one whose callee pin moved
3437 // is announced as a `rebase`, so `frames` names the callee actually
3438 // executing rather than the one the crashed segment entered
3439 // (07 §5.2 case (a)); the fold updates that frame's `irHash` in
3440 // place and touches nothing else.
3441 let open_pin = self.live_frames.remove(&key);
3442 let rebase = match &open_pin {
3443 None => false,
3444 Some(pin) => *pin != callee.ir_hash,
3445 };
3446 if open_pin.is_none() || rebase {
3447 self.append(
3448 &call_path,
3449 &RunLogPayload::CallFramePushed {
3450 frame: CallFrame {
3451 flow_id: callee.flow_id.clone(),
3452 ir_hash: callee.ir_hash.clone(),
3453 call_step_id: Some(step_id.clone()),
3454 inputs_snapshot: Value::Object(gated.clone()),
3455 vars: BTreeMap::new(),
3456 iter_stack: Vec::new(),
3457 next_index: 0,
3458 },
3459 rebase,
3460 },
3461 )?;
3462 }
3463
3464 // The callee body runs in a fresh scope: params = inputs, env
3465 // passes through read-only, the caller's steps/vars are invisible
3466 // (07 §1.2 — hard boundary).
3467 let mut callee_frame = FrameState::new(callee, call_path.clone(), gated, frame.depth + 1);
3468 let ctl = self
3469 .exec_body(&mut callee_frame, call_path.clone(), &callee.body, 0)
3470 .await?;
3471 match ctl {
3472 Ctl::Continue | Ctl::HaltFail => {}
3473 Ctl::Abort => {
3474 // Unwind the frame so the ledger stays balanced; an
3475 // aborted run makes no semantic claim.
3476 self.append(
3477 &call_path,
3478 &RunLogPayload::CallFramePopped { outputs: None },
3479 )?;
3480 self.append(
3481 &call_path,
3482 &RunLogPayload::StepExited {
3483 provider_state_summary: None,
3484 state: StepState::Aborted,
3485 output: None,
3486 localized: Vec::new(),
3487 localization_gaps: Vec::new(),
3488 },
3489 )?;
3490 return Ok(Ctl::Abort);
3491 }
3492 // Suspension/blocking leaves the frame live: resume falls back
3493 // into the exact position (07 §4.6), never restarts the frame.
3494 other => return Ok(other),
3495 }
3496
3497 // Outbound gate: declared outputs evaluated in the *callee* scope,
3498 // schema-validated, snapshotted (07 §1.1). Only a completed body
3499 // has outputs; a halted callee pops without them.
3500 let outputs = if matches!(ctl, Ctl::Continue) {
3501 match self.call_outputs_gate(callee, &callee_frame) {
3502 Ok(outputs) => Some(outputs),
3503 Err(message) => {
3504 self.append(
3505 &call_path,
3506 &RunLogPayload::CallFramePopped { outputs: None },
3507 )?;
3508 return self
3509 .settle_error(
3510 frame,
3511 &call_path,
3512 &step_id,
3513 VerdictStatus::Fail,
3514 format!("callee outputs failed the outbound gate: {message}"),
3515 )
3516 .await;
3517 }
3518 }
3519 } else {
3520 None
3521 };
3522 self.append(
3523 &call_path,
3524 &RunLogPayload::CallFramePopped {
3525 outputs: outputs.clone(),
3526 },
3527 )?;
3528
3529 // The call step's verdict *is* the callee's flow verdict
3530 // (spine §6.3); `degraded` propagates verbatim and participates in
3531 // the caller's fold.
3532 let callee_verdict = fold_flow_verdict(&callee_frame.fold, callee.verdict_policy);
3533 let mut status = None;
3534 let mut verdict_seq = None;
3535 if let Some(folded) = callee_verdict {
3536 let folded = FoldedVerdict {
3537 status: folded.status,
3538 degraded: folded.degraded,
3539 summary: format!(
3540 "callee '{}' flow verdict: {}",
3541 callee.flow_id, folded.summary
3542 ),
3543 };
3544 let seq = self
3545 .record_step_verdict(
3546 &call_path,
3547 &folded,
3548 Vec::new(),
3549 None,
3550 EvidenceManifest::default(),
3551 )
3552 .await?;
3553 verdict_seq = Some(seq);
3554 frame.seed_verdict(&step_id, folded.status, folded.degraded);
3555 status = Some(folded.status);
3556 }
3557
3558 // Handler consultation on the call step's own verdict (the callee
3559 // handled its internal failures itself; this hook is the caller's
3560 // policy about the aggregate). Re-invocation dispositions (retry /
3561 // repair-then-re-call) need the 07 §1 attempt-framed full re-call
3562 // — a typed M2 refusal, registered.
3563 if matches!(
3564 status,
3565 Some(VerdictStatus::Fail) | Some(VerdictStatus::Unknown)
3566 ) {
3567 let hook = if status == Some(VerdictStatus::Fail) {
3568 HandlerHook::OnFail
3569 } else {
3570 HandlerHook::OnUnknown
3571 };
3572 match self
3573 .consult_hook(frame, &call_path, step.base.handlers.as_deref(), hook, None)
3574 .await?
3575 {
3576 Consulted::None | Consulted::RepairFailed => {}
3577 Consulted::Continue => {
3578 self.append(
3579 &call_path,
3580 &RunLogPayload::StepExited {
3581 provider_state_summary: None,
3582 state: StepState::Judged,
3583 output: outputs.clone(),
3584 localized: Vec::new(),
3585 localization_gaps: Vec::new(),
3586 },
3587 )?;
3588 if let Some(outputs) = outputs {
3589 frame.outputs.insert(step_id.as_str().to_owned(), outputs);
3590 }
3591 return Ok(Ctl::Continue);
3592 }
3593 Consulted::Abort => {
3594 self.append(
3595 &call_path,
3596 &RunLogPayload::StepExited {
3597 provider_state_summary: None,
3598 state: StepState::Aborted,
3599 output: None,
3600 localized: Vec::new(),
3601 localization_gaps: Vec::new(),
3602 },
3603 )?;
3604 return Ok(Ctl::Abort);
3605 }
3606 Consulted::Escalated {
3607 status: ruled,
3608 summary: ruled_summary,
3609 evidence: ruling_evidence,
3610 } => {
3611 let folded = FoldedVerdict {
3612 status: ruled,
3613 degraded: false,
3614 summary: ruled_summary,
3615 };
3616 let supersedes = verdict_seq.map(|seq| format!("seq:{seq}"));
3617 let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
3618 let ruled_seq = self
3619 .record_step_verdict(&call_path, &folded, cited, supersedes, manifest)
3620 .await?;
3621 self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
3622 frame.reseed_last(&step_id, ruled, false);
3623 self.append(
3624 &call_path,
3625 &RunLogPayload::StepExited {
3626 provider_state_summary: None,
3627 state: StepState::Judged,
3628 output: outputs.clone(),
3629 localized: Vec::new(),
3630 localization_gaps: Vec::new(),
3631 },
3632 )?;
3633 if let Some(outputs) = outputs {
3634 frame.outputs.insert(step_id.as_str().to_owned(), outputs);
3635 }
3636 return Ok(if ruled == VerdictStatus::Fail {
3637 Ctl::HaltFail
3638 } else {
3639 Ctl::Continue
3640 });
3641 }
3642 Consulted::Retry(_) | Consulted::Repaired | Consulted::RepairDone => {
3643 return Err(RunnerError::M0Unsupported {
3644 detail: format!(
3645 "handler re-invocation dispositions on call step '{step_id}' need the 07 §1 \
3646 attempt-framed full re-call — not in the M2 subset \
3647 (escalate/continue/abort are supported)"
3648 ),
3649 });
3650 }
3651 Consulted::Pending(pending) => {
3652 return Ok(Ctl::AwaitHuman(pending));
3653 }
3654 Consulted::Propagate(inner) => return Ok(inner),
3655 }
3656 }
3657
3658 self.append(
3659 &call_path,
3660 &RunLogPayload::StepExited {
3661 provider_state_summary: None,
3662 state: StepState::Judged,
3663 output: outputs.clone(),
3664 localized: Vec::new(),
3665 localization_gaps: Vec::new(),
3666 },
3667 )?;
3668 if let Some(outputs) = outputs {
3669 frame.outputs.insert(step_id.as_str().to_owned(), outputs);
3670 }
3671 if matches!(ctl, Ctl::HaltFail) || status == Some(VerdictStatus::Fail) {
3672 Ok(Ctl::HaltFail)
3673 } else {
3674 Ok(Ctl::Continue)
3675 }
3676 }
3677
3678 /// The outbound gate: callee `outputs` declarations evaluated over the
3679 /// callee frame's scope and validated against their schemas.
3680 fn call_outputs_gate(
3681 &self,
3682 callee: &FlowIR,
3683 callee_frame: &FrameState<'a>,
3684 ) -> Result<Value, String> {
3685 let scope = callee_frame.scope(&self.env, None);
3686 let mut outputs = Map::new();
3687 for decl in &callee.outputs {
3688 let value = pointlock_expr::eval(&decl.from, &scope)
3689 .map_err(|error| format!("output '{}' evaluation failed: {error}", decl.name))?;
3690 jsonschema::validate(decl.schema.as_value(), &value)
3691 .map_err(|error| format!("output '{}' failed its schema: {error}", decl.name))?;
3692 outputs.insert(decl.name.as_str().to_owned(), value);
3693 }
3694 Ok(Value::Object(outputs))
3695 }
3696
3697 // ─── if steps ───────────────────────────────────────────────────────────
3698
3699 async fn exec_if(
3700 &mut self,
3701 frame: &mut FrameState<'a>,
3702 step_path: RunPath,
3703 step: &'a IfStepIR,
3704 ) -> Result<Ctl, RunnerError> {
3705 let step_id = step.base.step_id.clone();
3706 let key = instance_key(&step_path);
3707 // A suspension-opened span reuses the archived branch decision —
3708 // the snapshot rule (spine §6.6) applies to control values too.
3709 let cond_value = match self.open_span_inputs(&key) {
3710 Some(archived) => archived.get("cond").cloned().unwrap_or(Value::Null),
3711 None => {
3712 let scope = frame.scope(&self.env, None);
3713 match pointlock_expr::eval(&step.cond, &scope) {
3714 Ok(value) => value,
3715 Err(error) => {
3716 self.enter_step(&step_path, &step.base, Value::Null)?;
3717 return self
3718 .settle_error(
3719 frame,
3720 &step_path,
3721 &step_id,
3722 VerdictStatus::Fail,
3723 format!("if cond evaluation failed: {error}"),
3724 )
3725 .await;
3726 }
3727 }
3728 }
3729 };
3730 // Strict boolean (02 §4.5): anything else is a compiler/expression
3731 // bug signal — step fail, no branch is taken.
3732 let Some(cond) = cond_value.as_bool() else {
3733 self.enter_step(
3734 &step_path,
3735 &step.base,
3736 serde_json::json!({ "cond": cond_value }),
3737 )?;
3738 return self
3739 .settle_error(
3740 frame,
3741 &step_path,
3742 &step_id,
3743 VerdictStatus::Fail,
3744 format!("if cond evaluated to a non-boolean value: {cond_value}"),
3745 )
3746 .await;
3747 };
3748 self.enter_step(&step_path, &step.base, serde_json::json!({ "cond": cond }))?;
3749
3750 if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
3751 return Ok(ctl);
3752 }
3753
3754 let empty: &'a [StepIR] = &[];
3755 let (selected, unselected): (&'a [StepIR], &'a [StepIR]) = if cond {
3756 (&step.then, step.r#else.as_deref().unwrap_or(empty))
3757 } else {
3758 (step.r#else.as_deref().unwrap_or(empty), &step.then)
3759 };
3760 // The unselected branch's steps each leave an
3761 // entered(null)/exited(skipped) pair — ledger completeness (the
3762 // blocked precedent). Recorded before the selected branch runs so
3763 // a mid-branch suspension leaves a complete account.
3764 self.stash_open_span_summaries().await;
3765 self.record_pairs(&step_path, unselected, StepState::Skipped)?;
3766
3767 let ctl = self
3768 .exec_body(frame, step_path.clone(), selected, 0)
3769 .await?;
3770 match ctl {
3771 Ctl::Continue | Ctl::HaltFail => {
3772 // Containers yield no verdict of their own (R4): the exit
3773 // closes the span; child verdicts already folded.
3774 self.append(
3775 &step_path,
3776 &RunLogPayload::StepExited {
3777 provider_state_summary: None,
3778 state: StepState::Judged,
3779 output: None,
3780 localized: Vec::new(),
3781 localization_gaps: Vec::new(),
3782 },
3783 )?;
3784 Ok(ctl)
3785 }
3786 Ctl::Abort => {
3787 self.append(
3788 &step_path,
3789 &RunLogPayload::StepExited {
3790 provider_state_summary: None,
3791 state: StepState::Aborted,
3792 output: None,
3793 localized: Vec::new(),
3794 localization_gaps: Vec::new(),
3795 },
3796 )?;
3797 Ok(Ctl::Abort)
3798 }
3799 other => Ok(other),
3800 }
3801 }
3802
3803 // ─── foreach steps ──────────────────────────────────────────────────────
3804
3805 async fn exec_foreach(
3806 &mut self,
3807 frame: &mut FrameState<'a>,
3808 step_path: RunPath,
3809 step: &'a ForeachStepIR,
3810 ) -> Result<Ctl, RunnerError> {
3811 let step_id = step.base.step_id.clone();
3812 let key = instance_key(&step_path);
3813 let items_value = match self.open_span_inputs(&key) {
3814 Some(archived) => archived.get("items").cloned().unwrap_or(Value::Null),
3815 None => {
3816 let scope = frame.scope(&self.env, None);
3817 match pointlock_expr::eval(&step.items, &scope) {
3818 Ok(value) => value,
3819 Err(error) => {
3820 self.enter_step(&step_path, &step.base, Value::Null)?;
3821 return self
3822 .settle_error(
3823 frame,
3824 &step_path,
3825 &step_id,
3826 VerdictStatus::Fail,
3827 format!("foreach items evaluation failed: {error}"),
3828 )
3829 .await;
3830 }
3831 }
3832 }
3833 };
3834 let Some(items) = items_value.as_array().cloned() else {
3835 self.enter_step(
3836 &step_path,
3837 &step.base,
3838 serde_json::json!({ "items": items_value, "as": step.r#as.as_str() }),
3839 )?;
3840 return self
3841 .settle_error(
3842 frame,
3843 &step_path,
3844 &step_id,
3845 VerdictStatus::Fail,
3846 format!("foreach items evaluated to a non-array value: {items_value}"),
3847 )
3848 .await;
3849 };
3850 // The snapshot carries `{ items, as }`: the position authority for
3851 // the positional (index-keyed) resume regime and the fold's
3852 // IterState carrier.
3853 self.enter_step(
3854 &step_path,
3855 &step.base,
3856 serde_json::json!({ "items": items, "as": step.r#as.as_str() }),
3857 )?;
3858
3859 if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
3860 return Ok(ctl);
3861 }
3862
3863 for (index, item) in items.iter().enumerate() {
3864 frame
3865 .iters
3866 .push((step.r#as.as_str().to_owned(), item.clone()));
3867 let mut iter_prefix = step_path.clone();
3868 iter_prefix.push(PathFrame::Iteration {
3869 index: index as u64,
3870 key: None,
3871 });
3872 let ctl = self.exec_body(frame, iter_prefix, &step.body, 0).await?;
3873 frame.iters.pop();
3874 match ctl {
3875 Ctl::Continue => {}
3876 Ctl::HaltFail => {
3877 // The failing iteration already blocked its own tail;
3878 // later iterations never materialize as instances.
3879 self.append(
3880 &step_path,
3881 &RunLogPayload::StepExited {
3882 provider_state_summary: None,
3883 state: StepState::Judged,
3884 output: None,
3885 localized: Vec::new(),
3886 localization_gaps: Vec::new(),
3887 },
3888 )?;
3889 return Ok(Ctl::HaltFail);
3890 }
3891 Ctl::Abort => {
3892 self.append(
3893 &step_path,
3894 &RunLogPayload::StepExited {
3895 provider_state_summary: None,
3896 state: StepState::Aborted,
3897 output: None,
3898 localized: Vec::new(),
3899 localization_gaps: Vec::new(),
3900 },
3901 )?;
3902 return Ok(Ctl::Abort);
3903 }
3904 other => return Ok(other),
3905 }
3906 }
3907 self.append(
3908 &step_path,
3909 &RunLogPayload::StepExited {
3910 provider_state_summary: None,
3911 state: StepState::Judged,
3912 output: None,
3913 localized: Vec::new(),
3914 localization_gaps: Vec::new(),
3915 },
3916 )?;
3917 Ok(Ctl::Continue)
3918 }
3919
3920 // ─── let steps ──────────────────────────────────────────────────────────
3921
3922 async fn exec_let(
3923 &mut self,
3924 frame: &mut FrameState<'a>,
3925 step_path: RunPath,
3926 step: &'a LetStepIR,
3927 ) -> Result<Ctl, RunnerError> {
3928 let step_id = step.base.step_id.clone();
3929 let key = instance_key(&step_path);
3930 let evaluated = match self.open_span_inputs(&key) {
3931 // The archived snapshot *is* the bindings product (pure,
3932 // deterministic) — never re-evaluated on resume.
3933 Some(Value::Object(map)) => map,
3934 Some(_) | None => {
3935 let scope = frame.scope(&self.env, None);
3936 let mut evaluated = Map::new();
3937 let mut error = None;
3938 for (name, expr) in step.bindings.iter() {
3939 // SSA single assignment: rebinding is refused by the
3940 // compiler check phase; this is the runtime defense
3941 // line against hand-built IR.
3942 if frame.vars.contains_key(name.as_str()) {
3943 error = Some(format!(
3944 "binding '{name}' rebinds an existing var (SSA single assignment)"
3945 ));
3946 break;
3947 }
3948 match pointlock_expr::eval(expr, &scope) {
3949 Ok(value) => {
3950 evaluated.insert(name.as_str().to_owned(), value);
3951 }
3952 Err(eval_error) => {
3953 error =
3954 Some(format!("binding '{name}' evaluation failed: {eval_error}"));
3955 break;
3956 }
3957 }
3958 }
3959 if let Some(message) = error {
3960 self.enter_step(&step_path, &step.base, Value::Null)?;
3961 return self
3962 .settle_error(
3963 frame,
3964 &step_path,
3965 &step_id,
3966 VerdictStatus::Fail,
3967 format!("let bindings failed: {message}"),
3968 )
3969 .await;
3970 }
3971 evaluated
3972 }
3973 };
3974 // The ready snapshot carries the evaluated bindings — the resume
3975 // walk re-seeds `vars.*` from exactly this carrier.
3976 self.enter_step(&step_path, &step.base, Value::Object(evaluated.clone()))?;
3977 if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
3978 return Ok(ctl);
3979 }
3980 self.append(
3981 &step_path,
3982 &RunLogPayload::StepExited {
3983 provider_state_summary: None,
3984 state: StepState::Judged,
3985 output: None,
3986 localized: Vec::new(),
3987 localization_gaps: Vec::new(),
3988 },
3989 )?;
3990 for (name, value) in evaluated {
3991 frame.vars.insert(name, value);
3992 }
3993 Ok(Ctl::Continue)
3994 }
3995
3996 // ─── assert steps ───────────────────────────────────────────────────────
3997
3998 async fn exec_assert(
3999 &mut self,
4000 frame: &mut FrameState<'a>,
4001 step_path: RunPath,
4002 step: &'a AssertStepIR,
4003 ) -> Result<Ctl, RunnerError> {
4004 let step_id = step.base.step_id.clone();
4005 // An assert step resolves no input expressions; the span still
4006 // opens with an explicitly-null snapshot.
4007 self.enter_step(&step_path, &step.base, Value::Null)?;
4008 if let Some(ctl) = self.probe_or_note(frame, &step_path, &step.base).await? {
4009 return Ok(ctl);
4010 }
4011 let needs = VerifyNeeds::of(&step.assertions);
4012 let mut last_verdict_seq: Option<u64> = None;
4013 let mut seeded = false;
4014 let mut active_retry: Option<(RetryPolicy, u32)> = None;
4015 loop {
4016 let material = match &step.observe {
4017 // Fresh capture: one `session.observe`, localized through the
4018 // same observing pipeline as action observations.
4019 ObservationSource::Fresh(_) => {
4020 let mut anchor = step_path.clone();
4021 anchor.push(PathFrame::Phase {
4022 phase: Phase::Observe,
4023 });
4024 self.fresh_material(&needs, &anchor).await?
4025 }
4026 // Archive reuse: the referenced action step's localized
4027 // observation material — zero device I/O, offline
4028 // re-judgeable by construction.
4029 ObservationSource::FromStep(from) => {
4030 let which = from.which;
4031 match frame.observed.get(from.from_step.as_str()) {
4032 None => ObserveMaterial::absent(&format!(
4033 "step '{}' has no archived observation in this frame",
4034 from.from_step
4035 )),
4036 Some(observed) => {
4037 let wanted = match which {
4038 pointlock_ir::ObservationWhich::After => &observed.after_id,
4039 pointlock_ir::ObservationWhich::Before => &observed.before_id,
4040 };
4041 match wanted {
4042 None => ObserveMaterial::absent(&format!(
4043 "step '{}' recorded no {:?} observation",
4044 from.from_step, which
4045 )),
4046 Some(observation_id) => {
4047 match observed
4048 .observations
4049 .iter()
4050 .find(|record| record.observation_id == *observation_id)
4051 {
4052 None => ObserveMaterial::absent(
4053 "the referenced observation was never localized",
4054 ),
4055 Some(record) => {
4056 material_from_observation(self.store, record)
4057 }
4058 }
4059 }
4060 }
4061 }
4062 }
4063 }
4064 };
4065 let scope = frame.scope(&self.env, None);
4066 let mut outcomes = Vec::with_capacity(step.assertions.len());
4067 let mut degraded_verify = false;
4068 for assertion in &step.assertions {
4069 let evaluated = match &assertion.predicate {
4070 PredicateIR::Expr { expr } => EvaluatedAssertion {
4071 record: eval_expr_assertion(assertion, expr, &scope),
4072 degraded_verify: false,
4073 },
4074 _ => {
4075 eval_observed_assertion(assertion, &material, self.vision.as_deref()).await
4076 }
4077 };
4078 degraded_verify |= evaluated.degraded_verify;
4079 outcomes.push(evaluated.record);
4080 }
4081 for outcome in &outcomes {
4082 let mut path = step_path.clone();
4083 path.push(PathFrame::Phase {
4084 phase: Phase::Assert,
4085 });
4086 path.push(PathFrame::Assertion {
4087 assert_id: outcome.assert_id.clone(),
4088 });
4089 self.append(
4090 &path,
4091 &RunLogPayload::AssertionEvaluated {
4092 outcome: outcome.clone(),
4093 },
4094 )?;
4095 }
4096 // Assert steps declare ≥ 1 assertion, so a verdict always folds.
4097 let folded =
4098 fold_step_verdict(&outcomes, false, degraded_verify, frame.flow.verdict_policy);
4099 let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
4100 let seq = self
4101 .record_step_verdict(
4102 &step_path,
4103 &folded,
4104 Vec::new(),
4105 supersedes,
4106 EvidenceManifest::default(),
4107 )
4108 .await?;
4109 last_verdict_seq = Some(seq);
4110 if seeded {
4111 frame.reseed_last(&step_id, folded.status, folded.degraded);
4112 } else {
4113 frame.seed_verdict(&step_id, folded.status, folded.degraded);
4114 seeded = true;
4115 }
4116 if folded.status == VerdictStatus::Pass {
4117 self.append(
4118 &step_path,
4119 &RunLogPayload::StepExited {
4120 provider_state_summary: None,
4121 state: StepState::Judged,
4122 output: None,
4123 localized: Vec::new(),
4124 localization_gaps: Vec::new(),
4125 },
4126 )?;
4127 return Ok(Ctl::Continue);
4128 }
4129
4130 // In-force handler retry budget (observe + assert re-entry is
4131 // readonly by construction — always replay-safe).
4132 if let Some((policy, used)) = active_retry.take()
4133 && used < policy.max_attempts
4134 {
4135 self.backoff_policy(&policy, used).await;
4136 active_retry = Some((policy, used + 1));
4137 continue;
4138 }
4139
4140 let hook = if folded.status == VerdictStatus::Fail {
4141 HandlerHook::OnFail
4142 } else {
4143 HandlerHook::OnUnknown
4144 };
4145 match self
4146 .consult_hook(frame, &step_path, step.base.handlers.as_deref(), hook, None)
4147 .await?
4148 {
4149 Consulted::None | Consulted::RepairFailed => {
4150 self.append(
4151 &step_path,
4152 &RunLogPayload::StepExited {
4153 provider_state_summary: None,
4154 state: StepState::Judged,
4155 output: None,
4156 localized: Vec::new(),
4157 localization_gaps: Vec::new(),
4158 },
4159 )?;
4160 return Ok(if folded.status == VerdictStatus::Fail {
4161 Ctl::HaltFail
4162 } else {
4163 Ctl::Continue
4164 });
4165 }
4166 Consulted::Continue => {
4167 self.append(
4168 &step_path,
4169 &RunLogPayload::StepExited {
4170 provider_state_summary: None,
4171 state: StepState::Judged,
4172 output: None,
4173 localized: Vec::new(),
4174 localization_gaps: Vec::new(),
4175 },
4176 )?;
4177 return Ok(Ctl::Continue);
4178 }
4179 Consulted::Abort => {
4180 self.append(
4181 &step_path,
4182 &RunLogPayload::StepExited {
4183 provider_state_summary: None,
4184 state: StepState::Aborted,
4185 output: None,
4186 localized: Vec::new(),
4187 localization_gaps: Vec::new(),
4188 },
4189 )?;
4190 return Ok(Ctl::Abort);
4191 }
4192 Consulted::Escalated {
4193 status: ruled,
4194 summary: ruled_summary,
4195 evidence: ruling_evidence,
4196 } => {
4197 let folded = FoldedVerdict {
4198 status: ruled,
4199 degraded: false,
4200 summary: ruled_summary,
4201 };
4202 let supersedes = last_verdict_seq.map(|seq| format!("seq:{seq}"));
4203 let (cited, manifest) = escalate_verdict_material(&ruling_evidence);
4204 let ruled_seq = self
4205 .record_step_verdict(&step_path, &folded, cited, supersedes, manifest)
4206 .await?;
4207 self.link_ruling_evidence(ruled_seq, &ruling_evidence)?;
4208 frame.reseed_last(&step_id, ruled, false);
4209 self.append(
4210 &step_path,
4211 &RunLogPayload::StepExited {
4212 provider_state_summary: None,
4213 state: StepState::Judged,
4214 output: None,
4215 localized: Vec::new(),
4216 localization_gaps: Vec::new(),
4217 },
4218 )?;
4219 return Ok(if ruled == VerdictStatus::Fail {
4220 Ctl::HaltFail
4221 } else {
4222 Ctl::Continue
4223 });
4224 }
4225 Consulted::Retry(policy) => {
4226 self.backoff_policy(&policy, 0).await;
4227 active_retry = Some((policy, 1));
4228 }
4229 Consulted::Repaired | Consulted::RepairDone => {}
4230 Consulted::Pending(pending) => {
4231 return Ok(Ctl::AwaitHuman(pending));
4232 }
4233 Consulted::Propagate(ctl) => return Ok(ctl),
4234 }
4235 // Loop: re-observe and re-evaluate (fresh material each round;
4236 // fromStep archives are stable, so a re-round only makes sense
4237 // after a repair — both are honest re-judgements on new/declared
4238 // world state).
4239 }
4240 }
4241
4242 // ─── observing / localization ───────────────────────────────────────────
4243
4244 /// The observing phase: localize evidence (spine §6.6 — provider-side
4245 /// retention is not guaranteed) and record the before/after
4246 /// observations. Returns the provider asset refs cited by the verdict,
4247 /// the [`ObserveMaterial`] of the *after* observation, and the
4248 /// localized records (the `fromStep` archive of this step).
4249 async fn observing(
4250 &mut self,
4251 step: &'a ActionStepIR,
4252 step_path: &RunPath,
4253 attempt_n: u64,
4254 result: &ActionResult,
4255 settled_seq: u64,
4256 ) -> Result<(Vec<AssetRef>, ObserveMaterial, StepObs, EvidenceManifest), RunnerError> {
4257 let mut observe_path = step_path.clone();
4258 observe_path.push(PathFrame::Attempt { n: attempt_n });
4259 observe_path.push(PathFrame::Phase {
4260 phase: Phase::Observe,
4261 });
4262 let needs = VerifyNeeds::of(&step.assertions);
4263 let mut cited = Vec::new();
4264 let mut material =
4265 ObserveMaterial::absent("the action result carries no after observation");
4266 let mut observed = StepObs {
4267 observations: Vec::new(),
4268 before_id: None,
4269 after_id: None,
4270 };
4271 if let Some(observation) = &result.before {
4272 let record = self
4273 .localize_observation(observation, &mut cited, None)
4274 .await?;
4275 // file-before-row-before-log: the bytes are on disk and indexed
4276 // before this event references them.
4277 self.append(
4278 &observe_path,
4279 &RunLogPayload::ObservationRecorded {
4280 observation: record.clone(),
4281 },
4282 )?;
4283 observed.before_id = Some(record.observation_id.clone());
4284 observed.observations.push(record);
4285 }
4286 if let Some(observation) = &result.after {
4287 material = ObserveMaterial::default();
4288 let record = self
4289 .localize_observation(observation, &mut cited, Some((&needs, &mut material)))
4290 .await?;
4291 self.append(
4292 &observe_path,
4293 &RunLogPayload::ObservationRecorded {
4294 observation: record.clone(),
4295 },
4296 )?;
4297 observed.after_id = Some(record.observation_id.clone());
4298 observed.observations.push(record);
4299 }
4300 let mut manifest = EvidenceManifest::default();
4301 for asset in &result.evidence {
4302 // Bounded manifest (the cited-list cap's sibling): entries
4303 // beyond the cap are recorded as typed gaps — bounded DTOs
4304 // must not silently truncate (spine §10 bounded-render
4305 // discipline).
4306 if manifest.localized.len() >= VERDICT_EVIDENCE_MAX_ENTRIES {
4307 manifest.gaps.push(pointlock_ir::EvidenceGap {
4308 asset: asset.clone(),
4309 reason: format!(
4310 "evidence cap exceeded ({VERDICT_EVIDENCE_MAX_ENTRIES} max per judgment)"
4311 ),
4312 });
4313 continue;
4314 }
4315 // Auxiliary settlement evidence (item ③, 2026-07-18): a
4316 // success joins this judgment's localized manifest (and the
4317 // evidence_ref index); a failure is a TYPED gap on the
4318 // verdict record — never a silent omission (principle 4/R4).
4319 match self.try_localize(asset).await? {
4320 Ok((evidence, _bytes)) => {
4321 self.store.link_evidence(
4322 &self.run_id,
4323 settled_seq,
4324 &asset.id,
4325 &evidence.sha256,
4326 )?;
4327 cited.push(asset.clone());
4328 manifest.localized.push(evidence);
4329 }
4330 Err(reason) => {
4331 manifest.gaps.push(pointlock_ir::EvidenceGap {
4332 asset: asset.clone(),
4333 reason,
4334 });
4335 }
4336 }
4337 }
4338 Ok((cited, material, observed, manifest))
4339 }
4340
4341 /// Localizes one observation's evidence and builds its durable record.
4342 /// Omissions are typed data and pass through verbatim; a localization
4343 /// failure leaves the affected field absent and feeds the dependent
4344 /// verify channel a typed gap — the run is never aborted over it (M2
4345 /// degradation rule; principle 4 routes it to `unknown`).
4346 async fn localize_observation(
4347 &mut self,
4348 observation: &Observation,
4349 cited: &mut Vec<AssetRef>,
4350 mut material: Option<(&VerifyNeeds, &mut ObserveMaterial)>,
4351 ) -> Result<ObservationRecord, RunnerError> {
4352 let screenshot = match &observation.screenshot {
4353 Some(asset) => match self.try_localize(asset).await? {
4354 Ok((evidence, bytes)) => {
4355 cited.push(asset.clone());
4356 if let Some((needs, material)) = material.as_mut()
4357 && needs.vision
4358 {
4359 material.screenshot = Some((bytes, asset.media_type.clone()));
4360 }
4361 Some(evidence)
4362 }
4363 Err(gap) => {
4364 if let Some((needs, material)) = material.as_mut()
4365 && needs.vision
4366 {
4367 material.screenshot_gap = Some(gap);
4368 }
4369 None
4370 }
4371 },
4372 None => {
4373 if let Some((needs, material)) = material.as_mut()
4374 && needs.vision
4375 {
4376 material.screenshot_gap = Some(match observation.screenshot_omission {
4377 Some(reason) => {
4378 format!("screenshot omitted by the provider ({reason:?})")
4379 }
4380 None => "the observation carries no screenshot".to_owned(),
4381 });
4382 }
4383 None
4384 }
4385 };
4386 let mut ui_snapshot_omission = observation.ui_snapshot_omission;
4387 let ui_snapshot = match &observation.ui_snapshot {
4388 Some(snapshot) => {
4389 let wants_tree = matches!(&material, Some((needs, _)) if needs.ui_tree);
4390 let localized = if wants_tree {
4391 self.localize_ui_tree(
4392 observation,
4393 snapshot,
4394 &mut material,
4395 &mut ui_snapshot_omission,
4396 )
4397 .await?
4398 } else {
4399 // No assertion consumes the tree: localize the
4400 // provider-side evidence object as before (retention is
4401 // not guaranteed), without the `ui.snapshot.get` pull.
4402 // A failure is a gap, not an abort.
4403 self.try_localize(&snapshot.evidence)
4404 .await?
4405 .ok()
4406 .map(|(evidence, _bytes)| evidence)
4407 };
4408 // Cite only what actually landed (item ③ review fix): a
4409 // citation the local library cannot serve would be a
4410 // silent gallery omission. Citation granularity is the
4411 // ASSET (the pointer); the byte-exact truth (sha256 +
4412 // localPath of what was actually stored — the canonical
4413 // tree on the wants_tree path) lives on the record's
4414 // EvidenceRef, which is what consumers resolve.
4415 if localized.is_some() {
4416 cited.push(snapshot.evidence.clone());
4417 }
4418 localized
4419 }
4420 None => {
4421 if let Some((needs, material)) = material.as_mut()
4422 && needs.ui_tree
4423 {
4424 material.ui_tree_gap = Some(match observation.ui_snapshot_omission {
4425 Some(reason) => {
4426 format!("uiSnapshot omitted by the provider ({reason:?})")
4427 }
4428 None => "the observation carries no uiSnapshot".to_owned(),
4429 });
4430 }
4431 None
4432 }
4433 };
4434 // Finiteness guard: `scaleFactor` is the only f64 in the durable
4435 // record domain, and serde_json writes a non-finite f64 as `null`
4436 // — which round-trips into a permanent ledger read failure (every
4437 // refold/verify/projection of the run errors). A provider that
4438 // reports a non-finite viewport gets the field honestly absent
4439 // rather than a poisoned ledger (clamping would falsify evidence;
4440 // aborting the run over a cosmetic field would violate the M2
4441 // degradation rule).
4442 let viewport = observation
4443 .viewport
4444 .scale_factor
4445 .is_finite()
4446 .then(|| observation.viewport.clone());
4447 Ok(ObservationRecord {
4448 observation_id: observation.id.clone(),
4449 captured_at_ms: observation.captured_at_ms,
4450 viewport,
4451 screenshot,
4452 screenshot_omission: observation.screenshot_omission,
4453 ui_snapshot,
4454 ui_snapshot_omission,
4455 })
4456 }
4457
4458 /// Pulls the observation's normalized UI tree (`ui.snapshot.get`),
4459 /// localizes the canonical bytes, and feeds the verify-chain material.
4460 /// A typed `Unavailable` — or a provider error on the dereference — is
4461 /// data, not a run abort: it becomes the uiTree channel's gap (unknown
4462 /// propagation), never a fabricated tree.
4463 async fn localize_ui_tree(
4464 &mut self,
4465 observation: &Observation,
4466 snapshot: &pointlock_ir::UiSnapshotRef,
4467 material: &mut Option<(&VerifyNeeds, &mut ObserveMaterial)>,
4468 ui_snapshot_omission: &mut Option<UiSnapshotOmissionReason>,
4469 ) -> Result<Option<EvidenceRef>, RunnerError> {
4470 match self.session.ui_snapshot(&observation.id).await {
4471 Ok(UiSnapshotOutcome::Available { snapshot: tree }) => {
4472 let bytes =
4473 serde_json::to_vec(&tree).expect("a serde_json::Value always serializes");
4474 let put = self.store.put_evidence(&bytes, "application/json")?;
4475 if let Some((_, material)) = material.as_mut() {
4476 material.ui_tree = Some(bytes);
4477 }
4478 Ok(Some(EvidenceRef {
4479 asset: snapshot.evidence.clone(),
4480 sha256: put.sha256,
4481 local_path: put.local_path,
4482 }))
4483 }
4484 Ok(UiSnapshotOutcome::Unavailable { reason }) => {
4485 *ui_snapshot_omission = Some(reason);
4486 if let Some((_, material)) = material.as_mut() {
4487 material.ui_tree_gap =
4488 Some(format!("uiSnapshot dereference unavailable ({reason:?})"));
4489 }
4490 Ok(None)
4491 }
4492 Err(error) => {
4493 if let Some((_, material)) = material.as_mut() {
4494 material.ui_tree_gap = Some(format!("uiSnapshot dereference failed: {error}"));
4495 }
4496 Ok(None)
4497 }
4498 }
4499 }
4500
4501 /// Fetches an asset's bytes into the content-addressed evidence area.
4502 /// The outer `Result` is infrastructure (store I/O — still fatal); the
4503 /// inner one is the typed localization gap (fetch unsupported/ruptured,
4504 /// integrity mismatch — the run degrades, never aborts; 04 §4.3 is
4505 /// honored by *not using* mismatched bytes, with the reason on record).
4506 async fn try_localize(
4507 &mut self,
4508 asset: &AssetRef,
4509 ) -> Result<Result<(EvidenceRef, Vec<u8>), String>, RunnerError> {
4510 let mut stream = match self.session.fetch_evidence(asset).await {
4511 Ok(stream) => stream,
4512 Err(error) => {
4513 return Ok(Err(format!(
4514 "evidence fetch failed for asset {}: {error}",
4515 asset.id
4516 )));
4517 }
4518 };
4519 let mut bytes = Vec::new();
4520 while let Some(chunk) = stream.next().await {
4521 match chunk {
4522 Ok(part) => bytes.extend(part),
4523 Err(error) => {
4524 return Ok(Err(format!(
4525 "evidence stream failed for asset {}: {error}",
4526 asset.id
4527 )));
4528 }
4529 }
4530 }
4531 let put = self.store.put_evidence(&bytes, &asset.media_type)?;
4532 if let Some(expected) = &asset.sha256
4533 && expected != &put.sha256
4534 {
4535 return Ok(Err(format!(
4536 "evidence integrity failure for asset {}: sha256 {} != declared {expected}",
4537 asset.id, put.sha256
4538 )));
4539 }
4540 let evidence = EvidenceRef {
4541 asset: asset.clone(),
4542 sha256: put.sha256,
4543 local_path: put.local_path,
4544 };
4545 Ok(Ok((evidence, bytes)))
4546 }
4547
4548 /// Appends `verdictRecorded` and writes the verdict back through the
4549 /// provider (`verdict.record` — the daemon only validates and
4550 /// persists). Returns the `verdictRecorded` seq (evidence linking).
4551 async fn record_step_verdict(
4552 &mut self,
4553 path: &RunPath,
4554 folded: &FoldedVerdict,
4555 cited: Vec<AssetRef>,
4556 supersedes: Option<String>,
4557 manifest: EvidenceManifest,
4558 ) -> Result<u64, RunnerError> {
4559 let verdict = Verdict {
4560 status: folded.status,
4561 degraded: folded.degraded,
4562 // The local ledger keeps the FULL summary — the 16384-char
4563 // cap is a wire hard limit, applied at write-back only
4564 // (04 §5).
4565 summary: folded.summary.clone(),
4566 evidence: cited
4567 .into_iter()
4568 .take(VERDICT_EVIDENCE_MAX_ENTRIES)
4569 .collect(),
4570 supersedes,
4571 };
4572 // Failure-instant capture (07 §2.2): the verdict instant IS the
4573 // failure instant, so the capture runs FIRST — before the
4574 // write-back RPC below can delay it on a degraded daemon. The
4575 // profile rides the span's exit through the `append` attach; a
4576 // superseding pass discards it. Deliberate cost: a retry round
4577 // that will be superseded still pays a capture (bounded by the
4578 // 2s budget) — skipping it would need the consult outcome, which
4579 // is only known after the handler runs, and a wrong skip would
4580 // ship a summary-less fail exit. Correctness over the bounded
4581 // RPC.
4582 let key = instance_key(path);
4583 match verdict.status {
4584 VerdictStatus::Fail | VerdictStatus::Unknown => {
4585 let summary = self.capture_summary().await;
4586 self.pending_summaries.insert(key, summary);
4587 }
4588 VerdictStatus::Pass => {
4589 self.pending_summaries.remove(&key);
4590 }
4591 }
4592 // Remote archival before the append so its outcome can ride the
4593 // event; it is archival of an already-derived verdict, not a
4594 // world effect, so the actionIntent WAL discipline does not
4595 // apply. A failure never changes the local verdict and never
4596 // aborts the run — it is annotated here and surfaced by the
4597 // report (04 §5). Deliberate cost: on a hung daemon the
4598 // `verdictRecorded` `at_ms` trails the fold instant by the
4599 // bounded write-back budget — the price of carrying the outcome
4600 // on the event under the closed §6.1 vocabulary.
4601 let remote_archival_error = self.try_verdict_writeback(&verdict).await;
4602 let seq = self.append(
4603 path,
4604 &RunLogPayload::VerdictRecorded {
4605 verdict: verdict.clone(),
4606 localized: manifest.localized,
4607 localization_gaps: manifest.gaps,
4608 remote_archival_error,
4609 },
4610 )?;
4611 Ok(seq)
4612 }
4613
4614 /// `ProviderSession::record_verdict` write-back with the wire caps
4615 /// applied on the runner side (compaction is the runner's job,
4616 /// 04 §5). Returns the failure rendered for the ledger annotation —
4617 /// never an error: remote archival failure must not change the local
4618 /// verdict or abort the run (04 §5, the RunLog is the sole truth).
4619 async fn try_verdict_writeback(&mut self, verdict: &Verdict) -> Option<String> {
4620 self.session
4621 .record_verdict(VerdictWrite {
4622 status: verdict.status,
4623 summary: cap_wire_summary(verdict),
4624 evidence: verdict
4625 .evidence
4626 .iter()
4627 .take(VERDICT_EVIDENCE_MAX_ENTRIES)
4628 .cloned()
4629 .collect(),
4630 })
4631 .await
4632 .err()
4633 .map(|error| format!("remote archival failed: {error}"))
4634 }
4635
4636 /// Best-effort session teardown (04 §2.1: `end` must not block the
4637 /// runner's teardown when the transport is already gone).
4638 async fn end_session(&mut self, outcome: SessionOutcome) {
4639 let _ = self.session.end(outcome, None).await;
4640 }
4641}
4642
4643/// The inbound gate of a call step (07 §1.1): apply the callee's declared
4644/// param defaults, refuse undeclared inputs and missing required params,
4645/// and validate every present value against its `ParamDecl.schema`.
4646fn call_inputs_gate(
4647 callee: &FlowIR,
4648 inputs: Map<String, Value>,
4649) -> Result<Map<String, Value>, String> {
4650 for key in inputs.keys() {
4651 if !callee
4652 .params
4653 .iter()
4654 .any(|decl: &ParamDecl| decl.name.as_str() == key)
4655 {
4656 return Err(format!(
4657 "input '{key}' is not a declared param of callee '{}'",
4658 callee.flow_id
4659 ));
4660 }
4661 }
4662 let gated =
4663 params_with_defaults(callee, Value::Object(inputs)).map_err(|error| error.to_string())?;
4664 for decl in &callee.params {
4665 if let Some(value) = gated.get(decl.name.as_str()) {
4666 jsonschema::validate(decl.schema.as_value(), value)
4667 .map_err(|error| format!("param '{}' failed its schema: {error}", decl.name))?;
4668 }
4669 }
4670 Ok(gated)
4671}
4672
4673/// Applies declared param defaults over the supplied params/inputs;
4674/// missing required params without defaults are refused. Shared by the run
4675/// entry (run params) and the call step's inbound gate (07 §1.1).
4676pub(crate) fn params_with_defaults(
4677 flow: &FlowIR,
4678 params: Value,
4679) -> Result<Map<String, Value>, RunnerError> {
4680 let mut map = match params {
4681 Value::Object(map) => map,
4682 Value::Null => Map::new(),
4683 other => {
4684 return Err(RunnerError::InvalidParams {
4685 reason: format!("params must be a JSON object or null, got {other}"),
4686 });
4687 }
4688 };
4689 for decl in &flow.params {
4690 if map.contains_key(decl.name.as_str()) {
4691 continue;
4692 }
4693 if let Some(default) = &decl.default {
4694 map.insert(decl.name.as_str().to_owned(), default.clone());
4695 } else if decl.required {
4696 return Err(RunnerError::InvalidParams {
4697 reason: format!(
4698 "required param '{}' is missing and has no default",
4699 decl.name
4700 ),
4701 });
4702 }
4703 }
4704 Ok(map)
4705}
4706
4707/// Which observation channels a set of assertions' verify chains consume —
4708/// decides what the observing phase must localize into
4709/// [`ObserveMaterial`] and what a fresh observe must want.
4710pub(crate) struct VerifyNeeds {
4711 /// Some assertion's chain contains `uiTree`.
4712 pub ui_tree: bool,
4713 /// Some assertion's chain contains `vision`.
4714 pub vision: bool,
4715}
4716
4717impl VerifyNeeds {
4718 /// Scans the assertions (expr predicates consume no channel).
4719 pub fn of(assertions: &[AssertionIR]) -> Self {
4720 let mut needs = VerifyNeeds {
4721 ui_tree: false,
4722 vision: false,
4723 };
4724 for assertion in assertions {
4725 if matches!(assertion.predicate, PredicateIR::Expr { .. }) {
4726 continue;
4727 }
4728 for channel in &assertion.verify_via {
4729 match channel {
4730 VerifyChannel::UiTree => needs.ui_tree = true,
4731 VerifyChannel::Vision => needs.vision = true,
4732 VerifyChannel::Dom => {}
4733 }
4734 }
4735 }
4736 needs
4737 }
4738}
4739
4740/// Truncates a verdict summary to the provider wire cap (char-aware),
4741/// appending a content-hash pointer to the local full verdict when it
4742/// cuts (04 §5: the RunLog keeps the complete summary; the wire copy
4743/// points back at it).
4744pub(crate) fn cap_wire_summary(verdict: &Verdict) -> String {
4745 if verdict.summary.chars().count() <= VERDICT_SUMMARY_MAX_CHARS {
4746 return verdict.summary.clone();
4747 }
4748 let pointer = format!(
4749 " …[truncated; full local verdict {}]",
4750 pointlock_ir::domain_hash(
4751 "pointlock-runner/1/local-verdict",
4752 &serde_json::to_value(verdict).expect("a Verdict always serializes"),
4753 )
4754 );
4755 let keep = VERDICT_SUMMARY_MAX_CHARS.saturating_sub(pointer.chars().count());
4756 let mut capped: String = verdict.summary.chars().take(keep).collect();
4757 capped.push_str(&pointer);
4758 capped
4759}
4760
4761/// The backoff delay before retry number `tries + 1` (spine §3
4762/// `RetryPolicy.backoffMs`).
4763fn backoff_ms(policy: &pointlock_ir::RetryPolicy, tries: u32) -> u64 {
4764 match &policy.backoff_ms {
4765 pointlock_ir::BackoffMs::Fixed(number) => number.as_f64().unwrap_or(0.0) as u64,
4766 pointlock_ir::BackoffMs::Schedule(schedule) => {
4767 let initial = schedule.initial.as_f64().unwrap_or(0.0);
4768 let factor = schedule.factor.as_f64().unwrap_or(1.0);
4769 let max = schedule.max.as_f64().unwrap_or(f64::MAX);
4770 let exponent = tries.saturating_sub(1);
4771 (initial * factor.powi(exponent as i32)).min(max) as u64
4772 }
4773 }
4774}
4775
4776/// Whether an in-attempt retry is allowed (spine §6.5 mount point 1,
4777/// closed): only `action_failed_retryable`, `target_stale`, and — for
4778/// idempotent steps — `action_timed_out`, and only when the policy lists
4779/// the class and the budget is not exhausted.
4780fn retry_allowed(step: &ActionStepIR, class: ErrorClass, tries: u32) -> bool {
4781 let Some(policy) = &step.base.retry else {
4782 return false;
4783 };
4784 if tries >= policy.max_attempts {
4785 return false;
4786 }
4787 if !policy.retry_on.contains(&class) {
4788 return false;
4789 }
4790 match class {
4791 ErrorClass::ActionFailedRetryable | ErrorClass::TargetStale => true,
4792 ErrorClass::ActionTimedOut => step.idempotent,
4793 _ => false,
4794 }
4795}
4796
4797/// Maps a non-succeeded terminal onto the closed `ErrorClass` taxonomy
4798/// (spine §5). When the wire code spells a class verbatim it is adopted
4799/// (mirrors the store fold's best-effort rule); otherwise `failed` maps by
4800/// the daemon-declared `retryable` flag.
4801pub(crate) fn classify(outcome: &ActionOutcome) -> ErrorClass {
4802 match outcome {
4803 ActionOutcome::Succeeded { .. } => {
4804 unreachable!("classify is only called on non-succeeded terminals")
4805 }
4806 ActionOutcome::Failed { error } => code_spelled_class(&error.code).unwrap_or({
4807 if error.retryable {
4808 ErrorClass::ActionFailedRetryable
4809 } else {
4810 ErrorClass::ActionFailedFinal
4811 }
4812 }),
4813 ActionOutcome::TimedOut { .. } => ErrorClass::ActionTimedOut,
4814 ActionOutcome::Cancelled { .. } => ErrorClass::ActionCancelled,
4815 }
4816}
4817
4818fn code_spelled_class(code: &str) -> Option<ErrorClass> {
4819 serde_json::from_value(Value::String(code.to_owned())).ok()
4820}
4821
4822fn terminal_message(outcome: &ActionOutcome) -> String {
4823 let error: &ErrorInfo = match outcome {
4824 ActionOutcome::Failed { error }
4825 | ActionOutcome::Cancelled { error }
4826 | ActionOutcome::TimedOut { error } => error,
4827 ActionOutcome::Succeeded { .. } => {
4828 unreachable!("terminal_message is only called on non-succeeded terminals")
4829 }
4830 };
4831 format!("{} ({})", error.message, error.code)
4832}
4833
4834/// Whether the provider-reported execution mode is inside the attempt's
4835/// whitelist (§6.4 R-degrade). An absent execution report cannot be
4836/// audited and is accepted (the DeviceRail adapter always reports it).
4837fn execution_accepted(attempt: &BoundAttempt, execution: &Option<ActionExecution>) -> bool {
4838 match execution {
4839 None => true,
4840 Some(execution) => {
4841 let mode = match execution {
4842 ActionExecution::NativeSemantic { .. } => ExecutionMode::NativeSemantic,
4843 ActionExecution::WebSemantic { .. } => ExecutionMode::WebSemantic,
4844 ActionExecution::CoordinateFallback { .. } => ExecutionMode::CoordinateFallback,
4845 };
4846 attempt.accept_execution_modes.contains(&mode)
4847 }
4848 }
4849}
4850
4851/// Whether an action step is effectively mutating for the I2 replay gates
4852/// (mutating and not declared idempotent).
4853pub(crate) fn gated_mutating(step: &ActionStepIR) -> bool {
4854 step.effect == EffectClassAction::Mutating && !step.idempotent
4855}
4856
4857/// Whether an uncertain reconcile branch may replay the step
4858/// (07 §4.4: `idempotent: true` or `effect: "readonly"`).
4859pub(crate) fn replay_permitted(step: &ActionStepIR) -> bool {
4860 step.effect == EffectClassAction::Readonly || step.idempotent
4861}
4862
4863#[cfg(test)]
4864mod chain_start_tests {
4865 use super::*;
4866
4867 fn two_attempt_step() -> ActionStepIR {
4868 serde_json::from_value(serde_json::json!({
4869 "kind": "action",
4870 "stepId": "s1",
4871 "effectHash": format!("sha256:{}", "0".repeat(64)),
4872 "judgeHash": format!("sha256:{}", "0".repeat(64)),
4873 "checkpoint": true,
4874 "effect": "mutating",
4875 "idempotent": true,
4876 "binding": { "attempts": [ {
4877 "channel": "uiTree",
4878 "actionName": "tapElement",
4879 "args": {},
4880 "acceptExecutionModes": ["nativeSemantic"],
4881 "protection": "standard"
4882 }, {
4883 "channel": "uiTree",
4884 "actionName": "setElementValue",
4885 "args": {},
4886 "acceptExecutionModes": ["nativeSemantic"],
4887 "protection": "standard"
4888 } ] },
4889 "assertions": []
4890 }))
4891 .expect("fixture step")
4892 }
4893
4894 #[test]
4895 fn maps_recorded_positions_and_refuses_out_of_range() {
4896 let step = two_attempt_step();
4897 assert_eq!(chain_start(None, &step).expect("head"), 0);
4898 assert_eq!(chain_start(Some(1), &step).expect("first"), 0);
4899 assert_eq!(chain_start(Some(2), &step).expect("second"), 1);
4900 assert!(chain_start(Some(3), &step).is_err());
4901 assert!(chain_start(Some(0), &step).is_err());
4902 }
4903}
4904
4905#[cfg(test)]
4906mod tests {
4907 use super::*;
4908
4909 #[test]
4910 fn wire_summary_truncation_appends_the_local_verdict_pointer() {
4911 let verdict = Verdict {
4912 status: VerdictStatus::Fail,
4913 degraded: false,
4914 summary: "x".repeat(VERDICT_SUMMARY_MAX_CHARS + 100),
4915 evidence: Vec::new(),
4916 supersedes: None,
4917 };
4918 let capped = cap_wire_summary(&verdict);
4919 // Exactly at the wire cap — the fake/devicerail providers reject
4920 // anything above it fail-closed.
4921 assert_eq!(capped.chars().count(), VERDICT_SUMMARY_MAX_CHARS);
4922 // The 04 §5 pointer to the full local verdict rides the tail.
4923 assert!(
4924 capped.ends_with(']') && capped.contains("full local verdict sha256:"),
4925 "pointer missing: …{}",
4926 &capped[capped.len().saturating_sub(90)..]
4927 );
4928 assert!(capped.starts_with("xxx"));
4929
4930 // Negative control: an in-cap summary passes through verbatim,
4931 // pointer-free.
4932 let short = Verdict {
4933 summary: "all assertions passed".to_owned(),
4934 ..verdict
4935 };
4936 assert_eq!(cap_wire_summary(&short), "all assertions passed");
4937 }
4938}