#!/usr/bin/env python3
"""Real pair/worker/systemd effects in a fresh disconnected disposable PID1.

The fixture ELF and exact-document allowlist verifier are explicit doubles.
Crash injection wraps journal/rename only in a separately identified test bundle.
No production root key or full product binary is represented by this test.
"""
import hashlib
import io
import json
import os
from pathlib import Path
import shutil
import socket
import subprocess
import tarfile
import time

assert os.geteuid() == 0 and socket.gethostname() == 'public-ota-worker-fixture'
assert Path('/run/.containerenv').exists()
assert os.environ.get('ARCHY_PUBLIC_WORKER_PID1') == '1'
PACKAGE = Path('/opt/public-fixture')
ROOT = Path('/var/lib/archipelago-public-transactions')
SNAPSHOTS = Path('/var/lib/archipelago-public-updates')
DATA = Path('/var/lib/archipelago')
FRONTEND = Path('/opt/archipelago/web-ui')
BINARY = Path('/usr/local/bin/archipelago')
EVIDENCE = Path('/var/lib/public-worker-evidence')
EVIDENCE.mkdir(mode=0o700)


def run(*args, check=True, **kwargs):
    result = subprocess.run(args, capture_output=True, text=True, timeout=240, **kwargs)
    if check and result.returncode:
        raise AssertionError((args, result.returncode, result.stdout, result.stderr))
    return result


def write(path, content, mode=0o644):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(content.encode() if isinstance(content, str) else content)
    path.chmod(mode)


def wait(predicate, label, seconds=60):
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        result = predicate()
        if result:
            return result
        time.sleep(.05)
    raise AssertionError('timeout: ' + label + '; durable state=' + str(state()))


def state():
    try:
        return json.loads((ROOT / (ROOT / 'current').read_text() / 'state.json').read_bytes())
    except FileNotFoundError:
        return None


def terminal(phase):
    return wait(lambda: (s if (s := state()) and s['phase'] == phase else None), phase)


def manager_pid():
    return int(run('systemctl', 'show', 'archipelago.service', '-p', 'MainPID', '--value').stdout)


def pair():
    return hashlib.sha256(BINARY.read_bytes()).hexdigest(), (FRONTEND / 'index.html').read_bytes()


def preserved():
    return {name: (DATA / name).read_bytes() for name in
            ['wallet-fixture', 'app-data-fixture', 'stopped-apps.json', 'uninstalled-apps.json']}


def assert_preserved():
    assert preserved() == SENTINELS


preparation = (PACKAGE / 'public_update_prepare.py').read_text().rsplit("\nif __name__ == '__main__':", 1)[0]
worker = (PACKAGE / 'public_update_worker.py').read_text()
bundle = (preparation + '\n' + (PACKAGE / 'public_update_runtime.py').read_text()
          + '\n' + (PACKAGE / 'public_update_adopt.py').read_text() + '\n' + worker)
