use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use crate::event_log::{AnyEventLog, SqliteEventLog};
use crate::redact::current_policy;
use crate::session_timeline::{
query_session_timeline, SessionTimelineQuery, SessionTimelineSnapshot,
};
use super::persistence::load_run_record_snapshot;
use super::time::parse_timestamp_ms;
use super::{build_run_view_with_event_log, RunRecord, RunView, RunViewUsage, ViewProducer};
mod join_evidence;
#[cfg(test)]
mod join_evidence_tests;
use join_evidence::{project_join_evidence, JoinEvidenceProjection};
pub const RUN_REPORT_SCHEMA: &str = "harn.run_report.v1";
pub const RUN_REPORT_SCHEMA_VERSION: u32 = 1;
const MAX_RUN_TREE_DEPTH: usize = 64;
const MAX_RUN_TREE_NODES: usize = 1024;
const MAX_RUN_TREE_BYTES: usize = 32 * 1024 * 1024;
#[derive(Clone, Debug, Default)]
pub struct RunReportRequest {
pub run_record_path: PathBuf,
pub events_db: Option<PathBuf>,
pub allowed_roots: Vec<PathBuf>,
pub source_root: Option<PathBuf>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
#[serde(default)]
pub struct RunReport {
pub schema: String,
pub schema_version: u32,
pub producer: ViewProducer,
pub projection: RunReportProjection,
pub root_run_id: String,
pub agents: Vec<RunReportAgent>,
pub delegations: Vec<RunReportDelegation>,
pub llm_calls: Vec<RunReportLlmCall>,
pub coordination: RunReportCoordination,
pub timelines: Vec<SessionTimelineSnapshot>,
pub sources: Vec<RunReportSource>,
pub checks: Vec<RunReportCheck>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct RunReportProjection {
pub id: String,
pub hash: String,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
#[serde(default)]
pub struct RunReportAgent {
pub agent_id: String,
pub worker_id: Option<String>,
pub run_id: Option<String>,
pub session_id: Option<String>,
pub parent_agent_id: Option<String>,
pub status: String,
pub task: String,
pub started_at: Option<String>,
pub finished_at: Option<String>,
pub duration_ms: Option<u64>,
pub usage: RunViewUsage,
pub visible_output: Option<String>,
pub execution: Option<RunReportExecution>,
pub capability_policy: Option<super::super::CapabilityPolicy>,
pub mutation_scope: Option<String>,
pub approval_policy: Option<super::super::ToolApprovalPolicy>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct RunReportExecution {
pub cwd: Option<String>,
pub project_root: Option<String>,
pub repo_path: Option<String>,
pub worktree_path: Option<String>,
pub branch: Option<String>,
pub environment_policy: crate::security::EnvironmentPolicyKind,
pub grants: Vec<crate::security::GrantReceipt>,
}
impl From<&super::RunExecutionRecord> for RunReportExecution {
fn from(execution: &super::RunExecutionRecord) -> Self {
Self {
cwd: execution.cwd.clone(),
project_root: execution.project_root.clone(),
repo_path: execution.repo_path.clone(),
worktree_path: execution.worktree_path.clone(),
branch: execution.branch.clone(),
environment_policy: execution.environment_policy,
grants: execution.grants.clone(),
}
}
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct RunReportDelegation {
pub parent_agent_id: String,
pub child_agent_id: String,
pub worker_id: String,
pub status: String,
pub parent_observed_status: String,
pub child_observed_status: Option<String>,
pub started_at: Option<String>,
pub finished_at: Option<String>,
pub forward_pointer: bool,
pub back_pointer: bool,
pub session_pointer_consistent: Option<bool>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
#[serde(default)]
pub struct RunReportLlmCall {
pub agent_id: String,
pub call_id: String,
pub provider: Option<String>,
pub model: Option<String>,
pub start_ms: u64,
pub duration_ms: u64,
pub ttft_ms: Option<u64>,
pub input_tokens: Option<i64>,
pub output_tokens: Option<i64>,
pub cache_read_tokens: Option<i64>,
pub cache_write_tokens: Option<i64>,
pub cost_usd: Option<f64>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct RunReportCoordination {
pub spawned: usize,
pub terminal: usize,
pub open: usize,
pub orphaned: usize,
pub unjoined: Option<usize>,
pub max_concurrent_children: Option<usize>,
pub observed_wait_ms: Option<u64>,
pub observed_join_ms: Option<u64>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct RunReportSource {
pub id: String,
pub agent_id: Option<String>,
pub kind: String,
pub path: Option<String>,
pub sha256: Option<String>,
pub status: String,
pub error: Option<String>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct RunReportCheck {
pub code: String,
pub severity: String,
pub status: String,
pub agent_id: Option<String>,
pub message: String,
}
#[derive(Debug)]
pub enum RunReportError {
Read(String),
EventLog(String),
Encode(String),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RunReportValidationError {
Schema(String),
Hash(String),
Encode(String),
}
#[derive(Debug)]
struct LoadedRunTree {
records: Vec<RunRecord>,
paths: Vec<PathBuf>,
hashes: Vec<String>,
}
impl std::fmt::Display for RunReportError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Read(message) | Self::EventLog(message) | Self::Encode(message) => {
formatter.write_str(message)
}
}
}
}
impl std::error::Error for RunReportError {}
impl std::fmt::Display for RunReportValidationError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Schema(message) | Self::Hash(message) | Self::Encode(message) => {
formatter.write_str(message)
}
}
}
}
impl std::error::Error for RunReportValidationError {}
pub fn validate_run_report(report: &RunReport) -> Result<(), RunReportValidationError> {
if report.schema != RUN_REPORT_SCHEMA || report.schema_version != RUN_REPORT_SCHEMA_VERSION {
return Err(RunReportValidationError::Schema(format!(
"expected {RUN_REPORT_SCHEMA} schema version {RUN_REPORT_SCHEMA_VERSION}, got {:?} version {}",
report.schema, report.schema_version
)));
}
let expected = run_report_projection_hash(report)?;
if report.projection.hash != expected {
return Err(RunReportValidationError::Hash(format!(
"run report projection hash mismatch: expected {expected}, got {:?}",
report.projection.hash
)));
}
Ok(())
}
pub fn run_report_projection_hash(report: &RunReport) -> Result<String, RunReportValidationError> {
let value = serde_json::to_value(report)
.map_err(|error| RunReportValidationError::Encode(error.to_string()))?;
Ok(run_report_projection_hash_value(&value))
}
fn run_report_projection_hash_value(value: &Value) -> String {
let mut value = value.clone();
value["projection"]["hash"] = Value::String(String::new());
let digest = Sha256::digest(crate::canonical_json::to_vec(&value));
format!("sha256:{}", hex::encode(digest))
}
pub async fn build_run_report(request: RunReportRequest) -> Result<RunReport, RunReportError> {
let allowed_roots = canonical_allowed_roots(&request.allowed_roots)?;
let root_path = checked_path(&request.run_record_path, &allowed_roots)?;
let source_root = request
.source_root
.as_deref()
.and_then(|path| path.canonicalize().ok());
let events_db_path = request
.events_db
.as_deref()
.map(|path| checked_path(path, &allowed_roots))
.transpose()?;
let event_log = match events_db_path.as_deref() {
Some(path) => Some(AnyEventLog::Sqlite(
SqliteEventLog::open_read_only(path.to_path_buf(), 16)
.map_err(|error| RunReportError::EventLog(error.to_string()))?,
)),
None => None,
};
let tree = tokio::task::spawn_blocking(move || load_run_tree(&root_path, &allowed_roots))
.await
.map_err(|error| RunReportError::Read(format!("load run tree task: {error}")))??;
let root_run_id = tree
.records
.first()
.map(|record| record.id.clone())
.unwrap_or_default();
let mut views = Vec::with_capacity(tree.records.len());
let mut timelines = Vec::new();
for (record, path) in tree.records.iter().zip(&tree.paths) {
let view = build_run_view_with_event_log(record, None::<String>, event_log.as_ref())
.await
.map_err(|error| RunReportError::EventLog(error.to_string()))?;
let query = SessionTimelineQuery {
session_id: view.run.session_id.clone(),
run_id: Some(record.id.clone()),
run_path: Some(path.to_string_lossy().into_owned()),
..SessionTimelineQuery::default()
};
let timeline = query_session_timeline(event_log.as_ref(), Some(record), query)
.await
.map_err(|error| RunReportError::EventLog(error.to_string()))?;
if event_log.is_some() || !timeline.nodes.is_empty() {
timelines.push(timeline);
}
views.push(view);
}
let report = assemble_report(
root_run_id,
&tree.records,
&tree.paths,
&tree.hashes,
&views,
timelines,
source_root.as_deref(),
events_db_path.as_deref(),
)?;
let mut value =
serde_json::to_value(&report).map_err(|error| RunReportError::Encode(error.to_string()))?;
value["projection"]["hash"] = Value::String(String::new());
current_policy().redact_json_in_place(&mut value);
value["projection"]["hash"] = Value::String(run_report_projection_hash_value(&value));
serde_json::from_value(value).map_err(|error| RunReportError::Encode(error.to_string()))
}
fn assemble_report(
root_run_id: String,
records: &[RunRecord],
record_paths: &[PathBuf],
record_hashes: &[String],
views: &[RunView],
timelines: Vec<SessionTimelineSnapshot>,
source_root: Option<&Path>,
events_db_path: Option<&Path>,
) -> Result<RunReport, RunReportError> {
let views_by_id: BTreeMap<_, _> = views
.iter()
.map(|view| (view.run.run_id.as_str(), view))
.collect();
let records_by_id: BTreeMap<_, _> = records
.iter()
.map(|record| (record.id.as_str(), record))
.collect();
if records_by_id.len() != records.len() {
return Err(RunReportError::Read(
"run tree contains duplicate run ids; lineage would be ambiguous".to_string(),
));
}
let records_by_path: BTreeMap<_, _> = record_paths
.iter()
.zip(records)
.map(|(path, record)| (path.clone(), record))
.collect();
let mut checks = Vec::new();
let mut agents = Vec::new();
let mut delegations = Vec::new();
let mut llm_calls = Vec::new();
let mut sources = Vec::new();
let mut forward_edges = BTreeSet::new();
for (((record, path), hash), view) in records
.iter()
.zip(record_paths)
.zip(record_hashes)
.zip(views)
{
agents.push(agent_from_view(view, None, record.policy.clone()));
sources.push(source_for_snapshot(
format!("run:{}", record.id),
"run_record",
path,
hash,
source_root,
Some(format!("run:{}", record.id)),
));
if let Some(observability) = &record.observability {
for pointer in &observability.transcript_pointers {
sources.push(RunReportSource {
id: format!("run:{}:{}", record.id, pointer.id),
agent_id: Some(format!("run:{}", record.id)),
kind: pointer.kind.clone(),
path: pointer
.path
.as_deref()
.map(|path| display_path(Path::new(path), source_root)),
sha256: pointer
.descriptor
.as_ref()
.map(|descriptor| descriptor.sha256.clone()),
status: pointer.verification_status.clone(),
error: pointer.verification_error.clone(),
});
}
}
for span in &record.trace_spans {
if span.kind != "llm_call" {
continue;
}
llm_calls.push(llm_call_from_span(&record.id, span));
}
for child in &record.child_runs {
let path_record = child
.run_path
.as_deref()
.map(PathBuf::from)
.map(|child_path| resolve_child_path(path, child_path))
.and_then(|child_path| child_path.canonicalize().ok())
.and_then(|child_path| records_by_path.get(&child_path).copied());
let id_record = child
.run_id
.as_deref()
.and_then(|id| records_by_id.get(id).copied());
let child_record = path_record.or(id_record);
let child_agent_id = child_record
.map(|record| format!("run:{}", record.id))
.or_else(|| child.run_id.as_ref().map(|run_id| format!("run:{run_id}")))
.unwrap_or_else(|| format!("worker:{}", child.worker_id));
let child_view =
child_record.and_then(|record| views_by_id.get(record.id.as_str()).copied());
let back_pointer = child_record
.and_then(|child_record| child_record.parent_run_id.as_deref())
== Some(record.id.as_str());
let session_pointer_consistent = session_pointers_consistent(view, child, child_view);
let child_observed_status = child_record.map(|record| record.status.clone());
let status = child_observed_status
.as_deref()
.filter(|status| !status.is_empty())
.unwrap_or(&child.status)
.to_string();
delegations.push(RunReportDelegation {
parent_agent_id: format!("run:{}", record.id),
child_agent_id: child_agent_id.clone(),
worker_id: child.worker_id.clone(),
status,
parent_observed_status: child.status.clone(),
child_observed_status: child_observed_status.clone(),
started_at: child_record
.and_then(|record| nonempty(&record.started_at))
.or_else(|| nonempty(&child.started_at)),
finished_at: child_record
.and_then(|record| record.finished_at.clone())
.or_else(|| child.finished_at.clone()),
forward_pointer: true,
back_pointer,
session_pointer_consistent,
});
if let Some(child_record) = child_record {
forward_edges.insert((record.id.clone(), child_record.id.clone()));
if child.run_id.is_none() {
checks.push(check(
"child_run_id_missing",
"warning",
&child_agent_id,
format!(
"worker {} was correlated by run_path because its forward run_id is missing",
child.worker_id
),
));
} else if child.run_id.as_deref() != Some(child_record.id.as_str()) {
checks.push(check(
"child_run_id_mismatch",
"error",
&child_agent_id,
format!(
"worker {} points to run id {:?}, but run_path contains {}",
child.worker_id, child.run_id, child_record.id
),
));
}
if let Some(agent) = agents
.iter_mut()
.find(|agent| agent.agent_id == child_agent_id)
{
agent.worker_id = Some(child.worker_id.clone());
agent.mutation_scope = child.mutation_scope.clone();
agent.approval_policy = child.approval_policy.clone();
}
}
if child_record.is_none() {
agents.push(agent_from_child(child, &record.id));
checks.push(check(
"child_run_missing",
"error",
&child_agent_id,
format!(
"worker {} has no readable child run record",
child.worker_id
),
));
} else if !back_pointer {
checks.push(check(
"child_back_pointer_mismatch",
"error",
&child_agent_id,
format!("child does not point back to parent run {}", record.id),
));
}
if child_observed_status
.as_deref()
.is_some_and(|status| !child.status.is_empty() && status != child.status)
{
checks.push(check(
"child_status_mismatch",
"warning",
&child_agent_id,
format!(
"parent recorded status {:?}, while the child run recorded {:?}",
child.status, child_observed_status
),
));
}
if session_pointer_consistent == Some(false) {
checks.push(check(
"child_session_pointer_mismatch",
"error",
&child_agent_id,
"child and parent disagree about parent session".to_string(),
));
}
}
}
if let Some(path) = events_db_path {
sources.push(RunReportSource {
id: "events:sqlite".to_string(),
agent_id: None,
kind: "event_log".to_string(),
path: Some(display_path(path, source_root)),
sha256: None,
status: "queried_read_only".to_string(),
error: Some(
"live SQLite evidence is cursor-scoped; no unstable whole-file hash was claimed"
.to_string(),
),
});
}
for record in records {
if let Some(parent_id) = record.parent_run_id.as_deref() {
if !records_by_id.contains_key(parent_id) {
checks.push(check(
"parent_run_missing",
"error",
&format!("run:{}", record.id),
format!("parent run {parent_id} is not present in the report"),
));
} else if !forward_edges.contains(&(parent_id.to_string(), record.id.clone())) {
delegations.push(RunReportDelegation {
parent_agent_id: format!("run:{parent_id}"),
child_agent_id: format!("run:{}", record.id),
back_pointer: true,
..RunReportDelegation::default()
});
checks.push(check(
"parent_forward_pointer_missing",
"error",
&format!("run:{}", record.id),
format!("parent run {parent_id} does not list this child"),
));
}
}
}
for timeline in &timelines {
if !timeline.coverage.truncated {
continue;
}
let agent_id = timeline
.query
.run_id
.as_deref()
.map(|run_id| format!("run:{run_id}"))
.unwrap_or_else(|| format!("run:{root_run_id}"));
let availability = timeline
.coverage
.available
.map(|available| format!(" of {available} available"))
.unwrap_or_else(|| ", with the total available count unknown".to_string());
checks.push(check(
"timeline_truncated",
"warning",
&agent_id,
format!(
"timeline returned {}{}; later evidence may be omitted, so absence must not be inferred",
timeline.coverage.returned, availability
),
));
}
let join_evidence = project_join_evidence(&delegations, &timelines, events_db_path.is_some());
checks.extend(join_evidence.checks.iter().cloned());
let coordination = coordination_summary(&delegations, &checks, &join_evidence);
if !delegations.is_empty() {
if coordination.max_concurrent_children.is_none() {
checks.push(RunReportCheck {
code: "coordination_intervals_incomplete".to_string(),
severity: "info".to_string(),
status: "unavailable".to_string(),
agent_id: Some(format!("run:{root_run_id}")),
message: "one or more child intervals lack a parseable start or finish timestamp, so peak concurrency is unknown".to_string(),
});
}
checks.push(RunReportCheck {
code: "coordination_timing_unavailable".to_string(),
severity: "info".to_string(),
status: "unavailable".to_string(),
agent_id: Some(format!("run:{root_run_id}")),
message: if join_evidence.complete {
"join receipts measure terminal-to-collection lag, but no canonical wait-start event exists, so parent wait and result-processing time remain unknown".to_string()
} else {
"canonical join evidence is missing, malformed, or truncated, so unjoined children, parent wait, and terminal-to-collection lag remain unknown".to_string()
},
});
}
agents.sort_by(|left, right| left.agent_id.cmp(&right.agent_id));
agents.dedup_by(|left, right| left.agent_id == right.agent_id);
delegations.sort_by(|left, right| {
(&left.parent_agent_id, &left.child_agent_id, &left.worker_id).cmp(&(
&right.parent_agent_id,
&right.child_agent_id,
&right.worker_id,
))
});
llm_calls.sort_by_key(|call| (call.start_ms, call.call_id.clone()));
sources.sort_by(|left, right| left.id.cmp(&right.id));
checks.sort_by(|left, right| (&left.code, &left.agent_id).cmp(&(&right.code, &right.agent_id)));
Ok(RunReport {
schema: RUN_REPORT_SCHEMA.to_string(),
schema_version: RUN_REPORT_SCHEMA_VERSION,
producer: ViewProducer::default(),
projection: RunReportProjection {
id: format!("run_report:{root_run_id}"),
hash: String::new(),
},
root_run_id,
agents,
delegations,
llm_calls,
coordination,
timelines,
sources,
checks,
})
}
fn load_run_tree(path: &Path, allowed_roots: &[PathBuf]) -> Result<LoadedRunTree, RunReportError> {
let mut tree = LoadedRunTree {
records: Vec::new(),
paths: Vec::new(),
hashes: Vec::new(),
};
let mut seen_paths = BTreeSet::new();
let mut pending = vec![(path.to_path_buf(), 0_usize)];
let mut total_bytes = 0_usize;
while let Some((candidate, depth)) = pending.pop() {
if depth > MAX_RUN_TREE_DEPTH {
return Err(RunReportError::Read(format!(
"run tree exceeds maximum depth {MAX_RUN_TREE_DEPTH}"
)));
}
let path = checked_path(&candidate, allowed_roots)?;
if !seen_paths.insert(path.clone()) {
continue;
}
if seen_paths.len() > MAX_RUN_TREE_NODES {
return Err(RunReportError::Read(format!(
"run tree exceeds maximum node count {MAX_RUN_TREE_NODES}"
)));
}
let (record, bytes) = load_run_record_snapshot(&path).map_err(|error| {
RunReportError::Read(format!("load run record {}: {error}", path.display()))
})?;
total_bytes = total_bytes.saturating_add(bytes.len());
if total_bytes > MAX_RUN_TREE_BYTES {
return Err(RunReportError::Read(format!(
"run tree exceeds maximum persisted size {MAX_RUN_TREE_BYTES} bytes"
)));
}
let mut child_paths = record
.child_runs
.iter()
.filter_map(|child| child.run_path.as_deref())
.map(PathBuf::from)
.map(|child| resolve_child_path(&path, child))
.filter(|child| child.exists())
.map(|child| (child, depth + 1))
.collect::<Vec<_>>();
child_paths.reverse();
pending.extend(child_paths);
tree.hashes
.push(format!("sha256:{}", hex::encode(Sha256::digest(&bytes))));
tree.records.push(record);
tree.paths.push(path);
}
Ok(tree)
}
fn resolve_child_path(parent_path: &Path, child: PathBuf) -> PathBuf {
if child.is_absolute() {
child
} else {
parent_path.parent().unwrap_or(Path::new(".")).join(child)
}
}
fn canonical_allowed_roots(roots: &[PathBuf]) -> Result<Vec<PathBuf>, RunReportError> {
roots
.iter()
.map(|root| {
root.canonicalize().map_err(|error| {
RunReportError::Read(format!("resolve allowed root {}: {error}", root.display()))
})
})
.collect()
}
fn checked_path(path: &Path, allowed_roots: &[PathBuf]) -> Result<PathBuf, RunReportError> {
let canonical = path
.canonicalize()
.map_err(|error| RunReportError::Read(format!("resolve {}: {error}", path.display())))?;
if !allowed_roots.is_empty() && !allowed_roots.iter().any(|root| canonical.starts_with(root)) {
return Err(RunReportError::Read(format!(
"path {} is outside the report's allowed roots",
path.display()
)));
}
Ok(canonical)
}
pub(crate) fn read_checked_run_report_bytes(
path: &Path,
allowed_roots: &[PathBuf],
) -> Result<Vec<u8>, RunReportError> {
let allowed_roots = canonical_allowed_roots(allowed_roots)?;
let path = checked_path(path, &allowed_roots)?;
std::fs::read(&path).map_err(|error| {
RunReportError::Read(format!("read run report {}: {error}", path.display()))
})
}
fn agent_from_view(
view: &RunView,
worker_id: Option<String>,
capability_policy: super::super::CapabilityPolicy,
) -> RunReportAgent {
RunReportAgent {
agent_id: format!("run:{}", view.run.run_id),
worker_id,
run_id: Some(view.run.run_id.clone()),
session_id: view.run.session_id.clone(),
parent_agent_id: view
.run
.parent_run_id
.as_ref()
.map(|id| format!("run:{id}")),
status: view.run.status.clone(),
task: view.run.task.clone(),
started_at: nonempty(&view.run.started_at),
finished_at: view.run.finished_at.clone(),
duration_ms: view.run.duration_ms,
usage: view.usage.clone(),
visible_output: view.visible_text.clone(),
execution: view
.metadata
.execution
.as_ref()
.map(RunReportExecution::from),
capability_policy: Some(capability_policy),
mutation_scope: None,
approval_policy: None,
}
}
fn session_pointers_consistent(
parent_view: &RunView,
child: &super::RunChildRecord,
child_view: Option<&RunView>,
) -> Option<bool> {
let mut comparisons = Vec::new();
if let (Some(actual), Some(declared)) = (
parent_view.run.session_id.as_deref(),
child.parent_session_id.as_deref(),
) {
comparisons.push(actual == declared);
}
if let (Some(actual), Some(child_view)) = (parent_view.run.session_id.as_deref(), child_view) {
if let Some(declared) = child_view.run.parent_session_id.as_deref() {
comparisons.push(actual == declared);
}
}
if let (Some(declared), Some(child_view)) = (child.session_id.as_deref(), child_view) {
if let Some(actual) = child_view.run.session_id.as_deref() {
comparisons.push(actual == declared);
}
}
(!comparisons.is_empty()).then(|| comparisons.into_iter().all(|matches| matches))
}
fn agent_from_child(child: &super::RunChildRecord, parent_run_id: &str) -> RunReportAgent {
let policy = current_policy();
RunReportAgent {
agent_id: child
.run_id
.as_ref()
.map(|run_id| format!("run:{run_id}"))
.unwrap_or_else(|| format!("worker:{}", child.worker_id)),
worker_id: Some(child.worker_id.clone()),
run_id: child.run_id.clone(),
session_id: child.session_id.clone(),
parent_agent_id: Some(format!("run:{parent_run_id}")),
status: child.status.clone(),
task: policy.redact_string(&child.task).into_owned(),
started_at: nonempty(&child.started_at),
finished_at: child.finished_at.clone(),
execution: child.execution.as_ref().map(RunReportExecution::from),
capability_policy: None,
mutation_scope: child.mutation_scope.clone(),
approval_policy: child.approval_policy.clone(),
..RunReportAgent::default()
}
}
fn llm_call_from_span(run_id: &str, span: &super::RunTraceSpanRecord) -> RunReportLlmCall {
let integer = |key: &str| span.metadata.get(key).and_then(Value::as_i64);
let number = |key: &str| span.metadata.get(key).and_then(Value::as_f64);
let text = |key: &str| {
span.metadata
.get(key)
.and_then(Value::as_str)
.map(str::to_string)
};
RunReportLlmCall {
agent_id: format!("run:{run_id}"),
call_id: format!("{}:{}", span.trace_id, span.span_id),
provider: text(crate::tracing::meta::PROVIDER),
model: text(crate::tracing::meta::MODEL),
start_ms: span.start_ms,
duration_ms: span.duration_ms,
ttft_ms: span.ttft_ms,
input_tokens: integer(crate::tracing::meta::INPUT_TOKENS),
output_tokens: integer(crate::tracing::meta::OUTPUT_TOKENS),
cache_read_tokens: integer(crate::tracing::meta::CACHE_READ_TOKENS),
cache_write_tokens: integer(crate::tracing::meta::CACHE_WRITE_TOKENS),
cost_usd: span
.cost_usd
.or_else(|| number(crate::tracing::meta::COST_USD)),
}
}
fn coordination_summary(
delegations: &[RunReportDelegation],
checks: &[RunReportCheck],
join_evidence: &JoinEvidenceProjection,
) -> RunReportCoordination {
let terminal = delegations
.iter()
.filter(|delegation| {
crate::agent_events::WorkerEvent::status_is_terminal(&delegation.status)
})
.count();
let intervals: Vec<(i128, i128)> = delegations
.iter()
.filter_map(|delegation| {
let start = parse_timestamp_ms(delegation.started_at.as_deref()?)?;
let finish = parse_timestamp_ms(delegation.finished_at.as_deref()?)?;
(finish >= start).then_some((start, finish))
})
.collect();
let mut points = intervals
.iter()
.flat_map(|(start, finish)| [(*start, 1_i32), (*finish, -1_i32)])
.collect::<Vec<_>>();
points.sort_by(|left, right| left.0.cmp(&right.0).then_with(|| right.1.cmp(&left.1)));
let max_concurrent_children = if delegations.is_empty() {
Some(0)
} else if intervals.len() == delegations.len() {
Some(
points
.into_iter()
.fold((0_i32, 0_i32), |(active, max), (_, delta)| {
let active = active + delta;
(active, max.max(active))
})
.1
.max(0) as usize,
)
} else {
None
};
RunReportCoordination {
spawned: delegations.len(),
terminal,
open: delegations.len().saturating_sub(terminal),
orphaned: checks
.iter()
.filter(|check| check.code == "parent_run_missing")
.count(),
unjoined: join_evidence.complete.then(|| {
delegations
.iter()
.filter(|delegation| {
crate::agent_events::WorkerEvent::status_is_terminal(&delegation.status)
&& !join_evidence.joined(delegation)
})
.count()
}),
max_concurrent_children,
observed_wait_ms: None,
observed_join_ms: if join_evidence.complete {
join_evidence.max_terminal_to_collection_ms
} else {
None
},
}
}
fn source_for_snapshot(
id: String,
kind: &str,
path: &Path,
sha256: &str,
source_root: Option<&Path>,
agent_id: Option<String>,
) -> RunReportSource {
RunReportSource {
id,
agent_id,
kind: kind.to_string(),
path: Some(display_path(path, source_root)),
sha256: Some(sha256.to_string()),
status: "verified".to_string(),
error: None,
}
}
fn display_path(path: &Path, source_root: Option<&Path>) -> String {
let normalized = canonicalize_with_missing_suffix(path).unwrap_or_else(|| path.to_path_buf());
source_root
.and_then(|root| normalized.strip_prefix(root).ok())
.unwrap_or(&normalized)
.to_string_lossy()
.into_owned()
}
fn canonicalize_with_missing_suffix(path: &Path) -> Option<PathBuf> {
let mut cursor = path;
let mut suffix = Vec::new();
while !cursor.exists() {
suffix.push(cursor.file_name()?.to_os_string());
cursor = cursor.parent()?;
}
let mut normalized = cursor.canonicalize().ok()?;
for component in suffix.into_iter().rev() {
normalized.push(component);
}
Some(normalized)
}
fn check(code: &str, severity: &str, agent_id: &str, message: String) -> RunReportCheck {
RunReportCheck {
code: code.to_string(),
severity: severity.to_string(),
status: "failed".to_string(),
agent_id: Some(agent_id.to_string()),
message,
}
}
fn nonempty(value: &str) -> Option<String> {
(!value.is_empty()).then(|| value.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::orchestration::{save_run_record, RunChildRecord, RunTraceSpanRecord};
use std::fs;
fn temp_dir(label: &str) -> PathBuf {
let path = std::env::temp_dir().join(format!("harn-{label}-{}", uuid::Uuid::now_v7()));
fs::create_dir_all(&path).unwrap();
path
}
#[tokio::test]
async fn report_correlates_parent_and_child_bidirectionally() {
let dir = temp_dir("run-report-lineage");
let parent_path = dir.join("parent.json");
let child_path = dir.join("child.json");
let child = RunRecord {
type_name: "workflow_run".to_string(),
id: "child".to_string(),
workflow_id: "child-workflow".to_string(),
task: "child task".to_string(),
status: "completed".to_string(),
started_at: "2026-08-02T10:00:01Z".to_string(),
finished_at: Some("2026-08-02T10:00:03Z".to_string()),
parent_run_id: Some("parent".to_string()),
root_run_id: Some("parent".to_string()),
metadata: BTreeMap::from([
(
"session_id".to_string(),
Value::String("child-session".to_string()),
),
(
"parent_session_id".to_string(),
Value::String("parent-session".to_string()),
),
]),
..RunRecord::default()
};
let parent = RunRecord {
type_name: "workflow_run".to_string(),
id: "parent".to_string(),
workflow_id: "parent-workflow".to_string(),
task: "parent task".to_string(),
status: "completed".to_string(),
started_at: "2026-08-02T10:00:00Z".to_string(),
finished_at: Some("2026-08-02T10:00:04Z".to_string()),
root_run_id: Some("parent".to_string()),
child_runs: vec![RunChildRecord {
worker_id: "worker-1".to_string(),
worker_name: "child".to_string(),
task: "child task".to_string(),
status: "completed".to_string(),
started_at: "2026-08-02T10:00:01Z".to_string(),
finished_at: Some("2026-08-02T10:00:03Z".to_string()),
session_id: Some("child-session".to_string()),
parent_session_id: Some("parent-session".to_string()),
run_id: Some("child".to_string()),
run_path: Some(child_path.to_string_lossy().into_owned()),
..RunChildRecord::default()
}],
metadata: BTreeMap::from([(
"session_id".to_string(),
Value::String("parent-session".to_string()),
)]),
..RunRecord::default()
};
save_run_record(&child, Some(child_path.to_str().unwrap())).unwrap();
save_run_record(&parent, Some(parent_path.to_str().unwrap())).unwrap();
let report = build_run_report(RunReportRequest {
run_record_path: parent_path,
allowed_roots: vec![dir.clone()],
source_root: Some(dir.clone()),
..RunReportRequest::default()
})
.await
.unwrap();
assert_eq!(report.agents.len(), 2);
assert_eq!(report.delegations.len(), 1);
assert!(report.delegations[0].forward_pointer);
assert!(report.delegations[0].back_pointer);
assert_eq!(report.delegations[0].session_pointer_consistent, Some(true));
assert_eq!(report.coordination.max_concurrent_children, Some(1));
assert_eq!(report.coordination.unjoined, None);
assert!(report
.checks
.iter()
.any(|check| check.code == "coordination_timing_unavailable"));
assert!(report.projection.hash.starts_with("sha256:"));
assert_eq!(
report
.sources
.iter()
.map(|source| source.id.as_str())
.collect::<BTreeSet<_>>()
.len(),
report.sources.len()
);
assert!(
report.sources.iter().all(|source| {
source
.path
.as_deref()
.is_none_or(|path| !Path::new(path).is_absolute())
}),
"sources={:?}",
report.sources
);
fs::remove_dir_all(dir).unwrap();
}
#[tokio::test]
async fn report_flags_missing_child_record_without_inventing_a_join() {
let dir = temp_dir("run-report-missing-child");
let parent_path = dir.join("parent.json");
let parent = RunRecord {
type_name: "workflow_run".to_string(),
id: "parent".to_string(),
workflow_id: "parent-workflow".to_string(),
status: "completed".to_string(),
child_runs: vec![RunChildRecord {
worker_id: "worker-1".to_string(),
worker_name: "child".to_string(),
status: "running".to_string(),
run_id: Some("missing-child".to_string()),
run_path: Some(dir.join("missing.json").to_string_lossy().into_owned()),
..RunChildRecord::default()
}],
..RunRecord::default()
};
save_run_record(&parent, Some(parent_path.to_str().unwrap())).unwrap();
let report = build_run_report(RunReportRequest {
run_record_path: parent_path,
allowed_roots: vec![dir.clone()],
source_root: Some(dir.clone()),
..RunReportRequest::default()
})
.await
.unwrap();
assert_eq!(report.coordination.open, 1);
assert_eq!(report.coordination.unjoined, None);
assert!(report
.agents
.iter()
.any(|agent| agent.agent_id == "run:missing-child"));
assert!(report
.checks
.iter()
.any(|check| check.code == "child_run_missing"));
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn suspended_child_is_open_until_resumed_or_stopped() {
let coordination = coordination_summary(
&[RunReportDelegation {
status: "suspended".to_string(),
..RunReportDelegation::default()
}],
&[],
&JoinEvidenceProjection::default(),
);
assert_eq!(coordination.terminal, 0);
assert_eq!(coordination.open, 1);
}
#[tokio::test]
async fn report_recovers_missing_forward_run_id_from_child_path() {
let dir = temp_dir("run-report-path-lineage");
let parent_path = dir.join("parent.json");
let child_path = dir.join("child.json");
let child = RunRecord {
type_name: "workflow_run".to_string(),
id: "child".to_string(),
workflow_id: "child-workflow".to_string(),
status: "completed".to_string(),
parent_run_id: Some("parent".to_string()),
started_at: uuid::Uuid::now_v7().to_string(),
finished_at: Some(uuid::Uuid::now_v7().to_string()),
..RunRecord::default()
};
let parent = RunRecord {
type_name: "workflow_run".to_string(),
id: "parent".to_string(),
workflow_id: "parent-workflow".to_string(),
status: "completed".to_string(),
child_runs: vec![RunChildRecord {
worker_id: "worker-1".to_string(),
status: "running".to_string(),
run_id: None,
run_path: Some(child_path.to_string_lossy().into_owned()),
started_at: child.started_at.clone(),
finished_at: child.finished_at.clone(),
..RunChildRecord::default()
}],
..RunRecord::default()
};
save_run_record(&child, Some(child_path.to_str().unwrap())).unwrap();
save_run_record(&parent, Some(parent_path.to_str().unwrap())).unwrap();
let report = build_run_report(RunReportRequest {
run_record_path: parent_path,
allowed_roots: vec![dir.clone()],
source_root: Some(dir.clone()),
..RunReportRequest::default()
})
.await
.unwrap();
assert_eq!(report.agents.len(), 2);
assert_eq!(report.delegations.len(), 1);
assert_eq!(report.delegations[0].child_agent_id, "run:child");
assert_eq!(report.delegations[0].status, "completed");
assert_eq!(report.delegations[0].parent_observed_status, "running");
assert_eq!(report.coordination.terminal, 1);
assert_eq!(report.coordination.max_concurrent_children, Some(1));
assert!(report
.checks
.iter()
.any(|check| check.code == "child_run_id_missing"));
assert!(report
.checks
.iter()
.any(|check| check.code == "child_status_mismatch"));
fs::remove_dir_all(dir).unwrap();
}
#[tokio::test]
async fn report_flags_timeline_truncation_before_a_late_llm_call() {
let dir = temp_dir("run-report-timeline-truncation");
let run_path = dir.join("run.json");
let mut trace_spans = (1..=1025)
.map(|span_id| RunTraceSpanRecord {
trace_id: "trace-root".to_string(),
span_id,
kind: "import".to_string(),
name: format!("import-{span_id}"),
start_ms: span_id,
duration_ms: 1,
..RunTraceSpanRecord::default()
})
.collect::<Vec<_>>();
trace_spans.push(RunTraceSpanRecord {
trace_id: "trace-root".to_string(),
span_id: 1026,
kind: "llm_call".to_string(),
name: "late-llm-call".to_string(),
start_ms: 1026,
duration_ms: 2,
..RunTraceSpanRecord::default()
});
let run = RunRecord {
type_name: "workflow_run".to_string(),
id: "root".to_string(),
status: "completed".to_string(),
trace_spans,
..RunRecord::default()
};
save_run_record(&run, Some(run_path.to_str().unwrap())).unwrap();
let report = build_run_report(RunReportRequest {
run_record_path: run_path,
allowed_roots: vec![dir.clone()],
source_root: Some(dir.clone()),
..RunReportRequest::default()
})
.await
.unwrap();
assert_eq!(report.timelines.len(), 1);
let timeline = &report.timelines[0];
assert_eq!(timeline.coverage.returned, 1024);
assert_eq!(timeline.coverage.available, Some(1026));
assert!(timeline.coverage.truncated);
assert!(!timeline.nodes.iter().any(|node| node.kind == "llm_call"));
assert_eq!(report.llm_calls.len(), 1);
let check = report
.checks
.iter()
.find(|check| check.code == "timeline_truncated")
.expect("explicit truncation check");
assert_eq!(check.severity, "warning");
assert!(check.message.contains("1024 of 1026 available"));
assert!(check.message.contains("absence must not be inferred"));
fs::remove_dir_all(dir).unwrap();
}
#[tokio::test]
async fn report_redacts_returned_projection_and_hash_is_reproducible() {
let dir = temp_dir("run-report-redaction");
let run_path = dir.join("run.json");
let secret = "sk-proj-test-abcdefghijklmnopqrstuvwxyz123456";
let run = RunRecord {
type_name: "workflow_run".to_string(),
id: "root".to_string(),
workflow_id: "workflow".to_string(),
task: format!("do not expose {secret}"),
status: "completed".to_string(),
transcript: Some(serde_json::json!({
"events": [{
"kind": "message",
"role": "assistant",
"visibility": "public",
"blocks": [
{"type": "output_text", "text": format!("safe answer {secret}"), "visibility": "public"},
{"type": "reasoning", "text": "private chain of thought", "visibility": "private"}
]
}]
})),
..RunRecord::default()
};
fs::write(&run_path, serde_json::to_vec(&run).unwrap()).unwrap();
let report = build_run_report(RunReportRequest {
run_record_path: run_path,
allowed_roots: vec![dir.clone()],
source_root: Some(dir.clone()),
..RunReportRequest::default()
})
.await
.unwrap();
let rendered = serde_json::to_string(&report).unwrap();
assert!(!rendered.contains(secret));
let visible = report.agents[0]
.visible_output
.as_deref()
.expect("public transcript output");
assert!(visible.starts_with("safe answer "));
assert!(visible.contains("<redacted:openai_key:"));
assert!(!rendered.contains("private chain of thought"));
let expected_hash = report.projection.hash.clone();
let mut value = serde_json::to_value(&report).unwrap();
value["projection"]["hash"] = Value::String(String::new());
let actual_hash = format!(
"sha256:{}",
hex::encode(Sha256::digest(crate::canonical_json::to_vec(&value)))
);
assert_eq!(actual_hash, expected_hash);
fs::remove_dir_all(dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn snapshot_report_does_not_follow_implicit_sidecar_symlinks() {
let dir = temp_dir("run-report-sidecar-symlink");
let outside = temp_dir("run-report-sidecar-outside");
let run_path = dir.join("root.json");
let run = RunRecord {
type_name: "workflow_run".to_string(),
id: "root".to_string(),
workflow_id: "workflow".to_string(),
status: "completed".to_string(),
..RunRecord::default()
};
fs::write(&run_path, serde_json::to_vec(&run).unwrap()).unwrap();
fs::write(
outside.join("llm_transcript.jsonl"),
"{\"type\":\"daemon_event\",\"secret\":\"outside\"}\n",
)
.unwrap();
std::os::unix::fs::symlink(&outside, dir.join("root-llm")).unwrap();
let report = build_run_report(RunReportRequest {
run_record_path: run_path,
allowed_roots: vec![dir.clone()],
source_root: Some(dir.clone()),
..RunReportRequest::default()
})
.await
.unwrap();
assert_eq!(report.sources.len(), 1);
assert_eq!(report.sources[0].kind, "run_record");
fs::remove_dir_all(dir).unwrap();
fs::remove_dir_all(outside).unwrap();
}
}