plano/cli/planoai/local_agent_warning.py

"""Detect local-agent provider entries in a Plano config and warn the
operator that the host is about to spawn a local CLI binary with the same
filesystem, shell, and network capabilities as the user running planoai.

Local-agent providers (e.g. ``claude-cli``) are an entirely different
trust class from stateless network LLM providers (``openai``,
``anthropic``, ``gemini``, ...): the bridge runs inside brightstaff and
shells out to a local binary for every request, so a misconfigured
production deployment would expose the host to whatever the spawned
agent can do — which, for tools like Claude Code, is "anything the
operator can do at the shell".

This module is intentionally additive and side-effect free until the
caller invokes :func:`maybe_warn_local_agent_providers`. The set of
known local-agent provider interfaces lives in
:data:`LOCAL_AGENT_PROVIDER_INTERFACES`; adding a future entry (codex,
chatgpt-cli, opencode, hermes, ...) is a one-line change.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Iterable

from rich.console import Console
from rich.panel import Panel

from planoai.consts import PLANO_STATE_DIR

# Provider interfaces whose runtime spawns a local CLI subprocess with
# host filesystem / shell access. The string here is matched against the
# config's ``provider_interface`` field AND against the ``<prefix>/...``
# in ``model:`` and ``name:`` fields, so configs that rely on the
# Python-side autofill (``model: claude-cli/*`` only) are still detected
# before that autofill runs.
#
# Add new entries here as additional local-agent bridges are implemented
# (e.g. a future ``codex-cli`` or ``chatgpt-cli`` bridge that spawns the
# Codex CLI). This is the *only* line that needs to change to extend the
# warning's coverage.
LOCAL_AGENT_PROVIDER_INTERFACES: tuple[str, ...] = ("claude-cli",)

# Persistent ack lives next to the rest of the per-user planoai state
# (run/, bin/, plugins/, ...). Operators can ``rm`` this file to undo.
ACK_FILE_PATH = os.path.join(PLANO_STATE_DIR, "local_agent_ack.json")

# Env-var fallback for the ``--ack-local-agents`` CLI flag. Truthy values
# are 1/true/yes (case-insensitive); everything else is treated as unset.
ACK_ENV_VAR = "PLANO_ACK_LOCAL_AGENTS"

# Public docs page. The Sphinx source lives at
# ``docs/source/resources/local_agent_providers.rst`` and is published to
# https://docs.planoai.dev (CNAME at ``docs/CNAME``).
DOCS_LEARN_MORE = "https://docs.planoai.dev/resources/local_agent_providers.html"


@dataclass(frozen=True)
class LocalAgentProvider:
    """A single ``model_providers`` entry that resolves to a local-agent
    bridge. ``name`` and ``model`` come straight from the config, while
    ``interface`` is the canonical key used for ack persistence."""

    interface: str
    name: str
    model: str


def _truthy_env(value: str | None) -> bool:
    if not value:
        return False
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _interface_for_entry(entry: dict) -> str | None:
    """Return the canonical local-agent interface name for ``entry``, or
    ``None`` if the entry isn't a local-agent provider.

    Matching is intentionally permissive so that minimally-configured
    entries — i.e. just ``model: claude-cli/*`` before the Python
    autofill runs — are still detected. The first match wins and is
    returned; multiple matches against the same interface collapse.
    """

    if not isinstance(entry, dict):
        return None

    provider_interface = (entry.get("provider_interface") or "").strip()
    provider = (entry.get("provider") or "").strip()
    model = str(entry.get("model") or "").strip()
    name = str(entry.get("name") or "").strip()

    for interface in LOCAL_AGENT_PROVIDER_INTERFACES:
        if provider_interface == interface or provider == interface:
            return interface
        prefix = f"{interface}/"
        if model.startswith(prefix) or name.startswith(prefix):
            return interface

    return None


def detect_local_agent_providers(config: dict) -> list[LocalAgentProvider]:
    """Walk ``config`` and return every ``model_providers`` entry whose
    ``provider_interface`` falls in :data:`LOCAL_AGENT_PROVIDER_INTERFACES`.

    Order is preserved so the warning lists providers in declaration
    order. Both the new ``model_providers`` key and the legacy
    ``llm_providers`` key are consulted, mirroring the rest of the CLI.
    """

    if not isinstance(config, dict):
        return []

    providers = config.get("model_providers")
    if not isinstance(providers, list):
        providers = config.get("llm_providers") or []

    found: list[LocalAgentProvider] = []
    for entry in providers:
        interface = _interface_for_entry(entry)
        if interface is None:
            continue
        model = str(entry.get("model") or "").strip()
        name = str(entry.get("name") or "").strip() or model or interface
        found.append(LocalAgentProvider(interface=interface, name=name, model=model))
    return found


def _interfaces_in(providers: Iterable[LocalAgentProvider]) -> set[str]:
    return {p.interface for p in providers}


def load_acknowledged_interfaces(ack_path: str = ACK_FILE_PATH) -> set[str]:
    """Read the ack file and return the set of acknowledged provider
    interfaces. Missing or malformed files are treated as "no ack",
    never as a hard error, so a half-written ack file degrades to "warn
    again" instead of crashing ``planoai up``."""

    try:
        with open(ack_path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError):
        return set()

    if not isinstance(data, dict):
        return set()
    raw = data.get("acknowledged")
    if not isinstance(raw, list):
        return set()
    return {str(item) for item in raw if isinstance(item, str)}


def write_acknowledgement(
    interfaces: Iterable[str],
    ack_path: str = ACK_FILE_PATH,
) -> set[str]:
    """Persist ``interfaces`` (merged with anything already on disk) to
    the ack file. Returns the full acknowledged set after the write so
    callers can render an "acknowledged: X, Y" line."""

    merged = load_acknowledged_interfaces(ack_path) | set(interfaces)
    payload = {
        "acknowledged": sorted(merged),
        "ack_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
    }
    os.makedirs(os.path.dirname(ack_path), exist_ok=True)
    with open(ack_path, "w", encoding="utf-8") as f:
        json.dump(payload, f, indent=2, sort_keys=True)
        f.write("\n")
    return merged


def _render_panel(
    console: Console,
    pending: list[LocalAgentProvider],
) -> None:
    """Render the (small) reminder panel for ``pending``. Callers must
    ensure ``pending`` is non-empty.

    The panel is intentionally compact: the title names the interface(s),
    the body is two short lines (capability summary + dismiss hint), and
    the "Learn more" link points at the published Sphinx docs. Operators
    who want the full trust-model write-up follow the link.
    """

    interfaces = sorted({p.interface for p in pending})
    interfaces_csv = ", ".join(interfaces)

    # Show user-set names parenthetically, but skip ``<interface>/...``
    # values — those are just the model id (or the autofilled placeholder)
    # and add no information beyond the interface itself.
    extra_names = sorted(
        {
            p.name
            for p in pending
            if p.name
            and p.name != p.interface
            and not any(
                p.name.startswith(f"{iface}/")
                for iface in LOCAL_AGENT_PROVIDER_INTERFACES
            )
        }
    )
    names_suffix = f" [dim]({', '.join(extra_names)})[/dim]" if extra_names else ""

    plural = len(interfaces) > 1
    pronoun = "they spawn" if plural else "it spawns"

    body = (
        f"[bold]{interfaces_csv}[/bold]{names_suffix} is a local-agent provider — "
        f"{pronoun} a CLI subprocess that runs as you (full filesystem and shell "
        f"access). For local development only.\n\n"
        f"[dim]Learn more:[/dim] [link={DOCS_LEARN_MORE}]"
        f"{DOCS_LEARN_MORE}[/link]\n"
        f"[dim]Hide this:[/dim]  [cyan]planoai up --ack-local-agents[/cyan]"
    )

    console.print(
        Panel(
            body,
            title=f"⚠  Local-agent provider detected ({interfaces_csv})",
            title_align="left",
            border_style="yellow",
            padding=(0, 2),
        )
    )


def maybe_warn_local_agent_providers(
    config: dict,
    console: Console,
    *,
    ack_flag: bool = False,
    ack_path: str = ACK_FILE_PATH,
    env: dict | None = None,
) -> bool:
    """Show the local-agent warning panel if appropriate and return
    ``True`` iff the panel was rendered.

    Resolution order, top to bottom:

    1. No local-agent providers in config → no-op.
    2. ``ack_flag`` (the ``--ack-local-agents`` CLI flag) **or** the
       :data:`ACK_ENV_VAR` env var truthy → write/update the ack file
       so it covers every triggering interface, print one ✓ confirmation
       line, suppress the panel.
    3. Existing ack file already covers every triggering interface →
       print a single dim INFO line and suppress the panel.
    4. Otherwise → render the panel for the *un-acked* interfaces only
       (e.g. acknowledged ``claude-cli`` doesn't suppress a fresh
       warning when the operator later adds a hypothetical ``codex``).
    """

    env = env if env is not None else os.environ
    detected = detect_local_agent_providers(config)
    if not detected:
        return False

    ack_via_env = _truthy_env(env.get(ACK_ENV_VAR))
    if ack_flag or ack_via_env:
        new_set = _interfaces_in(detected)
        write_acknowledgement(new_set, ack_path=ack_path)
        ack_csv = ", ".join(sorted(new_set))
        console.print(
            f"[green]✓[/green] Acknowledged local-agent provider: "
            f"[bold]{ack_csv}[/bold] [dim](won't warn again)[/dim]"
        )
        return False

    acknowledged = load_acknowledged_interfaces(ack_path)
    pending = [p for p in detected if p.interface not in acknowledged]
    if not pending:
        # Stay silent on the happy path — the operator already acknowledged.
        # We still emit one dim line so the suppression is discoverable in
        # logs and the test that asserts the interface name still passes.
        ack_csv = ", ".join(sorted(_interfaces_in(detected)))
        console.print(f"[dim]local-agent provider: {ack_csv} (acknowledged)[/dim]")
        return False

    _render_panel(console, pending)
    return True


__all__ = [
    "ACK_ENV_VAR",
    "ACK_FILE_PATH",
    "DOCS_LEARN_MORE",
    "LOCAL_AGENT_PROVIDER_INTERFACES",
    "LocalAgentProvider",
    "detect_local_agent_providers",
    "load_acknowledged_interfaces",
    "maybe_warn_local_agent_providers",
    "write_acknowledgement",
]