diff --git a/packages/core/src/remote/snapshot.ts b/packages/core/src/remote/snapshot.ts index 106d5b6f..7ba19bcf 100644 --- a/packages/core/src/remote/snapshot.ts +++ b/packages/core/src/remote/snapshot.ts @@ -129,6 +129,14 @@ function columnsOf(table: string): { name: string; pk: number }[] { return all<{ name: string; pk: number }>(`PRAGMA table_info(${q(table)})`); } +/** The columns of each UNIQUE index of a table besides its primary key (a slug, a name). */ +function uniqueKeysOf(table: string): string[][] { + return all<{ name: string; unique: number; origin: string; partial: number }>(`PRAGMA index_list(${q(table)})`) + .filter((i) => i.unique === 1 && i.origin !== "pk" && !i.partial) + .map((i) => all<{ name: string | null }>(`PRAGMA index_info(${q(i.name)})`).map((c) => c.name)) + .filter((cols): cols is string[] => cols.length > 0 && cols.every((c) => c !== null)); +} + /** Fields of a settings section that keep the runner's value: they name programs and endpoints on that computer. */ function keptFields(section: string): string[] { return [...(EXECUTABLE_SETTINGS[section] ?? []), ...(RUNNER_SETTINGS[section] ?? [])]; @@ -273,6 +281,10 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str const removedAgents: { slug: string; repoPath: string }[] = []; /** An agent with an active run stood in the way: this sync isn't recorded as applied, so the next one tries again. */ let pending = false; + /** Rows of the copy left out for a row of this computer with the same name, and that row. */ + const sameName: { table: string; from: string; to: string }[] = []; + /** Per table, the rows this copy wrote (not this computer's own). */ + const copied = new Map>(); const touch = (spec: TableSpec, key: string) => { changed.add(spec.entity); touched.get(spec.table)!.add(key); @@ -321,6 +333,7 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str const rows: Record[] = []; const ownChanges = mineNow(spec.table); const ownRows = madeByMe(spec.table); + const uniqueKeys = !primary || spec.table === "agents" ? [] : uniqueKeysOf(spec.table); for (const raw of incoming as unknown[]) { if (!isObject(raw) || typeof raw[pk] !== "string" || !raw[pk]) continue; if (spec.only && raw[spec.only.column] !== spec.only.value) continue; @@ -329,6 +342,7 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str if (problem) warnings.push(problem); else rows.push(Object.fromEntries(Object.entries(raw).map(([k, v]) => [k, decodeValue(v)]))); } + copied.set(spec.table, new Set([...keys].filter((k) => !ownChanges.has(k) && !ownRows.has(k)))); // Delete before insert: a row that goes may hold a slug (UNIQUE) a row that comes needs — a runner's own default // agent and the controller's are both "godmode". @@ -375,6 +389,17 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str const dir = row.working_directory; if (dir != null && (typeof dir !== "string" || workingDirectoryProblem(dir))) row.working_directory = null; } + // A row this computer made itself holds the same name (both made "earshot" while apart): it stays, and what the + // copy hangs on the other one moves to it, so the rest of the copy still lands. + const holder = uniqueKeys + .filter((cols) => cols.every((c) => row[c] != null)) + .map((cols) => get>(`SELECT ${q(pk)} AS k FROM ${q(spec.table)} WHERE ${cols.map((c) => `${q(c)} = ?`).join(" AND ")} AND ${q(pk)} != ?`, ...cols.map((c) => row[c]!), key)) + .find((h) => h && (ownRows.has(String(h.k)) || ownChanges.has(String(h.k)))); + if (holder) { + sameName.push({ table: spec.table, from: key, to: String(holder.k) }); + warnings.push(`"${String(row.name ?? key)}" (${spec.table}) wasn't copied: this computer has its own one with the same name.`); + continue; + } // Columns this schema doesn't have are ignored; columns the snapshot lacks keep their default. const names = mirrored.filter((n) => n in row); const extra = spec.onInsert?.(row) ?? {}; @@ -393,6 +418,29 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str } } + // Only what the copy wrote moves, and only off a row that isn't here; one that can't move (its own name is taken + // there) is left to the orphan pass below. + for (const { table, from, to } of sameName) { + const parentPk = columnsOf(table).find((c) => c.pk)?.name; + if (!parentPk || get(`SELECT 1 FROM ${q(table)} WHERE ${q(parentPk)} = ?`, from)) continue; + for (const spec of TABLES) { + const pk = columnsOf(spec.table).find((c) => c.pk)?.name; + const mine = copied.get(spec.table); + if (!pk || !mine) continue; + for (const fk of all<{ table: string; from: string }>(`PRAGMA foreign_key_list(${q(spec.table)})`)) { + if (fk.table !== table) continue; + for (const child of all<{ k: string }>(`SELECT ${q(pk)} AS k FROM ${q(spec.table)} WHERE ${q(fk.from)} = ?`, from)) { + if (!mine.has(child.k)) continue; + try { + if (run(`UPDATE ${q(spec.table)} SET ${q(fk.from)} = ? WHERE ${q(pk)} = ?`, to, child.k).changes > 0) changed.add(spec.entity); + } catch (err) { + log.warn(`could not move ${spec.table} ${child.k} to ${table} ${to}`, err instanceof Error ? err.message : err); + } + } + } + } + } + // What belongs to an agent that wasn't copied (see the warnings) would fail the commit: leave it out too. Agents // themselves are never dropped here — their chats would go with them; a snapshot that broken fails as a whole. let orphans = 0; diff --git a/packages/core/test/remote-snapshot.test.ts b/packages/core/test/remote-snapshot.test.ts index 8aa756f8..606363a5 100644 --- a/packages/core/test/remote-snapshot.test.ts +++ b/packages/core/test/remote-snapshot.test.ts @@ -10,7 +10,9 @@ import { all, closeDb, get, getMeta, insert, openDb, run as sql } from "../src/d import { bus } from "../src/events/bus"; import { setLogLevel } from "../src/log"; import type { MemorySnapshot } from "../src/memory/files"; +import { clearGroup, setGroup } from "../src/remote/group"; import { mergeMemory, readMemoryState, writeMemoryState } from "../src/remote/memorySync"; +import { ensureReplica, resetReplicaCache } from "../src/remote/replica"; import { appliedDigest, applySnapshot, buildSnapshot, snapshotDigest, type ConfigSnapshot } from "../src/remote/snapshot"; import * as runner from "../src/runner/runner"; import { getSettings, resetSettingsCache, updateSettings } from "../src/services/settings"; @@ -676,6 +678,72 @@ describe("applying a snapshot on a runner", () => { expect(vault.isUnlocked()).toBe(true); }); + test("a secondary keeps a project it made with the same name as one of the primary's, and the rest of the copy lands", async () => { + const secondary = mkdtempSync(join(tmpdir(), "godmode-snapshot-secondary-")); + try { + use(controller); + insert("workspaces", { id: "ws_ears", name: "Ears", slug: "ears", created_at: TS, updated_at: TS }); + const firstCopy = buildSnapshot(); + use(secondary); + resetReplicaCache(); + ensureReplica(); + setGroup("secondary", "rnr_primary"); + await applySnapshot(firstCopy); + // Made here while the primary was away. + insert("projects", { id: "prj_here", workspace_id: "ws_ears", name: "Earshot", slug: "earshot", created_at: TS, updated_at: TS }); + + use(controller); + insert("projects", { id: "prj_there", workspace_id: "ws_ears", name: "Earshot", slug: "earshot", created_at: LATER, updated_at: LATER }); + sql("UPDATE agents SET project_id = 'prj_there', instructions = 'Work on Earshot.' WHERE id = 'agt_helper'"); + const copy = buildSnapshot(); + sql("UPDATE agents SET project_id = NULL, instructions = 'Answer briefly.' WHERE id = 'agt_helper'"); + sql("DELETE FROM workspaces WHERE id = 'ws_ears'"); + + use(secondary); + const result = await applySnapshot(copy); + expect(result.warnings).toEqual(['"Earshot" (projects) wasn\'t copied: this computer has its own one with the same name.']); + expect(exists("projects", "prj_here")).toBe(true); + expect(exists("projects", "prj_there")).toBe(false); + expect(row("agents", "agt_helper")).toMatchObject({ project_id: "prj_here", instructions: "Work on Earshot." }); + expect(appliedDigest()).toBe(copy.digest); + } finally { + clearGroup(); + use(runnerDir); + rmSync(secondary, { recursive: true, force: true }); + } + }); + + test("a workspace and a project both made on each side with the same names don't stop the copy", async () => { + const secondary = mkdtempSync(join(tmpdir(), "godmode-snapshot-secondary-")); + try { + use(secondary); + resetReplicaCache(); + ensureReplica(); + setGroup("secondary", "rnr_primary"); + insert("workspaces", { id: "ws_here", name: "Studio", slug: "studio", created_at: TS, updated_at: TS }); + insert("projects", { id: "prj_here", workspace_id: "ws_here", name: "Earshot", slug: "earshot", created_at: TS, updated_at: TS }); + + use(controller); + insert("workspaces", { id: "ws_there", name: "Studio", slug: "studio", created_at: LATER, updated_at: LATER }); + insert("projects", { id: "prj_there", workspace_id: "ws_there", name: "Earshot", slug: "earshot", created_at: LATER, updated_at: LATER }); + const copy = buildSnapshot(); + sql("DELETE FROM workspaces WHERE id = 'ws_there'"); + + use(secondary); + const result = await applySnapshot(copy); + expect(result.warnings.join("\n")).toContain('"Studio" (workspaces) wasn\'t copied'); + expect(row("projects", "prj_here")).toMatchObject({ workspace_id: "ws_here", slug: "earshot" }); + expect(exists("workspaces", "ws_there")).toBe(false); + expect(exists("projects", "prj_there")).toBe(false); + expect(exists("agents", "agt_helper")).toBe(true); + expect(appliedDigest()).toBe(copy.digest); + } finally { + clearGroup(); + use(runnerDir); + rmSync(secondary, { recursive: true, force: true }); + } + }); + test("the vault refuses a key that doesn't open its canary", async () => { const meta = vault.vaultMetaForBackup() as { kdf: string; wrappedDek: string; canary: string }; const err = await rejection(vault.adoptKey(randomKey().toString("base64"), meta));