From 12e1e33cc80d8babda008e2d65bdcb1f64675e3f Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN Date: Wed, 30 Sep 2026 20:17:06 -0400 Subject: [PATCH] TASK-pilot-executor-outcomes: require terminal executor evidence --- docs/TEST_INTEGRITY.md | 75 +++++++++++++++ src/cli.mjs | 17 +++- src/runtime/codex-protocol-reducer.mjs | 83 +++++++++++++++++ src/runtime/lifecycle.mjs | 46 +++++++++- test/fleet/executor-outcomes.test.mjs | 122 ++++++++++++++++++++++++- test/fleet/lifecycle.test.mjs | 32 ++++++- test/fleet/provider-updates.test.mjs | 2 +- 7 files changed, 363 insertions(+), 14 deletions(-) create mode 100644 src/runtime/codex-protocol-reducer.mjs diff --git a/docs/TEST_INTEGRITY.md b/docs/TEST_INTEGRITY.md index c867fa8..203e6ac 100644 --- a/docs/TEST_INTEGRITY.md +++ b/docs/TEST_INTEGRITY.md @@ -1,5 +1,80 @@ # Test Integrity Notes +## 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 diff --git a/src/cli.mjs b/src/cli.mjs index 6ad9f13..27f84c4 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'; @@ -46,6 +47,16 @@ 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. + if (launch.runtime === 'codex' && spawn === spawnSync) { + 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'; @@ -1110,7 +1121,7 @@ export async function runCli(argv = process.argv.slice(2), { } print(startFleet({ plan, controlPlane: control, adapters, - executor: (launch) => spawn(launch.command, launch.args, { + executor: (launch) => executeRuntimeLaunch(spawn, launch, { cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES, }), @@ -1444,7 +1455,7 @@ export async function runCli(argv = process.argv.slice(2), { } print(startFleet({ plan, controlPlane: control, adapters, - executor: (launch) => spawn(launch.command, launch.args, { + executor: (launch) => executeRuntimeLaunch(spawn, launch, { cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES, }), @@ -1619,7 +1630,7 @@ export async function runCli(argv = process.argv.slice(2), { } return startFleet({ plan, controlPlane: control, adapters, - executor: (launch) => spawn(launch.command, launch.args, { + executor: (launch) => executeRuntimeLaunch(spawn, launch, { cwd: launch.cwd, 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..3370a2b --- /dev/null +++ b/src/runtime/codex-protocol-reducer.mjs @@ -0,0 +1,83 @@ +#!/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; + +function safeCode(value) { + return typeof value === 'string' && /^[A-Z][A-Z0-9_.-]{0,79}$/.test(value) ? value : null; +} + +function emit(record) { + process.stdout.write(`${JSON.stringify(record)}\n`); +} + +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; + } + 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); + terminal = { type: 'turn.failed', ...(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, () => { 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 (threadStarted) emit({ type: 'thread.started', thread_id: threadStarted }); + if (childSpawnError) emit({ type: 'error', code: childSpawnError }); + else if (terminal) emit(terminal); + if (signal) exitWithSignal(signal); + else process.exit(Number.isInteger(code) && code >= 0 ? code : 1); +}); diff --git a/src/runtime/lifecycle.mjs b/src/runtime/lifecycle.mjs index 37e7713..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), @@ -483,7 +499,20 @@ function safeExecutorCode(value) { return typeof value === 'string' && /^[A-Z][A-Z0-9_.-]{0,79}$/.test(value) ? value : null; } -export function classifyExecutorOutcome(result) { +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); @@ -495,6 +524,12 @@ export function classifyExecutorOutcome(result) { }; } 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' }; @@ -536,8 +571,11 @@ 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 }); - const outcome = classifyExecutorOutcome(result); + 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 }); @@ -577,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/test/fleet/executor-outcomes.test.mjs b/test/fleet/executor-outcomes.test.mjs index ab7f7c1..ff71df8 100644 --- a/test/fleet/executor-outcomes.test.mjs +++ b/test/fleet/executor-outcomes.test.mjs @@ -1,10 +1,11 @@ import assert from 'node:assert/strict'; import { execFileSync, spawnSync } from 'node:child_process'; -import { mkdtempSync, writeFileSync } from 'node:fs'; +import { 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 } 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'; @@ -13,6 +14,8 @@ 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]); @@ -72,10 +75,16 @@ function expectInterruptedStart({ context, result, expectedReason, plan = planFo } finally { control.close(); } } -test('SCN-executor-interrupted-outcomes: a graceful timeout, signal, or ENOBUFS cannot create an idle turn or disclose output', () => { +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', 'process.on("SIGTERM", () => process.exit(0)); setInterval(() => {}, 1000);', - ], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], timeout: 1_000, killSignal: 'SIGTERM' }); + '-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' }); @@ -108,6 +117,111 @@ test('SCN-executor-interrupted-outcomes: a graceful timeout, signal, or 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-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-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 }); diff --git a/test/fleet/lifecycle.test.mjs b/test/fleet/lifecycle.test.mjs index b90260d..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), [ @@ -361,6 +361,34 @@ test('SCN-mixed-runtime: Codex and Claude share stable identities while planning 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 }); 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; } }; }