fix(goal): quiet waits and unify fleet NG ownership (#56, #57, #58)

scripts/test-fleet-goal-native.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""Native Pi discovery controls using owned source copies and real guard modules.
+
+No provider, session, private goal read or live mutation. The optional first
+argument is the installed brain root supplying wrapper/core/unslop modules.
+"""
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+
+repo = Path(__file__).resolve().parent.parent
+brain = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.home() / '.mosaic'
+with tempfile.TemporaryDirectory(prefix='goal58-native-') as temp:
+    root = Path(temp)
+    for name in ['goal', 'mosaic-core']:
+        shutil.copytree(repo / 'extensions' / name, root / '.pi/extensions' / name)
+    legacy = root / 'fleet/extensions/goal'
+    shutil.copytree(repo / 'extensions/goal', legacy)
+    for name in ['wrapper-guard', 'mosaic-core']:
+        (root / 'fleet/extensions' / name).symlink_to(brain / 'fleet/extensions' / name)
+    (root / 'tools').mkdir()
+    (root / 'tools/unslop-hook').symlink_to(brain / 'tools/unslop-hook')
+    (root / '.goal-discovery-owned-fixture').write_text('synthetic goal58 fixture')
+    settings = root / 'settings.json'
+    def check(cwd, mode, guard='wrapper-guard'):
+        settings.write_text(json.dumps({'extensions': [str(legacy), str(root / 'fleet/extensions' / guard)]}))
+        subprocess.run(['node', str(repo / 'scripts/test-goal-discovery.mjs'),
+                        str(cwd), str(settings), mode, str(root)], check=True, timeout=30,
+                       env={**os.environ, 'MOSAIC_GOAL_DISCOVERY_FIXTURE_ROOT': str(root)})
+    check(root, 'conflict')
+    legacy.rename(root / 'old-goal-source')
+    legacy.symlink_to(root / '.pi/extensions/goal')
+    check(root, 'unified')
+    check(root, 'unified', 'mosaic-core')
+    for relative in ['fleet/agents/topher', 'fleet/roles/interact']:
+        cwd = root / relative
+        (cwd / '.pi').mkdir(parents=True)
+        (cwd / '.pi/extensions').symlink_to(root / '.pi/extensions')
+        check(cwd, 'unified')
+    velma = root / 'fleet/agents/velma'
+    (velma / '.pi/extensions').mkdir(parents=True)
+    (velma / '.pi/extensions/goal').symlink_to(root / '.pi/extensions/goal')
+    shutil.copytree(repo / 'extensions/mosaic-core', velma / '.pi/extensions/mosaic-core')
+    check(velma, 'unified')
+    isolated = root / 'isolated'
+    isolated.mkdir()
+    check(isolated, 'unified')
+print('PASS native negative control and six unified discovery cases; no live source/state changes')

scripts/test-fleet-goal-ownership.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""Synthetic transactional controls for the #58 fleet source-alias repair."""
+import importlib.util
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+spec = importlib.util.spec_from_file_location('unify', Path(__file__).with_name('unify-fleet-goal.py'))
+u = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(u)
+
+
+class OwnershipTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(prefix='goal58-transaction-')
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        for p in (u.SHARED, *u.ALIASES):
+            directory = self.root / p
+            directory.mkdir(parents=True)
+            (directory / 'index.ts').write_text('// fixture ' + p)
+        launcher = self.root / u.LAUNCHER
+        launcher.parent.mkdir(parents=True)
+        before = (b'#!/bin/bash\nARGS=()\n'
+                  b'# Issue #57: Resume uses the operator-selected shared NG goal\n'
+                  b'if false; then :; fi\nfor f in "${PROMPT_FILES[@]}"; do :; done\n')
+        after = b'#!/bin/bash\nARGS=()\nfor f in "${PROMPT_FILES[@]}"; do :; done\n'
+        launcher.write_bytes(before)
+        launcher.chmod(0o755)
+        self.addCleanup(patch.stopall)
+        patch.object(u, 'HOTFIX_SHA', u.sha(before)).start()
+        patch.object(u, 'ORIGINAL_SHA', u.sha(after)).start()
+        self.settings = self.root / 'fleet/roles/code/.pi/agent/settings.json'
+        self.settings.parent.mkdir(parents=True)
+        self.settings.write_text(json.dumps({'extensions': ['~/.mosaic/fleet/extensions/goal'], 'packages': []}))
+        link = self.root / 'fleet/agents/joe/.pi/agent/settings.json'
+        link.parent.mkdir(parents=True)
+        link.symlink_to(self.settings)
+        template = self.root / 'fleet/templates/pi-settings.json.template'
+        template.parent.mkdir(parents=True)
+        template.write_text('{"extensions":["~/.mosaic/fleet/extensions/goal"]}')
+        self.env_file = self.root / 'fleet/agents/joe/launch.env'
+        self.env_file.write_text('# synthetic content must not be read\n')
+        self.states = [link.parent / 'goal-state.json', self.root / '.pi/state/goal/goal-state.fixture.json']
+        for state in self.states:
+            state.parent.mkdir(parents=True, exist_ok=True)
+            state.write_text('owned synthetic state sentinel')
+        for p in u.DEPENDENCIES:
+            target = self.root / p
+            if target.suffix == '.ts':
+                target.parent.mkdir(parents=True, exist_ok=True)
+                target.write_text('// synthetic guard')
+            else:
+                target.mkdir(parents=True, exist_ok=True)
+                (target / 'module.ts').write_text('// synthetic core')
+        self.plan = u.prepare(self.root)
+        self.raw = json.dumps(self.plan).encode()
+
+    def test_apply_restore_and_state_preservation(self):
+        backup = u.apply(self.plan, self.raw)
+        u.verify_after(self.plan)
+        self.assertEqual((backup / 'planned.json').read_bytes(), self.raw)
+        for state in self.states:
+            self.assertEqual(state.read_text(), 'owned synthetic state sentinel')
+        u.recover(self.plan, backup)
+        u.verify_before(self.plan)
+
+    def test_partial_failure_reverses_exchange(self):
+        original = u.exchange
+        calls = 0
+        def exchange(a, b):
+            nonlocal calls
+            calls += 1
+            if calls == 2:
+                raise OSError('injected second exchange failure')
+            original(a, b)
+        with patch.object(u, 'exchange', exchange):
+            with self.assertRaisesRegex(OSError, 'injected'):
+                u.apply(self.plan, self.raw)
+        u.verify_before(self.plan)
+        self.assertEqual(calls, 3)
+
+    def test_receipt_recovers_interrupted_partial_deployment(self):
+        backup = u.apply(self.plan, self.raw)
+        # Simulate interruption after aliases changed, before launcher exchange.
+        u.exchange(self.root / u.LAUNCHER, u.stage_path(backup, u.LAUNCHER))
+        u.recover(self.plan, backup)
+        u.verify_before(self.plan)
+
+    def test_settings_drift_refuses(self):
+        self.settings.write_text('{}')
+        with self.assertRaisesRegex(RuntimeError, 'inventory drift'):
+            u.apply(self.plan, self.raw)
+        self.assertFalse((self.root / u.ALIASES[0]).is_symlink())
+
+    def test_environment_contents_are_never_read(self):
+        original = Path.read_bytes
+        def read_bytes(path):
+            if path.name == 'launch.env':
+                self.fail('environment contents read')
+            return original(path)
+        with patch.object(Path, 'read_bytes', read_bytes):
+            u.verify_before(self.plan)
+
+    def test_environment_metadata_drift_refuses(self):
+        self.env_file.write_text('# changed synthetic environment\n')
+        with self.assertRaisesRegex(RuntimeError, 'inventory drift'):
+            u.apply(self.plan, self.raw)
+
+    def test_shared_source_drift_refuses(self):
+        (self.root / u.SHARED / 'index.ts').write_text('drift')
+        with self.assertRaisesRegex(RuntimeError, 'shared source drift'):
+            u.apply(self.plan, self.raw)
+
+    def test_dependency_drift_refuses(self):
+        (self.root / u.DEPENDENCIES[0] / 'module.ts').write_text('drift')
+        with self.assertRaisesRegex(RuntimeError, 'dependency drift'):
+            u.apply(self.plan, self.raw)
+
+    def test_target_drift_refuses(self):
+        (self.root / u.ALIASES[0] / 'index.ts').write_text('drift')
+        with self.assertRaisesRegex(RuntimeError, 'before drift'):
+            u.apply(self.plan, self.raw)
+
+    def test_mode_drift_refuses(self):
+        (self.root / u.ALIASES[0] / 'index.ts').chmod(0o700)
+        with self.assertRaisesRegex(RuntimeError, 'before drift'):
+            u.verify_before(self.plan)
+
+    def test_nested_symlink_refuses(self):
+        (self.root / u.ALIASES[0] / 'link').symlink_to(self.settings)
+        with self.assertRaisesRegex(RuntimeError, 'symlink/special'):
+            u.prepare(self.root)
+
+    def test_special_file_refuses(self):
+        os.mkfifo(self.root / u.ALIASES[0] / 'fifo')
+        with self.assertRaisesRegex(RuntimeError, 'symlink/special'):
+            u.prepare(self.root)
+
+    def test_repeat_apply_refuses(self):
+        u.apply(self.plan, self.raw)
+        with self.assertRaisesRegex(RuntimeError, 'non-canonical source'):
+            u.apply(self.plan, self.raw)
+
+    def test_rollback_validates_all_pairs_before_effect(self):
+        backup = u.apply(self.plan, self.raw)
+        (u.stage_path(backup, u.LAUNCHER)).write_text('backup drift')
+        with self.assertRaisesRegex(RuntimeError, 'rollback backup drift'):
+            u.recover(self.plan, backup)
+        u.verify_after(self.plan)
+
+    def test_rollback_refuses_alias_drift(self):
+        backup = u.apply(self.plan, self.raw)
+        target = self.root / u.ALIASES[1]
+        target.unlink()
+        target.symlink_to(self.root / 'unreviewed')
+        with self.assertRaises(RuntimeError):
+            u.recover(self.plan, backup)
+        self.assertTrue(u.is_alias(self.root / u.ALIASES[0], self.root / u.SHARED))
+
+    def test_concurrent_deployment_lock_refuses(self):
+        with u.deployment_lock(self.plan):
+            with self.assertRaises(BlockingIOError):
+                with u.deployment_lock(self.plan):
+                    self.fail('second writer acquired the lock')
+
+    def test_unknown_launcher_refuses(self):
+        (self.root / u.LAUNCHER).write_text('# not reviewed\n')
+        with self.assertRaisesRegex(RuntimeError, 'reviewed Resume hotfix'):
+            u.prepare(self.root)
+
+
+if __name__ == '__main__':
+    unittest.main()

scripts/test-goal-discovery.mjs

@@ -0,0 +1,81 @@
+#!/usr/bin/env node
+// No-model native resource-loader check: real project discovery plus seat extensions.
+// Usage: node scripts/test-goal-discovery.mjs <cwd> <settings.json> conflict|explicit|shared|unified [brain]
+// Reads only the extensions list; never copies credentials, starts sessions or runs tools.
+import assert from 'node:assert/strict';
+import { mkdtempSync, readFileSync, realpathSync, rmSync, existsSync, unlinkSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { dirname, join, relative } from 'node:path';
+import { pathToFileURL } from 'node:url';
+import { execFileSync } from 'node:child_process';
+import { randomUUID } from 'node:crypto';
+
+const [cwd, settingsPath, expected, brain = cwd] = process.argv.slice(2);
+assert.ok(cwd && settingsPath && ['conflict', 'explicit', 'shared', 'unified'].includes(expected));
+const wantsShared = expected === 'shared' || expected === 'unified';
+const extensions = JSON.parse(readFileSync(settingsPath, 'utf8')).extensions;
+assert.ok(Array.isArray(extensions) && extensions.every(x => typeof x === 'string'));
+const agentDir = mkdtempSync(join(tmpdir(), 'goal-discovery-'));
+process.env.PI_CODING_AGENT_DIR = agentDir;
+process.env.MOSAIC_LAUNCH_INCARNATION = 'discovery-test-' + randomUUID();
+process.env.PI_OFFLINE = '1';
+// Loading the shared extension must not quarantine any operator-owned legacy state.
+for (const dir of new Set([cwd, brain])) {
+  assert.equal(existsSync(join(dir, '.pi/state/goal/goal-state.json')), false);
+}
+try {
+  const binary = realpathSync(execFileSync('which', ['pi'], { encoding: 'utf8' }).trim());
+  const { DefaultResourceLoader, SettingsManager } = await import(pathToFileURL(join(dirname(binary), 'index.js')).href);
+  const settingsManager = SettingsManager.inMemory({ extensions });
+  settingsManager.setProjectTrusted(true);
+  const loader = new DefaultResourceLoader({ cwd, agentDir, settingsManager,
+    noExtensions: expected === 'explicit' || expected === 'shared',
+    additionalExtensionPaths: [...(expected === 'shared'
+      ? [join(brain, '.pi/extensions/goal'), join(brain, 'fleet/extensions/wrapper-guard')]
+      : expected === 'explicit' ? extensions : []), join(brain, 'tools/unslop-hook/extension.ts')],
+    noSkills: true, noPromptTemplates: true, noThemes: true, noContextFiles: true });
+  await loader.reload();
+  const loaded = loader.getExtensions();
+  if (expected === 'conflict') {
+    assert.ok(loaded.errors.some(e => e.error.includes('Tool "goal_report" conflicts with')), JSON.stringify(loaded.errors));
+    console.log('PASS negative control: combined discovery reproduces goal_report conflict');
+  } else {
+    assert.deepEqual(loaded.errors, []);
+    const owners = loaded.extensions.filter(e => e.tools.has('goal_report'));
+    assert.equal(owners.length, 1);
+    assert.equal(realpathSync(owners[0].path.replace(/\/index\.ts$/, '')),
+      realpathSync(join(brain, wantsShared ? '.pi/extensions/goal' : 'fleet/extensions/goal')));
+    const guardNames = extensions.filter(p => /\/(wrapper-guard|mosaic-core)$/.test(p)).map(p => p.split('/').at(-1));
+    assert.ok(guardNames.length, 'test must include the configured enforcement extension');
+    for (const name of guardNames) {
+      const guards = loaded.extensions.filter(e => e.path.replace(/\/index\.ts$/, '').endsWith('/' + name));
+      assert.equal(guards.length, 1, `${name} must remain loaded`);
+      assert.ok(guards[0].handlers.get('tool_call')?.length, `${name} interception must remain registered`);
+    }
+    assert.equal(owners[0].commands.has('goal'), true);
+    assert.equal(loaded.extensions.filter(e => e.path.endsWith('/unslop-hook/extension.ts')).length, 1,
+      'launcher unslop extension must remain loaded');
+    if (wantsShared) {
+      assert.ok(owners[0].shortcuts.has('alt+g'), 'NG full recall shortcut must be registered');
+    }
+    // Only the synthetic native suite opts into an owned fixture write. Live checks
+    // merely load registrations and never run a goal command or read private state.
+    if (process.env.MOSAIC_GOAL_DISCOVERY_FIXTURE_ROOT === brain) {
+      assert.ok(brain.startsWith(join(tmpdir(), 'goal58-native-')));
+      assert.equal(readFileSync(join(brain, '.goal-discovery-owned-fixture'), 'utf8'), 'synthetic goal58 fixture');
+      const filename = `goal-state.${process.env.MOSAIC_LAUNCH_INCARNATION}.json`;
+      const possible = [...new Set([join(brain, '.pi/state/goal', filename),
+        join(brain, 'fleet/state/goal', filename), join(cwd, '.pi/state/goal', filename)])];
+      assert.ok(possible.every(p => !existsSync(p)), 'fixture identity must be new');
+      await owners[0].commands.get('goal').handler('owned goal58 fixture', { mode: 'print', hasUI: false });
+      const written = possible.filter(existsSync);
+      assert.equal(written.length, 1, 'one incarnation state store per selected entrypoint');
+      assert.equal(JSON.parse(readFileSync(written[0], 'utf8')).text, 'owned goal58 fixture');
+      console.log(`PASS owned fixture state path: ${relative(brain, dirname(written[0]))}`);
+      unlinkSync(written[0]);
+    }
+    console.log(`PASS ${expected} discovery: exactly one ${wantsShared ? 'shared NG' : 'legacy'} goal_report and /goal; ${guardNames.join(', ')} interception and unslop retained`);
+  }
+} finally {
+  rmSync(agentDir, { recursive: true, force: true });
+}

scripts/test-goal-native.py

@@ -62,7 +62,7 @@ with tempfile.TemporaryDirectory(prefix="ng-native-pi-") as home:
         rpc.terminate()
         rpc.communicate(timeout=10)
 
-    for status, label, width, no_color in [("paused", "Paused", 120, False), ("blocked", "Blocked", 45, False), ("none", "Complete", 120, True), ("active", "Waiting", 45, False)]:
+    for status, label, width, no_color, timed in [("paused", "Paused", 120, False, False), ("blocked", "Blocked", 45, False, False), ("none", "Complete", 120, True, False), ("active", "Waiting", 45, False, True), ("active", "Waiting", 120, False, False)]:
         incarnation = "ng-native-test-" + uuid.uuid4().hex
         statefile = ROOT / ".pi/state/goal" / f"goal-state.{incarnation}.json"
         statefile.parent.mkdir(parents=True, exist_ok=True)
@@ -70,7 +70,10 @@ with tempfile.TemporaryDirectory(prefix="ng-native-pi-") as home:
         if status == "none":
             state.update(text="", lastOutcome=dict(text=TEXT, status="complete", evidence="native fixture", at="2026-09-06T00:00:00.000Z"))
         if status == "active":
-            state.update(waitTimeoutSeconds=3600, waitWakeUsed=False, activeWait=dict(owner="native fixture", nextCheck="manual acceptance", deadlineAt=int(time.time() * 1000) + 3600000, wakeSent=False))
+            state.update(activeWait=dict(owner="native fixture", nextCheck="manual acceptance"))
+            if timed:
+                state.update(waitTimeoutSeconds=3600, waitWakeUsed=False)
+                state["activeWait"].update(deadlineAt=int(time.time() * 1000) + 3600000, wakeSent=False)
         statefile.write_text(json.dumps(state))
         tty_env = {**env, "MOSAIC_LAUNCH_INCARNATION": incarnation}
         tty_env.pop("NO_COLOR", None)
@@ -98,7 +101,10 @@ with tempfile.TemporaryDirectory(prefix="ng-native-pi-") as home:
             Path("/tmp/ng-goal-native-entered.log").write_bytes(entered)
             os.write(master, b"\x1b[13u")
             receive(master, "END-OF-GOAL")
-            print(f"PASS native {label}, width={width}, NO_COLOR={no_color}: footer + /goal + Alt+G", flush=True)
+            if status == "active":
+                current = json.loads(statefile.read_text())
+                assert current == state, "waiting startup/recall must not inject checks or mutate the fixture"
+            print(f"PASS native {label}, width={width}, NO_COLOR={no_color}, timed={timed}: footer + /goal + Alt+G; zero wait checks", flush=True)
         finally:
             os.killpg(child.pid, signal.SIGTERM)
             child.wait(timeout=10)

scripts/unify-fleet-goal.py

@@ -0,0 +1,265 @@
+#!/usr/bin/env python3
+"""#58 local fleet repair. Default preflight; explicit pinned apply or rollback.
+
+Only two source directories and the Resume-hotfix launcher may change. Settings,
+credentials, state, sessions and enforcement extensions are never written.
+Linux renameat2 exchanges keep originals at the private backup paths, including
+when interrupted. A receipt can recover a partially applied transaction.
+"""
+import argparse
+from contextlib import contextmanager
+import ctypes
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import stat
+import uuid
+
+SHARED = '.pi/extensions/goal'
+ALIASES = ('fleet/extensions/goal', 'fleet/agents/velma/.pi/extensions/goal')
+LAUNCHER = 'fleet/bin/launch-seat.sh'
+DEPENDENCIES = ('.pi/extensions/mosaic-core', 'fleet/extensions/mosaic-core',
+                'fleet/agents/velma/.pi/extensions/mosaic-core',
+                'fleet/extensions/wrapper-guard/index.ts', 'tools/unslop-hook/extension.ts')
+HOTFIX_SHA = 'dd9e5ece5f1a86cc286027668560198bb7a44d2ab891f7b1ab4f89d7379ca0e6'
+ORIGINAL_SHA = '9352feed0acf9d449c26c0556ba00aba1c65ded43d376930a39ff6b5cee21986'
+
+
+def require(condition, message):
+    if not condition:
+        raise RuntimeError(message)
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def snapshot(path):
+    """Ordinary source files only. Do not call this on goal-state directories."""
+    require(path.resolve() == path, f'non-canonical source path: {path}')
+    result = {}
+    paths = [path] + (sorted(path.rglob('*')) if path.is_dir() else [])
+    for p in paths:
+        s = p.lstat()
+        kind = 'dir' if stat.S_ISDIR(s.st_mode) else 'file' if stat.S_ISREG(s.st_mode) else None
+        require(kind is not None, f'symlink/special source file refused: {p}')
+        require(s.st_uid == os.getuid(), f'foreign-owned source: {p}')
+        result[str(p.relative_to(path))] = {'type': kind, 'mode': stat.S_IMODE(s.st_mode)}
+        if kind == 'file':
+            result[str(p.relative_to(path))]['sha256'] = sha(p.read_bytes())
+    return result
+
+
+def inventory(root):
+    """Pin configuration without copying its contents or reading auth/state."""
+    paths = []
+    for group in ('agents', 'roles'):
+        paths += list((root / 'fleet' / group).glob('*/.pi/agent/settings.json'))
+    paths += [root / 'fleet/templates/pi-settings.json.template']
+    result = {}
+    for p in sorted(paths):
+        data = p.read_bytes()
+        entry = {'sha256': sha(data), 'resolved': str(p.resolve()),
+                 'link': os.readlink(p) if p.is_symlink() else None}
+        if p.name == 'settings.json':
+            settings = json.loads(data)
+            entry['extensions'] = settings.get('extensions', [])
+            require(not settings.get('packages'), f'unreviewed package source: {p}')
+        result[str(p.relative_to(root))] = entry
+    # Environment files may carry credentials. Pin filesystem identity/change
+    # metadata only; never read or copy their contents for this source repair.
+    for p in sorted((root / 'fleet/agents').glob('*/launch.env')):
+        s = p.stat()
+        result[str(p.relative_to(root))] = {
+            'resolved': str(p.resolve()), 'link': os.readlink(p) if p.is_symlink() else None,
+            'device': s.st_dev, 'inode': s.st_ino, 'size': s.st_size,
+            'mtime_ns': s.st_mtime_ns, 'ctime_ns': s.st_ctime_ns,
+        }
+    return result
+
+
+def restored_launcher(data):
+    require(sha(data) == HOTFIX_SHA, 'launcher is not the reviewed Resume hotfix')
+    start = data.index(b'# Issue #57: Resume uses the operator-selected shared NG goal')
+    end = data.index(b'for f in "${PROMPT_FILES[@]}"', start)
+    result = data[:start] + data[end:]
+    require(sha(result) == ORIGINAL_SHA, 'launcher restoration differs from original')
+    return result
+
+
+def prepare(root):
+    require(root.is_dir() and root.resolve() == root, 'root must be a canonical directory')
+    after = restored_launcher((root / LAUNCHER).read_bytes())
+    before = {p: snapshot(root / p) for p in (*ALIASES, LAUNCHER)}
+    launcher_after = {'.': {**before[LAUNCHER]['.'], 'sha256': sha(after)}}
+    return {'version': 1, 'root': str(root), 'shared': snapshot(root / SHARED),
+            'dependencies': {p: snapshot(root / p) for p in DEPENDENCIES},
+            'before': before, 'launcherAfter': launcher_after, 'inventory': inventory(root)}
+
+
+def validate_plan(plan):
+    require(plan['version'] == 1, 'unsupported plan')
+    root = Path(plan['root'])
+    require(root.is_absolute() and root.resolve() == root, 'invalid root')
+    require(set(plan['before']) == set((*ALIASES, LAUNCHER)), 'invalid mutation scope')
+    require(plan['before'][LAUNCHER]['.']['sha256'] == HOTFIX_SHA, 'unreviewed launcher pin')
+    require(plan['launcherAfter']['.']['sha256'] == ORIGINAL_SHA, 'unreviewed restoration pin')
+    return root
+
+
+def verify_common(plan):
+    root = validate_plan(plan)
+    require(snapshot(root / SHARED) == plan['shared'], 'shared source drift')
+    require(set(plan['dependencies']) == set(DEPENDENCIES), 'invalid dependency scope')
+    for p in DEPENDENCIES:
+        require(snapshot(root / p) == plan['dependencies'][p], f'dependency drift: {p}')
+    require(inventory(root) == plan['inventory'], 'configuration inventory drift')
+    return root
+
+
+def verify_before(plan):
+    root = verify_common(plan)
+    for p in (*ALIASES, LAUNCHER):
+        require(snapshot(root / p) == plan['before'][p], f'before drift: {p}')
+
+
+def is_alias(path, shared):
+    return path.is_symlink() and os.readlink(path) == str(shared) and path.resolve() == shared
+
+
+def verify_after(plan):
+    root = verify_common(plan)
+    for p in ALIASES:
+        require(is_alias(root / p, root / SHARED), f'alias drift: {p}')
+    require(snapshot(root / LAUNCHER) == plan['launcherAfter'], 'launcher after drift')
+
+
+def exchange(a, b):
+    fn = ctypes.CDLL(None, use_errno=True).renameat2
+    fn.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
+    fn.restype = ctypes.c_int
+    if fn(-100, os.fsencode(a), -100, os.fsencode(b), 2):
+        error = ctypes.get_errno()
+        raise OSError(error, os.strerror(error))
+
+
+def write_once(path, data):
+    with path.open('xb') as f:
+        f.write(data)
+        f.flush()
+        os.fsync(f.fileno())
+
+
+def stage_path(backup, relative):
+    return backup / ('launcher' if relative == LAUNCHER else 'legacy' if relative == ALIASES[0] else 'velma')
+
+
+def recover(plan, backup):
+    """Validate every pair before reverting any. Handles partial exchanges."""
+    root = verify_common(plan)
+    require(backup.parent == root / '.pi/goal-backups' and backup.resolve() == backup,
+            'backup outside approved directory')
+    swapped = []
+    for p in (*ALIASES, LAUNCHER):
+        target, stage = root / p, stage_path(backup, p)
+        if p in ALIASES:
+            target_after = is_alias(target, root / SHARED)
+            stage_after = is_alias(stage, root / SHARED)
+        else:
+            target_after = snapshot(target) == plan['launcherAfter']
+            stage_after = snapshot(stage) == plan['launcherAfter']
+        if target_after:
+            require(snapshot(stage) == plan['before'][p], f'rollback backup drift: {p}')
+            swapped.append(p)
+        else:
+            require(snapshot(target) == plan['before'][p] and stage_after, f'rollback pair drift: {p}')
+    for p in reversed(swapped):
+        exchange(root / p, stage_path(backup, p))
+    verify_before(plan)
+    write_once(backup / ('rollback-' + uuid.uuid4().hex + '.json'),
+               json.dumps({'status': 'rolled back', 'restored': swapped}).encode())
+
+
+def apply(plan, raw):
+    verify_before(plan)
+    root = Path(plan['root'])
+    backup_root = root / '.pi/goal-backups'
+    backup_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    require(backup_root.resolve() == backup_root, 'backup directory alias refused')
+    backup = backup_root / ('goal58-' + uuid.uuid4().hex)
+    backup.mkdir(mode=0o700)
+    for p in ALIASES:
+        stage_path(backup, p).symlink_to(root / SHARED, target_is_directory=True)
+    stage = stage_path(backup, LAUNCHER)
+    write_once(stage, restored_launcher((root / LAUNCHER).read_bytes()))
+    os.chmod(stage, plan['before'][LAUNCHER]['.']['mode'])
+    for p in (*ALIASES, LAUNCHER):
+        require((root / p).lstat().st_dev == stage_path(backup, p).lstat().st_dev, 'cross-device exchange refused')
+    write_once(backup / 'planned.json', raw)
+    print(f'Recovery receipt: {backup}', flush=True)
+    try:
+        verify_before(plan)
+        for p in (*ALIASES, LAUNCHER):
+            require(snapshot(root / p) == plan['before'][p], f'concurrent drift: {p}')
+            exchange(root / p, stage_path(backup, p))
+        verify_after(plan)
+        for p in (*ALIASES, LAUNCHER):
+            require(snapshot(stage_path(backup, p)) == plan['before'][p], f'backup mismatch: {p}')
+        write_once(backup / 'deployed.json', json.dumps({'status': 'deployed', 'planSha256': sha(raw)}).encode())
+    except BaseException:
+        recover(plan, backup)
+        raise
+    return backup
+
+
+@contextmanager
+def deployment_lock(plan):
+    root = verify_common(plan)
+    directory = root / '.pi/goal-backups'
+    directory.mkdir(mode=0o700, parents=True, exist_ok=True)
+    require(directory.resolve() == directory, 'backup directory alias refused')
+    fd = os.open(directory / 'goal58.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        yield
+    finally:
+        os.close(fd)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('mode', nargs='?', default='preflight', choices=['prepare', 'preflight', 'apply', 'verify', 'rollback'])
+    parser.add_argument('--root', type=Path)
+    parser.add_argument('--plan', type=Path, required=True)
+    parser.add_argument('--sha256')
+    parser.add_argument('--backup', type=Path)
+    args = parser.parse_args()
+    if args.mode == 'prepare':
+        require(args.root is not None, '--root required')
+        plan = prepare(args.root)
+        raw = (json.dumps(plan, indent=2, sort_keys=True) + '\n').encode()
+        write_once(args.plan, raw)
+        print('Prepared plan SHA256:', sha(raw))
+        return
+    raw = args.plan.read_bytes()
+    require(sha(raw) == args.sha256, 'reviewed plan SHA256 required')
+    plan = json.loads(raw)
+    if args.mode == 'preflight':
+        verify_before(plan)
+    elif args.mode == 'apply':
+        with deployment_lock(plan):
+            print('Deployed:', apply(plan, raw))
+    elif args.mode == 'verify':
+        verify_after(plan)
+    else:
+        require(args.backup is not None, '--backup required')
+        require((args.backup / 'planned.json').read_bytes() == raw, 'backup receipt mismatch')
+        with deployment_lock(plan):
+            recover(plan, args.backup)
+    print('PASS', args.mode)
+
+
+if __name__ == '__main__':
+    main()