creator/backend/agents.py

"""Provider layer: runs agent calls via the Claude CLI or OpenCode (MiniMax).

Both runners are independent. If a binary/key is missing, only the
respective provider fails — the other keeps running unchanged.

Role routing: config.resolve_role maps (run_provider, role) → (provider, model)
ACROSS stacks, so one run can generate on MiniMax and judge on Claude. If the
routed provider is unavailable, the call falls back to the run's provider.
"""

import asyncio
import heapq
import logging
import os
import re
import shutil
import signal
import sqlite3
import tempfile
import time
import urllib.request
from contextlib import asynccontextmanager
from pathlib import Path

from config import (PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS,
                    MAX_CONCURRENT_AGENTS_PER_TOPIC, MAX_CONCURRENT_INTERACTIVE,
                    resolve_role)

log = logging.getLogger("creator.agents")

_active_processes: dict[str, asyncio.subprocess.Process] = {}
_active_started: dict[str, float] = {}   # agent_key → wall-clock start (for the live runtime display)
_active_labels: dict[str, str] = {}      # agent_key → human-readable label (for display + events)


def active_agents(scope_prefix: str | None = None) -> list[dict]:
    """Currently running agents and how long they've been running. Filter by key prefix
    (e.g. f"blocks-{topic}-") for one topic. → [{key, label, runtime}] sorted longest-first."""
    now = time.time()
    out = [{"key": k, "label": _active_labels.get(k, ""), "runtime": round(now - t, 1)}
           for k, t in list(_active_started.items())
           if k in _active_processes and (not scope_prefix or k.startswith(scope_prefix))]
    return sorted(out, key=lambda a: -a["runtime"])

# Cancelled scopes (key prefixes, symmetric to kill_process). An agent whose
# key starts with one of these prefixes aborts BEFORE the spawn — so agents WAITING
# in the semaphore queue are also stopped immediately on abort instead of still starting.
_cancelled_prefixes: set[str] = set()


def cancel_scope(prefix: str) -> None:
    _cancelled_prefixes.add(prefix)


def clear_scope(prefix: str) -> None:
    _cancelled_prefixes.discard(prefix)


def _scope_cancelled(agent_key: str) -> bool:
    return any(agent_key.startswith(p) for p in _cancelled_prefixes)

# Caps the real CLI processes — independent of the pipeline semaphore in
# generator.py. The acquire happens BEFORE the spawn so that queue wait time
# does not count against the agent timeout.
class _PrioritySemaphore:
    """asyncio.Semaphore variant: when slots are scarce, the LOWEST priority number is served first
    (FIFO within the same priority). Lets earlier pipeline columns grab agents before later ones."""
    def __init__(self, value: int):
        self._value = value
        self._waiters: list = []  # heap of [priority, seq, future]
        self._seq = 0

    async def acquire(self, priority: int = 100):
        if self._value > 0:
            self._value -= 1
            return
        fut = asyncio.get_event_loop().create_future()
        entry = [priority, self._seq, fut]
        self._seq += 1
        heapq.heappush(self._waiters, entry)
        try:
            await fut  # release() hands us the slot directly (no value change)
        except BaseException:
            entry[2] = None  # tombstone so release() skips this dead waiter
            if fut.done() and not fut.cancelled():
                self.release()  # granted just before we were cancelled → pass it on
            raise

    def release(self):
        while self._waiters:
            entry = heapq.heappop(self._waiters)
            if entry[2] is not None and not entry[2].done():
                entry[2].set_result(None)  # hand the slot straight to the highest-priority waiter
                return
        self._value += 1


_batch_sem = _PrioritySemaphore(MAX_CONCURRENT_AGENTS)
_interactive_sem = asyncio.Semaphore(MAX_CONCURRENT_INTERACTIVE)

# Per-topic caps (lazily created): each topic gets its own priority semaphore of size
# MAX_CONCURRENT_AGENTS_PER_TOPIC, nested INSIDE the global _batch_sem. Priority-based too, so the
# per-topic queue can't undo the global priority when one topic is the only load.
_topic_sems: dict[str, _PrioritySemaphore] = {}

# Smaller index = higher priority. Board 1 (inventory) first — it feeds everything.
# Within board 2 the LATE stages win (outline → artefacts → … → subblocks): finish cards
# instead of opening new WIP, so the makespan tail block gets slots before fresh work.
_STAGE_PRIORITY = ("research", "ingest", "cluster", "pair", "clarify", "naming", "filter",
                   "dedup", "grouping", "gruppierung", "supplement", "outline", "artifact",
                   "question", "relevance", "level", "facts", "subblock")


