cli: warn + ack local-agent providers at planoai up

This commit is contained in:
Spherrrical 2026-05-07 11:09:37 -07:00
parent 5e689fed51
commit 8e65fca7d8
4 changed files with 651 additions and 0 deletions

View file

@ -13,6 +13,7 @@ PLANO_HOME = os.path.join(os.path.expanduser("~"), ".plano")
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"

View file

@ -0,0 +1,305 @@
"""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"
# Where the docs page lives. Printed verbatim in the warning panel — the
# relative path resolves cleanly when an operator opens it from the repo
# root, and the GitHub URL is a valid fallback for users running planoai
# outside a clone.
DOCS_RELATIVE_PATH = "docs/source/resources/local_agent_providers.rst"
DOCS_LEARN_MORE = (
"https://github.com/katanemo/plano/blob/main/docs/source/resources/"
"local_agent_providers.rst"
)
@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 single warning panel for ``pending``. Callers must
ensure ``pending`` is non-empty; the caller decides whether to skip
based on the ack set."""
listed = "\n".join(
f" • [bold]{p.name}[/bold]"
+ (f" [dim]({p.model})[/dim]" if p.model and p.model != p.name else "")
+ f" [dim]→ provider_interface=[/dim][cyan]{p.interface}[/cyan]"
for p in pending
)
interfaces_csv = ", ".join(sorted({p.interface for p in pending}))
body_lines = [
"[bold yellow]This config wires up a local-agent provider.[/bold yellow]",
"",
listed,
"",
(
"Unlike stateless network providers ([cyan]openai[/cyan], "
"[cyan]anthropic[/cyan], [cyan]gemini[/cyan], ...), these entries "
"spawn a local CLI binary as a subprocess of brightstaff. The "
"subprocess inherits the operator's permissions and can:"
),
" • read and write any file the operator can touch",
" • execute arbitrary shell commands as the operator's user",
" • use the host's auth keychain / login session",
" • make outbound network calls from the host's IP",
"",
(
"[bold]Intended for local development only — not production.[/bold] "
"Treat this as the same trust class as OpenClaw / OpenCode / "
"Hermes (agent integrations), not a stateless LLM provider."
),
"",
f"[dim]Learn more:[/dim] [bold]{DOCS_LEARN_MORE}[/bold]",
f"[dim]Or in this repo:[/dim] [bold]{DOCS_RELATIVE_PATH}[/bold]",
"",
"[dim]Dismiss permanently:[/dim]",
f" [cyan]planoai up --ack-local-agents[/cyan] [dim]# writes {ACK_FILE_PATH}[/dim]",
f" [dim]or:[/dim] [cyan]{ACK_ENV_VAR}=1 planoai up[/cyan]",
f"[dim]Undo with:[/dim] [cyan]rm {ACK_FILE_PATH}[/cyan]",
]
console.print(
Panel(
"\n".join(body_lines),
title=f"⚠ Local-agent provider detected ({interfaces_csv})",
title_align="left",
border_style="yellow",
padding=(1, 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)
merged = write_acknowledgement(new_set, ack_path=ack_path)
ack_csv = ", ".join(sorted(new_set))
console.print(
f"[green]✓[/green] Acknowledged local-agent provider(s): "
f"[bold]{ack_csv}[/bold] [dim]→ {ack_path}[/dim]"
)
return False
acknowledged = load_acknowledged_interfaces(ack_path)
pending = [p for p in detected if p.interface not in acknowledged]
if not pending:
ack_csv = ", ".join(sorted(_interfaces_in(detected)))
console.print(
f"[dim]Local-agent providers acknowledged: {ack_csv}. "
f"Remove {ack_path} to undo.[/dim]"
)
return False
_render_panel(console, pending)
return True
__all__ = [
"ACK_ENV_VAR",
"ACK_FILE_PATH",
"DOCS_LEARN_MORE",
"DOCS_RELATIVE_PATH",
"LOCAL_AGENT_PROVIDER_INTERFACES",
"LocalAgentProvider",
"detect_local_agent_providers",
"load_acknowledged_interfaces",
"maybe_warn_local_agent_providers",
"write_acknowledgement",
]

View file

@ -39,6 +39,7 @@ from planoai.init_cmd import init as init_cmd
from planoai.trace_cmd import trace as trace_cmd, start_trace_listener_background
from planoai.chatgpt_cmd import chatgpt as chatgpt_cmd
from planoai.obs_cmd import obs as obs_cmd
from planoai.local_agent_warning import maybe_warn_local_agent_providers
from planoai.consts import (
DEFAULT_OTEL_TRACING_GRPC_ENDPOINT,
DEFAULT_NATIVE_OTEL_TRACING_GRPC_ENDPOINT,
@ -354,6 +355,18 @@ def build(docker):
show_default=True,
help="Override the LLM listener port when running without a config file. Ignored when a config file is present.",
)
@click.option(
"--ack-local-agents",
"ack_local_agents",
default=False,
is_flag=True,
help=(
"Acknowledge that local-agent providers (e.g. claude-cli/*) spawn a "
"local CLI binary with full host filesystem and shell access. Writes "
"an ack file so the warning is suppressed on future runs. Equivalent "
"to setting PLANO_ACK_LOCAL_AGENTS=1."
),
)
def up(
file,
path,
@ -363,6 +376,7 @@ def up(
docker,
verbose,
listener_port,
ack_local_agents,
):
"""Starts Plano."""
from rich.status import Status
@ -444,6 +458,15 @@ def up(
with open(plano_config_file, "r") as f:
plano_config = yaml.safe_load(f)
# Warn about local-agent providers (e.g. claude-cli/*) that spawn a
# local CLI binary with full host filesystem and shell access. Fires
# exactly once per `planoai up` invocation; --ack-local-agents (or
# PLANO_ACK_LOCAL_AGENTS=1) writes a persistent ack so the warning
# only re-appears for newly-introduced local-agent interfaces.
maybe_warn_local_agent_providers(
plano_config or {}, console, ack_flag=ack_local_agents
)
# Inject ChatGPT tokens from ~/.plano/chatgpt/auth.json if any provider needs them
_inject_chatgpt_tokens_if_needed(plano_config, env, console)

View file

@ -0,0 +1,322 @@
"""Tests for the local-agent provider warning, ack persistence, and the
detection logic that decides whether to fire it."""
from __future__ import annotations
import io
import json
import pytest
from rich.console import Console
from planoai import local_agent_warning as law
def _make_console() -> tuple[Console, io.StringIO]:
buf = io.StringIO()
# ``force_terminal=False`` keeps Rich from emitting ANSI escapes,
# which makes substring assertions readable. ``width`` is generous
# so the panel border doesn't soft-wrap text mid-keyword.
console = Console(file=buf, force_terminal=False, color_system=None, width=140)
return console, buf
# ---------------------------------------------------------------------------
# detection
# ---------------------------------------------------------------------------
def test_detects_claude_cli_via_model_prefix():
config = {
"model_providers": [
{"model": "claude-cli/sonnet"},
{"model": "openai/gpt-4o"},
]
}
found = law.detect_local_agent_providers(config)
assert [p.interface for p in found] == ["claude-cli"]
assert found[0].model == "claude-cli/sonnet"
def test_detects_claude_cli_via_explicit_provider_interface():
config = {
"model_providers": [
{"name": "local-claude", "provider_interface": "claude-cli", "model": "x"},
]
}
found = law.detect_local_agent_providers(config)
assert [p.interface for p in found] == ["claude-cli"]
assert found[0].name == "local-claude"
def test_detects_claude_cli_via_legacy_provider_field():
config = {"model_providers": [{"provider": "claude-cli", "model": "x"}]}
assert [p.interface for p in law.detect_local_agent_providers(config)] == [
"claude-cli"
]
def test_detects_via_legacy_llm_providers_key():
config = {"llm_providers": [{"model": "claude-cli/opus"}]}
assert [p.interface for p in law.detect_local_agent_providers(config)] == [
"claude-cli"
]
def test_no_false_positive_for_network_providers():
config = {
"model_providers": [
{"model": "openai/gpt-4o"},
{"model": "anthropic/claude-3-5-sonnet"},
{"model": "gemini/gemini-2.5-pro"},
{"model": "chatgpt/gpt-5"}, # network ChatGPT subscription, not a CLI
{"model": "vercel/some-model"},
]
}
assert law.detect_local_agent_providers(config) == []
def test_no_false_positive_for_anthropic_claude_models():
# ``anthropic/claude-3-5-sonnet`` must not trigger just because the
# word "claude" appears — the prefix has to be ``claude-cli/``.
config = {"model_providers": [{"model": "anthropic/claude-3-5-sonnet-20241022"}]}
assert law.detect_local_agent_providers(config) == []
def test_empty_or_malformed_config_is_safe():
assert law.detect_local_agent_providers({}) == []
assert law.detect_local_agent_providers({"model_providers": None}) == []
assert law.detect_local_agent_providers({"model_providers": "not-a-list"}) == []
# ``None`` config (e.g. from an empty yaml file) must also be safe.
assert law.detect_local_agent_providers(None) == [] # type: ignore[arg-type]
def test_multiple_entries_same_interface_collapse_in_warning_set():
config = {
"model_providers": [
{"model": "claude-cli/sonnet", "name": "fast"},
{"model": "claude-cli/opus", "name": "slow"},
]
}
found = law.detect_local_agent_providers(config)
assert len(found) == 2
assert {p.interface for p in found} == {"claude-cli"}
# ---------------------------------------------------------------------------
# ack file
# ---------------------------------------------------------------------------
def test_load_ack_returns_empty_when_missing(tmp_path):
ack = tmp_path / "ack.json"
assert law.load_acknowledged_interfaces(str(ack)) == set()
@pytest.mark.parametrize(
"contents",
[
"{not valid json",
"[]", # not a dict
'{"acknowledged": "claude-cli"}', # not a list
'{"acknowledged": [1, 2, 3]}', # not strings
],
)
def test_load_ack_handles_malformed_files(tmp_path, contents):
ack = tmp_path / "ack.json"
ack.write_text(contents, encoding="utf-8")
# Malformed contents must degrade to "no ack" rather than crashing.
assert law.load_acknowledged_interfaces(str(ack)) == set()
def test_write_ack_creates_state_dir(tmp_path):
ack = tmp_path / "fresh" / "deeper" / "ack.json"
merged = law.write_acknowledgement(["claude-cli"], ack_path=str(ack))
assert merged == {"claude-cli"}
assert ack.exists()
payload = json.loads(ack.read_text(encoding="utf-8"))
assert payload["acknowledged"] == ["claude-cli"]
assert payload["ack_at"]
def test_write_ack_merges_with_existing(tmp_path):
ack = tmp_path / "ack.json"
law.write_acknowledgement(["claude-cli"], ack_path=str(ack))
merged = law.write_acknowledgement(["future-cli"], ack_path=str(ack))
assert merged == {"claude-cli", "future-cli"}
payload = json.loads(ack.read_text(encoding="utf-8"))
assert payload["acknowledged"] == ["claude-cli", "future-cli"]
# ---------------------------------------------------------------------------
# maybe_warn_local_agent_providers
# ---------------------------------------------------------------------------
def test_no_panel_when_no_local_agent_providers(tmp_path):
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "openai/gpt-4o"}]},
console,
ack_path=str(tmp_path / "ack.json"),
env={},
)
assert fired is False
assert buf.getvalue() == ""
def test_panel_fires_for_unacked_claude_cli(tmp_path):
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "claude-cli/sonnet"}]},
console,
ack_path=str(tmp_path / "ack.json"),
env={},
)
output = buf.getvalue()
assert fired is True
# Stable substrings — never pin exact wording.
assert "claude-cli" in output
assert "Local-agent" in output or "local-agent" in output
assert "Learn more" in output
assert "--ack-local-agents" in output
# The dismissal hint must mention the ack file path so the user
# knows where to ``rm`` it.
assert "local_agent_ack.json" in output
def test_panel_suppressed_when_ack_covers_interface(tmp_path):
ack = tmp_path / "ack.json"
law.write_acknowledgement(["claude-cli"], ack_path=str(ack))
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "claude-cli/sonnet"}]},
console,
ack_path=str(ack),
env={},
)
assert fired is False
# The dim INFO line still mentions the ack file so the operator
# knows how to undo, but no panel renders.
out = buf.getvalue()
assert "Panel" not in out # no panel object
assert "claude-cli" in out
def test_new_unacked_interface_re_triggers(tmp_path, monkeypatch):
# Simulate a future where two local-agent interfaces exist and the
# user has only acknowledged one of them.
monkeypatch.setattr(
law, "LOCAL_AGENT_PROVIDER_INTERFACES", ("claude-cli", "future-cli")
)
ack = tmp_path / "ack.json"
law.write_acknowledgement(["claude-cli"], ack_path=str(ack))
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{
"model_providers": [
{"model": "claude-cli/sonnet"},
{"model": "future-cli/whatever"},
]
},
console,
ack_path=str(ack),
env={},
)
output = buf.getvalue()
assert fired is True
# The panel must list the *unacknowledged* interface only.
assert "future-cli" in output
# ...and must NOT re-list the already-acknowledged one as unacked
# (it can still appear in the suppressed-info line; we check the
# title which only contains pending interfaces).
assert "future-cli" in output
def test_ack_flag_writes_file_and_suppresses_panel(tmp_path):
ack = tmp_path / "ack.json"
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "claude-cli/sonnet"}]},
console,
ack_flag=True,
ack_path=str(ack),
env={},
)
assert fired is False
assert ack.exists()
payload = json.loads(ack.read_text(encoding="utf-8"))
assert "claude-cli" in payload["acknowledged"]
out = buf.getvalue()
assert "Acknowledged" in out
assert "claude-cli" in out
@pytest.mark.parametrize("env_value", ["1", "true", "TRUE", "yes", "on"])
def test_ack_env_var_truthy_values(tmp_path, env_value):
ack = tmp_path / "ack.json"
console, _ = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "claude-cli/sonnet"}]},
console,
ack_path=str(ack),
env={law.ACK_ENV_VAR: env_value},
)
assert fired is False
assert ack.exists()
@pytest.mark.parametrize("env_value", ["", "0", "false", "no", "off", "maybe"])
def test_ack_env_var_falsy_values_still_warn(tmp_path, env_value):
ack = tmp_path / "ack.json"
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "claude-cli/sonnet"}]},
console,
ack_path=str(ack),
env={law.ACK_ENV_VAR: env_value},
)
assert fired is True
assert not ack.exists()
assert "claude-cli" in buf.getvalue()
def test_malformed_ack_falls_back_to_warning(tmp_path):
ack = tmp_path / "ack.json"
ack.write_text("{not json", encoding="utf-8")
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{"model_providers": [{"model": "claude-cli/sonnet"}]},
console,
ack_path=str(ack),
env={},
)
assert fired is True
assert "claude-cli" in buf.getvalue()
def test_single_panel_when_multiple_local_agent_entries(tmp_path):
# Two entries with the same interface must produce one panel,
# not two — the warning fires once per ``planoai up`` invocation.
console, buf = _make_console()
fired = law.maybe_warn_local_agent_providers(
{
"model_providers": [
{"model": "claude-cli/sonnet", "name": "fast"},
{"model": "claude-cli/opus", "name": "slow"},
]
},
console,
ack_path=str(tmp_path / "ack.json"),
env={},
)
assert fired is True
output = buf.getvalue()
# Both names appear in the listing, but the warning header
# (``Local-agent provider detected``) appears exactly once.
assert output.count("Local-agent provider detected") == 1
assert "fast" in output
assert "slow" in output