diff --git a/apps/desktop/src/components/runners/chat-move-progress.tsx b/apps/desktop/src/components/runners/chat-move-progress.tsx new file mode 100644 index 00000000..2914f316 --- /dev/null +++ b/apps/desktop/src/components/runners/chat-move-progress.tsx @@ -0,0 +1,134 @@ +import { useEffect, useRef, useState } from "react"; +import type { QueryClient } from "@tanstack/react-query"; +import type { ChatMove, ConversationWithMessages, MoveChatInput } from "@godmode/shared"; +import { ArrowRightLeft, X } from "lucide-react"; +import { toast } from "sonner"; +import { toastApiError } from "@/components/vault/vault-utils"; +import { api } from "@/lib/api"; +import { qk } from "@/lib/queryKeys"; +import { cn } from "@/lib/utils"; +import { setChatMove, useChatMoves } from "@/stores/chatMoves"; + +const MB = 1024 * 1024; + +function size(n: number): string { + if (n < MB) return `${Math.max(1, Math.round(n / 1024))} KB`; + return `${(n / MB).toFixed(n < 10 * MB ? 1 : 0)} MB`; +} + +function clock(seconds: number): string { + const s = Math.max(0, Math.round(seconds)); + return s < 60 ? `${s} s` : `${Math.floor(s / 60)}:${String(s % 60).padStart(2, "0")} min`; +} + +/** Copy speed from the first bytes of the copy on, once there is enough to tell. */ +function useRate(move: ChatMove | undefined): number | null { + const first = useRef<{ at: number; bytes: number } | null>(null); + const copying = move?.stage === "copying"; + if (!copying) first.current = null; + else if (!first.current && move.bytes > 0) first.current = { at: Date.now(), bytes: move.bytes }; + if (!copying || !first.current) return null; + const seconds = (Date.now() - first.current.at) / 1000; + return seconds >= 1.5 ? (move.bytes - first.current.bytes) / seconds : null; +} + +function useElapsed(since: number): number { + const [at, setAt] = useState(Date.now()); + useEffect(() => { + const t = setInterval(() => setAt(Date.now()), 1000); + return () => clearInterval(t); + }, []); + return (at - since) / 1000; +} + +function ChatMoveToast({ toastId, conversationId, dest, agentName }: { toastId: string; conversationId: string; dest: string; agentName: string }) { + const move = useChatMoves((s) => s.moves[conversationId]); + const started = useRef(Date.now()); + const seen = useRef(false); + if (move) seen.current = true; + const finishing = !move && seen.current; + const elapsed = useElapsed(started.current); + const rate = useRate(move); + + const percent = move?.stage === "copying" && move.total ? Math.min(100, (move.bytes / move.total) * 100) : finishing || move?.stage === "saving" || move?.stage === "continuing" ? 100 : null; + const left = move?.stage === "copying" && move.total && rate ? (move.total - move.bytes) / rate : null; + + let detail: string; + if (finishing) detail = "Almost there"; + else if (!move) detail = "Starting"; + else if (move.stage === "checking") detail = `Checking the chat on ${move.from}`; + else if (move.stage === "copying") detail = move.total ? `Copying the conversation · ${size(move.bytes)} of ${size(move.total)}` : `Copying the conversation · ${size(move.bytes)}`; + else if (move.stage === "sending") detail = `Sending it to ${move.to}`; + else if (move.stage === "saving") detail = `Saving it on ${dest}`; + else detail = `${agentName} picks up where it stopped`; + + return ( +
+
+ + + +
+

+ Moving chat to {dest} + {move?.stage === "copying" && percent !== null && {Math.floor(percent)}%} +

+

{detail}

+
+ +
+
+
+
+
+

+ Keep working, you'll see when it's there. + {left !== null ? `${clock(left)} left · ${size(rate!)}/s` : clock(elapsed)} +

+
+
+ ); +} + +/** + * Moves a chat without holding anything up: the dialog closes and a card in the corner shows how far the move is + * (from the server's `chat.move` events) until it arrived or failed. + */ +export function startChatMove( + qc: QueryClient, + opts: { conversationId: string; input: MoveChatInput; from: string; dest: string; agentName: string; working: boolean }, +) { + const { conversationId, dest, agentName, working } = opts; + const id = `chat-move:${conversationId}`; + setChatMove(conversationId, { conversationId, from: opts.from, to: dest, stage: "checking", bytes: 0, total: null, startedAt: new Date().toISOString() }); + toast.custom((t) => , { id, duration: Infinity, unstyled: true }); + api.conversations.move(conversationId, opts.input).then( + (moved) => { + setChatMove(conversationId, null); + qc.setQueryData(qk.conversation(conversationId), (old) => (old ? { ...old, ...moved } : old)); + void qc.invalidateQueries({ queryKey: qk.conversation(conversationId) }); + void qc.invalidateQueries({ queryKey: qk.conversationsAll }); + void qc.invalidateQueries({ queryKey: qk.runners }); + toast.dismiss(id); + toast.success(`Moved to ${dest}`, { + id: `${id}:done`, + description: working ? `${agentName} continues there where it stopped.` : `${agentName} continues this chat there, with the whole conversation.`, + }); + }, + (err) => { + setChatMove(conversationId, null); + toast.dismiss(id); + toastApiError(err, "Couldn't move the chat", qc); + }, + ); +} diff --git a/apps/desktop/src/components/runners/move-chat.tsx b/apps/desktop/src/components/runners/move-chat.tsx index a6b149ae..52fefdcb 100644 --- a/apps/desktop/src/components/runners/move-chat.tsx +++ b/apps/desktop/src/components/runners/move-chat.tsx @@ -1,8 +1,7 @@ import { useEffect, useRef, useState } from "react"; -import { useMutation, useQueryClient } from "@tanstack/react-query"; -import type { Conversation, ConversationWithMessages } from "@godmode/shared"; +import { useQueryClient } from "@tanstack/react-query"; +import type { Conversation } from "@godmode/shared"; import { ArrowRight, Laptop, MonitorSmartphone } from "lucide-react"; -import { toast } from "sonner"; import { AlertDialog, AlertDialogCancel, @@ -15,11 +14,10 @@ import { import { Button } from "@/components/ui/button"; import { Spinner } from "@/components/ui/spinner"; import { followupWhen } from "@/components/chat/followup"; -import { toastApiError } from "@/components/vault/vault-utils"; -import { api } from "@/lib/api"; import { placeName, type Place } from "@/lib/places"; -import { qk } from "@/lib/queryKeys"; import { cn } from "@/lib/utils"; +import { useChatMoves } from "@/stores/chatMoves"; +import { startChatMove } from "./chat-move-progress"; import { RunnerStateDot, RunnerTag, THIS_COMPUTER_INLINE } from "./runner-parts"; /** Why a runner's chat can't move right now (its run works or stands still there), or null. */ @@ -47,8 +45,8 @@ function PlaceTag({ place, active }: { place: Place | null; active?: boolean }) } /** - * Asks before a chat moves to another computer, then moves it with its Claude session. While a run works the move - * stops it and the agent continues there (`working`). + * Asks before a chat moves to another computer, then moves it with its Claude session in the background (a card in the + * corner shows how far it is). While a run works the move stops it and the agent continues there (`working`). */ export function MoveChatDialog({ conversation, @@ -82,23 +80,15 @@ export function MoveChatDialog({ const name = origin ? placeName(origin) : "The other computer"; const away = !!origin && origin.state !== "online"; const dest = to?.here ? THIS_COMPUTER_INLINE : to ? placeName(to) : "the other computer"; - const move = useMutation({ - mutationFn: () => api.conversations.move(conversation.id, { to: to!.runnerId, force: working || undefined }), - onSuccess: (moved) => { - qc.setQueryData(qk.conversation(conversation.id), (old) => (old ? { ...old, ...moved } : old)); - void qc.invalidateQueries({ queryKey: qk.conversation(conversation.id) }); - void qc.invalidateQueries({ queryKey: qk.conversationsAll }); - void qc.invalidateQueries({ queryKey: qk.runners }); - onOpenChange(false); - toast.success(`Moved to ${dest}`, { - description: working ? `${agentName} continues there where it stopped.` : `${agentName} continues this chat there, with the whole conversation.`, - }); - }, - onError: (err) => toastApiError(err, "Couldn't move the chat", qc), - }); + const moving = useChatMoves((s) => !!s.moves[conversation.id]); + const move = () => { + if (!to) return; + startChatMove(qc, { conversationId: conversation.id, input: { to: to.runnerId, force: working || undefined }, from: name, dest, agentName, working }); + onOpenChange(false); + }; return ( - !move.isPending && onOpenChange(next)}> + Move this chat to {dest}? @@ -146,9 +136,9 @@ export function MoveChatDialog({ - Cancel - diff --git a/apps/desktop/src/lib/realtime.ts b/apps/desktop/src/lib/realtime.ts index fcbb2e54..d4548a5e 100644 --- a/apps/desktop/src/lib/realtime.ts +++ b/apps/desktop/src/lib/realtime.ts @@ -2,6 +2,7 @@ import type { InvalidateQueryFilters, QueryClient } from "@tanstack/react-query" import { browserView, type AgentQuestion, type AutomationEvent, type BrowserProfile, type ClientEvent, type ConversationWithMessages, type DeviceGroup, type EntityName, type Message, type RemoteRunner, type ServerEvent, type Task, type TaskEvent, type Vm } from "@godmode/shared"; import { wsUrl } from "./core"; import { useLive } from "@/stores/live"; +import { setChatMove, useChatMoves } from "@/stores/chatMoves"; import { withPending } from "./pending-queue"; import { qk } from "./queryKeys"; @@ -214,6 +215,8 @@ function handle(qc: QueryClient, event: ServerEvent) { case "hello": // A run.started for each active run follows; whatever else still looks live ended while we were away. if (event.activeRunIds) live.retainRuns(event.activeRunIds); + // Moves still on their way are told again right after. + useChatMoves.setState({ moves: {} }); break; case "run.started": live.runStarted(event.run); @@ -406,6 +409,9 @@ function handle(qc: QueryClient, event: ServerEvent) { qc.invalidateQueries({ queryKey: qk.runners }); qc.invalidateQueries({ queryKey: qk.devices }); break; + case "chat.move": + setChatMove(event.conversationId, event.move); + break; case "devices.updated": { const before = qc.getQueryData(qk.devices); qc.setQueryData(qk.devices, event.group); diff --git a/apps/desktop/src/stores/chatMoves.ts b/apps/desktop/src/stores/chatMoves.ts new file mode 100644 index 00000000..19eb9dfd --- /dev/null +++ b/apps/desktop/src/stores/chatMoves.ts @@ -0,0 +1,14 @@ +import { create } from "zustand"; +import type { ChatMove } from "@godmode/shared"; + +/** Chats on their way between computers, as the server tells them (`chat.move`). */ +export const useChatMoves = create<{ moves: Record }>(() => ({ moves: {} })); + +export function setChatMove(conversationId: string, move: ChatMove | null) { + useChatMoves.setState(({ moves }) => { + const next = { ...moves }; + if (move) next[conversationId] = move; + else delete next[conversationId]; + return { moves: next }; + }); +} diff --git a/packages/core/src/cloud/scope.ts b/packages/core/src/cloud/scope.ts index e8dd0585..83f1398d 100644 --- a/packages/core/src/cloud/scope.ts +++ b/packages/core/src/cloud/scope.ts @@ -305,6 +305,7 @@ const RULES: [methods: string, path: string, rule: Rule][] = [ ["GET|PUT", "/api/link/agents/:id/memory", refused(COMPUTER_ONLY)], ["POST", "/api/link/browser/:profileId/cookies", refused(COMPUTER_ONLY)], ["GET", "/api/link/conversations/:id/handover", refused(COMPUTER_ONLY)], + ["GET", "/api/link/conversations/:id/handover/size", refused(COMPUTER_ONLY)], ["POST", "/api/link/exec", refused(COMPUTER_ONLY)], ["POST", "/api/link/forget", refused(COMPUTER_ONLY)], ["PUT", "/api/link/update/chunk", refused(COMPUTER_ONLY)], diff --git a/packages/core/src/remote/channel.ts b/packages/core/src/remote/channel.ts index 283c415b..48ad4a2b 100644 --- a/packages/core/src/remote/channel.ts +++ b/packages/core/src/remote/channel.ts @@ -285,6 +285,13 @@ export class SecureChannel { } while (offset < bytes.byteLength); } + /** Bytes of streams still arriving: how far a large answer is. */ + get incoming(): number { + let size = 0; + for (const stream of this.streams.values()) size += stream.size; + return size; + } + /** Ends the link from this side. Also what the owner calls when the transport closed underneath. */ close(): void { this.shutdown(null); diff --git a/packages/core/src/remote/chatMoves.ts b/packages/core/src/remote/chatMoves.ts new file mode 100644 index 00000000..375f2313 --- /dev/null +++ b/packages/core/src/remote/chatMoves.ts @@ -0,0 +1,53 @@ +/** + * Chats on their way between computers (owner: remote): how far each one is, told to the windows as `chat.move` + * events so a move shows its progress on a slow connection instead of a dialog that stands still. + */ +import type { ChatMove, ChatMoveStage } from "@godmode/shared"; +import { bus } from "../events/bus"; +import { addWelcomeEvents } from "../server/ws"; +import { now } from "../util"; + +const EMIT_EVERY_MS = 250; + +const moves = new Map(); +const lastEmit = new Map(); + +function emit(conversationId: string, force = false): void { + const at = Date.now(); + if (!force && at - (lastEmit.get(conversationId) ?? 0) < EMIT_EVERY_MS) return; + lastEmit.set(conversationId, at); + bus.emit({ type: "chat.move", conversationId, move: moves.get(conversationId) ?? null }); +} + +addWelcomeEvents(() => [...moves.values()].map((move) => ({ type: "chat.move", conversationId: move.conversationId, move }))); + +/** Runs the move with its progress told; the windows learn it ended (arrived or failed) when `fn` settles. */ +export async function trackMove(conversationId: string, from: string, to: string, fn: () => Promise): Promise { + if (moves.has(conversationId)) return fn(); + moves.set(conversationId, { conversationId, from, to, stage: "checking", bytes: 0, total: null, startedAt: now() }); + emit(conversationId, true); + try { + return await fn(); + } finally { + moves.delete(conversationId); + emit(conversationId, true); + lastEmit.delete(conversationId); + } +} + +/** A new step of a tracked move; nothing for a chat that isn't moving through `trackMove`. */ +export function moveStage(conversationId: string, stage: ChatMoveStage, total: number | null = null): void { + const move = moves.get(conversationId); + if (!move) return; + moves.set(conversationId, { ...move, stage, bytes: 0, total }); + emit(conversationId, true); +} + +export function moveBytes(conversationId: string, bytes: number): void { + const move = moves.get(conversationId); + if (!move) return; + const next = move.total ? Math.min(bytes, move.total) : bytes; + if (next <= move.bytes) return; + moves.set(conversationId, { ...move, bytes: next }); + emit(conversationId); +} diff --git a/packages/core/src/remote/devices.ts b/packages/core/src/remote/devices.ts index f29eab66..1a7a5bc5 100644 --- a/packages/core/src/remote/devices.ts +++ b/packages/core/src/remote/devices.ts @@ -65,6 +65,7 @@ import { HttpError, badRequest, now, parseJson, sleep } from "../util"; import { remoteRuns } from "./activeRuns"; import { canonicalKey } from "./crypto"; import { readHandover, writeSession, writeTranscript } from "./handover"; +import { moveStage, trackMove } from "./chatMoves"; import { loadIdentity } from "./identity"; import { forgetController, linkServerPort, runnerInfo, setPhoneInfo, startLinkServer, stopLinkServer } from "./linkServer"; import { adoptChat, catchUp, fetchConversation, importChat } from "./mirror"; @@ -1338,7 +1339,7 @@ async function pushChat(conversationId: string, targetId: string, extra: Message handover: readHandover(conversationId), followup: followup ? { note: followup.note, dueAt: followup.dueAt } : null, }, - { timeoutMs: 180_000 }, + { timeoutMs: 15 * 60_000 }, ); const state: RunnerChatState = { running: false, paused: null, followup: followup ? { note: followup.note, dueAt: followup.dueAt, createdAt: followup.createdAt } : null }; sql("UPDATE conversations SET runner_id = ?, runner_state = ?, working_directory = NULL, computer_target = NULL, updated_at = ? WHERE id = ?", targetId, JSON.stringify(state), now(), conversationId); @@ -1429,20 +1430,27 @@ export async function moveChat(conversationId: string, input: MoveChatInput): Pr const queued = owner === null ? listQueue(conversationId).map((q) => q.content) : []; if (queued.length && !input.force) throw new HttpError(409, "Messages are waiting in this chat's queue. Send or remove them first, or move it anyway.", "chat_queued"); - if (working) await stopWork(conversationId, owner, dest); - if (queued.length) clearQueue(conversationId); const resume = working || queued.length > 0; - if (owner !== null) await moveChatHere(conversationId); - - const blocks = marker(from, dest, resume); - if (to !== null) { - await pushChat(conversationId, to, resume ? null : blocks); - if (resume) await sendThere(conversationId, to, { content: `Continued on ${dest}`, prompt: continuePrompt(from, dest, queued), marker: blocks }); - } else if (resume) { - await sendMessage(conversationId, { content: `Continued on ${dest}`, prompt: continuePrompt(from, dest, queued), marker: blocks, trigger: "chat" }); - } else { - addMessage({ conversationId, role: "system", content: "", blocks }); - } + await trackMove(conversationId, from, dest, async () => { + if (working) await stopWork(conversationId, owner, dest); + if (queued.length) clearQueue(conversationId); + if (owner !== null) await moveChatHere(conversationId); + + const blocks = marker(from, dest, resume); + if (to !== null) { + moveStage(conversationId, "sending"); + await pushChat(conversationId, to, resume ? null : blocks); + if (resume) { + moveStage(conversationId, "continuing"); + await sendThere(conversationId, to, { content: `Continued on ${dest}`, prompt: continuePrompt(from, dest, queued), marker: blocks }); + } + } else if (resume) { + moveStage(conversationId, "continuing"); + await sendMessage(conversationId, { content: `Continued on ${dest}`, prompt: continuePrompt(from, dest, queued), marker: blocks, trigger: "chat" }); + } else { + addMessage({ conversationId, role: "system", content: "", blocks }); + } + }); log.info("moved a chat", { conversationId, from, to: dest, resume }); audit("user", "device.chat.move", conversationId, { from, to: dest, resume }); emitConversationUpdated(conversationId); diff --git a/packages/core/src/remote/linkClient.ts b/packages/core/src/remote/linkClient.ts index ab719026..76d6fa90 100644 --- a/packages/core/src/remote/linkClient.ts +++ b/packages/core/src/remote/linkClient.ts @@ -418,6 +418,11 @@ export class RemoteLink { if (!this.current || this._state.state !== "online") throw this.offlineError(); } + /** Bytes of answers still arriving from the runner (a chat's handover on a slow connection). */ + incomingBytes(): number { + return this.current?.channel.incoming ?? 0; + } + /** A request into the runner's API. Throws `runner_offline` when the link is down. */ request( method: string, diff --git a/packages/core/src/remote/runners.ts b/packages/core/src/remote/runners.ts index 5c71a3ea..551d5005 100644 --- a/packages/core/src/remote/runners.ts +++ b/packages/core/src/remote/runners.ts @@ -63,6 +63,7 @@ import { loadIdentity } from "./identity"; import { RemoteLink, type LinkState } from "./linkClient"; import { mergeMemory, readMemoryState, writeMemoryState } from "./memorySync"; import { parseHandover, writeSession, writeTranscript, type ChatHandover } from "./handover"; +import { moveBytes, moveStage } from "./chatMoves"; import { adoptChat, applyRunnerEvent, catchUp, chatMovedHere, forgetHeavyChats, reconcileConversation, runnerDisconnected, setMirrorHooks } from "./mirror"; import { cancelOffer, createOffer } from "./pairing"; import { forgetRunnerUpdates, maybeAutoUpdate, prepareRunnerUpdates, resetRunnerUpdates, runnerConnected, runnerUpdate, setUpdateHost, updateRunnerSoftware } from "./runnerUpdates"; @@ -991,6 +992,9 @@ function offline(err: unknown): boolean { return err instanceof HttpError && err.code === "runner_offline"; } +/** A session of up to 64 MB on a slow connection: the move shows how far it is meanwhile. */ +const HANDOVER_TIMEOUT_MS = 15 * 60_000; + /** Chats on their way here: nothing new starts in them on the runner meanwhile (routing.ts). */ const moving = new Set(); @@ -1022,11 +1026,20 @@ async function handoverFrom(r: RunnerRow, conversationId: string): Promise("GET", `/api/runs?conversationId=${path}&limit=500`).catch(() => undefined); const chat = reconcileConversation(r.id, remote, runs); if (chat.activeRunId || chat.running || chat.paused) throw busyOnRunner(r); + // A runner from before the size can't say how much comes: the move shows the bytes without a total. + const size = await l.json<{ bytes?: unknown }>("GET", `/api/link/conversations/${path}/handover/size`).catch(() => null); + moveStage(conversationId, "copying", typeof size?.bytes === "number" && size.bytes > 0 ? size.bytes : null); + const ticker = setInterval(() => moveBytes(conversationId, l.incomingBytes()), 250); + ticker.unref?.(); // A runner from before moving chats has no handover: the next turn here starts a new session with a recap. - return await l.json("GET", `/api/link/conversations/${path}/handover`, undefined, { timeoutMs: 120_000 }).then(parseHandover, (err) => { - if (gone(err)) return null; - throw err; - }); + try { + return await l.json("GET", `/api/link/conversations/${path}/handover`, undefined, { timeoutMs: HANDOVER_TIMEOUT_MS }).then(parseHandover, (err) => { + if (gone(err)) return null; + throw err; + }); + } finally { + clearInterval(ticker); + } } catch (err) { if (offline(err)) throw new HttpError(409, `${r.name} went away while the chat was moving. Nothing changed — try again.`, "runner_offline"); if (err instanceof HttpError && err.code === "runner_chat_busy") throw err; @@ -1068,6 +1081,7 @@ export async function moveChatHere(conversationId: string): Promise x.conversationId === conversationId)) throw busyOnRunner(r); + moveStage(conversationId, "saving"); const agent = getAgent(conv.agent_id); const session = !!handover?.sessionId && !!handover.session && writeSession(agent.workingDirectory ?? agent.repoPath, handover.sessionId, handover.session); if (handover?.transcript) writeTranscript(transcriptPath(agent, conversationId), handover.transcript); diff --git a/packages/core/src/server/routes/link.ts b/packages/core/src/server/routes/link.ts index 59650142..682c0bc2 100644 --- a/packages/core/src/server/routes/link.ts +++ b/packages/core/src/server/routes/link.ts @@ -80,6 +80,7 @@ export function registerLinkRoutes(app: Hono): void { // A chat of this runner moves to the controller: its Claude session and transcript go along. app.get("/api/link/conversations/:id/handover", (c) => c.json(readHandover(c.req.param("id")))); + app.get("/api/link/conversations/:id/handover/size", (c) => c.json({ bytes: Buffer.byteLength(JSON.stringify(readHandover(c.req.param("id")))) })); // The sessions the controller copies here (remote/browserSessions.ts); one from before sends no mode and replaces. app.post("/api/link/browser/:profileId/cookies", async (c) => { diff --git a/packages/core/test/chat-moves.test.ts b/packages/core/test/chat-moves.test.ts new file mode 100644 index 00000000..81cd35a2 --- /dev/null +++ b/packages/core/test/chat-moves.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, test } from "bun:test"; +import type { ChatMove, ServerEvent } from "@godmode/shared"; +import { bus } from "../src/events/bus"; +import { moveBytes, moveStage, trackMove } from "../src/remote/chatMoves"; + +function listen(): { moves: (ChatMove | null)[]; stop: () => void } { + const moves: (ChatMove | null)[] = []; + const stop = bus.on((e: ServerEvent) => { + if (e.type === "chat.move") moves.push(e.move); + }); + return { moves, stop }; +} + +describe("chat moves", () => { + test("a move tells each step and the bytes of the copy, then that it ended", async () => { + const { moves, stop } = listen(); + await trackMove("c1", "Laptop", "this Mac", async () => { + moveStage("c1", "copying", 1000); + moveBytes("c1", 400); + await Bun.sleep(300); + moveBytes("c1", 5000); + moveStage("c1", "saving"); + }); + stop(); + expect(moves.map((m) => m?.stage ?? null)).toEqual(["checking", "copying", "copying", "saving", null]); + const copied = moves.filter((m) => m?.stage === "copying" && m.bytes > 0); + expect(copied.at(-1)).toMatchObject({ from: "Laptop", to: "this Mac", bytes: 1000, total: 1000 }); + }); + + test("a failed move still tells that it ended, and progress outside a move is ignored", async () => { + const { moves, stop } = listen(); + moveStage("c2", "copying", 10); + await expect(trackMove("c2", "a", "b", async () => Promise.reject(new Error("gone")))).rejects.toThrow("gone"); + moveBytes("c2", 5); + stop(); + expect(moves.map((m) => m?.stage ?? null)).toEqual(["checking", null]); + }); + + test("bytes in quick succession are told a few times a second, not each time", async () => { + const { moves, stop } = listen(); + await trackMove("c3", "a", "b", async () => { + moveStage("c3", "copying", null); + for (let i = 1; i <= 50; i++) moveBytes("c3", i * 100); + }); + stop(); + expect(moves.length).toBeLessThan(6); + }); +}); diff --git a/packages/shared/src/devices.ts b/packages/shared/src/devices.ts index d146214d..1d3cc446 100644 --- a/packages/shared/src/devices.ts +++ b/packages/shared/src/devices.ts @@ -138,6 +138,22 @@ export interface MoveChatInput { force?: boolean; } +/** + * How far a chat on its way to another computer is: talking to where it is, copying its session and transcript + * (`bytes` of `total`, null when the other computer can't say), sending it on, saving it, continuing the run. + */ +export type ChatMoveStage = "checking" | "copying" | "sending" | "saving" | "continuing"; + +export interface ChatMove { + conversationId: ID; + from: string; + to: string; + stage: ChatMoveStage; + bytes: number; + total: number | null; + startedAt: string; +} + /** POST /api/devices/handoff */ export interface HandoffInput { /** The device whose chats move (its runner id), or null for the primary. */ diff --git a/packages/shared/src/events.ts b/packages/shared/src/events.ts index 436cc901..cd2da03c 100644 --- a/packages/shared/src/events.ts +++ b/packages/shared/src/events.ts @@ -3,7 +3,7 @@ import type { Task, TaskEvent } from "./tasks"; import type { Vm } from "./vm"; import type { MobileDevice } from "./mobile"; import type { RemoteRunner } from "./remote"; -import type { DeviceGroup } from "./devices"; +import type { ChatMove, DeviceGroup } from "./devices"; import type { Agent, AgentQuestion, @@ -170,6 +170,8 @@ export type ServerEvent = * the primary's Godmode (or its own while the primary is away), so everything it shows is fetched again. */ | { type: "devices.updated"; group: DeviceGroup } + /** A chat moving between computers made progress (`move`), or arrived or failed (`move: null`). */ + | { type: "chat.move"; conversationId: ID; move: ChatMove | null } /** Between the computers of a device group only: shared data changed here, the other one pulls it. */ | { type: "replica.changed"; seq: number } | { type: "entity.changed"; entity: EntityName };