Qt-removal R0: foundations - Python backend, single SPA, native pywebview shell, burn-down gate

backend/__init__.py

@@ -0,0 +1,14 @@
+"""Graphlink backend (Qt-removal plan R0, doc/QT_REMOVAL_PLAN.md).
+
+The Python half of the target architecture: FastAPI + a WebSocket event bus
+carrying full-state snapshots and intents between the domain services and the
+single React SPA. This package is the direct successor of the QWebChannel
+IslandBridge layer and deliberately preserves its wire semantics (versioned
+full-state envelopes, named intents) so the existing island payload schemas
+and generated TS types retarget without redesign.
+
+Zero Qt imports are permitted anywhere under this package - enforced by
+graphlink_app/tests/test_no_qt_anywhere.py.
+"""
+
+BACKEND_VERSION = "0.1.0"

backend/app.py

@@ -0,0 +1,140 @@
+"""FastAPI application: HTTP surface + the /ws WebSocket endpoint
+(Qt-removal plan R0).
+
+Serves three things:
+- /api/health - liveness + version (the desktop shell polls this at startup)
+- /ws?session=<id> - the event-bus WebSocket (state snapshots out, intents in)
+- / - the built SPA (static files), when a build directory exists
+
+Client -> server message kinds over /ws:
+  {"kind": "subscribe", "topics": ["system", ...]}      -> current snapshots
+  {"kind": "intent", "topic": t, "intent": name,
+   "args": [...], "id": optional}                        -> optional result
+Server -> client:
+  {"kind": "state", "topic": t, "payload": {...envelope...}}
+  {"kind": "result", "id": ..., "value": ...}            (only when id sent)
+  {"kind": "error", "id": ..., "error": "..."}           (bad topic/intent)
+
+R0 registers only the `system` topic (backend identity) and its `ping`
+intent - the acceptance round-trip. Real domain topics arrive per-phase
+(R1 canvas, R2 chrome, ...), each registering here exactly like system does.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from pathlib import Path
+
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.responses import FileResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+
+from backend import BACKEND_VERSION
+from backend.events import EventBus, SessionBus, UnknownIntentError, UnknownTopicError
+
+logger = logging.getLogger(__name__)
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SPA_DIST_DIR = REPO_ROOT / "web_ui" / "dist" / "app"
+
+
+def _configure_session(bus: SessionBus) -> None:
+    """Give every session the R0 topic surface. Later phases extend this
+    with canvas/chrome/node topics - one registrar, one place to read the
+    whole API surface."""
+
+    bus.register_topic(
+        "system",
+        lambda: {"app": "graphlink", "backendVersion": BACKEND_VERSION, "sessionId": bus.session_id},
+    )
+
+    def ping(*args):
+        # The R0 acceptance round-trip: echo + a server-side timestamp so the
+        # UI can prove the reply crossed the process boundary.
+        return {"echo": list(args), "serverTime": time.time()}
+
+    bus.register_intent("system", "ping", ping)
+
+
+def create_app(spa_dir: Path | None = None) -> FastAPI:
+    app = FastAPI(title="Graphlink backend", version=BACKEND_VERSION)
+    bus = EventBus(configure_session=_configure_session)
+    app.state.bus = bus
+
+    @app.get("/api/health")
+    async def health() -> JSONResponse:
+        return JSONResponse({"status": "ok", "app": "graphlink", "version": BACKEND_VERSION})
+
+    @app.websocket("/ws")
+    async def ws_endpoint(websocket: WebSocket) -> None:
+        session_id = websocket.query_params.get("session", "default")
+        session = bus.session(session_id)
+        await websocket.accept()
+        session.attach(websocket)
+        try:
+            while True:
+                message = await websocket.receive_json()
+                await _handle_message(session, websocket, message)
+        except WebSocketDisconnect:
+            pass
+        finally:
+            session.detach(websocket)
+
+    resolved_spa = SPA_DIST_DIR if spa_dir is None else spa_dir
+    if resolved_spa.is_dir():
+        # html=True serves index.html at / ; the explicit fallback below keeps
+        # deep links (client-side routes) working instead of 404ing.
+        app.mount("/assets", StaticFiles(directory=resolved_spa / "assets"), name="assets")
+
+        @app.get("/{full_path:path}")
+        async def spa(full_path: str) -> FileResponse:
+            candidate = (resolved_spa / full_path).resolve()
+            if full_path and candidate.is_file() and candidate.is_relative_to(resolved_spa.resolve()):
+                return FileResponse(candidate)
+            return FileResponse(resolved_spa / "index.html")
+    else:
+        logger.warning("SPA build not found at %s - only /api and /ws are served", resolved_spa)
+
+    return app
+
+
+async def _handle_message(session: SessionBus, websocket: WebSocket, message: dict) -> None:
+    kind = message.get("kind")
+    msg_id = message.get("id")
+
+    if kind == "subscribe":
+        topics = message.get("topics") or session.topic_names()
+        for topic in topics:
+            try:
+                await session.send_snapshot(topic, websocket)
+            except UnknownTopicError:
+                await websocket.send_json(
+                    {"kind": "error", "id": msg_id, "error": f"unknown topic: {topic}"}
+                )
+        return
+
+    if kind == "intent":
+        topic = message.get("topic", "")
+        intent = message.get("intent", "")
+        args = message.get("args") or []
+        try:
+            result = await session.dispatch_intent(topic, intent, args)
+        except (UnknownTopicError, UnknownIntentError) as exc:
+            await websocket.send_json({"kind": "error", "id": msg_id, "error": str(exc)})
+            return
+        except Exception:
+            # Handler bugs surface as errors to the caller, never as a dropped
+            # socket - and always land in the log.
+            logger.exception("intent %s/%s failed", topic, intent)
+            await websocket.send_json(
+                {"kind": "error", "id": msg_id, "error": f"intent failed: {topic}/{intent}"}
+            )
+            return
+        if msg_id is not None:
+            await websocket.send_json({"kind": "result", "id": msg_id, "value": result})
+        return
+
+    await websocket.send_json(
+        {"kind": "error", "id": msg_id, "error": f"unknown message kind: {kind!r}"}
+    )

backend/events.py

@@ -0,0 +1,171 @@
+"""WebSocket event bus: session-scoped topics carrying versioned full-state
+snapshots plus named intents (Qt-removal plan R0).
+
+Wire semantics are inherited from the IslandBridge/QWebChannel layer this
+replaces, because they were already the right shape and the frontend's
+generated validators depend on them:
+
+- Server -> client: full-state snapshots only, never diffs. Every snapshot
+  carries schemaVersion / minCompatibleSchemaVersion / revision, stamped
+  here exactly as IslandBridge.publish() stamped them (a reader may accept a
+  NEWER payload than it understands - additive-only guarantee - but must
+  refuse one older than its stated minimum).
+- Client -> server: named intents ("setGridSize", "ready", ...) addressed to
+  a topic, with positional JSON args - the successor of @Slot methods.
+
+Scoping: one SessionBus per session id. Topics, revisions, and connections
+are session-local, so two windows on different sessions never see each
+other's state. The bus is transport-agnostic: connections are anything with
+an async send_json(dict) - real WebSockets in app.py, plain recorders in
+tests.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import logging
+from typing import Any, Awaitable, Callable, Protocol
+
+logger = logging.getLogger(__name__)
+
+StateBuilder = Callable[[], dict[str, Any]]
+IntentHandler = Callable[..., Any | Awaitable[Any]]
+
+
+class Connection(Protocol):
+    async def send_json(self, data: dict[str, Any]) -> None: ...
+
+
+class UnknownTopicError(KeyError):
+    """An intent or publish referenced a topic nothing registered."""
+
+
+class UnknownIntentError(KeyError):
+    """A client sent an intent name the topic does not expose."""
+
+
+class _Topic:
+    __slots__ = ("name", "builder", "schema_version", "min_compatible", "revision")
+
+    def __init__(self, name: str, builder: StateBuilder, schema_version: int, min_compatible: int):
+        self.name = name
+        self.builder = builder
+        self.schema_version = schema_version
+        self.min_compatible = min_compatible
+        self.revision = 0
+
+    def snapshot(self) -> dict[str, Any]:
+        self.revision += 1
+        payload = dict(self.builder())
+        payload["schemaVersion"] = self.schema_version
+        payload["minCompatibleSchemaVersion"] = self.min_compatible
+        payload["revision"] = self.revision
+        return payload
+
+
+class SessionBus:
+    """Topics + connections + intent handlers for ONE session."""
+
+    def __init__(self, session_id: str):
+        self.session_id = session_id
+        self._topics: dict[str, _Topic] = {}
+        self._intents: dict[tuple[str, str], IntentHandler] = {}
+        self._connections: set[Connection] = set()
+
+    # -- registration ------------------------------------------------------
+
+    def register_topic(
+        self,
+        name: str,
+        builder: StateBuilder,
+        *,
+        schema_version: int = 1,
+        min_compatible: int = 1,
+    ) -> None:
+        assert name not in self._topics, f"topic {name!r} registered twice"
+        self._topics[name] = _Topic(name, builder, schema_version, min_compatible)
+
+    def register_intent(self, topic: str, intent: str, handler: IntentHandler) -> None:
+        key = (topic, intent)
+        assert key not in self._intents, f"intent {topic}/{intent} registered twice"
+        self._intents[key] = handler
+
+    # -- connections -------------------------------------------------------
+
+    def attach(self, conn: Connection) -> None:
+        self._connections.add(conn)
+
+    def detach(self, conn: Connection) -> None:
+        self._connections.discard(conn)
+
+    @property
+    def connection_count(self) -> int:
+        return len(self._connections)
+
+    # -- state flow --------------------------------------------------------
+
+    async def publish(self, topic: str) -> dict[str, Any]:
+        """Build a fresh snapshot for `topic` and broadcast it to every
+        attached connection. Returns the snapshot (tests + send-current-state
+        on subscribe both want it)."""
+        t = self._topics.get(topic)
+        if t is None:
+            raise UnknownTopicError(topic)
+        snapshot = t.snapshot()
+        message = {"kind": "state", "topic": topic, "payload": snapshot}
+        # Snapshot the set: a failed send detaches the connection mid-loop.
+        for conn in list(self._connections):
+            try:
+                await conn.send_json(message)
+            except Exception:
+                # A dead socket must never poison the broadcast for the rest.
+                logger.warning("dropping dead connection on session %s", self.session_id)
+                self.detach(conn)
+        return snapshot
+
+    async def send_snapshot(self, topic: str, conn: Connection) -> None:
+        """Send the current state of one topic to one connection (the
+        subscribe handshake - the successor of loadFinished -> publish())."""
+        t = self._topics.get(topic)
+        if t is None:
+            raise UnknownTopicError(topic)
+        await conn.send_json({"kind": "state", "topic": topic, "payload": t.snapshot()})
+
+    def topic_names(self) -> list[str]:
+        return sorted(self._topics)
+
+    async def dispatch_intent(self, topic: str, intent: str, args: list[Any]) -> Any:
+        """Run a registered intent handler. Sync and async handlers are both
+        supported; sync handlers run in a thread so a slow one cannot stall
+        the event loop (QThread's replacement in miniature)."""
+        handler = self._intents.get((topic, intent))
+        if handler is None:
+            if topic not in self._topics and not any(t == topic for t, _ in self._intents):
+                raise UnknownTopicError(topic)
+            raise UnknownIntentError(f"{topic}/{intent}")
+        if inspect.iscoroutinefunction(handler):
+            return await handler(*args)
+        return await asyncio.to_thread(handler, *args)
+
+
+class EventBus:
+    """All sessions. Session buses are created on first use and configured by
+    the app's registrar so every session exposes the same topic/intent
+    surface over its own state."""
+
+    def __init__(self, configure_session: Callable[[SessionBus], None] | None = None):
+        self._sessions: dict[str, SessionBus] = {}
+        self._configure_session = configure_session
+
+    def session(self, session_id: str = "default") -> SessionBus:
+        bus = self._sessions.get(session_id)
+        if bus is None:
+            bus = SessionBus(session_id)
+            if self._configure_session is not None:
+                self._configure_session(bus)
+            self._sessions[session_id] = bus
+        return bus
+
+    def session_ids(self) -> list[str]:
+        return sorted(self._sessions)