Claude Opus 4.8

# Shipment Damage Assessment

**Prepared for:** Museum Registrar / Insurance Reviewer
**Document type:** Incoming condition / triage report
**Basis:** Single photographic frame; no physical inspection performed

---

## 1) Visible Evidence Inventory

**Directly observed in image:**

- **Wooden crate** (pine/plywood construction), open-faced, resting on its side on a grey worktable. Lid/cover battens appear detached and resting on top of the crate rather than fastened.
- **Ceramic vessel** (small narrow-necked jar/pot, buff/tan color) seated upright on white foam inside the crate. **Visible dark crack lines** radiate across the body and around the neck/shoulder — a clear network of fracture lines is present on the surface.
- **White polyethylene/PE foam blocks** used as internal cushioning, partially cut/fitted; arrangement appears loose and not fully conforming to the object.
- **Loose stone/ceramic fragments** resting directly on the foam inside the crate (no individual wrapping visible).
- **Three sealed clear plastic bags** on the table containing additional stone/pottery fragments, each with a paper label inside.
- **Black electronic device** (handwritten card nearby reads "SHOCK SENSOR") — consistent with an impact/shock indicator, condition/state not legible.
- **Small white device** with cabling (lower center) — possibly a second sensor or data logger; label not legible.
- **Yellow tape measure** (scale tool).
- **Blue nitrile gloves** (one pair), unused appearance.
- **Multiple paper forms** — printed multi-field documents (condition reports / chain-of-custody / shipping paperwork). Text largely **illegible** at this resolution.
- **Handwritten index cards** partially legible: "SHOCK SENSOR," "CHAIN OF CUSTODY," "HUMIDITY / TEMPERATURE," and bag labels.
- **Background:** laboratory/collections storage area with shelving, white archival trays/boxes, additional tools.

**Not determinable from image:** sensor trigger status, exact fragment counts, label text, weights, prior condition baseline.

---

## 2) Suspected Damage Mechanisms (with confidence)

| Mechanism | Confidence | Reasoning |
|---|---|---|
| **Pre-existing or in-transit fracturing of the ceramic vessel** | **Medium-High** | Crack network is clearly visible on the body/neck. Cannot determine from one image whether these are old/stable cracks, restoration joins, or new transit damage. |
| **Object fragmentation / detached sherds** | **Medium** | Loose fragments present both inside crate and bagged. May be associated sherds of the vessel OR separate accessioned artifacts — unclear. |
| **Inadequate / non-conforming cushioning** | **Medium** | Foam blocks appear loose and not custom-cavity-cut to the vessel; vessel sits on a flat foam surface, allowing potential movement/tipping. |
| **Shock/impact event during transit** | **Low-Medium (Unverified)** | A shock sensor is present, implying impact monitoring was intended. Its triggered/untriggered state is not readable — this is the single most important unverified data point. |
| **Loose fragments contacting object (abrasion risk)** | **Low-Medium** | Unwrapped fragments resting on shared foam could rub or impact the vessel. |
| **Climate/humidity exposure** | **Low (Unknown)** | A temperature/humidity card is referenced; no readings visible. |

---

## 3) Immediate Handling Risks

- **Fragile cracked ceramic** — the visible fracture network means the vessel may be structurally unstable; lifting by the rim/neck could cause failure.
- **Loose unsecured fragments** that may shift, fall, or be lost during movement of the crate.
- **Open, possibly unsecured crate** — the lid is detached; the assembly should not be lifted or transported in current state.
- **Mixed object/fragment provenance risk** — fragments inside vs. bagged could be confused or co-mingled, threatening accession integrity.
- **Sensor data perishability** — shock/climate logger data could be lost or overwritten if not read and recorded promptly.

---

## 4) Missing Information Needed From Staff

1. **Shock sensor status** — was it triggered? Photograph/record the indicator immediately.
2. **Climate logger readings** — temp/humidity range during transit.
3. **Incoming condition report / baseline images** — are the vessel cracks pre-existing (documented prior) or new?
4. **Accession/inventory list** — how many fragments expected; do bagged fragments correspond to this object or separate items?
5. **Label/form transcription** — full text of the paper documents and bag labels.
6. **Chain-of-custody record** — handoff signatures, transit duration, carrier.
7. **Whether any disassembly/unpacking has already occurred** (i.e., is current layout "as received" or partially unpacked by staff?).

---

## 5) Recommended Triage Actions (Priority Order)

1. **Do not move the crate or vessel** until condition is documented and sensors read. Stabilize in place.
2. **Read and photograph the shock sensor and climate logger** before any further handling; record values to the condition file.
3. **High-resolution detail photography** of the vessel cracks (raking light), all fragments, and all labels/forms for the permanent record.
4. **Transcribe all paperwork and bag labels**; reconcile fragment counts against the accession/packing list.
5. **Segregate and individually support loose fragments** (cushioned trays, no co-mingling) and confirm whether they belong to the vessel.
6. **Stabilize the vessel** — assess crack stability before lift; use both-hands/under-base support, gloves (already staged), and a padded transfer tray. Defer lifting to a conservator if instability suspected.
7. **Quarantine for conservator review** — flag the ceramic for structural assessment before exhibition or further transport.
8. **Open a damage/discrepancy claim file** if cracks are confirmed new or fragment counts don't match.

---

## 6) Concise Insurance Summary

A ceramic vessel received in an open wooden crate displays a **visible network of cracks** across its body and neck; whether these are pre-existing or transit-related **cannot be determined from the image alone** and requires comparison to the incoming/pre-shipment condition report. Internal cushioning appears **loose and non-conforming**, and **unwrapped loose fragments** are present, indicating possible packing inadequacy and elevated handling risk. A **shock sensor and climate logger accompanied the shipment but their readings are not yet recorded** — these are critical to establishing whether a damaging event occurred in transit. **Recommendation:** withhold final liability determination pending (a) sensor data, (b) baseline condition comparison, and (c) fragment reconciliation. Object is currently classified **at-risk / fragile** and requires conservator stabilization. No conclusion of total loss is supported by available evidence.

---

## 7) JSON Object

```json
{
  "item_condition": {
    "primary_object": "ceramic vessel with visible crack network on body and neck",
    "stability": "uncertain - possibly fragile/unstable",
    "associated_fragments": "loose fragments present inside crate and in 3 sealed bags; provenance/count unverified",
    "damage_origin": "undetermined (pre-existing vs transit) - baseline comparison required",
    "confidence": "medium"
  },
  "packaging_condition": {
    "crate": "open wooden crate, lid/battens appear detached/unsecured",
    "cushioning": "PE foam blocks, loose and non-conforming to object",
    "fragment_protection": "inadequate - unwrapped fragments on shared foam",
    "assessment": "suboptimal / inadequate"
  },
  "visible_labels": [
    "SHOCK SENSOR (handwritten card)",
    "CHAIN OF CUSTODY (handwritten card)",
    "HUMIDITY / TEMPERATURE (handwritten card)",
    "bag labels present but text illegible",
    "multiple printed forms - text not legible at resolution"
  ],
  "risk_level": "high",
  "next_actions": [
    "Do not move crate or vessel until documented",
    "Read and photograph shock sensor and climate logger immediately",
    "High-resolution detail photography of cracks, fragments, labels",
    "Transcribe all paperwork; reconcile fragment counts to accession list",
    "Segregate and individually support loose fragments",
    "Conservator assessment of vessel crack stability before any lift",
    "Open damage/discrepancy claim file if cracks confirmed new"
  ]
}
```

---

*Note: All findings are derived from a single photograph. Sensor states, document text, and the pre-shipment condition baseline could not be verified and are explicitly flagged as uncertain. Physical inspection by a conservator is required to confirm damage origin and structural integrity.*

# Canal Lock Operations Schematic — Analysis

> **Reading-quality caveat (read first):** This is a photographed, folded, and heavily over-annotated paper schematic. The printed base text and the great majority of handwritten notes are **not legibly resolvable** at this image quality. Almost every alphanumeric ID below is partially or wholly illegible. I have transcribed only what I can defend and tagged everything else **UNCERTAIN**. Where I provide engineering logic (Sections 5–6), it is explicitly **inference**, not read from the page.

---

## 1) Plain-language summary

**Direct evidence:** The title block reads **"MUNICIPAL CANAL LOCK OPERATIONS"** with a sub-number that appears to be **`109.252.208704`** (UNCERTAIN). The drawing is a process/piping-and-control schematic showing multiple **pumps**, **gates**, and interconnecting runs traced over in **blue, green, orange/yellow, and red** highlighter — strongly suggesting separate fluid circuits and/or power-and-signal paths. A yellow **WARNING / Calibration** label is affixed mid-sheet, and "WARNING" is repeated lower on the drawing.

**Plausible interpretation (INFERENCE):** This documents the **control and fluid system for a municipal canal lock chamber** — i.e., the pumps and gate actuators that fill/drain the lock and open/close the lock gates, plus associated valves, sensors, and electrical feeds. The red handwriting (top-left: "PUMPS / VALVE SCHEDULE / Calibration") indicates the sheet is being used as a working **commissioning or calibration** reference, not a clean as-built.

---

## 2) Inventory of visible components

| Type | Apparent label | Location on sheet | Confidence |
|---|---|---|---|
| Pump | "PUMP B" | Center-top | Medium |
| Pump | "PUMP P-19" / "PUMP P 19" | Right of center, top | UNCERTAIN |
| Pump | "PUMP 122" | Center | UNCERTAIN |
| Pump | "PUMP P-2" or "PUMP 102" | Lower-center | UNCERTAIN |
| Pump | "PUMP P-9" / "P19 P9" | Right-center | UNCERTAIN |
| Pump | unlabeled symbol(s) | Multiple | — |
| Gate | "GATE 03" | Left | UNCERTAIN |
| Gate | "GATE B" | Center (referenced in red) | UNCERTAIN |
| Gate | "GATE 119" / "GATE B19" | Center | UNCERTAIN |
| Gate | "GATE 03" / "GAME 03" | Lower-right | UNCERTAIN |
| Valve | circle symbol marked "07" | Left-center | UNCERTAIN |
| Valve | numerous unlabeled valve symbols | Throughout | — |
| Sensor/instrument | tag bubbles with "15", "81", "67" | Scattered | UNCERTAIN — could be valve, gauge, or loop tags |
| Power feed | colored line runs (blue/green/orange/red) | Throughout | Cannot confirm which are power vs. fluid |
| Data table | parameter table, top-right | Top-right | Rows present but values illegible |
| Data table | second table | Right-center/lower | Illegible |
| Safety label | "WARNING / Calibration" sticker | Center | High (sticker present); body text illegible |

**Important:** I cannot confirm a component explicitly labeled **"Pump P-3"** or **"Gate B"** as called out in your question (Section 5). The closest readable tags are listed above and are UNCERTAIN. Section 5 therefore treats P-3 / Gate B as **named hypotheticals**.

---

## 3) Dependency map (bullet form)

**Caveat:** Connectivity below is inferred from highlighter routing and from standard lock architecture, **not** from legible line-by-line tracing.

- **Power feeds → Pumps → Gate actuators / fill-drain circuits**
  - Colored runs appear to fan out from a left-side source area toward pump and gate symbols (INFERENCE).
- **Pumps → Lock chamber level**
  - Fill/drain pumps drive chamber water level; gate operation is gated on level equalization (standard lock interlock — INFERENCE).
- **Gates ↔ Level sensors**
  - Gate "open" permissive normally depends on level/pressure equalization sensors (the "15/81/67" bubbles are candidate loop tags — UNCERTAIN).
- **Valves → Pump suction/discharge isolation**
  - Valve symbols (e.g., "07") sit inline on pump runs, implying isolation/throttling dependency (INFERENCE).
- **Calibration label → entire instrument set**
  - The Calibration sticker plus repeated "Calibration" notes imply sensors/valves are pending or mid-calibration, so **automatic interlocks may not be fully trustworthy right now** (INFERENCE — operationally significant).

---

## 4) Handwritten note conflicts & ambiguities

