cyclone/.superpowers/skills/cyclone-store/SKILL.md

---
name: cyclone-store
description: "Cyclone store write-paths, the pubsub event contract (claim_written / remittance_written / activity_recorded), and the SP21 store-split boundary map. Use when: touching store.py, adding a new entity, wiring a new write event, or splitting a store module."
---

# cyclone-store

Cyclone persists every parsed X12 batch through one facade,
`CycloneStore` (now exposed via `backend/src/cyclone/store/__init__.py` after the SP21 split landed; the public import surface `from cyclone.store import …` is unchanged). Every write inserts the row AND publishes a pubsub event on the in-process `EventBus` (`backend/src/cyclone/pubsub.py:20`) so live-tail pages see new rows the moment they land. The event contract is the seam between persistence and streaming — a wrong event name silently goes stale.

As of this writing: the store is a **16-module subpackage** under `backend/src/cyclone/store/` (SP21 split landed). Event kinds in use today: `claim_written`, `remittance_written`, `activity_recorded`, plus the SP-acks-tail additions `ack_written` and `ta1_ack_written` for the `/api/acks/stream` and `/api/ta1-acks/stream` endpoints. Next: **SP43** (after the SP42 doc-pass merges).

## When to use

- **Adding a new entity.** You need a new ORM model + write method +
  event kind + snapshot serializer and want to know where each piece
  lives (current monolith vs. post-SP21 module).
- **Wiring a new write event.** You're adding a `<entity>_written`
  event and need both the publish call in the store AND the
  subscribe call in the live-tail endpoint to stay in sync.
- **Debugging a write-path issue.** A page isn't reflecting new
  rows, the DB has the row but the stream is silent, or the publish
  raises and rolls back the transaction.
- **Splitting a store module.** You're moving a domain out of
  `store.py` into its own module under `backend/src/cyclone/store/`
  and need the SP21 module list + facade re-export rules.

## Conventions

1. **All writes go through `CycloneStore`.** Route handlers and
   parsers must not write directly to the ORM session. The facade
   opens a short-lived session via `db.SessionLocal()()`. Direct ORM
   access is the #1 way the live-tail contract gets bypassed (no
   event published → page silently goes stale). Read paths are
   similar — prefer the facade's `iter_*` / `get_*` methods over raw
   `s.execute(select(...))`.
2. **Every write publishes an event.** The event name matches the
   entity: `claim_written`, `remittance_written`,
   `activity_recorded` (the trailing `_recorded` signals a
   non-canonical row — activity events are derived, not first-class),
   plus `ack_written` and `ta1_ack_written` for the SP-acks-tail
   streams. New entities get `<entity>_written`; activity-style side
   rows get `<entity>_recorded`. Publish is **best-effort** —
   failures are logged but never roll back the persisted batch.
3. **Snapshot shape.** Each entity has a `to_ui_<entity>` serializer
   (plain Python function returning a dict) — currently
   `to_ui_claim`, `to_ui_remittance`, `to_ui_claim_from_orm`,
   `to_ui_remittance_from_orm`, `to_ui_provider`. Post-SP21 these
   move to `backend/src/cyclone/store/ui.py`. The serializer is the
   single source of truth for what the frontend sees — every event
   payload MUST match what the matching list endpoint returns for
   that row (`store.py:1052-1054`).
4. **SP21 boundaries.** Post-split, each domain lives in its own
   module under `backend/src/cyclone/store/`: `__init__.py` (facade
   + `CycloneStore` class), `acks.py`, `backfill.py`, `backups.py`,
   `batches.py`, `claim_acks.py`, `claim_detail.py`, `exceptions.py`,
   `inbox.py`, `kpis.py`, `orm_builders.py`, `providers.py`,
   `records.py`, `resubmissions.py`, `submission_dedup.py`, `ui.py`,
   `write.py` — 16 modules total. Cross-module writes go through
   `CycloneStore` facade methods, not direct module access. The
   facade re-exports every name callers currently import from
   `cyclone.store`. Full split plan:
   `docs/superpowers/plans/2026-06-21-cyclone-store-split.md`.
5. **No business logic in route handlers.** A route handler validates
   input, calls `store.<method>(...)`, passes the `event_bus`,
   returns the serialized result. Reconciliation, idempotency
   checks, CAS adjustment persistence — all live in the store.

## Patterns

### A `CycloneStore.add` write — publishes events from inserted rows

Taken from `backend/src/cyclone/store/write.py` (post-SP21). The method opens
a session, inserts rows, then runs a sync `_publish_events_sync`
after commit so subscribers can immediately re-fetch consistent data.

