From 18c4c169d88c72ef1fa1ec2914fcf4de29419db3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:35:17 -0400 Subject: [PATCH 1/7] Own WITHOUT ROWID once on the table capability --- src/agent_comms/cohort_schema.py | 10 ++++------ src/agent_comms/coordination_response.py | 5 ++--- src/agent_comms/native_pi.py | 5 ++--- src/agent_comms/native_prompt_binding.py | 5 ++--- src/agent_comms/native_runtime_input.py | 8 +++----- src/agent_comms/transcript_routes.py | 8 +++----- src/agent_comms/triage_native_sources.py | 5 ++--- src/agent_comms/typed_table.py | 6 ++++++ src/agent_comms/view_unread.py | 13 ++++++++++--- 9 files changed, 34 insertions(+), 31 deletions(-) diff --git a/src/agent_comms/cohort_schema.py b/src/agent_comms/cohort_schema.py index e4954330f9..7116e8b932 100644 --- a/src/agent_comms/cohort_schema.py +++ b/src/agent_comms/cohort_schema.py @@ -32,6 +32,7 @@ NoDelete, SharedFileTables, TypedTable, + WithoutRowid, ) if TYPE_CHECKING: @@ -75,7 +76,7 @@ class ClaimBatchAcknowledgements( GuardedColumns, NoDelete, CohortTable, - TypedTable, + WithoutRowid, declared_name="claim_batch_receipts", ): transition_column = "sealed" @@ -135,7 +136,6 @@ class ClaimBatchAcknowledgements( policy_version: Annotated[str, FrozenColumn(check="length(policy_version) BETWEEN 1 AND 256")] accepted_at_ms: Annotated[int, FrozenColumn(check="accepted_at_ms >= 0")] sealed: bool = False - without_rowid = True @classmethod def decision_hash(cls, delivery: Delivered) -> str: @@ -177,7 +177,7 @@ def require_selected_claim(self, assignment: WakeAssignment) -> None: raise ProjectionUnavailableError("selected source acknowledgement is unsealed or changed") @dataclass(frozen=True, kw_only=True) -class ClaimBatchMembers(GuardedInserts, ImmutableRows, CohortTable, TypedTable): +class ClaimBatchMembers(GuardedInserts, ImmutableRows, CohortTable, WithoutRowid): insert_condition = """((SELECT sealed FROM claim_batch_receipts r WHERE r.wire_root_id = NEW.wire_root_id AND r.wire_seq = NEW.wire_seq) != 0) @@ -198,7 +198,6 @@ class ClaimBatchMembers(GuardedInserts, ImmutableRows, CohortTable, TypedTable): ("claim_id",), ("wire_root_id", "wire_seq", "recipient_lookup"), ) - without_rowid = True @classmethod def references(cls): @@ -216,7 +215,7 @@ class CohortDeliveries( GuardedInserts, ImmutableRows, CohortTable, - TypedTable, + WithoutRowid, declared_name="cohort_delivery_receipts", ): """Sealed deliveries, including observers that receive no execution claim.""" @@ -255,7 +254,6 @@ class CohortDeliveries( ("wire_root_id", "wire_seq", "claim_id"), ) indexes = (Index(("recipient_lookup", "wire_seq")),) - without_rowid = True @classmethod def sealed_sequences_after( diff --git a/src/agent_comms/coordination_response.py b/src/agent_comms/coordination_response.py index 2da5b96167..bfcb3d527a 100644 --- a/src/agent_comms/coordination_response.py +++ b/src/agent_comms/coordination_response.py @@ -53,7 +53,7 @@ ImmutableRows, SqlValue, TypedRow, - TypedTable, + WithoutRowid, ) from .locked_store import RegistryFile @@ -65,7 +65,7 @@ class ResponseTable(ImmutableRows, PrivateCoordinatorTable): schema_error = PublicationActivationBlocked @dataclass(frozen=True) -class PublicationAppendDispatches(ResponseTable, TypedTable): +class PublicationAppendDispatches(ResponseTable, WithoutRowid): execution_id: Annotated[str, Column(primary_key=True)] exact_target: Annotated[ str, Column(primary_key=True, check="length(exact_target) BETWEEN 1 AND 256") @@ -76,7 +76,6 @@ class PublicationAppendDispatches(ResponseTable, TypedTable): owner_generation: Annotated[int, Column(check="owner_generation>0")] attempt_ordinal: Annotated[int, Column(check="attempt_ordinal>0")] dispatched_at_ms: Annotated[int, Column(check="dispatched_at_ms>=0")] - without_rowid = True @classmethod def references(cls): diff --git a/src/agent_comms/native_pi.py b/src/agent_comms/native_pi.py index f3de4c8b32..1586f90212 100644 --- a/src/agent_comms/native_pi.py +++ b/src/agent_comms/native_pi.py @@ -50,7 +50,7 @@ GuardedInserts, ImmutableRows, Index, - TypedTable, + WithoutRowid, ) if TYPE_CHECKING: @@ -215,7 +215,7 @@ def at(self, session_file: Path, *, session_entry_start: int) -> NativeContextPr @dataclass(frozen=True, slots=True) class NativeContextJournal( - GuardedInserts, ImmutableRows, NativeContextRecord, TypedTable, DurableTables + GuardedInserts, ImmutableRows, NativeContextRecord, WithoutRowid, DurableTables ): """One proof declaration owns wire facts and the indexed durable journal.""" @@ -230,7 +230,6 @@ class NativeContextJournal( schema: Literal[2] type: Literal["context_committed"] - without_rowid = True indexes = (Index(("request_generation",)), Index(("session_id",))) @classmethod diff --git a/src/agent_comms/native_prompt_binding.py b/src/agent_comms/native_prompt_binding.py index 92ccba6312..7b59afb3a6 100644 --- a/src/agent_comms/native_prompt_binding.py +++ b/src/agent_comms/native_prompt_binding.py @@ -38,7 +38,7 @@ IntegerStorage, TextStorage, TypedRow, - TypedTable, + WithoutRowid, ) if TYPE_CHECKING: @@ -52,7 +52,7 @@ def binding_store_path(store: Coordination) -> Path: @dataclass(frozen=True, slots=True) class PromptBinding( - ImmutableRows, DurableFile, SQLiteFile, NativeInputRecord, TypedTable, DurableTables + ImmutableRows, DurableFile, SQLiteFile, NativeInputRecord, WithoutRowid, DurableTables ): filename = "native_prompt_bindings.sqlite3" @@ -81,7 +81,6 @@ def execution(self) -> NativeInputExecution: except MalformedRecord as error: raise IdentityConflict("Prompt binding execution columns conflict with their stage") from error - without_rowid = True checks = ( "(stage='triage' AND execution_id IS NULL AND attempt_ordinal IS NULL) OR " "(stage='full' AND execution_id IS NOT NULL AND attempt_ordinal>0)", diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py index 33d696970e..90974227e8 100644 --- a/src/agent_comms/native_runtime_input.py +++ b/src/agent_comms/native_runtime_input.py @@ -52,7 +52,7 @@ StableReferenceColumns, TextStorage, TypedRow, - TypedTable, + WithoutRowid, WriteOnceColumns, sql_literal, ) @@ -86,7 +86,7 @@ class NativeRuntimeInput( NativeInputRecord, NativeInputContext, NativeRuntimeTable, - TypedTable, + WithoutRowid, ): @classmethod def notification_context_sql(cls) -> str: @@ -288,7 +288,6 @@ def published_replies(self, db, user, owner_lookup): Column(check=RecordedTriageText.sql_constraint("verdict"), storage=TextStorage, nullable=True), ] = AbsentTriageDecisionRecord() - without_rowid = True unique = (("execution_id", "attempt_ordinal"),) checks = ( "(stage='triage' AND execution_id IS NULL AND attempt_ordinal IS NULL) OR " @@ -342,7 +341,7 @@ class CurrentNativeCursor( NoDelete, NativeInputContext, NativeRuntimeTable, - TypedTable, + WithoutRowid, ): column_orders = (NonDecreasing(("covered_seq", "injected_seq")),) reference_column = 'injected_seq' @@ -395,7 +394,6 @@ def owner_identity(self): session_id: Annotated[NativeColumn, NativeContext(str), Column(storage=TextStorage, nullable=True)] request_generation: Annotated[NativeColumn, NativeContext(int), Column(storage=IntegerStorage, nullable=True)] - without_rowid = True checks = ( "(injected_seq=0 AND input_id IS NULL AND stage IS NULL " "AND session_id IS NULL AND request_generation IS NULL) OR " diff --git a/src/agent_comms/transcript_routes.py b/src/agent_comms/transcript_routes.py index 24fcc56e6f..b8dc3a4bf2 100644 --- a/src/agent_comms/transcript_routes.py +++ b/src/agent_comms/transcript_routes.py @@ -14,15 +14,14 @@ from .field_codec import MalformedRecord from .private_path import FileObservation, FileRevision from .routing import TurnRouting -from .typed_table import Column, DurableStore, TypedTable +from .typed_table import Column, DurableStore, TypedTable, WithoutRowid @dataclass(frozen=True) -class TranscriptRoute(TypedTable): +class TranscriptRoute(WithoutRowid): session_file: Annotated[str, Column(primary_key=True)] entry_id: Annotated[str, Column(primary_key=True)] routing: TurnRouting - without_rowid = True def require_publication(self, routing: TurnRouting) -> None: if self.routing.publications and self.routing != routing: @@ -30,14 +29,13 @@ def require_publication(self, routing: TurnRouting) -> None: @dataclass(frozen=True) -class InputDisplay(TypedTable): +class InputDisplay(WithoutRowid): """Owner-authored presentation for one native input; None text means internal.""" native_id: Annotated[str, Column(primary_key=True)] text: str | None routing: TurnRouting | None = None sent_text_digest: str | None = None - without_rowid = True checks = ("routing IS NULL OR sent_text_digest IS NOT NULL",) def matches(self, text: str) -> bool: diff --git a/src/agent_comms/triage_native_sources.py b/src/agent_comms/triage_native_sources.py index 4732af6952..6682041384 100644 --- a/src/agent_comms/triage_native_sources.py +++ b/src/agent_comms/triage_native_sources.py @@ -13,7 +13,7 @@ Column, GuardedInserts, ImmutableRows, - TypedTable, + WithoutRowid, ) @@ -22,7 +22,7 @@ class TriageNativeSources( GuardedInserts, ImmutableRows, NativeRuntimeTable, - TypedTable, + WithoutRowid, declared_name="native_runtime_triage_sources", ): insert_condition = """(NOT EXISTS (SELECT 1 FROM native_runtime_input n @@ -38,7 +38,6 @@ class TriageNativeSources( ] assignment_ids: tuple[str, ...] - without_rowid = True checks = ("json_array_length(assignment_ids)>0",) def __post_init__(self): diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index b08232fbf5..52d8b3acf8 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -2156,3 +2156,9 @@ def __init_subclass__(cls, **kwargs: object) -> None: cls.filename = ( f"{declared_words(cls.__name__)}.{hashlib.sha256(schema).hexdigest()[:12]}.sqlite3" ) + + +class WithoutRowid(TypedTable, declared_name=None): + """The table is stored directly by its declared primary key.""" + + without_rowid = True diff --git a/src/agent_comms/view_unread.py b/src/agent_comms/view_unread.py index 01838fe908..b341cd4dc0 100644 --- a/src/agent_comms/view_unread.py +++ b/src/agent_comms/view_unread.py @@ -23,7 +23,15 @@ from .read_ledger import ReadDocument, ReadLedger from .registry_document import RosterVisibility from .thread_owned_state import ThreadOwnedState -from .typed_table import Column, JsonStorage, SQLiteTableName, SQLiteUserVersion, TypedRow, TypedTable +from .typed_table import ( + Column, + JsonStorage, + SQLiteTableName, + SQLiteUserVersion, + TypedRow, + TypedTable, + WithoutRowid, +) from .ui_model.unread import CappedReplyUnread, IndexingUnread, ReplyUnread, UnreadPresentation from .watched import FileSource @@ -115,10 +123,9 @@ def prepare(self, cancelled: Event) -> ReplyIndexResult: @dataclass(frozen=True) -class TranscriptReply(ReplyIndexTable, TypedTable): +class TranscriptReply(ReplyIndexTable, WithoutRowid): source: Annotated[str, Column(primary_key=True)] end: Annotated[int, Column(primary_key=True)] - without_rowid = True @classmethod def after(cls, database: sqlite3.Connection, ends: Mapping[str, int]) -> dict[str, int]: From 5b1c98ae9eb9b67a8351e107a39d7dff7ef650e3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:43:48 -0400 Subject: [PATCH 2/7] Share fixed singleton keys and plain turn item identity --- src/agent_comms/cli_protocol.py | 5 ++-- src/agent_comms/context_store.py | 7 +++--- src/agent_comms/private_sidecar.py | 6 ++--- src/agent_comms/runtime_updates.py | 39 +++++++++--------------------- src/agent_comms/turn_item.py | 15 ++++++++++++ src/agent_comms/typed_table.py | 7 ++++++ 6 files changed, 41 insertions(+), 38 deletions(-) create mode 100644 src/agent_comms/turn_item.py diff --git a/src/agent_comms/cli_protocol.py b/src/agent_comms/cli_protocol.py index ac0fa08f16..c3aa226525 100644 --- a/src/agent_comms/cli_protocol.py +++ b/src/agent_comms/cli_protocol.py @@ -20,6 +20,7 @@ from .field_codec import TimestampText from .recorded_time import UnrecordedTimeError from .external_transcript import RequestedEarlierPage +from .turn_item import TurnItem if TYPE_CHECKING: from .child_process import AttachedChild @@ -222,11 +223,9 @@ async def handle(self, bridge: CliBridge, child: AttachedChild) -> None: await bridge.accept(self.message) -class EarlierItems(WatchLine): +class EarlierItems(WatchLine, TurnItem): """The pager's outstanding continuation, handled only by its socket-owning bridge.""" wire_tag = 'earlier-items' - turn_id: str - item_id: str request: RequestedEarlierPage @classmethod diff --git a/src/agent_comms/context_store.py b/src/agent_comms/context_store.py index 9ebb031928..b832e3a636 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 TYPE_CHECKING, Annotated, ClassVar, Self, cast from .errors import RelationViolationError, StoredHistoryUnreadable from .field_codec import ( @@ -37,7 +37,7 @@ if TYPE_CHECKING: from .registry_document import RegistrySnapshot -from .typed_table import Column, DurableStore, Generated, Index, TypedTable +from .typed_table import Column, DurableStore, Generated, Index, SingletonRow, TypedTable @dataclass(frozen=True) @@ -95,8 +95,7 @@ def page( @dataclass(frozen=True) -class ContextStoreInfo(ContextSourceState, TypedTable): - singleton: Annotated[Literal[1], Column(primary_key=True)] +class ContextStoreInfo(ContextSourceState, SingletonRow): source: ContextStoreIdentity @classmethod diff --git a/src/agent_comms/private_sidecar.py b/src/agent_comms/private_sidecar.py index 5c7c195f60..3049388fc4 100644 --- a/src/agent_comms/private_sidecar.py +++ b/src/agent_comms/private_sidecar.py @@ -25,7 +25,6 @@ from dataclasses import dataclass from functools import partial from pathlib import Path -from typing import Annotated, Literal from agent_comms.locked_store import SQLiteFile from agent_comms.coordination_errors import IdentityConflict @@ -34,8 +33,8 @@ from .store_files import StoreOwner from .typed_table import ( AllObjectsTables, - Column, ImmutableRows, + SingletonRow, SQLiteDatabase, SQLiteForeignKeys, SQLiteSchemaObject, @@ -204,8 +203,7 @@ def _connect(raw: bytes | None) -> sqlite3.Connection: @dataclass(frozen=True) -class SnapshotMeta(ImmutableRows, TypedTable): - singleton: Annotated[Literal[1], Column(primary_key=True)] +class SnapshotMeta(ImmutableRows, SingletonRow): ddl_digest: str @dataclass(frozen=True) diff --git a/src/agent_comms/runtime_updates.py b/src/agent_comms/runtime_updates.py index 6b6d6cbe43..97a6658a81 100644 --- a/src/agent_comms/runtime_updates.py +++ b/src/agent_comms/runtime_updates.py @@ -24,6 +24,7 @@ ) from .thread_text import NoThreadText, ThreadText from .turn_phase import IdlePhase, ModelWaitPhase, TurnPhase +from .turn_item import TurnItem if TYPE_CHECKING: from .cli_bridges.shared import TranscriptBridge @@ -150,36 +151,20 @@ def append(self, update: AppendDraftItem) -> tuple[RuntimeUpdate, ...]: return () -@dataclass(frozen=True) -class ItemKey: - """One item's identity: the turn that holds it and its producer's item id.""" - - turn_id: str - item_id: str - - -class KeyedItem(ItemKey): - """An update about one item, named by the turn that holds it and its producer's item id.""" - - @property - def key(self) -> ItemKey: - return ItemKey(self.turn_id, self.item_id) - - @dataclass(frozen=True) class ThreadOutput(activity.ThreadOutput): """Where a CLI's output lands in Core: finished items and current activity.""" transcript: ExternalTranscript agents: AgentActivity - drafts: dict[ItemKey, DraftAccumulator] = field(default_factory=dict) + drafts: dict[TurnItem, DraftAccumulator] = field(default_factory=dict) def publish_batch(self, updates: list[RuntimeUpdate]) -> None: assembled = [part for update in updates for part in update.accumulate(self)] for update in RuntimeUpdate.coalesced(assembled): update.publish(self) - def draft(self, key: ItemKey) -> DraftAccumulator: + def draft(self, key: TurnItem) -> DraftAccumulator: return self.drafts.get(key, NO_PREFIX) def start_item(self, update: StartedItem) -> tuple[RuntimeUpdate, ...]: @@ -235,7 +220,7 @@ def accumulate(self, output: ThreadOutput) -> tuple[RuntimeUpdate, ...]: async def observed(self, bridge: TranscriptBridge) -> None: # noqa: B027 - default hook """A source-owned notification may confirm delivery before publication.""" - def retain_before(self, later: set[ItemKey]) -> bool: + def retain_before(self, later: set[TurnItem]) -> bool: """Whether this update still needs publication before later updates.""" return True @@ -248,7 +233,7 @@ def add_to(self, history: ImportBuffer) -> None: # noqa: B027 - default hook @staticmethod def coalesced(updates: list[RuntimeUpdate]) -> list[RuntimeUpdate]: - later: set[ItemKey] = set() + later: set[TurnItem] = set() kept = [update for update in reversed(updates) if update.retain_before(later)] return list(reversed(kept)) @@ -276,7 +261,7 @@ def publish(self, output: ThreadOutput) -> None: @dataclass(frozen=True) -class ActiveItem(KeyedItem, StatusUpdate): +class ActiveItem(TurnItem, StatusUpdate): """An item the CLI started producing: what it is doing now.""" phase: TurnPhase @@ -289,7 +274,7 @@ def accumulate(self, output: ThreadOutput) -> tuple[RuntimeUpdate, ...]: @dataclass(frozen=True) -class FinishedItem(KeyedItem, RuntimeUpdate): +class FinishedItem(TurnItem, RuntimeUpdate): """One CLI item's complete presentation, identified by its producer.""" events: tuple[TranscriptEvent, ...] @@ -297,7 +282,7 @@ class FinishedItem(KeyedItem, RuntimeUpdate): def publish(self, output: ThreadOutput) -> None: output.transcript.finish(turn_id=self.turn_id, item_id=self.item_id, events=self.events) - def retain_before(self, later: set[ItemKey]) -> bool: + def retain_before(self, later: set[TurnItem]) -> bool: later.add(self.key) return True @@ -309,7 +294,7 @@ def add_to(self, history: ImportBuffer) -> None: @dataclass(frozen=True) -class PageItem(ItemKey): +class PageItem(TurnItem): """An item-list identity survives even when its update is invisible in the live view.""" update: RuntimeUpdate @@ -412,7 +397,7 @@ def add_to(self, history: ImportBuffer) -> None: @dataclass(frozen=True) -class DraftItem(KeyedItem, RuntimeUpdate): +class DraftItem(TurnItem, RuntimeUpdate): """A pending item's events, materialized only if this preview survives coalescing.""" accumulator: PresentableDraft @@ -424,7 +409,7 @@ def events(self) -> tuple[TranscriptEvent, ...]: def publish(self, output: ThreadOutput) -> None: output.transcript.draft(turn_id=self.turn_id, item_id=self.item_id, events=self.events) - def retain_before(self, later: set[ItemKey]) -> bool: + def retain_before(self, later: set[TurnItem]) -> bool: keep = self.key not in later later.add(self.key) return keep @@ -455,7 +440,7 @@ def accumulate(self, output: ThreadOutput) -> tuple[RuntimeUpdate, ...]: @dataclass(frozen=True) -class AppendDraftItem(KeyedItem, DraftUpdate): +class AppendDraftItem(TurnItem, DraftUpdate): text: str section: int = 0 diff --git a/src/agent_comms/turn_item.py b/src/agent_comms/turn_item.py new file mode 100644 index 0000000000..d5de80b5a8 --- /dev/null +++ b/src/agent_comms/turn_item.py @@ -0,0 +1,15 @@ +"""The producer's item identity within its native turn.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class TurnItem: + turn_id: str + item_id: str + + @property + def key(self) -> TurnItem: + return TurnItem(self.turn_id, self.item_id) diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 52d8b3acf8..062e75c600 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -2162,3 +2162,10 @@ class WithoutRowid(TypedTable, declared_name=None): """The table is stored directly by its declared primary key.""" without_rowid = True + + +@dataclass(frozen=True) +class SingletonRow(TypedTable, declared_name=None): + """The table has one fixed primary key, preceding its owned value fields.""" + + singleton: Annotated[Literal[1], Column(primary_key=True)] From 41123458987fcccad94212dada03ddcb366dfc0f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:56:45 -0400 Subject: [PATCH 3/7] Distinguish claimed identities, resource scopes and claim generations --- src/agent_comms/acp_extension.py | 18 +++++++++--------- src/agent_comms/channel_coding_tools.py | 12 ++++++------ src/agent_comms/coordination_snapshot.py | 2 +- src/agent_comms/envelope_claim_transitions.py | 14 +++++++------- src/agent_comms/following_inputs.py | 2 +- src/agent_comms/goal_failure_observation.py | 10 +++++----- src/agent_comms/input_attempt.py | 8 ++++---- src/agent_comms/input_origin.py | 8 ++++---- src/agent_comms/native_input_owner.py | 4 ++-- src/agent_comms/ordinary_turn_rules.py | 8 ++++---- src/agent_comms/owned_send.py | 10 +++++----- src/agent_comms/owned_turn.py | 4 ++-- src/agent_comms/private_send_rules.py | 4 ++-- src/agent_comms/publisher.py | 2 +- src/agent_comms/queued_input.py | 18 +++++++++--------- src/agent_comms/selected_actions.py | 4 ++-- src/agent_comms/selected_claim.py | 2 +- src/agent_comms/selected_tool_broker.py | 2 +- src/agent_comms/selected_turn.py | 2 +- src/agent_comms/selected_write_plan.py | 2 +- src/agent_comms/task_sources.py | 2 +- tests/test_acp_queue_contract.py | 12 ++++++------ tests/test_bus_resource_claims.py | 2 +- tests/test_claim_admission_verifier.py | 2 +- tests/test_current_input_origin.py | 4 ++-- tests/test_envelope_bus_integration.py | 6 +++--- tests/test_envelope_claim_transitions.py | 10 +++++----- tests/test_goal_input_review.py | 2 +- tests/test_input_documents.py | 2 +- tests/test_interrupted_delivery_recovery.py | 2 +- tests/test_retained_context_operations.py | 2 +- tests/test_stopped_claim_expiry.py | 2 +- tests/test_thread_identity.py | 4 +++- tests/test_tool_output.py | 2 +- tests/test_unscoped_claim_history.py | 4 ++-- 35 files changed, 98 insertions(+), 96 deletions(-) diff --git a/src/agent_comms/acp_extension.py b/src/agent_comms/acp_extension.py index 0d69abdcc5..8427f3cee0 100644 --- a/src/agent_comms/acp_extension.py +++ b/src/agent_comms/acp_extension.py @@ -200,7 +200,7 @@ def require_live(self): class LiveAttachmentScope(AttachmentScope): """Logical ACP attachment and executor evidence have separate identities.""" session_id: str - claim_scope: ClaimedIdentity + claimed_identity: ClaimedIdentity owner_pid: int @property @@ -209,11 +209,11 @@ def logical_key(self) -> tuple[str, ...]: ... @property def owner_created_at(self): - return self.claim_scope.incarnation.require_recorded().created_at + return self.claimed_identity.incarnation.require_recorded().created_at @property def claimed_generation(self): - return self.claim_scope.claimed_generation + return self.claimed_identity.claimed_generation def to_wire(self): return FieldCodec.encode_record(self) @@ -301,16 +301,16 @@ class NoCursorScope(CursorScope): class LiveCursorScope(CursorScope, LiveAttachmentScope): session_id: str wire_root_id: str - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] owner_pid: int def __post_init__(self): Id128Text.decode(self.wire_root_id) - self.claim_scope.incarnation.require_recorded() + self.claimed_identity.incarnation.require_recorded() if ( not self.session_id or self.owner_pid <= 0 - or self.claim_scope.claimed_generation <= 0 + or self.claimed_identity.claimed_generation <= 0 ): raise ValueError("Invalid native cursor scope") @@ -370,7 +370,7 @@ def validate_scope(self, scope: CursorScope) -> None: cursor = self.cursor if ( cursor.wire_root_id != scope.wire_root_id - or cursor.owner_thread != scope.claim_scope.incarnation.name + or cursor.owner_thread != scope.claimed_identity.incarnation.name or cursor.owner_claimed_generation != scope.claimed_generation or not 0 <= cursor.injected_seq <= cursor.covered_seq ): @@ -446,7 +446,7 @@ class NoQueueScope(QueueScope): @dataclass(frozen=True) class LiveQueueScope(QueueScope, LiveAttachmentScope): session_id: str - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] owner_pid: int @property @@ -549,7 +549,7 @@ def without_scope(self) -> None: raise ValueError('Available input queue has no original attachment scope.') def capture_human_input(self, comms, acquire_scope: Callable[[], QueueScope]) -> HumanInputOrigin: - return HumanInputOrigin.capture(comms, acquire_scope().claim_scope) + return HumanInputOrigin.capture(comms, acquire_scope().claimed_identity) @dataclass(frozen=True) diff --git a/src/agent_comms/channel_coding_tools.py b/src/agent_comms/channel_coding_tools.py index fda7f344e3..81f44818bf 100644 --- a/src/agent_comms/channel_coding_tools.py +++ b/src/agent_comms/channel_coding_tools.py @@ -100,7 +100,7 @@ def for_original(self, assignment, operation_id): @dataclass(frozen=True, slots=True) class SelectedCodingToolOwner(CodingToolOwner): store: Coordination - claim_scope: WakeClaim + resource_scope: WakeClaim session_dir: Path input_id: str goal_permission: TurnGoalPermission = field() @@ -115,8 +115,8 @@ def for_original(self, assignment, operation_id) -> CodingToolOwner: self.store.session._connection, self.input_id ): raise SelectedToolDenied("Tool original is absent from the reserved native batch") - return replace(self, claim_scope=replace( - self.claim_scope, + return replace(self, resource_scope=replace( + self.resource_scope, source_seq=assignment.wire_seq, source_message_id=assignment.message_id, wake_assignment_id=assignment.assignment_id, @@ -127,11 +127,11 @@ def for_original(self, assignment, operation_id) -> CodingToolOwner: def verify(self, store, turn, snapshot): if self.input_id != turn.command.input_id: raise SelectedToolDenied("Native tool input changed") - verify_sent_full_input(store, self.claim_scope, self.owner_name, self.input_id) - verify_selected_wake(self.comms, store, self.claim_scope, self.owner_name) + verify_sent_full_input(store, self.resource_scope, self.owner_name, self.input_id) + verify_selected_wake(self.comms, store, self.resource_scope, self.owner_name) def claim_resource(self, store, turn, resource, slot): - scope = replace(self.claim_scope, operation_id=slot) + scope = replace(self.resource_scope, operation_id=slot) return publish_selected_resource_claim(self.comms, store, scope, self.owner_name, resource) diff --git a/src/agent_comms/coordination_snapshot.py b/src/agent_comms/coordination_snapshot.py index a9a8ca174e..26c176857c 100644 --- a/src/agent_comms/coordination_snapshot.py +++ b/src/agent_comms/coordination_snapshot.py @@ -357,7 +357,7 @@ def release_resources(self, boundary: NativeBoundary, outcome: TerminalAttempt) current = bus.bus.claim_projection().live_in(boundary.registry) owned = tuple( claim for claim in current.values() - if claim.claim_scope.belongs_to(attempt) + if claim.resource_scope.belongs_to(attempt) ) if owned: bus.publisher.release_resources(boundary, owned, outcome.release_wording) diff --git a/src/agent_comms/envelope_claim_transitions.py b/src/agent_comms/envelope_claim_transitions.py index 87ffd2d288..d3c14b724f 100644 --- a/src/agent_comms/envelope_claim_transitions.py +++ b/src/agent_comms/envelope_claim_transitions.py @@ -274,7 +274,7 @@ class ClaimIdentity: @dataclass(frozen=True) class AcquiringClaim(ClaimIdentity): generation: Annotated[str, Id128Text] - claim_scope: Annotated[ClaimScope, WireName("admission")] + resource_scope: Annotated[ClaimScope, WireName("admission")] @dataclass(frozen=True) @@ -351,7 +351,7 @@ class Acquisition(ClaimTransition, declared_name=None): default=(), kw_only=True ) generation: Annotated[str, Id128Text] = field(kw_only=True) - claim_scope: ClassVar[ClaimScope] + resource_scope: ClassVar[ClaimScope] def __post_init__(self) -> None: super().__post_init__() @@ -372,7 +372,7 @@ def claimed(self, resource: str) -> ClaimOwner: @property def operation_ids(self) -> tuple[str, ...]: - return self.claim_scope.operation_ids + return self.resource_scope.operation_ids def require_holders(self, previous: ClaimProjection) -> None: for resource in self.claims: @@ -390,7 +390,7 @@ def project(self, previous: ClaimProjection) -> ClaimProjection: class WakeClaims(RecordMember, Acquisition, AcquiringClaim): """An acquisition published today; its row records the scope it serves.""" - claim_scope: Annotated[RecordedScope, WireName("admission")] + resource_scope: Annotated[RecordedScope, WireName("admission")] @dataclass(frozen=True) @@ -400,7 +400,7 @@ class UnscopedClaims(RecordMember, Acquisition): Only bus history holds this row; no writer publishes one. """ - claim_scope: ClassVar[ClaimScope] = UnrecordedScope() + resource_scope: ClassVar[ClaimScope] = UnrecordedScope() @dataclass(frozen=True) @@ -447,7 +447,7 @@ def __post_init__(self) -> None: def acquire( self, scope: RecordedScope, registry: RegistrySnapshot, publish: Callable[[], ClaimOwner] ) -> ClaimOwner: - return self if self.claim_scope.reuses(scope) and self.live(registry) else publish() + return self if self.resource_scope.reuses(scope) and self.live(registry) else publish() def require_holder(self, owner: str, incarnation: str) -> None: if (self.owner, self.incarnation) != (owner, incarnation): @@ -457,7 +457,7 @@ def live(self, registry: RegistrySnapshot) -> bool: incarnation = RecordedThreadIncarnation(self.owner, float(self.incarnation)) return any( thread.process_alive - and self.claim_scope.held_at(registry.claim_generation(thread.name)) + and self.resource_scope.held_at(registry.claim_generation(thread.name)) for thread in registry.active_threads(registry.threads_for(incarnation)) ) diff --git a/src/agent_comms/following_inputs.py b/src/agent_comms/following_inputs.py index 8868056c03..e5320c8319 100644 --- a/src/agent_comms/following_inputs.py +++ b/src/agent_comms/following_inputs.py @@ -90,7 +90,7 @@ async def open(self, session: TrackedTurnSession) -> None: session_id=self.session_id, thread=thread, turn=TurnId(lease.turn_id), - claim_scope=lease.claimed_generation, + claim_generation=lease.claimed_generation, original=original, goal_permit=NoLaunchPermit(), ) diff --git a/src/agent_comms/goal_failure_observation.py b/src/agent_comms/goal_failure_observation.py index 035bd0bdcc..31a8b211ba 100644 --- a/src/agent_comms/goal_failure_observation.py +++ b/src/agent_comms/goal_failure_observation.py @@ -56,8 +56,8 @@ class FailedTurnEvidence(GoalLedgerTable, TypedTable, GoalFailureEvidence): owner: str owner_created_at: float worktree: str - claim_scope: Annotated[ - int, WireName("admission"), Column(record_key=True, check="{claim_scope}>0") + claim_generation: Annotated[ + int, WireName("admission"), Column(record_key=True, check="{claim_generation}>0") ] turn_generation: Annotated[int, Column(check="turn_generation>0")] goal_revision: Annotated[int, Column(check="goal_revision>=0")] @@ -67,7 +67,7 @@ class FailedTurnEvidence(GoalLedgerTable, TypedTable, GoalFailureEvidence): def __post_init__(self) -> None: # These are the durable observation's constraints, independent of any # caller's authority. SQL/FieldCodec remains the single stored decoder. - if type(self.claim_scope) is not int or self.claim_scope <= 0: + if type(self.claim_generation) is not int or self.claim_generation <= 0: raise ValueError("Observation claim_scope must be positive") if type(self.turn_generation) is not int or self.turn_generation <= 0: raise ValueError("Observation turn generation must be positive") @@ -91,11 +91,11 @@ def incarnation(self) -> ThreadIncarnation: @property def claimed_identity(self) -> ClaimedIdentity: - return ClaimedIdentity(self.incarnation, self.claim_scope) + return ClaimedIdentity(self.incarnation, self.claim_generation) @property def turn(self) -> TurnFence: - return TurnFence(TurnIdentity(self.incarnation, self.turn_generation), self.turn_id, self.claim_scope) + return TurnFence(TurnIdentity(self.incarnation, self.turn_generation), self.turn_id, self.claim_generation) def matches_owner(self, owner: Thread, claim_scope: int | None) -> bool: if claim_scope is None: diff --git a/src/agent_comms/input_attempt.py b/src/agent_comms/input_attempt.py index 67e8a859dc..b3b0c84993 100644 --- a/src/agent_comms/input_attempt.py +++ b/src/agent_comms/input_attempt.py @@ -248,7 +248,7 @@ def require_original_provenance(self, source: InputProvenance): key: str sequence: int | None owner: str - claim_scope: Annotated[int, WireName("admission")] + claim_generation: Annotated[int, WireName("admission")] target: str source_text: str origin: Annotated[InputOrigin, WireOmitDefault()] = field( @@ -266,7 +266,7 @@ def __post_init__(self) -> None: for value in (self.key, self.owner, self.target, self.source_text): if not FieldCodec.decode(str, value): raise ValueError("Input identity text cannot be empty") - GenerationCounter.require_positive(self.claim_scope) + GenerationCounter.require_positive(self.claim_generation) except ValueError as error: raise ValueError("Invalid ACP input identity") from error if self.sequence is not None and ( @@ -288,7 +288,7 @@ def matches_owner(self, source_owner: ThreadIncarnation) -> bool: return self.origin.matches_owner(self.owner, source_owner) def matches_claim(self, claim_scope: int) -> bool: - return self.claim_scope == claim_scope + return self.claim_generation == claim_scope def _transition(self, target: type[StoredInput], **changes) -> StoredInput: values = {item.name: getattr(self, item.name) for item in fields(self)} @@ -334,7 +334,7 @@ def queued_for(self, owner: ThreadIncarnation, claim_scope: int, text: str) -> b def bind( self, *, claim_scope: int, turn_id: str, native_id: str, text: str ) -> InputTransition: - if self.claim_scope != claim_scope: + if self.claim_generation != claim_scope: return UnchangedInput(self) return MovedInput(self._transition( BoundUnknownInput, turn_id=turn_id, native_id=native_id, sent_digest=TextDigest.of(text) diff --git a/src/agent_comms/input_origin.py b/src/agent_comms/input_origin.py index b392102bf8..a4555d7695 100644 --- a/src/agent_comms/input_origin.py +++ b/src/agent_comms/input_origin.py @@ -120,7 +120,7 @@ class HumanInputOrigin(InputOrigin): root_id: Annotated[str, Id128Text] author: HumanOrigin - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] project: str goal: GoalCheckpoint @@ -154,7 +154,7 @@ def require_ingress(self, comms, snapshot, claim_scope, controller): comms.registry.store.private_guard_unlocked().require_root( self.root_id, RelationViolationError("User input origin belongs to another native namespace"), ) - if self.claim_scope != claim_scope: + if self.claimed_identity != claim_scope: raise RelationViolationError("User input origin attachment changed") self.author.require_registered(snapshot) owner = snapshot.require_active(claim_scope.incarnation.name) @@ -170,12 +170,12 @@ def require_human(self) -> HumanInputOrigin: return self def applies(self, owner: Thread, snapshot: RegistrySnapshot) -> bool: - return (self.claim_scope.incarnation.resolved(snapshot) == owner.incarnation + return (self.claimed_identity.incarnation.resolved(snapshot) == owner.incarnation and (self.project, self.goal) == (owner.worktree, owner.goal.captured)) def matches_owner(self, recorded_name, incarnation): """The captured original birth survives a rename, unlike a neutral name.""" - authorized = self.claim_scope.incarnation.require_recorded() + authorized = self.claimed_identity.incarnation.require_recorded() original = incarnation.require_recorded() return (recorded_name == authorized.name and authorized.created_at == original.created_at) diff --git a/src/agent_comms/native_input_owner.py b/src/agent_comms/native_input_owner.py index 3dbce8460a..4816ba6168 100644 --- a/src/agent_comms/native_input_owner.py +++ b/src/agent_comms/native_input_owner.py @@ -73,7 +73,7 @@ def require_claim(self, expected: Thread, claim_scope: int) -> None: try: GoalRegistryIdentityCheck( expected=expected, actual=self.thread, - expected_claim=claim_scope, claim_scope=self.claimed_generation, + expected_claim=claim_scope, claim_generation=self.claimed_generation, process=ProcessIdentity.capture(os.getpid()), goal_permission=OwnerGoalPermission(expected.goal), ).require_valid() @@ -112,7 +112,7 @@ def _require_current( expected=self.thread, actual=actual, expected_claim=self.claimed_generation, - claim_scope=claim_scope, + claim_generation=claim_scope, process=ProcessIdentity.capture(os.getpid()), goal_permission=self.goal_permission, ).require_valid() diff --git a/src/agent_comms/ordinary_turn_rules.py b/src/agent_comms/ordinary_turn_rules.py index 8271643f4b..0c67b6b695 100644 --- a/src/agent_comms/ordinary_turn_rules.py +++ b/src/agent_comms/ordinary_turn_rules.py @@ -38,7 +38,7 @@ class OrdinaryOwnerCheck(RuleCheck): current: Thread registry: RegistrySnapshot canonical: str - claim_scope: int + claim_generation: int turn: TurnId @@ -63,7 +63,7 @@ class OrdinaryClaimRule(ReservationRule): explanation = "The ordinary owner claim_scope generation changed." def violated(self, check: OrdinaryOwnerCheck) -> bool: - return check.registry.claim_generation(check.canonical) != check.claim_scope + return check.registry.claim_generation(check.canonical) != check.claim_generation class OrdinaryProcessRule(ReservationRule): @@ -122,7 +122,7 @@ def violated(self, check: MissingAcceptedInputCheck) -> bool: class AcceptedInputCheck(RuleCheck): acknowledgement: QueuedInput current: Thread - claim_scope: int + claim_generation: int input_id: str keys: tuple[str, ...] inputs: InputDrain @@ -134,7 +134,7 @@ class AcceptedInputAuthorityRule(ReservationRule): explanation = "The live input's accepted owner or goal authority changed." def violated(self, check: AcceptedInputCheck) -> bool: - return not check.acknowledgement.current(check.current, check.claim_scope) + return not check.acknowledgement.current(check.current, check.claim_generation) class AcceptedInputKeyRule(ReservationRule): diff --git a/src/agent_comms/owned_send.py b/src/agent_comms/owned_send.py index b7c7e58a56..382699c77e 100644 --- a/src/agent_comms/owned_send.py +++ b/src/agent_comms/owned_send.py @@ -53,7 +53,7 @@ class OwnedSend(CodingToolOwner): session_id: str thread: Thread turn: TurnId - claim_scope: int + claim_generation: int original: OriginalTurnInput goal_permit: GoalTurnPermit @@ -80,7 +80,7 @@ def _check_owner( current=current, registry=snapshot, canonical=canonical, - claim_scope=self.claim_scope, + claim_generation=self.claim_generation, turn=self.turn, ).require_valid() InputKeysCheck(source=source, text=text).require_valid() @@ -119,7 +119,7 @@ def _prepare( AcceptedInputCheck( acknowledgement=accepted, current=current, - claim_scope=self.claim_scope, + claim_generation=self.claim_generation, input_id=source.accepted_id, keys=source.keys, inputs=self.inputs, @@ -138,7 +138,7 @@ def _prepare( # never reinterpret a failed durable operation as a retryable policy refusal. if not self.inputs.dispositions.bind_turn_input( source.keys, - claim_scope=self.claim_scope, + claim_scope=self.claim_generation, turn_id=self.turn.value, native_id=native_id, text=sent_text, @@ -191,7 +191,7 @@ def claim_resource(self, store, turn, resource, slot): self.verify(store, turn, snapshot) owner = snapshot.require(self.owner_name) canonical = resource.normalized(Path(owner.worktree)) - scope = NativeClaim(owner_claimed_generation=self.claim_scope, + scope = NativeClaim(owner_claimed_generation=self.claim_generation, operation_id=slot, turn_id=self.turn.value, input_id=turn.command.input_id) return boundary.bus.publisher._publish_acquisition_unlocked( diff --git a/src/agent_comms/owned_turn.py b/src/agent_comms/owned_turn.py index ced5311af1..37e849cc91 100644 --- a/src/agent_comms/owned_turn.py +++ b/src/agent_comms/owned_turn.py @@ -167,7 +167,7 @@ def begin(self, resources: AsyncExitStack, permits: ExitStack) -> bool: key=self.runner.inputs.dispositions.bus_key(origin, self.thread), sequence=origin.seq, owner=self.thread_name, - claim_scope=snapshot.claim_generation(self.thread_name), + claim_generation=snapshot.claim_generation(self.thread_name), target=origin.target, source_text=ScheduledTurn.incoming( origin, aliases=snapshot.aliases @@ -345,7 +345,7 @@ async def stream(self): session_id=self.session_id, thread=self.thread, turn=TurnId(self.turn_id), - claim_scope=self.turn_lease.claimed_generation, + claim_generation=self.turn_lease.claimed_generation, goal_permit=self.goal_permit, original=replace(self.original, prompt=rendered.text), ) diff --git a/src/agent_comms/private_send_rules.py b/src/agent_comms/private_send_rules.py index 0e3a4ccac1..16597129fe 100644 --- a/src/agent_comms/private_send_rules.py +++ b/src/agent_comms/private_send_rules.py @@ -25,7 +25,7 @@ class RegistryIdentityCheck(RuleCheck): expected: Thread actual: Thread expected_claim: int - claim_scope: int | None + claim_generation: int | None process: ProcessIdentity goal_permission: GoalPermission @@ -119,7 +119,7 @@ class RegistryClaimRule(ReservationRule): explanation = "The registry claim_scope generation changed." def violated(self, check: RegistryIdentityCheck) -> bool: - return check.claim_scope != check.expected_claim + return check.claim_generation != check.expected_claim class RegistryRoleRule(ReservationRule): diff --git a/src/agent_comms/publisher.py b/src/agent_comms/publisher.py index 2f767e8c47..fb8b7e67fa 100644 --- a/src/agent_comms/publisher.py +++ b/src/agent_comms/publisher.py @@ -173,7 +173,7 @@ def _publish_acquisition_unlocked( build = self._claim_builder( snapshot, ClaimNoticeKind.claimed(owner), partial(WakeClaims, incarnation=str(owner.created_at), claims=(resource,), - generation=Id128Text.mint(), claim_scope=claim), activity=activity, + generation=Id128Text.mint(), resource_scope=claim), activity=activity, ) return self._claim_result(lambda: self.bus.acquire_resource_claim(resource, claim, snapshot, build)) diff --git a/src/agent_comms/queued_input.py b/src/agent_comms/queued_input.py index d0a0ca67a8..e5d2408acd 100644 --- a/src/agent_comms/queued_input.py +++ b/src/agent_comms/queued_input.py @@ -55,7 +55,7 @@ class InputHandoffRefused(RelationViolationError): @dataclass(frozen=True, slots=True) class QueuedInputContext: - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] goal: GoalDeclaration = field(default_factory=NoGoal) @classmethod @@ -66,14 +66,14 @@ def named(self, name: str) -> QueuedInputContext: """Use the caller's canonical registry name without changing authority.""" return replace( self, - claim_scope=replace( - self.claim_scope, - incarnation=replace(self.claim_scope.incarnation, name=name), + claimed_identity=replace( + self.claimed_identity, + incarnation=replace(self.claimed_identity.incarnation, name=name), ), ) def owns(self, claim_scope: ClaimedIdentity) -> bool: - return self.named(claim_scope.incarnation.name).claim_scope == claim_scope + return self.named(claim_scope.incarnation.name).claimed_identity == claim_scope @property def active_goal_id(self) -> str | None: @@ -98,7 +98,7 @@ def queue_items(self, scope: QueueScope) -> tuple[QueueItem, ...]: """Show the item only when its recorded input generation matches the queue.""" from .acp_extension import QueueItem - return (QueueItem(self.input_id, self.text),) if self.context.owns(scope.claim_scope) else () + return (QueueItem(self.input_id, self.text),) if self.context.owns(scope.claimed_identity) else () def source(self) -> AcceptedFollowingInput: return AcceptedFollowingInput( @@ -130,7 +130,7 @@ def capture( owner, snapshot.claimed_identity(canonical), ) - origin.require_ingress(inputs.comms, snapshot, context.claim_scope, controller) + origin.require_ingress(inputs.comms, snapshot, context.claimed_identity, controller) item = cls( text or prompt or "[image prompt]", echo, @@ -141,7 +141,7 @@ def capture( controller, ) row = ReservedInput( - item.key, None, canonical, context.claim_scope.claimed_generation, + item.key, None, canonical, context.claimed_identity.claimed_generation, canonical, item.text, origin=origin, ) document = inputs.dispositions.reserve_originals(row, ownership=ownership) @@ -362,7 +362,7 @@ def reserve_pending( (DeliveredMessageSegment.capture(source, assignment, owner),) ) row = ReservedInput( - key, message.seq, canonical, context.claim_scope.claimed_generation, + key, message.seq, canonical, context.claimed_identity.claimed_generation, message.target, ScheduledTurn.incoming(message, aliases=snapshot.aliases).prompt, origin=WireInputOrigin(root_id, message.reference), diff --git a/src/agent_comms/selected_actions.py b/src/agent_comms/selected_actions.py index 0e9bf49c2a..42c651628d 100644 --- a/src/agent_comms/selected_actions.py +++ b/src/agent_comms/selected_actions.py @@ -148,14 +148,14 @@ def apply(self, owner: CodingToolOwner) -> None: claimed = publish_selected_resource_claim( owner.comms, owner.store, - owner.claim_scope, + owner.resource_scope, owner.owner_name, self.resource, ) write_selected_claimed_file( owner.comms, owner.store, - owner.claim_scope, + owner.resource_scope, owner.owner_name, claimed, self.contents, diff --git a/src/agent_comms/selected_claim.py b/src/agent_comms/selected_claim.py index 9c330c27e6..496c1bc523 100644 --- a/src/agent_comms/selected_claim.py +++ b/src/agent_comms/selected_claim.py @@ -270,7 +270,7 @@ def write_selected_claimed_file( comms.bus.bus.writing() as db, ): normalized = ExistingFileClaim(Path(claimed.resource)).normalized(Path(owner.worktree)) - if claimed.claim_scope != claim_scope or BusSnapshot.from_connection(db).resource_owner(normalized) != claimed: + if claimed.resource_scope != claim_scope or BusSnapshot.from_connection(db).resource_owner(normalized) != claimed: raise IdentityConflict("Selected write has no current exact resource claim") with _opened_selected_file(Path(owner.worktree), normalized) as (fd, stable): os.ftruncate(fd, 0) diff --git a/src/agent_comms/selected_tool_broker.py b/src/agent_comms/selected_tool_broker.py index 7b0b4cf3d8..3d29829f8b 100644 --- a/src/agent_comms/selected_tool_broker.py +++ b/src/agent_comms/selected_tool_broker.py @@ -70,7 +70,7 @@ def mode(self, owner): return selected_tool_mode_for_owner( owner.comms, owner.store, - owner.claim_scope, + owner.resource_scope, owner.owner_name, owner.session_dir, owner.input_id, diff --git a/src/agent_comms/selected_turn.py b/src/agent_comms/selected_turn.py index e2b7cf0949..65f7bf07f6 100644 --- a/src/agent_comms/selected_turn.py +++ b/src/agent_comms/selected_turn.py @@ -165,7 +165,7 @@ def tool_owner(self, session, input_id, action) -> SelectedCodingToolOwner: return SelectedCodingToolOwner( comms=participant.comms, store=participant.store, - claim_scope=claim_scope, + resource_scope=claim_scope, owner_name=participant.owner.thread.name, session_dir=session.directory, input_id=input_id, diff --git a/src/agent_comms/selected_write_plan.py b/src/agent_comms/selected_write_plan.py index 5e6d2ee2b3..0e5724f028 100644 --- a/src/agent_comms/selected_write_plan.py +++ b/src/agent_comms/selected_write_plan.py @@ -53,7 +53,7 @@ class SelectedWriteBinding: root_id: str source: MessageReference assignment_id: str - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/task_sources.py b/src/agent_comms/task_sources.py index 6681d940d3..1383c9737e 100644 --- a/src/agent_comms/task_sources.py +++ b/src/agent_comms/task_sources.py @@ -599,7 +599,7 @@ def require_wording_publication(self, registry, original_source): original_author = RecordedThreadIncarnation(origin.author.sender, origin.author.created_at) if original_author.resolved(registry) != self.source_user.resolved(registry): raise RelationViolationError("USER input pin differs from its original human author") - if origin.claim_scope.incarnation.resolved(registry) != self.recipient.resolved(registry): + if origin.claimed_identity.incarnation.resolved(registry) != self.recipient.resolved(registry): raise RelationViolationError("USER input pin recipient did not receive its original input") if origin.project != self.scope.project: raise RelationViolationError("USER input pin belongs to another project") diff --git a/tests/test_acp_queue_contract.py b/tests/test_acp_queue_contract.py index d2efcbb622..27f6130d9a 100644 --- a/tests/test_acp_queue_contract.py +++ b/tests/test_acp_queue_contract.py @@ -62,7 +62,7 @@ async def test_real_acp_surrogate_queue_ingress_stays_unknown_and_attachable(tmp for f in decode_updates((await agent.sessions.metadata("beta", session_id="beta"))) if isinstance(f, QueueChangedUpdate) ) - assert state.scope.claim_scope.incarnation.name == "beta" + assert state.scope.claimed_identity.incarnation.name == "beta" assert isinstance(state.projection, UnavailableQueueProjection) assert exact in agent.inputs.state("beta").queued @@ -148,7 +148,7 @@ async def test_alias_maps_only_attachment_session_id(tmp_path): state = await agent.inputs.queue_state("beta") (rebased,) = decode_updates(present_session(encode_updates(state), "alias")) assert rebased.scope.session_id == "alias" - assert rebased.scope.claim_scope == state.scope.claim_scope + assert rebased.scope.claimed_identity == state.scope.claimed_identity assert rebased.projection is not None and rebased.projection == state.projection assert state.scope.session_id == "beta" @@ -160,12 +160,12 @@ def test_attachment_routing_rename_preserves_exact_executor_fences(): scopes = (LiveQueueScope("official-acp-id", original, 123), LiveCursorScope("official-acp-id", "a" * 32, original, 123)) for scope in scopes: - assert scope.relation(replace(scope, claim_scope=renamed)).current + assert scope.relation(replace(scope, claimed_identity=renamed)).current assert scope.relation(replace(scope, session_id="new-route")).foreign assert scope.relation(replace(scope, owner_pid=124)).ambiguous - assert scope.relation(replace(scope, claim_scope=replace(original, incarnation=replace(original.incarnation, created_at=1001.0)))).ambiguous - assert scope.relation(replace(scope, claim_scope=replace(original, claimed_generation=8))).newer - assert scope.relation(replace(scope, claim_scope=replace(original, claimed_generation=6))).older + assert scope.relation(replace(scope, claimed_identity=replace(original, incarnation=replace(original.incarnation, created_at=1001.0)))).ambiguous + assert scope.relation(replace(scope, claimed_identity=replace(original, claimed_generation=8))).newer + assert scope.relation(replace(scope, claimed_identity=replace(original, claimed_generation=6))).older assert scopes[1].relation(replace(scopes[1], wire_root_id="b" * 32)).foreign assert original == renamed # Birth owns identity; a name is only its route. assert original.incarnation.name != renamed.incarnation.name diff --git a/tests/test_bus_resource_claims.py b/tests/test_bus_resource_claims.py index 0d4c13bf10..de2c0e64f3 100644 --- a/tests/test_bus_resource_claims.py +++ b/tests/test_bus_resource_claims.py @@ -56,7 +56,7 @@ def build(seq: int, db: sqlite3.Connection) -> ClaimRecord: def acquisition(self, name: str, resources: tuple[str, ...], scope: NativeClaim) -> RecordBuilder: return self.builder(name, partial( - WakeClaims, claims=resources, generation=scope.operation_id, claim_scope=scope, + WakeClaims, claims=resources, generation=scope.operation_id, resource_scope=scope, )) def acquire(self, name: str, resource: str, scope: NativeClaim) -> ClaimOwner: diff --git a/tests/test_claim_admission_verifier.py b/tests/test_claim_admission_verifier.py index 7510471afd..a3db597acb 100644 --- a/tests/test_claim_admission_verifier.py +++ b/tests/test_claim_admission_verifier.py @@ -131,7 +131,7 @@ def test_selected_wake_verifier_refuses_no_wake_and_stale_authority( comms, store, claim_scope, owner.name, ExistingFileClaim(Path(resource)) ) selected_owner = Comms(root).bus.bus.claim_projection()[str(resource)] - assert selected_owner.claim_scope == claim_scope + assert selected_owner.resource_scope == claim_scope assert selected_owner.resource == str(resource) assert ( publish_selected_resource_claim( diff --git a/tests/test_current_input_origin.py b/tests/test_current_input_origin.py index 14eb23cd3e..2e7c86c70e 100644 --- a/tests/test_current_input_origin.py +++ b/tests/test_current_input_origin.py @@ -93,7 +93,7 @@ async def test_original_human_followup_survives_forwarding_reservation_and_nativ assert row.context_provenance() in contribution.provenance native_id = "b" * 32 sent = "User follow-up:\n" + command.user_text - assert inputs.bind(key, claim_scope=row.claim_scope, turn_id="held-native", native_id=native_id, text=sent) + assert inputs.bind(key, claim_scope=row.claim_generation, turn_id="held-native", native_id=native_id, text=sent) assert inputs.started(key, turn_id="held-native", native_id=native_id, text=sent) # Existing terminal settlement never rewrites a started original as # unsent or grants a replay; the native completion owns that. @@ -130,7 +130,7 @@ async def test_foreign_or_changed_origin_refused_before_any_reservation(live_own async def test_historical_unknown_author_stays_neutral_and_human_scope_invalidates(live_owner): comms, agent = live_owner origin = capture(comms) - neutral = ReservedInput("old", None, "beta", origin.claim_scope.claimed_generation, + neutral = ReservedInput("old", None, "beta", origin.claimed_identity.claimed_generation, "beta", "Exact old user-looking text") payload = FieldCodec.encode(neutral) assert "origin" not in payload diff --git a/tests/test_envelope_bus_integration.py b/tests/test_envelope_bus_integration.py index c65959a003..f4958417d4 100644 --- a/tests/test_envelope_bus_integration.py +++ b/tests/test_envelope_bus_integration.py @@ -81,10 +81,10 @@ def test_scoped_acquisition_is_one_original_and_conflict_appends_nothing(tmp_pat transition = sent.claim_transition assert transition.claims == (str(worktree / "a.py"),) assert (transition.message_id, transition.seq) == (sent.message_id, sent.seq) - assert isinstance(transition.claim_scope, NativeClaim) - assert transition.claim_scope.input_id == "input-alice" + assert isinstance(transition.resource_scope, NativeClaim) + assert transition.resource_scope.input_id == "input-alice" assert comms.bus.bus.revision() == sent.seq - assert comms.bus.bus.claim(sent.seq).claim_scope.input_id == "input-alice" + assert comms.bus.bus.claim(sent.seq).resource_scope.input_id == "input-alice" assert read_history(comms) == [sent] assert comms.bus.bus.claim_projection()[str(worktree / "a.py")].owner == "alice" with pytest.raises(ClaimConflict) as loss: diff --git a/tests/test_envelope_claim_transitions.py b/tests/test_envelope_claim_transitions.py index 2380007d9e..fa0283a624 100644 --- a/tests/test_envelope_claim_transitions.py +++ b/tests/test_envelope_claim_transitions.py @@ -33,7 +33,7 @@ def test_same_name_can_take_a_previous_birth_only_with_current_registry_evidence scope = NativeClaim(owner_claimed_generation=1, operation_id="c" * 32, turn_id="turn-owner", input_id="input-owner") held = WakeClaims("owner", "1.0", 1, "msg-1", claims=(resource,), - generation=G1, claim_scope=scope).project(ClaimProjection()) + generation=G1, resource_scope=scope).project(ClaimProjection()) original = held[resource] assert not original.live(registry) original.require_holder("owner", "1.0") @@ -41,7 +41,7 @@ def test_same_name_can_take_a_previous_birth_only_with_current_registry_evidence with pytest.raises(ClaimConflict): original.require_holder(owner, incarnation) successor = WakeClaims("owner", "2.0", 2, "msg-2", claims=(resource,), - generation=G2, claim_scope=scope) + generation=G2, resource_scope=scope) saved = FieldCodec.decode(ClaimTransition, FieldCodec.encode(successor)) moved = saved.apply(held, registry) assert moved[resource].incarnation == "2.0" and moved[resource].generation == G2 @@ -65,15 +65,15 @@ def test_selected_wake_binding_survives_claim_projection(tmp_path): attempt_ordinal=1, ) claimed = WakeClaims("owner", "epoch-1", 8, "msg-8", claims=(resource,), - generation=G1, claim_scope=claim_scope) + generation=G1, resource_scope=claim_scope) raw = FieldCodec.encode(claimed) decoded = FieldCodec.decode(ClaimTransition, raw) assert decoded == claimed - assert decoded.project(ClaimProjection())[resource].claim_scope == claim_scope + assert decoded.project(ClaimProjection())[resource].resource_scope == claim_scope message = Message("owner", "owner", "Claim file", ClaimNoticeKind, seq=8, timestamp=1.0) bound = ClaimMessage.of(message, replace(claimed, message_id=message.message_id)) assert Message.from_wire(bound.to_wire()) == bound - admission_key = FieldCodec.record_key(WakeClaims, "claim_scope") + admission_key = FieldCodec.record_key(WakeClaims, "resource_scope") invalid = {**raw, admission_key: {**raw[admission_key], "version": 99}} with pytest.raises(ValueError): FieldCodec.decode(ClaimTransition, invalid) diff --git a/tests/test_goal_input_review.py b/tests/test_goal_input_review.py index 6907c3ae05..7b6c8ad30d 100644 --- a/tests/test_goal_input_review.py +++ b/tests/test_goal_input_review.py @@ -106,7 +106,7 @@ def test_inspected_canonical_dependencies_allow_standby_without_replay(tmp_path, assert row.accepts_reservation assert not hasattr(row, "native_id") and not hasattr(row, "turn_id") assert row.source_text == message.body and not hasattr(row, "sent_text") - assert row.claim_scope == claim_scope and row.reviewed_for_goal(goal.id) + assert row.claim_generation == claim_scope and row.reviewed_for_goal(goal.id) assert not stored_inputs(ledger.root).rows["acp:owner-input"].goal_reviews generation = comms.goals.open_goal_store().snapshot(goal.id) assert generation is not None and generation.lifecycle.ready diff --git a/tests/test_input_documents.py b/tests/test_input_documents.py index 8c03303c70..6d6aebbf99 100644 --- a/tests/test_input_documents.py +++ b/tests/test_input_documents.py @@ -88,7 +88,7 @@ def test_saved_discriminator_and_optional_notices_roundtrip_without_replay(tmp_p from agent_comms.text_digest import TextDigest saved = FieldCodec.encode(BoundUnknownInput( - key="acp:old", sequence=None, owner="worker", claim_scope=1, + key="acp:old", sequence=None, owner="worker", claim_generation=1, target="worker", source_text="Retain me", turn_id="turn", native_id="a" * 32, sent_digest=TextDigest.of("Exact sent text"), notice_dismissed=True, goal_reviews={"goal": GoalInputDecision(goal_revision=7, turn_id="review")}, diff --git a/tests/test_interrupted_delivery_recovery.py b/tests/test_interrupted_delivery_recovery.py index 555ae70c1e..69a30e7037 100644 --- a/tests/test_interrupted_delivery_recovery.py +++ b/tests/test_interrupted_delivery_recovery.py @@ -59,7 +59,7 @@ async def test_interrupted_following_delivery_recovers_only_before_native_send( leased = comms.agents.begin_turn(owner.name, "interrupted-turn") assert agent.inputs.dispositions.bind( item.key, - claim_scope=row.claim_scope, + claim_scope=row.claim_generation, turn_id=leased.thread.require_turn_lease().turn_id, native_id="a" * 32, text=row.source_text, diff --git a/tests/test_retained_context_operations.py b/tests/test_retained_context_operations.py index 852fca2b54..000eb41ded 100644 --- a/tests/test_retained_context_operations.py +++ b/tests/test_retained_context_operations.py @@ -147,7 +147,7 @@ async def original_input_consumer_journey(tmp_path, command): reservation.release() assert agent.inputs.dispositions.record( "neutral-original", seq=None, owner="beta", - claim_scope=origin.claim_scope.claimed_generation, target="beta", text=wording) + claim_scope=origin.claimed_identity.claimed_generation, target="beta", text=wording) inputs = agent.inputs.dispositions original_inputs = inputs.path.read_bytes() other = comms.registry.declare(comms.registry.new_thread('unaddressed', frozenset(), origin.project)) diff --git a/tests/test_stopped_claim_expiry.py b/tests/test_stopped_claim_expiry.py index 4f0fa09d22..f129e3683a 100644 --- a/tests/test_stopped_claim_expiry.py +++ b/tests/test_stopped_claim_expiry.py @@ -33,7 +33,7 @@ def acquire(name, resources, operation): return comms.bus.publisher._publish_claim_transition_unlocked( boundary.registry, ClaimNoticeKind.claimed(owner), partial(WakeClaims, incarnation=str(owner.created_at), claims=resources, - generation=operation, claim_scope=scope), activity=boundary.activity, + generation=operation, resource_scope=scope), activity=boundary.activity, ) old_paths = tuple(str(tmp_path / name) for name in ("untouched.py", "other.py")) diff --git a/tests/test_thread_identity.py b/tests/test_thread_identity.py index 7bc6a58f8a..b11b940ea6 100644 --- a/tests/test_thread_identity.py +++ b/tests/test_thread_identity.py @@ -211,7 +211,9 @@ def test_claim_scope_cannot_decode_a_registry_process_owner(tmp_path): LiveCursorScope("official-session", "a" * 32, claim_scope, owner.require_process().pid)): encoded = FieldCodec.encode(scope) assert FieldCodec.decode(type(scope), encoded) == scope - encoded["claim_scope"] = FieldCodec.encode(registry_owner) + wire_key = FieldCodec.record_key(type(scope), "claimed_identity") + assert wire_key == "admission" + encoded[wire_key] = FieldCodec.encode(registry_owner) with pytest.raises(ValueError): FieldCodec.decode(type(scope), encoded) assert FieldCodec.decode(ClaimedIdentity, FieldCodec.encode(claim_scope)) == claim_scope diff --git a/tests/test_tool_output.py b/tests/test_tool_output.py index ff8f53959b..365e1fd3b4 100644 --- a/tests/test_tool_output.py +++ b/tests/test_tool_output.py @@ -25,7 +25,7 @@ def seed_unknown(comms, count, text): dispositions = InputDispositions(comms.root) rows = { f"bus:{index}": FieldCodec.encode(BoundUnknownInput( - key=f"bus:{index}", sequence=index, owner="b", claim_scope=index, + key=f"bus:{index}", sequence=index, owner="b", claim_generation=index, target="b", source_text=f"{index}: {text}", turn_id="private-turn", native_id="f" * 32, sent_digest=TextDigest.of("private-native-prompt"), )) diff --git a/tests/test_unscoped_claim_history.py b/tests/test_unscoped_claim_history.py index 727e2dc03f..cc3b0bfcb8 100644 --- a/tests/test_unscoped_claim_history.py +++ b/tests/test_unscoped_claim_history.py @@ -28,7 +28,7 @@ def test_unscoped_history_row_decodes_projects_never_live_and_releases(tmp_path) assert FieldCodec.encode(transition) == stored transition.require_new_operation(lambda operation: pytest.fail("no operation was recorded")) held = transition.project(ClaimProjection()) - assert held[resource].claim_scope == UnrecordedScope() + assert held[resource].resource_scope == UnrecordedScope() registry = comms.registry.snapshot() assert not held[resource].live(registry) @@ -37,7 +37,7 @@ def test_unscoped_history_row_decodes_projects_never_live_and_releases(tmp_path) taken = WakeClaims( owner="peer", incarnation=str(peer.created_at), seq=2, message_id="msg-2", claims=(resource,), generation=G2, - claim_scope=NativeClaim(owner_claimed_generation=1, operation_id="c" * 32, + resource_scope=NativeClaim(owner_claimed_generation=1, operation_id="c" * 32, turn_id="turn-2", input_id="input-2"), ).apply(held, registry) assert taken[resource].owner == "peer" From d5eac7a526328f18e13e1359d7a17d1703e39b46 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:21:31 -0400 Subject: [PATCH 4/7] Carry stored claim fields to their declared names --- src/agent_comms/acp_extension.py | 4 +- src/agent_comms/envelope_claim_transitions.py | 8 +- src/agent_comms/goal_failure_observation.py | 4 +- src/agent_comms/input_attempt.py | 2 +- src/agent_comms/input_origin.py | 4 +- src/agent_comms/native_runtime_input.py | 4 +- src/agent_comms/owner_lifecycle.py | 3 +- src/agent_comms/queued_input.py | 6 +- src/agent_comms/registry_document.py | 3 +- src/agent_comms/selected_write_plan.py | 8 +- src/agent_comms/thread_identity.py | 7 +- src/agent_comms/turn_lease.py | 5 +- tests/test_envelope_claim_transitions.py | 6 +- tests/test_registry_current_format.py | 2 +- tests/test_thread_identity.py | 2 +- tests/test_unscoped_claim_history.py | 4 +- tools/cutover/carry_bus_history.py | 25 +- tools/cutover/carry_claim_fields.py | 371 ++++++++++++++++++ 18 files changed, 420 insertions(+), 48 deletions(-) create mode 100644 tools/cutover/carry_claim_fields.py diff --git a/src/agent_comms/acp_extension.py b/src/agent_comms/acp_extension.py index cdd93cdca8..0e2df1139f 100644 --- a/src/agent_comms/acp_extension.py +++ b/src/agent_comms/acp_extension.py @@ -302,7 +302,7 @@ class NoCursorScope(CursorScope): class LiveCursorScope(CursorScope, LiveAttachmentScope): session_id: str wire_root_id: str - claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity owner: ProcessIdentity def __post_init__(self): @@ -446,7 +446,7 @@ class NoQueueScope(QueueScope): @dataclass(frozen=True) class LiveQueueScope(QueueScope, LiveAttachmentScope): session_id: str - claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity owner: ProcessIdentity @property diff --git a/src/agent_comms/envelope_claim_transitions.py b/src/agent_comms/envelope_claim_transitions.py index d3c14b724f..04f9ebd8a2 100644 --- a/src/agent_comms/envelope_claim_transitions.py +++ b/src/agent_comms/envelope_claim_transitions.py @@ -169,9 +169,7 @@ def held_at(self, claimed_generation: int) -> bool: class RecordedScope(ClaimScope, declared_name=None): """Every acquisition published today records the input generation and operation it serves.""" - owner_claimed_generation: Annotated[ - int, WireName("owner_admission_generation") - ] + owner_claimed_generation: int operation_id: Annotated[str, Id128Text] turn_id: str @@ -274,7 +272,7 @@ class ClaimIdentity: @dataclass(frozen=True) class AcquiringClaim(ClaimIdentity): generation: Annotated[str, Id128Text] - resource_scope: Annotated[ClaimScope, WireName("admission")] + resource_scope: ClaimScope @dataclass(frozen=True) @@ -390,7 +388,7 @@ def project(self, previous: ClaimProjection) -> ClaimProjection: class WakeClaims(RecordMember, Acquisition, AcquiringClaim): """An acquisition published today; its row records the scope it serves.""" - resource_scope: Annotated[RecordedScope, WireName("admission")] + resource_scope: RecordedScope @dataclass(frozen=True) diff --git a/src/agent_comms/goal_failure_observation.py b/src/agent_comms/goal_failure_observation.py index 31a8b211ba..c185cc3f04 100644 --- a/src/agent_comms/goal_failure_observation.py +++ b/src/agent_comms/goal_failure_observation.py @@ -13,7 +13,7 @@ from typing import TYPE_CHECKING, Annotated from .diagnostics import FailureReason -from .field_codec import Id128Text, WireName +from .field_codec import Id128Text from .goal_attempt_identity import GoalAttemptIdentity from .thread_identity import ( ClaimedIdentity, @@ -57,7 +57,7 @@ class FailedTurnEvidence(GoalLedgerTable, TypedTable, GoalFailureEvidence): owner_created_at: float worktree: str claim_generation: Annotated[ - int, WireName("admission"), Column(record_key=True, check="{claim_generation}>0") + int, Column(record_key=True, check="{claim_generation}>0") ] turn_generation: Annotated[int, Column(check="turn_generation>0")] goal_revision: Annotated[int, Column(check="goal_revision>=0")] diff --git a/src/agent_comms/input_attempt.py b/src/agent_comms/input_attempt.py index b3b0c84993..d2b5ea8c03 100644 --- a/src/agent_comms/input_attempt.py +++ b/src/agent_comms/input_attempt.py @@ -248,7 +248,7 @@ def require_original_provenance(self, source: InputProvenance): key: str sequence: int | None owner: str - claim_generation: Annotated[int, WireName("admission")] + claim_generation: int target: str source_text: str origin: Annotated[InputOrigin, WireOmitDefault()] = field( diff --git a/src/agent_comms/input_origin.py b/src/agent_comms/input_origin.py index a4555d7695..c0f85a9bfc 100644 --- a/src/agent_comms/input_origin.py +++ b/src/agent_comms/input_origin.py @@ -10,7 +10,7 @@ from .bus_publication import HumanOrigin from .declared_family import DeclaredFamily from .errors import RelationViolationError -from .field_codec import Id128Text, WireName +from .field_codec import Id128Text from .goals import GoalCheckpoint from .message_reference import MessageReference from .thread_identity import ClaimedIdentity @@ -120,7 +120,7 @@ class HumanInputOrigin(InputOrigin): root_id: Annotated[str, Id128Text] author: HumanOrigin - claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity project: str goal: GoalCheckpoint diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py index 90974227e8..077c9a8757 100644 --- a/src/agent_comms/native_runtime_input.py +++ b/src/agent_comms/native_runtime_input.py @@ -16,7 +16,7 @@ from agent_comms.private_runtime_schema import PrivateCoordinatorTable from .coordination_errors import IdentityConflict, PublicationActivationBlocked, StaleFence -from .field_codec import FieldCodec, MalformedRecord, NativeContext, WireName +from .field_codec import FieldCodec, MalformedRecord, NativeContext from .native_claim_epoch import NativeClaimEpoch, UnrecordedNativeClaim from .native_input_record import ( NativeColumn, @@ -264,7 +264,6 @@ def published_replies(self, db, user, owner_lookup): owner_token_digest: Annotated[str, FrozenColumn(check="length(owner_token_digest)=64")] sent_owner_claimed_generation: Annotated[ NativeClaimEpoch, - WireName("sent_owner_admission_generation"), Column( record_key=True, check="{sent_owner_claimed_generation}>0", @@ -365,7 +364,6 @@ def context_anchor(self) -> NativeColumn: owner_generation: Annotated[int, FrozenColumn(primary_key=True, check="owner_generation>0")] owner_claimed_generation: Annotated[ int, - WireName("owner_admission_generation"), FrozenColumn(record_key=True, primary_key=True, check="{owner_claimed_generation}>0"), ] diff --git a/src/agent_comms/owner_lifecycle.py b/src/agent_comms/owner_lifecycle.py index 4a02d8b060..0aafc481aa 100644 --- a/src/agent_comms/owner_lifecycle.py +++ b/src/agent_comms/owner_lifecycle.py @@ -13,7 +13,6 @@ from typing import TYPE_CHECKING, Annotated from uuid import uuid4 -from agent_comms import field_codec from .active_route import resolve_comms_route from agent_comms.attempt_recovery import RecoveryMonitorCapability @@ -95,7 +94,7 @@ def record_first_input(self, record) -> None: class OwnerRestartSelection: identity: OwnerIdentity process: ProcessIdentity - claimed_generation: Annotated[int, field_codec.WireName("admission_generation")] + claimed_generation: int interpreter: str @classmethod diff --git a/src/agent_comms/queued_input.py b/src/agent_comms/queued_input.py index e5d2408acd..4c9ce672f6 100644 --- a/src/agent_comms/queued_input.py +++ b/src/agent_comms/queued_input.py @@ -11,7 +11,7 @@ from contextlib import AsyncExitStack, ExitStack, asynccontextmanager from dataclasses import dataclass, field, replace from functools import partial -from typing import TYPE_CHECKING, Any, Generic, Self, TypeVar, Annotated +from typing import TYPE_CHECKING, Any, Generic, Self, TypeVar from acp import RequestError @@ -24,7 +24,7 @@ from .coordination_session import ReadNativeCoordinationSession from .coordinator import Coordination from .errors import RelationViolationError -from .field_codec import Id128Text, WireName +from .field_codec import Id128Text from .goals import GoalDeclaration, NoGoal from .image_inputs import ImageInput from .input_attempt import InputAttempt, ReservedInput @@ -55,7 +55,7 @@ class InputHandoffRefused(RelationViolationError): @dataclass(frozen=True, slots=True) class QueuedInputContext: - claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity goal: GoalDeclaration = field(default_factory=NoGoal) @classmethod diff --git a/src/agent_comms/registry_document.py b/src/agent_comms/registry_document.py index 2b3fb606db..769a943658 100644 --- a/src/agent_comms/registry_document.py +++ b/src/agent_comms/registry_document.py @@ -19,7 +19,6 @@ MalformedRecord, FieldPlan, SlicePolicy, - WireName, WireRequired, ) from .registration_change import ( @@ -323,7 +322,7 @@ class RegistryDocument(RegistryPresence[dict[str, ThreadStatus]]): default_factory=GenerationCounter ) claims: Annotated[ - GenerationCounter, WireName("admissions"), SlicePolicy(GenerationOf) + GenerationCounter, SlicePolicy(GenerationOf) ] = field(default_factory=GenerationCounter) _human: UserThreadLookup[str] = field(init=False, repr=False, compare=False) diff --git a/src/agent_comms/selected_write_plan.py b/src/agent_comms/selected_write_plan.py index 2ecbff2683..f4b3eac0f1 100644 --- a/src/agent_comms/selected_write_plan.py +++ b/src/agent_comms/selected_write_plan.py @@ -16,7 +16,7 @@ from contextlib import suppress from dataclasses import dataclass, fields from pathlib import Path -from typing import TYPE_CHECKING, Literal, Annotated +from typing import TYPE_CHECKING, Literal from .coordination_database import CoordinationStore from .locked_store import DurableFile, PatternFile @@ -33,7 +33,7 @@ from .coordination_cohort import sealed_cohort_assignments from .declared_family import DeclaredFamily from .envelope_claim_transitions import ExistingFileClaim -from .field_codec import FieldCodec, Id128Text, MalformedRecord, WireName +from .field_codec import FieldCodec, Id128Text, MalformedRecord from .message_reference import MessageReference from .private_path import PrivateDirectoryRole, PrivateFileRole from .thread_identity import ClaimedIdentity, RecordedThreadIncarnation @@ -54,7 +54,7 @@ class SelectedWriteBinding: root_id: str source: MessageReference assignment_id: str - claimed_identity: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity @dataclass(frozen=True, kw_only=True) @@ -69,7 +69,7 @@ class SelectedWriteIntent(DeclaredFamily, affix="WriteIntent"): claim_id: str owner: str incarnation: float - claim_epoch: Annotated[int, WireName("admission_epoch")] + claim_epoch: int expected_attempt_ordinal: Literal[1] operation_id: str resource: str diff --git a/src/agent_comms/thread_identity.py b/src/agent_comms/thread_identity.py index d6b96dfe14..35484efc26 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -12,7 +12,7 @@ from collections.abc import Iterable from dataclasses import dataclass, field, replace from functools import cached_property -from typing import TYPE_CHECKING, Callable, ClassVar, Self, TypeVar, Annotated +from typing import TYPE_CHECKING, Callable, ClassVar, Self, TypeVar from metaclass_registry import AutoRegisterMeta, RegistryConfig @@ -23,7 +23,6 @@ from .field_codec import ( FieldCodec, WireValue, - WireName, ) from .response_policy import CollectivePolicy, InformationalPolicy, ResponsePolicy @@ -350,7 +349,7 @@ class ClaimedIdentity: """A registry-allocated input generation, never an owner lease.""" incarnation: ThreadIncarnation - claimed_generation: Annotated[int, WireName("admission_generation")] + claimed_generation: int def includes(self, previous: ClaimedIdentity) -> bool: """Compare order only within this registry's input-generation counter.""" @@ -381,7 +380,7 @@ class TurnFence: identity: TurnIdentity turn_id: str - claimed_generation: Annotated[int, WireName("admission_generation")] + claimed_generation: int def renamed(self, name: str) -> Self: """Registry alias resolution changes the name, never incarnation or epochs.""" diff --git a/src/agent_comms/turn_lease.py b/src/agent_comms/turn_lease.py index ad212ffe0c..8006c5319c 100644 --- a/src/agent_comms/turn_lease.py +++ b/src/agent_comms/turn_lease.py @@ -17,7 +17,6 @@ RecordMember, SingleFieldValue, SubclassUnion, - WireName, WireTransient, boundary_decoder, ) @@ -212,9 +211,7 @@ class ActiveTurn(Executing, RecordMember, TurnDeclaration): owner: ProcessIdentity started_at: float = field(default_factory=time.time) routing: TurnRouting | None = None - claimed_generation: Annotated[ - TurnGeneration, WireName("admission_generation") - ] = UnrecordedGeneration() + claimed_generation: TurnGeneration = UnrecordedGeneration() turn_generation: TurnGeneration = UnrecordedGeneration() phase: TurnPhase = field(default_factory=PreparingPhase) def execution(self, source: GoalExecutionSource) -> GoalExecution: diff --git a/tests/test_envelope_claim_transitions.py b/tests/test_envelope_claim_transitions.py index fa0283a624..2a864b76f3 100644 --- a/tests/test_envelope_claim_transitions.py +++ b/tests/test_envelope_claim_transitions.py @@ -73,12 +73,12 @@ def test_selected_wake_binding_survives_claim_projection(tmp_path): message = Message("owner", "owner", "Claim file", ClaimNoticeKind, seq=8, timestamp=1.0) bound = ClaimMessage.of(message, replace(claimed, message_id=message.message_id)) assert Message.from_wire(bound.to_wire()) == bound - admission_key = FieldCodec.record_key(WakeClaims, "resource_scope") - invalid = {**raw, admission_key: {**raw[admission_key], "version": 99}} + scope_key = FieldCodec.record_key(WakeClaims, "resource_scope") + invalid = {**raw, scope_key: {**raw[scope_key], "version": 99}} with pytest.raises(ValueError): FieldCodec.decode(ClaimTransition, invalid) null_claim = raw.copy() - null_claim[admission_key] = None + null_claim[scope_key] = None with pytest.raises(ValueError): FieldCodec.decode(ClaimTransition, null_claim) diff --git a/tests/test_registry_current_format.py b/tests/test_registry_current_format.py index d7d6225e76..8fbc1e7a72 100644 --- a/tests/test_registry_current_format.py +++ b/tests/test_registry_current_format.py @@ -58,7 +58,7 @@ def test_incomplete_saved_identity_cannot_authorize_an_owner(tmp_path): registry = Registration(path) registry.register(registry.new_thread('owner', frozenset(), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()))) raw = json.loads(path.read_text()) - for field in ("owners", "admissions"): + for field in ("owners", "claims"): incomplete = dict(raw) incomplete.pop(field) path.write_text(json.dumps(incomplete)) diff --git a/tests/test_thread_identity.py b/tests/test_thread_identity.py index a9bf26e2e3..3034c32b9c 100644 --- a/tests/test_thread_identity.py +++ b/tests/test_thread_identity.py @@ -212,7 +212,7 @@ def test_claim_scope_cannot_decode_a_registry_process_owner(tmp_path): encoded = FieldCodec.encode(scope) assert FieldCodec.decode(type(scope), encoded) == scope wire_key = FieldCodec.record_key(type(scope), "claimed_identity") - assert wire_key == "admission" + assert wire_key == "claimed_identity" encoded[wire_key] = FieldCodec.encode(registry_owner) with pytest.raises(ValueError): FieldCodec.decode(type(scope), encoded) diff --git a/tests/test_unscoped_claim_history.py b/tests/test_unscoped_claim_history.py index cc3b0bfcb8..07c550ed40 100644 --- a/tests/test_unscoped_claim_history.py +++ b/tests/test_unscoped_claim_history.py @@ -49,6 +49,6 @@ def test_unscoped_history_row_decodes_projects_never_live_and_releases(tmp_path) # A row published today records its scope; history rows carry none. with pytest.raises(MalformedRecord): - FieldCodec.decode(ClaimTransition, {**stored, "admission": None}) + FieldCodec.decode(ClaimTransition, {**stored, "resource_scope": None}) scoped = FieldCodec.encode(taken[resource]) - assert scoped["admission"]["operation_id"] == "c" * 32 + assert scoped["resource_scope"]["operation_id"] == "c" * 32 diff --git a/tools/cutover/carry_bus_history.py b/tools/cutover/carry_bus_history.py index 19ec5a9327..377b009fdf 100644 --- a/tools/cutover/carry_bus_history.py +++ b/tools/cutover/carry_bus_history.py @@ -44,6 +44,7 @@ WireName, WireOmitDefault, WireOrder, + compile_decoder, ) from agent_comms.historical_views import HistoryArchive, HistorySource from agent_comms.installer.carry import Carried, Carry @@ -403,7 +404,10 @@ def from_wire(cls, data: object) -> Self: class HistoryScope(SubclassUnion, ABC, affix="HistoryScope"): def restored(self, scope: ClaimScope) -> Self: - return FieldCodec.decode_record(type(self), FieldCodec.encode(scope)) + raw = FieldCodec.encode(scope) + old = {"owner_admission_generation" if key == "owner_claimed_generation" else key: value + for key, value in raw.items()} + return FieldCodec.decode_record(type(self), old) @dataclass(frozen=True, kw_only=True) @@ -486,11 +490,17 @@ class HistoryClaim(SubclassUnion, ClaimIdentity, ABC, affix="HistoryClaim"): """Retired field-group order is part of the original envelope's physical form.""" def current(self) -> ClaimTransition: - return ClaimTransition.from_wire(FieldCodec.encode(self)) + from .carry_claim_fields import WIRE_NAMES + + raw = WIRE_NAMES.rewrite(compile_decoder(ClaimTransition), FieldCodec.encode(self)) + return ClaimTransition.from_wire(raw) + + def restored_wire(self, transition: ClaimTransition) -> dict[str, object]: + return FieldCodec.encode(transition) def restored(self, transition: ClaimTransition) -> Self: # The known old member owns layout; the new record supplies every value. - return FieldCodec.decode_record(type(self), FieldCodec.encode(transition)) + return FieldCodec.decode_record(type(self), self.restored_wire(transition)) class OrderedHistoryClaim(OrderedHistoryMember, HistoryClaim, declared_name=None): @@ -516,12 +526,13 @@ class ClaimGeneration: class ScopedClaimGeneration(ClaimGeneration, HistoryClaim, declared_name=None): scope: Annotated[HistoryScope, WireName("admission")] - def restored(self, transition: ClaimTransition) -> Self: + def restored_wire(self, transition: ClaimTransition) -> dict[str, object]: # This retired member decoded an acquiring scope, never a release. acquiring = cast(AcquiringClaim, transition) - return replace( - super().restored(transition), scope=self.scope.restored(acquiring.claim_scope) - ) + scope = FieldCodec.encode(self.scope.restored(acquiring.resource_scope)) + return {"admission" if key == "resource_scope" else key: + scope if key == "resource_scope" else value + for key, value in super().restored_wire(transition).items()} @dataclass(frozen=True, kw_only=True) diff --git a/tools/cutover/carry_claim_fields.py b/tools/cutover/carry_claim_fields.py new file mode 100644 index 0000000000..b46cac6a57 --- /dev/null +++ b/tools/cutover/carry_claim_fields.py @@ -0,0 +1,371 @@ +"""Give retained claim fields their declared names, once, under install maintenance. + +The old spelling is selected by the record declaration, never by the JSON value. +Opaque provider JSON is not a record and is never searched for matching strings. +Delete this step after every retained root has crossed this format change. +""" + +from __future__ import annotations + +import json +from contextlib import closing +from dataclasses import dataclass +from functools import cache +from pathlib import Path + +from agent_comms.acp_extension import LiveCursorScope, LiveQueueScope +from agent_comms.envelope_claim_transitions import AcquiringClaim, RecordedScope +from agent_comms.errors import RelationViolationError +from agent_comms.field_codec import ( + FieldCodec, + InlineFieldPlan, + InlineRecord, + MalformedRecord, + RecordMember, + SingleFieldValue, + SubclassUnion, + UnionDecoder, + ValueDecoder, + _ArrayDecoder, + _DictDecoder, + _FamilyRecordDecoder, + _RecordDecoder, + _RepresentedDecoder, + _StringKeyDictDecoder, + _PairKeyDictDecoder, + compile_decoder, +) +from agent_comms.goal_failure_observation import FailedTurnEvidence +from agent_comms.historical_views import HistoryArchive +from agent_comms.input_attempt import StoredInput +from agent_comms.input_origin import HumanInputOrigin +from agent_comms.installer.carry import Carried, Carry +from agent_comms.locked_store import JsonDocument, RegistryFile, RootFile, SQLiteFile +from agent_comms.mro_dispatch import MroDispatch, handles +from agent_comms.native_runtime_input import CurrentNativeCursor, NativeRuntimeInput +from agent_comms.owner_lifecycle import OwnerRestartSelection +from agent_comms.queued_input import QueuedInputContext +from agent_comms.registry_document import RegistryDocument +from agent_comms.registry_store import RegistryStore +from agent_comms.restart_queue import DIRECTORY, QueuedRestart +from agent_comms.selected_write_plan import ( + SelectedWriteBinding, + SelectedWriteIntent, + SelectedWritePlans, +) +from agent_comms.store_files import _atomic_write_text, _store_lock +from agent_comms.thread_identity import ClaimedIdentity, TurnFence +from agent_comms.turn_lease import ActiveTurn +from agent_comms.typed_table import HeldBusFile, JsonStorage, TypedTable, _identifier + +from .carry_bus_history import CarryCatalogSource +from .carry_context_request_key import ContextRequestKey +from .carry_owner_requests import OwnerRequests +from .carry_turn_owner import TurnOwner + + +@dataclass(frozen=True) +class RenamedField: + owner: type + name: str + old: str + + +# These are the retired declarations. Current spelling, JSON layout, SQL storage, +# descendants and consumers all come from FieldCodec/TypedTable, not this list. +RENAMED = ( + RenamedField(LiveCursorScope, "claimed_identity", "admission"), + RenamedField(LiveQueueScope, "claimed_identity", "admission"), + RenamedField(HumanInputOrigin, "claimed_identity", "admission"), + RenamedField(QueuedInputContext, "claimed_identity", "admission"), + RenamedField(SelectedWriteBinding, "claimed_identity", "admission"), + RenamedField(AcquiringClaim, "resource_scope", "admission"), + RenamedField(RecordedScope, "owner_claimed_generation", "owner_admission_generation"), + RenamedField(StoredInput, "claim_generation", "admission"), + RenamedField(FailedTurnEvidence, "claim_generation", "admission"), + RenamedField(ClaimedIdentity, "claimed_generation", "admission_generation"), + RenamedField(TurnFence, "claimed_generation", "admission_generation"), + RenamedField(ActiveTurn, "claimed_generation", "admission_generation"), + RenamedField(OwnerRestartSelection, "claimed_generation", "admission_generation"), + RenamedField(SelectedWriteIntent, "claim_epoch", "admission_epoch"), + RenamedField(RegistryDocument, "claims", "admissions"), + RenamedField(NativeRuntimeInput, "sent_owner_claimed_generation", "sent_owner_admission_generation"), + RenamedField(CurrentNativeCursor, "owner_claimed_generation", "owner_admission_generation"), +) + + +@cache +def renames(record: type) -> dict[str, str]: + return {item.name: item.old for item in RENAMED if issubclass(record, item.owner)} + + +class ClaimFieldNames(MroDispatch): + """Walk the existing codec's plans; only the retired field names change.""" + + @staticmethod + def represented(plan: _RepresentedDecoder) -> ValueDecoder: + owner = plan.representation + if issubclass(owner, RecordMember): + return _RecordDecoder(owner) + if issubclass(owner, SingleFieldValue): + return owner.wire_field().value + if issubclass(owner, SubclassUnion): + return ValueDecoder.for_members(owner.members_with(owner)) + if issubclass(owner, InlineRecord): + return _RecordDecoder(owner.wire_record()) + return plan.structural + + def children(self, plan: ValueDecoder) -> tuple[ValueDecoder, ...]: + return next(self.handlers_for(plan))(plan, None) + + @cache + def touches(self, plan: ValueDecoder, seen: frozenset[object] = frozenset()) -> bool: + if plan in seen: + return False + if isinstance(plan, _RecordDecoder) and renames(plan.target): + return True + return any(self.touches(child, seen | {plan}) for child in self.children(plan)) + + def rewrite(self, plan: ValueDecoder, raw: object) -> object: + if raw is None or not self.touches(plan): + return raw + return next(self.handlers_for(plan))(plan, raw) + + @handles(ValueDecoder) + def opaque(self, plan: ValueDecoder, raw: object) -> object: + return () if raw is None else raw + + @handles(_RepresentedDecoder) + def representation(self, plan: _RepresentedDecoder, raw: object) -> object: + child = self.represented(plan) + return (child,) if raw is None else self.rewrite(child, raw) + + @handles(_FamilyRecordDecoder) + def family(self, plan: _FamilyRecordDecoder, raw: object) -> object: + if raw is None: + return tuple(_RecordDecoder(member) for member in plan.target.members_with(plan.target)) + member, _ = plan.member(raw) + return self.rewrite(_RecordDecoder(member), raw) + + @handles(_RecordDecoder) + def record(self, plan: _RecordDecoder, raw: object) -> object: + entries = tuple(entry for entry in FieldCodec.field_plans(plan.target) + if entry.records and entry.declared.init) + if raw is None: + return tuple(entry.value for entry in entries) + if not isinstance(raw, dict): + raise MalformedRecord(f"Expected {plan.target.__name__} record") + result = dict(raw) + keys = renames(plan.target) + for entry in entries: + old = keys.get(entry.name) + if old is not None and old in result: + if entry.key in result: + raise RelationViolationError( + f"{plan.target.__name__} contains both {old} and {entry.key}") + result = {entry.key if key == old else key: value + for key, value in result.items()} + if isinstance(entry, InlineFieldPlan): + result = self.rewrite(entry.value, result) + elif entry.key in result: + result[entry.key] = self.rewrite(entry.value, result[entry.key]) + return result + + @handles(UnionDecoder) + def union(self, plan: UnionDecoder, raw: object) -> object: + if raw is None: + return plan.alternatives + answers = [] + for alternative in plan.alternatives: + try: + changed = self.rewrite(alternative, raw) + alternative.decode_bound(changed, {}) + except MalformedRecord: + continue + if changed not in answers: + answers.append(changed) + if len(answers) != 1: + raise RelationViolationError( + f"Claim field carry cannot select exactly one declared {plan.target}: " + f"{len(answers)} different results") + return answers[0] + + @handles(_ArrayDecoder) + def array(self, plan: _ArrayDecoder, raw: object) -> object: + if raw is None: + return plan.items + return [self.rewrite(child, item) for child, item in plan.plan_items(raw)] + + @handles(_DictDecoder) + def mapping(self, plan: _DictDecoder, raw: object) -> object: + return (plan.key, plan.value) + + @handles(_StringKeyDictDecoder) + def string_mapping(self, plan: _StringKeyDictDecoder, raw: object) -> object: + if raw is None: + return self.mapping(plan, raw) + return {key: self.rewrite(plan.value, item) for key, item in plan.entries(raw)} + + @handles(_PairKeyDictDecoder) + def pair_mapping(self, plan: _PairKeyDictDecoder, raw: object) -> object: + if raw is None: + return self.mapping(plan, raw) + return [[self.rewrite(plan.key, key), self.rewrite(plan.value, item)] + for key, item in plan.entries(raw)] + + +WIRE_NAMES = ClaimFieldNames() + + +class ClaimFields(Carry): + precedes = (OwnerRequests, TurnOwner, ContextRequestKey) + + @staticmethod + def documents(root: Path, files: dict[Path, type[RootFile]]) -> dict[Path, object]: + documents = {path: owner.record_type for path, owner in files.items() + if issubclass(owner, JsonDocument) and not issubclass(owner, RegistryStore)} + documents.update((path, SelectedWriteIntent) for path in SelectedWritePlans.matching_in( + root / SelectedWritePlans.root_directory)) + documents.update((path, QueuedRestart) for path in (root / DIRECTORY).rglob("*.json")) + return documents + + def roots(self) -> tuple[Path, ...]: + return (self.root, *(source.mapped_root(self.root, self.original_root) + for source in CarryCatalogSource.read(self.root, self.original_root))) + + def backup_files(self) -> tuple[Path, ...]: + paths = set() + for root in self.roots(): + files = RootFile.existing_in(root) + paths.update(path for path, owner in files.items() if issubclass(owner, SQLiteFile)) + paths.update(self.documents(root, files)) + paths.update(RegistryStore.at(root).written) + paths.add(HistoryArchive.path_in(self.root)) + for path in paths: + self.require_confined(self.root, path) + return tuple(sorted(path for path in paths if path.exists())) + + def carry(self, *, apply: bool) -> Carried: + if apply: + HeldBusFile(RegistryFile, RegistryFile.path_in(self.root)).require() + rewritten = [] + changed = 0 + visitor = WIRE_NAMES + with _store_lock(HistoryArchive.path_in(self.root)): + sources = CarryCatalogSource.read(self.root, self.original_root) + for root in self.roots(): + registry = RegistryStore.at(root) + if registry.path.exists(): + with registry.locked(): + registry.private_guard_unlocked() + raw = json.loads(registry.path.read_bytes()) + current = visitor.rewrite(compile_decoder(RegistryDocument), raw) + if current != raw: + changed += 1 + if apply: + registry._write_unlocked(json.dumps(current, indent=2)) + rewritten.extend(str(path.relative_to(self.root)) + for path in registry.written) + files = RootFile.existing_in(root) + for path, owner in files.items(): + self.require_confined(self.root, path) + if issubclass(owner, SQLiteFile): + count = self.database(path, visitor, apply=apply) + changed += count + if count and apply: + rewritten.append(str(path.relative_to(self.root))) + for path, declared in self.documents(root, files).items(): + self.require_confined(self.root, path) + plan = compile_decoder(declared) + if not visitor.touches(plan): + continue + with _store_lock(path): + raw = json.loads(path.read_bytes()) + current = visitor.rewrite(plan, raw) + count = int(current != raw) + if count and apply: + _atomic_write_text(path, json.dumps(current)) + changed += count + if count and apply: + rewritten.append(str(path.relative_to(self.root))) + if apply and rewritten: + rebound = tuple(source.carried_registry(source.mapped_root(self.root, self.original_root)) + for source in sources) + if rebound: + _atomic_write_text(HistoryArchive.path_in(self.root), + json.dumps(FieldCodec.encode(rebound))) + rewritten.append(HistoryArchive.filename) + return Carried(changed, rewritten=tuple(rewritten)) + + @staticmethod + def database(path: Path, visitor: ClaimFieldNames, *, apply: bool) -> int: + # Read only typed JSON columns that can contain these declarations. The + # multi-gigabyte message bodies are neither copied into RAM nor decoded. + pending = [] + rebuilds = [] + changed = 0 + with closing(SQLiteFile.read_snapshot(path)) as db: + tables = tuple(row for row in TypedTable.members_with(TypedTable) + if row.stored_columns(db)) + for row in tables: + stored = row.stored_columns(db) + renamed = {field.name: _identifier(old) + for field in row._fields() + if field.column.record_key + and (old := renames(row).get(field.name)) in stored} + if any(field.wire_name in stored for field in row._fields() + if field.name in renamed): + raise RelationViolationError(f"{row.__name__} contains old and new claim columns") + json_fields = tuple(field for field in row._fields() + if issubclass(field.storage, JsonStorage) and field.wire_name in stored + and visitor.touches(field.value)) + primary = tuple(field for field in row._fields() if field.column.primary_key) + if json_fields: + selected = (*primary, *json_fields) + columns = ",".join(renamed.get(field.name, _identifier(field.wire_name)) + for field in selected) + where = " AND ".join( + f"{renamed.get(field.name, _identifier(field.wire_name))} IS ?" + for field in primary) + if not primary: + raise RelationViolationError(f"{row.__name__} has no declared row key") + with closing(db.execute(f"SELECT {columns} FROM {_identifier(row.declared_name)}")) as rows: + for values in rows: + keys, payloads = values[:len(primary)], values[len(primary):] + updates = {} + for field, value in zip(json_fields, payloads, strict=True): + if value is None: + continue + raw = json.loads(value) + current = visitor.rewrite(field.value, raw) + if current != raw: + updates[field.wire_name] = json.dumps(current) + if updates: + assignments = ",".join(f"{_identifier(key)}=?" for key in updates) + updates_for_row = (f"UPDATE {_identifier(row.declared_name)} SET {assignments} WHERE {where}", + (*updates.values(), *keys)) + # Keep only changed rows, not the scanned source. + pending.append(updates_for_row) + changed += 1 + if renamed: + rebuilds.append((row, renamed)) + changed += 1 + if not apply or not changed: + return changed + with _store_lock(path), closing(SQLiteFile.writer(path, isolation_level=None)) as db: + triggers = db.execute("SELECT name,sql FROM sqlite_master WHERE type='trigger'").fetchall() + db.execute("PRAGMA foreign_keys=OFF") + with SQLiteFile.write_transaction(db): + for name, _ in triggers: + db.execute(f"DROP TRIGGER {_identifier(name)}") + for query, arguments in pending: + db.execute(query, arguments) + for row, renamed in rebuilds: + row.rebuild(db, renamed) + present = {name for (name,) in db.execute("SELECT name FROM sqlite_master WHERE type='trigger'")} + for name, sql in triggers: + if name not in present: + db.execute(sql) + if problems := db.execute("PRAGMA foreign_key_check").fetchall(): + raise RelationViolationError(f"Claim carry broke references in {path}: {problems}") + return changed From ab91e5d7ce05649a27e6ceeebc6fbbb701461e4b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:32:53 -0400 Subject: [PATCH 5/7] Combine failed-turn reason and claim column carry in one rebuild --- tools/cutover/carry_declared_failures.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tools/cutover/carry_declared_failures.py b/tools/cutover/carry_declared_failures.py index 830e1e2b22..123c8b75da 100644 --- a/tools/cutover/carry_declared_failures.py +++ b/tools/cutover/carry_declared_failures.py @@ -19,12 +19,17 @@ from agent_comms.installer.carry import Carried, Carry from agent_comms.locked_store import SQLiteFile from agent_comms.store_files import _atomic_write_text -from agent_comms.typed_table import HeldBusFile, sql_literal +from agent_comms.typed_table import HeldBusFile, _identifier, sql_literal from .carry_bus_history import CarryCatalogSource +from .carry_claim_fields import ClaimFields, renames class DeclaredFailures(Carry): + # Both changes affect one table. Rebuild it once with both source columns; + # ClaimFields then observes the already-carried SQL claim field. + precedes = (ClaimFields,) + old_names: ClassVar[dict[str, type[failures.TurnFailure]]] = { "backend_failed": failures.BackendFailed, "native_preflight_timeout": failures.NativePreflightTimeout, @@ -89,7 +94,14 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: f"WHEN {sql_literal(old)} THEN {sql_literal(kind)}" for old, kind in self.old_names.items() ) - FailedTurnEvidence.rebuild(prepared, {"failure": f"CASE reason {cases} END"}) + replacements = {"failure": f"CASE reason {cases} END"} + retired = renames(FailedTurnEvidence) + replacements.update({ + field.name: _identifier(retired[field.name]) + for field in FailedTurnEvidence._fields() + if field.name in retired and retired[field.name] in columns + }) + FailedTurnEvidence.rebuild(prepared, replacements) changed += 1 if AttemptRecord.stored_columns(prepared): for old, kind in self.old_names.items(): From e399a11542b88308922cf9c9b032a806ac9925c1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:39:19 -0400 Subject: [PATCH 6/7] Move stored-record traversal into its codec owners --- src/agent_comms/field_codec.py | 192 +++++++++++++++++++++++++++- tools/cutover/carry_bus_history.py | 5 +- tools/cutover/carry_claim_fields.py | 175 +++++-------------------- 3 files changed, 220 insertions(+), 152 deletions(-) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index 7cc42ffe9f..f01d704cef 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -48,6 +48,9 @@ T = TypeVar("T") +type WireRecordChange = Callable[[type, dict[str, object]], dict[str, object]] +type WireRecordFilter = Callable[[type], bool] + class MalformedRecord(ValueError): # noqa: N818 - names the refused data """Data that does not satisfy its declaration. @@ -508,6 +511,18 @@ def wire_keys(self) -> tuple[str, ...]: def encoded(self, value, *, key=None): return {key or self.key: value} + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return (self.value,) if self.wire_keys else () + + def rewrite_wire( + self, data: dict[str, object], change: WireRecordChange, select: WireRecordFilter, + ) -> dict[str, object]: + if self.key not in data or not self.wire_keys: + return data + current = self.value.rewrite_records(data[self.key], change, select) + return data if current is data[self.key] else {**data, self.key: current} + def schema_fields(self): annotation, declared = self.annotation, self.declared base = FieldCodec._representation(annotation)[0] @@ -680,6 +695,11 @@ def wire_keys(self) -> tuple[str, ...]: def encoded(self, value, *, key=None): return value + def rewrite_wire( + self, data: dict[str, object], change: WireRecordChange, select: WireRecordFilter, + ) -> dict[str, object]: + return cast(dict[str, object], self.value.rewrite_records(data, change, select)) + def captured(self, data: object) -> dict[str, object]: if not isinstance(data, dict): raise MalformedRecord("Expected a record object.") @@ -796,6 +816,11 @@ class FieldRepresentation(MroDispatch, ABC): field_plan: ClassVar[type[FieldPlan]] = DeclaredFieldPlan + @classmethod + def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]: + """A custom scalar form stays opaque; declared structural forms supply their plan.""" + return structural + @classmethod def decode_field(cls, data: object, structural: ValueDecoder[object]) -> object: handler = getattr(cls, next(cls.handler_names_for(type(data)))) @@ -894,6 +919,10 @@ def wire_field(cls): entry, = FieldCodec._record_plan(cls).fields return entry + @classmethod + def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]: + return cls.wire_field().value + def to_wire(self): entry = self.wire_field() return FieldCodec.encode_document(getattr(self, entry.name), entry.annotation) @@ -923,6 +952,10 @@ class SubclassUnion(DeclaredFamily, AcceptsNull, ExternalJsonShape, WireValue): def decodes_tagged_records(cls) -> bool: return False + @classmethod + def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]: + return ValueDecoder.for_members(cls.members_with(cls)) + @classmethod def from_wire(cls, data: object) -> Self: return ValueDecoder.for_members(cls.members_with(cls)).decode_bound(data, {}) @@ -962,6 +995,10 @@ def schema(cls) -> dict[str, object]: class RecordMember(WireValue): """A wire union's record case uses the declared record codec.""" + @classmethod + def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]: + return _member_decoder(cls) + @classmethod def from_wire(cls, data: object) -> Self: return FieldCodec.decode_record(cls, data) @@ -983,6 +1020,10 @@ class InlineRecord(WireValue, ExternalJsonShape): def wire_record(cls): return cls + @classmethod + def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]: + return _member_decoder(cls.wire_record()) + @classmethod def schema(cls): return FieldCodec.record_schema(cls.wire_record()) @@ -1736,6 +1777,46 @@ def decode(self, data: object) -> T: @abstractmethod def encode(self, value: T) -> object: ... + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return () + + def selects_record(self, select: WireRecordFilter) -> bool: + return False + + @cache + def contains_records( + self, select: WireRecordFilter, seen: frozenset[object] = frozenset(), + ) -> bool: + """Whether this immutable plan reaches a selected declared record; no payload read.""" + if self in seen: + return False + return self.selects_record(select) or any( + child.contains_records(select, seen | {self}) for child in self.wire_children + ) + + def rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + """Transform declared raw records without decoding opaque JSON or losing other keys. + + The caller owns the record change. This decoder owns wire layout, child + membership and union selection. ``select`` is a pure declaration predicate. + """ + if data is None or not self.contains_records(select): + return data + return self._rewrite_records(data, change, select) + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + return data + + def rewrite_document( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + return self.rewrite_records(data, change, select) + def decode_external(self, data: object, record: Mapping[str, object]) -> T: for decode in self.external_decoders(data): return self.decode_resolved(cast(T, decode(data, record))) @@ -1907,6 +1988,20 @@ class _RepresentedDecoder(ValueDecoder[Any]): representation: type[FieldRepresentation] structural: ValueDecoder[object] + @property + @cache + def wire_structure(self) -> ValueDecoder[object]: + return self.representation.wire_structure(self.structural) + + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return (self.wire_structure,) + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + return self.wire_structure.rewrite_records(data, change, select) + def external_decoders(self, data: object) -> Iterable[ExternalDecoder]: return FieldCodec.external_decoders(self.representation) @@ -2056,6 +2151,28 @@ class UnionDecoder[T](_ContextDecoder[T]): origins = (Union, types.UnionType) alternatives: tuple[ValueDecoder[T], ...] + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return self.alternatives + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + answers = [] + for alternative in self.alternatives: + try: + current = alternative.rewrite_records(data, change, select) + alternative.decode_bound(current, {}) + except MalformedRecord: + continue + if current not in answers: + answers.append(current) + if len(answers) != 1: + raise MalformedRecord( + f"Record rewrite cannot select exactly one declared {self.target}: " + f"{len(answers)} different results") + return answers[0] + @classmethod def from_annotation(cls, target, args): return cls(target, tuple(compile_decoder(item) for item in args)) @@ -2118,6 +2235,17 @@ class _ArrayDecoder(ValueDecoder[Any]): container: ClassVar[type] items: tuple[ValueDecoder[object], ...] + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return self.items + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + if not isinstance(data, list): + raise MalformedRecord("Expected a JSON array.") + return [plan.rewrite_records(value, change, select) for plan, value in self.plan_items(data)] + @classmethod def from_annotation(cls, target, args): return cls(target, tuple(compile_decoder(item) for item in (args or (Any,)))) @@ -2223,6 +2351,23 @@ class _DictDecoder(ValueDecoder[Any]): value: ValueDecoder[object] origins = (dict,) + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return self.key, self.value + + @staticmethod + @abstractmethod + def wire_mapping(entries: Iterable[tuple[object, object]]) -> object: ... + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + return self.wire_mapping( + (self.key.rewrite_records(key, change, select), + self.value.rewrite_records(value, change, select)) + for key, value in self.entries(data) + ) + def argument_options(self) -> dict[str, object]: return {"type": FieldCodec.parse} @@ -2269,8 +2414,12 @@ def decode_keys(self, data, keys, selections): class _StringKeyDictDecoder(_DictDecoder): entries = staticmethod(_object_entries) + @staticmethod + def wire_mapping(entries: Iterable[tuple[object, object]]) -> dict: + return dict(entries) + def encode(self, value: dict[str, object]) -> dict[str, object]: - return {key: self.value.encode(item) for key, item in value.items()} + return self.wire_mapping((key, self.value.encode(item)) for key, item in value.items()) def encode_document(self, value: dict[str, object]) -> dict[str, object]: return {key: self.value.encode_document(item) for key, item in value.items()} @@ -2283,9 +2432,13 @@ def decode_document(self, data: object) -> dict[str, object]: class _PairKeyDictDecoder(_DictDecoder): entries = staticmethod(_object_or_pair_entries) + @staticmethod + def wire_mapping(entries: Iterable[tuple[object, object]]) -> list: + return [[key, value] for key, value in entries] + def encode(self, value: dict[object, object]) -> list[list[object]]: - return [[self.key.encode(key), self.value.encode(item)] - for key, item in value.items()] + return self.wire_mapping((self.key.encode(key), self.value.encode(item)) + for key, item in value.items()) @dataclass(frozen=True, slots=True) @@ -2294,6 +2447,16 @@ class _FamilyRecordDecoder[Member: DeclaredFamily](_ContextDecoder[Member]): target: type[Member] + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return tuple(_member_decoder(member) for member in self.target.members_with(self.target)) + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + member, _ = self.member(data) + return _member_decoder(member).rewrite_records(data, change, select) + def encode(self, value: Member) -> dict[str, object]: return {**value.record_tag(), **_member_decoder(type(value)).encode(value)} @@ -2336,6 +2499,11 @@ def encode_document(self, value: DeclaredFamily) -> object: def decode_document(self, data: object) -> object: return _member_decoder(self.target).decode_bound(data, {}) + def rewrite_document( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + return _member_decoder(self.target).rewrite_records(data, change, select) + @dataclass(frozen=True, slots=True) class _RecordDecoder(_ContextDecoder[Any]): @@ -2344,6 +2512,24 @@ class _RecordDecoder(_ContextDecoder[Any]): target: type + @property + def wire_children(self) -> tuple[ValueDecoder[object], ...]: + return tuple(child for entry in FieldCodec._record_plan(self.target).bindings + for child in entry.wire_children) + + def selects_record(self, select: WireRecordFilter) -> bool: + return select(self.target) + + def _rewrite_records( + self, data: object, change: WireRecordChange, select: WireRecordFilter, + ) -> object: + if not isinstance(data, dict): + raise MalformedRecord(f"Expected {self.target.__name__} record") + result = change(self.target, data) + for entry in FieldCodec._record_plan(self.target).bindings: + result = entry.rewrite_wire(result, change, select) + return result + def encode(self, value: object) -> dict[str, object]: result = FieldCodec._record_plan(type(value)).encoded( value, "wire", lambda entry, current: entry.value.encode(current) diff --git a/tools/cutover/carry_bus_history.py b/tools/cutover/carry_bus_history.py index a59d133f89..ece90ef4c3 100644 --- a/tools/cutover/carry_bus_history.py +++ b/tools/cutover/carry_bus_history.py @@ -44,7 +44,6 @@ WireName, WireOmitDefault, WireOrder, - compile_decoder, ) from agent_comms.historical_views import HistoryArchive, HistorySource from agent_comms.installer.carry import Carried, Carry @@ -490,9 +489,9 @@ class HistoryClaim(SubclassUnion, ClaimIdentity, ABC, affix="HistoryClaim"): """Retired field-group order is part of the original envelope's physical form.""" def current(self) -> ClaimTransition: - from .carry_claim_fields import WIRE_NAMES + from .carry_claim_fields import ClaimFields - raw = WIRE_NAMES.rewrite(compile_decoder(ClaimTransition), FieldCodec.encode(self)) + raw = ClaimFields.convert(ClaimTransition, FieldCodec.encode(self)) return ClaimTransition.from_wire(raw) def restored_wire(self, transition: ClaimTransition) -> dict[str, object]: diff --git a/tools/cutover/carry_claim_fields.py b/tools/cutover/carry_claim_fields.py index a293d2f18f..e1930379c7 100644 --- a/tools/cutover/carry_claim_fields.py +++ b/tools/cutover/carry_claim_fields.py @@ -16,32 +16,13 @@ from agent_comms.acp_extension import LiveCursorScope, LiveQueueScope from agent_comms.envelope_claim_transitions import AcquiringClaim, RecordedScope from agent_comms.errors import RelationViolationError -from agent_comms.field_codec import ( - FieldCodec, - InlineFieldPlan, - InlineRecord, - MalformedRecord, - RecordMember, - SingleFieldValue, - SubclassUnion, - UnionDecoder, - ValueDecoder, - _ArrayDecoder, - _DictDecoder, - _FamilyRecordDecoder, - _RecordDecoder, - _RepresentedDecoder, - _StringKeyDictDecoder, - _PairKeyDictDecoder, - compile_decoder, -) +from agent_comms.field_codec import FieldCodec, compile_decoder from agent_comms.goal_failure_observation import FailedTurnEvidence from agent_comms.historical_views import HistoryArchive from agent_comms.input_attempt import StoredInput from agent_comms.input_origin import HumanInputOrigin from agent_comms.installer.carry import Carried, Carry from agent_comms.locked_store import JsonDocument, RegistryFile, RootFile, SQLiteFile -from agent_comms.mro_dispatch import MroDispatch, handles from agent_comms.native_runtime_input import CurrentNativeCursor, NativeRuntimeInput from agent_comms.owner_lifecycle import OwnerRestartSelection from agent_comms.queued_input import QueuedInputContext @@ -96,131 +77,34 @@ class RenamedField: ) -@cache -def renames(record: type) -> dict[str, str]: - return {item.name: item.old for item in RENAMED if issubclass(record, item.owner)} - - -class ClaimFieldNames(MroDispatch): - """Walk the existing codec's plans; only the retired field names change.""" - - @staticmethod - def represented(plan: _RepresentedDecoder) -> ValueDecoder: - owner = plan.representation - if issubclass(owner, RecordMember): - return _RecordDecoder(owner) - if issubclass(owner, SingleFieldValue): - return owner.wire_field().value - if issubclass(owner, SubclassUnion): - return ValueDecoder.for_members(owner.members_with(owner)) - if issubclass(owner, InlineRecord): - return _RecordDecoder(owner.wire_record()) - return plan.structural - - def children(self, plan: ValueDecoder) -> tuple[ValueDecoder, ...]: - return next(self.handlers_for(plan))(plan, None) +class ClaimFields(Carry): + precedes = (FinishedTurns, DeadTurns, OwnerRequests, TurnOwner, ContextRequestKey) + @classmethod @cache - def touches(self, plan: ValueDecoder, seen: frozenset[object] = frozenset()) -> bool: - if plan in seen: - return False - if isinstance(plan, _RecordDecoder) and renames(plan.target): - return True - return any(self.touches(child, seen | {plan}) for child in self.children(plan)) - - def rewrite(self, plan: ValueDecoder, raw: object) -> object: - if raw is None or not self.touches(plan): - return raw - return next(self.handlers_for(plan))(plan, raw) - - @handles(ValueDecoder) - def opaque(self, plan: ValueDecoder, raw: object) -> object: - return () if raw is None else raw - - @handles(_RepresentedDecoder) - def representation(self, plan: _RepresentedDecoder, raw: object) -> object: - child = self.represented(plan) - return (child,) if raw is None else self.rewrite(child, raw) - - @handles(_FamilyRecordDecoder) - def family(self, plan: _FamilyRecordDecoder, raw: object) -> object: - if raw is None: - return tuple(_RecordDecoder(member) for member in plan.target.members_with(plan.target)) - member, _ = plan.member(raw) - return self.rewrite(_RecordDecoder(member), raw) - - @handles(_RecordDecoder) - def record(self, plan: _RecordDecoder, raw: object) -> object: - entries = tuple(entry for entry in FieldCodec.field_plans(plan.target) - if entry.records and entry.declared.init) - if raw is None: - return tuple(entry.value for entry in entries) - if not isinstance(raw, dict): - raise MalformedRecord(f"Expected {plan.target.__name__} record") - result = dict(raw) - keys = renames(plan.target) - for entry in entries: - old = keys.get(entry.name) + def retired_fields(cls, record: type) -> dict[str, str]: + return {item.name: item.old for item in RENAMED if issubclass(record, item.owner)} + + @classmethod + def selects_record(cls, record: type) -> bool: + return bool(cls.retired_fields(record)) + + @classmethod + def rename_record(cls, record: type, raw: dict[str, object]) -> dict[str, object]: + result = raw + for entry in FieldCodec.field_plans(record): + old = cls.retired_fields(record).get(entry.name) if old is not None and old in result: if entry.key in result: raise RelationViolationError( - f"{plan.target.__name__} contains both {old} and {entry.key}") + f"{record.__name__} contains both {old} and {entry.key}") result = {entry.key if key == old else key: value for key, value in result.items()} - if isinstance(entry, InlineFieldPlan): - result = self.rewrite(entry.value, result) - elif entry.key in result: - result[entry.key] = self.rewrite(entry.value, result[entry.key]) return result - @handles(UnionDecoder) - def union(self, plan: UnionDecoder, raw: object) -> object: - if raw is None: - return plan.alternatives - answers = [] - for alternative in plan.alternatives: - try: - changed = self.rewrite(alternative, raw) - alternative.decode_bound(changed, {}) - except MalformedRecord: - continue - if changed not in answers: - answers.append(changed) - if len(answers) != 1: - raise RelationViolationError( - f"Claim field carry cannot select exactly one declared {plan.target}: " - f"{len(answers)} different results") - return answers[0] - - @handles(_ArrayDecoder) - def array(self, plan: _ArrayDecoder, raw: object) -> object: - if raw is None: - return plan.items - return [self.rewrite(child, item) for child, item in plan.plan_items(raw)] - - @handles(_DictDecoder) - def mapping(self, plan: _DictDecoder, raw: object) -> object: - return (plan.key, plan.value) - - @handles(_StringKeyDictDecoder) - def string_mapping(self, plan: _StringKeyDictDecoder, raw: object) -> object: - if raw is None: - return self.mapping(plan, raw) - return {key: self.rewrite(plan.value, item) for key, item in plan.entries(raw)} - - @handles(_PairKeyDictDecoder) - def pair_mapping(self, plan: _PairKeyDictDecoder, raw: object) -> object: - if raw is None: - return self.mapping(plan, raw) - return [[self.rewrite(plan.key, key), self.rewrite(plan.value, item)] - for key, item in plan.entries(raw)] - - -WIRE_NAMES = ClaimFieldNames() - - -class ClaimFields(Carry): - precedes = (FinishedTurns, DeadTurns, OwnerRequests, TurnOwner, ContextRequestKey) + @classmethod + def convert(cls, declared: object, raw: object) -> object: + return compile_decoder(declared).rewrite_records(raw, cls.rename_record, cls.selects_record) @staticmethod def documents(root: Path, files: dict[Path, type[RootFile]]) -> dict[Path, object]: @@ -252,7 +136,6 @@ def carry(self, *, apply: bool) -> Carried: HeldBusFile(RegistryFile, RegistryFile.path_in(self.root)).require() rewritten = [] changed = 0 - visitor = WIRE_NAMES with _store_lock(HistoryArchive.path_in(self.root)): sources = CarryCatalogSource.read(self.root, self.original_root) for root in self.roots(): @@ -261,7 +144,7 @@ def carry(self, *, apply: bool) -> Carried: with registry.locked(): registry.private_guard_unlocked() raw = json.loads(registry.path.read_bytes()) - current = visitor.rewrite(compile_decoder(RegistryDocument), raw) + current = self.convert(RegistryDocument, raw) if current != raw: changed += 1 if apply: @@ -272,18 +155,18 @@ def carry(self, *, apply: bool) -> Carried: for path, owner in files.items(): self.require_confined(self.root, path) if issubclass(owner, SQLiteFile): - count = self.database(path, visitor, apply=apply) + count = self.database(path, apply=apply) changed += count if count and apply: rewritten.append(str(path.relative_to(self.root))) for path, declared in self.documents(root, files).items(): self.require_confined(self.root, path) plan = compile_decoder(declared) - if not visitor.touches(plan): + if not plan.contains_records(self.selects_record): continue with _store_lock(path): raw = json.loads(path.read_bytes()) - current = visitor.rewrite(plan, raw) + current = plan.rewrite_document(raw, self.rename_record, self.selects_record) count = int(current != raw) if count and apply: AtomicFile(path).write_text(json.dumps(current)) @@ -299,8 +182,8 @@ def carry(self, *, apply: bool) -> Carried: rewritten.append(HistoryArchive.filename) return Carried(changed, rewritten=tuple(rewritten)) - @staticmethod - def database(path: Path, visitor: ClaimFieldNames, *, apply: bool) -> int: + @classmethod + def database(cls, path: Path, *, apply: bool) -> int: # Read only typed JSON columns that can contain these declarations. The # multi-gigabyte message bodies are neither copied into RAM nor decoded. pending = [] @@ -314,13 +197,13 @@ def database(path: Path, visitor: ClaimFieldNames, *, apply: bool) -> int: renamed = {field.name: _identifier(old) for field in row._fields() if field.column.record_key - and (old := renames(row).get(field.name)) in stored} + and (old := cls.retired_fields(row).get(field.name)) in stored} if any(field.wire_name in stored for field in row._fields() if field.name in renamed): raise RelationViolationError(f"{row.__name__} contains old and new claim columns") json_fields = tuple(field for field in row._fields() if issubclass(field.storage, JsonStorage) and field.wire_name in stored - and visitor.touches(field.value)) + and field.value.contains_records(cls.selects_record)) primary = tuple(field for field in row._fields() if field.column.primary_key) if json_fields: selected = (*primary, *json_fields) @@ -339,7 +222,7 @@ def database(path: Path, visitor: ClaimFieldNames, *, apply: bool) -> int: if value is None: continue raw = json.loads(value) - current = visitor.rewrite(field.value, raw) + current = field.value.rewrite_records(raw, cls.rename_record, cls.selects_record) if current != raw: updates[field.wire_name] = json.dumps(current) if updates: From ed944f943161262b668e24f65f9cdef1ba556288 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:45:39 -0400 Subject: [PATCH 7/7] Carry either failed-turn column through the same rebuild --- tools/cutover/carry_claim_fields.py | 3 --- tools/cutover/carry_declared_failures.py | 5 +++-- tools/cutover/retired_claim_fields.py | 11 +++++++---- 3 files changed, 10 insertions(+), 9 deletions(-) diff --git a/tools/cutover/carry_claim_fields.py b/tools/cutover/carry_claim_fields.py index 73d89b2144..ed802ed42b 100644 --- a/tools/cutover/carry_claim_fields.py +++ b/tools/cutover/carry_claim_fields.py @@ -127,9 +127,6 @@ def database(cls, path: Path, *, apply: bool) -> int: for row in tables: stored = row.stored_columns(db) renamed = cls.retired_columns(row, stored) - if any(field.wire_name in stored for field in row._fields() - if field.name in renamed): - raise RelationViolationError(f"{row.__name__} contains old and new claim columns") json_fields = tuple(field for field in row._fields() if issubclass(field.storage, JsonStorage) and field.wire_name in stored and field.value.contains_records(cls.selects_record)) diff --git a/tools/cutover/carry_declared_failures.py b/tools/cutover/carry_declared_failures.py index 45d2c60c77..9081ad69c3 100644 --- a/tools/cutover/carry_declared_failures.py +++ b/tools/cutover/carry_declared_failures.py @@ -87,6 +87,7 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: with closing(SQLiteFile.read_snapshot(path)) as original: original.backup(prepared) columns = FailedTurnEvidence.stored_columns(prepared) + replacements = self.retired_columns(FailedTurnEvidence, columns) if "reason" in columns: for (reason,) in prepared.execute( f'SELECT DISTINCT reason FROM "{FailedTurnEvidence.declared_name}"' @@ -97,8 +98,8 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: f"WHEN {sql_literal(old)} THEN {sql_literal(kind)}" for old, kind in self.old_names.items() ) - replacements = {"failure": f"CASE reason {cases} END"} - replacements.update(self.retired_columns(FailedTurnEvidence, columns)) + replacements["failure"] = f"CASE reason {cases} END" + if replacements: FailedTurnEvidence.rebuild(prepared, replacements) changed += 1 if AttemptRecord.stored_columns(prepared): diff --git a/tools/cutover/retired_claim_fields.py b/tools/cutover/retired_claim_fields.py index 79062ee9e3..e4baee179e 100644 --- a/tools/cutover/retired_claim_fields.py +++ b/tools/cutover/retired_claim_fields.py @@ -68,10 +68,13 @@ def selects_record(cls, record: type) -> bool: @classmethod def retired_columns(cls, record: type[TypedTable], stored: tuple[str, ...]) -> dict[str, str]: - return {field.name: _identifier(old) - for field in record._fields() - if field.column.record_key - and (old := cls.retired_fields(record).get(field.name)) in stored} + fields = record._fields() + changed = {field.name: _identifier(old) + for field in fields if field.column.record_key + and (old := cls.retired_fields(record).get(field.name)) in stored} + if any(field.wire_name in stored for field in fields if field.name in changed): + raise RelationViolationError(f"{record.__name__} contains old and new claim columns") + return changed @classmethod def rename_record(cls, record: type, raw: dict[str, object]) -> dict[str, object]: