Make automatic fleet concurrency owner-configurable up to seven

docs/TEST_INTEGRITY.md

@@ -1,5 +1,28 @@
 # Test Integrity Notes
 
+## 2026-10-03 — Owner-configured concurrency
+
+```yaml
+test_integrity_note:
+  change_type: new_tests
+  scenarios:
+    - id: SCN-runtime-continuation-concurrency
+      given: "Seven eligible identities and an approved 48-turn daily cap"
+      when: "The owner raises concurrency to seven, starts a bounded controller, then pauses"
+      then: "Seven pending starts occur before completion; a tighter limit restricts candidates; invalid settings refuse; pause preserves cap, concurrency and charged attempts"
+  observable_outcomes:
+    - "Pending launch count, policy capacity, ledger attempts and refusal codes"
+  determinism_controls:
+    - "Hermetic Git repository, fixed clock, controlled promises and no provider calls"
+  anti_cheat_rationale:
+    prevents:
+      - "Treating sequential starts as concurrency or resetting usage on configuration"
+      - "Tick limits exceeding owner-approved policy"
+  relaxation:
+    did_relax_any_assertion: false
+    if_true_explain_spec_basis: ""
+```
+
 ## 2026-10-03 — Visible automatic-work stops
 
 ```yaml

docs/spec_changes/2026-10-03-continuation-concurrency.md

@@ -0,0 +1,16 @@
+# Owner-configured continuation concurrency
+
+The owner can configure `torch continuation enable --concurrency 7 --yes`.
+The default remains three; valid settings are integers from one through seven.
+An explicit tick `--limit` may tighten but never exceed the approved policy.
+Each refill reads current policy; lowering capacity does not cancel active turns.
+Omitted configuration fields preserve the approved daily cap and concurrency.
+Configuration never resets the attempt ledger or bypasses physical identity guards.
+
+An already-running older controller cannot hot-load this change. Replace its
+timer, not its active service, and allow its native turns and controller lock to
+finish before the new controller takes over. No concurrent controllers may
+dispatch through the same project lock.
+
+Scenario SCN-runtime-continuation-concurrency proves seven simultaneous pending
+starts, tighter tick limits, invalid-policy refusal and preserved usage on pause.

site/console.js

@@ -1041,7 +1041,7 @@ function renderAttention(groups) {
 function render(snapshot) {
   const continuation = snapshot.continuation;
   setText('#continuation-status', continuation?.available
-    ? `Automatic turns: ${continuation.attempts}/${continuation.maxTurnsPerDay} used today (UTC); ${continuation.remainingTurns} left. ${continuation.stopReason === 'daily-turn-cap'
+    ? `Automatic turns: ${continuation.attempts}/${continuation.maxTurnsPerDay} used today (UTC); ${continuation.remainingTurns} left. Concurrency policy: up to ${continuation.maxConcurrency ?? 3}. ${continuation.stopReason === 'daily-turn-cap'
       ? 'No further automatic starts: the owner-controlled daily cap is reached. Running turns may finish.'
       : continuation.stopReason === 'paused' ? 'Owner policy is paused; no new automatic starts.' : 'Policy permits new eligible turns; this does not verify timer installation.'} ${continuation.held?.length ?? 0} unchanged-work hold(s) need Manager coordination. This is not measured token or cost usage.`
     : 'Automatic continuation policy is unavailable; do not assume the fleet can start more work.');

src/cli.mjs

@@ -98,7 +98,7 @@ Usage:
   torch setup [--parent <path>] [--dry-run] [--yes] [--json]
   torch up [--fresh] [--only <id,id>] [--dry-run] [--yes] [--json]
   torch recover-runtime --area <id> --unit <service> --invocation <id> [--expected-session <id> --expected-updated-at <time> --yes] [--json]
-  torch continuation <status|enable|pause|tick> [--max-turns-per-day <1-48>] [--limit <1-3>] [--yes] [--json]
+  torch continuation <status|enable|pause|tick> [--max-turns-per-day <1-48>] [--concurrency <1-7>] [--limit <1-7>] [--yes] [--json]
   torch down [--dry-run] [--yes] [--json]
   torch capture [--json]
   torch detach [--dry-run] [--yes] [--json]
