cleveragents-core/tools/_opencode_worker.py

"""Blocking OpenCode worker dispatcher.

Implements ``run_worker_blocking`` per
``docs/development/conflict-drive-plan.md`` § 6.2. The single entry
point spins up an OpenCode session, dispatches an agent prompt, polls
until the session is idle, performs tolerant JSON extraction on the
final assistant message, and tears the session down — all from a
deterministic Python driver with a watchdog timeout.

The expected caller is :mod:`tools.conflict_drive`, but the surface is
agent-agnostic and can be reused by other deterministic drivers that
need a single-shot LLM worker.

Outcomes
--------

The worker returns a :class:`WorkerResult` whose ``outcome`` is one of:

- ``resolved`` — every conflict was semantically resolved.
- ``unresolvable`` — the rebase finished with conflict markers
  committed; the driver will detect them via ``git diff --check`` and
  escalate via ``auto/needs-implementer``.
- ``rebase-failed`` — the worker could not progress the rebase at all.
- ``timeout`` — the watchdog wallclock fired (default 900 s).
- ``transport-error`` — OpenCode HTTP 5xx, network reset, or any
  unexpected I/O error during the dispatch lifecycle.

The driver maps these onto the failure-class table in plan § 5.1;
``verification-fail`` and ``lease-violation`` are driver-side
classifications, not worker outcomes.
"""
from __future__ import annotations

import json
import logging
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable, Literal


logger = logging.getLogger("opencode_worker")


Outcome = Literal[
    "resolved",
    "unresolvable",
    "rebase-failed",
    "timeout",
    "transport-error",
]


@dataclass
class WorkerResult:
    """Typed return value from :func:`run_worker_blocking`.

    ``raw_response`` carries the verbatim final-assistant message text
    purely for telemetry / debug; the driver never parses it directly
    (the typed ``outcome`` and ``files_touched`` fields above are the
    contract).
    """

    outcome: Outcome
    files_touched: list[str] = field(default_factory=list)
    wallclock_seconds: float = 0.0
    session_id: str = ""
    raw_response: str = ""


SessionStatus = Literal["completed", "timeout", "transport-error"]


@dataclass
class SessionResult:
    """Outcome-agnostic result from running a one-shot OpenCode session.

    Workers that do not emit a fixed JSON schema (review, implementation)
    use this directly; the conflict driver keeps using
    :func:`run_worker_blocking`, which classifies the parsed JSON into a
    domain-specific :class:`WorkerResult.outcome`.

    ``status`` describes only the *session* lifecycle: did the session
    finish normally (``completed``), hit the watchdog
    (``timeout``), or fail at the OpenCode HTTP layer
    (``transport-error``). ``parsed_json`` is the last parseable JSON
    object found in the final assistant message, or ``None`` if the
    worker did not emit one — which is the normal case for review /
    implementation workers.
    """

    status: SessionStatus
    wallclock_seconds: float = 0.0
    session_id: str = ""
    raw_response: str = ""
    parsed_json: dict[str, Any] | None = None


# ─── HTTP layer (kept intentionally minimal — no shared retries) ──────────
#
# We deliberately do NOT reuse ``_claim_runtime``'s state-change retry
# policy: OpenCode's writes are not idempotent (POST /session creates a
# new session every call) and a transient retry would multiply session
# leaks. Each request gets one shot; any error short-circuits to
# ``transport-error``.


# Tuple of exceptions every dispatch step catches. ``urllib`` raises
# ``HTTPError`` (subclass of ``URLError``) on non-2xx, so this covers
# both transport-level failures (DNS, connection refused) and OpenCode
# 5xx / 4xx.
_TRANSPORT_EXC: tuple[type[BaseException], ...] = (
    urllib.error.URLError,
    TimeoutError,
    OSError,
)


