Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 07b357de04 | |||
| ddbd119dcb |
@@ -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 () => {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 } },
|
||||
)
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user