diff --git a/packages/core/src/automations/conditions.ts b/packages/core/src/automations/conditions.ts index 9cce5332..5650c160 100644 --- a/packages/core/src/automations/conditions.ts +++ b/packages/core/src/automations/conditions.ts @@ -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). */ @@ -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`); diff --git a/packages/core/src/automations/events.ts b/packages/core/src/automations/events.ts index e99b2c3f..4f7c4568 100644 --- a/packages/core/src/automations/events.ts +++ b/packages/core/src/automations/events.ts @@ -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"); @@ -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); @@ -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.`, diff --git a/packages/core/src/memory/dreaming.ts b/packages/core/src/memory/dreaming.ts index 320ef7e0..ccedfc06 100644 --- a/packages/core/src/memory/dreaming.ts +++ b/packages/core/src/memory/dreaming.ts @@ -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"; @@ -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; } @@ -992,7 +998,7 @@ async function sweepOverflowing(at: Date): Promise { 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) { diff --git a/packages/core/src/remote/devices.ts b/packages/core/src/remote/devices.ts index 6e4fada6..f29eab66 100644 --- a/packages/core/src/remote/devices.ts +++ b/packages/core/src/remote/devices.ts @@ -435,6 +435,7 @@ async function primaryArrived(id: string): Promise { } 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(); @@ -444,6 +445,39 @@ async function primaryArrived(id: string): Promise { 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 { + 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 { // Tells the primary there is something to pull; then wait until it has it (it pulls when its link to here is up). diff --git a/packages/core/src/remote/localPaths.ts b/packages/core/src/remote/localPaths.ts new file mode 100644 index 00000000..2d3a4181 --- /dev/null +++ b/packages/core/src/remote/localPaths.ts @@ -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 = /(?()]+(?=\/)/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(); + const under = new RegExp(`(?()]+`, "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]; +} diff --git a/packages/core/src/remote/memorySync.ts b/packages/core/src/remote/memorySync.ts index 6bdfee61..178692f1 100644 --- a/packages/core/src/remote/memorySync.ts +++ b/packages/core/src/remote/memorySync.ts @@ -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}/`)); @@ -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; } diff --git a/packages/core/src/remote/runners.ts b/packages/core/src/remote/runners.ts index 5d24350c..5c71a3ea 100644 --- a/packages/core/src/remote/runners.ts +++ b/packages/core/src/remote/runners.ts @@ -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 { const remote = await l.json<{ digest: string; snapshot: Parameters[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[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; } diff --git a/packages/core/src/remote/standIn.ts b/packages/core/src/remote/standIn.ts index 0e2bd96e..24c96e7f 100644 --- a/packages/core/src/remote/standIn.ts +++ b/packages/core/src/remote/standIn.ts @@ -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"); @@ -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; +} diff --git a/packages/core/src/scheduler/scheduler.ts b/packages/core/src/scheduler/scheduler.ts index af873f5f..b11563d5 100644 --- a/packages/core/src/scheduler/scheduler.ts +++ b/packages/core/src/scheduler/scheduler.ts @@ -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"); @@ -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) { @@ -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"), diff --git a/packages/core/src/server/routes/link.ts b/packages/core/src/server/routes/link.ts index 190ffc2d..59650142 100644 --- a/packages/core/src/server/routes/link.ts +++ b/packages/core/src/server/routes/link.ts @@ -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"; @@ -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. diff --git a/packages/core/test/local-paths.test.ts b/packages/core/test/local-paths.test.ts new file mode 100644 index 00000000..5ada7bd9 --- /dev/null +++ b/packages/core/test/local-paths.test.ts @@ -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([]); + }); +}); diff --git a/packages/core/test/remote-snapshot.test.ts b/packages/core/test/remote-snapshot.test.ts index aceaa77d..8aa756f8 100644 --- a/packages/core/test/remote-snapshot.test.ts +++ b/packages/core/test/remote-snapshot.test.ts @@ -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" }); diff --git a/packages/core/test/scheduler.test.ts b/packages/core/test/scheduler.test.ts index 9bfdbfa8..ee7126e0 100644 --- a/packages/core/test/scheduler.test.ts +++ b/packages/core/test/scheduler.test.ts @@ -1,6 +1,6 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, setSystemTime, spyOn, test } from "bun:test"; import { mkdtempSync, rmSync } from "node:fs"; -import { tmpdir } from "node:os"; +import { homedir, tmpdir } from "node:os"; import { join } from "node:path"; import type { Agent, Conversation, Run } from "@godmode/shared"; import { loadConfig } from "../src/config"; @@ -15,6 +15,7 @@ import * as routines from "../src/services/routines"; import { createRoutine, getRoutine, runRoutineNow, updateRoutine } from "../src/services/routines"; import { catchUpMissed, dueSlot, reloadSchedules, runMissedSince, scheduledRoutines, setScheduleGroup, startScheduler, stopScheduler, triggerRoutine } from "../src/scheduler/scheduler"; import { HttpError, newId, now } from "../src/util"; +import { resetGroupCache, setGroup, setStandingIn, clearGroup } from "../src/remote/group"; import { Cron } from "croner"; let dataDir: string; @@ -378,6 +379,34 @@ describe("triggering", () => { for (const r of [hourly, ranThere, before]) updateRoutine(r.id, { enabled: false }); }); + test("a stand-in leaves automations of agents whose memory never came over, or whose folders aren't here, to the primary", async () => { + const ts = now(); + insert("runners", { id: "rnr_primary", name: "Primary", public_key: "pk-primary", port: 7787, kind: "device", created_at: ts, updated_at: ts }); + setGroup("secondary", "rnr_primary"); + setStandingIn(ts); + try { + const routine = createRoutine({ agentId: agent.id, name: "Mailing", cron: "0 6 * * *", prompt: "send the mails", enabled: true }); + expect((await catchHttp(triggerRoutine(routine.id, { scheduled: true }))).status).toBe(409); + expect(get<{ status: string; note: string }>("SELECT status, note FROM automation_events WHERE routine_id = ? ORDER BY rowid DESC LIMIT 1", routine.id)).toEqual({ + status: "skipped", + note: "Waits for the primary: this agent's memory never came over to this computer", + }); + + exec("INSERT INTO runner_memory (runner_id, agent_id, digest, snapshot, updated_at) VALUES ('rnr_primary', ?, 'd', '{}', ?)", agent.id, ts); + updateRoutine(routine.id, { prompt: `Read ${join(homedir(), `.godmode-missing-${newId("x")}`, "sent-log.md")} first` }); + expect((await catchHttp(triggerRoutine(routine.id, { scheduled: true }))).status).toBe(409); + + updateRoutine(routine.id, { prompt: "send the mails" }); + finish(await triggerRoutine(routine.id, { scheduled: true })); + expect(sent.map((m) => m.routineId)).toEqual([routine.id]); + updateRoutine(routine.id, { enabled: false }); + } finally { + clearGroup(); + resetGroupCache(); + exec("DELETE FROM runners WHERE id = 'rnr_primary'"); + } + }); + describe("slots missed while the scheduler wasn't running", () => { // Tuesday, 16:06 in Berlin: Godmode was off from 14:38 to 16:06. const wentAt = "2026-10-20T12:38:00.000Z"; diff --git a/packages/core/test/stand-in-waiting.test.ts b/packages/core/test/stand-in-waiting.test.ts new file mode 100644 index 00000000..99296996 --- /dev/null +++ b/packages/core/test/stand-in-waiting.test.ts @@ -0,0 +1,68 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { mkdirSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import type { Agent } from "@godmode/shared"; +import { makeAgent, setupEnv, until, type TestEnv } from "./fixtures/runner-harness"; +import { run as sql, setMeta } from "../src/db"; +import { getRun } from "../src/runner/runner"; +import { getConversation, startChat } from "../src/services/conversations"; +import { getFollowup, scheduleFollowup } from "../src/services/followups"; +import { pauseConversation, pauseOf, stopPauses } from "../src/services/pauses"; +import { stopWaitingWork } from "../src/remote/devices"; + +let env: TestEnv; +let agent: Agent; + +beforeAll(async () => { + env = await setupEnv("godmode-standin-waiting-"); + agent = await makeAgent({ name: "Stand-in Bot" }); +}); + +afterAll(async () => { + stopPauses(); + await env.close(); +}); + +const PRIMARY = "rnr_primarydesk01"; + +describe("chats a stand-in took over, once the primary is back", () => { + test("work that would go on by itself later stops; what waits for the human stays", async () => { + const since = new Date().toISOString(); + mkdirSync(env.stateDir, { recursive: true }); + writeFileSync(join(env.stateDir, "limit"), String(Math.floor(Date.now() / 1000) + 4 * 86_400)); + const limited = await startChat({ agentId: agent.id, content: "LIMIT_HIT send the follow-up emails" }); + await until(() => getRun(limited.run.id).status === "paused", 15_000, "the run to wait for the limit"); + expect(pauseOf(limited.conversation.id)).toMatchObject({ reason: "limit", auto: 1 }); + + const asked = await startChat({ agentId: agent.id, content: "WAIT_TO_FINISH look at the inbox" }); + await until(() => getRun(asked.run.id).status === "running" && getConversation(asked.conversation.id).messages.at(-1)!.blocks.length > 0, 10_000, "the run to start"); + await pauseConversation(asked.conversation.id); + await until(() => getRun(asked.run.id).status === "paused", 15_000, "the human's pause"); + + const later = await startChat({ agentId: agent.id, content: "hello" }); + await until(() => getRun(later.run.id).status === "succeeded", 10_000, "the run to finish"); + scheduleFollowup({ conversationId: later.conversation.id, agentId: agent.id, dueAt: new Date(Date.now() + 86_400_000), note: "post again" }); + + const untouched = await startChat({ agentId: agent.id, content: "LIMIT_HIT a chat of this computer" }); + await until(() => getRun(untouched.run.id).status === "paused", 15_000, "its run to wait for the limit"); + + const scheduled = await startChat({ agentId: agent.id, content: "LIMIT_HIT the 15:00 posting slot" }); + const older = await startChat({ agentId: agent.id, content: "LIMIT_HIT a slot from before" }); + await until(() => getRun(scheduled.run.id).status === "paused" && getRun(older.run.id).status === "paused", 15_000, "the slots to wait for the limit"); + sql("UPDATE runs SET trigger = 'routine' WHERE id IN (?, ?)", scheduled.run.id, older.run.id); + sql("UPDATE runs SET created_at = '2026-01-01T00:00:00.000Z' WHERE id = ?", older.run.id); + + setMeta(`link.release.${PRIMARY}`, JSON.stringify([limited.conversation.id, asked.conversation.id, later.conversation.id])); + expect(await stopWaitingWork(PRIMARY, "Desk", since)).toBe(3); + + expect(getRun(limited.run.id)).toMatchObject({ status: "cancelled" }); + expect(getRun(limited.run.id).error).toStartWith("Stopped: Desk is back"); + expect(pauseOf(limited.conversation.id)).toBeNull(); + expect(getConversation(limited.conversation.id).messages.at(-1)!.blocks.some((b) => b.type === "notice" && b.text.startsWith("Stopped: Desk is back"))).toBe(true); + expect(getFollowup(later.conversation.id)).toBeNull(); + expect(pauseOf(asked.conversation.id)).toMatchObject({ reason: "user" }); + expect(pauseOf(untouched.conversation.id)).toMatchObject({ reason: "limit" }); + expect(getRun(scheduled.run.id).status).toBe("cancelled"); + expect(pauseOf(older.conversation.id)).toMatchObject({ reason: "limit" }); + }, 60_000); +});