diff --git a/codex-rs/app-server/src/request_processors/thread_monitor_processor.rs b/codex-rs/app-server/src/request_processors/thread_monitor_processor.rs index 291b439c5..3a129e077 100644 --- a/codex-rs/app-server/src/request_processors/thread_monitor_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_monitor_processor.rs @@ -1,5 +1,10 @@ use super::thread_monitor_api::*; use super::*; +use codex_app_server_protocol::CommandExecutionApprovalDecision; +use codex_app_server_protocol::CommandExecutionRequestApprovalParams; +use codex_app_server_protocol::CommandExecutionRequestApprovalResponse; +use codex_app_server_protocol::ServerRequestPayload; +use codex_shell_command::parse_command::shlex_join; #[derive(Clone)] pub(crate) struct ThreadMonitorRequestProcessor { @@ -106,6 +111,17 @@ impl ThreadMonitorRequestProcessor { let state_db = self.prepare_monitor_mutation(thread_id).await?; self.ensure_thread_monitor_capacity(&state_db, thread_id) .await?; + let writer = self.current_monitor_writer(thread_id).await?; + let authorization = self + .authorize_monitor_command( + &request_id, + thread_id, + &writer, + /*generation*/ 0, + command.as_str(), + cwd.as_deref(), + ) + .await?; let monitor = state_db .thread_monitors() .create_thread_monitor(codex_state::ThreadMonitorCreateParams { @@ -117,6 +133,7 @@ impl ThreadMonitorRequestProcessor { routing, output_file, status: codex_state::ThreadMonitorStatus::Running, + authorization: Some(authorization), }) .await .map_err(|err| internal_error(format!("failed to create thread monitor: {err}")))?; @@ -257,9 +274,24 @@ impl ThreadMonitorRequestProcessor { let monitor_id = self .resolve_monitor_id_for_thread(&state_db, thread_id, params.monitor_id.as_str()) .await?; + let monitor = self + .load_monitor_for_thread(&state_db, thread_id, monitor_id.as_str()) + .await?; + let cwd = validate_optional_monitor_relative_path("monitor cwd", monitor.cwd.clone())?; + let writer = self.current_monitor_writer(thread_id).await?; + let authorization = self + .authorize_monitor_command( + &request_id, + thread_id, + &writer, + monitor.generation + 1, + monitor.command.as_str(), + cwd.as_deref(), + ) + .await?; let monitor = state_db .thread_monitors() - .restart_thread_monitor(monitor_id.as_str()) + .restart_thread_monitor(monitor_id.as_str(), monitor.generation, authorization) .await .map_err(|err| internal_error(format!("failed to restart thread monitor: {err}")))? .ok_or_else(|| invalid_request(format!("monitor not found: {monitor_id}")))?; @@ -333,6 +365,128 @@ impl ThreadMonitorRequestProcessor { Ok(()) } + async fn current_monitor_writer( + &self, + thread_id: ThreadId, + ) -> Result, JSONRPCErrorError> { + self.thread_manager + .get_thread(thread_id) + .await + .map_err(|_| { + invalid_request(format!( + "thread must be loaded to authorize monitor commands: {thread_id}" + )) + }) + } + + #[allow(clippy::too_many_arguments)] + async fn authorize_monitor_command( + &self, + request_id: &ConnectionRequestId, + thread_id: ThreadId, + writer: &Arc, + generation: i64, + command: &str, + cwd: Option<&str>, + ) -> Result { + let writer_fence = writer.monitor_writer_fence().to_string(); + let before = writer.config_snapshot().await; + let approval_cwd = monitor_approval_cwd(&before.cwd, cwd)?; + self.request_monitor_command_approval(request_id, thread_id, command, approval_cwd) + .await?; + + let current_writer = self.current_monitor_writer(thread_id).await?; + let after = current_writer.config_snapshot().await; + if writer_fence != current_writer.monitor_writer_fence() + || before.cwd != after.cwd + || before.permission_profile != after.permission_profile + { + return Err(invalid_request( + "monitor command authorization became stale before it could be recorded", + )); + } + Ok(codex_state::ThreadMonitorAuthorization::new( + thread_id, + generation, + command, + cwd, + writer_fence, + after.permission_profile, + after.cwd.display().to_string(), + )) + } + + async fn request_monitor_command_approval( + &self, + request_id: &ConnectionRequestId, + thread_id: ThreadId, + command: &str, + cwd: AbsolutePathBuf, + ) -> Result<(), JSONRPCErrorError> { + let command = codex_core::exec::persistent_shell_command_args(command); + let item_id = format!("monitor-authorization-{}", Uuid::new_v4()); + let params = CommandExecutionRequestApprovalParams { + thread_id: thread_id.to_string(), + turn_id: item_id.clone(), + item_id, + started_at_ms: Utc::now().timestamp_millis(), + approval_id: None, + reason: Some( + "Authorize this persistent monitor command. It may run in the background until stopped." + .to_string(), + ), + network_approval_context: None, + command: Some(shlex_join(&command)), + cwd: Some(cwd), + command_actions: None, + additional_permissions: None, + proposed_execpolicy_amendment: None, + proposed_network_policy_amendments: None, + available_decisions: Some(vec![ + CommandExecutionApprovalDecision::Accept, + CommandExecutionApprovalDecision::Decline, + CommandExecutionApprovalDecision::Cancel, + ]), + }; + let connection_ids = [request_id.connection_id]; + let (_, receiver) = self + .outgoing + .send_request_to_connections( + Some(&connection_ids), + ServerRequestPayload::CommandExecutionRequestApproval(params), + Some(thread_id), + ) + .await; + let response = receiver + .await + .map_err(|err| { + warn!( + thread_id = %thread_id, + "monitor command approval callback closed: {err}" + ); + invalid_request("monitor command approval was not granted") + })? + .map_err(|err| { + warn!( + thread_id = %thread_id, + "monitor command approval failed: {err:?}" + ); + invalid_request("monitor command approval was not granted") + })?; + let response = serde_json::from_value::(response) + .map_err(|err| { + warn!( + thread_id = %thread_id, + "invalid monitor command approval response: {err}" + ); + invalid_request("monitor command approval was not granted") + })?; + if !monitor_command_approval_granted(&response.decision) { + return Err(invalid_request("monitor command approval was not granted")); + } + Ok(()) + } + async fn prepare_monitor_mutation( &self, thread_id: ThreadId, @@ -579,3 +733,39 @@ fn parse_thread_id_for_monitor_request(thread_id: &str) -> Result, +) -> Result { + let cwd = match cwd { + Some(cwd) => thread_cwd.join(cwd), + None => thread_cwd.clone(), + }; + Ok(cwd) +} + +fn monitor_command_approval_granted(decision: &CommandExecutionApprovalDecision) -> bool { + matches!(decision, CommandExecutionApprovalDecision::Accept) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn monitor_command_approval_is_one_shot_and_fails_closed() { + assert!(monitor_command_approval_granted( + &CommandExecutionApprovalDecision::Accept + )); + assert!(!monitor_command_approval_granted( + &CommandExecutionApprovalDecision::AcceptForSession + )); + assert!(!monitor_command_approval_granted( + &CommandExecutionApprovalDecision::Decline + )); + assert!(!monitor_command_approval_granted( + &CommandExecutionApprovalDecision::Cancel + )); + } +} diff --git a/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs b/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs index 5df05078f..0e358e13d 100644 --- a/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs +++ b/codex-rs/app-server/src/request_processors/thread_monitor_runtime.rs @@ -2,21 +2,14 @@ use super::thread_monitor_api::api_thread_monitor_event_from_state; use super::thread_monitor_api::api_thread_monitor_from_state; use super::*; use codex_protocol::AgentPath; +use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::InterAgentCommunication; -#[cfg(not(target_os = "windows"))] -use codex_shell_command::shell_detect::ShellType; -#[cfg(not(target_os = "windows"))] -use codex_shell_command::shell_detect::get_shell; -#[cfg(not(target_os = "windows"))] -use codex_shell_command::shell_detect::ultimate_fallback_shell; use std::path::Component; use std::path::Path; use std::path::PathBuf; -use std::process::Stdio; use tokio::io::AsyncBufReadExt; use tokio::io::AsyncWriteExt; use tokio::io::BufReader; -use tokio::process::Command; const MONITOR_POLL_INTERVAL: Duration = Duration::from_secs(2); const MAX_MONITOR_EVENT_CHARS: usize = 8_000; @@ -231,25 +224,17 @@ impl ThreadMonitorRuntime { .await; return; }; - self.record_monitor_event( - &state_db, - &monitor, - codex_state::ThreadMonitorEventStream::System, - "monitor process starting", - ) - .await; - - let thread_cwd = match monitor_thread_cwd(&state_db, &monitor).await { - Ok(cwd) => cwd, + let (initial_snapshot, _) = match self.authorized_monitor_execution(&monitor).await { + Ok(context) => context, Err(err) => { - let error = monitor_error(format!("invalid monitor thread cwd: {err}")); + let error = monitor_error(err); self.mark_monitor_failed(&state_db, &monitor, error).await; self.remove_active_monitor(&monitor.monitor_id, monitor.generation) .await; return; } }; - let cwd = match resolve_monitor_cwd(&monitor, thread_cwd.as_path()).await { + let cwd = match resolve_monitor_cwd(&monitor, initial_snapshot.cwd.as_path()).await { Ok(cwd) => cwd, Err(err) => { let error = monitor_error(format!("invalid monitor cwd: {err}")); @@ -259,17 +244,62 @@ impl ThreadMonitorRuntime { return; } }; - let mut command = monitor_command(&monitor.command); - command - .current_dir(&cwd) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .stdin(Stdio::null()); - - let mut child = match command.spawn() { + let (snapshot, config) = match self.authorized_monitor_execution(&monitor).await { + Ok(context) => context, + Err(err) => { + let error = monitor_error(err); + self.mark_monitor_failed(&state_db, &monitor, error).await; + self.remove_active_monitor(&monitor.monitor_id, monitor.generation) + .await; + return; + } + }; + if initial_snapshot.cwd != snapshot.cwd { + let error = monitor_error( + "monitor command authorization became stale while resolving its cwd; restart the monitor to reauthorize" + .to_string(), + ); + self.mark_monitor_failed(&state_db, &monitor, error).await; + self.remove_active_monitor(&monitor.monitor_id, monitor.generation) + .await; + return; + } + let cwd = match AbsolutePathBuf::try_from(cwd) { + Ok(cwd) => cwd, + Err(err) => { + let error = monitor_error(format!("invalid absolute monitor cwd: {err}")); + self.mark_monitor_failed(&state_db, &monitor, error).await; + self.remove_active_monitor(&monitor.monitor_id, monitor.generation) + .await; + return; + } + }; + self.record_monitor_event( + &state_db, + &monitor, + codex_state::ThreadMonitorEventStream::System, + "monitor process starting", + ) + .await; + let env = create_env( + &config.permissions.shell_environment_policy, + Some(monitor.thread_id), + ); + let mut child = match codex_core::exec::spawn_streaming_command_under_sandbox( + codex_core::exec::persistent_shell_command_args(&monitor.command), + cwd.clone(), + env, + &snapshot.permission_profile, + &snapshot.cwd, + &config.codex_linux_sandbox_exe, + config.features.use_legacy_landlock(), + ) + .await + { Ok(child) => child, Err(err) => { - let error = monitor_error(format!("failed to start monitor command: {err}")); + let error = + monitor_error(format!("failed to start sandboxed monitor command: {err}")); self.mark_monitor_failed(&state_db, &monitor, error).await; self.remove_active_monitor(&monitor.monitor_id, monitor.generation) .await; @@ -298,7 +328,7 @@ impl ThreadMonitorRuntime { let runtime = self.clone(); let state_db = state_db.clone(); let monitor = monitor.clone(); - let cwd = cwd.clone(); + let cwd = cwd.to_path_buf(); let cancel_token = cancel_token.clone(); self.tasks.spawn(async move { runtime @@ -318,7 +348,7 @@ impl ThreadMonitorRuntime { let runtime = self.clone(); let state_db = state_db.clone(); let monitor = monitor.clone(); - let cwd = cwd.clone(); + let cwd = cwd.to_path_buf(); let cancel_token = cancel_token.clone(); self.tasks.spawn(async move { runtime @@ -369,6 +399,34 @@ impl ThreadMonitorRuntime { .await; } + async fn authorized_monitor_execution( + &self, + monitor: &codex_state::ThreadMonitor, + ) -> Result<(ThreadConfigSnapshot, Arc), String> { + let thread = self + .thread_manager + .get_thread(monitor.thread_id) + .await + .map_err(|_| { + "monitor command authorization has no current thread writer; restart the monitor to reauthorize" + .to_string() + })?; + let snapshot = thread.config_snapshot().await; + let thread_cwd = snapshot.cwd.display().to_string(); + if !monitor_authorization_is_current( + monitor, + thread.monitor_writer_fence(), + &snapshot.permission_profile, + thread_cwd.as_str(), + ) { + return Err( + "monitor command authorization is missing or stale; restart the monitor to reauthorize" + .to_string(), + ); + } + Ok((snapshot, thread.config().await)) + } + async fn read_monitor_stream( &self, state_db: StateDbHandle, @@ -635,20 +693,6 @@ impl ThreadMonitorRuntime { } } -async fn monitor_thread_cwd( - state_db: &StateDbHandle, - monitor: &codex_state::ThreadMonitor, -) -> anyhow::Result { - let metadata = state_db - .get_thread(monitor.thread_id) - .await? - .ok_or_else(|| anyhow::anyhow!("monitor thread metadata not found"))?; - if metadata.cwd.as_os_str().is_empty() { - anyhow::bail!("monitor thread cwd is empty"); - } - Ok(metadata.cwd) -} - /// Builds the mailbox communication injected into a thread for a monitor line. /// /// `trigger_turn` is `true` so an idle thread is woken to observe the event @@ -679,26 +723,6 @@ Output: ) } -fn monitor_command(command: &str) -> Command { - #[cfg(target_os = "windows")] - { - let mut cmd = Command::new("cmd"); - cmd.arg("/C").arg(command); - cmd - } - - #[cfg(not(target_os = "windows"))] - { - let shell = get_shell(ShellType::Bash, /*path*/ None) - .or_else(|| get_shell(ShellType::Zsh, /*path*/ None)) - .or_else(|| get_shell(ShellType::Sh, /*path*/ None)) - .unwrap_or_else(ultimate_fallback_shell); - let mut cmd = Command::new(shell.shell_path); - cmd.arg("-lc").arg(command); - cmd - } -} - async fn resolve_monitor_cwd( monitor: &codex_state::ThreadMonitor, fallback: &Path, @@ -715,6 +739,17 @@ async fn resolve_monitor_cwd( Ok(canonical_cwd) } +fn monitor_authorization_is_current( + monitor: &codex_state::ThreadMonitor, + writer_fence: &str, + permission_profile: &PermissionProfile, + thread_cwd: &str, +) -> bool { + monitor.authorization.as_ref().is_some_and(|authorization| { + authorization.authorizes(monitor, writer_fence, permission_profile, thread_cwd) + }) +} + fn resolve_monitor_relative_path( field_name: &str, base: &Path, @@ -755,6 +790,7 @@ fn truncate_chars(value: String, max_chars: usize) -> String { mod tests { use super::*; use pretty_assertions::assert_eq; + use tokio::process::Command; #[test] fn monitor_relative_path_resolver_stays_within_base() { @@ -802,44 +838,23 @@ mod tests { ); } - #[tokio::test] - async fn monitor_thread_cwd_reads_persisted_thread_metadata() -> anyhow::Result<()> { - let tempdir = tempfile::TempDir::new()?; - let app_server_cwd = tempdir.path().join("server"); - let thread_cwd = tempdir.path().join("thread"); - tokio::fs::create_dir_all(&app_server_cwd).await?; - tokio::fs::create_dir_all(&thread_cwd).await?; - let state_db = - codex_state::StateRuntime::init(tempdir.path().join("state"), "test-provider".into()) - .await?; - let thread_id = codex_protocol::ThreadId::new(); - let mut builder = codex_state::ThreadMetadataBuilder::new( - thread_id, - tempdir.path().join("rollout.jsonl"), - chrono::Utc::now(), - codex_protocol::protocol::SessionSource::default(), - ); - builder.cwd = thread_cwd.clone(); - state_db - .upsert_thread(&builder.build("test-provider")) - .await?; - let monitor = test_monitor_for_thread(thread_id, /*cwd*/ None); - - assert_eq!(monitor_thread_cwd(&state_db, &monitor).await?, thread_cwd); - assert_ne!( - monitor_thread_cwd(&state_db, &monitor).await?, - app_server_cwd - ); - Ok(()) - } - #[tokio::test] async fn monitor_command_supports_bash_source_when_bash_is_available() { - if get_shell(ShellType::Bash, /*path*/ None).is_none() { + if codex_shell_command::shell_detect::get_shell( + codex_shell_command::shell_detect::ShellType::Bash, + /*path*/ None, + ) + .is_none() + { return; } - - let output = monitor_command("source /dev/null && printf ok") + let command = + codex_core::exec::persistent_shell_command_args("source /dev/null && printf ok"); + let Some((program, args)) = command.split_first() else { + return; + }; + let output = Command::new(program) + .args(args) .output() .await .expect("monitor command should run"); @@ -852,6 +867,45 @@ mod tests { assert_eq!(String::from_utf8_lossy(&output.stdout), "ok"); } + #[test] + fn monitor_authorization_requires_current_fence_profile_and_subject() { + let permission_profile = PermissionProfile::read_only(); + let monitor = test_monitor(/*cwd*/ None); + assert!(monitor_authorization_is_current( + &monitor, + "writer-fence", + &permission_profile, + "/workspace", + )); + + let mut missing = monitor.clone(); + missing.authorization = None; + assert!(!monitor_authorization_is_current( + &missing, + "writer-fence", + &permission_profile, + "/workspace", + )); + assert!(!monitor_authorization_is_current( + &monitor, + "stale-writer", + &permission_profile, + "/workspace", + )); + assert!(!monitor_authorization_is_current( + &monitor, + "writer-fence", + &PermissionProfile::Disabled, + "/workspace", + )); + assert!(!monitor_authorization_is_current( + &monitor, + "writer-fence", + &permission_profile, + "/different-workspace", + )); + } + #[test] fn monitor_output_communication_uses_wake_if_idle_mailbox_shape() { let monitor = test_monitor(/*cwd*/ None); @@ -882,17 +936,29 @@ mod tests { cwd: Option<&str>, ) -> codex_state::ThreadMonitor { let now = chrono::Utc::now(); + let command = "printf ok"; + let cwd = cwd.map(str::to_string); + let authorization = codex_state::ThreadMonitorAuthorization::new( + thread_id, + /*generation*/ 1, + command, + cwd.as_deref(), + "writer-fence".to_string(), + PermissionProfile::read_only(), + "/workspace".to_string(), + ); codex_state::ThreadMonitor { thread_id, monitor_id: "monitor-id".to_string(), name: "monitor".to_string(), prompt: "watch".to_string(), - command: "printf ok".to_string(), - cwd: cwd.map(str::to_string), + command: command.to_string(), + cwd, routing: codex_state::ThreadMonitorRouting::File, output_file: Some("monitor.log".to_string()), status: codex_state::ThreadMonitorStatus::Running, generation: 1, + authorization: Some(authorization), process_id: None, last_event_at: None, last_error: None, diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index e12006ff9..a68dc5d65 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -703,6 +703,15 @@ impl CodexThread { self.codex.thread_config_snapshot().await } + /// Returns the process-local writer fence for monitor command authorization. + /// + /// The fence changes whenever this thread is materialized as a new live + /// session, so persisted monitor authorizations cannot be replayed by a + /// later app-server process without an explicit create/restart approval. + pub fn monitor_writer_fence(&self) -> &str { + self.codex.session.monitor_writer_fence() + } + /// Returns the files that supplied the thread's loaded model instructions. pub async fn instruction_sources(&self) -> Vec { self.codex.instruction_sources().await diff --git a/codex-rs/core/src/exec.rs b/codex-rs/core/src/exec.rs index 4d9f58454..e55f8f844 100644 --- a/codex-rs/core/src/exec.rs +++ b/codex-rs/core/src/exec.rs @@ -44,6 +44,12 @@ use codex_sandboxing::SandboxTransformRequest; use codex_sandboxing::SandboxType; use codex_sandboxing::SandboxablePreference; use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile; +#[cfg(not(target_os = "windows"))] +use codex_shell_command::shell_detect::ShellType; +#[cfg(not(target_os = "windows"))] +use codex_shell_command::shell_detect::get_shell; +#[cfg(not(target_os = "windows"))] +use codex_shell_command::shell_detect::ultimate_fallback_shell; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP; use codex_utils_pty::process_group::kill_child_process_group; @@ -81,6 +87,30 @@ pub(crate) const MAX_EXEC_OUTPUT_DELTAS_PER_CALL: usize = 10_000; // indefinitely, effectively hanging the whole agent. pub const IO_DRAIN_TIMEOUT_MS: u64 = 2_000; // 2 s should be plenty for local pipes +/// Builds the exact shell argv used for persistent model-designed commands. +/// +/// Approval and execution must both use this helper so the command the user +/// authorizes is the command the monitor runtime later passes to the sandbox. +pub fn persistent_shell_command_args(command: &str) -> Vec { + #[cfg(target_os = "windows")] + { + vec!["cmd".to_string(), "/C".to_string(), command.to_string()] + } + + #[cfg(not(target_os = "windows"))] + { + let shell = get_shell(ShellType::Bash, /*path*/ None) + .or_else(|| get_shell(ShellType::Zsh, /*path*/ None)) + .or_else(|| get_shell(ShellType::Sh, /*path*/ None)) + .unwrap_or_else(ultimate_fallback_shell); + vec![ + shell.shell_path.to_string_lossy().into_owned(), + "-lc".to_string(), + command.to_string(), + ] + } +} + #[derive(Debug)] pub struct ExecParams { pub command: Vec, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 9071ca8c0..14c471058 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1869,6 +1869,11 @@ impl Session { .clone() } + pub(crate) async fn thread_config_snapshot(&self) -> ThreadConfigSnapshot { + let state = self.state.lock().await; + state.session_configuration.thread_config_snapshot() + } + pub(crate) async fn provider(&self) -> ModelProviderInfo { let state = self.state.lock().await; state.session_configuration.provider.clone() diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index e77138556..610bee385 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -26,6 +26,7 @@ use tokio::sync::Semaphore; /// A session has at most 1 running task at a time, and can be interrupted by user input. pub(crate) struct Session { pub(crate) thread_id: ThreadId, + pub(crate) monitor_writer_fence: String, pub(crate) installation_id: String, pub(super) tx_event: Sender, pub(super) agent_status: watch::Sender, @@ -662,6 +663,10 @@ impl Session { self.services.agent_control.session_id() } + pub(crate) fn monitor_writer_fence(&self) -> &str { + self.monitor_writer_fence.as_str() + } + #[instrument(name = "session_init", level = "info", skip_all)] #[allow(clippy::too_many_arguments)] #[expect( @@ -1262,6 +1267,7 @@ impl Session { let sess = Arc::new(Session { thread_id, + monitor_writer_fence: uuid::Uuid::new_v4().to_string(), installation_id, tx_event: tx_event.clone(), agent_status, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ff713f5a6..d2c08a8a1 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6360,6 +6360,7 @@ async fn make_session_and_context_with_events() let session = Session { thread_id, + monitor_writer_fence: uuid::Uuid::new_v4().to_string(), installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, agent_status: agent_status_tx, @@ -8456,6 +8457,7 @@ where let session = Arc::new(Session { thread_id, + monitor_writer_fence: uuid::Uuid::new_v4().to_string(), installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, agent_status: agent_status_tx, @@ -8501,6 +8503,22 @@ pub(crate) async fn make_session_and_context_with_rx() -> ( make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await } +#[tokio::test] +async fn session_test_initializers_assign_distinct_monitor_writer_fences() { + let config_session = make_session_with_config(|_config| {}) + .await + .expect("create config-backed session"); + let (context_session, _turn_context, _rx_event) = + make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await; + + let config_fence = config_session.monitor_writer_fence(); + let context_fence = context_session.monitor_writer_fence(); + + assert!(!config_fence.is_empty()); + assert!(!context_fence.is_empty()); + assert_ne!(config_fence, context_fence); +} + #[tokio::test] async fn refresh_mcp_servers_is_deferred_until_next_turn() { let (session, turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/tools/handlers/monitor_control.rs b/codex-rs/core/src/tools/handlers/monitor_control.rs index 42ca1945a..9bee91410 100644 --- a/codex-rs/core/src/tools/handlers/monitor_control.rs +++ b/codex-rs/core/src/tools/handlers/monitor_control.rs @@ -1,6 +1,8 @@ //! Built-in model tool handler for managing thread monitors. use crate::function_tool::FunctionCallError; +use crate::session::session::Session; +use crate::session::turn_context::TurnContext; use crate::tools::context::FunctionToolOutput; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolPayload; @@ -11,8 +13,10 @@ use crate::tools::handlers::parse_arguments; use crate::tools::registry::CoreToolRuntime; use crate::tools::registry::ToolExecutor; use codex_protocol::ThreadId; +use codex_protocol::protocol::ReviewDecision; use codex_tools::ToolName; use codex_tools::ToolSpec; +use codex_utils_absolute_path::AbsolutePathBuf; use serde::Deserialize; use serde::Serialize; use serde_json::Value as JsonValue; @@ -129,7 +133,11 @@ impl ToolExecutor for ManageMonitorHandler { invocation: ToolInvocation, ) -> Result, FunctionCallError> { let ToolInvocation { - session, payload, .. + session, + turn, + call_id, + payload, + .. } = invocation; let arguments = match payload { @@ -146,7 +154,15 @@ impl ToolExecutor for ManageMonitorHandler { let state_db = session.state_db().ok_or_else(|| { FunctionCallError::Fatal("sqlite state db is unavailable for this session".to_string()) })?; - let response = manage_monitor(state_db, session.thread_id(), args).await?; + let response = manage_monitor( + state_db, + session.clone(), + turn, + call_id, + session.thread_id(), + args, + ) + .await?; monitor_response(response, verbose).map(boxed_tool_output) } } @@ -155,11 +171,16 @@ impl CoreToolRuntime for ManageMonitorHandler {} async fn manage_monitor( state_db: Arc, + session: Arc, + turn: Arc, + call_id: String, thread_id: ThreadId, args: ManageMonitorArgs, ) -> Result { match args.action { - MonitorAction::Create => create_monitor(state_db, thread_id, args).await, + MonitorAction::Create => { + create_monitor(state_db, &session, &turn, call_id.as_str(), thread_id, args).await + } MonitorAction::List => { let monitors = list_monitor_snapshots(&state_db, thread_id).await?; Ok(ManageMonitorResponse { @@ -175,13 +196,18 @@ async fn manage_monitor( } MonitorAction::Read => read_monitor(state_db, thread_id, args).await, MonitorAction::Stop => set_monitor_stopped(state_db, thread_id, args).await, - MonitorAction::Restart => restart_monitor(state_db, thread_id, args).await, + MonitorAction::Restart => { + restart_monitor(state_db, &session, &turn, call_id.as_str(), thread_id, args).await + } MonitorAction::Delete => delete_monitor(state_db, thread_id, args).await, } } async fn create_monitor( state_db: Arc, + session: &Arc, + turn: &Arc, + call_id: &str, thread_id: ThreadId, args: ManageMonitorArgs, ) -> Result { @@ -215,6 +241,16 @@ async fn create_monitor( let output_file = validate_optional_monitor_relative_path("output_file", args.output_file.as_deref())?; let output_file = validate_output_file_for_routing(routing, output_file)?; + let authorization = authorize_monitor_command( + session, + turn, + call_id, + thread_id, + /*generation*/ 0, + command.as_str(), + cwd.as_deref(), + ) + .await?; let monitor = state_db .thread_monitors() .create_thread_monitor(codex_state::ThreadMonitorCreateParams { @@ -226,6 +262,7 @@ async fn create_monitor( routing, output_file, status: codex_state::ThreadMonitorStatus::Running, + authorization: Some(authorization), }) .await .map_err(|err| { @@ -344,14 +381,28 @@ async fn set_monitor_stopped( async fn restart_monitor( state_db: Arc, + session: &Arc, + turn: &Arc, + call_id: &str, thread_id: ThreadId, args: ManageMonitorArgs, ) -> Result { let monitor_id = resolve_monitor_id(&state_db, thread_id, args.monitor_id.as_deref()).await?; - load_monitor_for_thread(&state_db, thread_id, monitor_id.as_str()).await?; + let monitor = load_monitor_for_thread(&state_db, thread_id, monitor_id.as_str()).await?; + let cwd = validate_optional_monitor_relative_path("cwd", monitor.cwd.as_deref())?; + let authorization = authorize_monitor_command( + session, + turn, + call_id, + thread_id, + monitor.generation + 1, + monitor.command.as_str(), + cwd.as_deref(), + ) + .await?; let monitor = state_db .thread_monitors() - .restart_thread_monitor(monitor_id.as_str()) + .restart_thread_monitor(monitor_id.as_str(), monitor.generation, authorization) .await .map_err(|err| FunctionCallError::Fatal(format!("failed to restart monitor: {err}")))? .ok_or_else(|| model_error(format!("monitor not found: {monitor_id}")))?; @@ -369,6 +420,73 @@ async fn restart_monitor( }) } +async fn authorize_monitor_command( + session: &Arc, + turn: &Arc, + call_id: &str, + thread_id: ThreadId, + generation: i64, + command: &str, + cwd: Option<&str>, +) -> Result { + let before = session.thread_config_snapshot().await; + let approval_cwd = monitor_approval_cwd(&before.cwd, cwd)?; + let decision = session + .request_command_approval( + turn, + call_id.to_string(), + /*approval_id*/ None, + crate::exec::persistent_shell_command_args(command), + approval_cwd, + Some( + "Authorize this persistent monitor command. It may run in the background until stopped." + .to_string(), + ), + /*network_approval_context*/ None, + /*proposed_execpolicy_amendment*/ None, + /*additional_permissions*/ None, + Some(vec![ + ReviewDecision::Approved, + ReviewDecision::Denied, + ReviewDecision::Abort, + ]), + ) + .await; + if !monitor_command_approval_granted(&decision) { + return Err(model_error("monitor command approval was not granted")); + } + + let after = session.thread_config_snapshot().await; + if before.cwd != after.cwd || before.permission_profile != after.permission_profile { + return Err(model_error( + "monitor command authorization became stale before it could be recorded", + )); + } + Ok(codex_state::ThreadMonitorAuthorization::new( + thread_id, + generation, + command, + cwd, + session.monitor_writer_fence().to_string(), + after.permission_profile, + after.cwd.display().to_string(), + )) +} + +fn monitor_approval_cwd( + thread_cwd: &AbsolutePathBuf, + cwd: Option<&str>, +) -> Result { + Ok(match cwd { + Some(cwd) => thread_cwd.join(cwd), + None => thread_cwd.clone(), + }) +} + +fn monitor_command_approval_granted(decision: &ReviewDecision) -> bool { + matches!(decision, ReviewDecision::Approved) +} + async fn delete_monitor( state_db: Arc, thread_id: ThreadId, @@ -760,6 +878,7 @@ mod tests { routing: codex_state::ThreadMonitorRouting::Stream, output_file: None, status: codex_state::ThreadMonitorStatus::Running, + authorization: None, }) .await .expect("monitor should be created") @@ -909,4 +1028,14 @@ mod tests { assert!(validate_optional_monitor_relative_path("output_file", Some("../out")).is_err()); assert!(validate_optional_monitor_relative_path("output_file", Some(".")).is_err()); } + + #[test] + fn monitor_command_approval_is_one_shot_and_fails_closed() { + assert!(monitor_command_approval_granted(&ReviewDecision::Approved)); + assert!(!monitor_command_approval_granted( + &ReviewDecision::ApprovedForSession + )); + assert!(!monitor_command_approval_granted(&ReviewDecision::Denied)); + assert!(!monitor_command_approval_granted(&ReviewDecision::Abort)); + } } diff --git a/codex-rs/state/migrations/0068_thread_monitor_authorization.sql b/codex-rs/state/migrations/0068_thread_monitor_authorization.sql new file mode 100644 index 000000000..bfff9573f --- /dev/null +++ b/codex-rs/state/migrations/0068_thread_monitor_authorization.sql @@ -0,0 +1,2 @@ +ALTER TABLE thread_monitors +ADD COLUMN authorization_json TEXT; diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index 6c407144c..b9fad6e44 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -134,6 +134,7 @@ pub use model::ThreadGoalStatus; pub use model::ThreadMetadata; pub use model::ThreadMetadataBuilder; pub use model::ThreadMonitor; +pub use model::ThreadMonitorAuthorization; pub use model::ThreadMonitorEvent; pub use model::ThreadMonitorEventStream; pub use model::ThreadMonitorRouting; diff --git a/codex-rs/state/src/model/mod.rs b/codex-rs/state/src/model/mod.rs index 190acf485..46fc5f344 100644 --- a/codex-rs/state/src/model/mod.rs +++ b/codex-rs/state/src/model/mod.rs @@ -107,6 +107,7 @@ pub use thread_metadata::ThreadMetadata; pub use thread_metadata::ThreadMetadataBuilder; pub use thread_metadata::ThreadsPage; pub use thread_monitor::ThreadMonitor; +pub use thread_monitor::ThreadMonitorAuthorization; pub use thread_monitor::ThreadMonitorEvent; pub use thread_monitor::ThreadMonitorEventStream; pub use thread_monitor::ThreadMonitorRouting; diff --git a/codex-rs/state/src/model/thread_monitor.rs b/codex-rs/state/src/model/thread_monitor.rs index 3c9985a95..6e958acd5 100644 --- a/codex-rs/state/src/model/thread_monitor.rs +++ b/codex-rs/state/src/model/thread_monitor.rs @@ -3,6 +3,11 @@ use anyhow::anyhow; use chrono::DateTime; use chrono::Utc; use codex_protocol::ThreadId; +use codex_protocol::models::PermissionProfile; +use serde::Deserialize; +use serde::Serialize; +use sha2::Digest; +use sha2::Sha256; use sqlx::Row; use sqlx::sqlite::SqliteRow; @@ -118,6 +123,7 @@ pub struct ThreadMonitor { pub output_file: Option, pub status: ThreadMonitorStatus, pub generation: i64, + pub authorization: Option, pub process_id: Option, pub last_event_at: Option>, pub last_error: Option, @@ -125,6 +131,79 @@ pub struct ThreadMonitor { pub updated_at: DateTime, } +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +pub struct ThreadMonitorAuthorization { + pub generation: i64, + pub writer_fence: String, + pub subject_digest: String, + pub permission_profile: PermissionProfile, + pub thread_cwd: String, +} + +impl ThreadMonitorAuthorization { + pub fn new( + thread_id: ThreadId, + generation: i64, + command: &str, + cwd: Option<&str>, + writer_fence: String, + permission_profile: PermissionProfile, + thread_cwd: String, + ) -> Self { + let subject_digest = thread_monitor_authorization_subject_digest( + thread_id, + generation, + command, + cwd, + &thread_cwd, + ); + Self { + generation, + writer_fence, + subject_digest, + permission_profile, + thread_cwd, + } + } + + pub fn authorizes( + &self, + monitor: &ThreadMonitor, + writer_fence: &str, + permission_profile: &PermissionProfile, + thread_cwd: &str, + ) -> bool { + self.generation == monitor.generation + && self.writer_fence == writer_fence + && self.permission_profile == *permission_profile + && self.thread_cwd == thread_cwd + && self.matches_subject( + monitor.thread_id, + monitor.generation, + monitor.command.as_str(), + monitor.cwd.as_deref(), + ) + } + + pub fn matches_subject( + &self, + thread_id: ThreadId, + generation: i64, + command: &str, + cwd: Option<&str>, + ) -> bool { + self.generation == generation + && self.subject_digest + == thread_monitor_authorization_subject_digest( + thread_id, + generation, + command, + cwd, + &self.thread_cwd, + ) + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct ThreadMonitorEvent { pub thread_id: ThreadId, @@ -146,6 +225,7 @@ pub(crate) struct ThreadMonitorRow { pub output_file: Option, pub status: String, pub generation: i64, + pub authorization_json: Option, pub process_id: Option, pub last_event_at_ms: Option, pub last_error: Option, @@ -166,6 +246,7 @@ impl ThreadMonitorRow { output_file: row.try_get("output_file")?, status: row.try_get("status")?, generation: row.try_get("generation")?, + authorization_json: row.try_get("authorization_json")?, process_id: row.try_get("process_id")?, last_event_at_ms: row.try_get("last_event_at_ms")?, last_error: row.try_get("last_error")?, @@ -190,6 +271,10 @@ impl TryFrom for ThreadMonitor { output_file: row.output_file, status: ThreadMonitorStatus::try_from(row.status.as_str())?, generation: row.generation, + authorization: row + .authorization_json + .as_deref() + .and_then(|value| serde_json::from_str(value).ok()), process_id: row.process_id, last_event_at: optional_epoch_millis_to_datetime(row.last_event_at_ms)?, last_error: row.last_error, @@ -236,6 +321,118 @@ impl TryFrom for ThreadMonitorEvent { } } +fn thread_monitor_authorization_subject_digest( + thread_id: ThreadId, + generation: i64, + command: &str, + cwd: Option<&str>, + thread_cwd: &str, +) -> String { + let mut hasher = Sha256::new(); + hash_monitor_authorization_field(&mut hasher, b"version", b"1"); + hash_monitor_authorization_field(&mut hasher, b"thread_id", thread_id.to_string().as_bytes()); + hash_monitor_authorization_field( + &mut hasher, + b"generation", + generation.to_string().as_bytes(), + ); + hash_monitor_authorization_field(&mut hasher, b"command", command.as_bytes()); + match cwd { + Some(cwd) => { + hash_monitor_authorization_field(&mut hasher, b"cwd_present", b"1"); + hash_monitor_authorization_field(&mut hasher, b"cwd", cwd.as_bytes()); + } + None => hash_monitor_authorization_field(&mut hasher, b"cwd_present", b"0"), + } + hash_monitor_authorization_field(&mut hasher, b"thread_cwd", thread_cwd.as_bytes()); + let digest = hasher.finalize(); + digest.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn hash_monitor_authorization_field(hasher: &mut Sha256, label: &[u8], value: &[u8]) { + hasher.update((label.len() as u64).to_le_bytes()); + hasher.update(label); + hasher.update((value.len() as u64).to_le_bytes()); + hasher.update(value); +} + fn optional_epoch_millis_to_datetime(value: Option) -> Result>> { value.map(epoch_millis_to_datetime).transpose() } + +#[cfg(test)] +mod tests { + use super::*; + + fn test_monitor(authorization: Option) -> ThreadMonitor { + let now = Utc::now(); + ThreadMonitor { + thread_id: ThreadId::new(), + monitor_id: "monitor-id".to_string(), + name: "monitor".to_string(), + prompt: "watch".to_string(), + command: "printf ok".to_string(), + cwd: Some("logs".to_string()), + routing: ThreadMonitorRouting::Stream, + output_file: None, + status: ThreadMonitorStatus::Running, + generation: 2, + authorization, + process_id: None, + last_event_at: None, + last_error: None, + created_at: now, + updated_at: now, + } + } + + #[test] + fn monitor_authorization_accepts_only_the_current_exact_subject_and_fence() { + let permission_profile = PermissionProfile::read_only(); + let thread_cwd = "/workspace".to_string(); + let mut monitor = test_monitor(/*authorization*/ None); + let authorization = ThreadMonitorAuthorization::new( + monitor.thread_id, + monitor.generation, + monitor.command.as_str(), + monitor.cwd.as_deref(), + "writer-fence".to_string(), + permission_profile.clone(), + thread_cwd.clone(), + ); + monitor.authorization = Some(authorization.clone()); + + assert!(authorization.authorizes( + &monitor, + "writer-fence", + &permission_profile, + thread_cwd.as_str(), + )); + assert!(!authorization.authorizes( + &monitor, + "stale-writer", + &permission_profile, + thread_cwd.as_str(), + )); + assert!(!authorization.authorizes( + &monitor, + "writer-fence", + &PermissionProfile::Disabled, + thread_cwd.as_str(), + )); + assert!(!authorization.authorizes( + &monitor, + "writer-fence", + &permission_profile, + "/different-workspace", + )); + + monitor.command = "printf changed".to_string(); + assert!(!authorization.authorizes( + &monitor, + "writer-fence", + &permission_profile, + thread_cwd.as_str(), + )); + } +} diff --git a/codex-rs/state/src/runtime/monitors.rs b/codex-rs/state/src/runtime/monitors.rs index 79bfaba8a..500223bd4 100644 --- a/codex-rs/state/src/runtime/monitors.rs +++ b/codex-rs/state/src/runtime/monitors.rs @@ -23,6 +23,7 @@ pub struct ThreadMonitorCreateParams { pub routing: crate::ThreadMonitorRouting, pub output_file: Option, pub status: crate::ThreadMonitorStatus, + pub authorization: Option, } pub struct ThreadMonitorUpdate { @@ -34,6 +35,7 @@ pub struct ThreadMonitorUpdate { pub output_file: Option>, pub status: Option, pub generation: Option, + pub authorization: Option, pub process_id: Option>, pub last_event_at: Option>>, pub last_error: Option>, @@ -53,6 +55,22 @@ impl MonitorStore { ) -> anyhow::Result { let monitor_id = Uuid::new_v4().to_string(); let now_ms = datetime_to_epoch_millis(Utc::now()); + let name = redact_state_string(params.name); + let prompt = redact_state_string(params.prompt); + let command = redact_state_string(params.command); + let cwd = redact_state_optional_string(params.cwd); + let output_file = redact_state_optional_string(params.output_file); + if let Some(authorization) = params.authorization.as_ref() + && !authorization.matches_subject( + params.thread_id, + /*generation*/ 0, + command.as_str(), + cwd.as_deref(), + ) + { + anyhow::bail!("thread monitor authorization does not match the create subject"); + } + let authorization_json = serialize_monitor_authorization(params.authorization.as_ref())?; let sql = thread_monitor_returning( r#" INSERT INTO thread_monitors ( @@ -65,22 +83,24 @@ INSERT INTO thread_monitors ( routing, output_file, status, + authorization_json, created_at_ms, updated_at_ms -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING "#, ); let row = sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(monitor_id) .bind(params.thread_id.to_string()) - .bind(redact_state_string(params.name)) - .bind(redact_state_string(params.prompt)) - .bind(redact_state_string(params.command)) - .bind(redact_state_optional_string(params.cwd)) + .bind(name) + .bind(prompt) + .bind(command) + .bind(cwd) .bind(params.routing.as_str()) - .bind(redact_state_optional_string(params.output_file)) + .bind(output_file) .bind(params.status.as_str()) + .bind(authorization_json) .bind(now_ms) .bind(now_ms) .fetch_one(self.pool.as_ref()) @@ -121,6 +141,7 @@ SELECT output_file, status, generation, + authorization_json, process_id, last_event_at_ms, last_error, @@ -151,6 +172,7 @@ SELECT output_file, status, generation, + authorization_json, process_id, last_event_at_ms, last_error, @@ -182,6 +204,7 @@ ORDER BY updated_at_ms, monitor_id let output_file = update.output_file.unwrap_or(existing.output_file); let status = update.status.unwrap_or(existing.status); let generation = update.generation.unwrap_or(existing.generation); + let authorization = update.authorization.or(existing.authorization); let process_id = update.process_id.unwrap_or(existing.process_id); let last_event_at = update.last_event_at.unwrap_or(existing.last_event_at); let last_error = update.last_error.unwrap_or(existing.last_error); @@ -191,6 +214,7 @@ ORDER BY updated_at_ms, monitor_id let cwd = redact_state_optional_string(cwd); let output_file = redact_state_optional_string(output_file); let last_error = redact_state_optional_string(last_error); + let authorization_json = serialize_monitor_authorization(authorization.as_ref())?; let sql = thread_monitor_returning( r#" UPDATE thread_monitors @@ -203,6 +227,7 @@ SET output_file = ?, status = ?, generation = ?, + authorization_json = ?, process_id = ?, last_event_at_ms = ?, last_error = ?, @@ -220,6 +245,7 @@ RETURNING .bind(output_file) .bind(status.as_str()) .bind(generation) + .bind(authorization_json) .bind(process_id) .bind(last_event_at.map(datetime_to_epoch_millis)) .bind(last_error) @@ -247,6 +273,7 @@ RETURNING output_file: None, status: Some(status), generation: None, + authorization: None, process_id: Some(None), last_event_at: None, last_error: Some(last_error), @@ -258,27 +285,51 @@ RETURNING pub async fn restart_thread_monitor( &self, monitor_id: &str, + expected_generation: i64, + authorization: crate::ThreadMonitorAuthorization, ) -> anyhow::Result> { let Some(existing) = self.get_thread_monitor(monitor_id).await? else { return Ok(None); }; - self.update_thread_monitor( - monitor_id, - ThreadMonitorUpdate { - name: None, - prompt: None, - command: None, - cwd: None, - routing: None, - output_file: None, - status: Some(crate::ThreadMonitorStatus::Running), - generation: Some(existing.generation + 1), - process_id: Some(None), - last_event_at: None, - last_error: Some(None), - }, - ) - .await + if existing.generation != expected_generation { + anyhow::bail!("thread monitor generation changed before restart authorization"); + } + let next_generation = expected_generation + 1; + if !authorization.matches_subject( + existing.thread_id, + next_generation, + existing.command.as_str(), + existing.cwd.as_deref(), + ) { + anyhow::bail!("thread monitor authorization does not match the restart subject"); + } + let authorization_json = serialize_monitor_authorization(Some(&authorization))?; + let sql = thread_monitor_returning( + r#" +UPDATE thread_monitors +SET + status = 'running', + generation = ?, + authorization_json = ?, + process_id = NULL, + last_error = NULL, + updated_at_ms = ? +WHERE monitor_id = ? AND generation = ? +RETURNING +"#, + ); + let row = sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(next_generation) + .bind(authorization_json) + .bind(datetime_to_epoch_millis(Utc::now())) + .bind(monitor_id) + .bind(expected_generation) + .fetch_optional(self.pool.as_ref()) + .await?; + match row { + Some(row) => Ok(Some(thread_monitor_from_row(&row)?)), + None => anyhow::bail!("thread monitor generation changed before restart commit"), + } } pub async fn mark_thread_monitor_started( @@ -304,6 +355,7 @@ RETURNING output_file: None, status: Some(crate::ThreadMonitorStatus::Running), generation: None, + authorization: None, process_id: Some(process_id), last_event_at: None, last_error: Some(None), @@ -467,6 +519,7 @@ fn thread_monitor_select_columns() -> &'static str { output_file, status, generation, + authorization_json, process_id, last_event_at_ms, last_error, @@ -487,6 +540,14 @@ fn thread_monitor_select_by_id(prefix: &'static str) -> String { ) } +fn serialize_monitor_authorization( + authorization: Option<&crate::ThreadMonitorAuthorization>, +) -> anyhow::Result> { + authorization + .map(crate::redacted_local_state_serialized_json_string) + .transpose() +} + #[cfg(test)] mod tests { use super::*; @@ -525,17 +586,27 @@ mod tests { runtime: &StateRuntime, thread_id: ThreadId, ) -> crate::ThreadMonitor { + let command = "while true; do echo ok; sleep 60; done"; runtime .thread_monitors() .create_thread_monitor(ThreadMonitorCreateParams { thread_id, name: "CI watcher".to_string(), prompt: "watch CI".to_string(), - command: "while true; do echo ok; sleep 60; done".to_string(), + command: command.to_string(), cwd: None, routing: crate::ThreadMonitorRouting::Stream, output_file: None, status: crate::ThreadMonitorStatus::Running, + authorization: Some(crate::ThreadMonitorAuthorization::new( + thread_id, + /*generation*/ 0, + command, + /*cwd*/ None, + "writer-fence".to_string(), + codex_protocol::models::PermissionProfile::read_only(), + "/workspace".to_string(), + )), }) .await .expect("monitor should be created") @@ -552,6 +623,7 @@ mod tests { assert_eq!("CI watcher", created.name); assert_eq!(crate::ThreadMonitorStatus::Running, created.status); assert_eq!(0, created.generation); + assert!(created.authorization.is_some()); let event = runtime .thread_monitors() @@ -576,14 +648,49 @@ mod tests { .expect("events should list"); assert_eq!(vec![event], events); + let restart_authorization = crate::ThreadMonitorAuthorization::new( + thread_id, + /*generation*/ 1, + created.command.as_str(), + created.cwd.as_deref(), + "writer-fence".to_string(), + codex_protocol::models::PermissionProfile::read_only(), + "/workspace".to_string(), + ); let restarted = runtime .thread_monitors() - .restart_thread_monitor(created.monitor_id.as_str()) + .restart_thread_monitor( + created.monitor_id.as_str(), + /*expected_generation*/ 0, + restart_authorization, + ) .await .expect("restart should succeed") .expect("monitor should exist"); assert_eq!(crate::ThreadMonitorStatus::Running, restarted.status); assert_eq!(1, restarted.generation); + assert!(restarted.authorization.is_some()); + let stale_authorization = crate::ThreadMonitorAuthorization::new( + thread_id, + /*generation*/ 1, + created.command.as_str(), + created.cwd.as_deref(), + "writer-fence".to_string(), + codex_protocol::models::PermissionProfile::read_only(), + "/workspace".to_string(), + ); + assert!( + runtime + .thread_monitors() + .restart_thread_monitor( + created.monitor_id.as_str(), + /*expected_generation*/ 0, + stale_authorization, + ) + .await + .is_err(), + "a stale generation must not commit a restart authorization" + ); let deleted = runtime .thread_monitors() @@ -619,6 +726,7 @@ mod tests { routing: crate::ThreadMonitorRouting::Stream, output_file: None, status: crate::ThreadMonitorStatus::Running, + authorization: None, }) .await .expect("monitor should be created"); diff --git a/codex-rs/state/src/runtime/workflow_automation.rs b/codex-rs/state/src/runtime/workflow_automation.rs index aa19ce5e0..c4fd630da 100644 --- a/codex-rs/state/src/runtime/workflow_automation.rs +++ b/codex-rs/state/src/runtime/workflow_automation.rs @@ -1325,6 +1325,7 @@ WHERE timer_id = ? routing: crate::ThreadMonitorRouting::Stream, output_file: None, status: crate::ThreadMonitorStatus::Running, + authorization: None, }) .await .expect("monitor should create"); @@ -1408,6 +1409,7 @@ WHERE timer_id = ? routing: crate::ThreadMonitorRouting::Stream, output_file: None, status: crate::ThreadMonitorStatus::Running, + authorization: None, }) .await .expect("monitor should create");