- **Pervasive illegibility:** Most red/blue cursive notes (left margin, lower area, top-right) cannot be transcribed. Do not assume any of them are non-critical. (UNCERTAIN)
- **Repeated "Calibration":** Appears top-left (red), on the yellow sticker, and lower-right — suggesting calibration status is **inconsistently recorded across the sheet** (which items are done vs. pending is unclear). Conflict risk: HIGH.
- **Two "WARNING" indications:** The yellow sticker reads "WARNION/WARNING" and a separate "WARNING" appears lower; their body text differs and neither is readable — possible **conflicting or superseded safety notes.**
- **Duplicate gate IDs:** "GATE 03" appears to be used in **two different locations** (left and lower-right). Either a numbering reuse or a transcription error — must be resolved before acting on any "GATE 03" instruction. (UNCERTAIN)
- **Pump ID ambiguity:** "P-2 / 102," "P-19 / P-9," "122" — the pump numbering is inconsistent between print and handwriting; risk of misidentifying a pump during a procedure. (UNCERTAIN)
- **Title sub-number** (`109.252.208704`) reads like an IP/asset string but is unverifiable. (UNCERTAIN)

---

## 5) Most likely failure sequence — *P-3 loses power while Gate B is closed*

> **This is engineering INFERENCE.** Neither "P-3" nor "Gate B" is confirmed legible on the sheet; treat as a scenario, not a read fact.

1. **Loss of P-3 output** → the fill/drain duty it serves stops; chamber level change halts or reverses (if P-3 was holding against a head).
2. **Gate B closed = chamber isolated** → no gravity equalization path through that gate; differential head across Gate B persists.
3. **Level/pressure imbalance** → upstream or downstream level diverges from setpoint; level sensors (candidate tags 15/81/67) flag deviation.
4. **Interlock response (IF calibrated):** gate-open permissives lock out — Gate B cannot be safely opened against the un-equalized head. **BUT** per the Calibration notes, interlocks may be untrusted → **manual verification required.**
5. **Backflow / check-valve loading:** with P-3 dead, discharge check valves and isolation valve ("07"?) take reverse pressure; risk of seal/check leak-back into the chamber.
6. **Cascade risk:** standby/parallel pump (P-2 / P-19?) may auto-start to cover duty → possible **overload of shared power feed** if P-3's fault is electrical (breaker trip propagation). 
7. **Worst case:** if no standby and Gate B stays sealed, the lock becomes **unable to equalize** → vessel transit blocked and sustained differential head on a closed gate (structural/seal stress).

---

## 6) Prioritized operator checklist — next 30 minutes

**Priority 1 — Safety & isolation (0–5 min)**
1. Confirm P-3 actual power status at the panel; identify if breaker tripped vs. upstream feed loss.
2. **Do NOT command Gate B open** until head differential across it is confirmed equalized by an independent (manual) reading — interlocks are calibration-suspect.
3. Verify the two "WARNING" notices for any active lockout/tagout before touching equipment.

**Priority 2 — Stabilize (5–15 min)**
4. Confirm whether a standby pump (P-2 / P-19 — UNCERTAIN IDs) auto-started; check shared feed amperage for overload.
5. Check discharge check valve / isolation valve "07" (UNCERTAIN) for back-leakage; isolate P-3 if leaking.
6. Manually log upstream and downstream levels to establish the real differential across Gate B.

**Priority 3 — Diagnose & resolve ID ambiguity (15–25 min)**
7. Resolve the **duplicate "GATE 03"** and **pump numbering** ambiguities against the master valve/pump schedule before any switching. 
8. Restore P-3 only after fault root-cause (electrical vs. mechanical) is known.

**Priority 4 — Document & escalate (25–30 min)**
9. Record the calibration status of all level/pressure loops; flag any interlock you had to bypass manually.
10. Escalate to supervising engineer if Gate B cannot be safely equalized or if shared power feed is overloaded.

---

### Confidence summary
- **High:** Document is a municipal canal-lock ops schematic; it is annotated; a calibration/warning sticker is present; multiple pumps and gates exist.
- **Medium/Low:** Any specific ID (pumps, gates, valves, the title number).
- **Inference only:** All connectivity (Section 3), the failure cascade (Section 5), and the checklist logic (Section 6).

If you can supply a higher-resolution or de-skewed scan of specific quadrants, I can convert many of the UNCERTAIN tags into verified reads.

# HelioLift Incident Review — Duplicate Dispatch Confirmations (Zone A12)

**Classification:** SEV-2 (regulatory-reportable, no patient harm) · **Status:** Contained pending durable fix · **IC:** AI Eng Lead

---

## 1) Executive Summary

At 02:10 UTC the ledger writer was enabled for 25% of dispatch traffic. Within ~3 minutes, Zone A12 showed **3.7% duplicate dispatch confirmations** but only **0.4% double launches**. The asymmetry is diagnostic: the **launch service deduplicates by `(patient_hash, seq)`** (a stable logical identity) and therefore suppressed almost all duplicate launches, while the **notification service deduplicates by `idempotency_key`**, which the writer generates **non-deterministically** (`patientHash:Date.now()`). Two writer nodes processing the same logical event produced two distinct keys, so notifications were not collapsed.

Three defects compounded:
1. **Non-deterministic idempotency key** — identity is wall-clock-based, not content-based.
2. **No database-level uniqueness** — `dispatch_ledger` has a *non-unique* index on `(patient_hash, seq)`; the "append-only" table physically permits duplicate state changes.
3. **Fail-open dedup on Redis timeout** — the Redis "last seq" gate is the only guard against duplicates, it is racy across nodes, and a timeout (observed on `aurora-4`) causes the stale-check to be skipped and the event committed anyway.

No patient harm occurred. The launch dedup key (correct identity) protected the safety-critical path. The fix is to make identity **deterministic and content-derived**, enforce it with a **unique constraint + `ON CONFLICT DO NOTHING`**, and demote Redis to an optimization (fail-closed to the DB authority). We can reach 100% ledger traffic within the week behind this guarantee.

---

## 2) Likely Root Cause

**Primary:** Identity for an append-only event was minted from `Date.now()`, so the *same logical event* received *different* idempotency keys on different nodes (`91f:1716171221102` vs `91f:1716171221188`). Any consumer keyed on `idempotency_key` (notifications) saw two distinct events.

**Enabling condition (concurrent writers):** During the 25% rollout, retry/failover routed the same logical event to two nodes (`aurora-7`, `aurora-2`) within 86 ms. The Redis "last seq" gate is:
- **Eventually consistent / racy** — both nodes can read `lastSeq < 44` before either writes.
- **Fail-open** — on the `aurora-4` timeout, `redis.get` did not return a guard value, the code path treated it as "no prior seq," and the duplicate `seq=44 CONFIRMED` was committed (`pg_commit_ok` immediately follows `redis_timeout`).

