hive-subagent-mcp: refuse an ACP interrupt with no turn in flight

`Canceller::cancel()` does nothing and returns `false` when no ACP turn
is in flight: from `start` until the run's task begins its turn, and
from a turn's end until `after_turn` releases the name. `interrupt`
ignored that return, recorded `Cancelled` and replied that the run was
cancelled, so a non-goal run went on to do its whole first turn while
`status` read as cancelled.

On ACP, `interrupt` now records the stop, cancels, and if nothing was
in flight withdraws the stop under the same `stops` lock, restores the
tracking entry, and refuses with the claude path's "still starting, try
again shortly". The claude arm is unchanged.

Review finding on #4824 (argus).

hive-subagent-mcp/src/session.rs

@@ -2200,8 +2200,8 @@ fn describe_stopped(name: &str, stop: &StopReason, turns: &str) -> String {
 ///
 /// # Errors
 ///
-/// An invalid name, nothing tracked under `name`, or `name` has no confirmed
-/// process right now (a spawn in flight, or a goal run between turns).
+/// An invalid name, nothing tracked under `name`, or nothing to signal right
+/// now (a spawn or ACP turn not yet live, or a goal run between turns).
 pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<String> {
     validate_name(name)?;
     let mut running = state.running.lock().unwrap_or_else(PoisonError::into_inner);
@@ -2209,29 +2209,46 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<Strin
         None => anyhow::bail!("no subagent named `{name}` is currently running"),
         Some(None) => {
             running.insert(name.to_owned(), None);
-            anyhow::bail!(
-                "subagent `{name}` is still starting — not yet confirmed running, try again \
-                 shortly"
-            );
+            return Err(still_starting(name));
         }
-        Some(Some(stopper)) => {
+        Some(Some(Stopper::Claude(cancel))) => {
             state.record_stop(name, StopReason::Cancelled);
             drop(running);
-            match stopper {
-                Stopper::Claude(cancel) => cancel.cancel(force),
-                // `force` has no stronger form on ACP: the runtime already kills
-                // an agent that ignores `session/cancel`.
-                Stopper::Acp(canceller) => {
-                    let _ = canceller.cancel();
+            cancel.cancel(force);
+        }
+        // `force` has no stronger form on ACP: the runtime already kills an
+        // agent that ignores `session/cancel`. `cancel()` is `false` when no
+        // turn is in flight (the agent not prompted yet, or a turn just
+        // ended), and then nothing was cancelled. The stop goes in before the
+        // cancel and comes out under the same `stops` lock, so
+        // `plan_after_turn` never reads a cancel that didn't happen.
+        Some(Some(Stopper::Acp(canceller))) => {
+            let mut stops = state.stops.lock().unwrap_or_else(PoisonError::into_inner);
+            let prior = stops.insert(name.to_owned(), StopReason::Cancelled);
+            if !canceller.cancel() {
+                match prior {
+                    Some(prior) => stops.insert(name.to_owned(), prior),
+                    None => stops.remove(name),
+                };
+                drop(stops);
+                running.insert(name.to_owned(), Some(Stopper::Acp(canceller)));
+                return Err(still_starting(name));
+            }
         }
     }
     Ok(format!(
-                "interrupt sent to subagent `{name}` — the run is cancelled, not just the turn \
-                 that was in flight, so no further goal turn will start. `continue` is what \
-                 restarts it, with a fresh turn allowance."
+        "interrupt sent to subagent `{name}` — the run is cancelled, not just the turn that was \
+         in flight, so no further goal turn will start. `continue` is what restarts it, with a \
+         fresh turn allowance."
     ))
 }
-    }
+
+/// `interrupt`'s refusal for a name that is claimed but has nothing to
+/// signal yet.
+fn still_starting(name: &str) -> anyhow::Error {
+    anyhow::anyhow!(
+        "subagent `{name}` is still starting — not yet confirmed running, try again shortly"
+    )
 }
 
 /// Push `name`'s one-shot end-of-turn todo. Best-effort: a connect/write
@@ -4308,6 +4325,31 @@ done
         std::fs::remove_dir_all(&dir).ok();
     }
 
+    #[tokio::test]
+    async fn an_interrupt_before_the_acp_turn_is_live_is_refused() {
+        let dir = scratch_dir("runtime-acp-early-cancel");
+        let state = State::new(PathBuf::from("/dev/null"), signal_url());
+        let state = Arc::new(on_acp_agent(state, &dir, "answer"));
+        assert!(state.reserve("n"));
+        start_on_runtime(&state, "n", claude_config(&dir), None, "go".into())
+            .expect("the ACP run starts");
+
+        // The test runtime is single-threaded, so the run's task has not
+        // been polled yet: no turn is in flight.
+        let err = interrupt(&state, "n", false).expect_err("nothing to cancel yet");
+        assert!(err.to_string().contains("still starting"), "{err}");
+        assert_eq!(state.stop_reason("n"), None);
+
+        until("the run ends", || state.occupancy("n").is_none()).await;
+        assert_eq!(lines(&dir.join("prompts")).len(), 1);
+        assert_eq!(
+            state.stop_reason("n"),
+            Some(StopReason::Done),
+            "the run was never marked cancelled"
+        );
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
     #[test]
     fn an_acp_subagent_gets_the_tools_a_claude_subagent_would() {
         let ask = |kind| PermissionAsk {