diff --git a/.gitleaks.toml b/.gitleaks.toml index 60cabd2423..95360d5fcb 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -22,6 +22,11 @@ regexes = [ # signature. generic-api-key matched it, and main's Secret Scan was red on # every push from 2026-10-16. The exact literal only, so a real token still fails. '''eyJhbGciOiJSUzI1NiJ9\.jwt''', + # A fake OpenRouter key in packages/seats/test/supervisor.test.js (#2701): the + # scrub test needs a value shaped like a real key so it can prove the key never + # reaches the record. generic-api-key matched it in a commit the scan reads, so + # an inline allow on today's line would not clear it. The exact literal only. + '''sk-or-v1-abcdefabcdefabcdef''', # The namespaced cloudflared ExternalSecret has `- secretKey: credentials.json` # (#2439) — a Secret KEY NAME (what the tunnel credentials.json is stored under), not # a value. gitleaks' kubernetes-secret-yaml rule captures `secretKey: credentials` (it diff --git a/packages/seats/README.md b/packages/seats/README.md index 8f86eff12d..990cbab2c0 100644 --- a/packages/seats/README.md +++ b/packages/seats/README.md @@ -1,6 +1,6 @@ # packages/seats -The adapters that run an agent CLI as a supervised seat: one prompt in, one CLI run, one reply out. The package holds the contract, the registry, the failure naming, the spawn observation, and the first adapters (`stub`, `hermes`). +The adapters that run an agent CLI as a supervised seat: one prompt in, one CLI run, one reply out. The package holds the contract, the registry, the failure naming, the spawn observation, the run supervisor (`superviseRun`: a turn with no progress signal for `stallMs` is traced as stuck, cancelled, and settled even if its adapter ignores the cancel), and the first adapters (`stub`, `hermes`). It is private and is not published on its own. `@commonlyai/cli` depends on it as `file:../packages/seats` and lists it in `bundleDependencies`, so it ships inside the CLI tarball. diff --git a/packages/seats/docs/record-v1.md b/packages/seats/docs/record-v1.md index 335fd202d8..26ab8f459f 100644 --- a/packages/seats/docs/record-v1.md +++ b/packages/seats/docs/record-v1.md @@ -333,7 +333,7 @@ built in Go runs its tests with `-race` in CI. | kind | data | default actor | |---|---|---| | `credential.granted` | `{names: [NAME]}` | `system` | -| `credential.denied` | `{name, reason}`, where `reason` is `not_granted` or `unresolved` | `system` | +| `credential.denied` | `{name, reason}`, where `reason` is `not_granted`, `unresolved` or `too_short` (a value under 8 characters) | `system` | These cover the environment credentials a run is started with. Values never appear here: `NAME` is an environment variable name matching @@ -434,11 +434,21 @@ What's new or changed in `EVENT_TYPES`: and a writer refuses one that is not name-shaped. - Agent-authored text, such as message chunks and tool titles, belongs to the agent and could repeat a value the agent holds. A writer that knows - the run's credential values replaces every occurrence of 8 characters or - more with `[redacted]` before the line is hashed. + the run's credential values replaces every occurrence of each, whatever + its length, with `[redacted]` before the line is hashed. A writer does + not grant a credential shorter than 8 characters (`credential.denied`, + `too_short`). + - Message and thought chunks are scrubbed per stream, not per line, so a + value split across chunks is caught. A writer holds back a chunk's tail + while it could begin a known value, and writes it with the stream's next + chunk. If the stream ends first, the held tail is a partial value, so the + writer writes `[redacted]` as a chunk of its own in its place, before the + next `attempt.started`, `attempt.settled` or `run.settled`. A short + ordinary tail that happened to begin a value is redacted too. A held tail + lives only in the writer's memory: a writer that crashes before the next + write loses it, and never writes it. - That replacement covers known values only. A value the writer was not - given, or one the agent transformed (base64, split across chunks), passes - through. + given, or one the agent transformed (base64, say), passes through. 2. **Readers ignore unknown fields and skip unknown kinds.** A reader that meets a kind it doesn't know still checks the chain over that line. 3. **`v` bumps only on a breaking change:** diff --git a/packages/seats/src/index.js b/packages/seats/src/index.js index fcfd9a0824..6f0be90d3c 100644 --- a/packages/seats/src/index.js +++ b/packages/seats/src/index.js @@ -14,3 +14,4 @@ export { RECORD_VERSION, RECORD_SCHEMA, KINDS, canonicalLine, appendEntry, verifyRecord, readRecord, hashLine, toolSurfaceOf, argsDigestOf, } from './record/format.js'; export { toOtlpJson } from './record/otel.js'; +export { superviseRun, SUPERVISOR_DEFAULTS, DEFAULT_RETRY_ON } from './supervisor.js'; diff --git a/packages/seats/src/runtime-observation.js b/packages/seats/src/runtime-observation.js index 59eb63ee8c..e9c7bfd48d 100644 --- a/packages/seats/src/runtime-observation.js +++ b/packages/seats/src/runtime-observation.js @@ -19,9 +19,11 @@ export const controlsFromArguments = (args, { modelFlag, effortFlag, valueFlags // Called at the child's successful OS spawn event, not when a requested // configuration is saved or when a turn is merely queued. Never send argv. +// The child rides along as a second argument so a supervisor can read its +// output as a liveness signal and kill it; a reporter reads only the first. export const observeSpawn = (child, onSpawn, controls) => { if (typeof onSpawn !== 'function') return; child.once('spawn', () => { - try { onSpawn({ model: controls.model || null, effort: controls.effort || null, at: new Date().toISOString() }); } catch { /* diagnostic only */ } + try { onSpawn({ model: controls.model || null, effort: controls.effort || null, at: new Date().toISOString() }, child); } catch { /* diagnostic only */ } }); }; diff --git a/packages/seats/src/supervisor.js b/packages/seats/src/supervisor.js new file mode 100644 index 0000000000..513a420432 --- /dev/null +++ b/packages/seats/src/supervisor.js @@ -0,0 +1,287 @@ +/** + * The run supervisor: one prompt, one traced run, and a guarantee that the run + * settles. A turn that hangs is named `stuck` in the trace, cancelled, and if + * the adapter will not let go, abandoned and settled anyway. Nothing this + * module starts is allowed to sit in "running" forever. + * + * Liveness is read from progress signals, not from the reply: any trace event + * the adapter emits, and any byte on the stdout/stderr of a child the adapter + * reports through `ctx.onSpawn`. A run that is busy but quiet past the window + * is indistinguishable from a hung one, which is why the window is long. + */ +import { RunFailure, runFailure, failureKindOf, FAILURE_KINDS } from './failures.js'; +import { redactKnownCredentials, scrubAdapterFailure, spawnCredentials } from './upstream-refusal.js'; + +export const SUPERVISOR_DEFAULTS = Object.freeze({ + // Longer than the quiet stretches a healthy turn has: a long-context first + // token, or a tool (an install, a test suite) whose output the CLI buffers. + stallMs: 120_000, + // Covers an adapter's own SIGTERM -> 5 s -> SIGKILL ladder with room to spare. + cancelGraceMs: 10_000, + // The turn budget the run loop already enforces (COMMONLY_AGENT_RUN_TIMEOUT_MS's + // fallback), so supervising a turn does not shorten or lengthen it. + timeoutMs: 15 * 60 * 1000, +}); + +export const DEFAULT_RETRY_ON = Object.freeze(['stuck', 'exited', 'timeout']); + +// setTimeout overflows past this and fires immediately. +const MAX_TIMER_MS = 2_147_483_647; +// The supervisor's own events; an adapter writing them would forge the run's outcome. +const SUPERVISOR_EVENTS = new Set(['run.started', 'run.settled']); + +const windowOf = (value, fallback) => { + if (value === undefined || value === null) return fallback; + const number = Number(value); + if (number === 0 || number === Infinity) return null; // explicitly off + if (!Number.isFinite(number) || number < 0) throw new Error(`supervisor window must be a non-negative number, got ${value}`); + return Math.min(number, MAX_TIMER_MS); +}; + +const resolvePolicy = (policy = {}) => { + const retry = policy.retry || {}; + return { + stallMs: windowOf(policy.stallMs, SUPERVISOR_DEFAULTS.stallMs), + cancelGraceMs: windowOf(policy.cancelGraceMs, SUPERVISOR_DEFAULTS.cancelGraceMs) ?? 0, + timeoutMs: windowOf(policy.timeoutMs, SUPERVISOR_DEFAULTS.timeoutMs), + retryMax: Math.max(0, Math.floor(Number(retry.max) || 0)), + retryOn: Array.isArray(retry.on) ? retry.on : DEFAULT_RETRY_ON, + onStalled: typeof policy.onStalled === 'function' ? policy.onStalled : null, + }; +}; + +const isChild = (value) => !!value && typeof value.kill === 'function' && typeof value.once === 'function'; +const isAlive = (child) => child.exitCode === null && child.signalCode === null; + +// The retryable flag in FAILURE_KINDS outranks the caller's list: a refusal +// listed in `on` is still never re-sent. +const shouldRetry = (failure, attempt, settings) => { + const kind = failureKindOf(failure); + return attempt <= settings.retryMax + && !!kind && FAILURE_KINDS[kind].retryable + && settings.retryOn.includes(kind) + && !failure.adapterIgnoredCancel; +}; + +const outcomeOf = (failure) => { + if (!failure) return 'ok'; + const kind = failureKindOf(failure); + if (kind === 'stuck') return 'stuck'; + if (kind === 'cancelled') return 'cancelled'; + return 'failed'; +}; + +// An adapter that threw a bare Error broke the RunFailure rule; its text is +// still scrubbed before it reaches the trace, and it carries no kind rather +// than a guessed one. A RunFailure's detail keeps its shape but loses any +// credential this spawn was handed, whoever built it. +const failureData = (failure, credentials) => (failure instanceof RunFailure + ? { kind: failure.kind, detail: redactKnownCredentials(failure.detail, credentials) } + : { kind: null, detail: scrubAdapterFailure(String(failure?.message ?? failure ?? ''), { credentials }).detail }); + +const sessionIdOf = (failure) => { + const id = failure?.sessionId ?? failure?.newSessionId; + return typeof id === 'string' && id ? id : null; +}; + +const ignoredCancel = (reason, graceMs) => { + const failure = new RunFailure(reason.kind, `${reason.detail}; the adapter did not settle within ${graceMs}ms of cancellation`); + failure.adapterIgnoredCancel = true; + return failure; +}; + +const callerCancel = (signal) => (failureKindOf(signal.reason) === 'cancelled' + ? signal.reason : runFailure('cancelled', 'cancelled by the caller')); + +/** One attempt. Resolves { result } or { failure }; never rejects, never stays pending. */ +const runAttempt = ({ adapter, prompt, ctx, run, settings, attempt, sessionId, external }) => new Promise((resolve) => { + const controller = new AbortController(); + const children = new Set(); + let live = true; + let abortedBy = null; + let lastSignal = Date.now(); + let stallTimer = null; + let wallTimer = null; + let graceTimer = null; + + const touch = () => { if (live) lastSignal = Date.now(); }; + + const killChildren = () => { + let killed = 0; + for (const child of children) { + if (!isAlive(child)) continue; + try { child.kill('SIGKILL'); killed += 1; } catch { /* already gone */ } + } + return killed; + }; + + const onExternal = () => cancel(callerCancel(external), { recovery: false }); + + const finish = (outcome) => { + if (!live) return; + live = false; + clearTimeout(stallTimer); + clearTimeout(wallTimer); + clearTimeout(graceTimer); + external?.removeEventListener?.('abort', onExternal); + // An adapter that honoured the abort can still leave its child behind. + if (abortedBy && !outcome.failure?.adapterIgnoredCancel && killChildren() > 0) { + run.emit('liveness.recovery', { action: 'kill', attempt }); + } + resolve(outcome); + }; + + function cancel(reason, { recovery = true } = {}) { + if (abortedBy || !live) return; + abortedBy = reason; + clearTimeout(stallTimer); + clearTimeout(wallTimer); + if (recovery) run.emit('liveness.recovery', { action: 'cancel', attempt }); + controller.abort(reason); + graceTimer = setTimeout(() => { + if (!live) return; + run.emit('liveness.recovery', { action: 'kill', attempt }); + killChildren(); + finish({ failure: ignoredCancel(reason, settings.cancelGraceMs) }); + }, settings.cancelGraceMs); + } + + const checkStall = () => { + if (!live || abortedBy) return; + const idleMs = Date.now() - lastSignal; + if (idleMs < settings.stallMs) { + stallTimer = setTimeout(checkStall, settings.stallMs - idleMs); + return; + } + const windowMs = settings.stallMs; + run.emit('liveness.stalled', { idleMs, windowMs }); + if (settings.onStalled) { + // A host's notifier failing must not keep the run from recovering. + try { + Promise.resolve(settings.onStalled({ runId: run.runId, adapter: adapter.name, attempt, idleMs, windowMs })).catch(() => {}); + } catch { /* notifier only */ } + } + cancel(runFailure('stuck', `no progress signal for ${idleMs}ms (window ${windowMs}ms)`)); + }; + + // Counts as progress only while this attempt is live, so a cancelled + // attempt's late output cannot keep the next one alive or reach the trace. + // The emit options (the entry's actor) pass through unchanged. `redact` is + // offered only when the run has one, so prepareRunEnv cannot hand its + // values to a no-op; it is not gated on `live`, since adding a value to + // scrub is never unsafe. + const trace = { + emit(type, data, options) { + if (!live || SUPERVISOR_EVENTS.has(type)) return null; + touch(); + return run.emit(type, data, options); + }, + ...(typeof run.redact === 'function' ? { redact: (values) => run.redact(values) } : {}), + }; + + // Wraps the caller's onSpawn: every argument reaches it unchanged. + const onSpawn = (...args) => { + if (live) { + touch(); + for (const child of args.filter(isChild)) { + children.add(child); + child.stdout?.on?.('data', touch); + child.stderr?.on?.('data', touch); + } + } + return typeof ctx.onSpawn === 'function' ? ctx.onSpawn(...args) : undefined; + }; + + if (external?.aborted) { + abortedBy = callerCancel(external); + finish({ failure: abortedBy }); + return; + } + external?.addEventListener?.('abort', onExternal, { once: true }); + if (settings.stallMs) stallTimer = setTimeout(checkStall, settings.stallMs); + if (settings.timeoutMs) { + wallTimer = setTimeout(() => cancel(runFailure('timeout', `exceeded the ${settings.timeoutMs}ms wall-clock budget`)), settings.timeoutMs); + } + + let pending; + try { + pending = Promise.resolve(adapter.spawn(prompt, { + ...ctx, ...(sessionId ? { sessionId } : {}), trace, signal: controller.signal, onSpawn, + })); + } catch (error) { + pending = Promise.reject(error); + } + // Once the supervisor has aborted, the abort's reason is the outcome, even + // if the adapter answers inside the grace window: the trace already says why. + pending.then( + (result) => finish(abortedBy ? { failure: abortedBy } : { result }), + (error) => finish({ failure: abortedBy || error }), + ); +}); + +/** + * Run one prompt under liveness supervision. + * + * @param {object} args + * @param {import('./contract.js').Adapter} args.adapter + * @param {string} args.prompt + * @param {object} [args.ctx] passed to adapter.spawn; `trace`, `signal` and `onSpawn` are wrapped. + * `ctx.signal`, when given, cancels the run (outcome `cancelled`). + * @param {ReturnType} args.store + * @param {object} [args.policy] + * @param {number} [args.policy.stallMs] no-progress window; 0 turns it off + * @param {number} [args.policy.cancelGraceMs] how long a cancelled adapter has to settle + * @param {number} [args.policy.timeoutMs] wall-clock budget per attempt; 0 turns it off + * @param {{max: number, on?: string[]}} [args.policy.retry] off unless max > 0 + * @param {(info: {runId, adapter, attempt, idleMs, windowMs}) => void} [args.policy.onStalled] + * @param {string} [args.runId] + * @param {string} [args.label] + * @returns {Promise<{ runId: string, outcome: string, attempts: number, result?: object, failure?: Error }>} + */ +export const superviseRun = async ({ adapter, prompt, ctx = {}, store, policy = {}, runId, label } = {}) => { + if (!adapter || typeof adapter.spawn !== 'function') throw new Error('superviseRun needs an adapter with spawn()'); + if (!store || typeof store.startRun !== 'function') throw new Error('superviseRun needs a trace store'); + const settings = resolvePolicy(policy); + const run = store.startRun({ runId, adapter: adapter.name, label }); + // The runtime token rides in ctx, not through the valet, so the store would + // not otherwise know to scrub it from agent text. + if (typeof run.redact === 'function' && typeof ctx.runtimeToken === 'string' && ctx.runtimeToken) { + run.redact([ctx.runtimeToken]); + } + const startedAt = Date.now(); + const external = ctx.signal; + let attempt = 0; + let sessionId = ctx.sessionId || null; + let last = { failure: null }; + + try { + for (;;) { + attempt += 1; + run.emit('run.started', { adapter: adapter.name, ...(sessionId ? { sessionId } : {}), attempt }); + last = await runAttempt({ adapter, prompt, ctx, run, settings, attempt, sessionId, external }); + if (!last.failure || external?.aborted || !shouldRetry(last.failure, attempt, settings)) break; + sessionId = sessionIdOf(last.failure) || sessionId; + run.emit('liveness.recovery', { action: 'retry', attempt: attempt + 1 }); + } + } catch (error) { + // The supervisor's own fault (a trace write failing, say) still settles the run. + last = { failure: error }; + } + + const outcome = outcomeOf(last.failure); + const credentials = spawnCredentials(ctx, [adapter.providerKeyEnv].filter(Boolean)); + try { + run.emit('run.settled', { + outcome, + ...(last.failure ? { failure: failureData(last.failure, credentials) } : {}), + durationMs: Date.now() - startedAt, + attempts: attempt, + }); + } catch { /* the outcome is still returned */ } + return { + runId: run.runId, + outcome, + attempts: attempt, + ...(last.failure ? { failure: last.failure } : { result: last.result }), + }; +}; diff --git a/packages/seats/src/trace/events.js b/packages/seats/src/trace/events.js index 2069eac6d0..0ababca921 100644 --- a/packages/seats/src/trace/events.js +++ b/packages/seats/src/trace/events.js @@ -23,7 +23,7 @@ export const EVENT_TYPES = Object.freeze([ 'liveness.stalled', // { idleMs, windowMs } 'liveness.recovery', // { action: 'cancel'|'kill'|'retry', attempt } 'credential.granted', // { names: [...] } names only, never values - 'credential.denied', // { name, reason?: 'not_granted'|'unresolved' } + 'credential.denied', // { name, reason?: 'not_granted'|'unresolved'|'too_short' } 'process.exited', // { code, signal } // Record kinds with no earlier trace name (docs/record-v1.md has the full data). 'task.claimed', // { leaseUntil? } mirrors the board's claim; the board stays the source of truth diff --git a/packages/seats/src/trace/store.js b/packages/seats/src/trace/store.js index 4608844385..5049f8d82a 100644 --- a/packages/seats/src/trace/store.js +++ b/packages/seats/src/trace/store.js @@ -14,7 +14,6 @@ import { join } from 'node:path'; import { appendEntry, hashLine, kindOfType, readRecord, traceTypeOf, validateActor, verifyRecord, BOARD_ACTOR, } from '../record/format.js'; -import { redactKnownCredentials } from '../upstream-refusal.js'; const RUN_ID = /^[A-Za-z0-9_-]{6,64}$/; const readJsonl = (path) => (existsSync(path) @@ -23,10 +22,22 @@ const readJsonl = (path) => (existsSync(path) }) : []); -/** Known credential values never reach a line, even inside agent text. */ +/** + * Known credential values never reach a line, even inside agent text. Every + * registered value is replaced, whatever its length: the eight-character + * floor in redactKnownCredentials belongs to the refusal-text reader, not to + * the record. `secrets` is kept longest first, so a value that contains a + * shorter one is replaced whole rather than in pieces. + */ +const scrubText = (text, secrets) => { + let out = text; + for (const value of secrets) out = out.split(value).join('[redacted]'); + return out; +}; + const scrub = (value, secrets) => { if (secrets.length === 0) return value; - if (typeof value === 'string') return redactKnownCredentials(value, secrets); + if (typeof value === 'string') return scrubText(value, secrets); if (Array.isArray(value)) return value.map((item) => scrub(item, secrets)); if (value && typeof value === 'object') { return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, scrub(item, secrets)])); @@ -34,6 +45,34 @@ const scrub = (value, secrets) => { return value; }; +/** + * Streamed agent text arrives in deltas, so a value can be split across two or + * more of them and match none on its own. Each stream (kind + actor) is + * scrubbed over a rolling window instead: the held tail is joined to the next + * delta, the join is scrubbed, and the longest suffix of the result that + * could still begin a known value is held back again. What is written is the + * rest. + * + * Why that is enough: an occurrence that started inside written text and ran + * past its end would make the written part plus the held tail a proper prefix + * of that value, a longer such suffix than the one held. So no value spans a + * written boundary, and each written piece is itself scrubbed. The tail is + * shorter than the longest value, and only as long as an actual prefix match. + */ +const STREAMED_KINDS = new Set(['agent.message_chunk', 'agent.thought_chunk']); +// The held text is written out before these, so it stays in its own attempt +// and is never lost at settle, whichever path (ok, failure, cancel, kill) got there. +const FLUSH_BEFORE = new Set(['attempt.started', 'attempt.settled', 'run.settled']); + +const heldLength = (text, secrets) => { + const longest = secrets.length ? secrets[0].length : 0; + for (let start = Math.max(0, text.length - longest + 1); start < text.length; start += 1) { + const tail = text.slice(start); + if (secrets.some((value) => value.length > tail.length && value.startsWith(tail))) return text.length - start; + } + return 0; +}; + /** What readRun has always returned, plus the envelope: `type` is the trace name. */ const asEvent = (entry) => (entry.kind ? { ...entry, type: traceTypeOf(entry.kind) } : entry); @@ -70,11 +109,12 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => { let seq = 0; let settled = false; let broken = false; - const secrets = []; + const secrets = []; // longest first; see scrubText + const streams = new Map(); // stream key → { kind, actor, held } const grants = new Map(); // open grants: grantId → { parentGrantId, binding, expiresAt } const startedAt = now(); - const write = (kind, data, actor) => { + const writeLine = (kind, data, actor) => { const line = appendEntry(last, { runId, taskId, podId, ts: now().toISOString(), kind, data: scrub(data, secrets), actor, }, { runActor }); @@ -97,6 +137,31 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => { return asEvent(entry); }; + /** A streamed delta's text, less a tail that could begin a known value. */ + const windowed = (kind, data, actor) => { + if (!STREAMED_KINDS.has(kind) || typeof data?.text !== 'string') return data; + const key = JSON.stringify([kind, actor ?? null]); + const joined = scrubText(`${streams.get(key)?.held ?? ''}${data.text}`, secrets); + const cut = joined.length - heldLength(joined, secrets); + if (cut < joined.length) streams.set(key, { kind, actor, held: joined.slice(cut) }); + else streams.delete(key); + return { ...data, text: joined.slice(0, cut) }; + }; + + const write = (kind, data, actor) => writeLine(kind, windowed(kind, data, actor), actor); + + /** + * The stream has ended, so each held tail can no longer complete a value. + * A tail is held only while it is a prefix of a known value, so it is part + * of a credential: write the marker in its place, never the prefix. + */ + const flushStreams = () => { + for (const [key, { kind, actor }] of streams) { + streams.delete(key); + writeLine(kind, { text: '[redacted]' }, actor); + } + }; + const revoke = (grantIds, reason, actor) => (grantIds.length ? write('grant.revoked', { grantIds, reason }, actor) : null); const emit = (type, data = {}, options = {}) => { @@ -104,6 +169,7 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => { if (!kind) throw new Error(`unknown trace event type: ${type}`); if (settled) return null; if (broken) throw new Error(`run ${runId}: the record is broken after a failed write`); + if (FLUSH_BEFORE.has(kind)) flushStreams(); const event = write(kind, data, options?.actor); if (kind === 'task.done') { // The board closed the task: every grant bound to it ends, in one write. @@ -142,8 +208,10 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => { return revoke([...grants].filter(([, g]) => g.expiresAt && Date.parse(g.expiresAt) <= cutoff).map(([id]) => id), 'expired'); }; + /** Every non-empty string is scrubbed from here on, whatever its length. */ const redact = (values = []) => { - for (const value of values) if (typeof value === 'string' && !secrets.includes(value)) secrets.push(value); + for (const value of values) if (typeof value === 'string' && value && !secrets.includes(value)) secrets.push(value); + secrets.sort((a, b) => b.length - a.length); }; append(indexPath, { diff --git a/packages/seats/src/upstream-refusal.js b/packages/seats/src/upstream-refusal.js index 598edb8a51..12f7f30cca 100644 --- a/packages/seats/src/upstream-refusal.js +++ b/packages/seats/src/upstream-refusal.js @@ -74,6 +74,9 @@ export const bodyFromError = (value) => { return text.replace(STATUS_PREFIX, '').trim(); }; +/** The shortest value the refusal-text reader treats as a credential. */ +export const MIN_CREDENTIAL_LENGTH = 8; + /** * Replace credential values this spawn was handed with a marker. * @@ -82,11 +85,16 @@ export const bodyFromError = (value) => { * value is in the lines around the secret. Values shorter than eight characters * are ignored: redacting a one-character "credential" would corrupt ordinary * prose, and no issued credential is that short. + * + * The floor belongs to this reader only. The trace store scrubs every value + * registered with it, whatever its length, and the valet refuses to grant a + * credential shorter than this (credential_denied), so no granted value is + * ever below the floor. */ export const redactKnownCredentials = (text, credentials = []) => { let out = String(text ?? ''); for (const value of credentials) { - if (typeof value !== 'string' || value.length < 8) continue; + if (typeof value !== 'string' || value.length < MIN_CREDENTIAL_LENGTH) continue; out = out.split(value).join('[redacted]'); } return out; @@ -350,5 +358,5 @@ export const spawnCredentials = (ctx = {}, envNames = []) => { return [ ctx.runtimeToken, ...envNames.map((name) => env[name]), - ].filter((value) => typeof value === 'string' && value.length >= 8); + ].filter((value) => typeof value === 'string' && value.length >= MIN_CREDENTIAL_LENGTH); }; diff --git a/packages/seats/src/valet/env.js b/packages/seats/src/valet/env.js index f819da8118..fe37dfefbe 100644 --- a/packages/seats/src/valet/env.js +++ b/packages/seats/src/valet/env.js @@ -16,6 +16,7 @@ import { homedir, tmpdir } from 'node:os'; import { delimiter, join, resolve, sep } from 'node:path'; import { traceOf } from '../trace/events.js'; import { runFailure } from '../failures.js'; +import { MIN_CREDENTIAL_LENGTH } from '../upstream-refusal.js'; import { validateGrant } from './grant.js'; // Same spirit as the goose adapter's SAFE_ENV: locale, terminal and TLS trust @@ -103,8 +104,10 @@ const grantIssued = (meta) => { * * `require` lists credential names the adapter cannot run without; one the * grant does not name fails the run as credential_denied before any spawn. - * A named credential that cannot be resolved fails it the same way: a run - * that silently lacks its key fails later and less legibly. + * A named credential that cannot be resolved, or that resolves to fewer than + * MIN_CREDENTIAL_LENGTH characters, fails it the same way: a run that + * silently lacks its key fails later and less legibly, and a value that short + * would slip past the refusal-text reader's redaction. * * Events go to `trace`, else ctx.trace: grant.issued (only with grantMeta, * first), then credential.granted {names} or credential.denied {name, reason}. @@ -120,9 +123,14 @@ export const prepareRunEnv = async ({ const issued = grantMeta ? grantIssued(grantMeta) : null; if (issued) hook.emit('grant.issued', issued); const realHomes = [...new Set([baseEnv.HOME, homedir()].filter(Boolean))]; + const DENIED = { + not_granted: 'is not in the run grant', + unresolved: 'could not be resolved', + too_short: `resolved to a value shorter than ${MIN_CREDENTIAL_LENGTH} characters`, + }; const deny = (name, reason, values) => { hook.emit('credential.denied', { name, reason }); - return runFailure('credential_denied', `${name} ${reason === 'not_granted' ? 'is not in the run grant' : 'could not be resolved'}`, { credentials: values }); + return runFailure('credential_denied', `${name} ${DENIED[reason]}`, { credentials: values }); }; const granted = new Set(valid.credentials.map((c) => c.name)); @@ -139,6 +147,9 @@ export const prepareRunEnv = async ({ // A resolver's error text may quote the secret it failed on; drop it. try { value = await resolveOne(spec, { baseEnv, resolveSecret }); } catch { value = undefined; } if (!usable(value)) throw deny(spec.name, 'unresolved', values); + // A value this short cannot be told apart from ordinary text, so the + // refusal-text reader would not redact it. Refuse it here, before spawn. + if (value.length < MIN_CREDENTIAL_LENGTH) throw deny(spec.name, 'too_short', values); values.push(value); env[spec.name] = value; } diff --git a/packages/seats/test/record-redact-stream.test.js b/packages/seats/test/record-redact-stream.test.js new file mode 100644 index 0000000000..ba178a30c3 --- /dev/null +++ b/packages/seats/test/record-redact-stream.test.js @@ -0,0 +1,258 @@ +/** + * The gaps the security gate on #2701 found in the store's redaction, each + * closed with a test that fails on the code it replaced: + * + * 1. a value split across streamed deltas matched none of them; + * 2. a value under eight characters was granted and then not scrubbed; + * 3. the runtime token and a RunFailure's detail never met the scrub. + * + * "Stored text" below is every chunk text the record holds for one stream, + * joined in order: what a reader reconstructs the agent's output from. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createTraceStore } from '../src/trace/store.js'; +import { superviseRun } from '../src/supervisor.js'; +import { prepareRunEnv } from '../src/valet/env.js'; +import { RunFailure } from '../src/failures.js'; + +const CANARY = 'sk-test-CANARY-7f3a91c2d4e5b6a8'; +const AGENT = { type: 'agent', id: 'u-agent-stream', owner: 'u-owner-stream' }; +const OTHER = { type: 'agent', id: 'u-agent-other', owner: 'u-owner-stream' }; +const FAST = { stallMs: 0, timeoutMs: 0, cancelGraceMs: 40 }; + +const withStore = async (fn) => { + const root = mkdtempSync(join(tmpdir(), 'seats-redact-stream-')); + try { return await fn(createTraceStore({ root }), root); } finally { rmSync(root, { recursive: true, force: true }); } +}; +const allBytes = (root) => readdirSync(root, { recursive: true, withFileTypes: true }) + .filter((d) => d.isFile()) + .map((d) => readFileSync(join(d.parentPath ?? d.path, d.name), 'utf8')) + .join('\n'); +const storedText = (store, runId, kind, actorId) => store.readRun(runId) + .filter((e) => e.kind === kind && (actorId === undefined || e.actor?.id === actorId)) + .map((e) => e.data.text) + .join(''); + +/** The checks every split case shares: not in the bytes, not in the joined text, and the marker present. */ +const assertScrubbed = (store, root, runId, streams) => { + assert.equal(allBytes(root).includes(CANARY), false, 'the value is in the record bytes'); + for (const { kind, actorId, expected } of streams) { + const text = storedText(store, runId, kind, actorId); + assert.equal(text.includes(CANARY), false, `the joined ${kind} text reconstructs the value: ${text}`); + assert.equal(text, expected, `the ${kind} stream lost or changed text`); + assert.ok(text.includes('[redacted]'), 'positive control: the scrub replaced the value'); + } + assert.equal(store.verifyRun(runId).ok, true, 'held text is written through the chain like any other line'); +}; + +const splits = (n) => Array.from({ length: n - 1 }, (_, i) => i + 1); + +test('rhea\'s case: a value split across two message deltas is scrubbed', () => withStore(async (store, root) => { + const run = store.startRun({ runId: 'run-split-rhea' }); + run.redact([CANARY]); + run.emit('agent.message_chunk', { text: 'the key is sk-test-CANA' }); + run.emit('agent.message_chunk', { text: 'RY-7f3a91c2d4e5b6a8 end' }); + const texts = store.readRun('run-split-rhea').map((e) => e.data.text); + assert.deepEqual(texts, ['the key is ', '[redacted] end']); + assert.equal(allBytes(root).includes(CANARY), false); +})); + +for (const kind of ['agent.message_chunk', 'agent.thought_chunk']) { + test(`${kind}: a value split in two, at every split point, never reaches the record`, () => withStore(async (store, root) => { + for (const at of splits(CANARY.length)) { + const runId = `run-two-${kind.length}-${at}`; + const run = store.startRun({ runId, actor: AGENT }); + run.redact([CANARY]); + run.emit('run.started', { adapter: 'fake', attempt: 1 }); + run.emit(kind, { text: `the key is ${CANARY.slice(0, at)}` }); + run.emit(kind, { text: `${CANARY.slice(at)} end` }); + run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 }); + assertScrubbed(store, root, runId, [{ kind, expected: 'the key is [redacted] end' }]); + } + })); +} + +test('a value split three ways, at every pair of split points, never reaches the record', () => withStore(async (store, root) => { + for (const a of splits(CANARY.length)) { + for (let b = a + 1; b < CANARY.length; b += 1) { + const runId = `run-three-${a}-${b}`; + const run = store.startRun({ runId }); + run.redact([CANARY]); + run.emit('agent.message_chunk', { text: `key ${CANARY.slice(0, a)}` }); + run.emit('agent.message_chunk', { text: CANARY.slice(a, b) }); + run.emit('agent.message_chunk', { text: `${CANARY.slice(b)}!` }); + run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 }); + assertScrubbed(store, root, runId, [{ kind: 'agent.message_chunk', expected: 'key [redacted]!' }]); + } + } +})); + +test('two streams interleaved, each splitting the value, are scrubbed apart', () => withStore(async (store, root) => { + for (const at of splits(CANARY.length)) { + // Message vs thought, and the same kind from two actors: four streams, one line each in turn. + const runId = `run-mix-${at}`; + const run = store.startRun({ runId, actor: AGENT }); + run.redact([CANARY]); + const head = `a ${CANARY.slice(0, at)}`; + const tail = `${CANARY.slice(at)} z`; + run.emit('agent.message_chunk', { text: head }); + run.emit('agent.thought_chunk', { text: head }); + run.emit('agent.message_chunk', { text: head }, { actor: OTHER }); + run.emit('usage', { inputTokens: 1 }); // a non-stream line between the halves releases nothing + run.emit('agent.thought_chunk', { text: tail }); + run.emit('agent.message_chunk', { text: tail }, { actor: OTHER }); + run.emit('agent.message_chunk', { text: tail }); + run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 }); + assertScrubbed(store, root, runId, [ + { kind: 'agent.message_chunk', actorId: AGENT.id, expected: 'a [redacted] z' }, + { kind: 'agent.message_chunk', actorId: OTHER.id, expected: 'a [redacted] z' }, + { kind: 'agent.thought_chunk', actorId: AGENT.id, expected: 'a [redacted] z' }, + ]); + } +})); + +test('a stream that ends mid-value writes the marker before settle, never the prefix', () => withStore(async (store, root) => { + const run = store.startRun({ runId: 'run-held-settle' }); + run.redact([CANARY]); + run.emit('agent.message_chunk', { text: `cut off at ${CANARY.slice(0, 12)}` }); + run.emit('run.settled', { outcome: 'failed', failure: { kind: 'exited', detail: 'gone' }, durationMs: 1, attempts: 1 }); + const events = store.readRun('run-held-settle'); + assert.deepEqual(events.map((e) => [e.kind, e.data.text]), [ + ['agent.message_chunk', 'cut off at '], + ['agent.message_chunk', '[redacted]'], + ['run.settled', undefined], + ]); + assert.equal(allBytes(root).includes(CANARY.slice(0, 12)), false); +})); + +// The security gate's probe: all but the last character, then the run ends. +test('a value cut one character short never reaches the record', () => withStore(async (store, root) => { + const run = store.startRun({ runId: 'run-held-short' }); + run.redact([CANARY]); + run.emit('agent.message_chunk', { text: `key: ${CANARY.slice(0, -1)}` }); + run.emit('liveness.recovery', { action: 'kill', attempt: 1 }); + run.emit('run.settled', { outcome: 'failed', failure: { kind: 'exited', detail: 'killed' }, durationMs: 1, attempts: 1 }); + assert.deepEqual(storedText(store, 'run-held-short', 'agent.message_chunk', undefined), 'key: [redacted]'); + assert.equal(allBytes(root).includes(CANARY.slice(0, -1)), false); + assert.equal(allBytes(root).includes(CANARY.slice(0, 8)), false); +})); + +// The value is registered as the run's runtime token, the one the supervisor itself hands the store. +test('a supervised run killed mid-value writes the marker for its held tail, before the settle and the retry', () => withStore(async (store, root) => { + let spawns = 0; + const adapter = { + name: 'split-then-hang', + spawn(prompt, ctx) { + spawns += 1; + ctx.trace.emit('agent.message_chunk', { text: `try ${spawns}: ${CANARY.slice(0, 10)}` }); + // Ignores the abort, so the supervisor's grace timer kills it. + return new Promise(() => {}); + }, + }; + const ctrl = new AbortController(); + setTimeout(() => ctrl.abort(), 30); + const out = await superviseRun({ + adapter, prompt: 'p', store, ctx: { signal: ctrl.signal, runtimeToken: CANARY }, policy: FAST, + }); + assert.equal(out.outcome, 'cancelled'); + const events = store.readRun(out.runId); + const settledAt = events.findIndex((e) => e.kind === 'run.settled'); + const chunks = events.filter((e) => e.kind === 'agent.message_chunk'); + assert.deepEqual(chunks.map((e) => e.data.text), ['try 1: ', '[redacted]']); + assert.ok(chunks.every((e) => e.seq < events[settledAt].seq), 'the marker landed before run.settled'); + assert.equal(allBytes(root).includes(CANARY.slice(0, 10)), false, 'no prefix of the value on disk'); + + // On retry, the first attempt's tail lands before the second attempt starts, + // not glued to the second's text. (After the retry line: an adapter may still + // stream during a cancel's grace window, so recovery lines do not release.) + let tries = 0; + const retrying = { + name: 'split-then-exit', + async spawn(prompt, ctx) { + tries += 1; + ctx.trace.emit('agent.message_chunk', { text: `try ${tries}: ${CANARY.slice(0, 10)}` }); + if (tries === 1) throw new RunFailure('exited', 'the CLI died'); + return {}; + }, + }; + const again = await superviseRun({ + adapter: retrying, prompt: 'p', store, ctx: { runtimeToken: CANARY }, policy: { ...FAST, retry: { max: 1 } }, + }); + assert.equal(again.outcome, 'ok'); + assert.deepEqual(store.readRun(again.runId).map((e) => e.kind === 'agent.message_chunk' ? e.data.text : e.kind), [ + 'attempt.started', 'try 1: ', 'liveness.recovery', '[redacted]', + 'attempt.started', 'try 2: ', '[redacted]', 'run.settled', + ]); + assert.equal(allBytes(root).includes(CANARY.slice(0, 10)), false); +})); + +test('the store scrubs a registered value of any length', () => withStore(async (store, root) => { + const run = store.startRun({ runId: 'run-short-1' }); + run.redact(['abc123']); + run.emit('agent.message_chunk', { text: 'value abc123 here' }); + run.emit('tool.update', { toolCallId: 't', title: 'abc123' }); + run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 }); + const bytes = allBytes(root); + assert.equal(bytes.includes('abc123'), false, 'a six-character value reached the record'); + assert.equal(storedText(store, 'run-short-1', 'agent.message_chunk'), 'value [redacted] here'); +})); + +test('the valet refuses a credential under eight characters as credential_denied, before any HOME', async () => { + const root = mkdtempSync(join(tmpdir(), 'seats-redact-short-')); + try { + const events = []; + const trace = { emit: (type, data) => events.push({ type, data }), redact: () => assert.fail('a refused value was registered') }; + await assert.rejects(prepareRunEnv({ + grant: { credentials: [{ name: 'SHORT_KEY', from: 'resolver' }] }, resolveSecret: async () => 'abc123', root, trace, + }), (error) => { + assert.ok(error instanceof RunFailure); + assert.equal(error.kind, 'credential_denied'); + assert.match(error.message, /SHORT_KEY resolved to a value shorter than 8 characters/); + assert.equal(error.message.includes('abc123'), false); + return true; + }); + assert.deepEqual(events, [{ type: 'credential.denied', data: { name: 'SHORT_KEY', reason: 'too_short' } }]); + assert.deepEqual(readdirSync(root), [], 'the refused prepare removed its HOME'); + // Positive control: eight characters is enough. + const ok = await prepareRunEnv({ + grant: { credentials: [{ name: 'SHORT_KEY', from: 'resolver' }] }, resolveSecret: async () => 'abcd1234', root, trace: { emit() {} }, + }); + assert.equal(ok.env.SHORT_KEY, 'abcd1234'); + await ok.cleanup(); + } finally { rmSync(root, { recursive: true, force: true }); } +}); + +const TOKEN = 'cm_agent_runtime-token-5e6f7a8b9c0d'; + +test('the run\'s runtime token is scrubbed from agent text', () => withStore(async (store, root) => { + const adapter = { + name: 'token-echo', + async spawn(prompt, ctx) { + ctx.trace.emit('agent.message_chunk', { text: `my token is ${ctx.runtimeToken}.` }); + return {}; + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, ctx: { runtimeToken: TOKEN }, policy: FAST }); + assert.equal(out.outcome, 'ok'); + assert.equal(allBytes(root).includes(TOKEN), false, 'the runtime token reached the record'); + assert.equal(storedText(store, out.runId, 'agent.message_chunk'), 'my token is [redacted].'); +})); + +test('a RunFailure\'s detail is scrubbed of the spawn\'s credentials before settle', () => withStore(async (store) => { + // A run with no redact hook: only failureData stands between the detail and the line. + const bare = { + startRun: (meta) => { + const { redact, ...run } = store.startRun(meta); + return run; + }, + }; + const adapter = { name: 'leaky-failure', async spawn() { throw new RunFailure('exited', `auth failed for ${TOKEN}`); } }; + const out = await superviseRun({ adapter, prompt: 'p', store: bare, ctx: { runtimeToken: TOKEN }, policy: FAST }); + assert.equal(out.outcome, 'failed'); + const [settled] = store.readRun(out.runId).filter((e) => e.kind === 'run.settled'); + assert.deepEqual(settled.data.failure, { kind: 'exited', detail: 'auth failed for [redacted]' }); +})); diff --git a/packages/seats/test/record-redact-supervised.test.js b/packages/seats/test/record-redact-supervised.test.js new file mode 100644 index 0000000000..5ec6d46f0b --- /dev/null +++ b/packages/seats/test/record-redact-supervised.test.js @@ -0,0 +1,186 @@ +/** + * A credential value the valet resolves for a supervised run never reaches + * the record, even when the agent repeats it. The value travels the real + * path: prepareRunEnv is called with the spawn ctx the supervisor hands the + * adapter, so it registers the value through the supervisor's trace wrapper, + * and the store scrubs it before a line is hashed or written. + * + * Everything is local and keyless: the fake provider answers on 127.0.0.1 and + * the "credential" is a fake value. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { superviseRun } from '../src/supervisor.js'; +import { createTraceStore } from '../src/trace/store.js'; +import { prepareRunEnv } from '../src/valet/env.js'; +import { createPermissionPolicy } from '../src/valet/permissions.js'; +import { toOtlpJson } from '../src/record/otel.js'; +import { runFailure } from '../src/failures.js'; +import { startFakeProvider } from '../test-support/fake-provider.js'; + +const CANARY = 'sk-test-CANARY-7f3a91c2d4e5b6a8'; +const KEY_NAME = 'FAKE_PROVIDER_KEY'; +const GRANT = { credentials: [{ name: KEY_NAME, from: 'resolver' }], tools: { allow: ['read'], ask: ['fetch'] } }; +const AGENT = { type: 'agent', id: 'u-agent-redact', owner: 'u-owner-redact', name: 'Redact test agent' }; +const OWNER = { id: 'u-owner-redact', role: 'agent-owner' }; +// The kinds the adapter writes itself, naming its actor. permission.* are the valet's. +const AGENT_KINDS = new Set(['agent.message_chunk', 'tool.call', 'tool.update', 'result.reported']); + +// The store's redaction is exact-match on known values (docs/record-v1.md §3, +// rule 1): a value the agent transforms, such as base64, is not claimed and +// is not asserted here. The canary is all unreserved characters, so its +// URL-encoded and JSON-escaped forms are the value itself; the check names +// them anyway so a change of canary cannot quietly drop them. +const FORMS = Object.freeze({ + raw: CANARY, + urlEncoded: encodeURIComponent(CANARY), + jsonEscaped: JSON.stringify(CANARY).slice(1, -1), +}); +const assertAbsent = (text, where) => { + for (const [form, value] of Object.entries(FORMS)) { + assert.equal(text.includes(value), false, `the ${form} credential value appears in ${where}`); + } +}; + +const allFiles = (dir) => readdirSync(dir, { recursive: true, withFileTypes: true }) + .filter((d) => d.isFile()) + .map((d) => join(d.parentPath ?? d.path, d.name)); + +/** The text an OpenAI-style SSE stream carried, one string per content delta. */ +const sseDeltas = async (response) => { + const body = await response.text(); + return body.split('\n') + .filter((line) => line.startsWith('data: ') && line !== 'data: [DONE]') + .map((line) => JSON.parse(line.slice(6)).choices?.[0]?.delta?.content) + .filter((text) => typeof text === 'string' && text.length > 0); +}; + +/** + * An adapter that does what a CLI adapter does with a granted key: asks the + * valet for its env, calls the provider with it, and streams what comes back. + * The provider is scripted to echo the key, so the key lands in agent text, a + * tool result, the question the valet's permission policy records, and the + * final result. + */ +const echoingAdapter = ({ baseURL, envRoot, fail, seen }) => ({ + name: 'echo-fake', + providerKeyEnv: KEY_NAME, + async spawn(prompt, ctx) { + const prepared = await prepareRunEnv({ + grant: GRANT, resolveSecret: async () => CANARY, root: envRoot, ctx, require: [KEY_NAME], + }); + try { + const key = prepared.env[KEY_NAME]; + seen.key = key; + const as = { actor: AGENT }; + ctx.trace.emit('turn.prompt_sent', { chars: prompt.length }); + const response = await fetch(`${baseURL}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${key}` }, + body: JSON.stringify({ model: 'fake-model', stream: true, messages: [{ role: 'user', content: `[[reply:the key is ${key}]]` }] }), + }); + const deltas = await sseDeltas(response); + seen.reply = deltas.join(''); + for (const text of deltas) ctx.trace.emit('agent.message_chunk', { text }, as); + ctx.trace.emit('tool.call', { + toolCallId: 'tc-1', title: `printenv ${KEY_NAME}`, kind: 'execute', status: 'pending', surface: 'native', + }, as); + ctx.trace.emit('tool.update', { toolCallId: 'tc-1', status: 'completed', title: `${KEY_NAME}=${key}` }, as); + const decide = createPermissionPolicy(GRANT, { + trace: ctx.trace, askedOf: OWNER, approve: async () => ({ approved: true, by: OWNER.id, role: OWNER.role }), + }); + const answer = await decide({ + toolCall: { toolCallId: 'tc-2', title: `Send ${key} to the deploy hook?`, kind: 'fetch', rawInput: { token: key } }, + options: [{ optionId: 'allow', kind: 'allow_once' }, { optionId: 'reject', kind: 'reject_once' }], + }); + seen.answer = answer; + if (fail) throw runFailure('exited', `the CLI exited after printing ${key}`); + ctx.trace.emit('agent.message_chunk', { text: `Done. I used ${key}.` }, as); + ctx.trace.emit('result.reported', { status: 'done', refs: [{ type: 'note', id: `used-${key}` }] }, as); + return { text: seen.reply }; + } finally { + await prepared.cleanup(); + } + }, +}); + +const withHarness = async (fn) => { + const root = mkdtempSync(join(tmpdir(), 'seats-redact-sup-')); + const fake = await startFakeProvider(); + try { + const traceRoot = join(root, 'trace'); + return await fn({ fake, root, traceRoot, store: createTraceStore({ root: traceRoot }) }); + } finally { + await fake.close(); + rmSync(root, { recursive: true, force: true }); + } +}; + +for (const fail of [false, true]) { + test(`a granted credential the agent echoes never reaches the record or its export (${fail ? 'failed' : 'ok'} run)`, () => withHarness(async ({ + fake, root, traceRoot, store, + }) => { + const seen = {}; + const out = await superviseRun({ + adapter: echoingAdapter({ baseURL: fake.baseURL, envRoot: root, fail, seen }), + prompt: 'deploy it', + store, + policy: { stallMs: 0, timeoutMs: 0 }, + }); + + // Positive controls: the value really reached the run and the agent really repeated it. + assert.equal(seen.key, CANARY, 'the valet resolved the planted value into the run env'); + assert.ok(seen.reply.includes(CANARY), 'the provider echoed the value back as agent text'); + assert.deepEqual(fake.requests.map((r) => r.authorization), ['present']); + assert.equal(out.outcome, fail ? 'failed' : 'ok'); + if (fail) assert.ok(out.failure.detail.includes(CANARY), 'the failure the run settled on quoted the value'); + + // Every byte the store wrote: the run file and the index. + const files = allFiles(traceRoot); + assert.deepEqual(files.map((f) => f.slice(traceRoot.length + 1)).sort(), ['index.jsonl', `runs/${out.runId}.jsonl`]); + const bytes = files.map((f) => readFileSync(f, 'utf8')).join('\n'); + assertAbsent(bytes, 'the record files'); + + const entries = store.readRun(out.runId); + // The export reads what the store hands back, so it is checked on its own, + // before anything else about those entries. + const otlp = JSON.stringify(toOtlpJson(entries)); + assertAbsent(otlp, 'the OTLP export'); + assert.ok(otlp.includes('[redacted]'), 'positive control: a scrubbed field reached the export'); + assert.ok(otlp.includes('seats.question'), 'positive control: the permission question reached the export'); + assert.ok(otlp.includes('Send [redacted] to the deploy hook?'), 'the exported question is the scrubbed one'); + + const byKind = (kind) => entries.filter((e) => e.kind === kind); + // Each place the value was planted holds the marker instead, so its absence above is a scrub, not a drop. + assert.ok(byKind('agent.message_chunk').some((e) => e.data.text.includes('[redacted]')), 'agent text was scrubbed'); + assert.equal(byKind('tool.update')[0].data.title, `${KEY_NAME}=[redacted]`, 'the tool result was scrubbed'); + const [asked] = byKind('permission.requested'); + assert.equal(asked.data.question, 'Send [redacted] to the deploy hook?', 'the permission question was scrubbed'); + assert.deepEqual(asked.data.askedOf, OWNER); + assert.match(asked.data.argsDigest, /^sha256:[0-9a-f]{64}$/, 'the arguments are recorded as a digest only'); + assert.deepEqual(seen.answer, { outcome: 'selected', optionId: 'allow' }); + const [settled] = byKind('run.settled'); + if (fail) { + assert.equal(settled.data.failure.detail, 'the CLI exited after printing [redacted]', 'the final failure was scrubbed'); + const indexRow = readFileSync(join(traceRoot, 'index.jsonl'), 'utf8').split('\n').filter(Boolean).map(JSON.parse).find((r) => r.outcome); + assert.equal(indexRow.failure.detail, 'the CLI exited after printing [redacted]', 'the index row was scrubbed'); + } else { + assert.equal(byKind('result.reported')[0].data.refs[0].id, 'used-[redacted]', 'the final result was scrubbed'); + assert.ok(byKind('agent.message_chunk').some((e) => e.data.text === 'Done. I used [redacted].')); + } + assert.deepEqual(byKind('credential.granted')[0].data, { names: [KEY_NAME] }, 'the valet recorded the name through the wrapper'); + + // Scrubbed before hashing: the chain over the scrubbed lines is intact. + const verified = store.verifyRun(out.runId); + assert.equal(verified.ok, true, JSON.stringify(verified.errors)); + assert.equal(verified.entries, entries.length); + + // The adapter's own entries carry the actor it named. + const agentEntries = entries.filter((e) => AGENT_KINDS.has(e.kind)); + assert.ok(agentEntries.length >= (fail ? 3 : 5), `${agentEntries.length} adapter entries`); + for (const entry of agentEntries) assert.deepEqual(entry.actor, AGENT, `actor on ${entry.kind} #${entry.seq}`); + })); +} diff --git a/packages/seats/test/supervisor-process.test.js b/packages/seats/test/supervisor-process.test.js new file mode 100644 index 0000000000..cb71038d67 --- /dev/null +++ b/packages/seats/test/supervisor-process.test.js @@ -0,0 +1,105 @@ +// Real processes against the fake provider: the hang is a genuine HTTP call +// that never answers, in a genuine child, so liveness has to fire on silence +// it did not stage, and the child has to actually die. +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { superviseRun } from '../src/supervisor.js'; +import { createTraceStore } from '../src/trace/store.js'; +import { runFailure } from '../src/failures.js'; +import { observeSpawn } from '../src/runtime-observation.js'; +import { startFakeProvider } from '../test-support/fake-provider.js'; + +// Says it is starting (a byte of progress), then waits on a completion the +// fake provider holds open forever. +const CHILD = ` +process.stdout.write('requesting\\n'); +fetch(process.argv[1] + '/chat/completions', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ model: 'fake-model', messages: [{ role: 'user', content: '[[stall]]' }] }), +}).then((r) => r.text()).then((t) => process.stdout.write(t)); +`; + +const nodeFetchAdapter = ({ honourAbort, children }) => ({ + name: 'node-fetch', + spawn(prompt, ctx) { + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, ['-e', CHILD, ctx.baseURL], { cwd: ctx.cwd, env: ctx.env, stdio: ['ignore', 'pipe', 'pipe'] }); + children.push(child); + observeSpawn(child, ctx.onSpawn, {}); + let out = ''; + child.stdout.on('data', (chunk) => { out += chunk; }); + if (honourAbort) ctx.signal.addEventListener('abort', () => child.kill('SIGTERM'), { once: true }); + child.on('close', (code, signal) => { + ctx.trace.emit('process.exited', { code, signal }); + if (ctx.signal.aborted) reject(ctx.signal.reason); + else if (code === 0) resolve({ text: out }); + else reject(runFailure('exited', `node exited ${code ?? signal}`)); + }); + }); + }, +}); + +const isRunning = (pid) => { + try { process.kill(pid, 0); return true; } catch (error) { return error.code !== 'ESRCH'; } +}; + +const withProvider = async (fn) => { + const home = mkdtempSync(join(tmpdir(), 'seats-sup-proc-')); + const fake = await startFakeProvider(); + try { + return await fn({ fake, home, store: createTraceStore({ root: join(home, 'trace') }) }); + } finally { + await fake.close(); + rmSync(home, { recursive: true, force: true }); + } +}; + +const settleChild = async (child) => { + if (child.exitCode === null && child.signalCode === null) { + let timer; + await Promise.race([once(child, 'exit'), new Promise((r) => { timer = setTimeout(r, 3000); })]); + clearTimeout(timer); + } +}; + +for (const honourAbort of [false, true]) { + test(`a real child hung on a stalled HTTP call is caught and killed (adapter ${honourAbort ? 'honours' : 'ignores'} abort)`, () => withProvider(async ({ fake, home, store }) => { + const children = []; + const ctx = { baseURL: fake.baseURL, cwd: home, env: { PATH: process.env.PATH, HOME: home } }; + const started = Date.now(); + const out = await superviseRun({ + adapter: nodeFetchAdapter({ honourAbort, children }), prompt: 'p', ctx, store, + policy: { stallMs: 400, cancelGraceMs: 400, timeoutMs: 10000 }, + }); + const elapsed = Date.now() - started; + + assert.equal(out.outcome, 'stuck'); + assert.equal(out.failure.kind, 'stuck'); + assert.ok(elapsed < 3000, `settled in ${elapsed}ms`); + assert.deepEqual(fake.requests.map((r) => r.directive), ['stall'], 'the child really made the call that hung'); + + assert.equal(children.length, 1); + const [child] = children; + await settleChild(child); + assert.ok(child.signalCode, `the child was killed by a signal (exit ${child.exitCode}, signal ${child.signalCode})`); + assert.equal(isRunning(child.pid), false, `pid ${child.pid} is gone`); + + const types = store.readRun(out.runId).map((e) => (e.type === 'liveness.recovery' ? `recovery:${e.data.action}` : e.type)); + if (honourAbort) { + assert.equal(child.signalCode, 'SIGTERM'); + assert.equal(out.failure.adapterIgnoredCancel, undefined); + assert.deepEqual(types, ['run.started', 'liveness.stalled', 'recovery:cancel', 'process.exited', 'run.settled']); + } else { + // Nobody but the supervisor stopped it: the kill came from the handle onSpawn gave it. + assert.equal(child.signalCode, 'SIGKILL'); + assert.equal(out.failure.adapterIgnoredCancel, true); + assert.deepEqual(types, ['run.started', 'liveness.stalled', 'recovery:cancel', 'recovery:kill', 'run.settled']); + } + })); +} diff --git a/packages/seats/test/supervisor.test.js b/packages/seats/test/supervisor.test.js new file mode 100644 index 0000000000..0c0e806d9b --- /dev/null +++ b/packages/seats/test/supervisor.test.js @@ -0,0 +1,394 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { superviseRun, SUPERVISOR_DEFAULTS, DEFAULT_RETRY_ON } from '../src/supervisor.js'; +import * as seats from '../src/index.js'; +import { createTraceStore } from '../src/trace/store.js'; +import { RunFailure, runFailure } from '../src/failures.js'; +import { observeSpawn } from '../src/runtime-observation.js'; + +// Small windows instead of a fake clock: every timer here is real, so the +// supervisor is exercised exactly as it runs, just in milliseconds. +const FAST = { stallMs: 60, cancelGraceMs: 60, timeoutMs: 0 }; +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +const withStore = async (fn) => { + const root = mkdtempSync(join(tmpdir(), 'seats-sup-')); + try { return await fn(createTraceStore({ root })); } finally { rmSync(root, { recursive: true, force: true }); } +}; +const typesOf = (store, runId) => store.readRun(runId).map((e) => ( + e.type === 'liveness.recovery' ? `recovery:${e.data.action}` : e.type)); +const settledOf = (store, runId) => store.readRun(runId).filter((e) => e.type === 'run.settled'); + +/** Hangs until aborted, then rejects with the signal's reason, as the seam asks. */ +const hangsHonouringAbort = (calls = []) => ({ + name: 'hang-honour', + spawn(prompt, ctx) { + calls.push(ctx); + return new Promise((_, reject) => { + ctx.signal.addEventListener('abort', () => reject(ctx.signal.reason), { once: true }); + }); + }, +}); + +/** A fake child: an EventEmitter with stdout/stderr and a kill that records. */ +const fakeChild = () => { + const child = new EventEmitter(); + child.stdout = new EventEmitter(); + child.stderr = new EventEmitter(); + child.exitCode = null; + child.signalCode = null; + child.kills = []; + child.kill = (signal) => { + child.kills.push(signal); + child.signalCode = signal; + setImmediate(() => child.emit('exit', null, signal)); + return true; + }; + return child; +}; + +test('defaults are the documented windows and retry is off', () => { + assert.deepEqual({ ...SUPERVISOR_DEFAULTS }, { stallMs: 120000, cancelGraceMs: 10000, timeoutMs: 900000 }); + assert.deepEqual([...DEFAULT_RETRY_ON], ['stuck', 'exited', 'timeout']); + assert.equal(seats.superviseRun, superviseRun); +}); + +test('a successful run: started, the adapter\'s events, settled ok — in that order', () => withStore(async (store) => { + const adapter = { + name: 'ok', + async spawn(prompt, ctx) { + ctx.trace.emit('turn.prompt_sent', { chars: prompt.length }); + ctx.trace.emit('agent.message_chunk', { text: 'PONG' }); + return { text: 'PONG', newSessionId: 's-1' }; + }, + }; + const out = await superviseRun({ adapter, prompt: 'ping', store, policy: FAST }); + assert.equal(out.outcome, 'ok'); + assert.deepEqual(out.result, { text: 'PONG', newSessionId: 's-1' }); + assert.equal(out.failure, undefined); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'turn.prompt_sent', 'agent.message_chunk', 'run.settled']); + const [settled] = settledOf(store, out.runId); + assert.equal(settled.data.outcome, 'ok'); + assert.equal(settled.data.attempts, 1); + assert.equal(typeof settled.data.durationMs, 'number'); + assert.equal(store.listRuns()[0].outcome, 'ok'); +})); + +test('a hung adapter that honours abort settles stuck, with the reason on its signal', () => withStore(async (store) => { + const calls = []; + const stalled = []; + const out = await superviseRun({ + adapter: hangsHonouringAbort(calls), prompt: 'p', store, + policy: { ...FAST, cancelGraceMs: 2000, onStalled: (info) => stalled.push(info) }, + }); + assert.equal(out.outcome, 'stuck'); + assert.ok(out.failure instanceof RunFailure); + assert.equal(out.failure.kind, 'stuck'); + assert.equal(out.failure.adapterIgnoredCancel, undefined); + assert.equal(calls[0].signal.reason.kind, 'stuck', 'the adapter sees a RunFailure as the abort reason'); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'liveness.stalled', 'recovery:cancel', 'run.settled']); + const events = store.readRun(out.runId); + const stall = events.find((e) => e.type === 'liveness.stalled').data; + assert.equal(stall.windowMs, 60); + assert.ok(stall.idleMs >= 60, `idleMs ${stall.idleMs} is at least the window`); + assert.deepEqual(events.find((e) => e.type === 'liveness.recovery').data, { action: 'cancel', attempt: 1 }); + assert.equal(settledOf(store, out.runId)[0].data.failure.kind, 'stuck'); + assert.equal(stalled.length, 1); + assert.equal(stalled[0].runId, out.runId); + assert.equal(stalled[0].attempt, 1); +})); + +test('a hung adapter that IGNORES abort still settles stuck, records kill, and kills its child', () => withStore(async (store) => { + const child = fakeChild(); + const adapter = { + name: 'hang-ignore', + spawn(prompt, ctx) { + ctx.onSpawn({ model: null, effort: null, at: 'x' }, child); + return new Promise(() => {}); // never settles, never looks at the signal + }, + }; + const started = Date.now(); + const out = await superviseRun({ adapter, prompt: 'p', store, policy: FAST }); + const elapsed = Date.now() - started; + assert.equal(out.outcome, 'stuck'); + assert.equal(out.failure.kind, 'stuck'); + assert.equal(out.failure.adapterIgnoredCancel, true); + assert.match(out.failure.detail, /did not settle within 60ms of cancellation/); + assert.deepEqual(child.kills, ['SIGKILL']); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'liveness.stalled', 'recovery:cancel', 'recovery:kill', 'run.settled']); + assert.ok(elapsed < 1000, `settled in ${elapsed}ms, not left running`); +})); + +test('an ignored cancellation is not retried even when the kind is listed', () => withStore(async (store) => { + let calls = 0; + const adapter = { name: 'hang-ignore', spawn() { calls += 1; return new Promise(() => {}); } }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { ...FAST, retry: { max: 2 } } }); + assert.equal(out.outcome, 'stuck'); + assert.equal(calls, 1, 'a second attempt could race the first, which may still be running'); +})); + +test('steady trace progress is never called stuck', () => withStore(async (store) => { + const adapter = { + name: 'steady', + async spawn(prompt, ctx) { + for (let i = 0; i < 15; i += 1) { // 300ms of work against a 60ms window + await delay(20); + ctx.trace.emit('agent.thought_chunk', { chars: 1 }); + } + return { text: 'done' }; + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: FAST }); + assert.equal(out.outcome, 'ok'); + assert.ok(!typesOf(store, out.runId).some((t) => t.startsWith('liveness') || t.startsWith('recovery'))); +})); + +test('child stdout/stderr bytes count as progress, and the caller\'s onSpawn still receives every argument', () => withStore(async (store) => { + const seen = []; + const adapter = { + name: 'bytes', + async spawn(prompt, ctx) { + const child = fakeChild(); + observeSpawn(child, ctx.onSpawn, { model: 'm' }); + child.emit('spawn'); + for (let i = 0; i < 15; i += 1) { + await delay(20); + (i % 2 ? child.stdout : child.stderr).emit('data', Buffer.from('.')); + } + return { text: 'done' }; + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, ctx: { onSpawn: (...args) => seen.push(args) }, policy: FAST }); + assert.equal(out.outcome, 'ok'); + assert.equal(seen.length, 1); + assert.equal(seen[0][0].model, 'm'); + assert.equal(typeof seen[0][1].kill, 'function', 'observeSpawn passes the child as the second argument'); + assert.ok(!typesOf(store, out.runId).includes('liveness.stalled')); +})); + +test('progress that stops part-way is caught, and idleMs is measured from the last signal', () => withStore(async (store) => { + const adapter = { + name: 'stops', + spawn(prompt, ctx) { + return new Promise((_, reject) => { + ctx.signal.addEventListener('abort', () => reject(ctx.signal.reason)); + (async () => { for (let i = 0; i < 5; i += 1) { await delay(20); ctx.trace.emit('agent.thought_chunk', { chars: 1 }); } })(); + }); + }, + }; + const started = Date.now(); + const out = await superviseRun({ adapter, prompt: 'p', store, policy: FAST }); + assert.equal(out.outcome, 'stuck'); + const stall = store.readRun(out.runId).find((e) => e.type === 'liveness.stalled').data; + assert.ok(stall.idleMs >= 60 && stall.idleMs < 200, `idleMs ${stall.idleMs}`); + assert.ok(Date.now() - started >= 160, 'the window restarted on each signal'); +})); + +test('a refusal is never retried, even when listed in retry.on', () => withStore(async (store) => { + let calls = 0; + const adapter = { name: 'refuses', async spawn() { calls += 1; throw runFailure('refusal', 'content policy'); } }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { ...FAST, retry: { max: 3, on: ['refusal', 'stuck'] } } }); + assert.equal(calls, 1); + assert.equal(out.outcome, 'failed'); + assert.equal(out.attempts, 1); + assert.equal(out.failure.kind, 'refusal'); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'run.settled']); + assert.deepEqual(settledOf(store, out.runId)[0].data.failure, { kind: 'refusal', detail: 'content policy' }); +})); + +test('stuck is retried once on the same runId, carrying the session, then succeeds', () => withStore(async (store) => { + const calls = []; + const adapter = { + name: 'flaky', + spawn(prompt, ctx) { + calls.push({ sessionId: ctx.sessionId }); + if (calls.length === 2) return Promise.resolve({ text: 'second time', newSessionId: 'sess-1' }); + return new Promise((_, reject) => { + ctx.signal.addEventListener('abort', () => { + const failure = ctx.signal.reason; + failure.sessionId = 'sess-1'; // the adapter knew its session when it failed + reject(failure); + }); + }); + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { ...FAST, retry: { max: 1 } } }); + assert.equal(out.outcome, 'ok'); + assert.equal(out.attempts, 2); + assert.equal(out.result.text, 'second time'); + assert.deepEqual(calls, [{ sessionId: undefined }, { sessionId: 'sess-1' }]); + assert.deepEqual(typesOf(store, out.runId), [ + 'run.started', 'liveness.stalled', 'recovery:cancel', 'recovery:retry', 'run.started', 'run.settled']); + const starts = store.readRun(out.runId).filter((e) => e.type === 'run.started').map((e) => e.data); + assert.deepEqual(starts, [{ adapter: 'flaky', attempt: 1 }, { adapter: 'flaky', sessionId: 'sess-1', attempt: 2 }]); + assert.equal(settledOf(store, out.runId)[0].data.attempts, 2); + assert.equal(store.listRuns().length, 1, 'a retry is the same run, not a new one'); +})); + +test('retry is off by default: a stuck run is attempted once', () => withStore(async (store) => { + const calls = []; + const out = await superviseRun({ adapter: hangsHonouringAbort(calls), prompt: 'p', store, policy: { ...FAST, cancelGraceMs: 1000 } }); + assert.equal(out.outcome, 'stuck'); + assert.equal(calls.length, 1); +})); + +test('retry stops at max and settles with the last failure', () => withStore(async (store) => { + const calls = []; + const out = await superviseRun({ adapter: hangsHonouringAbort(calls), prompt: 'p', store, policy: { ...FAST, cancelGraceMs: 1000, retry: { max: 2 } } }); + assert.equal(calls.length, 3); + assert.equal(out.attempts, 3); + assert.equal(out.outcome, 'stuck'); + assert.equal(settledOf(store, out.runId).length, 1); +})); + +test('a kind left out of retry.on is not retried', () => withStore(async (store) => { + let calls = 0; + const adapter = { name: 'exits', async spawn() { calls += 1; throw runFailure('exited', 'code 1'); } }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { ...FAST, retry: { max: 2, on: ['stuck'] } } }); + assert.equal(calls, 1); + assert.equal(out.outcome, 'failed'); +})); + +test('the wall-clock timeout cancels a busy run with kind timeout, not stuck', () => withStore(async (store) => { + const adapter = { + name: 'busy', + spawn(prompt, ctx) { + return new Promise((_, reject) => { + const tick = setInterval(() => ctx.trace.emit('agent.thought_chunk', { chars: 1 }), 10); + ctx.signal.addEventListener('abort', () => { clearInterval(tick); reject(ctx.signal.reason); }); + }); + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { stallMs: 1000, cancelGraceMs: 1000, timeoutMs: 120 } }); + assert.equal(out.outcome, 'failed'); + assert.equal(out.failure.kind, 'timeout'); + const types = typesOf(store, out.runId); + assert.ok(!types.includes('liveness.stalled')); + assert.deepEqual(types.filter((t) => !t.startsWith('agent.')), ['run.started', 'recovery:cancel', 'run.settled']); + assert.equal(settledOf(store, out.runId)[0].data.failure.kind, 'timeout'); +})); + +test('a timeout the adapter ignores still settles as timeout', () => withStore(async (store) => { + const adapter = { name: 'deaf', spawn: () => new Promise(() => {}) }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { stallMs: 0, cancelGraceMs: 40, timeoutMs: 60 } }); + assert.equal(out.failure.kind, 'timeout'); + assert.equal(out.failure.adapterIgnoredCancel, true); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'recovery:cancel', 'recovery:kill', 'run.settled']); +})); + +test('a caller\'s signal cancels the run as cancelled, and a pre-aborted signal never spawns', () => withStore(async (store) => { + const controller = new AbortController(); + setTimeout(() => controller.abort(), 30); + const out = await superviseRun({ adapter: hangsHonouringAbort(), prompt: 'p', store, ctx: { signal: controller.signal }, policy: { ...FAST, stallMs: 1000, retry: { max: 3 } } }); + assert.equal(out.outcome, 'cancelled'); + assert.equal(out.failure.kind, 'cancelled'); + assert.equal(out.attempts, 1); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'run.settled']); + + let spawned = 0; + const out2 = await superviseRun({ adapter: { name: 'x', spawn: () => { spawned += 1; return {}; } }, prompt: 'p', store, ctx: { signal: AbortSignal.abort() }, policy: FAST }); + assert.equal(out2.outcome, 'cancelled'); + assert.equal(spawned, 0); +})); + +test('a bare Error settles failed with no kind, scrubbed, and does not throw', () => withStore(async (store) => { + const key = 'sk-or-v1-abcdefabcdefabcdef'; + const adapter = { name: 'bare', providerKeyEnv: 'OPENROUTER_API_KEY', spawn() { throw new Error(`exploded with ${key}`); } }; + const out = await superviseRun({ adapter, prompt: 'p', store, ctx: { env: { OPENROUTER_API_KEY: key } }, policy: FAST }); + assert.equal(out.outcome, 'failed'); + const { failure } = settledOf(store, out.runId)[0].data; + assert.equal(failure.kind, null); + assert.doesNotMatch(failure.detail, /abcdefabcdef/); + assert.match(failure.detail, /exploded with \[redacted\]/); +})); + +test('an adapter cannot write the supervisor\'s run.* events', () => withStore(async (store) => { + const adapter = { + name: 'forger', + async spawn(prompt, ctx) { + assert.equal(ctx.trace.emit('run.settled', { outcome: 'ok', durationMs: 0, attempts: 1 }), null); + assert.equal(ctx.trace.emit('run.started', { attempt: 9 }), null); + throw runFailure('exited', 'code 2'); + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: FAST }); + assert.equal(out.outcome, 'failed'); + assert.deepEqual(typesOf(store, out.runId), ['run.started', 'run.settled']); +})); + +test('exactly one run.settled on every path, and a late adapter answer writes nothing', () => withStore(async (store) => { + const lateHook = []; + const late = { + name: 'late', + spawn(prompt, ctx) { + // Ignores the abort, then answers after the supervisor gave up on it. + return new Promise((resolve) => setTimeout(() => { + lateHook.push(ctx.trace.emit('agent.message_chunk', { text: 'too late' })); + resolve({ text: 'too late' }); + }, 250)); + }, + }; + const paths = [ + { adapter: { name: 'ok', spawn: async () => ({ text: 'x' }) }, outcome: 'ok' }, + { adapter: hangsHonouringAbort(), outcome: 'stuck' }, + { adapter: { name: 'deaf', spawn: () => new Promise(() => {}) }, outcome: 'stuck' }, + { adapter: late, outcome: 'stuck' }, + { adapter: { name: 'refuse', spawn: async () => { throw runFailure('refusal', 'no'); } }, outcome: 'failed', policy: { retry: { max: 2, on: ['refusal'] } } }, + { adapter: hangsHonouringAbort(), outcome: 'stuck', policy: { retry: { max: 1 } } }, + { adapter: { name: 'deaf', spawn: () => new Promise(() => {}) }, outcome: 'failed', policy: { stallMs: 0, timeoutMs: 50 } }, + { adapter: { name: 'bare', spawn: () => { throw new Error('boom'); } }, outcome: 'failed' }, + ]; + const runIds = []; + for (const path of paths) { + const out = await superviseRun({ adapter: path.adapter, prompt: 'p', store, policy: { ...FAST, ...path.policy } }); + assert.equal(out.outcome, path.outcome, path.adapter.name); + runIds.push(out.runId); + } + await delay(300); // let the late adapter answer + assert.deepEqual(lateHook, [null], 'the late trace write was dropped'); + for (const runId of runIds) { + assert.equal(settledOf(store, runId).length, 1, runId); + assert.equal(store.readRun(runId).at(-1).type, 'run.settled', 'nothing follows the settle'); + } + assert.ok(store.listRuns({ limit: 100 }).every((row) => row.outcome !== 'running')); +})); + +test('an adapter that answers after the supervisor aborted it still settles with the abort reason', () => withStore(async (store) => { + const adapter = { + name: 'answers-anyway', + spawn(prompt, ctx) { + return new Promise((resolve) => { + ctx.signal.addEventListener('abort', () => resolve({ text: 'a reply nobody waited for' })); + }); + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { ...FAST, cancelGraceMs: 1000 } }); + assert.equal(out.outcome, 'stuck'); + assert.equal(out.result, undefined); + assert.equal(settledOf(store, out.runId)[0].data.outcome, 'stuck'); +})); + +test('a cancelled attempt\'s late trace writes never reach the next attempt', () => withStore(async (store) => { + let calls = 0; + const adapter = { + name: 'stray', + spawn(prompt, ctx) { + calls += 1; + if (calls === 2) return delay(100).then(() => ({ text: 'ok' })); + return new Promise((_, reject) => { + ctx.signal.addEventListener('abort', () => { + reject(ctx.signal.reason); + setTimeout(() => ctx.trace.emit('agent.message_chunk', { text: 'stale' }), 30); + }); + }); + }, + }; + const out = await superviseRun({ adapter, prompt: 'p', store, policy: { ...FAST, stallMs: 200, retry: { max: 1 } } }); + assert.equal(out.outcome, 'ok'); + assert.equal(out.attempts, 2); + assert.ok(!store.readRun(out.runId).some((e) => e.data?.text === 'stale')); +}));