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}