def _agent_priority(key: str) -> int:
    for i, tag in enumerate(_STAGE_PRIORITY):
        if f"-{tag}-" in key or key.endswith(f"-{tag}"):
            return i
    return len(_STAGE_PRIORITY)  # unmatched keys (guide board, …) after everything


@asynccontextmanager
async def _batch_gate(scope: str | None, priority: int):
    """Per-topic slot FIRST (fair), then the GLOBAL slot by priority (earlier columns win when
    agents are scarce). Order matters — a waiter holds only its per-topic slot while queueing globally."""
    topic_sem = _topic_sems.setdefault(scope, _PrioritySemaphore(MAX_CONCURRENT_AGENTS_PER_TOPIC)) if scope else None
    if topic_sem is not None:
        await topic_sem.acquire(priority)
    await _batch_sem.acquire(priority)
    try:
        yield
    finally:
        _batch_sem.release()
        if topic_sem is not None:
            topic_sem.release()

# Space OpenCode starts: processes starting simultaneously collide on the internal
# session DB ("database is locked", exit after <1s). Token bucket instead of a lock
# held through spawn+sleep: the lock only assigns a start slot, the sleep happens
# outside — a wave of starts is spaced by OPENCODE_START_DELAY without a global convoy.
_opencode_start_lock = asyncio.Lock()
_OPENCODE_START_DELAY = float(os.getenv("OPENCODE_START_DELAY", "0.5"))
_opencode_next_start = 0.0


async def _opencode_slot() -> None:
    global _opencode_next_start
    loop = asyncio.get_running_loop()
    async with _opencode_start_lock:
        now = loop.time()
        start_at = max(now, _opencode_next_start)
        _opencode_next_start = start_at + _OPENCODE_START_DELAY
    await asyncio.sleep(max(0.0, start_at - now))

_SLIM_CONFIG = Path(__file__).resolve().parent.parent / "dev-ops" / "opencode-slim.json"

# Capability → Claude --allowedTools
_CLAUDE_TOOLS = {
    "full": "Write,Bash,Read,WebSearch,WebFetch",
    "files": "Read,Bash,Write",
    "read": "Read",
    "none": None,
}

# Capability → OpenCode agent (tool permissions defined in dev-ops/opencode.json)
_OPENCODE_AGENTS = {
    "full": "full",
    "files": "files",
    "read": "readonly",
    "none": "text",
}


def provider_available(provider: str) -> bool:
    cfg = PROVIDERS.get(provider)
    if not cfg:
        return False
    if shutil.which(cfg["cli"]) is None:
        return False
    env_key = cfg.get("env_key")
    if env_key and not os.environ.get(env_key):
        return False
    check_url = cfg.get("check_url")
    if check_url:
        try:
            urllib.request.urlopen(check_url, timeout=1)
        except Exception:
            return False
    return True


# Availability cache for role routing: the routed target is probed at most once per
# TTL (check_url providers would otherwise block the loop on every call).
_avail_cache: dict[str, tuple[float, bool]] = {}
_AVAIL_TTL = 60.0


def _available_cached(provider: str) -> bool:
    now = time.monotonic()
    hit = _avail_cache.get(provider)
    if hit and now - hit[0] < _AVAIL_TTL:
        return hit[1]
    ok = provider_available(provider)
    _avail_cache[provider] = (now, ok)
    return ok


def _kill(process) -> None:
    """Kill the agent and its child processes via the process group (otherwise the
    children spawned by the CLI survive, keep the pipes open and block communicate())."""
    try:
        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
    except (ProcessLookupError, PermissionError):
        try:
            process.kill()
        except ProcessLookupError:
            pass


def kill_process(agent_key_prefix: str) -> None:
    """Kill all active processes whose key starts with the prefix (covers -plan/-w1…)."""
    for key, process in list(_active_processes.items()):
        if process.returncode is not None:  # clean up dead entries while iterating
            _active_processes.pop(key, None)
            _active_started.pop(key, None)
            continue
        if key.startswith(agent_key_prefix):
            log.debug("kill agent %s", key)
            _kill(process)


# Event sink for the pipeline history (injected by main.py lifespan as database.add_event —
# agents.py stays DB-free). Called fire-and-forget for every finished BATCH agent.
on_event = None


