Files
codex/codex-rs/core/src/tasks/regular.rs
T
pakrym-oai fbd4efa9ed [codex] Use TurnInput for session task input (#24151)
## Why

The idea here is to erase the difference between initial and followup
inputs to a turn. Followup inputs are already represented as TurnInput.

Eventual goal is not to have explicit on task input at all and pull
everything from input Q.

## What Changed

- Changes `SessionTask::run` and the erased `AnySessionTask::run` path
to accept `Vec<TurnInput>`.
- Wraps user-submitted spawn input as `TurnInput::UserInput` at the
session task start boundary.
- Updates `run_turn` to record initial `TurnInput` using the same hook
and recording path used for pending input.
- Keeps review-specific conversion local to `ReviewTask`, where the
sub-Codex one-shot API still expects `Vec<UserInput>`.
- Moves the synthetic compact prompt into `CompactTask` and starts
compact tasks with empty task input.

## Validation

- `cargo check -p codex-core`
- `just test -p codex-core -E
'test(task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input)
| test(queued_response_items_for_next_turn_move_into_next_active_turn) |
test(steered_input_reopens_mailbox_delivery_for_current_turn)'`
2026-05-22 15:21:08 -07:00

91 lines
2.9 KiB
Rust

use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use crate::session::TurnInput;
use crate::session::turn::run_turn;
use crate::session::turn_context::TurnContext;
use crate::session_startup_prewarm::SessionStartupPrewarmResolution;
use crate::state::TaskKind;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnStartedEvent;
use tracing::Instrument;
use tracing::trace_span;
use super::SessionTask;
use super::SessionTaskContext;
#[derive(Default)]
pub(crate) struct RegularTask;
impl RegularTask {
pub(crate) fn new() -> Self {
Self
}
}
impl SessionTask for RegularTask {
fn kind(&self) -> TaskKind {
TaskKind::Regular
}
fn span_name(&self) -> &'static str {
"session_task.turn"
}
fn records_turn_token_usage_on_span(&self) -> bool {
true
}
async fn run(
self: Arc<Self>,
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
input: Vec<TurnInput>,
cancellation_token: CancellationToken,
) -> Option<String> {
let sess = session.clone_session();
let turn_extension_data = session.turn_extension_data();
let run_turn_span = trace_span!("run_turn");
// Regular turns emit `TurnStarted` inline so first-turn lifecycle does
// not wait on startup prewarm resolution.
let event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: ctx.sub_id.clone(),
trace_id: ctx.trace_id.clone(),
started_at: ctx.turn_timing_state.started_at_unix_secs().await,
model_context_window: ctx.model_context_window(),
collaboration_mode_kind: ctx.collaboration_mode.mode,
});
sess.send_event(ctx.as_ref(), event).await;
sess.set_server_reasoning_included(/*included*/ false).await;
let prewarmed_client_session = match sess
.consume_startup_prewarm_for_regular_turn(&cancellation_token)
.await
{
SessionStartupPrewarmResolution::Cancelled => return None,
SessionStartupPrewarmResolution::Unavailable { .. } => None,
SessionStartupPrewarmResolution::Ready(prewarmed_client_session) => {
Some(*prewarmed_client_session)
}
};
let mut next_input = input;
let mut prewarmed_client_session = prewarmed_client_session;
loop {
let last_agent_message = run_turn(
Arc::clone(&sess),
Arc::clone(&ctx),
Arc::clone(&turn_extension_data),
next_input,
prewarmed_client_session.take(),
cancellation_token.child_token(),
)
.instrument(run_turn_span.clone())
.await;
if !sess.input_queue.has_pending_input(&sess.active_turn).await {
return last_agent_message;
}
next_input = Vec::new();
}
}
}