File size: 3,192 Bytes
3c7404b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | #!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import subprocess
import sys
def _parse_opencode_json_stream(raw_stdout: str) -> dict:
text_parts: list[str] = []
saw_finish = False
malformed_lines = 0
for line in raw_stdout.splitlines():
line = line.strip()
if not line:
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
malformed_lines += 1
continue
if event.get("type") == "text":
part = event.get("part") or {}
text = part.get("text")
if isinstance(text, str):
text_parts.append(text)
if event.get("type") == "step_finish":
saw_finish = True
return {
"text": "".join(text_parts).strip(),
"saw_finish": saw_finish,
"malformed_lines": malformed_lines,
}
def main() -> int:
parser = argparse.ArgumentParser(description="Run opencode once via Python subprocess")
parser.add_argument("--model", required=True)
parser.add_argument("--agent", required=True)
parser.add_argument("--cwd", required=False)
parser.add_argument("--timeout-sec", type=int, default=600)
parser.add_argument("--prompt", required=True)
args = parser.parse_args()
command = [
"opencode",
"run",
"--pure",
"--format",
"json",
"--model",
args.model,
"--agent",
args.agent,
args.prompt,
]
try:
completed = subprocess.run(
command,
cwd=args.cwd or None,
capture_output=True,
text=True,
timeout=max(1, int(args.timeout_sec)),
check=False,
)
parsed = _parse_opencode_json_stream(completed.stdout)
completed_ok = bool(parsed["saw_finish"] or parsed["text"])
payload = {
"ok": completed.returncode == 0 and completed_ok,
"returncode": completed.returncode,
"stdout": parsed["text"],
"stderr": completed.stderr,
"timeout": False,
"step_finish_seen": parsed["saw_finish"],
"malformed_output_lines": parsed["malformed_lines"],
"raw_stdout": completed.stdout,
}
except subprocess.TimeoutExpired as exc:
raw_stdout = exc.stdout.decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "")
raw_stderr = exc.stderr.decode() if isinstance(exc.stderr, bytes) else (exc.stderr or "")
parsed = _parse_opencode_json_stream(raw_stdout)
timeout_ok = bool(parsed["saw_finish"] or (parsed["text"] and not raw_stderr.strip()))
payload = {
"ok": timeout_ok,
"returncode": 124,
"stdout": parsed["text"],
"stderr": raw_stderr,
"timeout": not timeout_ok,
"step_finish_seen": parsed["saw_finish"],
"malformed_output_lines": parsed["malformed_lines"],
"raw_stdout": raw_stdout,
}
sys.stdout.write(json.dumps(payload))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|