#!/usr/bin/python3
"""Actual normal HTTP/PTY, renewal and D09 owner acceptance in the private realm.

No replacement service, fake APP_READY, surrogate terminal, payment or private
owner wire injection. Lifecycle requests originate through the native FD10 RPC.
"""
import base64
import copy
import hashlib
import http.cookies
import json
import os
from pathlib import Path
import re
import runpy
import secrets
import shutil
import socket
import struct
import subprocess
import time
import urllib.request
import urllib.error

PACKAGE=Path('/opt/c01-fixture');EVIDENCE=Path('/var/lib/c01-normal')
scope=runpy.run_path(str(PACKAGE/'tests/regression/c01-compose.py'))
provision=scope['d09_module']();owner=provision.owner
import d09_store as store
import d09_wire as wire
generation,policy=scope['generation'],scope['policy']
sha=scope['sha'];dump=scope['dump'];run=scope['run']


def ledger():return json.loads((owner.ROOT/'ledger.json').read_text())
def binding():return ledger().get('active_binding') or json.loads(owner.BINDING.read_text())
def wait(body,label,seconds=180):
    end=time.monotonic()+seconds
    while time.monotonic()<end:
        value=body()
        if value:return value
        time.sleep(.2)
    raise AssertionError('actual acceptance timeout: '+label)


class NativeHTTP:
    def __init__(self):self.cookies={};self.sequence=0
    def request(self,path,value=None):
        headers={'Origin':'http://127.0.0.1:5678','Content-Type':'application/json'}
        if self.cookies.get('csrf_token'):headers['X-CSRF-Token']=self.cookies['csrf_token']
        if self.cookies:headers['Cookie']='; '.join(k+'='+v for k,v in self.cookies.items())
        request=urllib.request.Request('http://127.0.0.1:5678'+path,
            data=json.dumps(value).encode() if value is not None else None,headers=headers)
        with urllib.request.urlopen(request,timeout=40) as response:
            for raw in response.headers.get_all('Set-Cookie',[]):
                values=http.cookies.SimpleCookie();values.load(raw)
                self.cookies.update({k:v.value for k,v in values.items()})
            return json.loads(response.read(1024*1024))
    def rpc(self,method,params=None):
        self.sequence+=1
        result=self.request('/rpc/v1',{'jsonrpc':'2.0','id':self.sequence,'method':method,'params':params})
        if result.get('error') is not None:raise RuntimeError(method+': '+json.dumps(result['error']))
        return result['result']
    def login(self):
        self.cookies={}
        password=json.loads((PACKAGE/'baseline-auth.json').read_text())['password']
        self.rpc('auth.login',{'password':password})
        assert self.cookies,'real authenticated session cookie absent'


