fix: keep consecutive requests running and clear finished stop state

Wait for OpenCode session cancellation before reusing a conversation, and replace the session after unconfirmed cancellation. Preserve terminal message state when stale streaming snapshots arrive so finished conversations return to Send.

Validated with full production Electron UI and the configured real text model: queued requests complete, cancellation followed by immediate resend completes, and Stop clears. Focused tests: 362 passed; typecheck and targeted lint passed. Live dispatch counting limitations are recorded in the feature evidence.

Release note: Consecutive OpenCode requests no longer inherit a previous cancellation, and finished conversations no longer retain the Stop button after stale updates.

src/main/agent/opencode-runtime-lifecycle.test.ts

@@ -49,6 +49,9 @@ it.skipIf(!existsSync(binaryPath))(
     const gateway = new KnowledgeMcpGateway({} as never)
     await gateway.start()
     let openEventBodies = 0
+    let delayedAbort: Promise<void> | undefined
+    let abortStarted: (() => void) | undefined
+    let promptCount = 0
     const runtime = new OpenCodeRuntime({
       embedded: true, binaryPath: '', bundledBinaryPath: binaryPath,
       configPath: '', defaultWorkspace: workspace, knowledgeGateway: gateway,
@@ -61,8 +64,14 @@ it.skipIf(!existsSync(binaryPath))(
       createClient: (options) => createOpencodeClient({
         ...options,
         fetch: async (request, init) => {
-          const response = await fetch(request, init)
           const url = request instanceof Request ? request.url : request
+          const pathname = new URL(url).pathname
+          if (pathname.endsWith('/abort') && delayedAbort) {
+            abortStarted?.()
+            await delayedAbort
+          }
+          if (pathname.endsWith('/prompt_async')) promptCount++
+          const response = await fetch(request, init)
           if (new URL(url).pathname !== '/event' || !response.body) return response
           openEventBodies++
           const reader = response.body.getReader()
@@ -130,6 +139,30 @@ it.skipIf(!existsSync(binaryPath))(
         expect(run()).resolves.toBe('LIFECYCLE_OK')
       ])
       await expect.poll(() => openEventBodies).toBe(0)
+      let releaseAbort!: () => void
+      delayedAbort = new Promise<void>((resolve) => { releaseAbort = resolve })
+      const abortPending = new Promise<void>((resolve) => { abortStarted = resolve })
+      const reusedConversation = crypto.randomUUID()
+      cancelledController = new AbortController()
+      const first = run(cancelledController, true, reusedConversation).catch((error: unknown) => error)
+      let second: Promise<string> | undefined
+      try {
+        await abortPending
+        const beforeResend = promptCount
+        second = run(new AbortController(), false, reusedConversation)
+        await new Promise((resolve) => setTimeout(resolve, 100))
+        const beforeAbortCompletes = promptCount
+        releaseAbort()
+        expect(await first).toMatchObject({ name: 'AbortError' })
+        await expect(second).resolves.toBe('LIFECYCLE_OK')
+        expect(beforeAbortCompletes).toBe(beforeResend)
+        expect(promptCount).toBe(beforeResend + 1)
+        await expect.poll(() => openEventBodies).toBe(0)
+      } finally {
+        releaseAbort()
+        await Promise.allSettled([first, second])
+        delayedAbort = undefined
+      }
       const conversationId = crypto.randomUUID()
       await expect(run(new AbortController(), false, conversationId)).resolves.toBe('LIFECYCLE_OK')
       await expect(runtime.compactConversation({

src/main/agent/opencode-runtime.test.ts

@@ -1924,6 +1924,137 @@ describe("OpenCodeRuntime embedded launcher", () => {
     await runtime.dispose();
   });
 
+  it.each(["cancel", "stream-error"])("waits for delayed session abort before reusing a conversation after %s", async (outcome) => {
+    const setup = runClient([]);
+    const controller = new AbortController();
+    let releaseAbort!: () => void;
+    const abortGate = new Promise<void>((resolve) => { releaseAbort = resolve; });
+    let endFirst!: () => void;
+    const firstGate = new Promise<void>((resolve) => { endFirst = resolve; });
+    let secondActive = false;
+    let secondInterrupted = false;
+    let subscriptions = 0;
+    vi.mocked(setup.session.abort).mockImplementation(async () => {
+      await abortGate;
+      if (secondActive) secondInterrupted = true;
+      return { data: true, error: undefined } as never;
+    });
+    vi.mocked(setup.event.subscribe).mockImplementation(async () => {
+      const first = ++subscriptions === 1;
+      return {
+        stream: (async function* () {
+          if (first) {
+            await firstGate;
+            if (outcome === "stream-error") throw new Error("stream failed");
+            return;
+          }
+          secondActive = true;
+          await abortGate;
+          // Let every pending abort reach the server before B completes.
+          await new Promise<void>((resolve) => setTimeout(resolve, 0));
+          if (secondInterrupted) throw new Error("B interrupted by A's late abort");
+          yield { type: "session.idle", properties: { sessionID: "session-1" } };
+          secondActive = false;
+        })(),
+      } as never;
+    });
+    const runtime = embeddedRuntime(setup.client, {
+      embedded: false, baseUrl: "http://127.0.0.1:4096",
+    });
+    const collect = async (requestId: string, signal: AbortSignal) => {
+      const events: RuntimeEvent[] = [];
+      for await (const event of runtime.run({
+        requestId, conversationId: "shared-conversation", prompt: requestId,
+        workMode: "execute",
+      }, signal)) events.push(event);
+      return events;
+    };
+    const first = collect("A", controller.signal).catch((error: unknown) => error);
+    let second: Promise<unknown> | undefined;
+    try {
+      await vi.waitFor(() => expect(setup.session.promptAsync).toHaveBeenCalledTimes(1));
+      if (outcome === "cancel") controller.abort(new Error("A cancelled"));
+      endFirst();
+      second = collect("B", new AbortController().signal).catch((error: unknown) => error);
+      await vi.waitFor(() => expect(setup.session.abort).toHaveBeenCalled());
+      await new Promise<void>((resolve) => setTimeout(resolve, 20));
+      const promptsBeforeAbort = vi.mocked(setup.session.promptAsync).mock.calls.length;
+      releaseAbort();
+      expect(await first).toMatchObject({ message: outcome === "cancel" ? "A cancelled" : "stream failed" });
+      expect(await second).toEqual(expect.arrayContaining([expect.objectContaining({ type: "done" })]));
+      expect(promptsBeforeAbort).toBe(1);
+      expect(setup.session.abort).toHaveBeenCalledOnce();
+      expect(setup.session.create).toHaveBeenCalledOnce();
+      expect(setup.session.promptAsync).toHaveBeenCalledTimes(2);
+    } finally {
+      endFirst();
+      releaseAbort();
+      await Promise.all([first, second]);
+      await runtime.dispose();
+    }
+  });
+
+  it.each(["never-returns", "rejects", "error", "false"])("replaces the session after abort %s and lets the next request finish", async (outcome) => {
+    const setup = runClient([]);
+    const controller = new AbortController();
+    let abortSignal: AbortSignal | undefined;
+    let abortedAtStart: boolean | undefined;
+    vi.mocked(setup.session.abort).mockImplementation(async (_input, options) => {
+      abortSignal = options?.signal ?? undefined;
+      abortedAtStart = abortSignal?.aborted;
+      if (outcome === "never-returns") return new Promise<never>(() => undefined);
+      if (outcome === "rejects") throw new Error("abort failed");
+      return { data: false, error: outcome === "error" ? { message: "abort failed" } : undefined } as never;
+    });
+    vi.mocked(setup.session.create)
+      .mockResolvedValueOnce({ data: { id: "session-1" } } as never)
+      .mockResolvedValueOnce({ data: { id: "session-2" } } as never);
+    let subscriptions = 0;
+    vi.mocked(setup.event.subscribe).mockImplementation(async () => {
+      const first = ++subscriptions === 1;
+      return {
+        stream: (async function* () {
+          if (first) {
+            controller.abort(new Error("A cancelled"));
+            return;
+          }
+          yield { type: "session.idle", properties: { sessionID: "session-2" } };
+        })(),
+      } as never;
+    });
+    const runtime = embeddedRuntime(setup.client, {
+      embedded: false, baseUrl: "http://127.0.0.1:4096",
+    }, { controlRequestTimeoutMs: 20 });
+    const collect = async (requestId: string, signal: AbortSignal) => {
+      const events: RuntimeEvent[] = [];
+      for await (const event of runtime.run({
+        requestId, conversationId: "shared-conversation", prompt: requestId,
+        workMode: "execute",
+      }, signal)) events.push(event);
+      return events;
+    };
+    try {
+      let settled = false;
+      const first = collect("A", controller.signal).catch((error: unknown) => error)
+        .finally(() => { settled = true; });
+      await vi.waitFor(() => expect(settled).toBe(true));
+      expect(await first).toMatchObject({ message: "A cancelled" });
+      expect(abortedAtStart).toBe(false);
+      expect(abortSignal?.aborted).toBe(outcome === "never-returns");
+      expect(setup.session.abort).toHaveBeenCalledOnce();
+      await expect(collect("B", new AbortController().signal)).resolves.toEqual(
+        expect.arrayContaining([expect.objectContaining({ type: "done", sessionId: "session-2" })]),
+      );
+      expect(setup.session.create).toHaveBeenCalledTimes(2);
+      expect(setup.session.promptAsync).toHaveBeenLastCalledWith(
+        expect.objectContaining({ sessionID: "session-2" }),
+        expect.objectContaining({ signal: expect.any(AbortSignal) }),
+      );
+    } finally {
+      await runtime.dispose();
+    }
+  });
+
   it("loads assigned Skills before prompting", async () => {
     const child = fakeChild();
     const promptAsync = vi.fn().mockResolvedValue({ error: undefined });
@@ -4046,10 +4177,10 @@ describe("OpenCodeRuntime embedded permission mediation", () => {
     await expect(collectRun(runtime, "execute")).rejects.toThrow(
       "OpenCode 权限回复失败",
     );
-    expect(session.abort).toHaveBeenCalledWith({
-      sessionID: "session-1",
-      directory: process.cwd(),
-    });
+    expect(session.abort).toHaveBeenCalledWith(
+      { sessionID: "session-1", directory: process.cwd() },
+      { signal: expect.any(AbortSignal) },
+    );
     await runtime.dispose();
   });
 

src/main/agent/opencode-runtime.ts

@@ -2378,13 +2378,26 @@ export class OpenCodeRuntime implements AgentRuntime {
         AbortSignal.any([signal, subscriptionController.signal]),
       );
 
+      let sessionAbort: Promise<void> | undefined;
       const abortSession = (): void => {
-        void client.session
-          .abort({
-            sessionID: sessionId,
-            directory,
+        sessionAbort ??= this.controlRequest(
+          "取消会话",
+          (controlSignal) => client.session.abort(
+            { sessionID: sessionId, directory },
+            { signal: controlSignal },
+          ),
+        )
+          .then((response) => {
+            if (response.error || response.data !== true) {
+              throw new Error("OpenCode 会话取消失败");
+            }
           })
-          .catch(() => undefined);
+          .catch(() => {
+            // An unconfirmed abort must never target a reused session.
+            if (this.sessions.get(request.conversationId)?.id === sessionId) {
+              this.sessions.delete(request.conversationId);
+            }
+          });
       };
       signal.addEventListener("abort", abortSession, { once: true });
 
@@ -2994,6 +3007,8 @@ export class OpenCodeRuntime implements AgentRuntime {
         throw error;
       } finally {
         signal.removeEventListener("abort", abortSession);
+        // Keep the conversation locked until its session-wide abort settles.
+        await sessionAbort;
         for (const questionId of reportedQuestionIds.values()) {
           this.pendingQuestions.delete(questionId);
         }

src/renderer/src/App.test.tsx

@@ -6053,6 +6053,11 @@ describe("App", () => {
       queueItemId: secondItem.id,
       prompt: secondItem.label,
     });
+    act(() => {
+      agentListener?.({ requestId: run.mock.calls[1]![0].requestId, type: "done" });
+    });
+    await waitFor(() => expect(screen.queryByLabelText("停止生成")).not.toBeInTheDocument());
+    expect(screen.getByLabelText("发送")).toBeInTheDocument();
   });
 
   it("restores a queued message when Agent preflight rejects it", async () => {
@@ -7779,6 +7784,27 @@ describe("App", () => {
       ],
     });
 
+    it.each(["complete", "error"] as const)("does not revive a locally %s message from a newer conversation's stale streaming snapshot", async (state) => {
+      const started = await startStreamingRun();
+      act(() => {
+        agentListener?.({ requestId: started.requestId, type: "text", delta: "Retained terminal output" });
+        agentListener?.(state === "complete"
+          ? { requestId: started.requestId, type: "done" }
+          : { requestId: started.requestId, type: "error", status: "cancelled", message: "Cancelled response" });
+      });
+      await waitFor(() => expect(screen.queryByLabelText("停止生成")).not.toBeInTheDocument());
+      const snapshot = persistedSnapshot(started, { state: "streaming", content: "Stale partial output" });
+      snapshot.updatedAt = Date.now() + 10_000;
+      vi.mocked(api.conversations.list).mockResolvedValue([snapshot]);
+      const callsBeforeRefresh = vi.mocked(api.conversations.list).mock.calls.length;
+      await act(async () => { conversationsChangedListener?.(); });
+      await waitFor(() => expect(api.conversations.list).toHaveBeenCalledTimes(callsBeforeRefresh + 1));
+      expect(screen.queryByLabelText("停止生成")).not.toBeInTheDocument();
+      expect(screen.getByText("Retained terminal output")).toBeInTheDocument();
+      expect(screen.queryByText("Stale partial output")).not.toBeInTheDocument();
+      expect(screen.getByLabelText("发送")).toBeInTheDocument();
+    });
+
     it("converges a streaming message onto the persisted completed state after reconnecting", async () => {
       const started = await startStreamingRun();
       expect(conversationsChangedListener).toBeDefined();

src/renderer/src/App.tsx

@@ -1546,6 +1546,11 @@ function mergePersistedConversations(
       ...conversation.messages.map((persistedMessage) => {
         const localMessage = localMessageById.get(persistedMessage.id);
         const message = mergeMessageImageState(persistedMessage, persistedMessage, localMessage);
+        // A newer conversation timestamp can still contain an older message
+        // snapshot while a terminal event is being persisted.
+        if (localMessage && localMessage.state !== "streaming" && message.state === "streaming") {
+          return mergeMessageImageState(localMessage, persistedMessage, localMessage);
+        }
         if (!conversation.activeRequest && local.activeRequest?.messageId === message.id) {
           return { ...message, pendingQuestions: undefined };
         }