# Parse → Detect → Decide Workflow Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Replace the 409-on-persist workflow with parse-first, pre-flight dedup, structured 409 with full parse result + collision summary, `?force=true` to skip-and-continue, and `DELETE /api/batches/{id}` for cascade cleanup. Frontend gets an error panel with four actions (force-insert, open prior, delete prior, pick different file).

**Architecture:** Pre-flight dedup runs between validation and persist in both `parse_837` and `parse_835` endpoints. New `dedup.py` module owns `CollisionReport` + `preflight_837`/`preflight_835`. New `?force=true` query param bypasses pre-flight; the existing per-row `s.get(Claim, claim_id)` dedup in `store.add` silently skips colliding rows, and the 200 response includes `skipped_claim_ids`. New `DELETE /api/batches/{id}` endpoint hard-deletes with cascade; refuses if any claim is past `submitted` state. Frontend `ApiError` carries `collisions` + `parseResult`; `Upload.tsx` panel renders the full parse result and the four actions.

**Tech Stack:** Python 3.11+, SQLAlchemy 2.x, FastAPI, SQLite (sqlcipher3 optional), React 18 + TanStack Query + Radix UI + sonner, Vitest + React Testing Library, pytest.

**Spec:** `docs/superpowers/specs/2026-06-21-cyclone-parse-decide-workflow-design.md` — read fully before starting.

**Worktree setup (one-time):**

```bash
cd /Users/openclaw/dev/cyclone
# claims-unique-fix worktree already exists from prior SP work; Tasks 1.1 and 1.2 are done on it.
cd .worktrees/claims-unique-fix
# venv is symlinked to main repo's .venv
# Use PYTHONPATH override to make Python import the worktree's source, not the main repo's:
export PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src:$PYTHONPATH
```

All commits happen in this worktree. Merge to main via fast-forward when each phase ends.

---

## Phase 1 — Backend pre-flight dedup module + schema foundation (mostly done; **Task 1.3 is new foundation work added after spec/plan review revealed that `claims.id` PK prevents the workflow from firing**)

### Task 1.1: Migration 0013 drops inline UNIQUE — DONE

**Files:**
- `backend/src/cyclone/migrations/0013_drop_claims_unique_constraint.sql` (created)
- `backend/tests/test_db_migrate.py` (modified)

Commit `b6efd0e` on `claims-unique-fix`. No further action.

### Task 1.2: Store helpers `find_existing_batch_for_claim` / `find_existing_batch_for_remit` — DONE

**Files:**
- `backend/src/cyclone/store.py` (modified)
- `backend/tests/test_store.py` (modified)

Commit `890207f` on `claims-unique-fix`. No further action.

### Task 1.3: Migration 0014 — relax `claims` and `remittances` PKs to composite `(batch_id, id)`

The current schema has `claims.id` and `remittances.id` as single-column
PRIMARY KEYs. This makes the spec'd "cross-batch CLM01 collision" workflow
unreachable: the same CLM01 cannot exist in two batches, so the pre-flight
409 path can never fire, force-insert can never skip anything, and resubmits
are impossible. Migration 0014 relaxes the PKs to composite `(batch_id, id)`
and updates every FK that referenced the old single-column PK.

After 0014, the pre-flight dedup is genuinely exercisable end-to-end and
the entire workflow in the spec actually fires on real data.

**Files:**
- Create: `backend/src/cyclone/migrations/0014_relax_claims_remits_pk.sql`
- Modify: `backend/src/cyclone/db.py` (drop the redundant single-column `unique=True` on `id`, if any; add composite PK markers)
- Modify: `backend/tests/test_db_migrate.py` (0014 tests)

- [ ] **Step 1: Inspect the live schema and identify every FK**

Before writing the migration, enumerate every FK that points at `claims(id)` or `remittances(id)`. Run:

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -c "
from cyclone import db
from sqlalchemy import inspect
ins = inspect(db.engine())
for tbl in ins.get_table_names():
    for fk in ins.get_foreign_keys(tbl):
        if 'claims' in (fk.get('referred_table') or '') or 'remittances' in (fk.get('referred_table') or ''):
            print(tbl, fk['constrained_columns'], '->', fk['referred_table'], fk['referred_columns'])
"
```

You should see (at minimum) FKs from `remittances`, `matches`, `cas_adjustments`, `line_reconciliations`, `activity_events` (if FK-declared there), and `claims.matched_remittance_id`. Capture this list — the migration must update each one to include `batch_id`.

- [ ] **Step 2: Write the failing migration tests**

Add to `backend/tests/test_db_migrate.py`:

```python
def test_migration_0014_relaxes_claims_pk_to_composite(tmp_path, monkeypatch):
    """Migration 0014 changes claims PK from single-column id to (batch_id, id).

    After 0014, two Claim rows with the same id can coexist if they are
    in different batches. Same for Remittance. The dedup story moves to
    the application layer (preflight_837 / preflight_835) instead of the
    schema.
    """
    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
    from cyclone import db, migrations

    # Synthesize 0001 + 0013 + 0014 by calling the migration runner with
    # only those versions. Use the existing migrate runner pattern from
    # test_db_migrate.py.
    db._reset_for_tests()
    db.init_db()  # runs all migrations up to current

    # After full migration: try to insert two claims with the same id
    # in different batches.
    from datetime import datetime, timezone
    from cyclone.db import Batch, Claim

    with db.SessionLocal()() as s:
        s.add(Batch(id="B1", kind="837p", input_filename="b1.txt",
                    parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
                    raw_result_json={}))
        s.add(Batch(id="B2", kind="837p", input_filename="b2.txt",
                    parsed_at=datetime(2026, 1, 2, tzinfo=timezone.utc),
                    raw_result_json={}))
        s.add(Claim(id="CLM-A", batch_id="B1", patient_control_number="M1",
                    state_changed_at=datetime(2026, 1, 1, tzinfo=timezone.utc)))
        s.add(Claim(id="CLM-A", batch_id="B2", patient_control_number="M1",
                    state_changed_at=datetime(2026, 1, 2, tzinfo=timezone.utc)))
        s.commit()  # must NOT raise

    with db.SessionLocal()() as s:
        rows = s.execute(db.text("SELECT id, batch_id FROM claims WHERE id='CLM-A' ORDER BY batch_id")).all()
        assert rows == [("CLM-A", "B1"), ("CLM-A", "B2")]


def test_migration_0014_relaxes_remittances_pk_to_composite(tmp_path, monkeypatch):
    """Same shape for remittances: same CLP01 can exist in two batches."""
    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
    from cyclone import db
    from datetime import datetime, timezone
    from cyclone.db import Batch, Remittance

    db._reset_for_tests()
    db.init_db()

    with db.SessionLocal()() as s:
        s.add(Batch(id="B1", kind="835", input_filename="b1.txt",
                    parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
                    raw_result_json={}))
        s.add(Batch(id="B2", kind="835", input_filename="b2.txt",
                    parsed_at=datetime(2026, 1, 2, tzinfo=timezone.utc),
                    raw_result_json={}))
        s.add(Remittance(id="CLP-A", batch_id="B1",
                         payer_claim_control_number="CLP-A",
                         status_code="1",
                         received_at=datetime(2026, 1, 1, tzinfo=timezone.utc)))
        s.add(Remittance(id="CLP-A", batch_id="B2",
                         payer_claim_control_number="CLP-A",
                         status_code="1",
                         received_at=datetime(2026, 1, 2, tzinfo=timezone.utc)))
        s.commit()

    with db.SessionLocal()() as s:
        rows = s.execute(db.text("SELECT id, batch_id FROM remittances WHERE id='CLP-A' ORDER BY batch_id")).all()
        assert rows == [("CLP-A", "B1"), ("CLP-A", "B2")]


def test_migration_0014_preserves_existing_data(tmp_path, monkeypatch):
    """If the DB already has claim/remittance rows when 0014 runs, the rows
    survive the table recreation (INSERT INTO new SELECT * FROM old).
    """
    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
    # Apply migrations 0001 through 0013, insert a row, then apply 0014.
    # (See test_db_migrate.py for the migrate-to-version helper.)
    from cyclone import db
    from datetime import datetime, timezone
    from cyclone.db import Batch, Claim

    db._reset_for_tests()
    db.init_db()  # up to 0014

    with db.SessionLocal()() as s:
        s.add(Batch(id="B-OLD", kind="837p", input_filename="x.txt",
                    parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
                    raw_result_json={}))
        s.add(Claim(id="CLM-OLD", batch_id="B-OLD", patient_control_number="M",
                    state_changed_at=datetime(2026, 1, 1, tzinfo=timezone.utc)))
        s.commit()

    # Verify row still exists.
    with db.SessionLocal()() as s:
        row = s.execute(db.text("SELECT id, batch_id FROM claims WHERE id='CLM-OLD'")).first()
    assert row == ("CLM-OLD", "B-OLD")
```

- [ ] **Step 3: Run the new tests, expect FAIL**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_db_migrate.py -k "0014" --no-header -q
```

Expected: FAIL with `IntegrityError: UNIQUE constraint failed: claims.id` (the same row insertion that succeeds after 0014 fails before it).

- [ ] **Step 4: Write migration 0014**

Create `backend/src/cyclone/migrations/0014_relax_claims_remits_pk.sql`:

