diff --git a/solstone/apps/observer/routes.py b/solstone/apps/observer/routes.py index d1e273dc3..ba0612eb4 100644 --- a/solstone/apps/observer/routes.py +++ b/solstone/apps/observer/routes.py @@ -18,16 +18,19 @@ import base64 import json import logging import platform +import queue import re import secrets from pathlib import Path from typing import Any -from flask import Blueprint, jsonify, request +from flask import Blueprint, Response, jsonify, request, stream_with_context from werkzeug.utils import secure_filename +import solstone.convey.bridge as convey_bridge from solstone.apps.utils import log_app_action from solstone.convey import emit +from solstone.convey.copy import OBSERVER_CALLOSUM_LIVE_LABEL from solstone.observe.utils import ( MAX_SEGMENT_ATTEMPTS, compute_bytes_sha256, @@ -55,12 +58,17 @@ observer_bp = Blueprint( __name__, url_prefix="/app/observer", ) +OBSERVER_CALLOSUM_SSE_ROUTE = "/app/observer//callosum" +_OBSERVER_CALLOSUM_SSE_RULE = OBSERVER_CALLOSUM_SSE_ROUTE.removeprefix( + observer_bp.url_prefix or "" +) # Key length in bytes (256 bits = 32 bytes) KEY_BYTES = 32 ACTIVE_THRESHOLD_MS = 30_000 STALE_THRESHOLD_MS = 120_000 FUTURE_CLOCK_DRIFT_TOLERANCE_MS = 5 * 60 * 1000 +_SSE_HEARTBEAT_SECONDS = 20 OBSERVER_STATE_LABELS = { "connected": "Connected", @@ -152,8 +160,9 @@ def _serialize_observer(observer: dict[str, Any], current_now: int) -> dict[str, observer.get("revoked", False), current_now, ) + key_prefix = observer.get("key", "")[:8] return { - "key_prefix": observer.get("key", "")[:8], + "key_prefix": key_prefix, "name": observer.get("name", ""), "created_at": observer.get("created_at", 0), "last_seen": observer.get("last_seen"), @@ -162,6 +171,7 @@ def _serialize_observer(observer: dict[str, Any], current_now: int) -> dict[str, "revoked": observer.get("revoked", False), "revoked_at": observer.get("revoked_at"), "stats": observer.get("stats", {}), + "live": convey_bridge.subscription_count(key_prefix) > 0, **freshness, "label": OBSERVER_STATE_LABELS[str(freshness["state"])], } @@ -204,11 +214,86 @@ def api_list() -> Any: "active_ms": ACTIVE_THRESHOLD_MS, "stale_ms": STALE_THRESHOLD_MS, }, + "labels": { + "live": OBSERVER_CALLOSUM_LIVE_LABEL, + }, "observers": result, } ) +# LOCKED — wire format observer clients depend on. Field names and presence are the +# downstream contract. Adding a new field to a callosum event is permitted; renaming +# or removing existing fields requires a spec revision. +# +# Each SSE message body is a JSON object with at minimum: +# { +# "tract": str, # e.g. "chat", "observe", "cortex", "supervisor", ... +# "event": str, # the event name within the tract +# "ts": int, # millisecond timestamp +# ... event-specific fields, passed through as emitted by the bus +# } +# +# The feed does NOT add or remove fields relative to the bus payload. +# The feed does NOT filter events. +# The feed does NOT redact fields (v1 trust call; same trust boundary as the existing +# WebSocket bridge — observers are inside it). +@observer_bp.route(_OBSERVER_CALLOSUM_SSE_RULE, methods=["GET"]) +def callosum_sse(key: str) -> Any: + """Stream Callosum events to an authenticated observer process.""" + auth_key = _get_key(key) + if not auth_key: + return jsonify({"error": "Authorization required"}), 401 + + observer = load_observer(auth_key) + if not observer: + return jsonify({"error": "Invalid key"}), 401 + + if observer.get("revoked", False): + return jsonify({"error": "Observer revoked"}), 403 + + if not observer.get("enabled", True): + return jsonify({"error": "Observer disabled"}), 403 + + key_prefix = auth_key[:8] + handle = convey_bridge.register_sse_subscriber(key_prefix) + + def generate(): + try: + yield ": heartbeat\n\n" + while True: + if handle.dropped.is_set(): + return + try: + serialized_message = handle.queue.get( + timeout=_SSE_HEARTBEAT_SECONDS + ) + except queue.Empty: + current_observer = load_observer(auth_key) + if not current_observer: + return + if current_observer.get("revoked", False): + return + if not current_observer.get("enabled", True): + return + yield ": heartbeat\n\n" + continue + + if handle.dropped.is_set(): + return + yield f"data: {serialized_message}\n\n" + finally: + convey_bridge.unregister_sse_subscriber(handle) + + response = Response( + stream_with_context(generate()), + mimetype="text/event-stream", + ) + response.headers["Cache-Control"] = "no-cache" + response.headers["X-Accel-Buffering"] = "no" + return response + + @observer_bp.route("/api/create", methods=["POST"]) def api_create() -> Any: """Create a new observer registration.""" diff --git a/solstone/apps/observer/tests/test_callosum_sse.py b/solstone/apps/observer/tests/test_callosum_sse.py new file mode 100644 index 000000000..c837803ba --- /dev/null +++ b/solstone/apps/observer/tests/test_callosum_sse.py @@ -0,0 +1,204 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Tests for observer Callosum SSE route and bridge fan-out.""" + +from __future__ import annotations + +import json +import time +from collections.abc import Iterator + +import pytest + +import solstone.apps.observer.routes as routes_module +import solstone.convey.bridge as convey_bridge +from solstone.apps.observer.routes import OBSERVER_CALLOSUM_SSE_ROUTE +from solstone.apps.observer.utils import load_observer, save_observer + + +@pytest.fixture(autouse=True) +def clear_sse_subscribers() -> Iterator[None]: + with convey_bridge._SSE_LOCK: + convey_bridge._SSE_SUBSCRIBERS_BY_KEY.clear() + yield + with convey_bridge._SSE_LOCK: + convey_bridge._SSE_SUBSCRIBERS_BY_KEY.clear() + + +def _create_observer(env, name: str = "sse-test") -> tuple[str, str]: + resp = env.client.post( + "/app/observer/api/create", + json={"name": name}, + content_type="application/json", + ) + assert resp.status_code == 200 + data = resp.get_json() + return data["key"], data["key_prefix"] + + +def _route_for(key: str) -> str: + return OBSERVER_CALLOSUM_SSE_ROUTE.replace("", key) + + +def _next_chunk(response) -> str: + chunk = next(iter(response.response)) + if isinstance(chunk, bytes): + return chunk.decode("utf-8") + return str(chunk) + + +def _parse_sse_data(chunk: str) -> dict: + for line in chunk.splitlines(): + if line.startswith("data: "): + return json.loads(line[len("data: ") :]) + raise AssertionError(f"No data line found in chunk: {chunk!r}") + + +def _next_data(response) -> dict: + for chunk in response.response: + text = chunk.decode("utf-8") if isinstance(chunk, bytes) else str(chunk) + if "data: " in text: + return _parse_sse_data(text) + raise AssertionError("SSE stream ended before a data frame was received") + + +def test_callosum_sse_missing_key_returns_401(observer_env): + env = observer_env() + with env.app.test_request_context(_route_for("unused")): + response, status = routes_module.callosum_sse("") + assert status == 401 + assert response.get_json()["error"] == "Authorization required" + + +def test_callosum_sse_unknown_key_returns_401(observer_env): + env = observer_env() + resp = env.client.get(_route_for("unknown-key"), buffered=False) + assert resp.status_code == 401 + assert resp.get_json()["error"] == "Invalid key" + + +def test_callosum_sse_revoked_key_returns_403(observer_env): + env = observer_env() + key, key_prefix = _create_observer(env, "revoked-sse") + revoke = env.client.delete(f"/app/observer/api/{key_prefix}") + assert revoke.status_code == 200 + + resp = env.client.get(_route_for(key), buffered=False) + assert resp.status_code == 403 + assert resp.get_json()["error"] == "Observer revoked" + + +def test_callosum_sse_disabled_key_returns_403(observer_env): + env = observer_env() + key, _ = _create_observer(env, "disabled-sse") + observer = load_observer(key) + assert observer is not None + observer["enabled"] = False + assert save_observer(observer) + + resp = env.client.get(_route_for(key), buffered=False) + assert resp.status_code == 403 + assert resp.get_json()["error"] == "Observer disabled" + + +def test_callosum_sse_bearer_header_overrides_path_key(observer_env): + env = observer_env() + valid_key, _ = _create_observer(env, "valid-sse") + bogus_key = "bogus-key" + + resp = env.client.get( + _route_for(bogus_key), + headers={"Authorization": f"Bearer {valid_key}"}, + buffered=False, + ) + try: + assert resp.status_code == 200 + assert resp.content_type.startswith("text/event-stream") + finally: + resp.close() + + resp = env.client.get( + _route_for(valid_key), + headers={"Authorization": "Bearer invalid-key"}, + buffered=False, + ) + assert resp.status_code == 401 + assert resp.get_json()["error"] == "Invalid key" + + +def test_callosum_sse_success_content_type(observer_env): + env = observer_env() + key, _ = _create_observer(env) + + resp = env.client.get(_route_for(key), buffered=False) + try: + assert resp.status_code == 200 + assert resp.content_type.startswith("text/event-stream") + finally: + resp.close() + + +def test_callosum_sse_round_trip_payload(observer_env): + env = observer_env() + key, key_prefix = _create_observer(env) + resp = env.client.get(_route_for(key), buffered=False) + try: + assert resp.status_code == 200 + assert convey_bridge.subscription_count(key_prefix) == 1 + message = {"tract": "test", "event": "ping", "ts": 0, "extra": "value"} + convey_bridge._broadcast_callosum_event(message) + + parsed = _next_data(resp) + assert parsed == message + finally: + resp.close() + + +def test_callosum_sse_heartbeat(observer_env, monkeypatch): + env = observer_env() + key, _ = _create_observer(env) + monkeypatch.setattr(routes_module, "_SSE_HEARTBEAT_SECONDS", 0.01) + + resp = env.client.get(_route_for(key), buffered=False) + try: + assert resp.status_code == 200 + assert _next_chunk(resp) == ": heartbeat\n\n" + finally: + resp.close() + + +def test_sse_registry_lifecycle(): + handle = convey_bridge.register_sse_subscriber("aaaaaaaa") + assert convey_bridge.subscription_count("aaaaaaaa") == 1 + + convey_bridge.unregister_sse_subscriber(handle) + assert convey_bridge.subscription_count("aaaaaaaa") == 0 + + convey_bridge.unregister_sse_subscriber(handle) + assert convey_bridge.subscription_count("aaaaaaaa") == 0 + + +def test_slow_sse_subscriber_is_dropped_without_blocking_healthy_subscriber(): + slow = convey_bridge.register_sse_subscriber("aaaaaaaa") + healthy = convey_bridge.register_sse_subscriber("bbbbbbbb") + received: list[dict] = [] + + start = time.perf_counter() + for i in range(convey_bridge._SSE_QUEUE_MAXSIZE + 1): + convey_bridge._broadcast_callosum_event( + {"tract": "test", "event": "ping", "ts": i} + ) + received.append(json.loads(healthy.queue.get_nowait())) + elapsed = time.perf_counter() - start + + assert slow.dropped.is_set() + assert slow.drop_reason == "overflow" + assert "aaaaaaaa" not in convey_bridge._SSE_SUBSCRIBERS_BY_KEY + assert len(received) == convey_bridge._SSE_QUEUE_MAXSIZE + 1 + assert [message["ts"] for message in received] == list( + range(convey_bridge._SSE_QUEUE_MAXSIZE + 1) + ) + assert elapsed < 0.5 + + convey_bridge.unregister_sse_subscriber(healthy) diff --git a/solstone/apps/observer/tests/test_routes.py b/solstone/apps/observer/tests/test_routes.py index a5b3de576..41a3e927c 100644 --- a/solstone/apps/observer/tests/test_routes.py +++ b/solstone/apps/observer/tests/test_routes.py @@ -165,6 +165,9 @@ def test_api_list_empty(observer_env): "active_ms": 30000, "stale_ms": 120000, }, + "labels": { + "live": "live", + }, "observers": [], } diff --git a/solstone/convey/bridge.py b/solstone/convey/bridge.py index 6a0f862fc..d6b1e3df6 100644 --- a/solstone/convey/bridge.py +++ b/solstone/convey/bridge.py @@ -11,8 +11,10 @@ from __future__ import annotations import json import logging +import queue import threading import time +from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from flask_sock import Sock @@ -25,12 +27,27 @@ logger = logging.getLogger(__name__) _WATCH_LOCK = threading.Lock() _CALLOSUM_CONNECTION: Optional[CallosumConnection] = None _WEBSOCKET_CLIENTS: List[object] = [] +_SSE_QUEUE_MAXSIZE = 256 +_SSE_LOCK = threading.Lock() _STATE_CACHE: Dict[str, Any] = { "supervisor_status": None, "last_observe_ts": None, } +@dataclass(eq=False) +class _SseSubscriber: + key_prefix: str + queue: queue.Queue[str] = field( + default_factory=lambda: queue.Queue(maxsize=_SSE_QUEUE_MAXSIZE) + ) + dropped: threading.Event = field(default_factory=threading.Event) + drop_reason: str | None = None + + +_SSE_SUBSCRIBERS_BY_KEY: dict[str, set[_SseSubscriber]] = {} + + def _broadcast_to_websockets(event: dict) -> None: """Broadcast event to all connected WebSocket clients.""" msg = json.dumps(event) @@ -42,6 +59,58 @@ def _broadcast_to_websockets(event: dict) -> None: _WEBSOCKET_CLIENTS.remove(ws) +def register_sse_subscriber(key_prefix: str) -> _SseSubscriber: + """Register an SSE subscriber for a registered observer key prefix.""" + subscriber = _SseSubscriber(key_prefix=key_prefix) + with _SSE_LOCK: + _SSE_SUBSCRIBERS_BY_KEY.setdefault(key_prefix, set()).add(subscriber) + return subscriber + + +def unregister_sse_subscriber(handle: _SseSubscriber) -> None: + """Unregister an SSE subscriber handle. Safe to call more than once.""" + with _SSE_LOCK: + subscribers = _SSE_SUBSCRIBERS_BY_KEY.get(handle.key_prefix) + if not subscribers: + return + subscribers.discard(handle) + if not subscribers: + _SSE_SUBSCRIBERS_BY_KEY.pop(handle.key_prefix, None) + + +def subscription_count(key_prefix: str) -> int: + """Return the active SSE subscription count for an observer key prefix.""" + with _SSE_LOCK: + return len(_SSE_SUBSCRIBERS_BY_KEY.get(key_prefix, set())) + + +def _broadcast_to_sse_clients(message: dict) -> None: + """Broadcast a serialized Callosum event to all SSE subscribers.""" + with _SSE_LOCK: + subscribers = [ + subscriber + for subscribers_for_key in _SSE_SUBSCRIBERS_BY_KEY.values() + for subscriber in subscribers_for_key + ] + if not subscribers: + return + + serialized = json.dumps(message) + for subscriber in subscribers: + if subscriber.dropped.is_set(): + continue + try: + subscriber.queue.put_nowait(serialized) + except queue.Full: + subscriber.drop_reason = "overflow" + subscriber.dropped.set() + unregister_sse_subscriber(subscriber) + logger.info( + "Dropping slow observer callosum SSE subscriber key_prefix=%s", + subscriber.key_prefix, + ) + + def _broadcast_callosum_event(message: Dict[str, Any]) -> None: """Broadcast Callosum event to WebSocket clients and server-side handlers.""" # Update state cache @@ -58,6 +127,14 @@ def _broadcast_callosum_event(message: Dict[str, Any]) -> None: except Exception: # pragma: no cover - defensive against socket errors logger.exception("Failed to broadcast %s event", message.get("tract")) + # Broadcast to observer SSE clients + try: + _broadcast_to_sse_clients(message) + except Exception: # pragma: no cover - defensive against SSE errors + logger.exception( + "Failed to broadcast %s event to SSE clients", message.get("tract") + ) + # Dispatch to server-side app event handlers try: from solstone.apps.events import dispatch diff --git a/solstone/convey/copy.py b/solstone/convey/copy.py index 975d7fd40..e14cbaac0 100644 --- a/solstone/convey/copy.py +++ b/solstone/convey/copy.py @@ -22,6 +22,7 @@ SETTINGS_NETWORK_BUTTON_DISABLE = "restrict to localhost only" SETTINGS_NETWORK_NEEDS_PASSWORD = "set a password below first." SETTINGS_NETWORK_RESTARTING = "restarting convey…" SETTINGS_PASSWORD_HINT = "protects the web interface when network access is on. not required for localhost-only mode." +OBSERVER_CALLOSUM_LIVE_LABEL = "live" PAIRING_LOCALHOST_BANNER_TITLE = "convey is in localhost-only mode" PAIRING_LOCALHOST_BANNER_BODY_1 = ( "paired devices won't be able to connect until network access is enabled." @@ -34,6 +35,7 @@ PAIRING_NO_LAN_BANNER_BODY = "the QR code below uses localhost, which paired dev __all__ = [ "INIT_PASSWORD_HINT", + "OBSERVER_CALLOSUM_LIVE_LABEL", "PAIRING_LOCALHOST_BANNER_ACTION", "PAIRING_LOCALHOST_BANNER_BODY_1", "PAIRING_LOCALHOST_BANNER_BODY_2", diff --git a/tests/baselines/api/observer/list.json b/tests/baselines/api/observer/list.json index a99bc8151..a6ea4a6d8 100644 --- a/tests/baselines/api/observer/list.json +++ b/tests/baselines/api/observer/list.json @@ -1,4 +1,7 @@ { + "labels": { + "live": "live" + }, "observers": [], "thresholds": { "active_ms": 30000,