Prospective clock lifecycle controls

Registered trust commits510fbcb/d33fb33, productionc8fbf41. All new controls are NOT RUN. This plan supports X088 pre-flight; it establishes no native/compiler/package/browser outcome. The fixture shifts only a disposable target; no host clock change is admitted.

pre-flight: four literal SessionClockLifecycleTests at cleand33fb339db7a78af774e7a33911b2ec6dca22a8b, prospective trust510fbcb+d33; productionc8fbf41/oracle4817a0d0 unchanged. No package/browser/service/native compiler/host-clock change. Exact outer: python3 -B META/run-step-console-clock-lifecycle-red.py console-clock-lifecycle-red-01 1073741824 python3 -B -m unittest discover -s tools/ui -p test_session_clock.py -v -k SessionClockLifecycleTests; META=DATA/results/candidate-build-fresh-8fb6f413179b-01, cwd=DATA/scratch/console-access-acceptance, DATA=/run/media/kmosoti/data/FabricO11y. DriverSHA765c61db8facb00e64b742a59533d50eeadb979f6fad971805d39214b8f244c1; actual inner launcher python3 -B tools/resource_group.py -- python3 -B -m unittest discover -s tools/ui -p test_session_clock.py -v -k SessionClockLifecycleTests; default16/20GiB/no-swap/1800s, serial/full1GiB reserve/fresh95GB census. Literal real child ignoresTERM after leader exit; fake compiler writes diagnostic+partial-object then stalls under explicitly TEST-ONLY accelerated timer. Subreaper/owned child cleanup restores setting. No real240s/compiler-binding acceptance claimed. Frozen-image controls replace content or identical bytes/new inode and forge mapped-helper device/inode against the original frozen witness; exact refusal required, current-file agreement insufficient. TestsSHA76f937331e2ebe6e2da9ee121a39a0411c823f38bd788c3c4a36107cd3dcec49/protocolSHAdea94fa0109a4e757ac2c022b2829e2ec4073678feb077b1472b32a9ed6cdab3. Root captures actual intended red/stopped evidence before smallest production correction; actual source controls remain unchanged for green. Shared-cache deletion remainsHOLD perC047; this run cannot admit VM/pressure/soak or establish clock/native/browser acceptance.

Source /run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/test_session_clock.py; SHA256 76f937331e2ebe6e2da9ee121a39a0411c823f38bd788c3c4a36107cd3dcec49.

"""Independent literal oracle controls and separate fixture lifecycle regressions."""
import copy
from contextlib import contextmanager
import ctypes
import hashlib
import inspect
import os
from pathlib import Path
import signal
import subprocess
import sys
import tempfile
import threading
from time import monotonic as real_monotonic, sleep
import unittest
from unittest.mock import Mock, patch

import session_clock_oracle as oracle

ROOT=Path(__file__).resolve().parents[2]


def clock_rows():
    rows=[]
    for sequence,offset in enumerate((0,1,1799,1800,28799,28800)):
        for clock in (0,1,4):
            raw=100_500_000_000 if clock==0 else 200_500_000_000
            rows.append({'pid':700,'sequence':sequence,'offset_s':offset,'clock_id':clock,
                         'raw_ns':raw,'returned_ns':raw+(offset*1_000_000_000 if clock==0 else 0),
                         'syscall_before_ns':raw-1_000_000,'syscall_after_ns':raw+1_000_000})
    return rows


def isolation():
    ordinary={role:{'pid':pid,'helper_maps':[],'control_env_names':[],'control_fd_present':False,
                    'syscall_before_ns':100_000_000_000,'clock_ns':100_001_000_000,'syscall_after_ns':100_002_000_000}
              for pid,role in enumerate(('controller','browser','driver','companion'),701)}
    return {'target_pid':700,'target_executable_sha256':'a'*64,'target_helper_maps':['b'*64],
            'target_control_cloexec':True,'target_control_type':'AF_UNIX/SOCK_SEQPACKET','ordinary_processes':ordinary,
            'fork':{'pid':705,'target_offset_s':1,'raw_ns':100_000_000_000,'returned_ns':100_000_000_000,
                    'control_sequence_before':1,'control_sequence_after':1,'exec_helper_maps':[],
                    'exec_control_env_names':[],'exec_control_fd_present':False}}


