Compare commits

...

2 Commits

Author SHA1 Message Date
Kit Langton 07b357de04 refactor(opencode): drop AppRuntime from instance-runtime
Builds a local ManagedRuntime over InstanceLayer.layer with the shared
process-wide memoMap, so the four exported Promise helpers (load,
disposeInstance, disposeAllInstances, reloadInstance) no longer go
through AppRuntime. Service identity is preserved via memoMap; every
caller's behavior is unchanged.

One step in the broader AppRuntime removal. All public-facing exports
stay the same — callers in src/cli/bootstrap, src/cli/cmd/tui/worker,
src/acp/runtime, and the test fixtures don't need to change.
2026-05-20 20:50:49 -04:00
Kit Langton ddbd119dcb test: drop AppRuntime usage from tests
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<SseEvent>.

   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).
2026-05-20 20:50:32 -04:00
9 changed files with 457 additions and 628 deletions
@@ -1,16 +1,21 @@
import { AppRuntime } from "@/effect/app-runtime"
import { Effect, ManagedRuntime } from "effect"
import { memoMap } from "@opencode-ai/core/effect/memo-map"
import { type InstanceContext } from "./instance-context"
import { InstanceLayer } from "./instance-layer"
import { InstanceStore, type LoadInput } from "./instance-store"
// Bridge for Promise/ALS callers that cannot yet yield InstanceStore.Service.
// Delete this module once those callers are migrated to Effect boundaries that
// provide InstanceStore directly.
export const load = (input: LoadInput) => AppRuntime.runPromise(InstanceStore.Service.use((store) => store.load(input)))
export const disposeInstance = (ctx: InstanceContext) =>
AppRuntime.runPromise(InstanceStore.Service.use((store) => store.dispose(ctx)))
export const disposeAllInstances = () => AppRuntime.runPromise(InstanceStore.Service.use((store) => store.disposeAll()))
export const reloadInstance = (input: LoadInput) =>
AppRuntime.runPromise(InstanceStore.Service.use((store) => store.reload(input)))
const runtime = ManagedRuntime.make(InstanceLayer.layer, { memoMap })
const run = <A, E>(fn: (store: InstanceStore.Interface) => Effect.Effect<A, E>) =>
runtime.runPromise(InstanceStore.Service.use(fn))
export const load = (input: LoadInput) => run((store) => store.load(input))
export const disposeInstance = (ctx: InstanceContext) => run((store) => store.dispose(ctx))
export const disposeAllInstances = () => run((store) => store.disposeAll())
export const reloadInstance = (input: LoadInput) => run((store) => store.reload(input))
export * as InstanceRuntime from "./instance-runtime"
@@ -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<T>(fn: (ctx: InstanceContext) => T | Promise<T>) {
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 = <A, E>(effect: Effect.Effect<A, E, Workspace.Service>) => {
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 () => {
+31 -3
View File
@@ -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 = <A, E, R>(value: Body<A, E, R>) => Effect.suspend(() => (typeof value === "function" ? value() : value))
const run = <A, E, R, E2>(value: Body<A, E, R | Scope.Scope>, layer: Layer.Layer<R, E2>) =>
type Runner = <A, E, R, E2>(value: Body<A, E, R | Scope.Scope>, layer: Layer.Layer<R, E2>) => Promise<A>
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 = <A, E, R, E2>(value: Body<A, E, R | Scope.Scope>, layer: Layer.Layer
return yield* exit
}).pipe(Effect.runPromise)
const make = <R, E>(testLayer: Layer.Layer<R, E>, liveLayer: Layer.Layer<R, E>) => {
// 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 = <R, E>(testLayer: Layer.Layer<R, E>, liveLayer: Layer.Layer<R, E>, run: Runner = isolatedRun) => {
const effect = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test(name, () => run(value, testLayer), opts)
@@ -110,6 +131,13 @@ export const it = make(testEnv, liveEnv)
export const testEffect = <R, E>(layer: Layer.Layer<R, E>) =>
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 = <R, E>(layer: Layer.Layer<R, E>) =>
make(Layer.provideMerge(layer, testEnv), Layer.provideMerge(layer, liveEnv), sharedRun)
export const awaitWithTimeout = <A, E, R>(
self: Effect.Effect<A, E, R>,
message: string,
+63
View File
@@ -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<typeof SseEvent>
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<SseEvent>()
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<SseEvent>) =>
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<SseEvent>, 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")),
}),
)
@@ -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<string, string | undefined>()
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)
})
})
@@ -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<RuntimeFlags.Info> = {}) =>
Layer.provide(RuntimeFlags.layer(flags)),
)
const providerRuntime = ManagedRuntime.make(Layer.mergeAll(Provider.defaultLayer, Plugin.defaultLayer), { memoMap })
async function run<A, E>(ctx: InstanceContext, fn: (provider: Provider.Interface) => Effect.Effect<A, E, never>) {
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
@@ -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<typeof EventData>
type BusEvent = { type: string; properties: unknown }
afterEach(async () => {
@@ -55,82 +52,39 @@ afterEach(async () => {
await resetDatabase()
})
const inApp = <A, E>(eff: Effect.Effect<A, E, AppServices>) =>
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<never, Config.ConfigError, HttpServer.HttpServer> = 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<typeof PartID.ascending>) => {
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<Uint8Array>) =>
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<Uint8Array>, 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<BusEvent>()
@@ -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<Uint8Array>) {
let timeout: ReturnType<typeof setTimeout> | undefined
try {
return await Promise.race([
reader.read(),
new Promise<never>((_, 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<Uint8Array>) {
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<Uint8Array>, delay: number) {
let timeout: ReturnType<typeof setTimeout> | 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<never, Config.ConfigError, HttpServer.HttpServer> = 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 } },
)
})
+5 -3
View File
@@ -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)