Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions packages/provider-opencode/src/server/adapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";

import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost";
import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost";
Expand Down Expand Up @@ -2410,6 +2411,62 @@ describe("OpenCodeAdapterV2", () => {
),
);

it.effect("reports an unanswered session abort as an interrupt timeout", () =>
Effect.gen(function* () {
const events = asyncEventStream();
const abortCalled = promiseGate<void>();
let abortCalls = 0;
const harness = yield* makeOpenCodeRuntimeHarness("abort-timeout", "root", {
event: {
subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => {
options.signal?.addEventListener("abort", () => events.close(), { once: true });
return { stream: events.stream };
},
},
session: {
create: async () => ({ data: { id: "root", time: { created: 1, updated: 1 } } }),
get: async () => ({ data: { id: "root", time: { created: 1, updated: 1 } } }),
promptAsync: async () => ({ data: true }),
status: async () => ({ data: { root: { type: "busy" } } }),
messages: async () => ({ data: [] }),
children: async () => ({ data: [] }),
abort: async (_input: unknown, options?: { readonly signal?: AbortSignal }) => {
// Only the Stop request hangs; session teardown still gets an answer.
if (++abortCalls > 1) return { data: true };
abortCalled.resolve();
await new Promise<void>((_resolve, reject) =>
options?.signal?.addEventListener("abort", () => reject(options.signal!.reason), {
once: true,
}),
);
return { data: true };
},
},
mcp: { add: async () => ({ data: true }) },
});
yield* harness.startTurn();
const snapshot = yield* harness.runtime.readThreadSnapshot({
providerThread: harness.providerThread,
});
const turn = snapshot.providerTurns.at(-1)!;
const interrupt = yield* harness.runtime
.interruptTurn({ providerThread: harness.providerThread, providerTurnId: turn.id })
.pipe(Effect.flip, Effect.forkScoped);
yield* Effect.promise(() => abortCalled.promise);
yield* TestClock.adjust("10 seconds");

const error = yield* Fiber.join(interrupt);
assert.strictEqual(error._tag, "ProviderAdapterInterruptTimeoutError");
assert.deepInclude(error, {
providerThreadId: harness.providerThread.id,
providerTurnId: turn.id,
});
}).pipe(
Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)),
Effect.scoped,
),
);

it.effect("fails an active turn when the OpenCode event stream ends cleanly", () =>
Effect.gen(function* () {
const nativeEvents = asyncEventStream();
Expand Down
31 changes: 22 additions & 9 deletions packages/provider-opencode/src/server/adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ export const OPENCODE_PROVIDER = ProviderDriverKind.make("opencode");
export const OPENCODE_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(OPENCODE_PROVIDER);
export const OPENCODE_SDK_PROTOCOL = "opencode-sdk.sse" as const;
const DEFAULT_OPENCODE_SETTINGS = Schema.decodeSync(OpenCodeSettings)({});
const OPENCODE_STOP_CONFIRM_TIMEOUT_MS = 10_000;

const makeOpenCodeMessageId = Effect.fnUntraced(function* (
clock: Ref.Ref<{ readonly epochMillis: number; readonly counter: number }>,
Expand Down Expand Up @@ -3488,7 +3489,18 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function
yield* sdkCall("session.abort", { sessionID: sessionId }, (signal) =>
client.session.abort({ sessionID: sessionId }, { signal }),
).pipe(
Effect.timeout("10 seconds"),
Effect.timeoutOrElse({
duration: OPENCODE_STOP_CONFIRM_TIMEOUT_MS,
orElse: () =>
Effect.fail(
new ProviderAdapter.ProviderAdapterInterruptTimeoutError({
driver: OPENCODE_PROVIDER,
providerThreadId: interruptInput.providerThread.id,
providerTurnId: interruptInput.providerTurnId,
timeoutMs: OPENCODE_STOP_CONFIRM_TIMEOUT_MS,
}),
),
}),
Effect.catch((cause) =>
turn.finalized
? Effect.void
Expand All @@ -3503,14 +3515,15 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function
},
(effect, interruptInput) =>
effect.pipe(
Effect.mapError(
(cause) =>
new ProviderAdapter.ProviderAdapterInterruptError({
driver: OPENCODE_PROVIDER,
providerThreadId: interruptInput.providerThread.id,
providerTurnId: interruptInput.providerTurnId,
cause,
}),
Effect.mapError((cause) =>
cause._tag === "ProviderAdapterInterruptTimeoutError"
? cause
: new ProviderAdapter.ProviderAdapterInterruptError({
driver: OPENCODE_PROVIDER,
providerThreadId: interruptInput.providerThread.id,
providerTurnId: interruptInput.providerTurnId,
cause,
}),
),
),
),
Expand Down
Loading