async def run_agent(
    agent_key: str,
    prompt: str,
    timeout: int,
    provider: str = DEFAULT_PROVIDER,
    role: str = "fast",
    capabilities: str = "none",
    lane: str = "batch",
    scope: str | None = None,
    on_line=None,
    label: str = "",
) -> tuple[int, str, str]:
    if os.getenv("CREATOR_FAKE_AGENTS"):  # Sekunden-Smoke: deterministische Antworten statt LLM
        import fake_agents
        return await fake_agents.respond(agent_key, prompt, capabilities)
    if _scope_cancelled(agent_key):  # before queueing: don't even enter the queue
        return 1, "", "cancelled"
    if provider not in PROVIDERS:
        return 1, "", f"Unknown provider: {provider}"
    run_provider = provider
    provider, model = resolve_role(run_provider, role)
    if provider != run_provider and not _available_cached(provider):
        provider, model = run_provider, PROVIDERS[run_provider].get(role, "")
    if not model:
        return 1, "", f"No model for role '{role}' (provider: {provider})"
    if shutil.which(PROVIDERS[provider]["cli"]) is None:
        return 1, "", f"CLI '{PROVIDERS[provider]['cli']}' not installed (provider: {provider})"
    queued = time.monotonic()
    gate = _interactive_sem if lane == "interactive" else _batch_gate(scope, _agent_priority(agent_key))
    async with gate:
        if _scope_cancelled(agent_key):  # after the acquire: cancelled in the queue → no spawn
            return 1, "", "cancelled"
        wait_ms = int((time.monotonic() - queued) * 1000)
        start = time.monotonic()
        status = "error"
        rc = None
        err_tail = ""
        try:
            log.info("agent %s: %s %s (role %s)", agent_key, provider, model, role)
            if PROVIDERS[provider]["cli"] == "opencode":
                res = await _run_opencode(agent_key, prompt, timeout, provider, model, capabilities, on_line=on_line, label=label)
            else:
                res = await _run_claude_cli(agent_key, prompt, timeout, model, capabilities, label=label)
            rc = res[0]
            status = "ok" if rc == 0 else ("killed" if rc is not None and rc < 0 else "error")
            if rc not in (0, None) and rc >= 0:
                err_tail = (res[2] or res[1] or "").strip()[-300:]  # diagnosis: rc=1 without stderr is opaque
            return res
        except asyncio.TimeoutError:
            status = "timeout"
            raise
        except asyncio.CancelledError:
            status = "cancelled"
            raise
        finally:
            if on_event is not None and scope is not None:  # batch pipeline only, never fatal
                try:
                    meta = {"provider": provider, "model": model, "role": role, "rc": rc}
                    if err_tail:
                        meta["stderr"] = err_tail
                    if PROVIDERS[provider]["cli"] == "opencode":  # token accounting per agent
                        if (tok := await asyncio.to_thread(_session_tokens, agent_key)):
                            meta["tokens"] = tok
                    await on_event(topic=scope, kind="agent", key=agent_key, label=label,
                                   status=status, dur_ms=int((time.monotonic() - start) * 1000),
                                   wait_ms=wait_ms, meta=meta)
                except Exception:
                    log.debug("on_event failed", exc_info=True)


async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None, timeout: int, stagger: bool = False, on_line=None, label: str = "", env: dict | None = None) -> tuple[int, str, str]:
    start = time.monotonic()

    async def spawn():
        return await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE if stdin_data is not None else asyncio.subprocess.DEVNULL,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            start_new_session=True,  # own process group → killpg also kills child processes
            env=env,
        )

    if stagger:
        await _opencode_slot()  # spaced start slot; spawn itself is not serialized
    process = await spawn()
    # Collision-safe tracking: identical keys (e.g. same chunk label from parallel cards)
    # get a ~n suffix — prefix-based kill/cancel still matches, nothing becomes an orphan.
    track_key = agent_key
    n = 2
    while track_key in _active_processes:
        track_key = f"{agent_key}~{n}"
        n += 1
    _active_processes[track_key] = process
    _active_started[track_key] = time.time()
    _active_labels[track_key] = label
    try:
        try:
            if on_line is not None:
                # Streaming path: read stdout line by line, hand each raw line to on_line LIVE.
                out_chunks: list[str] = []

                async def _pump():
                    async for raw in process.stdout:
                        s = raw.decode("utf-8", errors="replace")
                        out_chunks.append(s)
                        try:
                            on_line(s)
                        except Exception:
                            log.debug("on_line callback failed", exc_info=True)
                await asyncio.wait_for(_pump(), timeout=timeout)
                await process.wait()
                stderr_b = await process.stderr.read()
                stdout, stderr = "".join(out_chunks).encode("utf-8"), stderr_b
            else:
                stdout, stderr = await asyncio.wait_for(
                    process.communicate(input=stdin_data),
                    timeout=timeout,
                )
        except asyncio.TimeoutError:
            _kill(process)
            try:
                await asyncio.wait_for(process.wait(), timeout=5)
            except asyncio.TimeoutError:
                pass
            log.info("agent %s: timeout after %ds", agent_key, timeout)
            raise
        log.info(
            "agent %s: exit %s after %.1fs (%d bytes stdout)",
            agent_key, process.returncode, time.monotonic() - start, len(stdout),
        )
        return process.returncode, stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace")
    finally:
        # Pop only on identity: a slot restart under the same key must not evict
        # the NEW process from tracking.
        if _active_processes.get(track_key) is process:
            del _active_processes[track_key]
            _active_started.pop(track_key, None)
            _active_labels.pop(track_key, None)