write(PACKAGE / 'owner.py', bundle, 0o444)
# This wrapper is fixture-only. The actual worker logic/files/systemd are real;
# the wrapper merely stops execution at a chosen durable boundary for SIGKILL.
hooks = '''
original_journal = journal
original_replace = os.replace
original_rename = os.rename
original_atomic = atomic
original_boot_recovery = retain_boot_recovery
def pause_at(name):
    fault = Path('/opt/public-fixture/fault')
    if fault.exists() and fault.read_text() == name:
        fault.rename(fault.with_name('fault-consumed'))
        Path('/opt/public-fixture/paused').write_text(name)
        import signal
        os.kill(os.getpid(), signal.SIGSTOP)
def journal(tx, value, phase):
    value = original_journal(tx, value, phase)
    pause_at(phase)
    return value
def replaced(source, target):
    original_replace(source, target)
    if Path(target) == FRONTEND: pause_at('frontend-installed')
    if Path(target) == BINARY: pause_at('binary-installed')
os.replace = replaced
def renamed(source, target):
    original_rename(source, target)
    if Path(target).name.startswith('.public-runtime-retired-'):
        pause_at('runtime-retired')
    if Path(target) == Path('/opt/archipelago/scripts'):
        pause_at('runtime-scripts-installed')
os.rename = renamed
def atomic(path, data, mode=0o600):
    original_atomic(path, data, mode)
    if path == TRANSACTIONS / 'current': pause_at('pointer-published')
def retain_boot_recovery(unit):
    original_boot_recovery(unit)
    pause_at('boot-recovery-linked')
worker_main()
'''
write(PACKAGE / 'fault-owner.py', bundle.rsplit("\nif __name__ == '__main__':", 1)[0] + hooks, 0o444)
download_hook = """
def public_download(url, path, size=None, sha256=None):
    prefix = 'https://source.archipelago-foundation.org/lfg2025/archy/releases/download/'
    assert url.startswith(prefix)
    raw = (Path('/opt/public-fixture/downloads') / url.removeprefix(prefix)).read_bytes()
    assert size is None or len(raw) == size
    assert sha256 is None or hashlib.sha256(raw).hexdigest() == sha256
    write_manifest(path, raw)
"""
prefix = bundle.rsplit("\nif __name__ == '__main__':", 1)[0]
write(PACKAGE / 'adopt-owner.py', prefix + download_hook + '\nworker_main()\n', 0o444)
write(PACKAGE / 'fault-adopt-owner.py', prefix + download_hook + hooks, 0o444)
write(PACKAGE / 'verify.py', '''import hashlib,pathlib,sys
# Exact-document authorization double, not release-root cryptography.
raw=pathlib.Path(sys.argv[1]).read_bytes()
assert hashlib.sha256(raw).hexdigest() in pathlib.Path('/opt/public-fixture/authorized-document').read_text().splitlines()
''', 0o444)
write(PACKAGE / 'frontend-server.py', '''from http.server import HTTPServer,BaseHTTPRequestHandler
from pathlib import Path
class Handler(BaseHTTPRequestHandler):
 def do_GET(self):
  data=Path('/opt/archipelago/web-ui/index.html').read_bytes()
  self.send_response(200);self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
HTTPServer(('127.0.0.1',80),Handler).serve_forever()
''')
run('systemctl', 'stop', 'archipelago.service', check=False)
run('useradd', '--system', '--no-create-home', 'archipelago', check=False)
# Clean only this disposable image's product owner configuration, never host.
for p in ['/etc/archipelago-d09', '/etc/archipelago-c01', '/var/lib/archipelago-d09', '/run/archipelago-d09',
          '/etc/systemd/system/archipelago.service.d']:
    path = Path(p)
    if path.is_symlink():
        path.unlink()
    elif path.exists():
        shutil.rmtree(path)
for p in ['/usr/libexec/archipelago-d09ctl', '/usr/libexec/archipelago-d09-owner',
          '/etc/systemd/system/archipelago-d09-lifecycle.service',
          '/usr/lib/systemd/system/archipelago-d09-lifecycle.service']:
    Path(p).unlink(missing_ok=True)
write('/etc/systemd/system/archipelago.service', '''[Unit]
Description=Public OTA fixture manager (not product)
[Service]
Type=notify
TimeoutStartSec=180
User=archipelago
ExecStart=/usr/local/bin/archipelago
Restart=no
TimeoutStopSec=5
''')
# The fixture must not inherit background product maintenance from its base
# image. Use explicit inert units so the transaction can preserve their states.
for unit in ('archipelago-doctor', 'archipelago-npm-bridge'):
    run('systemctl', 'stop', unit + '.timer', unit + '.service', check=False)
    write('/etc/systemd/system/' + unit + '.service', '[Service]\nType=oneshot\nExecStart=/usr/bin/true\n')
    write('/etc/systemd/system/' + unit + '.timer', '[Timer]\nOnActiveSec=1h\n')
write('/etc/systemd/system/public-fixture-frontend.service', '''[Service]
ExecStart=/usr/bin/python3 -I /opt/public-fixture/frontend-server.py
''')
run('systemctl', 'daemon-reload')
run('systemctl', 'start', 'public-fixture-frontend.service')
# Overlayfs cannot rename a directory inherited from its lower image layer.
# Materialize the disposable baseline in the writable layer, preserving its
# exact contents; installed disk filesystems do not have that fixture boundary.
for directory in ('apps', 'scripts', 'docker'):
    target = Path('/opt/archipelago') / directory
    if target.is_dir():
        saved = EVIDENCE / ('base-image-' + directory)
        shutil.copytree(target, saved)
        shutil.rmtree(target)
        shutil.copytree(saved, target)
