diff --git a/prompts/heartbeat.md b/prompts/heartbeat.md new file mode 100644 index 000000000..492ba3d13 --- /dev/null +++ b/prompts/heartbeat.md @@ -0,0 +1,28 @@ +You are Talon's background heartbeat agent. You run periodically (every {{intervalMinutes}} minutes) to perform maintenance tasks defined by the user. + +You have access ONLY to filesystem tools (Read, Write, Edit, Bash, Glob, Grep). Do NOT attempt to use any Telegram, MCP, or messaging tools. + +## Context + +- Workspace: `{{workspace}}` +- Memory file: `{{memoryFile}}` +- Logs directory: `{{logsDir}}` +- Last heartbeat: `{{lastRunIso}}` +- Run number: #{{runCount}} + +## Instructions + +Read the user-defined instructions file at `{{instructionsFile}}`. Follow whatever tasks are defined there. + +If the instructions file does not exist or is empty, perform these default tasks: + +1. **Review recent logs** — Check `{{logsDir}}/` for log files dated after `{{lastRunIso}}`. If `{{lastRunIso}}` is `never`, treat it as the beginning of time and review all available logs. Extract any new facts, preferences, or notable events. +2. **Update memory** — Merge any new information into `{{memoryFile}}`, keeping entries concise and factual. +3. **Workspace hygiene** — Note any issues but do not delete files unless the instructions explicitly say to. + +## Rules + +- Be surgical and precise. Do not rewrite files unnecessarily. +- Do not modify files outside the workspace unless the instructions explicitly allow it. +- Keep your work focused and efficient — you have a 10-minute time limit. +- When done, stop. The system handles all state tracking. diff --git a/src/__tests__/heartbeat.test.ts b/src/__tests__/heartbeat.test.ts new file mode 100644 index 000000000..43d500522 --- /dev/null +++ b/src/__tests__/heartbeat.test.ts @@ -0,0 +1,363 @@ +/** + * Tests for src/core/heartbeat.ts + * + * Covers: initHeartbeat, startHeartbeatTimer (double-start guard), + * forceHeartbeat (concurrency guard), state persistence semantics + * (success vs failure paths), and awaitCurrentRun. + * The actual SDK agent call is mocked to avoid spawning a real Claude agent. + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +// ── Mocks ───────────────────────────────────────────────────────────────── + +vi.mock("../util/log.js", () => ({ + log: vi.fn(), + logError: vi.fn(), + logWarn: vi.fn(), +})); + +const existsSyncMock = vi.fn(() => false); +const readFileSyncMock = vi.fn(() => "null"); +const mkdirSyncMock = vi.fn(); + +vi.mock("node:fs", () => ({ + existsSync: existsSyncMock, + readFileSync: readFileSyncMock, + mkdirSync: mkdirSyncMock, +})); + +const appendFileMock = vi.fn(async () => {}); +const mkdirAsyncMock = vi.fn(async () => undefined); +vi.mock("node:fs/promises", () => ({ + appendFile: appendFileMock, + mkdir: mkdirAsyncMock, +})); + +const writeAtomicSyncMock = vi.fn(); +vi.mock("write-file-atomic", () => ({ + default: { sync: writeAtomicSyncMock }, +})); + +// Mock the agent SDK so runHeartbeatAgent doesn't actually spawn Claude +// eslint-disable-next-line @typescript-eslint/no-explicit-any +const queryMock = vi.fn<() => AsyncGenerator>(async function* () { + // Yield nothing — simulates a clean run +}); +vi.mock("@anthropic-ai/claude-agent-sdk", () => ({ + query: queryMock, +})); + +vi.mock("../util/paths.js", () => ({ + files: { + heartbeatState: "/fake/.talon/workspace/memory/heartbeat_state.json", + dreamState: "/fake/.talon/workspace/memory/dream_state.json", + memory: "/fake/.talon/workspace/memory/memory.md", + log: "/fake/.talon/talon.log", + }, + dirs: { + root: "/fake/.talon", + logs: "/fake/.talon/workspace/logs", + workspace: "/fake/.talon/workspace", + data: "/fake/.talon/data", + memory: "/fake/.talon/workspace/memory", + prompts: "/fake/.talon/prompts", + }, +})); + +// ── Tests ───────────────────────────────────────────────────────────────── + +const { + initHeartbeat, + startHeartbeatTimer, + stopHeartbeatTimer, + forceHeartbeat, + getHeartbeatStatus, + awaitCurrentRun, +} = await import("../core/heartbeat.js"); + +describe("initHeartbeat", () => { + it("accepts a config object without throwing", () => { + expect(() => initHeartbeat({ model: "claude-sonnet-4-6" })).not.toThrow(); + }); + + it("accepts all optional config fields", () => { + expect(() => + initHeartbeat({ + model: "claude-sonnet-4-6", + heartbeatModel: "claude-haiku-4-5", + claudeBinary: "/usr/local/bin/claude", + workspace: "/tmp/test-workspace", + }), + ).not.toThrow(); + }); +}); + +describe("startHeartbeatTimer", () => { + beforeEach(() => { + vi.useFakeTimers(); + initHeartbeat({ model: "claude-sonnet-4-6" }); + }); + + afterEach(() => { + stopHeartbeatTimer(); + vi.useRealTimers(); + }); + + it("guards against double-start during startup delay", () => { + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); + + startHeartbeatTimer(60); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect(setIntervalSpy).not.toHaveBeenCalled(); + + // Calling again during the 5-minute startup delay should be a no-op + startHeartbeatTimer(60); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect(setIntervalSpy).not.toHaveBeenCalled(); + + setTimeoutSpy.mockRestore(); + setIntervalSpy.mockRestore(); + }); + + it("guards against double-start after interval is set", () => { + const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); + + startHeartbeatTimer(60); + expect(setIntervalSpy).not.toHaveBeenCalled(); + + // Advance past startup delay to create the interval timer + vi.advanceTimersByTime(5 * 60 * 1000 + 100); + expect(setIntervalSpy).toHaveBeenCalledTimes(1); + + // Now try to start again — should be a no-op (interval count stays at 1) + startHeartbeatTimer(60); + expect(setIntervalSpy).toHaveBeenCalledTimes(1); + + setIntervalSpy.mockRestore(); + }); +}); + +describe("forceHeartbeat", () => { + beforeEach(() => { + initHeartbeat({ model: "claude-sonnet-4-6" }); + existsSyncMock.mockReturnValue(false); + readFileSyncMock.mockReset(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + readFileSyncMock.mockImplementation(((filePath: string) => { + const p = String(filePath).replace(/\\/g, "/"); + if (p.endsWith("/heartbeat.md")) + return "heartbeat prompt {{workspace}} {{logsDir}} {{lastRunIso}} {{memoryFile}} {{instructionsFile}} {{runCount}} {{intervalMinutes}}"; + return "null"; + }) as any); + writeAtomicSyncMock.mockClear(); + queryMock.mockClear(); + }); + + it("writes heartbeat state twice (running then idle) on success", async () => { + await forceHeartbeat(); + + expect(writeAtomicSyncMock).toHaveBeenCalledTimes(2); + + const firstCall = JSON.parse( + writeAtomicSyncMock.mock.calls[0][1] as string, + ); + expect(firstCall.status).toBe("running"); + + const secondCall = JSON.parse( + writeAtomicSyncMock.mock.calls[1][1] as string, + ); + expect(secondCall.status).toBe("idle"); + }); + + it("increments run_count only on success", async () => { + await forceHeartbeat(); + + const finalState = JSON.parse( + writeAtomicSyncMock.mock.calls[ + writeAtomicSyncMock.mock.calls.length - 1 + ][1] as string, + ); + expect(finalState.run_count).toBe(1); + expect(finalState.status).toBe("idle"); + }); + + it("preserves previous last_run on failure", async () => { + const previousLastRun = Date.now() - 3600_000; + existsSyncMock.mockReturnValue(true); + readFileSyncMock.mockReturnValue( + JSON.stringify({ + last_run: previousLastRun, + status: "idle", + run_count: 5, + }), + ); + + // Make agent throw + queryMock.mockImplementationOnce(async function* () { + throw new Error("Agent exploded"); + }); + + await expect(forceHeartbeat()).rejects.toThrow("Agent exploded"); + + // The last write should preserve previous last_run and run_count + const finalState = JSON.parse( + writeAtomicSyncMock.mock.calls[ + writeAtomicSyncMock.mock.calls.length - 1 + ][1] as string, + ); + expect(finalState.last_run).toBe(previousLastRun); + expect(finalState.run_count).toBe(5); + expect(finalState.status).toBe("idle"); + }); + + it("sets last_started even on failure", async () => { + existsSyncMock.mockReturnValue(false); + + queryMock.mockImplementationOnce(async function* () { + throw new Error("Boom"); + }); + + await expect(forceHeartbeat()).rejects.toThrow("Boom"); + + const finalState = JSON.parse( + writeAtomicSyncMock.mock.calls[ + writeAtomicSyncMock.mock.calls.length - 1 + ][1] as string, + ); + expect(finalState.last_started).toBeGreaterThan(0); + }); + + it("rejects concurrent runs (concurrency guard)", async () => { + const firstRun = forceHeartbeat().catch(() => {}); + + // The running flag should now be true + await expect(forceHeartbeat()).rejects.toThrow("Heartbeat already running"); + await firstRun; + }); + + it("resolves successfully when agent mock yields no messages", async () => { + await expect(forceHeartbeat()).resolves.toBeUndefined(); + }); +}); + +describe("getHeartbeatStatus", () => { + it("returns null when no state file exists", () => { + existsSyncMock.mockReturnValue(false); + expect(getHeartbeatStatus()).toBeNull(); + }); + + it("returns parsed state when file exists", () => { + const state = { + last_run: Date.now(), + status: "idle", + run_count: 3, + }; + existsSyncMock.mockReturnValue(true); + readFileSyncMock.mockReturnValue(JSON.stringify(state)); + const result = getHeartbeatStatus(); + expect(result?.run_count).toBe(3); + expect(result?.status).toBe("idle"); + }); + + it("returns null for corrupt JSON", () => { + existsSyncMock.mockReturnValue(true); + readFileSyncMock.mockReturnValue("{ invalid json "); + expect(getHeartbeatStatus()).toBeNull(); + }); + + it("returns null when last_run is not a number", () => { + existsSyncMock.mockReturnValue(true); + readFileSyncMock.mockReturnValue( + JSON.stringify({ + last_run: "not-a-number", + status: "idle", + run_count: 0, + }), + ); + expect(getHeartbeatStatus()).toBeNull(); + }); + + it("returns null when run_count is not a number", () => { + existsSyncMock.mockReturnValue(true); + readFileSyncMock.mockReturnValue( + JSON.stringify({ + last_run: Date.now(), + status: "idle", + run_count: "five", + }), + ); + expect(getHeartbeatStatus()).toBeNull(); + }); + + it("returns null when status is invalid", () => { + existsSyncMock.mockReturnValue(true); + readFileSyncMock.mockReturnValue( + JSON.stringify({ + last_run: Date.now(), + status: "broken", + run_count: 1, + }), + ); + expect(getHeartbeatStatus()).toBeNull(); + }); + + it("returns null when last_run is non-finite", () => { + existsSyncMock.mockReturnValue(true); + // 1e309 parses to Infinity, which is typeof number but not finite + readFileSyncMock.mockReturnValue( + '{"last_run":1e309,"status":"idle","run_count":0}', + ); + expect(getHeartbeatStatus()).toBeNull(); + }); +}); + +describe("awaitCurrentRun", () => { + beforeEach(() => { + initHeartbeat({ model: "claude-sonnet-4-6" }); + existsSyncMock.mockReturnValue(false); + readFileSyncMock.mockReset(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + readFileSyncMock.mockImplementation(((filePath: string) => { + const p = String(filePath).replace(/\\/g, "/"); + if (p.endsWith("/heartbeat.md")) + return "heartbeat prompt {{workspace}} {{logsDir}} {{lastRunIso}} {{memoryFile}} {{instructionsFile}} {{runCount}} {{intervalMinutes}}"; + return "null"; + }) as any); + writeAtomicSyncMock.mockClear(); + }); + + it("resolves immediately when no run is in progress", async () => { + await expect(awaitCurrentRun()).resolves.toBeUndefined(); + }); + + it("waits for in-flight run to complete", async () => { + let resolveAgent!: () => void; + const agentPromise = new Promise((r) => { + resolveAgent = r; + }); + + queryMock.mockImplementationOnce(async function* () { + await agentPromise; + }); + + const runPromise = forceHeartbeat().catch(() => {}); + + // awaitCurrentRun should not resolve until the agent finishes + let awaited = false; + const awaitPromise = awaitCurrentRun().then(() => { + awaited = true; + }); + + // Give microtasks a chance to run + await new Promise((r) => setTimeout(r, 10)); + expect(awaited).toBe(false); + + // Now resolve the agent + resolveAgent(); + await awaitPromise; + await runPromise; + expect(awaited).toBe(true); + }); +}); diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 96ce6ad78..98718f89a 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -21,6 +21,7 @@ import { initDispatcher } from "./core/dispatcher.js"; import { initPulse, resetPulseTimer } from "./core/pulse.js"; import { initCron } from "./core/cron.js"; import { initDream } from "./core/dream.js"; +import { initHeartbeat } from "./core/heartbeat.js"; import { log } from "./util/log.js"; import type { TalonConfig } from "./util/config.js"; import type { QueryBackend, ContextManager } from "./core/types.js"; @@ -124,4 +125,10 @@ export async function initBackendAndDispatcher( claudeBinary: config.claudeBinary, workspace: config.workspace, }); + initHeartbeat({ + model: config.model, + heartbeatModel: config.heartbeatModel, + claudeBinary: config.claudeBinary, + workspace: config.workspace, + }); } diff --git a/src/core/heartbeat.ts b/src/core/heartbeat.ts new file mode 100644 index 000000000..00f3689d6 --- /dev/null +++ b/src/core/heartbeat.ts @@ -0,0 +1,512 @@ +/** + * Heartbeat — periodic background agent for user-defined maintenance tasks. + * + * Runs at a configurable interval (default: 60 minutes). + * The agent reads instructions from ~/.talon/workspace/heartbeat-instructions.md + * and executes them using filesystem-only tools (no Telegram/MCP access). + * + * Modeled after dream.ts but more general-purpose. + */ + +import { existsSync, readFileSync, mkdirSync } from "node:fs"; +import { appendFile, mkdir } from "node:fs/promises"; +import { resolve } from "node:path"; +import writeFileAtomic from "write-file-atomic"; +import { query } from "@anthropic-ai/claude-agent-sdk"; +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import { files as pathFiles, dirs } from "../util/paths.js"; +import { log, logError, logWarn } from "../util/log.js"; + +// ── Types ──────────────────────────────────────────────────────────────────── + +export type HeartbeatState = { + /** Unix millisecond timestamp of the last successfully completed heartbeat run. */ + last_run: number; + /** Human-readable ISO timestamp of the last successfully completed heartbeat run. */ + last_run_at?: string; + /** Unix millisecond timestamp of the last time a heartbeat was started (success or failure). */ + last_started?: number; + /** "idle" when no heartbeat is running, "running" while one is active. */ + status: "idle" | "running"; + /** Total number of successfully completed heartbeat runs. */ + run_count: number; +}; + +// ── Constants ──────────────────────────────────────────────────────────────── + +const HEARTBEAT_STATE_FILE = pathFiles.heartbeatState; +const HEARTBEAT_TIMEOUT_MS = 10 * 60 * 1000; // 10-minute max +const HEARTBEAT_LOGS_DIR = resolve(dirs.logs, "heartbeats"); +const STARTUP_DELAY_MS = 5 * 60 * 1000; // 5-minute delay before first run + +// ── State ──────────────────────────────────────────────────────────────────── + +let running = false; // in-process guard (one heartbeat at a time) +let currentRunPromise: Promise | null = null; // tracks in-flight run for graceful shutdown +let timer: ReturnType | null = null; +let startupTimer: ReturnType | null = null; +let intervalMinutesRef = 60; // stored from startHeartbeatTimer +let configRef: { + model?: string; + heartbeatModel?: string; + claudeBinary?: string; + workspace?: string; +} | null = null; + +export function initHeartbeat(cfg: { + model?: string; + /** Override model used specifically for heartbeat (e.g. haiku for cost savings). Falls back to main model. */ + heartbeatModel?: string; + claudeBinary?: string; + workspace?: string; +}): void { + configRef = cfg; +} + +// ── Public API ─────────────────────────────────────────────────────────────── + +/** + * Start the heartbeat timer. First run happens after a 5-minute startup delay, + * then repeats at the configured interval. + */ +export function startHeartbeatTimer(intervalMinutes: number): void { + if (timer || startupTimer) return; // already running or scheduled to start + + if (!Number.isFinite(intervalMinutes) || intervalMinutes <= 0) { + logWarn( + "heartbeat", + `Refusing to start heartbeat timer with invalid intervalMinutes: ${intervalMinutes}`, + ); + return; + } + + intervalMinutesRef = intervalMinutes; + const intervalMs = intervalMinutes * 60 * 1000; + log( + "heartbeat", + `Starting heartbeat timer (every ${intervalMinutes}min, first run in 5min)`, + ); + + startupTimer = setTimeout(() => { + startupTimer = null; + // Run immediately after startup delay + executeHeartbeat("auto").catch(() => {}); + + // Then set up the recurring interval + timer = setInterval(() => { + executeHeartbeat("auto").catch(() => {}); + }, intervalMs); + }, STARTUP_DELAY_MS); +} + +/** + * Stop the heartbeat timer. Does not wait for in-flight runs. + * Use awaitCurrentRun() after this to wait for a running heartbeat to finish. + */ +export function stopHeartbeatTimer(): void { + if (startupTimer) { + clearTimeout(startupTimer); + startupTimer = null; + } + if (timer) { + clearInterval(timer); + timer = null; + log("heartbeat", "Heartbeat timer stopped"); + } +} + +/** + * Wait for any in-flight heartbeat run to complete. + * Call after stopHeartbeatTimer() during graceful shutdown. + */ +export async function awaitCurrentRun(timeoutMs = 10_000): Promise { + if (currentRunPromise) { + log("heartbeat", "Waiting for in-flight heartbeat to complete..."); + try { + await Promise.race([ + currentRunPromise, + new Promise((resolve) => { + const t = setTimeout(() => { + logWarn( + "heartbeat", + "In-flight heartbeat did not finish within shutdown budget, proceeding", + ); + resolve(); + }, timeoutMs); + t.unref(); + }), + ]); + } catch { + // Already logged in executeHeartbeat + } + } +} + +/** + * Force a heartbeat run immediately. + * Returns a promise that resolves when the heartbeat completes. + * Throws if a heartbeat is already running. + */ +export async function forceHeartbeat(): Promise { + if (running) throw new Error("Heartbeat already running"); + await executeHeartbeat("forced"); +} + +/** + * Get the current heartbeat status. + */ +export function getHeartbeatStatus(): HeartbeatState | null { + return readHeartbeatState(); +} + +// ── Core execution ────────────────────────────────────────────────────────── + +async function executeHeartbeat(trigger: "auto" | "forced"): Promise { + if (running) return; + + const state = readHeartbeatState(); + const now = Date.now(); + const previousRunCount = state?.run_count ?? 0; + const previousLastRun = state?.last_run ?? 0; + + running = true; + // Mark as running with last_started, but preserve last_run from previous successful run + writeHeartbeatState({ + last_run: previousLastRun, + last_started: now, + status: "running", + run_count: previousRunCount, + }); + log( + "heartbeat", + `${trigger === "forced" ? "Force-triggering" : "Triggering"} heartbeat #${previousRunCount + 1} (last run: ${previousLastRun ? new Date(previousLastRun).toISOString() : "never"})`, + ); + + const run = (async () => { + try { + const heartbeatLogPath = await runHeartbeatAgent( + previousLastRun, + previousRunCount + 1, + ); + // Only update last_run and increment run_count on success + writeHeartbeatState({ + last_run: Date.now(), + last_started: now, + status: "idle", + run_count: previousRunCount + 1, + }); + log( + "heartbeat", + `Heartbeat #${previousRunCount + 1} complete (${trigger}), log: ${heartbeatLogPath}`, + ); + } catch (err) { + logError( + "heartbeat", + `Heartbeat #${previousRunCount + 1} failed (${trigger})`, + err, + ); + // On failure, revert to idle but keep previous last_run and run_count + writeHeartbeatState({ + last_run: previousLastRun, + last_started: now, + status: "idle", + run_count: previousRunCount, + }); + if (trigger === "forced") throw err; + } finally { + running = false; + currentRunPromise = null; + } + })(); + + currentRunPromise = run; + await run; +} + +// ── Heartbeat agent ───────────────────────────────────────────────────────── + +async function runHeartbeatAgent( + lastRunTimestamp: number, + runCount: number, +): Promise { + if (!configRef) { + throw new Error("Heartbeat agent not initialized"); + } + + const lastRunIso = + lastRunTimestamp > 0 ? new Date(lastRunTimestamp).toISOString() : "never"; + + const logsDir = dirs.logs; + const memoryFile = pathFiles.memory; + const workspace = configRef.workspace ?? dirs.workspace; + const instructionsFile = resolve(workspace, "heartbeat-instructions.md"); + + // Load prompt template from the prompts directory (seeded to ~/.talon/prompts/) + const promptPath = resolve(dirs.prompts, "heartbeat.md"); + + let prompt: string; + try { + prompt = readFileSync(promptPath, "utf-8") + .replace(/\{\{workspace\}\}/g, workspace) + .replace(/\{\{logsDir\}\}/g, logsDir) + .replace(/\{\{lastRunIso\}\}/g, lastRunIso) + .replace(/\{\{memoryFile\}\}/g, memoryFile) + .replace(/\{\{instructionsFile\}\}/g, instructionsFile) + .replace(/\{\{runCount\}\}/g, String(runCount)) + .replace(/\{\{intervalMinutes\}\}/g, String(intervalMinutesRef)); + } catch { + throw new Error(`Failed to read heartbeat prompt from ${promptPath}`); + } + + const model = + configRef.heartbeatModel ?? configRef.model ?? "claude-sonnet-4-6"; + + // Set up heartbeat log file + const heartbeatLogFile = await createHeartbeatLogFile(); + await appendHeartbeatLog( + heartbeatLogFile, + `# Heartbeat Run #${runCount} — ${new Date().toISOString()}\n`, + ); + await appendHeartbeatLog( + heartbeatLogFile, + `**Trigger:** ${lastRunIso === "never" ? "first run" : `last_run=${lastRunIso}`}, model=${model}\n`, + ); + await appendHeartbeatLog( + heartbeatLogFile, + `**Prompt:**\n\`\`\`\n${prompt}\n\`\`\`\n\n---\n`, + ); + + const options = { + model, + systemPrompt: + "You are a background heartbeat agent for Talon. Use only filesystem tools. Follow the user-defined instructions precisely. Be efficient — you have limited time.", + cwd: workspace, + permissionMode: "bypassPermissions" as const, + allowDangerouslySkipPermissions: true, + ...(configRef.claudeBinary + ? { pathToClaudeCodeExecutable: configRef.claudeBinary } + : {}), + // No MCP servers — filesystem tools only + mcpServers: {}, + disallowedTools: [ + "EnterPlanMode", + "ExitPlanMode", + "EnterWorktree", + "ExitWorktree", + "TodoWrite", + "TodoRead", + "TaskCreate", + "TaskUpdate", + "TaskGet", + "TaskList", + "TaskOutput", + "TaskStop", + "AskUserQuestion", + "Agent", + ], + }; + + // NOTE: The timeout races against the agent promise but cannot abort the + // underlying Claude subprocess (the Agent SDK does not expose an abort + // mechanism). On timeout, we still await the agent promise to ensure the + // running lock is not released while the subprocess is still active. + let timeoutHandle: ReturnType | null = null; + const timeoutPromise = new Promise((_, reject) => { + timeoutHandle = setTimeout( + () => reject(new Error("Heartbeat agent timed out")), + HEARTBEAT_TIMEOUT_MS, + ); + }); + + const agentPromise = (async () => { + const qi = query({ + prompt, + options: options as Parameters[0]["options"], + }); + for await (const msg of qi) { + await logHeartbeatMessage(heartbeatLogFile, msg); + } + await appendHeartbeatLog( + heartbeatLogFile, + `\n---\n**Heartbeat #${runCount} completed at ${new Date().toISOString()}**\n`, + ); + })(); + + try { + await Promise.race([agentPromise, timeoutPromise]); + } catch (err) { + await appendHeartbeatLog( + heartbeatLogFile, + `\n---\n**Heartbeat #${runCount} FAILED at ${new Date().toISOString()}:** ${err}\n`, + ); + // On timeout, wait for the agent to actually finish before releasing the lock + // to prevent overlapping heartbeat runs + await agentPromise.catch(() => {}); + throw err; + } finally { + if (timeoutHandle) clearTimeout(timeoutHandle); + } + + return heartbeatLogFile; +} + +// ── Logging helpers ───────────────────────────────────────────────────────── + +let heartbeatLogFileSequence = 0; + +async function createHeartbeatLogFile(): Promise { + if (!existsSync(HEARTBEAT_LOGS_DIR)) { + await mkdir(HEARTBEAT_LOGS_DIR, { recursive: true }); + } + const now = new Date(); + const ts = now.toISOString().replace(/[:.]/g, "-"); + const seq = heartbeatLogFileSequence++; + return resolve(HEARTBEAT_LOGS_DIR, `heartbeat-${ts}-${seq}.md`); +} + +async function appendHeartbeatLog( + logFile: string, + text: string, +): Promise { + try { + await appendFile(logFile, text); + } catch (err) { + logError("heartbeat", "Failed to write heartbeat log", err); + } +} + +async function logHeartbeatMessage( + logFile: string, + msg: SDKMessage, +): Promise { + try { + const ts = new Date().toISOString().slice(11, 19); + + switch (msg.type) { + case "assistant": { + const textBlocks = msg.message.content + .filter((b) => b.type === "text") + .map((b) => ("text" in b ? (b as { text: string }).text : "")); + const toolUseBlocks = msg.message.content + .filter((b) => b.type === "tool_use") + .map((b) => { + const tu = b as { name: string; input: unknown }; + return `**Tool call:** \`${tu.name}\`\n\`\`\`json\n${JSON.stringify(tu.input, null, 2)}\n\`\`\``; + }); + + if (textBlocks.length > 0) { + await appendHeartbeatLog( + logFile, + `\n## [${ts}] Assistant\n${textBlocks.join("\n")}\n`, + ); + } + if (toolUseBlocks.length > 0) { + await appendHeartbeatLog( + logFile, + `\n${toolUseBlocks.join("\n\n")}\n`, + ); + } + break; + } + case "result": { + const result = + "result" in msg + ? (msg as { result: string }).result + : JSON.stringify(msg); + const truncated = + result.length > 2000 + ? result.slice(0, 2000) + "\n... (truncated)" + : result; + await appendHeartbeatLog( + logFile, + `\n### [${ts}] Result (${msg.subtype})\n\`\`\`\n${truncated}\n\`\`\`\n`, + ); + break; + } + case "system": { + await appendHeartbeatLog( + logFile, + `\n### [${ts}] System (${msg.subtype})\n`, + ); + break; + } + case "user": { + if (msg.tool_use_result != null) { + const raw = + typeof msg.tool_use_result === "string" + ? msg.tool_use_result + : JSON.stringify(msg.tool_use_result, null, 2); + const truncated = + raw.length > 2000 ? raw.slice(0, 2000) + "\n... (truncated)" : raw; + await appendHeartbeatLog( + logFile, + `\n### [${ts}] Tool Result\n\`\`\`\n${truncated}\n\`\`\`\n`, + ); + } + break; + } + default: + break; + } + } catch { + // Don't let logging errors break the heartbeat + } +} + +// ── State helpers ──────────────────────────────────────────────────────────── + +function normalizeHeartbeatState(parsed: unknown): HeartbeatState | null { + if (!parsed || typeof parsed !== "object") return null; + + const candidate = parsed as Record; + const { last_run, last_run_at, last_started, status, run_count } = candidate; + + if (typeof last_run !== "number" || !Number.isFinite(last_run)) return null; + if (typeof run_count !== "number" || !Number.isFinite(run_count)) return null; + if (status !== "idle" && status !== "running") return null; + if (last_run_at !== undefined && typeof last_run_at !== "string") return null; + if ( + last_started !== undefined && + (typeof last_started !== "number" || !Number.isFinite(last_started)) + ) { + return null; + } + + return { + last_run, + run_count, + status, + ...(last_run_at !== undefined ? { last_run_at } : {}), + ...(last_started !== undefined ? { last_started } : {}), + }; +} + +function readHeartbeatState(): HeartbeatState | null { + try { + if (!existsSync(HEARTBEAT_STATE_FILE)) return null; + const raw = readFileSync(HEARTBEAT_STATE_FILE, "utf-8"); + return normalizeHeartbeatState(JSON.parse(raw)); + } catch { + return null; + } +} + +function writeHeartbeatState(state: HeartbeatState): void { + try { + const dir = resolve(HEARTBEAT_STATE_FILE, ".."); + if (!existsSync(dir)) mkdirSync(dir, { recursive: true }); + const { last_run_at: _lastRunAt, ...rest } = state; + const enriched: HeartbeatState = { + ...rest, + ...(state.last_run !== 0 + ? { last_run_at: new Date(state.last_run).toISOString() } + : {}), + }; + writeFileAtomic.sync( + HEARTBEAT_STATE_FILE, + JSON.stringify(enriched, null, 2) + "\n", + ); + } catch (err) { + logError("heartbeat", "Failed to write heartbeat state", err); + } +} diff --git a/src/index.ts b/src/index.ts index fed6398a7..a713f05d6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,6 +15,11 @@ import { flushHistory } from "./storage/history.js"; import { flushMediaIndex } from "./storage/media-index.js"; import { getActiveCount } from "./core/dispatcher.js"; import { startPulseTimer, stopPulseTimer } from "./core/pulse.js"; +import { + startHeartbeatTimer, + stopHeartbeatTimer, + awaitCurrentRun as awaitHeartbeat, +} from "./core/heartbeat.js"; import { startCronTimer, stopCronTimer } from "./core/cron.js"; import { startWatchdog, stopWatchdog } from "./util/watchdog.js"; import { log, logError, logWarn } from "./util/log.js"; @@ -97,6 +102,8 @@ async function gracefulShutdown(signal: string): Promise { await destroyPlugins(); } stopPulseTimer(); + stopHeartbeatTimer(); + await awaitHeartbeat(); stopCronTimer(); stopWatchdog(); stopUploadCleanup(); @@ -147,6 +154,7 @@ async function main(): Promise { log("bot", "Starting Talon..."); if (config.pulse) startPulseTimer(config.pulseIntervalMs); + if (config.heartbeat) startHeartbeatTimer(config.heartbeatIntervalMinutes); startCronTimer(); startWatchdog(config.workspace); startUploadCleanup(config.workspace); diff --git a/src/util/config.ts b/src/util/config.ts index 0891d7d7f..5d52976f0 100644 --- a/src/util/config.ts +++ b/src/util/config.ts @@ -35,6 +35,9 @@ const configSchema = z.object({ adminUserId: z.number().int().optional(), pulse: z.boolean().default(true), pulseIntervalMs: z.number().int().min(60000).default(300000), + heartbeat: z.boolean().default(false), + heartbeatIntervalMinutes: z.number().int().min(5).default(60), + heartbeatModel: z.string().optional(), // Model for heartbeat agent (defaults to main model) braveApiKey: z.string().optional(), searxngUrl: z.string().default("http://localhost:8080"), timezone: z.string().optional(), diff --git a/src/util/log.ts b/src/util/log.ts index de4da6808..8c0d713a1 100644 --- a/src/util/log.ts +++ b/src/util/log.ts @@ -35,6 +35,7 @@ export type LogComponent = | "commands" | "cron" | "dream" + | "heartbeat" | "dispatcher" | "gateway" | "plugin" diff --git a/src/util/paths.ts b/src/util/paths.ts index e558eaa64..fa7786b99 100644 --- a/src/util/paths.ts +++ b/src/util/paths.ts @@ -77,4 +77,11 @@ export const files = { pid: resolve(TALON_ROOT, "talon.pid"), /** Dream mode state: ~/.talon/workspace/memory/dream_state.json */ dreamState: resolve(TALON_ROOT, "workspace", "memory", "dream_state.json"), + /** Heartbeat state: ~/.talon/workspace/memory/heartbeat_state.json */ + heartbeatState: resolve( + TALON_ROOT, + "workspace", + "memory", + "heartbeat_state.json", + ), } as const;