**Why 3.7% vs 0.4%:**
- Launch path keyed on `(patient_hash, seq)` → stable identity → ~all dupes suppressed; the residual 0.4% reflects launches that crossed the dedup window or raced the launch service's own check before the first row was visible.
- Notification path keyed on `idempotency_key` → unstable identity → full 3.7% leaked.

**Not the cause:** NTP drift (<20 ms, well within tolerance) and "append-only semantics" themselves. The ledger model is sound; its *uniqueness invariant was never enforced in storage*.

**Transaction correctness note:** The writer issues `BEGIN`/`COMMIT` as separate statements over a pooled connection with no `try/catch`/`ROLLBACK`. A throw between `BEGIN` and `COMMIT` (e.g., the Redis call after the INSERT) can leak an open transaction or a half-applied Redis state. This is a latent reliability bug independent of the duplicate symptom.

---

## 3) Immediate Containment (0–4 h, no mission pause)

These are reversible, do not require schema locks, and do not pause emergency traffic.

1. **Freeze the rollout at 25%.** Do not advance until durable fix ships. (Config flag, no deploy.)
2. **Hotfix the idempotency key to deterministic** (`patientHash:seq:kind`) and deploy. This alone collapses duplicate notifications immediately, even before the DB constraint lands. See §5.
3. **Make Redis fail-closed.** On `redis.get` timeout/error, do **not** skip the guard — fall through to the DB-authoritative write path (which, once §6 ships, is idempotent via `ON CONFLICT`). Until the constraint exists, fail-closed means: route those events to the **legacy `dispatch_events`** path for that patient (which has a real PK), preserving the old invariant.
4. **Quarantine + reconcile notifications.** Run a one-shot job to collapse already-emitted duplicate confirmations by `(patient_hash, seq)` and suppress pending duplicates in the notification queue.
5. **Add a synthetic alert:** fire if `COUNT(*) > 1 GROUP BY (patient_hash, seq, event_kind)` appears in `dispatch_ledger` within any 5-min window.

**Containment exit criterion:** duplicate-confirmation rate in A12 < 0.1% over a 60-min window after the deterministic-key hotfix.

---

## 4) Durable Fix

| Layer | Change | Guarantee |
|---|---|---|
| **Identity** | Idempotency key = HMAC/content hash over the *logical* event `(patient_hash, seq, kind)` (payload excluded to tolerate retries with cosmetic differences). | Same logical event → same key on every node. |
| **Storage** | `UNIQUE (patient_hash, seq, event_kind)` **and** `UNIQUE (idempotency_key)` on `dispatch_ledger`, created `CONCURRENTLY`. | DB is the authority; concurrent writers cannot duplicate. |
| **Writer** | Single transaction, `INSERT ... ON CONFLICT DO NOTHING RETURNING`; report `written:false` on conflict. Wrap in `try/catch` with `ROLLBACK`. | Exactly-once append; no leaked transactions. |
| **Cache** | Redis demoted to a *hint* (fast-path skip of obviously-stale events). Source of truth = DB. Fail-closed: on Redis error, proceed to DB write (idempotent). | No fail-open duplicates. |
| **Consumers** | Standardize all downstream dedup on `(patient_hash, seq, event_kind)`; treat `idempotency_key` as an audit/transport token, not the dedup key. | Uniform identity across the system. |
| **Privacy** | Replace bare `patient_hash` with keyed HMAC + rotating epoch salt; add re-identification controls (§9). | Pseudonymization defensible to regulator. |

The decisive change is moving the uniqueness invariant **into the database**. Redis, NTP, and node coordination all become non-load-bearing for correctness.

---

## 5) TypeScript Patch

```ts
import { createHash } from 'node:crypto';

type EventKind = 'REQUESTED' | 'CONFIRMED' | 'LAUNCHED' | 'CANCELLED';

type DispatchEvent = {
  patientHash: string;
  seq: number;
  kind: EventKind;
  payload: Record<string, unknown>;
};

/**
 * Deterministic logical identity for an append-only event.
 * Derived ONLY from (patientHash, seq, kind) so that retries/failover on any
 * node produce the SAME key. Payload is intentionally excluded: cosmetic retry
 * differences (timestamps, trace ids) must not fork identity.
 *
 * If you require payload binding, hash a canonicalized payload behind a feature
 * flag and version the key (v2:) so consumers can migrate.
 */
function idempotencyKey(e: Pick<DispatchEvent, 'patientHash' | 'seq' | 'kind'>): string {
  return (
    'v2:' +
    createHash('sha256')
      .update(`${e.patientHash}\u0000${e.seq}\u0000${e.kind}`)
      .digest('hex')
  );
}

export async function writeDispatchEvent(event: DispatchEvent, ctx: Ctx) {
  const key = idempotencyKey(event);

  // Redis is a HINT ONLY. Never block correctness on it; never fail open.
  // On error we proceed to the DB, which is the authority via UNIQUE + ON CONFLICT.
  let cacheSaysStale = false;
  try {
    const lastSeq = await ctx.redis.get(`last:${event.patientHash}`);
    if (lastSeq && event.seq < Number(lastSeq)) {
      cacheSaysStale = true; // strictly-older is safe to skip; equality must hit DB
    }
  } catch (err) {
    ctx.logger.warn({ patient: event.patientHash, err: String(err) }, 'redis hint unavailable; proceeding to DB');
  }

  if (cacheSaysStale) {
    ctx.logger.info({ key, seq: event.seq }, 'stale event skipped via cache hint');
    return { written: false, reason: 'stale_hint' as const };
  }

  const client = await ctx.pg.connect(); // dedicated connection for the txn
  try {
    await client.query('BEGIN');

    const res = await client.query(
      `INSERT INTO dispatch_ledger
         (ledger_id, patient_hash, seq, event_kind, payload, idempotency_key, writer_node)
       VALUES (gen_random_uuid(), $1, $2, $3, $4, $5, $6)
       ON CONFLICT (patient_hash, seq, event_kind) DO NOTHING
       RETURNING ledger_id`,
      [event.patientHash, event.seq, event.kind, event.payload, key, ctx.nodeId]
    );

    await client.query('COMMIT');

    const written = res.rowCount === 1;

    // Best-effort cache update AFTER commit. Failure here is harmless.
    if (written) {
      ctx.redis
        .set(`last:${event.patientHash}`, String(event.seq), { EX: 600 })
        .catch((err) => ctx.logger.warn({ err: String(err) }, 'cache set failed (non-fatal)'));
    }

    return written
      ? { written: true as const, key }
      : { written: false as const, reason: 'duplicate_suppressed_by_db' as const, key };
  } catch (err) {
    await client.query('ROLLBACK').catch(() => void 0);
    ctx.logger.error({ key, seq: event.seq, err: String(err) }, 'ledger write failed; rolled back');
    throw err; // let the caller's retry policy re-drive; the write is idempotent
  } finally {
    client.release();
  }
}
```

