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

name, description

name	description
cyclone-store	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 (backend/src/cyclone/store.py:882). 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 single 2412-line module and the SP21 split into backend/src/cyclone/store/ is in progress. Three event kinds: claim_written, remittance_written, activity_recorded. Next: SP22.

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; current names at store.py:1072,1081,1096). 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 (store.py:1097-1098).
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), exceptions.py, records.py, orm_builders.py, ui.py, write.py, batches.py, claim_detail.py, acks.py, backups.py, inbox.py, providers.py. 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 list: docs/superpowers/plans/2026-06-21-cyclone-store-split.md:25-37.
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.py:898-1107. The method opens a session, inserts rows, then runs a sync _publish_events_sync after commit so subscribers can immediately re-fetch consistent data.

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

Taken from backend/src/cyclone/api.py:1380-1401. Two phases — eager snapshot, then live subscription — wrapped in StreamingResponse with media_type="application/x-ndjson".

@app.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")

_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.

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, currently in backend/src/cyclone/api.py (claims at :1398, remittances at :1891, activity at :2002). Any future split into backend/src/cyclone/api_routers/ must wire the same subscription.
• 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.