diff --git a/api/improve.api.md b/api/improve.api.md index 95c1ef0a..681bcf8d 100644 --- a/api/improve.api.md +++ b/api/improve.api.md @@ -190,7 +190,7 @@ export function approvePending(dir: string, id: string, approvedBy: string): Pro // @public export interface AuditEntry { - action: 'propose' | 'apply' | 'reject' | 'stale' | 'rollback' | 'retire' | 'guard-pass' | 'guard-fail' | 'pause' | 'resume' | 'canary-start' | 'canary-end'; + action: 'propose' | 'apply' | 'reject' | 'stale' | 'rollback' | 'retire' | 'guard-pass' | 'guard-fail' | 'unverified' | 'pause' | 'resume' | 'canary-start' | 'canary-end'; actor: 'gate' | 'canary' | 'monitor' | 'guard' | 'scheduler' | `person:${string}`; at: string; cycle?: string; @@ -321,6 +321,12 @@ export interface CaseVerdict { role: 'training' | 'holdout'; } +// @public +export function checkApplied(db: SqlDatabase, dir: string, settings: Pick, options: { + agent: string; + now?: number; +}): Promise; + // @public export function citedEntries(notes: string): string[]; @@ -879,6 +885,17 @@ type MessageRole = 'system' | 'user' | 'assistant' | 'tool'; // @public type ModerationCategory = 'hate' | 'harassment' | 'self-harm' | 'sexual' | 'sexual-minors' | 'violence' | 'illicit'; +// @public +export interface MonitorVerdict { + after?: WindowStat; + baselineRuns?: number; + before?: WindowStat; + reason?: string; + runs?: number; + state: 'idle' | 'pending' | 'kept' | 'kept-unverified' | 'regressed' | 'rolled-back'; + to?: string; +} + // @public interface ObjectDeltaEvent extends AgentEventBase<'object.delta'> { object: unknown; @@ -1478,6 +1495,16 @@ export function voteOnOutcome(dir: string, options: { entryId?: string; }): LibraryEntry[]; +// @public +export interface WindowStat { + lower: number; + mean: number; + pass3?: number; + passRate: number; + runs: number; + upper: number; +} + // (No @packageDocumentation comment for this package) ``` diff --git a/src/improve/apply.ts b/src/improve/apply.ts index ba096048..c2109096 100644 --- a/src/improve/apply.ts +++ b/src/improve/apply.ts @@ -222,7 +222,10 @@ export async function rollbackHarness(dir: string, to: string, by: RollbackBy): export function lastGoodRevision(dir: string): string | undefined { const revisions = listRevisions(dir); const latestGuard = new Map(); - for (const entry of readAudit(dir)) if ((entry.action === 'guard-pass' || entry.action === 'guard-fail') && entry.to !== undefined) latestGuard.set(entry.to, entry.action); + for (const entry of readAudit(dir)) { + if ((entry.action === 'guard-pass' || entry.action === 'guard-fail') && entry.to !== undefined) latestGuard.set(entry.to, entry.action); + if (entry.action === 'retire' && entry.from !== undefined) latestGuard.set(entry.from, 'guard-fail'); + } const passed = revisions.find((revision) => latestGuard.get(revision.id) === 'guard-pass')?.id; return passed ?? revisions.filter((revision) => revision.source === 'baseline').at(-1)?.id; } diff --git a/src/improve/audit.ts b/src/improve/audit.ts index b0a14e45..3760b901 100644 --- a/src/improve/audit.ts +++ b/src/improve/audit.ts @@ -13,7 +13,7 @@ export interface AuditEntry { /** Who or what made the change: a loop stage, or `person:`. */ actor: 'gate' | 'canary' | 'monitor' | 'guard' | 'scheduler' | `person:${string}`; /** What happened. */ - action: 'propose' | 'apply' | 'reject' | 'stale' | 'rollback' | 'retire' | 'guard-pass' | 'guard-fail' | 'pause' | 'resume' | 'canary-start' | 'canary-end'; + action: 'propose' | 'apply' | 'reject' | 'stale' | 'rollback' | 'retire' | 'guard-pass' | 'guard-fail' | 'unverified' | 'pause' | 'resume' | 'canary-start' | 'canary-end'; /** The revision before the change. */ from?: string; /** The revision after the change. */ diff --git a/src/improve/index.ts b/src/improve/index.ts index 42eca608..3955bbd7 100644 --- a/src/improve/index.ts +++ b/src/improve/index.ts @@ -19,6 +19,9 @@ export { currentRevision, harnessHash, listRevisions, readAgentHarness, snapshot export type { HarnessRevision } from './revisions'; export { applyProposal, lastGoodRevision, rollbackHarness } from './apply'; export type { ApplyInput, ApplyResult, Approval, RollbackBy, RollbackResult } from './apply'; +export { checkApplied } from './monitor'; +export type { MonitorVerdict } from './monitor'; +export type { WindowStat } from '../observe/continuousScore'; export { loadImproveSettings } from './settings'; export type { ImproveSettings } from './settings'; export { approvePending, listPending, rejectPending, savePending } from './pending'; diff --git a/src/improve/monitor.test.ts b/src/improve/monitor.test.ts new file mode 100644 index 00000000..609325b6 --- /dev/null +++ b/src/improve/monitor.test.ts @@ -0,0 +1,346 @@ +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 { ScorePoint } from '../observe/continuousScore'; +import type { IngestEvent } from '../observe/envelope'; +import { ingestBatch, openDb } from '../observe/store'; +import type { SqlDatabase } from '../storage/sqlite/driver'; +import * as applyModule from './apply'; +import { applyProposal, lastGoodRevision } from './apply'; +import { readAudit, verifyAudit } from './audit'; +import { checkApplied, judgeApplied } from './monitor'; +import { loadImproveSettings, type ImproveSettings } from './settings'; +import { currentRevision, readAgentHarness } from './revisions'; +import type { Harness, ImproveResult } from './types'; + +const DAY = 86_400_000; +let dir: string; +let db: SqlDatabase; +let traces = 0; + +const write = (file: string, text: string): void => { + mkdirSync(path.dirname(path.join(dir, file)), { recursive: true }); + writeFileSync(path.join(dir, file), text); +}; + +const summary = { runs: 3, excluded: 0, passRate: 0, noise: 0, tokensPerRun: 10 }; +const proposal = (harness: Harness): ImproveResult => ({ outcome: 'proposal', reason: 'Gate passed.', harness, base: summary, stages: [], spend: { tokens: 0 } }); + +async function apply(): Promise<{ from: string; to: string; at: number }> { + const base = readAgentHarness(dir); + const applied = await applyProposal(dir, { result: proposal({ ...base, instructions: 'Be careful. Check the receipt first.' }), base }, { mode: 'auto', by: 'gate' }); + const at = Date.parse(readAudit(dir).at(-1)!.at); + return { ...applied, at }; +} + +function runs(revision: string, outcomes: readonly number[], start: number, agent = 'support', duration = 10): void { + ingestBatch( + db, + outcomes.flatMap((value, i): IngestEvent[] => { + const trace = `t${traces++}`; + const at = start + i * 1000; + return [ + { type: 'observation', data: { obs_id: `root-${trace}`, trace_id: trace, agent_id: agent, kind: 'span', name: 'invoke_agent support', start: at, end: at + duration, status: 'ok', attrs: { 'lousho.run.finish_reason': 'stop', 'lousho.harness.revision': revision } } }, + { type: 'score', data: { score_id: `s-${trace}`, trace_id: trace, name: 'check', value, source: 'online', agent_id: agent, ts: at + duration + 10 } }, + ]; + }) + ); +} + +const rate = (passed: number, total: number): number[] => Array.from({ length: total }, (_, i) => (i < passed ? 1 : 0)); +const MONITOR_DEFAULTS: ImproveSettings['monitor'] = { minRuns: 20, window: 20, maxDrop: 0.05, autoRollback: true, giveUpDays: 7 }; +const check = (options: { monitor?: Partial; now?: number; agent?: string }) => + checkApplied(db, dir, { monitor: { ...MONITOR_DEFAULTS, ...options.monitor }, cooldownHours: 24 }, { agent: options.agent ?? 'support', ...(options.now !== undefined && { now: options.now }) }); + +beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'lousho-monitor-')); + write('instructions.md', 'Be careful.\n'); + db = openDb(':memory:'); +}); + +afterEach(() => { + db.close(); + rmSync(dir, { recursive: true, force: true }); +}); + +describe('checkApplied', () => { + it('pending below minRuns', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(5, 19), applied.at + 1000); + + expect(await check({ now: applied.at + DAY })).toEqual({ state: 'pending', runs: 19, baselineRuns: 20 }); + expect(readAudit(dir).map((entry) => entry.action)).toEqual(['apply']); + }); + + it('kept when the pass rate holds, audited as a guard pass', async () => { + const applied = await apply(); + runs(applied.from, rate(16, 20), applied.at - 100_000); + runs(applied.to, rate(17, 20), applied.at + 1000); + + const verdict = await check({ now: applied.at + DAY }); + expect(verdict).toMatchObject({ state: 'kept', before: { runs: 20, passRate: 0.8 }, after: { runs: 20, passRate: 0.85 } }); + expect(readAudit(dir).at(-1)).toMatchObject({ actor: 'monitor', action: 'guard-pass', from: applied.from, to: applied.to, evidence: { score: { before: 0.8, after: 0.85, runs: 20 } } }); + expect(currentRevision(dir)).toBe(applied.to); + }); + + it('rolls back on a 30-point drop with 20 runs each', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(12, 20), applied.at + 1000); + + const verdict = await check({ now: applied.at + DAY }); + expect(verdict).toMatchObject({ state: 'rolled-back', to: applied.from, reason: 'The pass rate fell from 90% to 60% over 20 runs.' }); + expect(currentRevision(dir)).toBe(applied.from); + expect(readFileSync(path.join(dir, 'instructions.md'), 'utf8')).toBe('Be careful.\n'); + expect(readAudit(dir).map((entry) => [entry.actor, entry.action])).toEqual([ + ['gate', 'apply'], + ['monitor', 'rollback'], + ['monitor', 'retire'], + ]); + expect(verifyAudit(dir)).toBeUndefined(); + }); + + it('does not roll back a 5-point dip within noise', async () => { + const applied = await apply(); + runs(applied.from, rate(17, 20), applied.at - 100_000); + runs(applied.to, rate(16, 20), applied.at + 1000); + + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'kept' }); + expect(currentRevision(dir)).toBe(applied.to); + }); + + it('waits for another look when a drop is not yet significant, then keeps it at the third', async () => { + const applied = await apply(); + const before = [...rate(17, 20), ...rate(17, 20), ...rate(17, 20)]; + runs(applied.from, before, applied.at - 100_000); + runs(applied.to, [...rate(14, 20), ...rate(16, 20), ...rate(16, 20)], applied.at + 1000); + + expect(judgeApplied(points(before), points(rate(14, 20)), MONITOR_DEFAULTS)).toEqual({ verdict: 'wait' }); + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'kept', after: { runs: 60 } }); + }); + + it('does not touch a hand-edited directory', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(6, 20), applied.at + 1000); + write('instructions.md', 'Edited by a person.\n'); + + expect(await check({ now: applied.at + DAY })).toEqual({ state: 'idle', reason: 'The directory changed by hand after the last apply, so the monitor leaves it alone.' }); + expect(readFileSync(path.join(dir, 'instructions.md'), 'utf8')).toBe('Edited by a person.\n'); + expect(readAudit(dir).map((entry) => entry.action)).toEqual(['apply']); + }); + + it('the M4 defaults from improve.json are the ones these tests use', () => { + expect(loadImproveSettings(dir, {}, {}).monitor).toEqual(MONITOR_DEFAULTS); + }); + + it('is idle with nothing applied', async () => { + expect(await check({})).toEqual({ state: 'idle', reason: 'No applied revision to watch.' }); + }); + + it('autoRollback false only reports', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(12, 20), applied.at + 1000); + + const verdict = await check({ monitor: { autoRollback: false }, now: applied.at + DAY }); + expect(verdict).toMatchObject({ state: 'regressed', reason: 'The pass rate fell from 90% to 60% over 20 runs.' }); + expect(currentRevision(dir)).toBe(applied.to); + expect(readAudit(dir).at(-1)).toMatchObject({ actor: 'monitor', action: 'guard-fail', reason: 'The pass rate fell from 90% to 60% over 20 runs. Automatic rollback is off.' }); + }); + + it('rollback writes a cooldown', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(12, 20), applied.at + 1000); + const now = applied.at + DAY; + + write('.lousho/state.json', JSON.stringify({ cycles: 2 })); + await check({ now }); + const state = JSON.parse(readFileSync(path.join(dir, '.lousho', 'state.json'), 'utf8')) as { cycles: number; cooldownUntil: string; monitor: { revision: string; verdict: { state: string } } }; + expect(state.cooldownUntil).toBe(new Date(now + DAY).toISOString()); + expect(state.monitor).toMatchObject({ revision: applied.to, verdict: { state: 'rolled-back' } }); + expect(state.cycles).toBe(2); + expect(lastGoodRevision(dir)).not.toBe(applied.to); + }); + + it('refuses to overwrite a state.json that is not a JSON object', async () => { + const applied = await apply(); + runs(applied.from, rate(16, 20), applied.at - 100_000); + runs(applied.to, rate(17, 20), applied.at + 1000); + write('.lousho/state.json', '[1, 2'); + + await expect(check({ now: applied.at + DAY })).rejects.toThrow(); + expect(readFileSync(path.join(dir, '.lousho', 'state.json'), 'utf8')).toBe('[1, 2'); + }); + + it('watches a revision again when it is applied again after a rollback', async () => { + const first = await apply(); + runs(first.from, rate(18, 20), first.at - 100_000); + runs(first.to, rate(12, 20), first.at + 1000); + expect(await check({ now: first.at + DAY })).toMatchObject({ state: 'rolled-back' }); + + const again = await apply(); + expect(again.to).toBe(first.to); + expect(await check({ now: again.at + DAY })).toMatchObject({ state: 'rolled-back' }); + expect(readAudit(dir).filter((entry) => entry.action === 'retire')).toHaveLength(2); + }); + + it("judges only the agent's own runs, found by id or registered name", async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(18, 20), applied.at + 1000); + runs(applied.to, rate(2, 20), applied.at + 1000, 'billing'); + ingestBatch(db, [{ type: 'agent.heartbeat', data: { agent_id: 'support', name: 'Support desk', ts: applied.at } }]); + + expect(await check({ now: applied.at + DAY, agent: 'Support desk' })).toMatchObject({ state: 'kept', after: { runs: 20, passRate: 0.9 } }); + }); + + it('leaves out a run that started before the apply and finished after it', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.from, rate(0, 5), applied.at - 500, 'support', 5000); + runs(applied.to, rate(18, 20), applied.at + 1000); + + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'kept', before: { runs: 20, passRate: 0.9 } }); + }); + + it('a refused rollback is audited once as a guard failure, beside the writer\'s own refusal, and reported', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(12, 20), applied.at + 1000); + rmSync(path.join(dir, '.lousho', 'revisions', `${applied.from}.json`), { force: true }); + + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'regressed', reason: 'The pass rate fell from 90% to 60% over 20 runs. The rollback was refused.' }); + expect(await check({ now: applied.at + DAY + 1 })).toMatchObject({ state: 'regressed' }); + expect(readAudit(dir).filter((entry) => entry.actor === 'monitor').map((entry) => entry.action)).toEqual(['reject', 'guard-fail']); + expect(currentRevision(dir)).toBe(applied.to); + }); + + it('refuses an empty agent and a name two agents share', async () => { + const applied = await apply(); + ingestBatch(db, [ + { type: 'agent.heartbeat', data: { agent_id: 'a1', name: 'twin', ts: applied.at } }, + { type: 'agent.heartbeat', data: { agent_id: 'a2', name: 'twin', ts: applied.at } }, + ]); + + await expect(check({ now: applied.at + DAY, agent: ' ' })).rejects.toThrow('empty name'); + await expect(check({ now: applied.at + DAY, agent: 'twin' })).rejects.toThrow("'twin' names 2 agents"); + }); + + it('settles any rollback failure once instead of retrying it every tick', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(12, 20), applied.at + 1000); + const failing = vi.spyOn(applyModule, 'rollbackHarness').mockRejectedValue(new Error('disk full')); + + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'regressed' }); + expect(await check({ now: applied.at + DAY + 1 })).toMatchObject({ state: 'regressed' }); + expect(failing).toHaveBeenCalledTimes(1); + expect(readAudit(dir).at(-1)).toMatchObject({ action: 'guard-fail', reason: expect.stringContaining('disk full') as unknown as string }); + failing.mockRestore(); + }); + + it('two concurrent checks audit one decision', async () => { + const applied = await apply(); + runs(applied.from, rate(16, 20), applied.at - 100_000); + runs(applied.to, rate(17, 20), applied.at + 1000); + + await Promise.all([check({ now: applied.at + DAY }), check({ now: applied.at + DAY })]); + expect(readAudit(dir).filter((entry) => entry.actor === 'monitor')).toHaveLength(1); + }); + + it('catches a continuous drop between the 0.05 limit and 0.15', async () => { + const applied = await apply(); + runs(applied.from, Array.from({ length: 20 }, () => 0.95), applied.at - 100_000); + runs(applied.to, Array.from({ length: 20 }, () => 0.85), applied.at + 1000); + + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'rolled-back', reason: 'The mean score fell from 0.95 to 0.85 over the newest 10 runs.' }); + }); + + it('a mean score that falls without crossing 0.5 is caught by the trajectory gate', async () => { + const applied = await apply(); + runs(applied.from, Array.from({ length: 20 }, () => 0.95), applied.at - 100_000); + runs(applied.to, Array.from({ length: 20 }, () => 0.7), applied.at + 1000); + + expect(await check({ now: applied.at + DAY })).toMatchObject({ state: 'rolled-back', reason: 'The mean score fell from 0.95 to 0.70 over the newest 10 runs.' }); + }); + + it('keeps a revision unverified after giveUpDays without enough runs, audited once and never rolled back', async () => { + const applied = await apply(); + runs(applied.to, rate(0, 5), applied.at + 1000); + + expect(await check({ now: applied.at + 6 * DAY })).toMatchObject({ state: 'pending', runs: 5 }); + expect(await check({ now: applied.at + 7 * DAY })).toEqual({ state: 'kept-unverified', runs: 5, baselineRuns: 0, reason: 'Kept unverified: 5 runs after 7 days, fewer than the 20 the monitor needs.' }); + expect(await check({ now: applied.at + 8 * DAY })).toMatchObject({ state: 'kept-unverified', runs: 5 }); + expect(readAudit(dir).filter((entry) => entry.action === 'unverified')).toHaveLength(1); + expect(currentRevision(dir)).toBe(applied.to); + }); + + it('a decided verdict is not judged or audited again', async () => { + const applied = await apply(); + runs(applied.from, rate(16, 20), applied.at - 100_000); + runs(applied.to, rate(17, 20), applied.at + 1000); + await check({ now: applied.at + DAY }); + runs(applied.to, rate(0, 40), applied.at + 100_000); + + expect(await check({ now: applied.at + 2 * DAY })).toMatchObject({ state: 'kept' }); + expect(readAudit(dir).filter((entry) => entry.actor === 'monitor')).toHaveLength(1); + }); + + it('after a rollback there is nothing left to watch', async () => { + const applied = await apply(); + runs(applied.from, rate(18, 20), applied.at - 100_000); + runs(applied.to, rate(12, 20), applied.at + 1000); + await check({ now: applied.at + DAY }); + + expect(await check({ now: applied.at + 2 * DAY })).toEqual({ state: 'idle', reason: 'No applied revision to watch.' }); + expect(existsSync(path.join(dir, '.lousho', 'state.json'))).toBe(true); + }); +}); + +function points(values: readonly number[]): ScorePoint[] { + return values.map((value, i) => ({ traceId: `p${i}`, at: i, value })); +} + +function seeded(seed: number): () => number { + let state = seed; + return () => { + state = (state * 1_664_525 + 1_013_904_223) >>> 0; + return state / 2 ** 32; + }; +} + +function rolledBack(random: () => number, before: number, after: number, continuous = false): boolean { + const value = (rate: number) => (continuous ? Math.min(1, Math.max(0, rate + (random() - 0.5) * 0.4)) : random() < rate ? 1 : 0); + const draw = (rate: number, count: number) => points(Array.from({ length: count }, () => value(rate))); + const earlier = draw(before, 60); + const later = draw(after, 60); + for (const runs of [20, 40, 60]) { + const judged = judgeApplied(earlier, later.slice(0, runs), MONITOR_DEFAULTS); + if (judged.verdict !== 'wait') return judged.verdict === 'regressed'; + } + return false; +} + +describe('judgeApplied over 1,000 seeded draws', () => { + it('false rollback is at most 5% on a stable 0.8 pass rate', () => { + const random = seeded(42); + const rollbacks = Array.from({ length: 1000 }, () => rolledBack(random, 0.8, 0.8)).filter(Boolean).length; + expect(rollbacks / 1000).toBeLessThanOrEqual(0.05); + }); + + it('detection is at least 90% for a true drop from 0.9 to 0.6', () => { + const random = seeded(7); + const rollbacks = Array.from({ length: 1000 }, () => rolledBack(random, 0.9, 0.6)).filter(Boolean).length; + expect(rollbacks / 1000).toBeGreaterThanOrEqual(0.9); + }); + + it('false rollback is at most 5% on a stable continuous score of 0.8, and a fall to 0.65 is caught', () => { + const random = seeded(11); + expect(Array.from({ length: 1000 }, () => rolledBack(random, 0.8, 0.8, true)).filter(Boolean).length / 1000).toBeLessThanOrEqual(0.05); + expect(Array.from({ length: 1000 }, () => rolledBack(random, 0.8, 0.65, true)).filter(Boolean).length / 1000).toBeGreaterThanOrEqual(0.9); + }); +}); diff --git a/src/improve/monitor.ts b/src/improve/monitor.ts new file mode 100644 index 00000000..734ab463 --- /dev/null +++ b/src/improve/monitor.ts @@ -0,0 +1,183 @@ +import { mkdirSync, readFileSync } 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 type { ImproveSettings } from './settings'; + +/** What the monitor decided about the current revision. */ +export interface MonitorVerdict { + /** `idle` with nothing to watch, `pending` until there are enough runs, then `kept`, `kept-unverified`, `regressed` (reported, not rolled back) or `rolled-back`. */ + state: 'idle' | 'pending' | 'kept' | 'kept-unverified' | 'regressed' | 'rolled-back'; + /** Why, for `idle`, `kept-unverified`, `regressed` and `rolled-back`. */ + reason?: string; + /** Finished runs under the current revision so far, for `pending` and `kept-unverified`. */ + runs?: number; + /** Finished runs of the revision before it, for `pending` and `kept-unverified`. */ + baselineRuns?: number; + /** The runs before the apply that were compared. */ + before?: WindowStat; + /** The runs after the apply that were compared. */ + after?: WindowStat; + /** The revision restored, for `rolled-back`. */ + to?: string; +} + +/** The comparison of the runs before and after an apply, without side effects. */ +export type MonitorJudgement = { verdict: 'wait' } | { verdict: 'kept'; before: WindowStat; after: WindowStat } | { verdict: 'regressed'; before: WindowStat; after: WindowStat; reason: string }; + +type MonitorSettings = ImproveSettings['monitor']; +type CheckSettings = Pick; + +const LOOKS = 3; +const LOOK_Z = 2.1; +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']; + +const percent = (value: number): string => `${Math.round(value * 100)}%`; +const binary = (points: readonly ScorePoint[]): boolean => points.every((point) => point.value === 0 || point.value === 1); +const passes = (points: readonly ScorePoint[]): number[] => points.map((point) => (point.value >= 0.5 ? 1 : 0)); + +function regression(before: readonly ScorePoint[], after: readonly ScorePoint[], b: WindowStat, a: WindowStat, maxDrop: number): string | undefined { + if (b.passRate - a.passRate > maxDrop && significantDrop(passes(before), passes(after), LOOK_Z)) return `The pass rate fell from ${percent(b.passRate)} to ${percent(a.passRate)} over ${a.runs} runs.`; + if (binary(before) && binary(after)) return undefined; + const gate = trajectoryGate([...before, ...after], { window: GATE_WINDOW, drop: maxDrop, z: LOOK_Z }); + if (b.mean - a.mean > maxDrop && gate.state === 'regression') return `The mean score fell from ${b.mean.toFixed(2)} to ${gate.mean.toFixed(2)} over the newest ${GATE_WINDOW} runs.`; + return undefined; +} + +/** Compares the first runs after an apply with the last runs before it, in up to three looks of `window` runs, at a per-look z of 2.1 that keeps the overall false rollback rate under one-sided 5%. */ +export function judgeApplied(before: readonly ScorePoint[], after: readonly ScorePoint[], settings: MonitorSettings): MonitorJudgement { + const minRuns = Math.max(settings.minRuns, settings.window); + if (after.length < minRuns || before.length < minRuns) return { verdict: 'wait' }; + const looks = Math.min(LOOKS, Math.floor(after.length / settings.window)); + const size = looks * settings.window; + const later = after.slice(0, size); + const earlier = before.slice(-size); + const b = windowSummary(earlier); + const a = windowSummary(later); + const reason = regression(earlier, later, b, a, settings.maxDrop); + if (reason !== undefined) return { verdict: 'regressed', before: b, after: a, reason }; + const dropped = b.passRate - a.passRate > settings.maxDrop || b.mean - a.mean > settings.maxDrop; + return dropped && looks < LOOKS ? { verdict: 'wait' } : { verdict: 'kept', before: b, after: a }; +} + +interface MonitorState { + cooldownUntil?: string; + monitor?: { revision: string; applySeq: number; at: string; verdict: MonitorVerdict }; + [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; +} + +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`); + }); +} + +function lastApply(dir: string, current: string): AuditEntry | string { + const change = readAudit(dir) + .filter((entry) => entry.action === 'apply' || entry.action === 'rollback') + .at(-1); + if (change?.action !== 'apply' || change.from === undefined) return 'No applied revision to watch.'; + if (change.to !== current) return 'The directory changed by hand after the last apply, so the monitor leaves it alone.'; + return change; +} + +function seriesAround(db: SqlDatabase, apply: AuditEntry, settings: MonitorSettings, agent: string): { before: ScorePoint[]; after: ScorePoint[] } { + const appliedAt = Date.parse(apply.at); + const before = finishedRuns(db, { agentId: agent, revision: apply.from, endedBefore: appliedAt, byEnd: true, limit: LOOKS * settings.window }).reverse().map(pointOf); + const after = finishedRuns(db, { agentId: agent, revision: apply.to, since: appliedAt, byEnd: true, limit: SERIES_LIMIT }).reverse().map(pointOf); + return { before, after }; +} + +const score = (before: WindowStat, after: WindowStat) => ({ score: { before: before.passRate, after: after.passRate, runs: after.runs } }); + +async function settle(dir: string, apply: AuditEntry, verdict: MonitorVerdict, now: number): Promise { + await writeState(dir, { monitor: { revision: apply.to as string, applySeq: apply.seq, at: new Date(now).toISOString(), verdict } }); + return verdict; +} + +async function rollBack(dir: string, apply: AuditEntry, judged: Extract, now: number, cooldownHours: number): Promise { + await writeState(dir, { cooldownUntil: new Date(now + cooldownHours * HOUR_MS).toISOString() }); + let to: string; + try { + to = (await rollbackHarness(dir, apply.from as string, { actor: 'monitor', reason: judged.reason, expectedFrom: apply.to })).to; + } catch (error) { + await appendAudit(dir, { actor: 'monitor', action: 'guard-fail', from: apply.from, to: apply.to, reason: `${judged.reason} The rollback was refused: ${error instanceof Error ? error.message : String(error)}`, evidence: score(judged.before, judged.after) }); + return settle(dir, apply, { state: 'regressed', before: judged.before, after: judged.after, reason: `${judged.reason} The rollback was refused.` }, now); + } + await appendAudit(dir, { actor: 'monitor', action: 'retire', from: apply.to, to, reason: judged.reason, evidence: score(judged.before, judged.after) }); + return settle(dir, apply, { state: 'rolled-back', to, before: judged.before, after: judged.after, reason: judged.reason }, now); +} + +async function decide(dir: string, apply: AuditEntry, judged: Exclude, settings: CheckSettings, now: number): Promise { + if (judged.verdict === 'kept') { + await appendAudit(dir, { actor: 'monitor', action: 'guard-pass', from: apply.from, to: apply.to, reason: `Kept: ${judged.after.runs} runs held the pass rate at ${percent(judged.after.passRate)} against ${percent(judged.before.passRate)}.`, evidence: score(judged.before, judged.after) }); + return settle(dir, apply, { state: 'kept', before: judged.before, after: judged.after }, now); + } + if (settings.monitor.autoRollback) return rollBack(dir, apply, judged, now, settings.cooldownHours); + await appendAudit(dir, { actor: 'monitor', action: 'guard-fail', from: apply.from, to: apply.to, reason: `${judged.reason} Automatic rollback is off.`, evidence: score(judged.before, judged.after) }); + return settle(dir, apply, { state: 'regressed', before: judged.before, after: judged.after, reason: judged.reason }, now); +} + +function unverifiedReason(after: number, before: number, settings: MonitorSettings): string { + const minRuns = Math.max(settings.minRuns, settings.window); + if (after < minRuns) return `Kept unverified: ${after} runs after ${settings.giveUpDays} days, fewer than the ${minRuns} the monitor needs.`; + if (before < minRuns) return `Kept unverified: the revision before it has only ${before} runs to compare with.`; + return `Kept unverified: after ${settings.giveUpDays} days the ${after} runs neither held nor clearly fell.`; +} + +function agentIdOf(db: SqlDatabase, agent: string): string { + if (agent.trim() === '') throw new ConfigurationError('checkApplied needs the agent whose runs it judges; got an empty name.', 'agent'); + if (db.prepare('SELECT 1 AS hit FROM agents WHERE agent_id = ?').get(agent) !== undefined) return agent; + const named = db.prepare('SELECT agent_id FROM agents WHERE name = ?').all(agent).map((row) => String(row.agent_id)); + if (named.length > 1) throw new ConfigurationError(`'${agent}' names ${named.length} agents (${named.join(', ')}); pass the agent_id.`, 'agent'); + return named[0] ?? agent; +} + +async function check(db: SqlDatabase, dir: string, settings: CheckSettings, now: number, name: string): Promise { + 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; + 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); + if (judged.verdict !== 'wait') return decide(dir, apply, judged, settings, now); + if (now - Date.parse(apply.at) < settings.monitor.giveUpDays * DAY_MS) return { state: 'pending', runs: after.length, baselineRuns: before.length }; + const reason = unverifiedReason(after.length, before.length, settings.monitor); + await appendAudit(dir, { actor: 'monitor', action: 'unverified', from: apply.from, to: apply.to, reason }); + return settle(dir, apply, { state: 'kept-unverified', runs: after.length, baselineRuns: before.length, reason }, now); +} + +/** Compares runs of `agent` (its id or registered name) under the current revision with its runs before the apply, and rolls back on a regression. */ +export async function checkApplied(db: SqlDatabase, dir: string, settings: Pick, options: { agent: string; now?: number }): Promise { + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + return withFileLock(path.join(dir, MONITOR_LOCK), () => check(db, dir, settings, options.now ?? Date.now(), options.agent)); +} diff --git a/src/observe/continuousScore.ts b/src/observe/continuousScore.ts index 5d237f8c..e20f6a4c 100644 --- a/src/observe/continuousScore.ts +++ b/src/observe/continuousScore.ts @@ -1,5 +1,5 @@ import type { SqlDatabase } from '../storage/sqlite/driver'; -import { finishedRuns } from './store'; +import { finishedRuns, type RevisionRun } from './store'; /** One run's value in the series: the mean of its online scores, or 0 when it failed and 1 when it passed with no score. */ export interface ScorePoint { @@ -41,8 +41,11 @@ export interface TrajectoryOptions { /** A window's summary; a value of 0.5 or more is a pass. */ export interface WindowStat { + /** How many runs the window holds. */ runs: number; + /** The mean point value. */ mean: number; + /** The share of runs that passed. */ passRate: number; /** pass^3, the chance that three independent runs all pass, estimated as C(passes, 3) / C(runs, 3); absent below three runs. */ pass3?: number; @@ -57,12 +60,17 @@ const PASS = 0.5; const EPSILON = 1e-9; const SERIES_DEFAULT_LIMIT = 1000; -/** The series for one agent (its exact agent id), oldest first. */ -export function scoreSeries(db: SqlDatabase, filter: { agent: string; revision?: string; since?: number; limit?: number }): ScorePoint[] { +/** The series for one agent (its exact agent id), or for every agent when it is left out, oldest first. */ +export function scoreSeries(db: SqlDatabase, filter: { agent?: string; revision?: string; since?: number; limit?: number }): ScorePoint[] { const { agent, ...rest } = filter; - return finishedRuns(db, { ...rest, agentId: agent, limit: filter.limit ?? SERIES_DEFAULT_LIMIT }) + return finishedRuns(db, { ...rest, ...(agent !== undefined && { agentId: agent }), limit: filter.limit ?? SERIES_DEFAULT_LIMIT }) .reverse() - .map((run) => ({ traceId: run.traceId, at: run.startedAt, ...(run.revision !== undefined && { revision: run.revision }), value: run.score ?? (run.failed ? 0 : 1) })); + .map(pointOf); +} + +/** A finished run as a point of the series. */ +export function pointOf(run: RevisionRun & { revision?: string }): ScorePoint { + return { traceId: run.traceId, at: run.startedAt, ...(run.revision !== undefined && { revision: run.revision }), value: run.score ?? (run.failed ? 0 : 1) }; } const meanOf = (values: readonly number[]): number => (values.length === 0 ? 0 : values.reduce((sum, value) => sum + value, 0) / values.length); diff --git a/src/observe/store.ts b/src/observe/store.ts index 92f3a6da..f5dcb96c 100644 --- a/src/observe/store.ts +++ b/src/observe/store.ts @@ -1280,23 +1280,26 @@ const ROOT_REVISION = "(CASE WHEN json_valid(o.attrs) THEN json_extract(o.attrs, const ROOTS_WHERE = "o.kind = 'span' AND o.parent_obs_id IS NULL AND o.end IS NOT NULL AND o.start >= ? "; const REVISION_WHERE = `AND (? IS NULL OR (instr(o.attrs, ?) > 0 AND ${ROOT_REVISION} = ?)) ORDER BY o.start DESC LIMIT ?`; -const ROOTS_BY_AGENT_ID = `SELECT o.obs_id, o.trace_id, o.start FROM observations o WHERE ${ROOTS_WHERE}AND o.agent_id = ? ${REVISION_WHERE}`; +const ROOTS_BY_AGENT_ID = `SELECT o.obs_id, o.trace_id, o.start FROM observations o WHERE ${ROOTS_WHERE}AND o.agent_id = ? AND o.end < ? ${REVISION_WHERE}`; +const ROOTS_BY_AGENT_ID_ENDED = ROOTS_BY_AGENT_ID.replace('ORDER BY o.start DESC', 'ORDER BY o.end DESC'); const ROOTS_BY_AGENT = `SELECT o.obs_id, o.trace_id, o.start FROM observations o LEFT JOIN agents a ON a.agent_id = o.agent_id WHERE ${ROOTS_WHERE}` + - `AND (? = '' OR o.agent_id = ? OR a.name = ?) ${REVISION_WHERE}`; + `AND (? = '' OR o.agent_id = ? OR a.name = ?) AND o.end < ? ${REVISION_WHERE}`; +const ROOTS_BY_AGENT_ENDED = ROOTS_BY_AGENT.replace('ORDER BY o.start DESC', 'ORDER BY o.end DESC'); const ROOT_REVISIONS = `SELECT o.obs_id, ${ROOT_REVISION} AS revision FROM observations o WHERE o.obs_id IN (SELECT value FROM json_each(?))`; const ONLINE_MEANS = "SELECT trace_id, AVG(value) AS mean FROM scores WHERE trace_id IN (SELECT value FROM json_each(?)) AND source = 'online' AND value IS NOT NULL GROUP BY trace_id"; -type RunFilter = { revision?: string; agent?: string; agentId?: string; since?: number; limit: number }; +type RunFilter = { revision?: string; agent?: string; agentId?: string; since?: number; endedBefore?: number; byEnd?: boolean; limit: number }; function rootRows(db: SqlDatabase, filter: RunFilter): SqlRow[] { const revision = filter.revision ?? null; const tail = [revision, revision, revision, filter.limit]; - if (filter.agentId !== undefined) return db.prepare(ROOTS_BY_AGENT_ID).all(filter.since ?? 0, filter.agentId, ...tail); + const endedBefore = filter.endedBefore ?? Number.MAX_SAFE_INTEGER; + if (filter.agentId !== undefined) return db.prepare(filter.byEnd === true ? ROOTS_BY_AGENT_ID_ENDED : ROOTS_BY_AGENT_ID).all(filter.since ?? 0, filter.agentId, endedBefore, ...tail); const agent = filter.agent ?? ''; - return db.prepare(ROOTS_BY_AGENT).all(filter.since ?? 0, agent, agent, agent, ...tail); + return db.prepare(filter.byEnd === true ? ROOTS_BY_AGENT_ENDED : ROOTS_BY_AGENT).all(filter.since ?? 0, agent, agent, agent, endedBefore, ...tail); } const ROOTS_AFTER = `SELECT COUNT(*) AS n FROM observations o WHERE ${ROOTS_WHERE.replace('o.start >= ?', 'o.start > ?')}AND o.agent_id = ?`;