feat(observability): stream scoped service health transitions

ods/extensions/services/dashboard-api/README.md

@@ -52,6 +52,7 @@ Environment variables (set in `.env`):
 | `GET` | `/status` | Yes | Full system status (all above combined) |
 | `GET` | `/api/status` | Yes | Dashboard-formatted status with inference metrics |
 | `GET` | `/api/host-agent/diagnostics` | Yes | Host-agent URL, gateway, auth, and live probe diagnostics |
+| `GET` | `/api/events/services` | Yes | SSE stream of deduplicated service-health transitions; repeat `service=` to scope the stream |
 
 ### Preflight
 

ods/extensions/services/dashboard-api/main.py

@@ -72,6 +72,7 @@
     tailscale,
     usage,
     node,
+    events,
 )
 from settings import (
     _ENV_ASSIGNMENT_RE, _ENV_COMMENTED_ASSIGNMENT_RE, _SETTINGS_APPLY_ALLOWED_SERVICES, _parse_env_text, _read_env_map_from_path,
@@ -1114,6 +1115,7 @@ def get_allowed_origins():
 app.include_router(tailscale.router)
 app.include_router(usage.router)
 app.include_router(node.router)
+app.include_router(events.router)
 
 
 # ================================================================

ods/extensions/services/dashboard-api/routers/events.py

@@ -0,0 +1,140 @@
+"""Server-sent service health events for operator automation."""
+
+import asyncio
+import hashlib
+import json
+import time
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, Header, Query, Request
+from fastapi.responses import StreamingResponse
+
+from helpers import get_cached_services
+from security import verify_api_key
+
+
+router = APIRouter(tags=["events"])
+
+
+class ServiceStateTracker:
+    """Convert cached health snapshots into deduplicated state transitions."""
+
+    def __init__(self) -> None:
+        self._fingerprint: str | None = None
+        self._states: dict[str, str] | None = None
+
+    def observe(self, statuses: list, *, cache_ready: bool) -> dict | None:
+        states = {
+            str(service.id): str(service.status)
+            for service in statuses
+        }
+        canonical = json.dumps(
+            {"cache_ready": cache_ready, "states": states},
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        fingerprint = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20]
+        if fingerprint == self._fingerprint:
+            return None
+
+        changes = []
+        if self._states is not None:
+            for service_id in sorted(set(self._states) | set(states)):
+                previous = self._states.get(service_id)
+                current = states.get(service_id)
+                if previous != current:
+                    changes.append({
+                        "id": service_id,
+                        "from": previous,
+                        "to": current or "removed",
+                    })
+
+        self._fingerprint = fingerprint
+        self._states = states
+        return {
+            "id": fingerprint,
+            "event": "services",
+            "data": {
+                "observed_at": datetime.now(timezone.utc).isoformat(),
+                "cache_ready": cache_ready,
+                "services": [
+                    {"id": service_id, "status": states[service_id]}
+                    for service_id in sorted(states)
+                ],
+                "changes": changes,
+            },
+        }
+
+
+def format_sse(observation: dict) -> str:
+    payload = json.dumps(observation["data"], sort_keys=True, separators=(",", ":"))
+    return (
+        f"id: {observation['id']}\n"
+        f"event: {observation['event']}\n"
+        f"data: {payload}\n\n"
+    )
+
+
+async def service_event_stream(
+    request: Request,
+    *,
+    poll_seconds: float,
+    heartbeat_seconds: float,
+    last_event_id: str | None,
+    max_events: int | None,
+    service_ids: frozenset[str] | None,
+):
+    tracker = ServiceStateTracker()
+    emitted = 0
+    last_heartbeat = time.monotonic()
+
+    while not await request.is_disconnected():
+        cached = get_cached_services()
+        statuses = cached or []
+        if service_ids is not None:
+            statuses = [
+                service
+                for service in statuses
+                if str(service.id) in service_ids
+            ]
+        observation = tracker.observe(statuses, cache_ready=cached is not None)
+        if observation is not None and observation["id"] != last_event_id:
+            yield format_sse(observation)
+            emitted += 1
+            last_event_id = observation["id"]
+            last_heartbeat = time.monotonic()
+            if max_events is not None and emitted >= max_events:
+                return
+        elif time.monotonic() - last_heartbeat >= heartbeat_seconds:
+            yield ": keep-alive\n\n"
+            last_heartbeat = time.monotonic()
+        await asyncio.sleep(poll_seconds)
+
+
+@router.get("/api/events/services")
+async def service_events(
+    request: Request,
+    poll_seconds: float = Query(2.0, ge=0.5, le=30.0),
+    heartbeat_seconds: float = Query(15.0, ge=5.0, le=120.0),
+    max_events: int | None = Query(None, ge=1, le=1000),
+    service: list[str] | None = Query(None),
+    last_event_id: str | None = Header(None, alias="Last-Event-ID"),
+    api_key: str = Depends(verify_api_key),
+):
+    """Stream deduplicated service-status snapshots and transitions as SSE."""
+    stream = service_event_stream(
+        request,
+        poll_seconds=poll_seconds,
+        heartbeat_seconds=heartbeat_seconds,
+        last_event_id=last_event_id,
+        max_events=max_events,
+        service_ids=frozenset(service) if service else None,
+    )
+    return StreamingResponse(
+        stream,
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )

ods/extensions/services/dashboard-api/tests/test_service_events.py

@@ -0,0 +1,156 @@
+"""Tests for the authenticated service-health event stream."""
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from routers.events import ServiceStateTracker, format_sse, service_event_stream
+
+
+def _service(service_id, status, response_time_ms=1):
+    return SimpleNamespace(id=service_id, status=status, response_time_ms=response_time_ms)
+
+
+def _event_data(event_text):
+    data_line = next(line for line in event_text.splitlines() if line.startswith("data: "))
+    return json.loads(data_line.removeprefix("data: "))
+
+
+def test_tracker_emits_only_state_changes_and_reports_transitions():
+    tracker = ServiceStateTracker()
+
+    initial = tracker.observe([_service("api", "healthy", 10)], cache_ready=True)
+    latency_only = tracker.observe([_service("api", "healthy", 900)], cache_ready=True)
+    changed = tracker.observe(
+        [_service("api", "unhealthy"), _service("worker", "healthy")],
+        cache_ready=True,
+    )
+
+    assert initial["data"]["changes"] == []
+    assert latency_only is None
+    assert changed["data"]["changes"] == [
+        {"id": "api", "from": "healthy", "to": "unhealthy"},
+        {"id": "worker", "from": None, "to": "healthy"},
+    ]
+
+
+def test_tracker_reports_removed_services_and_cache_readiness():
+    tracker = ServiceStateTracker()
+    waiting = tracker.observe([], cache_ready=False)
+    ready = tracker.observe([_service("api", "healthy")], cache_ready=True)
+    removed = tracker.observe([], cache_ready=True)
+
+    assert waiting["data"]["cache_ready"] is False
+    assert ready["data"]["changes"] == [
+        {"id": "api", "from": None, "to": "healthy"},
+    ]
+    assert removed["data"]["changes"] == [
+        {"id": "api", "from": "healthy", "to": "removed"},
+    ]
+
+
+def test_sse_format_has_resumable_id_event_and_compact_json():
+    observation = ServiceStateTracker().observe(
+        [_service("api", "healthy")],
+        cache_ready=True,
+    )
+
+    rendered = format_sse(observation)
+
+    assert rendered.startswith(f"id: {observation['id']}\nevent: services\n")
+    assert rendered.endswith("\n\n")
+    assert _event_data(rendered)["services"] == [{"id": "api", "status": "healthy"}]
+
+
+@pytest.mark.asyncio
+async def test_stream_resumes_after_last_event_id_and_waits_for_a_change(monkeypatch):
+    snapshots = iter([
+        [_service("api", "healthy")],
+        [_service("api", "unhealthy")],
+    ])
+    monkeypatch.setattr("routers.events.get_cached_services", lambda: next(snapshots))
+    previous = ServiceStateTracker().observe(
+        [_service("api", "healthy")],
+        cache_ready=True,
+    )
+
+    class ConnectedRequest:
+        async def is_disconnected(self):
+            return False
+
+    stream = service_event_stream(
+        ConnectedRequest(),
+        poll_seconds=0,
+        heartbeat_seconds=15,
+        last_event_id=previous["id"],
+        max_events=1,
+        service_ids=None,
+    )
+    resumed = await anext(stream)
+
+    assert _event_data(resumed)["changes"] == [
+        {"id": "api", "from": "healthy", "to": "unhealthy"},
+    ]
+
+
+def test_http_stream_can_emit_one_bounded_snapshot(test_client, monkeypatch):
+    monkeypatch.setattr(
+        "routers.events.get_cached_services",
+        lambda: [_service("dashboard-api", "healthy"), _service("llama-server", "starting")],
+    )
+
+    response = test_client.get(
+        "/api/events/services?max_events=1",
+        headers=test_client.auth_headers,
+    )
+
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/event-stream")
+    assert response.headers["cache-control"] == "no-cache"
+    assert response.headers["x-accel-buffering"] == "no"
+    assert "event: services" in response.text
+    assert _event_data(response.text)["services"] == [
+        {"id": "dashboard-api", "status": "healthy"},
+        {"id": "llama-server", "status": "starting"},
+    ]
+
+
+def test_http_stream_can_scope_snapshots_to_repeated_service_filters(
+    test_client,
+    monkeypatch,
+):
+    monkeypatch.setattr(
+        "routers.events.get_cached_services",
+        lambda: [
+            _service("dashboard-api", "healthy"),
+            _service("llama-server", "starting"),
+            _service("open-webui", "healthy"),
+        ],
+    )
+
+    response = test_client.get(
+        "/api/events/services?max_events=1&service=open-webui&service=llama-server",
+        headers=test_client.auth_headers,
+    )
+
+    assert response.status_code == 200
+    assert _event_data(response.text)["services"] == [
+        {"id": "llama-server", "status": "starting"},
+        {"id": "open-webui", "status": "healthy"},
+    ]
+
+
+def test_http_stream_requires_authentication(test_client):
+    response = test_client.get("/api/events/services?max_events=1")
+
+    assert response.status_code == 401
+
+
+def test_http_stream_validates_timing_and_event_bounds(test_client):
+    response = test_client.get(
+        "/api/events/services?poll_seconds=0.1&heartbeat_seconds=1&max_events=0",
+        headers=test_client.auth_headers,
+    )
+
+    assert response.status_code == 422