**Why these choices**
- `ON CONFLICT (patient_hash, seq, event_kind) DO NOTHING` makes concurrent double-writes from `aurora-7`/`aurora-2` collapse to one row, regardless of timing or Redis state.
- Strict `<` for the cache hint (not `<=`) prevents a stale cache from masking a legitimate equal-seq DB-level idempotency check.
- The dedicated connection + `try/catch/ROLLBACK/finally` removes the leaked-transaction failure mode.
- `RETURNING` lets us truthfully report whether we wrote, so the notification service can be driven *off the writer's result* rather than re-deriving identity.

---

## 6) SQL Migration Notes

Constraints must be added **without locking dispatch** (emergency missions cannot pause). Sequence matters: you cannot create a unique index over data that already contains duplicates.

```sql
-- STEP 0: Identify existing duplicates introduced during the 02:10–containment window.
-- (Audit artifact for the regulator; export before deleting.)
CREATE TABLE dispatch_ledger_dupe_audit AS
SELECT patient_hash, seq, event_kind, COUNT(*) AS n,
       array_agg(ledger_id ORDER BY created_at) AS ledger_ids,
       array_agg(idempotency_key) AS keys,
       min(created_at) AS first_seen, max(created_at) AS last_seen
FROM dispatch_ledger
GROUP BY patient_hash, seq, event_kind
HAVING COUNT(*) > 1;

-- STEP 1: Deduplicate in place, keeping the earliest row (first writer wins).
-- Run in batches in production to bound lock footprint.
WITH ranked AS (
  SELECT ledger_id,
         row_number() OVER (
           PARTITION BY patient_hash, seq, event_kind
           ORDER BY created_at, ledger_id
         ) AS rn
  FROM dispatch_ledger
)
DELETE FROM dispatch_ledger d
USING ranked r
WHERE d.ledger_id = r.ledger_id
  AND r.rn > 1;

-- STEP 2: Add the logical-identity uniqueness CONCURRENTLY (no table lock).
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS
  dispatch_ledger_logical_uq
  ON dispatch_ledger (patient_hash, seq, event_kind);

-- STEP 3: Enforce transport-token uniqueness (catches accidental key reuse).
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS
  dispatch_ledger_idem_uq
  ON dispatch_ledger (idempotency_key);

-- STEP 4: Drop the now-redundant non-unique index after the unique one is valid.
DROP INDEX CONCURRENTLY IF EXISTS dispatch_ledger_patient_seq_idx;

-- STEP 5: Guard append-only semantics at the storage layer.
-- Block UPDATE/DELETE on the ledger except via an explicit, audited admin role.
CREATE OR REPLACE FUNCTION ledger_is_append_only() RETURNS trigger AS $$
BEGIN
  RAISE EXCEPTION 'dispatch_ledger is append-only (% blocked)', TG_OP;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER dispatch_ledger_no_mutate
  BEFORE UPDATE OR DELETE ON dispatch_ledger
  FOR EACH ROW EXECUTE FUNCTION ledger_is_append_only();
-- NOTE: install this trigger AFTER STEP 1 dedup completes.
```

**Operational cautions**
- `CREATE UNIQUE INDEX CONCURRENTLY` cannot run inside a transaction block and will **fail (leaving an INVALID index)** if duplicates remain — hence Steps 0–1 first. If it fails, `DROP INDEX CONCURRENTLY` the invalid index and re-dedup.
- Run Step 1 in keyset-paginated batches (e.g., 10k rows) during low-traffic minutes to keep autovacuum and replication lag bounded.
- Keep `dispatch_events` writable as the **fallback path** until the ledger is at 100% and validated; do not drop it this week.
- After cutover, consider partitioning `dispatch_ledger` by `created_at` month to keep the unique-index maintenance cost flat.

---

## 7) Python Replay-Safety Validator

Validates that the ledger can be deterministically replayed into a single, consistent current state per patient — the core promise of the new architecture. Run pre-cutover and continuously in CI/canary.

