feat(webui): model usage, token cost, latency, and performance analytics (Closes #651)

Child of umbrella #655 (governed MCP restart coordination); builds on the
#657 restart-path inventory. Adds a central coordinator that evaluates live
control-plane state before a restart and returns a blast-radius impact
preview, so operators and the web console (#642/#652) can see what a restart
would disrupt before concurrent LLM work is destroyed.

Changes
- restart_coordinator.py (new) — pure classification: inventory -> impact
  report DTO (RestartImpactReport/SessionImpact/LeaseImpact). Verdicts:
  safe / unsafe / override. Never restarts anything; fails closed on an
  incomplete inventory.
- control_plane_db.py — additive ControlPlaneDB.list_sessions() read-only
  session inventory (the process-level unit a restart kills).
- gitea_mcp_server.py — new dry-run MCP tool gitea_request_mcp_restart:
  gathers sessions/leases/terminal-lock from the #613 DB, calls the
  coordinator, returns the report. Override authority is read from the
  environment, never self-asserted (#630/#710 F1 pattern). Apply is gated
  by a later drain proof (non-goal here).
- docs/mcp-restart-coordinator.md + docs/mcp-restart-impact-sample.json — doc
  and a real dry-run sample report.
- docs/mcp-tool-inventory.md — register the new tool (inventory sync).
- tests/test_restart_coordinator.py (new) — 15 tests: multi-session fixtures,
  deny-when-critical-section-open, fail-closed deny, override, terminal lock,
  stale heartbeat, JSON-serializable DTO, list_sessions.

Tests: pytest tests/test_restart_coordinator.py -> 15 passed. Full suite:
13 failed / 4753 passed; all 13 reproduce identically on clean master
@ef14622 (0 regressions). The residual test_issue_781 doc-registry failure
is a pre-existing baseline gap for gitea_rebind_dirty_same_claimant_author_session
(merged #864, undocumented on master) — out of scope for #658.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

control_plane_db.py

@@ -31,7 +31,7 @@ from typing import Any, Iterator, Sequence
 
 import dependency_graph
 
-SCHEMA_VERSION = 4
+SCHEMA_VERSION = 5
 
 # Assignable work kinds only — raw monitoring incidents are never work items.
 WORK_KINDS = frozenset({"issue", "pr"})
@@ -186,6 +186,34 @@ CREATE INDEX IF NOT EXISTS idx_dependency_edges_target
     ON dependency_edges(remote, org, repo, target_kind, target_number);
 CREATE INDEX IF NOT EXISTS idx_assignments_session ON assignments(session_id, status);
 CREATE INDEX IF NOT EXISTS idx_incident_gitea ON incident_links(gitea_org, gitea_repo, gitea_issue_number);
+
+-- Model usage, token cost, latency, and performance events (#651)
+CREATE TABLE IF NOT EXISTS usage_events (
+    usage_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    session_id TEXT,
+    remote TEXT NOT NULL DEFAULT 'dadeschools',
+    org TEXT NOT NULL DEFAULT '',
+    repo TEXT NOT NULL DEFAULT '',
+    project_id TEXT,
+    role TEXT NOT NULL DEFAULT 'unknown',
+    model TEXT NOT NULL DEFAULT 'unknown',
+    issue_number INTEGER,
+    pr_number INTEGER,
+    stage TEXT NOT NULL DEFAULT 'unknown',
+    input_tokens INTEGER,
+    output_tokens INTEGER,
+    total_tokens INTEGER,
+    estimated_cost_usd REAL,
+    latency_ms INTEGER,
+    duration_ms INTEGER,
+    status TEXT NOT NULL DEFAULT 'success',
+    metadata TEXT,
+    created_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_usage_events_scope ON usage_events(remote, org, repo);
+CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);
+CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);
 """
 
 
@@ -339,6 +367,7 @@ class ControlPlaneDB:
                 self._migrate_incident_links_null_scope(conn)
                 self._migrate_lease_lifecycle_columns(conn)
                 self._migrate_session_ownership_columns(conn)
+                self._migrate_usage_events_table(conn)
                 conn.execute(
                     "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
                     ("schema_version", str(SCHEMA_VERSION)),
@@ -518,6 +547,174 @@ class ControlPlaneDB:
                     f"UPDATE incident_links SET {col} = '' WHERE {col} IS NULL"
                 )
 
+    def _migrate_usage_events_table(self, conn: sqlite3.Connection) -> None:
+        """Create usage_events table and indexes if they do not exist (#651)."""
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS usage_events (
+                usage_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                session_id TEXT,
+                remote TEXT NOT NULL DEFAULT 'dadeschools',
+                org TEXT NOT NULL DEFAULT '',
+                repo TEXT NOT NULL DEFAULT '',
+                project_id TEXT,
+                role TEXT NOT NULL DEFAULT 'unknown',
+                model TEXT NOT NULL DEFAULT 'unknown',
+                issue_number INTEGER,
+                pr_number INTEGER,
+                stage TEXT NOT NULL DEFAULT 'unknown',
+                input_tokens INTEGER,
+                output_tokens INTEGER,
+                total_tokens INTEGER,
+                estimated_cost_usd REAL,
+                latency_ms INTEGER,
+                duration_ms INTEGER,
+                status TEXT NOT NULL DEFAULT 'success',
+                metadata TEXT,
+                created_at TEXT NOT NULL
+            );
+        """)
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_scope ON usage_events(remote, org, repo);")
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);")
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);")
+
+    def record_usage_event(
+        self,
+        *,
+        session_id: str | None = None,
+        remote: str = "dadeschools",
+        org: str = "",
+        repo: str = "",
+        project_id: str | None = None,
+        role: str = "unknown",
+        model: str = "unknown",
+        issue_number: int | None = None,
+        pr_number: int | None = None,
+        stage: str = "unknown",
+        input_tokens: int | None = None,
+        output_tokens: int | None = None,
+        total_tokens: int | None = None,
+        estimated_cost_usd: float | None = None,
+        latency_ms: int | None = None,
+        duration_ms: int | None = None,
+        status: str = "success",
+        metadata: str | dict[str, Any] | None = None,
+        created_at: str | None = None,
+    ) -> int:
+        """Record a model usage, token cost, latency, or stage performance event (#651)."""
+        ts = created_at or _ts()
+        meta_str: str | None = None
+        if metadata is not None:
+            from webui import console_redaction
+            redacted_meta = console_redaction.redact_payload(metadata)
+            if isinstance(redacted_meta, str):
+                meta_str = redacted_meta
+            else:
+                try:
+                    meta_str = json.dumps(redacted_meta, default=str)
+                except Exception:
+                    meta_str = str(redacted_meta)
+
+        if total_tokens is None and (input_tokens is not None or output_tokens is not None):
+            total_tokens = (input_tokens or 0) + (output_tokens or 0)
+
+        with self._tx(immediate=True) as conn:
+            cursor = conn.execute(
+                """
+                INSERT INTO usage_events (
+                    session_id, remote, org, repo, project_id, role, model,
+                    issue_number, pr_number, stage, input_tokens, output_tokens,
+                    total_tokens, estimated_cost_usd, latency_ms, duration_ms,
+                    status, metadata, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    session_id,
+                    remote,
+                    org,
+                    repo,
+                    project_id,
+                    role,
+                    model,
+                    issue_number,
+                    pr_number,
+                    stage,
+                    input_tokens,
+                    output_tokens,
+                    total_tokens,
+                    estimated_cost_usd,
+                    latency_ms,
+                    duration_ms,
+                    status,
+                    meta_str,
+                    ts,
+                ),
+            )
+            return cursor.lastrowid
+
+    def query_usage_events(
+        self,
+        *,
+        remote: str | None = None,
+        org: str | None = None,
+        repo: str | None = None,
+        project_id: str | None = None,
+        role: str | None = None,
+        model: str | None = None,
+        issue_number: int | None = None,
+        pr_number: int | None = None,
+        stage: str | None = None,
+        session_id: str | None = None,
+        limit: int = 500,
+        offset: int = 0,
+    ) -> list[dict[str, Any]]:
+        """Query stored usage events matching filters (#651)."""
+        conditions = []
+        params = []
+        if remote:
+            conditions.append("remote = ?")
+            params.append(remote)
+        if org:
+            conditions.append("org = ?")
+            params.append(org)
+        if repo:
+            conditions.append("repo = ?")
+            params.append(repo)
+        if project_id:
+            conditions.append("project_id = ?")
+            params.append(project_id)
+        if role:
+            conditions.append("role = ?")
+            params.append(role)
+        if model:
+            conditions.append("model = ?")
+            params.append(model)
+        if issue_number is not None:
+            conditions.append("issue_number = ?")
+            params.append(issue_number)
+        if pr_number is not None:
+            conditions.append("pr_number = ?")
+            params.append(pr_number)
+        if stage:
+            conditions.append("stage = ?")
+            params.append(stage)
+        if session_id:
+            conditions.append("session_id = ?")
+            params.append(session_id)
+
+        where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
+        sql = f"""
+            SELECT * FROM usage_events
+            {where_clause}
+            ORDER BY usage_id ASC
+            LIMIT ? OFFSET ?
+        """
+        params.extend([limit, offset])
+
+        with self._tx(immediate=False) as conn:
+            cursor = conn.execute(sql, params)
+            rows = cursor.fetchall()
+            return [dict(row) for row in rows]
+
     # ── sessions ──────────────────────────────────────────────────────────
 
     def upsert_session(
@@ -599,6 +796,35 @@ class ControlPlaneDB:
                 (_ts(), session_id),
             )
 
+    def list_sessions(
+        self,
+        *,
+        statuses: Sequence[str] | None = None,
+        limit: int = 500,
+    ) -> list[dict[str, Any]]:
+        """List session rows for restart / impact analysis (#658).
+
+        Read-only. Sessions are the process-level unit an MCP restart
+        disrupts, so the restart coordinator inventories them to compute blast
+        radius. Optional ``statuses`` filter (e.g. ``('active',)``) narrows to
+        live rows. Never returns secrets — only operational metadata.
+        """
+        clauses: list[str] = []
+        params: list[Any] = []
+        if statuses:
+            placeholders = ", ".join("?" for _ in statuses)
+            clauses.append(f"status IN ({placeholders})")
+            params.extend(statuses)
+        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
+        sql = (
+            f"SELECT * FROM sessions {where} "
+            "ORDER BY last_heartbeat_at DESC LIMIT ?"
+        )
+        params.append(max(1, int(limit)))
+        with self._tx(immediate=False) as conn:
+            rows = conn.execute(sql, params).fetchall()
+            return [dict(r) for r in rows]
+
     # ── work items ────────────────────────────────────────────────────────
 
     def upsert_work_item(

docs/mcp-restart-coordinator.md

@@ -0,0 +1,95 @@
+# MCP restart coordinator and impact analysis (#658)
+
+Before any sanctioned MCP restart, a central coordinator evaluates the live
+control-plane state and produces an **impact preview** so operators and the web
+console (#642 / #652) can see the blast radius *before* concurrent LLM work is
+disrupted. Uncoordinated restarts destroy in-flight author/reviewer/merger work
+and give operators no way to see what they are about to break.
+
+This lands the coordinator + impact DTO + a dry-run MCP tool. It is the single
+sanctioned entry point for restart evaluation post-#657 (which inventoried the
+restart/reload/kill paths). The **mutative apply** path — actually performing a
+restart — is a later child gated by a drain proof and is explicitly out of
+scope here.
+
+## Components
+
+| Piece | Where | Responsibility |
+|-------|-------|----------------|
+| `restart_coordinator.evaluate_restart_impact` | `restart_coordinator.py` | Pure classification: inventory → impact report DTO. No I/O, no restart. |
+| `RestartImpactReport` / `SessionImpact` / `LeaseImpact` | `restart_coordinator.py` | Console-facing DTO (`.as_dict()` is JSON-serializable). |
+| `ControlPlaneDB.list_sessions` | `control_plane_db.py` | Read-only session inventory (the process-level unit a restart kills). |
+| `gitea_request_mcp_restart` | `gitea_mcp_server.py` | MCP tool: gathers inventory from the #613 DB, calls the coordinator, returns the report. Dry-run only. |
+
+## Dimensions evaluated
+
+The coordinator classifies the inventory across the dimensions #658 requires:
+
+- **Sessions** — every active MCP session; a restart terminates all of them.
+  Liveness = `status == active` **and** the owner pid is alive **and** the
+  heartbeat is fresh (default window 15 min). Dead/stale sessions do not count
+  toward blast radius.
+- **Leases / locks** — control-plane leases joined with work items and their
+  freshness (`lease_lifecycle.classify_lease_freshness`). Only `active` (live
+  owner) leases are *disruptive*; expired / released / dead-process leases never
+  withhold a restart.
+- **Issue / PR work** — the issues and PRs behind disruptive leases.
+- **Mutations / critical sections** — a live lease carrying an author worktree
+  or a mutating phase (`implementing`, `publishing`, `merging`, …) is a
+  critical section a restart must not sever.
+- **Terminal (merge) lock** — an active terminal lock always makes a restart
+  unsafe.
+- **Prior recovery attempts** — narrower recovery already tried (e.g. sanctioned
+  client reconnects) is echoed so the operator sees the escalation history.
+
+## Verdict
+
+Exactly three verdicts, matching the acceptance criteria:
+
+| Verdict | `allow_restart` | Meaning |
+|---------|-----------------|---------|
+| `safe` | `true` | No other live sessions, no live leases, no terminal lock. |
+| `unsafe` | `false` | Live work would be disrupted and no operator override is present — **or** the inventory could not be completed (fail closed). |
+| `override` | `true` | Live work present, but an operator override accepts the blast radius. |
+
+`override_would_allow` tells the console whether an override path exists for the
+current state. `blast_radius` is a `none` / `low` / `medium` / `high` severity
+band derived from the affected session and work counts.
+
+### Fail closed
+
+If the control-plane inventory cannot be completed (DB unavailable, a listing
+failed), `inventory_complete` is `false` and the verdict is `unsafe` / deny. An
+incomplete evaluation must never green-light a restart.
+
+### Operator override authority
+
+Override authority is read from the environment variable
+`GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION` and **never** from a tool
+argument. A worker session cannot set an environment variable on an
+already-running daemon, so override cannot be self-asserted (same pattern as the
+#630 daemon-maintenance authorization). The `request_override` tool argument only
+expresses caller intent; it takes effect solely when the environment
+authorization is present.
+
+## The tool
+
+```text
+gitea_request_mcp_restart(remote, host, org, repo,
+                          dry_run=True, request_override=False,
+                          session_id=None, limit=200)
+```
+
+Read-only, dry-run, and it **never restarts anything**. `apply_supported` is
+always `false`; passing `dry_run=False` performs no restart and reports that
+apply is gated by a drain proof (a separate child).
+
+## Audit
+
+Every evaluation carries an `audit_record` (event, coordinator version, verdict,
+allow decision, blast radius, counts, timestamp) so restart decisions are
+auditable. No secrets flow through the coordinator — session ids, pids, and
+profiles are operational metadata only.
+
+A representative dry-run report is in
+[`mcp-restart-impact-sample.json`](./mcp-restart-impact-sample.json).

docs/mcp-restart-impact-sample.json

@@ -0,0 +1,148 @@
+{
+  "coordinator_version": "1.0.0-issue-658",
+  "evaluated_at": "2026-07-24T06:00:00+00:00",
+  "dry_run": true,
+  "restart_performed": false,
+  "inventory_complete": true,
+  "incomplete_reasons": [],
+  "verdict": "unsafe",
+  "allow_restart": false,
+  "override_would_allow": true,
+  "operator_override": false,
+  "blast_radius": "high",
+  "reasons": [
+    "live work would be disrupted; restart denied without operator override",
+    "1 critical section(s) in flight (active lease with a live owner)"
+  ],
+  "affected_sessions": [
+    {
+      "session_id": "prgs-author-30988-d6f43c25",
+      "role": "author",
+      "profile": "prgs-author",
+      "pid": 1,
+      "status": "active",
+      "alive": true,
+      "heartbeat_stale": false,
+      "is_requester": false,
+      "live": true
+    },
+    {
+      "session_id": "prgs-reviewer-4157-0ce9",
+      "role": "reviewer",
+      "profile": "prgs-reviewer",
+      "pid": 1,
+      "status": "active",
+      "alive": true,
+      "heartbeat_stale": false,
+      "is_requester": true,
+      "live": true
+    }
+  ],
+  "affected_leases": [
+    {
+      "lease_id": "lease-abc",
+      "session_id": "prgs-author-30988-d6f43c25",
+      "role": "author",
+      "phase": "implementing",
+      "freshness": "active",
+      "work_kind": "issue",
+      "work_number": 658,
+      "worktree_path": "/repo/branches/feat-issue-658",
+      "disruptive": true,
+      "is_mutation": true,
+      "is_critical_section": true
+    },
+    {
+      "lease_id": "lease-dead",
+      "session_id": "prgs-author-91485",
+      "role": "author",
+      "phase": "allocated",
+      "freshness": "stale_dead_process",
+      "work_kind": "issue",
+      "work_number": 651,
+      "worktree_path": null,
+      "disruptive": false,
+      "is_mutation": false,
+      "is_critical_section": false
+    }
+  ],
+  "critical_sections": [
+    {
+      "lease_id": "lease-abc",
+      "session_id": "prgs-author-30988-d6f43c25",
+      "role": "author",
+      "phase": "implementing",
+      "freshness": "active",
+      "work_kind": "issue",
+      "work_number": 658,
+      "worktree_path": "/repo/branches/feat-issue-658",
+      "disruptive": true,
+      "is_mutation": true,
+      "is_critical_section": true
+    }
+  ],
+  "affected_issues": [
+    658
+  ],
+  "affected_prs": [],
+  "mutations": [
+    {
+      "lease_id": "lease-abc",
+      "session_id": "prgs-author-30988-d6f43c25",
+      "role": "author",
+      "phase": "implementing",
+      "freshness": "active",
+      "work_kind": "issue",
+      "work_number": 658,
+      "worktree_path": "/repo/branches/feat-issue-658",
+      "disruptive": true,
+      "is_mutation": true,
+      "is_critical_section": true
+    }
+  ],
+  "terminal_lock": null,
+  "ack_state": {
+    "prgs-author-30988-d6f43c25": "pending"
+  },
+  "prior_recovery_attempts": [
+    {
+      "kind": "client_reconnect",
+      "at": "2026-07-24T06:00:00+00:00",
+      "outcome": "insufficient"
+    }
+  ],
+  "counts": {
+    "sessions_total": 2,
+    "sessions_live_other": 1,
+    "leases_total": 2,
+    "leases_disruptive": 1,
+    "critical_sections": 1,
+    "mutations": 1,
+    "affected_issues": 1,
+    "affected_prs": 0,
+    "prior_recovery_attempts": 1
+  },
+  "audit_record": {
+    "event": "restart_impact_evaluated",
+    "coordinator_version": "1.0.0-issue-658",
+    "evaluated_at": "2026-07-24T06:00:00+00:00",
+    "dry_run": true,
+    "operator_override": false,
+    "requesting_session_id": "prgs-reviewer-4157-0ce9",
+    "inventory_complete": true,
+    "verdict": "unsafe",
+    "allow_restart": false,
+    "blast_radius": "high",
+    "counts": {
+      "sessions_total": 2,
+      "sessions_live_other": 1,
+      "leases_total": 2,
+      "leases_disruptive": 1,
+      "critical_sections": 1,
+      "mutations": 1,
+      "affected_issues": 1,
+      "affected_prs": 0,
+      "prior_recovery_attempts": 1
+    }
+  }
+}

docs/mcp-tool-inventory.md

@@ -135,6 +135,7 @@ that gates each call, not which tools exist.
 - `gitea_release_merger_pr_lease`
 - `gitea_release_reviewer_pr_lease`
 - `gitea_release_workflow_lease`
+- `gitea_request_mcp_restart`
 - `gitea_resolve_task_capability`
 - `gitea_resume_review_draft`
 - `gitea_review_pr`

docs/observability/analytics-instrumentation.md

@@ -0,0 +1,125 @@
+# Model Usage, Token Cost, Latency, and Workflow Analytics (Phase 4)
+
+- **Tracking Issue:** [#651](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/651)
+- **Parent Epic:** [#631](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/651)
+- **Console Surface:** `/analytics`, `/api/v1/analytics`, `/api/v1/analytics/usage`
+
+## 1. Overview
+
+The Web Console Analytics module provides durable, aggregate visibility into **model usage, token cost, latency percentiles, and workflow-stage performance** across projects, worker roles, AI models, issues, and PRs.
+
+### Non-Goals
+- No mandatory client-side telemetry that leaks prompts or secret keys.
+- No third-party payment provider or billing integration.
+- No automatic model routing changes without controller policy (#647).
+
+---
+
+## 2. Event Schema (`usage_events`)
+
+Usage metrics are stored in the control-plane database under table `usage_events`.
+
+| Column | Type | Description |
+|---|---|---|
+| `usage_id` | `INTEGER` | Primary key (autoincrement) |
+| `session_id` | `TEXT` | Optional active session identifier |
+| `remote` | `TEXT` | Known Gitea instance (`dadeschools` or `prgs`) |
+| `org` | `TEXT` | Repository owner / organization |
+| `repo` | `TEXT` | Repository name |
+| `project_id` | `TEXT` | Optional project identifier |
+| `role` | `TEXT` | Active worker role (`author`, `reviewer`, `merger`, `reconciler`, `controller`) |
+| `model` | `TEXT` | LLM model identifier (e.g. `gemini-3.6-flash`, `claude-3-5-sonnet`) |
+| `issue_number` | `INTEGER` | Correlated Gitea issue number (optional) |
+| `pr_number` | `INTEGER` | Correlated Gitea PR number (optional) |
+| `stage` | `TEXT` | Workflow stage (`preflight`, `implementation`, `review`, `merge`, `reconciliation`) |
+| `input_tokens` | `INTEGER` | Input token count (optional / nullable) |
+| `output_tokens` | `INTEGER` | Output token count (optional / nullable) |
+| `total_tokens` | `INTEGER` | Total token count (optional / nullable) |
+| `estimated_cost_usd` | `REAL` | Estimated USD cost (optional / nullable) |
+| `latency_ms` | `INTEGER` | Request latency in milliseconds (optional / nullable) |
+| `duration_ms` | `INTEGER` | Stage execution duration in milliseconds (optional / nullable) |
+| `status` | `TEXT` | Outcome status (`success`, `failure`, `timeout`) |
+| `metadata` | `TEXT` | Redacted metadata or summary string |
+| `created_at` | `TEXT` | ISO 8601 UTC timestamp |
+
+---
+
+## 3. Handling of Missing Data ("Unknown" vs. Zero Fabrication)
+
+To ensure operational metrics accurately reflect evidence:
+- **Untracked or missing metrics are displayed as `Unknown`**, never zero-fabricated.
+- If an event omits `estimated_cost_usd`, `latency_ms`, or token counts, the aggregator marks those fields as missing (`None`) rather than defaulting to `0` or `$0.00`.
+- Summary tables and KPI cards explicitly indicate when data is unmeasured or partially reported.
+
+---
+
+## 4. Redaction & Security Rules
+
+Per `#633` security policy:
+- Free-text fields (`metadata`, `prompt_summary`, `session_id`) are run through `console_redaction.redact_text` before persistence and output serialization.
+- Secret tokens, keychain commands, authorization headers, passwords, and JWTs are stripped automatically.
+
+---
+
+## 5. Opt-in Instrumentation Guide
+
+Applications, MCP servers, and background sessions can report usage metrics through either Python API or HTTP ingestion.
+
+### Python Ingestion
+
+```python
+from webui.analytics_loader import record_usage
+
+record_usage(
+    remote="dadeschools",
+    org="Scaled-Tech-Consulting",
+    repo="Gitea-Tools",
+    role="author",
+    model="gemini-3.6-flash",
+    issue_number=651,
+    stage="implementation",
+    input_tokens=1420,
+    output_tokens=380,
+    total_tokens=1800,
+    estimated_cost_usd=0.00045,
+    latency_ms=320,
+    duration_ms=4500,
+    status="success",
+    metadata={"note": "Implementation of analytics module"},
+)
+```
+
+### HTTP Ingestion API
+
+```http
+POST /api/v1/analytics/usage HTTP/1.1
+Content-Type: application/json
+
+{
+  "remote": "dadeschools",
+  "org": "Scaled-Tech-Consulting",
+  "repo": "Gitea-Tools",
+  "role": "author",
+  "model": "gemini-3.6-flash",
+  "issue_number": 651,
+  "stage": "implementation",
+  "input_tokens": 1420,
+  "output_tokens": 380,
+  "total_tokens": 1800,
+  "estimated_cost_usd": 0.00045,
+  "latency_ms": 320,
+  "duration_ms": 4500,
+  "status": "success",
+  "metadata": "Analytics schema landed"
+}
+```
+
+---
+
+## 6. Querying Analytics API
+
+```http
+GET /api/v1/analytics?role=author&stage=implementation HTTP/1.1
+```
+
+Returns `AnalyticsSnapshot` JSON containing aggregations (`by_model`, `by_stage`, `by_role`, `by_work_item`, `by_project`) and latency percentiles (`p50`, `p90`, `p95`, `p99`).

gitea_mcp_server.py

@@ -2040,6 +2040,7 @@ import dependency_graph  # noqa: E402  # #784 durable dependency edges
 import control_plane_db  # noqa: E402
 import lease_lifecycle  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
+import restart_coordinator  # noqa: E402  # #658 MCP restart coordinator/impact
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
 import sentry_incident_bridge  # noqa: E402  (#607 Sentry→Gitea incident bridge)
@@ -22050,6 +22051,156 @@ def gitea_workflow_dashboard(
     return payload
 
 
+@mcp.tool()
+def gitea_request_mcp_restart(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    dry_run: bool = True,
+    request_override: bool = False,
+    session_id: str | None = None,
+    limit: int = 200,
+) -> dict:
+    """Evaluate a proposed MCP restart and return an impact preview (#658).
+
+    Central restart coordinator: gathers live control-plane state (sessions,
+    leases/locks, in-flight issue/PR work, mutations, worktrees) and returns a
+    blast-radius impact report with a ``safe`` / ``unsafe`` / ``override``
+    verdict, so the console (#642/#652) and operators can see what a restart
+    would disrupt *before* any concurrent LLM work is destroyed.
+
+    This tool is **dry-run and never restarts anything.** The mutative apply
+    path is a separate child gated by a drain proof (non-goal here); calling
+    with ``dry_run=False`` still performs no restart and reports that apply is
+    not yet available.
+
+    Operator override authority is read from the process environment
+    (``GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION``), never self-asserted by
+    the requesting session: ``request_override`` only expresses caller intent
+    and takes effect solely when that environment authorization is present.
+
+    Fails closed: if the control-plane inventory cannot be completed, the
+    verdict is ``unsafe`` / deny (an incomplete evaluation must never green-light
+    a restart).
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "read_only": True,
+            "dry_run": True,
+            "restart_performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+    except ValueError as exc:
+        return {
+            "success": False,
+            "read_only": True,
+            "dry_run": True,
+            "restart_performed": False,
+            "reasons": [str(exc)],
+        }
+
+    inventory_complete = True
+    incomplete_reasons: list[str] = []
+    sessions: list[dict] = []
+    leases: list[dict] = []
+    terminal_lock: dict | None = None
+
+    db, db_errs = _control_plane_db_or_error()
+    if db is None:
+        inventory_complete = False
+        incomplete_reasons.extend(
+            db_errs or ["control-plane DB unavailable; cannot evaluate restart"]
+        )
+    else:
+        try:
+            sessions = db.list_sessions(statuses=("active",), limit=max(1, int(limit)))
+        except Exception as exc:  # noqa: BLE001
+            inventory_complete = False
+            incomplete_reasons.append(
+                f"session inventory failed: {_redact(str(exc))}"
+            )
+        try:
+            lease_result = lease_lifecycle.list_active_leases(
+                db,
+                remote=remote if remote in REMOTES else remote,
+                org=o,
+                repo=r,
+                role=None,
+                include_non_active=False,
+                limit=max(1, int(limit)),
+            )
+            leases = list(lease_result.get("leases") or [])
+        except Exception as exc:  # noqa: BLE001
+            inventory_complete = False
+            incomplete_reasons.append(
+                f"lease inventory failed: {_redact(str(exc))}"
+            )
+        try:
+            terminal = db.get_active_terminal_lock(
+                remote=remote if remote in REMOTES else remote,
+                org=o,
+                repo=r,
+            )
+            if terminal:
+                terminal_lock = dict(terminal)
+        except Exception as exc:  # noqa: BLE001
+            inventory_complete = False
+            incomplete_reasons.append(
+                f"terminal lock lookup failed: {_redact(str(exc))}"
+            )
+
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip() or "session"
+    sid = (session_id or "").strip() or f"{profile_name}-{os.getpid()}"
+
+    # Override authority is read from the environment only — a worker session
+    # cannot set an env var for an already-running daemon, so it cannot be
+    # self-asserted the way a tool argument could (#630/#710 F1 pattern).
+    operator_authorized = bool(
+        (os.environ.get("GITEA_OPERATOR_RESTART_OVERRIDE_AUTHORIZATION") or "").strip()
+    )
+    operator_override = bool(request_override and operator_authorized)
+
+    inventory = {
+        "sessions": sessions,
+        "leases": leases,
+        "terminal_lock": terminal_lock,
+        "inventory_complete": inventory_complete,
+        "incomplete_reasons": incomplete_reasons,
+    }
+
+    report = restart_coordinator.evaluate_restart_impact(
+        inventory,
+        operator_override=operator_override,
+        requesting_session_id=sid,
+        dry_run=True,  # coordinator is always analysis-only (#658)
+    )
+
+    payload = report.as_dict()
+    payload["success"] = True
+    payload["read_only"] = True
+    payload["remote"] = remote
+    payload["org"] = o
+    payload["repo"] = r
+    payload["requesting_session_id"] = sid
+    payload["operator_override_requested"] = bool(request_override)
+    payload["operator_override_authorized"] = operator_authorized
+    payload["apply_supported"] = False
+    if not dry_run:
+        payload["reasons"] = list(payload.get("reasons") or []) + [
+            "apply requested but not supported: sanctioned restart apply is "
+            "gated by a drain proof (separate child); no restart performed (#658)"
+        ]
+    return payload
+
+
 @mcp.tool()
 def gitea_inspect_workflow_lease(
     lease_id: str,

pr_work_lease.py

@@ -228,25 +228,74 @@ def find_active_reviewer_lease(
     return None
 
 
+def _conflict_fix_chain_key(lease: dict) -> tuple | None:
+    """Identity of the lease chain a conflict-fix marker belongs to (#842).
+
+    Keyed by PR number, profile, head_before, and branch. Returns None when any
+    required component (pr_number, profile, head_before) is missing or malformed.
+    """
+    raw = lease.get("raw_fields") or {}
+    pr_number = lease.get("pr_number")
+    profile = (lease.get("profile") or "").strip().lower()
+    head_before = lease.get("head_before")
+    branch = (lease.get("branch") or raw.get("branch") or "").strip()
+    if not (pr_number and profile and head_before):
+        return None
+    return (pr_number, profile, head_before, branch)
+
+
+def _conflict_fix_chain_matches(key1: tuple, key2: tuple) -> bool:
+    """True when two conflict-fix chain keys refer to the same lease chain."""
+    pr1, profile1, head1, branch1 = key1
+    pr2, profile2, head2, branch2 = key2
+    if pr1 != pr2 or profile1 != profile2 or head1 != head2:
+        return False
+    if branch1 and branch2 and branch1 != branch2:
+        return False
+    return True
+
+
+def _conflict_fix_chain_terminated_after(entries: list[dict], index: int) -> bool:
+    """True when a later marker terminates the conflict-fix chain of ``entries[index]``.
+
+    Append-only newest-wins: a terminal marker (phase=released/blocked/done)
+    ends only its matching claim chain (#842).
+    """
+    key = _conflict_fix_chain_key(entries[index])
+    if key is None:
+        return False
+    for later in entries[index + 1:]:
+        phase = (later.get("phase") or "").strip().lower()
+        if phase not in _TERMINAL_CONFLICT_FIX_PHASES:
+            continue
+        later_key = _conflict_fix_chain_key(later)
+        if later_key and _conflict_fix_chain_matches(key, later_key):
+            return True
+    return False
+
+
 def find_active_conflict_fix_lease(
     comments: list[dict],
     *,
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Return the newest unexpired conflict-fix lease for *pr_number*, if any."""
+    """Return the newest unexpired, non-terminated conflict-fix lease for *pr_number*, if any."""
     now = now or datetime.now(timezone.utc)
     candidates = [
         entry for entry in _comment_entries(comments, pr_number=pr_number)
         if entry.get("lease_kind") == "conflict_fix"
     ]
-    for lease in reversed(candidates):
+    for index in range(len(candidates) - 1, -1, -1):
+        lease = candidates[index]
         if _lease_expired(lease, now=now):
             continue
         phase = (lease.get("phase") or "").strip().lower()
         if phase in _TERMINAL_CONFLICT_FIX_PHASES:
             continue
         if phase in _ACTIVE_CONFLICT_FIX_PHASES or phase:
+            if _conflict_fix_chain_terminated_after(candidates, index):
+                continue
             return lease
     return None
 

restart_coordinator.py

@@ -0,0 +1,451 @@
+"""MCP restart coordinator and impact analysis (#658).
+
+Before any sanctioned MCP restart, a central coordinator must evaluate the
+live control-plane state — active sessions, leases/locks, in-flight issue/PR
+work, mutations, worktrees, and recovery history — and produce an *impact
+preview* so operators (and the web console, #642/#652) can see the blast
+radius **before** concurrent LLM work is disrupted.
+
+Design rules (mirrors the read-only posture of ``workflow_dashboard`` /
+``lease_lifecycle``):
+
+* **Pure classification.** :func:`evaluate_restart_impact` takes an already
+  gathered inventory and returns a structured report. It never touches the
+  network, the filesystem, or a live process, so multi-session fixtures can
+  drive every branch in unit tests. The coordinator *never restarts anything*;
+  a mutative apply path is a later child gated by a drain proof (non-goal here).
+* **Fail closed.** If the inventory is not explicitly complete, the verdict is
+  ``unsafe`` / deny — an incomplete evaluation must never green-light a restart.
+* **No secrets.** Session ids, pids, and profiles are operational metadata, not
+  credentials; nothing secret flows through this module.
+
+The single sanctioned entry point post-#657 is the MCP tool
+``gitea_request_mcp_restart`` (dry-run by default), which gathers the inventory
+from the #613 control-plane DB and calls :func:`evaluate_restart_impact`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Mapping, Sequence
+
+import lease_lifecycle
+
+COORDINATOR_VERSION = "1.0.0-issue-658"
+
+# Restart verdicts. Exactly the three the acceptance criteria name.
+VERDICT_SAFE = "safe"
+VERDICT_UNSAFE = "unsafe"
+VERDICT_OVERRIDE = "override"
+
+# Blast-radius severity bands.
+BLAST_NONE = "none"
+BLAST_LOW = "low"
+BLAST_MEDIUM = "medium"
+BLAST_HIGH = "high"
+
+# A live lease with a live owner process is treated as active in-flight work.
+LEASE_FRESHNESS_LIVE = "active"
+
+# Default staleness window for a session heartbeat (seconds). A session whose
+# last heartbeat is older than this is not counted as live even if its row is
+# still marked ``active`` — it is assumed dead/detached.
+DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS = 900
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _parse_ts(value: str | None) -> datetime | None:
+    return lease_lifecycle._parse_ts(value)
+
+
+@dataclass(frozen=True)
+class SessionImpact:
+    """One MCP session a restart would terminate."""
+
+    session_id: str
+    role: str | None
+    profile: str | None
+    pid: int | None
+    status: str | None
+    alive: bool | None
+    heartbeat_stale: bool
+    is_requester: bool
+    live: bool
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "session_id": self.session_id,
+            "role": self.role,
+            "profile": self.profile,
+            "pid": self.pid,
+            "status": self.status,
+            "alive": self.alive,
+            "heartbeat_stale": self.heartbeat_stale,
+            "is_requester": self.is_requester,
+            "live": self.live,
+        }
+
+
+@dataclass(frozen=True)
+class LeaseImpact:
+    """One control-plane lease a restart would disrupt."""
+
+    lease_id: str | None
+    session_id: str | None
+    role: str | None
+    phase: str | None
+    freshness: str | None
+    work_kind: str | None
+    work_number: int | None
+    worktree_path: str | None
+    disruptive: bool
+    is_mutation: bool
+    is_critical_section: bool
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "lease_id": self.lease_id,
+            "session_id": self.session_id,
+            "role": self.role,
+            "phase": self.phase,
+            "freshness": self.freshness,
+            "work_kind": self.work_kind,
+            "work_number": self.work_number,
+            "worktree_path": self.worktree_path,
+            "disruptive": self.disruptive,
+            "is_mutation": self.is_mutation,
+            "is_critical_section": self.is_critical_section,
+        }
+
+
+@dataclass(frozen=True)
+class RestartImpactReport:
+    """Impact preview DTO returned to the console / operator (#642/#652)."""
+
+    coordinator_version: str
+    evaluated_at: str
+    dry_run: bool
+    restart_performed: bool
+    inventory_complete: bool
+    verdict: str
+    allow_restart: bool
+    override_would_allow: bool
+    operator_override: bool
+    blast_radius: str
+    reasons: list[str]
+    affected_sessions: list[SessionImpact]
+    affected_leases: list[LeaseImpact]
+    critical_sections: list[LeaseImpact]
+    affected_issues: list[int]
+    affected_prs: list[int]
+    mutations: list[LeaseImpact]
+    terminal_lock: dict[str, Any] | None
+    ack_state: dict[str, str]
+    prior_recovery_attempts: list[dict[str, Any]]
+    counts: dict[str, int]
+    audit_record: dict[str, Any]
+    incomplete_reasons: list[str] = field(default_factory=list)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "coordinator_version": self.coordinator_version,
+            "evaluated_at": self.evaluated_at,
+            "dry_run": self.dry_run,
+            "restart_performed": self.restart_performed,
+            "inventory_complete": self.inventory_complete,
+            "incomplete_reasons": list(self.incomplete_reasons),
+            "verdict": self.verdict,
+            "allow_restart": self.allow_restart,
+            "override_would_allow": self.override_would_allow,
+            "operator_override": self.operator_override,
+            "blast_radius": self.blast_radius,
+            "reasons": list(self.reasons),
+            "affected_sessions": [s.as_dict() for s in self.affected_sessions],
+            "affected_leases": [l.as_dict() for l in self.affected_leases],
+            "critical_sections": [l.as_dict() for l in self.critical_sections],
+            "affected_issues": list(self.affected_issues),
+            "affected_prs": list(self.affected_prs),
+            "mutations": [l.as_dict() for l in self.mutations],
+            "terminal_lock": self.terminal_lock,
+            "ack_state": dict(self.ack_state),
+            "prior_recovery_attempts": list(self.prior_recovery_attempts),
+            "counts": dict(self.counts),
+            "audit_record": dict(self.audit_record),
+        }
+
+
+def _classify_session(
+    row: Mapping[str, Any],
+    *,
+    now: datetime,
+    requesting_session_id: str | None,
+    heartbeat_stale_seconds: int,
+) -> SessionImpact:
+    session_id = str(row.get("session_id") or "")
+    pid = row.get("pid")
+    status = (row.get("status") or "").strip().lower() or None
+    alive = lease_lifecycle.is_process_alive(pid) if pid is not None else None
+    hb = _parse_ts(row.get("last_heartbeat_at"))
+    heartbeat_stale = bool(
+        hb is not None and (now - hb).total_seconds() > heartbeat_stale_seconds
+    )
+    live = bool(status == "active" and alive is not False and not heartbeat_stale)
+    return SessionImpact(
+        session_id=session_id,
+        role=row.get("role"),
+        profile=row.get("profile"),
+        pid=pid,
+        status=status,
+        alive=alive,
+        heartbeat_stale=heartbeat_stale,
+        is_requester=bool(
+            requesting_session_id and session_id == requesting_session_id
+        ),
+        live=live,
+    )
+
+
+# Lease phases that represent an active mutation in flight (as opposed to a
+# mere allocation/claim with no work committed yet). An active lease in any of
+# these phases is a critical section a restart must not sever.
+_MUTATING_PHASES = frozenset(
+    {
+        "implementing",
+        "publishing",
+        "pushing",
+        "committing",
+        "reviewing",
+        "merging",
+        "reconciling",
+        "conflict_fix",
+    }
+)
+
+
+def _classify_lease(row: Mapping[str, Any]) -> LeaseImpact:
+    freshness_obj = row.get("freshness")
+    if isinstance(freshness_obj, Mapping):
+        freshness = str(freshness_obj.get("freshness") or "").strip().lower() or None
+    else:
+        freshness = str(freshness_obj or "").strip().lower() or None
+    phase = (row.get("phase") or "").strip().lower() or None
+    worktree = row.get("worktree_path")
+    disruptive = freshness == LEASE_FRESHNESS_LIVE
+    # A live lease is a mutation-in-flight if it carries an author worktree or
+    # its phase names a mutating step. All disruptive leases are critical
+    # sections a restart would sever regardless.
+    is_mutation = bool(
+        disruptive and (bool(worktree) or (phase in _MUTATING_PHASES))
+    )
+    number = row.get("work_number")
+    try:
+        number = int(number) if number is not None else None
+    except (TypeError, ValueError):
+        number = None
+    return LeaseImpact(
+        lease_id=row.get("lease_id"),
+        session_id=row.get("session_id"),
+        role=row.get("role"),
+        phase=phase,
+        freshness=freshness,
+        work_kind=(str(row.get("work_kind") or "").strip().lower() or None),
+        work_number=number,
+        worktree_path=worktree,
+        disruptive=disruptive,
+        is_mutation=is_mutation,
+        is_critical_section=disruptive,
+    )
+
+
+def _blast_radius(*, session_count: int, work_count: int, mutation_count: int) -> str:
+    if mutation_count > 0 or work_count >= 3 or session_count >= 3:
+        return BLAST_HIGH
+    if work_count > 0 or session_count == 2:
+        return BLAST_MEDIUM
+    if session_count == 1:
+        return BLAST_LOW
+    return BLAST_NONE
+
+
+def evaluate_restart_impact(
+    inventory: Mapping[str, Any],
+    *,
+    now: datetime | None = None,
+    operator_override: bool = False,
+    requesting_session_id: str | None = None,
+    dry_run: bool = True,
+    session_heartbeat_stale_seconds: int = DEFAULT_SESSION_HEARTBEAT_STALE_SECONDS,
+) -> RestartImpactReport:
+    """Evaluate a proposed MCP restart and return an impact preview.
+
+    ``inventory`` is a mapping with:
+
+    * ``sessions`` — session rows (session_id, role, profile, pid, status,
+      last_heartbeat_at).
+    * ``leases`` — control-plane lease rows, each ideally carrying an enriched
+      ``freshness`` dict (as :func:`lease_lifecycle.list_active_leases` returns);
+      a bare string freshness is also accepted.
+    * ``terminal_lock`` — the active terminal (merge) lock, if any.
+    * ``prior_recovery_attempts`` — narrower recovery attempts already tried
+      (e.g. sanctioned reconnects) so the operator sees escalation history.
+    * ``inventory_complete`` — bool. **Must** be explicitly True; a missing or
+      falsy value forces a deny (fail closed).
+    * ``incomplete_reasons`` — optional reasons the inventory is incomplete.
+
+    The coordinator never restarts anything: ``restart_performed`` is always
+    False and the mutative apply path is a later drain-gated child.
+    """
+    moment = now or _utc_now()
+    reasons: list[str] = []
+
+    inventory_complete = bool(inventory.get("inventory_complete", False))
+    incomplete_reasons = [str(r) for r in (inventory.get("incomplete_reasons") or [])]
+
+    sessions_raw: Sequence[Mapping[str, Any]] = inventory.get("sessions") or []
+    leases_raw: Sequence[Mapping[str, Any]] = inventory.get("leases") or []
+    terminal_lock = inventory.get("terminal_lock") or None
+    prior_recovery_attempts = [
+        dict(a) for a in (inventory.get("prior_recovery_attempts") or [])
+    ]
+
+    session_impacts = [
+        _classify_session(
+            s,
+            now=moment,
+            requesting_session_id=requesting_session_id,
+            heartbeat_stale_seconds=session_heartbeat_stale_seconds,
+        )
+        for s in sessions_raw
+    ]
+    lease_impacts = [_classify_lease(l) for l in leases_raw]
+
+    # Only *other* live sessions and live leases constitute blast radius: a
+    # restart that would kill only the requesting session with no other work in
+    # flight is safe.
+    other_live_sessions = [
+        s for s in session_impacts if s.live and not s.is_requester
+    ]
+    disruptive_leases = [l for l in lease_impacts if l.disruptive]
+    critical_sections = [l for l in lease_impacts if l.is_critical_section]
+    mutations = [l for l in lease_impacts if l.is_mutation]
+
+    affected_issues = sorted(
+        {
+            l.work_number
+            for l in disruptive_leases
+            if l.work_kind == "issue" and l.work_number is not None
+        }
+    )
+    affected_prs = sorted(
+        {
+            l.work_number
+            for l in disruptive_leases
+            if l.work_kind == "pr" and l.work_number is not None
+        }
+    )
+
+    disruptive = bool(disruptive_leases or other_live_sessions or terminal_lock)
+
+    if not inventory_complete:
+        verdict = VERDICT_UNSAFE
+        allow_restart = False
+        reasons.append(
+            "inventory incomplete: restart evaluation cannot confirm blast "
+            "radius — deny (fail closed, #658)"
+        )
+        reasons.extend(incomplete_reasons)
+    elif not disruptive:
+        verdict = VERDICT_SAFE
+        allow_restart = True
+        reasons.append("no other live sessions, live leases, or terminal lock")
+    elif operator_override:
+        verdict = VERDICT_OVERRIDE
+        allow_restart = True
+        reasons.append(
+            "live work present; operator override accepts the blast radius"
+        )
+    else:
+        verdict = VERDICT_UNSAFE
+        allow_restart = False
+        reasons.append(
+            "live work would be disrupted; restart denied without operator "
+            "override"
+        )
+
+    if critical_sections and inventory_complete:
+        reasons.append(
+            f"{len(critical_sections)} critical section(s) in flight "
+            "(active lease with a live owner)"
+        )
+    if terminal_lock:
+        reasons.append("active terminal (merge) lock present")
+
+    override_would_allow = bool(inventory_complete and disruptive)
+
+    blast_radius = _blast_radius(
+        session_count=len(other_live_sessions),
+        work_count=len(affected_issues) + len(affected_prs),
+        mutation_count=len(mutations),
+    )
+
+    # Acknowledgement is a later child (drain protocol); expose per-session
+    # placeholders so the console can render the ack column now.
+    ack_state = {s.session_id: "pending" for s in other_live_sessions}
+
+    counts = {
+        "sessions_total": len(session_impacts),
+        "sessions_live_other": len(other_live_sessions),
+        "leases_total": len(lease_impacts),
+        "leases_disruptive": len(disruptive_leases),
+        "critical_sections": len(critical_sections),
+        "mutations": len(mutations),
+        "affected_issues": len(affected_issues),
+        "affected_prs": len(affected_prs),
+        "prior_recovery_attempts": len(prior_recovery_attempts),
+    }
+
+    audit_record = {
+        "event": "restart_impact_evaluated",
+        "coordinator_version": COORDINATOR_VERSION,
+        "evaluated_at": moment.isoformat(),
+        "dry_run": dry_run,
+        "operator_override": bool(operator_override),
+        "requesting_session_id": requesting_session_id,
+        "inventory_complete": inventory_complete,
+        "verdict": verdict,
+        "allow_restart": allow_restart,
+        "blast_radius": blast_radius,
+        "counts": counts,
+    }
+
+    return RestartImpactReport(
+        coordinator_version=COORDINATOR_VERSION,
+        evaluated_at=moment.isoformat(),
+        dry_run=dry_run,
+        restart_performed=False,
+        inventory_complete=inventory_complete,
+        verdict=verdict,
+        allow_restart=allow_restart,
+        override_would_allow=override_would_allow,
+        operator_override=bool(operator_override),
+        blast_radius=blast_radius,
+        reasons=reasons,
+        affected_sessions=session_impacts,
+        affected_leases=lease_impacts,
+        critical_sections=critical_sections,
+        affected_issues=affected_issues,
+        affected_prs=affected_prs,
+        mutations=mutations,
+        terminal_lock=dict(terminal_lock)
+        if isinstance(terminal_lock, Mapping)
+        else terminal_lock,
+        ack_state=ack_state,
+        prior_recovery_attempts=prior_recovery_attempts,
+        counts=counts,
+        audit_record=audit_record,
+        incomplete_reasons=incomplete_reasons,
+    )

tests/test_control_plane_db.py

@@ -36,7 +36,7 @@ class ControlPlaneDBTest(unittest.TestCase):
             rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
         finally:
             conn.close()
-        self.assertEqual(rows["schema_version"], "4")
+        self.assertEqual(rows["schema_version"], "5")
         self.assertIn("DB coordinates", rows["architecture"])
         self.assertIn("bridge", rows["architecture"].lower())
 

tests/test_pr_work_lease.py

@@ -19,6 +19,7 @@ from pr_work_lease import (  # noqa: E402
     assess_reviewer_mutation_blocked,
     assess_reviewer_stale_head_final_report,
     format_conflict_fix_lease_body,
+    find_active_conflict_fix_lease,
     parse_conflict_fix_lease_comment,
     parse_reviewer_lease_comment,
 )
@@ -203,5 +204,157 @@ class TestFormatLease(unittest.TestCase):
         self.assertEqual(parsed["pr_number"], 376)
 
 
+class TestConflictFixLeaseLifecycle(unittest.TestCase):
+    def test_claim_followed_by_matching_release(self):
+        claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
+        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+        release_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "branch: feat/fix-376",
+            "worktree: branches/fix-376",
+            "profile: prgs-author",
+            "phase: released",
+            f"head_before: {HEAD_A}",
+            f"head_after: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        comments = [{"body": claim_body}, {"body": release_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNone(lease)
+
+    def test_expired_claim_without_release(self):
+        past_expires = (NOW - timedelta(minutes=10)).isoformat().replace("+00:00", "Z")
+        claim_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "phase: claimed",
+            f"head_before: {HEAD_A}",
+            f"expires_at: {past_expires}",
+            "profile: prgs-author",
+        ])
+        comments = [{"body": claim_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNone(lease)
+
+    def test_mismatched_release_different_head(self):
+        claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
+        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+        release_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "profile: prgs-author",
+            "phase: released",
+            f"head_before: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        comments = [{"body": claim_body}, {"body": release_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease["phase"], "claimed")
+
+    def test_mismatched_release_different_branch(self):
+        claim_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "branch: feat/branch-A",
+            "phase: claimed",
+            f"head_before: {HEAD_A}",
+            f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
+            "profile: prgs-author",
+        ])
+        release_body = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "branch: feat/branch-B",
+            "phase: released",
+            f"head_before: {HEAD_A}",
+            f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
+            "profile: prgs-author",
+        ])
+        comments = [{"body": claim_body}, {"body": release_body}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease["phase"], "claimed")
+
+    def test_release_followed_by_newer_claim(self):
+        claim_1 = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
+        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
+        release_1 = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "profile: prgs-author",
+            "phase: released",
+            f"head_before: {HEAD_A}",
+            f"head_after: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        claim_2 = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "profile: prgs-author",
+            "phase: claimed",
+            f"head_before: {HEAD_B}",
+            f"expires_at: {expires}",
+        ])
+        comments = [{"body": claim_1}, {"body": release_1}, {"body": claim_2}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease["head_before"], HEAD_B)
+
+    def test_malformed_or_ambiguous_markers(self):
+        malformed_release = "\n".join([
+            CONFLICT_FIX_LEASE_MARKER,
+            "pr: #376",
+            "phase: released",
+            # missing head_before and profile
+        ])
+        claim_body = _conflict_fix_body(phase="claimed")
+        comments = [{"body": claim_body}, {"body": malformed_release}]
+        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
+        self.assertIsNotNone(lease)
+
+    def test_pr818_historical_sequence(self):
+        comment_14696 = "\n".join([
+            "<!-- mcp-conflict-fix-lease:v1 -->",
+            "pr: #818",
+            "branch: feat/issue-638-webui-app-shell-phase1",
+            "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
+            "profile: prgs-author",
+            "session_id: unknown",
+            "phase: claimed",
+            "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
+            "expires_at: 2026-07-23T07:12:13Z",
+            "reviewer_active: no",
+        ])
+        comment_14730 = "\n".join([
+            "<!-- mcp-conflict-fix-lease:v1 -->",
+            "pr: #818",
+            "branch: feat/issue-638-webui-app-shell-phase1",
+            "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
+            "profile: prgs-author",
+            "session_id: prgs-author-61241-e5129c60",
+            "phase: released",
+            "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
+            "head_after: 64b6eb5d5402663098de5ded3b0617cc3b3df98f",
+            "expires_at: 2026-07-23T06:05:00Z",
+            "reviewer_active: no",
+        ])
+        comments = [{"body": comment_14696}, {"body": comment_14730}]
+        check_now = datetime(2026, 7, 23, 6, 30, tzinfo=timezone.utc)
+        lease = find_active_conflict_fix_lease(comments, pr_number=818, now=check_now)
+        self.assertIsNone(lease)
+        
+        reviewer_gate = assess_reviewer_mutation_blocked(
+            pr_number=818,
+            comments=comments,
+            reviewed_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
+            live_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
+            mutation="approve",
+            now=check_now,
+        )
+        self.assertTrue(reviewer_gate["mutation_allowed"])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_restart_coordinator.py

@@ -0,0 +1,340 @@
+"""Tests for the MCP restart coordinator and impact analysis (#658).
+
+Multi-session fixtures exercise every verdict branch: safe, unsafe (live work),
+override, and the fail-closed deny on incomplete inventory. Also covers the
+critical-section deny path and the new ``ControlPlaneDB.list_sessions``.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+
+import restart_coordinator as rc
+from control_plane_db import ControlPlaneDB
+
+
+NOW = datetime(2026, 7, 24, 6, 0, 0, tzinfo=timezone.utc)
+
+
+def _ts(dt: datetime) -> str:
+    return dt.isoformat()
+
+
+def _live_pid() -> int:
+    return os.getpid()
+
+
+def _dead_pid() -> int:
+    # A pid that is essentially never alive. os.kill(0) on it raises
+    # ProcessLookupError → is_process_alive False.
+    return 2_000_000_000
+
+
+def _session(session_id, *, pid, status="active", heartbeat=None, role="author"):
+    return {
+        "session_id": session_id,
+        "role": role,
+        "profile": "prgs-author",
+        "pid": pid,
+        "status": status,
+        "last_heartbeat_at": _ts(heartbeat or NOW),
+    }
+
+
+def _lease(
+    lease_id,
+    *,
+    session_id,
+    freshness,
+    kind="issue",
+    number=658,
+    phase="allocated",
+    worktree=None,
+    role="author",
+):
+    return {
+        "lease_id": lease_id,
+        "session_id": session_id,
+        "role": role,
+        "phase": phase,
+        "work_kind": kind,
+        "work_number": number,
+        "worktree_path": worktree,
+        "freshness": {"freshness": freshness},
+    }
+
+
+class EvaluateRestartImpactTest(unittest.TestCase):
+    def test_incomplete_inventory_denies_fail_closed(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {"inventory_complete": False, "incomplete_reasons": ["db down"]},
+            now=NOW,
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
+        self.assertFalse(report.allow_restart)
+        self.assertFalse(report.restart_performed)
+        self.assertIn("db down", report.incomplete_reasons)
+        self.assertTrue(
+            any("fail closed" in reasoning for reasoning in report.reasons)
+        )
+
+    def test_missing_completeness_flag_denies(self) -> None:
+        # No inventory_complete key at all → treated as incomplete.
+        report = rc.evaluate_restart_impact({}, now=NOW)
+        self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
+        self.assertFalse(report.allow_restart)
+
+    def test_no_other_work_is_safe(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [_session("requester", pid=_live_pid())],
+                "leases": [],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_SAFE)
+        self.assertTrue(report.allow_restart)
+        self.assertEqual(report.blast_radius, rc.BLAST_NONE)
+        self.assertEqual(report.affected_issues, [])
+
+    def test_dead_foreign_session_and_lease_are_not_disruptive(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [
+                    _session("requester", pid=_live_pid()),
+                    _session("dead", pid=_dead_pid()),
+                ],
+                "leases": [
+                    _lease("l-dead", session_id="dead", freshness="stale_dead_process")
+                ],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_SAFE)
+        self.assertTrue(report.allow_restart)
+        self.assertEqual(report.counts["leases_disruptive"], 0)
+        self.assertEqual(report.counts["sessions_live_other"], 0)
+
+    def test_live_foreign_lease_denies_without_override(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [
+                    _session("requester", pid=_live_pid()),
+                    _session("worker", pid=_live_pid()),
+                ],
+                "leases": [
+                    _lease(
+                        "l1",
+                        session_id="worker",
+                        freshness="active",
+                        worktree="/tmp/wt-658",
+                        phase="implementing",
+                    )
+                ],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
+        self.assertFalse(report.allow_restart)
+        # Critical section detected: active lease with a live owner.
+        self.assertEqual(len(report.critical_sections), 1)
+        self.assertEqual(report.affected_issues, [658])
+        self.assertEqual(report.counts["mutations"], 1)
+        self.assertTrue(report.override_would_allow)
+        self.assertEqual(report.blast_radius, rc.BLAST_HIGH)
+        # Placeholder ack state for the affected session.
+        self.assertEqual(report.ack_state.get("worker"), "pending")
+
+    def test_operator_override_allows_despite_live_work(self) -> None:
+        inv = {
+            "inventory_complete": True,
+            "sessions": [
+                _session("requester", pid=_live_pid()),
+                _session("worker", pid=_live_pid()),
+            ],
+            "leases": [_lease("l1", session_id="worker", freshness="active")],
+        }
+        report = rc.evaluate_restart_impact(
+            inv,
+            now=NOW,
+            requesting_session_id="requester",
+            operator_override=True,
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_OVERRIDE)
+        self.assertTrue(report.allow_restart)
+        self.assertFalse(report.restart_performed)
+
+    def test_deny_when_critical_section_open(self) -> None:
+        # A single live author lease in a mutating phase is a critical section
+        # that must deny an un-overridden restart.
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [_session("worker", pid=_live_pid())],
+                "leases": [
+                    _lease(
+                        "l1",
+                        session_id="worker",
+                        freshness="active",
+                        phase="merging",
+                        kind="pr",
+                        number=900,
+                    )
+                ],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
+        self.assertFalse(report.allow_restart)
+        self.assertEqual(report.affected_prs, [900])
+        self.assertEqual(len(report.critical_sections), 1)
+
+    def test_terminal_lock_makes_restart_unsafe(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [_session("requester", pid=_live_pid())],
+                "leases": [],
+                "terminal_lock": {"terminal_pr": 812},
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
+        self.assertFalse(report.allow_restart)
+        self.assertIsNotNone(report.terminal_lock)
+        self.assertTrue(
+            any("terminal" in reasoning for reasoning in report.reasons)
+        )
+
+    def test_other_live_session_without_lease_is_disruptive(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [
+                    _session("requester", pid=_live_pid()),
+                    _session("idle-but-live", pid=_live_pid()),
+                ],
+                "leases": [],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
+        self.assertEqual(report.counts["sessions_live_other"], 1)
+
+    def test_stale_heartbeat_session_not_counted_live(self) -> None:
+        stale = NOW - timedelta(hours=2)
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [
+                    _session("requester", pid=_live_pid()),
+                    _session("stale", pid=_live_pid(), heartbeat=stale),
+                ],
+                "leases": [],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.verdict, rc.VERDICT_SAFE)
+        self.assertEqual(report.counts["sessions_live_other"], 0)
+
+    def test_prior_recovery_attempts_echoed(self) -> None:
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [_session("requester", pid=_live_pid())],
+                "leases": [],
+                "prior_recovery_attempts": [
+                    {"kind": "client_reconnect", "at": _ts(NOW)}
+                ],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(len(report.prior_recovery_attempts), 1)
+        self.assertEqual(report.counts["prior_recovery_attempts"], 1)
+
+    def test_bare_string_freshness_accepted(self) -> None:
+        lease = _lease("l1", session_id="worker", freshness="active")
+        lease["freshness"] = "active"  # bare string, not a dict
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [_session("worker", pid=_live_pid())],
+                "leases": [lease],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.counts["leases_disruptive"], 1)
+
+    def test_as_dict_is_serializable_dto(self) -> None:
+        import json
+
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": [_session("requester", pid=_live_pid())],
+                "leases": [],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        payload = report.as_dict()
+        # Round-trips through JSON — safe for the console DTO.
+        encoded = json.dumps(payload)
+        decoded = json.loads(encoded)
+        self.assertEqual(decoded["verdict"], rc.VERDICT_SAFE)
+        self.assertIn("audit_record", decoded)
+        self.assertEqual(decoded["audit_record"]["event"], "restart_impact_evaluated")
+        self.assertFalse(decoded["restart_performed"])
+        self.assertIn("coordinator_version", decoded)
+
+
+class ListSessionsTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_list_sessions_filters_by_status(self) -> None:
+        self.db.upsert_session(session_id="a", role="author", pid=1, status="active")
+        self.db.upsert_session(session_id="b", role="author", pid=2, status="ended")
+        active = self.db.list_sessions(statuses=("active",))
+        ids = {row["session_id"] for row in active}
+        self.assertEqual(ids, {"a"})
+        every = self.db.list_sessions()
+        self.assertEqual({row["session_id"] for row in every}, {"a", "b"})
+
+    def test_list_sessions_feeds_coordinator(self) -> None:
+        self.db.upsert_session(
+            session_id="requester", role="author", pid=os.getpid(), status="active"
+        )
+        report = rc.evaluate_restart_impact(
+            {
+                "inventory_complete": True,
+                "sessions": self.db.list_sessions(statuses=("active",)),
+                "leases": [],
+            },
+            now=NOW,
+            requesting_session_id="requester",
+        )
+        self.assertEqual(report.counts["sessions_total"], 1)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/test_webui_analytics.py

@@ -0,0 +1,196 @@
+"""Unit and integration tests for Model Usage & Performance Analytics (#651)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from starlette.testclient import TestClient
+
+import control_plane_db
+from webui.analytics_loader import (
+    ANALYTICS_SCHEMA_VERSION,
+    compute_percentile,
+    load_analytics,
+    record_usage,
+)
+from webui.app import create_app
+from webui import console_redaction
+
+
+class AnalyticsLoaderTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.temp_dir.name, "test_control_plane.sqlite3")
+        os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
+        self.db = control_plane_db.ControlPlaneDB(db_path=self.db_path)
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def test_compute_percentile(self) -> None:
+        self.assertIsNone(compute_percentile([], 50.0))
+        self.assertEqual(compute_percentile([100], 50.0), 100.0)
+
+        # 2 elements: [100, 200]
+        self.assertEqual(compute_percentile([100, 200], 50.0), 150.0)
+
+        # 100 elements: 1..100
+        vals = list(range(1, 101))
+        self.assertEqual(compute_percentile(vals, 50.0), 50.5)
+        self.assertAlmostEqual(compute_percentile(vals, 90.0), 90.1)
+
+    def test_record_and_aggregate_usage(self) -> None:
+        # Record event 1 (complete data)
+        u1 = record_usage(
+            db_path=self.db_path,
+            remote="dadeschools",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            role="author",
+            model="gemini-3.6-flash",
+            issue_number=651,
+            stage="implementation",
+            input_tokens=1000,
+            output_tokens=500,
+            estimated_cost_usd=0.0015,
+            latency_ms=200,
+            duration_ms=3000,
+            metadata={"secret_key": "secret123", "note": "token=secret123"},
+        )
+        self.assertGreater(u1, 0)
+
+        # Record event 2 (missing tokens and cost -> unknown)
+        u2 = record_usage(
+            db_path=self.db_path,
+            remote="dadeschools",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            role="reviewer",
+            model="claude-3-5-sonnet",
+            pr_number=846,
+            stage="review",
+            latency_ms=500,
+            duration_ms=6000,
+        )
+        self.assertGreater(u2, u1)
+
+        snapshot = load_analytics(
+            db_path=self.db_path,
+            remote="dadeschools",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+
+        self.assertTrue(snapshot.ok)
+        self.assertEqual(snapshot.schema_version, ANALYTICS_SCHEMA_VERSION)
+        self.assertEqual(snapshot.total_events, 2)
+
+        # Verify overall summary
+        summary = snapshot.overall_summary
+        self.assertEqual(summary.total_events, 2)
+        self.assertEqual(summary.events_with_tokens, 1)
+        self.assertEqual(summary.total_tokens, 1500)
+        self.assertEqual(summary.events_with_cost, 1)
+        self.assertEqual(summary.estimated_cost_usd, 0.0015)
+        self.assertEqual(summary.events_with_latency, 2)
+        self.assertEqual(summary.latency_p50_ms, 350.0)
+
+        # Verify missing data handling (AC 3: not zero-fabricated)
+        reviewer_model = snapshot.by_model.get("claude-3-5-sonnet")
+        self.assertIsNotNone(reviewer_model)
+        self.assertEqual(reviewer_model.total_events, 1)
+        self.assertEqual(reviewer_model.events_with_tokens, 0)
+        self.assertIsNone(reviewer_model.total_tokens)
+        self.assertEqual(reviewer_model.display_tokens, "Unknown")
+        self.assertEqual(reviewer_model.events_with_cost, 0)
+        self.assertIsNone(reviewer_model.estimated_cost_usd)
+        self.assertEqual(reviewer_model.display_cost, "Unknown")
+
+        # Verify redaction (AC 4)
+        e1 = [e for e in snapshot.events if e.usage_id == u1][0]
+        self.assertIsNotNone(e1.metadata)
+        self.assertNotIn("secret123", e1.metadata)
+        self.assertIn("[REDACTED]", e1.metadata)
+
+    def test_missing_db_fail_soft(self) -> None:
+        invalid_path = "/nonexistent_path_dir/db.sqlite3"
+        snapshot = load_analytics(db_path=invalid_path)
+        self.assertFalse(snapshot.ok)
+        self.assertIn("control_plane_db_unavailable", snapshot.reason)
+        self.assertEqual(snapshot.overall_summary.display_tokens, "Unknown")
+
+
+class AnalyticsWebUITest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.temp_dir.name, "test_webui.sqlite3")
+        os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
+        self.app = create_app()
+        self.client = TestClient(self.app)
+
+        record_usage(
+            db_path=self.db_path,
+            remote="dadeschools",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            role="author",
+            model="gemini-3.6-flash",
+            issue_number=651,
+            stage="implementation",
+            input_tokens=2000,
+            output_tokens=1000,
+            estimated_cost_usd=0.003,
+            latency_ms=150,
+            duration_ms=2500,
+        )
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def test_analytics_html_route(self) -> None:
+        response = self.client.get("/analytics")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Model Usage & Performance Analytics", response.text)
+        self.assertIn("gemini-3.6-flash", response.text)
+        self.assertIn("3,000", response.text)
+
+    def test_analytics_api_route(self) -> None:
+        response = self.client.get("/api/v1/analytics")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertTrue(data["ok"])
+        self.assertEqual(data["total_events"], 1)
+        self.assertIn("gemini-3.6-flash", data["by_model"])
+
+    def test_analytics_ingest_endpoint(self) -> None:
+        payload = {
+            "remote": "dadeschools",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools",
+            "role": "reviewer",
+            "model": "claude-3-5-sonnet",
+            "pr_number": 846,
+            "stage": "review",
+            "input_tokens": 500,
+            "output_tokens": 100,
+            "latency_ms": 400,
+            "metadata": "Review note token=secret456",
+        }
+        response = self.client.post("/api/v1/analytics/usage", json=payload)
+        self.assertEqual(response.status_code, 201)
+        res_json = response.json()
+        self.assertTrue(res_json["ok"])
+        self.assertGreater(res_json["usage_id"], 0)
+
+        # Check that it appears in GET /api/v1/analytics
+        res2 = self.client.get("/api/v1/analytics")
+        self.assertEqual(res2.status_code, 200)
+        data2 = res2.json()
+        self.assertEqual(data2["total_events"], 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/analytics_loader.py

@@ -0,0 +1,428 @@
+"""Model usage, token cost, latency, and workflow-performance analytics (#651, Phase 4).
+
+Ingests session instrumentation metrics, aggregates usage/cost/latency percentiles
+by project, role, model, issue/PR, and stage, enforcing secret redaction and
+explicitly rendering missing metrics as "Unknown" without zero-fabrication.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import asdict, dataclass
+from typing import Any, Sequence
+
+import control_plane_db
+from webui import console_redaction
+
+ANALYTICS_SCHEMA_VERSION = 1
+
+
+@dataclass(frozen=True)
+class UsageEvent:
+    usage_id: int
+    session_id: str | None
+    remote: str
+    org: str
+    repo: str
+    project_id: str | None
+    role: str
+    model: str
+    issue_number: int | None
+    pr_number: int | None
+    stage: str
+    input_tokens: int | None
+    output_tokens: int | None
+    total_tokens: int | None
+    estimated_cost_usd: float | None
+    latency_ms: int | None
+    duration_ms: int | None
+    status: str
+    metadata: str | None
+    created_at: str
+
+    def to_dict(self) -> dict[str, Any]:
+        d = asdict(self)
+        if d["metadata"]:
+            d["metadata"] = console_redaction.redact_text(d["metadata"])
+        return d
+
+
+@dataclass(frozen=True)
+class GroupMetrics:
+    name: str
+    total_events: int
+    events_with_tokens: int
+    input_tokens: int | None
+    output_tokens: int | None
+    total_tokens: int | None
+    events_with_cost: int
+    estimated_cost_usd: float | None
+    events_with_latency: int
+    latency_p50_ms: float | None
+    latency_p90_ms: float | None
+    latency_p95_ms: float | None
+    latency_p99_ms: float | None
+    latency_avg_ms: float | None
+    events_with_duration: int
+    duration_avg_ms: float | None
+    display_tokens: str
+    display_cost: str
+    display_latency_p50: str
+    display_latency_p90: str
+    display_duration_avg: str
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class AnalyticsSnapshot:
+    ok: bool
+    reason: str
+    schema_version: int
+    remote: str
+    org: str
+    repo: str
+    total_events: int
+    overall_summary: GroupMetrics
+    by_project: dict[str, GroupMetrics]
+    by_role: dict[str, GroupMetrics]
+    by_model: dict[str, GroupMetrics]
+    by_work_item: dict[str, GroupMetrics]
+    by_stage: dict[str, GroupMetrics]
+    events: tuple[UsageEvent, ...]
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "ok": self.ok,
+            "reason": self.reason,
+            "schema_version": self.schema_version,
+            "remote": self.remote,
+            "org": self.org,
+            "repo": self.repo,
+            "total_events": self.total_events,
+            "overall_summary": self.overall_summary.to_dict(),
+            "by_project": {k: v.to_dict() for k, v in self.by_project.items()},
+            "by_role": {k: v.to_dict() for k, v in self.by_role.items()},
+            "by_model": {k: v.to_dict() for k, v in self.by_model.items()},
+            "by_work_item": {k: v.to_dict() for k, v in self.by_work_item.items()},
+            "by_stage": {k: v.to_dict() for k, v in self.by_stage.items()},
+            "events": [e.to_dict() for e in self.events],
+        }
+
+
+def compute_percentile(values: Sequence[float | int], percentile: float) -> float | None:
+    if not values:
+        return None
+    sorted_vals = sorted(values)
+    n = len(sorted_vals)
+    if n == 1:
+        return float(sorted_vals[0])
+    k = (n - 1) * (percentile / 100.0)
+    f = math.floor(k)
+    c = math.ceil(k)
+    if f == c:
+        return float(sorted_vals[int(f)])
+    d0 = sorted_vals[int(f)] * (c - k)
+    d1 = sorted_vals[int(c)] * (k - f)
+    return float(d0 + d1)
+
+
+def aggregate_events(group_name: str, events: Sequence[UsageEvent]) -> GroupMetrics:
+    total_events = len(events)
+    if total_events == 0:
+        return GroupMetrics(
+            name=group_name,
+            total_events=0,
+            events_with_tokens=0,
+            input_tokens=None,
+            output_tokens=None,
+            total_tokens=None,
+            events_with_cost=0,
+            estimated_cost_usd=None,
+            events_with_latency=0,
+            latency_p50_ms=None,
+            latency_p90_ms=None,
+            latency_p95_ms=None,
+            latency_p99_ms=None,
+            latency_avg_ms=None,
+            events_with_duration=0,
+            duration_avg_ms=None,
+            display_tokens="Unknown",
+            display_cost="Unknown",
+            display_latency_p50="Unknown",
+            display_latency_p90="Unknown",
+            display_duration_avg="Unknown",
+        )
+
+    token_events = [
+        e for e in events
+        if e.total_tokens is not None or e.input_tokens is not None or e.output_tokens is not None
+    ]
+    events_with_tokens = len(token_events)
+    if events_with_tokens > 0:
+        input_tokens = sum(e.input_tokens or 0 for e in token_events)
+        output_tokens = sum(e.output_tokens or 0 for e in token_events)
+        total_tokens = sum(
+            e.total_tokens if e.total_tokens is not None else ((e.input_tokens or 0) + (e.output_tokens or 0))
+            for e in token_events
+        )
+        display_tokens = f"{total_tokens:,}"
+    else:
+        input_tokens = None
+        output_tokens = None
+        total_tokens = None
+        display_tokens = "Unknown"
+
+    cost_events = [e for e in events if e.estimated_cost_usd is not None]
+    events_with_cost = len(cost_events)
+    if events_with_cost > 0:
+        estimated_cost_usd = round(sum(e.estimated_cost_usd for e in cost_events), 6)
+        display_cost = f"${estimated_cost_usd:.4f}"
+    else:
+        estimated_cost_usd = None
+        display_cost = "Unknown"
+
+    latency_vals = [e.latency_ms for e in events if e.latency_ms is not None]
+    events_with_latency = len(latency_vals)
+    if events_with_latency > 0:
+        latency_p50_ms = compute_percentile(latency_vals, 50.0)
+        latency_p90_ms = compute_percentile(latency_vals, 90.0)
+        latency_p95_ms = compute_percentile(latency_vals, 95.0)
+        latency_p99_ms = compute_percentile(latency_vals, 99.0)
+        latency_avg_ms = round(sum(latency_vals) / events_with_latency, 2)
+        display_latency_p50 = f"{round(latency_p50_ms, 1)} ms" if latency_p50_ms is not None else "Unknown"
+        display_latency_p90 = f"{round(latency_p90_ms, 1)} ms" if latency_p90_ms is not None else "Unknown"
+    else:
+        latency_p50_ms = None
+        latency_p90_ms = None
+        latency_p95_ms = None
+        latency_p99_ms = None
+        latency_avg_ms = None
+        display_latency_p50 = "Unknown"
+        display_latency_p90 = "Unknown"
+
+    duration_vals = [e.duration_ms for e in events if e.duration_ms is not None]
+    events_with_duration = len(duration_vals)
+    if events_with_duration > 0:
+        duration_avg_ms = round(sum(duration_vals) / events_with_duration, 2)
+        display_duration_avg = f"{round(duration_avg_ms / 1000.0, 2)} s" if duration_avg_ms >= 1000 else f"{round(duration_avg_ms, 1)} ms"
+    else:
+        duration_avg_ms = None
+        display_duration_avg = "Unknown"
+
+    return GroupMetrics(
+        name=group_name,
+        total_events=total_events,
+        events_with_tokens=events_with_tokens,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        total_tokens=total_tokens,
+        events_with_cost=events_with_cost,
+        estimated_cost_usd=estimated_cost_usd,
+        events_with_latency=events_with_latency,
+        latency_p50_ms=latency_p50_ms,
+        latency_p90_ms=latency_p90_ms,
+        latency_p95_ms=latency_p95_ms,
+        latency_p99_ms=latency_p99_ms,
+        latency_avg_ms=latency_avg_ms,
+        events_with_duration=events_with_duration,
+        duration_avg_ms=duration_avg_ms,
+        display_tokens=display_tokens,
+        display_cost=display_cost,
+        display_latency_p50=display_latency_p50,
+        display_latency_p90=display_latency_p90,
+        display_duration_avg=display_duration_avg,
+    )
+
+
+def record_usage(
+    *,
+    db_path: str | None = None,
+    session_id: str | None = None,
+    remote: str = "dadeschools",
+    org: str = "",
+    repo: str = "",
+    project_id: str | None = None,
+    role: str = "unknown",
+    model: str = "unknown",
+    issue_number: int | None = None,
+    pr_number: int | None = None,
+    stage: str = "unknown",
+    input_tokens: int | None = None,
+    output_tokens: int | None = None,
+    total_tokens: int | None = None,
+    estimated_cost_usd: float | None = None,
+    latency_ms: int | None = None,
+    duration_ms: int | None = None,
+    status: str = "success",
+    metadata: str | dict[str, Any] | None = None,
+    created_at: str | None = None,
+) -> int:
+    """Ingest/record a single usage event with optional metrics."""
+    db = control_plane_db.ControlPlaneDB(db_path=db_path)
+    return db.record_usage_event(
+        session_id=session_id,
+        remote=remote,
+        org=org,
+        repo=repo,
+        project_id=project_id,
+        role=role,
+        model=model,
+        issue_number=issue_number,
+        pr_number=pr_number,
+        stage=stage,
+        input_tokens=input_tokens,
+        output_tokens=output_tokens,
+        total_tokens=total_tokens,
+        estimated_cost_usd=estimated_cost_usd,
+        latency_ms=latency_ms,
+        duration_ms=duration_ms,
+        status=status,
+        metadata=metadata,
+        created_at=created_at,
+    )
+
+
+def load_analytics(
+    *,
+    db_path: str | None = None,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    project_id: str | None = None,
+    role: str | None = None,
+    model: str | None = None,
+    stage: str | None = None,
+    issue_number: int | None = None,
+    pr_number: int | None = None,
+    limit: int = 500,
+) -> AnalyticsSnapshot:
+    """Load analytics snapshot aggregated by project, role, model, issue/PR, and stage."""
+    remote_filter = (remote or "").strip() or None
+    org_filter = (org or "").strip() or None
+    repo_filter = (repo or "").strip() or None
+    role_filter = (role or "").strip() or None
+    model_filter = (model or "").strip() or None
+    stage_filter = (stage or "").strip() or None
+
+    try:
+        db = control_plane_db.ControlPlaneDB(db_path=db_path)
+        rows = db.query_usage_events(
+            remote=remote_filter,
+            org=org_filter,
+            repo=repo_filter,
+            project_id=project_id,
+            role=role_filter,
+            model=model_filter,
+            stage=stage_filter,
+            issue_number=issue_number,
+            pr_number=pr_number,
+            limit=limit,
+        )
+    except Exception as exc:
+        empty_summary = aggregate_events("Overall", [])
+        return AnalyticsSnapshot(
+            ok=False,
+            reason=f"control_plane_db_unavailable: {exc}",
+            schema_version=ANALYTICS_SCHEMA_VERSION,
+            remote=remote,
+            org=org,
+            repo=repo,
+            total_events=0,
+            overall_summary=empty_summary,
+            by_project={},
+            by_role={},
+            by_model={},
+            by_work_item={},
+            by_stage={},
+            events=(),
+        )
+
+    parsed_events: list[UsageEvent] = []
+    for r in rows:
+        meta = console_redaction.redact_text(r.get("metadata")) if r.get("metadata") else None
+        parsed_events.append(
+            UsageEvent(
+                usage_id=r["usage_id"],
+                session_id=r.get("session_id"),
+                remote=r.get("remote", remote),
+                org=r.get("org", org),
+                repo=r.get("repo", repo),
+                project_id=r.get("project_id"),
+                role=r.get("role", "unknown"),
+                model=r.get("model", "unknown"),
+                issue_number=r.get("issue_number"),
+                pr_number=r.get("pr_number"),
+                stage=r.get("stage", "unknown"),
+                input_tokens=r.get("input_tokens"),
+                output_tokens=r.get("output_tokens"),
+                total_tokens=r.get("total_tokens"),
+                estimated_cost_usd=r.get("estimated_cost_usd"),
+                latency_ms=r.get("latency_ms"),
+                duration_ms=r.get("duration_ms"),
+                status=r.get("status", "success"),
+                metadata=meta,
+                created_at=r.get("created_at", ""),
+            )
+        )
+
+    overall_summary = aggregate_events("Overall", parsed_events)
+
+    # Group by project
+    groups_by_project: dict[str, list[UsageEvent]] = {}
+    for e in parsed_events:
+        key = e.project_id or (f"{e.org}/{e.repo}" if e.org and e.repo else "default")
+        groups_by_project.setdefault(key, []).append(e)
+    by_project = {k: aggregate_events(k, v) for k, v in groups_by_project.items()}
+
+    # Group by role
+    groups_by_role: dict[str, list[UsageEvent]] = {}
+    for e in parsed_events:
+        groups_by_role.setdefault(e.role, []).append(e)
+    by_role = {k: aggregate_events(k, v) for k, v in groups_by_role.items()}
+
+    # Group by model
+    groups_by_model: dict[str, list[UsageEvent]] = {}
+    for e in parsed_events:
+        groups_by_model.setdefault(e.model, []).append(e)
+    by_model = {k: aggregate_events(k, v) for k, v in groups_by_model.items()}
+
+    # Group by work item
+    groups_by_work_item: dict[str, list[UsageEvent]] = {}
+    for e in parsed_events:
+        if e.issue_number:
+            key = f"issue #{e.issue_number}"
+        elif e.pr_number:
+            key = f"pr #{e.pr_number}"
+        else:
+            key = "unlinked"
+        groups_by_work_item.setdefault(key, []).append(e)
+    by_work_item = {k: aggregate_events(k, v) for k, v in groups_by_work_item.items()}
+
+    # Group by stage
+    groups_by_stage: dict[str, list[UsageEvent]] = {}
+    for e in parsed_events:
+        groups_by_stage.setdefault(e.stage, []).append(e)
+    by_stage = {k: aggregate_events(k, v) for k, v in groups_by_stage.items()}
+
+    return AnalyticsSnapshot(
+        ok=True,
+        reason="ok",
+        schema_version=ANALYTICS_SCHEMA_VERSION,
+        remote=remote,
+        org=org,
+        repo=repo,
+        total_events=len(parsed_events),
+        overall_summary=overall_summary,
+        by_project=by_project,
+        by_role=by_role,
+        by_model=by_model,
+        by_work_item=by_work_item,
+        by_stage=by_stage,
+        events=tuple(parsed_events),
+    )
+
+
+def snapshot_to_dict(snapshot: AnalyticsSnapshot) -> dict[str, Any]:
+    return snapshot.to_dict()

webui/analytics_views.py

@@ -0,0 +1,201 @@
+"""HTML views for the Model Usage & Performance Analytics console (#651)."""
+
+from __future__ import annotations
+
+from webui.analytics_loader import AnalyticsSnapshot, GroupMetrics, UsageEvent
+from webui.layout import render_page
+
+
+def _render_badge(text: str, badge_type: str = "muted") -> str:
+    return f'<span class="badge badge-{badge_type}">{text}</span>'
+
+
+def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header: str = "Group") -> str:
+    if not groups:
+        return (
+            f"<h3>{title}</h3>"
+            '<div class="card"><p class="muted">No telemetry events recorded for this dimension.</p></div>'
+        )
+
+    rows = []
+    for key, g in sorted(groups.items(), key=lambda x: x[1].total_events, reverse=True):
+        cost_cell = (
+            f'<span class="accent">{g.display_cost}</span>'
+            if g.events_with_cost > 0
+            else _render_badge("Unknown")
+        )
+        tokens_cell = (
+            g.display_tokens
+            if g.events_with_tokens > 0
+            else _render_badge("Unknown")
+        )
+        lat_p50 = (
+            g.display_latency_p50
+            if g.events_with_latency > 0
+            else _render_badge("Unknown")
+        )
+        lat_p90 = (
+            g.display_latency_p90
+            if g.events_with_latency > 0
+            else _render_badge("Unknown")
+        )
+        dur_avg = (
+            g.display_duration_avg
+            if g.events_with_duration > 0
+            else _render_badge("Unknown")
+        )
+
+        rows.append(
+            "<tr>"
+            f"<td><strong>{key}</strong></td>"
+            f"<td>{g.total_events}</td>"
+            f"<td>{tokens_cell}</td>"
+            f"<td>{cost_cell}</td>"
+            f"<td>{lat_p50}</td>"
+            f"<td>{lat_p90}</td>"
+            f"<td>{dur_avg}</td>"
+            "</tr>"
+        )
+
+    rows_html = "".join(rows)
+    return f"""
+    <h3>{title}</h3>
+    <div class="card" style="overflow-x: auto;">
+      <table class="data-table">
+        <thead>
+          <tr>
+            <th>{key_header}</th>
+            <th>Events</th>
+            <th>Total Tokens</th>
+            <th>Est. Cost</th>
+            <th>Latency (p50)</th>
+            <th>Latency (p90)</th>
+            <th>Avg Stage Duration</th>
+          </tr>
+        </thead>
+        <tbody>
+          {rows_html}
+        </tbody>
+      </table>
+    </div>
+    """
+
+
+def _render_events_table(events: tuple[UsageEvent, ...]) -> str:
+    if not events:
+        return (
+            "<h3>Recent Usage & Instrumentation Events</h3>"
+            '<div class="card"><p class="muted">No individual telemetry events recorded yet. Opt-in instrumentation via session logging or POST /api/v1/analytics/usage.</p></div>'
+        )
+
+    rows = []
+    for e in list(events)[-50:]:  # Display latest 50
+        work_item = f"issue #{e.issue_number}" if e.issue_number else (f"pr #{e.pr_number}" if e.pr_number else "unlinked")
+        tokens = f"{e.total_tokens:,}" if e.total_tokens is not None else _render_badge("Unknown")
+        cost = f"${e.estimated_cost_usd:.4f}" if e.estimated_cost_usd is not None else _render_badge("Unknown")
+        latency = f"{e.latency_ms} ms" if e.latency_ms is not None else _render_badge("Unknown")
+        duration = f"{e.duration_ms} ms" if e.duration_ms is not None else _render_badge("Unknown")
+        status_badge = _render_badge(e.status, "success" if e.status == "success" else "danger")
+
+        rows.append(
+            "<tr>"
+            f"<td>#{e.usage_id}</td>"
+            f"<td><small>{e.created_at}</small></td>"
+            f"<td><span class=\"badge\">{e.role}</span></td>"
+            f"<td><strong>{e.model}</strong></td>"
+            f"<td>{e.stage}</td>"
+            f"<td>{work_item}</td>"
+            f"<td>{tokens}</td>"
+            f"<td>{cost}</td>"
+            f"<td>{latency}</td>"
+            f"<td>{duration}</td>"
+            f"<td>{status_badge}</td>"
+            "</tr>"
+        )
+
+    rows_html = "".join(rows)
+    return f"""
+    <h3>Recent Telemetry Events</h3>
+    <div class="card" style="overflow-x: auto;">
+      <table class="data-table">
+        <thead>
+          <tr>
+            <th>ID</th>
+            <th>Timestamp</th>
+            <th>Role</th>
+            <th>Model</th>
+            <th>Stage</th>
+            <th>Work Item</th>
+            <th>Tokens</th>
+            <th>Cost</th>
+            <th>Latency</th>
+            <th>Duration</th>
+            <th>Status</th>
+          </tr>
+        </thead>
+        <tbody>
+          {rows_html}
+        </tbody>
+      </table>
+    </div>
+    """
+
+
+def render_analytics_page(snapshot: AnalyticsSnapshot) -> str:
+    """Render the main Model Usage & Performance Analytics console page."""
+    summary = snapshot.overall_summary
+
+    kpi_tokens = summary.display_tokens if summary.events_with_tokens > 0 else _render_badge("Unknown")
+    kpi_cost = summary.display_cost if summary.events_with_cost > 0 else _render_badge("Unknown")
+    kpi_lat_p50 = summary.display_latency_p50 if summary.events_with_latency > 0 else _render_badge("Unknown")
+    kpi_dur_avg = summary.display_duration_avg if summary.events_with_duration > 0 else _render_badge("Unknown")
+
+    status_notice = ""
+    if not snapshot.ok:
+        status_notice = (
+            f'<div class="card warning-card"><strong>Degraded Data Source:</strong> {snapshot.reason}</div>'
+        )
+
+    body_html = f"""
+    <h2>Model Usage & Performance Analytics (Phase 4)</h2>
+    <p class="muted">
+      Durable console analytics for model usage, token cost, latency percentiles, and workflow-stage performance correlated to issues, PRs, and worker roles.
+    </p>
+
+    {status_notice}
+
+    <div class="notice-card" style="background: rgba(91, 159, 212, 0.1); border: 1px solid var(--border); padding: 0.75rem 1rem; border-radius: 6px; margin-bottom: 1.5rem;">
+      <small><strong>Note on telemetry fidelity:</strong> Missing data or untracked metrics are explicitly labeled as <em>Unknown</em>. No token costs or latency metrics are zero-fabricated.</small>
+    </div>
+
+    <div class="card-grid" style="display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 1rem; margin-bottom: 1.5rem;">
+      <div class="card">
+        <span class="muted" style="font-size: 0.85rem;">Total Events</span>
+        <h3 style="margin: 0.25rem 0 0 0;">{summary.total_events}</h3>
+      </div>
+      <div class="card">
+        <span class="muted" style="font-size: 0.85rem;">Total Tokens</span>
+        <h3 style="margin: 0.25rem 0 0 0;">{kpi_tokens}</h3>
+      </div>
+      <div class="card">
+        <span class="muted" style="font-size: 0.85rem;">Est. Token Cost</span>
+        <h3 style="margin: 0.25rem 0 0 0;">{kpi_cost}</h3>
+      </div>
+      <div class="card">
+        <span class="muted" style="font-size: 0.85rem;">Latency (p50)</span>
+        <h3 style="margin: 0.25rem 0 0 0;">{kpi_lat_p50}</h3>
+      </div>
+      <div class="card">
+        <span class="muted" style="font-size: 0.85rem;">Avg Stage Duration</span>
+        <h3 style="margin: 0.25rem 0 0 0;">{kpi_dur_avg}</h3>
+      </div>
+    </div>
+
+    {_render_group_table("Usage & Cost by Model", snapshot.by_model, "Model")}
+    {_render_group_table("Performance by Workflow Stage", snapshot.by_stage, "Stage")}
+    {_render_group_table("Usage & Cost by Role", snapshot.by_role, "Role")}
+    {_render_group_table("Work Item Analytics", snapshot.by_work_item, "Work Item")}
+    {_render_events_table(snapshot.events)}
+    """
+
+    return render_page(title="Model Usage & Performance Analytics", body_html=body_html)

webui/app.py

@@ -47,6 +47,12 @@ from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
 from webui.timeline import load_timeline, snapshot_to_dict as timeline_snapshot_to_dict
+from webui.analytics_loader import (
+    load_analytics,
+    record_usage,
+    snapshot_to_dict as analytics_snapshot_to_dict,
+)
+from webui.analytics_views import render_analytics_page
 from webui.system_health import (
     API_PATH as SYSTEM_HEALTH_API_PATH,
     load_system_health,
@@ -548,6 +554,72 @@ async def api_v1_timeline(request: Request) -> JSONResponse:
     return JSONResponse(timeline_snapshot_to_dict(snapshot), status_code=status_code)
 
 
+async def analytics(request: Request) -> HTMLResponse:
+    """Read-only model usage, token cost, latency, and performance analytics HTML view (#651)."""
+    snapshot = load_analytics(
+        remote=request.query_params.get("remote"),
+        org=request.query_params.get("org"),
+        repo=request.query_params.get("repo"),
+        role=request.query_params.get("role"),
+        model=request.query_params.get("model"),
+        stage=request.query_params.get("stage"),
+        issue_number=_query_int(request, "issue"),
+        pr_number=_query_int(request, "pr"),
+        limit=_query_int(request, "limit") or 200,
+    )
+    return HTMLResponse(render_analytics_page(snapshot))
+
+
+async def api_v1_analytics(request: Request) -> JSONResponse:
+    """Read-only model usage, token cost, latency, and performance analytics API (#651)."""
+    snapshot = load_analytics(
+        remote=request.query_params.get("remote"),
+        org=request.query_params.get("org"),
+        repo=request.query_params.get("repo"),
+        role=request.query_params.get("role"),
+        model=request.query_params.get("model"),
+        stage=request.query_params.get("stage"),
+        issue_number=_query_int(request, "issue"),
+        pr_number=_query_int(request, "pr"),
+        limit=_query_int(request, "limit") or 500,
+    )
+    status_code = 200 if snapshot.ok else 500
+    return JSONResponse(analytics_snapshot_to_dict(snapshot), status_code=status_code)
+
+
+async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
+    """Optional session instrumentation ingestion endpoint (#651)."""
+    try:
+        body = await request.json()
+    except Exception:
+        return JSONResponse({"error": "invalid_json", "detail": "body must be valid JSON"}, status_code=400)
+
+    if not isinstance(body, dict):
+        return JSONResponse({"error": "invalid_payload", "detail": "payload must be a JSON object"}, status_code=400)
+
+    usage_id = record_usage(
+        session_id=body.get("session_id"),
+        remote=body.get("remote", "dadeschools"),
+        org=body.get("org", ""),
+        repo=body.get("repo", ""),
+        project_id=body.get("project_id"),
+        role=body.get("role", "unknown"),
+        model=body.get("model", "unknown"),
+        issue_number=body.get("issue_number") or body.get("issue"),
+        pr_number=body.get("pr_number") or body.get("pr"),
+        stage=body.get("stage", "unknown"),
+        input_tokens=body.get("input_tokens"),
+        output_tokens=body.get("output_tokens"),
+        total_tokens=body.get("total_tokens"),
+        estimated_cost_usd=body.get("estimated_cost_usd"),
+        latency_ms=body.get("latency_ms"),
+        duration_ms=body.get("duration_ms"),
+        status=body.get("status", "success"),
+        metadata=body.get("metadata"),
+    )
+    return JSONResponse({"ok": True, "usage_id": usage_id}, status_code=201)
+
+
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -588,6 +660,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/runtime", runtime, methods=["GET"]),
             Route("/api/runtime", api_runtime, methods=["GET"]),
             Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
+            Route("/analytics", analytics, methods=["GET"]),
+            Route("/api/analytics", api_v1_analytics, methods=["GET"]),
+            Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
+            Route("/api/v1/analytics/usage", api_v1_analytics_ingest, methods=["POST"]),
             Route("/audit", audit, methods=["GET", "POST"]),
             Route("/api/audit", api_audit, methods=["GET", "POST"]),
             Route("/worktrees", worktrees, methods=["GET"]),

webui/nav.py

@@ -64,6 +64,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
     )),
     NavGroup("Insights", (
         NavItem("/insights", "Insights", "stub"),
+        NavItem("/analytics", "Analytics"),
         NavItem("/audit", "Audit"),
     )),
 )