feat(core): basic implementation of remote workspace support (#15120)

packages/opencode/src/control-plane/adaptors/index.ts

@@ -0,0 +1,10 @@
+import { WorktreeAdaptor } from "./worktree"
+import type { Config } from "../config"
+import type { Adaptor } from "./types"
+
+export function getAdaptor(config: Config): Adaptor {
+  switch (config.type) {
+    case "worktree":
+      return WorktreeAdaptor
+  }
+}

packages/opencode/src/control-plane/adaptors/types.ts

@@ -0,0 +1,7 @@
+import type { Config } from "../config"
+
+export type Adaptor<T extends Config = Config> = {
+  create(from: T, branch?: string | null): Promise<{ config: T; init: () => Promise<void> }>
+  remove(from: T): Promise<void>
+  request(from: T, method: string, url: string, data?: BodyInit, signal?: AbortSignal): Promise<Response | undefined>
+}

packages/opencode/src/control-plane/adaptors/worktree.ts

@@ -0,0 +1,26 @@
+import { Worktree } from "@/worktree"
+import type { Config } from "../config"
+import type { Adaptor } from "./types"
+
+type WorktreeConfig = Extract<Config, { type: "worktree" }>
+
+export const WorktreeAdaptor: Adaptor<WorktreeConfig> = {
+  async create(_from: WorktreeConfig, _branch: string) {
+    const next = await Worktree.create(undefined)
+    return {
+      config: {
+        type: "worktree",
+        directory: next.directory,
+      },
+      // Hack for now: `Worktree.create` puts all its async code in a
+      // `setTimeout` so it doesn't use this, but we should change that
+      init: async () => {},
+    }
+  },
+  async remove(config: WorktreeConfig) {
+    await Worktree.remove({ directory: config.directory })
+  },
+  async request(_from: WorktreeConfig, _method: string, _url: string, _data?: BodyInit, _signal?: AbortSignal) {
+    throw new Error("worktree does not support request")
+  },
+}

packages/opencode/src/control-plane/config.ts

@@ -0,0 +1,10 @@
+import z from "zod"
+
+export const Config = z.discriminatedUnion("type", [
+  z.object({
+    directory: z.string(),
+    type: z.literal("worktree"),
+  }),
+])
+
+export type Config = z.infer<typeof Config>

packages/opencode/src/control-plane/session-proxy-middleware.ts

@@ -0,0 +1,46 @@
+import { Instance } from "@/project/instance"
+import type { MiddlewareHandler } from "hono"
+import { Installation } from "../installation"
+import { getAdaptor } from "./adaptors"
+import { Workspace } from "./workspace"
+
+// This middleware forwards all non-GET requests if the workspace is a
+// remote. The remote workspace needs to handle session mutations
+async function proxySessionRequest(req: Request) {
+  if (req.method === "GET") return
+  if (!Instance.directory.startsWith("wrk_")) return
+
+  const workspace = await Workspace.get(Instance.directory)
+  if (!workspace) {
+    return new Response(`Workspace not found: ${Instance.directory}`, {
+      status: 500,
+      headers: {
+        "content-type": "text/plain; charset=utf-8",
+      },
+    })
+  }
+  if (workspace.config.type === "worktree") return
+
+  const url = new URL(req.url)
+  const body = req.method === "HEAD" ? undefined : await req.arrayBuffer()
+  return getAdaptor(workspace.config).request(
+    workspace.config,
+    req.method,
+    `${url.pathname}${url.search}`,
+    body,
+    req.signal,
+  )
+}
+
+export const SessionProxyMiddleware: MiddlewareHandler = async (c, next) => {
+  // Only available in development for now
+  if (!Installation.isLocal()) {
+    return next()
+  }
+
+  const response = await proxySessionRequest(c.req.raw)
+  if (response) {
+    return response
+  }
+  return next()
+}

packages/opencode/src/control-plane/sse.ts

@@ -0,0 +1,66 @@
+export async function parseSSE(
+  body: ReadableStream<Uint8Array>,
+  signal: AbortSignal,
+  onEvent: (event: unknown) => void,
+) {
+  const reader = body.getReader()
+  const decoder = new TextDecoder()
+  let buf = ""
+  let last = ""
+  let retry = 1000
+
+  const abort = () => {
+    void reader.cancel().catch(() => undefined)
+  }
+
+  signal.addEventListener("abort", abort)
+
+  try {
+    while (!signal.aborted) {
+      const chunk = await reader.read().catch(() => ({ done: true, value: undefined as Uint8Array | undefined }))
+      if (chunk.done) break
+
+      buf += decoder.decode(chunk.value, { stream: true })
+      buf = buf.replace(/\r\n/g, "\n").replace(/\r/g, "\n")
+
+      const chunks = buf.split("\n\n")
+      buf = chunks.pop() ?? ""
+
+      chunks.forEach((chunk) => {
+        const data: string[] = []
+        chunk.split("\n").forEach((line) => {
+          if (line.startsWith("data:")) {
+            data.push(line.replace(/^data:\s*/, ""))
+            return
+          }
+          if (line.startsWith("id:")) {
+            last = line.replace(/^id:\s*/, "")
+            return
+          }
+          if (line.startsWith("retry:")) {
+            const parsed = Number.parseInt(line.replace(/^retry:\s*/, ""), 10)
+            if (!Number.isNaN(parsed)) retry = parsed
+          }
+        })
+
+        if (!data.length) return
+        const raw = data.join("\n")
+        try {
+          onEvent(JSON.parse(raw))
+        } catch {
+          onEvent({
+            type: "sse.message",
+            properties: {
+              data: raw,
+              id: last || undefined,
+              retry,
+            },
+          })
+        }
+      })
+    }
+  } finally {
+    signal.removeEventListener("abort", abort)
+    reader.releaseLock()
+  }
+}

packages/opencode/src/control-plane/workspace-server/routes.ts

@@ -0,0 +1,33 @@
+import { GlobalBus } from "../../bus/global"
+import { Hono } from "hono"
+import { streamSSE } from "hono/streaming"
+
+export function WorkspaceServerRoutes() {
+  return new Hono().get("/event", async (c) => {
+    c.header("X-Accel-Buffering", "no")
+    c.header("X-Content-Type-Options", "nosniff")
+    return streamSSE(c, async (stream) => {
+      const send = async (event: unknown) => {
+        await stream.writeSSE({
+          data: JSON.stringify(event),
+        })
+      }
+      const handler = async (event: { directory?: string; payload: unknown }) => {
+        await send(event.payload)
+      }
+      GlobalBus.on("event", handler)
+      await send({ type: "server.connected", properties: {} })
+      const heartbeat = setInterval(() => {
+        void send({ type: "server.heartbeat", properties: {} })
+      }, 10_000)
+
+      await new Promise<void>((resolve) => {
+        stream.onAbort(() => {
+          clearInterval(heartbeat)
+          GlobalBus.off("event", handler)
+          resolve()
+        })
+      })
+    })
+  })
+}

packages/opencode/src/control-plane/workspace-server/server.ts

@@ -0,0 +1,24 @@
+import { Hono } from "hono"
+import { SessionRoutes } from "../../server/routes/session"
+import { WorkspaceServerRoutes } from "./routes"
+
+export namespace WorkspaceServer {
+  export function App() {
+    const session = new Hono()
+      .use("*", async (c, next) => {
+        if (c.req.method === "GET") return c.notFound()
+        await next()
+      })
+      .route("/", SessionRoutes())
+
+    return new Hono().route("/session", session).route("/", WorkspaceServerRoutes())
+  }
+
+  export function Listen(opts: { hostname: string; port: number }) {
+    return Bun.serve({
+      hostname: opts.hostname,
+      port: opts.port,
+      fetch: App().fetch,
+    })
+  }
+}

packages/opencode/src/control-plane/workspace.sql.ts

@@ -0,0 +1,12 @@
+import { sqliteTable, text } from "drizzle-orm/sqlite-core"
+import { ProjectTable } from "@/project/project.sql"
+import type { Config } from "./config"
+
+export const WorkspaceTable = sqliteTable("workspace", {
+  id: text().primaryKey(),
+  branch: text(),
+  project_id: text()
+    .notNull()
+    .references(() => ProjectTable.id, { onDelete: "cascade" }),
+  config: text({ mode: "json" }).notNull().$type<Config>(),
+})

packages/opencode/src/control-plane/workspace.ts

@@ -0,0 +1,160 @@
+import z from "zod"
+import { Identifier } from "@/id/id"
+import { fn } from "@/util/fn"
+import { Database, eq } from "@/storage/db"
+import { Project } from "@/project/project"
+import { BusEvent } from "@/bus/bus-event"
+import { GlobalBus } from "@/bus/global"
+import { Log } from "@/util/log"
+import { WorkspaceTable } from "./workspace.sql"
+import { Config } from "./config"
+import { getAdaptor } from "./adaptors"
+import { parseSSE } from "./sse"
+
+export namespace Workspace {
+  export const Event = {
+    Ready: BusEvent.define(
+      "workspace.ready",
+      z.object({
+        name: z.string(),
+      }),
+    ),
+    Failed: BusEvent.define(
+      "workspace.failed",
+      z.object({
+        message: z.string(),
+      }),
+    ),
+  }
+
+  export const Info = z
+    .object({
+      id: Identifier.schema("workspace"),
+      branch: z.string().nullable(),
+      projectID: z.string(),
+      config: Config,
+    })
+    .meta({
+      ref: "Workspace",
+    })
+  export type Info = z.infer<typeof Info>
+
+  function fromRow(row: typeof WorkspaceTable.$inferSelect): Info {
+    return {
+      id: row.id,
+      branch: row.branch,
+      projectID: row.project_id,
+      config: row.config,
+    }
+  }
+
+  export const create = fn(
+    z.object({
+      id: Identifier.schema("workspace").optional(),
+      projectID: Info.shape.projectID,
+      branch: Info.shape.branch,
+      config: Info.shape.config,
+    }),
+    async (input) => {
+      const id = Identifier.ascending("workspace", input.id)
+
+      const { config, init } = await getAdaptor(input.config).create(input.config, input.branch)
+
+      const info: Info = {
+        id,
+        projectID: input.projectID,
+        branch: input.branch,
+        config,
+      }
+
+      setTimeout(async () => {
+        await init()
+
+        Database.use((db) => {
+          db.insert(WorkspaceTable)
+            .values({
+              id: info.id,
+              branch: info.branch,
+              project_id: info.projectID,
+              config: info.config,
+            })
+            .run()
+        })
+
+        GlobalBus.emit("event", {
+          directory: id,
+          payload: {
+            type: Event.Ready.type,
+            properties: {},
+          },
+        })
+      }, 0)
+
+      return info
+    },
+  )
+
+  export function list(project: Project.Info) {
+    const rows = Database.use((db) =>
+      db.select().from(WorkspaceTable).where(eq(WorkspaceTable.project_id, project.id)).all(),
+    )
+    return rows.map(fromRow).sort((a, b) => a.id.localeCompare(b.id))
+  }
+
+  export const get = fn(Identifier.schema("workspace"), async (id) => {
+    const row = Database.use((db) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get())
+    if (!row) return
+    return fromRow(row)
+  })
+
+  export const remove = fn(Identifier.schema("workspace"), async (id) => {
+    const row = Database.use((db) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get())
+    if (row) {
+      const info = fromRow(row)
+      await getAdaptor(info.config).remove(info.config)
+      Database.use((db) => db.delete(WorkspaceTable).where(eq(WorkspaceTable.id, id)).run())
+      return info
+    }
+  })
+  const log = Log.create({ service: "workspace-sync" })
+
+  async function workspaceEventLoop(space: Info, stop: AbortSignal) {
+    while (!stop.aborted) {
+      const res = await getAdaptor(space.config)
+        .request(space.config, "GET", "/event", undefined, stop)
+        .catch(() => undefined)
+      if (!res || !res.ok || !res.body) {
+        await Bun.sleep(1000)
+        continue
+      }
+      await parseSSE(res.body, stop, (event) => {
+        GlobalBus.emit("event", {
+          directory: space.id,
+          payload: event,
+        })
+      })
+      // Wait 250ms and retry if SSE connection fails
+      await Bun.sleep(250)
+    }
+  }
+
+  export function startSyncing(project: Project.Info) {
+    const stop = new AbortController()
+    const spaces = list(project).filter((space) => space.config.type !== "worktree")
+
+    spaces.forEach((space) => {
+      void workspaceEventLoop(space, stop.signal).catch((error) => {
+        log.warn("workspace sync listener failed", {
+          workspaceID: space.id,
+          error,
+        })
+      })
+    })
+
+    return {
+      async stop() {
+        stop.abort()
+      },
+    }
+  }
+}