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
10 changes: 9 additions & 1 deletion packages/core/src/remote/replica.ts
Original file line number Diff line number Diff line change
Expand Up @@ -459,6 +459,8 @@ function emptyResult(): ApplyResult {
return { applied: 0, kept: 0, deferred: 0, failed: 0, entities: new Set(), agents: new Set(), removedAgents: [], keys: {} };
}

const stuckWarned = new Set<string>();

/**
* Write what `peer` changed. `peerWinsTies`: the peer is the primary. Each row is checked against its last change here
* first; a row that points at one that isn't here yet waits (replica_deferred) and is tried with the next pull.
Expand Down Expand Up @@ -491,7 +493,13 @@ export function applyChanges(peer: string, changes: ReplicaChange[], opts: Apply
} catch {
/* the database is in trouble: the next pull brings it again */
}
log.warn(`could not take a change of ${c.t} yet`, { peer, key: c.k, error: one instanceof Error ? one.message : String(one) });
const error = one instanceof Error ? one.message : String(one);
// Tried again with every pull: the same failure is worth one warning.
const seen = `${peer}${KEY_SEP}${c.t}${KEY_SEP}${c.k}${KEY_SEP}${error}`;
if (stuckWarned.size >= 1000) stuckWarned.clear();
const level = stuckWarned.has(seen) ? "info" : "warn";
stuckWarned.add(seen);
log[level](`could not take a change of ${c.t} yet`, { peer, key: c.k, error });
}
}
Object.assign(result, partial);
Expand Down
12 changes: 7 additions & 5 deletions packages/core/src/remote/runners.ts
Original file line number Diff line number Diff line change
Expand Up @@ -642,7 +642,7 @@ export function startRunners(): void {
for (const r of all<RunnerRow>("SELECT * FROM runners")) startLink(r);
setMirrorHooks({
runFinished: (runnerId, run) => {
void syncRunner(runnerId, { agentId: run.agentId }).catch((err) => log.warn(`could not merge memory after run ${run.id}`, err instanceof Error ? err.message : err));
void syncRunner(runnerId, { agentId: run.agentId, work: true }).catch((err) => logFailure(`could not merge memory after run ${run.id}`, err));
},
resync: (runnerId, runId) => links.get(runnerId)?.sendClientEvent({ type: "run.resync", runId }),
released: (runnerId, conversationId) => pendingReleases(runnerId).includes(conversationId),
Expand Down Expand Up @@ -741,7 +741,7 @@ export function linkGone(err: unknown): boolean {

function logFailure(text: string, err: unknown): void {
const detail = err instanceof Error ? err.message : err;
if (linkGone(err)) log.info(text, detail);
if (linkGone(err) || (err as { repeated?: boolean } | null)?.repeated) log.info(text, detail);
else log.warn(text, detail);
}

Expand Down Expand Up @@ -784,12 +784,14 @@ export function syncRunner(id: string, opts: { agentId?: string; profileId?: str
/* the next connect tells */
}
} catch (err) {
update("runners", id, { sync_error: `Couldn't copy your setup: ${message(err)}`, updated_at: now() });
const failure = Object.assign(new HttpError(502, `Couldn't copy your setup to ${r.name}: ${message(err)}`, "runner_sync_failed"), { cause: err });
const syncError = `Couldn't copy your setup: ${message(err)}`;
update("runners", id, { sync_error: syncError, updated_at: now() });
// The same failure as last time: it was warned about then, every edit and run tries again.
const failure = Object.assign(new HttpError(502, `Couldn't copy your setup to ${r.name}: ${message(err)}`, "runner_sync_failed"), { cause: err, repeated: r.sync_error === syncError });
// 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);
logFailure(`could not copy the setup to runner ${id}`, failure);
} finally {
syncing.delete(id);
emit(id);
Expand Down
22 changes: 11 additions & 11 deletions packages/core/test/remote-snapshot.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,8 @@ 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 { clearGroup, setGroup } from "../src/remote/group";
import { ensureReplica, madeHere, pendingKeys, 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";
import { HttpError } from "../src/util";
Expand Down Expand Up @@ -726,7 +724,7 @@ 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 () => {
test("a project a secondary made with the same name as one of the primary's moves to a free slug, and the rest of the copy lands", async () => {
const secondary = mkdtempSync(join(tmpdir(), "godmode-snapshot-secondary-"));
try {
use(controller);
Expand All @@ -749,10 +747,11 @@ describe("applying a snapshot on a runner", () => {

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(result.warnings).toEqual([]);
expect(row("projects", "prj_here").slug).toBe("earshot-2");
expect(row("projects", "prj_there").slug).toBe("earshot");
expect(pendingKeys("rnr_primary", "projects").has("prj_here")).toBe(true);
expect(row("agents", "agt_helper")).toMatchObject({ project_id: "prj_there", instructions: "Work on Earshot." });
expect(appliedDigest()).toBe(copy.digest);
} finally {
clearGroup();
Expand All @@ -779,10 +778,11 @@ describe("applying a snapshot on a runner", () => {

use(secondary);
const result = await applySnapshot(copy);
expect(result.warnings.join("\n")).toContain('"Studio" (workspaces) wasn\'t copied');
expect(result.warnings).toEqual([]);
expect(row("workspaces", "ws_here").slug).toBe("studio-2");
expect(row("workspaces", "ws_there").slug).toBe("studio");
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(row("projects", "prj_there")).toMatchObject({ workspace_id: "ws_there", slug: "earshot" });
expect(exists("agents", "agt_helper")).toBe(true);
expect(appliedDigest()).toBe(copy.digest);
} finally {
Expand Down
19 changes: 17 additions & 2 deletions packages/core/test/replica.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,12 @@
*/
import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test";
import type { Database } from "bun:sqlite";
import { mkdtempSync, rmSync } from "node:fs";
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { loadConfig } from "../src/config";
import { all, closeDb, get, openDb, run, setDb } from "../src/db";
import { setLogLevel } from "../src/log";
import { setFileLogLevel, setLogDir, setLogLevel } from "../src/log";
import { getSettings, resetSettingsCache, updateSettings } from "../src/services/settings";
import { applyChanges, changesFor, ensureReplica, pendingFor, pendingKeys, replicaSeq, replicaTables, resetReplicaCache, retryDeferred, type ReplicaChange } from "../src/remote/replica";

Expand Down Expand Up @@ -272,6 +272,21 @@ describe("between two computers", () => {
expect(all("SELECT * FROM replica_deferred")).toHaveLength(0);
});

test("a change that keeps failing is warned about once, its retries only at info", () => {
use(b);
const logs = join(dir, "logs");
mkdirSync(logs, { recursive: true });
setLogDir(logs);
setFileLogLevel("info");
const broken: ReplicaChange = { t: "goals", k: "gol_broken", at: T(40), row: { id: "gol_broken", title: null, created_at: T(40), updated_at: T(40) } };
expect(applyChanges("rnr_a", [broken], { peerWinsTies: false }).failed).toBe(1);
expect(retryDeferred("rnr_a", { peerWinsTies: false }).failed).toBe(1);
expect(retryDeferred("rnr_a", { peerWinsTies: false }).failed).toBe(1);
const file = join(logs, "godmode.jsonl");
const lines = existsSync(file) ? readFileSync(file, "utf8").trim().split("\n").map((l) => JSON.parse(l) as { level: string; msg: string }) : [];
expect(lines.filter((l) => l.msg === "could not take a change of goals yet").map((l) => l.level)).toEqual(["warn", "info", "info"]);
});

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) } };
Expand Down
Loading