Expose pinned 64-request per GPU migration helpers

eval-queue-64-5f912bc.py

@@ -0,0 +1,445 @@
+#!/usr/bin/env python3
+"""Two independent GPU workers; tier barriers; bounded resumable internal eval."""
+import argparse
+import concurrent.futures as cf
+from collections import deque
+import contextlib
+import fcntl
+import json
+import os
+from pathlib import Path
+import signal
+import statistics
+import subprocess
+import sys
+import tempfile
+import time
+import threading
+import urllib.request
+import contract_eval as E
+import queue_data as D
+from context_check import CONTEXT_LENGTH
+
+
+class Deadline(Exception): pass
+
+
+def atomic_json(path,value):
+    tmp=path.with_suffix(path.suffix+".tmp")
+    tmp.write_bytes(E.encoded(value)); os.replace(tmp,path)
+
+
+@contextlib.contextmanager
+def locked(path,blocking=True):
+    with path.open("a+") as f:
+        try: fcntl.flock(f,fcntl.LOCK_EX | (0 if blocking else fcntl.LOCK_NB))
+        except BlockingIOError:
+            yield None; return
+        try: yield f
+        finally: fcntl.flock(f,fcntl.LOCK_UN)
+
+
+def journal(path):
+    rows={}
+    if not path.exists(): return rows
+    end=0
+    with path.open("rb") as f:
+        while line:=f.readline():
+            try: row=json.loads(line)
+            except ValueError:
+                if f.read(1): raise ValueError("corrupt interior journal line")
+                if line.endswith(b"\n"): raise ValueError("corrupt complete journal line")
+                # Preserve only an interrupted final write, never silently lose a full row.
+                tail=path.with_name(path.name+".interrupted-tail")
+                if tail.exists(): raise ValueError("inspect previous interrupted tail first")
+                tail.write_bytes(line)
+                with path.open("r+b") as repair: repair.truncate(end)
+                break
+            if row["id"] in rows: raise ValueError("duplicate journal ID")
+            rows[row["id"]]=row; end=f.tell()
+    if rows and path.stat().st_size:
+        with path.open("rb") as f:
+            f.seek(-1,2); missing_newline=f.read(1)!=b"\n"
+        if missing_newline:
+            with path.open("ab") as f: f.write(b"\n"); f.flush(); os.fsync(f.fileno())
+    return rows
+
+
+def append(path,row):
+    with path.open("ab") as f:
+        f.write(E.encoded(row)+b"\n"); f.flush(); os.fsync(f.fileno())
+
+
+def stop_group(process):
+    # Only the process group created by this queue; never pkill/ray stop.
+    try: os.killpg(process.pid,signal.SIGTERM)
+    except ProcessLookupError: return
+    try: process.wait(timeout=3)
+    except subprocess.TimeoutExpired: pass
+    try: os.killpg(process.pid,signal.SIGKILL)
+    except ProcessLookupError: pass
+    try: process.wait(timeout=3)
+    except subprocess.TimeoutExpired: pass
+
+
+def score_job(python,job,deadline):
+    remaining=deadline-time.time()
+    if remaining<=0: raise Deadline()
+    timeout=min(120.,remaining)
+    env={"PATH":"/usr/bin:/bin","OPENBLAS_NUM_THREADS":"1","OMP_NUM_THREADS":"1",
+         "PYTHONDONTWRITEBYTECODE":"1","PYTHONUNBUFFERED":"1","LANG":"C.UTF-8"}
+    with tempfile.TemporaryDirectory(prefix="simct-score-") as work, tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err:
+        env["HOME"]=work; env["TMPDIR"]=work
+        p=subprocess.Popen([python,str(E.HERE/"internal_worker.py")],stdin=subprocess.PIPE,
+                           stdout=out,stderr=err,cwd=work,env=env,start_new_session=True)
+        try:
+            try: p.communicate(E.encoded(job),timeout=timeout)
+            except subprocess.TimeoutExpired:
+                if timeout<120: raise Deadline()
+                return {"passed":False,"timeout":True}
+            out.seek(0); raw=out.read(1024**2)
+            if p.returncode:
+                raise RuntimeError("scorer runtime error: "+raw.decode(errors="replace")[:1000])
+            result=json.loads(raw)
+            if type(result.get("passed")) is not bool: raise ValueError("invalid scorer result")
+            return result
+        finally: stop_group(p)
+
+
+def preflight(python):
+    cases=[("gsm8k",{"gold":"#### 2"},"#### 2","#### 3"),
+           ("math500",{"gold":"2"},r"\boxed{2}",r"\boxed{3}"),
+           ("mbpp",{"tests":["assert add(2,3)==5"],"setup":""},"def add(a,b): return a+b","def add(a,b): return a-b"),
+           ("live-code-bench-v6",{"tests":[{"input":"2 3\n","output":"5\n"}],"fn_name":None},"a,b=map(int,input().split());print(a+b)","print(0)"),
+           ("live-code-bench-v6",{"tests":[{"input":"2\n3","output":"5"}],"fn_name":"add"},"class Solution:\n def add(self,a,b): return a+b","class Solution:\n def add(self,a,b): return a-b")]
+    evidence=[]
+    for benchmark,item,good,bad in cases:
+        for correct,text in ((True,good),(False,bad)):
+            result=score_job(python,{"profile":D.PROFILE,"benchmark":benchmark,"item":item,"text":text},time.time()+150)
+            if result["passed"]!=correct or result.get("timeout"): raise RuntimeError("scorer qualification failed")
+            evidence.append({"benchmark":benchmark,"functional":bool(item.get("fn_name")),"correct":correct,**result})
+    return evidence
+
+
+def state_path(root): return root/"state.json"
+
+
+def read_state(root,plan_hash,plan):
+    path=state_path(root)
+    if path.exists():
+        state=E.read_json(path)
+        if state["plan_sha256"]!=plan_hash: raise ValueError("queue plan changed")
+        return state
+    start=time.time()
+    state={"plan_sha256":plan_hash,"started":start,"deadline":start+plan["hours"]*3600,
+           "admit_until":start+plan["admit_hours"]*3600,"jobs":{},"durations":[]}
+    atomic_json(path,state);return state
+
+
+def next_jobs(plan,state,now):
+    pending=[j for j in plan["jobs"] if state["jobs"].get(j["id"],{}).get("status")!="completed"]
+    if not pending or now>=state["admit_until"]: return []
+    if any(x.get("status")=="failed" for x in state["jobs"].values()):
+        raise RuntimeError("queue has a failed job; inspect error and use retry after fixing it")
+    tier=min(j["tier"] for j in pending)
+    estimate=max(state["durations"],default=0.)
+    if now+estimate>state["deadline"]: return []
+    return [j for j in pending if j["tier"]==tier]
+
+
+def cell_contract(plan_hash,job,data,benchmark,seed,server):
+    return {"plan_sha256":plan_hash,"checkpoint_sha256":job["checkpoint"]["sha256"],
+            "data_sha256":data[benchmark]["sha256"],"benchmark":benchmark,"seed":seed,
+            "profile":D.PROFILE,"server":server}
+
+
+def generate_one(base,item,benchmark,seed,deadline):
+    if time.time()>=deadline: raise Deadline()
+    payload=E.generation_payload("eval-gemma",item,benchmark,seed)
+    request=urllib.request.Request(base+"/v1/chat/completions",data=E.encoded(payload),headers={"Content-Type":"application/json"})
+    opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    with opener.open(request,timeout=min(600.,max(.1,deadline-time.time()))) as response:
+        result=json.load(response)
+    E.validate_response(result)
+    return {"id":item["id"],"seed":seed,"request_sha256":E.digest(E.encoded(payload)),"response":result}
+
+
+def run_cell(root,plan,plan_hash,job,benchmark,seed,base,server,args,deadline):
+    cell=root/"cells"/job["id"]/benchmark/str(seed);cell.mkdir(parents=True,exist_ok=True)
+    contract=cell_contract(plan_hash,job,plan["data"],benchmark,seed,server)
+    manifest=cell/"contract.json"
+    if manifest.exists():
+        if E.read_json(manifest)!=contract: raise ValueError("cell resume contract mismatch")
+    else: E.write_new(manifest,contract)
+    dataset=E.read_json(plan["data"][benchmark]["path"])
+    items={x["id"]:x for x in dataset["items"]}
+    if (cell/"metrics.json").exists():
+        complete=E.read_json(cell/"metrics.json")
+        for name in ("responses","scores"):
+            if E.file_hash(cell/(name+".jsonl"))!=complete[name+"_sha256"]:
+                raise ValueError("completed journal changed; no repair permitted")
+    responses=journal(cell/"responses.jsonl"); scores=journal(cell/"scores.jsonl")
+    if not set(scores)<=set(responses)<=set(items): raise ValueError("orphan/unknown results")
+    for key,row in responses.items():
+        expected=E.generation_payload("eval-gemma",items[key],benchmark,seed)
+        if row["seed"]!=seed or row["request_sha256"]!=E.digest(E.encoded(expected)): raise ValueError("response request mismatch")
+        E.validate_response(row["response"])
+    for key,row in scores.items():
+        if row["response_sha256"]!=E.digest(E.encoded(responses[key])) or type(row.get("passed")) is not bool:
+            raise ValueError("score does not match response")
+    if (cell/"metrics.json").exists():
+        m=E.read_json(cell/"metrics.json")
+        if len(scores)!=len(items) or m["responses_sha256"]!=E.file_hash(cell/"responses.jsonl") or m["scores_sha256"]!=E.file_hash(cell/"scores.jsonl"):
+            raise ValueError("completed cell integrity mismatch")
+        return m
+    started=time.time()
+    previous_seconds=E.read_json(cell/"timing.json").get("seconds",0.) if (cell/"timing.json").exists() else 0.
+    pending=iter([x for x in items.values() if x["id"] not in scores])
+    futures={}
+    def process_item(item):
+        # Data decoding remains in parent, one problem per scoring process.
+        row=responses.get(item["id"])
+        if row is None: row=generate_one(base,item,benchmark,seed,deadline)
+        return row
+    score_buffer=getattr(args,"score_buffer",64)
+    if score_buffer<args.concurrency: raise ValueError("score buffer must cover generation concurrency")
+    backlog=deque()
+    def grade_item(item):
+        # Decode potentially large/private tests only inside an active scorer slot.
+        row=responses[item["id"]]
+        payload={"profile":D.PROFILE,"benchmark":benchmark,"item":D.score_item(benchmark,item),
+                 "text":row["response"]["choices"][0]["message"]["content"]}
+        return score_job(args.score_python,payload,deadline)
+    print("PIPELINE",json.dumps({"generation_concurrency":args.concurrency,
+          "score_workers":args.score_workers,"score_buffer":score_buffer}),flush=True)
+    try:
+        with cf.ThreadPoolExecutor(max_workers=args.concurrency) as generation, cf.ThreadPoolExecutor(max_workers=args.score_workers) as grading:
+            exhausted=False
+            while futures or backlog or not exhausted:
+                if time.time()>=deadline: raise Deadline()
+                scoring=sum(kind=="score" for kind,_ in futures.values())
+                while backlog and scoring<args.score_workers:
+                    item=backlog.popleft()
+                    futures[grading.submit(grade_item,item)]=( "score",item)
+                    scoring+=1
+                generating=sum(kind=="generation" for kind,_ in futures.values())
+                # Reserve backlog capacity for in-flight generation; scorer backlog
+                # no longer consumes generation slots. Backpressure remains bounded.
+                while not exhausted and generating<args.concurrency and len(backlog)+generating<score_buffer:
+                    try: item=next(pending)
+                    except StopIteration: exhausted=True;break
+                    futures[generation.submit(process_item,item)]=("generation",item)
+                    generating+=1
+                if not futures: break
+                done,_=cf.wait(futures,timeout=min(1.,max(.1,deadline-time.time())),return_when=cf.FIRST_COMPLETED)
+                for future in done:
+                    kind,item=futures.pop(future);value=future.result()
+                    if kind=="generation":
+                        if item["id"] not in responses:
+                            append(cell/"responses.jsonl",value);responses[item["id"]]=value
+                        backlog.append(item)
+                    else:
+                        value.update(id=item["id"],response_sha256=E.digest(E.encoded(responses[item["id"]])))
+                        append(cell/"scores.jsonl",value);scores[item["id"]]=value
+                        if len(scores)%25==0: print(f"PROGRESS {job['id']} {benchmark} seed={seed} {len(scores)}/{len(items)}",flush=True)
+    finally:
+        atomic_json(cell/"timing.json",{"seconds":previous_seconds+time.time()-started})
+    if len(scores)!=len(items): raise ValueError("incomplete cell")
+    metrics={"status":"completed","contract":contract,"count":len(items),
+             "score":sum(x["passed"] for x in scores.values())/len(items),
+             "truncated_fraction":sum(x["response"]["choices"][0]["finish_reason"]=="length" for x in responses.values())/len(items),
+             "seconds":previous_seconds+time.time()-started,
+             "responses_sha256":E.file_hash(cell/"responses.jsonl"),"scores_sha256":E.file_hash(cell/"scores.jsonl")}
+    E.write_new(cell/"metrics.json",metrics)
+    print("CELL_COMPLETE",job["id"],benchmark,seed,metrics["score"],flush=True)
+    return metrics
+
+
+def run_checkpoint(root,plan,plan_hash,job,args,deadline):
+    actual=E.checkpoint_identity(job["checkpoint"]["path"])
+    if time.time()>=deadline: raise Deadline("wall budget during checkpoint verification")
+    if actual!=job["checkpoint"]: raise ValueError("checkpoint changed after plan")
+    port=31000+1000*args.gpu;base=f"http://127.0.0.1:{port}"
+    import socket
+    with socket.socket() as sock:
+        if sock.connect_ex(("127.0.0.1",port))==0: raise RuntimeError("server port already occupied")
+    env=dict(os.environ,CUDA_VISIBLE_DEVICES=str(args.gpu),HF_HUB_OFFLINE="1",TRANSFORMERS_OFFLINE="1",
+             HF_DATASETS_OFFLINE="1",OMP_NUM_THREADS="4",TOKENIZERS_PARALLELISM="false",PYTHONUNBUFFERED="1",
+             PYTHONPATH=str(E.HERE.parents[1]/"experiments/modal/vendor")+":"+str(E.HERE.parents[1]))
+    for name in ("HTTP_PROXY","HTTPS_PROXY","ALL_PROXY","http_proxy","https_proxy","all_proxy","WANDB_API_KEY","HF_TOKEN"):
+        env.pop(name,None)
+    env.pop("SGLANG_ALLOW_OVERWRITE_LONGER_CONTEXT_LEN",None)
+    subprocess.run(["bash",str(E.HERE.parents[1]/"experiments/runai/python-b200-host.sh"),
+                    str(E.HERE/"context_check.py"),"--plan",str(args.plan.resolve()),
+                    "--checkpoint",actual["path"]],env=dict(env,CUDA_VISIBLE_DEVICES=""),
+                   check=True,timeout=max(.1,min(300.,deadline-time.time())))
+    if time.time()>=deadline: raise Deadline("wall budget during context qualification")
+    log=root/(job["id"]+f"-gpu{args.gpu}-server.log")
+    command=["bash",str(E.HERE.parents[1]/"experiments/runai/python-b200-host.sh"),"-m","sglang.launch_server",
+             "--model-path",actual["path"],"--served-model-name","eval-gemma","--host","127.0.0.1","--port",str(port),
+             "--tp-size","1","--mem-fraction-static","0.8","--context-length",str(CONTEXT_LENGTH),
+             "--attention-backend","triton","--disable-cuda-graph","--random-seed","42"]
+    with log.open("ab") as output:
+        process=subprocess.Popen(command,env=env,stdout=output,stderr=subprocess.STDOUT,start_new_session=True)
+        watchdog=threading.Timer(max(0.,deadline-time.time()),stop_group,args=(process,))
+        watchdog.daemon=True; watchdog.start()
+        try:
+            startup=min(deadline,time.time()+900)
+            while True:
+                if time.time()>=deadline: raise Deadline("wall budget")
+                if time.time()>=startup: raise RuntimeError("server readiness exceeded 900 seconds; see "+str(log))
+                if process.poll() is not None: raise RuntimeError("SGLang exited; see "+str(log))
+                try:
+                    # Short health timeout; identity verification follows readiness.
+                    opener=urllib.request.build_opener(urllib.request.ProxyHandler({}))
+                    with opener.open(base+"/health",timeout=2): pass
+                    break
+                except (OSError,ValueError): time.sleep(2)
+            server=E.verify_server(base,actual["path"],"eval-gemma")
+            for seed in plan["seeds"]:
+                for benchmark in E.CAPS:
+                    run_cell(root,plan,plan_hash,job,benchmark,seed,base,server,args,deadline)
+            if E.verify_server(base,actual["path"],"eval-gemma")!=server: raise ValueError("server identity drift")
+        finally:
+            watchdog.cancel(); stop_group(process)
+
+
+def worker(args):
+    root=args.plan.resolve().parent;plan=E.read_json(args.plan);plan_hash=E.file_hash(args.plan)
+    if plan["profile"]!=D.PROFILE or plan["source"]!=D.script_hashes(): raise ValueError("queue source/profile changed")
+    for data in plan["data"].values():
+        if E.file_hash(data["path"])!=data["sha256"]: raise ValueError("prepared data changed")
+    qualification=preflight(args.score_python)
+    with locked(root/"state.lock"):
+        state=read_state(root,plan_hash,plan)
+        if "qualification" in state and state["qualification"]!=qualification: raise ValueError("grader version drift")
+        state["qualification"]=qualification;atomic_json(state_path(root),state)
+    gpu_lock=Path(tempfile.gettempdir())/f"simct-eval-gpu{args.gpu}.lock"
+    with locked(gpu_lock,blocking=False) as own:
+        if own is None: raise RuntimeError("another eval worker owns this GPU")
+        while True:
+            with locked(root/"state.lock"):
+                state=read_state(root,plan_hash,plan)
+                candidates=next_jobs(plan,state,time.time())
+                if not candidates:
+                    print("QUEUE_STOP admission/deadline/complete",flush=True);return
+            usage=subprocess.check_output(["nvidia-smi","-i",str(args.gpu),"--query-gpu=memory.used","--format=csv,noheader,nounits"],text=True).strip()
+            if int(usage)>=1024:
+                print("WAIT_GPU",args.gpu,usage,flush=True);time.sleep(15);continue
+            claimed=False
+            for job in candidates:
+                with locked(root/(job["id"]+".lock"),blocking=False) as lock:
+                    if lock is None: continue
+                    with locked(root/"state.lock"):
+                        state=read_state(root,plan_hash,plan)
+                        if job["id"] not in {j["id"] for j in next_jobs(plan,state,time.time())}: continue
+                        state["jobs"][job["id"]]={"status":"running","gpu":args.gpu,"started":time.time()}
+                        atomic_json(state_path(root),state)
+                    started=time.time();claimed=True
+                    try:
+                        run_checkpoint(root,plan,plan_hash,job,args,state["deadline"])
+                        result={"status":"completed","seconds":time.time()-started}
+                    except Deadline as exc:
+                        result={"status":"partial","reason":str(exc) or "wall budget"}
+                    except Exception as exc:
+                        result=({"status":"partial","reason":"wall budget"} if time.time()>=state["deadline"] else
+                                {"status":"failed","error":type(exc).__name__+": "+str(exc)})
+                    with locked(root/"state.lock"):
+                        state=read_state(root,plan_hash,plan);state["jobs"][job["id"]]=result
+                        if result["status"]=="completed": state["durations"].append(result["seconds"])
+                        atomic_json(state_path(root),state)
+                    print("JOB",job["id"],json.dumps(result),flush=True)
+                    if result["status"]=="failed": raise RuntimeError(result["error"])
+                    if result["status"]!="completed": return
+                    break
+            if not claimed: time.sleep(5)
+
+
+def summarize(args):
+    root=args.plan.resolve().parent;plan=E.read_json(args.plan);ph=E.file_hash(args.plan)
+    output={"profile":plan["profile"],"seeds":plan["seeds"],"checkpoints":{},"scope":plan["protocol"]["scope"]}
+    for job in plan["jobs"]:
+        benchmarks={}
+        for benchmark in E.CAPS:
+            paths=[root/"cells"/job["id"]/benchmark/str(seed)/"metrics.json" for seed in plan["seeds"]]
+            if not all(p.exists() for p in paths): continue
+            rows=[E.read_json(p) for p in paths]
+            for seed,row,path in zip(plan["seeds"],rows,paths):
+                c=row["contract"]
+                if c["data_sha256"]!=plan["data"][benchmark]["sha256"] or c["benchmark"]!=benchmark or c["profile"]!=D.PROFILE or row["count"]!=plan["data"][benchmark]["count"]:
+                    raise ValueError("summary data contract mismatch")
+                for name in ("responses","scores"):
+                    if E.file_hash(path.parent/(name+".jsonl"))!=row[name+"_sha256"]:
+                        raise ValueError("summary journal integrity mismatch")
+                if c["plan_sha256"]!=ph or c["seed"]!=seed or c["checkpoint_sha256"]!=job["checkpoint"]["sha256"] or row["status"]!="completed":
+                    raise ValueError("summary cell contract mismatch")
+            values=[x["score"] for x in rows]
+            benchmarks[benchmark]={"mean":statistics.mean(values),"sample_std":statistics.stdev(values),"scores":values}
+        result={"benchmarks":benchmarks,"complete":len(benchmarks)==4}
+        if result["complete"]: result["average"]=statistics.mean(x["mean"] for x in benchmarks.values())
+        output["checkpoints"][job["id"]]=result
+    atomic_json(root/"summary-3seeds.json",output);print(json.dumps(output,indent=2))
+
+
+def recover_startup(args):
+    """Fork only a zero-result failed-startup queue; retain its original clock."""
+    old=args.from_plan.resolve();oldroot=old.parent
+    with locked(oldroot/"state.lock"), contextlib.ExitStack() as locks:
+        plan=E.read_json(old);state=E.read_json(oldroot/"state.json")
+        if state["plan_sha256"]!=E.file_hash(old) or plan["profile"]!=D.PROFILE:
+            raise ValueError("old plan identity mismatch")
+        if state["durations"] or not state["jobs"] or any(x["status"]!="failed" for x in state["jobs"].values()):
+            raise ValueError("recovery requires exclusively failed startup jobs")
+        if list(oldroot.glob("cells/**/*.json*")):
+            raise ValueError("results exist; use an audited migration instead")
+        for job in plan["jobs"]:
+            if locks.enter_context(locked(oldroot/(job["id"]+".lock"),blocking=False)) is None:
+                raise ValueError("old worker still owns a checkpoint")
+        for data in plan["data"].values():
+            if E.file_hash(data["path"])!=data["sha256"]: raise ValueError("old data changed")
+        root=args.out.resolve();root.mkdir(parents=True,exist_ok=False)
+        plan["source"]=D.script_hashes();plan["protocol"]["context_length"]=CONTEXT_LENGTH
+        plan["recovery"]={"from_plan":str(old),"sha256":E.file_hash(old),"reason":"Gemma2 native context startup fix; original clock retained"}
+        E.write_new(root/"plan.json",plan)
+        state.update(plan_sha256=E.file_hash(root/"plan.json"),jobs={},durations=[])
+        E.write_new(root/"state.json",state)
+        print("RECOVERED_PLAN="+str(root/"plan.json"))
+        print("REMAINING_HOURS="+str(max(0.,state["deadline"]-time.time())/3600))
+        print("Original deadline retained; no GPU started.")
+
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__);sub=p.add_subparsers(dest="cmd",required=True)
+    q=sub.add_parser("prepare")
+    q.add_argument("--out",type=Path,required=True)
+    q.add_argument("--shared",type=Path,default=Path("/workspace/storage-shared/nlp/tungks"))
+    q.add_argument("--company-data",type=Path,default=Path("/workspace/storage-shared/nlp/hadv17/lm-eval/benchmarks"))
+    q.add_argument("--proxy",default="http://10.30.154.118:80")
+    q.add_argument("--hours",type=float,default=20.)
+    q.set_defaults(func=D.prepare)
+    q=sub.add_parser("preflight");q.add_argument("--score-python",default="/usr/bin/python3.12");q.set_defaults(func=lambda a:print(json.dumps(preflight(a.score_python),indent=2)))
+    q=sub.add_parser("worker")
+    q.add_argument("--plan",type=Path,required=True);q.add_argument("--gpu",type=int,choices=range(8),required=True)
+    q.add_argument("--score-python",default="/usr/bin/python3.12")
+    q.add_argument("--concurrency",type=int,default=8);q.add_argument("--score-workers",type=int,default=2)
+    q.add_argument("--score-buffer",type=int,default=64,help="Maximum waiting plus in-flight generated items; tests decoded only by active scorers")
+    q.add_argument("--internal-code-execution",action="store_true",required=True,
+                   help="Explicit company-internal profile; resource limits are not an OS sandbox")
+    q.set_defaults(func=worker)
+    q=sub.add_parser("summarize");q.add_argument("--plan",type=Path,required=True);q.set_defaults(func=summarize)
+    q=sub.add_parser("recover-startup");q.add_argument("--from-plan",type=Path,required=True);q.add_argument("--out",type=Path,required=True);q.set_defaults(func=recover_startup)
+    q=sub.add_parser("retry");q.add_argument("--plan",type=Path,required=True)
+    def retry(a):
+        root=a.plan.resolve().parent
+        with locked(root/"state.lock"):
+            s=E.read_json(state_path(root))
+            for key,value in list(s["jobs"].items()):
+                if value["status"]=="failed": del s["jobs"][key]
+            atomic_json(state_path(root),s)
+    q.set_defaults(func=retry)
+    a=p.parse_args()
+    if a.cmd=="prepare" and not 1<a.hours<=20: p.error("budget must be >1 and <=20 hours")
+    if a.cmd=="worker" and not (1<=a.concurrency<=64 and 1<=a.score_workers<=16 and a.concurrency<=a.score_buffer<=256): p.error("invalid concurrency")
+    a.func(a)
+
+if __name__=="__main__": main()

