fix(sdk): serialize chain reserve, sign, and persist

record_chained_event holds the outbox lock from chain read
through signing to commit so concurrent workers cannot reuse
the same chain position.

Signed-off-by: Nathan Gillett <nathan@intentproof.io>

Author: Nathan Gillett

src/intentproof/instrumentation.py

@@ -12,7 +12,7 @@
 import ulid as _ulid
 
 from intentproof import client
-from intentproof.signing import SENTINEL_PREV_HASH, event_content_hash, sign_event
+from intentproof.signing import event_content_hash, sign_event
 
 _correlation_id: ContextVar[str | None] = ContextVar(
     "intentproof_correlation_id", default=None
@@ -71,43 +71,35 @@ def _record_execution(
     error_obj: dict[str, Any] | None,
 ) -> None:
     outbox = client.get_outbox()
-    chain_pos = 1
-    prev_hash = SENTINEL_PREV_HASH
-    state = outbox.get_chain_state(correlation_id)
-    if state:
-        chain_pos = state["position"] + 1
-        prev_hash = state["hash"]
-
-    event: dict[str, Any] = {
-        "schema": "intentproof.event.v1",
-        "event_id": event_id,
-        "tenant_id": client.get_tenant_id(),
-        "instance_id": client.get_instance_id(),
-        "correlation_id": correlation_id,
-        "provenance_class": "sdk_attested_evidence",
-        "prev_event_hash": prev_hash,
-        "chain_position": chain_pos,
-        "intent": intent,
-        "action": action,
-        "status": status,
-        "started_at": _iso_timestamp(t0_ms),
-        "completed_at": _iso_timestamp(t1_ms),
-        "duration_ms": t1_ms - t0_ms,
-        "inputs": inputs,
-        "output": output if status == "ok" else None,
-        "error": error_obj,
-        "attributes": {},
-        "untrusted_payload": _untrusted_payload(inputs, output, status),
-        "spec_version": "1.0.0",
-        "sdk_version": client.SDK_VERSION,
-    }
-
-    signed = sign_event(event, client.get_private_key(), client.get_instance_id())
-    event_hash = event_content_hash(signed)
-
-    outbox.append_with_chain_state(
-        event_id, signed, correlation_id, chain_pos, event_hash
-    )
+
+    def build_signed(chain_pos: int, prev_hash: str) -> tuple[dict[str, Any], str]:
+        event: dict[str, Any] = {
+            "schema": "intentproof.event.v1",
+            "event_id": event_id,
+            "tenant_id": client.get_tenant_id(),
+            "instance_id": client.get_instance_id(),
+            "correlation_id": correlation_id,
+            "provenance_class": "sdk_attested_evidence",
+            "prev_event_hash": prev_hash,
+            "chain_position": chain_pos,
+            "intent": intent,
+            "action": action,
+            "status": status,
+            "started_at": _iso_timestamp(t0_ms),
+            "completed_at": _iso_timestamp(t1_ms),
+            "duration_ms": t1_ms - t0_ms,
+            "inputs": inputs,
+            "output": output if status == "ok" else None,
+            "error": error_obj,
+            "attributes": {},
+            "untrusted_payload": _untrusted_payload(inputs, output, status),
+            "spec_version": "1.0.0",
+            "sdk_version": client.SDK_VERSION,
+        }
+        signed = sign_event(event, client.get_private_key(), client.get_instance_id())
+        return signed, event_content_hash(signed)
+
+    signed = outbox.record_chained_event(correlation_id, event_id, build_signed)
 
     exporter = client.get_exporter()
     if exporter is not None:

src/intentproof/outbox.py

@@ -5,8 +5,11 @@
 import json
 import sqlite3
 import threading
+from collections.abc import Callable
 from typing import Any
 
+from intentproof.signing import SENTINEL_PREV_HASH
+
 
 class Outbox:
     def __init__(self, db_path: str) -> None:
@@ -38,6 +41,58 @@ def append(self, event_id: str, body: dict[str, Any]) -> None:
             )
             self._db.commit()
 
