Fix OpenCode custom command hangs (#1063)

* Fix OpenCode slash command turns

* ci: install opencode in server-tests so real OpenCode e2e runs

Server tests previously only installed the Claude Code CLI, so every
opencode-*.real.e2e.test.ts self-skipped on CI. The custom-command and
turn-lifecycle real tests cover the OpenCode session SSE rewrite, so
add opencode-ai globally and let them run. Big Pickle is free so no
provider secrets needed.

* Fix OpenCode unit tests exposed by enabling opencode in CI

server-tests previously skipped every OpenCodeAgentClient test because
opencode-ai was not installed. Now that CI runs them, four real issues
surface:

- Drop unused mkdirSync import (was used on a code path 1812b1489
  refactored away on main; the rebase merged imports without rechecking
  usage). Fixes the lint job.
- Fix slash-command-timeout test passing four ctor args instead of three,
  which silently dropped the TestOpenCodeRuntime and ran against the real
  opencode runtime.
- Relax the listModels contextWindowMaxTokens assertion. Some providers
  routed through OpenCode (OpenAI in particular) don't expose a numeric
  context window; assert the type only when the field is present.
- Swallow late notifySubscribers calls once close() flips the closed
  flag and clear subscribers up front. The session-lifetime SSE stream
  introduced by the prior commit means a session.error arriving after
  the owning test has moved on can rejectCompletion on a deferred no
  one is awaiting, surfacing as an unhandled rejection in a downstream
  test file.
This commit is contained in:
Mohamed Boudra
2026-05-17 19:51:20 +08:00
committed by GitHub
parent b3c272f720
commit 00e7ac9ec8
13 changed files with 394 additions and 173 deletions
+2 -2
View File
@@ -104,8 +104,8 @@ jobs:
- name: Install dependencies - name: Install dependencies
run: npm ci run: npm ci
- name: Install Claude Code CLI for provider tests - name: Install agent CLIs for provider tests
run: npm install -g @anthropic-ai/claude-code run: npm install -g @anthropic-ai/claude-code opencode-ai
- name: Build highlight dependency - name: Build highlight dependency
run: npm run build --workspace=@getpaseo/highlight run: npm run build --workspace=@getpaseo/highlight
@@ -1,8 +1,12 @@
import { beforeAll, beforeEach, describe, expect, test } from "vitest"; import { afterAll, beforeAll, beforeEach, describe, expect, test } from "vitest";
import pino from "pino"; import pino from "pino";
import { isCommandAvailable } from "../../../utils/executable.js"; import { isCommandAvailable } from "../../../utils/executable.js";
import { OpenCodeAgentClient } from "./opencode-agent.js"; import { OpenCodeAgentClient } from "./opencode-agent.js";
import { OpenCodeServerManager } from "./opencode/server-manager.js";
const BIG_PICKLE_MODEL = "opencode/big-pickle";
const logger = pino({ level: "silent" });
describe("opencode agent commands contract (real)", () => { describe("opencode agent commands contract (real)", () => {
let canRun = false; let canRun = false;
@@ -17,13 +21,18 @@ describe("opencode agent commands contract (real)", () => {
} }
}); });
afterAll(async () => {
await OpenCodeServerManager.getInstance(logger).shutdown();
});
test("lists slash commands with the expected contract", async () => { test("lists slash commands with the expected contract", async () => {
expect(await isCommandAvailable("opencode")).toBe(true); expect(await isCommandAvailable("opencode")).toBe(true);
const client = new OpenCodeAgentClient(pino({ level: "silent" })); const client = new OpenCodeAgentClient(logger);
const session = await client.createSession({ const session = await client.createSession({
provider: "opencode", provider: "opencode",
cwd: process.cwd(), cwd: process.cwd(),
model: BIG_PICKLE_MODEL,
modeId: "plan", modeId: "plan",
}); });
@@ -50,10 +59,11 @@ describe("opencode agent commands contract (real)", () => {
test("executes a slash command without arguments", async () => { test("executes a slash command without arguments", async () => {
expect(await isCommandAvailable("opencode")).toBe(true); expect(await isCommandAvailable("opencode")).toBe(true);
const client = new OpenCodeAgentClient(pino({ level: "silent" })); const client = new OpenCodeAgentClient(logger);
const session = await client.createSession({ const session = await client.createSession({
provider: "opencode", provider: "opencode",
cwd: process.cwd(), cwd: process.cwd(),
model: BIG_PICKLE_MODEL,
modeId: "plan", modeId: "plan",
}); });
@@ -0,0 +1,107 @@
import { existsSync } from "node:fs";
import { mkdir, mkdtemp, rm, rmdir, writeFile } from "node:fs/promises";
import { homedir, tmpdir } from "node:os";
import path from "node:path";
import { beforeAll, beforeEach, describe, expect, test } from "vitest";
import pino from "pino";
import { isCommandAvailable } from "../../../utils/executable.js";
import { createDaemonTestContext, type DaemonTestContext } from "../../test-utils/index.js";
import { OpenCodeAgentClient } from "./opencode-agent.js";
import { OpenCodeServerManager } from "./opencode/server-manager.js";
const COMMAND_NAME = "paseo-issue-903-big-pickle";
const COMMAND_FILE_NAME = `${COMMAND_NAME}.md`;
const BIG_PICKLE_MODEL = "opencode/big-pickle";
const EXPECTED_RESPONSE = "PASEO_ISSUE_903_BIG_PICKLE_OK";
describe("opencode custom command Big Pickle E2E (real)", () => {
let canRun = false;
beforeAll(async () => {
canRun = await isCommandAvailable("opencode");
});
beforeEach((context) => {
if (!canRun) {
context.skip();
}
});
test("executes a global custom command through Paseo using Big Pickle", async () => {
const commandDir = path.join(homedir(), ".config", "opencode", "command");
const commandFile = path.join(commandDir, COMMAND_FILE_NAME);
const commandDirExisted = existsSync(commandDir);
if (existsSync(commandFile)) {
throw new Error(`Refusing to overwrite existing OpenCode command file: ${commandFile}`);
}
const projectDir = await mkdtemp(path.join(tmpdir(), "paseo-opencode-big-pickle-"));
const logger = pino({ level: "silent" });
let ctx: DaemonTestContext | undefined;
try {
await mkdir(commandDir, { recursive: true });
await writeFile(
commandFile,
[
"---",
"description: Paseo issue 903 Big Pickle custom command",
"agent: build",
"---",
"",
"Reply exactly with this token and nothing else:",
EXPECTED_RESPONSE,
"",
].join("\n"),
);
ctx = await createDaemonTestContext({
logger,
agentClients: {
opencode: new OpenCodeAgentClient(logger),
},
});
const agent = await ctx.client.createAgent({
provider: "opencode",
cwd: projectDir,
model: BIG_PICKLE_MODEL,
modeId: "plan",
title: "OpenCode issue 903 Big Pickle custom command",
});
expect(agent.provider).toBe("opencode");
expect(agent.model).toBe(BIG_PICKLE_MODEL);
expect(agent.status).toBe("idle");
const commands = await ctx.client.listCommands(agent.id);
expect(commands.error).toBeNull();
expect(commands.commands).toEqual(
expect.arrayContaining([
expect.objectContaining({
name: COMMAND_NAME,
description: "Paseo issue 903 Big Pickle custom command",
}),
]),
);
await ctx.client.sendMessage(agent.id, `/${COMMAND_NAME}`);
const state = await ctx.client.waitForFinish(agent.id, 90_000);
expect(state.status).toBe("idle");
expect(state.error).toBeNull();
expect(state.final?.status).toBe("idle");
expect(state.lastMessage).toContain(EXPECTED_RESPONSE);
} finally {
await ctx?.cleanup();
await OpenCodeServerManager.getInstance(logger).shutdown();
await rm(commandFile, { force: true });
if (!commandDirExisted) {
await rmdir(commandDir).catch(() => undefined);
}
await rm(projectDir, { recursive: true, force: true });
}
}, 120_000);
});
@@ -4,7 +4,6 @@ import { createTestLogger } from "../../../test-utils/test-logger.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js";
import { OpenCodeAgentClient } from "./opencode-agent.js"; import { OpenCodeAgentClient } from "./opencode-agent.js";
import { import {
createEventStream,
idleEvent, idleEvent,
TestOpenCodeClient, TestOpenCodeClient,
TestOpenCodeRuntime, TestOpenCodeRuntime,
@@ -19,7 +18,7 @@ function mockOpenCodeClient(options: MockOpenCodeClientOptions = {}) {
const runtime = new TestOpenCodeRuntime(); const runtime = new TestOpenCodeRuntime();
const openCodeClient = new TestOpenCodeClient(); const openCodeClient = new TestOpenCodeClient();
openCodeClient.appAgentsResponse = { data: options.agents ?? [] }; openCodeClient.appAgentsResponse = { data: options.agents ?? [] };
openCodeClient.eventStream = createEventStream(options.events ?? [idleEvent()]); openCodeClient.sessionPromptAsyncEvents = options.events ?? [idleEvent()];
runtime.enqueueClient(openCodeClient); runtime.enqueueClient(openCodeClient);
return { openCodeClient, runtime }; return { openCodeClient, runtime };
@@ -3,6 +3,7 @@ import { describe, expect, test } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js"; import { createTestLogger } from "../../../test-utils/test-logger.js";
import { OpenCodeAgentClient } from "./opencode-agent.js"; import { OpenCodeAgentClient } from "./opencode-agent.js";
import { import {
idleEvent,
TestOpenCodeClient, TestOpenCodeClient,
TestOpenCodeRuntime, TestOpenCodeRuntime,
} from "./opencode/test-utils/test-opencode-runtime.js"; } from "./opencode/test-utils/test-opencode-runtime.js";
@@ -93,6 +94,41 @@ describe("OpenCodeAgentSession slash command timeout handling", () => {
usage: undefined, usage: undefined,
}); });
}); });
test("leaves successful slash command turns open until OpenCode emits idle", async () => {
const runtime = new TestOpenCodeRuntime();
const openCodeClient = createOpenCodeClientWithConnectedProvider();
openCodeClient.sessionCommandEvents = [];
openCodeClient.commandListResponse = {
data: [{ name: "help", description: "Show help", hints: [] }],
};
runtime.enqueueClient(openCodeClient);
const client = new OpenCodeAgentClient(createTestLogger(), undefined, { runtime });
const session = await client.createSession({ provider: "opencode", cwd: "/tmp" });
const runPromise = session.run("/help");
await nextTick();
await nextTick();
expect(openCodeClient.calls.sessionCommand).toHaveLength(1);
let settled = false;
void runPromise.then(() => {
settled = true;
return undefined;
});
await nextTick();
expect(settled).toBe(false);
openCodeClient.emitEvent(idleEvent());
await expect(runPromise).resolves.toMatchObject({
sessionId: "session-1",
finalText: "",
timeline: [],
usage: undefined,
});
});
}); });
function createOpenCodeClientWithConnectedProvider(): TestOpenCodeClient { function createOpenCodeClientWithConnectedProvider(): TestOpenCodeClient {
@@ -105,3 +141,7 @@ function createOpenCodeClientWithConnectedProvider(): TestOpenCodeClient {
}; };
return openCodeClient; return openCodeClient;
} }
function nextTick(): Promise<void> {
return new Promise((resolve) => setImmediate(resolve));
}
@@ -1,4 +1,4 @@
import { beforeAll, describe, expect, test, vi } from "vitest"; import { afterAll, describe, expect, test, vi } from "vitest";
import { existsSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { existsSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
import os from "node:os"; import os from "node:os";
import path from "node:path"; import path from "node:path";
@@ -11,6 +11,7 @@ import {
OpenCodeAgentClient, OpenCodeAgentClient,
translateOpenCodeEvent, translateOpenCodeEvent,
} from "./opencode-agent.js"; } from "./opencode-agent.js";
import { OpenCodeServerManager } from "./opencode/server-manager.js";
import { streamSession } from "./test-utils/session-stream-adapter.js"; import { streamSession } from "./test-utils/session-stream-adapter.js";
import { import {
TestOpenCodeClient, TestOpenCodeClient,
@@ -33,8 +34,7 @@ function tmpCwd(): string {
} }
} }
// Dynamic model selection - will be set in beforeAll const TEST_MODEL = "opencode/big-pickle";
let TEST_MODEL: string | undefined;
interface TurnResult { interface TurnResult {
events: AgentStreamEvent[]; events: AgentStreamEvent[];
@@ -82,14 +82,6 @@ async function collectTurnEvents(iterator: AsyncGenerator<AgentStreamEvent>): Pr
return result; return result;
} }
function createAsyncIterable<T>(items: T[]): AsyncIterable<T> {
return (async function* () {
for (const item of items) {
yield item;
}
})();
}
function isBinaryInstalled(binary: string): boolean { function isBinaryInstalled(binary: string): boolean {
try { try {
const out = execFileSync("which", [binary], { encoding: "utf8" }).trim(); const out = execFileSync("which", [binary], { encoding: "utf8" }).trim();
@@ -109,45 +101,9 @@ const hasOpenCode = isBinaryInstalled("opencode");
model: TEST_MODEL, model: TEST_MODEL,
}); });
beforeAll(async () => { afterAll(async () => {
const startTime = Date.now(); await OpenCodeServerManager.getInstance(logger).shutdown();
logger.info("beforeAll: Starting model selection"); });
const client = new OpenCodeAgentClient(logger);
const models = await client.listModels({ cwd: os.homedir(), force: false });
logger.info(
{ modelCount: models.length, elapsed: Date.now() - startTime },
"beforeAll: Retrieved models",
);
// Prefer cheap models that support tool use (required by OpenCode agents).
// Avoid free-tier OpenRouter models — they often lack tool-use support.
const fastModel = models.find(
(m) =>
m.id.includes("gpt-4.1-nano") ||
m.id.includes("gpt-4.1-mini") ||
m.id.includes("gpt-5-nano") ||
m.id.includes("gpt-5.4-mini") ||
m.id.includes("gpt-4o-mini"),
);
if (fastModel) {
TEST_MODEL = fastModel.id;
} else if (models.length > 0) {
// Fallback to any available model
TEST_MODEL = models[0].id;
} else {
throw new Error(
"No OpenCode models available. Please authenticate with a provider (e.g., set OPENAI_API_KEY).",
);
}
logger.info(
{ model: TEST_MODEL, totalElapsed: Date.now() - startTime },
"beforeAll: Selected OpenCode test model",
);
}, 30_000);
test("creates a session with valid id and provider", async () => { test("creates a session with valid id and provider", async () => {
const cwd = tmpCwd(); const cwd = tmpCwd();
@@ -200,6 +156,7 @@ const hasOpenCode = isBinaryInstalled("opencode");
// HARD ASSERT: At least one model is returned (OpenCode has connected providers) // HARD ASSERT: At least one model is returned (OpenCode has connected providers)
expect(models.length).toBeGreaterThan(0); expect(models.length).toBeGreaterThan(0);
expect(models.some((model) => model.id === TEST_MODEL)).toBe(true);
// HARD ASSERT: Each model has required fields with correct types // HARD ASSERT: Each model has required fields with correct types
for (const model of models) { for (const model of models) {
@@ -214,8 +171,12 @@ const hasOpenCode = isBinaryInstalled("opencode");
expect(model.metadata).toMatchObject({ expect(model.metadata).toMatchObject({
providerId: expect.any(String), providerId: expect.any(String),
modelId: expect.any(String), modelId: expect.any(String),
contextWindowMaxTokens: expect.any(Number),
}); });
// contextWindowMaxTokens is upstream-provided and may be absent for some
// OpenCode-routed providers; assert the type only when present.
if (model.metadata?.contextWindowMaxTokens !== undefined) {
expect(typeof model.metadata.contextWindowMaxTokens).toBe("number");
}
} }
}, 60_000); }, 60_000);
@@ -623,6 +584,7 @@ describe("OpenCode adapter context-window normalization", () => {
describe("OpenCode adapter startTurn error handling", () => { describe("OpenCode adapter startTurn error handling", () => {
test("emits turn_started before live OpenCode timeline items", async () => { test("emits turn_started before live OpenCode timeline items", async () => {
const eventsGate = createTestDeferred<void>();
const globalEvents = [ const globalEvents = [
{ {
payload: { payload: {
@@ -669,10 +631,18 @@ describe("OpenCode adapter startTurn error handling", () => {
]; ];
const fakeClient = { const fakeClient = {
global: { global: {
event: vi.fn().mockResolvedValue({ stream: createAsyncIterable(globalEvents) }), event: vi.fn().mockResolvedValue({
stream: (async function* () {
await eventsGate.promise;
yield* globalEvents;
})(),
}),
}, },
session: { session: {
promptAsync: vi.fn().mockResolvedValue({ data: {}, error: undefined }), promptAsync: vi.fn().mockImplementation(async () => {
eventsGate.resolve();
return { data: {}, error: undefined };
}),
}, },
} as never; } as never;
@@ -698,6 +668,7 @@ describe("OpenCode adapter startTurn error handling", () => {
}); });
test("unwraps OpenCode global event payloads during a turn", async () => { test("unwraps OpenCode global event payloads during a turn", async () => {
const eventsGate = createTestDeferred<void>();
const globalEvents = [ const globalEvents = [
{ {
payload: { payload: {
@@ -760,10 +731,18 @@ describe("OpenCode adapter startTurn error handling", () => {
subscribe: vi.fn(), subscribe: vi.fn(),
}, },
global: { global: {
event: vi.fn().mockResolvedValue({ stream: createAsyncIterable(globalEvents) }), event: vi.fn().mockResolvedValue({
stream: (async function* () {
await eventsGate.promise;
yield* globalEvents;
})(),
}),
}, },
session: { session: {
promptAsync: vi.fn().mockResolvedValue({ data: {}, error: undefined }), promptAsync: vi.fn().mockImplementation(async () => {
eventsGate.resolve();
return { data: {}, error: undefined };
}),
}, },
} as never; } as never;
@@ -790,11 +769,13 @@ describe("OpenCode adapter startTurn error handling", () => {
test("keeps a turn active while OpenCode is retrying", async () => { test("keeps a turn active while OpenCode is retrying", async () => {
vi.useFakeTimers(); vi.useFakeTimers();
const eventsGate = createTestDeferred<void>();
const retryStream: AsyncIterable<unknown> = { const retryStream: AsyncIterable<unknown> = {
[Symbol.asyncIterator]: () => { [Symbol.asyncIterator]: () => {
let emitted = false; let emitted = false;
return { return {
next: async () => { next: async () => {
await eventsGate.promise;
if (!emitted) { if (!emitted) {
emitted = true; emitted = true;
return { return {
@@ -826,7 +807,10 @@ describe("OpenCode adapter startTurn error handling", () => {
session: { session: {
abort: vi.fn().mockResolvedValue({ error: null }), abort: vi.fn().mockResolvedValue({ error: null }),
update: vi.fn().mockResolvedValue({ error: null }), update: vi.fn().mockResolvedValue({ error: null }),
promptAsync: vi.fn().mockResolvedValue({ data: {}, error: undefined }), promptAsync: vi.fn().mockImplementation(async () => {
eventsGate.resolve();
return { data: {}, error: undefined };
}),
}, },
} as never; } as never;
@@ -2267,6 +2267,9 @@ class OpenCodeAgentSession implements AgentSession {
private pendingChildToolPartsBySessionId = new Map<string, OpenCodeToolPartEventPart[]>(); private pendingChildToolPartsBySessionId = new Map<string, OpenCodeToolPartEventPart[]>();
private selectedModelContextWindowMaxTokens: number | undefined; private selectedModelContextWindowMaxTokens: number | undefined;
private releaseServer: (() => void) | null; private releaseServer: (() => void) | null;
private eventStreamAbortController: AbortController | null = null;
private eventStreamReady: Deferred<void> | null = null;
private closed = false;
private readonly persistSession: boolean; private readonly persistSession: boolean;
private deletedFromProvider = false; private deletedFromProvider = false;
constructor( constructor(
@@ -2290,6 +2293,7 @@ class OpenCodeAgentSession implements AgentSession {
this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens( this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens(
config.model, config.model,
); );
this.startEventStream();
} }
get id(): string | null { get id(): string | null {
@@ -2422,22 +2426,17 @@ class OpenCodeAgentSession implements AgentSession {
const effectiveVariant = thinkingOptionId ?? undefined; const effectiveVariant = thinkingOptionId ?? undefined;
const effectiveMode = resolveOpenCodeRuntimeAgentId(this.currentMode); const effectiveMode = resolveOpenCodeRuntimeAgentId(this.currentMode);
try {
await this.ensureEventStreamReady();
} catch (error) {
if (this.abortController === turnAbortController) {
this.abortController = null;
}
throw error;
}
const turnId = this.createTurnId(); const turnId = this.createTurnId();
this.activeForegroundTurnId = turnId; this.activeForegroundTurnId = turnId;
// OpenCode's /event SSE endpoint does NOT replay past events. If we send
// the prompt before our reader is connected, terminal events fired early
// by the server (e.g. session.error / session.idle for invalid model or
// mode) are missed and the turn hangs forever. Wait for the subscription
// to be established before sending anything.
const subscriptionReady = createDeferred<void>();
void this.consumeEventStream(turnId, turnAbortController, subscriptionReady);
try {
await subscriptionReady.promise;
} catch {
// consumeEventStream already finished the turn with the subscription error.
return { turnId };
}
this.notifySubscribers({ type: "turn_started", provider: "opencode" }, turnId); this.notifySubscribers({ type: "turn_started", provider: "opencode" }, turnId);
const slashCommand = await this.resolveSlashCommandInvocation(prompt); const slashCommand = await this.resolveSlashCommandInvocation(prompt);
@@ -2480,12 +2479,8 @@ class OpenCodeAgentSession implements AgentSession {
return { turnId }; return { turnId };
} }
// command() blocks until the server finishes processing. OpenCode's SSE // command() is only dispatch acknowledgement. OpenCode session events are
// endpoint does NOT replay past events, so if the command completes before // the source of truth for when the command turn becomes idle or fails.
// our SSE reader connects, we miss `session.idle` and the turn hangs.
// Handle both success and error in the response handler as a fallback —
// finishForegroundTurn's guard prevents duplicate terminal events if the
// SSE stream already delivered the event.
void this.client.session void this.client.session
.command({ .command({
sessionID: this.sessionId, sessionID: this.sessionId,
@@ -2514,11 +2509,6 @@ class OpenCodeAgentSession implements AgentSession {
{ type: "turn_failed", provider: "opencode", error: errorMsg }, { type: "turn_failed", provider: "opencode", error: errorMsg },
turnId, turnId,
); );
} else {
this.finishForegroundTurn(
{ type: "turn_completed", provider: "opencode", usage: undefined },
turnId,
);
} }
return; return;
}) })
@@ -2608,7 +2598,6 @@ class OpenCodeAgentSession implements AgentSession {
return { turnId }; return { turnId };
} }
subscribe(callback: (event: AgentStreamEvent) => void): () => void { subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback); this.subscribers.add(callback);
return () => { return () => {
@@ -2616,109 +2605,115 @@ class OpenCodeAgentSession implements AgentSession {
}; };
} }
private startEventStream(): void {
void this.ensureEventStreamReady().catch((error) => {
this.logger.warn({ err: error, sessionId: this.sessionId }, "OpenCode event stream failed");
});
}
private ensureEventStreamReady(): Promise<void> {
if (this.eventStreamReady) {
return this.eventStreamReady.promise;
}
const eventStreamAbortController = new AbortController();
const eventStreamReady = createDeferred<void>();
this.eventStreamAbortController = eventStreamAbortController;
this.eventStreamReady = eventStreamReady;
void this.consumeEventStream(eventStreamAbortController, eventStreamReady).finally(() => {
if (this.eventStreamAbortController === eventStreamAbortController) {
this.eventStreamAbortController = null;
this.eventStreamReady = null;
}
});
return eventStreamReady.promise;
}
private async consumeEventStream( private async consumeEventStream(
turnId: string, eventStreamAbortController: AbortController,
turnAbortController: AbortController, eventStreamReady: Deferred<void>,
subscriptionReady: Deferred<void>,
): Promise<void> { ): Promise<void> {
this.traceOpenCode("provider.opencode.subscribe.start", { this.traceOpenCode("provider.opencode.subscribe.start", {
turnId,
sessionId: this.sessionId, sessionId: this.sessionId,
cwd: this.config.cwd, cwd: this.config.cwd,
}); });
let eventStreamReadyResolved = false;
try { try {
const result = await this.client.global.event({ const result = await this.client.global.event({
signal: turnAbortController.signal, signal: eventStreamAbortController.signal,
sseMaxRetryAttempts: 0, sseMaxRetryAttempts: 0,
}); });
eventStreamReadyResolved = true;
this.traceOpenCode("provider.opencode.subscribe.ready", {
sessionId: this.sessionId,
});
eventStreamReady.resolve();
let eventCount = 0; let eventCount = 0;
let subscriptionReadyResolved = false;
for await (const rawEvent of result.stream) { for await (const rawEvent of result.stream) {
eventCount += 1; eventCount += 1;
if (!subscriptionReadyResolved) { await this.consumeOpenCodeStreamEvent({ rawEvent, eventCount });
subscriptionReadyResolved = true;
this.traceOpenCode("provider.opencode.subscribe.ready", {
turnId,
sessionId: this.sessionId,
});
subscriptionReady.resolve();
}
const shouldContinue = await this.consumeOpenCodeStreamEvent({
rawEvent,
eventCount,
turnId,
turnAbortController,
});
if (!shouldContinue) {
return;
}
} }
this.traceOpenCode("provider.opencode.stream.eof", { this.traceOpenCode("provider.opencode.stream.eof", {
turnId,
eventCount, eventCount,
aborted: turnAbortController.signal.aborted, aborted: eventStreamAbortController.signal.aborted,
stillActive: this.activeForegroundTurnId === turnId, activeTurnId: this.activeForegroundTurnId,
}); });
if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { if (!eventStreamAbortController.signal.aborted) {
this.traceOpenCode("provider.opencode.turn.fail_eof", { turnId, eventCount }); if (!eventStreamReadyResolved) {
if (!subscriptionReadyResolved) { eventStreamReady.reject(new Error("OpenCode event stream ended before it became ready"));
subscriptionReady.reject(new Error("OpenCode event stream ended before it became ready")); }
const activeTurnId = this.activeForegroundTurnId;
if (activeTurnId) {
this.traceOpenCode("provider.opencode.turn.fail_eof", {
turnId: activeTurnId,
eventCount,
});
this.finishForegroundTurn(
{
type: "turn_failed",
provider: "opencode",
error: "OpenCode event stream ended before the turn reached a terminal state",
},
activeTurnId,
);
} }
this.finishForegroundTurn(
{
type: "turn_failed",
provider: "opencode",
error: "OpenCode event stream ended before the turn reached a terminal state",
},
turnId,
);
} }
} catch (error) { } catch (error) {
this.traceOpenCode("provider.opencode.subscribe.error", { this.traceOpenCode("provider.opencode.subscribe.error", {
turnId, turnId: this.activeForegroundTurnId ?? undefined,
error: error:
error instanceof Error ? { name: error.name, message: error.message } : String(error), error instanceof Error ? { name: error.name, message: error.message } : String(error),
}); });
subscriptionReady.reject(error); if (!eventStreamReadyResolved) {
if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) { eventStreamReady.reject(error);
}
const activeTurnId = this.activeForegroundTurnId;
if (!eventStreamAbortController.signal.aborted && activeTurnId) {
this.finishForegroundTurn( this.finishForegroundTurn(
{ {
type: "turn_failed", type: "turn_failed",
provider: "opencode", provider: "opencode",
error: toDiagnosticErrorMessage(error), error: toDiagnosticErrorMessage(error),
}, },
turnId, activeTurnId,
); );
} }
} finally {
if (turnAbortController.signal.aborted) {
this.finishForegroundTurn(
{
type: "turn_canceled",
provider: "opencode",
reason: "interrupted",
},
turnId,
);
}
if (this.abortController === turnAbortController && this.activeForegroundTurnId !== turnId) {
this.abortController = null;
}
} }
} }
private async consumeOpenCodeStreamEvent(params: { private async consumeOpenCodeStreamEvent(params: {
rawEvent: unknown; rawEvent: unknown;
eventCount: number; eventCount: number;
turnId: string; }): Promise<void> {
turnAbortController: AbortController; const { rawEvent, eventCount } = params;
}): Promise<boolean> { const turnId = this.activeForegroundTurnId;
const { rawEvent, eventCount, turnId, turnAbortController } = params;
const event = unwrapOpenCodeGlobalEvent(rawEvent); const event = unwrapOpenCodeGlobalEvent(rawEvent);
this.traceOpenCode("provider.opencode.raw_event", { this.traceOpenCode("provider.opencode.raw_event", {
turnId, turnId: turnId ?? undefined,
n: eventCount, n: eventCount,
type: event?.type, type: event?.type,
rawType: readOpenCodeRecord(rawEvent)?.type, rawType: readOpenCodeRecord(rawEvent)?.type,
@@ -2727,16 +2722,15 @@ class OpenCodeAgentSession implements AgentSession {
properties: event?.properties, properties: event?.properties,
}); });
if (!event) { if (!event) {
return true; return;
} }
if (turnAbortController.signal.aborted || this.activeForegroundTurnId !== turnId) { if (!turnId) {
this.traceOpenCode("provider.opencode.event.skip", { this.traceOpenCode("provider.opencode.event.skip", {
turnId,
n: eventCount, n: eventCount,
aborted: turnAbortController.signal.aborted, reason: "no_active_turn",
activeTurnId: this.activeForegroundTurnId, type: event.type,
}); });
return false; return;
} }
const translated = await this.translateEvent(event); const translated = await this.translateEvent(event);
this.traceOpenCode("provider.opencode.parsed_event", { this.traceOpenCode("provider.opencode.parsed_event", {
@@ -2750,7 +2744,7 @@ class OpenCodeAgentSession implements AgentSession {
for (const e of translated) { for (const e of translated) {
if (this.activeForegroundTurnId !== turnId) { if (this.activeForegroundTurnId !== turnId) {
this.traceOpenCode("provider.opencode.parsed_event.skip_active", { turnId, type: e.type }); this.traceOpenCode("provider.opencode.parsed_event.skip_active", { turnId, type: e.type });
return false; return;
} }
if (e.type === "timeline" && e.item.type === "tool_call") { if (e.type === "timeline" && e.item.type === "tool_call") {
this.trackToolCall(e.item); this.trackToolCall(e.item);
@@ -2762,12 +2756,10 @@ class OpenCodeAgentSession implements AgentSession {
type: terminalEvent.type, type: terminalEvent.type,
}); });
this.finishForegroundTurn(terminalEvent, turnId); this.finishForegroundTurn(terminalEvent, turnId);
return false; return;
} }
this.notifySubscribers(e, turnId); this.notifySubscribers(e, turnId);
} }
return true;
} }
private finishForegroundTurn( private finishForegroundTurn(
@@ -2790,8 +2782,6 @@ class OpenCodeAgentSession implements AgentSession {
this.runningToolCalls.clear(); this.runningToolCalls.clear();
} }
this.activeForegroundTurnId = null; this.activeForegroundTurnId = null;
// Abort the SSE connection so the SDK tears down the underlying fetch.
this.abortController?.abort();
this.abortController = null; this.abortController = null;
this.notifySubscribers(event, turnId); this.notifySubscribers(event, turnId);
} }
@@ -2833,6 +2823,9 @@ class OpenCodeAgentSession implements AgentSession {
} }
private notifySubscribers(event: AgentStreamEvent, turnIdOverride?: string): void { private notifySubscribers(event: AgentStreamEvent, turnIdOverride?: string): void {
if (this.closed) {
return;
}
const turnId = turnIdOverride ?? this.activeForegroundTurnId; const turnId = turnIdOverride ?? this.activeForegroundTurnId;
const tagged = turnId ? { ...event, turnId } : event; const tagged = turnId ? { ...event, turnId } : event;
this.traceOpenCode("provider.opencode.event_emit", { this.traceOpenCode("provider.opencode.event_emit", {
@@ -3003,7 +2996,16 @@ class OpenCodeAgentSession implements AgentSession {
async close(): Promise<void> { async close(): Promise<void> {
try { try {
// Flip closed before clearing subscribers so any event the SDK delivers
// after the abort (between here and subscribers.clear) is swallowed by
// notifySubscribers instead of bubbling through provider-runner as an
// unhandled rejection in whichever test the daemon hops to next.
this.closed = true;
this.abortController?.abort(); this.abortController?.abort();
this.eventStreamAbortController?.abort();
this.eventStreamAbortController = null;
this.eventStreamReady = null;
this.subscribers.clear();
await reconcileOpenCodeSessionClose({ await reconcileOpenCodeSessionClose({
client: this.client, client: this.client,
sessionId: this.sessionId, sessionId: this.sessionId,
@@ -3011,7 +3013,6 @@ class OpenCodeAgentSession implements AgentSession {
logger: this.logger, logger: this.logger,
}); });
await this.deleteProviderSessionIfEphemeral(); await this.deleteProviderSessionIfEphemeral();
this.subscribers.clear();
this.activeForegroundTurnId = null; this.activeForegroundTurnId = null;
} finally { } finally {
this.releaseServer?.(); this.releaseServer?.();
@@ -1,15 +1,19 @@
import { beforeAll, beforeEach, describe, test, expect } from "vitest"; import { afterAll, beforeAll, beforeEach, describe, test, expect } from "vitest";
import { mkdtempSync, rmSync } from "node:fs"; import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
import path from "node:path"; import path from "node:path";
import pino from "pino"; import pino from "pino";
import { OpenCodeAgentClient } from "./opencode-agent.js"; import { OpenCodeAgentClient } from "./opencode-agent.js";
import { OpenCodeServerManager } from "./opencode/server-manager.js";
import { isProviderAvailable } from "../../daemon-e2e/agent-configs.js"; import { isProviderAvailable } from "../../daemon-e2e/agent-configs.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js";
const BIG_PICKLE_MODEL = "opencode/big-pickle";
describe("OpenCode assistant message", () => { describe("OpenCode assistant message", () => {
let canRun = false; let canRun = false;
const logger = pino({ level: "silent" });
beforeAll(async () => { beforeAll(async () => {
canRun = await isProviderAvailable("opencode"); canRun = await isProviderAvailable("opencode");
@@ -21,16 +25,19 @@ describe("OpenCode assistant message", () => {
} }
}); });
afterAll(async () => {
await OpenCodeServerManager.getInstance(logger).shutdown();
});
test("assistant_message appears in live stream with opencode/big-pickle", async () => { test("assistant_message appears in live stream with opencode/big-pickle", async () => {
const cwd = mkdtempSync(path.join(tmpdir(), "opencode-msg-")); const cwd = mkdtempSync(path.join(tmpdir(), "opencode-msg-"));
const logger = pino({ level: "silent" });
const client = new OpenCodeAgentClient(logger); const client = new OpenCodeAgentClient(logger);
try { try {
const session = await client.createSession({ const session = await client.createSession({
provider: "opencode", provider: "opencode",
cwd, cwd,
model: "opencode/big-pickle", model: BIG_PICKLE_MODEL,
modeId: "build", modeId: "build",
}); });
@@ -46,13 +53,14 @@ describe("OpenCode assistant message", () => {
test("streamHistory returns assistant_message after a completed turn", async () => { test("streamHistory returns assistant_message after a completed turn", async () => {
const cwd = mkdtempSync(path.join(tmpdir(), "opencode-history-")); const cwd = mkdtempSync(path.join(tmpdir(), "opencode-history-"));
const logger = pino({ level: "silent" });
const client = new OpenCodeAgentClient(logger); const client = new OpenCodeAgentClient(logger);
try { try {
const session = await client.createSession({ const session = await client.createSession({
provider: "opencode", provider: "opencode",
cwd, cwd,
model: BIG_PICKLE_MODEL,
modeId: "build",
}); });
const result = await session.run("Say hello back in one sentence."); const result = await session.run("Say hello back in one sentence.");
@@ -1,15 +1,19 @@
import { beforeAll, beforeEach, describe, test, expect } from "vitest"; import { afterAll, beforeAll, beforeEach, describe, test, expect } from "vitest";
import { mkdtempSync, rmSync } from "node:fs"; import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
import path from "node:path"; import path from "node:path";
import pino from "pino"; import pino from "pino";
import { OpenCodeAgentClient } from "./opencode-agent.js"; import { OpenCodeAgentClient } from "./opencode-agent.js";
import { OpenCodeServerManager } from "./opencode/server-manager.js";
import { isProviderAvailable } from "../../daemon-e2e/agent-configs.js"; import { isProviderAvailable } from "../../daemon-e2e/agent-configs.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js";
const BIG_PICKLE_MODEL = "opencode/big-pickle";
describe("OpenCode reasoning dedup", () => { describe("OpenCode reasoning dedup", () => {
let canRun = false; let canRun = false;
const logger = pino({ level: "silent" });
beforeAll(async () => { beforeAll(async () => {
canRun = await isProviderAvailable("opencode"); canRun = await isProviderAvailable("opencode");
@@ -21,16 +25,19 @@ describe("OpenCode reasoning dedup", () => {
} }
}); });
afterAll(async () => {
await OpenCodeServerManager.getInstance(logger).shutdown();
});
test("reasoning content is not duplicated as assistant_message", async () => { test("reasoning content is not duplicated as assistant_message", async () => {
const cwd = mkdtempSync(path.join(tmpdir(), "opencode-reasoning-dedup-")); const cwd = mkdtempSync(path.join(tmpdir(), "opencode-reasoning-dedup-"));
const logger = pino({ level: "silent" });
const client = new OpenCodeAgentClient(logger); const client = new OpenCodeAgentClient(logger);
try { try {
const session = await client.createSession({ const session = await client.createSession({
provider: "opencode", provider: "opencode",
cwd, cwd,
model: "opencode/gpt-5-nano", model: BIG_PICKLE_MODEL,
modeId: "build", modeId: "build",
}); });
@@ -106,6 +106,19 @@ describe("OpenCodeServerManager generations", () => {
expect(second.process.kill).toHaveBeenCalledWith("SIGTERM"); expect(second.process.kill).toHaveBeenCalledWith("SIGTERM");
}); });
test("shutdown still signals a process after an earlier kill signal if it has not exited", async () => {
const manager = createTestManager();
const first = createGeneration(4451);
stubGenerations(manager, [first]);
await manager.acquire({ force: false });
first.process.killed = true;
await manager.shutdown();
expect(first.process.kill).toHaveBeenCalledWith("SIGTERM");
});
test("repeated rotations leave zero unreferenced retired servers", async () => { test("repeated rotations leave zero unreferenced retired servers", async () => {
const manager = createTestManager(); const manager = createTestManager();
const first = createGeneration(4501); const first = createGeneration(4501);
@@ -277,7 +277,10 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
} }
private async killServer(server: OpenCodeServerGeneration): Promise<void> { private async killServer(server: OpenCodeServerGeneration): Promise<void> {
if (server.process.killed) { if (
(server.process.exitCode !== null && server.process.exitCode !== undefined) ||
(server.process.signalCode !== null && server.process.signalCode !== undefined)
) {
return; return;
} }
const result = await terminateWithTreeKill(server.process, { const result = await terminateWithTreeKill(server.process, {
@@ -65,7 +65,7 @@ export class TestOpenCodeClient {
appAgentsResponse: OpenCodeResponse = { data: [] }; appAgentsResponse: OpenCodeResponse = { data: [] };
commandListResponse: OpenCodeResponse = { data: [] }; commandListResponse: OpenCodeResponse = { data: [] };
eventStream: AsyncIterable<unknown> = createEventStream([idleEvent()]); eventStream: AsyncIterable<unknown>;
experimentalSessionListResponse: OpenCodeResponse = { data: [] }; experimentalSessionListResponse: OpenCodeResponse = { data: [] };
permissionReplyResponse: OpenCodeResponse = {}; permissionReplyResponse: OpenCodeResponse = {};
providerListResponse: OpenCodeResponse = { data: { connected: [], all: [] } }; providerListResponse: OpenCodeResponse = { data: { connected: [], all: [] } };
@@ -74,13 +74,24 @@ export class TestOpenCodeClient {
questionReplyResponse: OpenCodeResponse = {}; questionReplyResponse: OpenCodeResponse = {};
sessionAbortResponse: OpenCodeResponse = {}; sessionAbortResponse: OpenCodeResponse = {};
sessionCommandError: unknown = null; sessionCommandError: unknown = null;
sessionCommandEvents: unknown[] = [idleEvent()];
sessionCommandResponse: OpenCodeResponse = {}; sessionCommandResponse: OpenCodeResponse = {};
sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } }; sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } };
sessionDeleteResponse: OpenCodeResponse = {}; sessionDeleteResponse: OpenCodeResponse = {};
sessionMessagesResponse: OpenCodeResponse = { data: [] }; sessionMessagesResponse: OpenCodeResponse = { data: [] };
sessionPromptAsyncEvents: unknown[] = [idleEvent()];
sessionPromptAsyncResponse: OpenCodeResponse = {}; sessionPromptAsyncResponse: OpenCodeResponse = {};
sessionSummarizeResponse: OpenCodeResponse = { data: {} }; sessionSummarizeResponse: OpenCodeResponse = { data: {} };
sessionUpdateResponse: OpenCodeResponse = {}; sessionUpdateResponse: OpenCodeResponse = {};
private readonly queuedEventStream = createQueuedEventStream();
constructor() {
this.eventStream = this.queuedEventStream.stream;
}
emitEvent(event: unknown): void {
this.queuedEventStream.emit(event);
}
asSdkClient(): OpencodeClient { asSdkClient(): OpencodeClient {
return { return {
@@ -154,6 +165,9 @@ export class TestOpenCodeClient {
if (this.sessionCommandError) { if (this.sessionCommandError) {
throw this.sessionCommandError; throw this.sessionCommandError;
} }
for (const event of this.sessionCommandEvents) {
this.emitEvent(event);
}
return this.sessionCommandResponse; return this.sessionCommandResponse;
}, },
create: async (parameters: unknown) => { create: async (parameters: unknown) => {
@@ -170,6 +184,9 @@ export class TestOpenCodeClient {
}, },
promptAsync: async (parameters: unknown) => { promptAsync: async (parameters: unknown) => {
this.calls.sessionPromptAsync.push(parameters); this.calls.sessionPromptAsync.push(parameters);
for (const event of this.sessionPromptAsyncEvents) {
this.emitEvent(event);
}
return this.sessionPromptAsyncResponse; return this.sessionPromptAsyncResponse;
}, },
summarize: async (parameters: unknown) => { summarize: async (parameters: unknown) => {
@@ -193,6 +210,38 @@ export function createEventStream(events: unknown[]): AsyncGenerator<unknown> {
})(); })();
} }
function createQueuedEventStream(): {
stream: AsyncIterable<unknown>;
emit: (event: unknown) => void;
} {
const queue: unknown[] = [];
const waiters: Array<(result: IteratorResult<unknown>) => void> = [];
return {
stream: {
[Symbol.asyncIterator]: () => ({
next: () => {
const event = queue.shift();
if (event !== undefined) {
return Promise.resolve({ done: false, value: event });
}
return new Promise<IteratorResult<unknown>>((resolve) => {
waiters.push(resolve);
});
},
}),
},
emit: (event: unknown) => {
const waiter = waiters.shift();
if (waiter) {
waiter({ done: false, value: event });
return;
}
queue.push(event);
},
};
}
export function idleEvent(): unknown { export function idleEvent(): unknown {
return { return {
type: "session.idle", type: "session.idle",
@@ -50,7 +50,7 @@ export const agentConfigs = {
}, },
opencode: { opencode: {
provider: "opencode", provider: "opencode",
model: "opencode/glm-5-free", model: "opencode/big-pickle",
modes: { modes: {
full: "default", full: "default",
ask: "default", ask: "default",