TASK-pilot-executor-outcomes: fail closed on executor errors

docs/TEST_INTEGRITY.md

@@ -1,5 +1,35 @@
 # Test Integrity Notes
 
+## 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 — Exact observed-file ownership evidence
 
 ```yaml

src/adapters/codex.mjs

@@ -87,7 +87,7 @@ function redactDiagnosticText(value, limit = MAX_DIAGNOSTIC_TEXT_LENGTH) {
   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)\s*(?:=|:|\s)\s*)([^\s,;]+)/gi, '$1[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);
 }

src/cli.mjs

@@ -42,6 +42,10 @@ 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;
 import { OwnerDigestService } from './observability/owner-digest.mjs';
 import { ContextTelemetryService } from './telemetry/context.mjs';
 import { ScheduleService } from './schedules/service.mjs';
@@ -1108,6 +1112,7 @@ export async function runCli(argv = process.argv.slice(2), {
           plan, controlPlane: control, adapters,
           executor: (launch) => spawn(launch.command, launch.args, {
             cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'],
+            maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES,
           }),
         }), { json });
         return 0;
@@ -1441,6 +1446,7 @@ export async function runCli(argv = process.argv.slice(2), {
             plan, controlPlane: control, adapters,
             executor: (launch) => spawn(launch.command, launch.args, {
               cwd: launch.cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'],
+              maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES,
             }),
           }), { json });
         } else throw new TorchError(`Unknown fleet operation: ${operation}`, { code: 'UNKNOWN_COMMAND' });
@@ -1615,7 +1621,7 @@ export async function runCli(argv = process.argv.slice(2), {
                 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',
+                  timeout: 300_000, maxBuffer: MAX_RUNTIME_EXECUTOR_OUTPUT_BYTES, killSignal: 'SIGTERM',
                 }),
               });
             },

src/runtime/lifecycle.mjs

@@ -465,6 +465,61 @@ 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;
+}
+
+export function classifyExecutorOutcome(result) {
+  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) 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 canSafelyCaptureRuntimeId(result) {
+  const output = result?.stdout ?? result?.output;
+  return Buffer.byteLength(typeof output === 'string' ? output : String(output ?? ''), 'utf8')
+    <= MAX_EXECUTOR_CAPTURE_OUTPUT_BYTES;
+}
+
+function captureRuntimeId({ action, runtimeAdapter, result }) {
+  if (!action.requiresRuntimeIdCapture || !runtimeAdapter || !canSafelyCaptureRuntimeId(result)) {
+    return action.runtimeSessionId;
+  }
+  try {
+    return runtimeAdapter.captureRuntimeId({
+      areaId: action.areaId, runtimeSessionId: action.runtimeSessionId,
+      stdout: result?.stdout, output: result?.output,
+    }).runtimeSessionId;
+  } catch {
+    return action.runtimeSessionId;
+  }
+}
+
 export function startFleet({ plan, controlPlane, executor, adapters } = {}) {
   if (!plan?.canProceed) {
     throw new TorchError('Fleet startup plan has unresolved blockers', {
@@ -482,20 +537,25 @@ export function startFleet({ plan, controlPlane, executor, adapters } = {}) {
     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 runtimeAdapter = runtimes.get(action.runtime);
+    const outcome = classifyExecutorOutcome(result);
+    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;
       controlPlane.reportStatus({
-        areaId: action.areaId, state: 'offline', runtime: action.runtime,
-        runtimeSessionId: action.runtimeSessionId, summary: 'Runtime launch failed.',
+        areaId: action.areaId, state: 'working', runtime: action.runtime,
+        runtimeSessionId,
+        summary: 'Runtime executor completion is unknown; duplicate launch is blocked pending reconciliation.',
       });
       throw new TorchError(`Fleet startup failed for ${action.areaId}`, {
         code: 'FLEET_START_FAILED',
         details: {
-          areaId: action.areaId, status: result.status,
-          stderr: diagnostic ? diagnostic.stderr : result.stderr ?? null,
-          ...(diagnostic ? { diagnostic } : {}),
+          areaId: action.areaId, status: outcome.status, signal: outcome.signal,
+          executorOutcome: outcome,
+          ...(diagnostic
+            ? { diagnostic, stderr: diagnostic.stderr }
+            : { stderr: safeExecutorText(result?.stderr ?? result?.error?.message) }),
           started,
         },
       });

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) {

test/fleet/executor-outcomes.test.mjs

@@ -0,0 +1,117 @@
+import assert from 'node:assert/strict';
+import { execFileSync, spawnSync } from 'node:child_process';
+import { mkdtempSync, 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 { 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';
+
+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', '[email protected]']);
+  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' });
+  return { root, env, worker };
+}
+
+function planFor(context, runtimeSessionId = 'captured-native-session') {
+  return {
+    canProceed: true,
+    projectId: 'executor-outcomes-fixture',
+    actions: [{
+      areaId: context.worker, runtime: 'codex', runtimeSessionId,
+      requiresRuntimeIdCapture: false, 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 }) {
+  const control = openControlPlane({ repositoryRoot: context.root, env: context.env });
+  try {
+    let failure;
+    assert.throws(() => startFleet({
+      plan: planFor(context), 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, 'captured-native-session');
+    return failure;
+  } finally { control.close(); }
+}
+
+test('SCN-executor-interrupted-outcomes: a graceful timeout, signal, or ENOBUFS cannot create an idle turn or disclose output', () => {
+  const timeout = spawnSync(process.execPath, [
+    '-e', 'process.on("SIGTERM", () => process.exit(0)); setInterval(() => {}, 1000);',
+  ], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], timeout: 100, killSignal: 'SIGTERM' });
+  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/);
+});
+
+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(); }
+});