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
3 changes: 3 additions & 0 deletions packages/core/src/automations/conditions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import { createConversation, sendMessage, setConversationState } from "../servic
import { chatFitsAutomation } from "./conversation";
import { ensureRunListener, isAutomationBusy, receiveEvent } from "./events";
import { badRequest, conflict, now, truncate } from "../util";
import { standInHold } from "../remote/standIn";

const MAX_OBSERVATION = 2000;
/** Messages kept in a check conversation (older checks are pruned). */
Expand Down Expand Up @@ -90,6 +91,8 @@ export async function runConditionCheck(routineId: string, opts: { scheduled?: b
const agent = getAgent(routine.agentId);
if (!agent.enabled) throw conflict(`Agent "${agent.name}" is disabled`);
if (opts.scheduled && !routine.enabled) throw conflict(`“${routine.name}” is paused`);
const held = opts.scheduled ? standInHold(agent.id, routine.prompt) : null;
if (held) throw conflict(`“${routine.name}”: ${held}`);
const state = readTriggerState(routineId);
if (starting.has(routineId) || (state.checkRunId && runActive(state.checkRunId))) {
throw conflict(`“${routine.name}” is already checking its condition`);
Expand Down
13 changes: 11 additions & 2 deletions packages/core/src/automations/events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ import { remindWaitingAutomation } from "../services/questions";
import { onCheckRunFinished } from "./conditions";
import { HttpError, badRequest, conflict, newId, now, parseJson, truncate } from "../util";
import { isLicenseRequired, noteLicenseRefusal } from "../license/license";
import { localizePaths } from "../remote/localPaths";
import { standInHold } from "../remote/standIn";

const log = logger("automations");

Expand Down Expand Up @@ -375,6 +377,12 @@ export async function dispatch(routineId: string, opts: { byHuman?: boolean } =
emitRoutine(routineId);
return null;
}
const held = opts.byHuman ? null : standInHold(routine.agentId, routine.prompt);
if (held) {
setEventStatus(ids, "skipped", held);
emitRoutine(routineId);
return null;
}
if (rateLimited(routine)) return null;

dispatching.add(routineId);
Expand Down Expand Up @@ -464,9 +472,10 @@ function attr(value: string): string {
* Schedule ticks (and running a schedule by hand) get the plain prompt, like before automations.
*/
export function buildEventPrompt(routine: Routine, events: AutomationEvent[]): string {
if (routine.trigger.type === "schedule" || events.every((e) => e.source === "schedule")) return routine.prompt;
const prompt = localizePaths(routine.prompt);
if (routine.trigger.type === "schedule" || events.every((e) => e.source === "schedule")) return prompt;
const test = events.every((e) => e.source === "manual");
const parts: string[] = [routine.prompt.trim(), "", "---", ""];
const parts: string[] = [prompt.trim(), "", "---", ""];
if (test) {
parts.push(
`This is a test of the automation “${routine.name}”, started by the human — no real event happened. Do a dry run: check that you can reach everything the task needs and say briefly what you would do for a typical event, but don't send, post, buy, delete or change anything.`,
Expand Down
8 changes: 7 additions & 1 deletion packages/core/src/memory/dreaming.ts
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ import {
} from "./files";
import { exhaustedBudget } from "../services/budgets";
import { licenseBlocks } from "../license/license";
import { standInHold } from "../remote/standIn";
import { MAX_REWRITES, MAX_TIDY_CHANGES, RECENT_DAYS, applyTidy, canUndoWorkspace, undoWorkspace, workspaceChangedSince, workspaceInventory, type TidyOperation } from "./workspace";


Expand Down Expand Up @@ -946,6 +947,11 @@ async function dreamIfIdle(agent: Agent, settings: Settings, at: Date): Promise<
deferred.delete(agent.id);
// Dreams cost money too: none on schedule while a monthly budget is used up (Dream now still works).
if (exhaustedBudget(agent) || licenseBlocks()) return false;
const held = standInHold(agent.id);
if (held) {
log.info(`no scheduled dream for ${agent.slug}: ${held}`);
return false;
}
await startDream(agent.id, "schedule");
return true;
}
Expand Down Expand Up @@ -992,7 +998,7 @@ async function sweepOverflowing(at: Date): Promise<string[]> {
try {
if (deferred.has(agent.id) || at.getTime() - (earlyFailed.get(agent.id) ?? -Infinity) < EARLY_DREAM_EVERY_MS) continue;
if (!memoryOverflowing(agent, settings, at) || !agentIdle(agent.id, at)) continue;
if (exhaustedBudget(agent) || licenseBlocks()) continue;
if (exhaustedBudget(agent) || licenseBlocks() || standInHold(agent.id)) continue;
await startDream(agent.id, "size");
started.push(agent.id);
} catch (err) {
Expand Down
34 changes: 34 additions & 0 deletions packages/core/src/remote/devices.ts
Original file line number Diff line number Diff line change
Expand Up @@ -435,6 +435,7 @@ async function primaryArrived(id: string): Promise<void> {
} catch (err) {
log.info("could not take the primary's changes yet", { error: err instanceof Error ? err.message : String(err) });
}
await stopWaitingWork(id, primaryName(), standingInSince());
// Chats continued here while it was away (also before standing in began) live here now.
await releaseChatsNow(id);
const since = standingInSince();
Expand All @@ -444,6 +445,39 @@ async function primaryArrived(id: string): Promise<void> {
void catchUpWithPrimary(id);
}

/**
* On a secondary: work for `primary` doesn't go on by itself later (a usage limit or budget that resets, a follow-up)
* in chats it took over from there, or in automations it started while standing in since `since`. The primary is back
* and does its own work again; days later it would only be repeated. What waits for the human stays.
*/
export async function stopWaitingWork(primary: string, name: string, since: string | null = null): Promise<number> {
const reason = `Stopped: ${name} is back and does its own work again. Continue this chat yourself if it still needs it`;
const taken = takenFrom(primary);
const runs = new Set(
since
? all<{ run_id: string }>(
`SELECT p.run_id FROM paused_runs p JOIN runs r ON r.id = p.run_id
WHERE p.reason IN ('limit', 'budget') AND r.trigger IN ('routine', 'heartbeat', 'check', 'dream', 'followup') AND r.created_at >= ?`,
since,
).map((r) => r.run_id)
: [],
);
for (const id of taken) {
const paused = pauseOf(id);
if (paused && (paused.reason === "limit" || paused.reason === "budget")) runs.add(paused.run_id);
}
let stopped = 0;
for (const runId of runs) {
await cancelRun(runId, reason, { byHuman: true }).then(
() => stopped++,
(err) => log.info(`could not stop the waiting run ${runId}`, { error: err instanceof Error ? err.message : String(err) }),
);
}
for (const id of taken) if (cancelFollowup(id)) stopped++;
if (stopped) log.info("the primary is back; work waiting for it here stops", { stopped });
return stopped;
}

/** False when the primary went away again meanwhile: this computer keeps standing in. */
async function handBack(id: string, since: string): Promise<boolean> {
// Tells the primary there is something to pull; then wait until it has it (it pulls when its link to here is up).
Expand Down
26 changes: 26 additions & 0 deletions packages/core/src/remote/localPaths.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
/**
* Paths in an automation's prompt that name another computer's home folder (written on the primary of a device group,
* run on a secondary that stands in for it): the same place under this computer's home.
*/
import { existsSync } from "node:fs";
import { homedir } from "node:os";
import { dirname } from "node:path";

const HOME = /(?<![\w./~-])\/(?:Users|home)\/[^/\s"'`<>()]+(?=\/)/g;

export function localizePaths(text: string, home = homedir()): string {
return text.replace(HOME, (other) => (other === home || existsSync(other) ? other : home));
}

/** Folders under this computer's home the text names that aren't here (a file that isn't there yet is fine). */
export function missingFolders(text: string, home = homedir()): string[] {
const missing = new Set<string>();
const under = new RegExp(`(?<![\\w./~-])${home.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}/[^\\s"'\`<>()]+`, "g");
for (const [raw] of localizePaths(text, home).matchAll(under)) {
const path = raw.replace(/[.,;:!?]+$/, "");
if (existsSync(path)) continue;
const folder = path.endsWith("/") ? path.slice(0, -1) : dirname(path);
if (folder !== home && !existsSync(folder)) missing.add(folder);
}
return [...missing];
}
5 changes: 4 additions & 1 deletion packages/core/src/remote/memorySync.ts
Original file line number Diff line number Diff line change
Expand Up @@ -66,12 +66,14 @@ function mergeLines(local: string, remote: string): string {
* Pure. `base` = what both sides had at the last sync (null = first time). Per file: changed on one side only → that
* side; changed on both → local's lines plus the remote lines local doesn't have; deleted on one side and untouched on
* the other → deleted; deleted on one side but changed on the other → kept. Files either side couldn't read are left
* alone on both (they end up in `merged.skipped`).
* alone on both (they end up in `merged.skipped`). `owner`: on the first sync, the side whose version of a file both
* have wins (the primary's real memory over the template or a stand-in's first dream); without it, both are merged.
*/
export function mergeMemory(
base: MemorySnapshot | null,
local: MemorySnapshot,
remote: MemorySnapshot,
owner?: "local" | "remote",
): { merged: MemorySnapshot; changedLocal: boolean; changedRemote: boolean } {
const skipped = [...new Set([...local.skipped, ...remote.skipped])].sort();
const unknown = (path: string) => skipped.some((p) => path === p || path.startsWith(`${p}/`));
Expand All @@ -86,6 +88,7 @@ export function mergeMemory(
else if (mine === was) text = theirs;
else if (mine === null) text = theirs;
else if (theirs === null) text = mine;
else if (!base && owner) text = owner === "local" ? mine : theirs;
else text = mergeLines(mine, theirs);
if (text !== null) files[path] = text;
}
Expand Down
28 changes: 16 additions & 12 deletions packages/core/src/remote/runners.ts
Original file line number Diff line number Diff line change
Expand Up @@ -796,30 +796,34 @@ export function syncRunner(id: string, opts: { agentId?: string; profileId?: str
});
}

/** What this computer and `runnerId` last agreed on for an agent's memory: the base of their next merge. */
export function storeMemoryBase(runnerId: string, agentId: string, state: { digest: string; snapshot: unknown }): void {
sql(
"INSERT INTO runner_memory (runner_id, agent_id, digest, snapshot, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(runner_id, agent_id) DO UPDATE SET digest = excluded.digest, snapshot = excluded.snapshot, updated_at = excluded.updated_at",
runnerId,
agentId,
state.digest,
JSON.stringify(state.snapshot),
now(),
);
}

/** Three-way merge of an agent's memory files with the runner's; both sides end up with the same files. */
async function exchangeMemory(runnerId: string, l: RemoteLink, name: string, agentId: string): Promise<boolean> {
const remote = await l.json<{ digest: string; snapshot: Parameters<typeof mergeMemory>[2] }>("GET", `/api/link/agents/${encodeURIComponent(agentId)}/memory`);
const local = readMemoryState(agentId);
const baseRow = get<{ digest: string; snapshot: string }>("SELECT digest, snapshot FROM runner_memory WHERE runner_id = ? AND agent_id = ?", runnerId, agentId);
const store = (digest: string, snapshot: unknown) =>
sql(
"INSERT INTO runner_memory (runner_id, agent_id, digest, snapshot, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(runner_id, agent_id) DO UPDATE SET digest = excluded.digest, snapshot = excluded.snapshot, updated_at = excluded.updated_at",
runnerId,
agentId,
digest,
JSON.stringify(snapshot),
now(),
);
if (remote.digest === local.digest) {
if (baseRow?.digest !== local.digest) store(local.digest, local.snapshot);
if (baseRow?.digest !== local.digest) storeMemoryBase(runnerId, agentId, local);
return false;
}
const base = baseRow ? parseJson<Parameters<typeof mergeMemory>[0]>(baseRow.snapshot, null) : null;
const { merged, changedLocal, changedRemote } = mergeMemory(base, local.snapshot, remote.snapshot);
const owner = isSecondary() && runnerId === primaryRunnerId() ? "remote" : "local";
const { merged, changedLocal, changedRemote } = mergeMemory(base, local.snapshot, remote.snapshot, owner);
let digest = local.digest;
if (changedLocal) digest = (await writeMemoryState(agentId, merged, `Memory from ${name}`)).digest;
if (changedRemote) digest = (await l.json<{ digest: string }>("PUT", `/api/link/agents/${encodeURIComponent(agentId)}/memory`, { snapshot: merged })).digest;
store(digest, merged);
storeMemoryBase(runnerId, agentId, { digest, snapshot: merged });
return changedLocal || changedRemote;
}

Expand Down
21 changes: 19 additions & 2 deletions packages/core/src/remote/standIn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,11 @@
* working on the primary continue here from a recap. remote/devices.ts hands it all back when the primary returns.
*/
import type { StandIn } from "@godmode/shared";
import { deleteMeta, getMeta, setMeta } from "../db";
import { deleteMeta, get, getMeta, setMeta } from "../db";
import { logger } from "../log";
import { badRequest, now } from "../util";
import { isSecondary, setStandingIn, standingIn, standingInSince } from "./group";
import { isSecondary, primaryRunnerId, setStandingIn, standingIn, standingInSince } from "./group";
import { missingFolders } from "./localPaths";

const log = logger("devices");

Expand Down Expand Up @@ -156,3 +157,19 @@ export function forgetStandIn(): void {
deleteMeta(ENABLED_META);
deleteMeta(MINUTES_META);
}

/**
* Why a stand-in doesn't start an agent's dream or automation on its own, or null. The agent's memory never came over
* from the primary (its folder here is the empty template), or the prompt names folders this computer doesn't have:
* that work would run blind and its memory would clash with the primary's when it is back.
*/
export function standInHold(agentId: string, prompt?: string): string | null {
if (!standingIn()) return null;
const primary = primaryRunnerId();
if (primary && !get("SELECT 1 FROM runner_memory WHERE runner_id = ? AND agent_id = ?", primary, agentId)) {
return "Waits for the primary: this agent's memory never came over to this computer";
}
const missing = prompt ? missingFolders(prompt) : [];
if (missing.length) return `Waits for the primary: ${missing.join(", ")} ${missing.length === 1 ? "isn't" : "aren't"} on this computer`;
return null;
}
9 changes: 8 additions & 1 deletion packages/core/src/scheduler/scheduler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@ import { pausedRun } from "../services/pauses";
import { remindWaitingAutomation } from "../services/questions";
import { exhaustedBudget } from "../services/budgets";
import { isLicenseRequired, noteLicenseRefusal } from "../license/license";
import { localizePaths } from "../remote/localPaths";
import { standInHold } from "../remote/standIn";

const log = logger("scheduler");

Expand Down Expand Up @@ -132,6 +134,11 @@ export async function triggerRoutine(id: string, opts: { scheduled?: boolean; by
const agent = getAgent(routine.agentId);
if (!agent.enabled) throw conflict(`Agent "${agent.name}" is disabled`);
if (opts.scheduled && !routine.enabled) throw conflict(`Routine "${routine.name}" is disabled`);
const held = opts.scheduled ? standInHold(agent.id, routine.prompt) : null;
if (held) {
recordEvent(id, { source: "schedule", title: opts.title ?? "Scheduled time reached", status: "skipped", note: held });
throw conflict(`Routine "${routine.name}": ${held}`);
}
const active = triggering.has(id) ? null : activeMainRun(id);
if (triggering.has(id) || active) {
if (opts.scheduled) {
Expand All @@ -155,7 +162,7 @@ export async function triggerRoutine(id: string, opts: { scheduled?: boolean; by
nextRunAt(routine),
id,
);
const { run } = await sendMessage(conversationId, { content: routine.prompt, trigger: "routine", routineId: routine.id, source: "automation", byHuman: opts.byHuman });
const { run } = await sendMessage(conversationId, { content: localizePaths(routine.prompt), trigger: "routine", routineId: routine.id, source: "automation", byHuman: opts.byHuman });
recordEvent(id, {
source: opts.scheduled ? "schedule" : "manual",
title: opts.title ?? (opts.scheduled ? "Scheduled time reached" : "Started manually"),
Expand Down
7 changes: 6 additions & 1 deletion packages/core/src/server/routes/link.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import { readHandover } from "../../remote/handover";
import { fixCheck, runnerHealth } from "../../remote/health";
import { forgetController, runnerInfo } from "../../remote/linkServer";
import { readMemoryState, writeMemoryState } from "../../remote/memorySync";
import { storeMemoryBase } from "../../remote/runners";
import { MAX_UPDATE_BYTES, applyUpdate, downloadUpdate, receiveChunk } from "../../remote/selfUpdate";
import { applySnapshot, type ConfigSnapshot } from "../../remote/snapshot";
import { audit } from "../../services/audit";
Expand Down Expand Up @@ -70,7 +71,11 @@ export function registerLinkRoutes(app: Hono): void {

app.put("/api/link/agents/:id/memory", async (c) => {
const input = await body(c, z.object({ snapshot: z.object({ files: z.record(z.string(), z.unknown()) }).passthrough() }));
return c.json(await writeMemoryState(c.req.param("id"), input.snapshot as never, "Memory from Godmode"));
const state = await writeMemoryState(c.req.param("id"), input.snapshot as never, "Memory from Godmode");
// A device of the group: this side knows the agreed state too (and that the agent's memory came over).
const peer = deviceIdOf(viaLink(c) ?? "");
if (peer) storeMemoryBase(peer, c.req.param("id"), state);
return c.json(state);
});

// A chat of this runner moves to the controller: its Claude session and transcript go along.
Expand Down
27 changes: 27 additions & 0 deletions packages/core/test/local-paths.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
import { afterAll, describe, expect, test } from "bun:test";
import { mkdirSync, mkdtempSync, rmSync } from "node:fs";
import { homedir, tmpdir } from "node:os";
import { join } from "node:path";
import { localizePaths, missingFolders } from "../src/remote/localPaths";

const home = mkdtempSync(join(tmpdir(), "godmode-home-"));
mkdirSync(join(home, ".godmode", "shared", "sales"), { recursive: true });

afterAll(() => rmSync(home, { recursive: true, force: true }));

describe("paths of another computer's home", () => {
test("map to this computer's home; existing folders and this home stay", () => {
expect(localizePaths("Log to /Users/someone-else-xyz/.godmode/shared/sales/sent.md.", home)).toBe(`Log to ${home}/.godmode/shared/sales/sent.md.`);
expect(localizePaths(`See ${home}/notes.md and ${homedir()}/x`, home)).toBe(`See ${home}/notes.md and ${homedir()}/x`);
expect(localizePaths("cd /home/other-user-xyz/work && ls", home)).toBe(`cd ${home}/work && ls`);
expect(localizePaths("no paths here, just text/Users/a/b", home)).toBe("no paths here, just text/Users/a/b");
});

test("folders the prompt needs that aren't here are named; files still to be written are fine", () => {
expect(missingFolders("Append to /Users/other-xyz/.godmode/shared/sales/sent.md", home)).toEqual([]);
expect(missingFolders("Read /Users/other-xyz/.godmode/shared/leads/ki-leads.md, then /Users/other-xyz/.godmode/shared/leads/", home)).toEqual([
`${home}/.godmode/shared/leads`,
]);
expect(missingFolders("Nothing outside: /tmp/x/y.md", home)).toEqual([]);
});
});
9 changes: 9 additions & 0 deletions packages/core/test/remote-snapshot.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -766,6 +766,15 @@ describe("merging an agent's memory", () => {
expect(merged.files).toEqual({ "MEMORY.md": "local\nremote\n", "memory/a.md": "a\n", "memory/b.md": "b\n" });
});

test("on the first sync the owner's version of a file both have wins; files only one side has are kept", () => {
const primary = snap({ "MEMORY.md": "# Memory\n- real facts\n", "memory/leads.md": "leads\n" });
const standIn = snap({ "MEMORY.md": "# Memory\n- first dream from activity\n", "memory/dream.md": "d\n" });
expect(mergeMemory(null, standIn, primary, "remote").merged.files).toEqual({ "MEMORY.md": "# Memory\n- real facts\n", "memory/dream.md": "d\n", "memory/leads.md": "leads\n" });
expect(mergeMemory(null, primary, standIn, "local").merged.files["MEMORY.md"]).toBe("# Memory\n- real facts\n");
// With a base both sides worked on the same memory: their lines are merged as always.
expect(mergeMemory(primary, primary, standIn, "local").merged.files["MEMORY.md"]).toBe(standIn.files["MEMORY.md"]);
});

test("a file deleted on one side and untouched on the other is deleted", () => {
const base = snap({ "MEMORY.md": "a\n", "memory/old.md": "old\n", "memory/stale.md": "stale\n" });
const local = snap({ "MEMORY.md": "a\n", "memory/stale.md": "stale\n" });
Expand Down
Loading
Loading