DATA.mkdir(parents=True, exist_ok=True)
for name, raw in {'wallet-fixture': b'wallet must remain byte exact', 'app-data-fixture': b'persistent application data',
                  'stopped-apps.json': b'["stopped-app"]', 'uninstalled-apps.json': b'["uninstalled-app"]'}.items():
    write(DATA / name, raw, 0o600)
SENTINELS = preserved()


def reset(case, fail=False, bad_health=False, runtime=False):
    prior = state()
    if prior:
        run('systemctl', 'stop', prior['worker_unit'], check=False)
    run('systemctl', 'stop', 'archipelago.service', check=False)
    dropin = Path('/etc/systemd/system/archipelago.service.d/70-public-application-update.conf')
    for path in [dropin, *(Path('/etc/systemd/system') / (unit + '.d/70-public-application-update.conf')
            for unit in ('archipelago-doctor.timer','archipelago-doctor.service','archipelago-npm-bridge.timer','archipelago-npm-bridge.service'))]:
        path.unlink(missing_ok=True)
    for path in Path('/etc/systemd/system/basic.target.wants').glob('archipelago-public-recovery-*.service'):
        path.unlink()
    if ROOT.exists():
        ROOT.rename(EVIDENCE / ('transactions-before-' + case))
    if SNAPSHOTS.exists():
        SNAPSHOTS.rename(EVIDENCE / ('snapshots-before-' + case))
    if FRONTEND.exists():
        shutil.rmtree(FRONTEND)
    for name in ('fault', 'fault-consumed', 'paused'):
        (PACKAGE / name).unlink(missing_ok=True)
    write(BINARY, (PACKAGE / 'old-backend').read_bytes(), 0o755)
    write(FRONTEND / 'index.html', b'old frontend\n')
    write(FRONTEND / 'aiui/retained.html', b'locally installed UI extra')
    run('chown', '-R', 'archipelago:archipelago', '/opt/archipelago')
    run('systemctl', 'daemon-reload')
    run('systemctl', 'reset-failed', 'archipelago.service', check=False)
    run('systemctl', 'start', 'archipelago.service')
    wait(manager_pid, 'old manager running')
    staging = DATA / 'update-staging'
    if staging.exists():
        shutil.rmtree(staging)
    staging.mkdir()
    write(staging / 'archipelago', (PACKAGE / ('bad-backend' if fail else 'health-backend' if bad_health else 'new-backend')).read_bytes())
    archive_name = 'archipelago-frontend-2.0.4-alpha.tar.gz'
    with tarfile.open(staging / archive_name, 'w:gz') as archive:
        entries = [('index.html', b'new frontend\n'), ('archipelago-runtime/RELEASE-PAYLOAD.json', b'{}')]
        if runtime:
            # Real fixed destination domain, with inert fixture code/units.
            # No product executable/signature claim is made by this matrix.
            scope = {'Path': Path}
            exec((PACKAGE / 'public_update_runtime.py').read_text(), scope)
            for _, relative, kind, _ in scope['RUNTIME_OBJECTS'].values():
                if not relative:
                    continue
                name = 'archipelago-runtime/' + relative
                if kind == 'directory':
                    entries.append((name + '/fixture.txt', b'new runtime fixture\n'))
                elif relative.endswith('.service'):
                    entries.append((name, b'[Service]\nType=oneshot\nExecStart=/usr/bin/true\n'))
                elif relative.endswith('.timer'):
                    entries.append((name, b'[Timer]\nOnActiveSec=1h\n[Install]\nWantedBy=timers.target\n'))
                else:
                    entries.append((name, b'#!/bin/sh\nexit 0\n'))
            entries.append(('archipelago-runtime/image-recipe/configs/nginx-archipelago.conf', b'fixture template\n'))
        for name, raw in entries:
            member = tarfile.TarInfo(name); member.size = len(raw); member.mode = 0o644
            archive.addfile(member, io.BytesIO(raw))
    components = []
    for name in ['archipelago', archive_name]:
        raw = (staging / name).read_bytes()
        components.append({'name': name, 'new_version': '2.0.4-alpha', 'size_bytes': len(raw),
                           'sha256': hashlib.sha256(raw).hexdigest(),
                           'download_url': 'https://source.archipelago-foundation.org/lfg2025/archy/releases/download/v2.0.4-alpha/' + name})
    document = {'version': '2.0.4-alpha', 'application_update': {'version': 1, 'mode': 'signed-application-only'},
                'signed_by': 'did:key:z6Mkfu5LT8d4DjETtrkATvHh9Dvcbnr7zBCUwfau8Sw7DLWT', 'signature': '0' * 128,
                'components': components}
    if runtime:
        document['application_update'] = {'version': 2, 'mode': 'signed-application-runtime'}
    raw = json.dumps(document)
    write(PACKAGE / 'signed-manifest.json', raw)
    write(PACKAGE / 'authorized-document', hashlib.sha256(raw.encode()).hexdigest())
    ready = json.loads(run('python3', '-I', str(PACKAGE / 'public_update_prepare.py'), input=raw).stdout)
    return Path(ready['path']).name, pair()


