(MOT-4100) fix(harness): stop button — prompt cancellation end to end

console/web/src/components/chat/ChatView.tsx

@@ -179,6 +179,20 @@ export function ChatView({
   const serverWorking = conversation.status === 'working'
   const streamingIndicator = isStreaming || serverWorking
 
+  // Stop requested but not yet finalized server-side. Disables the stop button
+  // until status-changed flips the indicator off. The ref is the dedupe guard:
+  // synchronous (two clicks in one frame can't both pass, unlike closure
+  // state) and readable outside a state updater (React forbids side effects
+  // inside updaters — Strict Mode double-invokes them).
+  const [stopping, setStopping] = useState(false)
+  const stopRequestedRef = useRef(false)
+  useEffect(() => {
+    if (!streamingIndicator) {
+      stopRequestedRef.current = false
+      setStopping(false)
+    }
+  }, [streamingIndicator])
+
   // Messages queued mid-stream (MOT-3837): shown above the composer until the
   // harness drains them into the transcript. Each draft carries the predicted
   // entry id of its eventual transcript row.
@@ -1278,8 +1292,17 @@ export function ChatView({
   )
 
   const handleStop = useCallback(() => {
+    if (stopRequestedRef.current) return
+    stopRequestedRef.current = true
+    setStopping(true)
     abortRef.current?.abort()
-    void backend.abortRun?.(sessionId).catch(() => {})
+    // Re-enable the button if the stop RPC fails while the server still
+    // reports working — otherwise `stopping` never clears (the reset
+    // effect waits on the indicator) and the user can't retry.
+    void backend.abortRun?.(sessionId).catch(() => {
+      stopRequestedRef.current = false
+      setStopping(false)
+    })
   }, [backend, sessionId])
 
   // Rescue a parked stream loop: the session hit a terminal error server-side
@@ -1729,6 +1752,7 @@ export function ChatView({
             onTextChange={handleComposerTextChange}
             onSubmit={handleSubmit}
             onStop={handleStop}
+            stopping={stopping}
             queuedForEdit={backend.editQueued ? queuedForEdit : undefined}
             onEditQueued={backend.editQueued ? handleEditQueued : undefined}
             onBrowseChange={setBrowsedQueuedId}

console/web/src/components/chat/Composer.tsx

@@ -4,7 +4,7 @@ import {
   $getRoot,
   type LexicalEditor,
 } from 'lexical'
-import { ArrowUp, Square } from 'lucide-react'
+import { ArrowUp, Loader2, Square } from 'lucide-react'
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
 import { PermissionModePicker } from '@/components/permissions/PermissionModePicker'
 import type { PermissionMode } from '@/lib/backend/approval-settings'
@@ -87,6 +87,8 @@ interface ComposerProps {
   onPermissionModeChange: (next: PermissionMode) => void
   onSubmit: (payload: ComposerSubmitPayload) => void
   onStop?: () => void
+  /** A stop was requested and the server hasn't finalized the turn yet. */
+  stopping?: boolean
   isStreaming?: boolean
   /**
    * When true, the editor stays unlocked while streaming: a submit queues the
@@ -160,6 +162,7 @@ export function Composer({
   onPermissionModeChange,
   onSubmit,
   onStop,
+  stopping,
   isStreaming,
   queueWhileStreaming,
   blocked,
@@ -389,10 +392,15 @@ export function Composer({
             <button
               type="button"
               onClick={onStop}
-              aria-label="stop generating"
+              disabled={stopping}
+              aria-label={stopping ? 'stopping' : 'stop generating'}
               className={actionButtonClass}
             >
-              <Square size={16} aria-hidden className="fill-black/90" />
+              {stopping ? (
+                <Loader2 size={16} aria-hidden className="animate-spin" />
+              ) : (
+                <Square size={16} aria-hidden className="fill-black/90" />
+              )}
             </button>
           ) : (
             <button

console/web/src/lib/backend/translate.test.ts

@@ -177,7 +177,12 @@ describe('translateTurnSource — turn-completed', () => {
         completed({ status: 'cancelled', reason: 'stopped' }),
       ),
     ).toEqual([
-      { kind: 'stop-reason', reason: 'aborted', message: 'stopped' },
+      {
+        kind: 'stop-reason',
+        reason: 'aborted',
+        message: 'stopped',
+        entryId: 'e_t-1_stopped',
+      },
       { kind: 'assistant-end' },
     ])
   })

console/web/src/lib/backend/translate.ts

@@ -108,7 +108,14 @@ export function translateTurnSource(event: TurnSourceEvent): StreamEvent[] {
 function translateTurnCompleted(event: TurnCompletedEvent): StreamEvent[] {
   const out: StreamEvent[] = []
   if (event.status === 'cancelled') {
-    out.push({ kind: 'stop-reason', reason: 'aborted', message: event.reason })
+    out.push({
+      kind: 'stop-reason',
+      reason: 'aborted',
+      message: event.reason,
+      // Same id as the harness's durable "stopped" notice entry, so the live
+      // notice and the reconciled transcript row dedupe instead of doubling.
+      entryId: `e_${event.turn_id}_stopped`,
+    })
   } else if (event.status === 'failed' || event.result_error) {
     const message = event.result_error ?? event.reason
     out.push({

harness/src/clients/router.rs

@@ -19,7 +19,7 @@ use iii_sdk::protocol::TriggerRequest;
 use iii_sdk::IIIClient;
 use serde_json::{json, Value};
 use tokio::sync::mpsc;
-use tokio::sync::Notify;
+use tokio::sync::{watch, Notify};
 
 use crate::error::HarnessError;
 use crate::types::event::{AssistantMessageEvent, StopReason};
@@ -70,10 +70,17 @@ impl RouterClient {
     }
 
     /// Stream one assistant turn, forwarding coalesced partials to `sink`.
+    ///
+    /// `abort_rx` is the harness-local cancel backstop: `router::abort` is the
+    /// happy path (the router closes the stream → EOF), but it's a no-op when
+    /// it races ahead of the router's stream registration or after a router
+    /// restart — then only this local signal cuts the await (which otherwise
+    /// blocks up to `router_timeout_ms`).
     pub async fn chat(
         &self,
         params: ChatParams,
         sink: &dyn StreamSink,
+        mut abort_rx: watch::Receiver<bool>,
     ) -> Result<ChatOutcome, HarnessError> {
         let channel = create_channel(&self.iii, None)
             .await
@@ -113,6 +120,7 @@ impl RouterClient {
             }
         });
 
+        let request_id = params.request_id.clone();
         let mut payload = json!({
             "writer_ref": channel.writer_ref,
             "request_id": params.request_id,
@@ -186,6 +194,15 @@ impl RouterClient {
         // then fall through to the outcome handling below.
         let mut response: Option<Result<Value, iii_sdk::errors::Error>> = None;
         let mut drain_deadline = Instant::now();
+        // Resolves when harness::stop fires the turn's local cancel; a dropped
+        // sender (registry cleared — no stop can arrive) parks forever instead
+        // of busy-looping the select.
+        let abort_fired = async move {
+            if abort_rx.wait_for(|a| *a).await.is_err() {
+                std::future::pending::<()>().await;
+            }
+        };
+        tokio::pin!(abort_fired);
         loop {
             let frame = if response.is_some() {
                 match tokio::time::timeout(Duration::from_millis(750), rx.recv()).await {
@@ -208,6 +225,52 @@ impl RouterClient {
             } else {
                 tokio::select! {
                     f = rx.recv() => f,
+                    // Local cancel backstop: cut the await now, return the
+                    // partial as Aborted. No arm in the post-ack drain branch
+                    // above — that path is bounded (≤10s) already.
+                    _ = &mut abort_fired => {
+                        // notify_one, not notify_waiters: it stores a permit
+                        // when the pump task hasn't polled its notified() yet
+                        // (a pre-fired cancel can get here before the spawned
+                        // pump first runs), so the wakeup can't be lost and
+                        // pump.await can't hang on next_binary().
+                        cancel.notify_one();
+                        let _ = pump.await;
+                        // Kill the held-open router::chat task; the router's
+                        // own teardown was already requested via router::abort.
+                        trigger.abort();
+                        // Re-request the router abort: stop.rs's router::abort
+                        // can race ahead of the router's stream registration
+                        // and no-op — by now the stream is certainly
+                        // registered, so this lands and the provider stops
+                        // generating instead of streaming into the closed
+                        // channel until its own terminal. Detached: an abort
+                        // against a hung router must not block the stop path.
+                        let client = self.clone();
+                        let rid = request_id.clone();
+                        tokio::spawn(async move {
+                            client.abort(&rid).await;
+                        });
+                        // Prefer a terminal frame's complete message over the
+                        // tracker's partial: a stop landing between a Done
+                        // frame and EOF must not replace the full assistant
+                        // entry with the stale pre-Done partial.
+                        let mut message = final_message
+                            .or_else(|| tracker.current())
+                            .unwrap_or_else(|| {
+                                empty_assistant(
+                                    params.provider.as_deref().unwrap_or(""),
+                                    &params.model,
+                                )
+                            });
+                        message.stop_reason = StopReason::Aborted;
+                        return Ok(ChatOutcome {
+                            ok: false,
+                            stop_reason: Some(StopReason::Aborted),
+                            message,
+                            error: None,
+                        });
+                    }
                     r = &mut trigger => {
                         response = Some(r.map_err(|e| {
                             HarnessError::Internal(format!("router::chat task: {e}"))
@@ -273,7 +336,8 @@ impl RouterClient {
         }
 
         // Stream drained; stop the pump and collect the trigger ack.
-        cancel.notify_waiters();
+        // notify_one: latched — safe even if the pump task hasn't polled yet.
+        cancel.notify_one();
         let _ = pump.await;
         let response = match response {
             Some(r) => r,

harness/src/deps.rs

@@ -12,7 +12,7 @@ use crate::configuration::ConfigCell;
 use crate::discovery::{FunctionsCell, FunctionsSnapshot};
 use crate::events::TurnEvents;
 use crate::hooks::HookRegistry;
-use crate::locks::SessionLocks;
+use crate::locks::{SessionLocks, TurnCancels};
 use crate::subscriptions::SubscriptionRegistry;
 
 #[derive(Clone)]
@@ -23,6 +23,7 @@ pub struct Deps {
     pub events: TurnEvents,
     pub hooks: HookRegistry,
     pub locks: SessionLocks,
+    pub cancels: TurnCancels,
     pub subscriptions: Arc<SubscriptionRegistry>,
     /// react's per-subscription fire-rate breaker (loop breaker #3).
     pub react_gate: Arc<crate::functions::react::FireGate>,
@@ -43,6 +44,7 @@ impl Deps {
             events,
             hooks,
             locks: SessionLocks::new(),
+            cancels: TurnCancels::new(),
             subscriptions: Arc::new(SubscriptionRegistry::new()),
             react_gate: Arc::new(crate::functions::react::FireGate::default()),
         }

harness/src/functions/stop.rs

@@ -44,10 +44,21 @@ pub async fn handle(deps: &Deps, req: StopRequest) -> Result<StopResponse, Harne
     if record.status.is_terminal() {
         return Ok(StopResponse { stopping: false });
     }
+    // Idempotent: a prior stop already fired the cancel signal, cascaded to
+    // children, and requested the router abort — repeat clicks become one read.
+    if record.abort {
+        return Ok(StopResponse { stopping: true });
+    }
     // Pin the turn we observed so the write under the lock can't land on a newer
     // turn that started in between (matters when `turn_id` was omitted).
     let target_turn = record.turn_id.clone();
 
+    // In-process cancel signal, fired lock-free BEFORE anything that awaits:
+    // it cuts the in-flight `router.chat` await (backstop when router::abort
+    // is a no-op) and is observed between tool executions, where the durable
+    // flag write below is blocked on the session lock.
+    deps.cancels.fire(&target_turn);
+
     // Cascade to non-terminal spawned children BEFORE taking this session's
     // lock (harness.md § Cancellation cascade): each child stop acquires its
     // OWN session lock, so holding the parent lock here could deadlock against a
@@ -100,5 +111,15 @@ pub async fn handle(deps: &Deps, req: StopRequest) -> Result<StopResponse, Harne
     record.updated_at = crate::types::message::AgentMessage::now_ms();
     crate::state::put_turn(&deps.iii, &record, cfg.session_timeout_ms).await?;
 
+    // "stopping" ack on the existing phase-reason channel (status stays
+    // "working" — same semantics as "waiting for <model>"). UNDER the lock and
+    // after the terminal re-check: every finalizer emits its own set_status
+    // while holding this lock, so a lock-free ack here could land AFTER a
+    // concurrent finalize's "done" and leave the session stuck on "working".
+    let session = deps.session().await;
+    let _ = session
+        .set_status(&req.session_id, "working", Some("stopping"))
+        .await;
+
     Ok(StopResponse { stopping: true })
 }

harness/src/locks.rs

@@ -35,9 +35,55 @@ impl SessionLocks {
     }
 }
 
+/// Per-turn in-process cancel signals. `harness::stop` fires one lock-free so
+/// the running step can observe the stop where durable state can't reach it:
+/// inside the `router::chat` await (via `watch`) and between tool executions
+/// (via `is_fired`) — the tool phase holds the session lock, so the durable
+/// abort write is blocked behind it by construction. Level-triggered `watch`
+/// channels: a fire before subscribe is still observed. Keyed by turn_id so a
+/// stale fire can never cancel a newer turn. Same single-process caveat as
+/// `SessionLocks` above.
+#[derive(Clone, Default)]
+pub struct TurnCancels {
+    map: Arc<Mutex<HashMap<String, tokio::sync::watch::Sender<bool>>>>,
+}
+
+impl TurnCancels {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Signal cancellation for `turn_id` (idempotent).
+    pub fn fire(&self, turn_id: &str) {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.entry(turn_id.to_string())
+            .or_insert_with(|| tokio::sync::watch::channel(false).0)
+            .send_replace(true);
+    }
+
+    pub fn is_fired(&self, turn_id: &str) -> bool {
+        let map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.get(turn_id).is_some_and(|s| *s.borrow())
+    }
+
+    /// Subscribe to `turn_id`'s cancel signal, creating it on first use.
+    pub fn watch(&self, turn_id: &str) -> tokio::sync::watch::Receiver<bool> {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.entry(turn_id.to_string())
+            .or_insert_with(|| tokio::sync::watch::channel(false).0)
+            .subscribe()
+    }
+
+    /// Drop `turn_id`'s signal once the turn is terminal.
+    pub fn clear(&self, turn_id: &str) {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.remove(turn_id);
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use super::SessionLocks;
+    use super::{SessionLocks, TurnCancels};
     use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
     use std::sync::Arc;
 
@@ -84,4 +130,18 @@ mod tests {
         let _other = locks.guard("b").await;
         drop(held);
     }
+
+    /// The property the chat-abort backstop relies on: level-triggering. A fire
+    /// BEFORE anyone subscribes is still observed by a later watch/is_fired.
+    #[tokio::test]
+    async fn cancel_fired_before_subscribe_is_observed() {
+        let cancels = TurnCancels::new();
+        cancels.fire("t1");
+        assert!(cancels.is_fired("t1"));
+        let rx = cancels.watch("t1");
+        assert!(*rx.borrow());
+        assert!(!cancels.is_fired("t2"));
+        cancels.clear("t1");
+        assert!(!cancels.is_fired("t1"));
+    }
 }