From 7b6b0e58d024852896991b7bab1d1e006a05b36f Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 01:07:09 +0800 Subject: [PATCH 1/2] Handle Codex turn_aborted events --- bt-daemon/src/translate/codex.rs | 57 +++++++++++++++++++++--- bt-daemon/tests/codex_translator.rs | 69 +++++++++++++++++++++++++++++ bt-daemon/tests/replay.rs | 44 ++++++++++++++++-- 3 files changed, 160 insertions(+), 10 deletions(-) diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index eb0a1765..93d40be2 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -726,6 +726,7 @@ impl CodexTranslator { "user_message" => self.set_turn_input(scope, payload, ops), "token_count" => self.close_llm_with_tokens(scope, payload, ts, ops), "task_complete" => self.close_turn(scope, payload, ts, ops), + "turn_aborted" => self.abort_turn(scope, payload, ts, ops), _ => {} } } @@ -1285,6 +1286,22 @@ impl CodexTranslator { } fn close_turn(&mut self, scope: &mut Scope, payload: &Value, ts: i64, ops: &mut Vec) { + self.finish_turn(scope, payload, ts, None, ops); + } + + fn abort_turn(&mut self, scope: &mut Scope, payload: &Value, ts: i64, ops: &mut Vec) { + let reason = str_field(payload, "reason").unwrap_or_else(|| "turn_aborted".into()); + self.finish_turn(scope, payload, ts, Some(reason.as_str()), ops); + } + + fn finish_turn( + &mut self, + scope: &mut Scope, + payload: &Value, + ts: i64, + interruption_reason: Option<&str>, + ops: &mut Vec, + ) { let requested_turn_id = str_field(payload, "turn_id"); let turn_index = requested_turn_id .as_deref() @@ -1315,19 +1332,32 @@ impl CodexTranslator { } else { Some(llm_output(&llm.output)) }; + let metadata = if let Some(reason) = interruption_reason { + json!({ + "usage_unavailable_reason": "codex_turn_aborted_before_token_count", + "outcome": "interrupted", + "termination_reason": reason, + }) + } else { + json!({ + "usage_unavailable_reason": "codex_transcript_missing_token_count" + }) + }; ops.push(SpanOp::Merge(SpanRow { span_id: llm.span_id, root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![turn_span_id], - end_ms: Some(llm.last_output_ms), + end_ms: Some(if interruption_reason.is_some() { + ts.max(llm.start_ms) + } else { + llm.last_output_ms + }), output, - metadata: Some(json!({ - "usage_unavailable_reason": "codex_transcript_missing_token_count" - })), + metadata: Some(metadata), ..Default::default() })); } - self.close_tools_for_turn(scope, &turn_id, Some(ts), ops); + self.close_tools_for_turn(scope, &turn_id, Some(ts), interruption_reason, ops); let turn = scope.open_turns.remove(turn_index); scope.last_turn_end_ms = Some(ts); let output = str_field(payload, "last_agent_message") @@ -1336,12 +1366,15 @@ impl CodexTranslator { if let Some(output) = &output { scope.last_turn_output = Some(output.clone()); } + let metadata = interruption_reason + .map(|reason| json!({ "outcome": "interrupted", "termination_reason": reason })); ops.push(SpanOp::Merge(SpanRow { span_id: turn.span_id, root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![scope.turn_parent_span_id.clone()], end_ms: Some(ts), output, + metadata, ..Default::default() })); } @@ -1351,6 +1384,7 @@ impl CodexTranslator { scope: &mut Scope, turn_id: &str, end_ms: Option, + interruption_reason: Option<&str>, ops: &mut Vec, ) { let parent_span_ids = self.turn_parent_span_ids(turn_id); @@ -1362,6 +1396,13 @@ impl CodexTranslator { .collect(); for call_id in call_ids { if let Some((span_id, _, start_ms)) = scope.open_tools.remove(&call_id) { + let mut metadata = tool_approval_metadata(Some(ToolApproval::Approved)); + if let Some(reason) = interruption_reason { + if let Some(fields) = metadata.as_object_mut() { + fields.insert("outcome".into(), json!("interrupted")); + fields.insert("termination_reason".into(), json!(reason)); + } + } ops.push(SpanOp::Merge(SpanRow { span_id, root_span_id: self.effective_root_span_id.clone(), @@ -1370,8 +1411,10 @@ impl CodexTranslator { || start_ms.saturating_add(MAX_OPEN_TOOL_SPAN_MS), |end| end.min(start_ms.saturating_add(MAX_OPEN_TOOL_SPAN_MS)), )), - metadata: Some(tool_approval_metadata(Some(ToolApproval::Approved))), - error: Some(MISSING_TOOL_OUTPUT_ERROR.to_string()), + metadata: Some(metadata), + error: interruption_reason + .is_none() + .then(|| MISSING_TOOL_OUTPUT_ERROR.to_string()), ..Default::default() })); } diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index 18f0cf72..c2547df1 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -284,6 +284,75 @@ fn codex_happy_path_builds_session_turn_llm_tool_tree() { ); } +#[test] +fn codex_turn_aborted_closes_open_work_as_interrupted() { + let tmp = tempfile::tempdir().unwrap(); + let transcript = tmp.path().join("rollout.jsonl"); + let path = transcript.to_str().unwrap(); + 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": "event_msg", + "payload": { "type": "task_started", "turn_id": "t1" } }), + json!({ "timestamp": "2026-01-01T00:00:03Z", "type": "response_item", + "payload": { "type": "message", "role": "assistant", + "content": [{ "type": "output_text", "text": "Working" }] } }), + json!({ "timestamp": "2026-01-01T00:00:04Z", "type": "response_item", + "payload": { "type": "function_call", "call_id": "c1", "name": "shell", + "arguments": "{\"command\":\"sleep 10\"}", + "metadata": { "turn_id": "t1" } } }), + json!({ "timestamp": "2026-01-01T00:00:05Z", "type": "event_msg", + "payload": { "type": "turn_aborted", "turn_id": "t1", "reason": "interrupted" } }), + ] { + append(&transcript, record); + } + + let mut translator = Registry::default_agents().create("codex", "s"); + let ctx = SessionCtx { + session_id: "s".into(), + config: None, + }; + let mut ops = translator + .handle(&envelope("s", "SessionStart", path, json!({})), &ctx) + .unwrap(); + ops.extend(translator.flush(&ctx).unwrap()); + assert_merges_preserve_insert_identity(&ops); + let rows = reduce(ops); + let turn = find(&rows, SpanType::Task, "turn: t1"); + let abort_ms = 1_767_225_605_000; + assert_eq!(turn.end_ms, Some(abort_ms)); + assert_eq!( + turn.metadata.as_ref().unwrap()["outcome"], + json!("interrupted") + ); + assert_eq!( + turn.metadata.as_ref().unwrap()["termination_reason"], + json!("interrupted") + ); + assert_eq!(turn.error, None, "intentional interruption is not an error"); + + let llm = rows + .values() + .find(|row| row.span_type == SpanType::Llm) + .unwrap(); + assert_eq!(llm.end_ms, Some(abort_ms)); + assert_eq!( + llm.metadata.as_ref().unwrap()["outcome"], + json!("interrupted") + ); + + let tool = find(&rows, SpanType::Tool, "shell"); + assert_eq!(tool.end_ms, Some(abort_ms)); + assert_eq!( + tool.metadata.as_ref().unwrap()["outcome"], + json!("interrupted") + ); + assert_eq!( + tool.error, None, + "intentional interruption is not a tool failure" + ); +} + #[test] fn codex_root_preserves_canonical_source_across_lifecycle_events() { let tmp = tempfile::tempdir().unwrap(); diff --git a/bt-daemon/tests/replay.rs b/bt-daemon/tests/replay.rs index 317ff27a..0556bac3 100644 --- a/bt-daemon/tests/replay.rs +++ b/bt-daemon/tests/replay.rs @@ -477,9 +477,9 @@ async fn imports_native_codex_rollout_through_codex_translator() { json!({"timestamp":"2026-01-01T00:00:02Z","type":"turn_context","payload":{"model":"gpt-test"}}), json!({"timestamp":"2026-01-01T00:00:03Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn-1"}}), json!({"timestamp":"2026-01-01T00:00:04Z","type":"event_msg","payload":{"type":"user_message","message":"list files"}}), - json!({"timestamp":"2026-01-01T00:00:05Z","type":"response_item","payload":{"type":"function_call","call_id":"call-1","name":"shell","arguments":"{\"command\":\"ls\"}"}}), - json!({"timestamp":"2026-01-01T00:00:06Z","type":"response_item","payload":{"type":"function_call_output","call_id":"call-1","output":"README.md"}}), - json!({"timestamp":"2026-01-01T00:00:07Z","type":"event_msg","payload":{"type":"task_complete","last_agent_message":"Done"}}), + json!({"timestamp":"2026-01-01T00:00:05Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Working"}]}}), + json!({"timestamp":"2026-01-01T00:00:06Z","type":"response_item","payload":{"type":"function_call","call_id":"call-1","name":"shell","arguments":"{\"command\":\"sleep 10\"}"}}), + json!({"timestamp":"2026-01-01T00:00:07Z","type":"event_msg","payload":{"type":"turn_aborted","turn_id":"turn-1","reason":"interrupted"}}), json!({"timestamp":"2026-01-01T00:00:08Z","type":"turn_context","payload":{"model":"gpt-test"}}), json!({"timestamp":"2026-01-01T00:00:09Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn-2"}}), json!({"timestamp":"2026-01-01T00:00:10Z","type":"event_msg","payload":{"type":"user_message","message":"show status"}}), @@ -519,6 +519,44 @@ async fn imports_native_codex_rollout_through_codex_translator() { .collect::>(); assert!(turn_ids.contains(&"turn-1")); assert!(turn_ids.contains(&"turn-2")); + let interrupted_turn_id = rows + .iter() + .find(|row| row.pointer("/Insert/name").and_then(Value::as_str) == Some("turn: turn-1")) + .and_then(|row| row.pointer("/Insert/span_id")) + .and_then(Value::as_str) + .unwrap(); + let interrupted_turn = rows + .iter() + .find(|row| { + row.pointer("/Merge/span_id").and_then(Value::as_str) == Some(interrupted_turn_id) + && row + .pointer("/Merge/metadata/outcome") + .and_then(Value::as_str) + == Some("interrupted") + }) + .unwrap(); + assert_eq!( + interrupted_turn + .pointer("/Merge/metadata/outcome") + .and_then(Value::as_str), + Some("interrupted") + ); + assert_eq!( + interrupted_turn + .pointer("/Merge/metadata/termination_reason") + .and_then(Value::as_str), + Some("interrupted") + ); + assert!( + rows.iter().any(|row| { + row.pointer("/Merge/metadata/outcome") + .and_then(Value::as_str) + == Some("interrupted") + && row.pointer("/Merge/end_ms").and_then(Value::as_i64) == Some(1_767_225_607_000) + && row.pointer("/Merge/error").is_none() + }), + "open model/tool work should close at the native abort boundary without an error" + ); } /// Two `bt trace import` invocations for the same session id, each headed to From 547f8aaff71051d4a9559138c85c7300dae8b113 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 01:25:18 +0800 Subject: [PATCH 2/2] Keep Codex model spans outside tool runtime --- bt-daemon/src/translate/codex.rs | 8 ++++++-- bt-daemon/tests/codex_translator.rs | 24 ++++++++++++++++++++++-- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/bt-daemon/src/translate/codex.rs b/bt-daemon/src/translate/codex.rs index 93d40be2..ea5a8a6c 100644 --- a/bt-daemon/src/translate/codex.rs +++ b/bt-daemon/src/translate/codex.rs @@ -1318,6 +1318,10 @@ impl CodexTranslator { }; let turn_id = scope.open_turns[turn_index].turn_id.clone(); let turn_span_id = scope.open_turns[turn_index].span_id.clone(); + let has_open_tool = scope + .open_tools + .values() + .any(|(_, owner, _)| owner == &turn_id); if scope .open_llm @@ -1347,10 +1351,10 @@ impl CodexTranslator { span_id: llm.span_id, root_span_id: self.effective_root_span_id.clone(), parent_span_ids: vec![turn_span_id], - end_ms: Some(if interruption_reason.is_some() { + end_ms: Some(if interruption_reason.is_some() && !has_open_tool { ts.max(llm.start_ms) } else { - llm.last_output_ms + llm.last_output_ms.max(llm.start_ms) }), output, metadata: Some(metadata), diff --git a/bt-daemon/tests/codex_translator.rs b/bt-daemon/tests/codex_translator.rs index c2547df1..30d7751a 100644 --- a/bt-daemon/tests/codex_translator.rs +++ b/bt-daemon/tests/codex_translator.rs @@ -303,6 +303,13 @@ fn codex_turn_aborted_closes_open_work_as_interrupted() { "metadata": { "turn_id": "t1" } } }), json!({ "timestamp": "2026-01-01T00:00:05Z", "type": "event_msg", "payload": { "type": "turn_aborted", "turn_id": "t1", "reason": "interrupted" } }), + 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": "assistant", + "content": [{ "type": "output_text", "text": "Still thinking" }] } }), + json!({ "timestamp": "2026-01-01T00:00:08Z", "type": "event_msg", + "payload": { "type": "turn_aborted", "turn_id": "t2", "reason": "interrupted" } }), ] { append(&transcript, record); } @@ -331,11 +338,24 @@ fn codex_turn_aborted_closes_open_work_as_interrupted() { ); assert_eq!(turn.error, None, "intentional interruption is not an error"); + let active_turn = find(&rows, SpanType::Task, "turn: t2"); + let active_llm = rows + .values() + .find(|row| { + row.span_type == SpanType::Llm && row.parent_span_ids.contains(&active_turn.span_id) + }) + .unwrap(); + assert_eq!(active_llm.end_ms, Some(abort_ms + 3_000)); + let llm = rows .values() - .find(|row| row.span_type == SpanType::Llm) + .find(|row| row.span_type == SpanType::Llm && row.parent_span_ids.contains(&turn.span_id)) .unwrap(); - assert_eq!(llm.end_ms, Some(abort_ms)); + assert_eq!( + llm.end_ms, + Some(abort_ms - 1_000), + "the model span ends when it emits the tool call, not when the tool is aborted" + ); assert_eq!( llm.metadata.as_ref().unwrap()["outcome"], json!("interrupted")