engines/dotcache_runner.py

from __future__ import annotations

import json
import os
import shlex
import subprocess
import sys
from pathlib import Path
from typing import Any

from engines.live_request import default_dotcache_runner_script


def run_dotcache_live(request: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
    return _run_subprocess_runner(
        env_var="DOTCACHE_SPACE_DOTCACHE_RUNNER_CMD",
        request=request,
        runner_name="dotcache",
    )


def _run_subprocess_runner(*, env_var: str, request: dict[str, Any], runner_name: str) -> tuple[dict[str, Any], list[str]]:
    command = os.getenv(env_var)
    if command:
        argv = shlex.split(command)
    else:
        repo_root = Path(__file__).resolve().parents[1]
        argv = [sys.executable, str(default_dotcache_runner_script(repo_root))]
    completed = subprocess.run(
        argv,
        input=json.dumps(request),
        text=True,
        capture_output=True,
        check=False,
    )
    logs = [
        f"{runner_name} runner command: {' '.join(argv)}",
        f"{runner_name} runner exit code: {completed.returncode}",
    ]
    if completed.stderr.strip():
        logs.append(completed.stderr.strip())
    if completed.returncode != 0:
        raise RuntimeError("\n".join(logs))
    try:
        payload = json.loads(completed.stdout)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Live {runner_name} runner returned invalid JSON.") from exc
    return payload, logs