runtime: bound parallel continuation and ignore acknowledgement-only changes

docs/TEST_INTEGRITY.md

@@ -2846,6 +2846,10 @@ test_integrity_note:
       given: "Owner-enabled roster and a new durable signal"
       when: "Turns repeat, hit a cap or are paused"
       then: "Only new eligible signals launch within the owner daily cap"
+    - id: SCN-runtime-continuation-signals
+      given: "Two eligible identities with multiple unread messages"
+      when: "A bounded batch starts and only acknowledgements change afterward"
+      then: "Both starts occur concurrently, a competing controller refuses, and acknowledgements alone cause no wake"
     - id: SCN-runtime-continuation-pause
       given: "Two eligible turns"
       when: "The owner pauses during the first"

src/cli.mjs

@@ -14,7 +14,7 @@ import { startConsole } from './console/server.mjs';
 import { classifyRecoverability, createLocalCanonical, planLocalCanonical } from './canonical/local.mjs';
 import { openControlPlane } from './control-plane/service.mjs';
 import { fetchCanonicalObjects, planCanonicalFetch } from './canonical/fetch.mjs';
-import { spawnSync } from 'node:child_process';
+import { spawn as spawnProcess, spawnSync } from 'node:child_process';
 import { randomUUID } from 'node:crypto';
 import {
   mkdirSync, mkdtempSync, renameSync, rmSync, statSync, writeFileSync,
@@ -594,12 +594,15 @@ export async function runCli(argv = process.argv.slice(2), {
         }
         if (operation !== 'tick') throw new TorchError('Unknown continuation operation.', { code: 'UNKNOWN_COMMAND' });
         print(await runContinuation(control, backlog, { actorId: 'owner', limit: Number(optionValue(argv, '--limit') ?? 3),
-          launch: areaId => {
-            const result = spawn(process.execPath, [fileURLToPath(new URL('../bin/torch.mjs', import.meta.url)),
+          launch: areaId => new Promise(resolveTurn => {
+            const child = spawnProcess(process.execPath, [fileURLToPath(new URL('../bin/torch.mjs', import.meta.url)),
               'up', '--only', areaId, '--yes', '--json'], { cwd: control.repositoryRoot, env,
-              encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], maxBuffer: 256 * 1024 });
-            return result.error || result.signal ? null : result.status;
-          } }), { json });
+              stdio: ['ignore', 'pipe', 'pipe'] });
+            child.stdout.resume(); child.stderr.resume();
+            let failed = false;
+            child.on('error', () => { failed = true; });
+            child.on('close', (code, signal) => resolveTurn(failed || signal ? null : code));
+          }) }), { json });
         return 0;
       } finally { control.close(); }
     }

src/runtime/continuation.mjs

