PR1: Ingest correctness — durable per-record cursor + retry-dedup + run-id tiebreaker

CHANGELOG.md

@@ -0,0 +1,70 @@
+# Changelog
+
+All notable changes to the Context Intelligence Server are recorded here.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [7.0.0]
+
+### Changed (breaking)
+
+- **`QueueManager.commit()` now takes a required `cursor` argument** —
+  `commit(session_id, new_offset, cursor)`. The committed byte offset and the
+  cross-handler cursor are written in one atomic offset record so they can
+  never skew. This is a breaking change to the `QueueManager` Protocol that any
+  alternate backend must implement; all in-repo callers are updated. A legacy
+  bare-integer `.offset` written by an older build still reads its committed
+  position correctly, so an in-place upgrade is safe.
+- **`QueueManager` Protocol gains `read_cursor`** as a member every backend
+  must provide: it returns the persisted cursor a rebuilt worker resumes from.
+  Log-structured-queue details stay off the Protocol -- dead-line
+  reconciliation is a filesystem-backend concern (a broker backend has no
+  leading-dead-line window), so it is reached through the concrete backend, not
+  the Protocol.
+
+### Fixed
+
+- The durable cursor now survives every offset-mutation path that keeps the
+  session alive, not just `commit()`: idle compaction and the boot
+  `RESET_OFFSET` reclaim preserve a non-empty cursor instead of wiping it with
+  a bare offset write. Finalize's `delete_drained` still drops it, which is
+  terminal cleanup, not loss: the session is over.
+- Every write to a session's `.offset` now happens through one writer that
+  takes the session's file lock itself and stages through a per-write temp
+  file. Previously the lock was the caller's job, and the dead-letter
+  reconcile reached by `read_batch` took none -- a commit racing it was
+  silently rolled back to the offset the reconcile had read (measured: 248 of
+  300 concurrent runs), and the shared temp name let one writer publish
+  another's half-written record.
+- A corrupt/unparseable `.offset` file now quarantines the one affected drain
+  worker (logged, closed, deregistered) instead of crash-looping it; other
+  sessions keep draining. Every read in the drain loop is covered, including
+  the idle dry-exit recheck and the `session:end` tail drain, which previously
+  escaped to the supervisor.
+- Cross-handler run-id resolution is consistent across the orchestrator-run,
+  iteration, and content-block handlers, so a partial cursor after a worker
+  rebuild no longer drops the `HAS_PART` edge or orphans ContentBlock nodes.
+- `restore_cursor` deep-copies mutable cursor fields, so an in-place mutation on
+  a later retry attempt can no longer corrupt the pre-batch baseline the next
+  rollback restores.
+- A crash-then-respawn mid-isolation no longer re-dead-letters an
+  already-dead-lettered record: the drain worker reconciles the session's
+  leading dead lines on every (re)spawn, not only at boot.
+- Finalizing a session retires its dead letters out of the session's own name
+  instead of leaving them in place. A later session reusing the id no longer
+  reconciles against the previous session's dead payloads and commits past
+  events it never processed. The payloads are retained, still reported by
+  `GET /queues/dead-letter/{key}`, and still expire on their own schedule.
+- The Session node survives an exhausted-batch isolation: discarding the failed
+  batch's buffer now also invalidates the seen-session cache, so the isolated
+  re-dispatch re-issues the node instead of early-returning.
+
+## [6.7.3]
+
+### Added
+
+- Durable per-record cursor persisted atomically with the committed offset, so
+  a rebuilt drain worker resumes its cross-handler counters instead of
+  re-minting node ids. Retry-dedup and the run-id tiebreaker keep a re-delivered
+  batch idempotent.

context_intelligence_server/handlers/data_layer_2/content_block.py

@@ -49,11 +49,18 @@ async def __call__(self, event: str, data: dict[str, Any]) -> HookResult:
             return HookResult(action="continue")
 
         block_index = data.get("block_index")
-        iteration_id = self.services.data_layer_2.active_iteration_id
-        # ID format is "{session_id}::iteration::{n}"; [-1] extracts the iteration number.
-        # If the cursor format ever changes, this extraction must be updated to match.
-        iteration_n = iteration_id.split("::")[-1] if iteration_id else "0"
-        block_node_id = f"{session_id}::block::{iteration_n}::{block_index}"
+        # Key the block off the FULL active_iteration_id so a block inherits the
+        # iteration's run scope (including the run tiebreaker). Using only the
+        # trailing iteration number would let two runs that share an iteration
+        # number collide on the same block_node_id and MERGE-overwrite.
+        # Resolve the iteration id the same seq-aware way the Iteration handler
+        # does, so after a worker rebuild a block inherits the run's real
+        # iteration scope rather than the run-less fallback.
+        iteration_id = self.services.data_layer_2.resolve_active_iteration_id(
+            session_id
+        )
+        iteration_key = iteration_id if iteration_id else f"{session_id}::iteration::0"
+        block_node_id = f"{iteration_key}::block::{block_index}"
 
         if event == "content_block:start":
             await self._handle_start(session_id, block_node_id, block_index, data)

context_intelligence_server/handlers/data_layer_2/iteration.py

@@ -36,6 +36,18 @@ class IterationHandler:
     def __init__(self, services: HookStateService) -> None:
         self.services = services
 
+    def _current_iteration_scope(self) -> str:
+        """The 'run' | 'unscoped' discriminator, sourced from the SAME cursor
+        field (``execution_start_ts``) used to decide the node_id shape, so all
+        three upsert_node call sites (provider:request, llm:request,
+        llm:response) always agree.
+        """
+        return (
+            "run"
+            if self.services.data_layer_2.execution_start_ts is not None
+            else "unscoped"
+        )
+
     async def __call__(self, event: str, data: dict[str, Any]) -> HookResult:
         """Dispatch to the appropriate sub-handler.
 
@@ -64,22 +76,55 @@ async def _handle_provider_request(
     ) -> None:
         """Create Iteration node and set active_iteration_id cursor.
 
-        - Computes iteration_id as '{session_id}::iteration::{iteration_number}'
+        - Computes iteration_id as run-scoped, reusing the active run's full
+          orch_run_id (which carries the run tiebreaker):
+          '{orch_run_id}::iteration::{iteration_number}' when a run is active,
+          falling back to the bare '{session_id}::iteration::{iteration_number}'
+          shape when no run is active.
         - Sets active_iteration_id cursor on DataLayer2State
         - Creates Iteration:SST_EVENT node with session_id, iteration_number, started_at
