fix(monitors): fence persistent command authorization

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<Arc<CodexThread>, 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<CodexThread>,
+        generation: i64,
+        command: &str,
+        cwd: Option<&str>,
+    ) -> Result<codex_state::ThreadMonitorAuthorization, JSONRPCErrorError> {
+        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::<CommandExecutionRequestApprovalResponse>(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<ThreadId, JSON
     ThreadId::from_string(thread_id)
         .map_err(|err| invalid_request(format!("invalid thread id: {err}")))
 }
+
+fn monitor_approval_cwd(
+    thread_cwd: &AbsolutePathBuf,
+    cwd: Option<&str>,
+) -> Result<AbsolutePathBuf, JSONRPCErrorError> {
+    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
+        ));
+    }
+}