From f3fee73976b0f1075dca309e5a5ef53534d3107f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 31 Jul 2026 12:08:43 +0200 Subject: [PATCH] Stop OpenCode cancels from killing the next turn (#2662) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(opencode): prevent late aborts from canceling replacement turns OpenCode aborts are session-scoped, so provider idle alone does not make a session safe to reuse. Keep abort retries owned by the stopping turn and require both abort settlement and provider idle before dispatching the next runner operation. * fix(opencode): retain the stop barrier through provider idle Keep the stopping operation authoritative until both its session-scoped abort and provider-idle evidence settle, including stale idle-session interrupts. Also keep the real drift fixture credential-allowlisted and wait for command output pipes to drain. * test(opencode): launch Windows fixture shims safely * fix(opencode): recover stop observation safely * fix(opencode): preserve autonomous wake through stop * fix(opencode): retain deferred wake completion * fix(opencode): serialize stop boundary replay * test(opencode): use shared real e2e lane * fix(opencode): own stops started during replay * fix(opencode): retry stop observation on demand * fix(opencode): follow replacement stop boundaries * fix(opencode): retain wakes during stop recovery * fix(opencode): hold replay turns behind boundary * fix(opencode): filter canceled reconnect output * fix(opencode): bind status evidence to stop * fix(opencode): acknowledge owned cancellation * fix(opencode): reconcile stale busy evidence * fix(opencode): bind replay terminals to wakes * refactor(opencode): own the stop boundary in one explicit state The run boundary had grown into a mutable flag bag plus a raw-event buffer that was replayed and re-classified after the fact. Each new interleaving needed another predicate, and the classification kept losing autonomous wakes. Restore stopping as a turn-state variant that owns exactly two things: the session-scoped abort it issued (with its single owned retry) and the canceled run's terminal. Runner operations wait on both; a failed abort fails closed. Provider status reconciliation now ends the stop through the same transition an SSE terminal takes. The canceled run's residue is suppressed only up to that terminal. Events published afterwards are a new run by construction and take the normal live path, so a wake no longer waits for abort settlement and cannot be dropped by a retention heuristic. * fix(opencode): emit cancellation before adopting the next run The acknowledgement ran off a promise combinator, so a wake adopted from the event right after the terminal could reach subscribers before the canceled turn's turn_canceled. Acknowledge synchronously from whichever proof lands first instead. * docs(opencode): attach the stop doc comment to its interface * fix(opencode): retry the stop in progress instead of opening a second Pressing Stop again after an abort failed built a detached boundary: its fresh abort could succeed while the canceled turn kept waiting for an acknowledgement, so the daemon sat out the full cancel timeout before force-canceling. A repeated Stop now replaces the abort owned by the stop already in progress, which is what carries the acknowledgement and what lifts the fail-closed gate. * fix(opencode): accumulate every abort a stop has issued A repeated Stop replaced the stop's abort settlement instead of adding to it, so a first request still in flight — the slow-teardown case the cancel cap exists for — no longer held the gate. The replacement could start while that request was live and get cancelled by it, which is the race this boundary exists to close. Reuse now waits for every abort the stop issued, while only the newest can hold the gate closed. * fix(opencode): scope abort settlement to the session, not the stop Abort is session-scoped, so its settlement has to be too. Tracking it on the stop meant a new stop boundary silently dropped an older request that was still in flight, and that request could still cancel whatever run started next. The session now accumulates every abort it has issued; the stop is left owning only the run it is stopping. * fix(opencode): recheck the abort gate after waiting on it A waiting prompt read the settlement once, so a Stop issued while it waited was invisible to it: the older abort would settle, the prompt would start, and the newer one could still cancel it. Drain until the settlement observed is the current one. --- docs/providers.md | 8 + .../agent/providers/opencode-agent.test.ts | 717 +++++++++++++++++- .../server/agent/providers/opencode-agent.ts | 268 +++++-- .../providers/opencode-server-manager.test.ts | 17 +- .../providers/opencode/server-manager.ts | 4 + .../test-utils/test-opencode-harness.ts | 5 +- .../opencode-omo-autonomous.real.e2e.test.ts | 236 ++++++ .../opencode-omo-real-runtime.test.ts | 43 ++ .../daemon-e2e/opencode-omo-real-runtime.ts | 520 +++++++++++++ 9 files changed, 1730 insertions(+), 88 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts create mode 100644 packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.test.ts create mode 100644 packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.ts diff --git a/docs/providers.md b/docs/providers.md index 5bee5b40b..b5723733e 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -458,3 +458,11 @@ Tests use `isProviderAvailable(provider)` to skip when the binary or credentials **`defaultCommand` is a tuple.** The first element is the binary name, the rest are default arguments. The base class uses this to find the executable and spawn the process. **Runtime settings can override the command.** Users can configure custom binary paths or environment variables per provider via `ProviderRuntimeSettings`. Your factory in the registry should pass `runtimeSettings?.["your-provider"]` through to the constructor. + +**Session-scoped cancellation needs a stop boundary inside the provider.** Some agents cancel the whole session rather than one turn — OpenCode's `session.abort` is the example. A cancel that is still in flight when the next run starts will kill that replacement run, which is what makes "stop, then immediately prompt again" (`replaceRunning`, `notifyOnFinish` wakes, schedules) flaky. Own this in the provider session, not in `AgentManager`: + +- Model the stop as an explicit `stopping` turn-state variant carrying the canceled run's terminal and the cancellation the caller is still owed. Pressing Stop again retries the stop already in progress rather than opening a second one; never fire a detached retry, it will outlive its boundary. +- **Scope the cancel settlement the way the provider scopes the cancel.** If cancellation is session-scoped, so is its settlement: track it on the session, accumulating every request issued, and let it outlive the stop that issued it. A request still in flight lands on the runner whenever the server gets to it — however many stops have come and gone since. Scoping it to the current stop looks right and quietly drops older requests from the gate. Only the newest may hold the gate closed, since recovering from a failed cancel is what pressing Stop again is for. +- Gate the operations **the daemon issues** (prompt, slash command, summarize) on both the terminal and cancel settlement. Permission and question responses are not runner operations and must stay outside the gate, or an auto-approve deadlocks the stop. Runs the _provider_ starts on its own — plugin or autonomous wakes — are observed, not gated: the daemon does not choose when they begin, and holding their events back does not protect them from a cancel already in flight, it only hides a run that may already be dead. +- Fail closed: if the cancel never succeeded you never proved the run stopped, so refuse new runs until the next Stop issues a fresh cancel. `AgentManager` already turns a rejected `interrupt()` into a refused cancel. +- Suppress the canceled run's residue only until its authoritative terminal. Anything the provider publishes after that terminal is a new run by construction and must take the normal live path — buffering it and replaying it later is how autonomous/plugin wakes get lost. diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index b9af8193c..ecc1852b6 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -125,11 +125,13 @@ function userMessageEvents(params: { ]; } -function assistantTurnEvents({ +function assistantMessageEvents({ sessionId = "session-1", + messageId = "msg_assistant", text = "Hello from OpenCode", }: { sessionId?: string; + messageId?: string; text?: string; } = {}): unknown[] { return [ @@ -137,7 +139,7 @@ function assistantTurnEvents({ type: "message.updated", properties: { info: { - id: "msg_assistant", + id: messageId, sessionID: sessionId, role: "assistant", }, @@ -147,13 +149,25 @@ function assistantTurnEvents({ type: "message.part.delta", properties: { sessionID: sessionId, - messageID: "msg_assistant", - partID: "prt_text", + messageID: messageId, + partID: messageId === "msg_assistant" ? "prt_text" : `part_${messageId}`, field: "text", delta: text, }, }, - { type: "session.idle", properties: { sessionID: sessionId } }, + ]; +} + +function assistantTurnEvents( + params: { + sessionId?: string; + messageId?: string; + text?: string; + } = {}, +): unknown[] { + return [ + ...assistantMessageEvents(params), + { type: "session.idle", properties: { sessionID: params.sessionId ?? "session-1" } }, ]; } @@ -1901,27 +1915,603 @@ describe("OpenCode adapter startTurn error handling", () => { } }); - test("retries abort and waits for provider idle before starting the next prompt", async () => { + test("waits for the stop abort and provider idle before starting the next prompt", async () => { const { parent: session, openCode } = await createParentSession("ses_unit_test"); + const retryStarted = createTestDeferred(); + const settleRetry = createTestDeferred(); + const oldIdleConsumed = createTestDeferred(); + const replacementCompleted = createTestDeferred(); + const events: AgentStreamEvent[] = []; openCode.sessionPromptAsyncEvents = []; openCode.sessionAbortImplementation = async () => { if (openCode.calls.sessionAbort.length === 1) { throw new Error("abort transport failed"); } + retryStarted.resolve(); + await settleRetry.promise; + return { data: true }; + }; + session.subscribe((event) => { + events.push(event); + if (event.type === "provider_subagent") { + oldIdleConsumed.resolve(); + } + if (event.type === "turn_completed") { + replacementCompleted.resolve(); + } + }); + try { + await session.startTurn("first"); + const interrupt = session.interrupt(); + await retryStarted.promise; + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + openCode.emitEvent({ type: "session.idle", properties: { sessionID: "ses_unit_test" }, }); - return { data: true }; - }; - try { - await session.startTurn("first"); - await session.interrupt(); + openCode.emitEvent({ + type: "session.created", + properties: { + info: { + id: "ses_child_after_old_idle", + parentID: "ses_unit_test", + title: "Old idle drain marker", + directory: "/workspace/repo", + }, + }, + }); + await oldIdleConsumed.promise; + const replacement = session.startTurn("second"); + await new Promise((resolve) => setImmediate(resolve)); + + expect(openCode.calls.sessionAbort).toHaveLength(2); expect(openCode.calls.sessionPromptAsync).toHaveLength(1); - await session.startTurn("second"); - expect(openCode.calls.sessionAbort).toHaveLength(2); + settleRetry.resolve(); + await interrupt; + await replacement; expect(openCode.calls.sessionPromptAsync).toHaveLength(2); + + for (const event of assistantTurnEvents({ + sessionId: "ses_unit_test", + text: "Replacement completed safely.", + })) { + openCode.emitEvent(event); + } + await replacementCompleted.promise; + + expect( + turnEventSignatures(events.filter((event) => event.type !== "provider_subagent")), + ).toEqual([ + ["turn_started", "opencode-turn-0"], + ["turn_canceled", "opencode-turn-0"], + ["turn_started", "opencode-turn-1"], + ["timeline", "opencode-turn-1"], + ["turn_completed", "opencode-turn-1"], + ]); + } finally { + settleRetry.resolve(); + await session.close(); + } + }); + + test("streams an autonomous wake as soon as the canceled run reaches its terminal", async () => { + const { parent: session, openCode } = await createParentSession("ses_wake_after_terminal"); + const settleAbort = createTestDeferred(); + const events: AgentStreamEvent[] = []; + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + await settleAbort.promise; + return { data: true }; + }; + session.subscribe((event) => events.push(event)); + + try { + await session.startTurn("first"); + const interrupt = session.interrupt(); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(1)); + + for (const event of assistantMessageEvents({ + sessionId: "ses_wake_after_terminal", + messageId: "msg_canceled_assistant", + text: "Canceled residue.", + })) { + openCode.emitEvent(event); + } + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_wake_after_terminal" }, + }); + for (const event of userMessageEvents({ + sessionId: "ses_wake_after_terminal", + messageId: "msg_wake_after_terminal", + text: "Wake after the canceled terminal", + })) { + openCode.emitEvent(event); + } + for (const event of assistantMessageEvents({ + sessionId: "ses_wake_after_terminal", + messageId: "msg_wake_assistant", + text: "Wake output streams live.", + })) { + openCode.emitEvent(event); + } + + await vi.waitFor(() => + expect(events).toContainEqual( + expect.objectContaining({ + type: "timeline", + turnId: "opencode-turn-1", + item: expect.objectContaining({ text: "Wake output streams live." }), + }), + ), + ); + expect(events).not.toContainEqual( + expect.objectContaining({ + type: "timeline", + item: expect.objectContaining({ text: "Canceled residue." }), + }), + ); + + settleAbort.resolve(); + await interrupt; + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_wake_after_terminal" }, + }); + await vi.waitFor(() => + expect(events).toContainEqual( + expect.objectContaining({ type: "turn_completed", turnId: "opencode-turn-1" }), + ), + ); + await expect(session.startTurn("after the wake")).resolves.toEqual({ + turnId: "opencode-turn-2", + }); + } finally { + settleAbort.resolve(); + await session.close(); + } + }); + + test("preserves an autonomous wake that arrives after provider idle but before abort settlement", async () => { + const { parent: session, openCode } = await createParentSession("ses_deferred_autonomous"); + const settleAbort = createTestDeferred(); + const wakeEventsConsumed = createTestDeferred(); + const events: AgentStreamEvent[] = []; + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + await settleAbort.promise; + return { data: true }; + }; + session.subscribe((event) => { + events.push(event); + if ( + event.type === "provider_subagent" && + event.event.type === "upsert" && + event.event.id === "ses_deferred_wake_drain" + ) { + wakeEventsConsumed.resolve(); + } + }); + + try { + await session.startTurn("first"); + openCode.emitEvent({ + type: "message.part.updated", + properties: { + part: { + id: "prt_canceled_usage", + sessionID: "ses_deferred_autonomous", + messageID: "msg_canceled_assistant", + type: "step-finish", + tokens: { input: 100, output: 20, cache: { read: 50 } }, + }, + }, + }); + await vi.waitFor(() => + expect(events).toContainEqual( + expect.objectContaining({ + type: "usage_updated", + usage: expect.objectContaining({ + inputTokens: 100, + outputTokens: 20, + cachedInputTokens: 50, + }), + }), + ), + ); + const interrupt = session.interrupt(); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(1)); + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_deferred_autonomous" }, + }); + openCode.emitEvent({ + type: "message.updated", + properties: { + info: { + id: "msg_plugin_wake", + sessionID: "ses_deferred_autonomous", + role: "user", + }, + }, + }); + openCode.emitEvent({ + type: "message.part.updated", + properties: { + part: { + id: "prt_autonomous_usage", + sessionID: "ses_deferred_autonomous", + messageID: "msg_autonomous_assistant", + type: "step-finish", + tokens: { input: 7, output: 3 }, + }, + }, + }); + for (const event of assistantTurnEvents({ + sessionId: "ses_deferred_autonomous", + text: "Autonomous wake completed.", + })) { + openCode.emitEvent(event); + } + openCode.emitEvent({ + type: "session.created", + properties: { + info: { + id: "ses_deferred_wake_drain", + parentID: "ses_deferred_autonomous", + title: "Deferred wake drain marker", + directory: "/workspace/repo", + }, + }, + }); + await wakeEventsConsumed.promise; + + settleAbort.resolve(); + await interrupt; + await new Promise((resolve) => setImmediate(resolve)); + + expect(events).toContainEqual({ + type: "turn_started", + provider: "opencode", + turnId: "opencode-turn-1", + }); + await vi.waitFor(() => + expect(events).toContainEqual({ + type: "turn_completed", + provider: "opencode", + turnId: "opencode-turn-1", + usage: expect.objectContaining({ + inputTokens: 7, + outputTokens: 3, + contextWindowUsedTokens: 10, + }), + }), + ); + const autonomousCompletion = events.find( + (event) => event.type === "turn_completed" && event.turnId === "opencode-turn-1", + ); + expect(autonomousCompletion?.type).toBe("turn_completed"); + if (autonomousCompletion?.type === "turn_completed") { + expect(autonomousCompletion.usage?.cachedInputTokens).toBeUndefined(); + } + expect(events).toContainEqual( + expect.objectContaining({ + type: "timeline", + turnId: "opencode-turn-1", + item: expect.objectContaining({ + type: "assistant_message", + text: "Autonomous wake completed.", + }), + }), + ); + await expect(session.startTurn("after autonomous wake")).resolves.toEqual({ + turnId: "opencode-turn-2", + }); + } finally { + settleAbort.resolve(); + await session.close(); + } + }); + + test("waits for an idle-session abort before starting a prompt", async () => { + const { parent: session, openCode } = await createParentSession("ses_idle_abort"); + const settleAbort = createTestDeferred(); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + await settleAbort.promise; + return { data: true }; + }; + + try { + const interrupt = session.interrupt(); + const prompt = session.startTurn("next"); + await new Promise((resolve) => setImmediate(resolve)); + expect(openCode.calls.sessionPromptAsync).toHaveLength(0); + + settleAbort.resolve(); + await interrupt; + await prompt; + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + } finally { + settleAbort.resolve(); + await session.close(); + } + }); + + test("does not acknowledge cancellation when both abort attempts fail", async () => { + const { parent: session, openCode } = await createParentSession("ses_failed_idle_abort"); + const events: AgentStreamEvent[] = []; + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + throw new Error("abort failed"); + }; + session.subscribe((event) => events.push(event)); + + try { + await session.startTurn("still running upstream"); + await expect(session.interrupt()).rejects.toThrow("abort failed"); + + await expect(session.startTurn("unsafe replacement")).rejects.toThrow("abort failed"); + expect(openCode.calls.sessionAbort).toHaveLength(2); + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + expect(events.map((event) => event.type)).toEqual(["turn_started"]); + } finally { + await session.close(); + } + }); + + test("acknowledges the canceled turn when a repeated Stop finally aborts", async () => { + const { parent: session, openCode } = await createParentSession("ses_retried_stop"); + const events: AgentStreamEvent[] = []; + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + if (openCode.calls.sessionAbort.length <= 2) { + throw new Error("abort failed"); + } + return { data: true }; + }; + session.subscribe((event) => events.push(event)); + + try { + await session.startTurn("still running upstream"); + await expect(session.interrupt()).rejects.toThrow("abort failed"); + expect(events.map((event) => event.type)).toEqual(["turn_started"]); + + await session.interrupt(); + + expect(openCode.calls.sessionAbort).toHaveLength(3); + expect(events).toContainEqual({ + type: "turn_canceled", + provider: "opencode", + reason: "interrupted", + turnId: "opencode-turn-0", + }); + + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_retried_stop" }, + }); + await expect(session.startTurn("replacement")).resolves.toEqual({ + turnId: "opencode-turn-1", + }); + expect(openCode.calls.sessionPromptAsync).toHaveLength(2); + } finally { + await session.close(); + } + }); + + test("keeps a waiting replacement behind an abort issued while it waits", async () => { + const { parent: session, openCode } = await createParentSession("ses_stop_during_wait"); + const settleFirstAbort = createTestDeferred(); + const settleSecondAbort = createTestDeferred(); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + await (openCode.calls.sessionAbort.length === 1 + ? settleFirstAbort.promise + : settleSecondAbort.promise); + return { data: true }; + }; + + let replacement: Promise<{ turnId: string }> | undefined; + try { + await session.startTurn("first"); + const firstStop = session.interrupt(); + void firstStop.catch(() => undefined); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(1)); + + replacement = session.startTurn("replacement"); + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_stop_during_wait" }, + }); + + const secondStop = session.interrupt(); + void secondStop.catch(() => undefined); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(2)); + + settleFirstAbort.resolve(); + await firstStop; + await new Promise((resolve) => setImmediate(resolve)); + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + + settleSecondAbort.resolve(); + await secondStop; + await replacement; + expect(openCode.calls.sessionPromptAsync).toHaveLength(2); + } finally { + settleFirstAbort.resolve(); + settleSecondAbort.resolve(); + await replacement?.catch(() => undefined); + await session.close(); + } + }); + + test("keeps a replacement behind an abort left in flight by an earlier stop", async () => { + const { parent: session, openCode } = await createParentSession("ses_carried_abort"); + const settleFirstAbort = createTestDeferred(); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + if (openCode.calls.sessionAbort.length === 1) { + await settleFirstAbort.promise; + } + return { data: true }; + }; + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + let replacement: Promise<{ turnId: string }> | undefined; + try { + await session.startTurn("first"); + const firstStop = session.interrupt(); + void firstStop.catch(() => undefined); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(1)); + + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_carried_abort" }, + }); + for (const event of userMessageEvents({ + sessionId: "ses_carried_abort", + messageId: "msg_wake_between_stops", + text: "Autonomous wake between stops", + })) { + openCode.emitEvent(event); + } + await vi.waitFor(() => + expect(events).toContainEqual({ + type: "turn_started", + provider: "opencode", + turnId: "opencode-turn-1", + }), + ); + + await session.interrupt(); + expect(openCode.calls.sessionAbort).toHaveLength(2); + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_carried_abort" }, + }); + + replacement = session.startTurn("replacement"); + await new Promise((resolve) => setImmediate(resolve)); + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + + settleFirstAbort.resolve(); + await firstStop; + await replacement; + expect(openCode.calls.sessionPromptAsync).toHaveLength(2); + } finally { + settleFirstAbort.resolve(); + await replacement?.catch(() => undefined); + await session.close(); + } + }); + + test("keeps a replacement behind an earlier Stop's abort that is still in flight", async () => { + const { parent: session, openCode } = await createParentSession("ses_overlapping_stops"); + const settleFirstAbort = createTestDeferred(); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + if (openCode.calls.sessionAbort.length === 1) { + await settleFirstAbort.promise; + } + return { data: true }; + }; + + let replacement: Promise<{ turnId: string }> | undefined; + try { + await session.startTurn("first"); + const firstStop = session.interrupt(); + void firstStop.catch(() => undefined); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(1)); + + await session.interrupt(); + expect(openCode.calls.sessionAbort).toHaveLength(2); + openCode.emitEvent({ + type: "session.idle", + properties: { sessionID: "ses_overlapping_stops" }, + }); + + replacement = session.startTurn("replacement"); + await new Promise((resolve) => setImmediate(resolve)); + expect(openCode.calls.sessionPromptAsync).toHaveLength(1); + + settleFirstAbort.resolve(); + await firstStop; + await replacement; + expect(openCode.calls.sessionPromptAsync).toHaveLength(2); + } finally { + settleFirstAbort.resolve(); + await replacement?.catch(() => undefined); + await session.close(); + } + }); + + test("does not retry an owned abort after the session closes", async () => { + const { parent: session, openCode } = await createParentSession("ses_close_pending_abort"); + const settleOwnedAbort = createTestDeferred(); + openCode.sessionAbortImplementation = async () => { + if (openCode.calls.sessionAbort.length === 1) { + await settleOwnedAbort.promise; + throw new Error("late abort failure"); + } + return { data: true }; + }; + + const interrupt = session.interrupt(); + await vi.waitFor(() => expect(openCode.calls.sessionAbort).toHaveLength(1)); + await session.close(); + expect(openCode.calls.sessionAbort).toHaveLength(2); + + settleOwnedAbort.resolve(); + await expect(interrupt).rejects.toThrow("late abort failure"); + expect(openCode.calls.sessionAbort).toHaveLength(2); + }); + + test("fails closed when the SDK abort throws before returning a promise", async () => { + const openCode = new TestOpenCodeClient(); + const sdkClient = openCode.asSdkClient(); + const workingAbort = sdkClient.session.abort; + sdkClient.session.abort = (() => { + throw new Error("synchronous abort failure"); + }) as typeof sdkClient.session.abort; + const session = new __openCodeInternals.OpenCodeAgentSession( + { provider: "opencode", cwd: "/workspace/repo" }, + sdkClient, + "ses_sync_abort_failure", + createTestLogger(), + ); + + try { + await expect(session.interrupt()).rejects.toThrow("synchronous abort failure"); + + await expect(session.startTurn("unsafe replacement")).rejects.toThrow( + "synchronous abort failure", + ); + expect(openCode.calls.sessionPromptAsync).toHaveLength(0); + } finally { + sdkClient.session.abort = workingAbort; + await session.close(); + } + }); + + test("fails closed when idle-session abort attempts return SDK errors", async () => { + const { parent: session, openCode } = await createParentSession("ses_error_idle_abort"); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => ({ + error: new Error("abort response failed"), + }); + + try { + await expect(session.interrupt()).rejects.toThrow("abort response failed"); + + await expect(session.startTurn("unsafe replacement")).rejects.toThrow( + "abort response failed", + ); + expect(openCode.calls.sessionAbort).toHaveLength(2); + expect(openCode.calls.sessionPromptAsync).toHaveLength(0); } finally { await session.close(); } @@ -1948,21 +2538,91 @@ describe("OpenCode adapter startTurn error handling", () => { } }); - test("reconnects and confirms provider idle when the event stream ends while stopping", async () => { + test("keeps waiting for the stop terminal when the reconnect status probe fails", async () => { const firstStreamEnd = createTestDeferred(); + const settleAbort = createTestDeferred(); + const releaseTerminal = createTestDeferred(); let subscriptionCount = 0; const { parent: session, openCode } = await createParentSession( - "ses_stream_reconnect", + "ses_stop_status_probe_failure", (client) => { client.globalEventImplementation = async (options) => { subscriptionCount += 1; const signal = (options as { signal: AbortSignal }).signal; - const end = subscriptionCount === 1 ? firstStreamEnd.promise : waitForAbort(signal); return { stream: { async *[Symbol.asyncIterator]() { yield { type: "server.connected", properties: {} }; - await end; + if (subscriptionCount === 1) { + await firstStreamEnd.promise; + return; + } + await releaseTerminal.promise; + yield { + type: "session.idle", + properties: { sessionID: "ses_stop_status_probe_failure" }, + }; + await waitForAbort(signal); + }, + }, + }; + }; + client.sessionStatusImplementation = async () => { + throw new Error("provider status unavailable"); + }; + }, + ); + openCode.sessionPromptAsyncEvents = []; + openCode.sessionAbortImplementation = async () => { + await settleAbort.promise; + return { data: true }; + }; + + try { + await session.startTurn("first"); + const interrupt = session.interrupt(); + const replacement = session.startTurn("second"); + firstStreamEnd.resolve(); + await vi.waitFor(() => expect(openCode.calls.sessionStatus).toHaveLength(1)); + + settleAbort.resolve(); + await interrupt; + await new Promise((resolve) => setImmediate(resolve)); + + expect(openCode.calls.globalEvent).toHaveLength(2); + expect(openCode.calls.sessionStatus).toHaveLength(1); + + releaseTerminal.resolve(); + await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" }); + expect(openCode.calls.sessionPromptAsync).toHaveLength(2); + expect(openCode.calls.globalEvent).toHaveLength(2); + } finally { + settleAbort.resolve(); + releaseTerminal.resolve(); + await session.close(); + } + }, 15_000); + + test("does not reconnect the stop observer while closing", async () => { + const firstStreamEnd = createTestDeferred(); + const streamSignals: AbortSignal[] = []; + let subscriptionCount = 0; + const { parent: session, openCode } = await createParentSession( + "ses_close_while_stopping", + (client) => { + client.globalEventImplementation = async (options) => { + subscriptionCount += 1; + const signal = (options as { signal: AbortSignal }).signal; + streamSignals.push(signal); + return { + stream: { + async *[Symbol.asyncIterator]() { + yield { type: "server.connected", properties: {} }; + if (subscriptionCount === 1) { + await firstStreamEnd.promise; + return; + } + await waitForAbort(signal); }, }, }; @@ -1974,18 +2634,23 @@ describe("OpenCode adapter startTurn error handling", () => { try { await session.startTurn("first"); await session.interrupt(); - - const secondTurn = session.startTurn("second"); + const replacement = session.startTurn("second"); + void replacement.catch(() => undefined); firstStreamEnd.resolve(); - - await expect(secondTurn).resolves.toEqual({ turnId: "opencode-turn-1" }); - expect(openCode.calls.globalEvent).toHaveLength(2); - expect(openCode.calls.sessionStatus).toEqual([{ directory: "/workspace/repo" }]); - expect(openCode.calls.sessionPromptAsync).toHaveLength(2); - } finally { await session.close(); + + await expect(replacement).rejects.toThrow("OpenCode session is closed"); + expect(openCode.calls.globalEvent).toHaveLength(1); + } finally { + for (const signal of streamSignals) { + if (!signal.aborted) { + (signal as AbortSignal & { dispatchEvent: (event: Event) => boolean }).dispatchEvent( + new Event("abort"), + ); + } + } } - }, 15_000); + }); }); describe("OpenCodeAgentClient env", () => { diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 850cd0b4e..99c542bea 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -2760,7 +2760,25 @@ function createDeferred(): Deferred { type OpenCodeTurnState = | { status: "idle" } | { status: "running"; turnId: string } - | { status: "stopping"; idle: Deferred }; + | { status: "stopping"; stop: OpenCodeStop }; + +type OpenCodeRunnerStatus = "idle" | "busy" | "retry"; + +/** + * One in-flight stop of the OpenCode session runner. + * + * A stop tracks the run it is stopping: the terminal that run still owes, and + * the cancellation the caller is still owed. The aborts it issues are tracked by + * the session instead, because OpenCode's abort is session-scoped rather than + * turn-scoped and can outlive the stop that issued it. The runner is reusable + * only once both the terminal and every issued abort have settled. + */ +interface OpenCodeStop { + /** Foreground turn still owed a cancellation acknowledgement; cleared once emitted. */ + pendingCancellationTurnId: string | null; + /** Resolves when the canceled run publishes its authoritative terminal. */ + readonly terminal: Deferred; +} function unwrapOpenCodeGlobalEvent(event: unknown): OpenCodeEvent | null { const record = readOpenCodeRecord(event); @@ -2913,6 +2931,12 @@ class OpenCodeAgentSession implements AgentSession { private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); private nextTurnOrdinal = 0; private turnState: OpenCodeTurnState = { status: "idle" }; + /** + * Settlement of every session-scoped abort issued so far. It outlives the stop + * that issued it because a request still in flight can cancel a replacement + * run, and a rejection means we never proved the runner stopped. + */ + private abortSettlement: Promise = Promise.resolve(); private readonly runningToolCalls = new Map(); private subAgentsByCallId = new Map(); private subAgentCallIdByChildSessionId = new Map(); @@ -3009,22 +3033,31 @@ class OpenCodeAgentSession implements AgentSession { async interrupt(): Promise { const turnId = this.activeForegroundTurnId; this.abortController?.abort(); - if (turnId) { - this.beginStoppingTurn(turnId); - } + const abort = this.issueStop(turnId); // COMPAT(opencodeSlowAbort): OpenCode 1.14.42+ blocks session.abort until // the running tool actually stops, which can be tens of seconds for // long-running tools. Cap the wait so the user-visible cancel lands // quickly while still giving OpenCode a chance to confirm the abort // cleanly. Drop the timeout once upstream returns abort acknowledgement // before tool teardown. - const abortPromise = this.abortSession(turnId, "interrupt"); - await withTimeout(abortPromise, 2_000, "OpenCode session.abort").catch((error) => { - this.logger.warn( - { err: error, sessionId: this.sessionId, turnId }, - "OpenCode session.abort exceeded the cancel cap; proceeding with local cancel", - ); - }); + const settledAbort = abort.then( + () => undefined, + (error: unknown) => error, + ); + // Only the cap is tolerated. A settled failure means the runner may still be + // going, and the caller must hear about it. + const abortFailure = await withTimeout(settledAbort, 2_000, "OpenCode session.abort").catch( + (error) => { + this.logger.warn( + { err: error, sessionId: this.sessionId, turnId }, + "OpenCode session.abort did not settle within the cancel cap", + ); + return undefined; + }, + ); + if (abortFailure !== undefined) { + throw abortFailure; + } } async revertBoth(input: { messageId: string }): Promise { @@ -3042,73 +3075,118 @@ class OpenCodeAgentSession implements AgentSession { sessionID: this.sessionId, directory: this.config.cwd, }) - .then(() => undefined) + .then((response) => { + if (response.error) { + throw new Error(toDiagnosticErrorMessage(response.error)); + } + return undefined; + }) .catch((error) => { this.logger.warn( { err: error, sessionId: this.sessionId, turnId, reason }, "OpenCode session.abort rejected", ); + throw error; }); } + /** + * Gate every runner-affecting operation on the previous stop. OpenCode runs one + * runner per session and aborts it session-wide, so a replacement may start + * only once the canceled run published its terminal and our own abort settled. + * A failed abort never proved the runner stopped, so it fails closed until the + * next Stop issues a fresh one. + */ + private async awaitRunnerQuiescence(): Promise { + const providerIdle = this.waitUntilProviderIdle(); + // A failed abort is decisive on its own, so let it reject this wait without + // leaving the still-running observation unhandled. + void providerIdle.catch(() => undefined); + let observed = this.abortSettlement; + await Promise.all([observed, providerIdle]); + // Stop can issue a further abort while we waited, and the settlement is + // replaced rather than mutated, so drain until what we observed is current. + while (this.abortSettlement !== observed) { + observed = this.abortSettlement; + await observed; + } + } + private async waitUntilProviderIdle(): Promise { if (this.turnState.status !== "stopping") { return; } - - const stopping = this.turnState; - // OpenCode creates prompts before joining its single session runner. Sending - // while that runner is stopping can strand the new prompt behind the old - // run, so only provider-confirmed idle makes the session reusable. await withTimeout( - this.observeProviderStopBoundary(stopping), + this.observeProviderStopBoundary(this.turnState.stop), OPENCODE_PENDING_ABORT_START_TIMEOUT_MS, "OpenCode previous turn to stop", ); } - private async observeProviderStopBoundary( - stopping: Extract, - ): Promise { - while (this.turnState === stopping) { - void this.abortSession(null, "turn_start_boundary"); + private async observeProviderStopBoundary(stop: OpenCodeStop): Promise { + while (this.isStopping(stop)) { const eventStreamTask = this.eventStreamTask; const boundary = await (eventStreamTask ? Promise.race([ - stopping.idle.promise.then(() => "idle" as const), + stop.terminal.promise.then(() => "terminal" as const), eventStreamTask.then( () => "stream_ended" as const, () => "stream_ended" as const, ), ]) : Promise.resolve("stream_ended" as const)); - if (boundary === "idle") { + if (boundary === "terminal") { return; } - - await this.ensureEventStreamReady(); - const response = await this.client.session.status({ directory: this.config.cwd }); - if (response.error) { - throw new Error( - `Failed to confirm OpenCode session status: ${toDiagnosticErrorMessage(response.error)}`, - ); - } - const statuses = readOpenCodeRecord(response.data); - if (!statuses) { - throw new Error("OpenCode returned an invalid session status response"); - } - const status = readOpenCodeRecord(statuses[this.sessionId]); - const statusType = readNonEmptyString(status?.type); - if (!status || statusType === "idle") { - this.finishStoppingTurn(); - return; - } - if (statusType !== "busy" && statusType !== "retry") { - throw new Error(`OpenCode returned an unknown session status '${statusType ?? "missing"}'`); - } + await this.reconcileStopWithProviderStatus(stop); } } + /** + * The event stream dropped mid-stop, so the canceled run's terminal may have + * gone with it. `session.status` is the provider's authority on whether the + * runner is still going, and an idle answer ends the stop through the same + * transition an SSE terminal takes. A failed probe is not evidence either way, + * so the stop keeps waiting on the replacement stream instead of rejecting a + * caller that a later terminal would have released. + */ + private async reconcileStopWithProviderStatus(stop: OpenCodeStop): Promise { + await this.ensureEventStreamReady(); + try { + if ((await this.readProviderRunnerStatus()) === "idle") { + this.finishStoppingTurn(stop); + } + } catch (error) { + this.logger.warn( + { err: error, sessionId: this.sessionId, turnId: stop.pendingCancellationTurnId }, + "Failed to reconcile the OpenCode stop with provider session status", + ); + } + } + + private async readProviderRunnerStatus(): Promise { + const response = await this.client.session.status({ directory: this.config.cwd }); + if (response.error) { + throw new Error( + `Failed to confirm OpenCode session status: ${toDiagnosticErrorMessage(response.error)}`, + ); + } + const statuses = readOpenCodeRecord(response.data); + if (!statuses) { + throw new Error("OpenCode returned an invalid session status response"); + } + const status = readOpenCodeRecord(statuses[this.sessionId]); + // OpenCode drops idle sessions from the status map entirely. + if (!status) { + return "idle"; + } + const statusType = readNonEmptyString(status.type); + if (statusType !== "idle" && statusType !== "busy" && statusType !== "retry") { + throw new Error(`OpenCode returned an unknown session status '${statusType ?? "missing"}'`); + } + return statusType; + } + async startTurn( prompt: AgentPromptInput, options?: AgentRunOptions, @@ -3116,7 +3194,7 @@ class OpenCodeAgentSession implements AgentSession { if (this.turnState.status === "running") { throw new Error("A foreground turn is already active"); } - await this.waitUntilProviderIdle(); + await this.awaitRunnerQuiescence(); if (this.turnState.status !== "idle") { throw new Error("OpenCode is still stopping the previous turn"); } @@ -3451,6 +3529,9 @@ class OpenCodeAgentSession implements AgentSession { } private ensureEventStreamReady(): Promise { + if (this.closed) { + return Promise.reject(new Error("OpenCode session is closed")); + } if (this.eventStreamReady) { return this.eventStreamReady.promise; } @@ -3580,8 +3661,11 @@ class OpenCodeAgentSession implements AgentSession { this.turnState.status === "stopping" && getOpenCodeEventSessionId(event) === this.sessionId ) { + // Residue of the canceled run must not surface as a new turn. Its terminal + // is the authoritative end of the stop, so anything OpenCode publishes + // afterwards belongs to a new run by construction and takes the live path. if (isOpenCodeTerminalEvent(event, this.sessionId)) { - this.finishStoppingTurn(); + this.finishStoppingTurn(this.turnState.stop); } this.traceOpenCode("provider.opencode.event.skip", { n: eventCount, @@ -3720,29 +3804,96 @@ class OpenCodeAgentSession implements AgentSession { this.notifySubscribers(event, turnId); } - private beginStoppingTurn(turnId: string): void { - if (this.turnState.status !== "running" || this.turnState.turnId !== turnId) { - return; + private isStopping(stop: OpenCodeStop): boolean { + return this.turnState.status === "stopping" && this.turnState.stop === stop; + } + + /** Stops the session runner and returns the abort this Stop issued. */ + private issueStop(turnId: string | null): Promise { + if (this.turnState.status === "stopping") { + // Stop pressed again during a stop retries that same stop. There is one + // runner per session, so a second boundary would race the first — and + // after a failed abort, a fresh one is both the only proof that can still + // acknowledge the canceled turn and the only way out of fail-closed. + return this.issueOwnedAbort(this.turnState.stop); + } + const stop: OpenCodeStop = { + pendingCancellationTurnId: turnId, + terminal: createDeferred(), + }; + const abort = this.issueOwnedAbort(stop); + if (turnId) { + this.synthesizeInterruptedToolCalls(turnId); + // An idle session has no run to observe, so only abort settlement gates + // reuse there. A running one also owes the canceled run's terminal. + this.turnState = { status: "stopping", stop }; } - this.synthesizeInterruptedToolCalls(turnId); this.pendingUserMessageText = null; this.pendingClientMessageId = null; this.abortController = null; - this.turnState = { status: "stopping", idle: createDeferred() }; + return abort; + } + + /** Issues one more session-scoped abort for this stop and returns it. */ + private issueOwnedAbort(stop: OpenCodeStop): Promise { + const abort = this.runOwnedAbort(stop.pendingCancellationTurnId); + // Abort is session-scoped, so its settlement is too: an older request lands + // on the runner whenever the server gets to it, however many stops have come + // and gone since. Only the newest abort may hold the gate closed, since + // recovering from a failed one is what pressing Stop again is for. + const stillInFlight = this.abortSettlement.catch(() => undefined); + this.abortSettlement = Promise.all([stillInFlight, abort]).then(() => undefined); + void this.abortSettlement.catch(() => undefined); + // Cancellation is acknowledged as soon as an owned abort succeeds, or when + // the provider publishes the canceled run's terminal. + void abort.then( + () => this.acknowledgeCancellation(stop), + () => undefined, + ); + return abort; + } + + private acknowledgeCancellation(stop: OpenCodeStop): void { + const turnId = stop.pendingCancellationTurnId; + if (!turnId) { + return; + } + stop.pendingCancellationTurnId = null; this.notifySubscribers( { type: "turn_canceled", provider: "opencode", reason: "interrupted" }, turnId, ); } - private finishStoppingTurn(): void { - if (this.turnState.status !== "stopping") { + /** + * Issues the session-scoped abort for one stop, retrying it once. The stop owns + * every abort it needs: a detached retry would outlive its own boundary and + * cancel whichever run happened to be current when it landed. + */ + private runOwnedAbort(turnId: string | null): Promise { + // The turn hop also converts a synchronous SDK throw into a rejection. + return Promise.resolve() + .then(() => this.abortSession(turnId, "interrupt")) + .catch((error) => { + if (this.closed) { + throw error; + } + return this.abortSession(turnId, "stop_retry"); + }); + } + + private finishStoppingTurn(stop: OpenCodeStop): void { + if (!this.isStopping(stop)) { return; } - const stopping = this.turnState; + // Acknowledge before leaving the stopping state so a successor run adopted + // from the very next event cannot start ahead of the cancellation. + this.acknowledgeCancellation(stop); resetOpenCodeTurnTrackingState(this.createTranslationState()); + const contextWindowMaxTokens = this.resolveSelectedModelContextWindowMaxTokens(); + this.accumulatedUsage = contextWindowMaxTokens !== undefined ? { contextWindowMaxTokens } : {}; this.turnState = { status: "idle" }; - stopping.idle.resolve(); + stop.terminal.resolve(); } private trackToolCall(item: ToolCallTimelineItem): void { @@ -3991,9 +4142,6 @@ class OpenCodeAgentSession implements AgentSession { logger: this.logger, }); await this.deleteProviderSessionIfEphemeral(); - if (this.turnState.status === "stopping") { - this.turnState.idle.resolve(); - } this.turnState = { status: "idle" }; } finally { await this.releaseServer?.(); diff --git a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts index de094dba6..38f698b47 100644 --- a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts +++ b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts @@ -27,6 +27,16 @@ afterEach(() => { }); describe("OpenCodeServerManager generations", () => { + test("uses an explicit base environment for the server process", async () => { + const baseEnv = { HOME: "/isolated/home", PATH: "/isolated/bin" }; + const { manager, runtime } = createTestManager([4091], { baseEnv }); + + const acquisition = await manager.acquireCurrent(); + + expect(runtime.spawnCalls[0]?.options.baseEnv).toEqual(baseEnv); + await acquisition.release(); + }); + test("rotation creates a new current server without killing a referenced old server", async () => { const { manager, runtime } = createTestManager([4101, 4102]); @@ -334,7 +344,11 @@ describe.runIf(process.platform === "win32")( function createTestManager( ports: number[], - options: { autoAnnounce?: boolean; opencodeHomeDir?: string } = {}, + options: { + autoAnnounce?: boolean; + baseEnv?: Record; + opencodeHomeDir?: string; + } = {}, ): { manager: OpenCodeServerManager; runtime: FakeOpenCodeServerRuntime; @@ -346,6 +360,7 @@ function createTestManager( return { manager: new OpenCodeServerManager({ logger: createTestLogger(), + baseEnv: options.baseEnv, managedProcesses: runtime.managedProcesses, portAllocator: runtime.allocatePort, resolveCommandPrefix: runtime.resolveCommandPrefix, diff --git a/packages/server/src/server/agent/providers/opencode/server-manager.ts b/packages/server/src/server/agent/providers/opencode/server-manager.ts index d765d3d27..2baa07bbd 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -53,6 +53,7 @@ export type OpenCodeServerProcessSpawner = ( export interface OpenCodeServerManagerOptions { logger: Logger; + baseEnv?: SpawnProcessOptions["baseEnv"]; runtimeSettings?: ProviderRuntimeSettings; managedProcesses?: ManagedProcessRegistry; terminateProcess?: ProcessTerminator; @@ -70,6 +71,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { private startPromise: Promise | null = null; private newServerPromise: Promise | null = null; private readonly logger: Logger; + private readonly baseEnv?: SpawnProcessOptions["baseEnv"]; private readonly runtimeSettings?: ProviderRuntimeSettings; private readonly runtimeSettingsKey: string; private readonly managedProcesses?: ManagedProcessRegistry; @@ -81,6 +83,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { constructor(options: OpenCodeServerManagerOptions) { this.logger = options.logger; + this.baseEnv = options.baseEnv; this.runtimeSettings = options.runtimeSettings; this.runtimeSettingsKey = JSON.stringify(this.runtimeSettings ?? {}); this.managedProcesses = options.managedProcesses; @@ -298,6 +301,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { detached: process.platform !== "win32", stdio: ["ignore", "pipe", "pipe"], ...createProviderEnvSpec({ + baseEnv: this.baseEnv, runtimeSettings: this.runtimeSettings, overlays: [launchEnv], }), diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts index b59399279..7433532f0 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts @@ -124,6 +124,7 @@ export class TestOpenCodeClient { sessionPromptAsyncEvents: unknown[] = [idleEvent()]; sessionPromptAsyncResponse: OpenCodeResponse = {}; sessionStatusResponse: OpenCodeResponse = { data: {} }; + sessionStatusImplementation: ((parameters: unknown) => Promise) | null = null; sessionSummarizeEvents: unknown[] = [idleEvent()]; sessionSummarizeResponse: OpenCodeResponse = { data: {} }; sessionUpdateResponse: OpenCodeResponse = {}; @@ -260,7 +261,9 @@ export class TestOpenCodeClient { }, status: async (parameters: unknown) => { this.calls.sessionStatus.push(parameters); - return this.sessionStatusResponse; + return this.sessionStatusImplementation + ? await this.sessionStatusImplementation(parameters) + : this.sessionStatusResponse; }, summarize: async (parameters: unknown) => { this.calls.sessionSummarize.push(parameters); diff --git a/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts new file mode 100644 index 000000000..db3fb4482 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/opencode-omo-autonomous.real.e2e.test.ts @@ -0,0 +1,236 @@ +import { writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { expect, test } from "vitest"; + +import type { SessionOutboundMessage } from "../messages.js"; +import { createMessageCollector } from "../test-utils/message-collector.js"; +import { createOpenCodeOmoRealRuntime } from "./opencode-omo-real-runtime.js"; + +const CHILD_TOKEN = "PASEO_OMO_CHILD_READY"; +const ROOT_IDLE_TOKEN = "PASEO_OMO_ROOT_WAITING"; +const FINAL_TOKEN = "PASEO_OMO_AUTONOMOUS_FINAL"; +const ROOT_IDLE_BARRIER = ".paseo-omo-root-idle"; + +test("OpenCode remains idle while an OMO child works and completes the autonomous wake", async () => { + const runtime = await createOpenCodeOmoRealRuntime(); + const collector = createMessageCollector(runtime.client); + let passed = false; + + try { + const modeId = await resolveSisyphusMode(runtime.client, runtime.workspace); + const agent = await runtime.client.createAgent({ + provider: "opencode", + cwd: runtime.workspace, + title: "OpenCode OMO autonomous wake", + model: runtime.model, + modeId, + }); + + await runtime.client.sendMessage( + agent.id, + [ + "Use the OMO task tool exactly once with category quick and run_in_background=true.", + `Tell the child: use bash to wait until ${ROOT_IDLE_BARRIER} exists in the workspace, sleep 5, then return exactly ${CHILD_TOKEN}.`, + `After the background task starts, reply exactly ${ROOT_IDLE_TOKEN} and stop.`, + "When the background completion notification arrives, retrieve the task result with background_output.", + `After confirming ${CHILD_TOKEN}, reply exactly ${FINAL_TOKEN}.`, + ].join(" "), + ); + + const firstFinish = await runtime.client.waitForFinish(agent.id, 240_000); + expect(firstFinish.status).toBe("idle"); + const firstTimeline = await runtime.client.fetchAgentTimeline(agent.id, { limit: 240 }); + expect(assistantTexts(firstTimeline.entries).join("\n")).toContain(ROOT_IDLE_TOKEN); + await writeFile(join(runtime.workspace, ROOT_IDLE_BARRIER), "root idle\n", "utf8"); + + const completedChild = await waitForCompletedChildWithToken( + runtime.client, + agent.id, + CHILD_TOKEN, + ); + + const firstTerminal = firstTerminalIndex(collector.messages, agent.id); + await waitForMessage( + collector.messages, + (message, index) => + index > firstTerminal && isAgentStreamEvent(message, agent.id, "turn_started"), + "autonomous turn start", + ); + const wakeStartIndex = collector.messages.findIndex( + (message) => + isAgentStreamEvent(message, agent.id, "turn_started") && + collector.messages.indexOf(message) > firstTerminal, + ); + await waitForMessage( + collector.messages, + (message, index) => + index > wakeStartIndex && isAgentStreamEvent(message, agent.id, "turn_completed"), + "autonomous turn completion", + ); + const autonomousTerminalIndex = collector.messages.findIndex( + (message, index) => + index > wakeStartIndex && isAgentStreamEvent(message, agent.id, "turn_completed"), + ); + const finalTimeline = await runtime.client.fetchAgentTimeline(agent.id, { limit: 320 }); + expect(assistantTexts(finalTimeline.entries).at(-1)?.trim()).toBe(FINAL_TOKEN); + await waitForMessage( + collector.messages, + (message, index) => + index > autonomousTerminalIndex && isAgentStatus(message, agent.id, "idle"), + "idle after autonomous final output", + ); + const cancellationAfterWake = collector.messages + .slice(wakeStartIndex, autonomousTerminalIndex + 1) + .some((message) => isAgentStreamEvent(message, agent.id, "turn_canceled")); + const rootRunningIndex = collector.messages.findIndex((message) => + isAgentStatus(message, agent.id, "running"), + ); + const childRunningIndex = collector.messages.findIndex((message) => + isProviderSubagentStatus(message, agent.id, completedChild.id, "running"), + ); + const rootIdleIndex = collector.messages.findIndex( + (message, index) => index > childRunningIndex && isAgentStatus(message, agent.id, "idle"), + ); + const childCompletedIndex = collector.messages.findIndex((message) => + isProviderSubagentStatus(message, agent.id, completedChild.id, "completed"), + ); + expect(rootRunningIndex).toBeGreaterThan(-1); + expect(childRunningIndex).toBeGreaterThan(rootRunningIndex); + expect(rootIdleIndex).toBeGreaterThan(childRunningIndex); + expect(childCompletedIndex).toBeGreaterThan(rootIdleIndex); + expect(wakeStartIndex).toBeGreaterThan(-1); + expect(autonomousTerminalIndex).toBeGreaterThan(wakeStartIndex); + expect(cancellationAfterWake).toBe(false); + passed = true; + } catch (error) { + throw new Error( + `${error instanceof Error ? error.message : String(error)}. Retained real-test artifacts: ${runtime.artifacts}`, + { cause: error }, + ); + } finally { + collector.unsubscribe(); + await runtime.close(passed); + } +}, 2_100_000); + +function assistantTexts( + entries: Awaited< + ReturnType< + typeof import("../test-utils/daemon-client.js").DaemonClient.prototype.fetchAgentTimeline + > + >["entries"], +): string[] { + return entries.flatMap((entry) => + entry.item.type === "assistant_message" ? [entry.item.text] : [], + ); +} + +function isAgentStreamEvent( + message: SessionOutboundMessage, + agentId: string, + eventType: "turn_started" | "turn_completed" | "turn_failed" | "turn_canceled", +): boolean { + return ( + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === eventType + ); +} + +function isAgentStatus( + message: SessionOutboundMessage, + agentId: string, + status: "running" | "idle", +): boolean { + return ( + message.type === "agent_update" && + message.payload.kind === "upsert" && + message.payload.agent.id === agentId && + message.payload.agent.status === status + ); +} + +function isProviderSubagentStatus( + message: SessionOutboundMessage, + parentAgentId: string, + subagentId: string, + status: "running" | "completed", +): boolean { + return ( + message.type === "agent.provider_subagents.update" && + message.payload.kind === "upsert" && + message.payload.subagent.parentAgentId === parentAgentId && + message.payload.subagent.id === subagentId && + message.payload.subagent.status === status + ); +} + +function firstTerminalIndex(messages: SessionOutboundMessage[], agentId: string): number { + return messages.findIndex( + (message) => + isAgentStreamEvent(message, agentId, "turn_completed") || + isAgentStreamEvent(message, agentId, "turn_failed") || + isAgentStreamEvent(message, agentId, "turn_canceled"), + ); +} + +async function waitForCompletedChildWithToken( + client: import("../test-utils/daemon-client.js").DaemonClient, + parentAgentId: string, + token: string, +): Promise>["subagents"][number]> { + const deadline = Date.now() + 240_000; + while (Date.now() < deadline) { + const { subagents } = await client.listProviderSubagents(parentAgentId); + for (const subagent of subagents) { + const timeline = await client.fetchProviderSubagentTimeline(parentAgentId, subagent.id, { + direction: "tail", + limit: 0, + }); + if ( + subagent.status === "completed" && + (timeline.rows + .flatMap((row) => (row.item.type === "assistant_message" ? [row.item.text] : [])) + .join("") + .includes(token) || + timeline.rows.some( + (row) => + row.item.type === "tool_call" && + row.item.detail.type === "shell" && + row.item.detail.output?.includes(token), + )) + ) { + return subagent; + } + } + await new Promise((resolve) => setTimeout(resolve, 250)); + } + throw new Error(`Timed out waiting for completed OpenCode child output ${token}`); +} + +async function waitForMessage( + messages: SessionOutboundMessage[], + predicate: (message: SessionOutboundMessage, index: number) => boolean, + description: string, +): Promise { + const deadline = Date.now() + 240_000; + while (!messages.some(predicate)) { + if (Date.now() >= deadline) { + throw new Error(`Timed out waiting for ${description}`); + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } +} + +async function resolveSisyphusMode( + client: import("../test-utils/daemon-client.js").DaemonClient, + cwd: string, +): Promise { + const { modes } = await client.listProviderModes("opencode", { cwd }); + const matches = modes.filter((mode) => mode.label.toLowerCase().startsWith("sisyphus")); + if (matches.length !== 1 || !matches[0]) { + throw new Error(`Expected one OMO Sisyphus mode, received: ${JSON.stringify(modes)}`); + } + return matches[0].id; +} diff --git a/packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.test.ts b/packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.test.ts new file mode 100644 index 000000000..18fefb1d4 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, test } from "vitest"; + +import { + resolveCommandLaunch, + resolveWindowsHomeEnv, + withArtifactLocation, +} from "./opencode-omo-real-runtime.js"; + +describe("resolveCommandLaunch", () => { + test("launches Windows npm command shims through cmd.exe", () => { + expect(resolveCommandLaunch("C:\\fixture runtime\\omo.cmd", ["install"], "win32")).toEqual({ + command: process.env.COMSPEC || "cmd.exe", + args: ["/d", "/s", "/c", '"C:\\fixture runtime\\omo.cmd" install'], + shell: false, + }); + }); + + test("launches Windows executables directly", () => { + expect(resolveCommandLaunch("C:\\fixture runtime\\opencode.exe", [], "win32")).toEqual({ + command: "C:\\fixture runtime\\opencode.exe", + args: [], + shell: false, + }); + }); +}); + +test("includes retained artifacts in fixture setup failures", () => { + expect( + withArtifactLocation(new Error("Command timed out: npm-install.log"), "/tmp/artifacts").message, + ).toBe("Command timed out: npm-install.log. Retained real-test artifacts: /tmp/artifacts"); +}); + +test("isolates Windows native home resolution", () => { + expect(resolveWindowsHomeEnv("C:\\fixture\\home", "C:\\fixture\\tmp")).toEqual({ + USERPROFILE: "C:\\fixture\\home", + HOMEDRIVE: "C:", + HOMEPATH: "\\fixture\\home", + APPDATA: "C:\\fixture\\home\\AppData\\Roaming", + LOCALAPPDATA: "C:\\fixture\\home\\AppData\\Local", + TEMP: "C:\\fixture\\tmp", + TMP: "C:\\fixture\\tmp", + }); +}); diff --git a/packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.ts b/packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.ts new file mode 100644 index 000000000..923431f9b --- /dev/null +++ b/packages/server/src/server/daemon-e2e/opencode-omo-real-runtime.ts @@ -0,0 +1,520 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { Transform } from "node:stream"; + +import pino from "pino"; + +import { OpenCodeAgentClient } from "../agent/providers/opencode-agent.js"; +import { OpenCodeServerManager } from "../agent/providers/opencode/server-manager.js"; +import { terminateWithTreeKill } from "../../utils/tree-kill.js"; +import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; + +const PINNED_OPENCODE_VERSION = "1.18.9"; +const PINNED_OMO_VERSION = "4.19.3"; +const DEFAULT_OPENROUTER_MODEL = "openrouter/google/gemini-2.5-flash"; +const NO_AUTH_MODEL = "opencode/big-pickle"; +const COMMAND_TIMEOUT_MS = 180_000; + +interface RuntimePaths { + root: string; + home: string; + paseoHomeRoot: string; + xdgConfig: string; + xdgData: string; + xdgCache: string; + temporary: string; + npmCache: string; + runtime: string; + artifacts: string; + workspace: string; +} + +interface CommandInput { + command: string; + args: string[]; + cwd: string; + env: NodeJS.ProcessEnv; + artifactName: string; + artifacts: string; + secrets: string[]; + requiredOutput?: string; +} + +export interface OpenCodeOmoRealRuntime { + client: DaemonClient; + daemon: TestPaseoDaemon; + model: string; + workspace: string; + artifacts: string; + close: (passed: boolean) => Promise; +} + +export async function createOpenCodeOmoRealRuntime(): Promise { + const paths = createRuntimePaths(); + const openCodeVersion = + process.env.PASEO_REAL_OPENCODE_VERSION?.trim() || PINNED_OPENCODE_VERSION; + const omoVersion = process.env.PASEO_REAL_OMO_VERSION?.trim() || PINNED_OMO_VERSION; + const openRouterApiKey = process.env.OPENROUTER_API_KEY?.trim() || null; + const model = resolveModel(openRouterApiKey); + const secrets = collectEnvironmentSecrets(openRouterApiKey); + const runtimeEnv = buildRuntimeEnv(paths, openRouterApiKey); + const npmCli = process.env.npm_execpath; + if (!npmCli || !existsSync(npmCli)) { + throw new Error( + `The targeted npm script must provide npm_execpath. Retained real-test artifacts: ${paths.artifacts}`, + ); + } + + await runCommand({ + command: process.execPath, + args: [ + npmCli, + "install", + "--no-save", + "--no-package-lock", + "--prefix", + paths.runtime, + "--cache", + paths.npmCache, + `opencode-ai@${openCodeVersion}`, + `oh-my-openagent@${omoVersion}`, + ], + cwd: paths.root, + env: runtimeEnv, + artifactName: "npm-install.log", + artifacts: paths.artifacts, + secrets, + }); + + const executableSuffix = process.platform === "win32" ? ".cmd" : ""; + const openCodeCommand = path.join( + paths.runtime, + "node_modules", + ".bin", + `opencode${executableSuffix}`, + ); + const omoCommand = path.join(paths.runtime, "node_modules", ".bin", `omo${executableSuffix}`); + assertFixtureExecutable(openCodeCommand, `opencode-ai@${openCodeVersion}`); + assertFixtureExecutable(omoCommand, `oh-my-openagent@${omoVersion}`); + + await runCommand({ + command: omoCommand, + args: [ + "install", + "--no-tui", + "--platform=opencode", + "--claude=no", + "--openai=no", + "--gemini=no", + "--copilot=no", + "--opencode-zen=yes", + "--zai-coding-plan=no", + "--opencode-go=no", + "--kimi-for-coding=no", + ], + cwd: paths.workspace, + env: runtimeEnv, + artifactName: "omo-install.log", + artifacts: paths.artifacts, + secrets, + }); + pinOpenCodePlugin(paths.xdgConfig, omoVersion); + writeOmoModelConfig(paths.home, model); + + await runCommand({ + command: openCodeCommand, + args: ["agent", "list"], + cwd: paths.workspace, + env: runtimeEnv, + artifactName: "plugin-catalog-probe.log", + artifacts: paths.artifacts, + secrets, + }); + + if (model === NO_AUTH_MODEL) { + await runCommand({ + command: openCodeCommand, + args: ["run", "--model", model, "Reply with exactly: PASEO_BIG_PICKLE_PROBE_OK"], + cwd: paths.workspace, + env: runtimeEnv, + artifactName: "big-pickle-probe.log", + artifacts: paths.artifacts, + secrets, + requiredOutput: "PASEO_BIG_PICKLE_PROBE_OK", + }); + } + + const previousPaseoHome = process.env.PASEO_HOME; + let traceDestination: ReturnType | null = null; + let closeTrace: (() => void) | null = null; + let serverManager: OpenCodeServerManager | null = null; + let daemon: TestPaseoDaemon | null = null; + let client: DaemonClient | null = null; + try { + process.env.PASEO_HOME = path.join(paths.paseoHomeRoot, ".paseo"); + traceDestination = pino.destination({ + dest: path.join(paths.artifacts, "daemon.log"), + sync: true, + }); + const redactingTrace = new Transform({ + transform(chunk: Buffer, _encoding, callback) { + callback(null, redactSecrets(chunk.toString(), secrets)); + }, + }); + redactingTrace.pipe(traceDestination, { end: false }); + const logger = pino({ level: "trace" }, redactingTrace); + closeTrace = createTraceCloser(redactingTrace, traceDestination); + const runtimeSettings = { + command: { mode: "replace", argv: [openCodeCommand] }, + env: runtimeEnv, + } as const; + serverManager = new OpenCodeServerManager({ + logger, + runtimeSettings, + baseEnv: runtimeEnv, + resolveHomeDir: () => paths.home, + }); + const openCodeClient = new OpenCodeAgentClient(logger, runtimeSettings, { serverManager }); + daemon = await createTestPaseoDaemon({ + agentClients: { opencode: openCodeClient }, + logger, + paseoHomeRoot: paths.paseoHomeRoot, + staticDir: path.join(paths.root, "static"), + cleanup: false, + }); + client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + await client.connect(); + await client.fetchAgents({ subscribe: { subscriptionId: "opencode-omo-real" } }); + + return { + client, + daemon, + model, + workspace: paths.workspace, + artifacts: paths.artifacts, + close: async (passed) => { + try { + await client.close().catch(() => undefined); + await daemon.close().catch(() => undefined); + await serverManager.shutdown().catch(() => undefined); + closeTrace(); + if (passed) { + rmSync(paths.root, { recursive: true, force: true }); + } + } finally { + restoreEnvironment("PASEO_HOME", previousPaseoHome); + } + }, + }; + } catch (error) { + try { + await client?.close().catch(() => undefined); + await daemon?.close().catch(() => undefined); + await serverManager?.shutdown().catch(() => undefined); + if (closeTrace) { + closeTrace(); + } else { + traceDestination?.end(); + } + } finally { + restoreEnvironment("PASEO_HOME", previousPaseoHome); + } + throw withArtifactLocation(error, paths.artifacts); + } +} + +function createRuntimePaths(): RuntimePaths { + const root = mkdtempSync(path.join(tmpdir(), "paseo-real-opencode-omo-")); + const paths: RuntimePaths = { + root, + home: path.join(root, "home"), + paseoHomeRoot: path.join(root, "paseo-home"), + xdgConfig: path.join(root, "xdg", "config"), + xdgData: path.join(root, "xdg", "data"), + xdgCache: path.join(root, "xdg", "cache"), + temporary: path.join(root, "tmp"), + npmCache: path.join(root, "npm-cache"), + runtime: path.join(root, "runtime"), + artifacts: path.join(root, "artifacts"), + workspace: path.join(root, "workspace"), + }; + for (const directory of Object.values(paths)) { + mkdirSync(directory, { recursive: true }); + } + if (process.platform === "win32") { + const windowsHome = resolveWindowsHomeEnv(paths.home, paths.temporary); + mkdirSync(windowsHome.APPDATA, { recursive: true }); + mkdirSync(windowsHome.LOCALAPPDATA, { recursive: true }); + } + return paths; +} + +function resolveModel(openRouterApiKey: string | null): string { + const explicitModel = process.env.PASEO_REAL_OPENCODE_MODEL?.trim(); + if (explicitModel) { + return explicitModel; + } + return openRouterApiKey ? DEFAULT_OPENROUTER_MODEL : NO_AUTH_MODEL; +} + +function buildRuntimeEnv(paths: RuntimePaths, openRouterApiKey: string | null): NodeJS.ProcessEnv { + return { + LANG: process.env.LANG, + LC_ALL: process.env.LC_ALL, + TZ: process.env.TZ, + SystemRoot: process.env.SystemRoot, + COMSPEC: process.env.COMSPEC, + PATHEXT: process.env.PATHEXT, + NODE_EXTRA_CA_CERTS: process.env.NODE_EXTRA_CA_CERTS, + SSL_CERT_FILE: process.env.SSL_CERT_FILE, + SSL_CERT_DIR: process.env.SSL_CERT_DIR, + HOME: paths.home, + ...(process.platform === "win32" ? resolveWindowsHomeEnv(paths.home, paths.temporary) : {}), + PASEO_HOME: path.join(paths.paseoHomeRoot, ".paseo"), + XDG_CONFIG_HOME: paths.xdgConfig, + XDG_DATA_HOME: paths.xdgData, + XDG_CACHE_HOME: paths.xdgCache, + TMPDIR: paths.temporary, + npm_config_cache: paths.npmCache, + PATH: [path.join(paths.runtime, "node_modules", ".bin"), process.env.PATH] + .filter((entry): entry is string => Boolean(entry)) + .join(path.delimiter), + OPENCODE_DISABLE_AUTO_UPDATE: "1", + CI: "1", + FORCE_COLOR: "", + ...(openRouterApiKey ? { OPENROUTER_API_KEY: openRouterApiKey } : {}), + }; +} + +export function resolveWindowsHomeEnv( + home: string, + temporary: string, +): { + USERPROFILE: string; + HOMEDRIVE: string; + HOMEPATH: string; + APPDATA: string; + LOCALAPPDATA: string; + TEMP: string; + TMP: string; +} { + const root = path.win32.parse(home).root; + const relativeHome = path.win32.relative(root, home); + return { + USERPROFILE: home, + HOMEDRIVE: root.replace(/[\\/]$/, ""), + HOMEPATH: `\\${relativeHome}`, + APPDATA: path.win32.join(home, "AppData", "Roaming"), + LOCALAPPDATA: path.win32.join(home, "AppData", "Local"), + TEMP: temporary, + TMP: temporary, + }; +} + +async function runCommand(input: CommandInput): Promise { + const launch = resolveCommandLaunch(input.command, input.args); + const child = spawn(launch.command, launch.args, { + cwd: input.cwd, + env: input.env, + detached: process.platform !== "win32", + shell: launch.shell, + stdio: ["ignore", "pipe", "pipe"], + windowsHide: true, + }); + const stdout: Buffer[] = []; + const stderr: Buffer[] = []; + child.stdout?.on("data", (chunk: Buffer) => stdout.push(chunk)); + child.stderr?.on("data", (chunk: Buffer) => stderr.push(chunk)); + + let timeout: NodeJS.Timeout | null = null; + const exit = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>( + (resolve, reject) => { + child.once("error", reject); + child.once("close", (code, signal) => resolve({ code, signal })); + }, + ); + const timedOut = new Promise((_resolve, reject) => { + timeout = setTimeout( + () => reject(new Error(`Command timed out: ${input.artifactName}`)), + COMMAND_TIMEOUT_MS, + ); + }); + + let result: { code: number | null; signal: NodeJS.Signals | null }; + try { + result = await Promise.race([exit, timedOut]); + } catch (error) { + await stopChild(child); + writeCommandArtifact(input, stdout, stderr); + throw withArtifactLocation(error, input.artifacts); + } finally { + if (timeout) { + clearTimeout(timeout); + } + } + + const output = writeCommandArtifact(input, stdout, stderr); + if (result.code !== 0) { + throw new Error( + `Command failed (${result.code ?? result.signal ?? "unknown"}): ${input.artifactName}. Retained real-test artifacts: ${input.artifacts}`, + ); + } + if (input.requiredOutput && !output.includes(input.requiredOutput)) { + throw new Error( + `Command did not produce ${input.requiredOutput}: ${input.artifactName}. Retained real-test artifacts: ${input.artifacts}`, + ); + } +} + +export function resolveCommandLaunch( + command: string, + args: string[], + platform: NodeJS.Platform = process.platform, +): { command: string; args: string[]; shell: boolean } { + if (platform === "win32" && path.extname(command).toLowerCase() === ".cmd") { + return { + command: process.env.COMSPEC || "cmd.exe", + args: [ + "/d", + "/s", + "/c", + [ + quoteWindowsCommandArg(command, true), + ...args.map((arg) => quoteWindowsCommandArg(arg)), + ].join(" "), + ], + shell: false, + }; + } + return { + command, + args, + shell: false, + }; +} + +function quoteWindowsCommandArg(value: string, force = false): string { + if (!force && value.length > 0 && !/[\s"&|<>^()]/.test(value)) { + return value; + } + return `"${value.replaceAll('"', '""')}"`; +} + +function writeCommandArtifact(input: CommandInput, stdout: Buffer[], stderr: Buffer[]): string { + const combined = [Buffer.concat(stdout).toString(), Buffer.concat(stderr).toString()] + .filter((text) => text.length > 0) + .join("\n"); + writeFileSync( + path.join(input.artifacts, input.artifactName), + redactSecrets(combined, input.secrets), + ); + return combined; +} + +function redactSecrets(text: string, secrets: string[]): string { + return secrets.reduce((redacted, secret) => redacted.split(secret).join("[REDACTED]"), text); +} + +function collectEnvironmentSecrets(openRouterApiKey: string | null): string[] { + const secrets = Object.entries(process.env) + .filter(([name, value]) => /(?:AUTH|KEY|PASSWORD|SECRET|TOKEN)/i.test(name) && value) + .map(([, value]) => value) + .filter((value): value is string => typeof value === "string" && value.length >= 8); + if (openRouterApiKey) { + secrets.push(openRouterApiKey); + } + return [...new Set(secrets)]; +} + +function createTraceCloser( + trace: Transform, + destination: ReturnType, +): () => void { + let closed = false; + return () => { + if (closed) { + return; + } + closed = true; + trace.end(); + destination.flushSync(); + destination.end(); + }; +} + +async function stopChild(child: ChildProcess): Promise { + await terminateWithTreeKill(child, { + gracefulTimeoutMs: 2_000, + forceTimeoutMs: 2_000, + }); +} + +function assertFixtureExecutable(filePath: string, packageName: string): void { + if (!existsSync(filePath)) { + throw new Error(`Fixture install did not provide ${packageName} at ${filePath}`); + } +} + +function pinOpenCodePlugin(xdgConfig: string, omoVersion: string): void { + const configDirectory = path.join(xdgConfig, "opencode"); + const jsonPath = path.join(configDirectory, "opencode.json"); + mkdirSync(configDirectory, { recursive: true }); + writeFileSync( + jsonPath, + JSON.stringify( + { + $schema: "https://opencode.ai/config.json", + plugin: [`oh-my-openagent@${omoVersion}`], + }, + null, + 2, + ), + ); +} + +function writeOmoModelConfig(home: string, model: string): void { + const configDirectory = path.join(home, ".omo"); + mkdirSync(configDirectory, { recursive: true }); + writeFileSync( + path.join(configDirectory, "omo.jsonc"), + JSON.stringify( + { + "[opencode]": { + $schema: + "https://raw.githubusercontent.com/code-yeongyu/oh-my-openagent/dev/assets/omo.schema.json", + agents: { + sisyphus: { model, fallback_models: [{ model }] }, + "sisyphus-junior": { model, fallback_models: [{ model }] }, + }, + categories: { + quick: { model, fallback_models: [{ model }] }, + }, + opencode: { + background_task: { + providerConcurrency: { [model.split("/")[0] ?? "opencode"]: 2 }, + modelConcurrency: { [model]: 2 }, + }, + }, + }, + }, + null, + 2, + ), + ); +} + +function restoreEnvironment(name: string, previous: string | undefined): void { + if (previous === undefined) { + delete process.env[name]; + return; + } + process.env[name] = previous; +} + +export function withArtifactLocation(error: unknown, artifacts: string): Error { + const message = error instanceof Error ? error.message : String(error); + return new Error(`${message}. Retained real-test artifacts: ${artifacts}`, { cause: error }); +}