diff --git a/api/improve.api.md b/api/improve.api.md index 49ce3612..f728ca87 100644 --- a/api/improve.api.md +++ b/api/improve.api.md @@ -2544,6 +2544,7 @@ export interface WatchOptions extends Omit void; onCycle?: (cycle: HealCycle) => void; onError?: (error: unknown) => void; + registry?: string; reloadSettings?: () => ImproveSettings; store?: CheckpointStore; } diff --git a/docs/observability.md b/docs/observability.md index 9dc5cb9b..eb34d682 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -516,6 +516,17 @@ posted once per revision. The reason is plain text: links, images and markup are stripped. Each event is also one stderr line, and a webhook that fails never fails the cycle. +The sweep also records each `agent.score_regression` in the db's +`score_regressions` table, with or without webhooks. A running +`lousho improve --watch` polls it every tick and starts a cycle with +`trigger: 'score'` for a regression of the agent's current revision, instead +of waiting for new failed runs; every budget, the cooldown and the kill switch +still apply. The daemon never writes to an agent directory. A watcher lists +itself in `watchers.json` beside the db while it runs (informational: the sweep +records every regression, once per agent and revision, across daemon restarts). +A score cycle learns from the revision's failed and low-scored runs; a +regression with neither ends `idle`. + Insights averages runs/day over the days an agent was active in the window (from the later of the window start and the agent's first sighting), so an agent first seen today with 12 runs reports 12/day, not 12/7. The per-bucket diff --git a/llms-full.txt b/llms-full.txt index 9527176a..af2418fe 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -15679,6 +15679,17 @@ posted once per revision. The reason is plain text: links, images and markup are stripped. Each event is also one stderr line, and a webhook that fails never fails the cycle. +The sweep also records each `agent.score_regression` in the db's +`score_regressions` table, with or without webhooks. A running +`lousho improve --watch` polls it every tick and starts a cycle with +`trigger: 'score'` for a regression of the agent's current revision, instead +of waiting for new failed runs; every budget, the cooldown and the kill switch +still apply. The daemon never writes to an agent directory. A watcher lists +itself in `watchers.json` beside the db while it runs (informational: the sweep +records every regression, once per agent and revision, across daemon restarts). +A score cycle learns from the revision's failed and low-scored runs; a +regression with neither ends `idle`. + Insights averages runs/day over the days an agent was active in the window (from the later of the window start and the agent's first sighting), so an agent first seen today with 12 runs reports 12/day, not 12/7. The per-bucket diff --git a/src/cli/improve.ts b/src/cli/improve.ts index 48d29ac7..bb2b89a5 100644 --- a/src/cli/improve.ts +++ b/src/cli/improve.ts @@ -533,6 +533,7 @@ async function watchDir(args: ImproveArgs, io: ImproveIo, dir: string, config: C editor, settings, reloadSettings: () => loadImproveSettings(dir, args.approval ? { approval: args.approval } : {}), + registry: path.join(path.dirname(dbPathOf(args.db)), 'watchers.json'), agent, protect: args.protect, withRecoveries: args.withRecoveries === true, diff --git a/src/improve/watch.test.ts b/src/improve/watch.test.ts index cfa7ad56..37592ded 100644 --- a/src/improve/watch.test.ts +++ b/src/improve/watch.test.ts @@ -19,6 +19,7 @@ import { updateState } from './state'; import type { Harness } from './types'; import { isDefinedSchedule } from '../schedules/defineSchedule'; import { noticeEvent } from './notify'; +import { readWatchers, registerWatcher } from './watchers'; import { improveSchedule, watchAgent, watchContext, watchTick, type WatchOptions } from './watch'; vi.mock('./heal', async (original) => { @@ -264,6 +265,63 @@ describe('watchTick', () => { }); }); +describe('the score regression trigger', () => { + const mark = (at: number, revision: string | null = currentRevision(dir)) => db.prepare('INSERT INTO score_regressions (agent_id, revision, at, details) VALUES (?, ?, ?, ?)').run('support', revision, at, '{}'); + + it('a regression marker triggers a cycle with trigger score', async () => { + const context = watchContext(options()); + seedFailures(1); + expect((await watchTick(options(), context)).cycle?.outcome).toBe('idle'); + mark(Date.now()); + const scored = await watchTick(options(), context); + expect(vi.mocked(healOnce).mock.calls.at(-1)?.[0].trigger).toBe('score'); + expect(scored.cycle).toMatchObject({ trigger: 'score', outcome: 'applied' }); + await watchTick(options(), context); + expect(vi.mocked(healOnce).mock.calls.at(-1)?.[0].trigger).toBe('failures'); + }); + + it("ignores a marker for another revision, and the trigger respects the watcher's budgets", async () => { + const context = watchContext(options()); + seedFailures(1); + mark(Date.now(), 'f'.repeat(64)); + await watchTick(options(), context); + expect(vi.mocked(healOnce).mock.calls.at(-1)?.[0].trigger).toBe('failures'); + mark(Date.now() + 1); + expect((await watchTick(options(), context)).cycle).toMatchObject({ trigger: 'score', outcome: 'applied' }); + mark(Date.now() + 2, null); + expect((await watchTick(options(), context)).cycle).toMatchObject({ trigger: 'score', outcome: 'budget' }); + await appendAudit(dir, { actor: 'monitor', action: 'rollback', reason: 'Regressed.' }); + mark(Date.now() + 3, null); + expect((await watchTick(options(), context)).cycle).toMatchObject({ trigger: 'score', outcome: 'cooldown' }); + }); + + it('a paused tick keeps the marker for the next tick that runs', async () => { + const context = watchContext(options()); + seedFailures(1); + mark(Date.now()); + stop(); + expect((await watchTick(options(), context)).state).toBe('paused'); + rmSync(path.join(dir, '.lousho', 'improve.stop')); + expect((await watchTick(options(), context)).cycle).toMatchObject({ trigger: 'score', outcome: 'applied' }); + }); + + it('keeps one entry per agent directory and drops a dead watcher', async () => { + const registry = path.join(root, 'watchers.json'); + writeFileSync(registry, JSON.stringify({ watchers: [{ agent: 'old', dir: '/gone', owner: 'x', pid: 2 ** 22 + 12345, since: 'then' }] })); + await registerWatcher(registry, { agent: 'support', dir, owner: 'a' }); + await registerWatcher(registry, { agent: 'support', dir, owner: 'b' }); + expect(readWatchers(registry).map((entry) => entry.owner)).toEqual(['b']); + }); + + it('registers the watcher beside the observe db while it runs', async () => { + const registry = path.join(root, 'watchers.json'); + const watcher = watchAgent(options({ registry, intervalMs: 60_000 })); + await vi.waitFor(() => expect(readWatchers(registry)).toEqual([expect.objectContaining({ agent: 'support', dir })])); + await watcher.stop(); + expect(readWatchers(registry)).toEqual([]); + }); +}); + describe('notifications', () => { it('an apply posts one improve.applied', async () => { seedFailures(); diff --git a/src/improve/watch.ts b/src/improve/watch.ts index 33a2d976..2674768c 100644 --- a/src/improve/watch.ts +++ b/src/improve/watch.ts @@ -4,6 +4,7 @@ import type { CheckpointStore } from '../execution/checkpoint'; import { ConfigurationError } from '../execution/errors'; import { failureReportFromDb } from '../observe/failures'; import { runsByRevision } from '../observe/store'; +import { scoreRegressionsSince } from '../observe/sweep'; import type { LLMProvider } from '../providers'; import { createRateLimiter, withRateLimit } from '../providers/rateLimit'; import { resolveProvider } from '../providers/resolveProvider'; @@ -21,7 +22,8 @@ import { harnessHash, readAgentHarness } from './revisions'; import { loadImproveSettings, type ImproveSettings } from './settings'; import { readState, updateState } from './state'; import type { ImproveOptions } from './types'; -import { budgetLeft, guardScheduled, harvest, saveWatch, watchOf, type WatchContext, type WatchState } from './watchSteps'; +import { registerWatcher, unregisterWatcher } from './watchers'; +import { budgetLeft, guardScheduled, harvest, saveWatch, watchOf, type WatchContext, type WatchMemory, type WatchState } from './watchSteps'; /** Options for {@link watchAgent} and {@link improveSchedule}. */ export interface WatchOptions extends Omit { @@ -41,6 +43,8 @@ export interface WatchOptions extends Omit void; + /** A `watchers.json` beside the observe db, where the watcher registers `{ agent, dir }` while it runs. */ + registry?: string; } /** What one watch tick did. */ @@ -145,16 +149,23 @@ async function countOutage(dir: string, cycle: HealCycle, notify: Notify): Promi notify({ type: 'improve.paused', reason, cycle: cycle.id }); } +function sameFailures(last: WatchMemory, key: ReturnType): boolean { + if (last.signatures === undefined) return false; + return last.revision === key.revision && last.failed === key.failed && last.newest === key.newest && last.signatures.join('\n') === key.signatures.join('\n'); +} + async function runCycle(options: WatchOptions, context: WatchContext, trigger: HealCycle['trigger'], now: number): Promise { 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, alerter: _alerter, log: _log, ...heal } = options; - const cycle = await healOnce({ ...heal, editor: context.editor, trigger, maxTokens: budgetLeft(options.dir, options.settings, now), ...(context.signal && { signal: context.signal }) }); - if (!NOT_WORKED.includes(cycle.outcome)) await saveWatch(options.dir, { revision: key.revision, signatures: key.signatures, failed: key.failed, ...(key.newest && { newest: key.newest }) }); + const marked = scoreRegressionsSince(options.db, options.agent, last.scoreAt ?? 0) + .filter((mark) => mark.revision === null || mark.revision === key.revision) + .at(-1)?.at; + if (marked === undefined && sameFailures(last, key)) 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, alerter: _alerter, log: _log, registry: _registry, ...heal } = options; + const cycle = await healOnce({ ...heal, editor: context.editor, trigger: marked === undefined ? trigger : 'score', maxTokens: budgetLeft(options.dir, options.settings, now), ...(context.signal && { signal: context.signal }) }); + if (!NOT_WORKED.includes(cycle.outcome)) await saveWatch(options.dir, { revision: key.revision, signatures: key.signatures, failed: key.failed, ...(key.newest && { newest: key.newest }), ...(marked !== undefined && { scoreAt: marked }) }); await countOutage(options.dir, cycle, context.notify); const notice = cycleNotice(cycle, options.dir); if (notice) context.notify(notice); @@ -228,6 +239,7 @@ export function watchAgent(options: WatchOptions): { if (!stopped) timer = setTimeout(loop, options.intervalMs ?? DEFAULT_INTERVAL_MS); }); }; + const registered = options.registry ? registerWatcher(options.registry, { agent: options.agent, dir: options.dir, owner: context.owner }).catch(report) : Promise.resolve(); loop(); return { async stop() { @@ -236,6 +248,8 @@ export function watchAgent(options: WatchOptions): { controller.abort(); await running; await context.store.releaseLease?.(context.leaseId, context.owner); + await registered; + if (options.registry) await unregisterWatcher(options.registry, context.owner).catch(report); }, }; } diff --git a/src/improve/watchSteps.ts b/src/improve/watchSteps.ts index cfb30d8e..b589ea0f 100644 --- a/src/improve/watchSteps.ts +++ b/src/improve/watchSteps.ts @@ -37,6 +37,7 @@ export interface WatchMemory { spent?: number; harvested?: number; guardNotified?: string; + scoreAt?: number; } export interface WatchState { diff --git a/src/improve/watchers.ts b/src/improve/watchers.ts new file mode 100644 index 00000000..936cabdc --- /dev/null +++ b/src/improve/watchers.ts @@ -0,0 +1,57 @@ +import { mkdirSync, readFileSync } from 'node:fs'; +import path from 'node:path'; +import { writeFileAtomic } from '../storage/atomicFile'; +import { withFileLock } from '../storage/fileLock'; + +export interface WatcherEntry { + agent: string; + dir: string; + owner: string; + pid: number; + since: string; +} + +function alive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === 'EPERM'; + } +} + +function isEntry(item: unknown): item is WatcherEntry { + const entry = item as Partial | null; + return [entry?.agent, entry?.dir, entry?.owner].every((value) => typeof value === 'string'); +} + +function readEntries(file: string): WatcherEntry[] { + try { + const parsed: unknown = JSON.parse(readFileSync(file, 'utf8')); + const list = (parsed as { watchers?: unknown } | null)?.watchers; + return Array.isArray(list) ? list.filter(isEntry) : []; + } catch { + return []; + } +} + +async function update(file: string, change: (entries: WatcherEntry[]) => WatcherEntry[]): Promise { + mkdirSync(path.dirname(file), { recursive: true }); + await withFileLock(`${file}.lock`, async () => { + const live = readEntries(file).filter((item) => !Number.isInteger(item.pid) || alive(item.pid)); + await writeFileAtomic(file, `${JSON.stringify({ watchers: change(live) }, null, 2)}\n`); + }); +} + +export function readWatchers(file: string): WatcherEntry[] { + return readEntries(file); +} + +export async function registerWatcher(file: string, entry: Omit): Promise { + const dir = path.resolve(entry.dir); + await update(file, (entries) => [...entries.filter((item) => item.owner !== entry.owner && !(item.dir === dir && item.agent === entry.agent)), { ...entry, dir, pid: process.pid, since: new Date().toISOString() }]); +} + +export async function unregisterWatcher(file: string, owner: string): Promise { + await update(file, (entries) => entries.filter((item) => item.owner !== owner)); +} diff --git a/src/observe/schema.ts b/src/observe/schema.ts index 639e6370..bebe026d 100644 --- a/src/observe/schema.ts +++ b/src/observe/schema.ts @@ -277,6 +277,15 @@ export const OBSERVE_MIGRATIONS: readonly string[] = [ ` CREATE INDEX observations_agent_roots ON observations (agent_id, start) WHERE parent_obs_id IS NULL AND kind = 'span'; `, + ` + CREATE TABLE score_regressions ( + agent_id TEXT NOT NULL, + revision TEXT, + at INTEGER NOT NULL, + details TEXT NOT NULL, + PRIMARY KEY (agent_id, at) + ); + `, ]; /** diff --git a/src/observe/scoreRegressionAlert.test.ts b/src/observe/scoreRegressionAlert.test.ts index e6303cbf..bd59c852 100644 --- a/src/observe/scoreRegressionAlert.test.ts +++ b/src/observe/scoreRegressionAlert.test.ts @@ -1,9 +1,12 @@ +import { mkdirSync, mkdtempSync, readdirSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; import { afterEach, describe, expect, it, vi } from 'vitest'; import type { SqlDatabase } from '../storage/sqlite/driver'; import type { AlertEvent } from './alerts'; import type { IngestEvent } from './envelope'; import { ingestBatch, openDb } from './store'; -import { startStatusSweep, sweepAgents, type ScoreRegressionMemory } from './sweep'; +import { scoreRegressionsSince, startStatusSweep, sweepAgents, type ScoreRegressionMemory } from './sweep'; const T0 = 1_700_000_000_000; const R1 = 'a'.repeat(64); @@ -154,7 +157,7 @@ describe('the agent.score_regression alert', () => { db.close(); }); - it('the daemon sweep keeps edge memory only when it delivers alerts', () => { + it('the daemon sweep delivers a regression alert once', () => { const db = openDb(':memory:'); ingestBatch(db, [heartbeat(), ...runs(0, 15, R1, 0.9), ...runs(15, 15, R2, 0.6)]); const delivered: AlertEvent[] = []; @@ -169,6 +172,46 @@ describe('the agent.score_regression alert', () => { db.close(); }); + it('records no marker for a steady series', () => { + const db = openDb(':memory:'); + ingestBatch(db, [heartbeat(), ...runs(0, 60, R1, 0.9)]); + const sweep = startStatusSweep(db, { intervalMs: 0, now: () => NOW }); + sweep.run(NOW + 1); + sweep.stop(); + expect(scoreRegressionsSince(db, 'support', 0)).toEqual([]); + db.close(); + }); + + it('records each regression as a marker a watcher polls, even without an alert sink, and writes nothing else', () => { + const root = mkdtempSync(path.join(tmpdir(), 'lousho-sweep-')); + const agentDir = path.join(root, 'support'); + mkdirSync(agentDir); + writeFileSync(path.join(agentDir, 'instructions.md'), 'Be brief.\n'); + const db = openDb(path.join(root, 'observe.db')); + ingestBatch(db, [heartbeat(), ...runs(0, 15, R1, 0.9), ...runs(15, 15, R2, 0.6)]); + const before = readdirSync(root, { recursive: true }).map(String).sort(); + const sweep = startStatusSweep(db, { intervalMs: 0, now: () => NOW }); + ingestBatch(db, runs(30, 10, R2, 0.6)); + sweep.run(NOW + 1); + ingestBatch(db, runs(40, 10, R2, 0.6)); + sweep.run(NOW + 2); + sweep.stop(); + expect(scoreRegressionsSince(db, 'support', 0)).toEqual([{ at: NOW + 2, revision: R2 }]); + expect(scoreRegressionsSince(db, 'uuid-1', 0)).toEqual([{ at: NOW + 2, revision: R2 }]); + expect(scoreRegressionsSince(db, 'uuid-1', NOW + 2)).toEqual([]); + const restarted = startStatusSweep(db, { intervalMs: 0, now: () => NOW }); + for (let i = 0; i < 4; i++) { + ingestBatch(db, runs(50 + i * 10, 10, R2, 0.6)); + restarted.run(NOW + 10 + i); + } + restarted.stop(); + expect(scoreRegressionsSince(db, 'support', 0)).toHaveLength(1); + expect(readdirSync(agentDir)).toEqual(['instructions.md']); + expect(readdirSync(root, { recursive: true }).map(String).sort().filter((name) => !name.startsWith('observe.db'))).toEqual(before.filter((name) => !name.startsWith('observe.db'))); + db.close(); + rmSync(root, { recursive: true, force: true }); + }); + it('reads each agent through its root index and stays under 50 ms per agent at 10k scores', () => { const db = openDb(':memory:'); const agents = Array.from({ length: 20 }, (_, i) => `agent-${i}`); diff --git a/src/observe/store.ts b/src/observe/store.ts index f5dcb96c..75b2675a 100644 --- a/src/observe/store.ts +++ b/src/observe/store.ts @@ -653,6 +653,7 @@ export function pruneBefore(db: SqlDatabase, cutoffMs: number): PruneResult { const orphanScores = db .prepare('DELETE FROM scores WHERE created_at < ? AND NOT EXISTS (SELECT 1 FROM observations o WHERE o.trace_id = scores.trace_id)') .run(cutoffMs); + db.prepare('DELETE FROM score_regressions WHERE at < ?').run(cutoffMs); const sessions = db .prepare( `DELETE FROM sessions WHERE status != 'active' AND COALESCE(ended_at, started_at) < ? diff --git a/src/observe/sweep.ts b/src/observe/sweep.ts index 58544b80..002986eb 100644 --- a/src/observe/sweep.ts +++ b/src/observe/sweep.ts @@ -78,7 +78,7 @@ export interface SweepPassOptions { errorRateThreshold?: number; /** Edge memory for that alert - `agent_id` -> the day bucket already alerted on. */ errorRateState?: Map; - /** Edge memory for `agent.score_regression`, per `agent_id`: the revisions already alerted on, when the gate last judged and its streak. Without it the score pass is skipped. */ + /** Edge memory for `agent.score_regression`, per `agent_id`: the revisions already alerted on, when the gate last judged and its streak. Without it the score pass is skipped; `startStatusSweep` always passes one. */ scoreRegressionState?: Map; } @@ -109,6 +109,10 @@ function judgedRegression(db: SqlDatabase, agentId: string, memory: ScoreRegress return memory.streak >= SCORE_STREAK ? { points, gate } : undefined; } +function recorded(db: SqlDatabase, agentId: string, revision: string | null): boolean { + return db.prepare('SELECT 1 AS hit FROM score_regressions WHERE agent_id = ? AND revision IS ?').get(agentId, revision) !== undefined; +} + function scoreRegressionAlert(db: SqlDatabase, row: SqlRow, now: number, state: Map): AlertEvent | undefined { const agentId = String(row.agent_id); const memory = state.get(agentId) ?? { alerted: new Set(), streak: 0 }; @@ -117,7 +121,10 @@ function scoreRegressionAlert(db: SqlDatabase, row: SqlRow, now: number, state: if (judged === undefined) return undefined; const { points, gate } = judged; const revision = points.at(-1)?.revision ?? null; - if (memory.alerted.has(revision)) return undefined; + if (memory.alerted.has(revision) || recorded(db, agentId, revision)) { + memory.alerted.add(revision); + return undefined; + } memory.alerted.add(revision); return { type: 'agent.score_regression', @@ -279,7 +286,13 @@ export function sweepAgents(db: SqlDatabase, now: number = Date.now(), options: }); } } - alerts.push(...scoreRegressionAlerts(db, now, options.scoreRegressionState)); + const regressions = scoreRegressionAlerts(db, now, options.scoreRegressionState); + try { + recordRegressions(db, regressions); + } catch (error) { + console.error('[lousho observe] recording a score regression failed:', error); + } + alerts.push(...regressions); return { transitions, alerts }; } @@ -331,6 +344,23 @@ function idleSessionPass(db: SqlDatabase, idleMs: number | undefined): (at: numb }; } +function recordRegressions(db: SqlDatabase, alerts: readonly AlertEvent[]): void { + const insert = db.prepare('INSERT OR IGNORE INTO score_regressions (agent_id, revision, at, details) VALUES (?, ?, ?, ?)'); + for (const alert of alerts) { + const revision = alert.details.revision; + insert.run(alert.agent_id, typeof revision === 'string' ? revision : null, alert.at, JSON.stringify(alert.details)); + } +} + +export function scoreRegressionsSince(db: SqlDatabase, agent: string, since: number): { at: number; revision: string | null }[] { + return db + .prepare( + 'SELECT at, revision FROM score_regressions WHERE at > ? AND (agent_id = ? OR (NOT EXISTS (SELECT 1 FROM agents WHERE agent_id = ?) AND agent_id IN (SELECT agent_id FROM agents WHERE name = ?))) ORDER BY at' + ) + .all(since, agent, agent, agent) + .map((row) => ({ at: Number(row.at), revision: row.revision === null ? null : String(row.revision) })); +} + /** * The daemon's sweep loop. The first pass runs immediately: agents that died * while the daemon was down are marked (and reported) at boot, not one @@ -364,7 +394,7 @@ export function startStatusSweep(db: SqlDatabase, options: StatusSweepOptions = const result = sweepAgents(db, at, { ...(options.errorRateThreshold !== undefined ? { errorRateThreshold: options.errorRateThreshold } : {}), errorRateState: errorRateSeen, - ...(options.onAlerts !== undefined && { scoreRegressionState: scoreRegressionSeen }), + scoreRegressionState: scoreRegressionSeen, }); if (result.alerts.length > 0) options.onAlerts?.(result.alerts); return result;