diff --git a/packages/opencode/src/cli/cmd/tui/component/dialog-session-list.tsx b/packages/opencode/src/cli/cmd/tui/component/dialog-session-list.tsx index d177f7ff24..cd5834ab70 100644 --- a/packages/opencode/src/cli/cmd/tui/component/dialog-session-list.tsx +++ b/packages/opencode/src/cli/cmd/tui/component/dialog-session-list.tsx @@ -12,7 +12,7 @@ import { Flag } from "@opencode-ai/core/flag/flag" import { DialogSessionRename } from "./dialog-session-rename" import { createDebouncedSignal } from "../util/signal" import { useToast } from "../ui/toast" -import { DialogWorkspaceCreate, restoreWorkspaceSession } from "./dialog-workspace-create" +import { DialogWorkspaceSelect, type WorkspaceSelection, warpWorkspaceSession } from "./dialog-workspace-create" import { Spinner } from "./spinner" import { errorMessage } from "@/util/error" import { DialogSessionDeleteFailed } from "./dialog-session-delete-failed" @@ -45,6 +45,36 @@ export function DialogSessionList() { function recover(session: NonNullable[number]>) { const workspace = project.workspace.get(session.workspaceID!) const list = () => dialog.replace(() => ) + const warp = async (selection: WorkspaceSelection) => { + if (selection.type === "none") return + const workspaceID = await (async () => { + if (selection.type === "existing") return selection.workspaceID + const result = await sdk.client.experimental.workspace + .create({ type: selection.workspaceType, branch: null }) + .catch(() => undefined) + const workspace = result?.data + if (!workspace) { + toast.show({ + message: `Failed to create workspace: ${errorMessage(result?.error ?? "no response")}`, + variant: "error", + }) + return + } + await project.workspace.sync() + return workspace.id + })() + if (!workspaceID) return + await warpWorkspaceSession({ + dialog, + sdk, + sync, + project, + toast, + workspaceID, + sessionID: session.id, + done: list, + }) + } dialog.replace(() => ( { dialog.replace(() => ( - - restoreWorkspaceSession({ - dialog, - sdk, - sync, - project, - toast, - workspaceID, - sessionID: session.id, - done: list, - }) - } + { + void warp(selection) + }} /> )) return false diff --git a/packages/opencode/src/cli/cmd/tui/component/dialog-workspace-create.tsx b/packages/opencode/src/cli/cmd/tui/component/dialog-workspace-create.tsx index 28ba1b6991..5aeb8ee1c2 100644 --- a/packages/opencode/src/cli/cmd/tui/component/dialog-workspace-create.tsx +++ b/packages/opencode/src/cli/cmd/tui/component/dialog-workspace-create.tsx @@ -7,7 +7,6 @@ import { createMemo, createSignal, onMount } from "solid-js" import { errorMessage } from "@/util/error" import { useSDK } from "../context/sdk" import { useToast } from "../ui/toast" -import { WorkspaceLabel } from "./workspace-label" type Adaptor = { type: string @@ -34,7 +33,7 @@ export type WorkspaceSelection = type WorkspaceSelectValue = WorkspaceSelection | { type: "existing-list" } | { type: "loading" } type ExistingWorkspaceSelectValue = { workspace: Workspace } -export async function restoreWorkspaceSession(input: { +export async function warpWorkspaceSession(input: { dialog: ReturnType sdk: ReturnType sync: ReturnType @@ -43,91 +42,38 @@ export async function restoreWorkspaceSession(input: { workspaceID: string sessionID: string done?: () => void -}) { + showSuccessToast?: boolean +}): Promise { const result = await input.sdk.client.experimental.workspace - .sessionRestore({ id: input.workspaceID, sessionID: input.sessionID }) + .warp({ + id: input.workspaceID, + sessionID: input.sessionID, + }) .catch(() => undefined) - if (!result?.data) { + if (!result || result.error) { input.toast.show({ - message: `Failed to restore session: ${errorMessage(result?.error ?? "no response")}`, + message: `Failed to warp session: ${errorMessage(result?.error ?? "no response")}`, variant: "error", }) - return + return false } input.project.workspace.set(input.workspaceID) await input.sync.bootstrap({ fatal: false }).catch(() => undefined) - await Promise.all([input.project.workspace.sync(), input.sync.session.sync(input.sessionID)]) + await Promise.all([input.project.workspace.sync(), input.sync.session.refresh()]) - input.toast.show({ - message: "Session restored into the new workspace", - variant: "success", - }) + if (input.showSuccessToast !== false) { + input.toast.show({ + message: "Session warped into the new workspace", + variant: "success", + }) + } input.done?.() - if (input.done) return + if (input.done) return true input.dialog.clear() -} - -function DialogWorkspaceTypeSelect(props: { onSelect: (adaptor: Adaptor) => Promise | void }) { - const dialog = useDialog() - const sync = useSync() - const sdk = useSDK() - const toast = useToast() - const [adaptors, setAdaptors] = createSignal() - - onMount(() => { - dialog.setSize("medium") - void (async () => { - const dir = sync.path.directory || sdk.directory - const url = new URL("/experimental/workspace/adaptor", sdk.url) - if (dir) url.searchParams.set("directory", dir) - const res = await sdk - .fetch(url) - .then((x) => x.json() as Promise) - .catch(() => undefined) - if (!res) { - toast.show({ - message: "Failed to load workspace adaptors", - variant: "error", - }) - return - } - setAdaptors(res) - })() - }) - - const options = createMemo(() => { - const list = adaptors() - if (!list) { - return [ - { - title: "Loading workspaces...", - value: undefined, - description: "Fetching available workspace adaptors", - }, - ] - } - return list.map((item) => ({ - title: item.name, - value: item, - description: item.description, - })) - }) - - return ( - { - if (!option.value) return - void props.onSelect(option.value) - }} - /> - ) + return true } export function DialogWorkspaceSelect(props: { @@ -174,7 +120,15 @@ export function DialogWorkspaceSelect(props: { }, ] } - const workspaces = project.workspace.list() + const recent = sync.data.session + .toSorted((a, b) => b.time.updated - a.time.updated) + .flatMap((session) => (session.workspaceID ? [session.workspaceID] : [])) + .filter((workspaceID, index, list) => list.indexOf(workspaceID) === index) + .slice(0, 3) + .flatMap((workspaceID) => { + const workspace = project.workspace.get(workspaceID) + return workspace ? [workspace] : [] + }) return [ ...list.map((adaptor) => ({ title: adaptor.name, @@ -188,7 +142,7 @@ export function DialogWorkspaceSelect(props: { description: "Use the local project", category: "Choose workspace", }, - ...workspaces.slice(0, 3).map((workspace: Workspace) => ({ + ...recent.map((workspace: Workspace) => ({ title: workspace.name, description: `(${workspace.type})`, value: { @@ -265,66 +219,3 @@ function DialogExistingWorkspaceSelect(props: { onSelect: (selection: WorkspaceS /> ) } - -export function DialogWorkspaceCreate(props: { onSelect: (workspaceID: string) => Promise | void }) { - const dialog = useDialog() - const project = useProject() - const sdk = useSDK() - const toast = useToast() - const [creating, setCreating] = createSignal() - - onMount(() => { - dialog.setSize("medium") - }) - - const options = createMemo(() => { - const type = creating() - if (type) { - return [ - { - title: `Creating ${type} workspace...`, - value: "creating" as const, - description: "This can take a while for remote environments", - }, - ] - } - return [] - }) - - const create = async (type: string) => { - if (creating()) return - setCreating(type) - - const result = await sdk.client.experimental.workspace.create({ type, branch: null }).catch(() => { - toast.show({ - message: "Creating workspace failed", - variant: "error", - }) - return undefined - }) - - const workspace = result?.data - if (!workspace) { - setCreating(undefined) - toast.show({ - message: `Failed to create workspace: ${errorMessage(result?.error ?? "no response")}`, - variant: "error", - }) - return - } - - await project.workspace.sync() - await props.onSelect(workspace.id) - setCreating(undefined) - } - - return creating() ? ( - - ) : ( - { - void create(adaptor.type) - }} - /> - ) -} diff --git a/packages/opencode/src/cli/cmd/tui/component/prompt/index.tsx b/packages/opencode/src/cli/cmd/tui/component/prompt/index.tsx index 4adc3fb2ec..3f837866a4 100644 --- a/packages/opencode/src/cli/cmd/tui/component/prompt/index.tsx +++ b/packages/opencode/src/cli/cmd/tui/component/prompt/index.tsx @@ -41,12 +41,7 @@ import { useKV } from "../../context/kv" import { createFadeIn } from "../../util/signal" import { useTextareaKeybindings } from "../textarea-keybindings" import { DialogSkill } from "../dialog-skill" -import { - DialogWorkspaceCreate, - DialogWorkspaceSelect, - restoreWorkspaceSession, - type WorkspaceSelection, -} from "../dialog-workspace-create" +import { DialogWorkspaceSelect, warpWorkspaceSession, type WorkspaceSelection } from "../dialog-workspace-create" import { DialogWorkspaceUnavailable } from "../dialog-workspace-unavailable" import { useArgs } from "@tui/context/args" import { Flag } from "@opencode-ai/core/flag/flag" @@ -163,6 +158,7 @@ export function Prompt(props: PromptProps) { const [workspaceSelection, setWorkspaceSelection] = createSignal() const [workspaceCreating, setWorkspaceCreating] = createSignal(false) const [workspaceCreatingDots, setWorkspaceCreatingDots] = createSignal(3) + const [warpNotice, setWarpNotice] = createSignal() const currentProviderLabel = createMemo(() => local.model.parsed().provider) const hasRightContent = createMemo(() => Boolean(props.right)) const selectedWorkspace = () => props.workspaceSelection ?? workspaceSelection() @@ -177,6 +173,72 @@ export function Prompt(props: PromptProps) { props.onWorkspaceCreatingChange?.(creating) } + function showWarpNotice(name: string) { + setWarpNotice(`Warped to ${name}`) + setTimeout(() => setWarpNotice(undefined), 4000) + } + + async function createWorkspace(selection: Extract) { + setCreatingWorkspace(true) + const result = await sdk.client.experimental.workspace + .create({ type: selection.workspaceType, branch: null }) + .catch(() => undefined) + if (result == undefined || result.error || !result.data) { + selectWorkspace(undefined) + setCreatingWorkspace(false) + toast.show({ + message: "Creating workspace failed", + variant: "error", + }) + return + } + + await project.workspace.sync() + const workspace = result.data + selectWorkspace({ + type: "existing", + workspaceID: workspace.id, + workspaceType: workspace.type, + workspaceName: workspace.name, + }) + setCreatingWorkspace(false) + return workspace + } + + async function warpSession(selection: WorkspaceSelection) { + if (!props.sessionID) { + selectWorkspace(selection) + dialog.clear() + if (selection.type === "new") void createWorkspace(selection) + return + } + if (selection.type === "none") { + dialog.clear() + return + } + + selectWorkspace(selection) + dialog.clear() + + const workspace = + selection.type === "existing" + ? { id: selection.workspaceID, name: selection.workspaceName } + : await createWorkspace(selection) + if (!workspace) return + + const warped = await warpWorkspaceSession({ + dialog, + sdk, + sync, + project, + toast, + workspaceID: workspace.id, + sessionID: props.sessionID, + showSuccessToast: false, + }) + if (warped) showWarpNotice(workspace.name) + } + createEffect(() => { if (!workspaceCreating()) { setWorkspaceCreatingDots(3) @@ -219,8 +281,8 @@ export function Prompt(props: PromptProps) { createEffect(() => { if (!input || input.isDestroyed) return - if (props.disabled || workspaceCreating()) input.cursorColor = theme.backgroundElement - if (!props.disabled && !workspaceCreating()) input.cursorColor = theme.text + if (props.disabled) input.cursorColor = theme.backgroundElement + if (!props.disabled) input.cursorColor = theme.text }) const lastUserMessage = createMemo(() => { @@ -499,8 +561,7 @@ export function Prompt(props: PromptProps) { { - selectWorkspace(selection) - dialog.clear() + void warpSession(selection) }} /> )) @@ -563,7 +624,7 @@ export function Prompt(props: PromptProps) { createEffect(() => { if (!input || input.isDestroyed) return - if (props.visible === false || dialog.stack.length > 0 || workspaceCreating()) { + if (props.visible === false || dialog.stack.length > 0) { if (input.focused) input.blur() return } @@ -583,7 +644,7 @@ export function Prompt(props: PromptProps) { : undefined input.traits = { capture, - suspend: !!props.disabled || workspaceCreating() || store.mode === "shell", + suspend: !!props.disabled || store.mode === "shell", status: store.mode === "shell" ? "SHELL" : undefined, } }) @@ -722,6 +783,7 @@ export function Prompt(props: PromptProps) { ]) async function submit() { + setWarpNotice(undefined) // IME: double-defer may fire before onContentChange flushes the last // composed character (e.g. Korean hangul) to the store, so read // plainText directly and sync before any downstream reads. @@ -729,7 +791,8 @@ export function Prompt(props: PromptProps) { setStore("prompt", "input", input.plainText) syncExtmarksWithPromptParts() } - if (props.disabled || workspaceCreating()) return false + if (props.disabled) return false + if (workspaceCreating()) return false if (autocomplete?.visible) return false if (!store.prompt.input) return false const agent = local.agent.current() @@ -753,20 +816,14 @@ export function Prompt(props: PromptProps) { { dialog.replace(() => ( - - restoreWorkspaceSession({ - dialog, - sdk, - sync, - project, - toast, - workspaceID: nextWorkspaceID, - sessionID: props.sessionID!, - }) - } + { + void warpSession(selection) + }} /> )) + return false }} /> )) @@ -836,33 +893,13 @@ export function Prompt(props: PromptProps) { let sessionID = props.sessionID if (sessionID == null) { const workspace = selectedWorkspace() - if (workspace?.type === "new") { - setCreatingWorkspace(true) - await new Promise((resolve) => setTimeout(resolve, 10_000)) - } - const workspaceID = await iife(async () => { + const workspaceID = iife(() => { if (!workspace) return undefined if (workspace.type === "none") return undefined if (workspace.type === "existing") return workspace.workspaceID - - const result = await sdk.client.experimental.workspace - .create({ type: workspace.workspaceType, branch: null }) - .catch(() => undefined) - if (result == undefined || result.error || !result.data) return undefined - - await project.workspace.sync() - return result.data.id + return undefined }) - if (workspace?.type === "new" && !workspaceID) { - setCreatingWorkspace(false) - toast.show({ - message: "Creating workspace failed", - variant: "error", - }) - return true - } - const res = await sdk.client.session.create({ workspace: workspaceID }) if (res.error) { @@ -1092,6 +1129,7 @@ export function Prompt(props: PromptProps) { const selected = selectedWorkspace() if (!selected) return if (selected.type === "none") return + if (props.sessionID && !workspaceCreating()) return if (selected.type === "new") { return { type: "new", @@ -1184,7 +1222,7 @@ export function Prompt(props: PromptProps) { }} keyBindings={textareaKeybindings()} onKeyDown={async (e) => { - if (props.disabled || workspaceCreating()) { + if (props.disabled) { e.preventDefault() return } @@ -1267,7 +1305,7 @@ export function Prompt(props: PromptProps) { setTimeout(() => setTimeout(() => submit(), 0), 0) }} onPaste={async (event: PasteEvent) => { - if (props.disabled || workspaceCreating()) { + if (props.disabled) { event.preventDefault() return } @@ -1362,7 +1400,7 @@ export function Prompt(props: PromptProps) { }} onMouseDown={(r: MouseEvent) => r.target?.focus()} focusedBackgroundColor={theme.backgroundElement} - cursorColor={workspaceCreating() ? theme.backgroundElement : theme.text} + cursorColor={props.disabled ? theme.backgroundElement : theme.text} syntaxStyle={syntax()} /> @@ -1433,35 +1471,8 @@ export function Prompt(props: PromptProps) { - - {(workspace) => ( - - - - - - {(() => { - const item = workspace() - if (item.type === "new") { - if (workspaceCreating()) return `Creating ${item.workspaceType}${".".repeat(workspaceCreatingDots())}` - return ( - <> - Workspace (new {item.workspaceType}) - - ) - } - return ( - <> - Workspace {item.workspaceName} - - ) - })()} - - - )} - - - }> + + + + {(notice) => ( + + {notice()} + + )} + + + {(workspace) => ( + + + + + + {(() => { + const item = workspace() + if (item.type === "new") { + if (workspaceCreating()) + return `Creating ${item.workspaceType}${".".repeat(workspaceCreatingDots())}` + return ( + <> + Workspace (new {item.workspaceType}) + + ) + } + return ( + <> + Workspace {item.workspaceName} + + ) + })()} + + + )} + + {props.hint ?? } diff --git a/packages/opencode/src/control-plane/workspace.ts b/packages/opencode/src/control-plane/workspace.ts index 1dd5361500..427d20af95 100644 --- a/packages/opencode/src/control-plane/workspace.ts +++ b/packages/opencode/src/control-plane/workspace.ts @@ -1,4 +1,4 @@ -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 { fn } from "@/util/fn" import { Database } from "@/storage/db" @@ -27,7 +27,7 @@ import { errorData } from "@/util/error" import { makeRuntime } from "@/effect/run-service" import { waitEvent } from "./util" import { WorkspaceContext } from "./workspace-context" -import { NonNegativeInt, withStatics } from "@/util/schema" +import { withStatics } from "@/util/schema" import { zod as effectZod, zodObject } from "@/util/effect-zod" export const Info = WorkspaceInfoSchema @@ -39,13 +39,6 @@ export const ConnectionStatus = Schema.Struct({ }) export type ConnectionStatus = Schema.Schema.Type -const Restore = Schema.Struct({ - workspaceID: WorkspaceID, - sessionID: SessionID, - total: NonNegativeInt, - step: NonNegativeInt, -}) - export const Event = { Ready: BusEvent.define( "workspace.ready", @@ -59,7 +52,6 @@ export const Event = { message: Schema.String, }), ), - Restore: BusEvent.define("workspace.restore", Restore), Status: BusEvent.define("workspace.status", ConnectionStatus), } @@ -89,11 +81,11 @@ export const CreateInput = Schema.Struct({ }).pipe(withStatics((s) => ({ zod: effectZod(s), zodObject: zodObject(s) }))) export type CreateInput = Schema.Schema.Type -export const SessionRestoreInput = Schema.Struct({ +export const SessionWarpInput = Schema.Struct({ workspaceID: WorkspaceID, sessionID: SessionID, }).pipe(withStatics((s) => ({ zod: effectZod(s), zodObject: zodObject(s) }))) -export type SessionRestoreInput = Schema.Schema.Type +export type SessionWarpInput = Schema.Schema.Type export class SyncHttpError extends Schema.TaggedErrorClass()("WorkspaceSyncHttpError", { message: Schema.String, @@ -117,8 +109,8 @@ export class SessionEventsNotFoundError extends Schema.TaggedErrorClass()( - "WorkspaceSessionRestoreHttpError", +export class SessionWarpHttpError extends Schema.TaggedErrorClass()( + "WorkspaceSessionWarpHttpError", { message: Schema.String, workspaceID: WorkspaceID, @@ -139,17 +131,17 @@ export class SyncAbortedError extends Schema.TaggedErrorClass( }) {} 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 - readonly sessionRestore: (input: SessionRestoreInput) => Effect.Effect<{ total: number }, SessionRestoreError> + readonly sessionWarp: (input: SessionWarpInput) => Effect.Effect readonly list: (project: Project.Info) => Effect.Effect readonly get: (id: WorkspaceID) => Effect.Effect readonly remove: (id: WorkspaceID) => Effect.Effect @@ -501,9 +493,9 @@ 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, }) @@ -515,18 +507,71 @@ export const layer = Layer.effect( workspaceID: 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) { + yield* Effect.gen(function* () { + const adaptor = getAdaptor(previous.projectID, previous.type) + const target = yield* Effect.promise(() => Promise.resolve(adaptor.target(previous))) + if (target.type === "local") return + + const response = yield* http.execute( + HttpClientRequest.post(route(target.url, "/sync/erase"), { + headers: new Headers(target.headers), + body: HttpBody.jsonUnsafe({ sessionID: input.sessionID }), + }), + ) + if (response.status < 200 || response.status >= 300) { + const body = yield* response.text + log.warn("session warp erase failed", { + workspaceID: previous.id, + sessionID: input.sessionID, + status: response.status, + body, + }) + } + }).pipe( + Effect.catch((error) => + Effect.sync(() => { + log.warn("session warp erase unavailable", { + workspaceID: previous.id, + sessionID: input.sessionID, + error, + }) + }), + ), + ) + } + } + const adaptor = getAdaptor(space.projectID, space.type) const target = yield* Effect.promise(() => Promise.resolve(adaptor.target(space))) - // Update the local copy to point to the new workspace - yield* Effect.sync(() => - SyncEvent.run(Session.Event.Updated, { + if (target.type === "local") { + yield* Effect.sync(() => + SyncEvent.run(Session.Event.Updated, { + sessionID: input.sessionID, + info: { + workspaceID: input.workspaceID, + }, + }), + ) + + log.info("session warp complete", { + workspaceID: input.workspaceID, sessionID: input.sessionID, - info: { - workspaceID: input.workspaceID, - }, - }), - ) + target: target.directory, + }) + return + } const rows = yield* db((db) => db @@ -548,130 +593,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") { - SyncEvent.replayAll(events) - log.info("session restore batch replayed locally", { - workspaceID: input.workspaceID, - sessionID: input.sessionID, - step: i + 1, - total, - events: events.length, - }) - } else { - const url = route(target.url, "/sync/replay") - const 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 ${input.workspaceID}: HTTP ${response.status} ${body}`, + workspaceID: input.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 ${input.workspaceID}: HTTP ${response.status} ${body}`, + workspaceID: input.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), @@ -800,7 +810,7 @@ export const layer = Layer.effect( return Service.of({ create, - sessionRestore, + sessionWarp, list, get, remove, @@ -862,7 +872,7 @@ const { runPromise, runSync } = makeRuntime(Service, defaultLayer) export const create = fn(CreateInput.zod, (input) => runPromise((svc) => svc.create(input))) -export const sessionRestore = fn(SessionRestoreInput.zod, (input) => runPromise((svc) => svc.sessionRestore(input))) +export const sessionWarp = fn(SessionWarpInput.zod, (input) => runPromise((svc) => svc.sessionWarp(input))) export function list(project: Project.Info) { return Database.use((db) => diff --git a/packages/opencode/src/server/backend.ts b/packages/opencode/src/server/backend.ts index f456dc0be5..8c9986f094 100644 --- a/packages/opencode/src/server/backend.ts +++ b/packages/opencode/src/server/backend.ts @@ -11,7 +11,7 @@ export type Selection = { export type Attributes = ReturnType export function select(): Selection { - if (Flag.OPENCODE_EXPERIMENTAL_HTTPAPI) return { backend: "effect-httpapi", reason: "env" } + // if (Flag.OPENCODE_EXPERIMENTAL_HTTPAPI) return { backend: "effect-httpapi", reason: "env" } return { backend: "hono", reason: "stable" } } diff --git a/packages/opencode/src/server/routes/control/workspace.ts b/packages/opencode/src/server/routes/control/workspace.ts index 19fbc757fb..c631351955 100644 --- a/packages/opencode/src/server/routes/control/workspace.ts +++ b/packages/opencode/src/server/routes/control/workspace.ts @@ -8,10 +8,6 @@ import { zodObject } from "@/util/effect-zod" import { Instance } from "@/project/instance" import { errors } from "../../error" import { lazy } from "@/util/lazy" -import * as Log from "@opencode-ai/core/util/log" -import { errorData } from "@/util/error" - -const log = Log.create({ service: "server.workspace" }) export const WorkspaceRoutes = lazy(() => new Hono() @@ -142,56 +138,28 @@ export const WorkspaceRoutes = lazy(() => }, ) .post( - "/:id/session-restore", + "/:id/warp", describeRoute({ - summary: "Restore session into workspace", - description: "Replay a session's sync events into the target workspace in batches.", - operationId: "experimental.workspace.sessionRestore", + summary: "Warp session into workspace", + description: "Move a session's sync history into the target workspace.", + operationId: "experimental.workspace.warp", responses: { - 200: { - description: "Session replay started", - content: { - "application/json": { - schema: resolver( - z.object({ - total: z.number().int().min(0), - }), - ), - }, - }, + 204: { + description: "Session warped", }, ...errors(400), }, }), validator("param", z.object({ id: zodObject(Workspace.Info).shape.id })), - validator("json", Workspace.SessionRestoreInput.zodObject.omit({ workspaceID: true })), + validator("json", Workspace.SessionWarpInput.zodObject.omit({ workspaceID: true })), async (c) => { const { id } = c.req.valid("param") - const body = c.req.valid("json") as Omit - log.info("session restore route requested", { + const body = c.req.valid("json") as Omit + await Workspace.sessionWarp({ workspaceID: id, - sessionID: body.sessionID, - directory: Instance.directory, + ...body, }) - try { - const result = await Workspace.sessionRestore({ - workspaceID: id, - ...body, - }) - log.info("session restore route complete", { - workspaceID: id, - sessionID: body.sessionID, - total: result.total, - }) - return c.json(result) - } catch (err) { - log.error("session restore route failed", { - workspaceID: id, - sessionID: body.sessionID, - error: errorData(err), - }) - throw err - } + return c.body(null, 204) }, ), ) diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/workspace.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/workspace.ts index 268e84f2ec..aa79bc86f8 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/groups/workspace.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/groups/workspace.ts @@ -1,6 +1,5 @@ import { Workspace } from "@/control-plane/workspace" import { WorkspaceAdaptorEntry } from "@/control-plane/types" -import { NonNegativeInt } from "@/util/schema" import { Schema, Struct } from "effect" import { HttpApi, HttpApiEndpoint, HttpApiError, HttpApiGroup, OpenApi } from "effect/unstable/httpapi" import { Authorization } from "../middleware/authorization" @@ -10,17 +9,12 @@ import { described } from "./metadata" const root = "/experimental/workspace" export const CreatePayload = Schema.Struct(Struct.omit(Workspace.CreateInput.fields, ["projectID"])) -export const SessionRestorePayload = Schema.Struct(Struct.omit(Workspace.SessionRestoreInput.fields, ["workspaceID"])) -export const SessionRestoreResponse = Schema.Struct({ - total: NonNegativeInt, -}) export const WorkspacePaths = { adaptors: `${root}/adaptor`, list: root, status: `${root}/status`, remove: `${root}/:id`, - sessionRestore: `${root}/:id/session-restore`, } as const export const WorkspaceApi = HttpApi.make("workspace") @@ -76,18 +70,6 @@ export const WorkspaceApi = HttpApi.make("workspace") description: "Remove an existing workspace.", }), ), - HttpApiEndpoint.post("sessionRestore", WorkspacePaths.sessionRestore, { - params: { id: Workspace.Info.fields.id }, - payload: SessionRestorePayload, - success: described(SessionRestoreResponse, "Session replay started"), - error: HttpApiError.BadRequest, - }).annotateMerge( - OpenApi.annotations({ - identifier: "experimental.workspace.sessionRestore", - summary: "Restore session into workspace", - description: "Replay a session's sync events into the target workspace in batches.", - }), - ), ) .annotateMerge(OpenApi.annotations({ title: "workspace", description: "Experimental HttpApi workspace routes." })) .middleware(InstanceContextMiddleware) diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/workspace.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/workspace.ts index 9413c865d1..45b80ef2e7 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/workspace.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/workspace.ts @@ -5,7 +5,7 @@ import { Instance } from "@/project/instance" import { Effect } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { InstanceHttpApi } from "../api" -import { CreatePayload, SessionRestorePayload } from "../groups/workspace" +import { CreatePayload } from "../groups/workspace" export const workspaceHandlers = HttpApiBuilder.group(InstanceHttpApi, "workspace", (handlers) => Effect.gen(function* () { @@ -40,27 +40,11 @@ export const workspaceHandlers = HttpApiBuilder.group(InstanceHttpApi, "workspac return yield* Effect.promise(() => Instance.restore(instance, () => Workspace.remove(ctx.params.id))) }) - const sessionRestore = Effect.fn("WorkspaceHttpApi.sessionRestore")(function* (ctx: { - params: { id: Workspace.Info["id"] } - payload: typeof SessionRestorePayload.Type - }) { - const instance = yield* InstanceState.context - return yield* Effect.promise(() => - Instance.restore(instance, () => - Workspace.sessionRestore({ - workspaceID: ctx.params.id, - sessionID: ctx.payload.sessionID, - }), - ), - ) - }) - return handlers .handle("adaptors", adaptors) .handle("list", list) .handle("create", create) .handle("status", status) .handle("remove", remove) - .handle("sessionRestore", sessionRestore) }), ) diff --git a/packages/opencode/src/server/routes/instance/sync.ts b/packages/opencode/src/server/routes/instance/sync.ts index b480477774..82a91d5bcb 100644 --- a/packages/opencode/src/server/routes/instance/sync.ts +++ b/packages/opencode/src/server/routes/instance/sync.ts @@ -9,12 +9,15 @@ import { not } from "drizzle-orm" import { or } from "drizzle-orm" import { lte } from "drizzle-orm" import { eq } from "drizzle-orm" -import { EventTable } from "@/sync/event.sql" +import { EventSequenceTable, EventTable } from "@/sync/event.sql" import { lazy } from "@/util/lazy" import * as Log from "@opencode-ai/core/util/log" import { startWorkspaceSyncing } from "@/control-plane/workspace" import { Instance } from "@/project/instance" import { errors } from "../../error" +import { Session } from "@/session/session" +import { WorkspaceContext } from "@/control-plane/workspace-context" +import { SessionID } from "@/session/schema" const ReplayEvent = z.object({ id: z.string(), @@ -23,6 +26,9 @@ const ReplayEvent = z.object({ type: z.string(), data: z.record(z.string(), z.unknown()), }) +const SessionPayload = z.object({ + sessionID: SessionID.zod, +}) const log = Log.create({ service: "server.sync" }) @@ -105,6 +111,82 @@ export const SyncRoutes = lazy(() => }) }, ) + .post( + "/erase", + describeRoute({ + summary: "Erase session sync events", + description: "Erase all locally stored sync events for a session aggregate.", + operationId: "sync.erase", + responses: { + 200: { + description: "Erased session sync events", + content: { + "application/json": { + schema: resolver(SessionPayload), + }, + }, + }, + ...errors(400), + }, + }), + validator("json", SessionPayload), + async (c) => { + const body = c.req.valid("json") + Database.transaction((tx) => { + tx.delete(EventTable).where(eq(EventTable.aggregate_id, body.sessionID)).run() + tx.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, body.sessionID)).run() + }) + + log.info("sync events erased", { + sessionID: body.sessionID, + }) + + return c.json({ + sessionID: body.sessionID, + }) + }, + ) + .post( + "/steal", + describeRoute({ + summary: "Steal session into workspace", + description: "Update a session to belong to the current workspace through the sync event system.", + operationId: "sync.steal", + responses: { + 200: { + description: "Session stolen into workspace", + content: { + "application/json": { + schema: resolver(SessionPayload), + }, + }, + }, + ...errors(400), + }, + }), + validator("json", SessionPayload), + async (c) => { + const body = c.req.valid("json") + const workspaceID = WorkspaceContext.workspaceID + if (!workspaceID) throw new Error("Cannot steal session without workspace context") + + SyncEvent.run(Session.Event.Updated, { + sessionID: body.sessionID, + info: { + workspaceID, + }, + }) + + log.info("sync session stolen", { + sessionID: body.sessionID, + workspaceID, + }) + + return c.json({ + sessionID: body.sessionID, + }) + }, + ) .post( "/history", describeRoute({ diff --git a/packages/opencode/test/control-plane/workspace.test.ts b/packages/opencode/test/control-plane/workspace.test.ts index bd5c4df7d5..0561bb072e 100644 --- a/packages/opencode/test/control-plane/workspace.test.ts +++ b/packages/opencode/test/control-plane/workspace.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test" +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test" import fs from "node:fs/promises" import Http from "node:http" import path from "node:path" @@ -6,7 +6,7 @@ import { setTimeout as delay } from "node:timers/promises" import { NodeHttpServer } from "@effect/platform-node" import { Effect, Layer } from "effect" import { HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" -import { asc, eq } from "drizzle-orm" +import { eq } from "drizzle-orm" import * as Log from "@opencode-ai/core/util/log" import { Flag } from "@opencode-ai/core/flag/flag" import { GlobalBus, type GlobalEvent } from "@/bus/global" @@ -15,11 +15,10 @@ import { ProjectID } from "@/project/schema" import { ProjectTable } from "@/project/project.sql" import { Instance } from "@/project/instance" import { Session as SessionNs } from "@/session/session" -import { SessionID, MessageID, PartID } from "@/session/schema" +import { SessionID } from "@/session/schema" import { SessionTable } from "@/session/session.sql" -import { ModelID, ProviderID } from "@/provider/schema" import { SyncEvent } from "@/sync" -import { EventSequenceTable, EventTable } from "@/sync/event.sql" +import { EventSequenceTable } from "@/sync/event.sql" import { resetDatabase } from "../fixture/db" import { provideTmpdirInstance, tmpdir } from "../fixture/fixture" import { testEffect } from "../lib/effect" @@ -298,48 +297,14 @@ function sessionSequence(sessionID: SessionID) { )?.seq } -function eventRows(sessionID: SessionID) { - return Database.use((db) => - db - .select({ seq: EventTable.seq, type: EventTable.type, data: EventTable.data }) - .from(EventTable) - .where(eq(EventTable.aggregate_id, sessionID)) - .orderBy(asc(EventTable.seq)) - .all(), - ) -} - function sessionUpdatedType() { return SyncEvent.versionedType(SessionNs.Event.Updated.type, SessionNs.Event.Updated.version) } -function replaceSessionEvents(sessionID: SessionID, count: number) { - Database.use((db) => { - db.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, sessionID)).run() - if (count === 0) return - - db.insert(EventSequenceTable) - .values({ aggregate_id: sessionID, seq: count - 1 }) - .run() - db.insert(EventTable) - .values( - Array.from({ length: count }, (_, i) => ({ - id: `evt_${unique(`manual-${i}`)}`, - aggregate_id: sessionID, - seq: i, - type: sessionUpdatedType(), - data: { sessionID, info: { title: `manual ${i}` } }, - })), - ) - .run() - }) -} - describe("workspace-old schemas and exports", () => { test("keeps the historical event type names", () => { expect(WorkspaceOld.Event.Ready.type).toBe("workspace.ready") expect(WorkspaceOld.Event.Failed.type).toBe("workspace.failed") - expect(WorkspaceOld.Event.Restore.type).toBe("workspace.restore") expect(WorkspaceOld.Event.Status.type).toBe("workspace.status") }) @@ -357,16 +322,6 @@ describe("workspace-old schemas and exports", () => { expect(() => WorkspaceOld.CreateInput.zod.parse({ ...input, branch: 1 })).toThrow() }) - test("validates session restore input", () => { - const input = { - workspaceID: WorkspaceID.ascending("wrk_schema_restore"), - sessionID: SessionID.descending("ses_schema_restore"), - } - - expect(WorkspaceOld.SessionRestoreInput.zod.parse(input)).toEqual(input) - expect(() => WorkspaceOld.SessionRestoreInput.zod.parse({ ...input, workspaceID: "bad" })).toThrow() - expect(() => WorkspaceOld.SessionRestoreInput.zod.parse({ ...input, sessionID: "bad" })).toThrow() - }) }) describe("workspace-old CRUD", () => { @@ -1190,320 +1145,3 @@ describe("workspace-old waitForSync", () => { }) }, 7000) }) - -describe("workspace-old sessionRestore", () => { - test("throws when the workspace is missing", async () => { - await withInstance(async () => { - await expect( - WorkspaceOld.sessionRestore({ - workspaceID: WorkspaceID.ascending("wrk_restore_missing"), - sessionID: SessionID.descending("ses_restore_missing_workspace"), - }), - ).rejects.toThrow("Workspace not found: wrk_restore_missing") - }) - }) - - test("throws when switching a missing session fails", async () => { - await withInstance(async (dir) => { - const type = unique("restore-missing-session") - const info = workspaceInfo(Instance.project.id, type, { directory: dir }) - insertWorkspace(info) - registerAdaptor(Instance.project.id, type, localAdaptor(dir).adaptor) - - await expect( - WorkspaceOld.sessionRestore({ workspaceID: info.id, sessionID: SessionID.descending("ses_missing_restore") }), - ).rejects.toThrow("NotFoundError") - await WorkspaceOld.remove(info.id) - }) - }) - - it.live("posts remote replay batches of 10, emits progress, and includes the workspace update event", () => { - const replay: FetchCall[] = [] - return Effect.gen(function* () { - yield* HttpServer.serveEffect()( - Effect.gen(function* () { - const req = yield* HttpServerRequest.HttpServerRequest - const bodyText = yield* req.text - const call = { - url: new URL(req.url, "http://localhost"), - method: req.method, - headers: new Headers(req.headers), - bodyText, - json: bodyText ? JSON.parse(bodyText) : undefined, - } - if (call.url.pathname === "/restore/sync/replay") { - replay.push(call) - return HttpServerResponse.fromWeb(Response.json({ ok: true })) - } - return HttpServerResponse.text("unexpected", { status: 500 }) - }), - ) - const url = yield* serverUrl() - yield* provideTmpdirInstance( - (dir) => - Effect.gen(function* () { - const workspace = yield* WorkspaceOld.Service - const sessionSvc = yield* SessionNs.Service - const captured = captureGlobalEvents() - try { - const type = unique("restore-remote") - const info = workspaceInfo(Instance.project.id, type, { directory: dir }) - insertWorkspace(info) - registerAdaptor( - Instance.project.id, - type, - remoteAdaptor(`${url}/restore/?ignored=1#hash`, { - directory: dir, - headers: { authorization: "Bearer restore" }, - }).adaptor, - ) - const session = yield* sessionSvc.create({ title: "restore remote" }) - replaceSessionEvents(session.id, 24) - - const result = yield* workspace.sessionRestore({ workspaceID: info.id, sessionID: session.id }) - - expect(result).toEqual({ total: 3 }) - expect(replay).toHaveLength(3) - expect(replay.map((call) => call.url.pathname + call.url.search + call.url.hash)).toEqual([ - "/restore/sync/replay", - "/restore/sync/replay", - "/restore/sync/replay", - ]) - expect(replay.every((call) => call.headers.get("authorization") === "Bearer restore")).toBe(true) - expect(replay.every((call) => call.headers.get("content-type") === "application/json")).toBe(true) - expect(replay.map((call) => (call.json as { events: unknown[] }).events.length)).toEqual([10, 10, 5]) - expect(replay.map((call) => (call.json as { directory: string }).directory)).toEqual([dir, dir, dir]) - expect( - replay.flatMap((call) => - (call.json as { events: Array<{ seq: number }> }).events.map((event) => event.seq), - ), - ).toEqual(Array.from({ length: 25 }, (_, i) => i)) - expect( - (replay[2].json as { events: Array<{ seq: number; type: string; data: unknown }> }).events.at(-1), - ).toMatchObject({ - seq: 24, - type: sessionUpdatedType(), - data: { sessionID: session.id, info: { workspaceID: info.id } }, - }) - expect((yield* sessionSvc.get(session.id)).workspaceID).toBe(info.id) - expect( - captured.events - .filter( - (event) => event.workspace === info.id && event.payload.type === WorkspaceOld.Event.Restore.type, - ) - .map((event) => event.payload.properties.step), - ).toEqual([0, 1, 2, 3]) - yield* workspace.remove(info.id) - } finally { - captured.dispose() - } - }), - { git: true }, - ) - }) - }) - - it.live("remote restore sends an empty directory string when the workspace directory is null", () => { - const replay: FetchCall[] = [] - return Effect.gen(function* () { - yield* HttpServer.serveEffect()( - Effect.gen(function* () { - const req = yield* HttpServerRequest.HttpServerRequest - const bodyText = yield* req.text - replay.push({ - url: new URL(req.url, "http://localhost"), - method: req.method, - headers: new Headers(req.headers), - bodyText, - json: bodyText ? JSON.parse(bodyText) : undefined, - }) - return HttpServerResponse.fromWeb(Response.json({ ok: true })) - }), - ) - const url = yield* serverUrl() - yield* provideTmpdirInstance( - () => - Effect.gen(function* () { - const workspace = yield* WorkspaceOld.Service - const sessionSvc = yield* SessionNs.Service - const type = unique("restore-null-dir") - const info = workspaceInfo(Instance.project.id, type, { directory: null }) - insertWorkspace(info) - registerAdaptor(Instance.project.id, type, remoteAdaptor(`${url}/null-dir`, { directory: null }).adaptor) - const session = yield* sessionSvc.create({ title: "null dir" }) - replaceSessionEvents(session.id, 0) - - expect(yield* workspace.sessionRestore({ workspaceID: info.id, sessionID: session.id })).toEqual({ - total: 1, - }) - expect((replay[0].json as { directory: string }).directory).toBe("") - expect((replay[0].json as { events: unknown[] }).events).toHaveLength(1) - yield* workspace.remove(info.id) - }), - { git: true }, - ) - }) - }) - - it.live("remote restore failures include status and body and do not emit completed batch progress", () => { - const replay: FetchCall[] = [] - return Effect.gen(function* () { - yield* HttpServer.serveEffect()( - Effect.gen(function* () { - const req = yield* HttpServerRequest.HttpServerRequest - const bodyText = yield* req.text - replay.push({ - url: new URL(req.url, "http://localhost"), - method: req.method, - headers: new Headers(req.headers), - bodyText, - json: bodyText ? JSON.parse(bodyText) : undefined, - }) - return HttpServerResponse.text("replay failed", { status: 503 }) - }), - ) - const url = yield* serverUrl() - yield* provideTmpdirInstance( - (dir) => - Effect.gen(function* () { - const workspace = yield* WorkspaceOld.Service - const sessionSvc = yield* SessionNs.Service - const captured = captureGlobalEvents() - try { - const type = unique("restore-remote-fail") - const info = workspaceInfo(Instance.project.id, type, { directory: dir }) - insertWorkspace(info) - registerAdaptor(Instance.project.id, type, remoteAdaptor(`${url}/fail`, { directory: dir }).adaptor) - const session = yield* sessionSvc.create({ title: "restore fail" }) - replaceSessionEvents(session.id, 11) - - const error = yield* Effect.flip( - workspace.sessionRestore({ workspaceID: info.id, sessionID: session.id }), - ) - expect((error as Error).message).toContain( - `Failed to replay session ${session.id} into workspace ${info.id}: HTTP 503 replay failed`, - ) - - expect(replay).toHaveLength(1) - expect( - captured.events - .filter( - (event) => event.workspace === info.id && event.payload.type === WorkspaceOld.Event.Restore.type, - ) - .map((event) => event.payload.properties.step), - ).toEqual([0]) - yield* workspace.remove(info.id) - } finally { - captured.dispose() - } - }), - { git: true }, - ) - }) - }) - - test("local restore replays batches without fetch and emits progress", async () => { - await withInstance(async (dir) => { - const captured = captureGlobalEvents() - let fetchCallCount = 0 - const replayAll = spyOn(SyncEvent, "replayAll") - try { - using server = Bun.serve({ - port: 0, - fetch() { - fetchCallCount++ - return Response.json({ ok: true }) - }, - }) - const type = unique("restore-local") - const info = workspaceInfo(Instance.project.id, type, { directory: dir }) - insertWorkspace(info) - registerAdaptor(Instance.project.id, type, localAdaptor(dir).adaptor) - const session = await AppRuntime.runPromise( - SessionNs.Service.use((svc) => svc.create({ title: "restore local" })), - ) - replaceSessionEvents(session.id, 20) - - expect(await WorkspaceOld.sessionRestore({ workspaceID: info.id, sessionID: session.id })).toEqual({ total: 3 }) - - expect(fetchCallCount).toBe(0) - expect(replayAll).toHaveBeenCalledTimes(3) - expect(replayAll.mock.calls.map((call) => call[0].length)).toEqual([10, 10, 1]) - expect((await AppRuntime.runPromise(SessionNs.Service.use((svc) => svc.get(session.id)))).workspaceID).toBe( - info.id, - ) - expect(eventRows(session.id).map((row) => row.seq)).toEqual(Array.from({ length: 21 }, (_, i) => i)) - expect( - captured.events - .filter((event) => event.workspace === info.id && event.payload.type === WorkspaceOld.Event.Restore.type) - .map((event) => event.payload.properties.step), - ).toEqual([0, 1, 2, 3]) - await WorkspaceOld.remove(info.id) - } finally { - captured.dispose() - } - }) - }) - - it.live("session restore includes real message and part events in sequence order", () => { - const replay: FetchCall[] = [] - return Effect.gen(function* () { - yield* HttpServer.serveEffect()( - Effect.gen(function* () { - const req = yield* HttpServerRequest.HttpServerRequest - const bodyText = yield* req.text - replay.push({ - url: new URL(req.url, "http://localhost"), - method: req.method, - headers: new Headers(req.headers), - bodyText, - json: bodyText ? JSON.parse(bodyText) : undefined, - }) - return HttpServerResponse.fromWeb(Response.json({ ok: true })) - }), - ) - const url = yield* serverUrl() - yield* provideTmpdirInstance( - (dir) => - Effect.gen(function* () { - const workspace = yield* WorkspaceOld.Service - const sessionSvc = yield* SessionNs.Service - const type = unique("restore-real-events") - const info = workspaceInfo(Instance.project.id, type, { directory: dir }) - insertWorkspace(info) - registerAdaptor(Instance.project.id, type, remoteAdaptor(`${url}/real`, { directory: dir }).adaptor) - const session = yield* sessionSvc.create({ title: "real events" }) - for (let i = 0; i < 3; i++) { - const msg = yield* sessionSvc.updateMessage({ - id: MessageID.ascending(), - role: "user", - sessionID: session.id, - agent: "build", - model: { providerID: ProviderID.make("test"), modelID: ModelID.make("test") }, - time: { created: Date.now() }, - }) - yield* sessionSvc.updatePart({ - id: PartID.ascending(), - sessionID: session.id, - messageID: msg.id, - type: "text", - text: `message ${i}`, - }) - } - const before = eventRows(session.id) - - expect(yield* workspace.sessionRestore({ workspaceID: info.id, sessionID: session.id })).toEqual({ - total: 1, - }) - - const posted = (replay[0].json as { events: Array<{ seq: number; type: string }> }).events - expect(posted.map((event) => event.seq)).toEqual([...before.map((row) => row.seq), before.at(-1)!.seq + 1]) - expect(posted.map((event) => event.type).slice(0, -1)).toEqual(before.map((row) => row.type)) - expect(posted.at(-1)?.type).toBe(sessionUpdatedType()) - yield* workspace.remove(info.id) - }), - { git: true }, - ) - }) - }) -}) diff --git a/packages/opencode/test/server/httpapi-workspace.test.ts b/packages/opencode/test/server/httpapi-workspace.test.ts index 74dfbaef86..6b1efb2c42 100644 --- a/packages/opencode/test/server/httpapi-workspace.test.ts +++ b/packages/opencode/test/server/httpapi-workspace.test.ts @@ -179,18 +179,6 @@ describe("workspace HttpApi", () => { const workspace = (await created.json()) as Workspace.Info expect(workspace).toMatchObject({ type: "local-test", name: "local-test" }) - const session = await Instance.provide({ - directory: tmp.path, - fn: async () => runSession(Session.Service.use((svc) => svc.create({}))), - }) - const restored = await request(WorkspacePaths.sessionRestore.replace(":id", workspace.id), tmp.path, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ sessionID: session.id }), - }) - expect(restored.status).toBe(200) - expect((await restored.json()) as { total: number }).toMatchObject({ total: expect.any(Number) }) - const removed = await request(WorkspacePaths.remove.replace(":id", workspace.id), tmp.path, { method: "DELETE" }) expect(removed.status).toBe(200) expect(await removed.json()).toMatchObject({ id: workspace.id }) diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index 2da7c865d7..b3a15953ca 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -35,9 +35,9 @@ import type { ExperimentalWorkspaceListResponses, ExperimentalWorkspaceRemoveErrors, ExperimentalWorkspaceRemoveResponses, - ExperimentalWorkspaceSessionRestoreErrors, - ExperimentalWorkspaceSessionRestoreResponses, ExperimentalWorkspaceStatusResponses, + ExperimentalWorkspaceWarpErrors, + ExperimentalWorkspaceWarpResponses, FileListResponses, FilePartInput, FilePartSource, @@ -160,11 +160,15 @@ import type { SessionUpdateErrors, SessionUpdateResponses, SubtaskPartInput, + SyncEraseErrors, + SyncEraseResponses, SyncHistoryListErrors, SyncHistoryListResponses, SyncReplayErrors, SyncReplayResponses, SyncStartResponses, + SyncStealErrors, + SyncStealResponses, TextPartInput, ToolIdsErrors, ToolIdsResponses, @@ -689,11 +693,11 @@ export class Workspace extends HeyApiClient { } /** - * Restore session into workspace + * Warp session into workspace * - * Replay a session's sync events into the target workspace in batches. + * Move a session's sync history into the target workspace. */ - public sessionRestore( + public warp( parameters: { id: string directory?: string @@ -716,11 +720,11 @@ export class Workspace extends HeyApiClient { ], ) return (options?.client ?? this.client).post< - ExperimentalWorkspaceSessionRestoreResponses, - ExperimentalWorkspaceSessionRestoreErrors, + ExperimentalWorkspaceWarpResponses, + ExperimentalWorkspaceWarpErrors, ThrowOnError >({ - url: "/experimental/workspace/{id}/session-restore", + url: "/experimental/workspace/{id}/warp", ...options, ...params, headers: { @@ -3173,6 +3177,80 @@ export class Sync extends HeyApiClient { }) } + /** + * Erase session sync events + * + * Erase all locally stored sync events for a session aggregate. + */ + public erase( + parameters?: { + directory?: string + workspace?: string + sessionID?: string + }, + options?: Options, + ) { + const params = buildClientParams( + [parameters], + [ + { + args: [ + { in: "query", key: "directory" }, + { in: "query", key: "workspace" }, + { in: "body", key: "sessionID" }, + ], + }, + ], + ) + return (options?.client ?? this.client).post({ + url: "/sync/erase", + ...options, + ...params, + headers: { + "Content-Type": "application/json", + ...options?.headers, + ...params.headers, + }, + }) + } + + /** + * Steal session into workspace + * + * Update a session to belong to the current workspace through the sync event system. + */ + public steal( + parameters?: { + directory?: string + workspace?: string + sessionID?: string + }, + options?: Options, + ) { + const params = buildClientParams( + [parameters], + [ + { + args: [ + { in: "query", key: "directory" }, + { in: "query", key: "workspace" }, + { in: "body", key: "sessionID" }, + ], + }, + ], + ) + return (options?.client ?? this.client).post({ + url: "/sync/steal", + ...options, + ...params, + headers: { + "Content-Type": "application/json", + ...options?.headers, + ...params.headers, + }, + }) + } + private _history?: History get history(): History { return (this._history ??= new History({ client: this.client })) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 9bb1e50aac..a53bd34b58 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -466,16 +466,6 @@ export type EventWorkspaceFailed = { } } -export type EventWorkspaceRestore = { - type: "workspace.restore" - properties: { - workspaceID: string - sessionID: string - total: number - step: number - } -} - export type EventWorkspaceStatus = { type: "workspace.status" properties: { @@ -1143,7 +1133,6 @@ export type GlobalEvent = { | EventVcsBranchUpdated | EventWorkspaceReady | EventWorkspaceFailed - | EventWorkspaceRestore | EventWorkspaceStatus | EventWorktreeReady | EventWorktreeFailed @@ -2086,7 +2075,6 @@ export type Event = | EventVcsBranchUpdated | EventWorkspaceReady | EventWorkspaceFailed - | EventWorkspaceRestore | EventWorkspaceStatus | EventWorktreeReady | EventWorktreeFailed @@ -2564,7 +2552,7 @@ export type ExperimentalWorkspaceRemoveResponses = { export type ExperimentalWorkspaceRemoveResponse = ExperimentalWorkspaceRemoveResponses[keyof ExperimentalWorkspaceRemoveResponses] -export type ExperimentalWorkspaceSessionRestoreData = { +export type ExperimentalWorkspaceWarpData = { body?: { sessionID: string } @@ -2575,30 +2563,27 @@ export type ExperimentalWorkspaceSessionRestoreData = { directory?: string workspace?: string } - url: "/experimental/workspace/{id}/session-restore" + url: "/experimental/workspace/{id}/warp" } -export type ExperimentalWorkspaceSessionRestoreErrors = { +export type ExperimentalWorkspaceWarpErrors = { /** * Bad request */ 400: BadRequestError } -export type ExperimentalWorkspaceSessionRestoreError = - ExperimentalWorkspaceSessionRestoreErrors[keyof ExperimentalWorkspaceSessionRestoreErrors] +export type ExperimentalWorkspaceWarpError = ExperimentalWorkspaceWarpErrors[keyof ExperimentalWorkspaceWarpErrors] -export type ExperimentalWorkspaceSessionRestoreResponses = { +export type ExperimentalWorkspaceWarpResponses = { /** - * Session replay started + * Session warped */ - 200: { - total: number - } + 204: void } -export type ExperimentalWorkspaceSessionRestoreResponse = - ExperimentalWorkspaceSessionRestoreResponses[keyof ExperimentalWorkspaceSessionRestoreResponses] +export type ExperimentalWorkspaceWarpResponse = + ExperimentalWorkspaceWarpResponses[keyof ExperimentalWorkspaceWarpResponses] export type ProjectListData = { body?: never @@ -4626,6 +4611,70 @@ export type SyncReplayResponses = { export type SyncReplayResponse = SyncReplayResponses[keyof SyncReplayResponses] +export type SyncEraseData = { + body?: { + sessionID: string + } + path?: never + query?: { + directory?: string + workspace?: string + } + url: "/sync/erase" +} + +export type SyncEraseErrors = { + /** + * Bad request + */ + 400: BadRequestError +} + +export type SyncEraseError = SyncEraseErrors[keyof SyncEraseErrors] + +export type SyncEraseResponses = { + /** + * Erased session sync events + */ + 200: { + sessionID: string + } +} + +export type SyncEraseResponse = SyncEraseResponses[keyof SyncEraseResponses] + +export type SyncStealData = { + body?: { + sessionID: string + } + path?: never + query?: { + directory?: string + workspace?: string + } + url: "/sync/steal" +} + +export type SyncStealErrors = { + /** + * Bad request + */ + 400: BadRequestError +} + +export type SyncStealError = SyncStealErrors[keyof SyncStealErrors] + +export type SyncStealResponses = { + /** + * Session stolen into workspace + */ + 200: { + sessionID: string + } +} + +export type SyncStealResponse = SyncStealResponses[keyof SyncStealResponses] + export type SyncHistoryListData = { body?: { [key: string]: number