File size: 5,559 Bytes
ae60c27 4819e4b ae60c27 4819e4b ae60c27 | 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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | """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()]
|