```python
def add(
    self,
    record: BatchRecord,
    *,
    event_bus: "EventBus | None" = None,
) -> None:
    inserted_claim_ids: list[str] = []
    with db.SessionLocal()() as s:
        s.add(Batch(id=record.id, kind=record.kind, ...))
        if isinstance(record, BatchRecord837):
            for claim in record.result.claims:
                if s.get(Claim, claim.claim_id) is not None:
                    continue                       # idempotency: skip dupes
                s.add(_claim_837_row(claim, record.id))
                s.add(ActivityEvent(kind="claim_submitted", ...))
                inserted_claim_ids.append(claim.claim_id)
        # ... 835 branch + flush + cas adjustments ...
        s.commit()

    if event_bus is not None and inserted_claim_ids:
        self._publish_events_sync(event_bus, record, inserted_claim_ids)

def _publish_events_sync(self, event_bus, record, claim_ids):
    with db.SessionLocal()() as s:
        for cid in claim_ids:
            ui = to_ui_claim_from_orm(s.get(Claim, cid), ...)
            self._sync_publish(event_bus, "claim_written", ui)
        # ... remittance + activity loops ...
```

`EventBus.publish` is async but the body is pure sync `put_nowait`,
so the store calls `_sync_publish` directly to avoid forcing sync
FastAPI handlers to await.

### Backend `/api/<resource>/stream` endpoint — subscribes to the event

Streaming endpoints live in `backend/src/cyclone/api_routers/<topic>.py`
post-SP36 (e.g. `api_routers/claims.py`). Two phases — eager
snapshot, then live subscription — wrapped in `StreamingResponse`
with `media_type="application/x-ndjson"`.

```python
@router.get("/api/claims/stream")
async def claims_stream(request: Request, ...) -> StreamingResponse:
    bus: EventBus = request.app.state.event_bus

    async def gen() -> AsyncIterator[bytes]:
        rows = store.iter_claims(status=status, ...)         # 1. Snapshot
        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, ["claim_written"]):  # 2. Live
            yield chunk

    return StreamingResponse(gen(), media_type="application/x-ndjson")
```

The same shape is repeated in `api_routers/remittances.py`,
`api_routers/activity.py`, `api_routers/acks.py`, and
`api_routers/ta1_acks.py` (five streaming endpoints total).

`_ndjson_line` and `_tail_events` live in
`backend/src/cyclone/api_helpers.py`. The `["remittance_written"]`
and `["activity_recorded"]` subscriptions are at `api.py:1891,2002`.

### Backend test — asserts both the row AND the event landed

The autouse `conftest.py` (`backend/tests/conftest.py:20`) wires a
fresh `EventBus` onto `app.state` per test.

```python
def test_publishes_claim_written_event(client: TestClient) -> None:
    bus = app.state.event_bus
    resp = client.post("/api/parse-837",
                       files={"file": ("x.837", MINIMAL_837, "text/plain")})
    assert resp.status_code == 200
    assert list(store.iter_claims(limit=10))             # row landed

    queues = bus._subscribers.get("claim_written", [])    # event published
    assert queues
    evt = queues[0].get_nowait()
    assert evt["_kind"] == "claim_written"
```

## Anti-patterns

- **Don't read directly from the ORM in a route handler — go through
  the snapshot serializer.** A handler that does
  `with db.SessionLocal()() as s: row = s.get(Claim, cid); return row`
  bypasses `to_ui_claim_from_orm` and silently drifts from the event
  payload shape. Always call `store.get_claim_detail(cid)` (or the
  equivalent `get_*` facade method).
- **Don't introduce a new event name without updating the subscriber
  list.** Today every `<entity>_written` event has exactly one
  consumer — the matching `/api/<resource>/stream` endpoint, now
  split across `api_routers/{claims,remittances,activity,acks,ta1_acks}.py`
  (post-SP36 split). Any new event must wire its subscription in
  the matching router and add the resource name to the hook
  triplet in the corresponding page.
- **Don't merge a write method with its event publication into
  separate places.** `_publish_events_sync` lives next to `add` in
  `store.py:1042-1107` so reviewers see both halves of the contract
  in one diff. Post-SP21 the same rule applies.
- **Don't make the publish call blocking on commit failures.**
  Publish errors are caught and logged at `store.py:1097-1098`; a
  failing subscriber MUST NOT roll back the persisted batch. The
  batch is the source of truth; the event is the cache-invalidation
  hint.

## Related skills

- **`cyclone-edi`** — the parsed `ClaimOutput` / `ParseResult<EDI>`
  lands in the store via `CycloneStore.add`.
- **`cyclone-api-router`** — the route that calls `store.<method>(...)`
  and (for stream endpoints) subscribes to the matching
  `<entity>_written` event.
- **`cyclone-tail`** — the consumer side of the event contract;
  load when changing the wire format or adding a streaming hook.
- **`cyclone-tests`** — write-path tests live under `backend/tests/`
  and assert both the DB row and the event payload.