Merge branch 'dev' into ai-sdk-v6
This commit is contained in:
Aiden Cline
@@ -60,6 +60,8 @@ function toolEvent(
|
||||
const payload: EventMessagePartUpdated = {
|
||||
type: "message.part.updated",
|
||||
properties: {
|
||||
sessionID: sessionId,
|
||||
time: Date.now(),
|
||||
part: {
|
||||
id: `part_${opts.callID}`,
|
||||
sessionID: sessionId,
|
||||
|
||||
@@ -73,7 +73,6 @@ test("explore agent denies edit and write", async () => {
|
||||
expect(explore?.mode).toBe("subagent")
|
||||
expect(evalPerm(explore, "edit")).toBe("deny")
|
||||
expect(evalPerm(explore, "write")).toBe("deny")
|
||||
expect(evalPerm(explore, "todoread")).toBe("deny")
|
||||
expect(evalPerm(explore, "todowrite")).toBe("deny")
|
||||
},
|
||||
})
|
||||
@@ -102,7 +101,6 @@ test("general agent denies todo tools", async () => {
|
||||
expect(general).toBeDefined()
|
||||
expect(general?.mode).toBe("subagent")
|
||||
expect(general?.hidden).toBeUndefined()
|
||||
expect(evalPerm(general, "todoread")).toBe("deny")
|
||||
expect(evalPerm(general, "todowrite")).toBe("deny")
|
||||
},
|
||||
})
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
|
||||
import { describe, expect } from "bun:test"
|
||||
import { Deferred, Effect, Layer, Stream } from "effect"
|
||||
import z from "zod"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { BusEvent } from "../../src/bus/bus-event"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { provideInstance, provideTmpdirInstance, tmpdirScoped } from "../fixture/fixture"
|
||||
import { testEffect } from "../lib/effect"
|
||||
|
||||
const TestEvent = {
|
||||
Ping: BusEvent.define("test.effect.ping", z.object({ value: z.number() })),
|
||||
Pong: BusEvent.define("test.effect.pong", z.object({ message: z.string() })),
|
||||
}
|
||||
|
||||
const node = NodeChildProcessSpawner.layer.pipe(
|
||||
Layer.provideMerge(Layer.mergeAll(NodeFileSystem.layer, NodePath.layer)),
|
||||
)
|
||||
|
||||
const live = Layer.mergeAll(Bus.layer, node)
|
||||
|
||||
const it = testEffect(live)
|
||||
|
||||
describe("Bus (Effect-native)", () => {
|
||||
it.effect("publish + subscribe stream delivers events", () =>
|
||||
provideTmpdirInstance(() =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const received: number[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
|
||||
Effect.sync(() => {
|
||||
received.push(evt.properties.value)
|
||||
if (received.length === 2) Deferred.doneUnsafe(done, Effect.void)
|
||||
}),
|
||||
).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.sleep("10 millis")
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* bus.publish(TestEvent.Ping, { value: 2 })
|
||||
yield* Deferred.await(done)
|
||||
|
||||
expect(received).toEqual([1, 2])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("subscribe filters by event type", () =>
|
||||
provideTmpdirInstance(() =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const pings: number[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
|
||||
Effect.sync(() => {
|
||||
pings.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(done, Effect.void)
|
||||
}),
|
||||
).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.sleep("10 millis")
|
||||
yield* bus.publish(TestEvent.Pong, { message: "ignored" })
|
||||
yield* bus.publish(TestEvent.Ping, { value: 42 })
|
||||
yield* Deferred.await(done)
|
||||
|
||||
expect(pings).toEqual([42])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("subscribeAll receives all types", () =>
|
||||
provideTmpdirInstance(() =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const types: string[] = []
|
||||
const done = yield* Deferred.make<void>()
|
||||
|
||||
yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
|
||||
Effect.sync(() => {
|
||||
types.push(evt.type)
|
||||
if (types.length === 2) Deferred.doneUnsafe(done, Effect.void)
|
||||
}),
|
||||
).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.sleep("10 millis")
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* bus.publish(TestEvent.Pong, { message: "hi" })
|
||||
yield* Deferred.await(done)
|
||||
|
||||
expect(types).toContain("test.effect.ping")
|
||||
expect(types).toContain("test.effect.pong")
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("multiple subscribers each receive the event", () =>
|
||||
provideTmpdirInstance(() =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const a: number[] = []
|
||||
const b: number[] = []
|
||||
const doneA = yield* Deferred.make<void>()
|
||||
const doneB = yield* Deferred.make<void>()
|
||||
|
||||
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
|
||||
Effect.sync(() => {
|
||||
a.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(doneA, Effect.void)
|
||||
}),
|
||||
).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Stream.runForEach(bus.subscribe(TestEvent.Ping), (evt) =>
|
||||
Effect.sync(() => {
|
||||
b.push(evt.properties.value)
|
||||
Deferred.doneUnsafe(doneB, Effect.void)
|
||||
}),
|
||||
).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.sleep("10 millis")
|
||||
yield* bus.publish(TestEvent.Ping, { value: 99 })
|
||||
yield* Deferred.await(doneA)
|
||||
yield* Deferred.await(doneB)
|
||||
|
||||
expect(a).toEqual([99])
|
||||
expect(b).toEqual([99])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("subscribeAll stream sees InstanceDisposed on disposal", () =>
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped()
|
||||
const types: string[] = []
|
||||
const seen = yield* Deferred.make<void>()
|
||||
const disposed = yield* Deferred.make<void>()
|
||||
|
||||
// Set up subscriber inside the instance
|
||||
yield* Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
|
||||
yield* Stream.runForEach(bus.subscribeAll(), (evt) =>
|
||||
Effect.sync(() => {
|
||||
types.push(evt.type)
|
||||
if (evt.type === TestEvent.Ping.type) Deferred.doneUnsafe(seen, Effect.void)
|
||||
if (evt.type === Bus.InstanceDisposed.type) Deferred.doneUnsafe(disposed, Effect.void)
|
||||
}),
|
||||
).pipe(Effect.forkScoped)
|
||||
|
||||
yield* Effect.sleep("10 millis")
|
||||
yield* bus.publish(TestEvent.Ping, { value: 1 })
|
||||
yield* Deferred.await(seen)
|
||||
}).pipe(provideInstance(dir))
|
||||
|
||||
// Dispose from OUTSIDE the instance scope
|
||||
yield* Effect.promise(() => Instance.disposeAll())
|
||||
yield* Deferred.await(disposed).pipe(Effect.timeout("2 seconds"))
|
||||
|
||||
expect(types).toContain("test.effect.ping")
|
||||
expect(types).toContain(Bus.InstanceDisposed.type)
|
||||
}),
|
||||
)
|
||||
})
|
||||
@@ -0,0 +1,87 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import z from "zod"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { BusEvent } from "../../src/bus/bus-event"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
|
||||
const TestEvent = BusEvent.define("test.integration", z.object({ value: z.number() }))
|
||||
|
||||
function withInstance(directory: string, fn: () => Promise<void>) {
|
||||
return Instance.provide({ directory, fn })
|
||||
}
|
||||
|
||||
describe("Bus integration: acquireRelease subscriber pattern", () => {
|
||||
afterEach(() => Instance.disposeAll())
|
||||
|
||||
test("subscriber via callback facade receives events and cleans up on unsub", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
const unsub = Bus.subscribe(TestEvent, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 1 })
|
||||
await Bus.publish(TestEvent, { value: 2 })
|
||||
await Bun.sleep(10)
|
||||
|
||||
expect(received).toEqual([1, 2])
|
||||
|
||||
unsub()
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 3 })
|
||||
await Bun.sleep(10)
|
||||
|
||||
expect(received).toEqual([1, 2])
|
||||
})
|
||||
})
|
||||
|
||||
test("subscribeAll receives events from multiple types", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: Array<{ type: string; value?: number }> = []
|
||||
|
||||
const OtherEvent = BusEvent.define("test.other", z.object({ value: z.number() }))
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
received.push({ type: evt.type, value: evt.properties.value })
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 10 })
|
||||
await Bus.publish(OtherEvent, { value: 20 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(received).toEqual([
|
||||
{ type: "test.integration", value: 10 },
|
||||
{ type: "test.other", value: 20 },
|
||||
])
|
||||
})
|
||||
|
||||
test("subscriber cleanup on instance disposal interrupts the stream", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
let disposed = false
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
if (evt.type === Bus.InstanceDisposed.type) {
|
||||
disposed = true
|
||||
return
|
||||
}
|
||||
received.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
await Instance.disposeAll()
|
||||
await Bun.sleep(50)
|
||||
|
||||
expect(received).toEqual([1])
|
||||
expect(disposed).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,219 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import z from "zod"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { BusEvent } from "../../src/bus/bus-event"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
|
||||
const TestEvent = {
|
||||
Ping: BusEvent.define("test.ping", z.object({ value: z.number() })),
|
||||
Pong: BusEvent.define("test.pong", z.object({ message: z.string() })),
|
||||
}
|
||||
|
||||
function withInstance(directory: string, fn: () => Promise<void>) {
|
||||
return Instance.provide({ directory, fn })
|
||||
}
|
||||
|
||||
describe("Bus", () => {
|
||||
afterEach(() => Instance.disposeAll())
|
||||
|
||||
describe("publish + subscribe", () => {
|
||||
test("subscriber is live immediately after subscribe returns", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
})
|
||||
await Bus.publish(TestEvent.Ping, { value: 42 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(received).toEqual([42])
|
||||
})
|
||||
|
||||
test("subscriber receives matching events", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
})
|
||||
// Give the subscriber fiber time to start consuming
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 42 })
|
||||
await Bus.publish(TestEvent.Ping, { value: 99 })
|
||||
// Give subscriber time to process
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(received).toEqual([42, 99])
|
||||
})
|
||||
|
||||
test("subscriber does not receive events of other types", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const pings: number[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
pings.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Pong, { message: "hello" })
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(pings).toEqual([1])
|
||||
})
|
||||
|
||||
test("publish with no subscribers does not throw", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe("unsubscribe", () => {
|
||||
test("unsubscribe stops delivery", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: number[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
const unsub = Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
received.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
unsub()
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 2 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(received).toEqual([1])
|
||||
})
|
||||
})
|
||||
|
||||
describe("subscribeAll", () => {
|
||||
test("subscribeAll is live immediately after subscribe returns", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: string[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
received.push(evt.type)
|
||||
})
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(received).toEqual(["test.ping"])
|
||||
})
|
||||
|
||||
test("receives all event types", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: string[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
received.push(evt.type)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bus.publish(TestEvent.Pong, { message: "hi" })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(received).toContain("test.ping")
|
||||
expect(received).toContain("test.pong")
|
||||
})
|
||||
})
|
||||
|
||||
describe("multiple subscribers", () => {
|
||||
test("all subscribers for same event type are called", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const a: number[] = []
|
||||
const b: number[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
a.push(evt.properties.value)
|
||||
})
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
b.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 7 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(a).toEqual([7])
|
||||
expect(b).toEqual([7])
|
||||
})
|
||||
})
|
||||
|
||||
describe("instance isolation", () => {
|
||||
test("events in one directory do not reach subscribers in another", async () => {
|
||||
await using tmpA = await tmpdir()
|
||||
await using tmpB = await tmpdir()
|
||||
const receivedA: number[] = []
|
||||
const receivedB: number[] = []
|
||||
|
||||
await withInstance(tmpA.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
receivedA.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
await withInstance(tmpB.path, async () => {
|
||||
Bus.subscribe(TestEvent.Ping, (evt) => {
|
||||
receivedB.push(evt.properties.value)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
await withInstance(tmpA.path, async () => {
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
await withInstance(tmpB.path, async () => {
|
||||
await Bus.publish(TestEvent.Ping, { value: 2 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
expect(receivedA).toEqual([1])
|
||||
expect(receivedB).toEqual([2])
|
||||
})
|
||||
})
|
||||
|
||||
describe("instance disposal", () => {
|
||||
test("InstanceDisposed is delivered to wildcard subscribers before stream ends", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const received: string[] = []
|
||||
|
||||
await withInstance(tmp.path, async () => {
|
||||
Bus.subscribeAll((evt) => {
|
||||
received.push(evt.type)
|
||||
})
|
||||
await Bun.sleep(10)
|
||||
await Bus.publish(TestEvent.Ping, { value: 1 })
|
||||
await Bun.sleep(10)
|
||||
})
|
||||
|
||||
// Instance.disposeAll triggers the finalizer which publishes InstanceDisposed
|
||||
await Instance.disposeAll()
|
||||
await Bun.sleep(50)
|
||||
|
||||
expect(received).toContain("test.ping")
|
||||
expect(received).toContain(Bus.InstanceDisposed.type)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -34,7 +34,7 @@ async function check(map: (dir: string) => string) {
|
||||
await using tmp = await tmpdir({ git: true, config: { snapshot: true } })
|
||||
const prev = Global.Path.config
|
||||
;(Global.Path as { config: string }).config = globalTmp.path
|
||||
Config.global.reset()
|
||||
await Config.invalidate()
|
||||
try {
|
||||
await writeConfig(globalTmp.path, {
|
||||
$schema: "https://opencode.ai/config.json",
|
||||
@@ -52,7 +52,7 @@ async function check(map: (dir: string) => string) {
|
||||
} finally {
|
||||
await Instance.disposeAll()
|
||||
;(Global.Path as { config: string }).config = prev
|
||||
Config.global.reset()
|
||||
await Config.invalidate()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1400,7 +1400,6 @@ test("permission config preserves key order", async () => {
|
||||
external_directory: "ask",
|
||||
read: "allow",
|
||||
todowrite: "allow",
|
||||
todoread: "allow",
|
||||
"thoughts_*": "allow",
|
||||
"reasoning_model_*": "allow",
|
||||
"tools_*": "allow",
|
||||
@@ -1421,7 +1420,6 @@ test("permission config preserves key order", async () => {
|
||||
"external_directory",
|
||||
"read",
|
||||
"todowrite",
|
||||
"todoread",
|
||||
"thoughts_*",
|
||||
"reasoning_model_*",
|
||||
"tools_*",
|
||||
|
||||
@@ -1,159 +0,0 @@
|
||||
import { afterEach, describe, expect, mock, test } from "bun:test"
|
||||
import { WorkspaceID } from "../../src/control-plane/schema"
|
||||
import { Hono } from "hono"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
import { Project } from "../../src/project/project"
|
||||
import { WorkspaceTable } from "../../src/control-plane/workspace.sql"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { WorkspaceContext } from "../../src/control-plane/workspace-context"
|
||||
import { Database } from "../../src/storage/db"
|
||||
import { resetDatabase } from "../fixture/db"
|
||||
import * as adaptors from "../../src/control-plane/adaptors"
|
||||
import type { Adaptor } from "../../src/control-plane/types"
|
||||
import { Flag } from "../../src/flag/flag"
|
||||
|
||||
afterEach(async () => {
|
||||
mock.restore()
|
||||
await resetDatabase()
|
||||
})
|
||||
|
||||
const original = Flag.OPENCODE_EXPERIMENTAL_WORKSPACES
|
||||
// @ts-expect-error don't do this normally, but it works
|
||||
Flag.OPENCODE_EXPERIMENTAL_WORKSPACES = true
|
||||
|
||||
afterEach(() => {
|
||||
// @ts-expect-error don't do this normally, but it works
|
||||
Flag.OPENCODE_EXPERIMENTAL_WORKSPACES = original
|
||||
})
|
||||
|
||||
type State = {
|
||||
workspace?: "first" | "second"
|
||||
calls: Array<{ method: string; url: string; body?: string }>
|
||||
}
|
||||
|
||||
const remote = { type: "testing", name: "remote-a" } as unknown as typeof WorkspaceTable.$inferInsert
|
||||
|
||||
async function setup(state: State) {
|
||||
const TestAdaptor: Adaptor = {
|
||||
configure(config) {
|
||||
return config
|
||||
},
|
||||
async create() {
|
||||
throw new Error("not used")
|
||||
},
|
||||
async remove() {},
|
||||
|
||||
async fetch(_config: unknown, input: RequestInfo | URL, init?: RequestInit) {
|
||||
const url =
|
||||
input instanceof Request || input instanceof URL
|
||||
? input.toString()
|
||||
: new URL(input, "http://workspace.test").toString()
|
||||
const request = new Request(url, init)
|
||||
const body = request.method === "GET" || request.method === "HEAD" ? undefined : await request.text()
|
||||
state.calls.push({
|
||||
method: request.method,
|
||||
url: `${new URL(request.url).pathname}${new URL(request.url).search}`,
|
||||
body,
|
||||
})
|
||||
return new Response("proxied", { status: 202 })
|
||||
},
|
||||
}
|
||||
|
||||
adaptors.installAdaptor("testing", TestAdaptor)
|
||||
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
const { project } = await Project.fromDirectory(tmp.path)
|
||||
|
||||
const id1 = WorkspaceID.ascending()
|
||||
const id2 = WorkspaceID.ascending()
|
||||
|
||||
Database.use((db) =>
|
||||
db
|
||||
.insert(WorkspaceTable)
|
||||
.values([
|
||||
{
|
||||
id: id1,
|
||||
branch: "main",
|
||||
project_id: project.id,
|
||||
type: remote.type,
|
||||
name: remote.name,
|
||||
},
|
||||
{
|
||||
id: id2,
|
||||
branch: "main",
|
||||
project_id: project.id,
|
||||
type: "worktree",
|
||||
directory: tmp.path,
|
||||
name: "local",
|
||||
},
|
||||
])
|
||||
.run(),
|
||||
)
|
||||
|
||||
const { WorkspaceRouterMiddleware } = await import("../../src/control-plane/workspace-router-middleware")
|
||||
const app = new Hono().use(WorkspaceRouterMiddleware)
|
||||
|
||||
return {
|
||||
id1,
|
||||
id2,
|
||||
app,
|
||||
async request(input: RequestInfo | URL, init?: RequestInit) {
|
||||
return Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: async () =>
|
||||
WorkspaceContext.provide({
|
||||
workspaceID: state.workspace === "first" ? id1 : id2,
|
||||
fn: () => app.request(input, init),
|
||||
}),
|
||||
})
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
describe("control-plane/session-proxy-middleware", () => {
|
||||
test("forwards non-GET session requests for workspaces", async () => {
|
||||
const state: State = {
|
||||
workspace: "first",
|
||||
calls: [],
|
||||
}
|
||||
|
||||
const ctx = await setup(state)
|
||||
|
||||
ctx.app.post("/session/foo", (c) => c.text("local", 200))
|
||||
const response = await ctx.request("http://workspace.test/session/foo?x=1", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ hello: "world" }),
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
},
|
||||
})
|
||||
|
||||
expect(response.status).toBe(202)
|
||||
expect(await response.text()).toBe("proxied")
|
||||
expect(state.calls).toEqual([
|
||||
{
|
||||
method: "POST",
|
||||
url: "/session/foo?x=1",
|
||||
body: '{"hello":"world"}',
|
||||
},
|
||||
])
|
||||
})
|
||||
|
||||
// It will behave this way when we have syncing
|
||||
//
|
||||
// test("does not forward GET requests", async () => {
|
||||
// const state: State = {
|
||||
// workspace: "first",
|
||||
// calls: [],
|
||||
// }
|
||||
|
||||
// const ctx = await setup(state)
|
||||
|
||||
// ctx.app.get("/session/foo", (c) => c.text("local", 200))
|
||||
// const response = await ctx.request("http://workspace.test/session/foo?x=1")
|
||||
|
||||
// expect(response.status).toBe(200)
|
||||
// expect(await response.text()).toBe("local")
|
||||
// expect(state.calls).toEqual([])
|
||||
// })
|
||||
})
|
||||
@@ -1,70 +0,0 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import { Log } from "../../src/util/log"
|
||||
import { WorkspaceServer } from "../../src/control-plane/workspace-server/server"
|
||||
import { parseSSE } from "../../src/control-plane/sse"
|
||||
import { GlobalBus } from "../../src/bus/global"
|
||||
import { resetDatabase } from "../fixture/db"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
|
||||
afterEach(async () => {
|
||||
await resetDatabase()
|
||||
})
|
||||
|
||||
Log.init({ print: false })
|
||||
|
||||
describe("control-plane/workspace-server SSE", () => {
|
||||
test("streams GlobalBus events and parseSSE reads them", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
const app = WorkspaceServer.App()
|
||||
const stop = new AbortController()
|
||||
const seen: unknown[] = []
|
||||
try {
|
||||
const response = await app.request("/event", {
|
||||
signal: stop.signal,
|
||||
headers: {
|
||||
"x-opencode-workspace": "wrk_test_workspace",
|
||||
"x-opencode-directory": tmp.path,
|
||||
},
|
||||
})
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(response.body).toBeDefined()
|
||||
|
||||
const done = new Promise<void>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
reject(new Error("timed out waiting for workspace.test event"))
|
||||
}, 3000)
|
||||
|
||||
void parseSSE(response.body!, stop.signal, (event) => {
|
||||
seen.push(event)
|
||||
const next = event as { type?: string }
|
||||
if (next.type === "server.connected") {
|
||||
GlobalBus.emit("event", {
|
||||
payload: {
|
||||
type: "workspace.test",
|
||||
properties: { ok: true },
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
if (next.type !== "workspace.test") return
|
||||
clearTimeout(timeout)
|
||||
resolve()
|
||||
}).catch((error) => {
|
||||
clearTimeout(timeout)
|
||||
reject(error)
|
||||
})
|
||||
})
|
||||
|
||||
await done
|
||||
|
||||
expect(seen.some((event) => (event as { type?: string }).type === "server.connected")).toBe(true)
|
||||
expect(seen).toContainEqual({
|
||||
type: "workspace.test",
|
||||
properties: { ok: true },
|
||||
})
|
||||
} finally {
|
||||
stop.abort()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -1,99 +0,0 @@
|
||||
import { afterEach, describe, expect, mock, test } from "bun:test"
|
||||
import { WorkspaceID } from "../../src/control-plane/schema"
|
||||
import { Log } from "../../src/util/log"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
import { Project } from "../../src/project/project"
|
||||
import { Database } from "../../src/storage/db"
|
||||
import { WorkspaceTable } from "../../src/control-plane/workspace.sql"
|
||||
import { GlobalBus } from "../../src/bus/global"
|
||||
import { resetDatabase } from "../fixture/db"
|
||||
import * as adaptors from "../../src/control-plane/adaptors"
|
||||
import type { Adaptor } from "../../src/control-plane/types"
|
||||
|
||||
afterEach(async () => {
|
||||
mock.restore()
|
||||
await resetDatabase()
|
||||
})
|
||||
|
||||
Log.init({ print: false })
|
||||
|
||||
const remote = { type: "testing", name: "remote-a" } as unknown as typeof WorkspaceTable.$inferInsert
|
||||
|
||||
const TestAdaptor: Adaptor = {
|
||||
configure(config) {
|
||||
return config
|
||||
},
|
||||
async create() {
|
||||
throw new Error("not used")
|
||||
},
|
||||
async remove() {},
|
||||
async fetch(_config: unknown, _input: RequestInfo | URL, _init?: RequestInit) {
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
const encoder = new TextEncoder()
|
||||
controller.enqueue(encoder.encode('data: {"type":"remote.ready","properties":{}}\n\n'))
|
||||
controller.close()
|
||||
},
|
||||
})
|
||||
return new Response(body, {
|
||||
status: 200,
|
||||
headers: {
|
||||
"content-type": "text/event-stream",
|
||||
},
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
adaptors.installAdaptor("testing", TestAdaptor)
|
||||
|
||||
describe("control-plane/workspace.startSyncing", () => {
|
||||
test("syncs only remote workspaces and emits remote SSE events", async () => {
|
||||
const { Workspace } = await import("../../src/control-plane/workspace")
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
const { project } = await Project.fromDirectory(tmp.path)
|
||||
|
||||
const id1 = WorkspaceID.ascending()
|
||||
const id2 = WorkspaceID.ascending()
|
||||
|
||||
Database.use((db) =>
|
||||
db
|
||||
.insert(WorkspaceTable)
|
||||
.values([
|
||||
{
|
||||
id: id1,
|
||||
branch: "main",
|
||||
project_id: project.id,
|
||||
type: remote.type,
|
||||
name: remote.name,
|
||||
},
|
||||
{
|
||||
id: id2,
|
||||
branch: "main",
|
||||
project_id: project.id,
|
||||
type: "worktree",
|
||||
directory: tmp.path,
|
||||
name: "local",
|
||||
},
|
||||
])
|
||||
.run(),
|
||||
)
|
||||
|
||||
const done = new Promise<void>((resolve) => {
|
||||
const listener = (event: { directory?: string; payload: { type: string } }) => {
|
||||
if (event.directory !== id1) return
|
||||
if (event.payload.type !== "remote.ready") return
|
||||
GlobalBus.off("event", listener)
|
||||
resolve()
|
||||
}
|
||||
GlobalBus.on("event", listener)
|
||||
})
|
||||
|
||||
const sync = Workspace.startSyncing(project)
|
||||
await Promise.race([
|
||||
done,
|
||||
new Promise((_, reject) => setTimeout(() => reject(new Error("timed out waiting for sync event")), 2000)),
|
||||
])
|
||||
|
||||
await sync.stop()
|
||||
})
|
||||
})
|
||||
@@ -1,10 +1,10 @@
|
||||
import { expect, test } from "bun:test"
|
||||
import { Effect, Layer, ServiceMap } from "effect"
|
||||
import { makeRunPromise } from "../../src/effect/run-service"
|
||||
import { makeRuntime } from "../../src/effect/run-service"
|
||||
|
||||
class Shared extends ServiceMap.Service<Shared, { readonly id: number }>()("@test/Shared") {}
|
||||
|
||||
test("makeRunPromise shares dependent layers through the shared memo map", async () => {
|
||||
test("makeRuntime shares dependent layers through the shared memo map", async () => {
|
||||
let n = 0
|
||||
|
||||
const shared = Layer.effect(
|
||||
@@ -37,8 +37,8 @@ test("makeRunPromise shares dependent layers through the shared memo map", async
|
||||
}),
|
||||
).pipe(Layer.provide(shared))
|
||||
|
||||
const runOne = makeRunPromise(One, one)
|
||||
const runTwo = makeRunPromise(Two, two)
|
||||
const { runPromise: runOne } = makeRuntime(One, one)
|
||||
const { runPromise: runTwo } = makeRuntime(Two, two)
|
||||
|
||||
expect(await runOne((svc) => svc.get())).toBe(1)
|
||||
expect(await runTwo((svc) => svc.get())).toBe(1)
|
||||
|
||||
@@ -2,9 +2,8 @@ import { $ } from "bun"
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import fs from "fs/promises"
|
||||
import path from "path"
|
||||
import { Deferred, Effect, Option } from "effect"
|
||||
import { ConfigProvider, Deferred, Effect, Layer, ManagedRuntime, Option } from "effect"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
import { watcherConfigLayer, withServices } from "../fixture/instance"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { FileWatcher } from "../../src/file/watcher"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
@@ -16,20 +15,33 @@ const describeWatcher = FileWatcher.hasNativeBinding() && !process.env.CI ? desc
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const watcherConfigLayer = ConfigProvider.layer(
|
||||
ConfigProvider.fromUnknown({
|
||||
OPENCODE_EXPERIMENTAL_FILEWATCHER: "true",
|
||||
OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "false",
|
||||
}),
|
||||
)
|
||||
|
||||
type WatcherEvent = { file: string; event: "add" | "change" | "unlink" }
|
||||
|
||||
/** Run `body` with a live FileWatcher service. */
|
||||
function withWatcher<E>(directory: string, body: Effect.Effect<void, E>) {
|
||||
return withServices(
|
||||
return Instance.provide({
|
||||
directory,
|
||||
FileWatcher.layer,
|
||||
async (rt) => {
|
||||
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
|
||||
await Effect.runPromise(ready(directory))
|
||||
await Effect.runPromise(body)
|
||||
fn: async () => {
|
||||
const layer: Layer.Layer<FileWatcher.Service, never, never> = FileWatcher.layer.pipe(
|
||||
Layer.provide(watcherConfigLayer),
|
||||
)
|
||||
const rt = ManagedRuntime.make(layer)
|
||||
try {
|
||||
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
|
||||
await Effect.runPromise(ready(directory))
|
||||
await Effect.runPromise(body)
|
||||
} finally {
|
||||
await rt.dispose()
|
||||
}
|
||||
},
|
||||
{ provide: [watcherConfigLayer] },
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (evt: WatcherEvent) => void) {
|
||||
|
||||
@@ -2,7 +2,10 @@ import { $ } from "bun"
|
||||
import * as fs from "fs/promises"
|
||||
import os from "os"
|
||||
import path from "path"
|
||||
import { Effect, FileSystem, ServiceMap } from "effect"
|
||||
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
|
||||
import type { Config } from "../../src/config/config"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
|
||||
// Strip null bytes from paths (defensive fix for CI environment issues)
|
||||
function sanitizePath(p: string): string {
|
||||
@@ -71,3 +74,68 @@ export async function tmpdir<T>(options?: TmpDirOptions<T>) {
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/** Effectful scoped tmpdir. Cleaned up when the scope closes. Make sure these stay in sync */
|
||||
export function tmpdirScoped(options?: { git?: boolean; config?: Partial<Config.Info> }) {
|
||||
return Effect.gen(function* () {
|
||||
const fs = yield* FileSystem.FileSystem
|
||||
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
|
||||
const dir = yield* fs.makeTempDirectoryScoped({ prefix: "opencode-test-" })
|
||||
|
||||
const git = (...args: string[]) =>
|
||||
spawner.spawn(ChildProcess.make("git", args, { cwd: dir })).pipe(Effect.flatMap((handle) => handle.exitCode))
|
||||
|
||||
if (options?.git) {
|
||||
yield* git("init")
|
||||
yield* git("config", "core.fsmonitor", "false")
|
||||
yield* git("config", "user.email", "test@opencode.test")
|
||||
yield* git("config", "user.name", "Test")
|
||||
yield* git("commit", "--allow-empty", "-m", "root commit")
|
||||
}
|
||||
|
||||
if (options?.config) {
|
||||
yield* fs.writeFileString(
|
||||
path.join(dir, "opencode.json"),
|
||||
JSON.stringify({ $schema: "https://opencode.ai/config.json", ...options.config }),
|
||||
)
|
||||
}
|
||||
|
||||
return dir
|
||||
})
|
||||
}
|
||||
|
||||
export const provideInstance =
|
||||
(directory: string) =>
|
||||
<A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
|
||||
Effect.servicesWith((services: ServiceMap.ServiceMap<R>) =>
|
||||
Effect.promise<A>(async () =>
|
||||
Instance.provide({
|
||||
directory,
|
||||
fn: () => Effect.runPromiseWith(services)(self),
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
export function provideTmpdirInstance<A, E, R>(
|
||||
self: (path: string) => Effect.Effect<A, E, R>,
|
||||
options?: { git?: boolean; config?: Partial<Config.Info> },
|
||||
) {
|
||||
return Effect.gen(function* () {
|
||||
const path = yield* tmpdirScoped(options)
|
||||
let provided = false
|
||||
|
||||
yield* Effect.addFinalizer(() =>
|
||||
provided
|
||||
? Effect.promise(() =>
|
||||
Instance.provide({
|
||||
directory: path,
|
||||
fn: () => Instance.dispose(),
|
||||
}),
|
||||
).pipe(Effect.ignore)
|
||||
: Effect.void,
|
||||
)
|
||||
|
||||
provided = true
|
||||
return yield* self(path).pipe(provideInstance(path))
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,51 +0,0 @@
|
||||
import { ConfigProvider, Layer, ManagedRuntime } from "effect"
|
||||
import { InstanceContext } from "../../src/effect/instance-context"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
|
||||
/** ConfigProvider that enables the experimental file watcher. */
|
||||
export const watcherConfigLayer = ConfigProvider.layer(
|
||||
ConfigProvider.fromUnknown({
|
||||
OPENCODE_EXPERIMENTAL_FILEWATCHER: "true",
|
||||
OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "false",
|
||||
}),
|
||||
)
|
||||
|
||||
/**
|
||||
* Boot an Instance with the given service layers and run `body` with
|
||||
* the ManagedRuntime. Cleanup is automatic — the runtime is disposed
|
||||
* and Instance context is torn down when `body` completes.
|
||||
*
|
||||
* Layers may depend on InstanceContext (provided automatically).
|
||||
* Pass extra layers via `options.provide` (e.g. ConfigProvider.layer).
|
||||
*/
|
||||
export function withServices<S>(
|
||||
directory: string,
|
||||
layer: Layer.Layer<S, any, InstanceContext>,
|
||||
body: (rt: ManagedRuntime.ManagedRuntime<S, never>) => Promise<void>,
|
||||
options?: { provide?: Layer.Layer<never>[] },
|
||||
) {
|
||||
return Instance.provide({
|
||||
directory,
|
||||
fn: async () => {
|
||||
const ctx = Layer.sync(InstanceContext, () =>
|
||||
InstanceContext.of({
|
||||
directory: Instance.directory,
|
||||
worktree: Instance.worktree,
|
||||
project: Instance.project,
|
||||
}),
|
||||
)
|
||||
let resolved: Layer.Layer<S> = layer.pipe(Layer.provide(ctx)) as any
|
||||
if (options?.provide) {
|
||||
for (const l of options.provide) {
|
||||
resolved = resolved.pipe(Layer.provide(l)) as any
|
||||
}
|
||||
}
|
||||
const rt = ManagedRuntime.make(resolved)
|
||||
try {
|
||||
await body(rt)
|
||||
} finally {
|
||||
await rt.dispose()
|
||||
}
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -1,172 +1,182 @@
|
||||
import { Effect } from "effect"
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
import { withServices } from "../fixture/instance"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { File } from "../../src/file"
|
||||
import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
|
||||
import { describe, expect } from "bun:test"
|
||||
import { Effect, Layer } from "effect"
|
||||
import { provideTmpdirInstance } from "../fixture/fixture"
|
||||
import { testEffect } from "../lib/effect"
|
||||
import { Format } from "../../src/format"
|
||||
import * as Formatter from "../../src/format/formatter"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
|
||||
const node = NodeChildProcessSpawner.layer.pipe(
|
||||
Layer.provideMerge(Layer.mergeAll(NodeFileSystem.layer, NodePath.layer)),
|
||||
)
|
||||
|
||||
const it = testEffect(Layer.mergeAll(Format.layer, node))
|
||||
|
||||
describe("Format", () => {
|
||||
afterEach(async () => {
|
||||
await Instance.disposeAll()
|
||||
})
|
||||
it.effect("status() returns built-in formatters when no config overrides", () =>
|
||||
provideTmpdirInstance(() =>
|
||||
Format.Service.use((fmt) =>
|
||||
Effect.gen(function* () {
|
||||
const statuses = yield* fmt.status()
|
||||
expect(Array.isArray(statuses)).toBe(true)
|
||||
expect(statuses.length).toBeGreaterThan(0)
|
||||
|
||||
test("status() returns built-in formatters when no config overrides", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
for (const item of statuses) {
|
||||
expect(typeof item.name).toBe("string")
|
||||
expect(Array.isArray(item.extensions)).toBe(true)
|
||||
expect(typeof item.enabled).toBe("boolean")
|
||||
}
|
||||
|
||||
await withServices(tmp.path, Format.layer, async (rt) => {
|
||||
const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
|
||||
expect(Array.isArray(statuses)).toBe(true)
|
||||
expect(statuses.length).toBeGreaterThan(0)
|
||||
const gofmt = statuses.find((item) => item.name === "gofmt")
|
||||
expect(gofmt).toBeDefined()
|
||||
expect(gofmt!.extensions).toContain(".go")
|
||||
}),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
for (const s of statuses) {
|
||||
expect(typeof s.name).toBe("string")
|
||||
expect(Array.isArray(s.extensions)).toBe(true)
|
||||
expect(typeof s.enabled).toBe("boolean")
|
||||
}
|
||||
it.effect("status() returns empty list when formatter is disabled", () =>
|
||||
provideTmpdirInstance(
|
||||
() =>
|
||||
Format.Service.use((fmt) =>
|
||||
Effect.gen(function* () {
|
||||
expect(yield* fmt.status()).toEqual([])
|
||||
}),
|
||||
),
|
||||
{ config: { formatter: false } },
|
||||
),
|
||||
)
|
||||
|
||||
const gofmt = statuses.find((s) => s.name === "gofmt")
|
||||
expect(gofmt).toBeDefined()
|
||||
expect(gofmt!.extensions).toContain(".go")
|
||||
})
|
||||
})
|
||||
|
||||
test("status() returns empty list when formatter is disabled", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
config: { formatter: false },
|
||||
})
|
||||
|
||||
await withServices(tmp.path, Format.layer, async (rt) => {
|
||||
const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
|
||||
expect(statuses).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
test("status() excludes formatters marked as disabled in config", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
config: {
|
||||
formatter: {
|
||||
gofmt: { disabled: true },
|
||||
it.effect("status() excludes formatters marked as disabled in config", () =>
|
||||
provideTmpdirInstance(
|
||||
() =>
|
||||
Format.Service.use((fmt) =>
|
||||
Effect.gen(function* () {
|
||||
const statuses = yield* fmt.status()
|
||||
const gofmt = statuses.find((item) => item.name === "gofmt")
|
||||
expect(gofmt).toBeUndefined()
|
||||
}),
|
||||
),
|
||||
{
|
||||
config: {
|
||||
formatter: {
|
||||
gofmt: { disabled: true },
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
),
|
||||
)
|
||||
|
||||
await withServices(tmp.path, Format.layer, async (rt) => {
|
||||
const statuses = await rt.runPromise(Format.Service.use((s) => s.status()))
|
||||
const gofmt = statuses.find((s) => s.name === "gofmt")
|
||||
expect(gofmt).toBeUndefined()
|
||||
})
|
||||
})
|
||||
it.effect("service initializes without error", () =>
|
||||
provideTmpdirInstance(() => Format.Service.use(() => Effect.void)),
|
||||
)
|
||||
|
||||
test("service initializes without error", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await withServices(tmp.path, Format.layer, async (rt) => {
|
||||
await rt.runPromise(Format.Service.use(() => Effect.void))
|
||||
})
|
||||
})
|
||||
|
||||
test("status() initializes formatter state per directory", async () => {
|
||||
await using off = await tmpdir({
|
||||
config: { formatter: false },
|
||||
})
|
||||
await using on = await tmpdir()
|
||||
|
||||
const a = await Instance.provide({
|
||||
directory: off.path,
|
||||
fn: () => Format.status(),
|
||||
})
|
||||
const b = await Instance.provide({
|
||||
directory: on.path,
|
||||
fn: () => Format.status(),
|
||||
})
|
||||
|
||||
expect(a).toEqual([])
|
||||
expect(b.length).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
test("runs enabled checks for matching formatters in parallel", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
const file = `${tmp.path}/test.parallel`
|
||||
await Bun.write(file, "x")
|
||||
|
||||
const one = {
|
||||
extensions: Formatter.gofmt.extensions,
|
||||
enabled: Formatter.gofmt.enabled,
|
||||
command: Formatter.gofmt.command,
|
||||
}
|
||||
const two = {
|
||||
extensions: Formatter.mix.extensions,
|
||||
enabled: Formatter.mix.enabled,
|
||||
command: Formatter.mix.command,
|
||||
}
|
||||
|
||||
let active = 0
|
||||
let max = 0
|
||||
|
||||
Formatter.gofmt.extensions = [".parallel"]
|
||||
Formatter.mix.extensions = [".parallel"]
|
||||
Formatter.gofmt.command = ["sh", "-c", "true"]
|
||||
Formatter.mix.command = ["sh", "-c", "true"]
|
||||
Formatter.gofmt.enabled = async () => {
|
||||
active++
|
||||
max = Math.max(max, active)
|
||||
await Bun.sleep(20)
|
||||
active--
|
||||
return true
|
||||
}
|
||||
Formatter.mix.enabled = async () => {
|
||||
active++
|
||||
max = Math.max(max, active)
|
||||
await Bun.sleep(20)
|
||||
active--
|
||||
return true
|
||||
}
|
||||
|
||||
try {
|
||||
await withServices(tmp.path, Format.layer, async (rt) => {
|
||||
await rt.runPromise(Format.Service.use((s) => s.init()))
|
||||
await Bus.publish(File.Event.Edited, { file })
|
||||
it.effect("status() initializes formatter state per directory", () =>
|
||||
Effect.gen(function* () {
|
||||
const a = yield* provideTmpdirInstance(() => Format.Service.use((fmt) => fmt.status()), {
|
||||
config: { formatter: false },
|
||||
})
|
||||
} finally {
|
||||
Formatter.gofmt.extensions = one.extensions
|
||||
Formatter.gofmt.enabled = one.enabled
|
||||
Formatter.gofmt.command = one.command
|
||||
Formatter.mix.extensions = two.extensions
|
||||
Formatter.mix.enabled = two.enabled
|
||||
Formatter.mix.command = two.command
|
||||
}
|
||||
const b = yield* provideTmpdirInstance(() => Format.Service.use((fmt) => fmt.status()))
|
||||
|
||||
expect(max).toBe(2)
|
||||
})
|
||||
expect(a).toEqual([])
|
||||
expect(b.length).toBeGreaterThan(0)
|
||||
}),
|
||||
)
|
||||
|
||||
test("runs matching formatters sequentially for the same file", async () => {
|
||||
await using tmp = await tmpdir({
|
||||
config: {
|
||||
formatter: {
|
||||
first: {
|
||||
command: ["sh", "-c", 'sleep 0.05; v=$(cat "$1"); printf \'%sA\' "$v" > "$1"', "sh", "$FILE"],
|
||||
extensions: [".seq"],
|
||||
},
|
||||
second: {
|
||||
command: ["sh", "-c", 'v=$(cat "$1"); printf \'%sB\' "$v" > "$1"', "sh", "$FILE"],
|
||||
extensions: [".seq"],
|
||||
it.effect("runs enabled checks for matching formatters in parallel", () =>
|
||||
provideTmpdirInstance((path) =>
|
||||
Effect.gen(function* () {
|
||||
const file = `${path}/test.parallel`
|
||||
yield* Effect.promise(() => Bun.write(file, "x"))
|
||||
|
||||
const one = {
|
||||
extensions: Formatter.gofmt.extensions,
|
||||
enabled: Formatter.gofmt.enabled,
|
||||
command: Formatter.gofmt.command,
|
||||
}
|
||||
const two = {
|
||||
extensions: Formatter.mix.extensions,
|
||||
enabled: Formatter.mix.enabled,
|
||||
command: Formatter.mix.command,
|
||||
}
|
||||
|
||||
let active = 0
|
||||
let max = 0
|
||||
|
||||
yield* Effect.acquireUseRelease(
|
||||
Effect.sync(() => {
|
||||
Formatter.gofmt.extensions = [".parallel"]
|
||||
Formatter.mix.extensions = [".parallel"]
|
||||
Formatter.gofmt.command = ["sh", "-c", "true"]
|
||||
Formatter.mix.command = ["sh", "-c", "true"]
|
||||
Formatter.gofmt.enabled = async () => {
|
||||
active++
|
||||
max = Math.max(max, active)
|
||||
await Bun.sleep(20)
|
||||
active--
|
||||
return true
|
||||
}
|
||||
Formatter.mix.enabled = async () => {
|
||||
active++
|
||||
max = Math.max(max, active)
|
||||
await Bun.sleep(20)
|
||||
active--
|
||||
return true
|
||||
}
|
||||
}),
|
||||
() =>
|
||||
Format.Service.use((fmt) =>
|
||||
Effect.gen(function* () {
|
||||
yield* fmt.init()
|
||||
yield* fmt.file(file)
|
||||
}),
|
||||
),
|
||||
() =>
|
||||
Effect.sync(() => {
|
||||
Formatter.gofmt.extensions = one.extensions
|
||||
Formatter.gofmt.enabled = one.enabled
|
||||
Formatter.gofmt.command = one.command
|
||||
Formatter.mix.extensions = two.extensions
|
||||
Formatter.mix.enabled = two.enabled
|
||||
Formatter.mix.command = two.command
|
||||
}),
|
||||
)
|
||||
|
||||
expect(max).toBe(2)
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("runs matching formatters sequentially for the same file", () =>
|
||||
provideTmpdirInstance(
|
||||
(path) =>
|
||||
Effect.gen(function* () {
|
||||
const file = `${path}/test.seq`
|
||||
yield* Effect.promise(() => Bun.write(file, "x"))
|
||||
|
||||
yield* Format.Service.use((fmt) =>
|
||||
Effect.gen(function* () {
|
||||
yield* fmt.init()
|
||||
yield* fmt.file(file)
|
||||
}),
|
||||
)
|
||||
|
||||
expect(yield* Effect.promise(() => Bun.file(file).text())).toBe("xAB")
|
||||
}),
|
||||
{
|
||||
config: {
|
||||
formatter: {
|
||||
first: {
|
||||
command: ["sh", "-c", 'sleep 0.05; v=$(cat "$1"); printf \'%sA\' "$v" > "$1"', "sh", "$FILE"],
|
||||
extensions: [".seq"],
|
||||
},
|
||||
second: {
|
||||
command: ["sh", "-c", 'v=$(cat "$1"); printf \'%sB\' "$v" > "$1"', "sh", "$FILE"],
|
||||
extensions: [".seq"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
const file = `${tmp.path}/test.seq`
|
||||
await Bun.write(file, "x")
|
||||
|
||||
await withServices(tmp.path, Format.layer, async (rt) => {
|
||||
await rt.runPromise(Format.Service.use((s) => s.init()))
|
||||
await Bus.publish(File.Event.Edited, { file })
|
||||
})
|
||||
|
||||
expect(await Bun.file(file).text()).toBe("xAB")
|
||||
})
|
||||
),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import { describe, expect, spyOn, test } from "bun:test"
|
||||
import path from "path"
|
||||
import * as Lsp from "../../src/lsp/index"
|
||||
import { LSPServer } from "../../src/lsp/server"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
|
||||
describe("lsp.spawn", () => {
|
||||
test("does not spawn builtin LSP for files outside instance", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const spy = spyOn(LSPServer.Typescript, "spawn").mockResolvedValue(undefined)
|
||||
|
||||
try {
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: async () => {
|
||||
await Lsp.LSP.touchFile(path.join(tmp.path, "..", "outside.ts"))
|
||||
await Lsp.LSP.hover({
|
||||
file: path.join(tmp.path, "..", "hover.ts"),
|
||||
line: 0,
|
||||
character: 0,
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
expect(spy).toHaveBeenCalledTimes(0)
|
||||
} finally {
|
||||
spy.mockRestore()
|
||||
await Instance.disposeAll()
|
||||
}
|
||||
})
|
||||
|
||||
test("would spawn builtin LSP for files inside instance", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
const spy = spyOn(LSPServer.Typescript, "spawn").mockResolvedValue(undefined)
|
||||
|
||||
try {
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: async () => {
|
||||
await Lsp.LSP.hover({
|
||||
file: path.join(tmp.path, "src", "inside.ts"),
|
||||
line: 0,
|
||||
character: 0,
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
expect(spy).toHaveBeenCalledTimes(1)
|
||||
} finally {
|
||||
spy.mockRestore()
|
||||
await Instance.disposeAll()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,147 @@
|
||||
import { describe, expect, test, spyOn, beforeEach, afterEach } from "bun:test"
|
||||
import path from "path"
|
||||
import * as Lsp from "../../src/lsp/index"
|
||||
import { LSPServer } from "../../src/lsp/server"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
|
||||
function withInstance(fn: (dir: string) => Promise<void>) {
|
||||
return async () => {
|
||||
await using tmp = await tmpdir()
|
||||
try {
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: () => fn(tmp.path),
|
||||
})
|
||||
} finally {
|
||||
await Instance.disposeAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe("LSP service lifecycle", () => {
|
||||
let spawnSpy: ReturnType<typeof spyOn>
|
||||
|
||||
beforeEach(() => {
|
||||
spawnSpy = spyOn(LSPServer.Typescript, "spawn").mockResolvedValue(undefined)
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
spawnSpy.mockRestore()
|
||||
})
|
||||
|
||||
test(
|
||||
"init() completes without error",
|
||||
withInstance(async () => {
|
||||
await Lsp.LSP.init()
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"status() returns empty array initially",
|
||||
withInstance(async () => {
|
||||
const result = await Lsp.LSP.status()
|
||||
expect(Array.isArray(result)).toBe(true)
|
||||
expect(result.length).toBe(0)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"diagnostics() returns empty object initially",
|
||||
withInstance(async () => {
|
||||
const result = await Lsp.LSP.diagnostics()
|
||||
expect(typeof result).toBe("object")
|
||||
expect(Object.keys(result).length).toBe(0)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"hasClients() returns true for .ts files in instance",
|
||||
withInstance(async (dir) => {
|
||||
const result = await Lsp.LSP.hasClients(path.join(dir, "test.ts"))
|
||||
expect(result).toBe(true)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"hasClients() returns false for files outside instance",
|
||||
withInstance(async (dir) => {
|
||||
const result = await Lsp.LSP.hasClients(path.join(dir, "..", "outside.ts"))
|
||||
// hasClients checks servers but doesn't check containsPath — getClients does
|
||||
// So hasClients may return true even for outside files (it checks extension + root)
|
||||
// The guard is in getClients, not hasClients
|
||||
expect(typeof result).toBe("boolean")
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"workspaceSymbol() returns empty array with no clients",
|
||||
withInstance(async () => {
|
||||
const result = await Lsp.LSP.workspaceSymbol("test")
|
||||
expect(Array.isArray(result)).toBe(true)
|
||||
expect(result.length).toBe(0)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"definition() returns empty array for unknown file",
|
||||
withInstance(async (dir) => {
|
||||
const result = await Lsp.LSP.definition({
|
||||
file: path.join(dir, "nonexistent.ts"),
|
||||
line: 0,
|
||||
character: 0,
|
||||
})
|
||||
expect(Array.isArray(result)).toBe(true)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"references() returns empty array for unknown file",
|
||||
withInstance(async (dir) => {
|
||||
const result = await Lsp.LSP.references({
|
||||
file: path.join(dir, "nonexistent.ts"),
|
||||
line: 0,
|
||||
character: 0,
|
||||
})
|
||||
expect(Array.isArray(result)).toBe(true)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"multiple init() calls are idempotent",
|
||||
withInstance(async () => {
|
||||
await Lsp.LSP.init()
|
||||
await Lsp.LSP.init()
|
||||
await Lsp.LSP.init()
|
||||
// Should not throw or create duplicate state
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("LSP.Diagnostic", () => {
|
||||
test("pretty() formats error diagnostic", () => {
|
||||
const result = Lsp.LSP.Diagnostic.pretty({
|
||||
range: { start: { line: 9, character: 4 }, end: { line: 9, character: 10 } },
|
||||
message: "Type 'string' is not assignable to type 'number'",
|
||||
severity: 1,
|
||||
} as any)
|
||||
expect(result).toBe("ERROR [10:5] Type 'string' is not assignable to type 'number'")
|
||||
})
|
||||
|
||||
test("pretty() formats warning diagnostic", () => {
|
||||
const result = Lsp.LSP.Diagnostic.pretty({
|
||||
range: { start: { line: 0, character: 0 }, end: { line: 0, character: 5 } },
|
||||
message: "Unused variable",
|
||||
severity: 2,
|
||||
} as any)
|
||||
expect(result).toBe("WARN [1:1] Unused variable")
|
||||
})
|
||||
|
||||
test("pretty() defaults to ERROR when no severity", () => {
|
||||
const result = Lsp.LSP.Diagnostic.pretty({
|
||||
range: { start: { line: 0, character: 0 }, end: { line: 0, character: 1 } },
|
||||
message: "Something wrong",
|
||||
} as any)
|
||||
expect(result).toBe("ERROR [1:1] Something wrong")
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,750 @@
|
||||
import { test, expect, mock, beforeEach } from "bun:test"
|
||||
|
||||
// --- Mock infrastructure ---
|
||||
|
||||
// Per-client state for controlling mock behavior
|
||||
interface MockClientState {
|
||||
tools: Array<{ name: string; description?: string; inputSchema: object }>
|
||||
listToolsCalls: number
|
||||
listToolsShouldFail: boolean
|
||||
listToolsError: string
|
||||
listPromptsShouldFail: boolean
|
||||
listResourcesShouldFail: boolean
|
||||
prompts: Array<{ name: string; description?: string }>
|
||||
resources: Array<{ name: string; uri: string; description?: string }>
|
||||
closed: boolean
|
||||
notificationHandlers: Map<unknown, (...args: any[]) => any>
|
||||
}
|
||||
|
||||
const clientStates = new Map<string, MockClientState>()
|
||||
let lastCreatedClientName: string | undefined
|
||||
let connectShouldFail = false
|
||||
let connectShouldHang = false
|
||||
let connectError = "Mock transport cannot connect"
|
||||
// Tracks how many Client instances were created (detects leaks)
|
||||
let clientCreateCount = 0
|
||||
// Tracks how many times transport.close() is called across all mock transports
|
||||
let transportCloseCount = 0
|
||||
|
||||
function getOrCreateClientState(name?: string): MockClientState {
|
||||
const key = name ?? "default"
|
||||
let state = clientStates.get(key)
|
||||
if (!state) {
|
||||
state = {
|
||||
tools: [{ name: "test_tool", description: "A test tool", inputSchema: { type: "object", properties: {} } }],
|
||||
listToolsCalls: 0,
|
||||
listToolsShouldFail: false,
|
||||
listToolsError: "listTools failed",
|
||||
listPromptsShouldFail: false,
|
||||
listResourcesShouldFail: false,
|
||||
prompts: [],
|
||||
resources: [],
|
||||
closed: false,
|
||||
notificationHandlers: new Map(),
|
||||
}
|
||||
clientStates.set(key, state)
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
// Mock transport that succeeds or fails based on connectShouldFail / connectShouldHang
|
||||
class MockStdioTransport {
|
||||
stderr: null = null
|
||||
pid = 12345
|
||||
constructor(_opts: any) {}
|
||||
async start() {
|
||||
if (connectShouldHang) return new Promise<void>(() => {}) // never resolves
|
||||
if (connectShouldFail) throw new Error(connectError)
|
||||
}
|
||||
async close() {
|
||||
transportCloseCount++
|
||||
}
|
||||
}
|
||||
|
||||
class MockStreamableHTTP {
|
||||
constructor(_url: URL, _opts?: any) {}
|
||||
async start() {
|
||||
if (connectShouldHang) return new Promise<void>(() => {}) // never resolves
|
||||
if (connectShouldFail) throw new Error(connectError)
|
||||
}
|
||||
async close() {
|
||||
transportCloseCount++
|
||||
}
|
||||
async finishAuth() {}
|
||||
}
|
||||
|
||||
class MockSSE {
|
||||
constructor(_url: URL, _opts?: any) {}
|
||||
async start() {
|
||||
if (connectShouldHang) return new Promise<void>(() => {}) // never resolves
|
||||
if (connectShouldFail) throw new Error(connectError)
|
||||
}
|
||||
async close() {
|
||||
transportCloseCount++
|
||||
}
|
||||
}
|
||||
|
||||
mock.module("@modelcontextprotocol/sdk/client/stdio.js", () => ({
|
||||
StdioClientTransport: MockStdioTransport,
|
||||
}))
|
||||
|
||||
mock.module("@modelcontextprotocol/sdk/client/streamableHttp.js", () => ({
|
||||
StreamableHTTPClientTransport: MockStreamableHTTP,
|
||||
}))
|
||||
|
||||
mock.module("@modelcontextprotocol/sdk/client/sse.js", () => ({
|
||||
SSEClientTransport: MockSSE,
|
||||
}))
|
||||
|
||||
mock.module("@modelcontextprotocol/sdk/client/auth.js", () => ({
|
||||
UnauthorizedError: class extends Error {
|
||||
constructor() {
|
||||
super("Unauthorized")
|
||||
}
|
||||
},
|
||||
}))
|
||||
|
||||
// Mock Client that delegates to per-name MockClientState
|
||||
mock.module("@modelcontextprotocol/sdk/client/index.js", () => ({
|
||||
Client: class MockClient {
|
||||
_state!: MockClientState
|
||||
transport: any
|
||||
|
||||
constructor(_opts: any) {
|
||||
clientCreateCount++
|
||||
}
|
||||
|
||||
async connect(transport: { start: () => Promise<void> }) {
|
||||
this.transport = transport
|
||||
await transport.start()
|
||||
// After successful connect, bind to the last-created client name
|
||||
this._state = getOrCreateClientState(lastCreatedClientName)
|
||||
}
|
||||
|
||||
setNotificationHandler(schema: unknown, handler: (...args: any[]) => any) {
|
||||
this._state?.notificationHandlers.set(schema, handler)
|
||||
}
|
||||
|
||||
async listTools() {
|
||||
if (this._state) this._state.listToolsCalls++
|
||||
if (this._state?.listToolsShouldFail) {
|
||||
throw new Error(this._state.listToolsError)
|
||||
}
|
||||
return { tools: this._state?.tools ?? [] }
|
||||
}
|
||||
|
||||
async listPrompts() {
|
||||
if (this._state?.listPromptsShouldFail) {
|
||||
throw new Error("listPrompts failed")
|
||||
}
|
||||
return { prompts: this._state?.prompts ?? [] }
|
||||
}
|
||||
|
||||
async listResources() {
|
||||
if (this._state?.listResourcesShouldFail) {
|
||||
throw new Error("listResources failed")
|
||||
}
|
||||
return { resources: this._state?.resources ?? [] }
|
||||
}
|
||||
|
||||
async close() {
|
||||
if (this._state) this._state.closed = true
|
||||
}
|
||||
},
|
||||
}))
|
||||
|
||||
beforeEach(() => {
|
||||
clientStates.clear()
|
||||
lastCreatedClientName = undefined
|
||||
connectShouldFail = false
|
||||
connectShouldHang = false
|
||||
connectError = "Mock transport cannot connect"
|
||||
clientCreateCount = 0
|
||||
transportCloseCount = 0
|
||||
})
|
||||
|
||||
// Import after mocks
|
||||
const { MCP } = await import("../../src/mcp/index")
|
||||
const { Instance } = await import("../../src/project/instance")
|
||||
const { tmpdir } = await import("../fixture/fixture")
|
||||
|
||||
// --- Helper ---
|
||||
|
||||
function withInstance(config: Record<string, any>, fn: () => Promise<void>) {
|
||||
return async () => {
|
||||
await using tmp = await tmpdir({
|
||||
init: async (dir) => {
|
||||
await Bun.write(
|
||||
`${dir}/opencode.json`,
|
||||
JSON.stringify({
|
||||
$schema: "https://opencode.ai/config.json",
|
||||
mcp: config,
|
||||
}),
|
||||
)
|
||||
},
|
||||
})
|
||||
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: async () => {
|
||||
await fn()
|
||||
// dispose instance to clean up state between tests
|
||||
await Instance.dispose()
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Test: tools() are cached after connect
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"tools() reuses cached tool definitions after connect",
|
||||
withInstance({}, async () => {
|
||||
lastCreatedClientName = "my-server"
|
||||
const serverState = getOrCreateClientState("my-server")
|
||||
serverState.tools = [
|
||||
{ name: "do_thing", description: "does a thing", inputSchema: { type: "object", properties: {} } },
|
||||
]
|
||||
|
||||
// First: add the server successfully
|
||||
const addResult = await MCP.add("my-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
expect((addResult.status as any)["my-server"]?.status ?? (addResult.status as any).status).toBe("connected")
|
||||
|
||||
expect(serverState.listToolsCalls).toBe(1)
|
||||
|
||||
const toolsA = await MCP.tools()
|
||||
const toolsB = await MCP.tools()
|
||||
expect(Object.keys(toolsA).length).toBeGreaterThan(0)
|
||||
expect(Object.keys(toolsB).length).toBeGreaterThan(0)
|
||||
expect(serverState.listToolsCalls).toBe(1)
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: tool change notifications refresh the cache
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"tool change notifications refresh cached tool definitions",
|
||||
withInstance({}, async () => {
|
||||
lastCreatedClientName = "status-server"
|
||||
const serverState = getOrCreateClientState("status-server")
|
||||
|
||||
await MCP.add("status-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
const before = await MCP.tools()
|
||||
expect(Object.keys(before).some((key) => key.includes("test_tool"))).toBe(true)
|
||||
expect(serverState.listToolsCalls).toBe(1)
|
||||
|
||||
serverState.tools = [{ name: "next_tool", description: "next", inputSchema: { type: "object", properties: {} } }]
|
||||
|
||||
const handler = Array.from(serverState.notificationHandlers.values())[0]
|
||||
expect(handler).toBeDefined()
|
||||
await handler?.()
|
||||
|
||||
const after = await MCP.tools()
|
||||
expect(Object.keys(after).some((key) => key.includes("next_tool"))).toBe(true)
|
||||
expect(Object.keys(after).some((key) => key.includes("test_tool"))).toBe(false)
|
||||
expect(serverState.listToolsCalls).toBe(2)
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: connect() / disconnect() lifecycle
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"disconnect sets status to disabled and removes client",
|
||||
withInstance(
|
||||
{
|
||||
"disc-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "disc-server"
|
||||
getOrCreateClientState("disc-server")
|
||||
|
||||
await MCP.add("disc-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
const statusBefore = await MCP.status()
|
||||
expect(statusBefore["disc-server"]?.status).toBe("connected")
|
||||
|
||||
await MCP.disconnect("disc-server")
|
||||
|
||||
const statusAfter = await MCP.status()
|
||||
expect(statusAfter["disc-server"]?.status).toBe("disabled")
|
||||
|
||||
// Tools should be empty after disconnect
|
||||
const tools = await MCP.tools()
|
||||
const serverTools = Object.keys(tools).filter((k) => k.startsWith("disc-server"))
|
||||
expect(serverTools.length).toBe(0)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
test(
|
||||
"connect() after disconnect() re-establishes the server",
|
||||
withInstance(
|
||||
{
|
||||
"reconn-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "reconn-server"
|
||||
const serverState = getOrCreateClientState("reconn-server")
|
||||
serverState.tools = [{ name: "my_tool", description: "a tool", inputSchema: { type: "object", properties: {} } }]
|
||||
|
||||
await MCP.add("reconn-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
await MCP.disconnect("reconn-server")
|
||||
expect((await MCP.status())["reconn-server"]?.status).toBe("disabled")
|
||||
|
||||
// Reconnect
|
||||
await MCP.connect("reconn-server")
|
||||
expect((await MCP.status())["reconn-server"]?.status).toBe("connected")
|
||||
|
||||
const tools = await MCP.tools()
|
||||
expect(Object.keys(tools).some((k) => k.includes("my_tool"))).toBe(true)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: add() closes existing client before replacing
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"add() closes the old client when replacing a server",
|
||||
// Don't put the server in config — add it dynamically so we control
|
||||
// exactly which client instance is "first" vs "second".
|
||||
withInstance({}, async () => {
|
||||
lastCreatedClientName = "replace-server"
|
||||
const firstState = getOrCreateClientState("replace-server")
|
||||
|
||||
await MCP.add("replace-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
expect(firstState.closed).toBe(false)
|
||||
|
||||
// Create new state for second client
|
||||
clientStates.delete("replace-server")
|
||||
const secondState = getOrCreateClientState("replace-server")
|
||||
|
||||
// Re-add should close the first client
|
||||
await MCP.add("replace-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
expect(firstState.closed).toBe(true)
|
||||
expect(secondState.closed).toBe(false)
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: state init with mixed success/failure
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"init connects available servers even when one fails",
|
||||
withInstance(
|
||||
{
|
||||
"good-server": {
|
||||
type: "local",
|
||||
command: ["echo", "good"],
|
||||
},
|
||||
"bad-server": {
|
||||
type: "local",
|
||||
command: ["echo", "bad"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
// Set up good server
|
||||
const goodState = getOrCreateClientState("good-server")
|
||||
goodState.tools = [{ name: "good_tool", description: "works", inputSchema: { type: "object", properties: {} } }]
|
||||
|
||||
// Set up bad server - will fail on listTools during create()
|
||||
const badState = getOrCreateClientState("bad-server")
|
||||
badState.listToolsShouldFail = true
|
||||
|
||||
// Add good server first
|
||||
lastCreatedClientName = "good-server"
|
||||
await MCP.add("good-server", {
|
||||
type: "local",
|
||||
command: ["echo", "good"],
|
||||
})
|
||||
|
||||
// Add bad server - should fail but not affect good server
|
||||
lastCreatedClientName = "bad-server"
|
||||
await MCP.add("bad-server", {
|
||||
type: "local",
|
||||
command: ["echo", "bad"],
|
||||
})
|
||||
|
||||
const status = await MCP.status()
|
||||
expect(status["good-server"]?.status).toBe("connected")
|
||||
expect(status["bad-server"]?.status).toBe("failed")
|
||||
|
||||
// Good server's tools should still be available
|
||||
const tools = await MCP.tools()
|
||||
expect(Object.keys(tools).some((k) => k.includes("good_tool"))).toBe(true)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: disabled server via config
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"disabled server is marked as disabled without attempting connection",
|
||||
withInstance(
|
||||
{
|
||||
"disabled-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
enabled: false,
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
const countBefore = clientCreateCount
|
||||
|
||||
await MCP.add("disabled-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
enabled: false,
|
||||
} as any)
|
||||
|
||||
// No client should have been created
|
||||
expect(clientCreateCount).toBe(countBefore)
|
||||
|
||||
const status = await MCP.status()
|
||||
expect(status["disabled-server"]?.status).toBe("disabled")
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: prompts() and resources()
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"prompts() returns prompts from connected servers",
|
||||
withInstance(
|
||||
{
|
||||
"prompt-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "prompt-server"
|
||||
const serverState = getOrCreateClientState("prompt-server")
|
||||
serverState.prompts = [{ name: "my-prompt", description: "A test prompt" }]
|
||||
|
||||
await MCP.add("prompt-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
const prompts = await MCP.prompts()
|
||||
expect(Object.keys(prompts).length).toBe(1)
|
||||
const key = Object.keys(prompts)[0]
|
||||
expect(key).toContain("prompt-server")
|
||||
expect(key).toContain("my-prompt")
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
test(
|
||||
"resources() returns resources from connected servers",
|
||||
withInstance(
|
||||
{
|
||||
"resource-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "resource-server"
|
||||
const serverState = getOrCreateClientState("resource-server")
|
||||
serverState.resources = [{ name: "my-resource", uri: "file:///test.txt", description: "A test resource" }]
|
||||
|
||||
await MCP.add("resource-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
const resources = await MCP.resources()
|
||||
expect(Object.keys(resources).length).toBe(1)
|
||||
const key = Object.keys(resources)[0]
|
||||
expect(key).toContain("resource-server")
|
||||
expect(key).toContain("my-resource")
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
test(
|
||||
"prompts() skips disconnected servers",
|
||||
withInstance(
|
||||
{
|
||||
"prompt-disc-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "prompt-disc-server"
|
||||
const serverState = getOrCreateClientState("prompt-disc-server")
|
||||
serverState.prompts = [{ name: "hidden-prompt", description: "Should not appear" }]
|
||||
|
||||
await MCP.add("prompt-disc-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
await MCP.disconnect("prompt-disc-server")
|
||||
|
||||
const prompts = await MCP.prompts()
|
||||
expect(Object.keys(prompts).length).toBe(0)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: connect() on nonexistent server
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"connect() on nonexistent server does not throw",
|
||||
withInstance({}, async () => {
|
||||
// Should not throw
|
||||
await MCP.connect("nonexistent")
|
||||
const status = await MCP.status()
|
||||
expect(status["nonexistent"]).toBeUndefined()
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: disconnect() on nonexistent server
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"disconnect() on nonexistent server does not throw",
|
||||
withInstance({}, async () => {
|
||||
await MCP.disconnect("nonexistent")
|
||||
// Should complete without error
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: tools() with no MCP servers configured
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"tools() returns empty when no MCP servers are configured",
|
||||
withInstance({}, async () => {
|
||||
const tools = await MCP.tools()
|
||||
expect(Object.keys(tools).length).toBe(0)
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: connect failure during create()
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"server that fails to connect is marked as failed",
|
||||
withInstance(
|
||||
{
|
||||
"fail-connect": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "fail-connect"
|
||||
getOrCreateClientState("fail-connect")
|
||||
connectShouldFail = true
|
||||
connectError = "Connection refused"
|
||||
|
||||
await MCP.add("fail-connect", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
const status = await MCP.status()
|
||||
expect(status["fail-connect"]?.status).toBe("failed")
|
||||
if (status["fail-connect"]?.status === "failed") {
|
||||
expect(status["fail-connect"].error).toContain("Connection refused")
|
||||
}
|
||||
|
||||
// No tools should be available
|
||||
const tools = await MCP.tools()
|
||||
expect(Object.keys(tools).length).toBe(0)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Bug #5: McpOAuthCallback.cancelPending uses wrong key
|
||||
// ========================================================================
|
||||
|
||||
test("McpOAuthCallback.cancelPending is keyed by mcpName but pendingAuths uses oauthState", async () => {
|
||||
const { McpOAuthCallback } = await import("../../src/mcp/oauth-callback")
|
||||
|
||||
// Register a pending auth with an oauthState key, associated to an mcpName
|
||||
const oauthState = "abc123hexstate"
|
||||
const callbackPromise = McpOAuthCallback.waitForCallback(oauthState, "my-mcp-server")
|
||||
|
||||
// cancelPending is called with mcpName — should find the entry via reverse index
|
||||
McpOAuthCallback.cancelPending("my-mcp-server")
|
||||
|
||||
// The callback should still be pending because cancelPending looked up
|
||||
// "my-mcp-server" in a map keyed by "abc123hexstate"
|
||||
let resolved = false
|
||||
let rejected = false
|
||||
callbackPromise.then(() => (resolved = true)).catch(() => (rejected = true))
|
||||
|
||||
// Give it a tick
|
||||
await new Promise((r) => setTimeout(r, 50))
|
||||
|
||||
// cancelPending("my-mcp-server") should have rejected the pending callback
|
||||
expect(rejected).toBe(true)
|
||||
|
||||
await McpOAuthCallback.stop()
|
||||
})
|
||||
|
||||
// ========================================================================
|
||||
// Test: multiple tools from same server get correct name prefixes
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"tools() prefixes tool names with sanitized server name",
|
||||
withInstance(
|
||||
{
|
||||
"my.special-server": {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
},
|
||||
},
|
||||
async () => {
|
||||
lastCreatedClientName = "my.special-server"
|
||||
const serverState = getOrCreateClientState("my.special-server")
|
||||
serverState.tools = [
|
||||
{ name: "tool-a", description: "Tool A", inputSchema: { type: "object", properties: {} } },
|
||||
{ name: "tool.b", description: "Tool B", inputSchema: { type: "object", properties: {} } },
|
||||
]
|
||||
|
||||
await MCP.add("my.special-server", {
|
||||
type: "local",
|
||||
command: ["echo", "test"],
|
||||
})
|
||||
|
||||
const tools = await MCP.tools()
|
||||
const keys = Object.keys(tools)
|
||||
|
||||
// Server name dots should be replaced with underscores
|
||||
expect(keys.some((k) => k.startsWith("my_special-server_"))).toBe(true)
|
||||
// Tool name dots should be replaced with underscores
|
||||
expect(keys.some((k) => k.endsWith("tool_b"))).toBe(true)
|
||||
expect(keys.length).toBe(2)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: transport leak — local stdio timeout (#19168)
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"local stdio transport is closed when connect times out (no process leak)",
|
||||
withInstance({}, async () => {
|
||||
lastCreatedClientName = "hanging-server"
|
||||
getOrCreateClientState("hanging-server")
|
||||
connectShouldHang = true
|
||||
|
||||
const addResult = await MCP.add("hanging-server", {
|
||||
type: "local",
|
||||
command: ["node", "fake.js"],
|
||||
timeout: 100,
|
||||
})
|
||||
|
||||
const serverStatus = (addResult.status as any)["hanging-server"] ?? addResult.status
|
||||
expect(serverStatus.status).toBe("failed")
|
||||
expect(serverStatus.error).toContain("timed out")
|
||||
// Transport must be closed to avoid orphaned child process
|
||||
expect(transportCloseCount).toBeGreaterThanOrEqual(1)
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: transport leak — remote timeout (#19168)
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"remote transport is closed when connect times out",
|
||||
withInstance({}, async () => {
|
||||
lastCreatedClientName = "hanging-remote"
|
||||
getOrCreateClientState("hanging-remote")
|
||||
connectShouldHang = true
|
||||
|
||||
const addResult = await MCP.add("hanging-remote", {
|
||||
type: "remote",
|
||||
url: "http://localhost:9999/mcp",
|
||||
timeout: 100,
|
||||
oauth: false,
|
||||
})
|
||||
|
||||
const serverStatus = (addResult.status as any)["hanging-remote"] ?? addResult.status
|
||||
expect(serverStatus.status).toBe("failed")
|
||||
// Transport must be closed to avoid leaked HTTP connections
|
||||
expect(transportCloseCount).toBeGreaterThanOrEqual(1)
|
||||
}),
|
||||
)
|
||||
|
||||
// ========================================================================
|
||||
// Test: transport leak — failed remote transports not closed (#19168)
|
||||
// ========================================================================
|
||||
|
||||
test(
|
||||
"failed remote transport is closed before trying next transport",
|
||||
withInstance({}, async () => {
|
||||
lastCreatedClientName = "fail-remote"
|
||||
getOrCreateClientState("fail-remote")
|
||||
connectShouldFail = true
|
||||
connectError = "Connection refused"
|
||||
|
||||
const addResult = await MCP.add("fail-remote", {
|
||||
type: "remote",
|
||||
url: "http://localhost:9999/mcp",
|
||||
timeout: 5000,
|
||||
oauth: false,
|
||||
})
|
||||
|
||||
const serverStatus = (addResult.status as any)["fail-remote"] ?? addResult.status
|
||||
expect(serverStatus.status).toBe("failed")
|
||||
// Both StreamableHTTP and SSE transports should be closed
|
||||
expect(transportCloseCount).toBeGreaterThanOrEqual(2)
|
||||
}),
|
||||
)
|
||||
@@ -74,11 +74,17 @@ delete process.env["SAMBANOVA_API_KEY"]
|
||||
delete process.env["OPENCODE_SERVER_PASSWORD"]
|
||||
delete process.env["OPENCODE_SERVER_USERNAME"]
|
||||
|
||||
// Use in-memory sqlite
|
||||
process.env["OPENCODE_DB"] = ":memory:"
|
||||
|
||||
// Now safe to import from src/
|
||||
const { Log } = await import("../src/util/log")
|
||||
const { initProjectors } = await import("../src/server/projectors")
|
||||
|
||||
Log.init({
|
||||
print: false,
|
||||
dev: true,
|
||||
level: "DEBUG",
|
||||
})
|
||||
|
||||
initProjectors()
|
||||
|
||||
@@ -2,41 +2,28 @@ import { $ } from "bun"
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
import fs from "fs/promises"
|
||||
import path from "path"
|
||||
import { Effect, Layer, ManagedRuntime } from "effect"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
import { watcherConfigLayer, withServices } from "../fixture/instance"
|
||||
import { FileWatcher } from "../../src/file/watcher"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { GlobalBus } from "../../src/bus/global"
|
||||
import { Vcs } from "../../src/project/vcs"
|
||||
|
||||
// Skip in CI — native @parcel/watcher binding needed
|
||||
const describeVcs = FileWatcher.hasNativeBinding() && !process.env.CI ? describe : describe.skip
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function withVcs(
|
||||
directory: string,
|
||||
body: (rt: ManagedRuntime.ManagedRuntime<FileWatcher.Service | Vcs.Service, never>) => Promise<void>,
|
||||
) {
|
||||
return withServices(
|
||||
async function withVcs(directory: string, body: () => Promise<void>) {
|
||||
return Instance.provide({
|
||||
directory,
|
||||
Layer.merge(FileWatcher.layer, Vcs.layer),
|
||||
async (rt) => {
|
||||
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
|
||||
await rt.runPromise(Vcs.Service.use((s) => s.init()))
|
||||
fn: async () => {
|
||||
FileWatcher.init()
|
||||
Vcs.init()
|
||||
await Bun.sleep(500)
|
||||
await body(rt)
|
||||
await body()
|
||||
},
|
||||
{ provide: [watcherConfigLayer] },
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
type BranchEvent = { directory?: string; payload: { type: string; properties: { branch?: string } } }
|
||||
|
||||
/** Wait for a Vcs.Event.BranchUpdated event on GlobalBus, with retry polling as fallback */
|
||||
function nextBranchUpdate(directory: string, timeout = 10_000) {
|
||||
return new Promise<string | undefined>((resolve, reject) => {
|
||||
let settled = false
|
||||
@@ -62,10 +49,6 @@ function nextBranchUpdate(directory: string, timeout = 10_000) {
|
||||
})
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describeVcs("Vcs", () => {
|
||||
afterEach(async () => {
|
||||
await Instance.disposeAll()
|
||||
@@ -74,8 +57,8 @@ describeVcs("Vcs", () => {
|
||||
test("branch() returns current branch name", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
await withVcs(tmp.path, async (rt) => {
|
||||
const branch = await rt.runPromise(Vcs.Service.use((s) => s.branch()))
|
||||
await withVcs(tmp.path, async () => {
|
||||
const branch = await Vcs.branch()
|
||||
expect(branch).toBeDefined()
|
||||
expect(typeof branch).toBe("string")
|
||||
})
|
||||
@@ -84,8 +67,8 @@ describeVcs("Vcs", () => {
|
||||
test("branch() returns undefined for non-git directories", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await withVcs(tmp.path, async (rt) => {
|
||||
const branch = await rt.runPromise(Vcs.Service.use((s) => s.branch()))
|
||||
await withVcs(tmp.path, async () => {
|
||||
const branch = await Vcs.branch()
|
||||
expect(branch).toBeUndefined()
|
||||
})
|
||||
})
|
||||
@@ -99,7 +82,11 @@ describeVcs("Vcs", () => {
|
||||
const pending = nextBranchUpdate(tmp.path)
|
||||
|
||||
const head = path.join(tmp.path, ".git", "HEAD")
|
||||
await fs.writeFile(head, `ref: refs/heads/${branch}\n`)
|
||||
await fs.writeFile(
|
||||
head,
|
||||
`ref: refs/heads/${branch}
|
||||
`,
|
||||
)
|
||||
|
||||
const updated = await pending
|
||||
expect(updated).toBe(branch)
|
||||
@@ -111,14 +98,18 @@ describeVcs("Vcs", () => {
|
||||
const branch = `test-${Math.random().toString(36).slice(2)}`
|
||||
await $`git branch ${branch}`.cwd(tmp.path).quiet()
|
||||
|
||||
await withVcs(tmp.path, async (rt) => {
|
||||
await withVcs(tmp.path, async () => {
|
||||
const pending = nextBranchUpdate(tmp.path)
|
||||
|
||||
const head = path.join(tmp.path, ".git", "HEAD")
|
||||
await fs.writeFile(head, `ref: refs/heads/${branch}\n`)
|
||||
await fs.writeFile(
|
||||
head,
|
||||
`ref: refs/heads/${branch}
|
||||
`,
|
||||
)
|
||||
|
||||
await pending
|
||||
const current = await rt.runPromise(Vcs.Service.use((s) => s.branch()))
|
||||
const current = await Vcs.branch()
|
||||
expect(current).toBe(branch)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
import { $ } from "bun"
|
||||
import { afterEach, describe, expect, test } from "bun:test"
|
||||
|
||||
const wintest = process.platform !== "win32" ? test : test.skip
|
||||
import fs from "fs/promises"
|
||||
import path from "path"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { Worktree } from "../../src/worktree"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
|
||||
function withInstance(directory: string, fn: () => Promise<any>) {
|
||||
return Instance.provide({ directory, fn })
|
||||
}
|
||||
|
||||
function normalize(input: string) {
|
||||
return input.replace(/\\/g, "/").toLowerCase()
|
||||
}
|
||||
|
||||
async function waitReady() {
|
||||
const { GlobalBus } = await import("../../src/bus/global")
|
||||
|
||||
return await new Promise<{ name: string; branch: string }>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
GlobalBus.off("event", on)
|
||||
reject(new Error("timed out waiting for worktree.ready"))
|
||||
}, 10_000)
|
||||
|
||||
function on(evt: { directory?: string; payload: { type: string; properties: { name: string; branch: string } } }) {
|
||||
if (evt.payload.type !== Worktree.Event.Ready.type) return
|
||||
clearTimeout(timer)
|
||||
GlobalBus.off("event", on)
|
||||
resolve(evt.payload.properties)
|
||||
}
|
||||
|
||||
GlobalBus.on("event", on)
|
||||
})
|
||||
}
|
||||
|
||||
describe("Worktree", () => {
|
||||
afterEach(() => Instance.disposeAll())
|
||||
|
||||
describe("makeWorktreeInfo", () => {
|
||||
test("returns info with name, branch, and directory", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.makeWorktreeInfo())
|
||||
|
||||
expect(info.name).toBeDefined()
|
||||
expect(typeof info.name).toBe("string")
|
||||
expect(info.branch).toBe(`opencode/${info.name}`)
|
||||
expect(info.directory).toContain(info.name)
|
||||
})
|
||||
|
||||
test("uses provided name as base", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.makeWorktreeInfo("my-feature"))
|
||||
|
||||
expect(info.name).toBe("my-feature")
|
||||
expect(info.branch).toBe("opencode/my-feature")
|
||||
})
|
||||
|
||||
test("slugifies the provided name", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.makeWorktreeInfo("My Feature Branch!"))
|
||||
|
||||
expect(info.name).toBe("my-feature-branch")
|
||||
})
|
||||
|
||||
test("throws NotGitError for non-git directories", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await expect(withInstance(tmp.path, () => Worktree.makeWorktreeInfo())).rejects.toThrow("WorktreeNotGitError")
|
||||
})
|
||||
})
|
||||
|
||||
describe("create + remove lifecycle", () => {
|
||||
test("create returns worktree info and remove cleans up", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.create())
|
||||
|
||||
expect(info.name).toBeDefined()
|
||||
expect(info.branch).toStartWith("opencode/")
|
||||
expect(info.directory).toBeDefined()
|
||||
|
||||
// Wait for bootstrap to complete
|
||||
await Bun.sleep(1000)
|
||||
|
||||
const ok = await withInstance(tmp.path, () => Worktree.remove({ directory: info.directory }))
|
||||
expect(ok).toBe(true)
|
||||
})
|
||||
|
||||
test("create returns after setup and fires Event.Ready after bootstrap", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
const ready = waitReady()
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.create())
|
||||
|
||||
// create returns before bootstrap completes, but the worktree already exists
|
||||
expect(info.name).toBeDefined()
|
||||
expect(info.branch).toStartWith("opencode/")
|
||||
|
||||
const text = await $`git worktree list --porcelain`.cwd(tmp.path).quiet().text()
|
||||
const dir = await fs.realpath(info.directory).catch(() => info.directory)
|
||||
expect(normalize(text)).toContain(normalize(dir))
|
||||
|
||||
// Event.Ready fires after bootstrap finishes in the background
|
||||
const props = await ready
|
||||
expect(props.name).toBe(info.name)
|
||||
expect(props.branch).toBe(info.branch)
|
||||
|
||||
// Cleanup
|
||||
await withInstance(info.directory, () => Instance.dispose())
|
||||
await Bun.sleep(100)
|
||||
await withInstance(tmp.path, () => Worktree.remove({ directory: info.directory }))
|
||||
})
|
||||
|
||||
test("create with custom name", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
const ready = waitReady()
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.create({ name: "test-workspace" }))
|
||||
|
||||
expect(info.name).toBe("test-workspace")
|
||||
expect(info.branch).toBe("opencode/test-workspace")
|
||||
|
||||
// Cleanup
|
||||
await ready
|
||||
await withInstance(info.directory, () => Instance.dispose())
|
||||
await Bun.sleep(100)
|
||||
await withInstance(tmp.path, () => Worktree.remove({ directory: info.directory }))
|
||||
})
|
||||
})
|
||||
|
||||
describe("createFromInfo", () => {
|
||||
wintest("creates and bootstraps git worktree", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
const info = await withInstance(tmp.path, () => Worktree.makeWorktreeInfo("from-info-test"))
|
||||
await withInstance(tmp.path, () => Worktree.createFromInfo(info))
|
||||
|
||||
// Worktree should exist in git (normalize slashes for Windows)
|
||||
const list = await $`git worktree list --porcelain`.cwd(tmp.path).quiet().text()
|
||||
const normalizedList = list.replace(/\\/g, "/")
|
||||
const normalizedDir = info.directory.replace(/\\/g, "/")
|
||||
expect(normalizedList).toContain(normalizedDir)
|
||||
|
||||
// Cleanup
|
||||
await withInstance(tmp.path, () => Worktree.remove({ directory: info.directory }))
|
||||
})
|
||||
})
|
||||
|
||||
describe("remove edge cases", () => {
|
||||
test("remove non-existent directory succeeds silently", async () => {
|
||||
await using tmp = await tmpdir({ git: true })
|
||||
|
||||
const ok = await withInstance(tmp.path, () =>
|
||||
Worktree.remove({ directory: path.join(tmp.path, "does-not-exist") }),
|
||||
)
|
||||
expect(ok).toBe(true)
|
||||
})
|
||||
|
||||
test("throws NotGitError for non-git directories", async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await expect(withInstance(tmp.path, () => Worktree.remove({ directory: "/tmp/fake" }))).rejects.toThrow(
|
||||
"WorktreeNotGitError",
|
||||
)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -927,4 +927,31 @@ describe("session.message-v2.fromError", () => {
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
test("classifies ZlibError from fetch as retryable APIError", () => {
|
||||
const zlibError = new Error(
|
||||
'ZlibError fetching "https://opencode.cloudflare.dev/anthropic/messages". For more information, pass `verbose: true` in the second argument to fetch()',
|
||||
)
|
||||
;(zlibError as any).code = "ZlibError"
|
||||
;(zlibError as any).errno = 0
|
||||
;(zlibError as any).path = ""
|
||||
|
||||
const result = MessageV2.fromError(zlibError, { providerID })
|
||||
|
||||
expect(MessageV2.APIError.isInstance(result)).toBe(true)
|
||||
expect((result as MessageV2.APIError).data.isRetryable).toBe(true)
|
||||
expect((result as MessageV2.APIError).data.message).toInclude("decompression")
|
||||
})
|
||||
|
||||
test("classifies ZlibError as AbortedError when abort context is provided", () => {
|
||||
const zlibError = new Error(
|
||||
'ZlibError fetching "https://opencode.cloudflare.dev/anthropic/messages". For more information, pass `verbose: true` in the second argument to fetch()',
|
||||
)
|
||||
;(zlibError as any).code = "ZlibError"
|
||||
;(zlibError as any).errno = 0
|
||||
|
||||
const result = MessageV2.fromError(zlibError, { providerID, aborted: true })
|
||||
|
||||
expect(result.name).toBe("MessageAbortedError")
|
||||
})
|
||||
})
|
||||
|
||||
@@ -125,6 +125,18 @@ describe("session.retry.retryable", () => {
|
||||
|
||||
expect(SessionRetry.retryable(error)).toBeUndefined()
|
||||
})
|
||||
|
||||
test("retries ZlibError decompression failures", () => {
|
||||
const error = new MessageV2.APIError({
|
||||
message: "Response decompression failed",
|
||||
isRetryable: true,
|
||||
metadata: { code: "ZlibError" },
|
||||
}).toObject() as MessageV2.APIError
|
||||
|
||||
const retryable = SessionRetry.retryable(error)
|
||||
expect(retryable).toBeDefined()
|
||||
expect(retryable).toBe("Response decompression failed")
|
||||
})
|
||||
})
|
||||
|
||||
describe("session.message-v2.fromError", () => {
|
||||
|
||||
@@ -10,8 +10,8 @@ import { MessageID, PartID } from "../../src/session/schema"
|
||||
const projectRoot = path.join(__dirname, "../..")
|
||||
Log.init({ print: false })
|
||||
|
||||
describe("session.started event", () => {
|
||||
test("should emit session.started event when session is created", async () => {
|
||||
describe("session.created event", () => {
|
||||
test("should emit session.created event when session is created", async () => {
|
||||
await Instance.provide({
|
||||
directory: projectRoot,
|
||||
fn: async () => {
|
||||
@@ -41,14 +41,14 @@ describe("session.started event", () => {
|
||||
})
|
||||
})
|
||||
|
||||
test("session.started event should be emitted before session.updated", async () => {
|
||||
test("session.created event should be emitted before session.updated", async () => {
|
||||
await Instance.provide({
|
||||
directory: projectRoot,
|
||||
fn: async () => {
|
||||
const events: string[] = []
|
||||
|
||||
const unsubStarted = Bus.subscribe(Session.Event.Created, () => {
|
||||
events.push("started")
|
||||
const unsubCreated = Bus.subscribe(Session.Event.Created, () => {
|
||||
events.push("created")
|
||||
})
|
||||
|
||||
const unsubUpdated = Bus.subscribe(Session.Event.Updated, () => {
|
||||
@@ -59,12 +59,12 @@ describe("session.started event", () => {
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 100))
|
||||
|
||||
unsubStarted()
|
||||
unsubCreated()
|
||||
unsubUpdated()
|
||||
|
||||
expect(events).toContain("started")
|
||||
expect(events).toContain("created")
|
||||
expect(events).toContain("updated")
|
||||
expect(events.indexOf("started")).toBeLessThan(events.indexOf("updated"))
|
||||
expect(events.indexOf("created")).toBeLessThan(events.indexOf("updated"))
|
||||
|
||||
await Session.remove(session.id)
|
||||
},
|
||||
|
||||
@@ -181,7 +181,7 @@ test("symlink handling", async () => {
|
||||
})
|
||||
})
|
||||
|
||||
test("large file handling", async () => {
|
||||
test("file under size limit handling", async () => {
|
||||
await using tmp = await bootstrap()
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
@@ -196,6 +196,23 @@ test("large file handling", async () => {
|
||||
})
|
||||
})
|
||||
|
||||
test("large added files are skipped", async () => {
|
||||
await using tmp = await bootstrap()
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: async () => {
|
||||
const before = await Snapshot.track()
|
||||
expect(before).toBeTruthy()
|
||||
|
||||
await Filesystem.write(`${tmp.path}/huge.txt`, new Uint8Array(2 * 1024 * 1024 + 1))
|
||||
|
||||
expect((await Snapshot.patch(before!)).files).toEqual([])
|
||||
expect(await Snapshot.diff(before!)).toBe("")
|
||||
expect(await Snapshot.track()).toBe(before)
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
test("nested directory revert", async () => {
|
||||
await using tmp = await bootstrap()
|
||||
await Instance.provide({
|
||||
|
||||
@@ -6,14 +6,9 @@ import { Database } from "../../src/storage/db"
|
||||
|
||||
describe("Database.Path", () => {
|
||||
test("returns database path for the current channel", () => {
|
||||
const db = process.env["OPENCODE_DB"]
|
||||
const expected = db
|
||||
? path.isAbsolute(db)
|
||||
? db
|
||||
: path.join(Global.Path.data, db)
|
||||
: ["latest", "beta"].includes(Installation.CHANNEL)
|
||||
? path.join(Global.Path.data, "opencode.db")
|
||||
: path.join(Global.Path.data, `opencode-${Installation.CHANNEL.replace(/[^a-zA-Z0-9._-]/g, "-")}.db`)
|
||||
expect(Database.Path).toBe(expected)
|
||||
const expected = ["latest", "beta"].includes(Installation.CHANNEL)
|
||||
? path.join(Global.Path.data, "opencode.db")
|
||||
: path.join(Global.Path.data, `opencode-${Installation.CHANNEL.replace(/[^a-zA-Z0-9._-]/g, "-")}.db`)
|
||||
expect(Database.getChannelPath()).toBe(expected)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
import { describe, test, expect, beforeEach, afterEach, afterAll } from "bun:test"
|
||||
import { tmpdir } from "../fixture/fixture"
|
||||
import z from "zod"
|
||||
import { Bus } from "../../src/bus"
|
||||
import { Instance } from "../../src/project/instance"
|
||||
import { SyncEvent } from "../../src/sync"
|
||||
import { Database } from "../../src/storage/db"
|
||||
import { EventTable } from "../../src/sync/event.sql"
|
||||
import { Identifier } from "../../src/id/id"
|
||||
import { Flag } from "../../src/flag/flag"
|
||||
import { initProjectors } from "../../src/server/projectors"
|
||||
|
||||
const original = Flag.OPENCODE_EXPERIMENTAL_WORKSPACES
|
||||
|
||||
beforeEach(() => {
|
||||
Database.close()
|
||||
|
||||
// @ts-expect-error don't do this normally, but it works
|
||||
Flag.OPENCODE_EXPERIMENTAL_WORKSPACES = true
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
// @ts-expect-error don't do this normally, but it works
|
||||
Flag.OPENCODE_EXPERIMENTAL_WORKSPACES = original
|
||||
})
|
||||
|
||||
function withInstance(fn: () => void | Promise<void>) {
|
||||
return async () => {
|
||||
await using tmp = await tmpdir()
|
||||
|
||||
await Instance.provide({
|
||||
directory: tmp.path,
|
||||
fn: async () => {
|
||||
await fn()
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
describe("SyncEvent", () => {
|
||||
function setup() {
|
||||
SyncEvent.reset()
|
||||
|
||||
const Created = SyncEvent.define({
|
||||
type: "item.created",
|
||||
version: 1,
|
||||
aggregate: "id",
|
||||
schema: z.object({ id: z.string(), name: z.string() }),
|
||||
})
|
||||
const Sent = SyncEvent.define({
|
||||
type: "item.sent",
|
||||
version: 1,
|
||||
aggregate: "item_id",
|
||||
schema: z.object({ item_id: z.string(), to: z.string() }),
|
||||
})
|
||||
|
||||
SyncEvent.init({
|
||||
projectors: [SyncEvent.project(Created, () => {}), SyncEvent.project(Sent, () => {})],
|
||||
})
|
||||
|
||||
return { Created, Sent }
|
||||
}
|
||||
|
||||
afterAll(() => {
|
||||
SyncEvent.reset()
|
||||
initProjectors()
|
||||
})
|
||||
|
||||
describe("run", () => {
|
||||
test(
|
||||
"inserts event row",
|
||||
withInstance(() => {
|
||||
const { Created } = setup()
|
||||
SyncEvent.run(Created, { id: "evt_1", name: "first" })
|
||||
const rows = Database.use((db) => db.select().from(EventTable).all())
|
||||
expect(rows).toHaveLength(1)
|
||||
expect(rows[0].type).toBe("item.created.1")
|
||||
expect(rows[0].aggregate_id).toBe("evt_1")
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"increments seq per aggregate",
|
||||
withInstance(() => {
|
||||
const { Created } = setup()
|
||||
SyncEvent.run(Created, { id: "evt_1", name: "first" })
|
||||
SyncEvent.run(Created, { id: "evt_1", name: "second" })
|
||||
const rows = Database.use((db) => db.select().from(EventTable).all())
|
||||
expect(rows).toHaveLength(2)
|
||||
expect(rows[1].seq).toBe(rows[0].seq + 1)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"uses custom aggregate field from agg()",
|
||||
withInstance(() => {
|
||||
const { Sent } = setup()
|
||||
SyncEvent.run(Sent, { item_id: "evt_1", to: "james" })
|
||||
const rows = Database.use((db) => db.select().from(EventTable).all())
|
||||
expect(rows).toHaveLength(1)
|
||||
expect(rows[0].aggregate_id).toBe("evt_1")
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"emits events",
|
||||
withInstance(async () => {
|
||||
const { Created } = setup()
|
||||
const events: Array<{
|
||||
type: string
|
||||
properties: { id: string; name: string }
|
||||
}> = []
|
||||
const received = new Promise<void>((resolve) => {
|
||||
Bus.subscribeAll((event) => {
|
||||
events.push(event)
|
||||
resolve()
|
||||
})
|
||||
})
|
||||
|
||||
SyncEvent.run(Created, { id: "evt_1", name: "test" })
|
||||
|
||||
await received
|
||||
expect(events).toHaveLength(1)
|
||||
expect(events[0]).toEqual({
|
||||
type: "item.created",
|
||||
properties: {
|
||||
id: "evt_1",
|
||||
name: "test",
|
||||
},
|
||||
})
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("replay", () => {
|
||||
test(
|
||||
"inserts event from external payload",
|
||||
withInstance(() => {
|
||||
const id = Identifier.descending("message")
|
||||
SyncEvent.replay({
|
||||
id: "evt_1",
|
||||
type: "item.created.1",
|
||||
seq: 0,
|
||||
aggregateID: id,
|
||||
data: { id, name: "replayed" },
|
||||
})
|
||||
const rows = Database.use((db) => db.select().from(EventTable).all())
|
||||
expect(rows).toHaveLength(1)
|
||||
expect(rows[0].aggregate_id).toBe(id)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"throws on sequence mismatch",
|
||||
withInstance(() => {
|
||||
const id = Identifier.descending("message")
|
||||
SyncEvent.replay({
|
||||
id: "evt_1",
|
||||
type: "item.created.1",
|
||||
seq: 0,
|
||||
aggregateID: id,
|
||||
data: { id, name: "first" },
|
||||
})
|
||||
expect(() =>
|
||||
SyncEvent.replay({
|
||||
id: "evt_1",
|
||||
type: "item.created.1",
|
||||
seq: 5,
|
||||
aggregateID: id,
|
||||
data: { id, name: "bad" },
|
||||
}),
|
||||
).toThrow(/Sequence mismatch/)
|
||||
}),
|
||||
)
|
||||
|
||||
test(
|
||||
"throws on unknown event type",
|
||||
withInstance(() => {
|
||||
expect(() =>
|
||||
SyncEvent.replay({
|
||||
id: "evt_1",
|
||||
type: "unknown.event.1",
|
||||
seq: 0,
|
||||
aggregateID: "x",
|
||||
data: {},
|
||||
}),
|
||||
).toThrow(/Unknown event type/)
|
||||
}),
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -89,7 +89,6 @@ describe("tool.edit", () => {
|
||||
const { FileWatcher } = await import("../../src/file/watcher")
|
||||
|
||||
const events: string[] = []
|
||||
const unsubEdited = Bus.subscribe(File.Event.Edited, () => events.push("edited"))
|
||||
const unsubUpdated = Bus.subscribe(FileWatcher.Event.Updated, () => events.push("updated"))
|
||||
|
||||
const edit = await EditTool.init()
|
||||
@@ -102,9 +101,7 @@ describe("tool.edit", () => {
|
||||
ctx,
|
||||
)
|
||||
|
||||
expect(events).toContain("edited")
|
||||
expect(events).toContain("updated")
|
||||
unsubEdited()
|
||||
unsubUpdated()
|
||||
},
|
||||
})
|
||||
@@ -305,11 +302,9 @@ describe("tool.edit", () => {
|
||||
await FileTime.read(ctx.sessionID, filepath)
|
||||
|
||||
const { Bus } = await import("../../src/bus")
|
||||
const { File } = await import("../../src/file")
|
||||
const { FileWatcher } = await import("../../src/file/watcher")
|
||||
|
||||
const events: string[] = []
|
||||
const unsubEdited = Bus.subscribe(File.Event.Edited, () => events.push("edited"))
|
||||
const unsubUpdated = Bus.subscribe(FileWatcher.Event.Updated, () => events.push("updated"))
|
||||
|
||||
const edit = await EditTool.init()
|
||||
@@ -322,9 +317,7 @@ describe("tool.edit", () => {
|
||||
ctx,
|
||||
)
|
||||
|
||||
expect(events).toContain("edited")
|
||||
expect(events).toContain("updated")
|
||||
unsubEdited()
|
||||
unsubUpdated()
|
||||
},
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user