Key context requests by the incarnation, not its name (ContextRequestKeyPhase)

src/agent_comms/context_store.py

@@ -11,7 +11,7 @@
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import TYPE_CHECKING, Annotated, ClassVar, Literal, Self, cast
+from typing import Annotated, ClassVar, Literal, Self, cast
 
 from .errors import RelationViolationError, StoredHistoryUnreadable
 from .field_codec import (
@@ -35,8 +35,6 @@
     UnrecordedSegment,
 )
 
-if TYPE_CHECKING:
-    from .registry_document import RegistrySnapshot
 from .typed_table import Column, DurableStore, Generated, Index, TypedTable
 
 
@@ -62,7 +60,6 @@ def page(
         self,
         store: ContextStore,
         db: sqlite3.Connection,
-        snapshot: RegistrySnapshot,
         incarnation: ThreadIncarnation,
         cursor: ContextManifestCursor,
         previous: tuple[CapturedContext, ...],
@@ -85,7 +82,6 @@ def page(
         self,
         store: ContextStore,
         db: sqlite3.Connection,
-        snapshot: RegistrySnapshot,
         incarnation: ThreadIncarnation,
         cursor: ContextManifestCursor,
         previous: tuple[CapturedContext, ...],
@@ -120,14 +116,12 @@ def page(
         self,
         store: ContextStore,
         db: sqlite3.Connection,
-        snapshot: RegistrySnapshot,
         incarnation: ThreadIncarnation,
         cursor: ContextManifestCursor,
         previous: tuple[CapturedContext, ...],
         limit: int,
     ) -> ContextManifestPage:
-        owners = incarnation.recorded_names(snapshot)
-        where, parameters = cursor.selection(self.source, owners)
+        where, parameters = cursor.selection(self.source, incarnation)
         candidates = ContextRequest.select(
             db,
             where=where,
@@ -147,16 +141,13 @@ def page(
         older: BeforeContextManifest | LatestContextManifest = LATEST_CONTEXT
         newer: AfterContextManifest | LatestContextManifest = LATEST_CONTEXT
         if rows:
-            older = BeforeContextManifest(self.source, rows[0].id).available(db, owners)
-            newer = AfterContextManifest(self.source, rows[-1].id).available(db, owners)
+            older = BeforeContextManifest(self.source, rows[0].id).available(db, incarnation)
+            newer = AfterContextManifest(self.source, rows[-1].id).available(db, incarnation)
         retained = {item.record_id: item for item in previous if item.source == self.source}
         resources = tuple(
             retained[row.id] if row.id in retained else store.resolve(db, self.source, row)
             for row in rows
         )
-        for resource in resources:
-            if resource.value.thread.resolved(snapshot) != incarnation:
-                raise RelationViolationError("Context source differs from recorded owner")
         return ContextManifestPage(resources, older, newer)
 
 
@@ -210,35 +201,25 @@ def append(cls, db: sqlite3.Connection, segment: RecordedSegmentManifest) -> int
 class ContextRequest(TypedTable):
     id: Annotated[int, Column(primary_key=True, auto_increment=True)]
     identity: RecordedContextIdentity
-    thread_name: ClassVar[Generated] = Generated(
-        str,
-        Column(generated="json_extract(identity,'$.thread.name')"),
-    )
     thread_birth: ClassVar[Generated] = Generated(
         float,
         Column(generated="json_extract(identity,'$.thread.created_at')"),
     )
+    """The recorded incarnation's identity; its name is only a route."""
     versions: tuple[int, ...]
     size: int
-    indexes = (Index(("thread_name", "thread_birth", "id")),)
+    indexes = (Index(("thread_birth", "id")),)
 
     @classmethod
-    def owner_filter(
-        cls,
-        owners: tuple[ThreadIncarnation, ...],
-    ) -> tuple[str, tuple[object, ...]]:
-        return (
-            "(thread_name,thread_birth) IN (" + ",".join("(?,?)" for _ in owners) + ")",
-            tuple(value for owner in owners for value in (owner.name, owner.wire_birth)),
-        )
+    def owner_filter(cls, owner: ThreadIncarnation) -> tuple[str, tuple[object, ...]]:
+        return "thread_birth=?", (owner.require_recorded().created_at,)
 
     @classmethod
     def newest(cls, db: sqlite3.Connection, owner: ThreadIncarnation) -> list[ContextRequest]:
         """The thread's newest request, read through the owner index."""
         return cls.select(
             db,
-            thread_name=owner.name,
-            thread_birth=owner.wire_birth,
+            thread_birth=owner.require_recorded().created_at,
             order_by=("id",),
             descending=True,
             limit=1,
@@ -250,6 +231,7 @@ def append(cls, db: sqlite3.Connection, record: RecordedContextManifest) -> int:
         identity = FieldCodec.decode(
             RecordedContextIdentity, FieldCodec.project(record, "identity")
         )
+        identity.thread.require_recorded()
         return cast(
             int,
             cls.insert_values(
@@ -372,7 +354,7 @@ def has_position(self) -> bool: ...
 
     @abstractmethod
     def selection(
-        self, source: ContextStoreIdentity, owners: tuple[ThreadIncarnation, ...]
+        self, source: ContextStoreIdentity, owner: ThreadIncarnation
     ) -> tuple[str, tuple[object, ...]]: ...
 
 
@@ -382,9 +364,9 @@ def has_position(self) -> bool:
         return False
 
     def selection(
-        self, source: ContextStoreIdentity, owners: tuple[ThreadIncarnation, ...]
+        self, source: ContextStoreIdentity, owner: ThreadIncarnation
     ) -> tuple[str, tuple[object, ...]]:
-        return ContextRequest.owner_filter(owners)
+        return ContextRequest.owner_filter(owner)
 
 
 @dataclass(frozen=True)
@@ -398,19 +380,19 @@ def has_position(self) -> bool:
         return True
 
     def selection(
-        self, source: ContextStoreIdentity, owners: tuple[ThreadIncarnation, ...]
+        self, source: ContextStoreIdentity, owner: ThreadIncarnation
     ) -> tuple[str, tuple[object, ...]]:
         if self.source != source:
             raise ValueError("Recorded context page changed its source")
-        where, parameters = ContextRequest.owner_filter(owners)
+        where, parameters = ContextRequest.owner_filter(owner)
         return where + f" AND id {self.comparison} ?", (*parameters, self.record_id)
 
     def available(
         self,
         db: sqlite3.Connection,
-        owners: tuple[ThreadIncarnation, ...],
+        owner: ThreadIncarnation,
     ) -> Self | LatestContextManifest:
-        where, parameters = ContextRequest.owner_filter(owners)
+        where, parameters = ContextRequest.owner_filter(owner)
         return (
             self
             if ContextRequest.exists(
@@ -559,10 +541,9 @@ def page(
     ) -> ContextManifestPage:
         if not 1 <= limit <= self.max_page_rows:
             raise ValueError("Context page row limit exceeds its source budget")
-        snapshot = registry.snapshot()
-        incarnation = snapshot.require(name).incarnation
+        incarnation = registry.snapshot().require(name).incarnation
         selection = ContextManifestCursor.for_page(cursor)
         with self.snapshot() as db:
             return ContextStoreInfo.source_in(db).page(
-                self, db, snapshot, incarnation, selection, previous, limit
+                self, db, incarnation, selection, previous, limit
             )

tools/cutover/carry_context_request_key.py

@@ -0,0 +1,46 @@
+"""Rebuild the context request table on its incarnation key.
+
+Delete after every bus has run this step. The retired name column and its
+index are generated from the stored identity, so the rebuild keeps every row
+and its id; the declared table owns the new key.
+"""
+
+from __future__ import annotations
+
+from contextlib import closing
+from pathlib import Path
+
+from agent_comms.context_store import ContextRequest, ContextStore
+from agent_comms.installer.carry import Carried, Carry
+from agent_comms.locked_store import SQLiteFile
+from agent_comms.store_files import _store_lock
+from agent_comms.typed_table import SQLiteSchemaObject, TypedTable
+
+
+class ContextRequestKey(Carry):
+    @property
+    def path(self) -> Path:
+        return self.root / ContextStore.filename
+
+    def backup_files(self) -> tuple[Path, ...]:
+        self.require_confined(self.root, self.path)
+        return (self.path,) if self.path.exists() else ()
+
+    def carry(self, *, apply: bool) -> Carried:
+        if not self.path.exists():
+            return Carried(0)
+        with _store_lock(self.path), closing(
+            ContextStore.writer(self.path, isolation_level=None)
+        ) as db:
+            if not SQLiteSchemaObject.differences(
+                TypedTable.objects_of(db, (ContextRequest,)), ContextRequest.schema_objects()
+            ):
+                return Carried(0)
+            (rows,) = db.execute(f'SELECT count(*) FROM "{ContextRequest.declared_name}"').fetchone()
+            if not apply:
+                return Carried(1, {"context_requests": rows})
+            ContextStore.schema_change(self.path).require()
+            with SQLiteFile.write_transaction(db):
+                ContextRequest.rebuild(db)
+                ContextStore.require_schema(db)
+            return Carried(1, {"context_requests": rows}, (ContextStore.filename,))