diff --git a/docs/TEST_INTEGRITY.md b/docs/TEST_INTEGRITY.md index 0b481ef..33ab466 100644 --- a/docs/TEST_INTEGRITY.md +++ b/docs/TEST_INTEGRITY.md @@ -42,6 +42,146 @@ test_integrity_note: if_true_explain_spec_basis: "" ``` +## 2026-09-30 — Codex terminal completion correction + +```yaml +test_integrity_note: + change_type: spec_correction + scenarios: + - id: SCN-mixed-runtime + given: A status-zero Codex protocol stream containing thread.started and turn.completed + when: Fleet startup captures the provider-specific durable identity + then: The completed turn may record idle with the captured identity + - id: SCN-codex-terminal-completion + given: A status-zero Codex stream containing thread.started but no turn.completed + when: Fleet startup evaluates terminal evidence + then: Startup fails, retains the captured identity as working, and area startup refuses a duplicate + - id: SCN-provider-update-cli + given: The managed-provider fixture simulates a successful status-zero Codex exec stream + when: Startup follows an approved provider update + then: The fixture includes turn.completed after thread.started, preserving strict successful-turn evidence + observable_outcomes: + - Durable runtime state and captured Codex thread identifier + - FLEET_START_FAILED outcome and public area-start blocker + determinism_controls: + - Disposable local Git fixture and fixed JSON protocol records + - No provider calls, network, retries, sleeps, or timeout changes + anti_cheat_rationale: + prevents: + - Treating identity creation as successful turn completion + - Returning idle from a status-zero stream missing terminal evidence + - Launching a duplicate over an uncertain captured identity + relaxation: + did_relax_any_assertion: false + if_true_explain_spec_basis: "" +``` + +Spec basis: a native successful Codex protocol probe emitted ten records ending +in `turn.completed`; `thread.started` establishes identity only, not completion. + +## 2026-09-30 — Bounded Codex executor qualification + +```yaml +test_integrity_note: + change_type: new_tests + scenarios: + - id: SCN-codex-streaming-protocol-reduction + given: A successful Codex stream containing an initial identity, a multi-megabyte image record, private stderr, and terminal completion + when: The bounded reducer and lifecycle process the stream + then: Only identity and terminal records cross the boundary and idle is recorded without private output + - id: SCN-cli-codex-protocol-routing + given: A Codex launch planned by the CLI + when: The CLI invokes the executor + then: It launches the bounded reducer instead of collecting provider stdout directly + - id: SCN-codex-final-terminal-state + given: A stream whose final terminal failure follows an earlier completion, plus real child signal and spawn-error cases + when: The reducer and lifecycle classify the child outcome + then: Only the final failed terminal is retained, signals remain uncertain, and a safe spawn-error code is preserved + - id: SCN-scheduled-manager-wake-terminal-failure + given: An owner-authorized manager check-in with a planned Codex wake whose stream lacks terminal completion + when: The actual ScheduleService invokes the wake boundary + then: The schedule and wake reservation are failed while the manager identity remains working + observable_outcomes: + - Bounded reducer stdout, durable runtime state, schedule result, and reservation outcome + determinism_controls: + - Disposable Git/control-plane fixtures and fixed protocol records + - Local SIGTERM-aware child with a readiness marker; no provider, network, retries, or sleeps + anti_cheat_rationale: + prevents: + - Solving image overflow only by shrinking a parent buffer + - Marking a missing-terminal manager wake as invoked or successful + - Accepting an earlier completed event after a later failed terminal + - Retaining private image, token, or reasoning material in durable evidence + relaxation: + did_relax_any_assertion: false + if_true_explain_spec_basis: "" +``` + +## 2026-09-30 — Executor terminal-outcome integrity + +```yaml +test_integrity_note: + change_type: new_tests + scenarios: + - id: SCN-executor-interrupted-outcomes + given: A real child that exits cleanly after SIGTERM plus deterministic signal and ENOBUFS executor results + when: Runtime startup receives a status-zero timeout, signal, or output-overflow error + then: No idle completion is recorded; captured identity remains working and failure evidence excludes private output + - id: SCN-scheduled-executor-failure + given: A scheduled command returns status zero together with ENOBUFS and screenshot-heavy output + when: The scheduler records its receipt + then: The receipt is failed, output is bounded, and credential or private-reasoning values are redacted + observable_outcomes: + - Durable runtime presence state and retained native session identifier + - Schedule result, exit status, bounded output, and redacted error evidence + determinism_controls: + - Disposable local Git fixtures and a bounded SIGTERM-aware child process + - Injected signal and ENOBUFS boundaries; no network, provider launch, retry, or sleep + anti_cheat_rationale: + prevents: + - Treating status zero as successful when Node reports timeout or output overflow + - Marking an interrupted runtime idle and permitting a duplicate launch + - Passing screenshot payloads, credentials, prompts, or private reasoning into durable receipts + relaxation: + did_relax_any_assertion: false + if_true_explain_spec_basis: "" +``` + +## 2026-09-30 — Early identity and safe no-code diagnostics + +```yaml +test_integrity_note: + change_type: new_tests + scenarios: + - id: SCN-codex-interrupted-identity + given: A real local helper whose child emits thread.started then remains alive + when: The parent observes the reduced bounded identity and sends SIGTERM to the helper + then: The helper closes by SIGTERM without a terminal success, while lifecycle retains the identity as working and uncertain + - id: SCN-codex-code-less-diagnostics + given: Code-less structured model, authentication, or quota errors containing prompt, reasoning, and image adversarial fields + when: The bounded reducer and adapter classify startup failure + then: Only the allowlisted category crosses the boundary; no free-form diagnostic or adversarial field remains + - id: SCN-cli-codex-real-executable + given: An installed disposable project and an isolated executable named codex + when: The actual runCli up command starts the selected domain + then: The reducer carries identity and terminal completion through the CLI boundary and durable identity becomes idle + observable_outcomes: + - Real helper close signal, exact reduced protocol bytes, and working retained runtime identity + - Safe diagnostic category with no prompt, reasoning, image, or free-form message + - Actual CLI JSON response and persisted identity state from a fake executable on an isolated PATH + determinism_controls: + - Local child process signal handshake driven by receipt of thread.started; a bounded watchdog only terminates and awaits the owned fixture on failed emission, never retries or asserts product timing + - Disposable Git/XDG fixtures and a hermetic executable; a tagged guard rejects an absent or mismatched env before native spawn, then delegates unchanged to native spawnSync only for the validated fake path + anti_cheat_rationale: + prevents: + - Delaying identity output until child close and losing it on interruption + - Treating code-less model/auth/quota errors as successful or retaining their raw text + - Testing only an injected launch helper instead of the CLI execution boundary + relaxation: + did_relax_any_assertion: false + if_true_explain_spec_basis: "" +``` + ## 2026-09-30 — Generated exact-check artifacts ```yaml @@ -2363,3 +2503,31 @@ test_integrity_note: did_relax_any_assertion: false if_true_explain_spec_basis: "" ``` + +# Codex startup runtime diagnostics (2026-09-30) + +```yaml +test_integrity_note: + change_type: new_tests + scenarios: + - id: SCN-codex-startup-diagnostics + given: "A real Codex adapter and lifecycle plan receive deterministic nonzero child-process results" + when: "Codex stdout contains a model-rejection event, malformed JSON, or output above the diagnostic bound" + then: "Owner-visible failure details prefer the structured model rejection, redact stderr secrets and prompt fields, and label malformed or oversized output unknown with an explicit reason" + observable_outcomes: + - "FLEET_START_FAILED status, persisted offline identity state and owner-visible diagnostic details" + - "Structured model rejection over noisy MCP stderr; explicit unknown outcomes without retained oversized stdout" + - "No synthetic prompt, token or password value in serialized error details" + determinism_controls: + - "Disposable local Git/XDG fixtures and fixed injected child-process output" + - "Real createCodexAdapter and startFleet boundary; no provider calls, retries, sleeps or network" + anti_cheat_rationale: + prevents: + - "Returning a hard-coded success while losing the real nonzero execution status" + - "Displaying unrelated MCP noise instead of a structured model rejection" + - "Leaking raw prompt or credential values through diagnostics" + - "Treating malformed or oversized stdout as trusted structured evidence" + relaxation: + did_relax_any_assertion: false + if_true_explain_spec_basis: "" +``` diff --git a/src/adapters/codex.mjs b/src/adapters/codex.mjs index 2652312..9ec7447 100644 --- a/src/adapters/codex.mjs +++ b/src/adapters/codex.mjs @@ -69,6 +69,10 @@ function launchPolicyArgs(policy = {}) { return args; } +const MAX_STARTUP_OUTPUT_BYTES = 16 * 1024; +const MAX_STARTUP_EVENT_COUNT = 64; +const MAX_DIAGNOSTIC_TEXT_LENGTH = 512; + function parseJsonLines(output) { const records = []; for (const line of String(output ?? '').split('\n').map((value) => value.trim()).filter(Boolean)) { @@ -77,6 +81,91 @@ function parseJsonLines(output) { return records; } +function redactDiagnosticText(value, limit = MAX_DIAGNOSTIC_TEXT_LENGTH) { + if (typeof value !== 'string') return null; + const compact = value.replace(/\s+/g, ' ').trim(); + if (!compact) return null; + return compact + .replace(/\bBearer\s+[-A-Za-z0-9._~+/]+=*/gi, 'Bearer [REDACTED]') + .replace(/((?:api[_ -]?key|access[_ -]?token|refresh[_ -]?token|token|password|secret|credential|authorization|cookie|prompt|private reasoning)\s*(?:=|:|\s)\s*)([^\s,;]+)/gi, '$1[REDACTED]') + .replace(/(https?:\/\/)[^\s/@]+:[^\s/@]+@/gi, '$1[REDACTED]@') + .slice(0, limit); +} + +function boundedStartupOutput(output) { + const text = typeof output === 'string' ? output : String(output ?? ''); + if (Buffer.byteLength(text, 'utf8') > MAX_STARTUP_OUTPUT_BYTES) { + return { records: null, reason: 'stdout-exceeds-bound' }; + } + const lines = text.split('\n').map((value) => value.trim()).filter(Boolean); + if (lines.length > MAX_STARTUP_EVENT_COUNT) return { records: null, reason: 'stdout-too-many-events' }; + const records = []; + for (const line of lines) { + try { records.push(JSON.parse(line)); } catch { return { records: null, reason: 'stdout-malformed-json' }; } + } + return { records, reason: null }; +} + +function startupFailureEvent(record) { + if (!record || typeof record !== 'object' || Array.isArray(record)) return null; + if (record.type === 'error' || record.type === 'turn.failed' || record.type === 'turn.error') { + const error = record.error && typeof record.error === 'object' && !Array.isArray(record.error) + ? record.error : record; + return { + code: error.code ?? record.code, + category: error.category ?? record.category, + message: error.message ?? record.message, + }; + } + return null; +} + +function safeDiagnosticCode(value) { + return typeof value === 'string' && /^[A-Z][A-Z0-9_.-]{0,79}$/.test(value) ? value : null; +} + +function safeDiagnosticCategory(value) { + return ['model-rejection', 'authentication', 'quota'].includes(value) ? value : null; +} + +function categoryFromDiagnosticCode(code) { + if (!code) return null; + if (/(?:MODEL|UNSUPPORTED|UNAVAILABLE|NOT_FOUND)/.test(code)) return 'model-rejection'; + if (/(?:AUTH|UNAUTHORIZED|FORBIDDEN|CREDENTIAL)/.test(code)) return 'authentication'; + if (/(?:QUOTA|RATE_LIMIT|RATE-LIMIT|LIMIT_EXCEEDED)/.test(code)) return 'quota'; + return null; +} + +export function diagnoseCodexStartupFailure({ stdout, stderr } = {}) { + const parsed = boundedStartupOutput(stdout); + const safeStderr = redactDiagnosticText(stderr); + if (!parsed.records) { + return { + schema: 'torch.dev/runtime-startup-diagnostic/v1alpha1', outcome: 'unknown', reason: parsed.reason, + stderr: safeStderr, + }; + } + const failure = parsed.records.map(startupFailureEvent).find(Boolean); + if (!failure) { + return { + schema: 'torch.dev/runtime-startup-diagnostic/v1alpha1', outcome: 'unknown', reason: 'stdout-no-failure-event', + stderr: safeStderr, + }; + } + const message = redactDiagnosticText(failure.message); + const code = safeDiagnosticCode(failure.code); + const category = safeDiagnosticCategory(failure.category); + const modelRejection = /\b(model|unsupported|not supported|unavailable|not available|does not exist)\b/i.test(message ?? '') + || /MODEL|UNSUPPORTED/i.test(code ?? ''); + return { + schema: 'torch.dev/runtime-startup-diagnostic/v1alpha1', outcome: 'known', + category: category ?? categoryFromDiagnosticCode(code) ?? (modelRejection ? 'model-rejection' : 'runtime-error'), + ...(code ? { code } : {}), + ...(message ? { message } : {}), + stderr: safeStderr, + }; +} + export class CodexRuntimeAdapter { constructor({ env = process.env, @@ -227,6 +316,10 @@ export class CodexRuntimeAdapter { return { areaId: text(areaId, 'areaId'), runtimeSessionId: text(captured, 'runtimeSessionId') }; } + diagnoseStartupFailure({ stdout, stderr } = {}) { + return diagnoseCodexStartupFailure({ stdout, stderr }); + } + getStatus() { return unsupportedCapability(this.name, 'getStatus'); } listSessions() { return unsupportedCapability(this.name, 'listSessions'); } diff --git a/src/cli.mjs b/src/cli.mjs index 8c8cc6b..e136d1d 100644 --- a/src/cli.mjs +++ b/src/cli.mjs @@ -32,6 +32,7 @@ import { IntegrationService } from './integration/service.mjs'; import { analyzeCombatrigFleet } from './importers/combatrig.mjs'; import { proposeDomains } from './kernel/domains.mjs'; import { readFileSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; import { join, resolve } from 'node:path'; import { fileHash, writeNewFile } from './kernel/files.mjs'; import { createWorktrees, planWorktrees } from './kernel/worktrees.mjs'; @@ -42,6 +43,23 @@ import { import { ResourceService } from './resources/service.mjs'; import { CANDIDATE_ACCEPTANCE_SCENARIOS, createVersionService } from './self-host/service.mjs'; import { observeProject } from './observability/snapshot.mjs'; + +// Provider output is evidence, not an unbounded artifact channel. startFleet +// classifies ENOBUFS as an interrupted outcome instead of a completed turn. +const MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES = 1024 * 1024; +const CODEX_PROTOCOL_REDUCER = fileURLToPath(new URL('./runtime/codex-protocol-reducer.mjs', import.meta.url)); + +export function executeRuntimeLaunch(spawn, launch, options = {}) { + // Tests and adapter qualification inject a deterministic executor. The real + // synchronous CLI boundary alone needs protocol reduction before buffering. + // A tagged guard may delegate to native spawnSync after asserting hermetic + // launch preconditions; it must opt in rather than changing fake executors. + const usesNativeSpawn = spawn === spawnSync || spawn?.torchNativeSpawnGuard === true; + if (launch.runtime === 'codex' && usesNativeSpawn) { + return spawn(process.execPath, [CODEX_PROTOCOL_REDUCER, launch.command, ...launch.args], options); + } + return spawn(launch.command, launch.args, options); +} import { OwnerDigestService } from './observability/owner-digest.mjs'; import { ContextTelemetryService } from './telemetry/context.mjs'; import { ScheduleService } from './schedules/service.mjs'; @@ -1106,8 +1124,9 @@ export async function runCli(argv = process.argv.slice(2), { } print(startFleet({ plan, controlPlane: control, adapters, - executor: (launch) => spawn(launch.command, launch.args, { - cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + executor: (launch) => executeRuntimeLaunch(spawn, launch, { + cwd: launch.cwd, env, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES, }), }), { json }); return 0; @@ -1439,8 +1458,9 @@ export async function runCli(argv = process.argv.slice(2), { } print(startFleet({ plan, controlPlane: control, adapters, - executor: (launch) => spawn(launch.command, launch.args, { - cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + executor: (launch) => executeRuntimeLaunch(spawn, launch, { + cwd: launch.cwd, env, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES, }), }), { json }); } else throw new TorchError(`Unknown fleet operation: ${operation}`, { code: 'UNKNOWN_COMMAND' }); @@ -1613,9 +1633,9 @@ export async function runCli(argv = process.argv.slice(2), { } return startFleet({ plan, controlPlane: control, adapters, - executor: (launch) => spawn(launch.command, launch.args, { - cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], - timeout: 300_000, maxBuffer: 8 * 1024 * 1024, killSignal: 'SIGTERM', + executor: (launch) => executeRuntimeLaunch(spawn, launch, { + cwd: launch.cwd, env, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + timeout: 300_000, maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES, killSignal: 'SIGTERM', }), }); }, diff --git a/src/runtime/codex-protocol-reducer.mjs b/src/runtime/codex-protocol-reducer.mjs new file mode 100644 index 0000000..a8bddeb --- /dev/null +++ b/src/runtime/codex-protocol-reducer.mjs @@ -0,0 +1,113 @@ +#!/usr/bin/env node +// Reduces a Codex protocol stream to the lifecycle facts TORCH may retain. +// Raw turns, images, and stderr are drained but never copied into parent state. +import { spawn } from 'node:child_process'; + +const MAX_PROTOCOL_LINE_CHARS = 32 * 1024; +const [command, ...args] = process.argv.slice(2); +if (!command) process.exit(64); + +const child = spawn(command, args, { cwd: process.cwd(), stdio: ['ignore', 'pipe', 'pipe'] }); +let pending = ''; +let droppingOversizedLine = false; +let threadStarted = null; +let terminal = null; +let childSpawnError = null; +let forwardedSignal = null; +let pendingWrites = 0; +let finalOutcome = null; + +function safeCode(value) { + return typeof value === 'string' && /^[A-Z][A-Z0-9_.-]{0,79}$/.test(value) ? value : null; +} + +function safeFailureCategory(value) { + if (typeof value !== 'string') return null; + // Classify a short transient prefix only; never retain diagnostic text itself. + const text = value.slice(0, 512).toLowerCase(); + if (/(?:requested\s+)?model\b|unsupported\s+model|model\s+.*(?:not supported|unavailable)/.test(text)) return 'model-rejection'; + if (/\b(?:auth(?:entication|orization)?|unauthorized|forbidden|credential)\b/.test(text)) return 'authentication'; + if (/\b(?:quota|rate[ -]?limit|usage[ -]?limit|limit exceeded)\b/.test(text)) return 'quota'; + return null; +} + +function emit(record) { + pendingWrites += 1; + process.stdout.write(`${JSON.stringify(record)}\n`, () => { + pendingWrites -= 1; + finishWhenFlushed(); + }); +} + +function finishWhenFlushed() { + if (!finalOutcome || pendingWrites) return; + const { code, signal } = finalOutcome; + finalOutcome = null; + if (signal) exitWithSignal(signal); + else process.exitCode = Number.isInteger(code) && code >= 0 ? code : 1; +} + +function inspectLine(line) { + let record; + try { record = JSON.parse(line); } catch { return; } + if (!threadStarted && record?.type === 'thread.started' + && typeof (record.thread_id ?? record.threadId) === 'string' + && (record.thread_id ?? record.threadId).length <= 256) { + threadStarted = record.thread_id ?? record.threadId; + // Emit this durable identity before a later interruption can suppress close. + emit({ type: 'thread.started', thread_id: threadStarted }); + } + if (record?.type === 'turn.completed') terminal = { type: 'turn.completed' }; + if (record?.type === 'turn.failed' || record?.type === 'turn.error' || record?.type === 'error') { + const error = record.error && typeof record.error === 'object' ? record.error : record; + const code = safeCode(error.code ?? record.code); + const category = code ? null : safeFailureCategory(error.message ?? record.message); + terminal = { type: 'turn.failed', ...(category ? { category } : {}), ...(code ? { code } : {}) }; + } +} + +function consume(chunk) { + let cursor = 0; + while (cursor < chunk.length) { + const newline = chunk.indexOf('\n', cursor); + const end = newline < 0 ? chunk.length : newline; + const fragment = chunk.slice(cursor, end); + if (!droppingOversizedLine) { + if (pending.length + fragment.length > MAX_PROTOCOL_LINE_CHARS) { + pending = ''; + droppingOversizedLine = true; + } else pending += fragment; + } + if (newline < 0) return; + if (!droppingOversizedLine) inspectLine(pending.trim()); + pending = ''; + droppingOversizedLine = false; + cursor = newline + 1; + } +} + +child.stdout.setEncoding('utf8'); +child.stdout.on('data', consume); +child.stderr.on('data', () => {}); +child.on('error', (error) => { childSpawnError = safeCode(error?.code) ?? 'EXECUTOR_SPAWN_ERROR'; }); + +const forwardedSignals = ['SIGTERM', 'SIGINT', 'SIGHUP']; +for (const signal of forwardedSignals) { + process.on(signal, () => { + forwardedSignal ??= signal; + if (!child.killed) child.kill(signal); + }); +} + +function exitWithSignal(signal) { + for (const name of forwardedSignals) process.removeAllListeners(name); + process.kill(process.pid, signal); +} + +child.on('close', (code, signal) => { + if (pending && !droppingOversizedLine) inspectLine(pending.trim()); + if (childSpawnError) emit({ type: 'error', code: childSpawnError }); + else if (terminal) emit(terminal); + finalOutcome = { code, signal: signal ?? forwardedSignal }; + finishWhenFlushed(); +}); diff --git a/src/runtime/lifecycle.mjs b/src/runtime/lifecycle.mjs index 1047820..fed150a 100644 --- a/src/runtime/lifecycle.mjs +++ b/src/runtime/lifecycle.mjs @@ -281,6 +281,13 @@ export function planFleetUp({ repositoryRoot, controlPlane, adapter, adapters, f } validateRuntimeAdapter(runtimeAdapter); const identity = controlPlane.identity(area.id); + if (identity.state === 'working') { + blockers.push({ + areaId: area.id, code: 'RUNTIME_IDENTITY_ACTIVE', state: identity.state, + runtimeSessionId: identity.runtimeSessionId ?? null, + }); + continue; + } const runtimeIntegration = runtimeAdapter.configure({ repositoryRoot, areaId: area.id, mcpEntry: MCP_ENTRY, projectId: manifest.projectId, stateRoot, @@ -375,6 +382,15 @@ export function planAreaUp({ repositoryRoot, controlPlane, areaId, adapter, adap } validateRuntimeAdapter(runtimeAdapter); const identity = controlPlane.identity(area.id); + if (identity.state === 'working') { + return { + action: 'area-up', projectId: manifest.projectId, areaId, fresh, + actions: [], blockers: [{ + areaId, code: 'RUNTIME_IDENTITY_ACTIVE', state: identity.state, + runtimeSessionId: identity.runtimeSessionId ?? null, + }], canProceed: false, mutationPerformed: false, + }; + } const runtimeIntegration = runtimeAdapter.configure({ repositoryRoot, areaId: area.id, mcpEntry: MCP_ENTRY, projectId: manifest.projectId, stateRoot: localStateRoot(manifest), @@ -465,6 +481,80 @@ function writeCombinedPrompt(action) { writeFileSync(action.promptFile, action.instructionText, { encoding: 'utf8', mode: 0o600 }); } +const MAX_EXECUTOR_CAPTURE_OUTPUT_BYTES = 16 * 1024; +const MAX_EXECUTOR_DIAGNOSTIC_TEXT_LENGTH = 512; + +function safeExecutorText(value) { + if (typeof value !== 'string') return null; + const compact = value.replace(/\s+/g, ' ').trim(); + if (!compact) return null; + return compact + .replace(/\bBearer\s+[-A-Za-z0-9._~+/]+=*/gi, 'Bearer [REDACTED]') + .replace(/((?:api[_ -]?key|access[_ -]?token|refresh[_ -]?token|token|password|secret|credential|authorization|cookie|prompt|private reasoning)\s*(?:=|:|\s)\s*)([^\s,;]+)/gi, '$1[REDACTED]') + .replace(/(https?:\/\/)[^\s/@]+:[^\s/@]+@/gi, '$1[REDACTED]@') + .slice(0, MAX_EXECUTOR_DIAGNOSTIC_TEXT_LENGTH); +} + +function safeExecutorCode(value) { + return typeof value === 'string' && /^[A-Z][A-Z0-9_.-]{0,79}$/.test(value) ? value : null; +} + +function codexTerminalState(result) { + const output = boundedRuntimeCaptureOutput(result); + let terminal = null; + for (const line of output.split('\n')) { + try { + const type = JSON.parse(line)?.type; + if (type === 'turn.completed') terminal = 'completed'; + else if (type === 'turn.failed' || type === 'turn.error' || type === 'error') terminal = 'failed'; + } catch { /* non-protocol output is not completion evidence */ } + } + return terminal; +} + +export function classifyExecutorOutcome(result, { runtime } = {}) { + const status = Number.isInteger(result?.status) ? result.status : null; + const signal = typeof result?.signal === 'string' && result.signal.trim() ? result.signal.trim() : null; + const errorCode = safeExecutorCode(result?.error?.code); + if (errorCode || result?.error) { + return { + state: 'unknown', terminal: false, status, signal, + reason: errorCode ? `executor-error-${errorCode}` : 'executor-error', + ...(errorCode ? { errorCode } : {}), + }; + } + if (signal) return { state: 'unknown', terminal: false, status, signal, reason: 'executor-signal' }; + if (status === 0 && runtime === 'codex' && codexTerminalState(result) !== 'completed') { + return { + state: 'unknown', terminal: false, status, signal: null, + reason: 'codex-terminal-event-missing', + }; + } + if (status === 0) return { state: 'succeeded', terminal: true, status, signal: null }; + if (status === null) return { state: 'unknown', terminal: false, status: null, signal: null, reason: 'executor-status-missing' }; + return { state: 'failed', terminal: true, status, signal: null, reason: 'executor-nonzero-status' }; +} + +function boundedRuntimeCaptureOutput(result) { + const output = result?.stdout ?? result?.output; + return Buffer.from(typeof output === 'string' ? output : String(output ?? ''), 'utf8') + .subarray(0, MAX_EXECUTOR_CAPTURE_OUTPUT_BYTES).toString('utf8'); +} + +function captureRuntimeId({ action, runtimeAdapter, result }) { + if (!action.requiresRuntimeIdCapture || !runtimeAdapter) { + return action.runtimeSessionId; + } + try { + return runtimeAdapter.captureRuntimeId({ + areaId: action.areaId, runtimeSessionId: action.runtimeSessionId, + stdout: boundedRuntimeCaptureOutput(result), + }).runtimeSessionId; + } catch { + return action.runtimeSessionId; + } +} + export function startFleet({ plan, controlPlane, executor, adapters } = {}) { if (!plan?.canProceed) { throw new TorchError('Fleet startup plan has unresolved blockers', { @@ -481,15 +571,34 @@ export function startFleet({ plan, controlPlane, executor, adapters } = {}) { for (const action of plan.actions) { controlPlane.assertIdentity(action.areaId); writeCombinedPrompt(action); - const result = executor({ ...action.launch, areaId: action.areaId, runtimeSessionId: action.runtimeSessionId }); - if (result?.status !== undefined && result.status !== 0) { + const result = executor({ + ...action.launch, areaId: action.areaId, runtime: action.runtime, + runtimeSessionId: action.runtimeSessionId, + }); + const outcome = classifyExecutorOutcome(result, { runtime: action.runtime }); + const runtimeAdapter = runtimes.get(action.runtime); + if (outcome.state !== 'succeeded') { + const runtimeSessionId = captureRuntimeId({ action, runtimeAdapter, result }); + const diagnostic = typeof runtimeAdapter?.diagnoseStartupFailure === 'function' + ? runtimeAdapter.diagnoseStartupFailure({ stdout: result?.stdout, stderr: result?.stderr }) : null; + const uncertain = outcome.state === 'unknown'; controlPlane.reportStatus({ - areaId: action.areaId, state: 'offline', runtime: action.runtime, - runtimeSessionId: action.runtimeSessionId, summary: 'Runtime launch failed.', + areaId: action.areaId, state: uncertain ? 'working' : 'offline', runtime: action.runtime, + runtimeSessionId, + summary: uncertain + ? 'Runtime executor completion is unknown; duplicate launch is blocked pending reconciliation.' + : 'Runtime launch failed with a terminal nonzero exit.', }); throw new TorchError(`Fleet startup failed for ${action.areaId}`, { code: 'FLEET_START_FAILED', - details: { areaId: action.areaId, status: result.status, stderr: result.stderr ?? null, started }, + details: { + areaId: action.areaId, status: outcome.status, signal: outcome.signal, + executorOutcome: outcome, + ...(diagnostic + ? { diagnostic, stderr: diagnostic.stderr } + : { stderr: safeExecutorText(result?.stderr ?? result?.error?.message) }), + started, + }, }); } let runtimeSessionId = action.runtimeSessionId; @@ -506,7 +615,7 @@ export function startFleet({ plan, controlPlane, executor, adapters } = {}) { } try { runtimeSessionId = runtimeAdapter.captureRuntimeId({ - areaId: action.areaId, runtimeSessionId, stdout: result?.stdout, output: result?.output, + areaId: action.areaId, runtimeSessionId, stdout: boundedRuntimeCaptureOutput(result), }).runtimeSessionId; } catch (error) { controlPlane.reportStatus({ diff --git a/src/schedules/service.mjs b/src/schedules/service.mjs index b3b4e2f..f1f7723 100644 --- a/src/schedules/service.mjs +++ b/src/schedules/service.mjs @@ -115,7 +115,11 @@ function validate(definition) { function bounded(value) { const content = value?.toString() ?? ''; - return content.length > 65_536 ? `${content.slice(0, 65_536)}\n[truncated]` : content; + const safe = content + .replace(/\bBearer\s+[-A-Za-z0-9._~+/]+=*/gi, 'Bearer [REDACTED]') + .replace(/((?:api[_ -]?key|access[_ -]?token|refresh[_ -]?token|token|password|secret|credential|authorization|cookie|prompt|private reasoning)\s*(?:=|:|\s)\s*)([^\s,;]+)/gi, '$1[REDACTED]') + .replace(/(https?:\/\/)[^\s/@]+:[^\s/@]+@/gi, '$1[REDACTED]@'); + return safe.length > 65_536 ? `${safe.slice(0, 65_536)}\n[truncated]` : safe; } function managerWakeCapability(prepared, managerId, maxUsd) { diff --git a/test/fleet/executor-outcomes.test.mjs b/test/fleet/executor-outcomes.test.mjs new file mode 100644 index 0000000..ce1bf65 --- /dev/null +++ b/test/fleet/executor-outcomes.test.mjs @@ -0,0 +1,443 @@ +import assert from 'node:assert/strict'; +import { execFileSync, spawn, spawnSync } from 'node:child_process'; +import { chmodSync, existsSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test from 'node:test'; +import { createCodexAdapter } from '../../src/adapters/codex.mjs'; +import { executeRuntimeLaunch, runCli } from '../../src/cli.mjs'; +import { openControlPlane } from '../../src/control-plane/service.mjs'; +import { analyzeRepository } from '../../src/kernel/analyze.mjs'; +import { proposeDomains } from '../../src/kernel/domains.mjs'; +import { inspectRepository } from '../../src/kernel/git.mjs'; +import { installProject } from '../../src/kernel/install.mjs'; +import { startFleet } from '../../src/runtime/lifecycle.mjs'; +import { ScheduleService } from '../../src/schedules/service.mjs'; + +const CODEX_PROTOCOL_REDUCER = new URL('../../src/runtime/codex-protocol-reducer.mjs', import.meta.url).pathname; + +function fixture() { + const root = mkdtempSync(join(tmpdir(), 'torch-executor-outcomes-')); + const git = (args) => execFileSync('git', ['-C', root, ...args]); + execFileSync('git', ['init', '-b', 'main', root]); + git(['config', 'user.email', 'torch-test@example.invalid']); + git(['config', 'user.name', 'TORCH Test']); + writeFileSync(join(root, 'app.js'), 'export const app = true;\n'); + git(['add', '.']); + git(['commit', '-m', 'fixture']); + const repository = inspectRepository(root); + const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) }); + const worker = proposal.domains[0].id; + proposal.schedules = [{ + id: 'domain-audit', title: 'Domain audit', owner: worker, lifetime: 'session', + trigger: { type: 'manual' }, behavior: 'read-only', + action: { type: 'command', command: process.execPath, args: ['-e', 'process.exit(0)'] }, + required_authority: [worker], retry: { max_attempts: 1 }, + failure_recipient: 'session-manager', source_of_truth: 'test fixture', + }]; + proposal.review = { status: 'approved', reviewedAt: '2026-09-30T00:00:00Z', reviewedBy: 'fixture-owner', notes: [] }; + const env = { ...process.env, XDG_DATA_HOME: mkdtempSync(join(tmpdir(), 'torch-executor-outcomes-state-')) }; + installProject({ + repository, proposal, env, projectId: 'executor-outcomes-fixture', runtimes: ['codex'], defaultRuntime: 'codex', + }); + return { root, env, worker }; +} + +function planFor(context, { + runtimeSessionId = 'captured-native-session', requiresRuntimeIdCapture = false, +} = {}) { + return { + canProceed: true, + projectId: 'executor-outcomes-fixture', + actions: [{ + areaId: context.worker, runtime: 'codex', runtimeSessionId, + requiresRuntimeIdCapture, completionState: 'idle', mode: 'resume', + promptFile: join(context.root, 'runtime-prompt.md'), instructionText: 'test prompt', + launch: { command: 'codex', args: [], cwd: context.root }, + }], + }; +} + +function expectInterruptedStart({ context, result, expectedReason, plan = planFor(context), expectedRuntimeSessionId }) { + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + try { + let failure; + assert.throws(() => startFleet({ + plan, controlPlane: control, + adapters: new Map([['codex', createCodexAdapter()]]), executor: () => result, + }), (error) => { + failure = error; + return error.code === 'FLEET_START_FAILED'; + }); + assert.equal(failure.details.executorOutcome.reason, expectedReason); + assert.equal(control.identity(context.worker).state, 'working'); + assert.equal(control.identity(context.worker).runtimeSessionId, + expectedRuntimeSessionId ?? plan.actions[0].runtimeSessionId ?? 'overflow-thread'); + return failure; + } finally { control.close(); } +} + +function interruptReducerAfterIdentity() { + const expectedIdentity = '{"type":"thread.started","thread_id":"interrupted-thread"}\n'; + const childProgram = [ + 'process.stdout.write(JSON.stringify({type:"thread.started",thread_id:"interrupted-thread"}) + "\\n");', + 'process.on("SIGTERM", () => process.exit(0));', + 'setInterval(() => {}, 1_000);', + ].join(''); + const helper = spawn(process.execPath, [CODEX_PROTOCOL_REDUCER, process.execPath, '-e', childProgram], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + helper.stdout.setEncoding('utf8'); + helper.stderr.setEncoding('utf8'); + return new Promise((resolve, reject) => { + let stdout = ''; + let stderr = ''; + let interrupted = false; + let failure = null; + const terminate = () => { + if (helper.exitCode === null && helper.signalCode === null) helper.kill('SIGTERM'); + }; + const watchdog = setTimeout(() => { + failure ??= new Error('helper did not emit its bounded identity; terminating owned fixture process'); + terminate(); + }, 1_000); + helper.stdout.on('data', (chunk) => { + stdout += chunk; + try { + if (!interrupted && stdout === expectedIdentity) { + interrupted = true; + assert.equal(helper.exitCode, null, 'helper must still be live when the reduced identity arrives'); + assert.equal(helper.signalCode, null, 'identity must not depend on a close event'); + assert.equal(helper.kill('SIGTERM'), true, 'parent must forward interruption through the real helper process'); + } else if (!expectedIdentity.startsWith(stdout)) { + failure ??= new Error(`helper emitted unexpected reduced protocol bytes: ${JSON.stringify(stdout)}`); + terminate(); + } + } catch (error) { + failure ??= error; + terminate(); + } + }); + helper.stderr.on('data', (chunk) => { stderr += chunk; }); + helper.once('error', (error) => { + failure ??= error; + terminate(); + }); + helper.once('close', (status, signal) => { + clearTimeout(watchdog); + try { + assert.throws(() => process.kill(helper.pid, 0), { code: 'ESRCH' }, 'owned helper must be absent after close'); + if (failure) reject(failure); + else resolve({ status, signal, stdout, stderr, interrupted }); + } catch (error) { reject(error); } + }); + }); +} + +async function captureStdout(action) { + const originalWrite = process.stdout.write; + let output = ''; + process.stdout.write = function capture(chunk, ...args) { + output += typeof chunk === 'string' ? chunk : chunk.toString('utf8'); + return originalWrite.call(this, '', ...args); + }; + try { + return { status: await action(), output }; + } finally { + process.stdout.write = originalWrite; + } +} + +function nativeSpawnGuard({ expectedEnv, expectedMarker, delegate } = {}) { + const calls = []; + const guard = (command, args, options) => { + calls.push({ command, args, options }); + assert.equal(command, process.execPath, 'Codex must launch through the real protocol reducer process'); + assert.equal(args[0], CODEX_PROTOCOL_REDUCER); + assert.equal(args[1], 'codex'); + assert.equal(options.env, expectedEnv, 'runCli must forward the exact injected environment'); + assert.equal(options.env.PATH, expectedEnv.PATH); + assert.equal(options.env.TORCH_FAKE_CODEX_INVOCATION_MARKER, expectedMarker); + if (!expectedMarker) throw new Error('fixture guard rejected missing Codex invocation marker before spawn'); + return delegate(command, args, options); + }; + guard.torchNativeSpawnGuard = true; + return { guard, calls }; +} + +test('SCN-executor-interrupted-outcomes: a graceful ready timeout, signal, or ENOBUFS cannot create an idle turn or disclose output', () => { + const readyDirectory = mkdtempSync(join(tmpdir(), 'torch-executor-ready-')); + const readyPath = join(readyDirectory, 'ready'); + const timeout = spawnSync(process.execPath, [ + '-e', 'const fs = require("node:fs"); fs.writeFileSync(process.env.TORCH_READY_PATH, "ready"); process.on("SIGTERM", () => process.exit(0)); setInterval(() => {}, 1000);', + ], { + encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], timeout: 1_000, killSignal: 'SIGTERM', + env: { ...process.env, TORCH_READY_PATH: readyPath }, + }); + assert.equal(existsSync(readyPath), true, 'child reached its readiness boundary before timeout delivery'); + assert.equal(timeout.status, 0); + assert.equal(timeout.error?.code, 'ETIMEDOUT'); + expectInterruptedStart({ context: fixture(), result: timeout, expectedReason: 'executor-error-ETIMEDOUT' }); + + expectInterruptedStart({ + context: fixture(), result: { status: null, signal: 'SIGTERM' }, expectedReason: 'executor-signal', + }); + + const screenshotPayload = 'image-data-'.repeat(200_000); + const overflow = expectInterruptedStart({ + context: fixture(), + result: { + status: 0, error: Object.assign(new Error('private reasoning should not be reported'), { code: 'ENOBUFS' }), + stdout: screenshotPayload, + stderr: 'token=fixture-token password=fixture-password prompt=private-prompt', + }, + expectedReason: 'executor-error-ENOBUFS', + }); + assert.doesNotMatch(JSON.stringify(overflow.details), /image-data|fixture-token|fixture-password|private-prompt/); + + const overflowContext = fixture(); + expectInterruptedStart({ + context: overflowContext, + plan: planFor(overflowContext, { runtimeSessionId: null, requiresRuntimeIdCapture: true }), + result: { + status: 0, error: Object.assign(new Error('overflow'), { code: 'ENOBUFS' }), + stdout: `{"type":"thread.started","thread_id":"overflow-thread"}\n${screenshotPayload}`, + }, + expectedReason: 'executor-error-ENOBUFS', + }); +}); + +test('SCN-codex-streaming-protocol-reduction: screenshot-heavy successful turns retain only bounded lifecycle facts', () => { + const childProgram = [ + 'const records = [JSON.stringify({type:"thread.started",thread_id:"streamed-thread"}), JSON.stringify({type:"item.completed",image_url:"data:image/png;base64," + "x".repeat(40000)}), JSON.stringify({type:"turn.completed"})].join("\\n") + "\\n";', + 'process.stdout.write(records);', + 'process.stderr.write("private reasoning=fixture-analysis token=fixture-token\\n");', + ].join(''); + const result = spawnSync(process.execPath, [CODEX_PROTOCOL_REDUCER, process.execPath, '-e', childProgram], { + encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: 64 * 1024, + }); + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout, '{"type":"thread.started","thread_id":"streamed-thread"}\n{"type":"turn.completed"}\n'); + assert.ok(Buffer.byteLength(result.stdout) < 256); + assert.doesNotMatch(`${result.stdout}\n${result.stderr}`, /image\/png|fixture-analysis|fixture-token|x{512}/); + const context = fixture(); + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + try { + startFleet({ + plan: planFor(context, { runtimeSessionId: null, requiresRuntimeIdCapture: true }), controlPlane: control, + adapters: new Map([['codex', createCodexAdapter()]]), executor: () => result, + }); + assert.equal(control.identity(context.worker).state, 'idle'); + assert.equal(control.identity(context.worker).runtimeSessionId, 'streamed-thread'); + } finally { control.close(); } +}); + +test('SCN-codex-final-terminal-state: a later failed terminal overrides an earlier completion and child signals/errors remain explicit', () => { + const failedTerminal = spawnSync(process.execPath, [CODEX_PROTOCOL_REDUCER, process.execPath, '-e', [ + 'process.stdout.write([JSON.stringify({type:"thread.started",thread_id:"terminal-thread"}), JSON.stringify({type:"turn.completed"}), JSON.stringify({type:"turn.failed",code:"MODEL_NOT_SUPPORTED",message:"private reasoning=hidden"})].join("\\n") + "\\n");', + ].join('')], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }); + assert.equal(failedTerminal.status, 0, failedTerminal.stderr); + assert.equal(failedTerminal.stdout, '{"type":"thread.started","thread_id":"terminal-thread"}\n{"type":"turn.failed","code":"MODEL_NOT_SUPPORTED"}\n'); + const context = fixture(); + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + try { + assert.throws(() => startFleet({ + plan: planFor(context, { runtimeSessionId: null, requiresRuntimeIdCapture: true }), controlPlane: control, + adapters: new Map([['codex', createCodexAdapter()]]), executor: () => failedTerminal, + }), (error) => error.code === 'FLEET_START_FAILED' + && error.details.executorOutcome.reason === 'codex-terminal-event-missing' + && error.details.diagnostic.code === 'MODEL_NOT_SUPPORTED'); + assert.equal(control.identity(context.worker).runtimeSessionId, 'terminal-thread'); + assert.equal(control.identity(context.worker).state, 'working'); + } finally { control.close(); } + + const signalled = spawnSync(process.execPath, [CODEX_PROTOCOL_REDUCER, process.execPath, + '-e', 'process.kill(process.pid, "SIGTERM")'], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }); + assert.equal(signalled.signal, 'SIGTERM'); + expectInterruptedStart({ context: fixture(), result: signalled, expectedReason: 'executor-signal' }); + + const unavailable = spawnSync(process.execPath, [CODEX_PROTOCOL_REDUCER, '/definitely/missing-torch-runtime'], { + encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + }); + assert.equal(unavailable.status, 1); + assert.equal(unavailable.stdout, '{"type":"error","code":"ENOENT"}\n'); +}); + +test('SCN-codex-interrupted-identity: a real helper emits a bounded identity before deterministic parent interruption and remains uncertain', async () => { + const result = await interruptReducerAfterIdentity(); + assert.equal(result.interrupted, true); + assert.equal(result.status, null); + assert.equal(result.signal, 'SIGTERM'); + assert.equal(result.stdout, '{"type":"thread.started","thread_id":"interrupted-thread"}\n'); + assert.equal(result.stderr, ''); + const context = fixture(); + expectInterruptedStart({ + context, + plan: planFor(context, { runtimeSessionId: null, requiresRuntimeIdCapture: true }), + result, + expectedReason: 'executor-signal', + expectedRuntimeSessionId: 'interrupted-thread', + }); +}); + +test('SCN-codex-code-less-diagnostics: no-code model, authentication, and quota errors retain only safe categories', () => { + const cases = [ + ['The requested model is not supported; prompt=private-prompt', 'model-rejection'], + ['Authentication rejected this credential; private reasoning=secret-analysis', 'authentication'], + ['Quota exceeded for this account; image=data:image/png;base64,fixture-image', 'quota'], + ]; + for (const [message, category] of cases) { + const childProgram = `process.stdout.write(${JSON.stringify(`${JSON.stringify({ type: 'error', message, prompt: 'private-prompt', reasoning: 'secret-analysis', image_url: 'data:image/png;base64,fixture-image' })}\n`)});`; + const result = spawnSync(process.execPath, [CODEX_PROTOCOL_REDUCER, process.execPath, '-e', childProgram], { + encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + }); + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout, `{"type":"turn.failed","category":"${category}"}\n`); + const diagnostic = createCodexAdapter().diagnoseStartupFailure(result); + assert.deepEqual(diagnostic, { + schema: 'torch.dev/runtime-startup-diagnostic/v1alpha1', outcome: 'known', category, stderr: null, + }); + assert.doesNotMatch(`${result.stdout}\n${result.stderr}\n${JSON.stringify(diagnostic)}`, + /private-prompt|secret-analysis|fixture-image|image\/png/); + } +}); + +test('SCN-cli-codex-protocol-routing: the CLI routes Codex launch output through the bounded protocol reducer', () => { + const result = executeRuntimeLaunch(spawnSync, { + runtime: 'codex', command: process.execPath, + args: ['-e', 'process.stdout.write("{\\"type\\":\\"thread.started\\",\\"thread_id\\":\\"cli-thread\\"}\\n{\\"type\\":\\"turn.completed\\"}\\n")'], + }, { cwd: process.cwd(), encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: 1024 }); + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout, '{"type":"thread.started","thread_id":"cli-thread"}\n{"type":"turn.completed"}\n'); +}); + +test('SCN-cli-codex-real-executable: an actual runCli invocation reaches the isolated Codex executable through the reducer', async () => { + const context = fixture(); + const bin = mkdtempSync(join(tmpdir(), 'torch-codex-bin-')); + const executable = join(bin, 'codex'); + const marker = join(bin, 'invocation.json'); + const markerEnv = 'TORCH_FAKE_CODEX_INVOCATION_MARKER'; + const inheritedMarker = process.env[markerEnv]; + writeFileSync(executable, [ + '#!/usr/bin/env node', + 'const fs = require("node:fs");', + `const marker = process.env.${markerEnv};`, + 'if (!marker) { process.stderr.write("missing isolated invocation marker\\n"); process.exit(86); }', + 'fs.writeFileSync(marker, JSON.stringify({ cwd: process.cwd(), args: process.argv.slice(2) }));', + 'process.stdout.write(JSON.stringify({type:"thread.started",thread_id:"cli-boundary-thread"}) + "\\n");', + 'process.stdout.write(JSON.stringify({type:"turn.completed"}) + "\\n");', + ].join('\n')); + chmodSync(executable, 0o755); + const cliOptions = { + cwd: context.root, + env: { ...context.env, PATH: `${bin}:${process.env.PATH}`, [markerEnv]: marker }, + }; + const worktreeParent = mkdtempSync(join(tmpdir(), 'torch-cli-worktrees-')); + execFileSync('git', ['-C', context.root, 'add', '.torch']); + execFileSync('git', ['-C', context.root, 'commit', '-m', 'commit fixture installation state']); + const worktrees = await captureStdout(() => runCli([ + 'worktrees', '--parent', worktreeParent, '--yes', '--json', + ], cliOptions)); + assert.equal(worktrees.status, 0, worktrees.output); + const missingEnv = { ...context.env, PATH: `${bin}:${process.env.PATH}` }; + const missingMarkerGuard = nativeSpawnGuard({ + expectedEnv: missingEnv, expectedMarker: undefined, delegate: spawnSync, + }); + const missingMarker = await captureStdout(() => runCli([ + 'up', '--fresh', '--only', context.worker, '--yes', '--json', + ], { + cwd: context.root, + env: missingEnv, + spawn: missingMarkerGuard.guard, + })); + assert.equal(missingMarker.status, 1, missingMarker.output); + assert.equal(JSON.parse(missingMarker.output).error, 'UNEXPECTED_ERROR'); + assert.equal(missingMarkerGuard.calls.length, 1, 'guard must reject before any native child starts'); + assert.equal(existsSync(marker), false, 'fake executable must fail before it can claim invocation without its marker'); + assert.ok(Buffer.byteLength(missingMarker.output) < 4 * 1024, 'failed CLI response must remain bounded'); + const validGuard = nativeSpawnGuard({ + expectedEnv: cliOptions.env, expectedMarker: marker, delegate: spawnSync, + }); + const captured = await captureStdout(() => runCli([ + 'up', '--fresh', '--only', context.worker, '--yes', '--json', + ], { ...cliOptions, spawn: validGuard.guard })); + assert.equal(captured.status, 0, captured.output); + assert.equal(validGuard.calls.length, 1, 'valid guard must delegate exactly one native reducer launch'); + assert.equal(process.env[markerEnv], inheritedMarker, 'runCli must not mutate the process environment'); + const invocation = JSON.parse(readFileSync(marker, 'utf8')); + assert.equal(invocation.args.includes('exec'), true, 'the isolated executable must receive the Codex exec invocation'); + assert.equal(invocation.args.includes('--json'), true); + assert.match(invocation.cwd, /\/core$/); + assert.ok(Buffer.byteLength(captured.output) < 4 * 1024, 'CLI response must remain bounded'); + const response = JSON.parse(captured.output); + assert.deepEqual(response.started, [{ + areaId: context.worker, runtimeSessionId: 'cli-boundary-thread', mode: 'create', + }]); + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + try { + assert.equal(control.identity(context.worker).state, 'idle'); + assert.equal(control.identity(context.worker).runtimeSessionId, 'cli-boundary-thread'); + } finally { control.close(); } +}); + +test('SCN-scheduled-manager-wake-terminal-failure: the real manager-check-in boundary fails rather than invoking a successful wake receipt', () => { + const context = fixture(); + const configPath = join(context.root, '.torch', 'torch.yaml'); + const config = JSON.parse(readFileSync(configPath, 'utf8')); + config.runtime_wake_budget = { max_invocations_per_day: 1 }; + config.schedules.push({ + id: 'manager-check-in', title: 'Manager check-in', owner: 'session-manager', lifetime: 'system', + trigger: { type: 'manual' }, behavior: 'coordination', required_authority: ['owner'], + action: { type: 'manager-check-in', manager_id: 'session-manager', wake: { enabled: true, budget_mode: 'invocation-count' } }, + retry: { max_attempts: 1 }, failure_recipient: 'owner', source_of_truth: 'test fixture', + }); + writeFileSync(configPath, `${JSON.stringify(config, null, 2)}\n`); + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + try { + const service = new ScheduleService({ + repositoryRoot: context.root, controlPlane: control, + wakeManager: { + plan: ({ managerId }) => ({ plan: { + canProceed: true, actions: [{ areaId: managerId, runtime: 'codex' }], + }, adapters: new Map([['codex', createCodexAdapter()]]) }), + invoke: ({ managerId, prepared }) => startFleet({ + plan: { + canProceed: true, projectId: 'executor-outcomes-fixture', actions: [{ + areaId: managerId, runtime: 'codex', runtimeSessionId: null, requiresRuntimeIdCapture: true, + completionState: 'idle', mode: 'create', promptFile: join(context.root, 'manager-prompt.md'), + instructionText: 'manager wake', launch: { command: 'codex', args: [], cwd: context.root }, + }], + }, controlPlane: control, adapters: prepared.adapters, + executor: () => ({ status: 0, stdout: '{"type":"thread.started","thread_id":"manager-interrupted"}\n' }), + }), + }, + }); + const run = service.run({ scheduleId: 'manager-check-in', actorId: 'owner', approved: true }); + assert.equal(run.result, 'failed'); + assert.equal(control.identity('session-manager').state, 'working'); + assert.equal(control.identity('session-manager').runtimeSessionId, 'manager-interrupted'); + assert.equal(service.wakeReservations({ managerId: 'session-manager' })[0].outcome, 'failed'); + } finally { control.close(); } +}); + +test('SCN-scheduled-executor-failure: a zero-status executor error records failure and keeps output bounded and redacted', () => { + const context = fixture(); + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + try { + control.reportStatus({ areaId: 'session-manager', state: 'idle', summary: 'Fixture session is available.' }); + const service = new ScheduleService({ + repositoryRoot: context.root, controlPlane: control, + executor: () => ({ + status: 0, + error: Object.assign(new Error('private reasoning=fixture-analysis'), { code: 'ENOBUFS' }), + stdout: 'screenshot-data-'.repeat(200_000), + stderr: 'token=fixture-token password=fixture-password', + }), + }); + const run = service.run({ scheduleId: 'domain-audit', actorId: context.worker }); + assert.equal(run.result, 'failed'); + assert.equal(run.exitStatus, 0); + assert.match(run.stdout, /\[truncated\]/); + assert.doesNotMatch(`${run.stdout}\n${run.stderr}`, /fixture-token|fixture-password|fixture-analysis/); + } finally { control.close(); } +}); diff --git a/test/fleet/lifecycle.test.mjs b/test/fleet/lifecycle.test.mjs index 14effb5..28e5e07 100644 --- a/test/fleet/lifecycle.test.mjs +++ b/test/fleet/lifecycle.test.mjs @@ -15,7 +15,7 @@ import { defaultManagerCheckInSchedule, organizationGraphFromConfig } from '../. import { createWorktrees } from '../../src/kernel/worktrees.mjs'; import { ResourceService } from '../../src/resources/service.mjs'; import { - createFleetBrief, detachFleet, planFleetDetach, planFleetDown, planFleetUp, startFleet, stopFleet, + createFleetBrief, detachFleet, planAreaUp, planFleetDetach, planFleetDown, planFleetUp, startFleet, stopFleet, } from '../../src/runtime/lifecycle.mjs'; function fleetFixture({ workerRuntime = 'claude', hierarchy = false } = {}) { @@ -331,7 +331,7 @@ test('SCN-mixed-runtime: Codex and Claude share stable identities while planning const started = startFleet({ plan, controlPlane: control, adapters, executor: (launch) => launch.areaId === fixture.worker - ? { status: 0, stdout: '{"type":"thread.started","thread_id":"codex-worker-1"}\n' } + ? { status: 0, stdout: '{"type":"thread.started","thread_id":"codex-worker-1"}\n{"type":"turn.completed"}\n' } : { status: 0, stdout: 'claude-manager-1\n' }, }); assert.deepEqual(started.started.map((item) => item.runtimeSessionId), [ @@ -360,3 +360,124 @@ test('SCN-mixed-runtime: Codex and Claude share stable identities while planning assert.equal(control.identity(fixture.worker).state, 'offline'); control.close(); }); + +test('SCN-codex-terminal-completion: a captured Codex identity is not a completed turn without terminal evidence', () => { + const fixture = fleetFixture({ workerRuntime: 'codex' }); + const control = openControlPlane({ repositoryRoot: fixture.root, env: fixture.env }); + const adapters = new Map([ + ['claude', createClaudeAdapter()], + ['codex', createCodexAdapter({ executable: 'codex' })], + ]); + const plan = planFleetUp({ repositoryRoot: fixture.root, controlPlane: control, adapters, fresh: true }); + assert.throws(() => startFleet({ + plan, controlPlane: control, adapters, + executor: (launch) => launch.areaId === fixture.worker + ? { status: 0, stdout: '{"type":"thread.started","thread_id":"codex-interrupted-1"}\n' } + : { status: 0, stdout: 'claude-manager-1\n' }, + }), (error) => error.code === 'FLEET_START_FAILED' + && error.details.executorOutcome.reason === 'codex-terminal-event-missing'); + assert.equal(control.identity(fixture.worker).state, 'working'); + assert.equal(control.identity(fixture.worker).runtimeSessionId, 'codex-interrupted-1'); + const duplicate = planAreaUp({ + repositoryRoot: fixture.root, controlPlane: control, areaId: fixture.worker, adapters, + }); + assert.equal(duplicate.canProceed, false); + assert.deepEqual(duplicate.blockers, [{ + areaId: fixture.worker, code: 'RUNTIME_IDENTITY_ACTIVE', state: 'working', + runtimeSessionId: 'codex-interrupted-1', + }]); + control.close(); +}); + +test('SCN-codex-startup-diagnostics: owner-visible startup failures prefer bounded structured Codex evidence without disclosure', () => { + const modelFixture = fleetFixture({ workerRuntime: 'codex' }); + const modelControl = openControlPlane({ repositoryRoot: modelFixture.root, env: modelFixture.env }); + const modelAdapters = new Map([ + ['claude', createClaudeAdapter()], + ['codex', createCodexAdapter({ executable: 'codex' })], + ]); + const modelPlan = planFleetUp({ + repositoryRoot: modelFixture.root, controlPlane: modelControl, adapters: modelAdapters, fresh: true, + }); + let modelFailure; + assert.throws( + () => startFleet({ + plan: modelPlan, controlPlane: modelControl, adapters: modelAdapters, + executor: (launch) => launch.areaId === modelFixture.worker + ? { + status: 23, + stdout: '{"type":"error","code":"MODEL_NOT_SUPPORTED","message":"Requested model fixture-model is not supported","prompt":"fixture prompt must not leak"}\n', + stderr: 'Vercel MCP authentication warning: token=fixture-token password=fixture-password', + } + : { status: 0, stdout: 'claude-manager-1\n' }, + }), + (error) => { + modelFailure = error; + return error.code === 'FLEET_START_FAILED' && error.details.status === 23; + }, + ); + assert.deepEqual(modelFailure.details.diagnostic, { + schema: 'torch.dev/runtime-startup-diagnostic/v1alpha1', outcome: 'known', + category: 'model-rejection', code: 'MODEL_NOT_SUPPORTED', + message: 'Requested model fixture-model is not supported', + stderr: 'Vercel MCP authentication warning: token=[REDACTED] password=[REDACTED]', + }); + assert.equal(modelFailure.details.stderr, modelFailure.details.diagnostic.stderr); + assert.doesNotMatch(JSON.stringify(modelFailure.details), /fixture prompt|fixture-token|fixture-password/); + assert.equal(modelControl.identity(modelFixture.worker).state, 'offline'); + modelControl.close(); + + const malformedFixture = fleetFixture({ workerRuntime: 'codex' }); + const malformedControl = openControlPlane({ repositoryRoot: malformedFixture.root, env: malformedFixture.env }); + const malformedAdapters = new Map([ + ['claude', createClaudeAdapter()], + ['codex', createCodexAdapter({ executable: 'codex' })], + ]); + const malformedPlan = planFleetUp({ + repositoryRoot: malformedFixture.root, controlPlane: malformedControl, adapters: malformedAdapters, fresh: true, + }); + let malformedFailure; + assert.throws( + () => startFleet({ + plan: malformedPlan, controlPlane: malformedControl, adapters: malformedAdapters, + executor: (launch) => launch.areaId === malformedFixture.worker + ? { status: 24, stdout: '{malformed Codex event', stderr: 'MCP warning' } + : { status: 0, stdout: 'claude-manager-2\n' }, + }), + (error) => { + malformedFailure = error; + return error.code === 'FLEET_START_FAILED' && error.details.status === 24; + }, + ); + assert.equal(malformedFailure.details.diagnostic.outcome, 'unknown'); + assert.equal(malformedFailure.details.diagnostic.reason, 'stdout-malformed-json'); + assert.equal(malformedFailure.details.diagnostic.stderr, 'MCP warning'); + malformedControl.close(); + + const oversizedFixture = fleetFixture({ workerRuntime: 'codex' }); + const oversizedControl = openControlPlane({ repositoryRoot: oversizedFixture.root, env: oversizedFixture.env }); + const oversizedAdapters = new Map([ + ['claude', createClaudeAdapter()], + ['codex', createCodexAdapter({ executable: 'codex' })], + ]); + const oversizedPlan = planFleetUp({ + repositoryRoot: oversizedFixture.root, controlPlane: oversizedControl, adapters: oversizedAdapters, fresh: true, + }); + let oversizedFailure; + assert.throws( + () => startFleet({ + plan: oversizedPlan, controlPlane: oversizedControl, adapters: oversizedAdapters, + executor: (launch) => launch.areaId === oversizedFixture.worker + ? { status: 25, stdout: 'x'.repeat((16 * 1024) + 1), stderr: 'MCP warning' } + : { status: 0, stdout: 'claude-manager-3\n' }, + }), + (error) => { + oversizedFailure = error; + return error.code === 'FLEET_START_FAILED' && error.details.status === 25; + }, + ); + assert.equal(oversizedFailure.details.diagnostic.outcome, 'unknown'); + assert.equal(oversizedFailure.details.diagnostic.reason, 'stdout-exceeds-bound'); + assert.doesNotMatch(JSON.stringify(oversizedFailure.details), /x{512}/); + oversizedControl.close(); +}); diff --git a/test/fleet/provider-updates.test.mjs b/test/fleet/provider-updates.test.mjs index b79b8ba..7f363c0 100644 --- a/test/fleet/provider-updates.test.mjs +++ b/test/fleet/provider-updates.test.mjs @@ -34,7 +34,7 @@ function fixture() { return { status: 0, stdout: 'installed' }; } return { status: 0, stdout: args[0] === '--version' ? `provider ${version}` - : args.includes('exec') ? '{"type":"thread.started","thread_id":"provider-update-fixture-session"}\n' : 'help' }; + : args.includes('exec') ? '{"type":"thread.started","thread_id":"provider-update-fixture-session"}\n{"type":"turn.completed"}\n' : 'help' }; }; return { env, calls, runner, next: v => { version = v; }, breakInstall: () => { failInstall = true; } }; }