diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 57383e8ca8..c84c85628f 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -12,13 +12,13 @@ use crate::agent_communication::AgentCommunicationKind; use crate::codex_thread::ThreadConfigSnapshot; use crate::config::Config; use crate::config::RolloutBudgetConfig; +use crate::context::SubagentNotification; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::rollout_budget::RolloutBudget; use crate::session::emit_subagent_session_started; use crate::session::multi_agents::ResolvedMultiAgentV2UsageHints; use crate::session_prefix::format_inter_agent_completion_message; use crate::session_prefix::format_subagent_context_line; -use crate::session_prefix::format_subagent_notification_message; use crate::thread_manager::ResumeThreadWithHistoryOptions; use crate::thread_manager::ThreadIdGenerator; use crate::thread_manager::ThreadManagerState; @@ -592,12 +592,14 @@ impl AgentControl { .await; return; } - let message = format_subagent_notification_message(child_reference.as_str(), &status); let Ok(parent_thread) = state.get_thread(parent_thread_id).await else { return; }; parent_thread - .inject_user_message_without_turn(message) + .inject_fragment_without_turn(SubagentNotification::new( + child_reference.as_str(), + status, + )) .await; }); } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 882b39488d..04f7c9938b 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -119,6 +119,18 @@ fn rollout_response_item(item: ResponseItem) -> RolloutItem { RolloutItem::ResponseItem(item.into()) } +fn user_message(text: &str) -> ResponseItem { + ResponseItem::Message { + id: None, + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: text.to_string(), + }], + phase: None, + internal_chat_message_metadata_passthrough: None, + } +} + fn assistant_message(text: &str, phase: Option) -> ResponseItem { ResponseItem::Message { id: None, @@ -370,8 +382,9 @@ async fn persist_thread_for_tree_resume(thread: &Arc, message: &str .abort_all_tasks(TurnAbortReason::Interrupted) .await; thread - .inject_user_message_without_turn(message.to_string()) - .await; + .inject_response_items(vec![user_message(message)]) + .await + .expect("inject thread resume context"); thread .session .ensure_rollout_materialized(PersistContext::Standard) @@ -1013,8 +1026,9 @@ async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; parent_thread - .inject_user_message_without_turn("paginated parent context".to_string()) - .await; + .inject_response_items(vec![user_message("paginated parent context")]) + .await + .expect("inject paginated parent context"); let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-paginated".to_string(); parent_thread @@ -1181,8 +1195,9 @@ async fn spawn_agent_without_fork_from_paginated_parent_stays_fresh_and_paginate let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; parent_thread - .inject_user_message_without_turn("parent-only context".to_string()) - .await; + .inject_response_items(vec![user_message("parent-only context")]) + .await + .expect("inject parent-only context"); let child_thread_id = harness .spawn_anonymous_child( @@ -1336,7 +1351,11 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { let parent_thread_id = new_thread.thread_id; let parent_thread = new_thread.thread; parent_thread - .inject_user_message_without_turn("parent seed context".to_string()) + .session + .inject_no_new_turn( + vec![user_message("parent seed context")], + /*current_turn_context*/ None, + ) .await; let expected_parent_seed = parent_thread .session @@ -2240,8 +2259,9 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { let (parent_thread_id, parent_thread) = harness.start_thread().await; parent_thread - .inject_user_message_without_turn("old parent context".to_string()) - .await; + .inject_response_items(vec![user_message("old parent context")]) + .await + .expect("inject old parent context"); let queued_communication = InterAgentCommunication::new( AgentPath::root(), AgentPath::try_from("/root/worker").expect("agent path"), @@ -2274,8 +2294,9 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { ) .await; parent_thread - .inject_user_message_without_turn("current parent task".to_string()) - .await; + .inject_response_items(vec![user_message("current parent task")]) + .await + .expect("inject current parent task"); let spawn_turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-last-n".to_string(); parent_thread @@ -2406,8 +2427,9 @@ async fn spawn_agent_fork_last_n_turns_drops_parent_startup_prefix_when_under_li ) .await; parent_thread - .inject_user_message_without_turn("current parent task".to_string()) - .await; + .inject_response_items(vec![user_message("current parent task")]) + .await + .expect("inject current parent task"); let spawn_turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-last-n-under-limit".to_string(); parent_thread @@ -2510,8 +2532,9 @@ async fn spawn_agent_fork_last_n_turns_strips_parent_usage_hints() { let parent_thread_id = new_thread.thread_id; let parent_thread = new_thread.thread; parent_thread - .inject_user_message_without_turn("parent task".to_string()) - .await; + .inject_response_items(vec![user_message("parent task")]) + .await + .expect("inject parent task"); let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-last-n-usage-hints".to_string(); parent_thread diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 78dbe67fdf..a84a4f1888 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -1,5 +1,6 @@ use crate::agent::AgentStatus; use crate::config::ConstraintResult; +use crate::context::ContextualUserFragment; use crate::elicitation::ElicitationRegistration; use crate::session::SessionIo; use crate::session::SessionSettingsUpdate; @@ -25,7 +26,6 @@ use codex_protocol::error::Result as CodexResult; use codex_protocol::mcp::CallToolResult; use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::models::ActivePermissionProfile; -use codex_protocol::models::ContentItem; use codex_protocol::models::PermissionProfile; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; @@ -608,15 +608,9 @@ impl CodexThread { self.session.token_usage_info().await } - /// Records a user-role session-prefix message without creating a new user turn boundary. - pub(crate) async fn inject_user_message_without_turn(&self, message: String) { - let item = ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { text: message }], - phase: None, - internal_chat_message_metadata_passthrough: None, - }; + /// Records a context fragment without creating a new user turn boundary. + pub(crate) async fn inject_fragment_without_turn(&self, fragment: impl ContextualUserFragment) { + let item = ContextualUserFragment::into(fragment); self.session .inject_no_new_turn(vec![item], /*current_turn_context*/ None) .await; diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 50b8e003e4..4dbbd3a424 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -4,6 +4,8 @@ use std::time::Instant; use crate::Prompt; use crate::client::ModelClientSession; use crate::client_common::ResponseEvent; +use crate::context::CompactionSummary; +use crate::context::ContextualUserFragment; use crate::context::world_state::WorldState; use crate::hook_runtime::PostCompactHookOutcome; use crate::hook_runtime::PreCompactHookOutcome; @@ -723,13 +725,9 @@ fn build_compacted_history_with_limit( summary_text.to_string() }; - history.push(ResponseItemEnvelope::new(ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { text: summary_text }], - phase: None, - internal_chat_message_metadata_passthrough: None, - })); + history.push(ResponseItemEnvelope::new(ContextualUserFragment::into( + CompactionSummary::new(summary_text), + ))); history } diff --git a/codex-rs/core/src/context/compaction_summary.rs b/codex-rs/core/src/context/compaction_summary.rs new file mode 100644 index 0000000000..5c576275f9 --- /dev/null +++ b/codex-rs/core/src/context/compaction_summary.rs @@ -0,0 +1,37 @@ +use super::ContextualUserFragment; +use codex_protocol::models::ContentItemKind; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct CompactionSummary { + summary: String, +} + +impl CompactionSummary { + pub(crate) fn new(summary: impl Into) -> Self { + Self { + summary: summary.into(), + } + } +} + +impl ContextualUserFragment for CompactionSummary { + fn content_kind(&self) -> ContentItemKind { + ContentItemKind("compaction.summary".to_string()) + } + + fn role(&self) -> &'static str { + "user" + } + + fn markers(&self) -> (&'static str, &'static str) { + Self::type_markers() + } + + fn type_markers() -> (&'static str, &'static str) { + ("", "") + } + + fn body(&self) -> String { + self.summary.clone() + } +} diff --git a/codex-rs/core/src/context/guardian_approved_action.rs b/codex-rs/core/src/context/guardian_approved_action.rs new file mode 100644 index 0000000000..b206b058b3 --- /dev/null +++ b/codex-rs/core/src/context/guardian_approved_action.rs @@ -0,0 +1,48 @@ +use super::ContextualUserFragment; +use crate::guardian::AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX; +use codex_protocol::models::ContentItemKind; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct GuardianApprovedAction { + approved_action_json: String, +} + +impl GuardianApprovedAction { + pub(crate) fn new(approved_action_json: String) -> Self { + Self { + approved_action_json, + } + } +} + +impl ContextualUserFragment for GuardianApprovedAction { + fn content_kind(&self) -> ContentItemKind { + ContentItemKind("guardian.approved_action".to_string()) + } + + fn role(&self) -> &'static str { + "developer" + } + + fn markers(&self) -> (&'static str, &'static str) { + Self::type_markers() + } + + fn type_markers() -> (&'static str, &'static str) { + ("", "") + } + + fn body(&self) -> String { + let approval_prefix = AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX; + let approved_action_json = &self.approved_action_json; + format!( + r#"{approval_prefix} + +Treat this as approval to perform that exact action in the same context in which it was originally requested. +Do not assume this also authorizes similar operations with different payloads. + +Approved action: +{approved_action_json}"# + ) + } +} diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index ae8537bdad..a2a68544e7 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/codex-rs/core/src/context/mod.rs @@ -3,10 +3,12 @@ mod approved_command_prefix_saved; mod apps_instructions; mod available_plugins_instructions; +mod compaction_summary; mod contextual_user_message; mod current_time_reminder; mod environment_context; mod environments_instructions; +mod guardian_approved_action; mod guardian_followup_review_reminder; mod guardian_node_repl_policy; mod guardian_review_evidence; @@ -47,10 +49,12 @@ pub(crate) use available_plugins_instructions::AvailablePluginsInstructions; pub(crate) use codex_context_fragments::AdditionalContextDeveloperFragment; pub(crate) use codex_context_fragments::AdditionalContextUserFragment; pub use codex_context_fragments::ContextualUserFragment; +pub(crate) use compaction_summary::CompactionSummary; pub(crate) use contextual_user_message::is_contextual_user_fragment; pub(crate) use contextual_user_message::parse_visible_hook_prompt_message; pub(crate) use current_time_reminder::CurrentTimeReminder; pub(crate) use environments_instructions::EnvironmentsInstructions; +pub(crate) use guardian_approved_action::GuardianApprovedAction; pub(crate) use guardian_followup_review_reminder::GuardianFollowupReviewReminder; pub(crate) use guardian_node_repl_policy::GuardianNodeReplPolicy; pub use guardian_review_evidence::GuardianReviewEvidence; diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 644bc84fad..28fbc46b3c 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -15,6 +15,8 @@ use crate::session::thread_settings; use crate::session::turn_input; use crate::config::Config; +use crate::context::ContextualUserFragment; +use crate::context::GuardianApprovedAction; use crate::context::NodeReplReviewEvidence; use crate::review_prompts::resolve_review_request; use crate::session::spawn_review_thread; @@ -23,9 +25,6 @@ use crate::tasks::UserShellCommandMode; use crate::tasks::UserShellCommandTask; use crate::tasks::execute_user_shell_command; use codex_history::RolloutItem; -use codex_protocol::models::ContentItem; -use codex_protocol::models::ResponseInputItem; -use codex_protocol::models::ResponseItem; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; @@ -725,21 +724,9 @@ async fn approve_guardian_denied_action(sess: &Arc, event: GuardianAsse return; } }; - let approval_prefix = crate::guardian::AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX; - let text = format!( - r#"{approval_prefix} - -Treat this as approval to perform that exact action in the same context in which it was originally requested. -Do not assume this also authorizes similar operations with different payloads. - -Approved action: -{approved_action_json}"#, - ); - let items = vec![ResponseItem::from(ResponseInputItem::Message { - role: "developer".to_string(), - content: vec![ContentItem::InputText { text }], - phase: None, - })]; + let items = vec![ContextualUserFragment::into(GuardianApprovedAction::new( + approved_action_json, + ))]; sess.inject_no_new_turn(items, /*current_turn_context*/ None) .await; diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 1d9848511d..002b69265e 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -3,6 +3,8 @@ use super::*; use super::tests::build_world_state_from_turn_context; use super::tests::make_session_and_context; use super::tests::raw_history_items; +use crate::context::CompactionSummary; +use crate::context::ContextualUserFragment; use codex_history::CompactedItem; use codex_history::InitialHistory; use codex_history::ResponseItemEnvelope; @@ -1268,7 +1270,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_ reconstructed.history, annotated(vec![ user_message("before compact"), - user_message("legacy summary"), + ContextualUserFragment::into(CompactionSummary::new("legacy summary")), ]) ); assert!(reconstructed.reference_context_item.is_none()); diff --git a/codex-rs/core/src/session_prefix.rs b/codex-rs/core/src/session_prefix.rs index 7d4a74a02e..315ef1cabf 100644 --- a/codex-rs/core/src/session_prefix.rs +++ b/codex-rs/core/src/session_prefix.rs @@ -5,7 +5,6 @@ use codex_utils_output_truncation::truncate_text; use crate::context::ContextualUserFragment; use crate::context::InterAgentCompletionMessage; -use crate::context::SubagentNotification; const COMPLETION_MESSAGE_MAX_TOKENS: usize = 1_000; const COMPLETION_MESSAGE_ENVELOPE_TOKEN_RESERVE: usize = 100; @@ -17,13 +16,6 @@ const ERROR_NEXT_ACTION: &str = "This agent's turn failed. If you still need thi // messages but are not user intent. // TODO(jif) unify with structured schema -pub(crate) fn format_subagent_notification_message( - agent_reference: &str, - status: &AgentStatus, -) -> String { - SubagentNotification::new(agent_reference, status.clone()).render() -} - pub(crate) fn format_inter_agent_completion_message( task_name: AgentPath, sender: AgentPath, diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 8995126b7e..d94239d4ac 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -710,6 +710,29 @@ async fn summarize_context_three_requests_and_instructions() { regular_turn_context_count += 1; } RolloutItem::Compacted(ci) if ci.message == expected_summary_message => { + let summary_item = ci + .replacement_history + .as_ref() + .and_then(|history| history.last()) + .expect("compacted history should retain its summary"); + let summary_item = + serde_json::to_value(&summary_item.item).expect("serialize compacted summary"); + assert_eq!( + json!({ + "role": summary_item["role"], + "content": summary_item["content"], + "content_item_kinds": summary_item + ["internal_chat_message_metadata_passthrough"]["content_item_kinds"], + }), + json!({ + "role": "user", + "content": [{ + "type": "input_text", + "text": expected_summary_message, + }], + "content_item_kinds": ["compaction.summary"], + }) + ); saw_compacted_summary = true; } _ => {} diff --git a/codex-rs/core/tests/suite/context_annotations.rs b/codex-rs/core/tests/suite/context_annotations.rs index f7db8c50c2..e2b99dd998 100644 --- a/codex-rs/core/tests/suite/context_annotations.rs +++ b/codex-rs/core/tests/suite/context_annotations.rs @@ -8,6 +8,10 @@ use codex_protocol::openai_models::InputModality; use codex_protocol::protocol::AdditionalContextEntry; use codex_protocol::protocol::AdditionalContextKind; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::GuardianAssessmentAction; +use codex_protocol::protocol::GuardianAssessmentEvent; +use codex_protocol::protocol::GuardianAssessmentStatus; +use codex_protocol::protocol::Op; use codex_protocol::user_input::UserInput; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; @@ -17,6 +21,7 @@ use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; use std::collections::BTreeMap; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -66,6 +71,36 @@ async fn first_request_item_types_roles_and_content_annotations() -> Result<()> .build_with_auto_env(&server) .await?; + test.codex + .submit(Op::ApproveGuardianDeniedAction { + event: GuardianAssessmentEvent { + id: "guardian-review".to_string(), + target_item_id: None, + plugin_id: None, + script_path: None, + turn_id: "guardian-turn".to_string(), + started_at_ms: 0, + completed_at_ms: Some(1), + status: GuardianAssessmentStatus::Denied, + risk_level: None, + user_authorization: None, + rationale: None, + decision_source: None, + action: GuardianAssessmentAction::McpToolCall { + server: "example".to_string(), + tool_name: "write".to_string(), + connector_id: None, + connector_name: None, + tool_title: None, + }, + }, + }) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RawResponseItem(_)) + }) + .await; + test.codex .start_or_steer_turn( TurnInputRequest::user_input(vec![ @@ -104,8 +139,55 @@ async fn first_request_item_types_roles_and_content_annotations() -> Result<()> }) .await; - let items = response - .single_request() + let request = response.single_request(); + assert!(request.has_content_kinds(&["guardian.approved_action"])); + let mut guardian_item = request + .input() + .into_iter() + .next() + .expect("guardian approval should be the first context item"); + guardian_item + .as_object_mut() + .expect("guardian approval should be an object") + .remove("id"); + let guardian_metadata = guardian_item["internal_chat_message_metadata_passthrough"] + .as_object_mut() + .expect("guardian approval should have passthrough metadata"); + guardian_metadata.remove("turn_id"); + guardian_metadata.remove("create_time"); + let approved_action = serde_json::to_string_pretty(&serde_json::json!({ + "action": { + "type": "mcp_tool_call", + "server": "example", + "tool_name": "write", + "connector_id": null, + "connector_name": null, + "tool_title": null, + }, + "outcome": "allowed", + }))?; + assert_eq!( + guardian_item, + serde_json::json!({ + "type": "message", + "role": "developer", + "content": [{ + "type": "input_text", + "text": format!( + "The user has manually approved a specific action that was previously `Rejected`.\n\n\ + Treat this as approval to perform that exact action in the same context in which it was originally requested.\n\ + Do not assume this also authorizes similar operations with different payloads.\n\n\ + Approved action:\n\ + {approved_action}" + ), + }], + "internal_chat_message_metadata_passthrough": { + "content_item_kinds": ["guardian.approved_action"], + }, + }) + ); + + let items = request .input() .into_iter() .map(|item| { @@ -118,6 +200,7 @@ async fn first_request_item_types_roles_and_content_annotations() -> Result<()> .collect::>() .join("\n"); insta::assert_snapshot!(items, @r#" + message developer ["guardian.approved_action"] message developer ["generic.developer_instructions","token_budget.context_window_guidance","generic.permissions_instructions","environments.instructions"] message developer ["token_budget.context_window"] message developer ["multi_agent.usage_hint"] diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 9ffa69d844..2956af0a20 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -131,12 +131,6 @@ fn log_field<'a>(line: &'a str, name: &str) -> Option<&'a str> { .map(|value| value.trim_matches('"')) } -fn has_subagent_notification(req: &ResponsesRequest) -> bool { - req.message_input_texts("user") - .iter() - .any(|text| text.contains("")) -} - fn tool_parameter_description(tool: &Value, parameter_name: &str) -> Option { tool.get("parameters") .and_then(|parameters| parameters.get("properties")) @@ -884,7 +878,11 @@ async fn subagent_notification_is_included_without_wait() -> Result<()> { test.submit_turn(TURN_2_NO_WAIT_PROMPT).await?; let turn2_requests = wait_for_requests(&turn2).await?; - assert!(turn2_requests.iter().any(has_subagent_notification)); + assert!( + turn2_requests + .iter() + .any(|request| request.has_content_kinds(&["multi_agent.subagent_notification"])) + ); Ok(()) }