chore: initialize qiming workspace repository

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

@@ -0,0 +1,46 @@
+import { lazy } from "@/util/lazy"
+import type { ProjectID } from "@/project/schema"
+import type { WorkspaceAdaptor, WorkspaceAdaptorEntry } from "../types"
+
+const BUILTIN: Record<string, () => Promise<WorkspaceAdaptor>> = {
+  worktree: lazy(async () => (await import("./worktree")).WorktreeAdaptor),
+}
+
+const state = new Map<ProjectID, Map<string, WorkspaceAdaptor>>()
+
+export async function getAdaptor(projectID: ProjectID, type: string): Promise<WorkspaceAdaptor> {
+  const custom = state.get(projectID)?.get(type)
+  if (custom) return custom
+
+  const builtin = BUILTIN[type]
+  if (builtin) return builtin()
+
+  throw new Error(`Unknown workspace adaptor: ${type}`)
+}
+
+export async function listAdaptors(projectID: ProjectID): Promise<WorkspaceAdaptorEntry[]> {
+  const builtin = await Promise.all(
+    Object.entries(BUILTIN).map(async ([type, init]) => {
+      const adaptor = await init()
+      return {
+        type,
+        name: adaptor.name,
+        description: adaptor.description,
+      }
+    }),
+  )
+  const custom = [...(state.get(projectID)?.entries() ?? [])].map(([type, adaptor]) => ({
+    type,
+    name: adaptor.name,
+    description: adaptor.description,
+  }))
+  return [...builtin, ...custom]
+}
+
+// Plugins can be loaded per-project so we need to scope them. If you
+// want to install a global one pass `ProjectID.global`
+export function registerAdaptor(projectID: ProjectID, type: string, adaptor: WorkspaceAdaptor) {
+  const adaptors = state.get(projectID) ?? new Map<string, WorkspaceAdaptor>()
+  adaptors.set(type, adaptor)
+  state.set(projectID, adaptors)
+}

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

@@ -0,0 +1,48 @@
+import { Schema } from "effect"
+import { AppRuntime } from "@/effect/app-runtime"
+import { Worktree } from "@/worktree"
+import { type WorkspaceAdaptor, WorkspaceInfo } from "../types"
+
+const WorktreeConfig = Schema.Struct({
+  name: WorkspaceInfo.fields.name,
+  branch: Schema.String,
+  directory: Schema.String,
+})
+const decodeWorktreeConfig = Schema.decodeUnknownSync(WorktreeConfig)
+
+export const WorktreeAdaptor: WorkspaceAdaptor = {
+  name: "Worktree",
+  description: "Create a git worktree",
+  async configure(info) {
+    const worktree = await AppRuntime.runPromise(Worktree.Service.use((svc) => svc.makeWorktreeInfo()))
+    return {
+      ...info,
+      name: worktree.name,
+      branch: worktree.branch,
+      directory: worktree.directory,
+    }
+  },
+  async create(info) {
+    const config = decodeWorktreeConfig(info)
+    await AppRuntime.runPromise(
+      Worktree.Service.use((svc) =>
+        svc.createFromInfo({
+          name: config.name,
+          directory: config.directory,
+          branch: config.branch,
+        }),
+      ),
+    )
+  },
+  async remove(info) {
+    const config = decodeWorktreeConfig(info)
+    await AppRuntime.runPromise(Worktree.Service.use((svc) => svc.remove({ directory: config.directory })))
+  },
+  target(info) {
+    const config = decodeWorktreeConfig(info)
+    return {
+      type: "local",
+      directory: config.directory,
+    }
+  },
+}

qimingcode/packages/opencode/src/control-plane/dev/debug-workspace-plugin.ts

