From f823714b01220c875788e1c86cc160524d353ec9 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN Date: Fri, 2 Oct 2026 23:59:50 -0400 Subject: [PATCH] runtime: bound parallel continuation and ignore acknowledgement-only changes --- docs/TEST_INTEGRITY.md | 4 ++++ src/cli.mjs | 15 ++++++++------ src/runtime/continuation.mjs | 22 ++++++++++++-------- test/fleet/legacy-runtime-recovery.test.mjs | 23 +++++++++++++++++++++ test/fleet/mcp-adapter.test.mjs | 2 +- 5 files changed, 50 insertions(+), 16 deletions(-) diff --git a/docs/TEST_INTEGRITY.md b/docs/TEST_INTEGRITY.md index 822ecbe..aacb180 100644 --- a/docs/TEST_INTEGRITY.md +++ b/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" diff --git a/src/cli.mjs b/src/cli.mjs index 0b7a0f7..6a169fc 100644 --- a/src/cli.mjs +++ b/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(); } } diff --git a/src/runtime/continuation.mjs b/src/runtime/continuation.mjs index 85dcc33..b0fed80 100644 --- a/src/runtime/continuation.mjs +++ b/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); } } diff --git a/test/fleet/legacy-runtime-recovery.test.mjs b/test/fleet/legacy-runtime-recovery.test.mjs index 7f724b3..cf9cf7e 100644 --- a/test/fleet/legacy-runtime-recovery.test.mjs +++ b/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 { diff --git a/test/fleet/mcp-adapter.test.mjs b/test/fleet/mcp-adapter.test.mjs index 089f674..e196697 100644 --- a/test/fleet/mcp-adapter.test.mjs +++ b/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);