From 7fec35b445e6fe584b84e982b7fabce96e95bc35 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:50:44 -0400 Subject: [PATCH] Key context requests by the incarnation, not its name (ContextRequestKeyPhase) ContextRequest was indexed on (thread_name, thread_birth), and newest() filtered on the current name (IDEN-6). After a rename, latest() missed every earlier request, returned AbsentContext and restarted the turn counter at 1; paging fanned out over every alias in the registry to paper over it, then re-resolved each row's owner to check the fan-out. A recorded incarnation's identity is its birth: RecordedThreadIncarnation compares and hashes wire_birth, and its lookup is a hash of that birth. The table now keys on thread_birth alone; append and every read require a recorded incarnation. The thread_name column, the alias fan-out, the post-page owner recheck and the registry snapshot threaded into paging only to feed that fan-out are gone. Carry: tools/cutover/carry_context_request_key.py rebuilds the table through its declaration (the generated name column cannot be derived away), keeping every row and id. On a copy of the live store: 51355 rows, identical id/identity/versions/size digest before and after, second dry run reports 0. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/context_store.py | 57 ++++++++-------------- tools/cutover/carry_context_request_key.py | 46 +++++++++++++++++ 2 files changed, 65 insertions(+), 38 deletions(-) create mode 100644 tools/cutover/carry_context_request_key.py diff --git a/src/agent_comms/context_store.py b/src/agent_comms/context_store.py index 9ebb031928..4462a72683 100644 --- a/src/agent_comms/context_store.py +++ b/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 ) diff --git a/tools/cutover/carry_context_request_key.py b/tools/cutover/carry_context_request_key.py new file mode 100644 index 0000000000..eab80c62ec --- /dev/null +++ b/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,))