use std::collections::{BTreeMap, BTreeSet};
use pointlock_ir::{
ActionStepIR, AlignmentClass, AlignmentEntry, AlignmentReport, BindingState, CheckpointView,
FlowIR, Hash, PathFrame, ReconcileResult, RequiresConfirmation, RunLogPayload, RunPath,
SupervisePolicy, ir_hash,
};
use pointlock_provider_kit::{CancellationToken, ProviderSession, SessionOutcome};
use pointlock_store::{NewRun, Store, WriterLease};
use serde_json::{Map, Value};
use crate::align::{Alignment, Harvest, align, harvest, live_frame_pins, open_instances};
use crate::engine::{
Adopted, Execution, FrameState, FrontierWork, HumanRequestFact, RunOutcome, gated_mutating,
instance_key, is_history, now_ms, params_with_defaults, replay_permitted, root_path,
};
use crate::error::{BlockedReason, RunnerError};
use crate::load::{LoadedFlow, check_attestation, load};
use crate::scope::ScopeSeed;
pub struct RunOptions {
pub stop: CancellationToken,
pub run_id: Option<String>,
pub device_id: String,
pub platform: Option<String>,
pub vision: Option<std::sync::Arc<dyn pointlock_vision::VisionVerifier>>,
pub subflows: BTreeMap<Hash, FlowIR>,
pub supervise: Option<SupervisePolicy>,
pub clock: Option<std::sync::Arc<dyn Fn() -> u64 + Send + Sync>>,
pub stop_at: Option<String>,
pub stop_after: Option<String>,
}
impl RunOptions {
pub fn new(device_id: impl Into<String>) -> Self {
RunOptions {
stop: CancellationToken::new(),
run_id: None,
device_id: device_id.into(),
platform: None,
vision: None,
subflows: BTreeMap::new(),
supervise: None,
clock: None,
stop_at: None,
stop_after: None,
}
}
}
#[derive(Default)]
pub struct ResumeOptions {
pub stop: CancellationToken,
pub stop_at: Option<String>,
pub stop_after: Option<String>,
pub platform: Option<String>,
pub old_flow_ir: Option<FlowIR>,
pub supervise: Option<SupervisePolicy>,
pub vision: Option<std::sync::Arc<dyn pointlock_vision::VisionVerifier>>,
pub force_reexecute: Vec<String>,
pub force_stale_writer: bool,
pub allow_mutating_reexec: Vec<String>,
pub clock: Option<std::sync::Arc<dyn Fn() -> u64 + Send + Sync>>,
}
pub struct Runner;
impl Runner {
pub async fn run(
flow: &FlowIR,
params: Value,
session: Box<dyn ProviderSession>,
store: &mut Store,
opts: RunOptions,
) -> Result<RunOutcome, RunnerError> {
let RunOptions {
stop,
run_id,
device_id,
platform,
vision,
subflows,
supervise,
clock,
stop_at,
stop_after,
} = opts;
let loaded = load(flow, &subflows)?;
check_attestation(&loaded, session.attestation())?;
let params = params_with_defaults(flow, params)?;
let cursor = session.current_cursor().await?;
let initial_lineage = vec![cursor.session_id.clone()];
let run_id = run_id.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let _lease = WriterLease::acquire(store.root(), &run_id)?;
let run_id = store.begin_run(NewRun {
run_id: Some(run_id),
flow_id: flow.flow_id.clone(),
ir_hash: flow.ir_hash.clone(),
lockfile_digest: flow.lockfile_digest.clone(),
params_snapshot: Value::Object(params.clone()),
binding: BindingState {
device_id: device_id.clone(),
session_lineage: vec![cursor.session_id.clone()],
event_cursor: cursor,
},
created_at_ms: now_ms(),
})?;
store.append_event(
&run_id,
now_ms(),
&root_path(flow),
&RunLogPayload::RunStarted {
ir_hash: flow.ir_hash.clone(),
lockfile_digest: flow.lockfile_digest.clone(),
params_snapshot: Value::Object(params.clone()),
supervise_policy: supervise,
},
)?;
let env = env_bindings(&device_id, platform.as_deref(), &run_id);
let exec = Execution {
flows: &loaded,
session,
store,
run_id,
stop,
stop_at,
stop_after,
env,
session_lineage: initial_lineage,
pending_summaries: Default::default(),
attempt_base: Default::default(),
open_spans: Default::default(),
live_frames: Default::default(),
resumed: false,
authorized: BTreeSet::new(),
reentry_seen: false,
adoptable: Default::default(),
frontier: None,
vision,
supervise,
human: Default::default(),
settled: Default::default(),
recorded_verdicts: Default::default(),
hook_triggers: Default::default(),
clock,
};
let root = FrameState::new(flow, root_path(flow), params, 1);
exec.run(root, 0).await
}
pub async fn resume(
new_flow: &FlowIR,
run_id: &str,
session: Box<dyn ProviderSession>,
store: &mut Store,
opts: ResumeOptions,
) -> Result<RunOutcome, RunnerError> {
let subflows = BTreeMap::new();
Self::resume_with_subflows(new_flow, &subflows, run_id, session, store, opts).await
}
pub async fn resume_with_subflows(
new_flow: &FlowIR,
subflows: &BTreeMap<Hash, FlowIR>,
run_id: &str,
session: Box<dyn ProviderSession>,
store: &mut Store,
opts: ResumeOptions,
) -> Result<RunOutcome, RunnerError> {
let loaded = load(new_flow, subflows)?;
check_attestation(&loaded, session.attestation())?;
let view = store.rebuild_checkpoint(run_id)?;
let events = store.events(run_id)?;
let facts = harvest(&events);
let executing = executing_ir_hash(&facts, &view);
if let Some(old) = opts.old_flow_ir.as_ref() {
let computed = ir_hash(old);
if computed != *executing {
return Err(RunnerError::OldIrMismatch {
expected: executing.clone(),
computed,
});
}
}
let _lease = if opts.force_stale_writer {
None
} else {
Some(WriterLease::acquire(store.root(), run_id)?)
};
if is_same_ir_resume(&facts, &view, new_flow) {
resume_same_ir(loaded, view, facts, run_id, session, store, opts).await
} else {
resume_cross_ir(loaded, view, facts, run_id, session, store, opts).await
}
}
#[allow(clippy::too_many_arguments)]
pub async fn align_preview(
new_flow: &FlowIR,
subflows: &BTreeMap<Hash, FlowIR>,
run_id: &str,
store: &Store,
platform: Option<&str>,
vision: Option<&dyn pointlock_vision::VisionVerifier>,
forced: &[String],
old_flow_ir: Option<&FlowIR>,
) -> Result<AlignmentReport, RunnerError> {
let loaded = load(new_flow, subflows)?;
let view = store.rebuild_checkpoint(run_id)?;
let events = store.events(run_id)?;
let facts = harvest(&events);
let executing = executing_ir_hash(&facts, &view);
if let Some(old) = old_flow_ir {
let computed = ir_hash(old);
if computed != *executing {
return Err(RunnerError::OldIrMismatch {
expected: executing.clone(),
computed,
});
}
}
if let Some(live_hook) = facts
.live_frames
.iter()
.find(|path| path.iter().any(|f| matches!(f, PathFrame::Hook { .. })))
{
return Err(RunnerError::M0Unsupported {
detail: format!(
"resume across a live handler-repair frame ({}) is not in the M2 subset โ \
the repair flow suspended mid-flight; hook-aware frame re-entry is \
registered for the repair wave",
pointlock_ir::render_run_path(live_hook)
),
});
}
if is_same_ir_resume(&facts, &view, new_flow) {
return Ok(same_ir_report(new_flow, &facts, &view));
}
if !is_alignable_path(&view.frontier.run_path) {
return Err(RunnerError::M0Unsupported {
detail: "the run's frontier sits inside a handler frame".to_owned(),
});
}
if view
.human_pending
.as_ref()
.is_some_and(|pending| !is_alignable_path(&pending.run_path))
{
return Err(RunnerError::M0Unsupported {
detail: "a handler escalation is still awaiting an answer".to_owned(),
});
}
let seed = ScopeSeed::new(
params_object(&view),
&view.binding.device_id,
platform,
run_id,
);
match align(
&loaded,
&new_flow.body,
&view,
&facts,
&seed,
store,
vision,
&[],
forced,
old_flow_ir,
)
.await
{
Ok(alignment) => Ok(alignment.report),
Err(RunnerError::RequiresConfirmation { report }) => Ok(*report),
Err(other) => Err(other),
}
}
}
fn env_bindings(device_id: &str, platform: Option<&str>, run_id: &str) -> Vec<(String, Value)> {
let mut env = vec![
("deviceId".to_owned(), Value::String(device_id.to_owned())),
("runId".to_owned(), Value::String(run_id.to_owned())),
];
if let Some(platform) = platform {
env.push(("platform".to_owned(), Value::String(platform.to_owned())));
}
env
}
fn executing_ir_hash<'a>(facts: &'a Harvest, view: &'a CheckpointView) -> &'a Hash {
facts.executing_ir_hash.as_ref().unwrap_or(&view.ir_hash)
}
fn is_same_ir_resume(facts: &Harvest, view: &CheckpointView, new_flow: &FlowIR) -> bool {
!facts.cross_ir_resumed && view.ir_hash == new_flow.ir_hash
}
fn same_ir_report(new_flow: &FlowIR, facts: &Harvest, view: &CheckpointView) -> AlignmentReport {
let open = open_instances(facts);
let completed: BTreeMap<String, &pointlock_ir::StepRecord> = view
.completed
.iter()
.filter(|record| !open.contains(&instance_key(&record.run_path)))
.map(|record| (instance_key(&record.run_path), record))
.collect();
let mut entries = Vec::new();
for step in &new_flow.body {
let mut path = root_path(new_flow);
path.push(match step {
pointlock_ir::StepIR::Call(call) => PathFrame::Call {
step_id: Some(call.base.step_id.clone()),
callee_flow_id: call.flow_ref.flow_id.clone(),
callee_ir_hash: call.flow_ref.ir_hash.clone(),
},
other => PathFrame::Step {
step_id: other.step_id().clone(),
},
});
let key = instance_key(&path);
let adopted = completed.get(&key).is_some_and(|record| is_history(record));
entries.push(AlignmentEntry {
run_path: path.clone(),
step_id: step.step_id().clone(),
class: if adopted {
AlignmentClass::Reusable
} else {
AlignmentClass::New
},
reason: (!adopted).then(|| "no adoptable prior record".to_owned()),
});
}
AlignmentReport {
entries,
resume_point: Some(view.frontier.run_path.clone()),
requires_confirmation: Vec::new(),
}
}
async fn resume_same_ir(
loaded: LoadedFlow<'_>,
view: CheckpointView,
facts: Harvest,
run_id: &str,
mut session: Box<dyn ProviderSession>,
store: &mut Store,
opts: ResumeOptions,
) -> Result<RunOutcome, RunnerError> {
let new_flow = loaded.root;
let bind_cursor = store.run_meta(run_id)?.binding.event_cursor;
let open = open_instances(&facts);
let mut adoptable: BTreeMap<String, Adopted> = BTreeMap::new();
for record in &view.completed {
let key = instance_key(&record.run_path);
if open.contains(&key) {
continue;
}
adoptable.insert(
key.clone(),
Adopted {
record: record.clone(),
before_id: facts.before_observation.get(&key).cloned(),
after_id: facts.after_observation.get(&key).cloned(),
},
);
}
let open_spans: BTreeMap<String, Value> = facts
.open_spans
.iter()
.map(|path| {
let key = instance_key(path);
let inputs = facts
.entered_inputs
.get(&key)
.cloned()
.unwrap_or(Value::Null);
(key, inputs)
})
.collect();
if let Some(live_hook) = facts
.live_frames
.iter()
.find(|path| path.iter().any(|f| matches!(f, PathFrame::Hook { .. })))
{
return Err(RunnerError::M0Unsupported {
detail: format!(
"resume across a live handler-repair frame ({}) is not in the M2 subset โ \
the repair flow suspended mid-flight; hook-aware frame re-entry is \
registered for the repair wave",
pointlock_ir::render_run_path(live_hook)
),
});
}
let live_frames = live_frame_pins(&facts);
let mut report = same_ir_report(new_flow, &facts, &view);
let mut frontier_work = None;
let mut deferred_settle = None;
let mut pending_adjudication: Option<Box<Adjudication>> = None;
let mut blocked = None;
if let Some(intent) = &view.frontier.pending_intent {
let frontier_key = instance_key(&view.frontier.run_path);
let new_step = loaded.resolve_action(&view.frontier.run_path);
let effect_dirty = match (new_step, facts.entered_effect_hash.get(&frontier_key)) {
(Some(step), Some(archived)) => *archived != step.base.effect_hash,
_ => true,
};
let decision = match reconcile_frontier(
&mut session,
new_step,
true,
effect_dirty,
&view,
&facts,
&bind_cursor,
&mut report,
intent,
)
.await
{
Ok(decision) => decision,
Err(error) => {
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Err(error);
}
};
match decision {
FrontierDecision::Work(work) => frontier_work = Some((frontier_key, work)),
FrontierDecision::DeferredSettle(settle) => deferred_settle = Some(settle),
FrontierDecision::Adjudicate(adjudication) => pending_adjudication = Some(adjudication),
FrontierDecision::Blocked(reason) => blocked = Some(reason),
FrontierDecision::Nothing => {}
}
}
let resumed_cursor = session.current_cursor().await.ok();
store.append_event(
run_id,
now_ms(),
&root_path(new_flow),
&RunLogPayload::RunResumed {
alignment_report: report.clone(),
supervise_policy: opts.supervise,
event_cursor: resumed_cursor,
},
)?;
if let Some((path, call_id, outcome)) = deferred_settle {
store.append_event(
run_id,
now_ms(),
&path,
&RunLogPayload::ActionSettled {
call_id,
outcome: crate::engine::quarantine_unpersistable(*outcome),
},
)?;
}
if let Some(adjudication) = pending_adjudication {
let Adjudication {
run_path,
request,
pending,
} = *adjudication;
if let Some((request_id, prompt, presents)) = request {
store.append_event(
run_id,
now_ms(),
&run_path,
&RunLogPayload::HumanRequested {
request_id,
purpose: pointlock_ir::HumanPurpose::Step,
mode: Some(pointlock_ir::vocab::HumanMode::RepairWorld),
prompt,
presents,
decisions: Some(vec![
"adopt".to_owned(),
"redo".to_owned(),
"abort".to_owned(),
]),
output_schema: None,
deadline_at_ms: None,
},
)?;
}
let summary = crate::engine::capture_provider_state_summary(
session.as_ref(),
&view.binding.session_lineage,
&view.binding.device_id,
opts.platform.as_deref(),
)
.await;
store.append_event(
run_id,
now_ms(),
&root_path(new_flow),
&RunLogPayload::RunSuspended {
provider_state_summary: Some(summary),
reason: Some(format!(
"awaiting human adjudication (requestId {})",
pending.request_id
)),
},
)?;
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Ok(RunOutcome::AwaitingHuman { pending });
}
if let Some(reason) = blocked {
let summary = crate::engine::capture_provider_state_summary(
session.as_ref(),
&view.binding.session_lineage,
&view.binding.device_id,
opts.platform.as_deref(),
)
.await;
store.append_event(
run_id,
now_ms(),
&root_path(new_flow),
&RunLogPayload::RunSuspended {
provider_state_summary: Some(summary),
reason: Some(reason.to_string()),
},
)?;
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Ok(RunOutcome::Blocked { reason });
}
let params = params_object(&view);
let env = env_bindings(&view.binding.device_id, opts.platform.as_deref(), run_id);
let exec = Execution {
flows: &loaded,
session,
store,
run_id: run_id.to_owned(),
stop: opts.stop,
stop_at: opts.stop_at.clone(),
stop_after: opts.stop_after.clone(),
env,
attempt_base: facts.max_attempt.clone(),
open_spans,
live_frames,
resumed: true,
authorized: opts.allow_mutating_reexec.iter().cloned().collect(),
reentry_seen: false,
adoptable,
frontier: frontier_work,
session_lineage: view.binding.session_lineage.clone(),
pending_summaries: BTreeMap::new(),
vision: opts.vision.clone(),
supervise: opts.supervise,
human: facts.human_requests.clone(),
settled: facts.settled.clone(),
recorded_verdicts: facts.recorded_verdicts.clone(),
hook_triggers: facts.hook_triggers.clone(),
clock: opts.clock,
};
let root = FrameState::new(new_flow, root_path(new_flow), params, 1);
exec.run(root, 0).await
}
async fn resume_cross_ir(
loaded: LoadedFlow<'_>,
view: CheckpointView,
facts: Harvest,
run_id: &str,
mut session: Box<dyn ProviderSession>,
store: &mut Store,
opts: ResumeOptions,
) -> Result<RunOutcome, RunnerError> {
let new_flow = loaded.root;
let bind_cursor = store.run_meta(run_id)?.binding.event_cursor;
if !is_alignable_path(&view.frontier.run_path) {
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Err(RunnerError::M0Unsupported {
detail: format!(
"the run's frontier sits inside a handler frame ({}); hook-aware frame \
re-entry is registered for the repair wave",
pointlock_ir::render_run_path(&view.frontier.run_path)
),
});
}
if let Some(live_hook) = facts
.live_frames
.iter()
.find(|path| path.iter().any(|f| matches!(f, PathFrame::Hook { .. })))
{
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Err(RunnerError::M0Unsupported {
detail: format!(
"resume across a live handler-repair frame ({}) is not in the M2 subset โ \
the repair flow suspended mid-flight; hook-aware frame re-entry is \
registered for the repair wave",
pointlock_ir::render_run_path(live_hook)
),
});
}
if let Some(pending) = view
.human_pending
.as_ref()
.filter(|pending| !is_alignable_path(&pending.run_path))
{
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Err(RunnerError::M0Unsupported {
detail: format!(
"a handler escalation is still awaiting an answer ({}); resuming it under a \
repaired IR needs hook-aware re-entry, which is registered for the repair \
wave โ answer or let it time out first",
pointlock_ir::render_run_path(&pending.run_path)
),
});
}
let seed = ScopeSeed::new(
params_object(&view),
&view.binding.device_id,
opts.platform.as_deref(),
run_id,
);
let mut alignment = match align(
&loaded,
&new_flow.body,
&view,
&facts,
&seed,
store,
opts.vision.as_deref(),
&opts.allow_mutating_reexec,
&opts.force_reexecute,
opts.old_flow_ir.as_ref(),
)
.await
{
Ok(alignment) => alignment,
Err(error) => {
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Err(error);
}
};
let mut frontier_work = None;
let mut deferred_settle = None;
let mut pending_adjudication: Option<Box<Adjudication>> = None;
let mut blocked = None;
if let Some(intent) = &view.frontier.pending_intent {
let frontier_key = instance_key(&view.frontier.run_path);
let new_step = loaded.resolve_action(&view.frontier.run_path);
let at_resume = alignment.resume_key.as_deref() == Some(frontier_key.as_str());
let effect_dirty = match (new_step, facts.entered_effect_hash.get(&frontier_key)) {
(Some(step), Some(archived)) => *archived != step.base.effect_hash,
_ => true,
};
let decision = match reconcile_frontier(
&mut session,
new_step,
at_resume,
effect_dirty,
&view,
&facts,
&bind_cursor,
&mut alignment.report,
intent,
)
.await
{
Ok(decision) => decision,
Err(error) => {
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Err(error);
}
};
match decision {
FrontierDecision::Work(work) => frontier_work = Some((frontier_key, work)),
FrontierDecision::DeferredSettle(settle) => deferred_settle = Some(settle),
FrontierDecision::Adjudicate(adjudication) => pending_adjudication = Some(adjudication),
FrontierDecision::Blocked(reason) => blocked = Some(reason),
FrontierDecision::Nothing => {}
}
}
let resumed_cursor = session.current_cursor().await.ok();
store.append_event(
run_id,
now_ms(),
&root_path(new_flow),
&RunLogPayload::RunResumed {
alignment_report: alignment.report.clone(),
supervise_policy: opts.supervise,
event_cursor: resumed_cursor,
},
)?;
let rejudged = std::mem::take(&mut alignment.rejudged);
for rejudge in rejudged {
let remote_archival_error = session
.record_verdict(pointlock_provider_kit::VerdictWrite {
status: rejudge.verdict.status,
summary: crate::engine::cap_wire_summary(&rejudge.verdict),
evidence: rejudge
.verdict
.evidence
.iter()
.take(pointlock_provider_kit::VERDICT_EVIDENCE_MAX_ENTRIES)
.cloned()
.collect(),
})
.await
.err()
.map(|error| format!("remote archival failed: {error}"));
store.append_event(
run_id,
now_ms(),
&rejudge.run_path,
&RunLogPayload::VerdictRecorded {
verdict: rejudge.verdict.clone(),
localized: Vec::new(),
localization_gaps: Vec::new(),
remote_archival_error,
},
)?;
}
if let Some((path, call_id, outcome)) = deferred_settle {
store.append_event(
run_id,
now_ms(),
&path,
&RunLogPayload::ActionSettled {
call_id,
outcome: crate::engine::quarantine_unpersistable(*outcome),
},
)?;
}
if let Some(adjudication) = pending_adjudication {
let Adjudication {
run_path,
request,
pending,
} = *adjudication;
if let Some((request_id, prompt, presents)) = request {
store.append_event(
run_id,
now_ms(),
&run_path,
&RunLogPayload::HumanRequested {
request_id,
purpose: pointlock_ir::HumanPurpose::Step,
mode: Some(pointlock_ir::vocab::HumanMode::RepairWorld),
prompt,
presents,
decisions: Some(vec![
"adopt".to_owned(),
"redo".to_owned(),
"abort".to_owned(),
]),
output_schema: None,
deadline_at_ms: None,
},
)?;
}
let summary = crate::engine::capture_provider_state_summary(
session.as_ref(),
&view.binding.session_lineage,
&view.binding.device_id,
opts.platform.as_deref(),
)
.await;
store.append_event(
run_id,
now_ms(),
&root_path(new_flow),
&RunLogPayload::RunSuspended {
provider_state_summary: Some(summary),
reason: Some(format!(
"awaiting human adjudication (requestId {})",
pending.request_id
)),
},
)?;
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Ok(RunOutcome::AwaitingHuman { pending });
}
if let Some(reason) = blocked {
let summary = crate::engine::capture_provider_state_summary(
session.as_ref(),
&view.binding.session_lineage,
&view.binding.device_id,
opts.platform.as_deref(),
)
.await;
store.append_event(
run_id,
now_ms(),
&root_path(new_flow),
&RunLogPayload::RunSuspended {
provider_state_summary: Some(summary),
reason: Some(reason.to_string()),
},
)?;
let _ = session.end(SessionOutcome::Shutdown, None).await;
return Ok(RunOutcome::Blocked { reason });
}
let Alignment {
adoptable,
teardown,
..
} = alignment;
let torn = |key: &str| -> bool {
teardown
.as_deref()
.is_some_and(|call| key == call || crate::align::is_instance_descendant(call, key))
};
if teardown.is_some() {
let live_keys: BTreeSet<String> = facts
.live_frames
.iter()
.map(|path| instance_key(path))
.collect();
for span in facts.open_spans.iter().rev() {
let key = instance_key(span);
if !torn(&key) {
continue;
}
if live_keys.contains(&key) {
store.append_event(
run_id,
now_ms(),
span,
&RunLogPayload::CallFramePopped { outputs: None },
)?;
}
store.append_event(
run_id,
now_ms(),
span,
&RunLogPayload::StepExited {
provider_state_summary: None,
state: pointlock_ir::StepState::Aborted,
output: None,
localized: Vec::new(),
localization_gaps: Vec::new(),
},
)?;
}
}
let open_spans: BTreeMap<String, Value> = facts
.open_spans
.iter()
.filter(|path| !torn(&instance_key(path)))
.map(|path| {
let key = instance_key(path);
let inputs = facts
.entered_inputs
.get(&key)
.cloned()
.unwrap_or(Value::Null);
(key, inputs)
})
.collect();
let live_frames: BTreeMap<String, pointlock_ir::Hash> = live_frame_pins(&facts)
.into_iter()
.filter(|(key, _)| !torn(key))
.collect();
let params = params_object(&view);
let env = env_bindings(&view.binding.device_id, opts.platform.as_deref(), run_id);
let exec = Execution {
flows: &loaded,
session,
store,
run_id: run_id.to_owned(),
stop: opts.stop,
stop_at: opts.stop_at.clone(),
stop_after: opts.stop_after.clone(),
env,
attempt_base: facts.max_attempt.clone(),
open_spans,
live_frames,
resumed: true,
authorized: opts.allow_mutating_reexec.iter().cloned().collect(),
reentry_seen: false,
adoptable,
frontier: frontier_work,
session_lineage: view.binding.session_lineage.clone(),
pending_summaries: BTreeMap::new(),
vision: opts.vision.clone(),
supervise: opts.supervise,
human: facts.human_requests.clone(),
settled: facts.settled.clone(),
recorded_verdicts: facts.recorded_verdicts.clone(),
hook_triggers: facts.hook_triggers.clone(),
clock: opts.clock,
};
let root = FrameState::new(new_flow, root_path(new_flow), params, 1);
exec.run(root, 0).await
}
fn is_alignable_path(path: &RunPath) -> bool {
!path
.iter()
.any(|frame| matches!(frame, PathFrame::Hook { .. }))
}
struct Adjudication {
run_path: RunPath,
request: Option<(String, String, Value)>,
pending: pointlock_ir::HumanPending,
}
enum FrontierDecision {
Work(FrontierWork),
DeferredSettle(
(
pointlock_ir::RunPath,
String,
Box<pointlock_ir::ActionOutcome>,
),
),
Adjudicate(Box<Adjudication>),
Blocked(BlockedReason),
Nothing,
}
#[allow(clippy::too_many_arguments)]
async fn reconcile_frontier(
session: &mut Box<dyn ProviderSession>,
new_step: Option<&ActionStepIR>,
at_resume: bool,
effect_dirty: bool,
view: &CheckpointView,
facts: &Harvest,
bind_cursor: &pointlock_ir::EventCursor,
report: &mut AlignmentReport,
intent: &pointlock_ir::PendingIntent,
) -> Result<FrontierDecision, RunnerError> {
let issuing = facts
.intent_issuing
.get(&intent.call_id)
.cloned()
.unwrap_or(crate::align::IssuingCursor::Unknown);
let fate = match &issuing {
crate::align::IssuingCursor::FromBinding => {
session.reconcile(&intent.call_id, bind_cursor).await?
}
crate::align::IssuingCursor::Known(cursor) => {
session.reconcile(&intent.call_id, cursor).await?
}
crate::align::IssuingCursor::Unknown => ReconcileResult::LogUnavailable {
reason: "the issuing session is unknowable (a resume predating the \
eventCursor carrier intervened); refusing to reconcile with \
a fabricated credential"
.to_owned(),
},
};
let intent_path = facts
.intent_path
.get(&intent.call_id)
.cloned()
.unwrap_or_else(|| view.frontier.run_path.clone());
let mutating_gated = new_step.map(gated_mutating).unwrap_or(true);
match fate {
ReconcileResult::Completed { outcome } => {
if !effect_dirty && at_resume {
return Ok(FrontierDecision::Work(FrontierWork::Adopt {
call_id: intent.call_id.clone(),
intent_path,
outcome,
args: intent.args_snapshot.clone(),
chain_index: facts.intent_chain_index.get(&intent.call_id).copied(),
}));
}
let effect_possible = matches!(
outcome.as_ref(),
pointlock_ir::ActionOutcome::Succeeded { .. }
| pointlock_ir::ActionOutcome::TimedOut { .. }
);
if effect_possible && mutating_gated {
report.requires_confirmation.push(RequiresConfirmation {
run_path: view.frontier.run_path.clone(),
step_id: new_step.map(|step| step.base.step_id.clone()),
cause: "frontierUnknown".to_owned(),
reason: format!(
"callId {} reached a recorded {} terminal on the device but \
it is not adoptable; re-execution of the mutating step \
needs explicit authorization",
intent.call_id,
outcome.kind()
),
});
return Err(RunnerError::RequiresConfirmation {
report: Box::new(report.clone()),
});
}
Ok(FrontierDecision::DeferredSettle((
intent_path,
intent.call_id.clone(),
outcome,
)))
}
ReconcileResult::NeverDispatched => {
if !effect_dirty && at_resume {
Ok(FrontierDecision::Work(FrontierWork::Replay {
chain_index: facts.intent_chain_index.get(&intent.call_id).copied(),
args: intent.args_snapshot.clone(),
}))
} else {
Ok(FrontierDecision::Nothing)
}
}
ReconcileResult::StartedNoTerminal => uncertain_branch(
new_step,
intent,
&view.frontier.run_path,
facts,
report,
at_resume,
effect_dirty,
"startedNoTerminal",
facts.intent_chain_index.get(&intent.call_id).copied(),
),
ReconcileResult::LogUnavailable { reason } => uncertain_branch(
new_step,
intent,
&view.frontier.run_path,
facts,
report,
at_resume,
effect_dirty,
&format!("logUnavailable: {reason}"),
facts.intent_chain_index.get(&intent.call_id).copied(),
),
}
}
#[allow(clippy::too_many_arguments)]
fn uncertain_branch(
new_step: Option<&ActionStepIR>,
intent: &pointlock_ir::PendingIntent,
frontier_path: &RunPath,
facts: &Harvest,
report: &mut AlignmentReport,
at_resume: bool,
effect_dirty: bool,
fate: &str,
chain_index: Option<u32>,
) -> Result<FrontierDecision, RunnerError> {
let permitted = new_step.map(replay_permitted).unwrap_or(false);
if permitted {
if at_resume && !effect_dirty {
return Ok(FrontierDecision::Work(FrontierWork::Replay {
chain_index,
args: intent.args_snapshot.clone(),
}));
}
return Ok(FrontierDecision::Nothing);
}
let mut hook_path = frontier_path.clone();
hook_path.push(PathFrame::Hook {
hook: pointlock_ir::HandlerHook::OnResumeDrift,
trigger: 1,
});
hook_path.push(PathFrame::Step {
step_id: "adjudicate".try_into().expect("a fixed valid step id"),
});
let key = instance_key(&hook_path);
if let Some(fact) = facts.human_requests.get(&key)
&& fact.presents.get("callId").and_then(Value::as_str) == Some(intent.call_id.as_str())
{
match &fact.final_response {
None => {
return Ok(FrontierDecision::Adjudicate(Box::new(Adjudication {
run_path: hook_path.clone(),
request: None,
pending: pending_of(fact, &hook_path),
})));
}
Some(response) => {
let ruling = response.get("decision").and_then(Value::as_str);
match ruling {
Some("adopt") => {
if at_resume && !effect_dirty {
return Ok(FrontierDecision::Work(FrontierWork::ConfirmEffect {
message: format!(
"uncertain fate ({fate}) of callId {} adjudicated \
`adopt`",
intent.call_id
),
args: intent.args_snapshot.clone(),
}));
}
report.requires_confirmation.push(RequiresConfirmation {
run_path: frontier_path.clone(),
step_id: new_step.map(|step| step.base.step_id.clone()),
cause: "frontierUnknown".to_owned(),
reason: format!(
"callId {} was adjudicated `adopt` (the effect stands) but \
the step is not adoptable here; re-execution of the \
mutating step needs explicit authorization",
intent.call_id
),
});
return Err(RunnerError::RequiresConfirmation {
report: Box::new(report.clone()),
});
}
Some("redo") => {
if at_resume && !effect_dirty {
return Ok(FrontierDecision::Work(FrontierWork::Replay {
chain_index,
args: intent.args_snapshot.clone(),
}));
}
return Ok(FrontierDecision::Nothing);
}
Some("abort") => {
return Ok(FrontierDecision::Work(FrontierWork::AbortRuled {
args: intent.args_snapshot.clone(),
}));
}
other => {
return Ok(FrontierDecision::Blocked(BlockedReason::RequiresHuman {
call_id: intent.call_id.clone(),
detail: format!(
"adjudication response carries an unusable decision \
{other:?}; refusing to guess"
),
}));
}
}
}
}
}
let request_id = uuid::Uuid::new_v4().to_string();
let prompt = format!(
"the fate of callId {} is uncertain ({fate}) and the step is mutating and \
not idempotent โ automatic replay is forbidden (I2). Inspect the device, \
then rule: `adopt` (the effect happened; verify and continue), `redo` \
(the effect did not happen or you undid it; dispatch again), or `abort` \
(stop the run)",
intent.call_id
);
let presents = serde_json::json!({
"callId": intent.call_id,
"fate": fate,
"argsSnapshot": intent.args_snapshot,
});
let pending = pointlock_ir::HumanPending {
run_path: hook_path.clone(),
request_id: request_id.clone(),
purpose: pointlock_ir::HumanPurpose::Step,
mode: Some(pointlock_ir::vocab::HumanMode::RepairWorld),
prompt: prompt.clone(),
deadline_at_ms: None,
};
Ok(FrontierDecision::Adjudicate(Box::new(Adjudication {
run_path: hook_path,
request: Some((request_id, prompt, presents)),
pending,
})))
}
fn pending_of(fact: &HumanRequestFact, hook_path: &RunPath) -> pointlock_ir::HumanPending {
pointlock_ir::HumanPending {
run_path: hook_path.clone(),
request_id: fact.request_id.clone(),
purpose: fact.purpose,
mode: fact.mode,
prompt: fact.prompt.clone(),
deadline_at_ms: fact.deadline_at_ms,
}
}
fn params_object(view: &CheckpointView) -> Map<String, Value> {
match &view.params_snapshot {
Value::Object(map) => map.clone(),
_ => Map::new(),
}
}