Move request state into step context

This commit is contained in:
Ahmed Ibrahim
2026-07-08 21:49:50 -07:00
parent e4c7bdc387
commit ba9a2d63d7
57 changed files with 508 additions and 462 deletions
+7 -3
View File
@@ -11,6 +11,7 @@ use crate::config::ConfigBuilder;
use crate::context::ContextualUserFragment;
use crate::context::SubagentNotification;
use crate::init_state_db;
use crate::session::step_context::StepContext;
use crate::thread_manager::StartThreadOptions;
use assert_matches::assert_matches;
use codex_extension_api::ExtensionDataInit;
@@ -983,7 +984,8 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
],
)
.await;
let parent_reference_context_item = turn_context.to_turn_context_item();
let parent_reference_context_item =
StepContext::for_test(Arc::clone(&turn_context)).to_turn_context_item();
parent_thread
.codex
.session
@@ -1187,7 +1189,9 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() {
previous_window_id: None,
window_id: None,
}),
RolloutItem::TurnContext(turn_context.to_turn_context_item()),
RolloutItem::TurnContext(
StepContext::for_test(Arc::clone(&turn_context)).to_turn_context_item(),
),
RolloutItem::ResponseItem(spawn_agent_call(&parent_spawn_call_id)),
])
.await;
@@ -1375,7 +1379,7 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() {
.codex
.session
.persist_rollout_items(&[RolloutItem::TurnContext(
spawn_turn_context.to_turn_context_item(),
StepContext::for_test(Arc::clone(&spawn_turn_context)).to_turn_context_item(),
)])
.await;
parent_thread
+9 -24
View File
@@ -52,7 +52,6 @@ use crate::session::SUBMISSION_CHANNEL_CAPACITY;
use crate::session::emit_subagent_session_started;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use codex_login::AuthManager;
use codex_models_manager::manager::SharedModelsManager;
use codex_protocol::error::CodexErr;
@@ -79,12 +78,12 @@ pub(crate) async fn run_codex_thread_interactive(
auth_manager: Arc<AuthManager>,
models_manager: SharedModelsManager,
parent_session: Arc<Session>,
parent_ctx: Arc<TurnContext>,
step_context: Arc<StepContext>,
cancel_token: CancellationToken,
subagent_source: SubAgentSource,
initial_history: Option<InitialHistory>,
) -> Result<Codex, CodexErr> {
let parent_ctx = step_context.turn.as_ref();
let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let (tx_ops, rx_ops) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY);
let conversation_history = initial_history.unwrap_or(InitialHistory::New);
@@ -158,7 +157,6 @@ pub(crate) async fn run_codex_thread_interactive(
// Forward events from the sub-agent to the consumer, filtering approvals and
// routing them to the parent session for decisions.
let parent_session_clone = Arc::clone(&parent_session);
let parent_ctx_clone = Arc::clone(&parent_ctx);
let step_context_clone = Arc::clone(&step_context);
let codex_for_events = Arc::clone(&codex);
// Cache the child call's MCP metadata at begin time. The later legacy
@@ -170,7 +168,6 @@ pub(crate) async fn run_codex_thread_interactive(
codex_for_events,
tx_sub,
parent_session_clone,
parent_ctx_clone,
step_context_clone,
pending_mcp_invocations,
cancel_token_events,
@@ -203,7 +200,6 @@ pub(crate) async fn run_codex_thread_one_shot(
models_manager: SharedModelsManager,
input: Vec<UserInput>,
parent_session: Arc<Session>,
parent_ctx: Arc<TurnContext>,
step_context: Arc<StepContext>,
cancel_token: CancellationToken,
subagent_source: SubAgentSource,
@@ -218,7 +214,6 @@ pub(crate) async fn run_codex_thread_one_shot(
auth_manager,
models_manager,
parent_session,
parent_ctx,
step_context,
child_cancel.clone(),
subagent_source,
@@ -284,7 +279,6 @@ async fn forward_events(
codex: Arc<Codex>,
tx_sub: Sender<Event>,
parent_session: Arc<Session>,
parent_ctx: Arc<TurnContext>,
step_context: Arc<StepContext>,
pending_mcp_invocations: Arc<Mutex<HashMap<String, PendingMcpInvocation>>>,
cancel_token: CancellationToken,
@@ -321,7 +315,6 @@ async fn forward_events(
&codex,
id,
&parent_session,
&parent_ctx,
Arc::clone(&step_context),
event,
&cancel_token,
@@ -336,7 +329,6 @@ async fn forward_events(
&codex,
id,
&parent_session,
&parent_ctx,
Arc::clone(&step_context),
event,
&cancel_token,
@@ -350,7 +342,6 @@ async fn forward_events(
handle_request_permissions(
&codex,
&parent_session,
&parent_ctx,
&step_context,
event,
&cancel_token,
@@ -365,7 +356,6 @@ async fn forward_events(
&codex,
id,
&parent_session,
&parent_ctx,
Arc::clone(&step_context),
&pending_mcp_invocations,
event,
@@ -503,11 +493,11 @@ async fn handle_exec_approval(
codex: &Codex,
turn_id: String,
parent_session: &Arc<Session>,
parent_ctx: &Arc<TurnContext>,
step_context: Arc<StepContext>,
event: ExecApprovalRequestEvent,
cancel_token: &CancellationToken,
) {
let parent_ctx = &step_context.turn;
let approval_id_for_op = event.effective_approval_id();
let ExecApprovalRequestEvent {
call_id,
@@ -526,8 +516,7 @@ async fn handle_exec_approval(
let review_cancel = cancel_token.child_token();
let review_rx = spawn_approval_request_review(
Arc::clone(parent_session),
Arc::clone(parent_ctx),
step_context,
step_context.clone(),
new_guardian_review_id(),
GuardianApprovalRequest::Shell {
id: call_id.clone(),
@@ -590,11 +579,11 @@ async fn handle_patch_approval(
codex: &Codex,
_id: String,
parent_session: &Arc<Session>,
parent_ctx: &Arc<TurnContext>,
step_context: Arc<StepContext>,
event: ApplyPatchApprovalRequestEvent,
cancel_token: &CancellationToken,
) {
let parent_ctx = &step_context.turn;
let ApplyPatchApprovalRequestEvent {
call_id,
changes,
@@ -641,8 +630,7 @@ async fn handle_patch_approval(
.join("\n");
let review_rx = spawn_approval_request_review(
Arc::clone(parent_session),
Arc::clone(parent_ctx),
step_context,
step_context.clone(),
new_guardian_review_id(),
GuardianApprovalRequest::ApplyPatch {
id: approval_id.clone(),
@@ -694,16 +682,15 @@ async fn handle_request_user_input(
codex: &Codex,
id: String,
parent_session: &Arc<Session>,
parent_ctx: &Arc<TurnContext>,
step_context: Arc<StepContext>,
pending_mcp_invocations: &Arc<Mutex<HashMap<String, PendingMcpInvocation>>>,
event: RequestUserInputEvent,
cancel_token: &CancellationToken,
) {
let parent_ctx = &step_context.turn;
if let Some(response) = maybe_auto_review_mcp_request_user_input(
parent_session,
parent_ctx,
step_context,
Arc::clone(&step_context),
pending_mcp_invocations,
&event,
cancel_token,
@@ -739,12 +726,12 @@ async fn handle_request_user_input(
/// `McpToolCallBegin` in order to rebuild the full guardian review request.
async fn maybe_auto_review_mcp_request_user_input(
parent_session: &Arc<Session>,
parent_ctx: &Arc<TurnContext>,
step_context: Arc<StepContext>,
pending_mcp_invocations: &Arc<Mutex<HashMap<String, PendingMcpInvocation>>>,
event: &RequestUserInputEvent,
cancel_token: &CancellationToken,
) -> Option<RequestUserInputResponse> {
let parent_ctx = &step_context.turn;
// TODO(ccunningham): Support delegated MCP approval elicitations here too after
// coordinating with @fouad. Today guardian only auto-reviews the RequestUserInput
// compatibility path for delegated MCP approvals.
@@ -767,7 +754,6 @@ async fn maybe_auto_review_mcp_request_user_input(
let review_cancel = cancel_token.child_token();
let review_rx = spawn_approval_request_review(
Arc::clone(parent_session),
Arc::clone(parent_ctx),
step_context,
new_guardian_review_id(),
build_guardian_mcp_tool_review_request(&event.call_id, &invocation, metadata.as_ref()),
@@ -814,11 +800,11 @@ async fn maybe_auto_review_mcp_request_user_input(
async fn handle_request_permissions(
codex: &Codex,
parent_session: &Arc<Session>,
parent_ctx: &Arc<TurnContext>,
step_context: &Arc<StepContext>,
event: RequestPermissionsEvent,
cancel_token: &CancellationToken,
) {
let parent_ctx = &step_context.turn;
let call_id = event.call_id;
let args = RequestPermissionsArgs {
environment_id: event.environment_id,
@@ -830,7 +816,6 @@ async fn handle_request_permissions(
parent_ctx.cwd.clone()
});
let response_fut = parent_session.request_permissions_for_cwd(
parent_ctx,
step_context,
call_id.clone(),
args,
+7 -6
View File
@@ -1,6 +1,7 @@
use super::*;
use crate::mcp_tool_call::MCP_TOOL_APPROVAL_DECLINE_SYNTHETIC;
use crate::mcp_tool_call::MCP_TOOL_APPROVAL_QUESTION_ID_PREFIX;
use crate::session::step_context::StepContext;
use async_channel::bounded;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_protocol::config_types::ApprovalsReviewer;
@@ -69,7 +70,7 @@ async fn forward_events_filters_private_events_before_blocked_send_is_cancelled(
Arc::clone(&codex),
tx_out.clone(),
session,
ctx,
StepContext::for_test(ctx),
Arc::new(Mutex::new(HashMap::new())),
cancel.clone(),
));
@@ -194,7 +195,7 @@ async fn run_codex_thread_interactive_respects_pre_cancelled_spawn() {
Arc::clone(&parent_session.services.auth_manager),
Arc::clone(&parent_session.services.models_manager),
parent_session,
parent_ctx,
StepContext::for_test(parent_ctx),
cancel_token,
SubAgentSource::Review,
/*initial_history*/ None,
@@ -251,7 +252,7 @@ async fn handle_request_permissions_uses_tool_call_id_for_round_trip() {
handle_request_permissions(
codex.as_ref(),
&parent_session,
&parent_ctx,
&StepContext::for_test(Arc::clone(&parent_ctx)),
RequestPermissionsEvent {
call_id: request_call_id,
turn_id: "child-turn-1".to_string(),
@@ -341,7 +342,7 @@ async fn handle_exec_approval_uses_call_id_for_guardian_review_and_approval_id_f
codex.as_ref(),
"child-turn-1".to_string(),
&parent_session,
&parent_ctx,
StepContext::for_test(Arc::clone(&parent_ctx)),
ExecApprovalRequestEvent {
call_id: "command-item-1".to_string(),
approval_id: Some("callback-approval-1".to_string()),
@@ -449,7 +450,7 @@ async fn delegated_mcp_guardian_abort_returns_synthetic_decline_answer() {
let response = maybe_auto_review_mcp_request_user_input(
&parent_session,
&parent_ctx,
StepContext::for_test(Arc::clone(&parent_ctx)),
&pending_mcp_invocations,
&RequestUserInputEvent {
call_id: "call-1".to_string(),
@@ -513,7 +514,7 @@ async fn delegated_mcp_user_reviewer_returns_none_without_metadata() {
};
let response = maybe_auto_review_mcp_request_user_input(
&parent_session,
&parent_ctx,
StepContext::for_test(Arc::clone(&parent_ctx)),
&pending_mcp_invocations,
&event,
&cancel_token,
+7 -17
View File
@@ -70,7 +70,6 @@ pub(crate) enum InitialContextInjection {
pub(crate) async fn build_compaction_initial_context(
sess: &Session,
turn_context: &TurnContext,
step_context: &StepContext,
initial_context_injection: &InitialContextInjection,
) -> (Vec<ResponseItem>, Option<Arc<WorldState>>) {
@@ -78,11 +77,7 @@ pub(crate) async fn build_compaction_initial_context(
match initial_context_injection {
InitialContextInjection::BeforeLastUserMessage(world_state) => {
let items = sess
.build_initial_context_with_world_state(
turn_context,
world_state.as_ref(),
step_context,
)
.build_initial_context_with_world_state(world_state.as_ref(), step_context)
.await;
(items, Some(Arc::clone(world_state)))
}
@@ -96,12 +91,12 @@ pub(crate) fn should_use_remote_compact_task(provider: &ModelProviderInfo) -> bo
pub(crate) async fn run_inline_auto_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
step_context: Arc<StepContext>,
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
let turn_context = step_context.turn.as_ref();
let prompt = turn_context
.config
.compact_prompt
@@ -116,7 +111,6 @@ pub(crate) async fn run_inline_auto_compact_task(
run_compact_task_inner(
sess,
turn_context,
step_context,
input,
initial_context_injection,
@@ -130,10 +124,10 @@ pub(crate) async fn run_inline_auto_compact_task(
pub(crate) async fn run_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
step_context: Arc<StepContext>,
input: Vec<UserInput>,
) -> CodexResult<()> {
let turn_context = step_context.turn.clone();
let start_event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_context.sub_id.clone(),
trace_id: turn_context.trace_id.clone(),
@@ -144,7 +138,6 @@ pub(crate) async fn run_compact_task(
sess.send_event(&turn_context, start_event).await;
run_compact_task_inner(
sess.clone(),
turn_context,
step_context,
input,
InitialContextInjection::DoNotInject,
@@ -158,7 +151,6 @@ pub(crate) async fn run_compact_task(
async fn run_compact_task_inner(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
step_context: Arc<StepContext>,
input: Vec<UserInput>,
initial_context_injection: InitialContextInjection,
@@ -166,6 +158,7 @@ async fn run_compact_task_inner(
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
let turn_context = step_context.turn.clone();
let compaction_metadata =
CompactionTurnMetadata::new(trigger, reason, CompactionImplementation::Responses, phase);
let attempt = CompactionAnalyticsAttempt::begin(
@@ -195,7 +188,6 @@ async fn run_compact_task_inner(
}
let result = run_compact_task_inner_impl(
Arc::clone(&sess),
Arc::clone(&turn_context),
step_context,
input,
initial_context_injection,
@@ -231,12 +223,12 @@ async fn run_compact_task_inner(
async fn run_compact_task_inner_impl(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
step_context: Arc<StepContext>,
input: Vec<UserInput>,
initial_context_injection: InitialContextInjection,
compaction_metadata: CompactionTurnMetadata,
) -> CodexResult<String> {
let turn_context = step_context.turn.clone();
let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new());
sess.emit_turn_item_started(&turn_context, &compaction_item)
.await;
@@ -274,7 +266,6 @@ async fn run_compact_task_inner_impl(
};
let attempt_result = drain_to_completed(
&sess,
turn_context.as_ref(),
step_context.as_ref(),
&mut client_session,
&responses_metadata,
@@ -349,7 +340,6 @@ async fn run_compact_task_inner_impl(
let (initial_context, world_state_baseline) = build_compaction_initial_context(
sess.as_ref(),
turn_context.as_ref(),
step_context.as_ref(),
&initial_context_injection,
)
@@ -361,7 +351,7 @@ async fn run_compact_task_inner_impl(
let reference_context_item = match initial_context_injection {
InitialContextInjection::DoNotInject => None,
InitialContextInjection::BeforeLastUserMessage(_) => {
Some(turn_context.to_turn_context_item(&step_context))
Some(step_context.to_turn_context_item())
}
};
let compacted_item = CompactedItem {
@@ -675,12 +665,12 @@ fn build_compacted_history_with_limit(
async fn drain_to_completed(
sess: &Session,
turn_context: &TurnContext,
step_context: &StepContext,
client_session: &mut ModelClientSession,
responses_metadata: &CodexResponsesMetadata,
prompt: &Prompt,
) -> CodexResult<()> {
let turn_context = step_context.turn.as_ref();
let mut stream = client_session
.stream(
prompt,
+3 -10
View File
@@ -259,7 +259,6 @@ async fn run_remote_compact_task_inner_impl(
let (new_window_number, new_window_ids) = sess.advance_auto_compact_window().await;
let (new_history, world_state_baseline) = process_compacted_history(
sess.as_ref(),
compaction_turn_context.as_ref(),
step_context.as_ref(),
new_history,
&initial_context_injection,
@@ -269,7 +268,7 @@ async fn run_remote_compact_task_inner_impl(
let reference_context_item = match initial_context_injection {
InitialContextInjection::DoNotInject => None,
InitialContextInjection::BeforeLastUserMessage(_) => {
Some(compaction_turn_context.to_turn_context_item(step_context))
Some(step_context.to_turn_context_item())
}
};
let compacted_item = CompactedItem {
@@ -304,7 +303,6 @@ async fn run_remote_compact_task_inner_impl(
pub(crate) async fn process_compacted_history(
sess: &Session,
turn_context: &TurnContext,
step_context: &StepContext,
mut compacted_history: Vec<ResponseItem>,
initial_context_injection: &InitialContextInjection,
@@ -312,13 +310,8 @@ pub(crate) async fn process_compacted_history(
// Mid-turn compaction is the only path that must inject initial context above the last user
// message in the replacement history. Pre-turn compaction instead injects context after the
// compaction item, but mid-turn compaction keeps the compaction item last for model training.
let (initial_context, world_state_baseline) = build_compaction_initial_context(
sess,
turn_context,
step_context,
initial_context_injection,
)
.await;
let (initial_context, world_state_baseline) =
build_compaction_initial_context(sess, step_context, initial_context_injection).await;
compacted_history.retain(should_keep_compacted_history_item);
(
+2 -3
View File
@@ -283,7 +283,6 @@ async fn run_remote_compact_task_inner_impl(
let (new_window_number, new_window_ids) = sess.advance_auto_compact_window().await;
let (new_history, world_state_baseline) = process_compacted_history(
sess.as_ref(),
compaction_turn_context.as_ref(),
step_context.as_ref(),
compacted_history,
&initial_context_injection,
@@ -293,7 +292,7 @@ async fn run_remote_compact_task_inner_impl(
let reference_context_item = match initial_context_injection {
InitialContextInjection::DoNotInject => None,
InitialContextInjection::BeforeLastUserMessage(_) => {
Some(compaction_turn_context.to_turn_context_item(step_context))
Some(step_context.to_turn_context_item())
}
};
let compacted_item = CompactedItem {
@@ -330,12 +329,12 @@ struct RemoteCompactionV2Output {
async fn run_remote_compaction_request_v2(
sess: &Session,
turn_context: &TurnContext,
step_context: &StepContext,
client_session: &mut ModelClientSession,
prompt: &Prompt,
responses_metadata: &CodexResponsesMetadata,
) -> CodexResult<RemoteCompactionV2Output> {
let turn_context = step_context.turn.as_ref();
let max_retries = turn_context
.provider
.info()
@@ -101,7 +101,6 @@ pub(super) async fn run_remote_compact_v2_attempt(
};
let compaction_output_result = run_remote_compaction_request_v2(
sess,
turn_context.as_ref(),
step_context.as_ref(),
client_session,
&prompt,
+4 -2
View File
@@ -1,4 +1,5 @@
use super::*;
use crate::session::step_context::StepContext;
use crate::session::tests::build_world_state_from_turn_context;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::WireApi;
@@ -13,17 +14,18 @@ async fn process_compacted_history_with_test_session(
) -> (Vec<ResponseItem>, Vec<ResponseItem>) {
let (session, turn_context) = crate::session::tests::make_session_and_context().await;
let turn_context = Arc::new(turn_context);
let step_context = StepContext::for_test(turn_context.clone());
session
.set_previous_turn_settings(previous_turn_settings.cloned())
.await;
let world_state = Arc::new(build_world_state_from_turn_context(&session, &turn_context).await);
let initial_context = session
.build_initial_context_with_world_state(&turn_context, world_state.as_ref())
.build_initial_context_with_world_state(world_state.as_ref(), &step_context)
.await;
let initial_context_injection = InitialContextInjection::BeforeLastUserMessage(world_state);
let (refreshed, _) = crate::compact_remote::process_compacted_history(
&session,
&turn_context,
&step_context,
compacted_history,
&initial_context_injection,
)
+5 -21
View File
@@ -8,7 +8,6 @@ use crate::hook_runtime::run_post_compact_hooks;
use crate::hook_runtime::run_pre_compact_hooks;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use codex_analytics::CompactionTrigger;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
@@ -25,8 +24,8 @@ use codex_protocol::protocol::TurnStartedEvent;
pub(crate) async fn run_manual_compact_task(
sess: Arc<Session>,
step_context: Arc<StepContext>,
turn_context: Arc<TurnContext>,
) -> CodexResult<()> {
let turn_context = step_context.turn.clone();
let start_event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_context.sub_id.clone(),
trace_id: turn_context.trace_id.clone(),
@@ -38,14 +37,7 @@ pub(crate) async fn run_manual_compact_task(
// Manual compaction runs outside run_turn, so it captures its own current step.
let world_state = Arc::new(sess.build_world_state_for_step(&step_context).await);
run_compact_task_inner(
&sess,
&turn_context,
&step_context,
world_state,
CompactionTrigger::Manual,
)
.await
run_compact_task_inner(&sess, &step_context, world_state, CompactionTrigger::Manual).await
}
/// Runs token-budget inline auto-compaction as a normal compaction lifecycle.
@@ -58,30 +50,22 @@ pub(crate) async fn run_inline_auto_compact_task(
step_context: Arc<StepContext>,
initial_context_injection: InitialContextInjection,
) -> CodexResult<()> {
let turn_context = &step_context.turn;
let world_state = match initial_context_injection {
InitialContextInjection::BeforeLastUserMessage(world_state) => world_state,
InitialContextInjection::DoNotInject => {
Arc::new(sess.build_world_state_for_step(&step_context).await)
}
};
run_compact_task_inner(
&sess,
turn_context,
&step_context,
world_state,
CompactionTrigger::Auto,
)
.await
run_compact_task_inner(&sess, &step_context, world_state, CompactionTrigger::Auto).await
}
async fn run_compact_task_inner(
sess: &Arc<Session>,
turn_context: &Arc<TurnContext>,
step_context: &Arc<StepContext>,
world_state: Arc<WorldState>,
trigger: CompactionTrigger,
) -> CodexResult<()> {
let turn_context = &step_context.turn;
let pre_compact_outcome = run_pre_compact_hooks(sess, turn_context, trigger).await;
match pre_compact_outcome {
PreCompactHookOutcome::Continue => {}
@@ -91,7 +75,7 @@ async fn run_compact_task_inner(
let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new());
sess.emit_turn_item_started(turn_context, &compaction_item)
.await;
sess.start_new_context_window(turn_context.as_ref(), world_state, step_context.as_ref())
sess.start_new_context_window(world_state, step_context.as_ref())
.await;
sess.emit_turn_item_completed(turn_context, compaction_item)
.await;
+3 -4
View File
@@ -83,11 +83,10 @@ fn build_collaboration_mode_update_item(
fn build_multi_agent_mode_update_item(
previous: Option<&TurnContextItem>,
next: &TurnContext,
step_context: &StepContext,
) -> Option<String> {
let effective_multi_agent_mode =
crate::session::multi_agents::effective_multi_agent_mode(next, step_context);
crate::session::multi_agents::effective_multi_agent_mode(step_context);
let previous = previous?;
if previous.multi_agent_mode == effective_multi_agent_mode {
return None;
@@ -241,11 +240,11 @@ fn build_text_message(role: &str, text_sections: Vec<String>) -> Option<Response
pub(crate) fn build_settings_update_items(
previous: Option<&TurnContextItem>,
previous_turn_settings: Option<&PreviousTurnSettings>,
next: &TurnContext,
step_context: &StepContext,
exec_policy: &Policy,
personality_feature_enabled: bool,
) -> Vec<ResponseItem> {
let next = step_context.turn.as_ref();
// TODO(ccunningham): build_settings_update_items still does not cover every
// model-visible item emitted by build_initial_context. Persist the remaining
// inputs or add explicit replay events so fork/resume can diff everything
@@ -256,7 +255,7 @@ pub(crate) fn build_settings_update_items(
build_model_instructions_update_item(previous_turn_settings, next),
build_permissions_update_item(previous, next, exec_policy),
build_collaboration_mode_update_item(previous, next),
build_multi_agent_mode_update_item(previous, next, step_context),
build_multi_agent_mode_update_item(previous, step_context),
build_realtime_update_item(previous, previous_turn_settings, next),
build_personality_update_item(previous, next, personality_feature_enabled),
]
+4 -18
View File
@@ -275,7 +275,6 @@ pub(crate) async fn record_guardian_denial_for_test(
/// caller as distinct from explicit guardian denials.
async fn run_guardian_review(
session: Arc<Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
review_id: String,
request: GuardianApprovalRequest,
@@ -283,6 +282,7 @@ async fn run_guardian_review(
approval_request_source: GuardianApprovalRequestSource,
external_cancel: Option<CancellationToken>,
) -> ReviewDecision {
let turn = step_context.turn.clone();
let target_item_id = guardian_request_target_item_id(&request).map(str::to_string);
let assessment_turn_id = guardian_request_turn_id(&request, &turn.sub_id).to_string();
let action_summary = guardian_assessment_action(&request);
@@ -360,7 +360,6 @@ async fn run_guardian_review(
let terminal_action = action_summary.clone();
let (outcome, analytics_result) = Box::pin(run_guardian_review_session_with_retry(
session.clone(),
turn.clone(),
step_context.clone(),
request,
retry_reason.clone(),
@@ -596,7 +595,6 @@ async fn run_guardian_review(
/// Public entrypoint for approval requests that should be reviewed by guardian.
pub(crate) async fn review_approval_request(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
step_context: Arc<StepContext>,
review_id: String,
request: GuardianApprovalRequest,
@@ -606,7 +604,6 @@ pub(crate) async fn review_approval_request(
// guardian session state machine into their own async stack.
Box::pin(run_guardian_review(
Arc::clone(session),
Arc::clone(turn),
step_context,
review_id,
request,
@@ -619,7 +616,6 @@ pub(crate) async fn review_approval_request(
pub(crate) async fn review_approval_request_with_cancel(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
step_context: Arc<StepContext>,
review_id: String,
request: GuardianApprovalRequest,
@@ -629,7 +625,6 @@ pub(crate) async fn review_approval_request_with_cancel(
) -> ReviewDecision {
run_guardian_review(
Arc::clone(session),
Arc::clone(turn),
step_context,
review_id,
request,
@@ -642,7 +637,6 @@ pub(crate) async fn review_approval_request_with_cancel(
pub(crate) fn spawn_approval_request_review(
session: Arc<Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
review_id: String,
request: GuardianApprovalRequest,
@@ -661,7 +655,6 @@ pub(crate) fn spawn_approval_request_review(
};
let decision = runtime.block_on(review_approval_request_with_cancel(
&session,
&turn,
step_context,
review_id,
request,
@@ -686,9 +679,9 @@ pub(super) struct GuardianReviewSessionConfig {
pub(super) async fn guardian_review_session_config(
session: &Session,
turn: &TurnContext,
step_context: &StepContext,
) -> anyhow::Result<GuardianReviewSessionConfig> {
let turn = step_context.turn.as_ref();
let network_proxy = session.services.network_proxy.load_full();
let live_network_config = match network_proxy.as_ref() {
Some(network_proxy) => Some(network_proxy.proxy().current_cfg().await?),
@@ -778,7 +771,6 @@ pub(super) async fn guardian_review_session_config(
/// rules.
async fn run_guardian_review_session_before_deadline(
session: Arc<Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
request: GuardianApprovalRequest,
retry_reason: Option<String>,
@@ -786,12 +778,8 @@ async fn run_guardian_review_session_before_deadline(
external_cancel: Option<CancellationToken>,
deadline: Instant,
) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) {
let session_config = match guardian_review_session_config(
session.as_ref(),
turn.as_ref(),
step_context.as_ref(),
)
.await
let turn = step_context.turn.clone();
let session_config = match guardian_review_session_config(session.as_ref(), &step_context).await
{
Ok(session_config) => session_config,
Err(err) => {
@@ -879,7 +867,6 @@ async fn run_guardian_review_session_before_deadline(
pub(super) async fn run_guardian_review_session_with_retry(
session: Arc<Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
request: GuardianApprovalRequest,
retry_reason: Option<String>,
@@ -893,7 +880,6 @@ pub(super) async fn run_guardian_review_session_with_retry(
loop {
let (outcome, mut analytics_result) = run_guardian_review_session_before_deadline(
Arc::clone(&session),
Arc::clone(&turn),
Arc::clone(&step_context),
request.clone(),
retry_reason.clone(),
+6 -14
View File
@@ -297,15 +297,13 @@ impl GuardianReviewSessionManager {
pub(crate) fn initialize(
&self,
parent_session: Arc<Session>,
parent_turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
) -> BoxFuture<'_, anyhow::Result<()>> {
// Boxing breaks the Session::new -> Guardian -> Session::new future recursion.
Box::pin(async move {
let spawn_config =
guardian_review_session_config(&parent_session, &parent_turn, &step_context)
.await?
.spawn_config;
let spawn_config = guardian_review_session_config(&parent_session, &step_context)
.await?
.spawn_config;
let reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
&spawn_config,
parent_session.user_instructions().await,
@@ -314,7 +312,6 @@ impl GuardianReviewSessionManager {
let spawn_cancel_guard = spawn_cancel_token.clone().drop_guard();
let review_session = spawn_guardian_review_session(
&parent_session,
&parent_turn,
Arc::clone(&step_context),
spawn_config,
reuse_key,
@@ -400,7 +397,6 @@ impl GuardianReviewSessionManager {
&spawn_cancel_token,
Box::pin(spawn_guardian_review_session(
&params.parent_session,
&params.parent_turn,
Arc::clone(&params.step_context),
params.spawn_config.clone(),
next_reuse_key.clone(),
@@ -612,7 +608,6 @@ impl GuardianReviewSessionManager {
&spawn_cancel_token,
Box::pin(spawn_guardian_review_session(
&params.parent_session,
&params.parent_turn,
Arc::clone(&params.step_context),
fork_config,
reuse_key,
@@ -655,7 +650,6 @@ impl GuardianReviewSessionManager {
async fn spawn_guardian_review_session(
parent_session: &Arc<Session>,
parent_turn: &Arc<TurnContext>,
step_context: Arc<StepContext>,
spawn_config: Config,
reuse_key: GuardianReviewSessionReuseKey,
@@ -675,7 +669,6 @@ async fn spawn_guardian_review_session(
parent_session.services.auth_manager.clone(),
parent_session.services.models_manager.clone(),
Arc::clone(parent_session),
Arc::clone(parent_turn),
step_context,
cancel_token.clone(),
SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_string()),
@@ -1212,8 +1205,10 @@ mod tests {
async fn test_review_params() -> GuardianReviewSessionParams {
let (session, turn) = crate::session::tests::make_session_and_context().await;
let turn = Arc::new(turn);
let step_context = StepContext::for_test(Arc::clone(&turn));
let model = turn.model_info.slug.clone();
let reasoning_effort = turn.reasoning_effort.clone();
let reasoning_effort = step_context.reasoning_effort.clone();
let reasoning_summary = turn.reasoning_summary;
let personality = turn.personality;
#[allow(deprecated)]
@@ -1226,9 +1221,6 @@ mod tests {
)
.expect("guardian config");
let turn = Arc::new(turn);
let step_context = StepContext::for_test(Arc::clone(&turn));
GuardianReviewSessionParams {
parent_session: Arc::new(session),
parent_turn: Arc::clone(&turn),
+18 -17
View File
@@ -7,6 +7,7 @@ use crate::config::NetworkProxySpec;
use crate::config::test_config;
use crate::guardian::approval_request::guardian_request_target_item_id;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use crate::test_support;
use codex_analytics::GuardianApprovalRequestSource;
@@ -1151,7 +1152,7 @@ async fn cancelled_guardian_review_emits_terminal_abort_without_warning() {
let decision = review_approval_request_with_cancel(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-cancelled-guardian".to_string(),
GuardianApprovalRequest::ApplyPatch {
id: "patch-1".to_string(),
@@ -1471,7 +1472,7 @@ async fn guardian_request_model_for_auto_review(
let (outcome, analytics_result) = run_guardian_review_session_for_test(
Arc::clone(&session),
turn,
StepContext::for_test(turn),
GuardianApprovalRequest::Shell {
id: "shell-1".to_string(),
command: vec!["git".to_string(), "push".to_string()],
@@ -1722,7 +1723,7 @@ async fn guardian_review_request_layout_matches_model_visible_request_snapshot()
let outcome = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
request,
Some("Sandbox denied outbound git push to github.com.".to_string()),
guardian_output_schema(),
@@ -1924,7 +1925,7 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
};
let first_outcome = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
first_request,
Some("First retry reason".to_string()),
guardian_output_schema(),
@@ -1971,7 +1972,7 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
};
let second_outcome = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
second_request,
Some("Second retry reason".to_string()),
guardian_output_schema(),
@@ -2014,7 +2015,7 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
};
let third_outcome = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
third_request,
Some("Third retry reason".to_string()),
guardian_output_schema(),
@@ -2199,7 +2200,7 @@ async fn guardian_reused_trunk_ignores_stale_prior_turn_completion() -> anyhow::
let (session, turn) = guardian_test_session_and_turn(&server).await;
let first_outcome = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
GuardianApprovalRequest::Shell {
id: "shell-1".to_string(),
command: vec!["git".to_string(), "push".to_string()],
@@ -2242,7 +2243,7 @@ async fn guardian_reused_trunk_ignores_stale_prior_turn_completion() -> anyhow::
let second_outcome = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
GuardianApprovalRequest::Shell {
id: "shell-2".to_string(),
command: vec!["git".to_string(), "push".to_string()],
@@ -2321,7 +2322,7 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() ->
let decision = review_approval_request(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-shell-guardian-error".to_string(),
GuardianApprovalRequest::Shell {
id: "shell-guardian-error".to_string(),
@@ -2420,7 +2421,7 @@ async fn guardian_review_retries_transient_session_failure_then_approves() -> an
let (outcome, metadata) = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
guardian_shell_request("shell-session-retry"),
/*retry_reason*/ None,
guardian_output_schema(),
@@ -2461,7 +2462,7 @@ async fn guardian_review_does_not_retry_missing_assessment_payload() -> anyhow::
let decision = review_approval_request(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-missing-assessment".to_string(),
guardian_shell_request("shell-missing-assessment"),
/*retry_reason*/ None,
@@ -2511,7 +2512,7 @@ async fn guardian_review_retries_two_parse_failures_then_approves() -> anyhow::R
let (outcome, metadata) = run_guardian_review_session_for_test(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(turn.clone()),
guardian_shell_request("shell-parse-retry"),
/*retry_reason*/ None,
guardian_output_schema(),
@@ -2565,7 +2566,7 @@ async fn guardian_review_exhausts_three_failures_with_one_terminal_event() -> an
let decision = review_approval_request(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-exhausted-retry".to_string(),
guardian_shell_request("shell-exhausted-retry"),
/*retry_reason*/ None,
@@ -2616,7 +2617,7 @@ async fn guardian_review_does_not_retry_valid_denial() -> anyhow::Result<()> {
let decision = review_approval_request(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-valid-denial".to_string(),
guardian_shell_request("shell-valid-denial"),
/*retry_reason*/ None,
@@ -2719,7 +2720,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
assert_eq!(
review_approval_request(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-shell-guardian-1".to_string(),
initial_request,
/*retry_reason*/ None
@@ -2773,7 +2774,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
let mut second_review = tokio::spawn(async move {
review_approval_request(
&session_for_second,
&turn_for_second,
StepContext::for_test(turn_for_second.clone()),
"review-shell-guardian-2".to_string(),
second_request,
Some("trunk follow-up".to_string()),
@@ -2820,7 +2821,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
let third_decision = review_approval_request(
&session,
&turn,
StepContext::for_test(turn.clone()),
"review-shell-guardian-3".to_string(),
third_request,
Some("parallel follow-up".to_string()),
-1
View File
@@ -1274,7 +1274,6 @@ async fn maybe_request_mcp_tool_approval(
let review_id = new_guardian_review_id();
let decision = review_approval_request(
sess,
turn_context,
Arc::clone(step_context),
review_id.clone(),
build_guardian_mcp_tool_review_request(call_id, invocation, metadata),
+29 -19
View File
@@ -93,10 +93,10 @@ fn approval_metadata(
}
}
fn mcp_turn_metadata_context(turn_context: &TurnContext) -> McpTurnMetadataContext<'_> {
fn mcp_turn_metadata_context(step_context: &StepContext) -> McpTurnMetadataContext<'_> {
McpTurnMetadataContext {
model: turn_context.model_info.slug.as_str(),
reasoning_effort: turn_context.effective_reasoning_effort(),
model: step_context.turn.model_info.slug.as_str(),
reasoning_effort: step_context.effective_reasoning_effort(),
}
}
@@ -1092,13 +1092,15 @@ fn truncate_mcp_tool_result_for_event_bounds_large_error() {
#[tokio::test]
async fn mcp_tool_call_request_meta_includes_turn_metadata_for_custom_server() {
let (_, turn_context) = make_session_and_context().await;
let expected_turn_metadata = turn_context
let step_context = StepContext::for_test(Arc::new(turn_context));
let expected_turn_metadata = step_context
.turn
.turn_metadata_state
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&turn_context))
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&step_context))
.expect("turn metadata");
let meta = build_mcp_tool_call_request_meta(
&turn_context,
&step_context,
"custom_server",
"call-custom",
/*metadata*/ None,
@@ -1112,13 +1114,13 @@ async fn mcp_tool_call_request_meta_includes_turn_metadata_for_custom_server() {
turn_metadata
.get("model")
.and_then(serde_json::Value::as_str),
Some(turn_context.model_info.slug.as_str())
Some(step_context.turn.model_info.slug.as_str())
);
assert_eq!(
turn_metadata
.get("reasoning_effort")
.and_then(serde_json::Value::as_str),
turn_context
step_context
.effective_reasoning_effort()
.map(|effort| effort.to_string())
.as_deref()
@@ -1135,12 +1137,14 @@ async fn mcp_tool_call_request_meta_includes_turn_metadata_for_custom_server() {
#[tokio::test]
async fn mcp_tool_call_request_meta_includes_turn_started_at_unix_ms() {
let (_, turn_context) = make_session_and_context().await;
turn_context
let step_context = StepContext::for_test(Arc::new(turn_context));
step_context
.turn
.turn_metadata_state
.set_turn_started_at_unix_ms(/*turn_started_at_unix_ms*/ 1_700_000_000_123);
let meta = build_mcp_tool_call_request_meta(
&turn_context,
&step_context,
"custom_server",
"call-custom",
/*metadata*/ None,
@@ -1196,9 +1200,11 @@ async fn mcp_sandbox_cwd_is_none_for_unselected_server_environment() -> anyhow::
#[tokio::test]
async fn plugin_mcp_tool_call_request_meta_includes_plugin_id() {
let (_, turn_context) = make_session_and_context().await;
let expected_turn_metadata = turn_context
let step_context = StepContext::for_test(Arc::new(turn_context));
let expected_turn_metadata = step_context
.turn
.turn_metadata_state
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&turn_context))
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&step_context))
.expect("turn metadata");
let mut metadata = approval_metadata(
/*connector_id*/ None, /*connector_name*/ None,
@@ -1208,7 +1214,7 @@ async fn plugin_mcp_tool_call_request_meta_includes_plugin_id() {
metadata.plugin_id = Some("sample@test".to_string());
assert_eq!(
build_mcp_tool_call_request_meta(&turn_context, "sample", "call-plugin", Some(&metadata),),
build_mcp_tool_call_request_meta(&step_context, "sample", "call-plugin", Some(&metadata),),
Some(serde_json::json!({
crate::X_CODEX_TURN_METADATA_HEADER: expected_turn_metadata,
MCP_TOOL_PLUGIN_ID_META_KEY: "sample@test",
@@ -1315,9 +1321,11 @@ async fn mcp_tool_call_item_includes_app_identity() {
#[tokio::test]
async fn codex_apps_tool_call_request_meta_includes_turn_metadata_and_codex_apps_meta() {
let (_, turn_context) = make_session_and_context().await;
let expected_turn_metadata = turn_context
let step_context = StepContext::for_test(Arc::new(turn_context));
let expected_turn_metadata = step_context
.turn
.turn_metadata_state
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&turn_context))
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&step_context))
.expect("turn metadata");
let metadata = McpToolApprovalMetadata {
annotations: None,
@@ -1346,7 +1354,7 @@ async fn codex_apps_tool_call_request_meta_includes_turn_metadata_and_codex_apps
assert_eq!(
build_mcp_tool_call_request_meta(
&turn_context,
&step_context,
CODEX_APPS_MCP_SERVER_NAME,
"call_abc123xyz789",
Some(&metadata),
@@ -1366,14 +1374,16 @@ async fn codex_apps_tool_call_request_meta_includes_turn_metadata_and_codex_apps
#[tokio::test]
async fn codex_apps_tool_call_request_meta_includes_call_id_without_existing_codex_apps_meta() {
let (_, turn_context) = make_session_and_context().await;
let expected_turn_metadata = turn_context
let step_context = StepContext::for_test(Arc::new(turn_context));
let expected_turn_metadata = step_context
.turn
.turn_metadata_state
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&turn_context))
.current_meta_value_for_mcp_request(mcp_turn_metadata_context(&step_context))
.expect("turn metadata");
assert_eq!(
build_mcp_tool_call_request_meta(
&turn_context,
&step_context,
CODEX_APPS_MCP_SERVER_NAME,
"call_abc123xyz789",
/*metadata*/ None,
@@ -11,6 +11,7 @@ use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use super::tests::make_session_and_context_with_rx;
use crate::session::step_context::StepContext;
use crate::state::ActiveTurn;
async fn wait_until_held(pause_state: &mut watch::Receiver<bool>) {
@@ -100,12 +101,14 @@ async fn patch_approval_holds_an_elicitation_until_response() {
#[tokio::test]
async fn permission_request_holds_an_elicitation_until_response() {
let (session, turn_context, events) = make_session_and_context_with_rx().await;
let step_context = StepContext::for_test(turn_context.clone());
*session.active_turn.lock().await = Some(ActiveTurn::default());
let mut pause_state = session.subscribe_elicitation_pause_state();
let request = tokio::spawn({
let session = session.clone();
let turn_context = turn_context.clone();
let step_context = step_context.clone();
async move {
let environment = turn_context
.environments
@@ -114,7 +117,7 @@ async fn permission_request_holds_an_elicitation_until_response() {
.selection();
session
.request_permissions_for_environment(
&turn_context,
&step_context,
"call-1".to_string(),
RequestPermissionsArgs {
environment_id: None,
-1
View File
@@ -624,7 +624,6 @@ async fn review_guardian_mcp_elicitation(
let review_id = crate::guardian::new_guardian_review_id();
let decision = crate::guardian::review_approval_request(
&session,
&turn_context,
step_context,
review_id.clone(),
guardian_request,
+11 -31
View File
@@ -1728,7 +1728,6 @@ impl Session {
async fn build_settings_update_items(
&self,
reference_context_item: Option<&TurnContextItem>,
current_context: &TurnContext,
step_context: &StepContext,
) -> Vec<ResponseItem> {
// TODO: Make context updates a pure diff of persisted previous/current TurnContextItem
@@ -1743,7 +1742,6 @@ impl Session {
crate::context_manager::updates::build_settings_update_items(
reference_context_item,
previous_turn_settings.as_ref(),
current_context,
step_context,
exec_policy.as_ref(),
self.features.enabled(Feature::Personality),
@@ -2289,13 +2287,13 @@ impl Session {
)]
pub(crate) async fn request_permissions_for_environment(
self: &Arc<Self>,
turn_context: &Arc<TurnContext>,
step_context: &Arc<StepContext>,
call_id: String,
args: RequestPermissionsArgs,
environment: TurnEnvironmentSelection,
cancellation_token: CancellationToken,
) -> Option<RequestPermissionsResponse> {
let turn_context = &step_context.turn;
match turn_context.as_ref().approval_policy.value() {
AskForApproval::Never => {
return Some(RequestPermissionsResponse {
@@ -2339,7 +2337,6 @@ impl Session {
};
let review_id = crate::guardian::new_guardian_review_id();
let session = Arc::clone(self);
let turn = Arc::clone(turn_context);
let request = crate::guardian::GuardianApprovalRequest::RequestPermissions {
id: call_id,
turn_id: turn_context.sub_id.clone(),
@@ -2348,7 +2345,6 @@ impl Session {
};
let review_rx = crate::guardian::spawn_approval_request_review(
session,
turn,
step_context.clone(),
review_id,
request,
@@ -2459,13 +2455,13 @@ impl Session {
pub(crate) async fn request_permissions_for_cwd(
self: &Arc<Self>,
turn_context: &Arc<TurnContext>,
step_context: &Arc<StepContext>,
call_id: String,
args: RequestPermissionsArgs,
cwd: AbsolutePathBuf,
cancellation_token: CancellationToken,
) -> Option<RequestPermissionsResponse> {
let turn_context = &step_context.turn;
let turn_environment = match args.environment_id.as_deref() {
Some(environment_id) => turn_context
.environments
@@ -2484,7 +2480,6 @@ impl Session {
let mut environment = turn_environment.selection();
environment.cwd = PathUri::from_abs_path(&cwd);
self.request_permissions_for_environment(
turn_context,
step_context,
call_id,
args,
@@ -3189,20 +3184,19 @@ impl Session {
pub(crate) async fn build_initial_context_with_world_state(
&self,
turn_context: &TurnContext,
world_state: &WorldState,
step_context: &StepContext,
) -> Vec<ResponseItem> {
self.build_initial_context_with_world_state_and_mcp(turn_context, world_state, step_context)
self.build_initial_context_with_world_state_and_mcp(world_state, step_context)
.await
}
async fn build_initial_context_with_world_state_and_mcp(
&self,
turn_context: &TurnContext,
world_state: &WorldState,
step_context: &StepContext,
) -> Vec<ResponseItem> {
let turn_context = step_context.turn.as_ref();
let mcp = step_context.mcp.as_ref();
let mut developer_sections = Vec::<String>::with_capacity(8);
let mut contextual_user_sections = Vec::<String>::with_capacity(2);
@@ -3474,9 +3468,7 @@ impl Session {
{
items.push(usage_hint_message);
}
if let Some(multi_agent_mode) =
multi_agents::effective_multi_agent_mode(turn_context, step_context)
{
if let Some(multi_agent_mode) = multi_agents::effective_multi_agent_mode(step_context) {
items.push(ContextualUserFragment::into(
MultiAgentModeInstructions::new(multi_agent_mode),
));
@@ -3543,23 +3535,19 @@ impl Session {
pub(crate) async fn start_new_context_window(
&self,
turn_context: &TurnContext,
world_state: Arc<WorldState>,
step_context: &StepContext,
) -> u64 {
let turn_context = step_context.turn.as_ref();
let window = {
let mut state = self.state.lock().await;
state.start_new_context_window()
};
let (window_number, window_ids) = window;
let context_items = self
.build_initial_context_with_world_state(
turn_context,
world_state.as_ref(),
step_context,
)
.build_initial_context_with_world_state(world_state.as_ref(), step_context)
.await;
let turn_context_item = turn_context.to_turn_context_item(step_context);
let turn_context_item = step_context.to_turn_context_item();
self.replace_compacted_history(
turn_context,
context_items,
@@ -3607,18 +3595,14 @@ impl Session {
let state = self.state.lock().await;
state.reference_context_item()
};
let turn_context_item = turn_context.to_turn_context_item(step_context);
let turn_context_item = step_context.to_turn_context_item();
let turn_context_changed = reference_context_item.as_ref() != Some(&turn_context_item);
let should_inject_full_context = reference_context_item.is_none();
let world_state = Arc::new(self.build_world_state_for_step(step_context).await);
// Full initial context resets the baseline; later turns persist only its changes.
let (mut context_items, world_state_item) = if should_inject_full_context {
let context_items = self
.build_initial_context_with_world_state_and_mcp(
turn_context,
world_state.as_ref(),
step_context,
)
.build_initial_context_with_world_state_and_mcp(world_state.as_ref(), step_context)
.await;
let snapshot = world_state.snapshot();
self.state
@@ -3634,11 +3618,7 @@ impl Session {
// Steady-state path: append only built-in context diffs here; turn-scoped extension
// context is added below.
let mut context_items = self
.build_settings_update_items(
reference_context_item.as_ref(),
turn_context,
step_context,
)
.build_settings_update_items(reference_context_item.as_ref(), step_context)
.await;
let (world_state_items, world_state_item) = {
let mut state = self.state.lock().await;
+2 -4
View File
@@ -37,10 +37,8 @@ fn configured_usage_hint_text_for_source<'a>(
}
}
pub(crate) fn effective_multi_agent_mode(
turn_context: &TurnContext,
step_context: &StepContext,
) -> Option<MultiAgentMode> {
pub(crate) fn effective_multi_agent_mode(step_context: &StepContext) -> Option<MultiAgentMode> {
let turn_context = step_context.turn.as_ref();
if turn_context.multi_agent_version != MultiAgentVersion::V2 {
return None;
}
@@ -17,8 +17,13 @@ use codex_protocol::protocol::WorldStateItem;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::path::PathBuf;
use std::sync::Arc;
use uuid::Uuid;
fn test_step_context(turn_context: Arc<TurnContext>) -> Arc<StepContext> {
StepContext::for_test(turn_context)
}
fn user_message(text: &str) -> ResponseItem {
ResponseItem::Message {
id: None,
@@ -134,11 +139,11 @@ async fn record_initial_history_reconstructs_typed_inter_agent_message() {
#[tokio::test]
async fn record_initial_history_restores_world_state_baseline() {
let (session, turn_context) = make_session_and_context().await;
let turn_context = Arc::new(turn_context);
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let world_state = build_world_state_from_turn_context(&session, &turn_context).await;
let rollout_items = completed_user_turn_rollout(
turn_context.to_turn_context_item(),
test_step_context(turn_context.clone()).to_turn_context_item(),
vec![RolloutItem::WorldState(WorldStateItem::full(
world_state.snapshot().into_value(),
))],
@@ -162,7 +167,8 @@ async fn record_initial_history_restores_world_state_baseline() {
#[tokio::test]
async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previous_turn_settings()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_model = "previous-rollout-model";
let previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -183,7 +189,9 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previ
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let rollout_items = vec![RolloutItem::TurnContext(previous_context_item)];
@@ -208,7 +216,8 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previ
#[tokio::test]
async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lifecycle_turn_with_missing_turn_context_id()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_model = "previous-rollout-model";
let mut previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -229,7 +238,9 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let turn_id = previous_context_item
@@ -290,8 +301,9 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif
#[tokio::test]
async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_completed_turns() {
let (session, turn_context) = make_session_and_context().await;
let first_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let first_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
@@ -413,8 +425,9 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com
#[tokio::test]
async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_incomplete_turn() {
let (session, turn_context) = make_session_and_context().await;
let first_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let first_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
@@ -507,8 +520,9 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc
#[tokio::test]
async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metadata() {
let (session, turn_context) = make_session_and_context().await;
let first_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let first_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
@@ -633,8 +647,9 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
#[tokio::test]
async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() {
let (session, turn_context) = make_session_and_context().await;
let first_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let first_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
@@ -734,8 +749,9 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() {
#[tokio::test]
async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding_user_turns() {
let (session, turn_context) = make_session_and_context().await;
let only_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let only_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let only_turn_id = only_context_item
.turn_id
.clone()
@@ -788,8 +804,9 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding
#[tokio::test]
async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
let (session, turn_context) = make_session_and_context().await;
let previous_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let user_turn_id = previous_context_item
.turn_id
.clone()
@@ -863,8 +880,9 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
#[tokio::test]
async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_compaction_metadata() {
let (session, turn_context) = make_session_and_context().await;
let previous_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let previous_turn_id = previous_context_item
.turn_id
.clone()
@@ -959,8 +977,9 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp
#[tokio::test]
async fn record_initial_history_resumed_bare_turn_context_does_not_seed_reference_context_item() {
let (session, turn_context) = make_session_and_context().await;
let previous_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let rollout_items = vec![RolloutItem::TurnContext(previous_context_item.clone())];
session
@@ -976,8 +995,9 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_seed_referenc
#[tokio::test]
async fn record_initial_history_resumed_does_not_seed_reference_context_item_after_compaction() {
let (session, turn_context) = make_session_and_context().await;
let previous_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let rollout_items = vec![
RolloutItem::TurnContext(previous_context_item),
RolloutItem::Compacted(CompactedItem {
@@ -1004,7 +1024,8 @@ async fn record_initial_history_resumed_does_not_seed_reference_context_item_aft
#[tokio::test]
async fn reconstruct_history_restores_initial_window_from_session_meta() {
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let thread_id = ThreadId::default();
let initial_window_id = Uuid::now_v7();
let rollout_items = vec![RolloutItem::SessionMeta(SessionMetaLine {
@@ -1031,7 +1052,8 @@ async fn reconstruct_history_restores_initial_window_from_session_meta() {
#[tokio::test]
async fn reconstruct_history_prefers_compacted_window_over_session_meta() {
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let thread_id = ThreadId::default();
let initial_window_id = Uuid::now_v7();
let compacted_first_window_id = Uuid::now_v7();
@@ -1077,9 +1099,10 @@ async fn reconstruct_history_prefers_compacted_window_over_session_meta() {
#[tokio::test]
async fn reconstruct_history_replays_world_state_from_latest_compaction_window() {
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let rollout_items = completed_user_turn_rollout(
turn_context.to_turn_context_item(),
test_step_context(turn_context.clone()).to_turn_context_item(),
vec![
RolloutItem::WorldState(WorldStateItem::full(json!({
"environment": {"status": "old"}
@@ -1116,7 +1139,8 @@ async fn reconstruct_history_replays_world_state_from_latest_compaction_window()
#[tokio::test]
async fn reconstruct_history_preserves_legacy_compaction_count_with_session_meta_window() {
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let thread_id = ThreadId::default();
let initial_window_id = Uuid::now_v7();
let rollout_items = vec![
@@ -1154,7 +1178,8 @@ async fn reconstruct_history_preserves_legacy_compaction_count_with_session_meta
#[tokio::test]
async fn reconstruct_history_legacy_compaction_without_replacement_history_does_not_inject_current_initial_context()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let rollout_items = vec![
RolloutItem::ResponseItem(user_message("before compact")),
RolloutItem::ResponseItem(assistant_message("assistant reply")),
@@ -1185,8 +1210,9 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_
#[tokio::test]
async fn reconstruct_history_legacy_compaction_without_replacement_history_clears_later_reference_context_item()
{
let (session, turn_context) = make_session_and_context().await;
let current_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let current_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let current_turn_id = current_context_item
.turn_id
.clone()
@@ -1242,7 +1268,8 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear
#[tokio::test]
async fn record_initial_history_resumed_turn_context_after_compaction_reestablishes_reference_context_item()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_model = "previous-rollout-model";
let previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -1263,7 +1290,9 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let previous_turn_id = previous_context_item
@@ -1349,7 +1378,9 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}))
.expect("serialize expected reference context item")
@@ -1359,7 +1390,8 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
#[tokio::test]
async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_turn_for_compaction_accounting()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_model = "previous-rollout-model";
let previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -1380,7 +1412,9 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let previous_turn_id = previous_context_item
@@ -1478,8 +1512,9 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
#[tokio::test]
async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_for_later_turn_context()
{
let (session, turn_context) = make_session_and_context().await;
let previous_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let previous_turn_id = previous_context_item
.turn_id
.clone()
@@ -1506,7 +1541,9 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -1606,7 +1643,8 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
#[tokio::test]
async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clears_reference_context_item()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_model = "previous-rollout-model";
let previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -1627,7 +1665,9 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let previous_turn_id = previous_context_item
@@ -1716,8 +1756,9 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
#[tokio::test]
async fn record_initial_history_resumed_trailing_incomplete_turn_preserves_turn_context_item() {
let (session, turn_context) = make_session_and_context().await;
let current_context_item = turn_context.to_turn_context_item();
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let current_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
let current_turn_id = current_context_item
.turn_id
.clone()
@@ -1773,7 +1814,8 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_preserves_turn_
#[tokio::test]
async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clears_reference_context_item()
{
let (session, turn_context) = make_session_and_context().await;
let (session, turn_context_raw) = make_session_and_context().await;
let turn_context = Arc::new(turn_context_raw);
let previous_model = "previous-rollout-model";
let previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -1794,7 +1836,9 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let previous_turn_id = previous_context_item
+92 -50
View File
@@ -204,6 +204,10 @@ impl StepContext {
}
}
fn test_step_context(turn_context: Arc<TurnContext>) -> Arc<StepContext> {
StepContext::for_test(turn_context)
}
mod guardian_tests;
struct InstructionsTestCase {
@@ -1055,7 +1059,7 @@ async fn danger_full_access_tool_attempts_do_not_enforce_managed_network() -> an
id: ctx.call_id.to_string(),
command: Vec::new(),
#[allow(deprecated)]
cwd: ctx.turn.cwd.clone(),
cwd: ctx.step_context.turn.cwd.clone(),
sandbox_permissions: crate::sandboxing::SandboxPermissions::UseDefault,
additional_permissions: None,
justification: None,
@@ -1132,7 +1136,7 @@ async fn danger_full_access_tool_attempts_do_not_enforce_managed_network() -> an
let mut tool = ProbeToolRuntime::default();
let tool_ctx = crate::tools::sandboxing::ToolCtx {
session: Arc::clone(&session),
turn: Arc::clone(&turn),
step_context: StepContext::for_test(Arc::clone(&turn)),
call_id: "probe-call".to_string(),
tool_name: codex_tools::ToolName::plain("probe"),
};
@@ -2789,9 +2793,10 @@ async fn start_new_context_window_assigns_and_persists_item_ids() {
attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await;
let world_state =
Arc::new(build_world_state_from_turn_context(session.as_ref(), &turn_context).await);
let step_context = StepContext::for_test(turn_context.clone());
session
.start_new_context_window(turn_context.as_ref(), world_state)
.start_new_context_window(world_state, &step_context)
.await;
let live_history = session.clone_history().await;
@@ -3073,6 +3078,7 @@ async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result<
#[tokio::test]
async fn record_initial_history_forked_hydrates_previous_turn_settings() {
let (session, turn_context) = make_session_and_context().await;
let turn_context = Arc::new(turn_context);
let previous_model = "forked-rollout-model";
let previous_context_item = TurnContextItem {
turn_id: Some(turn_context.sub_id.clone()),
@@ -3093,7 +3099,9 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
effort: turn_context.reasoning_effort.clone(),
effort: test_step_context(turn_context.clone())
.reasoning_effort
.clone(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
let turn_id = previous_context_item
@@ -3175,8 +3183,11 @@ async fn thread_rollback_drops_last_turn_from_history() {
full_history.extend(initial_context.clone());
full_history.extend(turn_1.clone());
full_history.extend(turn_2);
sess.replace_history(full_history.clone(), Some(tc.to_turn_context_item()))
.await;
sess.replace_history(
full_history.clone(),
Some(test_step_context(tc.clone()).to_turn_context_item()),
)
.await;
let rollout_items: Vec<RolloutItem> = full_history
.into_iter()
.map(RolloutItem::ResponseItem)
@@ -3190,7 +3201,8 @@ async fn thread_rollback_drops_last_turn_from_history() {
.await;
{
let mut state = sess.state.lock().await;
state.set_reference_context_item(Some(tc.to_turn_context_item()));
state
.set_reference_context_item(Some(test_step_context(tc.clone()).to_turn_context_item()));
}
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
@@ -3235,8 +3247,11 @@ async fn thread_rollback_clears_history_when_num_turns_exceeds_existing_turns()
let mut full_history = Vec::new();
full_history.extend(initial_context.clone());
full_history.extend(turn_1);
sess.replace_history(full_history.clone(), Some(tc.to_turn_context_item()))
.await;
sess.replace_history(
full_history.clone(),
Some(test_step_context(tc.clone()).to_turn_context_item()),
)
.await;
let rollout_items: Vec<RolloutItem> = full_history
.into_iter()
.map(RolloutItem::ResponseItem)
@@ -3282,7 +3297,7 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context
)
.await;
let first_context_item = tc.to_turn_context_item();
let first_context_item = test_step_context(tc.clone()).to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
@@ -3404,7 +3419,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio
)
.await;
let first_context_item = tc.to_turn_context_item();
let first_context_item = test_step_context(tc.clone()).to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
@@ -3546,7 +3561,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() {
Arc::get_mut(&mut sess).expect("session should not have additional references"),
)
.await;
let turn_context_item = tc.to_turn_context_item();
let turn_context_item = test_step_context(tc.clone()).to_turn_context_item();
sess.persist_rollout_items(&[
RolloutItem::EventMsg(EventMsg::TurnStarted(
@@ -3969,10 +3984,13 @@ async fn includes_timed_out_message() {
#[tokio::test]
async fn turn_context_with_model_updates_model_fields() {
let (session, mut turn_context) = make_session_and_context().await;
turn_context.reasoning_effort = Some(ReasoningEffortConfig::Minimal);
Arc::make_mut(&mut turn_context.config).model_reasoning_effort =
Some(codex_protocol::openai_models::ReasoningEffort::Minimal);
let updated = turn_context
.with_model("gpt-5.4".to_string(), &session.services.models_manager)
.await;
let updated = Arc::new(updated);
let updated_step = test_step_context(updated.clone());
let expected_model_info = session
.services
.models_manager
@@ -3983,19 +4001,19 @@ async fn turn_context_with_model_updates_model_fields() {
.await;
assert_eq!(updated.config.model.as_deref(), Some("gpt-5.4"));
assert_eq!(updated.collaboration_mode.model(), "gpt-5.4");
assert_eq!(updated.collaboration_mode().model(), "gpt-5.4");
assert_eq!(updated.model_info, expected_model_info);
assert_eq!(
updated.reasoning_effort,
Some(ReasoningEffortConfig::Medium)
updated_step.reasoning_effort,
Some(codex_protocol::openai_models::ReasoningEffort::Medium)
);
assert_eq!(
updated.collaboration_mode.reasoning_effort(),
Some(ReasoningEffortConfig::Medium)
updated.collaboration_mode().reasoning_effort(),
Some(codex_protocol::openai_models::ReasoningEffort::Medium)
);
assert_eq!(
updated.config.model_reasoning_effort,
Some(ReasoningEffortConfig::Medium)
Some(codex_protocol::openai_models::ReasoningEffort::Medium)
);
}
@@ -5322,8 +5340,9 @@ async fn build_initial_context(
turn_context: &Arc<TurnContext>,
) -> Vec<ResponseItem> {
let world_state = build_world_state_from_turn_context(session, turn_context).await;
let step_context = StepContext::for_test(turn_context.clone());
session
.build_initial_context_with_world_state(turn_context.as_ref(), &world_state)
.build_initial_context_with_world_state(&world_state, &step_context)
.await
}
@@ -6096,6 +6115,7 @@ async fn request_permissions_emits_event_when_granular_policy_allows_requests()
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let step_context = StepContext::for_test(turn_context.as_ref().clone());
let call_id = "call-1".to_string();
let expected_response = codex_protocol::request_permissions::RequestPermissionsResponse {
permissions: RequestPermissionProfile {
@@ -6111,6 +6131,7 @@ async fn request_permissions_emits_event_when_granular_policy_allows_requests()
let handle = tokio::spawn({
let session = Arc::clone(&session);
let turn_context = Arc::clone(&turn_context);
let step_context = step_context.clone();
let call_id = call_id.clone();
async move {
let environment = turn_context
@@ -6120,7 +6141,7 @@ async fn request_permissions_emits_event_when_granular_policy_allows_requests()
.selection();
session
.request_permissions_for_environment(
&turn_context,
&step_context,
call_id,
codex_protocol::request_permissions::RequestPermissionsArgs {
environment_id: None,
@@ -6332,6 +6353,7 @@ async fn request_permissions_response_materializes_session_cwd_grants_before_rec
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let step_context = StepContext::for_test(turn_context.as_ref().clone());
let call_id = "call-1".to_string();
let requested_permissions = RequestPermissionProfile {
file_system: Some(FileSystemPermissions {
@@ -6349,6 +6371,7 @@ async fn request_permissions_response_materializes_session_cwd_grants_before_rec
let handle = tokio::spawn({
let session = Arc::clone(&session);
let turn_context = Arc::clone(&turn_context);
let step_context = step_context.clone();
let call_id = call_id.clone();
let requested_permissions = requested_permissions.clone();
async move {
@@ -6359,7 +6382,7 @@ async fn request_permissions_response_materializes_session_cwd_grants_before_rec
.selection();
session
.request_permissions_for_environment(
&turn_context,
&step_context,
call_id,
codex_protocol::request_permissions::RequestPermissionsArgs {
environment_id: None,
@@ -6442,6 +6465,7 @@ async fn request_permissions_is_auto_denied_when_granular_policy_blocks_tool_req
let session = Arc::new(session);
let turn_context = Arc::new(turn_context);
let step_context = StepContext::for_test(turn_context.as_ref().clone());
let call_id = "call-1".to_string();
let environment = turn_context
.environments
@@ -6450,7 +6474,7 @@ async fn request_permissions_is_auto_denied_when_granular_policy_blocks_tool_req
.selection();
let response = session
.request_permissions_for_environment(
&turn_context,
&step_context,
call_id,
codex_protocol::request_permissions::RequestPermissionsArgs {
environment_id: None,
@@ -8149,12 +8173,11 @@ async fn build_settings_update_items_emits_realtime_start_when_session_becomes_l
)
.await;
current_context.realtime_active = true;
let previous_step = StepContext::for_test(previous_context.clone());
let current_step = test_step_context(Arc::new(current_context));
let update_items = session
.build_settings_update_items(
Some(&previous_context.to_turn_context_item()),
&current_context,
)
.build_settings_update_items(Some(&previous_step.to_turn_context_item()), &current_step)
.await;
let developer_texts = developer_input_texts(&update_items);
@@ -8177,12 +8200,11 @@ async fn build_settings_update_items_emits_realtime_end_when_session_stops_being
)
.await;
current_context.realtime_active = false;
let previous_step = test_step_context(Arc::new(previous_context));
let current_step = test_step_context(Arc::new(current_context));
let update_items = session
.build_settings_update_items(
Some(&previous_context.to_turn_context_item()),
&current_context,
)
.build_settings_update_items(Some(&previous_step.to_turn_context_item()), &current_step)
.await;
let developer_texts = developer_input_texts(&update_items);
@@ -8197,7 +8219,9 @@ async fn build_settings_update_items_emits_realtime_end_when_session_stops_being
#[tokio::test]
async fn build_settings_update_items_uses_previous_turn_settings_for_realtime_end() {
let (session, previous_context) = make_session_and_context().await;
let mut previous_context_item = previous_context.to_turn_context_item();
let previous_context = Arc::new(previous_context);
let mut previous_context_item =
test_step_context(previous_context.clone()).to_turn_context_item();
previous_context_item.realtime_active = None;
let previous_turn_settings = PreviousTurnSettings {
model: previous_context.model_info.slug.clone(),
@@ -8211,12 +8235,13 @@ async fn build_settings_update_items_uses_previous_turn_settings_for_realtime_en
)
.await;
current_context.realtime_active = false;
let current_step = test_step_context(Arc::new(current_context));
session
.set_previous_turn_settings(Some(previous_turn_settings))
.await;
let update_items = session
.build_settings_update_items(Some(&previous_context_item), &current_context)
.build_settings_update_items(Some(&previous_context_item), &current_step)
.await;
let developer_texts = developer_input_texts(&update_items);
@@ -8243,7 +8268,7 @@ async fn build_initial_context_uses_previous_realtime_state() {
"expected initial context to describe active realtime state, got {developer_texts:?}"
);
let previous_context_item = turn_context.to_turn_context_item();
let previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
{
let mut state = session.state.lock().await;
state.set_reference_context_item(Some(previous_context_item));
@@ -8400,9 +8425,9 @@ async fn record_context_updates_includes_turn_context_fragments_on_steady_state_
.insert(TurnContextExtensionTestState {
expected_model_context_window: Some(50),
});
let mut previous_context_item = turn_context.to_turn_context_item();
previous_context_item.turn_id = Some("previous-turn-id".to_string());
let turn_context = Arc::new(turn_context);
let mut previous_context_item = test_step_context(turn_context.clone()).to_turn_context_item();
previous_context_item.turn_id = Some("previous-turn-id".to_string());
let world_state = build_world_state_from_turn_context(&session, &turn_context).await;
{
let mut state = session.state.lock().await;
@@ -8976,14 +9001,20 @@ async fn turn_context_item_stores_local_cwd() {
#[allow(deprecated)]
let local_cwd = turn_context.cwd.clone();
assert_eq!(turn_context.to_turn_context_item().cwd, local_cwd);
assert_eq!(
test_step_context(Arc::new(turn_context))
.to_turn_context_item()
.cwd,
local_cwd
);
}
#[tokio::test]
async fn turn_context_item_omits_legacy_equivalent_file_system_sandbox_policy() {
let (_session, turn_context) = make_session_and_context().await;
let turn_context = Arc::new(turn_context);
let item = turn_context.to_turn_context_item();
let item = test_step_context(turn_context.clone()).to_turn_context_item();
assert_eq!(item.file_system_sandbox_policy, None);
assert_eq!(
@@ -9001,8 +9032,9 @@ async fn turn_context_item_stores_split_file_system_sandbox_policy_when_differen
&file_system_sandbox_policy,
turn_context.network_sandbox_policy(),
);
let turn_context = Arc::new(turn_context);
let item = turn_context.to_turn_context_item();
let item = test_step_context(turn_context.clone()).to_turn_context_item();
assert_eq!(
item.file_system_sandbox_policy,
@@ -9030,8 +9062,10 @@ async fn record_context_updates_and_set_reference_context_item_injects_full_cont
let current_context = session.reference_context_item().await;
assert_eq!(
serde_json::to_value(current_context).expect("serialize current context item"),
serde_json::to_value(Some(turn_context.to_turn_context_item()))
.expect("serialize expected context item")
serde_json::to_value(Some(
test_step_context(turn_context.clone()).to_turn_context_item(),
))
.expect("serialize expected context item")
);
}
@@ -9082,8 +9116,8 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co
async fn record_context_updates_and_set_reference_context_item_persists_baseline_without_emitting_diffs()
{
let (mut session, turn_context) = make_session_and_context().await;
let previous_context_item = turn_context.to_turn_context_item();
let previous_context = Arc::new(turn_context);
let previous_context_item = test_step_context(previous_context.clone()).to_turn_context_item();
let world_state = build_world_state_from_turn_context(&session, &previous_context).await;
let mut turn_context = Arc::try_unwrap(previous_context)
.unwrap_or_else(|_| panic!("previous turn context should have no remaining references"));
@@ -9096,13 +9130,15 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
.set_world_state_baseline(world_state.snapshot());
}
let rollout_path = attach_thread_persistence(&mut session).await;
let turn_context = Arc::new(turn_context);
let current_step = test_step_context(turn_context.clone());
let update_items = session
.build_settings_update_items(Some(&previous_context_item), &turn_context)
.build_settings_update_items(Some(&previous_context_item), &current_step)
.await;
assert_eq!(update_items, Vec::new());
let turn_context = Arc::new(turn_context);
drop(current_step);
let step_context = StepContext::for_test(Arc::clone(&turn_context));
session
.record_context_updates_and_set_reference_context_item(&step_context)
@@ -9115,8 +9151,10 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
assert_eq!(
serde_json::to_value(session.reference_context_item().await)
.expect("serialize current context item"),
serde_json::to_value(Some(turn_context.to_turn_context_item()))
.expect("serialize expected context item")
serde_json::to_value(Some(
test_step_context(turn_context.clone()).to_turn_context_item(),
))
.expect("serialize expected context item")
);
session.ensure_rollout_materialized().await;
session.flush_rollout().await.expect("rollout should flush");
@@ -9134,8 +9172,10 @@ async fn record_context_updates_and_set_reference_context_item_persists_baseline
assert_eq!(
serde_json::to_value(persisted_turn_context)
.expect("serialize persisted turn context item"),
serde_json::to_value(Some(turn_context.to_turn_context_item()))
.expect("serialize expected turn context item")
serde_json::to_value(Some(
test_step_context(turn_context.clone()).to_turn_context_item(),
))
.expect("serialize expected turn context item")
);
}
@@ -9260,8 +9300,10 @@ async fn record_context_updates_and_set_reference_context_item_persists_full_rei
assert_eq!(
serde_json::to_value(persisted_turn_context)
.expect("serialize persisted turn context item"),
serde_json::to_value(Some(turn_context.to_turn_context_item()))
.expect("serialize expected turn context item")
serde_json::to_value(Some(
test_step_context(turn_context.clone()).to_turn_context_item(),
))
.expect("serialize expected turn context item")
);
}
@@ -117,6 +117,7 @@ async fn request_permissions_routes_to_guardian_when_reviewer_is_enabled() {
);
let session = Arc::new(session);
let turn_context = Arc::new(turn_context_raw);
let step_context = StepContext::for_test(turn_context.clone());
let requested_permissions = RequestPermissionProfile {
network: Some(NetworkPermissions {
@@ -132,7 +133,7 @@ async fn request_permissions_routes_to_guardian_when_reviewer_is_enabled() {
let response = tokio::time::timeout(
Duration::from_secs(45),
session.request_permissions_for_environment(
&turn_context,
&step_context,
"perm-call-1".to_string(),
RequestPermissionsArgs {
environment_id: None,
@@ -214,9 +215,11 @@ async fn request_permissions_guardian_review_stops_when_cancelled() {
..RequestPermissionProfile::default()
};
let cancellation_token = CancellationToken::new();
let step_context = StepContext::for_test(turn_context.clone());
let request_handle = tokio::spawn({
let session = Arc::clone(&session);
let turn_context = Arc::clone(&turn_context);
let step_context = step_context.clone();
let requested_permissions = requested_permissions.clone();
let cancellation_token = cancellation_token.clone();
async move {
@@ -227,7 +230,7 @@ async fn request_permissions_guardian_review_stops_when_cancelled() {
.selection();
session
.request_permissions_for_environment(
&turn_context,
&step_context,
"perm-call-cancelled".to_string(),
RequestPermissionsArgs {
environment_id: None,
@@ -531,12 +534,13 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message
turn_context.session_source = guardian_source;
turn_context.developer_instructions = Some(guardian_policy.clone());
let turn_context = Arc::new(turn_context);
let step_context = StepContext::for_test(turn_context.clone());
let world_state = Arc::new(build_world_state_from_turn_context(&session, &turn_context).await);
let initial_context_injection = InitialContextInjection::BeforeLastUserMessage(world_state);
let (refreshed, _) = crate::compact_remote::process_compacted_history(
&session,
&turn_context,
&step_context,
vec![
ResponseItem::Message {
id: None,
+5 -8
View File
@@ -206,8 +206,7 @@ pub(crate) async fn run_turn(
.await;
}
track_turn_resolved_config_analytics(&sess, &turn_context, &input, first_step_context.clone())
.await;
track_turn_resolved_config_analytics(&sess, &input, first_step_context.clone()).await;
let mut last_agent_message: Option<String> = None;
let mut stop_hook_active = false;
@@ -750,10 +749,10 @@ async fn build_extension_turn_input_items(
)]
async fn track_turn_resolved_config_analytics(
sess: &Session,
turn_context: &TurnContext,
input: &[TurnInput],
first_step_context: Arc<StepContext>,
) {
let turn_context = first_step_context.turn.as_ref();
let thread_config = {
let state = sess.state.lock().await;
state.session_configuration.thread_config_snapshot()
@@ -1030,7 +1029,6 @@ async fn run_auto_compact(
);
run_inline_auto_compact_task(
Arc::clone(sess),
Arc::clone(turn_context),
Arc::clone(&step_context),
initial_context_injection,
reason,
@@ -1169,7 +1167,6 @@ async fn run_sampling_request(
let err = match try_run_sampling_request(
tool_runtime.clone(),
Arc::clone(&sess),
Arc::clone(&turn_context),
step_context.clone(),
Arc::clone(&turn_store),
client_session,
@@ -1933,14 +1930,13 @@ async fn drain_in_flight(
#[instrument(level = "trace",
skip_all,
fields(
turn_id = %turn_context.sub_id,
model = %turn_context.model_info.slug
turn_id = %step_context.turn.sub_id,
model = %step_context.turn.model_info.slug
)
)]
async fn try_run_sampling_request(
tool_runtime: ToolCallRuntime,
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
step_context: Arc<StepContext>,
turn_store: Arc<codex_extension_api::ExtensionData>,
client_session: &mut ModelClientSession,
@@ -1949,6 +1945,7 @@ async fn try_run_sampling_request(
prompt: &Prompt,
cancellation_token: CancellationToken,
) -> CodexResult<SamplingRequestResult> {
let turn_context = step_context.turn.clone();
feedback_tags!(
model = turn_context.model_info.slug.clone(),
approval_policy = turn_context.approval_policy.value(),
+27 -22
View File
@@ -341,41 +341,46 @@ impl TurnContext {
(file_system_sandbox_policy != legacy_file_system_sandbox_policy)
.then_some(file_system_sandbox_policy)
}
}
pub(crate) fn to_turn_context_item(&self, step_context: &StepContext) -> TurnContextItem {
let workspace_roots = self.config.effective_workspace_roots();
impl StepContext {
pub(crate) fn to_turn_context_item(&self) -> TurnContextItem {
let turn = self.turn.as_ref();
let workspace_roots = turn.config.effective_workspace_roots();
#[allow(deprecated)]
let cwd = self.cwd.clone();
let cwd = turn.cwd.clone();
TurnContextItem {
turn_id: Some(self.sub_id.clone()),
turn_id: Some(turn.sub_id.clone()),
cwd,
workspace_roots: (!workspace_roots.is_empty()).then_some(workspace_roots),
current_date: self.current_date.clone(),
timezone: self.timezone.clone(),
approval_policy: self.approval_policy.value(),
sandbox_policy: self.sandbox_policy(),
permission_profile: Some(self.permission_profile()),
network: self.turn_context_network_item(),
file_system_sandbox_policy: self.non_legacy_file_system_sandbox_policy(),
model: self.model_info.slug.clone(),
comp_hash: self.model_info.comp_hash.clone(),
personality: self.personality,
current_date: turn.current_date.clone(),
timezone: turn.timezone.clone(),
approval_policy: turn.approval_policy.value(),
sandbox_policy: turn.sandbox_policy(),
permission_profile: Some(turn.permission_profile()),
network: turn.turn_context_network_item(),
file_system_sandbox_policy: turn.non_legacy_file_system_sandbox_policy(),
model: turn.model_info.slug.clone(),
comp_hash: turn.model_info.comp_hash.clone(),
personality: turn.personality,
collaboration_mode: Some(CollaborationMode {
mode: self.mode,
mode: turn.mode,
settings: Settings {
model: self.model_info.slug.clone(),
reasoning_effort: step_context.reasoning_effort.clone(),
developer_instructions: self.collaboration_mode_developer_instructions.clone(),
model: turn.model_info.slug.clone(),
reasoning_effort: self.reasoning_effort.clone(),
developer_instructions: turn.collaboration_mode_developer_instructions.clone(),
},
}),
multi_agent_version: Some(self.multi_agent_version),
multi_agent_mode: super::multi_agents::effective_multi_agent_mode(self, step_context),
realtime_active: Some(self.realtime_active),
effort: step_context.effective_reasoning_effort(),
multi_agent_version: Some(turn.multi_agent_version),
multi_agent_mode: super::multi_agents::effective_multi_agent_mode(self),
realtime_active: Some(turn.realtime_active),
effort: self.effective_reasoning_effort(),
summary: ReasoningSummaryConfig::Auto,
}
}
}
impl TurnContext {
fn turn_context_network_item(&self) -> Option<TurnContextNetworkItem> {
let network = self
.config
+1 -6
View File
@@ -256,16 +256,11 @@ async fn schedule_startup_prewarm_inner(
.await;
if routes_approval_to_guardian(&startup_turn_context) {
let guardian_session = Arc::clone(&session);
let guardian_parent_turn = Arc::clone(&startup_turn_context);
let guardian_step_context = Arc::clone(&step_context);
drop(tokio::spawn(async move {
if let Err(err) = guardian_session
.guardian_review_session
.initialize(
Arc::clone(&guardian_session),
guardian_parent_turn,
guardian_step_context,
)
.initialize(Arc::clone(&guardian_session), guardian_step_context)
.await
{
warn!("failed to initialize guardian review session: {err:#}");
+2 -3
View File
@@ -37,8 +37,7 @@ impl SessionTask for CompactTask {
.set_active_step_context(Arc::clone(&step_context))
.await;
if ctx.config.features.enabled(Feature::TokenBudget) {
crate::compact_token_budget::run_manual_compact_task(session, step_context, ctx)
.await?;
crate::compact_token_budget::run_manual_compact_task(session, step_context).await?;
return Ok(None);
}
@@ -78,7 +77,7 @@ impl SessionTask for CompactTask {
// Compaction prompt is synthesized; no UI element ranges to preserve.
text_elements: Vec::new(),
}];
crate::compact::run_compact_task(session.clone(), ctx, step_context, input).await
crate::compact::run_compact_task(session.clone(), step_context, input).await
};
if let Err(err @ CodexErr::TurnAborted) = result {
return Err(err);
+1 -3
View File
@@ -75,7 +75,6 @@ impl SessionTask for ReviewTask {
// Start sub-codex conversation and get the receiver for events.
let output = match start_review_conversation(
session.clone(),
ctx.clone(),
user_input,
step_context,
cancellation_token.clone(),
@@ -98,11 +97,11 @@ impl SessionTask for ReviewTask {
async fn start_review_conversation(
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
input: Vec<UserInput>,
step_context: Arc<StepContext>,
cancellation_token: CancellationToken,
) -> Option<async_channel::Receiver<Event>> {
let ctx = &step_context.turn;
let config = ctx.config.clone();
let mut sub_agent_config = config.as_ref().clone();
// Carry over review-only feature restrictions so the delegate cannot
@@ -132,7 +131,6 @@ async fn start_review_conversation(
session.models_manager(),
input,
session.clone_session(),
ctx.clone(),
step_context,
cancellation_token,
SubAgentSource::Review,
+3 -1
View File
@@ -4,6 +4,7 @@ use crate::init_state_db;
use crate::installation_id::INSTALLATION_ID_FILENAME;
use crate::rollout::RolloutRecorder;
use crate::session::session::SessionSettingsUpdate;
use crate::session::step_context::StepContext;
use crate::session::tests::build_world_state_from_turn_context;
use crate::session::tests::make_session_and_context;
use crate::tasks::InterruptedTurnHistoryMarker;
@@ -320,8 +321,9 @@ async fn ignores_session_prefix_messages_when_truncating() {
let (session, turn_context) = make_session_and_context().await;
let turn_context = Arc::new(turn_context);
let world_state = build_world_state_from_turn_context(&session, &turn_context).await;
let step_context = StepContext::for_test(turn_context.clone());
let mut items = session
.build_initial_context_with_world_state(&turn_context, &world_state)
.build_initial_context_with_world_state(&world_state, &step_context)
.await;
items.push(user_msg("feature request"));
items.push(assistant_msg("ack"));
@@ -1,4 +1,5 @@
use super::*;
use crate::session::step_context::StepContext;
use crate::session::tests::build_world_state_from_turn_context;
use crate::session::tests::make_session_and_context;
use codex_protocol::AgentPath;
@@ -271,8 +272,9 @@ async fn ignores_session_prefix_messages_when_truncating_rollout_from_start() {
let (session, turn_context) = make_session_and_context().await;
let turn_context = Arc::new(turn_context);
let world_state = build_world_state_from_turn_context(&session, &turn_context).await;
let step_context = StepContext::for_test(turn_context.clone());
let mut items = session
.build_initial_context_with_world_state(&turn_context, &world_state)
.build_initial_context_with_world_state(&world_state, &step_context)
.await;
items.push(user_msg("feature request"));
items.push(assistant_msg("ack"));
@@ -108,10 +108,10 @@ fn required_state_db(
async fn build_runner_options(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
step_context: &StepContext,
requested_concurrency: Option<usize>,
) -> Result<JobRunnerOptions, FunctionCallError> {
let turn = &step_context.turn;
let multi_agent_version = turn.multi_agent_version;
if multi_agent_version == MultiAgentVersion::Disabled {
return Err(FunctionCallError::RespondToModel(
@@ -126,7 +126,7 @@ async fn build_runner_options(
}
let max_concurrency = normalize_concurrency(requested_concurrency, agent_max_threads);
let base_instructions = session.get_base_instructions().await;
let spawn_config = build_agent_spawn_config(&base_instructions, turn.as_ref(), step_context)?;
let spawn_config = build_agent_spawn_config(&base_instructions, step_context)?;
Ok(JobRunnerOptions {
max_concurrency,
spawn_config,
@@ -35,7 +35,6 @@ impl SpawnAgentsOnCsvHandler {
) -> Result<Box<dyn crate::tools::context::ToolOutput>, FunctionCallError> {
let ToolInvocation {
session,
turn,
step_context,
payload,
..
@@ -50,7 +49,7 @@ impl SpawnAgentsOnCsvHandler {
}
};
handle(session, turn, step_context, arguments)
handle(session, step_context, arguments)
.await
.map(boxed_tool_output)
}
@@ -69,10 +68,10 @@ impl CoreToolRuntime for SpawnAgentsOnCsvHandler {
/// `report_agent_job_result`, then exported to CSV on completion.
pub async fn handle(
session: Arc<Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
arguments: String,
) -> Result<FunctionToolOutput, FunctionCallError> {
let turn = step_context.turn.clone();
let args: SpawnAgentsOnCsvArgs = parse_arguments(arguments.as_str())?;
if args.instruction.trim().is_empty() {
return Err(FunctionCallError::RespondToModel(
@@ -184,23 +183,17 @@ pub async fn handle(
})?;
let requested_concurrency = args.max_concurrency.or(args.max_workers);
let options = match build_runner_options(
&session,
&turn,
step_context.as_ref(),
requested_concurrency,
)
.await
{
Ok(options) => options,
Err(err) => {
let error_message = err.to_string();
let _ = db
.mark_agent_job_failed(job_id.as_str(), error_message.as_str())
.await;
return Err(err);
}
};
let options =
match build_runner_options(&session, step_context.as_ref(), requested_concurrency).await {
Ok(options) => options,
Err(err) => {
let error_message = err.to_string();
let _ = db
.mark_agent_job_failed(job_id.as_str(), error_message.as_str())
.await;
return Err(err);
}
};
db.mark_agent_job_running(job_id.as_str())
.await
.map_err(|err| {
@@ -11,6 +11,7 @@ use crate::apply_patch::InternalApplyPatchInvocation;
use crate::apply_patch::convert_apply_patch_to_protocol;
use crate::function_tool::FunctionCallError;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use crate::session::turn_context::TurnEnvironment;
use crate::tools::context::ApplyPatchToolOutput;
@@ -444,7 +445,7 @@ impl ApplyPatchHandler {
let mut runtime = ApplyPatchRuntime::new();
let tool_ctx = ToolCtx {
session: session.clone(),
turn: turn.clone(),
step_context: step_context.clone(),
call_id: call_id.clone(),
tool_name: tool_name.clone(),
};
@@ -549,11 +550,12 @@ pub(crate) async fn intercept_apply_patch(
fs: &dyn ExecutorFileSystem,
turn_environment: TurnEnvironment,
session: Arc<Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
tracker: Option<&SharedTurnDiffTracker>,
call_id: &str,
tool_name: &str,
) -> Result<Option<FunctionToolOutput>, FunctionCallError> {
let turn = step_context.turn.clone();
let sandbox = turn.file_system_sandbox_context(/*additional_permissions*/ None, cwd);
match codex_apply_patch::maybe_parse_apply_patch_verified(command, cwd, fs, Some(&sandbox))
.await
@@ -607,7 +609,7 @@ pub(crate) async fn intercept_apply_patch(
let mut runtime = ApplyPatchRuntime::new();
let tool_ctx = ToolCtx {
session: session.clone(),
turn: turn.clone(),
step_context,
call_id: call_id.to_string(),
tool_name: ToolName::plain(tool_name),
};
@@ -2,7 +2,6 @@ use super::*;
use crate::agent::next_thread_spawn_depth;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use crate::tools::handlers::multi_agents_spec::create_resume_agent_tool;
use codex_tools::ToolSpec;
use std::sync::Arc;
@@ -89,7 +88,6 @@ async fn handle_resume_agent(
let (receiver_agent, error) = if matches!(status, AgentStatus::NotFound) {
match Box::pin(try_resume_closed_agent(
&session,
&turn,
step_context.as_ref(),
receiver_thread_id,
child_depth,
@@ -190,12 +188,12 @@ impl ToolOutput for ResumeAgentResult {
async fn try_resume_closed_agent(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
step_context: &StepContext,
receiver_thread_id: ThreadId,
child_depth: i32,
) -> Result<(), FunctionCallError> {
let config = build_agent_resume_config(turn.as_ref(), step_context)?;
let turn = &step_context.turn;
let config = build_agent_resume_config(step_context)?;
Box::pin(session.services.agent_control.resume_agent_from_rollout(
config,
receiver_thread_id,
@@ -49,7 +49,7 @@ impl Handler {
.agent_control
.get_agent_metadata(receiver_thread_id);
if receiver_agent.is_some() {
let resume_config = build_agent_resume_config(turn.as_ref(), step_context.as_ref())?;
let resume_config = build_agent_resume_config(step_context.as_ref())?;
session
.services
.agent_control
@@ -89,7 +89,6 @@ async fn handle_spawn_agent(
.await;
let mut config = build_agent_spawn_config(
&session.get_base_instructions().await,
turn.as_ref(),
step_context.as_ref(),
)?;
if let Some(service_tier) = args.service_tier.as_ref() {
@@ -161,33 +161,29 @@ pub(crate) fn parse_collab_input(
/// the wrong provider or runtime policy.
pub(crate) fn build_agent_spawn_config(
base_instructions: &BaseInstructions,
turn: &TurnContext,
step_context: &StepContext,
) -> Result<Config, FunctionCallError> {
let mut config = build_agent_shared_config(turn, step_context)?;
let mut config = build_agent_shared_config(step_context)?;
config.base_instructions = Some(base_instructions.text.clone());
Ok(config)
}
pub(crate) fn build_agent_resume_config(
turn: &TurnContext,
step_context: &StepContext,
) -> Result<Config, FunctionCallError> {
let mut config = build_agent_shared_config(turn, step_context)?;
let mut config = build_agent_shared_config(step_context)?;
// For resume, keep base instructions sourced from rollout/session metadata.
config.base_instructions = None;
Ok(config)
}
fn build_agent_shared_config(
turn: &TurnContext,
step_context: &StepContext,
) -> Result<Config, FunctionCallError> {
fn build_agent_shared_config(step_context: &StepContext) -> Result<Config, FunctionCallError> {
let turn = step_context.turn.as_ref();
let base_config = turn.config.clone();
let mut config = (*base_config).clone();
config.model = Some(turn.model_info.slug.clone());
config.model_provider = turn.provider.info().clone();
config.model_reasoning_effort = step_context.effective_reasoning_effort().clone();
config.model_reasoning_effort = step_context.effective_reasoning_effort();
config.model_reasoning_summary = Some(turn.reasoning_summary);
config.developer_instructions = turn.developer_instructions.clone();
apply_spawn_agent_runtime_overrides(&mut config, turn)?;
@@ -4513,12 +4513,14 @@ async fn build_agent_spawn_config_uses_turn_context_values() {
.set(AskForApproval::OnRequest)
.expect("approval policy set");
let config = build_agent_spawn_config(&base_instructions, &turn).expect("spawn config");
let turn = Arc::new(turn);
let step_context = StepContext::for_test(turn.clone());
let config = build_agent_spawn_config(&base_instructions, &step_context).expect("spawn config");
let mut expected = (*turn.config).clone();
expected.base_instructions = Some(base_instructions.text);
expected.model = Some(turn.model_info.slug.clone());
expected.model_provider = turn.provider.info().clone();
expected.model_reasoning_effort = turn.reasoning_effort.clone();
expected.model_reasoning_effort = step_context.reasoning_effort.clone();
expected.model_reasoning_summary = Some(turn.reasoning_summary);
expected.developer_instructions = turn.developer_instructions.clone();
#[allow(deprecated)]
@@ -4547,13 +4549,15 @@ async fn build_agent_resume_config_clears_base_instructions() {
.set(AskForApproval::OnRequest)
.expect("approval policy set");
let config = build_agent_resume_config(&turn).expect("resume config");
let turn = Arc::new(turn);
let step_context = StepContext::for_test(turn.clone());
let config = build_agent_resume_config(&step_context).expect("resume config");
let mut expected = (*turn.config).clone();
expected.base_instructions = None;
expected.model = Some(turn.model_info.slug.clone());
expected.model_provider = turn.provider.info().clone();
expected.model_reasoning_effort = turn.reasoning_effort.clone();
expected.model_reasoning_effort = step_context.reasoning_effort.clone();
expected.model_reasoning_summary = Some(turn.reasoning_summary);
expected.developer_instructions = turn.developer_instructions.clone();
#[allow(deprecated)]
@@ -90,7 +90,7 @@ pub(crate) async fn handle_message_string_tool(
let receiver_agent_path = receiver_agent.agent_path.clone().ok_or_else(|| {
FunctionCallError::RespondToModel("target agent is missing an agent_path".to_string())
})?;
let resume_config = build_agent_resume_config(turn.as_ref(), step_context.as_ref())?;
let resume_config = build_agent_resume_config(step_context.as_ref())?;
session
.services
.agent_control
@@ -62,7 +62,6 @@ async fn handle_spawn_agent(
let child_depth = next_thread_spawn_depth(&session_source);
let mut config = build_agent_spawn_config(
&session.get_base_instructions().await,
turn.as_ref(),
step_context.as_ref(),
)?;
if let Some(service_tier) = args.service_tier.as_ref() {
@@ -46,7 +46,6 @@ impl RequestPermissionsHandler {
) -> Result<Box<dyn crate::tools::context::ToolOutput>, FunctionCallError> {
let ToolInvocation {
session,
turn,
step_context,
cancellation_token,
call_id,
@@ -94,7 +93,6 @@ impl RequestPermissionsHandler {
let response = session
.request_permissions_for_environment(
&turn,
&step_context,
call_id,
args,
+6 -5
View File
@@ -7,7 +7,7 @@ use tokio_util::sync::CancellationToken;
use crate::exec::ExecParams;
use crate::exec_policy::ExecApprovalRequest;
use crate::function_tool::FunctionCallError;
use crate::session::turn_context::TurnContext;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnEnvironment;
use crate::shell::ShellType;
use crate::tools::context::FunctionToolOutput;
@@ -53,7 +53,7 @@ struct RunExecLikeArgs {
additional_permissions: Option<AdditionalPermissionProfile>,
prefix_rule: Option<Vec<String>>,
session: Arc<crate::session::session::Session>,
turn: Arc<TurnContext>,
step_context: Arc<StepContext>,
turn_environment: TurnEnvironment,
tracker: crate::tools::context::SharedTurnDiffTracker,
call_id: String,
@@ -70,12 +70,13 @@ async fn run_exec_like(args: RunExecLikeArgs) -> Result<FunctionToolOutput, Func
additional_permissions,
prefix_rule,
session,
turn,
step_context,
turn_environment,
tracker,
call_id,
shell_runtime_backend,
} = args;
let turn = step_context.turn.clone();
let fs = turn_environment.environment.get_filesystem();
@@ -145,7 +146,7 @@ async fn run_exec_like(args: RunExecLikeArgs) -> Result<FunctionToolOutput, Func
fs.as_ref(),
turn_environment.clone(),
session.clone(),
turn.clone(),
step_context.clone(),
Some(&tracker),
&call_id,
tool_name.name.as_str(),
@@ -205,7 +206,7 @@ async fn run_exec_like(args: RunExecLikeArgs) -> Result<FunctionToolOutput, Func
let mut runtime = ShellRuntime::for_shell_command(shell_runtime_backend);
let tool_ctx = ToolCtx {
session: session.clone(),
turn: turn.clone(),
step_context,
call_id: call_id.clone(),
tool_name,
};
@@ -226,7 +226,7 @@ impl ShellCommandHandler {
additional_permissions: params.additional_permissions.clone(),
prefix_rule,
session,
turn,
step_context,
turn_environment,
tracker,
call_id,
@@ -127,7 +127,8 @@ impl ExecCommandHandler {
};
let manager: &UnifiedExecProcessManager = &session.services.unified_exec_manager;
let context = UnifiedExecContext::new(session.clone(), turn.clone(), call_id.clone());
let context =
UnifiedExecContext::new(session.clone(), step_context.clone(), call_id.clone());
let environment_args: ExecCommandEnvironmentArgs = parse_arguments(&arguments)?;
let Some(turn_environment) = resolve_tool_environment(
&step_context.environments,
@@ -194,7 +195,7 @@ impl ExecCommandHandler {
if let Some(native_cwd) = native_cwd.as_ref() {
maybe_emit_implicit_skill_invocation(
session.as_ref(),
context.turn.as_ref(),
context.step_context.turn.as_ref(),
&hook_command,
native_cwd,
)
@@ -275,11 +276,11 @@ impl ExecCommandHandler {
.requests_sandbox_override()
&& !effective_additional_permissions.permissions_preapproved
&& !matches!(
context.turn.approval_policy.value(),
context.step_context.turn.approval_policy.value(),
codex_protocol::protocol::AskForApproval::OnRequest
)
{
let approval_policy = context.turn.approval_policy.value();
let approval_policy = context.step_context.turn.approval_policy.value();
manager.release_process_id(process_id).await;
return Err(FunctionCallError::RespondToModel(format!(
"approval policy is {approval_policy:?}; reject command — you cannot ask for escalated permissions if the approval policy is {approval_policy:?}"
@@ -295,7 +296,7 @@ impl ExecCommandHandler {
|| {
normalize_and_validate_additional_permissions(
additional_permissions_allowed,
context.turn.approval_policy.value(),
context.step_context.turn.approval_policy.value(),
effective_additional_permissions.sandbox_permissions,
effective_additional_permissions.additional_permissions,
effective_additional_permissions.permissions_preapproved,
@@ -317,7 +318,7 @@ impl ExecCommandHandler {
fs.as_ref(),
turn_environment.clone(),
context.session.clone(),
context.turn.clone(),
context.step_context.clone(),
Some(&tracker),
&context.call_id,
"exec_command",
@@ -353,7 +354,7 @@ impl ExecCommandHandler {
sandbox_cwd: native_environment_cwd,
turn_environment: turn_environment.clone(),
shell_mode,
network: context.turn.network.clone(),
network: context.step_context.turn.network.clone(),
tty,
sandbox_permissions: effective_additional_permissions.sandbox_permissions,
additional_permissions: normalized_additional_permissions,
+14 -4
View File
@@ -8,6 +8,7 @@ use crate::guardian::routes_approval_to_guardian;
use crate::hook_runtime::run_permission_request_hooks;
use crate::network_policy_decision::denied_network_policy_message;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::tools::sandboxing::PermissionRequestPayload;
use crate::tools::sandboxing::ToolError;
use codex_hooks::PermissionRequestDecision;
@@ -239,6 +240,7 @@ impl PendingHostApproval {
struct ActiveNetworkApprovalCall {
registration_id: String,
turn_id: String,
step_context: Arc<StepContext>,
trigger: GuardianNetworkAccessTrigger,
command: String,
environment_id: String,
@@ -293,7 +295,7 @@ impl NetworkApprovalService {
async fn register_call(
&self,
registration_id: String,
turn_id: String,
step_context: Arc<StepContext>,
trigger: GuardianNetworkAccessTrigger,
command: String,
environment_id: String,
@@ -301,11 +303,13 @@ impl NetworkApprovalService {
) {
let mut calls = self.calls.lock().await;
let key = registration_id.clone();
let turn_id = step_context.turn.sub_id.clone();
calls.active_calls.insert(
key,
Arc::new(ActiveNetworkApprovalCall {
registration_id,
turn_id,
step_context,
trigger,
command,
environment_id,
@@ -632,9 +636,15 @@ impl NetworkApprovalService {
let use_guardian = routes_approval_to_guardian(&turn_context);
let guardian_review_id = use_guardian.then(new_guardian_review_id);
let approval_decision = if let Some(review_id) = guardian_review_id.clone() {
let Some(step_context) = owner_call
.as_ref()
.map(|owner_call| owner_call.step_context.clone())
else {
return NetworkDecision::deny(REASON_NOT_ALLOWED);
};
review_approval_request(
&session,
&turn_context,
step_context,
review_id,
GuardianApprovalRequest::NetworkAccess {
id: guardian_approval_id.clone(),
@@ -849,7 +859,7 @@ pub(crate) fn build_network_policy_decider(
pub(crate) async fn begin_network_approval(
session: &Session,
turn_id: &str,
step_context: Arc<StepContext>,
managed_network_active: bool,
selected_sandbox: SandboxType,
spec: Option<NetworkApprovalSpec>,
@@ -890,7 +900,7 @@ pub(crate) async fn begin_network_approval(
.network_approval
.register_call(
registration_id.clone(),
turn_id.to_string(),
step_context,
trigger,
command,
environment_id,
@@ -1,5 +1,7 @@
use super::*;
use crate::sandboxing::SandboxPermissions;
use crate::session::step_context::StepContext;
use crate::session::tests::make_session_and_context;
use codex_network_proxy::BlockedRequestArgs;
use codex_protocol::models::PermissionProfile;
use codex_protocol::permissions::NetworkSandboxPolicy;
@@ -296,10 +298,11 @@ async fn register_call_with_default_shell_trigger(
registration_id: &str,
) -> CancellationToken {
let cancellation_token = CancellationToken::new();
let (_session, turn) = make_session_and_context().await;
service
.register_call(
registration_id.to_string(),
"turn-1".to_string(),
StepContext::for_test(Arc::new(turn)),
GuardianNetworkAccessTrigger {
call_id: "call-1".to_string(),
tool_name: "shell_command".to_string(),
@@ -321,6 +324,7 @@ async fn register_call_with_default_shell_trigger(
#[tokio::test]
async fn active_call_preserves_triggering_command_context() {
let service = NetworkApprovalService::default();
let (_session, turn) = make_session_and_context().await;
let expected = GuardianNetworkAccessTrigger {
call_id: "call-1".to_string(),
tool_name: "shell_command".to_string(),
@@ -335,7 +339,7 @@ async fn active_call_preserves_triggering_command_context() {
service
.register_call(
"registration-1".to_string(),
"turn-1".to_string(),
StepContext::for_test(Arc::new(turn)),
expected.clone(),
"curl https://example.com".to_string(),
"remote".to_string(),
+7 -7
View File
@@ -71,7 +71,7 @@ impl ToolOrchestrator {
{
let network_approval = match begin_network_approval(
&tool_ctx.session,
&tool_ctx.turn.sub_id,
tool_ctx.step_context.clone(),
managed_network_active,
attempt.sandbox,
tool.network_approval_spec(req, tool_ctx),
@@ -84,7 +84,7 @@ impl ToolOrchestrator {
let attempt_tool_ctx = ToolCtx {
session: tool_ctx.session.clone(),
turn: tool_ctx.turn.clone(),
step_context: tool_ctx.step_context.clone(),
call_id: tool_ctx.call_id.clone(),
tool_name: tool_ctx.tool_name.clone(),
};
@@ -171,7 +171,7 @@ impl ToolOrchestrator {
let guardian_review_id = Some(new_guardian_review_id());
let approval_ctx = ApprovalCtx {
session: &tool_ctx.session,
turn: &tool_ctx.turn,
step_context: &tool_ctx.step_context,
call_id: &tool_ctx.call_id,
guardian_review_id: guardian_review_id.clone(),
retry_reason: None,
@@ -206,7 +206,7 @@ impl ToolOrchestrator {
let guardian_review_id = use_guardian.then(new_guardian_review_id);
let approval_ctx = ApprovalCtx {
session: &tool_ctx.session,
turn: &tool_ctx.turn,
step_context: &tool_ctx.step_context,
call_id: &tool_ctx.call_id,
guardian_review_id: guardian_review_id.clone(),
retry_reason: reason.clone(),
@@ -403,7 +403,7 @@ impl ToolOrchestrator {
let guardian_review_id = use_guardian.then(new_guardian_review_id);
let approval_ctx = ApprovalCtx {
session: &tool_ctx.session,
turn: &tool_ctx.turn,
step_context: &tool_ctx.step_context,
call_id: &tool_ctx.call_id,
guardian_review_id: guardian_review_id.clone(),
retry_reason: Some(retry_reason),
@@ -539,7 +539,7 @@ impl ToolOrchestrator {
let tool_name = flat_tool_name(&tool_ctx.tool_name);
match run_permission_request_hooks(
approval_ctx.session,
approval_ctx.turn,
&approval_ctx.step_context.turn,
permission_request_run_id,
permission_request,
)
@@ -579,7 +579,7 @@ impl ToolOrchestrator {
Ok(action) => {
review_approval_request(
approval_ctx.session,
approval_ctx.turn,
approval_ctx.step_context.clone(),
review_id,
action,
approval_ctx.retry_reason.clone(),
@@ -146,7 +146,7 @@ impl Approvable<ApplyPatchRequest> for ApplyPatchRuntime {
ctx: ApprovalCtx<'a>,
) -> BoxFuture<'a, ReviewDecision> {
let session = ctx.session;
let turn = ctx.turn;
let turn = &ctx.step_context.turn;
let call_id = ctx.call_id.to_string();
let retry_reason = ctx.retry_reason.clone();
let approval_keys = self.approval_keys(req);
+3 -3
View File
@@ -106,7 +106,7 @@ impl ShellRuntime {
fn stdout_stream(ctx: &ToolCtx) -> Option<crate::exec::StdoutStream> {
Some(crate::exec::StdoutStream {
sub_id: ctx.turn.sub_id.clone(),
sub_id: ctx.step_context.turn.sub_id.clone(),
call_id: ctx.call_id.clone(),
tx_event: ctx.session.get_tx_event(),
})
@@ -149,7 +149,7 @@ impl Approvable<ShellRequest> for ShellRuntime {
.clone()
.or_else(|| req.justification.clone());
let session = ctx.session;
let turn = ctx.turn;
let turn = &ctx.step_context.turn;
let call_id = ctx.call_id.to_string();
Box::pin(async move {
with_cached_approval(&session.services, "shell", keys, move || async move {
@@ -213,7 +213,7 @@ impl ToolRuntime<ShellRequest, ExecToolCallOutput> for ShellRuntime {
req: &ShellRequest,
ctx: &ToolCtx,
) -> Option<NetworkApprovalSpec> {
let file_system_sandbox_policy = ctx.turn.file_system_sandbox_policy();
let file_system_sandbox_policy = ctx.step_context.turn.file_system_sandbox_policy();
let sandbox_permissions = sandbox_permissions_preserving_denied_reads(
req.sandbox_permissions,
&file_system_sandbox_policy,
@@ -200,8 +200,8 @@ pub(super) async fn try_run_zsh_fork(
arg0,
sandbox_policy_cwd,
windows_sandbox_workspace_roots,
codex_linux_sandbox_exe: ctx.turn.config.codex_linux_sandbox_exe.clone(),
use_legacy_landlock: ctx.turn.config.features.use_legacy_landlock(),
codex_linux_sandbox_exe: ctx.step_context.turn.config.codex_linux_sandbox_exe.clone(),
use_legacy_landlock: ctx.step_context.turn.config.features.use_legacy_landlock(),
};
let main_execve_wrapper_exe = ctx
.session
@@ -235,11 +235,11 @@ pub(super) async fn try_run_zsh_fork(
let escalation_policy = CoreShellActionProvider {
policy: Arc::clone(&exec_policy),
session: Arc::clone(&ctx.session),
turn: Arc::clone(&ctx.turn),
step_context: Arc::clone(&ctx.step_context),
call_id: ctx.call_id.clone(),
environment_id: req.turn_environment.environment_id.clone(),
tool_name: GuardianCommandSource::Shell,
approval_policy: ctx.turn.approval_policy.value(),
approval_policy: ctx.step_context.turn.approval_policy.value(),
permission_profile: command_executor.permission_profile.clone(),
file_system_sandbox_policy: command_executor.file_system_sandbox_policy.clone(),
sandbox_permissions: req.sandbox_permissions,
@@ -313,17 +313,17 @@ pub(crate) async fn prepare_unified_exec_zsh_fork(
arg0: exec_request.arg0.clone(),
sandbox_policy_cwd,
windows_sandbox_workspace_roots: exec_request.windows_sandbox_workspace_roots.clone(),
codex_linux_sandbox_exe: ctx.turn.config.codex_linux_sandbox_exe.clone(),
use_legacy_landlock: ctx.turn.config.features.use_legacy_landlock(),
codex_linux_sandbox_exe: ctx.step_context.turn.config.codex_linux_sandbox_exe.clone(),
use_legacy_landlock: ctx.step_context.turn.config.features.use_legacy_landlock(),
};
let escalation_policy = CoreShellActionProvider {
policy: Arc::clone(&exec_policy),
session: Arc::clone(&ctx.session),
turn: Arc::clone(&ctx.turn),
step_context: Arc::clone(&ctx.step_context),
call_id: ctx.call_id.clone(),
environment_id: req.turn_environment.environment_id.clone(),
tool_name: GuardianCommandSource::UnifiedExec,
approval_policy: ctx.turn.approval_policy.value(),
approval_policy: ctx.step_context.turn.approval_policy.value(),
permission_profile: exec_request.permission_profile.clone(),
file_system_sandbox_policy: exec_request.file_system_sandbox_policy.clone(),
sandbox_permissions: req.sandbox_permissions,
@@ -354,7 +354,7 @@ pub(crate) async fn prepare_unified_exec_zsh_fork(
struct CoreShellActionProvider {
policy: Arc<RwLock<Policy>>,
session: Arc<crate::session::session::Session>,
turn: Arc<crate::session::turn_context::TurnContext>,
step_context: Arc<crate::session::step_context::StepContext>,
call_id: String,
environment_id: String,
tool_name: GuardianCommandSource,
@@ -450,7 +450,8 @@ impl CoreShellActionProvider {
let command = join_program_and_argv(program, argv);
let workdir = workdir.clone();
let session = self.session.clone();
let turn = self.turn.clone();
let step_context = self.step_context.clone();
let turn = step_context.turn.clone();
let call_id = self.call_id.clone();
let approval_id = Some(Uuid::new_v4().to_string());
let environment_id = Some(self.environment_id.clone());
@@ -493,7 +494,7 @@ impl CoreShellActionProvider {
if let Some(review_id) = guardian_review_id.clone() {
let decision = review_approval_request(
&session,
&turn,
step_context,
review_id.clone(),
GuardianApprovalRequest::Execve {
id: call_id.clone(),
@@ -651,7 +652,7 @@ impl CoreShellActionProvider {
InterceptedExecPolicyContext {
approval_policy: self.approval_policy,
permission_profile: self.permission_profile.clone(),
windows_sandbox_level: self.turn.windows_sandbox_level,
windows_sandbox_level: self.step_context.turn.windows_sandbox_level,
sandbox_permissions: self.approval_sandbox_permissions,
enable_shell_wrapper_parsing:
ENABLE_INTERCEPTED_EXEC_POLICY_SHELL_WRAPPER_PARSING,
@@ -9,6 +9,7 @@ use super::join_program_and_argv;
use super::map_exec_result;
use crate::config::Constrained;
use crate::sandboxing::SandboxPermissions;
use crate::session::step_context::StepContext;
use crate::session::tests::make_session_and_context;
use anyhow::Context;
use codex_execpolicy::Decision;
@@ -424,7 +425,7 @@ async fn preapproved_additional_permissions_escalate_intercepted_exec() -> anyho
let provider = CoreShellActionProvider {
policy: Arc::new(RwLock::new(codex_execpolicy::Policy::empty())),
session: Arc::new(session),
turn: Arc::new(turn_context),
step_context: StepContext::for_test(Arc::new(turn_context)),
call_id: "preapproved-additional-permissions".to_string(),
environment_id: "local".to_string(),
tool_name: GuardianCommandSource::Shell,
@@ -560,7 +561,7 @@ async fn execve_permission_request_hook_short_circuits_prompt() -> anyhow::Resul
let provider = CoreShellActionProvider {
policy: std::sync::Arc::new(RwLock::new(codex_execpolicy::Policy::empty())),
session: std::sync::Arc::new(session),
turn: std::sync::Arc::new(turn_context),
step_context: StepContext::for_test(std::sync::Arc::new(turn_context)),
call_id: "execve-hook-call".to_string(),
environment_id: "local".to_string(),
tool_name: GuardianCommandSource::Shell,
@@ -771,7 +772,7 @@ prefix_rule(pattern = ["{cat_path_literal}"], decision = "allow")
let provider = CoreShellActionProvider {
policy: Arc::new(RwLock::new(policy)),
session: Arc::new(session),
turn: Arc::new(turn_context),
step_context: StepContext::for_test(Arc::new(turn_context)),
call_id: "deny-read-prefix-allow".to_string(),
environment_id: "local".to_string(),
tool_name: GuardianCommandSource::Shell,
@@ -808,7 +809,7 @@ async fn denied_reads_keep_granular_sandbox_rejection_for_escalation() -> anyhow
let provider = CoreShellActionProvider {
policy: Arc::new(RwLock::new(PolicyParser::new().build())),
session: Arc::new(session),
turn: Arc::new(turn_context),
step_context: StepContext::for_test(Arc::new(turn_context)),
call_id: "deny-read-granular-sandbox-reject".to_string(),
environment_id: "local".to_string(),
tool_name: GuardianCommandSource::Shell,
@@ -174,7 +174,7 @@ impl Approvable<UnifiedExecRequest> for UnifiedExecRuntime<'_> {
) -> BoxFuture<'b, ReviewDecision> {
let keys = self.approval_keys(req);
let session = ctx.session;
let turn = ctx.turn;
let turn = &ctx.step_context.turn;
let call_id = ctx.call_id.to_string();
let command = req.command.clone();
let environment_id = Some(req.turn_environment.environment_id.clone());
@@ -264,7 +264,7 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
req: &UnifiedExecRequest,
ctx: &ToolCtx,
) -> Option<NetworkApprovalSpec> {
let file_system_sandbox_policy = ctx.turn.file_system_sandbox_policy();
let file_system_sandbox_policy = ctx.step_context.turn.file_system_sandbox_policy();
let sandbox_permissions = sandbox_permissions_preserving_denied_reads(
req.sandbox_permissions,
&file_system_sandbox_policy,
+1 -3
View File
@@ -8,7 +8,6 @@ use crate::sandboxing::ExecOptions;
use crate::sandboxing::SandboxPermissions;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnContext;
use crate::state::SessionServices;
use crate::tools::hook_names::HookToolName;
use crate::tools::network_approval::NetworkApprovalSpec;
@@ -121,7 +120,7 @@ where
#[derive(Clone)]
pub(crate) struct ApprovalCtx<'a> {
pub session: &'a Arc<Session>,
pub turn: &'a Arc<TurnContext>,
pub step_context: &'a Arc<StepContext>,
pub call_id: &'a str,
/// Guardian review lifecycle ID for this approval, when guardian is reviewing it.
///
@@ -384,7 +383,6 @@ pub(crate) trait Sandboxable {
pub(crate) struct ToolCtx {
pub session: Arc<Session>,
pub turn: Arc<TurnContext>,
pub step_context: Arc<StepContext>,
pub call_id: String,
pub tool_name: ToolName,
@@ -47,7 +47,7 @@ pub(crate) fn start_streaming_output(
let exit_token = process.cancellation_token();
let session_ref = Arc::clone(&context.session);
let turn_ref = Arc::clone(&context.turn);
let turn_ref = Arc::clone(&context.step_context.turn);
let call_id = context.call_id.clone();
tokio::spawn(async move {
+7 -4
View File
@@ -38,7 +38,6 @@ use tokio::sync::Mutex;
use crate::sandboxing::SandboxPermissions;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::session::turn_context::TurnEnvironment;
use crate::shell::ShellType;
use crate::tools::network_approval::DeferredNetworkApproval;
@@ -74,15 +73,19 @@ pub(crate) const MAX_UNIFIED_EXEC_PROCESSES: usize = 64;
pub(crate) struct UnifiedExecContext {
pub session: Arc<Session>,
pub turn: Arc<TurnContext>,
pub step_context: Arc<crate::session::step_context::StepContext>,
pub call_id: String,
}
impl UnifiedExecContext {
pub fn new(session: Arc<Session>, turn: Arc<TurnContext>, call_id: String) -> Self {
pub fn new(
session: Arc<Session>,
step_context: Arc<crate::session::step_context::StepContext>,
call_id: String,
) -> Self {
Self {
session,
turn,
step_context,
call_id,
}
}
+6 -2
View File
@@ -5,6 +5,7 @@ use crate::exec::ExecCapturePolicy;
use crate::exec::ExecExpiration;
use crate::sandboxing::ExecRequest;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use crate::session::tests::make_session_and_context;
use crate::session::turn_context::TurnContext;
use crate::tools::context::ExecCommandToolOutput;
@@ -121,8 +122,11 @@ async fn exec_command_with_tty(
)
.await?,
);
let context =
UnifiedExecContext::new(Arc::clone(session), Arc::clone(turn), "call".to_string());
let context = UnifiedExecContext::new(
Arc::clone(session),
StepContext::for_test(Arc::clone(turn)),
"call".to_string(),
);
let started_at = Instant::now();
let process_started_alive = !process.has_exited() && process.exit_code().is_none();
if process_started_alive {
@@ -331,7 +331,7 @@ async fn emit_failed_initial_exec_end_if_unstored(
emit_failed_exec_end_for_unified_exec(
Arc::clone(&context.session),
Arc::clone(&context.turn),
Arc::clone(&context.step_context.turn),
context.call_id.clone(),
request.command.clone(),
cwd,
@@ -435,7 +435,7 @@ impl UnifiedExecProcessManager {
let transcript = Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default()));
let event_ctx = ToolEventCtx::new(
context.session.as_ref(),
context.turn.as_ref(),
context.step_context.turn.as_ref(),
&context.call_id,
/*turn_diff_tracker*/ None,
);
@@ -599,7 +599,7 @@ impl UnifiedExecProcessManager {
let exit = exit_code.unwrap_or(-1);
emit_exec_end_for_unified_exec(
Arc::clone(&context.session),
Arc::clone(&context.turn),
Arc::clone(&context.step_context.turn),
context.call_id.clone(),
request.command.clone(),
cwd.clone(),
@@ -622,7 +622,12 @@ impl UnifiedExecProcessManager {
chunk_id,
wall_time,
raw_output: collected,
truncation_policy: context.turn.model_info.truncation_policy.into(),
truncation_policy: context
.step_context
.turn
.model_info
.truncation_policy
.into(),
max_output_tokens: request.max_output_tokens,
process_id: response_process_id,
exit_code,
@@ -897,7 +902,7 @@ impl UnifiedExecProcessManager {
spawn_exit_watcher(
Arc::clone(&process),
Arc::clone(&context.session),
Arc::clone(&context.turn),
Arc::clone(&context.step_context.turn),
context.call_id.clone(),
command.to_vec(),
cwd,
@@ -1106,7 +1111,12 @@ impl UnifiedExecProcessManager {
context: &UnifiedExecContext,
) -> Result<(UnifiedExecProcess, Option<DeferredNetworkApproval>), UnifiedExecError> {
let local_policy_env = create_env(
&context.turn.config.permissions.shell_environment_policy,
&context
.step_context
.turn
.config
.permissions
.shell_environment_policy,
/*thread_id*/ None,
);
let mut env = local_policy_env.clone();
@@ -1114,12 +1124,22 @@ impl UnifiedExecProcessManager {
CODEX_THREAD_ID_ENV_VAR.to_string(),
context.session.thread_id.to_string(),
);
let active_permission_profile = context.turn.config.permissions.active_permission_profile();
let active_permission_profile = context
.step_context
.turn
.config
.permissions
.active_permission_profile();
inject_permission_profile_env(&mut env, active_permission_profile.as_ref());
let env = apply_unified_exec_env(env);
let exec_server_env_config = ExecServerEnvConfig {
policy: exec_env_policy_from_shell_policy(
&context.turn.config.permissions.shell_environment_policy,
&context
.step_context
.turn
.config
.permissions
.shell_environment_policy,
),
local_policy_env,
};
@@ -1131,9 +1151,9 @@ impl UnifiedExecProcessManager {
.exec_policy
.create_exec_approval_requirement_for_command(ExecApprovalRequest {
command: &request.command,
approval_policy: context.turn.approval_policy.value(),
permission_profile: context.turn.permission_profile(),
windows_sandbox_level: context.turn.windows_sandbox_level,
approval_policy: context.step_context.turn.approval_policy.value(),
permission_profile: context.step_context.turn.permission_profile(),
windows_sandbox_level: context.step_context.turn.windows_sandbox_level,
sandbox_permissions: if request.additional_permissions_preapproved {
crate::sandboxing::SandboxPermissions::UseDefault
} else {
@@ -1153,6 +1173,7 @@ impl UnifiedExecProcessManager {
env,
exec_server_env_config: Some(exec_server_env_config),
explicit_env_overrides: context
.step_context
.turn
.config
.permissions
@@ -1170,7 +1191,7 @@ impl UnifiedExecProcessManager {
};
let tool_ctx = ToolCtx {
session: context.session.clone(),
turn: context.turn.clone(),
step_context: context.step_context.clone(),
call_id: context.call_id.clone(),
tool_name: ToolName::plain("exec_command"),
};
@@ -1179,8 +1200,8 @@ impl UnifiedExecProcessManager {
&mut runtime,
&req,
&tool_ctx,
&context.turn,
context.turn.approval_policy.value(),
&context.step_context.turn,
context.step_context.turn.approval_policy.value(),
)
.await
.map(|result| (result.output, result.deferred_network_approval))
@@ -1,4 +1,5 @@
use super::*;
use crate::session::step_context::StepContext;
use crate::unified_exec::clamp_yield_time;
use codex_network_proxy::ManagedNetworkSandboxContext;
use pretty_assertions::assert_eq;
@@ -261,7 +262,7 @@ async fn failed_initial_end_for_unstored_process_uses_fallback_output() {
let (session, turn, rx_event) = crate::session::tests::make_session_and_context_with_rx().await;
let context = UnifiedExecContext::new(
Arc::clone(&session),
Arc::clone(&turn),
StepContext::for_test(Arc::clone(&turn)),
"call-unified-denied".to_string(),
);
let request = ExecCommandRequest {