feat: add interrupt capabilities to send_input (#9276)

codex-rs/core/src/agent/control.rs

@@ -71,6 +71,12 @@ impl AgentControl {
         result
     }
 
+    /// Interrupt the current task for an existing agent thread.
+    pub(crate) async fn interrupt_agent(&self, agent_id: ThreadId) -> CodexResult<String> {
+        let state = self.upgrade()?;
+        state.send_op(agent_id, Op::Interrupt).await
+    }
+
     /// Submit a shutdown request to an existing agent thread.
     pub(crate) async fn shutdown_agent(&self, agent_id: ThreadId) -> CodexResult<String> {
         let state = self.upgrade()?;

codex-rs/core/src/tools/handlers/collab.rs

@@ -170,6 +170,8 @@ mod send_input {
     struct SendInputArgs {
         id: String,
         message: String,
+        #[serde(default)]
+        interrupt: bool,
     }
 
     #[derive(Debug, Serialize)]
@@ -191,6 +193,14 @@ mod send_input {
                 "Empty message can't be sent to an agent".to_string(),
             ));
         }
+        if args.interrupt {
+            session
+                .services
+                .agent_control
+                .interrupt_agent(receiver_thread_id)
+                .await
+                .map_err(|err| collab_agent_error(receiver_thread_id, err))?;
+        }
         session
             .send_event(
                 &turn,
@@ -723,6 +733,45 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn send_input_interrupts_before_prompt() {
+        let (mut session, turn) = make_session_and_context().await;
+        let manager = thread_manager();
+        session.services.agent_control = manager.agent_control();
+        let config = turn.client.config().as_ref().clone();
+        let thread = manager.start_thread(config).await.expect("start thread");
+        let agent_id = thread.thread_id;
+        let invocation = invocation(
+            Arc::new(session),
+            Arc::new(turn),
+            "send_input",
+            function_payload(json!({
+                "id": agent_id.to_string(),
+                "message": "hi",
+                "interrupt": true
+            })),
+        );
+        CollabHandler
+            .handle(invocation)
+            .await
+            .expect("send_input should succeed");
+
+        let ops = manager.captured_ops();
+        let ops_for_agent: Vec<&Op> = ops
+            .iter()
+            .filter_map(|(id, op)| (*id == agent_id).then_some(op))
+            .collect();
+        assert_eq!(ops_for_agent.len(), 2);
+        assert!(matches!(ops_for_agent[0], Op::Interrupt));
+        assert!(matches!(ops_for_agent[1], Op::UserInput { .. }));
+
+        let _ = thread
+            .thread
+            .submit(Op::Shutdown {})
+            .await
+            .expect("shutdown should submit");
+    }
+
     #[tokio::test]
     async fn wait_rejects_non_positive_timeout() {
         let (session, turn) = make_session_and_context().await;

codex-rs/core/src/tools/spec.rs

@@ -478,6 +478,15 @@ fn create_send_input_tool() -> ToolSpec {
             description: Some("Message to send to the agent.".to_string()),
         },
     );
+    properties.insert(
+        "interrupt".to_string(),
+        JsonSchema::Boolean {
+            description: Some(
+                "When true, interrupt the agent's current task before sending the message. When false (default), the message will be processed when the agent is done on its current task."
+                    .to_string(),
+            ),
+        },
+    );
 
     ToolSpec::Function(ResponsesApiTool {
         name: "send_input".to_string(),