341 lines
11 KiB
Python
Executable file
341 lines
11 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Run one fresh Codex, Claude, Gemini/agy, or Pi agent without polling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from typing import Any, Iterable
|
|
import uuid
|
|
|
|
|
|
AGENT_COMMAND = {"codex": "codex", "claude": "claude", "gemini": "agy", "pi": "pi"}
|
|
DEFAULT_AGENT = "codex"
|
|
DEFAULT_MODEL = "gpt-5.6-sol"
|
|
DEFAULT_REASONING_EFFORT = "xhigh"
|
|
LABEL_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$")
|
|
PROBE_EXPECTED = "MILESTONE_AGENT_READY"
|
|
|
|
|
|
class AgentRunError(RuntimeError):
|
|
"""One-shot runner contract error."""
|
|
|
|
|
|
def now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def emit(event: str, **payload: Any) -> None:
|
|
print(json.dumps({"event": event, **payload}, ensure_ascii=False, sort_keys=True), flush=True)
|
|
|
|
|
|
def atomic_json(path: Path, value: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + f".tmp.{os.getpid()}")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def process_start_token(pid: int) -> str | None:
|
|
"""Return a best-effort token that distinguishes PID reuse."""
|
|
stat = Path(f"/proc/{pid}/stat")
|
|
try:
|
|
remainder = stat.read_text(encoding="utf-8").rsplit(")", 1)[1].split()
|
|
return f"proc:{remainder[19]}"
|
|
except (OSError, IndexError):
|
|
return None
|
|
|
|
|
|
def workspace_root(raw: str) -> Path:
|
|
workspace = Path(raw).expanduser().resolve()
|
|
if not workspace.is_dir():
|
|
raise AgentRunError(f"workspace directory not found: {workspace}")
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--show-toplevel"],
|
|
cwd=workspace,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0 or Path(result.stdout.strip()).resolve() != workspace:
|
|
raise AgentRunError(f"workspace must be a git repository root: {workspace}")
|
|
return workspace
|
|
|
|
|
|
def state_root(workspace: Path) -> Path:
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--git-common-dir"],
|
|
cwd=workspace,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
if result.returncode == 0:
|
|
raw = Path(result.stdout.strip())
|
|
common = (workspace / raw).resolve() if not raw.is_absolute() else raw.resolve()
|
|
if os.access(common, os.W_OK):
|
|
return common / "epic-work-preparation"
|
|
fallback = Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local" / "state")))
|
|
identity = hashlib.sha256(str(workspace).encode()).hexdigest()[:16]
|
|
return fallback / "epic-work-preparation" / identity
|
|
|
|
|
|
def result_file(workspace: Path, raw: str | None) -> Path | None:
|
|
if raw is None:
|
|
return None
|
|
path = Path(raw).expanduser().resolve()
|
|
root = state_root(workspace).resolve()
|
|
try:
|
|
path.relative_to(root)
|
|
except ValueError as exc:
|
|
raise AgentRunError(f"--result-file must be inside runner state root: {root}") from exc
|
|
return path
|
|
|
|
|
|
def persist(locator: Path, result: Path | None, record: dict[str, Any]) -> None:
|
|
atomic_json(locator, record)
|
|
if result is not None:
|
|
atomic_json(result, record)
|
|
|
|
|
|
def prompt_text(args: argparse.Namespace) -> str:
|
|
if args.probe:
|
|
return (
|
|
"Reply only with the result of joining MILESTONE, _AGENT, and _READY. "
|
|
"Do not inspect files, call tools, or modify the workspace."
|
|
)
|
|
if args.prompt is not None:
|
|
return args.prompt
|
|
if args.prompt_file is None:
|
|
raise AgentRunError("--prompt or --prompt-file is required")
|
|
path = Path(args.prompt_file).expanduser().resolve()
|
|
if not path.is_file():
|
|
raise AgentRunError(f"prompt file not found: {path}")
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def build_command(
|
|
*,
|
|
agent: str,
|
|
prompt: str,
|
|
workspace: Path,
|
|
model: str | None,
|
|
reasoning_effort: str | None,
|
|
pi_provider: str | None,
|
|
session_id: str,
|
|
attempt_dir: Path,
|
|
probe: bool = False,
|
|
) -> list[str]:
|
|
if agent == "codex":
|
|
command = ["codex", "exec", "--json", "-C", str(workspace)]
|
|
if model:
|
|
command.extend(["-m", model])
|
|
if reasoning_effort:
|
|
command.extend(["-c", f'model_reasoning_effort="{reasoning_effort}"'])
|
|
if not probe:
|
|
command.append("--dangerously-bypass-approvals-and-sandbox")
|
|
command.append(prompt)
|
|
return command
|
|
if agent == "claude":
|
|
command = [
|
|
"claude",
|
|
"-p",
|
|
"--output-format",
|
|
"stream-json",
|
|
"--verbose",
|
|
"--session-id",
|
|
session_id,
|
|
]
|
|
if model:
|
|
command.extend(["--model", model])
|
|
if reasoning_effort:
|
|
command.extend(["--effort", reasoning_effort])
|
|
if not probe:
|
|
command.append("--dangerously-skip-permissions")
|
|
command.append(prompt)
|
|
return command
|
|
if agent == "gemini":
|
|
command = ["agy", "--print", prompt, "--print-timeout", "8h"]
|
|
if model:
|
|
command.extend(["--model", model])
|
|
if not probe:
|
|
command.append("--dangerously-skip-permissions")
|
|
command.extend(["--log-file", str(attempt_dir / "agy-cli.log")])
|
|
return command
|
|
if agent == "pi":
|
|
command = [
|
|
"pi",
|
|
"-p",
|
|
"--mode",
|
|
"json",
|
|
"--session-id",
|
|
session_id,
|
|
"--session-dir",
|
|
str(attempt_dir / "pi-sessions"),
|
|
]
|
|
if not probe:
|
|
command.append("--approve")
|
|
if pi_provider:
|
|
command.extend(["--provider", pi_provider])
|
|
if model:
|
|
command.extend(["--model", model])
|
|
if reasoning_effort:
|
|
command.extend(["--thinking", reasoning_effort])
|
|
command.append(prompt)
|
|
return command
|
|
raise AgentRunError(f"unsupported agent: {agent}")
|
|
|
|
|
|
def sanitized_command(command: list[str], prompt: str) -> list[str]:
|
|
return ["<prompt>" if value == prompt else value for value in command]
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
value = argparse.ArgumentParser(description=__doc__)
|
|
value.add_argument("--agent", choices=sorted(AGENT_COMMAND), default=DEFAULT_AGENT)
|
|
value.add_argument("--workspace", required=True)
|
|
prompt_group = value.add_mutually_exclusive_group()
|
|
prompt_group.add_argument("--prompt")
|
|
prompt_group.add_argument("--prompt-file")
|
|
value.add_argument("--model")
|
|
value.add_argument("--reasoning-effort", default=DEFAULT_REASONING_EFFORT)
|
|
value.add_argument("--pi-provider")
|
|
value.add_argument("--label", default="one-shot")
|
|
value.add_argument("--probe", action="store_true")
|
|
value.add_argument("--result-file")
|
|
return value
|
|
|
|
|
|
def execute(args: argparse.Namespace) -> int:
|
|
workspace = workspace_root(args.workspace)
|
|
if args.model is None and args.agent == DEFAULT_AGENT:
|
|
args.model = DEFAULT_MODEL
|
|
if not LABEL_PATTERN.fullmatch(args.label):
|
|
raise AgentRunError("--label may contain only letters, digits, dot, underscore, and hyphen")
|
|
prompt = prompt_text(args)
|
|
result = result_file(workspace, args.result_file)
|
|
executable = AGENT_COMMAND[args.agent]
|
|
resolved = shutil.which(executable)
|
|
if resolved is None:
|
|
raise AgentRunError(f"agent command not found: agent={args.agent} command={executable}")
|
|
|
|
execution_id = f"{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:12]}"
|
|
root = state_root(workspace)
|
|
attempt_dir = root / "runs" / f"{args.label}-{execution_id}"
|
|
attempt_dir.mkdir(parents=True, exist_ok=False)
|
|
stream = attempt_dir / "stream.log"
|
|
locator = attempt_dir / "locator.json"
|
|
session_id = str(uuid.uuid4())
|
|
command = build_command(
|
|
agent=args.agent,
|
|
prompt=prompt,
|
|
workspace=workspace,
|
|
model=args.model,
|
|
reasoning_effort=args.reasoning_effort,
|
|
pi_provider=args.pi_provider,
|
|
session_id=session_id,
|
|
attempt_dir=attempt_dir,
|
|
probe=args.probe,
|
|
)
|
|
record: dict[str, Any] = {
|
|
"execution_id": execution_id,
|
|
"label": args.label,
|
|
"workspace": str(workspace),
|
|
"agent": args.agent,
|
|
"command": sanitized_command(command, prompt),
|
|
"model": args.model,
|
|
"reasoning_effort": args.reasoning_effort,
|
|
"prompt_sha256": hashlib.sha256(prompt.encode()).hexdigest(),
|
|
"session_id": session_id,
|
|
"stream_log": str(stream),
|
|
"locator": str(locator),
|
|
"started_at": now(),
|
|
"status": "starting",
|
|
}
|
|
persist(locator, result, record)
|
|
emit(
|
|
"AGENT_STARTED",
|
|
agent=args.agent,
|
|
execution_id=execution_id,
|
|
label=args.label,
|
|
locator=str(locator),
|
|
model=args.model or "default",
|
|
)
|
|
with stream.open("wb") as output:
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
cwd=workspace,
|
|
env={
|
|
**os.environ,
|
|
"MILESTONE_PREPARATION_EXECUTION_ID": execution_id,
|
|
},
|
|
stdout=output,
|
|
stderr=subprocess.STDOUT,
|
|
start_new_session=True,
|
|
)
|
|
except OSError as exc:
|
|
record.update(status="failed", finished_at=now(), exit_code=127, error=str(exc))
|
|
persist(locator, result, record)
|
|
emit("AGENT_FINISHED", execution_id=execution_id, label=args.label, result="failed", exit_code=127)
|
|
return 127
|
|
record.update(
|
|
status="running",
|
|
agent_pid=process.pid,
|
|
agent_process_start_token=process_start_token(process.pid),
|
|
)
|
|
persist(locator, result, record)
|
|
try:
|
|
exit_code = process.wait()
|
|
except KeyboardInterrupt:
|
|
record.update(status="tracking", interrupted_at=now(), agent_pid=process.pid)
|
|
persist(locator, result, record)
|
|
emit(
|
|
"AGENT_TRACKING",
|
|
execution_id=execution_id,
|
|
label=args.label,
|
|
locator=str(locator),
|
|
pid=process.pid,
|
|
)
|
|
return 3
|
|
|
|
size = stream.stat().st_size
|
|
status = "succeeded" if exit_code == 0 and size > 0 else "failed"
|
|
if args.probe and status == "succeeded":
|
|
content = stream.read_text(encoding="utf-8", errors="replace")
|
|
if PROBE_EXPECTED not in content.upper():
|
|
status = "failed"
|
|
exit_code = 2
|
|
record.update(status=status, finished_at=now(), exit_code=exit_code, output_bytes=size)
|
|
persist(locator, result, record)
|
|
emit(
|
|
"AGENT_FINISHED",
|
|
execution_id=execution_id,
|
|
exit_code=exit_code,
|
|
label=args.label,
|
|
locator=str(locator),
|
|
result=status,
|
|
)
|
|
return exit_code if status == "succeeded" else (exit_code or 2)
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
args = parser().parse_args(argv)
|
|
try:
|
|
return execute(args)
|
|
except (AgentRunError, OSError) as exc:
|
|
emit("AGENT_FINISHED", label=getattr(args, "label", "one-shot"), result="failed", reason=str(exc))
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|