scripts/run_workflow_execution_cpu.py

"""Paired actual disposable tool execution in owned offline CPU containers."""
import hashlib,json,os,subprocess,time
from pathlib import Path


def main():
    env=dict(os.environ)
    for k in ('DOCKER_HOST','DOCKER_CONTEXT','DOCKER_TLS','DOCKER_TLS_VERIFY','DOCKER_CERT_PATH'):env.pop(k,None)
    D=['docker','--host=unix:///var/run/docker.sock']
    def run(args,**kw):return subprocess.run(D+args,env=env,capture_output=True,text=True,check=True,**kw).stdout
    image='needle-gateway-v2:workflow-qualification'
    for label in ('before','after'):
        cid=run(['run','-d','--cpus','2','--memory','16g','--network','none','--entrypoint','sleep',image,'infinity']).strip()
        try:
            checked={}
            for name in ('gateway.py','models.py','registry.py','laya.py','context_adapter.py','decision_policy.py','request_limits.py','intent_guard.py'):
                source=subprocess.check_output(['git','show','ef25f06:'+name]) if label=='before' else Path(name).read_bytes()
                if label=='before':
                    import tempfile
                    with tempfile.NamedTemporaryFile() as f:
                        f.write(source);f.flush();run(['cp',f.name,cid+':/srv/'+name])
                actual=run(['exec',cid,'sha256sum','/srv/'+name]).split()[0]
                assert actual==hashlib.sha256(source).hexdigest()
                checked[name]=actual
            if label=='after':
                for added in ('payload_transport.py','native_schema.py'):
                    actual=run(['exec',cid,'sha256sum','/srv/'+added]).split()[0]
                    assert actual==hashlib.sha256(Path(added).read_bytes()).hexdigest();checked[added]=actual
            for folder in ('scripts','benchmarks'):run(['cp',folder,cid+':/srv/'+folder])
            model=run(['exec',cid,'python','-c',"import hashlib,json,pathlib; p=pathlib.Path('/root/.cache/cactus-needle/v3/3.0.1'); print(json.dumps({n:hashlib.sha256((p/n).read_bytes()).hexdigest() for n in ('libneedle.so','needle3.cact')}))"])
            expected_assets={line.split()[1]:line.split()[0] for line in Path('needle-assets.sha256').read_text().splitlines() if line.strip()}
            assert json.loads(model)==expected_assets,'Official asset mismatch'
            start=time.perf_counter()
            # getrusage belongs to the process hosting TestClient/native inference;
            # child pytest/Git RSS is separate, not an invented server measurement.
            probe="import scripts.workflow_cpu_app;import json,resource;from scripts.coverage_workflow import run;d=run();d['host_self_peak_rss_kib']=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss;print(json.dumps(d))"
            data=json.loads(run(['exec',cid,'env','-u','NEEDLE3_LIB_PATH','python','-c',probe],timeout=300))
            data.update(total_wall_ms=(time.perf_counter()-start)*1000,verified_runtime_source_sha256=checked,model_assets_sha256=json.loads(model),image_id=run(['image','inspect','--format','{{.Id}}',image]).strip(),git_version=run(['exec',cid,'git','--version']).strip(),resources={'cpu_quota':2,'memory_gib':16,'network':'none','published_ports':[],'executor_in_same_cgroup':True,'sdk_date_fact':'date: 2026-10-17 Sat 00:00','clock_control_sha256':hashlib.sha256(Path('scripts/workflow_cpu_app.py').read_bytes()).hexdigest()})
            data['methodology']='Actual native gateway and allowlisted disposable Git/pytest execution inside the same 2-CPU/16-GiB offline cgroup. TestClient protocol-shaped HTTP; real MCP is qualified separately. Identical original workflow and test image, exact runtime source overlaid for baseline. Self peak RSS excludes subprocesses.'
            Path('benchmarks/workflow/results/'+label+'-workflow-cpu.json').write_text(json.dumps(data,indent=2)+'\n')
            print(label,data['workflow_success'],data['verification'],flush=True)
        finally:run(['rm','-f',cid])

if __name__=='__main__':main()