diff --git a/packages/core/src/automations/signInRetry.ts b/packages/core/src/automations/signInRetry.ts new file mode 100644 index 00000000..bf303ecf --- /dev/null +++ b/packages/core/src/automations/signInRetry.ts @@ -0,0 +1,160 @@ +/** + * Scheduled automations that failed because Claude was signed out run once more when it works again. The outage of a + * computer (this one or a runner) ends with the first run there that Claude answers; then every automation whose latest + * run failed over the sign-in gets one more run — unless its next time slot is less than two hours away, it is turned + * off now, or that run was itself such a second try (never more than once). + */ +import type { Run } from "@godmode/shared"; +import { modelProvider, runEndOf } from "@godmode/shared"; +import { all, get, getMeta, setMeta } from "../db"; +import { bus } from "../events/bus"; +import { logger } from "../log"; +import { triggerRoutine } from "../scheduler/scheduler"; +import { getRoutine } from "../services/routines"; +import { parseJson } from "../util"; +import { activeMainRun } from "./events"; + +const log = logger("sign-in-retry"); + +/** Failed runs looked at after an outage, and the runs that tried them again (newest last). */ +const HANDLED_META = "automations.signInRetried"; +/** Failed longer ago than this, the automation waits for its next time slot. */ +const MAX_AGE_MS = 24 * 3_600_000; +/** Its next time slot is this close: it runs then anyway. */ +const NEXT_SLOT_MS = 2 * 3_600_000; +const KEEP_HANDLED = 300; +const RETRY_TITLE = "Ran again once Claude was signed in"; + +interface FailedRun { + id: string; + routine_id: string; + error: string; + runner_id: string | null; +} + +function failedSignedOut(error: string | null | undefined): boolean { + return !!error && runEndOf(error)?.kind === "auth"; +} + +/** The automations' latest runs that failed over Claude's sign-in, on any computer. */ +function signedOutFailures(): FailedRun[] { + return all( + `SELECT r.id, r.routine_id, r.error, c.runner_id FROM runs r JOIN conversations c ON c.id = r.conversation_id + WHERE r.status = 'failed' AND r.trigger = 'routine' AND r.routine_id IS NOT NULL AND r.error IS NOT NULL AND r.finished_at > ? + AND r.id = (SELECT l.id FROM runs l WHERE l.routine_id = r.routine_id AND l.trigger = 'routine' ORDER BY l.created_at DESC, l.rowid DESC LIMIT 1) + ORDER BY +r.created_at, r.rowid`, + new Date(Date.now() - MAX_AGE_MS).toISOString(), + ).filter((r) => failedSignedOut(r.error)); +} + +function handled(): string[] { + return parseJson(getMeta(HANDLED_META), []); +} + +function markHandled(runIds: string[]): void { + setMeta(HANDLED_META, JSON.stringify([...handled(), ...runIds].slice(-KEEP_HANDLED))); +} + +/** Started as the one more run (here, or by another computer of the group that stood in). */ +function wasRetry(r: FailedRun): boolean { + return !!get("SELECT 1 FROM automation_events WHERE routine_id = ? AND run_id = ? AND title = ?", r.routine_id, r.id, RETRY_TITLE); +} + +/** Why the automation doesn't run again now, or null. */ +function skipReason(routineId: string): string | null { + let routine: ReturnType; + try { + routine = getRoutine(routineId); + } catch { + return "deleted"; + } + if (routine.trigger.type !== "schedule") return "not scheduled"; + if (!routine.enabled) return "turned off"; + if (get<{ enabled: number }>("SELECT enabled FROM agents WHERE id = ?", routine.agentId)?.enabled !== 1) return "agent turned off"; + if (activeMainRun(routineId)) return "already running"; + const next = routine.nextRunAt ? Date.parse(routine.nextRunAt) : Number.NaN; + if (Number.isFinite(next) && next - Date.now() < NEXT_SLOT_MS) return "next run soon"; + return null; +} + +/** + * Claude answered a run on `runnerId` (null: this computer): run the automations once more that failed there while it + * was signed out. Returns the runs it started. + */ +export async function retryAfterSignIn(runnerId: string | null): Promise { + const done = new Set(handled()); + // Each failure is looked at once: a second try that failed again waits for the automation's next time slot. + const failed = signedOutFailures().filter((r) => r.runner_id === runnerId && !done.has(r.id) && !wasRetry(r)); + if (!failed.length) return []; + markHandled(failed.map((r) => r.id)); + const started: Run[] = []; + const skipped: Record = {}; + for (const r of failed) { + const reason = skipReason(r.routine_id); + if (reason) { + skipped[r.routine_id] = reason; + continue; + } + try { + const run = await triggerRoutine(r.routine_id, { scheduled: true, title: RETRY_TITLE }); + started.push(run); + markHandled([run.id]); + } catch (err) { + skipped[r.routine_id] = err instanceof Error ? err.message : String(err); + } + } + log.info(`ran ${started.length} automation(s) again that failed while Claude was signed out, skipped ${Object.keys(skipped).length}`, { + ...(runnerId ? { runnerId } : {}), + runs: started.map((r) => r.id), + skipped, + }); + return started; +} + +let unsubscribe: (() => void) | null = null; +/** An automation may wait for Claude to work again: set at the start and by every one that fails over the sign-in. */ +let watching = false; +let queue: Promise = Promise.resolve(); + +function runnerOf(conversationId: string): string | null { + return get<{ runner_id: string | null }>("SELECT runner_id FROM conversations WHERE id = ?", conversationId)?.runner_id ?? null; +} + +function onFinished(run: Run): void { + if (run.status === "failed" && run.trigger === "routine" && failedSignedOut(run.error)) { + watching = true; + return; + } + // A run on Z.ai says nothing about Claude's sign-in. + if (run.status !== "succeeded" || !watching || modelProvider(run.model) === "zai") return; + const runnerId = runnerOf(run.conversationId); + queue = queue + .then(async () => { + if (!watching) return; + await retryAfterSignIn(runnerId); + const done = new Set(handled()); + if (!signedOutFailures().some((r) => !done.has(r.id) && !wasRetry(r))) watching = false; + }) + .catch((err) => log.warn("could not run the automations again that failed while Claude was signed out", err)); +} + +export function startSignInRetry(): void { + if (unsubscribe) return; + // What failed before this version ran is old news: it isn't started again hours later. + if (getMeta(HANDLED_META) === null) markHandled(signedOutFailures().map((r) => r.id)); + watching = true; + unsubscribe = bus.on((e) => { + if (e.type === "run.finished") onFinished(e.run); + }); +} + +export function stopSignInRetry(): void { + unsubscribe?.(); + unsubscribe = null; + watching = false; +} + +/** Tests: until the retries the last answered run started are through. */ +export function signInRetryIdle(): Promise { + return queue; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 31957412..eadf5fb0 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -31,6 +31,7 @@ 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 { startSignInRetry, stopSignInRetry } from "./automations/signInRetry"; import { startDreaming, stopDreaming } from "./memory/dreaming"; import { startFollowups, stopFollowups } from "./services/followups"; import { startHumanTasks, stopHumanTasks } from "./services/humanTasks"; @@ -211,6 +212,7 @@ async function serve(values: Record, role?: CoreConfig["role"]) 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(); startAutomationEvents(); @@ -224,6 +226,7 @@ async function serve(values: Record, role?: CoreConfig["role"]) if (!sharedOn) return; sharedOn = false; stopScheduler(); + stopSignInRetry(); stopRunNotices(); stopDreaming(); stopAppTriggers(); diff --git a/packages/core/src/scheduler/scheduler.ts b/packages/core/src/scheduler/scheduler.ts index feaa6a04..0e4e49c8 100644 --- a/packages/core/src/scheduler/scheduler.ts +++ b/packages/core/src/scheduler/scheduler.ts @@ -110,7 +110,7 @@ function nextRunAt(routine: Routine): string | null { * Run a schedule automation now: same path as a cron tick. Throws 409 when the agent is disabled or the routine is * already running. `scheduled` ticks additionally skip routines that were disabled meanwhile. */ -export async function triggerRoutine(id: string, opts: { scheduled?: boolean; byHuman?: boolean } = {}): Promise { +export async function triggerRoutine(id: string, opts: { scheduled?: boolean; byHuman?: boolean; title?: string } = {}): Promise { ensureRunListener(); const routine = getRoutine(id); if (routine.trigger.type !== "schedule") throw badRequest(`“${routine.name}” is not a scheduled automation`); @@ -143,7 +143,7 @@ export async function triggerRoutine(id: string, opts: { scheduled?: boolean; by const { run } = await sendMessage(conversationId, { content: routine.prompt, trigger: "routine", routineId: routine.id, source: "automation", byHuman: opts.byHuman }); recordEvent(id, { source: opts.scheduled ? "schedule" : "manual", - title: opts.scheduled ? "Scheduled time reached" : "Started manually", + title: opts.title ?? (opts.scheduled ? "Scheduled time reached" : "Started manually"), status: "running", runId: run.id, }); diff --git a/packages/core/src/services/claudeLogin.ts b/packages/core/src/services/claudeLogin.ts index 34e5da66..f044b256 100644 --- a/packages/core/src/services/claudeLogin.ts +++ b/packages/core/src/services/claudeLogin.ts @@ -10,9 +10,11 @@ import { statSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import type { ClaudeLoginStatus } from "@godmode/shared"; +import { config } from "../config"; import { deleteMeta, getMeta, setMeta } from "../db"; import { bus } from "../events/bus"; import { logger } from "../log"; +import { computerName } from "../mobile/devices"; import { childEnv, now, parseJson } from "../util"; import { resetDoctorCache, resolveClaudeBinary, runCommand, stripAnsi, toolPath } from "./doctor"; import { notify } from "./notifications"; @@ -113,8 +115,17 @@ export async function noteClaudeSignedOut({ apiKey, since }: { apiKey: boolean; if (claudeSignedOut() || session || works !== seen) return; log.warn(`a run found Claude Code signed out${apiKey ? " (Anthropic API key refused)" : ""}`); setSignedOut({ at: now(), apiKey, stamp }); - if (apiKey) notify("warning", "Claude refused the API key", "Your agents can't work until the Anthropic API key works again.", "/integrations?tab=api-keys"); - else notify("warning", "Claude is signed out", "Your agents can't work until you sign in to Claude again. It's one click in Godmode.", "/settings/system"); + // One notice for the whole outage, the runner's too (it reaches the computer it works for): runs fail until it's over. + const on = config().role === "runner" ? ` on ${computerName()}` : ""; + if (apiKey) notify("error", `Claude refused the API key${on}`, "Your agents and scheduled automations fail until the Anthropic API key works again.", "/integrations?tab=api-keys"); + else { + notify( + "error", + `Claude is signed out${on}`, + "Your agents and scheduled automations fail until you sign in again: run `claude` in a terminal and type /login, or sign in from Godmode with one click. Automations that failed meanwhile get one more run then, unless they are due again soon.", + "/settings/system", + ); + } } /** Claude answered a run: whatever found it signed out is over. */ diff --git a/packages/core/src/services/runNotices.ts b/packages/core/src/services/runNotices.ts index 5527d22e..e0e2870d 100644 --- a/packages/core/src/services/runNotices.ts +++ b/packages/core/src/services/runNotices.ts @@ -4,7 +4,8 @@ * itself (notify_user), or it stopped on a missing login (that has its own notice). */ import type { Run } from "@godmode/shared"; -import { get, run as sql } from "../db"; +import { runEndOf } from "@godmode/shared"; +import { all, get, run as sql } from "../db"; import { bus } from "../events/bus"; import { logger } from "../log"; import { isConversationViewed, setConversationViewHandler } from "../server/ws"; @@ -55,18 +56,50 @@ function automationNotice(r: Run, conversationId: string, viewed: boolean): void const setting = routine.notify ?? "failures"; if (setting === "never" || viewed || runNotifiedUser(r.id) || reportedMissingLogin(r.id)) return; if (r.status === "failed") { - const before = get<{ status: string }>( - "SELECT status FROM runs WHERE routine_id = ? AND trigger = 'routine' AND id != ? AND status IN ('succeeded', 'failed') ORDER BY created_at DESC LIMIT 1", + // Claude being signed out has a notice of its own for all the runs it fails (services/claudeLogin.ts), a moment later. + if (runEndOf(r.error ?? "")?.kind === "auth") { + setTimeout(() => { + try { + if (!signOutNoticed(r)) failureNotice(r, routine.name, conversationId); + } catch (err) { + log.warn(`could not handle the end of run ${r.id}`, err); + } + }, signOutWaitMs).unref?.(); + return; + } + // Once until it works again; a run that failed over the sign-in says nothing about that. + const before = all<{ status: string; error: string | null }>( + "SELECT status, error FROM runs WHERE routine_id = ? AND trigger = 'routine' AND id != ? AND status IN ('succeeded', 'failed') ORDER BY created_at DESC LIMIT 20", r.routineId!, r.id, - ); + ).find((b) => runEndOf(b.error ?? "")?.kind !== "auth"); if (before?.status === "failed") return; - notify("error", `“${routine.name}” failed`, shorten(r.error ?? "Its last run failed."), `/chat/${conversationId}`); + failureNotice(r, routine.name, conversationId); } else if (setting === "always") { notify("success", `“${routine.name}” finished`, shorten((r.result ?? "").replace(/\s+/g, " ").trim()), `/chat/${conversationId}`); } } +function failureNotice(r: Run, name: string, conversationId: string): void { + notify("error", `“${name}” failed`, shorten(r.error ?? "Its last run failed."), `/chat/${conversationId}`); +} + +/** Claude's sign-in notice went out since anything last worked before this run (on this computer or a runner). */ +function signOutNoticed(r: Run): boolean { + const worked = get<{ at: string }>("SELECT finished_at AS at FROM runs WHERE status = 'succeeded' AND finished_at <= ? ORDER BY finished_at DESC LIMIT 1", r.createdAt)?.at; + return !!get( + "SELECT 1 FROM notifications WHERE created_at >= ? AND (title LIKE 'Claude is signed out%' OR title LIKE 'Claude refused the API key%') LIMIT 1", + worked ?? new Date(Date.parse(r.createdAt) - 86_400_000).toISOString(), + ); +} + +let signOutWaitMs = 10_000; + +/** Tests: how long a failure over the sign-in waits for Claude's sign-in notice. */ +export function __setSignOutWaitForTests(ms: number): void { + signOutWaitMs = ms; +} + let unsubscribe: (() => void) | null = null; export function startRunNotices(): void { diff --git a/packages/core/test/sign-in-retry.test.ts b/packages/core/test/sign-in-retry.test.ts new file mode 100644 index 00000000..b51e09a7 --- /dev/null +++ b/packages/core/test/sign-in-retry.test.ts @@ -0,0 +1,91 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import type { Agent, Run } from "@godmode/shared"; +import { makeAgent, setupEnv, until, type TestEnv } from "./fixtures/runner-harness"; +import { all, getMeta, insert } from "../src/db"; +import { bus } from "../src/events/bus"; +import { automationConversation } from "../src/automations/conversation"; +import { newId, now } from "../src/util"; +import { createRoutine, updateRoutine } from "../src/services/routines"; +import { startChat } from "../src/services/conversations"; +import { claudeLoginStatus } from "../src/services/claudeLogin"; +import { __setSignOutWaitForTests, startRunNotices, stopRunNotices } from "../src/services/runNotices"; +import { triggerRoutine } from "../src/scheduler/scheduler"; +import { signInRetryIdle, startSignInRetry, stopSignInRetry } from "../src/automations/signInRetry"; +import { getRun, waitForRun } from "../src/runner/runner"; + +let env: TestEnv; +let agent: Agent; + +beforeAll(async () => { + env = await setupEnv("godmode-sign-in-retry-"); + agent = await makeAgent({ name: "Sign-in Retry Bot" }); + __setSignOutWaitForTests(300); + startRunNotices(); + startSignInRetry(); +}); + +afterAll(async () => { + stopSignInRetry(); + stopRunNotices(); + await env.close(); +}); + +const signedOutNotices = () => all<{ kind: string; body: string }>("SELECT kind, body FROM notifications WHERE title = 'Claude is signed out'"); +const runsOf = (routineId: string) => all<{ id: string; status: string }>("SELECT id, status FROM runs WHERE routine_id = ? AND trigger = 'routine' ORDER BY created_at, rowid", routineId); + +async function answered(): Promise { + const { run } = await startChat({ agentId: agent.id, content: "hello" }); + const finished = await waitForRun(run.id, 20_000); + expect(finished.status).toBe("succeeded"); + await Bun.sleep(100); + await signInRetryIdle(); + return finished; +} + +describe("automations that failed while Claude was signed out", () => { + test("one notice per outage, one more run once Claude answers, none when the next slot is near", async () => { + const yearly = createRoutine({ agentId: agent.id, name: "Yearly report", cron: "0 3 1 1 *", timezone: "UTC", prompt: "AUTH_EXPIRED" }); + const often = createRoutine({ agentId: agent.id, name: "Half-hourly check", cron: "*/30 * * * *", timezone: "UTC", prompt: "AUTH_EXPIRED" }); + const off = createRoutine({ agentId: agent.id, name: "Turned off later", cron: "0 4 1 1 *", timezone: "UTC", prompt: "AUTH_EXPIRED" }); + + for (const r of [yearly, often, off]) { + const run = await triggerRoutine(r.id); + expect((await waitForRun(run.id, 20_000)).status).toBe("failed"); + } + await until(() => !!claudeLoginStatus().signedOut); + await Bun.sleep(600); + // Several runs failed: one notice for all of them, none per automation. + expect(signedOutNotices()).toHaveLength(1); + expect(signedOutNotices()[0]!.kind).toBe("error"); + expect(signedOutNotices()[0]!.body).toContain("/login"); + expect(all("SELECT id FROM notifications WHERE title LIKE '“%” failed'")).toHaveLength(0); + + updateRoutine(off.id, { enabled: false }); + + await answered(); + await until(() => runsOf(yearly.id).length === 2); + expect(runsOf(often.id)).toHaveLength(1); + expect(runsOf(off.id)).toHaveLength(1); + const again = runsOf(yearly.id)[1]!; + expect(JSON.parse(getMeta("automations.signInRetried")!)).toContain(again.id); + + // The second try finds Claude signed out again: a new outage, a new notice. + expect((await waitForRun(again.id, 20_000)).status).toBe("failed"); + await until(() => signedOutNotices().length === 2); + + // It was the one more run: the next answer doesn't start it a third time. + await answered(); + await Bun.sleep(300); + expect(runsOf(yearly.id)).toHaveLength(2); + expect(runsOf(often.id)).toHaveLength(1); + }, 60_000); + + test("a failure that reads like a sign-in problem without a sign-in notice is still told", async () => { + const routine = createRoutine({ agentId: agent.id, name: "Lone failure", cron: "0 5 1 1 *", timezone: "UTC", prompt: "go" }); + const id = newId("run"); + const error = "Claude Code is not signed in (or the API key is invalid)."; + insert("runs", { id, agent_id: agent.id, conversation_id: automationConversation(routine), routine_id: routine.id, trigger: "routine", status: "failed", prompt: "go", error, created_at: now(), finished_at: now() }); + bus.emit({ type: "run.finished", run: getRun(id) }); + await until(() => all("SELECT id FROM notifications WHERE title = '“Lone failure” failed'").length === 1, 5_000); + }); +});