@@ -0,0 +1,73 @@
+import type { Plugin } from "@opencode-ai/plugin"
+import { rename, writeFile } from "node:fs/promises"
+import { randomInt } from "node:crypto"
+import { setTimeout as sleep } from "node:timers/promises"
+
+const DEV_DATA_FILE = "/tmp/opencode-workspace-dev-data.json"
+const DEV_DATA_TEMP_FILE = `${DEV_DATA_FILE}.tmp`
+
+async function waitForHealth(port: number) {
+  const url = `http://127.0.0.1:${port}/global/health`
+  const started = Date.now()
+
+  while (Date.now() - started < 30_000) {
+    try {
+      const response = await fetch(url)
+      if (response.ok) {
+        return
+      }
+    } catch {}
+
+    await sleep(250)
+  }
+
+  throw new Error(`Timed out waiting for debug server health check at ${url}`)
+}
+
+let PORT: number | undefined
+
+async function writeDebugData(port: number, id: string, env: Record<string, string | undefined>) {
+  await writeFile(
+    DEV_DATA_TEMP_FILE,
+    JSON.stringify(
+      {
+        port,
+        id,
+        env,
+      },
+      null,
+      2,
+    ),
+  )
+
+  await rename(DEV_DATA_TEMP_FILE, DEV_DATA_FILE)
+}
+
+export const DebugWorkspacePlugin: Plugin = async ({ experimental_workspace }) => {
+  experimental_workspace.register("debug", {
+    name: "Debug",
+    description: "Create a debugging server",
+    configure(config) {
+      return config
+    },
+    async create(config, env) {
+      const port = randomInt(5000, 9001)
+      PORT = port
+
+      await writeDebugData(port, config.id, env)
+
+      await waitForHealth(port)
+    },
+    async remove(_config) {},
+    target(_config) {
+      return {
+        type: "remote",
+        url: `http://localhost:${PORT!}/`,
+      }
+    },
+  })
+
+  return {}
+}
+
+export default DebugWorkspacePlugin

qimingcode/packages/opencode/src/control-plane/schema.ts

@@ -0,0 +1,18 @@
+import { Schema } from "effect"
+
+import { Identifier } from "@/id/id"
+import { zod, ZodOverride } from "@/util/effect-zod"
+import { withStatics } from "@/util/schema"
+
+const workspaceIdSchema = Schema.String.annotate({ [ZodOverride]: Identifier.schema("workspace") }).pipe(
+  Schema.brand("WorkspaceID"),
+)
+
+export type WorkspaceID = typeof workspaceIdSchema.Type
+
+export const WorkspaceID = workspaceIdSchema.pipe(
+  withStatics((schema: typeof workspaceIdSchema) => ({
+    ascending: (id?: string) => schema.make(Identifier.ascending("workspace", id)),
+    zod: zod(schema),
+  })),
+)

qimingcode/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()
+  }
+}

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

@@ -0,0 +1,45 @@
+import { Schema } from "effect"
+import { ProjectID } from "@/project/schema"
+import { WorkspaceID } from "./schema"
+import { zod } from "@/util/effect-zod"
+import { type DeepMutable, withStatics } from "@/util/schema"
+
+export const WorkspaceInfo = Schema.Struct({
+  id: WorkspaceID,
+  type: Schema.String,
+  name: Schema.String,
+  branch: Schema.NullOr(Schema.String),
+  directory: Schema.NullOr(Schema.String),
+  extra: Schema.NullOr(Schema.Unknown),
+  projectID: ProjectID,
+})
+  .annotate({ identifier: "Workspace" })
+  .pipe(withStatics((s) => ({ zod: zod(s) })))
+export type WorkspaceInfo = DeepMutable<Schema.Schema.Type<typeof WorkspaceInfo>>
+
+export const WorkspaceAdaptorEntry = Schema.Struct({
+  type: Schema.String,
+  name: Schema.String,
+  description: Schema.String,
+}).pipe(withStatics((s) => ({ zod: zod(s) })))
+export type WorkspaceAdaptorEntry = Schema.Schema.Type<typeof WorkspaceAdaptorEntry>
+
+export type Target =
+  | {
+      type: "local"
+      directory: string
+    }
+  | {
+      type: "remote"
+      url: string | URL
+      headers?: HeadersInit
+    }
+
+export type WorkspaceAdaptor = {
+  name: string
+  description: string
+  configure(info: WorkspaceInfo): WorkspaceInfo | Promise<WorkspaceInfo>
+  create(info: WorkspaceInfo, env: Record<string, string | undefined>, from?: WorkspaceInfo): Promise<void>
+  remove(info: WorkspaceInfo): Promise<void>
+  target(info: WorkspaceInfo): Target | Promise<Target>
+}

qimingcode/packages/opencode/src/control-plane/util.ts