def submit(identity, faults=False, check=True):
    return run('python3', '-I', str(PACKAGE / ('fault-owner.py' if faults else 'owner.py')),
               'submit', identity, check=check)


def save_case(name):
    assert_preserved()
    write(EVIDENCE / (name + '.json'), json.dumps({'state': state(), 'pair': [pair()[0], pair()[1].decode()],
                                                 'manager_pid': manager_pid(), 'preservation': 'byte-exact'}))
    print('PASS ' + name, flush=True)


if os.environ.get('ARCHY_PUBLIC_WORKER_RUNTIME_ONLY') != '1':
    identity, old = reset('happy')
    old_pid = manager_pid()
    submit(identity)
    s = terminal('committed')
    assert pair()[1] == b'new frontend\n' and pair()[0] != old[0]
    assert manager_pid() != old_pid
    assert (FRONTEND / 'aiui/retained.html').read_bytes() == b'locally installed UI extra'
    save_case('commit-with-manager-replacement')
    run('python3', '-I', str(PACKAGE / 'owner.py'), 'request-recovery')
    terminal('restored')
    assert pair() == old
    save_case('manual-exact-pair-recovery')

    identity, old = reset('failed-start', fail=True)
    submit(identity)
    s = terminal('restored')
    assert 'failure' in s and pair() == old
    save_case('failed-candidate-start-restores-both')

    identity, old = reset('wrong-health', bad_health=True)
    submit(identity)
    s = terminal('restored')
    assert 'version differs' in s['failure'] and pair() == old
    save_case('post-restart-verification-failure-restores-both')

    identity, old = reset('older-bootstrap')
    run('python3', '-I', str(PACKAGE / 'owner.py'), 'apply', input=(PACKAGE / 'signed-manifest.json').read_text())
    terminal('committed')
    save_case('standalone-bootstrap-without-new-backend-cli')

    for boundary in ['preparing', 'prepared', 'installing', 'frontend-installed', 'binary-installed', 'verifying']:
        identity, old = reset('kill-' + boundary)
        write(PACKAGE / 'fault', boundary)
        submit(identity, faults=True)
        wait(lambda: (PACKAGE / 'paused').exists(), 'worker paused at ' + boundary)
        s = state()
        run('systemctl', 'kill', '--kill-whom=main', '--signal=KILL', s['worker_unit'])
        terminal('aborted' if boundary == 'preparing' else 'restored')
        assert pair() == old
        wait(manager_pid, 'manager recovered')
        save_case('worker-death-' + boundary)

    # Boot recovery: stop the paused worker without letting its Restart= run; stop
    # the oneshot boot dependency, then request manager start through that dependency.
    # This exercises PID1 boot ordering with persistent on-disk state, not a fake
    # recovery receipt. A full container reboot is tested by the outer runner too.
    identity, old = reset('boot-order')
    write(PACKAGE / 'fault', 'frontend-installed')
    submit(identity, faults=True)
    wait(lambda: (PACKAGE / 'paused').exists(), 'mixed pair before boot recovery')
    s = state()
    run('systemctl', 'stop', s['worker_unit'])
    recovery_unit = 'archipelago-public-recovery-' + s['id'] + '.service'
    run('systemctl', 'stop', recovery_unit)
    run('systemctl', 'start', 'archipelago.service')
    terminal('restored')
    assert pair() == old
    save_case('pid1-recovery-before-manager-start')

    for index, marker in enumerate(['/etc/archipelago-d09', '/etc/archipelago-c01', '/var/lib/archipelago-d09', '/run/archipelago-d09',
                                  '/usr/libexec/archipelago-d09ctl',
                                  '/etc/systemd/system/archipelago-d09-lifecycle.service']):
        identity, old = reset('refuse-' + str(index))
        Path(marker).parent.mkdir(parents=True, exist_ok=True)
        Path(marker).symlink_to('/nonexistent-owner')
        try:
            result = submit(identity, check=False)
            assert result.returncode != 0 and pair() == old and state() is None
            assert_preserved()
        finally:
            Path(marker).unlink()
    print('PASS all partial/dangling protected owners refuse activation', flush=True)

    identity, old = reset('concurrent')
    write(PACKAGE / 'fault', 'prepared')
    submit(identity, faults=True)
    wait(lambda: (PACKAGE / 'paused').exists(), 'concurrent transaction fence')
    assert submit(identity, check=False).returncode != 0 and pair() == old
    run('systemctl', 'kill', '--kill-whom=main', '--signal=KILL', state()['worker_unit'])
    terminal('restored')
    save_case('concurrent-submission-refused')

    identity, old = reset('recovery-death')
    submit(identity, faults=True)
    terminal('committed')
    write(PACKAGE / 'fault', 'recovering')
    run('python3', '-I', str(PACKAGE / 'owner.py'), 'request-recovery')
    wait(lambda: (PACKAGE / 'paused').exists(), 'recovery journal boundary')
    run('systemctl', 'kill', '--kill-whom=main', '--signal=KILL', state()['worker_unit'])
    terminal('restored')
    assert pair() == old
    save_case('interrupted-recovery-resumes')

    identity, old = reset('late-owner')
    write(PACKAGE / 'fault', 'prepared')
    submit(identity, faults=True)
    wait(lambda: (PACKAGE / 'paused').exists(), 'late owner handoff')
    marker = Path('/etc/archipelago-d09')
    marker.mkdir()
    run('systemctl', 'kill', '--kill-whom=main', '--signal=KILL', state()['worker_unit'])
    time.sleep(4)
    assert pair() == old and state()['phase'] == 'prepared'
    run('systemctl', 'stop', state()['worker_unit'])
    marker.rmdir()
    assert_preserved()
    print('PASS protected owner appearing mid-transaction prevents fallback recovery', flush=True)

    identity, old = reset('tampered-snapshot')
    (SNAPSHOTS / identity / 'archipelago').write_bytes(b'corrupt artifact')
    assert submit(identity, check=False).returncode != 0 and pair() == old and state() is None
    assert_preserved()
    print('PASS tampered prepared artifact refused before manager stop', flush=True)

    identity, old = reset('preimage-corruption')
    submit(identity)
    s = terminal('committed')
    tx = ROOT / s['id']
    (tx / 'old-binary').write_bytes(b'root injected corruption')
    run('python3', '-I', str(PACKAGE / 'owner.py'), 'request-recovery')
    wait(lambda: state()['phase'] == 'recovering', 'recovery refusal')
    time.sleep(3)
    assert pair()[1] == b'new frontend\n'
    assert pair()[0] != old[0]
    assert_preserved()
    print('PASS corrupt protected preimage refuses without installing corrupted bytes', flush=True)

    identity, old = reset('foreign-code')
    submit(identity)
    terminal('committed')
    write(FRONTEND / 'index.html', b'foreign root edit')
    run('python3', '-I', str(PACKAGE / 'owner.py'), 'request-recovery')
    wait(lambda: state()['phase'] == 'recovering', 'foreign edit refusal')
    time.sleep(3)
    assert (FRONTEND / 'index.html').read_bytes() == b'foreign root edit'
    assert_preserved()
    print('PASS foreign installed code edit is retained, not overwritten', flush=True)

