lmin/zmo-cli
1
0
Fork 0
Code Issues Pull Requests Actions 6 Packages Projects Releases Wiki Activity

feat(core): session warping (#25768)

Browse Source
This commit is contained in:
James Long
2026-05-04 21:28:38 -04:00
committed by GitHub
parent 84afd2bef8
commit 22a4a9df8b
25 changed files with 4032 additions and 1095 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
packages/opencode/src/control-plane/workspace.ts
View File

@@ -1,10 +1,11 @@
import { Context, Effect, FiberMap, Layer, Schema, Stream } from "effect"
import { Context, Effect, FiberMap, Iterable, Layer, Schema, Stream } from "effect"
import { FetchHttpClient, HttpBody, HttpClient, HttpClientError, HttpClientRequest } from "effect/unstable/http"
import { Database } from "@/storage/db"
import { asc } from "drizzle-orm"
import { eq } from "drizzle-orm"
import { inArray } from "drizzle-orm"
import { Project } from "@/project/project"
import { Instance } from "@/project/instance"
import { BusEvent } from "@/bus/bus-event"
import { GlobalBus } from "@/bus/global"
import { Auth } from "@/auth"
@@ -20,6 +21,7 @@ import { getAdapter } from "./adapters"
import { type WorkspaceInfo, WorkspaceInfo as WorkspaceInfoSchema } from "./types"
import { WorkspaceID } from "./schema"
import { Session } from "@/session/session"
import { SessionPrompt } from "@/session/prompt"
import { SessionTable } from "@/session/session.sql"
import { SessionID } from "@/session/schema"
import { errorData } from "@/util/error"
@@ -38,13 +40,6 @@ export const ConnectionStatus = Schema.Struct({
})
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",
@@ -58,7 +53,6 @@ export const Event = {
message: Schema.String,
}),
),
Restore: BusEvent.define("workspace.restore", Restore),
Status: BusEvent.define("workspace.status", ConnectionStatus),
}
@@ -84,15 +78,15 @@ export const CreateInput = Schema.Struct({
type: Info.fields.type,
branch: Info.fields.branch,
projectID: ProjectID,
extra: Info.fields.extra,
extra: Schema.optional(Info.fields.extra),
}).pipe(withStatics((s) => ({ zod: effectZod(s), zodObject: zodObject(s) })))
export type CreateInput = Schema.Schema.Type<typeof CreateInput>
export const SessionRestoreInput = Schema.Struct({
workspaceID: WorkspaceID,
export const SessionWarpInput = Schema.Struct({
workspaceID: Schema.NullOr(WorkspaceID),
sessionID: SessionID,
}).pipe(withStatics((s) => ({ zod: effectZod(s), zodObject: zodObject(s) })))
export type SessionRestoreInput = Schema.Schema.Type<typeof SessionRestoreInput>
export type SessionWarpInput = Schema.Schema.Type<typeof SessionWarpInput>
export class SyncHttpError extends Schema.TaggedErrorClass<SyncHttpError>()("WorkspaceSyncHttpError", {
message: Schema.String,
@@ -116,8 +110,8 @@ export class SessionEventsNotFoundError extends Schema.TaggedErrorClass<SessionE
},
) {}
export class SessionRestoreHttpError extends Schema.TaggedErrorClass<SessionRestoreHttpError>()(
"WorkspaceSessionRestoreHttpError",
export class SessionWarpHttpError extends Schema.TaggedErrorClass<SessionWarpHttpError>()(
"WorkspaceSessionWarpHttpError",
{
message: Schema.String,
workspaceID: WorkspaceID,
@@ -138,17 +132,17 @@ export class SyncAbortedError extends Schema.TaggedErrorClass<SyncAbortedError>(
}) {}
type CreateError = Auth.AuthError
type SessionRestoreError =
type SessionWarpError =
| WorkspaceNotFoundError
| SessionEventsNotFoundError
| SessionRestoreHttpError
| SessionWarpHttpError
| HttpClientError.HttpClientError
type WaitForSyncError = SyncTimeoutError | SyncAbortedError
type SyncLoopError = SyncHttpError | HttpClientError.HttpClientError
export interface Interface {
readonly create: (input: CreateInput) => Effect.Effect<Info, CreateError>
readonly sessionRestore: (input: SessionRestoreInput) => Effect.Effect<{ total: number }, SessionRestoreError>
readonly sessionWarp: (input: SessionWarpInput) => Effect.Effect<void, SessionWarpError>
readonly list: (project: Project.Info) => Effect.Effect<Info[]>
readonly get: (id: WorkspaceID) => Effect.Effect<Info | undefined>
readonly remove: (id: WorkspaceID) => Effect.Effect<Info | undefined>
@@ -169,6 +163,7 @@ export const layer = Layer.effect(
Effect.gen(function* () {
const auth = yield* Auth.Service
const session = yield* Session.Service
const prompt = yield* SessionPrompt.Service
const http = yield* HttpClient.HttpClient
const sync = yield* SyncEvent.Service
const connections = new Map<WorkspaceID, ConnectionStatus>()
@@ -461,7 +456,7 @@ export const layer = Layer.effect(
const id = WorkspaceID.ascending(input.id)
const adapter = getAdapter(input.projectID, input.type)
const config = yield* EffectBridge.fromPromise(() =>
adapter.configure({ ...input, id, name: Slug.create(), directory: null }),
adapter.configure({ ...input, id, name: Slug.create(), directory: null, extra: input.extra ?? null }),
)
const info: Info = {
@@ -518,29 +513,93 @@ export const layer = Layer.effect(
return info
})
const sessionRestore = Effect.fn("Workspace.sessionRestore")(function* (input: SessionRestoreInput) {
const sessionWarp = Effect.fn("Workspace.sessionWarp")(function* (input: SessionWarpInput) {
return yield* Effect.gen(function* () {
log.info("session restore requested", {
log.info("session warp requested", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
})
const space = yield* get(input.workspaceID)
const current = yield* db((db) =>
db
.select({ workspaceID: SessionTable.workspace_id })
.from(SessionTable)
.where(eq(SessionTable.id, input.sessionID))
.get(),
)
if (current?.workspaceID) {
const previous = yield* get(current.workspaceID)
if (previous) {
const adapter = getAdapter(previous.projectID, previous.type)
const target = yield* EffectBridge.fromPromise(() => adapter.target(previous))
if (target.type === "remote") {
yield* syncHistory(previous, target.url, target.headers).pipe(
Effect.catch((error) =>
Effect.sync(() => {
log.warn("session warp final source sync failed", {
workspaceID: previous.id,
sessionID: input.sessionID,
error: errorData(error),
})
}),
),
)
} else {
yield* prompt.cancel(input.sessionID)
}
// "claim" this session so any future events coming from
// the old workspace are ignored
SyncEvent.claim(input.sessionID, input.workspaceID ?? Instance.project.id)
}
}
if (input.workspaceID === null) {
yield* Effect.sync(() =>
SyncEvent.run(Session.Event.Updated, {
sessionID: input.sessionID,
info: {
workspaceID: null,
},
}),
)
log.info("session warp complete", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
target: "local",
})
return
}
const workspaceID = input.workspaceID
const space = yield* get(workspaceID)
if (!space)
return yield* new WorkspaceNotFoundError({
message: `Workspace not found: ${input.workspaceID}`,
workspaceID: input.workspaceID,
message: `Workspace not found: ${workspaceID}`,
workspaceID,
})
const adapter = getAdapter(space.projectID, space.type)
const target = yield* EffectBridge.fromPromise(() => adapter.target(space))
yield* sync.run(Session.Event.Updated, {
sessionID: input.sessionID,
info: {
if (target.type === "local") {
yield* sync.run(Session.Event.Updated, {
sessionID: input.sessionID,
info: {
workspaceID: input.workspaceID,
},
})
log.info("session warp complete", {
workspaceID: input.workspaceID,
},
})
sessionID: input.sessionID,
target: target.directory,
})
return
}
const rows = yield* db((db) =>
db
@@ -562,130 +621,95 @@ export const layer = Layer.effect(
sessionID: input.sessionID,
})
const size = 10
// TODO: look into using effect APIs to process this in chunks
const sets = Array.from({ length: Math.ceil(rows.length / size) }, (_, i) =>
rows.slice(i * size, (i + 1) * size),
)
const total = sets.length
const batches = Iterable.chunksOf(rows, 10)
const total = Iterable.size(batches)
log.info("session restore prepared", {
log.info("session warp 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,
target: String(route(target.url, "/sync/replay")),
events: rows.length,
batches: total,
first: rows[0]?.seq,
last: rows.at(-1)?.seq,
})
yield* Effect.sync(() =>
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") {
yield* sync.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 res = yield* http.execute(
HttpClientRequest.post(url, {
headers: new Headers(target.headers),
body: HttpBody.jsonUnsafe({
directory: space.directory ?? "",
events,
yield* Effect.forEach(
batches,
(events, i) =>
Effect.gen(function* () {
const response = yield* http.execute(
HttpClientRequest.post(route(target.url, "/sync/replay"), {
headers: new Headers(target.headers),
body: HttpBody.jsonUnsafe({
directory: space.directory ?? "",
events,
}),
}),
}),
)
)
if (res.status < 200 || res.status >= 300) {
const body = yield* res.text
log.error("session restore batch failed", {
if (response.status < 200 || response.status >= 300) {
const body = yield* response.text
log.error("session warp batch failed", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
step: i + 1,
total,
status: response.status,
body,
})
return yield* new SessionWarpHttpError({
message: `Failed to warp session ${input.sessionID} into workspace ${workspaceID}: HTTP ${response.status} ${body}`,
workspaceID,
sessionID: input.sessionID,
status: response.status,
body,
})
}
log.info("session warp batch posted", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
step: i + 1,
total,
status: res.status,
body,
status: response.status,
})
return yield* new SessionRestoreHttpError({
message: `Failed to replay session ${input.sessionID} into workspace ${input.workspaceID}: HTTP ${res.status} ${body}`,
workspaceID: input.workspaceID,
sessionID: input.sessionID,
status: res.status,
body,
})
}
log.info("session restore batch posted", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
step: i + 1,
total,
status: res.status,
})
}
yield* Effect.sync(() =>
GlobalBus.emit("event", {
directory: "global",
workspace: input.workspaceID,
payload: {
type: Event.Restore.type,
properties: {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
total,
step: i + 1,
},
},
}),
)
{ discard: true },
)
const response = yield* http.execute(
HttpClientRequest.post(route(target.url, "/sync/steal"), {
headers: new Headers(target.headers),
body: HttpBody.jsonUnsafe({ sessionID: input.sessionID }),
}),
)
if (response.status < 200 || response.status >= 300) {
const body = yield* response.text
log.error("session warp steal failed", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
status: response.status,
body,
})
return yield* new SessionWarpHttpError({
message: `Failed to steal session ${input.sessionID} into workspace ${workspaceID}: HTTP ${response.status} ${body}`,
workspaceID,
sessionID: input.sessionID,
status: response.status,
body,
})
}
log.info("session restore complete", {
log.info("session warp complete", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
batches: total,
})
return { total }
}).pipe(
Effect.tapError((err) =>
Effect.sync(() =>
log.error("session restore failed", {
log.error("session warp failed", {
workspaceID: input.workspaceID,
sessionID: input.sessionID,
error: errorData(err),
@@ -814,7 +838,7 @@ export const layer = Layer.effect(
return Service.of({
create,
sessionRestore,
sessionWarp,
list,
get,
remove,
@@ -830,6 +854,7 @@ export const defaultLayer = layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(SessionPrompt.defaultLayer),
Layer.provide(FetchHttpClient.layer),
)