#!/usr/bin/python3
"""Normal caller/Bridge/authority-loop integration under the isolated runner.

The root coordinator is the production clock_refresh function. NTS measurement,
collector namespace transitions and outer cgroup management are fixture seams; peer credentials,
processes, issuer handoff, owner queue, storage, expiry and control protocol are real.
This is clock mechanism qualification, not actual namespace isolation coverage.
"""
import json,os,select,signal,socket,subprocess,sys,time
from dataclasses import replace
from decimal import Decimal
from pathlib import Path
from types import SimpleNamespace
sys.path.insert(0,str(Path(__file__).resolve().parents[2]/'scripts'))
import c01_supervisor as s
import c01_nts as nts
import c01_recovery as recovery
import c01_children as lifecycle

assert os.geteuid()==0 and os.environ.get('ARCHY_TEST_ISOLATED')=='1'
binary,case=sys.argv[1:];cfg=json.loads(Path(case).read_text());root=Path(cfg['root'])
APP=61021;AUTH=61022
def drop(uid):
    os.setgroups([]);os.setgid(uid);os.setuid(uid)
def native(role,uid,channel):
    name='c01_launcher::clock_refresh::tests::normal_refresh_'+role+'_process'
    env={**os.environ,'C01_NORMAL_REFRESH_CASE':case,'C01_NORMAL_REFRESH_FD':str(channel.fileno())}
    return subprocess.Popen([binary,'--exact',name,'--ignored','--test-threads=1','--nocapture'],
        env=env,pass_fds=(channel.fileno(),),preexec_fn=lambda:drop(uid),stdout=subprocess.DEVNULL)

def synthetic_measurement(policy,raw=False):
    # Explicit synthetic issuer input. No live NTS claim or local-clock fallback
    # exists in production: this replacement is confined to this isolated driver.
    begin=nts.clock_reading(raw=raw);end=nts.clock_reading(raw=raw)
    assessment=nts.Assessment('01020304','127.0.0.1',Decimal(end.wall)/nts.NANO,Decimal(0),1_000_000,(),())
    return nts.bind_observation(assessment,begin,end,recovery.boot_id(),recovery.boot_id(),policy)
nts.collect=synthetic_measurement

def fixture_collector_namespace():
    # The backend runner deliberately drops SYS_ADMIN. Do not expand its host
    # capabilities for a synthetic NTS test. Install this seam ONLY in the
    # forked fixture collector, never in production or the parent coordinator.
    network=s.c01_network;libc=s.LIBC
    expected=network.process_namespace(os.getpid())
    def same_namespace(pid,role):
        assert role=='managed' and pid==os.getpid()
        assert network.process_namespace(pid)==expected
        return expected
    def enter_host_worker(actual_libc):
        assert actual_libc is s.LIBC
        same_namespace(os.getpid(),'managed')
    class CollectorLibc:
        def __getattr__(self,name):return getattr(libc,name)
        def setns(self,fd,kind):
            assert kind==network.CLONE_NEWNET and network.identity(fd)==expected
            same_namespace(os.getpid(),'managed')
            return 0
    s.LIBC=CollectorLibc()
    s.c01_network=SimpleNamespace(identity=network.identity,CLONE_NEWNET=network.CLONE_NEWNET,
        enter_host_worker=enter_host_worker,require_private=same_namespace)

class FixtureChildren:
    def __init__(self):self.children={};self.results={};self.reserved={};self.stopping=False;self.spawned=0
    def spawn(self,role,body,keep=()):
        assert role=='collector' and role not in self.children and role not in self.results
        parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
        parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
        pid=os.fork()
        if pid==0:
            try:
                parent.close();s.close_except({child.fileno()})
                fixture_collector_namespace();body(child);os._exit(0)
            except BaseException:
                import traceback;traceback.print_exc();os._exit(111)
        child.close();record=SimpleNamespace(role=role,pid=pid,channel=parent,pidfd=os.pidfd_open(pid),process=None)
        self.children[role]=record;self.spawned+=1;return record
    def poll(self):
        record=self.children.get('collector')
        if record:
            pid,status=os.waitpid(record.pid,os.WNOHANG)
            if pid:
                self.results['collector']=os.waitstatus_to_exitcode(status)
                record.channel.close();os.close(record.pidfd);del self.children['collector']
    rearm_collector=lifecycle.Children.rearm_collector

class Owner:
    receive=s.Supervisor.receive;wait=s.Supervisor.wait;relay_collector=s.Supervisor.relay_collector
    def __init__(self):
        self.children=FixtureChildren();self.clock_sequence=0;self.power_state=None
        self.completed_one_shots={'collector'};self.wake_read,self.wake_write=os.pipe2(os.O_NONBLOCK|os.O_CLOEXEC)
    def tick(self):
        self.children.poll()
        assert self.children.results.get('collector',0)==0,'collector failed'
        return True

