/** * autoplan-shadow — shadow mode for the CEO-only decider (GST-537 PR 3, * docs/unattended.md#shadow-mode). In a driver run that turned it on at init * (`autoplan_shadow on` or GSTACK_AUTOPLAN_SHADOW=1; off by default), every * gate the driver publishes gets one hidden model prediction per question, * written before anyone answers: action (decide, send_back, hold), pick, * confidence, cites and reason, beside the free keep-direction rule's pick. * Nothing routes differently and a shadow failure never blocks a gate or an * answer. The decider prompt is the frozen, hand-written * autoplan/shadow-decider.md (its sha256 is pinned below and on every row). * * The publication is two-phase so the run lock is never held during the call: * `claimShadow` runs under the run lock of the command that wrote the gate and * takes a lease (`/shadow/r.pending`: pid, host, deadline; journaled * `shadow_pending`); `runShadow` calls the model with no lock held (each call * reserved under its own ledger id labeled `shadow` and journaled * `shadow_dispatched` before the send), then retakes the lock with a bounded * wait and publishes `/shadow/ceo-only.r.jsonl` with link (a second * writer never overwrites) plus `shadow_recorded` or `shadow_failed`. A live * lease makes every other writer pay nothing, and `answer` waits for it * (`shadowBeforeAnswer`). Only a dead lease (its pid gone, or past its * deadline) is a crash: its dispatched calls settle `unknown`, never released, * and `answer` then records the revision with every row `retrospective`. * This module is the only one that names the shadow folder's paths; the reader * for gstack-evals and the private validate check are in * lib/autoplan-shadow-reader.ts. */ import * as crypto from 'node:crypto'; import * as fs from 'node:fs'; import * as os from 'node:os'; import * as path from 'node:path'; import { extractHeading } from './autoplan-prompts'; import { GSTACK_VERSION } from './autoplan-migrate'; import { withRunLock } from './autoplan-run'; import { FILES, RunError, journal, now, readJournal, readState, voiceRows, type RunState } from './autoplan-state'; import { decisionId, recHash, ruleKept, type GateItem, type GateList } from './gate-list'; import type { FindingRow } from './headless-artifacts'; import { readJsonl } from './headless-artifacts'; import { apiKeyFor, callModel, type ApiCallResult } from './outside-voice-api'; import { modelFamily } from './outside-voice-runner'; import { estimateCostUsd, hasPricing } from './pricing'; import { readLedger, release, reserve, settle, summarize } from './spend-ledger'; import { resolveSourceRoot } from './source-root'; import { readConfigKey } from './state-root'; export const SHADOW_PROMPT_VERSION = 1; /** sha256 of autoplan/shadow-decider.md at SHADOW_PROMPT_VERSION; a prompt edit bumps both (test/autoplan-shadow.test.ts). */ export const SHADOW_PROMPT_SHA256 = 'e4857932ae831b7a6d933c42c182ccd7711c68338fec6901799264f645589344'; export const DEFAULT_SHADOW_MODEL = 'claude-opus-5-5'; /** One deadline for the call and its one retry. */ export const SHADOW_DEADLINE_MS = 90_000; export const SHADOW_MAX_TOKENS = 4_000; /** A reply cut at the output limit is asked once more at this limit, inside the same deadline. */ export const SHADOW_RAISED_MAX_TOKENS = 8_000; export const SHADOW_SINK = 'autoplan-shadow'; export const SHADOW_CONSENT = 'autoplan_shadow on'; const PLAN_CHARS = 40_000; const REASON_CHARS = 400; const RELOCK_MS = 30_000; const LEASE_GRACE_MS = 30_000; export const SHADOW_ACTIONS = ['decide', 'send_back', 'hold'] as const; const ROOT = path.resolve(import.meta.dir, '..'); export const shadowPromptFile = () => path.join(resolveSourceRoot(ROOT), 'autoplan', 'shadow-decider.md'); export const shadowDir = (out: string) => path.join(out, 'shadow'); export const recordFile = (out: string, rev: number) => path.join(shadowDir(out), `ceo-only.r${rev}.jsonl`); export const RECORD_RE = /^ceo-only\.r(\d+)\.jsonl$/; const leaseFile = (out: string, rev: number) => path.join(shadowDir(out), `r${rev}.pending`); const saltFile = (out: string) => path.join(shadowDir(out), 'salt'); const sha256 = (data: string | Buffer) => crypto.createHash('sha256').update(data).digest('hex'); const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); // --------------------------------------------------------------------------- // the switch, fixed at init // --------------------------------------------------------------------------- export interface ShadowSwitch { enabled: boolean; model?: string; reason?: string; source: 'env' | 'config' | 'default' } /** * GSTACK_AUTOPLAN_SHADOW (1|on|0|off), else gstack-config autoplan_shadow (default off). On also needs a priced model * (autoplan_shadow_model, default claude-opus-5-5), its family's API key, and a run that is neither --no-outside nor * CEO-only. Stored in run.json at init and never re-read. The line is printed only when shadow was asked for. */ export function shadowSwitch(o: { outsideEnabled: boolean; ceoOnly: boolean }, env: NodeJS.ProcessEnv = process.env): { shadow: ShadowSwitch; line?: string } { const fromEnv = !!env.GSTACK_AUTOPLAN_SHADOW?.trim(); const value = (fromEnv ? env.GSTACK_AUTOPLAN_SHADOW! : readConfigKey('autoplan_shadow', env) ?? '').trim().toLowerCase(); const source = fromEnv ? 'env' as const : value ? 'config' as const : 'default' as const; const label = fromEnv ? `GSTACK_AUTOPLAN_SHADOW=${value}` : `autoplan_shadow ${value}`; const off = (reason: string) => ({ shadow: { enabled: false, reason, source }, line: `SHADOW: off (${reason})` }); if (!value || value === 'off' || value === '0') return { shadow: { enabled: false, reason: source === 'default' ? 'off by default' : label, source } }; if (value !== 'on' && value !== '1') return off(`${label} is not on or off`); if (o.ceoOnly) return off('a CEO-only run writes no gate'); if (!o.outsideEnabled) return off('--no-outside'); const model = (readConfigKey('autoplan_shadow_model', env) ?? '').trim() || DEFAULT_SHADOW_MODEL; const family = modelFamily(model); const keyName = family === 'openai' ? 'OPENAI_API_KEY' : family === 'anthropic' ? 'ANTHROPIC_API_KEY' : `no API transport for family ${family}`; if (!apiKeyFor(family, env)) return off(`no API key for ${model} (${keyName})`); if (!hasPricing(model)) return off(`${model} has no price in lib/pricing.ts, so --spend-cap cannot bound it`); return { shadow: { enabled: true, model, source }, line: `SHADOW: on model=${model} (${label}); one hidden model call per gate page, counted against --spend-cap` }; } // --------------------------------------------------------------------------- // the lease // --------------------------------------------------------------------------- interface Lease { gate_rev: number; pid: number; host: string; started_at: string; deadline: string; retrospective: boolean } export interface ShadowClaim { gate_rev: number; retrospective: boolean } function readLease(out: string, rev: number): Lease | undefined { try { return JSON.parse(fs.readFileSync(leaseFile(out, rev), 'utf8')) as Lease; } catch { return undefined; } } /** Live: before its deadline plus a grace, and (on this host) its pid still running. */ function leaseLive(l: Lease, at = Date.now()): boolean { if (!(at <= Date.parse(l.deadline) + LEASE_GRACE_MS)) return false; if (l.host !== os.hostname()) return true; try { process.kill(l.pid, 0); return true; } catch (e: any) { return e.code === 'EPERM'; } } function dropLease(out: string, rev: number): void { const l = readLease(out, rev); if (l && l.pid === process.pid && l.host === os.hostname()) fs.rmSync(leaseFile(out, rev), { force: true }); } const resultRow = (rows: Array>, rev: number) => rows.find(r => (r.event === 'shadow_recorded' || r.event === 'shadow_failed') && r.gate_rev === rev); const callIds = (rows: Array>, rev: number) => rows.filter(r => r.event === 'shadow_dispatched' && r.gate_rev === rev).map(r => String(r.attempt)); /** The shadow folder: created on first use; a symlink or a non-directory is refused, never followed. */ function safeDir(out: string): string | undefined { const dir = shadowDir(out); try { const st = fs.lstatSync(dir); if (st.isSymbolicLink()) return `${dir} is a symlink`; if (!st.isDirectory()) return `${dir} is not a directory`; } catch (e: any) { if (e.code !== 'ENOENT') return `${dir}: ${e.code ?? e.message}`; fs.mkdirSync(dir, { mode: 0o700 }); } return undefined; } /** A dead lease: its dispatched calls settle unknown at their reservation (they may have been billed), the rest are released. */ function expireLease(out: string, rev: number, rows: Array>): void { const file = path.join(out, FILES.spend); const sent = new Set(callIds(rows, rev)); const open = readLedger(file).reserved.filter(r => r.attempt.startsWith(`shadow-r${rev}-`)); for (const r of open) if (sent.has(r.attempt)) settle(file, r.attempt, 'unknown'); else release(file, r.attempt); fs.rmSync(leaseFile(out, rev), { force: true }); journal(out, { event: 'shadow_expired', gate_rev: rev, settled_unknown: open.filter(r => sent.has(r.attempt)).map(r => r.attempt), released: open.filter(r => !sent.has(r.attempt)).map(r => r.attempt) }); } /** * Under the run lock: claim the current gate revision's shadow, or say why not (nothing printed when shadow is off, the * run is not at a gate, or the revision already has its result). A live lease means another writer is paying: this one * pays nothing. `retrospective` (answer's crash path) claims only a revision whose shadow_pending has no result. */ export function claimShadow(out: string, o: { retrospective?: boolean } = {}): { claim?: ShadowClaim; line?: string } { const state = readState(out); const rev = state.gate_rev; if (!state.shadow?.enabled || state.ceo_only || state.status !== 'gate_pending' || !rev) return {}; const rows = readJournal(out); if (resultRow(rows, rev) || fs.existsSync(recordFile(out, rev))) return {}; const pending = rows.some(r => r.event === 'shadow_pending' && r.gate_rev === rev); if (o.retrospective && !pending) return {}; const unsafe = safeDir(out); if (unsafe) { journal(out, { event: 'shadow_failed', gate_rev: rev, code: 'unsafe_dir', detail: unsafe }); return { line: `SHADOW: failed unsafe_dir (${unsafe})` }; } const held = readLease(out, rev); if (held && leaseLive(held)) return { line: `SHADOW: pending (gate_rev ${rev} is being recorded by pid ${held.pid} on ${held.host} until ${held.deadline}; this command pays nothing)` }; if (held || pending) expireLease(out, rev, rows); const started = Date.now(); const lease: Lease = { gate_rev: rev, pid: process.pid, host: os.hostname(), started_at: new Date(started).toISOString(), deadline: new Date(started + SHADOW_DEADLINE_MS + RELOCK_MS).toISOString(), retrospective: !!o.retrospective }; const staged = `${leaseFile(out, rev)}.${process.pid}.${crypto.randomBytes(4).toString('hex')}.tmp`; fs.writeFileSync(staged, JSON.stringify(lease)); try { fs.linkSync(staged, leaseFile(out, rev)); } catch (e: any) { if (e.code !== 'EEXIST') throw e; return { line: `SHADOW: pending (gate_rev ${rev} was claimed by another writer)` }; } finally { fs.rmSync(staged, { force: true }); } journal(out, { event: 'shadow_pending', ...lease }); return { claim: { gate_rev: rev, retrospective: lease.retrospective } }; } // --------------------------------------------------------------------------- // the page: one prompt per gate // --------------------------------------------------------------------------- interface PageFinding { id: string; severity: string; title: string } export interface PageItem { item: GateItem; findings: PageFinding[]; frozen: Record } export interface Page { prompt: string; items: PageItem[]; plan_sha256: string; prompt_sha256: string; request_sha256: string } /** Every finding the gate items can name: both voices of every phase, plus findings a deferred close carried. */ function findingIndex(state: RunState): Map { const rows: FindingRow[] = []; for (const phase of state.phase_order) { rows.push(...voiceRows(state, phase, 'native'), ...voiceRows(state, phase, 'outside')); const carried = state.phases[phase]?.deferred?.carried_file; if (carried && fs.existsSync(path.join(state.out, carried))) rows.push(...readJsonl(path.join(state.out, carried)).rows as unknown as FindingRow[]); } return new Map(rows.map(r => [r.id, r])); } /** The question as frozen in the record and sent in the prompt; p1 lists every finding the other items name. */ export function buildPage(state: RunState, list: GateList, deciderText: string): Page { const index = findingIndex(state); const own = (item: GateItem) => (item.origin?.split('|')[1] ?? '').split('+').filter(id => index.has(id)); const named = [...new Set(list.items.filter(i => i.id !== 'p1').flatMap(own))]; const items = list.items.map(item => { const findings = (item.id === 'p1' ? named : own(item)).map(id => { const f = index.get(id)!; return { id, severity: f.severity, title: f.title.split('\n')[0]!.slice(0, 200) }; }); const frozen = { id: item.id, item_class: item.item_class ?? null, kind: item.kind, title: item.title, why: item.why ?? null, phase: item.phase ?? null, options: item.options.map(o => ({ key: o.key, text: o.text, changes_plan: o.changes_plan ?? null })), recommended: item.recommended, recommended_basis: item.recommended_basis ?? null, findings }; return { item, findings, frozen }; }); const planText = fs.existsSync(state.active_plan) ? fs.readFileSync(state.active_plan, 'utf8') : ''; const plan = (extractHeading(planText, 'Implementation plan') ?? planText).slice(0, PLAN_CHARS); const slots: Record = { PLAN: plan, QUESTIONS: '```json\n' + JSON.stringify(items.map(i => i.frozen), null, 2) + '\n```', IDS: items.map(i => i.item.id).join(', ') }; const prompt = deciderText.replace(/\{\{(PLAN|QUESTIONS|IDS)\}\}/g, (_, k: string) => slots[k]!); return { prompt, items, plan_sha256: sha256(planText), prompt_sha256: sha256(deciderText), request_sha256: sha256(prompt) }; } // --------------------------------------------------------------------------- // the reply contract // --------------------------------------------------------------------------- export type Verdict = { ok: true; action: typeof SHADOW_ACTIONS[number]; pick: string; confidence: number; cites: string[]; reason: string } | { ok: false; why: string }; export interface ParsedPage { byId: Map; foreign: string[]; clean: boolean } function jsonArray(text: string): unknown { const fenced = /```(?:json)?\s*\n([\s\S]*?)```/.exec(text)?.[1]; for (const candidate of [fenced, text.trim(), text.slice(text.indexOf('['), text.lastIndexOf(']') + 1)]) { if (!candidate) continue; try { return JSON.parse(candidate); } catch { /* next candidate */ } } return undefined; } /** Exactly one valid object per requested id; a duplicate, omission or invalid field marks that item malformed, a foreign id the page. */ export function parsePage(text: string, items: PageItem[]): ParsedPage { const byId = new Map(); const parsed = jsonArray(text); if (!Array.isArray(parsed)) { for (const i of items) byId.set(i.item.id, { ok: false, why: 'no JSON array in the reply' }); return { byId, foreign: [], clean: false }; } const want = new Map(items.map(i => [i.item.id, i])); const seen = new Map(); const foreign: string[] = []; for (const obj of parsed) { const id = (obj as { id?: unknown })?.id; if (typeof id === 'string' && want.has(id)) (seen.get(id) ?? seen.set(id, []).get(id)!).push(obj); else foreign.push(typeof id === 'string' ? id : JSON.stringify(id ?? null)); } for (const [id, pi] of want) { const objs = seen.get(id) ?? []; byId.set(id, objs.length === 0 ? { ok: false, why: 'missing' } : objs.length > 1 ? { ok: false, why: 'duplicate' } : verdict(objs[0] as Record, pi)); } return { byId, foreign, clean: foreign.length === 0 && [...byId.values()].every(v => v.ok) }; } function verdict(o: Record, pi: PageItem): Verdict { const cites = Array.isArray(o.cites) ? o.cites : null; const allowed = new Set(pi.findings.map(f => f.id)); if (!SHADOW_ACTIONS.includes(o.action as typeof SHADOW_ACTIONS[number])) return { ok: false, why: `action ${JSON.stringify(o.action ?? null)} is not decide, send_back or hold` }; if (typeof o.pick !== 'string' || !pi.item.options.some(op => op.key === o.pick)) return { ok: false, why: `pick ${JSON.stringify(o.pick ?? null)} is not an option of ${pi.item.id}` }; if (typeof o.confidence !== 'number' || !Number.isFinite(o.confidence) || o.confidence < 0 || o.confidence > 1) return { ok: false, why: `confidence ${JSON.stringify(o.confidence ?? null)} is not a number from 0 to 1` }; if (!cites || cites.some(c => typeof c !== 'string' || !allowed.has(c))) return { ok: false, why: `cites ${JSON.stringify(o.cites ?? null)} name a finding outside ${pi.item.id}'s origin` }; if (typeof o.reason !== 'string' || o.reason.length > REASON_CHARS) return { ok: false, why: `reason is not a string of at most ${REASON_CHARS} characters` }; return { ok: true, action: o.action as typeof SHADOW_ACTIONS[number], pick: o.pick, confidence: o.confidence, cites: cites as string[], reason: o.reason }; } /** The 10% audit mark: sha256(salt + decision id) mod 1000 < 100. */ export function auditMark(salt: string, decision: string): boolean { return BigInt('0x' + sha256(salt + decision)) % 1000n < 100n; } // --------------------------------------------------------------------------- // the call and the record // --------------------------------------------------------------------------- export interface ShadowOutcome { status: 'recorded' | 'failed' | 'dropped'; code?: string; model?: string; usd?: number; malformed?: number; line: string } export interface RunShadowOptions { env?: NodeJS.ProcessEnv; deadlineMs?: number } class ShadowFailure extends Error { constructor(public code: string, detail: string) { super(detail); } } interface Reply { result: ApiCallResult; usd: number | 'unknown'; attempt: string } /** Classify a callModel error: no key, an HTTP status (not billed, settles 0), or a timeout or network loss (may be billed, settles unknown). */ function failureOf(e: any): { code: string; usd: number | 'unknown' } { const msg = String(e?.message ?? e); if (/^no API key/.test(msg)) return { code: 'no_key', usd: 0 }; if (/^(anthropic|openai) \d{3}:/.test(msg)) return { code: 'http', usd: 0 }; return { code: /timeout|abort/i.test(msg) || e?.name === 'AbortError' || e?.name === 'TimeoutError' ? 'timeout' : 'network', usd: 'unknown' }; } /** * Outside any lock: the page is asked once; a reply cut at the output limit is asked once more at 8,000 tokens, and a * malformed one once more as a fresh request; one 90 s deadline covers both. Each call is reserved under its own * ledger id (label shadow), journaled shadow_dispatched before the send and settled at its actual usage. Then the * lock is retaken with a bounded wait and the record published. Never throws: every failure is an outcome. */ export async function runShadow(out: string, claim: ShadowClaim, o: RunShadowOptions = {}): Promise { const env = o.env ?? process.env; const deadline = Date.now() + (o.deadlineMs ?? SHADOW_DEADLINE_MS); const rev = claim.gate_rev; const replies: Reply[] = []; try { const state = readState(out); const list = JSON.parse(fs.readFileSync(path.join(out, 'gate.json'), 'utf8')) as GateList; if (list.gate_rev !== rev) throw new ShadowFailure('stale_gate', `gate.json is gate_rev ${list.gate_rev}, the claim is ${rev}`); const model = state.shadow!.model!; const page = buildPage(state, list, fs.readFileSync(shadowPromptFile(), 'utf8')); writeAtomic(path.join(shadowDir(out), `r${rev}.prompt.md`), page.prompt); const inputTokens = Math.ceil(Buffer.byteLength(page.prompt) / 3); const estimate = (maxTokens: number) => estimateCostUsd({ input: inputTokens, output: maxTokens }, model); const ledgerFile = path.join(out, FILES.spend); const headroom = summarize(readLedger(ledgerFile, state.spend_cap_usd)).headroom_usd; const worst = estimate(SHADOW_MAX_TOKENS) + estimate(SHADOW_RAISED_MAX_TOKENS); if (headroom !== null && headroom < worst) throw new ShadowFailure('budget', `headroom ${headroom.toFixed(4)} < ${worst.toFixed(4)} (one call plus its retry)`); const ask = async (maxTokens: number): Promise => { const remaining = deadline - Date.now(); if (remaining <= 0) throw new ShadowFailure('timeout', `the ${(o.deadlineMs ?? SHADOW_DEADLINE_MS) / 1000} s deadline passed before the call`); const attempt = `shadow-r${rev}-${callIds(readJournal(out), rev).length + 1}`; try { reserve(ledgerFile, attempt, estimate(maxTokens), { cap: readState(out).spend_cap_usd, label: 'shadow' }); } catch (e: any) { throw new ShadowFailure('budget', e.message); } journal(out, { event: 'shadow_dispatched', gate_rev: rev, attempt, model, max_tokens: maxTokens, estimate_usd: estimate(maxTokens), retrospective: claim.retrospective }); let result: ApiCallResult; try { result = await callModel({ model, prompt: page.prompt, timeoutMs: remaining, env, maxTokens, receipt: { sink: SHADOW_SINK, consent: SHADOW_CONSENT, payloadClass: 'plan-and-gate' } }); } catch (e: any) { const f = failureOf(e); settle(ledgerFile, attempt, f.usd); throw new ShadowFailure(f.code, String(e?.message ?? e)); } const { input_tokens: inTok, output_tokens: outTok } = result.usage; const usd: number | 'unknown' = typeof inTok === 'number' && typeof outTok === 'number' ? estimateCostUsd({ input: inTok, output: outTok }, model) : 'unknown'; settle(ledgerFile, attempt, usd); writeAtomic(path.join(shadowDir(out), `${attempt}.reply.txt`), result.text); const reply = { result, usd, attempt }; replies.push(reply); return reply; }; const truncated = (r: Reply) => r.result.stop_reason === 'max_tokens' || r.result.stop_reason === 'length'; const first = await ask(SHADOW_MAX_TOKENS); let parsed: ParsedPage[]; if (truncated(first)) { const raised = await ask(SHADOW_RAISED_MAX_TOKENS); if (truncated(raised)) throw new ShadowFailure('truncated', `the reply was cut at ${SHADOW_RAISED_MAX_TOKENS} output tokens`); parsed = [parsePage(raised.result.text, page.items)]; } else { parsed = [parsePage(first.result.text, page.items)]; if (!parsed[0]!.clean) { try { const retry = await ask(SHADOW_MAX_TOKENS); if (!truncated(retry)) parsed.push(parsePage(retry.result.text, page.items)); } catch (e) { if (!(e instanceof ShadowFailure)) throw e; } } } return publish(out, claim, { rows: shadowRows(out, state, list, page, parsed, replies), model, replies }); } catch (e: any) { const failure = e instanceof ShadowFailure ? e : new ShadowFailure('internal', String(e?.message ?? e)); return publish(out, claim, { failure, replies }); } } const usdOf = (replies: Reply[]) => (replies.some(r => r.usd === 'unknown') ? 'unknown' as const : Math.round(replies.reduce((s, r) => s + (r.usd as number), 0) * 1e6) / 1e6); /** One row per gate item: the frozen question, what the model predicted (the retry's verdict when valid, else the first's), the free rule's pick and the audit mark. */ function shadowRows(out: string, state: RunState, list: GateList, page: Page, parsed: ParsedPage[], replies: Reply[]): Array> { const at = now(); const usage = replies.reduce((u, r) => ({ input_tokens: u.input_tokens + (r.result.usage.input_tokens ?? 0), output_tokens: u.output_tokens + (r.result.usage.output_tokens ?? 0) }), { input_tokens: 0, output_tokens: 0 }); return page.items.map(({ item, frozen }) => { const verdicts = parsed.map(p => p.byId.get(item.id)!); const k = verdicts.map((v, i) => [v, i] as const).reverse().find(([v]) => v.ok); const decision = decisionId(list.run, list.gate_rev, item.id); const base = { schema_version: 1, run: list.run, gate_rev: list.gate_rev, decision, item: item.id, item_class: item.item_class ?? null, phase: item.phase ?? null, ceo_eligible: item.item_class !== 'finding' && (item.id === 'p1' || item.phase === 'ceo'), item_frozen: frozen, rec_hash: recHash(list), rec_rev: list.rec_rev ?? 0, plan_sha256: page.plan_sha256, prompt_version: SHADOW_PROMPT_VERSION, prompt_sha256: page.prompt_sha256, request_sha256: page.request_sha256, model: replies[replies.length - 1]?.result.model ?? state.shadow?.model, gstack_version: GSTACK_VERSION, rule_pick: ruleKept(item), retrospective: false, page_usage: usage, page_usd: usdOf(replies), recorded_at: at, }; if (!k) return { ...base, status: 'malformed', malformed: (verdicts[verdicts.length - 1] as { why: string }).why, action: null, pick: null, confidence: null, cites: [], reason: null, audit: false }; const v = k[0] as Extract; return { ...base, status: 'recorded', action: v.action, pick: v.pick, confidence: v.confidence, cites: v.cites, reason: v.reason, reply: k[1] + 1, audit: v.action === 'decide' && auditMark(saltOf(out), decision) }; }); } /** The per-run audit salt, drawn once into the shadow folder (never run.json). */ function saltOf(out: string): string { try { return fs.readFileSync(saltFile(out), 'utf8').trim(); } catch { /* first use */ } try { fs.writeFileSync(saltFile(out), crypto.randomBytes(16).toString('hex') + '\n', { flag: 'wx', mode: 0o600 }); } catch (e: any) { if (e.code !== 'EEXIST') throw e; } return fs.readFileSync(saltFile(out), 'utf8').trim(); } function writeAtomic(file: string, text: string): void { const tmp = `${file}.${process.pid}.${crypto.randomBytes(4).toString('hex')}.tmp`; fs.writeFileSync(tmp, text); fs.renameSync(tmp, file); } /** Under the run lock (a bounded wait, never RUN_LOCKED at once): link the record, journal the result, drop the lease. */ function publish(out: string, claim: ShadowClaim, r: { rows?: Array>; model?: string; failure?: ShadowFailure; replies: Reply[] }): ShadowOutcome { const rev = claim.gate_rev; const usd = usdOf(r.replies); const attempts = r.replies.map(x => x.attempt); try { return withRunLock(out, () => { const rows = readJournal(out); if (fs.existsSync(recordFile(out, rev)) || resultRow(rows, rev)) { dropLease(out, rev); return { status: 'dropped' as const, line: `SHADOW: dropped (gate_rev ${rev} already has its result; this call's cost stays in the ledger)` }; } if (r.failure) { journal(out, { event: 'shadow_failed', gate_rev: rev, code: r.failure.code, detail: r.failure.message.slice(0, 500), attempts, usd, retrospective: claim.retrospective }); dropLease(out, rev); return { status: 'failed' as const, code: r.failure.code, line: `SHADOW: failed ${r.failure.code}` }; } const salted: Array> = r.rows!.map(row => ({ ...row, retrospective: claim.retrospective })); const text = salted.map(row => JSON.stringify(row)).join('\n') + '\n'; const final = recordFile(out, rev); for (const e of fs.readdirSync(shadowDir(out))) if (e.startsWith(`ceo-only.r${rev}.jsonl.`) && e.endsWith('.tmp')) fs.rmSync(path.join(shadowDir(out), e), { force: true }); const tmp = `${final}.${process.pid}.${crypto.randomBytes(4).toString('hex')}.tmp`; fs.writeFileSync(tmp, text, { flag: 'wx' }); try { fs.linkSync(tmp, final); } catch (e: any) { if (e.code !== 'EEXIST') throw e; dropLease(out, rev); return { status: 'dropped' as const, line: `SHADOW: dropped (gate_rev ${rev} was recorded by another writer)` }; } finally { fs.rmSync(tmp, { force: true }); } const malformed = salted.filter(row => row.status === 'malformed').length; journal(out, { event: 'shadow_recorded', gate_rev: rev, sha256: sha256(text), model: r.model, usd, items: salted.length, malformed, attempts, retrospective: claim.retrospective }); dropLease(out, rev); const usdText = usd === 'unknown' ? 'unknown' : usd.toFixed(4); return { status: 'recorded' as const, model: r.model, usd: usd === 'unknown' ? undefined : usd, malformed, line: `SHADOW: recorded model=${r.model} usd=${usdText}${malformed ? ` malformed=${malformed}/${salted.length}` : ''}${claim.retrospective ? ' retrospective' : ''}` }; }, { waitMs: RELOCK_MS }); } catch (e: any) { const code = e instanceof RunError && e.code === 'RUN_LOCKED' ? 'locked' : 'internal'; try { withRunLock(out, () => { if (!resultRow(readJournal(out), rev) && !fs.existsSync(recordFile(out, rev))) journal(out, { event: 'shadow_failed', gate_rev: rev, code, detail: String(e?.message ?? e).slice(0, 500), attempts, usd, retrospective: claim.retrospective }); dropLease(out, rev); }, { waitMs: 2_000 }); } catch { /* still locked: the lease expires and the next command settles the revision */ } return { status: 'failed', code, line: `SHADOW: failed ${code} (${e?.message ?? e}); the gate stands` }; } } /** * Before `answer` takes the lock: wait for a live lease to finish (up to its deadline), then, when the revision has a * shadow_pending and still no result (a crash), record it first with every row retrospective. Never throws. */ export async function shadowBeforeAnswer(out: string, o: RunShadowOptions = {}): Promise { try { const state = readState(out); const rev = state.gate_rev; if (!state.shadow?.enabled || state.status !== 'gate_pending' || !rev) return []; for (let l = readLease(out, rev); l && leaseLive(l); l = readLease(out, rev)) await sleep(200); const rows = readJournal(out); if (resultRow(rows, rev) || fs.existsSync(recordFile(out, rev)) || !rows.some(r => r.event === 'shadow_pending' && r.gate_rev === rev)) return []; const { claim, line } = withRunLock(out, () => claimShadow(out, { retrospective: true }), { waitMs: RELOCK_MS }); if (!claim) return line ? [line] : []; return [(await runShadow(out, claim, o)).line]; } catch (e: any) { return [`SHADOW: failed internal (${e?.message ?? e}); the answer proceeds`]; } }