diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 6d9be4ec02..c54bb99027 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4071,6 +4071,10 @@ ], "description": "Override the reasoning effort for this turn and subsequent turns." }, + "goal": { + "description": "Replace the thread's active goal with an objective derived from this turn's text input.", + "type": "boolean" + }, "input": { "items": { "$ref": "#/definitions/UserInput" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 0c29f4a097..f855515e77 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -18856,6 +18856,10 @@ ], "description": "Override the reasoning effort for this turn and subsequent turns." }, + "goal": { + "description": "Replace the thread's active goal with an objective derived from this turn's text input.", + "type": "boolean" + }, "input": { "items": { "$ref": "#/definitions/v2/UserInput" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index b378a54d16..810dc376b7 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -16673,6 +16673,10 @@ ], "description": "Override the reasoning effort for this turn and subsequent turns." }, + "goal": { + "description": "Replace the thread's active goal with an objective derived from this turn's text input.", + "type": "boolean" + }, "input": { "items": { "$ref": "#/definitions/UserInput" diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json index 070944a28b..4c5ad6957f 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json @@ -533,6 +533,10 @@ ], "description": "Override the reasoning effort for this turn and subsequent turns." }, + "goal": { + "description": "Replace the thread's active goal with an objective derived from this turn's text input.", + "type": "boolean" + }, "input": { "items": { "$ref": "#/definitions/UserInput" diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/TurnStartParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/TurnStartParams.ts index afe1ac6d94..4eadbdb331 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/TurnStartParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/TurnStartParams.ts @@ -11,6 +11,9 @@ import type { SandboxPolicy } from "./SandboxPolicy"; import type { UserInput } from "./UserInput"; export type TurnStartParams = {threadId: string, clientUserMessageId?: string | null, input: Array, /** + * Replace the thread's active goal with an objective derived from this turn's text input. + */ +goal?: boolean, /** * Override the working directory for this turn and subsequent turns. */ cwd?: string | null, /** diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 4a683be990..ea3aec9e71 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -3602,7 +3602,10 @@ fn turn_start_params_preserve_explicit_null_service_tier() { "serviceTier": null })) .expect("params should deserialize"); - assert_eq!(params.service_tier, Some(None)); + assert_eq!( + (params.service_tier.clone(), params.goal), + (Some(None), false) + ); let serialized = serde_json::to_value(¶ms).expect("params should serialize"); assert_eq!( @@ -3614,6 +3617,7 @@ fn turn_start_params_preserve_explicit_null_service_tier() { thread_id: "thread_123".to_string(), client_user_message_id: None, input: vec![], + goal: false, responsesapi_client_metadata: None, additional_context: None, environments: None, @@ -3633,7 +3637,13 @@ fn turn_start_params_preserve_explicit_null_service_tier() { }; let serialized_without_override = serde_json::to_value(&without_override).expect("params should serialize"); - assert_eq!(serialized_without_override.get("serviceTier"), None); + assert_eq!( + ( + serialized_without_override.get("serviceTier"), + serialized_without_override.get("goal"), + ), + (None, None) + ); } #[test] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs index 3bac7bf7ec..e61ca5f83f 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs @@ -68,6 +68,9 @@ pub struct TurnStartParams { #[ts(optional = nullable)] pub client_user_message_id: Option, pub input: Vec, + /// Replace the thread's active goal with an objective derived from this turn's text input. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub goal: bool, /// Optional turn-scoped Responses API client metadata. #[experimental("turn/start.responsesapiClientMetadata")] #[ts(optional = nullable)] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index d918f70b73..e3acce0ed6 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -158,7 +158,7 @@ Example with notification opt-out: - `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream. - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted. - `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success. -- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". +- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Set `goal: true` to replace the persisted thread goal using the ordered, non-empty text inputs as its objective before generation starts; ephemeral threads and input without text reject goal turns. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success. - `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Review and manual compaction turns reject `turn/steer`. - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`. @@ -659,6 +659,8 @@ Turns attach user input (text or images) to a thread and trigger Codex generatio You can optionally specify config overrides on the new turn. If specified, these settings become the default for subsequent turns on the same thread. `outputSchema` applies only to the current turn. Experimental `environments` is turn-scoped: omit it to inherit the thread's sticky environments, pass `[]` to run the turn with no environments, or pass explicit environment ids to override the sticky selection for this turn only. +Set `goal` to `true` to run the request as a goal. The server trims each non-empty text input, joins them in order with a blank line to form the objective, replaces any existing goal, and preserves the original input for the turn. Goal turns require an idle persisted thread, the goals feature to be enabled, and a collaboration mode other than plan mode. + `approvalsReviewer` accepts: - `"user"` — default. Review approval requests directly in the client. diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index 771b1fd128..53f8512e3b 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -658,6 +658,7 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> { text: "hello".to_string(), text_elements: Vec::new(), }], + goal: false, responsesapi_client_metadata: None, additional_context: None, cwd: None, diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 57205798b2..8817619b4e 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -362,6 +362,7 @@ use codex_models_manager::collaboration_mode_presets::builtin_collaboration_mode use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ForcedLoginMethod; +use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::config_types::TrustLevel; @@ -386,6 +387,7 @@ use codex_protocol::protocol::ConversationTextParams; use codex_protocol::protocol::EventMsg; #[cfg(test)] use codex_protocol::protocol::GitInfo as CoreGitInfo; +use codex_protocol::protocol::InitialGoal; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::McpAuthStatus as CoreMcpAuthStatus; use codex_protocol::protocol::Op; @@ -402,6 +404,7 @@ use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::W3cTraceContext; +use codex_protocol::protocol::validate_thread_goal_objective; use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use codex_protocol::user_input::UserInput as CoreInputItem; use codex_rmcp_client::perform_oauth_login_return_url; diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index ce98a2e8e0..bc17413289 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -394,6 +394,47 @@ impl TurnRequestProcessor { self.track_error_response(&request_id, error, /*error_type*/ None); })?; + let initial_goal = if params.goal { + if !self.config.features.enabled(Feature::Goals) { + return Err(invalid_request("goals feature is disabled")); + } + if thread.rollout_path().is_none() { + return Err(invalid_request(format!( + "ephemeral thread does not support goals: {thread_id}" + ))); + } + if matches!(thread.agent_status().await, AgentStatus::Running) { + return Err(invalid_request( + "cannot start a goal while another turn is active", + )); + } + let goal_mode = params + .collaboration_mode + .as_ref() + .map(|mode| mode.mode) + .unwrap_or(thread.config_snapshot().await.collaboration_mode.mode); + if goal_mode == ModeKind::Plan { + return Err(invalid_request("goal turns do not support plan mode")); + } + let objective = params + .input + .iter() + .filter_map(|item| match item { + V2UserInput::Text { text, .. } => Some(text.trim()), + V2UserInput::Image { .. } + | V2UserInput::LocalImage { .. } + | V2UserInput::Skill { .. } + | V2UserInput::Mention { .. } => None, + }) + .filter(|text| !text.is_empty()) + .collect::>() + .join("\n\n"); + validate_thread_goal_objective(&objective).map_err(invalid_request)?; + Some(InitialGoal { objective }) + } else { + None + }; + let environment_selections = self.parse_environment_selections(params.environments)?; // Map v2 input items to core input items. @@ -436,15 +477,20 @@ impl TurnRequestProcessor { additional_context, thread_settings, }; + let trace = self.request_trace_context(&request_id).await; let turn_id = thread .submit_user_input_with_client_user_message_id( turn_op, - self.request_trace_context(&request_id).await, + trace, client_user_message_id, + initial_goal, ) .await .map_err(|err| { - let error = internal_error(format!("failed to start turn: {err}")); + let error = match err { + CodexErr::InvalidRequest(message) => invalid_request(message), + err => internal_error(format!("failed to start turn: {err}")), + }; self.track_error_response(&request_id, &error, /*error_type*/ None); error })?; diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index aed3c58697..086d3564de 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -39,10 +39,13 @@ use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PatchApplyStatus; use codex_app_server_protocol::PatchChangeKind; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequestResolvedNotification; use codex_app_server_protocol::TextElement; +use codex_app_server_protocol::ThreadGoalStatus; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadSettingsUpdatedNotification; use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; @@ -60,6 +63,7 @@ use codex_core::personality_migration::PERSONALITY_MIGRATION_FILENAME; use codex_core::test_support::all_model_presets; use codex_features::FEATURES; use codex_features::Feature; +use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Personality; @@ -69,6 +73,7 @@ use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_DANGER_FULL_ACCESS; use codex_protocol::models::ImageDetail; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; +use codex_state::StateRuntime; use core_test_support::responses; use core_test_support::skip_if_no_network; use pretty_assertions::assert_eq; @@ -77,6 +82,7 @@ use serde_json::json; use std::collections::BTreeMap; use std::collections::HashMap; use std::path::Path; +use std::time::Duration; use tempfile::TempDir; use tokio::time::timeout; use wiremock::ResponseTemplate; @@ -307,6 +313,589 @@ async fn turn_start_with_empty_input_runs_model_request() -> Result<()> { Ok(()) } +#[tokio::test] +async fn turn_start_goal_replaces_existing_goal_and_continues() -> Result<()> { + let server = responses::start_mock_server().await; + let response_mock = responses::mount_sse_sequence( + &server, + vec![ + responses::sse(vec![ + responses::ev_assistant_message("msg-1", "Initial pass complete."), + responses::ev_completed("resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("resp-2"), + responses::ev_function_call( + "call-complete-goal", + "update_goal", + r#"{"status":"complete"}"#, + ), + responses::ev_completed("resp-2"), + ]), + responses::sse(vec![ + responses::ev_assistant_message("msg-2", "Goal complete."), + responses::ev_completed("resp-3"), + ]), + responses::sse(vec![ + responses::ev_assistant_message("msg-3", "Replacement initial pass complete."), + responses::ev_completed("resp-4"), + ]), + responses::sse(vec![ + responses::ev_response_created("resp-5"), + responses::ev_function_call( + "call-complete-replacement-goal", + "update_goal", + r#"{"status":"complete"}"#, + ), + responses::ev_completed("resp-5"), + ]), + responses::sse(vec![ + responses::ev_assistant_message("msg-4", "Replacement goal complete."), + responses::ev_completed("resp-6"), + ]), + ], + ) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + "never", + &BTreeMap::from([(Feature::Goals, true)]), + )?; + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_resp)?; + let goal_cwd = codex_home.path().join("goal-workspace"); + std::fs::create_dir_all(&goal_cwd)?; + let output_schema = json!({ + "type": "object", + "properties": {"summary": {"type": "string"}}, + "required": ["summary"], + "additionalProperties": false, + }); + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + client_user_message_id: None, + input: vec![ + V2UserInput::Text { + text: " Improve benchmark coverage ".to_string(), + text_elements: Vec::new(), + }, + V2UserInput::Text { + text: "Document the results".to_string(), + text_elements: Vec::new(), + }, + ], + goal: true, + cwd: Some(goal_cwd.clone()), + approval_policy: Some(codex_app_server_protocol::AskForApproval::Never), + permissions: Some(BUILT_IN_PERMISSION_PROFILE_DANGER_FULL_ACCESS.to_string()), + model: Some("mock-model".to_string()), + effort: Some(ReasoningEffort::High), + summary: Some(ReasoningSummary::Detailed), + output_schema: Some(output_schema.clone()), + ..Default::default() + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response(turn_resp)?; + + let notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/goal/updated"), + ) + .await??; + let notification: ServerNotification = notification.try_into()?; + let ServerNotification::ThreadGoalUpdated(notification) = notification else { + anyhow::bail!("expected thread goal update notification"); + }; + assert_eq!( + ( + notification.thread_id, + notification.turn_id, + notification.goal.objective, + notification.goal.status, + notification.goal.token_budget, + notification.goal.tokens_used, + notification.goal.time_used_seconds, + ), + ( + thread.id.clone(), + Some(turn.id.clone()), + "Improve benchmark coverage\n\nDocument the results".to_string(), + ThreadGoalStatus::Active, + None, + 0, + 0, + ) + ); + + let settings_notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/settings/updated"), + ) + .await??; + let settings_notification: ThreadSettingsUpdatedNotification = serde_json::from_value( + settings_notification + .params + .expect("thread/settings/updated params"), + )?; + assert_eq!( + ( + settings_notification.thread_id, + settings_notification.thread_settings.cwd, + settings_notification.thread_settings.approval_policy, + settings_notification.thread_settings.sandbox_policy, + settings_notification.thread_settings.model, + settings_notification.thread_settings.effort, + settings_notification.thread_settings.summary, + ), + ( + thread.id.clone(), + goal_cwd.try_into()?, + codex_app_server_protocol::AskForApproval::Never, + codex_app_server_protocol::SandboxPolicy::DangerFullAccess, + "mock-model".to_string(), + Some(ReasoningEffort::High), + Some(ReasoningSummary::Detailed), + ) + ); + + let started: TurnStartedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await?? + .params + .expect("turn/started params"), + )?; + let completed: TurnCompletedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await?? + .params + .expect("turn/completed params"), + )?; + assert_eq!( + ( + started.thread_id, + started.turn.id, + completed.thread_id, + completed.turn.id, + ), + ( + thread.id.clone(), + turn.id.clone(), + thread.id.clone(), + turn.id.clone(), + ) + ); + + let continuation_started: TurnStartedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await?? + .params + .expect("continuation turn/started params"), + )?; + let continuation_completed: TurnCompletedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await?? + .params + .expect("continuation turn/completed params"), + )?; + let continuation_turn_id = continuation_started.turn.id.clone(); + assert_eq!( + ( + continuation_started.thread_id, + continuation_completed.thread_id, + continuation_completed.turn.id, + continuation_turn_id != turn.id, + ), + ( + thread.id.clone(), + thread.id.clone(), + continuation_turn_id, + true, + ) + ); + + let requests = response_mock.requests(); + let initial_request = requests.first().expect("initial model request"); + let initial_body = initial_request.body_json(); + assert_eq!( + ( + initial_request.message_input_texts("user"), + initial_body.get("model"), + initial_body.pointer("/reasoning/effort"), + initial_body.pointer("/reasoning/summary"), + initial_body.pointer("/text/format/schema"), + requests + .get(1) + .expect("continuation model request") + .message_input_texts("user") + .into_iter() + .any(|text| text.contains("Continue working toward the active thread goal.")), + ), + ( + vec![ + " Improve benchmark coverage ".to_string(), + "Document the results".to_string(), + ], + Some(&json!("mock-model")), + Some(&json!("high")), + Some(&json!("detailed")), + Some(&output_schema), + true, + ) + ); + + let completed_goal_notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/goal/updated"), + ) + .await??; + let completed_goal_notification: ServerNotification = completed_goal_notification.try_into()?; + let ServerNotification::ThreadGoalUpdated(completed_goal_notification) = + completed_goal_notification + else { + anyhow::bail!("expected completed thread goal update notification"); + }; + let state_db = + StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".to_string()).await?; + let persisted_thread_id = ThreadId::from_string(&thread.id)?; + let first_goal = state_db + .thread_goals() + .get_thread_goal(persisted_thread_id) + .await? + .expect("first goal should be persisted"); + assert_eq!( + ( + completed_goal_notification.goal.status, + first_goal.objective.clone(), + first_goal.status, + ), + ( + ThreadGoalStatus::Complete, + "Improve benchmark coverage\n\nDocument the results".to_string(), + codex_state::ThreadGoalStatus::Complete, + ) + ); + + let replacement_turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + client_user_message_id: None, + input: vec![V2UserInput::Text { + text: "Publish the benchmark report".to_string(), + text_elements: Vec::new(), + }], + goal: true, + ..Default::default() + }) + .await?; + let replacement_turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(replacement_turn_req)), + ) + .await??; + let TurnStartResponse { + turn: replacement_turn, + } = to_response(replacement_turn_resp)?; + let replacement_goal_notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/goal/updated"), + ) + .await??; + let replacement_goal_notification: ServerNotification = + replacement_goal_notification.try_into()?; + let ServerNotification::ThreadGoalUpdated(replacement_goal_notification) = + replacement_goal_notification + else { + anyhow::bail!("expected replacement thread goal update notification"); + }; + let replacement_goal = state_db + .thread_goals() + .get_thread_goal(persisted_thread_id) + .await? + .expect("replacement goal should be persisted"); + assert_eq!( + ( + replacement_goal_notification.thread_id, + replacement_goal_notification.turn_id, + replacement_goal_notification.goal.objective, + replacement_goal_notification.goal.status, + replacement_goal.goal_id != first_goal.goal_id, + replacement_goal.tokens_used, + replacement_goal.time_used_seconds, + ), + ( + thread.id.clone(), + Some(replacement_turn.id.clone()), + "Publish the benchmark report".to_string(), + ThreadGoalStatus::Active, + true, + 0, + 0, + ) + ); + + let mut replacement_lifecycles = Vec::new(); + for _ in 0..2 { + let started: TurnStartedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await?? + .params + .expect("replacement turn/started params"), + )?; + let completed: TurnCompletedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await?? + .params + .expect("replacement turn/completed params"), + )?; + replacement_lifecycles.push((started.turn.id, completed.turn.id)); + } + assert_eq!( + ( + replacement_lifecycles[0].clone(), + replacement_lifecycles[1].0 != replacement_turn.id, + replacement_lifecycles[1].0 == replacement_lifecycles[1].1, + ), + ( + (replacement_turn.id.clone(), replacement_turn.id), + true, + true, + ) + ); + + Ok(()) +} + +#[tokio::test] +async fn turn_start_goal_validates_input_and_thread_persistence() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + "never", + &BTreeMap::from([(Feature::Goals, true)]), + )?; + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let persisted_thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let persisted_thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(persisted_thread_req)), + ) + .await??; + let ThreadStartResponse { + thread: persisted_thread, + .. + } = to_response(persisted_thread_resp)?; + let empty_goal_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: persisted_thread.id, + client_user_message_id: None, + input: Vec::new(), + goal: true, + ..Default::default() + }) + .await?; + let empty_goal_error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(empty_goal_req)), + ) + .await??; + + let ephemeral_thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ephemeral: Some(true), + ..Default::default() + }) + .await?; + let ephemeral_thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(ephemeral_thread_req)), + ) + .await??; + let ThreadStartResponse { + thread: ephemeral_thread, + .. + } = to_response(ephemeral_thread_resp)?; + + let ephemeral_goal_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: ephemeral_thread.id.clone(), + client_user_message_id: None, + input: vec![V2UserInput::Text { + text: "Improve benchmark coverage".to_string(), + text_elements: Vec::new(), + }], + goal: true, + ..Default::default() + }) + .await?; + let ephemeral_goal_error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(ephemeral_goal_req)), + ) + .await??; + assert_eq!( + ( + empty_goal_error.error.message, + ephemeral_goal_error.error.message, + ), + ( + "goal objective must not be empty".to_string(), + format!( + "ephemeral thread does not support goals: {}", + ephemeral_thread.id + ), + ) + ); + + Ok(()) +} + +#[tokio::test] +async fn concurrent_goal_starts_reject_the_loser_without_starting_a_turn() -> Result<()> { + let server = responses::start_mock_server().await; + responses::mount_response_once( + &server, + responses::sse_response(responses::sse(vec![ + responses::ev_assistant_message("msg-1", "Working."), + responses::ev_completed("resp-1"), + ])) + .set_delay(Duration::from_secs(1)), + ) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + "never", + &BTreeMap::from([(Feature::Goals, true)]), + )?; + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_resp)?; + + let first_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "First goal".to_string(), + text_elements: Vec::new(), + }], + goal: true, + ..Default::default() + }) + .await?; + let second_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "Second goal".to_string(), + text_elements: Vec::new(), + }], + goal: true, + ..Default::default() + }) + .await?; + + let first_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(first_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response(first_response)?; + let second_error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(second_req)), + ) + .await??; + let started: TurnStartedNotification = serde_json::from_value( + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await?? + .params + .expect("turn/started params"), + )?; + + assert_eq!( + ( + second_error.error.message, + started.thread_id, + started.turn.id, + ), + ( + "cannot start a goal while another turn is active".to_string(), + thread.id, + turn.id, + ) + ); + + Ok(()) +} + #[tokio::test] async fn turn_start_additional_context_flows_to_model_input() -> Result<()> { let responses = vec![create_final_assistant_message_sse_response("Done")?]; @@ -2327,6 +2916,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> { text: "first turn".to_string(), text_elements: Vec::new(), }], + goal: false, responsesapi_client_metadata: None, additional_context: None, cwd: Some(first_cwd.clone()), @@ -2371,6 +2961,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> { text: "second turn".to_string(), text_elements: Vec::new(), }], + goal: false, responsesapi_client_metadata: None, additional_context: None, cwd: Some(second_cwd.clone()), diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index 5a1473b057..b3e579b49e 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -225,6 +225,7 @@ pub(crate) async fn run_codex_thread_one_shot( id: "shutdown".to_string(), op: Op::Shutdown {}, client_user_message_id: None, + initial_goal: None, trace: None, }) .await; diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 7dabb09d33..01ecd80262 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -131,6 +131,7 @@ async fn forward_ops_preserves_submission_trace_context() { id: "sub-1".to_string(), op: Op::Interrupt, client_user_message_id: None, + initial_goal: None, trace: Some(codex_protocol::protocol::W3cTraceContext { traceparent: Some( "00-1234567890abcdef1234567890abcdef-1234567890abcdef-01".to_string(), diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index b3f9cb0e2e..ce9157b500 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -22,6 +22,7 @@ use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AdditionalContextEntry; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::Event; +use codex_protocol::protocol::InitialGoal; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::SandboxPolicy; @@ -233,9 +234,15 @@ impl CodexThread { op: Op, trace: Option, client_user_message_id: Option, + initial_goal: Option, ) -> CodexResult { self.codex - .submit_user_input_with_client_user_message_id(op, trace, client_user_message_id) + .submit_user_input_with_client_user_message_id( + op, + trace, + client_user_message_id, + initial_goal, + ) .await } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 7730a30ad7..bca44d1f2a 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -4,6 +4,7 @@ use crate::realtime_conversation::handle_start as handle_realtime_conversation_s use crate::realtime_conversation::handle_text as handle_realtime_conversation_text; use async_channel::Receiver; use codex_otel::set_parent_from_w3c_trace_context; +use codex_protocol::error::CodexErr; use codex_protocol::protocol::Submission; use tracing::Instrument; use tracing::debug_span; @@ -14,6 +15,7 @@ use crate::session::TurnInput; use crate::session::session::Session; use crate::session::session::SessionSettingsUpdate; +use super::initial_goal::InitialGoalStartError; use crate::config::Config; use crate::realtime_context::REALTIME_TURN_TOKEN_BUDGET; use crate::realtime_context::truncate_realtime_text_to_token_budget; @@ -21,6 +23,8 @@ use crate::realtime_conversation::REALTIME_USER_TEXT_PREFIX; use crate::realtime_conversation::prefix_realtime_v2_text; use crate::review_prompts::resolve_review_request; use crate::session::spawn_review_thread; +use crate::state::ActiveTurn; +use crate::state::TurnState; use crate::tasks::CompactTask; use crate::tasks::UserShellCommandMode; use crate::tasks::UserShellCommandTask; @@ -34,6 +38,7 @@ use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::GuardianAssessmentStatus; +use codex_protocol::protocol::InitialGoal; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::McpServerRefreshConfig; use codex_protocol::protocol::Op; @@ -61,6 +66,7 @@ use codex_rmcp_client::ElicitationAction; use codex_rmcp_client::ElicitationResponse; use serde_json::Value; use std::sync::Arc; +use tokio::sync::Mutex; use tracing::debug; use tracing::info; use tracing::warn; @@ -90,6 +96,7 @@ pub async fn user_input_or_turn( sub_id: String, op: Op, client_user_message_id: Option, + initial_goal: Option, ) { user_input_or_turn_inner( sess, @@ -97,10 +104,28 @@ pub async fn user_input_or_turn( op, /*mirror_user_text_to_realtime*/ Some(()), client_user_message_id, + initial_goal, ) .await; } +async fn clear_reserved_goal_turn(sess: &Session, turn_state: &Arc>) { + let cleared = { + let mut active_turn = sess.active_turn.lock().await; + if active_turn.as_ref().is_some_and(|active_turn| { + active_turn.task.is_none() && Arc::ptr_eq(&active_turn.turn_state, turn_state) + }) { + *active_turn = None; + true + } else { + false + } + }; + if cleared { + sess.emit_thread_idle_lifecycle_if_idle().await; + } +} + pub async fn update_thread_settings( sess: &Arc, sub_id: String, @@ -197,6 +222,7 @@ pub(super) async fn user_input_or_turn_inner( op: Op, mirror_user_text_to_realtime: Option<()>, client_user_message_id: Option, + initial_goal: Option, ) { let Op::UserInput { items, @@ -218,9 +244,67 @@ pub(super) async fn user_input_or_turn_inner( updates.final_output_json_schema = Some(final_output_json_schema); updates.environments = environments; - let Ok(current_context) = sess.new_turn_with_sub_id(sub_id.clone(), updates).await else { - // new_turn_with_sub_id already emits the error event. - return; + let mut reserved_goal_turn = None; + if initial_goal.is_some() { + let turn_state = { + let mut active_turn = sess.active_turn.lock().await; + if active_turn.is_some() { + None + } else { + let active_turn = active_turn.get_or_insert_with(ActiveTurn::default); + Some(Arc::clone(&active_turn.turn_state)) + } + }; + let Some(turn_state) = turn_state else { + sess.complete_initial_goal_start( + &sub_id, + Err(InitialGoalStartError::InvalidRequest( + "cannot start a goal while another turn is active".to_string(), + )), + ); + return; + }; + reserved_goal_turn = Some(turn_state); + } + + let current_context = if let Some(initial_goal) = initial_goal { + let prepared_turn = match sess.prepare_turn(updates).await { + Ok(prepared_turn) => prepared_turn, + Err(err) => { + if let Some(turn_state) = reserved_goal_turn.as_ref() { + clear_reserved_goal_turn(sess, turn_state).await; + } + let err = match err { + CodexErr::InvalidRequest(message) => { + InitialGoalStartError::InvalidRequest(message) + } + err => InitialGoalStartError::Internal(err.to_string()), + }; + sess.complete_initial_goal_start(&sub_id, Err(err)); + return; + } + }; + match sess + .prepare_initial_goal(&sub_id, &prepared_turn, &initial_goal) + .await + { + Ok(()) => {} + Err(err) => { + if let Some(turn_state) = reserved_goal_turn.as_ref() { + clear_reserved_goal_turn(sess, turn_state).await; + } + sess.complete_initial_goal_start(&sub_id, Err(err)); + return; + } + } + + sess.commit_prepared_turn(sub_id.clone(), prepared_turn) + .await + } else { + match sess.new_turn_with_sub_id(sub_id.clone(), updates).await { + Ok(current_context) => current_context, + Err(_) => return, + } }; if emit_thread_settings_applied { sess.send_event_raw(Event { @@ -241,6 +325,18 @@ pub(super) async fn user_input_or_turn_inner( ) .await { + Ok(_) if reserved_goal_turn.is_some() => { + if let Some(turn_state) = reserved_goal_turn.as_ref() { + clear_reserved_goal_turn(sess, turn_state).await; + } + sess.complete_initial_goal_start( + &sub_id, + Err(InitialGoalStartError::Internal( + "goal turn reservation was replaced before startup".to_string(), + )), + ); + return; + } Ok(_) => { current_context.session_telemetry.user_prompt(&items); Some(items) @@ -273,15 +369,35 @@ pub(super) async fn user_input_or_turn_inner( client_id: client_user_message_id, }); } - sess.spawn_task( - Arc::clone(¤t_context), - task_input, - crate::tasks::RegularTask::new(), - ) - .await; + if reserved_goal_turn.is_some() { + sess.start_task( + Arc::clone(¤t_context), + task_input, + crate::tasks::RegularTask::new(), + ) + .await; + sess.complete_initial_goal_start(&sub_id, Ok(())); + } else { + sess.spawn_task( + Arc::clone(¤t_context), + task_input, + crate::tasks::RegularTask::new(), + ) + .await; + } Some(accepted_items) } Err(err) => { + if let Some(turn_state) = reserved_goal_turn.as_ref() { + clear_reserved_goal_turn(sess, turn_state).await; + sess.complete_initial_goal_start( + &sub_id, + Err(InitialGoalStartError::Internal( + err.to_error_event().message, + )), + ); + return; + } sess.send_event_raw(Event { id: sub_id, msg: EventMsg::Error(err.to_error_event()), @@ -787,8 +903,14 @@ pub(super) async fn submission_loop( false } Op::UserInput { .. } => { - user_input_or_turn(&sess, sub.id.clone(), sub.op, sub.client_user_message_id) - .await; + user_input_or_turn( + &sess, + sub.id.clone(), + sub.op, + sub.client_user_message_id, + sub.initial_goal, + ) + .await; false } Op::ThreadSettings { thread_settings } => { diff --git a/codex-rs/core/src/session/initial_goal.rs b/codex-rs/core/src/session/initial_goal.rs new file mode 100644 index 0000000000..1a7ada8b89 --- /dev/null +++ b/codex-rs/core/src/session/initial_goal.rs @@ -0,0 +1,106 @@ +use std::collections::HashMap; +use std::sync::Mutex; +use std::sync::PoisonError; + +use codex_extension_api::InitialGoalError; +use codex_extension_api::InitialGoalInput; +use codex_protocol::protocol::InitialGoal; +use tokio::sync::oneshot; + +use super::session::Session; +use super::turn_context::PreparedTurn; + +pub(crate) enum InitialGoalStartError { + InvalidRequest(String), + Internal(String), +} + +#[derive(Default)] +pub(crate) struct InitialGoalStartAcks { + senders: Mutex>>>, +} + +impl InitialGoalStartAcks { + pub(crate) fn register( + &self, + turn_id: String, + ) -> oneshot::Receiver> { + let (sender, receiver) = oneshot::channel(); + self.senders + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(turn_id, sender); + receiver + } + + pub(crate) fn complete(&self, turn_id: &str, result: Result<(), InitialGoalStartError>) { + if let Some(sender) = self + .senders + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(turn_id) + { + let _ = sender.send(result); + } + } + + pub(crate) fn cancel(&self, turn_id: &str) { + self.senders + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(turn_id); + } +} + +impl Session { + pub(crate) async fn prepare_initial_goal( + &self, + turn_id: &str, + prepared_turn: &PreparedTurn, + goal: &InitialGoal, + ) -> Result<(), InitialGoalStartError> { + let contributor = self + .services + .extensions + .initial_goal_contributor() + .ok_or_else(|| { + InitialGoalStartError::Internal( + "goal extension is unavailable for this thread".to_string(), + ) + })?; + contributor + .replace_for_turn(InitialGoalInput { + turn_id, + goal, + collaboration_mode: prepared_turn.collaboration_mode(), + session_store: &self.services.session_extension_data, + thread_store: &self.services.thread_extension_data, + }) + .await + .map_err(|err| match err { + InitialGoalError::InvalidRequest(message) => { + InitialGoalStartError::InvalidRequest(message) + } + InitialGoalError::Internal(message) => InitialGoalStartError::Internal(message), + }) + } + + pub(crate) fn register_initial_goal_start_ack( + &self, + turn_id: String, + ) -> oneshot::Receiver> { + self.initial_goal_start_acks.register(turn_id) + } + + pub(crate) fn complete_initial_goal_start( + &self, + turn_id: &str, + result: Result<(), InitialGoalStartError>, + ) { + self.initial_goal_start_acks.complete(turn_id, result); + } + + pub(crate) fn cancel_initial_goal_start(&self, turn_id: &str) { + self.initial_goal_start_acks.cancel(turn_id); + } +} diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 37d55256f5..9c309eb63d 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -9,6 +9,7 @@ use std::sync::atomic::AtomicU64; use std::time::SystemTime; use std::time::UNIX_EPOCH; +use self::initial_goal::InitialGoalStartError; use crate::agent::AgentControl; use crate::agent::AgentStatus; use crate::agent::agent_status_from_event; @@ -196,6 +197,7 @@ use codex_protocol::exec_output::StreamOutput; mod config_lock; mod handlers; +mod initial_goal; mod inject; mod input_queue; mod mcp; @@ -337,6 +339,7 @@ use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecApprovalRequestEvent; +use codex_protocol::protocol::InitialGoal; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::McpServerRefreshConfig; use codex_protocol::protocol::ModelRerouteEvent; @@ -380,6 +383,17 @@ pub struct Codex { pub(crate) session_loop_termination: SessionLoopTermination, } +struct InitialGoalStartAckGuard { + session: Arc, + turn_id: String, +} + +impl Drop for InitialGoalStartAckGuard { + fn drop(&mut self) { + self.session.cancel_initial_goal_start(&self.turn_id); + } +} + pub(crate) type SessionLoopTermination = Shared>; /// Wrapper returned by [`Codex::spawn`] containing the spawned [`Codex`] and @@ -687,6 +701,7 @@ impl Codex { id: id.clone(), op, client_user_message_id: None, + initial_goal: None, trace, }; self.submit_with_id(sub).await?; @@ -698,16 +713,43 @@ impl Codex { op: Op, trace: Option, client_user_message_id: Option, + initial_goal: Option, ) -> CodexResult { debug_assert!(matches!(op, Op::UserInput { .. })); let id = Uuid::now_v7().to_string(); + let mut _goal_start_ack_guard = None; + let goal_start_ack = if initial_goal.is_some() { + let receiver = self.session.register_initial_goal_start_ack(id.clone()); + _goal_start_ack_guard = Some(InitialGoalStartAckGuard { + session: Arc::clone(&self.session), + turn_id: id.clone(), + }); + Some(receiver) + } else { + None + }; let sub = Submission { id: id.clone(), op, client_user_message_id, + initial_goal, trace, }; self.submit_with_id(sub).await?; + if let Some(goal_start_ack) = goal_start_ack { + let goal_start_result = tokio::select! { + result = goal_start_ack => { + result.map_err(|_| CodexErr::InternalAgentDied)? + } + () = self.session_loop_termination.clone() => { + return Err(CodexErr::InternalAgentDied); + } + }; + goal_start_result.map_err(|err| match err { + InitialGoalStartError::InvalidRequest(message) => CodexErr::InvalidRequest(message), + InitialGoalStartError::Internal(message) => CodexErr::Fatal(message), + })?; + } Ok(id) } @@ -1126,6 +1168,7 @@ impl Session { }, /*mirror_user_text_to_realtime*/ None, /*client_user_message_id*/ None, + /*initial_goal*/ None, ) .await; } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 122ff6aeb9..5020c4bd56 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1,3 +1,4 @@ +use super::initial_goal::InitialGoalStartAcks; use super::input_queue::InputQueue; use super::*; use crate::agents_md::LoadedAgentsMd; @@ -36,6 +37,7 @@ pub(crate) struct Session { pub(crate) conversation: Arc, pub(crate) active_turn: Mutex>, pub(crate) input_queue: InputQueue, + pub(crate) initial_goal_start_acks: InitialGoalStartAcks, pub(crate) guardian_review_session: GuardianReviewSessionManager, pub(crate) services: SessionServices, pub(super) next_internal_sub_id: AtomicU64, @@ -1057,6 +1059,7 @@ impl Session { conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: InputQueue::new(), + initial_goal_start_acks: InitialGoalStartAcks::default(), guardian_review_session: GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 366cf6f22a..4ef13dc3ba 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -4899,6 +4899,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: super::input_queue::InputQueue::new(), + initial_goal_start_acks: super::initial_goal::InitialGoalStartAcks::default(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), @@ -5825,6 +5826,7 @@ async fn submit_with_id_captures_current_span_trace_context() { id: "sub-1".into(), op: Op::Interrupt, client_user_message_id: None, + initial_goal: None, trace: None, }) .await @@ -5897,6 +5899,7 @@ fn submission_dispatch_span_prefers_submission_trace_context() { id: "sub-1".into(), op: Op::Interrupt, client_user_message_id: None, + initial_goal: None, trace: Some(submission_trace), }) }); @@ -5924,6 +5927,7 @@ fn submission_dispatch_span_uses_debug_for_realtime_audio() { }, }), client_user_message_id: None, + initial_goal: None, trace: None, }); @@ -5992,6 +5996,7 @@ async fn user_turn_updates_approvals_reviewer() { }, }, /*client_user_message_id*/ None, + /*initial_goal*/ None, ) .await; @@ -6286,6 +6291,7 @@ async fn spawn_task_turn_span_inherits_dispatch_trace_context() { id: "sub-1".into(), op: Op::Interrupt, client_user_message_id: None, + initial_goal: None, trace: Some(submission_trace.clone()), }); let dispatch_span_id = dispatch_span.context().span().span_context().span_id(); @@ -6976,6 +6982,7 @@ where conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: super::input_queue::InputQueue::new(), + initial_goal_start_acks: super::initial_goal::InitialGoalStartAcks::default(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 5f52912d59..7bf666c405 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -111,6 +111,24 @@ enum TurnMultiAgentRuntime { Preview, } +pub(crate) struct PreparedTurn { + session_configuration: SessionConfiguration, + turn_environments: ResolvedTurnEnvironments, + permission_profile_changed: bool, + previous_cwd: AbsolutePathBuf, + codex_home: AbsolutePathBuf, + session_source: SessionSource, + previous_config: Option, + new_config: Option, + final_output_json_schema: Option>, +} + +impl PreparedTurn { + pub(crate) fn collaboration_mode(&self) -> &CollaborationMode { + &self.session_configuration.collaboration_mode + } +} + impl TurnContext { pub(crate) fn permission_profile(&self) -> PermissionProfile { self.permission_profile.clone() @@ -582,63 +600,8 @@ impl Session { sub_id: String, updates: SessionSettingsUpdate, ) -> CodexResult> { - let notify_config_contributors = !self.services.extensions.config_contributors().is_empty(); - let update_result: CodexResult<_> = { - let mut state = self.state.lock().await; - match state.session_configuration.clone().apply(&updates) { - Ok(next) => { - let mut effective_environments = updates - .environments - .clone() - .unwrap_or_else(|| next.environments.clone()); - if updates.environments.is_none() { - Self::overlay_runtime_cwd_on_primary_environment( - &mut effective_environments, - &next.cwd, - ); - } - let turn_environments = - self.resolve_turn_environments(&effective_environments)?; - let previous_cwd = state.session_configuration.cwd.clone(); - let previous_permission_profile = - state.session_configuration.permission_profile(); - let next_permission_profile = next.permission_profile(); - let permission_profile_changed = - previous_permission_profile != next_permission_profile; - let codex_home = next.codex_home.clone(); - let session_source = next.session_source.clone(); - let previous_config = notify_config_contributors.then(|| { - Self::build_effective_session_config(&state.session_configuration) - }); - let new_config = notify_config_contributors - .then(|| Self::build_effective_session_config(&next)); - state.session_configuration = next.clone(); - Ok(( - next, - turn_environments, - permission_profile_changed, - previous_cwd, - codex_home, - session_source, - previous_config, - new_config, - )) - } - Err(err) => Err(CodexErr::InvalidRequest(err.to_string())), - } - }; - - let ( - session_configuration, - turn_environments, - permission_profile_changed, - previous_cwd, - codex_home, - session_source, - previous_config, - new_config, - ) = match update_result { - Ok(update) => update, + let prepared_turn = match self.prepare_turn(updates).await { + Ok(prepared_turn) => prepared_turn, Err(err) => { let message = err.to_string(); self.send_event_raw(Event { @@ -652,6 +615,74 @@ impl Session { return Err(CodexErr::InvalidRequest(message)); } }; + Ok(self.commit_prepared_turn(sub_id, prepared_turn).await) + } + + pub(crate) async fn prepare_turn( + &self, + updates: SessionSettingsUpdate, + ) -> CodexResult { + let notify_config_contributors = !self.services.extensions.config_contributors().is_empty(); + let state = self.state.lock().await; + let session_configuration = state + .session_configuration + .clone() + .apply(&updates) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; + let mut effective_environments = updates + .environments + .clone() + .unwrap_or_else(|| session_configuration.environments.clone()); + if updates.environments.is_none() { + Self::overlay_runtime_cwd_on_primary_environment( + &mut effective_environments, + &session_configuration.cwd, + ); + } + let turn_environments = self.resolve_turn_environments(&effective_environments)?; + let previous_cwd = state.session_configuration.cwd.clone(); + let previous_permission_profile = state.session_configuration.permission_profile(); + let next_permission_profile = session_configuration.permission_profile(); + let permission_profile_changed = previous_permission_profile != next_permission_profile; + let codex_home = session_configuration.codex_home.clone(); + let session_source = session_configuration.session_source.clone(); + let previous_config = notify_config_contributors + .then(|| Self::build_effective_session_config(&state.session_configuration)); + let new_config = notify_config_contributors + .then(|| Self::build_effective_session_config(&session_configuration)); + Ok(PreparedTurn { + session_configuration, + turn_environments, + permission_profile_changed, + previous_cwd, + codex_home, + session_source, + previous_config, + new_config, + final_output_json_schema: updates.final_output_json_schema, + }) + } + + pub(crate) async fn commit_prepared_turn( + &self, + sub_id: String, + prepared_turn: PreparedTurn, + ) -> Arc { + let PreparedTurn { + session_configuration, + turn_environments, + permission_profile_changed, + previous_cwd, + codex_home, + session_source, + previous_config, + new_config, + final_output_json_schema, + } = prepared_turn; + { + let mut state = self.state.lock().await; + state.session_configuration = session_configuration.clone(); + } self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref()); self.maybe_refresh_shell_snapshot_for_cwd( @@ -666,14 +697,13 @@ impl Session { .await; } - Ok(self - .new_turn_from_configuration( - sub_id, - session_configuration, - updates.final_output_json_schema, - turn_environments, - ) - .await) + self.new_turn_from_configuration( + sub_id, + session_configuration, + final_output_json_schema, + turn_environments, + ) + .await } fn resolve_turn_environments( diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index b8b87a85ec..17234b2f1d 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -868,6 +868,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { thread_id: primary_thread_id_for_span.clone(), client_user_message_id: None, input: items.into_iter().map(Into::into).collect(), + goal: false, responsesapi_client_metadata: None, additional_context: None, environments: None, diff --git a/codex-rs/ext/extension-api/src/contributors.rs b/codex-rs/ext/extension-api/src/contributors.rs index 8706e8ee7a..b7b19dabe1 100644 --- a/codex-rs/ext/extension-api/src/contributors.rs +++ b/codex-rs/ext/extension-api/src/contributors.rs @@ -10,12 +10,16 @@ use codex_tools::ToolExecutor; use crate::ExtensionData; +mod initial_goal; mod prompt; mod thread_lifecycle; mod tool_lifecycle; mod turn_input; mod turn_lifecycle; +pub use initial_goal::InitialGoalContributor; +pub use initial_goal::InitialGoalError; +pub use initial_goal::InitialGoalInput; pub use prompt::PromptFragment; pub use prompt::PromptSlot; pub use thread_lifecycle::ThreadIdleInput; diff --git a/codex-rs/ext/extension-api/src/contributors/initial_goal.rs b/codex-rs/ext/extension-api/src/contributors/initial_goal.rs new file mode 100644 index 0000000000..3c9363e1e9 --- /dev/null +++ b/codex-rs/ext/extension-api/src/contributors/initial_goal.rs @@ -0,0 +1,39 @@ +use std::future::Future; +use std::pin::Pin; + +use codex_protocol::config_types::CollaborationMode; +use codex_protocol::protocol::InitialGoal; + +use crate::ExtensionData; + +/// Input supplied before the host commits settings or starts an initial goal turn. +pub struct InitialGoalInput<'a> { + /// Stable host-owned turn identifier. + pub turn_id: &'a str, + /// Goal objective requested for this turn. + pub goal: &'a InitialGoal, + /// Effective collaboration mode prepared for this turn. + pub collaboration_mode: &'a CollaborationMode, + /// Store scoped to the host session runtime. + pub session_store: &'a ExtensionData, + /// Store scoped to this thread runtime. + pub thread_store: &'a ExtensionData, +} + +/// Error returned while preparing an initial goal turn. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum InitialGoalError { + /// The request is invalid and should be reported to the caller. + InvalidRequest(String), + /// Goal persistence or runtime preparation failed internally. + Internal(String), +} + +/// Extension contribution that atomically replaces a goal before a turn starts. +pub trait InitialGoalContributor: Send + Sync { + /// Persist and prepare the requested goal before the host commits the turn. + fn replace_for_turn<'a>( + &'a self, + input: InitialGoalInput<'a>, + ) -> Pin> + Send + 'a>>; +} diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index 7fa60c0fe7..1b0cec8890 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -31,6 +31,9 @@ pub use codex_tools::parse_tool_input_schema_without_compaction; pub use contributors::ApprovalReviewContributor; pub use contributors::ConfigContributor; pub use contributors::ContextContributor; +pub use contributors::InitialGoalContributor; +pub use contributors::InitialGoalError; +pub use contributors::InitialGoalInput; pub use contributors::PromptFragment; pub use contributors::PromptSlot; pub use contributors::ThreadIdleInput; diff --git a/codex-rs/ext/extension-api/src/registry.rs b/codex-rs/ext/extension-api/src/registry.rs index 0849387110..108077ecd7 100644 --- a/codex-rs/ext/extension-api/src/registry.rs +++ b/codex-rs/ext/extension-api/src/registry.rs @@ -7,6 +7,7 @@ use crate::ConfigContributor; use crate::ContextContributor; use crate::ExtensionData; use crate::ExtensionEventSink; +use crate::InitialGoalContributor; use crate::NoopExtensionEventSink; use crate::ThreadLifecycleContributor; use crate::TokenUsageContributor; @@ -19,6 +20,7 @@ use crate::TurnLifecycleContributor; /// Mutable registry used while hosts register typed runtime contributions. pub struct ExtensionRegistryBuilder { event_sink: Arc, + initial_goal_contributor: Option>, thread_lifecycle_contributors: Vec>>, turn_lifecycle_contributors: Vec>, config_contributors: Vec>>, @@ -35,6 +37,7 @@ impl Default for ExtensionRegistryBuilder { fn default() -> Self { Self { event_sink: Arc::new(NoopExtensionEventSink), + initial_goal_contributor: None, thread_lifecycle_contributors: Vec::new(), turn_lifecycle_contributors: Vec::new(), config_contributors: Vec::new(), @@ -81,6 +84,11 @@ impl ExtensionRegistryBuilder { self.thread_lifecycle_contributors.push(contributor); } + /// Registers the extension that owns persisted thread goals. + pub fn initial_goal_contributor(&mut self, contributor: Arc) { + self.initial_goal_contributor = Some(contributor); + } + /// Registers one turn-lifecycle contributor. pub fn turn_lifecycle_contributor(&mut self, contributor: Arc) { self.turn_lifecycle_contributors.push(contributor); @@ -125,6 +133,7 @@ impl ExtensionRegistryBuilder { pub fn build(self) -> ExtensionRegistry { ExtensionRegistry { event_sink: self.event_sink, + initial_goal_contributor: self.initial_goal_contributor, thread_lifecycle_contributors: self.thread_lifecycle_contributors, turn_lifecycle_contributors: self.turn_lifecycle_contributors, config_contributors: self.config_contributors, @@ -142,6 +151,7 @@ impl ExtensionRegistryBuilder { /// Immutable typed registry produced after extensions are installed. pub struct ExtensionRegistry { event_sink: Arc, + initial_goal_contributor: Option>, thread_lifecycle_contributors: Vec>>, turn_lifecycle_contributors: Vec>, config_contributors: Vec>>, @@ -160,6 +170,11 @@ impl ExtensionRegistry { Arc::clone(&self.event_sink) } + /// Returns the extension that owns initial goal replacement, when installed. + pub fn initial_goal_contributor(&self) -> Option<&Arc> { + self.initial_goal_contributor.as_ref() + } + /// Returns the registered thread-lifecycle contributors. pub fn thread_lifecycle_contributors(&self) -> &[Arc>] { &self.thread_lifecycle_contributors diff --git a/codex-rs/ext/goal/src/api.rs b/codex-rs/ext/goal/src/api.rs index 5123e6a5ce..58211f835f 100644 --- a/codex-rs/ext/goal/src/api.rs +++ b/codex-rs/ext/goal/src/api.rs @@ -236,6 +236,71 @@ impl GoalService { }) } + pub(crate) async fn replace_goal_for_turn_start( + &self, + state_db: &codex_state::StateRuntime, + thread_id: ThreadId, + objective: &str, + ) -> Result { + let objective = objective.trim(); + validate_thread_goal_objective(objective).map_err(GoalServiceError::InvalidRequest)?; + + let runtime = self.runtime_for_thread(thread_id); + let _goal_state_permit = match runtime.as_ref() { + Some(runtime) => Some( + runtime + .goal_state_permit() + .await + .map_err(GoalServiceError::Internal)?, + ), + None => None, + }; + let previous_state_goal = state_db + .thread_goals() + .get_thread_goal(thread_id) + .await + .map_err(|err| { + GoalServiceError::Internal(format!("failed to read thread goal: {err}")) + })?; + let previous_goal = previous_state_goal.as_ref().map(PreviousGoalSnapshot::from); + if let Some(runtime) = runtime.as_ref() { + runtime + .prepare_external_goal_mutation() + .await + .map_err(GoalServiceError::Internal)?; + } + + let goal = match state_db + .thread_goals() + .replace_thread_goal( + thread_id, + objective, + codex_state::ThreadGoalStatus::Active, + /*token_budget*/ None, + ) + .await + .map_err(|err| { + GoalServiceError::Internal(format!("failed to replace thread goal: {err}")) + }) { + Ok(goal) => goal, + Err(err) => { + if let Some(runtime) = runtime.as_ref() { + runtime.restore_goal_after_failed_mutation(previous_state_goal.as_ref()); + } + return Err(err); + } + }; + fill_empty_thread_preview_if_possible(state_db, thread_id, &goal).await; + if let Some(runtime) = runtime.as_ref() { + runtime.apply_initial_goal_set(&goal, previous_goal.as_ref()); + } + Ok(GoalSetOutcome { + goal: protocol_goal_from_state(goal.clone()), + state_goal: goal, + previous_goal, + }) + } + pub async fn clear_thread_goal( &self, state_db: &codex_state::StateRuntime, diff --git a/codex-rs/ext/goal/src/extension.rs b/codex-rs/ext/goal/src/extension.rs index 78ae0b5b44..55c8897840 100644 --- a/codex-rs/ext/goal/src/extension.rs +++ b/codex-rs/ext/goal/src/extension.rs @@ -7,6 +7,9 @@ use codex_extension_api::ConfigContributor; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::InitialGoalContributor; +use codex_extension_api::InitialGoalError; +use codex_extension_api::InitialGoalInput; use codex_extension_api::ThreadIdleInput; use codex_extension_api::ThreadLifecycleContributor; use codex_extension_api::ThreadResumeInput; @@ -181,6 +184,59 @@ where } } +impl InitialGoalContributor for GoalExtension +where + C: Send + Sync + 'static, +{ + fn replace_for_turn<'a>( + &'a self, + input: InitialGoalInput<'a>, + ) -> std::pin::Pin< + Box> + Send + 'a>, + > { + Box::pin(async move { + let Some(runtime) = goal_runtime_handle(input.thread_store) else { + return Err(InitialGoalError::Internal( + "goal runtime is unavailable for this thread".to_string(), + )); + }; + if !runtime.is_enabled() { + return Err(InitialGoalError::InvalidRequest( + "goals feature is disabled".to_string(), + )); + } + if input.collaboration_mode.mode == codex_protocol::config_types::ModeKind::Plan { + return Err(InitialGoalError::InvalidRequest( + "goal turns do not support plan mode".to_string(), + )); + } + + let outcome = self + .goal_service + .replace_goal_for_turn_start( + &self.state_dbs, + runtime.thread_id(), + &input.goal.objective, + ) + .await + .map_err(|err| match err { + crate::api::GoalServiceError::InvalidRequest(message) => { + InitialGoalError::InvalidRequest(message) + } + crate::api::GoalServiceError::Internal(message) => { + InitialGoalError::Internal(message) + } + })?; + self.event_emitter.thread_goal_updated( + input.turn_id, + Some(input.turn_id.to_string()), + outcome.goal, + ); + Ok(()) + }) + } +} + #[async_trait] impl TurnLifecycleContributor for GoalExtension where @@ -443,6 +499,7 @@ pub fn install_with_backend( goals_enabled, )); registry.thread_lifecycle_contributor(extension.clone()); + registry.initial_goal_contributor(extension.clone()); registry.config_contributor(extension.clone()); registry.turn_lifecycle_contributor(extension.clone()); registry.token_usage_contributor(extension.clone()); diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index 2641dfb949..72076e64db 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/codex-rs/ext/goal/src/runtime.rs @@ -211,6 +211,38 @@ impl GoalRuntimeHandle { Ok(()) } + pub(crate) fn apply_initial_goal_set( + &self, + goal: &codex_state::ThreadGoal, + previous_goal: Option<&PreviousGoalSnapshot>, + ) { + if !self.is_enabled() { + return; + } + + let replaced_existing_goal = + previous_goal.is_some_and(|previous_goal| previous_goal.goal_id != goal.goal_id); + if previous_goal.is_none() || replaced_existing_goal { + self.inner.metrics.record_created(); + } + self.inner + .accounting_state + .mark_idle_goal_active(goal.goal_id.clone()); + } + + pub(crate) fn restore_goal_after_failed_mutation( + &self, + goal: Option<&codex_state::ThreadGoal>, + ) { + match goal { + Some(goal) if goal.status == codex_state::ThreadGoalStatus::Active => self + .inner + .accounting_state + .mark_idle_goal_active(goal.goal_id.clone()), + Some(_) | None => self.inner.accounting_state.clear_active_goal(), + } + } + pub async fn apply_external_goal_clear(&self) -> Result<(), String> { if !self.is_enabled() { return Ok(()); diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index fd56805d4e..92b5e5a775 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -115,6 +115,7 @@ pub async fn run_codex_tool_session( thread_settings: Default::default(), }, client_user_message_id: None, + initial_goal: None, trace: None, }; diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs index 2f85b35143..6ddcd3ebcc 100644 --- a/codex-rs/mcp-server/src/message_processor.rs +++ b/codex-rs/mcp-server/src/message_processor.rs @@ -540,6 +540,7 @@ impl MessageProcessor { id: request_id_string, op: codex_protocol::protocol::Op::Interrupt, client_user_message_id: None, + initial_goal: None, trace: None, }) .await diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index c9e2e8bb28..d8f7ac5fd7 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -132,10 +132,17 @@ pub struct Submission { pub op: Op, /// Client-provided id for the user message represented by `Op::UserInput`. pub client_user_message_id: Option, + /// Goal to replace before processing this user-input submission. + pub initial_goal: Option, /// Optional W3C trace carrier propagated across async submission handoffs. pub trace: Option, } +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)] +pub struct InitialGoal { + pub objective: String, +} + #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] pub struct W3cTraceContext { #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 32db3d2cf7..0a175bc15d 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -710,6 +710,7 @@ impl AppServerSession { thread_id: thread_id.to_string(), client_user_message_id: None, input: items, + goal: false, responsesapi_client_metadata: None, additional_context: None, environments: None, diff --git a/sdk/python/docs/api-reference.md b/sdk/python/docs/api-reference.md index f003a28511..0ee6689c64 100644 --- a/sdk/python/docs/api-reference.md +++ b/sdk/python/docs/api-reference.md @@ -150,16 +150,16 @@ attempt. API-key login completes synchronously and does not return a handle. ### Thread -- `run(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnResult` -- `turn(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnHandle` +- `run(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnResult` +- `turn(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnHandle` - `read(*, include_turns: bool = False) -> ThreadReadResponse` - `set_name(name: str) -> ThreadSetNameResponse` - `compact() -> ThreadCompactStartResponse` ### AsyncThread -- `run(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[TurnResult]` -- `turn(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[AsyncTurnHandle]` +- `run(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[TurnResult]` +- `turn(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[AsyncTurnHandle]` - `read(*, include_turns: bool = False) -> Awaitable[ThreadReadResponse]` - `set_name(name: str) -> Awaitable[ThreadSetNameResponse]` - `compact() -> Awaitable[ThreadCompactStartResponse]` @@ -184,6 +184,10 @@ phase-less assistant message item. Use `turn(...)` when you need low-level turn control (`stream()`, `steer()`, `interrupt()`) before collecting the turn result. +Pass `goal=True` to either method for an objective that can continue through +multiple internal turns. Streaming, steering, interruption, and the returned +`TurnResult` still present one logical turn with one stable ID. + ## Sandbox Use `sandbox=` consistently on thread lifecycle methods and turns: diff --git a/sdk/python/docs/getting-started.md b/sdk/python/docs/getting-started.md index fb2c88b4c3..78547cde8d 100644 --- a/sdk/python/docs/getting-started.md +++ b/sdk/python/docs/getting-started.md @@ -72,6 +72,16 @@ with Codex() as codex: Use `Thread.turn(...)` when you need a `TurnHandle` for streaming, steering, or interrupting an active turn. +For a longer objective, opt into goal mode on the operation producing the +result: + +```python +result = thread.run("Improve the benchmark coverage in this repository.", goal=True) +``` + +Goal mode may continue working internally, but it streams and returns as one +logical turn. + ## 4. Choose Sandbox Access Use one enum for the initial thread and later turn overrides: diff --git a/sdk/python/examples/16_goal_turns/async.py b/sdk/python/examples/16_goal_turns/async.py new file mode 100644 index 0000000000..7a390d6323 --- /dev/null +++ b/sdk/python/examples/16_goal_turns/async.py @@ -0,0 +1,28 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src, runtime_config + +ensure_local_sdk_src() + +import asyncio + +from openai_codex import AsyncCodex + + +async def main() -> None: + async with AsyncCodex(config=runtime_config()) as codex: + thread = await codex.thread_start() + result = await thread.run( + "Improve the benchmark coverage in this repository.", + goal=True, + ) + print(result.final_response) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/16_goal_turns/sync.py b/sdk/python/examples/16_goal_turns/sync.py new file mode 100644 index 0000000000..d82f59c721 --- /dev/null +++ b/sdk/python/examples/16_goal_turns/sync.py @@ -0,0 +1,17 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src, runtime_config + +ensure_local_sdk_src() + +from openai_codex import Codex + +with Codex(config=runtime_config()) as codex: + thread = codex.thread_start() + result = thread.run("Improve the benchmark coverage in this repository.", goal=True) + print(result.final_response) diff --git a/sdk/python/examples/README.md b/sdk/python/examples/README.md index 8efeb75b14..bf04e64b8e 100644 --- a/sdk/python/examples/README.md +++ b/sdk/python/examples/README.md @@ -89,3 +89,5 @@ python examples/01_quickstart_constructor/async.py - separate `steer()` and `interrupt()` demos with concise summaries - `15_login_and_account/` - browser-login handle lifecycle, cancellation, and account inspection +- `16_goal_turns/` + - run a longer autonomous objective as one logical turn diff --git a/sdk/python/scripts/update_sdk_artifacts.py b/sdk/python/scripts/update_sdk_artifacts.py index a4af5d605e..683ed97247 100755 --- a/sdk/python/scripts/update_sdk_artifacts.py +++ b/sdk/python/scripts/update_sdk_artifacts.py @@ -527,6 +527,18 @@ def _normalized_schema_bundle_text(schema_dir: Path) -> str: schema = json.loads(schema_bundle_path(schema_dir).read_text()) definitions = schema.get("definitions", {}) if isinstance(definitions, dict): + turn_start = definitions.get("TurnStartParams") + if isinstance(turn_start, dict): + properties = turn_start.get("properties") + if isinstance(properties, dict): + properties["goal"] = { + "default": False, + "description": ( + "Replace the thread's active goal with an objective derived " + "from this turn's text input." + ), + "type": "boolean", + } for definition in definitions.values(): if isinstance(definition, dict): _flatten_string_enum_one_of(definition) @@ -648,6 +660,27 @@ def _notification_turn_id_specs( return (sorted(set(direct)), sorted(set(nested))) +def _notification_thread_id_specs( + schema_dir: Path, + specs: list[tuple[str, str]], +) -> list[str]: + """Return notification payloads that carry a direct thread id.""" + server_notifications = json.loads((schema_dir / "ServerNotification.json").read_text()) + definitions = server_notifications.get("definitions", {}) + if not isinstance(definitions, dict): + return [] + + direct: list[str] = [] + for _, class_name in specs: + definition = definitions.get(class_name) + if not isinstance(definition, dict): + continue + props = definition.get("properties", {}) + if isinstance(props, dict) and "threadId" in props: + direct.append(class_name) + return sorted(set(direct)) + + def _type_tuple_source(class_names: list[str]) -> str: """Render a generated tuple literal for notification payload classes.""" if not class_names: @@ -666,6 +699,7 @@ def generate_notification_registry(schema_dir: Path) -> None: schema_dir, specs, ) + direct_thread_id_types = _notification_thread_id_specs(schema_dir, specs) lines = [ "# Auto-generated by scripts/update_sdk_artifacts.py", @@ -697,6 +731,9 @@ def generate_notification_registry(schema_dir: Path) -> None: "NESTED_TURN_NOTIFICATION_TYPES: tuple[type[BaseModel], ...] = " f"{_type_tuple_source(nested_turn_types)}", "", + "DIRECT_THREAD_ID_NOTIFICATION_TYPES: tuple[type[BaseModel], ...] = " + f"{_type_tuple_source(direct_thread_id_types)}", + "", "", "def notification_turn_id(payload: BaseModel) -> str | None:", ' """Return the turn id carried by generated notification payload metadata."""', @@ -706,6 +743,13 @@ def generate_notification_registry(schema_dir: Path) -> None: " return payload.turn.id", " return None", "", + "", + "def notification_thread_id(payload: BaseModel) -> str | None:", + ' """Return the thread id carried by generated notification payload metadata."""', + " if isinstance(payload, DIRECT_THREAD_ID_NOTIFICATION_TYPES):", + " return payload.thread_id", + " return None", + "", ] ) @@ -1077,6 +1121,7 @@ def _render_thread_block( " self,", " input: RunInput,", " *,", + " goal: bool = False,", *_approval_mode_override_signature_lines(), *_kw_signature_lines(turn_fields), " ) -> TurnHandle:", @@ -1089,8 +1134,16 @@ def _render_thread_block( *_approval_mode_model_arg_lines(), *_model_arg_lines(turn_fields), " )", - " turn = self._client.turn_start(self.id, wire_input, params=params)", - " return TurnHandle(self._client, self.id, turn.turn.id)", + " goal_state = self._client.register_goal_operation(self.id) if goal else None", + " try:", + " turn = self._client.turn_start(self.id, wire_input, params=params, goal=goal)", + " except BaseException:", + " if goal_state is not None:", + " self._client.unregister_goal_operation(goal_state)", + " raise", + " if goal_state is not None:", + " self._client.bind_goal_operation(goal_state, turn.turn.id)", + " return TurnHandle(self._client, self.id, turn.turn.id, _goal=goal_state)", ] return "\n".join(lines) @@ -1103,6 +1156,7 @@ def _render_async_thread_block( " self,", " input: RunInput,", " *,", + " goal: bool = False,", *_approval_mode_override_signature_lines(), *_kw_signature_lines(turn_fields), " ) -> AsyncTurnHandle:", @@ -1116,12 +1170,21 @@ def _render_async_thread_block( *_approval_mode_model_arg_lines(), *_model_arg_lines(turn_fields), " )", - " turn = await self._codex._client.turn_start(", - " self.id,", - " wire_input,", - " params=params,", - " )", - " return AsyncTurnHandle(self._codex, self.id, turn.turn.id)", + " goal_state = self._codex._client.register_goal_operation(self.id) if goal else None", + " try:", + " turn = await self._codex._client.turn_start(", + " self.id,", + " wire_input,", + " params=params,", + " goal=goal,", + " )", + " except BaseException:", + " if goal_state is not None:", + " self._codex._client.unregister_goal_operation(goal_state)", + " raise", + " if goal_state is not None:", + " self._codex._client.bind_goal_operation(goal_state, turn.turn.id)", + " return AsyncTurnHandle(self._codex, self.id, turn.turn.id, _goal=goal_state)", ] return "\n".join(lines) @@ -1163,9 +1226,9 @@ def generate_public_api_flat_methods() -> None: turn_start_fields = _load_public_fields( "openai_codex.generated.v2_all", "TurnStartParams", - # Keep the wire model current without exposing this app-server field - # through the ergonomic Python API yet. - exclude={"thread_id", "input", "client_user_message_id", *approval_fields}, + # `goal` has a stable bool default and private routing setup, so render + # it explicitly rather than inheriting the generated model default. + exclude={"thread_id", "input", "client_user_message_id", "goal", *approval_fields}, ) turn_start_fields = _replace_public_sandbox_field(turn_start_fields, wire_name="sandbox_policy") diff --git a/sdk/python/src/openai_codex/_goal.py b/sdk/python/src/openai_codex/_goal.py new file mode 100644 index 0000000000..b9ef31eb52 --- /dev/null +++ b/sdk/python/src/openai_codex/_goal.py @@ -0,0 +1,382 @@ +import queue +import threading +from collections import deque +from dataclasses import dataclass, field +from typing import AsyncIterator, Awaitable, Callable, Iterator + +from .generated.notification_registry import notification_turn_id +from .generated.v2_all import ( + ItemCompletedNotification, + ThreadGoalClearedNotification, + ThreadGoalStatus, + ThreadGoalUpdatedNotification, + ThreadItem, + ThreadTokenUsage, + ThreadTokenUsageUpdatedNotification, + Turn, + TurnCompletedNotification, + TurnStartedNotification, + TurnStatus, +) +from .models import Notification, UnknownNotification + + +class _GoalStreamClosed(Exception): + """Wake a notification reader after its logical stream closes.""" + + +def _terminal_goal_status(status: ThreadGoalStatus | None) -> bool: + return status in { + ThreadGoalStatus.paused, + ThreadGoalStatus.blocked, + ThreadGoalStatus.usage_limited, + ThreadGoalStatus.budget_limited, + ThreadGoalStatus.complete, + } + + +@dataclass(slots=True) +class _GoalOperationState: + """Private state for one goal operation exposed as a logical turn.""" + + thread_id: str + logical_turn_id: str | None = None + current_turn_id: str | None = None + status: ThreadGoalStatus | None = None + started_turn: Turn | None = None + completed_turn: Turn | None = None + items: list[ThreadItem] = field(default_factory=list) + usage: ThreadTokenUsage | None = None + interrupted: bool = False + interrupt_requested: bool = False + cleared: bool = False + _condition: threading.Condition = field(default_factory=threading.Condition) + _notifications: queue.Queue[Notification | BaseException] = field(default_factory=queue.Queue) + _failure: BaseException | None = None + _finished: bool = False + + def bind(self, logical_turn_id: str) -> None: + with self._condition: + self.logical_turn_id = logical_turn_id + if self.current_turn_id is None and self.completed_turn is None: + self.current_turn_id = logical_turn_id + self._condition.notify_all() + + def observe(self, notification: Notification) -> None: + payload = notification.payload + with self._condition: + if isinstance(payload, TurnStartedNotification): + self.current_turn_id = payload.turn.id + if self.started_turn is None: + self.started_turn = payload.turn + elif isinstance(payload, TurnCompletedNotification): + self.completed_turn = payload.turn + if self.current_turn_id == payload.turn.id: + self.current_turn_id = None + elif isinstance(payload, ThreadGoalUpdatedNotification): + self.status = payload.goal.status + elif isinstance(payload, ThreadGoalClearedNotification): + self.cleared = True + elif isinstance(payload, ItemCompletedNotification): + self.items.append(payload.item) + elif isinstance(payload, ThreadTokenUsageUpdatedNotification): + self.usage = payload.token_usage + if ( + isinstance(payload, TurnCompletedNotification) + and payload.turn.status in {TurnStatus.failed, TurnStatus.interrupted} + ) or ( + self.current_turn_id is None + and self.completed_turn is not None + and (self.cleared or _terminal_goal_status(self.status)) + ): + self._finished = True + self._condition.notify_all() + self._notifications.put(notification) + + def fail(self, exc: BaseException) -> None: + with self._condition: + self._failure = exc + self._condition.notify_all() + self._notifications.put(exc) + + def next_notification(self) -> Notification: + item = self._notifications.get() + if isinstance(item, BaseException): + raise item + return item + + def finish(self) -> None: + """Mark the logical operation inactive and wake waiting controls.""" + with self._condition: + self._finished = True + self.current_turn_id = None + self._condition.notify_all() + + def is_finished(self) -> bool: + with self._condition: + return self._finished + + def begin_interrupt(self) -> bool: + with self._condition: + if self._finished: + return False + self.interrupt_requested = True + return True + + def confirm_interrupt(self) -> None: + with self._condition: + self.interrupted = True + self.interrupt_requested = False + + def cancel_interrupt(self) -> None: + with self._condition: + self.interrupt_requested = False + + def explicit_interrupt(self, status: ThreadGoalStatus | None) -> bool: + with self._condition: + return self.interrupted or ( + self.interrupt_requested and status == ThreadGoalStatus.paused + ) + + def active_turn(self, *, after: str | None = None) -> str | None: + """Wait for the current turn, or return None once the goal has ended.""" + with self._condition: + while True: + if self._failure is not None: + raise self._failure + if self._finished: + return None + if self.current_turn_id is not None and self.current_turn_id != after: + return self.current_turn_id + if self.cleared or _terminal_goal_status(self.status): + return None + self._condition.wait() + + def current_turn(self) -> str | None: + """Return the current physical turn without waiting for rollover.""" + with self._condition: + return self.current_turn_id + + def wake_notification_reader(self) -> None: + """Release a reader blocked after its stream has been closed.""" + self._notifications.put(_GoalStreamClosed()) + + +def _logical_notification(notification: Notification, logical_turn_id: str) -> Notification: + """Return a copy whose turn metadata uses the logical operation id.""" + payload = notification.payload + if isinstance(payload, UnknownNotification): + params = dict(payload.params) + if isinstance(params.get("turnId"), str): + params["turnId"] = logical_turn_id + turn = params.get("turn") + if isinstance(turn, dict) and isinstance(turn.get("id"), str): + params["turn"] = {**turn, "id": logical_turn_id} + return Notification(notification.method, UnknownNotification(params)) + + turn_id = notification_turn_id(payload) + if turn_id is None: + return notification + if hasattr(payload, "turn_id"): + return Notification( + notification.method, + payload.model_copy(update={"turn_id": logical_turn_id}), + ) + if hasattr(payload, "turn"): + logical_turn = payload.turn.model_copy(update={"id": logical_turn_id}) + return Notification( + notification.method, + payload.model_copy(update={"turn": logical_turn}), + ) + return notification + + +def _logical_completion( + completed: TurnCompletedNotification, + *, + logical_turn_id: str, + started: Turn | None, + interrupted: bool, +) -> TurnCompletedNotification: + """Coalesce the final physical completion into one logical completion.""" + final_turn = completed.turn + started_at = started.started_at if started is not None else final_turn.started_at + duration_ms = final_turn.duration_ms + if started_at is not None and final_turn.completed_at is not None: + duration_ms = max(0, final_turn.completed_at - started_at) * 1000 + updates: dict[str, object] = { + "id": logical_turn_id, + "started_at": started_at, + "duration_ms": duration_ms, + } + if interrupted: + updates["status"] = TurnStatus.interrupted + return completed.model_copy(update={"turn": final_turn.model_copy(update=updates)}) + + +@dataclass(slots=True) +class _GoalStreamCursor: + """Consume physical goal events as one ordered logical turn stream.""" + + state: _GoalOperationState + started: Turn | None = None + last_completed: TurnCompletedNotification | None = None + status: ThreadGoalStatus | None = None + active: bool = False + cleared: bool = False + + def process(self, notification: Notification) -> tuple[list[Notification], bool]: + logical_turn_id = self.state.logical_turn_id + if logical_turn_id is None: + raise RuntimeError("goal operation has not been bound to a logical turn id") + + payload = notification.payload + if isinstance(payload, TurnStartedNotification): + self.active = True + if self.started is not None: + return [], False + self.started = payload.turn + return [_logical_notification(notification, logical_turn_id)], False + + if isinstance(payload, TurnCompletedNotification): + self.active = False + self.last_completed = payload + if payload.turn.status in {TurnStatus.failed, TurnStatus.interrupted}: + self.state.finish() + return [self._completion(notification.method, payload)], True + if self.status is None and not self.cleared: + raise RuntimeError( + "the connected Codex runtime did not activate goal mode for this turn" + ) + if self.cleared or _terminal_goal_status(self.status): + self.state.finish() + return [self._completion(notification.method, payload)], True + return [], False + + events = [_logical_notification(notification, logical_turn_id)] + if isinstance(payload, ThreadGoalUpdatedNotification): + self.status = payload.goal.status + events = [] + elif isinstance(payload, ThreadGoalClearedNotification): + self.cleared = True + events = [] + + if ( + not self.active + and self.last_completed is not None + and (self.cleared or _terminal_goal_status(self.status)) + ): + self.state.finish() + events.append(self._completion("turn/completed", self.last_completed)) + return events, True + return events, False + + def _completion( + self, + method: str, + payload: TurnCompletedNotification, + ) -> Notification: + logical_turn_id = self.state.logical_turn_id + if logical_turn_id is None: + raise RuntimeError("goal operation has not been bound to a logical turn id") + return Notification( + method, + _logical_completion( + payload, + logical_turn_id=logical_turn_id, + started=self.started, + interrupted=self.state.explicit_interrupt(self.status), + ), + ) + + +@dataclass(slots=True) +class _GoalNotificationStream(Iterator[Notification]): + """Closeable synchronous view of one logical goal operation.""" + + state: _GoalOperationState + next_notification: Callable[[], Notification] + unregister: Callable[[], None] + _cursor: _GoalStreamCursor = field(init=False) + _pending: deque[Notification] = field(default_factory=deque) + _closed: bool = False + + def __post_init__(self) -> None: + self._cursor = _GoalStreamCursor(self.state) + + def __iter__(self) -> "_GoalNotificationStream": + return self + + def __next__(self) -> Notification: + if self._closed: + raise StopIteration + try: + while not self._pending: + events, completed = self._cursor.process(self.next_notification()) + self._pending.extend(events) + if completed: + self._finish() + return self._pending.popleft() + except _GoalStreamClosed: + self.close() + raise StopIteration from None + except BaseException: + self.close() + raise + + def _finish(self) -> None: + if self._closed: + return + self.state.finish() + self.state.wake_notification_reader() + self.unregister() + self._closed = True + + def close(self) -> None: + self._finish() + + +@dataclass(slots=True) +class _AsyncGoalNotificationStream(AsyncIterator[Notification]): + """Closeable asynchronous view of one logical goal operation.""" + + state: _GoalOperationState + next_notification: Callable[[], Awaitable[Notification]] + unregister: Callable[[], None] + _cursor: _GoalStreamCursor = field(init=False) + _pending: deque[Notification] = field(default_factory=deque) + _closed: bool = False + + def __post_init__(self) -> None: + self._cursor = _GoalStreamCursor(self.state) + + def __aiter__(self) -> "_AsyncGoalNotificationStream": + return self + + async def __anext__(self) -> Notification: + if self._closed: + raise StopAsyncIteration + try: + while not self._pending: + events, completed = self._cursor.process(await self.next_notification()) + self._pending.extend(events) + if completed: + self._finish() + return self._pending.popleft() + except _GoalStreamClosed: + await self.aclose() + raise StopAsyncIteration from None + except BaseException: + await self.aclose() + raise + + def _finish(self) -> None: + if self._closed: + return + self.state.finish() + self.state.wake_notification_reader() + self.unregister() + self._closed = True + + async def aclose(self) -> None: + self._finish() diff --git a/sdk/python/src/openai_codex/_message_router.py b/sdk/python/src/openai_codex/_message_router.py index b500f16c56..c77e9c9b08 100644 --- a/sdk/python/src/openai_codex/_message_router.py +++ b/sdk/python/src/openai_codex/_message_router.py @@ -4,8 +4,9 @@ import queue import threading from collections import deque +from ._goal import _GoalOperationState from .errors import CodexError, map_jsonrpc_error -from .generated.notification_registry import notification_turn_id +from .generated.notification_registry import notification_thread_id, notification_turn_id from .generated.v2_all import AccountLoginCompletedNotification from .models import JsonValue, Notification, UnknownNotification @@ -30,6 +31,7 @@ class MessageRouter: self._pending_login_notifications: dict[str, deque[Notification]] = {} self._turn_notifications: dict[str, queue.Queue[NotificationQueueItem]] = {} self._pending_turn_notifications: dict[str, deque[Notification]] = {} + self._goal_operations: dict[str, _GoalOperationState] = {} self._global_notifications: queue.Queue[NotificationQueueItem] = queue.Queue() def create_response_waiter(self, request_id: str) -> queue.Queue[ResponseQueueItem]: @@ -116,6 +118,21 @@ class MessageRouter: raise item return item + def register_goal(self, thread_id: str) -> _GoalOperationState: + """Register one thread-scoped logical goal operation before it starts.""" + state = _GoalOperationState(thread_id=thread_id) + with self._lock: + if thread_id in self._goal_operations: + raise RuntimeError(f"thread {thread_id!r} already has an active goal operation") + self._goal_operations[thread_id] = state + return state + + def unregister_goal(self, state: _GoalOperationState) -> None: + """Stop routing notifications to a completed logical goal operation.""" + with self._lock: + if self._goal_operations.get(state.thread_id) is state: + self._goal_operations.pop(state.thread_id) + def route_response(self, msg: dict[str, JsonValue]) -> None: """Deliver a JSON-RPC response or error to its request waiter.""" @@ -157,6 +174,17 @@ class MessageRouter: return turn_id = self._notification_turn_id(notification) + thread_id = self._notification_thread_id(notification) + if thread_id is not None: + with self._lock: + goal_state = self._goal_operations.get(thread_id) + if goal_state is not None and ( + turn_id is not None or notification.method.startswith("thread/goal/") + ): + goal_state.observe(notification) + if goal_state.is_finished(): + self.unregister_goal(goal_state) + return if turn_id is None: self._global_notifications.put(notification) return @@ -182,6 +210,8 @@ class MessageRouter: self._pending_login_notifications.clear() turn_queues = list(self._turn_notifications.values()) self._pending_turn_notifications.clear() + goal_operations = list(self._goal_operations.values()) + self._goal_operations.clear() # Put the same transport failure into every queue so no SDK call blocks # forever waiting for a response that cannot arrive. for waiter in response_waiters: @@ -190,6 +220,8 @@ class MessageRouter: login_queue.put(exc) for turn_queue in turn_queues: turn_queue.put(exc) + for goal_operation in goal_operations: + goal_operation.fail(exc) self._global_notifications.put(exc) def _notification_login_id(self, notification: Notification) -> str | None: @@ -220,3 +252,11 @@ class MessageRouter: return raw_nested_turn_id return None return notification_turn_id(payload) + + def _notification_thread_id(self, notification: Notification) -> str | None: + """Extract thread ids from known generated payloads or raw payloads.""" + payload = notification.payload + if isinstance(payload, UnknownNotification): + raw_thread_id = payload.params.get("threadId") + return raw_thread_id if isinstance(raw_thread_id, str) else None + return notification_thread_id(payload) diff --git a/sdk/python/src/openai_codex/api.py b/sdk/python/src/openai_codex/api.py index 6fc9a8243d..54852dedd1 100644 --- a/sdk/python/src/openai_codex/api.py +++ b/sdk/python/src/openai_codex/api.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import re from dataclasses import dataclass from typing import AsyncIterator, Iterator @@ -9,6 +10,11 @@ from ._approval_mode import ( _approval_mode_override_settings, _approval_mode_settings, ) +from ._goal import ( + _AsyncGoalNotificationStream, + _GoalNotificationStream, + _GoalOperationState, +) from ._initialize_metadata import validate_initialize_metadata from ._inputs import ( ImageInput as ImageInput, @@ -40,6 +46,7 @@ from ._run import ( from ._sandbox import Sandbox as Sandbox, _sandbox_mode, _sandbox_policy from .async_client import AsyncCodexClient from .client import CodexClient, CodexConfig +from .errors import InvalidRequestError from .generated.v2_all import ( ApiKeyLoginAccountParams, GetAccountParams, @@ -72,6 +79,19 @@ from .generated.v2_all import ( from .models import InitializeResponse, JsonObject, Notification +def _active_turn_id_from_error(exc: InvalidRequestError) -> str | None: + match = re.search(r" but found `?([^`]+)`?$", exc.message) + return match.group(1) if match is not None else None + + +def _inactive_turn_error() -> InvalidRequestError: + return InvalidRequestError(-32600, "no active turn to steer") + + +def _inactive_interrupt_error() -> InvalidRequestError: + return InvalidRequestError(-32600, "no active turn to interrupt") + + class Codex: """Synchronous client for creating threads and running Codex turns. @@ -541,6 +561,7 @@ class Thread: self, input: RunInput, *, + goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -554,6 +575,7 @@ class Thread: """Run a complete turn and collect its final result.""" turn = self.turn( input, + goal=goal, approval_mode=approval_mode, cwd=cwd, effort=effort, @@ -575,6 +597,7 @@ class Thread: self, input: RunInput, *, + goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -602,8 +625,16 @@ class Thread: service_tier=service_tier, summary=summary, ) - turn = self._client.turn_start(self.id, wire_input, params=params) - return TurnHandle(self._client, self.id, turn.turn.id) + goal_state = self._client.register_goal_operation(self.id) if goal else None + try: + turn = self._client.turn_start(self.id, wire_input, params=params, goal=goal) + except BaseException: + if goal_state is not None: + self._client.unregister_goal_operation(goal_state) + raise + if goal_state is not None: + self._client.bind_goal_operation(goal_state, turn.turn.id) + return TurnHandle(self._client, self.id, turn.turn.id, _goal=goal_state) # END GENERATED: Thread.flat_methods @@ -629,6 +660,7 @@ class AsyncThread: self, input: RunInput, *, + goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -642,6 +674,7 @@ class AsyncThread: """Run a complete turn asynchronously and collect its final result.""" turn = await self.turn( input, + goal=goal, approval_mode=approval_mode, cwd=cwd, effort=effort, @@ -663,6 +696,7 @@ class AsyncThread: self, input: RunInput, *, + goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -691,12 +725,21 @@ class AsyncThread: service_tier=service_tier, summary=summary, ) - turn = await self._codex._client.turn_start( - self.id, - wire_input, - params=params, - ) - return AsyncTurnHandle(self._codex, self.id, turn.turn.id) + goal_state = self._codex._client.register_goal_operation(self.id) if goal else None + try: + turn = await self._codex._client.turn_start( + self.id, + wire_input, + params=params, + goal=goal, + ) + except BaseException: + if goal_state is not None: + self._codex._client.unregister_goal_operation(goal_state) + raise + if goal_state is not None: + self._codex._client.bind_goal_operation(goal_state, turn.turn.id) + return AsyncTurnHandle(self._codex, self.id, turn.turn.id, _goal=goal_state) # END GENERATED: AsyncThread.flat_methods @@ -721,9 +764,30 @@ class TurnHandle: _client: CodexClient thread_id: str id: str + _goal: _GoalOperationState | None = None def steer(self, input: RunInput) -> TurnSteerResponse: """Send additional input to this active turn.""" + if self._goal is not None: + wire_input = _to_wire_input(_normalize_run_input(input)) + turn_id = self._goal.active_turn() + if turn_id is None: + raise _inactive_turn_error() + try: + response = self._client.turn_steer(self.thread_id, turn_id, wire_input) + except InvalidRequestError as exc: + if not ( + exc.message == "no active turn to steer" + or exc.message.startswith("expected active turn id") + ): + raise + next_turn_id = _active_turn_id_from_error(exc) + if next_turn_id is None: + next_turn_id = self._goal.active_turn(after=turn_id) + if next_turn_id is None: + raise _inactive_turn_error() from exc + response = self._client.turn_steer(self.thread_id, next_turn_id, wire_input) + return response.model_copy(update={"turn_id": self.id}) return self._client.turn_steer( self.thread_id, self.id, @@ -732,23 +796,61 @@ class TurnHandle: def interrupt(self) -> TurnInterruptResponse: """Request interruption of this active turn.""" + if self._goal is not None: + if not self._goal.begin_interrupt(): + raise _inactive_interrupt_error() + try: + self._client.pause_goal(self.thread_id) + except BaseException: + self._goal.cancel_interrupt() + raise + self._goal.confirm_interrupt() + turn_id = self._goal.current_turn() + if turn_id is None: + return TurnInterruptResponse() + try: + return self._client.turn_interrupt(self.thread_id, turn_id) + except InvalidRequestError as exc: + if exc.message == "no active turn to interrupt": + return TurnInterruptResponse() + if exc.message.startswith("expected active turn id"): + next_turn_id = _active_turn_id_from_error(exc) or self._goal.current_turn() + if next_turn_id is None or next_turn_id == turn_id: + return TurnInterruptResponse() + try: + return self._client.turn_interrupt(self.thread_id, next_turn_id) + except InvalidRequestError as retry_exc: + if retry_exc.message == "no active turn to interrupt": + return TurnInterruptResponse() + raise + raise return self._client.turn_interrupt(self.thread_id, self.id) def stream(self) -> Iterator[Notification]: """Yield only notifications routed to this turn handle.""" - self._client.register_turn_notifications(self.id) - try: - while True: - event = self._client.next_turn_notification(self.id) - yield event - if ( - event.method == "turn/completed" - and isinstance(event.payload, TurnCompletedNotification) - and event.payload.turn.id == self.id - ): - break - finally: - self._client.unregister_turn_notifications(self.id) + if self._goal is not None: + return _GoalNotificationStream( + self._goal, + lambda: self._client.next_goal_notification(self._goal), + lambda: self._client.unregister_goal_operation(self._goal), + ) + + def ordinary_stream() -> Iterator[Notification]: + self._client.register_turn_notifications(self.id) + try: + while True: + event = self._client.next_turn_notification(self.id) + yield event + if ( + event.method == "turn/completed" + and isinstance(event.payload, TurnCompletedNotification) + and event.payload.turn.id == self.id + ): + break + finally: + self._client.unregister_turn_notifications(self.id) + + return ordinary_stream() def run(self) -> TurnResult: """Consume the turn stream and return its completed result.""" @@ -766,10 +868,39 @@ class AsyncTurnHandle: _codex: AsyncCodex thread_id: str id: str + _goal: _GoalOperationState | None = None async def steer(self, input: RunInput) -> TurnSteerResponse: """Send additional input to this active turn.""" await self._codex._ensure_initialized() + if self._goal is not None: + wire_input = _to_wire_input(_normalize_run_input(input)) + turn_id = await asyncio.to_thread(self._goal.active_turn) + if turn_id is None: + raise _inactive_turn_error() + try: + response = await self._codex._client.turn_steer( + self.thread_id, + turn_id, + wire_input, + ) + except InvalidRequestError as exc: + if not ( + exc.message == "no active turn to steer" + or exc.message.startswith("expected active turn id") + ): + raise + next_turn_id = _active_turn_id_from_error(exc) + if next_turn_id is None: + next_turn_id = await asyncio.to_thread(self._goal.active_turn, after=turn_id) + if next_turn_id is None: + raise _inactive_turn_error() from exc + response = await self._codex._client.turn_steer( + self.thread_id, + next_turn_id, + wire_input, + ) + return response.model_copy(update={"turn_id": self.id}) return await self._codex._client.turn_steer( self.thread_id, self.id, @@ -779,24 +910,70 @@ class AsyncTurnHandle: async def interrupt(self) -> TurnInterruptResponse: """Request interruption of this active turn.""" await self._codex._ensure_initialized() + if self._goal is not None: + if not self._goal.begin_interrupt(): + raise _inactive_interrupt_error() + try: + await self._codex._client.pause_goal(self.thread_id) + except BaseException: + self._goal.cancel_interrupt() + raise + self._goal.confirm_interrupt() + turn_id = self._goal.current_turn() + if turn_id is None: + return TurnInterruptResponse() + try: + return await self._codex._client.turn_interrupt(self.thread_id, turn_id) + except InvalidRequestError as exc: + if exc.message == "no active turn to interrupt": + return TurnInterruptResponse() + if exc.message.startswith("expected active turn id"): + next_turn_id = _active_turn_id_from_error(exc) or self._goal.current_turn() + if next_turn_id is None or next_turn_id == turn_id: + return TurnInterruptResponse() + try: + return await self._codex._client.turn_interrupt( + self.thread_id, + next_turn_id, + ) + except InvalidRequestError as retry_exc: + if retry_exc.message == "no active turn to interrupt": + return TurnInterruptResponse() + raise + raise return await self._codex._client.turn_interrupt(self.thread_id, self.id) - async def stream(self) -> AsyncIterator[Notification]: + def stream(self) -> AsyncIterator[Notification]: """Yield only notifications routed to this async turn handle.""" - await self._codex._ensure_initialized() - self._codex._client.register_turn_notifications(self.id) - try: - while True: - event = await self._codex._client.next_turn_notification(self.id) - yield event - if ( - event.method == "turn/completed" - and isinstance(event.payload, TurnCompletedNotification) - and event.payload.turn.id == self.id - ): - break - finally: - self._codex._client.unregister_turn_notifications(self.id) + if self._goal is not None: + + async def next_goal_notification() -> Notification: + await self._codex._ensure_initialized() + return await self._codex._client.next_goal_notification(self._goal) + + return _AsyncGoalNotificationStream( + self._goal, + next_goal_notification, + lambda: self._codex._client.unregister_goal_operation(self._goal), + ) + + async def ordinary_stream() -> AsyncIterator[Notification]: + await self._codex._ensure_initialized() + self._codex._client.register_turn_notifications(self.id) + try: + while True: + event = await self._codex._client.next_turn_notification(self.id) + yield event + if ( + event.method == "turn/completed" + and isinstance(event.payload, TurnCompletedNotification) + and event.payload.turn.id == self.id + ): + break + finally: + self._codex._client.unregister_turn_notifications(self.id) + + return ordinary_stream() async def run(self) -> TurnResult: """Consume the turn stream and return its completed result.""" diff --git a/sdk/python/src/openai_codex/async_client.py b/sdk/python/src/openai_codex/async_client.py index f0b12b99ed..5ea34ead8e 100644 --- a/sdk/python/src/openai_codex/async_client.py +++ b/sdk/python/src/openai_codex/async_client.py @@ -6,6 +6,7 @@ from typing import AsyncIterator, Callable, ParamSpec, TypeVar from pydantic import BaseModel +from ._goal import _GoalOperationState from .client import CodexClient, CodexConfig from .generated.v2_all import ( AccountLoginCompletedNotification, @@ -21,6 +22,7 @@ from .generated.v2_all import ( ThreadCompactStartResponse, ThreadForkParams as V2ThreadForkParams, ThreadForkResponse, + ThreadGoalSetResponse, ThreadListParams as V2ThreadListParams, ThreadListResponse, ThreadReadResponse, @@ -107,6 +109,18 @@ class AsyncCodexClient: """Unregister a turn notification queue on the wrapped sync client.""" self._sync.unregister_turn_notifications(turn_id) + def register_goal_operation(self, thread_id: str) -> _GoalOperationState: + """Register a logical goal route on the wrapped sync client.""" + return self._sync.register_goal_operation(thread_id) + + def bind_goal_operation(self, state: _GoalOperationState, turn_id: str) -> None: + """Bind a logical goal route to its stable turn id.""" + self._sync.bind_goal_operation(state, turn_id) + + def unregister_goal_operation(self, state: _GoalOperationState) -> None: + """Release one logical goal route.""" + self._sync.unregister_goal_operation(state) + async def request( self, method: str, @@ -192,14 +206,26 @@ class AsyncCodexClient: """Start thread compaction using the wrapped sync client.""" return await self._call_sync(self._sync.thread_compact, thread_id) + async def pause_goal(self, thread_id: str) -> ThreadGoalSetResponse: + """Pause the active goal through the wrapped sync client.""" + return await self._call_sync(self._sync.pause_goal, thread_id) + async def turn_start( self, thread_id: str, input_items: list[JsonObject] | JsonObject | str, params: V2TurnStartParams | JsonObject | None = None, + *, + goal: bool = False, ) -> TurnStartResponse: """Start a turn using the wrapped sync client.""" - return await self._call_sync(self._sync.turn_start, thread_id, input_items, params) + return await self._call_sync( + self._sync.turn_start, + thread_id, + input_items, + params, + goal=goal, + ) async def turn_interrupt(self, thread_id: str, turn_id: str) -> TurnInterruptResponse: """Interrupt a turn using the wrapped sync client.""" @@ -256,6 +282,10 @@ class AsyncCodexClient: """Wait for the next notification routed to one turn.""" return await self._call_sync(self._sync.next_turn_notification, turn_id) + async def next_goal_notification(self, state: _GoalOperationState) -> Notification: + """Wait for the next notification in a logical goal turn.""" + return await self._call_sync(self._sync.next_goal_notification, state) + async def wait_for_login_completed( self, login_id: str, diff --git a/sdk/python/src/openai_codex/client.py b/sdk/python/src/openai_codex/client.py index 9951d6d1f3..c8e0b09191 100644 --- a/sdk/python/src/openai_codex/client.py +++ b/sdk/python/src/openai_codex/client.py @@ -10,6 +10,7 @@ from typing import Callable, Iterator, TypeVar from pydantic import BaseModel +from ._goal import _GoalOperationState from ._message_router import MessageRouter from ._version import __version__ as SDK_VERSION from .errors import CodexError, TransportClosedError @@ -30,6 +31,8 @@ from .generated.v2_all import ( ThreadCompactStartResponse, ThreadForkParams as V2ThreadForkParams, ThreadForkResponse, + ThreadGoalSetResponse, + ThreadGoalStatus, ThreadListParams as V2ThreadListParams, ThreadListResponse, ThreadReadResponse, @@ -352,6 +355,23 @@ class CodexClient: """Return the next routed notification for the requested turn id.""" return self._router.next_turn_notification(turn_id) + def register_goal_operation(self, thread_id: str) -> _GoalOperationState: + """Register a private thread-scoped route for a logical goal turn.""" + return self._router.register_goal(thread_id) + + def bind_goal_operation(self, state: _GoalOperationState, turn_id: str) -> None: + """Bind a pending goal route to its stable logical turn id.""" + state.bind(turn_id) + self.unregister_turn_notifications(turn_id) + + def unregister_goal_operation(self, state: _GoalOperationState) -> None: + """Release routing state for one logical goal turn.""" + self._router.unregister_goal(state) + + def next_goal_notification(self, state: _GoalOperationState) -> Notification: + """Wait for the next notification in a logical goal turn.""" + return state.next_notification() + def account_login_start( self, params: V2LoginAccountParams | JsonObject, @@ -452,11 +472,21 @@ class CodexClient: response_model=ThreadCompactStartResponse, ) + def pause_goal(self, thread_id: str) -> ThreadGoalSetResponse: + """Pause the active goal used by a logical goal turn.""" + return self.request( + "thread/goal/set", + {"threadId": thread_id, "status": ThreadGoalStatus.paused.value}, + response_model=ThreadGoalSetResponse, + ) + def turn_start( self, thread_id: str, input_items: list[JsonObject] | JsonObject | str, params: V2TurnStartParams | JsonObject | None = None, + *, + goal: bool = False, ) -> TurnStartResponse: """Start a turn and register its notification queue as early as possible.""" payload = { @@ -464,6 +494,11 @@ class CodexClient: "threadId": thread_id, "input": self._normalize_input_items(input_items), } + params_goal = payload.pop("goal", False) + if not isinstance(params_goal, bool): + raise TypeError("turn/start goal must be a bool") + if goal or params_goal: + payload["goal"] = True started = self.request("turn/start", payload, response_model=TurnStartResponse) self.register_turn_notifications(started.turn.id) return started diff --git a/sdk/python/src/openai_codex/generated/notification_registry.py b/sdk/python/src/openai_codex/generated/notification_registry.py index d5e620a7a3..190135a077 100644 --- a/sdk/python/src/openai_codex/generated/notification_registry.py +++ b/sdk/python/src/openai_codex/generated/notification_registry.py @@ -169,6 +169,53 @@ NESTED_TURN_NOTIFICATION_TYPES: tuple[type[BaseModel], ...] = ( TurnStartedNotification, ) +DIRECT_THREAD_ID_NOTIFICATION_TYPES: tuple[type[BaseModel], ...] = ( + AgentMessageDeltaNotification, + CommandExecutionOutputDeltaNotification, + ContextCompactedNotification, + ErrorNotification, + FileChangeOutputDeltaNotification, + FileChangePatchUpdatedNotification, + GuardianWarningNotification, + HookCompletedNotification, + HookStartedNotification, + ItemCompletedNotification, + ItemGuardianApprovalReviewCompletedNotification, + ItemGuardianApprovalReviewStartedNotification, + ItemStartedNotification, + McpToolCallProgressNotification, + ModelReroutedNotification, + ModelVerificationNotification, + PlanDeltaNotification, + ReasoningSummaryPartAddedNotification, + ReasoningSummaryTextDeltaNotification, + ReasoningTextDeltaNotification, + ServerRequestResolvedNotification, + TerminalInteractionNotification, + ThreadArchivedNotification, + ThreadClosedNotification, + ThreadGoalClearedNotification, + ThreadGoalUpdatedNotification, + ThreadNameUpdatedNotification, + ThreadRealtimeClosedNotification, + ThreadRealtimeErrorNotification, + ThreadRealtimeItemAddedNotification, + ThreadRealtimeOutputAudioDeltaNotification, + ThreadRealtimeSdpNotification, + ThreadRealtimeStartedNotification, + ThreadRealtimeTranscriptDeltaNotification, + ThreadRealtimeTranscriptDoneNotification, + ThreadSettingsUpdatedNotification, + ThreadStatusChangedNotification, + ThreadTokenUsageUpdatedNotification, + ThreadUnarchivedNotification, + TurnCompletedNotification, + TurnDiffUpdatedNotification, + TurnPlanUpdatedNotification, + TurnStartedNotification, + WarningNotification, +) + def notification_turn_id(payload: BaseModel) -> str | None: """Return the turn id carried by generated notification payload metadata.""" @@ -177,3 +224,10 @@ def notification_turn_id(payload: BaseModel) -> str | None: if isinstance(payload, NESTED_TURN_NOTIFICATION_TYPES): return payload.turn.id return None + + +def notification_thread_id(payload: BaseModel) -> str | None: + """Return the thread id carried by generated notification payload metadata.""" + if isinstance(payload, DIRECT_THREAD_ID_NOTIFICATION_TYPES): + return payload.thread_id + return None diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 15ede1801c..2bc193a5e5 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -6956,6 +6956,12 @@ class TurnStartParams(BaseModel): ReasoningEffort | None, Field(description="Override the reasoning effort for this turn and subsequent turns."), ] = None + goal: Annotated[ + bool | None, + Field( + description="Replace the thread's active goal with an objective derived from this turn's text input." + ), + ] = False input: list[UserInput] model: Annotated[ str | None, Field(description="Override the model for this turn and subsequent turns.") diff --git a/sdk/python/src/openai_codex/models.py b/sdk/python/src/openai_codex/models.py index 70c61d44cd..d9d15dc684 100644 --- a/sdk/python/src/openai_codex/models.py +++ b/sdk/python/src/openai_codex/models.py @@ -27,6 +27,8 @@ from .generated.v2_all import ( ReasoningSummaryTextDeltaNotification, ReasoningTextDeltaNotification, TerminalInteractionNotification, + ThreadGoalClearedNotification, + ThreadGoalUpdatedNotification, ThreadNameUpdatedNotification, ThreadStartedNotification, ThreadTokenUsageUpdatedNotification, @@ -70,6 +72,8 @@ NotificationPayload: TypeAlias = ( | ReasoningTextDeltaNotification | TerminalInteractionNotification | ThreadNameUpdatedNotification + | ThreadGoalClearedNotification + | ThreadGoalUpdatedNotification | ThreadStartedNotification | ThreadTokenUsageUpdatedNotification | TurnCompletedNotification diff --git a/sdk/python/tests/test_client_rpc_methods.py b/sdk/python/tests/test_client_rpc_methods.py index 73fdd452ff..406926ca64 100644 --- a/sdk/python/tests/test_client_rpc_methods.py +++ b/sdk/python/tests/test_client_rpc_methods.py @@ -3,7 +3,10 @@ from __future__ import annotations from pathlib import Path from openai_codex.client import CodexClient, _params_dict -from openai_codex.generated.notification_registry import notification_turn_id +from openai_codex.generated.notification_registry import ( + notification_thread_id, + notification_turn_id, +) from openai_codex.generated.v2_all import ( AgentMessageDeltaNotification, ApprovalsReviewer, @@ -11,6 +14,8 @@ from openai_codex.generated.v2_all import ( ThreadResumeResponse, ThreadTokenUsageUpdatedNotification, TurnCompletedNotification, + TurnStartParams, + TurnStartResponse, WarningNotification, ) from openai_codex.models import Notification, UnknownNotification @@ -26,11 +31,73 @@ def test_generated_params_models_are_snake_case_and_dump_by_alias() -> None: assert dumped == {"searchTerm": "needle", "limit": 5} +def test_turn_start_sends_goal_only_when_enabled() -> None: + """The low-level request should preserve false-by-omission wire behavior.""" + client = CodexClient() + requests: list[dict[str, object]] = [] + + def fake_request(method, params, *, response_model): + requests.append({"method": method, "params": params, "response_model": response_model}) + return TurnStartResponse.model_validate( + {"turn": {"id": f"turn-{len(requests)}", "items": [], "status": "inProgress"}} + ) + + client.request = fake_request # type: ignore[method-assign] + + client.turn_start("thread-1", "ordinary") + client.turn_start("thread-1", "goal", goal=True) + client.turn_start( + "thread-1", + "typed goal", + TurnStartParams(threadId="thread-1", input=[], goal=True), + ) + + assert requests == [ + { + "method": "turn/start", + "params": { + "threadId": "thread-1", + "input": [{"type": "text", "text": "ordinary"}], + }, + "response_model": TurnStartResponse, + }, + { + "method": "turn/start", + "params": { + "threadId": "thread-1", + "input": [{"type": "text", "text": "goal"}], + "goal": True, + }, + "response_model": TurnStartResponse, + }, + { + "method": "turn/start", + "params": { + "threadId": "thread-1", + "input": [{"type": "text", "text": "typed goal"}], + "goal": True, + }, + "response_model": TurnStartResponse, + }, + ] + + def test_generated_v2_bundle_has_single_shared_plan_type_definition() -> None: source = (ROOT / "src" / "openai_codex" / "generated" / "v2_all.py").read_text() assert source.count("class PlanType(") == 1 +def test_generated_turn_start_params_include_goal_mode() -> None: + assert ( + TurnStartParams( + threadId="thread-1", + input=[{"type": "text", "text": "goal"}], + goal=True, + ).goal + is True + ) + + def test_thread_resume_response_accepts_auto_review_reviewer() -> None: """Generated response models should keep accepting the auto review enum value.""" response = ThreadResumeResponse.model_validate( @@ -142,6 +209,24 @@ def test_generated_notification_turn_id_handles_known_payload_shapes() -> None: ] == ["turn-1", "turn-2", None] +def test_generated_notification_thread_id_handles_known_payload_shapes() -> None: + """Generated routing metadata should expose thread ids without field guessing.""" + direct = AgentMessageDeltaNotification.model_validate( + { + "delta": "hello", + "itemId": "item-1", + "threadId": "thread-1", + "turnId": "turn-1", + } + ) + unscoped = WarningNotification(message="heads up") + + assert [notification_thread_id(direct), notification_thread_id(unscoped)] == [ + "thread-1", + None, + ] + + def test_turn_notification_router_demuxes_registered_turns() -> None: """The router should deliver out-of-order turn events to the matching queues.""" client = CodexClient() diff --git a/sdk/python/tests/test_goal_turns.py b/sdk/python/tests/test_goal_turns.py new file mode 100644 index 0000000000..039d8ffc80 --- /dev/null +++ b/sdk/python/tests/test_goal_turns.py @@ -0,0 +1,598 @@ +import asyncio +import threading +from dataclasses import dataclass + +import pytest + +from openai_codex.api import AsyncTurnHandle, TurnHandle +from openai_codex.client import CodexClient +from openai_codex.errors import InvalidRequestError +from openai_codex.generated.notification_registry import notification_turn_id +from openai_codex.generated.v2_all import ( + TurnCompletedNotification, + TurnInterruptResponse, + TurnStatus, + TurnSteerResponse, +) + + +def _route(client: CodexClient, method: str, params: dict[str, object]) -> None: + client._router.route_notification(client._coerce_notification(method, params)) + + +def _turn(turn_id: str, status: str, *, started_at: int, completed_at: int | None = None): + turn: dict[str, object] = { + "id": turn_id, + "items": [], + "startedAt": started_at, + "status": status, + } + if completed_at is not None: + turn["completedAt"] = completed_at + return turn + + +def _goal(thread_id: str, status: str) -> dict[str, object]: + return { + "createdAt": 1, + "objective": "Improve benchmark coverage", + "status": status, + "threadId": thread_id, + "timeUsedSeconds": 0, + "tokensUsed": 0, + "updatedAt": 1, + } + + +def test_goal_turn_collects_continuations_as_one_logical_turn() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) + + _route( + client, + "thread/goal/updated", + {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, + ) + _route( + client, + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, + ) + _route( + client, + "item/completed", + { + "threadId": "thread-1", + "turnId": "turn-1", + "item": {"id": "message-1", "type": "agentMessage", "text": "working"}, + }, + ) + _route( + client, + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-1", "completed", started_at=10, completed_at=12), + }, + ) + _route( + client, + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-2", "inProgress", started_at=13)}, + ) + _route( + client, + "item/completed", + { + "threadId": "thread-1", + "turnId": "turn-2", + "item": { + "id": "message-2", + "type": "agentMessage", + "text": "done", + "phase": "final_answer", + }, + }, + ) + _route( + client, + "thread/goal/updated", + {"threadId": "thread-1", "turnId": "turn-2", "goal": _goal("thread-1", "complete")}, + ) + _route( + client, + "thread/tokenUsage/updated", + { + "threadId": "thread-1", + "turnId": "turn-2", + "tokenUsage": { + "last": { + "cachedInputTokens": 1, + "inputTokens": 2, + "outputTokens": 3, + "reasoningOutputTokens": 4, + "totalTokens": 9, + }, + "total": { + "cachedInputTokens": 10, + "inputTokens": 20, + "outputTokens": 30, + "reasoningOutputTokens": 40, + "totalTokens": 90, + }, + }, + }, + ) + _route( + client, + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-2", "completed", started_at=13, completed_at=15), + }, + ) + + result = handle.run() + + assert { + "id": result.id, + "status": result.status, + "started_at": result.started_at, + "completed_at": result.completed_at, + "duration_ms": result.duration_ms, + "final_response": result.final_response, + "item_count": len(result.items), + "total_tokens": result.usage.total.total_tokens if result.usage is not None else None, + "registered_goals": client._router._goal_operations, + } == { + "id": "turn-1", + "status": TurnStatus.completed, + "started_at": 10, + "completed_at": 15, + "duration_ms": 5000, + "final_response": "done", + "item_count": 2, + "total_tokens": 90, + "registered_goals": {}, + } + + +def test_goal_stream_hides_continuation_boundaries_and_rewrites_ids() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) + + for method, params in [ + ( + "thread/goal/updated", + {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, + ), + ( + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, + ), + ( + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), + }, + ), + ( + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-2", "inProgress", started_at=12)}, + ), + ( + "thread/goal/updated", + {"threadId": "thread-1", "turnId": "turn-2", "goal": _goal("thread-1", "complete")}, + ), + ( + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-2", "completed", started_at=12, completed_at=13), + }, + ), + ]: + _route(client, method, params) + + events = list(handle.stream()) + lifecycle = [event for event in events if event.method in {"turn/started", "turn/completed"}] + + assert { + "lifecycle": [event.method for event in lifecycle], + "turn_ids": [ + turn_id + for event in events + if (turn_id := notification_turn_id(event.payload)) is not None + ], + "completed": [ + event.payload.turn.status + for event in events + if isinstance(event.payload, TurnCompletedNotification) + ], + } == { + "lifecycle": ["turn/started", "turn/completed"], + "turn_ids": ["turn-1", "turn-1"], + "completed": [TurnStatus.completed], + } + + +def test_goal_router_releases_and_wakes_operations_on_transport_failure() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + failure = RuntimeError("transport closed") + + client._router.fail_all(failure) + + with pytest.raises(RuntimeError, match="transport closed"): + state.next_notification() + assert client._router._goal_operations == {} + + +def test_closing_unstarted_goal_stream_releases_route_and_controls() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) + + stream = handle.stream() + stream.close() + + assert { + "active_turn": state.active_turn(), + "registered_goals": client._router._goal_operations, + } == {"active_turn": None, "registered_goals": {}} + + +def test_failed_goal_start_propagates_turn_error() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) + failed_turn = _turn("turn-1", "failed", started_at=10, completed_at=11) + failed_turn["error"] = {"message": "failed to start goal"} + + _route( + client, + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, + ) + _route( + client, + "turn/completed", + {"threadId": "thread-1", "turn": failed_turn}, + ) + + with pytest.raises(RuntimeError, match="failed to start goal"): + handle.run() + assert client._router._goal_operations == {} + + +class _SyncControlClient: + def __init__(self, *, fail_first: bool = True) -> None: + self.steer_ids: list[str] = [] + self.interrupt_ids: list[str] = [] + self.paused = False + self.fail_first = fail_first + + def turn_steer(self, _thread_id, turn_id, _input): + self.steer_ids.append(turn_id) + if self.fail_first and len(self.steer_ids) == 1: + raise InvalidRequestError( + -32600, + "expected active turn id `turn-1` but found `turn-2`", + ) + return TurnSteerResponse(turn_id=turn_id) + + def pause_goal(self, _thread_id): + self.paused = True + + def turn_interrupt(self, _thread_id, turn_id): + self.interrupt_ids.append(turn_id) + if self.fail_first and len(self.interrupt_ids) == 1: + raise InvalidRequestError( + -32600, + "expected active turn id turn-1 but found turn-2", + ) + return TurnInterruptResponse() + + +def test_goal_controls_hide_rollover_turn_ids() -> None: + state = CodexClient().register_goal_operation("thread-1") + state.bind("turn-1") + client = _SyncControlClient() + handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + + steered = handle.steer("keep going") + interrupted = handle.interrupt() + + assert { + "steer_ids": client.steer_ids, + "public_steer_id": steered.turn_id, + "paused": client.paused, + "interrupt_ids": client.interrupt_ids, + "interrupt": interrupted.model_dump(mode="json"), + } == { + "steer_ids": ["turn-1", "turn-2"], + "public_steer_id": "turn-1", + "paused": True, + "interrupt_ids": ["turn-1", "turn-2"], + "interrupt": {}, + } + + +def test_finished_goal_controls_match_inactive_turn_errors() -> None: + state = CodexClient().register_goal_operation("thread-1") + state.bind("turn-1") + state.finish() + client = _SyncControlClient(fail_first=False) + handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + + with pytest.raises(InvalidRequestError, match="no active turn to steer"): + handle.steer("keep going") + with pytest.raises(InvalidRequestError, match="no active turn to interrupt"): + handle.interrupt() + assert client.paused is False + + +def test_goal_steer_waits_for_rollover_turn() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + control = _SyncControlClient(fail_first=False) + handle = TurnHandle(control, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + results: list[TurnSteerResponse] = [] + + _route( + client, + "thread/goal/updated", + {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, + ) + _route( + client, + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, + ) + _route( + client, + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), + }, + ) + + steering = threading.Thread(target=lambda: results.append(handle.steer("continue"))) + steering.start() + _route( + client, + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-2", "inProgress", started_at=12)}, + ) + steering.join(timeout=5) + + assert { + "alive": steering.is_alive(), + "steer_ids": control.steer_ids, + "public_turn_id": results[0].turn_id, + } == { + "alive": False, + "steer_ids": ["turn-2"], + "public_turn_id": "turn-1", + } + + +def test_goal_interrupt_succeeds_during_rollover() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + control = _SyncControlClient() + handle = TurnHandle(control, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + _route( + client, + "thread/goal/updated", + {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, + ) + _route( + client, + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, + ) + _route( + client, + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), + }, + ) + + response = handle.interrupt() + + assert { + "paused": control.paused, + "interrupt_ids": control.interrupt_ids, + "response": response.model_dump(mode="json"), + "interrupted": state.interrupted, + } == { + "paused": True, + "interrupt_ids": [], + "response": {}, + "interrupted": True, + } + + +class _AsyncGoalClient: + def __init__(self, state) -> None: + self.state = state + self.unregistered = False + + async def next_goal_notification(self, state): + return await asyncio.to_thread(state.next_notification) + + def unregister_goal_operation(self, _state) -> None: + self.unregistered = True + + +@dataclass(slots=True) +class _AsyncCodexStub: + _client: object + + async def _ensure_initialized(self) -> None: + return None + + +def test_async_goal_stream_matches_sync_logical_lifecycle() -> None: + async def scenario() -> None: + client = CodexClient() + state = client.register_goal_operation("thread-1") + state.bind("turn-1") + async_client = _AsyncGoalClient(state) + codex = _AsyncCodexStub(async_client) + handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + + for method, params in [ + ( + "thread/goal/updated", + { + "threadId": "thread-1", + "turnId": "turn-1", + "goal": _goal("thread-1", "active"), + }, + ), + ( + "turn/started", + {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, + ), + ( + "thread/goal/updated", + { + "threadId": "thread-1", + "turnId": "turn-1", + "goal": _goal("thread-1", "complete"), + }, + ), + ( + "turn/completed", + { + "threadId": "thread-1", + "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), + }, + ), + ]: + _route(client, method, params) + + events = [event async for event in handle.stream()] + + assert { + "methods": [event.method for event in events], + "ids": [notification_turn_id(event.payload) for event in events], + "unregistered": async_client.unregistered, + } == { + "methods": ["turn/started", "turn/completed"], + "ids": ["turn-1", "turn-1"], + "unregistered": True, + } + + asyncio.run(scenario()) + + +def test_cancelling_async_goal_stream_wakes_notification_reader() -> None: + async def scenario() -> None: + state = CodexClient().register_goal_operation("thread-1") + state.bind("turn-1") + async_client = _AsyncGoalClient(state) + codex = _AsyncCodexStub(async_client) + handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + stream = handle.stream() + reading = asyncio.create_task(anext(stream)) + await asyncio.sleep(0) + + reading.cancel() + with pytest.raises(asyncio.CancelledError): + await reading + + assert {"finished": state.is_finished(), "unregistered": async_client.unregistered} == { + "finished": True, + "unregistered": True, + } + + asyncio.run(scenario()) + + +class _AsyncControlClient: + def __init__(self) -> None: + self.steer_ids: list[str] = [] + self.interrupt_ids: list[str] = [] + self.paused = False + + async def turn_steer(self, _thread_id, turn_id, _input): + self.steer_ids.append(turn_id) + if len(self.steer_ids) == 1: + raise InvalidRequestError( + -32600, + "expected active turn id `turn-1` but found `turn-2`", + ) + return TurnSteerResponse(turn_id=turn_id) + + async def pause_goal(self, _thread_id): + self.paused = True + + async def turn_interrupt(self, _thread_id, turn_id): + self.interrupt_ids.append(turn_id) + if len(self.interrupt_ids) == 1: + raise InvalidRequestError( + -32600, + "expected active turn id turn-1 but found turn-2", + ) + return TurnInterruptResponse() + + +def test_async_goal_controls_hide_rollover_turn_ids() -> None: + async def scenario() -> None: + state = CodexClient().register_goal_operation("thread-1") + state.bind("turn-1") + client = _AsyncControlClient() + codex = _AsyncCodexStub(client) # type: ignore[arg-type] + handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + + steered = await handle.steer("keep going") + interrupted = await handle.interrupt() + + assert { + "steer_ids": client.steer_ids, + "public_steer_id": steered.turn_id, + "paused": client.paused, + "interrupt_ids": client.interrupt_ids, + "interrupt": interrupted.model_dump(mode="json"), + } == { + "steer_ids": ["turn-1", "turn-2"], + "public_steer_id": "turn-1", + "paused": True, + "interrupt_ids": ["turn-1", "turn-2"], + "interrupt": {}, + } + + asyncio.run(scenario()) + + +def test_finished_async_goal_controls_match_inactive_turn_errors() -> None: + async def scenario() -> None: + state = CodexClient().register_goal_operation("thread-1") + state.bind("turn-1") + state.finish() + client = _AsyncControlClient() + codex = _AsyncCodexStub(client) + handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] + + with pytest.raises(InvalidRequestError, match="no active turn to steer"): + await handle.steer("keep going") + with pytest.raises(InvalidRequestError, match="no active turn to interrupt"): + await handle.interrupt() + assert client.paused is False + + asyncio.run(scenario()) diff --git a/sdk/python/tests/test_public_api_signatures.py b/sdk/python/tests/test_public_api_signatures.py index c26ac90f6b..b735e41733 100644 --- a/sdk/python/tests/test_public_api_signatures.py +++ b/sdk/python/tests/test_public_api_signatures.py @@ -185,6 +185,19 @@ def test_turn_input_methods_accept_string_shortcut() -> None: ) +def test_turn_result_producers_expose_goal_mode() -> None: + """Goal mode belongs to APIs that produce a turn result or handle.""" + funcs = [Thread.run, Thread.turn, AsyncThread.run, AsyncThread.turn] + + assert { + fn: ( + inspect.signature(fn).parameters["goal"].annotation, + inspect.signature(fn).parameters["goal"].default, + ) + for fn in funcs + } == dict.fromkeys(funcs, ("bool", False)) + + def test_root_exports_approval_mode() -> None: """The root package should expose the high-level approval mode enum.""" assert [(mode.name, mode.value) for mode in ApprovalMode] == [ @@ -370,6 +383,7 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "thread_source", ], Thread.turn: [ + "goal", "approval_mode", "cwd", "effort", @@ -381,6 +395,7 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "summary", ], Thread.run: [ + "goal", "approval_mode", "cwd", "effort", @@ -445,6 +460,7 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "thread_source", ], AsyncThread.turn: [ + "goal", "approval_mode", "cwd", "effort", @@ -456,6 +472,7 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "summary", ], AsyncThread.run: [ + "goal", "approval_mode", "cwd", "effort",