feat(app-server): bind stdio interrupts to the registering connection (phase 6.2)

Adds a process-unique ConnectionId minted per stdio connection and records it on
InFlightTurn. interrupt_stdio_turn now refuses a different connection with a
typed not_turn_owner error (-32012); interrupts with no connection (HTTP, tests,
legacy) behave as before. A second registration for a live thread key is
refused instead of overwriting the first turn's entry, and the refused claim
cannot retire the other turn's registration. The full-tool turn/start gate stays
closed.

Verified (local, existing checks, source only): the stdio, interrupt,
prompt_request, socket_transport, full_control and goal checks and the
daemon_socket integration tests are in the 578-check run (577 passed; the one
failure was an unrelated docs row-length check, since fixed). Not verified: a
real two-connection native observer; a daemon guest interrupting an owner turn
now gets not_turn_owner, and no existing test relied on that.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>

crates/app-server/src/bridge.rs

@@ -303,6 +303,7 @@ impl RuntimeBridge {
                 },
                 runtime_thread_id: thread_id.to_string(),
                 turn_id: turn_id.clone(),
+                connection: None,
             };
             if let Err(interrupt) = interrupt_turn_request(&orphan).await {
                 tracing::warn!(
@@ -328,6 +329,7 @@ impl RuntimeBridge {
                     },
                     runtime_thread_id: thread_id.to_string(),
                     turn_id: turn_id.clone(),
+                    connection: None,
                 },
             );
         }
@@ -377,6 +379,7 @@ impl RuntimeBridge {
                 },
                 runtime_thread_id: thread_id.to_string(),
                 turn_id: turn_id.clone(),
