feat(observability): run trace tree, /metrics endpoint, hive trace CLI

docs/guide/cli-reference.md

@@ -89,6 +89,16 @@ Show detailed summary of a recorded run -- goals, decisions, tool usage.
 hive inspect <run-id>
 ```
 
+### `hive trace`
+
+Render a run's span tree -- run -> agent -> goal -> decision/tool -- derived
+from the structured logs. Goals show their outcome, decisions their token
+counts, tools a success/failure mark.
+
+```bash
+hive trace <run-id>
+```
+
 ### `hive replay`
 
 Replay a past session step by step.

docs/guide/rest-api.md

@@ -97,6 +97,8 @@ Interactive docs are auto-generated at `/docs` (Swagger) and `/redoc`.
 | `GET` | `/status` | Status of all agents |
 | `GET` | `/healthz` | Liveness + readiness (DB reachable) |
 | `GET` | `/runs`, `/runs/{id}` | Structured run logs |
+| `GET` | `/runs/{id}/trace` | Span tree derived from run logs (run -> agent -> goal -> decision/tool) |
+| `GET` | `/metrics` | Prometheus text metrics: agents by status + latest-run counters |
 | `GET` | `/approvals` | Global pending-approval queue |
 | `GET` | `/agents/{id}/approvals` | Pending approvals for one agent |
 | `POST` | `/agents/{id}/approvals/{approval_id}` | Approve or deny |

src/hive/cli/main.py

@@ -633,6 +633,44 @@ def inspect(run_id: str = typer.Argument(help="Run ID to inspect")) -> None:
             console.print(f"    {status_icon} [{g.event}] {obj}")
 
 
+@app.command()
+def trace(run_id: str = typer.Argument(help="Run ID to trace")) -> None:
+    """Render a run's span tree: run -> agent -> goal -> decision/tool."""
+    from rich.tree import Tree as RichTree
+
+    from hive.logging.trace import Span, TraceBuilder, children_of
+
+    builder = TraceBuilder(Path.cwd() / "logs")
+    spans = builder.build(run_id)
+    if not spans:
+        console.print(f"[red]Run not found: {run_id}[/red]")
+        raise typer.Exit(1)
+
+    style = {"run": "bold", "agent": "cyan", "goal": "green", "decision": "dim", "tool": "yellow"}
+
+    def label(s: Span) -> str:
+        extra = ""
+        if s.kind == "goal" and s.attributes.get("outcome"):
+            extra = f" [{s.attributes['outcome']}]"
+        elif s.kind == "decision":
+            tokens = (s.attributes.get("input_tokens") or 0) + (
+                s.attributes.get("output_tokens") or 0
+            )
+            extra = f" ({tokens} tok)" if tokens else ""
+        elif s.kind == "tool":
+            extra = " ✓" if s.attributes.get("success") else " ✗"
+        return f"[{style[s.kind]}]{s.name}{extra}[/{style[s.kind]}]"
+
+    def attach(node: RichTree, parent_id: str) -> None:
+        for child in children_of(spans, parent_id):
+            attach(node.add(label(child)), child.span_id)
+
+    root_span = spans[0]
+    tree = RichTree(label(root_span))
+    attach(tree, root_span.span_id)
+    console.print(tree)
+
+
 @app.command()
 def lives() -> None:
     """List all agent life directories."""

src/hive/logging/trace.py

@@ -0,0 +1,158 @@
+"""Trace tree derived from a run's structured JSONL logs.
+
+A pure data transform: ``TraceBuilder`` reads a finished (or in-progress) run
+via ``LogReader`` and derives a span tree -- run -> agent -> goal ->
+decision/tool -- from the correlation fields (``goal_id``, ``step_index``)
+the writers already record. The JSONL files stay the single source of truth;
+no new write path is introduced.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+from hive.logging.reader import LogReader
+
+SpanKind = Literal["run", "agent", "goal", "decision", "tool"]
+
+
+def _seg(value: str) -> str:
+    """Sanitize one span-id path segment.
+
+    Span ids join segments with ``/``; a literal ``/`` inside an agent or
+    goal id would let two distinct tuples collide on the same span id.
+    """
+    return value.replace("/", "%2F")
+
+
+class Span(BaseModel):
+    """One node in the derived trace tree.
+
+    ``span_id`` values are deterministic paths (``run/agent/goal/...``) so the
+    same logs always produce the same tree.
+    """
+
+    span_id: str
+    parent_span_id: str | None = None
+    name: str
+    kind: SpanKind
+    start: datetime | None = None
+    end: datetime | None = None
+    attributes: dict[str, Any] = Field(default_factory=dict)
+
+
+class TraceBuilder:
+    """Builds span trees from run logs."""
+
+    def __init__(self, logs_dir: Path):
+        self._reader = LogReader(logs_dir)
+
+    def build(self, run_id: str) -> list[Span]:
+        """Derive the span tree for ``run_id``.
+
+        Returns an empty list for an unknown run. Decisions/tools whose
+        ``goal_id`` matches no goal span (e.g. goal-generation decisions, or
+        logs from before correlation fields existed) attach to the agent span
+        rather than being dropped.
+        """
+        run = self._reader.get_run(run_id)
+        if run is None:
+            return []
+
+        run_span = Span(
+            span_id=run_id,
+            name=f"run {run_id}",
+            kind="run",
+            start=run.started_at,
+            attributes={"heartbeat": run.heartbeat, "profiles": run.profiles},
+        )
+        spans = [run_span]
+
+        for agent_id in self._reader.get_agent_ids(run_id):
+            agent_span_id = f"{run_id}/{_seg(agent_id)}"
+            spans.append(
+                Span(
+                    span_id=agent_span_id,
+                    parent_span_id=run_id,
+                    name=agent_id,
+                    kind="agent",
+                )
+            )
+
+            goal_spans: dict[str, Span] = {}
+            for goal in self._reader.get_agent_goals(run_id, agent_id):
+                existing = goal_spans.get(goal.goal_id)
+                if existing is None:
+                    span = Span(
+                        span_id=f"{agent_span_id}/{_seg(goal.goal_id)}",
+                        parent_span_id=agent_span_id,
+                        name=goal.objective or goal.goal_id,
+                        kind="goal",
+                        start=goal.ts if goal.event == "generated" else None,
+                        attributes={
+                            "goal_id": goal.goal_id,
+                            # "in_progress" until a terminal event closes the
+                            # span, so a crashed or mid-flight run is
+                            # distinguishable from a closed goal.
+                            "outcome": ("in_progress" if goal.event == "generated" else goal.event),
+                        },
+                    )
+                    goal_spans[goal.goal_id] = span
+                    spans.append(span)
+                else:
+                    # Later events (completed/abandoned) close and annotate the span.
+                    existing.attributes["outcome"] = goal.event
+                    if goal.event in ("completed", "abandoned"):
+                        existing.end = goal.ts
+                    if goal.objective and existing.name == goal.goal_id:
+                        existing.name = goal.objective
+            goal_span_ids = {gid: s.span_id for gid, s in goal_spans.items()}
+
+            for i, d in enumerate(self._reader.get_agent_decisions(run_id, agent_id)):
+                parent = goal_span_ids.get(d.goal_id, agent_span_id)
+                spans.append(
+                    Span(
+                        span_id=f"{parent}/d{i}",
+                        parent_span_id=parent,
+                        name=f"{d.decision_type} #{d.step_index or i}",
+                        kind="decision",
+                        start=d.ts,
+                        attributes={
+                            "model": d.model,
+                            "input_tokens": d.input_tokens,
+                            "output_tokens": d.output_tokens,
+                            "cost_usd": d.cost_usd,
+                            "duration_ms": d.duration_ms,
+                            "success": d.success,
+                        },
+                    )
+                )
+
+            for i, t in enumerate(self._reader.get_agent_tools(run_id, agent_id)):
+                parent = goal_span_ids.get(t.goal_id, agent_span_id)
+                spans.append(
+                    Span(
+                        span_id=f"{parent}/t{i}",
+                        parent_span_id=parent,
+                        name=t.tool_name,
+                        kind="tool",
+                        start=t.ts,
+                        attributes={
+                            "step_index": t.step_index,
+                            "success": t.success,
+                            "duration_ms": t.duration_ms,
+                            "error": t.error,
+                        },
+                    )
+                )
+
+        return spans
+
+
+def children_of(spans: list[Span], parent_id: str | None) -> list[Span]:
+    """Direct children of ``parent_id`` (None = roots), in log order."""
+    return [s for s in spans if s.parent_span_id == parent_id]

src/hive/server/routes/system.py

@@ -73,3 +73,60 @@ async def get_run(run_id: str, ctx: ServerContext = Depends(get_context)) -> dic
     if not summary:
         raise HTTPException(status_code=404, detail=f"run not found: {run_id}")
     return summary
+
+
+@router.get("/runs/{run_id}/trace")
+async def get_run_trace(
+    run_id: str, ctx: ServerContext = Depends(get_context)
+) -> list[dict[str, Any]]:
+    """Span tree (run -> agent -> goal -> decision/tool) derived from run logs."""
+    from hive.logging.trace import TraceBuilder
+
+    builder = TraceBuilder(ctx.root / "logs")
+    spans = await asyncio.to_thread(builder.build, run_id)
+    if not spans:
+        raise HTTPException(status_code=404, detail=f"run not found: {run_id}")
+    return [s.model_dump(mode="json") for s in spans]
+
+
+@router.get("/metrics", response_class=Response)
+async def metrics(ctx: ServerContext = Depends(get_context)) -> Response:
+    """Prometheus text-format metrics: agent statuses plus latest-run counters.
+
+    Rendered by hand -- the exposition text format is trivial and not worth a
+    dependency. The per-run values are snapshots of the most recent run's log
+    summary, exposed as gauges (they reset when a new run starts, so they are
+    deliberately NOT counters -- don't apply rate()/increase() to them).
+    """
+    from hive.logging.reader import LogReader
+
+    agents = await ctx.store.list_agents()
+    by_status: dict[str, int] = {}
+    for a in agents:
+        by_status[a.status.value] = by_status.get(a.status.value, 0) + 1
+
+    reader = LogReader(ctx.root / "logs")
+    runs = await asyncio.to_thread(reader.list_runs)
+    summary: dict[str, Any] = {}
+    if runs:
+        summary = await asyncio.to_thread(reader.get_summary, runs[0].run_id)
+
+    lines = [
+        "# HELP hive_agents Agents known to the store, by status.",
+        "# TYPE hive_agents gauge",
+    ]
+    for status_value, count in sorted(by_status.items()):
+        lines.append(f'hive_agents{{status="{status_value}"}} {count}')
+    snapshot_help = {
+        "goals_generated": "Goals generated in the latest run (snapshot; resets each run).",
+        "goals_completed": "Goals completed in the latest run (snapshot; resets each run).",
+        "goals_abandoned": "Goals abandoned in the latest run (snapshot; resets each run).",
+        "tool_calls": "Tool calls in the latest run (snapshot; resets each run).",
+        "total_tokens": "Tokens consumed in the latest run (snapshot; resets each run).",
+        "total_cost_usd": "Estimated cost (USD) of the latest run (snapshot; resets each run).",
+    }
+    for key, help_text in snapshot_help.items():
+        lines.append(f"# HELP hive_{key} {help_text}")
+        lines.append(f"# TYPE hive_{key} gauge")
+        lines.append(f"hive_{key} {summary.get(key, 0)}")
+    return Response("\n".join(lines) + "\n", media_type="text/plain; version=0.0.4")