Skip to content

Handle Codex turn_aborted as an interruption #206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Stephen Belanger (Qard) merged 2 commits into from
Oct 20, 2026
Merged

Handle Codex turn_aborted as an interruption #206

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7b6b0e5
Handle Codex turn_aborted events
Qard Oct 20, 2026
547f8aa
Keep Codex model spans outside tool runtime
Qard Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 54 additions & 7 deletions bt-daemon/src/translate/codex.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
_ => {}
}
}
Expand Down Expand Up @@ -1285,6 +1286,22 @@ impl CodexTranslator {
}

fn close_turn(&mut self, scope: &mut Scope, payload: &Value, ts: i64, ops: &mut Vec<SpanOp>) {
self.finish_turn(scope, payload, ts, None, ops);
}

fn abort_turn(&mut self, scope: &mut Scope, payload: &Value, ts: i64, ops: &mut Vec<SpanOp>) {
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<SpanOp>,
) {
let requested_turn_id = str_field(payload, "turn_id");
let turn_index = requested_turn_id
.as_deref()
Expand All @@ -1301,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
Expand All @@ -1315,19 +1336,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() && !has_open_tool {
ts.max(llm.start_ms)
} else {
llm.last_output_ms.max(llm.start_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")
Expand All @@ -1336,12 +1370,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()
}));
}
Expand All @@ -1351,6 +1388,7 @@ impl CodexTranslator {
scope: &mut Scope,
turn_id: &str,
end_ms: Option<i64>,
interruption_reason: Option<&str>,
ops: &mut Vec<SpanOp>,
) {
let parent_span_ids = self.turn_parent_span_ids(turn_id);
Expand All @@ -1362,6 +1400,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(),
Expand All @@ -1370,8 +1415,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()
}));
}
Expand Down
89 changes: 89 additions & 0 deletions bt-daemon/tests/codex_translator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,95 @@ 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" } }),
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);
}

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 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 && row.parent_span_ids.contains(&turn.span_id))
.unwrap();
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")
);

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();
Expand Down
44 changes: 41 additions & 3 deletions bt-daemon/tests/replay.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}}),
Expand Down Expand Up @@ -519,6 +519,44 @@ async fn imports_native_codex_rollout_through_codex_translator() {
.collect::<Vec<_>>();
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
Expand Down
Loading