95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from claridoc.providers.base import Provider, ProviderRequest, ProviderResponse, ProviderUnavailable, run_command
|
|
|
|
|
|
class CodexProvider(Provider):
|
|
"""Non-interactive adapter for `codex exec`.
|
|
|
|
The default sandbox is read-only because document generation only needs the
|
|
prompt and stdout. Override command/extra_args in pipeline configuration when
|
|
an organization's Codex wrapper uses different flags.
|
|
"""
|
|
|
|
def generate(self, request: ProviderRequest) -> ProviderResponse:
|
|
options = self.spec.options
|
|
binary = str(options.get("binary") or os.environ.get("CLARIDOC_CODEX_BIN") or "codex")
|
|
custom = options.get("command")
|
|
output_path: Path | None = None
|
|
if custom:
|
|
command = _command_list(custom)
|
|
else:
|
|
handle = tempfile.NamedTemporaryFile(prefix="claridoc-codex-", suffix=".txt", delete=False)
|
|
handle.close()
|
|
output_path = Path(handle.name)
|
|
command = [binary, "exec"]
|
|
sandbox = str(options.get("sandbox", "read-only"))
|
|
if sandbox:
|
|
command.extend(["--sandbox", sandbox])
|
|
if bool(options.get("skip_git_repo_check", True)):
|
|
command.append("--skip-git-repo-check")
|
|
if self.spec.model:
|
|
command.extend(["--model", self.spec.model])
|
|
command.extend(["--output-last-message", str(output_path)])
|
|
command.extend(_string_list(options.get("extra_args", []), "codex extra_args"))
|
|
command.append("-")
|
|
|
|
try:
|
|
completed = run_command(
|
|
command,
|
|
prompt=request.prompt,
|
|
cwd=request.workdir,
|
|
timeout_seconds=self.spec.timeout_seconds,
|
|
env=os.environ.copy(),
|
|
)
|
|
if output_path and output_path.exists():
|
|
text = output_path.read_text(encoding="utf-8").strip()
|
|
if not text:
|
|
text = completed.stdout.strip()
|
|
else:
|
|
text = completed.stdout.strip()
|
|
finally:
|
|
if output_path:
|
|
output_path.unlink(missing_ok=True)
|
|
if not text:
|
|
raise ProviderUnavailable("Codex returned an empty response")
|
|
return ProviderResponse(text=text, provider=self.name, model=self.spec.model, command=command)
|
|
|
|
def check(self) -> dict[str, Any]:
|
|
binary = str(self.spec.options.get("binary") or os.environ.get("CLARIDOC_CODEX_BIN") or "codex")
|
|
custom = self.spec.options.get("command")
|
|
executable = _command_list(custom)[0] if custom else binary
|
|
found = shutil.which(executable) if not Path(executable).is_file() else executable
|
|
return {
|
|
"provider": self.name,
|
|
"available": bool(found),
|
|
"executable": str(found or executable),
|
|
"mode": "custom-command" if custom else "codex exec",
|
|
"note": "Authentication is verified only by a live invocation.",
|
|
}
|
|
|
|
|
|
def _command_list(value: Any) -> list[str]:
|
|
if isinstance(value, str):
|
|
result = shlex.split(value)
|
|
elif isinstance(value, list):
|
|
result = [str(item) for item in value]
|
|
else:
|
|
raise ProviderUnavailable("codex options.command must be a string or array")
|
|
if not result:
|
|
raise ProviderUnavailable("codex options.command is empty")
|
|
return result
|
|
|
|
|
|
def _string_list(value: Any, name: str) -> list[str]:
|
|
if not isinstance(value, list):
|
|
raise ProviderUnavailable(f"{name} must be an array")
|
|
return [str(item) for item in value]
|