migrate-eval-64-5f912bc.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Stop exact old workers, fork verified journals, add SimCT; never launch GPUs."""
+import argparse, contextlib, copy, hashlib, json, os, shutil, signal, sys, time
+from pathlib import Path
+
+
+def proc(pid):
+    p=Path('/proc')/str(pid)
+    try:
+        raw=(p/'stat').read_text().rsplit(')',1)[1].split()
+        return dict(pid=int(pid),ppid=int(raw[1]),group=int(raw[2]),start=raw[19],state=raw[0],
+                    argv=[x.decode() for x in (p/'cmdline').read_bytes().split(b'\0') if x])
+    except (OSError, UnicodeError): return None
+
+
+def stop_workers(plan,queue,pids):
+    handles=[]; roots=[]
+    try:
+        for pid in pids:
+            info=proc(pid)
+            if info is None: continue
+            args=info['argv']
+            if str(queue) not in args or 'worker' not in args or '--plan' not in args or args[args.index('--plan')+1]!=str(plan):
+                raise ValueError('PID identity mismatch: '+str(pid))
+            fd=os.pidfd_open(pid); again=proc(pid)
+            if not again or again['start']!=info['start']: raise ValueError('PID changed')
+            handles.append(fd); roots.append(info)
+        # Freeze only validated workers so they cannot start a new job during teardown.
+        for fd in handles: signal.pidfd_send_signal(fd,signal.SIGSTOP)
+        table={int(p.name):proc(int(p.name)) for p in Path('/proc').iterdir() if p.name.isdigit()}
+        descendants={r['pid'] for r in roots}
+        while True:
+            more={pid for pid,x in table.items() if x and x['ppid'] in descendants}
+            if more<=descendants: break
+            descendants |= more
+        children=[]
+        for pid in descendants-{r['pid'] for r in roots}:
+            before=table[pid]
+            try: fd=os.pidfd_open(pid)
+            except ProcessLookupError: continue
+            after=proc(pid)
+            if not after or after['start']!=before['start']: os.close(fd);continue
+            children.append(fd)
+        try:
+            for fd in children+handles:
+                try: signal.pidfd_send_signal(fd,signal.SIGTERM)
+                except ProcessLookupError: pass
+            for fd in handles:
+                try: signal.pidfd_send_signal(fd,signal.SIGCONT)
+                except ProcessLookupError: pass
+            limit=time.monotonic()+30
+            while time.monotonic()<limit:
+                alive=[pid for pid in descendants if (x:=proc(pid)) and x['state']!='Z' and x['start']==table[pid]['start']]
+                if not alive: return
+                time.sleep(.5)
+            raise RuntimeError('Some owned processes remain; no migration performed: '+str(alive))
+        finally:
+            for fd in children: os.close(fd)
+    finally:
+        for fd in handles:
+            try: signal.pidfd_send_signal(fd,signal.SIGCONT)
+            except ProcessLookupError: pass
+            os.close(fd)
+
+
+def main():
+    a=argparse.ArgumentParser();a.add_argument('--plan',type=Path,required=True);a.add_argument('--source',type=Path,required=True)
+    a.add_argument('--out',type=Path,required=True);a.add_argument('--worker-pids',nargs='+',type=int)
+    a.add_argument('--queue-update',type=Path)
+    a.add_argument('--simct',type=Path);args=a.parse_args()
+    old=args.plan.resolve(strict=True); source=args.source.resolve(strict=True); out=args.out.resolve()
+    if out.exists() or out.is_relative_to(old.parent): raise ValueError('Output must be a new sibling directory')
+    sys.path.insert(0,str(source/'scripts/evaluation'))
+    import eval_queue as Q
+    E,D=Q.E,Q.D
+    plan=E.read_json(old); oldhash=E.file_hash(old)
+    if plan['source']!=D.script_hashes(): raise ValueError('Source differs from running plan')
+    expected_count=25 if args.queue_update else 17
+    allowed=('sft','atomic','fixed','simct') if args.queue_update else ('sft','atomic','fixed')
+    if len(plan['jobs'])!=expected_count or any(j['mode'] not in allowed for j in plan['jobs']): raise ValueError('Unexpected old jobs')
+    update_bytes=None
+    if args.queue_update:
+        update_bytes=args.queue_update.read_bytes()
+        if plan['source']['eval_queue.py'] not in ('c64478394072d27fff38b432e6976e900a9b9a6a9065105590257ab34e8fa36d','616acf5249f5b7365b3f93ff7a17546c96b9d71ffdf5306b7981a6079bd0ce3e','af7cdaa0fab262f597babc64a17d5e1861ec2fbb35644612a096785962678a51'): raise ValueError('Unsupported original queue version')
+        if hashlib.sha256(update_bytes).hexdigest()!='c4c23d92039237e6a0717f9221f302d4965df98a1e02544e869a3e79987cde6f': raise ValueError('Unsupported queue update')
+    initial_state=E.read_json(old.parent/'state.json')
+    if time.time()>=initial_state['admit_until']: raise ValueError('Admission expired; workers left untouched')
+    for j in plan['jobs']:
+        if j['mode'] in D.RUNS and D.RUNS[j['mode']] not in Path(j['checkpoint']['path']).parts: raise ValueError('Wrong historical MP run')
+    extra=[]
+    if not args.queue_update:
+        if args.simct is None: raise ValueError('--simct required')
+        summary=E.read_json(args.simct/'run-summary.json')
+        if summary['kd_algorithm']!='span_ctkd' or summary['status']!='completed' or summary['optimizer_updates']!=312: raise ValueError('SimCT summary invalid')
+        if summary['student']!=next(j['checkpoint']['path'] for j in plan['jobs'] if j['mode']=='sft'): raise ValueError('Different SFT initialization')
+        # Hash before stopping to avoid wasting idle GPU time on checkpoint inventory.
+        for step in (312,156,80,240,40,200,120,280):
+            print('HASH_SIMCT',step,flush=True)
+            identity=E.checkpoint_identity(args.simct/f'step{step}')
+            tier=next(j['tier'] for j in plan['jobs'] if j['mode']=='atomic' and j['step']==step)
+            extra.append(dict(id=f'simct-{step}',mode='simct',step=step,tier=tier,checkpoint=identity))
+    for data in plan['data'].values():
+        if E.file_hash(data['path'])!=data['sha256']: raise ValueError('Data changed')
+    pids=args.worker_pids
+    if pids is None:
+        pids=[]
+        for entry in Path('/proc').iterdir():
+            if not entry.name.isdigit(): continue
+            info=proc(int(entry.name))
+            if info and 'worker' in info['argv'] and '--plan' in info['argv']:
+                argv=info['argv']
+                if argv[argv.index('--plan')+1]==str(old): pids.append(info['pid'])
+        print('MATCHED_WORKERS',pids,flush=True)
+    stop_workers(old,source/'scripts/evaluation/eval_queue.py',pids)
+    with contextlib.ExitStack() as stack:
+        for gpu in (0,1):
+            if stack.enter_context(Q.locked(Path(f'/tmp/simct-eval-gpu{gpu}.lock'),blocking=False)) is None: raise ValueError('GPU worker still active')
+        stack.enter_context(Q.locked(old.parent/'state.lock'))
+        for j in plan['jobs']:
+            if stack.enter_context(Q.locked(old.parent/(j['id']+'.lock'),blocking=False)) is None: raise ValueError('Job still active')
+        state=E.read_json(old.parent/'state.json')
+        if state['plan_sha256']!=oldhash or E.file_hash(old)!=oldhash: raise ValueError('Plan changed')
+        if time.time()>=state['admit_until']: raise ValueError('Original admission window expired; no implicit extension')
+        out.mkdir()
+        receipt={'from_plan':str(old),'old_plan_sha256':oldhash,'source':str(source),'concurrency':16,'original_files':{},'status':'incomplete'}
+        E.write_new(out/'migration.json',receipt)
+        new=copy.deepcopy(plan);new['jobs']+=extra;new['jobs'].sort(key=lambda j:j['tier'])
+        if update_bytes is not None:
+            target=out/'source'
+            shutil.copytree(source,target,ignore=shutil.ignore_patterns('.git','__pycache__','remote_artifacts'))
+            (target/'scripts/evaluation/eval_queue.py').write_bytes(update_bytes)
+            new['source']={name:E.file_hash(target/'scripts/evaluation'/name) for name in plan['source']}
+            if {k for k in new['source'] if new['source'][k]!=plan['source'][k]}!={'eval_queue.py'}: raise ValueError('Unexpected source changes')
+            receipt.update(source=str(target),previous_source=str(source),concurrency_by_gpu={"0":64,"1":64},score_workers=16,score_buffer=128)
+            receipt.pop('concurrency',None)
+            new['execution_transition']={'generation_concurrency_by_gpu':{'0':64,'1':64},'score_workers':16,'score_buffer':128,'old_source':plan['source']}
+
+        new['migration']={'from_plan':str(old),'sha256':oldhash,'reason':('Decouple generation/scoring; preserve journals and clock' if update_bytes is not None else 'Add eight SimCT checkpoints; preserve journals and original clock')}
+        E.write_new(out/'plan.json',new);newhash=E.file_hash(out/'plan.json')
+        if (old.parent/'cells').exists(): shutil.copytree(old.parent/'cells',out/'cells')
+        jobs={j['id']:j for j in plan['jobs']};datasets={b:{x['id']:x for x in E.read_json(d['path'])['items']} for b,d in plan['data'].items()}
+        totals={'responses':0,'scores':0,'metrics':0}
+        for f in (old.parent/'cells').rglob('*'):
+            if f.is_file(): receipt['original_files'][str(f.relative_to(old.parent))]=E.file_hash(f)
+        for cell in (out/'cells').glob('*/*/*'):
+            if not cell.is_dir(): continue
+            jid,b,seed=cell.relative_to(out/'cells').parts;seed=int(seed)
+            if jid not in jobs or b not in datasets or seed not in plan['seeds']: raise ValueError('Unknown cell')
+            contract=E.read_json(cell/'contract.json')
+            expected=Q.cell_contract(oldhash,jobs[jid],plan['data'],b,seed,contract['server'])
+            if contract!=expected: raise ValueError('Old cell contract mismatch')
+            metrics=E.read_json(cell/'metrics.json') if (cell/'metrics.json').exists() else None
+            if metrics:
+                if metrics['contract']!=contract or metrics['status']!='completed': raise ValueError('Old metrics mismatch')
+                for name in ('responses','scores'):
+                    if E.file_hash(cell/(name+'.jsonl'))!=metrics[name+'_sha256']: raise ValueError('Completed journal changed')
+            responses=Q.journal(cell/'responses.jsonl');scores=Q.journal(cell/'scores.jsonl');items=datasets[b]
+            if not set(scores)<=set(responses)<=set(items): raise ValueError('Unknown result ID')
+            for key,row in responses.items():
+                payload=E.generation_payload('eval-gemma',items[key],b,seed)
+                if row['seed']!=seed or row['request_sha256']!=E.digest(E.encoded(payload)): raise ValueError('Request mismatch')
+                E.validate_response(row['response'])
+            for key,row in scores.items():
+                if type(row.get('passed')) is not bool or row['response_sha256']!=E.digest(E.encoded(responses[key])): raise ValueError('Score mismatch')
+            if metrics:
+                if len(scores)!=len(items) or metrics['count']!=len(items) or metrics['score']!=sum(x['passed'] for x in scores.values())/len(items): raise ValueError('Metric count/score mismatch')
+                metrics['contract']={**contract,'plan_sha256':newhash};Q.atomic_json(cell/'metrics.json',metrics);totals['metrics']+=1
+            Q.atomic_json(cell/'contract.json',{**contract,'plan_sha256':newhash})
+            totals['responses']+=len(responses);totals['scores']+=len(scores)
+        for rel,digest in receipt['original_files'].items():
+            if E.file_hash(old.parent/rel)!=digest: raise ValueError('Source journal changed during migration')
+        state['plan_sha256']=newhash
+        state['jobs']={k:v for k,v in state['jobs'].items() if v['status']=='completed'}
+        for jid in state['jobs']:
+            if not all((out/'cells'/jid/b/str(s)/'metrics.json').exists() for b in E.CAPS for s in plan['seeds']): raise ValueError('Completed job lacks cells')
+        E.write_new(out/'state.json',state)
+        receipt.update(status='verified',new_plan_sha256=newhash,totals=totals,deadline=state['deadline'])
+        Q.atomic_json(out/'migration.json',receipt)
+        print('MIGRATION_PASS',json.dumps(totals),flush=True);print('NEW_PLAN='+str(out/'plan.json'));print('NEW_SOURCE='+receipt['source']);print('No GPU launched. Original deadline retained.')
+
+if __name__=='__main__': main()