async def _run_claude_cli(agent_key: str, prompt: str, timeout: int, model: str, capabilities: str, label: str = "") -> tuple[int, str, str]:
    cfg = PROVIDERS["claude"]
    cmd = [cfg["cli"], "-p", "--model", model]
    tools = _CLAUDE_TOOLS.get(capabilities)
    if tools:
        cmd += ["--allowedTools", tools]
    cmd += ["--dangerously-skip-permissions"]
    return await _communicate(agent_key, cmd, prompt.encode("utf-8"), timeout, label=label)


_OPENCODE_DB = Path.home() / ".local" / "share" / "opencode" / "opencode.db"


def _session_tokens(agent_key: str) -> dict | None:
    """Token counters of the newest OpenCode session titled `agent_key` (set via run --title).
    Best-effort read-only lookup — None when DB/row is missing; never fails the agent.
    Timeouts count too: their sessions consumed tokens, that waste should be visible."""
    try:
        con = sqlite3.connect(f"file:{_OPENCODE_DB}?mode=ro", uri=True, timeout=1)
        try:
            row = con.execute(
                "SELECT tokens_input, tokens_output, tokens_reasoning,"
                " tokens_cache_read, tokens_cache_write"
                " FROM session WHERE title=? ORDER BY time_created DESC LIMIT 1",
                (agent_key,)).fetchone()
        finally:
            con.close()
    except Exception:
        return None
    if row is None:
        return None
    keys = ("input", "output", "reasoning", "cache_read", "cache_write")
    return {k: int(v or 0) for k, v in zip(keys, row)}


async def _run_opencode(agent_key: str, prompt: str, timeout: int, provider: str, model: str, capabilities: str, on_line=None, label: str = "") -> tuple[int, str, str]:
    cfg = PROVIDERS[provider]
    # Prompt via temp file instead of argv (ARG_MAX protection for large project prompts)
    with tempfile.NamedTemporaryFile("w", suffix=".md", delete=False, encoding="utf-8", dir=tempfile.gettempdir()) as f:
        f.write(prompt)
        prompt_path = Path(f.name)
    # The positional message MUST come before -f: -f is an array flag and
    # would otherwise eat the text as a second file name ("File not found").
    cmd = [
        cfg["cli"], "run",
        "Folge exakt den Anweisungen in der angehängten Datei. Sie sind der vollständige Auftrag.",
        "-m", model,
        "--agent", _OPENCODE_AGENTS.get(capabilities, "text"),
        "--dangerously-skip-permissions",
        "--title", agent_key,  # token accounting: joins the OpenCode session to our event
        "-f", str(prompt_path),
    ]
    if on_line is not None:
        cmd += ["--format", "json"]  # raw JSON events → parsed live by on_line
    env = None
    if capabilities != "full":
        # Batch agents (files/readonly/text) never use the web MCPs, but opencode starts
        # every configured MCP server PER PROCESS (~3 procs / ~300 MB each). Point them
        # at the mcp-free config copy; only `full` (research/supplement) keeps the servers.
        env = {**os.environ, "OPENCODE_CONFIG": str(_SLIM_CONFIG)}
    try:
        rc, stdout, stderr = await _communicate(agent_key, cmd, None, timeout, stagger=True, on_line=on_line, label=label, env=env)
        return rc, (stdout if on_line is not None else _clean_opencode_output(stdout)), stderr
    finally:
        prompt_path.unlink(missing_ok=True)


_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")


def _clean_opencode_output(text: str) -> str:
    """Strip ANSI codes and the leading banner ("> agent · model")."""
    text = _ANSI_RE.sub("", text)
    lines = text.splitlines()
    while lines and (not lines[0].strip() or lines[0].lstrip().startswith(">")):
        lines.pop(0)
    return "\n".join(lines).strip()