From d56dbf159bba62c662c55cb347b40a8936411688 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN Date: Sun, 27 Sep 2026 19:09:05 -0400 Subject: [PATCH] Strengthen fleet wind-down recovery --- README.md | 16 ++- docs/PORTABLE_AGENT_FLEET_SPEC.md | 11 ++ ...grity-note-2026-09-27-portable-kernel.yaml | 7 +- src/runtime/lifecycle.mjs | 126 ++++++++++++++++-- test/fleet/lifecycle.test.mjs | 26 +++- 5 files changed, 163 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index ac73e41..9facfab 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,9 @@ and the first Claude reference-fleet path: launch, without changing the user's global runtime configuration; - durable-first Codex steering with native `queue` notification when available; - worker-first/Session-Manager-last fresh start and resume planning; -- safe wind-down with active-work refusal and local resume snapshots. +- safe wind-down that refuses active work, Git operations, leases, check or + measurement guards, and in-flight landing, then records final worker status + plus local and tracked resume state. - shell-free configured checks with exact-commit receipts; - transactional FIFO resource queues and leases; - a native integration queue with durable authorization and fast-forward main protection; @@ -210,10 +212,14 @@ node bin/torch.mjs detach --yes --json ``` `capture` externalizes the current stable identities, runtime IDs, presence, -and worktree commits without stopping sessions. `detach` first enforces normal -wind-down safety, then marks runtime integration detached while preserving the -tracked organization, local state, branches, and worktrees. A later explicit -`up` reattaches the selected runtime sessions. +worktree commits, open integration, guards, and leases without stopping +sessions. `down` and `detach` refuse active work, Git operations, unreleased +leases, active checks or measurements, and in-flight landing. Each worker sends +the Session Manager a durable final status, and the manager stops last after +writing both the local snapshot and tracked `.torch/RESUME-BRIEF.md`. `detach` +then marks runtime integration detached while preserving the tracked +organization, local state, branches, and worktrees. A later explicit `up` +reattaches the selected runtime sessions. The Fleet can evolve when recurring work develops a coherent new boundary. The Session Manager may propose the domain through MCP or the CLI, but cannot diff --git a/docs/PORTABLE_AGENT_FLEET_SPEC.md b/docs/PORTABLE_AGENT_FLEET_SPEC.md index 31cc4e5..d10bd30 100644 --- a/docs/PORTABLE_AGENT_FLEET_SPEC.md +++ b/docs/PORTABLE_AGENT_FLEET_SPEC.md @@ -893,6 +893,17 @@ Before shutdown, every domain MUST: The Session Manager writes or validates `RESUME-BRIEF.md`, captures session mapping, records fleet-wide risks, and stops last. +Implementation checkpoint (2026-09-27): planned wind-down fails closed on +working, starting, or stopping identities; active Git operations; unreleased +resource leases; active check or measurement guards; and integration already +landing. Every non-offline worker sends a durable final status to the Session +Manager before runtime stop. TORCH preserves unfinished task identity, captures +runtime mappings plus worktrees, integration, guards, and leases locally, and +atomically writes `.torch/RESUME-BRIEF.md` in the Session Manager worktree +before the manager is considered stopped. This is scenario-tested with +virtualized runtime shutdown; live provider shutdown remains an operational +qualification gate. + ## 16. Git and worktree policy ### 16.1 Isolation diff --git a/reports/test-audit/test-integrity-note-2026-09-27-portable-kernel.yaml b/reports/test-audit/test-integrity-note-2026-09-27-portable-kernel.yaml index b09927e..58d6b74 100644 --- a/reports/test-audit/test-integrity-note-2026-09-27-portable-kernel.yaml +++ b/reports/test-audit/test-integrity-note-2026-09-27-portable-kernel.yaml @@ -103,9 +103,9 @@ test_integrity_note: when: the fleet starts fresh, winds down safely, and plans its next start then: workers start before the manager, runtime IDs persist, the manager stops last, and the next start resumes - id: SCN-fleet-wind-down-safety - given: a launch failure or a domain reporting active work + given: a launch failure, a domain reporting active work, an active Git operation, or an unreleased scarce resource when: startup or wind-down is attempted - then: manager launch is prevented after worker failure and active work blocks shutdown + then: manager launch is prevented after worker failure and every unresolved safety condition blocks shutdown - id: SCN-mixed-runtime given: one approved Codex worker domain and one Claude Session Manager with stable Fleet identities when: the mixed fleet starts and winds down through virtualized runtime boundaries @@ -254,7 +254,7 @@ test_integrity_note: - SQLite state after process restart - CLI and MCP output over real process boundaries - runtime command plans and execution order - - local resume snapshot and presence transitions + - local resume snapshot, tracked resume brief, final worker message, preserved task identity, and presence transitions - immutable check receipt commit and result - resource holder, FIFO position, expiry, and cancellation state - canonical main SHA before and after authorized landing @@ -299,6 +299,7 @@ test_integrity_note: - runtime sessions launching without their identity-bound TORCH tool surface - lost Codex thread IDs or native steering that bypasses durable messaging - manager-first startup or manager-first shutdown + - shutdown that loses unfinished task identity, skips final worker status, or proceeds through active work, Git operations, leases, checks, measurements, or landing - passing receipts detached from the commit actually measured - unfair or automatically stolen resource leases - integration based on branch labels, stale checks, or stale main diff --git a/src/runtime/lifecycle.mjs b/src/runtime/lifecycle.mjs index 812b2ac..6c7a596 100644 --- a/src/runtime/lifecycle.mjs +++ b/src/runtime/lifecycle.mjs @@ -95,22 +95,71 @@ function projectMetadata(stateRoot) { function captureSnapshot({ plan, controlPlane, now = () => new Date() }) { const snapshotPath = join(plan.stateRoot, 'sessions', 'resume.json'); - atomicJson(snapshotPath, { + const capturedAt = now().toISOString(); + const database = controlPlane.database; + const hasTable = (name) => Boolean(database.prepare( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", + ).get(name)); + const snapshot = { schema: 'torch.dev/runtime-snapshot/v1alpha1', projectId: plan.projectId, - capturedAt: now().toISOString(), agents: controlPlane.listAgents(), + capturedAt, agents: controlPlane.listAgents(), worktrees: plan.actions.map((action) => action.worktree).filter(Boolean), - }); - return snapshotPath; + resourceLeases: hasTable('resource_leases') ? database.prepare(` + SELECT resource_id AS resourceId, area_id AS areaId, acquired_at AS acquiredAt, + expires_at AS expiresAt FROM resource_leases WHERE released_at IS NULL ORDER BY resource_id, acquired_at + `).all() : [], + worktreeGuards: hasTable('worktree_guards') ? database.prepare(` + SELECT area_id AS areaId, guard_type AS type, reason, created_at AS createdAt + FROM worktree_guards WHERE released_at IS NULL ORDER BY created_at, id + `).all() : [], + integration: hasTable('integration_requests') ? database.prepare(` + SELECT id, source_area AS sourceArea, source_commit AS sourceCommit, state, reason + FROM integration_requests WHERE state NOT IN ('landed', 'superseded') ORDER BY created_at, id + `).all() : [], + }; + atomicJson(snapshotPath, snapshot); + return { snapshotPath, snapshot }; +} + +function writeResumeBrief(plan, snapshot) { + if (!plan.resumeBriefPath) return null; + const lines = [ + '# TORCH resume brief', '', + `Captured: ${snapshot.capturedAt}`, + `Project: ${snapshot.projectId}`, + '', '## Fleet identities', '', + ...snapshot.agents.map((agent) => + `- ${agent.areaId}: ${agent.state}; runtime=${agent.runtime ?? 'none'}; task=${agent.currentTask ?? agent.task ?? 'none'}; summary=${agent.summary ?? 'none'}`), + '', '## Worktrees', '', + ...(snapshot.worktrees.length ? snapshot.worktrees.map((worktree) => + `- ${worktree.areaId}: branch=${worktree.branch}; commit=${worktree.commit}; dirty=${worktree.dirtyEntries.length ? worktree.dirtyEntries.join(', ') : 'no'}`) : ['- No managed worktrees recorded.']), + '', '## Open integration', '', + ...(snapshot.integration.length ? snapshot.integration.map((request) => + `- ${request.id}: ${request.sourceArea} ${request.sourceCommit} is ${request.state}${request.reason ? ` (${request.reason})` : ''}`) : ['- None.']), + '', '## Guards and leases', '', + ...(snapshot.worktreeGuards.length ? snapshot.worktreeGuards.map((guard) => + `- guard: ${guard.areaId} ${guard.type} — ${guard.reason}`) : ['- No active worktree guards.']), + ...(snapshot.resourceLeases.length ? snapshot.resourceLeases.map((lease) => + `- lease: ${lease.areaId} holds ${lease.resourceId} until ${lease.expiresAt}`) : ['- No active resource leases.']), + '', 'Resume by checking live identity, inbox, backlog, worktree state, and exact check receipts before dispatch.', '', + ]; + const temporary = `${plan.resumeBriefPath}.tmp-${process.pid}`; + writeFileSync(temporary, `${lines.join('\n')}\n`, { encoding: 'utf8', mode: 0o600, flag: 'wx' }); + renameSync(temporary, plan.resumeBriefPath); + return plan.resumeBriefPath; } function inspectWorktree(worktree) { const git = (args) => execFileSync('git', ['-C', worktree.path, ...args], { encoding: 'utf8' }).trim(); + const operations = ['MERGE_HEAD', 'REBASE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD'] + .filter((name) => existsSync(git(['rev-parse', '--git-path', name]))); return { areaId: worktree.area, path: worktree.path, branch: git(['branch', '--show-current']), commit: git(['rev-parse', 'HEAD']), dirtyEntries: git(['status', '--porcelain']).split('\n').filter(Boolean), + operations, }; } @@ -334,12 +383,16 @@ export function planFleetDown({ repositoryRoot, controlPlane, adapters } = {}) { const worktrees = worktreesByArea(manifest); const runtimes = adapterMap({ adapters }); const adapterRegistryProvided = adapters !== undefined; + const database = controlPlane.database; + const hasTable = (name) => Boolean(database.prepare( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", + ).get(name)); const actions = orderedAreas(roster).map((area) => { const identity = controlPlane.identity(area.id); const worktree = worktrees.get(area.id); return { areaId: area.id, state: identity.state, runtime: identity.runtime, - runtimeSessionId: identity.runtimeSessionId, + runtimeSessionId: identity.runtimeSessionId, currentTask: identity.currentTask, requiresRuntimeStop: identity.runtimeSessionId ? runtimes.get(identity.runtime)?.capabilities.stopSession !== false : false, @@ -355,15 +408,40 @@ export function planFleetDown({ repositoryRoot, controlPlane, adapters } = {}) { blockers.push(...actions .filter((action) => action.runtimeAdapterMissing) .map((action) => ({ areaId: action.areaId, runtime: action.runtime, code: 'RUNTIME_ADAPTER_MISSING' }))); + blockers.push(...actions + .filter((action) => action.worktree?.operations.length) + .map((action) => ({ + areaId: action.areaId, code: 'GIT_OPERATION_ACTIVE', operations: action.worktree.operations, + }))); + if (hasTable('resource_leases')) { + blockers.push(...database.prepare(` + SELECT resource_id AS resourceId, area_id AS areaId, expires_at AS expiresAt + FROM resource_leases WHERE released_at IS NULL ORDER BY resource_id, acquired_at + `).all().map((lease) => ({ ...lease, code: 'RESOURCE_LEASE_ACTIVE' }))); + } + if (hasTable('worktree_guards')) { + blockers.push(...database.prepare(` + SELECT area_id AS areaId, guard_type AS type, reason + FROM worktree_guards WHERE released_at IS NULL AND guard_type IN ('check', 'measurement') + ORDER BY created_at, id + `).all().map((guard) => ({ ...guard, code: 'WORKTREE_GUARD_ACTIVE' }))); + } + if (hasTable('integration_requests')) { + blockers.push(...database.prepare(` + SELECT id, source_area AS areaId FROM integration_requests WHERE state = 'landing' ORDER BY created_at, id + `).all().map((request) => ({ ...request, code: 'INTEGRATION_LANDING_ACTIVE' }))); + } + const managerWorktree = worktrees.get('session-manager'); return { action: 'fleet-down', projectId: manifest.projectId, stateRoot: localStateRoot(manifest), + resumeBriefPath: managerWorktree ? join(managerWorktree.path, '.torch', 'RESUME-BRIEF.md') : null, actions, blockers, canProceed: blockers.length === 0, mutationPerformed: false, }; } export function stopFleet({ plan, controlPlane, stopRuntime, now = () => new Date() } = {}) { if (!plan?.canProceed) { - throw new TorchError('Fleet wind-down refused while sessions report active work', { + throw new TorchError('Fleet wind-down refused while safety conditions remain unresolved', { code: 'FLEET_NOT_READY_TO_STOP', details: plan?.blockers ?? [], }); } @@ -374,9 +452,21 @@ export function stopFleet({ plan, controlPlane, stopRuntime, now = () => new Dat } const stopped = []; for (const action of plan.actions) { + if (action.areaId !== 'session-manager' && action.state !== 'offline') { + controlPlane.sendMessage({ + sender: action.areaId, recipient: 'session-manager', + body: `Final status: ${action.state}. ${action.worktree + ? `Branch ${action.worktree.branch} at ${action.worktree.commit}; ${action.worktree.dirtyEntries.length} dirty entries.` + : 'No managed worktree.'}`, + references: { commit: action.worktree?.commit }, + }); + } if (!action.runtimeSessionId || action.state === 'offline' || action.requiresRuntimeStop === false) { if (action.state !== 'offline') { - controlPlane.reportStatus({ areaId: action.areaId, state: 'offline', summary: 'No resident runtime process to stop.' }); + controlPlane.reportStatus({ + areaId: action.areaId, state: 'offline', summary: 'No resident runtime process to stop.', + task: action.currentTask, + }); } stopped.push({ areaId: action.areaId, runtimeSessionId: action.runtimeSessionId, @@ -384,24 +474,34 @@ export function stopFleet({ plan, controlPlane, stopRuntime, now = () => new Dat }); continue; } - controlPlane.reportStatus({ areaId: action.areaId, state: 'stopping', summary: 'TORCH wind-down requested.' }); + controlPlane.reportStatus({ + areaId: action.areaId, state: 'stopping', summary: 'TORCH wind-down requested.', task: action.currentTask, + }); const result = stopRuntime(action); if (result?.stopped === false || (result?.status !== undefined && result.status !== 0)) { - controlPlane.reportStatus({ areaId: action.areaId, state: 'waiting', summary: 'Runtime stop failed.' }); + controlPlane.reportStatus({ + areaId: action.areaId, state: 'waiting', summary: 'Runtime stop failed.', task: action.currentTask, + }); throw new TorchError(`Fleet wind-down failed for ${action.areaId}`, { code: 'FLEET_STOP_FAILED', details: { areaId: action.areaId, result }, }); } - controlPlane.reportStatus({ areaId: action.areaId, state: 'offline', summary: 'Runtime stopped cleanly.' }); + controlPlane.reportStatus({ + areaId: action.areaId, state: 'offline', summary: 'Runtime stopped cleanly.', task: action.currentTask, + }); stopped.push({ areaId: action.areaId, runtimeSessionId: action.runtimeSessionId, alreadyOffline: false }); } - const snapshotPath = captureSnapshot({ plan, controlPlane, now }); - return { projectId: plan.projectId, stopped, snapshotPath, mutationPerformed: true }; + const captured = captureSnapshot({ plan, controlPlane, now }); + const resumeBriefPath = writeResumeBrief(plan, captured.snapshot); + return { + projectId: plan.projectId, stopped, snapshotPath: captured.snapshotPath, + resumeBriefPath, mutationPerformed: true, + }; } export function captureFleet({ repositoryRoot, controlPlane, now = () => new Date() } = {}) { const plan = planFleetDown({ repositoryRoot, controlPlane }); - const snapshotPath = captureSnapshot({ plan, controlPlane, now }); + const { snapshotPath } = captureSnapshot({ plan, controlPlane, now }); return { action: 'capture', projectId: plan.projectId, snapshotPath, agents: controlPlane.listAgents(), mutationPerformed: true, diff --git a/test/fleet/lifecycle.test.mjs b/test/fleet/lifecycle.test.mjs index b7d7014..19d2dd8 100644 --- a/test/fleet/lifecycle.test.mjs +++ b/test/fleet/lifecycle.test.mjs @@ -1,6 +1,6 @@ import assert from 'node:assert/strict'; import { execFileSync } from 'node:child_process'; -import { existsSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs'; +import { existsSync, mkdtempSync, readFileSync, unlinkSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { basename, join } from 'node:path'; import test from 'node:test'; @@ -12,6 +12,7 @@ import { proposeDomains } from '../../src/kernel/domains.mjs'; import { inspectRepository } from '../../src/kernel/git.mjs'; import { installProject } from '../../src/kernel/install.mjs'; import { createWorktrees } from '../../src/kernel/worktrees.mjs'; +import { ResourceService } from '../../src/resources/service.mjs'; import { detachFleet, planFleetDetach, planFleetDown, planFleetUp, startFleet, stopFleet, } from '../../src/runtime/lifecycle.mjs'; @@ -27,6 +28,8 @@ function fleetFixture({ workerRuntime = 'claude' } = {}) { let repository = inspectRepository(root); const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) }); proposal.domains[0].runtime = workerRuntime; + proposal.domains[0].resources = ['browser']; + proposal.resources = [{ id: 'browser', capacity: 1, queue: 'fifo', max_hold_seconds: 300 }]; proposal.review = { status: 'approved', reviewedAt: '2026-09-27T00:00:00Z', reviewedBy: 'fixture-owner', notes: [], }; @@ -63,7 +66,9 @@ test('SCN-fleet-fresh-resume: workers start before manager and stable identities assert.equal(existsSync(fresh.actions[0].promptFile), true); assert.equal(readFileSync(fresh.actions[0].promptFile, 'utf8').includes('Common fleet rules'), true); - control.reportStatus({ areaId: fixture.worker, state: 'idle', summary: 'Checkpoint safe.' }); + control.reportStatus({ + areaId: fixture.worker, state: 'idle', summary: 'Checkpoint safe.', task: 'TASK-RESUME', + }); control.reportStatus({ areaId: 'session-manager', state: 'idle', summary: 'Fleet safe.' }); const stoppedIds = []; const down = stopFleet({ @@ -73,6 +78,12 @@ test('SCN-fleet-fresh-resume: workers start before manager and stable identities assert.equal(down.stopped.at(-1).areaId, 'session-manager'); assert.equal(stoppedIds.length, fresh.actions.length); assert.equal(existsSync(down.snapshotPath), true); + assert.equal(existsSync(down.resumeBriefPath), true); + assert.match(readFileSync(down.resumeBriefPath, 'utf8'), /TASK-RESUME/); + assert.match(readFileSync(down.resumeBriefPath, 'utf8'), new RegExp(fixture.worker)); + assert.equal(control.identity(fixture.worker).currentTask, 'TASK-RESUME'); + assert.equal(control.readMessages({ recipient: 'session-manager' }) + .some((message) => message.body.startsWith('Final status:')), true); const detached = detachFleet({ plan: planFleetDetach({ repositoryRoot: fixture.root, controlPlane: control }), @@ -111,13 +122,24 @@ test('SCN-fleet-wind-down-safety: active work blocks runtime shutdown and failed && control.identity('session-manager').state === 'offline', ); control.reportStatus({ areaId: fixture.worker, state: 'working', summary: 'Unfinished change.' }); + const resources = new ResourceService({ repositoryRoot: fixture.root, controlPlane: control }); + resources.acquire({ resourceId: 'browser', areaId: fixture.worker }); + const workerPath = plan.actions.find((action) => action.areaId === fixture.worker).worktree; + const mergeHead = execFileSync('git', ['-C', workerPath, 'rev-parse', '--git-path', 'MERGE_HEAD'], { + encoding: 'utf8', + }).trim(); + writeFileSync(mergeHead, `${execFileSync('git', ['-C', fixture.root, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim()}\n`); const down = planFleetDown({ repositoryRoot: fixture.root, controlPlane: control }); assert.equal(down.canProceed, false); assert.equal(down.blockers.some((blocker) => blocker.areaId === fixture.worker && blocker.state === 'working'), true); + assert.equal(down.blockers.some((blocker) => blocker.code === 'RESOURCE_LEASE_ACTIVE'), true); + assert.equal(down.blockers.some((blocker) => blocker.code === 'GIT_OPERATION_ACTIVE'), true); assert.throws( () => stopFleet({ plan: down, controlPlane: control, stopRuntime: () => ({ stopped: true }) }), (error) => error.code === 'FLEET_NOT_READY_TO_STOP', ); + unlinkSync(mergeHead); + resources.release({ resourceId: 'browser', areaId: fixture.worker }); control.close(); });