owner=Owner();records=[]
try:
    parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET);parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
    authority_process=native('authority',AUTH,child);child.close()
    authority=SimpleNamespace(role='authority',pid=authority_process.pid,channel=parent,process=authority_process)
    owner.children.children['authority']=authority;records.append(authority)
    end=s.now()+120*10**9;owner.wait(authority,b'LISTENING',AUTH,AUTH,end)
    parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET);parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
    app_process=native('application',APP,child);child.close()
    app=SimpleNamespace(role='application',pid=app_process.pid,channel=parent,process=app_process)
    owner.children.children['application']=app;records.append(app)
    app_peer=dict(pid=app.pid,uid=APP,gid=APP);authority_peer=dict(pid=authority.pid,uid=AUTH,gid=AUTH)
    s.packet(authority.channel,app_peer);s.packet(app.channel,authority_peer)
    owner.wait(app,b'CONNECTED',APP,APP,end)
    context=json.loads(owner.receive(authority,AUTH,AUTH,end));assert context['status']=='EPOCH_CONTEXT'
    maximum=cfg['delegation']['max_awake'];maximum=maximum['secs']*10**9+maximum['nanos']
    q=recovery.PlatformQualification(os.uname().release,recovery.clocksource(),'ab'*32,10**9,250_000_000,0,maximum,cfg['delegation']['characterized'])
    prepared=SimpleNamespace(epoch=dict(delegation=cfg['delegation']),value=dict(binding=cfg['binding'],clock_domain='01'*16))
    collector=s.start_collector(owner,q,cfg['binding'],bytes([1])*16,authority_peer,'\0'+cfg['address'],end,prepared.epoch,context['challenge'])
    owner.relay_collector(collector,authority,AUTH,AUTH,end)
    owner.wait(authority,b'AUTHORIZED',AUTH,AUTH,end);owner.wait(collector,b'ADMIN_SENT',0,0,end);s.packet(collector.channel,b'RELEASE')
    s.packet(app.channel,b'INITIAL_AUTHORIZED')
    ready=False;failed=False;deadline=time.monotonic()+25
    result_file=root/'data/refresh-result.json'
    while time.monotonic()<deadline:
        owner.tick()
        if result_file.exists() and app_process.poll() is not None:break
        if select.select([app.channel],[],[],.05)[0]:
            # Exit can race the poll above. EOF is not an authenticated status
            # packet; accept it only with the completed fixture result and a
            # successful child exit (checked below).
            if not app.channel.recv(1,socket.MSG_PEEK):
                app_process.wait(timeout=5)
                assert result_file.exists(),'application exited without a fixture result'
                break
            packet=owner.receive(app,APP,APP,s.now()+120*10**9)
            if packet==b'FAILED':failed=True;break
            msg=json.loads(packet)
            if msg['status']=='APP_READY':ready=True
            elif msg['status']=='CLOCK_REFRESH':
                s.clock_refresh(owner,prepared,q,app,authority,app_peer,authority_peer,'\0'+cfg['address'],msg)
            else:raise AssertionError('unexpected normal app status '+repr(msg))
        if result_file.exists() and app_process.poll() is not None:break
    assert ready
    app_process.wait(timeout=5)
    assert app_process.returncode==0
    result=json.loads(result_file.read_text())
    if cfg['uncertain']:
        assert failed and owner.clock_sequence==0 and owner.children.spawned==1 and result['uncertain_effect']
    else:
        assert not failed and owner.clock_sequence==2 and owner.children.spawned==3
        assert result['pid']==app.pid and len(result['domains'])==3 and result['old_consumed_and_expired_refused'] and result['new_request_succeeded']
    print(json.dumps(dict(uncertain=cfg['uncertain'],application_pid=app.pid,authority_pid=authority.pid,
        refreshes=owner.clock_sequence,collectors=owner.children.spawned,normal_caller=True,normal_authority_loop=True,
        application_or_container_restart=False,qualification='clock-mechanism-only',
        namespace_transition='fixture-seam-no-setns',actual_namespace_isolation=False,
        source_root=str(Path(__file__).resolve().parents[2]),result=result)))
finally:
    for record in records:
        if record.process.poll() is None:record.process.terminate()
    for record in records:
        try:record.process.wait(timeout=5)
        except subprocess.TimeoutExpired:record.process.kill();record.process.wait()
        record.channel.close()
    collector=owner.children.children.get('collector')
    if collector:
        try:signal.pidfd_send_signal(collector.pidfd,signal.SIGKILL)
        except ProcessLookupError:pass
        os.waitpid(collector.pid,0);collector.channel.close();os.close(collector.pidfd)
    os.close(owner.wake_read);os.close(owner.wake_write)