feat(sp36): extract remittances + activity routers (NDJSON live-tail batch)

Tasks 9 + 10 combined per user direction (one commit for the streaming-NDJSON batch).

api_routers/remittances.py (new, 169 lines):
  - GET /api/remittances                  (paginated list + NDJSON variant)
  - GET /api/remittances/summary          (server-aggregated KPI tiles)
  - GET /api/remittances/stream           (NDJSON live-tail on remittance_written)
  - GET /api/remittances/{remittance_id}  (single remittance + labeled CAS)

api_routers/activity.py (new, 102 lines):
  - GET /api/activity         (paginated event list + NDJSON variant)
  - GET /api/activity/stream  (NDJSON live-tail on activity_recorded)

Both routers use router-level matrix_gate (single source of auth).

api.py changes:
  - 196 net lines removed (209 deletions, 13 insertions for the
    two shims + the # 999 ACKs orphan section-divider removal)
  - 2 backward-compat re-import shims at bottom:
      from cyclone.api_routers.remittances import remittances_stream
      from cyclone.api_routers.activity import activity_stream
    rationale: tests/test_api_stream_live.py imports these by name
    from cyclone.api; per SP-N invariant we don't rewrite tests for
    a structural refactor. (Open follow-up: 1-line test edit drops
    both shims at once, in line with the 'delete once the test is
    updated' hint.)
  - Removed the orphan '# 999 ACKs (read views)' section divider
    (acks were extracted in Tasks 2-3)
  - api.py: 3041 -> 2844 LOC (-197)

api_routers/__init__.py: registry extended (alphabetical) with
remittances + activity. 12 routers in registry.

Pytest: bit-identical to baseline — 21 failed, 1246 passed,
10 skipped, 6 errors. (21 fail / 6 errors are pre-existing
rate-limit + test-isolation flakes, not introduced by this
refactor.)

Live-tested via curl on the running container:
  GET /api/remittances         -> 401 (matrix_gate fires)
  GET /api/remittances/summary -> 401
  GET /api/remittances/stream  -> 401
  GET /api/remittances/<id>    -> 401
  POST /api/remittances        -> 405 (method-not-allowed)
  GET /api/activity            -> 401
  GET /api/activity/stream     -> 401
  POST /api/activity           -> 405

pr-reviewer: PASS

backend/src/cyclone/api.py

@@ -2445,134 +2445,6 @@ def _svc_to_dict(svc) -> dict:
     }
 
 
-@app.get("/api/remittances", dependencies=[Depends(matrix_gate)])
-def list_remittances(
-    request: Request,
-    batch_id: str | None = Query(None),
-    payer: str | None = Query(None),
-    claim_id: str | None = Query(None),
-    date_from: str | None = Query(None),
-    date_to: str | None = Query(None),
-    sort: str | None = Query(None),
-    order: str = Query("desc"),
-    limit: int = Query(100, ge=1, le=1000),
-    offset: int = Query(0, ge=0),
-) -> Any:
-    common = dict(
-        batch_id=batch_id,
-        payer=payer,
-        claim_id=claim_id,
-        date_from=date_from,
-        date_to=date_to,
-    )
-    items = list(store.iter_remittances(
-        sort=sort, order=order, limit=limit, offset=offset, **common,
-    ))
-    # SP27 Task 13b: count the full population, not a 100-row sample.
-    # See the matching note in list_claims — same silent-failure pattern.
-    total = store.count_remittances(**common)
-    returned = len(items)
-    has_more = total > offset + returned
-    if _wants_ndjson(request):
-        return StreamingResponse(
-            _ndjson_stream_list(items, total, returned, has_more),
-            media_type="application/x-ndjson",
-        )
-    return {
-        "items": items,
-        "total": total,
-        "returned": returned,
-        "has_more": has_more,
-    }
-
-
-@app.get("/api/remittances/summary", dependencies=[Depends(matrix_gate)])
-def remittances_summary(
-    batch_id: str | None = Query(None),
-    payer: str | None = Query(None),
-    claim_id: str | None = Query(None),
-    date_from: str | None = Query(None),
-    date_to: str | None = Query(None),
-) -> dict:
-    """Server-aggregated KPI tiles for the Remittances page.
-
-    Returns ``{count, total_paid, total_adjustments}`` over the
-    full filtered remittance population — NOT a page-limited
-    sample. The Remittances page consumes this for its "Total paid"
-    and "Adjustments" tiles so they can't silently understate the
-    true DB population the way a page-local ``items.reduce(...)``
-    would. Mirrors the silent-incompleteness fix that
-    ``/api/dashboard/kpis`` (commit ``59c3275``) and
-    ``/api/remittances`` (commit ``d81b6ed``) made for their tiles.
-
-    Same filter parameters as ``/api/remittances``. Always returns
-    a populated dict (``{"count": 0, "total_paid": 0,
-    "total_adjustments": 0}`` when no rows match) so the frontend
-    can render the tiles directly without a loading-vs-empty
-    branch.
-    """
-    return store.summarize_remittances(
-        batch_id=batch_id, payer=payer, claim_id=claim_id,
-        date_from=date_from, date_to=date_to,
-    )
-
-
-@app.get("/api/remittances/stream", dependencies=[Depends(matrix_gate)])
-async def remittances_stream(
-    request: Request,
-    payer: str | None = Query(None),
-    claim_id: str | None = Query(None),
-    date_from: str | None = Query(None),
-    date_to: str | None = Query(None),
-    sort: str | None = Query(None),
-    order: str = Query("desc"),
-    limit: int = Query(100, ge=1, le=1000),
-) -> StreamingResponse:
-    """Stream Remittances as NDJSON: snapshot first, then live events.
-
-    Subscribes to ``remittance_written``. Default sort is
-    ``-received_date`` (newest-first), matching the list endpoint's
-    most common sort.
-
-    NOTE: registered before ``/api/remittances/{remittance_id}`` so
-    the literal ``stream`` path segment doesn't get matched as a
-    remittance id.
-    """
-    bus: EventBus = request.app.state.event_bus
-
-    async def gen() -> AsyncIterator[bytes]:
-        rows = store.iter_remittances(
-            payer=payer, claim_id=claim_id,
-            date_from=date_from, date_to=date_to,
-            sort=sort or "-received_date", order=order, limit=limit,
-        )
-        for row in rows:
-            yield _ndjson_line({"type": "item", "data": row})
-        yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
-
-        async for chunk in _tail_events(request, bus, ["remittance_written"]):
-            yield chunk
-
-    return StreamingResponse(gen(), media_type="application/x-ndjson")
-
-
-@app.get("/api/remittances/{remittance_id}", dependencies=[Depends(matrix_gate)])
-def get_remittance(remittance_id: str) -> dict:
-    """Return one remittance with its labeled CAS ``adjustments`` array.
-
-    Path param is ``remittance_id`` (not ``id``) to avoid shadowing
-    FastAPI's internal ``id`` name and to keep OpenAPI docs self-
-    describing. Returns 404 when the remittance is missing — never 500.
-    """
-    body = store.get_remittance(remittance_id)
-    if body is None:
-        raise HTTPException(
-            status_code=404,
-            detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"},
-        )
-    return body
-
-
 @app.get("/api/providers", dependencies=[Depends(matrix_gate)])
 def list_providers(
     request: Request,
@@ -2603,74 +2475,6 @@ def list_providers(
     }
 
 
-@app.get("/api/activity", dependencies=[Depends(matrix_gate)])
-def list_activity(
-    request: Request,
-    kind: str | None = Query(None),
-    since: str | None = Query(None),
-    limit: int = Query(200, ge=1, le=5000),
-) -> Any:
-    events = store.recent_activity(limit=limit)
-    if kind is not None:
-        events = [e for e in events if e["kind"] == kind]
-    if since is not None:
-        events = [e for e in events if e["timestamp"] >= since]
-    total = len(events)
-    has_more = False
-    if _wants_ndjson(request):
-        return StreamingResponse(
-            _ndjson_stream_list(events, total, total, has_more),
-            media_type="application/x-ndjson",
-        )
-    return {
-        "items": events,
-        "total": total,
-        "returned": total,
-        "has_more": has_more,
-    }
-
-
-@app.get("/api/activity/stream", dependencies=[Depends(matrix_gate)])
-async def activity_stream(
-    request: Request,
-    kind: str | None = Query(None),
-    since: str | None = Query(None),
-    limit: int = Query(50, ge=1, le=5000),
-) -> StreamingResponse:
-    """Stream Activity events as NDJSON: snapshot first, then live events.
-
-    Subscribes to ``activity_recorded``. Default ``limit`` is 50
-    (smaller than the list endpoint's 200) because activity is
-    high-volume — callers usually want the most recent handful, not a
-    full replay.
-    """
-    bus: EventBus = request.app.state.event_bus
-
-    async def gen() -> AsyncIterator[bytes]:
-        # Snapshot reuses the same in-memory filter as ``list_activity``
-        # so the two endpoints are interchangeable for the snapshot
-        # half.
-        events = store.recent_activity(limit=limit)
-        if kind is not None:
-            events = [e for e in events if e["kind"] == kind]
-        if since is not None:
-            events = [e for e in events if e["timestamp"] >= since]
-        for ev in events:
-            yield _ndjson_line({"type": "item", "data": ev})
-        yield _ndjson_line({
-            "type": "snapshot_end", "data": {"count": len(events)},
-        })
-
-        async for chunk in _tail_events(request, bus, ["activity_recorded"]):
-            yield chunk
-
-    return StreamingResponse(gen(), media_type="application/x-ndjson")
-
-
-# --------------------------------------------------------------------------- #
-# 999 ACKs (read views)
-# --------------------------------------------------------------------------- #
-
 
 # ---------------------------------------------------------------------------
 # SP9: providers / payers / clearhouse endpoints
@@ -3037,5 +2841,18 @@ app.include_router(admin_users_router, dependencies=[Depends(matrix_gate)])
 # line once the test is updated to import from the new home.
 from cyclone.api_routers.payers import _clear_summary_cache  # noqa: F401  # SP36 backward-compat shim
 
+# SP36 Tasks 9+10 backward-compat: ``remittances_stream`` and
+# ``activity_stream`` moved into ``cyclone.api_routers.remittances``
+# and ``cyclone.api_routers.activity`` respectively (they are
+# single-router handlers per spec D4). The live-tail integration
+# test in ``tests/test_api_stream_live.py`` imports them by name
+# from this module to drive ``_call_endpoint(stream_fn, path)``.
+# Per the SP-N invariant that we don't rewrite tests for a
+# structural refactor, expose them on this module's namespace via
+# one-line re-imports. Delete these lines once the test is updated
+# to import from the new homes.
+from cyclone.api_routers.remittances import remittances_stream  # noqa: F401  # SP36 backward-compat shim
+from cyclone.api_routers.activity import activity_stream  # noqa: F401  # SP36 backward-compat shim
+
 
 __all__ = ["app"]

backend/src/cyclone/api_routers/__init__.py

@@ -12,6 +12,7 @@ from fastapi import APIRouter
 
 from cyclone.api_routers import (
     acks,
+    activity,
     admin,
     claim_acks,
     config,
@@ -20,11 +21,13 @@ from cyclone.api_routers import (
     health,
     payers,
     reconciliation,
+    remittances,
     ta1_acks,
 )
 
 routers: list[APIRouter] = [
     acks.router,        # gated
+    activity.router,    # gated
     admin.router,       # gated
     claim_acks.router,  # gated
     config.router,      # gated
@@ -33,6 +36,7 @@ routers: list[APIRouter] = [
     health.router,      # public — health probes must work pre-auth
     payers.router,      # gated
     reconciliation.router, # gated
+    remittances.router, # gated
     ta1_acks.router,    # gated
 ]
 

backend/src/cyclone/api_routers/activity.py

@@ -0,0 +1,102 @@
+"""``/api/activity`` and ``/api/activity/stream`` — operator-facing event log.
+
+Two endpoints, both gated by ``matrix_gate``:
+
+- ``GET /api/activity``          — paginated event list with ``kind`` /
+  ``since`` filters, plus an NDJSON variant when the caller sends
+  ``Accept: application/x-ndjson``.
+- ``GET /api/activity/stream``   — NDJSON live-tail: snapshot of the
+  most recent N events, then ``activity_recorded`` events as they
+  hit the store. Default ``limit`` is 50 (smaller than the list
+  endpoint's 200) because activity is high-volume — callers usually
+  want the most recent handful, not a full replay.
+
+The snapshot halves of ``/api/activity`` and ``/api/activity/stream``
+share the same in-memory filter logic (``kind`` + ``since``) so the
+two endpoints are interchangeable for the snapshot half.
+
+SP36 Task 10: this block moved here from ``api.py:2606`` (the
+``/api/activity*`` pair).
+"""
+from __future__ import annotations
+
+from typing import Any, AsyncIterator
+
+from fastapi import APIRouter, Depends, Query, Request
+from fastapi.responses import StreamingResponse
+
+from cyclone.api_helpers import (
+    ndjson_line as _ndjson_line,
+    ndjson_stream_list as _ndjson_stream_list,
+    tail_events as _tail_events,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.pubsub import EventBus
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/activity")
+def list_activity(
+    request: Request,
+    kind: str | None = Query(None),
+    since: str | None = Query(None),
+    limit: int = Query(200, ge=1, le=5000),
+) -> Any:
+    events = store.recent_activity(limit=limit)
+    if kind is not None:
+        events = [e for e in events if e["kind"] == kind]
+    if since is not None:
+        events = [e for e in events if e["timestamp"] >= since]
+    total = len(events)
+    has_more = False
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(events, total, total, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": events,
+        "total": total,
+        "returned": total,
+        "has_more": has_more,
+    }
+
+
+@router.get("/api/activity/stream")
+async def activity_stream(
+    request: Request,
+    kind: str | None = Query(None),
+    since: str | None = Query(None),
+    limit: int = Query(50, ge=1, le=5000),
+) -> StreamingResponse:
+    """Stream Activity events as NDJSON: snapshot first, then live events.
+
+    Subscribes to ``activity_recorded``. Default ``limit`` is 50
+    (smaller than the list endpoint's 200) because activity is
+    high-volume — callers usually want the most recent handful, not a
+    full replay.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        # Snapshot reuses the same in-memory filter as ``list_activity``
+        # so the two endpoints are interchangeable for the snapshot
+        # half.
+        events = store.recent_activity(limit=limit)
+        if kind is not None:
+            events = [e for e in events if e["kind"] == kind]
+        if since is not None:
+            events = [e for e in events if e["timestamp"] >= since]
+        for ev in events:
+            yield _ndjson_line({"type": "item", "data": ev})
+        yield _ndjson_line({
+            "type": "snapshot_end", "data": {"count": len(events)},
+        })
+
+        async for chunk in _tail_events(request, bus, ["activity_recorded"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")

backend/src/cyclone/api_routers/remittances.py

@@ -0,0 +1,169 @@
+"""``/api/remittances`` and ``/api/remittances/stream`` — read views over the Remittance population.
+
+Four endpoints, all gated by ``matrix_gate``:
+
+- ``GET /api/remittances``                — paginated list with filter+sort,
+  plus an NDJSON variant when the caller sends ``Accept: application/x-ndjson``.
+- ``GET /api/remittances/summary``        — server-aggregated KPI tiles
+  (``count``, ``total_paid``, ``total_adjustments``) over the full
+  filtered population — never a page-limited sample (SP27 fix).
+- ``GET /api/remittances/stream``         — NDJSON live-tail: snapshot
+  of currently-known rows, then ``remittance_written`` events as
+  they hit the store. Subscribed to by the Remittances page.
+- ``GET /api/remittances/{remittance_id}`` — one remittance with its
+  labeled CAS ``adjustments`` array. 404 on missing id (never 500).
+
+``/api/remittances/stream`` is registered before
+``/api/remittances/{remittance_id}`` so the literal ``stream`` path
+segment is not captured as a remittance id.
+
+SP36 Task 9: this block moved here from ``api.py:2448`` (the 4
+``/api/remittances*`` routes, with the streaming route's
+``NOTE: registered before…`` comment preserved verbatim).
+"""
+from __future__ import annotations
+
+from typing import Any, AsyncIterator
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.responses import StreamingResponse
+
+from cyclone.api_helpers import (
+    ndjson_line as _ndjson_line,
+    ndjson_stream_list as _ndjson_stream_list,
+    tail_events as _tail_events,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.pubsub import EventBus
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/remittances")
+def list_remittances(
+    request: Request,
+    batch_id: str | None = Query(None),
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+    offset: int = Query(0, ge=0),
+) -> Any:
+    common = dict(
+        batch_id=batch_id,
+        payer=payer,
+        claim_id=claim_id,
+        date_from=date_from,
+        date_to=date_to,
+    )
+    items = list(store.iter_remittances(
+        sort=sort, order=order, limit=limit, offset=offset, **common,
+    ))
+    # SP27 Task 13b: count the full population, not a 100-row sample.
+    # See the matching note in list_claims — same silent-failure pattern.
+    total = store.count_remittances(**common)
+    returned = len(items)
+    has_more = total > offset + returned
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(items, total, returned, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": items,
+        "total": total,
+        "returned": returned,
+        "has_more": has_more,
+    }
+
+
+@router.get("/api/remittances/summary")
+def remittances_summary(
+    batch_id: str | None = Query(None),
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+) -> dict:
+    """Server-aggregated KPI tiles for the Remittances page.
+
+    Returns ``{count, total_paid, total_adjustments}`` over the
+    full filtered remittance population — NOT a page-limited
+    sample. The Remittances page consumes this for its "Total paid"
+    and "Adjustments" tiles so they can't silently understate the
+    true DB population the way a page-local ``items.reduce(...)``
+    would. Mirrors the silent-incompleteness fix that
+    ``/api/dashboard/kpis`` (commit ``59c3275``) and
+    ``/api/remittances`` (commit ``d81b6ed``) made for their tiles.
+
+    Same filter parameters as ``/api/remittances``. Always returns
+    a populated dict (``{"count": 0, "total_paid": 0,
+    "total_adjustments": 0}`` when no rows match) so the frontend
+    can render the tiles directly without a loading-vs-empty
+    branch.
+    """
+    return store.summarize_remittances(
+        batch_id=batch_id, payer=payer, claim_id=claim_id,
+        date_from=date_from, date_to=date_to,
+    )
+
+
+@router.get("/api/remittances/stream")
+async def remittances_stream(
+    request: Request,
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> StreamingResponse:
+    """Stream Remittances as NDJSON: snapshot first, then live events.
+
+    Subscribes to ``remittance_written``. Default sort is
+    ``-received_date`` (newest-first), matching the list endpoint's
+    most common sort.
+
+    NOTE: registered before ``/api/remittances/{remittance_id}`` so
+    the literal ``stream`` path segment doesn't get matched as a
+    remittance id.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        rows = store.iter_remittances(
+            payer=payer, claim_id=claim_id,
+            date_from=date_from, date_to=date_to,
+            sort=sort or "-received_date", order=order, limit=limit,
+        )
+        for row in rows:
+            yield _ndjson_line({"type": "item", "data": row})
+        yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
+
+        async for chunk in _tail_events(request, bus, ["remittance_written"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")
+
+
+@router.get("/api/remittances/{remittance_id}")
+def get_remittance(remittance_id: str) -> dict:
+    """Return one remittance with its labeled CAS ``adjustments`` array.
+
+    Path param is ``remittance_id`` (not ``id``) to avoid shadowing
+    FastAPI's internal ``id`` name and to keep OpenAPI docs self-
+    describing. Returns 404 when the remittance is missing — never 500.
+    """
+    body = store.get_remittance(remittance_id)
+    if body is None:
+        raise HTTPException(
+            status_code=404,
+            detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"},
+        )
+    return body