m4
d4b53bfb08
Docker / build (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.12) (push) Has been cancelled
Test / pytest (windows-latest, 3.11) (push) Has been cancelled
Test / pytest (windows-latest, 3.12) (push) Has been cancelled
Lint / ruff (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.11) (push) Has been cancelled
Build / build (push) Has been cancelled
41 lines
2.2 KiB
Python
41 lines
2.2 KiB
Python
"""Isolated five-attempt budget for the new cross-Turn barrier target only."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
repo = Path(__file__).resolve().parents[2]
|
|
root = repo.parent / ".hermes/test-runtime/unified-execution/b04-checkpoint-writer"
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
node = "tests/test_checkpoint_writer_race.py::test_cross_turn_checkpoint_writer_barrier"
|
|
ledger = root / "attempts.jsonl"
|
|
records = [json.loads(line) for line in ledger.read_text().splitlines()] if ledger.exists() else []
|
|
attempt = 1 + sum(r.get("phase") == "start" for r in records)
|
|
assert attempt <= 5, "target budget exhausted"
|
|
home = root / f"home-{attempt}"
|
|
home.mkdir(mode=0o700)
|
|
env = {"HOME": str(home), "PATH": str(repo / ".venv/bin") + ":/usr/bin:/bin",
|
|
"PYTHONPATH": str(repo), "PYTHON_DOTENV_DISABLED": "1", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
|
|
"PYTHONDONTWRITEBYTECODE": "1", "EVOSCIENTIST_HOME": str(home),
|
|
"EVOSCIENTIST_CONFIG_DIR": str(home / "config"), "XDG_CONFIG_HOME": str(home / "config"),
|
|
"EVOSCIENTIST_DATA_DIR": str(home / "data"), "EVOSCIENTIST_WORKSPACE_DIR": str(home / "workspace"),
|
|
"EVOSCIENTIST_SKILLS_DIR": str(home / "skills"), "EVOSCIENTIST_MEMORIES_DIR": str(home / "memory")}
|
|
command = [str(repo / ".venv/bin/python"), "-m", "pytest", "--noconftest", "-p", "no:cacheprovider",
|
|
"-q", "-s", "--tb=short", node]
|
|
with ledger.open("a") as out:
|
|
if attempt == 1:
|
|
out.write(json.dumps(dict(phase="registered", node=node, used=0, limit=5)) + "\n")
|
|
out.write(json.dumps(dict(phase="start", node=node, attempt=attempt, stage=sys.argv[1], command=command)) + "\n")
|
|
try:
|
|
result = subprocess.run(command, cwd=repo, env=env, capture_output=True, text=True, timeout=60)
|
|
except subprocess.TimeoutExpired as exc:
|
|
output = exc.stdout or b""
|
|
result = subprocess.CompletedProcess(command, 124, output.decode() if isinstance(output, bytes) else output, "timeout")
|
|
log = root / f"attempt-{attempt}.log"
|
|
log.write_text(result.stdout + result.stderr)
|
|
os.chmod(log, 0o600)
|
|
with ledger.open("a") as out:
|
|
out.write(json.dumps(dict(phase="result", node=node, attempt=attempt, exit_code=result.returncode, log=str(log))) + "\n")
|
|
print(result.stdout + result.stderr)
|
|
sys.exit(result.returncode) |