@@ -0,0 +1,37 @@
+import { GlobalBus, type GlobalEvent } from "@/bus/global"
+
+export function waitEvent(input: { timeout: number; signal?: AbortSignal; fn: (event: GlobalEvent) => boolean }) {
+  if (input.signal?.aborted) return Promise.reject(input.signal.reason ?? new Error("Request aborted"))
+
+  return new Promise<void>((resolve, reject) => {
+    const abort = () => {
+      cleanup()
+      reject(input.signal?.reason ?? new Error("Request aborted"))
+    }
+
+    const handler = (event: GlobalEvent) => {
+      try {
+        if (!input.fn(event)) return
+        cleanup()
+        resolve()
+      } catch (error) {
+        cleanup()
+        reject(error)
+      }
+    }
+
+    const cleanup = () => {
+      clearTimeout(timeout)
+      GlobalBus.off("event", handler)
+      input.signal?.removeEventListener("abort", abort)
+    }
+
+    const timeout = setTimeout(() => {
+      cleanup()
+      reject(new Error("Timed out waiting for global event"))
+    }, input.timeout)
+
+    GlobalBus.on("event", handler)
+    input.signal?.addEventListener("abort", abort, { once: true })
+  })
+}

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

@@ -0,0 +1,26 @@
+import { LocalContext } from "../util"
+import type { WorkspaceID } from "../control-plane/schema"
+
+export interface WorkspaceContext {
+  workspaceID: WorkspaceID | undefined
+}
+
+const context = LocalContext.create<WorkspaceContext>("instance")
+
+export const WorkspaceContext = {
+  async provide<R>(input: { workspaceID?: WorkspaceID; fn: () => R }): Promise<R> {
+    return context.provide({ workspaceID: input.workspaceID }, () => input.fn())
+  },
+
+  restore<R>(workspaceID: WorkspaceID, fn: () => R): R {
+    return context.provide({ workspaceID }, fn)
+  },
+
+  get workspaceID() {
+    try {
+      return context.use().workspaceID
+    } catch {
+      return undefined
+    }
+  },
+}

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

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

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

