def ingest_batch(state, batch):
    return [state, report]

1. state holds cursor and the set of committed row ids
2. a row whose id is already committed is rejected as 'duplicate'
3. malformed rows are rejected with a reason, never skipped silently
4. the cursor advances past every row in the batch
5. replaying the same batch produces the same report (idempotent)
6. the report carries accepted and rejected counts

• rejected entries are (row_id, reason) tuples
• state and report are plain dicts/lists — JSON-safe
• committed ids are never forgotten