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