def sessions():
    metadata={'A':{'hash':'a'*64,'issued_s':1000,'expires_s':29800,'baseline_seen_s':1000},
              'B':{'hash':'b'*64,'issued_s':1000,'expires_s':29800,'baseline_seen_s':1000},
              'C':{'hash':'c'*64,'issued_s':10000,'expires_s':38800,'baseline_seen_s':10000}}
    rows=[]
    def add(cell,key,second,status,last_seen):
        anchor=metadata[key]
        rows.append({'cell':cell,'session_hash':anchor['hash'],'pid':700,'issued_s':anchor['issued_s'],
                     'expires_s':anchor['expires_s'],'last_seen_s':last_seen,'request_s':second,
                     'server_call_s':second,'status':status,'error':'sign in required' if status==401 else None})
    add('idle_before','A',2799,200,1000);add('idle_at','B',2800,401,1000)
    previous=10000
    for second in range(11000,38001,1000):add('absolute_keepalive','C',second,200,previous);previous=second
    add('absolute_before','C',38799,200,38000);add('absolute_at','C',38800,401,38799)
    return rows,metadata


class SessionClockTests(unittest.TestCase):
    def test_fixture_mechanism_sources_are_available_for_native_binding_admission(self):
        # This initial red is mechanism absence, never a reproduced product bug.
        for name in ('session_clock.py','session_clock.c','session_clock_probe.rs'):
            self.assertTrue((ROOT/'tools/ui'/name).is_file(), 'missing fixture mechanism: '+name)

    def test_literal_frames_and_bounds_reject_bad_wire_records(self):
        frames=[bytes.fromhex(line) for line in (
            '46434c4b7631000000000000000000000000000000000000',
            '46434c4b7631000001000000000000000100000000000000',
            '46434c4b7631000002000000000000000707000000000000',
            '46434c4b7631000003000000000000000807000000000000',
            '46434c4b7631000004000000000000007f70000000000000',
            '46434c4b7631000005000000000000008070000000000000')]
        oracle.check_frames(frames)
        maximum=[b'FCLKv1\0\0'+i.to_bytes(8,'little')+i.to_bytes(8,'little') for i in range(64)]
        oracle.check_frames(maximum)
        for bad,error in [([], 'frame-count'),(maximum+[maximum[-1]],'frame-count'),
                          ([frames[0][:-1]],'frame-length'),([frames[0]+b'x'],'frame-length'),
                          ([b'wrong-v1'+frames[0][8:]],'frame-version'),
                          ([frames[0][:16]+(1).to_bytes(8,'little')],'frame-offset'),
                          ([frames[0],frames[0]],'frame-sequence'),([frames[0],frames[2]],'frame-sequence'),
                          ([frames[0],frames[1],frames[2][:16]+bytes(8)],'frame-offset'),
                          ([frames[0],frames[1][:16]+(28861).to_bytes(8,'little')],'frame-offset'),
                          ([frames[0],frames[1][:16]+(2**64-1).to_bytes(8,'little')],'frame-offset')]:
            with self.subTest(error=error),self.assertRaisesRegex(ValueError,error):oracle.check_frames(bad)

    def test_independent_raw_clock_and_sleep_controls_reject_shifted_or_unbound_evidence(self):
        oracle.check_probe_clocks(clock_rows(),700)
        for index,field,value,error in [(3,'returned_ns',100_500_000_000,'clock-realtime-only'),
                                       (1,'returned_ns',201_500_000_000,'clock-realtime-only'),
                                       (3,'returned_ns',101_500_000_001,'clock-realtime-only'),
                                       (3,'pid',701,'clock-pid'),(3,'sequence',0,'clock-binding'),
                                       (0,'syscall_after_ns',102_000_000_000,'clock-bracket-inconclusive'),
                                       (0,'raw_ns',100_000_000_000,'clock-raw-bracket')]:
            changed=clock_rows();changed[index][field]=value
            with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):oracle.check_probe_clocks(changed,700)
        with self.assertRaisesRegex(ValueError,'clock-completeness'):oracle.check_probe_clocks(clock_rows()[:-1],700)
        sleep={'start_before_ns':200_000_000_000,'start_after_ns':200_001_000_000,
               'end_before_ns':200_502_000_000,'end_after_ns':200_503_000_000,'instant_elapsed_ns':502_000_000}
        oracle.check_sleep(sleep)
        with self.assertRaisesRegex(ValueError,'sleep-monotonic'):oracle.check_sleep(dict(sleep,instant_elapsed_ns=1_502_000_000))

    def test_target_maps_fork_and_child_escape_controls_are_independent(self):
        def check(row):oracle.check_isolation(row,700,'b'*64,'a'*64)
        check(isolation())
        for field,value,error in [('target_pid',701,'isolation-target'),('target_helper_maps',[],'isolation-interception'),
                                  ('target_helper_maps',['c'*64],'isolation-interception'),('target_control_cloexec',False,'isolation-control'),
                                  ('target_executable_sha256','d'*64,'isolation-target')]:
            changed=isolation();changed[field]=value
            with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):check(changed)
        for role in ('controller','browser','driver','companion'):
            for field,value in [('helper_maps',['b'*64]),('control_env_names',['FABRIC_CLOCK_FD']),('control_fd_present',True)]:
                changed=isolation();changed['ordinary_processes'][role][field]=value
                with self.subTest(role=role,field=field),self.assertRaisesRegex(ValueError,'isolation-escape'):check(changed)
            changed=isolation();changed['ordinary_processes'][role]['clock_ns']+=1_000_000_000
            with self.subTest(role=role),self.assertRaisesRegex(ValueError,'isolation-host-clock'):check(changed)
        for field,value,error in [('returned_ns',101_000_000_000,'isolation-fork-consumption'),
                                  ('control_sequence_after',2,'isolation-fork-consumption'),
                                  ('exec_helper_maps',['b'*64],'isolation-exec-escape'),
                                  ('exec_control_env_names',['LD_PRELOAD'],'isolation-exec-escape'),
                                  ('exec_control_env_names',['FABRIC_CLOCK_FD'],'isolation-exec-escape'),
                                  ('exec_control_fd_present',True,'isolation-exec-escape')]:
            changed=isolation();changed['fork'][field]=value
            with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):check(changed)

    def test_literal_session_boundaries_reject_wrong_refusal_baseline_or_token(self):
        rows,metadata=sessions();oracle.check_sessions(rows,metadata,700)
        for index,field,value,error in [(1,'status',200,'session-refusal'),(1,'status',429,'session-refusal'),
                                       (1,'status',403,'session-refusal'),(1,'status',None,'session-refusal'),
                                       (1,'last_seen_s',1001,'session-baseline'),(1,'request_s',2801,'session-boundary'),
                                       (-1,'server_call_s',38801,'session-boundary-inconclusive'),
                                       (-1,'session_hash','d'*64,'session-binding'),(-1,'issued_s',10001,'session-baseline'),
                                       (-1,'error','access denied','session-refusal')]:
            changed=copy.deepcopy(rows);changed[index][field]=value
            with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):oracle.check_sessions(changed,metadata,700)
        with self.assertRaisesRegex(ValueError,'session-completeness'):oracle.check_sessions(rows[:2]+rows[3:],metadata,700)


