#!/usr/bin/env python3
"""U23 public, real-process acceptance. Only owned files/processes on loopback."""
import hashlib, json, os, pathlib, platform, re, subprocess, tempfile, time
from datetime import datetime
ROOT = pathlib.Path(__file__).resolve().parent
JAVA_HOME = pathlib.Path(os.environ.get('JAVA_HOME') or subprocess.check_output(['/usr/libexec/java_home','-v','21'],text=True).strip())
ENV = dict(os.environ, JAVA_HOME=str(JAVA_HOME), PATH=str(JAVA_HOME/'bin')+os.pathsep+os.environ.get('PATH',''))
for key in list(ENV):
    if key.startswith(('LAB_NOTICE_', 'SPRING_APPLICATION_JSON','SPRING_PROFILES_ACTIVE','SPRING_CONFIG_')): ENV.pop(key)
VERSION = subprocess.check_output([str(JAVA_HOME/'bin/java'),'-version'],stderr=subprocess.STDOUT,text=True,env=ENV)
assert re.search(r'version "21[.]',VERSION),VERSION
LOGS=pathlib.Path.home()/'.cache/software-engineering-study/java-depth-impl/u23/runtime';LOGS.mkdir(parents=True,exist_ok=True)
RESULTS=[]
ACTIVE_INVOCATION=None
def observe(record):
    record["observedAt"]=datetime.now().astimezone().isoformat()
    record["invocationId"]=ACTIVE_INVOCATION
    RESULTS.append(record)
def command(name,args,expected=0,env=None,cwd=None,markers=()):
    started_at=datetime.now().astimezone().isoformat()
    p=subprocess.run([str(a) for a in args],env=env or ENV,cwd=cwd or ROOT,text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=180)
    (LOGS/(name+'.log')).write_text(p.stdout)
    assert p.returncode==expected,(name,p.returncode,p.stdout[-5000:])
    for m in markers: assert m in p.stdout,(name,m,p.stdout[-5000:])
    safe_args=[]
    hide_next=False
    for item in args:
        item=str(item)
        if hide_next:safe_args.append('<synthetic-redacted>');hide_next=False
        elif item in ('-storepass','-keypass'):safe_args.append(item);hide_next=True
        elif '.password=' in item:safe_args.append(item.split('=',1)[0]+'=<synthetic-redacted>')
        else:safe_args.append(item)
    record={'name':name,'exitCode':p.returncode,'markers':list(markers),'startedAt':started_at,'finishedAt':datetime.now().astimezone().isoformat(),'input':safe_args,'invocationId':ACTIVE_INVOCATION}
    if env:record['syntheticEnvironment']={key:env[key] for key in ['LAB_NOTICE_MESSAGE','LAB_NOTICE_CLIENTID','SPRING_APPLICATION_JSON'] if key in env}
    result_line=next((line[7:] for line in p.stdout.splitlines() if line.startswith('RESULT ')),None)
    if result_line is not None:record['observed']=json.loads(result_line)
    RESULTS.append(record)
    print('PASS',name,flush=True)
    return p.stdout
JAR=ROOT/'target/boot-runtime-1.jar'
def expect_source(data,kind):
    prop=data['effective']['lab.notice.message']
    assert prop['origin'] not in ('','null','None'),prop
    expected={'base':'application.properties','profile':'application-prod.properties','external-base':'application.properties','external-profile':'application-prod.properties','environment':'systemEnvironment','system':'systemProperties','application-json':'spring.application.json','command-line':'commandLineArgs','location-replaces':'replacement.properties','remove-overrides':'application-prod.properties'}.get(kind,'application.properties')
    assert expected in prop['source'],(kind,prop)
    if kind.startswith('external'):assert 'file [' in prop['source'] and 'boot-runtime-1.jar' not in prop['origin'],prop
def config(name,extra=(),env=None,expected='base',jvm=()):
    output=command(name,[JAVA_HOME/'bin/java',*jvm,'-jar',JAR,*extra],env=env,markers=['RESULT '])
    data=json.loads(next(line[7:] for line in output.splitlines() if line.startswith('RESULT ')))
    assert data['bound']['message']==expected,(name,data)
    assert data['effective']['lab.notice.message']['value']==expected,(name,data)
    expect_source(data,name)
    RESULTS[-1]['observed']=data
    return data
