diff --git a/api/improve.api.md b/api/improve.api.md index 045cf428..4c912917 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' | 'unverified' | 'pause' | 'resume' | 'canary-start' | 'canary-end'; + action: 'propose' | 'apply' | 'reject' | 'stale' | 'rollback' | 'retire' | 'guard-pass' | 'guard-fail' | 'unverified' | 'pause' | 'resume' | 'canary-start' | 'canary-end' | 'reflect-fail'; actor: 'gate' | 'canary' | 'monitor' | 'guard' | 'scheduler' | `person:${string}`; at: string; cycle?: string; @@ -303,14 +303,12 @@ export function casesFromFailures(db: SqlDatabase, filter?: { since?: number; agent?: string; minScore?: number; + traceIds?: readonly string[]; + recurring?: boolean; }): CapturedCase[]; // @public -export function casesFromRecoveries(db: SqlDatabase, filter?: { - since?: number; - agent?: string; - minScore?: number; -}): CapturedCase[]; +export function casesFromRecoveries(db: SqlDatabase, filter?: Parameters[1]): CapturedCase[]; // @public export interface CaseVerdict { @@ -590,6 +588,54 @@ interface HealActionEvent extends AgentEventBase<'heal.action'> { toolName: string; } +// @public +export interface HealCycle { + approval: ImproveSettings['approval']; + canary?: CanaryReplayResult; + endedAt?: string; + errors?: string[]; + from?: string; + guard?: GuardCorpusResult; + id: string; + outcome: 'idle' | 'paused' | 'busy' | 'budget' | 'cooldown' | Exclude | 'canary-failed' | 'applied' | 'pending' | 'guard-rolled-back' | 'stale' | 'refused' | 'interrupted'; + pending?: string; + reason: string; + result?: ImproveResult; + rolledBackTo?: string; + startedAt: string; + to?: string; + tokens: number; + trigger: 'failures' | 'score' | 'manual'; +} + +// @public +export function healOnce(options: HealOptions): Promise; + +// @public +export interface HealOptions { + agent?: string; + check: CaseCheck; + db: SqlDatabase; + dir: string; + editor: ImproveOptions['editor']; + editorHooks?: readonly AgentHook[]; + editorTimeoutMs?: number; + maxTokens?: number; + protect?: readonly string[]; + reflect?: boolean; + runCase: (harness: Harness, input: string) => Promise<{ + answer: string; + transcript: string; + tokens: number; + ended: string; + }>; + settings: ImproveSettings; + signal?: AbortSignal; + since?: number; + trigger?: HealCycle['trigger']; + withRecoveries?: boolean; +} + // @public interface HookContext { agentId?: string; @@ -675,6 +721,7 @@ export interface ImproveOptions { model: string; }; editorHooks?: readonly AgentHook[]; + editorTimeoutMs?: number; harness: Harness; holdout?: readonly string[]; library?: Readonly>; diff --git a/docs/cli.md b/docs/cli.md index a88e419a..bc479bb1 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -18,7 +18,7 @@ Run it with `npx lousho ` inside a project that has the SDK installed. | `lousho traces [id]` | List the recent runs saved by `fileTraceExporter()` (duration, model and tool calls, tokens, cost), or print one run as a span tree. | [Observability](./observability.md#local-traces) | | `lousho build --target= --agent=` | Build a deployable Node server, Docker image or Cloudflare Worker. | [Deployment](./deployment.md) | | `lousho studio` | Launch Agent Forge, the visual dashboard, on one local port. | [Agent Forge](./agent-forge.md) | -| `lousho improve --check ` | Experimental. Propose edits to an agent's instructions and skills from failed runs in your local observe database. Prints a review card and applies nothing. | [Below](#lousho-improve) | +| `lousho improve --check ` | Experimental. Run one self-healing cycle on an agent directory: propose edits to its instructions and skills from failed runs in your local observe database, gate them, replay them on fresh runs, and apply (auto mode) or queue (manual mode) one that passes. `--dry-run` prints the review card and applies nothing. | [Below](#lousho-improve) | Every command that takes a `` reads an agent spec file (`.yaml`, `.yml` or `.json`, see [Configuration](./configuration.md#agent-spec-files-agentspec)). @@ -375,9 +375,17 @@ export default (input, answer) => !answer.includes('I cannot help'); It takes the failed runs in your local observe database (default `~/.lousho/observe.db`, or `--db`), replays each recorded input through the agent, scores each answer with your check, and prints the review card from -`formatImproveResult()`. It never changes the agent: no file in the agent -directory is written, and there is no flag to apply a proposal. Review the card -and apply what you accept by hand. +`formatImproveResult()`. On an agent directory it then runs one self-healing +cycle with `healOnce()`: only failed runs of the current harness revision count, +a proposal that passes the gate is replayed on fresh runs (the canary), and in +auto mode (the default) it is written to `instructions.md` and +`skills//SKILL.md`, after which the corpus guard replays the protected +cases and rolls back on a regression. In manual mode it is queued for +`lousho improve approve`. Every step is audited, and `improve.json`'s daily caps, +the cooldown after a rollback and the `.lousho/improve.stop` kill switch apply. +`--dry-run` keeps the old behaviour: every failed run counts, the card is +printed, and nothing is written. An agent module, or an agent whose instructions +don't live in `instructions.md`, always runs dry. - `` is an agent directory or a `.ts`/`.js` agent module whose instructions are plain text. Editors may change only the instructions and diff --git a/docs/improve.md b/docs/improve.md index d1b5b680..59104273 100644 --- a/docs/improve.md +++ b/docs/improve.md @@ -346,7 +346,10 @@ npx lousho improve ./my-agent --check ./check.mjs --since 7d --out proposal.md It reads cases only from your local observe database, needs a module whose default export is your `CaseCheck` (return `failure` to name the check a failed answer missed), simulates tools other than loading skills -unless `--read-only` names them, and prints the review card. It applies nothing. +unless `--read-only` names them, and prints the review card. On an agent +directory it runs one self-healing cycle (`healOnce()`): in auto mode a proposal +that passes the gate and a replay canary is applied and guarded, and in manual +mode it is queued for approval. `--dry-run` applies nothing. `--max-tokens ` sets `maxTokens`; it has no default, so without it there is no cap. It is checked after each run or editor call, so the last one can go over, and a run that hits it ends `budget-exceeded`. diff --git a/llms-full.txt b/llms-full.txt index 8341fb6b..e74e27d9 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -5518,7 +5518,7 @@ Run it with `npx lousho ` inside a project that has the SDK installed. | `lousho traces [id]` | List the recent runs saved by `fileTraceExporter()` (duration, model and tool calls, tokens, cost), or print one run as a span tree. | [Observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md#local-traces) | | `lousho build --target= --agent=` | Build a deployable Node server, Docker image or Cloudflare Worker. | [Deployment](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md) | | `lousho studio` | Launch Agent Forge, the visual dashboard, on one local port. | [Agent Forge](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/agent-forge.md) | -| `lousho improve --check ` | Experimental. Propose edits to an agent's instructions and skills from failed runs in your local observe database. Prints a review card and applies nothing. | [Below](#lousho-improve) | +| `lousho improve --check ` | Experimental. Run one self-healing cycle on an agent directory: propose edits to its instructions and skills from failed runs in your local observe database, gate them, replay them on fresh runs, and apply (auto mode) or queue (manual mode) one that passes. `--dry-run` prints the review card and applies nothing. | [Below](#lousho-improve) | Every command that takes a `` reads an agent spec file (`.yaml`, `.yml` or `.json`, see [Configuration](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/configuration.md#agent-spec-files-agentspec)). @@ -5875,9 +5875,17 @@ export default (input, answer) => !answer.includes('I cannot help'); It takes the failed runs in your local observe database (default `~/.lousho/observe.db`, or `--db`), replays each recorded input through the agent, scores each answer with your check, and prints the review card from -`formatImproveResult()`. It never changes the agent: no file in the agent -directory is written, and there is no flag to apply a proposal. Review the card -and apply what you accept by hand. +`formatImproveResult()`. On an agent directory it then runs one self-healing +cycle with `healOnce()`: only failed runs of the current harness revision count, +a proposal that passes the gate is replayed on fresh runs (the canary), and in +auto mode (the default) it is written to `instructions.md` and +`skills//SKILL.md`, after which the corpus guard replays the protected +cases and rolls back on a regression. In manual mode it is queued for +`lousho improve approve`. Every step is audited, and `improve.json`'s daily caps, +the cooldown after a rollback and the `.lousho/improve.stop` kill switch apply. +`--dry-run` keeps the old behaviour: every failed run counts, the card is +printed, and nothing is written. An agent module, or an agent whose instructions +don't live in `instructions.md`, always runs dry. - `` is an agent directory or a `.ts`/`.js` agent module whose instructions are plain text. Editors may change only the instructions and @@ -13221,7 +13229,10 @@ npx lousho improve ./my-agent --check ./check.mjs --since 7d --out proposal.md It reads cases only from your local observe database, needs a module whose default export is your `CaseCheck` (return `failure` to name the check a failed answer missed), simulates tools other than loading skills -unless `--read-only` names them, and prints the review card. It applies nothing. +unless `--read-only` names them, and prints the review card. On an agent +directory it runs one self-healing cycle (`healOnce()`): in auto mode a proposal +that passes the gate and a replay canary is applied and guarded, and in manual +mode it is queued for approval. `--dry-run` applies nothing. `--max-tokens ` sets `maxTokens`; it has no default, so without it there is no cap. It is checked after each run or editor call, so the last one can go over, and a run that hits it ends `budget-exceeded`. diff --git a/src/cli/improve.test.ts b/src/cli/improve.test.ts index 30591b17..e982f662 100644 --- a/src/cli/improve.test.ts +++ b/src/cli/improve.test.ts @@ -9,6 +9,9 @@ import type { IngestEvent } from '../observe/envelope'; import { ingestBatch, openDb } from '../observe/store'; import { mockModel, type MockTurn } from '../testing'; import { defineTool } from '../tools/defineTool'; +import { readAudit } from '../improve/audit'; +import { listPending } from '../improve/pending'; +import { currentRevision } from '../improve/revisions'; import { parseImproveArgs, runImprove, type ImproveIo } from './improve'; const T0 = Date.now() - 60_000; @@ -80,6 +83,10 @@ function snapshot(dir: string): Record { const fixedWhenForbidden = (request: { messages: readonly unknown[] }) => ({ text: JSON.stringify(request.messages).includes('forbidden') ? 'ok, that is forbidden' : 'here is the doc' }); async function improve(args: string[], over: Partial = {}) { + return run(args.includes('--help') || args[0] === undefined ? args : [...args, '--dry-run'], over); +} + +async function run(args: string[], over: Partial = {}) { const out = sink(); const err = sink(); const io: ImproveIo = { @@ -203,7 +210,8 @@ describe('lousho improve', () => { const { code, stdout } = await improve(['--help']); expect(code).toBe(0); expect(stdout).toContain('Experimental'); - expect(stdout).toContain('never changes'); + expect(stdout).toContain('--dry-run'); + expect(stdout).not.toContain('never changes'); expect(stdout).toContain('[--max-tokens n]'); expect(stdout).toMatch(/--max-tokens n {2,}stop once n tokens are spent/); expect(stdout).toContain('no cap'); @@ -541,3 +549,74 @@ describe('lousho improve', () => { expect(stderr).toMatch(/pass:p1 was not protected: it passed 0 of 6 scored runs/); }); }); + +describe('lousho improve, one self-healing cycle', () => { + const editorFor = (instructions: string) => + mockModel( + [ + (request) => { + if (request.messages.at(-1)?.role === 'tool') return 'Traced to a failed run.'; + return { toolCalls: [{ name: 'write_file', args: { path: 'harness/instructions.md', content: instructions } }] }; + }, + ], + { onExhausted: 'repeat-last' } + ); + + function stamped(trace: string): IngestEvent[] { + const revision = currentRevision(agentDir); + return failedTrace(trace, JSON.stringify([{ role: 'user', parts: [{ type: 'text', content: 'can I read the secret doc?' }] }])).map((event) => + event.type === 'observation' && event.data.kind === 'span' ? { ...event, data: { ...event.data, attrs: { ...event.data.attrs, 'lousho.harness.revision': revision } } } : event + ); + } + + beforeEach(() => { + const db = openDb(dbPath); + ingestBatch(db, [...stamped('s1'), ...stamped('s2')]); + db.close(); + }); + + const instructions = () => fs.readFileSync(path.join(agentDir, 'instructions.md'), 'utf8').trim(); + + it('auto: applies a fix that passes the gate and the canary, and audits it', async () => { + const { code, stdout } = await run([agentDir, '--check', checkPath, '--db', dbPath], { editor: editorFor(FIX) }); + expect(code).toBe(0); + expect(stdout).toMatch(/lousho improve: applied: Applied [0-9a-f]{12}/); + expect(instructions()).toBe(FIX); + expect(readAudit(agentDir).map((entry) => entry.action)).toEqual(['canary-end', 'apply']); + }); + + it('manual: queues the fix for a person and writes nothing to the agent', async () => { + const { code, stdout } = await run([agentDir, '--check', checkPath, '--db', dbPath, '--approval', 'manual'], { editor: editorFor(FIX) }); + expect(code).toBe(0); + expect(stdout).toMatch(/lousho improve: pending: Queued for approval as [0-9a-f]{12}/); + expect(instructions()).toBe('You answer refund questions.'); + expect(listPending(agentDir)).toHaveLength(1); + }); + + it('--dry-run writes nothing', async () => { + const before = snapshot(root); + const { code, stdout } = await run([agentDir, '--check', checkPath, '--db', dbPath, '--dry-run'], { editor: editorFor(FIX) }); + expect(code).toBe(0); + expect(stdout).toContain('# Improve result: proposal'); + const after = snapshot(root); + expect(Object.keys(after).filter((file) => after[file] !== before[file] && !file.startsWith('observe.db'))).toEqual([]); + }); + + it('a module target is forced to dry run', async () => { + const file = path.join(root, 'agent.mjs'); + fs.writeFileSync(file, "export default { name: 'support', instructions: 'You answer refund questions.' };\n"); + const { code, stderr } = await run([file, '--check', checkPath, '--db', dbPath], { editor: editorFor(FIX) }); + expect(code).toBe(0); + expect(stderr).toMatch(/this is a dry run, because .*agent module/); + expect(fs.existsSync(path.join(root, '.lousho'))).toBe(false); + }); + + it('an agent whose instructions are not in instructions.md is forced to dry run', async () => { + fs.mkdirSync(path.join(agentDir, 'instructions')); + fs.writeFileSync(path.join(agentDir, 'instructions', 'claude.md'), 'You answer refund questions for Claude.'); + const { code, stderr } = await run([agentDir, '--check', checkPath, '--db', dbPath], { editor: editorFor(FIX) }); + expect(code).toBe(0); + expect(stderr).toMatch(/this is a dry run, because .*model-specific instructions/); + expect(instructions()).toBe('You answer refund questions.'); + }); +}); diff --git a/src/cli/improve.ts b/src/cli/improve.ts index 4544b174..e1da3dc6 100644 --- a/src/cli/improve.ts +++ b/src/cli/improve.ts @@ -23,7 +23,11 @@ import { type RecoveryEvidence, } from '../improve'; import { scoreCase } from '../improve/cases'; -import { recordedTurns } from '../improve/recoveries'; +import { agentInputOf } from '../improve/recoveries'; +import type { HealCycle } from '../improve/cycles'; +import { healOnce } from '../improve/heal'; +import { writtenHash } from '../improve/harnessWriter'; +import { readAgentHarness } from '../improve/revisions'; import { loadImproveSettings, type ImproveSettings } from '../improve/settings'; import { openDb } from '../observe/store'; import type { AgentInput, LLMProvider } from '../providers'; @@ -35,13 +39,22 @@ import { isImproveAdmin, runImproveAdmin } from './improveAdmin'; import { dbPathOf, resolveAgent, sinceMs } from './observe'; const USAGE = - 'Usage: lousho improve --check [--model provider/model] [--editor provider/model] [--db path] [--since d|h|m] [--agent X] [--protect traceId]... [--read-only tool]... [--max-tokens n] [--out file] [--with-recoveries] [--approval auto|manual]\n lousho improve revisions|rollback|audit|pending|approve|reject ...'; + 'Usage: lousho improve --check [--model provider/model] [--editor provider/model] [--db path] [--since d|h|m] [--agent X] [--protect traceId]... [--read-only tool]... [--max-tokens n] [--out file] [--with-recoveries] [--approval auto|manual] [--dry-run]\n lousho improve revisions|rollback|audit|pending|approve|reject ...'; const HELP = `${USAGE} -Experimental. Proposes edits to an agent's instructions and skills from failed runs in your own -local observe database, using improveAgent(). It never changes the agent: it prints a review card, -and --out writes the proposal to a file for a person to review and apply by hand. +Experimental. Runs one self-healing cycle on an agent directory. It captures failed runs of the +agent's current harness revision from your own local observe database, has editors propose edits +to its instructions and skills with improveAgent(), gates them, and replays the winner on fresh +runs (the canary). In auto mode (the default) a proposal that passes is written to instructions.md +and skills//SKILL.md, then the corpus guard replays the protected cases and rolls back on a +regression. In manual mode it is queued for 'lousho improve approve'. Every step is audited in +.lousho/audit.jsonl, and improve.json's daily caps, the cooldown after a rollback and the +.lousho/improve.stop kill switch apply. In auto mode only a failure seen in two or more runs +becomes a case (improve.json trigger.requireRecurring). Ctrl-C stops the cycle between steps; an +apply and its guard always finish. --dry-run only prints the review card and writes nothing; it +reads every failed run, whatever its revision, and ignores the kill switch and the daily caps. +An agent module, or instructions that don't live in instructions.md, always run dry. Runs replay each case's recorded input through the agent and are not traced. Tools other than load_skill and read_skill_file are simulated and never run unless --read-only names them, so only @@ -69,8 +82,10 @@ Options: --out file also write the proposal to this new file: the review card, or the JSON result for a .json path --approval auto|manual the approval mode, over LOUSHO_IMPROVE_APPROVAL and improve.json - (default auto). Manual keeps an edit that quotes a failed check - and flags it on the card for the person who approves it + (default auto). Manual queues a passing proposal for a person, and + keeps an edit that quotes a failed check, flagged on the card + --dry-run print the review card and change nothing, from every failed run + whatever its revision, as before self-healing -h, --help show this help Revisions, rollback, the audit log and pending proposals (run one with --help for its flags): @@ -100,6 +115,7 @@ const SPEC: CommandSpec = { 'max-tokens': { type: 'string' }, 'with-recoveries': { type: 'boolean' }, approval: { type: 'string' }, + 'dry-run': { type: 'boolean' }, }, }; @@ -119,6 +135,8 @@ export interface ImproveArgs { withRecoveries?: boolean; /** The approval mode from `--approval`. */ approval?: ImproveSettings['approval']; + /** `--dry-run`: print the card and write nothing. */ + dryRun?: boolean; help?: boolean; } @@ -168,6 +186,7 @@ export function parseImproveArgs(args: string[]): ImproveArgs { out: stringValue(values.out), maxTokens: maxTokensOf(stringValue(values['max-tokens'])), withRecoveries: values['with-recoveries'] === true, + dryRun: values['dry-run'] === true, ...approvalOf(stringValue(values.approval)), }; } @@ -217,21 +236,8 @@ function assertOutUsable(out: string, target: DevTarget, config: CreateAgentConf if (guarded.some((dir) => inside(file, fs.realpathSync(dir)))) throw usageError(SPEC, `--out must be outside the agent's files; ${out} is not.`); } -function messageOf(input: string): AgentInput { - let parsed: unknown; - try { - parsed = JSON.parse(input); - } catch { - return input; - } - if (typeof parsed === 'string') return parsed; - const messages = Array.isArray(parsed) ? recordedTurns(parsed) : []; - if (messages.length === 0) return input; - return messages.length === 1 && messages[0].role === 'user' ? messages[0].content : messages; -} - function readable(captured: CapturedCase, decoded: Map): CapturedCase { - const message = messageOf(captured.input); + const message = agentInputOf(captured.input); const input = typeof message === 'string' ? message : captured.input; decoded.set(input, message); return { ...captured, input }; @@ -259,7 +265,7 @@ function runCaseWith(config: CreateAgentConfig, args: ImproveArgs, io: ImproveIo return async (harness: Harness, input: string) => { const agent = createAgent({ ...rest, instructions: harness.instructions, skills: skillsOf(harness, directories), hooks } as CreateAgentConfig); try { - return caseRun(await agent.send(decoded.get(input) ?? input)); + return caseRun(await agent.send(decoded.get(input) ?? agentInputOf(input))); } finally { await agent.close().catch(() => undefined); } @@ -282,10 +288,14 @@ function unscoredOnError(probe: ImproveOptions['probe'], errors: Set): I }; } -function readCases(args: ImproveArgs): { failures: CapturedCase[]; passes: CapturedCase[] } { +function openObserveDb(args: ImproveArgs) { const file = dbPathOf(args.db); if (!fs.existsSync(file)) throw new ConfigurationError(`lousho improve: no observe database at ${file}. Run your agent with LOUSHO_OBSERVE=1 and 'lousho observe start' first, or pass --db.`, 'db'); - const db = openDb(file); + return openDb(file); +} + +function readCases(args: ImproveArgs): { failures: CapturedCase[]; passes: CapturedCase[] } { + const db = openObserveDb(args); try { const agent = args.agent === undefined ? undefined : resolveAgent(SPEC, db, args.agent); const filter = { since: args.since, agent }; @@ -332,11 +342,7 @@ async function withoutObserve(run: () => Promise): Promise { } } -async function improve(args: ImproveArgs, io: ImproveIo): Promise { - const target = detectTarget(args.path, 'improve'); - const config = await loadConfig(target); - if (args.out !== undefined) assertOutUsable(args.out, target, config); - const harness = baseHarness(config); +async function proposeOnly(args: ImproveArgs, io: ImproveIo, target: DevTarget, config: CreateAgentConfig, harness: Harness): Promise { const approval = args.approval ?? loadImproveSettings(target.kind === 'dir' ? target.path : path.dirname(target.path)).approval; const quotedChecks = approval === 'manual' ? 'flag' : 'reject'; const editor = editorOf(args, config, io); @@ -379,6 +385,91 @@ async function improve(args: ImproveArgs, io: ImproveIo): Promise { return result.outcome === 'outage' ? 1 : 0; } +function dryRunReason(target: DevTarget, harness: Harness): string | undefined { + if (target.kind !== 'dir') return `${target.path} is an agent module, whose files can't be changed in place`; + try { + if (writtenHash(readAgentHarness(target.path)) !== writtenHash(harness)) return "the agent's instructions or skills don't all come from instructions.md and skills//SKILL.md"; + } catch (error) { + if (error instanceof ConfigurationError) return error.message; + throw error; + } + return undefined; +} + +const NEXT_STEP: Partial> = { + pending: "Approve it with 'lousho improve approve --by ', or reject it with 'lousho improve reject'.", + applied: "See 'lousho improve audit '; undo it with 'lousho improve rollback --by '.", + paused: "Remove .lousho/improve.stop, unset LOUSHO_IMPROVE=off or set enabled to true in improve.json to resume.", +}; + +function reportCycle(cycle: HealCycle, args: ImproveArgs, io: ImproveIo): void { + if (cycle.result) io.stdout.write(formatImproveResult(cycle.result)); + const next = NEXT_STEP[cycle.outcome]; + io.stdout.write(`lousho improve: ${cycle.outcome}: ${cycle.reason}${next ? ` ${next}` : ''}\n`); + for (const message of cycle.errors ?? []) io.stderr.write(`lousho improve: a run failed and was left unscored: ${message}\n`); + if (args.out === undefined) return; + if (!cycle.result) { + io.stderr.write(`lousho improve: no proposal was made, so ${args.out} was not written.\n`); + return; + } + const card = formatImproveResult(cycle.result); + try { + fs.writeFileSync(path.resolve(args.out), args.out.toLowerCase().endsWith('.json') ? `${JSON.stringify(cycle.result, null, 2)}\n` : card, { flag: 'wx' }); + io.stderr.write(`lousho improve: wrote the proposal to ${args.out}.\n`); + } catch (error) { + io.stderr.write(`lousho improve: could not write ${args.out}: ${error instanceof Error ? error.message : String(error)}\n`); + } +} + +async function healDir(args: ImproveArgs, io: ImproveIo, dir: string, config: CreateAgentConfig): Promise { + const settings = loadImproveSettings(dir, args.approval ? { approval: args.approval } : {}); + const editor = editorOf(args, config, io); + const check = await loadCheck(args.check); + const db = openObserveDb(args); + const controller = new AbortController(); + const interrupt = () => controller.abort(); + process.once('SIGINT', interrupt); + try { + const agent = args.agent === undefined ? undefined : resolveAgent(SPEC, db, args.agent); + const budget = args.maxTokens === undefined ? "within improve.json's token caps" : `stopping once ${args.maxTokens} tokens are spent (--max-tokens)`; + const write = settings.approval === 'auto' ? 'applies a proposal that passes the gate and the canary' : 'queues a proposal that passes the gate and the canary for a person'; + io.stderr.write(`lousho improve: running one self-healing cycle in ${settings.approval} mode, which ${write}; agent runs and editor calls are billed to your models, ${budget}. Pass --dry-run to change nothing.\n`); + const cycle = await withoutObserve(() => + healOnce({ + dir, + db, + check, + runCase: runCaseWith(config, args, io, new Map()), + editor, + settings, + trigger: 'manual', + protect: args.protect, + withRecoveries: args.withRecoveries === true, + ...(agent !== undefined && { agent }), + ...(args.since !== undefined && { since: args.since }), + ...(args.maxTokens !== undefined && { maxTokens: args.maxTokens }), + signal: controller.signal, + }) + ); + reportCycle(cycle, args, io); + return cycle.outcome === 'outage' ? 1 : 0; + } finally { + process.off('SIGINT', interrupt); + db.close(); + } +} + +async function improve(args: ImproveArgs, io: ImproveIo): Promise { + const target = detectTarget(args.path, 'improve'); + const config = await loadConfig(target); + if (args.out !== undefined) assertOutUsable(args.out, target, config); + const harness = baseHarness(config); + const forced = args.dryRun ? undefined : dryRunReason(target, harness); + if (forced) io.stderr.write(`lousho improve: this is a dry run, because ${forced}. Nothing will be applied.\n`); + if (args.dryRun || forced || target.kind !== 'dir') return proposeOnly(args, io, target, config, harness); + return healDir(args, io, target.path, config); +} + /** Runs `lousho improve` with the arguments after `improve`; resolves with the exit code. */ export async function runImprove(args: string[], io: ImproveIo = { stdout: process.stdout, stderr: process.stderr }): Promise { try { diff --git a/src/improve/audit.ts b/src/improve/audit.ts index 0ab9ca07..72a9c0de 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' | 'unverified' | 'pause' | 'resume' | 'canary-start' | 'canary-end'; + action: 'propose' | 'apply' | 'reject' | 'stale' | 'rollback' | 'retire' | 'guard-pass' | 'guard-fail' | 'unverified' | 'pause' | 'resume' | 'canary-start' | 'canary-end' | 'reflect-fail'; /** The revision before the change. */ from?: string; /** The revision after the change. */ diff --git a/src/improve/cases.test.ts b/src/improve/cases.test.ts index 5b2751cb..f0e2766c 100644 --- a/src/improve/cases.test.ts +++ b/src/improve/cases.test.ts @@ -82,6 +82,14 @@ describe('casesFromFailures', () => { expect(casesFromFailures(db, { agent: 'billing' })).toEqual([]); db.close(); }); + + it('keeps only the named traces, and only recurring groups when asked', () => { + const db = database(); + expect(casesFromFailures(db, { traceIds: ['t2'] }).map((c) => c.id)).toEqual(['timeout:t2']); + expect(casesFromFailures(db, { recurring: true }).map((c) => c.id).sort()).toEqual(['timeout:t1', 'timeout:t2']); + expect(casesFromFailures(db, { traceIds: ['t1'], recurring: true })).toEqual([]); + db.close(); + }); }); describe('protectedCasesFromPasses', () => { diff --git a/src/improve/cases.ts b/src/improve/cases.ts index 7667a748..f1375dbc 100644 --- a/src/improve/cases.ts +++ b/src/improve/cases.ts @@ -52,10 +52,23 @@ function captureCase(db: SqlDatabase, traceId: string, prefix: string, kind: Cap } /** Turns failed runs in your local trace database into cases to learn from, at most 5 traces per failure class, skipping traces with no recorded input. */ -export function casesFromFailures(db: SqlDatabase, filter: { since?: number; agent?: string; minScore?: number } = {}): CapturedCase[] { +export function casesFromFailures( + db: SqlDatabase, + filter: { + since?: number; + agent?: string; + minScore?: number; + /** Only these traces, such as the runs of one harness revision. */ + traceIds?: readonly string[]; + /** Only failure groups seen in two or more traces. */ + recurring?: boolean; + } = {} +): CapturedCase[] { const seen = new Set(); const cases: CapturedCase[] = []; - for (const group of failureReportFromDb(db, filter).groups) { + const { recurring, ...report } = filter; + for (const group of failureReportFromDb(db, report).groups) { + if (recurring && !group.recurring) continue; for (const traceId of group.traceIds) { if (seen.has(traceId)) continue; seen.add(traceId); diff --git a/src/improve/cycles.test.ts b/src/improve/cycles.test.ts new file mode 100644 index 00000000..f20aba77 --- /dev/null +++ b/src/improve/cycles.test.ts @@ -0,0 +1,75 @@ +import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { appendAudit } from './audit'; +import { cooldownEnds, newCycleId, pausedBy, readCycles, usageToday, writeCycle, type HealCycle } from './cycles'; + +let dir: string; + +const cycle = (id: string, over: Partial = {}): HealCycle => ({ id, trigger: 'failures', approval: 'auto', outcome: 'applied', reason: 'Applied.', startedAt: new Date().toISOString(), tokens: 100, ...over }); + +beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'lousho-cycles-')); +}); + +afterEach(() => { + rmSync(dir, { recursive: true, force: true }); +}); + +describe('cycle records', () => { + it('ids sort by start time', () => { + expect(newCycleId(Date.UTC(2026, 10, 9, 8, 30, 5))).toMatch(/^20261109T083005Z-[0-9a-f]{6}$/); + expect(newCycleId(Date.UTC(2026, 10, 9)) < newCycleId(Date.UTC(2026, 10, 10))).toBe(true); + }); + + it('round-trips newest first, replaces a cycle saved twice, and skips broken files', () => { + writeCycle(dir, cycle('20261109T080000Z-aaaaaa', { outcome: 'interrupted' })); + writeCycle(dir, cycle('20261109T090000Z-bbbbbb')); + writeCycle(dir, cycle('20261109T080000Z-aaaaaa', { outcome: 'pending' })); + writeFileSync(path.join(dir, '.lousho', 'cycles', '20261109T100000Z-cccccc.json'), '{ broken'); + writeFileSync(path.join(dir, '.lousho', 'cycles', 'notes.json'), '{}'); + expect(readCycles(dir).map((item) => [item.id, item.outcome])).toEqual([ + ['20261109T090000Z-bbbbbb', 'applied'], + ['20261109T080000Z-aaaaaa', 'pending'], + ]); + expect(() => writeCycle(dir, cycle('../escape'))).toThrow(/not a cycle id/); + }); + + it('refuses a symlinked cycles folder', () => { + const elsewhere = mkdtempSync(path.join(tmpdir(), 'lousho-elsewhere-')); + mkdirSync(path.join(dir, '.lousho')); + symlinkSync(elsewhere, path.join(dir, '.lousho', 'cycles')); + expect(() => readCycles(dir)).toThrow(/symlink/); + rmSync(elsewhere, { recursive: true, force: true }); + }); +}); + +describe('budgets, cooldown and the kill switch', () => { + it("counts today's cycles, tokens and automatic applies only", async () => { + const now = Date.now(); + writeCycle(dir, cycle(newCycleId(now))); + writeCycle(dir, cycle(newCycleId(now - 2 * 86_400_000), { startedAt: new Date(now - 2 * 86_400_000).toISOString() })); + await appendAudit(dir, { actor: 'canary', action: 'apply', reason: 'Applied.' }); + await appendAudit(dir, { actor: 'person:ana', action: 'apply', reason: 'Applied.' }); + expect(usageToday(dir, now)).toEqual({ cycles: 1, tokens: 100, applies: 1 }); + }); + + it('a rollback starts a cooldown that ends after the set hours', async () => { + const now = Date.now(); + expect(cooldownEnds(dir, 24, now)).toBeUndefined(); + await appendAudit(dir, { actor: 'guard', action: 'rollback', reason: 'Regressed.' }); + expect(cooldownEnds(dir, 24, now)).toBeGreaterThan(now + 23 * 3_600_000); + expect(cooldownEnds(dir, 24, now + 25 * 3_600_000)).toBeUndefined(); + expect(cooldownEnds(dir, 0, now + 1000)).toBeUndefined(); + }); + + it('the stop file, LOUSHO_IMPROVE=off and enabled false each pause the loop', () => { + expect(pausedBy(dir, { enabled: true }, {})).toBeUndefined(); + expect(pausedBy(dir, { enabled: false }, {})).toMatch(/enabled to false/); + expect(pausedBy(dir, { enabled: true }, { LOUSHO_IMPROVE: ' OFF ' })).toMatch(/LOUSHO_IMPROVE/); + mkdirSync(path.join(dir, '.lousho')); + writeFileSync(path.join(dir, '.lousho', 'improve.stop'), 'maintenance'); + expect(pausedBy(dir, { enabled: true }, {})).toMatch(/improve\.stop/); + }); +}); diff --git a/src/improve/cycles.ts b/src/improve/cycles.ts new file mode 100644 index 00000000..1099dea9 --- /dev/null +++ b/src/improve/cycles.ts @@ -0,0 +1,142 @@ +import { randomBytes } from 'node:crypto'; +import { lstatSync, mkdirSync, readdirSync, readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs'; +import path from 'node:path'; +import { ConfigurationError } from '../execution/errors'; +import { readAudit } from './audit'; +import type { CanaryReplayResult } from './canary'; +import type { GuardCorpusResult } from './corpusGuard'; +import type { ImproveSettings } from './settings'; +import type { ImproveOutcome, ImproveResult } from './types'; + +/** What one self-healing cycle did. */ +export interface HealCycle { + /** The cycle id, which sorts by start time. */ + id: string; + /** What started it: new failed runs, a score regression, or a person running `lousho improve`. */ + trigger: 'failures' | 'score' | 'manual'; + /** The approval mode it ran under. */ + approval: ImproveSettings['approval']; + /** How it ended. `interrupted` is left on disk when a cycle stopped before it finished. */ + outcome: + | 'idle' + | 'paused' + | 'busy' + | 'budget' + | 'cooldown' + | Exclude + | 'canary-failed' + | 'applied' + | 'pending' + | 'guard-rolled-back' + | 'stale' + | 'refused' + | 'interrupted'; + /** Why, as plain text. */ + reason: string; + /** When it started, as an ISO timestamp. */ + startedAt: string; + /** When it ended, as an ISO timestamp. */ + endedAt?: string; + /** The improvement result, when the editors ran. */ + result?: ImproveResult; + /** The replay canary's verdict, when it ran. */ + canary?: CanaryReplayResult; + /** The corpus guard's verdict after an apply. */ + guard?: GuardCorpusResult; + /** The pending proposal id, in manual mode. */ + pending?: string; + /** The revision the cycle started from. */ + from?: string; + /** The revision it applied. */ + to?: string; + /** The revision the corpus guard rolled back to. */ + rolledBackTo?: string; + /** Tokens it spent on runs, editors and reflection. */ + tokens: number; + /** Runs that failed and were left unscored, one message each. */ + errors?: string[]; +} + +const LOUSHO = '.lousho'; +const FOLDER = path.join(LOUSHO, 'cycles'); +const STOP_FILE = path.join(LOUSHO, 'improve.stop'); +const ID = /^\d{8}T\d{6}Z-[0-9a-f]{6}$/; +const DAY_MS = 24 * 60 * 60 * 1000; +const HOUR_MS = 60 * 60 * 1000; +const AUTO_ACTORS: readonly string[] = ['gate', 'canary']; + +/** A new cycle id: its UTC start time and a random suffix. */ +export function newCycleId(now: number): string { + const stamp = new Date(now).toISOString().replace(/[-:]/g, '').replace(/\.\d{3}/, ''); + return `${stamp}-${randomBytes(3).toString('hex')}`; +} + +function folderOf(dir: string): string { + for (const part of [path.join(dir, LOUSHO), path.join(dir, FOLDER)]) { + if (lstatSync(part, { throwIfNoEntry: false })?.isSymbolicLink()) throw new ConfigurationError(`${part} is a symlink; cycle records must live inside the agent directory.`, 'cycles'); + } + return path.join(dir, FOLDER); +} + +/** Saves a cycle record as .lousho/cycles/.json, replacing an earlier save of the same cycle. */ +export function writeCycle(dir: string, cycle: HealCycle): void { + if (!ID.test(cycle.id)) throw new ConfigurationError(`writeCycle: '${cycle.id}' is not a cycle id.`, 'id'); + const folder = folderOf(dir); + mkdirSync(folder, { recursive: true }); + const file = path.join(folder, `${cycle.id}.json`); + const temp = `${file}.${process.pid}.${randomBytes(4).toString('hex')}.tmp`; + try { + writeFileSync(temp, `${JSON.stringify(cycle, null, 2)}\n`, { flag: 'wx', mode: 0o644 }); + renameSync(temp, file); + } catch (error) { + rmSync(temp, { force: true }); + throw error; + } +} + +function readCycle(file: string, id: string): HealCycle | undefined { + try { + const value = JSON.parse(readFileSync(file, 'utf8')) as HealCycle; + return value?.id === id && typeof value.startedAt === 'string' && typeof value.outcome === 'string' && Number.isFinite(value.tokens) ? value : undefined; + } catch { + return undefined; + } +} + +/** Saved cycle records, newest first; unreadable files are left out. */ +export function readCycles(dir: string): HealCycle[] { + const folder = folderOf(dir); + if (!lstatSync(folder, { throwIfNoEntry: false })?.isDirectory()) return []; + return readdirSync(folder) + .flatMap((name) => { + const id = name.slice(0, -'.json'.length); + return name.endsWith('.json') && ID.test(id) ? (readCycle(path.join(folder, name), id) ?? []) : []; + }) + .sort((a, b) => b.id.localeCompare(a.id)); +} + +const dayStart = (now: number): number => Math.floor(now / DAY_MS) * DAY_MS; + +/** Cycles, tokens and automatic applies so far in the current UTC day. */ +export function usageToday(dir: string, now: number): { cycles: number; tokens: number; applies: number } { + const since = dayStart(now); + const today = readCycles(dir).filter((cycle) => Date.parse(cycle.startedAt) >= since); + const applies = readAudit(dir).filter((entry) => entry.action === 'apply' && AUTO_ACTORS.includes(entry.actor) && Date.parse(entry.at) >= since).length; + return { cycles: today.length, tokens: today.reduce((sum, cycle) => sum + cycle.tokens, 0), applies }; +} + +/** When the cooldown after the latest rollback ends, in epoch ms, or `undefined` when none is active. */ +export function cooldownEnds(dir: string, hours: number, now: number): number | undefined { + const last = readAudit(dir).reverse().find((entry) => entry.action === 'rollback'); + if (!last) return undefined; + const ends = Date.parse(last.at) + hours * HOUR_MS; + return ends > now ? ends : undefined; +} + +/** Why the loop is switched off for a directory, or `undefined` when it may run: the stop file, `LOUSHO_IMPROVE=off`, or `enabled: false`. */ +export function pausedBy(dir: string, settings: Pick, env: NodeJS.ProcessEnv = process.env): string | undefined { + if (lstatSync(path.join(dir, STOP_FILE), { throwIfNoEntry: false })) return `${STOP_FILE} exists`; + if (env.LOUSHO_IMPROVE?.trim().toLowerCase() === 'off') return 'LOUSHO_IMPROVE is off'; + if (!settings.enabled) return 'improve.json sets enabled to false'; + return undefined; +} diff --git a/src/improve/heal.test.ts b/src/improve/heal.test.ts new file mode 100644 index 00000000..4c64a7fb --- /dev/null +++ b/src/improve/heal.test.ts @@ -0,0 +1,384 @@ +import { appendFileSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { IngestEvent } from '../observe/envelope'; +import { ingestBatch, openDb } from '../observe/store'; +import type { SqlDatabase } from '../storage/sqlite/driver'; +import { mockModel, type MockRequest, type MockTurn } from '../testing'; +import { readAudit } from './audit'; +import { canaryReplay } from './canary'; +import { guardCorpus } from './corpusGuard'; +import { readCorpus } from './corpus'; +import { readCycles, writeCycle } from './cycles'; +import { healOnce, type HealOptions } from './heal'; +import { listPending, rejectPending } from './pending'; +import { currentRevision } from './revisions'; +import { loadImproveSettings, type ImproveSettings } from './settings'; +import type { Harness } from './types'; + +vi.mock('./canary', async (original) => { + const actual = await original(); + return { ...actual, canaryReplay: vi.fn(actual.canaryReplay) }; +}); + +vi.mock('./corpusGuard', async (original) => { + const actual = await original(); + return { ...actual, guardCorpus: vi.fn(actual.guardCorpus) }; +}); + +const BASE = 'You answer refund questions.'; +const FIX = 'You answer refund questions. If a document is forbidden, say so and stop.'; +const WRONG = 'You answer refund questions. Always be brief.'; +const T0 = Date.now() - 60_000; + +let root: string; +let dir: string; +let db: SqlDatabase; + +const systemOf = (request: MockRequest): string => String(request.messages.find((message) => message.role === 'system')?.content ?? ''); +const isReflection = (request: MockRequest): boolean => systemOf(request).includes('finished attempt to improve'); +const isComposer = (request: MockRequest): boolean => systemOf(request).includes('You are the composer.'); + +function editorWriting(instructions: string, hooks: { compose?: () => void; reflect?: () => never } = {}): MockTurn { + return (request) => { + if (isReflection(request)) { + hooks.reflect?.(); + return 'NO_ENTRY'; + } + if (request.messages.at(-1)?.role === 'tool') return 'Traced to a failed run.'; + if (isComposer(request)) hooks.compose?.(); + return { toolCalls: [{ name: 'write_file', args: { path: 'harness/instructions.md', content: instructions } }] }; + }; +} + +const editor = (instructions = FIX, hooks: Parameters[1] = {}) => ({ provider: mockModel([editorWriting(instructions, hooks)], { onExhausted: 'repeat-last' }) }); + +const answers = (harness: Harness): string => (harness.instructions.includes('forbidden') ? 'ok, that is forbidden' : 'here is the doc'); + +const runCase: HealOptions['runCase'] = async (harness, input) => ({ answer: input.includes('refund window') ? 'five days' : answers(harness), transcript: `user: ${input}\nassistant: ${answers(harness)}`, tokens: 10, ended: 'stop' }); + +const check: HealOptions['check'] = (input, answer) => input.includes('refund window') || answer.startsWith('ok'); + +function root_(trace: string, revision: string, input: string): IngestEvent { + return { + type: 'observation', + data: { obs_id: `r-${trace}`, trace_id: trace, agent_id: 'support', kind: 'span', name: 'invoke_agent support', start: T0, end: T0 + 9, status: 'ok', input, output: 'done', attrs: { 'gen_ai.operation.name': 'invoke_agent', 'lousho.harness.revision': revision } }, + }; +} + +function failed(trace: string, revision = currentRevision(dir)): IngestEvent[] { + return [ + root_(trace, revision, JSON.stringify([{ role: 'user', parts: [{ type: 'text', content: 'can I read the secret doc?' }] }])), + { + type: 'observation', + data: { + obs_id: `t-${trace}`, + trace_id: trace, + parent_obs_id: `r-${trace}`, + agent_id: 'support', + kind: 'tool', + name: 'execute_tool lookup', + start: T0, + end: T0 + 5, + status: 'error', + attrs: { 'gen_ai.operation.name': 'execute_tool', 'gen_ai.tool.name': 'lookup', error: 'HTTP 403 Forbidden' }, + }, + }, + ]; +} + +const seedFailures = (traces = ['t1', 't2', 't3'], revision?: string) => ingestBatch(db, traces.flatMap((trace) => failed(trace, revision))); + +function settings(over: Partial = {}): ImproveSettings { + const loaded = loadImproveSettings(dir, {}, {}); + return { ...loaded, library: { enabled: false }, ...over }; +} + +const heal = (over: Partial = {}) => healOnce({ dir, db, check, runCase, editor: editor(), settings: settings(), ...over }); + +const instructions = () => readFileSync(path.join(dir, 'instructions.md'), 'utf8').trim(); +const actions = () => readAudit(dir).map((entry) => entry.action); + +function tree(base: string): Record { + const files: Record = {}; + for (const entry of readdirSync(base, { recursive: true, withFileTypes: true })) { + if (entry.isFile()) files[path.relative(base, path.join(entry.parentPath, entry.name))] = readFileSync(path.join(entry.parentPath, entry.name), 'utf8'); + } + return files; +} + +beforeEach(() => { + root = mkdtempSync(path.join(tmpdir(), 'lousho-heal-')); + dir = path.join(root, 'support'); + mkdirSync(path.join(dir, 'skills', 'refunds'), { recursive: true }); + writeFileSync(path.join(dir, 'instructions.md'), `${BASE}\n`); + writeFileSync(path.join(dir, 'skills', 'refunds', 'SKILL.md'), '---\nname: refunds\ndescription: How refunds work\n---\nRefunds take 5 days.\n'); + db = openDb(path.join(root, 'observe.db')); + ingestBatch(db, [root_('p1', currentRevision(dir), '"what is the refund window?"')]); +}); + +afterEach(() => { + db.close(); + vi.mocked(canaryReplay).mockClear(); + vi.mocked(guardCorpus).mockClear(); + vi.unstubAllEnvs(); + rmSync(root, { recursive: true, force: true }); +}); + +describe('healOnce', () => { + it('auto: a true fix is applied, guarded and audited', async () => { + seedFailures(); + const from = currentRevision(dir); + const cycle = await heal({ protect: ['p1'] }); + expect(cycle).toMatchObject({ outcome: 'applied', trigger: 'failures', approval: 'auto', from, to: currentRevision(dir) }); + expect(instructions()).toBe(FIX); + expect(cycle.canary?.passed).toBe(true); + expect(cycle.guard?.passed).toBe(true); + expect(readCorpus(dir).map((item) => item.id)).toEqual(['pass:p1']); + expect(actions()).toEqual(['canary-end', 'apply', 'guard-pass']); + expect(readAudit(dir).every((entry) => entry.cycle === undefined || entry.cycle === cycle.id)).toBe(true); + expect(readCycles(dir)).toEqual([expect.objectContaining({ id: cycle.id, outcome: 'applied', tokens: cycle.tokens })]); + expect(cycle.tokens).toBeGreaterThan(0); + }); + + it('manual: the same fix becomes a pending proposal and nothing is written', async () => { + seedFailures(); + const before = instructions(); + const cycle = await heal({ settings: settings({ approval: 'manual', trigger: { minNewFailures: 3, requireRecurring: false } }) }); + expect(cycle.outcome).toBe('pending'); + expect(instructions()).toBe(before); + expect(listPending(dir).map((item) => item.id)).toEqual([cycle.pending]); + expect(actions()).toEqual(['canary-end', 'propose']); + }); + + it('a canary failure applies nothing', async () => { + seedFailures(); + vi.mocked(canaryReplay).mockResolvedValueOnce({ passed: false, cases: [], reason: 'A training case fell below 5 in 6 fresh runs (x).', tokens: 7 }); + const cycle = await heal(); + expect(cycle.outcome).toBe('canary-failed'); + expect(instructions()).toBe(BASE); + expect(actions()).toEqual(['canary-end']); + }); + + it('runs the canary on fresh repeats above the gate, with the training and corpus cases', async () => { + seedFailures(); + await heal({ protect: ['p1'] }); + expect(vi.mocked(canaryReplay).mock.calls[0]?.[0]).toMatchObject({ firstRepeat: 100, runs: 6, protected: ['pass:p1'], training: expect.arrayContaining([expect.stringMatching(/:t[123]$/)]) }); + }); + + it('a guard regression rolls back in the same cycle and starts a cooldown', async () => { + seedFailures(); + const regressing: HealOptions['runCase'] = async (harness, input) => { + const run = await runCase(harness, input); + const applied = readAudit(dir).some((entry) => entry.action === 'apply'); + return applied && input.includes('refund window') && harness.instructions.includes('forbidden') ? { ...run, answer: 'no idea' } : run; + }; + const checkAnswer: HealOptions['check'] = (input, answer) => (input.includes('refund window') ? answer === 'five days' : answer.startsWith('ok')); + const from = currentRevision(dir); + const cycle = await heal({ protect: ['p1'], runCase: regressing, check: checkAnswer }); + expect(cycle).toMatchObject({ outcome: 'guard-rolled-back', rolledBackTo: from }); + expect(instructions()).toBe(BASE); + expect(actions()).toEqual(['canary-end', 'apply', 'guard-fail', 'rollback']); + seedFailures(['t4', 't5', 't6']); + expect((await heal()).outcome).toBe('cooldown'); + }); + + it('stale base is refused', async () => { + seedFailures(); + const cycle = await heal({ editor: editor(FIX, { compose: () => appendFileSync(path.join(dir, 'instructions.md'), 'A person edited this.\n') }) }); + expect(cycle.outcome).toBe('stale'); + expect(instructions()).toContain('A person edited this.'); + expect(actions()).toEqual(['canary-end', 'stale']); + }); + + it('failures from another revision are ignored', async () => { + seedFailures(['t1', 't2', 't3'], 'f'.repeat(64)); + const cycle = await heal(); + expect(cycle.outcome).toBe('idle'); + expect(instructions()).toBe(BASE); + expect(readCycles(dir)).toEqual([]); + }); + + it('is idle below minNewFailures, and counts only failures since the last cycle', async () => { + seedFailures(['t1', 't2']); + expect((await heal()).outcome).toBe('idle'); + expect((await heal({ trigger: 'manual' })).outcome).toBe('applied'); + }); + + it('auto mode needs a failure that recurs, unless a score regression started the cycle', async () => { + seedFailures(['t1']); + expect((await heal({ trigger: 'manual' })).outcome).toBe('idle'); + expect((await heal({ trigger: 'score' })).outcome).toBe('applied'); + }); + + it('the stop file, LOUSHO_IMPROVE=off and enabled false pause it', async () => { + seedFailures(); + vi.stubEnv('LOUSHO_IMPROVE', 'off'); + expect((await heal()).outcome).toBe('paused'); + vi.unstubAllEnvs(); + expect((await heal({ settings: settings({ enabled: false }) })).outcome).toBe('paused'); + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + writeFileSync(path.join(dir, '.lousho', 'improve.stop'), ''); + expect((await heal()).outcome).toBe('paused'); + expect(instructions()).toBe(BASE); + }); + + it('a pause that starts mid-cycle stops it before the write', async () => { + seedFailures(); + const pause = () => { + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + writeFileSync(path.join(dir, '.lousho', 'improve.stop'), ''); + }; + const cycle = await heal({ editor: editor(FIX, { compose: pause }) }); + expect(cycle.outcome).toBe('paused'); + expect(instructions()).toBe(BASE); + }); + + it('a second cycle on the same directory at once is busy', async () => { + seedFailures(); + const [first, second] = await Promise.all([heal(), heal()]); + expect([first.outcome, second.outcome].sort()).toEqual(['applied', 'busy']); + }); + + it('daily caps stop cycles, applies and tokens', async () => { + seedFailures(); + expect((await heal({ settings: settings({ budget: { maxTokensPerDay: 2_000_000, maxCyclesPerDay: 4, maxAppliesPerDay: 1 } }) })).outcome).toBe('applied'); + seedFailures(['t4', 't5', 't6'], currentRevision(dir)); + const capped = await heal({ settings: settings({ budget: { maxTokensPerDay: 2_000_000, maxCyclesPerDay: 4, maxAppliesPerDay: 1 } }) }); + expect(capped).toMatchObject({ outcome: 'budget', reason: expect.stringMatching(/automatic apply/) }); + expect((await heal({ settings: settings({ budget: { maxTokensPerDay: 2_000_000, maxCyclesPerDay: 1, maxAppliesPerDay: 9 } }) })).outcome).toBe('budget'); + expect((await heal({ settings: settings({ budget: { maxTokensPerDay: 10, maxCyclesPerDay: 9, maxAppliesPerDay: 9 } }) })).outcome).toBe('budget'); + }); + + it('a small token budget ends the cycle with budget-exceeded and applies nothing', async () => { + seedFailures(); + const cycle = await heal({ maxTokens: 25 }); + expect(cycle.outcome).toBe('budget-exceeded'); + expect(instructions()).toBe(BASE); + expect(readCycles(dir)[0]?.outcome).toBe('budget-exceeded'); + }); + + it('an outage, every failure passing on replay, and an editor that changes nothing end without a write', async () => { + seedFailures(); + const down: HealOptions['runCase'] = async () => { + throw new Error('provider down'); + }; + const outage = await heal({ runCase: down }); + expect(outage).toMatchObject({ outcome: 'outage', errors: ['provider down'] }); + const healthy: HealOptions['runCase'] = async (harness, input) => ({ ...(await runCase(harness, input)), answer: 'ok' }); + expect((await heal({ trigger: 'manual', runCase: healthy })).outcome).toBe('no-failures'); + expect((await heal({ trigger: 'manual', editor: editor(BASE) })).outcome).toBe('keep-base'); + expect(instructions()).toBe(BASE); + expect(actions()).toEqual([]); + }); + + it('a gate rejection is audited and reflected on, and a failed reflection is audited', async () => { + seedFailures(); + const failing = () => { + throw new Error('reflection model down'); + }; + const cycle = await heal({ editor: editor(WRONG, { reflect: failing }), settings: settings({ library: { enabled: true } }) }); + expect(cycle.outcome).toBe('keep-base'); + expect(actions()).toEqual(['reject', 'reflect-fail']); + expect(readAudit(dir)[1]?.reason).toMatch(/reflection failed/); + }); + + it('a proposal rejected before is refused when it comes back', async () => { + seedFailures(); + const manual = settings({ approval: 'manual', trigger: { minNewFailures: 3, requireRecurring: false } }); + const first = await heal({ settings: manual }); + await rejectPending(dir, first.pending as string, 'ana', 'not now'); + const again = await heal({ settings: manual, trigger: 'manual' }); + expect(again).toMatchObject({ outcome: 'refused', reason: expect.stringMatching(/rejected before/) }); + }); + + it('an aborted cycle is left on disk as interrupted', async () => { + seedFailures(); + const controller = new AbortController(); + const cycle = heal({ signal: controller.signal, editor: editor(FIX, { compose: () => controller.abort() }) }); + await expect(cycle).rejects.toThrow(); + expect(readCycles(dir)[0]?.outcome).toBe('interrupted'); + expect(instructions()).toBe(BASE); + }); + + it('undoes the apply when the guard throws, or cannot judge, without a library vote', async () => { + seedFailures(); + vi.mocked(guardCorpus).mockRejectedValueOnce(new Error('disk full')); + const thrown = await heal({ protect: ['p1'] }); + expect(thrown).toMatchObject({ outcome: 'guard-rolled-back', reason: expect.stringMatching(/could not run \(disk full\)/) }); + expect(instructions()).toBe(BASE); + expect(actions()).toEqual(['canary-end', 'apply', 'rollback']); + }); + + it('a guard that cannot judge counts as a failure', async () => { + seedFailures(); + vi.mocked(guardCorpus).mockResolvedValueOnce({ passed: false, cases: [], revision: 'x', runs: 0, tokens: 0, reason: 'Too few scored runs to judge pass:p1.' }); + const cycle = await heal({ protect: ['p1'] }); + expect(cycle).toMatchObject({ outcome: 'guard-rolled-back', reason: expect.stringMatching(/can't judge counts as a failure/) }); + expect(instructions()).toBe(BASE); + }); + + it('keeps the edit and says so when the rollback is refused', async () => { + seedFailures(); + vi.mocked(guardCorpus).mockImplementationOnce(async () => { + appendFileSync(path.join(dir, 'instructions.md'), 'A person edited this.\n'); + return { passed: false, cases: [{ caseId: 'pass:p1', role: 'holdout', base: { passed: 6, scored: 6 }, final: { passed: 0, scored: 6 }, reasons: [], class: 'regressed' }], revision: 'x', runs: 6, tokens: 0, reason: 'Corpus cases regressed.' }; + }); + const cycle = await heal({ protect: ['p1'] }); + expect(cycle).toMatchObject({ outcome: 'applied', reason: expect.stringMatching(/was refused/) }); + expect(instructions()).toContain('A person edited this.'); + }); + + it('guards an apply whose cycle stopped before its guard, at the start of the next cycle', async () => { + seedFailures(); + const first = await heal({ protect: ['p1'] }); + const { guard: _guard, ...unguarded } = first; + writeCycle(dir, { ...unguarded, outcome: 'interrupted' }); + const regressing: HealOptions['runCase'] = async (harness, input) => { + const run = await runCase(harness, input); + return input.includes('refund window') && harness.instructions.includes('forbidden') ? { ...run, answer: 'no idea' } : run; + }; + const checkAnswer: HealOptions['check'] = (input, answer) => (input.includes('refund window') ? answer === 'five days' : answer.startsWith('ok')); + const next = await heal({ runCase: regressing, check: checkAnswer }); + expect(next.outcome).toBe('cooldown'); + expect(readCycles(dir).find((item) => item.id === first.id)).toMatchObject({ outcome: 'guard-rolled-back', reason: expect.stringMatching(/^Guarded after an interruption/) }); + expect(instructions()).toBe(BASE); + }); + + it('stops before the editors when the kill switch flips during admission', async () => { + seedFailures(); + const pausing: HealOptions['runCase'] = async (harness, input) => { + mkdirSync(path.join(dir, '.lousho'), { recursive: true }); + writeFileSync(path.join(dir, '.lousho', 'improve.stop'), ''); + return runCase(harness, input); + }; + const cycle = await heal({ protect: ['p1'], runCase: pausing }); + expect(cycle).toMatchObject({ outcome: 'paused', reason: expect.stringMatching(/before the editors ran/) }); + expect(cycle.result).toBeUndefined(); + }); + + it('queues instead of applying when improve.json switches to manual mid-cycle', async () => { + seedFailures(); + const cycle = await heal({ editor: editor(FIX, { compose: () => writeFileSync(path.join(dir, 'improve.json'), '{ "approval": "manual" }') }) }); + expect(cycle).toMatchObject({ outcome: 'pending', approval: 'manual' }); + expect(instructions()).toBe(BASE); + }); + + it('applies nothing when the budget left cannot pay for the corpus guard', async () => { + seedFailures(); + vi.mocked(canaryReplay).mockResolvedValueOnce({ passed: true, cases: [{ caseId: 'x', role: 'training', base: { passed: 0, scored: 0 }, final: { passed: 6, scored: 6 }, reasons: [] }], reason: 'Held.', tokens: 1_000_000 }); + const cycle = await heal({ protect: ['p1'], maxTokens: 100_000 }); + expect(cycle).toMatchObject({ outcome: 'budget-exceeded', reason: expect.stringMatching(/corpus guard/) }); + expect(instructions()).toBe(BASE); + }); + + it('never writes outside instructions.md, skills/*/SKILL.md and .lousho/', async () => { + seedFailures(); + const before = tree(root); + await heal({ protect: ['p1'], settings: settings({ library: { enabled: true } }) }); + const changed = Object.keys(tree(root)).filter((file) => tree(root)[file] !== before[file]); + const allowed = /^support\/(instructions\.md|skills\/[^/]+\/SKILL\.md|\.lousho\/.*)$|^observe\.db/; + expect(changed.filter((file) => !allowed.test(file))).toEqual([]); + }); +}); diff --git a/src/improve/heal.ts b/src/improve/heal.ts new file mode 100644 index 00000000..e1f18420 --- /dev/null +++ b/src/improve/heal.ts @@ -0,0 +1,321 @@ +import { lstatSync, mkdirSync } from 'node:fs'; +import path from 'node:path'; +import { ConfigurationError, SDKError } from '../execution/errors'; +import type { AgentHook } from '../execution/hooks'; +import { runsByRevision } from '../observe/store'; +import { withFileLock } from '../storage/fileLock'; +import type { SqlDatabase } from '../storage/sqlite/driver'; +import { appendAudit } from './audit'; +import { canaryReplay, type CanaryReplayResult } from './canary'; +import { admitProtectedCases, casesFromFailures, protectedCasesFromPasses, scoreCase, type CapturedCase, type CaseCheck } from './cases'; +import { addToCorpus, readCorpus, type CorpusCase } from './corpus'; +import { cooldownEnds, newCycleId, pausedBy, readCycles, usageToday, writeCycle, type HealCycle } from './cycles'; +import { formatImproveResult } from './format'; +import { applyAndGuard, corpusCases, guardUnfinished } from './healApply'; +import { + BudgetSpent, + caseProbe, + checkpoint, + counts, + DEFAULT_EDITOR_TIMEOUT_MS, + fileApprovalOf, + finish, + libraryOf, + MAX_ALWAYS_ON, + messageOf, + metered, + recheck, + rejected, + shortId, + unscored, + type HealState, +} from './healState'; +import { improveAgent } from './improve'; +import { savePending } from './pending'; +import { casesFromRecoveries, readableInput } from './recoveries'; +import { harnessHash, readAgentHarness } from './revisions'; +import type { ImproveSettings } from './settings'; +import type { Harness, ImproveOptions, ImproveResult, RecoveryEvidence } from './types'; + +/** Options for {@link healOnce}. */ +export interface HealOptions { + /** The agent directory; its instructions must live in instructions.md. */ + dir: string; + /** The local observe database the failed runs come from. */ + db: SqlDatabase; + /** Scores each answer; it gets the user's text for a one-message run, else the recorded messages as JSON. */ + check: CaseCheck; + /** Runs the agent on one recorded input, as captured (JSON text), under a harness. */ + runCase: ( + harness: Harness, + input: string + ) => Promise<{ + /** The agent's final answer. */ + answer: string; + /** What happened, built with `renderTranscript`. */ + transcript: string; + /** Tokens the run used. */ + tokens: number; + /** How the run ended. */ + ended: string; + }>; + /** The model for the editors and the reflection. */ + editor: ImproveOptions['editor']; + /** The directory's settings, from `loadImproveSettings`. */ + settings: ImproveSettings; + /** What started the cycle (default `failures`, which waits for `settings.trigger.minNewFailures` new failed runs). */ + trigger?: HealCycle['trigger']; + /** Only runs of this agent, by id or registered name. */ + agent?: string; + /** Only runs that started at or after this time, in epoch ms. */ + since?: number; + /** Trace ids of runs that went well; each joins the corpus once it passes 6 of 6 fresh runs. */ + protect?: readonly string[]; + /** Show the editors each failed run's `.heal` retry whose answer passes `check`. */ + withRecoveries?: boolean; + /** Most tokens this cycle may spend, within the settings' caps. */ + maxTokens?: number; + /** Hooks for the editor and reflection sessions, such as a rate limit. */ + editorHooks?: readonly AgentHook[]; + /** Most milliseconds one editor or reflection session may run (default 600000). */ + editorTimeoutMs?: number; + /** Reflect on a decided outcome and store the lesson (default `settings.library.enabled`). */ + reflect?: boolean; + /** Stops the cycle between steps; an apply and its corpus guard always finish. */ + signal?: AbortSignal; +} + +const LOUSHO = '.lousho'; +const MAX_RUNS = 1000; +const ADMISSION_RUNS = 6; +const CANARY_FIRST_REPEAT = 100; +const UNLIMITED = Number.POSITIVE_INFINITY; + +interface Captured { + failures: CapturedCase[]; + passes: CapturedCase[]; + corpus: CorpusCase[]; +} + +type Start = Pick; + +function ended(start: Start, outcome: HealCycle['outcome'], reason: string): HealCycle { + return { ...start, outcome, reason, endedAt: new Date().toISOString() }; +} + +function lockFile(dir: string): string { + const lousho = path.join(dir, LOUSHO); + if (lstatSync(lousho, { throwIfNoEntry: false })?.isSymbolicLink()) throw new ConfigurationError(`${lousho} is a symlink; the self-healing state must live inside the agent directory.`, 'dir'); + mkdirSync(lousho, { recursive: true }); + return path.join(lousho, 'cycle.lock'); +} + +function blocked(options: HealOptions, start: Start, now: number): HealCycle | number { + const { dir, settings } = options; + const ends = cooldownEnds(dir, settings.cooldownHours, now); + if (ends !== undefined) return ended(start, 'cooldown', `A rollback started a ${settings.cooldownHours} hour cooldown, which ends at ${new Date(ends).toISOString()}.`); + const used = usageToday(dir, now); + const { budget } = settings; + if (used.cycles >= budget.maxCyclesPerDay) return ended(start, 'budget', `Today's ${budget.maxCyclesPerDay} cycles have run; the count resets at 00:00 UTC.`); + if (settings.approval === 'auto' && used.applies >= budget.maxAppliesPerDay) return ended(start, 'budget', `Today's ${budget.maxAppliesPerDay} automatic apply has been made; the count resets at 00:00 UTC.`); + const cap = Math.min(budget.maxTokensPerDay - used.tokens, budget.maxTokensPerCycle ?? UNLIMITED, options.maxTokens ?? UNLIMITED); + if (cap <= 0) return ended(start, 'budget', `The token budget is spent: ${used.tokens} of ${budget.maxTokensPerDay} tokens today.`); + return cap; +} + +function revisionTraces(options: HealOptions, revision: string, since: number | undefined): ReturnType { + return runsByRevision(options.db, { revision, limit: MAX_RUNS, ...(options.agent !== undefined && { agent: options.agent }), ...(since !== undefined && { since }) }); +} + +function enoughNewFailures(options: HealOptions, revision: string): boolean { + const last = readCycles(options.dir)[0]; + const since = Math.max(options.since ?? 0, last ? Date.parse(last.startedAt) : 0); + const failed = revisionTraces(options, revision, since).filter((run) => run.failed).length; + return failed >= options.settings.trigger.minNewFailures; +} + +const recurringOnly = (options: HealOptions, trigger: HealCycle['trigger']): boolean => options.settings.trigger.requireRecurring && trigger !== 'score'; + +function capture(options: HealOptions, trigger: HealCycle['trigger'], revision: string): Captured { + const corpus = readCorpus(options.dir); + const traceIds = revisionTraces(options, revision, options.since).map((run) => run.traceId); + if (traceIds.length === 0) return { failures: [], passes: [], corpus }; + const filter = { traceIds, recurring: recurringOnly(options, trigger), ...(options.since !== undefined && { since: options.since }) }; + const inCorpus = new Set(corpus.map((item) => item.traceId)); + const recovered = new Map(options.withRecoveries ? casesFromRecoveries(options.db, filter).map((captured) => [captured.id, captured]) : []); + const failures = casesFromFailures(options.db, filter) + .filter((captured) => !inCorpus.has(captured.traceId)) + .map((captured) => recovered.get(captured.id) ?? captured); + const failed = new Set(failures.map((captured) => captured.traceId)); + const passes = protectedCasesFromPasses(options.db, options.protect ?? []).filter((captured) => !failed.has(captured.traceId) && !inCorpus.has(captured.traceId)); + return { failures, passes, corpus }; +} + +function checkedRecoveries(captured: Captured, check: CaseCheck): RecoveryEvidence[] { + const guarded = new Set([...captured.passes, ...captured.corpus].map((item) => item.traceId)); + return captured.failures.flatMap(({ input, recovery }) => (recovery && !guarded.has(recovery.traceId) && scoreCase(check, readableInput(input), recovery.answer).passed === true ? [recovery] : [])); +} + +async function admit(state: HealState, captured: Captured): Promise { + if (captured.passes.length === 0) return captured.corpus; + const { admitted } = await admitProtectedCases({ harness: state.harness, cases: captured.passes, probe: metered(state), runs: ADMISSION_RUNS }); + if (admitted.length === 0) return captured.corpus; + const addedAt = new Date().toISOString(); + const revision = state.cycle.from as string; + const cases = admitted.map((item): CorpusCase => ({ id: item.id, traceId: item.traceId, input: item.input, addedAt, revision, origin: 'protected', baseline: { passed: ADMISSION_RUNS, scored: ADMISSION_RUNS } })); + return addToCorpus(state.dir, cases, state.settings.guard.maxCases); +} + +function propose(state: HealState, captured: Captured, holdout: readonly string[]): Promise { + const { options, settings } = state; + const recoveries = checkedRecoveries(captured, options.check); + const library = state.library?.files ?? {}; + return improveAgent({ + harness: state.harness, + cases: captured.failures.map((item) => item.id), + holdout, + probe: state.probe, + editor: options.editor, + quotedChecks: settings.approval === 'manual' ? 'flag' : 'reject', + maxTokens: state.cap - state.spent, + maxAlwaysOnChars: MAX_ALWAYS_ON, + editorTimeoutMs: options.editorTimeoutMs ?? DEFAULT_EDITOR_TIMEOUT_MS, + cacheDir: path.join(state.dir, LOUSHO, 'cache'), + ...(recoveries.length > 0 && { recoveries }), + ...(Object.keys(library).length > 0 && { library }), + ...(options.editorHooks && { editorHooks: options.editorHooks }), + ...(options.signal && { signal: options.signal }), + }); +} + +async function noProposal(state: HealState, result: ImproveResult & { outcome: Exclude }): Promise { + if (result.outcome !== 'keep-base' || !result.gate?.length) return finish(state, result.outcome, result.reason); + const { cycle } = state; + await appendAudit(state.dir, { actor: 'gate', action: 'reject', from: cycle.from, cycle: cycle.id, reason: `Not applied, the gate rejected it: ${result.reason}`, evidence: { gate: result.reason, cases: counts(result.cases) } }); + return rejected(state, result.outcome, result.reason); +} + +async function runCanary(state: HealState, result: ImproveResult, candidate: Harness, holdout: readonly string[]): Promise { + const training = (result.cases ?? []).filter((verdict) => verdict.role === 'training').map((verdict) => verdict.caseId); + const canary = await canaryReplay({ + base: state.harness, + candidate, + training, + protected: holdout, + probe: metered(state), + runs: state.settings.canary.replayRuns, + firstRepeat: CANARY_FIRST_REPEAT, + ...(state.options.signal && { signal: state.options.signal }), + }); + const verdict = canary.passed ? 'passed' : 'failed'; + await appendAudit(state.dir, { actor: 'canary', action: 'canary-end', from: state.cycle.from, to: harnessHash(candidate), cycle: state.cycle.id, reason: `The replay canary ${verdict}: ${canary.reason}`, evidence: { cases: counts(canary.cases) } }); + return canary; +} + +async function queue(state: HealState, result: ImproveResult): Promise { + try { + const pending = await savePending(state.dir, { result, base: state.harness, card: formatImproveResult(result), cycle: state.cycle.id, maxAlwaysOnChars: MAX_ALWAYS_ON }); + return finish(state, 'pending', `Queued for approval as ${pending.id}; approve it with lousho improve approve.`, { pending: pending.id }); + } catch (error) { + if (!(error instanceof ConfigurationError)) throw error; + return rejected(state, 'refused', error.message); + } +} + +function pausedStop(state: HealState, when: string): HealCycle | undefined { + const { paused } = recheck(state); + return paused ? finish(state, 'paused', `Paused ${when}: ${paused}. Nothing was applied.`) : undefined; +} + +async function decide(state: HealState, result: ImproveResult, candidate: Harness, holdout: readonly string[]): Promise { + const before = pausedStop(state, 'before the canary'); + if (before) return before; + const canary = await runCanary(state, result, candidate, holdout); + state.cycle.canary = canary; + checkpoint(state); + if (!canary.passed) return rejected(state, 'canary-failed', canary.reason); + const now = recheck(state); + if (now.paused) return finish(state, 'paused', `Paused before the write: ${now.paused}. Nothing was applied.`); + state.cycle.approval = now.approval; + return now.approval === 'auto' ? applyAndGuard(state, result) : queue(state, result); +} + +async function work(state: HealState, captured: Captured): Promise { + try { + const corpus = await admit(state, captured); + checkpoint(state); + const holdout = [...new Set(corpus.map((item) => item.id))]; + const paused = pausedStop(state, 'before the editors ran'); + if (paused) return paused; + if (state.spent >= state.cap) throw new BudgetSpent(); + const result = await propose(state, captured, holdout); + state.cycle.result = result; + state.spent += result.spend.tokens; + checkpoint(state); + if (result.outcome !== 'proposal') return await noProposal(state, { ...result, outcome: result.outcome }); + if (!result.harness) return finish(state, 'keep-base', result.reason); + return await decide(state, result, result.harness, holdout); + } catch (error) { + if (error instanceof BudgetSpent) return finish(state, 'budget-exceeded', `Stopped at the cycle's ${state.cap} token budget; nothing was applied.`); + throw error; + } +} + +function idleReason(options: HealOptions, trigger: HealCycle['trigger'], revision: string): string { + const twice = recurringOnly(options, trigger) ? ' that failed the same way in two or more runs' : ''; + return `No failed run under revision ${shortId(revision)} with a recorded input${twice}, so there is nothing to learn from.`; +} + +async function run(options: HealOptions, start: Start, harness: Harness, cap: number): Promise { + const revision = harnessHash(harness); + if (start.trigger === 'failures' && !enoughNewFailures(options, revision)) { + return ended(start, 'idle', `Fewer than ${options.settings.trigger.minNewFailures} new failed runs under revision ${shortId(revision)} since the last cycle.`); + } + const captured = capture(options, start.trigger, revision); + if (captured.failures.length === 0) return ended(start, 'idle', idleReason(options, start.trigger, revision)); + const cycle: HealCycle = { ...start, outcome: 'interrupted', reason: 'The cycle stopped before it finished.', from: revision }; + const errors = new Set(); + const probe = unscored(caseProbe([...captured.failures, ...captured.passes, ...corpusCases(captured.corpus)], options.runCase, options.check), errors, options.signal); + const fileApproval = fileApprovalOf(options.dir); + const state: HealState = { options, dir: options.dir, settings: options.settings, cycle, harness, cap, spent: 0, errors, probe, library: libraryOf(options.dir, options.settings), ...(fileApproval && { fileApproval }) }; + writeCycle(options.dir, cycle); + try { + const done = await work(state, captured); + writeCycle(options.dir, done); + return done; + } catch (error) { + writeCycle(options.dir, finish(state, 'interrupted', `The cycle stopped before it finished: ${messageOf(error)}`)); + throw error; + } +} + +async function locked(options: HealOptions, start: Start, now: number): Promise { + const harness = readAgentHarness(options.dir); + const recovered = await guardUnfinished(options, harness, UNLIMITED); + const current = recovered?.rolledBackTo ? readAgentHarness(options.dir) : harness; + const cap = blocked(options, start, now); + if (typeof cap !== 'number') return cap; + return run(options, start, current, cap); +} + +/** Runs one detect, capture, propose, gate, canary, apply-or-queue and guard cycle for an agent directory, and records it under .lousho/cycles/. */ +export async function healOnce(options: HealOptions): Promise { + const now = Date.now(); + const start: Start = { id: newCycleId(now), trigger: options.trigger ?? 'failures', approval: options.settings.approval, startedAt: new Date(now).toISOString(), tokens: 0 }; + const paused = pausedBy(options.dir, options.settings); + if (paused) return ended(start, 'paused', `Paused: ${paused}.`); + let held = false; + try { + return await withFileLock( + lockFile(options.dir), + () => { + held = true; + return locked(options, start, now); + }, + { timeoutMs: 0, busyCode: 'LOUSHO_STORAGE_BUSY' } + ); + } catch (error) { + if (!held && error instanceof SDKError && error.code === 'LOUSHO_STORAGE_BUSY') return ended(start, 'busy', 'Another cycle is running for this agent directory.'); + throw error; + } +} diff --git a/src/improve/healApply.ts b/src/improve/healApply.ts new file mode 100644 index 00000000..43f2253c --- /dev/null +++ b/src/improve/healApply.ts @@ -0,0 +1,96 @@ +import { ConfigurationError } from '../execution/errors'; +import { applyProposal, rollbackHarness } from './apply'; +import type { CapturedCase } from './cases'; +import { readCorpus, type CorpusCase } from './corpus'; +import { guardCorpus, type GuardCorpusResult } from './corpusGuard'; +import { readCycles, writeCycle, type HealCycle } from './cycles'; +import type { HealOptions } from './heal'; +import { STALE_FIELD } from './harnessWriter'; +import { caseProbe, checkpoint, counted, finish, libraryOf, MAX_ALWAYS_ON, messageOf, reflectAndVote, rejected, shortId, unscored, type HealState } from './healState'; +import { harnessHash } from './revisions'; +import type { Harness, ImproveResult } from './types'; + +interface GuardCheck { + guard?: GuardCorpusResult; + reason: string; + undo: boolean; + regressed: boolean; +} + +export function corpusCases(corpus: readonly CorpusCase[]): CapturedCase[] { + return corpus.map((item) => ({ id: item.id, input: item.input, traceId: item.traceId, kind: 'protected' })); +} + +async function runGuard(state: HealState): Promise { + let guard: GuardCorpusResult; + try { + guard = await guardCorpus(state.dir, { probe: counted(state), runs: state.settings.guard.runs, rollback: false }); + } catch (error) { + return { reason: `The corpus guard could not run (${messageOf(error)}), so the edit is undone.`, undo: true, regressed: false }; + } + if (guard.passed) return { guard, reason: guard.reason, undo: false, regressed: false }; + const regressed = guard.cases.some((item) => item.class === 'regressed'); + return { guard, reason: regressed ? guard.reason : `${guard.reason} A guard that can't judge counts as a failure, so the edit is undone.`, undo: true, regressed }; +} + +async function undoApply(state: HealState, to: string, reason: string): Promise<{ to: string } | { refused: string }> { + try { + return { to: (await rollbackHarness(state.dir, state.cycle.from as string, { actor: 'guard', reason, expectedFrom: to })).to }; + } catch (error) { + if (error instanceof ConfigurationError) return { refused: error.message }; + throw error; + } +} + +async function guardApplied(state: HealState, to: string): Promise { + const check = await runGuard(state); + if (check.guard) state.cycle.guard = check.guard; + checkpoint(state); + if (!check.undo) return finish(state, 'applied', `Applied ${shortId(to)}. ${check.reason}`); + const undone = await undoApply(state, to, check.reason); + if ('refused' in undone) return finish(state, 'applied', `Applied ${shortId(to)}. ${check.reason} The rollback to ${shortId(state.cycle.from)} was refused (${undone.refused}), so the edit stays and the monitor still checks it.`); + const reason = `Applied ${shortId(to)}, then rolled back to ${shortId(undone.to)}: ${check.reason}`; + if (check.regressed) await reflectAndVote(state, 'rolled-back'); + return finish(state, 'guard-rolled-back', reason, { rolledBackTo: undone.to }); +} + +function guardShortfall(state: HealState): string | undefined { + const { canary } = state.cycle; + const size = readCorpus(state.dir).length; + if (!canary || size === 0) return undefined; + const runs = canary.cases.reduce((sum, verdict) => sum + verdict.final.scored + verdict.base.scored, 0); + const needed = Math.ceil(size * state.settings.guard.runs * (canary.tokens / Math.max(1, runs))); + const left = state.cap - state.spent; + if (needed <= left) return undefined; + return `The corpus guard that must follow an apply needs about ${needed} tokens, more than the ${Math.max(0, left)} left, so nothing was applied.`; +} + +/** Applies a proposal that passed the gate and the canary, then guards it with the corpus and undoes it on a regression or a guard that can't judge. */ +export async function applyAndGuard(state: HealState, result: ImproveResult): Promise { + const shortfall = guardShortfall(state); + if (shortfall) return finish(state, 'budget-exceeded', shortfall); + let to: string; + try { + to = (await applyProposal(state.dir, { result, base: state.harness, cycle: state.cycle.id, maxAlwaysOnChars: MAX_ALWAYS_ON }, { mode: 'auto', by: 'canary' })).to; + } catch (error) { + if (!(error instanceof ConfigurationError)) throw error; + if (error.field === STALE_FIELD) return finish(state, 'stale', error.message); + return rejected(state, 'refused', error.message); + } + state.cycle.to = to; + checkpoint(state); + return guardApplied(state, to); +} + +/** Guards an apply whose cycle stopped before its guard ran, when the directory still holds that revision. */ +export async function guardUnfinished(options: HealOptions, harness: Harness, cap: number): Promise { + const last = readCycles(options.dir)[0]; + if (last?.outcome !== 'interrupted' || last.to === undefined || last.guard !== undefined || harnessHash(harness) !== last.to) return undefined; + const errors = new Set(); + const probe = unscored(caseProbe(corpusCases(readCorpus(options.dir)), options.runCase, options.check), errors, undefined); + const state: HealState = { options, dir: options.dir, settings: options.settings, cycle: last, harness, cap, spent: last.tokens, errors, probe, library: libraryOf(options.dir, options.settings) }; + const done = await guardApplied(state, last.to); + done.reason = `Guarded after an interruption. ${done.reason}`; + writeCycle(options.dir, done); + return done; +} diff --git a/src/improve/healState.ts b/src/improve/healState.ts new file mode 100644 index 00000000..05dbb0a7 --- /dev/null +++ b/src/improve/healState.ts @@ -0,0 +1,144 @@ +import { appendAudit } from './audit'; +import { probeCapturedCases, type CapturedCase, type CaseCheck } from './cases'; +import { pausedBy, writeCycle, type HealCycle } from './cycles'; +import type { HealOptions } from './heal'; +import { libraryFiles, libraryHome, type LibraryEntry } from './library'; +import { readableInput } from './recoveries'; +import { reflect, voteOnOutcome } from './reflect'; +import { loadImproveSettings, type ImproveSettings } from './settings'; +import type { CaseVerdict, Harness, ImproveOptions } from './types'; + +export const MAX_ALWAYS_ON = 8000; +export const DEFAULT_EDITOR_TIMEOUT_MS = 10 * 60 * 1000; +const MAX_ERRORS = 10; + +export class BudgetSpent extends Error {} + +export interface Library { + dir: string; + domain: string; + files: Record; +} + +export interface HealState { + options: HealOptions; + dir: string; + settings: ImproveSettings; + cycle: HealCycle; + harness: Harness; + cap: number; + spent: number; + errors: Set; + probe: ImproveOptions['probe']; + library?: Library; + fileApproval?: ImproveSettings['approval']; +} + +export const messageOf = (error: unknown): string => (error instanceof Error ? error.message : String(error)); + +export const shortId = (id: string | undefined): string => (id ?? '').slice(0, 12); + +export function caseProbe(cases: readonly CapturedCase[], runCase: HealOptions['runCase'], check: CaseCheck): ImproveOptions['probe'] { + return probeCapturedCases(cases, runCase, (input, answer) => check(readableInput(input), answer)); +} + +export function unscored(probe: ImproveOptions['probe'], errors: Set, signal: AbortSignal | undefined): ImproveOptions['probe'] { + return async (harness, caseId, repeat) => { + try { + return await probe(harness, caseId, repeat); + } catch (error) { + if (signal?.aborted) throw error; + errors.add(messageOf(error)); + return { passed: null, transcript: '', tokens: 0, ended: `error: ${messageOf(error)}` }; + } + }; +} + +export function counted(state: HealState): ImproveOptions['probe'] { + return async (harness, caseId, repeat) => { + const run = await state.probe(harness, caseId, repeat); + state.spent += run.tokens; + return run; + }; +} + +export function metered(state: HealState): ImproveOptions['probe'] { + const probe = counted(state); + return async (harness, caseId, repeat) => { + if (state.spent >= state.cap) throw new BudgetSpent(); + return probe(harness, caseId, repeat); + }; +} + +export function finish(state: HealState, outcome: HealCycle['outcome'], reason: string, extra: Partial = {}): HealCycle { + const errors = [...state.errors].slice(0, MAX_ERRORS); + return Object.assign(state.cycle, extra, { outcome, reason, endedAt: new Date().toISOString(), tokens: state.spent, ...(errors.length > 0 && { errors }) }); +} + +export function checkpoint(state: HealState): void { + writeCycle(state.dir, { ...state.cycle, tokens: state.spent }); +} + +export function fileApprovalOf(dir: string): ImproveSettings['approval'] | undefined { + try { + return loadImproveSettings(dir).approval; + } catch { + return undefined; + } +} + +export function recheck(state: HealState): { paused?: string; approval: ImproveSettings['approval'] } { + try { + const file = loadImproveSettings(state.dir); + const paused = pausedBy(state.dir, { enabled: state.settings.enabled && file.enabled }); + const switched = state.fileApproval === 'auto' && file.approval === 'manual'; + return { ...(paused && { paused }), approval: switched ? 'manual' : state.settings.approval }; + } catch (error) { + return { paused: `improve.json can't be read (${messageOf(error)})`, approval: 'manual' }; + } +} + +export function libraryOf(dir: string, settings: ImproveSettings): Library | undefined { + if (!settings.library.enabled) return undefined; + const home = libraryHome(dir, settings.library.domain); + return { ...home, files: libraryFiles(home.dir, home.domain) }; +} + +export const counts = (cases: readonly CaseVerdict[] | undefined) => cases?.map((verdict) => ({ caseId: verdict.caseId, base: `${verdict.base.passed}/${verdict.base.scored}`, final: `${verdict.final.passed}/${verdict.final.scored}` })); + +function sessionSignal(options: HealOptions): AbortSignal { + const timer = AbortSignal.timeout(options.editorTimeoutMs ?? DEFAULT_EDITOR_TIMEOUT_MS); + return options.signal ? AbortSignal.any([options.signal, timer]) : timer; +} + +async function learn(state: HealState, outcome: LibraryEntry['outcome']): Promise { + const { options, library, cycle } = state; + const result = cycle.result; + const remaining = state.cap - state.spent; + if (!library || !result || !(options.reflect ?? true) || remaining <= 0 || options.signal?.aborted) return undefined; + const reflection = await reflect({ + dir: library.dir, + domain: library.domain, + result, + outcome, + revision: cycle.from as string, + editor: options.editor, + ...(Number.isFinite(remaining) && { maxTokens: remaining }), + ...(options.editorHooks && { hooks: options.editorHooks }), + signal: sessionSignal(options), + }); + state.spent += reflection.tokens; + if (reflection.error !== undefined) await appendAudit(state.dir, { actor: 'scheduler', action: 'reflect-fail', from: cycle.from, cycle: cycle.id, reason: `The reflection failed, so no lesson was stored: ${reflection.error}` }); + return reflection.entry?.id; +} + +export async function rejected(state: HealState, outcome: HealCycle['outcome'], reason: string, extra: Partial = {}): Promise { + await learn(state, 'rejected'); + return finish(state, outcome, reason, extra); +} + +export async function reflectAndVote(state: HealState, outcome: 'rolled-back'): Promise { + const entryId = await learn(state, outcome); + const { library, cycle } = state; + if (library && cycle.result) voteOnOutcome(library.dir, { outcome, result: cycle.result, domain: library.domain, library: library.files, ...(entryId && { entryId }) }); +} diff --git a/src/improve/improve.test.ts b/src/improve/improve.test.ts index 3e52f6e9..2181cb8f 100644 --- a/src/improve/improve.test.ts +++ b/src/improve/improve.test.ts @@ -571,6 +571,13 @@ describe('improveAgent', () => { expect(sequential.stages).toEqual(result.stages); }); + it('ends as an outage when an editor session runs past editorTimeoutMs', async () => { + const slow: MockTurn = () => new Promise((resolve) => setTimeout(() => resolve('KEEP_BASE.'), 200)); + const result = await improveAgent(options({ editor: { provider: mockModel([slow], { onExhausted: 'repeat-last' }) }, editorTimeoutMs: 20 })); + expect(result).toMatchObject({ outcome: 'outage', reason: expect.stringMatching(/20 ms wall timeout/) }); + await expect(improveAgent(options({ editorTimeoutMs: 0 }))).rejects.toThrow(/editorTimeoutMs/); + }); + it('reports an outage when no probe could be scored, instead of a score of zero', async () => { const result = await improveAgent(options({ probe: async () => run(null) })); expect(result.outcome).toBe('outage'); diff --git a/src/improve/improve.ts b/src/improve/improve.ts index af046f0f..6783df6c 100644 --- a/src/improve/improve.ts +++ b/src/improve/improve.ts @@ -23,6 +23,8 @@ function assertOptions(options: ImproveOptions, repeats: number): void { const outside = Object.entries(options.library ?? {}).filter(([path, content]) => !LIBRARY_PATH.test(path) || !isLibraryFile(content)); if (outside.length > 0) throw new ConfigurationError(`improveAgent: library must hold files from libraryFiles, named context/library/.md, not ${outside.map(([path]) => path).join(', ')}.`, 'library'); assertProposers(options); + const timeout = options.editorTimeoutMs; + if (timeout !== undefined && (!Number.isInteger(timeout) || timeout <= 0)) throw new ConfigurationError(`improveAgent: editorTimeoutMs must be a positive whole number of milliseconds, got ${timeout}.`, 'editorTimeoutMs'); } const LIBRARY_PATH = /^context\/library\/[A-Za-z0-9_-][A-Za-z0-9._-]*\.md$/; diff --git a/src/improve/index.ts b/src/improve/index.ts index 3955bbd7..f3dc12f7 100644 --- a/src/improve/index.ts +++ b/src/improve/index.ts @@ -30,3 +30,6 @@ export { addToCorpus, readCorpus } from './corpus'; export type { CorpusCase } from './corpus'; export { guardCorpus } from './corpusGuard'; export type { GuardCorpusOptions, GuardCorpusResult, GuardedCase } from './corpusGuard'; +export { healOnce } from './heal'; +export type { HealOptions } from './heal'; +export type { HealCycle } from './cycles'; diff --git a/src/improve/recoveries.ts b/src/improve/recoveries.ts index a2e71304..52c35efd 100644 --- a/src/improve/recoveries.ts +++ b/src/improve/recoveries.ts @@ -1,5 +1,5 @@ import { failedTraceIdsFromDb } from '../observe/failures'; -import type { Message } from '../providers'; +import type { AgentInput, Message } from '../providers'; import type { SqlDatabase, SqlRow } from '../storage/sqlite/driver'; import { casesFromFailures, type CapturedCase } from './cases'; import { renderTranscript } from './evidence'; @@ -33,10 +33,28 @@ function turnOf(item: { role?: unknown; parts?: unknown } | null): Turn[] { return RECORDED_ROLES.has(role) && content ? [{ role: role as Message['role'], content }] : []; } -export function recordedTurns(items: readonly unknown[]): Turn[] { +function recordedTurns(items: readonly unknown[]): Turn[] { return items.flatMap((item) => turnOf(item as { role?: unknown; parts?: unknown } | null)); } +export function agentInputOf(input: string): AgentInput { + let value: unknown; + try { + value = JSON.parse(input); + } catch { + return input; + } + if (typeof value === 'string') return value; + const messages = Array.isArray(value) ? recordedTurns(value) : []; + if (messages.length === 0) return input; + return messages.length === 1 && messages[0].role === 'user' ? messages[0].content : messages; +} + +export function readableInput(input: string): string { + const message = agentInputOf(input); + return typeof message === 'string' ? message : input; +} + function recordedMessages(text: unknown, fallback: 'user' | 'assistant'): Turn[] { const value = parsed(text); if (typeof value === 'string') return [{ role: fallback, content: value }]; @@ -105,8 +123,9 @@ function healOf(db: SqlDatabase, run: Run, failed: ReadonlySet): { trace } /** Turns failed runs whose `.heal` fork from `agent.recover()` then finished with no recorded failure into cases to learn from, each carrying the retry as `recovery` evidence for the editors. */ -export function casesFromRecoveries(db: SqlDatabase, filter: { since?: number; agent?: string; minScore?: number } = {}): CapturedCase[] { - const failed = failedTraceIdsFromDb(db, filter); +export function casesFromRecoveries(db: SqlDatabase, filter: Parameters[1] = {}): CapturedCase[] { + const { recurring: _recurring, ...report } = filter; + const failed = failedTraceIdsFromDb(db, report); const latest = new Map(); for (const captured of casesFromFailures(db, filter)) { const run = runOf(db, captured.traceId); diff --git a/src/improve/run.ts b/src/improve/run.ts index 22339ec3..a43fd4fe 100644 --- a/src/improve/run.ts +++ b/src/improve/run.ts @@ -23,6 +23,12 @@ export class StopRun extends Error { } } +function sessionFailure(error: unknown, timedOut: boolean, timeout: number | undefined): unknown { + if (timedOut) return new StopRun('outage', `An editor session ran past its ${timeout} ms wall timeout.`); + if (error instanceof CompactedLLMProviderError && error.compacted.retryable) return new StopRun('outage', `The editor's provider failed: ${error.message}`); + return error; +} + const notesOf = (text: string) => /([\s\S]*?)<\/notes>/.exec(text)?.[1]?.trim() ?? text.trim(); interface EditContext { @@ -128,13 +134,15 @@ export class ImproveRun { maxSteps: maxEditorSteps ?? DEFAULT_MAX_EDITOR_STEPS, ...(editorHooks && { hooks: [...editorHooks] }), }); + const timeout = this.options.editorTimeoutMs; + const timer = timeout === undefined ? undefined : AbortSignal.timeout(timeout); + const stop = timer && signal ? AbortSignal.any([signal, timer]) : (timer ?? signal); try { - const result = await agent.send(input, signal ? { signal } : undefined); - signal?.throwIfAborted(); + const result = await agent.send(input, stop ? { signal: stop } : undefined); + stop?.throwIfAborted(); return { text: result.text ?? '', tokens: result.usage?.totalTokens ?? 0 }; } catch (error) { - if (error instanceof CompactedLLMProviderError && error.compacted.retryable) throw new StopRun('outage', `The editor's provider failed: ${error.message}`); - throw error; + throw sessionFailure(error, timer?.aborted === true && signal?.aborted !== true, timeout); } } diff --git a/src/improve/types.ts b/src/improve/types.ts index 5b0a81e2..590b4aec 100644 --- a/src/improve/types.ts +++ b/src/improve/types.ts @@ -83,6 +83,8 @@ export interface ImproveOptions { maxCostIncrease?: number; /** Most steps per editor session (default 30). */ maxEditorSteps?: number; + /** Most milliseconds one editor session may run; a session past it ends the run as an `outage` (default: no limit). */ + editorTimeoutMs?: number; /** Runs the editors even when every case passed, to test whether they rightly change nothing. */ alwaysPropose?: boolean; /** Hooks for the editor sessions, such as a rate-limit delay. */