fix: hash chain concurrency protection (v1.5.0)

Prevents race conditions where concurrent seal_and_store() calls could
fork the hash chain by claiming the same sequence number.

Defense in depth:
- UNIQUE(sequence) constraint on SQLite CapsuleModel
- UNIQUE(tenant_id, sequence) on PostgreSQL CapsuleModelPG
- Partial unique index for global chain (tenant_id IS NULL) on PG only
- Optimistic retry in seal_and_store() (3 attempts, then ChainConflictError)
- _is_integrity_error() detects violations across wrapper layers

New: ChainConflictError exception, 36 concurrency tests.
No public API changes. Backwards compatible.

Author: bradleygauthier

CHANGELOG.md

@@ -11,6 +11,32 @@ Versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ---
 
+## [1.5.0] - 2026-03-15
+
+Hash chain concurrency protection. Prevents race conditions where concurrent writes could fork the chain.
+
+### Added
+
+- **Optimistic retry in `seal_and_store()`** -- if a concurrent writer claims the same sequence number, the UNIQUE constraint rejects the duplicate and the method retries with the updated chain head. Up to 3 retries before raising `ChainConflictError`. Seal fields (hash, signature, metadata) are properly reset between attempts.
+- **`ChainConflictError` exception** -- raised when `seal_and_store()` exhausts all retries due to sustained concurrent writes for the same tenant chain. Subclass of `ChainError`; includes tenant ID in the message.
+- **`UNIQUE` constraint on sequence (SQLite)** -- `CapsuleModel` now enforces `UNIQUE(sequence)`, preventing duplicate sequence numbers at the database level. Defense-in-depth: even if application logic fails, the database rejects duplicates.
+- **`UNIQUE` constraint on tenant + sequence (PostgreSQL)** -- `CapsuleModelPG` now enforces `UNIQUE(tenant_id, sequence)`, scoped per tenant. Two tenants can independently have sequence 0, but the same tenant cannot have two capsules at the same sequence.
+- **Global chain protection (PostgreSQL)** -- a DDL event creates `CREATE UNIQUE INDEX ... WHERE tenant_id IS NULL` exclusively on PostgreSQL, preventing duplicate sequences in the global chain (tenant_id=NULL). Fires via `execute_if(dialect="postgresql")`; does not affect SQLite.
+- **`_is_integrity_error()` helper** -- detects `IntegrityError` and `UniqueViolationError` exceptions even when wrapped in `StorageError.__cause__` chains. Supports SQLAlchemy, asyncpg, and aiosqlite error types.
+- **36 new concurrency tests** (`test_chain_concurrency.py`) -- exception hierarchy (3), integrity error detection (6), retry behavior with mocks (7), UNIQUE constraint enforcement (2), end-to-end integration (6), model constraint verification (3), retry invariants including capsule identity preservation and warning emission (4), exception hierarchy completeness (3), CLI `_get_version()` (2).
+
+### Security
+
+- **TOCTOU race condition fixed** -- the `add()` → `seal()` → `store()` sequence previously had a time-of-check-time-of-use vulnerability where two concurrent writers could both read the same chain head and both store capsules with the same sequence number, silently forking the hash chain. The UNIQUE constraint converts this silent corruption into a detectable conflict, and the retry loop resolves it automatically.
+- **Defense in depth** -- the fix operates at two layers: database constraints (cannot be bypassed by application bugs) and application-level retry (handles the race transparently). Non-integrity errors (disk failures, network issues) propagate immediately without retry.
+
+### Migration
+
+- **New installations**: constraints are created automatically by `create_all()`.
+- **Existing databases**: run `ALTER TABLE capsules ADD CONSTRAINT uq_capsule_sequence UNIQUE (sequence)` (SQLite) or `ALTER TABLE quantumpipes_capsules ADD CONSTRAINT uq_capsule_tenant_sequence UNIQUE (tenant_id, sequence)` (PostgreSQL). If the table already contains duplicate sequences from a prior race condition, resolve duplicates before adding the constraint.
+
+---
+
 ## [1.4.0] - 2026-03-15
 
 Ecosystem expansion: Go verifier, LiteLLM integration, and negative conformance vectors.

README.md

@@ -96,6 +96,34 @@ Capsule #0          Capsule #1          Capsule #2
 
 No consensus mechanism. No distributed ledger. SHA3-256 hashes linking one record to the next.
 
