Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 af70a27b01 merge(master): resolve conflicts for PR #906 providers/insights
Merge prgs/master into feat/issue-650-providers-insights. Conflicts in
webui/app.py, webui/nav.py, and docs/webui-local-dev.md kept both:
- #650 providers/insights routes and docs
- #645 gitea linkage routes and docs

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-25 17:53:32 -04:00
sysadmin 9c69bfcd80 Merge pull request 'feat(webui): Gitea issue and PR linkage console (Closes #645)' (#904) from feat/issue-645-linkage-console into master 2026-07-25 16:32:21 -05:00
sysadminandClaude Opus 4.8 983e8ac2c7 feat(webui): AI-provider connections and evidence-backed insights (Closes #650)
Add Phase 4 advisory surfaces for declared AI-provider connection status
(no secrets, no live probe claims) and operational insights derived only
from durable traffic, health, provider, and analytics evidence.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-25 16:53:36 -04:00
6 changed files with 1412 additions and 14 deletions
+54 -1
View File
@@ -85,7 +85,10 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
| `/timeline` | Phase 1 shell stub — workflow event timeline |
| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
| `/insights` | Phase 1 shell stub — operational insights placeholder |
| `/providers` | AI-provider connection status (#650) — declared registry only, no secrets |
| `/api/v1/providers` | JSON provider connections; `502` when the registry cannot be loaded |
| `/insights` | Evidence-backed operational insights (#650) — advisory only |
| `/api/v1/insights` | JSON insights export with evidence refs and source availability |
Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with
`read-only-mvp`, except `/audit` and `/api/audit` which accept POST for
@@ -329,6 +332,56 @@ Honesty rules specific to this view:
The write-time redactor is a narrow denylist and is not relied on. The field
itself is kept — it is the `#630` evidence naming which daemon was killed.
## AI providers and operational insights (#650)
`/providers` and `/insights` are the Phase 4 **advisory** surfaces for AI-provider
connections and evidence-backed operational findings. They never expose API keys,
never mutate Gitea, and never authorize review, merge, or close.
### Provider connections (`/providers`)
Status is taken from the **worker registry** declaration (`webui/data/workers.registry.json`
or `WEBUI_WORKER_REGISTRY`):
| Field | Meaning |
|-------|---------|
| `connection_status` | `declared_available` or `declared_unavailable` from the registry `available` flag |
| `models` | Declared model list only (not a live vendor enumeration) |
| `worker_count` / `enabled_worker_count` | How many worker instances name this provider |
| `secrets_exposed` | Always `false` — credentials are never loaded |
Live executable health is **not** probed here (that belongs to the provider adapter
framework). The page states this probe limit explicitly so a green badge is not
misread as a process heartbeat.
`GET /api/v1/providers` returns the same model (`schema_version: 1`). It answers
`502` when the registry cannot be loaded so consumers cannot treat a fail-closed
payload as “no providers configured”.
### Operational insights (`/insights`)
Insights are pure functions over durable console evidence:
| Kind | Evidence source |
|------|-----------------|
| `blocked_queue_pressure` | Traffic control blocked bucket (issue/PR numbers + reasons) |
| `controller_attention` | Traffic control `needs_controller` items |
| `stale_runtime_risk` | System-health stale_runtime / mutation_safe |
| `provider_without_workers` | Declared-available providers with zero workers |
| `analytics_failure_pressure` | Analytics events with failure status (when loaded) |
Rules:
* Every insight carries at least one evidence ref (`kind` + `ref` + `detail`).
Evidence-less insights are refused, not emitted.
* `advisory_only` is always true; `claims_action_completed` is always false.
* Missing sources appear under `sources_unavailable` — never as a silent empty
“all clear”.
* Titles and details pass through console redaction before display.
`GET /api/v1/insights` exports the same model. The HTML page always renders
interpretation limits so operators know these cards do not override workflow gates.
## Gitea issue/PR linkage (#645)
`/gitea` is the Phase 3 read-only linkage console: which PR carries which issue,
+404
View File
@@ -0,0 +1,404 @@
"""Tests for AI-provider connections and evidence-backed insights (#650).
Covers acceptance criteria:
1. Provider connection status is redacted and accurate (declared registry only).
2. At least three insight types with evidence citations.
3. Insights never claim actions completed without proof.
4. Evidence requirement is enforced (no evidence-less insights).
5. Interpretation limits appear in docs-facing payloads.
"""
from __future__ import annotations
import json
import sys
import unittest
from dataclasses import dataclass
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from tests.webui_testclient import TestClient
from webui.app import create_app
from webui.insights_loader import (
CONFIDENCE_HIGH,
CONNECTION_DECLARED_AVAILABLE,
CONNECTION_DECLARED_UNAVAILABLE,
INSIGHT_BLOCKED_QUEUE,
INSIGHT_CONTROLLER_ATTENTION,
INSIGHT_PROVIDER_WITHOUT_WORKERS,
INSIGHT_STALE_RUNTIME,
ProviderConnection,
build_provider_connection,
generate_insights,
insight_blocked_queue,
insight_controller_attention,
insight_providers_without_workers,
insight_stale_runtime,
load_insights_snapshot,
load_provider_snapshot,
snapshot_insights_to_dict,
snapshot_providers_to_dict,
)
from webui.insights_views import render_insights_page, render_providers_page
from webui.nav import STUB_PAGES, nav_hrefs
from webui.worker_registry import ProviderRecord, WorkerRecord, ScheduleSpec, SchedulerSpec
def _provider(
provider_id: str = "claude",
*,
available: bool = True,
models: tuple[str, ...] = ("claude-opus-4-8",),
notes: str = "",
) -> ProviderRecord:
return ProviderRecord(
id=provider_id,
display_name=provider_id.title(),
vendor="TestVendor",
executable=provider_id,
available=available,
models=models,
notes=notes,
)
def _worker(
worker_id: str = "claude-author",
*,
provider: str = "claude",
enabled: bool = True,
) -> WorkerRecord:
return WorkerRecord(
id=worker_id,
display_name=worker_id,
provider=provider,
model="m1",
project="gitea-tools",
role="author",
namespace="gitea-author",
profile="prgs-author",
workflow="skills/llm-project-workflow/workflows/work-issue.md",
schedule=ScheduleSpec(kind="manual", seconds=None, expression=None),
timeout_seconds=3600,
enabled=enabled,
scheduler=SchedulerSpec(kind="manual", label=None),
notes="",
)
@dataclass(frozen=True)
class _TrafficItem:
kind: str
number: int
title: str = ""
traffic_state: str = "blocked"
expected_role: str = "author"
safe_for_roles: tuple[str, ...] = ()
badges: tuple[str, ...] = ()
block_reason: str | None = "dependency"
@dataclass(frozen=True)
class _Traffic:
blocked: tuple = ()
needs_controller: tuple = ()
inventory_complete: bool = True
fetch_error: str | None = None
@dataclass(frozen=True)
class _Stale:
daemon_head: str | None
checkout_head: str | None
remote_head: str | None
stale: bool
determinable: bool
mutation_safe: bool
@dataclass(frozen=True)
class _Health:
stale_runtime: _Stale | None
class TestProviderConnections(unittest.TestCase):
def test_available_provider_status(self):
conn = build_provider_connection(_provider(available=True), (_worker(),))
self.assertEqual(conn.connection_status, CONNECTION_DECLARED_AVAILABLE)
self.assertTrue(conn.available_declared)
self.assertEqual(conn.worker_count, 1)
self.assertEqual(conn.enabled_worker_count, 1)
self.assertFalse(conn.to_dict()["secrets_exposed"])
def test_unavailable_provider_status(self):
conn = build_provider_connection(_provider(available=False), ())
self.assertEqual(conn.connection_status, CONNECTION_DECLARED_UNAVAILABLE)
self.assertEqual(conn.worker_count, 0)
def test_notes_are_redacted(self):
conn = build_provider_connection(
_provider(notes="token=ghp_thisisnotarealsecretvalue0001"),
(),
)
self.assertNotIn("ghp_thisisnotarealsecretvalue0001", conn.notes)
self.assertNotIn(
"ghp_thisisnotarealsecretvalue0001",
json.dumps(conn.to_dict()),
)
def test_load_provider_snapshot_from_injected_registry(self):
from webui.worker_registry import WorkerRegistry
from pathlib import Path
registry = WorkerRegistry(
version=1,
revision=3,
updated_at="2026-07-25T00:00:00Z",
providers=(_provider("claude"), _provider("grok", available=False)),
workers=(_worker(),),
source_path=Path("/tmp/workers.registry.json"),
)
snapshot = load_provider_snapshot(registry=registry)
self.assertTrue(snapshot.ok)
self.assertEqual(snapshot.registry_revision, 3)
ids = {p.provider_id for p in snapshot.providers}
self.assertEqual(ids, {"claude", "grok"})
def test_registry_failure_is_fail_closed(self):
def _boom():
raise RuntimeError("disk gone")
snapshot = load_provider_snapshot(registry_loader=_boom)
self.assertFalse(snapshot.ok)
self.assertIn("unavailable", snapshot.fetch_error or "")
self.assertEqual(snapshot.providers, ())
class TestInsightGenerators(unittest.TestCase):
def test_blocked_queue_requires_evidence(self):
traffic = _Traffic(
blocked=(
_TrafficItem(kind="issue", number=647, block_reason="depends #646"),
_TrafficItem(kind="pr", number=902, block_reason="conflict"),
)
)
insight = insight_blocked_queue(traffic)
self.assertIsNotNone(insight)
self.assertEqual(insight.kind, INSIGHT_BLOCKED_QUEUE)
self.assertGreaterEqual(len(insight.evidence), 2)
self.assertTrue(insight.advisory_only)
self.assertFalse(insight.claims_action_completed)
refs = {e.ref for e in insight.evidence}
self.assertIn("#647", refs)
self.assertIn("#902", refs)
def test_empty_blocked_queue_yields_no_insight(self):
self.assertIsNone(insight_blocked_queue(_Traffic()))
def test_controller_attention_insight(self):
traffic = _Traffic(
needs_controller=(_TrafficItem(kind="issue", number=100, traffic_state="needs_controller"),)
)
insight = insight_controller_attention(traffic)
self.assertEqual(insight.kind, INSIGHT_CONTROLLER_ATTENTION)
self.assertEqual(insight.evidence[0].ref, "#100")
def test_stale_runtime_insight(self):
health = _Health(
stale_runtime=_Stale(
daemon_head="aaa",
checkout_head="bbb",
remote_head="ccc",
stale=True,
determinable=True,
mutation_safe=False,
)
)
insight = insight_stale_runtime(health)
self.assertEqual(insight.kind, INSIGHT_STALE_RUNTIME)
self.assertIn("stale", insight.evidence[0].detail)
self.assertFalse(insight.claims_action_completed)
def test_mutation_safe_runtime_yields_no_insight(self):
health = _Health(
stale_runtime=_Stale(
daemon_head="aaa",
checkout_head="aaa",
remote_head="aaa",
stale=False,
determinable=True,
mutation_safe=True,
)
)
self.assertIsNone(insight_stale_runtime(health))
def test_provider_without_workers(self):
providers = (
build_provider_connection(_provider("claude"), (_worker(),)),
build_provider_connection(_provider("grok"), ()),
)
insight = insight_providers_without_workers(providers)
self.assertEqual(insight.kind, INSIGHT_PROVIDER_WITHOUT_WORKERS)
self.assertEqual(insight.evidence[0].ref, "grok")
def test_generate_insights_composes_three_kinds(self):
from webui.worker_registry import WorkerRegistry
from pathlib import Path
registry = WorkerRegistry(
version=1,
revision=1,
updated_at="2026-07-25T00:00:00Z",
providers=(_provider("lonely"),),
workers=(),
source_path=Path("/tmp/w.json"),
)
provider_snapshot = load_provider_snapshot(registry=registry)
traffic = _Traffic(
blocked=(_TrafficItem(kind="issue", number=1),),
needs_controller=(_TrafficItem(kind="issue", number=2),),
)
health = _Health(
stale_runtime=_Stale("a", "b", "c", True, True, False)
)
insights, used, unavailable = generate_insights(
traffic=traffic,
health=health,
provider_snapshot=provider_snapshot,
analytics=None,
)
kinds = {i.kind for i in insights}
self.assertIn(INSIGHT_BLOCKED_QUEUE, kinds)
self.assertIn(INSIGHT_CONTROLLER_ATTENTION, kinds)
self.assertIn(INSIGHT_STALE_RUNTIME, kinds)
self.assertIn(INSIGHT_PROVIDER_WITHOUT_WORKERS, kinds)
self.assertGreaterEqual(len(kinds), 3)
self.assertIn("traffic", used)
self.assertIn("system_health", used)
self.assertIn("providers", used)
self.assertTrue(any(u["source"] == "analytics" for u in unavailable))
for insight in insights:
self.assertTrue(insight.advisory_only)
self.assertFalse(insight.claims_action_completed)
self.assertGreaterEqual(len(insight.evidence), 1)
def test_missing_source_is_reported_not_as_healthy_empty(self):
insights, used, unavailable = generate_insights(
traffic=None,
health=None,
provider_snapshot=None,
analytics=None,
)
self.assertEqual(insights, ())
self.assertEqual(used, ())
self.assertEqual(len(unavailable), 4)
class TestViewsAndRoutes(unittest.TestCase):
def test_providers_page_renders_connections(self):
from webui.worker_registry import WorkerRegistry
from pathlib import Path
registry = WorkerRegistry(
version=1,
revision=1,
updated_at="2026-07-25T00:00:00Z",
providers=(_provider("claude"),),
workers=(_worker(),),
source_path=Path("/tmp/w.json"),
)
snapshot = load_provider_snapshot(registry=registry)
html = render_providers_page(snapshot)
self.assertIn("AI provider connections", html)
self.assertIn("claude", html)
self.assertIn("declared_available", html)
self.assertIn("Interpretation limits", html)
self.assertNotIn("api_key", html.lower())
def test_failed_provider_snapshot_renders_no_table(self):
snapshot = load_provider_snapshot(registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("x")))
html = render_providers_page(snapshot)
self.assertIn("unavailable", html.lower())
self.assertNotIn("<tbody><tr><td><code>", html)
def test_insights_page_lists_evidence(self):
traffic = _Traffic(blocked=(_TrafficItem(kind="issue", number=42),))
snapshot = load_insights_snapshot(
traffic=traffic,
health=_Health(None),
provider_snapshot=load_provider_snapshot(
registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("skip"))
),
analytics=None,
load_live=False,
)
html = render_insights_page(snapshot)
self.assertIn("#42", html)
self.assertIn("advisory only", html.lower())
self.assertIn("Evidence", html)
def test_nav_exposes_live_insights_and_providers(self):
self.assertIn("/insights", nav_hrefs())
self.assertIn("/providers", nav_hrefs())
self.assertNotIn("/insights", STUB_PAGES)
def test_routes_are_read_only_and_export_json(self):
client = TestClient(create_app())
# Use live registry from package data — should be ok.
with mock.patch(
"webui.app.load_provider_snapshot",
return_value=load_provider_snapshot(
registry=__import__(
"webui.worker_registry", fromlist=["load_registry"]
).load_registry()
),
):
response = client.get("/providers")
self.assertEqual(response.status_code, 200)
self.assertIn("provider", response.text.lower())
api = client.get("/api/v1/providers")
self.assertEqual(api.status_code, 200)
payload = api.json()
self.assertTrue(payload["ok"])
self.assertIn("interpretation_limits", payload)
self.assertTrue(all(not p.get("secrets_exposed") for p in payload["providers"]))
with mock.patch(
"webui.app.load_insights_snapshot",
return_value=load_insights_snapshot(
traffic=_Traffic(blocked=(_TrafficItem(kind="issue", number=7),)),
health=_Health(None),
provider_snapshot=load_provider_snapshot(
registry_loader=lambda: (_ for _ in ()).throw(RuntimeError("x"))
),
analytics=None,
load_live=False,
),
):
page = client.get("/insights")
self.assertEqual(page.status_code, 200)
self.assertIn("#7", page.text)
api = client.get("/api/v1/insights")
self.assertEqual(api.status_code, 200)
body = api.json()
self.assertTrue(body["ok"])
self.assertTrue(all(i["advisory_only"] for i in body["insights"]))
self.assertTrue(all(not i["claims_action_completed"] for i in body["insights"]))
self.assertTrue(all(i["evidence"] for i in body["insights"]))
for path in ("/providers", "/api/v1/providers", "/insights", "/api/v1/insights"):
with self.subTest(path=path):
self.assertEqual(client.post(path).status_code, 405)
def test_home_nav_links_providers_and_insights(self):
home = TestClient(create_app()).get("/").text
self.assertIn('href="/providers"', home)
self.assertIn('href="/insights"', home)
if __name__ == "__main__":
unittest.main()
+39 -2
View File
@@ -53,6 +53,13 @@ from webui.session_loader import (
snapshot_to_dict as session_view_snapshot_to_dict,
)
from webui.session_views import render_sessions_page
from webui.insights_loader import (
load_insights_snapshot,
load_provider_snapshot,
snapshot_insights_to_dict,
snapshot_providers_to_dict,
)
from webui.insights_views import render_insights_page, render_providers_page
from webui.linkage_loader import (
load_linkage_snapshot,
snapshot_to_dict as linkage_snapshot_to_dict,
@@ -347,6 +354,34 @@ async def api_sessions(_request: Request) -> JSONResponse:
return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
async def providers(_request: Request) -> HTMLResponse:
"""AI-provider connection status (#650) — declared registry only, no secrets."""
return HTMLResponse(render_providers_page(load_provider_snapshot()))
async def api_v1_providers(_request: Request) -> JSONResponse:
"""JSON export of declared AI-provider connections (#650)."""
snapshot = load_provider_snapshot()
return JSONResponse(
snapshot_providers_to_dict(snapshot),
status_code=200 if snapshot.ok else 502,
)
async def insights(_request: Request) -> HTMLResponse:
"""Evidence-backed operational insights (#650) — advisory only."""
return HTMLResponse(render_insights_page(load_insights_snapshot()))
async def api_v1_insights(_request: Request) -> JSONResponse:
"""JSON export of evidence-backed insights (#650)."""
snapshot = load_insights_snapshot()
return JSONResponse(
snapshot_insights_to_dict(snapshot),
status_code=200 if snapshot.ok else 502,
)
def _linkage_snapshot(request: Request):
"""Load one linkage snapshot from the request's scope and focus parameters."""
return load_linkage_snapshot(
@@ -380,8 +415,6 @@ async def api_v1_gitea_linkage(request: Request) -> JSONResponse:
linkage_snapshot_to_dict(snapshot),
status_code=200 if snapshot.ok else 502,
)
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
if request.method == "GET":
return "", None
@@ -825,6 +858,10 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/sessions", api_sessions, methods=["GET"]),
Route("/api/v1/sessions", api_sessions, methods=["GET"]),
Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
Route("/providers", providers, methods=["GET"]),
Route("/api/v1/providers", api_v1_providers, methods=["GET"]),
Route("/insights", insights, methods=["GET"]),
Route("/api/v1/insights", api_v1_insights, methods=["GET"]),
Route("/gitea", gitea_linkage, methods=["GET"]),
Route("/api/v1/gitea/linkage", api_v1_gitea_linkage, methods=["GET"]),
Route("/analytics", analytics, methods=["GET"]),
+713
View File
@@ -0,0 +1,713 @@
"""AI-provider connections and evidence-backed operational insights (#650, Phase 4).
Operators need two related, **advisory** surfaces:
1. **Provider connection status** — which AI runtimes are *declared* in the
worker registry (#798), without ever exposing API keys or inventing a live
probe that this process cannot perform.
2. **Evidence-backed insights** — short cards derived only from durable
console evidence (traffic, system health, analytics, the same registry).
Every insight carries explicit evidence refs (issue/PR/provider/event ids).
Insights never claim that a workflow action completed without proof, and
they never mutate anything.
Design rules matching the rest of the console:
- **Read-only.** No endpoint registered here mutates Gitea, the control plane,
or the registry.
- **Advisory only.** Insights carry ``advisory_only=True`` and never emit an
"action completed" claim. The allocator, review, and merge paths remain the
only authorities for work selection and terminal state.
- **Qualified absence.** When a source could not run, the insight list says so
rather than inventing an empty-and-healthy fleet or zero blocked items.
- **Redaction.** Free-text titles, reasons, and notes pass through
``webui.console_redaction`` before they leave this module.
- **No secrets.** Provider records are taken from the credential-free worker
registry. Keys never appear in this surface.
Non-goals (from the issue): free-form chatbot that overrides gates, secret
provider keys in the UI, auto-merge or auto-close from insights.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any, Callable, Sequence
from webui import console_redaction
from webui.worker_registry import (
ProviderRecord,
WorkerRegistry,
WorkerRecord,
load_registry as load_worker_registry,
workers_for_provider,
)
INSIGHTS_SCHEMA_VERSION = 1
# Provider connection vocabulary. Declared availability is not a live probe —
# the worker registry owns the declaration, and adapters (#800) own live checks.
CONNECTION_DECLARED_AVAILABLE = "declared_available"
CONNECTION_DECLARED_UNAVAILABLE = "declared_unavailable"
CONNECTION_REGISTRY_UNAVAILABLE = "registry_unavailable"
# Insight kinds. Each generator is a pure function over one evidence source.
INSIGHT_BLOCKED_QUEUE = "blocked_queue_pressure"
INSIGHT_CONTROLLER_ATTENTION = "controller_attention"
INSIGHT_STALE_RUNTIME = "stale_runtime_risk"
INSIGHT_PROVIDER_WITHOUT_WORKERS = "provider_without_workers"
INSIGHT_ANALYTICS_FAILURE_RATE = "analytics_failure_pressure"
SEVERITY_INFO = "info"
SEVERITY_WARN = "warn"
SEVERITY_CRITICAL = "critical"
SEVERITY_UNPROVEN = "unproven"
CONFIDENCE_HIGH = "high"
CONFIDENCE_MEDIUM = "medium"
CONFIDENCE_LOW = "low"
CONFIDENCE_UNPROVEN = "unproven"
def _redact(value: Any) -> Any:
if value is None:
return None
return console_redaction.redact_text(str(value))
def _offline_test_mode() -> bool:
return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
"1",
"true",
"yes",
"on",
}
# --- Provider connection status ------------------------------------------------
@dataclass(frozen=True)
class ProviderConnection:
"""One AI provider's declared connection status (no secrets, no live probe)."""
provider_id: str
display_name: str
vendor: str
executable: str
connection_status: str
available_declared: bool
models: tuple[str, ...]
worker_count: int
enabled_worker_count: int
notes: str
#: Explicit statement of what was *not* proven (live process health, etc.).
probe_limit: str
def to_dict(self) -> dict[str, Any]:
return {
"provider_id": self.provider_id,
"display_name": self.display_name,
"vendor": self.vendor,
"executable": self.executable,
"connection_status": self.connection_status,
"available_declared": self.available_declared,
"models": list(self.models),
"worker_count": self.worker_count,
"enabled_worker_count": self.enabled_worker_count,
"notes": self.notes,
"probe_limit": self.probe_limit,
# Always true for this surface: keys are never loaded.
"secrets_exposed": False,
}
@dataclass(frozen=True)
class ProviderSnapshot:
ok: bool
providers: tuple[ProviderConnection, ...] = ()
registry_revision: int | None = None
registry_path: str | None = None
fetch_error: str | None = None
schema_version: int = INSIGHTS_SCHEMA_VERSION
def to_dict(self) -> dict[str, Any]:
return {
"ok": self.ok,
"schema_version": self.schema_version,
"registry_revision": self.registry_revision,
"registry_path": self.registry_path,
"fetch_error": self.fetch_error,
"providers": [p.to_dict() for p in self.providers],
"interpretation_limits": [
"connection_status reflects the worker registry declaration only",
"no API keys or credential material are loaded or rendered",
"live executable health is not probed on this surface (#800 owns that)",
],
}
_PROBE_LIMIT = (
"Declared status only. This console does not probe the provider executable "
"or call vendor APIs; live health belongs to the provider adapter framework."
)
def connection_status_for(provider: ProviderRecord) -> str:
return (
CONNECTION_DECLARED_AVAILABLE
if provider.available
else CONNECTION_DECLARED_UNAVAILABLE
)
def build_provider_connection(
provider: ProviderRecord,
workers: Sequence[WorkerRecord],
) -> ProviderConnection:
enabled = sum(1 for worker in workers if worker.enabled)
return ProviderConnection(
provider_id=provider.id,
display_name=str(_redact(provider.display_name) or provider.id),
vendor=str(_redact(provider.vendor) or ""),
executable=str(_redact(provider.executable) or ""),
connection_status=connection_status_for(provider),
available_declared=bool(provider.available),
models=tuple(str(_redact(m) or m) for m in provider.models),
worker_count=len(workers),
enabled_worker_count=enabled,
notes=str(_redact(provider.notes) or ""),
probe_limit=_PROBE_LIMIT,
)
def load_provider_snapshot(
*,
registry: WorkerRegistry | None = None,
registry_loader: Callable[[], WorkerRegistry] | None = None,
) -> ProviderSnapshot:
"""Load declared provider connections. Never raises for missing registry."""
if registry is None:
loader = registry_loader or load_worker_registry
try:
if _offline_test_mode() and registry_loader is None:
return ProviderSnapshot(
ok=False,
fetch_error=(
"provider registry not loaded in offline test mode "
"(inject a registry for unit tests)"
),
)
registry = loader()
except Exception as exc: # fail soft — operator-visible reason
return ProviderSnapshot(
ok=False,
fetch_error=str(_redact(f"worker registry unavailable: {exc}")),
)
connections = tuple(
build_provider_connection(provider, workers_for_provider(registry, provider.id))
for provider in registry.providers
)
return ProviderSnapshot(
ok=True,
providers=connections,
registry_revision=registry.revision,
registry_path=str(registry.source_path),
)
# --- Evidence-backed insights --------------------------------------------------
@dataclass(frozen=True)
class EvidenceRef:
"""One durable reference an insight is allowed to cite."""
kind: str # issue | pr | provider | health | analytics | traffic
ref: str
detail: str
def to_dict(self) -> dict[str, Any]:
return {
"kind": self.kind,
"ref": self.ref,
"detail": str(_redact(self.detail) or ""),
}
@dataclass(frozen=True)
class Insight:
"""One advisory finding. Never a claim that an action completed."""
insight_id: str
kind: str
severity: str
confidence: str
title: str
summary: str
evidence: tuple[EvidenceRef, ...]
advisory_only: bool = True
claims_action_completed: bool = False
def to_dict(self) -> dict[str, Any]:
return {
"insight_id": self.insight_id,
"kind": self.kind,
"severity": self.severity,
"confidence": self.confidence,
"title": str(_redact(self.title) or ""),
"summary": str(_redact(self.summary) or ""),
"evidence": [item.to_dict() for item in self.evidence],
"advisory_only": self.advisory_only,
"claims_action_completed": self.claims_action_completed,
}
@dataclass(frozen=True)
class InsightsSnapshot:
ok: bool
insights: tuple[Insight, ...] = ()
sources_used: tuple[str, ...] = ()
sources_unavailable: tuple[dict[str, str], ...] = ()
fetch_error: str | None = None
schema_version: int = INSIGHTS_SCHEMA_VERSION
def to_dict(self) -> dict[str, Any]:
return {
"ok": self.ok,
"schema_version": self.schema_version,
"insights": [insight.to_dict() for insight in self.insights],
"sources_used": list(self.sources_used),
"sources_unavailable": list(self.sources_unavailable),
"fetch_error": self.fetch_error,
"interpretation_limits": [
"insights are advisory only and never authorize merge, review, or close",
"an insight without evidence refs is refused rather than emitted",
"a missing source is listed under sources_unavailable, not as an empty success",
"insights never claim a workflow action completed",
],
}
def _require_evidence(evidence: Sequence[EvidenceRef]) -> tuple[EvidenceRef, ...]:
"""Fail closed: an insight with no evidence must not be emitted."""
items = tuple(evidence)
if not items:
raise ValueError("insight requires at least one evidence ref")
return items
def insight_blocked_queue(traffic: Any) -> Insight | None:
"""Traffic blocked bucket pressure with per-item evidence."""
blocked = tuple(getattr(traffic, "blocked", ()) or ())
if not blocked:
return None
evidence = []
for item in blocked[:20]:
kind = str(getattr(item, "kind", "issue") or "issue")
number = int(getattr(item, "number", 0) or 0)
if number <= 0:
continue
reason = getattr(item, "block_reason", None) or "blocked"
evidence.append(
EvidenceRef(
kind=kind,
ref=f"#{number}",
detail=f"traffic_state=blocked; reason={reason}",
)
)
if not evidence:
return None
count = len(blocked)
severity = SEVERITY_CRITICAL if count >= 10 else SEVERITY_WARN
return Insight(
insight_id=f"{INSIGHT_BLOCKED_QUEUE}:{count}",
kind=INSIGHT_BLOCKED_QUEUE,
severity=severity,
confidence=(
CONFIDENCE_HIGH
if getattr(traffic, "inventory_complete", False)
else CONFIDENCE_MEDIUM
),
title=f"{count} blocked work item(s) in traffic control",
summary=(
f"Traffic control reports {count} blocked item(s). "
"This is an observation of the loaded window, not a claim that "
"any remediation ran."
),
evidence=_require_evidence(evidence),
)
def insight_controller_attention(traffic: Any) -> Insight | None:
needs = tuple(getattr(traffic, "needs_controller", ()) or ())
if not needs:
return None
evidence = []
for item in needs[:20]:
kind = str(getattr(item, "kind", "issue") or "issue")
number = int(getattr(item, "number", 0) or 0)
if number <= 0:
continue
evidence.append(
EvidenceRef(
kind=kind,
ref=f"#{number}",
detail="traffic_state=needs_controller",
)
)
if not evidence:
return None
count = len(needs)
return Insight(
insight_id=f"{INSIGHT_CONTROLLER_ATTENTION}:{count}",
kind=INSIGHT_CONTROLLER_ATTENTION,
severity=SEVERITY_WARN if count else SEVERITY_INFO,
confidence=(
CONFIDENCE_HIGH
if getattr(traffic, "inventory_complete", False)
else CONFIDENCE_MEDIUM
),
title=f"{count} item(s) need controller attention",
summary=(
f"Traffic control marks {count} item(s) as needs_controller. "
"Advisory only — the controller allocator remains the authority "
"for routing."
),
evidence=_require_evidence(evidence),
)
def insight_stale_runtime(health: Any) -> Insight | None:
stale = getattr(health, "stale_runtime", None)
if stale is None:
return None
mutation_safe = bool(getattr(stale, "mutation_safe", False))
is_stale = bool(getattr(stale, "stale", False))
determinable = bool(getattr(stale, "determinable", False))
if mutation_safe and not is_stale:
return None
daemon = getattr(stale, "daemon_head", None) or "unknown"
checkout = getattr(stale, "checkout_head", None) or "unknown"
remote = getattr(stale, "remote_head", None) or "unknown"
if not determinable:
severity = SEVERITY_UNPROVEN
confidence = CONFIDENCE_UNPROVEN
title = "Runtime parity is not determinable"
summary = (
"System health could not prove mutation_safe. This is not proof "
"that the runtime is stale — only that parity was unproven."
)
else:
severity = SEVERITY_CRITICAL if is_stale else SEVERITY_WARN
confidence = CONFIDENCE_HIGH
title = "Stale or mutation-unsafe runtime"
summary = (
"System health reports a runtime that is not mutation_safe. "
"No restart or recovery is claimed by this insight."
)
return Insight(
insight_id=f"{INSIGHT_STALE_RUNTIME}:{daemon}:{checkout}",
kind=INSIGHT_STALE_RUNTIME,
severity=severity,
confidence=confidence,
title=title,
summary=summary,
evidence=_require_evidence(
(
EvidenceRef(
kind="health",
ref="stale_runtime",
detail=(
f"stale={is_stale}; mutation_safe={mutation_safe}; "
f"determinable={determinable}; daemon={daemon}; "
f"checkout={checkout}; remote={remote}"
),
),
)
),
)
def insight_providers_without_workers(
providers: Sequence[ProviderConnection],
) -> Insight | None:
lonely = [
provider
for provider in providers
if provider.available_declared and provider.worker_count == 0
]
if not lonely:
return None
evidence = tuple(
EvidenceRef(
kind="provider",
ref=provider.provider_id,
detail=(
f"available_declared=true; worker_count=0; "
f"vendor={provider.vendor}"
),
)
for provider in lonely
)
return Insight(
insight_id=f"{INSIGHT_PROVIDER_WITHOUT_WORKERS}:{len(lonely)}",
kind=INSIGHT_PROVIDER_WITHOUT_WORKERS,
severity=SEVERITY_INFO,
confidence=CONFIDENCE_HIGH,
title=f"{len(lonely)} declared-available provider(s) have no workers",
summary=(
"The worker registry declares these providers available but no "
"worker instance names them. This is a configuration observation, "
"not a claim that a provider process is running or idle."
),
evidence=_require_evidence(evidence),
)
def insight_analytics_failures(analytics: Any) -> Insight | None:
"""Flag elevated non-ok stage status in analytics when events exist."""
if analytics is None or not getattr(analytics, "ok", False):
return None
events = tuple(getattr(analytics, "events", ()) or ())
if not events:
return None
failed = [
event
for event in events
if str(getattr(event, "status", "") or "").lower()
in {"error", "failed", "failure"}
]
if not failed:
return None
# Cap evidence so a large window stays readable.
evidence = []
for event in failed[:20]:
usage_id = getattr(event, "usage_id", None)
issue = getattr(event, "issue_number", None)
pr = getattr(event, "pr_number", None)
if pr is not None:
ref_kind, ref = "pr", f"#{int(pr)}"
elif issue is not None:
ref_kind, ref = "issue", f"#{int(issue)}"
else:
ref_kind, ref = "analytics", f"usage:{usage_id}"
evidence.append(
EvidenceRef(
kind=ref_kind,
ref=ref,
detail=(
f"status={getattr(event, 'status', '')}; "
f"stage={getattr(event, 'stage', '')}; "
f"model={getattr(event, 'model', '')}"
),
)
)
if not evidence:
return None
rate = len(failed) / max(len(events), 1)
return Insight(
insight_id=f"{INSIGHT_ANALYTICS_FAILURE_RATE}:{len(failed)}:{len(events)}",
kind=INSIGHT_ANALYTICS_FAILURE_RATE,
severity=SEVERITY_WARN if rate >= 0.1 else SEVERITY_INFO,
confidence=CONFIDENCE_MEDIUM,
title=f"{len(failed)} analytics event(s) reported failure status",
summary=(
f"{len(failed)} of {len(events)} loaded analytics events carry a "
"failure status. Advisory only — this is not a gate decision."
),
evidence=_require_evidence(evidence),
)
def generate_insights(
*,
traffic: Any | None = None,
health: Any | None = None,
provider_snapshot: ProviderSnapshot | None = None,
analytics: Any | None = None,
) -> tuple[tuple[Insight, ...], tuple[str, ...], tuple[dict[str, str], ...]]:
"""Pure multi-source insight generation. Never mutates inputs."""
insights: list[Insight] = []
used: list[str] = []
unavailable: list[dict[str, str]] = []
if traffic is None:
unavailable.append(
{"source": "traffic", "reason": "traffic snapshot not supplied"}
)
elif getattr(traffic, "fetch_error", None):
unavailable.append(
{
"source": "traffic",
"reason": str(_redact(traffic.fetch_error) or "traffic fetch failed"),
}
)
else:
used.append("traffic")
for builder in (insight_blocked_queue, insight_controller_attention):
try:
item = builder(traffic)
except ValueError:
continue
if item is not None:
insights.append(item)
if health is None:
unavailable.append(
{"source": "system_health", "reason": "system health snapshot not supplied"}
)
else:
used.append("system_health")
try:
item = insight_stale_runtime(health)
except ValueError:
item = None
if item is not None:
insights.append(item)
if provider_snapshot is None:
unavailable.append(
{"source": "providers", "reason": "provider snapshot not supplied"}
)
elif not provider_snapshot.ok:
unavailable.append(
{
"source": "providers",
"reason": str(
_redact(provider_snapshot.fetch_error)
or "provider registry unavailable"
),
}
)
else:
used.append("providers")
try:
item = insight_providers_without_workers(provider_snapshot.providers)
except ValueError:
item = None
if item is not None:
insights.append(item)
if analytics is None:
unavailable.append(
{"source": "analytics", "reason": "analytics snapshot not supplied"}
)
elif not getattr(analytics, "ok", False):
unavailable.append(
{
"source": "analytics",
"reason": str(
_redact(getattr(analytics, "fetch_error", None))
or "analytics snapshot not ok"
),
}
)
else:
used.append("analytics")
try:
item = insight_analytics_failures(analytics)
except ValueError:
item = None
if item is not None:
insights.append(item)
# Stable ordering: severity then kind.
_sev_rank = {
SEVERITY_CRITICAL: 0,
SEVERITY_WARN: 1,
SEVERITY_INFO: 2,
SEVERITY_UNPROVEN: 3,
}
insights.sort(key=lambda i: (_sev_rank.get(i.severity, 9), i.kind, i.insight_id))
return tuple(insights), tuple(used), tuple(unavailable)
def load_insights_snapshot(
*,
traffic: Any | None = None,
health: Any | None = None,
provider_snapshot: ProviderSnapshot | None = None,
analytics: Any | None = None,
load_live: bool = True,
) -> InsightsSnapshot:
"""Compose insights from injected or live console evidence sources."""
sources_unavailable: list[dict[str, str]] = []
if load_live and traffic is None and not _offline_test_mode():
try:
from webui.traffic_loader import load_traffic_snapshot
traffic = load_traffic_snapshot()
except Exception as exc: # fail soft
sources_unavailable.append(
{
"source": "traffic",
"reason": str(_redact(f"traffic load failed: {exc}")),
}
)
traffic = None
if load_live and health is None and not _offline_test_mode():
try:
from webui.system_health import load_system_health
health = load_system_health()
except Exception as exc:
sources_unavailable.append(
{
"source": "system_health",
"reason": str(_redact(f"system health load failed: {exc}")),
}
)
health = None
if provider_snapshot is None:
provider_snapshot = load_provider_snapshot()
if load_live and analytics is None and not _offline_test_mode():
try:
from webui.analytics_loader import load_analytics
analytics = load_analytics()
except Exception as exc:
sources_unavailable.append(
{
"source": "analytics",
"reason": str(_redact(f"analytics load failed: {exc}")),
}
)
analytics = None
insights, used, unavailable = generate_insights(
traffic=traffic,
health=health,
provider_snapshot=provider_snapshot,
analytics=analytics,
)
merged_unavailable = tuple(sources_unavailable) + unavailable
# ok when at least one source contributed or we can honestly report absence.
ok = bool(used) or bool(merged_unavailable)
return InsightsSnapshot(
ok=ok,
insights=insights,
sources_used=used,
sources_unavailable=merged_unavailable,
fetch_error=None
if used
else (
"no evidence sources produced a usable snapshot"
if merged_unavailable
else "no insight sources ran"
),
)
def snapshot_providers_to_dict(snapshot: ProviderSnapshot) -> dict[str, Any]:
return snapshot.to_dict()
def snapshot_insights_to_dict(snapshot: InsightsSnapshot) -> dict[str, Any]:
return snapshot.to_dict()
+196
View File
@@ -0,0 +1,196 @@
"""HTML views for AI-provider connections and operational insights (#650)."""
from __future__ import annotations
from html import escape
from webui.insights_loader import (
CONNECTION_DECLARED_AVAILABLE,
CONNECTION_DECLARED_UNAVAILABLE,
InsightsSnapshot,
ProviderSnapshot,
SEVERITY_CRITICAL,
SEVERITY_INFO,
SEVERITY_UNPROVEN,
SEVERITY_WARN,
)
from webui.layout import render_page
_SEVERITY_CSS = {
SEVERITY_CRITICAL: "badge-blocked",
SEVERITY_WARN: "badge-health-degraded",
SEVERITY_INFO: "badge-health-ok",
SEVERITY_UNPROVEN: "badge-health-unproven",
}
_CONN_CSS = {
CONNECTION_DECLARED_AVAILABLE: "badge-health-ok",
CONNECTION_DECLARED_UNAVAILABLE: "badge-health-degraded",
"registry_unavailable": "badge-blocked",
}
def _badge(text: str, css: str) -> str:
return f'<span class="badge {css}">{escape(text)}</span>'
def _limits_card(lines: list[str], *, title: str) -> str:
items = "".join(f"<li>{escape(line)}</li>" for line in lines)
return f"""<div class="prompt-card">
<h3>{escape(title)}</h3>
<ul class="reasons">{items}</ul>
<p class="muted">Advisory surface only — no review, merge, close, or provider
mutation is available here.</p>
</div>"""
def render_providers_page(snapshot: ProviderSnapshot) -> str:
"""Render the AI-provider connections page."""
if not snapshot.ok:
body = f"""<h2>AI provider connections</h2>
<p class="meta">Phase 4 read-only provider status (#650).</p>
<div class="health-card health-stale">
<strong>Provider registry unavailable:</strong>
{escape(snapshot.fetch_error or "registry could not be loaded")}.
No connection table is rendered — an empty table would claim that no
providers are configured.
</div>
{_limits_card([
"connection_status reflects the worker registry declaration only",
"no API keys or credential material are loaded or rendered",
"live executable health is not probed on this surface",
], title="Interpretation limits")}
"""
return render_page(title="Providers", body_html=body)
rows = []
for provider in snapshot.providers:
models = (
", ".join(f"<code>{escape(m)}</code>" for m in provider.models)
if provider.models
else '<span class="muted">none declared</span>'
)
rows.append(
"<tr>"
f"<td><code>{escape(provider.provider_id)}</code></td>"
f"<td>{escape(provider.display_name)}</td>"
f"<td>{escape(provider.vendor)}</td>"
f"<td><code>{escape(provider.executable)}</code></td>"
f"<td>{_badge(provider.connection_status, _CONN_CSS.get(provider.connection_status, 'badge-health-skipped'))}</td>"
f"<td>{provider.worker_count} "
f"({provider.enabled_worker_count} enabled)</td>"
f"<td>{models}</td>"
"</tr>"
)
table = (
"".join(rows)
if rows
else '<tr><td colspan="7" class="muted">No providers declared in the registry.</td></tr>'
)
body = f"""<h2>AI provider connections</h2>
<p class="meta">Phase 4 read-only provider status (#650). Registry revision
<code>{escape(str(snapshot.registry_revision))}</code>.
Declared status only — secrets never load.</p>
<div class="health-card">
<h3>Declared connections</h3>
<table class="registry">
<thead>
<tr>
<th>Provider</th><th>Name</th><th>Vendor</th><th>Executable</th>
<th>Connection</th><th>Workers</th><th>Models (declared)</th>
</tr>
</thead>
<tbody>{table}</tbody>
</table>
<p class="muted">{escape(snapshot.providers[0].probe_limit if snapshot.providers else "")}</p>
</div>
{_limits_card([
"connection_status reflects the worker registry declaration only",
"no API keys or credential material are loaded or rendered",
"live executable health is not probed on this surface (#800 owns that)",
], title="Interpretation limits")}
<p class="muted">Related: <a href="/insights">Operational insights</a> ·
<a href="/analytics">Analytics</a></p>
"""
return render_page(title="Providers", body_html=body)
def _evidence_list(insight) -> str:
items = "".join(
f"<li><code>{escape(ref.kind)}:{escape(ref.ref)}</code> — "
f"{escape(ref.detail)}</li>"
for ref in insight.evidence
)
return f'<ul class="reasons">{items}</ul>'
def render_insights_page(snapshot: InsightsSnapshot) -> str:
"""Render the operational insights page."""
if not snapshot.ok and not snapshot.insights:
body = f"""<h2>Operational insights</h2>
<p class="meta">Phase 4 evidence-backed insights (#650).</p>
<div class="health-card health-stale">
<strong>Insights unavailable:</strong>
{escape(snapshot.fetch_error or "no sources ran")}.
</div>
{_limits_card([
"insights are advisory only and never authorize merge, review, or close",
"an insight without evidence refs is refused rather than emitted",
], title="Interpretation limits")}
"""
return render_page(title="Insights", body_html=body)
source_bits = []
if snapshot.sources_used:
source_bits.append(
"sources used: " + ", ".join(f"<code>{escape(s)}</code>" for s in snapshot.sources_used)
)
if snapshot.sources_unavailable:
missing = "; ".join(
f"{escape(item.get('source', '?'))}: {escape(item.get('reason', ''))}"
for item in snapshot.sources_unavailable
)
source_bits.append(f"sources unavailable: {missing}")
cards = []
for insight in snapshot.insights:
cards.append(
f"""<div class="prompt-card">
<h3>{_badge(insight.severity, _SEVERITY_CSS.get(insight.severity, "badge-health-skipped"))}
{escape(insight.title)}</h3>
<p class="meta"><code>{escape(insight.kind)}</code> · confidence
<code>{escape(insight.confidence)}</code> ·
{_badge("advisory only", "badge-health-skipped")} ·
{_badge("no action claimed", "badge-health-ok")}</p>
<p>{escape(insight.summary)}</p>
<h4>Evidence</h4>
{_evidence_list(insight)}
</div>"""
)
if not cards:
cards.append(
'<div class="prompt-card"><p class="muted">No insights met the '
"evidence threshold in the loaded sources. That is not a claim "
"that the fleet is healthy — only that no qualifying pattern was "
"found.</p></div>"
)
body = f"""<h2>Operational insights</h2>
<p class="meta">Phase 4 evidence-backed insights (#650). Derived only from
durable console evidence; never invents policy or completes workflow actions.</p>
<p class="muted">{" · ".join(source_bits) if source_bits else ""}</p>
{"".join(cards)}
{_limits_card([
"insights are advisory only and never authorize merge, review, or close",
"an insight without evidence refs is refused rather than emitted",
"a missing source is listed as unavailable, not as an empty success",
"insights never claim a workflow action completed",
], title="Interpretation limits")}
<p class="muted">Related: <a href="/providers">Provider connections</a> ·
<a href="/traffic">Traffic</a> · <a href="/system-health">System health</a> ·
<a href="/analytics">Analytics</a></p>
"""
return render_page(title="Insights", body_html=body)
+6 -11
View File
@@ -4,12 +4,11 @@ Single source of truth for the console navigation so ``webui/layout.py`` and
the ``webui/app.py`` route table stay aligned with epic #631. Read-only: every
destination is a GET view or a Phase 1 placeholder. No mutation links.
Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
Nav groups follow the #631 information architecture: Health, Traffic,
Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
Insights (placeholder), joined by the Phase 3 Gitea linkage group (#645).
Later-phase surfaces are declared as ``stub`` items and
backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
instead of a 404.
Gitea linkage (#645) plus Phase 4 Insights/Providers (#650). Later-phase
surfaces are declared as ``stub`` items and backed by ``STUB_PAGES`` so their
nav links resolve to a graceful placeholder instead of a 404.
"""
from __future__ import annotations
@@ -69,7 +68,8 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
NavItem("/prompts", "Prompts"),
)),
NavGroup("Insights", (
NavItem("/insights", "Insights", "stub"),
NavItem("/insights", "Insights"),
NavItem("/providers", "Providers"),
NavItem("/analytics", "Analytics"),
NavItem("/audit", "Audit"),
)),
@@ -93,11 +93,6 @@ STUB_PAGES: dict[str, tuple[str, str]] = {
"Policy",
"Capability and role policy surface. Placeholder until a later phase.",
),
"/insights": (
"Insights",
"Aggregate operational insights and trends. Placeholder until a later "
"phase.",
),
}