@@ -0,0 +1,617 @@
+import { Schema } from "effect"
+import { setTimeout as sleep } from "node:timers/promises"
+import { fn } from "@/util/fn"
+import { Database, asc, eq, inArray } from "@/storage"
+import { Project } from "@/project"
+import { BusEvent } from "@/bus/bus-event"
+import { GlobalBus } from "@/bus/global"
+import { Auth } from "@/auth"
+import { SyncEvent } from "@/sync"
+import { EventSequenceTable, EventTable } from "@/sync/event.sql"
+import { Flag } from "@opencode-ai/core/flag/flag"
+import { Log } from "@/util"
+import { Filesystem } from "@/util"
+import { ProjectID } from "@/project/schema"
+import { Slug } from "@opencode-ai/core/util/slug"
+import { WorkspaceTable } from "./workspace.sql"
+import { getAdaptor } from "./adaptors"
+import { type WorkspaceInfo, WorkspaceInfo as WorkspaceInfoSchema } from "./types"
+import { WorkspaceID } from "./schema"
+import { parseSSE } from "./sse"
+import { Session } from "@/session"
+import { SessionTable } from "@/session/session.sql"
+import { SessionID } from "@/session/schema"
+import { errorData } from "@/util/error"
+import { AppRuntime } from "@/effect/app-runtime"
+import { waitEvent } from "./util"
+import { WorkspaceContext } from "./workspace-context"
+import { NonNegativeInt, withStatics } from "@/util/schema"
+import { zod as effectZod, zodObject } from "@/util/effect-zod"
+
+export const Info = WorkspaceInfoSchema
+export type Info = WorkspaceInfo
+
+export const ConnectionStatus = Schema.Struct({
+  workspaceID: WorkspaceID,
+  status: Schema.Literals(["connected", "connecting", "disconnected", "error"]),
+})
+export type ConnectionStatus = Schema.Schema.Type<typeof ConnectionStatus>
+
+const Restore = Schema.Struct({
+  workspaceID: WorkspaceID,
+  sessionID: SessionID,
+  total: NonNegativeInt,
+  step: NonNegativeInt,
+})
+
+export const Event = {
+  Ready: BusEvent.define(
+    "workspace.ready",
+    Schema.Struct({
+      name: Schema.String,
+    }),
+  ),
+  Failed: BusEvent.define(
+    "workspace.failed",
+    Schema.Struct({
+      message: Schema.String,
+    }),
+  ),
+  Restore: BusEvent.define("workspace.restore", Restore),
+  Status: BusEvent.define("workspace.status", ConnectionStatus),
+}
+
+function fromRow(row: typeof WorkspaceTable.$inferSelect): Info {
+  return {
+    id: row.id,
+    type: row.type,
+    branch: row.branch,
+    name: row.name,
+    directory: row.directory,
+    extra: row.extra,
+    projectID: row.project_id,
+  }
+}
+
+export const CreateInput = Schema.Struct({
+  id: Schema.optional(WorkspaceID),
+  type: Info.fields.type,
+  branch: Info.fields.branch,
+  projectID: ProjectID,
+  extra: Info.fields.extra,
+}).pipe(withStatics((s) => ({ zod: effectZod(s), zodObject: zodObject(s) })))
+export type CreateInput = Schema.Schema.Type<typeof CreateInput>
+
+export const create = fn(CreateInput.zod, async (input) => {
+  const id = WorkspaceID.ascending(input.id)
+  const adaptor = await getAdaptor(input.projectID, input.type)
+
+  const config = await adaptor.configure({ ...input, id, name: Slug.create(), directory: null })
+
+  const info: Info = {
+    id,
+    type: config.type,
+    branch: config.branch ?? null,
+    name: config.name ?? null,
+    directory: config.directory ?? null,
+    extra: config.extra ?? null,
+    projectID: input.projectID,
+  }
+
+  Database.use((db) => {
+    db.insert(WorkspaceTable)
+      .values({
+        id: info.id,
+        type: info.type,
+        branch: info.branch,
+        name: info.name,
+        directory: info.directory,
+        extra: info.extra,
+        project_id: info.projectID,
+      })
+      .run()
+  })
+
+  const env = {
+    OPENCODE_AUTH_CONTENT: JSON.stringify(await AppRuntime.runPromise(Auth.Service.use((auth) => auth.all()))),
+    OPENCODE_WORKSPACE_ID: config.id,
+    OPENCODE_EXPERIMENTAL_WORKSPACES: "true",
+    OTEL_EXPORTER_OTLP_HEADERS: process.env.OTEL_EXPORTER_OTLP_HEADERS,
+    OTEL_EXPORTER_OTLP_ENDPOINT: process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
+    OTEL_RESOURCE_ATTRIBUTES: process.env.OTEL_RESOURCE_ATTRIBUTES,
+  }
+  await adaptor.create(config, env)
+
+  startSync(info)
+
+  await waitEvent({
+    timeout: TIMEOUT,
+    fn(event) {
+      if (event.workspace === info.id && event.payload.type === Event.Status.type) {
+        const { status } = event.payload.properties
+        return status === "error" || status === "connected"
+      }
+      return false
+    },
+  })
+
+  return info
+})
+
+export const SessionRestoreInput = Schema.Struct({
+  workspaceID: WorkspaceID,
+  sessionID: SessionID,
+}).pipe(withStatics((s) => ({ zod: effectZod(s), zodObject: zodObject(s) })))
+export type SessionRestoreInput = Schema.Schema.Type<typeof SessionRestoreInput>
+
+export const sessionRestore = fn(SessionRestoreInput.zod, async (input) => {
+  log.info("session restore requested", {
+    workspaceID: input.workspaceID,
+    sessionID: input.sessionID,
+  })
+  try {
+    const space = await get(input.workspaceID)
+    if (!space) throw new Error(`Workspace not found: ${input.workspaceID}`)
+
+    const adaptor = await getAdaptor(space.projectID, space.type)
+    const target = await adaptor.target(space)
+
+    // Need to switch the workspace of the session
+    SyncEvent.run(Session.Event.Updated, {
+      sessionID: input.sessionID,
+      info: {
+        workspaceID: input.workspaceID,
+      },
+    })
+
+    const rows = Database.use((db) =>
+      db
+        .select({
+          id: EventTable.id,
+          aggregateID: EventTable.aggregate_id,
+          seq: EventTable.seq,
+          type: EventTable.type,
+          data: EventTable.data,
+        })
+        .from(EventTable)
+        .where(eq(EventTable.aggregate_id, input.sessionID))
+        .orderBy(asc(EventTable.seq))
+        .all(),
+    )
+    if (rows.length === 0) throw new Error(`No events found for session: ${input.sessionID}`)
+
+    const all = rows
+
+    const size = 10
+    const sets = Array.from({ length: Math.ceil(all.length / size) }, (_, i) => all.slice(i * size, (i + 1) * size))
+    const total = sets.length
+    log.info("session restore prepared", {
+      workspaceID: input.workspaceID,
+      sessionID: input.sessionID,
+      workspaceType: space.type,
+      directory: space.directory,
+      target: target.type === "remote" ? String(route(target.url, "/sync/replay")) : target.directory,
+      events: all.length,
+      batches: total,
+      first: all[0]?.seq,
+      last: all.at(-1)?.seq,
+    })
+    GlobalBus.emit("event", {
+      directory: "global",
+      workspace: input.workspaceID,
+      payload: {
+        type: Event.Restore.type,
+        properties: {
+          workspaceID: input.workspaceID,
+          sessionID: input.sessionID,
+          total,
+          step: 0,
+        },
+      },
+    })
+    for (const [i, events] of sets.entries()) {
+      log.info("session restore batch starting", {
+        workspaceID: input.workspaceID,
+        sessionID: input.sessionID,
+        step: i + 1,
+        total,
+        events: events.length,
+        first: events[0]?.seq,
+        last: events.at(-1)?.seq,
+        target: target.type === "remote" ? String(route(target.url, "/sync/replay")) : target.directory,
+      })
+      if (target.type === "local") {
+        SyncEvent.replayAll(events)
+        log.info("session restore batch replayed locally", {
+          workspaceID: input.workspaceID,
+          sessionID: input.sessionID,
+          step: i + 1,
+          total,
+          events: events.length,
+        })
+      } else {
+        const url = route(target.url, "/sync/replay")
+        const headers = new Headers(target.headers)
+        headers.set("content-type", "application/json")
+        const res = await fetch(url, {
+          method: "POST",
+          headers,
+          body: JSON.stringify({
+            directory: space.directory ?? "",
+            events,
+          }),
+        })
+        if (!res.ok) {
+          const body = await res.text()
+          log.error("session restore batch failed", {
+            workspaceID: input.workspaceID,
+            sessionID: input.sessionID,
+            step: i + 1,
+            total,
+            status: res.status,
+            body,
+          })
+          throw new Error(
+            `Failed to replay session ${input.sessionID} into workspace ${input.workspaceID}: HTTP ${res.status} ${body}`,
+          )
+        }
+        log.info("session restore batch posted", {
+          workspaceID: input.workspaceID,
+          sessionID: input.sessionID,
+          step: i + 1,
+          total,
+          status: res.status,
+        })
+      }
+      GlobalBus.emit("event", {
+        directory: "global",
+        workspace: input.workspaceID,
+        payload: {
+          type: Event.Restore.type,
+          properties: {
+            workspaceID: input.workspaceID,
+            sessionID: input.sessionID,
+            total,
+            step: i + 1,
+          },
+        },
+      })
+    }
+
+    log.info("session restore complete", {
+      workspaceID: input.workspaceID,
+      sessionID: input.sessionID,
+      batches: total,
+    })
+
+    return {
+      total,
+    }
+  } catch (err) {
+    log.error("session restore failed", {
+      workspaceID: input.workspaceID,
+      sessionID: input.sessionID,
+      error: errorData(err),
+    })
+    throw err
+  }
+})
+
+export function list(project: Project.Info) {
+  const rows = Database.use((db) =>
+    db.select().from(WorkspaceTable).where(eq(WorkspaceTable.project_id, project.id)).all(),
+  )
+  const spaces = rows.map(fromRow).sort((a, b) => a.id.localeCompare(b.id))
+  return spaces
+}
+
+export const get = fn(WorkspaceID.zod, 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(WorkspaceID.zod, async (id) => {
+  const sessions = Database.use((db) =>
+    db.select({ id: SessionTable.id }).from(SessionTable).where(eq(SessionTable.workspace_id, id)).all(),
+  )
+  for (const session of sessions) {
+    await AppRuntime.runPromise(Session.Service.use((svc) => svc.remove(session.id)))
+  }
+
+  const row = Database.use((db) => db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get())
+
+  if (row) {
+    stopSync(id)
+
+    const info = fromRow(row)
+    try {
+      const adaptor = await getAdaptor(info.projectID, row.type)
+      await adaptor.remove(info)
+    } catch {
+      log.error("adaptor not available when removing workspace", { type: row.type })
+    }
+    Database.use((db) => db.delete(WorkspaceTable).where(eq(WorkspaceTable.id, id)).run())
+    return info
+  }
+})
+
+const connections = new Map<WorkspaceID, ConnectionStatus>()
+const aborts = new Map<WorkspaceID, AbortController>()
+const TIMEOUT = 5000
+
+function setStatus(id: WorkspaceID, status: ConnectionStatus["status"]) {
+  const prev = connections.get(id)
+  if (prev?.status === status) return
+  const next = { workspaceID: id, status }
+  connections.set(id, next)
+
+  if (status === "error") {
+    aborts.delete(id)
+  }
+
+  GlobalBus.emit("event", {
+    directory: "global",
+    workspace: id,
+    payload: {
+      type: Event.Status.type,
+      properties: next,
+    },
+  })
+}
+
+export function status(): ConnectionStatus[] {
+  return [...connections.values()]
+}
+
+function synced(state: Record<string, number>) {
+  const ids = Object.keys(state)
+  if (ids.length === 0) return true
+
+  const done = Object.fromEntries(
+    Database.use((db) =>
+      db
+        .select({
+          id: EventSequenceTable.aggregate_id,
+          seq: EventSequenceTable.seq,
+        })
+        .from(EventSequenceTable)
+        .where(inArray(EventSequenceTable.aggregate_id, ids))
+        .all(),
+    ).map((row) => [row.id, row.seq]),
+  ) as Record<string, number>
+
+  return ids.every((id) => {
+    return (done[id] ?? -1) >= state[id]
+  })
+}
+
+export async function isSyncing(workspaceID: WorkspaceID) {
+  return aborts.has(workspaceID)
+}
+
+export async function waitForSync(workspaceID: WorkspaceID, state: Record<string, number>, signal?: AbortSignal) {
+  if (synced(state)) return
+
+  try {
+    await waitEvent({
+      timeout: TIMEOUT,
+      signal,
+      fn(event) {
+        if (event.workspace !== workspaceID && event.payload.type !== "sync") {
+          return false
+        }
+        return synced(state)
+      },
+    })
+  } catch {
+    if (signal?.aborted) throw signal.reason ?? new Error("Request aborted")
+    throw new Error(`Timed out waiting for sync fence: ${JSON.stringify(state)}`)
+  }
+}
+
+const log = Log.create({ service: "workspace-sync" })
+
+function route(url: string | URL, path: string) {
+  const next = new URL(url)
+  next.pathname = `${next.pathname.replace(/\/$/, "")}${path}`
+  next.search = ""
+  next.hash = ""
+  return next
+}
+
+async function connectSSE(url: URL | string, headers: HeadersInit | undefined, signal: AbortSignal) {
+  const res = await fetch(route(url, "/global/event"), {
+    method: "GET",
+    headers,
+    signal,
+  })
+
+  if (!res.ok) throw new Error(`Workspace sync HTTP failure: ${res.status}`)
+  if (!res.body) throw new Error("No response body from global sync")
+
+  return res.body
+}
+
+async function syncHistory(space: Info, url: URL | string, headers: HeadersInit | undefined, signal: AbortSignal) {
+  const sessionIDs = Database.use((db) =>
+    db
+      .select({ id: SessionTable.id })
+      .from(SessionTable)
+      .where(eq(SessionTable.workspace_id, space.id))
+      .all()
+      .map((row) => row.id),
+  )
+  const state = sessionIDs.length
+    ? Object.fromEntries(
+        Database.use((db) =>
+          db.select().from(EventSequenceTable).where(inArray(EventSequenceTable.aggregate_id, sessionIDs)).all(),
+        ).map((row) => [row.aggregate_id, row.seq]),
+      )
+    : {}
+
+  log.info("syncing workspace history", {
+    workspaceID: space.id,
+    sessions: sessionIDs.length,
+    known: Object.keys(state).length,
+  })
+
+  const requestHeaders = new Headers(headers)
+  requestHeaders.set("content-type", "application/json")
+
+  const res = await fetch(route(url, "/sync/history"), {
+    method: "POST",
+    headers: requestHeaders,
+    body: JSON.stringify(state),
+    signal,
+  })
+
+  if (!res.ok) {
+    const body = await res.text()
+    throw new Error(`Workspace history HTTP failure: ${res.status} ${body}`)
+  }
+
+  const events = await res.json()
+
+  return WorkspaceContext.provide({
+    workspaceID: space.id,
+    fn: () => {
+      for (const event of events) {
+        SyncEvent.replay(
+          {
+            id: event.id,
+            aggregateID: event.aggregate_id,
+            seq: event.seq,
+            type: event.type,
+            data: event.data,
+          },
+          { publish: true },
+        )
+      }
+    },
+  })
+
+  log.info("workspace history synced", {
+    workspaceID: space.id,
+    events: events.length,
+  })
+}
+
+async function syncWorkspaceLoop(space: Info, signal: AbortSignal) {
+  const adaptor = await getAdaptor(space.projectID, space.type)
+  const target = await adaptor.target(space)
+
+  if (target.type === "local") return null
+
+  let attempt = 0
+
+  while (!signal.aborted) {
+    log.info("connecting to global sync", { workspace: space.name })
+    setStatus(space.id, "connecting")
+
+    let stream
+    try {
+      stream = await connectSSE(target.url, target.headers, signal)
+      await syncHistory(space, target.url, target.headers, signal)
+    } catch (err) {
+      stream = null
+      setStatus(space.id, "error")
+      log.info("failed to connect to global sync", {
+        workspace: space.name,
+        err,
+      })
+    }
+
+    if (stream) {
+      attempt = 0
+
+      log.info("global sync connected", { workspace: space.name })
+      setStatus(space.id, "connected")
+
+      await parseSSE(stream, signal, (evt: any) => {
+        try {
+          if (!("payload" in evt)) return
+          if (evt.payload.type === "server.heartbeat") return
+
+          if (evt.payload.type === "sync") {
+            SyncEvent.replay(evt.payload.syncEvent as SyncEvent.SerializedEvent)
+          }
+
+          GlobalBus.emit("event", {
+            directory: evt.directory,
+            project: evt.project,
+            workspace: space.id,
+            payload: evt.payload,
+          })
+        } catch (err) {
+          log.info("failed to replay global event", {
+            workspaceID: space.id,
+            error: err,
+          })
+        }
+      })
+
+      log.info("disconnected from global sync: " + space.id)
+      setStatus(space.id, "disconnected")
+    }
+
+    // Back off reconnect attempts up to 2 minutes while the workspace
+    // stays unavailable.
+    await sleep(Math.min(120_000, 1_000 * 2 ** attempt))
+    attempt += 1
+  }
+}
+
+async function startSync(space: Info) {
+  if (!Flag.OPENCODE_EXPERIMENTAL_WORKSPACES) return
+
+  const adaptor = await getAdaptor(space.projectID, space.type)
+  const target = await adaptor.target(space)
+
+  if (target.type === "local") {
+    void Filesystem.exists(target.directory).then((exists) => {
+      setStatus(space.id, exists ? "connected" : "error")
+    })
+    return
+  }
+
+  if (aborts.has(space.id)) return true
+
+  setStatus(space.id, "disconnected")
+
+  const abort = new AbortController()
+  aborts.set(space.id, abort)
+
+  void syncWorkspaceLoop(space, abort.signal).catch((error) => {
+    aborts.delete(space.id)
+
+    setStatus(space.id, "error")
+    log.warn("workspace listener failed", {
+      workspaceID: space.id,
+      error,
+    })
+  })
+}
+
+function stopSync(id: WorkspaceID) {
+  aborts.get(id)?.abort()
+  aborts.delete(id)
+  connections.delete(id)
+}
+
+export function startWorkspaceSyncing(projectID: ProjectID) {
+  const spaces = Database.use((db) =>
+    db
+      .select({ workspace: WorkspaceTable })
+      .from(WorkspaceTable)
+      .innerJoin(SessionTable, eq(SessionTable.workspace_id, WorkspaceTable.id))
+      .where(eq(WorkspaceTable.project_id, projectID))
+      .all(),
+  )
+
+  for (const row of new Map(spaces.map((row) => [row.workspace.id, row.workspace])).values()) {
+    void startSync(fromRow(row))
+  }
+}
+
+export * as Workspace from "./workspace"