diff --git a/packages/agent-connector/src/adapters/base.js b/packages/agent-connector/src/adapters/base.js index 4276758bc..3eb88ede4 100644 --- a/packages/agent-connector/src/adapters/base.js +++ b/packages/agent-connector/src/adapters/base.js @@ -91,6 +91,28 @@ class BaseAdapter { // Per-channel task tracking for parallel execution this._channelBusy = new Set(); this._channelQueues = {}; + // ── Stop watermarks ── + // A stop control event carries the SERVER timestamp of the moment the user + // hit Stop. Any message posted at or before that instant belongs to the turn + // the user just cancelled, so it must not start a fresh run — even if it + // reaches us afterwards, which is the common case: the message is posted, + // the user immediately hits Stop, and only then do we poll the message. + // Comparing server-to-server timestamps keeps this free of clock skew. + // + // Watermarks are NEVER cleared when a newer message passes. A newer message + // already fails the `<=` test on its own, and dropping the mark would let a + // message that was queued BEFORE the stop through when the worker drains + // that queue later. + this._stopWatermarks = new Map(); // channel → server ms + this._stopWatermarkAll = 0; // stop with no channel (whole agent) + // Newest timestamp among messages we have accepted for processing. The poll + // cursor never goes backwards, so this is the floor below which the server + // can no longer deliver anything — see _pruneStopWatermarks. + this._lastDispatchedMessageTs = 0; + // channel → the watermark we already announced a stop for, so a burst of + // dropped messages posts one notice while a genuinely new stop (larger + // watermark) is still allowed to announce itself. + this._stopNoticedAt = new Map(); // Cached workspace.browser_enabled. Populated lazily on first read so we // don't pay an HTTP roundtrip per message — adapters that toggle the // workspace flag must reconnect/restart to pick up the change (matches @@ -339,6 +361,10 @@ class BaseAdapter { if (ev.id) this._lastControlId = ev.id; const payload = ev.payload || {}; const action = payload.action; + // Record the watermark BEFORE the adapter-specific handler runs, so a + // message that lands while that handler is still killing processes is + // already covered. + if (action === 'stop') this._markStopWatermark(payload, ev.timestamp); if (action === 'set_mode') { const newMode = payload.mode || 'execute'; if ((newMode === 'execute' || newMode === 'plan') && newMode !== this._mode) { @@ -353,6 +379,117 @@ class BaseAdapter { } catch {} } + /** + * Remember when a stop was issued, using the control event's own server + * timestamp. An unusable timestamp is skipped rather than substituted with + * `Date.now()` — mixing this machine's clock into a server timeline would + * drop or admit the wrong messages under clock skew. + */ + _markStopWatermark(payload, timestamp) { + const ts = Number(timestamp); + if (!Number.isFinite(ts) || ts <= 0) { + this._log('Stop control event carried no usable timestamp — no watermark set'); + return; + } + const channel = payload && (payload.channel || payload.sessionId); + if (!channel) { + this._stopWatermarkAll = Math.max(this._stopWatermarkAll, ts); + return; + } + this._stopWatermarks.set(String(channel), ts); + } + + /** + * Forget watermarks that can no longer match anything, so a long-lived agent + * does not accumulate one entry per channel it was ever stopped in. + * + * A watermark is only dropped when it is provably spent. The poll cursor is + * monotonic in (timestamp, id) and shared by every channel, so once a message + * timestamped T has been dispatched the server can never hand us one older + * than T again — for any channel. What the server can no longer deliver, only + * our own queue can, so a watermark below T is spent once nothing at or below + * it is still queued. + * + * Evicting by count instead would be a correctness bug, not a safety net: the + * mark it discarded might be the one still holding back a queued message. + */ + _pruneStopWatermarks() { + if (this._stopWatermarks.size === 0) return; + const delivered = this._lastDispatchedMessageTs; + if (!delivered) return; + for (const [channel, watermark] of this._stopWatermarks) { + if (watermark >= delivered) continue; + // A message already inside _handleMessage has left the queue but has not + // finished: it re-checks the watermark before it starts work, so the mark + // has to outlive the turn. Another channel's newer message would + // otherwise prune it out from under that check. The worker prunes again + // once it is no longer busy. + if (this._channelBusy.has(channel)) continue; + const queue = this._channelQueues[channel] || []; + const stillHeld = queue.some((m) => { + const ts = m && m.createdAt ? Date.parse(m.createdAt) : NaN; + return !Number.isFinite(ts) || ts <= watermark; + }); + if (stillHeld) continue; + this._stopWatermarks.delete(channel); + this._stopNoticedAt.delete(channel); + } + } + + _stopWatermarkFor(channel) { + return Math.max(this._stopWatermarks.get(channel) || 0, this._stopWatermarkAll); + } + + /** + * True when *msg* was posted at or before the last stop for its channel, i.e. + * the user cancelled the turn this message belongs to. A message with no + * usable timestamp is never dropped — silently swallowing what someone typed + * is worse than running one extra turn. + */ + _isStoppedOut(channel, msg) { + const watermark = this._stopWatermarkFor(channel); + if (!watermark) return false; + const ts = msg && msg.createdAt ? Date.parse(msg.createdAt) : NaN; + if (!Number.isFinite(ts)) return false; + return ts <= watermark; + } + + /** + * Re-check, part way through handling *msg*, whether the user stopped this + * turn in the meantime. + * + * The dispatch-time check cannot cover the whole turn. Between it and the + * moment the CLI is actually spawned an adapter awaits several round trips + * (auto-title, a status ping), and a stop landing in that window finds no + * process to kill — so without this the run starts anyway and its answer is + * posted after "Execution stopped by user.". Call it immediately before + * starting the work, and again before posting a result. + */ + _turnWasStopped(channel, msg) { + return this._isStoppedOut(channel, msg); + } + + /** + * Announce "stopped" in *channel*, at most once per stop. + * + * The workspace UI keeps its Stop button in a disabled "Stopping…" state + * until a non-status message lands in the thread, so a stop that posts + * nothing locks that button. Every stop path must end here — including the + * ones where there was no process to kill. + * + * Dedup is keyed on the current watermark rather than a flag someone has to + * remember to reset: a later stop raises the watermark and is free to + * announce itself, while several messages dropped by the same stop share one + * notice. + */ + async _postStopNotice(channel, message = 'Execution stopped by user.') { + if (!channel) return; + const watermark = this._stopWatermarkFor(channel); + if (this._stopNoticedAt.get(channel) === watermark) return; + this._stopNoticedAt.set(channel, watermark); + try { await this.sendResponse(channel, message); } catch {} + } + /** * Handle adapter-specific control actions. Override in subclasses to add * per-adapter actions (`stop`, `restart`, …); always call @@ -760,6 +897,21 @@ class BaseAdapter { channel = msg.sessionId; } + // The user hit Stop after posting this. Killing the subprocess alone would + // not have helped — this message had not started running yet, so without + // this check it spawns a fresh run one poll later and Stop looks broken. + if (this._isStoppedOut(channel, msg)) { + this._log(`Dropping message posted before Stop in ${channel}`); + await this._postStopNotice(channel); + return; + } + + const dispatchedTs = msg && msg.createdAt ? Date.parse(msg.createdAt) : NaN; + if (Number.isFinite(dispatchedTs)) { + this._lastDispatchedMessageTs = Math.max(this._lastDispatchedMessageTs, dispatchedTs); + this._pruneStopWatermarks(); + } + if (this._channelBusy.has(channel)) { // A routine that's already running must not stack up. Routine fires are // periodic, so a fire that arrives while the previous run is still going @@ -813,6 +965,13 @@ class BaseAdapter { const queue = this._channelQueues[channel]; if (!queue || queue.length === 0) break; const nextMsg = queue.shift(); + // Queued before the stop that just landed — drop it rather than running + // the work the user cancelled. + if (this._isStoppedOut(channel, nextMsg)) { + this._log(`Dropping queued message posted before Stop in ${channel}`); + await this._postStopNotice(channel); + continue; + } if (nextMsg._queueId) { try { await this.sendStatus(channel, 'processing queued message', { queue_id: nextMsg._queueId, queue_status: 'processed' }); } catch {} } @@ -826,6 +985,8 @@ class BaseAdapter { } } this._channelBusy.delete(channel); + // Safe to reconsider this channel's watermark now that nothing is in flight. + this._pruneStopWatermarks(); } // ------------------------------------------------------------------ diff --git a/packages/agent-connector/src/adapters/claude.js b/packages/agent-connector/src/adapters/claude.js index 52292bdaa..30887c597 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)}...`); @@ -1156,6 +1148,16 @@ class ClaudeAdapter extends BaseAdapter { ? { enabled: true, entryId: glossary.entryId, content: glossary.content, scope: glossary.scope || 'channel' } : null; + // The stop may have arrived during the auto-title, status and knowledge + // round trips above, at a moment when there was no process for it to kill. + // Handing the message to a process now would run exactly what the user + // cancelled. + if (this._turnWasStopped(msgChannel, msg)) { + this._log(`Not starting a run in ${msgChannel} — the user stopped it first`); + await this._postStopNotice(msgChannel); + return; + } + // ── Persistent process fast-path ── // If we have a living persistent process for this channel, send via stdin // instead of spawning a new CLI (saves ~2s startup time). diff --git a/packages/agent-connector/src/adapters/cline.js b/packages/agent-connector/src/adapters/cline.js index 325a38ab2..a16022e10 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.'); } @@ -528,6 +533,14 @@ class ClineAdapter extends BaseAdapter { // One retry: if resuming a stale session fails, retry once fresh. for (let attempt = 0; attempt < 2; attempt++) { + // Re-checked every attempt. The stop can land during the awaits above, + // when there is no process to kill, and again between the first spawn and + // the stale-session retry — either way the work must not start. + if (this._turnWasStopped(channel, msg)) { + this._log(`Not starting a run in ${channel} — the user stopped it first`); + await this._postStopNotice(channel); + return; + } const resumeId = attempt === 0 ? this._resumableSession(channel, workingDir) : null; // Build the prompt. Resuming → Cline already has history, send the bare diff --git a/packages/agent-connector/src/adapters/copilot.js b/packages/agent-connector/src/adapters/copilot.js index feae3a9d0..5d31b38d4 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)) { @@ -449,6 +458,14 @@ class CopilotAdapter extends BaseAdapter { // Up to 2 attempts: resume first, then a fresh session if resume was stale. for (let attempt = 0; attempt < 2; attempt++) { + // Re-checked every attempt. The stop can land during the awaits above, + // when there is no process to kill, and again between the first spawn and + // the stale-session retry — either way the work must not start. + if (this._turnWasStopped(channel, msg)) { + this._log(`Not starting a run in ${channel} — the user stopped it first`); + await this._postStopNotice(channel); + return; + } const skipResume = attempt > 0; const args = this._buildArgs(fullPrompt, channel, { skipResume }); this._logSpawn(channel, args, skipResume); diff --git a/packages/agent-connector/src/adapters/cursor.js b/packages/agent-connector/src/adapters/cursor.js index be01d595e..7fe801a3e 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.'); } @@ -426,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 0c9496ab1..434729ea7 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) { @@ -552,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/src/adapters/openclaw.js b/packages/agent-connector/src/adapters/openclaw.js index 71c2d6685..9ea0544db 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,24 +248,117 @@ 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...'); + // 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 { 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._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}`); 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 +481,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 +524,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 +536,7 @@ class OpenClawAdapter extends BaseAdapter { clearInterval(pollInterval); clearTimeout(killTimeout); closeFd(); + releaseProc(); try { fs.unlinkSync(stderrFile); } catch {} reject(err); }); @@ -433,6 +546,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..b5f9d9a04 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; @@ -1395,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 71926824f..d886ad70a 100644 --- a/packages/agent-connector/test/stop-control.test.js +++ b/packages/agent-connector/test/stop-control.test.js @@ -8,6 +8,13 @@ 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'); +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)); @@ -466,4 +473,534 @@ 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); + }); + + // ── 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')), + [], + ); + }); + + // ── 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); + }); + + 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/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 new file mode 100644 index 000000000..4df763eba --- /dev/null +++ b/packages/go/web/lib/stop-requests.test.ts @@ -0,0 +1,69 @@ +import { describe, it, expect } from 'vitest'; +import { StopRequestTracker, confirmsStop } 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']); + }); +}); + +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 new file mode 100644 index 000000000..547f100b7 --- /dev/null +++ b/packages/go/web/lib/stop-requests.ts @@ -0,0 +1,49 @@ +/** + * 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); + } +} + +/** + * 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 46a79be65..0d2dff266 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, 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'; @@ -11,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 { @@ -35,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; @@ -85,6 +88,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 +149,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([]); @@ -193,8 +208,12 @@ 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; const next = new Set(prev); @@ -238,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)); @@ -256,12 +290,7 @@ 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); - }) - : agents; + const generation = stopRequestsRef.current.claim(sessionIds); const sendStop = () => Promise.allSettled( targetAgents.map((a) => { @@ -271,13 +300,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 @@ -414,7 +472,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; }) @@ -423,7 +482,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) { @@ -443,6 +502,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,12 +515,15 @@ export function WorkspaceProvider({ newActive.add(sid); } } else { - 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 new file mode 100644 index 000000000..4df763eba --- /dev/null +++ b/workspace/frontend/lib/stop-requests.test.ts @@ -0,0 +1,69 @@ +import { describe, it, expect } from 'vitest'; +import { StopRequestTracker, confirmsStop } 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']); + }); +}); + +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 new file mode 100644 index 000000000..547f100b7 --- /dev/null +++ b/workspace/frontend/lib/stop-requests.ts @@ -0,0 +1,49 @@ +/** + * 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); + } +} + +/** + * 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 d32763db1..0e2c6c6ab 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, confirmsStop } from './stop-requests'; import { capture, group } from './analytics'; import { useOpenAgentsAuth } from './openagents-auth-context'; import { generateUserId, getStoredIdentity, storeIdentity } from './identity'; @@ -112,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 { @@ -136,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; @@ -245,6 +248,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 +340,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([]); @@ -490,8 +505,12 @@ 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; const next = new Set(prev); @@ -535,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)); @@ -553,12 +587,7 @@ 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); - }) - : agents; + const generation = stopRequestsRef.current.claim(sessionIds); const sendStop = () => Promise.allSettled( targetAgents.map((a) => { @@ -568,13 +597,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 @@ -692,7 +750,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; }) @@ -701,7 +760,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) { @@ -721,6 +780,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,12 +793,15 @@ export function WorkspaceProvider({ newActive.add(sid); } } else { - 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