```python
#!/usr/bin/env python3
"""
replay_validator.py — verifies dispatch_ledger is safe to replay.

Checks:
  1. Logical-identity uniqueness: no duplicate (patient_hash, seq, event_kind).
  2. Idempotency-key uniqueness and determinism (key derivable from identity).
  3. Sequence monotonicity / no gaps that imply lost events (configurable).
  4. State-machine legality: REQUESTED -> CONFIRMED -> LAUNCHED, with CANCELLED
     permitted from any non-terminal state. No transition after a terminal state.
  5. Deterministic fold: replaying in (seq) order yields exactly one final state.

Exit code 0 = replay-safe. Non-zero = violations (printed as JSON for audit).
"""
from __future__ import annotations
import hashlib
import json
import sys
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Iterable

KIND_ORDER = {"REQUESTED": 0, "CONFIRMED": 1, "LAUNCHED": 2, "CANCELLED": 99}
TERMINAL = {"LAUNCHED", "CANCELLED"}
LEGAL_TRANSITIONS = {
    None: {"REQUESTED"},
    "REQUESTED": {"CONFIRMED", "CANCELLED"},
    "CONFIRMED": {"LAUNCHED", "CANCELLED"},
    "LAUNCHED": set(),      # terminal
    "CANCELLED": set(),     # terminal
}


@dataclass
class LedgerRow:
    patient_hash: str
    seq: int
    event_kind: str
    idempotency_key: str
    payload: dict


def expected_key(row: LedgerRow) -> str:
    h = hashlib.sha256(

# HelixQueue: Duplicate Assignments & Missing Audit Rows — Senior Engineering Response

## Context Assumptions (state explicitly)

- Batches arrive as Redis stream entries; workers consume via a consumer group; HelixQueue's scheduler reads, assigns to a GPU worker, writes a `batch_assignments` row and an `audit_log` row.
- "International sample identifiers" means non-ASCII Unicode strings now appear in IDs/keys.
- Current assignment likely uses a `SELECT ... then INSERT/UPDATE` pattern without `SELECT FOR UPDATE`/upsert, and Redis `XREADGROUP` without disciplined `XACK`/`XAUTOCLAIM`.
- "Missing audit rows" implies the audit write is in a separate transaction or fire-and-forget, not atomic with the assignment.

---

## 1) Likely Root Causes (ranked by probability)

| # | Root cause | Why it fits the symptoms | Confidence |
|---|-----------|--------------------------|-----------|
| 1 | **Unicode non-normalized keys** used for dedup/idempotency. Same logical ID arrives as NFC vs NFD (or with differing case/width), so dedup checks miss and produce a second assignment. | Bug appeared *exactly* when international IDs were added. Two byte-different-but-equal strings defeat any equality-based guard. | **High** |
| 2 | **Non-atomic claim**: read-modify-write race (`SELECT unassigned` → app logic → `INSERT assignment`) without `FOR UPDATE SKIP LOCKED` or a unique constraint. Two scheduler iterations / two pods grab the same batch under load. | Duplicates only under heavy load is the classic lost-update signature. | **High** |
| 3 | **Redis consumer-group redelivery on restart**: messages read but not `XACK`ed before a worker dies are redelivered (or `XAUTOCLAIM`ed), reprocessed, and create a second assignment because DB side isn't idempotent. | "Workers restart mid-batch" is the precise trigger for pending-entry redelivery. | **High** |
| 4 | **Audit write outside the assignment transaction** (separate `commit`, async task, or after Redis ack). On crash between commits, assignment exists but audit row is lost. | Explains *missing audit rows* specifically tied to restarts. | **Medium-High** |
| 5 | **PostgreSQL collation / `citext` / case-fold mismatch** on the ID column, or an index using a collation that treats distinct Unicode forms as different, so a `UNIQUE` constraint silently allows near-duplicates. | Secondary to #1; surfaces if a constraint exists but uses the wrong normalization. | **Medium** |
| 6 | **Idempotency key derived from mutable/locale-dependent data** (e.g., includes timestamp, truncated string, or `str.lower()` that's locale-broken for non-ASCII). | Would intermittently change keys for the same logical work. | **Low-Medium** |

**Primary hypothesis:** #1 + #2 + #3 compound — non-normalized keys remove the safety net, and the claim/redelivery race exercises it. #4 is an independent atomicity bug producing the missing-audit symptom.

---

## 2) Minimal Reproduction Plan

Goal: deterministically reproduce both symptoms in CI, fast.

**A. Unicode dedup miss (no load needed)**
```python
nfc = "Sämple-Ω-001"                     # composed
nfd = unicodedata.normalize("NFD", nfc)  # decomposed, byte-different
assert nfc != nfd                        # different bytes
submit_batch(sample_id=nfc)
submit_batch(sample_id=nfd)              # should be deduped; currently isn't
assert assignment_count(logical_id) == 1 # FAILS today
```

**B. Concurrent claim race**
- Seed 1 unassigned batch.
- Launch N=50 scheduler coroutines/threads calling `assign_next()` concurrently (use `asyncio.gather` or a `ThreadPoolExecutor`; add a `pg_sleep`/`asyncio.sleep` between the SELECT and INSERT to widen the window).
- Assert exactly one assignment row. Currently >1.

**C. Crash-during-batch redelivery**
- Use a test Redis with a consumer group.
- Worker reads an entry (`XREADGROUP`), writes assignment, **does not XACK**, then `os._exit(1)` (simulate crash).
- Restart, run `XAUTOCLAIM` of pending entries.
- Assert assignment processed exactly once and audit row present.

**D. Atomicity of audit**
- Inject a fault (monkeypatch) that raises *after* the assignment commit but *before* the audit write.
- Assert: with the redesign, either both rows exist or neither (no orphan assignment).

Wrap A–D as pytest cases; B/C run under `pytest -p no:randomly` with a real Postgres + Redis via `testcontainers`. Target total runtime < 60s.

---

## 3) Transaction-Safe Redesign

### Core principles
1. **Canonicalize all external identifiers** at the boundary: `unicodedata.normalize("NFC", s).strip()`, reject empty/control chars. Store the canonical form; keep the raw form in a separate column for forensics.
2. **Deterministic idempotency key** derived only from canonical, immutable fields.
3. **Single atomic transaction** writes assignment + audit + state transition. Redis `XACK` happens *only after* the DB commit succeeds.
4. **Explicit state machine** with DB-enforced legal transitions.
5. **Claim via `INSERT ... ON CONFLICT DO NOTHING`** keyed on idempotency key — the database is the arbiter, not application logic.

### Idempotency key
```python
def make_idempotency_key(batch_id: str, sample_id: str, attempt_epoch: str) -> str:
    canon = lambda s: unicodedata.normalize("NFC", s).strip()
    payload = "\x1f".join([canon(batch_id), canon(sample_id), attempt_epoch])
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
```
- `attempt_epoch` is a stable per-batch token (e.g., the Redis stream message ID), **not** wall-clock — so redelivery of the *same* message yields the *same* key, while a legitimately new submission yields a new key.

### State machine
```
PENDING ──claim──▶ ASSIGNED ──start──▶ RUNNING ──finish──▶ COMPLETED
   ▲                  │                    │
   └──────release─────┴────────fail────────┴──▶ FAILED ──retry──▶ PENDING
```
Illegal transitions rejected by a CHECK + trigger (below). A worker restart leaves a row in `ASSIGNED`/`RUNNING` with a `lease_expires_at`; a reaper releases expired leases back to `PENDING`.

---

## 4) PostgreSQL Schema Changes

```sql
-- 0. Normalization-safe ID columns. citext only case-folds; it does NOT
--    Unicode-normalize. So we store an app-normalized canonical column.
CREATE TYPE batch_state AS ENUM
  ('PENDING','ASSIGNED','RUNNING','COMPLETED','FAILED');

CREATE TABLE batches (
    id                BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sample_id_raw     TEXT NOT NULL,                 -- exactly as received
    sample_id_canon   TEXT NOT NULL,                 -- NFC-normalized, trimmed
    state             batch_state NOT NULL DEFAULT 'PENDING',
    worker_id         TEXT,
    lease_expires_at  TIMESTAMPTZ,
    attempt           INT NOT NULL DEFAULT 0,
    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT sample_canon_nonempty CHECK (length(sample_id_canon) > 0)
);

