From ddbd119dcb8a53625052b2d6072c8bad9914f9bf Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 20:12:51 -0400 Subject: [PATCH] test: drop AppRuntime usage from tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes all direct AppRuntime usage from packages/opencode/test — `grep -r AppRuntime packages/opencode/test` now returns nothing. Two patterns are applied: 1. Event tests rewritten in the httpapi-cors.test.ts style. The two /event SSE tests now serve HttpApiApp.routes on NodeHttpServer.layerTest and hit them via HttpClient. Pub/sub identity with the in-process routes is preserved via a new opt-in `testEffectShared` (in test/lib/effect.ts) that builds the test layer through the shared process-wide memoMap so Bus.defaultLayer resolves to the same Bus.Service the routes subscribed to. The SSE reader helpers move to test/lib/sse.ts and use HttpClient + Effect.Stream + Queue. The D7 diagnostic case is removed: the AppRuntime-vs-test-runtime distinction it diagnosed no longer exists. 2. Surgical swap in the remaining four files (provider/{amazon-bedrock,provider}, session/llm, control-plane/workspace). Each `AppRuntime.runPromise(...)` becomes a module-level `ManagedRuntime.make(Service.defaultLayer, { memoMap })`. The shared memoMap preserves service identity, so behavior is unchanged. Tests: event (9/9), amazon-bedrock (19/19), provider (84/84), workspace (35/35); 147 pass across the 5 affected files. The 3 pre-existing failures in session/llm.test.ts are independent of this change (verified by stashing the diff). --- .../test/control-plane/workspace.test.ts | 18 +- packages/opencode/test/lib/effect.ts | 34 +- packages/opencode/test/lib/sse.ts | 63 ++ .../test/provider/amazon-bedrock.test.ts | 577 ++++++------------ .../opencode/test/provider/provider.test.ts | 10 +- .../server/httpapi-event-diagnostics.test.ts | 176 ++---- .../test/server/httpapi-event.test.ts | 180 +++--- packages/opencode/test/session/llm.test.ts | 8 +- 8 files changed, 445 insertions(+), 621 deletions(-) create mode 100644 packages/opencode/test/lib/sse.ts diff --git a/packages/opencode/test/control-plane/workspace.test.ts b/packages/opencode/test/control-plane/workspace.test.ts index ae8fb97a51a..f4a6541c5df 100644 --- a/packages/opencode/test/control-plane/workspace.test.ts +++ b/packages/opencode/test/control-plane/workspace.test.ts @@ -5,7 +5,8 @@ import Http from "node:http" import path from "node:path" import { setTimeout as delay } from "node:timers/promises" import { NodeHttpServer } from "@effect/platform-node" -import { Effect, Layer, Schema } from "effect" +import { Effect, Layer, ManagedRuntime, Schema } from "effect" +import { memoMap } from "@opencode-ai/core/effect/memo-map" import { FetchHttpClient, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" import { eq } from "drizzle-orm" import { AppFileSystem } from "@opencode-ai/core/filesystem" @@ -29,8 +30,8 @@ import { WorkspaceID } from "../../src/control-plane/schema" import { WorkspaceTable } from "../../src/control-plane/workspace.sql" import type { Target, WorkspaceAdapter, WorkspaceInfo } from "../../src/control-plane/types" import * as Workspace from "../../src/control-plane/workspace" -import { AppRuntime } from "@/effect/app-runtime" import { InstanceStore } from "@/project/instance-store" +import { InstanceLayer } from "@/project/instance-layer" import { InstanceBootstrap } from "@/project/bootstrap" import { Auth } from "@/auth" import { SessionPrompt } from "@/session/prompt" @@ -121,9 +122,14 @@ afterEach(async () => { await resetDatabase() }) +const runtime = ManagedRuntime.make( + Layer.mergeAll(InstanceLayer.layer, Workspace.defaultLayer, SessionNs.defaultLayer), + { memoMap }, +) + async function withInstance(fn: (ctx: InstanceContext) => T | Promise) { await using tmp = await tmpdir({ git: true }) - const ctx = await AppRuntime.runPromise(InstanceStore.Service.use((store) => store.load({ directory: tmp.path }))) + const ctx = await runtime.runPromise(InstanceStore.Service.use((store) => store.load({ directory: tmp.path }))) return await context.provide(ctx, () => fn(ctx)) } @@ -149,7 +155,7 @@ function currentInstance() { const runWorkspace = (effect: Effect.Effect) => { const ctx = currentInstance() - return AppRuntime.runPromise(ctx ? effect.pipe(Effect.provideService(InstanceRef, ctx)) : effect) + return runtime.runPromise(ctx ? effect.pipe(Effect.provideService(InstanceRef, ctx)) : effect) } const createWorkspace = (input: Workspace.CreateInput) => runWorkspace(Workspace.Service.use((workspace) => workspace.create(input))) @@ -932,12 +938,12 @@ describe("workspace CRUD", () => { const previous = workspaceInfo(projectID, previousType) insertWorkspace(previous) registerAdapter(projectID, previousType, localAdapter(workspaceTmp.path, { createDir: false }).adapter) - const session = await AppRuntime.runPromise( + const session = await runtime.runPromise( SessionNs.Service.use((svc) => svc.create({})).pipe(Effect.provideService(InstanceRef, instance)), ) attachSessionToWorkspace(session.id, previous.id) - const workspaceCtx = await AppRuntime.runPromise( + const workspaceCtx = await runtime.runPromise( InstanceStore.Service.use((store) => store.load({ directory: workspaceTmp.path })), ) const workspaceProjectID = await context.provide(workspaceCtx, async () => { diff --git a/packages/opencode/test/lib/effect.ts b/packages/opencode/test/lib/effect.ts index 6ad2838fcae..37b7b95a2d8 100644 --- a/packages/opencode/test/lib/effect.ts +++ b/packages/opencode/test/lib/effect.ts @@ -1,8 +1,9 @@ import { test, type TestOptions } from "bun:test" import { Cause, Duration, Effect, Exit, Layer } from "effect" -import type * as Scope from "effect/Scope" +import * as Scope from "effect/Scope" import * as TestClock from "effect/testing/TestClock" import * as TestConsole from "effect/testing/TestConsole" +import { memoMap } from "@opencode-ai/core/effect/memo-map" import type { Config } from "@/config/config" import { TestInstance, withTmpdirInstance } from "../fixture/fixture" @@ -24,7 +25,9 @@ function instanceArgs( const body = (value: Body) => Effect.suspend(() => (typeof value === "function" ? value() : value)) -const run = (value: Body, layer: Layer.Layer) => +type Runner = (value: Body, layer: Layer.Layer) => Promise + +const isolatedRun: Runner = (value, layer) => Effect.gen(function* () { const exit = yield* body(value).pipe(Effect.scoped, Effect.provide(layer), Effect.exit) if (Exit.isFailure(exit)) { @@ -35,7 +38,25 @@ const run = (value: Body, layer: Layer.Layer return yield* exit }).pipe(Effect.runPromise) -const make = (testLayer: Layer.Layer, liveLayer: Layer.Layer) => { +// Builds the layer through the shared process-wide memoMap so cached services +// (Bus, Session, …) match Server.Default's instances. Use for tests that +// publish to an in-process HTTP server and need pub/sub identity with the +// server's handlers. +const sharedRun: Runner = (value, layer) => + Effect.gen(function* () { + const scope = yield* Scope.make() + const ctx = yield* Layer.buildWithMemoMap(layer, memoMap, scope) + const exit = yield* body(value).pipe(Effect.scoped, Effect.provide(ctx), Effect.exit) + yield* Scope.close(scope, Exit.void) + if (Exit.isFailure(exit)) { + for (const err of Cause.prettyErrors(exit.cause)) { + yield* Effect.logError(err) + } + } + return yield* exit + }).pipe(Effect.runPromise) + +const make = (testLayer: Layer.Layer, liveLayer: Layer.Layer, run: Runner = isolatedRun) => { const effect = (name: string, value: Body, opts?: number | TestOptions) => test(name, () => run(value, testLayer), opts) @@ -110,6 +131,13 @@ export const it = make(testEnv, liveEnv) export const testEffect = (layer: Layer.Layer) => make(Layer.provideMerge(layer, testEnv), Layer.provideMerge(layer, liveEnv)) +// Variant of `testEffect` that builds the test layer through the shared +// process-wide memoMap so services like Bus/Session resolve to the same +// instances Server.Default uses. Use when a test needs pub/sub identity with +// an in-process HTTP server — most tests should stick with `testEffect`. +export const testEffectShared = (layer: Layer.Layer) => + make(Layer.provideMerge(layer, testEnv), Layer.provideMerge(layer, liveEnv), sharedRun) + export const awaitWithTimeout = ( self: Effect.Effect, message: string, diff --git a/packages/opencode/test/lib/sse.ts b/packages/opencode/test/lib/sse.ts new file mode 100644 index 00000000000..cc076963cba --- /dev/null +++ b/packages/opencode/test/lib/sse.ts @@ -0,0 +1,63 @@ +import { Effect, Queue, Schema, Stream } from "effect" +import { HttpClient, HttpClientRequest } from "effect/unstable/http" +import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" + +export const SseEvent = Schema.Struct({ + id: Schema.optional(Schema.String), + type: Schema.String, + properties: Schema.Record(Schema.String, Schema.Any), +}) + +export type SseEvent = Schema.Schema.Type + +function decodeFrames(text: string): SseEvent[] { + return text + .split(/\n\n+/) + .map((part) => part.trim()) + .filter((part) => part.length > 0) + .map((part) => Schema.decodeUnknownSync(SseEvent)(JSON.parse(part.replace(/^data: /, "")))) +} + +/** + * Opens a scoped subscription to the instance `/event` SSE stream and returns + * a Queue of decoded events. The underlying request and decoder fiber are + * released when the test scope closes. + */ +export const openInstanceEventStream = (directory: string) => + Effect.gen(function* () { + const response = yield* HttpClientRequest.get(EventPaths.event).pipe( + HttpClientRequest.setHeader("x-opencode-directory", directory), + HttpClient.execute, + ) + const queue = yield* Queue.unbounded() + yield* response.stream.pipe( + Stream.decodeText({ encoding: "utf-8" }), + Stream.flatMap((text) => Stream.fromIterable(decodeFrames(text))), + Stream.runForEach((event) => Queue.offer(queue, event)), + Effect.forkScoped, + ) + return queue + }) + +export const readNextEvent = (queue: Queue.Queue) => + Queue.take(queue).pipe( + Effect.timeoutOrElse({ + duration: "3 seconds", + orElse: () => Effect.fail(new Error("timed out reading SSE event")), + }), + ) + +export const collectUntilEvent = (queue: Queue.Queue, predicate: (event: SseEvent) => boolean) => + Effect.gen(function* () { + const events: SseEvent[] = [] + while (true) { + const event = yield* readNextEvent(queue) + events.push(event) + if (predicate(event)) return events + } + }).pipe( + Effect.timeoutOrElse({ + duration: "4 seconds", + orElse: () => Effect.fail(new Error("collectUntilEvent deadline exceeded")), + }), + ) diff --git a/packages/opencode/test/provider/amazon-bedrock.test.ts b/packages/opencode/test/provider/amazon-bedrock.test.ts index 26bb520fc19..cb94e8cd970 100644 --- a/packages/opencode/test/provider/amazon-bedrock.test.ts +++ b/packages/opencode/test/provider/amazon-bedrock.test.ts @@ -1,31 +1,25 @@ -import { afterEach, test, expect, describe } from "bun:test" +import { afterEach, describe, expect, test } from "bun:test" +import { Effect, Layer } from "effect" import path from "path" import { unlink } from "fs/promises" - -import { ProviderID } from "../../src/provider/schema" -import { disposeAllInstances, tmpdir, withTestInstance } from "../fixture/fixture" -import type { InstanceContext } from "../../src/project/instance-context" -import { Provider } from "@/provider/provider" -import { Env } from "../../src/env" import { Global } from "@opencode-ai/core/global" import { Filesystem } from "@/util/filesystem" -import { Effect } from "effect" -import { AppRuntime } from "../../src/effect/app-runtime" -import { InstanceRef } from "../../src/effect/instance-ref" -import { makeRuntime } from "../../src/effect/run-service" +import { Env } from "../../src/env" +import { Provider } from "@/provider/provider" +import { ProviderID } from "../../src/provider/schema" +import { disposeAllInstances } from "../fixture/fixture" +import { testEffect } from "../lib/effect" + +const it = testEffect(Layer.mergeAll(Provider.defaultLayer, Env.defaultLayer)) -const env = makeRuntime(Env.Service, Env.defaultLayer) const originalEnv = new Map() -function rememberEnv(k: string) { - if (!originalEnv.has(k)) originalEnv.set(k, process.env[k]) -} - -const set = (ctx: InstanceContext, k: string, v: string) => { - rememberEnv(k) - process.env[k] = v - return env.runSync((svc) => svc.set(k, v).pipe(Effect.provideService(InstanceRef, ctx))) -} +const set = (k: string, v: string) => + Effect.gen(function* () { + if (!originalEnv.has(k)) originalEnv.set(k, process.env[k]) + process.env[k] = v + yield* Env.Service.use((svc) => svc.set(k, v)) + }) afterEach(async () => { for (const [key, value] of originalEnv) { @@ -36,427 +30,250 @@ afterEach(async () => { await disposeAllInstances() }) -async function list(ctx: InstanceContext) { - return AppRuntime.runPromise( - Effect.gen(function* () { - const provider = yield* Provider.Service - return yield* provider.list() - }).pipe(Effect.provideService(InstanceRef, ctx)), - ) -} +const list = Provider.Service.use((svc) => svc.list()) -test("Bedrock: config region takes precedence over AWS_REGION env var", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "eu-west-1", - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_REGION", "us-east-1") - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) - expect(providers[ProviderID.amazonBedrock]).toBeDefined() - expect(providers[ProviderID.amazonBedrock].options?.region).toBe("eu-west-1") - }, - }) -}) - -test("Bedrock: falls back to AWS_REGION env var when no config region", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_REGION", "eu-west-1") - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) - expect(providers[ProviderID.amazonBedrock]).toBeDefined() - expect(providers[ProviderID.amazonBedrock].options?.region).toBe("eu-west-1") - }, - }) -}) - -test("Bedrock: loads when bearer token from auth.json is present", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "eu-west-1", - }, - }, - }, - }), - ) - }, - }) - - const authPath = path.join(Global.Path.data, "auth.json") - - // Save original auth.json if it exists - let originalAuth: string | undefined - try { - originalAuth = await Filesystem.readText(authPath) - } catch { - // File doesn't exist, that's fine - } - - try { - // Write test auth.json - await Filesystem.write( - authPath, - JSON.stringify({ - "amazon-bedrock": { - type: "api", - key: "test-bearer-token", - }, - }), - ) - - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "") - set(ctx, "AWS_ACCESS_KEY_ID", "") - set(ctx, "AWS_BEARER_TOKEN_BEDROCK", "") - const providers = await list(ctx) - expect(providers[ProviderID.amazonBedrock]).toBeDefined() - expect(providers[ProviderID.amazonBedrock].options?.region).toBe("eu-west-1") - }, - }) - } finally { - // Restore original or delete - if (originalAuth !== undefined) { - await Filesystem.write(authPath, originalAuth) - } else { +const withAuthJson = (contents: string) => + Effect.acquireRelease( + Effect.promise(async () => { + const authPath = path.join(Global.Path.data, "auth.json") + let original: string | undefined try { - await unlink(authPath) + original = await Filesystem.readText(authPath) } catch { - // Ignore errors if file doesn't exist + original = undefined } - } - } -}) + await Filesystem.write(authPath, contents) + return { authPath, original } + }), + ({ authPath, original }) => + Effect.promise(async () => { + if (original !== undefined) { + await Filesystem.write(authPath, original) + return + } + await unlink(authPath).catch(() => undefined) + }), + ) -test("Bedrock: config profile takes precedence over AWS_PROFILE env var", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - profile: "my-custom-profile", - region: "us-east-1", - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "default") - set(ctx, "AWS_ACCESS_KEY_ID", "test-key-id") - const providers = await list(ctx) +it.instance( + "Bedrock: config region takes precedence over AWS_REGION env var", + () => + Effect.gen(function* () { + yield* set("AWS_REGION", "us-east-1") + yield* set("AWS_PROFILE", "default") + const providers = yield* list + expect(providers[ProviderID.amazonBedrock]).toBeDefined() + expect(providers[ProviderID.amazonBedrock].options?.region).toBe("eu-west-1") + }), + { config: { provider: { "amazon-bedrock": { options: { region: "eu-west-1" } } } } }, +) + +it.instance( + "Bedrock: falls back to AWS_REGION env var when no config region", + () => + Effect.gen(function* () { + yield* set("AWS_REGION", "eu-west-1") + yield* set("AWS_PROFILE", "default") + const providers = yield* list + expect(providers[ProviderID.amazonBedrock]).toBeDefined() + expect(providers[ProviderID.amazonBedrock].options?.region).toBe("eu-west-1") + }), +) + +it.instance( + "Bedrock: loads when bearer token from auth.json is present", + () => + Effect.gen(function* () { + yield* withAuthJson(JSON.stringify({ "amazon-bedrock": { type: "api", key: "test-bearer-token" } })) + yield* set("AWS_PROFILE", "") + yield* set("AWS_ACCESS_KEY_ID", "") + yield* set("AWS_BEARER_TOKEN_BEDROCK", "") + const providers = yield* list + expect(providers[ProviderID.amazonBedrock]).toBeDefined() + expect(providers[ProviderID.amazonBedrock].options?.region).toBe("eu-west-1") + }), + { config: { provider: { "amazon-bedrock": { options: { region: "eu-west-1" } } } } }, +) + +it.instance( + "Bedrock: config profile takes precedence over AWS_PROFILE env var", + () => + Effect.gen(function* () { + yield* set("AWS_PROFILE", "default") + yield* set("AWS_ACCESS_KEY_ID", "test-key-id") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() expect(providers[ProviderID.amazonBedrock].options?.region).toBe("us-east-1") + }), + { + config: { + provider: { "amazon-bedrock": { options: { profile: "my-custom-profile", region: "us-east-1" } } }, }, - }) -}) + }, +) -test("Bedrock: includes custom endpoint in options when specified", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - endpoint: "https://bedrock-runtime.us-east-1.vpce-xxxxx.amazonaws.com", - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) +it.instance( + "Bedrock: includes custom endpoint in options when specified", + () => + Effect.gen(function* () { + yield* set("AWS_PROFILE", "default") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() expect(providers[ProviderID.amazonBedrock].options?.endpoint).toBe( "https://bedrock-runtime.us-east-1.vpce-xxxxx.amazonaws.com", ) + }), + { + config: { + provider: { + "amazon-bedrock": { + options: { endpoint: "https://bedrock-runtime.us-east-1.vpce-xxxxx.amazonaws.com" }, + }, + }, }, - }) -}) + }, +) -test("Bedrock: autoloads when AWS_WEB_IDENTITY_TOKEN_FILE is present", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "us-east-1", - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_WEB_IDENTITY_TOKEN_FILE", "/var/run/secrets/eks.amazonaws.com/serviceaccount/token") - set(ctx, "AWS_ROLE_ARN", "arn:aws:iam::123456789012:role/my-eks-role") - set(ctx, "AWS_PROFILE", "") - set(ctx, "AWS_ACCESS_KEY_ID", "") - const providers = await list(ctx) +it.instance( + "Bedrock: autoloads when AWS_WEB_IDENTITY_TOKEN_FILE is present", + () => + Effect.gen(function* () { + yield* set("AWS_WEB_IDENTITY_TOKEN_FILE", "/var/run/secrets/eks.amazonaws.com/serviceaccount/token") + yield* set("AWS_ROLE_ARN", "arn:aws:iam::123456789012:role/my-eks-role") + yield* set("AWS_PROFILE", "") + yield* set("AWS_ACCESS_KEY_ID", "") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() expect(providers[ProviderID.amazonBedrock].options?.region).toBe("us-east-1") - }, - }) -}) + }), + { config: { provider: { "amazon-bedrock": { options: { region: "us-east-1" } } } } }, +) -// Tests for cross-region inference profile prefix handling -// Models from models.dev may come with prefixes already (e.g., us., eu., global.) -// These should NOT be double-prefixed when passed to the SDK +// Cross-region inference profile prefix handling. +// Models from models.dev may come with prefixes already (e.g. us., eu., global.). +// These should NOT be double-prefixed when passed to the SDK. -test("Bedrock: model with us. prefix should not be double-prefixed", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "us-east-1", - }, - models: { - "us.anthropic.claude-opus-4-5-20251101-v1:0": { - name: "Claude Opus 4.5 (US)", - }, - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) +it.instance( + "Bedrock: model with us. prefix should not be double-prefixed", + () => + Effect.gen(function* () { + yield* set("AWS_PROFILE", "default") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() - // The model should exist with the us. prefix expect(providers[ProviderID.amazonBedrock].models["us.anthropic.claude-opus-4-5-20251101-v1:0"]).toBeDefined() + }), + { + config: { + provider: { + "amazon-bedrock": { + options: { region: "us-east-1" }, + models: { "us.anthropic.claude-opus-4-5-20251101-v1:0": { name: "Claude Opus 4.5 (US)" } }, + }, + }, }, - }) -}) + }, +) -test("Bedrock: model with global. prefix should not be prefixed", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "us-east-1", - }, - models: { - "global.anthropic.claude-opus-4-5-20251101-v1:0": { - name: "Claude Opus 4.5 (Global)", - }, - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) +it.instance( + "Bedrock: model with global. prefix should not be prefixed", + () => + Effect.gen(function* () { + yield* set("AWS_PROFILE", "default") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() expect(providers[ProviderID.amazonBedrock].models["global.anthropic.claude-opus-4-5-20251101-v1:0"]).toBeDefined() + }), + { + config: { + provider: { + "amazon-bedrock": { + options: { region: "us-east-1" }, + models: { "global.anthropic.claude-opus-4-5-20251101-v1:0": { name: "Claude Opus 4.5 (Global)" } }, + }, + }, }, - }) -}) + }, +) -test("Bedrock: model with eu. prefix should not be double-prefixed", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "eu-west-1", - }, - models: { - "eu.anthropic.claude-opus-4-5-20251101-v1:0": { - name: "Claude Opus 4.5 (EU)", - }, - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) +it.instance( + "Bedrock: model with eu. prefix should not be double-prefixed", + () => + Effect.gen(function* () { + yield* set("AWS_PROFILE", "default") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() expect(providers[ProviderID.amazonBedrock].models["eu.anthropic.claude-opus-4-5-20251101-v1:0"]).toBeDefined() + }), + { + config: { + provider: { + "amazon-bedrock": { + options: { region: "eu-west-1" }, + models: { "eu.anthropic.claude-opus-4-5-20251101-v1:0": { name: "Claude Opus 4.5 (EU)" } }, + }, + }, }, - }) -}) + }, +) -test("Bedrock: model without prefix in US region should get us. prefix added", async () => { - await using tmp = await tmpdir({ - init: async (dir) => { - await Filesystem.write( - path.join(dir, "opencode.json"), - JSON.stringify({ - $schema: "https://opencode.ai/config.json", - provider: { - "amazon-bedrock": { - options: { - region: "us-east-1", - }, - models: { - "anthropic.claude-opus-4-5-20251101-v1:0": { - name: "Claude Opus 4.5", - }, - }, - }, - }, - }), - ) - }, - }) - await withTestInstance({ - directory: tmp.path, - fn: async (ctx) => { - set(ctx, "AWS_PROFILE", "default") - const providers = await list(ctx) +it.instance( + "Bedrock: model without prefix in US region should get us. prefix added", + () => + Effect.gen(function* () { + yield* set("AWS_PROFILE", "default") + const providers = yield* list expect(providers[ProviderID.amazonBedrock]).toBeDefined() - // Non-prefixed model should still be registered expect(providers[ProviderID.amazonBedrock].models["anthropic.claude-opus-4-5-20251101-v1:0"]).toBeDefined() + }), + { + config: { + provider: { + "amazon-bedrock": { + options: { region: "us-east-1" }, + models: { "anthropic.claude-opus-4-5-20251101-v1:0": { name: "Claude Opus 4.5" } }, + }, + }, }, - }) -}) - -// Direct unit tests for cross-region inference profile prefix handling -// These test the prefix detection logic used in getModel + }, +) +// Direct unit tests for cross-region inference profile prefix detection. describe("Bedrock cross-region prefix detection", () => { const crossRegionPrefixes = ["global.", "us.", "eu.", "jp.", "apac.", "au."] test("should detect global. prefix", () => { - const modelID = "global.anthropic.claude-opus-4-5-20251101-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(true) + expect(crossRegionPrefixes.some((p) => "global.anthropic.claude-opus-4-5-20251101-v1:0".startsWith(p))).toBe(true) }) test("should detect us. prefix", () => { - const modelID = "us.anthropic.claude-opus-4-5-20251101-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(true) + expect(crossRegionPrefixes.some((p) => "us.anthropic.claude-opus-4-5-20251101-v1:0".startsWith(p))).toBe(true) }) test("should detect eu. prefix", () => { - const modelID = "eu.anthropic.claude-opus-4-5-20251101-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(true) + expect(crossRegionPrefixes.some((p) => "eu.anthropic.claude-opus-4-5-20251101-v1:0".startsWith(p))).toBe(true) }) test("should detect jp. prefix", () => { - const modelID = "jp.anthropic.claude-sonnet-4-20250514-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(true) + expect(crossRegionPrefixes.some((p) => "jp.anthropic.claude-sonnet-4-20250514-v1:0".startsWith(p))).toBe(true) }) test("should detect apac. prefix", () => { - const modelID = "apac.anthropic.claude-sonnet-4-20250514-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(true) + expect(crossRegionPrefixes.some((p) => "apac.anthropic.claude-sonnet-4-20250514-v1:0".startsWith(p))).toBe(true) }) test("should detect au. prefix", () => { - const modelID = "au.anthropic.claude-sonnet-4-5-20250929-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(true) + expect(crossRegionPrefixes.some((p) => "au.anthropic.claude-sonnet-4-5-20250929-v1:0".startsWith(p))).toBe(true) }) test("should NOT detect prefix for non-prefixed model", () => { - const modelID = "anthropic.claude-opus-4-5-20251101-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(false) + expect(crossRegionPrefixes.some((p) => "anthropic.claude-opus-4-5-20251101-v1:0".startsWith(p))).toBe(false) }) test("should NOT detect prefix for amazon nova models", () => { - const modelID = "amazon.nova-pro-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(false) + expect(crossRegionPrefixes.some((p) => "amazon.nova-pro-v1:0".startsWith(p))).toBe(false) }) test("should NOT detect prefix for cohere models", () => { - const modelID = "cohere.command-r-plus-v1:0" - const hasPrefix = crossRegionPrefixes.some((prefix) => modelID.startsWith(prefix)) - expect(hasPrefix).toBe(false) + expect(crossRegionPrefixes.some((p) => "cohere.command-r-plus-v1:0".startsWith(p))).toBe(false) }) }) diff --git a/packages/opencode/test/provider/provider.test.ts b/packages/opencode/test/provider/provider.test.ts index 5215b094ab2..971801f289c 100644 --- a/packages/opencode/test/provider/provider.test.ts +++ b/packages/opencode/test/provider/provider.test.ts @@ -12,8 +12,8 @@ import { Provider } from "@/provider/provider" import { ProviderID, ModelID } from "../../src/provider/schema" import { Filesystem } from "@/util/filesystem" import { Env } from "../../src/env" -import { Effect, Layer } from "effect" -import { AppRuntime } from "../../src/effect/app-runtime" +import { Effect, Layer, ManagedRuntime } from "effect" +import { memoMap } from "@opencode-ai/core/effect/memo-map" import { InstanceRef } from "../../src/effect/instance-ref" import { makeRuntime } from "../../src/effect/run-service" import { testEffect } from "../lib/effect" @@ -66,8 +66,10 @@ const providerLayer = (flags: Partial = {}) => Layer.provide(RuntimeFlags.layer(flags)), ) +const providerRuntime = ManagedRuntime.make(Layer.mergeAll(Provider.defaultLayer, Plugin.defaultLayer), { memoMap }) + async function run(ctx: InstanceContext, fn: (provider: Provider.Interface) => Effect.Effect) { - return AppRuntime.runPromise( + return providerRuntime.runPromise( Effect.gen(function* () { const provider = yield* Provider.Service return yield* fn(provider) @@ -2509,7 +2511,7 @@ test("plugin config providers persist after instance dispose", async () => { const first = await withTestInstance({ directory: tmp.path, fn: async (ctx) => - AppRuntime.runPromise( + providerRuntime.runPromise( Effect.gen(function* () { const plugin = yield* Plugin.Service const provider = yield* Provider.Service diff --git a/packages/opencode/test/server/httpapi-event-diagnostics.test.ts b/packages/opencode/test/server/httpapi-event-diagnostics.test.ts index 95a6941a4af..d6bb53eacf7 100644 --- a/packages/opencode/test/server/httpapi-event-diagnostics.test.ts +++ b/packages/opencode/test/server/httpapi-event-diagnostics.test.ts @@ -1,53 +1,50 @@ // Diagnostic suite for /event SSE delivery. // // Each test isolates ONE variable in the publisher chain while keeping the -// subscriber path constant (raw `app().request` reading the SSE body — no SDK -// consumer involvement). The pass/fail pattern across tests tells us where the -// bug lives: +// subscriber path constant (HttpClient against the in-process HttpApiApp +// routes, reading the SSE body via openInstanceEventStream). The pass/fail +// pattern across tests tells us where the bug lives: // -// D1 (baseline): publish via Bus.Service.use via AppRuntime — mirror of the -// existing httpapi-event.test.ts test 3. Confirms /event SSE delivery -// works for a SOME publish path. +// D1 (baseline): publish via Bus.Service.use — mirror of the existing +// httpapi-event.test.ts test 3. Confirms /event SSE delivery works for +// a SOME publish path. // // D2: publish N times in quick succession via Bus.Service.use. If the bus // subscription is acquired correctly there should be no message loss. // -// D3: publish via SyncEvent.use.run via AppRuntime — exercises the same path -// the HTTP handlers use (Session.updatePart → sync.run → bus.publish) -// without the HTTP roundtrip. Tells us whether the sync path itself can -// deliver in-process. +// D3: publish via SyncEvent.use.run — exercises the same path the HTTP +// handlers use (Session.updatePart → sync.run → bus.publish) without the +// HTTP roundtrip. Tells us whether the sync path itself can deliver +// in-process. // -// D4: publish via SyncEvent.use.run from a fresh `Effect.provide` scope -// (mimicking what happens if a handler's layer was scoped per-request). +// D4: publish via SyncEvent.use.run; subscriber is an in-process Bus +// callback. Confirms pub/sub identity end-to-end without /event SSE. // // D5: in-process Bus.Service callback subscriber AND raw /event SSE subscriber // receive the same publish. If both receive: no bug. If only the // callback receives: the /event handler has an acquisition race. +// +// D6: same as D5 but the callback subscriber is attached AFTER /event SSE +// subscription is established. Order-of-setup variable. +import { NodeHttpServer, NodeServices } from "@effect/platform-node" import { afterEach, describe, expect } from "bun:test" -import { Deferred, Effect, Schema } from "effect" +import { Config, Deferred, Effect, Layer } from "effect" +import { HttpRouter, HttpServer } from "effect/unstable/http" +import * as Socket from "effect/unstable/socket/Socket" import * as Log from "@opencode-ai/core/util/log" import { Bus } from "../../src/bus" -import { type AppServices, AppRuntime } from "../../src/effect/app-runtime" -import { InstanceRef } from "../../src/effect/instance-ref" -import { Server } from "../../src/server/server" import { Event as ServerEvent } from "../../src/server/event" -import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" +import { HttpApiApp } from "../../src/server/routes/instance/httpapi/server" import { MessageV2 } from "../../src/session/message-v2" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SyncEvent } from "../../src/sync" import { resetDatabase } from "../fixture/db" import { disposeAllInstances, TestInstance } from "../fixture/fixture" -import { it } from "../lib/effect" +import { testEffectShared } from "../lib/effect" +import { collectUntilEvent, openInstanceEventStream, readNextEvent, type SseEvent } from "../lib/sse" void Log.init({ print: false }) -const EventData = Schema.Struct({ - id: Schema.optional(Schema.String), - type: Schema.String, - properties: Schema.Record(Schema.String, Schema.Any), -}) - -type SseEvent = Schema.Schema.Type type BusEvent = { type: string; properties: unknown } afterEach(async () => { @@ -55,82 +52,39 @@ afterEach(async () => { await resetDatabase() }) -const inApp = (eff: Effect.Effect) => - Effect.gen(function* () { - const ctx = yield* InstanceRef - if (!ctx) return yield* Effect.die("InstanceRef not provided in test scope") - return yield* Effect.promise(() => AppRuntime.runPromise(eff.pipe(Effect.provideService(InstanceRef, ctx)))) - }) +const servedRoutes: Layer.Layer = HttpRouter.serve( + HttpApiApp.routes, + { disableListenLog: true, disableLogger: true }, +) -const publishConnected = inApp(Bus.Service.use((svc) => svc.publish(ServerEvent.Connected, {}))) +const it = testEffectShared( + Layer.mergeAll( + Bus.defaultLayer, + SyncEvent.defaultLayer, + servedRoutes.pipe( + Layer.provide(Socket.layerWebSocketConstructorGlobal), + Layer.provideMerge(NodeHttpServer.layerTest), + Layer.provideMerge(NodeServices.layer), + ), + ), +) + +const publishConnected = Bus.Service.use((svc) => svc.publish(ServerEvent.Connected, {})) const publishPartUpdated = (partID: ReturnType) => { const sessionID = SessionID.make(`ses_${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}`) - return inApp( - SyncEvent.use.run(MessageV2.Event.PartUpdated, { - sessionID, - part: { id: partID, sessionID, messageID: MessageID.ascending(), type: "text", text: "diag" }, - time: Date.now(), - }), - ) + return SyncEvent.use.run(MessageV2.Event.PartUpdated, { + sessionID, + part: { id: partID, sessionID, messageID: MessageID.ascending(), type: "text", text: "diag" }, + time: Date.now(), + }) } const subscribeAllCallback = (handler: (event: BusEvent) => void) => - Effect.acquireRelease(inApp(Bus.Service.use((svc) => svc.subscribeAllCallback(handler))), (dispose) => + Effect.acquireRelease(Bus.Service.use((svc) => svc.subscribeAllCallback(handler)), (dispose) => Effect.sync(dispose), ) -const openEventStream = (directory: string) => - Effect.gen(function* () { - const response = yield* Effect.promise(async () => - Server.Default().app.request(EventPaths.event, { headers: { "x-opencode-directory": directory } }), - ) - if (!response.body) return yield* Effect.die("missing SSE response body") - const reader = response.body.getReader() - yield* Effect.addFinalizer(() => Effect.promise(() => reader.cancel().catch(() => undefined))) - return reader - }) - -const decoder = new TextDecoder() - -function decodeFrame(value: Uint8Array): SseEvent[] { - return decoder - .decode(value) - .split(/\n\n+/) - .map((part) => part.trim()) - .filter((part) => part.length > 0) - .map((part) => Schema.decodeUnknownSync(EventData)(JSON.parse(part.replace(/^data: /, "")))) -} - -const readNextEvent = (reader: ReadableStreamDefaultReader) => - Effect.promise(() => reader.read()).pipe( - Effect.timeoutOrElse({ - duration: "3 seconds", - orElse: () => Effect.fail(new Error("timed out reading SSE chunk")), - }), - Effect.flatMap((result) => { - if (result.done || !result.value) return Effect.fail(new Error("event stream closed")) - const frames = decodeFrame(result.value) - if (frames.length === 0) return Effect.fail(new Error("empty SSE frame")) - return Effect.succeed(frames[0]) - }), - ) - -const collectUntilEvent = (reader: ReadableStreamDefaultReader, predicate: (event: SseEvent) => boolean) => - Effect.gen(function* () { - const events: SseEvent[] = [] - while (true) { - const event = yield* readNextEvent(reader) - events.push(event) - if (predicate(event)) return events - } - }).pipe( - Effect.timeoutOrElse({ - duration: "4 seconds", - orElse: () => Effect.fail(new Error("collectUntil deadline exceeded")), - }), - ) - const isPartUpdated = (event: { type: string }) => event.type === MessageV2.Event.PartUpdated.type describe("/event SSE delivery diagnostics", () => { @@ -142,7 +96,7 @@ describe("/event SSE delivery diagnostics", () => { () => Effect.gen(function* () { const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) + const reader = yield* openInstanceEventStream(directory) expect((yield* readNextEvent(reader)).type).toBe("server.connected") yield* publishConnected @@ -157,7 +111,7 @@ describe("/event SSE delivery diagnostics", () => { () => Effect.gen(function* () { const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) + const reader = yield* openInstanceEventStream(directory) expect((yield* readNextEvent(reader)).type).toBe("server.connected") const N = 5 @@ -172,13 +126,13 @@ describe("/event SSE delivery diagnostics", () => { // The critical test. If D1 passes but this fails, the bus-identity fix is // incomplete OR the sync.run publish path doesn't reach the same bus - // /event subscribes to, even within the same AppRuntime. + // /event subscribes to, even when both share the memoMap. it.instance( "D3: delivers a SyncEvent published via SyncEvent.use.run after server.connected", () => Effect.gen(function* () { const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) + const reader = yield* openInstanceEventStream(directory) expect((yield* readNextEvent(reader)).type).toBe("server.connected") const partID = PartID.ascending() @@ -186,7 +140,8 @@ describe("/event SSE delivery diagnostics", () => { const collected = yield* collectUntilEvent(reader, isPartUpdated) const updated = collected.find(isPartUpdated) - expect(updated?.properties.part.id).toBe(partID) + expect(updated).toBeDefined() + expect((updated as SseEvent).properties.part.id).toBe(partID) }), { git: true, config: { formatter: false, lsp: false } }, ) @@ -216,7 +171,7 @@ describe("/event SSE delivery diagnostics", () => { }), ) expect(event.type).toBe(MessageV2.Event.PartUpdated.type) - expect(event.properties).toMatchObject({ part: { id: partID } }) + expect((event.properties as { part: { id: string } }).part.id).toBe(partID) }), { git: true, config: { formatter: false, lsp: false } }, ) @@ -233,7 +188,7 @@ describe("/event SSE delivery diagnostics", () => { yield* subscribeAllCallback((event) => { if (isPartUpdated(event)) Deferred.doneUnsafe(callbackReceived, Effect.succeed(event)) }) - const reader = yield* openEventStream(directory) + const reader = yield* openInstanceEventStream(directory) expect((yield* readNextEvent(reader)).type).toBe("server.connected") const partID = PartID.ascending() @@ -255,29 +210,6 @@ describe("/event SSE delivery diagnostics", () => { { git: true, config: { formatter: false, lsp: false } }, ) - // D7: like D5 but the "second subscriber" is a NO-OP AppRuntime.runPromise - // call (no PubSub.subscribe). If D7 passes, the specific subscribeAllCallback - // is what breaks SSE — not arbitrary AppRuntime usage. If D7 fails, anything - // running through AppRuntime concurrently with /event SSE breaks delivery. - it.instance( - "D7: SSE receives sync.run publish even with concurrent no-op AppRuntime activity", - () => - Effect.gen(function* () { - const { directory } = yield* TestInstance - yield* inApp(Effect.void) - - const reader = yield* openEventStream(directory) - expect((yield* readNextEvent(reader)).type).toBe("server.connected") - - const partID = PartID.ascending() - yield* publishPartUpdated(partID) - - const collected = yield* collectUntilEvent(reader, isPartUpdated) - expect(collected.find(isPartUpdated)).toBeDefined() - }), - { git: true, config: { formatter: false, lsp: false } }, - ) - // D6: same as D5 but the callback subscriber is attached AFTER /event SSE // subscription is established. If D5 fails and D6 passes, the order of // subscriber setup is the determining factor. @@ -286,7 +218,7 @@ describe("/event SSE delivery diagnostics", () => { () => Effect.gen(function* () { const { directory } = yield* TestInstance - const reader = yield* openEventStream(directory) + const reader = yield* openInstanceEventStream(directory) expect((yield* readNextEvent(reader)).type).toBe("server.connected") const callbackReceived = yield* Deferred.make() diff --git a/packages/opencode/test/server/httpapi-event.test.ts b/packages/opencode/test/server/httpapi-event.test.ts index fcf7b59ff39..7d632dd9065 100644 --- a/packages/opencode/test/server/httpapi-event.test.ts +++ b/packages/opencode/test/server/httpapi-event.test.ts @@ -1,121 +1,95 @@ -import { afterEach, describe, expect, test } from "bun:test" -import { Bus } from "../../src/bus" -import { AppRuntime } from "../../src/effect/app-runtime" -import { InstanceRef } from "../../src/effect/instance-ref" -import { Server } from "../../src/server/server" -import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" -import { Event as ServerEvent } from "../../src/server/event" +import { NodeHttpServer, NodeServices } from "@effect/platform-node" +import { afterEach, describe, expect } from "bun:test" +import { Config, Effect, Layer, Queue } from "effect" +import { HttpClient, HttpClientRequest, HttpRouter, HttpServer } from "effect/unstable/http" +import * as Socket from "effect/unstable/socket/Socket" import * as Log from "@opencode-ai/core/util/log" -import { Effect, Schema } from "effect" +import { Bus } from "../../src/bus" +import { Event as ServerEvent } from "../../src/server/event" +import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" +import { HttpApiApp } from "../../src/server/routes/instance/httpapi/server" import { resetDatabase } from "../fixture/db" -import { disposeAllInstances, reloadTestInstance, tmpdir } from "../fixture/fixture" +import { disposeAllInstances, TestInstance } from "../fixture/fixture" +import { testEffectShared } from "../lib/effect" +import { openInstanceEventStream, readNextEvent } from "../lib/sse" void Log.init({ print: false }) -function app() { - return Server.Default().app -} - -const EventData = Schema.Struct({ - id: Schema.optional(Schema.String), - type: Schema.String, - properties: Schema.Record(Schema.String, Schema.Any), -}) - -async function readChunk(reader: ReadableStreamDefaultReader) { - let timeout: ReturnType | undefined - try { - return await Promise.race([ - reader.read(), - new Promise((_, reject) => { - timeout = setTimeout(() => reject(new Error("timed out waiting for event")), 5_000) - }), - ]) - } finally { - if (timeout) clearTimeout(timeout) - } -} - -async function readFirstEvent(response: Response) { - if (!response.body) throw new Error("missing response body") - const reader = response.body.getReader() - try { - return await readEvent(reader) - } finally { - await reader.cancel() - } -} - -async function readEvent(reader: ReadableStreamDefaultReader) { - const result = await readChunk(reader) - if (result.done || !result.value) throw new Error("event stream closed") - return Schema.decodeUnknownSync(EventData)(JSON.parse(new TextDecoder().decode(result.value).replace(/^data: /, ""))) -} - -async function readStatusWithin(reader: ReadableStreamDefaultReader, delay: number) { - let timeout: ReturnType | undefined - try { - return await Promise.race([ - reader.read().then((result) => (result.done ? "closed" : "event")), - new Promise<"open">((resolve) => { - timeout = setTimeout(() => resolve("open"), delay) - }), - ]) - } finally { - if (timeout) clearTimeout(timeout) - } -} - afterEach(async () => { await disposeAllInstances() await resetDatabase() }) +const servedRoutes: Layer.Layer = HttpRouter.serve( + HttpApiApp.routes, + { disableListenLog: true, disableLogger: true }, +) + +const it = testEffectShared( + Layer.mergeAll( + Bus.defaultLayer, + servedRoutes.pipe( + Layer.provide(Socket.layerWebSocketConstructorGlobal), + Layer.provideMerge(NodeHttpServer.layerTest), + Layer.provideMerge(NodeServices.layer), + ), + ), +) + describe("event HttpApi", () => { - test("serves event stream", async () => { - await using tmp = await tmpdir({ git: true, config: { formatter: false, lsp: false } }) - const response = await app().request(EventPaths.event, { headers: { "x-opencode-directory": tmp.path } }) + it.instance( + "serves event stream with correct headers and initial server.connected", + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + const response = yield* HttpClientRequest.get(EventPaths.event).pipe( + HttpClientRequest.setHeader("x-opencode-directory", directory), + HttpClient.execute, + ) - expect(response.status).toBe(200) - expect(response.headers.get("content-type")).toContain("text/event-stream") - expect(response.headers.get("cache-control")).toBe("no-cache, no-transform") - expect(response.headers.get("x-accel-buffering")).toBe("no") - expect(response.headers.get("x-content-type-options")).toBe("nosniff") - expect(await readFirstEvent(response)).toMatchObject({ type: "server.connected", properties: {} }) - }) + expect(response.status).toBe(200) + expect(response.headers["content-type"]).toContain("text/event-stream") + expect(response.headers["cache-control"]).toBe("no-cache, no-transform") + expect(response.headers["x-accel-buffering"]).toBe("no") + expect(response.headers["x-content-type-options"]).toBe("nosniff") - test("keeps the event stream open after the initial event", async () => { - await using tmp = await tmpdir({ git: true, config: { formatter: false, lsp: false } }) - const response = await app().request(EventPaths.event, { headers: { "x-opencode-directory": tmp.path } }) - if (!response.body) throw new Error("missing response body") + // Read one event to also verify the stream is wired up; the response is + // scope-bound so the connection closes when the test ends. + const events = yield* openInstanceEventStream(directory) + expect(yield* readNextEvent(events)).toMatchObject({ type: "server.connected", properties: {} }) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) - const reader = response.body.getReader() - try { - expect(await readEvent(reader)).toMatchObject({ type: "server.connected", properties: {} }) - expect(await readStatusWithin(reader, 250)).toBe("open") - } finally { - await reader.cancel() - } - }) + it.instance( + "keeps the event stream open after the initial event", + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + const events = yield* openInstanceEventStream(directory) + expect(yield* readNextEvent(events)).toMatchObject({ type: "server.connected", properties: {} }) - test("delivers instance bus events after the initial event", async () => { - await using tmp = await tmpdir({ git: true, config: { formatter: false, lsp: false } }) - const response = await app().request(EventPaths.event, { headers: { "x-opencode-directory": tmp.path } }) - if (!response.body) throw new Error("missing response body") + // If no second event arrives within 250ms, the stream is still open. + const status = yield* Queue.take(events).pipe( + Effect.map(() => "event" as const), + Effect.timeoutOrElse({ duration: "250 millis", orElse: () => Effect.succeed("open" as const) }), + ) + expect(status).toBe("open") + }), + { git: true, config: { formatter: false, lsp: false } }, + ) - const reader = response.body.getReader() - try { - expect(await readEvent(reader)).toMatchObject({ type: "server.connected", properties: {} }) + it.instance( + "delivers instance bus events after the initial event", + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + const events = yield* openInstanceEventStream(directory) + expect(yield* readNextEvent(events)).toMatchObject({ type: "server.connected", properties: {} }) - const next = readEvent(reader) - const ctx = await reloadTestInstance({ directory: tmp.path }) - await AppRuntime.runPromise( - Bus.Service.use((svc) => svc.publish(ServerEvent.Connected, {})).pipe(Effect.provideService(InstanceRef, ctx)), - ) - - expect(await next).toMatchObject({ type: "server.connected", properties: {} }) - } finally { - await reader.cancel() - } - }) + yield* Bus.Service.use((svc) => svc.publish(ServerEvent.Connected, {})) + expect(yield* readNextEvent(events)).toMatchObject({ type: "server.connected", properties: {} }) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) }) diff --git a/packages/opencode/test/session/llm.test.ts b/packages/opencode/test/session/llm.test.ts index 5ad1ae21772..5e8eb03f9e2 100644 --- a/packages/opencode/test/session/llm.test.ts +++ b/packages/opencode/test/session/llm.test.ts @@ -1,7 +1,8 @@ import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test" import path from "path" import { tool, type ModelMessage } from "ai" -import { Cause, Effect, Exit, Layer, Stream } from "effect" +import { Cause, Effect, Exit, Layer, ManagedRuntime, Stream } from "effect" +import { memoMap } from "@opencode-ai/core/effect/memo-map" import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http" import z from "zod" import { makeRuntime } from "../../src/effect/run-service" @@ -21,7 +22,6 @@ import { tmpdir, withTestInstance } from "../fixture/fixture" import type { Agent } from "../../src/agent/agent" import { MessageV2 } from "../../src/session/message-v2" import { SessionID, MessageID } from "../../src/session/schema" -import { AppRuntime } from "../../src/effect/app-runtime" import { RuntimeFlags } from "@/effect/runtime-flags" import { Permission } from "@/permission" import { LLMAISDK } from "@/session/llm/ai-sdk" @@ -50,12 +50,14 @@ const openAIConfig = (model: ModelsDev.Provider["models"][string], baseURL: stri } } +const providerRuntime = ManagedRuntime.make(Provider.defaultLayer, { memoMap }) + async function getModel(providerID: ProviderID, modelID: ModelID, ctx: InstanceContext) { const effect = Effect.gen(function* () { const provider = yield* Provider.Service return yield* provider.getModel(providerID, modelID) }) - return AppRuntime.runPromise(effect.pipe(Effect.provideService(InstanceRef, ctx))) + return providerRuntime.runPromise(effect.pipe(Effect.provideService(InstanceRef, ctx))) } const llm = makeRuntime(LLM.Service, LLM.defaultLayer)