import contextlib, signal, ssl, urllib.request, urllib.error
@contextlib.contextmanager
def server(name,extra=(),env=None):
    path=LOGS/(name+'.log')
    with path.open('w') as log:
        launched_at=datetime.now().astimezone().isoformat()
        p=subprocess.Popen([str(JAVA_HOME/'bin/java'),'-jar',str(JAR),'--lab.mode=server',*extra],env=env or ENV,cwd=ROOT,stdout=log,stderr=subprocess.STDOUT,text=True)
        try:
            end=time.monotonic()+20
            while time.monotonic()<end:
                text=path.read_text()
                match=re.search(r'BOUND (\d+)',text)
                if match and 'EVENT ready' in text:break
                if p.poll() is not None:raise AssertionError((name,p.returncode,text[-5000:]))
                time.sleep(.05)
            else:raise AssertionError((name,'readiness timed out',path.read_text()[-5000:]))
            url=('https' if any(a.startswith('--server.ssl.bundle=') for a in extra) else 'http')+'://localhost:'+match[1]
            yield url,p
        finally:
            if p.poll() is None:
                p.terminate()
                try:p.wait(timeout=12)
                except subprocess.TimeoutExpired:
                    p.kill();p.wait(timeout=5)
                    raise AssertionError((name,'required forced kill during cleanup'))
            assert p.poll() is not None,(name,'unreaped process')
    observe({'name':name+'-process','exitCode':p.returncode,'cleaned':True,'startedAt':launched_at,'finishedAt':datetime.now().astimezone().isoformat()})
def request(url,path='/lab/state',data=None,headers=None,context=None):
    req=urllib.request.Request(url+path,data=data,headers=headers or {})
    try:
        with urllib.request.urlopen(req,timeout=8,context=context) as response:return response.status,response.read()
    except urllib.error.HTTPError as e:return e.code,e.read()

def configuration_cases():
    config('base')

    # EX1: each process reads real files/env/system properties; no MapPropertySource substitutes.
    profile=config('profile',['--spring.profiles.active=prod'],expected='profile')
    assert profile['bound']['labels']==['profile-only'],profile
    assert profile['bound']['routes']=={'base':'base-route','shared':'profile-shared','profile':'profile-route'},profile
    with tempfile.TemporaryDirectory(prefix='u23-config-') as temp:
        work=pathlib.Path(temp)
        (work/'application.properties').write_text('lab.notice.message=external\n')
        (work/'application-prod.properties').write_text('lab.notice.message=external-profile\n')
        extra=['--spring.config.additional-location=file:'+str(work)+'/', '--spring.profiles.active=prod']
        config('external-profile',extra,expected='external-profile')
        config('external-base',[extra[0]],expected='external')
        env=dict(ENV,LAB_NOTICE_MESSAGE='env',LAB_NOTICE_CLIENTID='env-client')
        d=config('environment',extra,env,expected='env')
        assert d['bound']['clientId']=='env-client',d
        system=[JAVA_HOME/'bin/java','-Dlab.notice.message=system','-jar',JAR,*extra]
        output=command('system',system,env=env,markers=['RESULT '])
        d=json.loads(next(x[7:] for x in output.splitlines() if x.startswith('RESULT ')))
        assert d['bound']['message']==d['effective']['lab.notice.message']['value']=='system',d
        expect_source(d,'system')
        RESULTS[-1]['observed']=d
        env_json=dict(env,SPRING_APPLICATION_JSON=json.dumps({'lab':{'notice':{'message':'json'}}}))
        output=command('application-json',system,env=env_json,markers=['RESULT '])
        d=json.loads(next(x[7:] for x in output.splitlines() if x.startswith('RESULT ')))
        assert d['bound']['message']==d['effective']['lab.notice.message']['value']=='json',d
        expect_source(d,'application-json')
        RESULTS[-1]['observed']=d
        config('command-line',extra+['--lab.notice.message=cli'],env_json,expected='cli',jvm=['-Dlab.notice.message=system'])
        # Removing high-priority sources restores file/profile result in a fresh process.
        config('remove-overrides',extra,expected='external-profile')
        full=work/'replacement.properties'
        full.write_text('spring.autoconfigure.exclude=org.springframework.boot.jdbc.autoconfigure.DataSourceAutoConfiguration\nlab.notice.message=replacement\nlab.notice.token=synthetic\nlab.notice.timeout=750ms\nlab.notice.workers=1\nlab.notice.nested.region=replacement-region\n')
        d=config('location-replaces',['--spring.config.location=file:'+str(full)],expected='replacement')
        assert d['bound']['labels'] is None and d['bound']['routes'] is None,d
        assert d['bound']['timeoutMillis']==750 and d['bound']['limitBytes']==1024,d
        missing_token=work/'no-token.properties'
        missing_token.write_text('\n'.join(line for line in full.read_text().splitlines() if not line.startswith('lab.notice.token='))+'\n')
        missing_output=command('missing-required-field',[JAVA_HOME/'bin/java','-jar',JAR,'--spring.config.location=file:'+str(missing_token)],expected=1,markers=['BindValidationException','lab.notice.token','Value: \"null\"','EVENT prepared'])
        assert 'EVENT started' not in missing_output and 'EVENT runner' not in missing_output,missing_output
        missing=work/'missing/'
        output=command('missing-required-location',[JAVA_HOME/'bin/java','-jar',JAR,'--spring.config.additional-location=file:'+str(missing)+'/'],expected=1,markers=['ConfigDataLocationNotFoundException','EVENTS [environment-event-before-postprocessors, failed]'])
        assert 'EVENT prepared' not in output and 'EVENT started' not in output,output
        config('optional-location-recovery',['--spring.config.additional-location=optional:file:'+str(missing)+'/'])
    for name,key,value,root_cause in [
        ('blank-required-field','token','','BindValidationException'),
        ('invalid-duration','timeout','abc','ConversionFailedException'),
        ('constraint-min','workers','0','BindValidationException'),
        ('nested-validation','nested.region','','BindValidationException'),
    ]:
        output=command(name,[JAVA_HOME/'bin/java','-jar',JAR,'--lab.notice.'+key+'='+value],expected=1,markers=[root_cause,'EVENT prepared','ConfigurationPropertiesBindException'])
        assert 'EVENT started' not in output and 'EVENT ready' not in output,output
    config('binding-recovery',['--lab.notice.timeout=750ms','--lab.notice.workers=3'])