-- Enforce uniqueness on the canonical form (prevents NFC/NFD twins).
CREATE UNIQUE INDEX uq_batches_sample_canon ON batches (sample_id_canon);

-- 1. Assignments: the idempotency key is the dedup arbiter.
CREATE TABLE batch_assignments (
    id               BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    batch_id         BIGINT NOT NULL REFERENCES batches(id),
    worker_id        TEXT   NOT NULL,
    idempotency_key  TEXT   NOT NULL,
    stream_msg_id    TEXT   NOT NULL,
    state            batch_state NOT NULL DEFAULT 'ASSIGNED',
    created_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT uq_assignment_idem UNIQUE (idempotency_key)  -- THE guard
);

-- One active (non-terminal) assignment per batch.
CREATE UNIQUE INDEX uq_one_active_assignment
    ON batch_assignments (batch_id)
    WHERE state IN ('ASSIGNED','RUNNING');

-- 2. Audit log, written in the SAME transaction.
CREATE TABLE audit_log (
    id               BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    batch_id         BIGINT NOT NULL REFERENCES batches(id),
    assignment_id    BIGINT REFERENCES batch_assignments(id),
    from_state       batch_state,
    to_state         batch_state NOT NULL,
    worker_id        TEXT,
    idempotency_key  TEXT NOT NULL,
    detail           JSONB NOT NULL DEFAULT '{}'::jsonb,
    created_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
    -- Make audit idempotent too, so retries don't double-log.
    CONSTRAINT uq_audit_event UNIQUE (idempotency_key, to_state)
);

-- 3. Enforce legal transitions at the DB level.
CREATE OR REPLACE FUNCTION enforce_batch_transition() RETURNS trigger AS $$
BEGIN
  IF OLD.state = NEW.state THEN RETURN NEW; END IF;
  IF NOT (
       (OLD.state='PENDING'   AND NEW.state IN ('ASSIGNED'))
    OR (OLD.state='ASSIGNED'  AND NEW.state IN ('RUNNING','PENDING','FAILED'))
    OR (OLD.state='RUNNING'   AND NEW.state IN ('COMPLETED','FAILED','PENDING'))
    OR (OLD.state='FAILED'    AND NEW.state IN ('PENDING'))
  ) THEN
    RAISE EXCEPTION 'illegal transition % -> %', OLD.state, NEW.state;
  END IF;
  NEW.updated_at = now();
  RETURN NEW;
END $$ LANGUAGE plpgsql;

CREATE TRIGGER trg_batch_transition
  BEFORE UPDATE OF state ON batches
  FOR EACH ROW EXECUTE FUNCTION enforce_batch_transition();
```

Key points:
- `uq_assignment_idem` makes **duplicate assignment physically impossible**.
- `uq_one_active_assignment` (partial unique) guarantees only one in-flight assignment per batch even if keys differ.
- `uq_audit_event` makes the **audit insert idempotent** so retries can't double-write and crashes can't drop it (it's in the same tx).
- Uniqueness on `sample_id_canon` kills the NFC/NFD twin problem.

---

## 5) Corrected Scheduler Loop (Python pseudocode)

```python
import unicodedata, hashlib
from sqlalchemy import text

LEASE_SECONDS = 120

def canon(s: str) -> str:
    s = unicodedata.normalize("NFC", s).strip()
    if not s or any(unicodedata.category(c).startswith("C") for c in s):
        raise ValueError("invalid identifier")
    return s

def idem_key(sample_canon: str, stream_msg_id: str) -> str:
    payload = f"{sample_canon}\x1f{stream_msg_id}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()

async def run_scheduler(group, consumer, worker_id):
    while not shutdown.is_set():
        # 1. Reclaim crashed peers' pending work (idempotent by design).
        await redis.xautoclaim(STREAM, group, consumer,
                               min_idle_time=LEASE_SECONDS*1000, start_id="0-0",
                               count=64)
        # 2. Read new + reclaimed entries.
        entries = await redis.xreadgroup(group, consumer,
                                         {STREAM: ">"}, count=32, block=2000)
        for msg_id, fields in entries:
            try:
                await process_one(msg_id, fields, worker_id)
                await redis.xack(STREAM, group, msg_id)   # ACK only after commit
            except RetryableError:
                pass        # leave pending; xautoclaim will redeliver
            except PoisonError:
                await dead_letter(msg_id, fields)
                await redis.xack(STREAM, group, msg_id)

async def process_one(msg_id, fields, worker_id):
    sample = canon(fields["sample_id"])
    key = idem_key(sample, msg_id)

    async with engine.begin() as tx:          # single atomic transaction
        # Upsert batch by canonical id; lock the row.
        batch = await tx.execute(text("""
            INSERT INTO batches (sample_id_raw, sample_id_canon)
            VALUES (:raw, :canon)
            ON CONFLICT (sample_id_canon) DO UPDATE
              SET updated_at = now()
            RETURNING id, state
        """), {"raw": fields["sample_id"], "canon": sample}).one()

        # Atomic claim: the unique idempotency_key is the arbiter.
        res = await tx.execute(text("""
            INSERT INTO batch_assignments
              (batch_id, worker_id, idempotency_key, stream_msg_id, state)
            VALUES (:b, :w, :k, :m, 'ASSIGNED')
            ON CONFLICT (idempotency_key) DO NOTHING
            RETURNING id
        """), {"b": batch.id, "w": worker_id, "k": key, "m": msg_id})
        row = res.first()

        if row is None:
            # Already assigned by us before a crash -> safe no-op (idempotent).
            return

        assignment_id = row.id

        # Transition batch state with row lock (FOR UPDATE via the trigger path).
        await tx.execute(text("""
            UPDATE batches
               SET state='ASSIGNED', worker_id=:w,
                   lease_expires_at = now() + (:lease || ' seconds')::interval,
                   attempt = attempt + 1
             WHERE id=:b AND state IN ('PENDING','FAILED')
        """), {"w": worker_id, "lease": LEASE_SECONDS, "b": batch.id})

        # Audit IN THE SAME TX, idempotent on (key, to_state).
        await tx.execute(text("""
            INSERT INTO audit_log
              (batch_id, assignment_id, from_state, to_state,
               worker_id, idempotency_key, detail)
            VALUES (:b, :a, :frm, 'ASSIGNED', :w, :k, :d)
            ON CONFLICT (idempotency_key, to_state) DO NOTHING
        """), {"b": batch.id, "a": assignment_id, "frm": batch.state,
               "w": worker_id, "k": key, "d": json_dumps({"msg_id": msg_id})})
        # tx commits here. Only now do we XACK (in caller).

