fix: immediate stop feedback and synthetic execution_complete (#79)

* fix: immediate stop feedback and synthetic execution_complete

stopExecution() now marks the processing message as failed, broadcasts
a synthetic execution_complete (so all clients flush buffered tokens),
and sends processing_status: false immediately.

processSandboxEvent() guards execution_complete with isStillProcessing
so the bridge's late execution_complete doesn't double-update a stopped
message. Queue draining and snapshots still run regardless.

* fix: address PR review feedback

- Remove unnecessary `as ServerMessage` cast on synthetic execution_complete
  broadcast — the type already matches the ServerMessage union
- Broadcast execution_complete to clients in the normal (non-stopped) completion
  path — the early return skipped the generic broadcast, so normal completions
  were never sent to WebSocket clients
- Update unit test header comment to accurately describe what's tested
  (repository methods, not DO behavior)

* chore: clarify why stopExecution notifies slack-bot directly

* refactor: simplify completionMessageId and add queue-drain integration test

- Remove dead `?? event.messageId` fallback (messageId already resolves it)
- Add integration test verifying queue drain dispatches next message when
  sandbox WS is connected after stop + late execution_complete

Author: ColeMurray

packages/control-plane/src/session/durable-object.ts

@@ -1042,14 +1042,18 @@ export class SessionDO extends DurableObject<Env> {
 
     // Handle specific event types
     if (event.type === "execution_complete") {
-      // Use the resolved messageId (which now correctly prioritizes event.messageId)
-      const completionMessageId = messageId ?? event.messageId;
-      const status = event.success ? "completed" : "failed";
+      // messageId already incorporates event.messageId (line above), so no extra fallback needed
+      const completionMessageId = messageId;
 
-      if (completionMessageId) {
+      // Only update message status if it's still processing (not already stopped)
+      const isStillProcessing =
+        completionMessageId != null && processingMessage?.id === completionMessageId;
+
+      if (isStillProcessing) {
+        // Normal path: message still processing, complete it as before
+        const status = event.success ? "completed" : "failed";
         this.repository.updateMessageCompletion(completionMessageId, status, now);
 
-        // Emit prompt.complete wide event with duration metrics
         const timestamps = this.repository.getMessageTimestamps(completionMessageId);
         const totalDurationMs = timestamps ? now - timestamps.created_at : undefined;
         const processingDurationMs =
@@ -1069,30 +1073,24 @@ export class SessionDO extends DurableObject<Env> {
           queue_duration_ms: queueDurationMs,
         });
 
-        // Broadcast processing status change (after DB update so getIsProcessing is accurate)
+        this.broadcast({ type: "sandbox_event", event });
         this.broadcast({ type: "processing_status", isProcessing: this.getIsProcessing() });
-
-        // Notify slack-bot of completion (fire-and-forget with retry)
         this.ctx.waitUntil(this.notifySlackBot(completionMessageId, event.success));
       } else {
-        this.log.warn("prompt.complete", {
+        // Stopped path: message was already marked failed by stopExecution()
+        this.log.info("prompt.complete", {
           event: "prompt.complete",
-          outcome: "error",
-          error_reason: "no_message_id",
+          message_id: completionMessageId,
+          outcome: "already_stopped",
         });
       }
 
-      // Take snapshot after execution completes (per Ramp spec)
-      // "When the agent is finished making changes, we take another snapshot"
-      // Use fire-and-forget so snapshot doesn't block the response to the user
+      // Always run these regardless of stop (snapshot, activity, queue drain)
       this.ctx.waitUntil(this.triggerSnapshot("execution_complete"));
-
-      // Reset activity timer - give user time to review output before inactivity timeout
       this.updateLastActivity(now);
       await this.scheduleInactivityCheck();
-
-      // Process next in queue
       await this.processMessageQueue();
+      return; // execution_complete handling is done; skip the generic broadcast below
     }
 
     if (event.type === "git_sync") {
@@ -1108,7 +1106,7 @@ export class SessionDO extends DurableObject<Env> {
       this.handlePushEvent(event);
     }
 
-    // Broadcast to clients
+    // Broadcast to clients (all non-execution_complete events)
     this.broadcast({ type: "sandbox_event", event });
   }
 
@@ -1305,10 +1303,42 @@ export class SessionDO extends DurableObject<Env> {
 
   /**
    * Stop current execution.
-   * Sends stop command to sandbox, which should respond with execution_complete.
-   * The processing status will be updated when execution_complete is received.
+   * Marks the processing message as failed, broadcasts synthetic execution_complete
+   * so all clients flush buffered tokens, and forwards stop to the sandbox.
    */
   private async stopExecution(): Promise<void> {
+    const now = Date.now();
+    const processingMessage = this.repository.getProcessingMessage();
+
+    if (processingMessage) {
+      this.repository.updateMessageCompletion(processingMessage.id, "failed", now);
+      this.log.info("prompt.stopped", {
+        event: "prompt.stopped",
+        message_id: processingMessage.id,
+      });
+
+      // Broadcast synthetic execution_complete so ALL clients flush buffered tokens.
+      // (The stop-clicking client flushes locally, but other connected clients don't.)
+      this.broadcast({
+        type: "sandbox_event",
+        event: {
+          type: "execution_complete",
+          messageId: processingMessage.id,
+          success: false,
+          sandboxId: "",
+          timestamp: now / 1000,
+        },
+      });
+
+      // Notify slack-bot now because the bridge's late execution_complete will hit
+      // the "already_stopped" branch in processSandboxEvent() which skips notification.
+      this.ctx.waitUntil(this.notifySlackBot(processingMessage.id, false));
+    }
+
+    // Immediate client feedback
+    this.broadcast({ type: "processing_status", isProcessing: false });
+
+    // Forward stop to sandbox (bridge cancels its task)
     const sandboxWs = this.wsManager.getSandboxSocket();
     if (sandboxWs) {
       this.wsManager.send(sandboxWs, { type: "stop" });
@@ -2031,8 +2061,8 @@ export class SessionDO extends DurableObject<Env> {
     }
   }
 
-  private handleStop(): Response {
-    this.stopExecution();
+  private async handleStop(): Promise<Response> {
+    await this.stopExecution();
     return Response.json({ status: "stopping" });
   }
 

packages/control-plane/src/session/stop-execution.test.ts

@@ -0,0 +1,121 @@
+/**
+ * Unit tests for the stop-execution–related repository behavior.
+ *
+ * These tests exercise SessionRepository methods (e.g. getProcessingMessage()
+ * and updateMessageCompletion()) that are used by stopExecution() and the
+ * execution_complete guard in processSandboxEvent().
+ *
+ * We focus here on the repository-level interactions and state transitions
+ * by directly calling the repository methods and verifying their effects.
+ */
+
+import { describe, it, expect, beforeEach } from "vitest";
+import { SessionRepository, type SqlStorage, type SqlResult } from "./repository";
+
+/**
+ * Create a mock SqlStorage that tracks calls and can return configurable data.
+ */
+function createMockSql() {
+  const calls: Array<{ query: string; params: unknown[] }> = [];
+  const mockData: Map<string, unknown[]> = new Map();
+  let oneValue: unknown = null;
+
+  const sql: SqlStorage = {
+    exec(query: string, ...params: unknown[]): SqlResult {
+      calls.push({ query, params });
+      const data = mockData.get(query) ?? [];
+      return {
+        toArray: () => data,
+        one: () => oneValue,
+      };
+    },
+  };
+
+  return {
+    sql,
+    calls,
+    setData(query: string, data: unknown[]) {
+      mockData.set(query, data);
+    },
+    setOne(value: unknown) {
+      oneValue = value;
+    },
+    reset() {
+      calls.length = 0;
+      mockData.clear();
+      oneValue = null;
+    },
+  };
+}
+
+describe("Stop execution - repository interactions", () => {
+  let mock: ReturnType<typeof createMockSql>;
+  let repo: SessionRepository;
+
+  beforeEach(() => {
+    mock = createMockSql();
+    repo = new SessionRepository(mock.sql);
+  });
+
+  describe("getProcessingMessage", () => {
+    it("returns message when one is processing", () => {
+      mock.setData(`SELECT id FROM messages WHERE status = 'processing' LIMIT 1`, [
+        { id: "msg-1" },
+      ]);
+      const result = repo.getProcessingMessage();
+      expect(result).toEqual({ id: "msg-1" });
+    });
+
+    it("returns null when no message is processing", () => {
+      mock.setData(`SELECT id FROM messages WHERE status = 'processing' LIMIT 1`, []);
+      expect(repo.getProcessingMessage()).toBeNull();
+    });
+  });
+
+  describe("updateMessageCompletion", () => {
+    it("calls SQL with correct parameters for failed status", () => {
+      repo.updateMessageCompletion("msg-1", "failed", 1000);
+
+      const call = mock.calls.find((c) => c.query.includes("UPDATE messages SET status"));
+      expect(call).toBeDefined();
+      expect(call!.params).toContain("failed");
+      expect(call!.params).toContain("msg-1");
+      expect(call!.params).toContain(1000);
+    });
+
+    it("calls SQL with correct parameters for completed status", () => {
+      repo.updateMessageCompletion("msg-2", "completed", 2000);
+
+      const call = mock.calls.find((c) => c.query.includes("UPDATE messages SET status"));
+      expect(call).toBeDefined();
+      expect(call!.params).toContain("completed");
+      expect(call!.params).toContain("msg-2");
+    });
+  });
+
+  describe("stopExecution state machine", () => {
+    it("marks processing message as failed, then getProcessingMessage returns null", () => {
+      // First call: message is processing
+      mock.setData(`SELECT id FROM messages WHERE status = 'processing' LIMIT 1`, [
+        { id: "msg-1" },
+      ]);
+      const processing = repo.getProcessingMessage();
+      expect(processing).toEqual({ id: "msg-1" });
+
+      // Mark as failed
+      repo.updateMessageCompletion("msg-1", "failed", Date.now());
+
+      // After update, simulate no processing messages
+      mock.setData(`SELECT id FROM messages WHERE status = 'processing' LIMIT 1`, []);
+      expect(repo.getProcessingMessage()).toBeNull();
+    });
+
+    it("does not error when no processing message exists", () => {
+      mock.setData(`SELECT id FROM messages WHERE status = 'processing' LIMIT 1`, []);
+
+      const processing = repo.getProcessingMessage();
+      expect(processing).toBeNull();
+      // No updateMessageCompletion call needed - this is the idempotent case
+    });
+  });
+});