From 2f448b656982855840418ed010ce9885cf6e76f7 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Mon, 9 Nov 2026 01:54:56 +0300 Subject: [PATCH 1/2] feat(improve): lousho improve --watch with budgets, rate limits, lease and kill switch Co-Authored-By: Claude Opus 5.5 --- docs/cli.md | 18 +++ src/cli/improve.test.ts | 48 ++++++ src/cli/improve.ts | 93 ++++++++++- src/cli/improveAdmin.ts | 3 +- src/cli/improvePause.ts | 46 ++++++ src/cli/init/telegramTemplate.ts | 21 ++- src/improve/cycles.ts | 9 +- src/improve/heal.ts | 8 +- src/improve/index.ts | 3 + src/improve/monitor.ts | 26 +--- src/improve/pause.ts | 38 +++++ src/improve/settings.test.ts | 2 + src/improve/settings.ts | 7 + src/improve/state.ts | 37 +++++ src/improve/watch.test.ts | 257 +++++++++++++++++++++++++++++++ src/improve/watch.ts | 198 ++++++++++++++++++++++++ src/improve/watchSteps.ts | 137 ++++++++++++++++ 17 files changed, 912 insertions(+), 39 deletions(-) create mode 100644 src/cli/improvePause.ts create mode 100644 src/improve/pause.ts create mode 100644 src/improve/state.ts create mode 100644 src/improve/watch.test.ts create mode 100644 src/improve/watch.ts create mode 100644 src/improve/watchSteps.ts diff --git a/docs/cli.md b/docs/cli.md index bc479bb1..7135448b 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -495,6 +495,23 @@ npx lousho improve reject ./my-agent 3f9c0a1b2d4e --by ali --reason "too broad" and moves the file to `.lousho/proposals/rejected/`. A rejected proposal is never queued again. +`--watch` keeps the loop running on an agent directory: + +```bash +npx lousho improve ./my-agent --check ./check.mjs --watch --interval 15m +npx lousho improve pause ./my-agent --reason "release freeze" --by ali +npx lousho improve resume ./my-agent --by ali +``` + +Each tick checks the kill switch (`.lousho/improve.stop`, `LOUSHO_IMPROVE=off` +or `"enabled": false`, audited once as `pause`), takes a lease so only one +watcher runs per directory, runs the post-apply monitor, replays the corpus +every `guard.everyHours`, and then runs one cycle within the daily caps. A tick +with the same failures as the last one and no new failed run is skipped. +Three outages in a row write the stop file. Editor calls are limited to +`rateLimit.rpm` a minute. Ctrl-C stops after the current step. `pause` writes +the stop file and `resume` removes it; both are audited. + ### `improve.json` Self-healing settings live in `improve.json` beside the agent. Every key is @@ -510,6 +527,7 @@ optional, and an unknown key or a bad value is an error "monitor": { "minRuns": 20, "window": 20, "maxDrop": 0.05, "autoRollback": true, "giveUpDays": 7 }, "guard": { "runs": 3, "everyHours": 24, "maxCases": 30 }, "cooldownHours": 24, + "rateLimit": { "rpm": 20 }, "library": { "enabled": true } } ``` diff --git a/src/cli/improve.test.ts b/src/cli/improve.test.ts index e982f662..c4725661 100644 --- a/src/cli/improve.test.ts +++ b/src/cli/improve.test.ts @@ -611,6 +611,54 @@ describe('lousho improve, one self-healing cycle', () => { expect(fs.existsSync(path.join(root, '.lousho'))).toBe(false); }); + it('--watch runs cycles until stopped, and refuses a dry run', async () => { + const db = openDb(dbPath); + ingestBatch(db, stamped('s3')); + db.close(); + let stopNow: () => void = () => undefined; + const stop = new Promise((resolve) => { + stopNow = resolve; + }); + const lines: string[] = []; + const stdout = new Writable({ + write(chunk, _encoding, callback) { + lines.push(String(chunk)); + if (String(chunk).startsWith('lousho improve: ')) stopNow(); + callback(); + }, + }); + const code = await runImprove([agentDir, '--check', checkPath, '--db', dbPath, '--watch', '--interval', '1h', '--agent', 'support'], { + stdout, + stderr: sink().stream, + overrides: { provider: mockModel([fixedWhenForbidden], { onExhausted: 'repeat-last' }) }, + editor: editorFor(FIX), + stop, + }); + expect(code).toBe(0); + expect(lines.join('')).toMatch(/lousho improve: applied: Applied/); + expect(instructions()).toBe(FIX); + const dry = await run([agentDir, '--check', checkPath, '--db', dbPath, '--watch', '--dry-run'], { editor: editorFor(FIX) }); + expect(dry).toMatchObject({ code: 2, stderr: expect.stringMatching(/--watch needs an agent directory/) }); + expect(() => parseImproveArgs([agentDir, '--check', checkPath, '--interval', 'soon'])).toThrow(/--interval must be a positive duration/); + expect(parseImproveArgs([agentDir, '--check', checkPath, '--interval', '30s']).intervalMs).toBe(30_000); + }); + + it('pause writes the kill switch once and resume removes it, both audited', async () => { + const paused = await run(['pause', agentDir, '--reason', 'maintenance', '--by', 'ana']); + expect(paused).toMatchObject({ code: 0, stdout: expect.stringMatching(/Paused self-healing/) }); + expect(fs.readFileSync(path.join(agentDir, '.lousho', 'improve.stop'), 'utf8')).toBe('maintenance\n'); + expect((await run(['pause', agentDir])).stdout).toMatch(/already paused/); + const cycle = await run([agentDir, '--check', checkPath, '--db', dbPath], { editor: editorFor(FIX) }); + expect(cycle.stdout).toMatch(/lousho improve: paused: Paused: \.lousho\/improve\.stop exists/); + expect(instructions()).toBe('You answer refund questions.'); + expect((await run(['resume', agentDir, '--by', 'ana'])).stdout).toMatch(/Resumed self-healing/); + expect(fs.existsSync(path.join(agentDir, '.lousho', 'improve.stop'))).toBe(false); + expect(readAudit(agentDir).map((entry) => [entry.action, entry.actor])).toEqual([ + ['pause', 'person:ana'], + ['resume', 'person:ana'], + ]); + }); + it('an agent whose instructions are not in instructions.md is forced to dry run', async () => { fs.mkdirSync(path.join(agentDir, 'instructions')); fs.writeFileSync(path.join(agentDir, 'instructions', 'claude.md'), 'You answer refund questions for Claude.'); diff --git a/src/cli/improve.ts b/src/cli/improve.ts index e1da3dc6..54d6623c 100644 --- a/src/cli/improve.ts +++ b/src/cli/improve.ts @@ -26,6 +26,7 @@ import { scoreCase } from '../improve/cases'; import { agentInputOf } from '../improve/recoveries'; import type { HealCycle } from '../improve/cycles'; import { healOnce } from '../improve/heal'; +import { watchAgent } from '../improve/watch'; import { writtenHash } from '../improve/harnessWriter'; import { readAgentHarness } from '../improve/revisions'; import { loadImproveSettings, type ImproveSettings } from '../improve/settings'; @@ -39,7 +40,7 @@ import { isImproveAdmin, runImproveAdmin } from './improveAdmin'; import { dbPathOf, resolveAgent, sinceMs } from './observe'; const USAGE = - 'Usage: lousho improve --check [--model provider/model] [--editor provider/model] [--db path] [--since d|h|m] [--agent X] [--protect traceId]... [--read-only tool]... [--max-tokens n] [--out file] [--with-recoveries] [--approval auto|manual] [--dry-run]\n lousho improve revisions|rollback|audit|pending|approve|reject ...'; + 'Usage: lousho improve --check [--model provider/model] [--editor provider/model] [--db path] [--since d|h|m] [--agent X] [--protect traceId]... [--read-only tool]... [--max-tokens n] [--out file] [--with-recoveries] [--approval auto|manual] [--dry-run] [--watch [--interval 15m]]\n lousho improve revisions|rollback|audit|pending|approve|reject|pause|resume ...'; const HELP = `${USAGE} @@ -86,6 +87,12 @@ Options: keeps an edit that quotes a failed check, flagged on the card --dry-run print the review card and change nothing, from every failed run whatever its revision, as before self-healing + --watch keep running: every --interval (default 15m) check the kill + switch, take the one-watcher lease, run the post-apply monitor, + the corpus guard every guard.everyHours, then a cycle within + the daily caps. --agent names the agent whose runs the monitor + judges (default: the agent's name). Ctrl-C stops after the step + --interval m the tick interval for --watch, as s, m or h -h, --help show this help Revisions, rollback, the audit log and pending proposals (run one with --help for its flags): @@ -95,6 +102,8 @@ Revisions, rollback, the audit log and pending proposals (run one with --help fo lousho improve pending [--json] lousho improve approve --by lousho improve reject --by [--reason text] + lousho improve pause [--reason text] [--by ] write the kill switch file + lousho improve resume [--by ] remove it See docs/improve.md.`; @@ -116,6 +125,8 @@ const SPEC: CommandSpec = { 'with-recoveries': { type: 'boolean' }, approval: { type: 'string' }, 'dry-run': { type: 'boolean' }, + watch: { type: 'boolean' }, + interval: { type: 'string' }, }, }; @@ -137,6 +148,10 @@ export interface ImproveArgs { approval?: ImproveSettings['approval']; /** `--dry-run`: print the card and write nothing. */ dryRun?: boolean; + /** `--watch`: keep running cycles on a timer. */ + watch?: boolean; + /** `--interval`, in milliseconds. */ + intervalMs?: number; help?: boolean; } @@ -147,6 +162,8 @@ export interface ImproveIo { overrides?: CreateAgentConfig; /** The editors' provider, instead of `--editor` (tests pass a mock). */ editor?: LLMProvider; + /** Resolves when `--watch` should stop, instead of SIGINT or SIGTERM (tests pass one). */ + stop?: Promise; } const listValue = (value: unknown): string[] => (Array.isArray(value) ? value.filter((item): item is string => typeof item === 'string') : []); @@ -158,6 +175,16 @@ function maxTokensOf(text: string | undefined): number | undefined { return tokens; } +const INTERVAL_UNIT_MS: Record = { s: 1000, m: 60_000, h: 3_600_000 }; + +function intervalOf(text: string | undefined): Pick { + if (text === undefined) return {}; + const match = /^(\d+)([smh])$/.exec(text); + const count = Number(match?.[1] ?? 0); + if (match === null || count <= 0) throw usageError(SPEC, `--interval must be a positive duration like '15m', '30s' or '1h', not '${text}'.`); + return { intervalMs: count * INTERVAL_UNIT_MS[match[2]] }; +} + function approvalOf(text: string | undefined): Pick { if (text === undefined) return {}; if (text !== 'auto' && text !== 'manual') throw usageError(SPEC, `--approval must be 'auto' or 'manual', not '${text}'.`); @@ -187,6 +214,8 @@ export function parseImproveArgs(args: string[]): ImproveArgs { maxTokens: maxTokensOf(stringValue(values['max-tokens'])), withRecoveries: values['with-recoveries'] === true, dryRun: values['dry-run'] === true, + watch: values.watch === true, + ...intervalOf(stringValue(values.interval)), ...approvalOf(stringValue(values.approval)), }; } @@ -421,11 +450,15 @@ function reportCycle(cycle: HealCycle, args: ImproveArgs, io: ImproveIo): void { } } -async function healDir(args: ImproveArgs, io: ImproveIo, dir: string, config: CreateAgentConfig): Promise { +async function cycleInputs(args: ImproveArgs, io: ImproveIo, dir: string, config: CreateAgentConfig) { const settings = loadImproveSettings(dir, args.approval ? { approval: args.approval } : {}); const editor = editorOf(args, config, io); const check = await loadCheck(args.check); - const db = openObserveDb(args); + return { settings, editor, check, db: openObserveDb(args) }; +} + +async function healDir(args: ImproveArgs, io: ImproveIo, dir: string, config: CreateAgentConfig): Promise { + const { settings, editor, check, db } = await cycleInputs(args, io, dir, config); const controller = new AbortController(); const interrupt = () => controller.abort(); process.once('SIGINT', interrupt); @@ -459,15 +492,67 @@ async function healDir(args: ImproveArgs, io: ImproveIo, dir: string, config: Cr } } +function stopRequested(io: ImproveIo): { stopped: Promise; dispose(): void } { + if (io.stop) return { stopped: io.stop, dispose: () => undefined }; + let done: () => void = () => undefined; + const stopped = new Promise((resolve) => { + done = resolve; + }); + process.once('SIGINT', done); + process.once('SIGTERM', done); + return { + stopped, + dispose: () => { + process.off('SIGINT', done); + process.off('SIGTERM', done); + }, + }; +} + +async function watchDir(args: ImproveArgs, io: ImproveIo, dir: string, config: CreateAgentConfig): Promise { + const { settings, editor, check, db } = await cycleInputs(args, io, dir, config); + const name = args.agent ?? config.name; + if (name === undefined) throw usageError(SPEC, "--watch needs --agent: the agent doesn't name itself, so the monitor can't find its runs."); + const stop = stopRequested(io); + try { + const agent = resolveAgent(SPEC, db, name); + io.stderr.write(`lousho improve: watching ${dir} in ${settings.approval} mode every ${(args.intervalMs ?? 900_000) / 1000}s; 'lousho improve pause ${dir}' or Ctrl-C stops it.\n`); + await withoutObserve(async () => { + const watcher = watchAgent({ + dir, + db, + check, + runCase: runCaseWith(config, args, io, new Map()), + editor, + settings, + agent, + protect: args.protect, + withRecoveries: args.withRecoveries === true, + ...(args.since !== undefined && { since: args.since }), + ...(args.intervalMs !== undefined && { intervalMs: args.intervalMs }), + onCycle: (cycle) => io.stdout.write(`lousho improve: ${cycle.outcome}: ${cycle.reason}\n`), + onError: (error) => io.stderr.write(`lousho improve: a watch tick failed: ${error instanceof Error ? error.message : String(error)}\n`), + }); + await stop.stopped; + await watcher.stop(); + }); + return 0; + } finally { + stop.dispose(); + db.close(); + } +} + async function improve(args: ImproveArgs, io: ImproveIo): Promise { const target = detectTarget(args.path, 'improve'); const config = await loadConfig(target); if (args.out !== undefined) assertOutUsable(args.out, target, config); const harness = baseHarness(config); const forced = args.dryRun ? undefined : dryRunReason(target, harness); + if (args.watch && (args.dryRun || forced || target.kind !== 'dir')) throw usageError(SPEC, `--watch needs an agent directory whose instructions live in instructions.md, and can't be a dry run${forced ? `: ${forced}` : ''}.`); if (forced) io.stderr.write(`lousho improve: this is a dry run, because ${forced}. Nothing will be applied.\n`); if (args.dryRun || forced || target.kind !== 'dir') return proposeOnly(args, io, target, config, harness); - return healDir(args, io, target.path, config); + return args.watch ? watchDir(args, io, target.path, config) : healDir(args, io, target.path, config); } /** Runs `lousho improve` with the arguments after `improve`; resolves with the exit code. */ diff --git a/src/cli/improveAdmin.ts b/src/cli/improveAdmin.ts index 6cbee375..9700d495 100644 --- a/src/cli/improveAdmin.ts +++ b/src/cli/improveAdmin.ts @@ -6,6 +6,7 @@ import { readAudit, verifyAudit, type AuditEntry } from '../improve/audit'; import { approvePending, isPendingId, listPending, readPendingProposal, rejectPending } from '../improve/pending'; import { currentRevision, listRevisions, type HarnessRevision } from '../improve/revisions'; import { parseCommand, stringValue, usageError, type CommandSpec } from './args'; +import { pause, resume } from './improvePause'; interface AdminIo { stdout: Writable; @@ -186,7 +187,7 @@ async function reject(args: readonly string[], io: AdminIo): Promise { return 0; } -const COMMANDS: Record number | Promise> = { revisions, rollback, audit, pending, approve, reject }; +const COMMANDS: Record number | Promise> = { revisions, rollback, audit, pending, approve, reject, pause, resume }; /** Whether the arguments after `improve` name an admin subcommand rather than an agent path. */ export function isImproveAdmin(args: readonly string[]): boolean { diff --git a/src/cli/improvePause.ts b/src/cli/improvePause.ts new file mode 100644 index 00000000..8cb0f127 --- /dev/null +++ b/src/cli/improvePause.ts @@ -0,0 +1,46 @@ +import * as path from 'node:path'; +import type { Writable } from 'node:stream'; +import { pauseImprove, resumeImprove } from '../improve/pause'; +import { parseCommand, stringValue, usageError, type CommandSpec, type ParsedCommand } from './args'; + +interface PauseIo { + stdout: Writable; + stderr: Writable; +} + +const PAUSE: CommandSpec = { + command: 'improve pause', + usage: 'Usage: lousho improve pause [--reason text] [--by ]', + positionals: 1, + options: { reason: { type: 'string' }, by: { type: 'string' } }, +}; +const RESUME: CommandSpec = { command: 'improve resume', usage: 'Usage: lousho improve resume [--by ]', positionals: 1, options: { by: { type: 'string' } } }; + +function dirOf(spec: CommandSpec, args: readonly string[], io: PauseIo): { dir: string; values: ParsedCommand['values'] } | undefined { + const { values, positionals, help } = parseCommand(spec, [...args]); + if (help) { + io.stdout.write(`${spec.usage}\n`); + return undefined; + } + if (positionals.length !== 1) throw usageError(spec, 'an agent directory is required.'); + return { dir: path.resolve(positionals[0]), values }; +} + +export async function pause(args: readonly string[], io: PauseIo): Promise { + const parsed = dirOf(PAUSE, args, io); + if (!parsed) return 0; + const by = stringValue(parsed.values.by); + const reason = stringValue(parsed.values.reason); + const done = await pauseImprove(parsed.dir, { ...(by !== undefined && { by }), ...(reason !== undefined && { reason }) }); + io.stdout.write(done ? `Paused self-healing in ${parsed.dir}: wrote .lousho/improve.stop. A running watcher stops at its next tick.\n` : `Self-healing in ${parsed.dir} was already paused.\n`); + return 0; +} + +export async function resume(args: readonly string[], io: PauseIo): Promise { + const parsed = dirOf(RESUME, args, io); + if (!parsed) return 0; + const by = stringValue(parsed.values.by); + const done = await resumeImprove(parsed.dir, by !== undefined ? { by } : {}); + io.stdout.write(done ? `Resumed self-healing in ${parsed.dir}: removed .lousho/improve.stop.\n` : `Self-healing in ${parsed.dir} was not paused by the stop file.\n`); + return 0; +} diff --git a/src/cli/init/telegramTemplate.ts b/src/cli/init/telegramTemplate.ts index 8ab65a7b..9b22b8b0 100644 --- a/src/cli/init/telegramTemplate.ts +++ b/src/cli/init/telegramTemplate.ts @@ -222,7 +222,7 @@ const BRIDGE_TS = [ ' * - everyone else is logged and ignored — this bot can run tools on your', ' * machine, strangers must not get turns.', ' */', - "import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';", + "import { mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs';", "import { basename, dirname, join } from 'node:path';", "import type { Agent, Session } from 'lousho';", 'import {', @@ -238,7 +238,8 @@ const BRIDGE_TS = [ "} from './api.js';", "import { addRule, describeRule, findMatch, loadRules, removeRule } from '../approvals.js';", '', - "const STATE_FILE = '.lousho/state.json';", + "const STATE_FILE = '.lousho/telegram.json';", + "const LEGACY_STATE_FILE = '.lousho/state.json';", "const INBOX = '.lousho/inbox';", 'const ensureDir = (p: string) => (mkdirSync(p, { recursive: true }), p);', 'const STARTED_AT = Date.now();', @@ -247,16 +248,24 @@ const BRIDGE_TS = [ ' ownerChatId?: number;', ' ownerUserId?: number;', '}', - 'const loadState = (): BridgeState => {', + 'const readJson = (file: string): BridgeState | undefined => {', ' try {', - " return JSON.parse(readFileSync(STATE_FILE, 'utf8'));", + " return JSON.parse(readFileSync(file, 'utf8'));", ' } catch {', - ' return {};', + ' return undefined;', ' }', '};', + 'const loadState = (): BridgeState => {', + ' const state = readJson(STATE_FILE);', + ' if (state) return state;', + ' const { ownerChatId, ownerUserId } = readJson(LEGACY_STATE_FILE) ?? {};', + ' return { ownerChatId, ownerUserId };', + '};', 'const saveState = (patch: BridgeState) => {', ' ensureDir(dirname(STATE_FILE));', - ' writeFileSync(STATE_FILE, JSON.stringify({ ...loadState(), ...patch }));', + ' const temp = `${STATE_FILE}.${process.pid}.tmp`;', + ' writeFileSync(temp, JSON.stringify({ ...loadState(), ...patch }));', + ' renameSync(temp, STATE_FILE);', '};', '', "const botUsername = () => process.env.BOT_USERNAME ?? '';", diff --git a/src/improve/cycles.ts b/src/improve/cycles.ts index 1099dea9..13fdccb6 100644 --- a/src/improve/cycles.ts +++ b/src/improve/cycles.ts @@ -65,6 +65,13 @@ const DAY_MS = 24 * 60 * 60 * 1000; const HOUR_MS = 60 * 60 * 1000; const AUTO_ACTORS: readonly string[] = ['gate', 'canary']; +export function ownLousho(dir: string): string { + const lousho = path.join(dir, LOUSHO); + if (lstatSync(lousho, { throwIfNoEntry: false })?.isSymbolicLink()) throw new ConfigurationError(`${lousho} is a symlink; the self-healing state must live inside the agent directory.`, 'dir'); + mkdirSync(lousho, { recursive: true }); + return lousho; +} + /** A new cycle id: its UTC start time and a random suffix. */ export function newCycleId(now: number): string { const stamp = new Date(now).toISOString().replace(/[-:]/g, '').replace(/\.\d{3}/, ''); @@ -115,7 +122,7 @@ export function readCycles(dir: string): HealCycle[] { .sort((a, b) => b.id.localeCompare(a.id)); } -const dayStart = (now: number): number => Math.floor(now / DAY_MS) * DAY_MS; +export const dayStart = (now: number): number => Math.floor(now / DAY_MS) * DAY_MS; /** Cycles, tokens and automatic applies so far in the current UTC day. */ export function usageToday(dir: string, now: number): { cycles: number; tokens: number; applies: number } { diff --git a/src/improve/heal.ts b/src/improve/heal.ts index e1f18420..bb156b1f 100644 --- a/src/improve/heal.ts +++ b/src/improve/heal.ts @@ -1,4 +1,3 @@ -import { lstatSync, mkdirSync } from 'node:fs'; import path from 'node:path'; import { ConfigurationError, SDKError } from '../execution/errors'; import type { AgentHook } from '../execution/hooks'; @@ -9,7 +8,7 @@ import { appendAudit } from './audit'; import { canaryReplay, type CanaryReplayResult } from './canary'; import { admitProtectedCases, casesFromFailures, protectedCasesFromPasses, scoreCase, type CapturedCase, type CaseCheck } from './cases'; import { addToCorpus, readCorpus, type CorpusCase } from './corpus'; -import { cooldownEnds, newCycleId, pausedBy, readCycles, usageToday, writeCycle, type HealCycle } from './cycles'; +import { cooldownEnds, newCycleId, ownLousho, pausedBy, readCycles, usageToday, writeCycle, type HealCycle } from './cycles'; import { formatImproveResult } from './format'; import { applyAndGuard, corpusCases, guardUnfinished } from './healApply'; import { @@ -104,10 +103,7 @@ function ended(start: Start, outcome: HealCycle['outcome'], reason: string): Hea } function lockFile(dir: string): string { - const lousho = path.join(dir, LOUSHO); - if (lstatSync(lousho, { throwIfNoEntry: false })?.isSymbolicLink()) throw new ConfigurationError(`${lousho} is a symlink; the self-healing state must live inside the agent directory.`, 'dir'); - mkdirSync(lousho, { recursive: true }); - return path.join(lousho, 'cycle.lock'); + return path.join(ownLousho(dir), 'cycle.lock'); } function blocked(options: HealOptions, start: Start, now: number): HealCycle | number { diff --git a/src/improve/index.ts b/src/improve/index.ts index f3dc12f7..1e120a79 100644 --- a/src/improve/index.ts +++ b/src/improve/index.ts @@ -33,3 +33,6 @@ export type { GuardCorpusOptions, GuardCorpusResult, GuardedCase } from './corpu export { healOnce } from './heal'; export type { HealOptions } from './heal'; export type { HealCycle } from './cycles'; +export { improveSchedule, watchAgent } from './watch'; +export type { WatchOptions, WatchTick } from './watch'; +export { pauseImprove, resumeImprove } from './pause'; diff --git a/src/improve/monitor.ts b/src/improve/monitor.ts index 734ab463..22ef86ac 100644 --- a/src/improve/monitor.ts +++ b/src/improve/monitor.ts @@ -1,14 +1,14 @@ -import { mkdirSync, readFileSync } from 'node:fs'; +import { mkdirSync } from 'node:fs'; import path from 'node:path'; import { pointOf, significantDrop, trajectoryGate, windowSummary, type ScorePoint, type WindowStat } from '../observe/continuousScore'; import { finishedRuns } from '../observe/store'; import { ConfigurationError } from '../execution/errors'; import type { SqlDatabase } from '../storage/sqlite/driver'; -import { writeFileAtomic } from '../storage/atomicFile'; import { withFileLock } from '../storage/fileLock'; import { rollbackHarness } from './apply'; import { appendAudit, readAudit, type AuditEntry } from './audit'; import { currentRevision } from './revisions'; +import { readState, updateState } from './state'; import type { ImproveSettings } from './settings'; /** What the monitor decided about the current revision. */ @@ -41,8 +41,6 @@ const GATE_WINDOW = 10; const DAY_MS = 86_400_000; const HOUR_MS = 3_600_000; const SERIES_LIMIT = 100_000; -const STATE_FILE = path.join('.lousho', 'state.json'); -const STATE_LOCK = path.join('.lousho', 'state.lock'); const MONITOR_LOCK = path.join('.lousho', 'monitor.lock'); const FINAL: readonly MonitorVerdict['state'][] = ['kept', 'kept-unverified', 'regressed', 'rolled-back']; @@ -80,24 +78,10 @@ interface MonitorState { [key: string]: unknown; } -function readState(dir: string): MonitorState { - const file = path.join(dir, STATE_FILE); - let text: string; - try { - text = readFileSync(file, 'utf8'); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') return {}; - throw error; - } - const parsed: unknown = JSON.parse(text); - if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) throw new ConfigurationError(`${file} is not a JSON object; fix or delete it before the monitor runs.`, 'state', 'LOUSHO_CONFIG_INVALID'); - return parsed as MonitorState; -} +const readMonitorState = (dir: string): Partial => readState(dir); async function writeState(dir: string, change: Partial): Promise { - await withFileLock(path.join(dir, STATE_LOCK), async () => { - await writeFileAtomic(path.join(dir, STATE_FILE), `${JSON.stringify({ ...readState(dir), ...change }, null, 2)}\n`); - }); + await updateState(dir, change); } function lastApply(dir: string, current: string): AuditEntry | string { @@ -165,7 +149,7 @@ async function check(db: SqlDatabase, dir: string, settings: CheckSettings, now: const agent = agentIdOf(db, name); const apply = lastApply(dir, currentRevision(dir)); if (typeof apply === 'string') return { state: 'idle', reason: apply }; - const stored = readState(dir).monitor; + const stored = readMonitorState(dir).monitor; if (stored?.applySeq === apply.seq && FINAL.includes(stored.verdict.state)) return stored.verdict; const { before, after } = seriesAround(db, apply, settings.monitor, agent); const judged = judgeApplied(before, after, settings.monitor); diff --git a/src/improve/pause.ts b/src/improve/pause.ts new file mode 100644 index 00000000..28d899ad --- /dev/null +++ b/src/improve/pause.ts @@ -0,0 +1,38 @@ +import { lstatSync, rmSync, writeFileSync } from 'node:fs'; +import path from 'node:path'; +import { appendAudit, type AuditEntry } from './audit'; +import { ownLousho } from './cycles'; +import { updateState } from './state'; +import type { WatchState } from './watchSteps'; + +const STOP_FILE = path.join('.lousho', 'improve.stop'); + +function stopFile(dir: string): string { + return path.join(ownLousho(dir), 'improve.stop'); +} + +const actorOf = (by: string | undefined): AuditEntry['actor'] => (by === undefined || by.trim() === '' ? 'scheduler' : `person:${by.trim()}`); + +/** Turns the self-healing loop off for a directory by writing .lousho/improve.stop, and audits `pause` once. */ +export async function pauseImprove(dir: string, options: { by?: string; reason?: string } = {}): Promise { + const file = stopFile(dir); + if (lstatSync(file, { throwIfNoEntry: false })) return false; + const reason = options.reason?.trim() || 'Paused from the command line.'; + writeFileSync(file, `${reason}\n`, { flag: 'wx' }); + await appendAudit(dir, { actor: actorOf(options.by), action: 'pause', reason }); + await updateState(dir, (current) => ({ watch: { ...current.watch, paused: `${STOP_FILE} exists` } })); + return true; +} + +/** Removes .lousho/improve.stop and audits `resume`; `improve.json` `enabled: false` and `LOUSHO_IMPROVE=off` still stop the loop. */ +export async function resumeImprove(dir: string, options: { by?: string } = {}): Promise { + const file = stopFile(dir); + if (!lstatSync(file, { throwIfNoEntry: false })) return false; + rmSync(file, { force: true }); + await appendAudit(dir, { actor: actorOf(options.by), action: 'resume', reason: 'The stop file was removed.' }); + await updateState(dir, (current) => { + const { paused: _paused, ...rest } = current.watch ?? {}; + return { watch: rest }; + }); + return true; +} diff --git a/src/improve/settings.test.ts b/src/improve/settings.test.ts index 62d5d426..6f16eb4c 100644 --- a/src/improve/settings.test.ts +++ b/src/improve/settings.test.ts @@ -28,6 +28,7 @@ describe('loadImproveSettings', () => { monitor: { minRuns: 20, window: 20, maxDrop: 0.05, autoRollback: true, giveUpDays: 7 }, guard: { runs: 3, everyHours: 24, maxCases: 30 }, cooldownHours: 24, + rateLimit: { rpm: 20 }, library: { enabled: true }, }); }); @@ -73,6 +74,7 @@ describe('loadImproveSettings', () => { monitor: { minRuns: 10, window: 15, maxDrop: 0.1, autoRollback: false, giveUpDays: 3 }, guard: { runs: 2, everyHours: 12, maxCases: 40 }, cooldownHours: 6, + rateLimit: { rpm: 5 }, library: { enabled: false, domain: 'billing' }, }; file({ $schema: './improve.schema.json', ...settings }); diff --git a/src/improve/settings.ts b/src/improve/settings.ts index 9ee12732..b6f4c3dd 100644 --- a/src/improve/settings.ts +++ b/src/improve/settings.ts @@ -63,6 +63,11 @@ export interface ImproveSettings { }; /** Hours without cycles after a rollback (default 24). */ cooldownHours: number; + /** The watcher's limit on editor and reflection model calls. */ + rateLimit: { + /** Model calls per minute (default 20). */ + rpm: number; + }; /** The lesson library editors read. */ library: { /** Use the library (default `true`). */ @@ -92,6 +97,7 @@ const fileSchema = z .optional(), guard: z.object({ runs: count.optional(), everyHours: z.number().positive().finite().optional(), maxCases: count.optional() }).strict().optional(), cooldownHours: z.number().min(0).finite().optional(), + rateLimit: z.object({ rpm: count.optional() }).strict().optional(), library: z.object({ enabled: z.boolean().optional(), domain: z.string().regex(/^[a-z0-9][a-z0-9-]*$/).optional() }).strict().optional(), }) .strict(); @@ -144,6 +150,7 @@ export function loadImproveSettings(dir: string, overrides: Partial(dir: string): Partial { + const file = path.join(dir, STATE_FILE); + let text: string; + try { + text = readFileSync(file, 'utf8'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return {}; + throw error; + } + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + parsed = undefined; + } + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) throw new ConfigurationError(`${file} is not a JSON object; fix or delete it before the self-healing loop runs.`, 'state', 'LOUSHO_CONFIG_INVALID'); + return parsed as Partial; +} + +export async function updateState(dir: string, change: Partial | ((current: Partial) => Partial)): Promise> { + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + return withFileLock(path.join(dir, STATE_LOCK), async () => { + const current = readState(dir); + const next = { ...current, ...(typeof change === 'function' ? change(current) : change) }; + await writeFileAtomic(path.join(dir, STATE_FILE), `${JSON.stringify(next, null, 2)}\n`); + return next; + }); +} diff --git a/src/improve/watch.test.ts b/src/improve/watch.test.ts new file mode 100644 index 00000000..2098c2a1 --- /dev/null +++ b/src/improve/watch.test.ts @@ -0,0 +1,257 @@ +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { IngestEvent } from '../observe/envelope'; +import { ingestBatch, openDb } from '../observe/store'; +import type { SqlDatabase } from '../storage/sqlite/driver'; +import { mockModel, type MockRequest } from '../testing'; +import { readAudit } from './audit'; +import { readCorpus } from './corpus'; +import { readCycles } from './cycles'; +import { healOnce, type HealOptions } from './heal'; +import { checkApplied } from './monitor'; +import { currentRevision } from './revisions'; +import { loadImproveSettings, type ImproveSettings } from './settings'; +import { updateState } from './state'; +import type { Harness } from './types'; +import { isDefinedSchedule } from '../schedules/defineSchedule'; +import { improveSchedule, watchAgent, watchContext, watchTick, type WatchOptions } from './watch'; + +vi.mock('./heal', async (original) => { + const actual = await original(); + return { ...actual, healOnce: vi.fn(actual.healOnce) }; +}); + +vi.mock('./monitor', async (original) => { + const actual = await original(); + return { ...actual, checkApplied: vi.fn(actual.checkApplied) }; +}); + +const BASE = 'You answer refund questions.'; +const FIX = 'You answer refund questions. If a document is forbidden, say so and stop.'; +const T0 = Date.now() - 60_000; + +let root: string; +let dir: string; +let db: SqlDatabase; +let traces = 0; + +const systemOf = (request: MockRequest): string => String(request.messages.find((message) => message.role === 'system')?.content ?? ''); + +const editor = () => + mockModel( + [ + (request) => { + if (systemOf(request).includes('finished attempt to improve')) return 'NO_ENTRY'; + if (request.messages.at(-1)?.role === 'tool') return 'Traced to a failed run.'; + return { toolCalls: [{ name: 'write_file', args: { path: 'harness/instructions.md', content: FIX } }] }; + }, + ], + { onExhausted: 'repeat-last' } + ); + +const answers = (harness: Harness): string => (harness.instructions.includes('forbidden') ? 'ok, that is forbidden' : 'here is the doc'); +const runCase: HealOptions['runCase'] = async (harness, input) => ({ answer: input.includes('refund window') ? 'five days' : answers(harness), transcript: `assistant: ${answers(harness)}`, tokens: 10, ended: 'stop' }); +const check: HealOptions['check'] = (input, answer) => input.includes('refund window') || answer.startsWith('ok'); + +function root_(trace: string, input: string, start = T0): IngestEvent { + return { + type: 'observation', + data: { obs_id: `r-${trace}`, trace_id: trace, agent_id: 'support', kind: 'span', name: 'invoke_agent support', start, end: start + 9, status: 'ok', input, output: 'done', attrs: { 'gen_ai.operation.name': 'invoke_agent', 'lousho.harness.revision': currentRevision(dir) } }, + }; +} + +function seedFailures(count = 3): void { + const events = Array.from({ length: count }, (): IngestEvent[] => { + const trace = `t${traces++}`; + const at = Date.now() + traces; + return [ + root_(trace, '"can I read the secret doc?"', at), + { + type: 'observation', + data: { obs_id: `x-${trace}`, trace_id: trace, parent_obs_id: `r-${trace}`, agent_id: 'support', kind: 'tool', name: 'execute_tool lookup', start: at, end: at + 5, status: 'error', attrs: { 'gen_ai.operation.name': 'execute_tool', 'gen_ai.tool.name': 'lookup', error: 'HTTP 403 Forbidden' } }, + }, + ]; + }).flat(); + ingestBatch(db, events); +} + +function settings(over: Partial = {}): ImproveSettings { + return { ...loadImproveSettings(dir, {}, {}), library: { enabled: false }, ...over }; +} + +const options = (over: Partial = {}): WatchOptions => ({ dir, db, check, runCase, editor: { provider: editor() }, settings: settings(), agent: 'support', ...over }); + +const tick = (over: Partial = {}, context = watchContext(options(over))) => watchTick(options(over), context); + +const actions = () => readAudit(dir).map((entry) => entry.action); +const stop = () => { + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + writeFileSync(path.join(dir, '.lousho', 'improve.stop'), 'maintenance\n'); +}; + +beforeEach(() => { + root = mkdtempSync(path.join(tmpdir(), 'lousho-watch-')); + dir = path.join(root, 'support'); + mkdirSync(dir, { recursive: true }); + writeFileSync(path.join(dir, 'instructions.md'), `${BASE}\n`); + db = openDb(path.join(root, 'observe.db')); + ingestBatch(db, [root_('p1', '"what is the refund window?"')]); +}); + +afterEach(() => { + db.close(); + vi.mocked(healOnce).mockClear(); + vi.mocked(checkApplied).mockClear(); + vi.useRealTimers(); + rmSync(root, { recursive: true, force: true }); +}); + +describe('watchTick', () => { + it('the stop file prevents any cycle and is audited once', async () => { + seedFailures(); + stop(); + expect((await tick()).state).toBe('paused'); + expect((await tick()).state).toBe('paused'); + expect(healOnce).not.toHaveBeenCalled(); + expect(actions()).toEqual(['pause']); + rmSync(path.join(dir, '.lousho', 'improve.stop')); + const resumed = await tick(); + expect(resumed.cycle?.outcome).toBe('applied'); + expect(actions().slice(0, 2)).toEqual(['pause', 'resume']); + }); + + it('LOUSHO_IMPROVE=off and enabled false pause it too', async () => { + vi.stubEnv('LOUSHO_IMPROVE', 'off'); + expect((await tick()).state).toBe('paused'); + vi.unstubAllEnvs(); + expect((await tick({ settings: settings({ enabled: false }) })).state).toBe('paused'); + expect(actions()).toEqual(['pause']); + }); + + it('the second watcher on the same dir does nothing', async () => { + seedFailures(); + const first = await tick(); + expect(first.cycle?.outcome).toBe('applied'); + const second = await tick(); + expect(second).toMatchObject({ state: 'leased' }); + expect(healOnce).toHaveBeenCalledTimes(1); + }); + + it('maxAppliesPerDay 1 blocks a second apply the same day', async () => { + const context = watchContext(options()); + seedFailures(); + expect((await watchTick(options(), context)).cycle?.outcome).toBe('applied'); + seedFailures(); + expect((await watchTick(options(), context)).cycle).toMatchObject({ outcome: 'budget', reason: expect.stringMatching(/automatic apply/) }); + }); + + it('cooldown after rollback blocks cycles', async () => { + seedFailures(); + await updateState(dir, { cooldownUntil: new Date(Date.now() + 3_600_000).toISOString() }); + const result = await tick(); + expect(result.reason).toMatch(/cooldown/); + expect(healOnce).not.toHaveBeenCalled(); + }); + + it('the token budget passed to healOnce shrinks as spent', async () => { + const context = watchContext(options()); + const manual = settings({ approval: 'manual', trigger: { minNewFailures: 3, requireRecurring: false } }); + seedFailures(); + await watchTick(options({ settings: manual }), context); + seedFailures(); + await watchTick(options({ settings: manual }), context); + const [first, second] = vi.mocked(healOnce).mock.calls.map(([call]) => call.maxTokens as number); + const spent = readCycles(dir).find((cycle) => cycle.outcome !== 'interrupted')?.tokens ?? 0; + expect(first).toBe(2_000_000); + expect(spent).toBeGreaterThan(0); + expect(second).toBeLessThan(first); + }); + + it('same signatures twice skip the second cycle', async () => { + const context = watchContext(options()); + const manual = settings({ approval: 'manual', trigger: { minNewFailures: 3, requireRecurring: false } }); + seedFailures(); + expect((await watchTick(options({ settings: manual }), context)).cycle?.outcome).toBe('pending'); + expect((await watchTick(options({ settings: manual }), context)).state).toBe('deduped'); + expect(healOnce).toHaveBeenCalledTimes(1); + }); + + it('three outages in a row pause it with an audit entry', async () => { + const context = watchContext(options()); + const down: HealOptions['runCase'] = async () => { + throw new Error('provider down'); + }; + for (let i = 0; i < 3; i++) { + seedFailures(); + expect((await watchTick(options({ runCase: down }), context)).cycle?.outcome).toBe('outage'); + } + expect(existsSync(path.join(dir, '.lousho', 'improve.stop'))).toBe(true); + expect(actions()).toEqual(['pause']); + seedFailures(); + expect((await watchTick(options({ runCase: down }), context)).state).toBe('paused'); + expect(actions()).toEqual(['pause']); + }); + + it('runs the corpus guard once guard.everyHours have passed', async () => { + const context = watchContext(options()); + seedFailures(); + const first = await watchTick(options({ protect: ['p1'] }), context); + expect(first.cycle?.outcome).toBe('applied'); + expect(first.guard).toBeUndefined(); + const second = await watchTick(options(), context); + expect(second.guard).toMatchObject({ passed: true }); + expect((await watchTick(options(), context)).guard).toBeUndefined(); + }); + + it('adds the cases a kept edit fixed to the corpus once the monitor keeps it, once', async () => { + const context = watchContext(options()); + seedFailures(); + const applied = await watchTick(options(), context); + expect(applied.cycle?.outcome).toBe('applied'); + const applySeq = readAudit(dir).find((entry) => entry.action === 'apply')?.seq as number; + await updateState(dir, { monitor: { revision: currentRevision(dir), applySeq, at: new Date().toISOString(), verdict: { state: 'kept' } } }); + vi.mocked(checkApplied).mockResolvedValueOnce({ state: 'kept' }).mockResolvedValueOnce({ state: 'kept' }); + await watchTick(options(), context); + const fixed = readCorpus(dir).filter((item) => item.origin === 'fixed'); + expect(fixed.length).toBeGreaterThan(0); + expect(fixed.every((item) => item.revision === currentRevision(dir))).toBe(true); + await watchTick(options(), context); + expect(readCorpus(dir).filter((item) => item.origin === 'fixed')).toHaveLength(fixed.length); + }); +}); + +describe('improveSchedule', () => { + it('is a schedule whose run is one watch tick', async () => { + seedFailures(); + const schedule = improveSchedule({ ...options(), cron: '0 * * * *' }); + expect(isDefinedSchedule(schedule)).toBe(true); + expect(schedule).toMatchObject({ cron: '0 * * * *', name: 'improve' }); + await schedule.run?.({ agent: undefined as never, firedAt: new Date(), name: 'improve' }); + expect(healOnce).toHaveBeenCalledTimes(1); + }); +}); + +describe('watchAgent', () => { + it('ticks on its interval and stop() waits for the cycle in progress, then exits', async () => { + seedFailures(); + let release: () => void = () => undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const slow: HealOptions['runCase'] = async (harness, input) => { + await gate; + return runCase(harness, input); + }; + const cycles: string[] = []; + const watcher = watchAgent(options({ runCase: slow, intervalMs: 60_000, onCycle: (cycle) => cycles.push(cycle.outcome) })); + await vi.waitFor(() => expect(healOnce).toHaveBeenCalledTimes(1)); + const stopping = watcher.stop(); + release(); + await stopping; + expect(healOnce).toHaveBeenCalledTimes(1); + expect(readFileSync(path.join(dir, 'instructions.md'), 'utf8').trim()).toBe(BASE); + expect(readCycles(dir)[0]?.outcome).toBe('interrupted'); + }); +}); diff --git a/src/improve/watch.ts b/src/improve/watch.ts new file mode 100644 index 00000000..5ae61010 --- /dev/null +++ b/src/improve/watch.ts @@ -0,0 +1,198 @@ +import { randomUUID } from 'node:crypto'; +import { lstatSync, mkdirSync, writeFileSync } from 'node:fs'; +import path from 'node:path'; +import type { CheckpointStore } from '../execution/checkpoint'; +import { failureReportFromDb } from '../observe/failures'; +import { runsByRevision } from '../observe/store'; +import type { LLMProvider } from '../providers'; +import { createRateLimiter, withRateLimit } from '../providers/rateLimit'; +import { resolveProvider } from '../providers/resolveProvider'; +import { defineSchedule, type DefinedSchedule } from '../schedules/defineSchedule'; +import { fileStore } from '../storage/fileStore'; +import { appendAudit } from './audit'; +import { pausedBy, type HealCycle } from './cycles'; +import { healOnce, type HealOptions } from './heal'; +import { budgetLeft, guardScheduled, harvest, saveWatch, watchOf, type WatchContext, type WatchState } from './watchSteps'; +import { checkApplied, type MonitorVerdict } from './monitor'; +import { harnessHash, readAgentHarness } from './revisions'; +import { loadImproveSettings } from './settings'; +import { readState, updateState } from './state'; +import type { GuardCorpusResult } from './corpusGuard'; +import type { ImproveOptions } from './types'; + +/** Options for {@link watchAgent} and {@link improveSchedule}. */ +export interface WatchOptions extends Omit { + /** The agent's id or registered name in the observe database; the monitor judges its runs. */ + agent: string; + /** Milliseconds between ticks (default 15 minutes). */ + intervalMs?: number; + /** Called with each cycle that ran. */ + onCycle?: (cycle: HealCycle) => void; + /** Called with an error a tick threw; the watcher keeps running (default: printed to stderr). */ + onError?: (error: unknown) => void; + /** Holds the one-watcher-per-directory lease (default a file store in `.lousho/watch`). */ + store?: CheckpointStore; +} + +/** What one watch tick did. */ +export interface WatchTick { + /** `paused` by the kill switch, `leased` when another watcher holds the directory, `deduped` when nothing new arrived, else `ran`. */ + state: 'paused' | 'leased' | 'deduped' | 'ran'; + /** Why, as plain text. */ + reason: string; + /** The monitor's verdict on the last apply. */ + monitor?: MonitorVerdict; + /** The scheduled corpus guard, when it ran. */ + guard?: GuardCorpusResult; + /** The cycle, when one ran. */ + cycle?: HealCycle; +} + +const DEFAULT_INTERVAL_MS = 15 * 60 * 1000; +const MAX_OUTAGES = 3; +const LEASE_ID = 'improve-watch'; +const MAX_RUNS = 1000; +const STOP_FILE = path.join('.lousho', 'improve.stop'); + +function rateLimited(editor: ImproveOptions['editor'], rpm: number): ImproveOptions['editor'] { + const provider: LLMProvider = 'provider' in editor ? editor.provider : resolveProvider(editor.model); + return { provider: withRateLimit(provider, createRateLimiter({ requestsPerMinute: rpm })) }; +} + +/** The tick's shared parts: the lease owner and the rate-limited editor. */ +export function watchContext(options: WatchOptions, signal?: AbortSignal): WatchContext { + const store = options.store ?? fileStore(path.join(options.dir, '.lousho', 'watch')).checkpoints; + return { owner: `${process.pid}:watch:${randomUUID()}`, store, editor: rateLimited(options.editor, options.settings.rateLimit.rpm), ...(signal && { signal }) }; +} + +function pausedReason(options: WatchOptions): string | undefined { + try { + return pausedBy(options.dir, { enabled: options.settings.enabled && loadImproveSettings(options.dir).enabled }); + } catch (error) { + return `improve.json can't be read (${error instanceof Error ? error.message : String(error)})`; + } +} + +async function pauseOnce(dir: string, reason: string): Promise { + if (watchOf(dir).paused === undefined) { + await appendAudit(dir, { actor: 'scheduler', action: 'pause', reason: `The self-healing loop is paused: ${reason}.` }); + await saveWatch(dir, { paused: reason }); + } + return { state: 'paused', reason: `Paused: ${reason}.` }; +} + +async function resumeIfPaused(dir: string): Promise { + if (watchOf(dir).paused === undefined) return; + await appendAudit(dir, { actor: 'scheduler', action: 'resume', reason: 'The kill switch is off, so the self-healing loop runs again.' }); + await updateState(dir, (current) => { + const { paused: _paused, ...rest } = current.watch ?? {}; + return { watch: rest }; + }); +} + +function failureSignatures(options: WatchOptions): { revision: string; signatures: string[]; failed: number } { + const revision = harnessHash(readAgentHarness(options.dir)); + const runs = runsByRevision(options.db, { revision, agent: options.agent, limit: MAX_RUNS }); + const traceIds = runs.map((run) => run.traceId); + const signatures = traceIds.length === 0 ? [] : [...new Set(failureReportFromDb(options.db, { traceIds }).groups.map((group) => group.signature))].sort(); + return { revision, signatures, failed: runs.filter((run) => run.failed).length }; +} + +function stateCooldown(dir: string, now: number): boolean { + const until = Date.parse(readState(dir).cooldownUntil ?? ''); + return Number.isFinite(until) && until > now; +} + +async function countOutage(dir: string, cycle: HealCycle): Promise { + if (cycle.outcome !== 'outage') { + if (['idle', 'paused', 'busy', 'budget', 'cooldown'].includes(cycle.outcome)) return; + await saveWatch(dir, { outages: 0 }); + return; + } + const outages = (watchOf(dir).outages ?? 0) + 1; + await saveWatch(dir, { outages }); + if (outages < MAX_OUTAGES) return; + const reason = `${MAX_OUTAGES} cycles in a row ended in a provider outage`; + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + if (!lstatSync(path.join(dir, STOP_FILE), { throwIfNoEntry: false })) writeFileSync(path.join(dir, STOP_FILE), `${reason}\n`, { flag: 'wx' }); + await appendAudit(dir, { actor: 'scheduler', action: 'pause', reason: `The self-healing loop paused itself: ${reason}.` }); + await saveWatch(dir, { paused: `${STOP_FILE} exists`, outages: 0 }); +} + +async function runCycle(options: WatchOptions, context: WatchContext, trigger: HealCycle['trigger'], now: number): Promise { + if (stateCooldown(options.dir, now)) return { state: 'ran', reason: `The monitor's cooldown runs until ${readState(options.dir).cooldownUntil}, so no cycle ran.` }; + const { revision, signatures, failed } = failureSignatures(options); + const watch = watchOf(options.dir); + const same = watch.signatures !== undefined && watch.signatures.join('\n') === signatures.join('\n'); + if (same && watch.revision === revision && watch.failed === failed) return { state: 'deduped', reason: 'The same failures as the last cycle and nothing new, so no cycle ran.' }; + const { intervalMs: _interval, onCycle, onError: _onError, store: _store, ...heal } = options; + const cycle = await healOnce({ ...heal, editor: context.editor, trigger, maxTokens: budgetLeft(options.dir, options.settings, now), ...(context.signal && { signal: context.signal }) }); + await saveWatch(options.dir, { revision, signatures, failed }); + await countOutage(options.dir, cycle); + onCycle?.(cycle); + return { state: 'ran', reason: cycle.reason, cycle }; +} + +/** Runs one watch tick: the kill switch, the lease, the monitor, the scheduled corpus guard, then a cycle within the budgets. */ +export async function watchTick(options: WatchOptions, context: WatchContext, trigger: HealCycle['trigger'] = 'failures'): Promise { + const now = Date.now(); + const paused = pausedReason(options); + if (paused) return pauseOnce(options.dir, paused); + await resumeIfPaused(options.dir); + const ttl = 2 * (options.intervalMs ?? DEFAULT_INTERVAL_MS); + if (context.store.acquireLease && !(await context.store.acquireLease(LEASE_ID, context.owner, ttl))) { + return { state: 'leased', reason: 'Another watcher holds this agent directory.' }; + } + const monitor = await checkApplied(options.db, options.dir, options.settings, { agent: options.agent, now }); + await harvest(options, monitor, now); + const guard = await guardScheduled(options, now); + const ran = await runCycle(options, context, trigger, now); + return { ...ran, monitor, ...(guard && { guard }) }; +} + +/** Runs healOnce, the monitor and the corpus guard on a timer, within the settings' budgets; stops on the kill switch. */ +export function watchAgent(options: WatchOptions): { stop(): Promise } { + const controller = new AbortController(); + const context = watchContext(options, controller.signal); + const report = options.onError ?? ((error: unknown) => process.stderr.write(`lousho improve --watch: a tick failed: ${error instanceof Error ? error.message : String(error)}\n`)); + let timer: ReturnType | undefined; + let stopped = false; + let running: Promise = Promise.resolve(); + const loop = (): void => { + running = watchTick(options, context) + .then(() => undefined, (error: unknown) => (controller.signal.aborted ? undefined : report(error))) + .finally(() => { + if (!stopped) timer = setTimeout(loop, options.intervalMs ?? DEFAULT_INTERVAL_MS); + }); + }; + loop(); + return { + async stop() { + stopped = true; + if (timer) clearTimeout(timer); + controller.abort(); + await running; + await context.store.releaseLease?.(LEASE_ID, context.owner); + }, + }; +} + +/** A defineSchedule() schedule that runs one watch tick, for agents that already use schedules/. */ +export function improveSchedule( + options: WatchOptions & { + /** When to tick (default every 15 minutes). */ + cron?: string; + /** The schedule's name (default `improve`). */ + name?: string; + } +): DefinedSchedule { + const { cron = '*/15 * * * *', name = 'improve', ...watch } = options; + const context = watchContext(watch); + return defineSchedule({ + cron, + name, + run: async () => { + await watchTick(watch, context); + }, + }); +} diff --git a/src/improve/watchSteps.ts b/src/improve/watchSteps.ts new file mode 100644 index 00000000..8d61c036 --- /dev/null +++ b/src/improve/watchSteps.ts @@ -0,0 +1,137 @@ +import type { CheckpointStore } from '../execution/checkpoint'; +import { exampleFromTrace } from '../observe/datasets'; +import { readAudit } from './audit'; +import { admitProtectedCases, type CapturedCase } from './cases'; +import { addToCorpus, readCorpus, type CorpusCase } from './corpus'; +import { guardCorpus, type GuardCorpusResult } from './corpusGuard'; +import { dayStart, readCycles, usageToday, type HealCycle } from './cycles'; +import { corpusCases } from './healApply'; +import { caseProbe, libraryOf, unscored } from './healState'; +import type { MonitorVerdict } from './monitor'; +import { voteOnOutcome } from './reflect'; +import { harnessHash, readAgentHarness } from './revisions'; +import type { ImproveSettings } from './settings'; +import { readState, updateState } from './state'; +import type { ImproveOptions } from './types'; +import type { WatchOptions } from './watch'; + +export interface WatchContext { + owner: string; + store: CheckpointStore; + editor: ImproveOptions['editor']; + signal?: AbortSignal; +} + +export interface WatchMemory { + paused?: string; + signatures?: string[]; + revision?: string; + failed?: number; + outages?: number; + guardAt?: number; + spentDay?: number; + spent?: number; + harvested?: number; +} + +export interface WatchState { + cooldownUntil?: string; + monitor?: { applySeq?: number }; + watch?: WatchMemory; +} + +const HOUR_MS = 60 * 60 * 1000; +const ADMISSION_RUNS = 6; +const GUARD_FIRST_REPEAT = 200; +const GUARD_RANGE = 20; + +export const watchOf = (dir: string): WatchMemory => readState(dir).watch ?? {}; + +export async function saveWatch(dir: string, change: Partial): Promise { + await updateState(dir, (current) => ({ watch: { ...current.watch, ...change } })); +} + +export function budgetLeft(dir: string, settings: ImproveSettings, now: number): number { + const watch = watchOf(dir); + const spent = watch.spentDay === dayStart(now) ? (watch.spent ?? 0) : 0; + return settings.budget.maxTokensPerDay - spent - usageToday(dir, now).tokens; +} + +async function addSpent(dir: string, tokens: number, now: number): Promise { + if (tokens <= 0) return; + const day = dayStart(now); + await updateState(dir, (current) => { + const watch = current.watch ?? {}; + return { watch: { ...watch, spentDay: day, spent: (watch.spentDay === day ? (watch.spent ?? 0) : 0) + tokens } }; + }); +} + +function counting(probe: ImproveOptions['probe'], spent: { tokens: number }): ImproveOptions['probe'] { + return async (harness, caseId, repeat) => { + const run = await probe(harness, caseId, repeat); + spent.tokens += run.tokens; + return run; + }; +} + +function corpusProbe(options: WatchOptions, cases: readonly CapturedCase[], spent: { tokens: number }): ImproveOptions['probe'] { + return counting(unscored(caseProbe(cases, options.runCase, options.check), new Set(), undefined), spent); +} + +function cycleOfApply(dir: string, applySeq: number): HealCycle | undefined { + const id = readAudit(dir).find((entry) => entry.seq === applySeq)?.cycle; + return id === undefined ? undefined : readCycles(dir).find((cycle) => cycle.id === id); +} + +function fixedCases(options: WatchOptions, cycle: HealCycle): CapturedCase[] { + const kept = new Set(readCorpus(options.dir).map((item) => item.traceId)); + return (cycle.result?.cases ?? []).flatMap((verdict) => { + const traceId = verdict.caseId.slice(verdict.caseId.indexOf(':') + 1); + if (verdict.role !== 'training' || kept.has(traceId)) return []; + const example = exampleFromTrace(options.db, traceId); + return typeof example === 'string' ? [] : [{ id: verdict.caseId, input: example.input, traceId, kind: 'protected' as const }]; + }); +} + +async function admitFixed(options: WatchOptions, cycle: HealCycle, now: number): Promise { + const cases = fixedCases(options, cycle); + if (cases.length === 0 || budgetLeft(options.dir, options.settings, now) <= 0) return; + const spent = { tokens: 0 }; + const harness = readAgentHarness(options.dir); + const { admitted } = await admitProtectedCases({ harness, cases, probe: corpusProbe(options, cases, spent), runs: ADMISSION_RUNS }); + await addSpent(options.dir, spent.tokens, now); + const revision = harnessHash(harness); + const addedAt = new Date(now).toISOString(); + const fixed = admitted.map((item): CorpusCase => ({ id: item.id, traceId: item.traceId, input: item.input, addedAt, revision, origin: 'fixed', baseline: { passed: ADMISSION_RUNS, scored: ADMISSION_RUNS } })); + if (fixed.length > 0) await addToCorpus(options.dir, fixed, options.settings.guard.maxCases); +} + +/** Votes once on the library entries behind an apply the monitor decided, and adds the cases a kept edit fixed to the corpus once they pass 6 of 6. */ +export async function harvest(options: WatchOptions, monitor: MonitorVerdict, now: number): Promise { + if (monitor.state !== 'kept' && monitor.state !== 'rolled-back') return; + const state = readState(options.dir); + const applySeq = state.monitor?.applySeq; + if (applySeq === undefined || state.watch?.harvested === applySeq) return; + await saveWatch(options.dir, { harvested: applySeq }); + const cycle = cycleOfApply(options.dir, applySeq); + if (!cycle?.result) return; + const library = libraryOf(options.dir, options.settings); + if (library) voteOnOutcome(library.dir, { outcome: monitor.state, result: cycle.result, domain: library.domain }); + if (monitor.state === 'kept') await admitFixed(options, cycle, now); +} + +/** Replays the corpus under the current revision once `guard.everyHours` have passed, rolling back on a regression. */ +export async function guardScheduled(options: WatchOptions, now: number): Promise { + const { dir, settings } = options; + const last = watchOf(dir).guardAt; + if (last !== undefined && now - last < settings.guard.everyHours * HOUR_MS) return undefined; + const corpus = readCorpus(dir); + if (corpus.length === 0 || budgetLeft(dir, settings, now) <= 0) return undefined; + const revision = harnessHash(readAgentHarness(dir)); + const earlier = readAudit(dir).filter((entry) => (entry.action === 'guard-pass' || entry.action === 'guard-fail') && entry.to === revision).length; + const spent = { tokens: 0 }; + const guard = await guardCorpus(dir, { probe: corpusProbe(options, corpusCases(corpus), spent), runs: settings.guard.runs, rollback: true, firstRepeat: GUARD_FIRST_REPEAT + earlier * GUARD_RANGE }); + await addSpent(dir, spent.tokens, now); + await saveWatch(dir, { guardAt: now }); + return guard; +} From cb640ec7ef3d9a08d0792b6681f95358d7bb79e3 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Mon, 9 Nov 2026 02:08:02 +0300 Subject: [PATCH 2/2] fix(improve): watch review fixes Retry after an outage or a blocked cycle, renew the lease during a tick, a per-directory lease id, meter the watcher's own spend into the day's usage, check stop and pause between steps, re-read settings each tick, and keep the Telegram owner when telegram.json is empty. Co-Authored-By: Claude Opus 5.5 --- api/improve.api.md | 942 ++++++++++++++++++++++++++++++- llms-full.txt | 18 + src/cli/improve.ts | 12 +- src/cli/init/telegramTemplate.ts | 4 +- src/improve/cycles.ts | 7 +- src/improve/healApply.ts | 2 - src/improve/watch.test.ts | 54 +- src/improve/watch.ts | 144 +++-- src/improve/watchSteps.ts | 85 ++- 9 files changed, 1175 insertions(+), 93 deletions(-) diff --git a/api/improve.api.md b/api/improve.api.md index 4c912917..fa6d9818 100644 --- a/api/improve.api.md +++ b/api/improve.api.md @@ -33,6 +33,31 @@ export function admitProtectedCases(options: { // @public const AGENT_EVENT_SCHEMA_VERSION: 1; +// @public +interface AgentApprovals { + answer(reply: { + id: string; + answer: string; + }, options?: ResolveApprovalOptions): Promise; + get(id: string): Promise; + list(): Promise; + resolve(decision: ApprovalDecision, options?: ResolveApprovalOptions): Promise; + resolveAll(decisions: readonly ApprovalDecision[], options?: ResolveApprovalOptions): Promise; + streamAnswer(reply: { + id: string; + answer: string; + }, options?: ResolveApprovalOptions): AgentRun; + streamResolve(decision: ApprovalDecision, options?: ResolveApprovalOptions): AgentRun; +} + +// @public +type AgentCompaction = boolean | AgentCompactionOptions; + +// @public +interface AgentCompactionOptions extends Pick { + summarizer?: LLMProvider | string; +} + // @public interface AgentDrift { instructions: boolean; @@ -111,6 +136,16 @@ interface AgentEventUsage { totalTokens: number; } +// @public +interface AgentFingerprint { + hash: string; + instructions: string; + model?: string; + tools: Record; + // (undocumented) + version: number; +} + // @public interface AgentHook { name: string; @@ -120,6 +155,54 @@ interface AgentHook { preToolCall?(ctx: ToolCallHookContext): MaybePromise; } +// @public +type AgentInput = string | ContentPart[] | Message[]; + +// @public +interface AgentOAuth { + complete(params: OAuthCallbackParams): Promise; + mcpSignInUrl(server: string): Promise; + signInUrl(provider: OAuthProvider): Promise; +} + +// @public +interface AgentRun extends AsyncIterable { + enqueue(input: AgentInput): EnqueueResult; + readonly result: Promise>; + readonly runId: string; + steer(input: AgentInput): SteerResult; +} + +// @public +class AgentSession { + constructor(run: SessionRunner, options?: SessionOptions, streamRun?: SessionStreamRunner, spawn?: SessionSpawner); + protected readonly checkpointStore: CheckpointStore | undefined; + clear(): Promise; + compact(options?: SessionCompactOptions): Promise; + protected continueTurn(next: () => Promise): Promise; + protected readonly currentPermissionMode: () => PermissionMode; + discardPending(): Promise; + fork(options: SessionForkOptions): Promise>; + protected readonly forwardToListeners: (event: AgentEvent) => void; + history(): Promise; + // (undocumented) + readonly id: string; + load(): Promise; + get messages(): readonly Message[]; + on(listener: (event: AgentEvent) => void): () => void; + protected pausedTurnFound(_approvalId: string): void; + pending(): Promise; + protected pendingTurnDiscarded(_approvalIds: string[]): Promise; + get permissionMode(): PermissionMode; + resume(options?: { + signal?: AbortSignal; + }): Promise; + send(input: AgentInput, options?: SessionSendOptions): Promise>; + setPermissionMode(mode: PermissionMode): void; + stream(input: AgentInput, options?: SessionSendOptions): AgentRun; + protected get turnEvents(): ((event: AgentEvent) => void) | undefined; +} + // @public export function appendAudit(dir: string, entry: AuditInput): Promise; @@ -150,6 +233,18 @@ export type Approval = { approvedBy: string; }; +// @public +interface ApprovalDecision { + // (undocumented) + approved: boolean; + args?: Record; + // (undocumented) + id: string; + // (undocumented) + note?: string; + remember?: 'session'; +} + // @public type ApprovalKind = 'tool' | 'question' | 'sign-in'; @@ -218,6 +313,25 @@ export interface AuditEntry { // @public export type AuditInput = Omit; +// @public +type BackgroundTaskStatus = 'queued' | 'running' | 'done' | 'failed' | 'cancelled' | 'awaiting-approval'; + +// @public +interface BackgroundTaskView { + // (undocumented) + agent: string; + approvalId?: string; + elapsedMs: number; + error?: string; + result?: string; + // (undocumented) + status: BackgroundTaskStatus | 'timeout'; + // (undocumented) + taskId: string; + // (undocumented) + toolName?: string; +} + // @public interface BudgetExceeded { // (undocumented) @@ -235,6 +349,22 @@ interface BudgetExceededEvent extends AgentEventBase<'budget.exceeded'>, BudgetE // @public type BudgetLimit = Exclude; +// @public +interface BudgetSpent { + // (undocumented) + costUsd?: number; + // (undocumented) + durationMs: number; + // (undocumented) + inputTokens: number; + // (undocumented) + outputTokens: number; + // (undocumented) + steps: number; + // (undocumented) + totalTokens: number; +} + // @public export function cachedProbe(probe: ImproveOptions['probe'], options: CachedProbeOptions): ImproveOptions['probe']; @@ -325,6 +455,98 @@ export function checkApplied(db: SqlDatabase, dir: string, settings: Pick; +// @public +interface Checkpoint { + agentFingerprint?: AgentFingerprint; + // (undocumented) + agentId: string; + approvalId?: string; + approvalIds?: string[]; + approvalKind?: ApprovalKind; + attempts?: number; + businessState?: unknown; + delegatedInFlight?: CheckpointUsage[]; + // (undocumented) + finishReason?: string; + flow?: FlowCheckpointState; + lastError?: CheckpointError; + // (undocumented) + messages: Message[]; + metadata?: Record; + principal?: Principal; + runConfig?: unknown; + // (undocumented) + sessionId: string; + startedToolCallIds?: string[]; + status?: CheckpointStatus; + // (undocumented) + stepIndex: number; + stepUsage?: StepUsage[]; + // (undocumented) + toolCalls: unknown[]; + unstarted?: boolean; + usage: CheckpointUsage; +} + +// @public +interface CheckpointDeleteOptions { + keepHistory?: boolean; +} + +// @public +interface CheckpointError { + at: string; + category?: string; + code?: string; + // (undocumented) + message: string; + retryable: boolean; +} + +// @public +interface CheckpointHistoryEntry { + checkpoint: Checkpoint; + savedAt: string; + status: CheckpointStatus; + step: number; +} + +// @public +interface CheckpointHistoryOptions { + limit?: number; +} + +// @public +interface CheckpointListEntry { + checkpoint: Checkpoint; + sessionId: string; + status: CheckpointStatus; +} + +// @public +interface CheckpointListOptions { + status?: CheckpointStatus | readonly CheckpointStatus[]; +} + +// @public +type CheckpointStatus = 'in-progress' | 'awaiting-approval' | 'finished'; + +// @public +interface CheckpointStore { + acquireLease?(sessionId: string, owner: string, ttlMs: number): Promise; + delete(sessionId: string, options?: CheckpointDeleteOptions): Promise; + history?(sessionId: string, options?: CheckpointHistoryOptions): Promise; + list?(options?: CheckpointListOptions): Promise; + load(sessionId: string): Promise; + releaseLease?(sessionId: string, owner: string): Promise; + renewLease?(sessionId: string, owner: string, ttlMs: number): Promise; + save(sessionId: string, checkpoint: Checkpoint): Promise; + saveIfLeased?(sessionId: string, checkpoint: Checkpoint, owner: string): Promise; +} + +// @public +type CheckpointUsage = Pick & Partial; + // @public export function citedEntries(notes: string): string[]; @@ -349,6 +571,48 @@ interface CompactionDoneEvent extends AgentEventBase<'compaction.done'> { unchanged?: boolean; } +// @public +interface CompactionHookOptions extends Omit { + onCompaction?: (info: CompactionInfo) => void; +} + +// @public +interface CompactionInfo { + appliedStrategy?: string; + error?: Error; + // (undocumented) + prunedToolCallIds: string[]; + strategy: string; + summary?: string; + // (undocumented) + tokensAfter: number; + // (undocumented) + tokensBefore: number; +} + +// @public +interface CompactionInput { + contextWindow: number; + estimateTokens: CompactionTokenCounter; + generate?: (provider: LLMProvider, request: GenerateOptions) => Promise; + messages: Message[]; + protectedTokens: number; + signal?: AbortSignal; + thresholdTokens: number; +} + +// @public +interface CompactionResult { + error?: Error; + messages: Message[]; + prunedToolCallIds: string[]; + summary?: string; + // (undocumented) + tokensAfter: number; + // (undocumented) + tokensBefore: number; +} + // @public interface CompactionStartEvent extends AgentEventBase<'compaction.start'> { contextWindow: number; @@ -358,6 +622,27 @@ interface CompactionStartEvent extends AgentEventBase<'compaction.start'> { trigger?: 'manual'; } +// @public +interface CompactionStrategy { + // (undocumented) + compact(input: CompactionInput): CompactionResult | Promise; + // (undocumented) + name: string; +} + +// @public +type CompactionTokenCounter = (input: Message | Message[]) => number; + +// @public +interface CompactMessagesOptions { + contextWindow?: number; + model?: string; + protectedTokens?: number; + reserveOutputTokens?: number; + strategy?: CompactionStrategy; + thresholdPercent?: number; +} + // @public type ContentPart = TextContentPart | ImageContentPart | FileContentPart; @@ -385,15 +670,76 @@ export function createShadowHook(options: { recorded?: Readonly>; }): AgentHook; +// @public +type CredentialOwner = 'app' | 'user'; + // @public export function currentRevision(dir: string): string; +// @public +type DefinedSchedule = Readonly; + +// @public +interface DelegatedUsage { + costPartial?: boolean; + costUsd: number | undefined; + // (undocumented) + estimated: boolean; + // (undocumented) + inputTokens: number; + // (undocumented) + modelCalls: number; + // (undocumented) + outputTokens: number; + runs: number; + toolErrors?: number; + // (undocumented) + totalTokens: number; +} + // @public export const emptyLedger: () => ImproveLedger; +// @public +interface EnqueueResult { + applied: false | Promise; + id: string; +} + // @public type ExecutionFinishReason = GenerateResult['finishReason'] | 'awaiting-approval' | 'aborted' | 'max-steps' | 'output-invalid' | 'budget-exceeded' | 'guardrail'; +// @public +interface ExecutionResult { + abortReason?: { + name: string; + message: string; + }; + agentName?: string; + approvalId?: string; + approvalIds?: string[]; + backgroundTasks?: BackgroundTaskView[]; + budget?: BudgetExceeded; + // (undocumented) + finishReason: ExecutionFinishReason; + guardrail?: GuardrailTrip; + // (undocumented) + messages: Message[]; + object?: TObject; + outputError?: OutputError; + outputRepaired?: true; + reasoning?: string; + // (undocumented) + steps: number; + stepUsage?: StepUsage[]; + // (undocumented) + text: string; + // (undocumented) + toolCalls: ToolCall[]; + toolErrors?: number; + usage: RunUsage; +} + // @public interface FileContentPart { // (undocumented) @@ -406,6 +752,57 @@ interface FileContentPart { type: 'file'; } +// @public +interface FlowCheckpointState { + choices: Record; + code: string; + completedNodeIds: string[]; + nodeResults: Record; + output?: unknown; + pendingApproval?: FlowPendingApproval; + steps: number; + structure?: string; + usage: ProviderUsage; + variables: Record; +} + +// @public +interface FlowPendingApproval { + approvalId: string; + args: Record; + nodeId: string; + toolName: string; +} + +// @public +interface ForkOptions { + checkpointStore: CheckpointStore; + fromStep: number; + newSessionId?: string; + // (undocumented) + patch?: ForkPatch; + sessionId: string; +} + +// @public +interface ForkPatch { + appendInput?: string; + businessState?: unknown; + messages?: (messages: Message[]) => Message[]; + toolResult?: { + toolCallId: string; + result: unknown; + }; +} + +// @public +interface ForkResult { + checkpoint: Checkpoint; + // (undocumented) + sessionId: string; + step: number; +} + // @public export function formatImproveResult(result: ImproveResult): string; @@ -788,6 +1185,12 @@ export interface ImproveResult { stages: ImproveStage[]; } +// @public +export function improveSchedule(options: WatchOptions & { + cron?: string; + name?: string; +}): DefinedSchedule; + // @public export interface ImproveSettings { approval: 'auto' | 'manual'; @@ -822,6 +1225,9 @@ export interface ImproveSettings { autoRollback: boolean; giveUpDays: number; }; + rateLimit: { + rpm: number; + }; trigger: { minNewFailures: number; requireRecurring: boolean; @@ -845,6 +1251,30 @@ interface InputAppliedEvent extends AgentEventBase<'input.applied'> { step: number; } +// @public +class InputQueue { + // @internal + callOutput(): void; + // @internal + close(): void; + // @internal + closeAfter(run: Promise): void; + // @internal + endPhase(): void; + // @internal + listen(listener: (input: QueuedInput) => void): void; + // @internal + get messages(): Message[]; + push(input: AgentInput): EnqueueResult; + // @internal + startBatch(): AbortSignal; + // @internal + startCall(): AbortSignal; + steer(input: AgentInput): SteerResult; + // @internal + take(): QueuedInput[]; +} + // @public interface InputQueuedEvent extends AgentEventBase<'input.queued'> { id: string; @@ -935,6 +1365,34 @@ interface Message { // @public type MessageRole = 'system' | 'user' | 'assistant' | 'tool'; +// @public +interface ModelSettings { + frequencyPenalty?: number; + maxTokens?: number; + presencePenalty?: number; + seed?: number; + stop?: string[]; + temperature?: number; + toolChoice?: 'auto' | 'required' | 'none' | { + type: 'function'; + function: { + name: string; + }; + }; + topP?: number; +} + +// @public +interface ModelUsage { + calls: number; + costPartial?: boolean; + costUsd?: number; + // (undocumented) + inputTokens: number; + // (undocumented) + outputTokens: number; +} + // @public type ModerationCategory = 'hate' | 'harassment' | 'self-harm' | 'sexual' | 'sexual-minors' | 'violence' | 'illicit'; @@ -950,8 +1408,100 @@ export interface MonitorVerdict { } // @public -interface ObjectDeltaEvent extends AgentEventBase<'object.delta'> { - object: unknown; +interface OAuthCallbackParams { + // (undocumented) + code?: string; + error?: string; + principal?: Principal; + // (undocumented) + state: string; +} + +// @public +type OAuthClientAuth = 'client_secret_post' | 'client_secret_basic'; + +// @public +interface OAuthCompleteResult { + approvalId?: string; + displayName?: string; + // (undocumented) + outcome: 'signed-in' | 'declined'; + provider: string; + // (undocumented) + sessionId?: string; +} + +// @public +interface OAuthProvider extends Readonly> { + // (undocumented) + readonly clientAuth: OAuthClientAuth; + // (undocumented) + readonly credentialOwner: CredentialOwner; + // (undocumented) + readonly displayName: string; + // (undocumented) + readonly kind: 'oauth-provider'; + // (undocumented) + readonly scopes: readonly string[]; +} + +// @public +interface OAuthProviderOptions { + authorizationParams?: Record; + authorizationUrl: string; + clientAuth?: OAuthClientAuth; + // (undocumented) + clientId: string; + clientSecret?: string; + credentialOwner?: CredentialOwner; + displayName?: string; + fetch?: typeof fetch; + name: string; + redirectUri: string; + // (undocumented) + scopes?: readonly string[]; + tokenUrl: string; +} + +// @public +interface ObjectDeltaEvent extends AgentEventBase<'object.delta'> { + object: unknown; +} + +// @public +interface OutputError { + issues: ToolArgumentIssue[]; + kind: 'invalid' | 'truncated'; + message: string; +} + +// @public +export function pauseImprove(dir: string, options?: { + by?: string; + reason?: string; +}): Promise; + +// @public +interface PendingApproval { + // (undocumented) + agentId?: string; + // (undocumented) + args: Record; + // (undocumented) + createdAt: string; + expiresAt?: string; + // (undocumented) + id: string; + kind?: ApprovalKind; + principal?: Principal; + question?: ApprovalQuestion; + sessionId?: string; + signIn?: ApprovalSignIn; + subagentPath?: string[]; + // (undocumented) + toolCallId: string; + // (undocumented) + toolName: string; } // @public @@ -967,9 +1517,50 @@ export interface PendingProposal { staleAt?: string; } +// @public +interface PendingRun { + approvalId?: string; + approvalIds?: string[]; + approvalKind?: ApprovalKind; + attempts?: number; + checkpointId: string; + kind: 'session' | 'run'; + lastError?: CheckpointError; + sessionId: string; + status: 'in-progress' | 'awaiting-approval'; + step: number; +} + +// @public +interface PendingTurn { + // (undocumented) + approvalId?: string; + approvalIds?: string[]; + approvalKind?: ApprovalKind; + idempotencyKey?: string; + status: 'in-progress' | 'awaiting-approval'; +} + // @public type PermissionAction = 'allow' | 'deny' | 'ask'; +// @public +interface PermissionAuditContext { + // (undocumented) + principal?: Readonly; +} + +// @public +interface PermissionContext { + principal?: Readonly; + // (undocumented) + sessionId?: string; + // (undocumented) + toolCallId: string; + // (undocumented) + toolName: string; +} + // @public type PermissionDecision = PermissionAction | 'default'; @@ -999,6 +1590,39 @@ interface PermissionDecisionEvent extends AgentEventBase<'permission.decision'>, // @public type PermissionMode = 'default' | 'plan' | 'acceptEdits' | 'dontAsk'; +// @public +interface PermissionModeChange { + at: string; + // (undocumented) + from: PermissionMode; + // (undocumented) + sessionId: string; + // (undocumented) + to: PermissionMode; +} + +// @public +interface PermissionOptions { + onPermissionDecision?: (entry: PermissionDecisionEntry, context: PermissionAuditContext) => void; + onPermissionModeChange?: (change: PermissionModeChange) => void; + permissionMode?: PermissionMode | (() => PermissionMode); + permissions?: readonly PermissionRule[]; +} + +// @public +interface PermissionRule { + // (undocumented) + action: PermissionAction; + reason?: string; + // (undocumented) + tool: PermissionToolMatcher; + ttlMs?: number; + when?: (args: Record, ctx: PermissionContext) => boolean | Promise; +} + +// @public +type PermissionToolMatcher = string | readonly string[] | RegExp; + // @public type PiiType = 'email' | 'phone' | 'credit-card' | 'iban' | 'us-ssn' | 'ip-address'; @@ -1067,6 +1691,15 @@ export interface ProbeSummary { // @public type PromptCachingOption = 'auto' | false; +// @public +interface PromptScheduleInput extends ScheduleBase { + // (undocumented) + prompt: string; + // (undocumented) + run?: never; + sharedSession?: boolean; +} + // @public export function protectedCasesFromPasses(db: SqlDatabase, traceIds: readonly string[]): CapturedCase[]; @@ -1111,6 +1744,16 @@ interface ProviderUsage { // @public export function pruneCache(dir: string, limits?: CacheLimits): void; +// @public +interface QueuedInput { + // (undocumented) + id: string; + // (undocumented) + messages: Message[]; + steered?: 'immediate' | 'queued'; + text: string; +} + // @public export function readAgentHarness(dir: string): Harness; @@ -1160,6 +1803,18 @@ interface ReasoningSettings { // @public type ReasoningStartEvent = AgentEventBase<'reasoning.start'>; +// @public +interface RecoverOptions { + intervalMs?: number; + maxAttempts?: number; + onError?: (error: unknown, run?: PendingRun) => void; + onOutage?: (paused: boolean) => void; + onQuarantined?: (run: PendingRun) => void; + onRecovered?: (run: PendingRun, result: TResult) => void; + retryWithDiagnosis?: boolean; + signal?: AbortSignal; +} + // @public export interface RecoveryEvidence { answer: string; @@ -1195,6 +1850,18 @@ export function renderTranscript(messages: readonly Message[], options?: { maxResultChars?: number; }): string; +// @public +interface ResolveApprovalOptions { + principal?: Principal; + // (undocumented) + signal?: AbortSignal; +} + +// @public +export function resumeImprove(dir: string, options?: { + by?: string; +}): Promise; + // @public export interface RollbackBy { actor: AuditEntry['actor']; @@ -1234,6 +1901,22 @@ interface RunLimits { onExceeded?: 'stop' | 'throw'; } +// @public +interface RunRecovery { + readonly done: Promise; + stop(): Promise; +} + +// @public +interface RunScheduleInput extends ScheduleBase { + // (undocumented) + prompt?: never; + // (undocumented) + run: (ctx: ScheduleContext) => Promise; + // (undocumented) + sharedSession?: never; +} + // @public interface RunStartEvent extends AgentEventBase<'run.start'> { agentId?: string; @@ -1241,6 +1924,30 @@ interface RunStartEvent extends AgentEventBase<'run.start'> { agentName: string; } +// @public +interface RunUsage { + byModel: Record; + cachedInputTokens?: number; + cacheWriteTokens?: number; + // @deprecated (undocumented) + completionTokens: number; + costPartial?: boolean; + costUsd: number | undefined; + delegated?: DelegatedUsage; + estimated: boolean; + hostedToolCalls?: Partial>; + // (undocumented) + inputTokens: number; + modelCalls: number; + // (undocumented) + outputTokens: number; + // @deprecated (undocumented) + promptTokens: number; + reasoningTokens?: number; + // (undocumented) + totalTokens: number; +} + // @public export function savePending(dir: string, input: { result: ImproveResult; @@ -1250,6 +1957,23 @@ export function savePending(dir: string, input: { maxAlwaysOnChars?: number; }): Promise; +// @public (undocumented) +interface ScheduleBase { + cron: string; + name?: string; + timezone?: string; +} + +// @public +interface ScheduleContext { + agent: SimpleAgent; + firedAt: Date; + name: string; +} + +// @public +type ScheduleInput = PromptScheduleInput | RunScheduleInput; + // @public interface SchemaIssue { // (undocumented) @@ -1260,6 +1984,23 @@ interface SchemaIssue { }>; } +// @public +interface SendOptions { + approvalTtlMs?: number; + instructions?: string; + maxSteps?: number; + metadata?: Record; + modelSettings?: ModelSettings; + onEvent?: (event: AgentEvent) => void; + parentSpanId?: string; + permissionMode?: PermissionMode; + principal?: Principal; + reasoning?: ReasoningOption; + sessionId?: string; + signal?: AbortSignal; + throwOnError?: boolean; +} + // @public interface ServedBy { model?: string; @@ -1267,6 +2008,145 @@ interface ServedBy { provider: string; } +// @public +interface SessionBudget { + // (undocumented) + limits: RunLimits; + // (undocumented) + spent: BudgetSpent; +} + +// @public +interface SessionCompactOptions extends Pick { + // (undocumented) + signal?: AbortSignal; +} + +// @public +interface SessionCompactResult { + error?: { + message: string; + }; + // (undocumented) + messagesAfter: number; + // (undocumented) + messagesBefore: number; + strategy: string; + // (undocumented) + tokensAfter: number; + // (undocumented) + tokensBefore: number; +} + +// @public +interface SessionForkOptions { + fromStep: number; + id?: string; + patch?: Pick; +} + +// @public +interface SessionHistoryStep { + messageIndex: number; + step: number; + text: string; + // (undocumented) + toolCalls: Array<{ + id: string; + name: string; + args: unknown; + result?: string; + }>; + turn: number; +} + +// @public +interface SessionOptions { + checkpointStore?: CheckpointStore; + compaction?: AgentCompaction; + id?: string; + limits?: RunLimits; + onPermissionModeChange?: PermissionOptions['onPermissionModeChange']; + permissionMode?: PermissionOptions['permissionMode']; + store?: SessionStore | SessionStores; + turnHistoryLimit?: number; + turnPolicy?: 'queue' | 'steer' | 'wait'; +} + +// @public +type SessionRunner = (input: Message[], signal?: AbortSignal, checkpoint?: SessionTurnOptions, call?: SessionTurnCall) => Promise; + +// @public +interface SessionSendOptions { + metadata?: Record; + principal?: Principal; + // (undocumented) + signal?: AbortSignal; +} + +// @public +type SessionSpawner = (options: SessionOptions) => AgentSession; + +// @public +interface SessionStore { + delete(id: string): Promise; + load(id: string): Promise; + save(id: string, messages: readonly Message[]): Promise; + saveIf?(id: string, expectedRevision: string, messages: readonly Message[]): Promise; +} + +// @public +interface SessionStores { + // (undocumented) + checkpoints?: CheckpointStore; + // (undocumented) + sessions: SessionStore; +} + +// @public +type SessionStreamRunner = (input: Message[], signal?: AbortSignal, checkpoint?: SessionTurnOptions, call?: SessionTurnCall) => AgentRun; + +// @public +interface SessionTurnCall { + // (undocumented) + input: AgentInput; + // (undocumented) + metadata?: Record; + principal?: Principal; +} + +// @public +interface SessionTurnCheckpoint { + // (undocumented) + checkpointStore: CheckpointStore; + sessionId: string; +} + +// @public +type SessionTurnOptions = Partial & { + sessionBudget?: SessionBudget; + inputQueue?: InputQueue; + permissionMode?: PermissionOptions['permissionMode']; + onAgentEvent?: (event: AgentEvent) => void; +}; + +// @public +interface SimpleAgent { + approvals: AgentApprovals; + close: () => Promise; + fork: (sessionId: string, options: Omit) => Promise; + oauth: AgentOAuth; + pending: () => Promise; + ready: () => Promise; + recover: (options?: RecoverOptions>) => RunRecovery; + resume: (sessionId: string, options?: { + signal?: AbortSignal; + }) => Promise | null>; + send: (message: AgentInput, options?: SendOptions) => Promise>; + session: (options?: SessionOptions) => AgentSession; + stream: (message: AgentInput, options?: SendOptions) => AgentRun; +} + // @public export function snapshotRevision(dir: string, source: HarnessRevision['source'], parent?: string): HarnessRevision; @@ -1313,6 +2193,14 @@ interface StandardSchemaV1 { }; } +// @public +interface SteerResult { + // (undocumented) + applied: 'immediate' | 'queued' | false; + id: string; + joined: Promise; +} + // @public interface StepDoneEvent extends AgentEventBase<'step.done'> { finishReason: ExecutionFinishReason | 'steered'; @@ -1327,6 +2215,16 @@ interface StepStartEvent extends AgentEventBase<'step.start'> { step: number; } +// @public +interface StepUsage { + costUsd?: number; + estimated: boolean; + model: string; + step: number; + // (undocumented) + usage: Usage; +} + // @public interface StreamChunk { // (undocumented) @@ -1427,6 +2325,12 @@ interface TodoUpdatedEvent extends AgentEventBase<'todo.updated'> { toolCallId: string; } +// @public +interface ToolArgumentIssue { + message: string; + path: string; +} + // @public interface ToolCall { // (undocumented) @@ -1534,6 +2438,16 @@ interface ToolStartEvent extends AgentEventBase<'tool.start'> { toolName: string; } +// @public +interface Usage { + cachedInputTokens?: number; + cacheWriteTokens?: number; + inputTokens: number; + outputTokens: number; + reasoningTokens?: number; + totalTokens: number; +} + // @public export function verifyAudit(dir: string): number | undefined; @@ -1549,6 +2463,30 @@ export function voteOnOutcome(dir: string, options: { entryId?: string; }): LibraryEntry[]; +// @public +export function watchAgent(options: WatchOptions): { + stop(): Promise; +}; + +// @public +export interface WatchOptions extends Omit { + agent: string; + intervalMs?: number; + onCycle?: (cycle: HealCycle) => void; + onError?: (error: unknown) => void; + reloadSettings?: () => ImproveSettings; + store?: CheckpointStore; +} + +// @public +export interface WatchTick { + cycle?: HealCycle; + guard?: GuardCorpusResult; + monitor?: MonitorVerdict; + reason: string; + state: 'paused' | 'stopped' | 'leased' | 'cooldown' | 'deduped' | 'ran'; +} + // @public export interface WindowStat { lower: number; diff --git a/llms-full.txt b/llms-full.txt index e74e27d9..1160e3a3 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -5995,6 +5995,23 @@ npx lousho improve reject ./my-agent 3f9c0a1b2d4e --by ali --reason "too broad" and moves the file to `.lousho/proposals/rejected/`. A rejected proposal is never queued again. +`--watch` keeps the loop running on an agent directory: + +```bash +npx lousho improve ./my-agent --check ./check.mjs --watch --interval 15m +npx lousho improve pause ./my-agent --reason "release freeze" --by ali +npx lousho improve resume ./my-agent --by ali +``` + +Each tick checks the kill switch (`.lousho/improve.stop`, `LOUSHO_IMPROVE=off` +or `"enabled": false`, audited once as `pause`), takes a lease so only one +watcher runs per directory, runs the post-apply monitor, replays the corpus +every `guard.everyHours`, and then runs one cycle within the daily caps. A tick +with the same failures as the last one and no new failed run is skipped. +Three outages in a row write the stop file. Editor calls are limited to +`rateLimit.rpm` a minute. Ctrl-C stops after the current step. `pause` writes +the stop file and `resume` removes it; both are audited. + ### `improve.json` Self-healing settings live in `improve.json` beside the agent. Every key is @@ -6010,6 +6027,7 @@ optional, and an unknown key or a bad value is an error "monitor": { "minRuns": 20, "window": 20, "maxDrop": 0.05, "autoRollback": true, "giveUpDays": 7 }, "guard": { "runs": 3, "everyHours": 24, "maxCases": 30 }, "cooldownHours": 24, + "rateLimit": { "rpm": 20 }, "library": { "enabled": true } } ``` diff --git a/src/cli/improve.ts b/src/cli/improve.ts index 54d6623c..6c122d1d 100644 --- a/src/cli/improve.ts +++ b/src/cli/improve.ts @@ -525,6 +525,7 @@ async function watchDir(args: ImproveArgs, io: ImproveIo, dir: string, config: C runCase: runCaseWith(config, args, io, new Map()), editor, settings, + reloadSettings: () => loadImproveSettings(dir, args.approval ? { approval: args.approval } : {}), agent, protect: args.protect, withRecoveries: args.withRecoveries === true, @@ -543,13 +544,22 @@ async function watchDir(args: ImproveArgs, io: ImproveIo, dir: string, config: C } } +function assertWatchArgs(args: ImproveArgs, target: DevTarget, forced: string | undefined): void { + if (!args.watch) { + if (args.intervalMs !== undefined) throw usageError(SPEC, '--interval only applies with --watch.'); + return; + } + if (args.maxTokens !== undefined || args.out !== undefined) throw usageError(SPEC, '--watch takes its token budget from improve.json and writes no --out file; drop --max-tokens and --out.'); + if (args.dryRun || forced || target.kind !== 'dir') throw usageError(SPEC, `--watch needs an agent directory whose instructions live in instructions.md, and can't be a dry run${forced ? `: ${forced}` : ''}.`); +} + async function improve(args: ImproveArgs, io: ImproveIo): Promise { const target = detectTarget(args.path, 'improve'); const config = await loadConfig(target); if (args.out !== undefined) assertOutUsable(args.out, target, config); const harness = baseHarness(config); const forced = args.dryRun ? undefined : dryRunReason(target, harness); - if (args.watch && (args.dryRun || forced || target.kind !== 'dir')) throw usageError(SPEC, `--watch needs an agent directory whose instructions live in instructions.md, and can't be a dry run${forced ? `: ${forced}` : ''}.`); + assertWatchArgs(args, target, forced); if (forced) io.stderr.write(`lousho improve: this is a dry run, because ${forced}. Nothing will be applied.\n`); if (args.dryRun || forced || target.kind !== 'dir') return proposeOnly(args, io, target, config, harness); return args.watch ? watchDir(args, io, target.path, config) : healDir(args, io, target.path, config); diff --git a/src/cli/init/telegramTemplate.ts b/src/cli/init/telegramTemplate.ts index 9b22b8b0..1262623d 100644 --- a/src/cli/init/telegramTemplate.ts +++ b/src/cli/init/telegramTemplate.ts @@ -256,10 +256,8 @@ const BRIDGE_TS = [ ' }', '};', 'const loadState = (): BridgeState => {', - ' const state = readJson(STATE_FILE);', - ' if (state) return state;', ' const { ownerChatId, ownerUserId } = readJson(LEGACY_STATE_FILE) ?? {};', - ' return { ownerChatId, ownerUserId };', + ' return { ownerChatId, ownerUserId, ...readJson(STATE_FILE) };', '};', 'const saveState = (patch: BridgeState) => {', ' ensureDir(dirname(STATE_FILE));', diff --git a/src/improve/cycles.ts b/src/improve/cycles.ts index 13fdccb6..eff36d2e 100644 --- a/src/improve/cycles.ts +++ b/src/improve/cycles.ts @@ -3,6 +3,7 @@ import { lstatSync, mkdirSync, readdirSync, readFileSync, renameSync, rmSync, wr import path from 'node:path'; import { ConfigurationError } from '../execution/errors'; import { readAudit } from './audit'; +import { readState } from './state'; import type { CanaryReplayResult } from './canary'; import type { GuardCorpusResult } from './corpusGuard'; import type { ImproveSettings } from './settings'; @@ -124,12 +125,14 @@ export function readCycles(dir: string): HealCycle[] { export const dayStart = (now: number): number => Math.floor(now / DAY_MS) * DAY_MS; -/** Cycles, tokens and automatic applies so far in the current UTC day. */ +/** Cycles, tokens (the watcher's own guard and corpus runs included) and automatic applies so far in the current UTC day. */ export function usageToday(dir: string, now: number): { cycles: number; tokens: number; applies: number } { const since = dayStart(now); const today = readCycles(dir).filter((cycle) => Date.parse(cycle.startedAt) >= since); const applies = readAudit(dir).filter((entry) => entry.action === 'apply' && AUTO_ACTORS.includes(entry.actor) && Date.parse(entry.at) >= since).length; - return { cycles: today.length, tokens: today.reduce((sum, cycle) => sum + cycle.tokens, 0), applies }; + const watch = readState<{ watch?: { spentDay?: number; spent?: number } }>(dir).watch; + const watcher = watch?.spentDay === since ? (watch.spent ?? 0) : 0; + return { cycles: today.length, tokens: watcher + today.reduce((sum, cycle) => sum + cycle.tokens, 0), applies }; } /** When the cooldown after the latest rollback ends, in epoch ms, or `undefined` when none is active. */ diff --git a/src/improve/healApply.ts b/src/improve/healApply.ts index 43f2253c..fdd720a4 100644 --- a/src/improve/healApply.ts +++ b/src/improve/healApply.ts @@ -65,7 +65,6 @@ function guardShortfall(state: HealState): string | undefined { return `The corpus guard that must follow an apply needs about ${needed} tokens, more than the ${Math.max(0, left)} left, so nothing was applied.`; } -/** Applies a proposal that passed the gate and the canary, then guards it with the corpus and undoes it on a regression or a guard that can't judge. */ export async function applyAndGuard(state: HealState, result: ImproveResult): Promise { const shortfall = guardShortfall(state); if (shortfall) return finish(state, 'budget-exceeded', shortfall); @@ -82,7 +81,6 @@ export async function applyAndGuard(state: HealState, result: ImproveResult): Pr return guardApplied(state, to); } -/** Guards an apply whose cycle stopped before its guard ran, when the directory still holds that revision. */ export async function guardUnfinished(options: HealOptions, harness: Harness, cap: number): Promise { const last = readCycles(options.dir)[0]; if (last?.outcome !== 'interrupted' || last.to === undefined || last.guard !== undefined || harnessHash(harness) !== last.to) return undefined; diff --git a/src/improve/watch.test.ts b/src/improve/watch.test.ts index 2098c2a1..e838d71e 100644 --- a/src/improve/watch.test.ts +++ b/src/improve/watch.test.ts @@ -6,9 +6,10 @@ import type { IngestEvent } from '../observe/envelope'; import { ingestBatch, openDb } from '../observe/store'; import type { SqlDatabase } from '../storage/sqlite/driver'; import { mockModel, type MockRequest } from '../testing'; -import { readAudit } from './audit'; +import { appendAudit, readAudit } from './audit'; import { readCorpus } from './corpus'; -import { readCycles } from './cycles'; +import { readCycles, usageToday } from './cycles'; +import { budgetLeft } from './watchSteps'; import { healOnce, type HealOptions } from './heal'; import { checkApplied } from './monitor'; import { currentRevision } from './revisions'; @@ -183,17 +184,52 @@ describe('watchTick', () => { const down: HealOptions['runCase'] = async () => { throw new Error('provider down'); }; - for (let i = 0; i < 3; i++) { - seedFailures(); - expect((await watchTick(options({ runCase: down }), context)).cycle?.outcome).toBe('outage'); - } + seedFailures(); + for (let i = 0; i < 3; i++) expect((await watchTick(options({ runCase: down, settings: settings({ trigger: { minNewFailures: 3, requireRecurring: true } }) }), context, 'manual')).cycle?.outcome).toBe('outage'); expect(existsSync(path.join(dir, '.lousho', 'improve.stop'))).toBe(true); expect(actions()).toEqual(['pause']); - seedFailures(); expect((await watchTick(options({ runCase: down }), context)).state).toBe('paused'); expect(actions()).toEqual(['pause']); }); + it('maxCyclesPerDay, the daily token cap and an audited rollback each stop the cycle', async () => { + const context = watchContext(options()); + seedFailures(); + expect((await watchTick(options({ settings: settings({ budget: { maxTokensPerDay: 10, maxCyclesPerDay: 4, maxAppliesPerDay: 1 } }) }), context)).cycle).toMatchObject({ outcome: 'budget-exceeded' }); + expect((await watchTick(options({ settings: settings({ budget: { maxTokensPerDay: 2_000_000, maxCyclesPerDay: 1, maxAppliesPerDay: 1 } }) }), context, 'manual')).cycle).toMatchObject({ outcome: 'budget', reason: expect.stringMatching(/cycles have run/) }); + await appendAudit(dir, { actor: 'monitor', action: 'rollback', reason: 'Regressed.' }); + expect((await watchTick(options(), context, 'manual')).cycle).toMatchObject({ outcome: 'cooldown' }); + }); + + it("counts the watcher's own guard runs toward the day's tokens", async () => { + const context = watchContext(options()); + seedFailures(); + await watchTick(options({ protect: ['p1'] }), context); + const before = usageToday(dir, Date.now()).tokens; + const guarded = await watchTick(options(), context); + expect(guarded.guard?.tokens).toBeGreaterThan(0); + expect(usageToday(dir, Date.now()).tokens).toBe(before + (guarded.guard?.tokens ?? 0)); + expect(budgetLeft(dir, settings(), Date.now())).toBe(2_000_000 - usageToday(dir, Date.now()).tokens); + }); + + it('renews its lease while a long tick runs, so a second watcher stays out', async () => { + seedFailures(); + let release: () => void = () => undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const slow: HealOptions['runCase'] = async (harness, input) => { + await gate; + return runCase(harness, input); + }; + const short = { intervalMs: 30 }; + const first = watchTick(options({ ...short, runCase: slow }), watchContext(options(short))); + await new Promise((resolve) => setTimeout(resolve, 150)); + expect((await watchTick(options(short), watchContext(options(short)))).state).toBe('leased'); + release(); + expect((await first).cycle?.outcome).toBe('applied'); + }); + it('runs the corpus guard once guard.everyHours have passed', async () => { const context = watchContext(options()); seedFailures(); @@ -219,6 +255,10 @@ describe('watchTick', () => { expect(fixed.every((item) => item.revision === currentRevision(dir))).toBe(true); await watchTick(options(), context); expect(readCorpus(dir).filter((item) => item.origin === 'fixed')).toHaveLength(fixed.length); + vi.mocked(checkApplied).mockResolvedValueOnce({ state: 'rolled-back' }); + await updateState(dir, { monitor: undefined }); + await watchTick(options(), context); + expect(readCorpus(dir).filter((item) => item.origin === 'fixed')).toHaveLength(fixed.length); }); }); diff --git a/src/improve/watch.ts b/src/improve/watch.ts index 5ae61010..10574401 100644 --- a/src/improve/watch.ts +++ b/src/improve/watch.ts @@ -1,7 +1,7 @@ -import { randomUUID } from 'node:crypto'; -import { lstatSync, mkdirSync, writeFileSync } from 'node:fs'; +import { createHash, randomUUID } from 'node:crypto'; import path from 'node:path'; import type { CheckpointStore } from '../execution/checkpoint'; +import { ConfigurationError } from '../execution/errors'; import { failureReportFromDb } from '../observe/failures'; import { runsByRevision } from '../observe/store'; import type { LLMProvider } from '../providers'; @@ -10,34 +10,37 @@ import { resolveProvider } from '../providers/resolveProvider'; import { defineSchedule, type DefinedSchedule } from '../schedules/defineSchedule'; import { fileStore } from '../storage/fileStore'; import { appendAudit } from './audit'; +import type { GuardCorpusResult } from './corpusGuard'; import { pausedBy, type HealCycle } from './cycles'; import { healOnce, type HealOptions } from './heal'; -import { budgetLeft, guardScheduled, harvest, saveWatch, watchOf, type WatchContext, type WatchState } from './watchSteps'; import { checkApplied, type MonitorVerdict } from './monitor'; +import { pauseImprove } from './pause'; import { harnessHash, readAgentHarness } from './revisions'; -import { loadImproveSettings } from './settings'; +import { loadImproveSettings, type ImproveSettings } from './settings'; import { readState, updateState } from './state'; -import type { GuardCorpusResult } from './corpusGuard'; import type { ImproveOptions } from './types'; +import { budgetLeft, guardScheduled, harvest, saveWatch, watchOf, type WatchContext, type WatchState } from './watchSteps'; /** Options for {@link watchAgent} and {@link improveSchedule}. */ export interface WatchOptions extends Omit { /** The agent's id or registered name in the observe database; the monitor judges its runs. */ agent: string; - /** Milliseconds between ticks (default 15 minutes). */ + /** Milliseconds between ticks (default 15 minutes); the lease lasts twice this and is renewed while a tick runs. */ intervalMs?: number; + /** Reads the settings afresh each tick, such as `() => loadImproveSettings(dir)`; without it `settings` is used, with `enabled` re-read from improve.json. */ + reloadSettings?: () => ImproveSettings; /** Called with each cycle that ran. */ onCycle?: (cycle: HealCycle) => void; /** Called with an error a tick threw; the watcher keeps running (default: printed to stderr). */ onError?: (error: unknown) => void; - /** Holds the one-watcher-per-directory lease (default a file store in `.lousho/watch`). */ + /** Holds the one-watcher-per-directory lease; it must support leases (default a file store in `.lousho/watch`). */ store?: CheckpointStore; } /** What one watch tick did. */ export interface WatchTick { - /** `paused` by the kill switch, `leased` when another watcher holds the directory, `deduped` when nothing new arrived, else `ran`. */ - state: 'paused' | 'leased' | 'deduped' | 'ran'; + /** `paused` by the kill switch, `stopped` when the watcher was stopped, `leased` when another watcher holds the directory, `cooldown` after a rollback, `deduped` when nothing new arrived, else `ran`. */ + state: 'paused' | 'stopped' | 'leased' | 'cooldown' | 'deduped' | 'ran'; /** Why, as plain text. */ reason: string; /** The monitor's verdict on the last apply. */ @@ -50,26 +53,32 @@ export interface WatchTick { const DEFAULT_INTERVAL_MS = 15 * 60 * 1000; const MAX_OUTAGES = 3; -const LEASE_ID = 'improve-watch'; const MAX_RUNS = 1000; -const STOP_FILE = path.join('.lousho', 'improve.stop'); +const RENEW_FRACTION = 3; +const MIN_RENEW_MS = 10; +const NOT_WORKED: readonly HealCycle['outcome'][] = ['outage', 'busy', 'budget', 'budget-exceeded', 'cooldown', 'paused', 'interrupted']; +const UNCOUNTED: readonly HealCycle['outcome'][] = ['idle', 'paused', 'busy', 'budget', 'cooldown']; + +const messageOf = (error: unknown): string => (error instanceof Error ? error.message : String(error)); function rateLimited(editor: ImproveOptions['editor'], rpm: number): ImproveOptions['editor'] { const provider: LLMProvider = 'provider' in editor ? editor.provider : resolveProvider(editor.model); return { provider: withRateLimit(provider, createRateLimiter({ requestsPerMinute: rpm })) }; } -/** The tick's shared parts: the lease owner and the rate-limited editor. */ export function watchContext(options: WatchOptions, signal?: AbortSignal): WatchContext { const store = options.store ?? fileStore(path.join(options.dir, '.lousho', 'watch')).checkpoints; - return { owner: `${process.pid}:watch:${randomUUID()}`, store, editor: rateLimited(options.editor, options.settings.rateLimit.rpm), ...(signal && { signal }) }; + if (!store.acquireLease) throw new ConfigurationError('watchAgent: the store must support leases (acquireLease), so only one watcher runs per agent directory.', 'store'); + const leaseId = `improve-${createHash('sha256').update(path.resolve(options.dir)).digest('hex').slice(0, 32)}`; + return { owner: `${process.pid}:watch:${randomUUID()}`, leaseId, store, editor: rateLimited(options.editor, options.settings.rateLimit.rpm), ...(signal && { signal }) }; } -function pausedReason(options: WatchOptions): string | undefined { +function settingsOf(options: WatchOptions): ImproveSettings | string { try { - return pausedBy(options.dir, { enabled: options.settings.enabled && loadImproveSettings(options.dir).enabled }); + if (options.reloadSettings) return options.reloadSettings(); + return { ...options.settings, enabled: options.settings.enabled && loadImproveSettings(options.dir).enabled }; } catch (error) { - return `improve.json can't be read (${error instanceof Error ? error.message : String(error)})`; + return `improve.json can't be read (${messageOf(error)})`; } } @@ -90,77 +99,118 @@ async function resumeIfPaused(dir: string): Promise { }); } -function failureSignatures(options: WatchOptions): { revision: string; signatures: string[]; failed: number } { +function halted(options: WatchOptions, signal: AbortSignal | undefined): WatchTick | undefined { + if (signal?.aborted) return { state: 'stopped', reason: 'The watcher stopped, or lost its lease, before the next step.' }; + const settings = settingsOf(options); + const paused = typeof settings === 'string' ? settings : pausedBy(options.dir, settings); + return paused ? { state: 'paused', reason: `Paused before the next step: ${paused}.` } : undefined; +} + +function failureKey(options: WatchOptions): { revision: string; signatures: string[]; failed: number; newest?: string } { const revision = harnessHash(readAgentHarness(options.dir)); const runs = runsByRevision(options.db, { revision, agent: options.agent, limit: MAX_RUNS }); const traceIds = runs.map((run) => run.traceId); const signatures = traceIds.length === 0 ? [] : [...new Set(failureReportFromDb(options.db, { traceIds }).groups.map((group) => group.signature))].sort(); - return { revision, signatures, failed: runs.filter((run) => run.failed).length }; + const failed = runs.filter((run) => run.failed); + return { revision, signatures, failed: failed.length, ...(failed[0] && { newest: failed[0].traceId }) }; } -function stateCooldown(dir: string, now: number): boolean { - const until = Date.parse(readState(dir).cooldownUntil ?? ''); - return Number.isFinite(until) && until > now; +function cooldownUntil(dir: string, now: number): string | undefined { + const until = readState(dir).cooldownUntil; + const at = Date.parse(until ?? ''); + return Number.isFinite(at) && at > now ? until : undefined; } async function countOutage(dir: string, cycle: HealCycle): Promise { + if (UNCOUNTED.includes(cycle.outcome)) return; if (cycle.outcome !== 'outage') { - if (['idle', 'paused', 'busy', 'budget', 'cooldown'].includes(cycle.outcome)) return; await saveWatch(dir, { outages: 0 }); return; } const outages = (watchOf(dir).outages ?? 0) + 1; await saveWatch(dir, { outages }); if (outages < MAX_OUTAGES) return; - const reason = `${MAX_OUTAGES} cycles in a row ended in a provider outage`; - mkdirSync(path.join(dir, '.lousho'), { recursive: true }); - if (!lstatSync(path.join(dir, STOP_FILE), { throwIfNoEntry: false })) writeFileSync(path.join(dir, STOP_FILE), `${reason}\n`, { flag: 'wx' }); - await appendAudit(dir, { actor: 'scheduler', action: 'pause', reason: `The self-healing loop paused itself: ${reason}.` }); - await saveWatch(dir, { paused: `${STOP_FILE} exists`, outages: 0 }); + await pauseImprove(dir, { reason: `The self-healing loop paused itself: ${MAX_OUTAGES} cycles in a row ended in a provider outage.` }); + await saveWatch(dir, { outages: 0 }); } async function runCycle(options: WatchOptions, context: WatchContext, trigger: HealCycle['trigger'], now: number): Promise { - if (stateCooldown(options.dir, now)) return { state: 'ran', reason: `The monitor's cooldown runs until ${readState(options.dir).cooldownUntil}, so no cycle ran.` }; - const { revision, signatures, failed } = failureSignatures(options); - const watch = watchOf(options.dir); - const same = watch.signatures !== undefined && watch.signatures.join('\n') === signatures.join('\n'); - if (same && watch.revision === revision && watch.failed === failed) return { state: 'deduped', reason: 'The same failures as the last cycle and nothing new, so no cycle ran.' }; - const { intervalMs: _interval, onCycle, onError: _onError, store: _store, ...heal } = options; + const until = cooldownUntil(options.dir, now); + if (until) return { state: 'cooldown', reason: `The monitor's cooldown runs until ${until}, so no cycle ran.` }; + const key = failureKey(options); + const last = watchOf(options.dir); + const same = last.revision === key.revision && last.failed === key.failed && last.newest === key.newest && (last.signatures ?? []).join('\n') === key.signatures.join('\n'); + if (same && last.signatures !== undefined) return { state: 'deduped', reason: 'The same failures as the last cycle and nothing new, so no cycle ran.' }; + const { intervalMs: _interval, reloadSettings: _reload, onCycle, onError: _onError, store: _store, ...heal } = options; const cycle = await healOnce({ ...heal, editor: context.editor, trigger, maxTokens: budgetLeft(options.dir, options.settings, now), ...(context.signal && { signal: context.signal }) }); - await saveWatch(options.dir, { revision, signatures, failed }); + if (!NOT_WORKED.includes(cycle.outcome)) await saveWatch(options.dir, { revision: key.revision, signatures: key.signatures, failed: key.failed, ...(key.newest && { newest: key.newest }) }); await countOutage(options.dir, cycle); onCycle?.(cycle); return { state: 'ran', reason: cycle.reason, cycle }; } +function holdLease(context: WatchContext, ttl: number): { signal: AbortSignal; release(): void } { + const lost = new AbortController(); + const renew = () => { + context.store.acquireLease?.(context.leaseId, context.owner, ttl).then( + (held) => (held ? undefined : lost.abort()), + () => lost.abort() + ); + }; + const timer = setInterval(renew, Math.max(MIN_RENEW_MS, Math.floor(ttl / RENEW_FRACTION))); + (timer as { unref?: () => void }).unref?.(); + return { signal: context.signal ? AbortSignal.any([context.signal, lost.signal]) : lost.signal, release: () => clearInterval(timer) }; +} + +async function steps(options: WatchOptions, context: WatchContext, trigger: HealCycle['trigger'], now: number): Promise { + const monitor = await checkApplied(options.db, options.dir, options.settings, { agent: options.agent, now }); + const stopAfterMonitor = halted(options, context.signal); + if (stopAfterMonitor) return { ...stopAfterMonitor, monitor }; + await harvest(options, monitor, now, context.signal); + const stopAfterHarvest = halted(options, context.signal); + if (stopAfterHarvest) return { ...stopAfterHarvest, monitor }; + const guard = await guardScheduled(options, now, context.signal); + const stopAfterGuard = halted(options, context.signal); + if (stopAfterGuard) return { ...stopAfterGuard, monitor, ...(guard && { guard }) }; + const ran = await runCycle(options, context, trigger, now); + return { ...ran, monitor, ...(guard && { guard }) }; +} + /** Runs one watch tick: the kill switch, the lease, the monitor, the scheduled corpus guard, then a cycle within the budgets. */ export async function watchTick(options: WatchOptions, context: WatchContext, trigger: HealCycle['trigger'] = 'failures'): Promise { const now = Date.now(); - const paused = pausedReason(options); + const settings = settingsOf(options); + const paused = typeof settings === 'string' ? settings : pausedBy(options.dir, settings); if (paused) return pauseOnce(options.dir, paused); await resumeIfPaused(options.dir); + if (context.signal?.aborted) return { state: 'stopped', reason: 'The watcher stopped.' }; const ttl = 2 * (options.intervalMs ?? DEFAULT_INTERVAL_MS); - if (context.store.acquireLease && !(await context.store.acquireLease(LEASE_ID, context.owner, ttl))) { - return { state: 'leased', reason: 'Another watcher holds this agent directory.' }; + if (!(await context.store.acquireLease?.(context.leaseId, context.owner, ttl))) return { state: 'leased', reason: 'Another watcher holds this agent directory.' }; + const lease = holdLease(context, ttl); + try { + return await steps({ ...options, settings: settings as ImproveSettings }, { ...context, signal: lease.signal }, trigger, now); + } finally { + lease.release(); } - const monitor = await checkApplied(options.db, options.dir, options.settings, { agent: options.agent, now }); - await harvest(options, monitor, now); - const guard = await guardScheduled(options, now); - const ran = await runCycle(options, context, trigger, now); - return { ...ran, monitor, ...(guard && { guard }) }; } /** Runs healOnce, the monitor and the corpus guard on a timer, within the settings' budgets; stops on the kill switch. */ -export function watchAgent(options: WatchOptions): { stop(): Promise } { +export function watchAgent(options: WatchOptions): { + /** Stops after the current step, waits for the tick in progress and releases the lease. */ + stop(): Promise; +} { const controller = new AbortController(); const context = watchContext(options, controller.signal); - const report = options.onError ?? ((error: unknown) => process.stderr.write(`lousho improve --watch: a tick failed: ${error instanceof Error ? error.message : String(error)}\n`)); + const report = options.onError ?? ((error: unknown) => process.stderr.write(`lousho improve --watch: a tick failed: ${messageOf(error)}\n`)); let timer: ReturnType | undefined; let stopped = false; let running: Promise = Promise.resolve(); const loop = (): void => { running = watchTick(options, context) - .then(() => undefined, (error: unknown) => (controller.signal.aborted ? undefined : report(error))) + .then( + () => undefined, + (error: unknown) => (controller.signal.aborted ? undefined : report(error)) + ) .finally(() => { if (!stopped) timer = setTimeout(loop, options.intervalMs ?? DEFAULT_INTERVAL_MS); }); @@ -172,7 +222,7 @@ export function watchAgent(options: WatchOptions): { stop(): Promise } { if (timer) clearTimeout(timer); controller.abort(); await running; - await context.store.releaseLease?.(LEASE_ID, context.owner); + await context.store.releaseLease?.(context.leaseId, context.owner); }, }; } @@ -180,7 +230,7 @@ export function watchAgent(options: WatchOptions): { stop(): Promise } { /** A defineSchedule() schedule that runs one watch tick, for agents that already use schedules/. */ export function improveSchedule( options: WatchOptions & { - /** When to tick (default every 15 minutes). */ + /** When to tick (default every 15 minutes); set `intervalMs` to the same spacing so the lease spans two ticks. */ cron?: string; /** The schedule's name (default `improve`). */ name?: string; diff --git a/src/improve/watchSteps.ts b/src/improve/watchSteps.ts index 8d61c036..c68001b6 100644 --- a/src/improve/watchSteps.ts +++ b/src/improve/watchSteps.ts @@ -6,7 +6,7 @@ import { addToCorpus, readCorpus, type CorpusCase } from './corpus'; import { guardCorpus, type GuardCorpusResult } from './corpusGuard'; import { dayStart, readCycles, usageToday, type HealCycle } from './cycles'; import { corpusCases } from './healApply'; -import { caseProbe, libraryOf, unscored } from './healState'; +import { BudgetSpent, caseProbe, libraryOf, unscored } from './healState'; import type { MonitorVerdict } from './monitor'; import { voteOnOutcome } from './reflect'; import { harnessHash, readAgentHarness } from './revisions'; @@ -17,6 +17,7 @@ import type { WatchOptions } from './watch'; export interface WatchContext { owner: string; + leaseId: string; store: CheckpointStore; editor: ImproveOptions['editor']; signal?: AbortSignal; @@ -24,9 +25,10 @@ export interface WatchContext { export interface WatchMemory { paused?: string; - signatures?: string[]; revision?: string; + signatures?: string[]; failed?: number; + newest?: string; outages?: number; guardAt?: number; spentDay?: number; @@ -52,9 +54,7 @@ export async function saveWatch(dir: string, change: Partial): Prom } export function budgetLeft(dir: string, settings: ImproveSettings, now: number): number { - const watch = watchOf(dir); - const spent = watch.spentDay === dayStart(now) ? (watch.spent ?? 0) : 0; - return settings.budget.maxTokensPerDay - spent - usageToday(dir, now).tokens; + return settings.budget.maxTokensPerDay - usageToday(dir, now).tokens; } async function addSpent(dir: string, tokens: number, now: number): Promise { @@ -66,16 +66,35 @@ async function addSpent(dir: string, tokens: number, now: number): Promise }); } -function counting(probe: ImproveOptions['probe'], spent: { tokens: number }): ImproveOptions['probe'] { +interface Spend { + tokens: number; + limit: number; +} + +function metered(probe: ImproveOptions['probe'], spend: Spend): ImproveOptions['probe'] { return async (harness, caseId, repeat) => { + if (spend.tokens >= spend.limit) throw new BudgetSpent(); const run = await probe(harness, caseId, repeat); - spent.tokens += run.tokens; + spend.tokens += run.tokens; return run; }; } -function corpusProbe(options: WatchOptions, cases: readonly CapturedCase[], spent: { tokens: number }): ImproveOptions['probe'] { - return counting(unscored(caseProbe(cases, options.runCase, options.check), new Set(), undefined), spent); +async function withSpend(options: WatchOptions, now: number, task: (spend: Spend) => Promise): Promise { + const spend: Spend = { tokens: 0, limit: budgetLeft(options.dir, options.settings, now) }; + if (spend.limit <= 0) return undefined; + try { + return await task(spend); + } catch (error) { + if (error instanceof BudgetSpent) return undefined; + throw error; + } finally { + await addSpent(options.dir, spend.tokens, now); + } +} + +function corpusProbe(options: WatchOptions, cases: readonly CapturedCase[], spend: Spend, signal: AbortSignal | undefined): ImproveOptions['probe'] { + return metered(unscored(caseProbe(cases, options.runCase, options.check), new Set(), signal), spend); } function cycleOfApply(dir: string, applySeq: number): HealCycle | undefined { @@ -93,45 +112,53 @@ function fixedCases(options: WatchOptions, cycle: HealCycle): CapturedCase[] { }); } -async function admitFixed(options: WatchOptions, cycle: HealCycle, now: number): Promise { +async function admitFixed(options: WatchOptions, cycle: HealCycle, now: number, signal: AbortSignal | undefined): Promise { const cases = fixedCases(options, cycle); - if (cases.length === 0 || budgetLeft(options.dir, options.settings, now) <= 0) return; - const spent = { tokens: 0 }; + if (cases.length === 0) return; const harness = readAgentHarness(options.dir); - const { admitted } = await admitProtectedCases({ harness, cases, probe: corpusProbe(options, cases, spent), runs: ADMISSION_RUNS }); - await addSpent(options.dir, spent.tokens, now); + const admitted = await withSpend(options, now, async (spend) => (await admitProtectedCases({ harness, cases, probe: corpusProbe(options, cases, spend, signal), runs: ADMISSION_RUNS })).admitted); + if (!admitted || admitted.length === 0) return; const revision = harnessHash(harness); const addedAt = new Date(now).toISOString(); const fixed = admitted.map((item): CorpusCase => ({ id: item.id, traceId: item.traceId, input: item.input, addedAt, revision, origin: 'fixed', baseline: { passed: ADMISSION_RUNS, scored: ADMISSION_RUNS } })); - if (fixed.length > 0) await addToCorpus(options.dir, fixed, options.settings.guard.maxCases); + await addToCorpus(options.dir, fixed, options.settings.guard.maxCases); +} + +function harvestClaim(current: Partial): number | undefined { + const applySeq = current.monitor?.applySeq; + return applySeq !== undefined && current.watch?.harvested !== applySeq ? applySeq : undefined; +} + +async function claimHarvest(dir: string): Promise { + let claimed: number | undefined; + await updateState(dir, (current) => { + claimed = harvestClaim(current); + return claimed === undefined ? {} : { watch: { ...current.watch, harvested: claimed } }; + }); + return claimed; } -/** Votes once on the library entries behind an apply the monitor decided, and adds the cases a kept edit fixed to the corpus once they pass 6 of 6. */ -export async function harvest(options: WatchOptions, monitor: MonitorVerdict, now: number): Promise { +export async function harvest(options: WatchOptions, monitor: MonitorVerdict, now: number, signal?: AbortSignal): Promise { if (monitor.state !== 'kept' && monitor.state !== 'rolled-back') return; - const state = readState(options.dir); - const applySeq = state.monitor?.applySeq; - if (applySeq === undefined || state.watch?.harvested === applySeq) return; - await saveWatch(options.dir, { harvested: applySeq }); + const applySeq = await claimHarvest(options.dir); + if (applySeq === undefined) return; const cycle = cycleOfApply(options.dir, applySeq); if (!cycle?.result) return; const library = libraryOf(options.dir, options.settings); if (library) voteOnOutcome(library.dir, { outcome: monitor.state, result: cycle.result, domain: library.domain }); - if (monitor.state === 'kept') await admitFixed(options, cycle, now); + if (monitor.state === 'kept') await admitFixed(options, cycle, now, signal); } -/** Replays the corpus under the current revision once `guard.everyHours` have passed, rolling back on a regression. */ -export async function guardScheduled(options: WatchOptions, now: number): Promise { +export async function guardScheduled(options: WatchOptions, now: number, signal?: AbortSignal): Promise { const { dir, settings } = options; const last = watchOf(dir).guardAt; if (last !== undefined && now - last < settings.guard.everyHours * HOUR_MS) return undefined; const corpus = readCorpus(dir); - if (corpus.length === 0 || budgetLeft(dir, settings, now) <= 0) return undefined; + if (corpus.length === 0) return undefined; const revision = harnessHash(readAgentHarness(dir)); const earlier = readAudit(dir).filter((entry) => (entry.action === 'guard-pass' || entry.action === 'guard-fail') && entry.to === revision).length; - const spent = { tokens: 0 }; - const guard = await guardCorpus(dir, { probe: corpusProbe(options, corpusCases(corpus), spent), runs: settings.guard.runs, rollback: true, firstRepeat: GUARD_FIRST_REPEAT + earlier * GUARD_RANGE }); - await addSpent(dir, spent.tokens, now); - await saveWatch(dir, { guardAt: now }); + const firstRepeat = GUARD_FIRST_REPEAT + earlier * GUARD_RANGE; + const guard = await withSpend(options, now, (spend) => guardCorpus(dir, { probe: corpusProbe(options, corpusCases(corpus), spend, signal), runs: settings.guard.runs, rollback: true, firstRepeat, ...(signal && { signal }) })); + if (guard) await saveWatch(dir, { guardAt: now }); return guard; }