apunkt/plano
1
0
Fork 0
mirror of https://github.com/katanemo/plano.git synced 2026-07-23 16:51:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

Introduce brand new CLI experience with tracing and quickstart (#724)

Browse source 
Release hardens tracing and routing: clearer CLI, modular internals, updated demos/docs/tests, and improved multi-agent reliability.

Co-authored-by: Adil Hafeez <adil.hafeez@gmail.com>
This commit is contained in:
Musa 2026-02-10 13:17:43 -08:00 • committed by GitHub
parent 5394ef5770
commit e3bf2b7f71
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 2429 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

3
cli/planoai/consts.py
View file

@ -1,5 +1,8 @@
import os
# Brand color - Plano purple
PLANO_COLOR = "#969FF4"
SERVICE_NAME_ARCHGW = "plano"
PLANO_DOCKER_NAME = "plano"
PLANO_DOCKER_IMAGE = os.getenv("PLANO_DOCKER_IMAGE", "katanemo/plano:0.4.4")

2
cli/planoai/core.py
View file

@ -1,8 +1,8 @@
import json
import subprocess
import os
import time
import sys
import time
import yaml
from planoai.utils import convert_legacy_listeners, getLogger

4
cli/planoai/docker_cli.py
View file

@ -35,7 +35,7 @@ def docker_stop_container(container: str) -> str:
def docker_remove_container(container: str) -> str:
result = subprocess.run(
["docker", "rm", container], capture_output=True, text=True, check=False
["docker", "rm", "-f", container], capture_output=True, text=True, check=False
)
return result.returncode
@ -48,7 +48,7 @@ def docker_start_plano_detached(
env_args = [item for key, value in env.items() for item in ["-e", f"{key}={value}"]]
port_mappings = [
f"{12001}:{12001}",
"12001:12001",
"19901:9901",
]

303
cli/planoai/init_cmd.py Normal file
View file

@ -0,0 +1,303 @@
import os
from importlib import resources
from dataclasses import dataclass
from pathlib import Path
import rich_click as click
from rich.console import Console
from rich.panel import Panel
from planoai.consts import PLANO_COLOR
@dataclass(frozen=True)
class Template:
"""
A Plano config template.
- id: stable identifier used by --template
- title/description: UI strings
- yaml_text: embedded template contents (works in PyPI installs)
"""
id: str
title: str
description: str
yaml_text: str
_TEMPLATE_PACKAGE = "planoai.templates"
def _load_template_yaml(filename: str) -> str:
return resources.files(_TEMPLATE_PACKAGE).joinpath(filename).read_text("utf-8")
BUILTIN_TEMPLATES: list[Template] = [
Template(
id="sub_agent_orchestration",
title="Sub Agent Orchestration",
description="multi-agent routing across specialized agents",
yaml_text=_load_template_yaml("sub_agent_orchestration.yaml"),
),
Template(
id="coding_agent_routing",
title="Coding Agent Routing",
description="routing preferences + model aliases for coding tasks",
yaml_text=_load_template_yaml("coding_agent_routing.yaml"),
),
Template(
id="preference_aware_routing",
title="Preference-aware LLM routing",
description="automatic LLM routing based on preferences",
yaml_text=_load_template_yaml("preference_aware_routing.yaml"),
),
Template(
id="filter_chain_guardrails",
title="Guardrails via Filter Chains",
description="input guards, query rewrite, and context building",
yaml_text=_load_template_yaml("filter_chain_guardrails.yaml"),
),
Template(
id="conversational_state_v1_responses",
title="Conversational State via v1/responses",
description="stateful responses with memory-backed storage",
yaml_text=_load_template_yaml("conversational_state_v1_responses.yaml"),
),
]
def _get_templates() -> list[Template]:
return list(BUILTIN_TEMPLATES)
def _resolve_template(template_id: str | None) -> Template | None:
if not template_id:
return None
templates = _get_templates()
for t in templates:
if t.id == template_id:
return t
return None
def _ensure_parent_dir(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
def _write_clean_config(path: Path, force: bool) -> None:
_ensure_parent_dir(path)
if path.exists() and not force:
raise FileExistsError(str(path))
# user asked for NOTHING in it: empty file, with just a newline for POSIX friendliness
path.write_text("\n", encoding="utf-8")
def _write_template_config(path: Path, template: Template, force: bool) -> str:
_ensure_parent_dir(path)
if path.exists() and not force:
raise FileExistsError(str(path))
path.write_text(template.yaml_text, encoding="utf-8")
return "builtin"
def _print_config_preview(console: Console, text: str, max_lines: int = 28) -> None:
lines = text.strip("\n").splitlines()
preview_lines = lines[:max_lines]
if len(lines) > max_lines:
preview_lines.append("... (truncated)")
preview = "\n".join(preview_lines).strip("\n")
if not preview:
preview = "(empty)"
console.print(
Panel(
preview,
title="Config preview",
border_style="dim",
title_align="left",
)
)
def _questionary_style():
# prompt_toolkit style string format
from prompt_toolkit.styles import Style
return Style.from_dict(
{
"qmark": f"fg:{PLANO_COLOR} bold",
"question": "bold",
"answer": f"fg:{PLANO_COLOR} bold",
"pointer": f"fg:{PLANO_COLOR} bold",
"highlighted": f"fg:{PLANO_COLOR} bold",
"selected": f"fg:{PLANO_COLOR}",
"instruction": "fg:#888888",
"text": "",
"disabled": "fg:#666666",
}
)
def _force_truecolor_for_prompt_toolkit() -> None:
"""
Ensure prompt_toolkit uses truecolor so our brand hex (#969FF4) renders correctly.
Without this, some terminals or environments downgrade to 8-bit and the color
can look like a generic blue.
"""
# Only set if user hasn't explicitly chosen a depth.
os.environ.setdefault("PROMPT_TOOLKIT_COLOR_DEPTH", "DEPTH_24_BIT")
@click.command()
@click.option(
"--template",
"template_id_or_path",
default=None,
help="Create config.yaml from a built-in template id.",
)
@click.option(
"--clean",
is_flag=True,
help="Create an empty config.yaml with no contents.",
)
@click.option(
"--output",
"-o",
"output_path",
default="config.yaml",
show_default=True,
help="Where to write the generated config.",
)
@click.option(
"--force",
is_flag=True,
help="Overwrite existing config file if it already exists.",
)
@click.option(
"--list-templates",
is_flag=True,
help="List available template ids and exit.",
)
@click.pass_context
def init(ctx, template_id_or_path, clean, output_path, force, list_templates):
"""Initialize a Plano config quickly (arrow-key interactive wizard by default)."""
import sys
console = Console()
if clean and template_id_or_path:
raise click.UsageError("Use either --clean or --template, not both.")
templates = _get_templates()
if list_templates:
console.print(f"[bold {PLANO_COLOR}]Available templates[/bold {PLANO_COLOR}]\n")
for t in templates:
console.print(f" [bold]{t.id}[/bold] - {t.description}")
return
out_path = Path(output_path).expanduser()
# Non-interactive fast paths
if clean or template_id_or_path:
if clean:
try:
_write_clean_config(out_path, force=force)
except FileExistsError:
raise click.ClickException(
f"Refusing to overwrite existing file: {out_path} (use --force)"
)
console.print(f"[green]✓[/green] Wrote [bold]{out_path}[/bold]")
_print_config_preview(console, out_path.read_text(encoding="utf-8"))
return
template = _resolve_template(template_id_or_path)
if not template:
raise click.ClickException(
f"Unknown template: {template_id_or_path}\n"
f"Run: planoai init --list-templates"
)
try:
_write_template_config(out_path, template, force=force)
except FileExistsError:
raise click.ClickException(
f"Refusing to overwrite existing file: {out_path} (use --force)"
)
console.print(
f"[green]✓[/green] Wrote [bold]{out_path}[/bold] [dim]({template.id})[/dim]"
)
_print_config_preview(console, template.yaml_text)
return
# Interactive wizard
if not (sys.stdin.isatty() and sys.stdout.isatty()):
raise click.ClickException(
"Interactive mode requires a TTY.\n"
"Use one of:\n"
" planoai init --template <id>\n"
" planoai init --clean\n"
" planoai init --list-templates"
)
_force_truecolor_for_prompt_toolkit()
# Lazy import so non-interactive users don't pay the import/compat cost
import questionary
from questionary import Choice
# Step 1: choose template (or clean)
template_choices: list[Choice] = [
Choice("Create a clean config.yaml (empty)", value="clean"),
]
for t in templates:
label = f"{t.title} — {t.description}"
template_choices.append(Choice(label, value=t))
selected = questionary.select(
"Choose a template",
choices=template_choices,
style=_questionary_style(),
pointer="❯",
use_indicator=True,
).ask()
if not selected:
console.print("[dim]Cancelled.[/dim]")
return
# Step 2: output path (default: config.yaml)
out_answer = questionary.text(
"Where should I write the config?",
default=str(out_path),
style=_questionary_style(),
).ask()
if not out_answer:
console.print("[dim]Cancelled.[/dim]")
return
out_path = Path(out_answer).expanduser()
if out_path.exists() and not force:
overwrite = questionary.confirm(
f"{out_path} already exists. Overwrite?",
default=False,
style=_questionary_style(),
).ask()
if not overwrite:
console.print("[dim]Cancelled.[/dim]")
return
force = True
if selected == "clean":
_write_clean_config(out_path, force=True)
console.print(f"[green]✓[/green] Wrote [bold]{out_path}[/bold]")
_print_config_preview(console, out_path.read_text(encoding="utf-8"))
return
template = selected
_write_template_config(out_path, template, force=True)
console.print(
f"[green]✓[/green] Wrote [bold]{out_path}[/bold] [dim]({template.id})[/dim]"
)
_print_config_preview(console, template.yaml_text)

266
cli/planoai/main.py
View file

@ -1,11 +1,12 @@
import click
import os
import sys
import subprocess
import multiprocessing
import importlib.metadata
import json
import subprocess
import sys
import rich_click as click
from planoai import targets
# Brand color - Plano purple
PLANO_COLOR = "#969FF4"
from planoai.docker_cli import (
docker_validate_plano_schema,
stream_gateway_logs,
@ -14,7 +15,6 @@ from planoai.docker_cli import (
from planoai.utils import (
getLogger,
get_llm_provider_access_keys,
has_ingress_listener,
load_env_file_to_dict,
set_log_level,
stream_access_logs,
@ -26,60 +26,106 @@ from planoai.core import (
stop_docker_container,
start_cli_agent,
)
from planoai.init_cmd import init as init_cmd
from planoai.trace_cmd import trace as trace_cmd, start_trace_listener_background
from planoai.consts import (
DEFAULT_OTEL_TRACING_GRPC_ENDPOINT,
PLANO_DOCKER_IMAGE,
PLANO_DOCKER_NAME,
SERVICE_NAME_ARCHGW,
)
from planoai.rich_click_config import configure_rich_click
from planoai.versioning import check_version_status, get_latest_version, get_version
log = getLogger(__name__)
# ref https://patorjk.com/software/taag/#p=display&f=Doom&t=Plano&x=none&v=4&h=4&w=80&we=false
logo = r"""
______ _
| ___ \ |
| |_/ / | __ _ _ __ ___
| __/| |/ _` | '_ \ / _ \
| | | | (_| | | | | (_) |
\_| |_|\__,_|_| |_|\___/
"""
def _is_port_in_use(port: int) -> bool:
"""Check if a TCP port is already bound on localhost."""
import socket
# Command to build plano Docker images
ARCHGW_DOCKERFILE = "./Dockerfile"
def get_version():
try:
# First try to get version from package metadata (for installed packages)
version = importlib.metadata.version("planoai")
return version
except importlib.metadata.PackageNotFoundError:
# Fallback to version defined in __init__.py (for development)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try:
from planoai import __version__
s.bind(("0.0.0.0", port))
return False
except OSError:
return True
return __version__
except ImportError:
return "version not found"
# ref https://patorjk.com/software/taag/#p=display&f=Doom&t=Plano&x=none&v=4&h=4&w=80&we=false
LOGO = f"""[bold {PLANO_COLOR}]
______ _
| ___ \\ |
| |_/ / | __ _ _ __ ___
| __/| |/ _` | '_ \\ / _ \\
| | | | (_| | | | | (_) |
\\_| |_|\\__,_|_| |_|\\___/
[/bold {PLANO_COLOR}]"""
def _console():
from rich.console import Console
return Console()
def _print_cli_header(console) -> None:
console.print(
f"\n[bold {PLANO_COLOR}]Plano CLI[/bold {PLANO_COLOR}] [dim]v{get_version()}[/dim]\n"
)
def _print_missing_keys(console, missing_keys: list[str]) -> None:
console.print(f"\n[red]✗[/red] [red]Missing API keys![/red]\n")
for key in missing_keys:
console.print(f" [red]•[/red] [bold]{key}[/bold] not found")
console.print(f"\n[dim]Set the environment variable(s):[/dim]")
for key in missing_keys:
console.print(f' [cyan]export {key}="your-api-key"[/cyan]')
console.print(f"\n[dim]Or create a .env file in the config directory.[/dim]\n")
def _print_version(console, current_version: str) -> None:
console.print(
f"[bold {PLANO_COLOR}]plano[/bold {PLANO_COLOR}] version [cyan]{current_version}[/cyan]"
)
def _maybe_check_updates(console, current_version: str) -> None:
if os.environ.get("PLANO_SKIP_VERSION_CHECK"):
return
latest_version = get_latest_version()
status = check_version_status(current_version, latest_version)
if status["is_outdated"]:
console.print(
f"\n[yellow]⚠ Update available:[/yellow] [bold]{status['latest']}[/bold]"
)
console.print("[dim]Run: uv pip install --upgrade planoai[/dim]")
elif latest_version:
console.print(f"[dim]✓ You're up to date[/dim]")
configure_rich_click(PLANO_COLOR)
@click.group(invoke_without_command=True)
@click.option("--version", is_flag=True, help="Show the plano cli version and exit.")
@click.option("--version", is_flag=True, help="Show the Plano CLI version and exit.")
@click.pass_context
def main(ctx, version):
# Set log level from LOG_LEVEL env var only
set_log_level(os.environ.get("LOG_LEVEL", "info"))
console = _console()
if version:
click.echo(f"plano cli version: {get_version()}")
current_version = get_version()
_print_version(console, current_version)
_maybe_check_updates(console, current_version)
ctx.exit()
log.info(f"Starting plano cli version: {get_version()}")
if ctx.invoked_subcommand is None:
click.echo("""Plano (AI-native proxy and dataplane for agentic apps) CLI""")
click.echo(logo)
console.print(LOGO)
console.print("[dim]The Delivery Infrastructure for Agentic Apps[/dim]\n")
click.echo(ctx.get_help())
@ -133,81 +179,156 @@ def build():
help="Run Plano in the foreground. Default is False",
is_flag=True,
)
def up(file, path, foreground):
@click.option(
"--with-tracing",
default=False,
help="Start a local OTLP trace collector on port 4317.",
is_flag=True,
)
@click.option(
"--tracing-port",
default=4317,
type=int,
help="Port for the OTLP trace collector (default: 4317).",
show_default=True,
)
def up(file, path, foreground, with_tracing, tracing_port):
"""Starts Plano."""
from rich.status import Status
console = _console()
_print_cli_header(console)
# Use the utility function to find config file
arch_config_file = find_config_file(path, file)
# Check if the file exists
if not os.path.exists(arch_config_file):
log.info(f"Error: {arch_config_file} does not exist.")
return
log.info(f"Validating {arch_config_file}")
(
validation_return_code,
validation_stdout,
validation_stderr,
) = docker_validate_plano_schema(arch_config_file)
if validation_return_code != 0:
log.info(f"Error: Validation failed. Exiting")
log.info(f"Validation stdout: {validation_stdout}")
log.info(f"Validation stderr: {validation_stderr}")
console.print(
f"[red]✗[/red] Config file not found: [dim]{arch_config_file}[/dim]"
)
sys.exit(1)
# Set the ARCH_CONFIG_FILE environment variable
with Status(
"[dim]Validating configuration[/dim]", spinner="dots", spinner_style="dim"
):
(
validation_return_code,
_,
validation_stderr,
) = docker_validate_plano_schema(arch_config_file)
if validation_return_code != 0:
console.print(f"[red]✗[/red] Validation failed")
if validation_stderr:
console.print(f" [dim]{validation_stderr.strip()}[/dim]")
sys.exit(1)
console.print(f"[green]✓[/green] Configuration valid")
# Set up environment
env_stage = {
"OTEL_TRACING_GRPC_ENDPOINT": DEFAULT_OTEL_TRACING_GRPC_ENDPOINT,
}
env = os.environ.copy()
# Remove PATH variable if present
env.pop("PATH", None)
# check if access_keys are preesnt in the config file
access_keys = get_llm_provider_access_keys(arch_config_file=arch_config_file)
# remove duplicates
# Check access keys
access_keys = get_llm_provider_access_keys(arch_config_file=arch_config_file)
access_keys = set(access_keys)
# remove the $ from the access_keys
access_keys = [item[1:] if item.startswith("$") else item for item in access_keys]
missing_keys = []
if access_keys:
if file:
app_env_file = os.path.join(
os.path.dirname(os.path.abspath(file)), ".env"
) # check the .env file in the path
app_env_file = os.path.join(os.path.dirname(os.path.abspath(file)), ".env")
else:
app_env_file = os.path.abspath(os.path.join(path, ".env"))
if not os.path.exists(
app_env_file
): # check to see if the environment variables in the current environment or not
if not os.path.exists(app_env_file):
for access_key in access_keys:
if env.get(access_key) is None:
log.info(f"Access Key: {access_key} not found. Exiting Start")
sys.exit(1)
missing_keys.append(access_key)
else:
env_stage[access_key] = env.get(access_key)
else: # .env file exists, use that to send parameters to Arch
else:
env_file_dict = load_env_file_to_dict(app_env_file)
for access_key in access_keys:
if env_file_dict.get(access_key) is None:
log.info(f"Access Key: {access_key} not found. Exiting Start")
sys.exit(1)
missing_keys.append(access_key)
else:
env_stage[access_key] = env_file_dict[access_key]
if missing_keys:
_print_missing_keys(console, missing_keys)
sys.exit(1)
# Pass log level to the Docker container — supervisord uses LOG_LEVEL
# to set RUST_LOG (brightstaff) and envoy component log levels
env_stage["LOG_LEVEL"] = os.environ.get("LOG_LEVEL", "info")
# Start the local OTLP trace collector if --with-tracing is set
trace_server = None
if with_tracing:
if _is_port_in_use(tracing_port):
# A listener is already running (e.g. `planoai trace listen`)
console.print(
f"[green]✓[/green] Trace collector already running on port [cyan]{tracing_port}[/cyan]"
)
else:
try:
trace_server = start_trace_listener_background(grpc_port=tracing_port)
console.print(
f"[green]✓[/green] Trace collector listening on [cyan]0.0.0.0:{tracing_port}[/cyan]"
)
except Exception as e:
console.print(
f"[red]✗[/red] Failed to start trace collector on port {tracing_port}: {e}"
)
console.print(
f"\n[dim]Check if another process is using port {tracing_port}:[/dim]"
)
console.print(f" [cyan]lsof -i :{tracing_port}[/cyan]")
console.print(f"\n[dim]Or use a different port:[/dim]")
console.print(
f" [cyan]planoai up --with-tracing --tracing-port 4318[/cyan]\n"
)
sys.exit(1)
# Update the OTEL endpoint so the gateway sends traces to the right port
env_stage[
"OTEL_TRACING_GRPC_ENDPOINT"
] = f"http://host.docker.internal:{tracing_port}"
env.update(env_stage)
start_arch(arch_config_file, env, foreground=foreground)
try:
start_arch(arch_config_file, env, foreground=foreground)
# When tracing is enabled but --foreground is not, keep the process
# alive so the OTLP collector continues to receive spans.
if trace_server is not None and not foreground:
console.print(
f"[dim]Plano is running. Trace collector active on port {tracing_port}. Press Ctrl+C to stop.[/dim]"
)
trace_server.wait_for_termination()
except KeyboardInterrupt:
if trace_server is not None:
console.print(f"\n[dim]Stopping trace collector...[/dim]")
finally:
if trace_server is not None:
trace_server.stop(grace=2)
@click.command()
def down():
"""Stops Arch."""
stop_docker_container()
"""Stops Plano."""
console = _console()
_print_cli_header(console)
with console.status(
f"[{PLANO_COLOR}]Shutting down Plano...[/{PLANO_COLOR}]", spinner="dots"
):
stop_docker_container()
@click.command()
@ -306,12 +427,15 @@ def cli_agent(type, file, path, settings):
sys.exit(1)
# add commands to the main group
main.add_command(up)
main.add_command(down)
main.add_command(build)
main.add_command(logs)
main.add_command(cli_agent)
main.add_command(generate_prompt_targets)
main.add_command(init_cmd, name="init")
main.add_command(trace_cmd, name="trace")
if __name__ == "__main__":
main()

71
cli/planoai/rich_click_config.py Normal file
View file

@ -0,0 +1,71 @@
import rich_click as click
def configure_rich_click(plano_color: str) -> None:
click.rich_click.USE_RICH_MARKUP = True
click.rich_click.USE_MARKDOWN = False
click.rich_click.SHOW_ARGUMENTS = True
click.rich_click.GROUP_ARGUMENTS_OPTIONS = True
click.rich_click.STYLE_ERRORS_SUGGESTION = "dim italic"
click.rich_click.ERRORS_SUGGESTION = (
"Try running the '--help' flag for more information."
)
click.rich_click.ERRORS_EPILOGUE = ""
# Custom colors matching Plano brand.
click.rich_click.STYLE_OPTION = f"dim {plano_color}"
click.rich_click.STYLE_ARGUMENT = f"dim {plano_color}"
click.rich_click.STYLE_COMMAND = f"bold {plano_color}"
click.rich_click.STYLE_SWITCH = "bold green"
click.rich_click.STYLE_METAVAR = "bold yellow"
click.rich_click.STYLE_USAGE = "bold"
click.rich_click.STYLE_USAGE_COMMAND = f"bold dim {plano_color}"
click.rich_click.STYLE_HELPTEXT_FIRST_LINE = "white italic"
click.rich_click.STYLE_HELPTEXT = ""
click.rich_click.STYLE_HEADER_TEXT = "bold"
click.rich_click.STYLE_FOOTER_TEXT = "dim"
click.rich_click.STYLE_OPTIONS_PANEL_BORDER = "dim"
click.rich_click.ALIGN_OPTIONS_PANEL = "left"
click.rich_click.MAX_WIDTH = 100
# Option groups for better organization.
click.rich_click.OPTION_GROUPS = {
"planoai up": [
{
"name": "Configuration",
"options": ["--path", "file"],
},
{
"name": "Runtime Options",
"options": ["--foreground", "--with-tracing", "--tracing-port"],
},
],
"planoai logs": [
{
"name": "Log Options",
"options": ["--debug", "--follow"],
},
],
}
# Command groups for main help.
click.rich_click.COMMAND_GROUPS = {
"planoai": [
{
"name": "Gateway Commands",
"commands": ["up", "down", "build", "logs"],
},
{
"name": "Agent Commands",
"commands": ["cli-agent"],
},
{
"name": "Observability",
"commands": ["trace"],
},
{
"name": "Utilities",
"commands": ["generate-prompt-targets"],
},
],
}

43
cli/planoai/templates/coding_agent_routing.yaml Normal file
View file

@ -0,0 +1,43 @@
version: v0.1
listeners:
egress_traffic:
address: 0.0.0.0
port: 12000
message_format: openai
timeout: 30s
llm_providers:
# OpenAI Models
- model: openai/gpt-5-2025-08-07
access_key: $OPENAI_API_KEY
routing_preferences:
- name: code generation
description: generating new code snippets, functions, or boilerplate based on user prompts or requirements
- model: openai/gpt-4.1-2025-04-14
access_key: $OPENAI_API_KEY
routing_preferences:
- name: code understanding
description: understand and explain existing code snippets, functions, or libraries
# Anthropic Models
- model: anthropic/claude-sonnet-4-5
default: true
access_key: $ANTHROPIC_API_KEY
- model: anthropic/claude-haiku-4-5
access_key: $ANTHROPIC_API_KEY
# Ollama Models
- model: ollama/llama3.1
base_url: http://host.docker.internal:11434
# Model aliases - friendly names that map to actual provider names
model_aliases:
# Alias for a small faster Claude model
arch.claude.code.small.fast:
target: claude-haiku-4-5
tracing:
random_sampling: 100

25
cli/planoai/templates/conversational_state_v1_responses.yaml Normal file
View file

@ -0,0 +1,25 @@
version: v0.1
listeners:
egress_traffic:
address: 0.0.0.0
port: 12000
message_format: openai
timeout: 30s
llm_providers:
# OpenAI Models
- model: openai/gpt-5-mini-2025-08-07
access_key: $OPENAI_API_KEY
default: true
# Anthropic Models
- model: anthropic/claude-sonnet-4-20250514
access_key: $ANTHROPIC_API_KEY
# State storage configuration for v1/responses API
# Manages conversation state for multi-turn conversations
state_storage:
# Type: memory | postgres
type: memory

50
cli/planoai/templates/filter_chain_guardrails.yaml Normal file
View file

@ -0,0 +1,50 @@
version: v0.3.0
agents:
- id: rag_agent
url: http://rag-agents:10505
filters:
- id: input_guards
url: http://rag-agents:10500
type: http
# type: mcp (default)
# transport: streamable-http (default)
# tool: input_guards (default - same as filter id)
- id: query_rewriter
url: http://rag-agents:10501
type: http
# type: mcp (default)
# transport: streamable-http (default)
# tool: query_rewriter (default - same as filter id)
- id: context_builder
url: http://rag-agents:10502
type: http
model_providers:
- model: openai/gpt-4o-mini
access_key: $OPENAI_API_KEY
default: true
- model: openai/gpt-4o
access_key: $OPENAI_API_KEY
model_aliases:
fast-llm:
target: gpt-4o-mini
smart-llm:
target: gpt-4o
listeners:
- type: agent
name: agent_1
port: 8001
router: plano_orchestrator_v1
agents:
- id: rag_agent
description: virtual assistant for retrieval augmented generation tasks
filter_chain:
- input_guards
- query_rewriter
- context_builder
tracing:
random_sampling: 100

29
cli/planoai/templates/preference_aware_routing.yaml Normal file
View file

@ -0,0 +1,29 @@
version: v0.1.0
listeners:
egress_traffic:
address: 0.0.0.0
port: 12000
message_format: openai
timeout: 30s
llm_providers:
- model: openai/gpt-4o-mini
access_key: $OPENAI_API_KEY
default: true
- model: openai/gpt-4o
access_key: $OPENAI_API_KEY
routing_preferences:
- name: code understanding
description: understand and explain existing code snippets, functions, or libraries
- model: anthropic/claude-sonnet-4-20250514
access_key: $ANTHROPIC_API_KEY
routing_preferences:
- name: code generation
description: generating new code snippets, functions, or boilerplate based on user prompts or requirements
tracing:
random_sampling: 100

57
cli/planoai/templates/sub_agent_orchestration.yaml Normal file
View file

@ -0,0 +1,57 @@
version: v0.3.0
agents:
- id: weather_agent
url: http://langchain-weather-agent:10510
- id: flight_agent
url: http://crewai-flight-agent:10520
model_providers:
- model: openai/gpt-4o
access_key: $OPENAI_API_KEY
default: true
- model: openai/gpt-4o-mini
access_key: $OPENAI_API_KEY # smaller, faster, cheaper model for extracting entities like location
listeners:
- type: agent
name: travel_booking_service
port: 8001
router: plano_orchestrator_v1
agents:
- id: weather_agent
description: |
WeatherAgent is a specialized AI assistant for real-time weather information and forecasts. It provides accurate weather data for any city worldwide using the Open-Meteo API, helping travelers plan their trips with up-to-date weather conditions.
Capabilities:
* Get real-time weather conditions and multi-day forecasts for any city worldwide using Open-Meteo API (free, no API key needed)
* Provides current temperature
* Provides multi-day forecasts
* Provides weather conditions
* Provides sunrise/sunset times
* Provides detailed weather information
* Understands conversation context to resolve location references from previous messages
* Handles weather-related questions including "What's the weather in [city]?", "What's the forecast for [city]?", "How's the weather in [city]?"
* When queries include both weather and other travel questions (e.g., flights, currency), this agent answers ONLY the weather part
- id: flight_agent
description: |
FlightAgent is an AI-powered tool specialized in providing live flight information between airports. It leverages the FlightAware AeroAPI to deliver real-time flight status, gate information, and delay updates.
Capabilities:
* Get live flight information between airports using FlightAware AeroAPI
* Shows real-time flight status
* Shows scheduled/estimated/actual departure and arrival times
* Shows gate and terminal information
* Shows delays
* Shows aircraft type
* Shows flight status
* Automatically resolves city names to airport codes (IATA/ICAO)
* Understands conversation context to infer origin/destination from follow-up questions
* Handles flight-related questions including "What flights go from [city] to [city]?", "Do flights go to [city]?", "Are there direct flights from [city]?"
* When queries include both flight and other travel questions (e.g., weather, currency), this agent answers ONLY the flight part
tracing:
random_sampling: 100

971
cli/planoai/trace_cmd.py Normal file
View file

@ -0,0 +1,971 @@
import json
import os
import re
import string
import threading
import time
from collections import OrderedDict
from concurrent import futures
from dataclasses import dataclass
from datetime import datetime, timezone
from fnmatch import fnmatch
from typing import Any
import grpc
import rich_click as click
from opentelemetry.proto.collector.trace.v1 import (
trace_service_pb2,
trace_service_pb2_grpc,
)
from rich.console import Console
from rich.text import Text
from rich.tree import Tree
from planoai.consts import PLANO_COLOR
DEFAULT_GRPC_PORT = 4317
MAX_TRACES = 50
MAX_SPANS_PER_TRACE = 500
@dataclass
class TraceSummary:
trace_id: str
start_ns: int
end_ns: int
@property
def total_ms(self) -> float:
return max(0, (self.end_ns - self.start_ns) / 1_000_000)
@property
def timestamp(self) -> str:
if self.start_ns <= 0:
return "unknown"
dt = datetime.fromtimestamp(self.start_ns / 1_000_000_000, tz=timezone.utc)
return dt.astimezone().strftime("%Y-%m-%d %H:%M:%S")
def _parse_filter_patterns(filter_patterns: tuple[str, ...]) -> list[str]:
parts: list[str] = []
for raw in filter_patterns:
for token in raw.split(","):
part = token.strip()
if not part:
raise ValueError("Filter contains empty tokens.")
parts.append(part)
return parts
def _is_hex(value: str, length: int) -> bool:
if len(value) != length:
return False
return all(char in string.hexdigits for char in value)
def _parse_where_filters(where_filters: tuple[str, ...]) -> list[tuple[str, str]]:
parsed: list[tuple[str, str]] = []
invalid: list[str] = []
key_pattern = re.compile(r"^[A-Za-z0-9_.:-]+$")
for raw in where_filters:
if raw.count("=") != 1:
invalid.append(raw)
continue
key, value = raw.split("=", 1)
key = key.strip()
value = value.strip()
if not key or not value or not key_pattern.match(key):
invalid.append(raw)
continue
parsed.append((key, value))
if invalid:
invalid_list = ", ".join(invalid)
raise click.ClickException(
f"Invalid --where filter(s): {invalid_list}. Use key=value."
)
return parsed
def _collect_attr_keys(traces: list[dict[str, Any]]) -> set[str]:
keys: set[str] = set()
for trace in traces:
for span in trace.get("spans", []):
for item in span.get("attributes", []):
key = item.get("key")
if key:
keys.add(str(key))
return keys
def _fetch_traces_raw() -> list[dict[str, Any]]:
port = os.environ.get("PLANO_TRACE_PORT", str(DEFAULT_GRPC_PORT))
target = f"127.0.0.1:{port}"
try:
channel = grpc.insecure_channel(target)
stub = channel.unary_unary(
"/plano.TraceQuery/GetTraces",
request_serializer=lambda x: x,
response_deserializer=lambda x: x,
)
response = stub(b"", timeout=3)
channel.close()
data = json.loads(response)
traces = data.get("traces", [])
if isinstance(traces, list):
return traces
except Exception:
pass
return []
def _attrs(span: dict[str, Any]) -> dict[str, str]:
attrs = {}
for item in span.get("attributes", []):
key = item.get("key")
value_obj = item.get("value", {})
value = value_obj.get("stringValue")
if value is None and "intValue" in value_obj:
value = value_obj.get("intValue")
if value is None and "doubleValue" in value_obj:
value = value_obj.get("doubleValue")
if value is None and "boolValue" in value_obj:
value = value_obj.get("boolValue")
if key is not None and value is not None:
attrs[str(key)] = str(value)
return attrs
def _safe_int(value: Any, default: int = 0) -> int:
try:
return int(value)
except (TypeError, ValueError):
return default
def _parse_since_seconds(value: str | None) -> int | None:
if not value:
return None
value = value.strip()
if not value:
return None
if len(value) < 2:
return None
number, unit = value[:-1], value[-1]
try:
qty = int(number)
except ValueError:
return None
multiplier = {"m": 60, "h": 60 * 60, "d": 60 * 60 * 24}.get(unit)
if multiplier is None:
return None
return qty * multiplier
def _matches_pattern(value: str, pattern: str) -> bool:
if pattern == "*":
return True
if "*" not in pattern:
return value == pattern
parts = [part for part in pattern.split("*") if part]
if not parts:
return True
remaining = value
for idx, part in enumerate(parts):
pos = remaining.find(part)
if pos == -1:
return False
if idx == 0 and not pattern.startswith("*") and pos != 0:
return False
remaining = remaining[pos + len(part) :]
if not pattern.endswith("*") and remaining:
return False
return True
def _attribute_map(span: dict[str, Any]) -> dict[str, str]:
attrs = {}
for item in span.get("attributes", []):
key = item.get("key")
value_obj = item.get("value", {})
value = value_obj.get("stringValue")
if value is None and "intValue" in value_obj:
value = value_obj.get("intValue")
if value is None and "doubleValue" in value_obj:
value = value_obj.get("doubleValue")
if value is None and "boolValue" in value_obj:
value = value_obj.get("boolValue")
if key is not None and value is not None:
attrs[str(key)] = str(value)
return attrs
def _filter_attributes(span: dict[str, Any], patterns: list[str]) -> dict[str, Any]:
if not patterns:
return span
attributes = span.get("attributes", [])
filtered = [
item
for item in attributes
if any(
_matches_pattern(str(item.get("key", "")), pattern) for pattern in patterns
)
]
cloned = dict(span)
cloned["attributes"] = filtered
return cloned
def _filter_traces(
traces: list[dict[str, Any]],
filter_patterns: list[str],
where_filters: list[tuple[str, str]],
since_seconds: int | None,
) -> tuple[list[dict[str, Any]], list[str]]:
now_nanos = int(time.time() * 1_000_000_000)
since_nanos = now_nanos - (since_seconds * 1_000_000_000) if since_seconds else None
filtered_traces: list[dict[str, Any]] = []
for trace in traces:
spans = trace.get("spans", []) or []
if since_nanos is not None:
spans = [
span
for span in spans
if _safe_int(span.get("startTimeUnixNano", 0)) >= since_nanos
]
if filter_patterns:
spans = [_filter_attributes(span, filter_patterns) for span in spans]
if not spans:
continue
candidate = dict(trace)
candidate["spans"] = spans
filtered_traces.append(candidate)
if where_filters:
def matches_where(trace: dict[str, Any]) -> bool:
for key, value in where_filters:
if not any(
_attribute_map(span).get(key) == value
for span in trace.get("spans", [])
):
return False
return True
filtered_traces = [trace for trace in filtered_traces if matches_where(trace)]
trace_ids = [trace.get("trace_id", "") for trace in filtered_traces]
return filtered_traces, trace_ids
class _TraceStore:
"""Thread-safe in-memory store backed by a fixed-length deque.
Spans may arrive with **different** ``traceId`` values but are
linked via ``parentSpanId``. This store groups them into logical
traces by following parent-child span relationships, so all
connected spans end up under a single trace group regardless of
the ``traceId`` they were emitted with.
"""
def __init__(self, max_traces: int = MAX_TRACES) -> None:
self._traces: OrderedDict[str, dict[str, Any]] = OrderedDict()
self._seen_spans: dict[str, set[str]] = {}
# span_id → group key (the trace_id used as the dict key)
self._span_to_group: dict[str, str] = {}
# parent_span_id → group key for spans whose parent arrived first
self._parent_to_group: dict[str, str] = {}
self._max_traces = max_traces
self._lock = threading.Lock()
def _evict_oldest(self) -> None:
"""Remove the oldest trace group (caller must hold *_lock*)."""
if not self._traces:
return
oldest_id, oldest = self._traces.popitem(last=False)
self._seen_spans.pop(oldest_id, None)
for span in oldest.get("spans", []):
sid = span.get("spanId", "")
self._span_to_group.pop(sid, None)
self._parent_to_group.pop(sid, None)
def _merge_groups(self, src_key: str, dst_key: str) -> None:
"""Move all spans from *src_key* group into *dst_key* (caller holds lock)."""
if src_key == dst_key or src_key not in self._traces:
return
src = self._traces.pop(src_key)
dst = self._traces[dst_key]
dst_seen = self._seen_spans[dst_key]
src_seen = self._seen_spans.pop(src_key, set())
for span in src.get("spans", []):
sid = span.get("spanId", "")
if sid and sid not in dst_seen:
dst["spans"].append(span)
dst_seen.add(sid)
self._span_to_group[sid] = dst_key
for sid in src_seen:
self._span_to_group[sid] = dst_key
# Update parent→group mappings that pointed to src.
for pid, gid in list(self._parent_to_group.items()):
if gid == src_key:
self._parent_to_group[pid] = dst_key
def merge_spans(self, trace_id: str, spans: list[dict[str, Any]]) -> None:
"""Merge *spans* into the correct trace group.
The group is determined by following ``parentSpanId`` /
``spanId`` links, falling back to *trace_id* when no link
exists.
"""
with self._lock:
for span in spans:
span_id = span.get("spanId", "")
parent_id = span.get("parentSpanId", "")
# Determine which group this span belongs to.
group_key: str | None = None
# 1. Does the parent already live in a group?
if parent_id and parent_id in self._span_to_group:
group_key = self._span_to_group[parent_id]
# 2. Is this span already known as a parent of another group?
if group_key is None and span_id and span_id in self._parent_to_group:
group_key = self._parent_to_group.pop(span_id)
# 3. Fall back to the wire trace_id.
if group_key is None:
group_key = trace_id
# Create the group if needed.
if group_key not in self._traces:
if len(self._traces) >= self._max_traces:
self._evict_oldest()
self._traces[group_key] = {"trace_id": group_key, "spans": []}
self._seen_spans[group_key] = set()
else:
self._traces.move_to_end(group_key)
# Insert span (deduplicate).
seen = self._seen_spans[group_key]
if span_id and span_id in seen:
continue
if span_id:
seen.add(span_id)
self._span_to_group[span_id] = group_key
if len(self._traces[group_key]["spans"]) < MAX_SPANS_PER_TRACE:
self._traces[group_key]["spans"].append(span)
# Record parent link so future spans can find this group.
if parent_id and parent_id not in self._span_to_group:
self._parent_to_group[parent_id] = group_key
# If this span's span_id is the parent of an existing
# *different* group, merge that group into this one.
if span_id and span_id in self._parent_to_group:
other = self._parent_to_group.pop(span_id)
if other != group_key and other in self._traces:
self._merge_groups(other, group_key)
def snapshot(self) -> list[dict[str, Any]]:
"""Return traces ordered newest-first."""
with self._lock:
traces = list(self._traces.values())
traces.reverse()
return traces
_TRACE_STORE = _TraceStore()
def _anyvalue_to_python(value_obj: Any) -> Any:
"""Convert an opentelemetry AnyValue protobuf to a Python primitive."""
if hasattr(value_obj, "string_value") and value_obj.HasField("value"):
kind = value_obj.WhichOneof("value")
if kind == "string_value":
return value_obj.string_value
if kind == "int_value":
return value_obj.int_value
if kind == "double_value":
return value_obj.double_value
if kind == "bool_value":
return value_obj.bool_value
return None
def _proto_span_to_dict(span: Any, service_name: str) -> dict[str, Any]:
"""Convert a protobuf Span message to the dict format used internally."""
span_dict: dict[str, Any] = {
"traceId": span.trace_id.hex(),
"spanId": span.span_id.hex(),
"parentSpanId": span.parent_span_id.hex() if span.parent_span_id else "",
"name": span.name,
"startTimeUnixNano": str(span.start_time_unix_nano),
"endTimeUnixNano": str(span.end_time_unix_nano),
"service": service_name,
"attributes": [],
}
for kv in span.attributes:
py_val = _anyvalue_to_python(kv.value)
if py_val is not None:
value_dict: dict[str, Any] = {}
if isinstance(py_val, str):
value_dict["stringValue"] = py_val
elif isinstance(py_val, bool):
value_dict["boolValue"] = py_val
elif isinstance(py_val, int):
value_dict["intValue"] = str(py_val)
elif isinstance(py_val, float):
value_dict["doubleValue"] = py_val
span_dict["attributes"].append({"key": kv.key, "value": value_dict})
return span_dict
class _OTLPTraceServicer(trace_service_pb2_grpc.TraceServiceServicer):
"""gRPC servicer that receives OTLP ExportTraceServiceRequest and
merges incoming spans into the global _TRACE_STORE by trace_id."""
_console = Console(stderr=True)
def Export(self, request, context): # noqa: N802
for resource_spans in request.resource_spans:
service_name = "unknown"
for attr in resource_spans.resource.attributes:
if attr.key == "service.name":
val = _anyvalue_to_python(attr.value)
if val is not None:
service_name = str(val)
break
for scope_spans in resource_spans.scope_spans:
for span in scope_spans.spans:
trace_id = span.trace_id.hex()
if not trace_id:
continue
span_dict = _proto_span_to_dict(span, service_name)
_TRACE_STORE.merge_spans(trace_id, [span_dict])
short_id = trace_id[:8]
short_span = span.span_id.hex()[:8]
span_start = (
datetime.fromtimestamp(
span.start_time_unix_nano / 1_000_000_000, tz=timezone.utc
)
.astimezone()
.strftime("%H:%M:%S.%f")[:-3]
)
dur_ns = span.end_time_unix_nano - span.start_time_unix_nano
dur_s = dur_ns / 1_000_000_000
dur_str = f"{dur_s:.3f}".rstrip("0").rstrip(".")
dur_str = f"{dur_str}s"
self._console.print(
f"[dim]{span_start}[/dim], "
f"trace=[yellow]{short_id}[/yellow], "
f"span=[yellow]{short_span}[/yellow], "
f"[bold {_service_color(service_name)}]{service_name}[/bold {_service_color(service_name)}] "
f"[cyan]{span.name}[/cyan] "
f"[dim]({dur_str})[/dim]"
)
return trace_service_pb2.ExportTraceServiceResponse()
class _TraceQueryHandler(grpc.GenericRpcHandler):
"""gRPC handler that serves stored traces to the CLI show command."""
def service(self, handler_call_details):
if handler_call_details.method == "/plano.TraceQuery/GetTraces":
return grpc.unary_unary_rpc_method_handler(
self._get_traces,
request_deserializer=lambda x: x,
response_serializer=lambda x: x,
)
return None
@staticmethod
def _get_traces(_request, _context):
traces = _TRACE_STORE.snapshot()
return json.dumps({"traces": traces}, separators=(",", ":")).encode("utf-8")
def _create_trace_server(host: str, grpc_port: int) -> grpc.Server:
"""Create, bind, and start an OTLP/gRPC trace-collection server.
Returns the running ``grpc.Server``. The caller is responsible
for calling ``server.stop()`` when done.
"""
grpc_server = grpc.server(
futures.ThreadPoolExecutor(max_workers=4),
handlers=[_TraceQueryHandler()],
)
trace_service_pb2_grpc.add_TraceServiceServicer_to_server(
_OTLPTraceServicer(), grpc_server
)
grpc_server.add_insecure_port(f"{host}:{grpc_port}")
grpc_server.start()
return grpc_server
def _start_trace_listener(host: str, grpc_port: int) -> None:
"""Start the OTLP/gRPC listener and block until interrupted."""
console = Console()
grpc_server = _create_trace_server(host, grpc_port)
console.print()
console.print(f"[bold {PLANO_COLOR}]Listening for traces...[/bold {PLANO_COLOR}]")
console.print(
f"[green]●[/green] gRPC (OTLP receiver) on [cyan]{host}:{grpc_port}[/cyan]"
)
console.print("[dim]Press Ctrl+C to stop.[/dim]")
console.print()
try:
grpc_server.wait_for_termination()
except KeyboardInterrupt:
pass
finally:
grpc_server.stop(grace=2)
def start_trace_listener_background(
host: str = "0.0.0.0", grpc_port: int = DEFAULT_GRPC_PORT
) -> grpc.Server:
"""Start the trace listener in the background (non-blocking).
Returns the running ``grpc.Server`` so the caller can call
``server.stop()`` later.
"""
return _create_trace_server(host, grpc_port)
def _span_time_ns(span: dict[str, Any], key: str) -> int:
try:
return int(span.get(key, 0))
except (TypeError, ValueError):
return 0
def _trace_id_short(trace_id: str) -> str:
return trace_id[:8] if trace_id else "unknown"
def _trace_summary(trace: dict[str, Any]) -> TraceSummary:
spans = trace.get("spans", [])
start_ns = min((_span_time_ns(s, "startTimeUnixNano") for s in spans), default=0)
end_ns = max((_span_time_ns(s, "endTimeUnixNano") for s in spans), default=0)
return TraceSummary(
trace_id=trace.get("trace_id", "unknown"),
start_ns=start_ns,
end_ns=end_ns,
)
def _service_color(service: str) -> str:
service = service.lower()
if "inbound" in service:
return "white"
if "outbound" in service:
return "white"
if "orchestrator" in service:
return PLANO_COLOR
if "routing" in service:
return "magenta"
if "agent" in service:
return "cyan"
if "llm" in service:
return "green"
return "white"
# Attributes to show for inbound/outbound spans when not verbose (trimmed view).
_INBOUND_OUTBOUND_ATTR_KEYS = (
"http.method",
"http.target",
"http.status_code",
"url.scheme",
"guid:x-request-id",
"request_size",
"response_size",
)
def _trim_attrs_for_display(
attrs: dict[str, str], service: str, verbose: bool
) -> dict[str, str]:
if verbose:
return attrs
if "inbound" in service.lower() or "outbound" in service.lower():
attrs = {k: v for k, v in attrs.items() if k in _INBOUND_OUTBOUND_ATTR_KEYS}
return {k: v for k, v in attrs.items() if k != "service.name.override"}
def _sorted_attr_items(attrs: dict[str, str]) -> list[tuple[str, str]]:
priority = [
"http.method",
"http.target",
"http.status_code",
"guid:x-request-id",
"request_size",
"response_size",
"routing.determination_ms",
"route.selected_model",
"selection.agents",
"selection.agent_count",
"agent.name",
"agent.sequence",
"duration_ms",
"llm.model",
"llm.is_streaming",
"llm.time_to_first_token",
"llm.duration_ms",
"llm.response_bytes",
]
prioritized = [(k, attrs[k]) for k in priority if k in attrs]
prioritized_keys = {k for k, _ in prioritized}
remaining = [(k, v) for k, v in attrs.items() if k not in prioritized_keys]
remaining.sort(key=lambda item: item[0])
return prioritized + remaining
def _display_attr_value(key: str, value: str) -> str:
if key == "http.status_code" and value != "200":
return f"{value} ⚠️"
return value
def _build_tree(trace: dict[str, Any], console: Console, verbose: bool = False) -> None:
spans = trace.get("spans", [])
if not spans:
console.print("[yellow]No spans found for this trace.[/yellow]")
return
start_ns = min((_span_time_ns(s, "startTimeUnixNano") for s in spans), default=0)
end_ns = max((_span_time_ns(s, "endTimeUnixNano") for s in spans), default=0)
total_ms = max(0, (end_ns - start_ns) / 1_000_000)
trace_id = trace.get("trace_id", "unknown")
console.print(
f"\n[bold]Trace:[/bold] {trace_id} [dim]({total_ms:.0f}ms total)[/dim]\n"
)
spans.sort(key=lambda s: _span_time_ns(s, "startTimeUnixNano"))
tree = Tree("", guide_style="dim")
for span in spans:
service = span.get("service", "plano(unknown)")
name = span.get("name", "")
offset_ms = max(
0, (_span_time_ns(span, "startTimeUnixNano") - start_ns) / 1_000_000
)
color = _service_color(service)
label = Text(f"{offset_ms:.0f}ms ", style="yellow")
label.append(service, style=f"bold {color}")
if name:
label.append(f" {name}", style="dim white")
node = tree.add(label)
attrs = _trim_attrs_for_display(_attrs(span), service, verbose)
sorted_items = list(_sorted_attr_items(attrs))
for idx, (key, value) in enumerate(sorted_items):
attr_line = Text()
attr_line.append(f"{key}: ", style="white")
attr_line.append(
_display_attr_value(key, str(value)),
style=f"{PLANO_COLOR}",
)
if idx == len(sorted_items) - 1:
attr_line.append("\n")
node.add(attr_line)
console.print(tree)
console.print()
def _select_request(
console: Console, traces: list[dict[str, Any]]
) -> dict[str, Any] | None:
try:
import questionary
from questionary import Choice
from prompt_toolkit.styles import Style
except ImportError as exc:
raise click.ClickException(
"Interactive selection requires 'questionary'. "
"Install it or rerun with --json."
) from exc
if not traces:
return None
style = Style.from_dict(
{
"qmark": f"fg:{PLANO_COLOR} bold",
"question": "bold",
"answer": f"fg:{PLANO_COLOR} bold",
"pointer": f"fg:{PLANO_COLOR} bold",
"highlighted": f"fg:{PLANO_COLOR} bold",
"selected": f"fg:{PLANO_COLOR}",
"instruction": "fg:#888888",
"text": "",
"disabled": "fg:#666666",
}
)
choices = []
for trace in traces:
summary = _trace_summary(trace)
label = f"{_trace_id_short(summary.trace_id)} ({summary.total_ms:.0f}ms total • {summary.timestamp})"
choices.append(Choice(label, value=trace))
selected = questionary.select(
"Select a trace to view:",
choices=choices,
style=style,
pointer="❯",
).ask()
if not selected:
console.print("[dim]Cancelled.[/dim]")
return None
return selected
@click.argument("target", required=False)
@click.option(
"--filter",
"filter_patterns",
multiple=True,
help=(
"Limit displayed attributes to matching keys "
"(wildcards supported). Repeatable."
),
)
@click.option(
"--where",
"where_filters",
multiple=True,
help="Match traces that contain key=value. Repeatable (AND semantics).",
)
@click.option("--list", "list_only", is_flag=True, help="List trace IDs only.")
@click.option(
"--no-interactive",
is_flag=True,
help="Disable interactive prompts and selections.",
)
@click.option("--limit", type=int, default=None, help="Limit results.")
@click.option("--since", default=None, help="Look back window (e.g. 5m, 2h, 1d).")
@click.option("--json", "json_out", is_flag=True, help="Output raw JSON.")
@click.option(
"--verbose",
"-v",
is_flag=True,
help="Show all span attributes; default trims inbound/outbound to a few keys.",
)
def _run_trace_show(
target,
filter_patterns,
where_filters,
list_only,
no_interactive,
limit,
since,
json_out,
verbose,
):
"""Trace requests from the local OTLP listener."""
console = Console()
try:
patterns = _parse_filter_patterns(filter_patterns)
except ValueError as exc:
raise click.ClickException(str(exc)) from exc
parsed_where = _parse_where_filters(where_filters)
if limit is not None and limit < 0:
raise click.ClickException("Limit must be greater than or equal to 0.")
since_seconds = _parse_since_seconds(since)
if target is None:
target = "any" if list_only or since or limit else "last"
if list_only and target not in (None, "last", "any"):
raise click.ClickException("Target and --list cannot be used together.")
short_target = None
if isinstance(target, str) and target not in ("last", "any"):
target_lower = target.lower()
if len(target_lower) == 8:
if not _is_hex(target_lower, 8) or target_lower == "00000000":
raise click.ClickException("Short trace ID must be 8 hex characters.")
short_target = target_lower
elif len(target_lower) == 32:
if not _is_hex(target_lower, 32) or target_lower == "0" * 32:
raise click.ClickException("Trace ID must be 32 hex characters.")
else:
raise click.ClickException("Trace ID must be 8 or 32 hex characters.")
traces_raw = _fetch_traces_raw()
if traces_raw:
available_keys = _collect_attr_keys(traces_raw)
if parsed_where:
missing_keys = [key for key, _ in parsed_where if key not in available_keys]
if missing_keys:
missing_list = ", ".join(missing_keys)
raise click.ClickException(f"Unknown --where key(s): {missing_list}")
if patterns:
unmatched = [
pattern
for pattern in patterns
if not any(fnmatch(key, pattern) for key in available_keys)
]
if unmatched:
unmatched_list = ", ".join(unmatched)
console.print(
f"[yellow]Warning:[/yellow] Filter key(s) not found: {unmatched_list}. "
"Returning unfiltered traces."
)
traces, trace_ids = _filter_traces(
traces_raw, patterns, parsed_where, since_seconds
)
if target == "last":
traces = traces[:1]
trace_ids = trace_ids[:1]
elif target not in (None, "any") and short_target is None:
traces = [trace for trace in traces if trace.get("trace_id") == target]
trace_ids = [trace.get("trace_id") for trace in traces]
if short_target:
traces = [
trace
for trace in traces
if trace.get("trace_id", "").lower().startswith(short_target)
]
trace_ids = [trace.get("trace_id") for trace in traces]
if limit is not None:
if list_only:
trace_ids = trace_ids[:limit]
else:
traces = traces[:limit]
if json_out:
if list_only:
console.print_json(data={"trace_ids": trace_ids})
else:
console.print_json(data={"traces": traces})
return
if list_only:
if traces and console.is_terminal and not no_interactive:
selected = _select_request(console, traces)
if selected:
_build_tree(selected, console, verbose=verbose)
return
if traces:
trace_ids = [_trace_id_short(_trace_summary(t).trace_id) for t in traces]
if not trace_ids:
console.print("[yellow]No trace IDs found.[/yellow]")
return
console.print("\n[bold]Trace IDs:[/bold]")
for trace_id in trace_ids:
console.print(f" [dim]-[/dim] {trace_id}")
return
if not traces:
console.print("[yellow]No traces found.[/yellow]")
return
trace_obj = traces[0]
_build_tree(trace_obj, console, verbose=verbose)
@click.group(invoke_without_command=True)
@click.argument("target", required=False)
@click.option(
"--filter",
"filter_patterns",
multiple=True,
help=(
"Limit displayed attributes to matching keys "
"(wildcards supported). Repeatable."
),
)
@click.option(
"--where",
"where_filters",
multiple=True,
help="Match traces that contain key=value. Repeatable (AND semantics).",
)
@click.option("--list", "list_only", is_flag=True, help="List trace IDs only.")
@click.option(
"--no-interactive",
is_flag=True,
help="Disable interactive prompts and selections.",
)
@click.option("--limit", type=int, default=None, help="Limit results.")
@click.option("--since", default=None, help="Look back window (e.g. 5m, 2h, 1d).")
@click.option("--json", "json_out", is_flag=True, help="Output raw JSON.")
@click.option(
"--verbose",
"-v",
is_flag=True,
help="Show all span attributes; default trims inbound/outbound to a few keys.",
)
@click.pass_context
def trace(
ctx,
target,
filter_patterns,
where_filters,
list_only,
no_interactive,
limit,
since,
json_out,
verbose,
):
"""Trace requests from the local OTLP listener."""
if ctx.invoked_subcommand:
return
if target == "listen" and not any(
[
filter_patterns,
where_filters,
list_only,
no_interactive,
limit,
since,
json_out,
verbose,
]
):
_start_trace_listener("0.0.0.0", DEFAULT_GRPC_PORT)
return
_run_trace_show(
target,
filter_patterns,
where_filters,
list_only,
no_interactive,
limit,
since,
json_out,
verbose,
)
@trace.command("listen")
@click.option("--host", default="0.0.0.0", show_default=True)
@click.option(
"--port",
type=int,
default=DEFAULT_GRPC_PORT,
show_default=True,
help="gRPC port for receiving OTLP traces.",
)
def trace_listen(host: str, port: int) -> None:
"""Listen for OTLP/gRPC traces."""
_start_trace_listener(host, port)

70
cli/planoai/versioning.py Normal file
View file

@ -0,0 +1,70 @@
import importlib.metadata
import re
PYPI_PACKAGE_NAME = "planoai"
PYPI_URL = f"https://pypi.org/pypi/{PYPI_PACKAGE_NAME}/json"
def get_version() -> str:
try:
# First try package metadata (installed package).
return importlib.metadata.version(PYPI_PACKAGE_NAME)
except importlib.metadata.PackageNotFoundError:
# Fallback to local development version.
try:
from planoai import __version__
return __version__
except ImportError:
return "version not found"
def get_latest_version(timeout: float = 2.0) -> str | None:
"""Fetch the latest version from PyPI."""
import requests
try:
response = requests.get(PYPI_URL, timeout=timeout)
if response.status_code == 200:
data = response.json()
return data.get("info", {}).get("version")
except (requests.RequestException, ValueError):
# Network error or invalid JSON - fail silently.
return None
return None
def parse_version(version_str: str) -> tuple[int, ...]:
"""Parse version string into a comparable tuple."""
clean_version = re.split(r"[a-zA-Z]", version_str)[0]
parts = clean_version.split(".")
return tuple(int(p) for p in parts if p.isdigit())
def check_version_status(
current: str, latest: str | None
) -> dict[str, str | bool | None]:
"""Compare current version with latest and return status metadata."""
if latest is None:
return {
"is_outdated": False,
"current": current,
"latest": None,
"message": None,
}
try:
is_outdated = parse_version(current) < parse_version(latest)
return {
"is_outdated": is_outdated,
"current": current,
"latest": latest,
"message": f"Update available: {latest}" if is_outdated else None,
}
except (ValueError, TypeError):
return {
"is_outdated": False,
"current": current,
"latest": latest,
"message": None,
}