diff --git a/README.md b/README.md index 214c1bf..a4afc85 100644 --- a/README.md +++ b/README.md @@ -46,6 +46,7 @@ and the first Claude reference-fleet path: - 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; +- domain-owned convergence plans that defer during checks, measurements, pins, dirty work, and Git operations; - a local bare canonical Git option with honest one-disk/local-remote grading; - mutation audit events tied to project and Fleet identity. - isolated candidate version stores with explicit compatibility declarations; diff --git a/package.json b/package.json index 77c9427..42f2d43 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ "./adapters/codex": "./src/adapters/codex.mjs", "./canonical/local": "./src/canonical/local.mjs", "./checks": "./src/checks/service.mjs", + "./convergence": "./src/convergence/service.mjs", "./console": "./src/console/server.mjs", "./design/brief": "./src/design/brief.mjs", "./design/architect": "./src/design/architect.mjs", @@ -60,6 +61,7 @@ "src/canonical/", "src/checks/", "src/console/", + "src/convergence/", "src/control-plane/", "src/design/", "src/evolution/", @@ -85,8 +87,8 @@ "torch": "node bin/torch.mjs", "check": "node --check bin/torch.mjs && node --check src/cli.mjs && node --check src/mcp/server.mjs && node --check src/console/server.mjs && node --check site/app.js && node --check site/console.js", "site:serve": "node bin/torch.mjs console serve", - "format": "eslint src/adapters src/backlog src/canonical src/checks src/console src/control-plane src/design src/evolution src/importers src/integration src/kernel src/mcp src/observability src/resources src/runtime src/schedules src/self-host src/telemetry src/cli.mjs bin/torch.mjs site/*.js test/fleet --fix", - "lint": "eslint src/adapters src/backlog src/canonical src/checks src/console src/control-plane src/design src/evolution src/importers src/integration src/kernel src/mcp src/observability src/resources src/runtime src/schedules src/self-host src/telemetry src/cli.mjs bin/torch.mjs site/*.js test/fleet" + "format": "eslint src/adapters src/backlog src/canonical src/checks src/console src/convergence src/control-plane src/design src/evolution src/importers src/integration src/kernel src/mcp src/observability src/resources src/runtime src/schedules src/self-host src/telemetry src/cli.mjs bin/torch.mjs site/*.js test/fleet --fix", + "lint": "eslint src/adapters src/backlog src/canonical src/checks src/console src/convergence src/control-plane src/design src/evolution src/importers src/integration src/kernel src/mcp src/observability src/resources src/runtime src/schedules src/self-host src/telemetry src/cli.mjs bin/torch.mjs site/*.js test/fleet" }, "dependencies": { "@modelcontextprotocol/server": "^2.1.0", 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 b495bed..fe1f763 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 @@ -126,6 +126,10 @@ test_integrity_note: given: a converged domain commit with every required exact-SHA receipt when: the domain requests integration and the Session Manager authorizes and lands it then: main fast-forwards only while source, target, receipts, cleanliness, and authority remain valid + - id: SCN-safe-convergence + given: a managed domain worktree behind canonical main + when: a check, measurement, pin, dirty state, or Git operation could make convergence unsafe + then: TORCH defers the merge, releases automatic check guards, and only the owning clean worktree may intentionally merge canonical state - id: SCN-local-canonical given: an installed project with no forge or remote when: the owner creates a local canonical Git store and later makes and pushes another commit diff --git a/src/checks/service.mjs b/src/checks/service.mjs index 9516898..229b0fc 100644 --- a/src/checks/service.mjs +++ b/src/checks/service.mjs @@ -5,6 +5,7 @@ import { join } from 'node:path'; import { TorchError } from '../kernel/errors.mjs'; import { readInstallManifest } from '../kernel/install.mjs'; import { loadFleetDefinition } from '../kernel/worktrees.mjs'; +import { beginWorktreeGuard, endWorktreeGuard } from '../convergence/service.mjs'; function git(root, args) { const result = spawnSync('git', ['-C', root, ...args], { encoding: 'utf8' }); @@ -109,9 +110,20 @@ export class CheckService { } const receiptId = this.idFactory(); const startedAt = this.clock().toISOString(); - const result = this.executor(plan.definition.command, plan.definition.args ?? [], { - cwd: plan.worktree, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + beginWorktreeGuard(this.controlPlane, { + areaId: plan.areaId, type: 'check', reason: `Running ${checkId}`, + clock: this.clock, idFactory: this.idFactory, }); + let result; + try { + result = this.executor(plan.definition.command, plan.definition.args ?? [], { + cwd: plan.worktree, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'], + }); + } finally { + endWorktreeGuard(this.controlPlane, { + areaId: plan.areaId, type: 'check', clock: this.clock, missingOk: true, + }); + } const finishedAt = this.clock().toISOString(); const afterCommit = git(plan.worktree, ['rev-parse', 'HEAD']); const afterDirty = git(plan.worktree, ['status', '--porcelain']).split('\n').filter(Boolean); diff --git a/src/cli.mjs b/src/cli.mjs index badd201..2df5312 100644 --- a/src/cli.mjs +++ b/src/cli.mjs @@ -30,6 +30,7 @@ import { createArchitectRunPlan, loadArchitectArtifact, runSessionArchitect, validateArchitectProposal, } from './design/architect.mjs'; import { FleetEvolutionService } from './evolution/service.mjs'; +import { ConvergenceService } from './convergence/service.mjs'; const HELP = `TORCH — portable agent fleet @@ -73,6 +74,11 @@ Usage: torch resources acquire --id --area [--json] torch resources release --id --area [--json] torch resources cancel --id --area [--json] + torch converge plan --area [--json] + torch converge run --area --yes [--json] + torch converge guards --area [--json] + torch converge hold --area --type --reason [--json] + torch converge release --area --type [--json] torch integrate list [--state ] [--json] torch integrate request --area [--commit ] [--json] torch integrate evaluate --request [--json] @@ -777,6 +783,43 @@ export async function runCli(argv = process.argv.slice(2), { control.close(); } } + if (command === 'converge') { + const operation = argv[1] ?? 'plan'; + const control = openControlPlane({ repositoryRoot: repository.root, env }); + try { + const convergence = new ConvergenceService({ repositoryRoot: repository.root, controlPlane: control }); + const areaId = optionValue(argv, '--area'); + if (operation === 'plan') print(convergence.plan({ areaId }), { json }); + else if (operation === 'guards') print({ guards: convergence.guards(areaId) }, { json }); + else if (operation === 'hold') { + const type = optionValue(argv, '--type'); + if (!['measurement', 'pin'].includes(type)) { + throw new TorchError('CLI guards may only hold measurement or pin; check guards are automatic.', { + code: 'WORKTREE_GUARD_INVALID', details: { type: type ?? null }, + }); + } + print(convergence.hold({ areaId, type, reason: optionValue(argv, '--reason') }), { json }); + } else if (operation === 'release') { + const type = optionValue(argv, '--type'); + if (!['measurement', 'pin'].includes(type)) { + throw new TorchError('CLI guards may only release measurement or pin; check guards are automatic.', { + code: 'WORKTREE_GUARD_INVALID', details: { type: type ?? null }, + }); + } + print(convergence.release({ areaId, type }), { json }); + } else if (operation === 'run') { + if (!argv.includes('--yes')) { + throw new TorchError('Convergence merges canonical state into a domain worktree. Review converge plan, then re-run with --yes.', { + code: 'APPROVAL_REQUIRED', + }); + } + print(convergence.converge({ areaId }), { json }); + } else throw new TorchError(`Unknown converge operation: ${operation}`, { code: 'UNKNOWN_COMMAND' }); + return 0; + } finally { + control.close(); + } + } if (command === 'resources') { const operation = argv[1] ?? 'list'; const control = openControlPlane({ repositoryRoot: repository.root, env }); diff --git a/src/convergence/service.mjs b/src/convergence/service.mjs new file mode 100644 index 0000000..a48ee13 --- /dev/null +++ b/src/convergence/service.mjs @@ -0,0 +1,164 @@ +import { randomUUID } from 'node:crypto'; +import { existsSync } from 'node:fs'; +import { spawnSync } from 'node:child_process'; +import { TorchError } from '../kernel/errors.mjs'; +import { readInstallManifest } from '../kernel/install.mjs'; +import { loadFleetDefinition } from '../kernel/worktrees.mjs'; + +const GUARD_TYPES = Object.freeze(['check', 'measurement', 'pin']); + +function git(root, args, { allowFailure = false } = {}) { + const result = spawnSync('git', ['-C', root, ...args], { encoding: 'utf8' }); + if (result.status !== 0 && !allowFailure) { + throw new TorchError(`Git command failed: git ${args.join(' ')}`, { + code: 'GIT_COMMAND_FAILED', details: result.stderr?.trim() || null, + }); + } + return { status: result.status, stdout: result.stdout?.trim() ?? '', stderr: result.stderr?.trim() ?? '' }; +} + +export function initializeWorktreeGuards(database) { + database.exec(` + CREATE TABLE IF NOT EXISTS worktree_guards ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + area_id TEXT NOT NULL, + guard_type TEXT NOT NULL, + reason TEXT NOT NULL, + created_at TEXT NOT NULL, + released_at TEXT + ); + CREATE UNIQUE INDEX IF NOT EXISTS one_open_guard_per_type + ON worktree_guards(area_id, guard_type) WHERE released_at IS NULL; + `); +} + +export function beginWorktreeGuard(controlPlane, { + areaId, type, reason, clock = () => new Date(), idFactory = randomUUID, +} = {}) { + const area = controlPlane.assertIdentity(areaId); + if (!GUARD_TYPES.includes(type)) throw new TorchError(`Invalid worktree guard: ${type}`, { code: 'WORKTREE_GUARD_INVALID' }); + if (typeof reason !== 'string' || !reason.trim()) throw new TorchError('Worktree guard reason is required', { code: 'WORKTREE_GUARD_INVALID' }); + initializeWorktreeGuards(controlPlane.database); + const guard = { + id: idFactory(), projectId: controlPlane.projectId, areaId: area, type, + reason: reason.trim(), createdAt: clock().toISOString(), releasedAt: null, + }; + try { + controlPlane.database.prepare(` + INSERT INTO worktree_guards (id, project_id, area_id, guard_type, reason, created_at) + VALUES (?, ?, ?, ?, ?, ?) + `).run(guard.id, guard.projectId, guard.areaId, guard.type, guard.reason, guard.createdAt); + } catch (error) { + throw new TorchError(`${area} already has an active ${type} guard`, { + code: 'WORKTREE_GUARD_EXISTS', details: error.message, + }); + } + controlPlane.audit({ + actorId: area, operation: 'worktree-guard.begin', entityType: 'worktree-guard', entityId: guard.id, + details: { type, reason: guard.reason }, + }); + return guard; +} + +export function endWorktreeGuard(controlPlane, { + areaId, type, clock = () => new Date(), missingOk = false, +} = {}) { + const area = controlPlane.assertIdentity(areaId); + if (!GUARD_TYPES.includes(type)) throw new TorchError(`Invalid worktree guard: ${type}`, { code: 'WORKTREE_GUARD_INVALID' }); + initializeWorktreeGuards(controlPlane.database); + const guard = controlPlane.database.prepare(` + SELECT * FROM worktree_guards WHERE area_id = ? AND guard_type = ? AND released_at IS NULL + `).get(area, type); + if (!guard) { + if (missingOk) return null; + throw new TorchError(`${area} has no active ${type} guard`, { code: 'WORKTREE_GUARD_NOT_FOUND' }); + } + const releasedAt = clock().toISOString(); + controlPlane.database.prepare('UPDATE worktree_guards SET released_at = ? WHERE id = ?').run(releasedAt, guard.id); + controlPlane.audit({ + actorId: area, operation: 'worktree-guard.end', entityType: 'worktree-guard', entityId: guard.id, + details: { type }, + }); + return { id: guard.id, areaId: area, type, reason: guard.reason, createdAt: guard.created_at, releasedAt }; +} + +export class ConvergenceService { + constructor({ repositoryRoot, controlPlane, clock = () => new Date(), idFactory = randomUUID } = {}) { + this.repositoryRoot = repositoryRoot; + this.controlPlane = controlPlane; + this.clock = clock; + this.idFactory = idFactory; + const { config } = loadFleetDefinition(repositoryRoot); + this.targetBranch = config.project.main_branch; + this.policy = config.synchronization ?? {}; + const manifest = readInstallManifest(repositoryRoot); + this.worktrees = new Map((manifest.external ?? []) + .filter((entry) => entry.type === 'worktree') + .map((entry) => [entry.area, entry])); + initializeWorktreeGuards(controlPlane.database); + } + + guards(areaId) { + const area = this.controlPlane.assertIdentity(areaId); + return this.controlPlane.database.prepare(` + SELECT id, area_id AS areaId, guard_type AS type, reason, created_at AS createdAt + FROM worktree_guards WHERE area_id = ? AND released_at IS NULL ORDER BY created_at, id + `).all(area); + } + + hold({ areaId, type, reason } = {}) { + return beginWorktreeGuard(this.controlPlane, { + areaId, type, reason, clock: this.clock, idFactory: this.idFactory, + }); + } + + release({ areaId, type } = {}) { + return endWorktreeGuard(this.controlPlane, { areaId, type, clock: this.clock }); + } + + plan({ areaId } = {}) { + const area = this.controlPlane.assertIdentity(areaId); + const worktree = this.worktrees.get(area); + if (!worktree) throw new TorchError(`No managed worktree for ${area}`, { code: 'WORKTREE_MISSING' }); + const blockers = []; + const actualBranch = git(worktree.path, ['branch', '--show-current']).stdout; + if (actualBranch !== worktree.branch) blockers.push({ code: 'BRANCH_MISMATCH', expected: worktree.branch, actual: actualBranch || null }); + const dirtyEntries = git(worktree.path, ['status', '--porcelain=v1', '--untracked-files=all']).stdout.split('\n').filter(Boolean); + if (dirtyEntries.length) blockers.push({ code: 'WORKTREE_DIRTY', entries: dirtyEntries }); + const operationPaths = ['MERGE_HEAD', 'REBASE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD'] + .map((name) => ({ name, path: git(worktree.path, ['rev-parse', '--git-path', name]).stdout })) + .filter((entry) => entry.path && existsSync(entry.path)); + if (operationPaths.length) blockers.push({ code: 'GIT_OPERATION_ACTIVE', operations: operationPaths.map((entry) => entry.name) }); + const guards = this.guards(area); + for (const guard of guards) blockers.push({ code: `WORKTREE_${guard.type.toUpperCase()}_ACTIVE`, guard }); + const sourceCommit = git(worktree.path, ['rev-parse', 'HEAD']).stdout; + const targetCommit = git(this.repositoryRoot, ['rev-parse', this.targetBranch]).stdout; + const behind = Number(git(this.repositoryRoot, ['rev-list', '--count', `${sourceCommit}..${targetCommit}`]).stdout || 0); + const ahead = Number(git(this.repositoryRoot, ['rev-list', '--count', `${targetCommit}..${sourceCommit}`]).stdout || 0); + return { + action: 'merge-canonical-into-domain', areaId: area, worktree: worktree.path, + branch: worktree.branch, targetBranch: this.targetBranch, sourceCommit, targetCommit, + ahead, behind, upToDate: behind === 0, blockers, + canProceed: blockers.length === 0, mutationPerformed: false, + }; + } + + converge({ areaId } = {}) { + const plan = this.plan({ areaId }); + if (!plan.canProceed) throw new TorchError('Worktree convergence is blocked', { code: 'CONVERGENCE_BLOCKED', details: plan.blockers }); + if (plan.upToDate) return { ...plan, changed: false }; + const merged = git(plan.worktree, ['merge', '--no-edit', plan.targetCommit], { allowFailure: true }); + if (merged.status !== 0) { + throw new TorchError('Canonical merge requires domain-owned conflict resolution', { + code: 'CONVERGENCE_CONFLICT', details: { stderr: merged.stderr, worktree: plan.worktree }, + }); + } + const commit = git(plan.worktree, ['rev-parse', 'HEAD']).stdout; + this.controlPlane.audit({ + actorId: plan.areaId, operation: 'worktree.converge', entityType: 'worktree', entityId: plan.areaId, + details: { from: plan.sourceCommit, target: plan.targetCommit, commit }, + }); + return { ...plan, commit, changed: commit !== plan.sourceCommit, mutationPerformed: true }; + } +} diff --git a/src/kernel/doctor.mjs b/src/kernel/doctor.mjs index ba0bda1..eb5f42d 100644 --- a/src/kernel/doctor.mjs +++ b/src/kernel/doctor.mjs @@ -100,6 +100,13 @@ export function diagnoseProject({ repository, env = process.env }) { `).all(); if (requests.length) findings.push({ severity: 'info', code: 'INTEGRATION_QUEUE', requests }); } + if (hasTable('worktree_guards')) { + const 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(); + if (guards.length) findings.push({ severity: 'info', code: 'WORKTREE_GUARDS', guards }); + } database.close(); } catch (error) { findings.push({ severity: 'error', code: 'CONTROL_PLANE_INVALID', path: databasePath, message: error.message }); diff --git a/test/fleet/integration-canonical.test.mjs b/test/fleet/integration-canonical.test.mjs index b1df78b..0ed37f0 100644 --- a/test/fleet/integration-canonical.test.mjs +++ b/test/fleet/integration-canonical.test.mjs @@ -1,12 +1,13 @@ 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'; import { BacklogService } from '../../src/backlog/service.mjs'; import { createLocalCanonical, classifyRecoverability, planLocalCanonical } from '../../src/canonical/local.mjs'; import { CheckService } from '../../src/checks/service.mjs'; +import { ConvergenceService } from '../../src/convergence/service.mjs'; import { openControlPlane } from '../../src/control-plane/service.mjs'; import { analyzeRepository } from '../../src/kernel/analyze.mjs'; import { proposeDomains } from '../../src/kernel/domains.mjs'; @@ -146,3 +147,58 @@ test('SCN-local-canonical: a no-forge project gains a bare canonical remote and execFileSync('git', ['-C', context.root, 'push', 'torch-canonical', 'main']); assert.equal(classifyRecoverability({ repositoryRoot: context.root }).level, 'LOCAL-REMOTE'); }); + +test('SCN-safe-convergence: only the owning clean and unguarded worktree merges canonical state', () => { + const context = integrationFixture(); + const control = openControlPlane({ repositoryRoot: context.root, env: context.env }); + const convergence = new ConvergenceService({ repositoryRoot: context.root, controlPlane: control }); + const checks = new CheckService({ + repositoryRoot: context.root, + controlPlane: control, + executor: () => { + const duringCheck = convergence.plan({ areaId: context.worker }); + assert.equal(duringCheck.blockers.some((blocker) => blocker.code === 'WORKTREE_CHECK_ACTIVE'), true); + return { status: 0, stdout: 'pass', stderr: '' }; + }, + }); + checks.run({ checkId: 'verify-pass', areaId: context.worker }); + assert.deepEqual(convergence.guards(context.worker), []); + + writeFileSync(join(context.root, 'main-only.js'), 'export const canonical = true;\n'); + execFileSync('git', ['-C', context.root, 'add', 'main-only.js']); + execFileSync('git', ['-C', context.root, 'commit', '-m', 'canonical advanced']); + assert.equal(convergence.plan({ areaId: context.worker }).behind, 1); + + convergence.hold({ areaId: context.worker, type: 'measurement', reason: 'Stable benchmark inputs' }); + const guarded = convergence.plan({ areaId: context.worker }); + assert.equal(guarded.canProceed, false); + assert.equal(guarded.blockers.some((blocker) => blocker.code === 'WORKTREE_MEASUREMENT_ACTIVE'), true); + assert.throws( + () => convergence.converge({ areaId: context.worker }), + (error) => error.code === 'CONVERGENCE_BLOCKED', + ); + convergence.release({ areaId: context.worker, type: 'measurement' }); + + convergence.hold({ areaId: context.worker, type: 'pin', reason: 'Owner-reviewed historical comparison' }); + assert.equal(convergence.plan({ areaId: context.worker }).blockers + .some((blocker) => blocker.code === 'WORKTREE_PIN_ACTIVE'), true); + convergence.release({ areaId: context.worker, type: 'pin' }); + + writeFileSync(join(context.worktree.path, 'uncommitted.txt'), 'do not merge around this\n'); + assert.equal(convergence.plan({ areaId: context.worker }).blockers + .some((blocker) => blocker.code === 'WORKTREE_DIRTY'), true); + unlinkSync(join(context.worktree.path, 'uncommitted.txt')); + + const mergeHead = execFileSync('git', ['-C', context.worktree.path, 'rev-parse', '--git-path', 'MERGE_HEAD'], { encoding: 'utf8' }).trim(); + writeFileSync(mergeHead, `${execFileSync('git', ['-C', context.root, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim()}\n`); + assert.equal(convergence.plan({ areaId: context.worker }).blockers + .some((blocker) => blocker.code === 'GIT_OPERATION_ACTIVE'), true); + unlinkSync(mergeHead); + + const result = convergence.converge({ areaId: context.worker }); + assert.equal(result.changed, true); + assert.equal(result.mutationPerformed, true); + assert.equal(readFileSync(join(context.worktree.path, 'main-only.js'), 'utf8'), 'export const canonical = true;\n'); + assert.equal(convergence.plan({ areaId: context.worker }).upToDate, true); + control.close(); +});