feat(improve): post-apply monitor and automatic rollback

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<ImproveSettings, 'monitor' | 'cooldownHours'>, options: {
+    agent: string;
+    now?: number;
+}): Promise<MonitorVerdict>;
+
 // @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)
 
 ```

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<string, AuditEntry['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);
+  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;
 }

src/improve/audit.ts

@@ -13,7 +13,7 @@ export interface AuditEntry {
   /** Who or what made the change: a loop stage, or `person:<name>`. */
   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. */

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';

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<ImproveSettings, 'monitor' | 'cooldownHours'>;
+
+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<MonitorState>): Promise<void> {
+  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<MonitorVerdict> {
+  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<MonitorJudgement, { verdict: 'regressed' }>, now: number, cooldownHours: number): Promise<MonitorVerdict> {
+  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<MonitorJudgement, { verdict: 'wait' }>, settings: CheckSettings, now: number): Promise<MonitorVerdict> {
+  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<MonitorVerdict> {
+  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<ImproveSettings, 'monitor' | 'cooldownHours'>, options: { agent: string; now?: number }): Promise<MonitorVerdict> {
+  mkdirSync(path.join(dir, '.lousho'), { recursive: true });
+  return withFileLock(path.join(dir, MONITOR_LOCK), () => check(db, dir, settings, options.now ?? Date.now(), options.agent));
+}

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);