TASK-bounded-routine-coordination: merge canonical 286bfa72

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'); }

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',
                 }),
               });
             },

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();
+});

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

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