import { assert, describe, it } from "@effect/vitest";import type { ToolPart } from "@opencode-ai/sdk/v2";import { NodeId, OpenCodeSettings, ProjectId, ProviderInstanceId, ProviderSessionId, ProviderThreadId, ProviderTurnId, RunAttemptId, RunId, MessageId, ThreadId, type OrchestrationV2ProviderTurn,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Fiber from "effect/Fiber";import * as Exit from "effect/Exit";import * as Queue from "effect/Queue";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as TestClock from "effect/testing/TestClock"; import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts";import type { OpenCodeRuntimeShape } from "../../provider/opencodeRuntime.ts";import type { ServerConfig } from "../../config.ts";import { IdAllocatorV2, layer as idAllocatorLayer } from "../IdAllocator.ts"; import { advanceOpenCodePromptAdmission, cancelOpenCodePromptAdmission, openCodeBoundaryAfterProviderTurn, openCodeChildPermissionRules, openCodePermissionRules, openCodePermissionRequestKind, openCodeToolProjectionKind, makeOpenCodeProtocolLogger, makeOpenCodeAdapterV2, OPENCODE_PROVIDER, OpenCodeProviderCapabilitiesV2, reconcileOpenCodePromptAdmissionStatus,} from "./OpenCodeAdapterV2.ts";import { ProviderAdapterV2RuntimePolicy } from "../ProviderAdapter.ts"; const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));const OPEN_CODE_TEST_SETTINGS = Schema.decodeSync(OpenCodeSettings)({ serverUrl: "http://test.invalid",}); function promiseGate<A>() { let resolve!: (value: A) => void; const promise = new Promise<A>((done) => { resolve = done; }); return { promise, resolve };} function asyncEventStream() { const values: Array<{ value: unknown; handled: ReturnType<typeof promiseGate<void>> }> = []; const waiters: Array<(value: IteratorResult<unknown>) => void> = []; let previousHandled: ReturnType<typeof promiseGate<void>> | undefined; return { close() { for (const waiter of waiters.splice(0)) waiter({ done: true, value: undefined }); }, push(value: unknown) { const handled = promiseGate<void>(); const waiter = waiters.shift(); if (waiter) { previousHandled = handled; waiter({ done: false, value }); } else values.push({ value, handled }); return handled.promise; }, stream: { [Symbol.asyncIterator]() { return { next: () => { previousHandled?.resolve(); const entry = values.shift(); if (entry !== undefined) { previousHandled = entry.handled; return Promise.resolve({ done: false as const, value: entry.value }); } return new Promise<IteratorResult<unknown>>((resolve) => waiters.push(resolve)); }, }; }, }, };} function runtimePolicy( runtimeMode: ProviderAdapterV2RuntimePolicy["runtimeMode"], override: Partial<ProviderAdapterV2RuntimePolicy> = {},): ProviderAdapterV2RuntimePolicy { return ProviderAdapterV2RuntimePolicy.make({ runtimeMode, interactionMode: "default", cwd: null, ...override, });} function permissionAction(rules: ReturnType<typeof openCodePermissionRules>, permission: string) { return rules.findLast((rule) => rule.permission === "*" || rule.permission === permission) ?.action;} function providerTurn(input: { readonly id: string; readonly ordinal: number; readonly nativeId: string | null;}): OrchestrationV2ProviderTurn { return { id: ProviderTurnId.make(input.id), providerThreadId: ProviderThreadId.make("provider-thread:opencode-test"), nodeId: NodeId.make(`node:${input.id}`), runAttemptId: null, nativeTurnRef: input.nativeId === null ? null : { driver: OPENCODE_PROVIDER, nativeId: input.nativeId, strength: "weak" }, ordinal: input.ordinal, status: "completed", startedAt: null, completedAt: null, };} const makeOpenCodeRuntimeHarness = Effect.fn("makeOpenCodeRuntimeHarness")(function* ( suffix: string, nativeSessionId: string, client: object,) { const idAllocator = yield* IdAllocatorV2; const instanceId = ProviderInstanceId.make(`opencode-${suffix}`); const threadId = ThreadId.make(`thread-opencode-${suffix}`); const modelSelection = { instanceId, model: "anthropic/claude-sonnet", options: [], }; const policy = runtimePolicy("full-access", { cwd: "/workspace" }); const adapter = makeOpenCodeAdapterV2({ instanceId, settings: OPEN_CODE_TEST_SETTINGS, environment: {}, runtime: { connectToOpenCodeServer: () => Effect.succeed({ url: "http://test.invalid", external: true }), createOpenCodeSdkClient: () => client, } as unknown as OpenCodeRuntimeShape, idAllocator, serverConfig: { cwd: "/workspace", attachmentsDir: "/tmp/attachments", } as ServerConfig["Service"], }); const runtime = yield* adapter.openSession({ threadId, providerSessionId: ProviderSessionId.make(`session-opencode-${suffix}`), modelSelection, runtimePolicy: policy, }); const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy: policy, }); const now = yield* DateTime.now; const startTurn = (text = "hello") => runtime.startTurn({ appThread: { id: threadId, projectId: ProjectId.make(`project-opencode-${suffix}`), title: suffix, providerInstanceId: modelSelection.instanceId, modelSelection, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, activeProviderThreadId: providerThread.id, lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, forkedFrom: null, createdBy: "user", creationSource: "web", createdAt: now, updatedAt: now, archivedAt: null, settledOverride: null, settledAt: null, lastVisitedAt: null, deletedAt: null, }, threadId, runId: RunId.make(`run-opencode-${suffix}`), runOrdinal: 1, providerTurnOrdinal: 1, attemptId: RunAttemptId.make(`attempt-opencode-${suffix}`), rootNodeId: NodeId.make(`node-opencode-${suffix}`), providerThread, message: { createdBy: "user", creationSource: "web", messageId: MessageId.make(`message-opencode-${suffix}`), text, attachments: [], }, modelSelection, runtimePolicy: policy, }); return { nativeSessionId, now, policy, providerThread, runId: RunId.make(`run-opencode-${suffix}`), runtime, startTurn, threadId, };}); describe("OpenCodeAdapterV2", () => { it.effect( "preserves tool lifecycle, approval kinds, and late assistant text without cached tool payloads", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); const nativeSessionId = "native-opencode-tool-lifecycle"; const harness = yield* makeOpenCodeRuntimeHarness("tool-lifecycle", nativeSessionId, { event: { subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); return { stream: nativeEvents.stream }; }, }, session: { create: async () => ({ data: { id: nativeSessionId, time: { created: 1, updated: 1 } }, }), promptAsync: async () => ({ data: true }), }, }); yield* harness.startTurn(); const received = yield* harness.runtime.events.pipe( Stream.takeUntil( (event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction", ), Stream.runCollect, Effect.forkScoped, ); const states = [ { status: "pending", input: { command: "pwd" }, raw: "" }, { status: "running", input: { command: "pwd" }, title: "Working directory", time: { start: 1 }, }, { status: "completed", input: { command: "pwd" }, output: "/repo\n", title: "Working directory", metadata: {}, time: { start: 1, end: 2 }, }, { status: "error", input: { command: "pwd" }, error: "Command failed", time: { start: 3, end: 4 }, }, ] satisfies ReadonlyArray<ToolPart["state"]>; for (const state of states) { yield* Effect.promise(() => nativeEvents.push({ type: "message.part.updated", properties: { sessionID: nativeSessionId, part: { id: state.status === "error" ? "part-failed" : "part-working", sessionID: nativeSessionId, messageID: "late-assistant", type: "tool", callID: state.status === "error" ? "call-failed" : "call-working", tool: "bash", state, }, }, }), ); } yield* Effect.promise(() => nativeEvents.push({ type: "message.part.updated", properties: { sessionID: nativeSessionId, part: { id: "part-read", sessionID: nativeSessionId, messageID: "late-assistant", type: "tool", callID: "call-read", tool: "read", state: { status: "pending", input: { filePath: "/outside/file" }, raw: "" }, }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "permission.asked", properties: { id: "read-permission", sessionID: nativeSessionId, permission: "external_directory", patterns: ["/outside/*"], always: [], metadata: {}, tool: { messageID: "late-assistant", callID: "call-read" }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "message.part.updated", properties: { sessionID: nativeSessionId, part: { id: "part-text", sessionID: nativeSessionId, messageID: "late-assistant", type: "text", text: "Tool results received", time: { start: 4 }, }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "message.updated", properties: { sessionID: nativeSessionId, info: { id: "late-assistant", sessionID: nativeSessionId, role: "assistant", time: { created: 1, completed: 4 }, }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "session.compacted", properties: { sessionID: nativeSessionId }, }), ); const events = yield* Fiber.join(received); const items = events.flatMap((event) => event.type === "turn_item.updated" ? [event.turnItem] : [], ); const commands = items.filter((item) => item.type === "command_execution"); assert.deepEqual( commands.map((item) => item.status), ["pending", "running", "completed", "failed"], ); assert.deepEqual( commands.map((item) => item.input), ["pwd", "pwd", "pwd", "pwd"], ); assert.equal(commands[2]?.output, "/repo\n"); assert.equal(commands[3]?.output, "Command failed"); assert.isTrue( events.some( (event) => event.type === "runtime_request.updated" && event.runtimeRequest.kind === "file-read", ), ); const assistant = items.filter((item) => item.type === "assistant_message"); assert.equal(assistant.at(-1)?.text, "Tool results received"); assert.equal(assistant.at(-1)?.status, "completed"); }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped), ); it.effect("compacts with the native summarize API and emits a completed compaction", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); const summarizeCalls: Array<unknown> = []; let promptCalls = 0; const harness = yield* makeOpenCodeRuntimeHarness("compact", "native-opencode-compact", { event: { subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); return { stream: nativeEvents.stream }; }, }, session: { create: async () => ({ data: { id: "native-opencode-compact", time: { created: 1, updated: 1 } }, }), summarize: async (input: unknown) => { summarizeCalls.push(input); return { data: true }; }, promptAsync: async () => { promptCalls += 1; return { data: true }; }, }, }); yield* harness.startTurn("/compact"); const events = yield* harness.runtime.events.pipe( Stream.takeUntil((event) => event.type === "turn.terminal"), Stream.runCollect, ); assert.deepEqual(summarizeCalls, [ { sessionID: "native-opencode-compact", providerID: "anthropic", modelID: "claude-sonnet", auto: false, }, ]); assert.equal(promptCalls, 0); assert.equal( events.filter( (event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction" && event.turnItem.status === "completed", ).length, 1, ); assert.isTrue( events.some((event) => event.type === "turn.terminal" && event.status === "completed"), ); }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped), ); it.effect("keeps a newly admitted prompt alive across stale idle and delayed busy evidence", () => Effect.gen(function* () { const idAllocator = yield* IdAllocatorV2; const nativeEvents = asyncEventStream(); const prompt = promiseGate<void>(); const promptStarted = promiseGate<void>(); const steerPrompt = promiseGate<void>(); const steerPromptStarted = promiseGate<void>(); const stopPromptStarted = promiseGate<void>(); const statusCalled = promiseGate<void>(); const steerStatusCalled = promiseGate<void>(); const abortCalled = promiseGate<void>(); let status: "idle" | "busy" = "busy"; let promptCalls = 0; let statusCalls = 0; const promptInputs: Array<{ readonly messageID?: string }> = []; const client = { event: { subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); return { stream: nativeEvents.stream }; }, }, session: { create: async () => ({ data: { id: "native-opencode-race", time: { created: 1, updated: 1 } }, }), promptAsync: async ( input: { readonly messageID?: string }, options?: { readonly signal?: AbortSignal }, ) => { promptInputs.push(input); promptCalls += 1; if (promptCalls === 1) { promptStarted.resolve(); await prompt.promise; } else if (promptCalls === 2) { steerPromptStarted.resolve(); await steerPrompt.promise; } else { stopPromptStarted.resolve(); await new Promise<void>((_resolve, reject) => options?.signal?.addEventListener("abort", () => reject(options.signal!.reason), { once: true, }), ); } return { data: true }; }, status: async () => ({ data: (() => { statusCalls += 1; if (statusCalls === 1) statusCalled.resolve();Lines 1–500 of 1394. Load more
import { assert, describe, it } from "@effect/vitest"; import type { ToolPart } from "@opencode-ai/sdk/v2"; import { NodeId, OpenCodeSettings, ProjectId, ProviderInstanceId, ProviderSessionId, ProviderThreadId, ProviderTurnId, RunAttemptId, RunId, MessageId, ThreadId, type OrchestrationV2ProviderTurn, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Exit from "effect/Exit"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import type { EventNdjsonLogger } from "../../provider/Layers/EventNdjsonLogger.ts"; import type { OpenCodeRuntimeShape } from "../../provider/opencodeRuntime.ts"; import type { ServerConfig } from "../../config.ts"; import { IdAllocatorV2, layer as idAllocatorLayer } from "../IdAllocator.ts"; import { advanceOpenCodePromptAdmission, cancelOpenCodePromptAdmission, openCodeBoundaryAfterProviderTurn, openCodeChildPermissionRules, openCodePermissionRules, openCodePermissionRequestKind, openCodeToolProjectionKind, makeOpenCodeProtocolLogger, makeOpenCodeAdapterV2, OPENCODE_PROVIDER, OpenCodeProviderCapabilitiesV2, reconcileOpenCodePromptAdmissionStatus, } from "./OpenCodeAdapterV2.ts"; import { ProviderAdapterV2RuntimePolicy } from "../ProviderAdapter.ts"; const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const OPEN_CODE_TEST_SETTINGS = Schema.decodeSync(OpenCodeSettings)({ serverUrl: "http://test.invalid", }); function promiseGate<A>() { let resolve!: (value: A) => void; const promise = new Promise<A>((done) => { resolve = done; }); return { promise, resolve }; } function asyncEventStream() { const values: Array<{ value: unknown; handled: ReturnType<typeof promiseGate<void>> }> = []; const waiters: Array<(value: IteratorResult<unknown>) => void> = []; let previousHandled: ReturnType<typeof promiseGate<void>> | undefined; return { close() { for (const waiter of waiters.splice(0)) waiter({ done: true, value: undefined }); }, push(value: unknown) { const handled = promiseGate<void>(); const waiter = waiters.shift(); if (waiter) { previousHandled = handled; waiter({ done: false, value }); } else values.push({ value, handled }); return handled.promise; }, stream: { [Symbol.asyncIterator]() { return { next: () => { previousHandled?.resolve(); const entry = values.shift(); if (entry !== undefined) { previousHandled = entry.handled; return Promise.resolve({ done: false as const, value: entry.value }); } return new Promise<IteratorResult<unknown>>((resolve) => waiters.push(resolve)); }, }; }, }, }; } function runtimePolicy( runtimeMode: ProviderAdapterV2RuntimePolicy["runtimeMode"], override: Partial<ProviderAdapterV2RuntimePolicy> = {}, ): ProviderAdapterV2RuntimePolicy { return ProviderAdapterV2RuntimePolicy.make({ runtimeMode, interactionMode: "default", cwd: null, ...override, }); } function permissionAction(rules: ReturnType<typeof openCodePermissionRules>, permission: string) { return rules.findLast((rule) => rule.permission === "*" || rule.permission === permission) ?.action; } function providerTurn(input: { readonly id: string; readonly ordinal: number; readonly nativeId: string | null; }): OrchestrationV2ProviderTurn { return { id: ProviderTurnId.make(input.id), providerThreadId: ProviderThreadId.make("provider-thread:opencode-test"), nodeId: NodeId.make(`node:${input.id}`), runAttemptId: null, nativeTurnRef: input.nativeId === null ? null : { driver: OPENCODE_PROVIDER, nativeId: input.nativeId, strength: "weak" }, ordinal: input.ordinal, status: "completed", startedAt: null, completedAt: null, }; } const makeOpenCodeRuntimeHarness = Effect.fn("makeOpenCodeRuntimeHarness")(function* ( suffix: string, nativeSessionId: string, client: object, ) { const idAllocator = yield* IdAllocatorV2; const instanceId = ProviderInstanceId.make(`opencode-${suffix}`); const threadId = ThreadId.make(`thread-opencode-${suffix}`); const modelSelection = { instanceId, model: "anthropic/claude-sonnet", options: [], }; const policy = runtimePolicy("full-access", { cwd: "/workspace" }); const adapter = makeOpenCodeAdapterV2({ instanceId, settings: OPEN_CODE_TEST_SETTINGS, environment: {}, runtime: { connectToOpenCodeServer: () => Effect.succeed({ url: "http://test.invalid", external: true }), createOpenCodeSdkClient: () => client, } as unknown as OpenCodeRuntimeShape, idAllocator, serverConfig: { cwd: "/workspace", attachmentsDir: "/tmp/attachments", } as ServerConfig["Service"], }); const runtime = yield* adapter.openSession({ threadId, providerSessionId: ProviderSessionId.make(`session-opencode-${suffix}`), modelSelection, runtimePolicy: policy, }); const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy: policy, }); const now = yield* DateTime.now; const startTurn = (text = "hello") => runtime.startTurn({ appThread: { id: threadId, projectId: ProjectId.make(`project-opencode-${suffix}`), title: suffix, providerInstanceId: modelSelection.instanceId, modelSelection, runtimeMode: "full-access", interactionMode: "default", branch: null, worktreePath: null, activeProviderThreadId: providerThread.id, lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, forkedFrom: null, createdBy: "user", creationSource: "web", createdAt: now, updatedAt: now, archivedAt: null, settledOverride: null, settledAt: null, lastVisitedAt: null, deletedAt: null, }, threadId, runId: RunId.make(`run-opencode-${suffix}`), runOrdinal: 1, providerTurnOrdinal: 1, attemptId: RunAttemptId.make(`attempt-opencode-${suffix}`), rootNodeId: NodeId.make(`node-opencode-${suffix}`), providerThread, message: { createdBy: "user", creationSource: "web", messageId: MessageId.make(`message-opencode-${suffix}`), text, attachments: [], }, modelSelection, runtimePolicy: policy, }); return { nativeSessionId, now, policy, providerThread, runId: RunId.make(`run-opencode-${suffix}`), runtime, startTurn, threadId, }; }); describe("OpenCodeAdapterV2", () => { it.effect( "preserves tool lifecycle, approval kinds, and late assistant text without cached tool payloads", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); const nativeSessionId = "native-opencode-tool-lifecycle"; const harness = yield* makeOpenCodeRuntimeHarness("tool-lifecycle", nativeSessionId, { event: { subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); return { stream: nativeEvents.stream }; }, }, session: { create: async () => ({ data: { id: nativeSessionId, time: { created: 1, updated: 1 } }, }), promptAsync: async () => ({ data: true }), }, }); yield* harness.startTurn(); const received = yield* harness.runtime.events.pipe( Stream.takeUntil( (event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction", ), Stream.runCollect, Effect.forkScoped, ); const states = [ { status: "pending", input: { command: "pwd" }, raw: "" }, { status: "running", input: { command: "pwd" }, title: "Working directory", time: { start: 1 }, }, { status: "completed", input: { command: "pwd" }, output: "/repo\n", title: "Working directory", metadata: {}, time: { start: 1, end: 2 }, }, { status: "error", input: { command: "pwd" }, error: "Command failed", time: { start: 3, end: 4 }, }, ] satisfies ReadonlyArray<ToolPart["state"]>; for (const state of states) { yield* Effect.promise(() => nativeEvents.push({ type: "message.part.updated", properties: { sessionID: nativeSessionId, part: { id: state.status === "error" ? "part-failed" : "part-working", sessionID: nativeSessionId, messageID: "late-assistant", type: "tool", callID: state.status === "error" ? "call-failed" : "call-working", tool: "bash", state, }, }, }), ); } yield* Effect.promise(() => nativeEvents.push({ type: "message.part.updated", properties: { sessionID: nativeSessionId, part: { id: "part-read", sessionID: nativeSessionId, messageID: "late-assistant", type: "tool", callID: "call-read", tool: "read", state: { status: "pending", input: { filePath: "/outside/file" }, raw: "" }, }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "permission.asked", properties: { id: "read-permission", sessionID: nativeSessionId, permission: "external_directory", patterns: ["/outside/*"], always: [], metadata: {}, tool: { messageID: "late-assistant", callID: "call-read" }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "message.part.updated", properties: { sessionID: nativeSessionId, part: { id: "part-text", sessionID: nativeSessionId, messageID: "late-assistant", type: "text", text: "Tool results received", time: { start: 4 }, }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "message.updated", properties: { sessionID: nativeSessionId, info: { id: "late-assistant", sessionID: nativeSessionId, role: "assistant", time: { created: 1, completed: 4 }, }, }, }), ); yield* Effect.promise(() => nativeEvents.push({ type: "session.compacted", properties: { sessionID: nativeSessionId }, }), ); const events = yield* Fiber.join(received); const items = events.flatMap((event) => event.type === "turn_item.updated" ? [event.turnItem] : [], ); const commands = items.filter((item) => item.type === "command_execution"); assert.deepEqual( commands.map((item) => item.status), ["pending", "running", "completed", "failed"], ); assert.deepEqual( commands.map((item) => item.input), ["pwd", "pwd", "pwd", "pwd"], ); assert.equal(commands[2]?.output, "/repo\n"); assert.equal(commands[3]?.output, "Command failed"); assert.isTrue( events.some( (event) => event.type === "runtime_request.updated" && event.runtimeRequest.kind === "file-read", ), ); const assistant = items.filter((item) => item.type === "assistant_message"); assert.equal(assistant.at(-1)?.text, "Tool results received"); assert.equal(assistant.at(-1)?.status, "completed"); }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped), ); it.effect("compacts with the native summarize API and emits a completed compaction", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); const summarizeCalls: Array<unknown> = []; let promptCalls = 0; const harness = yield* makeOpenCodeRuntimeHarness("compact", "native-opencode-compact", { event: { subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); return { stream: nativeEvents.stream }; }, }, session: { create: async () => ({ data: { id: "native-opencode-compact", time: { created: 1, updated: 1 } }, }), summarize: async (input: unknown) => { summarizeCalls.push(input); return { data: true }; }, promptAsync: async () => { promptCalls += 1; return { data: true }; }, }, }); yield* harness.startTurn("/compact"); const events = yield* harness.runtime.events.pipe( Stream.takeUntil((event) => event.type === "turn.terminal"), Stream.runCollect, ); assert.deepEqual(summarizeCalls, [ { sessionID: "native-opencode-compact", providerID: "anthropic", modelID: "claude-sonnet", auto: false, }, ]); assert.equal(promptCalls, 0); assert.equal( events.filter( (event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction" && event.turnItem.status === "completed", ).length, 1, ); assert.isTrue( events.some((event) => event.type === "turn.terminal" && event.status === "completed"), ); }).pipe(Effect.provide(idAllocatorLayer), Effect.scoped), ); it.effect("keeps a newly admitted prompt alive across stale idle and delayed busy evidence", () => Effect.gen(function* () { const idAllocator = yield* IdAllocatorV2; const nativeEvents = asyncEventStream(); const prompt = promiseGate<void>(); const promptStarted = promiseGate<void>(); const steerPrompt = promiseGate<void>(); const steerPromptStarted = promiseGate<void>(); const stopPromptStarted = promiseGate<void>(); const statusCalled = promiseGate<void>(); const steerStatusCalled = promiseGate<void>(); const abortCalled = promiseGate<void>(); let status: "idle" | "busy" = "busy"; let promptCalls = 0; let statusCalls = 0; const promptInputs: Array<{ readonly messageID?: string }> = []; const client = { event: { subscribe: async (_input: unknown, options: { signal?: AbortSignal }) => { options.signal?.addEventListener("abort", () => nativeEvents.close(), { once: true }); return { stream: nativeEvents.stream }; }, }, session: { create: async () => ({ data: { id: "native-opencode-race", time: { created: 1, updated: 1 } }, }), promptAsync: async ( input: { readonly messageID?: string }, options?: { readonly signal?: AbortSignal }, ) => { promptInputs.push(input); promptCalls += 1; if (promptCalls === 1) { promptStarted.resolve(); await prompt.promise; } else if (promptCalls === 2) { steerPromptStarted.resolve(); await steerPrompt.promise; } else { stopPromptStarted.resolve(); await new Promise<void>((_resolve, reject) => options?.signal?.addEventListener("abort", () => reject(options.signal!.reason), { once: true, }), ); } return { data: true }; }, status: async () => ({ data: (() => { statusCalls += 1; if (statusCalls === 1) statusCalled.resolve();