Add dependency-light TUI snapshot scaffold

pyproject.toml

@@ -50,6 +50,7 @@ matrix = [
 
 [project.scripts]
 agent-term = "agent_term.cli:main"
+agent-term-snapshot = "agent_term.snapshot_cli:main"
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/agent_term/snapshot_cli.py

@@ -0,0 +1,41 @@
+"""CLI entry point for rendering an AgentTerm operator snapshot."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from agent_term.store import DEFAULT_DB_PATH, EventStore
+from agent_term.tui_model import TuiSnapshotBuilder
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="agent-term-snapshot",
+        description="Render a dependency-light AgentTerm operator snapshot from the local event log.",
+    )
+    parser.add_argument(
+        "--db",
+        default=str(DEFAULT_DB_PATH),
+        help="Path to the local AgentTerm SQLite event log.",
+    )
+    parser.add_argument("channel", nargs="?", help="Optional channel/room to render.")
+    parser.add_argument("--limit", type=int, default=100, help="Maximum events to render.")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    store = EventStore(Path(args.db))
+    try:
+        events = store.tail(channel=args.channel, limit=args.limit)
+        snapshot = TuiSnapshotBuilder().build(events)
+        print(snapshot.render_text())
+        return 0
+    finally:
+        store.close()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

src/agent_term/tui_model.py

@@ -0,0 +1,214 @@
+"""Terminal UI view-model primitives.
+
+This is intentionally dependency-light. A future Textual application can render this
+model, but the grouping, status classification, and safety affordances are tested here
+without requiring a live terminal session in CI.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Iterable
+
+from agent_term.events import AgentTermEvent
+
+
+PANE_ORDER = (
+    "rooms",
+    "threads",
+    "agents",
+    "approvals",
+    "workrooms",
+    "topics",
+    "context",
+    "semantic",
+    "investigations",
+    "graphs",
+    "shells",
+    "runs",
+    "evidence",
+    "events",
+)
+
+
+@dataclass(frozen=True)
+class TuiLine:
+    """One rendered line in a TUI pane."""
+
+    text: str
+    status: str = "info"
+    event_id: str | None = None
+    source: str | None = None
+    kind: str | None = None
+    metadata: dict[str, object] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class TuiPane:
+    """A named TUI pane."""
+
+    name: str
+    title: str
+    lines: tuple[TuiLine, ...] = ()
+
+
+@dataclass(frozen=True)
+class TuiSnapshot:
+    """Complete dependency-free AgentTerm TUI snapshot."""
+
+    panes: tuple[TuiPane, ...]
+
+    def pane(self, name: str) -> TuiPane:
+        for pane in self.panes:
+            if pane.name == name:
+                return pane
+        raise KeyError(f"unknown pane: {name}")
+
+    def render_text(self) -> str:
+        blocks: list[str] = []
+        for pane in self.panes:
+            blocks.append(f"[{pane.title}]")
+            if not pane.lines:
+                blocks.append("  <empty>")
+                continue
+            for line in pane.lines:
+                blocks.append(f"  {line.status.upper()}: {line.text}")
+        return "\n".join(blocks)
+
+
+class TuiSnapshotBuilder:
+    """Builds an operator-oriented TUI snapshot from AgentTerm events."""
+
+    def build(self, events: Iterable[AgentTermEvent]) -> TuiSnapshot:
+        pane_lines: dict[str, list[TuiLine]] = {name: [] for name in PANE_ORDER}
+        seen_rooms: set[str] = set()
+        seen_threads: set[str] = set()
+
+        for event in events:
+            room_line = _room_line(event)
+            if room_line.text not in seen_rooms:
+                pane_lines["rooms"].append(room_line)
+                seen_rooms.add(room_line.text)
+
+            if event.thread_id and event.thread_id not in seen_threads:
+                pane_lines["threads"].append(
+                    TuiLine(
+                        text=f"{event.thread_id} in {event.channel}",
+                        status="info",
+                        event_id=event.event_id,
+                        source=event.source,
+                        kind=event.kind,
+                    )
+                )
+                seen_threads.add(event.thread_id)
+
+            pane_name = classify_event(event)
+            pane_lines[pane_name].append(event_line(event))
+            pane_lines["events"].append(event_line(event))
+
+        panes = tuple(
+            TuiPane(name=name, title=title_for_pane(name), lines=tuple(pane_lines[name]))
+            for name in PANE_ORDER
+        )
+        return TuiSnapshot(panes=panes)
+
+
+def classify_event(event: AgentTermEvent) -> str:
+    if event.source in {"agent-registry", "hermes", "codex", "claude-code", "openclaw"}:
+        return "agents"
+    if event.source in {"github", "ci", "mcp", "local-process"}:
+        return "agents"
+    if event.source == "policy-fabric" or event.kind in {"decision", "policy_check"}:
+        return "approvals"
+    if event.source == "prophet-workspace" or event.kind == "workroom":
+        return "workrooms"
+    if event.source == "slash-topics" or event.kind in {"topic_scope", "topic_membrane"}:
+        return "topics"
+    if event.source == "memory-mesh" or event.kind in {"memory_recall", "memory_write", "context_pack"}:
+        return "context"
+    if event.source == "new-hope" or event.kind in {"semantic_thread", "claim", "citation"}:
+        return "semantic"
+    if event.source in {"holmes", "sherlock-search"}:
+        return "investigations"
+    if event.source == "meshrush" or event.kind in {"graph_view", "graph_artifact"}:
+        return "graphs"
+    if event.source == "cloudshell-fog" or event.kind in {"shell_session", "shell_attach"}:
+        return "shells"
+    if event.source == "agentplane" or event.kind in {"validation", "placement", "run", "replay"}:
+        return "runs"
+    if _has_evidence(event):
+        return "evidence"
+    return "events"
+
+
+def event_line(event: AgentTermEvent) -> TuiLine:
+    status = status_for_event(event)
+    prefix = _event_prefix(event)
+    text = f"{prefix}{event.body}"
+    return TuiLine(
+        text=text,
+        status=status,
+        event_id=event.event_id,
+        source=event.source,
+        kind=event.kind,
+        metadata=event.metadata,
+    )
+
+
+def status_for_event(event: AgentTermEvent) -> str:
+    metadata = event.metadata
+    if bool(metadata.get("revoked")):
+        return "revoked"
+    if metadata.get("deny_reason") or metadata.get("admission_status") == "denied":
+        return "denied"
+    if metadata.get("policy_status") == "pending" or metadata.get("approval_required"):
+        return "pending"
+    if metadata.get("fail_closed"):
+        return "blocked"
+    if metadata.get("matrix_sensitive_context_allowed") is False:
+        return "blocked"
+    if metadata.get("dispatch_status") == "invoked":
+        return "active"
+    if metadata.get("cloudshell_status") in {"running", "attach_prepared"}:
+        return "active"
+    if metadata.get("agentplane_status") in {"completed", "placed", "valid"}:
+        return "active"
+    return "info"
+
+
+def title_for_pane(name: str) -> str:
+    titles = {
+        "rooms": "Matrix Rooms / Channels",
+        "threads": "Threads / Work Orders",
+        "agents": "Agents / Grants / Revocation",
+        "approvals": "Approvals / Policy Fabric",
+        "workrooms": "Prophet Workrooms",
+        "topics": "Slash Topics",
+        "context": "Memory / Context",
+        "semantic": "New Hope Semantic Objects",
+        "investigations": "Holmes / Sherlock",
+        "graphs": "MeshRush Graphs",
+        "shells": "cloudshell-fog",
+        "runs": "AgentPlane Runs",
+        "evidence": "Evidence",
+        "events": "Event Log",
+    }
+    return titles[name]
+
+
+def _room_line(event: AgentTermEvent) -> TuiLine:
+    room = str(event.metadata.get("matrix_room_alias") or event.channel)
+    return TuiLine(text=room, status="info", event_id=event.event_id, source=event.source)
+
+
+def _event_prefix(event: AgentTermEvent) -> str:
+    agent = event.metadata.get("agent_id")
+    workroom = event.metadata.get("workroom")
+    topic = event.metadata.get("topic_scope")
+    pieces = [piece for piece in (agent, workroom, topic) if piece]
+    return f"({' / '.join(str(piece) for piece in pieces)}) " if pieces else ""
+
+
+def _has_evidence(event: AgentTermEvent) -> bool:
+    artifacts = event.metadata.get("artifacts")
+    return bool(artifacts or event.metadata.get("artifact_ref") or event.metadata.get("evidence_ref"))

tests/test_tui_model.py

@@ -0,0 +1,99 @@
+from agent_term.events import AgentTermEvent
+from agent_term.tui_model import TuiSnapshotBuilder, classify_event, status_for_event
+
+
+def event(
+    source: str,
+    kind: str,
+    body: str = "body",
+    metadata: dict[str, object] | None = None,
+    thread_id: str | None = None,
+) -> AgentTermEvent:
+    return AgentTermEvent(
+        channel=f"!{source}",
+        sender="@operator",
+        kind=kind,
+        source=source,
+        body=body,
+        thread_id=thread_id,
+        metadata=metadata or {},
+    )
+
+
+def test_classifies_sourceos_control_panes():
+    assert classify_event(event("agent-registry", "agent_identity")) == "agents"
+    assert classify_event(event("policy-fabric", "decision")) == "approvals"
+    assert classify_event(event("prophet-workspace", "workroom")) == "workrooms"
+    assert classify_event(event("slash-topics", "topic_scope")) == "topics"
+    assert classify_event(event("memory-mesh", "memory_recall")) == "context"
+    assert classify_event(event("new-hope", "semantic_thread")) == "semantic"
+    assert classify_event(event("holmes", "investigation")) == "investigations"
+    assert classify_event(event("sherlock-search", "search_packet")) == "investigations"
+    assert classify_event(event("meshrush", "graph_view")) == "graphs"
+    assert classify_event(event("cloudshell-fog", "shell_session")) == "shells"
+    assert classify_event(event("agentplane", "run")) == "runs"
+
+
+def test_status_for_event_surfaces_operator_risk_states():
+    assert status_for_event(event("policy-fabric", "decision", metadata={"deny_reason": "no"})) == "denied"
+    assert status_for_event(event("agent-registry", "agent_identity", metadata={"revoked": True})) == "revoked"
+    assert status_for_event(event("memory-mesh", "memory_recall", metadata={"approval_required": True})) == "pending"
+    assert status_for_event(event("matrix", "matrix_emit", metadata={"matrix_sensitive_context_allowed": False})) == "blocked"
+    assert status_for_event(event("codex", "agent_message", metadata={"dispatch_status": "invoked"})) == "active"
+
+
+def test_snapshot_builds_rooms_threads_and_domain_panes():
+    events = [
+        event(
+            "matrix",
+            "matrix_room_event",
+            "Matrix message",
+            metadata={"matrix_room_alias": "#sourceos-ops:example.org"},
+            thread_id="$root",
+        ),
+        event(
+            "agent-registry",
+            "agent_identity",
+            "Resolve agent.codex",
+            metadata={"agent_id": "agent.codex", "session_id": "session-1"},
+        ),
+        event(
+            "policy-fabric",
+            "decision",
+            "Denied memory recall",
+            metadata={"deny_reason": "no_policy_decision"},
+        ),
+        event(
+            "prophet-workspace",
+            "workroom",
+            "Bind workroom",
+            metadata={"workroom": "pi-demo"},
+        ),
+        event(
+            "agentplane",
+            "run",
+            "Run complete",
+            metadata={"agentplane_status": "completed", "artifacts": [{"artifact_kind": "RunArtifact"}]},
+        ),
+    ]
+
+    snapshot = TuiSnapshotBuilder().build(events)
+
+    assert snapshot.pane("rooms").lines[0].text == "#sourceos-ops:example.org"
+    assert snapshot.pane("threads").lines[0].text == "$root in !matrix"
+    assert snapshot.pane("agents").lines[0].text == "(agent.codex) Resolve agent.codex"
+    assert snapshot.pane("approvals").lines[0].status == "denied"
+    assert snapshot.pane("workrooms").lines[0].text == "(pi-demo) Bind workroom"
+    assert snapshot.pane("runs").lines[0].status == "active"
+
+
+def test_snapshot_render_text_is_operator_readable():
+    snapshot = TuiSnapshotBuilder().build(
+        [event("cloudshell-fog", "shell_session", "Shell ready", metadata={"cloudshell_status": "running"})]
+    )
+
+    rendered = snapshot.render_text()
+
+    assert "[cloudshell-fog]" in rendered
+    assert "ACTIVE: Shell ready" in rendered
+    assert "[Matrix Rooms / Channels]" in rendered