def task_cases():
    # Public Boot task executor: bounded platform rejection, unbounded queue, virtual branch, user/force.
    command('platform-bounded',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=tasks','--spring.task.execution.pool.core-size=1','--spring.task.execution.pool.max-size=1','--spring.task.execution.pool.queue-capacity=1'],markers=['RESULT ','"completed":2'])
    command('platform-unbounded',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=tasks','--lab.task-case=unbounded','--spring.task.execution.pool.core-size=1','--spring.task.execution.pool.max-size=4'],markers=['RESULT ','"completed":3','"queuedWhileHeld":2'])
    command('virtual-bounded',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=tasks','--spring.threads.virtual.enabled=true','--spring.task.execution.pool.core-size=9','--spring.task.execution.simple.concurrency-limit=1','--spring.task.execution.simple.reject-tasks-when-limit-reached=true'],markers=['RESULT ','"workerVirtual":true','"completed":1','"concurrencyLimit":1'])
    command('user-executor-backoff',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=tasks','--lab.user-executor=true'],markers=['RESULT ','"userBackoff":true','"applicationTaskExecutor":false'])
    command('force-with-user',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=tasks','--lab.user-executor=true','--spring.task.execution.mode=force','--spring.task.execution.pool.core-size=1','--spring.task.execution.pool.max-size=1','--spring.task.execution.pool.queue-capacity=1'],markers=['RESULT ','"userExecutorPresent":true','"completed":2'])
    command('cron-invalid',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.cron-invalid=true'],expected=1,markers=['initialDelay','cron','EVENT prepared'])
    config('cron-recovery',['--lab.cron-invalid=false'])
    command('mysql-pool',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=db','--spring.autoconfigure.exclude=','--spring.datasource.url=jdbc:mysql://127.0.0.1:3306/?sslMode=DISABLED&allowPublicKeyRetrieval=true','--spring.datasource.username=root','--spring.datasource.hikari.maximum-pool-size=1','--spring.datasource.hikari.minimum-idle=0','--spring.datasource.hikari.connection-timeout=300'],markers=['RESULT ','"recovered":true','SQLTransientConnectionException'])

def http_cases():
    with server('http-container') as (url,process):
        status,body=request(url)
        assert status==200,(status,body)
        state=json.loads(body)
        assert state['connector']['maxThreads']==4 and state['connector']['maxConnections']==16,state
        observe({'name':'real-http-state','exitCode':0,'observed':state})
    # Structured logging is a logging format, independent of HTTP entity JSON.
    structured=command('structured-log',[JAVA_HOME/'bin/java','-jar',JAR,'--logging.structured.format.console=logstash'],markers=['RESULT '])
    log_lines=[json.loads(line) for line in structured.splitlines() if line.startswith('{')]
    assert any('U23_LOG_PROBE' in x.get('message','') for x in log_lines),log_lines
    assert 'synthetic-secret-not-for-output' not in structured
    suppressed=command('logging-level',[JAVA_HOME/'bin/java','-jar',JAR,'--logging.structured.format.console=logstash','--logging.level.study.runtime=error'],markers=['RESULT '])
    assert 'U23_LOG_PROBE' not in suppressed,suppressed
    with server('json-strict',['--spring.jackson.deserialization.fail-on-unknown-properties=true','--spring.jackson.default-property-inclusion=non_null']) as (url,process):
        status,body=request(url,'/lab/json',b'{"value":"ok","unexpected":1}',{'Content-Type':'application/json'})
        assert status==400,(status,body)
        status,body=request(url,'/lab/json',b'{"value":"ok"}',{'Content-Type':'application/json'})
        assert status==200 and json.loads(body)=={'value':'ok'},(status,body)
        observe({'name':'jackson3-strict-inclusion','exitCode':0,'observed':{'unknownStatus':400,'validStatus':200,'reply':json.loads(body)}})

def multipart_cases():
    with server('json-relaxed') as (url,process):
        status,body=request(url,'/lab/json',b'{"value":"ok","unexpected":1}',{'Content-Type':'application/json'})
        assert status==200 and json.loads(body)=={'value':'ok','optional':None},(status,body)
        def multipart(size):
            boundary='u23-bounded-boundary'
            data=('--'+boundary+'\r\nContent-Disposition: form-data; name="file"; filename="sample.txt"\r\nContent-Type: text/plain\r\n\r\n').encode()+b'x'*size+('\r\n--'+boundary+'--\r\n').encode()
            return request(url,'/lab/upload',data,{'Content-Type':'multipart/form-data; boundary='+boundary})
        small_status,body=multipart(128);assert small_status==200 and json.loads(body)['bytes']==128,(small_status,body)
        large_status,body=multipart(2048);assert large_status==413,(large_status,body)
        assert 'MaxUploadSizeExceededException' in (LOGS/'json-relaxed.log').read_text(), 'multipart rejection was not the expected size exception'
        _,state=request(url);assert json.loads(state)['uploads']==1,state
        error_status,_=request(url,'/lab/error');assert error_status==500,error_status
        _,trace=request(url,'/lab/trace');trace=json.loads(trace)
        assert any('REQUEST /lab/error' in item for item in trace),trace
        assert any('ERROR /error original=/lab/error' in item for item in trace),trace
        observe({'name':'multipart-and-error-dispatch','exitCode':0,'observed':{'smallBytes':128,'smallStatus':small_status,'largeBytes':2048,'largeStatus':large_status,'controllerCalls':1,'errorStatus':error_status,'dispatches':trace}})

def tls_cases():
    # Real TLS: isolated certificate material, no global trust changes, never a -k success path.
    import socket
    with tempfile.TemporaryDirectory(prefix='u23-tls-') as temp:
        tls=pathlib.Path(temp);store=tls/'server.p12';certificate=tls/'server.pem';password='u23-synthetic-password'
        command('keytool-generate',[JAVA_HOME/'bin/keytool','-genkeypair','-alias','u23','-keyalg','RSA','-keysize','2048','-validity','2','-storetype','PKCS12','-keystore',store,'-storepass',password,'-keypass',password,'-dname','CN=localhost','-ext','SAN=dns:localhost,ip:127.0.0.1'])
        command('keytool-export',[JAVA_HOME/'bin/keytool','-exportcert','-rfc','-alias','u23','-keystore',store,'-storepass',password,'-file',certificate])
        tls_args=['--server.ssl.bundle=u23','--spring.ssl.bundle.jks.u23.keystore.location=file:'+str(store),'--spring.ssl.bundle.jks.u23.keystore.type=PKCS12','--spring.ssl.bundle.jks.u23.keystore.password='+password]
        trust=ssl.create_default_context(cafile=str(certificate))
        with server('tls-server',tls_args) as (url,process):
            untrusted=False
            try:request(url)
            except urllib.error.URLError as error:
                assert isinstance(error.reason,ssl.SSLCertVerificationError),repr(error)
                untrusted=True
            assert untrusted,'default trust unexpectedly accepted self-signed cert'
            status,body=request(url,context=trust);assert status==200,(status,body)
            host_mismatch=False
            port=int(url.rsplit(':',1)[1])
            try:
                with socket.create_connection(('127.0.0.1',port),timeout=3) as plain:
                    with trust.wrap_socket(plain,server_hostname='wrong-host.example'):pass
            except ssl.SSLCertVerificationError:host_mismatch=True
            assert host_mismatch,'hostname checking was bypassed'
            observe({'name':'tls-trust-hostname','exitCode':0,'observed':{'defaultTrustRejected':True,'explicitCAStatus':status,'wrongHostnameRejected':True}})
        wrong=[a for a in tls_args if '.password=' not in a]+['--spring.ssl.bundle.jks.u23.keystore.password=incorrect']
        command('tls-wrong-password',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=server',*wrong],expected=1,markers=['password was incorrect','EVENT prepared'])
        with server('tls-password-recovery',tls_args) as (url,process):
            status,_=request(url,context=trust);assert status==200,status
    assert not tls.exists(),'owned certificate directory leaked'
    observe({'name':'tls-directory-cleanup','exitCode':0,'removed':True})

def shutdown_cases():
    # Real in-flight request + SIGTERM, observe normal completion and phase-budget overrun separately.
    import concurrent.futures
    for name,millis,budget in [('graceful-normal',1200,'4s'),('graceful-over-budget',4500,'300ms')]:
        extra=['--spring.lifecycle.timeout-per-shutdown-phase='+budget,'--logging.level.org.springframework.boot.tomcat.GracefulShutdown=info']
        with server(name,extra) as (url,process):
            def slow_request():
                try:
                    status,body=request(url,'/lab/slow?millis='+str(millis))
                    return {'status':status,'body':body.decode()}
                except (OSError,TimeoutError) as error:return {'error':type(error).__name__,'message':str(error)}
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as client:
                future=client.submit(slow_request)
                deadline=time.monotonic()+4
                while time.monotonic()<deadline:
                    _,body=request(url)
                    if json.loads(body)['active']==1:break
                    time.sleep(.02)
                else:raise AssertionError('slow request did not enter handler')
                start=time.monotonic();process.send_signal(signal.SIGTERM)
                refused=False
                while time.monotonic()-start<1:
                    try:request(url)
                    except (OSError,TimeoutError):refused=True;break
                    time.sleep(.02)
                assert refused,(name,'new connections were not rejected after shutdown began')
                process.wait(timeout=12)
                observed=future.result(timeout=10)
                elapsed=round((time.monotonic()-start)*1000)
            log=(LOGS/(name+'.log')).read_text()
            if name=='graceful-normal':
                assert observed.get('status')==200 and json.loads(observed['body'])=={'completed':True},observed
                assert 'SLOW COMPLETED' in log and 'Graceful shutdown complete' in log,log[-3000:]
            else:
                assert 'within timeout of 300ms' in log or 'Graceful shutdown aborted' in log,log[-5000:]
            assert process.returncode in (0,143),(name,process.returncode)
            observe({'name':name,'exitCode':process.returncode,'observed':{'requestMillis':millis,'phaseBudget':budget,'stopElapsedMillis':elapsed,'newConnectionRejected':refused,'inFlight':observed,'handlerCompleted':'SLOW COMPLETED' in log,'phaseBudgetExpired':'within timeout of 300ms' in log or 'Graceful shutdown aborted' in log}})

def scheduler_case():
    command('scheduler-explicit-pool',[JAVA_HOME/'bin/java','-jar',JAR,'--lab.mode=tasks','--spring.task.execution.pool.core-size=1','--spring.task.execution.pool.max-size=1','--spring.task.execution.pool.queue-capacity=1','--spring.task.scheduling.pool.size=3'],markers=['RESULT ','"schedulingPoolSize":3','"completed":2'])

if __name__=='__main__':
    import argparse
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--group',choices=['all','config','tasks','http','multipart','tls','shutdown','scheduler'],default='all')
    parser.add_argument('--skip-build',action='store_true',help='Use the existing fixed-version archive for assertion-only reruns')
    args=parser.parse_args()
    ACTIVE_INVOCATION=datetime.now().astimezone().isoformat()
    if not args.skip_build:command('build',['mvn','-q','clean','package'])
    assert JAR.is_file(),'Build first; --skip-build requires an existing archive'
    groups={'config':configuration_cases,'tasks':task_cases,'http':http_cases,'multipart':multipart_cases,'tls':tls_cases,'shutdown':shutdown_cases,'scheduler':scheduler_case}
    for name,action in groups.items():
        if args.group in ('all',name):action()
    if args.group=='all':
        command('dependency-tree',['mvn','dependency:tree','-Dincludes=org.springframework.boot:*,org.springframework:spring-core,tools.jackson.core:*,org.apache.tomcat.embed:*,com.zaxxer:HikariCP,com.mysql:mysql-connector-j'],markers=['spring-boot:jar:4.0.9','spring-core:jar:7.0.10','HikariCP:jar:7.0.2','tomcat-embed-core:jar:11.0.27'])
    (ROOT/'evidence').mkdir(exist_ok=True)
    evidence_path=ROOT/'evidence/u23-2026-10-26.json'
    if args.group!='all' and evidence_path.exists():
        previous=json.loads(evidence_path.read_text())
        for record in previous['results']:
            record.setdefault('sourceInvocationCompletedAt',previous['generatedAt'])
        for record in RESULTS:record['sourceInvocationId']=ACTIVE_INVOCATION
        by_name={record['name']:record for record in previous['results']}
        by_name.update({record['name']:record for record in RESULTS})
        merged=list(by_name.values())
    else:merged=RESULTS
    # Backfill observations from the original successful raw logs when a group only
    # strengthens assertions; this does not pretend that an unchanged case was rerun.
    for record in merged:
        raw=LOGS/(record['name']+'.log')
        if not raw.is_file():continue
        record['rawLogSha256AtCollection']=hashlib.sha256(raw.read_bytes()).hexdigest()
        record['hashCollectedAt']=datetime.now().astimezone().isoformat()
        record['rawLog']=str(raw)
        if 'observed' not in record:
            line=next((line[7:] for line in raw.read_text().splitlines() if line.startswith('RESULT ')),None)
            if line is not None:record['observed']=json.loads(line)
    evidence={'ticket':'U23 #60','generatedAt':datetime.now().astimezone().isoformat(),'environment':{'os':platform.platform(),'arch':platform.machine(),'jdk':VERSION.strip(),'boot':'4.0.9','framework':'7.0.10'},'input':'JAVA_HOME=$(/usr/libexec/java_home -v21) python3 labs/boot-runtime/run.py','latestInvocation':{'id':ACTIVE_INVOCATION,'group':args.group,'skipBuild':args.skip_build,'finishedAt':datetime.now().astimezone().isoformat()},'artifactSha256':hashlib.sha256(JAR.read_bytes()).hexdigest(),'runnerSha256':hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(),'exitCode':0,'rawLogs':str(LOGS),'results':merged,'limits':['Loopback real Tomcat/MySQL/TLS only; not production capacity, K8s ingress propagation or distributed topology','MySQL is SELECT1/borrow timeout only: no transaction/lock/recovery claims','Client disconnect from over-budget shutdown is observed, not a rollback or persisted-task guarantee','WebClient/Jetty/Netty/native/production cert rotation are outside this run']}
    evidence_path.write_text(json.dumps(evidence,ensure_ascii=False,indent=2)+'\n')
    print('PASS U23 '+args.group+', evidence written',flush=True)