diff --git a/docs/features/assistant-workbar/progress.md b/docs/features/assistant-workbar/progress.md index 1e9e25c9..a6f85d14 100644 --- a/docs/features/assistant-workbar/progress.md +++ b/docs/features/assistant-workbar/progress.md @@ -46,9 +46,13 @@ ## 2026-09-08 OpenCode 复用与原生控件恢复 -- `OpenCodeRuntime` 复用缓存的 embedded server 前执行有超时的健康检查;健康则保留 - client,检查失败、抛错或超时则清除 client、会话映射、会话初始化及待答问题,关闭旧 - server 后重建。并发调用共享检查与重建,调用方取消只结束自身等待。 +- `OpenCodeRuntime` 仅在没有活动请求时探测缓存的 embedded server;活动对话或 + Compact 期间直接复用服务,避免新会话的一次短探测失败关闭其他会话正在使用的进程。 + 空闲探测失败时仍清除旧 client 和会话映射并重建;并发调用共享检查与重建。 + 进程退出或服务关闭会中止绑定它的对话与 Compact,事件流重连等待也可立即取消, + 请求明确报错,不自动重放已执行命令。本次修复后的真实模型验证共 6 次:正常并发、 + 已复现场景下的共享服务保护均正常完成;主动结束测试进程后旧请求明确失败,新请求 + 在替换服务上完成。此前“新会话完成、旧会话持续等待”的故障不再出现。 - 请求级 MCP 注册与断开通过实例内 `mutateMcp` 队列串行执行。仅实际发出 `mcp.add` 的名称进入清理列表;注册尚在排队时取消不追加断开。断开的 1 秒超时从出队后开始。 - `App` 按本机/远程、连接方式、项目和 Runtime 选择组成 scope 缓存原生清单,用请求 diff --git a/src/main/agent/opencode-runtime.test.ts b/src/main/agent/opencode-runtime.test.ts index aed8b011..c5422aa8 100644 --- a/src/main/agent/opencode-runtime.test.ts +++ b/src/main/agent/opencode-runtime.test.ts @@ -1227,6 +1227,179 @@ describe("OpenCodeRuntime embedded launcher", () => { } }); + it("does not replace a shared server while another conversation is running", async () => { + const setup = runClient([ + { type: "session.idle", properties: { sessionID: "session-1" } }, + ]); + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + vi.mocked(setup.event.subscribe).mockImplementationOnce(async () => ({ + stream: (async function* () { + await gate; + yield { type: "session.idle", properties: { sessionID: "session-1" } }; + })(), + }) as never); + const child = fakeChild(); + const replacement = fakeChild(43); + const { deps, checkServerHealth, spawnMock } = dependencies(child, { + createClient: () => setup.client, + }); + spawnMock.mockReturnValueOnce(child).mockReturnValue(replacement); + const runtime = new OpenCodeRuntime(options(), deps); + const first = collectRun(runtime); + try { + await vi.waitFor(() => expect(setup.session.promptAsync).toHaveBeenCalledOnce()); + checkServerHealth.mockResolvedValueOnce(false); + const peerEvents: RuntimeEvent[] = []; + for await (const event of runtime.run( + { + requestId: "peer", conversationId: "peer-conversation", + prompt: "test", workMode: "execute", + }, + new AbortController().signal, + )) peerEvents.push(event); + await expect(runtime.testConnection()).resolves.toMatchObject({ available: true }); + expect(peerEvents.at(-1)).toMatchObject({ type: "done" }); + expect(checkServerHealth).toHaveBeenCalledOnce(); + expect(child.kill).not.toHaveBeenCalled(); + expect(spawnMock).toHaveBeenCalledOnce(); + release(); + await expect(first).resolves.toContainEqual(expect.objectContaining({ type: "done" })); + await expect(runtime.testConnection()).resolves.toMatchObject({ available: true }); + expect(child.kill).toHaveBeenCalledOnce(); + expect(spawnMock).toHaveBeenCalledTimes(2); + } finally { + release(); + await first.catch(() => undefined); + await runtime.dispose(); + } + }); + + it.each(["exit", "dispose"])( + "settles an active request when its embedded server ends through %s", + async (ending) => { + const setup = runClient([]); + vi.mocked(setup.event.subscribe).mockImplementation(async (_input, options) => ({ + stream: (async function* () { + const signal = options!.signal!; + await new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + yield { type: "session.idle", properties: { sessionID: "session-1" } }; + })(), + }) as never); + const child = fakeChild(); + const { deps } = dependencies(child, { createClient: () => setup.client }); + const runtime = new OpenCodeRuntime(options(), deps); + const controller = new AbortController(); + let settled = false; + const events: RuntimeEvent[] = []; + const result = (async () => { + try { + for await (const event of runtime.run( + { requestId: "active", conversationId: "active", prompt: "test", workMode: "execute" }, + controller.signal, + )) events.push(event); + } catch (error) { + return error; + } finally { + settled = true; + } + })(); + try { + await vi.waitFor(() => expect(setup.session.promptAsync).toHaveBeenCalledOnce()); + if (ending === "exit") closeChild(child, 1); + else await runtime.dispose(); + await vi.waitFor(() => expect(settled).toBe(true), { timeout: 300 }); + await expect(result).resolves.toMatchObject({ + message: expect.stringMatching(/OpenCode 服务.*(?:退出|关闭)/u), + }); + expect(controller.signal.aborted).toBe(false); + expect(events.some((event) => event.type === "done")).toBe(false); + } finally { + controller.abort(); + await result; + await runtime.dispose(); + } + }, + ); + + it("cancels an event-stream retry delay when its server exits", async () => { + const setup = runClient([]); + let retrying = false; + vi.mocked(setup.event.subscribe).mockImplementation(async (_input, options) => ({ + stream: (async function* () { + retrying = true; + await (options as { sseSleepFn: (milliseconds: number) => Promise }) + .sseSleepFn(30_000); + yield { type: "session.idle", properties: { sessionID: "session-1" } }; + })(), + }) as never); + const child = fakeChild(); + const { deps } = dependencies(child, { createClient: () => setup.client }); + const runtime = new OpenCodeRuntime(options(), deps); + const result = collectRun(runtime).catch((error: unknown) => error); + try { + await vi.waitFor(() => expect(retrying).toBe(true)); + closeChild(child, 1); + await expect(result).resolves.toMatchObject({ + message: expect.stringContaining("OpenCode 服务已退出"), + }); + } finally { + await runtime.dispose(); + } + }); + + it.each(["complete", "exit"])( + "protects an active Compact until it ends through %s", + async (ending) => { + const setup = runClient([ + { type: "session.idle", properties: { sessionID: "session-1" } }, + ]); + let release!: () => void; + const summarize = vi.fn((_input, options: { signal: AbortSignal }) => + new Promise((resolve, reject) => { + release = () => resolve({ data: true }); + options.signal.addEventListener("abort", () => reject(options.signal.reason), { once: true }); + })); + Object.assign(setup.client, { + v2: { session: { context: vi.fn().mockResolvedValue({ + data: { data: [{ type: "assistant", model: { providerID: "test", id: "test" } }] }, + }) } }, + }); + Object.assign(setup.client.session, { summarize }); + const child = fakeChild(); + const { deps, checkServerHealth } = dependencies(child, { createClient: () => setup.client }); + const runtime = new OpenCodeRuntime(options(), deps); + try { + await collectRun(runtime); + const compact = runtime.compactConversation({ + requestId: "compact", conversationId: "conversation-1", + runtimeSelection: { provider: "opencode" }, history: [], historyMessageIds: [], + }, new AbortController().signal).catch((error: unknown) => error); + await vi.waitFor(() => expect(summarize).toHaveBeenCalledOnce()); + const probes = checkServerHealth.mock.calls.length; + await expect(runtime.testConnection()).resolves.toMatchObject({ available: true }); + expect(checkServerHealth).toHaveBeenCalledTimes(probes); + expect(child.kill).not.toHaveBeenCalled(); + if (ending === "exit") { + closeChild(child, 1); + await expect(compact).resolves.toMatchObject({ + message: expect.stringContaining("OpenCode 服务已退出"), + }); + } else { + release(); + await expect(compact).resolves.toMatchObject({ result: { compacted: true } }); + await runtime.testConnection(); + expect(checkServerHealth).toHaveBeenCalledTimes(probes + 1); + } + } finally { + release?.(); + await runtime.dispose(); + } + }, + ); + it.each(["unhealthy", "rejected", "hung"])( "replaces a %s cached server once for concurrent callers", async (failure) => { diff --git a/src/main/agent/opencode-runtime.ts b/src/main/agent/opencode-runtime.ts index 3b9c3a4d..eb0a2c6b 100644 --- a/src/main/agent/opencode-runtime.ts +++ b/src/main/agent/opencode-runtime.ts @@ -20,6 +20,7 @@ import { } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join, resolve } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; import { parse as parseYaml, stringify as stringifyYaml } from "yaml"; import type { AgentQuestionAnswer, @@ -168,6 +169,8 @@ type OpenCodeProviderDescriptor = { type OpenCodeServer = { url: string; authorization: string; + signal: AbortSignal; + activeRequests: number; close: () => Promise; }; @@ -287,6 +290,14 @@ function executionDeadlineLabel(milliseconds: number): string { : `${milliseconds} 毫秒`; } +function eventSubscriptionOptions(signal: AbortSignal) { + return { + signal, + sseSleepFn: (milliseconds: number) => + delay(milliseconds, undefined, { signal }), + }; +} + function awaitWithAbort( operation: Promise, signal: AbortSignal, @@ -1171,10 +1182,24 @@ export class OpenCodeRuntime implements AgentRuntime { } stdout?.resume(); stderr?.resume(); + const lifetime = new AbortController(); + child.once("close", (code: number | null) => { + lifetime.abort(new Error( + `OpenCode 服务已退出(code ${code ?? "unknown"}),请重试当前请求`, + )); + }); + child.once("error", () => { + lifetime.abort(new Error( + "OpenCode 服务发生错误并已关闭,请重试当前请求", + )); + }); resolveServer({ url, authorization, + signal: lifetime.signal, + activeRequests: 0, close: async () => { + lifetime.abort(new Error("OpenCode 服务已关闭,请重试当前请求")); try { const exited = this.waitForExit(child); this.terminate(child); @@ -1315,6 +1340,10 @@ export class OpenCodeRuntime implements AgentRuntime { signal?: AbortSignal, ): Promise { if (this.client && this.server) { + // A short probe must not replace a server still serving another request. + if (this.server.activeRequests > 0 && !this.server.signal.aborted) { + return this.client; + } const probeSignal = AbortSignal.any([ ...(signal ? [signal] : []), AbortSignal.timeout(STARTUP_PROBE_TIMEOUT_MS), @@ -2027,6 +2056,27 @@ export class OpenCodeRuntime implements AgentRuntime { throw new Error("当前模型连接未启用图像输入"); } const client = await this.getClient(signal); + const server = this.server; + const requestSignal = server + ? AbortSignal.any([signal, server.signal]) + : signal; + if (server) server.activeRequests += 1; + try { + requestSignal.throwIfAborted(); + yield* this.runWithClient(request, requestSignal, client); + } catch (error) { + requestSignal.throwIfAborted(); + throw error; + } finally { + if (server) server.activeRequests -= 1; + } + } + + private async *runWithClient( + request: AgentExecutionRequest, + signal: AbortSignal, + client: OpencodeClient, + ): AsyncGenerator { const directory = this.options.defaultWorkspace; const runtimeControl = request.runtimeControl?.provider === "opencode" @@ -2290,7 +2340,10 @@ export class OpenCodeRuntime implements AgentRuntime { const subscription = await this.controlRequest( "订阅事件流", (controlSignal) => - client.event.subscribe({ directory }, { signal: controlSignal }), + client.event.subscribe( + { directory }, + eventSubscriptionOptions(controlSignal), + ), AbortSignal.any([signal, subscriptionController.signal]), ); @@ -2942,10 +2995,11 @@ export class OpenCodeRuntime implements AgentRuntime { configuredTimeout, ); deadlineTimer?.unref?.(); - const executionSignal = deadline + let executionSignal = deadline ? AbortSignal.any([signal, deadline.signal]) : signal; let releaseConversation: (() => void) | undefined; + let server: OpenCodeServer | undefined; try { releaseConversation = await this.acquireConversationRun( request.conversationId, @@ -2954,6 +3008,13 @@ export class OpenCodeRuntime implements AgentRuntime { const client = this.sessions.has(request.conversationId) ? await this.getClient(executionSignal) : undefined; + if (client) { + server = this.server; + if (server) { + server.activeRequests += 1; + executionSignal = AbortSignal.any([executionSignal, server.signal]); + } + } const sessionId = this.sessions.get(request.conversationId); if (!sessionId || !client) { return { @@ -3012,7 +3073,7 @@ export class OpenCodeRuntime implements AgentRuntime { (controlSignal) => client.event.subscribe( { directory: this.options.defaultWorkspace }, - { signal: controlSignal }, + eventSubscriptionOptions(controlSignal), ), subscriptionSignal, ); @@ -3094,11 +3155,13 @@ export class OpenCodeRuntime implements AgentRuntime { if (deadline?.signal.aborted && !signal.aborted) { throw deadline.signal.reason; } + executionSignal.throwIfAborted(); throw error; } finally { if (deadlineTimer) { clearTimeout(deadlineTimer); } + if (server) server.activeRequests -= 1; releaseConversation?.(); } }