content(cash-stats,journal): devices-primary cassette data flow (#156)

Cash Stats now reads devices (record of reference) with an incident fallback;
journal's CassetteInventoryService dual-writes incident + journal.cassette-baseline
to devices. Documents the mask-aware RCY parse, the removed duplicate parser, the
new topics, and the nginx /journal/ path quirk.

Refs #156

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/src/main/resources/content/dashboard/cash-stats/architect.md

@@ -12,9 +12,17 @@ How the **Cash Stats** page is built end-to-end. It is a thin read view over dat
 
 | Service | Role for this feature |
 |---------|----------------------|
-| `hiveops-dashboard` (SPA, 5181) | Renders `CashStatus.svelte`; no backend |
-| `hiveops-incident` (8080) | Owns `journal_events` + `atms`; serves the read API |
-| `hiveops-journal` (8087, **upload** role) | Parses ATM journals, extracts cassette counts, **writes into incident's DB** |
+| `hiveops-dashboard` (SPA, 5181) | Renders `CashStatus.svelte`; no backend. Reads **devices first, incident as fallback** (#156) |
+| `hiveops-devices` (8096) | **Record of reference (#156)** — owns `device_cassette_config`; serves `/api/cassette-inventory/fleet` |
+| `hiveops-incident` (8080) | Owns `journal_events` + `atms`; serves the legacy read API (fallback for machines devices can't cover) |
+| `hiveops-journal` (8087, **upload** role) | Parses ATM journals, extracts cassette counts, **dual-writes** to incident's DB *and* publishes a baseline to devices |
+
+> **#156 migration state.** Devices is the record of reference, but the journal still writes incident's
+> `journal_events` in parallel, and the dashboard falls back to incident for any machine devices doesn't
+> yet cover. Devices only covers machines that emit a journal **RCY / `[DENOMINATION]`** baseline (Hyosung
+> recyclers). Agent-bulk (`HIVEOPS_AGENT`), config-only (`CASSETTE_CONFIG`) and non-Hyosung formats
+> (NCR / ATEC / TCR) stay on the incident fallback. At cutover: prod devices covered ~197 of 237, plus 11
+> devices-only — merge shows the union, **nothing drops**.
 
 ## Data flow
 
@@ -25,15 +33,25 @@ journal.parse-requests.{INSTITUTION} ────────────┘
         │ worker parses file, persists JournalTransaction (STARTUP has rawLines)
         ▼
 journal.file-parsed  ──►  CassetteInventoryService (upload role, groupId journal-cassette-inventory)
-        │  extracts [DENOMINATION] / RCY USD / MIX / RETRACT slots from the newest STARTUP
-        ▼
-  incident DB: journal_events  (event_type='CASSETTE_INVENTORY', event_source='JOURNAL')
-        ▲
-        │  GET /api/journal-events/fleet/cassette-inventory   (+ GET /api/atms)
-CashStatus.svelte  ── merges events with ATM list, colours pills, sorts empty→low→ok→config→nodata
+        │  extractSlots(): [DENOMINATION] / RCY USD (mask-aware) / MIX / RETRACT from the newest STARTUP
+        ├──────────────────────────────►  incident DB: journal_events (event_source='JOURNAL')   [legacy]
+        │
+        └── journal.cassette-baseline ──►  devices: CassetteBaselineConsumer → device_cassette_config  [#156]
+                                           (+ journal.cassette-remain → live counts, + replenishment events)
+        ▲                                        ▲
+        │ GET /incident/api/journal-events/      │ GET /devices/api/cassette-inventory/fleet   [primary]
+        │     fleet/cassette-inventory [fallback]│
+CashStatus.svelte ── reads devices first, falls back to incident for uncovered ATMs (keyed by name),
+                     groups by ATM, colours pills, sorts empty→low→ok→config→nodata
 ```
 
-Key point: this is the **executor→Kafka→record-store** pattern applied across a service boundary. `hiveops-journal` is the executor (parse), `journal.file-parsed` is the event, and the "record store" it writes to is **incident's `journal_events` table** via a second read/write JDBC datasource (`incidentJdbcTemplate`), not its own DB. See [platform.kafka], [platform.data-architecture].
+Key point: **one extractor, two sinks (#156).** `CassetteInventoryService.extractSlots()` is the single
+parse of the STARTUP cash report; it writes incident's `journal_events` (legacy, via `incidentJdbcTemplate`)
+**and** publishes the same denominated slots to devices on `journal.cassette-baseline`. Devices is the
+record of reference; incident is kept only as the transitional read fallback. The masked-column RCY format
+(`RCY USD 1 ***************1536 0000` — INIT/DISP/STOR hidden, trailing **CST** = current count) is handled
+by the mask-aware `RCY_LINE` here; the earlier duplicate baseline parser in `JournalParser` rejected the
+mask and was removed. See [platform.kafka], [platform.data-architecture].
 
 ## Kafka
 
@@ -53,12 +71,15 @@ Key point: this is the **executor→Kafka→record-store** pattern applied acros
 3. `resolveNumericId(atmId)` → `SELECT id FROM atms WHERE atm_id=?` (incident). Null → skip + warn.
 4. **Freshness guard:** if incident already has a newer `CASSETTE_INVENTORY.event_time` for that ATM, skip.
 5. Otherwise `DELETE` existing `CASSETTE_INVENTORY` rows for the ATM, then `INSERT` fresh slots (`event_source='JOURNAL'`, `cassette_currency='USD'`).
+6. **#156 — publish to devices:** `publishDenominationBaselineToDevices()` takes the same slots, drops MIX/RETRACT (no denomination), and publishes a `CassetteBaselineEvent` on `journal.cassette-baseline` (denomination-ascending, positions 1..N, stamped with the restart time). Devices' `CassetteBaselineConsumer` upserts `device_cassette_config` — denomination is authoritative; bill count is seeded on new rows only and thereafter owned by the live Remain Count consumer.
 
-`backfillAll()` does the same across every ATM with STARTUPs (admin sync endpoint).
+`backfillAll()` does the same (both sinks) across every ATM with STARTUPs — the admin sync endpoint is how the devices denomination map gets seeded from history (offset-reset is `latest`).
 
-## Read path (incident → dashboard)
+## Read path (devices primary, incident fallback → dashboard)
 
-`JournalEventController.getFleetCassetteInventory()` → `JournalEventService.getFleetCassetteInventory(scope)`:
+**#156:** `dashboardAPI.getFleetCassetteInventory()` (in `api.ts`) fetches **both** `/devices/api/cassette-inventory/fleet` and `/incident/api/journal-events/fleet/cassette-inventory`, keys rows by ATM **name**, and returns devices rows plus incident rows for any machine devices doesn't cover. Devices wins; each source degrades independently (a failed call yields no rows, not an error). This is what keeps agent-bulk / config-only / non-Hyosung machines on screen while devices coverage grows.
+
+Both services scope the same way. Incident side — `JournalEventController.getFleetCassetteInventory()` → `JournalEventService.getFleetCassetteInventory(scope)`:
 - Scope from `InstitutionContext.resolveScope()` → resolved to institution *names* via `institutionKeyService`.
 - Repo query returns the **latest event per (atm_id, cassettePosition)** using `MAX(id)` grouped by `(atm.id, COALESCE(cassettePosition,''))`, with `JOIN FETCH e.atm` (no N+1). Three variants: unrestricted / single-institution / `institution IN (...)`.
 - For ATMs with **no** live inventory events, it synthesizes slots from `AtmProperties`:
@@ -82,7 +103,8 @@ The **Total** shown under each ATM name is **derived in the component**, not ser
 
 | Table | Service / DB | Access |
 |-------|--------------|--------|
-| `journal_events` | incident (`hiveiq_incident`) | read by API; **written by journal** via `incidentJdbcTemplate` |
+| `device_cassette_config` | **devices (`hiveiq_devices`)** | **record of reference (#156)** — written by baseline/remain/replenishment consumers, served by `/api/cassette-inventory/fleet` |
+| `journal_events` | incident (`hiveiq_incident`) | read by the fallback API; **written by journal** via `incidentJdbcTemplate` (legacy, kept during transition) |
 | `atms` | incident | read (both by API and by journal's `resolveNumericId`) |
 | `journal_transactions` | journal (`hiveiq_journal`) | read by `CassetteInventoryService` (STARTUP rawLines) |
 
@@ -90,8 +112,9 @@ The **Total** shown under each ATM name is **derived in the component**, not ser
 
 | Method | Path | Service | Notes |
 |--------|------|---------|-------|
-| GET | `/api/journal-events/fleet/cassette-inventory` | incident | institution-scoped fleet inventory |
-| GET | `/api/atms` | incident | ATM list for merge/labels |
-| POST | `/api/journal/admin/sync-cassette-inventory` | journal | admin backfill (`MSP_ADMIN`/`BCOS_ADMIN`) |
+| GET | `/devices/api/cassette-inventory/fleet` | devices | **primary (#156)** — institution-scoped fleet inventory |
+| GET | `/incident/api/journal-events/fleet/cassette-inventory` | incident | fallback for machines devices can't cover |
+| GET | `/devices/api/atms` | devices | ATM list for merge/labels |
+| POST | `/journal/admin/sync-cassette-inventory` | journal | admin backfill → both sinks (`MSP_ADMIN`/`BCOS_ADMIN`). Note: prod/dev nginx maps `/journal/` → `/api/journal/`, so the external path omits the second `/api/journal` |
 
 Cross-links: [platform.data-architecture], [platform.kafka], [platform.service-ownership].

backend/src/main/resources/content/dashboard/cash-stats/claude.md

@@ -10,27 +10,33 @@ Terse agent reference for `dashboard.cash-stats`. Read view over cassette bill c
 
 ## Ownership
 - **Frontend:** `hiveops-dashboard` → `frontend/src/components/Dashboard/CashStatus.svelte` (no backend).
-- **Read API owner:** `hiveops-incident` — `JournalEventController` / `JournalEventService`.
-- **Data producer:** `hiveops-journal` (upload role) — `CassetteInventoryService` writes into incident's DB.
-- Dashboard SPA has **no server**; all data is incident's.
+- **Record of reference (#156):** `hiveops-devices` — `device_cassette_config`, served by `FleetCassetteController` at `/api/cassette-inventory/fleet`.
+- **Fallback read API:** `hiveops-incident` — `JournalEventController`/`JournalEventService` (transitional; covers machines devices can't).
+- **Data producer:** `hiveops-journal` (upload role) — `CassetteInventoryService` **dual-writes**: incident DB (legacy) + `journal.cassette-baseline` → devices.
+- Dashboard SPA has **no server**; `api.ts::getFleetCassetteInventory` reads devices first, falls back to incident for uncovered ATMs (keyed by name).
 
 ## Endpoints (METHOD path → service)
 | Method | Path | Service | Auth |
 |--------|------|---------|------|
-| GET | `/api/journal-events/fleet/cassette-inventory` | incident | any app role, institution-scoped |
-| GET | `/api/atms` | incident | any app role |
-| POST | `/api/journal/admin/sync-cassette-inventory` | journal | `MSP_ADMIN`/`BCOS_ADMIN` |
-| POST | `/api/journal/admin/reconcile` | journal | `MSP_ADMIN`/`BCOS_ADMIN` (run post-deploy) |
+| GET | `/devices/api/cassette-inventory/fleet` | devices | **primary (#156)**, institution-scoped |
+| GET | `/incident/api/journal-events/fleet/cassette-inventory` | incident | fallback, institution-scoped |
+| GET | `/devices/api/atms` | devices | ATM list for merge/labels |
+| POST | `/journal/admin/sync-cassette-inventory` | journal | `MSP_ADMIN`/`BCOS_ADMIN` — backfills both sinks |
+| POST | `/journal/admin/reconcile` | journal | `MSP_ADMIN`/`BCOS_ADMIN` (run post-deploy) |
 
-- Frontend calls these via configured API base (`window.__APP_CONFIG__.apiUrl`); paths above are the incident controller mappings.
+- Frontend calls via `window.__APP_CONFIG__.apiUrl`; the `${GW}` root is the incident base with `/incident(/api)` stripped, so devices/incident are addressed by service prefix.
+- **nginx quirk:** prod *and* dev map external `/journal/` → service `/api/journal/`. So the external admin path is `/journal/admin/…`, **not** `/journal/api/journal/admin/…` (that 404s with a doubled `/api/journal`).
 
 ## Tables
-- `journal_events` — incident DB (`hiveiq_incident`). Filter `event_type='CASSETTE_INVENTORY'`. **Written by journal**, read by incident API.
+- `device_cassette_config` — devices DB (`hiveiq_devices`). **Record of reference (#156).** Denomination from the baseline (authoritative); bill count seeded on new rows, then owned by the Remain Count consumer.
+- `journal_events` — incident DB (`hiveiq_incident`). Filter `event_type='CASSETTE_INVENTORY'`. Written by journal (legacy), read by the fallback API.
 - `atms` — incident DB. Journal resolves `atm_id` (string) → `id` (numeric) here.
 - `journal_transactions` — journal DB (`hiveiq_journal`). Source STARTUP `rawLines`.
 
 ## Topics
 - `journal.file-parsed` — consumed by `CassetteInventoryService` (groupId `journal-cassette-inventory`, `auto.offset.reset=latest`).
+- `journal.cassette-baseline` (#156) — produced by `CassetteInventoryService`, consumed by devices `CassetteBaselineConsumer` (denomination map).
+- `journal.cassette-remain` (#156) — per-withdrawal live counts, consumed by devices `CassetteRemainConsumer` (time-guarded, monotonic by withdrawal time).
 - No topic is consumed/produced by the dashboard itself.
 
 ## Roles
@@ -57,8 +63,12 @@ Terse agent reference for `dashboard.cash-stats`. Read view over cassette bill c
 - `CassetteInventoryService` runs **only in journal upload role** (`journal.role=upload`).
 - ATMs whose `atm_id` isn't in incident `atms` get skipped (logged warn), so they show as "No Data."
 
+## #156 coverage
+- Devices only covers machines that emit a journal **RCY / `[DENOMINATION]`** baseline (Hyosung recyclers). `HIVEOPS_AGENT` bulk, `CASSETTE_CONFIG` config-only, and NCR/ATEC/TCR formats have no such baseline → they ride the **incident fallback**. Full devices parity is not achievable via journal for those; that is why the fallback exists.
+- Real masked RCY format: `RCY USD 1 ***************1536 0000` — INIT/DISP/STOR masked, trailing digits = **CST** (current count). Parsed by the mask-aware `RCY_LINE` in `CassetteInventoryService`. The old `JournalParser.extractCassetteBaseline` duplicate (strict numeric columns) was removed — do not reintroduce a second baseline parser.
+
 ## Do NOT
-- Do NOT add a cassette/cash endpoint to `hiveops-dashboard` — it has no backend; use incident.
+- Do NOT add a cassette/cash endpoint to `hiveops-dashboard` — it has no backend; read devices (primary) then incident (fallback).
 - Do NOT add a backend/DTO field for the grand total — it is derived in the component; the backend deliberately returns raw per-slot rows.
 - Do NOT include MIX/RETRACT slots in the total (null denomination → they would silently contribute `0` and understate nothing visibly); exclude them and flag with `*`.
 - Do NOT write `CASSETTE_INVENTORY` rows from anywhere but journal's `CassetteInventoryService` (single writer owns the delete+insert).

backend/src/main/resources/content/dashboard/cash-stats/internal.md

@@ -11,12 +11,14 @@ Ops / support view for the **Cash Stats** page on `dashboard.bcos.cloud`. It sho
 ## Who serves this
 
 - **Frontend:** `hiveops-dashboard` (SPA, port 5181) — `CashStatus.svelte`. No backend of its own.
-- **Data API:** `hiveops-incident` backend (port 8080) — `JournalEventController`.
-- **Data producer:** `hiveops-journal` (upload role) parses ATM STARTUP journals and writes `CASSETTE_INVENTORY` rows into the **incident** DB.
+- **Primary data API (#156):** `hiveops-devices` (port 8096) — `FleetCassetteController`, `device_cassette_config` (the record of reference).
+- **Fallback data API:** `hiveops-incident` (port 8080) — `JournalEventController` (covers machines devices can't).
+- **Data producer:** `hiveops-journal` (upload role) parses ATM STARTUP journals and **dual-writes** — `CASSETTE_INVENTORY` rows into incident *and* a baseline to devices via `journal.cassette-baseline`.
 
-The page makes exactly two calls on load and every 60 s:
-- `GET /api/journal-events/fleet/cassette-inventory` (incident)
-- `GET /api/atms` (incident)
+The page makes three calls on load and every 60 s (devices + incident merged client-side, devices wins; plus the ATM list):
+- `GET /devices/api/cassette-inventory/fleet` (devices, **primary**)
+- `GET /incident/api/journal-events/fleet/cassette-inventory` (incident, **fallback** for uncovered ATMs)
+- `GET /devices/api/atms` (ATM list for labels)
 
 ## Roles required
 
@@ -26,8 +28,9 @@ The page makes exactly two calls on load and every 60 s:
 
 ## First things to check when it misbehaves
 
-1. **Whole page errors / red banner** → incident backend down or unreachable. `curl` the two endpoints with a real JWT; expect `200` + JSON array.
-2. **Everything shows "No Data" / "No inventory reported"** → no `CASSETTE_INVENTORY` rows in incident `journal_events`. Usual cause: journal parsing stalled or the cassette-inventory Kafka consumer isn't running. Check `hiveops-journal` (upload role) is up and that files are parsing (`parse_status=PARSED`).
+1. **Whole page errors / red banner** → devices *and* incident both unreachable (the merge tolerates one being down). `curl` both fleet endpoints with a real JWT; expect `200` + JSON array.
+2. **Everything shows "No Data" / "No inventory reported"** → neither devices `device_cassette_config` nor incident `journal_events` has rows. Usual cause: journal parsing stalled or a cassette Kafka consumer isn't running. Check `hiveops-journal` (upload role) is up and files are parsing (`parse_status=PARSED`); then run the backfill.
+2b. **A machine dropped off vs the old (incident-only) view** → it's covered by incident but not devices, and the incident fallback failed or was filtered. Confirm the incident fleet endpoint returns it; remember devices only covers RCY/`[DENOMINATION]` machines, everything else relies on the fallback.
 3. **One ATM stuck stale / never updates** → its STARTUP journals aren't parsing, or the ATM's `atm_id` string doesn't resolve to a numeric id in incident's `atms` table (`resolveNumericId` returns null → row skipped, logged as a warning).
 4. **Counts look wrong for a recycler** → the journal parser reads the `[DENOMINATION]` table (or falls back to `RCY USD` lines) out of the STARTUP record; a firmware format it doesn't match yields missing/zero slots.
 5. **Cassettes show but say "(no data)"** → those are **config-only** synthetic slots. The ATM has a cassette *configuration* (agent-synced or a manually applied template) but has reported no actual bill count yet. Not an error.
@@ -36,9 +39,10 @@ The page makes exactly two calls on load and every 60 s:
 
 ## Admin actions (backend, not on the page)
 
-- **Force a fleet re-derive of cash inventory** from the latest STARTUP journal of every ATM:
-  `POST /api/journal/admin/sync-cassette-inventory` (journal service, `MSP_ADMIN`/`BCOS_ADMIN`). Runs `CassetteInventoryService.backfillAll()`.
-- After any journal deploy, run the standard journal **reconcile** (`POST /api/journal/admin/reconcile`) so stranded files finish parsing — otherwise new inventory never lands.
+- **Force a fleet re-derive of cash inventory** (writes both incident and the devices baseline) from the latest STARTUP journal of every ATM:
+  `POST /journal/admin/sync-cassette-inventory` (journal, `MSP_ADMIN`/`BCOS_ADMIN`). Runs `CassetteInventoryService.backfillAll()`. This is how the devices denomination map is seeded from history (consumers are `offset-reset=latest`).
+- After any journal deploy, run the standard journal **reconcile** (`POST /journal/admin/reconcile`) so stranded files finish parsing — otherwise new inventory never lands.
+- **nginx path quirk:** external `/journal/…` maps to service `/api/journal/…`, so the admin paths are `/journal/admin/…` (a `/journal/api/journal/admin/…` URL 404s).
 
 ## Common failure modes → fix
 

backend/src/main/resources/content/technical/journal/overview.md

@@ -48,6 +48,14 @@ Topic constants defined in `kafka/JournalKafkaConfig.java`. See [platform.kafka]
 | `journal.replenishments` | `ReplenishmentEventProducer` |
 | `journal.file-parsed` | `FileParsedEventProducer` |
 | `hiveops.journal.transactions.recorded` | `JournalTransactionEventProducer` |
+| `journal.cassette-baseline` (#156) | `CassetteBaselineProducer` — from `CassetteInventoryService` (upload role) on `file-parsed`; the RCY/`[DENOMINATION]` denomination baseline for devices |
+| `journal.cassette-remain` (#156) | `CassetteRemainProducer` — from `JournalParseConsumer` (**worker** role); the latest withdrawal's per-cassette Remain Count for devices live counts |
+
+> **#156 — devices cassette feed.** `CassetteInventoryService.extractSlots()` is the single cassette parser
+> (mask-aware `RCY USD`, `[DENOMINATION]` table, MIX/RETRACT). It dual-writes: incident `journal_events`
+> (legacy) and `journal.cassette-baseline` → devices `device_cassette_config`. The earlier duplicate parser
+> `JournalParser.extractCassetteBaseline` (which rejected the masked column format) was removed. See
+> [dashboard.cash-stats].
 
 Static topic `journal.parse-requests` is also declared. Per-institution topics are created at startup from `JOURNAL_INSTITUTIONS` env plus an always-present `...UNASSIGNED`.