-        - Conditionally creates E06: OrchestratorRun -[:HAS_PART {sst_semantic: 'CONTAINS'}]->
-          Iteration when execution_start_ts cursor is set
+        - Conditionally creates the OrchestratorRun -[:HAS_PART]-> Iteration edge
+          when a run is active.
+
+        Without run-scoping, iteration_number alone (a counter scoped to the whole
+        session, not the run) can repeat across orchestrator runs -- e.g. after a
+        drainer restart resets the in-memory counter -- causing distinct runs'
+        Iteration nodes to MERGE onto the same node_id and their usage figures to
+        clobber each other.
         """
         # Increment counter to get the next iteration number
         self.services.data_layer_2.iteration_count += 1
         iteration_number = self.services.data_layer_2.iteration_count
 
         timestamp: str = data.get("timestamp", "")
-        iteration_id = f"{session_id}::iteration::{iteration_number}"
+
+        # Resolve the run id the same seq-aware way every other handler does:
+        # after a worker rebuild active_orch_run_id may be absent while the run
+        # is still active, in which case this re-derives the run's real id (with
+        # its seq) rather than falling through to the seq-less, run-less shape.
+        orch_run_id = self.services.data_layer_2.resolve_active_orch_run_id(session_id)
+        if orch_run_id is not None:
+            iteration_id = f"{orch_run_id}::iteration::{iteration_number}"
+        else:
+            iteration_id = f"{session_id}::iteration::{iteration_number}"
 
         # Set cursor so llm:request and llm:response can find this iteration
         self.services.data_layer_2.active_iteration_id = iteration_id
 
+        # A queryable discriminator between a run-scoped iteration and a
+        # legitimate loop-basic session with no active orchestrator run.
+        iteration_scope = self._current_iteration_scope()
+        if iteration_scope == "unscoped":
+            # INFO not WARNING: a loop-basic session with no execution:start is
+            # a normal case, not an alert-worthy anomaly.
+            logger.info(
+                "unscoped_iteration_emitted session=%s iteration_number=%d iteration_id=%s",
+                session_id,
+                iteration_number,
+                iteration_id,
+                extra={"session_id": session_id},
+            )
+
         # Create the Iteration node
         await self.services.graph.upsert_node(
             iteration_id,
@@ -88,13 +133,12 @@ async def _handle_provider_request(
                 "session_id": session_id,
                 "iteration_number": iteration_number,
                 "started_at": timestamp,
+                "iteration_scope": iteration_scope,
             },
         )
 
         # E06 (conditional): OrchestratorRun -[:HAS_PART {sst_semantic: 'CONTAINS'}]-> Iteration
-        execution_start_ts = self.services.data_layer_2.execution_start_ts
-        if execution_start_ts is not None:
-            orch_run_id = f"{session_id}::orch_run::{execution_start_ts}"
+        if orch_run_id is not None:
             await self.services.graph.upsert_edge(
                 orch_run_id,
                 iteration_id,
@@ -127,6 +171,12 @@ async def _handle_llm_request(self, data: dict[str, Any]) -> None:
                 "model": data.get("model"),
                 "message_count": data.get("message_count"),
                 "has_system": data.get("has_system"),
+                # Stamp independently of provider:request's own write -- an
+                # Iteration node must never be created/updated without a scope
+                # value, even if this write is the first one
+                # to ever reach the node (e.g. a dead-lettered provider:request
+                # whose cursor mutation nonetheless survived).
+                "iteration_scope": self._current_iteration_scope(),
             },
         )
 
@@ -157,6 +207,9 @@ async def _handle_llm_response(self, data: dict[str, Any]) -> None:
                 "usage_input": usage.get("input_tokens"),
                 "usage_output": usage.get("output_tokens"),
                 "usage_cache_write": usage.get("cache_creation_input_tokens"),
+                # See _handle_llm_request -- same completeness rationale applies
+                # to this, the third of the three sites.
+                "iteration_scope": self._current_iteration_scope(),
             },
         )
 

context_intelligence_server/handlers/data_layer_2/orchestrator_run.py

@@ -62,23 +62,41 @@ async def __call__(self, event: str, data: dict[str, Any]) -> HookResult:
     # Sub-handlers
     # ------------------------------------------------------------------
 
+    def _resolve_run_id(self, session_id: str) -> str:
+        """Resolve the OrchestratorRun id an enrichment event must target.
+
+        Delegates to the shared seq-aware resolver so every handler agrees on
+        the id shape. The enrichment callers only reach here after guarding on
+        ``execution_start_ts is not None`` (a run is active), so the resolver
+        never returns None here.
+        """
+        run_id = self.services.data_layer_2.resolve_active_orch_run_id(session_id)
+        assert run_id is not None  # guarded by execution_start_ts on every caller
+        return run_id
+
     async def _handle_execution_start(
         self, session_id: str, data: dict[str, Any]
     ) -> None:
         """Create OrchestratorRun node and wire E01 (and optionally E14).
 
