diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 458e63efca..156379baf5 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4409,6 +4409,17 @@ }, "ThreadGoalClearParams": { "properties": { + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "threadId": { "type": "string" } @@ -4429,6 +4440,14 @@ ], "type": "object" }, + "ThreadGoalMutationOrigin": { + "description": "Distinguishes explicit user actions from automatic goal lifecycle mutations.", + "enum": [ + "user", + "automatic" + ], + "type": "string" + }, "ThreadGoalSetParams": { "properties": { "objective": { @@ -4437,6 +4456,17 @@ "null" ] }, + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "status": { "anyOf": [ { 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 663cc473da..04ae2b9286 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 @@ -21560,6 +21560,17 @@ "ThreadGoalClearParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/v2/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "threadId": { "type": "string" } @@ -21626,6 +21637,14 @@ "title": "ThreadGoalGetResponse", "type": "object" }, + "ThreadGoalMutationOrigin": { + "description": "Distinguishes explicit user actions from automatic goal lifecycle mutations.", + "enum": [ + "user", + "automatic" + ], + "type": "string" + }, "ThreadGoalSetParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { @@ -21635,6 +21654,17 @@ "null" ] }, + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/v2/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "status": { "anyOf": [ { 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 8856accc95..fcae8cbdd3 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 @@ -19274,6 +19274,17 @@ "ThreadGoalClearParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "threadId": { "type": "string" } @@ -19340,6 +19351,14 @@ "title": "ThreadGoalGetResponse", "type": "object" }, + "ThreadGoalMutationOrigin": { + "description": "Distinguishes explicit user actions from automatic goal lifecycle mutations.", + "enum": [ + "user", + "automatic" + ], + "type": "string" + }, "ThreadGoalSetParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { @@ -19349,6 +19368,17 @@ "null" ] }, + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "status": { "anyOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalClearParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalClearParams.json index 99b32a7316..ffef6d9d32 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalClearParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalClearParams.json @@ -1,6 +1,27 @@ { "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "ThreadGoalMutationOrigin": { + "description": "Distinguishes explicit user actions from automatic goal lifecycle mutations.", + "enum": [ + "user", + "automatic" + ], + "type": "string" + } + }, "properties": { + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "threadId": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalSetParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalSetParams.json index 0087827c21..b1aad0e236 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalSetParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadGoalSetParams.json @@ -1,6 +1,14 @@ { "$schema": "http://json-schema.org/draft-07/schema#", "definitions": { + "ThreadGoalMutationOrigin": { + "description": "Distinguishes explicit user actions from automatic goal lifecycle mutations.", + "enum": [ + "user", + "automatic" + ], + "type": "string" + }, "ThreadGoalStatus": { "enum": [ "active", @@ -20,6 +28,17 @@ "null" ] }, + "origin": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadGoalMutationOrigin" + }, + { + "type": "null" + } + ], + "description": "Missing provenance does not supply user authorization." + }, "status": { "anyOf": [ { diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 2c53aa1783..9e6ab232b7 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 1efda781c9..6de2c0c546 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalClearParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalClearParams.ts index efefc2538a..1e944736d4 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalClearParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalClearParams.ts @@ -1,5 +1,10 @@ // GENERATED CODE! DO NOT MODIFY BY HAND! // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadGoalMutationOrigin } from "./ThreadGoalMutationOrigin"; -export type ThreadGoalClearParams = { threadId: string, }; +export type ThreadGoalClearParams = { threadId: string, +/** + * Missing provenance does not supply user authorization. + */ +origin?: ThreadGoalMutationOrigin | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalMutationOrigin.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalMutationOrigin.ts new file mode 100644 index 0000000000..742419cba3 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalMutationOrigin.ts @@ -0,0 +1,8 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Distinguishes explicit user actions from automatic goal lifecycle mutations. + */ +export type ThreadGoalMutationOrigin = "user" | "automatic"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalSetParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalSetParams.ts index b92720c178..d6edcbddfd 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalSetParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadGoalSetParams.ts @@ -1,6 +1,11 @@ // GENERATED CODE! DO NOT MODIFY BY HAND! // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadGoalMutationOrigin } from "./ThreadGoalMutationOrigin"; import type { ThreadGoalStatus } from "./ThreadGoalStatus"; -export type ThreadGoalSetParams = { threadId: string, objective?: string | null, status?: ThreadGoalStatus | null, tokenBudget?: number | null, }; +export type ThreadGoalSetParams = { threadId: string, +/** + * Missing provenance does not supply user authorization. + */ +origin?: ThreadGoalMutationOrigin | null, objective?: string | null, status?: ThreadGoalStatus | null, tokenBudget?: number | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index b96f29ce44..24dc8f30a7 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -505,6 +505,7 @@ export type { ThreadGoalClearResponse } from "./ThreadGoalClearResponse"; export type { ThreadGoalClearedNotification } from "./ThreadGoalClearedNotification"; export type { ThreadGoalGetParams } from "./ThreadGoalGetParams"; export type { ThreadGoalGetResponse } from "./ThreadGoalGetResponse"; +export type { ThreadGoalMutationOrigin } from "./ThreadGoalMutationOrigin"; export type { ThreadGoalSetParams } from "./ThreadGoalSetParams"; export type { ThreadGoalSetResponse } from "./ThreadGoalSetResponse"; export type { ThreadGoalStatus } from "./ThreadGoalStatus"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index c6d6fad791..1aec8bbd45 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -2522,6 +2522,7 @@ mod tests { let thread_goal_set = ClientRequest::ThreadGoalSet { request_id: request_id(), params: v2::ThreadGoalSetParams { + origin: None, thread_id: "goal-thread".to_string(), objective: Some("ship it".to_string()), status: None, @@ -4440,6 +4441,7 @@ mod tests { let set_request = ClientRequest::ThreadGoalSet { request_id: RequestId::Integer(1), params: v2::ThreadGoalSetParams { + origin: None, thread_id: "thr_123".to_string(), objective: Some("ship goal mode".to_string()), status: Some(v2::ThreadGoalStatus::Active), @@ -4455,6 +4457,7 @@ mod tests { let clear_request = ClientRequest::ThreadGoalClear { request_id: RequestId::Integer(3), params: v2::ThreadGoalClearParams { + origin: None, thread_id: "thr_123".to_string(), }, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 2e166d5506..9eee37dcac 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -841,11 +841,24 @@ impl From for ThreadGoal { } } +/// Distinguishes explicit user actions from automatic goal lifecycle mutations. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum ThreadGoalMutationOrigin { + User, + Automatic, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadGoalSetParams { pub thread_id: String, + /// Missing provenance does not supply user authorization. + #[ts(optional = nullable)] + pub origin: Option, #[ts(optional = nullable)] pub objective: Option, #[ts(optional = nullable)] @@ -886,6 +899,9 @@ pub struct ThreadGoalGetResponse { #[ts(export_to = "v2/")] pub struct ThreadGoalClearParams { pub thread_id: String, + /// Missing provenance does not supply user authorization. + #[ts(optional = nullable)] + pub origin: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index 6d52f5b52f..bada44de51 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -2,6 +2,8 @@ use super::thread_input::DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR; use super::thread_input::can_accept_direct_input; use super::thread_input::ensure_direct_input_allowed; use super::*; +use codex_app_server_protocol::ThreadGoalMutationOrigin; +use codex_core::context::UserGoalUpdate; use codex_goal_extension::GoalObjectiveUpdate; use codex_goal_extension::GoalService; use codex_goal_extension::GoalServiceError; @@ -9,9 +11,11 @@ use codex_goal_extension::GoalSetRequest; use codex_goal_extension::GoalTokenBudgetUpdate; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; -use codex_protocol::protocol::ThreadSettingsAppliedEvent; -use codex_protocol::protocol::ThreadSettingsSnapshot; use codex_rollout::RolloutRecorder; +use codex_rollout::WriterLockCoordinator; + +#[path = "thread_goal_user_context.rs"] +mod user_context; enum GoalAccess { Read, @@ -27,6 +31,7 @@ pub(crate) struct ThreadGoalRequestProcessor { state_db: Option, goal_service: Arc, config_manager: ConfigManager, + writer_locks: Arc, } impl ThreadGoalRequestProcessor { @@ -39,6 +44,7 @@ impl ThreadGoalRequestProcessor { goal_service: Arc, config_manager: ConfigManager, ) -> Self { + let writer_locks = Arc::new(WriterLockCoordinator::new(&config.codex_home)); Self { thread_manager, outgoing, @@ -47,6 +53,7 @@ impl ThreadGoalRequestProcessor { state_db, goal_service, config_manager, + writer_locks, } } @@ -133,6 +140,7 @@ impl ThreadGoalRequestProcessor { } let thread_id = parse_thread_id_for_request(params.thread_id.as_str())?; + let _goal_resume_guard = self.thread_state_manager.lock_goal_resume(thread_id).await; let state_db = self .state_db_for_materialized_thread(thread_id, GoalAccess::Mutate) .await?; @@ -169,7 +177,7 @@ impl ThreadGoalRequestProcessor { thread_state.listener_command_tx() }; let status = params.status.map(ThreadGoalStatus::to_core); - let objective = params.objective.as_deref(); + let objective = params.objective.as_deref().map(str::trim); let outcome = self .goal_service @@ -187,45 +195,44 @@ impl ThreadGoalRequestProcessor { }, max_goal_token_budget, }, + Box::pin(async { + if params.origin == Some(ThreadGoalMutationOrigin::User) + && (objective.is_some() || status.is_some()) + { + self.record_user_goal_update( + thread_id, + UserGoalUpdate::Set { + objective: objective.map(str::to_owned), + status, + }, + ) + .await?; + } + Ok(()) + }), ) .await .map_err(goal_service_error)?; let goal = ThreadGoal::from(outcome.goal.clone()); let persist_result = match self.thread_manager.get_thread(thread_id).await { - Ok(thread) => match thread.rollout_path() { - Some(path) if codex_rollout::existing_rollout_path(&path).await.is_none() => { - // Goal-first threads need their settings captured when the goal creates the - // rollout. Once materialized, normal settings updates own this event. - let persisted_settings = thread.thread_settings_snapshot().await; - let items = [ - thread_settings_applied_item(thread_id, persisted_settings.clone()), - outcome.thread_goal_updated_item(), - ]; - match thread.append_rollout_items(&items).await { - Err(err) => Err(err), - Ok(()) => { - // Catch up a settings update queued while the rollout materialized. - let current_settings = thread.thread_settings_snapshot().await; - if current_settings == persisted_settings { - Ok(()) - } else { - thread - .append_rollout_items(&[thread_settings_applied_item( - thread_id, - current_settings, - )]) - .await - } - } + Ok(thread) => { + let needs_settings_checkpoint = match thread.rollout_path() { + Some(path) => codex_rollout::existing_rollout_path(&path).await.is_none(), + None => false, + }; + match thread + .append_rollout_items(&[outcome.thread_goal_updated_item()]) + .await + { + // Automatic and legacy callers still need goal-first settings persisted, + // even though they do not record a user instruction. + Ok(()) if needs_settings_checkpoint => { + thread.checkpoint_thread_settings().await } + result => result, } - Some(_) | None => { - thread - .append_rollout_items(&[outcome.thread_goal_updated_item()]) - .await - } - }, + } Err(_) => Ok(()), }; if let Err(err) = persist_result { @@ -275,6 +282,7 @@ impl ThreadGoalRequestProcessor { } let thread_id = parse_thread_id_for_request(params.thread_id.as_str())?; + let _goal_resume_guard = self.thread_state_manager.lock_goal_resume(thread_id).await; let state_db = self .state_db_for_materialized_thread(thread_id, GoalAccess::Mutate) .await?; @@ -288,7 +296,17 @@ impl ThreadGoalRequestProcessor { }; let cleared = self .goal_service - .clear_thread_goal(&state_db, thread_id) + .clear_thread_goal( + &state_db, + thread_id, + Box::pin(async { + if params.origin == Some(ThreadGoalMutationOrigin::User) { + self.record_user_goal_update(thread_id, UserGoalUpdate::Clear) + .await?; + } + Ok(()) + }), + ) .await .map_err(goal_service_error)?; @@ -504,18 +522,6 @@ impl ThreadGoalRequestProcessor { } } -fn thread_settings_applied_item( - thread_id: ThreadId, - thread_settings: ThreadSettingsSnapshot, -) -> RolloutItem { - RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied( - ThreadSettingsAppliedEvent { - thread_id: Some(thread_id), - thread_settings, - }, - )) -} - pub(super) fn api_thread_goal_from_state(goal: codex_state::ThreadGoal) -> ThreadGoal { ThreadGoal { thread_id: goal.thread_id.to_string(), diff --git a/codex-rs/app-server/src/request_processors/thread_goal_user_context.rs b/codex-rs/app-server/src/request_processors/thread_goal_user_context.rs new file mode 100644 index 0000000000..c27c4a21ce --- /dev/null +++ b/codex-rs/app-server/src/request_processors/thread_goal_user_context.rs @@ -0,0 +1,51 @@ +//! Persists explicit user goal mutations with user provenance before goal work resumes. +//! Unloaded threads receive the same item on replay; tool-created goals never use this path. + +use super::*; +use codex_core::context::ContextualUserFragment; +use codex_core::context::UserGoalUpdate; +use codex_rollout::RolloutRecorderParams; + +impl ThreadGoalRequestProcessor { + pub(super) async fn record_user_goal_update( + &self, + thread_id: ThreadId, + update: UserGoalUpdate, + ) -> Result<(), GoalServiceError> { + if let Ok(thread) = self.thread_manager.get_thread(thread_id).await { + return thread.record_user_goal_update(update).await.map_err(|err| { + GoalServiceError::Internal(format!("failed to record goal instruction: {err}")) + }); + } + let item = ContextualUserFragment::into(update); + let record_error = + |err| GoalServiceError::Internal(format!("failed to record goal instruction: {err}")); + let writer_lock = Arc::new(self.writer_locks.acquire(thread_id).map_err(record_error)?); + let path = codex_rollout::find_thread_path_by_id_str( + &self.config.codex_home, + &thread_id.to_string(), + self.state_db.as_deref(), + ) + .await + .map_err(|err| { + GoalServiceError::Internal(format!("failed to locate thread id {thread_id}: {err}")) + })? + .ok_or_else(|| { + GoalServiceError::InvalidRequest(format!("thread not found: {thread_id}")) + })?; + // The writer owns the lock through background IO, even if this RPC is cancelled. + let recorder = RolloutRecorder::new_with_writer_lock( + self.config.as_ref(), + RolloutRecorderParams::resume(path), + writer_lock, + ) + .await + .map_err(record_error)?; + recorder + .record_canonical_items(&[RolloutItem::ResponseItem(item.into())]) + .await + .map_err(record_error)?; + // Shutdown flushes queued items before acknowledging the goal mutation. + recorder.shutdown().await.map_err(record_error) + } +} diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 289cb63953..e3ce2b0b22 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3749,6 +3749,17 @@ impl ThreadRequestProcessor { } }; let (thread_history, resume_source_thread) = resume_result?; + // Path-based resume can use an empty request thread ID. Coordinate once its real + // identity is known; unrelated loaded threads never wait for this cold startup. + let _goal_resume_guard = if let InitialHistory::Resumed(resumed) = &thread_history { + Some( + self.thread_state_manager + .lock_goal_resume(resumed.conversation_id) + .await, + ) + } else { + None + }; if let InitialHistory::Resumed(resumed) = &thread_history && self .pending_thread_unloads @@ -3948,7 +3959,8 @@ impl ThreadRequestProcessor { let mut config = match prepared_config.take() { Some(prepared) if prepared.state == config_state => prepared.config, _ => { - // Config loading can call back into Desktop; release the permit during host work. + // Config loading can call back into Desktop; release both locks during host work. + drop(_goal_resume_guard); drop(_thread_list_state_permit); let config = self .config_manager diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 8cfbc00983..a19c41b19f 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -99,6 +99,7 @@ pub(crate) struct TurnSummary { #[derive(Default)] pub(crate) struct ThreadState { + goal_resume_lock: Arc>, pub(crate) pending_interrupts: PendingInterruptQueue, pub(crate) turn_summary: TurnSummary, pub(crate) last_terminal_turn_id: Option, @@ -361,6 +362,21 @@ pub(crate) struct ThreadStateManager { } impl ThreadStateManager { + /// Coordinates goal edits with cold/path-based resume for just this thread. + pub(crate) async fn lock_goal_resume( + &self, + thread_id: ThreadId, + ) -> tokio::sync::OwnedMutexGuard<()> { + let lock = self + .thread_state(thread_id) + .await + .lock() + .await + .goal_resume_lock + .clone(); + lock.lock_owned().await + } + pub(crate) fn new() -> Self { Self::default() } diff --git a/codex-rs/app-server/tests/suite/v2/guardian_goal_tests.rs b/codex-rs/app-server/tests/suite/v2/guardian_goal_tests.rs new file mode 100644 index 0000000000..358114e4de --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/guardian_goal_tests.rs @@ -0,0 +1,131 @@ +//! Goal API edits and clears enter model history across loaded and cold-resume paths. + +use super::*; +use codex_app_server_protocol::ThreadGoalSetResponse; +use pretty_assertions::assert_eq; +use test_case::test_case; + +#[test_case(false; "loaded")] +#[test_case(true; "unloaded with path resume")] +#[tokio::test] +async fn user_goal_updates_survive_resume_and_clear(unloaded: bool) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let mock = responses::mount_sse_once( + &server, + responses::sse(vec![responses::ev_completed("done")]), + ) + .await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .with_provider_config("supports_websockets = false") + .enable_feature(Feature::Goals) + .disable_feature(Feature::EnableRequestCompression) + .write(codex_home.path())?; + let mut app = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build_initialized_with_timeout(TIMEOUT) + .await?; + let thread_id = if unloaded { + create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "", + Some("mock_provider"), + /*git_info*/ None, + )? + } else { + app.start_thread(ThreadStartParams::default()) + .await? + .thread + .id + }; + // Automatic lifecycle calls mutate goal state without creating user authorization. + let id = app + .send_raw_request( + "thread/goal/set", + Some(json!({ + "threadId": thread_id, "objective": "Automatically generated goal.", + "status": "paused", "origin": "automatic", + })), + ) + .await?; + let _: ThreadGoalSetResponse = app.read_response(id).await?; + let mut resume_request = None; + for objective in [ + "Send the approved report.", + "Draft the report, but do not send it.", + ] { + if unloaded && objective.starts_with("Draft") { + let path = codex_rollout::find_thread_path_by_id_str( + codex_home.path(), + &thread_id, + /*state_db_ctx*/ None, + ) + .await? + .expect("unloaded rollout"); + resume_request = Some( + app.send_thread_resume_request(ThreadResumeParams { + thread_id: String::new(), + path: Some(path), + ..Default::default() + }) + .await?, + ); + } + let id = app + .send_raw_request( + "thread/goal/set", + Some(json!({ + "threadId": thread_id, "objective": objective, "status": "paused", "origin": "user", + })), + ) + .await?; + let _: ThreadGoalSetResponse = app.read_response(id).await?; + } + if let Some(id) = resume_request { + let _: ThreadResumeResponse = app.read_response(id).await?; + } + let id = app + .send_raw_request( + "thread/goal/clear", + Some(json!({"threadId": thread_id, "origin": "user"})), + ) + .await?; + let _: codex_app_server_protocol::ThreadGoalClearResponse = app.read_response(id).await?; + let id = app + .send_turn_start_request(TurnStartParams { + thread_id, + input: vec![UserInput::Text { + text: "What is the status?".to_owned(), + text_elements: vec![], + }], + ..Default::default() + }) + .await?; + let _: TurnStartResponse = app.read_response(id).await?; + let completed: TurnCompletedNotification = + timeout(TIMEOUT, app.read_notification("turn/completed")).await??; + assert_eq!(completed.turn.status, TurnStatus::Completed); + let history = mock.single_request().message_input_texts("user").join("\n"); + assert!( + !history.contains("Automatically generated goal."), + "{history}" + ); + let old = history + .find("User set the goal: \"Send the approved report.\"") + .expect("original user goal"); + let new = history + .find("User set the goal: \"Draft the report, but do not send it.\"") + .expect("updated user goal"); + assert!(old < new, "{history}"); + assert!( + history.contains("User set goal status: \"paused\"."), + "{history}" + ); + assert!(history.contains("User cleared the goal."), "{history}"); + app.shutdown_gracefully().await?; + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs index b769a1682c..b084707a4e 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs @@ -78,6 +78,9 @@ use super::mcp_tool::start_mcp_server_with_tools; #[path = "guardian_sync_session_tests.rs"] mod sync_sessions; +#[path = "guardian_goal_tests.rs"] +mod goals; + #[path = "guardian_v2_history_tests.rs"] mod history; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index c54fd5454d..7a00247221 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -755,8 +755,7 @@ impl CodexThread { &self, update: crate::context::UserGoalUpdate, ) -> CodexResult<()> { - self.session.record_user_goal_update(update).await; - self.checkpoint_preparation().await?; + self.session.record_user_goal_update(update).await?; Ok(()) } diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index a937b39db0..6de37b1e0a 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -1,3 +1,5 @@ +use crate::context::UserGoalUpdate; +use std::collections::HashSet; use std::sync::Arc; use std::time::Instant; @@ -30,6 +32,7 @@ use codex_analytics::CompactionTrigger; use codex_analytics::now_unix_seconds; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; +use codex_protocol::ResponseItemId; use codex_protocol::error::CodexErr; use codex_protocol::error::CodexErrorDetails; use codex_protocol::error::Result as CodexResult; @@ -76,6 +79,7 @@ pub(crate) enum InitialContextInjection { /// `Session::replace_compacted_history` assigns missing item IDs before constructing the persisted /// `CompactedItem`, ensuring the live and persisted histories remain identical. pub(crate) struct CompactedHistoryMetadata { + pub(crate) input_goal_ids: HashSet, pub(crate) message: String, pub(crate) window_number: u64, pub(crate) window_ids: AutoCompactWindowIds, @@ -255,6 +259,7 @@ async fn run_compact_task_inner_impl( let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input); let mut history = sess.clone_history().await; + let input_goal_ids = UserGoalUpdate::message_ids(history.raw_items()); history.record_items( &[initial_input_for_turn.into()], turn_context.model_info().truncation_policy.into(), @@ -381,6 +386,7 @@ async fn run_compact_task_inner_impl( reference_context_item, world_state_baseline, CompactedHistoryMetadata { + input_goal_ids, message: summary_text, window_number, window_ids, diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 013a4faca4..56e4aace1b 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -293,6 +293,7 @@ async fn run_remote_compact_task_inner_impl( } }; let RemoteCompactV2Attempt { + input_goal_ids, trace_input_history, prompt_input, prompt_input_metadata, @@ -361,6 +362,7 @@ async fn run_remote_compact_task_inner_impl( reference_context_item, world_state_baseline, CompactedHistoryMetadata { + input_goal_ids, message: String::new(), window_number: new_window_number, window_ids: new_window_ids, diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index 60b5b1b772..6f881e2868 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -1,3 +1,6 @@ +use crate::context::UserGoalUpdate; +use codex_protocol::ResponseItemId; +use std::collections::HashSet; use std::sync::Arc; use super::RemoteCompactionV2Output; @@ -17,6 +20,7 @@ use codex_rollout_trace::CompactionTraceContext; use tracing::info; pub(super) struct RemoteCompactV2Attempt { + pub(super) input_goal_ids: HashSet, pub(super) trace_input_history: Option>, pub(super) prompt_input: Vec, pub(super) prompt_input_metadata: Vec>, @@ -37,6 +41,7 @@ pub(super) async fn run_remote_compact_v2_attempt( ) -> CodexResult { let turn_context = &step_context.turn; let mut history = sess.clone_history().await; + let input_goal_ids = UserGoalUpdate::message_ids(history.raw_items()); let base_instructions = sess.get_prompt_base_instructions().await; let (rewritten_outputs, estimated_deleted_tokens) = trim_function_call_history_to_fit_context_window( @@ -122,6 +127,7 @@ pub(super) async fn run_remote_compact_v2_attempt( let mut prompt_input = prompt.input; prompt_input.pop(); Ok(RemoteCompactV2Attempt { + input_goal_ids, trace_input_history, prompt_input, prompt_input_metadata, diff --git a/codex-rs/core/src/context/user_goal.rs b/codex-rs/core/src/context/user_goal.rs index d20504159f..76d7c1c532 100644 --- a/codex-rs/core/src/context/user_goal.rs +++ b/codex-rs/core/src/context/user_goal.rs @@ -3,10 +3,12 @@ //! Oversized objectives are omitted whole so truncation cannot turn a restriction into a grant. use super::ContextualUserFragment; +use codex_protocol::ResponseItemId; use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::ThreadGoalStatus; +use std::collections::HashSet; const MAX_OBJECTIVE_BYTES: usize = 700; @@ -23,6 +25,17 @@ pub enum UserGoalUpdate { impl UserGoalUpdate { pub(crate) const OMITTED_OBJECTIVE_KIND: &str = "user.goal.omitted"; + /// Captures accepted goal instructions at a compaction input boundary. + pub(crate) fn message_ids<'a>( + items: impl Iterator, + ) -> HashSet { + items + .filter(|item| Self::message_text(item).is_some()) + .filter_map(ResponseItem::id) + .cloned() + .collect() + } + /// Reads only a host-annotated goal instruction, never a matching text wrapper alone. pub(crate) fn message_text(item: &ResponseItem) -> Option<&str> { let ResponseItem::Message { diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 0a2eaf9445..7414760c66 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -85,6 +85,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() { /*reference_context_item*/ None, /*world_state_baseline*/ None, crate::compact::CompactedHistoryMetadata { + input_goal_ids: Default::default(), message: String::new(), window_number, window_ids, diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 5e9ec6813f..88792b6dda 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -2657,6 +2657,7 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: /*reference_context_item*/ None, /*world_state_baseline*/ None, crate::compact::CompactedHistoryMetadata { + input_goal_ids: Default::default(), message: String::new(), window_number, window_ids, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index b876976846..4708d2921e 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4055,7 +4055,6 @@ impl Session { .get_or_insert_default() .compaction_model_hash = metadata.compaction_model_hash; } - let replacement_history = items.clone(); // Wait for accepted updates to finish persisting, then keep later updates from // overtaking the current settings snapshot while its checkpoint is written. let _settings_guard = thread_settings::acquire_persistence_lock(self).await; @@ -4076,6 +4075,26 @@ impl Session { } (!retained.is_empty()).then(|| WorldStateSnapshot::from(&retained)) }); + // Goal edits are published outside the running task. Keep edits accepted after + // the compaction input snapshot, in their original order, after its replacement. + let replacement_goal_ids = crate::context::UserGoalUpdate::message_ids( + items.iter().map(|envelope| &envelope.item), + ); + items.extend( + state + .history + .annotated_items() + .iter() + .filter(|envelope| { + crate::context::UserGoalUpdate::message_text(&envelope.item).is_some() + && envelope.item.id().is_some_and(|id| { + !metadata.input_goal_ids.contains(id) + && !replacement_goal_ids.contains(id) + }) + }) + .cloned(), + ); + let replacement_history = items.clone(); state.replace_annotated_history( items, reference_context_item.clone(), @@ -4530,9 +4549,10 @@ impl Session { world_state: Arc, ) -> u64 { let turn_context = step_context.turn.as_ref(); + let history = self.clone_history().await; + let input_goal_ids = crate::context::UserGoalUpdate::message_ids(history.raw_items()); let retained_client_developer_messages = if self.enabled(Feature::RetainClientDeveloperMessages) { - let history = self.clone_history().await; crate::compact_remote_v2::truncate_retained_messages_for_remote_compaction( history .annotated_items() @@ -4565,6 +4585,7 @@ impl Session { Some(turn_context_item), Some(world_state), CompactedHistoryMetadata { + input_goal_ids, message: String::new(), window_number, window_ids, diff --git a/codex-rs/core/src/session/retained_context.rs b/codex-rs/core/src/session/retained_context.rs index c1724370e1..8c4203bf13 100644 --- a/codex-rs/core/src/session/retained_context.rs +++ b/codex-rs/core/src/session/retained_context.rs @@ -1,5 +1,6 @@ //! Orders retained inputs at acceptance, assistant messages at stream start, //! and Code Mode messages at confirmed delivery. +//! Explicit goal authorization becomes live only after its persistence barrier succeeds. use std::collections::HashMap; use std::future::Future; @@ -11,10 +12,13 @@ use std::task::Poll; use crate::context::ContextualUserFragment; use crate::context::UserGoalUpdate; +use codex_history::CodexHarnessMetadata; +use codex_history::ResponseItemEnvelope; use codex_history::RetainedContextEvent; use codex_history::RetainedUserMessage; use codex_history::RolloutItem; use codex_protocol::models::ResponseItem; +use codex_thread_store::PersistContext; use tokio::sync::Semaphore; use tokio::sync::watch; use tokio::task::JoinHandle; @@ -77,17 +81,47 @@ impl Session { } /// Records authorization without adding pending input or reopening an active turn. - pub(crate) async fn record_user_goal_update(&self, update: UserGoalUpdate) { + pub(crate) async fn record_user_goal_update( + &self, + update: UserGoalUpdate, + ) -> std::io::Result<()> { // Goal metadata must not initialize a model step, even on a goal-first thread. // Keep context construction off callers' stacks, including the TUI RPC dispatcher. let context = Box::pin(self.new_inject_items_context()).await; + // Earlier Code Mode messages may need the persistence lock before this can return. + let user_input_order = self.reserve_user_input_order().await; let _guard = thread_settings::acquire_persistence_lock(self).await; - self.record_conversation_items( - &context, - context.model_info(), - &[ContextualUserFragment::into(update)], - ) - .await; + let mut item = ContextualUserFragment::into(update); + Self::stamp_response_item_for_history(&mut item, &context.sub_id); + Self::assign_missing_response_item_id(&mut item); + let mut item = ResponseItemEnvelope { + item, + metadata: Some(CodexHarnessMetadata { + user_input_order: Some(user_input_order), + ..Default::default() + }), + }; + if let Some(live_thread) = self.live_thread() { + // Capture settings under the same permit as the instruction, including when a + // goal creates the rollout. No fallible settings catch-up runs after publication. + live_thread + .append_items(&[ + RolloutItem::EventMsg(thread_settings::applied_event(self).await), + RolloutItem::ResponseItem(item.clone()), + ]) + .await + .map_err(std::io::Error::other)?; + } + // A failed append or checkpoint must not change live authorization or its revision. + self.try_ensure_rollout_materialized(PersistContext::ThreadPreparation) + .await?; + self.state.lock().await.history.record_annotated_items( + std::slice::from_mut(&mut item), + context.model_info().truncation_policy.into(), + ); + self.send_raw_response_items(&context, std::slice::from_ref(&item.item)) + .await; + Ok(()) } pub(crate) async fn reserve_user_input_order(&self) -> u64 { diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 68a9fef1c0..966ce1f92d 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1192,6 +1192,35 @@ impl Session { e })?; let session_result: anyhow::Result> = async { + if let InitialHistory::Resumed(resumed) = &mut initial_history + // A header identifies a stored snapshot. Headerless histories are explicitly + // supplied by callers (for example, imports) and must not be replaced from disk. + && matches!( + resumed.history.first(), + Some(RolloutItem::SessionMeta(meta)) if meta.meta.id == resumed.conversation_id + ) + && let Some(live_thread) = live_thread.as_ref() + && thread_store.as_any().is::() + { + // The initial snapshot predates writer ownership. A cold goal edit may have + // completed in between, including while V1 resume_agent was loading history. + // Reload under the live writer's ownership before publishing replayed context. + let items = match session_configuration.history_mode { + ThreadHistoryMode::Legacy => { + live_thread.load_history(/*include_archived*/ true).await?.items + } + ThreadHistoryMode::Paginated => { + thread_store + .load_latest_model_context(codex_thread_store::LoadThreadHistoryParams { + thread_id, + include_archived: true, + }) + .await? + .items + } + }; + resumed.history = Arc::new(items); + } let rollout_path = if let Some(live_thread) = live_thread.as_ref() { live_thread.local_rollout_path().await? } else { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 8221619a21..ccc0bfd554 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5869,6 +5869,7 @@ async fn settings_checkpoint_waits_for_accepted_settings_persistence() { /*reference_context_item*/ None, /*world_state_baseline*/ None, CompactedHistoryMetadata { + input_goal_ids: Default::default(), message: "summary".to_string(), window_number, window_ids, @@ -6002,6 +6003,7 @@ async fn mcp_attribution_checkpoints_cover_batch_prefixes_compaction_and_restore /*reference_context_item*/ None, /*world_state_baseline*/ None, CompactedHistoryMetadata { + input_goal_ids: Default::default(), message: "summary".to_string(), window_number, window_ids, @@ -6125,7 +6127,25 @@ async fn compaction_persists_resume_metadata_and_companion_records() { thread_settings: session.thread_settings_snapshot().await, }; + let mut first_live_history = None; for with_baselines in [true, false] { + let input_goal_ids = + crate::context::UserGoalUpdate::message_ids(session.clone_history().await.raw_items()); + // The goal edit is accepted after the compaction input was captured. + let accepted_goal = if with_baselines { + session + .record_user_goal_update(crate::context::UserGoalUpdate::Clear) + .await + .expect("record concurrent clear"); + session + .clone_history() + .await + .annotated_items() + .last() + .cloned() + } else { + None + }; let (window_number, window_ids) = session.advance_auto_compact_window().await; session .replace_compacted_history( @@ -6133,6 +6153,7 @@ async fn compaction_persists_resume_metadata_and_companion_records() { with_baselines.then_some(turn_context_baseline.clone()), with_baselines.then_some(Arc::clone(&world_state)), CompactedHistoryMetadata { + input_goal_ids, message: String::new(), window_number, window_ids, @@ -6142,6 +6163,11 @@ async fn compaction_persists_resume_metadata_and_companion_records() { }, ) .await; + if let Some(accepted_goal) = accepted_goal { + let live_history = session.clone_history().await.annotated_items().to_vec(); + assert_eq!(&live_history[1..], &[accepted_goal]); + first_live_history = Some(live_history); + } } session.flush_rollout().await.expect("flush checkpoints"); @@ -6163,6 +6189,7 @@ async fn compaction_persists_resume_metadata_and_companion_records() { else { panic!("unexpected compaction records: {compaction_items:#?}"); }; + assert_eq!(first.replacement_history, first_live_history); assert_eq!(first.resume_metadata.as_ref(), Some(&expected)); assert_eq!(second.resume_metadata.as_ref(), Some(&expected)); assert_eq!( diff --git a/codex-rs/core/tests/suite/guardian_root_handoff.rs b/codex-rs/core/tests/suite/guardian_root_handoff.rs index 310f950a6c..7f8fca4328 100644 --- a/codex-rs/core/tests/suite/guardian_root_handoff.rs +++ b/codex-rs/core/tests/suite/guardian_root_handoff.rs @@ -260,13 +260,15 @@ pub(crate) async fn handoff_scenario() -> Result> { let checkpoint: CompactedItem = serde_json::from_value(json!({ "message":"Root checkpoint.", "replacement_history":[], "retained_context":history.retained_context(), }))?; + test.codex + .append_rollout_items(&[RolloutItem::Compacted(checkpoint)]) + .await?; test.codex.flush_rollout().await?; - let mut rollout = test + let rollout = test .codex .load_history(/*include_archived*/ false) .await? .items; - rollout.push(RolloutItem::Compacted(checkpoint)); test.codex = crate::suite::guardian_checkpoint_migration::resume(&test, &test.codex, rollout).await?; expected.insert( diff --git a/codex-rs/ext/goal/src/api.rs b/codex-rs/ext/goal/src/api.rs index b414e9ccfb..a2bd783dc8 100644 --- a/codex-rs/ext/goal/src/api.rs +++ b/codex-rs/ext/goal/src/api.rs @@ -141,10 +141,14 @@ impl GoalService { .map_err(|err| GoalServiceError::Internal(format!("failed to read thread goal: {err}"))) } + /// Records the user's instruction after validation and before changing goal state. + /// The recording future runs under the runtime's goal mutation permit; failure leaves + /// the goal unchanged. Recorded user intent remains valid if the database write fails. pub async fn set_thread_goal( &self, state_db: &codex_state::StateRuntime, request: GoalSetRequest<'_>, + record_user_instruction: impl Future> + Send, ) -> Result { let GoalSetRequest { thread_id, @@ -191,14 +195,21 @@ impl GoalService { tracing::warn!("failed to prepare external goal mutation: {err}"); } + let existing_goal = state_db + .thread_goals() + .get_thread_goal(thread_id) + .await + .map_err(|err| { + GoalServiceError::Internal(format!("failed to read thread goal: {err}")) + })?; + if objective.is_none() && existing_goal.is_none() { + return Err(GoalServiceError::InvalidRequest(format!( + "cannot update goal for thread {thread_id}: no goal exists" + ))); + } + record_user_instruction.await?; + let (goal, previous_goal) = if let Some(objective) = objective { - let existing_goal = state_db - .thread_goals() - .get_thread_goal(thread_id) - .await - .map_err(|err| { - GoalServiceError::Internal(format!("failed to read thread goal: {err}")) - })?; if let Some(existing_goal) = existing_goal.as_ref() { let previous_goal = PreviousGoalSnapshot::from(existing_goal); state_db @@ -238,18 +249,11 @@ impl GoalService { .map(|goal| (goal, None))? } } else { - let existing_goal = state_db - .thread_goals() - .get_thread_goal(thread_id) - .await - .map_err(|err| { - GoalServiceError::Internal(format!("failed to read thread goal: {err}")) - })? - .ok_or_else(|| { - GoalServiceError::InvalidRequest(format!( - "cannot update goal for thread {thread_id}: no goal exists" - )) - })?; + let existing_goal = existing_goal.ok_or_else(|| { + GoalServiceError::InvalidRequest(format!( + "cannot update goal for thread {thread_id}: no goal exists" + )) + })?; let previous_goal = PreviousGoalSnapshot::from(&existing_goal); let expected_goal_id = existing_goal.goal_id.clone(); state_db @@ -289,10 +293,12 @@ impl GoalService { }) } + /// Records a clear instruction even when no goal exists, then deletes goal state under the mutation permit. pub async fn clear_thread_goal( &self, state_db: &codex_state::StateRuntime, thread_id: ThreadId, + record_user_instruction: impl Future> + Send, ) -> Result { let runtime = self.runtime_for_thread(thread_id); // Hold this through the prepare/write window so idle continuation cannot @@ -312,6 +318,9 @@ impl GoalService { tracing::warn!("failed to prepare external goal mutation: {err}"); } + // A failed set can leave authorization in history without a goal row. + record_user_instruction.await?; + let cleared_goal = state_db .thread_goals() .delete_thread_goal(thread_id) diff --git a/codex-rs/ext/goal/tests/goal_extension_backend.rs b/codex-rs/ext/goal/tests/goal_extension_backend.rs index e0a952d1a8..c9a855b555 100644 --- a/codex-rs/ext/goal/tests/goal_extension_backend.rs +++ b/codex-rs/ext/goal/tests/goal_extension_backend.rs @@ -31,6 +31,7 @@ use codex_goal_extension::GoalExtensionConfig; use codex_goal_extension::GoalObjectiveUpdate; use codex_goal_extension::GoalRuntimeHandle; use codex_goal_extension::GoalService; +use codex_goal_extension::GoalServiceError; use codex_goal_extension::GoalSetRequest; use codex_goal_extension::GoalTokenBudgetUpdate; use codex_goal_extension::install_with_backend; @@ -1342,6 +1343,7 @@ async fn goal_service_external_set_active_preserves_concurrent_usage() -> anyhow token_budget: GoalTokenBudgetUpdate::Keep, max_goal_token_budget: None, }, + std::future::ready(Ok(())), ) .await?; harness @@ -1416,7 +1418,7 @@ async fn thread_stop_unregisters_goal_runtime_from_service() -> anyhow::Result<( assert!( harness .goal_service - .clear_thread_goal(runtime.as_ref(), thread_id) + .clear_thread_goal(runtime.as_ref(), thread_id, std::future::ready(Ok(()))) .await? ); assert_eq!(Vec::::new(), harness.sink.goal_events()); @@ -1483,17 +1485,28 @@ async fn goal_service_sets_gets_and_clears_thread_goal() -> anyhow::Result<()> { seed_thread_metadata(runtime.as_ref(), thread_id).await?; let api = GoalService::new(); - let set = api + let request = GoalSetRequest { + thread_id, + objective: GoalObjectiveUpdate::Set(" ship goal API ownership "), + status: None, + token_budget: GoalTokenBudgetUpdate::Set(Some(123)), + max_goal_token_budget: None, + }; + let failure = GoalServiceError::Internal("instruction recording failed".to_owned()); + let result = api .set_thread_goal( runtime.as_ref(), - GoalSetRequest { - thread_id, - objective: GoalObjectiveUpdate::Set(" ship goal API ownership "), - status: None, - token_budget: GoalTokenBudgetUpdate::Set(Some(123)), - max_goal_token_budget: None, - }, + request, + std::future::ready(Err(failure.clone())), ) + .await; + assert_eq!(result.unwrap_err(), failure); + assert_eq!( + api.get_thread_goal(runtime.as_ref(), thread_id).await?, + None + ); + let set = api + .set_thread_goal(runtime.as_ref(), request, std::future::ready(Ok(()))) .await?; let get = api .get_thread_goal(runtime.as_ref(), thread_id) @@ -1510,12 +1523,32 @@ async fn goal_service_sets_gets_and_clears_thread_goal() -> anyhow::Result<()> { assert_eq!(Some(123), get.token_budget); assert_eq!(Some("ship goal API ownership"), metadata.preview.as_deref()); - assert!(api.clear_thread_goal(runtime.as_ref(), thread_id).await?); + assert_eq!( + api.clear_thread_goal( + runtime.as_ref(), + thread_id, + std::future::ready(Err(failure.clone())) + ) + .await, + Err(failure), + ); + assert_eq!( + api.get_thread_goal(runtime.as_ref(), thread_id).await?, + Some(get) + ); + + assert!( + api.clear_thread_goal(runtime.as_ref(), thread_id, std::future::ready(Ok(()))) + .await? + ); assert_eq!( None, api.get_thread_goal(runtime.as_ref(), thread_id).await? ); - assert!(!api.clear_thread_goal(runtime.as_ref(), thread_id).await?); + assert!( + !api.clear_thread_goal(runtime.as_ref(), thread_id, std::future::ready(Ok(()))) + .await? + ); Ok(()) } @@ -1537,6 +1570,7 @@ async fn goal_service_enforces_maximum_token_budget_on_creation_and_updates() -> token_budget: GoalTokenBudgetUpdate::Keep, max_goal_token_budget: Some(100), }, + std::future::ready(Ok(())), ) .await?; assert_eq!(goal.goal.token_budget, Some(100)); @@ -1551,6 +1585,7 @@ async fn goal_service_enforces_maximum_token_budget_on_creation_and_updates() -> token_budget: GoalTokenBudgetUpdate::Set(Some(101)), max_goal_token_budget: Some(100), }, + std::future::ready(Ok(())), ) .await .expect_err("goal budget above the configured maximum should fail"); @@ -1577,6 +1612,7 @@ async fn goal_service_enforces_maximum_token_budget_on_creation_and_updates() -> token_budget: GoalTokenBudgetUpdate::Set(Some(99)), max_goal_token_budget: Some(100), }, + std::future::ready(Ok(())), ) .await?; assert_eq!(goal.goal.token_budget, Some(99)); @@ -1591,6 +1627,7 @@ async fn goal_service_enforces_maximum_token_budget_on_creation_and_updates() -> token_budget: GoalTokenBudgetUpdate::Set(None), max_goal_token_budget: Some(100), }, + std::future::ready(Ok(())), ) .await?; assert_eq!(goal.goal.token_budget, Some(100)); diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index f9c6f5fd77..fac4b63702 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -1493,6 +1493,7 @@ impl AppServerSession { .request_typed(ClientRequest::ThreadGoalSet { request_id, params: ThreadGoalSetParams { + origin: Some(codex_app_server_protocol::ThreadGoalMutationOrigin::User), thread_id: thread_id.to_string(), objective, status, @@ -1512,6 +1513,7 @@ impl AppServerSession { .request_typed(ClientRequest::ThreadGoalClear { request_id, params: ThreadGoalClearParams { + origin: Some(codex_app_server_protocol::ThreadGoalMutationOrigin::User), thread_id: thread_id.to_string(), }, }) diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 6a2edc4725..e2fef8e383 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -5247,13 +5247,6 @@ class ThreadExtra(BaseModel): ) -class ThreadGoalClearParams(BaseModel): - model_config = ConfigDict( - populate_by_name=True, - ) - thread_id: Annotated[str, Field(alias="threadId")] - - class ThreadGoalClearResponse(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -5275,6 +5268,11 @@ class ThreadGoalGetParams(BaseModel): thread_id: Annotated[str, Field(alias="threadId")] +class ThreadGoalMutationOrigin(Enum): + user = "user" + automatic = "automatic" + + class ThreadGoalStatus(Enum): active = "active" paused = "paused" @@ -6867,15 +6865,6 @@ class ThreadGoalGetRequest(BaseModel): params: ThreadGoalGetParams -class ThreadGoalClearRequest(BaseModel): - model_config = ConfigDict( - populate_by_name=True, - ) - id: RequestId - method: Annotated[Literal["thread/goal/clear"], Field(title="Thread/goal/clearRequestMethod")] - params: ThreadGoalClearParams - - class ThreadMetadataUpdateRequest(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -9812,6 +9801,17 @@ class ThreadGoal(BaseModel): updated_at: Annotated[int, Field(alias="updatedAt")] +class ThreadGoalClearParams(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + origin: Annotated[ + ThreadGoalMutationOrigin | None, + Field(description="Missing provenance does not supply user authorization."), + ] = None + thread_id: Annotated[str, Field(alias="threadId")] + + class ThreadGoalGetResponse(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -9824,6 +9824,10 @@ class ThreadGoalSetParams(BaseModel): populate_by_name=True, ) objective: str | None = None + origin: Annotated[ + ThreadGoalMutationOrigin | None, + Field(description="Missing provenance does not supply user authorization."), + ] = None status: ThreadGoalStatus | None = None thread_id: Annotated[str, Field(alias="threadId")] token_budget: Annotated[int | None, Field(alias="tokenBudget")] = None @@ -10524,6 +10528,15 @@ class ThreadGoalSetRequest(BaseModel): params: ThreadGoalSetParams +class ThreadGoalClearRequest(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + id: RequestId + method: Annotated[Literal["thread/goal/clear"], Field(title="Thread/goal/clearRequestMethod")] + params: ThreadGoalClearParams + + class ThreadListRequest(BaseModel): model_config = ConfigDict( populate_by_name=True,