@contextmanager
def lifecycle_fixture():
    libc=ctypes.CDLL(None,use_errno=True);previous=ctypes.c_int()
    if libc.prctl(37,ctypes.byref(previous),0,0,0) or libc.prctl(36,1,0,0,0):raise RuntimeError('fixture subreaper unavailable')
    try:
        with tempfile.TemporaryDirectory(prefix='clock-lifecycle-',dir=os.environ['FABRIC_SCRATCH_ROOT']) as directory:yield Path(directory)
    finally:
        if libc.prctl(36,previous.value,0,0,0):raise RuntimeError('fixture subreaper restoration failed')


def marker_pid(marker):
    deadline=real_monotonic()+2
    while real_monotonic()<deadline:
        try:
            text=marker.read_text()
            if text.isdigit():return int(text)
        except FileNotFoundError:pass
        sleep(.005)
    raise RuntimeError('bounded fixture PID witness unavailable')


def reap_fixture(pid):
    deadline=real_monotonic()+5
    while real_monotonic()<deadline:
        try:
            if os.waitpid(pid,os.WNOHANG)[0]:return
        except ChildProcessError:
            if not Path(f'/proc/{pid}').exists():return
        sleep(.005)


def stop_fixture(pid):
    try:
        if os.waitpid(pid,os.WNOHANG)[0]:return
        os.kill(pid,signal.SIGKILL);os.waitpid(pid,0)
    except (ProcessLookupError,ChildProcessError):pass


