def run(
self,
input: str,
context: Optional[AgentContext] = None,
**kwargs: Any,
) -> AgentResult:
"""Run a coding task through opencode and return the assistant result."""
denied = self._execution_denied_result()
if denied is not None:
return denied
self._emit_turn_start(input)
# Resolve which opencode provider/model to address. Fail clearly rather
# than letting opencode 500 on an unregistered provider.
if self._provider_base_url:
prov, mid = self._provider_id, self._model_id
elif "/" in self._model_id:
prov, _, mid = self._model_id.partition("/")
else:
self._emit_turn_end(turns=1, error=True)
return AgentResult(
content=(
f"OpenCodeAgent could not determine an opencode provider for "
f"model {self._model_id!r}: no OpenAI-compatible base URL could "
f"be derived from the engine. Pass provider_base_url=..., or use "
f"a 'provider/model' that opencode already knows."
),
turns=1,
metadata={"error": True},
)
model_spec_v1 = {"providerID": prov, "modelID": mid}
model_spec_v2 = {"providerID": prov, "id": mid}
try:
self._ensure_server()
except RuntimeError as exc:
self._emit_turn_end(turns=1, error=True)
return AgentResult(content=str(exc), turns=1, metadata={"error": True})
data: dict = {}
turn_parts: List[dict] = []
active_version = self._resolved_api_version or "v1"
session_id = ""
try:
with self._client() as c:
version = self._resolved_api_version
# Session creation with auto-fallback
if version == "v2":
ses = c.post(
"/api/session",
json={
"title": input[:80],
"agent": self._agent,
"model": model_spec_v2,
},
)
ses.raise_for_status()
ses_data = ses.json()
session_id = ses_data.get("id") or (ses_data.get("data") or {}).get(
"id"
)
elif version == "v1":
ses = c.post("/session", json={"title": input[:80]})
ses.raise_for_status()
session_id = ses.json().get("id")
else:
# auto-detection: try V1 first, fallback to V2 on 404
try:
ses = c.post("/session", json={"title": input[:80]})
ses.raise_for_status()
session_id = ses.json().get("id")
self._resolved_api_version = "v1"
except Exception as ses_exc:
status = getattr(
getattr(ses_exc, "response", None), "status_code", None
)
if status == 404:
logger.info(
"OpenCode /session returned 404; "
"falling back to /api/session (v2)"
)
ses = c.post(
"/api/session",
json={
"title": input[:80],
"agent": self._agent,
"model": model_spec_v2,
},
)
ses.raise_for_status()
ses_data = ses.json()
session_id = ses_data.get("id") or (
ses_data.get("data") or {}
).get("id")
self._resolved_api_version = "v2"
else:
raise
if not session_id:
raise RuntimeError(
"Could not obtain session ID from opencode server"
)
self._current_session_id = session_id
active_version = self._resolved_api_version or "v1"
# Prompt / message dispatch
if active_version == "v2":
resp = c.post(
f"/api/session/{session_id}/prompt",
json={"text": input},
)
resp.raise_for_status()
# V2 admits a queued user input, not a completed assistant
# reply. Both synchronization and transcript retrieval must
# succeed before we can report a completed turn.
wait = c.post(f"/api/experimental/session/{session_id}/wait")
wait.raise_for_status()
terminal: Optional[dict] = None
for message in _fetch_v2_messages(c, session_id):
if message.get("type") == "idle":
terminal = message
continue
if message.get("type") != "assistant":
continue
terminal = None
parts = message.get("content")
if not isinstance(parts, list):
raise RuntimeError(
"OpenCode V2 returned invalid assistant content"
)
turn_parts.extend(parts)
data = message
if not data:
raise RuntimeError(
"OpenCode V2 completed without an assistant reply"
)
# The wait route reports idle even when execution failed.
# The terminal transcript marker records the actual outcome;
# a provider finish reason alone does not describe the turn.
if terminal is None:
raise RuntimeError(
"OpenCode V2 returned an incomplete turn transcript"
)
if terminal.get("outcome") != "succeeded" or terminal.get("error"):
error = terminal.get("error")
detail = (
error.get("message", "Turn failed")
if isinstance(error, dict)
else f"Turn {terminal.get('outcome', 'did not complete')}"
)
raise RuntimeError(f"OpenCode V2 failed: {detail}")
if data.get("error") or data.get("finish") in (
"error",
"content-filter",
):
error = data.get("error")
detail = (
error.get("message", "Assistant generation failed")
if isinstance(error, dict)
else "Assistant generation failed"
)
raise RuntimeError(f"OpenCode V2 assistant failed: {detail}")
if data.get("finish") not in ("stop", "length", "tool-calls"):
raise RuntimeError(
"OpenCode V2 returned an unfinished assistant reply"
)
else:
body: dict = {
"agent": self._agent,
"model": model_spec_v1,
"parts": [{"type": "text", "text": input}],
}
resp = c.post(f"/session/{session_id}/message", json=body)
resp.raise_for_status()
data = resp.json()
turn_parts = list(data.get("parts", []))
try:
msgs = c.get(f"/session/{session_id}/message").json()
if isinstance(msgs, list):
turn_parts = [
part
for mm in msgs
if isinstance(mm, dict)
for part in (mm.get("parts") or mm.get("content") or [])
]
except Exception as get_exc:
logger.debug("opencode message fetch failed: %s", get_exc)
except Exception as exc:
logger.error("opencode run failed: %s", exc, exc_info=True)
self._emit_turn_end(turns=1, error=True)
return AgentResult(
content=f"opencode agent failed: {exc}",
turns=1,
metadata={"error": True},
)
info = data.get("info", {}) if isinstance(data, dict) else {}
model_obj = data.get("model", {}) if isinstance(data.get("model"), dict) else {}
final_parts = data.get("content") or data.get("parts") or []
content = (
_extract_text(final_parts)
if isinstance(final_parts, list)
else (data.get("text", "") if isinstance(data, dict) else "")
)
tool_results = _extract_tool_results(turn_parts)
self._emit_turn_end(turns=1)
return AgentResult(
content=content,
tool_results=tool_results,
turns=1,
metadata={
"finish": data.get("finish") or info.get("finish"),
"tokens": data.get("tokens") or info.get("tokens"),
"provider_id": (
model_obj.get("providerID") or info.get("providerID", prov)
),
"model_id": (
model_obj.get("id")
or model_obj.get("modelID")
or info.get("modelID", mid)
),
"session_id": (
data.get("sessionID") or info.get("sessionID", session_id or "")
),
"agent": data.get("agent") or self._agent,
"api_version": active_version,
},
)