This commit is contained in:
Roy Han
2026-03-05 11:23:09 -08:00
parent 3eb9115cef
commit 2330d8b4dd
8 changed files with 352 additions and 29 deletions
@@ -1265,6 +1265,7 @@ mod tests {
item: CoreTurnItem::UserMessage(CoreUserMessageItem {
id: "user-item-id".to_string(),
content: Vec::new(),
metadata: None,
}),
}),
EventMsg::TurnComplete(TurnCompleteEvent {
@@ -4511,6 +4511,7 @@ mod tests {
path: "app://demo-app".to_string(),
},
],
metadata: None,
});
assert_eq!(
+248 -21
View File
@@ -78,6 +78,10 @@ use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::items::PlanItem;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::items::UserMessageMetadata;
use codex_protocol::items::UserMessagePersonality;
use codex_protocol::items::UserMessageSandboxPolicy;
use codex_protocol::items::UserMessageType;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::PermissionProfile;
@@ -151,6 +155,8 @@ use crate::config::types::McpServerConfig;
use crate::config::types::ShellEnvironmentPolicy;
use crate::context_manager::ContextManager;
use crate::context_manager::TotalTokenUsageBreakdown;
use crate::contextual_user_message::AGENTS_MD_FRAGMENT;
use crate::contextual_user_message::ENVIRONMENT_CONTEXT_FRAGMENT;
use crate::environment_context::EnvironmentContext;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
@@ -169,6 +175,110 @@ pub enum SteerInputError {
EmptyInput,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum UserMessageItemSource {
Prompt,
PromptQueued,
PromptSteering,
}
impl From<PendingInputSource> for UserMessageItemSource {
fn from(value: PendingInputSource) -> Self {
match value {
PendingInputSource::Queued => Self::PromptQueued,
PendingInputSource::Steering => Self::PromptSteering,
}
}
}
fn sandbox_policy_item_metadata(policy: &SandboxPolicy) -> UserMessageSandboxPolicy {
match policy {
SandboxPolicy::ReadOnly { .. } => UserMessageSandboxPolicy::ReadOnly,
SandboxPolicy::WorkspaceWrite { .. } => UserMessageSandboxPolicy::Sandbox,
SandboxPolicy::DangerFullAccess | SandboxPolicy::ExternalSandbox { .. } => {
UserMessageSandboxPolicy::FullAccess
}
}
}
fn personality_item_metadata(personality: Option<Personality>) -> Option<UserMessagePersonality> {
match personality {
Some(Personality::Friendly) => Some(UserMessagePersonality::Friendly),
Some(Personality::Pragmatic) => Some(UserMessagePersonality::Pragmatic),
Some(Personality::None) | None => None,
}
}
fn is_ide_context_user_input(input: &[UserInput]) -> bool {
input.iter().any(|entry| match entry {
UserInput::Text { text_elements, .. } => !text_elements.is_empty(),
UserInput::Skill { .. } | UserInput::Mention { .. } => true,
UserInput::Image { .. } | UserInput::LocalImage { .. } => false,
_ => false,
})
}
fn is_agents_md_message(input: &[UserInput]) -> bool {
input.iter().any(|entry| match entry {
UserInput::Text { text, .. } => AGENTS_MD_FRAGMENT.matches_text(text),
UserInput::Image { .. }
| UserInput::LocalImage { .. }
| UserInput::Skill { .. }
| UserInput::Mention { .. } => false,
_ => false,
})
}
fn is_environment_context_message(input: &[UserInput]) -> bool {
input.iter().any(|entry| match entry {
UserInput::Text { text, .. } => ENVIRONMENT_CONTEXT_FRAGMENT.matches_text(text),
UserInput::Image { .. }
| UserInput::LocalImage { .. }
| UserInput::Skill { .. }
| UserInput::Mention { .. } => false,
_ => false,
})
}
fn user_message_type_metadata(
input: &[UserInput],
source: UserMessageItemSource,
) -> UserMessageType {
if is_environment_context_message(input) {
return UserMessageType::EnvironmentContext;
}
if is_agents_md_message(input) {
// POC note: we can reliably detect AGENTS.md payloads, but not yet distinguish
// between default and custom sources in all call paths.
return UserMessageType::AgentsMdCustom;
}
if is_ide_context_user_input(input) {
return UserMessageType::PromptWithIdeContext;
}
match source {
UserMessageItemSource::Prompt => UserMessageType::Prompt,
UserMessageItemSource::PromptQueued => UserMessageType::PromptQueued,
UserMessageItemSource::PromptSteering => UserMessageType::PromptSteering,
}
}
fn build_user_message_metadata(
turn_context: &TurnContext,
input: &[UserInput],
source: UserMessageItemSource,
) -> UserMessageMetadata {
UserMessageMetadata {
is_plan_mode: Some(turn_context.collaboration_mode.mode == ModeKind::Plan),
is_tool_call_escalated: None,
escalation_status: None,
sandbox_policy: Some(sandbox_policy_item_metadata(
turn_context.sandbox_policy.get(),
)),
user_message_type: Some(user_message_type_metadata(input, source)),
personality: personality_item_metadata(turn_context.personality),
}
}
/// Notes from the previous real user turn.
///
/// Conceptually this is the same role that `previous_model` used to fill, but
@@ -262,6 +372,8 @@ use crate::skills::injection::app_id_from_path;
use crate::skills::injection::tool_kind_for_path;
use crate::skills::resolve_skill_dependencies_for_turn;
use crate::state::ActiveTurn;
use crate::state::PendingInputItem;
use crate::state::PendingInputSource;
use crate::state::SessionServices;
use crate::state::SessionState;
use crate::state_db;
@@ -3336,13 +3448,16 @@ impl Session {
turn_context: &TurnContext,
input: &[UserInput],
response_item: ResponseItem,
source: UserMessageItemSource,
) {
// Persist the user message to history, but emit the turn item from `UserInput` so
// UI-only `text_elements` are preserved. `ResponseItem::Message` does not carry
// those spans, and `record_response_item_and_emit_turn_item` would drop them.
self.record_conversation_items(turn_context, std::slice::from_ref(&response_item))
.await;
let turn_item = TurnItem::UserMessage(UserMessageItem::new(input));
let metadata = build_user_message_metadata(turn_context, input, source);
let turn_item =
TurnItem::UserMessage(UserMessageItem::new_with_metadata(input, Some(metadata)));
self.emit_turn_item_started(turn_context, &turn_item).await;
self.emit_turn_item_completed(turn_context, turn_item).await;
self.ensure_rollout_materialized().await;
@@ -3436,7 +3551,7 @@ impl Session {
}
let mut turn_state = active_turn.turn_state.lock().await;
turn_state.push_pending_input(input.into());
turn_state.push_pending_input(input.into(), PendingInputSource::Steering);
Ok(active_turn_id.clone())
}
@@ -3450,7 +3565,7 @@ impl Session {
Some(at) => {
let mut ts = at.turn_state.lock().await;
for item in input {
ts.push_pending_input(item);
ts.push_pending_input(item, PendingInputSource::Queued);
}
Ok(())
}
@@ -3458,7 +3573,7 @@ impl Session {
}
}
pub async fn get_pending_input(&self) -> Vec<ResponseInputItem> {
pub async fn get_pending_input(&self) -> Vec<PendingInputItem> {
let mut active = self.active_turn.lock().await;
match active.as_mut() {
Some(at) => {
@@ -5028,8 +5143,13 @@ pub(crate) async fn run_turn(
let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input.clone());
let response_item: ResponseItem = initial_input_for_turn.clone().into();
sess.record_user_prompt_and_emit_turn_item(turn_context.as_ref(), &input, response_item)
.await;
sess.record_user_prompt_and_emit_turn_item(
turn_context.as_ref(),
&input,
response_item,
UserMessageItemSource::Prompt,
)
.await;
// Track the previous-turn baseline from the regular user-turn path only so
// standalone tasks (compact/shell/review/undo) cannot suppress future
// model/realtime injections.
@@ -5065,17 +5185,23 @@ pub(crate) async fn run_turn(
.get_pending_input()
.await
.into_iter()
.map(ResponseItem::from)
.collect::<Vec<ResponseItem>>();
.map(|pending| {
(
UserMessageItemSource::from(pending.source),
ResponseItem::from(pending.input),
)
})
.collect::<Vec<(UserMessageItemSource, ResponseItem)>>();
if !pending_response_items.is_empty() {
for response_item in pending_response_items {
for (source, response_item) in pending_response_items {
if let Some(TurnItem::UserMessage(user_message)) = parse_turn_item(&response_item) {
// todo(aibrahim): move pending input to be UserInput only to keep TextElements. context: https://github.com/openai/codex/pull/10656#discussion_r2765522480
sess.record_user_prompt_and_emit_turn_item(
turn_context.as_ref(),
&user_message.content,
response_item,
source,
)
.await;
} else {
@@ -9628,30 +9754,70 @@ mod tests {
.await
.expect("expected item started event")
.expect("channel open");
assert!(matches!(
second.msg,
EventMsg::ItemStarted(ItemStartedEvent {
item: TurnItem::UserMessage(UserMessageItem { content, .. }),
..
}) if content == vec![UserInput::Text {
let EventMsg::ItemStarted(ItemStartedEvent {
item:
TurnItem::UserMessage(UserMessageItem {
content,
metadata: Some(metadata),
..
}),
..
}) = second.msg
else {
panic!("expected started user message item");
};
assert_eq!(
content,
vec![UserInput::Text {
text: "late pending input".to_string(),
text_elements: Vec::new(),
}]
);
assert_eq!(
metadata.user_message_type,
Some(UserMessageType::PromptQueued)
);
assert_eq!(metadata.is_plan_mode, Some(false));
assert!(matches!(
metadata.sandbox_policy,
Some(UserMessageSandboxPolicy::ReadOnly)
| Some(UserMessageSandboxPolicy::Sandbox)
| Some(UserMessageSandboxPolicy::FullAccess)
));
let third = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("expected item completed event")
.expect("channel open");
assert!(matches!(
third.msg,
EventMsg::ItemCompleted(ItemCompletedEvent {
item: TurnItem::UserMessage(UserMessageItem { content, .. }),
..
}) if content == vec![UserInput::Text {
let EventMsg::ItemCompleted(ItemCompletedEvent {
item:
TurnItem::UserMessage(UserMessageItem {
content,
metadata: Some(metadata),
..
}),
..
}) = third.msg
else {
panic!("expected completed user message item");
};
assert_eq!(
content,
vec![UserInput::Text {
text: "late pending input".to_string(),
text_elements: Vec::new(),
}]
);
assert_eq!(
metadata.user_message_type,
Some(UserMessageType::PromptQueued)
);
assert_eq!(metadata.is_plan_mode, Some(false));
assert!(matches!(
metadata.sandbox_policy,
Some(UserMessageSandboxPolicy::ReadOnly)
| Some(UserMessageSandboxPolicy::Sandbox)
| Some(UserMessageSandboxPolicy::FullAccess)
));
let fourth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
@@ -9684,6 +9850,67 @@ mod tests {
));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn task_finish_marks_steer_input_as_prompt_steering_metadata() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let input = vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}];
sess.spawn_task(
Arc::clone(&tc),
input,
NeverEndingTask {
kind: TaskKind::Regular,
listen_to_cancellation_token: false,
},
)
.await;
while rx.try_recv().is_ok() {}
let steer_turn_id = sess
.steer_input(
vec![UserInput::Text {
text: "steered input".to_string(),
text_elements: Vec::new(),
}],
Some(tc.sub_id.as_str()),
)
.await
.expect("steer input should enqueue while task is active");
assert_eq!(steer_turn_id, tc.sub_id);
sess.on_task_finished(Arc::clone(&tc), None).await;
let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("expected raw response item event")
.expect("channel open");
assert!(matches!(first.msg, EventMsg::RawResponseItem(_)));
let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("expected item started event")
.expect("channel open");
let EventMsg::ItemStarted(ItemStartedEvent {
item:
TurnItem::UserMessage(UserMessageItem {
metadata: Some(metadata),
..
}),
..
}) = second.msg
else {
panic!("expected started user message item");
};
assert_eq!(
metadata.user_message_type,
Some(UserMessageType::PromptSteering)
);
assert_eq!(metadata.is_plan_mode, Some(false));
}
#[tokio::test]
async fn steer_input_requires_active_turn() {
let (sess, _tc, _rx) = make_session_and_context_with_rx().await;
+2
View File
@@ -5,5 +5,7 @@ mod turn;
pub(crate) use service::SessionServices;
pub(crate) use session::SessionState;
pub(crate) use turn::ActiveTurn;
pub(crate) use turn::PendingInputItem;
pub(crate) use turn::PendingInputSource;
pub(crate) use turn::RunningTask;
pub(crate) use turn::TaskKind;
+20 -4
View File
@@ -40,6 +40,18 @@ pub(crate) enum TaskKind {
Compact,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum PendingInputSource {
Queued,
Steering,
}
#[derive(Clone, Debug)]
pub(crate) struct PendingInputItem {
pub(crate) input: ResponseInputItem,
pub(crate) source: PendingInputSource,
}
pub(crate) struct RunningTask {
pub(crate) done: Arc<Notify>,
pub(crate) kind: TaskKind,
@@ -73,7 +85,7 @@ pub(crate) struct TurnState {
pending_approvals: HashMap<String, oneshot::Sender<ReviewDecision>>,
pending_user_input: HashMap<String, oneshot::Sender<RequestUserInputResponse>>,
pending_dynamic_tools: HashMap<String, oneshot::Sender<DynamicToolResponse>>,
pending_input: Vec<ResponseInputItem>,
pending_input: Vec<PendingInputItem>,
pub(crate) tool_calls: u64,
pub(crate) token_usage_at_turn_start: TokenUsage,
}
@@ -131,11 +143,15 @@ impl TurnState {
self.pending_dynamic_tools.remove(key)
}
pub(crate) fn push_pending_input(&mut self, input: ResponseInputItem) {
self.pending_input.push(input);
pub(crate) fn push_pending_input(
&mut self,
input: ResponseInputItem,
source: PendingInputSource,
) {
self.pending_input.push(PendingInputItem { input, source });
}
pub(crate) fn take_pending_input(&mut self) -> Vec<ResponseInputItem> {
pub(crate) fn take_pending_input(&mut self) -> Vec<PendingInputItem> {
if self.pending_input.is_empty() {
Vec::with_capacity(0)
} else {
+11 -4
View File
@@ -21,6 +21,7 @@ use tracing::warn;
use crate::AuthManager;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::codex::UserMessageItemSource;
use crate::contextual_user_message::TURN_ABORTED_OPEN_TAG;
use crate::event_mapping::parse_turn_item;
use crate::models_manager::manager::ModelsManager;
@@ -29,11 +30,11 @@ use crate::protocol::TurnAbortReason;
use crate::protocol::TurnAbortedEvent;
use crate::protocol::TurnCompleteEvent;
use crate::state::ActiveTurn;
use crate::state::PendingInputItem;
use crate::state::RunningTask;
use crate::state::TaskKind;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::user_input::UserInput;
@@ -210,7 +211,7 @@ impl Session {
.cancel_git_enrichment_task();
let mut active = self.active_turn.lock().await;
let mut pending_input = Vec::<ResponseInputItem>::new();
let mut pending_input = Vec::<PendingInputItem>::new();
let mut should_clear_active_turn = false;
let mut token_usage_at_turn_start = None;
let mut turn_tool_calls = 0_u64;
@@ -230,9 +231,14 @@ impl Session {
if !pending_input.is_empty() {
let pending_response_items = pending_input
.into_iter()
.map(ResponseItem::from)
.map(|pending| {
(
UserMessageItemSource::from(pending.source),
ResponseItem::from(pending.input),
)
})
.collect::<Vec<_>>();
for response_item in pending_response_items {
for (source, response_item) in pending_response_items {
if let Some(TurnItem::UserMessage(user_message)) = parse_turn_item(&response_item) {
// Keep leftover user input on the same persistence + lifecycle path as the
// normal pre-sampling drain. This helper records the response item once, then
@@ -241,6 +247,7 @@ impl Session {
turn_context.as_ref(),
&user_message.content,
response_item,
source,
)
.await;
} else {
+68
View File
@@ -29,10 +29,73 @@ pub enum TurnItem {
ContextCompaction(ContextCompactionItem),
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum UserMessageType {
Prompt,
PromptSteering,
PromptQueued,
PromptWithIdeContext,
AgentsMdDefault,
AgentsMdCustom,
EnvironmentContext,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum EscalationStatus {
Approved,
Rejected,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum UserMessageSandboxPolicy {
ReadOnly,
Sandbox,
FullAccess,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum UserMessagePersonality {
Friendly,
Pragmatic,
}
#[derive(Debug, Clone, Default, Deserialize, Serialize, TS, JsonSchema, PartialEq, Eq)]
pub struct UserMessageMetadata {
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub is_plan_mode: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub is_tool_call_escalated: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub escalation_status: Option<EscalationStatus>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub sandbox_policy: Option<UserMessageSandboxPolicy>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub user_message_type: Option<UserMessageType>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub personality: Option<UserMessagePersonality>,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema)]
pub struct UserMessageItem {
pub id: String,
pub content: Vec<UserInput>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
pub metadata: Option<UserMessageMetadata>,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS, JsonSchema)]
@@ -116,9 +179,14 @@ impl Default for ContextCompactionItem {
impl UserMessageItem {
pub fn new(content: &[UserInput]) -> Self {
Self::new_with_metadata(content, None)
}
pub fn new_with_metadata(content: &[UserInput], metadata: Option<UserMessageMetadata>) -> Self {
Self {
id: uuid::Uuid::new_v4().to_string(),
content: content.to_vec(),
metadata,
}
}
+1
View File
@@ -3424,6 +3424,7 @@ fn complete_user_message_for_inputs(chat: &mut ChatWidget, item_id: &str, conten
item: TurnItem::UserMessage(UserMessageItem {
id: item_id.to_string(),
content,
metadata: None,
}),
}),
});