Files
codex/codex-rs/protocol/src/turn_input.rs
Eric Traut 7c098d8741 Gate new turn submissions on host shutdown admission (#43943)
## Why

Hosts need to stop new turn-input work during shutdown without consuming pending input or preventing already-running delegated work from finishing.

## What changed

- Add an optional `TurnStartAdmission` extension gate, checked before reserving or starting a new turn. Hosts without a gate retain existing behavior.
- Return `NotSubmittedReason::ServerDraining` for refused starts and surface an app-server error instructing clients to reconnect and retry.
- Keep steering, parent-delegated subagent input, and memory-only mailbox wakeups available during drain, while gating automatic starts.
- Close realtime conversations with an ordered handoff, error, and close event sequence when a handoff is refused during drain.

## Testing

Add regression coverage for rejected input staying out of subsequent requests, persisted queue items remaining available for later starts, delegated agent and review work completing during drain, mailbox wakeups, and realtime handoff error ordering.

GitOrigin-RevId: 03dbdcd71eab200e597c0649e6eb39bd92dbc82f
2026-09-08 23:45:10 +00:00

247 lines
9.0 KiB
Rust

//! Turn-input request and result types shared by Core's submission APIs.
use crate::models::ResponseItem;
use crate::protocol::AdditionalContextEntry;
use crate::protocol::InterAgentCommunication;
use crate::protocol::NonSteerableTurnKind;
use crate::protocol::ThreadSettingsOverrides;
use crate::protocol::W3cTraceContext;
use crate::user_input::UserInput;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
use std::collections::BTreeMap;
use std::collections::HashMap;
use ts_rs::TS;
/// Result of stopping an unfinished root turn so another worker can recover it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SuspendTurnOutcome {
Suspended {
turn_id: String,
},
NotActive,
/// A currently loaded descendant would remain running after root handoff.
HasLiveDescendants,
UnsupportedTask,
}
/// Input consumed by a regular turn.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum TurnInput {
UserInput {
content: Vec<UserInput>,
client_id: Option<String>,
},
ResponseItem(ResponseItem),
InterAgentCommunication(InterAgentCommunication),
}
/// One turn input and the context that follows it through submission.
///
/// Callers choose start-or-steer, idle-start, or steer-only behavior through
/// the corresponding `CodexThread` method.
#[derive(Clone, Debug)]
pub struct TurnInputRequest {
pub input: TurnInput,
pub thread_settings: ThreadSettingsOverrides,
pub start: TurnStartOptions,
pub additional_context: BTreeMap<String, AdditionalContextEntry>,
pub responsesapi_client_metadata: Option<HashMap<String, String>>,
pub trace: Option<W3cTraceContext>,
}
/// Request to resume sampling for an interrupted regular turn.
///
/// Sampling restarts under `turn_id`, which must be the ID already recorded
/// for that turn.
#[derive(Clone, Debug)]
pub struct RecoverTurnRequest {
pub turn_id: String,
pub thread_settings: ThreadSettingsOverrides,
pub trace: Option<W3cTraceContext>,
/// Program recorded in the interrupted turn's persisted context.
pub cyber_access_program: Option<CyberAccessProgram>,
}
impl TurnInputRequest {
/// Creates turn input that can be passed to one of the submission methods.
pub fn new(input: TurnInput) -> Self {
Self {
input,
thread_settings: ThreadSettingsOverrides::default(),
start: TurnStartOptions::default(),
additional_context: BTreeMap::new(),
responsesapi_client_metadata: None,
trace: None,
}
}
/// Creates ordinary user input without a client-provided message id.
pub fn user_input(content: Vec<UserInput>) -> Self {
Self::new(TurnInput::UserInput {
content,
client_id: None,
})
}
/// Persistent thread settings applied when Core accepts this input.
///
/// Settings are applied for both `Started` and `Steered`. A steered input
/// cannot change its already-active turn context, so those settings apply
/// to subsequent turns. `NotSubmitted` leaves them unapplied.
pub fn with_thread_settings(mut self, thread_settings: ThreadSettingsOverrides) -> Self {
self.thread_settings = thread_settings;
self
}
/// Options consulted only by start-capable submission methods when this
/// request starts a turn.
pub fn on_start(mut self, start: TurnStartOptions) -> Self {
self.start = start;
self
}
/// Context merged whether this request starts or steers.
pub fn with_additional_context(
mut self,
additional_context: BTreeMap<String, AdditionalContextEntry>,
) -> Self {
self.additional_context = additional_context;
self
}
/// Responses metadata attached whether this request starts or steers.
pub fn with_responses_metadata(
mut self,
responsesapi_client_metadata: Option<HashMap<String, String>>,
) -> Self {
self.responsesapi_client_metadata = responsesapi_client_metadata;
self
}
/// Trace context used when this request crosses the session loop.
pub fn with_trace(mut self, trace: Option<W3cTraceContext>) -> Self {
self.trace = trace;
self
}
}
/// How Core should route submitted turn input.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TurnInputMode {
/// Start a regular turn when idle, otherwise steer the active regular turn.
StartOrSteer,
/// Start only when the thread is idle.
StartIfIdle,
/// Steer only if this exact turn is active.
Steer { expected_turn_id: String },
}
/// Requested cyber treatment for a ChatGPT-authenticated Codex turn.
/// Authorization and model-tier restrictions remain server-owned.
#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum CyberAccessProgram {
Standard,
DaybreakBlue,
DaybreakRed,
}
/// Options for the new-turn branch of a submission.
///
/// Core only records these options when input starts a turn. When input steers,
/// `final_output_json_schema` is a compatibility requirement: Core only
/// accepts the steer if the active turn already uses the same schema. For
/// child input, Core also compares root lineage to detect ambiguity.
#[derive(Clone, Debug, Default)]
pub struct TurnStartOptions {
/// Source classification for the caller that starts a new turn.
/// Ignored when the submitted input steers an active turn.
pub turn_trigger: Option<String>,
/// Structured-output schema for a new turn. When steering, Core rejects
/// the input if the active turn uses a different schema.
pub final_output_json_schema: Option<Value>,
/// Service tier for a new turn, without changing the thread's preference.
pub service_tier: Option<String>,
/// Parent turn lineage recorded if this request starts a new turn.
pub parent_turn_id: Option<String>,
/// Causal root turn lineage recorded if this request starts a new turn.
pub root_turn_id: Option<String>,
/// Explicit cyber treatment for this turn. Omission preserves the backend's
/// automatic behavior.
pub cyber_access_program: Option<CyberAccessProgram>,
}
/// What Core did with input submitted through `start_or_steer_turn`.
///
/// Started and Steered only mean Core accepted the input for turn processing. They
/// do not wait for user-prompt hooks, updating the in-memory model context,
/// rollout persistence, or sampling.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TurnInputSubmission {
/// Core started a turn. Persistent thread settings and start options were applied.
Started { turn_id: String },
/// Core steered an active turn. Persistent thread settings were applied for
/// subsequent turns. No new turn was created, so lineage metadata was not
/// recorded. If the request included `final_output_json_schema`, the active
/// turn already used the same schema.
Steered { turn_id: String },
/// Core rejected the input without applying settings or start options.
NotSubmitted { reason: NotSubmittedReason },
}
/// What Core did with input submitted only for an idle turn start.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum StartIfIdleSubmission {
/// Core started a turn. Persistent thread settings and start options were applied.
Started { turn_id: String },
/// Core rejected the input without applying settings or start options.
NotSubmitted { reason: NotSubmittedReason },
}
/// What Core did with input submitted only for steering.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SteerSubmission {
/// Core steered an active turn. Persistent thread settings were applied for
/// subsequent turns.
Steered { turn_id: String },
/// Core rejected the input without applying settings.
NotSubmitted { reason: NotSubmittedReason },
}
/// Why Core did not accept submitted turn input for turn processing.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum NotSubmittedReason {
/// The host is draining and no longer permits new regular turns.
ServerDraining,
/// `start_turn_if_idle` found an active turn.
NotIdle,
/// `start_turn_if_idle` yielded to higher-priority trigger-turn mailbox input.
PendingTriggerTurn,
/// `start_turn_if_idle` received automatic non-user input for a turn that
/// would run in Plan mode.
PlanMode,
/// `steer_turn` found no active turn.
NoActiveTurn,
/// `steer_turn` found a different active turn.
ExpectedTurnMismatch { expected: String, actual: String },
/// `start_or_steer_turn` or `steer_turn` found an active turn that does not accept steering.
ActiveTurnNotSteerable { turn_kind: NonSteerableTurnKind },
/// `start_or_steer_turn` or `steer_turn` required a final output schema
/// that differs from the active turn's schema.
ActiveTurnOutputSchemaMismatch,
/// `start_or_steer_turn` or `steer_turn` reached a steering path with empty user input.
EmptyInput,
}