diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 99ce055..045fd92 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -183,6 +183,32 @@ jobs: rustup component add clippy rustfmt - name: Install latest coding agents run: npm install --prefix "${{ runner.temp }}/coding-agents" --no-save --no-package-lock --no-audit --no-fund --cache "${{ runner.temp }}/npm-cache" @openai/codex@latest @anthropic-ai/claude-code@latest opencode-ai@latest @earendil-works/pi-coding-agent@latest + - name: Ensure Codex platform binary is available + shell: bash + run: | + set -euo pipefail + agent_dir="${{ runner.temp }}/coding-agents" + codex_version="$(node -p 'require(process.argv[1]).version' "$(npm root --prefix "$agent_dir")/@openai/codex/package.json")" + codex_platform="$(node -p '`${process.platform}-${process.arch}`')" + platform_dir="$agent_dir/node_modules/@openai/codex-$codex_platform" + platform_spec="@openai/codex-$codex_platform@npm:@openai/codex@$codex_version-$codex_platform" + if test -d "$platform_dir"; then + exit 0 + fi + for attempt in 1 2 3 4 5 6; do + if npm install --prefix "$agent_dir" --no-save --no-package-lock --no-audit --no-fund \ + --cache "${{ runner.temp }}/npm-cache" \ + "@openai/codex@$codex_version" "$platform_spec" \ + @anthropic-ai/claude-code@latest opencode-ai@latest \ + @earendil-works/pi-coding-agent@latest; then + exit 0 + fi + if test "$attempt" = 6; then + echo "Codex platform package remained unavailable: $platform_spec" >&2 + exit 1 + fi + sleep 10 + done - name: Set up pnpm uses: pnpm/action-setup@fc06bc1257f339d1d5d8b3a19a8cae5388b55320 # v5.0.0 with: diff --git a/bt-daemon/src/journal.rs b/bt-daemon/src/journal.rs index be0448d..e91b2d9 100644 --- a/bt-daemon/src/journal.rs +++ b/bt-daemon/src/journal.rs @@ -211,6 +211,7 @@ pub async fn gc_old_managed_runs(data_dir: &Path, max_age: std::time::Duration) pub struct JournalWriter { file: tokio::fs::File, position: u64, + pi_context: Vec, } impl JournalWriter { @@ -224,7 +225,11 @@ impl JournalWriter { .open(path) .await?; let position = file.metadata().await?.len(); - Ok(Self { file, position }) + Ok(Self { + file, + position, + pi_context: Vec::new(), + }) } pub(crate) fn position(&self) -> u64 { @@ -241,7 +246,9 @@ impl JournalWriter { /// [`crate::transcript_mirror`]) and by age-based GC, and replay reads it /// as a stream so a large journal never becomes a large allocation. pub async fn append(&mut self, env: &Envelope) -> anyhow::Result { - let mut line = serde_json::to_vec(&env.redacted())?; + let mut redacted = env.redacted(); + compact_pi_payload(&mut redacted, &mut self.pi_context); + let mut line = serde_json::to_vec(&redacted)?; line.push(b'\n'); self.file.write_all(&line).await?; self.file.flush().await?; @@ -300,6 +307,7 @@ pub struct JournalReader { path: PathBuf, line_no: usize, position: u64, + pi_context: Vec, } impl JournalReader { @@ -321,6 +329,7 @@ impl JournalReader { path: path.to_path_buf(), line_no: 0, position: 0, + pi_context: Vec::new(), })) } @@ -379,9 +388,13 @@ impl JournalReader { through: checkpoint.through, } } else { - JournalRecord::Event(serde_json::from_value(value).map_err(|error| { + let mut event = serde_json::from_value(value).map_err(|error| { + anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) + })?; + expand_pi_payload(&mut event, &mut self.pi_context).map_err(|error| { anyhow::anyhow!("journal {}:{}: {error}", self.path.display(), self.line_no) - })?) + })?; + JournalRecord::Event(event) }; return Ok(Some(JournalRecordEntry { record, @@ -413,6 +426,113 @@ impl JournalReader { } } +const PI_MESSAGES_DELTA: &str = "_bt_messages_delta"; + +fn compact_pi_payload(event: &mut RedactedEnvelope, previous: &mut Vec) { + if event.source != "pi" { + return; + } + let Some(native) = event + .payload + .get_mut("event") + .and_then(serde_json::Value::as_object_mut) + else { + return; + }; + match event.event.as_str() { + "context" => { + let Some(messages) = native + .remove("messages") + .and_then(|messages| messages.as_array().cloned()) + else { + return; + }; + let common_prefix = previous + .iter() + .zip(&messages) + .take_while(|(left, right)| left == right) + .count(); + native.insert( + PI_MESSAGES_DELTA.into(), + serde_json::json!({ + "common_prefix": common_prefix, + "suffix": messages[common_prefix..], + }), + ); + *previous = messages; + } + "before_provider_request" => { + if let Some(payload) = native + .get_mut("payload") + .and_then(serde_json::Value::as_object_mut) + { + payload.remove("messages"); + } + } + "agent_end" => { + // The translator only consumes willRetry from this lifecycle event. + native.remove("messages"); + } + "message_update" => { + // Pi repeats the progressively growing assistant message on every + // streaming update. Translation only needs the update type to mark + // time-to-first-token; the completed message arrives separately in + // message_end. + native.remove("message"); + if let Some(update) = native + .get_mut("assistantMessageEvent") + .and_then(serde_json::Value::as_object_mut) + { + update.retain(|key, _| key == "type"); + } + native.retain(|key, _| key == "type" || key == "assistantMessageEvent"); + } + _ => {} + } +} + +fn expand_pi_payload( + event: &mut RedactedEnvelope, + previous: &mut Vec, +) -> anyhow::Result<()> { + if event.source != "pi" || event.event != "context" { + return Ok(()); + } + let Some(native) = event + .payload + .get_mut("event") + .and_then(serde_json::Value::as_object_mut) + else { + return Ok(()); + }; + let Some(delta) = native.remove(PI_MESSAGES_DELTA) else { + if let Some(messages) = native.get("messages").and_then(serde_json::Value::as_array) { + *previous = messages.clone(); + } + return Ok(()); + }; + let common_prefix = delta + .get("common_prefix") + .and_then(serde_json::Value::as_u64) + .and_then(|value| usize::try_from(value).ok()) + .ok_or_else(|| anyhow::anyhow!("invalid Pi context common prefix"))?; + anyhow::ensure!( + common_prefix <= previous.len(), + "Pi context common prefix exceeds prior context" + ); + let suffix = delta + .get("suffix") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| anyhow::anyhow!("invalid Pi context suffix"))?; + previous.truncate(common_prefix); + previous.extend(suffix.iter().cloned()); + native.insert( + "messages".into(), + serde_json::Value::Array(previous.clone()), + ); + Ok(()) +} + /// Best-effort age-based journal collection. A failed stat/remove is logged /// and ignored; stale state must never prevent the daemon from serving hooks. pub async fn gc_old_journals(data_dir: &Path, max_age: std::time::Duration) { @@ -475,6 +595,120 @@ pub fn envelope_from_redacted(r: RedactedEnvelope) -> Envelope { mod tests { use super::*; + fn pi_context(messages: Vec, ts_ms: i64) -> Envelope { + Envelope { + source: "pi".into(), + source_version: None, + plugin_version: None, + session_id: "pi-session".into(), + event: "context".into(), + ts_ms, + managed_run_id: None, + capture: None, + payload: serde_json::json!({"event":{"type":"context","messages":messages}}), + route: None, + config: None, + } + } + + #[tokio::test] + async fn pi_context_history_is_delta_encoded_and_replayed() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("pi.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let mut messages = Vec::new(); + let mut naive_bytes = 0usize; + for index in 0..20 { + messages.push(serde_json::json!({ + "role":"user", + "content": format!("{index}:{}", "x".repeat(4096)), + })); + let event = pi_context(messages.clone(), index); + naive_bytes += serde_json::to_vec(&event.redacted()).unwrap().len() + 1; + writer.append(&event).await.unwrap(); + } + let compacted = vec![ + serde_json::json!({"role":"compactionSummary","summary":"bounded"}), + serde_json::json!({"role":"user","content":"after"}), + ]; + let event = pi_context(compacted.clone(), 20); + naive_bytes += serde_json::to_vec(&event.redacted()).unwrap().len() + 1; + writer.append(&event).await.unwrap(); + drop(writer); + + let stored = tokio::fs::read(&path).await.unwrap(); + assert!( + stored.len() < naive_bytes / 3, + "Pi journal did not eliminate cumulative context copies: stored={} naive={naive_bytes}", + stored.len() + ); + assert!(!String::from_utf8_lossy(&stored).contains("\"messages\"")); + + let through = stored.len() as u64; + let mut reader = JournalReader::open(&path, through).await.unwrap().unwrap(); + let mut contexts = Vec::new(); + while let Some(entry) = reader.next_entry().await.unwrap() { + contexts.push( + entry + .payload + .pointer("/event/messages") + .and_then(serde_json::Value::as_array) + .cloned() + .unwrap(), + ); + } + assert_eq!(contexts.len(), 21); + assert_eq!(contexts[19].len(), 20); + assert_eq!(contexts[20], compacted); + } + + #[tokio::test] + async fn pi_streaming_updates_do_not_persist_growing_partial_messages() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("pi.ndjson"); + let mut writer = JournalWriter::open_path(&path).await.unwrap(); + let marker = "growing-partial-marker".repeat(4096); + let event = Envelope { + source: "pi".into(), + source_version: None, + plugin_version: None, + session_id: "pi-session".into(), + event: "message_update".into(), + ts_ms: 1, + managed_run_id: None, + capture: None, + payload: serde_json::json!({ + "event": { + "type": "message_update", + "assistantMessageEvent": { + "type": "text_delta", + "partial": marker, + }, + "message": {"role": "assistant", "content": marker}, + } + }), + route: None, + config: None, + }; + writer.append(&event).await.unwrap(); + drop(writer); + + let stored = tokio::fs::read(&path).await.unwrap(); + assert!(!String::from_utf8_lossy(&stored).contains("growing-partial-marker")); + + let mut reader = JournalReader::open(&path, stored.len() as u64) + .await + .unwrap() + .unwrap(); + let replayed = reader.next_entry().await.unwrap().unwrap(); + assert_eq!( + replayed + .payload + .pointer("/event/assistantMessageEvent/type"), + Some(&serde_json::json!("text_delta")) + ); + } + #[tokio::test] async fn source_journals_are_distinct_and_migrate_legacy_history() { let temp = tempfile::tempdir().unwrap(); diff --git a/bt-daemon/src/transcript_import/antigravity.rs b/bt-daemon/src/transcript_import/antigravity.rs index e37cd21..bd52456 100644 --- a/bt-daemon/src/transcript_import/antigravity.rs +++ b/bt-daemon/src/transcript_import/antigravity.rs @@ -2,48 +2,36 @@ use super::{envelope, validate_session_id}; use crate::wire::Envelope; use anyhow::bail; use serde_json::{json, Value}; -use std::collections::{HashSet, VecDeque}; +use std::collections::VecDeque; use std::path::{Path, PathBuf}; const TRANSCRIPT_NAME: &str = "transcript_full.jsonl"; #[derive(Default)] pub(super) struct Tail { - started: bool, - reported_tools: HashSet, - last_len: u64, + emitted: usize, + stopped: bool, } impl Tail { pub(super) fn poll( &mut self, events: Vec, - len: u64, + _len: u64, finalize: bool, ) -> anyhow::Result> { - if events.len() < 3 { - bail!("Antigravity import did not produce session boundary events"); - } - let mut out = Vec::new(); - if !self.started { - out.push(events[0].clone()); - self.started = true; - } else if len != self.last_len { - let mut checkpoint = events[0].clone(); - checkpoint.event = "ImportCheckpoint".into(); - out.push(checkpoint); - } - let end = events.len() - 2; - for event in &events[1..end] { - let step = event.payload.get("stepIdx").and_then(Value::as_i64); - if step.is_some_and(|step| self.reported_tools.insert(step)) { - out.push(event.clone()); - } + let Some((stop, active)) = events.split_last() else { + bail!("Antigravity import did not produce a session boundary event"); + }; + if self.emitted > active.len() { + bail!("Antigravity import event stream shrank without a reset"); } - if finalize { - out.extend_from_slice(&events[end..]); + let mut out = active[self.emitted..].to_vec(); + self.emitted = active.len(); + if finalize && !self.stopped { + out.push(stop.clone()); + self.stopped = true; } - self.last_len = len; Ok(out) } } @@ -81,25 +69,66 @@ fn is_transcript_path(path: &Path, session_id: &str) -> bool { == Some(session_id) } -pub(super) fn envelopes(path: &Path, records: &[Value]) -> anyhow::Result> { +pub(super) fn envelopes( + path: &Path, + records: &[Value], + record_end_offsets: &[u64], +) -> anyhow::Result> { + if records.len() != record_end_offsets.len() { + bail!("Antigravity transcript record offsets do not match parsed records"); + } let session_id = transcript_session_id(path) .ok_or_else(|| anyhow::anyhow!("invalid Antigravity transcript path {}", path.display()))?; if records.is_empty() { bail!("Antigravity transcript {} is empty", path.display()); } let (start_ms, end_ms) = timestamp_bounds_created(records); - let transcript_path = path.to_string_lossy(); - let mut events = vec![envelope( - "antigravity", - None, - &session_id, - "PreInvocation", - start_ms, - json!({"conversationId": session_id, "invocationNum": 0, "initialNumSteps": 0, - "transcriptPath": transcript_path}), - )]; + let transcript_path = path.to_string_lossy().into_owned(); + let mut events = Vec::new(); let mut planned_calls = VecDeque::new(); - for record in records { + let mut invocation_num = 0i64; + for (index, record) in records.iter().enumerate() { + let record_type = record.get("type").and_then(Value::as_str); + let step = record + .get("step_index") + .or_else(|| record.get("stepIndex")) + .and_then(Value::as_i64) + .unwrap_or(index as i64); + let ts = created_at_ms(record).unwrap_or(start_ms); + if record_type == Some("PLANNER_RESPONSE") { + let before = index + .checked_sub(1) + .and_then(|index| record_end_offsets.get(index)) + .copied() + .unwrap_or(0); + events.push(envelope( + "antigravity", + None, + &session_id, + "PreInvocation", + ts, + bounded_payload( + &session_id, + &transcript_path, + before, + json!({"invocationNum": invocation_num, "initialNumSteps": step}), + ), + )); + events.push(envelope( + "antigravity", + None, + &session_id, + "PostInvocation", + ts, + bounded_payload( + &session_id, + &transcript_path, + record_end_offsets[index], + json!({"invocationNum": invocation_num, "initialNumSteps": step}), + ), + )); + invocation_num += 1; + } if let Some(calls) = record .get("tool_calls") .or_else(|| record.get("toolCalls")) @@ -110,50 +139,58 @@ pub(super) fn envelopes(path: &Path, records: &[Value]) -> anyhow::Result Value { + let mut payload = json!({ + "conversationId": session_id, + "transcriptPath": transcript_path, + "_bt_transcript_observation": { + "path": transcript_path, + "observed_bytes": through, + } + }); + if let (Value::Object(payload), Value::Object(extra)) = (&mut payload, extra) { + payload.extend(extra); + } + payload +} + fn is_denied(record: &Value) -> bool { record .get("content") diff --git a/bt-daemon/src/transcript_import/claude.rs b/bt-daemon/src/transcript_import/claude.rs index 6132f6b..ddcc6dd 100644 --- a/bt-daemon/src/transcript_import/claude.rs +++ b/bt-daemon/src/transcript_import/claude.rs @@ -383,6 +383,9 @@ fn is_real_user(record: &Value) -> bool { if record.get("type").and_then(Value::as_str) != Some("user") { return false; } + if record.get("isCompactSummary").and_then(Value::as_bool) == Some(true) { + return false; + } !record .pointer("/message/content") .and_then(Value::as_array) diff --git a/bt-daemon/src/transcript_import/mod.rs b/bt-daemon/src/transcript_import/mod.rs index bb4c391..5d045b6 100644 --- a/bt-daemon/src/transcript_import/mod.rs +++ b/bt-daemon/src/transcript_import/mod.rs @@ -216,7 +216,9 @@ fn envelopes_from_records( &records.end_offsets, records.read_offset, ), - ImportSource::Antigravity => antigravity::envelopes(path, &records.values), + ImportSource::Antigravity => { + antigravity::envelopes(path, &records.values, &records.end_offsets) + } } } @@ -810,7 +812,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["PreInvocation"] + vec!["PreInvocation", "PostInvocation"] ); records.push(json!({"step_index":2,"source":"MODEL","type":"LIST_DIRECTORY","created_at":"2026-01-01T00:00:03Z","content":"result"})); @@ -821,7 +823,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["ImportCheckpoint", "PostToolUse"] + vec!["PostToolUse"] ); assert!(tail.poll(false).unwrap().is_empty()); assert_eq!( @@ -830,7 +832,7 @@ mod tests { .iter() .map(|event| event.event.as_str()) .collect::>(), - vec!["PostInvocation", "Stop"] + vec!["Stop"] ); } diff --git a/bt-daemon/src/translate/antigravity.rs b/bt-daemon/src/translate/antigravity.rs index 2b330c4..1115ce8 100644 --- a/bt-daemon/src/translate/antigravity.rs +++ b/bt-daemon/src/translate/antigravity.rs @@ -189,7 +189,20 @@ impl AntigravityTranslator { self.start_turn(clean_user_input(record_content(&record)), ts_ms, ops); } - if let Some(message) = transcript_message(&record, &record_type, &source) { + if record_type == "CHECKPOINT" { + if let Some(mut message) = transcript_message(&record, &record_type, &source) { + if let Some(message) = message.as_object_mut() { + message.insert("message_type".into(), json!("compaction_summary")); + } + self.history.clear(); + self.history.push(message); + // A checkpoint can be observed while an invocation is open. Its + // replacement window becomes the new origin for output slicing. + for invocation in self.invocations.values_mut() { + invocation.history_start = 0; + } + } + } else if let Some(message) = transcript_message(&record, &record_type, &source) { if record_type == "PLANNER_RESPONSE" { if let Some(turn) = &mut self.turn { turn.last_output = message.get("content").cloned(); diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index ad5e58f..57c2843 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -62,7 +62,78 @@ struct PendingTool { enum PendingHistory { Main, - Owned(Vec), + Owned(MessageHistory), +} + +#[derive(Default)] +struct MessageHistory { + messages: Vec, + preserved_after_compaction: Vec, +} + +struct HistoryMessage { + value: Value, + source_uuids: Vec, +} + +impl MessageHistory { + fn active(&self) -> Vec { + self.messages + .iter() + .map(|message| message.value.clone()) + .collect() + } + + fn push(&mut self, value: Value, source_uuid: Option) -> usize { + let index = self.messages.len(); + self.messages.push(HistoryMessage { + value, + source_uuids: source_uuid.into_iter().collect(), + }); + index + } + + fn update(&mut self, index: usize, value: Value, source_uuid: Option) { + let Some(message) = self.messages.get_mut(index) else { + return; + }; + message.value = value; + if let Some(source_uuid) = source_uuid { + if !message.source_uuids.contains(&source_uuid) { + message.source_uuids.push(source_uuid); + } + } + } + + fn observe_compact_boundary(&mut self, record: &Value) { + self.preserved_after_compaction = record + .pointer("/compactMetadata/preservedMessages/uuids") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_str) + .map(str::to_owned) + .collect(); + } + + fn begin_compacted(&mut self, summary: Value, source_uuid: Option) { + let preserved = std::mem::take(&mut self.preserved_after_compaction); + let kept = self + .messages + .drain(..) + .filter(|message| { + message + .source_uuids + .iter() + .any(|uuid| preserved.contains(uuid)) + }) + .collect::>(); + self.messages.push(HistoryMessage { + value: summary, + source_uuids: source_uuid.into_iter().collect(), + }); + self.messages.extend(kept); + } } struct PendingEmission { @@ -86,7 +157,7 @@ struct ClaudeTranslator { tool_seq: u32, main_transcript: Option, transcripts: HashMap, - main_history: Vec, + main_history: MessageHistory, emitted_requests: RecentSet, emitted_tools: RecentSet, pending_tools: HashMap, @@ -116,7 +187,7 @@ impl ClaudeTranslator { tool_seq: 0, main_transcript: None, transcripts: HashMap::new(), - main_history: Vec::new(), + main_history: MessageHistory::default(), emitted_requests: RecentSet::default(), emitted_tools: RecentSet::default(), pending_tools: HashMap::new(), @@ -436,7 +507,7 @@ impl ClaudeTranslator { let records = std::mem::take(&mut cursor.buffered); self.queue_transcript( records, - PendingHistory::Owned(Vec::new()), + PendingHistory::Owned(MessageHistory::default()), format!("subagent:{agent_id}"), parent.clone(), self.current_cwd.clone(), @@ -539,7 +610,7 @@ impl ClaudeTranslator { } fn release_terminal_state(&mut self) { - self.main_history.clear(); + self.main_history = MessageHistory::default(); self.transcripts.clear(); self.subagents.clear(); self.pending_tools.clear(); @@ -753,6 +824,19 @@ impl AgentTranslator for ClaudeTranslator { &mut ops, ), "PermissionDenied" => self.finish_tool(event, ToolApproval::Denied, None, &mut ops), + "PostCompact" => { + if let Some(parent) = self + .turn + .as_ref() + .map(|turn| turn.id.clone()) + .or_else(|| self.last_turn_id.clone()) + { + // PostCompact is the durable point at which Claude's + // synthetic compact-summary transcript record becomes the + // first message in the active history window. + self.emit_main(&parent, &mut ops); + } + } "SubagentStart" => { if let Some(agent_id) = string_field(&event.payload, "agent_id") { self.ensure_subagent(&agent_id, event, &mut ops); @@ -867,10 +951,10 @@ fn assistant_request_id(record: &Value) -> Option { struct ParsedTranscript { calls: Vec, tools: Vec, - history: Vec, + history: MessageHistory, } -fn parse_transcript(records: &[Value], mut history: Vec) -> ParsedTranscript { +fn parse_transcript(records: &[Value], mut history: MessageHistory) -> ParsedTranscript { let mut calls = Vec::::new(); let mut call_indexes = HashMap::::new(); let mut assistant_history_indexes = HashMap::::new(); @@ -894,17 +978,18 @@ fn parse_transcript(records: &[Value], mut history: Vec) -> ParsedTranscr calls.push(LlmCall::new( request_id.clone(), parse_timestamp_ms(record).unwrap_or(0), - history.clone(), + history.active(), )); index }); calls[index].observe(record); let output = calls[index].output_message(); + let source_uuid = string_field(record, "uuid"); if let Some(history_index) = assistant_history_indexes.get(&request_id) { - history[*history_index] = output; + history.update(*history_index, output, source_uuid); } else { - assistant_history_indexes.insert(request_id.clone(), history.len()); - history.push(output); + let history_index = history.push(output, source_uuid); + assistant_history_indexes.insert(request_id.clone(), history_index); } if let Some(content) = record.pointer("/message/content").and_then(Value::as_array) { @@ -937,6 +1022,28 @@ fn parse_transcript(records: &[Value], mut history: Vec) -> ParsedTranscr } } Some("user") => { + // Claude persists a synthetic user message containing the new + // active context immediately after a compact_boundary record. + // It is the first model-visible message after the boundary. + if record.get("isCompactSummary").and_then(Value::as_bool) == Some(true) { + let content = record + .pointer("/message/content") + .cloned() + .unwrap_or(Value::Null); + history.begin_compacted( + json!({ + "role": "user", + "content": if content.is_null() { + json!("[compaction summary unavailable]") + } else { + content + }, + "message_type": "compaction_summary" + }), + string_field(record, "uuid"), + ); + continue; + } let content = record .pointer("/message/content") .cloned() @@ -950,11 +1057,14 @@ fn parse_transcript(records: &[Value], mut history: Vec) -> ParsedTranscr had_tool_result = true; let call_id = string_field(block, "tool_use_id").unwrap_or_default(); let result = block.get("content").cloned().unwrap_or(Value::Null); - history.push(json!({ - "role": "tool", - "tool_call_id": call_id, - "content": result - })); + history.push( + json!({ + "role": "tool", + "tool_call_id": call_id, + "content": result + }), + string_field(record, "uuid"), + ); if let Some(tool) = tools.get_mut(&call_id) { tool.output = Some(result); tool.end_ms = parse_timestamp_ms(record).unwrap_or(tool.start_ms); @@ -968,12 +1078,23 @@ fn parse_transcript(records: &[Value], mut history: Vec) -> ParsedTranscr } } if !had_tool_result { - history.push(json!({ "role": "user", "content": content })); + history.push( + json!({ "role": "user", "content": content }), + string_field(record, "uuid"), + ); } } else if !content.is_null() { - history.push(json!({ "role": "user", "content": content })); + history.push( + json!({ "role": "user", "content": content }), + string_field(record, "uuid"), + ); } } + Some("system") + if record.get("subtype").and_then(Value::as_str) == Some("compact_boundary") => + { + history.observe_compact_boundary(record); + } _ => {} } } @@ -991,6 +1112,9 @@ fn is_real_user_record(record: &Value) -> bool { if record.get("type").and_then(Value::as_str) != Some("user") { return false; } + if record.get("isCompactSummary").and_then(Value::as_bool) == Some(true) { + return false; + } !record .pointer("/message/content") .and_then(Value::as_array) diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 844491a..70a1c96 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -109,7 +109,7 @@ struct Scope { model: Option, current_cwd: Option, open_turns: Vec, - conversation_history: Vec, + message_history: Vec, open_llm: Option, open_tools: HashMap, // call_id -> (tool span_id, turn_id) last_turn_end_ms: Option, @@ -724,7 +724,7 @@ impl CodexTranslator { let Some(index) = index else { return; }; - let input = Value::Array(scope.conversation_history.clone()); + let input = Value::Array(scope.message_history.clone()); let turn = &mut scope.open_turns[index]; let seq = turn.llm_seq; turn.llm_seq += 1; @@ -788,7 +788,7 @@ impl CodexTranslator { } } } - scope.conversation_history.push(msg); + scope.message_history.push(msg); } fn on_reasoning(&mut self, scope: &mut Scope, payload: &Value, ts: i64, ops: &mut Vec) { @@ -816,7 +816,7 @@ impl CodexTranslator { if let Some(llm) = &mut scope.open_llm { llm.output.push(item.clone()); } - scope.conversation_history.push(item); + scope.message_history.push(item); } fn on_tool_call(&mut self, scope: &mut Scope, payload: &Value, ts: i64, ops: &mut Vec) { @@ -849,8 +849,6 @@ impl CodexTranslator { "function": { "name": tool_name, "arguments": args_string }, }], }); - scope.conversation_history.push(tool_call_message.clone()); - let Some(call_id) = call_id else { return }; let turn_id = payload .get("metadata") @@ -872,9 +870,10 @@ impl CodexTranslator { self.ensure_llm(scope, Some(&turn_id), ts, ops); if let Some(llm) = &mut scope.open_llm { - llm.output.push(tool_call_message); + llm.output.push(tool_call_message.clone()); llm.last_output_ms = llm.last_output_ms.max(ts); } + scope.message_history.push(tool_call_message); let span_id = ids::span_id(&self.session_id, &format!("tool:{call_id}")); // spawn_agent: remember which turn ran it, so a later SubagentStart can @@ -1147,9 +1146,10 @@ impl CodexTranslator { ..Default::default() })); - // Synthetic llm span for the compaction call: before/after context. + // Synthetic llm span for the compaction call. The transcript exposes + // the replacement context, but not the exact request that produced it, + // so do not attach reconstructed pre-compaction history as LLM input. let start = turn_last_child.unwrap_or(turn_start); - let before = scope.conversation_history.clone(); let span_id = ids::span_id(&self.session_id, &format!("llm:{turn_id}:compaction")); let name = scope .model @@ -1162,17 +1162,16 @@ impl CodexTranslator { name: name.clone(), span_type: SpanType::Llm, start_ms: Some(start), - input: Some(json!({ "messages_before_compaction": before.len(), "history": before })), output: Some(compaction_output(replacement.as_ref())), metadata: Some(json!({ "model": scope.model, "turn_id": turn_id, "compaction": true })), ..Default::default() })); - if let Some(replacement) = replacement { - // Native compaction is a semantic memory barrier: future requests - // use only the replacement context, so the pre-compaction Values - // can be dropped as soon as their one compaction span is emitted. - scope.conversation_history = replacement; - } + // The ordered Compacted record is the history cutoff. The journal is + // still append-only, but translator state only needs the active window. + scope.message_history.clear(); + scope + .message_history + .extend(compaction_history(payload, replacement.as_ref())); let _ = (turn_span, name); scope.open_llm = Some(OpenLlm { span_id, @@ -1309,7 +1308,7 @@ impl Scope { model: None, current_cwd: None, open_turns: Vec::new(), - conversation_history: Vec::new(), + message_history: Vec::new(), open_llm: None, open_tools: HashMap::new(), last_turn_end_ms: None, @@ -1369,7 +1368,7 @@ fn push_tool_result(scope: &mut Scope, call_id: Option<&str>, payload: &Value) { .as_str() .map(str::to_string) .unwrap_or_else(|| serde_json::to_string(&output).unwrap_or_else(|_| "null".to_string())); - scope.conversation_history.push(json!({ + scope.message_history.push(json!({ "role": "tool", "content": content, "tool_call_id": call_id.unwrap_or_default(), @@ -1639,6 +1638,50 @@ fn compaction_output(replacement: Option<&Vec>) -> Value { }) } +fn compaction_history(payload: &Value, replacement: Option<&Vec>) -> Vec { + let items = replacement.map(Vec::as_slice).unwrap_or_default(); + if let Some(summary_index) = items + .iter() + .rposition(|item| item.get("type").and_then(Value::as_str) == Some("compaction")) + { + let mut active = Vec::with_capacity(items.len()); + active.push(items[summary_index].clone()); + active.extend( + items + .iter() + .enumerate() + .filter(|(index, _)| *index != summary_index) + .map(|(_, item)| item.clone()), + ); + return active; + } + if let Some(message) = payload + .get("message") + .and_then(Value::as_str) + .filter(|message| !message.is_empty()) + { + let mut active = Vec::with_capacity(items.len() + 1); + active.push(json!({ + "role": "user", + "content": message, + "message_type": "compaction_summary", + })); + active.extend(items.iter().cloned()); + return active; + } + if !items.is_empty() { + // Without a typed compaction item or explicit summary message, there + // is no evidence that any ordinary entry is the summary. Preserve the + // native replacement history order exactly. + return items.to_vec(); + } + vec![json!({ + "role": "user", + "content": "[compaction summary unavailable]", + "message_type": "compaction_summary", + })] +} + fn parse_ts(rec: &Value) -> Option { let s = rec.get("timestamp").and_then(Value::as_str)?; chrono::DateTime::parse_from_rfc3339(s) diff --git a/bt-daemon/src/translate/opencode.rs b/bt-daemon/src/translate/opencode.rs index 3bb751c..85bc7c6 100644 --- a/bt-daemon/src/translate/opencode.rs +++ b/bt-daemon/src/translate/opencode.rs @@ -59,6 +59,114 @@ struct NativeSession { tool_message_ids: HashMap, denied_tools: HashSet, completed_messages: RecentSet, + history: MessageHistory, + user_message_ids: HashSet, + user_parts: HashMap>, + history_tool_results: HashMap>, +} + +#[derive(Default)] +struct MessageHistory { + entries: Vec, + pending_compaction: Option, +} + +struct HistoryEntry { + message_id: String, + values: Vec, +} + +struct CompactionBoundary { + tail_start_id: Option, +} + +impl MessageHistory { + fn active(&self) -> Vec { + self.entries + .iter() + .flat_map(|entry| entry.values.iter().cloned()) + .collect() + } + + fn upsert(&mut self, message_id: &str, values: Vec) { + if let Some(entry) = self + .entries + .iter_mut() + .find(|entry| entry.message_id == message_id) + { + entry.values = values; + return; + } + self.entries.push(HistoryEntry { + message_id: message_id.to_string(), + values, + }); + } + + fn ensure(&mut self, message_id: &str) { + if !self + .entries + .iter() + .any(|entry| entry.message_id == message_id) + { + self.entries.push(HistoryEntry { + message_id: message_id.to_string(), + values: Vec::new(), + }); + } + } + + fn remove(&mut self, message_id: &str) { + self.entries.retain(|entry| entry.message_id != message_id); + } + + fn observe_compaction(&mut self, message_id: &str, tail_start_id: Option) { + // The native compaction trigger is control flow, not part of the + // provider-visible conversational history we expose on later spans. + self.remove(message_id); + self.pending_compaction = Some(CompactionBoundary { tail_start_id }); + } + + fn compaction_input(&self) -> Vec { + let Some(tail_start_id) = self + .pending_compaction + .as_ref() + .and_then(|boundary| boundary.tail_start_id.as_deref()) + else { + return self.active(); + }; + let end = self + .entries + .iter() + .position(|entry| entry.message_id == tail_start_id) + .unwrap_or(self.entries.len()); + self.entries[..end] + .iter() + .flat_map(|entry| entry.values.iter().cloned()) + .collect() + } + + fn begin_compacted(&mut self, message_id: &str, summary: Value) { + let tail_start_id = self + .pending_compaction + .take() + .and_then(|boundary| boundary.tail_start_id); + let kept = tail_start_id + .as_deref() + .and_then(|tail| { + self.entries + .iter() + .position(|entry| entry.message_id == tail) + }) + .map(|start| self.entries.drain(start..).collect::>()) + .unwrap_or_default(); + self.entries.clear(); + self.entries.push(HistoryEntry { + message_id: message_id.to_string(), + values: vec![summary], + }); + self.entries.extend(kept); + } } #[derive(Clone, Default)] @@ -94,6 +202,7 @@ impl AgentTranslator for OpenCodeTranslator { "permission.asked" => self.permission_asked(event), "permission.replied" => self.permission_replied(event), "session.idle" => self.finish_session_event(event, false, None), + "session.compacted" => Vec::new(), "session.deleted" => self.finish_session_event(event, true, None), "session.error" => { let error = format_error( @@ -265,6 +374,42 @@ impl OpenCodeTranslator { .join("\n") }) .unwrap_or_default(); + let message_id = output + .get("message") + .and_then(|message| message.get("id")) + .and_then(Value::as_str) + .map(str::to_owned) + .unwrap_or_else(|| format!("turn:{}:user", state.turn_number + 1)); + { + state.user_message_ids.insert(message_id.to_string()); + state.user_parts.insert( + message_id.clone(), + output + .get("parts") + .and_then(Value::as_array) + .into_iter() + .flatten() + .enumerate() + .filter(|(_, part)| part.get("type").and_then(Value::as_str) == Some("text")) + .filter_map(|(index, part)| { + let text = part.get("text").and_then(Value::as_str)?; + let id = part + .get("id") + .and_then(Value::as_str) + .map(str::to_owned) + .unwrap_or_else(|| format!("text:{index}")); + Some((id, text.to_string())) + }) + .collect(), + ); + state.history.upsert( + &message_id, + (!input.is_empty()) + .then(|| json!({"role":"user","content":input})) + .into_iter() + .collect(), + ); + } state.current_input = Some(input.clone()); state.current_turn_span_id = Some(turn_id.clone()); let model = event @@ -323,9 +468,31 @@ impl OpenCodeTranslator { match part.get("type").and_then(Value::as_str) { Some("text") => { if let Some(text) = part.get("text").and_then(Value::as_str) { - state.output_parts.insert(message_id.into(), text.into()); - if part.pointer("/time/end").is_some() { - state.current_output = Some(text.into()); + if state.user_message_ids.contains(message_id) { + let part_id = part.get("id").and_then(Value::as_str).unwrap_or("text"); + let parts = state.user_parts.entry(message_id.into()).or_default(); + if let Some((_, current)) = parts.iter_mut().find(|(id, _)| id == part_id) { + *current = text.to_string(); + } else { + parts.push((part_id.to_string(), text.to_string())); + } + let content = parts + .iter() + .map(|(_, text)| text.as_str()) + .collect::>() + .join("\n"); + state.history.upsert( + message_id, + (!content.is_empty()) + .then(|| json!({"role":"user","content":content})) + .into_iter() + .collect(), + ); + } else { + state.output_parts.insert(message_id.into(), text.into()); + if part.pointer("/time/end").is_some() { + state.current_output = Some(text.into()); + } } } } @@ -356,16 +523,34 @@ impl OpenCodeTranslator { Some("completed") => { if let Some(v) = part.pointer("/state/output") { state.tool_outputs.insert(call_id.into(), v.clone()); + state + .history_tool_results + .entry(message_id.into()) + .or_default() + .insert(call_id.into(), v.clone()); } } Some("error") => { + let error = format_error(part.pointer("/state/error")); + state.tool_errors.insert(call_id.into(), error.clone()); state - .tool_errors - .insert(call_id.into(), format_error(part.pointer("/state/error"))); + .history_tool_results + .entry(message_id.into()) + .or_default() + .insert(call_id.into(), Value::String(error)); } _ => {} } } + Some("compaction") => { + state.user_parts.remove(message_id); + state.history.observe_compaction( + message_id, + part.get("tail_start_id") + .and_then(Value::as_str) + .map(str::to_owned), + ); + } _ => {} } Vec::new() @@ -377,11 +562,6 @@ impl OpenCodeTranslator { .pointer("/properties/info") .or_else(|| event.payload.get("info")); let Some(info) = info else { return Vec::new() }; - if info.get("role").and_then(Value::as_str) != Some("assistant") - || info.pointer("/time/completed").is_none() - { - return Vec::new(); - } let Some(sid) = info.get("sessionID").and_then(Value::as_str) else { return Vec::new(); }; @@ -391,20 +571,29 @@ impl OpenCodeTranslator { let Some(state) = self.sessions.get_mut(sid) else { return Vec::new(); }; + match info.get("role").and_then(Value::as_str) { + Some("user") => { + state.user_message_ids.insert(mid.to_string()); + state.history.ensure(mid); + return Vec::new(); + } + Some("assistant") if info.pointer("/time/completed").is_some() => {} + _ => return Vec::new(), + } if !state.completed_messages.insert(mid.into()) { return Vec::new(); } - let Some(turn) = state.current_turn_span_id.clone() else { - return Vec::new(); - }; + let turn = state.current_turn_span_id.clone(); let cache_read = num(info, "/tokens/cache/read"); let cache_write = num(info, "/tokens/cache/write"); let prompt = num(info, "/tokens/input") + cache_read + cache_write; let completion = num(info, "/tokens/output"); let reasoning = num(info, "/tokens/reasoning"); - let mut assistant = json!({"role":"assistant","content":state.output_parts.remove(mid).unwrap_or_default()}); - if let Some(calls) = state.tool_calls.remove(mid) { - assistant["tool_calls"] = Value::Array(calls) + let content = state.output_parts.remove(mid).unwrap_or_default(); + let calls = state.tool_calls.remove(mid).unwrap_or_default(); + let mut assistant = json!({"role":"assistant","content":content}); + if !calls.is_empty() { + assistant["tool_calls"] = Value::Array(calls.clone()) } if let Some(reason) = state.reasoning_parts.remove(mid) { assistant["reasoning"] = json!([{"id":"reasoning","content":reason}]) @@ -412,13 +601,45 @@ impl OpenCodeTranslator { state .tool_message_ids .retain(|_, message_id| message_id != mid); + let mut history_values = (!content.is_empty() || !calls.is_empty()) + .then(|| assistant.clone()) + .into_iter() + .collect::>(); + let mut results = state.history_tool_results.remove(mid).unwrap_or_default(); + for call in &calls { + let Some(call_id) = call.get("id").and_then(Value::as_str) else { + continue; + }; + if let Some(result) = results.remove(call_id) { + history_values.push(json!({ + "role":"tool", + "tool_call_id":call_id, + "content":result + })); + } + } let mut input = Vec::new(); if let Some(system) = &state.system_prompt { input.push(json!({"role":"system","content":system})) } - if let Some(user) = &state.current_input { - input.push(json!({"role":"user","content":user})) + let is_compaction_summary = info.get("summary").and_then(Value::as_bool) == Some(true); + if is_compaction_summary { + input.extend(state.history.compaction_input()); + state.history.begin_compacted( + mid, + json!({ + "role":"assistant", + "content":content, + "message_type":"compaction_summary" + }), + ); + } else { + input.extend(state.history.active()); + state.history.upsert(mid, history_values); } + let Some(turn) = turn else { + return Vec::new(); + }; let provider = info .get("providerID") .and_then(Value::as_str) diff --git a/bt-daemon/src/translate/pi.rs b/bt-daemon/src/translate/pi.rs index 994ea30..99da04c 100644 --- a/bt-daemon/src/translate/pi.rs +++ b/bt-daemon/src/translate/pi.rs @@ -36,6 +36,7 @@ impl TranslatorFactory for PiTranslatorFactory { pending_llms: Vec::new(), tools: HashMap::new(), compaction: None, + active_compaction_message: None, branch_summary: None, last_ts: 0, thinking_level: None, @@ -69,6 +70,7 @@ struct PiTranslator { pending_llms: Vec, tools: HashMap, compaction: Option<(String, i64, Value)>, + active_compaction_message: Option, branch_summary: Option<(String, i64, Value)>, last_ts: i64, thinking_level: Option, @@ -101,10 +103,11 @@ impl AgentTranslator for PiTranslator { self.compaction = Some(( ids::span_id(&self.session_id, &format!("compaction:{}", envelope.ts_ms)), envelope.ts_ms, - event.clone(), + special_input(event, true), )) } "session_compact" => { + self.active_compaction_message = compaction_message(event); ops.extend(self.finish_special("Compaction", true, event, envelope.ts_ms)) } "session_before_tree" @@ -119,13 +122,17 @@ impl AgentTranslator for PiTranslator { &format!("branch-summary:{}", envelope.ts_ms), ), envelope.ts_ms, - event.clone(), + special_input(event, false), )) } - "session_tree" - if event.get("summaryEntry").is_some() || self.branch_summary.is_some() => - { - ops.extend(self.finish_special("Branch Summary", false, event, envelope.ts_ms)) + "session_tree" => { + // Tree navigation can select a branch with a different (or no) + // active compaction. The next native context event is + // authoritative for that branch. + self.active_compaction_message = None; + if event.get("summaryEntry").is_some() || self.branch_summary.is_some() { + ops.extend(self.finish_special("Branch Summary", false, event, envelope.ts_ms)) + } } "session_shutdown" => { ops.extend(self.close_turn(envelope.ts_ms, None)); @@ -294,7 +301,28 @@ impl PiTranslator { ops } fn capture_context(&mut self, event: &Value, ts: i64) { - let input = event.get("messages").cloned().unwrap_or_else(|| json!([])); + let mut messages = event + .get("messages") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let native_compaction = messages.iter().position(|message| { + message.get("role").and_then(Value::as_str) == Some("compactionSummary") + }); + match native_compaction { + Some(0) => self.active_compaction_message = messages.first().cloned(), + Some(index) => { + let message = messages.remove(index); + self.active_compaction_message = Some(message.clone()); + messages.insert(0, message); + } + None => { + if let Some(compaction) = &self.active_compaction_message { + messages.insert(0, compaction.clone()); + } + } + } + let input = Value::Array(messages); self.pending_llms.push(PendingLlm { start_ms: ts, input, @@ -304,7 +332,13 @@ impl PiTranslator { } fn provider_request(&mut self, event: &Value) { if let Some(call) = self.pending_llms.last_mut() { - call.provider = Some(event.clone()) + let mut provider = event.clone(); + if let Some(payload) = provider.get_mut("payload").and_then(Value::as_object_mut) { + // The authoritative provider-visible messages are already the + // LLM span input captured by the preceding context event. + payload.remove("messages"); + } + call.provider = Some(provider) } } fn streaming_update(&mut self, event: &Value, ts: i64) { @@ -562,6 +596,7 @@ impl PiTranslator { fn close_root(&mut self, ts: i64) -> SpanOp { self.opened = false; self.compaction = None; + self.active_compaction_message = None; self.branch_summary = None; SpanOp::Merge(SpanRow { span_id: self.root_span_id.clone(), @@ -575,6 +610,58 @@ impl PiTranslator { } } +fn compaction_message(event: &Value) -> Option { + let entry = event.get("compactionEntry")?; + let summary = entry.get("summary")?.clone(); + Some(json!({ + "role": "compactionSummary", + "summary": summary, + "tokensBefore": entry.get("tokensBefore").cloned().unwrap_or(Value::Null), + "timestamp": entry.get("timestamp").cloned().unwrap_or(Value::Null), + })) +} + +fn special_input(event: &Value, compaction: bool) -> Value { + if compaction { + let preparation = event.get("preparation"); + json!({ + "reason": event.get("reason"), + "willRetry": event.get("willRetry"), + "customInstructions": event.get("customInstructions"), + "tokensBefore": preparation.and_then(|value| value.get("tokensBefore")), + "firstKeptEntryId": preparation.and_then(|value| value.get("firstKeptEntryId")), + "isSplitTurn": preparation.and_then(|value| value.get("isSplitTurn")), + "messagesToSummarizeCount": preparation + .and_then(|value| value.get("messagesToSummarize")) + .and_then(Value::as_array) + .map(Vec::len), + "turnPrefixMessagesCount": preparation + .and_then(|value| value.get("turnPrefixMessages")) + .and_then(Value::as_array) + .map(Vec::len), + "branchEntryCount": event + .get("branchEntries") + .and_then(Value::as_array) + .map(Vec::len), + }) + } else { + let preparation = event.get("preparation"); + json!({ + "targetId": preparation.and_then(|value| value.get("targetId")), + "oldLeafId": preparation.and_then(|value| value.get("oldLeafId")), + "commonAncestorId": preparation.and_then(|value| value.get("commonAncestorId")), + "userWantsSummary": preparation.and_then(|value| value.get("userWantsSummary")), + "customInstructions": preparation.and_then(|value| value.get("customInstructions")), + "replaceInstructions": preparation.and_then(|value| value.get("replaceInstructions")), + "label": preparation.and_then(|value| value.get("label")), + "entriesToSummarizeCount": preparation + .and_then(|value| value.get("entriesToSummarize")) + .and_then(Value::as_array) + .map(Vec::len), + }) + } +} + fn num(v: &Value, key: &str) -> i64 { v.get(key).and_then(Value::as_i64).unwrap_or(0) } diff --git a/bt-daemon/tests/antigravity_translator.rs b/bt-daemon/tests/antigravity_translator.rs index a0dbe8e..550f252 100644 --- a/bt-daemon/tests/antigravity_translator.rs +++ b/bt-daemon/tests/antigravity_translator.rs @@ -518,6 +518,117 @@ fn real_cli_schema_recovers_messages_and_post_only_tool() { assert_eq!(tool.metadata.as_ref().unwrap()["tool_approval"], "approved"); } +#[test] +fn checkpoint_replaces_prior_llm_history() { + let records = vec![ + json!({ + "step_index": 0, + "source": "USER_EXPLICIT", + "type": "USER_INPUT", + "status": "DONE", + "content": "Old request that was compacted" + }), + json!({ + "step_index": 1, + "source": "MODEL", + "type": "PLANNER_RESPONSE", + "status": "DONE", + "content": "Old response that was compacted" + }), + json!({ + "step_index": 2, + "source": "SYSTEM", + "type": "CHECKPOINT", + "status": "DONE", + "content": "The earlier parts of this conversation have been truncated due to its long length. Summary of the retained context." + }), + json!({ + "step_index": 3, + "source": "USER_EXPLICIT", + "type": "USER_INPUT", + "status": "DONE", + "content": "Continue from the checkpoint" + }), + json!({ + "step_index": 4, + "source": "MODEL", + "type": "PLANNER_RESPONSE", + "status": "DONE", + "content": "Continued response" + }), + ]; + let (transcript, boundary) = jsonl(&records); + let path = "/tmp/conversation/transcript_full.jsonl"; + let registry = Registry::default_agents(); + let mut translator = registry.create("antigravity", "checkpoint-conversation"); + let ctx = SessionCtx { + session_id: "checkpoint-conversation".into(), + config: None, + }; + + let mut ops = translator + .handle( + &event( + "PreInvocation", + 100, + path, + &transcript, + boundary[3], + json!({"invocationNum":0,"initialNumSteps":4}), + ), + &ctx, + ) + .unwrap(); + ops.extend( + translator + .handle( + &event( + "PostInvocation", + 200, + path, + &transcript, + boundary[4], + json!({"invocationNum":0,"initialNumSteps":4}), + ), + &ctx, + ) + .unwrap(), + ); + + let rows = reduce(ops); + let llm = rows + .values() + .find(|row| row.span_type == SpanType::Llm) + .unwrap(); + assert_eq!( + llm.input, + Some(json!([ + { + "role": "system", + "content": "The earlier parts of this conversation have been truncated due to its long length. Summary of the retained context.", + "step_type": "CHECKPOINT", + "message_type": "compaction_summary" + }, + { + "role": "user", + "content": "Continue from the checkpoint", + "step_type": "USER_INPUT" + } + ])) + ); + assert_eq!( + llm.output, + Some(json!([{ + "role": "assistant", + "content": "Continued response", + "step_type": "PLANNER_RESPONSE" + }])) + ); + let input = serde_json::to_string(llm.input.as_ref().unwrap()).unwrap(); + assert!(!input.contains("Old request")); + assert!(!input.contains("Old response")); +} + #[test] fn resumed_process_reuses_invocation_zero_without_reparenting_to_turn_one() { let records = vec![ diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index 1b39eee..c81f64d 100644 --- a/bt-daemon/tests/claude_translator.rs +++ b/bt-daemon/tests/claude_translator.rs @@ -775,3 +775,224 @@ fn claude_large_catch_up_emits_one_historical_snapshot_per_batch() { } assert_eq!(llm_count, CALLS); } + +#[test] +fn claude_post_compact_replaces_old_prefix_and_preserves_recent_window() { + let base = chrono::DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z") + .unwrap() + .timestamp_millis(); + let transcript = tempfile::NamedTempFile::new().unwrap(); + let transcript_path = transcript.path().to_str().unwrap(); + let initial = [ + json!({ + "type":"user", + "uuid":"old-user", + "timestamp":"2026-01-01T00:00:01Z", + "message":{"role":"user","content":"old question"} + }), + json!({ + "type":"assistant", + "uuid":"old-assistant", + "timestamp":"2026-01-01T00:00:02Z", + "message":{"id":"old-request","model":"claude-test","role":"assistant","content":[{"type":"text","text":"old answer"}],"usage":{"input_tokens":1,"output_tokens":1}} + }), + json!({ + "type":"user", + "uuid":"recent-user", + "timestamp":"2026-01-01T00:00:02.100Z", + "message":{"role":"user","content":"recent question"} + }), + json!({ + "type":"assistant", + "uuid":"recent-assistant-text", + "timestamp":"2026-01-01T00:00:02.200Z", + "message":{"id":"recent-request","model":"claude-test","role":"assistant","content":[{"type":"text","text":"recent answer"}],"usage":{"input_tokens":1,"output_tokens":1}} + }), + json!({ + "type":"assistant", + "uuid":"recent-assistant-tool", + "timestamp":"2026-01-01T00:00:02.300Z", + "message":{"id":"recent-request","model":"claude-test","role":"assistant","content":[{"type":"tool_use","id":"tool-1","name":"Read","input":{"file_path":"README.md"}}],"usage":{"input_tokens":1,"output_tokens":1}} + }), + json!({ + "type":"user", + "uuid":"recent-tool-result", + "timestamp":"2026-01-01T00:00:02.400Z", + "message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"tool-1","content":"contents"}]} + }), + json!({ + "type":"attachment", + "uuid":"recent-attachment", + "timestamp":"2026-01-01T00:00:02.500Z", + "message":{"role":"user","content":"attachment metadata"} + }), + ]; + std::fs::write( + transcript.path(), + initial + .iter() + .map(Value::to_string) + .collect::>() + .join("\n"), + ) + .unwrap(); + + let registry = Registry::default_agents(); + let mut translator = registry.create("claude-code", "compacted"); + let ctx = SessionCtx { + session_id: "compacted".into(), + config: None, + }; + let envelope = |event: &str, ts_ms: i64, prompt: Option<&str>| Envelope { + source: "claude-code".into(), + source_version: None, + plugin_version: None, + session_id: "compacted".into(), + event: event.into(), + ts_ms, + managed_run_id: None, + payload: json!({ + "session_id":"compacted", + "transcript_path":transcript_path, + "prompt":prompt, + }), + route: None, + config: None, + capture: None, + }; + let mut ops = Vec::new(); + for event in [ + envelope("UserPromptSubmit", base + 1_000, Some("old question")), + envelope("Stop", base + 2_000, None), + ] { + ops.extend(translator.handle(&event, &ctx).unwrap()); + while let Some(batch) = translator.drain_pending(&ctx).unwrap() { + ops.extend(batch); + } + } + + let compacted = [ + json!({ + "type":"system", + "subtype":"compact_boundary", + "timestamp":"2026-01-01T00:00:03Z", + "compactMetadata":{ + "trigger":"manual", + "preservedSegment":{ + "headUuid":"recent-user", + "anchorUuid":"compact-summary", + "tailUuid":"recent-attachment" + }, + "preservedMessages":{ + "anchorUuid":"compact-summary", + "uuids":[ + "recent-user", + "recent-assistant-text", + "recent-assistant-tool", + "recent-tool-result", + "recent-attachment" + ], + "allUuids":[ + "recent-user", + "recent-assistant-text", + "recent-assistant-tool", + "recent-tool-result", + "internal-queue-record", + "recent-attachment" + ] + } + } + }), + json!({ + "type":"user", + "uuid":"compact-summary", + "isCompactSummary":true, + "timestamp":"2026-01-01T00:00:04Z", + "message":{"role":"user","content":"compact summary"} + }), + ]; + let previous = std::fs::read_to_string(transcript.path()).unwrap(); + std::fs::write( + transcript.path(), + format!( + "{previous}\n{}", + compacted + .iter() + .map(Value::to_string) + .collect::>() + .join("\n") + ), + ) + .unwrap(); + ops.extend( + translator + .handle(&envelope("PostCompact", base + 4_000, None), &ctx) + .unwrap(), + ); + while let Some(batch) = translator.drain_pending(&ctx).unwrap() { + ops.extend(batch); + } + + let post_compact = [ + json!({ + "type":"user", + "timestamp":"2026-01-01T00:00:05Z", + "message":{"role":"user","content":"new question"} + }), + json!({ + "type":"assistant", + "timestamp":"2026-01-01T00:00:06Z", + "message":{"id":"new-request","model":"claude-test","role":"assistant","content":[{"type":"text","text":"new answer"}],"usage":{"input_tokens":2,"output_tokens":1}} + }), + ]; + let previous = std::fs::read_to_string(transcript.path()).unwrap(); + std::fs::write( + transcript.path(), + format!( + "{previous}\n{}", + post_compact + .iter() + .map(Value::to_string) + .collect::>() + .join("\n") + ), + ) + .unwrap(); + for event in [ + envelope("UserPromptSubmit", base + 5_000, Some("new question")), + envelope("Stop", base + 6_000, None), + ] { + ops.extend(translator.handle(&event, &ctx).unwrap()); + while let Some(batch) = translator.drain_pending(&ctx).unwrap() { + ops.extend(batch); + } + } + + let rows = reduce(ops); + let llm = rows + .values() + .find(|row| { + row.span_type == SpanType::Llm + && row.metadata.as_ref().unwrap()["request_id"] == "new-request" + }) + .unwrap(); + let messages = llm.input.as_ref().unwrap().as_array().unwrap(); + assert_eq!(messages.len(), 5); + assert_eq!(messages[0]["message_type"], "compaction_summary"); + assert_eq!(messages[0]["content"], "compact summary"); + assert_eq!(messages[1]["role"], "user"); + assert_eq!(messages[1]["content"], "recent question"); + assert_eq!(messages[2]["role"], "assistant"); + assert_eq!(messages[2]["content"], "recent answer"); + assert_eq!(messages[2]["tool_calls"][0]["id"], "tool-1"); + assert_eq!(messages[3]["role"], "tool"); + assert_eq!(messages[3]["tool_call_id"], "tool-1"); + assert_eq!(messages[3]["content"], "contents"); + assert_eq!(messages[4]["content"], "new question"); + assert!(!messages.iter().any(|message| { + matches!( + message.get("content").and_then(Value::as_str), + Some("old question" | "old answer" | "attachment metadata") + ) + })); +} diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index b972c7b..b0045b3 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -753,6 +753,14 @@ fn tool_and_llm_payloads_preserve_original_contract() { .collect(); llms.sort_by_key(|row| row.start_ms); assert_eq!(llms.len(), 2); + assert!(llms[0] + .input + .as_ref() + .unwrap() + .as_array() + .unwrap() + .iter() + .all(|message| message.get("tool_calls").is_none())); assert_eq!( llms[0].output.as_ref().unwrap()["tool_calls"][0]["function"]["arguments"], json!("{\"cmd\":\"cat /tmp/review/SKILL.md\",\"sandbox_permissions\":\"require_escalated\",\"justification\":\"Need access\",\"prefix_rule\":[\"cat\"]}") @@ -910,10 +918,14 @@ fn codex_compaction_replaces_history_for_following_llms() { json!({ "timestamp": "2026-01-01T00:00:03Z", "type": "event_msg", "payload": { "type": "task_started", "turn_id": "t1" } }), json!({ "timestamp": "2026-01-01T00:00:04Z", "type": "response_item", "payload": { "type": "message", "role": "assistant", "content": [{ "type": "output_text", "text": "discard me" }] } }), json!({ "timestamp": "2026-01-01T00:00:05Z", "type": "event_msg", "payload": { "type": "token_count", "info": { "last_token_usage": { "input_tokens": 10, "output_tokens": 2 } } } }), - json!({ "timestamp": "2026-01-01T00:00:06Z", "type": "compacted", "payload": { "replacement_history": [{ "role": "user", "content": "compacted context" }] } }), + json!({ "timestamp": "2026-01-01T00:00:06Z", "type": "compacted", "payload": { "replacement_history": [ + { "role": "developer", "content": "retained preamble" }, + { "type": "compaction", "encrypted_content": "opaque-summary" } + ] } }), json!({ "timestamp": "2026-01-01T00:00:07Z", "type": "event_msg", "payload": { "type": "token_count", "info": { "last_token_usage": { "input_tokens": 5, "output_tokens": 1 } } } }), json!({ "timestamp": "2026-01-01T00:00:08Z", "type": "event_msg", "payload": { "type": "task_complete", "turn_id": "t1" } }), json!({ "timestamp": "2026-01-01T00:00:09Z", "type": "event_msg", "payload": { "type": "task_started", "turn_id": "t2" } }), + json!({ "timestamp": "2026-01-01T00:00:09.500Z", "type": "response_item", "payload": { "type": "message", "role": "user", "content": [{ "type": "input_text", "text": "after compaction" }] } }), json!({ "timestamp": "2026-01-01T00:00:10Z", "type": "response_item", "payload": { "type": "message", "role": "assistant", "content": [{ "type": "output_text", "text": "after" }] } }), json!({ "timestamp": "2026-01-01T00:00:11Z", "type": "event_msg", "payload": { "type": "token_count", "info": { "last_token_usage": { "input_tokens": 6, "output_tokens": 1 } } } }), ] { @@ -947,7 +959,68 @@ fn codex_compaction_replaces_history_for_following_llms() { let input = following.input.as_ref().unwrap().as_array().unwrap(); assert_eq!( input, - &[json!({ "role": "user", "content": "compacted context" })] + &[ + json!({ "type": "compaction", "encrypted_content": "opaque-summary" }), + json!({ "role": "developer", "content": "retained preamble" }), + json!({ "role": "user", "content": "after compaction" }) + ] + ); +} + +#[test] +fn codex_untagged_replacement_history_preserves_native_order() { + let tmp = tempfile::tempdir().unwrap(); + let transcript = tmp.path().join("rollout.jsonl"); + for record in [ + json!({ "timestamp": "2026-01-01T00:00:01Z", "type": "session_meta", "payload": { "id": "s", "cwd": "/x/app" } }), + json!({ "timestamp": "2026-01-01T00:00:02Z", "type": "turn_context", "payload": { "model": "gpt-5.5" } }), + json!({ "timestamp": "2026-01-01T00:00:03Z", "type": "event_msg", "payload": { "type": "task_started", "turn_id": "t1" } }), + json!({ "timestamp": "2026-01-01T00:00:04Z", "type": "compacted", "payload": { "replacement_history": [ + { "role": "developer", "content": "first retained message" }, + { "role": "user", "content": "second retained message" }, + { "role": "assistant", "content": "third retained message" } + ] } }), + json!({ "timestamp": "2026-01-01T00:00:05Z", "type": "event_msg", "payload": { "type": "task_complete", "turn_id": "t1" } }), + json!({ "timestamp": "2026-01-01T00:00:06Z", "type": "event_msg", "payload": { "type": "task_started", "turn_id": "t2" } }), + json!({ "timestamp": "2026-01-01T00:00:07Z", "type": "response_item", "payload": { "type": "message", "role": "user", "content": [{ "type": "input_text", "text": "after compaction" }] } }), + json!({ "timestamp": "2026-01-01T00:00:08Z", "type": "response_item", "payload": { "type": "message", "role": "assistant", "content": [{ "type": "output_text", "text": "after" }] } }), + json!({ "timestamp": "2026-01-01T00:00:09Z", "type": "event_msg", "payload": { "type": "token_count", "info": { "last_token_usage": { "input_tokens": 6, "output_tokens": 1 } } } }), + ] { + append(&transcript, record); + } + + let registry = Registry::default_agents(); + let mut translator = registry.create("codex", "s"); + let ctx = SessionCtx { + session_id: "s".into(), + config: None, + }; + let rows = reduce( + translator + .handle( + &envelope("s", "SessionStart", transcript.to_str().unwrap(), json!({})), + &ctx, + ) + .unwrap(), + ); + let following = rows + .values() + .find(|row| { + row.span_type == SpanType::Llm + && row + .metadata + .as_ref() + .is_some_and(|metadata| metadata["turn_id"] == json!("t2")) + }) + .unwrap(); + assert_eq!( + following.input.as_ref().unwrap(), + &json!([ + { "role": "developer", "content": "first retained message" }, + { "role": "user", "content": "second retained message" }, + { "role": "assistant", "content": "third retained message" }, + { "role": "user", "content": "after compaction" } + ]) ); } diff --git a/bt-daemon/tests/opencode_translator.rs b/bt-daemon/tests/opencode_translator.rs index a0e3803..5070678 100644 --- a/bt-daemon/tests/opencode_translator.rs +++ b/bt-daemon/tests/opencode_translator.rs @@ -120,6 +120,131 @@ fn opencode_builds_turn_llm_tool_and_closes_the_session() { .all(|r| r.end_ms.is_some())); } +#[test] +fn opencode_compaction_replaces_old_prefix_and_preserves_native_tail() { + let registry = Registry::default_agents(); + let mut translator = registry.create("opencode", "root-session"); + let ctx = SessionCtx { + session_id: "root-session".into(), + config: None, + }; + let events = vec![ + event( + "session.created", + 1, + json!({"properties":{"info":{"id":"native"}}}), + ), + event( + "chat.message", + 2, + json!({"input":{"sessionID":"native"},"output":{"message":{"id":"old-user"},"parts":[{"type":"text","text":"old question"}]}}), + ), + event( + "message.part.updated", + 3, + json!({"properties":{"part":{"sessionID":"native","messageID":"old-assistant","type":"text","text":"old answer","time":{"end":3}}}}), + ), + event( + "message.updated", + 4, + json!({"properties":{"info":{"id":"old-assistant","sessionID":"native","role":"assistant","providerID":"openai","modelID":"gpt-5","time":{"created":2,"completed":4},"tokens":{}}}}), + ), + event( + "chat.message", + 5, + json!({"input":{"sessionID":"native"},"output":{"message":{"id":"recent-user"},"parts":[{"type":"text","text":"recent question"}]}}), + ), + event( + "message.part.updated", + 6, + json!({"properties":{"part":{"sessionID":"native","messageID":"recent-assistant","type":"text","text":"recent answer"}}}), + ), + event( + "message.part.updated", + 7, + json!({"properties":{"part":{"sessionID":"native","messageID":"recent-assistant","type":"tool","callID":"tool-1","tool":"read","state":{"status":"completed","input":{"path":"README.md"},"output":"contents"}}}}), + ), + event( + "message.updated", + 8, + json!({"properties":{"info":{"id":"recent-assistant","sessionID":"native","role":"assistant","providerID":"openai","modelID":"gpt-5","time":{"created":5,"completed":8},"tokens":{}}}}), + ), + event( + "message.updated", + 9, + json!({"properties":{"info":{"id":"compact-trigger","sessionID":"native","role":"user","time":{"created":9}}}}), + ), + event( + "message.part.updated", + 10, + json!({"properties":{"part":{"id":"compaction-part","sessionID":"native","messageID":"compact-trigger","type":"compaction","auto":true,"overflow":false,"tail_start_id":"recent-user"}}}), + ), + event( + "message.part.updated", + 11, + json!({"properties":{"part":{"sessionID":"native","messageID":"compact-summary","type":"text","text":"summary of old work","time":{"end":11}}}}), + ), + event( + "message.updated", + 12, + json!({"properties":{"info":{"id":"compact-summary","parentID":"compact-trigger","sessionID":"native","role":"assistant","mode":"compaction","summary":true,"providerID":"openai","modelID":"gpt-5","time":{"created":10,"completed":12},"tokens":{}}}}), + ), + event( + "session.compacted", + 13, + json!({"properties":{"sessionID":"native"}}), + ), + event( + "message.updated", + 14, + json!({"properties":{"info":{"id":"continue-user","sessionID":"native","role":"user","time":{"created":14}}}}), + ), + event( + "message.part.updated", + 15, + json!({"properties":{"part":{"sessionID":"native","messageID":"continue-user","type":"text","text":"Continue if there are next steps","synthetic":true}}}), + ), + event( + "message.part.updated", + 16, + json!({"properties":{"part":{"sessionID":"native","messageID":"new-assistant","type":"text","text":"new answer","time":{"end":16}}}}), + ), + event( + "message.updated", + 17, + json!({"properties":{"info":{"id":"new-assistant","sessionID":"native","role":"assistant","providerID":"openai","modelID":"gpt-5","time":{"created":15,"completed":17},"tokens":{}}}}), + ), + ]; + let mut ops = Vec::new(); + for event in events { + ops.extend(translator.handle(&event, &ctx).unwrap()); + } + let rows = reduce(ops); + let llm = rows + .values() + .find(|row| { + row.span_type == SpanType::Llm + && row.metadata.as_ref().unwrap()["message_id"] == "new-assistant" + }) + .unwrap(); + let messages = llm.input.as_ref().unwrap().as_array().unwrap(); + assert_eq!(messages.len(), 5); + assert_eq!(messages[0]["message_type"], "compaction_summary"); + assert_eq!(messages[0]["content"], "summary of old work"); + assert_eq!(messages[1]["content"], "recent question"); + assert_eq!(messages[2]["content"], "recent answer"); + assert_eq!(messages[2]["tool_calls"][0]["id"], "tool-1"); + assert_eq!(messages[3]["role"], "tool"); + assert_eq!(messages[3]["content"], "contents"); + assert_eq!(messages[4]["content"], "Continue if there are next steps"); + assert!(!messages.iter().any(|message| { + matches!( + message.get("content").and_then(serde_json::Value::as_str), + Some("old question" | "old answer") + ) + })); +} + #[test] fn opencode_child_sessions_share_the_parent_trace_root() { let registry = Registry::default_agents(); diff --git a/bt-daemon/tests/pi_translator.rs b/bt-daemon/tests/pi_translator.rs index 8df7c87..1c4a8a2 100644 --- a/bt-daemon/tests/pi_translator.rs +++ b/bt-daemon/tests/pi_translator.rs @@ -70,6 +70,11 @@ fn pi_builds_turn_llm_tool_compaction_and_shutdown_spans() { 3, json!({"messages":[{"role":"user","content":"inspect"}]}), ), + event( + "before_provider_request", + 3, + json!({"payload":{"model":"gpt-5","messages":[{"role":"user","content":"inspect"}]}}), + ), event( "message_update", 4, @@ -103,15 +108,37 @@ fn pi_builds_turn_llm_tool_compaction_and_shutdown_spans() { event( "session_before_compact", 10, - json!({"preparation":{"tokensBefore":100}}), + json!({ + "preparation":{ + "tokensBefore":100, + "firstKeptEntryId":"kept-1", + "messagesToSummarize":[{"role":"user","content":"large history"}], + "turnPrefixMessages":[{"role":"assistant","content":"prefix"}] + }, + "branchEntries":[{"type":"message","message":{"role":"user","content":"large history"}}] + }), ), event( "session_compact", 11, - json!({"compactionEntry":{"summary":"short"}}), + json!({"compactionEntry":{"summary":"short","tokensBefore":100,"timestamp":"2026-01-01T00:00:11Z"}}), ), event("agent_end", 12, json!({"messages":[]})), - event("session_shutdown", 13, json!({"reason":"quit"})), + event("before_agent_start", 13, json!({"prompt":"continue"})), + // Pi normally puts compactionSummary first itself. Omitting it here + // verifies that session_compact still establishes the active base. + event( + "context", + 14, + json!({"messages":[{"role":"user","content":"continue"}]}), + ), + event( + "message_end", + 15, + json!({"message":{"role":"assistant","provider":"openai","model":"gpt-5","content":[{"type":"text","text":"continued"}],"usage":{"input":3,"output":1,"totalTokens":4}}}), + ), + event("agent_end", 16, json!({"messages":[]})), + event("session_shutdown", 17, json!({"reason":"quit"})), ]; let mut ops = Vec::new(); for event in events { @@ -122,20 +149,77 @@ fn pi_builds_turn_llm_tool_compaction_and_shutdown_spans() { rows.values() .filter(|r| r.span_type == SpanType::Llm) .count(), - 1 + 2 + ); + let compacted_llm = rows + .values() + .find(|r| { + r.span_type == SpanType::Llm + && r.input.as_ref().is_some_and(|input| { + input + .as_array() + .and_then(|messages| messages.first()) + .and_then(|message| message.get("role")) + == Some(&json!("compactionSummary")) + }) + }) + .unwrap(); + assert_eq!(compacted_llm.input.as_ref().unwrap()[0]["summary"], "short"); + assert_eq!( + compacted_llm.input.as_ref().unwrap()[1]["content"], + "continue" ); - let llm = rows + let first_llm = rows .values() - .find(|r| r.span_type == SpanType::Llm) + .find(|r| { + r.span_type == SpanType::Llm + && r.input + .as_ref() + .is_some_and(|input| input[0]["content"] == "inspect") + }) .unwrap(); - assert_eq!(llm.metrics.as_ref().unwrap()["prompt_tokens"], 8); - assert_eq!(llm.metrics.as_ref().unwrap()["time_to_first_token"], 0.001); + assert_eq!(first_llm.metrics.as_ref().unwrap()["prompt_tokens"], 8); + assert_eq!( + first_llm.metrics.as_ref().unwrap()["time_to_first_token"], + 0.001 + ); + assert_eq!( + first_llm.metadata.as_ref().unwrap()["provider_request"]["payload"]["model"], + "gpt-5" + ); + assert!( + first_llm.metadata.as_ref().unwrap()["provider_request"]["payload"] + .get("messages") + .is_none() + ); let tool = rows.values().find(|r| r.name == "skill: review").unwrap(); assert_eq!(tool.name, "skill: review"); assert_eq!(tool.metadata.as_ref().unwrap()["tool_approval"], "approved"); let failed_tool = rows.values().find(|r| r.name == "write").unwrap(); assert_eq!(failed_tool.error.as_deref(), Some("permission denied")); assert!(rows.values().any(|r| r.name == "Compaction")); + let compaction = rows.values().find(|r| r.name == "Compaction").unwrap(); + assert_eq!( + compaction.input.as_ref().unwrap()["messagesToSummarizeCount"], + 1 + ); + assert_eq!( + compaction.input.as_ref().unwrap()["turnPrefixMessagesCount"], + 1 + ); + assert_eq!(compaction.input.as_ref().unwrap()["branchEntryCount"], 1); + assert!(compaction + .input + .as_ref() + .unwrap() + .get("preparation") + .is_none()); + assert!(compaction + .input + .as_ref() + .unwrap() + .get("branchEntries") + .is_none()); assert!(rows .values() .filter(|r| r.span_type == SpanType::Task) diff --git a/bt-daemon/tests/replay.rs b/bt-daemon/tests/replay.rs index 2fa3b48..22450e8 100644 --- a/bt-daemon/tests/replay.rs +++ b/bt-daemon/tests/replay.rs @@ -76,7 +76,7 @@ async fn importing_antigravity_transcript_uses_the_production_translator() { .unwrap(); let output_rows = rows(&output.join("antigravity-import.ndjson")); assert_eq!(inserted(&output_rows, "task"), 2); - assert_eq!(inserted(&output_rows, "llm"), 1); + assert_eq!(inserted(&output_rows, "llm"), 2); assert!(output_rows .iter() .any(|row| { row.pointer("/Insert/input").and_then(Value::as_str) == Some("trace this") })); @@ -90,6 +90,79 @@ async fn importing_antigravity_transcript_uses_the_production_translator() { })); } +#[tokio::test] +async fn antigravity_import_applies_checkpoint_only_to_following_llms() { + let tmp = tempfile::tempdir().unwrap(); + let transcript = tmp + .path() + .join("brain/antigravity-compact-import/.system_generated/logs/transcript_full.jsonl"); + std::fs::create_dir_all(transcript.parent().unwrap()).unwrap(); + write_jsonl( + &transcript, + &[ + json!({"step_index":0,"source":"USER_EXPLICIT","type":"USER_INPUT","status":"DONE","created_at":"2026-01-01T00:00:01Z","content":"old request"}), + json!({"step_index":1,"source":"MODEL","type":"PLANNER_RESPONSE","status":"DONE","created_at":"2026-01-01T00:00:02Z","content":"old response"}), + json!({"step_index":2,"source":"SYSTEM","type":"CHECKPOINT","status":"DONE","created_at":"2026-01-01T00:00:03Z","content":"summary of old context"}), + json!({"step_index":3,"source":"USER_EXPLICIT","type":"USER_INPUT","status":"DONE","created_at":"2026-01-01T00:00:04Z","content":"new request"}), + json!({"step_index":4,"source":"MODEL","type":"PLANNER_RESPONSE","status":"DONE","created_at":"2026-01-01T00:00:05Z","content":"new response"}), + json!({"step_index":5,"source":"SYSTEM","type":"CHECKPOINT","status":"DONE","created_at":"2026-01-01T00:00:06Z","content":"newer summary"}), + json!({"step_index":6,"source":"USER_EXPLICIT","type":"USER_INPUT","status":"DONE","created_at":"2026-01-01T00:00:07Z","content":"newest request"}), + json!({"step_index":7,"source":"MODEL","type":"PLANNER_RESPONSE","status":"DONE","created_at":"2026-01-01T00:00:08Z","content":"newest response"}), + ], + ); + let output = tmp.path().join("spans"); + import_transcript( + &transcript, + ImportSource::Antigravity, + options(&output), + None, + false, + ) + .await + .unwrap(); + + let output_rows = rows(&output.join("antigravity-compact-import.ndjson")); + let llm_inputs = output_rows + .iter() + .filter(|row| row.pointer("/Insert/span_type").and_then(Value::as_str) == Some("llm")) + .filter_map(|row| row.pointer("/Insert/input")) + .collect::>(); + assert_eq!(llm_inputs.len(), 3); + assert_eq!( + llm_inputs[0], + &json!([{"role":"user","content":"old request","step_type":"USER_INPUT"}]) + ); + assert_eq!( + llm_inputs[1], + &json!([ + {"role":"system","content":"summary of old context","step_type":"CHECKPOINT","message_type":"compaction_summary"}, + {"role":"user","content":"new request","step_type":"USER_INPUT"} + ]) + ); + assert_eq!( + llm_inputs[2], + &json!([ + {"role":"system","content":"newer summary","step_type":"CHECKPOINT","message_type":"compaction_summary"}, + {"role":"user","content":"newest request","step_type":"USER_INPUT"} + ]) + ); + assert!(output_rows.iter().any(|row| { + row.pointer("/Merge/output/0/content") + .and_then(Value::as_str) + == Some("old response") + })); + assert!(output_rows.iter().any(|row| { + row.pointer("/Merge/output/0/content") + .and_then(Value::as_str) + == Some("new response") + })); + assert!(output_rows.iter().any(|row| { + row.pointer("/Merge/output/0/content") + .and_then(Value::as_str) + == Some("newest response") + })); +} + fn options(output: &std::path::Path) -> ServeOptions { ServeOptions { version: "test".into(), diff --git a/src/plugins/opencode/content/src/tracing/daemon.ts b/src/plugins/opencode/content/src/tracing/daemon.ts index c409459..ec51029 100644 --- a/src/plugins/opencode/content/src/tracing/daemon.ts +++ b/src/plugins/opencode/content/src/tracing/daemon.ts @@ -17,6 +17,7 @@ interface TracingRouteConfig { const FORWARDED_NATIVE_EVENTS = new Set([ "session.created", "session.idle", + "session.compacted", "session.deleted", "session.error", "message.part.updated",