fix(workspace): make Stop actually cancel the turn, not just kill the process

packages/agent-connector/src/adapters/base.js

@@ -91,6 +91,28 @@ class BaseAdapter {
     // Per-channel task tracking for parallel execution
     this._channelBusy = new Set();
     this._channelQueues = {};
+    // ── Stop watermarks ──
+    // A stop control event carries the SERVER timestamp of the moment the user
+    // hit Stop. Any message posted at or before that instant belongs to the turn
+    // the user just cancelled, so it must not start a fresh run — even if it
+    // reaches us afterwards, which is the common case: the message is posted,
+    // the user immediately hits Stop, and only then do we poll the message.
+    // Comparing server-to-server timestamps keeps this free of clock skew.
+    //
+    // Watermarks are NEVER cleared when a newer message passes. A newer message
+    // already fails the `<=` test on its own, and dropping the mark would let a
+    // message that was queued BEFORE the stop through when the worker drains
+    // that queue later.
+    this._stopWatermarks = new Map();  // channel → server ms
+    this._stopWatermarkAll = 0;        // stop with no channel (whole agent)
+    // Newest timestamp among messages we have accepted for processing. The poll
+    // cursor never goes backwards, so this is the floor below which the server
+    // can no longer deliver anything — see _pruneStopWatermarks.
+    this._lastDispatchedMessageTs = 0;
+    // channel → the watermark we already announced a stop for, so a burst of
+    // dropped messages posts one notice while a genuinely new stop (larger
+    // watermark) is still allowed to announce itself.
+    this._stopNoticedAt = new Map();
     // Cached workspace.browser_enabled. Populated lazily on first read so we
     // don't pay an HTTP roundtrip per message — adapters that toggle the
     // workspace flag must reconnect/restart to pick up the change (matches
@@ -339,6 +361,10 @@ class BaseAdapter {
         if (ev.id) this._lastControlId = ev.id;
         const payload = ev.payload || {};
         const action = payload.action;
+        // Record the watermark BEFORE the adapter-specific handler runs, so a
+        // message that lands while that handler is still killing processes is
+        // already covered.
+        if (action === 'stop') this._markStopWatermark(payload, ev.timestamp);
         if (action === 'set_mode') {
           const newMode = payload.mode || 'execute';
           if ((newMode === 'execute' || newMode === 'plan') && newMode !== this._mode) {
@@ -353,6 +379,117 @@ class BaseAdapter {
     } catch {}
   }
 
+  /**
+   * Remember when a stop was issued, using the control event's own server
+   * timestamp. An unusable timestamp is skipped rather than substituted with
+   * `Date.now()` — mixing this machine's clock into a server timeline would
+   * drop or admit the wrong messages under clock skew.
+   */
+  _markStopWatermark(payload, timestamp) {
+    const ts = Number(timestamp);
+    if (!Number.isFinite(ts) || ts <= 0) {
+      this._log('Stop control event carried no usable timestamp — no watermark set');
+      return;
+    }
+    const channel = payload && (payload.channel || payload.sessionId);
+    if (!channel) {
+      this._stopWatermarkAll = Math.max(this._stopWatermarkAll, ts);
+      return;
+    }
+    this._stopWatermarks.set(String(channel), ts);
+  }
+
+  /**
+   * Forget watermarks that can no longer match anything, so a long-lived agent
+   * does not accumulate one entry per channel it was ever stopped in.
+   *
+   * A watermark is only dropped when it is provably spent. The poll cursor is
+   * monotonic in (timestamp, id) and shared by every channel, so once a message
+   * timestamped T has been dispatched the server can never hand us one older
+   * than T again — for any channel. What the server can no longer deliver, only
+   * our own queue can, so a watermark below T is spent once nothing at or below
+   * it is still queued.
+   *
+   * Evicting by count instead would be a correctness bug, not a safety net: the
+   * mark it discarded might be the one still holding back a queued message.
+   */
+  _pruneStopWatermarks() {
+    if (this._stopWatermarks.size === 0) return;
+    const delivered = this._lastDispatchedMessageTs;
+    if (!delivered) return;
+    for (const [channel, watermark] of this._stopWatermarks) {
+      if (watermark >= delivered) continue;
+      // A message already inside _handleMessage has left the queue but has not
+      // finished: it re-checks the watermark before it starts work, so the mark
+      // has to outlive the turn. Another channel's newer message would
+      // otherwise prune it out from under that check. The worker prunes again
+      // once it is no longer busy.
+      if (this._channelBusy.has(channel)) continue;
+      const queue = this._channelQueues[channel] || [];
+      const stillHeld = queue.some((m) => {
+        const ts = m && m.createdAt ? Date.parse(m.createdAt) : NaN;
+        return !Number.isFinite(ts) || ts <= watermark;
+      });
+      if (stillHeld) continue;
+      this._stopWatermarks.delete(channel);
+      this._stopNoticedAt.delete(channel);
+    }
+  }
+
+  _stopWatermarkFor(channel) {
+    return Math.max(this._stopWatermarks.get(channel) || 0, this._stopWatermarkAll);
+  }
+
+  /**
+   * True when *msg* was posted at or before the last stop for its channel, i.e.
+   * the user cancelled the turn this message belongs to. A message with no
+   * usable timestamp is never dropped — silently swallowing what someone typed
+   * is worse than running one extra turn.
+   */
+  _isStoppedOut(channel, msg) {
+    const watermark = this._stopWatermarkFor(channel);
+    if (!watermark) return false;
+    const ts = msg && msg.createdAt ? Date.parse(msg.createdAt) : NaN;
+    if (!Number.isFinite(ts)) return false;
+    return ts <= watermark;
+  }
+
+  /**
+   * Re-check, part way through handling *msg*, whether the user stopped this
+   * turn in the meantime.
+   *
+   * The dispatch-time check cannot cover the whole turn. Between it and the
+   * moment the CLI is actually spawned an adapter awaits several round trips
+   * (auto-title, a status ping), and a stop landing in that window finds no
+   * process to kill — so without this the run starts anyway and its answer is
+   * posted after "Execution stopped by user.". Call it immediately before
+   * starting the work, and again before posting a result.
+   */
+  _turnWasStopped(channel, msg) {
+    return this._isStoppedOut(channel, msg);
+  }
+
+  /**
+   * Announce "stopped" in *channel*, at most once per stop.
+   *
+   * The workspace UI keeps its Stop button in a disabled "Stopping…" state
+   * until a non-status message lands in the thread, so a stop that posts
+   * nothing locks that button. Every stop path must end here — including the
+   * ones where there was no process to kill.
+   *
+   * Dedup is keyed on the current watermark rather than a flag someone has to
+   * remember to reset: a later stop raises the watermark and is free to
+   * announce itself, while several messages dropped by the same stop share one
+   * notice.
+   */
+  async _postStopNotice(channel, message = 'Execution stopped by user.') {
+    if (!channel) return;
+    const watermark = this._stopWatermarkFor(channel);
+    if (this._stopNoticedAt.get(channel) === watermark) return;
+    this._stopNoticedAt.set(channel, watermark);
+    try { await this.sendResponse(channel, message); } catch {}
+  }
+
   /**
    * Handle adapter-specific control actions. Override in subclasses to add
    * per-adapter actions (`stop`, `restart`, …); always call
@@ -760,6 +897,21 @@ class BaseAdapter {
       channel = msg.sessionId;
     }
 
+    // The user hit Stop after posting this. Killing the subprocess alone would
+    // not have helped — this message had not started running yet, so without
+    // this check it spawns a fresh run one poll later and Stop looks broken.
+    if (this._isStoppedOut(channel, msg)) {
+      this._log(`Dropping message posted before Stop in ${channel}`);
+      await this._postStopNotice(channel);
+      return;
+    }
+
+    const dispatchedTs = msg && msg.createdAt ? Date.parse(msg.createdAt) : NaN;
+    if (Number.isFinite(dispatchedTs)) {
+      this._lastDispatchedMessageTs = Math.max(this._lastDispatchedMessageTs, dispatchedTs);
+      this._pruneStopWatermarks();
+    }
+
     if (this._channelBusy.has(channel)) {
       // A routine that's already running must not stack up. Routine fires are
       // periodic, so a fire that arrives while the previous run is still going
@@ -813,6 +965,13 @@ class BaseAdapter {
       const queue = this._channelQueues[channel];
       if (!queue || queue.length === 0) break;
       const nextMsg = queue.shift();
+      // Queued before the stop that just landed — drop it rather than running
+      // the work the user cancelled.
+      if (this._isStoppedOut(channel, nextMsg)) {
+        this._log(`Dropping queued message posted before Stop in ${channel}`);
+        await this._postStopNotice(channel);
+        continue;
+      }
       if (nextMsg._queueId) {
         try { await this.sendStatus(channel, 'processing queued message', { queue_id: nextMsg._queueId, queue_status: 'processed' }); } catch {}
       }
@@ -826,6 +985,8 @@ class BaseAdapter {
       }
     }
     this._channelBusy.delete(channel);
+    // Safe to reconsider this channel's watermark now that nothing is in flight.
+    this._pruneStopWatermarks();
   }
 
   // ------------------------------------------------------------------

packages/agent-connector/src/adapters/claude.js

@@ -92,12 +92,6 @@ class ClaudeAdapter extends BaseAdapter {
     this._channelSessions = {}; // channel → Claude CLI session_id
     this._channelProcesses = {}; // channel → child process
     this._stoppingChannels = new Set();
-    // Channels that have already announced "Execution stopped by user." for the
-    // current stop. Two paths race to post it (the control-action handler that
-    // kills the process, and the in-flight message handler that sees
-    // pp.userStopped after exit), so this dedups to a single notice. Reset when
-    // a new message starts processing in the channel.
-    this._stopNoticeSent = new Set();
     this._persistentProcs = {}; // channel → { proc, lineBuffer, pendingLines, idleTimer, messageResolve }
     // Knowledge pinning (decision log + glossary) lives in BaseAdapter; this
     // adapter fetches directly in _handleMessage because the result also
@@ -138,16 +132,22 @@ class ClaudeAdapter extends BaseAdapter {
     if (action === 'stop') {
       const channel = (payload && typeof payload === 'object') ? payload.channel : null;
       if (channel) {
+        // Scoped to this channel and nothing else. The previous shape keyed the
+        // per-channel branch on `_channelProcesses[channel]` being set, so a
+        // stop naming a channel that happened to be idle fell through to the
+        // stop-everything branch and killed unrelated threads.
         const pp = this._persistentProcs[channel];
         if (pp) pp.userStopped = true;
-      }
-      if (channel && this._channelProcesses[channel]) {
-        this._log(`Stopping process for channel=${channel}`);
         this._stoppingChannels.add(channel);
-        const proc = this._channelProcesses[channel];
-        await this._stopProcess(proc);
-        delete this._channelProcesses[channel];
         delete this._channelQueues[channel];
+        const proc = this._channelProcesses[channel];
+        if (proc) {
+          this._log(`Stopping process for channel=${channel}`);
+          await this._stopProcess(proc);
+          delete this._channelProcesses[channel];
+        }
+        // Announced whether or not anything was running: an unacknowledged stop
+        // leaves the UI's Stop button disabled at "Stopping…" forever.
         await this._postStopNotice(channel);
       } else {
         for (const pp of Object.values(this._persistentProcs)) pp.userStopped = true;
@@ -312,24 +312,17 @@ class ClaudeAdapter extends BaseAdapter {
     );
   }
 
-  /**
-   * Post "Execution stopped by user." at most once per channel for a given
-   * stop. The control-action handler and the in-flight message handler both
-   * race to announce a stop; without this guard the user sees it twice. The
-   * guard is reset when a new message starts processing in the channel.
-   */
-  async _postStopNotice(channel) {
-    if (!channel || this._stopNoticeSent.has(channel)) return;
-    this._stopNoticeSent.add(channel);
-    try { await this.sendResponse(channel, 'Execution stopped by user.'); } catch {}
-  }
-
   async _stopAllProcesses(completionMessage = 'Execution stopped.') {
     for (const channel of Object.keys(this._persistentProcs)) {
       this._killPersistentProc(channel);
     }
     const entries = Object.entries(this._channelProcesses);
-    if (!entries.length) return;
+    if (!entries.length) {
+      // Nothing was running, but the user still asked for a stop and the UI is
+      // waiting on an acknowledgement.
+      await this._postStopNotice(this.channelName, completionMessage);
+      return;
+    }
     this._log(`Stopping ${entries.length} running process(es)...`);
     for (const [channel, proc] of entries) {
       this._stoppingChannels.add(channel);
@@ -1105,7 +1098,6 @@ class ClaudeAdapter extends BaseAdapter {
 
     const msgChannel = msg.sessionId || this.channelName;
     this._stoppingChannels.delete(msgChannel);
-    this._stopNoticeSent.delete(msgChannel);
     const sender = msg.senderName || msg.senderType || 'user';
     this._log(`Processing message from ${sender} in ${msgChannel}: ${content.slice(0, 80)}...`);
 
@@ -1156,6 +1148,16 @@ class ClaudeAdapter extends BaseAdapter {
       ? { enabled: true, entryId: glossary.entryId, content: glossary.content, scope: glossary.scope || 'channel' }
       : null;
 
+    // The stop may have arrived during the auto-title, status and knowledge
+    // round trips above, at a moment when there was no process for it to kill.
+    // Handing the message to a process now would run exactly what the user
+    // cancelled.
+    if (this._turnWasStopped(msgChannel, msg)) {
+      this._log(`Not starting a run in ${msgChannel} — the user stopped it first`);
+      await this._postStopNotice(msgChannel);
+      return;
+    }
+
     // ── Persistent process fast-path ──
     // If we have a living persistent process for this channel, send via stdin
     // instead of spawning a new CLI (saves ~2s startup time).

packages/agent-connector/src/adapters/cline.js

@@ -134,12 +134,17 @@ class ClineAdapter extends BaseAdapter {
   async _onControlAction(action, payload) {
     if (action === 'stop') {
       const channel = (payload && typeof payload === 'object') ? payload.channel : null;
-      if (channel && this._channelProcesses[channel]) {
+      if (channel) {
+      // Scoped to the named channel whether or not anything is running there.
+      // Keying the per-channel branch on a live process meant a stop naming an
+      // idle channel fell through and killed every other channel's work.
         this._stoppingChannels.add(channel);
-        await this._stopProcess(this._channelProcesses[channel]);
-        delete this._channelProcesses[channel];
         delete this._channelQueues[channel];
-        try { await this.sendResponse(channel, 'Execution stopped by user.'); } catch {}
+        if (this._channelProcesses[channel]) {
+          await this._stopProcess(this._channelProcesses[channel]);
+          delete this._channelProcesses[channel];
+        }
+        await this._postStopNotice(channel);
       } else {
         await this._stopAllProcesses('Execution stopped by user.');
       }
@@ -528,6 +533,14 @@ class ClineAdapter extends BaseAdapter {
 
     // One retry: if resuming a stale session fails, retry once fresh.
     for (let attempt = 0; attempt < 2; attempt++) {
+      // Re-checked every attempt. The stop can land during the awaits above,
+      // when there is no process to kill, and again between the first spawn and
+      // the stale-session retry — either way the work must not start.
+      if (this._turnWasStopped(channel, msg)) {
+        this._log(`Not starting a run in ${channel} — the user stopped it first`);
+        await this._postStopNotice(channel);
+        return;
+      }
       const resumeId = attempt === 0 ? this._resumableSession(channel, workingDir) : null;
 
       // Build the prompt. Resuming → Cline already has history, send the bare

packages/agent-connector/src/adapters/copilot.js

@@ -338,11 +338,20 @@ class CopilotAdapter extends BaseAdapter {
   async _onControlAction(action, payload) {
     if (action === 'stop') {
       const channel = (payload && typeof payload === 'object') ? payload.channel : null;
-      if (channel && this._channelProcesses[channel]) {
+      if (channel) {
+      // Scoped to the named channel whether or not anything is running there.
+      // Keying the per-channel branch on a live process meant a stop naming an
+      // idle channel fell through and killed every other channel's work.
         this._stoppingChannels.add(channel);
-        await this._stopProcess(this._channelProcesses[channel]);
-        delete this._channelProcesses[channel];
         delete this._channelQueues[channel];
+        if (this._channelProcesses[channel]) {
+          await this._stopProcess(this._channelProcesses[channel]);
+          delete this._channelProcesses[channel];
+        }
+        // Announced as a status, not a response — this adapter deliberately
+        // posts nothing of type 'response' after a user stop. The wording still
+        // carries "stopped", which is what the workspace UI matches on to
+        // release its Stop button.
         try { await this.sendStatus(channel, 'Execution stopped by user'); } catch {}
       } else {
         for (const [ch, proc] of Object.entries(this._channelProcesses)) {
@@ -449,6 +458,14 @@ class CopilotAdapter extends BaseAdapter {
 
     // Up to 2 attempts: resume first, then a fresh session if resume was stale.
     for (let attempt = 0; attempt < 2; attempt++) {
+      // Re-checked every attempt. The stop can land during the awaits above,
+      // when there is no process to kill, and again between the first spawn and
+      // the stale-session retry — either way the work must not start.
+      if (this._turnWasStopped(channel, msg)) {
+        this._log(`Not starting a run in ${channel} — the user stopped it first`);
+        await this._postStopNotice(channel);
+        return;
+      }
       const skipResume = attempt > 0;
       const args = this._buildArgs(fullPrompt, channel, { skipResume });
       this._logSpawn(channel, args, skipResume);