#!/usr/bin/env python3
# Evidence only: uses the pre-existing isolated runtime fixture, never the live T3 home.
import os, pathlib, subprocess, time, re, json
# Fixture source dependencies, included verbatim for public reproducibility.
LAUNCH_TS = 'import {main} from "/home/ubuntu/wt/issue-24755/apps/server/src/serviceLauncher.ts"; await main();\n'
RUNTIME_WRAPPER = '#!/bin/sh\nexec /home/ubuntu/.local/share/vite-plus/bin/node /home/ubuntu/wt/issue-24755/apps/server/src/bin.ts serve --port 19855 --host 127.0.0.1 --base-dir /tmp/t3-24755-repro --no-browser\n'
W='/home/ubuntu/wt/issue-24755'
H='/tmp/t3-24755-repro'
N='/home/ubuntu/.local/share/vite-plus/bin/node'
U='t3-24755-evidence.service'
P=pathlib.Path(H+'/userdata/logs/boot-service.log')
OUT=pathlib.Path('/tmp/t3-24755-evidence/runtime-sanitized.txt')
lines=[]
def record(s):
    lines.append(s); OUT.write_text('\n'.join(lines)+'\n'); print(s,flush=True)
def call(args):
    return subprocess.run(args,check=True,capture_output=True,text=True).stdout.strip()
def identity(s): return f'dev={s.st_dev} inode={s.st_ino} bytes={s.st_size}'
def active(): return call(['systemctl','--user','is-active',U])
def healthy():
    # Read locally; publish only the matched Listening line, never pairing URLs or state.
    data=P.read_bytes().decode(errors='replace')
    return re.search(r'(?m)^\[[^\n]{0,100}Listening on http://(?:localhost|127\.0\.0\.1):19855',data)
assert pathlib.Path(H+'/launch.ts').read_text() == LAUNCH_TS
assert pathlib.Path(H+'/runtime/versions/0.1.2/t3').read_text() == RUNTIME_WRAPPER
record('Runtime evidence: actual serviceLauncher.main + actual bin.ts serve; isolated fixture '+H)
record('Head '+call(['git','-C',W,'rev-parse','HEAD']))
record('Environment '+call(['uname','-srmo'])+'; node '+call([N,'--version']))
record('Service manager '+call(['systemctl','--version']).splitlines()[0])
pre=b'T3_24755_STARTUP_PREMARKER_20261025\n'
post=b'T3_24755_STARTUP_POSTMARKER_20261025\n'
second=b'T3_24755_MAINTENANCE_PREMARKER_20261025\n'
after=b'T3_24755_MAINTENANCE_POSTMARKER_20261025\n'
with P.open('wb') as f:
    f.write(pre); f.truncate(17*1024*1024)
fd=os.open(P,os.O_WRONLY|os.O_APPEND)
record('Before startup retained O_APPEND fd: '+identity(os.fstat(fd)))
record('Before startup live path: '+identity(P.stat()))
args=['systemd-run','--user','--unit='+U,'--property=WorkingDirectory='+W,'--property=StandardOutput=append:'+str(P),'--property=StandardError=append:'+str(P),'--setenv=T3CODE_HOME='+H,'--setenv=PATH='+os.environ['PATH'],N,H+'/launch.ts']
record('Command: systemd-run --user --unit='+U+' --property=WorkingDirectory='+W+' --property=StandardOutput=append:'+str(P)+' --property=StandardError=append:'+str(P)+' --setenv=T3CODE_HOME='+H+' --setenv=PATH=[inherited, omitted] '+N+' '+H+'/launch.ts')
started=False
try:
    call(args); started=True
    # Bounded runtime readiness observation, not a test-suite sleep/poll assertion.
    for _ in range(120):
        if healthy(): break
        time.sleep(.5)
    match=healthy(); assert match, 'No healthy Listening observed'
    record('Unit state: '+active())
    record('Healthy log: '+match.group(0))
    live=P.stat(); retained=os.fstat(fd)
    record('After startup retained fd: '+identity(retained))
    record('After startup live path: '+identity(live))
    assert (live.st_dev,live.st_ino)==(retained.st_dev,retained.st_ino)
    archive=pathlib.Path(str(P)+'.1')
    assert archive.read_bytes().startswith(pre)
    assert live.st_size<17*1024*1024
    record('Startup archive: bytes='+str(archive.stat().st_size)+'; premarker verified; live reduced')
    os.write(fd,post)
    assert post in P.read_bytes()
    record('Unique startup postmarker written through retained fd and verified in live path')
    os.write(fd,second)
    os.write(fd,b'X'*(11*1024*1024))
    record('Before maintenance unit state: '+active()+'; live '+identity(P.stat()))
    seam=pathlib.Path('/tmp/t3-24755-evidence/maintenance.ts')
    seam.write_text('import { boundServiceLog } from '+json.dumps(W+'/apps/server/src/serviceLauncher.ts')+'; await boundServiceLog('+json.dumps(H)+');\n')
    record('Command: '+N+' '+str(seam)+' (explicit exported maintenance seam; not timer firing)')
    call([N,str(seam)])
    live=P.stat(); retained=os.fstat(fd)
    record('After maintenance retained fd: '+identity(retained))
    record('After maintenance live path: '+identity(live))
    assert (live.st_dev,live.st_ino)==(retained.st_dev,retained.st_ino)
    data=archive.read_bytes()
    assert post in data and second in data
    assert live.st_size<10*1024*1024
    record('Second archive bytes='+str(len(data))+'; startup postmarker and maintenance premarker verified')
    os.write(fd,after)
    assert after in P.read_bytes()
    record('Unique maintenance postmarker written through original retained fd and verified in live path')
    record('After maintenance unit state: '+active())
    record('Timer wiring inspected: 10 minute setInterval calls boundServiceLog; startup calls same seam. No 10 minute timer wait performed.')
finally:
    os.close(fd)
    if started:
        subprocess.run(['systemctl','--user','stop',U],check=True)
        state=subprocess.run(['systemctl','--user','is-active',U],capture_output=True,text=True).stdout.strip()
        record('Owned unit stopped: '+state)