# A killed submission must have durable boot recovery even before all service
# dependencies or the authoritative pointer are published.
for boundary in ('boot-recovery-linked', 'pointer-published'):
    identity, old = reset('schedule-' + boundary, runtime=True)
    write(PACKAGE / 'fault', boundary)
    with (EVIDENCE / ('schedule-' + boundary + '.log')).open('w') as output:
        caller = subprocess.Popen(['python3', '-I', str(PACKAGE / 'fault-owner.py'), 'submit', identity],
                                  stdout=output, stderr=subprocess.STDOUT)
        wait(lambda: (PACKAGE / 'paused').exists(), boundary + ' caller interruption')
        caller.kill(); caller.wait(timeout=10)
    transactions = [p for p in ROOT.iterdir() if p.is_dir() and len(p.name) == 32]
    assert len(transactions) == 1
    selected = json.loads((transactions[0] / 'state.json').read_bytes())
    recovery_unit = 'archipelago-public-recovery-' + selected['id'] + '.service'
    assert (Path('/etc/systemd/system/basic.target.wants') / recovery_unit).is_symlink()
    # A new PID1 loads the persisted unit, not the pre-interruption arm command.
    run('systemctl', 'daemon-reload')
    run('systemctl', 'restart', recovery_unit)
    terminal('aborted')
    assert pair() == old
    save_case('submission-death-' + boundary)