@@ -590,10 +590,11 @@ export async function runCli(argv = process.argv.slice(2), {
         }
         if (operation === 'enable' || operation === 'pause') {
           print(configureContinuation(control, { actorId: 'owner', enabled: operation === 'enable',
-            maxTurnsPerDay: Number(optionValue(argv, '--max-turns-per-day') ?? 12) }), { json }); return 0;
+            maxTurnsPerDay: optionValue(argv, '--max-turns-per-day') === undefined ? undefined : Number(optionValue(argv, '--max-turns-per-day')),
+            maxConcurrency: optionValue(argv, '--concurrency') === undefined ? undefined : Number(optionValue(argv, '--concurrency')) }), { json }); return 0;
         }
         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),
+        print(await runContinuation(control, backlog, { actorId: 'owner', limit: optionValue(argv, '--limit') === undefined ? undefined : Number(optionValue(argv, '--limit')),
           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,

src/runtime/continuation.mjs

@@ -11,11 +11,12 @@ function occupiedAreas(control) {
 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: {}, progress: {},
+  const state = { enabled: false, maxTurnsPerDay: 12, maxConcurrency: 3, 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.maxConcurrency) || state.maxConcurrency < 1 || state.maxConcurrency > 7
     || !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' });
@@ -25,22 +26,26 @@ function read(control) {
 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 }
+  const value = policy ? { enabled: state.enabled, maxTurnsPerDay: state.maxTurnsPerDay, maxConcurrency: state.maxConcurrency, areas: state.areas }
     : { day: state.day, attempts: state.attempts, handled: state.handled, progress: state.progress };
   writeFileSync(temporary, JSON.stringify(value), { flag: 'wx', mode: 0o600 });
   renameSync(temporary, path);
 }
 
-export function configureContinuation(control, { actorId, enabled, maxTurnsPerDay = 12 } = {}) {
+export function configureContinuation(control, { actorId, enabled, maxTurnsPerDay, maxConcurrency } = {}) {
   control.assertOwnerActor(actorId);
-  if (typeof enabled !== 'boolean' || !Number.isInteger(maxTurnsPerDay) || maxTurnsPerDay < 1 || maxTurnsPerDay > 48) {
-    throw new TorchError('Continuation requires an explicit enable/pause and 1–48 daily turns.', { code: 'CONTINUATION_POLICY_INVALID' });
+  const previous = read(control);
+  maxTurnsPerDay ??= previous.maxTurnsPerDay;
+  maxConcurrency ??= previous.maxConcurrency;
+  if (typeof enabled !== 'boolean' || !Number.isInteger(maxTurnsPerDay) || maxTurnsPerDay < 1 || maxTurnsPerDay > 48
+    || !Number.isInteger(maxConcurrency) || maxConcurrency < 1 || maxConcurrency > 7) {
+    throw new TorchError('Continuation requires explicit enable/pause, 1–48 daily turns and 1–7 concurrent turns.', { code: 'CONTINUATION_POLICY_INVALID' });
   }
-  const state = { ...read(control), enabled, maxTurnsPerDay, areas: control.listAgents().map(a => a.areaId) };
+  const state = { ...previous, enabled, maxTurnsPerDay, maxConcurrency, areas: control.listAgents().map(a => a.areaId) };
   save(control, state, true);
   control.auditOwnerAction({ actorId, operation: 'runtime.continuation.configure', entityType: 'project', entityId: control.projectId,
-    details: { enabled, maxTurnsPerDay, areas: state.areas } });
-  return { enabled, maxTurnsPerDay, areas: state.areas, sessionsStarted: false };
+    details: { enabled, maxTurnsPerDay, maxConcurrency, areas: state.areas } });
+  return { enabled, maxTurnsPerDay, maxConcurrency, areas: state.areas, sessionsStarted: false };
 }
 
 export function observeContinuation({ stateRoot, tasks = [], now = () => new Date() } = {}) {
@@ -54,7 +59,7 @@ export function observeContinuation({ stateRoot, tasks = [], now = () => new Dat
       return active.length && progress?.status === 0 && progress.noProgress >= 2 && progress.taskKey === taskKey(active);
     }).map(([areaId]) => ({ areaId, reason: 'no-progress-needs-coordination' }));
     return { available: existsSync(join(stateRoot, 'continuation-policy.json')), enabled: state.enabled,
-      day, dayTimezone: 'UTC', attempts, maxTurnsPerDay: state.maxTurnsPerDay, remainingTurns,
+      day, dayTimezone: 'UTC', attempts, maxTurnsPerDay: state.maxTurnsPerDay, maxConcurrency: state.maxConcurrency, remainingTurns,
       stopReason: !state.enabled ? 'paused' : remainingTurns === 0 ? 'daily-turn-cap' : null,
       capacityHeldBy: occupiedAreas({ stateRoot }), held,
       budgetKind: 'automatic-turn-cap', manualTurnsIncluded: false, tokenOrCostGovernance: 'unknown',
@@ -64,14 +69,15 @@ export function observeContinuation({ stateRoot, tasks = [], now = () => new Dat
   }
 }
 
-export function planContinuation(control, backlog, { at = new Date(), limit = 3 } = {}) {
+export function planContinuation(control, backlog, { at = new Date(), limit } = {}) {
   const state = read(control);
   const day = at.toISOString().slice(0, 10);
   const attempts = state.day === day ? state.attempts : 0;
-  if (!Number.isInteger(limit) || limit < 1 || limit > 3) throw new TorchError('Continuation limit must be 1–3.', { code: 'CONTINUATION_POLICY_INVALID' });
+  if (limit !== undefined && (!Number.isInteger(limit) || limit < 1 || limit > 7)) throw new TorchError('Continuation limit must be 1–7.', { code: 'CONTINUATION_POLICY_INVALID' });
+  const capacity = Math.min(limit ?? state.maxConcurrency, state.maxConcurrency);
   const metadata = JSON.parse(readFileSync(join(control.stateRoot, 'project.json'), 'utf8'));
   if (!state.enabled || metadata.detachedAt || attempts >= state.maxTurnsPerDay) {
-    return { enabled: state.enabled, reason: metadata.detachedAt ? 'detached' : !state.enabled ? 'paused' : 'daily-turn-cap', candidates: [], attempts };
+    return { enabled: state.enabled, maxConcurrency: capacity, reason: metadata.detachedAt ? 'detached' : !state.enabled ? 'paused' : 'daily-turn-cap', candidates: [], attempts };
   }
   const tasks = backlog.list();
   const candidates = [];
@@ -101,12 +107,12 @@ export function planContinuation(control, backlog, { at = new Date(), limit = 3
   const ordinal = area => state.handled[area]?.dispatchedDay === day ? state.handled[area].dispatchedOrdinal ?? 0 : 0;
   candidates.sort((a, b) => ordinal(a.areaId) - ordinal(b.areaId)
     || Number(b.areaId === 'session-manager') - Number(a.areaId === 'session-manager'));
-  return { enabled: true, day, attempts, remainingTurns: state.maxTurnsPerDay - attempts,
+  return { enabled: true, maxConcurrency: capacity, day, attempts, remainingTurns: state.maxTurnsPerDay - attempts,
     occupiedAreas: occupiedAreas(control), held,
-    candidates: candidates.slice(0, Math.min(limit, state.maxTurnsPerDay - attempts)), mutationPerformed: false };
+    candidates: candidates.slice(0, Math.min(capacity, state.maxTurnsPerDay - attempts)), mutationPerformed: false };
 }
 
-export async function runContinuation(control, backlog, { actorId, launch, at, clock = () => at ?? new Date(), limit = 3 } = {}) {
+export async function runContinuation(control, backlog, { actorId, launch, at, clock = () => at ?? new Date(), limit } = {}) {
   control.assertOwnerActor(actorId);
   if (typeof launch !== 'function') throw new TorchError('Continuation needs an authorized launcher.', { code: 'RUNTIME_EXECUTION_NOT_AUTHORIZED' });
   const lock = join(control.stateRoot, 'continuation.lock');
@@ -121,9 +127,9 @@ export async function runContinuation(control, backlog, { actorId, launch, at, c
     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 current = planContinuation(control, backlog, { at: clock(), limit });
       let occupied = new Set([...occupiedAreas(control), ...pending.keys()]);
-      while (occupied.size < limit && current.enabled && !current.reason) {
+      while (occupied.size < current.maxConcurrency && current.enabled && !current.reason) {
         const candidate = current.candidates.find(c => !occupied.has(c.areaId));
         if (!candidate) break;
         const state = read(control);
@@ -136,10 +142,10 @@ export async function runContinuation(control, backlog, { actorId, launch, at, c
         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 });
+        current = planContinuation(control, backlog, { at: clock(), limit });
         occupied = new Set([...occupiedAreas(control), ...pending.keys()]);
       }
-      reason = current.reason ?? (occupied.size >= limit ? 'capacity-full' : null);
+      reason = current.reason ?? (occupied.size >= current.maxConcurrency ? 'capacity-full' : null);
       if (!pending.size) break;
       const completed = await Promise.race(pending.values());
       pending.delete(completed.areaId);

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, observeContinuation, planContinuation, runContinuation } from '../../src/runtime/continuation.mjs';
 
-function fixture({ recoveryWorktree = true, extraWorker = false } = {}) {
+function fixture({ recoveryWorktree = true, extraWorker = false, workerCount = 0 } = {}) {
   const root = mkdtempSync(join(tmpdir(), 'torch-legacy-recovery-'));
   for (const args of [['init', '-b', 'main'], ['config', 'user.email', '[email protected]'],
     ['config', 'user.name', 'Recovery Fixture']]) execFileSync('git', args, { cwd: root, stdio: 'pipe' });
@@ -24,6 +24,7 @@ function fixture({ recoveryWorktree = true, extraWorker = false } = {}) {
   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/**'] });
+  for (let i = 0; i < workerCount; i++) proposal.domains.push({ ...proposal.domains[0], id: `worker-${i}`, title: `Worker ${i}`, owned_paths: [`worker-${i}/**`] });
   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' });
@@ -218,6 +219,36 @@ test('SCN-runtime-continuation: explicit enable, event deduplication, daily cap
   } finally { f.control.close(); }
 });
 
+test('SCN-runtime-continuation-concurrency: seven authorized starts preserve budget and obey tighter limits', async () => {
+  const f = fixture({ workerCount: 5 });
+  const backlog = { list: () => [] };
+  const at = new Date('2026-10-03T00:00:00Z');
+  try {
+    for (const agent of f.control.listAgents()) {
+      f.control.reportStatus({ areaId: agent.areaId, state: 'idle', summary: 'Ready' });
+      f.control.sendOwnerRequest({ actorId: 'owner', recipient: agent.areaId, body: 'Review work' });
+    }
+    configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 48 });
+    assert.equal(planContinuation(f.control, backlog, { at }).maxConcurrency, 3);
+    assert.equal(planContinuation(f.control, backlog, { at, limit: 7 }).candidates.length, 3, 'Tick cannot exceed approved capacity');
+    configureContinuation(f.control, { actorId: 'owner', enabled: true, maxConcurrency: 7 });
+    assert.equal(observeContinuation({ stateRoot: f.control.stateRoot }).maxTurnsPerDay, 48);
+    assert.equal(planContinuation(f.control, backlog, { at, limit: 2 }).candidates.length, 2);
+    for (const value of [0, 8, 1.5]) assert.throws(() => configureContinuation(f.control,
+      { actorId: 'owner', enabled: true, maxConcurrency: value }), { code: 'CONTINUATION_POLICY_INVALID' });
+    const completions = [];
+    const running = runContinuation(f.control, backlog, { actorId: 'owner', at,
+      launch: () => new Promise(resolve => completions.push(resolve)) });
+    assert.equal(completions.length, 7, 'All seven start before any finishes');
+    assert.equal(planContinuation(f.control, backlog, { at }).attempts, 7);
+    configureContinuation(f.control, { actorId: 'owner', enabled: false });
+    assert.equal(observeContinuation({ stateRoot: f.control.stateRoot, now: () => at }).attempts, 7);
+    assert.equal(observeContinuation({ stateRoot: f.control.stateRoot }).maxConcurrency, 7);
+    completions.forEach(resolve => resolve(0));
+    assert.equal((await running).launched.length, 7);
+  } 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 {