+                connection: None,
             };
             if let Err(error) = interrupt_turn_request(&orphan).await {
                 tracing::warn!(

crates/app-server/src/connection.rs

@@ -2,12 +2,41 @@ use crate::{DispatchError, params_or_object, parse_params};
 use codewhale_app_server_protocol::lifecycle::{InitializeParams, InitializeResponse};
 use serde_json::Value;
 
-#[derive(Default)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub(crate) struct ConnectionId(u64);
+
+impl ConnectionId {
+    fn mint() -> Self {
+        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
+        Self(NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed))
+    }
+}
+
+impl std::fmt::Display for ConnectionId {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
 pub(crate) struct ConnectionSession {
+    id: ConnectionId,
     initialization: Option<InitializeParams>,
 }
 
+impl Default for ConnectionSession {
+    fn default() -> Self {
+        Self {
+            id: ConnectionId::mint(),
+            initialization: None,
+        }
+    }
+}
+
 impl ConnectionSession {
+    pub(crate) fn id(&self) -> ConnectionId {
+        self.id
+    }
+
     pub(crate) fn initialized(&self) -> bool {
         self.initialization.is_some()
     }

crates/app-server/src/daemon_socket.rs

@@ -312,6 +312,7 @@ where
                     request.params,
                     AppTransport::Socket,
                     None,
+                    None,
                 )
                 .await
                 {

crates/app-server/src/lib.rs

@@ -370,6 +370,7 @@ struct InFlightTurn {
     /// Thread id as the *runtime* knows it, not the stdio-facing id.
     runtime_thread_id: String,
     turn_id: String,
+    connection: Option<connection::ConnectionId>,
 }
 
 type TurnRegistry = Arc<Mutex<HashMap<String, InFlightTurn>>>;
@@ -1000,6 +1001,19 @@ impl DispatchError {
         }
     }
 
+    /// Server error (-32000..-32099): only the connection that started a
+    /// turn may interrupt it.
+    fn not_turn_owner() -> Self {
+        Self {
+            wire: RpcErrorDetails {
+                code: NOT_DAEMON_OWNER_CODE,
+                message: "only the connection that started this turn may interrupt it".to_string(),
+                data: Some(json!({ "error": "not_turn_owner" })),
+            },
+            transport_error: None,
+        }
+    }
+
     /// Server error (-32000..-32099): the client's expected daemon version
     /// does not match the running binary (bundle skew).
     #[cfg(any(unix, windows))]

crates/app-server/src/stdio.rs

@@ -1,5 +1,5 @@
 use crate::config_rpc::process_app_request;
-use crate::turns::{NativeTurn, handle_prompt_request, handle_thread_request, run_native_turn};
+use crate::turns::{NativeTurn, handle_prompt_request, handle_thread_request, run_native_turn_for};
 use crate::{
     AppState, AppTransport, ConfigGetParams, ConfigSetParams, DispatchError, InFlightTurn, RpcCall,
     RuntimeOwnerRequest, RuntimeThreadHint, ShutdownAuthority, StdioDispatchResult, StdioLoopExit,
@@ -190,7 +190,10 @@ where
         };
         if request.method == "initialize" {
             let response = match connection.initialize(request.params) {
-                Ok(result) => jsonrpc_result(id, result),
+                Ok(result) => {
+                    tracing::debug!(connection = %connection.id(), "control connection initialized");
+                    jsonrpc_result(id, result)
+                }
                 Err(error) => jsonrpc_error(id, error),
             };
             write_stdio_line(&mut writer, &response).await?;
@@ -231,6 +234,7 @@ where
                 request.params,
                 policy.transport,
                 Some(cancellation.clone()),
+                Some(connection.id()),
             );
             tokio::pin!(dispatch);
             loop {
@@ -256,7 +260,7 @@ where
                             Ok(Some(line)) => {
                                 if matches!(parse_stdio_line(&line), ParsedStdioLine::Request(ref request) if is_control_detach(request)) {
                                     stdin_open=false; drop(input_claim.take());
-                                } else if let Err(error) = handle_line_during_turn(state, &line, &mut pending, &mut pending_bytes, policy).await {
+                                } else if let Err(error) = handle_line_during_turn(state, &line, &mut pending, &mut pending_bytes, policy, Some(connection.id())).await {
                                     stdin_open = false;
                                     drop(input_claim.take());
                                     cancellation.cancel();
@@ -275,6 +279,7 @@ where
                 request.params,
                 policy.transport,
                 None,
+                Some(connection.id()),
             )
             .await
         };
@@ -380,6 +385,7 @@ pub(crate) async fn handle_line_during_turn(
     pending: &mut VecDeque<(PendingStdioWork, usize)>,
     pending_bytes: &mut usize,
     policy: StdioLoopPolicy,
+    connection: Option<crate::connection::ConnectionId>,
 ) -> Result<()> {
     let request = match parse_stdio_line(line) {
         ParsedStdioLine::Blank => return Ok(()),
@@ -396,13 +402,15 @@ pub(crate) async fn handle_line_during_turn(
             let response = match parse_params::<ThreadInterruptParams>(params_or_object(
                 request.params.clone(),
             )) {
-                Ok(parsed) => match interrupt_stdio_turn(state, &parsed.thread_id).await {
-                    Ok(interrupted) => jsonrpc_result(
-                        id,
-                        json!({ "thread_id": parsed.thread_id, "interrupted": interrupted }),
-                    ),
-                    Err(err) => jsonrpc_error(id, err),
-                },
+                Ok(parsed) => {
+                    match interrupt_stdio_turn(state, &parsed.thread_id, connection).await {
+                        Ok(interrupted) => jsonrpc_result(
+                            id,
+                            json!({ "thread_id": parsed.thread_id, "interrupted": interrupted }),
+                        ),
+                        Err(err) => jsonrpc_error(id, err),
+                    }
+                }
                 Err(err) => jsonrpc_error(id, err),
             };
             queue_stdio_work(
@@ -543,8 +551,9 @@ pub(crate) async fn handle_stdio_thread_message<W: AsyncWrite + Unpin>(
     writer: &mut W,
     parsed: ThreadMessageParams,
     cancellation: Option<CancellationToken>,
+    connection: Option<crate::connection::ConnectionId>,
 ) -> std::result::Result<Value, DispatchError> {
-    let mut result = run_native_turn(
+    let mut result = run_native_turn_for(
         state,
         writer,
         NativeTurn {
@@ -559,6 +568,7 @@ pub(crate) async fn handle_stdio_thread_message<W: AsyncWrite + Unpin>(
             cancellation,
         },
         None,
+        connection,
     )
     .await?;
     if let Some(object) = result.as_object_mut() {
@@ -690,10 +700,16 @@ pub(crate) async fn interrupt_turn_request(
 pub(crate) async fn interrupt_stdio_turn(
     state: &AppState,
     thread_id: &str,
+    requester: Option<crate::connection::ConnectionId>,
 ) -> std::result::Result<bool, DispatchError> {
     let Some(turn) = state.in_flight_turns.lock().await.get(thread_id).cloned() else {
         return Ok(false);
     };
+    if let (Some(owner), Some(requester)) = (turn.connection, requester)
+        && owner != requester
+    {
+        return Err(DispatchError::not_turn_owner());
+    }
     let interrupted = interrupt_turn_request(&turn).await?;
     retire_turn_registration(&state.in_flight_turns, thread_id, &turn).await;
     Ok(interrupted)
@@ -779,8 +795,16 @@ pub(crate) async fn dispatch_stdio_request(
     params: Value,
 ) -> std::result::Result<StdioDispatchResult, DispatchError> {
     let mut sink = tokio::io::sink();
-    dispatch_stdio_request_with_writer(state, &mut sink, method, params, AppTransport::Stdio, None)
-        .await
+    dispatch_stdio_request_with_writer(
+        state,
+        &mut sink,
+        method,
+        params,
+        AppTransport::Stdio,
+        None,
+        None,
+    )
+    .await
 }
 
 pub(crate) async fn dispatch_stdio_app_request(
@@ -803,6 +827,7 @@ pub(crate) async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
     params: Value,
     transport: AppTransport,
     cancellation: Option<CancellationToken>,
+    connection: Option<crate::connection::ConnectionId>,
 ) -> std::result::Result<StdioDispatchResult, DispatchError> {
     let outcome = match method {
         "healthz" | "app/healthz" => StdioDispatchResult {
@@ -921,6 +946,7 @@ pub(crate) async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
                         max_output_tokens,
                     },
                     cancellation.clone(),
+                    connection,
                 )
                 .await?;
                 return Ok(StdioDispatchResult {
@@ -1178,8 +1204,14 @@ pub(crate) async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
         }
         "thread/message" => {
             let parsed: ThreadMessageParams = parse_params(params_or_object(params))?;
-            let response =
-                handle_stdio_thread_message(state, writer, parsed, cancellation.clone()).await?;
+            let response = handle_stdio_thread_message(
+                state,
+                writer,
+                parsed,
+                cancellation.clone(),
+                connection,
+            )
+            .await?;
             StdioDispatchResult {
                 result: response,
                 should_exit: false,
@@ -1250,7 +1282,7 @@ pub(crate) async fn dispatch_stdio_request_with_writer<W: AsyncWrite + Unpin>(
         }
         "thread/interrupt" => {
             let parsed: ThreadInterruptParams = parse_params(params_or_object(params))?;
-            let interrupted = interrupt_stdio_turn(state, &parsed.thread_id).await?;
+            let interrupted = interrupt_stdio_turn(state, &parsed.thread_id, connection).await?;
             StdioDispatchResult {
                 result: json!({
                     "thread_id": parsed.thread_id,

crates/app-server/src/tests.rs

@@ -204,6 +204,7 @@ async fn full_control_queue_keeps_interrupt_priority_and_guest_shutdown_refusal(
         },
         runtime_thread_id: "runtime".into(),
         turn_id: "turn".into(),
+        connection: None,
     };
     state
         .in_flight_turns
@@ -231,7 +232,8 @@ async fn full_control_queue_keeps_interrupt_priority_and_guest_shutdown_refusal(
             r#"{"id":1,"method":"shutdown","params":{}}"#,
             &mut pending,
             &mut bytes,
-            guest
+            guest,
+            None
         )
         .await
         .is_err()
@@ -247,7 +249,8 @@ async fn full_control_queue_keeps_interrupt_priority_and_guest_shutdown_refusal(
             r#"{"id":2,"method":"thread/interrupt","params":{"thread_id":"client"}}"#,
             &mut pending,
             &mut bytes,
-            guest
+            guest,
+            None
         )
         .await
         .is_err()
@@ -276,7 +279,8 @@ async fn full_control_queue_keeps_interrupt_priority_and_guest_shutdown_refusal(
             r#"{"id":3,"method":"shutdown","params":{}}"#,
             &mut pending,
             &mut bytes,
-            owner
+            owner,
+            None
         )
         .await
         .is_err()
@@ -2145,7 +2149,7 @@ async fn interrupt_stops_a_turn_that_would_otherwise_stream_forever() {
                         "turn_runaway"
                     );
                     assert!(
-                        interrupt_stdio_turn(&state, "thr_a")
+                        interrupt_stdio_turn(&state, "thr_a", None)
                             .await
                             .expect("explicit recovery interrupt")
                     );
@@ -2759,6 +2763,7 @@ async fn prompt_request_executes_a_genuine_model_turn() {
         json!({ "allowed_tools": [], "prompt": "what is 2+2" }),
         AppTransport::Stdio,
         None,
+        None,
     )
     .await
     .expect("prompt/request dispatch");
@@ -3148,6 +3153,7 @@ async fn socket_transport_advertises_daemon_attach() {
         json!({}),
         AppTransport::Socket,
         None,
+        None,
     )
     .await
     .expect("capabilities dispatch");

crates/app-server/src/turns.rs

@@ -59,6 +59,16 @@ pub(crate) async fn run_native_turn<W: AsyncWrite + Unpin>(
     writer: &mut W,
     turn: NativeTurn<'_>,
     transcript: Option<&mut TurnTranscript>,
+) -> std::result::Result<Value, DispatchError> {
+    run_native_turn_for(state, writer, turn, transcript, None).await
+}
+
+pub(crate) async fn run_native_turn_for<W: AsyncWrite + Unpin>(
+    state: &AppState,
+    writer: &mut W,
+    turn: NativeTurn<'_>,
+    transcript: Option<&mut TurnTranscript>,
+    connection: Option<crate::connection::ConnectionId>,
 ) -> std::result::Result<Value, DispatchError> {
     if !matches!(turn.allowed_tools, Some([])) {
         return Err(DispatchError::unsupported_interactive_execution());
@@ -153,6 +163,7 @@ pub(crate) async fn run_native_turn<W: AsyncWrite + Unpin>(
             writer,
             registration,
             transcript,
+            connection,
         )
         .await
         .map_err(|error| {
@@ -385,6 +396,7 @@ async fn message_native_thread<W: AsyncWrite + Unpin>(
     writer: &mut W,
     registration: Option<(TurnRegistry, String)>,
     mut transcript: Option<&mut TurnTranscript>,
+    connection: Option<crate::connection::ConnectionId>,
 ) -> Result<Value> {
     anyhow::ensure!(
         !input
@@ -436,6 +448,7 @@ async fn message_native_thread<W: AsyncWrite + Unpin>(
         owner: TurnExecutionOwner::Native(owner),
         runtime_thread_id: thread_id.to_owned(),
         turn_id: turn_id.clone(),
+        connection,
     };
     let mut claim = NativeTurnClaim {
         live,
@@ -459,7 +472,14 @@ async fn message_native_thread<W: AsyncWrite + Unpin>(
     }
     let result = async {
         if let Some((registry, key)) = claim.registration.as_ref() {
-            registry.lock().await.insert(key.clone(), claim.live.clone());
+            match registry.lock().await.entry(key.clone()) {
+                std::collections::hash_map::Entry::Occupied(_) => anyhow::bail!(
+                    "a turn is already registered for thread {key}; refusing a second registration"
+                ),
+                std::collections::hash_map::Entry::Vacant(slot) => {
+                    slot.insert(claim.live.clone());
+                }
+            }
             claim.registration_published = true;
         }
         anyhow::ensure!(