| """Local OpenCode CLI access for the LLM test endpoints. |
| |
| Configuration via environment variables: |
| OPENCODE_BIN: executable name or path (default: "opencode") |
| OPENCODE_TIMEOUT: per-invocation timeout in seconds (default: 120) |
| """ |
|
|
| import asyncio |
| import json |
| import os |
| import shutil |
| import signal |
| import subprocess |
|
|
| from fastapi import HTTPException |
|
|
| DEFAULT_TIMEOUT = 120.0 |
| MAX_ERROR_CHARS = 4000 |
|
|
|
|
| def _resolve_binary() -> str: |
| configured = os.environ.get("OPENCODE_BIN", "").strip() |
| binary = configured or "opencode" |
| resolved = shutil.which(binary) |
| if not resolved: |
| raise HTTPException( |
| status_code=500, |
| detail=f"OpenCode CLI not found ({binary!r}); install it or set OPENCODE_BIN", |
| ) |
| return resolved |
|
|
|
|
| def _timeout_seconds() -> float: |
| raw = os.environ.get("OPENCODE_TIMEOUT", "").strip() |
| if not raw: |
| return DEFAULT_TIMEOUT |
| try: |
| value = float(raw) |
| except ValueError: |
| return DEFAULT_TIMEOUT |
| return value if value > 0 else DEFAULT_TIMEOUT |
|
|
|
|
| def _subprocess_options() -> dict: |
| if os.name == "nt": |
| return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} |
| return {"start_new_session": True} |
|
|
|
|
| def _kill(process: asyncio.subprocess.Process) -> None: |
| if process.returncode is not None: |
| return |
| try: |
| if os.name == "nt": |
| process.kill() |
| else: |
| os.killpg(process.pid, signal.SIGKILL) |
| except ProcessLookupError: |
| pass |
|
|
|
|
| def _json_events(stdout: str) -> list[dict]: |
| events: list[dict] = [] |
| for line in stdout.splitlines(): |
| line = line.strip() |
| if not line: |
| continue |
| try: |
| event = json.loads(line) |
| except json.JSONDecodeError: |
| continue |
| if isinstance(event, dict): |
| events.append(event) |
| return events |
|
|
|
|
| def _extract_text(stdout: str) -> str: |
| texts: list[str] = [] |
| for event in _json_events(stdout): |
| if event.get("type") != "text": |
| continue |
| part = event.get("part") |
| if not isinstance(part, dict): |
| continue |
| text = part.get("text") |
| if isinstance(text, str) and text.strip(): |
| texts.append(text.strip()) |
| return "\n".join(texts).strip() |
|
|
|
|
| def _event_error(stdout: str) -> str: |
| for event in _json_events(stdout): |
| if event.get("type") != "error": |
| continue |
| error = event.get("error") |
| if isinstance(error, dict): |
| message = error.get("message") |
| if isinstance(message, str) and message.strip(): |
| return message.strip() |
| if isinstance(error, str) and error.strip(): |
| return error.strip() |
| return "" |
|
|
|
|
| def _detail(stdout: str, stderr: str) -> str: |
| detail = stderr.strip() or _event_error(stdout) or stdout.strip() or "no output" |
| if len(detail) > MAX_ERROR_CHARS: |
| detail = detail[:MAX_ERROR_CHARS].rstrip() + "..." |
| return detail |
|
|
|
|
| async def _run(args: list[str]) -> tuple[int, str, str]: |
| binary = _resolve_binary() |
| timeout = _timeout_seconds() |
| env = os.environ.copy() |
| env.setdefault("NO_COLOR", "1") |
|
|
| try: |
| process = await asyncio.create_subprocess_exec( |
| binary, |
| *args, |
| stdin=asyncio.subprocess.DEVNULL, |
| stdout=asyncio.subprocess.PIPE, |
| stderr=asyncio.subprocess.PIPE, |
| env=env, |
| **_subprocess_options(), |
| ) |
| except OSError as exc: |
| raise HTTPException( |
| status_code=500, |
| detail=f"Could not start OpenCode CLI: {exc}", |
| ) from exc |
|
|
| try: |
| stdout_bytes, stderr_bytes = await asyncio.wait_for( |
| process.communicate(), timeout=timeout |
| ) |
| except asyncio.TimeoutError: |
| _kill(process) |
| await process.wait() |
| raise HTTPException( |
| status_code=504, |
| detail=f"OpenCode CLI timed out after {timeout:g} seconds", |
| ) from None |
|
|
| return ( |
| process.returncode if process.returncode is not None else -1, |
| stdout_bytes.decode("utf-8", errors="replace"), |
| stderr_bytes.decode("utf-8", errors="replace"), |
| ) |
|
|
|
|
| async def run_opencode(prompt: str, model: str | None = None) -> str: |
| """Run one prompt through OpenCode's plan agent, optionally with a model.""" |
| if not prompt.strip(): |
| raise HTTPException(status_code=400, detail="prompt must not be empty") |
|
|
| args = ["run", "--standalone", "--agent", "plan", "--format", "json"] |
| if model and model.strip(): |
| args.extend(["--model", model.strip()]) |
| args.extend(["--", prompt]) |
|
|
| returncode, stdout, stderr = await _run(args) |
| if returncode != 0: |
| raise HTTPException( |
| status_code=502, |
| detail=f"OpenCode CLI failed: {_detail(stdout, stderr)}", |
| ) |
|
|
| content = _extract_text(stdout) |
| if not content: |
| raise HTTPException( |
| status_code=502, |
| detail=f"OpenCode CLI returned an empty response: {_detail(stdout, stderr)}", |
| ) |
| return content |
|
|
|
|
| async def list_opencode_models() -> list[str]: |
| """Return the model selectors available to the local OpenCode CLI.""" |
| returncode, stdout, stderr = await _run(["models", "--standalone"]) |
| if returncode != 0: |
| raise HTTPException( |
| status_code=502, |
| detail=f"OpenCode CLI failed: {_detail(stdout, stderr)}", |
| ) |
| return [line.strip() for line in stdout.splitlines() if line.strip()] |
|
|