# Lease reaper (separate task): release expired in-flight work.
async def reap_expired():
    while not shutdown.is_set():
        async with engine.begin() as tx:
            stale = await tx.execute(text("""
                UPDATE batches SET state='PENDING', worker_id=NULL,
                                   lease_expires_at=NULL
                 WHERE state IN ('ASSIGNED','RUNNING')
                   AND lease_expires_at < now()
              RETURNING id, sample_id_canon
            """)).all()
            for b in stale:
                await tx.execute(text("""
                    UPDATE batch_assignments SET state='FAILED'
                     WHERE batch_id=:b AND state IN ('ASSIGNED','RUNNING')
                """), {"b": b.id})
        await asyncio.sleep(10)
```

Correctness argument:
- **Exactly-once effect**: the assignment INSERT is the commit point; redelivery hits `ON CONFLICT DO NOTHING` → no duplicate. XACK after commit ⇒ no lost work, at most safe reprocessing.
- **Atomic audit**: same transaction ⇒ assignment and audit are all-or-nothing.
- **Crash safety**: a crash before commit leaves the message pending → `xautoclaim` redelivers → idempotent replay. A crash after commit but before XACK → redelivery → no-op replay.

---

## 6) Migration & Rollout Plan (this week)

**Day 1 — Schema (online, backward-compatible)**
- Add new columns/tables with `IS NULL` defaults; do **not** drop anything yet.
- Backfill `sample_id_canon = normalize(sample_id_raw)` in batches of ~5k rows (`WHERE sample_id_canon IS NULL`) to avoid long locks.
- Detect existing twins *before* adding the unique index:
  ```sql
  SELECT sample_id_canon, count(*) FROM batches
  GROUP BY sample_id_canon HAVING count(*) > 1;
  ```
  Resolve duplicates manually, then `CREATE UNIQUE INDEX CONCURRENTLY`.

**Day 2 — Dual-write, read-old**
- Deploy code that writes idempotency keys + audit in the new atomic path **behind a feature flag** (`HELIX_IDEMPOTENT_ASSIGN`). Keep old reader for compatibility.
- Run reproduction suite (Section 2) against staging.

**Day 3 — Canary**
- Enable flag for 5% of workers (one consumer). Watch dashboards (Section 7) for `assign_conflict_total` (should be > 0 and rising = guard working) and zero `audit_missing_total`.

**Day 4 — Ramp**
- 25% → 50% → 100% as metrics stay clean for ≥2h at each step.

**Day 5 — Cleanup**
- Add the partial unique index and transition trigger if deferred.
- Remove old non-atomic code path; keep the flag as a kill-switch for one release.

**Rollback:** flip flag off → reverts to legacy path; new constraints are additive and don't break old writes (old path also writes canonical column via a small shim). Keep the `CREATE INDEX CONCURRENTLY` reversible with `DROP INDEX CONCURRENTLY`.

---

## 7) Observability Signals & Alerts

**Metrics (Prometheus naming)**
| Metric | Type | Meaning / Use |
|--------|------|----------------|
| `helix_assign_conflict_total` | counter | `ON CONFLICT DO NOTHING` hits = redelivery/dedup working. Sudden spike ⇒ excessive redelivery. |
| `helix_duplicate_active_assignments` | gauge | result of a periodic query for batches with >1 active assignment. **Must be 0.** |
| `helix_audit_missing_total` | counter | assignments lacking matching audit row (periodic reconciliation). **Alert on >0.** |
| `helix_stream_pending` | gauge | `XPENDING` count per consumer. Rising ⇒ stuck workers. |
| `helix_lease_reaped_total` | counter | leases reclaimed; correlates with worker restarts. |
| `helix_process_latency_seconds` | histogram | `process_one` duration. |
| `helix_id_normalization_changed_total` | counter | inputs where raw ≠ canon. Confirms international IDs are flowing. |
| `helix_xack_after_commit_failures` | counter | XACK failures post-commit (rare but causes safe reprocessing). |

**Reconciliation queries (run every 1–5 min)**
```sql
-- Duplicate active assignments (should always be empty)
SELECT batch_id FROM batch_assignments
 WHERE state IN ('ASSIGNED','RUNNING')
 GROUP BY batch_id HAVING count(*) > 1;

-- Assignment without audit
SELECT a.id FROM batch_assignments a
 LEFT JOIN audit_log l ON l.assignment_id = a.id
 WHERE l.id IS NULL;
```

**Alerts**
- `helix_duplicate_active_assignments > 0` → page (SEV2). Indicates constraint or logic regression.
- `helix_audit_missing_total > 0` (5m) → page.
- `helix_stream_pending` growing for >10m → warn (stuck consumer / poison message).
- `rate(helix_lease_reaped_total[5m])` abnormally high → warn (worker crash loop).
- Log every `ValueError` from `canon()` with the raw bytes (hex) for forensics; alert on rate spike (malformed IDs upstream).

---

## 8) Risk Register

| ID | Risk | Likelihood | Impact | Mitigation | Owner |
|----|------|-----------|--------|-----------|-------|
| R1 | Backfill reveals existing NFC/NFD duplicates blocking the unique index | High | Med | Pre-scan query Day 1; manual/automated merge before `CREATE UNIQUE INDEX CONCURRENTLY` | DBA |
| R2 | `attempt_epoch`/`stream_msg_id` not stable across legitimate resubmits → false dedup (work skipped) | Med | High | Key on stream message ID (stable per delivery); legitimately new work gets a new message ID. Add reconciliation alert for stalled PENDING batches | Backend |
| R3 | `XACK` after commit fails (Redis blip) → safe reprocessing but extra load | Med | Low | Idempotent replay handles it; metric `xack_after_commit_failures`; retry XACK with backoff | Backend |
| R4 | Transition trigger rejects a legitimate concurrent path, raising errors under load | Low | Med | Make UPD