55 lines
2.6 KiB
Python
55 lines
2.6 KiB
Python
"""arm-runner: arm commit 을 worktree 로 격리 체크아웃(clean 유지), external workspace 에 brief 주입,
|
|
10-step 의미단계 시퀀스를 stage별 별도 process 로 실행(대화 미상속, 원장+Accepted 만 소비). 외부웹은
|
|
evidence-pack 으로 봉인, HUMAN gate 는 사전승인 receipt(전 arm 동일)로 통과."""
|
|
import os
|
|
import subprocess
|
|
|
|
STAGES = [
|
|
{"name": "ground", "command": "ground"},
|
|
{"name": "decide", "command": "decide"},
|
|
{"name": "design-direction", "command": "design-direction"},
|
|
{"name": "design-system-dryrun", "command": "design-system", "dry-run": True},
|
|
]
|
|
|
|
|
|
def evidence_env(controller_evidence_dir):
|
|
return {"BENCHMARK_EVIDENCE_PACK": controller_evidence_dir, "ORGOS_EXTERNAL_WEB": "denied"}
|
|
|
|
|
|
def human_receipt(run_id, brief_sha, arm_ids):
|
|
return {"decision-policy": "pre-authorized-for-benchmark",
|
|
"accepted-scope": {"benchmark-run-id": run_id, "brief-sha256": brief_sha, "arm-ids": list(arm_ids)},
|
|
"forbidden": ["external-side-effect", "deployment", "real-purchase",
|
|
"account-change", "prod-resource-create"]}
|
|
|
|
|
|
def worktree_clean(worktree):
|
|
r = subprocess.run(["git", "status", "--porcelain"], cwd=worktree, capture_output=True, text=True)
|
|
return r.returncode == 0 and r.stdout.strip() == ""
|
|
|
|
|
|
def setup_worktree(run_id, arm, commit, root):
|
|
from . import paths
|
|
wt = paths.worktree_dir(run_id, arm)
|
|
os.makedirs(os.path.dirname(wt), exist_ok=True)
|
|
subprocess.run(["git", "worktree", "add", "--detach", wt, commit],
|
|
cwd=root, capture_output=True, text=True, check=True)
|
|
return wt
|
|
|
|
|
|
def run_stage(worktree, workspace, stage, env, exec_fn):
|
|
"""stage 를 별도 process 로 실행(exec_fn 주입 — 실제는 claude CLI, 테스트는 mock). 산출물·exit-code
|
|
기록. 실패(exit!=0)면 호출부가 다음 stage 를 진행하지 않는다(억지 진행 금지)."""
|
|
from . import probe
|
|
cmd_path = os.path.join(worktree, ".claude", "commands", f"{stage['command']}.md")
|
|
body = open(cmd_path, encoding="utf-8").read() if os.path.exists(cmd_path) else f"# /{stage['command']}"
|
|
brief = os.path.join(workspace, "brief.md")
|
|
inv = probe.build_stage_invocation(stage["command"], body, brief)
|
|
full_env = dict(os.environ); full_env.update(env); full_env["ORGOS_WORKSPACE"] = workspace
|
|
if stage.get("dry-run"):
|
|
full_env["ORGOS_DRY_RUN"] = "true"
|
|
res = exec_fn(inv["argv"], worktree, full_env)
|
|
return {"stage": stage["name"], "exit-code": res.get("exit-code", 0),
|
|
"artifacts": res.get("artifacts", []), "retries": res.get("retries", 0),
|
|
"transcript": res.get("transcript", [])}
|