def _request(
    method: str,
    url: str,
    *,
    body: Any | None = None,
    timeout: int = 30,
) -> Any:
    """Issue a single HTTP request to the OpenCode server.

    Returns the parsed JSON body on 2xx; raises ``urllib.error.HTTPError``
    (a subclass of ``URLError``) on non-2xx, and lets all other
    transport-level exceptions propagate. Callers catch
    :data:`_TRANSPORT_EXC` and convert to ``transport-error``.
    """
    data = json.dumps(body).encode() if body is not None else None
    headers = {}
    if data is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        payload = resp.read()
        if not payload:
            return None
        ct = resp.headers.get("Content-Type", "")
        if "application/json" in ct:
            try:
                return json.loads(payload)
            except (ValueError, TypeError):
                return None
        return payload


# ─── Tolerant JSON extraction ──────────────────────────────────────────────
#
# The worker's prompt asks for "exactly one JSON object somewhere in
# your final assistant message". Real LLMs surround it with prose, code
# fences, trailing whitespace, and frequently emit stray ``{`` inside
# explanatory prose ("set REPO_DIR={your repo path}"). We scan
# left-to-right looking at every ``{`` independently — taking only the
# LAST successfully-parsed object — so a single unmatched opener
# earlier in the message cannot shadow the structured exit at the end.

def _extract_last_json_object(text: str) -> dict[str, Any] | None:
    """Return the last ``{...}`` substring of ``text`` that parses as a
    JSON object, or ``None`` if no such substring exists.

    Strategy: walk left-to-right; at every ``{`` start, run a
    string-aware bracket matcher to see whether THIS opener closes at
    depth 0 within the buffer. If it does, attempt ``json.loads`` on
    the span and record success; in any case we then advance to the
    next character (NOT to ``end + 1``) so an unmatched outer ``{``
    cannot hide a valid object that nests inside it or that follows
    it later. The LAST successful parse wins so the worker's
    structured exit (always at the end of the message by prompt
    convention) is what we return.

    Cost: best-case O(N) when no stray openers appear; worst case
    O(N²) when the buffer contains many unterminated ``{`` runs (the
    pathological run-of-N-opens-no-closes string). LLM output is
    bounded by ``max_tokens`` so the worst case is irrelevant in
    practice; the pre-P1-6 implementation was the same complexity for
    a much weaker reason (quadratic shrink). Replacing the unmatched-
    opener ``break`` with ``i += 1`` is what closes the regression
    flagged in the architect's review.

    Tolerates code fences (``\\`\\`\\`json`` …), trailing prose, partial
    fragments, multiple JSON objects, and stray ``{`` inside prose.
    Does NOT tolerate JSON arrays at the outer level — by contract the
    worker emits an object.
    """
    if not text:
        return None
    last_match: dict[str, Any] | None = None
    n = len(text)
    i = 0
    while i < n:
        if text[i] != "{":
            i += 1
            continue
        # Scan forward from here, respecting JSON string syntax, until
        # we either close at depth 0 (complete object) or run off the
        # end (truncated). Inside a string, ``{`` and ``}`` are literal
        # data and must not move the depth counter.
        depth = 0
        in_string = False
        escape = False
        end = -1
        for j in range(i, n):
            ch = text[j]
            if in_string:
                if escape:
                    escape = False
                elif ch == "\\":
                    escape = True
                elif ch == '"':
                    in_string = False
                continue
            if ch == '"':
                in_string = True
                continue
            if ch == "{":
                depth += 1
            elif ch == "}":
                depth -= 1
                if depth == 0:
                    end = j + 1
                    break
        if end < 0:
            # Unmatched opener — every attempt that includes this ``{``
            # in its span will fail. Advance ONE position so subsequent
            # ``{``s (which may close fine) still get a chance. The
            # earlier ``break`` here was a regression: it caused
            # ``"prose with { stray. Then {valid: true}"`` to return
            # None instead of finding the trailing object.
            i += 1
            continue
        slice_ = text[i:end]
        try:
            parsed = json.loads(slice_)
        except (ValueError, TypeError):
            parsed = None
        if isinstance(parsed, dict):
            last_match = parsed
            # The outer span parsed cleanly — skip past it. Without this
            # the scan would re-enter the body and overwrite ``last_match``
            # with a nested sub-object, returning ``{"inner": 1}`` for
            # input ``{"outer": {"inner": 1}}``. The contract is "last
            # complete object", not "last opener".
            i = end
        else:
            # Span closed at depth 0 but did NOT parse (e.g. invalid
            # syntax inside a stray-bracketed prose block). Advance one
            # position so a valid object that nests inside this
            # unparseable span is still discoverable.
            i += 1
    return last_match