-        - Computes orch_run_id as '{session_id}::orch_run::{timestamp}'
+        - Computes orch_run_id as '{session_id}::orch_run::{timestamp}::{seq}'
         - Sets execution_start_ts cursor on DataLayer2State
         - Creates OrchestratorRun:SST_EVENT node with session_id + started_at
         - Creates E01: Session -[:HAS_EXECUTION {sst_semantic: 'CONTAINS'}]-> OrchestratorRun
         - Conditionally creates E14: Prompt -[:TRIGGERS {sst_semantic: 'LEADS_TO'}]->
           OrchestratorRun when last_prompt_id cursor is set
         """
         timestamp: str = data.get("timestamp", "")
-        orch_run_id = f"{session_id}::orch_run::{timestamp}"
-
-        # Store cursor so execution:end and orchestrator:complete can find this run
+        # Tiebreaker: two runs sharing an identical timestamp (coarse clock or a
+        # replayed execution:start) must not collide on the same orch_run_id.
+        self.services.data_layer_2.orch_run_seq += 1
+        seq = self.services.data_layer_2.orch_run_seq
+        orch_run_id = f"{session_id}::orch_run::{timestamp}::{seq}"
+
+        # Store cursors so execution:end, orchestrator:complete, and the
+        # Iteration/ContentBlock handlers all reuse this exact id.
         self.services.data_layer_2.execution_start_ts = timestamp
+        self.services.data_layer_2.active_orch_run_id = orch_run_id
 
         # Create the OrchestratorRun node
         await self.services.graph.upsert_node(
@@ -130,7 +148,7 @@ async def _handle_execution_end(
         if ts is None:
             return
 
-        orch_run_id = f"{session_id}::orch_run::{ts}"
+        orch_run_id = self._resolve_run_id(session_id)
         timestamp: str = data.get("timestamp", "")
 
         node_data: dict[str, Any] = {
@@ -167,7 +185,7 @@ async def _handle_orchestrator_complete(
         if ts is None:
             return
 
-        orch_run_id = f"{session_id}::orch_run::{ts}"
+        orch_run_id = self._resolve_run_id(session_id)
         timestamp: str = data.get("timestamp", "")
         orchestrator: str = data.get("orchestrator", "")
         turn_count: Any = data.get("turn_count")
@@ -213,3 +231,4 @@ async def _handle_orchestrator_complete(
         # Update cursors
         self.services.data_layer_2.last_completed_orch_run_id = orch_run_id
         self.services.data_layer_2.execution_start_ts = None
+        self.services.data_layer_2.active_orch_run_id = None

context_intelligence_server/handlers/data_layer_2/state.py

@@ -16,6 +16,19 @@ class DataLayer2State:
     # OrchestratorRun identity
     execution_start_ts: str | None = None
 
+    # Monotonic per-session run counter, incremented on each execution:start and
+    # folded into orch_run_id. Two runs that share an identical execution_start_ts
+    # (coarse clock, or a replayed execution:start) would otherwise collide on the
+    # same orch_run_id and MERGE-overwrite each other's nodes. Lives here so the
+    # durable cursor persists it across a worker rebuild.
+    orch_run_seq: int = 0
+
+    # The full orch_run_id of the active run, including the tiebreaker. Set at
+    # execution:start and read back by execution:end/orchestrator:complete and
+    # the Iteration/ContentBlock handlers, so every node in one run agrees on the
+    # same id without any site recomputing it from execution_start_ts alone.
+    active_orch_run_id: str | None = None
+
     # Iteration cursor read by ContentBlockHandler + ToolCallHandler
     active_iteration_id: str | None = None
 
@@ -28,6 +41,48 @@ class DataLayer2State:
     # E15 OrchestratorRun→Prompt turn-flow cursor
     last_completed_orch_run_id: str | None = None
 
-    # Iteration counter — incremented on each provider:request; used to compute
-    # iteration_id as '{session_id}::iteration::{iteration_count}'
+    # Iteration counter — incremented on each provider:request; combined with
+    # execution_start_ts (via IterationHandler) to compute the run-scoped
+    # iteration_id '{session_id}::orch_run::{execution_start_ts}::{seq}::iteration::{iteration_count}'
+    # (falls back to the bare '{session_id}::iteration::{iteration_count}' shape when no
+    # orchestrator run is active). Scoped per-session, not reset per run: uniqueness across
+    # runs comes from the orch_run_id prefix, not from this counter —
+    # resetting it would collide with ContentBlockHandler's block_node_id derivation, which
+    # keys solely off this counter's value, not the run.
     iteration_count: int = 0
+
+    def resolve_active_orch_run_id(self, session_id: str) -> str | None:
+        """Return the active run's orch_run_id, or None when no run is active.
+
+        Prefers the stored ``active_orch_run_id`` cursor. When that is absent
+        but a run IS active (``execution_start_ts`` set) — a partial cursor
+        after a worker rebuild — re-derives the same
+        ``{session_id}::orch_run::{execution_start_ts}::{seq}`` shape
+        ``execution:start`` produces, so every handler in one run agrees on the
+        id and the HAS_PART edge is never dropped onto a non-existent node.
+        Returns None only when no run is active (a loop-basic session).
+        """
+        if self.active_orch_run_id is not None:
+            return self.active_orch_run_id
+        if self.execution_start_ts is not None:
+            return (
+                f"{session_id}::orch_run::{self.execution_start_ts}"
+                f"::{self.orch_run_seq}"
+            )
+        return None
+
+    def resolve_active_iteration_id(self, session_id: str) -> str | None:
+        """Return the active iteration_id, or None when no iteration is active.
+
+        Prefers the stored ``active_iteration_id`` cursor. When that is absent
+        but a run+iteration is reconstructible (a partial cursor after a
+        rebuild), re-derives ``{orch_run_id}::iteration::{iteration_count}``
+        from the seq-aware run id, so a ContentBlock inherits the same run
+        scope every other handler agrees on rather than an orphan key.
+        """
+        if self.active_iteration_id is not None:
+            return self.active_iteration_id
+        run_id = self.resolve_active_orch_run_id(session_id)
+        if run_id is not None and self.iteration_count > 0:
+            return f"{run_id}::iteration::{self.iteration_count}"
+        return None