From 4550b1af3aac6380b64128891e419ba7aef48e0b Mon Sep 17 00:00:00 2001 From: ar9av Date: Fri, 31 Jul 2026 22:14:48 -0700 Subject: [PATCH 1/2] feat(term): full-screen terminal console for agents, sessions and cost MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `prismor term`, a curses console over the existing store: an agent -> session tree, a live event tail scoped to the selection, policy precedence, and per-session token cost. Renderer only — every number comes from prismor.runtime.store, so secrets stay cloaked and no new SQL is introduced. Degrades to a plain table when stdout isn't a tty, so it stays scriptable. Cost (prismor/runtime/cost.py) joins two sources, because the event store records what an agent did but never how many tokens it took: live prices from aipricing.guru's published feed (cached to ~/.prismor/pricing-cache.json, 12h TTL, serves stale when offline) and usage from Claude Code's own transcripts, which key on the same session ids Prismor already records. Sessions without a transcript render as unknown, never $0.00. The feed publishes no cache-write rate, so that component uses Anthropic's standard 1.25x input multiplier and totals are labelled estimates. Kept responsive under measurement, not assumption: - get_aggregate_stats (~900ms) is off the startup and draw paths entirely; the KPI panel paints immediately and fills in on idle - events paginate to the visible row count rather than a fixed 200 (55ms -> 11ms per keypress) - navigation is debounced 120ms, so holding j/k queues no queries - session pricing runs in small batches on idle - the screen repaints only when something changed Time to first paint 0.93s -> 0.11s; 40 keys with no gap settle in 1.2s. get_events_page derives total/pages from a window that grows with page depth, so those numbers are reported as a floor ("page 3 of 207+") rather than presented as an exact count. Session-scoped events are sliced locally and do show exact pages. --- docs/cli-reference.md | 2 + prismor/runtime/cli.py | 18 + prismor/runtime/cost.py | 291 +++++++ prismor/runtime/immunity_cli.py | 2 +- prismor/runtime/term.py | 1254 +++++++++++++++++++++++++++++++ 5 files changed, 1566 insertions(+), 1 deletion(-) create mode 100644 prismor/runtime/cost.py create mode 100644 prismor/runtime/term.py diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 4c13e77..9e00ecd 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -58,6 +58,7 @@ prismor │ ├─ deps Check project deps vs. threat feed │ ├─ analyze / ingest Run the engine over a JSONL session │ ├─ sessions / session List / show stored sessions +│ ├─ term Full-screen console (agents, sessions, live tail) │ ├─ trail verify · show · checkpoint — signed audit trail │ ├─ attest [verify|coverage] Signed evidence bundle + framework coverage │ ├─ discover Sweep host for ungoverned AI agents (shadow AI) @@ -160,6 +161,7 @@ Full policy model, rule schema, and the default rule list: [Prismor](prismor-run | `prismor attest coverage` | `--json`, `--workspace` | Show which compliance-framework controls the active policy covers (OWASP LLM/Agentic, NIST AI RMF, EU AI Act). | | `prismor discover` | `--json`, `--workspace` | Sweep this host for AI agents and flag any running without Prismor hooks (shadow AI). Host-local, read-only. See [Host discovery](attestation-bundle.md#host-discovery). | | `prismor status --all` | `--days N` | Terminal overview of every registered workspace. See [Dashboard](dashboard.md). | +| `prismor term` | — | Full-screen terminal console: agent → session tree, live event tail, policy precedence, per-session token cost. Falls back to a plain table when stdout isn't a tty. | | `prismor dashboard` | `--port`, `--host`, `--no-open` | Local web dashboard at `http://127.0.0.1:7070` (opens a browser tab). See [Dashboard](dashboard.md). | | `prismor serve` | `--port`, `--host`, `--no-open` | _Deprecated_ alias of `dashboard --no-open` (headless server only). | diff --git a/prismor/runtime/cli.py b/prismor/runtime/cli.py index a260a53..731d561 100644 --- a/prismor/runtime/cli.py +++ b/prismor/runtime/cli.py @@ -955,6 +955,18 @@ def main(argv: Optional[List[str]] = None) -> None: _print_status_overview(workspace) return + # ── term: full-screen terminal console (curses) ───────────────────── + if args.command == "term": + from prismor.runtime.term import run_term + registered = list_registered_workspaces() + if not registered: + sys.stderr.write( + "[prismor] Warning: no registered workspaces found.\n" + " Run 'prismor install-hooks' in a project first to collect data.\n" + ) + run_term() + return + # ── analyze ──────────────────────────────────────────────────────── if args.command == "analyze": # Accept `analyze ` as shorthand for `analyze --input `. @@ -2532,6 +2544,12 @@ def build_parser() -> argparse.ArgumentParser: help="With --all: show activity for the last N days (default: 7)", ) + # ── term ─────────────────────────────────────────────────────────── + subparsers.add_parser( + "term", + help="Full-screen terminal console: agents, sessions, live event tail", + ) + # ── analyze ──────────────────────────────────────────────────────── analyze = subparsers.add_parser("analyze", help="Analyze a session (or current session if no --input)") analyze.add_argument("file", nargs="?", help="Path to JSONL session file (same as --input). If omitted, analyzes most recent session") diff --git a/prismor/runtime/cost.py b/prismor/runtime/cost.py new file mode 100644 index 0000000..83c9647 --- /dev/null +++ b/prismor/runtime/cost.py @@ -0,0 +1,291 @@ +"""Token cost for agent sessions — live prices × real usage. + +Prismor's own event store records *what* an agent did, never how many tokens +it took: there is no usage column and no usage payload anywhere in the events +table. So cost is joined from two outside sources: + + prices aipricing.guru's published feed (``/api/pricing.json``), cached on + disk so the TUI still renders offline. + usage the agent's own transcript. Claude Code writes one JSONL per session + under ``~/.claude/projects//.jsonl`` with a + ``message.usage`` block per assistant turn — and the session ids + there are the same ids Prismor records, so they join directly. + +Only Claude Code keeps transcripts in a known location, so cost is reported +for Claude Code sessions and left unknown (not zero) for every other +framework. See :func:`session_cost`. +""" + +from __future__ import annotations + +import json +import re +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +PRICING_URL = "https://www.aipricing.guru/api/pricing.json" + +# Refetch prices at most this often; the feed updates on publish, not per-minute. +PRICING_TTL_SECONDS = 12 * 3600 +PRICING_TIMEOUT_SECONDS = 6 + +# The feed publishes inputPerM / cachedInputPerM / outputPerM but no cache +# *write* rate, which Anthropic bills above base input. 1.25x is the standard +# 5-minute-TTL multiplier; 1-hour-TTL caching costs 2x, so totals for +# long-lived caches are an underestimate. Surfaced as an estimate in the UI. +CACHE_WRITE_MULTIPLIER = 1.25 + +_CLAUDE_PROJECTS = Path.home() / ".claude" / "projects" + + +# ── Prices ──────────────────────────────────────────────────────────────────── + +def _cache_path() -> Path: + from prismor.runtime.store import prismor_home + return prismor_home() / "pricing-cache.json" + + +def _read_cache() -> Optional[Dict[str, Any]]: + path = _cache_path() + if not path.exists(): + return None + try: + return json.loads(path.read_text(encoding="utf-8")) + except Exception: + return None + + +def _fetch_pricing() -> Dict[str, Any]: + import urllib.request + + request = urllib.request.Request( + PRICING_URL, + headers={"Accept": "application/json", "User-Agent": "prismor-term"}, + ) + with urllib.request.urlopen(request, timeout=PRICING_TIMEOUT_SECONDS) as response: + return json.loads(response.read().decode("utf-8")) + + +def load_pricing(force: bool = False) -> Dict[str, Any]: + """Return ``{models: {id: pricing}, fetched_at, source, error}``. + + Never raises and never blocks the UI for longer than the HTTP timeout: on + any network failure it falls back to the on-disk cache, and only reports + ``source="unavailable"`` when there is no cache either. + """ + cached = _read_cache() + fresh_enough = ( + cached + and not force + and (time.time() - float(cached.get("fetched_at", 0))) < PRICING_TTL_SECONDS + ) + if fresh_enough: + return {**cached, "source": "cache"} + + try: + payload = _fetch_pricing() + except Exception as exc: + if cached: + return {**cached, "source": "stale", "error": str(exc)} + return {"models": {}, "fetched_at": 0, "source": "unavailable", "error": str(exc)} + + models: Dict[str, Dict[str, float]] = {} + for entry in payload.get("models", []): + model_id = str(entry.get("id", "")).lower() + pricing = entry.get("pricing") or {} + if not model_id or not pricing: + continue + models[model_id] = { + "input": float(pricing.get("inputPerM") or 0.0), + "cached": float(pricing.get("cachedInputPerM") or 0.0), + "output": float(pricing.get("outputPerM") or 0.0), + } + + result = { + "models": models, + "fetched_at": time.time(), + "upstream_updated": payload.get("lastUpdated") or payload.get("updated") or "", + "source": "live", + } + try: + _cache_path().write_text(json.dumps(result), encoding="utf-8") + except Exception: + pass + return result + + +def _model_candidates(model: str) -> List[str]: + """Transcript model ids → feed ids. + + Handles the shapes Claude Code actually writes: ``claude-opus-5``, + ``claude-opus-5[1m]`` (context-window suffix), ``claude-opus-4-8`` + (dashed minor) and ``claude-opus-4-20250514`` (dated release). + """ + base = str(model or "").strip().lower() + if not base: + return [] + base = re.sub(r"\[.*?\]$", "", base) + candidates = [base] + undated = re.sub(r"-\d{8}$", "", base) + if undated != base: + candidates.append(undated) + for candidate in list(candidates): + dotted = re.sub(r"(\d)-(\d)$", r"\1.\2", candidate) + if dotted != candidate: + candidates.append(dotted) + return candidates + + +def price_for(model: str, pricing: Dict[str, Any]) -> Optional[Dict[str, float]]: + models = pricing.get("models") or {} + for candidate in _model_candidates(model): + if candidate in models: + return models[candidate] + return None + + +# ── Usage ───────────────────────────────────────────────────────────────────── + +_transcript_index: Dict[str, Path] = {} +_index_built_at: float = 0.0 + + +def find_transcript(session_id: str) -> Optional[Path]: + """Locate a Claude Code transcript by session id. + + Indexes ``~/.claude/projects`` once and rebuilds only when a lookup misses, + so a live session that started after the index was built is still found. + """ + global _transcript_index, _index_built_at + + if not session_id: + return None + hit = _transcript_index.get(session_id) + if hit is not None and hit.exists(): + return hit + + if not _CLAUDE_PROJECTS.is_dir(): + return None + # Rebuild at most once every few seconds — a miss is usually a genuine + # non-Claude-Code session, and rescanning per row would be wasteful. + if time.time() - _index_built_at < 5.0 and hit is None and _transcript_index: + return None + index: Dict[str, Path] = {} + try: + for path in _CLAUDE_PROJECTS.glob("*/*.jsonl"): + index[path.stem] = path + except Exception: + return None + _transcript_index = index + _index_built_at = time.time() + return index.get(session_id) + + +def session_usage(session_id: str) -> Optional[Dict[str, Any]]: + """Sum token usage for one session, or None when there's no transcript.""" + path = find_transcript(session_id) + if path is None: + return None + + totals = {"input": 0, "output": 0, "cache_read": 0, "cache_write": 0} + by_model: Dict[str, Dict[str, int]] = {} + turns = 0 + + try: + with path.open("r", encoding="utf-8", errors="replace") as handle: + for line in handle: + try: + record = json.loads(line) + except Exception: + continue + message = record.get("message") + if not isinstance(message, dict): + continue + usage = message.get("usage") + if not isinstance(usage, dict): + continue + model = str(message.get("model") or "unknown") + bucket = by_model.setdefault( + model, {"input": 0, "output": 0, "cache_read": 0, "cache_write": 0} + ) + pairs = ( + ("input", "input_tokens"), + ("output", "output_tokens"), + ("cache_read", "cache_read_input_tokens"), + ("cache_write", "cache_creation_input_tokens"), + ) + for key, field in pairs: + value = int(usage.get(field) or 0) + bucket[key] += value + totals[key] += value + turns += 1 + except Exception: + return None + + if not turns: + return None + return {"totals": totals, "by_model": by_model, "turns": turns, "path": str(path)} + + +def session_cost(session_id: str, pricing: Dict[str, Any]) -> Dict[str, Any]: + """Cost for one session. + + ``known`` is False when there's no transcript (non-Claude-Code agent, or a + deleted one) — the caller must render that as unknown, never as $0.00. + ``priced`` is False when usage exists but no model in it matched the feed. + """ + usage = session_usage(session_id) + if usage is None: + return {"known": False, "priced": False, "usd": 0.0} + + total = 0.0 + priced_any = False + unpriced: List[str] = [] + + for model, tokens in usage["by_model"].items(): + rate = price_for(model, pricing) + if rate is None: + unpriced.append(model) + continue + priced_any = True + total += ( + tokens["input"] * rate["input"] + + tokens["output"] * rate["output"] + + tokens["cache_read"] * rate["cached"] + + tokens["cache_write"] * rate["input"] * CACHE_WRITE_MULTIPLIER + ) / 1_000_000 + + return { + "known": True, + "priced": priced_any, + "usd": total, + "totals": usage["totals"], + "models": sorted(usage["by_model"].keys()), + "unpriced": unpriced, + "turns": usage["turns"], + } + + +# ── Formatting ──────────────────────────────────────────────────────────────── + +def fmt_usd(amount: float, compact: bool = False) -> str: + if not compact: + return f"${amount:,.2f}" + if amount >= 1000: + return f"${amount / 1000:.1f}k" + if amount >= 10: + return f"${amount:.0f}" + if amount >= 0.01: + return f"${amount:.2f}" + return "$0" + + +def fmt_tokens(count: int) -> str: + if count >= 1_000_000_000: + return f"{count / 1e9:.1f}B" + if count >= 1_000_000: + return f"{count / 1e6:.1f}M" + if count >= 1_000: + return f"{count / 1e3:.1f}k" + return str(count) diff --git a/prismor/runtime/immunity_cli.py b/prismor/runtime/immunity_cli.py index 0027092..577425d 100644 --- a/prismor/runtime/immunity_cli.py +++ b/prismor/runtime/immunity_cli.py @@ -137,7 +137,7 @@ def main(argv: Optional[List[str]] = None) -> None: # owns; any introspected command NOT named here lands in the "More" catch-all, # so a new prismor.runtime.cli subcommand can never silently vanish from help. _HELP_GROUPS = [ - ("Quick start", ["setup", "status", "dashboard", "audit", "update", "pause", "pause-hard", "resume"]), + ("Quick start", ["setup", "status", "term", "dashboard", "audit", "update", "pause", "pause-hard", "resume"]), ("Runtime protection", ["check", "semantic-check", "scan", "deps", "sandbox", "policy"]), ("Sessions & forensics", ["analyze", "ingest", "sessions", "session"]), ("Hooks", ["install-hooks", "uninstall-hooks", "mcp-gateway"]), diff --git a/prismor/runtime/term.py b/prismor/runtime/term.py new file mode 100644 index 0000000..d2582f9 --- /dev/null +++ b/prismor/runtime/term.py @@ -0,0 +1,1254 @@ +"""prismor term — full-screen terminal console for agents, sessions and events. + +The web dashboard (`prismor dashboard`) and this share one data layer: every +number on screen comes from :mod:`prismor.runtime.store`, which already fans +out across every registered workspace DB and re-cloaks secrets on read. This +module is purely a renderer — it never queries SQLite itself. + +Layout:: + + ┌ header: enrollment · winning policy layer · agent count ──────────────┐ + │ Agents ▸ sessions │ selected node detail │ + │ (tree, j/k) ├───────────────────────────────────────────────────┤ + │ │ Events (live tail, scoped to the selection) │ + └ footer: keybinds ─────────────────────────────────────────────────────┘ + +The left pane is a two-level tree: agents expand into their sessions. What the +event tail shows follows the selected node — all agents, one agent, or one +session. + +Degrades to a plain text dump when curses is unusable (no tty, no +windows-curses, terminal too small) so it stays scriptable. +""" + +from __future__ import annotations + +import textwrap +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +# How stale an agent's last_seen can be before it stops counting as live. +_LIVE_WINDOW_SECONDS = 300 +_IDLE_WINDOW_SECONDS = 86400 + +# Seconds between automatic refetches while follow mode is on. +_FOLLOW_INTERVAL = 2.0 +# Poll granularity for getch(). Short so idle-debounced work fires promptly; +# a wakeup that finds no key and nothing dirty costs nothing. +_TICK_MS = 60 +# Navigation redraws immediately from cached rows and only re-queries once the +# selection has been still this long — so holding j/k never queues store calls. +_SETTLE_SECONDS = 0.12 + +# The 24h aggregate is expensive (~900ms); reuse it this long before refetching. +_STATS_TTL_SECONDS = 30.0 + +# Sessions priced per idle tick, so pricing a wide tree never blocks a keypress. +_COST_BATCH = 6 + +# Session paging. ``get_sessions_page`` has no agent filter, so an agent's +# sessions are found by walking pages and filtering client-side. Both caps are +# surfaced in the UI rather than silently truncating. +_SESSION_PAGE_SIZE = 200 +_SESSION_MAX_PAGES = 5 +_SESSION_MAX_PER_AGENT = 50 + +# ``get_session_scoped_detail`` returns at most this many events (store-side). +_SESSION_EVENT_CAP = 60 + +# Agent frameworks whose session ids are resumable Claude Code conversations. +_CLAUDE_FRAMEWORKS = {"claude", "claude-code"} + +# Session sort orders. Sessions are always *fetched* newest-first (so a +# truncated load holds the most recent ones); these re-order what was loaded. +_SORTS = ("recent", "risk", "findings", "cost") +_SORT_LABELS = {"recent": "last run", "risk": "risk", + "findings": "findings", "cost": "cost"} + + +# ── Data ────────────────────────────────────────────────────────────────────── + +def _age_seconds(ts: str) -> Optional[int]: + """Seconds since an ISO timestamp, or None if it can't be parsed.""" + if not ts: + return None + try: + from datetime import datetime, timezone + dt = datetime.fromisoformat(str(ts).replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return int((datetime.now(timezone.utc) - dt).total_seconds()) + except Exception: + return None + + +def _agent_status(agent: Dict[str, Any]) -> Tuple[str, str]: + """Map last_seen → (label, color key). + + There is no liveness signal in the store — sessions record timestamps, not + a heartbeat — so "Active" here means *recently seen*, not *running now*. + """ + age = _age_seconds(agent.get("last_seen", "")) + if age is None: + return "Unknown", "dim" + if age <= _LIVE_WINDOW_SECONDS: + return "Active", "green" + if age <= _IDLE_WINDOW_SECONDS: + return "Idle", "yellow" + return "Dormant", "dim" + + +def _relative(ts: str) -> str: + from prismor.runtime.store import _relative_time_store + return _relative_time_store(ts) if ts else "never" + + +def fmt_usd(amount: float, compact: bool = False) -> str: + from prismor.runtime.cost import fmt_usd as _fmt + return _fmt(amount, compact) + + +def _risk_color(score: int) -> str: + if score >= 70: + return "red" + if score >= 40: + return "yellow" + if score > 0: + return "white" + return "dim" + + +def _compact_age(relative: str) -> str: + """'2 days ago' / '1d ago' → '1d'. Keeps the session row narrow.""" + return str(relative or "").replace(" ago", "").replace(" ", "")[:4] or "—" + + +def _sort_sessions( + items: List[Dict[str, Any]], order: str, costs: Optional[Dict[str, Any]] = None +) -> List[Dict[str, Any]]: + """Re-order loaded sessions. ``updatedAtAbs`` is lexicographically sortable.""" + costs = costs or {} + if order == "risk": + key = lambda s: (-(s.get("riskScore") or 0), s.get("updatedAtAbs") or "") # noqa: E731 + elif order == "findings": + key = lambda s: (-(s.get("findingsCount") or 0), s.get("updatedAtAbs") or "") # noqa: E731 + elif order == "cost": + # Sessions with no transcript have unknown cost, not zero — sort them + # last rather than letting them rank as "cheapest". + key = lambda s: ( # noqa: E731 + -((costs.get(s.get("sessionId")) or {}).get("usd") or 0.0), + s.get("updatedAtAbs") or "", + ) + else: + return sorted(items, key=lambda s: s.get("updatedAtAbs") or "", reverse=True) + return sorted(items, key=key) + + +def _resume_blocker(session: Dict[str, Any], agent: Dict[str, Any]) -> Optional[str]: + """Why this session can't be resumed in Claude Code, or None if it can.""" + import shutil + + framework = str(agent.get("framework") or "").lower() + if framework not in _CLAUDE_FRAMEWORKS: + return f"resume is Claude Code only — this session ran under '{framework or 'unknown'}'" + if not shutil.which("claude"): + return "the `claude` CLI is not on PATH" + workspace = str(session.get("workspace") or "") + if not workspace or not Path(workspace).is_dir(): + return f"workspace no longer exists: {workspace or '(unknown)'}" + return None + + +def _agent_event_key(agent: Optional[Dict[str, Any]]) -> str: + """Which value to pass to ``get_events_page(agent=...)``. + + Events carry ``sessions.agent`` (the framework id), while the agents + overview keys on ``agent_name`` and only falls back to ``agent``. Filter on + the framework so labelled agent instances still match their own events. + """ + if not agent: + return "" + return str(agent.get("framework") or agent.get("name") or "") + + +def _safe(fn, default): + try: + return fn() + except Exception: + return default + + +def _fetch_base() -> Dict[str, Any]: + """Agents, policy chain and enrollment — all sub-5ms queries. + + Deliberately excludes ``get_aggregate_stats``, which costs ~900ms (it scans + every session and event in the window) and feeds only the four KPI numbers + on the "All agents" panel. That is fetched on demand by :func:`_fetch_stats` + so startup and refresh stay instant. + """ + from prismor.runtime import store + return { + "agents": _safe(store.get_agents_overview, []), + "policy": _safe(store.get_policy_precedence, {"winner": "default", "chain": []}), + "enrollment": _safe(store.get_enrollment, None), + } + + +def _fetch_stats() -> Dict[str, Any]: + """The expensive 24h aggregate. Only called when its panel is on screen.""" + from prismor.runtime import store + return _safe(lambda: store.get_aggregate_stats(24), {}) + + +def _fetch_sessions_for_agent(agent: Dict[str, Any]) -> Dict[str, Any]: + """Sessions belonging to one agent. + + ``get_sessions_page`` can't filter by agent, so walk pages (newest first) + and match client-side on the agent label, falling back to the framework id + for unlabelled sessions. Bounded by ``_SESSION_MAX_PAGES`` — the returned + ``truncated`` flag tells the UI to say so rather than imply completeness. + """ + from prismor.runtime import store + + name = str(agent.get("name", "")) + framework = str(agent.get("framework", "")) + found: List[Dict[str, Any]] = [] + scanned = 0 + total = 0 + truncated = False + + for page in range(1, _SESSION_MAX_PAGES + 1): + result = _safe( + lambda: store.get_sessions_page( + page=page, limit=_SESSION_PAGE_SIZE, sort="updatedAt", direction="desc" + ), + {"items": [], "total": 0, "pages": 1}, + ) + items = result.get("items", []) + total = result.get("total", 0) + scanned += len(items) + for s in items: + label = s.get("agentName") or s.get("agent") or "" + if label == name or (not s.get("agentName") and s.get("agent") == framework): + found.append(s) + if len(found) >= _SESSION_MAX_PER_AGENT: + truncated = True + break + if truncated or page >= result.get("pages", 1): + break + else: + truncated = True + + if scanned < total and len(found) < _SESSION_MAX_PER_AGENT: + truncated = True + + return {"items": found, "truncated": truncated, "scanned": scanned, "total": total} + + +def _normalize_session_event(ev: Dict[str, Any], session: Dict[str, Any]) -> Dict[str, Any]: + """Give a scoped-detail event the same shape as a ``get_events_page`` item. + + The two store calls return overlapping but not identical dicts — scoped + detail omits agent/session/workspace since they're implied by the query. + """ + out = dict(ev) + out.setdefault("agent", session.get("agent", "")) + out.setdefault("actionType", ev.get("type", "")) + out.setdefault("sessionId", session.get("sessionId", "")) + out.setdefault("workspace", session.get("workspace", "")) + return out + + +def _fetch_events( + node: Dict[str, Any], verdict: str, page: int = 1, limit: int = 20 +) -> Dict[str, Any]: + """One screenful of events for the selected tree node. + + ``limit`` is the number of rows actually visible, not a fixed 200. That + matters: ``get_events_page`` scales its internal scan with ``page * limit``, + so asking for a screenful costs ~8ms where asking for 200 rows costs + 35-70ms — and this runs on every navigation keypress. + + A session's events come from ``get_session_scoped_detail`` (the only + session-scoped read the store offers). It returns at most 60 and does its + own verdict-free query, so filtering and paging happen here. + """ + from prismor.runtime import store + + kind = node.get("kind") + limit = max(1, limit) + page = max(1, page) + + if kind == "session": + session = node["session"] + detail = _safe( + lambda: store.get_session_scoped_detail( + Path(session.get("workspace") or "."), session.get("sessionId", "") + ), + {"recent_events": [], "paused": False, "scoped": {}}, + ) + raw = detail.get("recent_events", []) + items = [_normalize_session_event(e, session) for e in raw] + if verdict == "blocked": + items = [e for e in items if e.get("verdict") == "blocked"] + elif verdict == "allowed": + items = [e for e in items if e.get("verdict") != "blocked"] + total = len(items) + pages = max(1, (total + limit - 1) // limit) + page = min(page, pages) + return { + "items": items[(page - 1) * limit: page * limit], + "total": total, + "page": page, + "pages": pages, + "exact": True, # sliced locally from a complete list + "has_next": page < pages, + "capped": len(raw) >= _SESSION_EVENT_CAP, + "session_detail": detail, + } + + agent_key = _agent_event_key(node.get("agent")) if kind == "agent" else "" + result = _safe( + lambda: store.get_events_page( + page=page, limit=limit, verdict=verdict, agent=agent_key + ), + {"items": [], "total": 0, "page": 1, "pages": 1}, + ) + # ``get_events_page`` derives total/pages from an internal window that grows + # with page depth, so both climb as you page deeper (200 → 276 → 621...). + # They are a floor on what exists, never a true count — reported as "N+" + # rather than dressed up as a fixed page count. + items = result.get("items", []) + return { + "items": items, + "total": result.get("total", 0), + "page": result.get("page", 1), + "pages": result.get("pages", 1), + "exact": False, + "has_next": len(items) >= limit, + "capped": False, + "session_detail": None, + } + + +# ── Plain-text fallback ─────────────────────────────────────────────────────── + +def _render_plain() -> None: + """Non-interactive dump — used when curses can't run (pipes, CI, no tty).""" + from prismor.runtime import store + + base = _fetch_base() + enrollment = base["enrollment"] + org = enrollment.get("org_id") if enrollment else None + print(f"prismor term — {org or 'local (not enrolled)'}") + print(f"policy: {base['policy'].get('winner', 'default')}") + print() + agents = base["agents"] + if not agents: + print("No agents recorded yet. Run `prismor install-hooks` in a project first.") + return + print("AGENT FRAMEWORK STATUS CALLS FLAGGED LAST SEEN") + for a in agents: + label, _ = _agent_status(a) + print( + f"{str(a.get('name', ''))[:20]:<20} {str(a.get('framework', ''))[:14]:<14} " + f"{label:<9} {a.get('total_calls', 0):>5} {a.get('blocked_calls', 0):>7} " + f"{_relative(a.get('last_seen', ''))}" + ) + print() + sessions = _safe(lambda: store.get_sessions_page(page=1, limit=10), {"items": [], "total": 0}) + print(f"{sessions.get('total', 0)} sessions — 10 most recent:") + for s in sessions.get("items", []): + print( + f" {str(s.get('sessionId', ''))[:16]:<16} {str(s.get('agentName', ''))[:16]:<16} " + f"risk {s.get('riskScore', 0):>3} {s.get('findingsCount', 0):>2} findings " + f"{s.get('updatedAt', '')}" + ) + print() + print("Run `prismor term` in a tty for the live tree view.") + + +# ── Curses app ──────────────────────────────────────────────────────────────── + +def _run_curses() -> bool: + try: + import curses + except ImportError: + return False # stock Windows Python without windows-curses + + state: Dict[str, Any] = { + "sel": 0, # index into the flattened tree + "top": 0, # tree scroll offset + "expanded": set(), # agent names currently expanded + "sessions": {}, # agent name → _fetch_sessions_for_agent() result + "ev_sel": 0, + "ev_page": 1, + "ev_rows": 20, # visible event rows — becomes the fetch limit + "dirty": False, # selection changed; re-query once it settles + "redraw": True, # screen is stale; repaint on the next loop pass + "last_input_at": 0.0, + "stats": None, # lazily fetched 24h aggregate + "stats_at": 0.0, + "want_stats": False, # request pending; the idle tick does the work + "focus": "tree", # tree | events + "follow": True, + "verdict": "", # "" | blocked | allowed + "sort": "recent", # session order: recent | risk | findings | cost + "pricing": None, # live token prices (see prismor.runtime.cost) + "costs": {}, # session id → session_cost() result + "mode": "main", # main | detail | policy | confirm + "confirm": None, # pending {"prompt", "apply"} + "flash": "", # transient status message + "base": None, + "events": None, + } + + def draw(stdscr) -> None: + curses.curs_set(0) + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_RED, -1) + curses.init_pair(2, curses.COLOR_YELLOW, -1) + curses.init_pair(3, curses.COLOR_WHITE, -1) + curses.init_pair(4, curses.COLOR_CYAN, -1) + curses.init_pair(5, curses.COLOR_GREEN, -1) + curses.init_pair(6, curses.COLOR_MAGENTA, -1) + stdscr.keypad(True) + stdscr.timeout(_TICK_MS) + + colors = { + "red": curses.color_pair(1), + "yellow": curses.color_pair(2), + "white": curses.color_pair(3), + "cyan": curses.color_pair(4), + "green": curses.color_pair(5), + "magenta": curses.color_pair(6), + "dim": curses.A_DIM, + } + + def put(y: int, x: int, text: str, attr: int = 0) -> None: + """addstr that never raises at the screen edge.""" + h, w = stdscr.getmaxyx() + if y < 0 or y >= h or x < 0 or x >= w: + return + try: + stdscr.addstr(y, x, str(text)[: max(0, w - x - 1)], attr) + except curses.error: + pass + + # ── tree model ── + + def build_tree() -> List[Dict[str, Any]]: + """Flatten agents (+ expanded sessions) into selectable rows.""" + rows: List[Dict[str, Any]] = [{"kind": "all", "depth": 0}] + for agent in state["base"]["agents"]: + name = str(agent.get("name", "")) + rows.append({"kind": "agent", "agent": agent, "name": name, "depth": 0}) + if name not in state["expanded"]: + continue + bundle = state["sessions"].get(name) + if bundle is None: + rows.append({"kind": "loading", "depth": 1}) + continue + for session in _sort_sessions(bundle["items"], state["sort"], state["costs"]): + rows.append({ + "kind": "session", "session": session, + "agent": agent, "depth": 1, + }) + if not bundle["items"]: + rows.append({"kind": "empty", "depth": 1}) + elif bundle["truncated"]: + rows.append({"kind": "note", "depth": 1, + "text": f"… showing first {len(bundle['items'])}"}) + return rows + + def current_node() -> Dict[str, Any]: + rows = build_tree() + if not rows: + return {"kind": "all"} + return rows[min(state["sel"], len(rows) - 1)] + + def refetch_events() -> None: + state["events"] = _fetch_events( + current_node(), state["verdict"], state["ev_page"], state["ev_rows"] + ) + state["events"]["fetched_at"] = time.monotonic() + state["ev_page"] = state["events"].get("page", 1) + state["ev_sel"] = min(state["ev_sel"], max(0, len(state["events"]["items"]) - 1)) + state["dirty"] = False + + def mark_dirty() -> None: + """Selection changed: redraw now, re-query once the user settles.""" + state["dirty"] = True + state["redraw"] = True + state["last_input_at"] = time.monotonic() + + def settled() -> bool: + return time.monotonic() - state["last_input_at"] >= _SETTLE_SECONDS + + def price_some(sessions: List[Dict[str, Any]], budget: int = _COST_BATCH) -> bool: + """Price up to ``budget`` unpriced sessions. True if any work was done. + + Each transcript is ~5ms, so a small batch per idle tick keeps the UI + responsive even when an agent has 50 sessions to price. + """ + from prismor.runtime import cost as cost_mod + pricing = state["pricing"] or {"models": {}} + done = 0 + for session in sessions: + if done >= budget: + break + sid = session.get("sessionId") or "" + if not sid or sid in state["costs"]: + continue + state["costs"][sid] = _safe( + lambda: cost_mod.session_cost(sid, pricing), + {"known": False, "priced": False, "usd": 0.0}, + ) + done += 1 + return done > 0 + + def pending_cost_sessions() -> List[Dict[str, Any]]: + """Sessions on screen that still need pricing (visible rows first).""" + pending = [] + for row in build_tree(): + if row["kind"] != "session": + continue + sid = row["session"].get("sessionId") + if sid and sid not in state["costs"]: + pending.append(row["session"]) + return pending + + def get_stats() -> Optional[Dict[str, Any]]: + """The 24h aggregate — never computed on the draw path. + + This is the one ~900ms query in the app. Requesting it here only + sets a flag; the idle tick does the work, so selecting "All agents" + paints instantly and fills in a moment later. + """ + now = time.monotonic() + if state["stats"] is None or now - state["stats_at"] >= _STATS_TTL_SECONDS: + state["want_stats"] = True + return state["stats"] + + def load_sessions(agent: Dict[str, Any]) -> None: + name = str(agent.get("name", "")) + if name in state["sessions"]: + return + state["flash"] = f"loading sessions for {name}…" + draw_main() + state["sessions"][name] = _fetch_sessions_for_agent(agent) + # Price only the first screenful now; the rest fills in on idle. + price_some(state["sessions"][name]["items"]) + state["flash"] = "" + + from prismor.runtime import cost as cost_mod + state["pricing"] = _safe(cost_mod.load_pricing, + {"models": {}, "source": "unavailable"}) + state["base"] = _fetch_base() + state["events"] = _fetch_events({"kind": "all"}, "", 1, state["ev_rows"]) + state["events"]["fetched_at"] = time.monotonic() + + # ── panes ── + + def draw_header() -> None: + h, w = stdscr.getmaxyx() + base = state["base"] + org = (base["enrollment"] or {}).get("org_id") or "local" + policy = base["policy"].get("winner", "default") + follow = "FOLLOW" if state["follow"] else "PAUSED" + prices = {"live": "prices live", "cache": "prices cached", + "stale": "prices STALE", "unavailable": "prices n/a"}.get( + (state["pricing"] or {}).get("source", ""), "prices n/a") + bar = ( + f" >_ PRISMOR TERM │ Org: {org} │ Policy: {policy} " + f"│ Agents: {len(base['agents'])} │ {prices} │ {follow} " + ) + put(0, 0, bar.ljust(w - 1), curses.A_REVERSE | curses.A_BOLD) + + def draw_footer(hint: str) -> None: + h, w = stdscr.getmaxyx() + if state["flash"]: + put(h - 1, 0, f" {state['flash']} ".ljust(w - 1)[: w - 1], + colors["yellow"] | curses.A_BOLD) + return + put(h - 1, 0, hint.ljust(w - 1)[: w - 1], colors["cyan"]) + + def draw_tree(lw: int, top_y: int, bot_y: int) -> None: + rows = build_tree() + active = state["focus"] == "tree" + put(top_y, 0, + f" Agents ▸ sessions ↓{_SORT_LABELS[state['sort']]}" + (" ◂" if active else ""), + curses.A_BOLD | (colors["cyan"] if active else 0)) + put(top_y + 1, 0, "─" * (lw - 1), colors["dim"]) + + visible = max(1, bot_y - (top_y + 2)) + state["sel"] = max(0, min(state["sel"], len(rows) - 1)) + if state["sel"] < state["top"]: + state["top"] = state["sel"] + if state["sel"] >= state["top"] + visible: + state["top"] = state["sel"] - visible + 1 + + for i, row in enumerate(rows[state["top"]: state["top"] + visible]): + idx = state["top"] + i + kind = row["kind"] + color = colors["white"] + if kind == "all": + line = " All agents" + elif kind == "agent": + agent = row["agent"] + _, ckey = _agent_status(agent) + color = colors[ckey] + caret = "▾" if row["name"] in state["expanded"] else "▸" + dot = {"Active": "●", "Idle": "◐"}.get(_agent_status(agent)[0], "○") + counts = f"{agent.get('total_calls', 0)}/{agent.get('blocked_calls', 0)}" + head = f" {caret} {dot} {row['name'][: max(4, lw - 16)]}" + line = head + " " * max(1, lw - 2 - len(head) - len(counts)) + counts + elif kind == "session": + session = row["session"] + risk = session.get("riskScore", 0) or 0 + color = colors[_risk_color(risk)] + sid = str(session.get("sessionId", ""))[:10] + sid_full = session.get("sessionId") + if sid_full not in state["costs"]: + spend = " ·" # not priced yet (fills in on idle) + else: + money = state["costs"][sid_full] or {} + spend = (fmt_usd(money["usd"], compact=True) + if money.get("priced") else " —") + tail = (f"r{risk} {session.get('findingsCount', 0)}f " + f"{spend} {_compact_age(session.get('updatedAt', ''))}") + head = f" {sid}" + line = head + " " * max(1, lw - 2 - len(head) - len(tail)) + tail + elif kind == "loading": + line, color = " loading…", colors["dim"] + elif kind == "empty": + line, color = " no sessions", colors["dim"] + else: + line, color = f" {row.get('text', '')}", colors["dim"] + + attr = color | (curses.A_REVERSE if active and idx == state["sel"] else 0) + put(top_y + 2 + i, 0, line.ljust(lw - 1)[: lw - 1], attr) + + def draw_detail(x0: int, y0: int, width: int) -> int: + node = current_node() + base = state["base"] + put(y0, x0, " Detail", curses.A_BOLD) + put(y0 + 1, x0, "─" * (width - 1), colors["dim"]) + y = y0 + 2 + + if node["kind"] == "session": + session = node["session"] + detail = (state["events"] or {}).get("session_detail") or {} + scoped = detail.get("scoped") or {} + paused = bool(detail.get("paused")) + risk = session.get("riskScore", 0) or 0 + scope_bits = [] + for field, label in ( + ("allowed_tools", "tools"), ("deny_tools", "deny-tools"), + ("allowed_paths", "paths"), ("deny_network", "no-net"), + ): + if scoped.get(field): + scope_bits.append(label) + fields = [ + ("Session", str(session.get("sessionId", ""))), + ("Agent", f"{session.get('agentName', '')} ({session.get('agent', '')})"), + ("Risk", f"{risk}/100 · {session.get('findingsCount', 0)} findings"), + ("Workspace", str(session.get("workspaceName") or session.get("workspace", ""))), + ("Started", f"{session.get('startedAt', '')} · updated {session.get('updatedAt', '')}"), + ("Immunity", "PAUSED" if paused else "active"), + ] + if scope_bits: + fields.append(("Scope", ", ".join(scope_bits))) + + from prismor.runtime.cost import fmt_tokens + money = state["costs"].get(session.get("sessionId")) or {} + if money.get("priced"): + totals = money.get("totals", {}) + fields.append(( + "Cost", + f"{fmt_usd(money['usd'])} est · {money.get('turns', 0)} turns " + f"· {', '.join(money.get('models', []))}", + )) + fields.append(( + "Tokens", + f"in {fmt_tokens(totals.get('input', 0))} · " + f"out {fmt_tokens(totals.get('output', 0))} · " + f"cache r{fmt_tokens(totals.get('cache_read', 0))}" + f"/w{fmt_tokens(totals.get('cache_write', 0))}", + )) + elif money.get("known"): + fields.append(("Cost", "usage found, but no live price for its model")) + else: + fields.append(("Cost", "unknown — no Claude Code transcript")) + elif node["kind"] == "agent": + agent = node["agent"] + label, _ = _agent_status(agent) + total = agent.get("total_calls", 0) or 0 + blocked = agent.get("blocked_calls", 0) or 0 + rate = f"{(blocked / total * 100):.0f}%" if total else "—" + fields = [ + ("Name", str(agent.get("name", "unknown"))), + ("Framework", str(agent.get("framework") or "—")), + ("Status", f"{label} (last seen {_relative(agent.get('last_seen', ''))})"), + ("Sessions", f"{total} total · {blocked} flagged · {rate} flag rate"), + ] + bundle = state["sessions"].get(str(agent.get("name", ""))) + if bundle: + priced = [ + state["costs"].get(s.get("sessionId")) or {} + for s in bundle["items"] + ] + known = [c for c in priced if c.get("priced")] + if known: + spend = sum(c["usd"] for c in known) + fields.append(( + "Cost", + f"{fmt_usd(spend)} est across {len(known)} of " + f"{len(bundle['items'])} loaded sessions", + )) + else: + # Fetched lazily — this is the only ~900ms query in the app, so + # it runs when its panel is first shown, not at startup. + stats = get_stats() + if stats is None: + fields = [ + ("Scope", "All agents (no filter)"), + ("Sessions", "computing 24h totals…"), + ("Tool calls", "…"), + ("Prevented", "…"), + ] + else: + kpis = stats.get("kpis", {}) + fields = [ + ("Scope", "All agents (no filter)"), + ("Sessions", f"{kpis.get('activeSessions', 0)} active in 24h"), + ("Tool calls", f"{kpis.get('toolCallsInspected24h', 0)} inspected in 24h"), + ("Prevented", + f"{kpis.get('dangerousCommandsPrevented24h', 0)} dangerous in 24h"), + ] + + for name, value in fields: + put(y, x0 + 1, f"{name:<11}", colors["dim"]) + attr = 0 + if name == "Status" and "Active" in str(value): + attr = colors["green"] + elif name == "Immunity": + attr = colors["yellow"] | curses.A_BOLD if value == "PAUSED" else colors["green"] + elif name == "Risk": + attr = colors[_risk_color(int(str(value).split("/")[0] or 0))] + put(y, x0 + 13, str(value), attr) + y += 1 + return y + 1 + + def draw_events(x0: int, y0: int, bot_y: int, width: int) -> None: + bundle = state["events"] or {"items": [], "capped": False} + events = bundle["items"] + node = current_node() + active = state["focus"] == "events" + scope = {"session": "session", "agent": "agent"}.get(node["kind"], "all agents") + cap = " · store caps at 60" if bundle.get("capped") else "" + total = bundle.get("total", len(events)) + if bundle.get("exact"): + where = f"page {bundle.get('page', 1)}/{bundle.get('pages', 1)} of {total}" + else: + # total is a growing floor, not a count — say so with "+". + where = f"page {bundle.get('page', 1)} · {len(events)} of {total}+" + head = (f" Events [{where} · {scope} · " + f"filter: {state['verdict'] or 'all'}{cap}]" + (" ◂" if active else "")) + put(y0, x0, head, curses.A_BOLD | (colors["cyan"] if active else 0)) + put(y0 + 1, x0, "─" * (width - 1), colors["dim"]) + + # The visible row count *is* the page size — recorded here so the + # next fetch asks for exactly one screenful. + visible = max(1, bot_y - (y0 + 2)) + if visible != state["ev_rows"]: + state["ev_rows"] = visible + mark_dirty() + + if not events: + put(y0 + 2, x0 + 1, "No events for this selection.", colors["dim"]) + return + + for i, ev in enumerate(events[:visible]): + idx = i + blocked = ev.get("verdict") == "blocked" + sev = str(ev.get("severity", "low")).lower() + if blocked: + color = colors["red"] if sev in ("critical", "high") else colors["yellow"] + else: + color = colors["dim"] + ts = str(ev.get("tsAbs", ""))[11:19] or str(ev.get("ts", ""))[:8] + agent = str(ev.get("agent", ""))[:10] + verdict = "BLOCK" if blocked else "allow" + line = f" {ts:<8} {agent:<10} {verdict:<5} {ev.get('action', '')}" + attr = color | (curses.A_REVERSE if active and idx == state["ev_sel"] else 0) + put(y0 + 2 + i, x0, line.ljust(width - 1)[: width - 1], attr) + + def draw_main() -> None: + stdscr.erase() + h, w = stdscr.getmaxyx() + lw = max(26, min(36, w // 3)) + draw_header() + draw_tree(lw, 1, h - 1) + for y in range(1, h - 1): + put(y, lw - 1, "│", colors["dim"]) + x0 = lw + 1 + width = w - x0 + ev_y = draw_detail(x0, 1, width) + draw_events(x0, ev_y, h - 1, width) + node = current_node() + session_hint = " [P] Pause [R] Resume in claude" if node["kind"] == "session" else "" + draw_footer( + " [j/k] Move [→/←] Expand [Tab] Pane [[/]] Page [Enter] Detail " + f"[s] Sort [f] Follow [v] Verdict [p] Policy{session_hint} [q] Quit " + ) + stdscr.refresh() + + def draw_event_detail() -> None: + stdscr.erase() + h, w = stdscr.getmaxyx() + events = (state["events"] or {}).get("items", []) + if not events: + state["mode"] = "main" + return + ev = events[min(state["ev_sel"], len(events) - 1)] + policy = ev.get("policy", {}) or {} + blocked = ev.get("verdict") == "blocked" + put(0, 0, f" Event {state['ev_sel'] + 1}/{len(events)} ".ljust(w - 1), + curses.A_REVERSE | curses.A_BOLD) + y = 2 + fields = [ + ("verdict", "BLOCKED" if blocked else "allowed"), + ("time", ev.get("tsAbs", "")), + ("agent", ev.get("agent", "")), + ("type", ev.get("actionType", "")), + ("tool", ev.get("toolTag", "") or "—"), + ("session", ev.get("sessionId", "")), + ("workspace", ev.get("workspace", "")), + ] + if blocked or policy.get("ruleId"): + fields += [ + ("rule", policy.get("ruleId", "") or "—"), + ("category", policy.get("category", "") or "—"), + ("action", policy.get("action", "") or "—"), + ("source", policy.get("source", "") or "—"), + ] + for label, value in fields: + if y >= h - 2: + break + put(y, 1, f"{label}:", colors["cyan"]) + attr = colors["red"] if label == "verdict" and blocked else 0 + put(y, 13, str(value), attr) + y += 1 + y += 1 + if policy.get("title") and y < h - 2: + for line in textwrap.wrap(str(policy["title"]), max(w - 3, 10)): + if y >= h - 2: + break + put(y, 1, line, curses.A_BOLD) + y += 1 + y += 1 + put(y, 1, "command / detail:", colors["cyan"]) + y += 1 + for line in textwrap.wrap(str(ev.get("action", "")), max(w - 5, 10)): + if y >= h - 2: + break + put(y, 3, line) + y += 1 + evidence = policy.get("evidence") + if evidence and y < h - 3: + y += 1 + put(y, 1, "evidence:", colors["cyan"]) + y += 1 + for line in textwrap.wrap(str(evidence), max(w - 5, 10)): + if y >= h - 2: + break + put(y, 3, line, colors["dim"]) + y += 1 + draw_footer(" [j/k] Prev/Next event · [Esc/b] Back · [q] Quit ") + stdscr.refresh() + + def draw_policy() -> None: + stdscr.erase() + h, w = stdscr.getmaxyx() + chain = state["base"]["policy"].get("chain", []) + winner = state["base"]["policy"].get("winner", "default") + put(0, 0, f" Policy precedence — winner: {winner} ".ljust(w - 1), + curses.A_REVERSE | curses.A_BOLD) + y = 2 + for layer in chain: + if y >= h - 2: + break + mark = "▶" if layer.get("winning") else " " + exists = "active" if layer.get("exists") else "not set" + attr = curses.A_BOLD | colors["green"] if layer.get("winning") else colors["dim"] + put(y, 1, f"{mark} {str(layer.get('label', '')):<20} {exists:<9} " + f"mode={layer.get('mode', '')}", attr) + y += 1 + for line in textwrap.wrap(str(layer.get("summary") or ""), max(w - 8, 10))[:2]: + if y >= h - 2: + break + put(y, 5, line, colors["dim"]) + y += 1 + if layer.get("path") and y < h - 2: + put(y, 5, str(layer["path"]), colors["dim"]) + y += 1 + y += 1 + draw_footer(" [Esc/p] Back · [q] Quit ") + stdscr.refresh() + + def draw_confirm() -> None: + h, w = stdscr.getmaxyx() + pending = state["confirm"] or {} + lines = pending.get("prompt", []) + box_w = min(w - 6, max(40, max((len(l) for l in lines), default=40) + 6)) + box_h = len(lines) + 4 + y0 = max(1, (h - box_h) // 2) + x0 = max(1, (w - box_w) // 2) + # Blank the interior first — this draws over the live main view, + # and without it the event tail bleeds through the box. + for i in range(1, box_h - 1): + put(y0 + i, x0, " " * box_w) + + def edges(row: int) -> None: + put(row, x0, "│", colors["yellow"]) + put(row, x0 + box_w - 1, "│", colors["yellow"]) + + put(y0, x0, "╭" + "─" * (box_w - 2) + "╮", colors["yellow"]) + for i, line in enumerate(lines): + edges(y0 + 1 + i) + put(y0 + 1 + i, x0 + 2, line, curses.A_BOLD if i == 0 else 0) + edges(y0 + box_h - 3) + edges(y0 + box_h - 2) + put(y0 + box_h - 2, x0 + 2, "[y] confirm [n] cancel", colors["cyan"]) + put(y0 + box_h - 1, x0, "╰" + "─" * (box_w - 2) + "╯", colors["yellow"]) + stdscr.refresh() + + def ask_pause_toggle() -> None: + """Queue a pause/resume confirm for the selected session.""" + node = current_node() + if node["kind"] != "session": + return + session = node["session"] + detail = (state["events"] or {}).get("session_detail") or {} + paused = bool(detail.get("paused")) + action = "resume" if paused else "pause" + sid = str(session.get("sessionId", "")) + + def apply() -> str: + from prismor.runtime import store + result = store.update_session_control( + Path(session.get("workspace") or "."), sid, action + ) + if not result.get("ok"): + return f"failed: {result.get('error', 'unknown error')}" + refetch_events() + return f"session {sid[:12]} — immunity {action}d" + + state["confirm"] = { + "prompt": [ + f"{action.capitalize()} immunity for session {sid[:12]}…?", + "", + ("This stops screening for that session only." + if action == "pause" else + "This re-enables screening for that session."), + ], + "apply": apply, + } + state["mode"] = "confirm" + + def launch_claude(session: Dict[str, Any]) -> str: + """Hand the terminal to `claude --resume `, then take it back. + + curses owns the tty, so the child would render into a broken + terminal without ``endwin()`` first — and the tree would come back + as garbage without ``reset_prog_mode()`` after. Session ids from + the Claude Code hooks *are* Claude Code conversation ids, and + --resume is project-scoped, so it runs in the session's workspace. + """ + import subprocess + + sid = str(session.get("sessionId", "")) + workspace = str(session.get("workspace") or "") + + curses.def_prog_mode() + curses.endwin() + try: + result = subprocess.call(["claude", "--resume", sid], cwd=workspace) + except Exception as exc: + result = -1 + error = str(exc) + else: + error = "" + finally: + curses.reset_prog_mode() + stdscr.clear() + stdscr.refresh() + + if error: + return f"could not launch claude: {error}" + if result != 0: + return f"claude exited with status {result}" + # The conversation may have added events while we were away. + state["base"] = _fetch_base() + refetch_events() + return f"returned from claude session {sid[:12]}" + + def ask_resume() -> None: + """Queue a confirm for resuming the selected session in Claude Code.""" + node = current_node() + if node["kind"] != "session": + state["flash"] = "select a session first" + return + session, agent = node["session"], node["agent"] + blocker = _resume_blocker(session, agent) + if blocker: + state["flash"] = blocker + return + sid = str(session.get("sessionId", "")) + workspace = str(session.get("workspace") or "") + home = str(Path.home()) + state["confirm"] = { + "prompt": [ + f"Resume session {sid[:12]}… in Claude Code?", + "", + f"cwd: {workspace.replace(home, '~')}", + "prismor term returns when you exit claude.", + ], + "apply": lambda: launch_claude(session), + } + state["mode"] = "confirm" + + # ── event loop ── + + while True: + h, w = stdscr.getmaxyx() + if h < 10 or w < 50: + stdscr.erase() + put(0, 0, "Terminal too small — need at least 50x10.") + stdscr.refresh() + if stdscr.getch() in (ord("q"), 27): + return + continue + + # Only repaint when something actually changed. Without this the + # whole screen redraws every tick (~16x/sec) while idle, which + # burns CPU and makes modals flicker as they are erased and + # re-stacked on top of the main view each frame. + if state["redraw"]: + if state["mode"] == "detail": + draw_event_detail() + elif state["mode"] == "policy": + draw_policy() + elif state["mode"] == "confirm": + draw_main() + draw_confirm() + else: + draw_main() + state["redraw"] = False + + key = stdscr.getch() + + if key == -1: # idle tick — do deferred work only when input settled + if state["mode"] != "main" or not settled(): + continue + if state["dirty"]: + refetch_events() + state["redraw"] = True + continue + if state["follow"]: + if time.monotonic() - state["events"]["fetched_at"] >= _FOLLOW_INTERVAL: + refetch_events() + state["redraw"] = True + continue + if state["want_stats"]: + state["want_stats"] = False + state["stats"] = _fetch_stats() + state["stats_at"] = time.monotonic() + state["redraw"] = True + continue + # Nothing urgent: spend the idle slice pricing sessions. + pending = pending_cost_sessions() + if pending and price_some(pending): + state["redraw"] = True + continue + + state["last_input_at"] = time.monotonic() + state["redraw"] = True # any keypress repaints + + if state["mode"] == "confirm": + if key in (ord("y"), ord("Y")): + pending = state["confirm"] + state["confirm"] = None + state["mode"] = "main" + if pending: + state["flash"] = pending["apply"]() + elif key in (ord("n"), ord("N"), 27, ord("q")): + state["confirm"] = None + state["mode"] = "main" + continue + + if state["flash"]: + state["flash"] = "" + + if key == ord("q"): + return + + if state["mode"] == "policy": + if key in (27, ord("p"), ord("b")): + state["mode"] = "main" + continue + + if state["mode"] == "detail": + events = (state["events"] or {}).get("items", []) + if key in (27, ord("b"), curses.KEY_BACKSPACE, 127): + state["mode"] = "main" + elif key in (curses.KEY_DOWN, ord("j")): + state["ev_sel"] = min(len(events) - 1, state["ev_sel"] + 1) + elif key in (curses.KEY_UP, ord("k")): + state["ev_sel"] = max(0, state["ev_sel"] - 1) + continue + + # main view + rows = build_tree() + node = rows[min(state["sel"], len(rows) - 1)] if rows else {"kind": "all"} + + def move_tree(delta: int) -> None: + """Move the cursor and redraw now; the query happens on settle.""" + state["sel"] = max(0, min(len(rows) - 1, state["sel"] + delta)) + state["ev_sel"] = 0 + state["ev_page"] = 1 + mark_dirty() + + def turn_page(delta: int) -> None: + bundle = state["events"] or {} + if delta > 0 and not bundle.get("has_next"): + return # short page = genuine end + nxt = max(1, state["ev_page"] + delta) + if bundle.get("exact"): + nxt = min(nxt, bundle.get("pages", 1)) + if nxt != state["ev_page"]: + state["ev_page"] = nxt + state["ev_sel"] = 0 + mark_dirty() + + if key == ord("\t"): + state["focus"] = "events" if state["focus"] == "tree" else "tree" + elif key == ord("f"): + state["follow"] = not state["follow"] + elif key == ord("p"): + state["mode"] = "policy" + elif key == ord("P"): + ask_pause_toggle() + elif key == ord("R"): + ask_resume() + elif key == ord("s"): + state["sort"] = _SORTS[(_SORTS.index(state["sort"]) + 1) % len(_SORTS)] + state["ev_sel"] = 0 + state["ev_page"] = 1 + mark_dirty() + elif key == ord("r"): + from prismor.runtime import cost as cost_mod + state["pricing"] = _safe(lambda: cost_mod.load_pricing(force=True), + state["pricing"]) + state["base"] = _fetch_base() + state["sessions"].clear() + state["costs"].clear() + state["stats"] = None; state["want_stats"] = True + refetch_events() + elif key == ord("v"): + state["verdict"] = {"": "blocked", "blocked": "allowed", "allowed": ""}[state["verdict"]] + state["ev_sel"] = 0 + state["ev_page"] = 1 + mark_dirty() + elif key in (ord("]"), curses.KEY_NPAGE): + turn_page(1) + elif key in (ord("["), curses.KEY_PPAGE): + turn_page(-1) + elif key in (curses.KEY_RIGHT, ord("l")): + if node["kind"] == "agent": + name = node["name"] + if name not in state["expanded"]: + state["expanded"].add(name) + load_sessions(node["agent"]) + elif key in (curses.KEY_LEFT, ord("h")): + if node["kind"] == "agent": + state["expanded"].discard(node["name"]) + elif node["kind"] in ("session", "loading", "empty", "note"): + # jump back up to the owning agent row and collapse it + for i in range(state["sel"], -1, -1): + if rows[i]["kind"] == "agent": + state["expanded"].discard(rows[i]["name"]) + state["sel"] = i + mark_dirty() + break + elif key in (curses.KEY_ENTER, 10, 13): + if state["focus"] == "events" and (state["events"] or {}).get("items"): + state["mode"] = "detail" + elif node["kind"] == "agent": + name = node["name"] + if name in state["expanded"]: + state["expanded"].discard(name) + else: + state["expanded"].add(name) + load_sessions(node["agent"]) + else: + state["focus"] = "events" + elif key in (curses.KEY_DOWN, ord("j")): + if state["focus"] == "tree": + move_tree(1) + else: + items = (state["events"] or {}).get("items", []) + if state["ev_sel"] + 1 >= len(items): + turn_page(1) # roll onto the next page + else: + state["ev_sel"] += 1 + elif key in (curses.KEY_UP, ord("k")): + if state["focus"] == "tree": + move_tree(-1) + elif state["ev_sel"] == 0: + turn_page(-1) + else: + state["ev_sel"] -= 1 + elif key == ord("g"): + if state["focus"] == "events": + state["ev_sel"] = 0 + state["ev_page"] = 1 + mark_dirty() + else: + state["sel"] = 0 + mark_dirty() + elif key == ord("G"): + if state["focus"] == "events": + bundle = state["events"] or {} + if bundle.get("exact"): + state["ev_page"] = bundle.get("pages", 1) + state["ev_sel"] = 0 + mark_dirty() + else: + # Last page is unknowable for store-backed paging; + # jump to the end of what's loaded instead. + state["ev_sel"] = max(0, len(bundle.get("items", [])) - 1) + else: + state["sel"] = len(rows) - 1 + mark_dirty() + + try: + curses.wrapper(draw) + except curses.error: + return False + except KeyboardInterrupt: + pass + return True + + +def run_term() -> None: + """Entry point for ``prismor term``.""" + import sys + + if not sys.stdout.isatty() or not sys.stdin.isatty(): + _render_plain() + return + if not _run_curses(): + _render_plain() From 8347563b0b432f1c19ea1accb79053b674510965 Mon Sep 17 00:00:00 2001 From: ar9av Date: Sun, 2 Aug 2026 12:27:27 -0700 Subject: [PATCH 2/2] docs(term): add Terminal Console guide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Full page for `prismor term`: layout, the complete key map, how session loading and its 50-per-agent limit behave, pausing a session, resuming one in Claude Code, and how cost is derived. Documents the caveats the UI already marks rather than leaving them to be discovered: cost exists only for agents that keep transcripts (unknown is rendered as "—", never $0.00), agent totals cover loaded sessions only, cache-write uses the 1.25x multiplier because the price feed omits that rate, and the figures are list-price API costs rather than a bill. Also records why event counts read "207+" for agent scope but are exact for session scope. Linked from the CLI reference, the dashboard doc (same underlying store), and the README doc list. --- README.md | 1 + docs/cli-reference.md | 3 +- docs/dashboard.md | 1 + docs/terminal-console.md | 181 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 185 insertions(+), 1 deletion(-) create mode 100644 docs/terminal-console.md diff --git a/README.md b/README.md index 85e2564..adcace4 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,7 @@ For the Skill, curl, and git-clone alternatives, plus PEP 668 systems and secret - ⚖️ [Layered Policy & Exemptions](docs/policy-layers-and-exemptions.md) covers per-rule observe/enforce, the non-overridable floor, and admin-granted, time-boxed exemptions across org / project / repo layers - 📡 [Live Telemetry](docs/live-telemetry.md) covers the optional enterprise control-plane link — device enrollment, signed remote policy, and redacted telemetry streamed to a self-hosted org dashboard - 📊 [Dashboard](docs/dashboard.md) covers the terminal and local web dashboards plus session forensics +- 🖥️ [Terminal Console](docs/terminal-console.md) is `prismor term` — a full-screen agent → session tree with a live event tail, policy precedence, and per-session token cost, without leaving the terminal - 🧾 [Signed Audit Trail](docs/audit-trail.md) hash-chains and Ed25519-signs every agent action locally, so `prismor trail verify` proves the history hasn't been edited, deleted, or rewritten - 📑 [Attestation Bundle](docs/attestation-bundle.md) packages posture, agent inventory, host discovery, framework-control coverage (OWASP LLM/Agentic, NIST AI RMF, EU AI Act), and the trail anchor into one Ed25519-signed file an auditor re-verifies with `prismor attest verify` - 🔦 [Host Discovery](docs/attestation-bundle.md#host-discovery) sweeps the machine with `prismor discover` and flags any AI agent running without Prismor hooks (shadow AI) diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 9e00ecd..b937274 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -161,7 +161,7 @@ Full policy model, rule schema, and the default rule list: [Prismor](prismor-run | `prismor attest coverage` | `--json`, `--workspace` | Show which compliance-framework controls the active policy covers (OWASP LLM/Agentic, NIST AI RMF, EU AI Act). | | `prismor discover` | `--json`, `--workspace` | Sweep this host for AI agents and flag any running without Prismor hooks (shadow AI). Host-local, read-only. See [Host discovery](attestation-bundle.md#host-discovery). | | `prismor status --all` | `--days N` | Terminal overview of every registered workspace. See [Dashboard](dashboard.md). | -| `prismor term` | — | Full-screen terminal console: agent → session tree, live event tail, policy precedence, per-session token cost. Falls back to a plain table when stdout isn't a tty. | +| `prismor term` | — | Full-screen terminal console: agent → session tree, live event tail, policy precedence, per-session token cost. Falls back to a plain table when stdout isn't a tty. See [Terminal Console](terminal-console.md). | | `prismor dashboard` | `--port`, `--host`, `--no-open` | Local web dashboard at `http://127.0.0.1:7070` (opens a browser tab). See [Dashboard](dashboard.md). | | `prismor serve` | `--port`, `--host`, `--no-open` | _Deprecated_ alias of `dashboard --no-open` (headless server only). | @@ -270,5 +270,6 @@ Scoring table, IOC feed, ecosystem support: [Supply Chain](supply-chain.md). - [Skill Scanner](skill-scanner.md) — MCP + skill risk scanning - [Sweep & Cloak](sweep-and-cloak.md) — secret prevention - [Semantic Guard](semantic-guard.md) — LLM-assisted injection defense +- [Terminal Console](terminal-console.md) — `prismor term`, the full-screen agent/session console - [Canary](canary.md) · [IAM](iam.md) · [Scoped Agent](scoped-agent.md) · [Learning](learning.md) · [Dashboard](dashboard.md) - [Docker & Containers](docker.md) · [Architecture](architecture.md) diff --git a/docs/dashboard.md b/docs/dashboard.md index 27ed6de..75a3f4a 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -159,6 +159,7 @@ viewer, with full rule metadata. ## See also +- [Terminal Console](terminal-console.md) — `prismor term`, a full-screen console over this same store: agent → session tree, live event tail, per-session cost - [Prismor](prismor-runtime.md) — session-log schema and the audit command - [Learning](learning.md) — mines this same history for new rules - [CLI Reference](cli-reference.md) — all commands at a glance diff --git a/docs/terminal-console.md b/docs/terminal-console.md new file mode 100644 index 0000000..e4d405b --- /dev/null +++ b/docs/terminal-console.md @@ -0,0 +1,181 @@ +# Terminal Console (`prismor term`) + +`prismor term` is a full-screen console for everything Prismor has recorded: a +tree of your agents and their sessions, a live event tail, the policy layer in +effect, and what each session cost in tokens. + +It is a **renderer only**. Every number comes from +[`prismor/runtime/store.py`](../prismor/runtime/store.py) — the same data layer +behind [`prismor dashboard`](dashboard.md) — which aggregates across every +registered workspace and re-cloaks secrets on read, so evidence panes show +`@@SECRET:NAME@@` placeholders and never real values. + +Implementation: [`prismor/runtime/term.py`](../prismor/runtime/term.py), cost in +[`prismor/runtime/cost.py`](../prismor/runtime/cost.py). + +```bash +prismor term +``` + +Needs a real terminal. Piped or redirected (CI, `| head`, no tty), it prints a +plain agent + session table instead, so it stays scriptable. + +--- + +## The layout + +``` + >_ PRISMOR TERM │ Org: acme │ Policy: default │ Agents: 35 │ prices live │ FOLLOW + Agents ▸ sessions ↓last run │ Detail +────────────────────────────────│────────────────────────────────────────────── + All agents │ Session d981011d-6f54-4516-a698-2afe8db… + ▾ ○ claude 512/229 │ Agent claude (claude) + 6a5d59a6-e r90 2f — 1d │ Risk 70/100 · 1 findings + d981011d-6 r70 1f $426 2d │ Immunity active + 19d34f10-6 r0 0f $0.05 1d │ Cost $426.00 est · 1629 turns · claude-opus-5 + ▸ ○ codex 64/32 │ Tokens in 3.1k · out 1.4M · cache r687.6M/w7.4M + ▸ ○ langchain 11/11 │ + │ Events [page 1/3 of 60 · session · filter: all] + │─────────────────────────────────────────────── + │ 23:08:18 claude BLOCK shell: rm -rf / + │ 23:08:01 claude allow prompt + [j/k] Move [→/←] Expand [Tab] Pane [[/]] Page [Enter] Detail [s] Sort … +``` + +**Header** — org (or `local` when unenrolled), the winning policy layer, agent +count, token-price freshness, and whether the tail is following. + +**Left pane** — a two-level tree. Agents show `total/flagged` call counts and a +status dot. Expand one and its sessions appear beneath, each with risk, +findings, cost, and age: `r90 2f $426 2d`. + +**Right pane** — detail for whatever is selected, above an event tail scoped to +the same thing: all agents, one agent, or one session. + +--- + +## Keys + +| Key | Does | +|---|---| +| `j` / `k`, `↑` / `↓` | Move in the focused pane | +| `→` / `l`, `←` / `h` | Expand / collapse an agent's sessions | +| `Tab` | Switch focus between the tree and the event tail | +| `Enter` | On an agent: expand. On an event: full detail (rule, category, evidence) | +| `[` / `]`, `PgUp` / `PgDn` | Page the event tail | +| `g` / `G` | Jump to top / bottom | +| `s` | Cycle session sort: last run → risk → findings → cost | +| `f` | Toggle follow (live tail, refreshes every 2s) | +| `v` | Cycle verdict filter: all → blocked → allowed | +| `p` | Policy precedence overlay | +| `P` | Pause / resume immunity for the selected session (asks first) | +| `R` | Resume the selected session in Claude Code (asks first) | +| `r` | Refresh everything, including token prices | +| `q` | Quit | + +--- + +## Sessions + +Expanding an agent loads its sessions newest-first. Two limits are worth knowing +because the UI states them rather than hiding them: + +- At most **50 sessions per agent** are loaded (scanning up to 5 pages). When + that bites, the tree shows `… showing first 50`. An agent with 512 sessions + gives you the 50 most recent, not a random 50. +- `s` re-sorts **what is loaded**, not the full history. Sorting by cost ranks + within those 50 most-recent sessions. + +Selecting a session scopes the event tail to it and shows its risk, workspace, +scoped rules, and immunity state. + +### Pausing a session + +`P` toggles immunity for one session via the same control the web dashboard +uses. It confirms first, and it writes: paused sessions stop being screened +until you resume them. It affects that session only — not the agent, not the +workspace. + +### Resuming a session in Claude Code + +`R` hands the terminal to `claude --resume `, run in that session's +original working directory. `prismor term` restores itself when you exit claude. + +This works because Claude Code's conversation ids *are* the session ids Prismor +records. It refuses, with the reason, when the session ran under a different +framework, when the `claude` CLI isn't on `PATH`, or when the recorded workspace +no longer exists (common for sessions that ran in temp directories). + +--- + +## Cost + +Prismor's event store records *what* an agent did, never how many tokens it +took — there is no usage column and no usage payload in any event. So cost is +joined from two outside sources: + +| | Source | +|---|---| +| **Prices** | The published feed at `https://www.aipricing.guru/api/pricing.json`, cached to `~/.prismor/pricing-cache.json` for 12h | +| **Usage** | The agent's own transcript — Claude Code writes `~/.claude/projects//.jsonl` with a `usage` block per turn | + +The header reports which prices are in play: `prices live` (just fetched), +`prices cached` (within TTL), `prices STALE` (fetch failed, serving the cache), +or `prices n/a` (no network and no cache — costs render unpriced). Nothing +blocks on the network; the fetch has a 6s timeout and falls back. + +**Read the numbers with these caveats**, all of which the UI marks: + +- Cost is shown only for agents that keep transcripts — Claude Code today. + Other frameworks show `—` (unknown), never `$0.00`. A `·` means *not priced + yet*; it fills in within a second. +- Agent-level totals say `across N of M loaded sessions`. They are not a + lifetime total for that agent. +- The price feed publishes no **cache-write** rate, so that component uses + Anthropic's standard 1.25x input multiplier. That is the 5-minute-TTL figure; + 1-hour-TTL caching bills 2x, so long-cache sessions are underestimated. + Everything is labelled `est` for this reason. +- These are **list-price API costs**. On a subscription plan your real marginal + cost was the flat fee — the figure answers "what would this have cost on the + API", which is useful for attribution, not a bill. + +--- + +## Event paging + +The tail fetches exactly one screenful at a time, so paging stays fast on a +large store. + +Counts are reported honestly, and they differ by scope: + +- **Session-scoped** events are read in full (the store returns up to 60) and + sliced locally, so the count is exact: `page 1/3 of 60`. +- **Agent- and all-scoped** events come from a store query whose reported total + grows as you page deeper — it is a floor, not a count. The header says + `page 3 · 23 of 207+` and `]` simply stops when a page comes back short. + +--- + +## Performance notes + +The console is built to never block on a keypress: + +- Navigation redraws immediately from cached rows and re-queries only once the + selection has been still for ~120ms, so holding `j` never queues queries. +- The 24-hour aggregate (the "All agents" KPI panel) is the one expensive query + in the app. It is never run on startup or during a redraw — the panel paints + as `computing 24h totals…` and fills in when idle. +- Session pricing runs in small batches while idle. +- The screen repaints only when something changed. + +If the terminal is smaller than 50x10 the console says so rather than drawing a +broken frame. + +--- + +## See also + +- [Dashboard & Sessions](dashboard.md) — the web dashboard and session forensics over the same store +- [Policy Layers & Exemptions](policy-layers-and-exemptions.md) — what the `p` overlay is showing +- [Scoped Agent](scoped-agent.md) — the per-session controls `P` toggles +- [CLI Reference](cli-reference.md) — all commands at a glance