def terminal(http):
    record=http.request('/api/terminal/sessions',{'name':'D09 native acceptance'})
    assert record['workspace']=='/home/archipelago/Work' and record['state']=='running'
    nonce=secrets.token_hex(12);marker='C01_PTY_'+nonce
    channel=socket.create_connection(('127.0.0.1',5678),timeout=10)
    key=base64.b64encode(secrets.token_bytes(16)).decode()
    cookie='; '.join(k+'='+v for k,v in http.cookies.items())
    channel.sendall(('GET /ws/terminal?session='+record['id']+' HTTP/1.1\r\nHost: 127.0.0.1:5678\r\n'
        'Origin: http://127.0.0.1:5678\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n'
        'Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: '+key+'\r\nCookie: '+cookie+'\r\n\r\n').encode())
    data=b''
    while b'\r\n\r\n' not in data:data+=channel.recv(4096)
    header,buffer=data.split(b'\r\n\r\n',1)
    assert b' 101 ' in header
    accept=base64.b64encode(hashlib.sha1((key+'258EAFA5-E914-47DA-95CA-C5AB0DC85B11').encode()).digest())
    assert accept in header
    def send(value,opcode=1):
        raw=json.dumps(value).encode() if opcode==1 else value;mask=secrets.token_bytes(4)
        prefix=bytes([0x80|opcode,0x80|len(raw)]) if len(raw)<126 else bytes([0x80|opcode,0x80|126])+struct.pack('!H',len(raw))
        channel.sendall(prefix+mask+bytes(c^mask[i%4] for i,c in enumerate(raw)))
    command=('printf "%s\\n" "$PWD" > /home/archipelago/Work/'+nonce+'; '
             'printf ok > /home/archipelago/.local/state/archipelago/sessions/'+nonce+'; '
             'if printf bad > /opt/archipelago/c01-writable-fixture; then printf BAD_WRITE; '
              'else printf DENIED_WRITE; fi; printf "\\n'+marker+'\\n"')
    if os.environ.get('C01_HISTORY_MIGRATION')=='1':
        command=(command.rsplit('printf "\\n'+marker+'\\n"',1)[0]+
            '/usr/bin/podman inspect --format \'{"id":"{{.Id}}","state":{{json .State}}}\' fixture-stopped '
            '> /home/archipelago/Work/'+nonce+'-container; '
            '/usr/bin/podman container exists fixture-removed; '
            'printf "%s\\n" "$?" > /home/archipelago/Work/'+nonce+'-absent; '
            'printf "\\n'+marker+'\\n"')
    yaya_declarations=EVIDENCE/'yaya-container-declarations.json'
    if yaya_declarations.exists():
        declarations=json.loads(yaya_declarations.read_text())['records']
        assert len(declarations)==38 and all(re.fullmatch(r'yaya-declaration-[0-9]+',row['name']) for row in declarations)
        command=(command.rsplit('printf "\\n'+marker+'\\n"',1)[0]+
            '/usr/bin/podman inspect --format \'{"id":"{{.Id}}","state":{{json .State}}}\' '+
            ' '.join(row['name'] for row in declarations)+' > /home/archipelago/Work/'+nonce+'-yaya; '+
            'printf "\\n'+marker+'\\n"')
    try:
        send({'type':'input','data':command});send({'type':'input','data':'\r'})
        def take(count):
            nonlocal buffer
            while len(buffer)<count:
                part=channel.recv(65536);assert part,'native terminal EOF';buffer+=part
                assert len(buffer)<=1024*1024
            result,buffer=buffer[:count],buffer[count:];return result
        output='';deadline=time.monotonic()+20
        while time.monotonic()<deadline:
            first,second=take(2);length=second&127
            if length==126:length=struct.unpack('!H',take(2))[0]
            elif length==127:length=struct.unpack('!Q',take(8))[0]
            assert length<=1024*1024 and not second&128
            payload=take(length)
            if first&15==9:send(payload,10);continue
            assert first&15==1,'unexpected native terminal frame'
            message=json.loads(payload);output=message.get('data','')
            # Do not mistake the shell's echoed command for command completion.
            work=Path('/var/lib/c01-fixture/home/Work')/nonce
            session=Path('/var/lib/c01-fixture/home/.local/state/archipelago/sessions')/nonce
            if marker in output and work.exists() and session.exists():
                assert work.read_text()=='/home/archipelago/Work\n' and session.read_text()=='ok'
                assert Path('/opt/archipelago/c01-writable-fixture').read_text()=='retained old-principal-owned fixture\n'
                if os.environ.get('C01_HISTORY_MIGRATION')=='1':
                    original=json.loads((EVIDENCE/'migration-seed.json').read_text())
                    state=json.loads(work.with_name(nonce+'-container').read_text())
                    assert state['id']==original['stopped_container_id']
                    assert state['state']['Status'] in ('created','configured') and not state['state']['Running']
                    assert state['state']['Pid']==0 and state['state']['StartedAt'].startswith('0001-01-01')
                    assert work.with_name(nonce+'-absent').read_text()=='1\n'
                if yaya_declarations.exists():
                    actual=[json.loads(line) for line in work.with_name(nonce+'-yaya').read_text().splitlines()]
                    assert len(actual)==38
                    for observed,expected in zip(actual,declarations):
                        assert observed['id']==expected['id'] and observed['state']['Status']=='exited'
                        assert observed['state']['Pid']==0 and not observed['state']['Running']
                        assert observed['state']['StartedAt']==expected['last_started'],'stopped Yaya projection restarted'
                return {'session':record['id'],'workspace':record['workspace'],'pty_output_sha256':hashlib.sha256(output.encode()).hexdigest(),
                        'work_probe':nonce,'denied_package_write':True}
        raise AssertionError('native PTY command did not complete')
    finally:channel.close()


def signature(path):
    m=path.stat();return [m.st_dev,m.st_ino,m.st_uid,m.st_gid,m.st_mode,sha(path)]


