Repository navigation
feat(w1): agent terminal renderer — JSON, whoami, remote gateway #1101
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| """Tests for agent_terminal_theme.py — W1 Agent Theming renderer.""" | ||
| import json | ||
| import os | ||
| import subprocess | ||
| import sys | ||
|
|
||
| import pytest | ||
|
|
||
| _SCRIPT = os.path.join(os.path.dirname(__file__), "..", "tools", "agent_terminal_theme.py") | ||
|
|
||
|
|
||
| def _run(args: list[str], env_override: dict[str, str] | None = None) -> subprocess.CompletedProcess: | ||
| """Run the theme script and return result.""" | ||
| env = {**os.environ, "PYTHONUTF8": "1", **(env_override or {})} | ||
| return subprocess.run( | ||
| [sys.executable, _SCRIPT, *args], | ||
| capture_output=True, text=True, encoding="utf-8", errors="replace", | ||
| env=env, timeout=10, | ||
| ) | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Banner output | ||
| # --------------------------------------------------------------------------- | ||
| class TestBanner: | ||
| def test_known_agent_banner(self): | ||
| result = _run(["--agent", "4090-claude", "--banner"]) | ||
| assert result.returncode == 0 | ||
| assert "4090 Claude" in result.stdout | ||
| assert "noise-reducer" in result.stdout | ||
|
|
||
| def test_unknown_agent_fails(self): | ||
| result = _run(["--agent", "nonexistent-agent", "--banner"]) | ||
| assert result.returncode == 1 | ||
| assert "not found" in result.stderr | ||
|
|
||
| def test_no_color_strips_ansi(self): | ||
| result = _run(["--agent", "claude-opus", "--banner"], env_override={"NO_COLOR": "1"}) | ||
| assert result.returncode == 0 | ||
| assert "\033[" not in result.stdout | ||
| assert "Claude Opus" in result.stdout | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Status bar | ||
| # --------------------------------------------------------------------------- | ||
| class TestStatusBar: | ||
| def test_status_output(self): | ||
| result = _run(["--agent", "4090-claude", "--status", "P7 live, 11/20 TAC"]) | ||
| assert result.returncode == 0 | ||
| assert "P7 live" in result.stdout | ||
| assert "4090 Claude" in result.stdout | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Session header | ||
| # --------------------------------------------------------------------------- | ||
| class TestSessionHeader: | ||
| def test_session_with_node(self): | ||
| result = _run(["--agent", "4090-claude", "--session"]) | ||
| assert result.returncode == 0 | ||
| assert "RTX 4090" in result.stdout | ||
| assert "noise-reducer" in result.stdout | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # JSON output | ||
| # --------------------------------------------------------------------------- | ||
| class TestJsonOutput: | ||
| def test_agent_json(self): | ||
| result = _run(["--agent", "4090-claude", "--json"]) | ||
| assert result.returncode == 0 | ||
| data = json.loads(result.stdout) | ||
| assert data["agent_id"] == "4090-claude" | ||
| assert data["color"] == "#0D9488" | ||
| assert data["glyph"] == "\u25c9" | ||
| assert "noise-reducer" == data["specialization"] | ||
|
|
||
| def test_session_json_includes_node(self): | ||
| result = _run(["--agent", "z890-claude", "--session", "--json"]) | ||
| assert result.returncode == 0 | ||
| data = json.loads(result.stdout) | ||
| assert data["agent_id"] == "z890-claude" | ||
| assert "node" in data | ||
| assert data["node"]["hardware"]["gpu"] == "RTX 3090 Ti 24GB" | ||
|
|
||
| def test_roster_json(self): | ||
| result = _run(["--roster", "--json"]) | ||
| assert result.returncode == 0 | ||
| data = json.loads(result.stdout) | ||
| assert isinstance(data, list) | ||
| assert len(data) >= 10 | ||
| ids = [a["agent_id"] for a in data] | ||
| assert "claude-opus" in ids | ||
| assert "4090-claude" in ids | ||
|
|
||
| def test_demo_json(self): | ||
| result = _run(["--demo", "--json"]) | ||
| assert result.returncode == 0 | ||
| data = json.loads(result.stdout) | ||
| assert isinstance(data, list) | ||
| assert all("glyph" in a for a in data) | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Whoami | ||
| # --------------------------------------------------------------------------- | ||
| class TestWhoami: | ||
| def test_whoami_with_env(self): | ||
| result = _run(["--whoami", "--json"], env_override={"PMOVES_AGENT_ID": "4090-claude"}) | ||
| assert result.returncode == 0 | ||
| data = json.loads(result.stdout) | ||
| assert data["agent_id"] == "4090-claude" | ||
| assert data["source"] == "env" | ||
| assert data["theme"]["color"] == "#0D9488" | ||
|
|
||
| def test_whoami_heuristic_fallback(self): | ||
| """Without PMOVES_AGENT_ID, whoami uses hostname heuristic.""" | ||
| env = {k: v for k, v in os.environ.items() if k != "PMOVES_AGENT_ID"} | ||
| env["PYTHONUTF8"] = "1" | ||
| result = subprocess.run( | ||
| [sys.executable, _SCRIPT, "--whoami", "--json"], | ||
| capture_output=True, text=True, encoding="utf-8", errors="replace", | ||
| env=env, timeout=10, | ||
| ) | ||
| assert result.returncode == 0 | ||
| data = json.loads(result.stdout) | ||
| assert "agent_id" in data | ||
| assert data["source"] == "heuristic" | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Remote fallback | ||
| # --------------------------------------------------------------------------- | ||
| class TestRemoteFallback: | ||
| def test_remote_falls_back_to_local(self): | ||
| """When gateway is unreachable, --remote should fallback to local YAML.""" | ||
| result = _run( | ||
| ["--agent", "claude-opus", "--remote", "--banner"], | ||
| env_override={"BOTZ_GATEWAY_URL": "http://localhost:19999"}, | ||
| ) | ||
| assert result.returncode == 0 | ||
| assert "Claude Opus" in result.stdout | ||
| assert "falling back" in result.stderr.lower() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Demo and roster | ||
| # --------------------------------------------------------------------------- | ||
| class TestDemoRoster: | ||
| def test_demo_renders_all(self): | ||
| result = _run(["--demo"]) | ||
| assert result.returncode == 0 | ||
| assert "Claude Opus" in result.stdout | ||
| assert "4090 Claude" in result.stdout | ||
| assert "Agent Roster" in result.stdout | ||
|
|
||
| def test_roster_compact(self): | ||
| result = _run(["--roster"]) | ||
| assert result.returncode == 0 | ||
| assert "Agent Roster" in result.stdout | ||
|
|
||
| def test_no_args_shows_help(self): | ||
| result = _run([]) | ||
| assert result.returncode == 1 | ||
| assert "usage:" in result.stdout.lower() or "usage:" in result.stderr.lower() |
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -11,13 +11,18 @@ | |||||
| python pmoves/tools/agent_terminal_theme.py --agent 4090-claude | ||||||
| python pmoves/tools/agent_terminal_theme.py --agent claude-opus --banner | ||||||
| python pmoves/tools/agent_terminal_theme.py --agent z890-claude --status "19 containers healthy" | ||||||
| python pmoves/tools/agent_terminal_theme.py --whoami | ||||||
| python pmoves/tools/agent_terminal_theme.py --agent 4090-claude --json | ||||||
| python pmoves/tools/agent_terminal_theme.py --agent z890-claude --remote --banner | ||||||
| """ | ||||||
| from __future__ import annotations | ||||||
|
|
||||||
| import argparse | ||||||
| import json as json_mod | ||||||
| import os | ||||||
| import socket | ||||||
| import sys | ||||||
| from dataclasses import dataclass | ||||||
| from dataclasses import dataclass, asdict | ||||||
| from pathlib import Path | ||||||
| from typing import Any, Dict, List, Optional | ||||||
|
|
||||||
|
|
@@ -185,6 +190,72 @@ def load_node(node_id: str) -> Optional[NodeSpec]: | |||||
| ) | ||||||
|
|
||||||
|
|
||||||
| # --------------------------------------------------------------------------- | ||||||
| # Remote loader (BoTZ Gateway at :8054) | ||||||
| # --------------------------------------------------------------------------- | ||||||
| _BOTZ_GATEWAY_URL = os.getenv("BOTZ_GATEWAY_URL", "http://localhost:8054") | ||||||
|
|
||||||
|
|
||||||
| def _fetch_remote_theme(agent_id: str) -> Optional[AgentTheme]: | ||||||
| """Fetch agent theme from BoTZ Gateway API. Returns None on failure.""" | ||||||
| try: | ||||||
| from urllib.request import urlopen, Request | ||||||
| from urllib.error import URLError | ||||||
| except ImportError: | ||||||
| return None | ||||||
| try: | ||||||
| req = Request(f"{_BOTZ_GATEWAY_URL}/v1/agent/theme/{agent_id}") | ||||||
| req.add_header("Accept", "application/json") | ||||||
| with urlopen(req, timeout=3) as resp: | ||||||
| data = json_mod.loads(resp.read()) | ||||||
| return AgentTheme( | ||||||
| agent_id=data.get("agent_id", agent_id), | ||||||
| display_name=data.get("display_name", agent_id), | ||||||
| glyph=data.get("glyph", "?"), | ||||||
| color=data.get("color", "#FFFFFF"), | ||||||
| accent=data.get("accent", "#CCCCCC"), | ||||||
| voice=data.get("voice", "analytical"), | ||||||
| resonance=data.get("resonance", []), | ||||||
| description=data.get("description", ""), | ||||||
| ) | ||||||
| except (URLError, OSError, json_mod.JSONDecodeError, KeyError): | ||||||
| return None | ||||||
|
Comment on lines
+199
to
+222
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Validate URL scheme to prevent SSRF via The static analysis tool (Ruff S310) correctly flags that 🛡️ Proposed fix to validate URL scheme def _fetch_remote_theme(agent_id: str) -> Optional[AgentTheme]:
"""Fetch agent theme from BoTZ Gateway API. Returns None on failure."""
try:
from urllib.request import urlopen, Request
from urllib.error import URLError
+ from urllib.parse import urlparse
except ImportError:
return None
+ url = f"{_BOTZ_GATEWAY_URL}/v1/agent/theme/{agent_id}"
+ if urlparse(url).scheme not in ("http", "https"):
+ return None
try:
- req = Request(f"{_BOTZ_GATEWAY_URL}/v1/agent/theme/{agent_id}")
+ req = Request(url)
req.add_header("Accept", "application/json")
with urlopen(req, timeout=3) as resp:
data = json_mod.loads(resp.read())🧰 Tools🪛 Ruff (0.15.6)[error] 207-207: Audit URL open for permitted schemes. Allowing use of (S310) [error] 209-209: Audit URL open for permitted schemes. Allowing use of (S310) 🤖 Prompt for AI Agents |
||||||
|
|
||||||
|
|
||||||
| def _resolve_whoami() -> str: | ||||||
| """Resolve current agent identity from environment, hostname, or gateway.""" | ||||||
| # 1. Explicit env var (set by BoTZ session or Claude Code hook) | ||||||
| agent_id = os.environ.get("PMOVES_AGENT_ID") | ||||||
| if agent_id: | ||||||
| return agent_id | ||||||
|
|
||||||
| # 2. Hostname heuristic — match against known node patterns | ||||||
| hostname = socket.gethostname().lower() | ||||||
| _HOST_MAP = { | ||||||
| "z890": "z890-claude", | ||||||
| "pmoves-z890": "z890-claude", | ||||||
| "powerfulmoves": "5090-claude", | ||||||
| "pmoves-powerfulmoves": "5090-claude", | ||||||
| "laptop": "4090-claude", | ||||||
| "pmoves-laptop": "4090-claude", | ||||||
| } | ||||||
| for pattern, aid in _HOST_MAP.items(): | ||||||
| if pattern in hostname: | ||||||
| return aid | ||||||
|
|
||||||
| # 3. Gateway fallback | ||||||
| try: | ||||||
| from urllib.request import urlopen, Request | ||||||
| from urllib.error import URLError | ||||||
| req = Request(f"{_BOTZ_GATEWAY_URL}/v1/agent/whoami") | ||||||
| req.add_header("Accept", "application/json") | ||||||
| with urlopen(req, timeout=3) as resp: | ||||||
| data = json_mod.loads(resp.read()) | ||||||
| return data.get("agent_id", "unknown") | ||||||
| except Exception: | ||||||
| return "unknown" | ||||||
|
Comment on lines
+246
to
+256
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Validate URL scheme and narrow exception handling in gateway fallback. Same SSRF concern applies here. Additionally, the blind 🛡️ Proposed fix # 3. Gateway fallback
try:
from urllib.request import urlopen, Request
from urllib.error import URLError
+ from urllib.parse import urlparse
+ url = f"{_BOTZ_GATEWAY_URL}/v1/agent/whoami"
+ if urlparse(url).scheme not in ("http", "https"):
+ return "unknown"
- req = Request(f"{_BOTZ_GATEWAY_URL}/v1/agent/whoami")
+ req = Request(url)
req.add_header("Accept", "application/json")
with urlopen(req, timeout=3) as resp:
data = json_mod.loads(resp.read())
return data.get("agent_id", "unknown")
- except Exception:
+ except (URLError, OSError, json_mod.JSONDecodeError, KeyError, ValueError):
return "unknown"🧰 Tools🪛 Ruff (0.15.6)[error] 250-250: Audit URL open for permitted schemes. Allowing use of (S310) [error] 252-252: Audit URL open for permitted schemes. Allowing use of (S310) [warning] 255-255: Do not catch blind exception: (BLE001) 🤖 Prompt for AI Agents |
||||||
|
|
||||||
|
|
||||||
| # --------------------------------------------------------------------------- | ||||||
| # Renderers | ||||||
| # --------------------------------------------------------------------------- | ||||||
|
|
@@ -272,14 +343,45 @@ def main() -> int: | |||||
| parser.add_argument("--session", action="store_true", help="Render session header") | ||||||
| parser.add_argument("--roster", action="store_true", help="Render all agents roster") | ||||||
| parser.add_argument("--demo", action="store_true", help="Render demo of all agents") | ||||||
| parser.add_argument("--json", action="store_true", dest="json_out", help="Output as JSON (for BoTZ CLI / P7 integration)") | ||||||
| parser.add_argument("--remote", action="store_true", help="Fetch theme from BoTZ Gateway instead of local YAML") | ||||||
| parser.add_argument("--whoami", action="store_true", help="Resolve and display current agent identity") | ||||||
|
|
||||||
| args = parser.parse_args() | ||||||
|
|
||||||
| # --- whoami: resolve identity and display --- | ||||||
| if args.whoami: | ||||||
| agent_id = _resolve_whoami() | ||||||
| agent = load_agent(agent_id) | ||||||
| node = load_node(agent_id) | ||||||
| if args.json_out: | ||||||
| result = {"agent_id": agent_id, "source": "env" if os.environ.get("PMOVES_AGENT_ID") else "heuristic"} | ||||||
| if agent: | ||||||
| result["theme"] = {"glyph": agent.glyph, "color": agent.color, "accent": agent.accent, "voice": agent.voice} | ||||||
| result["display_name"] = agent.display_name | ||||||
| result["specialization"] = agent.specialization | ||||||
| result["resonance"] = agent.resonance | ||||||
| if node: | ||||||
| result["node"] = {"hardware": node.hardware, "specialization": node.specialization} | ||||||
| print(json_mod.dumps(result, indent=2)) | ||||||
|
Comment on lines
+352
to
+366
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Source detection is inaccurate when gateway fallback is used. The Consider having ♻️ Proposed approach-def _resolve_whoami() -> str:
+def _resolve_whoami() -> tuple[str, str]:
- """Resolve current agent identity from environment, hostname, or gateway."""
+ """Resolve current agent identity. Returns (agent_id, source)."""
# 1. Explicit env var
agent_id = os.environ.get("PMOVES_AGENT_ID")
if agent_id:
- return agent_id
+ return agent_id, "env"
# 2. Hostname heuristic
...
for pattern, aid in _HOST_MAP.items():
if pattern in hostname:
- return aid
+ return aid, "heuristic"
# 3. Gateway fallback
...
- return data.get("agent_id", "unknown")
+ return data.get("agent_id", "unknown"), "gateway"
except ...:
- return "unknown"
+ return "unknown", "unknown"Then update the caller: - agent_id = _resolve_whoami()
+ agent_id, source = _resolve_whoami()
...
- result = {"agent_id": agent_id, "source": "env" if os.environ.get("PMOVES_AGENT_ID") else "heuristic"}
+ result = {"agent_id": agent_id, "source": source}🤖 Prompt for AI Agents |
||||||
| else: | ||||||
| if agent: | ||||||
| print(render_banner(agent)) | ||||||
| if node: | ||||||
| print() | ||||||
| print(session_header(agent, node)) | ||||||
| else: | ||||||
| print(f"Resolved agent: {agent_id} (not in signatures)") | ||||||
| return 0 | ||||||
|
|
||||||
| if args.demo: | ||||||
| agents = load_all_agents() | ||||||
| if not agents: | ||||||
| print("ERROR: Could not load agent_signatures.yaml", file=sys.stderr) | ||||||
| return 1 | ||||||
| if args.json_out: | ||||||
| print(json_mod.dumps([asdict(a) for a in agents], indent=2)) | ||||||
| return 0 | ||||||
| for agent in agents: | ||||||
| print(render_banner(agent)) | ||||||
| print() | ||||||
|
|
@@ -288,18 +390,41 @@ def main() -> int: | |||||
|
|
||||||
| if args.roster: | ||||||
| agents = load_all_agents() | ||||||
| if args.json_out: | ||||||
| print(json_mod.dumps([{"agent_id": a.agent_id, "glyph": a.glyph, "color": a.color, "specialization": a.specialization} for a in agents], indent=2)) | ||||||
| return 0 | ||||||
| print(render_roster(agents)) | ||||||
| return 0 | ||||||
|
|
||||||
| if not args.agent: | ||||||
| parser.print_help() | ||||||
| return 1 | ||||||
|
|
||||||
| agent = load_agent(args.agent) | ||||||
| # --- Load agent (local or remote) --- | ||||||
| agent = None | ||||||
| if args.remote: | ||||||
| agent = _fetch_remote_theme(args.agent) | ||||||
| if agent: | ||||||
| print(f"{_DIM}(remote: {_BOTZ_GATEWAY_URL}){_RESET}", file=sys.stderr) | ||||||
| else: | ||||||
| print(f"WARNING: Gateway unreachable, falling back to local YAML", file=sys.stderr) | ||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Remove unnecessary f-string prefix. This f-string has no placeholders. Ruff F541 correctly flags this as extraneous. ✏️ Proposed fix- print(f"WARNING: Gateway unreachable, falling back to local YAML", file=sys.stderr)
+ print("WARNING: Gateway unreachable, falling back to local YAML", file=sys.stderr)📝 Committable suggestion
Suggested change
🧰 Tools🪛 Ruff (0.15.6)[error] 410-410: f-string without any placeholders Remove extraneous (F541) 🤖 Prompt for AI Agents |
||||||
| if not agent: | ||||||
| agent = load_agent(args.agent) | ||||||
| if not agent: | ||||||
| print(f"ERROR: Agent '{args.agent}' not found in {_SIGNATURES_PATH}", file=sys.stderr) | ||||||
| return 1 | ||||||
|
|
||||||
| # --- JSON output mode --- | ||||||
| if args.json_out: | ||||||
| result = asdict(agent) | ||||||
| if args.session: | ||||||
| node = load_node(args.node or args.agent) | ||||||
| if node: | ||||||
| result["node"] = asdict(node) | ||||||
| print(json_mod.dumps(result, indent=2)) | ||||||
|
Comment on lines
+419
to
+424
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
When Useful? React with 👍 / 👎. |
||||||
| return 0 | ||||||
|
|
||||||
| # --- ANSI render modes --- | ||||||
| if args.session: | ||||||
| node = load_node(args.node or args.agent) | ||||||
| print(session_header(agent, node)) | ||||||
|
|
||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Duplicate CLAIM entry — remove this line.
Line 99 is an exact duplicate of the existing claim at line 89. Both have identical timestamp, agent, scope, and branch. This appears to be an accidental duplication that should be removed to maintain a clean claim register.
✏️ Proposed fix
📝 Committable suggestion
🤖 Prompt for AI Agents