use std::collections::BTreeMap;
use pointlock_ir::{
ActionOutcome, AssertionOutcomeRecord, ErrorInfo, EvidenceRef, FlowIR, JsonPointer,
ObservationRecord, PathFrame, RunLogEvent, RunLogPayload, RunPath, SourceMapEntry, StepIR,
StepRecord, StepState, StepVerdict, Verdict, parse_run_path, render_parsed_run_path,
render_run_path,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use super::ProjectionVersion;
use crate::error::StoreError;
use crate::store::Store;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AttemptView {
#[serde(skip_serializing_if = "Option::is_none")]
pub n: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub chain_index: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub channel: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub action_name: Option<String>,
pub call_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub outcome: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_class: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub execution_mode: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub fallback_reason: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<ErrorInfo>,
pub args_snapshot: Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub started_at_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub finished_at_ms: Option<u64>,
pub intent_seq: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub settled_seq: Option<u64>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub evidence: Vec<pointlock_ir::AssetRef>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct VerdictRecordView {
pub seq: u64,
pub at_ms: u64,
pub verdict: Verdict,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub localized: Vec<EvidenceRef>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub localization_gaps: Vec<pointlock_ir::EvidenceGap>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct EvidenceGapView {
pub asset: pointlock_ir::AssetRef,
pub reason: String,
pub seq: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct EvidenceItemView {
#[serde(flatten)]
pub reference: EvidenceRef,
pub source: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct HandlerTriggerView {
pub hook: String,
pub trigger: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub disposition: Option<String>,
pub seq: u64,
pub at_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SourceLocation {
pub ir_path: JsonPointer,
pub entry: SourceMapEntry,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct FrameEnvironment {
pub inputs_snapshot: Value,
pub vars: BTreeMap<String, Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub provider_state_summary: Option<pointlock_ir::ProviderStateSummary>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct StepDossierView {
pub projection_version: ProjectionVersion,
pub run_id: String,
pub run_path: String,
pub run_path_frames: RunPath,
pub step_id: String,
pub effect_hash: String,
pub judge_hash: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub ir_node: Option<StepIR>,
#[serde(skip_serializing_if = "Option::is_none")]
pub source: Option<SourceLocation>,
pub resolved_inputs: Value,
pub preflight: Vec<AssertionOutcomeRecord>,
pub attempts: Vec<AttemptView>,
#[serde(skip_serializing_if = "Option::is_none")]
pub output: Option<Value>,
pub observations: Vec<ObservationRecord>,
pub assertion_outcomes: Vec<AssertionOutcomeRecord>,
#[serde(skip_serializing_if = "Option::is_none")]
pub verdict: Option<StepVerdict>,
pub verdict_history: Vec<VerdictRecordView>,
pub handler_triggers: Vec<HandlerTriggerView>,
pub evidence: Vec<EvidenceItemView>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub evidence_gaps: Vec<EvidenceGapView>,
#[serde(skip_serializing_if = "Option::is_none")]
pub provider_state_summary: Option<pointlock_ir::ProviderStateSummary>,
#[serde(skip_serializing_if = "Option::is_none")]
pub frame: Option<FrameEnvironment>,
#[serde(skip_serializing_if = "Option::is_none")]
pub state: Option<StepState>,
}
fn instance_path(path: &[PathFrame]) -> Vec<PathFrame> {
path.iter()
.filter(|frame| {
!matches!(
frame,
PathFrame::Attempt { .. } | PathFrame::Phase { .. } | PathFrame::Assertion { .. }
)
})
.cloned()
.collect()
}
fn attempt_of(path: &[PathFrame]) -> Option<u64> {
path.iter().rev().find_map(|frame| match frame {
PathFrame::Attempt { n } => Some(*n),
_ => None,
})
}
fn wire<T: Serialize>(value: &T) -> String {
serde_json::to_value(value)
.ok()
.and_then(|v| v.as_str().map(str::to_owned))
.unwrap_or_default()
}
fn entered_instances(events: &[RunLogEvent]) -> Vec<(String, RunPath)> {
let mut seen = Vec::new();
let mut keys = std::collections::BTreeSet::new();
for event in events {
if matches!(event.payload, RunLogPayload::StepEntered { .. }) {
let instance = instance_path(&event.run_path);
let key = render_run_path(&instance);
if keys.insert(key.clone()) {
seen.push((key, instance));
}
}
}
seen
}
pub fn locate_step(store: &Store, run_id: &str, step: &str) -> Result<RunPath, StoreError> {
let events = store.events(run_id)?;
let instances = entered_instances(&events);
if step.trim_start().starts_with('[') {
let frames: RunPath = serde_json::from_str(step).map_err(|err| StoreError::BadRunPath {
input: step.to_owned(),
message: format!("not a JSON PathFrame[] document: {err}"),
})?;
let wanted = render_run_path(&instance_path(&frames));
return instances
.into_iter()
.find(|(key, _)| *key == wanted)
.map(|(_, path)| path)
.ok_or_else(|| StoreError::UnknownStepInstance {
run_id: run_id.to_owned(),
path: wanted,
});
}
if step.contains('/') || step.contains('@') {
let parsed = parse_run_path(step).map_err(|err| StoreError::BadRunPath {
input: step.to_owned(),
message: format!("{} at offset {}", err.message, err.offset),
})?;
let wanted = render_parsed_run_path(&instance_parsed(&parsed));
return instances
.into_iter()
.find(|(key, _)| *key == wanted)
.map(|(_, path)| path)
.ok_or_else(|| StoreError::UnknownStepInstance {
run_id: run_id.to_owned(),
path: step.to_owned(),
});
}
let matches: Vec<(String, RunPath)> = instances
.into_iter()
.filter(|(_, path)| {
instance_step_id(path)
.map(|step_id| step_id.as_ref() == step)
.unwrap_or(false)
})
.collect();
match matches.len() {
0 => Err(StoreError::UnknownStepInstance {
run_id: run_id.to_owned(),
path: step.to_owned(),
}),
1 => Ok(matches.into_iter().next().expect("len checked").1),
_ => Err(StoreError::AmbiguousStep {
run_id: run_id.to_owned(),
step: step.to_owned(),
candidates: matches.into_iter().map(|(key, _)| key).collect(),
}),
}
}
fn instance_step_id(path: &[PathFrame]) -> Option<&pointlock_ir::StepId> {
path.iter().rev().find_map(|frame| match frame {
PathFrame::Step { step_id } => Some(step_id),
PathFrame::Call {
step_id: Some(step_id),
..
} => Some(step_id),
_ => None,
})
}
fn instance_parsed(path: &[pointlock_ir::ParsedPathFrame]) -> Vec<pointlock_ir::ParsedPathFrame> {
use pointlock_ir::ParsedPathFrame as P;
path.iter()
.filter(|frame| {
!matches!(
frame,
P::Attempt { .. } | P::Phase { .. } | P::Assertion { .. }
)
})
.cloned()
.collect()
}
pub fn step_dossier(
store: &Store,
run_id: &str,
path: &[PathFrame],
artifacts: &[FlowIR],
) -> Result<StepDossierView, StoreError> {
let events = store.events(run_id)?;
let instance = instance_path(path);
let key = render_run_path(&instance);
let step_id =
instance_step_id(&instance)
.cloned()
.ok_or_else(|| StoreError::UnknownStepInstance {
run_id: run_id.to_owned(),
path: key.clone(),
})?;
let mut entered: Option<(String, String, Value)> = None; let mut preflight = Vec::new();
let mut attempts: Vec<AttemptView> = Vec::new();
let mut by_call: BTreeMap<String, usize> = BTreeMap::new();
let mut observations = Vec::new();
let mut assertion_outcomes = Vec::new();
let mut verdict_history = Vec::new();
let mut handler_triggers = Vec::new();
let mut evidence: Vec<EvidenceItemView> = Vec::new();
let mut evidence_gaps: Vec<EvidenceGapView> = Vec::new();
let mut output = None;
let mut state: Option<StepState> = None;
let mut step_summary: Option<pointlock_ir::ProviderStateSummary> = None;
let mut touched = false;
for event in &events {
if render_run_path(&instance_path(&event.run_path)) != key {
continue;
}
touched = true;
match &event.payload {
RunLogPayload::StepEntered {
effect_hash,
judge_hash,
resolved_inputs,
..
} => {
entered = Some((
effect_hash.to_string(),
judge_hash.to_string(),
resolved_inputs.clone(),
));
}
RunLogPayload::PreflightProbed { outcomes } => {
preflight.extend(outcomes.iter().cloned());
}
RunLogPayload::ActionIntent {
call_id,
args_snapshot,
chain_index,
channel,
action_name,
} => {
by_call.insert(call_id.clone(), attempts.len());
attempts.push(AttemptView {
n: attempt_of(&event.run_path),
chain_index: *chain_index,
channel: channel.as_ref().map(wire),
action_name: action_name.as_ref().map(|name| name.as_str().to_owned()),
call_id: call_id.clone(),
outcome: None,
error_class: None,
execution_mode: None,
fallback_reason: None,
error: None,
args_snapshot: args_snapshot.clone(),
started_at_ms: None,
finished_at_ms: None,
intent_seq: event.seq,
settled_seq: None,
evidence: Vec::new(),
});
}
RunLogPayload::ActionSettled { call_id, outcome } => {
if let Some(&index) = by_call.get(call_id) {
let attempt = &mut attempts[index];
attempt.outcome = Some(outcome.kind().to_owned());
if let ActionOutcome::Succeeded { result } = outcome {
attempt.evidence = result.evidence.clone();
}
attempt.settled_seq = Some(event.seq);
match outcome {
ActionOutcome::Succeeded { result } => {
attempt.started_at_ms = Some(result.started_at_ms);
attempt.finished_at_ms = Some(result.finished_at_ms);
attempt.execution_mode =
result.execution.as_ref().map(|execution| match execution {
pointlock_ir::ActionExecution::NativeSemantic { .. } => {
"nativeSemantic".to_owned()
}
pointlock_ir::ActionExecution::WebSemantic { .. } => {
"webSemantic".to_owned()
}
pointlock_ir::ActionExecution::CoordinateFallback {
fallback_reason,
..
} => {
attempt.fallback_reason = Some(wire(fallback_reason));
"coordinateFallback".to_owned()
}
});
}
ActionOutcome::Failed { error }
| ActionOutcome::Cancelled { error }
| ActionOutcome::TimedOut { error } => {
attempt.error = Some(error.clone());
}
}
}
}
RunLogPayload::ObservationRecorded { observation } => {
if let Some(evidence_ref) = &observation.screenshot {
evidence.push(EvidenceItemView {
reference: evidence_ref.clone(),
source: format!("observation:{}/screenshot", observation.observation_id),
});
}
if let Some(evidence_ref) = &observation.ui_snapshot {
evidence.push(EvidenceItemView {
reference: evidence_ref.clone(),
source: format!("observation:{}/uiSnapshot", observation.observation_id),
});
}
observations.push(observation.clone());
}
RunLogPayload::AssertionEvaluated { outcome } => {
assertion_outcomes.push(outcome.clone());
}
RunLogPayload::VerdictRecorded {
verdict,
localized,
localization_gaps,
remote_archival_error: _,
} => {
for entry in localized {
let duplicate = evidence.iter().any(|existing| {
existing.reference.sha256 == entry.sha256
&& existing.reference.asset.id == entry.asset.id
});
if !duplicate {
evidence.push(EvidenceItemView {
reference: entry.clone(),
source: format!("verdict@seq:{}", event.seq),
});
}
}
for gap in localization_gaps {
evidence_gaps.push(EvidenceGapView {
asset: gap.asset.clone(),
reason: gap.reason.clone(),
seq: event.seq,
});
}
verdict_history.push(VerdictRecordView {
seq: event.seq,
at_ms: event.at_ms,
verdict: verdict.clone(),
localized: localized.clone(),
localization_gaps: localization_gaps.clone(),
});
}
RunLogPayload::HandlerTriggered {
hook,
trigger,
disposition,
} => {
handler_triggers.push(HandlerTriggerView {
hook: wire(hook),
trigger: *trigger,
disposition: disposition.clone(),
seq: event.seq,
at_ms: event.at_ms,
});
}
RunLogPayload::StepExited {
provider_state_summary,
state: exit_state,
output: exit_output,
localized,
localization_gaps,
} => {
state = Some(*exit_state);
if exit_output.is_some() {
output = exit_output.clone();
}
if provider_state_summary.is_some() {
step_summary = provider_state_summary.clone();
}
for entry in localized {
let duplicate = evidence.iter().any(|existing| {
existing.reference.sha256 == entry.sha256
&& existing.reference.asset.id == entry.asset.id
});
if !duplicate {
evidence.push(EvidenceItemView {
reference: entry.clone(),
source: format!("exit@seq:{}", event.seq),
});
}
}
for gap in localization_gaps {
evidence_gaps.push(EvidenceGapView {
asset: gap.asset.clone(),
reason: gap.reason.clone(),
seq: event.seq,
});
}
}
_ => {}
}
}
if !touched || entered.is_none() {
return Err(StoreError::UnknownStepInstance {
run_id: run_id.to_owned(),
path: key.clone(),
});
}
let (effect_hash, judge_hash, resolved_inputs) = entered.expect("checked above");
let checkpoint = store.materialized_checkpoint(run_id)?;
if let Some((_, view)) = &checkpoint {
if let Some(record) = view
.completed
.iter()
.find(|record: &&StepRecord| render_run_path(&record.run_path) == key)
{
for recorded in &record.attempts {
if let Some(&index) = by_call.get(&recorded.call_id) {
let attempt = &mut attempts[index];
attempt.error_class = recorded.error_class.as_ref().map(wire);
if attempt.execution_mode.is_none() {
attempt.execution_mode = recorded.execution_mode.as_ref().map(wire);
}
if attempt.fallback_reason.is_none() {
attempt.fallback_reason = recorded.fallback_reason.as_ref().map(wire);
}
}
}
}
if state.is_none() && render_run_path(&instance_path(&view.frontier.run_path)) == key {
state = Some(view.frontier.state);
}
}
let verdict = verdict_history.last().map(|record| StepVerdict {
status: record.verdict.status,
degraded: record.verdict.degraded,
supersedes: record.verdict.supersedes.clone(),
});
let scan = match instance.last() {
Some(PathFrame::Call { .. }) => &instance[..instance.len() - 1],
_ => instance.as_slice(),
};
let governing_hash = scan.iter().rev().find_map(|frame| match frame {
PathFrame::Flow { ir_hash, .. } => Some(ir_hash.clone()),
PathFrame::Call { callee_ir_hash, .. } => Some(callee_ir_hash.clone()),
_ => None,
});
let mut ir_node = None;
let mut source = None;
if let Some(hash) = governing_hash
&& let Some(flow) = artifacts.iter().find(|flow| flow.ir_hash == hash)
&& let Some((node, pointer)) = find_step(&flow.body, "/body", step_id.as_ref())
{
source = flow
.source_map
.iter()
.find(|entry| entry.ir_path.as_ref() == pointer)
.map(|entry| SourceLocation {
ir_path: entry.ir_path.clone(),
entry: entry.clone(),
});
ir_node = Some(node.clone());
}
let frame = checkpoint
.as_ref()
.and_then(|(_, view)| frame_environment(&instance, &view.frames))
.map(|mut environment| {
environment.provider_state_summary = step_summary.clone();
environment
});
Ok(StepDossierView {
projection_version: ProjectionVersion,
run_id: run_id.to_owned(),
run_path: key,
run_path_frames: instance,
step_id: step_id.to_string(),
effect_hash,
judge_hash,
ir_node,
source,
resolved_inputs,
preflight,
attempts,
output,
observations,
assertion_outcomes,
verdict,
verdict_history,
handler_triggers,
evidence,
evidence_gaps,
provider_state_summary: step_summary,
frame,
state,
})
}
fn frame_environment(
instance: &[PathFrame],
frames: &[pointlock_ir::CallFrame],
) -> Option<FrameEnvironment> {
let mut level = 0usize;
for (index, path_frame) in instance.iter().enumerate() {
match path_frame {
PathFrame::Flow { ir_hash, flow_id } => {
let frame = frames.first()?;
if frame.ir_hash != *ir_hash
|| frame.flow_id != *flow_id
|| frame.call_step_id.is_some()
{
return None;
}
}
PathFrame::Call {
step_id,
callee_flow_id,
callee_ir_hash,
} => {
if index + 1 == instance.len() {
break;
}
level += 1;
let frame = frames.get(level)?;
if frame.call_step_id != *step_id
|| frame.flow_id != *callee_flow_id
|| frame.ir_hash != *callee_ir_hash
{
return None;
}
}
_ => {}
}
}
frames.get(level).map(|frame| FrameEnvironment {
inputs_snapshot: frame.inputs_snapshot.clone(),
vars: frame.vars.clone(),
provider_state_summary: None,
})
}
fn find_step<'a>(body: &'a [StepIR], prefix: &str, step_id: &str) -> Option<(&'a StepIR, String)> {
for (index, step) in body.iter().enumerate() {
let pointer = format!("{prefix}/{index}");
if step.step_id().as_ref() == step_id {
return Some((step, pointer));
}
match step {
StepIR::If(nested) => {
if let Some(found) = find_step(&nested.then, &format!("{pointer}/then"), step_id) {
return Some(found);
}
if let Some(else_body) = nested.r#else.as_deref()
&& let Some(found) = find_step(else_body, &format!("{pointer}/else"), step_id)
{
return Some(found);
}
}
StepIR::Foreach(nested) => {
if let Some(found) = find_step(&nested.body, &format!("{pointer}/body"), step_id) {
return Some(found);
}
}
_ => {}
}
}
None
}