Classify additional generated context fragments (#40275)

## What changed

- Represent compaction summaries and Guardian-approved actions as typed contextual fragments.
- Inject subagent notifications through the same fragment path.
- Annotate the resulting request items with `compaction.summary`, `guardian.approved_action`, and `multi_agent.subagent_notification` content kinds while preserving their existing roles and text.

## Testing

- Verify the content-kind metadata on compacted summaries, approved actions, and subagent notifications.

GitOrigin-RevId: 01c93e7f08b2dc1c1e85ba69e82bfb5f56e6cca5
This commit is contained in:
pakrym-oai
2026-08-23 20:08:40 +00:00
committed by copyberry
parent e21bc763a7
commit 287587c32c
13 changed files with 262 additions and 71 deletions
+5 -3
View File
@@ -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;
});
}
+38 -15
View File
@@ -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<MessagePhase>) -> ResponseItem {
ResponseItem::Message {
id: None,
@@ -370,8 +382,9 @@ async fn persist_thread_for_tree_resume(thread: &Arc<CodexThread>, 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
+4 -10
View File
@@ -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;
+5 -7
View File
@@ -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
}
@@ -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<String>) -> 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()
}
}
@@ -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}"#
)
}
}
+4
View File
@@ -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;
+5 -18
View File
@@ -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<Session>, 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;
@@ -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());
-8
View File
@@ -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,
+23
View File
@@ -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;
}
_ => {}
@@ -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::<Vec<_>>()
.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"]
@@ -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("<subagent_notification>"))
}
fn tool_parameter_description(tool: &Value, parameter_name: &str) -> Option<String> {
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(())
}