use crate::{
agent::steering::STEERING_HEADER,
output::redact_sensitive_text,
providers::{ChatMessage, ProviderConversationItem, ProviderToolResult},
sessions::{
BoundedReadError, Session, SessionEvent, SessionEventKind, SessionReadDiagnostic,
latest_valid_compaction_checkpoint,
},
};
use serde_json::{Value, json};
use std::{
fs,
io::{Read, Seek, SeekFrom},
path::{Path, PathBuf},
};
use super::{canonical_json, event_line, is_local_only_session_event};
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ConversationReplay {
pub items: Vec<ProviderConversationItem>,
pub legacy_lossy_events: Vec<String>,
pub session_read_diagnostics: Vec<SessionReadDiagnostic>,
pub replay_diagnostics: Vec<String>,
}
pub(crate) const REPLAY_TOOL_RESULT_OUTPUT_CHAR_LIMIT: usize = 24_000;
pub(crate) const REPLAY_JSONL_MAX_LINES: usize = 100_000;
pub(crate) const REPLAY_JSONL_MAX_BYTES: usize = 32 * 1024 * 1024;
const UNCOMMITTED_TURN_PROVIDER_VISIBLE_CHAR_LIMIT: usize = 128_000;
const FAILED_TURN_RECOVERY_SUMMARY_CHAR_LIMIT: usize = 12_000;
const FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT: usize = 4_000;
const FAILED_TURN_RECOVERY_TOOL_OUTPUT_CHAR_LIMIT: usize = 3_000;
const CHILD_SESSION_RECOVERY_MAX_LINES: usize = 200;
const CHILD_SESSION_RECOVERY_MAX_BYTES: usize = 128 * 1024;
fn sanitize_provider_response_item(mut item: Value) -> Value {
fn remove_encrypted_content(value: &mut Value) {
match value {
Value::Object(object) => {
object.remove("encrypted_content");
for child in object.values_mut() {
remove_encrypted_content(child);
}
}
Value::Array(items) => {
for item in items {
remove_encrypted_content(item);
}
}
_ => {}
}
}
remove_encrypted_content(&mut item);
item
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ConversationReplayLimits {
pub(crate) max_lines: usize,
pub(crate) max_bytes: usize,
}
impl Default for ConversationReplayLimits {
fn default() -> Self {
Self {
max_lines: REPLAY_JSONL_MAX_LINES,
max_bytes: REPLAY_JSONL_MAX_BYTES,
}
}
}
pub(crate) fn build_conversation_replay(
session: Option<&Session>,
) -> anyhow::Result<ConversationReplay> {
build_conversation_replay_with_limits(session, ConversationReplayLimits::default())
}
pub(crate) fn build_conversation_replay_with_limits(
session: Option<&Session>,
limits: ConversationReplayLimits,
) -> anyhow::Result<ConversationReplay> {
let Some(session) = session else {
return Ok(ConversationReplay {
items: Vec::new(),
legacy_lossy_events: Vec::new(),
session_read_diagnostics: Vec::new(),
replay_diagnostics: Vec::new(),
});
};
let tolerant = session
.read_events_tolerant_bounded(limits.max_lines, limits.max_bytes)
.map_err(|error| {
if error.downcast_ref::<BoundedReadError>().is_some() {
anyhow::anyhow!("session replay exceeded bounded history budget; durable session history was preserved; start /new and inspect or back up session data")
} else {
error
}
})?;
let sessions_root = session.path().parent().map(Path::to_path_buf);
let mut replay = build_conversation_replay_from_events_with_sessions_root(
session.id(),
&tolerant.events,
sessions_root.as_deref(),
);
replay.session_read_diagnostics = tolerant.diagnostics;
Ok(replay)
}
pub(crate) fn build_conversation_replay_from_events(
session_id: &str,
events: &[SessionEvent],
) -> ConversationReplay {
build_conversation_replay_from_events_with_sessions_root(session_id, events, None)
}
fn build_conversation_replay_from_events_with_sessions_root(
session_id: &str,
events: &[SessionEvent],
sessions_root: Option<&Path>,
) -> ConversationReplay {
let mut builder = ConversationReplayBuilder {
sessions_root: sessions_root.map(Path::to_path_buf),
..ConversationReplayBuilder::default()
};
let (checkpoint, mut replay_diagnostics) =
latest_valid_compaction_checkpoint(session_id, events);
let start_index = checkpoint
.as_ref()
.map(|checkpoint| checkpoint.cutoff_event_count)
.unwrap_or(0);
if let Some(checkpoint) = checkpoint {
builder.push_compaction_summary(&checkpoint.summary);
}
for event in events.iter().skip(start_index) {
builder.push_event(event);
}
builder.finish();
replay_diagnostics.extend(builder.replay_diagnostics);
ConversationReplay {
items: builder.items,
legacy_lossy_events: builder.legacy_lossy_events,
session_read_diagnostics: Vec::new(),
replay_diagnostics,
}
}
#[derive(Default)]
struct ConversationReplayBuilder {
items: Vec<ProviderConversationItem>,
legacy_lossy_events: Vec<String>,
replay_diagnostics: Vec<String>,
pending_assistant: String,
assistant_chunks_in_current_response: bool,
exact_provider_call_ids: std::collections::HashSet<String>,
pending_tool_call_ids: std::collections::HashSet<String>,
active_turn: Option<ReplayTurn>,
sessions_root: Option<PathBuf>,
}
#[derive(Default)]
struct ReplayTurn {
items: Vec<ProviderConversationItem>,
saw_completed_assistant: bool,
saw_provider_success_completion: bool,
saw_provider_failure_diagnostic: bool,
saw_turn_incomplete_marker: bool,
saw_turn_cancelled_marker: bool,
saw_turn_failed_marker: bool,
saw_turn_complete_marker: bool,
provider_failure_categories: Vec<&'static str>,
abort_recoveries: Vec<AbortRecovery>,
provider_visible_chars: usize,
original_provider_visible_chars: usize,
provider_activity_after_user: bool,
has_ttsr_retry_context: bool,
cwd: Option<PathBuf>,
tool_call_ids: std::collections::HashSet<String>,
}
#[derive(Default)]
struct AbortRecovery {
observed_reasoning_delta: bool,
unsafe_tool_call_progress: bool,
reasoning_preview: Option<String>,
partial_tool_call_summary: Option<String>,
}
impl ConversationReplayBuilder {
fn push_compaction_summary(&mut self, summary: &str) {
self.items.push(ProviderConversationItem::Message(ChatMessage::user(
format!("Session compaction summary. Earlier session events are represented only by this summary:\n\n{summary}"),
)));
}
fn push_event(&mut self, event: &SessionEvent) {
match event.kind() {
Some(SessionEventKind::AssistantChunk) => {
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
if !text.trim().is_empty() {
self.pending_assistant.push_str(text);
self.assistant_chunks_in_current_response = true;
}
} else {
self.push_legacy_note(event);
}
}
Some(SessionEventKind::AssistantOutput)
if !self.assistant_chunks_in_current_response =>
{
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
if let Some(turn) = self.active_turn.as_mut() {
turn.saw_completed_assistant = true;
}
if !text.trim().is_empty() {
self.push_message(ChatMessage::assistant(text));
}
} else {
self.push_legacy_note(event);
}
}
Some(SessionEventKind::AssistantOutput) => {
if let Some(turn) = self.active_turn.as_mut() {
turn.saw_completed_assistant = true;
}
}
Some(SessionEventKind::UserInput) => {
self.flush_assistant_chunks();
self.assistant_chunks_in_current_response = false;
if self.user_input_continues_active_turn(event) {
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
self.push_message(ChatMessage::user(text));
} else {
self.push_legacy_note(event);
}
return;
}
self.finalize_active_turn(false);
self.active_turn = Some(ReplayTurn {
cwd: Some(event.cwd.clone()),
..ReplayTurn::default()
});
if let Some(text) = event.payload.get("text").and_then(Value::as_str) {
self.push_message(ChatMessage::user(text));
} else {
self.push_legacy_note(event);
}
}
Some(SessionEventKind::ProviderResponseItem) => {
self.flush_assistant_chunks();
if let Some(item) = event
.payload
.get("item")
.cloned()
.map(sanitize_provider_response_item)
{
let item_type = item.get("type").and_then(Value::as_str);
if item_type == Some("function_call")
&& let Some(call_id) = item.get("call_id").and_then(Value::as_str)
{
self.register_tool_call_id(call_id);
}
if item_type == Some("function_call_output") {
let Some(call_id) = item.get("call_id").and_then(Value::as_str) else {
self.push_legacy_note(event);
return;
};
if !self.pending_tool_call_ids.remove(call_id) {
self.replay_diagnostics.push(format!(
"downgraded_orphan_provider_tool_result call_id={}",
sanitize_recovery_text(call_id, 256)
));
self.push_legacy_note(event);
return;
}
}
if provider_response_item_is_success_completion(&item)
&& let Some(turn) = self.active_turn.as_mut()
{
turn.saw_provider_success_completion = true;
}
self.push_replay_item(ProviderConversationItem::ResponseItem(item));
} else {
self.push_legacy_note(event);
}
}
Some(SessionEventKind::ToolCall) => {
self.flush_assistant_chunks();
self.push_tool_call(event);
}
Some(SessionEventKind::ToolResult) => {
self.flush_assistant_chunks();
self.push_tool_result(event);
}
Some(SessionEventKind::ProviderContextItem) => {
self.flush_assistant_chunks();
self.push_provider_context_item(event);
}
Some(SessionEventKind::Rewind) => {
self.flush_assistant_chunks();
if let Some(content) = crate::checkpoints::replay_rewind_context(&event.payload) {
self.push_message(ChatMessage::user(content));
} else {
self.push_legacy_note(event);
}
}
Some(SessionEventKind::Compaction | SessionEventKind::ReasoningSummary) => {}
Some(SessionEventKind::AbortRecovery) => {
if let Some(turn) = self.active_turn.as_mut() {
turn.abort_recoveries
.push(abort_recovery_from_payload(&event.payload));
}
}
Some(SessionEventKind::TtsrInjection) => {
if self.active_turn.is_some()
&& let Some(reminder) = event.payload.get("reminder").and_then(Value::as_str)
&& !reminder.trim().is_empty()
{
if let Some(turn) = self.active_turn.as_mut() {
turn.has_ttsr_retry_context = true;
}
self.push_message(ChatMessage::system(reminder));
}
}
Some(SessionEventKind::TurnStatus) => {
if let Some(turn) = self.active_turn.as_mut() {
match event.payload.get("status").and_then(Value::as_str) {
Some("incomplete") => turn.saw_turn_incomplete_marker = true,
Some("cancelled" | "canceled") => {
turn.saw_turn_cancelled_marker = true;
if !self.assistant_chunks_in_current_response
&& self.pending_assistant.trim().is_empty()
&& let Some(text) =
event.payload.get("assistant_text").and_then(Value::as_str)
&& !text.trim().is_empty()
{
self.pending_assistant.push_str(text);
self.assistant_chunks_in_current_response = true;
}
}
Some("failed") => {
turn.saw_turn_failed_marker = true;
if !self.assistant_chunks_in_current_response
&& self.pending_assistant.trim().is_empty()
&& let Some(text) =
event.payload.get("assistant_text").and_then(Value::as_str)
&& !text.trim().is_empty()
{
self.pending_assistant.push_str(text);
self.assistant_chunks_in_current_response = true;
}
}
Some("complete") => turn.saw_turn_complete_marker = true,
_ => {}
}
}
}
_ if is_local_only_session_event(event) => {
if let Some(category) = diagnostic_provider_failure_category(event)
&& let Some(turn) = self.active_turn.as_mut()
{
turn.saw_provider_failure_diagnostic = true;
if !turn.provider_failure_categories.contains(&category) {
turn.provider_failure_categories.push(category);
}
}
}
_ => {
self.flush_assistant_chunks();
self.push_legacy_note(event);
}
}
}
fn finish(&mut self) {
self.flush_assistant_chunks();
self.finalize_active_turn(true);
}
fn user_input_continues_active_turn(&self, event: &SessionEvent) -> bool {
let Some(text) = event.payload.get("text").and_then(Value::as_str) else {
return false;
};
let is_steering = text.starts_with(STEERING_HEADER);
if !(is_steering || auto_recovery_continue(event)) {
return false;
}
self.active_turn.as_ref().is_some_and(|turn| {
turn.provider_activity_after_user
&& (is_steering || !turn.saw_completed_assistant)
&& !turn.saw_provider_success_completion
&& !turn.saw_provider_failure_diagnostic
&& !turn.saw_turn_incomplete_marker
&& !turn.saw_turn_cancelled_marker
&& !turn.saw_turn_failed_marker
&& !turn.saw_turn_complete_marker
})
}
fn finalize_active_turn(&mut self, eof: bool) {
let Some(turn) = self.active_turn.take() else {
return;
};
for call_id in &turn.tool_call_ids {
self.pending_tool_call_ids.remove(call_id);
self.exact_provider_call_ids.remove(call_id);
}
let turn_incomplete = turn.saw_turn_incomplete_marker;
let turn_partial_terminal = turn.saw_turn_cancelled_marker || turn.saw_turn_failed_marker;
let committed = turn.saw_completed_assistant
|| turn.saw_provider_success_completion
|| turn.saw_turn_complete_marker;
let oversized =
turn.original_provider_visible_chars > UNCOMMITTED_TURN_PROVIDER_VISIBLE_CHAR_LIMIT;
let unsafe_tool_call_repair_failure = turn_has_unsafe_tool_call_progress(&turn);
if committed || (eof && turn.has_ttsr_retry_context) {
self.items.extend(turn.items);
} else if turn.provider_activity_after_user {
if turn_partial_terminal {
self.items
.extend(safe_durable_pre_failure_items(&turn.items));
if unsafe_tool_call_repair_failure {
self.replay_diagnostics.push(format!(
"replayed_partial_historical_turn_without_recovery_summary unsafe_tool_call_progress=true cancelled={} failed={} chars={}",
turn.saw_turn_cancelled_marker,
turn.saw_turn_failed_marker,
turn.original_provider_visible_chars
));
}
if !unsafe_tool_call_repair_failure
&& turn_needs_recovery_summary(&turn, self.sessions_root.as_deref())
&& let Some(summary) = failed_turn_recovery_summary(
&turn,
oversized,
self.sessions_root.as_deref(),
)
{
let summary_chars = summary.chars().count();
self.items
.push(ProviderConversationItem::Message(ChatMessage::user(
summary,
)));
self.replay_diagnostics.push(format!(
"recovered_partial_historical_turn cancelled={} failed={} chars={} summary_chars={}",
turn.saw_turn_cancelled_marker,
turn.saw_turn_failed_marker,
turn.original_provider_visible_chars,
summary_chars
));
}
self.replay_diagnostics.push(format!(
"replayed_partial_historical_turn cancelled={} failed={} chars={}",
turn.saw_turn_cancelled_marker,
turn.saw_turn_failed_marker,
turn.original_provider_visible_chars
));
} else if turn.saw_provider_failure_diagnostic {
if let Some(summary) =
failed_turn_recovery_summary(&turn, oversized, self.sessions_root.as_deref())
{
let summary_chars = summary.chars().count();
self.items
.push(ProviderConversationItem::Message(ChatMessage::user(
summary,
)));
self.replay_diagnostics.push(format!(
"recovered_failed_historical_turn provider_failure=true oversized={} chars={} limit={} summary_chars={}",
oversized,
turn.original_provider_visible_chars,
UNCOMMITTED_TURN_PROVIDER_VISIBLE_CHAR_LIMIT,
summary_chars
));
} else {
self.replay_diagnostics.push(format!(
"skipped_failed_historical_turn provider_failure=true oversized={} chars={} limit={}",
oversized,
turn.original_provider_visible_chars,
UNCOMMITTED_TURN_PROVIDER_VISIBLE_CHAR_LIMIT
));
}
} else if oversized {
self.replay_diagnostics.push(format!(
"skipped_failed_historical_turn provider_failure=false oversized=true chars={} limit={}",
turn.original_provider_visible_chars,
UNCOMMITTED_TURN_PROVIDER_VISIBLE_CHAR_LIMIT
));
} else if eof {
self.replay_diagnostics.push(format!(
"dropped_uncommitted_eof_provider_activity chars={}",
turn.original_provider_visible_chars
));
}
if eof
&& !turn.saw_provider_failure_diagnostic
&& !oversized
&& !turn_incomplete
&& !turn_partial_terminal
{
self.items.extend(pending_user_only_items(&turn.items));
} else if turn_incomplete {
self.replay_diagnostics.push(format!(
"dropped_incomplete_historical_turn provider_activity_after_user=true chars={}",
turn.original_provider_visible_chars
));
}
} else if turn_incomplete {
self.replay_diagnostics
.push("dropped_incomplete_historical_user_only_turn".to_string());
} else if turn_partial_terminal && !turn.abort_recoveries.is_empty() {
if turn_has_unsafe_tool_call_progress(&turn) {
self.items.extend(pending_user_only_items(&turn.items));
self.replay_diagnostics.push(format!(
"replayed_partial_historical_user_only_turn_without_recovery_summary unsafe_tool_call_progress=true cancelled={} failed={}",
turn.saw_turn_cancelled_marker, turn.saw_turn_failed_marker
));
} else if let Some(summary) =
failed_turn_recovery_summary(&turn, oversized, self.sessions_root.as_deref())
{
self.items
.push(ProviderConversationItem::Message(ChatMessage::user(
summary,
)));
self.replay_diagnostics.push(format!(
"recovered_partial_historical_user_only_turn cancelled={} failed={}",
turn.saw_turn_cancelled_marker, turn.saw_turn_failed_marker
));
}
} else if turn.saw_provider_failure_diagnostic || oversized {
self.replay_diagnostics.push(format!(
"skipped_failed_historical_user_only_turn provider_failure={} oversized={} chars={} limit={}",
turn.saw_provider_failure_diagnostic,
oversized,
turn.original_provider_visible_chars,
UNCOMMITTED_TURN_PROVIDER_VISIBLE_CHAR_LIMIT
));
} else {
self.items.extend(turn.items);
}
}
fn push_message(&mut self, message: ChatMessage) {
self.push_replay_item(ProviderConversationItem::Message(message));
}
fn push_replay_item(&mut self, item: ProviderConversationItem) {
if let Some(turn) = self.active_turn.as_mut() {
if !matches!(
item,
ProviderConversationItem::Message(ChatMessage {
role: crate::providers::MessageRole::User,
..
})
) {
turn.provider_activity_after_user = true;
}
let visible_chars = replay_item_visible_chars(&item);
turn.provider_visible_chars = turn.provider_visible_chars.saturating_add(visible_chars);
turn.original_provider_visible_chars = turn
.original_provider_visible_chars
.saturating_add(visible_chars);
turn.items.push(item);
} else {
self.items.push(item);
}
}
fn flush_assistant_chunks(&mut self) {
if self.pending_assistant.trim().is_empty() {
self.pending_assistant.clear();
return;
}
let text = std::mem::take(&mut self.pending_assistant);
self.push_message(ChatMessage::assistant(text));
}
fn register_tool_call_id(&mut self, call_id: &str) {
self.exact_provider_call_ids.insert(call_id.to_string());
self.pending_tool_call_ids.insert(call_id.to_string());
if let Some(turn) = self.active_turn.as_mut() {
turn.tool_call_ids.insert(call_id.to_string());
}
}
fn push_tool_call(&mut self, event: &SessionEvent) {
let Some(call_id) = event.payload.get("id").and_then(Value::as_str) else {
self.push_legacy_note(event);
return;
};
if call_id.trim().is_empty() || self.exact_provider_call_ids.contains(call_id) {
return;
}
let Some(name) = event.payload.get("name").and_then(Value::as_str) else {
self.push_legacy_note(event);
return;
};
let arguments = event
.payload
.get("arguments")
.cloned()
.unwrap_or(Value::Null);
self.register_tool_call_id(call_id);
self.push_replay_item(ProviderConversationItem::ResponseItem(json!({
"type": "function_call",
"call_id": call_id,
"name": name,
"arguments": arguments_json_text(&arguments),
"status": "completed",
})));
}
fn push_tool_result(&mut self, event: &SessionEvent) {
let Some(call_id) = event.payload.get("call_id").and_then(Value::as_str) else {
self.push_legacy_note(event);
return;
};
if call_id.trim().is_empty() {
self.push_legacy_note(event);
return;
}
if !self.pending_tool_call_ids.remove(call_id) {
self.replay_diagnostics.push(format!(
"downgraded_orphan_session_tool_result call_id={}",
sanitize_recovery_text(call_id, 256)
));
self.push_legacy_note(event);
return;
}
let result = &event.payload["result"];
let output = result
.get("content")
.and_then(Value::as_str)
.or_else(|| result.get("output").and_then(Value::as_str));
let Some(output) = output else {
self.push_legacy_note(event);
return;
};
let tool_name = result
.get("tool_name")
.and_then(Value::as_str)
.unwrap_or("unknown")
.to_string();
let success = result
.get("success")
.and_then(Value::as_bool)
.unwrap_or(false);
let original_chars = output.chars().count();
let output = compact_historical_tool_output(output, &tool_name, success);
if original_chars > REPLAY_TOOL_RESULT_OUTPUT_CHAR_LIMIT {
self.replay_diagnostics.push(format!(
"compacted_historical_tool_result_output tool={tool_name} success={success} chars={original_chars} limit={REPLAY_TOOL_RESULT_OUTPUT_CHAR_LIMIT}"
));
}
self.push_tool_result_item(
ProviderToolResult {
call_id: call_id.to_string(),
tool_name,
success,
output,
},
original_chars,
);
}
fn push_provider_context_item(&mut self, event: &SessionEvent) {
let role = event.payload.get("role").and_then(Value::as_str);
let Some(content) = event.payload.get("content").and_then(Value::as_str) else {
self.push_legacy_note(event);
return;
};
match role {
Some("user") => self.push_message(ChatMessage::user(content)),
_ => self.push_legacy_note(event),
}
}
fn push_tool_result_item(&mut self, result: ProviderToolResult, original_chars: usize) {
let item = ProviderConversationItem::ToolResult(result);
if let Some(turn) = self.active_turn.as_mut() {
turn.provider_activity_after_user = true;
turn.provider_visible_chars = turn
.provider_visible_chars
.saturating_add(replay_item_visible_chars(&item));
turn.original_provider_visible_chars = turn
.original_provider_visible_chars
.saturating_add(original_chars);
turn.items.push(item);
} else {
self.items.push(item);
}
}
fn push_legacy_note(&mut self, event: &SessionEvent) {
let content = event_line(event);
self.legacy_lossy_events.push(event.event_type.clone());
self.push_replay_item(ProviderConversationItem::LegacyReplayNote {
event_type: event.event_type.clone(),
content,
});
}
}
fn auto_recovery_continue(event: &SessionEvent) -> bool {
event.payload.get("auto_recovery").and_then(Value::as_bool) == Some(true)
&& event.payload.get("reason").and_then(Value::as_str)
== Some("incomplete_semantic_progress_timeout")
&& event.payload.get("text").and_then(Value::as_str) == Some("Continue")
}
fn safe_durable_pre_failure_items(
items: &[ProviderConversationItem],
) -> Vec<ProviderConversationItem> {
let completed_call_ids = items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::ToolResult(result) => Some(result.call_id.as_str()),
ProviderConversationItem::ResponseItem(value)
if value.get("type").and_then(Value::as_str) == Some("function_call_output") =>
{
value.get("call_id").and_then(Value::as_str)
}
_ => None,
})
.collect::<std::collections::HashSet<_>>();
items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::Message(message)
if matches!(
message.role,
crate::providers::MessageRole::User | crate::providers::MessageRole::Assistant
) =>
{
Some(ProviderConversationItem::Message(message.clone()))
}
ProviderConversationItem::ResponseItem(value)
if matches!(
value.get("type").and_then(Value::as_str),
Some("function_call" | "function_call_output")
) =>
{
let call_id = value.get("call_id").and_then(Value::as_str)?;
completed_call_ids
.contains(call_id)
.then(|| ProviderConversationItem::ResponseItem(value.clone()))
}
ProviderConversationItem::ToolResult(result)
if completed_call_ids.contains(result.call_id.as_str()) =>
{
Some(item.clone())
}
_ => None,
})
.collect()
}
fn pending_user_only_items(items: &[ProviderConversationItem]) -> Vec<ProviderConversationItem> {
items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::User =>
{
Some(ProviderConversationItem::Message(message.clone()))
}
_ => None,
})
.collect()
}
fn arguments_json_text(arguments: &Value) -> String {
match arguments {
Value::String(text) => text.clone(),
value => value.to_string(),
}
}
fn provider_response_item_is_success_completion(item: &Value) -> bool {
let item_type = item.get("type").and_then(Value::as_str);
let status = item.get("status").and_then(Value::as_str);
matches!(
status,
Some("completed" | "complete" | "succeeded" | "success")
) && matches!(
item_type,
Some("message" | "output_text" | "assistant_message")
)
}
fn diagnostic_provider_failure_category(event: &SessionEvent) -> Option<&'static str> {
if event.kind() != Some(SessionEventKind::Diagnostic) {
return None;
}
let mut text = event.payload.to_string();
text.make_ascii_lowercase();
if [
"request or response body error",
"request body",
"response body",
"failed to read provider error body",
]
.iter()
.any(|needle| text.contains(needle))
{
return Some("provider_body_error");
}
if text.contains("response.failed") || text.contains("failed or incomplete response") {
return Some("provider_response_failed");
}
if text.contains("response.incomplete") || text.contains("incomplete response") {
return Some("provider_response_incomplete");
}
if text.contains("provider stream no semantic progress before timeout")
|| text.contains("provider stream idle timeout")
{
return Some("provider_stream_timeout");
}
if text.contains("provider stream ended") {
return Some("provider_stream_ended");
}
None
}
fn abort_recovery_from_payload(payload: &Value) -> AbortRecovery {
AbortRecovery {
observed_reasoning_delta: payload
.get("observed_reasoning_delta")
.and_then(Value::as_bool)
.unwrap_or(false),
unsafe_tool_call_progress: payload
.get("unsafe_tool_call_progress")
.and_then(Value::as_bool)
.unwrap_or(false),
reasoning_preview: payload
.get("reasoning_preview")
.and_then(Value::as_str)
.map(|text| sanitize_recovery_text(text, FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT)),
partial_tool_call_summary: payload
.get("partial_tool_call_summary")
.and_then(Value::as_str)
.map(|text| sanitize_recovery_text(text, 512)),
}
}
fn turn_needs_recovery_summary(turn: &ReplayTurn, sessions_root: Option<&Path>) -> bool {
!turn.abort_recoveries.is_empty()
|| failed_turn_tool_results(&turn.items)
.iter()
.any(|result| subagent_child_recovery_summary(&result.output, sessions_root).is_some())
|| !failed_turn_tool_calls_without_results(&turn.items).is_empty()
|| !failed_turn_legacy_notes(&turn.items).is_empty()
}
fn turn_has_unsafe_tool_call_progress(turn: &ReplayTurn) -> bool {
turn.abort_recoveries
.iter()
.any(|recovery| recovery.unsafe_tool_call_progress)
}
fn failed_turn_recovery_summary(
turn: &ReplayTurn,
oversized: bool,
sessions_root: Option<&Path>,
) -> Option<String> {
if turn.items.is_empty() {
return None;
}
let mut summary = String::new();
summary.push_str(
"Session recovery summary for failed historical turn; not exact structured replay.\n",
);
summary.push_str(
"Completed paired tool calls/results are replayed; unpaired or unsafe structured progress is summarized as best-effort context only.\n",
);
if !turn.abort_recoveries.is_empty() {
summary.push_str("Abort recovery facts:\n");
for recovery in &turn.abort_recoveries {
summary.push_str("- reasoning_delta_seen=");
summary.push_str(&recovery.observed_reasoning_delta.to_string());
summary.push_str(" unsafe_tool_call_progress=");
summary.push_str(&recovery.unsafe_tool_call_progress.to_string());
summary.push('\n');
if let Some(preview) = &recovery.reasoning_preview {
summary.push_str(" reasoning preview:\n");
summary.push_str(&indent_recovery_block(&sanitize_recovery_text(
preview,
FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT,
)));
}
if let Some(tool_summary) = &recovery.partial_tool_call_summary {
summary.push_str(" tool progress: ");
summary.push_str(&sanitize_recovery_text(tool_summary, 512));
summary.push('\n');
}
}
}
if let Some(last_category) = turn.provider_failure_categories.last() {
summary.push_str("Last diagnostic/error category: ");
summary.push_str(last_category);
summary.push('\n');
}
if !turn.provider_failure_categories.is_empty() {
summary.push_str("Diagnostic categories: ");
summary.push_str(&turn.provider_failure_categories.join(", "));
summary.push('\n');
}
if let Some(cwd) = &turn.cwd {
summary.push_str("CWD: ");
summary.push_str(&sanitize_recovery_text(
&cwd.display().to_string(),
FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT,
));
summary.push('\n');
}
let objectives = failed_turn_user_objectives(&turn.items);
if !objectives.is_empty() {
summary.push_str("User objective(s):\n");
for objective in objectives {
summary.push_str("- ");
summary.push_str(&sanitize_recovery_text(
&objective,
FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT,
));
summary.push('\n');
}
}
let tool_results = failed_turn_tool_results(&turn.items);
if !tool_results.is_empty() {
summary.push_str("Tool/subagent results:\n");
for result in tool_results {
summary.push_str("- tool=");
summary.push_str(&sanitize_recovery_text(&result.tool_name, 256));
summary.push_str(" success=");
summary.push_str(&result.success.to_string());
summary.push('\n');
let child_summary = subagent_child_recovery_summary(&result.output, sessions_root);
if oversized {
summary.push_str(
" output: [omitted because failed turn exceeded recovery safety limit]\n",
);
} else if child_summary.is_some() {
summary.push_str(
" output: [omitted because child session recovery snapshot is available]\n",
);
} else {
summary.push_str(" output:\n");
summary.push_str(&indent_recovery_block(&sanitize_recovery_text(
&result.output,
FAILED_TURN_RECOVERY_TOOL_OUTPUT_CHAR_LIMIT,
)));
}
if let Some(child_summary) = child_summary {
summary.push_str(&child_summary);
}
}
}
let tool_calls_without_results = failed_turn_tool_calls_without_results(&turn.items);
if !tool_calls_without_results.is_empty() {
summary.push_str("Tool calls without recovered result:\n");
for name in tool_calls_without_results {
summary.push_str("- tool=");
summary.push_str(&sanitize_recovery_text(&name, 256));
summary.push('\n');
}
}
let legacy_notes = failed_turn_legacy_notes(&turn.items);
if !legacy_notes.is_empty() {
summary.push_str("Downgraded/orphan session events:\n");
for note in legacy_notes {
summary.push_str("- ");
summary.push_str(&sanitize_recovery_text(
¬e,
FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT,
));
summary.push('\n');
}
}
summary.push_str(
"Pending next action: Continue from this recovered context and the later user prompt.\n",
);
let summary = sanitize_recovery_text(&summary, FAILED_TURN_RECOVERY_SUMMARY_CHAR_LIMIT);
if summary.trim().is_empty() {
None
} else {
Some(summary)
}
}
fn subagent_child_recovery_summary(output: &str, sessions_root: Option<&Path>) -> Option<String> {
let sessions_root = sessions_root?;
let value: Value = serde_json::from_str(output).ok()?;
let results = value.get("results")?.as_array()?;
let mut summary = String::new();
for result in results {
if result.get("status").and_then(Value::as_str) != Some("failed") {
continue;
}
let Some(child) = child_session_file_recovery_snapshot(result, sessions_root) else {
continue;
};
summary.push_str(" child session recovery snapshot:\n");
summary.push_str(&indent_recovery_block(&child));
}
(!summary.trim().is_empty()).then_some(summary)
}
fn child_session_file_recovery_snapshot(result: &Value, sessions_root: &Path) -> Option<String> {
let id = result
.get("session_id")
.and_then(Value::as_str)
.or_else(|| {
result
.get("session_path")
.and_then(Value::as_str)
.and_then(|path| Path::new(path).file_stem())
.and_then(std::ffi::OsStr::to_str)
})?;
crate::sessions::validate_session_id(id.to_string()).ok()?;
let path = sessions_root.join("subagents").join(format!("{id}.jsonl"));
let trusted_root = fs::canonicalize(sessions_root.join("subagents")).ok()?;
let trusted_path = fs::canonicalize(&path).ok()?;
if !trusted_path.starts_with(&trusted_root) {
return None;
}
let persisted_path = result.get("session_path").and_then(Value::as_str)?;
let persisted_path = fs::canonicalize(persisted_path).ok()?;
if persisted_path != trusted_path {
return None;
}
let lines = child_session_tail_lines(
&trusted_path,
CHILD_SESSION_RECOVERY_MAX_LINES,
CHILD_SESSION_RECOVERY_MAX_BYTES,
)?;
let mut assistant_chunks = String::new();
let mut authoritative_output = None;
let mut abort_recoveries = Vec::new();
for line in lines {
let Ok(event) = serde_json::from_str::<SessionEvent>(&line) else {
continue;
};
match event.kind() {
Some(SessionEventKind::AssistantChunk) => {
if let Some(text) = event.payload.get("text").and_then(Value::as_str)
&& !text.trim().is_empty()
{
assistant_chunks.push_str(text);
}
}
Some(SessionEventKind::AssistantOutput) => {
if let Some(text) = event.payload.get("text").and_then(Value::as_str)
&& !text.trim().is_empty()
{
authoritative_output = Some(text.to_string());
}
}
Some(SessionEventKind::AbortRecovery) => {
abort_recoveries.push(event.payload.to_string());
}
_ => {}
}
}
let mut snapshot = authoritative_output.unwrap_or(assistant_chunks);
for recovery in abort_recoveries {
if !snapshot.is_empty() {
snapshot.push('\n');
}
snapshot.push_str("abort recovery: ");
snapshot.push_str(&recovery);
}
let snapshot = sanitize_recovery_text(&snapshot, FAILED_TURN_RECOVERY_FIELD_CHAR_LIMIT);
(!snapshot.trim().is_empty()).then_some(snapshot)
}
fn child_session_tail_lines(
path: &Path,
max_lines: usize,
max_bytes: usize,
) -> Option<Vec<String>> {
if max_lines == 0 || max_bytes == 0 {
return Some(Vec::new());
}
let mut file = fs::File::open(path).ok()?;
let file_len = file.metadata().ok()?.len();
let max_bytes_u64 = u64::try_from(max_bytes).ok()?;
let start = file_len.saturating_sub(max_bytes_u64);
file.seek(SeekFrom::Start(start)).ok()?;
let mut bytes = Vec::with_capacity(usize::try_from(file_len - start).ok()?);
file.take(max_bytes_u64).read_to_end(&mut bytes).ok()?;
let text = String::from_utf8_lossy(&bytes);
let lines = if start > 0 {
text.split_once('\n').map_or("", |(_, rest)| rest)
} else {
text.as_ref()
};
Some(
lines
.lines()
.rev()
.take(max_lines)
.map(ToString::to_string)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect(),
)
}
fn failed_turn_user_objectives(items: &[ProviderConversationItem]) -> Vec<String> {
items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::User =>
{
Some(message.content.clone())
}
_ => None,
})
.collect()
}
fn failed_turn_tool_results(items: &[ProviderConversationItem]) -> Vec<&ProviderToolResult> {
items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::ToolResult(result) => Some(result),
_ => None,
})
.collect()
}
fn failed_turn_tool_calls_without_results(items: &[ProviderConversationItem]) -> Vec<String> {
let result_call_ids = items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::ToolResult(result) => Some(result.call_id.as_str()),
_ => None,
})
.collect::<std::collections::HashSet<_>>();
items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::ResponseItem(value)
if value.get("type").and_then(Value::as_str) == Some("function_call") =>
{
let call_id = value
.get("call_id")
.and_then(Value::as_str)
.unwrap_or_default();
if result_call_ids.contains(call_id) {
None
} else {
value
.get("name")
.and_then(Value::as_str)
.map(ToString::to_string)
}
}
_ => None,
})
.collect()
}
fn failed_turn_legacy_notes(items: &[ProviderConversationItem]) -> Vec<String> {
items
.iter()
.filter_map(|item| match item {
ProviderConversationItem::LegacyReplayNote {
event_type,
content,
} => Some(format!("{event_type}: {content}")),
_ => None,
})
.collect()
}
fn indent_recovery_block(text: &str) -> String {
let mut indented = text
.lines()
.map(|line| format!(" {line}"))
.collect::<Vec<_>>()
.join("\n");
indented.push('\n');
indented
}
fn sanitize_recovery_text(text: &str, char_limit: usize) -> String {
let normalized = text
.chars()
.map(|ch| {
if ch.is_control() && !matches!(ch, '\n' | '\t') {
' '
} else {
ch
}
})
.collect::<String>();
let redacted = redact_sensitive_text(normalized.trim());
limit_recovery_chars(&redacted, char_limit)
}
fn limit_recovery_chars(text: &str, char_limit: usize) -> String {
let char_count = text.chars().count();
if char_count <= char_limit {
return text.to_string();
}
let head = char_limit.saturating_sub(160);
let prefix: String = text.chars().take(head).collect();
format!(
"{prefix}\n[recovery summary truncated: original_chars={char_count} limit={char_limit}]"
)
}
fn replay_item_visible_chars(item: &ProviderConversationItem) -> usize {
match item {
ProviderConversationItem::Message(message) => message.content.chars().count(),
ProviderConversationItem::ResponseItem(value) => canonical_json(value).chars().count(),
ProviderConversationItem::ToolResult(result) => result.output.chars().count(),
ProviderConversationItem::LegacyReplayNote { content, .. } => content.chars().count(),
}
}
pub(crate) fn compact_historical_tool_output(
output: &str,
tool_name: &str,
success: bool,
) -> String {
let original_chars = output.chars().count();
if original_chars <= REPLAY_TOOL_RESULT_OUTPUT_CHAR_LIMIT {
return output.to_string();
}
let edge_chars = REPLAY_TOOL_RESULT_OUTPUT_CHAR_LIMIT / 4;
let prefix: String = output.chars().take(edge_chars).collect();
let suffix_start = output
.char_indices()
.nth_back(edge_chars.saturating_sub(1))
.map(|(index, _)| index)
.unwrap_or(output.len());
let suffix = &output[suffix_start..];
format!(
"[historical replay compacted tool result: tool={tool_name} success={success} original_chars={original_chars} limit={REPLAY_TOOL_RESULT_OUTPUT_CHAR_LIMIT}]\n{prefix}\n[... historical replay output omitted ...]\n{suffix}"
)
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::Utc;
use serde_json::json;
fn event(kind: SessionEventKind, payload: Value) -> SessionEvent {
SessionEvent {
event_type: kind.as_str().to_string(),
timestamp: Utc::now(),
session_id: "session".to_string(),
cwd: PathBuf::from("/tmp/replay-test"),
session_path: Some(PathBuf::from("/tmp/replay-test")),
payload,
}
}
#[test]
fn ttsr_injection_replays_hidden_system_reminder() {
let replay = build_conversation_replay_from_events(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"do safe thing"})),
event(
SessionEventKind::TtsrInjection,
json!({
"schema_version": 1,
"turn_index": 0,
"request_sequence": 0,
"rule_source": "builtin",
"rule_pattern": "danger",
"matched_text_redacted": "[REDACTED]",
"reminder": "System TTSR reminder",
"aborted_turn": 0
}),
),
],
);
assert!(replay.legacy_lossy_events.is_empty());
assert!(replay.items.iter().any(|item| matches!(
item,
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::System
&& message.content == "System TTSR reminder"
)));
assert!(!replay.items.iter().any(|item| matches!(
item,
ProviderConversationItem::LegacyReplayNote { event_type, .. } if event_type == "ttsr_injection"
)));
}
#[test]
fn rewind_events_add_sanitized_context_only() {
let replay = build_conversation_replay_from_events(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"change files"})),
event(SessionEventKind::AssistantOutput, json!({"text":"done"})),
event(
SessionEventKind::Rewind,
json!({
"target_turn": 1,
"latest_turn": 1,
"paths": [
{"path":"src/lib.rs", "status":"restored"},
{"path":"../auth.json", "status":"skip_unavailable"}
],
"counts": {"Restored": 1}
}),
),
],
);
let summary = replay
.items
.iter()
.find_map(|item| match item {
ProviderConversationItem::Message(message)
if message.content.contains("Local filesystem rewind applied") =>
{
Some(message.content.as_str())
}
_ => None,
})
.unwrap();
assert!(summary.contains("src/lib.rs"), "{summary}");
assert!(summary.contains("<redacted>"), "{summary}");
assert!(!summary.contains("auth.json"), "{summary}");
assert!(!summary.contains("old bytes"), "{summary}");
assert!(!summary.contains("@@"), "{summary}");
}
#[test]
fn failed_empty_terminal_status_retains_user_objective() {
let replay = build_conversation_replay_from_events(
"session",
&[
event(
SessionEventKind::UserInput,
json!({"text":"keep objective"}),
),
event(
SessionEventKind::TurnStatus,
json!({"status":"failed","assistant_text":""}),
),
],
);
assert!(replay.items.iter().any(|item| matches!(
item,
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::User
&& message.content == "keep objective"
)));
}
#[test]
fn unsafe_abort_recovery_without_assistant_items_replays_user_without_summary() {
let replay = build_conversation_replay_from_events(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"recover me"})),
event(
SessionEventKind::TurnStatus,
json!({"status":"cancelled","assistant_text":""}),
),
event(
SessionEventKind::AbortRecovery,
json!({
"observed_reasoning_delta": true,
"unsafe_tool_call_progress": true,
"reasoning_preview": "safe preview",
"partial_tool_call_summary": "partial provider tool-call progress observed; raw arguments omitted"
}),
),
],
);
assert!(replay.items.iter().any(|item| matches!(
item,
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::User
&& message.content == "recover me"
)));
let replay_debug = format!("{replay:?}");
assert!(!replay_debug.contains("Session recovery summary for failed historical turn"));
assert!(!replay_debug.contains("Abort recovery facts"));
assert!(!replay_debug.contains("raw arguments omitted"));
}
#[test]
fn partial_tool_call_progress_is_summarized_not_structured_replayed() {
let replay = build_conversation_replay_from_events(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"call tool"})),
event(
SessionEventKind::ToolCall,
json!({"id":"call_1","name":"read","arguments":{"path":"file.txt"}}),
),
event(
SessionEventKind::TurnStatus,
json!({"status":"failed","assistant_text":""}),
),
],
);
assert!(!replay.items.iter().any(|item| matches!(
item,
ProviderConversationItem::ResponseItem(value)
if value.get("type").and_then(Value::as_str) == Some("function_call")
)));
assert!(replay.items.iter().any(|item| matches!(
item,
ProviderConversationItem::Message(message)
if message.content.contains("Tool calls without recovered result")
&& message.content.contains("read")
)));
}
#[test]
fn failed_subagents_tool_result_recovers_child_session_snapshot() {
let temp = tempfile::TempDir::new().unwrap();
let child_dir = temp.path().join("subagents");
fs::create_dir_all(&child_dir).unwrap();
let child_path = child_dir.join("child.jsonl");
let child_event = SessionEvent {
event_type: SessionEventKind::AssistantChunk.as_str().to_string(),
timestamp: Utc::now(),
session_id: "child".to_string(),
cwd: temp.path().to_path_buf(),
session_path: Some(temp.path().to_path_buf()),
payload: json!({"text":"child checkpoint"}),
};
let child_output_event = SessionEvent {
event_type: SessionEventKind::AssistantOutput.as_str().to_string(),
timestamp: Utc::now(),
session_id: "child".to_string(),
cwd: temp.path().to_path_buf(),
session_path: Some(temp.path().to_path_buf()),
payload: json!({"text":" \t"}),
};
fs::write(
&child_path,
format!(
"{}\n{}\n",
serde_json::to_string(&child_event).unwrap(),
serde_json::to_string(&child_output_event).unwrap()
),
)
.unwrap();
let tool_output = json!({
"summary":{"total":1,"completed":0,"failed":1},
"results":[{
"id":"g1",
"status":"failed",
"intent":"child",
"agent":null,
"identity":null,
"cwd": temp.path().to_string_lossy(),
"session_id":"child",
"session_path": child_path.to_string_lossy(),
"changed_files":[],
"output":"child checkpoint",
"output_truncated":false,
"error":"planned"
}]
})
.to_string();
let replay = build_conversation_replay_from_events_with_sessions_root(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"parent"})),
event(
SessionEventKind::ToolCall,
json!({"id":"subagents_1","name":"subagents","arguments":{}}),
),
event(
SessionEventKind::ToolResult,
json!({"call_id":"subagents_1","result":{"tool_name":"subagents","success":false,"content":tool_output}}),
),
event(
SessionEventKind::TurnStatus,
json!({"status":"failed","assistant_text":""}),
),
],
Some(temp.path()),
);
let summary = replay
.items
.iter()
.find_map(|item| match item {
ProviderConversationItem::Message(message)
if message.content.contains("child session recovery snapshot") =>
{
Some(message.content.as_str())
}
_ => None,
})
.unwrap();
assert!(summary.contains("child checkpoint"), "{summary}");
assert_eq!(summary.matches("child checkpoint").count(), 1, "{summary}");
}
#[test]
fn failed_subagents_tool_result_ignores_session_path_outside_trusted_root() {
let temp = tempfile::TempDir::new().unwrap();
let trusted_child_dir = temp.path().join("subagents");
fs::create_dir_all(&trusted_child_dir).unwrap();
let trusted_child_path = trusted_child_dir.join("child.jsonl");
fs::write(&trusted_child_path, "").unwrap();
let attacker_dir = tempfile::TempDir::new().unwrap();
let attacker_path = attacker_dir.path().join("child.jsonl");
let attacker_event = SessionEvent {
event_type: SessionEventKind::AssistantChunk.as_str().to_string(),
timestamp: Utc::now(),
session_id: "child".to_string(),
cwd: attacker_dir.path().to_path_buf(),
session_path: Some(attacker_path.clone()),
payload: json!({"text":"malicious child checkpoint"}),
};
fs::write(
&attacker_path,
format!("{}\n", serde_json::to_string(&attacker_event).unwrap()),
)
.unwrap();
let tool_output = json!({
"summary":{"total":1,"completed":0,"failed":1},
"results":[{
"id":"g1",
"status":"failed",
"intent":"child",
"agent":null,
"identity":null,
"cwd": attacker_dir.path().to_string_lossy(),
"session_id":"child",
"session_path": attacker_path.to_string_lossy(),
"changed_files":[],
"output":"",
"output_truncated":false,
"error":"planned"
}]
})
.to_string();
let replay = build_conversation_replay_from_events_with_sessions_root(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"parent"})),
event(
SessionEventKind::ToolCall,
json!({"id":"subagents_1","name":"subagents","arguments":{}}),
),
event(
SessionEventKind::ToolResult,
json!({"call_id":"subagents_1","result":{"tool_name":"subagents","success":false,"content":tool_output}}),
),
event(
SessionEventKind::TurnStatus,
json!({"status":"failed","assistant_text":""}),
),
],
Some(temp.path()),
);
let replay_debug = format!("{replay:?}");
assert!(
!replay_debug.contains("child session recovery snapshot"),
"{replay_debug}"
);
assert!(
!replay_debug.contains("malicious child checkpoint"),
"{replay_debug}"
);
}
#[test]
fn failed_subagents_tool_result_without_sessions_root_does_not_trust_child_path() {
let temp = tempfile::TempDir::new().unwrap();
let child_dir = temp.path().join("subagents");
fs::create_dir_all(&child_dir).unwrap();
let child_path = child_dir.join("child.jsonl");
let child_event = SessionEvent {
event_type: SessionEventKind::AssistantChunk.as_str().to_string(),
timestamp: Utc::now(),
session_id: "child".to_string(),
cwd: temp.path().to_path_buf(),
session_path: Some(temp.path().to_path_buf()),
payload: json!({"text":"untrusted child checkpoint"}),
};
fs::write(
&child_path,
format!("{}\n", serde_json::to_string(&child_event).unwrap()),
)
.unwrap();
let tool_output = json!({
"summary":{"total":1,"completed":0,"failed":1},
"results":[{
"id":"g1",
"status":"failed",
"intent":"child",
"session_id":"child",
"session_path": child_path.to_string_lossy(),
"error":"planned"
}]
})
.to_string();
let replay = build_conversation_replay_from_events(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"parent"})),
event(
SessionEventKind::ToolCall,
json!({"id":"subagents_1","name":"subagents","arguments":{}}),
),
event(
SessionEventKind::ToolResult,
json!({"call_id":"subagents_1","result":{"tool_name":"subagents","success":false,"content":tool_output}}),
),
event(
SessionEventKind::TurnStatus,
json!({"status":"failed","assistant_text":""}),
),
],
);
let replay_debug = format!("{replay:?}");
assert!(
!replay_debug.contains("child session recovery snapshot"),
"{replay_debug}"
);
assert!(
!replay_debug.contains("untrusted child checkpoint"),
"{replay_debug}"
);
}
#[test]
fn oversized_child_session_snapshot_reads_only_bounded_tail() {
let temp = tempfile::TempDir::new().unwrap();
let child_dir = temp.path().join("subagents");
fs::create_dir_all(&child_dir).unwrap();
let child_path = child_dir.join("child.jsonl");
let old_event = SessionEvent {
event_type: SessionEventKind::AssistantChunk.as_str().to_string(),
timestamp: Utc::now(),
session_id: "child".to_string(),
cwd: temp.path().to_path_buf(),
session_path: Some(temp.path().to_path_buf()),
payload: json!({"text":"outside-cap-secret-should-not-appear"}),
};
let recent_event = SessionEvent {
event_type: SessionEventKind::AssistantChunk.as_str().to_string(),
timestamp: Utc::now(),
session_id: "child".to_string(),
cwd: temp.path().to_path_buf(),
session_path: Some(temp.path().to_path_buf()),
payload: json!({"text":"recent child checkpoint"}),
};
let mut content = String::new();
content.push_str(&serde_json::to_string(&old_event).unwrap());
content.push('\n');
content.push_str(
&"padding outside snapshot window\n".repeat(CHILD_SESSION_RECOVERY_MAX_BYTES / 4),
);
content.push_str(&serde_json::to_string(&recent_event).unwrap());
content.push('\n');
fs::write(&child_path, content).unwrap();
assert!(
fs::metadata(&child_path).unwrap().len()
> u64::try_from(CHILD_SESSION_RECOVERY_MAX_BYTES).unwrap()
);
let tool_output = json!({
"summary":{"total":1,"completed":0,"failed":1},
"results":[{
"id":"g1",
"status":"failed",
"intent":"child",
"agent":null,
"identity":null,
"cwd": temp.path().to_string_lossy(),
"session_id":"child",
"session_path": child_path.to_string_lossy(),
"changed_files":[],
"output":"",
"output_truncated":false,
"error":"planned"
}]
})
.to_string();
let replay = build_conversation_replay_from_events_with_sessions_root(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"parent"})),
event(
SessionEventKind::ToolCall,
json!({"id":"subagents_1","name":"subagents","arguments":{}}),
),
event(
SessionEventKind::ToolResult,
json!({"call_id":"subagents_1","result":{"tool_name":"subagents","success":false,"content":tool_output}}),
),
event(
SessionEventKind::TurnStatus,
json!({"status":"failed","assistant_text":""}),
),
],
Some(temp.path()),
);
let summary = replay
.items
.iter()
.find_map(|item| match item {
ProviderConversationItem::Message(message)
if message.content.contains("child session recovery snapshot") =>
{
Some(message.content.as_str())
}
_ => None,
})
.unwrap();
assert!(summary.contains("recent child checkpoint"), "{summary}");
assert!(
!summary.contains("outside-cap-secret-should-not-appear"),
"{summary}"
);
}
#[test]
fn successful_turn_replay_stays_exact() {
let replay = build_conversation_replay_from_events(
"session",
&[
event(SessionEventKind::UserInput, json!({"text":"hello"})),
event(SessionEventKind::AssistantOutput, json!({"text":"world"})),
],
);
assert_eq!(replay.items.len(), 2);
assert!(matches!(
&replay.items[0],
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::User && message.content == "hello"
));
assert!(matches!(
&replay.items[1],
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::Assistant && message.content == "world"
));
}
}