+### Concurrency Protection (v1.5.0+)
+
+Concurrent writes to the same chain are handled automatically by `seal_and_store()`:
+
+1. **UNIQUE constraint** — the database rejects duplicate sequence numbers (`UNIQUE(sequence)` on SQLite, `UNIQUE(tenant_id, sequence)` on PostgreSQL)
+2. **Optimistic retry** — on conflict, `seal_and_store()` re-reads the chain head, recomputes the sequence, re-seals, and retries (up to 3 attempts)
+3. **`ChainConflictError`** — raised if all retries are exhausted under extreme contention
+
+```python
+from qp_capsule import Capsule, CapsuleChain, CapsuleStorage, Seal
+from qp_capsule.exceptions import ChainConflictError
+
+storage = CapsuleStorage()
+chain = CapsuleChain(storage)
+seal = Seal()
+
+# Safe for concurrent use — retries automatically on sequence conflict
+capsule = await chain.seal_and_store(
+    Capsule(trigger=TriggerSection(source="agent", request="deploy")),
+    seal=seal,
+    tenant_id="org-123",
+)
+
+# Multi-tenant: each tenant has an independent chain
+await chain.seal_and_store(capsule_a, seal=seal, tenant_id="tenant-a")
+await chain.seal_and_store(capsule_b, seal=seal, tenant_id="tenant-b")
+```
+
 ---
 
 ## Cryptographic Seal

reference/python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "qp-capsule"
-version = "1.4.0"
+version = "1.5.0"
 description = "Capsule Protocol Specification (CPS) — tamper-evident audit records for AI operations. Create, seal, verify, and chain Capsules in Python."
 readme = "README.md"
 license = "Apache-2.0"

reference/python/src/qp_capsule/__init__.py

@@ -36,7 +36,7 @@
 Spec: https://github.com/quantumpipes/capsule
 """
 
-__version__ = "1.4.0"
+__version__ = "1.5.0"
 __author__ = "Quantum Pipes Technologies, LLC"
 __license__ = "Apache-2.0"
 
@@ -56,6 +56,7 @@
 )
 from qp_capsule.exceptions import (
     CapsuleError,
+    ChainConflictError,
     ChainError,
     KeyringError,
     SealError,
@@ -109,6 +110,7 @@
     "CapsuleError",
     "SealError",
     "ChainError",
+    "ChainConflictError",
     "StorageError",
     "KeyringError",
     # High-Level API

reference/python/src/qp_capsule/chain.py

@@ -16,20 +16,31 @@
     - Integrity verification: Anyone can verify the chain
 
 The chain is the memory of the system made immutable.
+
+Concurrency:
+    seal_and_store() uses optimistic retry — if a concurrent writer claims
+    the same sequence number, the UNIQUE constraint rejects the duplicate
+    and the method retries with the updated chain head.
 """
 
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from qp_capsule.exceptions import ChainConflictError
 from qp_capsule.seal import compute_hash
 
 if TYPE_CHECKING:
     from qp_capsule.capsule import Capsule
     from qp_capsule.protocol import CapsuleStorageProtocol
     from qp_capsule.seal import Seal
 
+_log = logging.getLogger(__name__)
+
+_MAX_CHAIN_RETRIES = 3
+
 
 @dataclass
 class ChainVerificationResult:
@@ -238,10 +249,57 @@ async def seal_and_store(
         seal: Seal | None = None,
         tenant_id: str | None = None,
     ) -> Capsule:
-        """Chain, seal, and store a capsule in one call."""
+        """
+        Chain, seal, and store a Capsule in one call.
+
+        Uses optimistic retry to handle concurrent writes: if another writer
+        claims the same sequence number, the UNIQUE constraint on the storage
+        backend rejects the duplicate and this method retries with the updated
+        chain head. Retries up to ``_MAX_CHAIN_RETRIES`` times.
+
+        Raises:
+            ChainConflictError: If all retries are exhausted.
+        """
         from qp_capsule.seal import Seal as SealCls
 
-        capsule = await self.add(capsule, tenant_id=tenant_id)
         seal_instance = seal or SealCls()
