Skip to main content

pointlock_runner/
runner.rs

1//! The public entry points: [`Runner::run`] and [`Runner::resume`] /
2//! [`Runner::resume_with_subflows`] (spine §6; 07 §4–§5).
3//!
4//! Since M2 the entry points accept the resolved subflow registry
5//! (`Map<irHash, FlowIR>`, provided by the assembly layer; every entry
6//! self-verifies at load). Resume comes in two regimes:
7//!
8//! - **Same-IR resume** (the common suspend/crash continuation): every
9//!   completed step *instance* is adopted by its exact run path — the
10//!   walk falls back into open call frames and foreach iterations without
11//!   restarting any frame (07 §4.6), with the archived control snapshots
12//!   (cond / items / inputs) never re-evaluated (I3).
13//! - **Cross-IR resume** (repair): the 07 §5.2 *flat* subset — alignment,
14//!   offline re-judge and the `requiresConfirmation` gates over top-level
15//!   action steps. The nested rules (call down-drill, per-iteration
16//!   alignment, order-consistency) land with the repair wave; the
17//!   combination is refused with a typed error, never silently guessed.
18
19use std::collections::{BTreeMap, BTreeSet};
20
21use pointlock_ir::{
22    ActionStepIR, AlignmentClass, AlignmentEntry, AlignmentReport, BindingState, CheckpointView,
23    FlowIR, Hash, PathFrame, ReconcileResult, RequiresConfirmation, RunLogPayload, RunPath,
24    SupervisePolicy, ir_hash,
25};
26use pointlock_provider_kit::{CancellationToken, ProviderSession, SessionOutcome};
27use pointlock_store::{NewRun, Store};
28use serde_json::{Map, Value};
29
30use crate::align::{Alignment, Harvest, align, harvest, live_frame_pins, open_instances};
31use crate::engine::{
32    Adopted, Execution, FrameState, FrontierWork, HumanRequestFact, RunOutcome, gated_mutating,
33    instance_key, is_history, now_ms, params_with_defaults, replay_permitted, root_path,
34};
35use crate::error::{BlockedReason, RunnerError};
36use crate::load::{LoadedFlow, check_attestation, load};
37use crate::scope::ScopeSeed;
38
39/// Options of [`Runner::run`].
40pub struct RunOptions {
41    /// Cooperative stop token, honored at step boundaries
42    /// (`runSuspended` → [`RunOutcome::Suspended`]).
43    pub stop: CancellationToken,
44    /// Explicit run id; a UUIDv4 is generated when absent.
45    pub run_id: Option<String>,
46    /// The bound device (checkpoint hard binding; also `env.deviceId`).
47    pub device_id: String,
48    /// The device platform for `env.platform`, when known (comes from the
49    /// lockfile at the assembly layer; the SPI attestation does not carry
50    /// it).
51    pub platform: Option<String>,
52    /// The vision verifier consulted by `vision` verify-chain tails.
53    /// `None` is equivalent to
54    /// [`pointlock_vision::StubVisionVerifier`]: the vision channel cannot
55    /// complete and reports `"vision verifier not configured"` — the chain
56    /// degrades honestly toward `unknown` (principle 4).
57    pub vision: Option<std::sync::Arc<dyn pointlock_vision::VisionVerifier>>,
58    /// The resolved subflow registry keyed by `irHash` (07 §1.3): every
59    /// callee the flow's `subflows` table pins must be present; entries
60    /// self-verify at load. Empty for flows without subflows.
61    pub subflows: BTreeMap<Hash, FlowIR>,
62    /// This segment's supervision policy (R13, spine §6.9): recorded in
63    /// `runStarted.supervisePolicy` (explicitly `null` when absent) and
64    /// gates action-step dispatch (`mutating` gates mutating steps,
65    /// `all` every action step). Per segment, never inherited.
66    pub supervise: Option<SupervisePolicy>,
67    /// Injectable wall clock for human-deadline computation and lazy
68    /// timeout settlement (tests); `None` uses the system clock.
69    pub clock: Option<std::sync::Arc<dyn Fn() -> u64 + Send + Sync>>,
70}
71
72impl RunOptions {
73    /// Options with a fresh stop token, no explicit run id, no vision
74    /// verifier (stub-equivalent), an empty subflow registry, no
75    /// supervision and the system clock.
76    pub fn new(device_id: impl Into<String>) -> Self {
77        RunOptions {
78            stop: CancellationToken::new(),
79            run_id: None,
80            device_id: device_id.into(),
81            platform: None,
82            vision: None,
83            subflows: BTreeMap::new(),
84            supervise: None,
85            clock: None,
86        }
87    }
88}
89
90/// Options of [`Runner::resume`].
91#[derive(Default)]
92pub struct ResumeOptions {
93    /// Cooperative stop token (see [`RunOptions::stop`]).
94    pub stop: CancellationToken,
95    /// `env.platform`, when known.
96    pub platform: Option<String>,
97    /// The FlowIR the run originally executed — optional, and worth
98    /// supplying. Alignment reads the archived execution-time per-step
99    /// hashes from the checkpoint's `StepRecord`s (harvested from
100    /// `stepEntered`, spine §6.1 M1 note), so cross-IR resume works
101    /// without it; supplying it additionally unlocks the preflight-only
102    /// sub-domain comparison (07 §5.3 / 02 §12.3 ruling 6) — a
103    /// `judgeDirty` step whose only change is `preflight` adopts its
104    /// archived verdict outright instead of re-judging or re-executing.
105    /// Verified against the checkpoint's `irHash`
106    /// ([`RunnerError::OldIrMismatch`]); a mismatch is a caller error,
107    /// surfaced not ignored.
108    pub old_flow_ir: Option<FlowIR>,
109    /// This segment's supervision policy (R13, spine §6.9): recorded in
110    /// `runResumed.supervisePolicy` (explicitly `null` when absent).
111    /// Per segment, never inherited — an unset value means this segment
112    /// runs unsupervised regardless of previous segments; a supervision
113    /// request already pending still settles by its arbitrated response.
114    pub supervise: Option<SupervisePolicy>,
115    /// The vision verifier of this segment (see [`RunOptions::vision`]);
116    /// `None` is stub-equivalent — vision tails degrade to `unknown`.
117    pub vision: Option<std::sync::Arc<dyn pointlock_vision::VisionVerifier>>,
118    /// Step ids the author FORCES back to execution this segment
119    /// (07 §5.3, the CLI's repeatable `--force-reexecute <stepId>`):
120    /// each named step classifies `effectDirty` regardless of its hashes,
121    /// so the resume point rolls back to the earliest of them and they
122    /// re-run against the live world.
123    ///
124    /// The escape hatch for a re-judge the author rejects — an offline
125    /// re-judge that can only reach `unknown` because the archive lacks
126    /// the observation channel the new assertion needs (「缺料 →
127    /// unknown」), or an adopted result the author no longer trusts. It
128    /// upgrades the CLASSIFICATION only: a forced step that is mutating
129    /// and already effective still walks the 07 §5.4 gate and needs
130    /// `--allow-mutating-reexec` besides — forcing says "run it again",
131    /// authorizing says "yes, even though the world holds its effect".
132    /// Like the authorization list it covers this resume only, and it is
133    /// cross-IR vocabulary: a same-IR resume has no classification to
134    /// upgrade.
135    pub force_reexecute: Vec<String>,
136    /// Step ids the author explicitly authorized for mutating
137    /// re-execution this segment (07 §5.4 step 2, the CLI's repeatable
138    /// `--allow-mutating-reexec <stepId>`).
139    ///
140    /// Each id releases exactly one `requiresConfirmation` entry; there is
141    /// no wildcard, and the authorization covers **this resume only** —
142    /// nothing about it is persisted, so the next resume re-gates from
143    /// scratch. An id naming no gated step is refused rather than ignored:
144    /// silently accepting it would let an author believe they had cleared
145    /// something they had not.
146    ///
147    /// Releasing the gate does not skip the world check: the step still
148    /// enters `probing` and evaluates its `preflight` (§5.4 step 3), which
149    /// is what meets the residue of the earlier effect.
150    pub allow_mutating_reexec: Vec<String>,
151    /// Injectable wall clock (see [`RunOptions::clock`]).
152    pub clock: Option<std::sync::Arc<dyn Fn() -> u64 + Send + Sync>>,
153}
154
155/// The runner: executes a sealed [`FlowIR`] against an open provider
156/// session, journaling every transition into the single-writer store.
157/// Entry signatures accept only `FlowIR`, never strings (principles 1/2).
158pub struct Runner;
159
160impl Runner {
161    /// Runs a flow from the beginning (spine §6.2 pipeline; §6.1 event
162    /// vocabulary). This segment's `supervisePolicy` is recorded verbatim
163    /// in `runStarted` — explicitly `null` when unsupervised (R13,
164    /// per-segment self-describing ledger).
165    pub async fn run(
166        flow: &FlowIR,
167        params: Value,
168        session: Box<dyn ProviderSession>,
169        store: &mut Store,
170        opts: RunOptions,
171    ) -> Result<RunOutcome, RunnerError> {
172        let RunOptions {
173            stop,
174            run_id,
175            device_id,
176            platform,
177            vision,
178            subflows,
179            supervise,
180            clock,
181        } = opts;
182        let loaded = load(flow, &subflows)?;
183        check_attestation(&loaded, session.attestation())?;
184        let params = params_with_defaults(flow, params)?;
185
186        let cursor = session.current_cursor().await?;
187        let initial_lineage = vec![cursor.session_id.clone()];
188        let run_id = store.begin_run(NewRun {
189            run_id,
190            flow_id: flow.flow_id.clone(),
191            ir_hash: flow.ir_hash.clone(),
192            lockfile_digest: flow.lockfile_digest.clone(),
193            params_snapshot: Value::Object(params.clone()),
194            binding: BindingState {
195                device_id: device_id.clone(),
196                session_lineage: vec![cursor.session_id.clone()],
197                event_cursor: cursor,
198            },
199            created_at_ms: now_ms(),
200        })?;
201        store.append_event(
202            &run_id,
203            now_ms(),
204            &root_path(flow),
205            &RunLogPayload::RunStarted {
206                ir_hash: flow.ir_hash.clone(),
207                lockfile_digest: flow.lockfile_digest.clone(),
208                params_snapshot: Value::Object(params.clone()),
209                // R13: this segment's real policy — explicitly null when
210                // unsupervised (per-segment, self-describing).
211                supervise_policy: supervise,
212            },
213        )?;
214
215        let env = env_bindings(&device_id, platform.as_deref(), &run_id);
216        let exec = Execution {
217            flows: &loaded,
218            session,
219            store,
220            run_id,
221            stop,
222            env,
223            session_lineage: initial_lineage,
224            pending_summaries: Default::default(),
225            attempt_base: Default::default(),
226            open_spans: Default::default(),
227            live_frames: Default::default(),
228            // A fresh run never re-touches a world it stopped watching.
229            resumed: false,
230            authorized: BTreeSet::new(),
231            reentry_seen: false,
232            adoptable: Default::default(),
233            frontier: None,
234            vision,
235            supervise,
236            human: Default::default(),
237            settled: Default::default(),
238            recorded_verdicts: Default::default(),
239            hook_triggers: Default::default(),
240            clock,
241        };
242        let root = FrameState::new(flow, root_path(flow), params, 1);
243        exec.run(root, 0).await
244    }
245
246    /// Resumes a run (07 §4) without subflows. Legality ⟺ (A) every
247    /// completed record is still recognized under the (possibly repaired)
248    /// new IR — recorded as `alignmentReport` in `runResumed`; (B) a
249    /// pending intent on the frontier has been reconciled
250    /// (`ProviderSession::reconcile`); (C) the world passes the resume
251    /// probes — the first to-execute step's declared `preflight` runs
252    /// before its act; a step without probes resumes honestly `unprobed`
253    /// (I3).
254    pub async fn resume(
255        new_flow: &FlowIR,
256        run_id: &str,
257        session: Box<dyn ProviderSession>,
258        store: &mut Store,
259        opts: ResumeOptions,
260    ) -> Result<RunOutcome, RunnerError> {
261        let subflows = BTreeMap::new();
262        Self::resume_with_subflows(new_flow, &subflows, run_id, session, store, opts).await
263    }
264
265    /// [`Runner::resume`] with a resolved subflow registry — required when
266    /// the (new) flow pins callees; see [`RunOptions::subflows`].
267    pub async fn resume_with_subflows(
268        new_flow: &FlowIR,
269        subflows: &BTreeMap<Hash, FlowIR>,
270        run_id: &str,
271        session: Box<dyn ProviderSession>,
272        store: &mut Store,
273        opts: ResumeOptions,
274    ) -> Result<RunOutcome, RunnerError> {
275        let loaded = load(new_flow, subflows)?;
276        check_attestation(&loaded, session.attestation())?;
277        let view = store.rebuild_checkpoint(run_id)?;
278        let events = store.events(run_id)?;
279        let facts = harvest(&events);
280        let executing = executing_ir_hash(&facts, &view);
281
282        // The optional old-IR integrity check: when the caller supplies
283        // one, verify it is the IR the run executed — the LAST segment's
284        // IR, not the bind-time one (a mismatched old IR is a caller
285        // error, surfaced not ignored).
286        if let Some(old) = opts.old_flow_ir.as_ref() {
287            let computed = ir_hash(old);
288            if computed != *executing {
289                return Err(RunnerError::OldIrMismatch {
290                    expected: executing.clone(),
291                    computed,
292                });
293            }
294        }
295
296        if is_same_ir_resume(&facts, &view, new_flow) {
297            resume_same_ir(loaded, view, facts, run_id, session, store, opts).await
298        } else {
299            resume_cross_ir(loaded, view, facts, run_id, session, store, opts).await
300        }
301    }
302
303    /// The READ-ONLY alignment preview (08 §2.7): the resume path's
304    /// classification verbatim — same-IR trivial adoption or the flat
305    /// cross-IR `align` — but no session, no attestation, no writes, no
306    /// commitment. The preview is not a promise: the world can drift
307    /// between preview and resume; the resume-time preflight probes stay
308    /// the final judge. A confirmation-gated alignment is a preview
309    /// RESULT here (the report shows what the real resume would refuse),
310    /// not an error.
311    #[allow(clippy::too_many_arguments)]
312    pub async fn align_preview(
313        new_flow: &FlowIR,
314        subflows: &BTreeMap<Hash, FlowIR>,
315        run_id: &str,
316        store: &Store,
317        platform: Option<&str>,
318        vision: Option<&dyn pointlock_vision::VisionVerifier>,
319        forced: &[String],
320        old_flow_ir: Option<&FlowIR>,
321    ) -> Result<AlignmentReport, RunnerError> {
322        let loaded = load(new_flow, subflows)?;
323        let view = store.rebuild_checkpoint(run_id)?;
324        let events = store.events(run_id)?;
325        let facts = harvest(&events);
326        let executing = executing_ir_hash(&facts, &view);
327
328        // The same old-IR integrity check the real resume applies: a
329        // mismatched old IR is a caller error, and rehearsing with it
330        // would classify against the wrong sub-domains.
331        if let Some(old) = old_flow_ir {
332            let computed = ir_hash(old);
333            if computed != *executing {
334                return Err(RunnerError::OldIrMismatch {
335                    expected: executing.clone(),
336                    computed,
337                });
338            }
339        }
340
341        // The preview mirrors resume's typed refusals — a clean rehearsal
342        // of a resume the runner would categorically refuse is a lie.
343        if let Some(live_hook) = facts
344            .live_frames
345            .iter()
346            .find(|path| path.iter().any(|f| matches!(f, PathFrame::Hook { .. })))
347        {
348            return Err(RunnerError::M0Unsupported {
349                detail: format!(
350                    "resume across a live handler-repair frame ({}) is not in the M2 subset — \
351                     the repair flow suspended mid-flight; hook-aware frame re-entry is \
352                     registered for the repair wave",
353                    pointlock_ir::render_run_path(live_hook)
354                ),
355            });
356        }
357
358        if is_same_ir_resume(&facts, &view, new_flow) {
359            return Ok(same_ir_report(new_flow, &facts, &view));
360        }
361
362        if !is_alignable_path(&view.frontier.run_path) {
363            return Err(RunnerError::M0Unsupported {
364                detail: "the run's frontier sits inside a handler frame".to_owned(),
365            });
366        }
367        // Mirrors resume_cross_ir's third hook state (an escalate human
368        // still awaiting an answer): rehearsing a resume the runner would
369        // categorically refuse is a lie.
370        if view
371            .human_pending
372            .as_ref()
373            .is_some_and(|pending| !is_alignable_path(&pending.run_path))
374        {
375            return Err(RunnerError::M0Unsupported {
376                detail: "a handler escalation is still awaiting an answer".to_owned(),
377            });
378        }
379
380        // `env.platform` comes from the caller (the serve endpoint reads
381        // it from the SAME lockfile the resume assembly uses); when
382        // absent, an expr predicate referencing it re-judges to unknown
383        // in the preview (fail-closed) while the real resume would judge
384        // it — pass the platform to keep the rehearsal faithful.
385        let seed = ScopeSeed::new(
386            params_object(&view),
387            &view.binding.device_id,
388            platform,
389            run_id,
390        );
391        match align(
392            &loaded,
393            &new_flow.body,
394            &view,
395            &facts,
396            &seed,
397            store,
398            vision,
399            // A preview shows what WOULD gate: it authorizes nothing. The
400            // FORCED list and the old IR it does take — the rehearsal must
401            // classify exactly as the real resume will.
402            &[],
403            forced,
404            old_flow_ir,
405        )
406        .await
407        {
408            Ok(alignment) => Ok(alignment.report),
409            Err(RunnerError::RequiresConfirmation { report }) => Ok(*report),
410            Err(other) => Err(other),
411        }
412    }
413}
414
415/// `env.*` bindings: `deviceId`, `runId`, and `platform` when known (the
416/// platform comes from the assembly layer — the SPI attestation does not
417/// carry it). Run-constant, read-only pass-through across frames (07 §1.2).
418fn env_bindings(device_id: &str, platform: Option<&str>, run_id: &str) -> Vec<(String, Value)> {
419    let mut env = vec![
420        ("deviceId".to_owned(), Value::String(device_id.to_owned())),
421        ("runId".to_owned(), Value::String(run_id.to_owned())),
422    ];
423    if let Some(platform) = platform {
424        env.push(("platform".to_owned(), Value::String(platform.to_owned())));
425    }
426    env
427}
428
429/// The IR the run is currently executing under: the last `runResumed`
430/// segment's root flow hash when one exists, else the bind-time run-row
431/// hash. The fold never rebases `view.ir_hash` past `runStarted`, so after
432/// a cross-IR segment the checkpoint hash names the ORIGINAL artifact —
433/// the `--old-ir` check must name the last executing IR instead.
434fn executing_ir_hash<'a>(facts: &'a Harvest, view: &'a CheckpointView) -> &'a Hash {
435    facts.executing_ir_hash.as_ref().unwrap_or(&view.ir_hash)
436}
437
438/// Whether the trivial same-IR adoption is sound: the new IR is the one
439/// the run was bound to AND every segment so far executed under it. A
440/// ledger that has executed under another IR holds records of more than
441/// one IR, and adopting them by instance key alone would either credit
442/// the original artifact with a repaired segment's results or adopt
443/// positionally invalidated first-life records (07 §5.2) — such a resume
444/// is classified by `align` whatever IR it names.
445fn is_same_ir_resume(facts: &Harvest, view: &CheckpointView, new_flow: &FlowIR) -> bool {
446    !facts.cross_ir_resumed && view.ir_hash == new_flow.ir_hash
447}
448
449/// The same-IR alignment report: top-level instances with execution
450/// history are trivially reusable (identical hashes by construction);
451/// the rest re-execute as `new`. Shared by [`resume_same_ir`] and the
452/// read-only [`Runner::align_preview`] — one classification truth source,
453/// so the liveness rule ([`open_instances`]) is applied here too (R13).
454fn same_ir_report(new_flow: &FlowIR, facts: &Harvest, view: &CheckpointView) -> AlignmentReport {
455    let open = open_instances(facts);
456    let completed: BTreeMap<String, &pointlock_ir::StepRecord> = view
457        .completed
458        .iter()
459        .filter(|record| !open.contains(&instance_key(&record.run_path)))
460        .map(|record| (instance_key(&record.run_path), record))
461        .collect();
462    let mut entries = Vec::new();
463    for step in &new_flow.body {
464        let mut path = root_path(new_flow);
465        path.push(match step {
466            pointlock_ir::StepIR::Call(call) => PathFrame::Call {
467                step_id: Some(call.base.step_id.clone()),
468                callee_flow_id: call.flow_ref.flow_id.clone(),
469                callee_ir_hash: call.flow_ref.ir_hash.clone(),
470            },
471            other => PathFrame::Step {
472                step_id: other.step_id().clone(),
473            },
474        });
475        let key = instance_key(&path);
476        let adopted = completed.get(&key).is_some_and(|record| is_history(record));
477        entries.push(AlignmentEntry {
478            run_path: path.clone(),
479            step_id: step.step_id().clone(),
480            class: if adopted {
481                AlignmentClass::Reusable
482            } else {
483                AlignmentClass::New
484            },
485            reason: (!adopted).then(|| "no adoptable prior record".to_owned()),
486        });
487    }
488    AlignmentReport {
489        entries,
490        resume_point: Some(view.frontier.run_path.clone()),
491        requires_confirmation: Vec::new(),
492    }
493}
494
495// ─── same-IR resume: frame-precise adoption (07 §4.6) ───────────────────────
496
497/// Resumes under the identical IR: every completed step instance is
498/// adopted by its exact run path; open spans and live call frames are
499/// re-entered without re-appending their events; the walk lands on the
500/// frontier position inside any depth of nesting — no frame restarts, no
501/// snapshot re-evaluation.
502async fn resume_same_ir(
503    loaded: LoadedFlow<'_>,
504    view: CheckpointView,
505    facts: Harvest,
506    run_id: &str,
507    mut session: Box<dyn ProviderSession>,
508    store: &mut Store,
509    opts: ResumeOptions,
510) -> Result<RunOutcome, RunnerError> {
511    let new_flow = loaded.root;
512    // The bind-time binding cursor (run-row meta, written once at
513    // begin_run, never rewritten): the issuing credential of intents
514    // dispatched before any resume (07 §4.5).
515    let bind_cursor = store.run_meta(run_id)?.binding.event_cursor;
516    // Adoption set: completed instances keyed by their instance path.
517    // Liveness wins (07 §5.2): an instance left with an open span was
518    // re-executed after its record was written and suspended inside that
519    // re-execution; the open span (and its frontier work) is the live
520    // state, so the stale record is never adoptable.
521    let open = open_instances(&facts);
522    let mut adoptable: BTreeMap<String, Adopted> = BTreeMap::new();
523    for record in &view.completed {
524        let key = instance_key(&record.run_path);
525        if open.contains(&key) {
526            continue;
527        }
528        adoptable.insert(
529            key.clone(),
530            Adopted {
531                record: record.clone(),
532                before_id: facts.before_observation.get(&key).cloned(),
533                after_id: facts.after_observation.get(&key).cloned(),
534            },
535        );
536    }
537    let open_spans: BTreeMap<String, Value> = facts
538        .open_spans
539        .iter()
540        .map(|path| {
541            let key = instance_key(path);
542            let inputs = facts
543                .entered_inputs
544                .get(&key)
545                .cloned()
546                .unwrap_or(Value::Null);
547            (key, inputs)
548        })
549        .collect();
550    // A live hook-launched repair frame (a suspension *inside* a repair
551    // subflow) needs hook-aware frame re-entry — a typed M2 refusal, never
552    // a guess (the repair's own records stay archived and honest).
553    if let Some(live_hook) = facts
554        .live_frames
555        .iter()
556        .find(|path| path.iter().any(|f| matches!(f, PathFrame::Hook { .. })))
557    {
558        return Err(RunnerError::M0Unsupported {
559            detail: format!(
560                "resume across a live handler-repair frame ({}) is not in the M2 subset — \
561                 the repair flow suspended mid-flight; hook-aware frame re-entry is \
562                 registered for the repair wave",
563                pointlock_ir::render_run_path(live_hook)
564            ),
565        });
566    }
567    let live_frames = live_frame_pins(&facts);
568
569    // The alignment report of a same-IR resume (shared with the
570    // read-only preview — one classification truth source).
571    let mut report = same_ir_report(new_flow, &facts, &view);
572
573    // (B) unconditional reconcile of a pending intent (07 §4.1/§4.4). The
574    // frontier step is where the walk will land (everything before it is
575    // adopted), so `at_resume` holds by construction.
576    let mut frontier_work = None;
577    let mut deferred_settle = None;
578    let mut pending_adjudication: Option<Box<Adjudication>> = None;
579    let mut blocked = None;
580    if let Some(intent) = &view.frontier.pending_intent {
581        let frontier_key = instance_key(&view.frontier.run_path);
582        let new_step = loaded.resolve_action(&view.frontier.run_path);
583        // Same-IR: the archived entered hash matches the resolved step's
584        // by construction; a missing carrier fails closed (dirty).
585        let effect_dirty = match (new_step, facts.entered_effect_hash.get(&frontier_key)) {
586            (Some(step), Some(archived)) => *archived != step.base.effect_hash,
587            _ => true,
588        };
589        let decision = match reconcile_frontier(
590            &mut session,
591            new_step,
592            true,
593            effect_dirty,
594            &view,
595            &facts,
596            &bind_cursor,
597            &mut report,
598            intent,
599        )
600        .await
601        {
602            Ok(decision) => decision,
603            Err(error) => {
604                let _ = session.end(SessionOutcome::Shutdown, None).await;
605                return Err(error);
606            }
607        };
608        match decision {
609            FrontierDecision::Work(work) => frontier_work = Some((frontier_key, work)),
610            FrontierDecision::DeferredSettle(settle) => deferred_settle = Some(settle),
611            FrontierDecision::Adjudicate(adjudication) => pending_adjudication = Some(adjudication),
612            FrontierDecision::Blocked(reason) => blocked = Some(reason),
613            FrontierDecision::Nothing => {}
614        }
615    }
616
617    // The segment header: runResumed carries the alignment report, this
618    // segment's supervisePolicy (explicitly null when unsupervised —
619    // R13), and the new generation's reseeded cursor (07 §4.5: taken
620    // after the reconcile decisions, before this append; absent when the
621    // RPC fails — honest, never stale).
622    let resumed_cursor = session.current_cursor().await.ok();
623    store.append_event(
624        run_id,
625        now_ms(),
626        &root_path(new_flow),
627        &RunLogPayload::RunResumed {
628            alignment_report: report.clone(),
629            supervise_policy: opts.supervise,
630            event_cursor: resumed_cursor,
631        },
632    )?;
633
634    // A reconciled completed terminal that cannot be adopted at the
635    // resume point is still recorded — the ledger closes the intent and
636    // keeps the world fact as evidence (07 §4.1).
637    if let Some((path, call_id, outcome)) = deferred_settle {
638        store.append_event(
639            run_id,
640            now_ms(),
641            &path,
642            &RunLogPayload::ActionSettled {
643                call_id,
644                outcome: crate::engine::quarantine_unpersistable(*outcome),
645            },
646        )?;
647    }
648
649    if let Some(adjudication) = pending_adjudication {
650        // Phase 1 of the 07 §4.4 default escalation: the request (fresh or
651        // re-awaited) is the segment's outcome — the run suspends
652        // `awaitingHuman` and the answer arrives through the ordinary
653        // arbitration channel, durable for the next resume to consume.
654        let Adjudication {
655            run_path,
656            request,
657            pending,
658        } = *adjudication;
659        if let Some((request_id, prompt, presents)) = request {
660            store.append_event(
661                run_id,
662                now_ms(),
663                &run_path,
664                &RunLogPayload::HumanRequested {
665                    request_id,
666                    purpose: pointlock_ir::HumanPurpose::Step,
667                    mode: Some(pointlock_ir::vocab::HumanMode::RepairWorld),
668                    prompt,
669                    presents,
670                    decisions: Some(vec![
671                        "adopt".to_owned(),
672                        "redo".to_owned(),
673                        "abort".to_owned(),
674                    ]),
675                    output_schema: None,
676                    deadline_at_ms: None,
677                },
678            )?;
679        }
680        let summary = crate::engine::capture_provider_state_summary(
681            session.as_ref(),
682            &view.binding.session_lineage,
683            &view.binding.device_id,
684            opts.platform.as_deref(),
685        )
686        .await;
687        store.append_event(
688            run_id,
689            now_ms(),
690            &root_path(new_flow),
691            &RunLogPayload::RunSuspended {
692                provider_state_summary: Some(summary),
693                reason: Some(format!(
694                    "awaiting human adjudication (requestId {})",
695                    pending.request_id
696                )),
697            },
698        )?;
699        let _ = session.end(SessionOutcome::Shutdown, None).await;
700        return Ok(RunOutcome::AwaitingHuman { pending });
701    }
702
703    if let Some(reason) = blocked {
704        // Suspension-instant profile (07 §2.2): the session is still
705        // live at this pre-Execution blocked refusal.
706        let summary = crate::engine::capture_provider_state_summary(
707            session.as_ref(),
708            &view.binding.session_lineage,
709            &view.binding.device_id,
710            opts.platform.as_deref(),
711        )
712        .await;
713        store.append_event(
714            run_id,
715            now_ms(),
716            &root_path(new_flow),
717            &RunLogPayload::RunSuspended {
718                provider_state_summary: Some(summary),
719                reason: Some(reason.to_string()),
720            },
721        )?;
722        let _ = session.end(SessionOutcome::Shutdown, None).await;
723        return Ok(RunOutcome::Blocked { reason });
724    }
725
726    let params = params_object(&view);
727    let env = env_bindings(&view.binding.device_id, opts.platform.as_deref(), run_id);
728    let exec = Execution {
729        flows: &loaded,
730        session,
731        store,
732        run_id: run_id.to_owned(),
733        stop: opts.stop,
734        env,
735        attempt_base: facts.max_attempt.clone(),
736        open_spans,
737        live_frames,
738        resumed: true,
739        authorized: opts.allow_mutating_reexec.iter().cloned().collect(),
740        reentry_seen: false,
741        adoptable,
742        frontier: frontier_work,
743        session_lineage: view.binding.session_lineage.clone(),
744        pending_summaries: BTreeMap::new(),
745        vision: opts.vision.clone(),
746        supervise: opts.supervise,
747        human: facts.human_requests.clone(),
748        settled: facts.settled.clone(),
749        recorded_verdicts: facts.recorded_verdicts.clone(),
750        hook_triggers: facts.hook_triggers.clone(),
751        clock: opts.clock,
752    };
753    let root = FrameState::new(new_flow, root_path(new_flow), params, 1);
754    exec.run(root, 0).await
755}
756
757// ─── cross-IR resume: the flat alignment subset (07 §5.2) ────────────────────
758
759/// Resumes under a repaired IR. M2 subset: the old records and the new
760/// body must both be flat top-level action steps; anything nested is a
761/// typed refusal (the 07 §5.2 nested alignment rules land with the repair
762/// wave).
763async fn resume_cross_ir(
764    loaded: LoadedFlow<'_>,
765    view: CheckpointView,
766    facts: Harvest,
767    run_id: &str,
768    mut session: Box<dyn ProviderSession>,
769    store: &mut Store,
770    opts: ResumeOptions,
771) -> Result<RunOutcome, RunnerError> {
772    let new_flow = loaded.root;
773    // Bind-time credential, as in resume_same_ir (07 §4.5).
774    let bind_cursor = store.run_meta(run_id)?.binding.event_cursor;
775    // The FRONTIER may not sit inside a handler frame: resolving it means
776    // walking a path `resolve_step` refuses by construction, and the
777    // reconcile below would then have no step to reconcile against.
778    // Completed hook-framed records are a different matter — see
779    // [`is_alignable_path`].
780    if !is_alignable_path(&view.frontier.run_path) {
781        let _ = session.end(SessionOutcome::Shutdown, None).await;
782        return Err(RunnerError::M0Unsupported {
783            detail: format!(
784                "the run's frontier sits inside a handler frame ({}); hook-aware frame \
785                 re-entry is registered for the repair wave",
786                pointlock_ir::render_run_path(&view.frontier.run_path)
787            ),
788        });
789    }
790
791    // Unfinished handler work in ANY of its three shapes is a categorical
792    // refusal, and it is settled BEFORE alignment runs — `align` can return
793    // `RequiresConfirmation`, and letting that mask a resume the runner
794    // would refuse outright would tell the operator to authorize step ids
795    // for something that can never proceed. It is also the order
796    // `align_preview` uses, and the preview promises to mirror resume's
797    // typed refusals.
798    //
799    // (i) a repair subflow suspended mid-flight — its call frame is still
800    // open.
801    if let Some(live_hook) = facts
802        .live_frames
803        .iter()
804        .find(|path| path.iter().any(|f| matches!(f, PathFrame::Hook { .. })))
805    {
806        let _ = session.end(SessionOutcome::Shutdown, None).await;
807        return Err(RunnerError::M0Unsupported {
808            detail: format!(
809                "resume across a live handler-repair frame ({}) is not in the M2 subset — \
810                 the repair flow suspended mid-flight; hook-aware frame re-entry is \
811                 registered for the repair wave",
812                pointlock_ir::render_run_path(live_hook)
813            ),
814        });
815    }
816    // (ii) an escalate hook human still awaiting an answer. It leaves NO
817    // other trace: it opens no span and pushes no frame (「hook humans are
818    // not body steps」), so `live_frames`, `frontier` and `completed` are
819    // all blind to it — `humanPending` is the only carrier. Cross-IR it is
820    // genuinely unsafe: the continuation is looked up by an instance key
821    // rebuilt from the NEW host path, so renaming the host mints a SECOND
822    // request and strands the first unanswerable, and deleting the host
823    // strands it forever. Same-IR rebuilds the same key and settles
824    // correctly, which is why this refusal lives here and not there.
825    if let Some(pending) = view
826        .human_pending
827        .as_ref()
828        .filter(|pending| !is_alignable_path(&pending.run_path))
829    {
830        let _ = session.end(SessionOutcome::Shutdown, None).await;
831        return Err(RunnerError::M0Unsupported {
832            detail: format!(
833                "a handler escalation is still awaiting an answer ({}); resuming it under a \
834                 repaired IR needs hook-aware re-entry, which is registered for the repair \
835                 wave — answer or let it time out first",
836                pointlock_ir::render_run_path(&pending.run_path)
837            ),
838        });
839    }
840
841    let seed = ScopeSeed::new(
842        params_object(&view),
843        &view.binding.device_id,
844        opts.platform.as_deref(),
845        run_id,
846    );
847
848    // (A) alignment first (07 §4.1). Classification runs on the archived
849    // per-step hashes the fold harvested from `stepEntered` (spine §6.1
850    // M1 note) — the old FlowIR is not required.
851    let mut alignment = match align(
852        &loaded,
853        &new_flow.body,
854        &view,
855        &facts,
856        &seed,
857        store,
858        opts.vision.as_deref(),
859        &opts.allow_mutating_reexec,
860        &opts.force_reexecute,
861        opts.old_flow_ir.as_ref(),
862    )
863    .await
864    {
865        Ok(alignment) => alignment,
866        Err(error) => {
867            // A pre-header refusal must not leak the opened session
868            // (best-effort teardown, 04 §2.1).
869            let _ = session.end(SessionOutcome::Shutdown, None).await;
870            return Err(error);
871        }
872    };
873
874    // (B) unconditional reconcile of a pending intent (07 §4.1/§4.4).
875    let mut frontier_work = None;
876    let mut deferred_settle = None;
877    let mut pending_adjudication: Option<Box<Adjudication>> = None;
878    let mut blocked = None;
879    if let Some(intent) = &view.frontier.pending_intent {
880        let frontier_key = instance_key(&view.frontier.run_path);
881        // Resolved by PATH, not by a flat id scan: the frontier can sit
882        // inside a branch, and same-IR resume already resolves it this way.
883        let new_step = loaded.resolve_action(&view.frontier.run_path);
884        // The frontier IS the resume point when its instance is the one
885        // alignment named. Instance keys, not body indices: the comparison
886        // has to keep working once the resume point can sit inside a
887        // callee or an iteration.
888        let at_resume = alignment.resume_key.as_deref() == Some(frontier_key.as_str());
889        // §4.1 cross semantics: an effect-dirty frontier step's old result
890        // is never adopted — it is the product of the old binding. A
891        // missing hash or a frontier step absent from the new IR fails
892        // closed (dirty).
893        let effect_dirty = match (new_step, facts.entered_effect_hash.get(&frontier_key)) {
894            (Some(step), Some(archived)) => *archived != step.base.effect_hash,
895            _ => true,
896        };
897        let decision = match reconcile_frontier(
898            &mut session,
899            new_step,
900            at_resume,
901            effect_dirty,
902            &view,
903            &facts,
904            &bind_cursor,
905            &mut alignment.report,
906            intent,
907        )
908        .await
909        {
910            Ok(decision) => decision,
911            Err(error) => {
912                let _ = session.end(SessionOutcome::Shutdown, None).await;
913                return Err(error);
914            }
915        };
916        match decision {
917            FrontierDecision::Work(work) => frontier_work = Some((frontier_key, work)),
918            FrontierDecision::DeferredSettle(settle) => deferred_settle = Some(settle),
919            FrontierDecision::Adjudicate(adjudication) => pending_adjudication = Some(adjudication),
920            FrontierDecision::Blocked(reason) => blocked = Some(reason),
921            FrontierDecision::Nothing => {}
922        }
923    }
924
925    // The segment header (see the same-IR site for the cursor semantics).
926    let resumed_cursor = session.current_cursor().await.ok();
927    store.append_event(
928        run_id,
929        now_ms(),
930        &root_path(new_flow),
931        &RunLogPayload::RunResumed {
932            alignment_report: alignment.report.clone(),
933            supervise_policy: opts.supervise,
934            event_cursor: resumed_cursor,
935        },
936    )?;
937
938    // Offline re-judgements: new verdicts with `supersedes` lineage,
939    // anchored at the old records' run paths (the fold re-projects the
940    // completed records); written back via the *current* session
941    // (07 §5.3 — cross-session write-back is sound, the daemon only
942    // persists).
943    let rejudged = std::mem::take(&mut alignment.rejudged);
944    for rejudge in rejudged {
945        // Remote archival first so its outcome rides the event; a
946        // failure is annotation material, never a resume error (04 §5 —
947        // the RunLog is the sole truth). Wire caps applied here like on
948        // every other write-back: compaction is the runner's job (04 §5).
949        let remote_archival_error = session
950            .record_verdict(pointlock_provider_kit::VerdictWrite {
951                status: rejudge.verdict.status,
952                summary: crate::engine::cap_wire_summary(&rejudge.verdict),
953                evidence: rejudge
954                    .verdict
955                    .evidence
956                    .iter()
957                    .take(pointlock_provider_kit::VERDICT_EVIDENCE_MAX_ENTRIES)
958                    .cloned()
959                    .collect(),
960            })
961            .await
962            .err()
963            .map(|error| format!("remote archival failed: {error}"));
964        store.append_event(
965            run_id,
966            now_ms(),
967            &rejudge.run_path,
968            &RunLogPayload::VerdictRecorded {
969                verdict: rejudge.verdict.clone(),
970                localized: Vec::new(),
971                localization_gaps: Vec::new(),
972                remote_archival_error,
973            },
974        )?;
975    }
976
977    // A reconciled completed terminal that cannot be adopted at the
978    // resume point is still recorded — the ledger closes the intent
979    // and keeps the world fact as evidence (07 §4.1).
980    if let Some((path, call_id, outcome)) = deferred_settle {
981        store.append_event(
982            run_id,
983            now_ms(),
984            &path,
985            &RunLogPayload::ActionSettled {
986                call_id,
987                outcome: crate::engine::quarantine_unpersistable(*outcome),
988            },
989        )?;
990    }
991
992    if let Some(adjudication) = pending_adjudication {
993        // Phase 1 of the 07 §4.4 default escalation: the request (fresh or
994        // re-awaited) is the segment's outcome — the run suspends
995        // `awaitingHuman` and the answer arrives through the ordinary
996        // arbitration channel, durable for the next resume to consume.
997        let Adjudication {
998            run_path,
999            request,
1000            pending,
1001        } = *adjudication;
1002        if let Some((request_id, prompt, presents)) = request {
1003            store.append_event(
1004                run_id,
1005                now_ms(),
1006                &run_path,
1007                &RunLogPayload::HumanRequested {
1008                    request_id,
1009                    purpose: pointlock_ir::HumanPurpose::Step,
1010                    mode: Some(pointlock_ir::vocab::HumanMode::RepairWorld),
1011                    prompt,
1012                    presents,
1013                    decisions: Some(vec![
1014                        "adopt".to_owned(),
1015                        "redo".to_owned(),
1016                        "abort".to_owned(),
1017                    ]),
1018                    output_schema: None,
1019                    deadline_at_ms: None,
1020                },
1021            )?;
1022        }
1023        let summary = crate::engine::capture_provider_state_summary(
1024            session.as_ref(),
1025            &view.binding.session_lineage,
1026            &view.binding.device_id,
1027            opts.platform.as_deref(),
1028        )
1029        .await;
1030        store.append_event(
1031            run_id,
1032            now_ms(),
1033            &root_path(new_flow),
1034            &RunLogPayload::RunSuspended {
1035                provider_state_summary: Some(summary),
1036                reason: Some(format!(
1037                    "awaiting human adjudication (requestId {})",
1038                    pending.request_id
1039                )),
1040            },
1041        )?;
1042        let _ = session.end(SessionOutcome::Shutdown, None).await;
1043        return Ok(RunOutcome::AwaitingHuman { pending });
1044    }
1045
1046    if let Some(reason) = blocked {
1047        // Suspension-instant profile (07 §2.2): the session is still
1048        // live at this pre-Execution blocked refusal.
1049        let summary = crate::engine::capture_provider_state_summary(
1050            session.as_ref(),
1051            &view.binding.session_lineage,
1052            &view.binding.device_id,
1053            opts.platform.as_deref(),
1054        )
1055        .await;
1056        store.append_event(
1057            run_id,
1058            now_ms(),
1059            &root_path(new_flow),
1060            &RunLogPayload::RunSuspended {
1061                provider_state_summary: Some(summary),
1062                reason: Some(reason.to_string()),
1063            },
1064        )?;
1065        let _ = session.end(SessionOutcome::Shutdown, None).await;
1066        return Ok(RunOutcome::Blocked { reason });
1067    }
1068
1069    let Alignment {
1070        adoptable,
1071        teardown,
1072        ..
1073    } = alignment;
1074    // 07 §5.2 case (b): dismantle the stale frame ON THE LEDGER before
1075    // execution starts, mirroring `exec_call`'s abort unwind exactly —
1076    // close the open spans innermost-first (the fold's exit pairing is
1077    // LIFO), pop each live frame right before its own call span closes,
1078    // and let the call step exit `aborted` (an aborted execution makes no
1079    // semantic claim; nothing here is adoptable history). Emitted AFTER
1080    // the deferred settle above, so a terminal the reconcile closed lands
1081    // on the still-open frontier span and is archived with it.
1082    //
1083    // The suspension chain is one nested sequence, so "the torn-down
1084    // subtree" is precisely the open spans at or under the call's own key.
1085    let torn = |key: &str| -> bool {
1086        teardown
1087            .as_deref()
1088            .is_some_and(|call| key == call || crate::align::is_instance_descendant(call, key))
1089    };
1090    if teardown.is_some() {
1091        let live_keys: BTreeSet<String> = facts
1092            .live_frames
1093            .iter()
1094            .map(|path| instance_key(path))
1095            .collect();
1096        for span in facts.open_spans.iter().rev() {
1097            let key = instance_key(span);
1098            if !torn(&key) {
1099                continue;
1100            }
1101            if live_keys.contains(&key) {
1102                // The span belongs to a call step whose frame is open: the
1103                // frame pops first, the span closes second — the exact
1104                // unwind order of a live abort.
1105                store.append_event(
1106                    run_id,
1107                    now_ms(),
1108                    span,
1109                    &RunLogPayload::CallFramePopped { outputs: None },
1110                )?;
1111            }
1112            store.append_event(
1113                run_id,
1114                now_ms(),
1115                span,
1116                &RunLogPayload::StepExited {
1117                    provider_state_summary: None,
1118                    state: pointlock_ir::StepState::Aborted,
1119                    output: None,
1120                    localized: Vec::new(),
1121                    localization_gaps: Vec::new(),
1122                },
1123            )?;
1124        }
1125    }
1126    let open_spans: BTreeMap<String, Value> = facts
1127        .open_spans
1128        .iter()
1129        .filter(|path| !torn(&instance_key(path)))
1130        .map(|path| {
1131            let key = instance_key(path);
1132            let inputs = facts
1133                .entered_inputs
1134                .get(&key)
1135                .cloned()
1136                .unwrap_or(Value::Null);
1137            (key, inputs)
1138        })
1139        .collect();
1140    // The torn-down frame is gone from the ledger; handing its pin to the
1141    // engine would make `exec_call` skip the push for a frame that no
1142    // longer exists.
1143    let live_frames: BTreeMap<String, pointlock_ir::Hash> = live_frame_pins(&facts)
1144        .into_iter()
1145        .filter(|(key, _)| !torn(key))
1146        .collect();
1147    let params = params_object(&view);
1148    let env = env_bindings(&view.binding.device_id, opts.platform.as_deref(), run_id);
1149    let exec = Execution {
1150        flows: &loaded,
1151        session,
1152        store,
1153        run_id: run_id.to_owned(),
1154        stop: opts.stop,
1155        env,
1156        attempt_base: facts.max_attempt.clone(),
1157        open_spans,
1158        // Live call frames must not be pushed again on resume (07 §4.6);
1159        // the pin lets `exec_call` tell a plain re-entry from one that has
1160        // to rebase the frame onto a repaired callee (07 §5.2 case (a)).
1161        // The torn-down frame (case (b)) is filtered out above.
1162        live_frames,
1163        resumed: true,
1164        authorized: opts.allow_mutating_reexec.iter().cloned().collect(),
1165        reentry_seen: false,
1166        adoptable,
1167        frontier: frontier_work,
1168        session_lineage: view.binding.session_lineage.clone(),
1169        pending_summaries: BTreeMap::new(),
1170        vision: opts.vision.clone(),
1171        supervise: opts.supervise,
1172        human: facts.human_requests.clone(),
1173        settled: facts.settled.clone(),
1174        recorded_verdicts: facts.recorded_verdicts.clone(),
1175        hook_triggers: facts.hook_triggers.clone(),
1176        clock: opts.clock,
1177    };
1178    // Execution restarts at the top of the body; the adoption set does the
1179    // skipping, seeding each adopted step's output/verdict into its OWN
1180    // frame as it is reached. That is the same mechanism same-IR resume
1181    // uses, and the only one that can express a resume point at depth.
1182    let root = FrameState::new(new_flow, root_path(new_flow), params, 1);
1183    exec.run(root, 0).await
1184}
1185
1186/// Whether cross-IR alignment can ADDRESS this path.
1187///
1188/// Exactly a hook guard, and says so rather than re-listing the seven
1189/// frames it accepts: the walker descends `if` branch bodies, `foreach`
1190/// rounds, and — under the case (a) down-drill — callee bodies, addressing
1191/// every one of them by instance key, so `flow`/`step`/`call`/`iteration`
1192/// (and the attempt/phase/assertion suffixes) are all classifiable. `hook`
1193/// is the one frame shape nothing addresses.
1194///
1195/// Applied to the FRONTIER only. Completed hook-framed records are not
1196/// refused — 07 §5.2's last bullet rules 「hook 帧下的记录(handler 审计
1197/// 痕)不参与对齐复用……旧 hook 记录一律归档」: archive them, do not refuse
1198/// the resume. Refusing cost a real case — a run whose `onFail` repair
1199/// subflow completed could never be repaired cross-IR afterwards — and
1200/// archival is already structural rather than a promise:
1201/// - they are never ADOPTED: adoption is keyed by instance, and node keys
1202///   come from `child_frame`, which emits only `step`/`call`/`iteration`
1203///   segments. `instance_key` renders a hook frame as `/hook:<Hook>:<n>`,
1204///   which no `StepId` can spell, so no node key can ever collide;
1205/// - they are never ORPHAN-reported: the only hook-framed `StepRecord`s
1206///   come from a repair subflow's body, whose path always carries the
1207///   handler-launched `call` frame, and the orphan pass skips records
1208///   under a call frame the walk did not descend into. An escalate human
1209///   writes `humanRequested` and no step span at all, so it contributes
1210///   no record to misreport.
1211///
1212/// A LIVE hook frame is still refused, separately and before this: a
1213/// repair subflow suspended mid-flight needs hook-aware frame re-entry,
1214/// which is the repair wave's.
1215fn is_alignable_path(path: &RunPath) -> bool {
1216    !path
1217        .iter()
1218        .any(|frame| matches!(frame, PathFrame::Hook { .. }))
1219}
1220
1221/// A pending human adjudication of an uncertain reconcile (07 §4.4): the
1222/// run suspends `awaitingHuman` on a synthesized `repairWorld` request
1223/// whose vocabulary is `adopt | redo | abort` (00 §6.7-B). Paired to its
1224/// intent BY CALL ID (carried in `presents`), so an answer ruled for one
1225/// dispatch can never be replayed onto a later one.
1226struct Adjudication {
1227    /// The hook-framed anchor (`<frontier>/hook:OnResumeDrift:1/adjudicate`).
1228    run_path: RunPath,
1229    /// A fresh request to append — `(requestId, prompt, presents)`; `None`
1230    /// when an unanswered request for this callId is already on the ledger
1231    /// and the segment simply re-awaits it.
1232    request: Option<(String, String, Value)>,
1233    /// What the segment reports as the pending interaction.
1234    pending: pointlock_ir::HumanPending,
1235}
1236
1237/// What the frontier reconcile decided.
1238enum FrontierDecision {
1239    /// Mid-flight work for the resume step.
1240    Work(FrontierWork),
1241    /// Close the intent in the ledger with the archived terminal; the step
1242    /// re-executes fresh.
1243    DeferredSettle(
1244        (
1245            pointlock_ir::RunPath,
1246            String,
1247            Box<pointlock_ir::ActionOutcome>,
1248        ),
1249    ),
1250    /// Human adjudication required: suspend `awaitingHuman` on the
1251    /// adjudication request (fresh or re-awaited).
1252    Adjudicate(Box<Adjudication>),
1253    /// Human adjudication impossible to even request (defense line).
1254    Blocked(BlockedReason),
1255    /// Nothing to carry over (e.g. neverDispatched off the resume point).
1256    Nothing,
1257}
1258
1259/// Applies the 07 §4.4 decision table to a pending intent. `new_step` is
1260/// the frontier step as resolved in the new IR (nested paths supported);
1261/// `at_resume` states whether execution will land exactly on it;
1262/// `effect_dirty` is the §4.1 cross-semantics discriminator.
1263#[allow(clippy::too_many_arguments)]
1264async fn reconcile_frontier(
1265    session: &mut Box<dyn ProviderSession>,
1266    new_step: Option<&ActionStepIR>,
1267    at_resume: bool,
1268    effect_dirty: bool,
1269    view: &CheckpointView,
1270    facts: &Harvest,
1271    bind_cursor: &pointlock_ir::EventCursor,
1272    report: &mut AlignmentReport,
1273    intent: &pointlock_ir::PendingIntent,
1274) -> Result<FrontierDecision, RunnerError> {
1275    // The issuing credential (07 §4.5): per-intent exact state from the
1276    // ledger scan. `FromBinding` (no resume preceded the intent) resolves
1277    // to the BIND-TIME cursor — the run-row binding written once at
1278    // begin_run — NOT the folded view's cursor, which every
1279    // cursor-bearing resume reseeds to the newest generation (a
1280    // generation that never issued this intent). A missing harvest entry
1281    // means the ledger cannot attest the issuing generation at all:
1282    // fail-closed to Unknown, never a fabricated credential. Unknown is
1283    // answered with the uncertain branch WITHOUT an RPC.
1284    let issuing = facts
1285        .intent_issuing
1286        .get(&intent.call_id)
1287        .cloned()
1288        .unwrap_or(crate::align::IssuingCursor::Unknown);
1289    let fate = match &issuing {
1290        crate::align::IssuingCursor::FromBinding => {
1291            session.reconcile(&intent.call_id, bind_cursor).await?
1292        }
1293        crate::align::IssuingCursor::Known(cursor) => {
1294            session.reconcile(&intent.call_id, cursor).await?
1295        }
1296        crate::align::IssuingCursor::Unknown => ReconcileResult::LogUnavailable {
1297            reason: "the issuing session is unknowable (a resume predating the \
1298                     eventCursor carrier intervened); refusing to reconcile with \
1299                     a fabricated credential"
1300                .to_owned(),
1301        },
1302    };
1303    let intent_path = facts
1304        .intent_path
1305        .get(&intent.call_id)
1306        .cloned()
1307        .unwrap_or_else(|| view.frontier.run_path.clone());
1308    let mutating_gated = new_step.map(gated_mutating).unwrap_or(true);
1309
1310    match fate {
1311        ReconcileResult::Completed { outcome } => {
1312            if !effect_dirty && at_resume {
1313                // The archived terminal — whatever its four-way
1314                // discriminant — is adopted and disposed through the same
1315                // settled-outcome path as a live execute (§6.7-B).
1316                return Ok(FrontierDecision::Work(FrontierWork::Adopt {
1317                    call_id: intent.call_id.clone(),
1318                    intent_path,
1319                    outcome,
1320                    args: intent.args_snapshot.clone(),
1321                    chain_index: facts.intent_chain_index.get(&intent.call_id).copied(),
1322                }));
1323            }
1324            // Not adoptable at the resume point. Whether re-execution
1325            // risks a second effect follows the 07 §5.4 criterion: only a
1326            // succeeded or timedOut terminal can have mutated the world;
1327            // an archived failed/cancelled left no effect to double.
1328            let effect_possible = matches!(
1329                outcome.as_ref(),
1330                pointlock_ir::ActionOutcome::Succeeded { .. }
1331                    | pointlock_ir::ActionOutcome::TimedOut { .. }
1332            );
1333            if effect_possible && mutating_gated {
1334                // The old action (possibly) took effect but its terminal
1335                // cannot be adopted (effect-dirty or positionally
1336                // invalidated): re-execution is a second effect —
1337                // 07 §5.4 `frontierUnknown`, fail-closed.
1338                report.requires_confirmation.push(RequiresConfirmation {
1339                    run_path: view.frontier.run_path.clone(),
1340                    step_id: new_step.map(|step| step.base.step_id.clone()),
1341                    cause: "frontierUnknown".to_owned(),
1342                    reason: format!(
1343                        "callId {} reached a recorded {} terminal on the device but \
1344                         it is not adoptable; re-execution of the mutating step \
1345                         needs explicit authorization",
1346                        intent.call_id,
1347                        outcome.kind()
1348                    ),
1349                });
1350                return Err(RunnerError::RequiresConfirmation {
1351                    report: Box::new(report.clone()),
1352                });
1353            }
1354            Ok(FrontierDecision::DeferredSettle((
1355                intent_path,
1356                intent.call_id.clone(),
1357                outcome,
1358            )))
1359        }
1360        ReconcileResult::NeverDispatched => {
1361            if !effect_dirty && at_resume {
1362                // Safe replay: archived args, new callId, new WAL intent.
1363                Ok(FrontierDecision::Work(FrontierWork::Replay {
1364                    chain_index: facts.intent_chain_index.get(&intent.call_id).copied(),
1365                    args: intent.args_snapshot.clone(),
1366                }))
1367            } else {
1368                // The step re-executes fresh from ready (nothing happened
1369                // in the world).
1370                Ok(FrontierDecision::Nothing)
1371            }
1372        }
1373        ReconcileResult::StartedNoTerminal => uncertain_branch(
1374            new_step,
1375            intent,
1376            &view.frontier.run_path,
1377            facts,
1378            report,
1379            at_resume,
1380            effect_dirty,
1381            "startedNoTerminal",
1382            facts.intent_chain_index.get(&intent.call_id).copied(),
1383        ),
1384        ReconcileResult::LogUnavailable { reason } => uncertain_branch(
1385            new_step,
1386            intent,
1387            &view.frontier.run_path,
1388            facts,
1389            report,
1390            at_resume,
1391            effect_dirty,
1392            &format!("logUnavailable: {reason}"),
1393            facts.intent_chain_index.get(&intent.call_id).copied(),
1394        ),
1395    }
1396}
1397
1398/// The uncertain reconcile branch (07 §4.4): replay only with the explicit
1399/// author permission (`idempotent` / `readonly`); otherwise the DEFAULT
1400/// `onResumeDrift` escalation — a synthesized `repairWorld` human rules
1401/// `adopt | redo | abort` over the presented callId (00 §6.7-B). The
1402/// request and its answer live on the ordinary human ledger
1403/// (`humanRequested`/`humanResponded`), so the operator answers through
1404/// the same channels as any other wait and the ruling is durable: a crash
1405/// after the answer re-derives the same disposition.
1406///
1407/// A DECLARED `onResumeDrift` binding keeps serving the probe-drift ladder
1408/// it was written for; routing the reconcile adjudication through custom
1409/// bindings is registered for the repair wave.
1410#[allow(clippy::too_many_arguments)]
1411fn uncertain_branch(
1412    new_step: Option<&ActionStepIR>,
1413    intent: &pointlock_ir::PendingIntent,
1414    frontier_path: &RunPath,
1415    facts: &Harvest,
1416    report: &mut AlignmentReport,
1417    at_resume: bool,
1418    effect_dirty: bool,
1419    fate: &str,
1420    chain_index: Option<u32>,
1421) -> Result<FrontierDecision, RunnerError> {
1422    let permitted = new_step.map(replay_permitted).unwrap_or(false);
1423    if permitted {
1424        if at_resume && !effect_dirty {
1425            return Ok(FrontierDecision::Work(FrontierWork::Replay {
1426                chain_index,
1427                args: intent.args_snapshot.clone(),
1428            }));
1429        }
1430        // Fresh re-execution is equally safe for readonly/idempotent.
1431        return Ok(FrontierDecision::Nothing);
1432    }
1433
1434    // The adjudication anchor: one hook-framed instance under the frontier
1435    // step. The leaf id is fixed — identity per INTENT comes from the
1436    // callId carried in `presents`, checked below, so an answer ruled for
1437    // an earlier dispatch is never replayed onto this one.
1438    let mut hook_path = frontier_path.clone();
1439    hook_path.push(PathFrame::Hook {
1440        hook: pointlock_ir::HandlerHook::OnResumeDrift,
1441        trigger: 1,
1442    });
1443    hook_path.push(PathFrame::Step {
1444        step_id: "adjudicate".try_into().expect("a fixed valid step id"),
1445    });
1446    let key = instance_key(&hook_path);
1447
1448    if let Some(fact) = facts.human_requests.get(&key)
1449        && fact.presents.get("callId").and_then(Value::as_str) == Some(intent.call_id.as_str())
1450    {
1451        match &fact.final_response {
1452            None => {
1453                // Asked and unanswered: re-await the same request, no
1454                // duplicate append.
1455                return Ok(FrontierDecision::Adjudicate(Box::new(Adjudication {
1456                    run_path: hook_path.clone(),
1457                    request: None,
1458                    pending: pending_of(fact, &hook_path),
1459                })));
1460            }
1461            Some(response) => {
1462                let ruling = response.get("decision").and_then(Value::as_str);
1463                match ruling {
1464                    Some("adopt") => {
1465                        if at_resume && !effect_dirty {
1466                            // The ruled effect stands; the step's own
1467                            // assertions verify it over a fresh
1468                            // observation ([`FrontierWork::ConfirmEffect`]).
1469                            return Ok(FrontierDecision::Work(FrontierWork::ConfirmEffect {
1470                                message: format!(
1471                                    "uncertain fate ({fate}) of callId {} adjudicated \
1472                                     `adopt`",
1473                                    intent.call_id
1474                                ),
1475                                args: intent.args_snapshot.clone(),
1476                            }));
1477                        }
1478                        // Adopted effect on a step that must nonetheless
1479                        // re-execute (effect-dirty / positionally
1480                        // invalidated): a second effect — the 07 §5.4
1481                        // frontierUnknown gate, same as an unadoptable
1482                        // recorded terminal.
1483                        report.requires_confirmation.push(RequiresConfirmation {
1484                            run_path: frontier_path.clone(),
1485                            step_id: new_step.map(|step| step.base.step_id.clone()),
1486                            cause: "frontierUnknown".to_owned(),
1487                            reason: format!(
1488                                "callId {} was adjudicated `adopt` (the effect stands) but \
1489                                 the step is not adoptable here; re-execution of the \
1490                                 mutating step needs explicit authorization",
1491                                intent.call_id
1492                            ),
1493                        });
1494                        return Err(RunnerError::RequiresConfirmation {
1495                            report: Box::new(report.clone()),
1496                        });
1497                    }
1498                    Some("redo") => {
1499                        // I2 source (iv): the human's redo IS the license.
1500                        if at_resume && !effect_dirty {
1501                            return Ok(FrontierDecision::Work(FrontierWork::Replay {
1502                                chain_index,
1503                                args: intent.args_snapshot.clone(),
1504                            }));
1505                        }
1506                        return Ok(FrontierDecision::Nothing);
1507                    }
1508                    Some("abort") => {
1509                        return Ok(FrontierDecision::Work(FrontierWork::AbortRuled {
1510                            args: intent.args_snapshot.clone(),
1511                        }));
1512                    }
1513                    other => {
1514                        // The store arbitrates against the declared
1515                        // vocabulary, so this is a ledger anomaly — the
1516                        // defense line blocks rather than guesses.
1517                        return Ok(FrontierDecision::Blocked(BlockedReason::RequiresHuman {
1518                            call_id: intent.call_id.clone(),
1519                            detail: format!(
1520                                "adjudication response carries an unusable decision \
1521                                 {other:?}; refusing to guess"
1522                            ),
1523                        }));
1524                    }
1525                }
1526            }
1527        }
1528    }
1529
1530    // No adjudication asked yet (or the one on the ledger belongs to an
1531    // earlier dispatch): mint the request.
1532    let request_id = uuid::Uuid::new_v4().to_string();
1533    let prompt = format!(
1534        "the fate of callId {} is uncertain ({fate}) and the step is mutating and \
1535         not idempotent — automatic replay is forbidden (I2). Inspect the device, \
1536         then rule: `adopt` (the effect happened; verify and continue), `redo` \
1537         (the effect did not happen or you undid it; dispatch again), or `abort` \
1538         (stop the run)",
1539        intent.call_id
1540    );
1541    let presents = serde_json::json!({
1542        "callId": intent.call_id,
1543        "fate": fate,
1544        "argsSnapshot": intent.args_snapshot,
1545    });
1546    let pending = pointlock_ir::HumanPending {
1547        run_path: hook_path.clone(),
1548        request_id: request_id.clone(),
1549        purpose: pointlock_ir::HumanPurpose::Step,
1550        mode: Some(pointlock_ir::vocab::HumanMode::RepairWorld),
1551        prompt: prompt.clone(),
1552        deadline_at_ms: None,
1553    };
1554    Ok(FrontierDecision::Adjudicate(Box::new(Adjudication {
1555        run_path: hook_path,
1556        request: Some((request_id, prompt, presents)),
1557        pending,
1558    })))
1559}
1560
1561/// The pending descriptor of an already-asked adjudication.
1562fn pending_of(fact: &HumanRequestFact, hook_path: &RunPath) -> pointlock_ir::HumanPending {
1563    pointlock_ir::HumanPending {
1564        run_path: hook_path.clone(),
1565        request_id: fact.request_id.clone(),
1566        purpose: fact.purpose,
1567        mode: fact.mode,
1568        prompt: fact.prompt.clone(),
1569        deadline_at_ms: fact.deadline_at_ms,
1570    }
1571}
1572
1573/// The params snapshot of a checkpoint as an object map (it was written by
1574/// `Runner::run` as an object; anything else folds to empty).
1575fn params_object(view: &CheckpointView) -> Map<String, Value> {
1576    match &view.params_snapshot {
1577        Value::Object(map) => map.clone(),
1578        _ => Map::new(),
1579    }
1580}