From faeb08c1e139590a1e01929d2bf289c2a6c47e2e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 15 Jan 2026 14:59:07 +0000 Subject: [PATCH] feat: add interrupt capabilities to `send_input` (#9276) --- codex-rs/core/src/agent/control.rs | 6 +++ codex-rs/core/src/tools/handlers/collab.rs | 49 ++++++++++++++++++++++ codex-rs/core/src/tools/spec.rs | 9 ++++ 3 files changed, 64 insertions(+) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index c8ba7fdda..7083081dc 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/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 { + 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 { let state = self.upgrade()?; diff --git a/codex-rs/core/src/tools/handlers/collab.rs b/codex-rs/core/src/tools/handlers/collab.rs index 3c24cf137..df3f2901d 100644 --- a/codex-rs/core/src/tools/handlers/collab.rs +++ b/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; diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index 7e14e7697..1825a12cc 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/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(),