openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: dispatch queued turns from app-server

Browse Source
This commit is contained in:
efrazer-oai
2026-05-30 15:35:54 -07:00
committed by Edward Frazer
parent ee4634a18f
commit ef8672f22b
15 changed files with 1092 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/message_processor.rs
+2
View File
@@ -429,6 +429,7 @@ impl MessageProcessor {
config_manager.clone(),
Arc::clone(&pending_thread_unloads),
thread_state_manager.clone(),
state_db.clone(),
thread_watch_manager.clone(),
Arc::clone(&thread_list_state_permit),
Arc::clone(&skills_watcher),
@@ -439,6 +440,7 @@ impl MessageProcessor {
config_manager.clone(),
state_db.clone(),
thread_state_manager.clone(),
turn_processor.clone(),
);
let thread_processor = ThreadRequestProcessor::new(
auth_manager.clone(),
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+27
View File
@@ -1,4 +1,5 @@
use super::*;
use crate::thread_state::TerminalTurnOutcome;
pub(super) const THREAD_UNLOADING_DELAY: Duration = Duration::from_secs(30 * 60);
@@ -13,6 +14,7 @@ pub(super) struct ListenerTaskContext {
pub(super) fallback_model_provider: String,
pub(super) codex_home: PathBuf,
pub(super) skills_watcher: Arc<SkillsWatcher>,
pub(super) thread_queue_processor: ThreadQueueRequestProcessor,
}
struct UnloadingState {
@@ -260,6 +262,7 @@ pub(super) async fn ensure_listener_task_running(
thread_list_state_permit,
fallback_model_provider,
codex_home,
thread_queue_processor,
..
} = listener_task_context;
let outgoing_for_task = Arc::clone(&outgoing);
@@ -284,6 +287,7 @@ pub(super) async fn ensure_listener_task_running(
&thread_watch_manager,
&outgoing_for_task,
&pending_thread_unloads,
&thread_queue_processor,
listener_command,
)
.await;
@@ -339,6 +343,23 @@ pub(super) async fn ensure_listener_task_running(
fallback_model_provider.clone(),
)
.await;
let terminal_outcome = match &event.msg {
EventMsg::TurnComplete(_) => Some(TerminalTurnOutcome::Completed),
EventMsg::TurnAborted(_) => Some(TerminalTurnOutcome::Aborted),
EventMsg::Error(error) if error.affects_turn_status() => {
Some(TerminalTurnOutcome::Error)
}
_ => None,
};
if let Some(outcome) = terminal_outcome {
thread_queue_processor
.settle_dispatch_after_terminal_turn(
conversation_id,
event.id.as_str(),
outcome,
)
.await;
}
}
unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => {
if !unloading_watchers_open {
@@ -454,6 +475,7 @@ pub(super) async fn handle_thread_listener_command(
thread_watch_manager: &ThreadWatchManager,
outgoing: &Arc<OutgoingMessageSender>,
pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
thread_queue_processor: &ThreadQueueRequestProcessor,
listener_command: ThreadListenerCommand,
) {
match listener_command {
@@ -467,6 +489,7 @@ pub(super) async fn handle_thread_listener_command(
thread_watch_manager,
outgoing,
pending_thread_unloads,
thread_queue_processor,
*resume_request,
)
.await;
@@ -524,6 +547,7 @@ pub(super) async fn handle_pending_thread_resume_request(
thread_watch_manager: &ThreadWatchManager,
outgoing: &Arc<OutgoingMessageSender>,
pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
thread_queue_processor: &ThreadQueueRequestProcessor,
pending: crate::thread_state::PendingThreadResumeRequest,
) {
let active_turn = {
@@ -697,6 +721,9 @@ pub(super) async fn handle_pending_thread_resume_request(
{
tracing::warn!("failed to continue active goal after running-thread resume: {err}");
}
thread_queue_processor
.emit_resume_queue_snapshot_and_drain(conversation_id)
.await;
}
pub(super) async fn send_thread_goal_snapshot_notification(
codex-rs/app-server/src/request_processors/thread_processor.rs
+6 -3
View File
@@ -789,6 +789,7 @@ impl ThreadRequestProcessor {
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.to_path_buf(),
skills_watcher: Arc::clone(&self.skills_watcher),
thread_queue_processor: self.thread_queue_processor.clone(),
}
}
@@ -889,6 +890,7 @@ impl ThreadRequestProcessor {
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.to_path_buf(),
skills_watcher: Arc::clone(&self.skills_watcher),
thread_queue_processor: self.thread_queue_processor.clone(),
};
let request_trace = request_context.request_trace();
let config_manager = self.config_manager.clone();
@@ -2647,9 +2649,10 @@ impl ThreadRequestProcessor {
config_snapshot.active_permission_profile,
);
let token_usage_thread = include_turns.then(|| thread.clone());
let response_rollout_items = response_history.get_rollout_items();
let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() {
match build_thread_resume_initial_turns_page(
&response_history.get_rollout_items(),
&response_rollout_items,
thread.status.clone(),
/*has_live_running_thread*/ false,
/*active_turn*/ None,
@@ -2693,7 +2696,7 @@ impl ThreadRequestProcessor {
// rebuilding history only to attribute a replayed usage update.
if let Some(token_usage_thread) = token_usage_thread {
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
&response_history.get_rollout_items(),
&response_rollout_items,
token_usage_thread.turns.as_slice(),
);
// The client needs restored usage before it starts another turn.
@@ -2713,7 +2716,7 @@ impl ThreadRequestProcessor {
.emit_resume_goal_snapshot_and_continue(thread_id, codex_thread.as_ref())
.await;
self.thread_queue_processor
.emit_resume_queue_snapshot(thread_id)
.recover_resume_queue_snapshot_and_drain(thread_id, &response_rollout_items)
.await;
}
Err(err) => {
codex-rs/app-server/src/request_processors/thread_queue_processor.rs
+283 -7
View File
@@ -1,7 +1,9 @@
use super::*;
use crate::thread_state::TerminalTurnOutcome;
const THREAD_QUEUE_LIST_DEFAULT_LIMIT: usize = 25;
const THREAD_QUEUE_LIST_MAX_LIMIT: usize = 100;
const THREAD_QUEUE_MAX_SUBMISSION_BYTES: usize = MAX_USER_INPUT_TEXT_CHARS;
#[derive(Clone)]
pub(crate) struct ThreadQueueRequestProcessor {
@@ -10,6 +12,7 @@ pub(crate) struct ThreadQueueRequestProcessor {
config_manager: ConfigManager,
state_db: Option<StateDbHandle>,
thread_state_manager: ThreadStateManager,
turn_processor: TurnRequestProcessor,
}
impl ThreadQueueRequestProcessor {
@@ -19,6 +22,7 @@ impl ThreadQueueRequestProcessor {
config_manager: ConfigManager,
state_db: Option<StateDbHandle>,
thread_state_manager: ThreadStateManager,
turn_processor: TurnRequestProcessor,
) -> Self {
Self {
thread_manager,
@@ -26,6 +30,7 @@ impl ThreadQueueRequestProcessor {
config_manager,
state_db,
thread_state_manager,
turn_processor,
}
}
@@ -70,6 +75,11 @@ impl ThreadQueueRequestProcessor {
let payload = serde_json::to_vec(&submission).map_err(|err| {
internal_error(format!("failed to serialize queued turn payload: {err}"))
})?;
if payload.len() > THREAD_QUEUE_MAX_SUBMISSION_BYTES {
return Err(invalid_params(format!(
"Queued turn submission exceeds the maximum length of {THREAD_QUEUE_MAX_SUBMISSION_BYTES} bytes."
)));
}
let record = self
.state_db()?
.thread_queue()
@@ -86,6 +96,7 @@ impl ThreadQueueRequestProcessor {
)
.await;
self.emit_thread_queue_changed(thread_id).await;
self.drain_thread_queue_if_idle(thread_id).await;
Ok(None)
}
@@ -139,6 +150,7 @@ impl ThreadQueueRequestProcessor {
.await;
if deleted {
self.emit_thread_queue_changed(thread_id).await;
self.drain_thread_queue_if_idle(thread_id).await;
}
Ok(None)
}
@@ -168,22 +180,266 @@ impl ThreadQueueRequestProcessor {
},
)
.await;
self.send_thread_queue_changed(thread_id, queued_turns)
.await;
self.emit_thread_queue_changed(thread_id).await;
self.drain_thread_queue_if_idle(thread_id).await;
Ok(None)
}
pub(crate) async fn emit_resume_queue_snapshot(&self, thread_id: ThreadId) {
pub(crate) async fn recover_resume_queue_snapshot_and_drain(
&self,
thread_id: ThreadId,
rollout_items: &[RolloutItem],
) {
if self.require_enabled().await.is_err() {
return;
}
for turn in codex_app_server_protocol::build_turns_from_rollout_items(rollout_items) {
let outcome = match turn.status {
TurnStatus::Completed => Some(TerminalTurnOutcome::Completed),
TurnStatus::Interrupted => Some(TerminalTurnOutcome::Aborted),
TurnStatus::Failed => Some(TerminalTurnOutcome::Error),
TurnStatus::InProgress => None,
};
if let Some(outcome) = outcome {
self.settle_dispatch_claim(thread_id, turn.id.as_str(), outcome)
.await;
}
}
let Some(state_db) = self.state_db.as_ref() else {
return;
};
let failure = turn_error("queued turn dispatch was interrupted while app-server restarted");
let failure_json = match serde_json::to_vec(&failure) {
Ok(failure_json) => failure_json,
Err(err) => {
tracing::warn!("failed to serialize queued turn recovery failure: {err}");
return;
}
};
match state_db
.thread_queue()
.recover_dispatching_thread_queued_turns(thread_id, failure_json.as_slice())
.await
{
Ok(_) => {}
Err(err) => {
tracing::warn!("failed to recover queued turns for thread {thread_id}: {err}");
return;
}
}
self.emit_thread_queue_changed(thread_id).await;
self.drain_thread_queue_if_idle(thread_id).await;
}
pub(crate) async fn emit_resume_queue_snapshot_and_drain(&self, thread_id: ThreadId) {
if self.require_enabled().await.is_err() {
return;
}
self.emit_thread_queue_changed(thread_id).await;
self.drain_thread_queue_if_idle(thread_id).await;
}
pub(crate) async fn settle_dispatch_after_terminal_turn(
&self,
thread_id: ThreadId,
turn_id: &str,
outcome: TerminalTurnOutcome,
) {
self.settle_dispatch_claim(thread_id, turn_id, outcome)
.await;
self.drain_thread_queue_if_idle(thread_id).await;
}
async fn settle_dispatch_claim(
&self,
thread_id: ThreadId,
turn_id: &str,
outcome: TerminalTurnOutcome,
) {
let Some(state_db) = self.state_db.as_ref() else {
return;
};
let result = match outcome {
TerminalTurnOutcome::Completed => {
state_db
.thread_queue()
.remove_dispatching_thread_queued_turn(thread_id, turn_id)
.await
}
TerminalTurnOutcome::Aborted | TerminalTurnOutcome::Error => {
let message = match outcome {
TerminalTurnOutcome::Aborted => "queued turn was aborted",
TerminalTurnOutcome::Error => "queued turn failed before it started",
TerminalTurnOutcome::Completed => unreachable!(),
};
let failure_json = match serde_json::to_vec(&turn_error(message)) {
Ok(failure_json) => failure_json,
Err(err) => {
tracing::warn!("failed to serialize queued turn failure: {err}");
return;
}
};
state_db
.thread_queue()
.mark_dispatching_thread_queued_turn_failed(
thread_id,
turn_id,
failure_json.as_slice(),
)
.await
}
};
match result {
Ok(true) => self.emit_thread_queue_changed(thread_id).await,
Ok(false) => {}
Err(err) => {
tracing::warn!(
"failed to settle queued dispatch claim for thread {thread_id}: {err}"
);
}
}
}
async fn drain_thread_queue_if_idle(&self, thread_id: ThreadId) {
if self.require_enabled().await.is_err() {
return;
}
let Some(state_db) = self.state_db.as_ref() else {
return;
};
loop {
let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {
return;
};
if matches!(thread.agent_status().await, AgentStatus::Running) {
return;
}
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
{
let thread_state = thread_state.lock().await;
if thread_state.active_turn_snapshot().is_some()
|| !matches!(
thread_state.pending_turn_starts,
crate::thread_state::PendingTurnStarts::None
)
{
return;
}
}
let record = match state_db
.thread_queue()
.claim_head_thread_queued_turn(thread_id)
.await
{
Ok(Some(record)) => record,
Ok(None) => return,
Err(err) => {
tracing::warn!("failed to claim queued turn for thread {thread_id}: {err}");
return;
}
};
self.emit_thread_queue_changed(thread_id).await;
let submission = match serde_json::from_slice::<TurnSubmission>(
record.turn_submission_jsonb.as_slice(),
) {
Ok(submission) => submission,
Err(err) => {
self.fail_dispatch(
thread_id,
record.queued_turn_id.as_str(),
turn_error(format!("queued turn payload could not be read: {err}")),
)
.await;
return;
}
};
match self
.turn_processor
.queued_turn_start(thread_id, submission)
.await
{
Ok(response) => {
let turn_id = response.turn.id;
match state_db
.thread_queue()
.set_dispatching_thread_queued_turn_turn_id(
record.queued_turn_id.as_str(),
turn_id.as_str(),
)
.await
{
Ok(true) => {}
Ok(false) => {
tracing::warn!(
"queued turn {} lost its dispatch claim before turn {turn_id} was recorded",
record.queued_turn_id
);
return;
}
Err(err) => {
tracing::warn!(
"failed to record dispatch turn {turn_id} for queued turn {}: {err}",
record.queued_turn_id
);
return;
}
}
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
let terminal_outcome = {
let thread_state = thread_state.lock().await;
thread_state.terminal_outcome_for_turn(turn_id.as_str())
};
if let Some(outcome) = terminal_outcome {
self.settle_dispatch_claim(thread_id, turn_id.as_str(), outcome)
.await;
continue;
}
return;
}
Err(err) => {
self.fail_dispatch(
thread_id,
record.queued_turn_id.as_str(),
turn_error(format!(
"queued turn could not start: {message}",
message = err.message
)),
)
.await;
return;
}
}
}
}
async fn fail_dispatch(&self, thread_id: ThreadId, queued_turn_id: &str, error: TurnError) {
let Some(state_db) = self.state_db.as_ref() else {
return;
};
let failure_json = match serde_json::to_vec(&error) {
Ok(failure_json) => failure_json,
Err(err) => {
tracing::warn!("failed to serialize queued turn failure: {err}");
return;
}
};
match state_db
.thread_queue()
.mark_thread_queued_turn_failed(queued_turn_id, failure_json.as_slice())
.await
{
Ok(true) => self.emit_thread_queue_changed(thread_id).await,
Ok(false) => tracing::warn!(
"queued turn {queued_turn_id} could not be marked failed because its dispatch claim disappeared"
),
Err(err) => tracing::warn!("failed to mark queued turn {queued_turn_id} failed: {err}"),
}
}
async fn emit_thread_queue_changed(&self, thread_id: ThreadId) {
match self.list_visible_queued_turns(thread_id).await {
Ok(queued_turns) => {
self.send_thread_queue_changed(thread_id, queued_turns)
match self.queue_snapshot(thread_id).await {
Ok((queued_turns, dispatching_queued_turn_id)) => {
self.send_thread_queue_changed(thread_id, queued_turns, dispatching_queued_turn_id)
.await;
}
Err(err) => {
@@ -192,7 +448,12 @@ impl ThreadQueueRequestProcessor {
}
}
async fn send_thread_queue_changed(&self, thread_id: ThreadId, queued_turns: Vec<QueuedTurn>) {
async fn send_thread_queue_changed(
&self,
thread_id: ThreadId,
queued_turns: Vec<QueuedTurn>,
dispatching_queued_turn_id: Option<String>,
) {
let subscribed_connection_ids = self
.thread_state_manager
.subscribed_connection_ids(thread_id)
@@ -203,6 +464,7 @@ impl ThreadQueueRequestProcessor {
ServerNotification::ThreadQueueChanged(ThreadQueueChangedNotification {
thread_id: thread_id.to_string(),
queued_turns,
dispatching_queued_turn_id,
}),
)
.await;
@@ -225,6 +487,20 @@ impl ThreadQueueRequestProcessor {
.map(queued_turn_from_state)
.collect()
}
async fn queue_snapshot(
&self,
thread_id: ThreadId,
) -> Result<(Vec<QueuedTurn>, Option<String>), JSONRPCErrorError> {
let queued_turns = self.list_visible_queued_turns(thread_id).await?;
let dispatching_queued_turn_id = self
.state_db()?
.thread_queue()
.dispatching_thread_queued_turn_id(thread_id)
.await
.map_err(|err| internal_error(format!("failed to read queued dispatch: {err}")))?;
Ok((queued_turns, dispatching_queued_turn_id))
}
}
fn parse_queue_thread_id(thread_id: &str) -> Result<ThreadId, JSONRPCErrorError> {
codex-rs/app-server/src/request_processors/turn_processor.rs
+75 -9
View File
@@ -13,6 +13,7 @@ pub(crate) struct TurnRequestProcessor {
config_manager: ConfigManager,
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_state_manager: ThreadStateManager,
state_db: Option<StateDbHandle>,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<Semaphore>,
skills_watcher: Arc<SkillsWatcher>,
@@ -83,6 +84,7 @@ impl TurnRequestProcessor {
config_manager: ConfigManager,
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_state_manager: ThreadStateManager,
state_db: Option<StateDbHandle>,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<Semaphore>,
skills_watcher: Arc<SkillsWatcher>,
@@ -97,6 +99,7 @@ impl TurnRequestProcessor {
config_manager,
pending_thread_unloads,
thread_state_manager,
state_db,
thread_watch_manager,
thread_list_state_permit,
skills_watcher,
@@ -120,6 +123,27 @@ impl TurnRequestProcessor {
.map(|response| Some(response.into()))
}
pub(crate) async fn queued_turn_start(
&self,
thread_id: ThreadId,
submission: TurnSubmission,
) -> Result<TurnStartResponse, JSONRPCErrorError> {
Self::validate_v2_input_limit(&submission.input)?;
let thread_id_string = thread_id.to_string();
let (_, thread) = self.load_thread(&thread_id_string).await?;
let params = TurnStartParams {
thread_id: thread_id_string,
input: submission.input,
responsesapi_client_metadata: submission.responsesapi_client_metadata,
additional_context: submission.additional_context,
environments: submission.environments,
output_schema: submission.output_schema,
..Default::default()
};
self.start_turn_from_params(thread_id, thread, params, /*trace_context*/ None)
.await
}
pub(crate) async fn thread_inject_items(
&self,
params: ThreadInjectItemsParams,
@@ -407,7 +431,48 @@ impl TurnRequestProcessor {
.inspect_err(|error| {
self.track_error_response(&request_id, error, /*error_type*/ None);
})?;
let response = self
.start_turn_from_params(
thread_id,
thread,
params,
self.request_trace_context(&request_id).await,
)
.await?;
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
{
let mut thread_state = thread_state.lock().await;
let turn_lifecycle_already_visible = thread_state
.active_turn_snapshot()
.is_some_and(|turn| turn.id == response.turn.id)
|| thread_state.last_terminal_turn_id.as_deref() == Some(response.turn.id.as_str());
if !turn_lifecycle_already_visible {
match &mut thread_state.pending_turn_starts {
crate::thread_state::PendingTurnStarts::None => {
thread_state.pending_turn_starts =
crate::thread_state::PendingTurnStarts::WaitingForLifecycle {
turn_ids: HashSet::from([response.turn.id.clone()]),
};
}
crate::thread_state::PendingTurnStarts::WaitingForLifecycle { turn_ids } => {
turn_ids.insert(response.turn.id.clone());
}
}
}
}
self.outgoing
.record_request_turn_id(&request_id, &response.turn.id)
.await;
Ok(response)
}
async fn start_turn_from_params(
&self,
thread_id: ThreadId,
thread: Arc<CodexThread>,
params: TurnStartParams,
trace_context: Option<W3cTraceContext>,
) -> Result<TurnStartResponse, JSONRPCErrorError> {
let environment_selections = self.parse_environment_selections(params.environments)?;
// Map v2 input items to core input items.
@@ -452,15 +517,11 @@ impl TurnRequestProcessor {
let turn_id = thread
.submit_user_input_with_client_user_message_id(
turn_op,
self.request_trace_context(&request_id).await,
trace_context,
client_user_message_id,
)
.await
.map_err(|err| {
let error = internal_error(format!("failed to start turn: {err}"));
self.track_error_response(&request_id, &error, /*error_type*/ None);
error
})?;
.map_err(|err| internal_error(format!("failed to start turn: {err}")))?;
if turn_has_input {
let config_snapshot = thread.config_snapshot().await;
@@ -474,9 +535,6 @@ impl TurnRequestProcessor {
);
}
self.outgoing
.record_request_turn_id(&request_id, &turn_id)
.await;
let turn = Turn {
id: turn_id,
items: vec![],
@@ -1279,6 +1337,14 @@ impl TurnRequestProcessor {
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.to_path_buf(),
skills_watcher: Arc::clone(&self.skills_watcher),
thread_queue_processor: ThreadQueueRequestProcessor::new(
Arc::clone(&self.thread_manager),
Arc::clone(&self.outgoing),
self.config_manager.clone(),
self.state_db.clone(),
self.thread_state_manager.clone(),
self.clone(),
),
}
}
codex-rs/app-server/src/thread_state.rs
+121 -4
View File
@@ -26,6 +26,28 @@ use tracing::error;
type PendingInterruptQueue = Vec<ConnectionRequestId>;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TerminalTurnOutcome {
Completed,
Aborted,
Error,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct LastTerminalTurn {
turn_id: String,
outcome: TerminalTurnOutcome,
}
#[derive(Default)]
pub(crate) enum PendingTurnStarts {
#[default]
None,
WaitingForLifecycle {
turn_ids: HashSet<String>,
},
}
pub(crate) struct PendingThreadResumeRequest {
pub(crate) request_id: ConnectionRequestId,
pub(crate) history_items: Vec<RolloutItem>,
@@ -74,8 +96,10 @@ pub(crate) struct TurnSummary {
pub(crate) struct ThreadState {
pub(crate) pending_interrupts: PendingInterruptQueue,
pub(crate) pending_rollbacks: Option<ConnectionRequestId>,
pub(crate) pending_turn_starts: PendingTurnStarts,
pub(crate) turn_summary: TurnSummary,
pub(crate) last_terminal_turn_id: Option<String>,
last_terminal_turn: Option<LastTerminalTurn>,
pub(crate) cancel_tx: Option<oneshot::Sender<()>>,
pub(crate) experimental_raw_events: bool,
pub(crate) listener_generation: u64,
@@ -118,6 +142,7 @@ impl ThreadState {
let _ = cancel_tx.send(());
}
self.listener_command_tx = None;
self.pending_turn_starts = PendingTurnStarts::None;
self.current_turn_history.reset();
self.listener_thread = None;
self.watch_registration = WatchRegistration::default();
@@ -137,17 +162,54 @@ impl ThreadState {
self.current_turn_history.active_turn_snapshot()
}
pub(crate) fn track_current_turn_event(&mut self, event_turn_id: &str, event: &EventMsg) {
pub(crate) fn terminal_outcome_for_turn(&self, turn_id: &str) -> Option<TerminalTurnOutcome> {
self.last_terminal_turn
.as_ref()
.filter(|terminal_turn| terminal_turn.turn_id == turn_id)
.map(|terminal_turn| terminal_turn.outcome)
}
pub(crate) fn track_current_turn_event(
&mut self,
event_turn_id: &str,
event: &EventMsg,
) -> bool {
if let EventMsg::TurnStarted(payload) = event {
self.turn_summary.started_at = payload.started_at;
}
self.current_turn_history.handle_event(event);
if matches!(event, EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_))
&& !self.current_turn_history.has_active_turn()
{
let pending_turn_start_resolved = matches!(
event,
EventMsg::TurnStarted(_) | EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_)
) || matches!(event, EventMsg::Error(error) if error.affects_turn_status());
let pending_turn_starts_cleared = match &mut self.pending_turn_starts {
PendingTurnStarts::None => false,
PendingTurnStarts::WaitingForLifecycle { turn_ids } => {
pending_turn_start_resolved && turn_ids.remove(event_turn_id) && turn_ids.is_empty()
}
};
if pending_turn_starts_cleared {
self.pending_turn_starts = PendingTurnStarts::None;
}
let terminal_outcome = match event {
EventMsg::TurnComplete(_) if !self.current_turn_history.has_active_turn() => {
Some(TerminalTurnOutcome::Completed)
}
EventMsg::TurnAborted(_) => Some(TerminalTurnOutcome::Aborted),
EventMsg::Error(error) if error.affects_turn_status() => {
Some(TerminalTurnOutcome::Error)
}
_ => None,
};
if let Some(outcome) = terminal_outcome {
self.last_terminal_turn = Some(LastTerminalTurn {
turn_id: event_turn_id.to_string(),
outcome,
});
self.last_terminal_turn_id = Some(event_turn_id.to_string());
self.current_turn_history.reset();
}
pending_turn_starts_cleared
}
pub(crate) fn note_thread_settings(&mut self, thread_settings: ThreadSettings) -> bool {
@@ -198,8 +260,63 @@ mod tests {
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnStartedEvent;
use pretty_assertions::assert_eq;
fn turn_started(turn_id: &str) -> EventMsg {
EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_id.to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
})
}
#[test]
fn aborted_turn_clears_active_snapshot() {
let mut state = ThreadState::default();
state.track_current_turn_event("turn-1", &turn_started("turn-1"));
state.track_current_turn_event(
"turn-1",
&EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some("turn-1".to_string()),
reason: TurnAbortReason::Interrupted,
completed_at: None,
duration_ms: None,
}),
);
assert!(state.active_turn_snapshot().is_none());
assert_eq!(
state.terminal_outcome_for_turn("turn-1"),
Some(TerminalTurnOutcome::Aborted)
);
}
#[test]
fn status_affecting_error_clears_active_snapshot() {
let mut state = ThreadState::default();
state.track_current_turn_event("turn-1", &turn_started("turn-1"));
state.track_current_turn_event(
"turn-1",
&EventMsg::Error(ErrorEvent {
message: "boom".to_string(),
codex_error_info: None,
}),
);
assert!(state.active_turn_snapshot().is_none());
assert_eq!(
state.terminal_outcome_for_turn("turn-1"),
Some(TerminalTurnOutcome::Error)
);
assert_eq!(state.last_terminal_turn_id.as_deref(), Some("turn-1"));
}
#[test]
fn note_thread_settings_reports_only_effective_changes() {
let mut state = ThreadState::default();