-        capsule = seal_instance.seal(capsule)
-        return await self.storage.store(capsule, tenant_id=tenant_id)
+
+        for attempt in range(_MAX_CHAIN_RETRIES):
+            capsule = await self.add(capsule, tenant_id=tenant_id)
+            capsule = seal_instance.seal(capsule)
+
+            try:
+                return await self.storage.store(capsule, tenant_id=tenant_id)
+            except Exception as exc:
+                if not _is_integrity_error(exc):
+                    raise
+
+                _log.warning(
+                    "Chain sequence conflict (attempt %d/%d, seq=%d, tenant=%s), retrying",
+                    attempt + 1,
+                    _MAX_CHAIN_RETRIES,
+                    capsule.sequence,
+                    tenant_id,
+                )
+                capsule.hash = ""
+                capsule.signature = ""
+                capsule.signature_pq = ""
+                capsule.signed_at = None
+                capsule.signed_by = ""
+
+        raise ChainConflictError(
+            f"Failed to add Capsule to chain after {_MAX_CHAIN_RETRIES} retries "
+            f"(concurrent writes for tenant={tenant_id!r})"
+        )
+
+
+def _is_integrity_error(exc: BaseException) -> bool:
+    """Check if an exception is a database integrity/unique-constraint violation."""
+    from qp_capsule.exceptions import StorageError
+
+    name = type(exc).__name__
+    if name in ("IntegrityError", "UniqueViolationError"):
+        return True
+    if isinstance(exc, StorageError) and exc.__cause__ is not None:
+        return _is_integrity_error(exc.__cause__)
+    return False

reference/python/src/qp_capsule/cli.py

@@ -41,6 +41,11 @@
 _NO_COLOR = False
 
 
+def _get_version() -> str:
+    from qp_capsule import __version__
+    return __version__
+
+
 # ---------------------------------------------------------------------------
 # ANSI helpers (stdlib only -- no rich, no click)
 # ---------------------------------------------------------------------------
@@ -563,7 +568,7 @@ def _build_parser() -> argparse.ArgumentParser:
     )
     parser.add_argument(
         "--version", action="version",
-        version=f"capsule {__import__('qp_capsule').__version__}",
+        version=f"capsule {_get_version()}",
     )
     sub = parser.add_subparsers(dest="command")
 

reference/python/src/qp_capsule/exceptions.py

@@ -21,6 +21,10 @@ class ChainError(CapsuleError):
     """Hash chain integrity error."""
 
 
+class ChainConflictError(ChainError):
+    """Concurrent write detected — sequence conflict after max retries."""
+
+
 class StorageError(CapsuleError):
     """Capsule storage operation failed."""
 

reference/python/src/qp_capsule/storage.py

@@ -25,7 +25,7 @@
 from typing import Any
 from uuid import UUID
 
-from sqlalchemy import Integer, String, Text, desc, func, select
+from sqlalchemy import Integer, String, Text, UniqueConstraint, desc, func, select
 from sqlalchemy.ext.asyncio import (
     AsyncEngine,
     AsyncSession,
@@ -53,6 +53,9 @@ class CapsuleModel(Base):
     """
 
     __tablename__ = "capsules"
+    __table_args__ = (
+        UniqueConstraint("sequence", name="uq_capsule_sequence"),
+    )
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True)
     type: Mapped[str] = mapped_column(String(20))

reference/python/src/qp_capsule/storage_pg.py

@@ -27,7 +27,7 @@
 from datetime import datetime
 from uuid import UUID
 
-from sqlalchemy import Integer, String, Text, desc, select
+from sqlalchemy import DDL, Integer, String, Text, UniqueConstraint, desc, event, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
@@ -49,6 +49,9 @@ class CapsuleModelPG(PGBase):
     """
 
     __tablename__ = "quantumpipes_capsules"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "sequence", name="uq_capsule_tenant_sequence"),
+    )
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True)
     type: Mapped[str] = mapped_column(String(20))
@@ -65,6 +68,16 @@ class CapsuleModelPG(PGBase):
     tenant_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
 
 
+event.listen(
+    CapsuleModelPG.__table__,
+    "after_create",
+    DDL(
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_capsule_global_sequence "
+        "ON quantumpipes_capsules (sequence) WHERE tenant_id IS NULL"
+    ).execute_if(dialect="postgresql"),
+)
+
+
 class PostgresCapsuleStorage:
     """
     Capsule storage using PostgreSQL.