85 lines
2.2 KiB
Python
85 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import abc
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Sequence
|
|
|
|
from claridoc.models import ProviderSpec
|
|
|
|
|
|
class ProviderError(RuntimeError):
|
|
"""Base provider invocation error."""
|
|
|
|
|
|
class ProviderUnavailable(ProviderError):
|
|
"""Raised when a provider binary, SDK, or authentication surface is unavailable."""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProviderRequest:
|
|
stage: str
|
|
prompt: str
|
|
workdir: Path
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProviderResponse:
|
|
text: str
|
|
provider: str
|
|
model: str = ""
|
|
command: list[str] = field(default_factory=list)
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class Provider(abc.ABC):
|
|
def __init__(self, spec: ProviderSpec):
|
|
self.spec = spec
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.spec.provider
|
|
|
|
@abc.abstractmethod
|
|
def generate(self, request: ProviderRequest) -> ProviderResponse:
|
|
raise NotImplementedError
|
|
|
|
@abc.abstractmethod
|
|
def check(self) -> dict[str, Any]:
|
|
raise NotImplementedError
|
|
|
|
|
|
def run_command(
|
|
command: Sequence[str],
|
|
*,
|
|
prompt: str,
|
|
cwd: Path,
|
|
timeout_seconds: int,
|
|
env: dict[str, str] | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
try:
|
|
completed = subprocess.run(
|
|
list(command),
|
|
input=prompt,
|
|
text=True,
|
|
capture_output=True,
|
|
cwd=cwd,
|
|
timeout=timeout_seconds,
|
|
check=False,
|
|
env=env,
|
|
)
|
|
except FileNotFoundError as exc:
|
|
raise ProviderUnavailable(f"provider executable not found: {command[0]}") from exc
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise ProviderError(f"provider timed out after {timeout_seconds}s: {command[0]}") from exc
|
|
if completed.returncode != 0:
|
|
stderr = completed.stderr.strip()
|
|
stdout = completed.stdout.strip()
|
|
detail = stderr or stdout or "no diagnostic output"
|
|
if len(detail) > 2000:
|
|
detail = detail[-2000:]
|
|
raise ProviderError(f"provider exited with code {completed.returncode}: {detail}")
|
|
return completed
|