openAgent

1import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"
2import { $ } from "bun"
3import fs from "node:fs/promises"
4import Http from "node:http"
5import path from "node:path"
6import { NodeHttpServer } from "@effect/platform-node"
7import { Effect, Exit, Fiber, Layer, Schema } from "effect"
8import { HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
9import { eq } from "drizzle-orm"
10import { GlobalBus, type GlobalEvent } from "@/bus/global"
11import { Project } from "@/project/project"
12import { Database } from "@opencode-ai/core/database/database"
13import { ProjectV2 } from "@opencode-ai/core/project"
14import { ProjectTable } from "@opencode-ai/core/project/sql"
15import { AbsolutePath } from "@opencode-ai/core/schema"
16import { Session as SessionNs } from "@/session/session"
17import { SessionID } from "@/session/schema"
18import { SessionTable } from "@opencode-ai/core/session/sql"
19import { SessionProjector } from "@opencode-ai/core/session/projector"
20import { EventSequenceTable } from "@opencode-ai/core/event/sql"
21import { resetDatabase } from "../fixture/db"
22import { disposeAllInstances, provideTmpdirInstance, requireInstance, TestInstance } from "../fixture/fixture"
23import { testEffect } from "../lib/effect"
24import { registerAdapter } from "../../src/control-plane/adapters"
25import { WorkspaceV2 } from "@opencode-ai/core/workspace"
26import { WorkspaceTable } from "@opencode-ai/core/control-plane/workspace.sql"
27import type { Target, WorkspaceAdapter, WorkspaceInfo } from "../../src/control-plane/types"
28import * as Workspace from "../../src/control-plane/workspace"
29import { InstanceStore } from "@/project/instance-store"
30import { InstanceBootstrap } from "@/project/bootstrap"
31import { RuntimeFlags } from "@/effect/runtime-flags"
32import { Ripgrep } from "@opencode-ai/core/ripgrep"
33import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
34import { LayerNode } from "@opencode-ai/core/effect/layer-node"
35 
36const originalEnv = {
37  OPENCODE_AUTH_CONTENT: process.env.OPENCODE_AUTH_CONTENT,
38  OPENCODE_EXPERIMENTAL_WORKSPACES: process.env.OPENCODE_EXPERIMENTAL_WORKSPACES,
39  OTEL_EXPORTER_OTLP_HEADERS: process.env.OTEL_EXPORTER_OTLP_HEADERS,
40  OTEL_EXPORTER_OTLP_ENDPOINT: process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
41  OTEL_RESOURCE_ATTRIBUTES: process.env.OTEL_RESOURCE_ATTRIBUTES,
42}
43 
44const workspaceLayer = (experimentalWorkspaces: boolean) =>
45  AppNodeBuilder.build(
46    LayerNode.group([
47      Workspace.node,
48      SessionNs.node,
49      SessionProjector.node,
50      Database.node,
51      InstanceStore.node,
52      Ripgrep.node,
53    ]),
54    [
55      [RuntimeFlags.node, RuntimeFlags.layer({ experimentalWorkspaces })],
56      [
57        InstanceStore.bootstrapNode,
58        Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void })),
59      ],
60    ],
61  )
62 
63const testServerLayer = Layer.mergeAll(
64  NodeHttpServer.layer(Http.createServer, { host: "127.0.0.1", port: 0 }),
65  workspaceLayer(true),
66)
67const it = testEffect(testServerLayer)
68 
69type RecordedCreate = {
70  info: WorkspaceInfo
71  env: Record<string, string | undefined>
72  from?: WorkspaceInfo
73}
74 
75type RecordedAdapter = {
76  adapter: WorkspaceAdapter
77  calls: {
78    configure: WorkspaceInfo[]
79    create: RecordedCreate[]
80    list: number
81    remove: WorkspaceInfo[]
82    target: WorkspaceInfo[]
83  }
84}
85 
86type FetchCall = {
87  url: URL
88  method: string
89  headers: Headers
90  bodyText?: string
91  json?: unknown
92}
93 
94function unique(prefix: string) {
95  return `${prefix}-${Math.random().toString(36).slice(2)}`
96}
97 
98function restoreEnv() {
99  Object.entries(originalEnv).forEach(([key, value]) => {
100    if (value === undefined) {
101      delete process.env[key]
102      return
103    }
104    process.env[key] = value
105  })
106}
107 
108beforeEach(() => {
109  restoreEnv()
110  process.env.OPENCODE_EXPERIMENTAL_WORKSPACES = "true"
111})
112 
113afterEach(async () => {
114  mock.restore()
115  await disposeAllInstances()
116  restoreEnv()
117  await resetDatabase()
118})
119 
120async function initGitRepo(dir: string) {
121  await fs.mkdir(dir, { recursive: true })
122  await $`git init`.cwd(dir).quiet()
123  await $`git config core.fsmonitor false`.cwd(dir).quiet()
124  await $`git config commit.gpgsign false`.cwd(dir).quiet()
125  await $`git config user.email "test@opencode.test"`.cwd(dir).quiet()
126  await $`git config user.name "Test"`.cwd(dir).quiet()
127  await fs.writeFile(path.join(dir, "tracked.txt"), "base\n")
128  await $`git add tracked.txt`.cwd(dir).quiet()
129  await $`git commit -m "base"`.cwd(dir).quiet()
130}
131 
132const startWorkspaceSyncingWithFlag = (projectID: ProjectV2.ID, experimentalWorkspaces: boolean) =>
133  Effect.runPromise(
134    Workspace.use.startWorkspaceSyncing(projectID).pipe(Effect.provide(workspaceLayer(experimentalWorkspaces))),
135  )
136 
137const listWithFlag = (project: Project.Info, experimentalWorkspaces: boolean) =>
138  Effect.runPromise(Workspace.use.list(project).pipe(Effect.provide(workspaceLayer(experimentalWorkspaces))))
139 
140function captureGlobalEvents() {
141  const events: GlobalEvent[] = []
142  const handler = (event: GlobalEvent) => events.push(event)
143  GlobalBus.on("event", handler)
144  return {
145    events,
146    dispose() {
147      GlobalBus.off("event", handler)
148    },
149  }
150}
151 
152function expectExitContains(exit: Exit.Exit<unknown, unknown>, ...messages: string[]) {
153  expect(Exit.isFailure(exit)).toBe(true)
154  if (!Exit.isFailure(exit)) return
155  for (const message of messages) expect(String(exit.cause)).toContain(message)
156}
157 
158function eventuallyEffect(effect: Effect.Effect<void>, timeout = 1500) {
159  return Effect.gen(function* () {
160    const started = Date.now()
161    let last: unknown
162    while (Date.now() - started < timeout) {
163      const exit = yield* Effect.exit(effect)
164      if (exit._tag === "Success") return
165      last = exit.cause
166      yield* Effect.sleep("10 millis")
167    }
168    throw last ?? new Error("Timed out waiting for condition")
169  })
170}
171 
172function recordedAdapter(input: {
173  target: (info: WorkspaceInfo) => Target | Promise<Target>
174  configure?: (info: WorkspaceInfo) => WorkspaceInfo | Promise<WorkspaceInfo>
175  create?: (info: WorkspaceInfo, env: Record<string, string | undefined>, from?: WorkspaceInfo) => Promise<void>
176  list?: () => Omit<WorkspaceInfo, "id">[] | Promise<Omit<WorkspaceInfo, "id">[]>
177  remove?: (info: WorkspaceInfo) => Promise<void>
178}): RecordedAdapter {
179  const calls: RecordedAdapter["calls"] = {
180    configure: [],
181    create: [],
182    list: 0,
183    remove: [],
184    target: [],
185  }
186 
187  return {
188    calls,
189    adapter: {
190      name: "recorded",
191      description: "recorded",
192      configure(info) {
193        calls.configure.push(structuredClone(info))
194        return input.configure?.(info) ?? info
195      },
196      async create(info, env, from) {
197        calls.create.push({
198          info: structuredClone(info),
199          env: { ...env },
200          from: from ? structuredClone(from) : undefined,
201        })
202        await input.create?.(info, env, from)
203      },
204      ...(input.list
205        ? {
206            async list() {
207              calls.list += 1
208              return input.list?.() ?? []
209            },
210          }
211        : {}),
212      async remove(info) {
213        calls.remove.push(structuredClone(info))
214        await input.remove?.(info)
215      },
216      target(info) {
217        calls.target.push(structuredClone(info))
218        return input.target(info)
219      },
220    },
221  }
222}
223 
224function localAdapter(dir: string, input?: { createDir?: boolean; remove?: (info: WorkspaceInfo) => Promise<void> }) {
225  return recordedAdapter({
226    configure(info) {
227      return { ...info, directory: dir }
228    },
229    async create() {
230      if (input?.createDir === false) return
231      await fs.mkdir(dir, { recursive: true })
232    },
233    remove: input?.remove,
234    target() {
235      return { type: "local", directory: dir }
236    },
237  })
238}
239 
240function remoteAdapter(url: string, input?: { directory?: string | null; headers?: HeadersInit }) {
241  return recordedAdapter({
242    configure(info) {
243      return { ...info, directory: input?.directory ?? info.directory }
244    },
245    target() {
246      return { type: "remote", url, headers: input?.headers }
247    },
248  })
249}
250 
251function eventStreamResponse(events: unknown[] = [], keepOpen = true) {
252  const encoder = new TextEncoder()
253  return new Response(
254    new ReadableStream<Uint8Array>({
255      start(controller) {
256        if (keepOpen) controller.enqueue(encoder.encode(":\n\n"))
257        events.forEach((event) => controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)))
258        if (!keepOpen) controller.close()
259      },
260    }),
261    { status: 200, headers: { "content-type": "text/event-stream" } },
262  )
263}
264 
265function serverUrl() {
266  return Effect.gen(function* () {
267    return HttpServer.formatAddress((yield* HttpServer.HttpServer).address)
268  })
269}
270 
271function workspaceInfo(projectID: ProjectV2.ID, type: string, input?: Partial<Workspace.Info>): Workspace.Info {
272  return {
273    id: input?.id ?? WorkspaceV2.ID.ascending(),
274    type,
275    name: input?.name ?? unique("workspace"),
276    branch: input?.branch ?? null,
277    directory: input?.directory ?? null,
278    extra: input?.extra ?? null,
279    projectID,
280    timeUsed: input?.timeUsed ?? Date.now(),
281  }
282}
283 
284function insertWorkspace(info: Workspace.Info) {
285  return Database.Service.use(({ db }) =>
286    db
287      .insert(WorkspaceTable)
288      .values({
289        id: info.id,
290        type: info.type,
291        branch: info.branch,
292        name: info.name,
293        directory: info.directory,
294        extra: info.extra,
295        project_id: info.projectID,
296        time_used: info.timeUsed,
297      })
298      .run()
299      .pipe(Effect.orDie),
300  )
301}
302 
303function insertProject(id: ProjectV2.ID, worktree: string) {
304  return Database.Service.use(({ db }) =>
305    db
306      .insert(ProjectTable)
307      .values({
308        id,
309        worktree: AbsolutePath.make(worktree),
310        vcs: null,
311        name: null,
312        time_created: Date.now(),
313        time_updated: Date.now(),
314        sandboxes: [],
315      })
316      .run()
317      .pipe(Effect.orDie),
318  )
319}
320 
321function attachSessionToWorkspace(sessionID: SessionID, workspaceID: WorkspaceV2.ID) {
322  return Database.Service.use(({ db }) =>
323    db
324      .update(SessionTable)
325      .set({ workspace_id: workspaceID })
326      .where(eq(SessionTable.id, sessionID))
327      .run()
328      .pipe(Effect.orDie),
329  )
330}
331 
332function sessionSequence(sessionID: SessionID) {
333  return Database.Service.use(({ db }) =>
334    db
335      .select({ seq: EventSequenceTable.seq })
336      .from(EventSequenceTable)
337      .where(eq(EventSequenceTable.aggregate_id, sessionID))
338      .get()
339      .pipe(
340        Effect.orDie,
341        Effect.map((row) => row?.seq),
342      ),
343  )
344}
345 
346function sessionSequenceOwner(sessionID: SessionID) {
347  return Database.Service.use(({ db }) =>
348    db
349      .select({ ownerID: EventSequenceTable.owner_id })
350      .from(EventSequenceTable)
351      .where(eq(EventSequenceTable.aggregate_id, sessionID))
352      .get()
353      .pipe(
354        Effect.orDie,
355        Effect.map((row) => row?.ownerID),
356      ),
357  )
358}
359 
360describe("workspace schemas and exports", () => {
361  test("keeps the historical event type names", () => {
362    expect(Workspace.Event.Ready.type).toBe("workspace.ready")
363    expect(Workspace.Event.Failed.type).toBe("workspace.failed")
364    expect(Workspace.Event.Status.type).toBe("workspace.status")
365  })
366 
367  test("validates create input with workspace id, project id, branch, type, and extra", () => {
368    const input = {
369      id: WorkspaceV2.ID.ascending("wrk_schema_create"),
370      type: "worktree",
371      branch: "feature/schema",
372      projectID: ProjectV2.ID.make("project-schema"),
373      extra: { nested: true },
374    }
375 
376    const decode = Schema.decodeUnknownSync(Workspace.CreateInput)
377    expect(decode(input)).toEqual(input)
378    expect(() => decode({ ...input, id: 1 })).toThrow()
379    expect(() => decode({ ...input, branch: 1 })).toThrow()
380  })
381})
382 
383describe("workspace CRUD", () => {
384  it.instance(
385    "get returns undefined for a missing workspace",
386    () =>
387      Effect.gen(function* () {
388        const workspace = yield* Workspace.Service
389        expect(yield* workspace.get(WorkspaceV2.ID.ascending("wrk_missing_get"))).toBeUndefined()
390      }),
391    { git: true },
392  )
393 
394  it.instance(
395    "list maps database rows, filters by project, and sorts by id",
396    () =>
397      Effect.gen(function* () {
398        const instance = yield* requireInstance
399        const workspace = yield* Workspace.Service
400        const otherProjectID = ProjectV2.ID.make("project-other")
401        yield* insertProject(otherProjectID, "/tmp/other")
402        const a = workspaceInfo(instance.project.id, "manual", {
403          id: WorkspaceV2.ID.ascending("wrk_a_list"),
404          branch: "a",
405          directory: "/a",
406          extra: { a: true },
407        })
408        const b = workspaceInfo(instance.project.id, "manual", {
409          id: WorkspaceV2.ID.ascending("wrk_b_list"),
410          branch: "b",
411          directory: "/b",
412          extra: ["b"],
413        })
414        const other = workspaceInfo(otherProjectID, "manual", { id: WorkspaceV2.ID.ascending("wrk_c_list") })
415        yield* insertWorkspace(b)
416        yield* insertWorkspace(other)
417        yield* insertWorkspace(a)
418 
419        expect(yield* workspace.list(instance.project)).toEqual([a, b])
420      }),
421    { git: true },
422  )
423 
424  it.instance(
425    "list is disabled by the experimental workspace flag",
426    () =>
427      Effect.gen(function* () {
428        const instance = yield* requireInstance
429        yield* insertWorkspace(workspaceInfo(instance.project.id, "manual"))
430 
431        expect(yield* Effect.promise(() => listWithFlag(instance.project, false))).toEqual([])
432      }),
433    { git: true },
434  )
435 
436  it.instance(
437    "create configures, persists, creates, starts local sync, and passes environment",
438    () =>
439      Effect.gen(function* () {
440        const instance = yield* requireInstance
441        const workspace = yield* Workspace.Service
442        process.env.OPENCODE_AUTH_CONTENT = JSON.stringify({ test: { type: "api", key: "secret" } })
443        process.env.OTEL_EXPORTER_OTLP_HEADERS = "authorization=otel"
444        process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "https://otel.test"
445        process.env.OTEL_RESOURCE_ATTRIBUTES = "service.name=opencode-test"
446 
447        const workspaceID = WorkspaceV2.ID.ascending("wrk_create_local")
448        const type = unique("create-local")
449        const targetDir = path.join(instance.directory, "created-local")
450        const recorded = recordedAdapter({
451          configure(info) {
452            return {
453              ...info,
454              branch: "configured-branch",
455              name: "Configured Name",
456              directory: targetDir,
457              extra: { configured: true },
458            }
459          },
460          async create() {
461            await fs.mkdir(targetDir, { recursive: true })
462          },
463          target() {
464            return { type: "local", directory: targetDir }
465          },
466        })
467        registerAdapter(instance.project.id, type, recorded.adapter)
468 
469        const info = yield* workspace.create({
470          id: workspaceID,
471          type,
472          branch: null,
473          projectID: instance.project.id,
474          extra: null,
475        })
476 
477        expect(info).toEqual({
478          id: workspaceID,
479          type,
480          branch: "configured-branch",
481          name: "Configured Name",
482          directory: targetDir,
483          extra: { configured: true },
484          projectID: instance.project.id,
485          timeUsed: info.timeUsed,
486        })
487        expect(yield* workspace.get(workspaceID)).toEqual(info)
488        expect(yield* workspace.list(instance.project)).toEqual([info])
489        expect(recorded.calls.configure).toHaveLength(1)
490        expect(recorded.calls.configure[0]).toMatchObject({ id: workspaceID, type, directory: null })
491        expect(recorded.calls.create).toHaveLength(1)
492        expect(recorded.calls.create[0].info).toEqual({
493          id: workspaceID,
494          type,
495          branch: "configured-branch",
496          name: "Configured Name",
497          directory: targetDir,
498          extra: { configured: true },
499          projectID: instance.project.id,
500        })
501        expect(JSON.parse(recorded.calls.create[0].env.OPENCODE_AUTH_CONTENT ?? "{}")).toEqual({
502          test: { type: "api", key: "secret" },
503        })
504        expect(recorded.calls.create[0].env.OPENCODE_WORKSPACE_ID).toBe(workspaceID)
505        expect(recorded.calls.create[0].env.OPENCODE_EXPERIMENTAL_WORKSPACES).toBe("true")
506        expect(recorded.calls.create[0].env.OTEL_EXPORTER_OTLP_HEADERS).toBe("authorization=otel")
507        expect(recorded.calls.create[0].env.OTEL_EXPORTER_OTLP_ENDPOINT).toBe("https://otel.test")
508        expect(recorded.calls.create[0].env.OTEL_RESOURCE_ATTRIBUTES).toBe("service.name=opencode-test")
509        expect((yield* workspace.status()).find((item) => item.workspaceID === workspaceID)?.status).toBe("connected")
510 
511        yield* workspace.remove(workspaceID)
512        expect((yield* workspace.status()).find((item) => item.workspaceID === workspaceID)?.status).toBeUndefined()
513      }),
514    { git: true },
515  )
516 
517  it.instance(
518    "create propagates configure failures and does not insert a workspace",
519    () =>
520      Effect.gen(function* () {
521        const instance = yield* requireInstance
522        const workspace = yield* Workspace.Service
523        const type = unique("configure-failure")
524        registerAdapter(
525          instance.project.id,
526          type,
527          recordedAdapter({
528            configure() {
529              throw new Error("configure exploded")
530            },
531            target() {
532              return { type: "local", directory: "/unused" }
533            },
534          }).adapter,
535        )
536 
537        expectExitContains(
538          yield* Effect.exit(workspace.create({ type, branch: null, projectID: instance.project.id, extra: null })),
539          "configure exploded",
540        )
541        expect(yield* workspace.list(instance.project)).toEqual([])
542      }),
543    { git: true },
544  )
545 
546  it.instance(
547    "create leaves the inserted row when adapter create fails",
548    () =>
549      Effect.gen(function* () {
550        const instance = yield* requireInstance
551        const workspace = yield* Workspace.Service
552        const type = unique("create-failure")
553        const recorded = recordedAdapter({
554          async create() {
555            throw new Error("create exploded")
556          },
557          target() {
558            return { type: "local", directory: "/unused" }
559          },
560        })
561        registerAdapter(instance.project.id, type, recorded.adapter)
562 
563        expectExitContains(
564          yield* Effect.exit(
565            workspace.create({ type, branch: "branch", projectID: instance.project.id, extra: { x: 1 } }),
566          ),
567          "create exploded",
568        )
569 
570        const rows = yield* workspace.list(instance.project)
571        expect(rows).toHaveLength(1)
572        expect(rows[0]).toMatchObject({ type, branch: "branch", extra: { x: 1 } })
573        expect(recorded.calls.target).toHaveLength(0)
574        yield* workspace.remove(rows[0].id)
575      }),
576    { git: true },
577  )
578 
579  it.instance(
580    "create returns after a local workspace reports error",
581    () =>
582      Effect.gen(function* () {
583        const instance = yield* requireInstance
584        const workspace = yield* Workspace.Service
585        const type = unique("local-error")
586        const missing = path.join(instance.directory, "missing-local-target")
587        const recorded = localAdapter(missing, { createDir: false })
588        registerAdapter(instance.project.id, type, recorded.adapter)
589 
590        const info = yield* workspace.create({ type, branch: null, projectID: instance.project.id, extra: null })
591 
592        expect(info.directory).toBe(missing)
593        expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBe("error")
594        yield* workspace.remove(info.id)
595      }),
596    { git: true },
597  )
598 
599  it.instance(
600    "syncList registers adapter-listed workspaces that are missing by name",
601    () =>
602      Effect.gen(function* () {
603        const instance = yield* requireInstance
604        const workspace = yield* Workspace.Service
605        const type = unique("list-sync")
606        const existing = workspaceInfo(instance.project.id, type, {
607          id: WorkspaceV2.ID.ascending("wrk_list_sync_existing"),
608          name: "existing",
609          directory: path.join(instance.directory, "existing"),
610        })
611        yield* insertWorkspace(existing)
612 
613        const discovered = {
614          type,
615          name: "discovered",
616          branch: "feature/discovered",
617          directory: path.join(instance.directory, "discovered"),
618          extra: { source: "adapter" },
619          projectID: instance.project.id,
620        }
621        const recorded = recordedAdapter({
622          list() {
623            return [
624              {
625                type,
626                name: existing.name,
627                branch: "ignored",
628                directory: path.join(instance.directory, "ignored"),
629                extra: null,
630                projectID: instance.project.id,
631              },
632              discovered,
633            ]
634          },
635          target(info) {
636            return { type: "local", directory: info.directory ?? instance.directory }
637          },
638        })
639        registerAdapter(instance.project.id, type, recorded.adapter)
640 
641        yield* workspace.syncList(instance.project)
642        const synced = (yield* workspace.list(instance.project)).filter((item) => item.name === discovered.name)
643 
644        expect(synced).toHaveLength(1)
645        expect(synced[0]).toMatchObject(discovered)
646        expect(synced[0]?.id).toStartWith("wrk_")
647        expect(yield* workspace.list(instance.project)).toEqual(expect.arrayContaining([existing, synced[0]]))
648        expect(recorded.calls.list).toBe(1)
649        expect(recorded.calls.configure).toHaveLength(0)
650        expect(recorded.calls.create).toHaveLength(0)
651        expect(recorded.calls.target).toHaveLength(1)
652      }),
653    { git: true },
654  )
655 
656  it.instance(
657    "syncList calls every registered adapter with a list method",
658    () =>
659      Effect.gen(function* () {
660        const instance = yield* requireInstance
661        const workspace = yield* Workspace.Service
662        const typeA = unique("list-sync-a")
663        const typeB = unique("list-sync-b")
664        const adapterA = recordedAdapter({
665          list() {
666            return [
667              {
668                type: typeA,
669                name: "adapter-a",
670                branch: null,
671                directory: path.join(instance.directory, "adapter-a"),
672                extra: null,
673                projectID: instance.project.id,
674              },
675            ]
676          },
677          target(info) {
678            return { type: "local", directory: info.directory ?? instance.directory }
679          },
680        })
681        const adapterB = recordedAdapter({
682          list() {
683            return [
684              {
685                type: typeB,
686                name: "adapter-b",
687                branch: null,
688                directory: path.join(instance.directory, "adapter-b"),
689                extra: null,
690                projectID: instance.project.id,
691              },
692            ]
693          },
694          target(info) {
695            return { type: "local", directory: info.directory ?? instance.directory }
696          },
697        })
698        const noList = recordedAdapter({
699          target() {
700            return { type: "local", directory: instance.directory }
701          },
702        })
703        registerAdapter(instance.project.id, typeA, adapterA.adapter)
704        registerAdapter(instance.project.id, typeB, adapterB.adapter)
705        registerAdapter(instance.project.id, unique("list-sync-none"), noList.adapter)
706 
707        yield* workspace.syncList(instance.project)
708        const synced = yield* workspace.list(instance.project)
709 
710        expect(
711          synced
712            .filter((item) => item.type === typeA || item.type === typeB)
713            .map((item) => item.name)
714            .toSorted(),
715        ).toEqual(["adapter-a", "adapter-b"])
716        expect(adapterA.calls.list).toBe(1)
717        expect(adapterB.calls.list).toBe(1)
718        expect(noList.calls.list).toBe(0)
719      }),
720    { git: true },
721  )
722 
723  it.live("remote create connects to routed event and history endpoints", () => {
724    const calls: FetchCall[] = []
725    return Effect.gen(function* () {
726      yield* HttpServer.serveEffect()(
727        Effect.gen(function* () {
728          const req = yield* HttpServerRequest.HttpServerRequest
729          const bodyText = yield* req.text
730          const call = {
731            url: new URL(req.url, "http://localhost"),
732            method: req.method,
733            headers: new Headers(req.headers),
734            bodyText,
735            json: bodyText ? JSON.parse(bodyText) : undefined,
736          }
737          calls.push(call)
738          if (call.url.pathname === "/base/global/event")
739            return HttpServerResponse.fromWeb(eventStreamResponse([], false))
740          if (call.url.pathname === "/base/sync/history") return yield* HttpServerResponse.json([])
741          return HttpServerResponse.text("unexpected", { status: 500 })
742        }),
743      )
744      const url = yield* serverUrl()
745      yield* provideTmpdirInstance(
746        (dir) =>
747          Effect.gen(function* () {
748            const workspace = yield* Workspace.Service
749            const instance = yield* requireInstance
750            const type = unique("remote-create")
751            const recorded = remoteAdapter(`${url}/base/?ignored=1#hash`, { directory: dir })
752            registerAdapter(instance.project.id, type, recorded.adapter)
753 
754            const info = yield* workspace.create({ type, branch: null, projectID: instance.project.id, extra: null })
755 
756            expect(
757              calls.map((call) => `${call.method} ${call.url.pathname}${call.url.search}${call.url.hash}`),
758            ).toEqual(["GET /base/global/event", "POST /base/sync/history"])
759            expect(calls[1].json).toEqual({})
760            expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBe("connected")
761            expect(yield* workspace.isSyncing(info.id)).toBe(true)
762 
763            yield* workspace.remove(info.id)
764            expect(yield* workspace.isSyncing(info.id)).toBe(false)
765            expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined()
766          }),
767        { git: true },
768      )
769    })
770  })
771 
772  it.instance(
773    "remove returns undefined for a missing workspace",
774    () =>
775      Effect.gen(function* () {
776        const workspace = yield* Workspace.Service
777        expect(yield* workspace.remove(WorkspaceV2.ID.ascending("wrk_missing_remove"))).toBeUndefined()
778      }),
779    { git: true },
780  )
781 
782  it.instance(
783    "remove deletes the workspace, associated sessions, adapter resources, and status",
784    () => {
785      return Effect.gen(function* () {
786        const { directory: dir } = yield* TestInstance
787        const instance = yield* requireInstance
788        const workspace = yield* Workspace.Service
789        const sessionSvc = yield* SessionNs.Service
790        const type = unique("remove-local")
791        const recorded = localAdapter(path.join(dir, "remove-local"))
792        registerAdapter(instance.project.id, type, recorded.adapter)
793        const info = yield* workspace.create({ type, branch: null, projectID: instance.project.id, extra: null })
794        const one = yield* sessionSvc.create({})
795        const two = yield* sessionSvc.create({})
796        yield* attachSessionToWorkspace(one.id, info.id)
797        yield* attachSessionToWorkspace(two.id, info.id)
798 
799        const removed = yield* workspace.remove(info.id)
800 
801        expect(removed).toEqual(info)
802        expect(yield* workspace.get(info.id)).toBeUndefined()
803        expect(recorded.calls.remove).toEqual([info])
804        expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined()
805        const { db } = yield* Database.Service
806        expect(
807          yield* db
808            .select({ id: SessionTable.id })
809            .from(SessionTable)
810            .where(eq(SessionTable.workspace_id, info.id))
811            .all()
812            .pipe(Effect.orDie),
813        ).toEqual([])
814      })
815    },
816    { git: true },
817  )
818 
819  it.instance(
820    "remove still deletes the row when the adapter cannot remove resources",
821    () =>
822      Effect.gen(function* () {
823        const instance = yield* requireInstance
824        const workspace = yield* Workspace.Service
825        const type = unique("remove-throws")
826        const info = workspaceInfo(instance.project.id, type, { id: WorkspaceV2.ID.ascending("wrk_remove_throws") })
827        registerAdapter(
828          instance.project.id,
829          type,
830          recordedAdapter({
831            async remove() {
832              throw new Error("remove exploded")
833            },
834            target() {
835              return { type: "local", directory: "/unused" }
836            },
837          }).adapter,
838        )
839        yield* insertWorkspace(info)
840 
841        expect(yield* workspace.remove(info.id)).toEqual(info)
842        expect(yield* workspace.get(info.id)).toBeUndefined()
843      }),
844    { git: true },
845  )
846 
847  it.instance(
848    "sessionWarp moves a session into a local workspace and claims ownership",
849    () => {
850      return Effect.gen(function* () {
851        const { directory: dir } = yield* TestInstance
852        const instance = yield* requireInstance
853        const workspace = yield* Workspace.Service
854        const sessionSvc = yield* SessionNs.Service
855        const previousType = unique("warp-prev-local")
856        const targetType = unique("warp-target-local")
857        const previous = workspaceInfo(instance.project.id, previousType)
858        const target = workspaceInfo(instance.project.id, targetType)
859        yield* insertWorkspace(previous)
860        yield* insertWorkspace(target)
861        registerAdapter(instance.project.id, previousType, localAdapter(path.join(dir, "warp-prev-local")).adapter)
862        registerAdapter(instance.project.id, targetType, localAdapter(path.join(dir, "warp-target-local")).adapter)
863        const session = yield* sessionSvc.create({})
864        yield* attachSessionToWorkspace(session.id, previous.id)
865 
866        yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id })
867 
868        const { db } = yield* Database.Service
869        expect(
870          (yield* db
871            .select({ workspaceID: SessionTable.workspace_id })
872            .from(SessionTable)
873            .where(eq(SessionTable.id, session.id))
874            .get()
875            .pipe(Effect.orDie))?.workspaceID,
876        ).toBe(target.id)
877        expect(yield* sessionSequenceOwner(session.id)).toBe(target.id)
878      })
879    },
880    { git: true },
881  )
882 
883  it.instance(
884    "sessionWarp applies source workspace patch to local target workspace",
885    () => {
886      return Effect.gen(function* () {
887        const { directory: dir } = yield* TestInstance
888        const instance = yield* requireInstance
889        const workspace = yield* Workspace.Service
890        const sessionSvc = yield* SessionNs.Service
891        const previousType = unique("warp-patch-prev-local")
892        const targetType = unique("warp-patch-target-local")
893        const previousDir = path.join(dir, "warp-patch-prev-local")
894        const targetDir = path.join(dir, "warp-patch-target-local")
895        yield* Effect.promise(() => initGitRepo(previousDir))
896        yield* Effect.promise(() => initGitRepo(targetDir))
897        yield* Effect.promise(() => fs.writeFile(path.join(previousDir, "tracked.txt"), "changed\n"))
898        yield* Effect.promise(() => fs.writeFile(path.join(previousDir, "new.txt"), "new\n"))
899 
900        const previous = workspaceInfo(instance.project.id, previousType)
901        const target = workspaceInfo(instance.project.id, targetType)
902        yield* insertWorkspace(previous)
903        yield* insertWorkspace(target)
904        registerAdapter(instance.project.id, previousType, localAdapter(previousDir, { createDir: false }).adapter)
905        registerAdapter(instance.project.id, targetType, localAdapter(targetDir, { createDir: false }).adapter)
906        const session = yield* sessionSvc.create({})
907        yield* attachSessionToWorkspace(session.id, previous.id)
908 
909        yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id, copyChanges: true })
910 
911        expect(yield* Effect.promise(() => fs.readFile(path.join(targetDir, "tracked.txt"), "utf8"))).toBe("changed\n")
912        expect(yield* Effect.promise(() => fs.readFile(path.join(targetDir, "new.txt"), "utf8"))).toBe("new\n")
913      })
914    },
915    { git: true },
916  )
917 
918  it.instance(
919    "sessionWarp detaches a session to the local project and claims project ownership",
920    () => {
921      return Effect.gen(function* () {
922        const { directory: dir } = yield* TestInstance
923        const instance = yield* requireInstance
924        const workspace = yield* Workspace.Service
925        const sessionSvc = yield* SessionNs.Service
926        const previousType = unique("warp-detach-local")
927        const previous = workspaceInfo(instance.project.id, previousType)
928        yield* insertWorkspace(previous)
929        registerAdapter(instance.project.id, previousType, localAdapter(path.join(dir, "warp-detach-local")).adapter)
930        const session = yield* sessionSvc.create({})
931        yield* attachSessionToWorkspace(session.id, previous.id)
932 
933        yield* workspace.sessionWarp({ workspaceID: null, sessionID: session.id })
934 
935        const { db } = yield* Database.Service
936        expect(
937          (yield* db
938            .select({ workspaceID: SessionTable.workspace_id })
939            .from(SessionTable)
940            .where(eq(SessionTable.id, session.id))
941            .get()
942            .pipe(Effect.orDie))?.workspaceID,
943        ).toBeNull()
944        expect(yield* sessionSequenceOwner(session.id)).toBe(instance.project.id)
945      })
946    },
947    { git: true },
948  )
949 
950  const itCrossInstance = process.platform === "win32" ? it.instance.skip : it.instance
951  itCrossInstance(
952    "sessionWarp detaches to the source project when invoked from a workspace instance",
953    () =>
954      Effect.gen(function* () {
955        const instance = yield* requireInstance
956        const projectID = instance.project.id
957        const workspace = yield* Workspace.Service
958        const sessionSvc = yield* SessionNs.Service
959        const previousType = unique("warp-detach-workspace-instance")
960        const previous = workspaceInfo(projectID, previousType)
961        yield* insertWorkspace(previous)
962        const session = yield* sessionSvc.create({})
963        yield* attachSessionToWorkspace(session.id, previous.id)
964 
965        const workspaceProjectID = yield* provideTmpdirInstance(
966          (workspaceDir) =>
967            Effect.gen(function* () {
968              registerAdapter(projectID, previousType, localAdapter(workspaceDir, { createDir: false }).adapter)
969              const workspaceCtx = yield* requireInstance
970              expect(workspaceCtx.project.id).not.toBe(projectID)
971              yield* workspace.sessionWarp({ workspaceID: null, sessionID: session.id })
972              return workspaceCtx.project.id
973            }),
974          { git: true },
975        )
976 
977        const { db } = yield* Database.Service
978        expect(
979          (yield* db
980            .select({ workspaceID: SessionTable.workspace_id })
981            .from(SessionTable)
982            .where(eq(SessionTable.id, session.id))
983            .get()
984            .pipe(Effect.orDie))?.workspaceID,
985        ).toBeNull()
986        expect(yield* sessionSequenceOwner(session.id)).toBe(projectID)
987        expect(yield* sessionSequenceOwner(session.id)).not.toBe(workspaceProjectID)
988      }),
989    { git: true },
990  )
991 
992  it.live("sessionWarp syncs previous remote history, replays it, steals, and claims the sequence", () => {
993    const calls: FetchCall[] = []
994    let historySessionID: SessionID | undefined
995    let historySession: SessionNs.Info | undefined
996    let historyNextSeq = 0
997    return Effect.gen(function* () {
998      yield* HttpServer.serveEffect()(
999        Effect.gen(function* () {
1000          const req = yield* HttpServerRequest.HttpServerRequest
1001          const bodyText = yield* req.text
1002          const call = {
1003            url: new URL(req.url, "http://localhost"),
1004            method: req.method,
1005            headers: new Headers(req.headers),
1006            bodyText,
1007            json: bodyText ? JSON.parse(bodyText) : undefined,
1008          }
1009          calls.push(call)
1010          if (call.url.pathname === "/warp-source/sync/history") {
1011            return yield* HttpServerResponse.json([
1012              {
1013                id: `evt_${unique("warp-source-history")}`,
1014                aggregate_id: historySessionID!,
1015                seq: historyNextSeq,
1016                type: "session.updated.1",
1017                data: { sessionID: historySessionID!, info: historySession! },
1018              },
1019            ])
1020          }
1021          if (call.url.pathname === "/warp-source/vcs/diff/raw") return HttpServerResponse.text("remote patch")
1022          if (call.url.pathname === "/warp-target/sync/replay")
1023            return yield* HttpServerResponse.json({ sessionID: "ok" })
1024          if (call.url.pathname === "/warp-target/sync/steal")
1025            return yield* HttpServerResponse.json({ sessionID: "ok" })
1026          if (call.url.pathname === "/warp-target/vcs/apply") return yield* HttpServerResponse.json({ applied: true })
1027          return HttpServerResponse.text("unexpected", { status: 500 })
1028        }),
1029      )
1030      const url = yield* serverUrl()
1031      yield* provideTmpdirInstance(
1032        () =>
1033          Effect.gen(function* () {
1034            const workspace = yield* Workspace.Service
1035            const sessionSvc = yield* SessionNs.Service
1036            const instance = yield* requireInstance
1037            const previousType = unique("warp-remote-source")
1038            const targetType = unique("warp-remote-target")
1039            const previous = workspaceInfo(instance.project.id, previousType)
1040            const target = workspaceInfo(instance.project.id, targetType, { directory: "remote-target-dir" })
1041            yield* insertWorkspace(previous)
1042            yield* insertWorkspace(target)
1043            registerAdapter(instance.project.id, previousType, remoteAdapter(`${url}/warp-source`).adapter)
1044            registerAdapter(instance.project.id, targetType, remoteAdapter(`${url}/warp-target`).adapter)
1045            const session = yield* sessionSvc.create({})
1046            yield* attachSessionToWorkspace(session.id, previous.id)
1047            historySessionID = session.id
1048            historySession = { ...session, workspaceID: previous.id, title: "from source history" }
1049            historyNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1
1050 
1051            yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id, copyChanges: true })
1052 
1053            expect(calls.map((call) => `${call.method} ${call.url.pathname}`)).toEqual([
1054              "POST /warp-source/sync/history",
1055              "GET /warp-source/vcs/diff/raw",
1056              "POST /warp-target/vcs/apply",
1057              "POST /warp-target/sync/replay",
1058              "POST /warp-target/sync/steal",
1059            ])
1060            expect(calls[0].json).toEqual({ [session.id]: historyNextSeq - 1 })
1061            expect(calls[2].json).toEqual({ patch: "remote patch" })
1062            expect(calls[3].json).toMatchObject({
1063              directory: "remote-target-dir",
1064              events: [
1065                {
1066                  aggregateID: session.id,
1067                  seq: 0,
1068                  type: "session.created.1",
1069                },
1070                {
1071                  aggregateID: session.id,
1072                  seq: historyNextSeq,
1073                  type: "session.updated.1",
1074                },
1075              ],
1076            })
1077            expect(calls[4].json).toEqual({ sessionID: session.id })
1078            expect((yield* sessionSvc.get(session.id)).title).toBe("from source history")
1079            expect(yield* sessionSequenceOwner(session.id)).toBe(target.id)
1080          }),
1081        { git: true },
1082      )
1083    })
1084  })
1085})
1086 
1087describe("workspace sync state", () => {
1088  it.instance(
1089    "startWorkspaceSyncing is disabled by the experimental workspace flag",
1090    () =>
1091      Effect.gen(function* () {
1092        const { directory: dir } = yield* TestInstance
1093        const instance = yield* requireInstance
1094        const workspace = yield* Workspace.Service
1095        const sessionSvc = yield* SessionNs.Service
1096        const type = unique("flag-disabled")
1097        const info = workspaceInfo(instance.project.id, type)
1098        const session = yield* sessionSvc.create({})
1099        yield* attachSessionToWorkspace(session.id, info.id)
1100        yield* insertWorkspace(info)
1101        registerAdapter(instance.project.id, type, localAdapter(path.join(dir, "flag-disabled")).adapter)
1102 
1103        yield* Effect.promise(() => startWorkspaceSyncingWithFlag(instance.project.id, false))
1104        yield* Effect.sleep("25 millis")
1105 
1106        expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined()
1107      }),
1108    { git: true },
1109  )
1110 
1111  it.instance(
1112    "startWorkspaceSyncing starts all workspaces",
1113    () =>
1114      Effect.gen(function* () {
1115        const { directory: dir } = yield* TestInstance
1116        const instance = yield* requireInstance
1117        const workspace = yield* Workspace.Service
1118        const projectID = instance.project.id
1119        const firstType = unique("first")
1120        const secondType = unique("second")
1121        const first = workspaceInfo(projectID, firstType)
1122        const second = workspaceInfo(projectID, secondType)
1123        yield* Effect.promise(() => fs.mkdir(path.join(dir, "first"), { recursive: true }))
1124        yield* Effect.promise(() => fs.mkdir(path.join(dir, "second"), { recursive: true }))
1125        yield* insertWorkspace(first)
1126        yield* insertWorkspace(second)
1127        registerAdapter(projectID, firstType, localAdapter(path.join(dir, "first")).adapter)
1128        registerAdapter(projectID, secondType, localAdapter(path.join(dir, "second")).adapter)
1129        yield* Effect.addFinalizer(() =>
1130          Effect.all([workspace.remove(first.id), workspace.remove(second.id)], { discard: true }).pipe(Effect.ignore),
1131        )
1132 
1133        yield* workspace.startWorkspaceSyncing(projectID)
1134 
1135        yield* eventuallyEffect(
1136          Effect.gen(function* () {
1137            const status = yield* workspace.status()
1138            expect(status.find((item) => item.workspaceID === first.id)?.status).toBe("connected")
1139            expect(status.find((item) => item.workspaceID === second.id)?.status).toBe("connected")
1140          }),
1141        )
1142      }),
1143    { git: true },
1144  )
1145 
1146  it.instance(
1147    "local start reports error when the target directory is missing",
1148    () =>
1149      Effect.gen(function* () {
1150        const { directory: dir } = yield* TestInstance
1151        const instance = yield* requireInstance
1152        const workspace = yield* Workspace.Service
1153        const sessionSvc = yield* SessionNs.Service
1154        const type = unique("missing-local")
1155        const info = workspaceInfo(instance.project.id, type)
1156        yield* insertWorkspace(info)
1157        registerAdapter(
1158          instance.project.id,
1159          type,
1160          localAdapter(path.join(dir, "missing-target"), { createDir: false }).adapter,
1161        )
1162        yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
1163 
1164        yield* workspace.startWorkspaceSyncing(instance.project.id)
1165 
1166        yield* eventuallyEffect(
1167          Effect.gen(function* () {
1168            const status = yield* workspace.status()
1169            expect(status.find((item) => item.workspaceID === info.id)?.status).toBe("error")
1170          }),
1171        )
1172        expect(yield* workspace.isSyncing(info.id)).toBe(false)
1173        yield* workspace.remove(info.id)
1174      }),
1175    { git: true },
1176  )
1177 
1178  it.instance(
1179    "duplicate local status updates are suppressed",
1180    () =>
1181      Effect.gen(function* () {
1182        const { directory: dir } = yield* TestInstance
1183        const instance = yield* requireInstance
1184        const workspace = yield* Workspace.Service
1185        const sessionSvc = yield* SessionNs.Service
1186        const captured = captureGlobalEvents()
1187        yield* Effect.addFinalizer(() => Effect.sync(() => captured.dispose()))
1188        const type = unique("dedupe-local")
1189        const info = workspaceInfo(instance.project.id, type)
1190        const target = path.join(dir, "dedupe-local")
1191        yield* Effect.promise(() => fs.mkdir(target, { recursive: true }))
1192        yield* insertWorkspace(info)
1193        registerAdapter(instance.project.id, type, localAdapter(target).adapter)
1194        yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
1195 
1196        yield* workspace.startWorkspaceSyncing(instance.project.id)
1197        yield* workspace.startWorkspaceSyncing(instance.project.id)
1198 
1199        yield* eventuallyEffect(
1200          Effect.gen(function* () {
1201            const status = yield* workspace.status()
1202            expect(status.find((item) => item.workspaceID === info.id)?.status).toBe("connected")
1203          }),
1204        )
1205        expect(
1206          captured.events.filter(
1207            (event) => event.workspace === info.id && event.payload.type === Workspace.Event.Status.type,
1208          ),
1209        ).toHaveLength(1)
1210        yield* workspace.remove(info.id)
1211      }),
1212    { git: true },
1213  )
1214 
1215  it.live("remote start emits disconnected, connecting, and connected then refuses duplicate listeners", () => {
1216    const calls: FetchCall[] = []
1217    return Effect.gen(function* () {
1218      yield* HttpServer.serveEffect()(
1219        Effect.gen(function* () {
1220          const req = yield* HttpServerRequest.HttpServerRequest
1221          const bodyText = yield* req.text
1222          const call = {
1223            url: new URL(req.url, "http://localhost"),
1224            method: req.method,
1225            headers: new Headers(req.headers),
1226            bodyText,
1227            json: bodyText ? JSON.parse(bodyText) : undefined,
1228          }
1229          calls.push(call)
1230          if (call.url.pathname === "/sync/global/event") return HttpServerResponse.fromWeb(eventStreamResponse())
1231          if (call.url.pathname === "/sync/sync/history") return HttpServerResponse.fromWeb(Response.json([]))
1232          return HttpServerResponse.text("unexpected", { status: 500 })
1233        }),
1234      )
1235      const url = yield* serverUrl()
1236      yield* provideTmpdirInstance(
1237        () =>
1238          Effect.gen(function* () {
1239            const workspace = yield* Workspace.Service
1240            const sessionSvc = yield* SessionNs.Service
1241            const instance = yield* requireInstance
1242            const captured = captureGlobalEvents()
1243            try {
1244              const type = unique("remote-start")
1245              const info = workspaceInfo(instance.project.id, type)
1246              yield* insertWorkspace(info)
1247              registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sync`).adapter)
1248              yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
1249 
1250              yield* workspace.startWorkspaceSyncing(instance.project.id)
1251              yield* eventuallyEffect(
1252                Effect.gen(function* () {
1253                  expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBe(
1254                    "connected",
1255                  )
1256                }),
1257              )
1258              yield* workspace.startWorkspaceSyncing(instance.project.id)
1259              yield* Effect.sleep("25 millis")
1260 
1261              expect(
1262                captured.events
1263                  .filter((event) => event.workspace === info.id && event.payload.type === Workspace.Event.Status.type)
1264                  .map((event) => event.payload.properties.status),
1265              ).toEqual(["disconnected", "connecting", "connected"])
1266              expect(calls.filter((call) => call.url.pathname === "/sync/global/event")).toHaveLength(1)
1267              expect(calls.filter((call) => call.url.pathname === "/sync/sync/history")).toHaveLength(1)
1268              expect(yield* workspace.isSyncing(info.id)).toBe(true)
1269 
1270              yield* workspace.remove(info.id)
1271              expect(yield* workspace.isSyncing(info.id)).toBe(false)
1272            } finally {
1273              captured.dispose()
1274            }
1275          }),
1276        { git: true },
1277      )
1278    })
1279  })
1280 
1281  it.live("remote connection HTTP failures set error and clear syncing", () =>
1282    Effect.gen(function* () {
1283      yield* HttpServer.serveEffect()(
1284        Effect.gen(function* () {
1285          const req = yield* HttpServerRequest.HttpServerRequest
1286          if (new URL(req.url, "http://localhost").pathname === "/failed/global/event")
1287            return HttpServerResponse.text("nope", { status: 503 })
1288          return HttpServerResponse.fromWeb(Response.json([]))
1289        }),
1290      )
1291      const url = yield* serverUrl()
1292      yield* provideTmpdirInstance(
1293        () =>
1294          Effect.gen(function* () {
1295            const workspace = yield* Workspace.Service
1296            const sessionSvc = yield* SessionNs.Service
1297            const instance = yield* requireInstance
1298            const type = unique("remote-connect-fail")
1299            const info = workspaceInfo(instance.project.id, type)
1300            yield* insertWorkspace(info)
1301            registerAdapter(instance.project.id, type, remoteAdapter(`${url}/failed`).adapter)
1302            yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
1303 
1304            yield* workspace.startWorkspaceSyncing(instance.project.id)
1305 
1306            yield* eventuallyEffect(
1307              Effect.gen(function* () {
1308                expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBe("error")
1309              }),
1310            )
1311            expect(yield* workspace.isSyncing(info.id)).toBe(false)
1312            yield* workspace.remove(info.id)
1313          }),
1314        { git: true },
1315      )
1316    }),
1317  )
1318 
1319  it.live("remote history HTTP failures set error", () =>
1320    Effect.gen(function* () {
1321      yield* HttpServer.serveEffect()(
1322        Effect.gen(function* () {
1323          const req = yield* HttpServerRequest.HttpServerRequest
1324          const url = new URL(req.url, "http://localhost")
1325          if (url.pathname === "/history-failed/global/event")
1326            return HttpServerResponse.fromWeb(eventStreamResponse([], false))
1327          if (url.pathname === "/history-failed/sync/history")
1328            return HttpServerResponse.text("history failed", { status: 500 })
1329          return HttpServerResponse.fromWeb(Response.json([]))
1330        }),
1331      )
1332      const url = yield* serverUrl()
1333      yield* provideTmpdirInstance(
1334        () =>
1335          Effect.gen(function* () {
1336            const workspace = yield* Workspace.Service
1337            const sessionSvc = yield* SessionNs.Service
1338            const instance = yield* requireInstance
1339            const type = unique("remote-history-fail")
1340            const info = workspaceInfo(instance.project.id, type)
1341            yield* insertWorkspace(info)
1342            registerAdapter(instance.project.id, type, remoteAdapter(`${url}/history-failed`).adapter)
1343            yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
1344 
1345            yield* workspace.startWorkspaceSyncing(instance.project.id)
1346 
1347            yield* eventuallyEffect(
1348              Effect.gen(function* () {
1349                expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBe("error")
1350              }),
1351            )
1352            expect(yield* workspace.isSyncing(info.id)).toBe(false)
1353            yield* workspace.remove(info.id)
1354          }),
1355        { git: true },
1356      )
1357    }),
1358  )
1359 
1360  it.live("sync history sends the local sequence fence and replays returned events in workspace context", () => {
1361    const historyBodies: unknown[] = []
1362    let historySessionID: SessionID | undefined
1363    let historySession: SessionNs.Info | undefined
1364    let historyNextSeq = 0
1365    return Effect.gen(function* () {
1366      yield* HttpServer.serveEffect()(
1367        Effect.gen(function* () {
1368          const req = yield* HttpServerRequest.HttpServerRequest
1369          const bodyText = yield* req.text
1370          const url = new URL(req.url, "http://localhost")
1371          if (url.pathname === "/history/global/event") return HttpServerResponse.fromWeb(eventStreamResponse())
1372          if (url.pathname === "/history/sync/history") {
1373            historyBodies.push(bodyText ? JSON.parse(bodyText) : undefined)
1374            return HttpServerResponse.fromWeb(
1375              Response.json([
1376                {
1377                  id: `evt_${unique("history")}`,
1378                  aggregate_id: historySessionID!,
1379                  seq: historyNextSeq,
1380                  type: "session.updated.1",
1381                  data: { sessionID: historySessionID!, info: historySession! },
1382                },
1383              ]),
1384            )
1385          }
1386          return HttpServerResponse.text("unexpected", { status: 500 })
1387        }),
1388      )
1389      const url = yield* serverUrl()
1390      yield* provideTmpdirInstance(
1391        () =>
1392          Effect.gen(function* () {
1393            const workspace = yield* Workspace.Service
1394            const sessionSvc = yield* SessionNs.Service
1395            const instance = yield* requireInstance
1396            const captured = captureGlobalEvents()
1397            try {
1398              const type = unique("history-replay")
1399              const info = workspaceInfo(instance.project.id, type)
1400              yield* insertWorkspace(info)
1401              registerAdapter(instance.project.id, type, remoteAdapter(`${url}/history`).adapter)
1402              const session = yield* sessionSvc.create({ title: "before history" })
1403              yield* attachSessionToWorkspace(session.id, info.id)
1404              historySessionID = session.id
1405              historySession = { ...session, workspaceID: info.id, title: "from history" }
1406              historyNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1
1407 
1408              yield* workspace.startWorkspaceSyncing(instance.project.id)
1409 
1410              yield* eventuallyEffect(
1411                Effect.gen(function* () {
1412                  expect((yield* sessionSvc.get(session.id).pipe(Effect.orDie)).title).toBe("from history")
1413                }),
1414              )
1415              expect(historyBodies).toEqual([{ [session.id]: historyNextSeq - 1 }])
1416              expect(
1417                captured.events.some(
1418                  (event) =>
1419                    event.workspace === info.id &&
1420                    event.payload.type === "session.updated" &&
1421                    event.payload.properties.sessionID === session.id &&
1422                    event.payload.properties.info.title === "from history",
1423                ),
1424              ).toBe(true)
1425              yield* workspace.remove(info.id)
1426            } finally {
1427              captured.dispose()
1428            }
1429          }),
1430        { git: true },
1431      )
1432    })
1433  })
1434 
1435  it.live("SSE forwards non-heartbeat events and ignores heartbeats", () =>
1436    Effect.gen(function* () {
1437      yield* HttpServer.serveEffect()(
1438        Effect.gen(function* () {
1439          const req = yield* HttpServerRequest.HttpServerRequest
1440          const url = new URL(req.url, "http://localhost")
1441          if (url.pathname === "/sse-forward/global/event")
1442            return HttpServerResponse.fromWeb(
1443              eventStreamResponse(
1444                [
1445                  { directory: "remote-dir", project: "remote-project", payload: { type: "server.heartbeat" } },
1446                  {
1447                    directory: "remote-dir",
1448                    project: "remote-project",
1449                    payload: { type: "custom.remote", properties: { ok: true } },
1450                  },
1451                ],
1452                false,
1453              ),
1454            )
1455          if (url.pathname === "/sse-forward/sync/history") return HttpServerResponse.fromWeb(Response.json([]))
1456          return HttpServerResponse.text("unexpected", { status: 500 })
1457        }),
1458      )
1459      const url = yield* serverUrl()
1460      yield* provideTmpdirInstance(
1461        () =>
1462          Effect.gen(function* () {
1463            const workspace = yield* Workspace.Service
1464            const sessionSvc = yield* SessionNs.Service
1465            const instance = yield* requireInstance
1466            const captured = captureGlobalEvents()
1467            try {
1468              const type = unique("sse-forward")
1469              const info = workspaceInfo(instance.project.id, type)
1470              yield* insertWorkspace(info)
1471              registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sse-forward`).adapter)
1472              yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
1473 
1474              yield* workspace.startWorkspaceSyncing(instance.project.id)
1475 
1476              yield* eventuallyEffect(
1477                Effect.sync(() =>
1478                  expect(
1479                    captured.events.some(
1480                      (event) => event.workspace === info.id && event.payload.type === "custom.remote",
1481                    ),
1482                  ).toBe(true),
1483                ),
1484              )
1485              expect(
1486                captured.events.some(
1487                  (event) => event.workspace === info.id && event.payload.type === "server.heartbeat",
1488                ),
1489              ).toBe(false)
1490              expect(
1491                captured.events.find((event) => event.workspace === info.id && event.payload.type === "custom.remote"),
1492              ).toMatchObject({
1493                directory: "remote-dir",
1494                project: "remote-project",
1495                payload: { properties: { ok: true } },
1496              })
1497              yield* workspace.remove(info.id)
1498            } finally {
1499              captured.dispose()
1500            }
1501          }),
1502        { git: true },
1503      )
1504    }),
1505  )
1506 
1507  it.live("SSE sync events are replayed and forwarded", () => {
1508    let sseSessionID: SessionID | undefined
1509    let sseSession: SessionNs.Info | undefined
1510    let sseNextSeq = 0
1511    return Effect.gen(function* () {
1512      yield* HttpServer.serveEffect()(
1513        Effect.gen(function* () {
1514          const req = yield* HttpServerRequest.HttpServerRequest
1515          const url = new URL(req.url, "http://localhost")
1516          if (url.pathname === "/sse-sync/global/event")
1517            return HttpServerResponse.fromWeb(
1518              eventStreamResponse(
1519                [
1520                  {
1521                    directory: "remote-dir",
1522                    project: "remote-project",
1523                    payload: {
1524                      type: "sync",
1525                      syncEvent: {
1526                        id: `evt_${unique("sse")}`,
1527                        aggregateID: sseSessionID!,
1528                        seq: sseNextSeq,
1529                        type: "session.updated.1",
1530                        data: { sessionID: sseSessionID!, info: sseSession! },
1531                      },
1532                    },
1533                  },
1534                ],
1535                false,
1536              ),
1537            )
1538          if (url.pathname === "/sse-sync/sync/history") return HttpServerResponse.fromWeb(Response.json([]))
1539          return HttpServerResponse.text("unexpected", { status: 500 })
1540        }),
1541      )
1542      const url = yield* serverUrl()
1543      yield* provideTmpdirInstance(
1544        () =>
1545          Effect.gen(function* () {
1546            const workspace = yield* Workspace.Service
1547            const sessionSvc = yield* SessionNs.Service
1548            const instance = yield* requireInstance
1549            const captured = captureGlobalEvents()
1550            try {
1551              const type = unique("sse-sync")
1552              const info = workspaceInfo(instance.project.id, type)
1553              yield* insertWorkspace(info)
1554              registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sse-sync`).adapter)
1555              const session = yield* sessionSvc.create({ title: "before sse" })
1556              yield* attachSessionToWorkspace(session.id, info.id)
1557              sseSessionID = session.id
1558              sseSession = { ...session, workspaceID: info.id, title: "from sse" }
1559              sseNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1
1560 
1561              yield* workspace.startWorkspaceSyncing(instance.project.id)
1562 
1563              yield* eventuallyEffect(
1564                Effect.gen(function* () {
1565                  expect((yield* sessionSvc.get(session.id).pipe(Effect.orDie)).title).toBe("from sse")
1566                }),
1567              )
1568              expect(
1569                captured.events.some(
1570                  (event) =>
1571                    event.workspace === info.id &&
1572                    event.payload.type === "sync" &&
1573                    event.payload.syncEvent.seq === sseNextSeq,
1574                ),
1575              ).toBe(true)
1576              yield* workspace.remove(info.id)
1577            } finally {
1578              captured.dispose()
1579            }
1580          }),
1581        { git: true },
1582      )
1583    })
1584  })
1585})
1586 
1587describe("workspace waitForSync", () => {
1588  it.instance(
1589    "returns immediately for an empty fence",
1590    () =>
1591      Effect.gen(function* () {
1592        const workspace = yield* Workspace.Service
1593        expect(yield* workspace.waitForSync(WorkspaceV2.ID.ascending("wrk_wait_empty"), {})).toBeUndefined()
1594      }),
1595    { git: true },
1596  )
1597 
1598  it.instance(
1599    "returns immediately when the stored sequence already satisfies the fence",
1600    () =>
1601      Effect.gen(function* () {
1602        const workspace = yield* Workspace.Service
1603        const sessionID = SessionID.descending("ses_wait_done")
1604        const { db } = yield* Database.Service
1605        yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run().pipe(Effect.orDie)
1606 
1607        expect(
1608          yield* workspace.waitForSync(WorkspaceV2.ID.ascending("wrk_wait_done"), { [sessionID]: 4 }),
1609        ).toBeUndefined()
1610        expect(
1611          yield* workspace.waitForSync(WorkspaceV2.ID.ascending("wrk_wait_done_2"), { [sessionID]: 3 }),
1612        ).toBeUndefined()
1613      }),
1614    { git: true },
1615  )
1616 
1617  it.instance(
1618    "waits until the database reaches the requested sequence and a workspace event arrives",
1619    () =>
1620      Effect.gen(function* () {
1621        const workspace = yield* Workspace.Service
1622        const workspaceID = WorkspaceV2.ID.ascending("wrk_wait_event")
1623        const sessionID = SessionID.descending("ses_wait_event")
1624        const { db } = yield* Database.Service
1625        yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run().pipe(Effect.orDie)
1626 
1627        yield* Effect.all(
1628          [
1629            workspace.waitForSync(workspaceID, { [sessionID]: 2 }),
1630            Effect.gen(function* () {
1631              yield* Effect.sleep("10 millis")
1632              yield* db
1633                .update(EventSequenceTable)
1634                .set({ seq: 2 })
1635                .where(eq(EventSequenceTable.aggregate_id, sessionID))
1636                .run()
1637                .pipe(Effect.orDie)
1638              GlobalBus.emit("event", { workspace: workspaceID, payload: { type: "anything" } })
1639            }),
1640          ],
1641          { concurrency: "unbounded" },
1642        )
1643      }),
1644    { git: true },
1645  )
1646 
1647  it.instance(
1648    "a sync event for a different workspace can also release the fence",
1649    () =>
1650      Effect.gen(function* () {
1651        const workspace = yield* Workspace.Service
1652        const workspaceID = WorkspaceV2.ID.ascending("wrk_wait_sync_any")
1653        const sessionID = SessionID.descending("ses_wait_sync_any")
1654        const { db } = yield* Database.Service
1655        yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run().pipe(Effect.orDie)
1656 
1657        yield* Effect.all(
1658          [
1659            workspace.waitForSync(workspaceID, { [sessionID]: 1 }),
1660            Effect.gen(function* () {
1661              yield* Effect.sleep("10 millis")
1662              yield* db
1663                .update(EventSequenceTable)
1664                .set({ seq: 1 })
1665                .where(eq(EventSequenceTable.aggregate_id, sessionID))
1666                .run()
1667                .pipe(Effect.orDie)
1668              GlobalBus.emit("event", {
1669                workspace: WorkspaceV2.ID.ascending("wrk_other_workspace"),
1670                payload: { type: "sync" },
1671              })
1672            }),
1673          ],
1674          { concurrency: "unbounded" },
1675        )
1676      }),
1677    { git: true },
1678  )
1679 
1680  it.instance(
1681    "rejects with the abort reason when aborted",
1682    () =>
1683      Effect.gen(function* () {
1684        const workspace = yield* Workspace.Service
1685        const abort = new AbortController()
1686        const reason = new Error("caller aborted")
1687        const fiber = yield* Effect.forkChild(
1688          workspace.waitForSync(
1689            WorkspaceV2.ID.ascending("wrk_wait_abort"),
1690            { [SessionID.descending("ses_wait_abort")]: 1 },
1691            abort.signal,
1692          ),
1693        )
1694        abort.abort(reason)
1695 
1696        expectExitContains(yield* Fiber.await(fiber), "WorkspaceSyncAbortedError", reason.message)
1697      }),
1698    { git: true },
1699  )
1700 
1701  it.instance(
1702    "times out with the requested fence in the error message",
1703    () =>
1704      Effect.gen(function* () {
1705        const workspace = yield* Workspace.Service
1706        const sessionID = SessionID.descending("ses_wait_timeout")
1707        expectExitContains(
1708          yield* Effect.exit(
1709            workspace.waitForSync(WorkspaceV2.ID.ascending("wrk_wait_timeout"), { [sessionID]: 1 }, undefined, 25),
1710          ),
1711          `Timed out waiting for sync fence: {"${sessionID}":1}`,
1712        )
1713      }),
1714    { git: true },
1715    7000,
1716  )
1717})
1718