def _last_assistant_text(messages: list[dict[str, Any]]) -> str:
    """Return the concatenated text of the most recent assistant message
    in ``messages`` (oldest-first ordering, as returned by
    ``GET /session/{id}/message``). Empty string when no assistant
    message exists yet.
    """
    for m in reversed(messages):
        info = m.get("info") or {}
        if info.get("role") != "assistant":
            continue
        parts = m.get("parts") or []
        # Concatenate visible text parts only; ignore reasoning / tool /
        # patch / step-* entries — none of them are the structured exit
        # the worker is asked to emit.
        text_parts = [p.get("text", "") for p in parts if p.get("type") == "text"]
        return "\n".join(t for t in text_parts if t)
    return ""


# ─── Idle detection ────────────────────────────────────────────────────────


def _session_busy_state(server_url: str, session_id: str) -> str:
    """Return ``"busy"``, ``"idle"``, or ``"unknown"`` for ``session_id``.

    ``GET /session/status`` returns a map of ``session_id -> {"type":
    "busy"|"idle"}``. The map is updated asynchronously by OpenCode after
    a session is created, so a missing entry is **ambiguous**: it can
    mean either "session has not been registered yet" (immediately after
    ``POST /session/{id}/prompt_async``) or "session has run to
    completion and been removed" (a real terminal state). Distinguishing
    these two cases is the caller's responsibility.

    This helper returns ``"unknown"`` for a missing entry and lets
    :func:`run_session_blocking` apply the
    ``seen_busy_at_least_once`` heuristic that disambiguates the
    pre-launch race from a genuine terminal idle.

    Mirrors the data shape of
    ``.opencode/skills/auto-agents-system/scripts/session_wait_till_idle.ts``;
    that script tolerates the race window because its 5 s poll interval
    + ``Promise.all`` round-trip lets OpenCode register the session
    before the first observation.
    """
    statuses = _request("GET", f"{server_url}/session/status") or {}
    if not isinstance(statuses, dict):
        return "unknown"
    entry = statuses.get(session_id)
    if not isinstance(entry, dict):
        return "unknown"
    return "busy" if entry.get("type") == "busy" else "idle"


# Wait at most this many seconds for the session to first appear as
# ``busy`` after ``POST /session/{id}/prompt_async``. If the session
# never becomes busy within this window we assume the prompt never
# reached the agent (OpenCode internal error, malformed agent, etc.)
# and short-circuit with ``transport-error`` rather than silently
# reporting ``completed`` with an empty raw_response.
_STARTUP_GRACE_SECONDS = 30.0


# ─── Public entry point ────────────────────────────────────────────────────


def list_sessions(server_url: str) -> list[dict[str, Any]]:
    """Return ``GET /session`` from an OpenCode server, or ``[]`` on
    transport / parse failure.

    Public wrapper around the internal :func:`_request` helper so other
    deterministic drivers (e.g. the dispatcher coexistence guard) can
    enumerate live sessions without reaching into private API.
    Failures are swallowed and returned as an empty list because the
    sole consumer today is best-effort startup detection; if your use
    case needs the distinction between "no sessions" and "server
    unreachable", call :func:`_request` directly and handle
    :data:`_TRANSPORT_EXC`.
    """
    try:
        sessions = _request("GET", f"{server_url}/session")
    except _TRANSPORT_EXC as e:
        logger.warning("OpenCode list_sessions failed (%s): %s", server_url, e)
        return []
    if not isinstance(sessions, list):
        return []
    return [s for s in sessions if isinstance(s, dict)]


