diff --git a/packages/core/src/remote/replica.ts b/packages/core/src/remote/replica.ts index a0753125..4010b89c 100644 --- a/packages/core/src/remote/replica.ts +++ b/packages/core/src/remote/replica.ts @@ -19,7 +19,7 @@ import { logger } from "../log"; import { now } from "../util"; import { listActiveRuns } from "../runner/runner"; import { isSecondary, primaryRunnerId } from "./group"; -import { LOCAL_SETTINGS, SETUP_TABLES, unsafe, withOwnFields } from "./snapshot"; +import { freeSlug, LOCAL_SETTINGS, SETUP_TABLES, SLUG_SCOPES, slugHolder, unsafe, withOwnFields } from "./snapshot"; const log = logger("replica"); @@ -241,6 +241,18 @@ export function asPeer(origin: string, at: string | null, fn: () => T): T { } } +/** Run `fn` as this computer's own change, also inside `asPeer`: what it writes goes to the other computers. */ +export function asOwn(fn: () => T): T { + const ctx = get<{ origin: string | null; at: string | null }>("SELECT origin, at FROM replica_ctx WHERE id = 1"); + if (!ctx) return fn(); + sql("UPDATE replica_ctx SET origin = NULL, at = NULL WHERE id = 1"); + try { + return fn(); + } finally { + sql("UPDATE replica_ctx SET origin = ?, at = ? WHERE id = 1", ctx.origin, ctx.at); + } +} + /** The last change this computer counted (local or taken from another one). */ export function replicaSeq(): number { try { @@ -574,6 +586,11 @@ function applyOne(peer: string, c: ReplicaChange, opts: ApplyOptions, incoming: opts.onMissing?.(missing.table, missing.id); return; } + // Both computers made "earshot" while apart: the one that comes second here becomes "earshot-2" (the setup copy + // takes the new slug back to the peer). Only once the whole pull is in: a slug may be free by its end (a rename). + if (opts.retry && SLUG_SCOPES[t.table] && t.key.length === 1 && t.key[0] === "id" && row.slug != null && slugHolder(t.table, row, c.k)) { + row.slug = freeSlug(t.table, row, c.k); + } const isNew = !get(`SELECT 1 FROM ${q(t.table)} WHERE ${keyWhere(t)}`, ...parts); const extra = isNew ? (t.onInsert?.(row) ?? {}) : {}; const names = Object.keys(row); diff --git a/packages/core/src/remote/runners.ts b/packages/core/src/remote/runners.ts index ffc2ec20..1bc649cb 100644 --- a/packages/core/src/remote/runners.ts +++ b/packages/core/src/remote/runners.ts @@ -747,9 +747,10 @@ function logFailure(text: string, err: unknown): void { /** * Bring a runner up to date: the setup when it differs (or `force`), then — for a chat that starts — the agent's memory - * and the browser sessions of the profile it uses. Returns what was done; a failed setup copy throws. + * and the browser sessions of the profile it uses. Returns what was done; a failed setup copy throws, except for `work` + * on a runner that has an earlier setup and the vault key. */ -export function syncRunner(id: string, opts: { agentId?: string; profileId?: string | null; force?: boolean } = {}): Promise<{ config: boolean; memory: boolean; cookies: boolean; warnings: string[] }> { +export function syncRunner(id: string, opts: { agentId?: string; profileId?: string | null; force?: boolean; work?: boolean } = {}): Promise<{ config: boolean; memory: boolean; cookies: boolean; warnings: string[] }> { return serialized(id, async () => { const l = link(id); l.assertOnline(); @@ -784,7 +785,11 @@ export function syncRunner(id: string, opts: { agentId?: string; profileId?: str } } catch (err) { update("runners", id, { sync_error: `Couldn't copy your setup: ${message(err)}`, updated_at: now() }); - throw Object.assign(new HttpError(502, `Couldn't copy your setup to ${r.name}: ${message(err)}`, "runner_sync_failed"), { cause: err }); + const failure = Object.assign(new HttpError(502, `Couldn't copy your setup to ${r.name}: ${message(err)}`, "runner_sync_failed"), { cause: err }); + // Work goes ahead on the setup the runner has (the runner card shows why) — not when it never got one or lacks the key. + if (!opts.work || needsKey || !r.synced_at || linkGone(err)) throw failure; + warnings.push(failure.message); + logFailure(failure.message, err); } finally { syncing.delete(id); emit(id); @@ -926,7 +931,7 @@ export async function startRemoteChat(runnerId: string, input: StartChatInput): profileId = null; } } - await syncRunner(runnerId, { agentId, profileId }); + await syncRunner(runnerId, { agentId, profileId, work: true }); const { runnerId: _runner, workingDirectory, computerTarget, vmId, androidDeviceIds, ...rest } = input; // Another computer of the group is where the app that asked runs: its folders, screen, VMs and phones are named there. const own = row(runnerId)?.kind === "device" ? { workingDirectory, computerTarget, vmId, androidDeviceIds } : {}; @@ -949,7 +954,7 @@ export async function prepareRemoteMessage(conversationId: string): Promise = { workspaces: [], projects: ["workspace_id"] }; + +const SLUG_SEP = "\u001f"; + +export function slugKey(table: string, row: Record, slug: unknown = row.slug): string { + return [...SLUG_SCOPES[table]!.map((c) => String(row[c] ?? "")), String(slug)].join(SLUG_SEP); +} + +/** The id of another row of `table` that holds `row`'s slug. */ +export function slugHolder(table: string, row: Record, id: string, slug: unknown = row.slug): string | null { + const scope = SLUG_SCOPES[table]!; + const where = [...scope, "slug"].map((c) => `${q(c)} IS ?`).join(" AND "); + const params = [...scope.map((c) => (row[c] ?? null) as Value), slug as Value]; + return get<{ id: string }>(`SELECT id FROM ${q(table)} WHERE ${where} AND id != ?`, ...params, id)?.id ?? null; +} + +/** `row`'s slug, or the first "slug-2", "slug-3"… no other row holds and `claimed` doesn't name. */ +export function freeSlug(table: string, row: Record, id: string, claimed: Set = new Set()): string { + const base = String(row.slug); + let candidate = base; + for (let i = 2; claimed.has(slugKey(table, row, candidate)) || slugHolder(table, row, id, candidate); i++) candidate = `${base}-${i}`; + return candidate; +} + /** Why a row can't be copied: its slug or id becomes a folder name on the runner (an agent's id does too, for its browser files). */ export function unsafe(table: string, row: Record): string | null { const label = typeof row.name === "string" && row.name ? row.name : String(row.id); @@ -358,6 +386,25 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str } } + if (SLUG_SCOPES[spec.table] && pk === "id") { + const written = new Set(rows.map((r) => String(r.id)).filter((k) => !ownChanges.has(k))); + const claimed = new Map(rows.filter((r) => written.has(String(r.id))).map((r) => [slugKey(spec.table, r), String(r.id)])); + for (const local of all>(`SELECT * FROM ${q(spec.table)}`)) { + const key = String(local.id); + const holder = claimed.get(slugKey(spec.table, local)); + if (!holder || holder === key) continue; + if (written.has(key)) { + // It gets its new slug from this copy too, maybe later in the loop: out of the way until then. + run(`UPDATE ${q(spec.table)} SET slug = ? WHERE id = ?`, `${String(local.slug)}~${key}`, key); + continue; + } + // Made here and not with the primary yet: it yields, and the new slug goes to the primary as this computer's change. + const slug = freeSlug(spec.table, local, key, new Set(claimed.keys())); + asOwn(() => run(`UPDATE ${q(spec.table)} SET slug = ? WHERE id = ?`, slug, key)); + touch(spec, key); + } + } + for (const row of rows) { const key = String(row[pk]); if (ownChanges.has(key)) { diff --git a/packages/core/test/remote-snapshot.test.ts b/packages/core/test/remote-snapshot.test.ts index 8aa756f8..9fe8d78a 100644 --- a/packages/core/test/remote-snapshot.test.ts +++ b/packages/core/test/remote-snapshot.test.ts @@ -12,6 +12,8 @@ import { setLogLevel } from "../src/log"; import type { MemorySnapshot } from "../src/memory/files"; import { mergeMemory, readMemoryState, writeMemoryState } from "../src/remote/memorySync"; import { appliedDigest, applySnapshot, buildSnapshot, snapshotDigest, type ConfigSnapshot } from "../src/remote/snapshot"; +import { clearGroup, setGroup } from "../src/remote/group"; +import { ensureReplica, madeHere, pendingKeys, resetReplicaCache } from "../src/remote/replica"; import * as runner from "../src/runner/runner"; import { getSettings, resetSettingsCache, updateSettings } from "../src/services/settings"; import { HttpError } from "../src/util"; @@ -613,6 +615,52 @@ describe("applying a snapshot on a runner", () => { expect(get("SELECT 1 FROM settings WHERE key = 'future'")).toBeNull(); }); + test("projects that swapped slugs on the controller swap on the runner too", async () => { + await sync(() => { + insert("workspaces", { id: "ws_apps", name: "Apps", slug: "apps", created_at: TS, updated_at: TS }); + insert("projects", { id: "prj_one", workspace_id: "ws_apps", name: "One", slug: "alpha", created_at: TS, updated_at: TS }); + insert("projects", { id: "prj_two", workspace_id: "ws_apps", name: "Two", slug: "beta", created_at: TS, updated_at: TS }); + }); + const result = await sync(() => { + sql("UPDATE projects SET slug = 'swap' WHERE id = 'prj_one'"); + sql("UPDATE projects SET slug = 'alpha' WHERE id = 'prj_two'"); + sql("UPDATE projects SET slug = 'beta' WHERE id = 'prj_one'"); + }); + expect(result.warnings).toEqual([]); + expect(row("projects", "prj_one").slug).toBe("beta"); + expect(row("projects", "prj_two").slug).toBe("alpha"); + }); + + test("a project a secondary made while apart yields its slug to the primary's and still goes to the primary", async () => { + const secondary = mkdtempSync(join(tmpdir(), "godmode-snapshot-secondary-")); + try { + use(secondary); + setGroup("secondary", "rnr_primary"); + resetReplicaCache(); + ensureReplica(); + use(controller); + const before = buildSnapshot(); + use(secondary); + await applySnapshot(before); + insert("projects", { id: "prj_mine", workspace_id: "ws_apps", name: "Earshot", slug: "earshot", created_at: LATER, updated_at: LATER }); + use(controller); + insert("projects", { id: "prj_theirs", workspace_id: "ws_apps", name: "Earshot", slug: "earshot", created_at: LATER, updated_at: LATER }); + const snapshot = buildSnapshot(); + use(secondary); + expect((await applySnapshot(snapshot)).warnings).toEqual([]); + expect(row("projects", "prj_theirs").slug).toBe("earshot"); + expect(row("projects", "prj_mine").slug).toBe("earshot-2"); + expect(pendingKeys("rnr_primary", "projects").has("prj_mine")).toBe(true); + expect(madeHere("projects").has("prj_mine")).toBe(true); + } finally { + clearGroup(); + resetReplicaCache(); + use(controller); + sql("DELETE FROM projects WHERE id = 'prj_theirs'"); + rmSync(secondary, { recursive: true, force: true }); + } + }); + test("unsafe ids are skipped with a warning and nothing is written outside the data directory", async () => { use(controller); const snapshot = buildSnapshot(); diff --git a/packages/core/test/replica.test.ts b/packages/core/test/replica.test.ts index 48a598c4..9f787441 100644 --- a/packages/core/test/replica.test.ts +++ b/packages/core/test/replica.test.ts @@ -248,6 +248,30 @@ describe("between two computers", () => { expect(pendingFor("rnr_b")).toBe(0); }); + test("a project both computers made while apart lands with its own slug instead of failing", () => { + use(b); + run("DELETE FROM projects WHERE workspace_id = 'wsp_apps'"); + run("DELETE FROM workspaces WHERE id = 'wsp_apps'"); + run("INSERT INTO workspaces (id, name, slug, created_at, updated_at) VALUES ('wsp_apps', 'Apps', 'apps', ?, ?)", T(0), T(0)); + run("INSERT INTO projects (id, workspace_id, name, slug, created_at, updated_at) VALUES ('prj_here', 'wsp_apps', 'Earshot', 'earshot', ?, ?)", T(1), T(1)); + const theirs: ReplicaChange = { + t: "projects", + k: "prj_there", + at: T(30), + row: { id: "prj_there", workspace_id: "wsp_apps", name: "Earshot", slug: "earshot", created_at: T(30), updated_at: T(30) }, + }; + // Within the pull the slug may still be freed by a rename that comes later: it waits for the end of the pull. + expect(applyChanges("rnr_a", [theirs], { peerWinsTies: false }).failed).toBe(1); + const res = retryDeferred("rnr_a", { peerWinsTies: false }); + expect(res.applied).toBe(1); + expect(res.failed).toBe(0); + expect(all("SELECT id, slug FROM projects WHERE workspace_id = 'wsp_apps' ORDER BY id")).toEqual([ + { id: "prj_here", slug: "earshot" }, + { id: "prj_there", slug: "earshot-2" }, + ]); + expect(all("SELECT * FROM replica_deferred")).toHaveLength(0); + }); + test("an agent whose folder name isn't safe never lands", () => { use(b); const evil: ReplicaChange = { t: "agents", k: "agt_evil", at: T(20), row: { id: "agt_evil", name: "Evil", slug: "../../outside", created_at: T(0), updated_at: T(0) } };