11 - API (feat): Use Litestart Channels/Subscriptions

api/docs/adr/008-local-mode-composition.md

@@ -0,0 +1,44 @@
+---
+date: 2026-07-08
+---
+
+# ADR-008 - Local mode via composition, not conditionals
+
+## Context and Problem Statement
+
+The API supports a local development mode (`--path <damnit-dir>` / `DW_API_DAMNIT_PATH`): no OAuth, no MyMdC, a synthetic development user, and a single fixed proposal directory instead of GPFS discovery.
+
+A mode can be implemented in two ways. One branches on `settings.is_local` at every affected call site. The other selects implementations once, at composition time. Scattered branching means every new code path must remember both modes, local behaviour is defined piecemeal, and the two modes can drift apart silently. Selecting implementations once means local behaviour is defined in one reviewable place, and both modes share the same interfaces by construction.
+
+## Considered Options
+
+- Branch on `settings.is_local` at each affected call site.
+- Select implementations once, in the composition root.
+
+## Decision Outcome
+
+Chosen option: "select implementations once, in the composition root", because it keeps local behaviour in one place and makes the two modes share interfaces by construction.
+
+- `settings.is_local` is read only in the composition root (`main.py` / `state.py`) and in the `Settings` property that defines it ([ADR-003](003-injected-settings.md)).
+- It selects implementations: a path locator ([ADR-004](004-proposal-path-locator.md)), a MyMdC client, an auth controller, and a store backend.
+- Feature code depends on the selected collaborator through injection, never on the mode.
+
+### Consequences
+
+- Good: adding a code path cannot silently break local mode, because there is no second branch to forget.
+- Good: local and production can differ only in implementation, never in interface shape; each collaborator is independently testable and tests get the local composition for free.
+- Bad: a few more small interfaces exist (user provider, authorisation policy, proposals provider) rather than inline conditionals.
+
+## Details
+
+The composition maps each mode-dependent concern to a production and a local implementation. Some concerns are still branched inline rather than composed.
+
+| Concern | Production | Local |
+|---|---|---|
+| Proposal path resolution | GPFS path locator ([ADR-004](004-proposal-path-locator.md)) | fixed path locator for the given directory |
+| MyMdC | HTTP client | mock client synthesising the single local proposal |
+| Auth routes | OAuth controller | no-auth controller |
+| Current user | session-derived user info | synthetic development user |
+| Authorisation | proposal-membership policy | allow-all policy |
+| Proposal metadata | MyMdC fetch plus app-DB cache | synthetic metadata for the local directory |
+| Stores | file-backed | in-memory |

api/docs/adr/009-channels-subscriptions.md

@@ -0,0 +1,40 @@
+---
+date: 2026-07-08
+---
+
+# ADR-009 - Subscriptions: channel consumers, composition-selected publisher
+
+## Context and Problem Statement
+
+The frontend's live table updates use a GraphQL subscription (`latest_data`) over `graphql-transport-ws`. DAMNIT proposal databases are plain SQLite files written by an external process. There is no change feed to subscribe to today, so the server must produce one by polling.
+
+That will not stay true. Candidate event sources exist on the horizon: the facility Kafka bus, or Postgres `LISTEN/NOTIFY` should the application database move to Postgres. The subscription design must not weld resolvers to any one change-detection mechanism.
+
+Two forces shape the design. Naive per-client polling multiplies identical reads: N subscribers to one proposal would issue N queries per tick against a GPFS-hosted SQLite file. And any process-local coordination state is a deployment constraint that must be a recorded decision with a retirement path, not an accident.
+
+## Considered Options
+
+- Per-client polling loops inside the resolver.
+- Subscribers consume a channel; a composition-selected publisher produces events.
+
+## Decision Outcome
+
+Chosen option: "subscribers consume a channel; a composition-selected publisher produces events", because it keeps subscribers stable while the event source changes and coalesces reads structurally.
+
+- Subscribers consume a per-proposal channel on Litestar's `ChannelsPlugin` and fan events to their client. Resolvers hold no polling loops, no cursors, and no knowledge of how an event was produced.
+- Exactly one publisher per deployment is chosen in the composition root, the same composition-root selection as [ADR-008](008-local-mode-composition.md). Today it is a SQLite poller; a Postgres `LISTEN/NOTIFY` consumer or a Kafka bridge can replace it behind the same contract without touching subscribers.
+- The poller reads each watched proposal once per tick through the repository ([ADR-005](005-repository-pattern.md)) and publishes new rows, so one poll serves every subscriber. Its per-proposal high-water mark is publisher-internal state, deleted with the poller when a push publisher lands.
+- Each client passes its own `timestamp`; the resolver filters the shared event per client, so late joiners receive neither stale rows nor duplicates.
+- Persistent publisher failures publish an error event and terminate affected subscriptions with a typed error ([ADR-001](001-error-classes.md)), so clients resubscribe deliberately rather than silently receiving nothing.
+
+### Consequences
+
+- Good: subscriber code is stable across the polling to push migration; the Postgres-versus-Kafka choice is deferred without accruing rewrite cost.
+- Good: under polling, load scales with the number of watched proposals, not connected clients; under a push publisher the per-tick GPFS reads disappear.
+- Bad: the process-local channels backend forces single-worker deployment for now, a constraint that must stay loud until retired.
+
+## Details
+
+The single-worker retirement plan, in order: move sessions and tokens to a store-backed server-side session; move the channels backend to a shared backend; make the publisher push-based (deleting the poller and its cursors) or, if polling must persist, run it once per deployment with store-backed cursors. After those steps nothing process-local remains that is not a cache (the per-repository caches of [ADR-005](005-repository-pattern.md)), and the startup guard is removed.
+
+The event contract is that every new run is published to its proposal's channel exactly once, in order. Push publishers satisfy it natively. The polling publisher satisfies it with the high-water mark, initialised from the proposal's current max timestamp and advanced as new rows are seen.

