fix: prevent parallel OpenCode conversations from hanging

Starting another conversation could replace a busy local OpenCode server
after a brief health-check failure, leaving the existing conversation
reconnecting to a dead endpoint.

Keep active conversations and native Compact operations on their current
server. Propagate server exit and shutdown to bound requests and cancel
event-stream retry waits so failures are explicit instead of hanging.
Do not automatically replay commands.

Release note: 修复同时使用多个本机 OpenCode 会话时旧会话可能一直等待的问题。服务异常退出时会明确提示失败,不会自动重复执行命令。
This commit is contained in:
mesalogo
2026-09-08 21:10:35 +08:00
parent 28de0c45ba
commit c1f4fd0580
3 changed files with 246 additions and 6 deletions
+7 -3
View File
@@ -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 缓存原生清单,用请求
+173
View File
@@ -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<void>((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<void>((_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<void> })
.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) => {
+66 -3
View File
@@ -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<void>;
};
@@ -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<T>(
operation: Promise<T>,
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<OpencodeClient> {
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<RuntimeEvent, void, void> {
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?.();
}
}