diff --git a/apps/desktop/src/components/agents/routine-dialog.tsx b/apps/desktop/src/components/agents/routine-dialog.tsx index 3c734637..59c07ca3 100644 --- a/apps/desktop/src/components/agents/routine-dialog.tsx +++ b/apps/desktop/src/components/agents/routine-dialog.tsx @@ -3,11 +3,11 @@ import { useMutation, useQueryClient } from "@tanstack/react-query"; import { AnimatePresence, motion } from "motion/react"; import { toast } from "sonner"; import { ChevronRight, CircleCheck } from "lucide-react"; -import type { ComposioConnection, Routine, RoutineInput, RoutineTrigger, RoutineTriggerType, RoutineNotify } from "@godmode/shared"; +import type { ComposioConnection, GroupDevice, Routine, RoutineInput, RoutineTrigger, RoutineTriggerType, RoutineNotify } from "@godmode/shared"; import { findModel } from "@godmode/shared"; import { api, errorMessage } from "@/lib/api"; import { qk } from "@/lib/queryKeys"; -import { useAllAgents, useModelCatalog } from "@/lib/hooks"; +import { useAllAgents, useDeviceGroup, useModelCatalog } from "@/lib/hooks"; import { modKey } from "@/lib/desktop"; import { useEasyMode } from "@/lib/mode"; import { clearDraft, useDraft } from "@/lib/drafts"; @@ -78,6 +78,8 @@ export interface RoutineDraft { enabled: boolean; reuseConversation: boolean; notify: RoutineNotify; + /** Device groups: "primary" or a device id; null = any. */ + runOn: string | null; } /** Like the core: events get a conversation each (named after the event); schedules and conditions keep one. */ @@ -108,6 +110,7 @@ function initialDraft(routine: Routine | null | undefined, agentId: string | und enabled: routine?.enabled ?? initial?.enabled ?? true, reuseConversation: routine?.reuseConversation ?? initial?.reuseConversation ?? defaultReuse(triggerType), notify: routine?.notify ?? "failures", + runOn: routine?.runOn ?? initial?.runOn ?? null, }; } @@ -167,6 +170,8 @@ export function RoutineDialog({ const type = draft.triggerType; const meta = TRIGGER_TYPES[type]; const eventTrigger = isEventTrigger(type); + const { data: group } = useDeviceGroup(); + const pinnable = !!group?.role && !eventTrigger; const appSchema = useAppTriggerSchema(draft.app); const windowProblem = useMemo( @@ -233,6 +238,7 @@ export function RoutineDialog({ enabled: draft.enabled, reuseConversation: draft.reuseConversation, notify: draft.notify, + runOn: pinnable ? draft.runOn : undefined, }; if (!routine) return api.routines.create(input); if (!routine.fillOnly || !reveal) return api.routines.update(routine.id, input); @@ -618,6 +624,8 @@ export function RoutineDialog({

+ {pinnable && set("runOn", v)} />} + {kept.saved ? ( void }) { + const devices = list.map((d) => ({ + value: d.id ?? "primary", + name: d.name, + tag: [d.role === "primary" ? "primary" : null, d.self ? "this computer" : null].filter(Boolean).join(", "), + online: d.state === "online", + })); + const picked = value ? devices.find((d) => d.value === value) : null; + return ( +
+ + +

+ {!value + ? "Runs where your schedules run: the primary, or another computer while it stands in." + : picked + ? `Only ${picked.name} runs it, also while another computer stands in. A time it misses while away runs when it's back${picked.online ? "" : " (it's offline now)"}.` + : "Pinned to a computer that isn't listed: one that left the group (the primary runs it then), or one this computer can't see while the primary is away."} +

+
+ ); +} + function RawCronField({ id, label, value, onChange }: { id: string; label: string; value: string; onChange: (cron: string) => void }) { const problem = validateCron(value); return ( diff --git a/packages/core/src/automations/conversation.ts b/packages/core/src/automations/conversation.ts index 69a89597..e3b34a74 100644 --- a/packages/core/src/automations/conversation.ts +++ b/packages/core/src/automations/conversation.ts @@ -32,9 +32,15 @@ export function automationAgentReveals(routine: Pick): boole * became fill-only (or stopped being it) starts a new chat, and nothing read in the old one comes along. For an * agent that reads no raw secrets anyway, every chat of its own fits. */ -export function chatFitsAutomation(routine: Pick, conversationId: string): boolean { - const existing = get<{ secret_access: string | null }>("SELECT secret_access FROM conversations WHERE id = ? AND agent_id = ?", conversationId, routine.agentId); +export function chatFitsAutomation(routine: Pick & { runOn?: string | null }, conversationId: string): boolean { + const existing = get<{ secret_access: string | null; runner_id: string | null }>( + "SELECT secret_access, runner_id FROM conversations WHERE id = ? AND agent_id = ?", + conversationId, + routine.agentId, + ); if (!existing) return false; + // Pinned to one computer: a chat that lives on another one would run there. + if (routine.runOn && existing.runner_id) return false; return (existing.secret_access === "fill") === routine.fillOnly || !automationAgentReveals(routine); } diff --git a/packages/core/src/cloud/scope.ts b/packages/core/src/cloud/scope.ts index cf329943..0170bbef 100644 --- a/packages/core/src/cloud/scope.ts +++ b/packages/core/src/cloud/scope.ts @@ -307,6 +307,7 @@ const RULES: [methods: string, path: string, rule: Rule][] = [ ["GET", "/api/link/conversations/:id/handover", refused(COMPUTER_ONLY)], ["GET", "/api/link/conversations/:id/handover/size", refused(COMPUTER_ONLY)], ["POST", "/api/link/exec", refused(COMPUTER_ONLY)], + ["POST", "/api/link/routines/:id/run", refused(COMPUTER_ONLY)], ["POST", "/api/link/forget", refused(COMPUTER_ONLY)], ["PUT", "/api/link/update/chunk", refused(COMPUTER_ONLY)], ["POST", "/api/link/update/apply", refused(COMPUTER_ONLY)], diff --git a/packages/core/src/db/migrations.ts b/packages/core/src/db/migrations.ts index 0dfbad71..a06ae4ea 100644 --- a/packages/core/src/db/migrations.ts +++ b/packages/core/src/db/migrations.ts @@ -1691,6 +1691,14 @@ CREATE TABLE artifact_versions ( created_at TEXT NOT NULL, PRIMARY KEY (artifact_id, version) ); +`, + }, + { + id: 103, + name: "routine_run_on", + sql: /* sql */ ` +-- Device groups: the computer an automation's schedule runs on ("primary" or a device id; NULL = any). +ALTER TABLE routines ADD COLUMN run_on TEXT; `, }, ]; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index eadf5fb0..e8172f7a 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -30,7 +30,7 @@ import { ensureDefaultAgent } from "./agents/service"; import { recoverInterruptedRuns, shutdownRunner } from "./runner/runner"; import { interruptedWork, resumeInterruptedWork, startOutageRecovery, stopOutageRecovery } from "./services/resume"; import { clearRunMods, startModUpdates, stopModUpdates } from "./mods/service"; -import { startScheduler, stopScheduler } from "./scheduler/scheduler"; +import { reloadSchedules, startScheduler } from "./scheduler/scheduler"; import { startSignInRetry, stopSignInRetry } from "./automations/signInRetry"; import { startDreaming, stopDreaming } from "./memory/dreaming"; import { startFollowups, stopFollowups } from "./services/followups"; @@ -211,7 +211,6 @@ async function serve(values: Record, role?: CoreConfig["role"]) if (sharedOn) return; sharedOn = true; // A runner's runs are told by the computer it works for (they arrive there as its own runs). - startScheduler(); startSignInRetry(); startRunNotices(); startDreaming(); @@ -225,7 +224,6 @@ async function serve(values: Record, role?: CoreConfig["role"]) const stopShared = async () => { if (!sharedOn) return; sharedOn = false; - stopScheduler(); stopSignInRetry(); stopRunNotices(); stopDreaming(); @@ -237,11 +235,15 @@ async function serve(values: Record, role?: CoreConfig["role"]) stopCalls(); }; if (!runner) { - if (runsShared()) void groupSettled().then(startShared); + // Every computer of a group fires the automations pinned to it; the rest only where the shared work runs. + const settled = groupSettled(); + void settled.then(startScheduler); + if (runsShared()) void settled.then(startShared); startModUpdates(); onGroupChange(() => { if (runsShared()) startShared(); else void stopShared().catch((err) => log.warn("could not hand the shared work to the primary", err)); + reloadSchedules(); }); } // Adopt VMs that kept running while Godmode was closed. diff --git a/packages/core/src/mcp/tools.ts b/packages/core/src/mcp/tools.ts index e41feb98..c6ad1856 100644 --- a/packages/core/src/mcp/tools.ts +++ b/packages/core/src/mcp/tools.ts @@ -828,6 +828,14 @@ const routineFields = { .boolean() .optional() .describe("Keep one conversation for all runs (continuity). Default: true for schedule and condition, false for app and webhook (one conversation per event)."), + runOn: z + .string() + .max(100) + .nullable() + .optional() + .describe( + 'Device groups, schedule and condition triggers: the only computer that runs it — "primary", a device id or computer name from Settings → Devices; null = any (the primary, or a secondary standing in for it). Pin automations that need files, folders or browser profiles of one computer.', + ), }; function triggerSummary(r: Routine) { @@ -858,6 +866,7 @@ function routineSummary(r: Routine, promptMax = 500) { timezone: r.timezone, enabled: r.enabled, reuseConversation: r.reuseConversation, + ...(r.runOn ? { runOn: r.runOn } : {}), status: r.triggerStatus.state, ...(r.triggerStatus.message ? { statusMessage: r.triggerStatus.message } : {}), lastRunAt: r.lastRunAt, @@ -2451,7 +2460,7 @@ Make it look like a polished 2026 product: clean layout, real hierarchy, readabl defineTool({ name: "routine_update", description: - "Change an automation's trigger, schedule, prompt, filter, name or enabled state. Changing one of an agent that may read secrets in plain text while you can't makes it fill-only (pausing it doesn't).", + "Change an automation's trigger, schedule, prompt, filter, name, enabled state or the computer it runs on (runOn). Changing one of an agent that may read secrets in plain text while you can't makes it fill-only (pausing it or picking its computer doesn't).", schema: z.object({ routineId: z.string(), name: routineFields.name.optional(), @@ -2462,6 +2471,7 @@ Make it look like a polished 2026 product: clean layout, real hierarchy, readabl timezone: routineFields.timezone, enabled: routineFields.enabled, reuseConversation: routineFields.reuseConversation, + runOn: routineFields.runOn, }), when: managesSetup, run: async ({ routineId, trigger, ...patch }, { agent, ctx }) => { @@ -2476,8 +2486,8 @@ Make it look like a polished 2026 product: clean layout, real hierarchy, readabl const owner = getAgent(latest.agentId); const secrets = automationSecrets(agent, ctx, owner, "change its automations"); if (typeof secrets === "string") return fail(secrets); - // What this caller writes or starts runs without raw secrets from now on. Pausing does neither. - const pauseOnly = fields.length === 1 && patch.enabled === false; + // What this caller writes or starts runs without raw secrets from now on. Pausing does neither, nor picking the computer. + const pauseOnly = fields.length > 0 && fields.every((f) => f === "runOn" || (f === "enabled" && patch.enabled === false)); const fillOnly = secrets.fillOnly && !pauseOnly; const r = updateRoutine(routineId, { ...patch, ...(resolved ? { trigger: resolved } : {}), ...(fillOnly ? { fillOnly } : {}) }); const became = fillOnly && !latest.fillOnly; diff --git a/packages/core/src/remote/group.ts b/packages/core/src/remote/group.ts index 60a070db..dd7a74ae 100644 --- a/packages/core/src/remote/group.ts +++ b/packages/core/src/remote/group.ts @@ -9,7 +9,7 @@ * `link.*` meta never goes into backups, like the pairings it describes. */ import type { DeviceRole } from "@godmode/shared"; -import { deleteMeta, getMeta, setMeta } from "../db"; +import { deleteMeta, get, getMeta, setMeta } from "../db"; const ROLE_META = "link.group.role"; const PRIMARY_META = "link.group.primary"; @@ -97,6 +97,26 @@ export function runsShared(): boolean { return !isSecondary() || standingIn(); } +/** + * Whether this computer runs the schedule of an automation pinned to `runOn` ("primary", a device id, or null = any). + * Pinned ones run on their computer only, standing in or not; outside a group every pin is moot. On the primary, a pin + * to a computer that left the group runs here. + */ +export function runsRoutineHere(runOn: string | null | undefined): boolean { + const role = groupRole(); + if (!role) return true; + if (!runOn) return runsShared(); + if (role === "primary") { + if (runOn === "primary") return true; + try { + return !get("SELECT 1 FROM runners WHERE id = ? AND kind = 'device'", runOn); + } catch { + return false; + } + } + return runOn === selfDeviceId(); +} + export function setStandingIn(since: string | null): void { if (since) setMeta(STAND_IN_META, since); else deleteMeta(STAND_IN_META); diff --git a/packages/core/src/remote/routinePins.ts b/packages/core/src/remote/routinePins.ts new file mode 100644 index 00000000..49b35ce0 --- /dev/null +++ b/packages/core/src/remote/routinePins.ts @@ -0,0 +1,69 @@ +/** + * Automations pinned to one computer of a device group (owner: remote): `routines.run_on` is "primary" or a device's id + * in the primary's runner list (`GroupDevice.id`), the same on every computer. The scheduler fires a pinned automation + * only there (group.ts `runsRoutineHere`), and Run now elsewhere goes there over the link. + */ +import type { Run } from "@godmode/shared"; +import { all, get } from "../db"; +import { computerName } from "../mobile/devices"; +import { HttpError, badRequest } from "../util"; +import { groupRole, isPrimary, primaryRunnerId, selfDeviceId } from "./group"; + +const DEVICE_ID = /^[A-Za-z0-9_-]{1,100}$/; + +function rowName(id: string | null): string | null { + return id ? (get<{ name: string }>("SELECT name FROM runners WHERE id = ?", id)?.name ?? null) : null; +} + +/** + * A pin as given (by the app, an agent: "primary", a device id, a computer's name, or what a secondary calls its + * primary's runner) as it is stored. undefined = unchanged, null = any computer. + */ +export function resolveRunOn(input: string | null | undefined): string | null | undefined { + if (input === undefined) return undefined; + const value = (input ?? "").trim(); + if (!value || value.toLowerCase() === "any") return null; + const role = groupRole(); + if (!role) throw badRequest("This computer isn't part of a device group: its automations run here."); + const same = (a: string | null | undefined) => !!a && a.toLowerCase() === value.toLowerCase(); + const primaryId = role === "secondary" ? primaryRunnerId() : null; + if (same("primary") || same(primaryId) || same(role === "primary" ? computerName() : rowName(primaryId))) return "primary"; + if (role === "secondary" && (same(selfDeviceId()) || same(computerName()))) return selfDeviceId(); + const devices = all<{ id: string; name: string }>("SELECT id, name FROM runners WHERE kind = 'device'"); + const device = devices.find((d) => same(d.id)) ?? devices.find((d) => same(d.name)); + if (device) return device.id; + // A secondary knows the other secondaries only through its primary: their ids stand as given. + if (role === "secondary" && DEVICE_ID.test(value) && value.includes("_")) return value; + throw badRequest(`Unknown computer “${value}”: give "primary", a computer of the device group (Settings → Devices) or null for any.`); +} + +/** The computer an automation is pinned to, as this one calls it. */ +export function runOnName(runOn: string): string { + if (runOn === "primary") return isPrimary() ? computerName() : (rowName(primaryRunnerId()) ?? "the primary"); + if (runOn === selfDeviceId()) return computerName(); + return rowName(runOn) ?? "the computer it is pinned to"; +} + +/** + * Run now for an automation pinned to another computer: the request goes there (a secondary sends it to its primary, + * which passes it on unless it is the pinned one). Refused clearly while that computer can't be reached. + */ +export async function runPinnedElsewhere(routine: { id: string; name: string; runOn: string }, opts: { byHuman?: boolean }): Promise { + const target = isPrimary() ? routine.runOn : primaryRunnerId(); + const where = runOnName(routine.runOn); + // A secondary reaches the other secondaries through its primary. + const via = !isPrimary() && routine.runOn !== "primary" ? `${runOnName("primary")}, which passes it on there,` : where; + const offline = () => + new HttpError(409, `“${routine.name}” runs on ${where} only, and ${via} can't be reached right now. It runs there at its next time, or set Run on to Any device.`, "device_offline"); + if (!target) throw offline(); + const { link } = await import("./runners"); + let l: ReturnType; + try { + l = link(target); + l.assertOnline(); + } catch { + throw offline(); + } + const here = isPrimary() || routine.runOn === "primary"; + return l.json("POST", `/api/link/routines/${encodeURIComponent(routine.id)}/run`, { here, byHuman: opts.byHuman === true }, { timeoutMs: 120_000 }); +} diff --git a/packages/core/src/scheduler/scheduler.ts b/packages/core/src/scheduler/scheduler.ts index b11563d5..7ffb22f7 100644 --- a/packages/core/src/scheduler/scheduler.ts +++ b/packages/core/src/scheduler/scheduler.ts @@ -4,7 +4,9 @@ * a condition tick starts a condition check (automations/conditions.ts). Schedules with a random start window get a * one-off job at the start drawn for their next time slot, replaced by the next one when it fires; with several runs per * window every part of the window is such a slot. A slot that fell while the scheduler wasn't running (Godmode was off, - * the computer slept) runs once when it is back (catchUpMissed). + * the computer slept) runs once when it is back (catchUpMissed). In a device group an automation pinned to one computer + * (`run_on`) fires there only, also while another computer stands in for the primary; the rest fire where the group's + * schedules run (remote/group.ts `runsRoutineHere`). */ import { Cron } from "croner"; import type { Routine, Run } from "@godmode/shared"; @@ -25,6 +27,8 @@ import { exhaustedBudget } from "../services/budgets"; import { isLicenseRequired, noteLicenseRefusal } from "../license/license"; import { localizePaths } from "../remote/localPaths"; import { standInHold } from "../remote/standIn"; +import { runsRoutineHere } from "../remote/group"; +import { runOnName } from "../remote/routinePins"; const log = logger("scheduler"); @@ -51,6 +55,7 @@ interface WantedRoutine { type: string; start_window: number | null; runs: number | null; + run_on: string | null; } /** A schedule's start window and runs per window as the scheduler uses them: out-of-range values count as none. */ @@ -134,6 +139,9 @@ 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`); + if (opts.scheduled && !runsRoutineHere(routine.runOn)) { + throw conflict(`Routine "${routine.name}" runs on ${routine.runOn ? runOnName(routine.runOn) : "the computer that runs the group's schedules"}`); + } 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 }); @@ -191,7 +199,10 @@ export async function triggerRoutine(id: string, opts: { scheduled?: boolean; by interface ScheduleGroup { active: () => boolean; settle: () => Promise; - /** This computer schedules only while it stands in for its primary: runMissedSince covers what came due. */ + /** + * A secondary: what came due for the group while it stood in is runMissedSince's; it catches up only automations + * pinned to it. + */ standingIn?: () => boolean; } @@ -232,6 +243,10 @@ async function onTick(routineId: string, scheduledAt: Date | null = null, slot: reloadSchedules(); return; } + if (!runsRoutineHere(routine.runOn)) { + reloadSchedules(); + return; + } // Another computer of the group ran this time already (it stood in while this one was away). if (late && routine.lastRunAt && Date.parse(routine.lastRunAt) >= scheduledAt!.getTime() - LATE_MS) { log.info(`routine ${routineId} already ran at ${routine.lastRunAt}; this late tick is skipped`); @@ -265,14 +280,16 @@ interface ScheduleRow { cron: string; timezone: string; trigger: string; + run_on: string | null; created_at: string; } +/** Enabled schedules this computer fires. */ function scheduleRows(): ScheduleRow[] { return all( - `SELECT r.id, r.cron, r.timezone, r.trigger, r.created_at FROM routines r JOIN agents a ON a.id = r.agent_id + `SELECT r.id, r.cron, r.timezone, r.trigger, r.run_on, r.created_at FROM routines r JOIN agents a ON a.id = r.agent_id WHERE r.enabled = 1 AND a.enabled = 1 AND CASE WHEN json_valid(r.trigger) THEN json_extract(r.trigger, '$.type') END = 'schedule'`, - ); + ).filter((r) => runsRoutineHere(r.run_on)); } /** @@ -318,6 +335,8 @@ export async function runMissedSince(since: string): Promise { let ran = 0; for (const r of scheduleRows()) { if (gen !== generation) break; + // Pinned here: this computer's own, caught up like any of its schedules (catchUpMissed). + if (r.run_on || !runsRoutineHere(null)) continue; const due = lastSlot(r, Date.now()); // A drawn start still to come: this computer's own random start runs it. if (!due || due.slot < from - LATE_MS || due.start > Date.now() || slotTaken(r.id, due.slot)) continue; @@ -360,13 +379,13 @@ function beat(): void { */ export async function catchUpMissed(reason: "start" | "wake", aliveAt: number): Promise { if (!Number.isFinite(aliveAt) || aliveAt <= 0) return 0; - if (group?.standingIn?.()) return 0; const gen = generation; if (group?.active()) await group.settle().catch(() => undefined); if (!started) return 0; let ran = 0; for (const r of scheduleRows()) { if (gen !== generation) break; + if (!r.run_on && group?.standingIn?.()) continue; const t = Date.now(); const due = lastSlot(r, t); if (!due || due.start > t || due.start <= aliveAt - LATE_MS || Date.parse(r.created_at) >= due.slot || slotTaken(r.id, due.slot)) continue; @@ -397,14 +416,14 @@ export async function catchUpMissed(reason: "start" | "wake", aliveAt: number): export function reloadSchedules(): void { if (!started) return; const wanted = all( - `SELECT id, cron, timezone, type, start_window, runs FROM ( - SELECT r.id, r.cron, r.timezone, + `SELECT id, cron, timezone, type, start_window, runs, run_on FROM ( + SELECT r.id, r.cron, r.timezone, r.run_on, CASE WHEN json_valid(r.trigger) THEN json_extract(r.trigger, '$.type') END AS type, CASE WHEN json_valid(r.trigger) THEN json_extract(r.trigger, '$.startWindowMinutes') END AS start_window, CASE WHEN json_valid(r.trigger) THEN json_extract(r.trigger, '$.runsPerWindow') END AS runs FROM routines r JOIN agents a ON a.id = r.agent_id WHERE r.enabled = 1 AND a.enabled = 1) WHERE type IN ('schedule', 'condition')`, - ); + ).filter((r) => runsRoutineHere(r.run_on)); const byId = new Map(wanted.map((r) => [r.id, r])); for (const [id, scheduled] of jobs) { @@ -425,8 +444,9 @@ export function reloadSchedules(): void { } } - // Persist next run times so the UI shows when each routine fires next. - for (const row of all<{ id: string; next_run_at: string | null }>("SELECT id, next_run_at FROM routines")) { + // Persist next run times so the UI shows when each routine fires next. Another computer's pinned ones are its own. + for (const row of all<{ id: string; next_run_at: string | null; run_on: string | null }>("SELECT id, next_run_at, run_on FROM routines")) { + if (!runsRoutineHere(row.run_on)) continue; const next = jobs.get(row.id)?.job.nextRun()?.toISOString() ?? null; if (next === row.next_run_at) continue; exec("UPDATE routines SET next_run_at = ? WHERE id = ?", next, row.id); @@ -439,11 +459,16 @@ export function reloadSchedules(): void { * while that run still stands still. */ function repairStaleStatuses() { + const here = all<{ id: string; run_on: string | null }>("SELECT id, run_on FROM routines WHERE last_status IN ('queued', 'running', 'paused')") + .filter((r) => runsRoutineHere(r.run_on)) + .map((r) => r.id); + if (!here.length) return; exec( `UPDATE routines SET last_status = COALESCE( (SELECT status FROM runs WHERE runs.routine_id = routines.id AND runs.trigger = 'routine' ORDER BY created_at DESC LIMIT 1), 'failed') - WHERE last_status IN ('queued', 'running', 'paused') + WHERE last_status IN ('queued', 'running', 'paused') AND id IN (SELECT value FROM json_each(?)) AND NOT EXISTS (SELECT 1 FROM runs WHERE runs.routine_id = routines.id AND runs.trigger = 'routine' AND runs.status IN ('queued', 'running'))`, + JSON.stringify(here), ); } diff --git a/packages/core/src/server/routes/agents.ts b/packages/core/src/server/routes/agents.ts index 6dc66be6..ff140c28 100644 --- a/packages/core/src/server/routes/agents.ts +++ b/packages/core/src/server/routes/agents.ts @@ -189,6 +189,7 @@ export const routineSchema = z.object({ reuseConversation: z.boolean().optional(), notify: z.enum(["always", "failures", "never"]).optional(), fillOnly: z.boolean().optional(), + runOn: z.string().trim().max(100).nullable().optional(), }); function scopeParam(value: string | undefined): string | null | "all" { diff --git a/packages/core/src/server/routes/link.ts b/packages/core/src/server/routes/link.ts index 682c0bc2..f2fcd70b 100644 --- a/packages/core/src/server/routes/link.ts +++ b/packages/core/src/server/routes/link.ts @@ -23,6 +23,7 @@ import { callForDevice } from "../../remote/deviceTools"; import { welcomeFor } from "../ws"; import { get } from "../../db"; import { getConversationSummary, sendMessage } from "../../services/conversations"; +import { runRoutineNow } from "../../services/routines"; import type { MessageBlock, RunTrigger } from "@godmode/shared"; import { cancelRun } from "../../runner/runner"; import { body, z } from "../validate"; @@ -249,6 +250,13 @@ export function registerLinkRoutes(app: Hono): void { return c.json({ ...result, conversation: getConversationSummary(id) }); }); + // Run now for an automation pinned to one computer, passed on by another computer of the group (remote/routinePins.ts). + app.post("/api/link/routines/:id/run", async (c) => { + groupPeer(c); + const input = await body(c, z.object({ here: z.boolean().optional(), byHuman: z.boolean().optional() })); + return c.json(await runRoutineNow(c.req.param("id"), { byHuman: input.byHuman === true, here: input.here === true })); + }); + app.post("/api/link/forget", (c) => { const controllerId = viaLink(c); // Answered first: forgetting closes the link this answer travels on. diff --git a/packages/core/src/services/routines.ts b/packages/core/src/services/routines.ts index 35b964e2..8ab3c54f 100644 --- a/packages/core/src/services/routines.ts +++ b/packages/core/src/services/routines.ts @@ -20,6 +20,8 @@ import { cancelPendingEvents, sendTestEvent } from "../automations/events"; import { clearWebhookToken, issueWebhookToken, webhookPathOf } from "../automations/webhooks"; import { appTriggerHealth, getTriggerType, missingConfigFields, requestAppTriggerSync } from "../integrations/composioTriggers"; import { badRequest, conflict, newId, notFound, now, parseJson } from "../util"; +import { runsRoutineHere } from "../remote/group"; +import { resolveRunOn, runPinnedElsewhere } from "../remote/routinePins"; const log = logger("routines"); @@ -36,6 +38,7 @@ export interface RoutineRow { reuse_conversation: number; notify?: string | null; secret_access?: string | null; + run_on?: string | null; conversation_id: string | null; last_run_at: string | null; next_run_at: string | null; @@ -147,6 +150,7 @@ function toModel(r: RoutineRow, pending = 0): Routine { reuseConversation: bool(r.reuse_conversation), notify: r.notify === "always" || r.notify === "never" ? r.notify : "failures", fillOnly: r.secret_access === "fill", + runOn: r.run_on ?? null, conversationId: r.conversation_id, lastRunAt: r.last_run_at, nextRunAt: r.next_run_at, @@ -463,6 +467,7 @@ function syncRoutinesFile(agentId: string): void { reuseConversation: r.reuseConversation, notify: r.notify, fillOnly: r.fillOnly, + runOn: r.runOn, createdAt: r.createdAt, updatedAt: r.updatedAt, })); @@ -552,6 +557,7 @@ export function createRoutine(input: RoutineInput): Routine { const { cron, next } = scheduleFor(id, trigger, input.cron, timezone); const filter = trigger.type === "app" || trigger.type === "webhook" ? normalizeFilter(input.filter) : ""; const enabled = input.enabled !== false; + const runOn = resolveRunOn(input.runOn) ?? null; // The webhook secret is sealed with the vault key: fail before anything is stored when the vault is locked. const webhook = trigger.type === "webhook" ? issueWebhookToken(id) : null; const ts = now(); @@ -569,6 +575,7 @@ export function createRoutine(input: RoutineInput): Routine { reuse_conversation: int(input.reuseConversation ?? usesCron(trigger))!, notify: input.notify ?? "failures", secret_access: input.fillOnly ? "fill" : null, + run_on: runOn, conversation_id: null, last_run_at: null, next_run_at: enabled && agent.enabled && next ? next.toISOString() : null, @@ -620,6 +627,8 @@ export function updateRoutine(id: string, patch: Partial): Routine reuse_conversation: int(reuse), notify: patch.notify, secret_access: patch.fillOnly === undefined ? undefined : patch.fillOnly ? "fill" : null, + // Unchanged stays as it is: a pin to a computer that left the group can't be given again, but saves. + run_on: patch.runOn === current.runOn ? undefined : resolveRunOn(patch.runOn), updated_at: now(), next_run_at: enabled && agent.enabled && next ? next.toISOString() : null, webhook_token_hash: webhook?.hash, @@ -678,8 +687,12 @@ export function deleteRoutine(id: string): void { * test event. Throws 409 when it is already running. */ /** `byHuman`: the human clicked Run now (a scheduled automation then runs although a monthly budget is used up). */ -export async function runRoutineNow(id: string, opts: { byHuman?: boolean } = {}): Promise { +/** `here`: another computer of the group sent it here as the one it is pinned to (routes/link.ts): no further hop. */ +export async function runRoutineNow(id: string, opts: { byHuman?: boolean; here?: boolean } = {}): Promise { const routine = getRoutine(id); + if (routine.runOn && usesCron(routine.trigger) && !opts.here && !runsRoutineHere(routine.runOn)) { + return runPinnedElsewhere({ id, name: routine.name, runOn: routine.runOn }, opts); + } switch (routine.trigger.type) { case "condition": return runConditionCheck(id, { manual: true }); diff --git a/packages/core/test/replica.test.ts b/packages/core/test/replica.test.ts index 9d054071..48a598c4 100644 --- a/packages/core/test/replica.test.ts +++ b/packages/core/test/replica.test.ts @@ -135,6 +135,35 @@ describe("between two computers", () => { expect(pendingFor("rnr_a")).toBe(0); }); + test("the computer an automation is pinned to goes to the other computer", () => { + for (const db of [a, b]) { + use(db); + run("PRAGMA foreign_keys = OFF"); + const cols = all<{ name: string; type: string; notnull: number; dflt_value: string | null; pk: number }>('PRAGMA table_info("agents")').filter((c) => c.pk || (c.notnull && c.dflt_value === null)); + run(`INSERT OR REPLACE INTO agents (${cols.map((c) => `"${c.name}"`).join(", ")}) VALUES (${cols.map(() => "?").join(", ")})`, ...cols.map((c) => (c.name === "id" ? "agt_x" : /INT|REAL|NUM/i.test(c.type) ? 1 : `v_${c.name}`))); + run("INSERT OR REPLACE INTO routines (id, agent_id, name, cron, timezone, prompt, created_at, updated_at) VALUES ('rtn_pin', 'agt_x', 'Posts', '0 12 * * *', 'UTC', 'post', ?, ?)", T(0), T(0)); + run("PRAGMA foreign_keys = ON"); + } + use(a); + const start = replicaSeq(); + run("UPDATE routines SET run_on = 'primary' WHERE id = 'rtn_pin'"); + const page = changesFor("rnr_b", start, { setup: true }); + expect(page.changes.map((c) => c.k)).toContain("rtn_pin"); + use(b); + run("PRAGMA foreign_keys = OFF"); + try { + applyChanges("rnr_a", page.changes, { peerWinsTies: true }); + } finally { + run("PRAGMA foreign_keys = ON"); + } + expect(get<{ run_on: string | null }>("SELECT run_on FROM routines WHERE id = 'rtn_pin'")?.run_on).toBe("primary"); + for (const db of [a, b]) { + use(db); + run("DELETE FROM routines WHERE id = 'rtn_pin'"); + run("DELETE FROM agents WHERE id = 'agt_x'"); + } + }); + test("the newer change of a row wins, on both sides", () => { use(a); goal("gol_4", "Old"); diff --git a/packages/core/test/scheduler.test.ts b/packages/core/test/scheduler.test.ts index ee7126e0..9f058472 100644 --- a/packages/core/test/scheduler.test.ts +++ b/packages/core/test/scheduler.test.ts @@ -4,7 +4,7 @@ import { homedir, tmpdir } from "node:os"; import { join } from "node:path"; import type { Agent, Conversation, Run } from "@godmode/shared"; import { loadConfig } from "../src/config"; -import { closeDb, get, insert, openDb, run as exec } from "../src/db"; +import { closeDb, deleteMeta, get, insert, openDb, run as exec, setMeta } from "../src/db"; import { bus } from "../src/events/bus"; import { setLogLevel } from "../src/log"; import { resetSettingsCache } from "../src/services/settings"; @@ -407,6 +407,171 @@ describe("triggering", () => { } }); + describe("automations pinned to one computer", () => { + const ts = now(); + + beforeEach(() => { + stopScheduler(); + exec("UPDATE routines SET enabled = 0"); + insert("runners", { id: "rnr_primary", name: "Desk Mac", public_key: "pk-primary", port: 7787, kind: "device", created_at: ts, updated_at: ts }); + insert("runners", { id: "rnr_laptop", name: "Laptop", public_key: "pk-laptop", port: 7788, kind: "device", created_at: ts, updated_at: ts }); + exec("INSERT OR IGNORE INTO runner_memory (runner_id, agent_id, digest, snapshot, updated_at) VALUES ('rnr_primary', ?, 'd', '{}', ?)", agent.id, ts); + }); + + afterEach(() => { + stopScheduler(); + setScheduleGroup(null); + clearGroup(); + resetGroupCache(); + deleteMeta("link.group.self"); + exec("UPDATE routines SET enabled = 0, run_on = NULL"); + exec("DELETE FROM runners WHERE id IN ('rnr_primary', 'rnr_laptop')"); + }); + + /** A secondary (its id in the primary's list: rnr_self) that may stand in for its primary. */ + function secondary(standingIn: boolean) { + setGroup("secondary", "rnr_primary"); + setMeta("link.group.self", "rnr_self"); + if (standingIn) setStandingIn(now()); + } + + const scheduled = () => scheduledRoutines().map((s) => s.routineId); + + test("names, ids and the primary's runner resolve to the stored pin", () => { + expect(() => createRoutine({ agentId: agent.id, name: "Alone", cron: "0 6 * * *", prompt: "x", runOn: "primary" })).toThrow("isn't part of a device group"); + secondary(false); + const r = createRoutine({ agentId: agent.id, name: "Posts", cron: "0 12 * * *", prompt: "post", runOn: "Desk Mac", enabled: false }); + expect(r.runOn).toBe("primary"); + expect(updateRoutine(r.id, { runOn: "rnr_primary" }).runOn).toBe("primary"); + expect(updateRoutine(r.id, { runOn: "rnr_self" }).runOn).toBe("rnr_self"); + expect(updateRoutine(r.id, { runOn: "laptop" }).runOn).toBe("rnr_laptop"); + expect(updateRoutine(r.id, { name: "Posts 2" }).runOn).toBe("rnr_laptop"); + expect(updateRoutine(r.id, { runOn: "any" }).runOn).toBeNull(); + expect(() => updateRoutine(r.id, { runOn: "Kitchen PC" })).toThrow("Unknown computer"); + }); + + test("a stand-in never fires an automation pinned to the primary, nor catches it up", async () => { + secondary(true); + const slot = dueSlot(new Cron("0 * * * *", { timezone: "UTC", paused: true }), new Date())!; + const pinned = createRoutine({ agentId: agent.id, name: "Social posts", cron: "0 * * * *", timezone: "UTC", prompt: "post", runOn: "primary" }); + const shared = createRoutine({ agentId: agent.id, name: "Inbox", cron: "0 * * * *", timezone: "UTC", prompt: "inbox" }); + const theirNext = "2030-01-01T12:00:00.000Z"; + exec("UPDATE routines SET next_run_at = ? WHERE id = ?", theirNext, pinned.id); + startScheduler(); + expect(scheduled()).toContain(shared.id); + expect(scheduled()).not.toContain(pinned.id); + // Its next run is the primary's to tell. + expect(getRoutine(pinned.id).nextRunAt).toBe(theirNext); + + expect(await runMissedSince(new Date(slot.getTime() - 5 * 60_000).toISOString())).toBe(1); + expect(sent.map((m) => m.routineId)).toEqual([shared.id]); + expect((await catchHttp(triggerRoutine(pinned.id, { scheduled: true }))).status).toBe(409); + expect(getRoutine(pinned.id).lastRunAt).toBeNull(); + }); + + test("a secondary fires the automations pinned to it, also while it doesn't stand in", async () => { + secondary(false); + const mine = createRoutine({ agentId: agent.id, name: "Mine", cron: "0 6 * * *", timezone: "UTC", prompt: "x", runOn: "rnr_self" }); + const other = createRoutine({ agentId: agent.id, name: "Laptop's", cron: "0 6 * * *", timezone: "UTC", prompt: "x", runOn: "rnr_laptop" }); + const shared = createRoutine({ agentId: agent.id, name: "Shared", cron: "0 6 * * *", timezone: "UTC", prompt: "x" }); + startScheduler(); + expect(scheduled()).toEqual([mine.id]); + setStandingIn(now()); + reloadSchedules(); + expect(scheduled().sort()).toEqual([mine.id, shared.id].sort()); + expect(scheduled()).not.toContain(other.id); + setStandingIn(null); + reloadSchedules(); + expect(scheduled()).toEqual([mine.id]); + }); + + test("the primary leaves automations pinned to another computer to it, unless that one left the group", () => { + setGroup("primary"); + const laptop = createRoutine({ agentId: agent.id, name: "On the laptop", cron: "0 6 * * *", timezone: "UTC", prompt: "x", runOn: "Laptop" }); + const here = createRoutine({ agentId: agent.id, name: "Here", cron: "0 6 * * *", timezone: "UTC", prompt: "x", runOn: "primary" }); + startScheduler(); + expect(scheduled()).toContain(here.id); + expect(scheduled()).not.toContain(laptop.id); + exec("DELETE FROM runners WHERE id = 'rnr_laptop'"); + reloadSchedules(); + expect(scheduled()).toContain(laptop.id); + }); + + test("an automation pinned to a computer that left the group still saves", () => { + setGroup("primary"); + const r = createRoutine({ agentId: agent.id, name: "Laptop's", cron: "0 6 * * *", timezone: "UTC", prompt: "x", runOn: "rnr_laptop", enabled: false }); + exec("DELETE FROM runners WHERE id = 'rnr_laptop'"); + expect(updateRoutine(r.id, { prompt: "y", runOn: "rnr_laptop" }).runOn).toBe("rnr_laptop"); + expect(() => updateRoutine(r.id, { runOn: "rnr_gone" })).toThrow("Unknown computer"); + }); + + test("a pinned automation doesn't go on in a chat that lives on another computer", async () => { + secondary(false); + const r = createRoutine({ agentId: agent.id, name: "Posts", cron: "0 12 * * *", prompt: "post", runOn: "rnr_self" }); + finish(await triggerRoutine(r.id)); + const first = getRoutine(r.id).conversationId!; + // The chat moved to the primary meanwhile. + exec("UPDATE conversations SET runner_id = 'rnr_primary' WHERE id = ?", first); + finish(await triggerRoutine(r.id, { scheduled: true })); + expect(sent).toHaveLength(2); + expect(sent[1]!.conversationId).not.toBe(first); + expect(getRoutine(r.id).conversationId).toBe(sent[1]!.conversationId); + }); + + test("Run now elsewhere: refused clearly while the pinned computer can't be reached", async () => { + secondary(false); + const r = createRoutine({ agentId: agent.id, name: "Social posts", cron: "0 12 * * *", prompt: "post", runOn: "primary" }); + const err = await catchHttp(runRoutineNow(r.id)); + expect(err.status).toBe(409); + expect(err.code).toBe("device_offline"); + expect(err.message).toContain("Desk Mac"); + expect(sent).toHaveLength(0); + // Sent here as the pinned computer's (the other one passed it on): it runs here. + finish(await runRoutineNow(r.id, { here: true })); + expect(sent.map((m) => m.routineId)).toEqual([r.id]); + }); + + test("the pinned primary catches up a slot it missed while away, once; a stand-in that was there didn't run it", async () => { + setGroup("primary"); + setSystemTime(new Date("2026-10-20T12:38:00.000Z")); + try { + startScheduler(); + stopScheduler(); + const r = createRoutine({ agentId: agent.id, name: "Social posts", cron: "0 15 * * 1-5", timezone: "Europe/Berlin", prompt: "post", runOn: "primary" }); + exec("UPDATE routines SET created_at = ? WHERE id = ?", "2026-10-01T00:00:00.000Z", r.id); + setScheduleGroup({ active: () => true, settle: async () => {} }); + setSystemTime(new Date("2026-10-20T14:06:01.000Z")); + startScheduler(); + await Bun.sleep(20); + expect(sent.filter((m) => m.routineId === r.id)).toHaveLength(1); + expect(await catchUpMissed("wake", Date.parse("2026-10-20T12:38:00.000Z"))).toBe(0); + expect(sent.filter((m) => m.routineId === r.id)).toHaveLength(1); + } finally { + setSystemTime(); + } + }); + + test("a secondary catches up the automations pinned to it after a sleep, not the group's", async () => { + secondary(false); + setSystemTime(new Date("2026-10-20T12:38:00.000Z")); + try { + startScheduler(); + stopScheduler(); + const mine = createRoutine({ agentId: agent.id, name: "Mine", cron: "0 15 * * 1-5", timezone: "Europe/Berlin", prompt: "x", runOn: "rnr_self" }); + const shared = createRoutine({ agentId: agent.id, name: "Shared", cron: "0 15 * * 1-5", timezone: "Europe/Berlin", prompt: "x" }); + exec("UPDATE routines SET created_at = ? WHERE id IN (?, ?)", "2026-10-01T00:00:00.000Z", mine.id, shared.id); + setStandingIn(now()); + setScheduleGroup({ active: () => true, settle: async () => {}, standingIn: () => true }); + setSystemTime(new Date("2026-10-20T14:06:01.000Z")); + startScheduler(); + await Bun.sleep(20); + expect(sent.map((m) => m.routineId)).toEqual([mine.id]); + } finally { + setSystemTime(); + } + }); + }); + 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/shared/src/api.ts b/packages/shared/src/api.ts index 9a0a7552..d77f8c63 100644 --- a/packages/shared/src/api.ts +++ b/packages/shared/src/api.ts @@ -4,6 +4,7 @@ import type { AgentCharacter } from "./character"; import type { AgentComputerConfig, ComputerTarget } from "./computer"; import type { RoutineNotify, + RoutineRunOn, Agent, AgentBrowserConfig, AgentPermissions, @@ -290,6 +291,8 @@ export interface RoutineInput { notify?: RoutineNotify; /** Runs without raw secrets (see `Routine.fillOnly`). */ fillOnly?: boolean; + /** See `Routine.runOn`. */ + runOn?: RoutineRunOn | null; } export interface WebhookRotateResult { diff --git a/packages/shared/src/models.ts b/packages/shared/src/models.ts index fc439c07..155c14c2 100644 --- a/packages/shared/src/models.ts +++ b/packages/shared/src/models.ts @@ -284,6 +284,9 @@ export interface RoutineTriggerStatus { /** When an automation tells the human that a run ended. A failure is told once until a run succeeds again. */ export type RoutineNotify = "always" | "failures" | "never"; +/** "primary" or the id of a device of the group (its id in the primary's runner list). */ +export type RoutineRunOn = "primary" | ID; + export interface Routine { id: ID; agentId: ID; @@ -305,6 +308,11 @@ export interface Routine { * read them itself. Logins are still filled into pages. Only the human switches it off. */ fillOnly: boolean; + /** + * Device groups: the computer that runs its schedule or checks — "primary" or a device's id (`GroupDevice.id`); null = + * whichever computer runs the group's schedules (the primary, or a secondary standing in for it). + */ + runOn: RoutineRunOn | null; conversationId: ID | null; lastRunAt: ISODate | null; /** Next scheduled run (schedule) or check (condition). */