feat(improve): start a cycle on the observe daemon's score regression

api/improve.api.md

@@ -2544,6 +2544,7 @@ export interface WatchOptions extends Omit<HealOptions, 'trigger' | 'maxTokens'
     log?: (line: string) => void;
     onCycle?: (cycle: HealCycle) => void;
     onError?: (error: unknown) => void;
+    registry?: string;
     reloadSettings?: () => ImproveSettings;
     store?: CheckpointStore;
 }

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

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

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,

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

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<HealOptions, 'trigger' | 'maxTokens' | 'agent'> {
@@ -41,6 +43,8 @@ export interface WatchOptions extends Omit<HealOptions, 'trigger' | 'maxTokens'
   alerter?: Alerter;
   /** Writes one line per `improve.*` event (default stderr). */
   log?: (line: string) => 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<typeof failureKey>): 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<WatchTick> {
   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);
     },
   };
 }

src/improve/watchSteps.ts

@@ -37,6 +37,7 @@ export interface WatchMemory {
   spent?: number;
   harvested?: number;
   guardNotified?: string;
+  scoreAt?: number;
 }
 
 export interface WatchState {

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<WatcherEntry> | 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<void> {
+  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<WatcherEntry, 'since' | 'pid'>): Promise<void> {
+  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<void> {
+  await update(file, (entries) => entries.filter((item) => item.owner !== owner));
+}

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)
+  );
+  `,
 ];
 
 /**

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

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