diff --git a/packages/core/src/workspace.ts b/packages/core/src/workspace.ts index a71b84bfba7..daad50ceab3 100644 --- a/packages/core/src/workspace.ts +++ b/packages/core/src/workspace.ts @@ -7,6 +7,8 @@ import { Clock, Context, Deferred, Duration, Effect, Exit, FiberSet, Layer, Ref, import { systemError } from "effect/PlatformError" import { make } from "effect/unstable/process/ChildProcessSpawner" import type { EnvironmentDriver } from "./environment/driver.js" +import { execDefaults } from "./environment/exec-defaults.js" +import { Failed, type Files, type FilesImpl } from "./environment/files.js" import { Database } from "./database/database.js" import { KeyedMutex } from "./effect/keyed-mutex.js" import { WorkspaceDriver } from "./workspace/driver.js" @@ -60,6 +62,7 @@ export class Service extends Context.Service()("@opencode/Wo interface Connection { readonly driver: WorkspaceDriver.Interface readonly environment: EnvironmentDriver.Driver + readonly files: Files readonly saveBinding: (binding: WorkspaceDriver.Binding) => Effect.Effect readonly lastActivity: Ref.Ref readonly active: Ref.Ref @@ -166,6 +169,10 @@ const layer = (options: Options) => const connection: Connection = { driver, environment, + files: { + ...execDefaults(environment.spawner), + ...environment.overrides, + }, saveBinding: persistBinding, lastActivity: yield* Ref.make(now), active: yield* Ref.make(0), @@ -252,41 +259,62 @@ const layer = (options: Options) => }), provision, connect: Effect.fn("Workspace.connect")(function* (workspaceID) { - const spawner = make((command) => - Effect.acquireRelease( - // A live connection implies the binding is already persisted, so skip the provision hop. - Effect.suspend(() => (connections.has(workspaceID) ? Effect.void : provision(workspaceID))).pipe( - Effect.andThen( - locks.withLock(workspaceID)( - Effect.gen(function* () { - const connection = yield* open(workspaceID) - yield* Ref.set(connection.lastActivity, yield* Clock.currentTimeMillis) - yield* Ref.update(connection.active, (active) => active + 1) - return connection - }), - ), - ), - Effect.mapError((cause) => - systemError({ - _tag: "Unknown", - module: "Workspace", - method: "spawn", - description: `Failed to wake workspace ${workspaceID}`, - cause, + const acquire = Effect.acquireRelease( + // A live connection implies the binding is already persisted, so skip the provision hop. + Effect.suspend(() => (connections.has(workspaceID) ? Effect.void : provision(workspaceID))).pipe( + Effect.andThen( + locks.withLock(workspaceID)( + Effect.gen(function* () { + const connection = yield* open(workspaceID) + yield* Ref.set(connection.lastActivity, yield* Clock.currentTimeMillis) + yield* Ref.update(connection.active, (active) => active + 1) + return connection }), ), ), - (connection) => - locks.withLock(workspaceID)( - Effect.gen(function* () { - yield* Ref.update(connection.active, (active) => active - 1) - yield* Ref.set(connection.lastActivity, yield* Clock.currentTimeMillis) - }), - ), - ).pipe(Effect.flatMap((connection) => connection.environment.spawner.spawn(command))), + ), + (connection) => + locks.withLock(workspaceID)( + Effect.gen(function* () { + yield* Ref.update(connection.active, (active) => active - 1) + yield* Ref.set(connection.lastActivity, yield* Clock.currentTimeMillis) + }), + ), ) - // Overrides are connection-bound; per-spawn routing is required before any driver ships them, so they are deliberately omitted. - return { spawner } + const spawner = make((command) => + acquire.pipe( + Effect.mapError((cause) => + systemError({ + _tag: "Unknown", + module: "Workspace", + method: "spawn", + description: `Failed to wake workspace ${workspaceID}`, + cause, + }), + ), + Effect.flatMap((connection) => connection.environment.spawner.spawn(command)), + ), + ) + const withFiles = ( + path: string, + use: (files: Files) => Effect.Effect, + ): Effect.Effect => + Effect.scoped( + acquire.pipe( + Effect.mapError((cause) => new Failed({ path, cause })), + Effect.flatMap((connection) => use(connection.files)), + ), + ) + const overrides: FilesImpl = { + read: (path, range) => withFiles(path, (files) => files.read(path, range)), + write: (path, bytes) => withFiles(path, (files) => files.write(path, bytes)), + stat: (path) => withFiles(path, (files) => files.stat(path)), + list: (path) => withFiles(path, (files) => files.list(path)), + remove: (path) => withFiles(path, (files) => files.remove(path)), + move: (from, to) => withFiles(from, (files) => files.move(from, to)), + mkdir: (path) => withFiles(path, (files) => files.mkdir(path)), + } + return { spawner, overrides } }), destroy: Effect.fn("Workspace.destroy")(function* (workspaceID) { // Settling the shared attempt cancels its racing provision body and fails diff --git a/packages/core/test/workspace.test.ts b/packages/core/test/workspace.test.ts index 8666b05e226..76d58275284 100644 --- a/packages/core/test/workspace.test.ts +++ b/packages/core/test/workspace.test.ts @@ -1,15 +1,16 @@ import { 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 { Failed, makeFiles, makeMemoryDriver, NotFound } 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 { Deferred, Effect, Fiber, Sink, Stream } from "effect" import { TestClock } from "effect/testing" import { ChildProcess } from "effect/unstable/process" +import { ExitCode, make as makeSpawner, makeHandle, ProcessId } from "effect/unstable/process/ChildProcessSpawner" import { testEffect } from "./lib/effect" const calls: Array<{ readonly operation: string; readonly binding?: WorkspaceDriver.Binding | null }> = [] @@ -17,16 +18,17 @@ const memory = makeMemoryDriver() let failConnect = false let create: WorkspaceDriver.Interface["create"] = ({ workspaceID }) => Effect.succeed({ binding: { workspaceID, generation: 0 } }) +let connectDriver: WorkspaceDriver.Interface["connect"] = () => Effect.succeed(memory) const driver = WorkspaceDriver.make({ create: (input) => { calls.push({ operation: "create" }) return create(input) }, - connect: ({ binding }) => { - calls.push({ operation: "connect", binding }) + connect: (input) => { + calls.push({ operation: "connect", binding: input.binding }) if (failConnect) return Effect.fail(new WorkspaceDriver.Error({ message: "wake failed" })) - return Effect.succeed(memory) + return connectDriver(input) }, suspendForIdle: ({ binding, saveBinding }) => { calls.push({ operation: "suspendForIdle", binding }) @@ -49,6 +51,7 @@ beforeEach(() => { calls.splice(0) failConnect = false create = ({ workspaceID }) => Effect.succeed({ binding: { workspaceID, generation: 0 } }) + connectDriver = () => Effect.succeed(memory) }) const gateCreate = Effect.fnUntraced(function* () { @@ -381,3 +384,160 @@ it.effect("surfaces wake failures through the spawn error channel", () => }) }), ) + +it.effect("routes file overrides through the workspace connection without spawning processes", () => + Effect.gen(function* () { + const workspace = yield* Workspace.Service + const workspaceID = yield* workspace.create({ provider: "fake" }) + const files = makeFiles(yield* workspace.connect(workspaceID)) + + expect(calls).toEqual([]) + const root = `/override-${workspaceID}` + const payload = new TextEncoder().encode("hello workspace") + + yield* files.mkdir(root) + expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) + + yield* files.write(`${root}/note.txt`, payload) + expect(yield* files.stat(`${root}/note.txt`)).toMatchObject({ + type: "file", + size: payload.byteLength, + }) + expect((yield* files.read(`${root}/note.txt`, { offset: 0, length: 5 })).bytes).toEqual( + new TextEncoder().encode("hello"), + ) + expect(yield* files.list(root)).toEqual([{ name: "note.txt", type: "file" }]) + + yield* files.move(`${root}/note.txt`, `${root}/renamed.txt`) + expect(yield* files.list(root)).toEqual([{ name: "renamed.txt", type: "file" }]) + + yield* files.remove(root) + expect(yield* files.stat(root).pipe(Effect.flip)).toEqual(new NotFound({ path: root })) + expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) + }), +) + +it.effect("falls back to execDefaults over the connection spawner for unprovided overrides", () => + Effect.gen(function* () { + const spawned: string[] = [] + const written: Array<{ readonly path: string; readonly size: number }> = [] + const spawner = makeSpawner((command) => + Effect.sync(() => { + if (command._tag === "StandardCommand") { + spawned.push(command.args.at(3) ?? command.command) + } + return makeHandle({ + pid: ProcessId(1), + exitCode: Effect.succeed(ExitCode(0)), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + stdin: Sink.drain, + stdout: Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }) + }), + ) + connectDriver = () => + Effect.succeed({ + spawner, + overrides: { + write: (path, bytes) => + Effect.sync(() => { + written.push({ path, size: bytes.byteLength }) + }), + }, + }) + + const workspace = yield* Workspace.Service + const workspaceID = yield* workspace.create({ provider: "fake" }) + const files = makeFiles(yield* workspace.connect(workspaceID)) + + yield* files.write("/workspace/direct.txt", new Uint8Array([1, 2, 3])) + expect(written).toEqual([{ path: "/workspace/direct.txt", size: 3 }]) + expect(spawned).toEqual([]) + expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) + + yield* files.mkdir("/workspace/sub") + yield* files.remove("/workspace/direct.txt") + expect(spawned).toEqual(["/workspace/sub", "/workspace/direct.txt"]) + expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) + }), +) + +it.effect("tracks active operations and lastActivity across file overrides", () => + Effect.gen(function* () { + const started = yield* Deferred.make() + const release = yield* Deferred.make() + connectDriver = () => + Effect.succeed({ + spawner: memory.spawner, + overrides: { + ...memory.overrides, + read: (path) => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.as({ + info: { type: "file" as const, size: 2, mtimeMs: 0 }, + bytes: new Uint8Array([1, 2]), + }), + ), + }, + }) + + const workspace = yield* Workspace.Service + const workspaceID = yield* workspace.create({ provider: "fake" }) + const files = makeFiles(yield* workspace.connect(workspaceID)) + + const reading = yield* files.read("/workspace/in-flight.txt").pipe(Effect.forkScoped({ startImmediately: true })) + yield* Deferred.await(started) + + yield* TestClock.adjust("6 minutes") + expect(calls.map((call) => call.operation)).toEqual(["create", "connect"]) + + yield* Deferred.succeed(release, undefined) + expect((yield* Fiber.join(reading)).bytes).toEqual(new Uint8Array([1, 2])) + + 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, workspaceID)).get(), + ).pipe(Effect.orDie) + expect(stored?.last_used_at).toBe(6 * 60 * 1000) + }), +) + +it.effect("surfaces wake failures during file overrides as Environment.Failed", () => + Effect.gen(function* () { + const workspace = yield* Workspace.Service + const created = yield* workspace.provision(yield* workspace.create({ provider: "fake" })) + const files = makeFiles(yield* workspace.connect(created.id)) + yield* files.mkdir("/warm") + + yield* TestClock.adjust("6 minutes") + failConnect = true + + const readError = yield* files.read("/target.txt").pipe(Effect.flip) + expect(readError).toEqual( + new Failed({ + path: "/target.txt", + cause: new WorkspaceDriver.Error({ message: "wake failed" }), + }), + ) + + const moveError = yield* files.move("/from.txt", "/to.txt").pipe(Effect.flip) + expect(moveError).toEqual( + new Failed({ + path: "/from.txt", + cause: new WorkspaceDriver.Error({ message: "wake failed" }), + }), + ) + }), +) diff --git a/packages/sdk/src/effect/index.ts b/packages/sdk/src/effect/index.ts index 6523aa5fbab..b6cf0b0944a 100644 --- a/packages/sdk/src/effect/index.ts +++ b/packages/sdk/src/effect/index.ts @@ -1,5 +1,8 @@ export * as OpenCode from "./opencode" export * as Tool from "./tool" +export { EnvironmentDriver } from "@opencode/core/environment/driver" +export { EnvironmentFiles } from "@opencode/core/environment/files" +export { WorkspaceDriver } from "@opencode/core/workspace/driver" export { ClientError } from "@opencode/client/effect" export type { OpenCodeEvent } from "@opencode/client/effect" diff --git a/packages/sdk/test/contract-identity.test.ts b/packages/sdk/test/contract-identity.test.ts index 6e6920e5054..6149bb05fa8 100644 --- a/packages/sdk/test/contract-identity.test.ts +++ b/packages/sdk/test/contract-identity.test.ts @@ -1,5 +1,8 @@ import { expect, test } from "bun:test" +import { EnvironmentDriver } from "@opencode/core/environment/driver" +import { EnvironmentFiles } from "@opencode/core/environment/files" import { Location as CoreLocation } from "@opencode/core/location" +import { WorkspaceDriver } from "@opencode/core/workspace/driver" import { SessionInbox as CoreSessionInbox } from "@opencode/core/session/inbox" import { SessionMessage as CoreSessionMessage } from "@opencode/core/session/message" import { Agent } from "@opencode/schema/agent" @@ -20,6 +23,7 @@ import { ClientApi, groupNames, promiseOmitEndpoints } from "@opencode/protocol/ import { compile, emitPromise } from "@opencode/httpapi-codegen" const SDK = await import("../src/index") +const SDKEffect = await import("../src/effect") const CoreAgent = await import("@opencode/core/agent") const CoreModel = await import("@opencode/core/model") const CoreProject = await import("@opencode/core/project") @@ -98,3 +102,9 @@ test("client and Server contracts generate identically", () => { expect(emitPromise(client)).toEqual(emitPromise(server)) }) + +test("re-exports workspace and environment driver contracts from @opencode/sdk/effect", () => { + expect(SDKEffect.WorkspaceDriver).toBe(WorkspaceDriver) + expect(SDKEffect.EnvironmentDriver).toBe(EnvironmentDriver) + expect(SDKEffect.EnvironmentFiles).toBe(EnvironmentFiles) +})