activity-core/src/activity_core/llm_client.py

"""llm-connect adapter for instruction execution.

activity-core deliberately talks to llm-connect over its small HTTP surface
instead of importing provider-specific SDKs. This keeps the activity worker on
owned infrastructure while leaving provider selection, API keys, and model
routing behind the existing llm-connect boundary.
"""

from __future__ import annotations

import os
from typing import Any

import httpx


class DisabledLLMClient:
    """LLM client used when no llm-connect endpoint is configured."""

    last_response_metadata: dict[str, Any] | None = None

    def complete(
        self,
        prompt: str,
        model: str = "",
        config: dict[str, Any] | None = None,
    ) -> str:  # noqa: ARG002
        raise RuntimeError("LLM_CONNECT_URL is not configured")


class LLMConnectClient:
    """Small synchronous client for llm-connect server mode."""

    def __init__(self, base_url: str, timeout_seconds: float = 300.0) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout_seconds = timeout_seconds
        self.last_response_metadata: dict[str, Any] | None = None

    def complete(
        self,
        prompt: str,
        model: str = "",
        config: dict[str, Any] | None = None,
    ) -> str:
        run_config = dict(config or {})
        if model and "model_name" not in run_config:
            run_config["model_name"] = model
        run_config.setdefault("timeout_seconds", int(self.timeout_seconds))
        payload: dict[str, Any] = {
            "prompt": prompt,
            "config": run_config,
        }
        resp = httpx.post(
            f"{self.base_url}/execute",
            json=payload,
            timeout=self.timeout_seconds,
        )
        if resp.status_code >= 400:
            raise RuntimeError(_llm_connect_error_text(resp))
        data = resp.json()
        self.last_response_metadata = _extract_response_metadata(data)
        content = data.get("content")
        if not isinstance(content, str):
            raise ValueError("llm-connect response missing string content")
        return content


_SAFE_ERROR_KEYS = ("error", "provider_status", "provider", "model")
_MAX_ERROR_MESSAGE_CHARS = 400


def _llm_connect_error_text(resp: httpx.Response) -> str:
    """Describe an llm-connect error using its own diagnostic body.

    llm-connect maps every provider API error onto HTTP 502 and puts the real
    cause in the body (``error``, ``provider_status``) — see
    ``llm_connect/server.py::_error_response``. Reporting only the transport
    status makes a rejected provider key look identical to a dead gateway,
    which is exactly the ambiguity ACTIVITY-WP-0031-T03 evidence must not have.

    llm-connect sanitizes its own ``message`` before returning it; this keeps a
    bounded copy of that text plus an allowlist of non-secret fields.
    """
    base = f"llm-connect returned HTTP {resp.status_code}"
    try:
        body = resp.json()
    except ValueError:
        return base
    if not isinstance(body, dict):
        return base

    parts = [
        f"{key}={body[key]}"
        for key in _SAFE_ERROR_KEYS
        if body.get(key) not in (None, "")
    ]
    message = body.get("message")
    if isinstance(message, str) and message.strip():
        parts.append(f"message={message.strip()[:_MAX_ERROR_MESSAGE_CHARS]}")
    if not parts:
        return base
    return f"{base}: " + "; ".join(parts)


_SAFE_RESPONSE_METADATA_KEYS = {
    "finish_reason",
    "usage",
    "model",
    "model_name",
    "provider",
    "request_id",
    "response_id",
    "trace_id",
    "latency_ms",
    "duration_ms",
    "elapsed_ms",
    "created",
    "created_at",
}


def _extract_response_metadata(data: dict[str, Any]) -> dict[str, Any]:
    """Keep non-secret llm-connect diagnostics alongside the returned content."""
    return {
        key: value for key, value in data.items()
        if key in _SAFE_RESPONSE_METADATA_KEYS and _json_safe(value)
    }


def _json_safe(value: Any) -> bool:
    try:
        import json

        json.dumps(value)
    except (TypeError, ValueError):
        return False
    return True


def get_llm_client() -> DisabledLLMClient | LLMConnectClient:
    base_url = os.environ.get("LLM_CONNECT_URL", "").strip()
    if not base_url:
        return DisabledLLMClient()
    timeout = float(os.environ.get("LLM_CONNECT_TIMEOUT_SECONDS", "300"))
    return LLMConnectClient(base_url, timeout)