def run_session_blocking(
    *,
    server_url: str,
    agent: str,
    tag: str,
    prompt: str,
    timeout_seconds: int = 900,
    poll_interval_seconds: float = 2.0,
    on_poll: Callable[[], None] | None = None,
) -> SessionResult:
    """Spin up a one-shot OpenCode agent session, wait for it to finish,
    and return a typed :class:`SessionResult`.

    This is the outcome-agnostic entry point used by drivers whose
    workers do not emit a fixed ``{"outcome": ...}`` JSON schema (review,
    implementation). For the conflict driver, see
    :func:`run_worker_blocking`, which thin-wraps this function and adds
    the conflict-specific JSON-outcome classification.

    Lifecycle (matches plan § 6.2):

    1. ``POST /session`` with title ``"[{tag}] {agent}"`` → ``session_id``.
       The ``[TAG]`` prefix is the same convention used by the
       TypeScript ``session_start.ts`` helper so existing operator
       tooling (``session_find_by_tag``) finds these sessions.
    2. ``POST /session/{id}/prompt_async`` with the agent name + prompt.
    3. Poll ``GET /session/status`` every ``poll_interval_seconds`` until
       the session is idle. The watchdog wallclock at ``timeout_seconds``
       short-circuits with ``status="timeout"``; the session is asked to
       abort and DELETEd in the finally block.
    4. ``GET /session/{id}/message`` to read the final assistant turn.
    5. Tolerant JSON extraction (last parseable ``{...}`` substring of the
       last assistant text) → ``parsed_json``. May be ``None`` for
       workers that do not emit a JSON object — that is not a failure.
    6. ``DELETE /session/{id}`` in a finally block. On DELETE failure log
       a warning and continue — the session leaks until OpenCode server
       restart. Acceptable for this driver's expected dispatch rate
       (single-digit per day per plan § 6.2).

    Any HTTP / network error short-circuits to ``status="transport-error"``;
    the finally block still attempts session teardown.

    ``on_poll`` is invoked once per polling iteration (i.e., approximately
    every ``poll_interval_seconds``) before the session-status GET is
    issued. The dispatcher uses this to refresh its liveness heartbeat
    while a long-running worker is in flight; without it, the launcher
    or systemd unit would treat a 30-minute review session as a hung
    process and restart the dispatcher mid-cycle, orphaning the OpenCode
    session and its ``auto/claimed-*`` lock. Exceptions raised from
    ``on_poll`` are logged and swallowed so a transient heartbeat-write
    failure (disk full, EROFS, EACCES) cannot mask a successful worker
    completion. The callback fires:

    - On every iteration of the polling loop, including the first one
      after the post-``prompt_async`` warmup sleep.
    - Before the deadline check, so even a callback that takes longer
      than the remaining wallclock budget still gets at least one
      attempt to update liveness state before timeout aborts the
      session.

    The callback should be idempotent and side-effect-only (write a
    file, touch a TCP socket, etc.) — its return value is ignored.
    """
    started_at = time.monotonic()
    session_id = ""

    def _elapsed() -> float:
        return time.monotonic() - started_at

    try:
        title = f"[{tag}] {agent}"
        try:
            session = _request(
                "POST", f"{server_url}/session", body={"title": title}
            )
        except _TRANSPORT_EXC as e:
            logger.warning("OpenCode session create failed: %s", e)
            return SessionResult(
                status="transport-error",
                wallclock_seconds=_elapsed(),
                raw_response=str(e),
            )
        if not isinstance(session, dict) or not session.get("id"):
            logger.warning(
                "OpenCode session create returned malformed payload: %r", session
            )
            return SessionResult(
                status="transport-error",
                wallclock_seconds=_elapsed(),
                raw_response=json.dumps(session) if session is not None else "",
            )
        session_id = str(session["id"])
        logger.info(
            "OpenCode session %s created (tag=%s, agent=%s)",
            session_id,
            tag,
            agent,
        )

        try:
            _request(
                "POST",
                f"{server_url}/session/{session_id}/prompt_async",
                body={"agent": agent, "parts": [{"type": "text", "text": prompt}]},
            )
        except _TRANSPORT_EXC as e:
            logger.warning(
                "OpenCode prompt_async failed for session %s: %s", session_id, e
            )
            return SessionResult(
                status="transport-error",
                wallclock_seconds=_elapsed(),
                session_id=session_id,
                raw_response=str(e),
            )

        deadline = started_at + timeout_seconds
        # Pre-sleep so OpenCode has a chance to register the freshly
        # dispatched session as ``busy``. Without this the first
        # status poll fires before the agent starts and the
        # "missing entry == idle" interpretation in older code paths
        # would short-circuit the loop with no work done.
        time.sleep(min(poll_interval_seconds, max(0.0, deadline - time.monotonic())))

        seen_busy = False
        warmup_deadline = started_at + _STARTUP_GRACE_SECONDS
        while True:
            if on_poll is not None:
                try:
                    on_poll()
                except Exception as e:  # noqa: BLE001 - callback errors must not kill the worker
                    logger.warning(
                        "on_poll callback raised %s; continuing — heartbeat "
                        "may be stale but the worker session is unaffected",
                        type(e).__name__,
                        exc_info=True,
                    )
            if time.monotonic() >= deadline:
                logger.warning(
                    "OpenCode worker timed out after %.0f s (session=%s)",
                    timeout_seconds,
                    session_id,
                )
                try:
                    _request(
                        "POST",
                        f"{server_url}/session/{session_id}/abort",
                    )
                except _TRANSPORT_EXC as e:
                    logger.warning(
                        "OpenCode abort failed for session %s: %s",
                        session_id,
                        e,
                    )
                return SessionResult(
                    status="timeout",
                    wallclock_seconds=_elapsed(),
                    session_id=session_id,
                )
            try:
                state = _session_busy_state(server_url, session_id)
            except _TRANSPORT_EXC as e:
                logger.warning(
                    "OpenCode status poll failed for session %s: %s",
                    session_id,
                    e,
                )
                return SessionResult(
                    status="transport-error",
                    wallclock_seconds=_elapsed(),
                    session_id=session_id,
                    raw_response=str(e),
                )
            if state == "busy":
                seen_busy = True
            elif state == "idle" and seen_busy:
                # Genuine terminal idle: the session was observed busy
                # at some point and is now idle. Done.
                break
            else:
                # Either ``state == "idle"`` and we've never seen busy
                # (pre-launch race: ``prompt_async`` accepted but the
                # status map briefly lists the session as idle before
                # the agent boots), or ``state == "unknown"`` (entry
                # not in the status map at all). Both states are
                # acceptable transiently; both must resolve to busy
                # before the warmup deadline or we treat this as a
                # transport error rather than silently reporting an
                # empty completion. The empty-completion footgun is
                # exactly what bit dispatch_review's first cycle on
                # 2026-05-07 — a reviewer "completed" in 0.025 s of
                # wallclock with raw_response="".
                if time.monotonic() >= warmup_deadline:
                    logger.warning(
                        "OpenCode session %s never transitioned to busy "
                        "within %.0fs of prompt_async — treating as "
                        "transport-error (agent dispatch failed)",
                        session_id,
                        _STARTUP_GRACE_SECONDS,
                    )
                    return SessionResult(
                        status="transport-error",
                        wallclock_seconds=_elapsed(),
                        session_id=session_id,
                        raw_response="session never became busy",
                    )
            time.sleep(poll_interval_seconds)

        try:
            messages = _request(
                "GET", f"{server_url}/session/{session_id}/message"
            )
        except _TRANSPORT_EXC as e:
            logger.warning(
                "OpenCode message fetch failed for session %s: %s",
                session_id,
                e,
            )
            return SessionResult(
                status="transport-error",
                wallclock_seconds=_elapsed(),
                session_id=session_id,
                raw_response=str(e),
            )
        if not isinstance(messages, list):
            messages = []
        raw_text = _last_assistant_text(messages)
        parsed = _extract_last_json_object(raw_text)
        return SessionResult(
            status="completed",
            wallclock_seconds=_elapsed(),
            session_id=session_id,
            raw_response=raw_text,
            parsed_json=parsed,
        )

    finally:
        if session_id:
            try:
                _request("DELETE", f"{server_url}/session/{session_id}")
            except _TRANSPORT_EXC as e:
                logger.warning(
                    "OpenCode session DELETE failed for %s: %s — leaking "
                    "session, will be reclaimed at server restart",
                    session_id,
                    e,
                )