api/docs/architecture.md

@@ -58,7 +58,7 @@ The key rules are:
 3. **Composition root reads settings:** everything else receives configuration as parameters (see [ADR-003](adr/003-injected-settings.md)).
 4. **Authorisation applied at the edge:** routes and resolvers use dependencies and permission classes.
   - This means that services should not apply authorisation rules themselves.
-5. **No `if settings.is_local:` outside the composition root:** Local mode is selected by composition, not conditionals throughout the codebase.
+5. **No `if settings.is_local:` outside the composition root:** Local mode is selected by composition, not conditionals throughout the codebase (see [ADR-008](adr/008-local-mode-composition.md)).
 
 Note that these are currently only enforced by convention/review. Import linter/archetecture check tool is planned to be added.
 

api/pyproject.toml

@@ -79,8 +79,8 @@ dev = [
 default-groups = []
 
 [tool.pytest.ini_options]
-# Session-scoped event loop so alru_cache instances bind once and never
-# see a loop change between tests (no AlruCacheLoopResetWarning).
+# Session-scoped event loop so async fixtures (e.g. the channels plugin) and
+# tests share one loop across the suite, with no loop change between tests.
 asyncio_default_fixture_loop_scope = "session"
 asyncio_default_test_loop_scope = "session"
 

api/src/damnit_api/graphql/dependencies.py

@@ -1,15 +1,23 @@
 """Litestar dependency helpers for GraphQL subscription state."""
 
+from litestar.channels import ChannelsPlugin
 from litestar.datastructures import State
 
-from .subscriptions import SubscriptionCursors
+from .publisher import RunUpdatePublisher
 
 
-def get_subscription_cursors(state: State) -> SubscriptionCursors:
-    """Provide the subscription cursors from the application state."""
-    return state.app_state.subscription_cursors  # type: ignore[attr-defined]
+def get_channels(state: State) -> ChannelsPlugin:
+    """Provide the channels plugin from the application state."""
+    return state.app_state.channels  # type: ignore[attr-defined]
 
 
-# Plain type alias; Litestar injects by the parameter name `subscription_cursors`.
-SubscriptionCursorsDep = SubscriptionCursors
-"""Type alias for the subscription cursors dependency."""
+def get_run_update_publisher(state: State) -> RunUpdatePublisher:
+    """Provide the run-update publisher from the application state."""
+    return state.app_state.run_update_publisher  # type: ignore[attr-defined]
+
+
+# Plain type aliases; Litestar injects by the parameter name.
+ChannelsDep = ChannelsPlugin
+"""Type alias for the channels-plugin dependency."""
+RunUpdatePublisherDep = RunUpdatePublisher
+"""Type alias for the run-update-publisher dependency."""

api/src/damnit_api/graphql/publisher.py

@@ -0,0 +1,165 @@
+"""Run-update publishers: composition-selected producers of channel events.
+
+Subscription resolvers consume per-proposal channels (ADR-009); exactly one
+publisher per deployment produces the events. The SQLite poller below is the
+default; a Postgres LISTEN/NOTIFY consumer or a Kafka bridge slot in behind
+the same contract in the composition root (ADR-008) without touching
+subscribers.
+"""
+
+import asyncio
+import dataclasses
+from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+
+from .. import get_logger
+from ..runs.types import DamnitRun
+from ..shared.models import ProposalNumber
+
+if TYPE_CHECKING:
+    from litestar.channels import ChannelsPlugin
+
+    from ..runs.repository import DamnitRepositoryRegistry
+
+logger = get_logger()
+
+POLLING_INTERVAL = 1.0  # seconds
+MAX_CONSECUTIVE_FAILURES = 5
+
+
+def proposal_channel(proposal_number: ProposalNumber) -> str:
+    """Name of the run-updates channel for a proposal."""
+    return f"proposal:{proposal_number}"
+
+
+# `runtime_checkable` so Litestar's msgspec-based signature validation
+# (`isinstance` on injected Protocols) doesn't raise on every request.
+@runtime_checkable
+class RunUpdatePublisher(Protocol):
+    """Produce run-update events on per-proposal channels.
+
+    The event payload is the snapshot dict consumed by `latest_data`
+    (`runs`/`run_timestamps`/`max_timestamp`/`metadata`), or
+    `{"error": {...}}` when the publisher gives up on a proposal.
+    """
+
+    def watch(self, proposal_number: ProposalNumber) -> None: ...
+    async def aclose(self) -> None: ...
+
+
+class SqlitePollingRunUpdatePublisher:
+    """Polls DAMNIT databases and publishes new-run snapshots per proposal.
+
+    One poll task per watched proposal regardless of subscriber count —
+    coalescing is structural (one publisher, N channel subscribers). The
+    per-proposal high-water mark keeps poll cost proportional to new data;
+    it is an implementation detail of *this* publisher and is deleted with
+    it under push-based backends (ADR-009).
+    """
+
+    def __init__(
+        self,
+        channels: "ChannelsPlugin",
+        repositories: "DamnitRepositoryRegistry",
+        *,
+        interval: float = POLLING_INTERVAL,
+        max_consecutive_failures: int = MAX_CONSECUTIVE_FAILURES,
+    ) -> None:
+        self._channels = channels
+        self._repositories = repositories
+        self._interval = interval
+        self._max_consecutive_failures = max_consecutive_failures
+        self._cursors: dict[ProposalNumber, float] = {}
+        self._tasks: dict[ProposalNumber, asyncio.Task] = {}
+
+    def watch(self, proposal_number: ProposalNumber) -> None:
+        """Ensure a poll task is running for the proposal."""
+        task = self._tasks.get(proposal_number)
+        if task is None or task.done():
+            self._tasks[proposal_number] = asyncio.create_task(
+                self._poll_loop(proposal_number)
+            )
+
+    async def aclose(self) -> None:
+        for task in self._tasks.values():
+            task.cancel()
+        await asyncio.gather(*self._tasks.values(), return_exceptions=True)
+        self._tasks.clear()
+
+    async def _poll_loop(self, proposal_number: ProposalNumber) -> None:
+        failures = 0
+        while True:
+            await asyncio.sleep(self._interval)
+            try:
+                snapshot = await self._poll(proposal_number)
+            except Exception:
+                failures += 1
+                logger.exception(
+                    "Run-update poll failed",
+                    proposal=proposal_number,
+                    consecutive_failures=failures,
+                )
+                if failures >= self._max_consecutive_failures:
+                    # Persistent failure: tell subscribers to terminate with
+                    # a typed error (ADR-001) instead of silently going quiet.
+                    self._channels.publish(
+                        {
+                            "error": {
+                                "message": (
+                                    "Run updates unavailable after "
+                                    f"{failures} consecutive poll failures"
+                                ),
+                            }
+                        },
+                        proposal_channel(proposal_number),
+                    )
+                    self._tasks.pop(proposal_number, None)
+                    return
+                continue
+
+            failures = 0
+            if snapshot is not None:
+                self._channels.publish(snapshot, proposal_channel(proposal_number))
+
+    async def _poll(self, proposal_number: ProposalNumber) -> dict[str, Any] | None:
+        """One poll tick: return a snapshot of rows newer than the cursor."""
+        repo = self._repositories.get(proposal_number)
+
+        # Initialize cursor from the current max timestamp on first visit
+        if proposal_number not in self._cursors:
+            metadata = await repo.get_metadata()
+            self._cursors[proposal_number] = metadata.timestamp
+
+        start_at = self._cursors[proposal_number]
+        records = await repo.get_latest_runs(start_at=start_at)
+        if not records:
+            return None
+
+        runs: dict[int, Any] = {}
+        run_timestamps: dict[int, float] = {}
+        for record in records:
+            runs[record.run] = DamnitRun.resolve_record(record)
+            if record.variables:
+                run_timestamps[record.run] = max(
+                    vv.timestamp for vv in record.variables.values()
+                )
+
+        if not runs:
+            return None
+
+        latest_ts = max(run_timestamps.values(), default=start_at)
+
+        repo.invalidate_metadata_cache()
+        metadata = await repo.get_metadata()
+        self._cursors[proposal_number] = latest_ts
+        meta_dict = dataclasses.asdict(metadata)
+
+        return {
+            "runs": runs,
+            "run_timestamps": run_timestamps,
+            "max_timestamp": latest_ts,
+            "metadata": {
+                "runs": sorted(set(metadata.runs) | set(runs.keys())),
+                "variables": meta_dict["variables"],
+                "timestamp": latest_ts * 1000,  # ms for JS
+            },
+        }