diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index aa9d93c0a8..44b7e44a47 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 9eee37dcac..c7f22e2930 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1589,8 +1589,8 @@ pub struct ThreadSearchResponse { #[ts(export_to = "v2/")] pub struct ThreadSearchOccurrencesParams { pub thread_id: String, - /// Case-insensitive literal substring to find in visible user messages and final assistant - /// messages. + /// Case-insensitive literal substring to find in visible user messages and both partial and + /// final assistant answers. pub search_term: String, /// Opaque cursor returned by a previous call for the same thread and search term. #[ts(optional = nullable)] diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index f8cc0ec24d..3ac95063e4 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -669,6 +669,20 @@ async fn thread_search_occurrences_reads_paginated_projection() -> Result<()> { questions: None, }), ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::AgentMessage(AgentMessageItem { + id: "partial-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "stable partial answer".to_string(), + }], + phase: Some(MessagePhase::PartialAnswer), + memory_citation: None, + delivery: None, + questions: None, + }), + ), paginated_completed_item( thread_id, "turn-1", @@ -761,6 +775,24 @@ async fn thread_search_occurrences_reads_paginated_projection() -> Result<()> { assert_eq!(data[2].snippet_match_range.end, 15); assert_eq!(next_cursor, None); + let request_id = mcp + .send_thread_search_occurrences_request(ThreadSearchOccurrencesParams { + thread_id: thread_id.to_string(), + search_term: "partial".to_string(), + cursor: None, + limit: Some(3), + }) + .await?; + let ThreadSearchOccurrencesResponse { data, next_cursor } = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; + assert_eq!( + data.iter() + .map(|occurrence| (occurrence.item_id.as_str(), occurrence.snippet.as_str())) + .collect::>(), + vec![("partial-1", "stable partial answer")] + ); + assert_eq!(next_cursor, None); + let fork_request_id = mcp .send_thread_fork_request(ThreadForkParams { thread_id: thread_id.to_string(), diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 42090f9180..007898db4b 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -243,7 +243,7 @@ impl RealtimeStreamedItem { fn output_prefix(&self) -> &'static str { if self.prefix_final_message && self.sent_bytes == 0 - && !matches!(self.phase, Some(MessagePhase::Commentary)) + && matches!(self.phase, Some(MessagePhase::FinalAnswer) | None) { AGENT_FINAL_MESSAGE_PREFIX } else { @@ -928,9 +928,9 @@ impl RealtimeConversationManager { if handoff.client_managed_handoffs { return Ok(()); } - // TODO: Preserve explicit PartialAnswer phases through completed and streamed BEM - // routing, with integration coverage asserting the speakable channel. - let phase = if handoff.routes_handoff_by_bem() { + let phase = if handoff.routes_handoff_by_bem() + && !matches!(phase, Some(MessagePhase::PartialAnswer)) + { match bem_message_phase( &output_text, &handoff.codex_response_handoff_channel_prefixes, @@ -944,9 +944,10 @@ impl RealtimeConversationManager { } else { phase }; - // TODO: Treat PartialAnswer as nonterminal for V1 handoffs, omitting the final-message - // prefix for streamed, completed-item, and standalone output. - let is_commentary = matches!(phase, Some(MessagePhase::Commentary)); + let is_nonterminal = matches!( + phase, + Some(MessagePhase::Commentary | MessagePhase::PartialAnswer) + ); let active_handoff = handoff.stream.lock().await.active_handoff.clone(); let output = match active_handoff { Some(handoff_id) => { @@ -963,7 +964,7 @@ impl RealtimeConversationManager { ), phase, } - } else if handoff.event_parser == RealtimeEventParser::V1 && is_commentary { + } else if handoff.event_parser == RealtimeEventParser::V1 && is_nonterminal { RealtimeOutbound::HandoffAppend { handoff_id, text: output_text, @@ -990,7 +991,8 @@ impl RealtimeConversationManager { } } else { RealtimeOutbound::StandaloneHandoff { - text: if handoff.event_parser == RealtimeEventParser::V1 && !is_commentary { + text: if handoff.event_parser == RealtimeEventParser::V1 && !is_nonterminal + { format!("{AGENT_FINAL_MESSAGE_PREFIX}{output_text}") } else { output_text @@ -1032,14 +1034,14 @@ impl RealtimeConversationManager { let Some(handoff_id) = stream.active_handoff.clone() else { return; }; + // An explicit stable fragment is answer text, even when it starts + // with characters that legacy BEM treats as a private channel tag. + let parse_bem = handoff.routes_handoff_by_bem() + && !matches!(phase, Some(MessagePhase::PartialAnswer)); let mut streamed_item = RealtimeStreamedItem { handoff_id, - phase: if handoff.routes_handoff_by_bem() { - None - } else { - phase - }, - bem_channel_parser: handoff.routes_handoff_by_bem().then(|| { + phase: if parse_bem { None } else { phase }, + bem_channel_parser: parse_bem.then(|| { BemChannelParser::new(Arc::clone( &handoff.codex_response_handoff_channel_prefixes, )) diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index e7e8fc03dc..16af93eb53 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -26,6 +26,7 @@ use pretty_assertions::assert_eq; use std::collections::BTreeMap; use std::sync::Arc; use std::time::Instant; +use test_case::test_case; use tokio::sync::Mutex; #[test] @@ -231,11 +232,17 @@ async fn clears_active_handoff_explicitly() { assert_eq!(state.stream.lock().await.active_handoff.clone(), None); } -#[test] -fn streamed_handoff_preserves_a_bounded_final_tail() { +#[test_case(None, true; "legacy")] +#[test_case(Some(MessagePhase::FinalAnswer), true; "final_answer")] +#[test_case(Some(MessagePhase::Commentary), false; "commentary")] +#[test_case(Some(MessagePhase::PartialAnswer), false; "partial_answer")] +fn streamed_handoff_preserves_a_bounded_final_tail( + phase: Option, + expect_final_prefix: bool, +) { let mut item = RealtimeStreamedItem { handoff_id: "handoff_1".to_string(), - phase: Some(MessagePhase::FinalAnswer), + phase, bem_channel_parser: None, prefix_final_message: true, sent_bytes: 0, @@ -256,7 +263,12 @@ fn streamed_handoff_preserves_a_bounded_final_tail() { let output = format!("{first}{final_chunk}"); assert!(output.len() <= 4_000); - assert!(output.starts_with(&format!("{AGENT_FINAL_MESSAGE_PREFIX}HEAD"))); + let prefix = if expect_final_prefix { + AGENT_FINAL_MESSAGE_PREFIX + } else { + "" + }; + assert!(output.starts_with(&format!("{prefix}HEAD"))); assert!(output.contains(HANDOFF_STREAM_TRUNCATION_MARKER)); assert!(output.ends_with("TAIL")); } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 0dbb6df747..cebce3c147 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -4012,27 +4012,35 @@ async fn realtime_v2_noop_tool_call_returns_empty_function_output_without_respon Ok(()) } +#[test_matrix([None, Some("final_answer"), Some("commentary"), Some("partial_answer")], [false, true])] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Result<()> { +async fn conversation_mirrors_assistant_message_text_to_realtime_handoff( + phase: Option<&str>, + active_handoff: bool, +) -> Result<()> { skip_if_no_network!(Ok(())); let api_server = start_mock_server().await; + let mut message = responses::ev_assistant_message("msg_1", "assistant says hi"); + if let Some(phase) = phase { + message["item"]["phase"] = json!(phase); + } let _response_mock = responses::mount_sse_once( &api_server, responses::sse(vec![ responses::ev_response_created("resp_1"), - responses::ev_assistant_message("msg_1", "assistant says hi"), + message, responses::ev_completed("resp_1"), ]), ) .await; - let realtime_server = start_websocket_server(vec![vec![ - vec![ - json!({ - "type": "session.updated", - "session": { "id": "sess_1", "instructions": "backend prompt" } - }), + let mut realtime_events = vec![json!({ + "type": "session.updated", + "session": { "id": "sess_1", "instructions": "backend prompt" } + })]; + if active_handoff { + realtime_events.extend([ json!({ "type": "conversation.input_transcript.delta", "delta": "delegate hello" @@ -4043,10 +4051,9 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re "item_id": "item_1", "input_transcript": "delegate hello" }), - ], - vec![], - ]]) - .await; + ]); + } + let realtime_server = start_websocket_server(vec![vec![realtime_events, vec![]]]).await; let mut builder = test_codex().with_config({ let realtime_base_url = realtime_server.uri().to_string(); @@ -4055,7 +4062,7 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re config.realtime.version = RealtimeWsVersion::V1; } }); - let test = builder.build(&api_server).await?; + let test = builder.build_with_auto_env(&api_server).await?; test.codex .submit(Op::RealtimeConversationStart(ConversationStartParams { @@ -4095,28 +4102,31 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re .await; assert_eq!(session_updated, "sess_1"); - let _ = wait_for_event_match(&test.codex, |msg| match msg { - EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { - payload: RealtimeEvent::HandoffRequested(handoff), - }) if handoff.handoff_id == "handoff_1" => Some(()), - _ => None, - }) - .await; + if active_handoff { + let _ = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HandoffRequested(handoff), + }) if handoff.handoff_id == "handoff_1" => Some(()), + _ => None, + }) + .await; + } else { + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "hello without a realtime delegation".to_string(), + text_elements: Vec::new(), + }])) + .await?; + } wait_for_event(&test.codex, |event| { matches!(event, EventMsg::TurnComplete(_)) }) .await; - let deadline = tokio::time::Instant::now() + Duration::from_secs(2); - while tokio::time::Instant::now() < deadline { - let connections = realtime_server.connections(); - if connections.len() == 1 && connections[0].len() >= 2 { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - + let append = realtime_server + .wait_for_request(/*connection_index*/ 0, /*request_index*/ 1) + .await; let realtime_connections = realtime_server.connections(); assert_eq!(realtime_connections.len(), 1); assert_eq!(realtime_connections[0].len(), 2); @@ -4124,17 +4134,18 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re realtime_connections[0][0].body_json()["type"].as_str(), Some("session.update") ); + let expected_text = if matches!(phase, Some("commentary" | "partial_answer")) { + "assistant says hi" + } else { + "\"Agent Final Message\":\n\nassistant says hi" + }; assert_eq!( - realtime_connections[0][1].body_json()["type"].as_str(), - Some("conversation.handoff.append") - ); - assert_eq!( - realtime_connections[0][1].body_json()["handoff_id"].as_str(), - Some("handoff_1") - ); - assert_eq!( - realtime_connections[0][1].body_json()["output_text"].as_str(), - Some("\"Agent Final Message\":\n\nassistant says hi") + append.body_json(), + json!({ + "type": "conversation.handoff.append", + "handoff_id": if active_handoff { "handoff_1" } else { "codex" }, + "output_text": expected_text, + }) ); realtime_server.shutdown().await; @@ -4272,8 +4283,12 @@ async fn conversation_relays_only_public_reasoning_as_quiet_status(enabled: bool Ok(()) } +#[test_matrix([None, Some("partial_answer")], [false, true])] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> Result<()> { +async fn conversation_routes_assistant_messages_for_v3_handoff( + phase: Option<&str>, + streamed: bool, +) -> Result<()> { skip_if_no_network!(Ok(())); let initial_commentary_text = "[PRO"; @@ -4282,33 +4297,42 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R let commentary_text = format!("{initial_commentary_text}{first_commentary_delta}{second_commentary_delta}"); let (gate_commentary_done_tx, gate_commentary_done_rx) = oneshot::channel(); - let commentary_item_added = + let mut commentary_item_added = responses::ev_message_item_added("msg_commentary", initial_commentary_text); - let commentary_item_done = responses::ev_assistant_message("msg_commentary", &commentary_text); + let mut commentary_item_done = + responses::ev_assistant_message("msg_commentary", &commentary_text); + if let Some(phase) = phase { + commentary_item_added["item"]["phase"] = json!(phase); + commentary_item_done["item"]["phase"] = json!(phase); + } let initial_final_text = "[DO"; let final_delta = "NE]done"; let final_text = format!("{initial_final_text}{final_delta}"); let final_item_added = responses::ev_message_item_added("msg_final", initial_final_text); let final_item_done = responses::ev_assistant_message("msg_final", &final_text); - let response_chunks = vec![ + let mut response_chunks = vec![StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_response_created("resp_stream")), + }]; + if streamed { + response_chunks.extend([ + StreamingSseChunk { + gate: None, + body: sse_event(commentary_item_added), + }, + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_output_text_delta(first_commentary_delta)), + }, + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_output_text_delta(&second_commentary_delta)), + }, + ]); + } + response_chunks.extend([ StreamingSseChunk { - gate: None, - body: sse_event(responses::ev_response_created("resp_stream")), - }, - StreamingSseChunk { - gate: None, - body: sse_event(commentary_item_added), - }, - StreamingSseChunk { - gate: None, - body: sse_event(responses::ev_output_text_delta(first_commentary_delta)), - }, - StreamingSseChunk { - gate: None, - body: sse_event(responses::ev_output_text_delta(&second_commentary_delta)), - }, - StreamingSseChunk { - gate: Some(gate_commentary_done_rx), + gate: streamed.then_some(gate_commentary_done_rx), body: sse_event(commentary_item_done), }, StreamingSseChunk { @@ -4327,7 +4351,7 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R gate: None, body: sse_event(responses::ev_completed("resp_stream")), }, - ]; + ]); let (api_server, completions) = start_streaming_sse_server(vec![response_chunks]).await; let realtime_server = start_websocket_server(vec![vec![ @@ -4410,15 +4434,17 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R }) .await; - let _ = wait_for_event_match(&test.codex, |msg| match msg { - EventMsg::AgentMessageContentDelta(event) - if event.item_id == "msg_commentary" && event.delta == first_commentary_delta => - { - Some(()) - } - _ => None, - }) - .await; + if streamed { + let _ = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::AgentMessageContentDelta(event) + if event.item_id == "msg_commentary" && event.delta == first_commentary_delta => + { + Some(()) + } + _ => None, + }) + .await; + } assert_eq!( wait_for_websocket_request( &realtime_server, @@ -4430,7 +4456,7 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R json!({ "type": "delegation.context.append", "delegation_item_id": "delegation_stream", - "channel": "commentary", + "channel": if phase.is_some() { "speakable" } else { "commentary" }, "content": [{ "type": "input_text", "text": commentary_text }] }) ); @@ -4477,20 +4503,38 @@ async fn conversation_flushes_assistant_deltas_every_200ms_for_v3_handoff() -> R Ok(()) } +#[test_matrix(["commentary", "partial_answer"], [false, true])] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> Result<()> { +async fn conversation_handoff_persists_across_item_done_until_turn_complete( + phase: &str, + streamed: bool, +) -> Result<()> { skip_if_no_network!(Ok(())); let (gate_second_message_tx, gate_second_message_rx) = oneshot::channel(); let mut commentary_message = responses::ev_assistant_message("msg-1", "assistant message 1"); - commentary_message["item"]["phase"] = json!("commentary"); + commentary_message["item"]["phase"] = json!(phase); let mut final_message = responses::ev_assistant_message("msg-2", "assistant message 2"); final_message["item"]["phase"] = json!("final_answer"); - let first_chunks = vec![ - StreamingSseChunk { - gate: None, - body: sse_event(responses::ev_response_created("resp-1")), - }, + let mut first_chunks = vec![StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_response_created("resp-1")), + }]; + if streamed { + let mut added = responses::ev_message_item_added("msg-1", ""); + added["item"]["phase"] = json!(phase); + first_chunks.extend([ + StreamingSseChunk { + gate: None, + body: sse_event(added), + }, + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_output_text_delta("assistant message 1")), + }, + ]); + } + first_chunks.extend([ StreamingSseChunk { gate: None, body: sse_event(commentary_message), @@ -4503,7 +4547,7 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> gate: None, body: sse_event(responses::ev_completed("resp-1")), }, - ]; + ]); let (api_server, completions) = start_streaming_sse_server(vec![first_chunks]).await; let realtime_server = start_websocket_server(vec![vec![ diff --git a/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs b/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs index 142c1ea930..8f56890de4 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation/normalize.rs @@ -377,7 +377,9 @@ fn role_from_str(role: &str) -> Option { fn channel_from_phase(phase: &str) -> Option { match phase { "commentary" => Some(ConversationChannel::Commentary), - "final_answer" => Some(ConversationChannel::Final), + // Both answer phases are Harmony final-channel text; only the latter + // declares a terminal answer. + "partial_answer" | "final_answer" => Some(ConversationChannel::Final), "summary" => Some(ConversationChannel::Summary), _ => None, } diff --git a/codex-rs/rollout-trace/src/reducer/conversation_tests.rs b/codex-rs/rollout-trace/src/reducer/conversation_tests.rs index 4c4f7ace97..4a58e50304 100644 --- a/codex-rs/rollout-trace/src/reducer/conversation_tests.rs +++ b/codex-rs/rollout-trace/src/reducer/conversation_tests.rs @@ -89,6 +89,13 @@ fn response_outputs_enter_thread_conversation_on_completion() -> anyhow::Result< { "type": "message", "role": "assistant", + "phase": "partial_answer", + "content": [{"type": "output_text", "text": "first tests passed"}] + }, + { + "type": "message", + "role": "assistant", + "phase": "final_answer", "content": [{"type": "output_text", "text": "tests passed"}] } ] @@ -101,7 +108,30 @@ fn response_outputs_enter_thread_conversation_on_completion() -> anyhow::Result< let mut expected_thread_items = inference.request_item_ids.clone(); expected_thread_items.extend(inference.response_item_ids.clone()); - assert_eq!(inference.response_item_ids.len(), 1); + assert_eq!( + inference + .response_item_ids + .iter() + .map(|id| { + let item = &rollout.conversation_items[id]; + (item.channel.clone(), item.body.parts.clone()) + }) + .collect::>(), + vec![ + ( + Some(ConversationChannel::Final), + vec![ConversationPart::Text { + text: "first tests passed".into(), + }] + ), + ( + Some(ConversationChannel::Final), + vec![ConversationPart::Text { + text: "tests passed".into(), + }] + ), + ], + ); assert_eq!( rollout.threads["thread-root"].conversation_item_ids, expected_thread_items, diff --git a/codex-rs/thread-store/src/local/thread_history/search.rs b/codex-rs/thread-store/src/local/thread_history/search.rs index 7fa10e2419..0fe9f4de4f 100644 --- a/codex-rs/thread-store/src/local/thread_history/search.rs +++ b/codex-rs/thread-store/src/local/thread_history/search.rs @@ -125,7 +125,13 @@ FROM ( ON turns.thread_id = items.thread_id AND turns.turn_id = items.turn_id WHERE items.thread_id = ? - AND items.item_type = 'userMessage' + AND ( + items.item_type = 'userMessage' + OR ( + items.item_type = 'agentMessage' + AND json_extract(items.item_json, '$.phase') = 'partial_answer' + ) + ) AND items.rollout_ordinal >= ? AND items.rollout_ordinal < ? AND turns.rollout_ordinal >= ? diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index 058cc29a12..b7c0658be6 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -1769,7 +1769,7 @@ WHERE thread_id = ? AND turn_id = ? } #[tokio::test] -async fn summary_items_use_final_answers_and_ignore_commentary() { +async fn summary_items_use_final_answers_and_ignore_nonterminal_messages() { let home = TempDir::new().expect("temp dir"); let config = test_config(home.path()); let thread_id = ThreadId::default(); @@ -1815,6 +1815,11 @@ async fn summary_items_use_final_answers_and_ignore_commentary() { "turn-1", agent_message("commentary-1", MessagePhase::Commentary), ), + completed_item( + thread_id, + "turn-1", + agent_message("partial-1", MessagePhase::PartialAnswer), + ), completed_item( thread_id, "turn-1", @@ -1877,11 +1882,16 @@ WHERE thread_id = ? AND turn_id = ? "turn-2", agent_message("commentary-2", MessagePhase::Commentary), ), + completed_item( + thread_id, + "turn-2", + agent_message("partial-2", MessagePhase::PartialAnswer), + ), turn_completed("turn-2"), ], }) .await - .expect("append commentary-only turn"); + .expect("append turn without a final answer"); let summary = store .list_turns(ListTurnsParams {