Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions packages/core/src/remote/snapshot.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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] ?? [])];
Expand Down Expand Up @@ -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<string, Set<string>>();
const touch = (spec: TableSpec, key: string) => {
changed.add(spec.entity);
touched.get(spec.table)!.add(key);
Expand Down Expand Up @@ -321,6 +333,7 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str
const rows: Record<string, Value>[] = [];
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;
Expand All @@ -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".
Expand Down Expand Up @@ -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<Record<string, unknown>>(`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) ?? {};
Expand All @@ -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;
Expand Down
68 changes: 68 additions & 0 deletions packages/core/test/remote-snapshot.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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));
Expand Down
Loading