Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384import { beforeEach, expect } from "bun:test"import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"import { Database } from "@opencode/core/database/database"import { makeMemoryDriver } from "@opencode/core/environment/index"import { Workspace } from "@opencode/core/workspace"import { WorkspaceDriver } from "@opencode/core/workspace/driver"import { WorkspaceTable } from "@opencode/core/workspace/sql"import { LayerNode } from "@opencode/util/effect/layer-node"import { eq } from "drizzle-orm"import { Deferred, Effect, Fiber } from "effect"import { TestClock } from "effect/testing"import { ChildProcess } from "effect/unstable/process"import { testEffect } from "./lib/effect"
const calls: Array<{ readonly operation: string; readonly binding?: WorkspaceDriver.Binding | null }> = []const memory = makeMemoryDriver()let failConnect = falselet create: WorkspaceDriver.Interface["create"] = ({ workspaceID }) => Effect.succeed({ binding: { workspaceID, generation: 0 } })
const driver = WorkspaceDriver.make({ create: (input) => { calls.push({ operation: "create" }) return create(input) }, connect: ({ binding }) => { calls.push({ operation: "connect", binding }) if (failConnect) return Effect.fail(new WorkspaceDriver.Error({ message: "wake failed" })) return Effect.succeed(memory) }, suspendForIdle: ({ binding, saveBinding }) => { calls.push({ operation: "suspendForIdle", binding }) return saveBinding({ ...binding, generation: Number(binding.generation) + 1, suspended: true }) }, destroy: ({ binding }) => { calls.push({ operation: "destroy", binding }) return Effect.void },})
const it = testEffect( AppNodeBuilder.build( LayerNode.group([Database.node, Workspace.configured({ idleThreshold: "5 minutes", pollInterval: "1 minute" })]), [WorkspaceDriver.node.replace(WorkspaceDriver.registryNode({ fake: driver, other: driver }))], ),)
beforeEach(() => { calls.splice(0) failConnect = false create = ({ workspaceID }) => Effect.succeed({ binding: { workspaceID, generation: 0 } })})
const gateCreate = Effect.fnUntraced(function* () { const started = yield* Deferred.make<void>() const release = yield* Deferred.make<void>() create = ({ workspaceID }) => Deferred.succeed(started, undefined).pipe( Effect.andThen(Deferred.await(release)), Effect.as({ binding: { workspaceID, generation: 0 } }), ) return { started, release }})
it.effect("rejects unregistered workspace providers", () => Effect.gen(function* () { const registry = WorkspaceDriver.registry({ fake: driver })
for (const provider of ["missing", "constructor", "toString", "__proto__"]) { expect(yield* registry.get(provider).pipe(Effect.flip)).toEqual( new WorkspaceDriver.ProviderNotFound({ provider }), ) } }),)
it.effect("creates and persists an ID without provisioning", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" })
expect(workspaceID.startsWith("wrk_")).toBe(true) expect(calls).toEqual([]) expect( yield* Database.Service.use(({ db }) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, workspaceID)).get(), ).pipe(Effect.orDie), ).toMatchObject({ id: workspaceID, provider: "fake", binding: null }) }),)
it.effect("creates a workspace with a caller-supplied ID", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const id = Workspace.ID.create()
expect(yield* workspace.create({ id, provider: "fake" })).toBe(id) expect( yield* Database.Service.use(({ db }) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get(), ).pipe(Effect.orDie), ).toMatchObject({ id, provider: "fake", binding: null }) }),)
it.effect("reuses a caller-supplied ID with the same provider", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const id = Workspace.ID.create()
expect(yield* workspace.create({ id, provider: "fake" })).toBe(id) expect(yield* workspace.create({ id, provider: "fake" })).toBe(id) expect( yield* Database.Service.use(({ db }) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).all(), ).pipe(Effect.orDie), ).toHaveLength(1) expect(calls).toEqual([]) }),)
it.effect("rejects a caller-supplied ID already assigned to another provider", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const id = Workspace.ID.create() yield* workspace.create({ id, provider: "fake" })
expect(yield* workspace.create({ id, provider: "other" }).pipe(Effect.flip)).toEqual( new Workspace.CreateConflict({ workspaceID: id, provider: "other", existingProvider: "fake" }), ) }),)
it.effect("resolves an existing caller-supplied ID before provider lookup", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const id = Workspace.ID.create() yield* Database.Service.use(({ db }) => db .insert(WorkspaceTable) .values({ id, provider: "missing", binding: null, created_at: 0, last_used_at: 0 }) .run(), ).pipe(Effect.orDie)
expect(yield* workspace.create({ id, provider: "missing" })).toBe(id) expect(yield* workspace.create({ id, provider: "another-missing" }).pipe(Effect.flip)).toEqual( new Workspace.CreateConflict({ workspaceID: id, provider: "another-missing", existingProvider: "missing", }), ) }),)
it.effect("destroys an unprovisioned workspace through the driver with a null binding", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" })
expect(yield* workspace.destroy(workspaceID)).toEqual({ destroyed: true }) expect(calls).toEqual([{ operation: "destroy", binding: null }]) expect( yield* Database.Service.use(({ db }) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, workspaceID)).get(), ).pipe(Effect.orDie), ).toBeUndefined() }),)
it.effect("succeeds without calling the driver when the workspace does not exist", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = Workspace.ID.create()
expect(yield* workspace.destroy(workspaceID)).toEqual({ destroyed: false }) expect(calls).toEqual([]) }),)
it.effect("reports whether destroy removed an existing workspace", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" })
expect(yield* workspace.destroy(workspaceID)).toEqual({ destroyed: true }) expect(yield* workspace.destroy(workspaceID)).toEqual({ destroyed: false }) expect(calls).toEqual([{ operation: "destroy", binding: null }]) }),)
it.effect("starts eager provisioning in the background and lets callers join it", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const gate = yield* gateCreate()
const eager = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Deferred.await(gate.started) const waiter = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Effect.yieldNow expect(calls.map((call) => call.operation)).toEqual(["create"])
yield* Deferred.succeed(gate.release, undefined) const [eagerResult, waiterResult] = yield* Effect.all([Fiber.join(eager), Fiber.join(waiter)]) expect(eagerResult).toEqual(waiterResult) expect(eagerResult.binding).toEqual({ workspaceID, generation: 0 }) }),)
it.effect("starts lazy provisioning on the first spawn", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const environment = yield* workspace.connect(workspaceID) const gate = yield* gateCreate()
expect(calls).toEqual([]) const spawned = yield* Effect.scoped(environment.spawner.spawn(ChildProcess.make("lazy"))).pipe( Effect.forkScoped({ startImmediately: true }), ) yield* Deferred.await(gate.started) expect(calls.map((call) => call.operation)).toEqual(["create"]) yield* Deferred.succeed(gate.release, undefined) yield* Fiber.await(spawned) expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) }),)
it.effect("shares provisioning between concurrent first spawns", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const environment = yield* workspace.connect(workspaceID) const gate = yield* gateCreate()
const spawned = yield* Effect.all( ["first", "second"].map((command) => Effect.scoped(environment.spawner.spawn(ChildProcess.make(command))).pipe( Effect.forkScoped({ startImmediately: true }), ), ), ) yield* Deferred.await(gate.started) yield* Effect.yieldNow expect(calls.map((call) => call.operation)).toEqual(["create"])
yield* Deferred.succeed(gate.release, undefined) yield* Effect.forEach(spawned, Fiber.await) expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) }),)
it.effect("keeps shared provisioning alive when a waiter is interrupted", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const gate = yield* gateCreate()
const owner = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Deferred.await(gate.started) const waiter = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Fiber.interrupt(waiter) expect(calls.map((call) => call.operation)).toEqual(["create"])
yield* Deferred.succeed(gate.release, undefined) expect((yield* Fiber.join(owner)).binding).toEqual({ workspaceID, generation: 0 }) expect(calls.map((call) => call.operation)).toEqual(["create"]) }),)
it.effect("interrupts in-flight provisioning on destroy and fails waiters with NotFound", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const gate = yield* gateCreate()
const waiter = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Deferred.await(gate.started) yield* workspace.destroy(workspaceID)
expect(yield* Fiber.join(waiter).pipe(Effect.flip)).toEqual(new Workspace.NotFound({ workspaceID })) expect(calls.map((call) => call.operation)).toEqual(["create", "destroy"]) expect(calls.at(-1)?.binding).toBeNull() expect( yield* Database.Service.use(({ db }) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, workspaceID)).get(), ).pipe(Effect.orDie), ).toBeUndefined() }),)
it.effect("shares a failed attempt and retries the same workspace ID", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const started = yield* Deferred.make<void>() const release = yield* Deferred.make<void>() let fail = true create = ({ workspaceID }) => Deferred.succeed(started, undefined).pipe( Effect.andThen(Deferred.await(release)), Effect.andThen( Effect.suspend(() => fail ? Effect.fail(new WorkspaceDriver.Error({ message: "create failed" })) : Effect.succeed({ binding: { workspaceID, generation: 0 } }), ), ), )
const first = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Deferred.await(started) const second = yield* workspace.provision(workspaceID).pipe(Effect.forkScoped({ startImmediately: true })) yield* Deferred.succeed(release, undefined) const [firstExit, secondExit] = yield* Effect.all([Fiber.await(first), Fiber.await(second)]) expect(firstExit._tag).toBe("Failure") expect(secondExit._tag).toBe("Failure") expect(calls.map((call) => call.operation)).toEqual(["create"])
fail = false expect((yield* workspace.provision(workspaceID)).binding).toEqual({ workspaceID, generation: 0 }) expect(calls.map((call) => call.operation)).toEqual(["create", "create"]) }),)
it.effect("persists the workspace lifecycle and reconnects after idle suspension", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const workspaceID = yield* workspace.create({ provider: "fake" }) const created = yield* workspace.provision(workspaceID)
expect(created.id).toBe(workspaceID) expect(created.binding).toEqual({ workspaceID: created.id, generation: 0 })
const environment = yield* workspace.connect(created.id) expect(calls.map((call) => call.operation)).toEqual(["create"])
yield* TestClock.adjust("4 minutes") yield* Effect.scoped(environment.spawner.spawn(ChildProcess.make("activity"))).pipe(Effect.exit) yield* TestClock.adjust("4 minutes") expect(calls.map((call) => call.operation)).toEqual(["create", "connect"])
yield* TestClock.adjust("2 minutes") expect(calls.map((call) => call.operation)).toEqual(["create", "connect", "suspendForIdle"])
const stored = yield* Database.Service.use(({ db }) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, created.id)).get(), ).pipe(Effect.orDie) expect(stored?.binding).toEqual({ workspaceID: created.id, generation: 1, suspended: true }) expect(stored?.last_used_at).toBe(4 * 60 * 1000)
yield* Effect.scoped(environment.spawner.spawn(ChildProcess.make("wake"))).pipe(Effect.exit) expect(calls.map((call) => call.operation)).toEqual(["create", "connect", "suspendForIdle", "connect"]) expect(calls.at(-1)?.binding).toEqual({ workspaceID: created.id, generation: 1, suspended: true })
yield* workspace.destroy(created.id) expect(calls.at(-1)?.operation).toBe("destroy") }),)
it.effect("surfaces wake failures through the spawn error channel", () => Effect.gen(function* () { const workspace = yield* Workspace.Service const created = yield* workspace.provision(yield* workspace.create({ provider: "fake" })) const environment = yield* workspace.connect(created.id) yield* Effect.scoped(environment.spawner.spawn(ChildProcess.make("connect"))).pipe(Effect.exit)
yield* TestClock.adjust("6 minutes") failConnect = true
const error = yield* Effect.scoped(environment.spawner.spawn(ChildProcess.make("wake"))).pipe(Effect.flip) expect(error).toMatchObject({ _tag: "PlatformError", reason: { _tag: "Unknown", module: "Workspace", method: "spawn", description: `Failed to wake workspace ${created.id}`, }, }) }),)