def run_worker_blocking(
    *,
    server_url: str,
    agent: str,
    tag: str,
    prompt: str,
    timeout_seconds: int = 900,
    poll_interval_seconds: float = 2.0,
    on_poll: Callable[[], None] | None = None,
) -> WorkerResult:
    """Run a one-shot OpenCode agent and classify the result against the
    conflict-driver outcome schema.

    Thin wrapper around :func:`run_session_blocking` that interprets the
    last parseable JSON object (per plan § 6.2 step 5) into one of:

    - ``resolved`` / ``unresolvable`` / ``rebase-failed`` from the
      worker's structured exit, or
    - ``rebase-failed`` if the session completed but the JSON contract
      was violated (no JSON, missing ``outcome`` key, or unrecognized
      value), or
    - ``timeout`` / ``transport-error`` propagated from the session
      lifecycle.

    Drivers whose workers do not emit this schema (review, implementer)
    should call :func:`run_session_blocking` directly and treat
    ``status == "completed"`` as success.

    The ``on_poll`` callback is forwarded verbatim to
    :func:`run_session_blocking`; see that function's docstring for the
    contract. Used by ``conflict_drive.py`` to refresh its heartbeat
    while a long-running rebase + conflict-resolve session is in flight.
    """
    session = run_session_blocking(
        server_url=server_url,
        agent=agent,
        tag=tag,
        prompt=prompt,
        timeout_seconds=timeout_seconds,
        poll_interval_seconds=poll_interval_seconds,
        on_poll=on_poll,
    )
    if session.status == "timeout":
        return WorkerResult(
            outcome="timeout",
            wallclock_seconds=session.wallclock_seconds,
            session_id=session.session_id,
            raw_response=session.raw_response,
        )
    if session.status == "transport-error":
        return WorkerResult(
            outcome="transport-error",
            wallclock_seconds=session.wallclock_seconds,
            session_id=session.session_id,
            raw_response=session.raw_response,
        )
    parsed = session.parsed_json or {}
    outcome_str = parsed.get("outcome")
    if outcome_str not in ("resolved", "unresolvable", "rebase-failed"):
        logger.warning(
            "OpenCode worker returned unexpected outcome %r — classifying "
            "as rebase-failed (session=%s)",
            outcome_str,
            session.session_id,
        )
        outcome_str = "rebase-failed"
    files_touched = parsed.get("files_touched") or []
    if not isinstance(files_touched, list):
        files_touched = []
    return WorkerResult(
        outcome=outcome_str,  # type: ignore[arg-type]
        files_touched=[str(f) for f in files_touched],
        wallclock_seconds=session.wallclock_seconds,
        session_id=session.session_id,
        raw_response=session.raw_response,
    )