1"""Workspace composition (controller → node), collection (node → controller) and artefact hashing.2The workspace never contains the oracle; it contains the public bundle, COMPLETION_SCHEMA.json, PROMPT.md, SYSTEM.md,3(A3/A4) PORTAL.md, (A4) tools/selfcheck.py."""4from __future__ import annotations5import hashlib, json, os, shutil, tempfile6from .config import REPO, NODE_HOME, NODE_PYTHON7from .arms import ARMS, system_prompt, USER_PROMPT8from . import launcher910PUBLIC_EXCLUDES = ("generate.py", "__pycache__", ".DS_Store")1112def scenario_dir(task_id: str) -> str:13 return os.path.join(REPO, "benchmark", "scenarios", task_id)1415def load_task(task_id: str) -> dict:16 return json.load(open(os.path.join(scenario_dir(task_id), "task.json"), encoding="utf-8"))1718def compose_local(task_id: str, arm: str, run_id: str, portal_url: str | None, portal_user: dict | None) -> str:19 """Build the workspace in a temp dir on the controller and return its path."""20 tmp = tempfile.mkdtemp(prefix=f"wp42ws_{run_id}_")21 src = scenario_dir(task_id)22 for name in os.listdir(src):23 if name in PUBLIC_EXCLUDES: continue24 p = os.path.join(src, name)25 if name == "portal" and not ARMS[arm]["portal"]: continue26 (shutil.copytree if os.path.isdir(p) else shutil.copy2)(p, os.path.join(tmp, name))27 shutil.copy2(os.path.join(REPO, "benchmark", "COMPLETION_SCHEMA.json"), os.path.join(tmp, "COMPLETION_SCHEMA.json"))28 # The worker's task.json must not reveal the design (planted errors, injection, depth, seed): keep only what a worker legitimately needs.29 t = json.load(open(os.path.join(tmp, "task.json"), encoding="utf-8"))30 public = {k: t[k] for k in ("task_id", "occupation", "organisation", "title", "version", "required_deliverables", "applications_involved") if k in t}31 json.dump(public, open(os.path.join(tmp, "task.json"), "w"), indent=1)32 os.makedirs(os.path.join(tmp, "out"), exist_ok=True)33 open(os.path.join(tmp, "PROMPT.md"), "w").write(USER_PROMPT)34 open(os.path.join(tmp, "SYSTEM.md"), "w").write(system_prompt(arm))35 if ARMS[arm]["portal"]:36 u = portal_user or {"username": "analyst", "password": "analyst"}37 open(os.path.join(tmp, "PORTAL.md"), "w").write(f"# Internal enterprise portal\n\nURL: {portal_url}\nUsername: {u['username']}\nPassword: {u['password']}\n\nPages: /documents (library, downloads), /records/<entity> and /api/<entity> (JSON), /forms/<form_id> (HTML form; or POST JSON to the same URL). Health: /health. Log in first (POST /login with username/password; the session cookie is then required).\n")38 if ARMS[arm]["recovery"]:39 os.makedirs(os.path.join(tmp, "tools"), exist_ok=True)40 shutil.copy2(os.path.join(REPO, "benchmark", "tools", "selfcheck.py"), os.path.join(tmp, "tools", "selfcheck.py"))41 # Public bundle hash manifest (what the worker saw)42 open(os.path.join(tmp, "BUNDLE_MANIFEST.json"), "w").write(json.dumps(hash_tree(tmp, skip=("out",)), indent=1))43 return tmp4445def hash_tree(root: str, skip=()) -> dict:46 out = {}47 for dp, dns, fns in os.walk(root):48 dns[:] = [d for d in dns if d not in skip]49 for fn in sorted(fns):50 p = os.path.join(dp, fn); h = hashlib.sha256()51 with open(p, "rb") as f:52 for ch in iter(lambda: f.read(1 << 20), b""): h.update(ch)53 out[os.path.relpath(p, root)] = {"sha256": h.hexdigest(), "bytes": os.path.getsize(p)}54 return out5556def node_workspace(run_id: str) -> str:57 return f"{NODE_HOME}/runs/{run_id}"5859def push(node: str, local_ws: str, run_id: str) -> str:60 ws = node_workspace(run_id)61 launcher.ssh(node, f"mkdir -p {ws}", check=True)62 launcher.rsync_to(node, local_ws, ws)63 return ws6465def collect(node: str, run_id: str, dest_dir: str) -> dict:66 """Copy out/ back, hash it, and verify the inputs were not modified (compare with the manifest)."""67 ws = node_workspace(run_id)68 launcher.rsync_from(node, f"{ws}/out", os.path.join(dest_dir, "out"))69 n_remote = launcher.ssh(node, f"find {ws}/out -type f | wc -l").stdout.strip()70 n_local = sum(len(f) for _, _, f in os.walk(os.path.join(dest_dir, "out")))71 if n_remote.isdigit() and int(n_remote) != n_local:72 raise launcher.LaunchError(f"collection mismatch on {node}: {n_remote} remote files vs {n_local} local")73 launcher.rsync_from(node, f"{ws}/inputs", os.path.join(dest_dir, "inputs_after"))74 r = launcher.ssh(node, f"cat {ws}/BUNDLE_MANIFEST.json")75 try: manifest = json.loads(r.stdout)76 except Exception: manifest = {} # noqa: BLE00177 after = hash_tree(os.path.join(dest_dir, "inputs_after"))78 changed = [k for k, v in manifest.items() if k.startswith("inputs/") and after.get(k[len("inputs/"):], {}).get("sha256") != v["sha256"]]79 arts = hash_tree(os.path.join(dest_dir, "out"))80 shutil.rmtree(os.path.join(dest_dir, "inputs_after"), ignore_errors=True)81 return {"artifacts": arts, "inputs_changed": changed, "manifest": manifest}8283def reset(node: str, run_id: str) -> None:84 ws = node_workspace(run_id)85 launcher.ssh(node, f"rm -rf {ws}")8687def start_portal(node: str, run_id: str, port: int) -> str:88 """Start the portal on the node (background, loopback) and return the URL after a health check."""89 ws = node_workspace(run_id)90 cmd = (f"cd {ws} && (PYTHONPATH={NODE_HOME}/tools nohup {NODE_PYTHON} -m wp42.portal --workspace {ws} --port {port} "91 f"> {ws}/.portal.log 2>&1 & echo $! > {ws}/.portal.pid); sleep 2; curl -s -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{port}/health")92 r = launcher.ssh(node, cmd, timeout=60)93 if "200" not in r.stdout: raise launcher.LaunchError(f"portal did not start on {node}:{port}: {r.stdout[-200:]} {r.stderr[-300:]}")94 return f"http://127.0.0.1:{port}"9596def stop_portal(node: str, run_id: str) -> None:97 ws = node_workspace(run_id)98 launcher.ssh(node, f"test -f {ws}/.portal.pid && kill $(cat {ws}/.portal.pid) 2>/dev/null; true")99100def sync_tools(node: str) -> None:101 """Install the controller package (for the portal only) under ~/wp42/tools on the node."""102 launcher.ssh(node, f"mkdir -p {NODE_HOME}/tools/wp42", check=True)103 launcher.rsync_to(node, os.path.join(REPO, "src", "wp42"), f"{NODE_HOME}/tools/wp42", excludes=("__pycache__",))104