@@ -54,10 +54,13 @@ export function planContinuation(control, backlog, { at = new Date(), limit = 3
       || existsSync(join(control.stateRoot, 'sessions', 'turn-guards', areaId))) continue;
     const messages = control.readMessages({ recipient: areaId, unacknowledgedOnly: true, limit: 1000 });
     const active = tasks.filter(t => t.owner === areaId && ['assigned', 'in_progress'].includes(t.state));
-    if (!messages.length && !active.length) continue;
-    const fingerprint = createHash('sha256').update(JSON.stringify({ messages: messages.map(m => m.id), tasks: active.map(t => [t.id, t.revision]) })).digest('hex');
-    if (state.handled[areaId] === fingerprint) continue;
-    candidates.push({ areaId, fingerprint, messageCount: messages.length, taskIds: active.map(t => t.id) });
+    const previous = state.handled[areaId] ?? {};
+    const freshMessages = messages.filter(m => !previous.messageIds?.includes(m.id));
+    const freshTasks = active.filter(t => previous.taskRevisions?.[t.id] !== t.revision);
+    if (!freshMessages.length && !freshTasks.length) continue;
+    const signal = { messageIds: messages.map(m => m.id), taskRevisions: Object.fromEntries(active.map(t => [t.id, t.revision])) };
+    const fingerprint = createHash('sha256').update(JSON.stringify(signal)).digest('hex');
+    candidates.push({ areaId, fingerprint, signal, messageCount: freshMessages.length, taskIds: freshTasks.map(t => t.id) });
   }
   candidates.sort((a, b) => Number(b.areaId === 'session-manager') - Number(a.areaId === 'session-manager'));
   return { enabled: true, day, attempts, remainingTurns: state.maxTurnsPerDay - attempts,
@@ -74,7 +77,7 @@ export async function runContinuation(control, backlog, { actorId, launch, at, c
   }
   try {
     const plan = planContinuation(control, backlog, { at: clock(), limit });
-    const launched = [];
+    const pending = [];
     for (const candidate of plan.candidates) {
       // Fresh plan and policy check before each launch; a concurrent pause wins.
       const current = planContinuation(control, backlog, { at: clock(), limit: 3 });
@@ -82,11 +85,12 @@ export async function runContinuation(control, backlog, { actorId, launch, at, c
       const state = read(control);
       if (state.day !== current.day) { state.day = current.day; state.attempts = 0; }
       state.attempts++;
-      state.handled[candidate.areaId] = candidate.fingerprint;
+      state.handled[candidate.areaId] = candidate.signal;
       save(control, state); // Charge/reserve before launch; interrupted attempts are never refunded/retried silently.
-      const status = await launch(candidate.areaId);
-      launched.push({ areaId: candidate.areaId, status });
+      let result;
+      try { result = launch(candidate.areaId); } catch { result = null; }
+      pending.push(Promise.resolve(result).catch(() => null).then(status => ({ areaId: candidate.areaId, status })));
     }
-    return { launched, reason: plan.reason ?? null };
+    return { launched: await Promise.all(pending), reason: plan.reason ?? null };
   } finally { rmdirSync(lock); }
 }

test/fleet/legacy-runtime-recovery.test.mjs

@@ -147,6 +147,29 @@ test('SCN-runtime-continuation: explicit enable, event deduplication, daily cap
   } finally { f.control.close(); }
 });
 
+test('SCN-runtime-continuation-signals: acknowledgements do not wake; eligible turns start concurrently within the limit', async () => {
+  const f = fixture();
+  try {
+    f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' });
+    const messages = [];
+    for (const recipient of [f.areaId, 'session-manager']) {
+      messages.push(f.control.sendOwnerRequest({ actorId: 'owner', recipient, body: 'Review current work' }));
+      f.control.sendOwnerRequest({ actorId: 'owner', recipient, body: 'Second handoff' });
+    }
+    configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 4 });
+    const completions = [];
+    const running = runContinuation(f.control, { list: () => [] }, { actorId: 'owner', limit: 2,
+      launch: () => new Promise(resolve => completions.push(resolve)) });
+    assert.equal(completions.length, 2, 'Both starts occur without waiting for the first provider to finish');
+    assert.equal((await runContinuation(f.control, { list: () => [] }, { actorId: 'owner', launch: () => 0 })).reason,
+      'controller-active-or-unreconciled');
+    completions.forEach(resolve => resolve(0));
+    assert.equal((await running).launched.length, 2);
+    for (const message of messages) f.control.ackMessage({ recipient: message.recipient, messageId: message.id });
+    assert.equal(planContinuation(f.control, { list: () => [] }).candidates.length, 0);
+  } finally { f.control.close(); }
+});
+
 test('SCN-runtime-continuation-pause: a pause during one turn prevents another queued turn', async () => {
   const f = fixture();
   try {

test/fleet/mcp-adapter.test.mjs

@@ -259,7 +259,7 @@ test('SCN-codex-mcp-runtime-dir: only a valid Linux parent runtime directory is
     areaId: context.worker, runtimeSessionId: 'thread-runtime-dir', worktree: '/tmp/codex-runtime-dir',
     message: 'Resume.', mcp,
   });
-  const expected = `mcp_servers.${mcp.name}.env_vars=[\"XDG_RUNTIME_DIR\"]`;
+  const expected = `mcp_servers.${mcp.name}.env_vars=["XDG_RUNTIME_DIR"]`;
   for (const launch of [create, resume]) {
     assert.equal(launch.launch.args.includes(expected), true);
     assert.equal(launch.launch.args.some((arg) => /env_vars=/.test(arg) && arg !== expected), false);