use std::collections::{BTreeMap, BTreeSet};
use pointlock_ir::{
ActionOutcomeKind, ActionStepIR, AlignmentClass, AlignmentEntry, AlignmentReport, CallStepIR,
CheckpointView, FlowIR, HandlerHook, Hash, PathFrame, PredicateIR, RequiresConfirmation,
RunLogEvent, RunLogPayload, RunPath, StepIR, StepRecord, Verdict, VerdictPolicy, VerdictStatus,
effect_hash,
};
use pointlock_store::Store;
use serde_json::Value;
use crate::engine::{
Adopted, HumanRequestFact, attempt_of, child_frame, gated_mutating, instance_key, is_history,
root_path,
};
use crate::error::RunnerError;
use crate::judge::{eval_expr_assertion, fold_step_verdict, project_output};
use crate::observe_eval::{
EvaluatedAssertion, ObserveMaterial, eval_observed_assertion, material_from_observation,
};
use crate::scope::ScopeSeed;
pub(crate) fn live_frame_pins(facts: &Harvest) -> BTreeMap<String, Hash> {
facts
.live_frames
.iter()
.filter_map(|path| match path.last() {
Some(PathFrame::Call { callee_ir_hash, .. }) => {
Some((instance_key(path), callee_ir_hash.clone()))
}
_ => None,
})
.collect()
}
fn call_head_unchanged(call: &CallStepIR, archived_pin: &Hash, archived_effect: &Hash) -> bool {
let mut probe = call.clone();
probe.flow_ref.ir_hash = archived_pin.clone();
effect_hash(&StepIR::Call(probe)) == *archived_effect
}
fn drillable<'a>(
call: &CallStepIR,
key: &str,
live_pins: &BTreeMap<String, Hash>,
facts: &Harvest,
loaded: &crate::load::LoadedFlow<'a>,
) -> Option<&'a FlowIR> {
let archived_pin = live_pins.get(key)?;
let archived_effect = facts.entered_effect_hash.get(key)?;
if !call_head_unchanged(call, archived_pin, archived_effect) {
return None;
}
loaded.try_callee(&call.flow_ref)
}
fn old_step_at<'a>(old_root: &'a FlowIR, path: &RunPath) -> Option<&'a StepIR> {
let mut bodies: Vec<&'a [StepIR]> = vec![&old_root.body];
let mut current: Option<&'a StepIR> = None;
for frame in path {
match frame {
PathFrame::Flow { .. } | PathFrame::Iteration { .. } => {}
PathFrame::Attempt { .. } | PathFrame::Phase { .. } | PathFrame::Assertion { .. } => {
break;
}
PathFrame::Hook { .. } | PathFrame::Call { .. } => return None,
PathFrame::Step { step_id } => {
let step = bodies
.iter()
.flat_map(|body| body.iter())
.find(|step| step.step_id() == step_id)?;
bodies = match step {
StepIR::If(branch) => {
let mut inner: Vec<&'a [StepIR]> = vec![&branch.then];
if let Some(otherwise) = &branch.r#else {
inner.push(otherwise);
}
inner
}
StepIR::Foreach(each) => vec![&each.body],
_ => Vec::new(),
};
current = Some(step);
}
}
}
current
}
fn preflight_only_change(old_root: Option<&FlowIR>, record: &StepRecord, step: &StepIR) -> bool {
let Some(old_step) = old_root.and_then(|old| old_step_at(old, &record.run_path)) else {
return false;
};
old_step.kind() == step.kind()
&& pointlock_ir::judge_hash(old_step) == record.judge_hash
&& pointlock_ir::judge_subdomain_sans_preflight(old_step)
== pointlock_ir::judge_subdomain_sans_preflight(step)
}
fn call_pin(step: &StepIR) -> Hash {
match step {
StepIR::Call(call) => call.flow_ref.ir_hash.clone(),
other => unreachable!("only a call node is ever drilled, got {}", other.kind()),
}
}
pub(crate) fn is_instance_descendant(outer: &str, inner: &str) -> bool {
inner.len() > outer.len()
&& inner.starts_with(outer)
&& matches!(inner.as_bytes()[outer.len()], b'/' | b'[')
}
pub(crate) struct Node<'a> {
pub path: RunPath,
pub step: &'a StepIR,
}
#[allow(clippy::too_many_arguments)]
fn flatten<'a>(
loaded: &crate::load::LoadedFlow<'a>,
facts: &Harvest,
live_pins: &BTreeMap<String, Hash>,
prefix: &RunPath,
body: &'a [StepIR],
completed: &BTreeMap<String, &StepRecord>,
descended: &mut BTreeSet<String>,
out: &mut Vec<Node<'a>>,
) {
for step in body {
let mut path = prefix.clone();
path.push(child_frame(step));
let key = instance_key(&path);
let record = completed.get(&key).copied();
out.push(Node {
path: path.clone(),
step,
});
if let StepIR::Call(call) = step {
if let Some(callee) = drillable(call, &key, live_pins, facts, loaded) {
descended.insert(key);
flatten(
loaded,
facts,
live_pins,
&path,
&callee.body,
completed,
descended,
out,
);
}
continue;
}
let Some(record) = record else { continue };
match step {
StepIR::If(branch) => {
if record.effect_hash != step.base().effect_hash {
continue;
}
let Some(taken) = record.resolved_inputs.get("cond").and_then(Value::as_bool)
else {
continue;
};
let selected: &'a [StepIR] = if taken {
&branch.then
} else {
branch.r#else.as_deref().unwrap_or(&[])
};
descended.insert(key);
flatten(
loaded, facts, live_pins, &path, selected, completed, descended, out,
);
}
StepIR::Foreach(each) => {
if record.effect_hash != step.base().effect_hash {
continue;
}
let Some(rounds) = record
.resolved_inputs
.get("items")
.and_then(Value::as_array)
.map(Vec::len)
else {
continue;
};
descended.insert(key);
for index in 0..rounds {
let mut round = path.clone();
round.push(PathFrame::Iteration {
index: index as u64,
key: None,
});
flatten(
loaded, facts, live_pins, &round, &each.body, completed, descended, out,
);
}
}
_ => {}
}
}
}
fn gated_effect(step: &StepIR, descended: bool, loaded: &crate::load::LoadedFlow<'_>) -> bool {
fn body_mutates(body: &[StepIR], loaded: &crate::load::LoadedFlow<'_>, depth: usize) -> bool {
if depth > 32 {
return true; }
body.iter().any(|step| match step {
StepIR::Action(action) => gated_mutating(action),
StepIR::If(branch) => {
body_mutates(&branch.then, loaded, depth + 1)
|| branch
.r#else
.as_deref()
.is_some_and(|otherwise| body_mutates(otherwise, loaded, depth + 1))
}
StepIR::Foreach(each) => body_mutates(&each.body, loaded, depth + 1),
StepIR::Call(call) => match loaded.try_callee(&call.flow_ref) {
Some(callee) => body_mutates(&callee.body, loaded, depth + 1),
None => true,
},
_ => false,
})
}
match step {
StepIR::Action(action) => gated_mutating(action),
StepIR::If(_) | StepIR::Foreach(_) | StepIR::Call(_) if descended => false,
StepIR::If(branch) => {
body_mutates(&branch.then, loaded, 0)
|| branch
.r#else
.as_deref()
.is_some_and(|otherwise| body_mutates(otherwise, loaded, 0))
}
StepIR::Foreach(each) => body_mutates(&each.body, loaded, 0),
StepIR::Call(call) => match loaded.try_callee(&call.flow_ref) {
Some(callee) => body_mutates(&callee.body, loaded, 0),
None => true,
},
_ => false,
}
}
fn is_container(step: &StepIR) -> bool {
matches!(step, StepIR::If(_) | StepIR::Foreach(_) | StepIR::Call(_))
}
fn is_descendant(outer: &RunPath, inner: &RunPath) -> bool {
inner.len() > outer.len() && inner.starts_with(outer)
}
pub(crate) fn hook_trigger_key(instance: &str, hook: HandlerHook) -> String {
let wire = serde_json::to_value(hook).expect("HandlerHook serializes");
format!(
"{instance}|{}",
wire.as_str().expect("HandlerHook is a string literal")
)
}
pub(crate) fn retain_other_instances(counters: &mut BTreeMap<String, u64>, instance: &str) {
let prefix = format!("{instance}|");
counters.retain(|key, _| !key.starts_with(&prefix));
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum IssuingCursor {
FromBinding,
Known(pointlock_ir::EventCursor),
Unknown,
}
pub(crate) struct Harvest {
pub intent_issuing: BTreeMap<String, IssuingCursor>,
pub intent_chain_index: BTreeMap<String, u32>,
pub raw_output: BTreeMap<String, Value>,
pub after_observation: BTreeMap<String, String>,
pub before_observation: BTreeMap<String, String>,
pub max_attempt: BTreeMap<String, u64>,
pub verdict_seq: BTreeMap<String, u64>,
pub intent_path: BTreeMap<String, RunPath>,
pub entered_effect_hash: BTreeMap<String, Hash>,
pub entered_inputs: BTreeMap<String, Value>,
pub entered_seq: BTreeMap<String, u64>,
pub open_spans: Vec<RunPath>,
pub live_frames: Vec<RunPath>,
pub human_requests: BTreeMap<String, HumanRequestFact>,
pub settled: BTreeMap<String, SettledFact>,
pub recorded_verdicts: BTreeMap<String, (VerdictStatus, bool)>,
pub hook_triggers: BTreeMap<String, u64>,
pub executing_ir_hash: Option<Hash>,
pub cross_ir_resumed: bool,
}
#[derive(Debug, Clone)]
pub(crate) struct SettledFact {
pub outcome: pointlock_ir::ActionOutcome,
}
pub(crate) fn harvest(events: &[RunLogEvent]) -> Harvest {
let mut facts = Harvest {
intent_issuing: BTreeMap::new(),
intent_chain_index: BTreeMap::new(),
raw_output: BTreeMap::new(),
after_observation: BTreeMap::new(),
before_observation: BTreeMap::new(),
max_attempt: BTreeMap::new(),
verdict_seq: BTreeMap::new(),
intent_path: BTreeMap::new(),
entered_effect_hash: BTreeMap::new(),
entered_inputs: BTreeMap::new(),
entered_seq: BTreeMap::new(),
open_spans: Vec::new(),
live_frames: Vec::new(),
human_requests: BTreeMap::new(),
settled: BTreeMap::new(),
recorded_verdicts: BTreeMap::new(),
hook_triggers: BTreeMap::new(),
executing_ir_hash: None,
cross_ir_resumed: false,
};
let bind_ir_hash = events
.first()
.and_then(|event| match event.run_path.first() {
Some(PathFrame::Flow { ir_hash, .. }) => Some(ir_hash.clone()),
_ => None,
});
let mut request_index: BTreeMap<String, String> = BTreeMap::new();
let mut issuing = IssuingCursor::FromBinding;
for event in events {
match &event.payload {
RunLogPayload::StepEntered {
effect_hash,
resolved_inputs,
..
} => {
let key = instance_key(&event.run_path);
facts.open_spans.push(event.run_path.clone());
facts.settled.remove(&key);
facts.recorded_verdicts.remove(&key);
facts.human_requests.remove(&key);
retain_other_instances(&mut facts.hook_triggers, &key);
facts
.entered_effect_hash
.insert(key.clone(), effect_hash.clone());
facts
.entered_inputs
.insert(key.clone(), resolved_inputs.clone());
facts.entered_seq.insert(key, event.seq);
}
RunLogPayload::StepExited { .. } => {
let key = instance_key(&event.run_path);
if let Some(index) = facts
.open_spans
.iter()
.rposition(|span| instance_key(span) == key)
{
facts.open_spans.remove(index);
}
}
RunLogPayload::CallFramePushed { rebase, .. } => {
if !rebase {
facts.live_frames.push(event.run_path.clone());
}
}
RunLogPayload::CallFramePopped { .. } => {
facts.live_frames.pop();
}
RunLogPayload::RunResumed { event_cursor, .. } => {
if let Some(PathFrame::Flow { ir_hash, .. }) = event.run_path.first() {
if bind_ir_hash.as_ref() != Some(ir_hash) {
facts.cross_ir_resumed = true;
}
facts.executing_ir_hash = Some(ir_hash.clone());
}
issuing = match event_cursor {
Some(cursor) => IssuingCursor::Known(cursor.clone()),
None => IssuingCursor::Unknown,
};
}
RunLogPayload::ActionIntent {
call_id,
chain_index,
..
} => {
if let Some(index) = chain_index {
facts.intent_chain_index.insert(call_id.clone(), *index);
}
facts
.intent_issuing
.insert(call_id.clone(), issuing.clone());
facts
.intent_path
.insert(call_id.clone(), event.run_path.clone());
let key = instance_key(&event.run_path);
let n = attempt_of(&event.run_path).unwrap_or(1);
let entry = facts.max_attempt.entry(key).or_insert(0);
*entry = (*entry).max(n);
}
RunLogPayload::ActionSettled { outcome, .. } => {
let key = instance_key(&event.run_path);
facts.settled.insert(
key.clone(),
SettledFact {
outcome: outcome.clone(),
},
);
let pointlock_ir::ActionOutcome::Succeeded { result } = outcome else {
continue;
};
facts.raw_output.insert(key.clone(), result.output.clone());
match &result.after {
Some(after) => {
facts
.after_observation
.insert(key.clone(), after.id.clone());
}
None => {
facts.after_observation.remove(&key);
}
}
match &result.before {
Some(before) => {
facts.before_observation.insert(key, before.id.clone());
}
None => {
facts.before_observation.remove(&key);
}
}
}
RunLogPayload::VerdictRecorded { verdict, .. } => {
let key = instance_key(&event.run_path);
facts.verdict_seq.insert(key.clone(), event.seq);
facts
.recorded_verdicts
.insert(key, (verdict.status, verdict.degraded));
}
RunLogPayload::HandlerTriggered { hook, trigger, .. } => {
let key = hook_trigger_key(&instance_key(&event.run_path), *hook);
let entry = facts.hook_triggers.entry(key).or_insert(0);
*entry = (*entry).max(*trigger);
}
RunLogPayload::HumanRequested {
request_id,
purpose,
mode,
prompt,
presents,
decisions: _,
output_schema: _,
deadline_at_ms,
} => {
let key = instance_key(&event.run_path);
request_index.insert(request_id.clone(), key.clone());
facts.human_requests.insert(
key,
HumanRequestFact {
request_id: request_id.clone(),
run_path: event.run_path.clone(),
purpose: *purpose,
mode: *mode,
prompt: prompt.clone(),
presents: presents.clone(),
deadline_at_ms: *deadline_at_ms,
final_response: None,
final_actor: None,
},
);
}
RunLogPayload::HumanResponded {
request_id,
purpose,
response,
actor,
} => {
let is_final = !(*purpose == pointlock_ir::HumanPurpose::Supervision
&& response.get("decision").and_then(Value::as_str) == Some("suspend"));
if is_final
&& let Some(key) = request_index.get(request_id)
&& let Some(fact) = facts.human_requests.get_mut(key)
&& fact.request_id == *request_id
{
fact.final_response = Some(response.clone());
fact.final_actor = Some(actor.clone());
}
}
_ => {}
}
}
facts
}
pub(crate) struct Rejudged {
pub run_path: RunPath,
pub verdict: Verdict,
}
pub(crate) struct Alignment {
pub report: AlignmentReport,
pub resume_key: Option<String>,
pub rejudged: Vec<Rejudged>,
pub adoptable: BTreeMap<String, Adopted>,
pub teardown: Option<String>,
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn align(
loaded: &crate::load::LoadedFlow<'_>,
body: &[StepIR],
view: &CheckpointView,
facts: &Harvest,
seed: &ScopeSeed,
store: &Store,
vision: Option<&dyn pointlock_vision::VisionVerifier>,
authorized: &[String],
forced: &[String],
old_root: Option<&FlowIR>,
) -> Result<Alignment, RunnerError> {
let new_flow = loaded.root;
let policy = new_flow.verdict_policy;
let completed = completed_by_instance(view, facts);
let mut nodes = Vec::new();
let mut descended = BTreeSet::new();
let live_pins = live_frame_pins(facts);
flatten(
loaded,
facts,
&live_pins,
&root_path(new_flow),
body,
&completed,
&mut descended,
&mut nodes,
);
let inversion = order_inversion(&nodes, &completed, facts);
let mut entries = Vec::new();
let mut rejudged = Vec::new();
let mut outputs: BTreeMap<String, Value> = BTreeMap::new();
let mut verdicts: BTreeMap<String, (VerdictStatus, bool)> = BTreeMap::new();
let mut adoptable: BTreeMap<String, Adopted> = BTreeMap::new();
let mut first_dirty: Option<usize> = None;
let mut teardown: Option<String> = None;
for (index, node) in nodes.iter().enumerate() {
let step = node.step;
let step_id = step.step_id().clone();
let key = instance_key(&node.path);
if matches!(step, StepIR::Call(_))
&& live_pins.contains_key(&key)
&& !descended.contains(&key)
{
entries.push(AlignmentEntry {
run_path: node.path.clone(),
step_id,
class: AlignmentClass::EffectDirty,
reason: Some(
"call arguments changed while the callee frame was still open; the stale \
frame is torn down and the callee re-called with freshly evaluated \
inputs (07 §5.2 case (b))"
.to_owned(),
),
});
first_dirty.get_or_insert(index);
teardown = Some(key);
continue;
}
if matches!(step, StepIR::Call(_)) && descended.contains(&key) {
let pin_moved = live_pins.get(&key) != Some(&call_pin(step));
entries.push(AlignmentEntry {
run_path: node.path.clone(),
step_id,
class: AlignmentClass::EffectDirty,
reason: Some(if pin_moved {
"callee content changed, arguments untouched; the live callee frame is \
re-entered and aligned step by step (07 §5.2 case (a) down-drill)"
.to_owned()
} else {
"unchanged call, frame still open; re-entered and aligned step by step"
.to_owned()
}),
});
continue;
}
let Some(record) = completed.get(&key) else {
entries.push(AlignmentEntry {
run_path: node.path.clone(),
step_id,
class: AlignmentClass::New,
reason: Some("no adoptable prior record".to_owned()),
});
first_dirty.get_or_insert(index);
continue;
};
let class = {
let effect_same = record.effect_hash == step.base().effect_hash;
let judge_same = record.judge_hash == step.base().judge_hash;
match (effect_same, judge_same) {
(true, true) => AlignmentClass::Reusable,
(true, false) => AlignmentClass::JudgeDirty,
(false, _) => AlignmentClass::EffectDirty,
}
};
let reordered = inversion.is_some_and(|(from, _, _)| index >= from);
let forced_hit = forced.iter().any(|name| name == step_id.as_str());
let class = if reordered || forced_hit {
AlignmentClass::EffectDirty
} else {
class
};
match class {
AlignmentClass::Reusable => {
let reason = if first_dirty.is_some() {
Some("positionally invalidated: after the resume point".to_owned())
} else {
adopt(
step,
&key,
record,
facts,
seed,
&mut outputs,
&mut verdicts,
&mut adoptable,
None,
);
None
};
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason,
});
}
AlignmentClass::JudgeDirty => {
if first_dirty.is_some() {
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason: Some(
"judgeHash changed; re-executes after the resume point".to_owned(),
),
});
continue;
}
if preflight_only_change(old_root, record, step) {
adopt(
step,
&key,
record,
facts,
seed,
&mut outputs,
&mut verdicts,
&mut adoptable,
None,
);
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason: Some("preflightChanged".to_owned()),
});
continue;
}
if node
.path
.iter()
.any(|frame| matches!(frame, PathFrame::Iteration { .. }))
{
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class: AlignmentClass::EffectDirty,
reason: Some(
"judgeHash changed inside a foreach round; re-executed rather than \
re-judged (an offline re-judge there has no `iter` binding)"
.to_owned(),
),
});
first_dirty.get_or_insert(index);
continue;
}
if matches!(step, StepIR::Assert(_)) {
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class: AlignmentClass::EffectDirty,
reason: Some(
"judgeHash changed on an assert step; re-observed rather than \
re-judged"
.to_owned(),
),
});
first_dirty.get_or_insert(index);
continue;
}
let StepIR::Action(action) = step else {
adopt(
step,
&key,
record,
facts,
seed,
&mut outputs,
&mut verdicts,
&mut adoptable,
None,
);
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason: Some("preflightChanged".to_owned()),
});
continue;
};
let verdict = rejudge(
action, &key, record, facts, seed, &outputs, &verdicts, policy, store, vision,
)
.await;
let failed = verdict.status == VerdictStatus::Fail;
rejudged.push(Rejudged {
run_path: record.run_path.clone(),
verdict: verdict.clone(),
});
if failed {
first_dirty = Some(index);
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason: Some("offline re-judge failed; step re-executes".to_owned()),
});
} else {
adopt(
step,
&key,
record,
facts,
seed,
&mut outputs,
&mut verdicts,
&mut adoptable,
Some((verdict.status, verdict.degraded)),
);
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason: Some("judgeHash changed; re-judged offline".to_owned()),
});
}
}
AlignmentClass::EffectDirty => {
let reason = match inversion {
Some((_, earlier_id, later_id)) if reordered => Some(format!(
"order invalidated: '{later_id}' ran before '{earlier_id}' in the old \
run, reversing their order in the new IR; no record is adopted from \
there on"
)),
_ if forced_hit => {
Some("forced re-execution (--force-reexecute, 07 §5.3)".to_owned())
}
_ => Some("effectHash changed".to_owned()),
};
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id,
class,
reason,
});
first_dirty.get_or_insert(index);
}
AlignmentClass::New | AlignmentClass::Orphaned => {
unreachable!("classified above")
}
}
}
for node in nodes.iter().rev() {
if !matches!(node.step, StepIR::If(_) | StepIR::Foreach(_)) {
continue;
}
let key = instance_key(&node.path);
if !adoptable.contains_key(&key) {
continue;
}
let subtree_intact = nodes
.iter()
.filter(|other| is_descendant(&node.path, &other.path))
.all(|child| adoptable.contains_key(&instance_key(&child.path)));
if !subtree_intact {
adoptable.remove(&key);
}
}
let node_keys: BTreeSet<String> = nodes.iter().map(|node| instance_key(&node.path)).collect();
for record in completed.values() {
if record
.run_path
.iter()
.any(|frame| matches!(frame, PathFrame::Hook { .. }))
{
continue;
}
let key = instance_key(&record.run_path);
if node_keys.contains(&key) {
continue;
}
let enclosing = record.run_path.len().saturating_sub(1);
let inside_undrilled_frame = record.run_path[..enclosing]
.iter()
.enumerate()
.filter(|(_, frame)| matches!(frame, PathFrame::Call { .. }))
.any(|(index, _)| !descended.contains(&instance_key(&record.run_path[..=index])));
if inside_undrilled_frame {
continue;
}
entries.push(AlignmentEntry {
run_path: record.run_path.clone(),
step_id: record.step_id.clone(),
class: AlignmentClass::Orphaned,
reason: Some("absent from the new IR".to_owned()),
});
}
let resume_idx = first_dirty.unwrap_or_else(|| {
let cursor = view
.frames
.first()
.map(|frame| frame.next_index as usize)
.unwrap_or(0);
nodes
.iter()
.enumerate()
.filter(|(_, node)| node.path.len() == 2)
.nth(cursor)
.map(|(index, _)| index)
.unwrap_or(nodes.len())
});
let mut class_by_instance: BTreeMap<String, AlignmentClass> = BTreeMap::new();
for entry in &entries {
class_by_instance
.entry(instance_key(&entry.run_path))
.or_insert(entry.class);
}
let mut requires_confirmation = Vec::new();
for (index, node) in nodes.iter().enumerate().skip(resume_idx) {
let sid = node.step.step_id().as_str();
let key = instance_key(&node.path);
if teardown.as_deref() == Some(key.as_str()) {
let effective = gated_effect(node.step, false, loaded)
&& completed.iter().any(|(other, inner)| {
is_instance_descendant(&key, other)
&& !inner
.run_path
.iter()
.any(|frame| matches!(frame, PathFrame::Hook { .. }))
&& prior_effect_possible(inner)
});
if effective && !authorized.iter().any(|allowed| allowed == sid) {
let run_path = facts
.live_frames
.iter()
.find(|path| instance_key(path) == key)
.cloned()
.unwrap_or_else(|| node.path.clone());
requires_confirmation.push(RequiresConfirmation {
run_path,
step_id: Some(node.step.step_id().clone()),
cause: "mutatingReexec".to_owned(),
reason: format!(
"re-calling '{sid}' (traversal index {index}) tears down a frame whose \
completed steps include a mutating, non-idempotent action that took \
effect; the re-call needs explicit authorization (07 §5.2 case (b))"
),
});
}
continue;
}
let Some(record) = completed.get(&key) else {
continue;
};
if !gated_effect(node.step, descended.contains(&key), loaded) {
continue;
}
let effective = prior_effect_possible(record)
|| (is_container(node.step)
&& completed.iter().any(|(other, inner)| {
is_instance_descendant(&key, other)
&& !inner
.run_path
.iter()
.any(|frame| matches!(frame, PathFrame::Hook { .. }))
&& prior_effect_possible(inner)
}));
if !effective {
continue;
}
if authorized.iter().any(|allowed| allowed == sid) {
continue;
}
let cause = if inversion.is_some_and(|(from, _, _)| index >= from) {
"orderInvalidated"
} else {
match class_by_instance.get(&key) {
Some(AlignmentClass::EffectDirty) => "mutatingReexec",
_ => "positionalReplay",
}
};
requires_confirmation.push(RequiresConfirmation {
run_path: record.run_path.clone(),
step_id: Some(node.step.step_id().clone()),
cause: cause.to_owned(),
reason: if is_container(node.step) {
format!(
"re-executing '{sid}' (traversal index {index}) replays steps inside it \
that are mutating, not idempotent, and whose prior effect may be in the \
world; it needs explicit authorization"
)
} else {
format!(
"step '{sid}' (traversal index {index}) is mutating, not idempotent, and \
its prior effect may be in the world; re-execution needs explicit \
authorization"
)
},
});
}
if let Some(stranded) = facts.live_frames.iter().find(|path| {
let key = instance_key(path);
!path
.iter()
.any(|frame| matches!(frame, PathFrame::Hook { .. }))
&& !descended.contains(&key)
&& teardown.as_deref() != Some(key.as_str())
}) {
return Err(RunnerError::M0Unsupported {
detail: format!(
"live frame {} is neither re-entered nor torn down — the new IR's walk never \
reached its call site (a container above it changed shape); resume cannot \
address it",
pointlock_ir::render_run_path(stranded)
),
});
}
let resume_point = nodes.get(resume_idx).map(|node| node.path.clone());
let report = AlignmentReport {
entries,
resume_point,
requires_confirmation,
};
if !report.requires_confirmation.is_empty() {
return Err(RunnerError::RequiresConfirmation {
report: Box::new(report),
});
}
Ok(Alignment {
report,
resume_key: nodes.get(resume_idx).map(|node| instance_key(&node.path)),
rejudged,
adoptable,
teardown,
})
}
fn order_inversion<'a>(
nodes: &'a [Node<'a>],
completed: &BTreeMap<String, &StepRecord>,
facts: &Harvest,
) -> Option<(usize, &'a str, &'a str)> {
let mut previous: Option<(&'a str, u64)> = None;
for (index, node) in nodes.iter().enumerate() {
let sid = node.step.step_id().as_str();
let key = instance_key(&node.path);
if !completed.contains_key(&key) {
continue;
}
let Some(&seq) = facts.entered_seq.get(&key) else {
continue;
};
if let Some((earlier_id, earlier_seq)) = previous
&& seq <= earlier_seq
{
return Some((index, earlier_id, sid));
}
previous = Some((sid, seq));
}
None
}
fn completed_by_instance<'v>(
view: &'v CheckpointView,
facts: &Harvest,
) -> BTreeMap<String, &'v StepRecord> {
let open = open_instances(facts);
let mut map = BTreeMap::new();
for record in &view.completed {
if !is_history(record) {
continue;
}
let key = instance_key(&record.run_path);
if open.contains(&key) {
continue;
}
map.insert(key, record);
}
map
}
pub(crate) fn open_instances(facts: &Harvest) -> BTreeSet<String> {
facts
.open_spans
.iter()
.map(|path| instance_key(path))
.collect()
}
#[allow(clippy::too_many_arguments)]
fn adopt(
step: &StepIR,
key: &str,
record: &StepRecord,
facts: &Harvest,
seed: &ScopeSeed,
outputs: &mut BTreeMap<String, Value>,
verdicts: &mut BTreeMap<String, (VerdictStatus, bool)>,
adoptable: &mut BTreeMap<String, Adopted>,
override_verdict: Option<(VerdictStatus, bool)>,
) {
let sid = step.step_id().as_str();
if let (StepIR::Action(action), Some(raw)) = (step, facts.raw_output.get(key)) {
let scope = seed.scope(outputs, verdicts, Some((sid, raw)));
if let Ok(projected) = project_output(action, raw, &scope) {
outputs.insert(sid.to_owned(), projected);
}
}
if let Some((status, degraded)) = override_verdict {
verdicts.insert(sid.to_owned(), (status, degraded));
} else if let Some(verdict) = &record.verdict {
verdicts.insert(sid.to_owned(), (verdict.status, verdict.degraded));
}
let mut adopted_record = record.clone();
if let Some((status, degraded)) = override_verdict
&& let Some(verdict) = &mut adopted_record.verdict
{
verdict.status = status;
verdict.degraded = degraded;
}
adoptable.insert(
key.to_owned(),
Adopted {
record: adopted_record,
before_id: facts.before_observation.get(key).cloned(),
after_id: facts.after_observation.get(key).cloned(),
},
);
}
#[allow(clippy::too_many_arguments)]
async fn rejudge(
step: &ActionStepIR,
key: &str,
record: &StepRecord,
facts: &Harvest,
seed: &ScopeSeed,
outputs: &BTreeMap<String, Value>,
verdicts: &BTreeMap<String, (VerdictStatus, bool)>,
policy: VerdictPolicy,
store: &Store,
vision: Option<&dyn pointlock_vision::VisionVerifier>,
) -> Verdict {
let sid = step.base.step_id.as_str();
let degraded = record
.verdict
.as_ref()
.map(|verdict| verdict.degraded)
.unwrap_or(false);
let supersedes = facts.verdict_seq.get(key).map(|seq| format!("seq:{seq}"));
let folded = match facts.raw_output.get(key) {
None => crate::judge::FoldedVerdict {
status: VerdictStatus::Unknown,
degraded,
summary: "offline re-judge: no archived output for this step (missing input ⇒ \
unknown)"
.to_owned(),
},
Some(raw) => {
let raw_scope = seed.scope(outputs, verdicts, Some((sid, raw)));
match project_output(step, raw, &raw_scope) {
Err(error) => crate::judge::FoldedVerdict {
status: VerdictStatus::Unknown,
degraded,
summary: format!("offline re-judge: output projection failed: {error}"),
},
Ok(projected) => {
let scope = seed.scope(outputs, verdicts, Some((sid, &projected)));
let material = archived_material(
record,
facts.after_observation.get(key).map(String::as_str),
store,
);
let mut assertion_outcomes = Vec::with_capacity(step.assertions.len());
let mut degraded_verify = false;
for assertion in &step.assertions {
let evaluated = match &assertion.predicate {
PredicateIR::Expr { expr } => EvaluatedAssertion {
record: eval_expr_assertion(assertion, expr, &scope),
degraded_verify: false,
},
_ => eval_observed_assertion(assertion, &material, vision).await,
};
degraded_verify |= evaluated.degraded_verify;
assertion_outcomes.push(evaluated.record);
}
if assertion_outcomes.is_empty() {
crate::judge::FoldedVerdict {
status: VerdictStatus::Unknown,
degraded,
summary: "offline re-judge: the new step declares no assertions"
.to_owned(),
}
} else {
let mut folded = fold_step_verdict(
&assertion_outcomes,
degraded,
degraded_verify,
policy,
);
folded.summary =
format!("offline re-judge over archived output: {}", folded.summary);
folded
}
}
}
}
};
Verdict {
status: folded.status,
degraded: folded.degraded,
summary: folded.summary,
evidence: Vec::new(),
supersedes,
}
}
fn archived_material(
record: &StepRecord,
after_id: Option<&str>,
store: &Store,
) -> ObserveMaterial {
let Some(after_id) = after_id else {
return ObserveMaterial::absent(
"offline re-judge: the archived run recorded no after observation",
);
};
let Some(observation) = record
.observations
.iter()
.find(|observation| observation.observation_id == after_id)
else {
return ObserveMaterial::absent(
"offline re-judge: the after observation was never localized",
);
};
material_from_observation(store, observation)
}
fn prior_effect_possible(record: &StepRecord) -> bool {
record.attempts.iter().any(|attempt| {
matches!(
attempt.outcome,
ActionOutcomeKind::Succeeded | ActionOutcomeKind::TimedOut
)
})
}
#[cfg(test)]
mod tests {
use pointlock_ir::{FlowId, RunLogEvent, StepId, StepState};
use serde_json::json;
use super::*;
fn hash(fill: char) -> Hash {
Hash::new(format!("sha256:{}", fill.to_string().repeat(64))).expect("valid hash")
}
fn event(seq: u64, run_path: RunPath, payload: RunLogPayload) -> RunLogEvent {
RunLogEvent {
run_id: "run-1".to_owned(),
seq,
at_ms: seq,
run_path,
payload,
}
}
fn step_entered(id: &str) -> RunLogPayload {
RunLogPayload::StepEntered {
step_id: StepId::new(id).expect("valid step id"),
effect_hash: hash('c'),
judge_hash: hash('d'),
resolved_inputs: json!({}),
}
}
fn step_exited() -> RunLogPayload {
RunLogPayload::StepExited {
provider_state_summary: None,
state: StepState::Judged,
output: None,
localized: Vec::new(),
localization_gaps: Vec::new(),
}
}
#[test]
fn a_fresh_step_entered_resets_the_instance_handler_trigger_budget() {
let flow = PathFrame::Flow {
flow_id: FlowId::new("checkout").expect("valid flow id"),
ir_hash: hash('a'),
};
let step = |id: &str| PathFrame::Step {
step_id: StepId::new(id).expect("valid step id"),
};
let pay = vec![flow.clone(), step("pay")];
let other = vec![flow.clone(), step("other")];
let triggered = |trigger: u64| RunLogPayload::HandlerTriggered {
hook: HandlerHook::OnFail,
trigger,
disposition: Some("retry".to_owned()),
};
let life_one = vec![
event(1, pay.clone(), step_entered("pay")),
event(2, pay.clone(), triggered(1)),
event(3, pay.clone(), triggered(2)),
event(4, pay.clone(), step_exited()),
event(5, other.clone(), step_entered("other")),
event(6, other.clone(), triggered(1)),
event(7, other.clone(), step_exited()),
];
let pay_key = hook_trigger_key(&instance_key(&pay), HandlerHook::OnFail);
let other_key = hook_trigger_key(&instance_key(&other), HandlerHook::OnFail);
let facts = harvest(&life_one);
assert_eq!(
facts.hook_triggers.get(&pay_key),
Some(&2),
"life 1 exhausted"
);
assert_eq!(facts.hook_triggers.get(&other_key), Some(&1));
let mut life_two = life_one;
life_two.push(event(8, pay.clone(), step_entered("pay")));
let facts = harvest(&life_two);
assert_eq!(
facts.hook_triggers.get(&pay_key),
None,
"a new life starts with a fresh budget"
);
assert_eq!(
facts.hook_triggers.get(&other_key),
Some(&1),
"another instance's counter is untouched"
);
}
#[test]
fn a_container_exit_closes_its_own_span_over_an_orphaned_open_span() {
let flow = PathFrame::Flow {
flow_id: FlowId::new("checkout").expect("valid flow id"),
ir_hash: hash('a'),
};
let step = |id: &str| PathFrame::Step {
step_id: StepId::new(id).expect("valid step id"),
};
let iteration = PathFrame::Iteration {
index: 0,
key: None,
};
let container: RunPath = vec![flow.clone(), step("each")];
let orphan: RunPath = vec![flow.clone(), step("each"), iteration.clone(), step("x")];
let renamed: RunPath = vec![flow, step("each"), iteration, step("x2")];
let events = [
event(1, container.clone(), step_entered("each")),
event(2, orphan.clone(), step_entered("x")),
event(3, renamed.clone(), step_entered("x2")),
event(4, renamed, step_exited()),
event(5, container, step_exited()),
];
let facts = harvest(&events);
assert_eq!(
facts.open_spans,
vec![orphan],
"only the orphan stays open; the container's exit closed the container"
);
}
#[test]
fn order_inversion_reasons_with_the_latest_life_of_a_reexecuted_step() {
let flow = PathFrame::Flow {
flow_id: FlowId::new("checkout").expect("valid flow id"),
ir_hash: hash('a'),
};
let step_path = |id: &str| -> RunPath {
vec![
flow.clone(),
PathFrame::Step {
step_id: StepId::new(id).expect("valid step id"),
},
]
};
let s1 = step_path("s1");
let s2 = step_path("s2");
let events = [
event(1, s1.clone(), step_entered("s1")),
event(2, s1.clone(), step_exited()),
event(3, s2.clone(), step_entered("s2")),
event(4, s2.clone(), step_exited()),
event(5, s1.clone(), step_entered("s1")),
event(6, s1.clone(), step_exited()),
];
let facts = harvest(&events);
let step_ir = |id: &str| -> StepIR {
serde_json::from_value(json!({
"kind": "let", "stepId": id, "effectHash": hash('c').as_str(),
"judgeHash": hash('d').as_str(), "checkpoint": false,
"bindings": { "x": { "lit": 1 } }
}))
.expect("a minimal let step")
};
let (ir1, ir2) = (step_ir("s1"), step_ir("s2"));
let nodes = [
Node {
path: s1.clone(),
step: &ir1,
},
Node {
path: s2.clone(),
step: &ir2,
},
];
let record = |path: &RunPath, id: &str| StepRecord {
run_path: path.clone(),
step_id: StepId::new(id).expect("valid step id"),
effect_hash: hash('c'),
judge_hash: hash('d'),
attempts: Vec::new(),
resolved_inputs: json!({}),
output: None,
observations: Vec::new(),
evidence: Vec::new(),
assertion_outcomes: Vec::new(),
verdict: None,
};
let (r1, r2) = (record(&s1, "s1"), record(&s2, "s2"));
let completed: BTreeMap<String, &StepRecord> =
[(instance_key(&s1), &r1), (instance_key(&s2), &r2)]
.into_iter()
.collect();
assert_eq!(
order_inversion(&nodes, &completed, &facts),
Some((1, "s1", "s2")),
"s1's latest life (seq 5) ran after s2 (seq 3)"
);
}
}