diff --git a/docs/TEST_INTEGRITY.md b/docs/TEST_INTEGRITY.md index 48c6ef6..8687437 100644 --- a/docs/TEST_INTEGRITY.md +++ b/docs/TEST_INTEGRITY.md @@ -1,5 +1,38 @@ # Test Integrity Notes +## 2026-10-03 — Rolling continuation and bounded no-progress escalation + +```yaml +test_integrity_note: + change_type: new_tests + scenarios: + - id: SCN-runtime-continuation-refill + given: "A long manager turn and a short specialist turn occupy two slots; a third identity waits" + when: "Only the short turn completes" + then: "The third starts before manager completion and the daily attempt cap still holds" + - id: SCN-runtime-continuation-no-progress + given: "An unchanged unfinished eligible assignment" + when: "Two successful executor turns make no recorded task progress" + then: "Exactly two specialist turns occur, one durable coordination request is sent, unchanged-work repeats stop, new revisions may resume and unmet dependencies remain ineligible" + - id: SCN-runtime-continuation-capacity + given: "An owner-started executor already occupies the configured automatic capacity" + when: "The continuation controller runs" + then: "No additional automatic turn starts and no attempt is charged" + observable_outcomes: + - "Actual launch ordering, daily charges, coordination messages and explicit held-work reason" + determinism_controls: + - "Hermetic installed Git projects, manually resolved provider promises and task fixtures; no sleep, network or live model" + anti_cheat_rationale: + prevents: + - "Replacing the batch barrier by raising concurrency" + - "Claiming refill only after all turns finish" + - "Spending unlimited turns on unchanged work or ignoring dependency gates" + - "Ignoring physical executors started outside the controller" + relaxation: + did_relax_any_assertion: false + if_true_explain_spec_basis: "" +``` + ## 2026-10-03 — Process-verified runtime presence ```yaml diff --git a/docs/spec_changes/2026-10-03-rolling-continuation.md b/docs/spec_changes/2026-10-03-rolling-continuation.md new file mode 100644 index 0000000..3933ee0 --- /dev/null +++ b/docs/spec_changes/2026-10-03-rolling-continuation.md @@ -0,0 +1,15 @@ +# Rolling continuation without a whole-batch barrier + +Owner-requested fix for `TASK-d95e72d0-816d-4fc2-acaf-b73642fd4881`. + +The serialized controller reserves and charges each launch before dispatch. It waits for the first pending turn to complete, records its outcome, re-reads current policy/backlog/messages and immediately fills any available slot. A long manager turn no longer prevents other eligible identities from running. The controller lock still excludes competing automatic controllers; pending launches and all observed physical turn guards count toward automatic capacity. Owner-requested manual batches are separate authorization and are not silently included in the daily automatic attempt ledger. + +The same approved roster, per-identity mutex, pause, detached state and daily cap remain authoritative. A pause prevents additional launches and never kills an already-started turn. Unknown or failed executor outcomes are not silently retried. Retained crashed guards/controller locks remain fail-closed and require supported recovery. + +Successful turns may continue unchanged assigned/in-progress work if recorded dependencies are complete. After two consecutive successful turns leave the same task IDs/revisions unchanged, further unchanged-work retries are held and one durable request asks the Manager to resolve the blocker. New task revisions or fresh messages can justify another bounded turn. This is a conservative no-progress signal, not proof that no files changed or that a task is done; the Manager must inspect evidence. Waiting, blocked, unassigned, completed or unmet-dependency work is not automatically converted into implementation work. Existing one-assignment rules and self-claim policy are not changed. + +Resume instructions now request eligible implementation/verification/coordination rather than just a status report. They do not grant additional ownership or approval authority. All output completion remains exact-evidence gated. + +Status exposes occupied guard identities and no-progress holds. Installed release and reboot persistence remain unqualified. Existing native pilot service executes this source candidate on its next activation; the daily limit remains 12 automatic attempts, not a token-cost governor. + +Deterministic acceptance scenarios and anti-cheat notes are in docs/TEST_INTEGRITY.md. Independent live qualification remains necessary; do not close broader event-driven or persistent-operations umbrellas based on this source fix alone. diff --git a/src/runtime/continuation.mjs b/src/runtime/continuation.mjs index b0fed80..8365e4c 100644 --- a/src/runtime/continuation.mjs +++ b/src/runtime/continuation.mjs @@ -1,17 +1,23 @@ import { createHash, randomUUID } from 'node:crypto'; -import { existsSync, mkdirSync, readFileSync, renameSync, rmdirSync, writeFileSync } from 'node:fs'; +import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmdirSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { TorchError } from '../kernel/errors.mjs'; function statePath(control) { return join(control.stateRoot, 'continuation.json'); } +function occupiedAreas(control) { + const path = join(control.stateRoot, 'sessions', 'turn-guards'); + return existsSync(path) ? readdirSync(path, { withFileTypes: true }).filter(d => d.isDirectory()).map(d => d.name) : []; +} +function taskKey(tasks) { return JSON.stringify(tasks.map(t => [t.id, t.revision]).sort((a, b) => a[0].localeCompare(b[0]))); } function read(control) { const policyPath = join(control.stateRoot, 'continuation-policy.json'); - const state = { enabled: false, maxTurnsPerDay: 12, areas: [], day: null, attempts: 0, handled: {}, + const state = { enabled: false, maxTurnsPerDay: 12, areas: [], day: null, attempts: 0, handled: {}, progress: {}, ...(existsSync(statePath(control)) ? JSON.parse(readFileSync(statePath(control), 'utf8')) : {}), ...(existsSync(policyPath) ? JSON.parse(readFileSync(policyPath, 'utf8')) : {}) }; if (typeof state.enabled !== 'boolean' || !Number.isInteger(state.maxTurnsPerDay) || state.maxTurnsPerDay < 1 || state.maxTurnsPerDay > 48 || !Array.isArray(state.areas) - || !Number.isInteger(state.attempts) || state.attempts < 0 || !state.handled || typeof state.handled !== 'object') { + || !Number.isInteger(state.attempts) || state.attempts < 0 || !state.handled || typeof state.handled !== 'object' + || !state.progress || typeof state.progress !== 'object') { throw new TorchError('Continuation policy or attempt ledger is invalid; launch refused.', { code: 'CONTINUATION_POLICY_INVALID' }); } return state; @@ -20,7 +26,7 @@ function save(control, state, policy = false) { const path = policy ? join(control.stateRoot, 'continuation-policy.json') : statePath(control); const temporary = `${path}.${randomUUID()}.tmp`; const value = policy ? { enabled: state.enabled, maxTurnsPerDay: state.maxTurnsPerDay, areas: state.areas } - : { day: state.day, attempts: state.attempts, handled: state.handled }; + : { day: state.day, attempts: state.attempts, handled: state.handled, progress: state.progress }; writeFileSync(temporary, JSON.stringify(value), { flag: 'wx', mode: 0o600 }); renameSync(temporary, path); } @@ -48,22 +54,32 @@ export function planContinuation(control, backlog, { at = new Date(), limit = 3 } const tasks = backlog.list(); const candidates = []; + const held = []; for (const areaId of state.areas) { const agent = control.getAgent(areaId); if (!['idle', 'offline', 'waiting'].includes(agent.state) || 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)); + const active = tasks.filter(t => t.owner === areaId && ['assigned', 'in_progress'].includes(t.state) + && (t.dependencies ?? []).every(id => tasks.find(t => t.id === id)?.state === 'completed')); 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 progress = state.progress[areaId]; + const continueWork = active.length > 0 && progress?.status === 0 + && progress.taskKey === taskKey(active) && progress.noProgress < 2; + if (!freshMessages.length && !freshTasks.length && !continueWork) { + if (active.length && progress?.noProgress >= 2) held.push({ areaId, reason: 'no-progress-needs-coordination', taskIds: active.map(t => t.id) }); + 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.push({ areaId, fingerprint, signal, workKey: taskKey(active), + messageCount: freshMessages.length, taskIds: active.map(t => t.id), continueWork }); } candidates.sort((a, b) => Number(b.areaId === 'session-manager') - Number(a.areaId === 'session-manager')); return { enabled: true, day, attempts, remainingTurns: state.maxTurnsPerDay - attempts, + occupiedAreas: occupiedAreas(control), held, candidates: candidates.slice(0, Math.min(limit, state.maxTurnsPerDay - attempts)), mutationPerformed: false }; } @@ -76,21 +92,47 @@ export async function runContinuation(control, backlog, { actorId, launch, at, c throw error; } try { - const plan = planContinuation(control, backlog, { at: clock(), limit }); - 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 }); - if (!current.candidates.some(c => c.areaId === candidate.areaId && c.fingerprint === candidate.fingerprint)) continue; + const pending = new Map(); + const launched = []; + let reason = null; + while (true) { + // Refill immediately after EACH completion. Pending promises and physical + // guards both count, including manual executors outside this controller. + let current = planContinuation(control, backlog, { at: clock(), limit: 3 }); + let occupied = new Set([...occupiedAreas(control), ...pending.keys()]); + while (occupied.size < limit && current.enabled && !current.reason) { + const candidate = current.candidates.find(c => !occupied.has(c.areaId)); + if (!candidate) break; + const state = read(control); + if (state.day !== current.day) { state.day = current.day; state.attempts = 0; } + state.attempts++; + state.handled[candidate.areaId] = candidate.signal; + save(control, state); // Reserve before launch; unknown attempts are never refunded. + let result; + try { result = launch(candidate.areaId); } catch { result = null; } + pending.set(candidate.areaId, Promise.resolve(result).catch(() => null) + .then(status => ({ areaId: candidate.areaId, status, candidate }))); + current = planContinuation(control, backlog, { at: clock(), limit: 3 }); + occupied = new Set([...occupiedAreas(control), ...pending.keys()]); + } + reason = current.reason ?? (occupied.size >= limit ? 'capacity-full' : null); + if (!pending.size) break; + const completed = await Promise.race(pending.values()); + pending.delete(completed.areaId); + launched.push({ areaId: completed.areaId, status: completed.status }); const state = read(control); - if (state.day !== current.day) { state.day = current.day; state.attempts = 0; } - state.attempts++; - state.handled[candidate.areaId] = candidate.signal; - save(control, state); // Charge/reserve before launch; interrupted attempts are never refunded/retried silently. - let result; - try { result = launch(candidate.areaId); } catch { result = null; } - pending.push(Promise.resolve(result).catch(() => null).then(status => ({ areaId: candidate.areaId, status }))); + const active = backlog.list().filter(t => t.owner === completed.areaId && ['assigned', 'in_progress'].includes(t.state)); + const key = taskKey(active); + const unchanged = active.length > 0 && key === completed.candidate.workKey; + const prior = state.progress[completed.areaId]; + const noProgress = unchanged ? (prior?.taskKey === key ? prior.noProgress : 0) + 1 : 0; + state.progress[completed.areaId] = { status: completed.status, taskKey: key, noProgress, completedAt: clock().toISOString() }; + save(control, state); + if (completed.status === 0 && noProgress === 2 && typeof control.sendOwnerRequest === 'function') { + control.sendOwnerRequest({ actorId, recipient: 'session-manager', + body: `TORCH continuation requires coordination: ${completed.areaId} completed two turns without advancing assigned task state (${active.map(t => t.id).join(', ')}). Automatic unchanged-work retries are held. Inspect the blocker and route a concrete decision; do not mark work done without evidence.` }); + } } - return { launched: await Promise.all(pending), reason: plan.reason ?? null }; + return { launched, reason }; } finally { rmdirSync(lock); } } diff --git a/src/runtime/lifecycle.mjs b/src/runtime/lifecycle.mjs index 0683d2c..cec10a7 100644 --- a/src/runtime/lifecycle.mjs +++ b/src/runtime/lifecycle.mjs @@ -160,11 +160,11 @@ function startupMessage(areaId, fresh, instructionDigest) { if (areaId === 'session-manager') { return fresh ? `Start the TORCH Fleet. Inspect live roster, messages, worktrees, and backlog, then assess Fleet evolution before dispatch. ${currentBrief}` - : `Resume the TORCH Fleet from durable state. Reconcile live status and assess Fleet evolution before dispatch. ${currentBrief}`; + : `Resume the TORCH Fleet from durable state. Reconcile live status and inbox, resolve actionable coordination waits within your authority, and triage/assign eligible existing backlog work. Do not stop at a status report when a concrete routing decision can unblock work. Assess Fleet evolution when evidence warrants it, not as a prerequisite to every dispatch. Preserve ownership, approvals and dependency gates. ${currentBrief}`; } return fresh ? `Start this TORCH domain. Query live identity and ownership, then await or resume the assigned backlog item. ${currentBrief}` - : `Resume this TORCH domain from durable state. Read the inbox, confirm ownership, and report current evidence. ${currentBrief}`; + : `Resume this TORCH domain from durable state. Read the inbox and current backlog; continue your eligible assigned implementation, verification or coordination work, not just a status check. Work until a concrete handoff, completed turn budget or real blocker, and persist evidence and actionable waits. Never bypass dependencies, ownership or approval gates. If no eligible work exists, idle is valid. ${currentBrief}`; } function atomicJson(path, value) { diff --git a/test/fleet/legacy-runtime-recovery.test.mjs b/test/fleet/legacy-runtime-recovery.test.mjs index a5718f1..16c143c 100644 --- a/test/fleet/legacy-runtime-recovery.test.mjs +++ b/test/fleet/legacy-runtime-recovery.test.mjs @@ -14,7 +14,7 @@ import { observeRuntimeTurn, withRuntimeTurnGuard } from '../../src/runtime/turn import { observeProject } from '../../src/observability/snapshot.mjs'; import { configureContinuation, planContinuation, runContinuation } from '../../src/runtime/continuation.mjs'; -function fixture({ recoveryWorktree = true } = {}) { +function fixture({ recoveryWorktree = true, extraWorker = false } = {}) { const root = mkdtempSync(join(tmpdir(), 'torch-legacy-recovery-')); for (const args of [['init', '-b', 'main'], ['config', 'user.email', 'test@example.invalid'], ['config', 'user.name', 'Recovery Fixture']]) execFileSync('git', args, { cwd: root, stdio: 'pipe' }); @@ -23,6 +23,7 @@ function fixture({ recoveryWorktree = true } = {}) { execFileSync('git', ['commit', '-m', 'fixture'], { cwd: root, stdio: 'pipe' }); const repository = inspectRepository(root); const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) }); + if (extraWorker) proposal.domains.push({ ...proposal.domains[0], id: 'extra-worker', title: 'Extra worker', owned_paths: ['extra/**'] }); proposal.review = { status: 'approved', reviewedAt: '2026-10-03T00:00:00Z', reviewedBy: 'owner', notes: [] }; const env = { ...process.env, XDG_DATA_HOME: join(root, '.data') }; installProject({ repository, proposal, env, projectId: 'recovery-fixture' }); @@ -253,6 +254,70 @@ test('SCN-runtime-continuation-pause: a pause during one turn prevents another q } finally { f.control.close(); } }); +test('SCN-runtime-continuation-refill: a short turn frees capacity while the manager remains running', async () => { + const f = fixture({ extraWorker: true }); + try { + f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' }); + for (const recipient of ['session-manager', f.areaId, 'extra-worker']) { + f.control.sendOwnerRequest({ actorId: 'owner', recipient, body: 'Eligible handoff' }); + } + configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 3 }); + const starts = []; + const finishes = new Map(); + let thirdStarted; + const third = new Promise(resolve => { thirdStarted = resolve; }); + const run = runContinuation(f.control, { list: () => [] }, { actorId: 'owner', limit: 2, + launch: area => { starts.push(area); if (starts.length === 3) thirdStarted(); + return new Promise(resolve => finishes.set(area, resolve)); } }); + assert.deepEqual(starts, ['session-manager', f.areaId]); + finishes.get(f.areaId)(0); + await third; + assert.deepEqual(starts, ['session-manager', f.areaId, 'extra-worker']); + assert.equal(finishes.has('session-manager'), true, 'Manager completion was not required for refill'); + finishes.get('extra-worker')(0); + finishes.get('session-manager')(0); + assert.equal((await run).launched.length, 3); + assert.equal(planContinuation(f.control, { list: () => [] }).reason, 'daily-turn-cap'); + } finally { f.control.close(); } +}); + +test('SCN-runtime-continuation-no-progress: unfinished assignment continues once then escalates rather than busy-looping', async () => { + const f = fixture(); + try { + f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' }); + configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 8 }); + const task = { id: 'TASK-continue', owner: f.areaId, state: 'assigned', revision: 1, dependencies: [] }; + let specialistTurns = 0; + const result = await runContinuation(f.control, { list: () => [task] }, { actorId: 'owner', limit: 1, + launch: area => { if (area === f.areaId) specialistTurns++; return 0; } }); + assert.equal(specialistTurns, 2); + assert.equal(result.launched.filter(t => t.areaId === f.areaId).length, 2); + const messages = f.control.readMessages({ recipient: 'session-manager', unacknowledgedOnly: true, limit: 1000 }); + assert.equal(messages.filter(m => m.body.includes('two turns without advancing')).length, 1); + assert.equal(planContinuation(f.control, { list: () => [task] }).candidates.some(c => c.areaId === f.areaId), false); + task.revision++; + assert.equal(planContinuation(f.control, { list: () => [task] }).candidates.some(c => c.areaId === f.areaId), true); + task.dependencies = ['TASK-unfinished']; + assert.equal(planContinuation(f.control, { list: () => [task] }).candidates.some(c => c.areaId === f.areaId), false); + } finally { f.control.close(); } +}); + +test('SCN-runtime-continuation-capacity: a manually held executor counts against automatic fleet capacity', async () => { + const f = fixture(); + try { + f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' }); + f.control.sendOwnerRequest({ actorId: 'owner', recipient: f.areaId, body: 'Eligible handoff' }); + configureContinuation(f.control, { actorId: 'owner', enabled: true }); + let running; + withRuntimeTurnGuard(f.control, 'session-manager', () => { + running = runContinuation(f.control, { list: () => [] }, { actorId: 'owner', limit: 1, + launch: () => { throw new Error('Capacity must not be exceeded'); } }); + }); + assert.deepEqual((await running).launched, []); + assert.equal(planContinuation(f.control, { list: () => [] }).attempts, 0); + } finally { f.control.close(); } +}); + test('SCN-legacy-runtime-recovery-project: evidence cannot be applied to a different checkout with cloned project/session fields', () => { const first = fixture(); const second = fixture();