# Runtime ownership is tested with the same real PID1/journal worker, including
# crashes between directory renames. The unchanged pair-only matrix above also
# remains a distinct result rather than being called a product upgrade.
for boundary in (None, 'runtime-retired', 'runtime-scripts-installed'):
    label = 'runtime-' + (boundary or 'commit')
    identity, old = reset(label, runtime=True)
    routing = Path('/etc/nginx/sites-available/archipelago')
    routing.parent.mkdir(parents=True, exist_ok=True)
    # The fixture frontend owns port80; nginx syntax validation still uses an
    # actual self-contained configuration, without starting a second server.
    write('/etc/nginx/nginx.conf', 'events {}\nhttp { include /etc/nginx/sites-enabled/*; }\n')
    enabled = Path('/etc/nginx/sites-enabled'); enabled.mkdir(parents=True, exist_ok=True)
    for entry in enabled.iterdir():
        entry.unlink()
    write(routing, 'server { listen 8088; location /operator { return 200 "retained"; } }\n')
    (enabled / 'archipelago').symlink_to(routing)
    routing_before = routing.read_bytes()
    if boundary:
        write(PACKAGE / 'fault', boundary)
    submit(identity, faults=bool(boundary))
    if boundary:
        wait(lambda: (PACKAGE / 'paused').exists(), label + ' journaled interruption')
        run('systemctl', 'kill', '--kill-whom=main', '--signal=KILL', state()['worker_unit'])
    else:
        s = terminal('committed')
        assert s['runtime'] and Path('/etc/archipelago/public-update-runtime.json').exists()
        assert (FRONTEND / 'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf').read_bytes() == routing_before
        run('python3', '-I', str(PACKAGE / 'owner.py'), 'request-recovery')
    terminal('restored')
    assert pair() == old and routing.read_bytes() == routing_before
    assert not Path('/etc/archipelago/public-update-runtime.json').exists()
    save_case(label)

