Add durable NestQuest OpenCode controller and role launchers (#149)
Co-authored-by: orchestrator-agent <orchestrator@cubecraftlabs.com> Co-committed-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
This commit was merged in pull request #149.
This commit is contained in:
Executable
+20
@@ -0,0 +1,20 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
if [[ $# -ne 4 ]]; then
|
||||
echo "usage: nq-agent developer|approver WORKTREE TITLE PROMPT_FILE" >&2
|
||||
exit 64
|
||||
fi
|
||||
|
||||
role=$1
|
||||
worktree=$2
|
||||
title=$3
|
||||
prompt_file=$4
|
||||
case "$role" in developer|approver) ;; *) echo "invalid role: $role" >&2; exit 64 ;; esac
|
||||
[[ -d "$worktree" && -f "$prompt_file" ]] || { echo "missing worktree or prompt" >&2; exit 66; }
|
||||
|
||||
repo_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd -P)
|
||||
export OPENCODE_CONFIG_DIR="$repo_dir/.opencode"
|
||||
opencode_bin=${OPENCODE_BIN:-/home/overseer/.opencode/bin/opencode}
|
||||
|
||||
exec "$opencode_bin" run --pure --dir "$worktree" --agent "$role" --title "$title" --format json < "$prompt_file"
|
||||
Executable
+131
@@ -0,0 +1,131 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Inspect or merge a reviewed NestQuest feature PR into dev only."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
|
||||
API = "https://code.cubecraftlabs.com/api/v1/repos/Maniacs_Mansion/NestQuest"
|
||||
CREDENTIALS = Path("/home/overseer/.llms/credentials.env")
|
||||
FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||
|
||||
|
||||
def token_from_file() -> str:
|
||||
for line in CREDENTIALS.read_text().splitlines():
|
||||
if line.startswith("GITEA_APPROVER_AGENT_TOKEN="):
|
||||
return line.split("=", 1)[1].strip().strip("\"'")
|
||||
raise RuntimeError("GITEA_APPROVER_AGENT_TOKEN is unavailable")
|
||||
|
||||
|
||||
def request(path: str, token: str, payload: dict | None = None) -> dict | list:
|
||||
body = None if payload is None else json.dumps(payload).encode()
|
||||
headers = {"Authorization": "token " + token, "Accept": "application/json"}
|
||||
if body is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
call = urllib.request.Request(API + path, data=body, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(call, timeout=20) as response:
|
||||
raw = response.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
except urllib.error.HTTPError as error:
|
||||
raise RuntimeError(f"Gitea returned HTTP {error.code}") from None
|
||||
|
||||
|
||||
def inspect(pr_number: int, token: str) -> dict:
|
||||
pr = request(f"/pulls/{pr_number}", token)
|
||||
head = (pr.get("head") or {}).get("sha")
|
||||
base = (pr.get("base") or {}).get("ref")
|
||||
if not isinstance(head, str) or not FULL_SHA.fullmatch(head):
|
||||
raise RuntimeError("Gitea returned an invalid PR head")
|
||||
status = request(f"/commits/{head}/status", token)
|
||||
return {
|
||||
"number": pr.get("number"),
|
||||
"base": base,
|
||||
"head": head,
|
||||
"state": pr.get("state"),
|
||||
"merged": pr.get("merged"),
|
||||
"mergeable": pr.get("mergeable"),
|
||||
"check_count": status.get("total_count") or 0,
|
||||
"check_state": status.get("state"),
|
||||
}
|
||||
|
||||
|
||||
def validate_review(path: Path, head: str) -> None:
|
||||
if not path.is_file() or not path.resolve().is_relative_to(Path("/tmp/opencode").resolve()):
|
||||
raise RuntimeError("review evidence must be a file under /tmp/opencode")
|
||||
text = path.read_text(errors="replace")
|
||||
reviewed = re.findall(r"(?m)^REVIEWED_HEAD:\s*([0-9a-f]{40})\s*$", text)
|
||||
verdicts = re.findall(r"(?m)^REVIEW_VERDICT:\s*(APPROVED|CHANGES_REQUIRED)\s*$", text)
|
||||
if len(reviewed) != 1 or len(verdicts) != 1 or reviewed[0] != head or verdicts[0] != "APPROVED":
|
||||
raise RuntimeError("independent review does not approve this exact head")
|
||||
|
||||
|
||||
def has_platform_review(pr_number: int, head: str, token: str) -> bool:
|
||||
reviews = request(f"/pulls/{pr_number}/reviews", token)
|
||||
if not isinstance(reviews, list):
|
||||
raise RuntimeError("Gitea returned invalid review records")
|
||||
by_reviewer = [
|
||||
review for review in reviews
|
||||
if isinstance(review, dict)
|
||||
and (review.get("user") or {}).get("login") == "review_agent"
|
||||
and review.get("commit_id") == head
|
||||
]
|
||||
if not by_reviewer:
|
||||
return False
|
||||
latest = max(by_reviewer, key=lambda review: int(review.get("id") or 0))
|
||||
return latest.get("state") == "APPROVED"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("operation", choices=("inspect", "merge"))
|
||||
parser.add_argument("pr", type=int)
|
||||
parser.add_argument("reviewed_head", nargs="?")
|
||||
parser.add_argument("review_file", nargs="?", type=Path)
|
||||
args = parser.parse_args()
|
||||
if args.pr < 1:
|
||||
parser.error("PR number must be positive")
|
||||
if args.operation == "merge" and (not args.reviewed_head or not args.review_file):
|
||||
parser.error("merge requires reviewed_head and review_file")
|
||||
token = token_from_file()
|
||||
before = inspect(args.pr, token)
|
||||
if args.operation == "inspect":
|
||||
print(json.dumps(before, sort_keys=True))
|
||||
return 0
|
||||
if not FULL_SHA.fullmatch(args.reviewed_head):
|
||||
parser.error("reviewed head must be a full SHA")
|
||||
validate_review(args.review_file, args.reviewed_head)
|
||||
if not has_platform_review(args.pr, args.reviewed_head, token):
|
||||
raise RuntimeError("Gitea lacks an exact-head approval by review_agent")
|
||||
if before["base"] != "dev" or before["head"] != args.reviewed_head:
|
||||
raise RuntimeError("PR target or current head differs from the approved review")
|
||||
if before["state"] != "open" or before["merged"] or before["mergeable"] is not True:
|
||||
raise RuntimeError("PR is closed, merged, or not mergeable")
|
||||
if before["check_count"] and before["check_state"] != "success":
|
||||
raise RuntimeError("a configured commit check has not passed")
|
||||
# Gitea rechecks branch protection at the merge endpoint. The caller must
|
||||
# independently verify Maestro and task-level evidence before invoking us.
|
||||
# Gitea checks head_commit_id atomically while processing the merge.
|
||||
request(f"/pulls/{args.pr}/merge", token,
|
||||
{"do": "squash", "head_commit_id": args.reviewed_head})
|
||||
after = inspect(args.pr, token)
|
||||
if after["merged"] is not True or after["head"] != args.reviewed_head:
|
||||
raise RuntimeError("merge result is missing or differs from the reviewed head")
|
||||
print(json.dumps(after, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
raise SystemExit(main())
|
||||
except (OSError, ValueError, RuntimeError) as error:
|
||||
print(f"approval gate failed: {error}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
Executable
+277
@@ -0,0 +1,277 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Durable, single-owner controller for one NestQuest OpenCode tick at a time."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import fcntl
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import selectors
|
||||
import signal
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
|
||||
DEFAULT_PROJECT = Path("/mnt/ai-storage/projects/NestQuest")
|
||||
DEFAULT_STATE = Path.home() / ".local/state/nestquest-controller"
|
||||
STATE_RE = re.compile(r"(?m)^ORCHESTRATOR_STATE: (CONTINUE|QUIESCENT|BLOCKED)\s*$")
|
||||
MAX_TEXT = 64_000
|
||||
MAX_LINE = 4_000_000
|
||||
CTXD_PROJECT = "nestquest-house-chaos-coordination"
|
||||
CTXD_FILE = "OPENCODE-ORCHESTRATOR-PROMPT.MD"
|
||||
|
||||
|
||||
def check_ctxd() -> int:
|
||||
"""Use OpenCode's configured endpoint, never an unrelated host DB."""
|
||||
config = json.loads((Path.home() / ".config/opencode/opencode.json").read_text())
|
||||
mcp = config["mcp"]["ctxd"]
|
||||
configured_url = mcp["url"].rstrip("/")
|
||||
if not configured_url.endswith("/mcp"):
|
||||
raise RuntimeError("CTXD MCP URL must end in /mcp")
|
||||
endpoint = configured_url.removesuffix("/mcp")
|
||||
url = f"{endpoint}/projects/{CTXD_PROJECT}/files/{CTXD_FILE}"
|
||||
with urllib.request.urlopen(
|
||||
urllib.request.Request(url, headers=mcp["headers"]), timeout=8
|
||||
) as response:
|
||||
record = json.load(response)
|
||||
if (
|
||||
record.get("project_id") != CTXD_PROJECT
|
||||
or record.get("file_path") != CTXD_FILE
|
||||
or "CONTROLLER_TICK" not in record.get("content", "")
|
||||
):
|
||||
raise RuntimeError("CTXD returned the wrong NestQuest prompt")
|
||||
return int(record["version"])
|
||||
|
||||
|
||||
def interactive_nestquest_session(
|
||||
project: Path,
|
||||
proc_root: Path = Path("/proc"),
|
||||
database: Path = Path.home() / ".local/share/opencode/opencode.db",
|
||||
) -> bool:
|
||||
"""Avoid racing a user-owned OpenCode TUI for the same repository."""
|
||||
try:
|
||||
db = sqlite3.connect(f"file:{database}?mode=ro", uri=True)
|
||||
except sqlite3.Error as error:
|
||||
raise RuntimeError("OpenCode session database is unavailable") from error
|
||||
try:
|
||||
for entry in proc_root.iterdir():
|
||||
if not entry.name.isdecimal() or int(entry.name) == os.getpid():
|
||||
continue
|
||||
try:
|
||||
args = (entry / "cmdline").read_bytes().split(b"\0")
|
||||
args = [a.decode(errors="replace") for a in args if a]
|
||||
if not args or Path(args[0]).name != "opencode":
|
||||
continue
|
||||
if len(args) > 1 and args[1] in ("run", "debug"):
|
||||
continue
|
||||
session = None
|
||||
for flag in ("-s", "--session"):
|
||||
if flag in args and args.index(flag) + 1 < len(args):
|
||||
session = args[args.index(flag) + 1]
|
||||
break
|
||||
if session:
|
||||
row = db.execute("SELECT directory FROM session WHERE id=?", (session,)).fetchone()
|
||||
if row and Path(row[0]).resolve() == project.resolve():
|
||||
return True
|
||||
elif "--dir" in args and args.index("--dir") + 1 < len(args):
|
||||
if Path(args[args.index("--dir") + 1]).resolve() == project.resolve():
|
||||
return True
|
||||
elif (entry / "cwd").resolve() == project.resolve():
|
||||
return True
|
||||
except sqlite3.Error as error:
|
||||
raise RuntimeError("OpenCode session lookup failed") from error
|
||||
except TypeError as error:
|
||||
raise RuntimeError("OpenCode session directory is invalid") from error
|
||||
except (OSError, PermissionError):
|
||||
continue
|
||||
finally:
|
||||
db.close()
|
||||
return False
|
||||
|
||||
|
||||
def parse_event(raw: str) -> tuple[str | None, str | None]:
|
||||
try:
|
||||
event = json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
return None, None
|
||||
if not isinstance(event, dict):
|
||||
return None, None
|
||||
kind = event.get("type")
|
||||
part = event.get("part") or {}
|
||||
if kind == "text" and isinstance(part, dict):
|
||||
return part.get("text"), None
|
||||
if kind == "error":
|
||||
error = event.get("error") or {}
|
||||
known = {"ProviderAuthError", "UnknownError", "MessageOutputLengthError", "MessageAbortedError", "StructuredOutputError", "ContextOverflowError", "APIError"}
|
||||
name = error.get("name") if isinstance(error, dict) else None
|
||||
return None, name if name in known else "OpenCode error"
|
||||
return None, None
|
||||
|
||||
|
||||
def run_tick(command: list[str], env: dict[str, str], prompt: str, timeout: int) -> tuple[int, str, str | None]:
|
||||
process = subprocess.Popen(
|
||||
command,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
env=env,
|
||||
start_new_session=True,
|
||||
)
|
||||
assert process.stdin is not None and process.stdout is not None
|
||||
try:
|
||||
process.stdin.write(prompt.encode())
|
||||
except BrokenPipeError:
|
||||
pass
|
||||
finally:
|
||||
process.stdin.close()
|
||||
selector = selectors.DefaultSelector()
|
||||
selector.register(process.stdout, selectors.EVENT_READ)
|
||||
deadline = time.monotonic() + timeout
|
||||
buffer = b""
|
||||
texts: list[str] = []
|
||||
last_error: str | None = None
|
||||
diagnostic_lines = 0
|
||||
try:
|
||||
while selector.get_map():
|
||||
if time.monotonic() >= deadline:
|
||||
try:
|
||||
os.killpg(process.pid, signal.SIGKILL)
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
process.wait()
|
||||
return 124, "".join(texts)[-MAX_TEXT:], "controller tick exceeded documented timeout"
|
||||
for key, _ in selector.select(timeout=1):
|
||||
chunk = os.read(key.fd, 65536)
|
||||
if not chunk:
|
||||
selector.unregister(key.fileobj)
|
||||
continue
|
||||
buffer += chunk
|
||||
while b"\n" in buffer:
|
||||
line, buffer = buffer.split(b"\n", 1)
|
||||
text, error = parse_event(line.decode(errors="replace"))
|
||||
if not line.lstrip().startswith(b"{"):
|
||||
diagnostic_lines += 1
|
||||
if text:
|
||||
texts.append(text)
|
||||
if sum(map(len, texts)) > MAX_TEXT:
|
||||
texts = ["".join(texts)[-MAX_TEXT:]]
|
||||
if error:
|
||||
last_error = error
|
||||
if len(buffer) > MAX_LINE:
|
||||
buffer = b"" # A tool-output event is not a controller verdict.
|
||||
if buffer:
|
||||
text, error = parse_event(buffer.decode(errors="replace"))
|
||||
if text:
|
||||
texts.append(text)
|
||||
if error:
|
||||
last_error = error
|
||||
code = process.wait()
|
||||
if code and last_error is None and diagnostic_lines:
|
||||
last_error = f"OpenCode emitted {diagnostic_lines} non-JSON diagnostic lines"
|
||||
return code, "".join(texts)[-MAX_TEXT:], last_error
|
||||
finally:
|
||||
selector.close()
|
||||
process.stdout.close()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--project", type=Path, default=DEFAULT_PROJECT)
|
||||
parser.add_argument("--state-dir", type=Path, default=DEFAULT_STATE)
|
||||
parser.add_argument("--once", action="store_true", help="run only one model turn")
|
||||
parser.add_argument("--check", action="store_true", help="validate paths without calling a model")
|
||||
parser.add_argument("--check-interactive", action="store_true", help="report an open NestQuest TUI without calling a model")
|
||||
parser.add_argument("--tick-timeout", type=int, default=6 * 3600)
|
||||
args = parser.parse_args()
|
||||
|
||||
project = args.project.resolve()
|
||||
control_home = Path(__file__).resolve().parent.parent
|
||||
config_dir = control_home / ".opencode"
|
||||
binary = Path(os.environ.get("OPENCODE_BIN", "/home/overseer/.opencode/bin/opencode"))
|
||||
if not (project.is_dir() and config_dir.is_dir() and binary.is_file()):
|
||||
parser.error("project, agent configuration, or OpenCode binary is missing")
|
||||
if args.check:
|
||||
try:
|
||||
ctxd_version = check_ctxd()
|
||||
except (OSError, KeyError, ValueError, RuntimeError) as error:
|
||||
print(f"CTXD preflight failed: {type(error).__name__}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"ready: project={project} configuration={config_dir} binary={binary} ctxd_prompt_version={ctxd_version}")
|
||||
return 0
|
||||
if args.check_interactive:
|
||||
try:
|
||||
active = interactive_nestquest_session(project)
|
||||
except RuntimeError:
|
||||
print("OpenCode session state unavailable", file=sys.stderr)
|
||||
return 77
|
||||
print("interactive NestQuest session open" if active else "no interactive NestQuest session")
|
||||
return 0
|
||||
|
||||
args.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
os.chmod(args.state_dir, 0o700)
|
||||
with (args.state_dir / "controller.lock").open("w") as lock:
|
||||
try:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except BlockingIOError:
|
||||
print("another NestQuest controller is running", file=sys.stderr)
|
||||
return 75
|
||||
command = [str(binary), "run", "--pure", "--dir", str(project), "--agent", "orchestrator", "--format", "json", "--title", "NestQuest controller tick"]
|
||||
env = os.environ.copy()
|
||||
env["OPENCODE_CONFIG_DIR"] = str(config_dir)
|
||||
env["NESTQUEST_CONTROL_HOME"] = str(control_home)
|
||||
failures = 0
|
||||
while True:
|
||||
try:
|
||||
interactive = interactive_nestquest_session(project)
|
||||
except RuntimeError:
|
||||
print("OpenCode session state unavailable; waiting", file=sys.stderr, flush=True)
|
||||
if args.once:
|
||||
return 77
|
||||
time.sleep(60)
|
||||
continue
|
||||
if interactive:
|
||||
print("interactive NestQuest session still open; waiting", flush=True)
|
||||
if args.once:
|
||||
return 75
|
||||
time.sleep(60)
|
||||
continue
|
||||
prompt = (
|
||||
"Run one NestQuest controller tick. Read CTXD and Maestro afresh, "
|
||||
"reconcile Git and PR state, then execute the next eligible action. "
|
||||
"Use NESTQUEST_CONTROL_HOME for role launch scripts. "
|
||||
"Return a final ORCHESTRATOR_STATE line as your agent instructions require."
|
||||
)
|
||||
status, answer, error = run_tick(command, env, prompt, args.tick_timeout)
|
||||
states = STATE_RE.findall(answer)
|
||||
state = states[-1] if status == 0 and error is None and states else "ERROR"
|
||||
if state == "CONTINUE":
|
||||
failures = 0
|
||||
elif state == "QUIESCENT":
|
||||
failures = 0
|
||||
elif state == "BLOCKED":
|
||||
failures = 0
|
||||
else:
|
||||
failures += 1
|
||||
summary = {"time": int(time.time()), "state": state, "exit_code": status, "error": error, "failure_count": failures, "answer_tail": answer[-4000:]}
|
||||
target = args.state_dir / "last-run.json"
|
||||
temporary = args.state_dir / "last-run.json.tmp"
|
||||
temporary.write_text(json.dumps(summary, indent=2) + "\n")
|
||||
os.chmod(temporary, 0o600)
|
||||
temporary.replace(target)
|
||||
print(f"tick state={state} exit={status} failures={failures}", flush=True)
|
||||
if args.once:
|
||||
return 0 if state in ("CONTINUE", "QUIESCENT") else 77
|
||||
if state == "BLOCKED" or failures >= 3:
|
||||
print("controller blocked; inspect last-run.json", file=sys.stderr)
|
||||
return 77
|
||||
time.sleep(300 if state == "QUIESCENT" else min(120, 10 * 2**failures))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Executable
+111
@@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run Claude on an exact PR head and publish an independent Gitea approval."""
|
||||
from __future__ import annotations
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
API = "https://code.cubecraftlabs.com/api/v1"
|
||||
REPO = "/repos/Maniacs_Mansion/NestQuest"
|
||||
CREDENTIALS = Path("/home/overseer/.llms/credentials.env")
|
||||
CLAUDE = "/home/overseer/.local/bin/claude"
|
||||
SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||
|
||||
def git(worktree: Path, *args: str) -> str:
|
||||
return subprocess.check_output(["git", "-C", str(worktree), *args], text=True).strip()
|
||||
|
||||
def token() -> str:
|
||||
for line in CREDENTIALS.read_text().splitlines():
|
||||
if line.startswith("GITEA_REVIEW_AGENT_TOKEN="):
|
||||
return line.split("=", 1)[1].strip().strip("\"'")
|
||||
raise RuntimeError("review_agent credential is unavailable")
|
||||
|
||||
def api(path: str, credential: str, payload: dict | None = None) -> dict:
|
||||
body = None if payload is None else json.dumps(payload).encode()
|
||||
headers = {"Authorization": "token " + credential, "Accept": "application/json"}
|
||||
if body is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
call = urllib.request.Request(API + path, data=body, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(call, timeout=20) as response:
|
||||
return json.loads(response.read())
|
||||
except urllib.error.HTTPError as error:
|
||||
raise RuntimeError(f"Gitea returned HTTP {error.code}") from None
|
||||
|
||||
def verdict(output: str, head: str) -> str:
|
||||
reviewed = re.findall(r"(?m)^REVIEWED_HEAD:\s*([0-9a-f]{40})\s*$", output)
|
||||
decisions = re.findall(r"(?m)^REVIEW_VERDICT:\s*(APPROVED|CHANGES_REQUIRED)\s*$", output)
|
||||
if reviewed != [head] or len(decisions) != 1:
|
||||
raise RuntimeError("Claude did not return one verdict for the exact reviewed head")
|
||||
return decisions[0]
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("worktree", type=Path)
|
||||
parser.add_argument("base_ref")
|
||||
parser.add_argument("head_ref")
|
||||
parser.add_argument("prompt_file", type=Path)
|
||||
parser.add_argument("pr", type=int)
|
||||
parser.add_argument("output_file", type=Path)
|
||||
args = parser.parse_args()
|
||||
if not args.worktree.is_dir() or not args.prompt_file.is_file() or args.pr < 1:
|
||||
parser.error("valid worktree, prompt file, and PR number required")
|
||||
if not args.output_file.resolve().is_relative_to(Path("/tmp/opencode").resolve()):
|
||||
parser.error("review output must be under /tmp/opencode")
|
||||
base = git(args.worktree, "rev-parse", "--verify", args.base_ref + "^{commit}")
|
||||
head = git(args.worktree, "rev-parse", "--verify", args.head_ref + "^{commit}")
|
||||
if not SHA.fullmatch(head) or git(args.worktree, "rev-parse", "HEAD") != head:
|
||||
raise RuntimeError("worktree HEAD differs from requested head")
|
||||
subprocess.run(["git", "-C", str(args.worktree), "merge-base", "--is-ancestor", base, head], check=True)
|
||||
prompt = args.prompt_file.read_text() + (
|
||||
f"\n\nReview only the delta from base {base} to exact head {head}. "
|
||||
"Verify current HEAD before issuing a verdict. Read only; do not modify files. "
|
||||
"End with two plain lines: REVIEWED_HEAD: <full 40-character SHA> "
|
||||
"and REVIEW_VERDICT: APPROVED or REVIEW_VERDICT: CHANGES_REQUIRED."
|
||||
)
|
||||
result = subprocess.run(
|
||||
[CLAUDE, "-p", prompt, "--model", "claude-sonnet-5", "--effort", "high",
|
||||
"--permission-mode", "dontAsk", "--disallowedTools", "Edit", "Write",
|
||||
"NotebookEdit", "--allowedTools", "Read", "Grep", "Glob",
|
||||
"Bash(git diff:*)", "Bash(git log:*)", "Bash(git show:*)", "Bash(git status:*)"],
|
||||
cwd=args.worktree, text=True, capture_output=True, timeout=1200,
|
||||
)
|
||||
args.output_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output_file.write_text(result.stdout)
|
||||
os.chmod(args.output_file, 0o600)
|
||||
if result.returncode:
|
||||
raise RuntimeError(f"Claude exited {result.returncode}: {result.stderr[-300:]}")
|
||||
decision = verdict(result.stdout, head)
|
||||
if git(args.worktree, "rev-parse", "HEAD") != head:
|
||||
raise RuntimeError("worktree HEAD changed during review")
|
||||
print(result.stdout, end="")
|
||||
if decision != "APPROVED":
|
||||
return 2
|
||||
credential = token()
|
||||
if api("/user", credential).get("login") != "review_agent":
|
||||
raise RuntimeError("review credential belongs to wrong account")
|
||||
pr = api(f"{REPO}/pulls/{args.pr}", credential)
|
||||
if pr.get("state") != "open" or (pr.get("head") or {}).get("sha") != head:
|
||||
raise RuntimeError("PR head changed before review publication")
|
||||
published = api(
|
||||
f"{REPO}/pulls/{args.pr}/reviews", credential,
|
||||
{"body": "Independent Claude Sonnet 5 review completed for this exact commit.",
|
||||
"commit_id": head, "event": "APPROVED"},
|
||||
)
|
||||
if published.get("state") != "APPROVED" or published.get("commit_id") != head:
|
||||
raise RuntimeError("Gitea did not confirm exact-head approval")
|
||||
return 0
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
raise SystemExit(main())
|
||||
except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError,
|
||||
subprocess.TimeoutExpired) as error:
|
||||
print(f"review gate failed: {error}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
Reference in New Issue
Block a user