class SessionClockLifecycleTests(unittest.TestCase):
    def test_exited_leader_does_not_leave_term_resistant_group_member(self):
        import session_clock as sut
        with lifecycle_fixture() as root:
            marker=root/'child.pid'
            program='import os,signal,sys,time;from pathlib import Path\nchild=os.fork()\nif not child:\n signal.signal(signal.SIGTERM,signal.SIG_IGN);Path(sys.argv[1]).write_text(str(os.getpid()))\n while True:signal.pause()\nwhile not Path(sys.argv[1]).exists():time.sleep(.005)\nos._exit(0)\n'
            process=subprocess.Popen([sys.executable,'-c',program,str(marker)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,start_new_session=True)
            pid=marker_pid(marker);process.wait(timeout=2);receiver=threading.Thread(target=reap_fixture,args=(pid,),daemon=True);receiver.start()
            fixture=sut.SessionClock.__new__(sut.SessionClock);fixture.process=process;fixture.receiver=receiver;fixture.parent=Mock();fixture.stderr=Mock()
            try:
                error=None
                try:fixture.close()
                except ValueError as failure:error=str(failure)
                observed=(error,receiver.is_alive())
            finally:
                stop_fixture(pid);receiver.join(timeout=2);process.stdin.close();process.stdout.close()
            self.assertEqual(observed,(None,False),'whole owned group must stop even after its leader has exited')

    def test_compiler_timeout_stops_descendant_and_preserves_output_and_partial_build(self):
        import session_clock as sut
        with lifecycle_fixture() as root:
            marker=root/'compiler-child.pid';compiler=root/'fixture-compiler';output=root/'receipt'
            compiler.write_text('#!'+sys.executable+'\nimport os,signal,sys,time\nfrom pathlib import Path\n'
                'if "-vV" in sys.argv:print("release: 1.99.0\\ncommit-hash: b940084d7");sys.exit(0)\n'
                'if "--version" in sys.argv:print("literal fixture compiler");sys.exit(0)\n'
                'Path(sys.argv[sys.argv.index("-o")+1]).write_bytes(b"partial-object-public-evidence")\n'
                'print("compiler-timeout-public-output",file=sys.stderr,flush=True)\nchild=os.fork()\n'
                'if not child:\n signal.signal(signal.SIGTERM,signal.SIG_IGN);Path(os.environ["FCLK_TEST_MARKER"]).write_text(str(os.getpid()))\n'
                'while True:signal.pause()\n');compiler.chmod(0o700)
            def watcher():
                reap_fixture(marker_pid(marker))
            receiver=threading.Thread(target=watcher,daemon=True);receiver.start();calls=0
            def clock():
                nonlocal calls
                calls+=1;return real_monotonic()+(0 if calls==1 else 239.25)
            try:
                with patch.dict(os.environ,FCLK_TEST_MARKER=str(marker)),patch.object(sut.time,'monotonic',clock),patch.object(sys,'argv',['session-clock','--output',str(output),'--rustc',str(compiler),'--cc',str(compiler)]):
                    with self.assertRaises(subprocess.TimeoutExpired):sut.main()
                receiver.join(timeout=1);pid=marker_pid(marker)
                blob=b''.join(path.read_bytes() for path in output.rglob('*') if path.is_file() and path.stat().st_size<=2*1024**2)
                observed=(Path(f'/proc/{pid}').exists(),b'compiler-timeout-public-output' in blob,b'partial-object-public-evidence' in blob)
            finally:
                if marker.exists():stop_fixture(marker_pid(marker))
                receiver.join(timeout=2)
            self.assertEqual(observed,(False,True,True),'timeout must stop descendants and archive captured output plus partial build evidence')

    def test_replaced_images_are_rejected_against_the_original_frozen_witness(self):
        import session_clock as sut
        with lifecycle_fixture() as root:
            executable=root/'probe';helper=root/'shim.so';libc=root/'libc.so.6';proc=root/'proc';proc.mkdir();libc.write_bytes(b'literal libc witness')
            (proc/'exe').symlink_to(executable);(proc/'stat').write_text('700 (fixture) S '+'0 '*18+'123\n');(proc/'cgroup').write_text('0::/literal-contained-fixture\n')
            for field,message in [('executable','isolation-target-frozen'),('helper','isolation-helper-frozen')]:
                with self.subTest(field=field):
                    executable.write_bytes(b'literal frozen executable');helper.write_bytes(b'literal frozen helper')
                    expected={'executable_sha256':hashlib.sha256(b'literal frozen executable').hexdigest(),'helper_sha256':hashlib.sha256(b'literal frozen helper').hexdigest(),
                              'executable_identity':{'device':executable.stat().st_dev,'inode':executable.stat().st_ino},'helper_identity':{'device':helper.stat().st_dev,'inode':helper.stat().st_ino}}
                    stat=helper.stat();(proc/'maps').write_text(f'1000-2000 r-xp 00000000 {os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x} {stat.st_ino} {helper}\n2000-3000 r-xp 00000000 00:00 1 {libc}\n')
                    (executable if field=='executable' else helper).write_bytes(b'literal replacement after freeze')
                    # Legacy checker lacks the witness parameter and admits the real changed files.
                    options={'expected':expected} if 'expected' in inspect.signature(sut.process_binding).parameters else {}
                    refusal=None
                    with patch.object(sut,'Path',side_effect=lambda value:proc if str(value)=='/proc/700' else Path(value)):
                        try:sut.process_binding(700,executable,helper,**options)
                        except ValueError as failure:refusal=str(failure)
                    self.assertEqual(refusal,message,'current path agreement cannot replace the pre-launch frozen hashes and identities')

    def test_same_bytes_replacement_and_forged_mapping_reject_frozen_identity(self):
        import session_clock as sut
        with lifecycle_fixture() as root:
            executable=root/'probe';helper=root/'shim.so';libc=root/'libc.so.6';proc=root/'proc';proc.mkdir();libc.write_bytes(b'literal libc witness')
            (proc/'exe').symlink_to(executable);(proc/'stat').write_text('700 (fixture) S '+'0 '*18+'123\n');(proc/'cgroup').write_text('0::/literal-contained-fixture\n')
            for case in ('executable-replaced','helper-replaced','mapping-device','mapping-inode'):
                with self.subTest(case=case):
                    executable.write_bytes(b'literal frozen executable');helper.write_bytes(b'literal frozen helper')
                    expected={'executable_sha256':hashlib.sha256(b'literal frozen executable').hexdigest(),'helper_sha256':hashlib.sha256(b'literal frozen helper').hexdigest(),
                              'executable_identity':{'device':executable.stat().st_dev,'inode':executable.stat().st_ino},'helper_identity':{'device':helper.stat().st_dev,'inode':helper.stat().st_ino}}
                    if case.endswith('-replaced'):
                        selected=executable if case=='executable-replaced' else helper;frozen=selected.stat();replacement=root/'same-bytes-new-inode'
                        replacement.write_bytes(selected.read_bytes());self.assertNotEqual(replacement.stat().st_ino,frozen.st_ino)
                        replacement.replace(selected)
                        self.assertEqual(hashlib.sha256(selected.read_bytes()).hexdigest(),expected['executable_sha256' if selected==executable else 'helper_sha256'])
                        self.assertNotEqual(selected.stat().st_ino,frozen.st_ino)
                    stat=helper.stat();major=os.major(stat.st_dev)+(1 if case=='mapping-device' else 0);inode=stat.st_ino+(1 if case=='mapping-inode' else 0)
                    (proc/'maps').write_text(f'1000-2000 r-xp 00000000 {major:02x}:{os.minor(stat.st_dev):02x} {inode} {helper}\n2000-3000 r-xp 00000000 00:00 1 {libc}\n')
                    options={'expected':expected} if 'expected' in inspect.signature(sut.process_binding).parameters else {}
                    refusal=None
                    with patch.object(sut,'Path',side_effect=lambda value:proc if str(value)=='/proc/700' else Path(value)):
                        try:sut.process_binding(700,executable,helper,**options)
                        except ValueError as failure:refusal=str(failure)
                    self.assertEqual(refusal,'isolation-target-frozen' if case=='executable-replaced' else 'isolation-helper-frozen',
                                     'unchanged bytes cannot replace frozen file or loaded mapping device/inode witnesses')


if __name__=='__main__':unittest.main()

Source /run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/session_clock.py; SHA256 7d0a436a068574a4d5f74fa70da9aff534c21a88e10eed438f5a751085f32c3e.

"""Disposable target-only clock fixture and contained native binding admission."""
import argparse
import ctypes
import hashlib
import json
import os
from pathlib import Path
import platform
import select
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
import session_clock_oracle as oracle

HERE=Path(__file__).resolve().parent
sys.path.insert(0,str(HERE.parents[1]))
from resource_group import STORAGE, require_limits


def digest(path):
    with Path(path).open('rb') as stream:return hashlib.file_digest(stream,'sha256').hexdigest()


class KernelTime(ctypes.Structure):_fields_=[('sec',ctypes.c_long),('nsec',ctypes.c_long)]
def kernel_time():
    value=KernelTime();oracle.require(ctypes.CDLL(None,use_errno=True).syscall(228,0,ctypes.byref(value))==0,'fixture-raw-clock');return value.sec*1000000000+value.nsec


def process_binding(pid, executable, helper, intercepted=True):
    proc=Path(f'/proc/{pid}');actual=proc/'exe'
    oracle.require(actual.resolve()==executable.resolve() and digest(actual)==digest(executable),'isolation-target')
    rows=[line.split(maxsplit=5) for line in (proc/'maps').read_text().splitlines()]
    mappings={(row[3],row[4],row[5]) for row in rows if len(row)==6 and row[5]==str(helper)};stat=helper.stat()
    oracle.require(mappings==({(f'{os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x}',str(stat.st_ino),str(helper))} if intercepted else set()),'isolation-interception')
    libc=next(Path(row[5]) for row in rows if len(row)==6 and row[5].endswith('/libc.so.6'))
    return {'pid':pid,'executable_sha256':digest(actual),'helper_sha256':digest(helper) if mappings else None,'libc':str(libc),'libc_sha256':digest(libc),
            'start_ticks':int((proc/'stat').read_text().rsplit(')',1)[1].split()[19]),'cgroup':(proc/'cgroup').read_text(),'helper_maps':sorted(mappings)}


class SessionClock:
    def __init__(self, executable, helper, directory, deadline, initial=None, socket_type=socket.SOCK_SEQPACKET, preload=True, invalid_fd=False):
        self.deadline=deadline;self.frames=[];self.rows=[];self.bytes=0;self.error=None;self.buffer=b''
        env=dict(os.environ);oracle.require(not any(name in env for name in ('LD_PRELOAD','LD_AUDIT','FABRIC_CLOCK_FD')),'fixture-environment')
        self.parent,self.child=socket.socketpair(socket.AF_UNIX,socket_type|socket.SOCK_NONBLOCK|socket.SOCK_CLOEXEC)
        self.fd=self.child.fileno();self.socket_identity=os.readlink(f'/proc/self/fd/{self.fd}')
        self.parent.send(initial if initial is not None else self.frame(0));self.stderr=directory.open('xb')
        if preload:env.update(LD_PRELOAD=str(helper),FABRIC_CLOCK_FD=str(1048575 if invalid_fd else self.fd))
        self.argv=[str(executable),'interactive',str(self.fd),str(helper),self.socket_identity]
        try:self.process=subprocess.Popen(self.argv,env=env,pass_fds=(self.fd,),stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=self.stderr,start_new_session=True)
        except BaseException:self.parent.close();self.child.close();self.stderr.close();raise
        self.child.close();self.receiver=threading.Thread(target=self.receive,daemon=True);self.receiver.start()
    def frame(self,offset):
        frame=b'FCLKv1\0\0'+len(self.frames).to_bytes(8,'little')+offset.to_bytes(8,'little')
        oracle.check_frames(self.frames+[frame]);self.frames.append(frame);return frame
    def advance(self,offset):oracle.require(self.parent.send(self.frame(offset))==24,'fixture-control-send')
    def receive(self):
        try:
            while True:
                if not select.select([self.parent],[],[],.1)[0]:continue
                raw=self.parent.recv(257)
                if not raw:return
                self.bytes+=len(raw);oracle.require(len(raw)<=256 and self.bytes<=1048576,'fixture-ledger-bound')
                self.rows.append(json.loads(raw))
        except BaseException as error:
            self.error=type(error).__name__+': '+str(error)
            try:os.killpg(self.process.pid,signal.SIGTERM)
            except ProcessLookupError:pass
    def line(self):
        while b'\n' not in self.buffer:
            oracle.require(self.error is None,'fixture-ledger-error')
            remaining=self.deadline-time.monotonic();oracle.require(remaining>0,'fixture-deadline')
            oracle.require(select.select([self.process.stdout],[],[],remaining)[0],'fixture-deadline')
            raw=os.read(self.process.stdout.fileno(),4096);oracle.require(bool(raw),'fixture-probe-eof')
            self.buffer+=raw;oracle.require(len(self.buffer)<=65536,'fixture-probe-bound')
        raw,self.buffer=self.buffer.split(b'\n',1);return json.loads(raw)
    def command(self,name):self.process.stdin.write((name+'\n').encode());self.process.stdin.flush()
    def close(self):
        try:os.killpg(self.process.pid,signal.SIGTERM)
        except ProcessLookupError:pass
        try:self.process.wait(timeout=3)
        except subprocess.TimeoutExpired:os.killpg(self.process.pid,signal.SIGKILL);self.process.wait(timeout=3)
        self.receiver.join(timeout=1);self.parent.close();self.process.stdin.close();self.process.stdout.close();self.stderr.close()
        oracle.require(not self.receiver.is_alive(),'fixture-receiver-stop')
        try:os.killpg(self.process.pid,0)
        except ProcessLookupError:return
        raise ValueError('fixture-process-group-stop')


def native_admission(executable,helper,output,deadline,report):
    target=SessionClock(executable,helper,output/'positive.stderr',deadline);report['clocks']=[];report['system_time']=[]
    try:
        ready=target.line();report['ready']=ready
        oracle.require(ready=={'kind':'ready','pid':target.process.pid,'env_clear':True,'cloexec':True,'socket_type':5} and os.readlink(f'/proc/{target.process.pid}/fd/{target.fd}')==target.socket_identity and int(Path(f'/proc/{target.process.pid}/fdinfo/{target.fd}').read_text().split('flags:',1)[1].split()[0],8)&os.O_CLOEXEC,'isolation-control')
        report['target']=process_binding(target.process.pid,executable,helper);report['controller']=process_binding(os.getpid(),Path(sys.executable),helper,False);report['host_clocks']=[]
        report['control_fd']={'identity':target.socket_identity,'fdinfo':Path(f'/proc/{target.process.pid}/fdinfo/{target.fd}').read_text()};report['probe_argv']=target.argv
        for sequence,offset in enumerate((0,1,1799,1800,28799,28800)):
            if sequence:target.advance(offset)
            if sequence==2:
                target.command('fork');fork,child,done=(target.line() for _ in range(3));report['fork']=[fork,child,done]
                row=fork['row'];oracle.require(row['pid']==child['pid']==done['fork_pid']!=target.process.pid and done['status']==0,'isolation-fork-pid')
                oracle.require(row['offset_s']==1 and row['sequence']==done['sequence_before']==1 and row['returned_ns']==row['raw_ns'],'isolation-fork-consumption')
                oracle.require(row['syscall_before_ns']<=row['raw_ns']<=row['syscall_after_ns'] and row['syscall_after_ns']-row['syscall_before_ns']<=1000000000,'isolation-host-clock')
                oracle.require(child['kind']=='exec' and child['env_clear'] and not child['helper_mapped'] and not child['control_fd_present'],'isolation-exec-escape')
                oracle.require(child['syscall_before_ns']<=child['clock_ns']<=child['syscall_after_ns'] and child['syscall_after_ns']-child['syscall_before_ns']<=1000000000,'isolation-host-clock')
            target.command('clock')
            for _ in range(3):
                value=target.line();oracle.require(value['kind']=='clock','clock-binding');report['clocks'].append(value['row'])
            system=target.line();oracle.require(system['kind']=='system','clock-binding');report['system_time'].append(system['row'])
            before=kernel_time();clock=time.time_ns();after=kernel_time();oracle.require(before<=clock<=after and after-before<=1000000000 and (not report['host_clocks'] or before>=report['host_clocks'][-1]['syscall_after_ns']),'isolation-host-clock');report['host_clocks'].append({'sequence':sequence,'syscall_before_ns':before,'clock_ns':clock,'syscall_after_ns':after})
        target.command('sleep');sleep=target.line();oracle.require(sleep['kind']=='sleep','sleep-fields');report['sleep']=sleep['row']
        oracle.check_frames(target.frames);oracle.check_probe_clocks(report['clocks'],target.process.pid);oracle.check_sleep(report['sleep'])
        for row,clock in zip(report['system_time'],report['clocks'][::3]):
            oracle.require((row['pid'],row['sequence'],row['offset_s'],row['clock_id'])==(clock['pid'],clock['sequence'],clock['offset_s'],0),'clock-binding')
            oracle.require(row['syscall_before_ns']<=row['raw_ns']<=row['syscall_after_ns'] and row['syscall_after_ns']-row['syscall_before_ns']<=1000000000,'clock-raw-bracket')
            oracle.require(row['returned_ns']==row['raw_ns']+row['offset_s']*1000000000,'clock-realtime-only')
        target.command('quit');oracle.require(target.process.wait(timeout=max(.001,deadline-time.monotonic()))==0,'fixture-probe-exit')
    finally:
        target.close();report['ledger']=target.rows;report['frames']=[frame.hex() for frame in target.frames]
    oracle.require(target.error is None,'fixture-ledger-error')
    for row in report['clocks']+report['system_time']:
        oracle.require({key:value for key,value in row.items() if not key.startswith('syscall_')} in report['ledger'],'clock-ledger-binding')
    report['controls']=[]
    initial=b'FCLKv1\0\0'+bytes(16)
    controls=[('short',initial[:-1],'FCLK-FRAME'),('oversize',initial+b'x','FCLK-FRAME'),('version',b'bad-v001'+bytes(16),'FCLK-FRAME'),
              ('initial-sequence',initial[:8]+(1).to_bytes(8,'little')+bytes(8),'FCLK-SEQUENCE'),('initial-offset',initial[:16]+(1).to_bytes(8,'little'),'FCLK-OFFSET'),
              ('skip',None,'FCLK-SEQUENCE'),('duplicate',None,'FCLK-SEQUENCE'),('decrease',None,'FCLK-OFFSET'),('cap',None,'FCLK-OFFSET'),('overflow',None,'FCLK-OFFSET'),
              ('count',None,'FCLK-COUNT'),('death',None,'FCLK-EOF'),('closed',None,'FCLK-RECV'),('descriptor',None,'FCLK-FD'),('type',None,'FCLK-TYPE'),('unpreloaded',None,'isolation-interception')]
    for name,packet,expected in controls:
        error_path=output/(name+'.stderr');current=SessionClock(executable,helper,error_path,deadline,initial=packet,socket_type=socket.SOCK_STREAM if name=='type' else socket.SOCK_SEQPACKET,preload=name!='unpreloaded',invalid_fd=name=='descriptor')
        try:
            if name=='unpreloaded':
                oracle.require(current.line()=={'kind':'absent','pid':current.process.pid},'fixture-unpreloaded')
                try:process_binding(current.process.pid,executable,helper)
                except ValueError as error:oracle.require(str(error)==expected,'fixture-wrong-refusal')
                else:raise ValueError('fixture-unpreloaded-admitted')
            else:
                if packet is None and name not in ('type','descriptor'):
                    current.line()
                    if name=='death':current.parent.shutdown(socket.SHUT_WR)
                    elif name=='decrease':current.parent.send(initial[:8]+(1).to_bytes(8,'little')+(1).to_bytes(8,'little'));current.parent.send(initial[:8]+(2).to_bytes(8,'little')+bytes(8))
                    else:
                        values=[(i,0) for i in range(1,65)] if name=='count' else [(2 if name=='skip' else 0 if name=='duplicate' else 1,28861 if name=='cap' else 2**64-1 if name=='overflow' else 0)]
                        if name!='closed':
                            for seq,value in values:current.parent.send(initial[:8]+seq.to_bytes(8,'little')+value.to_bytes(8,'little'))
                    current.command('close' if name=='closed' else 'clock')
                oracle.require(current.process.wait(timeout=max(.001,deadline-time.monotonic()))==78,'fixture-control-exit')
        finally:current.close()
        stderr=error_path.read_text();oracle.require(name=='unpreloaded' or stderr.strip()==expected,'fixture-wrong-refusal')
        report['controls'].append({'case':name,'refusal':expected,'exit':current.process.returncode,'stderr':stderr,'ledger':current.rows})


def main():
    parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--output',type=Path,required=True);parser.add_argument('--rustc',required=True);parser.add_argument('--cc',default='gcc');args=parser.parse_args()
    require_limits();oracle.require(platform.system()=='Linux' and platform.machine()=='x86_64','fixture-platform')
    output=args.output;oracle.require(output.is_absolute() and output.resolve()==output and output.is_relative_to(STORAGE),'fixture-output');output.mkdir(mode=0o700)
    scratch=Path(os.environ['FABRIC_SCRATCH_ROOT']);oracle.require(scratch.is_dir() and scratch.is_relative_to(STORAGE),'fixture-scratch')
    directory=Path(tempfile.mkdtemp(prefix='session-clock-',dir=scratch));deadline=time.monotonic()+240;report={'scope':'native binding only; actual packaged browser session checks pending','state':'failed','exit':1,'commands':[]}
    def run(command):
        report['commands'].append(command);value=subprocess.run(command,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=max(.001,deadline-time.monotonic()))
        oracle.require(len(value.stdout)<=1048576,'fixture-tool-output');report.setdefault('tool_outputs',[]).append({'command':command,'exit':value.returncode,'output':value.stdout.decode(errors='replace')});oracle.require(value.returncode==0,'fixture-tool-exit');return value.stdout.decode()
    try:
        compiler=Path(shutil.which(args.cc)).resolve();rustc=Path(shutil.which(args.rustc)).resolve();version=run([str(rustc),'-vV'])
        oracle.require('release: 1.99.0' in version and 'commit-hash: b940084d7' in version,'fixture-toolchain')
        run([str(compiler),'--version']);helper=directory/'session-clock.so';executable=directory/'session-clock-probe'
        run([str(compiler),'-shared','-fPIC','-pthread','-O2','-Wall','-Wextra','-Werror','-o',str(helper),str(HERE/'session_clock.c'),'-ldl'])
        run([str(rustc),'--edition=2021','-C','opt-level=1','-o',str(executable),str(HERE/'session_clock_probe.rs')])
        dynamic=run(['readelf','--dyn-syms','--wide',str(executable)]);oracle.require('clock_gettime@' in dynamic,'fixture-dynamic-clock-import')
        interpreter=run(['readelf','-lW',str(executable)]).split('Requesting program interpreter: ',1)[1].split(']',1)[0]
        report['hashes']={str(path):digest(path) for path in (compiler,rustc,helper,executable,Path(interpreter),HERE/'session_clock.py',HERE/'session_clock.c',HERE/'session_clock_probe.rs')}
        oracle.require(sum(path.stat().st_size for path in directory.rglob('*') if path.is_file())<=512*1024**2,'fixture-storage-bound')
        native_admission(executable,helper,output,deadline,report)
        report['build_id_notes']={str(path):run(['readelf','-nW',str(path)]) for path in (helper,executable,Path(interpreter),Path(report['target']['libc']))};report.update(state='completed',exit=0)
    except BaseException as error:report['error']=type(error).__name__+': '+str(error);raise
    finally:
        shutil.rmtree(directory);report['cleanup']={'owned_fixture':str(directory),'absent':not directory.exists()};(output/'receipt.json').write_text(json.dumps(report,indent=2)+'\n');print(str(output/'receipt.json'))
    return report['exit']


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