def history_fields():
    value=ledger();active=binding();index={k:active[k] for k in ('target_binding_digest','composition_digest','transaction_id')}
    return dict(index,predecessor_epoch_id=value['epoch'],head_sequence=value['sequence'],
        preserved_head_digest=wire.digest(wire.canonical(value['receipt'])) if value['receipt'] else sha(owner.ROOT/'ledger.json'),
        preserved_journal_digest=wire.digest(wire.canonical({'receipts':value['journal'],'intents':value['intents']})),
        preserved_consumed_history_digest=wire.digest(wire.canonical(owner.consumption_history(value))),
        preserved_target_index_digest=wire.digest(wire.canonical(index)))


def external_record(fields):
    directory=PACKAGE/'external-history';directory.mkdir(mode=0o700,exist_ok=True)
    name=secrets.token_hex(16);history=directory/(name+'-history.json')
    store.atomic(history,wire.canonical(dict(fields,sequence=1)))
    record=directory/(name+'-record.json')
    store.atomic(record,wire.canonical(dict(fields,schema_version=1,external_history_sequence=1,
        external_history_digest=sha(history),observed_boot_id=owner.boot_id())))
    return record,history


def main():
    assert os.geteuid()==0 and Path('/run/.containerenv').exists()
    assert socket.gethostname()=='c01-prerequisite-qualification'
    http=NativeHTTP()
    wait(lambda: ledger()['prerequisite']['state']=='STARTED','actual prerequisite observation')
    before_auth=(owner.ROOT/'ledger.json').read_bytes()
    try:http.rpc('update.activate-admitted')
    except urllib.error.HTTPError as error:assert error.code in (401,403)
    else:raise AssertionError('unauthenticated admitted activation accepted')
    assert (owner.ROOT/'ledger.json').read_bytes()==before_auth
    http.login()
    retained={'work':Path('/var/lib/c01-fixture/home/Work/operator.txt'),
              'history':Path('/var/lib/c01-fixture/home/.bash_history'),
               'custom_fragments':Path('/etc/systemd/system/archipelago.service.d/45-retained-operator.conf')}
    if os.environ.get('C01_HISTORY_MIGRATION')=='1':
        launch=json.loads(policy.LAUNCH.read_text());manager=Path(launch['manager_backing']);runtime=Path(launch['runtime_data_dir'])
        retained.update(legacy_history=manager/'federation/pending_requests.json',
            user_stopped=manager/'user-stopped.json',user_uninstalled=manager/'user-uninstalled.json',
            wallet_metadata=manager/'wallet/migration-preserved.bin',runtime_wallet_metadata=runtime/'lnd/migration-preserved.bin',
            removed_data=runtime/'fixture-removed/migration-preserved.bin',subuid_data=runtime/'subuid-preserved.bin')
        yaya_proof=EVIDENCE/'yaya-container-declarations.json'
        if yaya_proof.exists():
            for record in json.loads(yaya_proof.read_text())['records']:
                for index,mount in enumerate(record['retained_mounts']):
                    retained['yaya-'+record['name']+'-'+str(index)]=Path(mount['synthetic_source'])/'retained-metadata'
    if os.environ.get('C01_RUNNING_SERVICES')=='1':
        retained.update(fips_public=Path('/etc/fips/fips.pub'),fips_config=Path('/etc/fips/fips.yaml'),
            wg_public=Path('/etc/archipelago-fips-wg/wg.public'),
            tor_public=Path('/var/lib/tor/hidden_service_archipelago/hostname'),
            tor_config=Path('/etc/tor/torrc'),nm_config=Path('/etc/NetworkManager/NetworkManager.conf'))
    before={k:signature(p) for k,p in retained.items()}
    initial=owner.service();assert initial['ActiveState']=='active'
    dump(EVIDENCE/'initial-terminal.json',terminal(http))
    baseline_journal=run('journalctl','-u','archipelago.service','--no-pager').stdout.decode()
    app_pid=re.findall(r'C01_BASELINE_READY application_pid=(\d+)',baseline_journal)[-1]
    baseline_renewals=re.findall(r'C01_CLOCK_REFRESH_COMMITTED application_pid=(\d+) epoch=(\S+)',baseline_journal)
    prior_renewals=baseline_renewals;successful_epochs=set()
    successes=0;refusals=[];deadline=time.monotonic()+400
    while time.monotonic()<deadline:
        assert owner.service()['InvocationID']==initial['InvocationID'],'normal renewal replaced manager'
        read_ok=False
        try:
            http.rpc('federation.list-pending-requests');successes+=1;read_ok=True
        except Exception as error:refusals.append((type(error).__name__+': '+str(error))[:512])
        log=run('journalctl','-u','archipelago.service','--no-pager').stdout.decode()
        renewals=re.findall(r'C01_CLOCK_REFRESH_COMMITTED application_pid=(\d+) epoch=(\S+)',log)
        if read_ok and prior_renewals and renewals and prior_renewals[-1]==renewals[-1]:
            successful_epochs.add(renewals[-1][1])
        normal_renewals=renewals[len(baseline_renewals):]
        if len(normal_renewals)>=2 and all(epoch in successful_epochs for _,epoch in normal_renewals[-2:]):
            assert all(pid==app_pid for pid,_ in renewals) and len({e for _,e in renewals})>=2
            break
        prior_renewals=renewals
        time.sleep(2.3)
    else:
        dump(EVIDENCE/'normal-renewals-refusal.json',{'successful_native_reads':successes,
            'refusals':refusals,'renewals':renewals,'successful_read_epochs':sorted(successful_epochs)})
        raise AssertionError('two post-readiness clock-only renewals with successful reads not observed')
    dump(EVIDENCE/'normal-renewals.json',{'application_pid':app_pid,'manager_invocation':initial['InvocationID'],
        'renewals':normal_renewals,'baseline_renewals':baseline_renewals,
        'successful_read_epochs':sorted(successful_epochs),
        'successful_native_reads':successes,'transient_read_refusals':refusals})

    def phase(name):
        receipt=ledger().get('receipt');return receipt if receipt and receipt['phase']==name else None
    def exchange(request):
        with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as channel:
            channel.settimeout(5);channel.connect(str(owner.SOCKET));channel.sendall(wire.canonical(request));channel.shutdown(socket.SHUT_WR)
            raw=b''
            while True:
                part=channel.recv(8192)
                if not part:break
                raw+=part;assert len(raw)<=wire.MAX
            return wire.parse(raw)
    def publish(value):
        raw=wire.canonical(value);digest=wire.digest(raw)
        store.atomic(owner.ROOT/'observations'/(digest+'.json'),raw);return digest
    def observe():
        receipt=wait(lambda: phase('AWAITING_OBSERVATION'),'native generation ready',240)
        assert owner.service()['ActiveState']=='active' and owner.service()['InvocationID']==receipt['observed_invocation_id']
        http.login();proof=terminal(http)
        after={k:signature(p) for k,p in retained.items()};assert after==before
        base={'transaction_id':receipt['transaction_id'],'observed_invocation_id':receipt['observed_invocation_id']}
        if os.environ.get('C01_HISTORY_MIGRATION')=='1':
            extended=dict(base,preserved={k:{'before':before[k],'after':after[k]} for k in before})
            proof['migration_preservation_digest']=publish(extended)
            dump(EVIDENCE/('history-preservation-'+receipt['observed_invocation_id']+'.json'),extended)
        namespace=publish(dict(base,qualification='actual native HTTP default terminal and WS PTY commands',
            default_terminal_created=True,default_terminal_workspace=proof['workspace'],
            namespace_writable=['/home/archipelago/Work','/home/archipelago/.local/state/archipelago/sessions'],native_proof=proof))
        preservation=publish(dict(base,preserved={k:{'before':wire.digest(wire.canonical(before[k])),
            'after':wire.digest(wire.canonical(after[k]))} for k in ('work','history','custom_fragments')}))
        observation={k:receipt[k] for k in ('transaction_id','target_binding_digest','composition_digest',
            'observed_invocation_id','effective_config_observation_digest','config_current_digest')}
        observation.update(namespace_observation_digest=namespace,preservation_observation_digest=preservation)
        value=binding()
        request=dict(schema_version=1,operation='ACK_OBSERVATION',route='OBSERVER',reason='OBSERVATION_READY',
            transaction_id=value['transaction_id'],request_id=secrets.token_hex(16),target_binding_digest=value['target_binding_digest'],
            composition_digest=value['composition_digest'],grant_digest=ledger().get('notice_grant') or value['grants']['OBSERVER'],expected_sequence=ledger()['sequence'],
            requester_invocation_id=None,observer_receipt_digest=publish(observation),selected_release_digest=None,staging_generation=None)
        assert exchange(request)['status']=='ACCEPTED'
        return wait(lambda: phase('COMPLETE'),'independent native observation accepted')
    def action(method):
        try:http.rpc(method)
        except (OSError,TimeoutError):pass # The actual owner may retire this HTTP invocation before reply.

    unchanged=(owner.ROOT/'ledger.json').read_bytes()
    try:http.rpc('update.activate-admitted',{'path':'/unappointed'})
    except RuntimeError:pass
    else:raise AssertionError('caller-selected activation argument accepted')
    assert (owner.ROOT/'ledger.json').read_bytes()==unchanged
    first_binding=copy.deepcopy(binding())
    action('update.activate-admitted');b_receipt=observe()
    assert sha(owner.DESTINATIONS['backend'])==first_binding['artifacts']['backend']
    wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive','B worker drain')
    # Retry through the new native B process, not a root-injected request. Its
    # retained request must resolve by lookup without another dispatch/grant.
    b_invocation=owner.service()['InvocationID'];b_history=(owner.ROOT/'ledger.json').read_bytes()
    http.rpc('update.activate-admitted')
    assert owner.service()['InvocationID']==b_invocation and (owner.ROOT/'ledger.json').read_bytes()==b_history

    # Root coordinator supplies a new, history-bound appointment. It cannot
    # reset the old ledger or manufacture an application/observer result.
    incoming=owner.ROOT/'incoming';archive=PACKAGE/'retained-incoming-B';archive.mkdir(mode=0o700)
    for path in incoming.iterdir():path.rename(archive/path.name)
    old_launch=json.loads(policy.LAUNCH.read_text())
    for label in ('new','restore'):
        for name,path in owner.destinations(first_binding).items():
            target=incoming/(label+'-'+name)
            if path.is_dir():shutil.copytree(path,target)
            elif path.exists():shutil.copy2(path,target)
    with (incoming/'new-backend').open('ab') as stream:stream.write(b'\nC01-disposable-generation-C\n')
    new_launch=dict(old_launch,native_sha256=sha(incoming/'new-backend'))
    dump(incoming/'new-launch',new_launch)
    dump(incoming/'exclusion.json',json.loads((policy.STATE/'journal.json').read_text()))
    effective=wire.digest(owner.effective_configuration().encode())
    successor=copy.deepcopy(first_binding)
    successor.pop('requester_uid');successor.pop('requester_gid')
    successor.update(transaction_id=secrets.token_hex(16),admission_epoch_id=secrets.token_hex(16),composition_digest=secrets.token_hex(32),
        grants={k:secrets.token_hex(32) for k in first_binding['grants']},
        artifacts={'backend':sha(incoming/'new-backend'),'frontend':owner.object_digest(incoming/'new-frontend')},
        restore_artifacts=first_binding['artifacts'],effective_config={'APPLY':effective,'RESTORE':effective},
        exclusion_evidence_digest=sha(incoming/'exclusion.json'),
        prerequisite={'grant_digest':secrets.token_hex(32),'backend_sha256':sha(owner.DESTINATIONS['backend']),'effective_config_digest':effective})
    successor['release_identity']=['fixture-C',[['archipelago','fixture-C',successor['artifacts']['backend'],(incoming/'new-backend').stat().st_size]]]
    successor['objects']={name:{'old':owner.object_digest(path),'new':owner.object_digest(incoming/('new-'+name)),
        'restore':owner.object_digest(incoming/('restore-'+name))} for name,path in owner.destinations(successor).items()}
    successor['c01']['generations']={name:{'native_sha256':launch['native_sha256'],'launch_sha256':wire.digest(generation.canonical(launch))}
        for name,launch in [('old',old_launch),('new',new_launch),('restore',old_launch)]}
    for operation,a,b in [('APPLY',old_launch,new_launch),('RESTORE',new_launch,old_launch)]:
        record={'version':1,'policy_sha256':sha(policy.CONFIG),'runtime_sha256':sha(policy.RUNTIME),
            'transaction_id':successor['transaction_id'],'action':operation,'grant_digest':secrets.token_hex(32),
            'from_launch':a,'to_launch':b,'from_launch_sha256':wire.digest(generation.canonical(a)),
            'to_launch_sha256':wire.digest(generation.canonical(b)),
            'from_workspace_digest':successor['objects']['config']['old' if operation=='APPLY' else 'new'],
            'to_workspace_digest':successor['objects']['config']['new' if operation=='APPLY' else 'restore']}
        raw=generation.canonical(record);record['external_history_digest']=wire.digest(raw)
        store.atomic(incoming/('c01-history-'+wire.digest(raw)+'.json'),raw)
        raw=generation.canonical(record);digest=wire.digest(raw)
        store.atomic(incoming/('c01-generation-'+digest+'.json'),raw);successor['c01']['transitions'][operation]=digest
    dump(incoming/'admission.json',successor)
    protected=dict(successor,requester_uid=first_binding['requester_uid'],requester_gid=first_binding['requester_gid'])
    fields=dict(history_fields(),successor_binding_digest=wire.digest(wire.canonical(protected)),terminal_disposition='APPLY')
    record,history=external_record(fields)
    original_journal=list(ledger()['journal']);provision.successor(incoming/'admission.json',record,history)
    assert ledger()['journal']==original_journal
    unchanged=(owner.ROOT/'ledger.json').read_bytes();provision.successor(incoming/'admission.json',record,history)
    assert unchanged==(owner.ROOT/'ledger.json').read_bytes()
    action('update.activate-admitted');c_receipt=observe()
    assert sha(owner.DESTINATIONS['backend'])==successor['artifacts']['backend']
    wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive','C worker drain')
    action('update.recover-admitted');restored=observe()
    assert restored['config_action']=='RESTORE' and sha(owner.DESTINATIONS['backend'])==first_binding['artifacts']['backend']
    wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive','restore worker drain')
    previous_invocation=owner.service()['InvocationID'];saved=(owner.ROOT/'ledger.json').read_bytes()
    run('systemctl','restart','archipelago-d09-lifecycle.service')
    wait(lambda: owner.service()['MainPID']=='0' and owner.service()['ActiveState'] in ('inactive','failed'),'owner-loss retirement')
    assert (owner.ROOT/'ledger.json').read_bytes()==saved,'owner restart changed retained history'
    probe=owner.Owner.__new__(owner.Owner)
    drain_waits=[]
    def drained():
        try:return probe.drained_disposition(ledger())
        except ValueError as error:
            reason=str(error)
            if reason not in ('management process/job not drained','prerequisite process still exists',
                'worker descendants not retired','worker process/job not drained','worker process still exists'):
                raise
            if reason not in drain_waits:drain_waits.append(reason)
            return None
    disposition=wait(drained,'owned worker drain proof')
    fields=dict(history_fields(),allowed_action='REACTIVATE_RETAINED',new_epoch_id=secrets.token_hex(16),
        new_grant_digest=secrets.token_hex(32),execution_grant_digest=secrets.token_hex(32),disposition_digest=disposition)
    record,history=external_record(fields);provision.readmit(record,history,activation=True)
    recovered=observe();fresh_invocation=owner.service()['InvocationID']
    assert fresh_invocation!=previous_invocation
    provision.readmit(record,history,activation=True)
    assert owner.service()['InvocationID']==fresh_invocation,'readmission replay dispatched twice'
    dump(EVIDENCE/'normal-generations.json',{'generations':['A','B','C','B'],'B':b_receipt,'C':c_receipt,'restored':restored,
        'owner_recovery':recovered,'recovery_invocation':fresh_invocation,'drain_waits':drain_waits,
        'preservation':before,'journal_entries':len(ledger()['journal']),'consumed_grants':len(ledger()['grants'])})
    if os.environ.get('C01_RUNNING_SERVICES')=='1':
        import c01_admission
        launch=json.loads(policy.LAUNCH.read_text())
        wait(lambda: run('systemctl','show',ledger()['execution']['unit'],'-p','ActiveState','--value').stdout.strip()==b'inactive',
             'final recovery worker drain')
        observations=[]
        # Actual producer loss must retire the application, not retarget a
        # restarted daemon behind its already-issued capability.
        run('systemctl','stop','archipelago-fips.service')
        wait(lambda: owner.service()['MainPID']=='0' and owner.service()['ActiveState'] in ('failed','inactive'),
             'required FIPS producer-loss manager retirement')
        for name,unit in c01_admission.UNITS.items():
            if name!='fips':run('systemctl','stop',unit)
            try:
                contract=c01_admission.Contract(launch)
            except (ValueError,OSError) as error:
                observations.append({'service':name,'unit':unit,'admission_refused':True,'reason':str(error)[:512]})
            else:
                contract.close();raise AssertionError('running service profile accepted stopped '+name)
            run('systemctl','start',unit)
        assert owner.service()['MainPID']=='0','service restoration silently reactivated manager'
        assert {k:signature(p) for k,p in retained.items()}==before
        dump(EVIDENCE/'running-service-loss.json',{'producer_loss_retired_manager':True,
             'automatic_manager_reactivation':False,'refusals':observations,'preservation_unchanged':True})
    return 0


if __name__=='__main__':raise SystemExit(main())