feat(httpapi): bridge sync routes (#24484)

packages/opencode/src/server/routes/instance/httpapi/mcp.ts

@@ -139,9 +139,10 @@ export const mcpHandlers = Layer.unwrap(
     })
 
     const add = Effect.fn("McpHttpApi.add")(function* (ctx: { payload: typeof AddPayload.Type }) {
-      const payload = Schema.decodeUnknownSync(AddPayload)(ctx.payload)
-      const result = (yield* mcp.add(payload.name, payload.config)).status
-      return Schema.decodeUnknownSync(StatusMap)("status" in result ? { [payload.name]: result } : result)
+      const result = (yield* mcp.add(ctx.payload.name, ctx.payload.config)).status
+      return yield* Schema.decodeUnknownEffect(StatusMap)("status" in result ? { [ctx.payload.name]: result } : result).pipe(
+        Effect.mapError(() => new HttpApiError.BadRequest({})),
+      )
     })
 
     const authStart = Effect.fn("McpHttpApi.authStart")(function* (ctx: { params: { name: string } }) {

packages/opencode/src/server/routes/instance/httpapi/server.ts

@@ -18,6 +18,7 @@ import { PermissionApi, permissionHandlers } from "./permission"
 import { ProjectApi, projectHandlers } from "./project"
 import { ProviderApi, providerHandlers } from "./provider"
 import { QuestionApi, questionHandlers } from "./question"
+import { SyncApi, syncHandlers } from "./sync"
 import { WorkspaceApi, workspaceHandlers } from "./workspace"
 import { disposeMiddleware } from "./lifecycle"
 import { memoMap } from "@opencode-ai/core/effect/memo-map"
@@ -73,6 +74,7 @@ export const routes = Layer.mergeAll(
   HttpApiBuilder.layer(QuestionApi).pipe(Layer.provide(questionHandlers)),
   HttpApiBuilder.layer(PermissionApi).pipe(Layer.provide(permissionHandlers)),
   HttpApiBuilder.layer(ProviderApi).pipe(Layer.provide(providerHandlers)),
+  HttpApiBuilder.layer(SyncApi).pipe(Layer.provide(syncHandlers)),
   HttpApiBuilder.layer(WorkspaceApi).pipe(Layer.provide(workspaceHandlers)),
 ).pipe(
   Layer.provide(authorizationLayer),

packages/opencode/src/server/routes/instance/httpapi/sync.ts

@@ -0,0 +1,130 @@
+import { startWorkspaceSyncing } from "@/control-plane/workspace"
+import * as InstanceState from "@/effect/instance-state"
+import { Database, asc, and, eq, lte, not, or } from "@/storage"
+import { SyncEvent } from "@/sync"
+import { EventTable } from "@/sync/event.sql"
+import { Effect, Layer, Schema } from "effect"
+import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
+import { Authorization } from "./auth"
+
+const root = "/sync"
+const ReplayEvent = Schema.Struct({
+  id: Schema.String,
+  aggregateID: Schema.String,
+  seq: Schema.Number,
+  type: Schema.String,
+  data: Schema.Record(Schema.String, Schema.Unknown),
+}).annotate({ identifier: "SyncReplayEvent" })
+const ReplayPayload = Schema.Struct({
+  directory: Schema.String,
+  events: Schema.NonEmptyArray(ReplayEvent),
+}).annotate({ identifier: "SyncReplayInput" })
+const ReplayResponse = Schema.Struct({
+  sessionID: Schema.String,
+}).annotate({ identifier: "SyncReplayResponse" })
+const HistoryPayload = Schema.Record(Schema.String, Schema.Number)
+const HistoryEvent = Schema.Struct({
+  id: Schema.String,
+  aggregate_id: Schema.String,
+  seq: Schema.Number,
+  type: Schema.String,
+  data: Schema.Record(Schema.String, Schema.Unknown),
+}).annotate({ identifier: "SyncHistoryEvent" })
+
+export const SyncPaths = {
+  start: `${root}/start`,
+  replay: `${root}/replay`,
+  history: `${root}/history`,
+} as const
+
+export const SyncApi = HttpApi.make("sync")
+  .add(
+    HttpApiGroup.make("sync")
+      .add(
+        HttpApiEndpoint.post("start", SyncPaths.start, {
+          success: Schema.Boolean,
+        }).annotateMerge(
+          OpenApi.annotations({
+            identifier: "sync.start",
+            summary: "Start workspace sync",
+            description: "Start sync loops for workspaces in the current project that have active sessions.",
+          }),
+        ),
+        HttpApiEndpoint.post("replay", SyncPaths.replay, {
+          payload: ReplayPayload,
+          success: ReplayResponse,
+        }).annotateMerge(
+          OpenApi.annotations({
+            identifier: "sync.replay",
+            summary: "Replay sync events",
+            description: "Validate and replay a complete sync event history.",
+          }),
+        ),
+        HttpApiEndpoint.post("history", SyncPaths.history, {
+          payload: HistoryPayload,
+          success: Schema.Array(HistoryEvent),
+        }).annotateMerge(
+          OpenApi.annotations({
+            identifier: "sync.history.list",
+            summary: "List sync events",
+            description:
+              "List sync events for all aggregates. Keys are aggregate IDs the client already knows about, values are the last known sequence ID. Events with seq > value are returned for those aggregates. Aggregates not listed in the input get their full history.",
+          }),
+        ),
+      )
+      .annotateMerge(
+        OpenApi.annotations({
+          title: "sync",
+          description: "Experimental HttpApi sync routes.",
+        }),
+      )
+      .middleware(Authorization),
+  )
+  .annotateMerge(
+    OpenApi.annotations({
+      title: "opencode experimental HttpApi",
+      version: "0.0.1",
+      description: "Experimental HttpApi surface for selected instance routes.",
+    }),
+  )
+
+export const syncHandlers = Layer.unwrap(
+  Effect.gen(function* () {
+    const start = Effect.fn("SyncHttpApi.start")(function* () {
+      startWorkspaceSyncing((yield* InstanceState.context).project.id)
+      return true
+    })
+
+    const replay = Effect.fn("SyncHttpApi.replay")(function* (ctx: { payload: typeof ReplayPayload.Type }) {
+      const events: SyncEvent.SerializedEvent[] = ctx.payload.events.map((event) => ({
+        id: event.id,
+        aggregateID: event.aggregateID,
+        seq: event.seq,
+        type: event.type,
+        data: { ...event.data },
+      }))
+      SyncEvent.replayAll(events)
+      return { sessionID: events[0].aggregateID }
+    })
+
+    const history = Effect.fn("SyncHttpApi.history")(function* (ctx: { payload: typeof HistoryPayload.Type }) {
+      const exclude = Object.entries(ctx.payload)
+      return Database.use((db) =>
+        db
+          .select()
+          .from(EventTable)
+          .where(
+            exclude.length > 0
+              ? not(or(...exclude.map(([id, seq]) => and(eq(EventTable.aggregate_id, id), lte(EventTable.seq, seq))))!)
+              : undefined,
+          )
+          .orderBy(asc(EventTable.seq))
+          .all(),
+      )
+    })
+
+    return HttpApiBuilder.group(SyncApi, "sync", (handlers) =>
+      handlers.handle("start", start).handle("replay", replay).handle("history", history),
+    )
+  }),
+)

packages/opencode/src/server/routes/instance/httpapi/workspace.ts

@@ -123,7 +123,7 @@ export const workspaceHandlers = Layer.unwrap(
       return yield* Effect.promise(() =>
         Instance.restore(instance, () =>
           Workspace.create({
-            ...Schema.decodeUnknownSync(CreatePayload)(ctx.payload),
+            ...ctx.payload,
             projectID: instance.project.id,
           }),
         ),

packages/opencode/src/server/routes/instance/index.ts

@@ -20,6 +20,7 @@ import { ExperimentalPaths } from "./httpapi/experimental"
 import { FilePaths } from "./httpapi/file"
 import { InstancePaths } from "./httpapi/instance"
 import { McpPaths } from "./httpapi/mcp"
+import { SyncPaths } from "./httpapi/sync"
 import { ProjectRoutes } from "./project"
 import { SessionRoutes } from "./session"
 import { PtyRoutes } from "./pty"
@@ -89,6 +90,9 @@ export const InstanceRoutes = (upgrade: UpgradeWebSocket): Hono => {
     app.delete(McpPaths.auth, (c) => handler(c.req.raw, context))
     app.post(McpPaths.connect, (c) => handler(c.req.raw, context))
     app.post(McpPaths.disconnect, (c) => handler(c.req.raw, context))
+    app.post(SyncPaths.start, (c) => handler(c.req.raw, context))
+    app.post(SyncPaths.replay, (c) => handler(c.req.raw, context))
+    app.post(SyncPaths.history, (c) => handler(c.req.raw, context))
   }
 
   return app