# Reproduce the historical installer, including staging deletion and the
# successor waiting in Type=notify activating. Signature/network are explicitly
# fixture doubles; actual files, service ownership and recovery are real.
for boundary in (None, 'preparing', 'prepared', 'runtime-scripts-installed'):
    label = 'legacy-adoption-' + (boundary or 'commit')
    identity, old_pair = reset(label, runtime=True)
    staging = DATA / 'update-staging'
    published = PACKAGE / 'downloads'
    if published.exists():
        published.rename(EVIDENCE / ('downloads-before-' + label))
    current_dir = published / 'v2.0.4-alpha'; current_dir.mkdir(parents=True)
    old_dir = published / 'v1.9.1-alpha'; old_dir.mkdir()
    document = json.loads((PACKAGE / 'signed-manifest.json').read_bytes())
    for component in document['components']:
        shutil.copyfile(staging / component['name'], current_dir / component['name'])
    shutil.copyfile(PACKAGE / 'signed-manifest.json', current_dir / 'manifest.json')
    shutil.copyfile(BINARY, old_dir / 'archipelago')
    with tarfile.open(old_dir / 'archipelago-frontend-1.9.1-alpha.tar.gz', 'w:gz') as archive:
        archive.add(FRONTEND, arcname='.')
    old_doc = {'version': '1.9.1-alpha', 'signed_by': document['signed_by'],
               'signature': '0' * 128, 'components': []}
    for name in ('archipelago', 'archipelago-frontend-1.9.1-alpha.tar.gz'):
        raw = (old_dir / name).read_bytes()
        old_doc['components'].append({'name': name, 'new_version': '1.9.1-alpha',
            'size_bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest(),
            'download_url': 'https://source.archipelago-foundation.org/lfg2025/archy/releases/download/v1.9.1-alpha/' + name})
    old_raw = json.dumps(old_doc).encode()
    write(current_dir / 'legacy-manifest-1.9.1-alpha.json', old_raw)
    with (PACKAGE / 'authorized-document').open('a') as allowlist:
        allowlist.write('\n' + hashlib.sha256(old_raw).hexdigest())
    run('systemctl', 'stop', 'archipelago.service')
    write(DATA / 'update-backup/archipelago', BINARY.read_bytes(), 0o755)
    previous_frontend = Path('/opt/archipelago/web-ui.bak')
    if previous_frontend.exists():
        previous_frontend.rename(EVIDENCE / ('legacy-frontend-before-' + label))
    FRONTEND.rename(previous_frontend)
    FRONTEND.mkdir()
    with tarfile.open(staging / 'archipelago-frontend-2.0.4-alpha.tar.gz') as archive:
        archive.extractall(FRONTEND, filter='data')
    shutil.copytree(previous_frontend / 'aiui', FRONTEND / 'aiui')
    for directory, _, files in os.walk(FRONTEND):
        Path(directory).chmod(0o755)
        for filename in files: (Path(directory) / filename).chmod(0o644)
    write(BINARY, (staging / 'archipelago').read_bytes(), 0o755)
    shutil.rmtree(staging)
    write(DATA / 'update-pending-verify.json', json.dumps({'previous_version': '1.9.1-alpha', 'new_version': '2.0.4-alpha'}))
    run('systemctl', 'start', '--no-block', 'archipelago.service')
    wait(manager_pid, 'first-generation starting manager')
    assert run('systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '--value').stdout.strip() == 'activating'
    if boundary: write(PACKAGE / 'fault', boundary)
    run('python3', '-I', str(PACKAGE / ('fault-adopt-owner.py' if boundary else 'adopt-owner.py')), 'adopt')
    if boundary:
        wait(lambda: (PACKAGE / 'paused').exists(), label + ' durable interruption')
        run('systemctl', 'kill', '--kill-whom=main', '--signal=KILL', state()['worker_unit'])
    else:
        terminal('committed')
        assert not (DATA / 'update-pending-verify.json').exists()
        run('python3', '-I', str(PACKAGE / 'adopt-owner.py'), 'request-recovery')
    terminal('restored')
    assert pair() == old_pair
    assert not (DATA / 'update-pending-verify.json').exists()
    assert not Path('/etc/archipelago/public-update-runtime.json').exists()
    save_case(label)
    if boundary is None:
        # Retry through the older updater after recovery, retaining the previous
        # owner's real journal and drop-ins instead of resetting the fixture.
        run('systemctl', 'stop', 'archipelago.service')
        write(DATA / 'update-backup/archipelago', BINARY.read_bytes(), 0o755)
        previous_frontend.rename(EVIDENCE / 'legacy-before-retry')
        FRONTEND.rename(previous_frontend)
        FRONTEND.mkdir()
        with tarfile.open(current_dir / 'archipelago-frontend-2.0.4-alpha.tar.gz') as archive:
            archive.extractall(FRONTEND, filter='data')
        shutil.copytree(previous_frontend / 'aiui', FRONTEND / 'aiui')
        for directory, _, files in os.walk(FRONTEND):
            Path(directory).chmod(0o755)
            for filename in files: (Path(directory) / filename).chmod(0o644)
        write(BINARY, (current_dir / 'archipelago').read_bytes(), 0o755)
        write(DATA / 'update-pending-verify.json', json.dumps({'previous_version': '1.9.1-alpha', 'new_version': '2.0.4-alpha'}))
        run('systemctl', 'start', '--no-block', 'archipelago.service')
        wait(manager_pid, 'retry starting manager')
        prior_id = state()['id']
        run('python3', '-I', str(PACKAGE / 'adopt-owner.py'), 'adopt')
        terminal('committed')
        assert state()['id'] != prior_id
        run('python3', '-I', str(PACKAGE / 'adopt-owner.py'), 'request-recovery')
        terminal('restored')
        assert pair() == old_pair
        save_case('legacy-retry-after-rollback-with-retained-owner')

# Leave a genuinely interrupted mixed pair for the outer container restart.
identity, old = reset('full-reboot', runtime=True)
# A real enabled maintenance timer witnesses the restored code domain on boot.
write(PACKAGE / 'boot-maintenance-witness.py', """import hashlib,json,time
from pathlib import Path
root=Path('/var/lib/archipelago-public-transactions')
state=json.loads((root/(root/'current').read_text()/'state.json').read_bytes()) if (root/'current').exists() else None
value={'phase':state['phase'] if state else None,
       'binary':hashlib.sha256(Path('/usr/local/bin/archipelago').read_bytes()).hexdigest(),
       'runtime_record':Path('/etc/archipelago/public-update-runtime.json').exists()}
with Path('/var/lib/public-worker-evidence/boot-maintenance.jsonl').open('a') as stream:stream.write(json.dumps(value)+'\\n')
""")
write('/etc/systemd/system/archipelago-npm-bridge.service', '[Service]\nType=oneshot\nExecStart=/usr/bin/python3 -I /opt/public-fixture/boot-maintenance-witness.py\n')
write('/etc/systemd/system/archipelago-npm-bridge.timer', '[Timer]\nOnBootSec=1s\n[Install]\nWantedBy=timers.target\n')
run('systemctl', 'daemon-reload')
run('systemctl', 'enable', 'archipelago-npm-bridge.timer')
run('systemctl', 'start', 'archipelago-npm-bridge.timer')
wait(lambda: run('systemctl','show','archipelago-npm-bridge.service','-p','ActiveState','--value').stdout.strip() == 'inactive', 'baseline maintenance completed')
write(PACKAGE / 'fault', 'frontend-installed')
submit(identity, faults=True)
wait(lambda: (PACKAGE / 'paused').exists(), 'mixed pair awaiting full PID1 restart')
run('systemctl', 'stop', state()['worker_unit'])
write(EVIDENCE / 'boot-maintenance.jsonl', '')
write(PACKAGE / 'reboot-expected.json', json.dumps({'binary': old[0], 'frontend': old[1].decode(),
      'boot_id': Path('/proc/sys/kernel/random/boot_id').read_text(), 'pid1_start': Path('/proc/1/stat').read_text()}))
write(PACKAGE / 'after-reboot.py', '''import hashlib,json,time
from pathlib import Path
root=Path('/var/lib/archipelago-public-transactions')
expected=json.loads(Path('/opt/public-fixture/reboot-expected.json').read_bytes())
assert Path('/proc/1/stat').read_text()!=expected['pid1_start'], 'PID1 did not restart'
deadline=time.monotonic()+60
while time.monotonic()<deadline:
 state=json.loads((root/(root/'current').read_text()/'state.json').read_bytes())
 if state['phase']=='restored':break
 time.sleep(.1)
else:raise AssertionError(state)
assert hashlib.sha256(Path('/usr/local/bin/archipelago').read_bytes()).hexdigest()==expected['binary']
assert Path('/opt/archipelago/web-ui/index.html').read_text()==expected['frontend']
assert not Path('/etc/archipelago/public-update-runtime.json').exists()
observations=Path('/var/lib/public-worker-evidence/boot-maintenance.jsonl')
for _ in range(100):
 if observations.exists() and observations.read_text().strip():break
 time.sleep(.1)
seen=[json.loads(line) for line in observations.read_text().splitlines()]
assert seen and all(item['binary']==expected['binary'] and not item['runtime_record'] and item['phase'] in ('restored-pending-start','restored') for item in seen), seen
print('PASS maintenance timer starts only after runtime recovery',flush=True)
Path('/var/lib/public-worker-evidence/full-pid1-restart.json').write_text(json.dumps(state))
print('PASS full PID1 restart restores interrupted pair before verified manager startup')
''')
write(EVIDENCE / 'complete.json', json.dumps({'passed': True, 'signature_verifier': 'exact-document double',
                                            'manager': 'fixture ELF, not product binary', 'pid1': 'real systemd'}))