use std::collections::HashMap;
use std::path::Path;
use crate::event;
#[cfg(test)]
use crate::event_log::reader::parse_json_lines;
use crate::event_log::reader::read_event_envelopes;
use crate::message::{Message, MessagePart};
use crate::nodegraph;
use crate::provider;
use crate::session::SessionOpenError;
#[derive(Debug, Clone)]
pub enum TranscriptEntry {
Message {
message: Message,
flow_run_id: Option<String>,
},
ToolTiming {
tool_use_id: String,
elapsed_ms: u64,
},
ActivitySummary {
turn: crate::activity::ActivitySummary,
session: crate::activity::ActivitySummary,
turn_files: Vec<String>,
session_files: Vec<String>,
},
CompactionSummary {
range_start: usize,
range_end: usize,
compacted_count: usize,
before_tokens: u64,
after_tokens: u64,
summary: String,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
DiffPreview {
tool_use_id: Option<String>,
title: String,
old_content: Option<String>,
new_content: Option<String>,
unified_diff: Option<String>,
},
FileEditApplied {
turn_id: Option<String>,
flow_run_id: Option<String>,
tool_use_id: Option<String>,
tool_name: String,
path: String,
metrics: crate::activity::EditMetrics,
},
FlowGraph {
run_id: String,
flow_name: String,
graph: nodegraph::FlowGraph,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
FlowStart {
run_id: String,
flow_name: String,
parent_run_id: Option<String>,
parent_node_id: Option<String>,
spawned: bool,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
FlowNodeStart {
run_id: String,
node_id: String,
kind: nodegraph::NodeKind,
label: String,
parent_node_id: Option<String>,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
FlowNodeEnd {
run_id: String,
node_id: String,
status: event::FlowNodeStatus,
output_preview: Option<String>,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
ToolNode {
run_id: String,
parent_node_id: String,
tool_use_id: String,
tool_name: String,
args_preview: String,
call_intent: Option<crate::message::ToolCallIntent>,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
FlowDone {
run_id: String,
ok: bool,
cancelled: bool,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
LlmCall {
model: String,
provider: String,
context_call_purpose: crate::context_plan::ContextCallPurpose,
context_call_scope: crate::context_plan::ContextCallScope,
usage: provider::TokenUsage,
wallclock_ms: u64,
ttft_ms: Option<u64>,
tokens_per_second: Option<f64>,
run_id: Option<event::FlowRunId>,
node_id: Option<String>,
ts: Option<chrono::DateTime<chrono::Utc>>,
},
PermissionRequest {
identity: crate::workflow::WorkflowPermissionIdentity,
payload: Box<crate::permission_audit::PermissionRequestAudit>,
state: crate::workflow::WorkflowPermissionState,
},
PermissionGroup {
payload: crate::permission_audit::PermissionGroupAudit,
resolved: bool,
},
TerminalFinalState {
handle: String,
screen: crate::tools::term::TerminalScreen,
},
MermaidDiagram {
source: String,
},
}
pub fn replay_messages_from(path: &Path) -> Result<Vec<Message>, SessionOpenError> {
Ok(replay_messages_with_seq(path)?
.into_iter()
.map(|(_, msg)| msg)
.collect())
}
pub fn replay_messages_with_seq(path: &Path) -> Result<Vec<(u64, Message)>, SessionOpenError> {
let envelopes = read_event_envelopes(path)?;
Ok(envelopes.as_slice().to_messages_with_seq())
}
pub fn replay_all_messages_with_seq(path: &Path) -> Result<Vec<(u64, Message)>, SessionOpenError> {
let envelopes = read_event_envelopes(path)?;
let spawned_flow_ids = spawned_flow_ids(&envelopes);
let mut messages = Vec::new();
let mut positions = HashMap::new();
for env in &envelopes {
match &env.event {
crate::event::Event::UserMsg {
message,
flow_run_id,
..
}
| crate::event::Event::AssistantMsg {
message,
flow_run_id,
..
}
| crate::event::Event::ToolResultMsg {
message,
flow_run_id,
..
} if message_belongs_to_root(flow_run_id.as_ref(), &spawned_flow_ids) => {
positions.insert(env.seq, messages.len());
messages.push((env.seq, message.clone()));
}
crate::event::Event::SystemMsg {
message,
flow_run_id,
..
} if message_belongs_to_root(flow_run_id.as_ref(), &spawned_flow_ids) => {
positions.insert(env.seq, messages.len());
messages.push((env.seq, message.clone()));
}
crate::event::Event::AttachmentDegraded {
message_seq,
part_index,
file_basename,
reason,
..
} => {
apply_attachment_degradation(
&mut messages,
&positions,
*message_seq,
*part_index,
file_basename,
reason,
);
}
_ => {}
}
}
Ok(messages)
}
#[derive(Debug, Clone)]
pub struct AttachmentPatch {
part_index: usize,
file_basename: String,
reason: String,
}
#[cfg(test)]
pub fn parse_ts(v: &serde_json::Value) -> Option<chrono::DateTime<chrono::Utc>> {
v.get("ts")?
.as_str()
.and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
.map(|dt| dt.with_timezone(&chrono::Utc))
}
#[cfg(test)]
pub fn parse_context_compact_event(v: &serde_json::Value) -> Option<CompactReplayEvent> {
if v["type"].as_str() != Some("context_compact") {
return None;
}
Some(CompactReplayEvent {
range_start: v["compacted_range_start"].as_u64().unwrap_or(0) as usize,
range_end: v["compacted_range_end"].as_u64().unwrap_or(0) as usize,
replacement_msg_seq: v["replacement_msg_seq"].as_u64(),
})
}
#[cfg(test)]
fn raw_event_belongs_to_root(
value: &serde_json::Value,
spawned_flow_ids: &std::collections::HashSet<crate::event::FlowRunId>,
) -> bool {
value["flow_run_id"]
.as_str()
.and_then(|raw| uuid::Uuid::parse_str(raw).ok())
.map(crate::event::FlowRunId)
.is_none_or(|run_id| !spawned_flow_ids.contains(&run_id))
}
#[cfg(test)]
#[derive(Debug, Clone)]
pub struct CompactReplayEvent {
range_start: usize,
range_end: usize,
replacement_msg_seq: Option<u64>,
}
#[derive(Clone, Copy)]
struct TranscriptMessageSlot {
seq: u64,
output_index: usize,
}
fn push_transcript_message(
out: &mut Vec<TranscriptEntry>,
messages: &mut Vec<TranscriptMessageSlot>,
positions: &mut HashMap<u64, usize>,
seq: u64,
entry: TranscriptEntry,
) {
positions.insert(seq, messages.len());
messages.push(TranscriptMessageSlot {
seq,
output_index: out.len(),
});
out.push(entry);
}
fn compact_transcript_messages(
out: &mut Vec<TranscriptEntry>,
messages: &mut Vec<TranscriptMessageSlot>,
positions: &mut HashMap<u64, usize>,
range_start: usize,
range_end: usize,
replacement_seq: u64,
) -> bool {
if range_start > range_end || range_end >= messages.len() {
return false;
}
let Some(replacement_position) = positions.get(&replacement_seq).copied() else {
return false;
};
let replacement_output_index = messages[replacement_position].output_index;
let mut replacement_entry = Some(out[replacement_output_index].clone());
let insertion_output_index = messages[range_start].output_index;
let removed_output_indices = messages[range_start..=range_end]
.iter()
.map(|slot| slot.output_index)
.chain(std::iter::once(replacement_output_index))
.collect::<std::collections::HashSet<_>>();
let old_len = out.len();
let mut old_to_new = vec![None; old_len];
let mut replacement_new_index = None;
let mut compacted = Vec::with_capacity(
old_len
.saturating_sub(removed_output_indices.len())
.saturating_add(1),
);
for (old_index, entry) in out.drain(..).enumerate() {
if old_index == insertion_output_index {
replacement_new_index = Some(compacted.len());
compacted.push(
replacement_entry
.take()
.expect("replacement inserted exactly once"),
);
}
if removed_output_indices.contains(&old_index) {
continue;
}
old_to_new[old_index] = Some(compacted.len());
compacted.push(entry);
}
let replacement_new_index = replacement_new_index.expect("message slot belongs to output");
*out = compacted;
let mut compacted_messages = Vec::with_capacity(
messages
.len()
.saturating_sub(range_end - range_start)
.saturating_sub(usize::from(
replacement_position < range_start || replacement_position > range_end,
)),
);
for (message_index, slot) in messages.iter().copied().enumerate() {
if message_index == range_start {
compacted_messages.push(TranscriptMessageSlot {
seq: replacement_seq,
output_index: replacement_new_index,
});
}
if (range_start..=range_end).contains(&message_index)
|| message_index == replacement_position
{
continue;
}
compacted_messages.push(TranscriptMessageSlot {
seq: slot.seq,
output_index: old_to_new[slot.output_index]
.expect("retained message has a retained output entry"),
});
}
*messages = compacted_messages;
positions.clear();
positions.extend(
messages
.iter()
.enumerate()
.map(|(index, slot)| (slot.seq, index)),
);
true
}
#[cfg(test)]
pub fn collect_attachment_patches(
values: &[serde_json::Value],
) -> HashMap<u64, Vec<AttachmentPatch>> {
let mut map: HashMap<u64, Vec<AttachmentPatch>> = HashMap::new();
for v in values {
if v["type"].as_str() == Some("attachment_degraded") {
let Some(msg_seq) = v["message_seq"].as_u64() else {
continue;
};
let Some(part_index) = v["part_index"].as_u64() else {
continue;
};
let file_basename = v["file_basename"].as_str().unwrap_or("").to_string();
let reason = v["reason"].as_str().unwrap_or("degraded").to_string();
map.entry(msg_seq).or_default().push(AttachmentPatch {
part_index: part_index as usize,
file_basename,
reason,
});
}
}
map
}
pub fn apply_attachment_patches(msg: &mut Message, patches: &[AttachmentPatch]) {
for p in patches {
if let Some(part) = msg.parts.get_mut(p.part_index) {
*part = MessagePart::Text {
text: format!(
"[attachment unavailable: {} — {}]",
p.file_basename, p.reason
),
};
}
}
}
fn legacy_permission_payload(
run_id: event::FlowRunId,
tool_use_id: String,
tool_name: &str,
reason: Option<&str>,
actor_label: &str,
at: chrono::DateTime<chrono::Utc>,
) -> crate::permission_audit::PermissionRequestAudit {
use crate::permission_audit::{
PermissionAuditTarget, PermissionPolicyReference, PermissionProjectionActor,
PermissionProvenanceSummary, PermissionRequestAudit,
};
PermissionRequestAudit {
request_id: None,
revision: 0,
session_id: "legacy:unknown".into(),
requesting_run_id: run_id.clone(),
parent_run_id: None,
root_run_id: run_id,
tool_use_id,
tool: if tool_name.is_empty() {
"unknown legacy tool".into()
} else {
tool_name.into()
},
call_intent: None,
tier: crate::tool::Tier::Zero,
execution_boundary: Default::default(),
provenance: PermissionProvenanceSummary {
cwd: None,
path: None,
path_origin: Some("legacy_unknown".into()),
workspace_id: None,
workspace_root: None,
repository_root: None,
network: false,
risks: Default::default(),
targets: Vec::new(),
},
target: PermissionAuditTarget::User,
group_ids: Vec::new(),
policy: PermissionPolicyReference {
snapshot_id: "legacy:unknown".into(),
rule_id: "legacy approval event; policy unavailable".into(),
},
escalation_path: Vec::new(),
decision_id: None,
actor: Some(PermissionProjectionActor::UnknownLegacy {
label: actor_label.into(),
}),
scope: None,
reason: reason.map(String::from),
at,
}
}
pub fn replay_transcript_from(path: &Path) -> Result<Vec<TranscriptEntry>, SessionOpenError> {
let mut entries = Vec::new();
let mut observer = |entry| entries.push(entry);
crate::event_log::replay::SessionReplay::from_path(path, Some(&mut observer))?;
Ok(entries)
}
#[cfg(test)]
fn replay_transcript_from_raw(path: &Path) -> Result<Vec<TranscriptEntry>, SessionOpenError> {
let text = match std::fs::read_to_string(path) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(e) => {
return Err(SessionOpenError::Replay {
path: path.to_path_buf(),
source: e,
});
}
};
let values = parse_json_lines(&text);
let mut known_flow_ids = std::collections::HashSet::new();
let mut flow_children =
std::collections::HashMap::<crate::event::FlowRunId, Vec<crate::event::FlowRunId>>::new();
let mut spawned_flow_ids = std::collections::HashSet::new();
for value in &values {
if value["type"].as_str() != Some("flow_start") {
continue;
}
let Some(raw_run_id) = value["run_id"].as_str() else {
continue;
};
let Ok(run_id) = uuid::Uuid::parse_str(raw_run_id) else {
continue;
};
let run_id = crate::event::FlowRunId(run_id);
let parent = value["parent_run_id"]
.as_str()
.and_then(|raw| uuid::Uuid::parse_str(raw).ok())
.map(crate::event::FlowRunId);
known_flow_ids.insert(run_id.clone());
if let Some(parent) = parent {
flow_children
.entry(parent)
.or_default()
.push(run_id.clone());
}
if value["spawned"].as_bool().unwrap_or(false) {
spawned_flow_ids.insert(run_id);
}
}
let mut queue = std::collections::VecDeque::from_iter(spawned_flow_ids.iter().cloned());
while let Some(parent) = queue.pop_front() {
if let Some(descendants) = flow_children.get(&parent) {
for descendant in descendants {
if spawned_flow_ids.insert(descendant.clone()) {
queue.push_back(descendant.clone());
}
}
}
}
let mut tool_started_at = HashMap::new();
for value in &values {
if value["type"].as_str() != Some("assistant_msg") {
continue;
}
let Some(ts) = parse_ts(value) else {
continue;
};
let Some(message) = value
.get("message")
.and_then(|message| serde_json::from_value::<Message>(message.clone()).ok())
else {
continue;
};
for part in message.parts {
if let MessagePart::ToolUse { id, .. } = part {
tool_started_at.insert(id, ts);
}
}
}
let patches = collect_attachment_patches(&values);
let mut out = Vec::new();
let mut messages = Vec::new();
let mut message_positions = HashMap::new();
let mut pending_permissions: std::collections::BTreeMap<
crate::workflow::WorkflowPermissionIdentity,
crate::permission_audit::PermissionRequestAudit,
> = std::collections::BTreeMap::new();
let mut legacy_pending: std::collections::HashMap<
(String, String),
Vec<crate::workflow::WorkflowPermissionIdentity>,
> = std::collections::HashMap::new();
let mut canonical_permissions = std::collections::HashSet::new();
let mut session_activity = crate::activity::ActivityAccumulator::default();
let mut turn_activity = crate::activity::ActivityAccumulator::default();
for v in &values {
let ty = v["type"].as_str().unwrap_or("");
if let Ok(envelope) = serde_json::from_value::<crate::event::EventEnvelope>(v.clone()) {
if matches!(&envelope.event, crate::event::Event::TurnStart { .. }) {
turn_activity = crate::activity::ActivityAccumulator::default();
} else {
session_activity.observe(&envelope.event);
turn_activity.observe(&envelope.event);
}
}
match ty {
"user_msg" | "assistant_msg" | "tool_result_msg" | "system_msg" => {
if let Some(m) = v.get("message")
&& let Ok(mut msg) = serde_json::from_value::<Message>(m.clone())
{
let seq = v["seq"].as_u64().unwrap_or(0);
let belongs_to_root = raw_event_belongs_to_root(v, &spawned_flow_ids);
if let Some(ps) = patches.get(&seq) {
apply_attachment_patches(&mut msg, ps);
}
let flow_run_id = v["flow_run_id"].as_str().and_then(|raw| {
let run_id = uuid::Uuid::parse_str(raw).ok()?;
let run_id = crate::event::FlowRunId(run_id);
if known_flow_ids.contains(&run_id) && !spawned_flow_ids.contains(&run_id) {
None
} else {
Some(raw.to_string())
}
});
let timing_ids = if ty == "tool_result_msg" {
msg.parts
.iter()
.filter_map(|part| match part {
MessagePart::ToolResult { tool_use_id, .. } => {
Some(tool_use_id.clone())
}
_ => None,
})
.collect::<Vec<_>>()
} else {
Vec::new()
};
let entry = TranscriptEntry::Message {
message: msg,
flow_run_id,
};
if belongs_to_root {
push_transcript_message(
&mut out,
&mut messages,
&mut message_positions,
seq,
entry,
);
} else {
out.push(entry);
}
if let Some(finished_at) = parse_ts(v) {
for tool_use_id in timing_ids {
if let Some(started_at) = tool_started_at.get(&tool_use_id) {
let elapsed_ms = finished_at
.signed_duration_since(*started_at)
.num_milliseconds()
.max(0) as u64;
out.push(TranscriptEntry::ToolTiming {
tool_use_id,
elapsed_ms,
});
}
}
}
}
}
"turn_start" => {}
"turn_end" => out.push(TranscriptEntry::ActivitySummary {
turn: turn_activity.summary(),
session: session_activity.summary(),
turn_files: turn_activity.file_paths(),
session_files: session_activity.file_paths(),
}),
"context_compact" => {
if !raw_event_belongs_to_root(v, &spawned_flow_ids) {
continue;
}
let Some(event) = parse_context_compact_event(v) else {
continue;
};
let Some(replacement_seq) = event.replacement_msg_seq else {
continue;
};
compact_transcript_messages(
&mut out,
&mut messages,
&mut message_positions,
event.range_start,
event.range_end,
replacement_seq,
);
}
"compaction_summary" => {
if !raw_event_belongs_to_root(v, &spawned_flow_ids) {
continue;
}
out.push(TranscriptEntry::CompactionSummary {
range_start: v["range_start"].as_u64().unwrap_or(0) as usize,
range_end: v["range_end"].as_u64().unwrap_or(0) as usize,
compacted_count: v["compacted_count"].as_u64().unwrap_or(0) as usize,
before_tokens: v["before_tokens"].as_u64().unwrap_or(0),
after_tokens: v["after_tokens"].as_u64().unwrap_or(0),
summary: v["summary"].as_str().unwrap_or("").to_string(),
ts: parse_ts(v),
});
}
"diff_preview" => {
out.push(TranscriptEntry::DiffPreview {
tool_use_id: v["tool_use_id"].as_str().map(String::from),
title: v["title"].as_str().unwrap_or("").to_string(),
old_content: v["old_content"].as_str().map(String::from),
new_content: v["new_content"].as_str().map(String::from),
unified_diff: v["unified_diff"].as_str().map(String::from),
});
}
"flow_graph" => {
let run_id = v["run_id"].as_str().unwrap_or("").to_string();
let flow_name = v
.get("graph")
.and_then(|g| g["flow_name"].as_str())
.unwrap_or("")
.to_string();
let ts = parse_ts(v);
if let Some(g) = v.get("graph")
&& let Ok(graph) = serde_json::from_value::<nodegraph::FlowGraph>(g.clone())
{
out.push(TranscriptEntry::FlowGraph {
run_id,
flow_name,
graph,
ts,
});
}
}
"flow_start" => {
let run_id = v["run_id"].as_str().unwrap_or("").to_string();
let flow_name = v["flow_name"].as_str().unwrap_or("").to_string();
let parent_run_id = v["parent_run_id"].as_str().map(String::from);
let parent_node_id = v["parent_node_id"].as_str().map(String::from);
let spawned = v.get("spawned").and_then(|s| s.as_bool()).unwrap_or(false);
let ts = parse_ts(v);
out.push(TranscriptEntry::FlowStart {
run_id,
flow_name,
parent_run_id,
parent_node_id,
spawned,
ts,
});
}
"flow_node_start" => {
let run_id = v["run_id"].as_str().unwrap_or("").to_string();
let node_id = v["node_id"].as_str().unwrap_or("").to_string();
let label = v["label"].as_str().unwrap_or(&node_id).to_string();
let parent_node_id = v["parent_node_id"].as_str().map(String::from);
let kind = v
.get("kind")
.and_then(|k| serde_json::from_value(k.clone()).ok())
.unwrap_or(nodegraph::NodeKind::UserConfirm);
let ts = parse_ts(v);
out.push(TranscriptEntry::FlowNodeStart {
run_id,
node_id,
kind,
label,
parent_node_id,
ts,
});
}
"flow_node_end" => {
let run_id = v["run_id"].as_str().unwrap_or("").to_string();
let node_id = v["node_id"].as_str().unwrap_or("").to_string();
let status: event::FlowNodeStatus = v
.get("status")
.and_then(|s| serde_json::from_value(s.clone()).ok())
.unwrap_or(event::FlowNodeStatus::Ok);
let output_preview = v["output_preview"].as_str().map(String::from);
let ts = parse_ts(v);
out.push(TranscriptEntry::FlowNodeEnd {
run_id,
node_id,
status,
output_preview,
ts,
});
}
"tool_node" => {
let run_id = v["run_id"].as_str().unwrap_or("").to_string();
let parent_node_id = v["parent_node_id"].as_str().unwrap_or("").to_string();
let tool_use_id = v["tool_use_id"].as_str().unwrap_or("").to_string();
let tool_name = v["tool_name"].as_str().unwrap_or("").to_string();
let args_preview = v["args_preview"].as_str().unwrap_or("").to_string();
let call_intent = v
.get("call_intent")
.and_then(|value| serde_json::from_value(value.clone()).ok());
let ts = parse_ts(v);
out.push(TranscriptEntry::ToolNode {
run_id,
parent_node_id,
tool_use_id,
tool_name,
args_preview,
call_intent,
ts,
});
}
"flow_end" => {
let run_id = v["run_id"].as_str().unwrap_or("").to_string();
let ok = v["status"]["kind"].as_str() == Some("ok");
let cancelled = v["status"]["kind"].as_str() == Some("cancelled");
let ts = parse_ts(v);
out.push(TranscriptEntry::FlowDone {
run_id,
ok,
cancelled,
ts,
});
}
"llm_call" => {
let model = v["model"].as_str().unwrap_or("").to_string();
let provider = v["provider"].as_str().unwrap_or("").to_string();
let context_call_purpose = v
.get("context_call_purpose")
.and_then(|value| serde_json::from_value(value.clone()).ok())
.unwrap_or_default();
let context_call_scope = v
.get("context_call_identity")
.and_then(|identity| identity.get("scope"))
.and_then(|value| serde_json::from_value(value.clone()).ok())
.unwrap_or_else(|| {
if v["run_id"].is_null() {
crate::context_plan::ContextCallScope::Detached
} else {
crate::context_plan::ContextCallScope::Root
}
});
let usage: provider::TokenUsage = v
.get("usage")
.and_then(|u| serde_json::from_value(u.clone()).ok())
.unwrap_or_default();
let wallclock_ms = v["wallclock_ms"].as_u64().unwrap_or(0);
let ttft_ms = v["ttft_ms"].as_u64();
let tokens_per_second = v["tokens_per_second"].as_f64();
let run_id = v["run_id"]
.as_str()
.and_then(|s| uuid::Uuid::parse_str(s).ok())
.map(event::FlowRunId);
let node_id = v["node_id"].as_str().map(String::from);
let ts = parse_ts(v);
out.push(TranscriptEntry::LlmCall {
model,
provider,
context_call_purpose,
context_call_scope,
usage,
wallclock_ms,
ttft_ms,
tokens_per_second,
run_id,
node_id,
ts,
});
}
"tool_pending_approval" | "tool_approved" | "tool_denied" => {
let Some(run_id) = v["run_id"]
.as_str()
.and_then(|raw| uuid::Uuid::parse_str(raw).ok())
.map(event::FlowRunId)
else {
continue;
};
let tool_use_id = v["tool_use_id"].as_str().unwrap_or_default().to_string();
if tool_use_id.is_empty() {
continue;
}
let seq = v["seq"].as_u64().unwrap_or(0);
let correlation = (run_id.0.to_string(), tool_use_id.clone());
if canonical_permissions.contains(&correlation) {
continue;
}
let identity = if ty == "tool_pending_approval" {
let identity = crate::workflow::WorkflowPermissionIdentity::Legacy {
seq,
run_id: run_id.0.to_string(),
tool_use_id: tool_use_id.clone(),
};
legacy_pending
.entry(correlation.clone())
.or_default()
.push(identity.clone());
identity
} else {
legacy_pending
.get_mut(&correlation)
.and_then(Vec::pop)
.unwrap_or_else(|| crate::workflow::WorkflowPermissionIdentity::Legacy {
seq,
run_id: run_id.0.to_string(),
tool_use_id: tool_use_id.clone(),
})
};
let state = match ty {
"tool_pending_approval" => crate::workflow::WorkflowPermissionState::Pending,
"tool_approved" => crate::workflow::WorkflowPermissionState::Approved,
"tool_denied" => crate::workflow::WorkflowPermissionState::Denied,
_ => unreachable!(),
};
let actor_label = match ty {
"tool_pending_approval" => "legacy approval actor unavailable",
"tool_approved" => "legacy approver unavailable",
"tool_denied" => "legacy denier unavailable",
_ => unreachable!(),
};
let at = parse_ts(v).unwrap_or_else(chrono::Utc::now);
let payload = if state.is_pending() {
legacy_permission_payload(
run_id,
tool_use_id,
v["tool_name"].as_str().unwrap_or_default(),
v["reason"].as_str(),
actor_label,
at,
)
} else if let Some(pending) = pending_permissions.get(&identity) {
let mut payload = pending.clone();
payload.actor = Some(
crate::permission_audit::PermissionProjectionActor::UnknownLegacy {
label: actor_label.into(),
},
);
payload.reason = v["reason"].as_str().map(String::from);
payload.at = at;
payload
} else {
legacy_permission_payload(
run_id,
tool_use_id,
v["tool_name"].as_str().unwrap_or_default(),
v["reason"].as_str(),
actor_label,
at,
)
};
if state.is_pending() {
pending_permissions.insert(identity.clone(), payload.clone());
} else {
pending_permissions.remove(&identity);
}
out.push(TranscriptEntry::PermissionRequest {
identity,
payload: Box::new(payload),
state,
});
}
"permission_request_created"
| "permission_request_targeted"
| "permission_request_deferred"
| "permission_request_approved"
| "permission_request_denied"
| "permission_request_cancelled"
| "unrestricted_execution" => {
let Some(payload) = v.get("payload").and_then(|payload| {
serde_json::from_value::<crate::permission_audit::PermissionRequestAudit>(
payload.clone(),
)
.ok()
}) else {
continue;
};
let state = match ty {
"permission_request_created"
| "permission_request_targeted"
| "permission_request_deferred" => {
crate::workflow::WorkflowPermissionState::Pending
}
"permission_request_approved" => {
crate::workflow::WorkflowPermissionState::Approved
}
"permission_request_denied" => crate::workflow::WorkflowPermissionState::Denied,
"permission_request_cancelled" => {
crate::workflow::WorkflowPermissionState::Cancelled
}
"unrestricted_execution" => {
crate::workflow::WorkflowPermissionState::Unrestricted
}
_ => unreachable!(),
};
let Some(request_id) = payload.request_id.clone() else {
continue;
};
canonical_permissions.insert((
payload.requesting_run_id.0.to_string(),
payload.tool_use_id.clone(),
));
let identity =
crate::workflow::WorkflowPermissionIdentity::Canonical { request_id };
if state.is_pending() {
pending_permissions.insert(identity.clone(), payload.clone());
} else {
pending_permissions.remove(&identity);
}
out.push(TranscriptEntry::PermissionRequest {
identity,
payload: Box::new(payload),
state,
});
}
"permission_group_created"
| "permission_group_updated"
| "permission_group_resolved" => {
if let Some(payload) = v.get("payload").and_then(|payload| {
serde_json::from_value::<crate::permission_audit::PermissionGroupAudit>(
payload.clone(),
)
.ok()
}) {
out.push(TranscriptEntry::PermissionGroup {
payload,
resolved: ty == "permission_group_resolved",
});
}
}
"terminal_final_state" => {
let handle = v["handle"].as_str().unwrap_or("").to_string();
if let Some(screen) = v.get("screen")
&& let Ok(screen) =
serde_json::from_value::<crate::tools::term::TerminalScreen>(screen.clone())
{
out.push(TranscriptEntry::TerminalFinalState { handle, screen });
}
}
"mermaid_diagram" => {
if let Some(source) = v.get("source").and_then(|s| s.as_str()) {
out.push(TranscriptEntry::MermaidDiagram {
source: source.to_string(),
});
}
}
_ => {}
}
}
out.extend(
pending_permissions
.into_iter()
.map(|(identity, mut payload)| {
payload.reason = Some("interrupted at end of persisted history".into());
TranscriptEntry::PermissionRequest {
identity,
payload: Box::new(payload),
state: crate::workflow::WorkflowPermissionState::Interrupted,
}
}),
);
Ok(out)
}
pub(crate) fn project_transcript_records(
records: &[crate::event_log::reader::ReplayRecord],
ownership: &crate::event_log::replay::FlowOwnership,
) -> Vec<TranscriptEntry> {
let mut patches: HashMap<u64, Vec<AttachmentPatch>> = HashMap::new();
let mut tool_started_at = HashMap::new();
for record in records {
if let crate::event::Event::AttachmentDegraded {
message_seq,
part_index,
file_basename,
reason,
..
} = &record.envelope.event
{
patches
.entry(*message_seq)
.or_default()
.push(AttachmentPatch {
part_index: *part_index,
file_basename: file_basename.clone(),
reason: reason.clone(),
});
}
if let crate::event::Event::AssistantMsg { message, .. } = &record.envelope.event
&& let Some(ts) = record.persisted_ts
{
for part in &message.parts {
if let MessagePart::ToolUse { id, .. } = part {
tool_started_at.insert(id.clone(), ts);
}
}
}
}
let mut out = Vec::new();
let mut messages = Vec::new();
let mut message_positions = HashMap::new();
let mut pending_permissions = std::collections::BTreeMap::<
crate::workflow::WorkflowPermissionIdentity,
crate::permission_audit::PermissionRequestAudit,
>::new();
let mut legacy_pending = std::collections::HashMap::<
(String, String),
Vec<crate::workflow::WorkflowPermissionIdentity>,
>::new();
let mut canonical_permissions = std::collections::HashSet::new();
let mut session_activity = crate::activity::ActivityAccumulator::default();
let mut turn_activity = crate::activity::ActivityAccumulator::default();
for record in records {
let seq = record.envelope.seq;
let ts = record.persisted_ts;
if matches!(
&record.envelope.event,
crate::event::Event::TurnStart { .. }
) {
turn_activity = crate::activity::ActivityAccumulator::default();
} else {
session_activity.observe(&record.envelope.event);
turn_activity.observe(&record.envelope.event);
}
match &record.envelope.event {
crate::event::Event::UserMsg {
message,
flow_run_id,
..
}
| crate::event::Event::AssistantMsg {
message,
flow_run_id,
..
}
| crate::event::Event::ToolResultMsg {
message,
flow_run_id,
..
}
| crate::event::Event::SystemMsg {
message,
flow_run_id,
..
} => {
let mut message = message.clone();
let belongs_to_root =
message_belongs_to_root(flow_run_id.as_ref(), &ownership.spawned);
if let Some(patches) = patches.get(&seq) {
apply_attachment_patches(&mut message, patches);
}
let flow_run_id = flow_run_id.as_ref().and_then(|run_id| {
if ownership.known.contains(run_id) && !ownership.spawned.contains(run_id) {
None
} else {
Some(run_id.0.to_string())
}
});
let entry = TranscriptEntry::Message {
message,
flow_run_id,
};
if belongs_to_root {
push_transcript_message(
&mut out,
&mut messages,
&mut message_positions,
seq,
entry,
);
} else {
out.push(entry);
}
if let crate::event::Event::ToolResultMsg { message, .. } = &record.envelope.event
&& let Some(finished_at) = ts
{
for part in &message.parts {
if let MessagePart::ToolResult { tool_use_id, .. } = part
&& let Some(started_at) = tool_started_at.get(tool_use_id)
{
let elapsed_ms = finished_at
.signed_duration_since(*started_at)
.num_milliseconds()
.max(0) as u64;
out.push(TranscriptEntry::ToolTiming {
tool_use_id: tool_use_id.clone(),
elapsed_ms,
});
}
}
}
}
crate::event::Event::TurnStart { .. } => {}
crate::event::Event::TurnEnd { .. } => {
out.push(TranscriptEntry::ActivitySummary {
turn: turn_activity.summary(),
session: session_activity.summary(),
turn_files: turn_activity.file_paths(),
session_files: session_activity.file_paths(),
});
}
crate::event::Event::ContextCompact {
flow_run_id,
compacted_range_start,
compacted_range_end,
replacement_msg_seq,
..
} => {
if !message_belongs_to_root(flow_run_id.as_ref(), &ownership.spawned) {
continue;
}
let range_start = *compacted_range_start as usize;
let range_end = *compacted_range_end as usize;
let Some(replacement_seq) = replacement_msg_seq else {
continue;
};
compact_transcript_messages(
&mut out,
&mut messages,
&mut message_positions,
range_start,
range_end,
*replacement_seq,
);
}
crate::event::Event::CompactionSummary {
flow_run_id,
range_start,
range_end,
compacted_count,
before_tokens,
after_tokens,
summary,
..
} => {
if !message_belongs_to_root(flow_run_id.as_ref(), &ownership.spawned) {
continue;
}
out.push(TranscriptEntry::CompactionSummary {
range_start: *range_start as usize,
range_end: *range_end as usize,
compacted_count: *compacted_count,
before_tokens: *before_tokens,
after_tokens: *after_tokens,
summary: summary.clone(),
ts,
});
}
crate::event::Event::DiffPreview {
tool_use_id,
title,
old_content,
new_content,
unified_diff,
..
} => out.push(TranscriptEntry::DiffPreview {
tool_use_id: tool_use_id.clone(),
title: title.clone(),
old_content: old_content.clone(),
new_content: new_content.clone(),
unified_diff: unified_diff.clone(),
}),
crate::event::Event::FileEditApplied {
turn_id,
flow_run_id,
tool_use_id,
tool_name,
path,
metrics,
} => out.push(TranscriptEntry::FileEditApplied {
turn_id: turn_id.as_ref().map(ToString::to_string),
flow_run_id: flow_run_id.as_ref().map(ToString::to_string),
tool_use_id: tool_use_id.clone(),
tool_name: tool_name.clone(),
path: path.clone(),
metrics: *metrics,
}),
crate::event::Event::FlowGraph { run_id, graph } => {
out.push(TranscriptEntry::FlowGraph {
run_id: run_id.0.to_string(),
flow_name: graph.flow_name.clone(),
graph: graph.clone(),
ts,
});
}
crate::event::Event::FlowStart {
run_id,
flow_name,
parent_run_id,
parent_node_id,
spawned,
} => out.push(TranscriptEntry::FlowStart {
run_id: run_id.0.to_string(),
flow_name: flow_name.clone(),
parent_run_id: parent_run_id.as_ref().map(|run_id| run_id.0.to_string()),
parent_node_id: parent_node_id.clone(),
spawned: *spawned,
ts,
}),
crate::event::Event::FlowNodeStart {
run_id,
node_id,
kind,
label,
parent_node_id,
} => out.push(TranscriptEntry::FlowNodeStart {
run_id: run_id.0.to_string(),
node_id: node_id.clone(),
kind: kind.clone(),
label: if label.is_empty() {
node_id.clone()
} else {
label.clone()
},
parent_node_id: parent_node_id.clone(),
ts,
}),
crate::event::Event::FlowNodeEnd {
run_id,
node_id,
status,
output_preview,
} => out.push(TranscriptEntry::FlowNodeEnd {
run_id: run_id.0.to_string(),
node_id: node_id.clone(),
status: status.clone(),
output_preview: output_preview.clone(),
ts,
}),
crate::event::Event::ToolNode {
run_id,
parent_node_id,
tool_use_id,
tool_name,
args_preview,
call_intent,
} => out.push(TranscriptEntry::ToolNode {
run_id: run_id.0.to_string(),
parent_node_id: parent_node_id.clone(),
tool_use_id: tool_use_id.clone(),
tool_name: tool_name.clone(),
args_preview: args_preview.clone(),
call_intent: call_intent.clone(),
ts,
}),
crate::event::Event::FlowEnd { run_id, status, .. } => {
out.push(TranscriptEntry::FlowDone {
run_id: run_id.0.to_string(),
ok: matches!(status, crate::event::FlowStatus::Ok),
cancelled: matches!(status, crate::event::FlowStatus::Cancelled),
ts,
});
}
crate::event::Event::LlmCall {
model,
provider,
context_call_purpose,
context_call_identity,
usage,
wallclock_ms,
ttft_ms,
tokens_per_second,
run_id,
node_id,
..
} => {
let context_call_scope = context_call_identity.as_ref().map_or_else(
|| {
if run_id.is_none() {
crate::context_plan::ContextCallScope::Detached
} else {
crate::context_plan::ContextCallScope::Root
}
},
|identity| identity.scope,
);
out.push(TranscriptEntry::LlmCall {
model: model.clone(),
provider: provider.clone(),
context_call_purpose: context_call_purpose.unwrap_or_default(),
context_call_scope,
usage: usage.clone(),
wallclock_ms: *wallclock_ms,
ttft_ms: *ttft_ms,
tokens_per_second: *tokens_per_second,
run_id: run_id.clone(),
node_id: node_id.clone(),
ts,
});
}
crate::event::Event::ToolPendingApproval {
run_id,
tool_use_id,
..
}
| crate::event::Event::ToolApproved {
run_id,
tool_use_id,
..
}
| crate::event::Event::ToolDenied {
run_id,
tool_use_id,
..
} => {
if tool_use_id.is_empty() {
continue;
}
let correlation = (run_id.0.to_string(), tool_use_id.clone());
if canonical_permissions.contains(&correlation) {
continue;
}
let (state, actor_label, tool_name, reason) = match &record.envelope.event {
crate::event::Event::ToolPendingApproval { tool_name, .. } => (
crate::workflow::WorkflowPermissionState::Pending,
"legacy approval actor unavailable",
tool_name.as_str(),
None,
),
crate::event::Event::ToolApproved { .. } => (
crate::workflow::WorkflowPermissionState::Approved,
"legacy approver unavailable",
"",
None,
),
crate::event::Event::ToolDenied { reason, .. } => (
crate::workflow::WorkflowPermissionState::Denied,
"legacy denier unavailable",
"",
Some(reason.as_str()),
),
_ => unreachable!(),
};
let pending = state.is_pending();
let identity = if pending {
let identity = crate::workflow::WorkflowPermissionIdentity::Legacy {
seq,
run_id: run_id.0.to_string(),
tool_use_id: tool_use_id.clone(),
};
legacy_pending
.entry(correlation.clone())
.or_default()
.push(identity.clone());
identity
} else {
legacy_pending
.get_mut(&correlation)
.and_then(Vec::pop)
.unwrap_or_else(|| crate::workflow::WorkflowPermissionIdentity::Legacy {
seq,
run_id: run_id.0.to_string(),
tool_use_id: tool_use_id.clone(),
})
};
let at = ts.unwrap_or_else(chrono::Utc::now);
let payload = if state.is_pending() {
legacy_permission_payload(
run_id.clone(),
tool_use_id.clone(),
tool_name,
reason,
actor_label,
at,
)
} else if let Some(pending) = pending_permissions.get(&identity) {
let mut payload = pending.clone();
payload.actor = Some(
crate::permission_audit::PermissionProjectionActor::UnknownLegacy {
label: actor_label.into(),
},
);
payload.reason = reason.map(String::from);
payload.at = at;
payload
} else {
legacy_permission_payload(
run_id.clone(),
tool_use_id.clone(),
tool_name,
reason,
actor_label,
at,
)
};
if state.is_pending() {
pending_permissions.insert(identity.clone(), payload.clone());
} else {
pending_permissions.remove(&identity);
}
out.push(TranscriptEntry::PermissionRequest {
identity,
payload: Box::new(payload),
state,
});
}
crate::event::Event::PermissionRequestCreated { payload }
| crate::event::Event::PermissionRequestTargeted { payload }
| crate::event::Event::PermissionRequestDeferred { payload }
| crate::event::Event::PermissionRequestApproved { payload }
| crate::event::Event::PermissionRequestDenied { payload }
| crate::event::Event::PermissionRequestCancelled { payload }
| crate::event::Event::UnrestrictedExecution { payload } => {
let state = match &record.envelope.event {
crate::event::Event::PermissionRequestCreated { .. }
| crate::event::Event::PermissionRequestTargeted { .. }
| crate::event::Event::PermissionRequestDeferred { .. } => {
crate::workflow::WorkflowPermissionState::Pending
}
crate::event::Event::PermissionRequestApproved { .. } => {
crate::workflow::WorkflowPermissionState::Approved
}
crate::event::Event::PermissionRequestDenied { .. } => {
crate::workflow::WorkflowPermissionState::Denied
}
crate::event::Event::PermissionRequestCancelled { .. } => {
crate::workflow::WorkflowPermissionState::Cancelled
}
crate::event::Event::UnrestrictedExecution { .. } => {
crate::workflow::WorkflowPermissionState::Unrestricted
}
_ => unreachable!(),
};
let Some(request_id) = payload.request_id.clone() else {
continue;
};
canonical_permissions.insert((
payload.requesting_run_id.0.to_string(),
payload.tool_use_id.clone(),
));
let identity =
crate::workflow::WorkflowPermissionIdentity::Canonical { request_id };
if state.is_pending() {
pending_permissions.insert(identity.clone(), payload.clone());
} else {
pending_permissions.remove(&identity);
}
out.push(TranscriptEntry::PermissionRequest {
identity,
payload: Box::new(payload.clone()),
state,
});
}
crate::event::Event::PermissionGroupCreated { payload }
| crate::event::Event::PermissionGroupUpdated { payload }
| crate::event::Event::PermissionGroupResolved { payload } => {
out.push(TranscriptEntry::PermissionGroup {
payload: payload.clone(),
resolved: matches!(
&record.envelope.event,
crate::event::Event::PermissionGroupResolved { .. }
),
});
}
crate::event::Event::TerminalFinalState { handle, screen, .. } => {
out.push(TranscriptEntry::TerminalFinalState {
handle: handle.clone(),
screen: screen.clone(),
});
}
crate::event::Event::MermaidDiagram { source } => {
out.push(TranscriptEntry::MermaidDiagram {
source: source.clone(),
});
}
_ => {}
}
}
out.extend(
pending_permissions
.into_iter()
.map(|(identity, mut payload)| {
payload.reason = Some("interrupted at end of persisted history".into());
TranscriptEntry::PermissionRequest {
identity,
payload: Box::new(payload),
state: crate::workflow::WorkflowPermissionState::Interrupted,
}
}),
);
out
}
pub trait MessageProjection {
fn to_messages(&self) -> Vec<Message>;
fn to_messages_with_seq(&self) -> Vec<(u64, Message)>;
}
impl MessageProjection for [crate::event::EventEnvelope] {
fn to_messages(&self) -> Vec<Message> {
self.to_messages_with_seq()
.into_iter()
.map(|(_, msg)| msg)
.collect()
}
fn to_messages_with_seq(&self) -> Vec<(u64, Message)> {
let spawned_flow_ids = spawned_flow_ids(self);
let mut acc: Vec<(u64, Message)> = Vec::new();
let mut positions = HashMap::new();
for env in self {
apply_envelope_to_messages(env, &spawned_flow_ids, &mut acc, &mut positions);
}
acc
}
}
pub(crate) fn spawned_flow_ids(
envelopes: &[crate::event::EventEnvelope],
) -> std::collections::HashSet<crate::event::FlowRunId> {
let mut children =
std::collections::HashMap::<crate::event::FlowRunId, Vec<crate::event::FlowRunId>>::new();
let mut spawned = std::collections::HashSet::new();
for env in envelopes {
if let crate::event::Event::FlowStart {
run_id,
parent_run_id,
spawned: is_spawned,
..
} = &env.event
{
if let Some(parent_run_id) = parent_run_id {
children
.entry(parent_run_id.clone())
.or_default()
.push(run_id.clone());
}
if *is_spawned {
spawned.insert(run_id.clone());
}
}
}
let mut queue = std::collections::VecDeque::from_iter(spawned.iter().cloned());
while let Some(parent) = queue.pop_front() {
if let Some(descendants) = children.get(&parent) {
for descendant in descendants {
if spawned.insert(descendant.clone()) {
queue.push_back(descendant.clone());
}
}
}
}
spawned
}
pub(crate) fn message_positions(acc: &[(u64, Message)]) -> HashMap<u64, usize> {
acc.iter()
.enumerate()
.map(|(index, (seq, _))| (*seq, index))
.collect()
}
fn rebuild_message_positions(acc: &[(u64, Message)], positions: &mut HashMap<u64, usize>) {
positions.clear();
positions.extend(
acc.iter()
.enumerate()
.map(|(index, (seq, _))| (*seq, index)),
);
}
pub(crate) fn apply_attachment_degradation(
acc: &mut [(u64, Message)],
positions: &HashMap<u64, usize>,
message_seq: u64,
part_index: usize,
file_basename: &str,
reason: &str,
) -> bool {
let Some(message_index) = positions.get(&message_seq).copied() else {
return false;
};
let Some(part) = acc
.get_mut(message_index)
.and_then(|(_, message)| message.parts.get_mut(part_index))
else {
return false;
};
let replacement = MessagePart::Text {
text: format!("[attachment unavailable: {} — {}]", file_basename, reason),
};
if *part == replacement {
return false;
}
*part = replacement;
true
}
pub(crate) fn message_belongs_to_root(
flow_run_id: Option<&crate::event::FlowRunId>,
spawned_flow_ids: &std::collections::HashSet<crate::event::FlowRunId>,
) -> bool {
flow_run_id.is_none_or(|run_id| !spawned_flow_ids.contains(run_id))
}
pub(crate) fn apply_envelope_to_messages(
env: &crate::event::EventEnvelope,
spawned_flow_ids: &std::collections::HashSet<crate::event::FlowRunId>,
acc: &mut Vec<(u64, Message)>,
positions: &mut HashMap<u64, usize>,
) -> bool {
match &env.event {
crate::event::Event::UserMsg {
message,
flow_run_id,
..
}
| crate::event::Event::AssistantMsg {
message,
flow_run_id,
..
}
| crate::event::Event::ToolResultMsg {
message,
flow_run_id,
..
} if message_belongs_to_root(flow_run_id.as_ref(), spawned_flow_ids) => {
positions.insert(env.seq, acc.len());
acc.push((env.seq, message.clone()));
true
}
crate::event::Event::SystemMsg {
message,
flow_run_id,
..
} if message_belongs_to_root(flow_run_id.as_ref(), spawned_flow_ids) => {
positions.insert(env.seq, acc.len());
acc.push((env.seq, message.clone()));
true
}
crate::event::Event::ContextCompact {
flow_run_id,
compacted_range_start,
compacted_range_end,
replacement_msg_seq,
summary_text,
after_tokens,
before_tokens,
..
} if message_belongs_to_root(flow_run_id.as_ref(), spawned_flow_ids) => {
let range_start = *compacted_range_start as usize;
let range_end = *compacted_range_end as usize;
if range_start > range_end || range_end >= acc.len() {
return false;
}
let Some(rep_seq) = replacement_msg_seq else {
return false;
};
let Some(rep_idx) = positions.get(rep_seq).copied() else {
return false;
};
if *after_tokens >= *before_tokens {
return false;
}
let removed_count = range_end - range_start + 1;
let replacement = if let Some(summary) = summary_text {
(
*rep_seq,
Message::system_compact_summary(
crate::event::TurnId::now(),
summary.clone(),
range_start as u64,
range_end as u64,
removed_count,
),
)
} else {
acc[rep_idx].clone()
};
let (adjusted_start, adjusted_end) = if rep_idx < range_start {
acc.remove(rep_idx);
(range_start - 1, range_end - 1)
} else if rep_idx > range_end {
acc.remove(rep_idx);
(range_start, range_end)
} else {
(range_start, range_end)
};
acc.splice(adjusted_start..=adjusted_end, [replacement]);
rebuild_message_positions(acc, positions);
true
}
crate::event::Event::Checkpoint {
flow_run_id,
messages,
..
} if message_belongs_to_root(flow_run_id.as_ref(), spawned_flow_ids) => {
let checkpoint = messages
.iter()
.cloned()
.enumerate()
.map(|(index, message)| (u64::MAX.saturating_sub(index as u64), message))
.collect::<Vec<_>>();
if *acc == checkpoint {
false
} else {
*acc = checkpoint;
rebuild_message_positions(acc, positions);
true
}
}
crate::event::Event::AttachmentDegraded {
message_seq,
part_index,
file_basename,
reason,
..
} => apply_attachment_degradation(
acc,
positions,
*message_seq,
*part_index,
file_basename,
reason,
),
_ => false,
}
}
#[cfg(test)]
mod tests {
use crate::event::{Event, EventEnvelope, FlowRunId, TurnId};
use crate::message::{Message, MessageOrigin, MessagePart, MessageRole};
use uuid::Uuid;
fn message(role: MessageRole, text: &str) -> Message {
Message {
role,
parts: vec![MessagePart::Text {
text: text.to_string(),
}],
turn_id: TurnId::now(),
origin: MessageOrigin::User,
}
}
fn flow_start(run_id: FlowRunId, parent_run_id: Option<FlowRunId>, spawned: bool) -> Event {
Event::FlowStart {
run_id,
flow_name: "test".into(),
spawned,
parent_run_id,
parent_node_id: None,
}
}
fn canonical_permission_payload(
run_id: FlowRunId,
tool_use_id: &str,
actor: crate::permission_audit::PermissionProjectionActor,
) -> crate::permission_audit::PermissionRequestAudit {
crate::permission_audit::PermissionRequestAudit {
request_id: Some(crate::permission::PermissionRequestId(Uuid::now_v7())),
revision: 1,
session_id: "session".into(),
requesting_run_id: run_id.clone(),
parent_run_id: None,
root_run_id: run_id,
tool_use_id: tool_use_id.into(),
tool: "fs.read".into(),
call_intent: None,
tier: crate::tool::Tier::Two,
execution_boundary: Default::default(),
provenance: Default::default(),
target: crate::permission_audit::PermissionAuditTarget::User,
group_ids: Vec::new(),
policy: crate::permission_audit::PermissionPolicyReference {
snapshot_id: "snapshot".into(),
rule_id: "rule".into(),
},
escalation_path: Vec::new(),
decision_id: Some(format!("decision-{tool_use_id}")),
actor: Some(actor),
scope: None,
reason: None,
at: chrono::Utc::now(),
}
}
#[test]
fn replay_excludes_subagent_messages() {
let root = FlowRunId(Uuid::now_v7());
let child = FlowRunId(Uuid::now_v7());
let envelopes = vec![
EventEnvelope::new(1, flow_start(root.clone(), None, false)),
EventEnvelope::new(2, flow_start(child.clone(), Some(root), true)),
EventEnvelope::new(
3,
Event::AssistantMsg {
turn_id: TurnId::now(),
flow_run_id: Some(child),
message: message(MessageRole::Assistant, "child"),
},
),
];
assert!(super::MessageProjection::to_messages(envelopes.as_slice()).is_empty());
}
#[test]
fn transcript_replay_ignores_spawned_compaction_events() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("events.jsonl");
let child = FlowRunId::now();
let events = [
EventEnvelope::new(
1,
Event::UserMsg {
turn_id: TurnId::now(),
flow_run_id: None,
message: message(MessageRole::User, "root user"),
},
),
EventEnvelope::new(2, flow_start(child.clone(), None, true)),
EventEnvelope::new(
3,
Event::SystemMsg {
turn_id: TurnId::now(),
flow_run_id: Some(child.clone()),
message: Message::system_compact_summary(
TurnId::now(),
"child summary",
0,
0,
1,
),
},
),
EventEnvelope::new(
4,
Event::ContextCompact {
session_id: "session".into(),
flow_run_id: Some(child.clone()),
before_tokens: 100,
after_tokens: 10,
compacted_range_start: 0,
compacted_range_end: 0,
summary_text: Some("child summary".into()),
replacement_msg_seq: Some(3),
},
),
EventEnvelope::new(
5,
Event::CompactionSummary {
session_id: "session".into(),
flow_run_id: Some(child),
range_start: 0,
range_end: 0,
compacted_count: 1,
before_tokens: 100,
after_tokens: 10,
summary: "child summary".into(),
},
),
];
let jsonl = events
.iter()
.map(serde_json::to_string)
.collect::<Result<Vec<_>, _>>()
.unwrap()
.join("\n");
std::fs::write(&path, jsonl).unwrap();
let raw_entries = super::replay_transcript_from_raw(&path).unwrap();
let entries = super::replay_transcript_from(&path).unwrap();
assert_eq!(format!("{raw_entries:#?}"), format!("{entries:#?}"));
assert!(entries.iter().any(|entry| matches!(
entry,
super::TranscriptEntry::Message { message, flow_run_id: None }
if message.text_concat() == "root user"
)));
assert!(entries.iter().any(|entry| matches!(
entry,
super::TranscriptEntry::FlowStart { spawned: true, .. }
)));
assert!(
!entries
.iter()
.any(|entry| matches!(entry, super::TranscriptEntry::CompactionSummary { .. }))
);
}
#[test]
fn transcript_compaction_preserves_interleaved_non_message_entries() {
let spawned = FlowRunId::now();
let events = [
EventEnvelope::new(
1,
Event::UserMsg {
turn_id: TurnId::now(),
flow_run_id: None,
message: message(MessageRole::User, "old user"),
},
),
EventEnvelope::new(2, flow_start(spawned.clone(), None, true)),
EventEnvelope::new(
3,
Event::AssistantMsg {
turn_id: TurnId::now(),
flow_run_id: Some(spawned.clone()),
message: message(MessageRole::Assistant, "spawned assistant"),
},
),
EventEnvelope::new(
4,
Event::AssistantMsg {
turn_id: TurnId::now(),
flow_run_id: None,
message: message(MessageRole::Assistant, "old assistant"),
},
),
EventEnvelope::new(
5,
Event::SystemMsg {
turn_id: TurnId::now(),
flow_run_id: None,
message: Message::system_compact_summary(TurnId::now(), "summary", 0, 1, 2),
},
),
EventEnvelope::new(
6,
Event::ContextCompact {
session_id: "session".into(),
flow_run_id: None,
before_tokens: 100,
after_tokens: 10,
compacted_range_start: 0,
compacted_range_end: 1,
summary_text: Some("summary".into()),
replacement_msg_seq: Some(5),
},
),
];
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("events.jsonl");
std::fs::write(
&path,
events
.iter()
.map(serde_json::to_string)
.collect::<Result<Vec<_>, _>>()
.unwrap()
.join("\n"),
)
.unwrap();
let raw_entries = super::replay_transcript_from_raw(&path).unwrap();
let entries = super::replay_transcript_from(&path).unwrap();
assert_eq!(format!("{raw_entries:#?}"), format!("{entries:#?}"));
assert_eq!(entries.len(), 3);
assert!(matches!(
&entries[0],
super::TranscriptEntry::Message { message, .. }
if message.text_concat() == "summary"
));
assert!(matches!(
&entries[1],
super::TranscriptEntry::FlowStart { run_id, .. } if run_id == &spawned.0.to_string()
));
assert!(matches!(
&entries[2],
super::TranscriptEntry::Message { message, flow_run_id: Some(run_id) }
if message.text_concat() == "spawned assistant"
&& run_id == &spawned.0.to_string()
));
}
#[test]
fn late_attachment_degradation_updates_all_replay_views() {
use crate::message::{ImageData, ImageSource};
use crate::provider::ImageDetail;
let image = Message {
role: MessageRole::User,
parts: vec![MessagePart::Image {
source: ImageSource {
media_type: "image/png".into(),
data: ImageData::Path {
path: "/tmp/missing.png".into(),
},
detail: ImageDetail::Auto,
},
}],
turn_id: TurnId::now(),
origin: MessageOrigin::User,
};
let events = vec![
EventEnvelope::new(
1,
Event::UserMsg {
turn_id: image.turn_id.clone(),
flow_run_id: None,
message: image,
},
),
EventEnvelope::new(
2,
Event::AttachmentDegraded {
turn_id: None,
flow_run_id: None,
message_seq: 1,
part_index: 0,
file_basename: "missing.png".into(),
reason: "unreadable".into(),
},
),
];
let jsonl = events
.iter()
.map(serde_json::to_string)
.collect::<Result<Vec<_>, _>>()
.unwrap()
.join("\n");
let replay =
crate::event_log::replay::SessionReplay::from_reader(std::io::Cursor::new(jsonl), None)
.unwrap();
assert!(
replay.compacted_messages[0]
.1
.text_concat()
.contains("missing.png")
);
assert!(
replay.all_messages[0]
.1
.text_concat()
.contains("missing.png")
);
let transcript = crate::event_log::replay::transcript_from_envelopes(&events);
assert!(matches!(
&transcript[0],
super::TranscriptEntry::Message { message, .. }
if message.text_concat().contains("missing.png")
));
}
#[test]
fn streaming_transcript_preserves_missing_legacy_optional_fields() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("events.jsonl");
let run_id = Uuid::now_v7();
let turn_id = TurnId::now();
let lines = [
serde_json::json!({
"type": "flow_start",
"seq": 1,
"run_id": run_id,
}),
serde_json::json!({
"type": "flow_node_start",
"seq": 2,
"run_id": run_id,
"node_id": "legacy-node",
}),
serde_json::json!({
"type": "assistant_msg",
"seq": 3,
"turn_id": turn_id,
"message": message(MessageRole::Assistant, "legacy"),
}),
];
std::fs::write(
&path,
lines
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
)
.unwrap();
let raw_entries = super::replay_transcript_from_raw(&path).unwrap();
let entries = super::replay_transcript_from(&path).unwrap();
assert_eq!(format!("{raw_entries:#?}"), format!("{entries:#?}"));
}
#[test]
fn replay_excludes_ordinary_subflow_execution_messages() {
let root = FlowRunId(Uuid::now_v7());
let child = FlowRunId(Uuid::now_v7());
let envelopes = vec![
EventEnvelope::new(1, flow_start(root.clone(), None, false)),
EventEnvelope::new(2, flow_start(child.clone(), Some(root), false)),
EventEnvelope::new(
3,
Event::AssistantMsg {
turn_id: TurnId::now(),
flow_run_id: Some(child),
message: message(MessageRole::Assistant, "ordinary child"),
},
),
];
let messages = super::MessageProjection::to_messages(envelopes.as_slice());
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].text_concat(), "ordinary child");
}
#[test]
fn replay_excludes_descendants_of_spawned_flows() {
let root = FlowRunId(Uuid::now_v7());
let spawned = FlowRunId(Uuid::now_v7());
let descendant = FlowRunId(Uuid::now_v7());
let envelopes = vec![
EventEnvelope::new(1, flow_start(root.clone(), None, false)),
EventEnvelope::new(2, flow_start(spawned.clone(), Some(root), true)),
EventEnvelope::new(3, flow_start(descendant.clone(), Some(spawned), false)),
EventEnvelope::new(
4,
Event::AssistantMsg {
turn_id: TurnId::now(),
flow_run_id: Some(descendant),
message: message(MessageRole::Assistant, "spawned descendant"),
},
),
];
assert!(super::MessageProjection::to_messages(envelopes.as_slice()).is_empty());
}
#[test]
fn replay_keeps_unknown_nonspawned_message() {
let orphan = FlowRunId(Uuid::now_v7());
let envelopes = vec![EventEnvelope::new(
1,
Event::AssistantMsg {
turn_id: TurnId::now(),
flow_run_id: Some(orphan),
message: message(MessageRole::Assistant, "orphan"),
},
)];
let messages = super::MessageProjection::to_messages(envelopes.as_slice());
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].text_concat(), "orphan");
}
#[test]
fn s8_immediate_canonical_decisions_suppress_paired_legacy_identities() {
use crate::permission_audit::PermissionProjectionActor;
use crate::workflow::{WorkflowPermissionIdentity, WorkflowPermissionState};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("events.jsonl");
let run = FlowRunId(Uuid::now_v7());
let cases = [
(
"auto",
"permission_request_approved",
"tool_approved",
PermissionProjectionActor::Policy {
policy_version: "snapshot".into(),
rule_id: "auto".into(),
},
WorkflowPermissionState::Approved,
),
(
"grant",
"permission_request_approved",
"tool_approved",
PermissionProjectionActor::User {
session_id: "session".into(),
principal_id: Some("grant-owner".into()),
},
WorkflowPermissionState::Approved,
),
(
"denied",
"permission_request_denied",
"tool_denied",
PermissionProjectionActor::Policy {
policy_version: "snapshot".into(),
rule_id: "deny".into(),
},
WorkflowPermissionState::Denied,
),
(
"unrestricted",
"unrestricted_execution",
"tool_approved",
PermissionProjectionActor::Policy {
policy_version: "snapshot".into(),
rule_id: "unrestricted".into(),
},
WorkflowPermissionState::Unrestricted,
),
];
let mut lines = Vec::new();
for (index, (tool_use_id, canonical_type, legacy_type, actor, _)) in
cases.iter().enumerate()
{
let payload = canonical_permission_payload(run.clone(), tool_use_id, actor.clone());
let seq = (index * 2 + 1) as u64;
lines.push(
serde_json::json!({"type":canonical_type,"seq":seq,"payload":payload}).to_string(),
);
lines.push(
serde_json::json!({"type":legacy_type,"seq":seq + 1,"run_id":run.0.to_string(),"tool_use_id":tool_use_id,"decided_by":"legacy-adapter","reason":"legacy denial"})
.to_string(),
);
}
std::fs::write(&path, lines.join("\n")).unwrap();
let raw_entries = super::replay_transcript_from_raw(&path).unwrap();
let entries = super::replay_transcript_from(&path).unwrap();
assert_eq!(format!("{raw_entries:#?}"), format!("{entries:#?}"));
let permissions = entries
.iter()
.filter_map(|entry| match entry {
super::TranscriptEntry::PermissionRequest {
identity,
payload,
state,
} => Some((identity, payload, state)),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(permissions.len(), cases.len());
for ((identity, payload, state), (tool_use_id, _, _, actor, expected_state)) in
permissions.into_iter().zip(cases)
{
assert!(matches!(
identity,
WorkflowPermissionIdentity::Canonical { .. }
));
assert_eq!(payload.tool_use_id, tool_use_id);
assert_eq!(payload.actor.as_ref(), Some(&actor));
assert_eq!(*state, expected_state);
}
}
#[test]
fn legacy_approval_replay_correlates_repeated_exact_tool_lifecycles() {
use crate::permission_audit::PermissionProjectionActor;
use crate::workflow::{WorkflowPermissionIdentity, WorkflowPermissionState};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("events.jsonl");
let run = Uuid::now_v7().to_string();
let lines = [
serde_json::json!({"type":"tool_pending_approval","seq":10,"run_id":run,"tool_use_id":"same","tool_name":"fs.write","args_preview":"{}","level":"approve"}),
serde_json::json!({"type":"tool_approved","seq":11,"run_id":run,"tool_use_id":"same","decided_by":"user"}),
serde_json::json!({"type":"tool_pending_approval","seq":12,"run_id":run,"tool_use_id":"same","tool_name":"fs.edit","args_preview":"{}","level":"approve"}),
serde_json::json!({"type":"tool_denied","seq":13,"run_id":run,"tool_use_id":"same","reason":"no"}),
serde_json::json!({"type":"tool_approved","seq":14,"run_id":run,"tool_use_id":"orphan","decided_by":"user"}),
];
std::fs::write(
&path,
lines
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
)
.unwrap();
let entries = super::replay_transcript_from(&path).unwrap();
let finals = entries
.iter()
.filter_map(|entry| match entry {
super::TranscriptEntry::PermissionRequest {
identity,
payload,
state,
} if !state.is_pending() => Some((identity, payload, state)),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(finals.len(), 3);
assert_eq!(
finals[0].0,
&WorkflowPermissionIdentity::Legacy {
seq: 10,
run_id: run.clone(),
tool_use_id: "same".into(),
}
);
assert_eq!(finals[0].1.tool, "fs.write");
assert_eq!(*finals[0].2, WorkflowPermissionState::Approved);
assert_eq!(
finals[1].0,
&WorkflowPermissionIdentity::Legacy {
seq: 12,
run_id: run.clone(),
tool_use_id: "same".into(),
}
);
assert_eq!(finals[1].1.tool, "fs.edit");
assert_eq!(*finals[1].2, WorkflowPermissionState::Denied);
assert_eq!(
finals[2].0,
&WorkflowPermissionIdentity::Legacy {
seq: 14,
run_id: run,
tool_use_id: "orphan".into(),
}
);
assert!(finals.iter().all(|(_, payload, _)| {
payload.request_id.is_none()
&& matches!(
payload.actor,
Some(PermissionProjectionActor::UnknownLegacy { .. })
)
}));
}
#[test]
fn unresolved_legacy_pending_replays_as_interrupted_with_origin_identity() {
use crate::workflow::{WorkflowPermissionIdentity, WorkflowPermissionState};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("events.jsonl");
let run = Uuid::now_v7().to_string();
std::fs::write(
&path,
serde_json::json!({"type":"tool_pending_approval","seq":21,"run_id":run,"tool_use_id":"pending","tool_name":"bash.spawn","args_preview":"{}","level":"approve"}).to_string(),
).unwrap();
let entries = super::replay_transcript_from(&path).unwrap();
let (identity, payload) = entries
.iter()
.find_map(|entry| match entry {
super::TranscriptEntry::PermissionRequest {
identity,
payload,
state: WorkflowPermissionState::Interrupted,
} => Some((identity, payload)),
_ => None,
})
.unwrap();
assert_eq!(
identity,
&WorkflowPermissionIdentity::Legacy {
seq: 21,
run_id: run,
tool_use_id: "pending".into(),
}
);
assert_eq!(payload.tool, "bash.spawn");
}
#[test]
fn transcript_reconstructs_tool_timing_and_turn_activity() {
let turn_id = TurnId::now();
let run_id = FlowRunId::now();
let tool_use_id = "edit-1".to_string();
let started_at = chrono::Utc::now();
let finished_at = started_at + chrono::Duration::milliseconds(1_500);
let mut events = vec![
EventEnvelope::new(
1,
Event::TurnStart {
turn_id: turn_id.clone(),
},
),
EventEnvelope::new(2, flow_start(run_id.clone(), None, false)),
EventEnvelope::new(
3,
Event::AssistantMsg {
turn_id: turn_id.clone(),
flow_run_id: Some(run_id.clone()),
message: Message {
role: MessageRole::Assistant,
parts: vec![MessagePart::ToolUse {
id: tool_use_id.clone(),
name: "fs.edit".into(),
input: serde_json::json!({"path": "/repo/src/lib.rs"}),
intent: None,
}],
turn_id: turn_id.clone(),
origin: MessageOrigin::User,
},
},
),
EventEnvelope::new(
4,
Event::ToolNode {
run_id,
parent_node_id: "node".into(),
tool_use_id: tool_use_id.clone(),
tool_name: "fs.edit".into(),
args_preview: "{}".into(),
call_intent: None,
},
),
EventEnvelope::new(
5,
Event::FileEditApplied {
turn_id: Some(turn_id.clone()),
flow_run_id: None,
tool_use_id: Some(tool_use_id.clone()),
tool_name: "fs.edit".into(),
path: "/repo/src/lib.rs".into(),
metrics: crate::activity::EditMetrics {
hunks: 2,
insertions: 7,
deletions: 3,
},
},
),
EventEnvelope::new(
6,
Event::ToolResultMsg {
turn_id: turn_id.clone(),
flow_run_id: None,
message: Message {
role: MessageRole::Tool,
parts: vec![MessagePart::ToolResult {
tool_use_id: tool_use_id.clone(),
content: "null".into(),
is_error: false,
}],
turn_id: turn_id.clone(),
origin: MessageOrigin::User,
},
},
),
EventEnvelope::new(7, Event::TurnEnd { turn_id }),
];
events[2].ts = started_at;
events[5].ts = finished_at;
let entries = crate::event_log::replay::transcript_from_envelopes(&events);
assert!(entries.iter().any(|entry| matches!(
entry,
super::TranscriptEntry::ToolTiming {
tool_use_id: id,
elapsed_ms: 1_500,
} if id == &tool_use_id
)));
let (turn, session, turn_files, session_files) = entries
.iter()
.find_map(|entry| match entry {
super::TranscriptEntry::ActivitySummary {
turn,
session,
turn_files,
session_files,
} => Some((turn, session, turn_files, session_files)),
_ => None,
})
.unwrap();
assert_eq!(turn.attempted_calls, 1);
assert_eq!(turn.completed_calls, 1);
assert_eq!(turn.applied_edits, 1);
assert_eq!((turn.hunks, turn.insertions, turn.deletions), (2, 7, 3));
assert_eq!(session, turn);
assert_eq!(turn_files, &["/repo/src/lib.rs"]);
assert_eq!(session_files, turn_files);
}
}