mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-02 16:26:14 -04:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ae2ecd1ed3 | |||
| 159d271e1e | |||
| 762850dfe5 |
@@ -7,9 +7,7 @@ import { eq } from "drizzle-orm"
|
||||
import { GlobalBus } from "@/bus/global"
|
||||
import { Bus as ProjectBus } from "@/bus"
|
||||
import { BusEvent } from "@/bus/bus-event"
|
||||
import type { InstanceContext } from "@/project/instance-context"
|
||||
import { EventSequenceTable, EventTable } from "./event.sql"
|
||||
import type { WorkspaceID } from "@/control-plane/schema"
|
||||
import { EventID } from "./schema"
|
||||
import { Context, Effect, Layer, Schema as EffectSchema } from "effect"
|
||||
import type { DeepMutable } from "@opencode-ai/core/schema"
|
||||
@@ -17,7 +15,7 @@ import { EventV2 } from "@opencode-ai/core/event"
|
||||
import { serviceUse } from "@/effect/service-use"
|
||||
import { InstanceState } from "@/effect/instance-state"
|
||||
import { RuntimeFlags } from "@/effect/runtime-flags"
|
||||
import { attachWith } from "@/effect/run-service"
|
||||
import { EffectBridge } from "@/effect/bridge"
|
||||
|
||||
// Keep `Event["data"]` mutable because projectors mutate the persisted shape
|
||||
// when writing to the database. Bus payloads (`Properties`) stay readonly —
|
||||
@@ -51,10 +49,6 @@ export type SerializedEvent<Def extends Definition = Definition> = Event<Def> &
|
||||
|
||||
type ProjectorFunc = (db: Database.TxOrDb, data: unknown, event: Event) => void
|
||||
type ConvertEvent = (type: string, data: Event["data"]) => unknown | Promise<unknown>
|
||||
type PublishContext = {
|
||||
instance?: InstanceContext
|
||||
workspace?: WorkspaceID
|
||||
}
|
||||
|
||||
export interface Interface {
|
||||
readonly run: <Def extends Definition>(
|
||||
@@ -107,16 +101,14 @@ export const layer = Layer.effect(Service)(
|
||||
}
|
||||
|
||||
const publish = !!options?.publish
|
||||
const context = publish
|
||||
? {
|
||||
instance: yield* InstanceState.context,
|
||||
workspace: yield* InstanceState.workspaceID,
|
||||
}
|
||||
: undefined
|
||||
// Bridge captures handler-fiber refs (InstanceRef/WorkspaceRef) and the
|
||||
// full Effect context, so the forked publish + GlobalBus emit run with
|
||||
// the right state without a per-call attachWith.
|
||||
const bridge = yield* EffectBridge.make()
|
||||
process(def, event, {
|
||||
bus,
|
||||
bridge,
|
||||
publish,
|
||||
context,
|
||||
ownerID: options?.ownerID,
|
||||
experimentalWorkspaces: flags.experimentalWorkspaces,
|
||||
})
|
||||
@@ -154,12 +146,7 @@ export const layer = Layer.effect(Service)(
|
||||
}
|
||||
|
||||
const { publish = true } = options || {}
|
||||
const context = publish
|
||||
? {
|
||||
instance: yield* InstanceState.context,
|
||||
workspace: yield* InstanceState.workspaceID,
|
||||
}
|
||||
: undefined
|
||||
const bridge = yield* EffectBridge.make()
|
||||
|
||||
// Note that this is an "immediate" transaction which is critical.
|
||||
// We need to make sure we can safely read and write with nothing
|
||||
@@ -175,7 +162,7 @@ export const layer = Layer.effect(Service)(
|
||||
const seq = row?.seq != null ? row.seq + 1 : 0
|
||||
|
||||
const event = { id, seq, aggregateID: agg, data }
|
||||
process(def, event, { bus, publish, context, experimentalWorkspaces: flags.experimentalWorkspaces })
|
||||
process(def, event, { bus, bridge, publish, experimentalWorkspaces: flags.experimentalWorkspaces })
|
||||
},
|
||||
{
|
||||
behavior: "immediate",
|
||||
@@ -308,8 +295,8 @@ function process<Def extends Definition>(
|
||||
event: Event<Def>,
|
||||
options: {
|
||||
bus: ProjectBus.Interface
|
||||
bridge: EffectBridge.Shape
|
||||
publish: boolean
|
||||
context?: PublishContext
|
||||
ownerID?: string
|
||||
experimentalWorkspaces: boolean
|
||||
},
|
||||
@@ -351,37 +338,36 @@ function process<Def extends Definition>(
|
||||
}
|
||||
|
||||
Database.effect(() => {
|
||||
if (options?.publish) {
|
||||
if (!options.context?.instance) {
|
||||
throw new Error("SyncEvent.process: publish requires instance context")
|
||||
}
|
||||
|
||||
const result = convertEvent(def.type, event.data)
|
||||
const publish = (data: unknown) =>
|
||||
Effect.runPromise(
|
||||
attachWith(options.bus.publish(def, data as Properties<Def>, { id: event.id }), {
|
||||
instance: options.context?.instance,
|
||||
workspace: options.context?.workspace,
|
||||
}),
|
||||
)
|
||||
if (result instanceof Promise) {
|
||||
void result.then(publish)
|
||||
} else {
|
||||
void publish(result)
|
||||
}
|
||||
|
||||
GlobalBus.emit("event", {
|
||||
directory: options.context.instance.directory,
|
||||
project: options.context.instance.project.id,
|
||||
workspace: options.context.workspace,
|
||||
payload: {
|
||||
type: "sync",
|
||||
syncEvent: {
|
||||
type: versionedType(def.type, def.version),
|
||||
...event,
|
||||
},
|
||||
},
|
||||
})
|
||||
if (!options.publish) return
|
||||
const result = convertEvent(def.type, event.data)
|
||||
// The bridge was built inside the caller's fiber so it already carries
|
||||
// InstanceRef/WorkspaceRef and the full Effect context. Both the bus
|
||||
// publish and the GlobalBus emit run inside the forked Effect so they
|
||||
// share the same instance/workspace lookup.
|
||||
const publish = (data: unknown) =>
|
||||
options.bridge.fork(
|
||||
Effect.gen(function* () {
|
||||
yield* options.bus.publish(def, data as Properties<Def>, { id: event.id })
|
||||
const instance = yield* InstanceState.context
|
||||
const workspace = yield* InstanceState.workspaceID
|
||||
GlobalBus.emit("event", {
|
||||
directory: instance.directory,
|
||||
project: instance.project.id,
|
||||
workspace,
|
||||
payload: {
|
||||
type: "sync",
|
||||
syncEvent: {
|
||||
type: versionedType(def.type, def.version),
|
||||
...event,
|
||||
},
|
||||
},
|
||||
})
|
||||
}),
|
||||
)
|
||||
if (result instanceof Promise) {
|
||||
void result.then(publish)
|
||||
} else {
|
||||
publish(result)
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -228,81 +228,65 @@ it.instance(
|
||||
},
|
||||
)
|
||||
|
||||
test("custom model alias via config", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
init: async (dir) => {
|
||||
await Bun.write(
|
||||
path.join(dir, "opencode.json"),
|
||||
JSON.stringify({
|
||||
$schema: "https://opencode.ai/config.json",
|
||||
provider: {
|
||||
anthropic: {
|
||||
models: {
|
||||
"my-alias": {
|
||||
id: "claude-sonnet-4-20250514",
|
||||
name: "My Custom Alias",
|
||||
},
|
||||
},
|
||||
it.instance(
|
||||
"custom model alias via config",
|
||||
Effect.gen(function* () {
|
||||
yield* setProcessEnv("ANTHROPIC_API_KEY", "test-api-key")
|
||||
const providers = yield* Provider.Service.use((provider) => provider.list())
|
||||
expect(providers[ProviderID.anthropic]).toBeDefined()
|
||||
expect(providers[ProviderID.anthropic].models["my-alias"]).toBeDefined()
|
||||
expect(providers[ProviderID.anthropic].models["my-alias"].name).toBe("My Custom Alias")
|
||||
}),
|
||||
{
|
||||
config: {
|
||||
provider: {
|
||||
anthropic: {
|
||||
models: {
|
||||
"my-alias": {
|
||||
id: "claude-sonnet-4-20250514",
|
||||
name: "My Custom Alias",
|
||||
},
|
||||
},
|
||||
}),
|
||||
)
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
await withTestInstance({
|
||||
directory: tmp.path,
|
||||
fn: async (ctx) => {
|
||||
await set(ctx, "ANTHROPIC_API_KEY", "test-api-key")
|
||||
const providers = await list(ctx)
|
||||
expect(providers[ProviderID.anthropic]).toBeDefined()
|
||||
expect(providers[ProviderID.anthropic].models["my-alias"]).toBeDefined()
|
||||
expect(providers[ProviderID.anthropic].models["my-alias"].name).toBe("My Custom Alias")
|
||||
},
|
||||
})
|
||||
})
|
||||
},
|
||||
)
|
||||
|
||||
test("custom provider with npm package", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
init: async (dir) => {
|
||||
await Bun.write(
|
||||
path.join(dir, "opencode.json"),
|
||||
JSON.stringify({
|
||||
$schema: "https://opencode.ai/config.json",
|
||||
provider: {
|
||||
"custom-provider": {
|
||||
name: "Custom Provider",
|
||||
npm: "@ai-sdk/openai-compatible",
|
||||
api: "https://api.custom.com/v1",
|
||||
env: ["CUSTOM_API_KEY"],
|
||||
models: {
|
||||
"custom-model": {
|
||||
name: "Custom Model",
|
||||
tool_call: true,
|
||||
limit: {
|
||||
context: 128000,
|
||||
output: 4096,
|
||||
},
|
||||
},
|
||||
},
|
||||
options: {
|
||||
apiKey: "custom-key",
|
||||
it.instance(
|
||||
"custom provider with npm package",
|
||||
Effect.gen(function* () {
|
||||
const providers = yield* Provider.Service.use((provider) => provider.list())
|
||||
expect(providers[ProviderID.make("custom-provider")]).toBeDefined()
|
||||
expect(providers[ProviderID.make("custom-provider")].name).toBe("Custom Provider")
|
||||
expect(providers[ProviderID.make("custom-provider")].models["custom-model"]).toBeDefined()
|
||||
}),
|
||||
{
|
||||
config: {
|
||||
provider: {
|
||||
"custom-provider": {
|
||||
name: "Custom Provider",
|
||||
npm: "@ai-sdk/openai-compatible",
|
||||
api: "https://api.custom.com/v1",
|
||||
env: ["CUSTOM_API_KEY"],
|
||||
models: {
|
||||
"custom-model": {
|
||||
name: "Custom Model",
|
||||
tool_call: true,
|
||||
limit: {
|
||||
context: 128000,
|
||||
output: 4096,
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
)
|
||||
options: {
|
||||
apiKey: "custom-key",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
await withTestInstance({
|
||||
directory: tmp.path,
|
||||
fn: async (ctx) => {
|
||||
const providers = await list(ctx)
|
||||
expect(providers[ProviderID.make("custom-provider")]).toBeDefined()
|
||||
expect(providers[ProviderID.make("custom-provider")].name).toBe("Custom Provider")
|
||||
expect(providers[ProviderID.make("custom-provider")].models["custom-model"]).toBeDefined()
|
||||
},
|
||||
})
|
||||
})
|
||||
},
|
||||
)
|
||||
|
||||
it.instance(
|
||||
"filters alpha provider models by default",
|
||||
@@ -324,66 +308,58 @@ experimentalModels.instance(
|
||||
{ config: alphaProviderConfig },
|
||||
)
|
||||
|
||||
test("custom DeepSeek openai-compatible model defaults interleaved reasoning field", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
init: async (dir) => {
|
||||
await Bun.write(
|
||||
path.join(dir, "opencode.json"),
|
||||
JSON.stringify({
|
||||
$schema: "https://opencode.ai/config.json",
|
||||
provider: {
|
||||
"custom-provider": {
|
||||
name: "Custom Provider",
|
||||
npm: "@ai-sdk/openai-compatible",
|
||||
api: "https://api.custom.com/v1",
|
||||
models: {
|
||||
"deepseek-r1": {
|
||||
name: "DeepSeek R1",
|
||||
},
|
||||
"deepseek-details": {
|
||||
name: "DeepSeek Details",
|
||||
interleaved: { field: "reasoning_details" },
|
||||
},
|
||||
"custom-model": {
|
||||
name: "Custom Model",
|
||||
},
|
||||
},
|
||||
options: {
|
||||
apiKey: "custom-key",
|
||||
},
|
||||
it.instance(
|
||||
"custom DeepSeek openai-compatible model defaults interleaved reasoning field",
|
||||
Effect.gen(function* () {
|
||||
const providers = yield* Provider.Service.use((provider) => provider.list())
|
||||
const provider = providers[ProviderID.make("custom-provider")]
|
||||
expect(provider.models["deepseek-r1"].capabilities.interleaved).toEqual({ field: "reasoning_content" })
|
||||
expect(provider.models["deepseek-details"].capabilities.interleaved).toEqual({ field: "reasoning_details" })
|
||||
expect(provider.models["custom-model"].capabilities.interleaved).toBe(false)
|
||||
expect(providers[ProviderID.make("custom-anthropic-provider")].models["deepseek-r1"].capabilities.interleaved).toBe(
|
||||
false,
|
||||
)
|
||||
}),
|
||||
{
|
||||
config: {
|
||||
provider: {
|
||||
"custom-provider": {
|
||||
name: "Custom Provider",
|
||||
npm: "@ai-sdk/openai-compatible",
|
||||
api: "https://api.custom.com/v1",
|
||||
models: {
|
||||
"deepseek-r1": {
|
||||
name: "DeepSeek R1",
|
||||
},
|
||||
"custom-anthropic-provider": {
|
||||
name: "Custom Anthropic Provider",
|
||||
npm: "@ai-sdk/anthropic",
|
||||
api: "https://api.custom.com/v1",
|
||||
models: {
|
||||
"deepseek-r1": {
|
||||
name: "DeepSeek R1",
|
||||
},
|
||||
},
|
||||
options: {
|
||||
apiKey: "custom-key",
|
||||
},
|
||||
"deepseek-details": {
|
||||
name: "DeepSeek Details",
|
||||
interleaved: { field: "reasoning_details" },
|
||||
},
|
||||
"custom-model": {
|
||||
name: "Custom Model",
|
||||
},
|
||||
},
|
||||
}),
|
||||
)
|
||||
options: {
|
||||
apiKey: "custom-key",
|
||||
},
|
||||
},
|
||||
"custom-anthropic-provider": {
|
||||
name: "Custom Anthropic Provider",
|
||||
npm: "@ai-sdk/anthropic",
|
||||
api: "https://api.custom.com/v1",
|
||||
models: {
|
||||
"deepseek-r1": {
|
||||
name: "DeepSeek R1",
|
||||
},
|
||||
},
|
||||
options: {
|
||||
apiKey: "custom-key",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
await withTestInstance({
|
||||
directory: tmp.path,
|
||||
fn: async (ctx) => {
|
||||
const providers = await list(ctx)
|
||||
const provider = providers[ProviderID.make("custom-provider")]
|
||||
expect(provider.models["deepseek-r1"].capabilities.interleaved).toEqual({ field: "reasoning_content" })
|
||||
expect(provider.models["deepseek-details"].capabilities.interleaved).toEqual({ field: "reasoning_details" })
|
||||
expect(provider.models["custom-model"].capabilities.interleaved).toBe(false)
|
||||
expect(
|
||||
providers[ProviderID.make("custom-anthropic-provider")].models["deepseek-r1"].capabilities.interleaved,
|
||||
).toBe(false)
|
||||
},
|
||||
})
|
||||
})
|
||||
},
|
||||
)
|
||||
|
||||
test("env variable takes precedence, config merges options", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
|
||||
@@ -1,14 +1,15 @@
|
||||
import { describe, expect, beforeEach, afterAll } from "bun:test"
|
||||
import { provideTmpdirInstance } from "../fixture/fixture"
|
||||
import { Effect, Layer, Schema } from "effect"
|
||||
import { Deferred, Effect, Layer, Schema } from "effect"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { GlobalBus, type GlobalEvent } from "../../src/bus/global"
|
||||
import { SyncEvent } from "../../src/sync"
|
||||
import { Database, eq } from "@/storage/db"
|
||||
import { EventSequenceTable, EventTable } from "../../src/sync/event.sql"
|
||||
import { MessageID } from "../../src/session/schema"
|
||||
import { initProjectors } from "../../src/server/projectors"
|
||||
import { testEffect } from "../lib/effect"
|
||||
import { awaitWithTimeout, testEffect } from "../lib/effect"
|
||||
import { RuntimeFlags } from "@/effect/runtime-flags"
|
||||
|
||||
const it = testEffect(
|
||||
@@ -139,6 +140,43 @@ describe("SyncEvent", () => {
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
// Regression for the EffectBridge migration. GlobalBus.emit used to fire
|
||||
// synchronously inside the Database.effect post-commit callback. After the
|
||||
// migration it fires inside the forked publish Effect, AFTER bus.publish
|
||||
// completes. Consumers don't care about microsecond-level ordering, but
|
||||
// we still need to prove the emit actually fires.
|
||||
it.live(
|
||||
"emits sync events to GlobalBus after publishing to ProjectBus",
|
||||
provideTmpdirInstance(() =>
|
||||
Effect.gen(function* () {
|
||||
const { Created } = setup()
|
||||
// Filter for OUR specific event in the handler so we ignore any
|
||||
// stray sync events from other tests' lingering forks.
|
||||
const received = yield* Deferred.make<GlobalEvent>()
|
||||
const handler = (evt: GlobalEvent) => {
|
||||
if (evt.payload?.type === "sync" && evt.payload?.syncEvent?.type === "item.created.1") {
|
||||
Deferred.doneUnsafe(received, Effect.succeed(evt))
|
||||
}
|
||||
}
|
||||
GlobalBus.on("event", handler)
|
||||
try {
|
||||
yield* SyncEvent.use.run(Created, { id: "evt_global_1", name: "global" })
|
||||
const event = yield* awaitWithTimeout(
|
||||
Deferred.await(received),
|
||||
"timed out waiting for sync event on GlobalBus",
|
||||
"2 seconds",
|
||||
)
|
||||
expect(event.payload).toMatchObject({
|
||||
type: "sync",
|
||||
syncEvent: { type: "item.created.1", data: { id: "evt_global_1", name: "global" } },
|
||||
})
|
||||
} finally {
|
||||
GlobalBus.off("event", handler)
|
||||
}
|
||||
}),
|
||||
),
|
||||
)
|
||||
})
|
||||
|
||||
describe("replay", () => {
|
||||
|
||||
@@ -69,6 +69,7 @@ Repeated setup work, long sleeps/timeouts, serial integration tests, filesystem/
|
||||
| HTTP listen PTY ticket tests restart the same listener topology twice | Folded directory-scoped ticket regression into the broader unsafe-ticket test | 7.051s | 6.170s | keep | Two targeted reruns passed after the change: 6.76s, 6.17s; still covers mint failure and successful same-directory upgrade. |
|
||||
| File watcher readiness can write before async native subscriptions are active | Retried short readiness writes and accepted symlink-realpath HEAD events | failed | 4.62s | keep | Three sequential focused watcher runs passed: 4.62s, 4.57s, 4.64s; full suite no longer failed in `watcher.test.ts`. |
|
||||
| First provider config/env/filtering block can use Effect-aware instance fixtures | Migrated six `tmpdir` + `withTestInstance` cases to `it.instance` | 6.06s | 6.07s | keep | Neutral timing, but removes manual config file writes and instance plumbing; use as the pattern for later provider slices. |
|
||||
| Custom provider/model config cases can use Effect-aware instance fixtures | Migrated three more config-heavy provider cases to `it.instance` | 6.07s | 6.12s | keep | Neutral timing within noise, but continues removing manual config file writes on top of the first provider fixture PR. |
|
||||
|
||||
## Profiling Results
|
||||
|
||||
|
||||
Reference in New Issue
Block a user