```sql
-- version: 14
-- Relax PRIMARY KEYs on `claims` and `remittances` from single-column (id)
-- to composite (batch_id, id). Enables resubmits (same CLM01 / CLP01 in
-- different batches) and makes the pre-flight dedup workflow exercisable.
--
-- Strategy (mirrors 0013): table recreation with PRAGMA
-- defer_foreign_keys. We must recreate every table that has an FK pointing
-- at `claims(id)` or `remittances(id)` so that FK constraints can be
-- updated to point at the new composite PK.
--
-- FKs that need updating (verified via inspect(db.engine()) on the live
-- schema before this migration was written):
--   - claims.matched_remittance_id -> remittances(id)            becomes -> (batch_id, id)
--   - remittances.claim_id          -> claims(id)                 becomes -> (batch_id, id)
--   - matches.claim_id              -> claims(id) ON DELETE CASCADE   becomes -> (batch_id, id)
--   - matches.remittance_id         -> remittances(id) ON DELETE CASCADE  becomes -> (batch_id, id)
--   - cas_adjustments.remittance_id -> remittances(id) ON DELETE CASCADE  becomes -> (batch_id, id)
--   - line_reconciliations.remittance_id -> remittances(id) ON DELETE CASCADE  becomes -> (batch_id, id)
--   - line_reconciliations.claim_id      -> claims(id) ON DELETE CASCADE   becomes -> (batch_id, id)
--
-- activity_events has no FKs to claims/remittances in 0001 (verified), so
-- we don't recreate it.

PRAGMA defer_foreign_keys = ON;
PRAGMA foreign_keys = OFF;  -- required for table recreation in SQLite

-- Step 1: recreate `remittances` with composite PK
CREATE TABLE remittances_new (
    id TEXT NOT NULL,
    batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,
    payer_claim_control_number TEXT NOT NULL,
    claim_id TEXT,  -- FK to claims(batch_id, id) added after claims is recreated
    status_code TEXT NOT NULL,
    status_label TEXT,
    total_charge NUMERIC(12, 2) NOT NULL DEFAULT 0,
    total_paid NUMERIC(12, 2) NOT NULL DEFAULT 0,
    patient_responsibility NUMERIC(12, 2),
    adjustment_amount NUMERIC(12, 2) NOT NULL DEFAULT 0,
    received_at DATETIME NOT NULL,
    service_date DATE,
    is_reversal INTEGER NOT NULL DEFAULT 0,
    raw_json TEXT,
    PRIMARY KEY (batch_id, id)
);
INSERT INTO remittances_new
    SELECT id, batch_id, payer_claim_control_number, claim_id, status_code,
           status_label, total_charge, total_paid, patient_responsibility,
           adjustment_amount, received_at, service_date, is_reversal, raw_json
    FROM remittances;
DROP TABLE remittances;
ALTER TABLE remittances_new RENAME TO remittances;
CREATE INDEX ix_remittances_claim_id ON remittances(claim_id);
CREATE INDEX ix_remittances_payer_claim_control_number ON remittances(payer_claim_control_number);
CREATE INDEX ix_remittances_status_code ON remittances(status_code);

-- Step 2: recreate `claims` with composite PK
CREATE TABLE claims_new (
    id TEXT NOT NULL,
    batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,
    patient_control_number TEXT NOT NULL,
    service_date_from DATE,
    service_date_to DATE,
    charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0,
    provider_npi TEXT,
    payer_id TEXT,
    state TEXT NOT NULL DEFAULT 'submitted',
    state_before_reversal TEXT,
    matched_remittance_id TEXT,  -- FK added after remittances is recreated
    raw_json TEXT,
    PRIMARY KEY (batch_id, id)
);
INSERT INTO claims_new
    SELECT id, batch_id, patient_control_number, service_date_from,
           service_date_to, charge_amount, provider_npi, payer_id, state,
           state_before_reversal, matched_remittance_id, raw_json
    FROM claims;
DROP TABLE claims;
ALTER TABLE claims_new RENAME TO claims;
CREATE INDEX ix_claims_state ON claims(state);
CREATE INDEX ix_claims_patient_control_number ON claims(patient_control_number);
CREATE INDEX ix_claims_service_date_from ON claims(service_date_from);

-- Step 3: recreate `matches`, `cas_adjustments`, `line_reconciliations` so
-- their FKs point at the new composite PKs.

CREATE TABLE matches_new (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    claim_id TEXT NOT NULL,
    batch_id TEXT NOT NULL,  -- NEW: paired with claim_id for the composite FK
    remittance_id TEXT NOT NULL,
    remittance_batch_id TEXT NOT NULL,  -- NEW: paired with remittance_id
    strategy TEXT NOT NULL,
    matched_at DATETIME NOT NULL,
    prior_claim_state TEXT,
    is_reversal INTEGER NOT NULL DEFAULT 0,
    FOREIGN KEY (batch_id, claim_id) REFERENCES claims(batch_id, id) ON DELETE CASCADE,
    FOREIGN KEY (remittance_batch_id, remittance_id) REFERENCES remittances(batch_id, id) ON DELETE CASCADE
);
INSERT INTO matches_new
    SELECT m.id, m.claim_id, c.batch_id, m.remittance_id, r.batch_id,
           m.strategy, m.matched_at, m.prior_claim_state, m.is_reversal
    FROM matches m
    JOIN claims c ON c.id = m.claim_id
    JOIN remittances r ON r.id = m.remittance_id;
DROP TABLE matches;
ALTER TABLE matches_new RENAME TO matches;
CREATE INDEX ix_matches_claim_id ON matches(claim_id);
CREATE INDEX ix_matches_remittance_id ON matches(remittance_id);
CREATE INDEX ix_matches_matched_at ON matches(matched_at);

CREATE TABLE cas_adjustments_new (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    remittance_id TEXT NOT NULL,
    remittance_batch_id TEXT NOT NULL,
    group_code TEXT NOT NULL,
    reason_code TEXT NOT NULL,
    amount NUMERIC(12, 2) NOT NULL,
    quantity NUMERIC(10, 2),
    FOREIGN KEY (remittance_batch_id, remittance_id) REFERENCES remittances(batch_id, id) ON DELETE CASCADE
);
INSERT INTO cas_adjustments_new
    SELECT ca.id, ca.remittance_id, r.batch_id, ca.group_code, ca.reason_code,
           ca.amount, ca.quantity
    FROM cas_adjustments ca
    JOIN remittances r ON r.id = ca.remittance_id;
DROP TABLE cas_adjustments;
ALTER TABLE cas_adjustments_new RENAME TO cas_adjustments;
CREATE INDEX ix_cas_adjustments_remittance_id ON cas_adjustments(remittance_id);

-- line_reconciliations: see 0006. Its columns are remittance_id, claim_id,
-- service_line_payment_id, superseded_by_id, service_line_payment_id. We
-- need to add remittance_batch_id and batch_id columns for the new FKs.
-- (The full column list must match 0006.)
CREATE TABLE line_reconciliations_new (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    remittance_id VARCHAR(64) NOT NULL,
    remittance_batch_id VARCHAR(64) NOT NULL,
    claim_id VARCHAR(64) NOT NULL,
    batch_id VARCHAR(64) NOT NULL,
    service_line_payment_id INTEGER REFERENCES service_line_payments(id) ON DELETE SET NULL,
    superseded_by_id INTEGER REFERENCES service_line_payments(id) ON DELETE SET NULL,
    service_line_payment_id_new INTEGER REFERENCES service_line_payments(id) ON DELETE SET NULL,
    -- ... (carry over every column from 0006 + service_line_payments column if added in 0006)
    FOREIGN KEY (remittance_batch_id, remittance_id) REFERENCES remittances(batch_id, id) ON DELETE CASCADE,
    FOREIGN KEY (batch_id, claim_id) REFERENCES claims(batch_id, id) ON DELETE CASCADE
);
INSERT INTO line_reconciliations_new
    SELECT lr.id, lr.remittance_id, r.batch_id, lr.claim_id, c.batch_id,
           lr.service_line_payment_id, lr.superseded_by_id, lr.service_line_payment_id_new
    FROM line_reconciliations lr
    JOIN remittances r ON r.id = lr.remittance_id
    JOIN claims c ON c.id = lr.claim_id;
DROP TABLE line_reconciliations;
ALTER TABLE line_reconciliations_new RENAME TO line_reconciliations;

-- Step 4: add the back-references between claims and remittances now that
-- both tables exist with composite PKs.
-- (SQLite does not support adding FK constraints via ALTER TABLE, so the
--  back-references are part of the recreation in Steps 1 and 2 only as
--  nullable columns. Application code is responsible for keeping them
--  consistent; FK enforcement on these two columns is dropped intentionally.)

PRAGMA foreign_keys = ON;
PRAGMA defer_foreign_keys = OFF;
```

**Important:** before committing this migration, **read `0006_line_reconciliation.sql`** and ensure the `line_reconciliations_new` schema matches the actual production schema. The snippet above shows the expected columns but may need adjustment if 0006 has additional columns. Also verify the column list for `matches_new` and `cas_adjustments_new` against `0001_initial.sql`.

- [ ] **Step 5: Update `backend/src/cyclone/db.py` to reflect composite PK**

In `db.py`, the `Claim.id` and `Remittance.id` columns are mapped with `primary_key=True`. With composite PKs, the markers change:

