From c30ac4213b417d6dd945ce2dae3e36daa4024526 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Tue, 1 Sep 2026 06:34:25 +0000 Subject: [PATCH 1/3] fix the workspace Stop button so it actually cancels the turn Stop killed the running subprocess and nothing else, so a message posted just before the click started a fresh run one poll later and the button looked inert. Adapters now record the stop control event's own server timestamp as a per-channel watermark and drop any message posted at or before it, both when a message is dispatched and when the channel worker drains its queue. The watermark is never cleared by a newer message passing, because a message queued before the stop is only seen at drain time, after the newer one has already gone through. A stop that had nothing to kill used to post nothing at all, which left the UI's button disabled at "Stopping..." forever since it only clears on a non-status message and the background poll skips the open thread. Every stop path now acknowledges, deduplicated on the current watermark so a burst of dropped messages posts one notice while a genuinely newer stop can still announce itself. Claude's per-channel branch was keyed on that channel having a live process, so a stop naming an idle channel fell through to the stop-everything branch and killed unrelated threads. It is now scoped to the named channel whether or not anything is running there. The frontend falls back to broadcasting a stop when a thread's participant list is not available yet, which was previously a silent no-op, and is only safe now that a channel-named stop stays in its channel. Each Stop click takes a generation and its timers act only while they still own the session, so a first click's give-up timer can no longer clear a second click's state. Sessions whose stop is never acknowledged go back to active rather than merely losing the latch, since the button renders on active-or-stopping and clearing alone just hid it. OpenClaw never tracked its CLI child, so Stop was completely inert there. It now registers the process per channel, releases it by identity so a late exit cannot unregister its successor, spawns into its own process group so tool subprocesses go down with it, tears children down on daemon shutdown, and suppresses both the error and the empty-response reply that a user stop would otherwise produce. Both copies of the workspace context are updated; packages/go/web carried a byte-identical stopAllAgents. --- packages/agent-connector/src/adapters/base.js | 108 ++++++++ .../agent-connector/src/adapters/claude.js | 44 ++-- .../agent-connector/src/adapters/openclaw.js | 99 +++++++- packages/agent-connector/src/adapters/pi.js | 24 +- .../agent-connector/test/stop-control.test.js | 230 ++++++++++++++++++ packages/go/web/lib/stop-requests.test.ts | 47 ++++ packages/go/web/lib/stop-requests.ts | 33 +++ packages/go/web/lib/workspace-context.tsx | 63 ++++- workspace/frontend/lib/stop-requests.test.ts | 47 ++++ workspace/frontend/lib/stop-requests.ts | 33 +++ workspace/frontend/lib/workspace-context.tsx | 63 ++++- 11 files changed, 737 insertions(+), 54 deletions(-) create mode 100644 packages/go/web/lib/stop-requests.test.ts create mode 100644 packages/go/web/lib/stop-requests.ts create mode 100644 workspace/frontend/lib/stop-requests.test.ts create mode 100644 workspace/frontend/lib/stop-requests.ts diff --git a/packages/agent-connector/src/adapters/base.js b/packages/agent-connector/src/adapters/base.js index 4276758bc..cf7f7c097 100644 --- a/packages/agent-connector/src/adapters/base.js +++ b/packages/agent-connector/src/adapters/base.js @@ -91,6 +91,24 @@ 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) + // 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 +357,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 +375,76 @@ class BaseAdapter { } catch {} } + /** Cap on remembered watermarks — a long-lived agent must not grow one entry + * per channel forever. Map preserves insertion order, so the oldest mark is + * the first key. */ + static get STOP_WATERMARK_LIMIT() { return 200; } + + /** + * 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; + } + const key = String(channel); + // Re-insert so the key moves to the back of the eviction order. + this._stopWatermarks.delete(key); + this._stopWatermarks.set(key, ts); + while (this._stopWatermarks.size > BaseAdapter.STOP_WATERMARK_LIMIT) { + this._stopWatermarks.delete(this._stopWatermarks.keys().next().value); + } + } + + _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; + } + + /** + * 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 +852,15 @@ 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; + } + 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 +914,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 {} } diff --git a/packages/agent-connector/src/adapters/claude.js b/packages/agent-connector/src/adapters/claude.js index 52292bdaa..7d47bcd92 100644 --- a/packages/agent-connector/src/adapters/claude.js +++ b/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)}...`); diff --git a/packages/agent-connector/src/adapters/openclaw.js b/packages/agent-connector/src/adapters/openclaw.js index 71c2d6685..b3f64e066 100644 --- a/packages/agent-connector/src/adapters/openclaw.js +++ b/packages/agent-connector/src/adapters/openclaw.js @@ -38,6 +38,14 @@ class OpenClawAdapter extends BaseAdapter { this.openclawAgentId = opts.openclawAgentId || 'main'; this.disabledModules = opts.disabledModules || new Set(); + // channel → in-flight CLI child. Without this a stop had nothing to act on: + // the adapter never tracked its subprocess, so the Stop button was inert + // for openclaw agents. + this._channelProcesses = {}; + // Channels whose current run was killed by a stop. Read by _handleMessage + // so the kill does not also surface as an error or a "no response" reply. + this._stoppingChannels = new Set(); + // Find the openclaw binary — always use CLI/gateway mode for full tool support this._openclawBinary = this._findOpenclawBinary(); @@ -240,6 +248,7 @@ class OpenClawAdapter extends BaseAdapter { } const sender = msg.senderName || msg.senderType || 'user'; this._log(`Processing message from ${sender} in ${msgChannel}: ${content.slice(0, 80)}...`); + this._stoppingChannels.delete(msgChannel); await this._autoTitleChannel(msgChannel, content); await this.sendStatus(msgChannel, 'thinking...'); @@ -249,15 +258,90 @@ class OpenClawAdapter extends BaseAdapter { if (responseText) { await this.sendResponse(msgChannel, responseText); - } else { + } else if (!this._stoppingChannels.has(msgChannel)) { await this.sendResponse(msgChannel, 'No response generated. Please try again.'); } } catch (e) { + // A stop kills the CLI, which exits non-zero and rejects here. The user + // already got "Execution stopped by user." — an error on top of it reads + // as though the stop broke something. + if (this._stoppingChannels.has(msgChannel)) { + this._log(`Run in ${msgChannel} ended because the user stopped it`); + return; + } this._log(`Error handling message: ${e.message}`); await this.sendError(msgChannel, `Error processing message: ${e.message}`); } } + // ------------------------------------------------------------------ + // Control actions + // ------------------------------------------------------------------ + + async _onControlAction(action, payload) { + if (action === 'stop') { + const channel = (payload && typeof payload === 'object') ? payload.channel : null; + if (channel) { + this._stoppingChannels.add(channel); + delete this._channelQueues[channel]; + await this._stopProcess(channel); + await this._postStopNotice(channel); + } else { + for (const ch of Object.keys(this._channelProcesses)) { + this._stoppingChannels.add(ch); + delete this._channelQueues[ch]; + await this._stopProcess(ch); + await this._postStopNotice(ch); + } + await this._postStopNotice(this.channelName); + } + return; + } + await super._onControlAction(action, payload); + } + + /** + * Kill the CLI running in *channel*, giving it the process group's SIGTERM + * first so its own tool subprocesses go down with it, then SIGKILL. Mirrors + * ClaudeAdapter._stopProcess; returns once the child is reaped or the grace + * period lapses. + */ + async _stopProcess(channel) { + const proc = this._channelProcesses[channel]; + if (!proc || proc.exitCode !== null) return; + this._log(`Stopping OpenClaw CLI for channel=${channel}`); + try { + if (IS_WINDOWS) { + try { execSync(`taskkill /F /T /PID ${proc.pid}`, { timeout: 5000 }); } catch {} + return; + } + try { process.kill(-proc.pid, 'SIGTERM'); } catch { proc.kill('SIGTERM'); } + await new Promise((resolve) => { + let done = false; + const finish = () => { if (!done) { done = true; resolve(); } }; + const timeout = setTimeout(() => { + try { process.kill(-proc.pid, 'SIGKILL'); } catch { proc.kill('SIGKILL'); } + const reap = setTimeout(finish, 1000); + proc.once('exit', () => { clearTimeout(reap); finish(); }); + }, 1500); + proc.once('exit', () => { clearTimeout(timeout); finish(); }); + }); + } catch {} + } + + /** + * Daemon shutdown. Without this the CLI children outlive the daemon and the + * thread's last event stays a `status`, so the workspace shows it running + * forever. + */ + stop() { + for (const channel of Object.keys(this._channelProcesses)) { + this._stoppingChannels.add(channel); + this._stopProcess(channel).catch(() => {}); + } + super.stop(); + } + // ------------------------------------------------------------------ // CLI mode (openclaw agent --local) // ------------------------------------------------------------------ @@ -380,9 +464,19 @@ class OpenClawAdapter extends BaseAdapter { cwd: this.workingDir || process.env.HOME || '/', timeout: 600000, windowsHide: true, + // Own process group on POSIX so a stop can signal the CLI's own + // children (tool subprocesses) instead of orphaning them. + detached: !IS_WINDOWS, }); if (proc.stdout) proc.stdout.on('data', (d) => { output += d; }); + this._channelProcesses[channel] = proc; + // Identity-checked so a late exit from a superseded process cannot + // unregister the run that replaced it. + const releaseProc = () => { + if (this._channelProcesses[channel] === proc) delete this._channelProcesses[channel]; + }; + // Poll stderr file every 500ms for tool events let stderrOffset = 0; const pollInterval = setInterval(() => { @@ -413,6 +507,7 @@ class OpenClawAdapter extends BaseAdapter { settled = true; clearInterval(pollInterval); closeFd(); + releaseProc(); try { fs.unlinkSync(stderrFile); } catch {} try { proc.kill(); } catch {} reject(new Error('CLI timed out after 600 seconds')); @@ -424,6 +519,7 @@ class OpenClawAdapter extends BaseAdapter { clearInterval(pollInterval); clearTimeout(killTimeout); closeFd(); + releaseProc(); try { fs.unlinkSync(stderrFile); } catch {} reject(err); }); @@ -433,6 +529,7 @@ class OpenClawAdapter extends BaseAdapter { clearInterval(pollInterval); clearTimeout(killTimeout); closeFd(); + releaseProc(); // Read full stderr content (contains JSON output + trace lines) let stderrContent = ''; try { diff --git a/packages/agent-connector/src/adapters/pi.js b/packages/agent-connector/src/adapters/pi.js index cb3e0e4d2..5ad83d05d 100644 --- a/packages/agent-connector/src/adapters/pi.js +++ b/packages/agent-connector/src/adapters/pi.js @@ -143,9 +143,6 @@ class PiAdapter extends BaseAdapter { // channel → child process (BaseAdapter/stop paths read this) this._channelProcesses = {}; this._stoppingChannels = new Set(); - // Dedup for "Execution stopped by user." — the control handler and the - // in-flight message handler both race to announce it (see claude.js). - this._stopNoticeSent = new Set(); this._sessionsFile = path.join( os.homedir(), '.openagents', 'sessions', @@ -525,13 +522,15 @@ class PiAdapter extends BaseAdapter { if (channel) { const pp = this._persistentProcs[channel]; if (pp) pp.userStopped = true; + this._stoppingChannels.add(channel); + delete this._channelQueues[channel]; if (this._channelProcesses[channel]) { this._log(`Stopping Pi for channel=${channel}`); - this._stoppingChannels.add(channel); await this._abortChannel(channel); - delete this._channelQueues[channel]; - await this._postStopNotice(channel); } + // Acknowledged even with nothing running — the UI's Stop button stays + // disabled at "Stopping…" until something non-status lands. + await this._postStopNotice(channel); } else { for (const pp of Object.values(this._persistentProcs)) pp.userStopped = true; await this._stopAllProcesses('Execution stopped by user.'); @@ -580,7 +579,10 @@ class PiAdapter extends BaseAdapter { async _stopAllProcesses(message = 'Execution stopped.') { const channels = Object.keys(this._persistentProcs); - if (!channels.length) return; + if (!channels.length) { + await this._postStopNotice(this.channelName, message); + return; + } this._log(`Stopping ${channels.length} Pi process(es)...`); for (const channel of channels) { this._stoppingChannels.add(channel); @@ -590,13 +592,6 @@ class PiAdapter extends BaseAdapter { } } - /** Post "Execution stopped by user." at most once per stop, per 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 {} - } - /** * Stop ONE channel: ask Pi to `abort` over RPC first (so it can unwind its * tool work), give the turn a short grace period to settle, then terminate @@ -1342,7 +1337,6 @@ class PiAdapter extends BaseAdapter { const channel = msg.sessionId || this.channelName || 'general'; this._stoppingChannels.delete(channel); - this._stopNoticeSent.delete(channel); if (!content && !attachments.length) return; diff --git a/packages/agent-connector/test/stop-control.test.js b/packages/agent-connector/test/stop-control.test.js index 71926824f..6d0069ae8 100644 --- a/packages/agent-connector/test/stop-control.test.js +++ b/packages/agent-connector/test/stop-control.test.js @@ -8,6 +8,8 @@ const { spawn } = require('node:child_process'); const BaseAdapter = require('../src/adapters/base'); const ClaudeAdapter = require('../src/adapters/claude'); const OpenCodeAdapter = require('../src/adapters/opencode'); +const OpenClawAdapter = require('../src/adapters/openclaw'); +const PiAdapter = require('../src/adapters/pi'); function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -466,4 +468,232 @@ describe('agent stop control', () => { await adapter._stopProcess(proc); } }); + + // ── Stop watermarks ──────────────────────────────────────────────── + // The failure these cover: a message is posted, the user hits Stop before the + // adapter has polled it, and the message then starts a fresh run — so Stop + // looks like it did nothing. + + function baseAdapter() { + const adapter = new BaseAdapter({ + workspaceId: 'ws', + channelName: 'thread', + token: 'token', + agentName: 'agent', + }); + adapter.sendResponse = async (channel, content) => { + adapter.__responses = adapter.__responses || []; + adapter.__responses.push({ channel, content }); + }; + adapter.__responses = []; + return adapter; + } + + const at = (ms) => new Date(ms).toISOString(); + + it('drops a message posted before the stop and announces it once', async () => { + const adapter = baseAdapter(); + const handled = []; + adapter._handleMessage = async (m) => handled.push(m); + + adapter._markStopWatermark({ channel: 'thread' }, 1_000); + await adapter._dispatchMessage({ sessionId: 'thread', content: 'a', createdAt: at(900) }); + await adapter._dispatchMessage({ sessionId: 'thread', content: 'b', createdAt: at(950) }); + + assert.deepEqual(handled, []); + assert.deepEqual(adapter.__responses, [ + { channel: 'thread', content: 'Execution stopped by user.' }, + ]); + }); + + it('keeps the watermark when a newer message passes, so queued older work stays dropped', async () => { + // Regression: clearing the watermark on the first message that passed let a + // message queued BEFORE the stop through when the worker drained the queue. + const adapter = baseAdapter(); + const handled = []; + adapter._handleMessage = async (m) => handled.push(m.content); + adapter._prefetchPinnedContext = async () => {}; + + const older = { sessionId: 'thread', content: 'older', createdAt: at(900) }; + const newer = { sessionId: 'thread', content: 'newer', createdAt: at(2_000) }; + const newest = { sessionId: 'thread', content: 'newest', createdAt: at(3_000) }; + + adapter._markStopWatermark({ channel: 'thread' }, 1_000); + + // `older` was queued before the stop; `newer` arrives after it and passes. + adapter._channelQueues.thread = [older, newer]; + await adapter._channelWorker('thread', newest); + + // The watermark survives a passing message, so the drain still drops `older`. + assert.equal(adapter._stopWatermarkFor('thread'), 1_000); + assert.deepEqual(handled, ['newest', 'newer']); + assert.deepEqual(adapter._channelQueues.thread, []); + assert.equal(adapter.__responses.length, 1, 'one notice for the dropped message'); + }); + + it('treats a message from the same millisecond as the stop as cancelled', async () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'thread' }, 1_000); + assert.equal(adapter._isStoppedOut('thread', { createdAt: at(1_000) }), true); + assert.equal(adapter._isStoppedOut('thread', { createdAt: at(1_001) }), false); + }); + + it('never drops a message that carries no timestamp', () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'thread' }, 1_000); + assert.equal(adapter._isStoppedOut('thread', { content: 'x' }), false); + assert.equal(adapter._isStoppedOut('thread', { createdAt: 'not-a-date' }), false); + }); + + it('ignores a control event with an unusable timestamp instead of using the local clock', () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'thread' }, undefined); + adapter._markStopWatermark({ channel: 'thread' }, 0); + adapter._markStopWatermark({ channel: 'thread' }, 'later'); + assert.equal(adapter._stopWatermarkFor('thread'), 0); + }); + + it('applies a channel-less stop to every channel', () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({}, 5_000); + assert.equal(adapter._isStoppedOut('anything', { createdAt: at(4_999) }), true); + assert.equal(adapter._isStoppedOut('anything', { createdAt: at(5_001) }), false); + }); + + it('announces once per stop but again after a newer stop', async () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'thread' }, 1_000); + await adapter._postStopNotice('thread'); + await adapter._postStopNotice('thread'); + assert.equal(adapter.__responses.length, 1); + + adapter._markStopWatermark({ channel: 'thread' }, 2_000); + await adapter._postStopNotice('thread'); + assert.equal(adapter.__responses.length, 2); + }); + + it('evicts the oldest watermark past the cap', () => { + const adapter = baseAdapter(); + const limit = BaseAdapter.STOP_WATERMARK_LIMIT; + for (let i = 0; i < limit + 5; i++) { + adapter._markStopWatermark({ channel: `c${i}` }, 1_000 + i); + } + assert.equal(adapter._stopWatermarks.size, limit); + assert.equal(adapter._stopWatermarkFor('c0'), 0); + assert.equal(adapter._stopWatermarkFor(`c${limit + 4}`), 1_000 + limit + 4); + }); + + // ── Acknowledging a stop that had nothing to kill ────────────────── + // A stop that posts nothing leaves the UI's button disabled at "Stopping…". + + it('Claude acknowledges a stop naming an idle channel without touching other channels', async () => { + const adapter = new ClaudeAdapter({ + workspaceId: 'ws', + channelName: 'thread', + token: 'token', + agentName: 'claude', + }); + const busy = new EventEmitter(); + busy.pid = 99999993; + busy.exitCode = null; + adapter._channelProcesses.busyChannel = busy; + adapter._stopProcess = async () => {}; + const responses = []; + adapter.sendResponse = async (channel, content) => responses.push({ channel, content }); + + await adapter._onControlAction('stop', { channel: 'idleChannel' }); + + assert.ok(adapter._channelProcesses.busyChannel, 'unrelated channel must keep running'); + assert.equal(adapter._stoppingChannels.has('busyChannel'), false); + assert.deepEqual(responses, [{ channel: 'idleChannel', content: 'Execution stopped by user.' }]); + }); + + it('Claude acknowledges a stop when nothing at all is running', async () => { + const adapter = new ClaudeAdapter({ + workspaceId: 'ws', + channelName: 'thread', + token: 'token', + agentName: 'claude', + }); + const responses = []; + adapter.sendResponse = async (channel, content) => responses.push({ channel, content }); + + await adapter._onControlAction('stop', {}); + + assert.deepEqual(responses, [{ channel: 'thread', content: 'Execution stopped by user.' }]); + }); + + it('Pi acknowledges a stop naming an idle channel', async () => { + const adapter = new PiAdapter({ + workspaceId: 'ws', + channelName: 'thread', + token: 'token', + agentName: 'pi', + }); + const responses = []; + adapter.sendResponse = async (channel, content) => responses.push({ channel, content }); + + await adapter._onControlAction('stop', { channel: 'idleChannel' }); + + assert.deepEqual(responses, [{ channel: 'idleChannel', content: 'Execution stopped by user.' }]); + }); + + // ── OpenClaw ─────────────────────────────────────────────────────── + + it('OpenClaw stop kills the tracked process and suppresses the resulting error', async () => { + const adapter = new OpenClawAdapter({ + workspaceId: 'ws', + channelName: 'thread', + token: 'token', + agentName: 'openclaw', + }); + const responses = []; + const errors = []; + adapter.sendResponse = async (channel, content) => responses.push({ channel, content }); + adapter.sendError = async (channel, content) => errors.push({ channel, content }); + adapter.sendStatus = async () => {}; + adapter._autoTitleChannel = async () => {}; + + const proc = new EventEmitter(); + proc.pid = 99999994; + proc.exitCode = null; + adapter._channelProcesses.thread = proc; + adapter._stopProcess = async (channel) => { delete adapter._channelProcesses[channel]; }; + + // The CLI exits non-zero when killed, which _runCliAgent turns into a reject. + adapter._runCliAgent = async () => { throw new Error('CLI exited null: killed'); }; + + const handling = adapter._handleMessage({ sessionId: 'thread', content: 'go' }); + await adapter._onControlAction('stop', { channel: 'thread' }); + await handling; + + assert.equal(adapter._channelProcesses.thread, undefined); + assert.deepEqual(errors, [], 'a user stop must not also report an error'); + assert.deepEqual(responses, [{ channel: 'thread', content: 'Execution stopped by user.' }]); + }); + + it('OpenClaw suppresses the empty-response reply when the user stopped the run', async () => { + const adapter = new OpenClawAdapter({ + workspaceId: 'ws', + channelName: 'thread', + token: 'token', + agentName: 'openclaw', + }); + const responses = []; + adapter.sendResponse = async (channel, content) => responses.push({ channel, content }); + adapter.sendError = async () => {}; + adapter.sendStatus = async () => {}; + adapter._autoTitleChannel = async () => {}; + adapter._stopProcess = async () => {}; + adapter._runCliAgent = async () => ''; + + const handling = adapter._handleMessage({ sessionId: 'thread', content: 'go' }); + await adapter._onControlAction('stop', { channel: 'thread' }); + await handling; + + assert.deepEqual( + responses.filter((r) => r.content.startsWith('No response generated')), + [], + ); + }); }); diff --git a/packages/go/web/lib/stop-requests.test.ts b/packages/go/web/lib/stop-requests.test.ts new file mode 100644 index 000000000..858e2b8fd --- /dev/null +++ b/packages/go/web/lib/stop-requests.test.ts @@ -0,0 +1,47 @@ +import { describe, it, expect } from 'vitest'; +import { StopRequestTracker } from './stop-requests'; + +describe('StopRequestTracker', () => { + it('reports a fresh Stop as owning its sessions', () => { + const tracker = new StopRequestTracker(); + const gen = tracker.claim(['a', 'b']); + expect(tracker.owned(['a', 'b'], gen)).toEqual(['a', 'b']); + }); + + it('hands ownership to the newer Stop so the older one stops acting', () => { + // Regression: the first Stop's give-up timer fires seconds after a second + // Stop began, and used to clear the second Stop's state. + const tracker = new StopRequestTracker(); + const first = tracker.claim(['a']); + const second = tracker.claim(['a']); + + expect(tracker.owned(['a'], first)).toEqual([]); + expect(tracker.owned(['a'], second)).toEqual(['a']); + }); + + it('leaves sessions the newer Stop did not claim with their original owner', () => { + const tracker = new StopRequestTracker(); + const first = tracker.claim(['a', 'b']); + tracker.claim(['b']); + + expect(tracker.owned(['a', 'b'], first)).toEqual(['a']); + }); + + it('retires a session once the stop is acknowledged', () => { + const tracker = new StopRequestTracker(); + const gen = tracker.claim(['a']); + tracker.release('a'); + expect(tracker.owned(['a'], gen)).toEqual([]); + }); + + it('never reuses a generation, so a released session cannot be reclaimed by an old timer', () => { + const tracker = new StopRequestTracker(); + const first = tracker.claim(['a']); + tracker.release('a'); + const second = tracker.claim(['a']); + + expect(second).not.toBe(first); + expect(tracker.owned(['a'], first)).toEqual([]); + expect(tracker.owned(['a'], second)).toEqual(['a']); + }); +}); diff --git a/packages/go/web/lib/stop-requests.ts b/packages/go/web/lib/stop-requests.ts new file mode 100644 index 000000000..c8b0144ad --- /dev/null +++ b/packages/go/web/lib/stop-requests.ts @@ -0,0 +1,33 @@ +/** + * Tracks which Stop click currently owns each session. + * + * A Stop schedules timers that fire seconds later — one to re-send the control + * event, one to give up waiting for an acknowledgement. By then the user may + * have started new work and pressed Stop again, and the first click's timers + * would clear the second click's state. Generations let a timer tell "still + * mine" from "superseded" before it touches anything. + */ +export class StopRequestTracker { + private generation = 0; + private owners = new Map(); + + /** Claim these sessions for a new Stop. Returns that Stop's generation. */ + claim(sessionIds: string[]): number { + const generation = ++this.generation; + for (const id of sessionIds) this.owners.set(id, generation); + return generation; + } + + /** The subset of `sessionIds` still owned by `generation`. */ + owned(sessionIds: string[], generation: number): string[] { + return sessionIds.filter((id) => this.owners.get(id) === generation); + } + + /** + * Retire a session — the agent acknowledged the stop, or we gave up on it. + * Any timer still holding a generation for it becomes a no-op. + */ + release(sessionId: string): void { + this.owners.delete(sessionId); + } +} diff --git a/packages/go/web/lib/workspace-context.tsx b/packages/go/web/lib/workspace-context.tsx index 46a79be65..7d0ec8b6a 100644 --- a/packages/go/web/lib/workspace-context.tsx +++ b/packages/go/web/lib/workspace-context.tsx @@ -3,6 +3,7 @@ import React, { createContext, useContext, useCallback, useEffect, useRef, useState } from 'react'; import { toast } from 'sonner'; import { workspaceApi } from './api'; +import { StopRequestTracker } from './stop-requests'; import { useOpenAgentsAuth } from './openagents-auth-context'; import { networkAgentToWorkspaceAgent, networkChannelToSession } from './types'; import type { BrowserPersistentContext, BrowserTab, DMConversation, RoutineItem, TodoItem, Workspace, WorkspaceAgent, WorkspaceCollaborator, WorkspaceFile, WorkspaceSession } from './types'; @@ -85,6 +86,14 @@ interface WorkspaceContextValue { setNotificationSound: (enabled: boolean) => void; } +/** + * How long a Stop waits for the agent to acknowledge before the UI stops + * believing it. The agent normally answers within a second or two; past this + * the session is treated as "stop unconfirmed" and the button becomes + * clickable again rather than staying disabled forever. + */ +const STOP_ACK_TIMEOUT_MS = 12_000; + const WorkspaceContext = createContext(null); export function useWorkspace() { @@ -138,6 +147,10 @@ export function WorkspaceProvider({ const [stoppingSessionIds, setStoppingSessionIds] = useState>(new Set()); const stoppingSessionIdsRef = useRef(stoppingSessionIds); stoppingSessionIdsRef.current = stoppingSessionIds; + // Ownership of each session's in-flight Stop. Its retry and give-up timers + // only act while the session still belongs to that Stop, so a first click's + // timers cannot clear the state of a second, unrelated click. + const stopRequestsRef = useRef(new StopRequestTracker()); const [completedSessionIds, setCompletedSessionIds] = useState>(new Set()); const [agentModes, setAgentModes] = useState>({}); const [files, setFiles] = useState([]); @@ -195,6 +208,7 @@ export function WorkspaceProvider({ const updateLastMessage = useCallback((sessionId: string, senderName: string, content: string, isStatus?: boolean) => { if (!isStatus || /stopped|stopping failed/i.test(content)) { + stopRequestsRef.current.release(sessionId); setStoppingSessionIds((prev) => { if (!prev.has(sessionId)) return prev; const next = new Set(prev); @@ -256,11 +270,17 @@ export function WorkspaceProvider({ return next; }); - const targetAgents = targetSessionId - ? agents.filter((a) => { - const session = sessions.find((s) => s.sessionId === targetSessionId); - return session && (session.participants || []).includes(a.agentName); - }) + const generation = stopRequestsRef.current.claim(sessionIds); + + const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; + const participants = session?.participants || []; + // Falls back to every agent when the participant list is unavailable — a + // thread created moments ago, or one discovery hasn't returned yet. Sending + // one stop too many is recoverable; sending none looks like a dead button. + // Safe only because each adapter now scopes a channel-named stop to that + // channel instead of stopping everything it is running. + const targetAgents = targetSessionId && participants.length > 0 + ? agents.filter((a) => participants.includes(a.agentName)) : agents; const sendStop = () => Promise.allSettled( @@ -271,13 +291,42 @@ export function WorkspaceProvider({ ); await sendStop(); + /** Sessions this particular Stop still owns — a later Stop takes them over. */ + const stillOurs = () => stopRequestsRef.current.owned(sessionIds, generation); + window.setTimeout(() => { + const ours = stillOurs(); + if (ours.length === 0) return; setStoppingSessionIds((prevStopping) => { - const stillStopping = sessionIds.filter((sid) => prevStopping.has(sid)); + const stillStopping = ours.filter((sid) => prevStopping.has(sid)); if (stillStopping.length > 0) void sendStop(); return prevStopping; }); }, 3000); + + window.setTimeout(() => { + const ours = stillOurs(); + if (ours.length === 0) return; + ours.forEach((sid) => stopRequestsRef.current.release(sid)); + // Read once, before either setState — the ref tracks rendered state, so + // deciding this inside an updater would depend on commit ordering. + const unconfirmed = ours.filter((sid) => stoppingSessionIdsRef.current.has(sid)); + if (unconfirmed.length === 0) return; + // The stop was never acknowledged. Release the latch AND put the session + // back to active: clearing the latch alone would just hide the button, + // since it renders on active-or-stopping. This returns it to a clickable + // "Stop" so the user can retry. + setStoppingSessionIds((prev) => { + const next = new Set(prev); + unconfirmed.forEach((sid) => next.delete(sid)); + return next; + }); + setActiveSessionIds((prev) => { + const next = new Set(prev); + unconfirmed.forEach((sid) => next.add(sid)); + return next; + }); + }, STOP_ACK_TIMEOUT_MS); }, [activeSessionIds, agents, sessions]); // Configure API client on mount @@ -443,6 +492,7 @@ export function WorkspaceProvider({ if (info.isStatus) { if (isStopping) { if (/stopped|stopping failed/i.test(info.content)) { + stopRequestsRef.current.release(sid); setStoppingSessionIds((s) => { if (!s.has(sid)) return s; const next = new Set(s); @@ -455,6 +505,7 @@ export function WorkspaceProvider({ newActive.add(sid); } } else { + stopRequestsRef.current.release(sid); setStoppingSessionIds((s) => { if (!s.has(sid)) return s; const next = new Set(s); diff --git a/workspace/frontend/lib/stop-requests.test.ts b/workspace/frontend/lib/stop-requests.test.ts new file mode 100644 index 000000000..858e2b8fd --- /dev/null +++ b/workspace/frontend/lib/stop-requests.test.ts @@ -0,0 +1,47 @@ +import { describe, it, expect } from 'vitest'; +import { StopRequestTracker } from './stop-requests'; + +describe('StopRequestTracker', () => { + it('reports a fresh Stop as owning its sessions', () => { + const tracker = new StopRequestTracker(); + const gen = tracker.claim(['a', 'b']); + expect(tracker.owned(['a', 'b'], gen)).toEqual(['a', 'b']); + }); + + it('hands ownership to the newer Stop so the older one stops acting', () => { + // Regression: the first Stop's give-up timer fires seconds after a second + // Stop began, and used to clear the second Stop's state. + const tracker = new StopRequestTracker(); + const first = tracker.claim(['a']); + const second = tracker.claim(['a']); + + expect(tracker.owned(['a'], first)).toEqual([]); + expect(tracker.owned(['a'], second)).toEqual(['a']); + }); + + it('leaves sessions the newer Stop did not claim with their original owner', () => { + const tracker = new StopRequestTracker(); + const first = tracker.claim(['a', 'b']); + tracker.claim(['b']); + + expect(tracker.owned(['a', 'b'], first)).toEqual(['a']); + }); + + it('retires a session once the stop is acknowledged', () => { + const tracker = new StopRequestTracker(); + const gen = tracker.claim(['a']); + tracker.release('a'); + expect(tracker.owned(['a'], gen)).toEqual([]); + }); + + it('never reuses a generation, so a released session cannot be reclaimed by an old timer', () => { + const tracker = new StopRequestTracker(); + const first = tracker.claim(['a']); + tracker.release('a'); + const second = tracker.claim(['a']); + + expect(second).not.toBe(first); + expect(tracker.owned(['a'], first)).toEqual([]); + expect(tracker.owned(['a'], second)).toEqual(['a']); + }); +}); diff --git a/workspace/frontend/lib/stop-requests.ts b/workspace/frontend/lib/stop-requests.ts new file mode 100644 index 000000000..c8b0144ad --- /dev/null +++ b/workspace/frontend/lib/stop-requests.ts @@ -0,0 +1,33 @@ +/** + * Tracks which Stop click currently owns each session. + * + * A Stop schedules timers that fire seconds later — one to re-send the control + * event, one to give up waiting for an acknowledgement. By then the user may + * have started new work and pressed Stop again, and the first click's timers + * would clear the second click's state. Generations let a timer tell "still + * mine" from "superseded" before it touches anything. + */ +export class StopRequestTracker { + private generation = 0; + private owners = new Map(); + + /** Claim these sessions for a new Stop. Returns that Stop's generation. */ + claim(sessionIds: string[]): number { + const generation = ++this.generation; + for (const id of sessionIds) this.owners.set(id, generation); + return generation; + } + + /** The subset of `sessionIds` still owned by `generation`. */ + owned(sessionIds: string[], generation: number): string[] { + return sessionIds.filter((id) => this.owners.get(id) === generation); + } + + /** + * Retire a session — the agent acknowledged the stop, or we gave up on it. + * Any timer still holding a generation for it becomes a no-op. + */ + release(sessionId: string): void { + this.owners.delete(sessionId); + } +} diff --git a/workspace/frontend/lib/workspace-context.tsx b/workspace/frontend/lib/workspace-context.tsx index d32763db1..2f0cd1198 100644 --- a/workspace/frontend/lib/workspace-context.tsx +++ b/workspace/frontend/lib/workspace-context.tsx @@ -2,6 +2,7 @@ import React, { createContext, useContext, useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { workspaceApi } from './api'; +import { StopRequestTracker } from './stop-requests'; import { capture, group } from './analytics'; import { useOpenAgentsAuth } from './openagents-auth-context'; import { generateUserId, getStoredIdentity, storeIdentity } from './identity'; @@ -245,6 +246,14 @@ interface WorkspaceContextValue { setNotificationSound: (enabled: boolean) => void; } +/** + * How long a Stop waits for the agent to acknowledge before the UI stops + * believing it. The agent normally answers within a second or two; past this + * the session is treated as "stop unconfirmed" and the button becomes + * clickable again rather than staying disabled forever. + */ +const STOP_ACK_TIMEOUT_MS = 12_000; + const WorkspaceContext = createContext(null); export function useWorkspace() { @@ -329,6 +338,10 @@ export function WorkspaceProvider({ const [stoppingSessionIds, setStoppingSessionIds] = useState>(new Set()); const stoppingSessionIdsRef = useRef(stoppingSessionIds); stoppingSessionIdsRef.current = stoppingSessionIds; + // Ownership of each session's in-flight Stop. Its retry and give-up timers + // only act while the session still belongs to that Stop, so a first click's + // timers cannot clear the state of a second, unrelated click. + const stopRequestsRef = useRef(new StopRequestTracker()); const [completedSessionIds, setCompletedSessionIds] = useState>(new Set()); const [agentModes, setAgentModes] = useState>({}); const [files, setFiles] = useState([]); @@ -492,6 +505,7 @@ export function WorkspaceProvider({ const updateLastMessage = useCallback((sessionId: string, senderName: string, content: string, isStatus?: boolean) => { if (!isStatus || /stopped|stopping failed/i.test(content)) { + stopRequestsRef.current.release(sessionId); setStoppingSessionIds((prev) => { if (!prev.has(sessionId)) return prev; const next = new Set(prev); @@ -553,11 +567,17 @@ export function WorkspaceProvider({ return next; }); - const targetAgents = targetSessionId - ? agents.filter((a) => { - const session = sessions.find((s) => s.sessionId === targetSessionId); - return session && (session.participants || []).includes(a.agentName); - }) + const generation = stopRequestsRef.current.claim(sessionIds); + + const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; + const participants = session?.participants || []; + // Falls back to every agent when the participant list is unavailable — a + // thread created moments ago, or one discovery hasn't returned yet. Sending + // one stop too many is recoverable; sending none looks like a dead button. + // Safe only because each adapter now scopes a channel-named stop to that + // channel instead of stopping everything it is running. + const targetAgents = targetSessionId && participants.length > 0 + ? agents.filter((a) => participants.includes(a.agentName)) : agents; const sendStop = () => Promise.allSettled( @@ -568,13 +588,42 @@ export function WorkspaceProvider({ ); await sendStop(); + /** Sessions this particular Stop still owns — a later Stop takes them over. */ + const stillOurs = () => stopRequestsRef.current.owned(sessionIds, generation); + window.setTimeout(() => { + const ours = stillOurs(); + if (ours.length === 0) return; setStoppingSessionIds((prevStopping) => { - const stillStopping = sessionIds.filter((sid) => prevStopping.has(sid)); + const stillStopping = ours.filter((sid) => prevStopping.has(sid)); if (stillStopping.length > 0) void sendStop(); return prevStopping; }); }, 3000); + + window.setTimeout(() => { + const ours = stillOurs(); + if (ours.length === 0) return; + ours.forEach((sid) => stopRequestsRef.current.release(sid)); + // Read once, before either setState — the ref tracks rendered state, so + // deciding this inside an updater would depend on commit ordering. + const unconfirmed = ours.filter((sid) => stoppingSessionIdsRef.current.has(sid)); + if (unconfirmed.length === 0) return; + // The stop was never acknowledged. Release the latch AND put the session + // back to active: clearing the latch alone would just hide the button, + // since it renders on active-or-stopping. This returns it to a clickable + // "Stop" so the user can retry. + setStoppingSessionIds((prev) => { + const next = new Set(prev); + unconfirmed.forEach((sid) => next.delete(sid)); + return next; + }); + setActiveSessionIds((prev) => { + const next = new Set(prev); + unconfirmed.forEach((sid) => next.add(sid)); + return next; + }); + }, STOP_ACK_TIMEOUT_MS); }, [activeSessionIds, agents, sessions]); // Configure API client on mount @@ -721,6 +770,7 @@ export function WorkspaceProvider({ if (info.isStatus) { if (isStopping) { if (/stopped|stopping failed/i.test(info.content)) { + stopRequestsRef.current.release(sid); setStoppingSessionIds((s) => { if (!s.has(sid)) return s; const next = new Set(s); @@ -733,6 +783,7 @@ export function WorkspaceProvider({ newActive.add(sid); } } else { + stopRequestsRef.current.release(sid); setStoppingSessionIds((s) => { if (!s.has(sid)) return s; const next = new Set(s); From de9249ca85ee6f9e500abca168cd31aab6f5ffe3 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Tue, 1 Sep 2026 07:55:57 +0000 Subject: [PATCH 2/3] address review findings on the Stop fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four problems, all confirmed against the code before fixing. The broadcast fallback for a missing participant list is removed. It was only safe if every adapter kept a channel-named stop inside that channel, and most do not — aider, amp, gemini, codex, mini, antigravity, llm-direct and hermes ignore the channel entirely, so telling an uninvolved agent to stop would have killed its work in some other thread. When no participant can be resolved the latch is released right away instead, which is what the stuck button needed in the first place. Cursor, Copilot, Cline and Kimi carried the same shape Claude did, where the per-channel branch was gated on that channel having a live process and a stop naming an idle channel fell through to stop-everything. They are now scoped the same way. Copilot keeps announcing via a status rather than a response, which its own test pins deliberately; the wording still carries "stopped", which is what the UI matches on. A stop could still slip through the window between the dispatch-time check and the actual spawn, because handling a message first awaits auto-title and a status ping. A stop landing there found no process to kill, so it only posted a notice and the run started anyway, ending with the full answer posted after "Execution stopped by user." OpenClaw, Claude and Pi now re-check immediately before starting work, and OpenClaw checks again before posting a result. Any non-status message used to confirm a stop, so the background poll could see the user's own just-sent message and clear the latch, cancelling the retry and give-up timers with nothing actually stopped. Only an agent message confirms now. The predicate is extracted so it can be tested. Watermark eviction by count could discard the mark still holding back a queued message, which contradicts the guarantee that a watermark outlives a passing message. Pruning is now driven by what the poll cursor has provably moved past, and only when nothing at or below the mark is still queued, so it stays bounded without being able to forget a live mark. Notice bookkeeping is pruned alongside it rather than growing forever. --- packages/agent-connector/src/adapters/base.js | 67 +++++- .../agent-connector/src/adapters/claude.js | 10 + .../agent-connector/src/adapters/cline.js | 13 +- .../agent-connector/src/adapters/copilot.js | 15 +- .../agent-connector/src/adapters/cursor.js | 13 +- packages/agent-connector/src/adapters/kimi.js | 18 +- .../agent-connector/src/adapters/openclaw.js | 21 +- packages/agent-connector/src/adapters/pi.js | 9 + .../agent-connector/test/stop-control.test.js | 194 +++++++++++++++++- packages/go/web/components/chat/chat-view.tsx | 4 +- packages/go/web/lib/stop-requests.test.ts | 24 ++- packages/go/web/lib/stop-requests.ts | 16 ++ packages/go/web/lib/workspace-context.tsx | 59 ++++-- .../frontend/components/chat/chat-view.tsx | 4 +- workspace/frontend/lib/stop-requests.test.ts | 24 ++- workspace/frontend/lib/stop-requests.ts | 16 ++ workspace/frontend/lib/workspace-context.tsx | 59 ++++-- 17 files changed, 483 insertions(+), 83 deletions(-) diff --git a/packages/agent-connector/src/adapters/base.js b/packages/agent-connector/src/adapters/base.js index cf7f7c097..61768c28f 100644 --- a/packages/agent-connector/src/adapters/base.js +++ b/packages/agent-connector/src/adapters/base.js @@ -105,6 +105,10 @@ class BaseAdapter { // 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. @@ -375,11 +379,6 @@ class BaseAdapter { } catch {} } - /** Cap on remembered watermarks — a long-lived agent must not grow one entry - * per channel forever. Map preserves insertion order, so the oldest mark is - * the first key. */ - static get STOP_WATERMARK_LIMIT() { return 200; } - /** * Remember when a stop was issued, using the control event's own server * timestamp. An unusable timestamp is skipped rather than substituted with @@ -397,12 +396,37 @@ class BaseAdapter { this._stopWatermarkAll = Math.max(this._stopWatermarkAll, ts); return; } - const key = String(channel); - // Re-insert so the key moves to the back of the eviction order. - this._stopWatermarks.delete(key); - this._stopWatermarks.set(key, ts); - while (this._stopWatermarks.size > BaseAdapter.STOP_WATERMARK_LIMIT) { - this._stopWatermarks.delete(this._stopWatermarks.keys().next().value); + 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; + 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); } } @@ -424,6 +448,21 @@ class BaseAdapter { 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. * @@ -861,6 +900,12 @@ class BaseAdapter { 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 diff --git a/packages/agent-connector/src/adapters/claude.js b/packages/agent-connector/src/adapters/claude.js index 7d47bcd92..30887c597 100644 --- a/packages/agent-connector/src/adapters/claude.js +++ b/packages/agent-connector/src/adapters/claude.js @@ -1148,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). diff --git a/packages/agent-connector/src/adapters/cline.js b/packages/agent-connector/src/adapters/cline.js index 325a38ab2..18aadf835 100644 --- a/packages/agent-connector/src/adapters/cline.js +++ b/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.'); } diff --git a/packages/agent-connector/src/adapters/copilot.js b/packages/agent-connector/src/adapters/copilot.js index feae3a9d0..395684826 100644 --- a/packages/agent-connector/src/adapters/copilot.js +++ b/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)) { diff --git a/packages/agent-connector/src/adapters/cursor.js b/packages/agent-connector/src/adapters/cursor.js index be01d595e..d63992769 100644 --- a/packages/agent-connector/src/adapters/cursor.js +++ b/packages/agent-connector/src/adapters/cursor.js @@ -60,12 +60,17 @@ class CursorAdapter 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.'); } catch {} + if (this._channelProcesses[channel]) { + await this._stopProcess(this._channelProcesses[channel]); + delete this._channelProcesses[channel]; + } + await this._postStopNotice(channel, 'Execution stopped.'); } else { await this._stopAllProcesses('Execution stopped.'); } diff --git a/packages/agent-connector/src/adapters/kimi.js b/packages/agent-connector/src/adapters/kimi.js index 0c9496ab1..8e880842b 100644 --- a/packages/agent-connector/src/adapters/kimi.js +++ b/packages/agent-connector/src/adapters/kimi.js @@ -168,12 +168,22 @@ class KimiAdapter extends LlmDirectAdapter { 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); + return; + } + // No CLI for this channel — the direct-API path may still have a turn + // in flight for it, so let base handle it, then acknowledge. + await super._onControlAction(action, payload); + await this._postStopNotice(channel); return; } if (Object.keys(this._channelProcesses).length) { diff --git a/packages/agent-connector/src/adapters/openclaw.js b/packages/agent-connector/src/adapters/openclaw.js index b3f64e066..9ea0544db 100644 --- a/packages/agent-connector/src/adapters/openclaw.js +++ b/packages/agent-connector/src/adapters/openclaw.js @@ -253,20 +253,37 @@ class OpenClawAdapter extends BaseAdapter { await this._autoTitleChannel(msgChannel, content); await this.sendStatus(msgChannel, 'thinking...'); + // The stop may have landed during those two round trips, when there was no + // process for it to kill. Starting the CLI now would run exactly the work + // 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; + } + try { const responseText = await this._runCliAgent(content, msgChannel); + // Checked again: a stop during the run leaves the CLI's partial answer + // worthless, and posting it after "Execution stopped by user." reads as + // though the stop did nothing. + if (this._stoppingChannels.has(msgChannel) || this._turnWasStopped(msgChannel, msg)) { + await this._postStopNotice(msgChannel); + return; + } if (responseText) { await this.sendResponse(msgChannel, responseText); - } else if (!this._stoppingChannels.has(msgChannel)) { + } else { await this.sendResponse(msgChannel, 'No response generated. Please try again.'); } } catch (e) { // A stop kills the CLI, which exits non-zero and rejects here. The user // already got "Execution stopped by user." — an error on top of it reads // as though the stop broke something. - if (this._stoppingChannels.has(msgChannel)) { + if (this._stoppingChannels.has(msgChannel) || this._turnWasStopped(msgChannel, msg)) { this._log(`Run in ${msgChannel} ended because the user stopped it`); + await this._postStopNotice(msgChannel); return; } this._log(`Error handling message: ${e.message}`); diff --git a/packages/agent-connector/src/adapters/pi.js b/packages/agent-connector/src/adapters/pi.js index 5ad83d05d..b5f9d9a04 100644 --- a/packages/agent-connector/src/adapters/pi.js +++ b/packages/agent-connector/src/adapters/pi.js @@ -1389,6 +1389,15 @@ class PiAdapter extends BaseAdapter { browserEnabled, }); + // The stop may have arrived during the status and prompt-building round + // trips above, when there was no process for it to kill. Starting the CLI + // now would run exactly what the user cancelled. + if (this._turnWasStopped(channel, msg)) { + this._log(`Not starting a run in ${channel} — the user stopped it first`); + await this._postStopNotice(channel); + return; + } + let pp; try { pp = await this._ensureProc(channel, workingDir, systemPrompt); diff --git a/packages/agent-connector/test/stop-control.test.js b/packages/agent-connector/test/stop-control.test.js index 6d0069ae8..deba461d7 100644 --- a/packages/agent-connector/test/stop-control.test.js +++ b/packages/agent-connector/test/stop-control.test.js @@ -10,6 +10,10 @@ const ClaudeAdapter = require('../src/adapters/claude'); const OpenCodeAdapter = require('../src/adapters/opencode'); const OpenClawAdapter = require('../src/adapters/openclaw'); const PiAdapter = require('../src/adapters/pi'); +const CursorAdapter = require('../src/adapters/cursor'); +const CopilotAdapter = require('../src/adapters/copilot'); +const ClineAdapter = require('../src/adapters/cline'); +const KimiAdapter = require('../src/adapters/kimi'); function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -572,17 +576,6 @@ describe('agent stop control', () => { assert.equal(adapter.__responses.length, 2); }); - it('evicts the oldest watermark past the cap', () => { - const adapter = baseAdapter(); - const limit = BaseAdapter.STOP_WATERMARK_LIMIT; - for (let i = 0; i < limit + 5; i++) { - adapter._markStopWatermark({ channel: `c${i}` }, 1_000 + i); - } - assert.equal(adapter._stopWatermarks.size, limit); - assert.equal(adapter._stopWatermarkFor('c0'), 0); - assert.equal(adapter._stopWatermarkFor(`c${limit + 4}`), 1_000 + limit + 4); - }); - // ── Acknowledging a stop that had nothing to kill ────────────────── // A stop that posts nothing leaves the UI's button disabled at "Stopping…". @@ -696,4 +689,183 @@ describe('agent stop control', () => { [], ); }); + + // ── A stop arriving before the CLI is spawned ────────────────────── + // The dispatch-time check cannot cover the awaits inside _handleMessage + // (auto-title, status ping). A stop landing there finds nothing to kill. + + it('OpenClaw does not start a run, or post a late answer, when the stop lands pre-spawn', async () => { + const adapter = new OpenClawAdapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'oc', + }); + adapter._log = () => {}; + const posted = []; + adapter.sendResponse = async (c, x) => posted.push(x); + adapter.sendError = async (c, x) => posted.push(`ERR ${x}`); + adapter.sendStatus = async () => {}; + adapter._stopProcess = async () => {}; + + let ran = 0; + adapter._autoTitleChannel = async () => { + adapter._markStopWatermark({ channel: 'thread' }, 2_000); + await adapter._onControlAction('stop', { channel: 'thread' }); + }; + adapter._runCliAgent = async () => { ran++; return 'late result'; }; + + await adapter._handleMessage({ + sessionId: 'thread', content: 'go', createdAt: new Date(1_000).toISOString(), + }); + + assert.equal(ran, 0, 'the CLI must not be spawned after the stop'); + assert.deepEqual(posted, ['Execution stopped by user.']); + }); + + it('OpenClaw suppresses a result produced by a run the user stopped mid-flight', async () => { + const adapter = new OpenClawAdapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'oc', + }); + adapter._log = () => {}; + const posted = []; + adapter.sendResponse = async (c, x) => posted.push(x); + adapter.sendError = async (c, x) => posted.push(`ERR ${x}`); + adapter.sendStatus = async () => {}; + adapter._autoTitleChannel = async () => {}; + adapter._stopProcess = async () => {}; + adapter._runCliAgent = async () => { + adapter._markStopWatermark({ channel: 'thread' }, 2_000); + await adapter._onControlAction('stop', { channel: 'thread' }); + return 'answer the user no longer wants'; + }; + + await adapter._handleMessage({ + sessionId: 'thread', content: 'go', createdAt: new Date(1_000).toISOString(), + }); + + assert.deepEqual(posted, ['Execution stopped by user.']); + }); + + it('Claude does not reach its spawn path when the stop lands pre-spawn', async () => { + const adapter = new ClaudeAdapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'claude', + }); + adapter._log = () => {}; + const posted = []; + adapter.sendResponse = async (c, x) => posted.push(x); + adapter.sendStatus = async () => { + adapter._markStopWatermark({ channel: 'thread' }, 2_000); + await adapter._onControlAction('stop', { channel: 'thread' }); + }; + adapter.client.getSession = async () => ({ title: '', titleManuallySet: true }); + adapter._fetchDecisionLog = async () => ({ available: false }); + adapter._fetchGlossary = async () => ({ available: false }); + let spawned = 0; + adapter._spawnPersistentProc = async () => { spawned++; throw new Error('must not spawn'); }; + + await adapter._handleMessage({ + sessionId: 'thread', content: 'go', createdAt: new Date(1_000).toISOString(), + }); + + assert.equal(spawned, 0); + assert.deepEqual(posted, ['Execution stopped by user.']); + }); + + // ── Channel scoping across adapters ──────────────────────────────── + // A stop naming an idle channel used to fall through to stop-everything. + + for (const [name, Adapter] of [ + ['Cursor', CursorAdapter], ['Copilot', CopilotAdapter], + ['Cline', ClineAdapter], ['Kimi', KimiAdapter], + ]) { + it(`${name} keeps other channels running when the stop names an idle channel`, async () => { + const adapter = new Adapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'a', + }); + adapter._log = () => {}; + adapter.sendResponse = async () => {}; + adapter.sendStatus = async () => {}; + const killed = []; + adapter._stopProcess = async (proc) => killed.push(proc.__name); + + for (const ch of ['busyA', 'busyB']) { + const proc = new EventEmitter(); + proc.pid = 999000; + proc.exitCode = null; + proc.__name = ch; + adapter._channelProcesses[ch] = proc; + } + + await adapter._onControlAction('stop', { channel: 'idleC' }); + + assert.deepEqual(killed, [], `${name} must not kill unrelated channels`); + assert.deepEqual(Object.keys(adapter._channelProcesses).sort(), ['busyA', 'busyB']); + }); + + it(`${name} still kills exactly the named busy channel`, async () => { + const adapter = new Adapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'a', + }); + adapter._log = () => {}; + adapter.sendResponse = async () => {}; + adapter.sendStatus = async () => {}; + const killed = []; + adapter._stopProcess = async (proc) => killed.push(proc.__name); + + for (const ch of ['busyA', 'busyB']) { + const proc = new EventEmitter(); + proc.pid = 999001; + proc.exitCode = null; + proc.__name = ch; + adapter._channelProcesses[ch] = proc; + } + + await adapter._onControlAction('stop', { channel: 'busyA' }); + + assert.deepEqual(killed, ['busyA']); + assert.deepEqual(Object.keys(adapter._channelProcesses), ['busyB']); + }); + } + + // ── Watermark pruning ────────────────────────────────────────────── + + it('keeps a watermark while a message old enough to be blocked is still queued', () => { + // Regression: evicting by count could discard the mark still holding back + // a queued message, which would then run on the next drain. + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'held' }, 1_000); + adapter._channelQueues.held = [{ createdAt: at(900) }]; + adapter._lastDispatchedMessageTs = 5_000; + + adapter._pruneStopWatermarks(); + assert.equal(adapter._stopWatermarkFor('held'), 1_000); + }); + + it('keeps a watermark until the poll cursor has provably moved past it', () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'c' }, 1_000); + adapter._lastDispatchedMessageTs = 1_000; // not yet strictly past + + adapter._pruneStopWatermarks(); + assert.equal(adapter._stopWatermarkFor('c'), 1_000); + }); + + it('forgets a watermark once it is spent, and its notice bookkeeping with it', async () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'c' }, 1_000); + await adapter._postStopNotice('c'); + adapter._lastDispatchedMessageTs = 2_000; + + adapter._pruneStopWatermarks(); + assert.equal(adapter._stopWatermarks.has('c'), false); + assert.equal(adapter._stopNoticedAt.has('c'), false); + }); + + it('never forgets a watermark for a queued message with no timestamp', () => { + const adapter = baseAdapter(); + adapter._markStopWatermark({ channel: 'c' }, 1_000); + adapter._channelQueues.c = [{ content: 'no timestamp' }]; + adapter._lastDispatchedMessageTs = 9_000; + + adapter._pruneStopWatermarks(); + assert.equal(adapter._stopWatermarkFor('c'), 1_000); + }); }); diff --git a/packages/go/web/components/chat/chat-view.tsx b/packages/go/web/components/chat/chat-view.tsx index 6370614c3..cd540178e 100644 --- a/packages/go/web/components/chat/chat-view.tsx +++ b/packages/go/web/components/chat/chat-view.tsx @@ -253,9 +253,9 @@ export function ChatView() { lastMsg.messageType === 'thinking' || lastMsg.messageType === 'loading' ); - updateLastMessage(currentSessionId, lastMsg.senderName, lastMsg.content, isWorking); + updateLastMessage(currentSessionId, lastMsg.senderName, lastMsg.content, isWorking, lastMsg.senderType === 'agent'); } else { - updateLastMessage(currentSessionId, '', ''); + updateLastMessage(currentSessionId, '', '', false, false); } }, [currentSessionId, displayMessages, updateLastMessage]); // eslint-disable-line react-hooks/exhaustive-deps diff --git a/packages/go/web/lib/stop-requests.test.ts b/packages/go/web/lib/stop-requests.test.ts index 858e2b8fd..4df763eba 100644 --- a/packages/go/web/lib/stop-requests.test.ts +++ b/packages/go/web/lib/stop-requests.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from 'vitest'; -import { StopRequestTracker } from './stop-requests'; +import { StopRequestTracker, confirmsStop } from './stop-requests'; describe('StopRequestTracker', () => { it('reports a fresh Stop as owning its sessions', () => { @@ -45,3 +45,25 @@ describe('StopRequestTracker', () => { expect(tracker.owned(['a'], second)).toEqual(['a']); }); }); + +describe('confirmsStop', () => { + it('rejects a human message', () => { + // Regression: the background poll can see the user's own just-sent message + // first, which used to clear the latch with nothing actually stopped. + expect(confirmsStop({ isAgent: false, isStatus: false, content: 'do the thing' })).toBe(false); + }); + + it('rejects a message with no known sender kind', () => { + expect(confirmsStop({ isStatus: false, content: 'anything' })).toBe(false); + }); + + it('accepts an agent reply', () => { + expect(confirmsStop({ isAgent: true, isStatus: false, content: 'Execution stopped by user.' })).toBe(true); + }); + + it('accepts an agent status only when it reports a terminal stop', () => { + expect(confirmsStop({ isAgent: true, isStatus: true, content: 'Bash > ls' })).toBe(false); + expect(confirmsStop({ isAgent: true, isStatus: true, content: 'stopped' })).toBe(true); + expect(confirmsStop({ isAgent: true, isStatus: true, content: 'stopping failed' })).toBe(true); + }); +}); diff --git a/packages/go/web/lib/stop-requests.ts b/packages/go/web/lib/stop-requests.ts index c8b0144ad..547f100b7 100644 --- a/packages/go/web/lib/stop-requests.ts +++ b/packages/go/web/lib/stop-requests.ts @@ -31,3 +31,19 @@ export class StopRequestTracker { this.owners.delete(sessionId); } } + +/** + * Whether a message arriving in a thread confirms that a pending Stop took + * effect. + * + * Only the agent can confirm. Accepting any non-status message let the user's + * own just-sent message — which the background poll often sees first — clear + * the latch and cancel the retry and give-up timers while nothing had actually + * stopped. + */ +export function confirmsStop( + { isAgent, isStatus, content }: { isAgent?: boolean; isStatus?: boolean; content: string }, +): boolean { + if (!isAgent) return false; + return !isStatus || /stopped|stopping failed/i.test(content); +} diff --git a/packages/go/web/lib/workspace-context.tsx b/packages/go/web/lib/workspace-context.tsx index 7d0ec8b6a..bd96537a9 100644 --- a/packages/go/web/lib/workspace-context.tsx +++ b/packages/go/web/lib/workspace-context.tsx @@ -3,7 +3,7 @@ import React, { createContext, useContext, useCallback, useEffect, useRef, useState } from 'react'; import { toast } from 'sonner'; import { workspaceApi } from './api'; -import { StopRequestTracker } from './stop-requests'; +import { StopRequestTracker, confirmsStop } from './stop-requests'; import { useOpenAgentsAuth } from './openagents-auth-context'; import { networkAgentToWorkspaceAgent, networkChannelToSession } from './types'; import type { BrowserPersistentContext, BrowserTab, DMConversation, RoutineItem, TodoItem, Workspace, WorkspaceAgent, WorkspaceCollaborator, WorkspaceFile, WorkspaceSession } from './types'; @@ -12,6 +12,8 @@ interface LastMessageInfo { content: string; senderName: string; isStatus?: boolean; + /** Sender is an agent. Only an agent's message can confirm a stop. */ + isAgent?: boolean; } interface WorkspaceContextValue { @@ -36,7 +38,7 @@ interface WorkspaceContextValue { monitorMode: boolean; acknowledgeCompletion: (sessionId: string) => void; agentModes: Record; - updateLastMessage: (sessionId: string, senderName: string, content: string, isStatus?: boolean) => void; + updateLastMessage: (sessionId: string, senderName: string, content: string, isStatus?: boolean, isAgent?: boolean) => void; setSessionActive: (sessionId: string, active: boolean) => void; updateAgentMode: (agentName: string, mode: string) => void; stopAllAgents: (sessionId?: string) => Promise; @@ -206,8 +208,11 @@ export function WorkspaceProvider({ try { localStorage.setItem('oa_notification_sound', String(enabled)); } catch {} }, []); - const updateLastMessage = useCallback((sessionId: string, senderName: string, content: string, isStatus?: boolean) => { - if (!isStatus || /stopped|stopping failed/i.test(content)) { + const updateLastMessage = useCallback((sessionId: string, senderName: string, content: string, isStatus?: boolean, isAgent?: boolean) => { + // Only the agent can confirm a stop. Treating any non-status message as + // confirmation let the user's own just-sent message clear the latch and + // cancel the retry and give-up timers, with nothing actually stopped. + if (confirmsStop({ isAgent, isStatus, content })) { stopRequestsRef.current.release(sessionId); setStoppingSessionIds((prev) => { if (!prev.has(sessionId)) return prev; @@ -274,15 +279,28 @@ export function WorkspaceProvider({ const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; const participants = session?.participants || []; - // Falls back to every agent when the participant list is unavailable — a - // thread created moments ago, or one discovery hasn't returned yet. Sending - // one stop too many is recoverable; sending none looks like a dead button. - // Safe only because each adapter now scopes a channel-named stop to that - // channel instead of stopping everything it is running. - const targetAgents = targetSessionId && participants.length > 0 + // Deliberately NOT falling back to every agent when the participant list is + // missing. Several adapters still ignore the channel a stop names and stop + // everything they are running, so broadcasting would let one thread's Stop + // kill an uninvolved agent's work in another thread. The give-up path below + // covers the "nobody to tell" case instead. + const targetAgents = targetSessionId ? agents.filter((a) => participants.includes(a.agentName)) : agents; + if (targetAgents.length === 0) { + // No agent to ask, so nothing will ever acknowledge. Release immediately + // rather than leaving the button disabled for the full timeout. + sessionIds.forEach((sid) => stopRequestsRef.current.release(sid)); + setStoppingSessionIds((prev) => { + if (!sessionIds.some((sid) => prev.has(sid))) return prev; + const next = new Set(prev); + sessionIds.forEach((sid) => next.delete(sid)); + return next; + }); + return; + } + const sendStop = () => Promise.allSettled( targetAgents.map((a) => { const channel = targetSessionId || undefined; @@ -463,7 +481,8 @@ export function WorkspaceProvider({ const content = payload?.content || ''; const msgType = payload?.message_type || 'chat'; const isStatus = msgType === 'status' || msgType === 'thinking'; - return { sessionId: ch.sessionId, senderName: sender, content, isStatus }; + const isAgent = (pick.source || '').startsWith('openagents:'); + return { sessionId: ch.sessionId, senderName: sender, content, isStatus, isAgent }; } catch { /* ignore */ } return null; }) @@ -472,7 +491,7 @@ export function WorkspaceProvider({ for (let i = 0; i < previews.length; i++) { const p = previews[i]; if (p && p.content) { - batch[p.sessionId] = { senderName: p.senderName, content: p.content.slice(0, 100), isStatus: p.isStatus }; + batch[p.sessionId] = { senderName: p.senderName, content: p.content.slice(0, 100), isStatus: p.isStatus, isAgent: p.isAgent }; } // Mark timestamp as known only after successful fetch (so failures retry next poll) if (p) { @@ -505,13 +524,15 @@ export function WorkspaceProvider({ newActive.add(sid); } } else { - stopRequestsRef.current.release(sid); - setStoppingSessionIds((s) => { - if (!s.has(sid)) return s; - const next = new Set(s); - next.delete(sid); - return next; - }); + if (confirmsStop({ isAgent: info.isAgent, isStatus: false, content: info.content })) { + stopRequestsRef.current.release(sid); + setStoppingSessionIds((s) => { + if (!s.has(sid)) return s; + const next = new Set(s); + next.delete(sid); + return next; + }); + } // Latest event is a real message — session is not working. // Always clear active so the shimmer doesn't stick when the // status→chat transition happens between polls or while diff --git a/workspace/frontend/components/chat/chat-view.tsx b/workspace/frontend/components/chat/chat-view.tsx index e599fc17d..b1134ab23 100644 --- a/workspace/frontend/components/chat/chat-view.tsx +++ b/workspace/frontend/components/chat/chat-view.tsx @@ -408,9 +408,9 @@ export function ChatView() { lastMsg.messageType === 'thinking' || lastMsg.messageType === 'loading' ); - updateLastMessage(currentSessionId, lastMsg.senderName, lastMsg.content, isWorking); + updateLastMessage(currentSessionId, lastMsg.senderName, lastMsg.content, isWorking, lastMsg.senderType === 'agent'); } else { - updateLastMessage(currentSessionId, '', ''); + updateLastMessage(currentSessionId, '', '', false, false); } }, [currentSessionId, displayMessages, updateLastMessage]); // eslint-disable-line react-hooks/exhaustive-deps diff --git a/workspace/frontend/lib/stop-requests.test.ts b/workspace/frontend/lib/stop-requests.test.ts index 858e2b8fd..4df763eba 100644 --- a/workspace/frontend/lib/stop-requests.test.ts +++ b/workspace/frontend/lib/stop-requests.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from 'vitest'; -import { StopRequestTracker } from './stop-requests'; +import { StopRequestTracker, confirmsStop } from './stop-requests'; describe('StopRequestTracker', () => { it('reports a fresh Stop as owning its sessions', () => { @@ -45,3 +45,25 @@ describe('StopRequestTracker', () => { expect(tracker.owned(['a'], second)).toEqual(['a']); }); }); + +describe('confirmsStop', () => { + it('rejects a human message', () => { + // Regression: the background poll can see the user's own just-sent message + // first, which used to clear the latch with nothing actually stopped. + expect(confirmsStop({ isAgent: false, isStatus: false, content: 'do the thing' })).toBe(false); + }); + + it('rejects a message with no known sender kind', () => { + expect(confirmsStop({ isStatus: false, content: 'anything' })).toBe(false); + }); + + it('accepts an agent reply', () => { + expect(confirmsStop({ isAgent: true, isStatus: false, content: 'Execution stopped by user.' })).toBe(true); + }); + + it('accepts an agent status only when it reports a terminal stop', () => { + expect(confirmsStop({ isAgent: true, isStatus: true, content: 'Bash > ls' })).toBe(false); + expect(confirmsStop({ isAgent: true, isStatus: true, content: 'stopped' })).toBe(true); + expect(confirmsStop({ isAgent: true, isStatus: true, content: 'stopping failed' })).toBe(true); + }); +}); diff --git a/workspace/frontend/lib/stop-requests.ts b/workspace/frontend/lib/stop-requests.ts index c8b0144ad..547f100b7 100644 --- a/workspace/frontend/lib/stop-requests.ts +++ b/workspace/frontend/lib/stop-requests.ts @@ -31,3 +31,19 @@ export class StopRequestTracker { this.owners.delete(sessionId); } } + +/** + * Whether a message arriving in a thread confirms that a pending Stop took + * effect. + * + * Only the agent can confirm. Accepting any non-status message let the user's + * own just-sent message — which the background poll often sees first — clear + * the latch and cancel the retry and give-up timers while nothing had actually + * stopped. + */ +export function confirmsStop( + { isAgent, isStatus, content }: { isAgent?: boolean; isStatus?: boolean; content: string }, +): boolean { + if (!isAgent) return false; + return !isStatus || /stopped|stopping failed/i.test(content); +} diff --git a/workspace/frontend/lib/workspace-context.tsx b/workspace/frontend/lib/workspace-context.tsx index 2f0cd1198..13c7a4658 100644 --- a/workspace/frontend/lib/workspace-context.tsx +++ b/workspace/frontend/lib/workspace-context.tsx @@ -2,7 +2,7 @@ import React, { createContext, useContext, useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { workspaceApi } from './api'; -import { StopRequestTracker } from './stop-requests'; +import { StopRequestTracker, confirmsStop } from './stop-requests'; import { capture, group } from './analytics'; import { useOpenAgentsAuth } from './openagents-auth-context'; import { generateUserId, getStoredIdentity, storeIdentity } from './identity'; @@ -113,6 +113,8 @@ interface LastMessageInfo { content: string; senderName: string; isStatus?: boolean; + /** Sender is an agent. Only an agent's message can confirm a stop. */ + isAgent?: boolean; } interface WorkspaceContextValue { @@ -137,7 +139,7 @@ interface WorkspaceContextValue { monitorMode: boolean; acknowledgeCompletion: (sessionId: string) => void; agentModes: Record; - updateLastMessage: (sessionId: string, senderName: string, content: string, isStatus?: boolean) => void; + updateLastMessage: (sessionId: string, senderName: string, content: string, isStatus?: boolean, isAgent?: boolean) => void; setSessionActive: (sessionId: string, active: boolean) => void; updateAgentMode: (agentName: string, mode: string) => void; stopAllAgents: (sessionId?: string) => Promise; @@ -503,8 +505,11 @@ export function WorkspaceProvider({ }; }, [currentUser.id, currentUser.name]); - const updateLastMessage = useCallback((sessionId: string, senderName: string, content: string, isStatus?: boolean) => { - if (!isStatus || /stopped|stopping failed/i.test(content)) { + const updateLastMessage = useCallback((sessionId: string, senderName: string, content: string, isStatus?: boolean, isAgent?: boolean) => { + // Only the agent can confirm a stop. Treating any non-status message as + // confirmation let the user's own just-sent message clear the latch and + // cancel the retry and give-up timers, with nothing actually stopped. + if (confirmsStop({ isAgent, isStatus, content })) { stopRequestsRef.current.release(sessionId); setStoppingSessionIds((prev) => { if (!prev.has(sessionId)) return prev; @@ -571,15 +576,28 @@ export function WorkspaceProvider({ const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; const participants = session?.participants || []; - // Falls back to every agent when the participant list is unavailable — a - // thread created moments ago, or one discovery hasn't returned yet. Sending - // one stop too many is recoverable; sending none looks like a dead button. - // Safe only because each adapter now scopes a channel-named stop to that - // channel instead of stopping everything it is running. - const targetAgents = targetSessionId && participants.length > 0 + // Deliberately NOT falling back to every agent when the participant list is + // missing. Several adapters still ignore the channel a stop names and stop + // everything they are running, so broadcasting would let one thread's Stop + // kill an uninvolved agent's work in another thread. The give-up path below + // covers the "nobody to tell" case instead. + const targetAgents = targetSessionId ? agents.filter((a) => participants.includes(a.agentName)) : agents; + if (targetAgents.length === 0) { + // No agent to ask, so nothing will ever acknowledge. Release immediately + // rather than leaving the button disabled for the full timeout. + sessionIds.forEach((sid) => stopRequestsRef.current.release(sid)); + setStoppingSessionIds((prev) => { + if (!sessionIds.some((sid) => prev.has(sid))) return prev; + const next = new Set(prev); + sessionIds.forEach((sid) => next.delete(sid)); + return next; + }); + return; + } + const sendStop = () => Promise.allSettled( targetAgents.map((a) => { const channel = targetSessionId || undefined; @@ -741,7 +759,8 @@ export function WorkspaceProvider({ const content = payload?.content || ''; const msgType = payload?.message_type || 'chat'; const isStatus = msgType === 'status' || msgType === 'thinking'; - return { sessionId: ch.sessionId, senderName: sender, content, isStatus }; + const isAgent = (pick.source || '').startsWith('openagents:'); + return { sessionId: ch.sessionId, senderName: sender, content, isStatus, isAgent }; } catch { /* ignore */ } return null; }) @@ -750,7 +769,7 @@ export function WorkspaceProvider({ for (let i = 0; i < previews.length; i++) { const p = previews[i]; if (p && p.content) { - batch[p.sessionId] = { senderName: p.senderName, content: p.content.slice(0, 100), isStatus: p.isStatus }; + batch[p.sessionId] = { senderName: p.senderName, content: p.content.slice(0, 100), isStatus: p.isStatus, isAgent: p.isAgent }; } // Mark timestamp as known only after successful fetch (so failures retry next poll) if (p) { @@ -783,13 +802,15 @@ export function WorkspaceProvider({ newActive.add(sid); } } else { - stopRequestsRef.current.release(sid); - setStoppingSessionIds((s) => { - if (!s.has(sid)) return s; - const next = new Set(s); - next.delete(sid); - return next; - }); + if (confirmsStop({ isAgent: info.isAgent, isStatus: false, content: info.content })) { + stopRequestsRef.current.release(sid); + setStoppingSessionIds((s) => { + if (!s.has(sid)) return s; + const next = new Set(s); + next.delete(sid); + return next; + }); + } // Latest event is a real message — session is not working. // Always clear active so the shimmer doesn't stick when the // status→chat transition happens between polls or while From bfe6c0df955ecc92e35efb84ba14fb18dacd9004 Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Tue, 1 Sep 2026 08:43:40 +0000 Subject: [PATCH 3/3] close the remaining Stop gaps found in review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pruning missed the message a channel is currently handling. Once handed to the worker it is no longer queued, so another channel's newer message could advance the cursor floor and drop the mark that message's own pre-spawn check depends on, and the run the user had cancelled started anyway. A busy channel now keeps its watermark, and the worker prunes again after it clears the busy flag. Minimal repro went from watermark 0 and stopped false to watermark 1000 and stopped true. Kimi's direct-API fallback was still global. With no CLI for the named channel it delegates to the direct adapter, whose in-flight request set carried no channel at all, so one thread's Stop destroyed every thread's request. Requests are now tagged with the channel that owns them and a named stop only destroys that channel's. This fixes every agent running in direct mode, not just Kimi. The four adapters scoped last round could still reach their CLI after a stop, through the same window already closed for Claude, Pi and OpenClaw — a stop arriving during auto-title or the status ping finds nothing to kill. Each now re-checks at the head of its attempt loop, which covers both the first spawn and the stale-session retry. The direct adapter checks before its API call and before posting a result. With no agent to send a stop to, the UI was left half-changed: the session had already been pulled out of active and its preview frozen at "Stopping...", so the thread lost its Stop button and never got it back. Targets are now resolved before any of that state is touched, so this case leaves the UI exactly as it was. --- packages/agent-connector/src/adapters/base.js | 8 ++ .../agent-connector/src/adapters/cline.js | 8 ++ .../agent-connector/src/adapters/copilot.js | 8 ++ .../agent-connector/src/adapters/cursor.js | 8 ++ packages/agent-connector/src/adapters/kimi.js | 8 ++ .../src/adapters/llm-direct.js | 34 ++++- .../agent-connector/test/stop-control.test.js | 135 ++++++++++++++++++ packages/go/web/lib/workspace-context.tsx | 39 ++--- workspace/frontend/lib/workspace-context.tsx | 39 ++--- 9 files changed, 236 insertions(+), 51 deletions(-) diff --git a/packages/agent-connector/src/adapters/base.js b/packages/agent-connector/src/adapters/base.js index 61768c28f..3eb88ede4 100644 --- a/packages/agent-connector/src/adapters/base.js +++ b/packages/agent-connector/src/adapters/base.js @@ -419,6 +419,12 @@ class BaseAdapter { 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; @@ -979,6 +985,8 @@ class BaseAdapter { } } this._channelBusy.delete(channel); + // Safe to reconsider this channel's watermark now that nothing is in flight. + this._pruneStopWatermarks(); } // ------------------------------------------------------------------ diff --git a/packages/agent-connector/src/adapters/cline.js b/packages/agent-connector/src/adapters/cline.js index 18aadf835..a16022e10 100644 --- a/packages/agent-connector/src/adapters/cline.js +++ b/packages/agent-connector/src/adapters/cline.js @@ -533,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 diff --git a/packages/agent-connector/src/adapters/copilot.js b/packages/agent-connector/src/adapters/copilot.js index 395684826..5d31b38d4 100644 --- a/packages/agent-connector/src/adapters/copilot.js +++ b/packages/agent-connector/src/adapters/copilot.js @@ -458,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); diff --git a/packages/agent-connector/src/adapters/cursor.js b/packages/agent-connector/src/adapters/cursor.js index d63992769..7fe801a3e 100644 --- a/packages/agent-connector/src/adapters/cursor.js +++ b/packages/agent-connector/src/adapters/cursor.js @@ -431,6 +431,14 @@ class CursorAdapter extends BaseAdapter { let effectiveContent = content; 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(msgChannel, msg)) { + this._log(`Not starting a run in ${msgChannel} — the user stopped it first`); + await this._postStopNotice(msgChannel); + return; + } if (attempt > 0) { try { const recap = await this._buildChannelRecap(msgChannel, content); diff --git a/packages/agent-connector/src/adapters/kimi.js b/packages/agent-connector/src/adapters/kimi.js index 8e880842b..434729ea7 100644 --- a/packages/agent-connector/src/adapters/kimi.js +++ b/packages/agent-connector/src/adapters/kimi.js @@ -562,6 +562,14 @@ class KimiAdapter extends LlmDirectAdapter { // 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; // Resuming → Kimi already has history, send the bare turn. Fresh → diff --git a/packages/agent-connector/src/adapters/llm-direct.js b/packages/agent-connector/src/adapters/llm-direct.js index 8ad772961..93a5da14d 100644 --- a/packages/agent-connector/src/adapters/llm-direct.js +++ b/packages/agent-connector/src/adapters/llm-direct.js @@ -62,13 +62,18 @@ class LlmDirectAdapter extends BaseAdapter { this._activeRequests.clear(); } - async _onControlAction(action, _payload) { + async _onControlAction(action, payload) { if (action === 'stop') { - for (const req of this._activeRequests) { + const channel = (payload && typeof payload === 'object') ? payload.channel : null; + for (const req of [...this._activeRequests]) { + if (channel && req.__oaChannel !== channel) continue; try { req.destroy(new Error('LLM API request stopped')); } catch {} + this._activeRequests.delete(req); } - this._activeRequests.clear(); + if (channel) await this._postStopNotice(channel); + return; } + await super._onControlAction(action, payload); } _buildSystemPrompt(channelName) { @@ -108,8 +113,20 @@ class LlmDirectAdapter extends BaseAdapter { return; } + // The stop may have arrived during the auto-title and status round trips + // above, when there was no request for it to destroy. + if (this._turnWasStopped(msgChannel, msg)) { + this._log(`Not calling the API for ${msgChannel} — the user stopped it first`); + await this._postStopNotice(msgChannel); + return; + } + const responseText = await this._callCompletionApi(content, msgChannel); + if (this._turnWasStopped(msgChannel, msg)) { + await this._postStopNotice(msgChannel); + return; + } if (responseText) { this._conversationHistory.push({ role: 'user', content }); this._conversationHistory.push({ role: 'assistant', content: responseText }); @@ -121,6 +138,13 @@ class LlmDirectAdapter extends BaseAdapter { await this.sendResponse(msgChannel, 'No response generated. Please try again.'); } } catch (e) { + // A stop destroys the request, which surfaces here as an error the user + // already knows about. + if (this._turnWasStopped(msgChannel, msg)) { + this._log(`Request in ${msgChannel} ended because the user stopped it`); + await this._postStopNotice(msgChannel); + return; + } this._log(`Error handling message: ${e.message}`); await this.sendError(msgChannel, `Error processing message: ${e.message}`); } @@ -199,6 +223,10 @@ class LlmDirectAdapter extends BaseAdapter { }); }); + // Tagged so a stop naming one channel does not tear down another + // channel's request — the set is shared by every channel this agent + // serves. + req.__oaChannel = channel; this._activeRequests.add(req); req.on('error', (err) => { this._activeRequests.delete(req); diff --git a/packages/agent-connector/test/stop-control.test.js b/packages/agent-connector/test/stop-control.test.js index deba461d7..d886ad70a 100644 --- a/packages/agent-connector/test/stop-control.test.js +++ b/packages/agent-connector/test/stop-control.test.js @@ -14,6 +14,7 @@ const CursorAdapter = require('../src/adapters/cursor'); const CopilotAdapter = require('../src/adapters/copilot'); const ClineAdapter = require('../src/adapters/cline'); const KimiAdapter = require('../src/adapters/kimi'); +const LlmDirectAdapter = require('../src/adapters/llm-direct'); function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -868,4 +869,138 @@ describe('agent stop control', () => { adapter._pruneStopWatermarks(); assert.equal(adapter._stopWatermarkFor('c'), 1_000); }); + + it('keeps a watermark while its channel is still handling a message', async () => { + // Regression: a message inside _handleMessage has left the queue but has + // not run yet. Another channel's newer message advanced the cursor floor + // and pruned the mark out from under that message's pre-spawn check. + const adapter = baseAdapter(); + adapter._prefetchPinnedContext = async () => {}; + + const msgA = { sessionId: 'A', content: 'a', createdAt: at(900) }; + let guardSaw = null; + adapter._handleMessage = async (m) => { + if (m.sessionId !== 'A') return; + await new Promise((r) => setTimeout(r, 60)); + guardSaw = adapter._turnWasStopped('A', msgA); + }; + + adapter._dispatchMessage(msgA); + await new Promise((r) => setTimeout(r, 10)); + adapter._markStopWatermark({ channel: 'A' }, 1_000); + await adapter._dispatchMessage({ sessionId: 'B', content: 'b', createdAt: at(5_000) }); + + assert.equal(adapter._stopWatermarkFor('A'), 1_000, 'busy channel keeps its mark'); + await new Promise((r) => setTimeout(r, 120)); + assert.equal(guardSaw, true, 'the pre-spawn check must still see the stop'); + }); + + it('prunes a channel once its worker is done with it', async () => { + const adapter = baseAdapter(); + adapter._prefetchPinnedContext = async () => {}; + adapter._handleMessage = async () => {}; + + adapter._markStopWatermark({ channel: 'A' }, 1_000); + adapter._lastDispatchedMessageTs = 5_000; + await adapter._channelWorker('A', { sessionId: 'A', content: 'x', createdAt: at(5_000) }); + + assert.equal(adapter._stopWatermarks.has('A'), false); + }); + + // ── Direct-API adapters ─────────────────────────────────────────── + + it('LLM direct stop destroys only the named channel\'s request', async () => { + const adapter = new LlmDirectAdapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'd', + }); + adapter._log = () => {}; + adapter.sendResponse = async () => {}; + + const destroyed = []; + const mkReq = (channel) => { + const req = { __oaChannel: channel, destroy() { destroyed.push(channel); } }; + adapter._activeRequests.add(req); + return req; + }; + mkReq('A'); + mkReq('B'); + + await adapter._onControlAction('stop', { channel: 'A' }); + + assert.deepEqual(destroyed, ['A'], 'channel B\'s request must survive'); + assert.equal(adapter._activeRequests.size, 1); + }); + + it('LLM direct does not call the API when the stop lands pre-request', async () => { + const adapter = new LlmDirectAdapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'd', + }); + adapter._log = () => {}; + const posted = []; + adapter.sendResponse = async (c, x) => posted.push(x); + adapter.sendError = async (c, x) => posted.push(`ERR ${x}`); + adapter._directMode = true; + adapter.sendStatus = async () => { + adapter._markStopWatermark({ channel: 'thread' }, 2_000); + }; + adapter._autoTitleChannel = async () => {}; + let called = 0; + adapter._callCompletionApi = async () => { called++; return 'late'; }; + + await adapter._handleMessage({ + sessionId: 'thread', content: 'go', createdAt: new Date(1_000).toISOString(), + }); + + assert.equal(called, 0); + assert.deepEqual(posted, ['Execution stopped by user.']); + }); + + // ── Pre-spawn window in the CLI adapters ────────────────────────── + + for (const [name, Adapter, chVar] of [ + ['Cursor', CursorAdapter, 'msgChannel'], ['Cline', ClineAdapter, 'channel'], + ['Copilot', CopilotAdapter, 'channel'], ['Kimi', KimiAdapter, 'channel'], + ]) { + it(`${name} does not spawn when the stop lands during the pre-spawn awaits`, async () => { + const adapter = new Adapter({ + workspaceId: 'ws', channelName: 'thread', token: 'token', agentName: 'a', + }); + adapter._log = () => {}; + const posted = []; + adapter.sendResponse = async (c, x) => posted.push(x); + adapter.sendStatus = async () => { + adapter._markStopWatermark({ channel: 'thread' }, 2_000); + }; + adapter.sendError = async (c, x) => posted.push(`ERR ${x}`); + adapter._autoTitleChannel = async () => {}; + + const msg = { sessionId: 'thread', content: 'go', createdAt: new Date(1_000).toISOString() }; + + // These adapters bail out early when their CLI is absent, which is the + // case on a test machine. Pretend it is installed so the guard under test + // is actually reached. + adapter._copilotBin = '/nonexistent/copilot'; + adapter._findClineBinary = () => '/nonexistent/cline'; + adapter._findKimiBinary = () => '/nonexistent/kimi'; + adapter._findCursorBinary = () => '/nonexistent/cursor'; + adapter._directMode = false; + adapter._resumableSession = () => null; + adapter._buildSystemContext = () => ''; + adapter._contextHeader = () => ''; + adapter._writeSkillFile = () => {}; + + // Whatever the adapter uses to reach its CLI must never be entered. + let spawned = 0; + for (const hook of ['_runTurn', '_spawnCli', '_runCli', '_spawnTurn', '_runOnce']) { + if (typeof adapter[hook] === 'function') { + adapter[hook] = async () => { spawned++; throw new Error(`${name} spawned after stop`); }; + } + } + + await adapter._handleMessage(msg); + assert.equal(spawned, 0, `${name} reached its CLI after the stop`); + assert.ok(posted.every((m) => !String(m).startsWith('ERR ')), `${name} posted an error: ${posted}`); + assert.deepEqual(posted, ['Execution stopped by user.']); + }); + } }); diff --git a/packages/go/web/lib/workspace-context.tsx b/packages/go/web/lib/workspace-context.tsx index bd96537a9..0d2dff266 100644 --- a/packages/go/web/lib/workspace-context.tsx +++ b/packages/go/web/lib/workspace-context.tsx @@ -257,6 +257,21 @@ export function WorkspaceProvider({ : Array.from(activeSessionIds); if (sessionIds.length === 0) return; + const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; + const participants = session?.participants || []; + // Deliberately NOT falling back to every agent when the participant list is + // missing. Several adapters still ignore the channel a stop names and stop + // everything they are running, so broadcasting would let one thread's Stop + // kill an uninvolved agent's work in another thread. + const targetAgents = targetSessionId + ? agents.filter((a) => participants.includes(a.agentName)) + : agents; + + // Resolved BEFORE anything is shown as stopping. With no agent to ask, + // nothing would ever acknowledge, and flipping the UI first would strand + // the thread with no Stop button and a preview frozen at "Stopping...". + if (targetAgents.length === 0) return; + setStoppingSessionIds((prev) => { const next = new Set(prev); sessionIds.forEach((sid) => next.add(sid)); @@ -277,30 +292,6 @@ export function WorkspaceProvider({ const generation = stopRequestsRef.current.claim(sessionIds); - const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; - const participants = session?.participants || []; - // Deliberately NOT falling back to every agent when the participant list is - // missing. Several adapters still ignore the channel a stop names and stop - // everything they are running, so broadcasting would let one thread's Stop - // kill an uninvolved agent's work in another thread. The give-up path below - // covers the "nobody to tell" case instead. - const targetAgents = targetSessionId - ? agents.filter((a) => participants.includes(a.agentName)) - : agents; - - if (targetAgents.length === 0) { - // No agent to ask, so nothing will ever acknowledge. Release immediately - // rather than leaving the button disabled for the full timeout. - sessionIds.forEach((sid) => stopRequestsRef.current.release(sid)); - setStoppingSessionIds((prev) => { - if (!sessionIds.some((sid) => prev.has(sid))) return prev; - const next = new Set(prev); - sessionIds.forEach((sid) => next.delete(sid)); - return next; - }); - return; - } - const sendStop = () => Promise.allSettled( targetAgents.map((a) => { const channel = targetSessionId || undefined; diff --git a/workspace/frontend/lib/workspace-context.tsx b/workspace/frontend/lib/workspace-context.tsx index 13c7a4658..0e2c6c6ab 100644 --- a/workspace/frontend/lib/workspace-context.tsx +++ b/workspace/frontend/lib/workspace-context.tsx @@ -554,6 +554,21 @@ export function WorkspaceProvider({ : Array.from(activeSessionIds); if (sessionIds.length === 0) return; + const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; + const participants = session?.participants || []; + // Deliberately NOT falling back to every agent when the participant list is + // missing. Several adapters still ignore the channel a stop names and stop + // everything they are running, so broadcasting would let one thread's Stop + // kill an uninvolved agent's work in another thread. + const targetAgents = targetSessionId + ? agents.filter((a) => participants.includes(a.agentName)) + : agents; + + // Resolved BEFORE anything is shown as stopping. With no agent to ask, + // nothing would ever acknowledge, and flipping the UI first would strand + // the thread with no Stop button and a preview frozen at "Stopping...". + if (targetAgents.length === 0) return; + setStoppingSessionIds((prev) => { const next = new Set(prev); sessionIds.forEach((sid) => next.add(sid)); @@ -574,30 +589,6 @@ export function WorkspaceProvider({ const generation = stopRequestsRef.current.claim(sessionIds); - const session = targetSessionId ? sessions.find((s) => s.sessionId === targetSessionId) : null; - const participants = session?.participants || []; - // Deliberately NOT falling back to every agent when the participant list is - // missing. Several adapters still ignore the channel a stop names and stop - // everything they are running, so broadcasting would let one thread's Stop - // kill an uninvolved agent's work in another thread. The give-up path below - // covers the "nobody to tell" case instead. - const targetAgents = targetSessionId - ? agents.filter((a) => participants.includes(a.agentName)) - : agents; - - if (targetAgents.length === 0) { - // No agent to ask, so nothing will ever acknowledge. Release immediately - // rather than leaving the button disabled for the full timeout. - sessionIds.forEach((sid) => stopRequestsRef.current.release(sid)); - setStoppingSessionIds((prev) => { - if (!sessionIds.some((sid) => prev.has(sid))) return prev; - const next = new Set(prev); - sessionIds.forEach((sid) => next.delete(sid)); - return next; - }); - return; - } - const sendStop = () => Promise.allSettled( targetAgents.map((a) => { const channel = targetSessionId || undefined;