Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,10 @@ COPY --from=envoy /usr/local/bin/envoy /usr/local/bin/envoy

WORKDIR /app

RUN pip install --no-cache-dir uv
# Pin uv to >=0.11.11; older versions bundle rustls-webpki 0.103.10 which is
# flagged by GHSA-82j2-j2ch-gfr8 (DoS via panic on malformed CRL BIT STRING).
# uv 0.11.11+ ships rustls-webpki 0.103.13.
RUN pip install --no-cache-dir 'uv>=0.11.11'

COPY cli/pyproject.toml ./
COPY cli/uv.lock ./
Expand Down
59 changes: 59 additions & 0 deletions cli/planoai/config_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,11 +39,64 @@
CHATGPT_DEFAULT_ORIGINATOR = "codex_cli_rs"
CHATGPT_DEFAULT_USER_AGENT = "codex_cli_rs/0.0.0 (Unknown 0; unknown) unknown"

# Local-only bridge that runs Claude Code CLI as a subprocess. Hosted by
# brightstaff on this loopback address; the Python CLI auto-fills the matching
# provider fields below and tells the launcher to enable the bridge.
CLAUDE_CLI_DEFAULT_BASE_URL = "http://127.0.0.1:14001"
CLAUDE_CLI_DEFAULT_LISTEN_ADDR = "127.0.0.1:14001"
CLAUDE_CLI_DEFAULT_NAME = "claude-cli/*"
CLAUDE_CLI_DEFAULT_ACCESS_KEY_PLACEHOLDER = "claude-cli-local"

SUPPORTED_PROVIDERS = (
SUPPORTED_PROVIDERS_WITHOUT_BASE_URL + SUPPORTED_PROVIDERS_WITH_BASE_URL
)


def _is_claude_cli_provider(model_provider):
"""Return True iff this provider entry refers to the local claude-cli
bridge. Triggered by any of `model`, `name`, or `provider_interface`
matching the `claude-cli/...` namespace.
"""
model = (model_provider.get("model") or "").strip()
name = (model_provider.get("name") or "").strip()
interface = (model_provider.get("provider_interface") or "").strip()
return (
model.startswith("claude-cli/")
or name.startswith("claude-cli/")
or interface == "claude-cli"
)


def _apply_claude_cli_autofill(model_provider):
"""Fill in implicit fields for `claude-cli/*` provider entries so the
user only has to write `model: claude-cli/*` (or any `claude-cli/...`)
and everything else is wired automatically: a localhost cluster pointing
at the brightstaff bridge, the `claude-cli` provider_interface, and a
placeholder access key so downstream validation does not reject the entry.

Returns True iff this entry was treated as a claude-cli provider (so the
caller can flip the launcher's `needs_claude_cli_runtime` flag).
"""
if not _is_claude_cli_provider(model_provider):
return False

if not model_provider.get("name"):
model_provider["name"] = model_provider.get("model") or CLAUDE_CLI_DEFAULT_NAME
if not model_provider.get("provider_interface"):
model_provider["provider_interface"] = "claude-cli"
if not model_provider.get("base_url"):
model_provider["base_url"] = CLAUDE_CLI_DEFAULT_BASE_URL
# Keep passthrough_auth users alone; the bridge ignores the access key
# anyway (it uses the host's `claude auth login` keychain), so a
# placeholder is fine for everyone else.
if not model_provider.get("access_key") and not model_provider.get(
"passthrough_auth"
):
model_provider["access_key"] = CLAUDE_CLI_DEFAULT_ACCESS_KEY_PLACEHOLDER

return True


def get_endpoint_and_port(endpoint, protocol):
endpoint_tokens = endpoint.split(":")
if len(endpoint_tokens) > 1:
Expand Down Expand Up @@ -329,6 +382,12 @@ def validate_and_render_schema():
name = listener.get("name", None)

for model_provider in listener.get("model_providers", []):
# Auto-fill the implicit fields for `claude-cli/*` providers
# before the rest of the loop runs validation. This makes
# `model_providers: [{model: claude-cli/*}]` a fully-formed
# entry by the time we reach the wildcard checks below.
_apply_claude_cli_autofill(model_provider)

if model_provider.get("usage", None):
llms_with_usage.append(model_provider["name"])
if model_provider.get("name") in model_provider_name_set:
Expand Down
1 change: 1 addition & 0 deletions cli/planoai/consts.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
PLANO_RUN_DIR = os.path.join(PLANO_HOME, "run")
PLANO_BIN_DIR = os.path.join(PLANO_HOME, "bin")
PLANO_PLUGINS_DIR = os.path.join(PLANO_HOME, "plugins")
PLANO_STATE_DIR = os.path.join(PLANO_HOME, "state")
ENVOY_VERSION = "v1.37.0" # keep in sync with Dockerfile ARG ENVOY_VERSION
NATIVE_PID_FILE = os.path.join(PLANO_RUN_DIR, "plano.pid")
DEFAULT_NATIVE_OTEL_TRACING_GRPC_ENDPOINT = "http://localhost:4317"
Expand Down
297 changes: 297 additions & 0 deletions cli/planoai/local_agent_warning.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,297 @@
"""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",
]
Loading
Loading