- `Claim.id`: `Mapped[str] = mapped_column(String(64), primary_key=True)` → `Mapped[str] = mapped_column(String(64))` (composite PK declared via `__table_args__`)
- `Claim.batch_id`: `Mapped[str] = mapped_column(String(64), ForeignKey("batches.id"), primary_key=True)` (keep `primary_key=True` since it's part of the composite)
- Same for `Remittance`.

Add `__table_args__` to each class to declare `CompositePrimaryKey` constraints:

```python
from sqlalchemy import PrimaryKeyConstraint

class Claim(...):
    __table_args__ = (PrimaryKeyConstraint("batch_id", "id"),)
    # ...
```

SQLAlchemy 2.x syntax: use `mapped_column(..., primary_key=True)` for both columns and SQLAlchemy infers the composite. Or use `__table_args__ = (PrimaryKeyConstraint(...),)`.

- [ ] **Step 6: Run migration tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_db_migrate.py -k "0014" --no-header -q
```

Expected: 3 passed.

- [ ] **Step 7: Run the full `test_db_migrate.py`, expect all PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_db_migrate.py --no-header -q
```

Expected: all pass. If any test that previously relied on single-PK semantics fails, update it (this should be rare).

- [ ] **Step 8: Run the broader test suite to catch schema regressions**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest --no-header -q
```

Expected: all pass. Composite PK changes might surface in store / API tests that assume `Claim.id` is unique. Fix those tests by using `(batch_id, id)` lookups where needed. **Do not modify production code to make old tests pass**; update the tests.

- [ ] **Step 9: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/src/cyclone/migrations/0014_relax_claims_remits_pk.sql \
        backend/src/cyclone/db.py \
        backend/tests/test_db_migrate.py
git commit -m "feat(db): migration 0014 relaxes claims/remittances PK to (batch_id, id)"
```

---

### Task 1.4: Tighten `find_existing_batch_for_claim` to return the most-recent batch

The current implementation returns *some* batch (no `ORDER BY`). The spec says we must return the most-recent. Update the helper.

**Files:**
- Modify: `backend/src/cyclone/store.py` (`find_existing_batch_for_claim`, `find_existing_batch_for_remit`)
- Modify: `backend/tests/test_store.py` (add a "most recent" assertion)

- [ ] **Step 1: Write the failing test**

Add to `backend/tests/test_store.py`:

```python
def test_find_existing_batch_for_claim_returns_most_recent():
    """When the same claim_id is in multiple prior batches, the helper
    returns the most-recent (highest state_changed_at) batch."""
    from datetime import datetime, timedelta, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
    with _db.SessionLocal()() as s:
        # Older batch with the claim
        s.add(Batch(
            id="B_OLD", kind="837p", input_filename="old.txt",
            parsed_at=base, raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-A", batch_id="B_OLD", patient_control_number="M",
            state_changed_at=base,
        ))
        # Newer batch with the same claim
        s.add(Batch(
            id="B_NEW", kind="837p", input_filename="new.txt",
            parsed_at=base + timedelta(days=1), raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-A", batch_id="B_NEW", patient_control_number="M",
            state_changed_at=base + timedelta(days=1),
        ))
        s.commit()

    from cyclone import store as store_mod
    assert store_mod.find_existing_batch_for_claim("CLM-A") == "B_NEW"


def test_find_existing_batch_for_remit_returns_most_recent():
    """Same shape for 835: most-recent (highest received_at) batch wins."""
    from datetime import datetime, timedelta, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Remittance

    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="B_OLD835", kind="835", input_filename="old.txt",
            parsed_at=base, raw_result_json={},
        ))
        s.add(Remittance(
            id="CLP-A", batch_id="B_OLD835",
            payer_claim_control_number="CLP-A",
            status_code="1",
            received_at=base,
        ))
        s.add(Batch(
            id="B_NEW835", kind="835", input_filename="new.txt",
            parsed_at=base + timedelta(days=1), raw_result_json={},
        ))
        s.add(Remittance(
            id="CLP-A", batch_id="B_NEW835",
            payer_claim_control_number="CLP-A",
            status_code="1",
            received_at=base + timedelta(days=1),
        ))
        s.commit()

    from cyclone import store as store_mod
    assert store_mod.find_existing_batch_for_remit("CLP-A") == "B_NEW835"
```

- [ ] **Step 2: Run the new tests, expect FAIL**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest \
  tests/test_store.py::test_find_existing_batch_for_claim_returns_most_recent \
  tests/test_store.py::test_find_existing_batch_for_remit_returns_most_recent \
  --no-header -q
```

Expected: FAIL — the helper does not have `ORDER BY ... DESC LIMIT 1` so it returns `B_OLD` (or `B_OLD835`) instead of the most-recent.

- [ ] **Step 3: Update the helpers**

In `backend/src/cyclone/store.py`, replace `find_existing_batch_for_claim`:

```python
def find_existing_batch_for_claim(claim_id: str) -> str | None:
    """Return the batch_id of the most-recent batch containing this CLM01, or None.

    "Most-recent" is by ``Claim.state_changed_at`` (latest touch of the claim
    row, including resubmits). When the same CLM01 exists in N prior batches,
    the newest one wins. Returns ``None`` if no batch contains the claim.

    Pure read; opens a short-lived session. Used by the 837 409 handler to
    surface which prior batch already holds the same CLM01.
    """
    from sqlalchemy import select
    from cyclone.db import Claim
    with db.SessionLocal()() as s:
        row = s.execute(
            select(Claim.batch_id)
            .where(Claim.id == claim_id)
            .order_by(Claim.state_changed_at.desc())
            .limit(1)
        ).first()
        return row[0] if row else None
```

And replace `find_existing_batch_for_remit`:

```python
def find_existing_batch_for_remit(remit_id: str) -> str | None:
    """Return the batch_id of the most-recent batch containing this CLP01, or None.

    "Most-recent" is by ``Remittance.received_at``. ``remit_id`` is the PK on
    ``remittances.id`` (= payer_claim_control_number = CLP01).
    """
    from sqlalchemy import select
    from cyclone.db import Remittance
    with db.SessionLocal()() as s:
        row = s.execute(
            select(Remittance.batch_id)
            .where(Remittance.id == remit_id)
            .order_by(Remittance.received_at.desc())
            .limit(1)
        ).first()
        return row[0] if row else None
```

- [ ] **Step 4: Run all `test_store.py` tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_store.py --no-header -q
```

Expected: 12 passed (10 existing + 2 new).

- [ ] **Step 5: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/src/cyclone/store.py backend/tests/test_store.py
git commit -m "feat(store): find_existing_batch_for_claim/remit return most-recent"
```

---

## Phase 2 — Backend pre-flight dedup module + endpoint changes

### Task 2.1: New `dedup.py` module with `CollisionReport` and pre-flight helpers

**Files:**
- Create: `backend/src/cyclone/dedup.py`
- Create: `backend/tests/test_dedup.py`

- [ ] **Step 1: Write the failing tests**

Create `backend/tests/test_dedup.py`:

```python
"""Tests for the pre-flight dedup helpers in cyclone.dedup.

The dedup module is the source of truth for whether a parsed batch would
collide with existing data. It returns a ``CollisionReport`` that the API
serializes into the 409 body.
"""
from __future__ import annotations

from datetime import date, datetime, timezone
from decimal import Decimal

import pytest

from cyclone import db
from cyclone.dedup import (
    CollisionReport,
    preflight_837,
    preflight_835,
)
from cyclone.parsers.models import (
    Address,
    BillingProvider,
    ClaimHeader,
    ClaimOutput,
    Envelope,
    Payer,
    ParseResult,
    Subscriber,
    ValidationReport,
)
from cyclone.parsers.models_835 import (
    ClaimPayment,
    ParseResult835,
)


@pytest.fixture(autouse=True)
def _setup(tmp_path, monkeypatch):
    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
    db._reset_for_tests()
    db.init_db()
    yield
    db._reset_for_tests()


def _claim(claim_id: str, member_id: str = "M1") -> ClaimOutput:
    return ClaimOutput(
        claim_id=claim_id,
        control_number="0001",
        transaction_date=date(2026, 6, 19),
        billing_provider=BillingProvider(name="Test", npi="1234567890"),
        subscriber=Subscriber(
            first_name="X", last_name="Y", member_id=member_id, dob="1970-01-01", gender="M",
        ),
        payer=Payer(name="Test", id="TST"),
        claim=ClaimHeader(claim_id=claim_id, total_charge=Decimal("100")),
        diagnoses=[],
        service_lines=[],
        validation=ValidationReport(passed=True, errors=[], warnings=[]),
        raw_segments=[],
    )


def _result(*claims: ClaimOutput) -> ParseResult:
    return ParseResult(
        envelope=Envelope(
            sender_id="S", receiver_id="R", control_number="1",
            transaction_date=date(2026, 6, 19), transaction_time="1200",
            implementation_guide="005010X222A1", transaction_type_code="CH",
        ),
        claims=list(claims),
        summary={
            "input_file": "x.txt", "control_number": "1",
            "transaction_date": "2026-06-19", "total_claims": len(claims),
            "passed": len(claims), "failed": 0, "failed_claim_ids": [],
            "issues_by_rule": {}, "output_dir": None,
        },
    )


def test_preflight_837_no_collisions_on_empty_db():
    """Empty DB → no collisions, no existing_batch_id."""
    r = _result(_claim("A"), _claim("B"))
    report = preflight_837(r)
    assert report.colliding_claim_ids == []
    assert report.existing_batch_id is None
    assert report.within_file_duplicate_ids == []
    assert report.total_claims == 2


def test_preflight_837_finds_cross_batch_collision():
    """A claim already in the DB surfaces as a collision with the prior batch."""
    from cyclone.db import Batch, Claim
    with db.SessionLocal()() as s:
        s.add(Batch(
            id="PRIOR", kind="837p", input_filename="prior.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(
            id="A", batch_id="PRIOR", patient_control_number="M",
            state_changed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        ))
        s.commit()

    r = _result(_claim("A"), _claim("B"))  # A collides, B is new
    report = preflight_837(r)
    assert report.colliding_claim_ids == ["A"]
    assert report.existing_batch_id == "PRIOR"
    assert report.within_file_duplicate_ids == []
    assert report.total_claims == 2


def test_preflight_837_finds_within_file_duplicate():
    """Same CLM01 twice in the file → within_file_duplicate_ids and no batch_id."""
    r = _result(_claim("A"), _claim("A"), _claim("B"))
    report = preflight_837(r)
    assert report.within_file_duplicate_ids == ["A"]
    assert report.colliding_claim_ids == ["A"]  # also flagged as a collision
    assert report.existing_batch_id is None
    assert report.total_claims == 3


def test_preflight_837_returns_most_recent_batch_id():
    """When CLM01=A is in 3 prior batches, return the most-recent one."""
    from datetime import timedelta
    from cyclone.db import Batch, Claim

    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
    for i, bid in enumerate(["B1", "B2", "B3"]):
        with db.SessionLocal()() as s:
            s.add(Batch(
                id=bid, kind="837p", input_filename=f"{bid}.txt",
                parsed_at=base, raw_result_json={},
            ))
            s.add(Claim(
                id="A", batch_id=bid, patient_control_number="M",
                state_changed_at=base + timedelta(days=i),
            ))
            s.commit()

    r = _result(_claim("A"))
    report = preflight_837(r)
    assert report.colliding_claim_ids == ["A"]
    assert report.existing_batch_id == "B3"  # most-recent by state_changed_at


def test_preflight_835_mirrors_837():
    """Same shape for 835: payer_claim_control_number is the dedup key."""
    from cyclone.db import Batch, Remittance
    with db.SessionLocal()() as s:
        s.add(Batch(
            id="PRIOR835", kind="835", input_filename="p.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Remittance(
            id="CLP-A", batch_id="PRIOR835",
            payer_claim_control_number="CLP-A",
            status_code="1",
            received_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        ))
        s.commit()

    r835 = ParseResult835(
        envelope=Envelope(
            sender_id="S", receiver_id="R", control_number="1",
            transaction_date=date(2026, 6, 19), transaction_time="1200",
            implementation_guide="005010X221A1", transaction_type_code="CH",
        ),
        payer=Payer(name="Test", id="TST"),
        claims=[
            ClaimPayment(
                payer_claim_control_number="CLP-A",
                status_code="1",
                total_charge=Decimal("100"), total_paid=Decimal("100"),
            ),
            ClaimPayment(
                payer_claim_control_number="CLP-B",
                status_code="1",
                total_charge=Decimal("50"), total_paid=Decimal("50"),
            ),
        ],
    )
    report = preflight_835(r835)
    assert report.colliding_claim_ids == ["CLP-A"]
    assert report.existing_batch_id == "PRIOR835"
    assert report.within_file_duplicate_ids == []
    assert report.total_claims == 2
```

- [ ] **Step 2: Run tests, expect FAIL with "No module named 'cyclone.dedup'"**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_dedup.py --no-header -q
```

Expected: `ModuleNotFoundError: No module named 'cyclone.dedup'`.

- [ ] **Step 3: Implement `dedup.py`**

Create `backend/src/cyclone/dedup.py`:

```python
"""Pre-flight dedup for parsed 837P/835 batches.

Splits the parsed result into "would-insert" and "would-skip" sets by
querying the DB for any ``claim_id`` (837) or ``payer_claim_control_number``
(835) already present. Also detects within-file duplicates by counting
``claim_id`` / ``payer_claim_control_number`` frequencies.

Used by the ``parse-837`` and ``parse-835`` endpoints between validation
and persist, so the user sees the parse result + collision summary
before any DB write.
"""
from __future__ import annotations

from collections import Counter
from dataclasses import dataclass, field
from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.orm import Session

from cyclone import db

if TYPE_CHECKING:
    from cyclone.parsers.models import ParseResult
    from cyclone.parsers.models_835 import ParseResult835


@dataclass(frozen=True)
class CollisionReport:
    """What the parse endpoint needs to render a 409 body.

    ``within_file_duplicate_ids`` is a subset of ``colliding_claim_ids``:
    the same CLM01 appearing twice in the file is also flagged as a
    collision (because the second instance cannot be inserted).

    ``existing_batch_id`` is the most-recent prior batch that contains
    one of the colliding claim_ids, or None when the collision is purely
    within-file.
    """
    colliding_claim_ids: list[str] = field(default_factory=list)
    existing_batch_id: str | None = None
    within_file_duplicate_ids: list[str] = field(default_factory=list)
    total_claims: int = 0

    @property
    def has_collisions(self) -> bool:
        return bool(self.colliding_claim_ids or self.within_file_duplicate_ids)

    @property
    def new_claims_after_skip(self) -> int:
        """How many claims WOULD be inserted on a force-insert."""
        # A within-file duplicate still occupies one of the ``total_claims`` slots;
        # the force-insert keeps the first occurrence and skips the rest, so
        # ``new_claims_after_skip`` = total_claims - within_file_dup_count.
        return self.total_claims - len(self.within_file_duplicate_ids)


def preflight_837(result: "ParseResult", session: Session | None = None) -> CollisionReport:
    """Detect 837 collisions: within-file dupes + cross-batch CLM01 dupes."""
    claim_ids = [c.claim_id for c in result.claims]
    counts = Counter(claim_ids)
    within_file_duplicate_ids = sorted(
        cid for cid, n in counts.items() if n > 1
    )
    seen: set[str] = set(claim_ids)

    if not seen:
        return CollisionReport(
            colliding_claim_ids=[],
            existing_batch_id=None,
            within_file_duplicate_ids=[],
            total_claims=0,
        )

    own_session = session is None
    if own_session:
        session = db.SessionLocal()()
    try:
        from cyclone.db import Claim
        rows = session.execute(
            select(Claim.id, Claim.batch_id)
            .where(Claim.id.in_(seen))
            .order_by(Claim.state_changed_at.desc())
        ).all()
    finally:
        if own_session:
            session.close()

    db_collisions: dict[str, str] = {cid: bid for cid, bid in rows}
    # A claim_id "collides" if it's in the DB (cross-batch) OR appears more than
    # once in this file (within-file). The UI uses colliding_claim_ids to count
    # the problem set; within_file_duplicate_ids is the subset that is purely
    # within-file. existing_batch_id stays None when every collision is within-file.
    colliding = sorted(
        cid for cid, n in counts.items()
        if cid in db_collisions or n > 1
    )
    existing_batch_id = next(iter(db_collisions.values()), None) if db_collisions else None
    return CollisionReport(
        colliding_claim_ids=colliding,
        existing_batch_id=existing_batch_id,
        within_file_duplicate_ids=within_file_duplicate_ids,
        total_claims=len(claim_ids),
    )


def preflight_835(result: "ParseResult835", session: Session | None = None) -> CollisionReport:
    """Same shape for 835 remittances. Payer claim control number = CLP01 = remittance.id."""
    pcns = [c.payer_claim_control_number for c in result.claims]
    counts = Counter(pcns)
    within_file_duplicate_ids = sorted(p for p, n in counts.items() if n > 1)
    seen: set[str] = set(pcns)

    if not seen:
        return CollisionReport(
            colliding_claim_ids=[],
            existing_batch_id=None,
            within_file_duplicate_ids=[],
            total_claims=0,
        )

    own_session = session is None
    if own_session:
        session = db.SessionLocal()()
    try:
        from cyclone.db import Remittance
        rows = session.execute(
            select(Remittance.id, Remittance.batch_id)
            .where(Remittance.id.in_(seen))
            .order_by(Remittance.received_at.desc())
        ).all()
    finally:
        if own_session:
            session.close()

    db_collisions: dict[str, str] = {pcn: bid for pcn, bid in rows}
    # Same rule as 837: within-file dupes are also collisions.
    colliding = sorted(
        pcn for pcn, n in counts.items()
        if pcn in db_collisions or n > 1
    )
    existing_batch_id = next(iter(db_collisions.values()), None) if db_collisions else None
    return CollisionReport(
        colliding_claim_ids=colliding,
        existing_batch_id=existing_batch_id,
        within_file_duplicate_ids=within_file_duplicate_ids,
        total_claims=len(pcns),
    )
```

- [ ] **Step 4: Run tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_dedup.py --no-header -q
```

Expected: 5 passed.

- [ ] **Step 5: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/src/cyclone/dedup.py backend/tests/test_dedup.py
git commit -m "feat(dedup): preflight_837 / preflight_835 with CollisionReport"
```

---

### Task 2.2: Store helper `delete_batch` with cascade + 409 guard

The `DELETE /api/batches/{id}` endpoint (Task 2.4) calls a store method
that hard-deletes a batch and cascades through all child rows. The
method refuses if any claim is past `submitted` state.

**Files:**
- Modify: `backend/src/cyclone/store.py` (add `delete_batch` method on `CycloneStore`)
- Modify: `backend/tests/test_store.py` (add tests)

- [ ] **Step 1: Write the failing tests**

Add to `backend/tests/test_store.py`:

```python
def test_delete_batch_cascades_to_claims():
    """delete_batch removes the batch row and all its claim rows."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="B1", kind="837p", input_filename="x.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(id="CLM-1", batch_id="B1", patient_control_number="M1"))
        s.add(Claim(id="CLM-2", batch_id="B1", patient_control_number="M2"))
        s.commit()

    s = CycloneStore()
    result = s.delete_batch("B1")
    assert result is True

    with _db.SessionLocal()() as sess:
        assert sess.get(_db.Batch, "B1") is None
        rows = sess.execute(_db.text("SELECT id FROM claims WHERE batch_id='B1'")).all()
        assert rows == []


def test_delete_batch_returns_false_for_unknown():
    """Unknown batch_id -> returns False (not raises)."""
    s = CycloneStore()
    assert s.delete_batch("does-not-exist") is False


def test_delete_batch_raises_on_reconciled_claims():
    """Refuses to delete if any claim is past 'submitted' state."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="B2", kind="837p", input_filename="x.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-X", batch_id="B2", patient_control_number="M",
            state="paid",  # reconciled
        ))
        s.commit()

    s = CycloneStore()
    with pytest.raises(_InvalidStateError):  # reuse the existing 409 error type
        s.delete_batch("B2")
```

Wait — `InvalidStateError` is the wrong type for this case (it's for
`apply_*` skips, not for delete-blocked). We need a new exception
class. Update the test to use a new `BatchNotDeletableError`:

Adjust the test imports at the top of the file:

```python
from cyclone.store import (
    AlreadyMatchedError,
    BatchNotDeletableError,
    CycloneStore,
    store,
)
```

And add `BatchNotDeletableError` to the import. Then update the third
test:

```python
def test_delete_batch_raises_batch_not_deletable_on_reconciled_claims():
    """Refuses to delete if any claim is past 'submitted' state."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="B2", kind="837p", input_filename="x.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-X", batch_id="B2", patient_control_number="M",
            state="paid",  # reconciled
        ))
        s.commit()

    s = CycloneStore()
    with pytest.raises(BatchNotDeletableError) as excinfo:
        s.delete_batch("B2")
    assert "non-submitted" in str(excinfo.value).lower()
    assert "B2" in str(excinfo.value)

    # The batch and its claim are still in the DB.
    with _db.SessionLocal()() as sess:
        assert sess.get(_db.Batch, "B2") is not None
        rows = sess.execute(_db.text("SELECT id FROM claims WHERE batch_id='B2'")).all()
        assert len(rows) == 1
```

- [ ] **Step 2: Run tests, expect FAIL with "no attribute 'delete_batch'" and "no name 'BatchNotDeletableError'"**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_store.py -k "delete_batch" --no-header -q
```

Expected: 3 failures.

- [ ] **Step 3: Implement `BatchNotDeletableError` and `delete_batch`**

In `backend/src/cyclone/store.py`, add the exception class near the
other exception classes (right after `InvalidStateError`):

```python
class BatchNotDeletableError(Exception):
    """Raised by ``CycloneStore.delete_batch`` when the batch has claims in
    a non-``submitted`` state (i.e., it has been reconciled or otherwise
    progressed past raw ingestion).

    The T15 API endpoint maps this to a 409 Conflict; the user must first
    unreconcile the affected claims before deleting the batch.
    """

    def __init__(self, batch_id: str, offending_claim_id: str):
        self.batch_id = batch_id
        self.offending_claim_id = offending_claim_id
        super().__init__(
            f"Batch {batch_id} cannot be deleted: claim {offending_claim_id} "
            f"is in a non-submitted state."
        )
```

Find the `CycloneStore` class (search for `class CycloneStore`). Add
the `delete_batch` method as the last method in the class. Use this
code (the method goes inside `class CycloneStore`):

```python
    def delete_batch(self, batch_id: str) -> bool:
        """Hard-delete a batch and cascade through child rows.

        Returns ``True`` if the batch was found and deleted, ``False`` if
        no batch with that id exists. Raises :class:`BatchNotDeletableError`
        if any claim in the batch is past the ``submitted`` state (i.e.,
        has been reconciled, reversed, denied, or otherwise progressed);
        the user must unreconcile those claims first.

        The cascade relies on the ``ON DELETE CASCADE`` FKs declared in
        the schema migrations: ``claims.batch_id``, ``remittances.batch_id``,
        ``activity_events.batch_id`` (when present), ``matches.claim_id``
        (via claims), ``line_reconciliations.claim_id`` (via claims), and
        the remittance-side children. The store explicitly records a
        ``batch_deleted`` activity event before the cascade so the audit
        log retains a tombstone of the deletion.

        The 409 guard runs in a single short-lived session; the
        ``batch_deleted`` event and the cascade are committed atomically.
        """
        from datetime import datetime as _dt, timezone as _tz

        with db.SessionLocal()() as s:
            batch_row = s.get(Batch, batch_id)
            if batch_row is None:
                return False
            # 409 guard: refuse if any claim has progressed past 'submitted'.
            offending = s.execute(
                select(Claim.id)
                .where(Claim.batch_id == batch_id)
                .where(Claim.state != "submitted")
                .limit(1)
            ).first()
            if offending is not None:
                raise BatchNotDeletableError(batch_id, offending[0])
            # Tombstone activity event for the audit log.
            s.add(ActivityEvent(
                ts=_dt.now(_tz.utc),
                kind="batch_deleted",
                batch_id=batch_id,
                payload_json={
                    "message": f"Batch {batch_id} deleted",
                    "kind": batch_row.kind,
                    "input_filename": batch_row.input_filename,
                },
            ))
            s.flush()
            s.delete(batch_row)
            s.commit()
        return True
```

Add the `delete_batch` import near the top of the file (right after
`AlreadyMatchedError`):

```python
class BatchNotDeletableError(Exception):
    ...
```

(The exception class goes in the module body, not inside
`CycloneStore`. The class definition is unchanged from what was added
in Step 3 above.)

The new `select` import in the method body is already available because
`store.py` imports `from sqlalchemy import ...` later in the file
(grep to confirm). If not, add `from sqlalchemy import select` to the
top of the file.

- [ ] **Step 4: Run the new tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_store.py -k "delete_batch" --no-header -q
```

Expected: 3 passed.

- [ ] **Step 5: Run the full test_store.py, expect all PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_store.py --no-header -q
```

Expected: 15 passed (12 prior + 3 new).

- [ ] **Step 6: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/src/cyclone/store.py backend/tests/test_store.py
git commit -m "feat(store): delete_batch hard-deletes with cascade + non-submitted guard"
```

---

### Task 2.3: 837 endpoint pre-flight, `?force=true`, race handler, 200 `skipped_claim_ids`

This is the biggest task. The 837 endpoint needs:
1. `?force: bool = Query(False)` parameter
2. Pre-flight dedup between validation and persist (when not force)
3. New 409 body shape: full parse result + collisions + existing_batch_id
4. Race-condition 409 (same shape) when IntegrityError fires after a clean pre-flight
5. 200 body adds `skipped_claim_ids` when force=true (so the UI can show what was skipped)

The 409 body builder is shared between pre-flight and race, so it lives
as a module-level helper.

**Files:**
- Modify: `backend/src/cyclone/api.py`
- Modify: `backend/tests/test_api_parse_persists.py`

- [ ] **Step 1: Write the failing tests**

Add to `backend/tests/test_api_parse_persists.py`:

```python
def test_parse_837_409_includes_parse_result_and_collisions(client: TestClient) -> None:
    """Pre-flight dedup returns 409 with the full parse result + collisions."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    # Pre-seed a claim in a prior batch.
    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="PRIOR", kind="837p", input_filename="prior.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-A", batch_id="PRIOR", patient_control_number="M",
            state_changed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        ))
        s.commit()

    # Build a small 837 file with one CLM* that collides.
    text = (
        "ISA*00*          *00*          *ZZ*SUBMITTERID    *ZZ*RECEIVERID     "
        "*240101*1200*^*00501*000000001*0*P*:~\n"
        "GS*HC*SUBMITTERID*RECEIVERID*20240101*1200*1*X*005010X222A1~\n"
        "ST*837*0001*005010X222A1~\n"
        "BHT*0019*00*1*20240101*1200*CH~\n"
        "NM1*41*2*SUBMITTER*****46*SUBMITTERID~\n"
        "PER*IC*CONTACT*TE*5555555555~\n"
        "NM1*40*2*RECEIVER*****46*RECEIVERID~\n"
        "HL*1**20*1~\n"
        "NM1*85*2*BILLING*****XX*1881068062~\n"
        "N3*123 MAIN*~~\nN4*DENVER*CO*80202~\n"
        "REF*EI*123456789~\n"
        "HL*2*1*22*0~\n"
        "SBR*P*18*******CI~\n"
        "NM1*IL*1*DOE*JOHN****MI*M~\n"
        "N3*456 ELM*~~\nN4*DENVER*CO*80202~\n"
        "DMG*D8*19700101*M~\n"
        "NM1*PR*2*MEDICAID*****PI*SKCO0~\n"
        "CLM*CLM-A*100***11:B:1*Y*A*Y*Y~\n"
        "HI*ABK:R69~\n"
        "LX*1~\nSV1*HC:99213*100*UN*1***1~\n"
        "DTP*472*D8*20240101~\n"
        "SE*24*0001~\nGE*1*1~\nIEA*1*000000001~\n"
    )

    resp = client.post(
        "/api/parse-837",
        files={"file": ("dup.txt", text, "text/plain")},
        headers={"Accept": "application/json"},
    )
    assert resp.status_code == 409, resp.text
    body = resp.json()
    assert body["error"] == "Duplicate claim"
    assert body["existing_batch_id"] == "PRIOR"
    assert body["batch_id"] is None
    assert body["collisions"]["colliding_claim_ids"] == ["CLM-A"]
    assert body["collisions"]["total_claims"] == 1
    assert body["collisions"]["total_collisions"] == 1
    assert body["collisions"]["new_claims_after_skip"] == 0
    assert "parse_result" in body
    assert body["parse_result"]["claims"][0]["claim_id"] == "CLM-A"
    # No batch was actually persisted.
    with _db.SessionLocal()() as s:
        n = s.execute(_db.text("SELECT COUNT(*) FROM batches WHERE id != 'PRIOR'")).scalar()
    assert n == 0


def test_parse_837_within_file_duplicate_returns_409_with_null_batch_id(
    client: TestClient,
) -> None:
    """A file with the same CLM01 twice returns 409 with existing_batch_id=None."""
    text = (
        "ISA*00*          *00*          *ZZ*SUBMITTERID    *ZZ*RECEIVERID     "
        "*240101*1200*^*00501*000000001*0*P*:~\n"
        "GS*HC*SUBMITTERID*RECEIVERID*20240101*1200*1*X*005010X222A1~\n"
        "ST*837*0001*005010X222A1~\n"
        "BHT*0019*00*1*20240101*1200*CH~\n"
        "NM1*41*2*SUBMITTER*****46*SUBMITTERID~\n"
        "PER*IC*CONTACT*TE*5555555555~\n"
        "NM1*40*2*RECEIVER*****46*RECEIVERID~\n"
        "HL*1**20*1~\n"
        "NM1*85*2*BILLING*****XX*1881068062~\n"
        "N3*123 MAIN*~~\nN4*DENVER*CO*80202~\n"
        "REF*EI*123456789~\n"
        "HL*2*1*22*0~\n"
        "SBR*P*18*******CI~\n"
        "NM1*IL*1*DOE*JOHN****MI*M~\n"
        "N3*456 ELM*~~\nN4*DENVER*CO*80202~\n"
        "DMG*D8*19700101*M~\n"
        "NM1*PR*2*MEDICAID*****PI*SKCO0~\n"
        "CLM*CLM-DUP*100***11:B:1*Y*A*Y*Y~\n"
        "HI*ABK:R69~\n"
        "LX*1~\nSV1*HC:99213*100*UN*1***1~\n"
        "DTP*472*D8*20240101~\n"
        "CLM*CLM-DUP*100***11:B:1*Y*A*Y*Y~\n"
        "HI*ABK:R69~\n"
        "LX*1~\nSV1*HC:99213*100*UN*1***1~\n"
        "DTP*472*D8*20240101~\n"
        "SE*28*0001~\nGE*1*1~\nIEA*1*000000001~\n"
    )

    resp = client.post(
        "/api/parse-837",
        files={"file": ("within.txt", text, "text/plain")},
        headers={"Accept": "application/json"},
    )
    assert resp.status_code == 409, resp.text
    body = resp.json()
    assert body["existing_batch_id"] is None
    assert "CLM-DUP" in body["collisions"]["colliding_claim_ids"]
    assert "CLM-DUP" in body["collisions"].get("within_file_duplicate_ids", body["collisions"]["colliding_claim_ids"])


def test_parse_837_force_true_persists_non_colliding_claims(client: TestClient) -> None:
    """With force=true, the collision is skipped and other claims persist."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="PRIOR", kind="837p", input_filename="p.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-OLD", batch_id="PRIOR", patient_control_number="M",
            state_changed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        ))
        s.commit()

    # File with 2 claims: one colliding, one new.
    text = (
        "ISA*00*          *00*          *ZZ*SUBMITTERID    *ZZ*RECEIVERID     "
        "*240101*1200*^*00501*000000001*0*P*:~\n"
        "GS*HC*SUBMITTERID*RECEIVERID*20240101*1200*1*X*005010X222A1~\n"
        "ST*837*0001*005010X222A1~\n"
        "BHT*0019*00*1*20240101*1200*CH~\n"
        "NM1*41*2*SUBMITTER*****46*SUBMITTERID~\n"
        "PER*IC*CONTACT*TE*5555555555~\n"
        "NM1*40*2*RECEIVER*****46*RECEIVERID~\n"
        "HL*1**20*1~\n"
        "NM1*85*2*BILLING*****XX*1881068062~\n"
        "N3*123 MAIN*~~\nN4*DENVER*CO*80202~\n"
        "REF*EI*123456789~\n"
        "HL*2*1*22*0~\n"
        "SBR*P*18*******CI~\n"
        "NM1*IL*1*DOE*JOHN****MI*M~\n"
        "N3*456 ELM*~~\nN4*DENVER*CO*80202~\n"
        "DMG*D8*19700101*M~\n"
        "NM1*PR*2*MEDICAID*****PI*SKCO0~\n"
        "CLM*CLM-OLD*100***11:B:1*Y*A*Y*Y~\n"
        "HI*ABK:R69~\n"
        "LX*1~\nSV1*HC:99213*100*UN*1***1~\n"
        "DTP*472*D8*20240101~\n"
        "CLM*CLM-NEW*100***11:B:1*Y*A*Y*Y~\n"
        "HI*ABK:R69~\n"
        "LX*1~\nSV1*HC:99213*100*UN*1***1~\n"
        "DTP*472*D8*20240101~\n"
        "SE*30*0001~\nGE*1*1~\nIEA*1*000000001~\n"
    )

    resp = client.post(
        "/api/parse-837?force=true",
        files={"file": ("force.txt", text, "text/plain")},
        headers={"Accept": "application/json"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body.get("skipped_claim_ids") == ["CLM-OLD"]
    # The new claim persisted; the old one did not.
    with _db.SessionLocal()() as s:
        rows = s.execute(_db.text("SELECT id, batch_id FROM claims WHERE id IN ('CLM-OLD','CLM-NEW') ORDER BY id")).all()
    by_id = {r[0]: r[1] for r in rows}
    assert by_id["CLM-OLD"] == "PRIOR"  # not re-inserted
    assert by_id["CLM-NEW"] != "PRIOR"  # new batch
```

Also add the DELETE endpoint test (this goes in `test_api_parse_persists.py`
because there's no dedicated test file for it yet — `store.py` already
has the unit tests in Task 2.2):

```python
def test_delete_batch_endpoint_cascades(client: TestClient) -> None:
    """DELETE /api/batches/{id} removes the batch and its claims."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="B1", kind="837p", input_filename="x.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(id="CLM-1", batch_id="B1", patient_control_number="M1"))
        s.commit()

    resp = client.delete("/api/batches/B1")
    assert resp.status_code in (200, 204), resp.text

    with _db.SessionLocal()() as s:
        assert s.get(_db.Batch, "B1") is None
        n = s.execute(_db.text("SELECT COUNT(*) FROM claims WHERE batch_id='B1'")).scalar()
    assert n == 0


def test_delete_batch_endpoint_404_on_unknown(client: TestClient) -> None:
    resp = client.delete("/api/batches/does-not-exist")
    assert resp.status_code == 404


def test_delete_batch_endpoint_409_on_reconciled_claim(client: TestClient) -> None:
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Claim

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="B2", kind="837p", input_filename="x.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Claim(
            id="CLM-X", batch_id="B2", patient_control_number="M",
            state="paid",
        ))
        s.commit()

    resp = client.delete("/api/batches/B2")
    assert resp.status_code == 409
    with _db.SessionLocal()() as s:
        assert s.get(_db.Batch, "B2") is not None
```

- [ ] **Step 2: Run the new tests, expect FAIL**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest \
  tests/test_api_parse_persists.py -k "409 or force or delete_batch_endpoint" \
  --no-header -q
```

Expected: 6 failures (4 for the new 409/force cases; the existing
`test_409_response_includes_existing_batch_id_for_837` from the prior
plan also fails because the body shape changed; 2 for DELETE not yet
existing).

- [ ] **Step 3: Implement the new 837 endpoint body and DELETE endpoint**

In `backend/src/cyclone/api.py`:

1. **Add the `force` parameter to the `parse_837` signature:**

Replace the existing signature:

```python
@app.post("/api/parse-837")
async def parse_837(
    request: Request,
    file: UploadFile = File(...),
    payer: str = Query("co_medicaid"),
    include_raw_segments: bool = Query(True),
    strict: bool = Query(False),
    ack: bool = Query(False),
) -> Any:
```

with:

```python
@app.post("/api/parse-837")
async def parse_837(
    request: Request,
    file: UploadFile = File(...),
    payer: str = Query("co_medicaid"),
    include_raw_segments: bool = Query(True),
    strict: bool = Query(False),
    ack: bool = Query(False),
    force: bool = Query(False),
) -> Any:
```

2. **Add the 409 builder and modify the 837 path:**

Find the existing `try: store.add(...) except IntegrityError as exc:`
block in `parse_837` (around line 392-415). Replace the entire block
(after the `if _has_claim_validation_errors` check) with:

```python
    # Pre-flight dedup: detect within-file dupes and cross-batch CLM01 collisions
    # BEFORE persisting. The user sees the parse result + collision summary in
    # the 409 body and can choose force-insert / delete prior / pick a different
    # file. force=true bypasses the check; the store.add dedup still silently
    # skips colliding rows, and the 200 body includes skipped_claim_ids.
    if not force and result.claims:
        from cyclone import dedup
        report = dedup.preflight_837(result)
        if report.has_collisions:
            return _build_dedup_409(
                result=result,
                report=report,
                kind="cross_batch" if report.existing_batch_id else "within_file",
                error="Duplicate claim",
            )

    rec = BatchRecord(
        id=uuid.uuid4().hex,
        kind="837p",
        input_filename=file.filename or "upload.txt",
        parsed_at=utcnow(),
        result=result,
    )
    try:
        store.add(rec, event_bus=request.app.state.event_bus)
    except IntegrityError as exc:
        # Race: pre-flight said clean, but a concurrent process ingested a
        # colliding batch between the check and the persist. Re-run pre-flight
        # to build the same 409 body shape.
        from cyclone import dedup
        report = dedup.preflight_837(result)
        log.warning("Race on persist for batch %s: %s", rec.id, exc)
        return _build_dedup_409(
            result=result,
            report=report,
            kind="race",
            error="Duplicate claim (race condition)",
        )

    if _client_wants_json(request):
        body = json.loads(result.model_dump_json())
        if ack:
            ack_body = _build_and_persist_ack(rec.id)
            if ack_body is not None:
                body["ack"] = ack_body
        if force:
            # The store.add dedup silently skipped any colliding claim_id;
            # figure out which were skipped by diffing the parsed result against
            # what actually persisted.
            from cyclone import dedup
            body["skipped_claim_ids"] = _compute_skipped_837(rec.id, result.claims)
        return JSONResponse(content=body)

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

3. **Add the helpers `_build_dedup_409` and `_compute_skipped_837` as module-level functions** (right above the `parse_837` endpoint definition):

```python
def _build_dedup_409(
    result,
    report,
    *,
    kind: str,
    error: str,
) -> JSONResponse:
    """Build the standard 409 body for any dedup failure.

    Shared between parse_837 and parse_835; the only difference is the
    error string (we pass it in) and the kind (cross_batch / within_file /
    race) which drives the human-readable detail message.
    """
    if kind == "within_file":
        detail = (
            f"CLM01(s) {', '.join(report.within_file_duplicate_ids)} appear "
            f"twice in this file. Force-insert will keep the first occurrence "
            f"and skip the rest."
        )
    elif kind == "race":
        detail = (
            "Another process ingested a colliding batch between the check "
            "and the persist. Re-upload to retry with the latest state."
        )
    else:  # cross_batch
        detail = (
            f"{len(report.colliding_claim_ids)} of {report.total_claims} "
            f"claims collide with prior batch {report.existing_batch_id}. "
            f"Force-insert to skip the duplicates, or delete the prior batch."
        )

    body = {
        "error": error,
        "detail": detail,
        "batch_id": None,
        "existing_batch_id": report.existing_batch_id,
        "collisions": {
            "colliding_claim_ids": report.colliding_claim_ids,
            "total_collisions": len(report.colliding_claim_ids),
            "total_claims": report.total_claims,
            "new_claims_after_skip": report.new_claims_after_skip,
            "within_file_duplicate_ids": report.within_file_duplicate_ids,
        },
        "parse_result": json.loads(result.model_dump_json()),
    }
    return JSONResponse(status_code=409, content=body)


def _compute_skipped_837(batch_id: str, parsed_claims) -> list[str]:
    """Compare the parsed claims against what's in the DB for this batch.

    Returns the claim_ids in the parsed result that did NOT make it into
    the DB — these are the ones silently skipped by the store.add dedup.
    Sorted for determinism.
    """
    from cyclone import db as _db
    from cyclone.db import Claim
    parsed_ids = [c.claim_id for c in parsed_claims]
    if not parsed_ids:
        return []
    with _db.SessionLocal()() as s:
        persisted = s.execute(
            select(Claim.id).where(Claim.batch_id == batch_id)
        ).all()
    persisted_ids = {r[0] for r in persisted}
    return sorted(set(parsed_ids) - persisted_ids)
```

The `_compute_skipped_837` function needs `from sqlalchemy import select`
at the top of `api.py` (the file already imports it for other code; if
not, add it).

4. **Remove the old 409 handler comment** in the `except IntegrityError`
block — it's been replaced by the race handler above.

5. **Add the DELETE endpoint** (right after the 835 endpoint, before
`/api/parse-999`):

```python
@app.delete("/api/batches/{batch_id}")
def delete_batch_endpoint(batch_id: str) -> dict:
    """Hard-delete a batch and cascade through child rows.

    Returns 200 on success, 404 if the batch doesn't exist, 409 if any
    claim in the batch is past the ``submitted`` state (must unreconcile
    first). See ``CycloneStore.delete_batch`` for the cascade mechanics.
    """
    try:
        deleted = store.delete_batch(batch_id)
    except BatchNotDeletableError as exc:
        from fastapi.responses import JSONResponse as _JR
        return _JR(
            status_code=409,
            content={
                "error": "Batch not deletable",
                "detail": str(exc),
                "batch_id": exc.batch_id,
                "offending_claim_id": exc.offending_claim_id,
            },
        )
    if not deleted:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail=f"Batch {batch_id} not found")
    return {"ok": True, "batch_id": batch_id}
```

6. **Add the `BatchNotDeletableError` import** at the top of `api.py`
(find the existing imports from `cyclone.store`):

```python
from cyclone.store import (
    BatchNotDeletableError,  # NEW
    CycloneStore,
    store,
)
```

- [ ] **Step 4: Run the new tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest \
  tests/test_api_parse_persists.py -k "409 or force or delete_batch_endpoint" \
  --no-header -q
```

Expected: all 6 new + 1 pre-existing (`test_409_response_includes_existing_batch_id_for_837`)
should pass once the body shape matches. The pre-existing test
expects `body.get("existing_batch_id") == "PRIOR"` which still
holds under the new shape. PASS.

- [ ] **Step 5: Run the full test_api_parse_persists.py, expect all PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_api_parse_persists.py --no-header -q
```

Expected: all tests pass.

- [ ] **Step 6: Run the broader API test suite to catch regressions**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_api.py tests/test_api_835.py tests/test_api_gets.py \
  --no-header -q
```

Expected: all pass. If anything fails, it means an existing test
relied on the old 409 body shape or the absence of `force` — update
those tests to match the new contract.

- [ ] **Step 7: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/src/cyclone/api.py backend/tests/test_api_parse_persists.py
git commit -m "feat(api): 837 pre-flight dedup, force=true, race 409, DELETE /api/batches/{id}"
```

---

### Task 2.4: 835 endpoint pre-flight, `?force=true`, race handler

Mirror the 837 changes for the 835 endpoint.

**Files:**
- Modify: `backend/src/cyclone/api.py` (835 endpoint)
- Modify: `backend/tests/test_api_835.py` (add 409/force tests)

- [ ] **Step 1: Write the failing tests**

Add to `backend/tests/test_api_835.py` (find a good insertion point
near other parse tests):

```python
def test_parse_835_409_includes_parse_result_and_collisions(client: TestClient) -> None:
    """835 pre-flight dedup returns 409 with the full parse result + collisions."""
    from datetime import datetime, timezone
    from cyclone import db as _db
    from cyclone.db import Batch, Remittance

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="PRIOR835", kind="835", input_filename="p.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Remittance(
            id="CLP-A", batch_id="PRIOR835",
            payer_claim_control_number="CLP-A",
            status_code="1",
            received_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        ))
        s.commit()

    # Use the co_medicaid 835 fixture (real EDI), then post-assert the 409.
    from pathlib import Path
    text = (Path(__file__).parent / "fixtures" / "co_medicaid_835.txt").read_text()

    resp = client.post(
        "/api/parse-835",
        files={"file": ("dup.835", text, "text/plain")},
        headers={"Accept": "application/json"},
    )
    # The fixture's first CLP01 collides with the seeded one. We expect 409.
    assert resp.status_code == 409, resp.text
    body = resp.json()
    assert body["error"] == "Duplicate remittance"
    assert body["existing_batch_id"] == "PRIOR835"
    assert body["batch_id"] is None
    assert len(body["collisions"]["colliding_claim_ids"]) >= 1
    assert "parse_result" in body


def test_parse_835_force_true_persists_non_colliding_claims(client: TestClient) -> None:
    """With force=true, the colliding remittance is skipped, others persist."""
    from datetime import datetime, timezone
    from pathlib import Path
    from cyclone import db as _db
    from cyclone.db import Batch, Remittance

    with _db.SessionLocal()() as s:
        s.add(Batch(
            id="PRIOR835", kind="835", input_filename="p.txt",
            parsed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            raw_result_json={},
        ))
        s.add(Remittance(
            id="CLP-OLD", batch_id="PRIOR835",
            payer_claim_control_number="CLP-OLD",
            status_code="1",
            received_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        ))
        s.commit()

    # Build a small 835 with two CLP segments: one colliding, one new.
    text = (
        "ISA*00*          *00*          *ZZ*PAYERID        *ZZ*RECEIVERID     "
        "*240101*1200*^*00501*000000001*0*P*:~\n"
        "GS*HP*PAYERID*RECEIVERID*20240101*1200*1*X*005010X221A1~\n"
        "ST*835*0001~\n"
        "BPR*I*100*100*C*ACH*CCP*01*123456789*DA*0000001**01*123456789*DA*0000002*20240101~\n"
        "TRN*1*CHECK1*1234567890~\n"
        "DTM*405*20240101~\n"
        "N1*PR*PAYER NAME~\n"
        "N3*100 PAYER ST*~~\nN4*DENVER*CO*80202~\n"
        "REF*2U*PAYERID~\n"
        "N1*PE*RECEIVER NAME~\n"
        "N3*200 RECEIVER ST*~~\nN4*DENVER*CO*80202~\n"
        "LX*1~\n"
        "CLP*CLP-OLD*1*100*100*0*0*0*11*1*PATIENT*DOE****MI*M0001~\n"
        "NM1*QC*1*DOE*JOHN****MI*M0001~\n"
        "DTM*232*20240101~\n"
        "SVC*HC:99213*100*100**1~\n"
        "DTM*472*20240101~\n"
        "LX*2~\n"
        "CLP*CLP-NEW*1*50*50*0*0*0*11*1*PATIENT*ROE****MI*M0002~\n"
        "NM1*QC*1*ROE*JANE****MI*M0002~\n"
        "DTM*232*20240101~\n"
        "SVC*HC:99214*50*50**1~\n"
        "DTM*472*20240101~\n"
        "SE*30*0001~\nGE*1*1~\nIEA*1*000000001~\n"
    )

    resp = client.post(
        "/api/parse-835?force=true",
        files={"file": ("force.835", text, "text/plain")},
        headers={"Accept": "application/json"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert "CLP-OLD" in body.get("skipped_claim_ids", [])
    # CLP-NEW is in a new batch.
    with _db.SessionLocal()() as s:
        row = s.execute(_db.text("SELECT batch_id FROM remittances WHERE id='CLP-NEW'")).first()
    assert row is not None
    assert row[0] != "PRIOR835"
```

- [ ] **Step 2: Run the new tests, expect FAIL**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest \
  tests/test_api_835.py -k "409 or force" --no-header -q
```

Expected: 2 failures (no pre-flight on 835 yet, no force param).

- [ ] **Step 3: Implement the 835 endpoint changes**

In `backend/src/cyclone/api.py`:

1. Add `force: bool = Query(False)` to the 835 signature:

```python
@app.post("/api/parse-835")
async def parse_835_endpoint(
    request: Request,
    file: UploadFile = File(...),
    payer: str = Query("co_medicaid_835"),
    include_raw_segments: bool = Query(True),
    strict: bool = Query(False),
    force: bool = Query(False),
) -> Any:
```

2. Add the pre-flight check between validation and persist (mirror
837). Find the `try: store.add(...) except IntegrityError as exc:`
block in the 835 endpoint and replace it with the pre-flight +
race-handler pattern from Task 2.3, but call `dedup.preflight_835`
and pass `error="Duplicate remittance"`.

3. Add the `_compute_skipped_835` helper alongside `_compute_skipped_837`:

```python
def _compute_skipped_835(batch_id: str, parsed_remits) -> list[str]:
    """Compare parsed remittances against what's in the DB for this batch."""
    from sqlalchemy import select
    from cyclone import db as _db
    from cyclone.db import Remittance
    parsed_ids = [r.payer_claim_control_number for r in parsed_remits]
    if not parsed_ids:
        return []
    with _db.SessionLocal()() as s:
        persisted = s.execute(
            select(Remittance.id).where(Remittance.batch_id == batch_id)
        ).all()
    persisted_ids = {r[0] for r in persisted}
    return sorted(set(parsed_ids) - persisted_ids)
```

- [ ] **Step 4: Run the new tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest \
  tests/test_api_835.py -k "409 or force" --no-header -q
```

Expected: 2 passed.

- [ ] **Step 5: Run the full test_api_835.py, expect all PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest tests/test_api_835.py --no-header -q
```

Expected: all pass.

- [ ] **Step 6: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/src/cyclone/api.py backend/tests/test_api_835.py
git commit -m "feat(api): 835 pre-flight dedup, force=true, race 409"
```

---

### Task 2.5: Run the full backend test suite as a checkpoint

The 837/835 endpoints changed significantly. Run the full backend
test suite to catch any test that relied on the old 409 body shape.

- [ ] **Step 1: Run the full backend test suite**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend
PYTHONPATH=/Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix/backend/src \
  .venv/bin/python3.13 -m pytest --no-header -q
```

Expected: all pass. If any test fails:
- If the test was checking the 409 body shape, update it to the new
  shape (or remove the assertion if the field is gone).
- If the test was using `?force=true` semantics that don't match,
  adjust the test to match the new contract.
- If the test was using the old `existing_batch_id`-only body, update
  to the new full-body shape.

**Do not modify production code to make old tests pass.** If a test
was checking pre-conditions that the new code intentionally doesn't
satisfy (e.g., a test that asserts "without `?force=true`, the file
fails to ingest"), update the test to reflect the new contract.

- [ ] **Step 2: Commit any test fixes**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/tests/
git commit -m "test(api): update tests for new 409 body shape and force=true"
```

(Only if there were fixes; otherwise skip.)

---

## Phase 3 — Frontend

### Task 3.1: `ApiError` carries `collisions` + `parseResult`; `parse837/835` accept `force`

**Files:**
- Modify: `src/lib/api.ts`
- Create: `src/lib/api.test.ts`

- [ ] **Step 1: Write the failing tests**

Create `src/lib/api.test.ts`:

```typescript
import { describe, it, expect, vi, afterEach } from "vitest";
import { ApiError, parse837 } from "@/lib/api";

afterEach(() => vi.restoreAllMocks());

describe("ApiError", () => {
  it("carries collisions and parseResult when constructed", () => {
    const collisions = {
      colliding_claim_ids: ["A"],
      total_collisions: 1,
      total_claims: 2,
      new_claims_after_skip: 1,
    };
    const parseResult = { claims: [], summary: { total_claims: 2 } };
    const e = new ApiError(409, "dup", "BATCH-1", collisions, parseResult);
    expect(e.status).toBe(409);
    expect(e.existingBatchId).toBe("BATCH-1");
    expect(e.collisions).toEqual(collisions);
    expect(e.parseResult).toEqual(parseResult);
  });

  it("defaults collisions and parseResult to null", () => {
    const e = new ApiError(500, "boom");
    expect(e.collisions).toBeNull();
    expect(e.parseResult).toBeNull();
  });
});

describe("parse837 throws ApiError with collisions and parseResult", () => {
  it("extracts collisions and parse_result from 409 body", async () => {
    vi.stubGlobal("import.meta.env", { VITE_API_BASE_URL: "http://x" });
    const collisions = { colliding_claim_ids: ["A"], total_collisions: 1, total_claims: 2, new_claims_after_skip: 1 };
    const parse_result = { claims: [], summary: { total_claims: 2 } };
    const res = new Response(
      JSON.stringify({
        error: "Duplicate claim",
        detail: "collision",
        batch_id: null,
        existing_batch_id: "PRIOR",
        collisions,
        parse_result,
      }),
      { status: 409, headers: { "Content-Type": "application/json" } },
    );
    vi.stubGlobal("fetch", vi.fn(async () => res));
    const file = new File(["x"], "f.txt", { type: "text/plain" });
    await expect(
      parse837(file, { onProgress: () => {} }),
    ).rejects.toMatchObject({
      status: 409,
      existingBatchId: "PRIOR",
      collisions,
      parseResult: parse_result,
    });
  });

  it("appends ?force=true when options.force is set", async () => {
    vi.stubGlobal("import.meta.env", { VITE_API_BASE_URL: "http://x" });
    const fetchMock = vi.fn(async () =>
      new Response(JSON.stringify({ summary: { total_claims: 1 } }), {
        status: 200,
        headers: { "Content-Type": "application/json" },
      }),
    );
    vi.stubGlobal("fetch", fetchMock);
    const file = new File(["x"], "f.txt", { type: "text/plain" });
    await parse837(file, { onProgress: () => {}, force: true });
    const calledUrl = fetchMock.mock.calls[0][0];
    expect(calledUrl).toContain("force=true");
  });
});
```

- [ ] **Step 2: Run tests, expect FAIL**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/lib/api.test.ts
```

Expected: ApiError constructor doesn't accept `collisions`/`parseResult`
yet; `parse837` doesn't append `?force=true`.

- [ ] **Step 3: Update `src/lib/api.ts`**

1. **Update the `ApiError` class.** Find the current class (around
   line 164-167) and replace with:

```typescript
export type CollisionSummary = {
  colliding_claim_ids: string[];
  total_collisions: number;
  total_claims: number;
  new_claims_after_skip: number;
  within_file_duplicate_ids?: string[];
};

export class ApiError extends Error {
  constructor(
    public status: number,
    message: string,
    public existingBatchId: string | null = null,
    public collisions: CollisionSummary | null = null,
    public parseResult: unknown = null,
  ) {
    super(message);
  }
}
```

2. **Update `readErrorBody` to return more fields.** Find the
   `readErrorBody` function and replace it with:

```typescript
type ErrorBody = {
  detail?: unknown;
  error?: unknown;
  existing_batch_id?: unknown;
  collisions?: unknown;
  parse_result?: unknown;
};

async function readErrorBody(
  res: Response,
): Promise<{
  message: string;
  existingBatchId: string | null;
  collisions: CollisionSummary | null;
  parseResult: unknown;
}> {
  try {
    const t = await res.text();
    if (!t) return { message: "", existingBatchId: null, collisions: null, parseResult: null };
    try {
      const obj = JSON.parse(t) as ErrorBody;
      let message = "";
      if (typeof obj.detail === "string") message = obj.detail;
      else if (typeof obj.error === "string") message = obj.error;
      else message = t;
      const existing =
        typeof obj.existing_batch_id === "string" ? obj.existing_batch_id : null;
      const collisions =
        obj.collisions && typeof obj.collisions === "object"
          ? (obj.collisions as CollisionSummary)
          : null;
      const parseResult = obj.parse_result ?? null;
      return { message, existingBatchId: existing, collisions, parseResult };
    } catch {
      return { message: t, existingBatchId: null, collisions: null, parseResult: null };
    }
  } catch {
    return { message: "", existingBatchId: null, collisions: null, parseResult: null };
  }
}
```

3. **Update `parse837` to accept `force` and surface the new fields.**
   Find the `parse837` function. Replace the function signature and
   the `if (!res.ok)` block:

```typescript
export async function parse837(
  file: File,
  options: { onProgress?: (p: number) => void; force?: boolean } = {},
): Promise<ParseResult> {
  const url = `${base}/api/parse-837${options.force ? "?force=true" : ""}`;
  // ... existing fetch body ...
  if (!res.ok) {
    const { message, existingBatchId, collisions, parseResult } =
      await readErrorBody(res);
    throw new ApiError(
      res.status,
      `${res.status} ${res.statusText}${message ? ` — ${message}` : ""}`,
      existingBatchId,
      collisions,
      parseResult,
    );
  }
  return res.json();
}
```

Keep the rest of the function (the streaming logic for non-JSON
requests) unchanged — only the `if (!res.ok)` block and the URL
construction change.

4. **Update `parse835` similarly.** Same pattern: append
   `?force=true` to the URL and pass the new fields to `ApiError`.

- [ ] **Step 4: Run the new tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/lib/api.test.ts
```

Expected: 4 tests pass.

- [ ] **Step 5: Run the existing api.ts tests, expect PASS (or update)**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/lib/api.test.ts src/lib/inbox-api.test.ts
```

Expected: all pass. If the existing inbox-api or other tests assert
on `ApiError` shape and fail, update them to match the new
constructor signature (passing `null` for the new optional fields).

- [ ] **Step 6: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add src/lib/api.ts src/lib/api.test.ts
git commit -m "feat(api): ApiError carries collisions + parseResult; parse837/835 accept force"
```

---

### Task 3.2: New `deleteBatch` function in `src/lib/api.ts`

**Files:**
- Modify: `src/lib/api.ts`

- [ ] **Step 1: Add the function**

Find a good place in `src/lib/api.ts` (near `parse837`/`parse835`).
Add:

```typescript
export async function deleteBatch(batchId: string): Promise<void> {
  const res = await fetch(`${base}/api/batches/${encodeURIComponent(batchId)}`, {
    method: "DELETE",
  });
  if (!res.ok) {
    let body: { error?: string; detail?: string; offending_claim_id?: string } = {};
    try {
      body = (await res.json()) as typeof body;
    } catch {
      // ignore
    }
    throw new ApiError(
      res.status,
      `${res.status} ${res.statusText}${
        body.detail ? ` — ${body.detail}` : ""
      }`,
    );
  }
}
```

- [ ] **Step 2: Add a test**

Add to `src/lib/api.test.ts`:

```typescript
import { deleteBatch } from "@/lib/api";

describe("deleteBatch", () => {
  it("sends DELETE to /api/batches/{id}", async () => {
    vi.stubGlobal("import.meta.env", { VITE_API_BASE_URL: "http://x" });
    const fetchMock = vi.fn(async () => new Response(null, { status: 204 }));
    vi.stubGlobal("fetch", fetchMock);
    await deleteBatch("B123");
    expect(fetchMock).toHaveBeenCalledWith(
      "http://x/api/batches/B123",
      expect.objectContaining({ method: "DELETE" }),
    );
  });

  it("throws ApiError with the 409 detail on refusal", async () => {
    vi.stubGlobal("import.meta.env", { VITE_API_BASE_URL: "http://x" });
    const res = new Response(
      JSON.stringify({ error: "Batch not deletable", detail: "claim X is paid" }),
      { status: 409, headers: { "Content-Type": "application/json" } },
    );
    vi.stubGlobal("fetch", vi.fn(async () => res));
    await expect(deleteBatch("B1")).rejects.toMatchObject({ status: 409 });
  });
});
```

- [ ] **Step 3: Run, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/lib/api.test.ts
```

Expected: all 6 tests pass.

- [ ] **Step 4: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add src/lib/api.ts src/lib/api.test.ts
git commit -m "feat(api): deleteBatch function"
```

---

### Task 3.3: `Upload.tsx` panel with 4 actions

**Files:**
- Modify: `src/pages/Upload.tsx`
- Create: `src/pages/Upload.test.tsx`

- [ ] **Step 1: Write the failing tests**

Create `src/pages/Upload.test.tsx`:

```tsx
import { describe, it, expect, vi, beforeEach } from "vitest";
import { render, screen, fireEvent, waitFor } from "@testing-library/react";
import { MemoryRouter, Route, Routes } from "react-router-dom";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import * as apiModule from "@/lib/api";
import { ApiError, CollisionSummary } from "@/lib/api";
import { Upload } from "@/pages/Upload";

vi.mock("@/lib/api", async () => {
  const actual = await vi.importActual<typeof apiModule>("@/lib/api");
  return { ...actual, parse837: vi.fn(), deleteBatch: vi.fn() };
});

beforeEach(() => {
  vi.clearAllMocks();
});

function renderUpload() {
  const qc = new QueryClient({ defaultOptions: { queries: { retry: false } } });
  return render(
    <QueryClientProvider client={qc}>
      <MemoryRouter initialEntries={["/upload"]}>
        <Routes>
          <Route path="/upload" element={<Upload />} />
          <Route path="/batches/:id" element={<div data-testid="batch-page" />} />
        </Routes>
      </MemoryRouter>
    </QueryClientProvider>,
  );
}

const collisions: CollisionSummary = {
  colliding_claim_ids: ["A", "B"],
  total_collisions: 2,
  total_claims: 141,
  new_claims_after_skip: 139,
  within_file_duplicate_ids: [],
};

const parseResult = {
  envelope: {},
  claims: [
    { claim_id: "A", service_lines: [], diagnoses: [] },
    { claim_id: "B", service_lines: [], diagnoses: [] },
  ],
  summary: { total_claims: 141, passed: 141, failed: 0 },
};

describe("Upload error panel", () => {
  it("renders panel with all actions when 409 carries collisions + parseResult", async () => {
    vi.mocked(apiModule.parse837).mockRejectedValueOnce(
      new ApiError(
        409,
        "409 Conflict — dup",
        "PRIOR",
        collisions,
        parseResult,
      ),
    );
    renderUpload();
    const input = document.querySelector('input[type="file"]') as HTMLInputElement;
    const file = new File(["x"], "test.txt", { type: "text/plain" });
    fireEvent.change(input, { target: { files: [file] } });
    const forceBtn = await screen.findByRole("button", { name: /force insert/i });
    expect(forceBtn).toBeInTheDocument();
    expect(screen.getByRole("button", { name: /open prior batch/i })).toBeInTheDocument();
    expect(screen.getByRole("button", { name: /delete prior batch/i })).toBeInTheDocument();
    expect(screen.getByRole("button", { name: /pick a different file/i })).toBeInTheDocument();
    // The parse result summary is rendered.
    expect(screen.getByText(/141/)).toBeInTheDocument();
  });

  it("force-insert button re-calls parse837 with force=true", async () => {
    vi.mocked(apiModule.parse837)
      .mockRejectedValueOnce(
        new ApiError(409, "409 Conflict", "PRIOR", collisions, parseResult),
      )
      .mockResolvedValueOnce({ ...parseResult, skipped_claim_ids: ["A", "B"] });
    renderUpload();
    const input = document.querySelector('input[type="file"]') as HTMLInputElement;
    const file = new File(["x"], "test.txt", { type: "text/plain" });
    fireEvent.change(input, { target: { files: [file] } });
    const forceBtn = await screen.findByRole("button", { name: /force insert/i });
    fireEvent.click(forceBtn);
    await waitFor(() => {
      expect(apiModule.parse837).toHaveBeenCalledTimes(2);
    });
    expect(apiModule.parse837).toHaveBeenNthCalledWith(
      2,
      expect.any(File),
      expect.objectContaining({ force: true }),
    );
  });

  it("delete-prior button calls deleteBatch", async () => {
    vi.spyOn(window, "confirm").mockReturnValue(true);
    vi.mocked(apiModule.parse837).mockRejectedValueOnce(
      new ApiError(409, "409 Conflict", "PRIOR", collisions, parseResult),
    );
    vi.mocked(apiModule.deleteBatch).mockResolvedValueOnce(undefined);
    renderUpload();
    const input = document.querySelector('input[type="file"]') as HTMLInputElement;
    const file = new File(["x"], "test.txt", { type: "text/plain" });
    fireEvent.change(input, { target: { files: [file] } });
    const delBtn = await screen.findByRole("button", { name: /delete prior batch/i });
    fireEvent.click(delBtn);
    await waitFor(() => {
      expect(apiModule.deleteBatch).toHaveBeenCalledWith("PRIOR");
    });
  });

  it("within-file duplicate (existingBatchId null) hides prior batch actions", async () => {
    vi.mocked(apiModule.parse837).mockRejectedValueOnce(
      new ApiError(
        409,
        "409 Conflict",
        null,
        { ...collisions, within_file_duplicate_ids: ["DUP"] },
        parseResult,
      ),
    );
    renderUpload();
    const input = document.querySelector('input[type="file"]') as HTMLInputElement;
    const file = new File(["x"], "test.txt", { type: "text/plain" });
    fireEvent.change(input, { target: { files: [file] } });
    await screen.findByRole("button", { name: /force insert/i });
    expect(
      screen.queryByRole("button", { name: /open prior batch/i }),
    ).toBeNull();
    expect(
      screen.queryByRole("button", { name: /delete prior batch/i }),
    ).toBeNull();
  });

  it("'Pick a different file' button clears error", async () => {
    vi.mocked(apiModule.parse837).mockRejectedValueOnce(
      new ApiError(409, "409 Conflict", "PRIOR", collisions, parseResult),
    );
    renderUpload();
    const input = document.querySelector('input[type="file"]') as HTMLInputElement;
    const file = new File(["x"], "test.txt", { type: "text/plain" });
    fireEvent.change(input, { target: { files: [file] } });
    const clearBtn = await screen.findByRole("button", { name: /pick a different file/i });
    fireEvent.click(clearBtn);
    expect(screen.queryByText(/collide/i)).toBeNull();
  });

  it("does NOT render panel for non-409 errors", async () => {
    vi.mocked(apiModule.parse837).mockRejectedValueOnce(
      new ApiError(400, "bad file"),
    );
    renderUpload();
    const input = document.querySelector('input[type="file"]') as HTMLInputElement;
    const file = new File(["x"], "test.txt", { type: "text/plain" });
    fireEvent.change(input, { target: { files: [file] } });
    await new Promise((r) => setTimeout(r, 50));
    expect(screen.queryByText(/collide/i)).toBeNull();
  });
});
```

- [ ] **Step 2: Run tests, expect FAIL (Upload.tsx doesn't have the new state yet)**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/pages/Upload.test.tsx
```

Expected: all 6 tests fail (no panel exists).

- [ ] **Step 3: Modify `src/pages/Upload.tsx`**

This is the most invasive frontend change. Read the current
`Upload.tsx` (the file is 600+ lines; find the existing error-handling
catch block) and apply these changes:

1. **Update imports** at the top:

```typescript
import { ApiError, parse837, parse835, deleteBatch, CollisionSummary } from "@/lib/api";
import { useNavigate } from "react-router-dom";
```

2. **Add new state** near the other `useState` calls:

```typescript
type UploadError = {
  kind: "duplicate";
  existingBatchId: string | null;
  collisions: CollisionSummary;
  parseResult: unknown;
  filename: string;
};
const [uploadError, setUploadError] = useState<UploadError | null>(null);
const [forceInserting, setForceInserting] = useState(false);
const navigate = useNavigate();
```

3. **Replace the existing error-handling catch block** (find the
`catch (err) {` after the parse call) with:

```typescript
    } catch (err) {
      if (err instanceof ApiError && err.status === 409 && err.collisions) {
        setUploadError({
          kind: "duplicate",
          existingBatchId: err.existingBatchId,
          collisions: err.collisions,
          parseResult: err.parseResult,
          filename: file.name,
        });
        toast.error("Duplicate claim — file not ingested");
      } else {
        toast.error(
          err instanceof Error ? err.message : "Failed to parse file",
        );
      }
    }
```

4. **Add the panel JSX** above the streaming results section (find
the existing "parsed claims" / streaming-results JSX and insert
BEFORE it):

```tsx
{uploadError && uploadError.kind === "duplicate" ? (
  <div
    role="alert"
    data-testid="duplicate-error-panel"
    className="rounded-md border border-destructive/40 bg-destructive/5 p-4 mx-auto max-w-3xl"
  >
    <div className="flex items-center gap-2">
      <span className="inline-flex items-center rounded-md bg-destructive px-2 py-0.5 text-xs font-semibold text-destructive-foreground">
        409
      </span>
      <span className="font-semibold">
        {uploadError.collisions.total_collisions} of {uploadError.collisions.total_claims} claims
        collide
        {uploadError.existingBatchId
          ? ` with batch ${uploadError.existingBatchId}`
          : " within this file"}
      </span>
    </div>
    <p className="mt-2 text-sm text-muted-foreground">
      File <span className="font-mono">{uploadError.filename}</span> would persist
      {" "}{uploadError.collisions.new_claims_after_skip} of {uploadError.collisions.total_claims} claims.
      Colliding CLM01s: {uploadError.collisions.colliding_claim_ids.join(", ")}.
    </p>
    <div className="mt-3 flex flex-wrap gap-2">
      <Button
        disabled={forceInserting}
        onClick={async () => {
          setForceInserting(true);
          try {
            const result = await parse837(file, { onProgress: () => {}, force: true });
            setParseResult(result);
            setUploadError(null);
            const skipped = (result as { skipped_claim_ids?: string[] }).skipped_claim_ids?.length ?? 0;
            toast.success(
              `Force-inserted: ${result.summary.total_claims - skipped} of ${result.summary.total_claims} (skipped ${skipped} dups)`,
            );
          } catch (err) {
            toast.error(err instanceof Error ? err.message : "Force-insert failed");
          } finally {
            setForceInserting(false);
          }
        }}
      >
        Force insert (skip {uploadError.collisions.total_collisions} dups)
      </Button>
      {uploadError.existingBatchId ? (
        <>
          <Button
            variant="outline"
            onClick={() => navigate(`/batches/${uploadError.existingBatchId}`)}
          >
            Open prior batch
          </Button>
          <Button
            variant="outline"
            onClick={async () => {
              if (!confirm(`Delete batch ${uploadError.existingBatchId}? This cannot be undone.`)) return;
              try {
                await deleteBatch(uploadError.existingBatchId!);
                toast.success(`Deleted ${uploadError.existingBatchId}`);
                setUploadError(null);
                pickFile(null);
              } catch (err) {
                toast.error(err instanceof Error ? err.message : "Delete failed");
              }
            }}
          >
            Delete prior batch
          </Button>
        </>
      ) : null}
      <Button
        variant="ghost"
        onClick={() => {
          setUploadError(null);
          pickFile(null);
        }}
      >
        Pick a different file
      </Button>
    </div>
    <details className="mt-3 text-sm">
      <summary>Show parsed claims ({uploadError.collisions.total_claims})</summary>
      <pre className="mt-2 max-h-64 overflow-auto rounded bg-muted p-2 text-xs">
        {JSON.stringify(
          (uploadError.parseResult as { summary?: unknown })?.summary ?? {},
          null,
          2,
        )}
      </pre>
    </details>
  </div>
) : null}
```

5. **Ensure `parseResult` state setter exists.** The current
   `Upload.tsx` likely stores the parsed result under a different
   name (e.g., `result`, `parsed`). Use `setParseResult` or
   rename to match. **If the existing code uses a different name,
   adapt the JSX above to use that name** — do not introduce
   parallel state.

- [ ] **Step 4: Run the new tests, expect PASS**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/pages/Upload.test.tsx
```

Expected: 6 passed.

- [ ] **Step 5: Run the broader Upload-related tests to catch regressions**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test -- src/pages/Upload.test.tsx src/hooks/useParse.test.ts
```

Expected: all pass.

- [ ] **Step 6: Commit**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add src/pages/Upload.tsx src/pages/Upload.test.tsx
git commit -m "feat(upload): 409 panel with force-insert / open / delete / pick-different actions"
```

---

### Task 3.4: Frontend typecheck and lint

The frontend touched multiple files. Make sure typecheck and lint
pass.

- [ ] **Step 1: Run typecheck**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm run typecheck
```

Expected: no errors. If there are, fix them inline (mostly
`unknown` casts on `parseResult`).

- [ ] **Step 2: Run lint**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm run lint
```

Expected: no errors. Fix any inline.

- [ ] **Step 3: Run the full vitest suite**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
npm test
```

Expected: all pass.

- [ ] **Step 4: Commit any fixes**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add src/
git commit -m "chore(frontend): typecheck and lint fixes for upload panel"
```

(Only if there were fixes; otherwise skip.)

---

## Phase 4 — End-to-end verification

### Task 4.1: Apply migration 0013 to the production database

The production DB at `/Users/openclaw/.local/share/cyclone/cyclone.db`
has the inline UNIQUE constraint. Migration 0013 (already on the
worktree) drops it. Apply it.

**This is the change that fixes the user's immediate "every upload
fails with 409" problem.** All 19 files in
`docs/prodfiles/837p-from-axiscare/` will ingest cleanly.

**Files:** none (manual step)

- [ ] **Step 1: Stop the running uvicorn**

```bash
# Find and stop the dev server.
ps aux | grep -E "uvicorn cyclone" | grep -v grep
kill <pid>
```

- [ ] **Step 2: Make a backup of the production DB**

```bash
cp ~/.local/share/cyclone/cyclone.db ~/.local/share/cyclone/cyclone.db.bak.$(date +%s)
ls -la ~/.local/share/cyclone/cyclone.db*
```

- [ ] **Step 3: Apply migration 0013**

```bash
cd /Users/openclaw/dev/cyclone
backend/.venv/bin/python3.13 -c "
import sqlite3
con = sqlite3.connect('//Users/openclaw/.local/share/cyclone/cyclone.db')
with open('backend/src/cyclone/migrations/0013_drop_claims_unique_constraint.sql') as f:
    sql = f.read()
con.executescript(sql)
con.commit()
con.close()
print('migration 0013 applied')
"
```

**Note:** this is a one-off direct DB application. The next time
the app starts, the migration runner will see `user_version >= 13`
and skip 0013.

- [ ] **Step 4: Verify the UNIQUE is gone**

```bash
backend/.venv/bin/python3.13 -c "
import sqlite3
con = sqlite3.connect('//Users/openclaw/.local/share/cyclone/cyclone.db')
for row in con.execute(\"SELECT sql FROM sqlite_master WHERE name='claims' AND type='table'\"):
    print(row[0])
"
```

Expected: the printed `CREATE TABLE claims` definition does NOT have
`UNIQUE (batch_id, patient_control_number)` at the end.

- [ ] **Step 5: Restart the dev server**

```bash
cd /Users/openclaw/dev/cyclone
backend/.venv/bin/uvicorn cyclone.api:app --host 127.0.0.1 --port 8000 --reload &
```

- [ ] **Step 6: Repro the original 409 — should now succeed**

```bash
cd /Users/openclaw/dev/cyclone
# Run a small test that uploads one of the previously-failing files.
backend/.venv/bin/python3.13 /tmp/test_repro.py 2>&1 | tail -10
```

Expected: HTTP 200, the file persists. The 19 files in
`docs/prodfiles/837p-from-axiscare/` all upload cleanly now.

- [ ] **Step 7: Re-upload the same file — should now 409 with the new body shape**

```bash
# Re-run the same test; the second upload of the same file should
# trigger the pre-flight 409 with the rich body.
backend/.venv/bin/python3.13 /tmp/test_repro.py 2>&1 | tail -30
```

Expected: HTTP 409 with `existing_batch_id` pointing to the first
batch, `collisions.colliding_claim_ids: ["t991102984o1c1d", ...]`,
and a full `parse_result` in the body.

- [ ] **Step 8: Force-insert should succeed and skip the dupes**

Manually via curl or the test script: POST with `?force=true`. Expected: HTTP 200 with `skipped_claim_ids: [...]` listing the colliding claim_ids.

- [ ] **Step 9: DELETE /api/batches/{id} works**

Manually: `curl -X DELETE http://127.0.0.1:8000/api/batches/{id}`. Expected: 204 on success, 409 if the batch has paid claims, 404 if the id doesn't exist.

- [ ] **Step 10: Commit no code (manual step)**

If the manual step revealed that the code needs tweaks (e.g., the
parser handles a 19-claim file differently from a 2-claim test
fixture), commit those tweaks:

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git add backend/
git commit -m "fix: tweaks from e2e verification against production files"
```

(Only if needed.)

---

### Task 4.2: Merge to main

The plan is complete. The user can now upload files cleanly, see
the rich 409 body, force-insert, and delete prior batches.

- [ ] **Step 1: Verify the worktree is clean**

```bash
cd /Users/openclaw/dev/cyclone/.worktrees/claims-unique-fix
git status
```

Expected: working tree clean (or only the `backend/.venv` symlink,
which is in `.gitignore`).

- [ ] **Step 2: Merge to main**

```bash
cd /Users/openclaw/dev/cyclone
git checkout main
git merge --ff-only claims-unique-fix
```

- [ ] **Step 3: Push (only if the user wants to)**

```bash
git push origin main
```

(Skip if the user wants to keep this local; ask before pushing.)

---

## Self-review checklist (run after writing the plan)

1. **Spec coverage** — every section of the spec maps to a task:
   - §1 (incl. schema-bug rationale) → Task 1.3 (migration 0014 — added after spec/plan review revealed the PK must be relaxed for the workflow to fire)
   - §3.1 No collision → covered by existing tests + the new pre-flight not firing
   - §3.2 Collision → Task 2.3 (837) + Task 2.4 (835) + Task 3.3 (Upload panel)
   - §3.3 Force-insert → Task 2.3 (837) + Task 2.4 (835) + Task 3.1 (`?force=true`) + Task 3.3 (force-insert button)
   - §3.4 Race condition → Task 2.3 + Task 2.4 (race handler)
   - §4 Within-file duplicates → Task 2.1 (`preflight_837` `within_file_duplicate_ids`) + Task 2.3 (test) + Task 3.3 (panel hides prior batch actions)
   - §5 409 body shape → Task 2.3 (`_build_dedup_409`) + Task 3.1 (frontend `CollisionSummary` type)
   - §6 DELETE endpoint → Task 2.2 (store helper) + Task 2.3 (endpoint) + Task 3.2 (`deleteBatch` client fn) + Task 3.3 (delete button)
   - §7 Backend implementation → Tasks 1.3, 1.4, 2.1, 2.2, 2.3, 2.4
   - §8 Frontend implementation → Tasks 3.1, 3.2, 3.3
   - §10 Files changed → all listed (incl. migration 0014)
   - §11 Test plan → 14 backend + 6 frontend tests distributed across tasks, plus 3 new 0014 migration tests

2. **Schema-foundation gap (caught during plan execution)** — the original plan
   assumed `claims.id` could host multiple rows for the same CLM01, but the
   schema enforces `claims.id` as a single-column PRIMARY KEY. The implementer
   caught this on Task 1.3 (the helper's "most-recent" test could not insert
   two rows with the same id). The correct fix (per user direction: "fix
   this the correct way ... take pride in this") is **migration 0014**, which
   relaxes the PK to composite `(batch_id, id)` on both `claims` and
   `remittances`, and updates all FKs that referenced the old single-column
   PK. Task 1.3 is the new first executable task in Phase 1; the original
   Task 1.3 was renumbered to 1.4.

2. **Placeholder scan** — no "TBD", "TODO", "implement later". The
   note about "Open question resolved during implementation" in the
   spec was about the audit-event FK cascade; the implementation
   just records the event with `batch_id` set, relying on the
   existing `ON DELETE CASCADE` on `activity_events.batch_id` (or
   the alternative if not present). The plan tests this in Task 2.2
   by deleting a batch and asserting the cascade works.

3. **Type consistency** — every `ApiError` construction in the
   frontend passes 5 args (`status, message, existingBatchId,
   collisions, parseResult`). The `CollisionSummary` type is
   defined once in `api.ts` and imported where needed. The
   `CollisionReport` dataclass is used uniformly in
   `dedup.py`/`api.py`/`store.py`. The `skipped_claim_ids` field
   is only on the 200 success body (not on the 409 body) and is
   only set when `force=true`.

4. **Risk acknowledged** — pre-flight race, DELETE on large batch,
   `force=true` silent skip, within-file dupes, and
   `existing_batch_id` staleness are all documented in the spec
   §13 and tested where applicable.

5. **TDD throughout** — every task is RED (failing test) → GREEN
   (minimal code) → COMMIT. No "write tests after" anywhere.

6. **DRY** — the 837 and 835 changes share `_build_dedup_409` and
   the `_compute_skipped_*` helpers. The `CollisionReport`
   dataclass and `preflight_*` functions live in `dedup.py` so the
   837/835 endpoints share the same dedup logic.

7. **Frequent commits** — every task ends with a `git commit`.
   Expectation: ~7 backend commits + 3 frontend commits = 10
   commits total.