+    def _next_chain_link(self, correlation_id: str) -> tuple[int, str]:
+        row = self._db.execute(
+            "SELECT last_position, last_hash FROM chains WHERE correlation_id = ?",
+            (correlation_id,),
+        ).fetchone()
+        if row is None:
+            return 1, SENTINEL_PREV_HASH
+        return row[0] + 1, row[1]
+
+    def _persist_chain_event(
+        self,
+        event_id: str,
+        body: dict[str, Any],
+        correlation_id: str,
+        position: int,
+        event_hash: str,
+    ) -> None:
+        self._db.execute(
+            "INSERT INTO events (event_id, body) VALUES (?, ?)",
+            (event_id, json.dumps(body)),
+        )
+        self._db.execute(
+            """
+            INSERT INTO chains (correlation_id, last_position, last_hash)
+            VALUES (?, ?, ?)
+            ON CONFLICT(correlation_id) DO UPDATE SET
+                last_position = excluded.last_position,
+                last_hash = excluded.last_hash
+            """,
+            (correlation_id, position, event_hash),
+        )
+
+    def record_chained_event(
+        self,
+        correlation_id: str,
+        event_id: str,
+        build_signed: Callable[[int, str], tuple[dict[str, Any], str]],
+    ) -> dict[str, Any]:
+        """Reserve chain slot, sign, and persist under one lock."""
+        with self._lock:
+            chain_pos, prev_hash = self._next_chain_link(correlation_id)
+            signed, event_hash = build_signed(chain_pos, prev_hash)
+            try:
+                self._persist_chain_event(
+                    event_id, signed, correlation_id, chain_pos, event_hash
+                )
+                self._db.commit()
+            except Exception:
+                self._db.rollback()
+                raise
+            return signed
+
     def append_with_chain_state(
         self,
         event_id: str,
@@ -49,19 +104,8 @@ def append_with_chain_state(
         """Persist event and chain head in one transaction."""
         with self._lock:
             try:
-                self._db.execute(
-                    "INSERT INTO events (event_id, body) VALUES (?, ?)",
-                    (event_id, json.dumps(body)),
-                )
-                self._db.execute(
-                    """
-                    INSERT INTO chains (correlation_id, last_position, last_hash)
-                    VALUES (?, ?, ?)
-                    ON CONFLICT(correlation_id) DO UPDATE SET
-                        last_position = excluded.last_position,
-                        last_hash = excluded.last_hash
-                    """,
-                    (correlation_id, position, event_hash),
+                self._persist_chain_event(
+                    event_id, body, correlation_id, position, event_hash
                 )
                 self._db.commit()
             except Exception:

tests/test_sdk.py

@@ -79,6 +79,48 @@ def test_produces_signed_event_with_sentinel_prev_hash(
     assert ev["untrusted_payload"] is True
 
 
+def test_concurrent_wrap_preserves_chain_positions(
+    sdk_dirs: tuple[str, str],
+) -> None:
+    from intentproof.signing import event_content_hash
+
+    db_path, data_dir = sdk_dirs
+    configure(db_path=db_path, data_dir=data_dir, tenant_id="tnt_a")
+    fn = wrap(intent="Test", action="test.action", fn=lambda x: x)
+    errors: list[BaseException] = []
+    barrier = threading.Barrier(2)
+
+    def worker() -> None:
+        try:
+            barrier.wait(timeout=2.0)
+            for _ in range(25):
+                run_with_correlation_id("corr-race", lambda: fn(1))
+        except BaseException as exc:
+            errors.append(exc)
+
+    threads = [threading.Thread(target=worker) for _ in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=5.0)
+
+    assert not errors
+    events = [
+        e
+        for e in client.get_outbox().get_events()
+        if e["correlation_id"] == "corr-race"
+    ]
+    assert len(events) == 50
+    positions = sorted(e["chain_position"] for e in events)
+    assert positions == list(range(1, 51))
+
+    by_pos = {e["chain_position"]: e for e in events}
+    assert by_pos[1]["prev_event_hash"].startswith("sha256:")
+    for pos in range(2, 51):
+        prev_hash = event_content_hash(by_pos[pos - 1])
+        assert by_pos[pos]["prev_event_hash"] == prev_hash
+
+
 def test_wrap_records_from_worker_thread(sdk_dirs: tuple[str, str]) -> None:
     db_path, data_dir = sdk_dirs
     configure(db_path=db_path, data_dir=data_dir, tenant_id="tnt_a")