diff --git a/src/agent_comms/acp_extension.py b/src/agent_comms/acp_extension.py index 34fba48c9a..0e2df1139f 100644 --- a/src/agent_comms/acp_extension.py +++ b/src/agent_comms/acp_extension.py @@ -201,7 +201,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: ProcessIdentity @property @@ -210,11 +210,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) @@ -302,15 +302,15 @@ class NoCursorScope(CursorScope): class LiveCursorScope(CursorScope, LiveAttachmentScope): session_id: str wire_root_id: str - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity owner: ProcessIdentity 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.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: ClaimedIdentity owner: ProcessIdentity @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/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/cohort_schema.py b/src/agent_comms/cohort_schema.py index 73609f473c..0ee3cc5105 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/context_store.py b/src/agent_comms/context_store.py index 4462a72683..b3a95f4f5f 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 Annotated, ClassVar, Literal, Self, cast +from typing import Annotated, ClassVar, Self, cast from .errors import RelationViolationError, StoredHistoryUnreadable from .field_codec import ( @@ -35,7 +35,7 @@ UnrecordedSegment, ) -from .typed_table import Column, DurableStore, Generated, Index, TypedTable +from .typed_table import Column, DurableStore, Generated, Index, SingletonRow, TypedTable @dataclass(frozen=True) @@ -91,8 +91,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/coordination_response.py b/src/agent_comms/coordination_response.py index dbc9d73be4..a881f4df49 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/coordination_snapshot.py b/src/agent_comms/coordination_snapshot.py index f8ac5189f7..1ca66516c7 100644 --- a/src/agent_comms/coordination_snapshot.py +++ b/src/agent_comms/coordination_snapshot.py @@ -355,7 +355,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..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] - claim_scope: Annotated[ClaimScope, WireName("admission")] + resource_scope: ClaimScope @dataclass(frozen=True) @@ -351,7 +349,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 +370,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 +388,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: RecordedScope @dataclass(frozen=True) @@ -400,7 +398,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 +445,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 +455,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/field_codec.py b/src/agent_comms/field_codec.py index 7cc42ffe9f..a4db3ba982 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,45 @@ 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 + + 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 +1987,19 @@ class _RepresentedDecoder(ValueDecoder[Any]): representation: type[FieldRepresentation] structural: ValueDecoder[object] + @property + 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 +2149,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 +2233,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 +2349,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 +2412,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 +2430,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 +2445,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 +2497,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 +2510,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/src/agent_comms/following_inputs.py b/src/agent_comms/following_inputs.py index 503d49f244..172ddd18dd 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 03f26f02d2..b396197f11 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 .turn_failure import TurnFailure -from .field_codec import Id128Text, WireName +from .field_codec import Id128Text from .goal_attempt_identity import GoalAttemptIdentity from .thread_identity import ( ClaimedIdentity, @@ -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, 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") @@ -89,11 +89,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..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_scope: Annotated[int, WireName("admission")] + claim_generation: int 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..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 - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity 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 c9d66cf089..049ef38b1d 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/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 cc915b13f4..e836e7c4a9 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, @@ -53,7 +53,7 @@ StableReferenceColumns, TextStorage, TypedRow, - TypedTable, + WithoutRowid, WriteOnceColumns, sql_literal, ) @@ -87,7 +87,7 @@ class NativeRuntimeInput( NativeInputRecord, NativeInputContext, NativeRuntimeTable, - TypedTable, + WithoutRowid, ): @classmethod def notification_context_sql(cls) -> str: @@ -265,7 +265,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", @@ -289,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 " @@ -343,7 +341,7 @@ class CurrentNativeCursor( NoDelete, NativeInputContext, NativeRuntimeTable, - TypedTable, + WithoutRowid, ): column_orders = (NonDecreasing(("covered_seq", "injected_seq")),) reference_column = 'injected_seq' @@ -367,7 +365,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"), ] @@ -396,7 +393,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/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/owner_lifecycle.py b/src/agent_comms/owner_lifecycle.py index 26c1a19546..2cf6080375 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/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/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/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..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: - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity 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/registry_document.py b/src/agent_comms/registry_document.py index 04a61d3f2c..97c7fd61fd 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/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/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 e8fc10b09b..ef369de003 100644 --- a/src/agent_comms/selected_tool_broker.py +++ b/src/agent_comms/selected_tool_broker.py @@ -72,7 +72,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 2c4fd31aeb..e09c0ef2f3 100644 --- a/src/agent_comms/selected_write_plan.py +++ b/src/agent_comms/selected_write_plan.py @@ -15,9 +15,8 @@ from contextlib import suppress from dataclasses import dataclass, fields from pathlib import Path -from typing import TYPE_CHECKING, Annotated, Literal +from typing import TYPE_CHECKING, Literal -from agent_comms.child_process import Platform from .coordination_database import CoordinationStore from .locked_store import DurableFile, PatternFile from .store_files import AtomicFile @@ -34,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 @@ -55,7 +54,7 @@ class SelectedWriteBinding: root_id: str source: MessageReference assignment_id: str - claim_scope: Annotated[ClaimedIdentity, WireName("admission")] + claimed_identity: ClaimedIdentity @dataclass(frozen=True, kw_only=True) @@ -70,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/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/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/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 9b4ab4aa87..68d8848331 100644 --- a/src/agent_comms/triage_native_sources.py +++ b/src/agent_comms/triage_native_sources.py @@ -14,7 +14,7 @@ Column, GuardedInserts, ImmutableRows, - TypedTable, + WithoutRowid, ) @@ -23,7 +23,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 @@ -39,7 +39,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/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/turn_lease.py b/src/agent_comms/turn_lease.py index 234f2f6b40..4f54e6b0a1 100644 --- a/src/agent_comms/turn_lease.py +++ b/src/agent_comms/turn_lease.py @@ -18,7 +18,6 @@ RecordMember, SingleFieldValue, SubclassUnion, - WireName, WireTransient, boundary_decoder, ) @@ -213,9 +212,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/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py index 04c4777692..cf697d228c 100644 --- a/src/agent_comms/typed_table.py +++ b/src/agent_comms/typed_table.py @@ -2196,3 +2196,16 @@ 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 + + +@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)] 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]: diff --git a/tests/test_acp_queue_contract.py b/tests/test_acp_queue_contract.py index 4bebd429f6..fae8258976 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, ProcessIdentity(123, 1)), LiveCursorScope("official-acp-id", "a" * 32, original, ProcessIdentity(123, 1))) 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=ProcessIdentity(124, 1))).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 17589592ee..955b0e8485 100644 --- a/tests/test_bus_resource_claims.py +++ b/tests/test_bus_resource_claims.py @@ -63,7 +63,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 e4aa7f57c6..7cd757819e 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..2a864b76f3 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,20 +65,20 @@ 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") - 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_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_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_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 1e96db0b6b..3034c32b9c 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())): 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 == "claimed_identity" + 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..07c550ed40 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" @@ -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 e22e44881d..0f3b03bb73 100644 --- a/tools/cutover/carry_bus_history.py +++ b/tools/cutover/carry_bus_history.py @@ -67,6 +67,8 @@ ) +from .retired_claim_fields import RetiredClaimFields + @dataclass(frozen=True) class HistoryMetadata(BusFile): """Only the retired stream boundary is needed, never its root ID or activation.""" @@ -403,7 +405,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 +491,15 @@ 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)) + raw = RetiredClaimFields.convert(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 +525,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..ed802ed42b --- /dev/null +++ b/tools/cutover/carry_claim_fields.py @@ -0,0 +1,182 @@ +"""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 pathlib import Path + +from agent_comms.errors import RelationViolationError +from agent_comms.field_codec import FieldCodec, compile_decoder +from agent_comms.historical_views import HistoryArchive +from agent_comms.installer.carry import Carried, Carry +from agent_comms.locked_store import JsonDocument, RegistryFile, RootFile, SQLiteFile +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 ( + SelectedWriteIntent, + SelectedWritePlans, +) +from agent_comms.store_files import AtomicFile, _store_lock +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_dead_turns import DeadTurns +from .carry_finished_turns import FinishedTurns +from .carry_owner_requests import OwnerRequests +from .carry_turn_owner import TurnOwner +from .retired_claim_fields import RetiredClaimFields + + +class ClaimFields(RetiredClaimFields, Carry): + precedes = (FinishedTurns, DeadTurns, 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( + SelectedWritePlans.directory_in(root))) + 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: list[Path] = [] + changed = 0 + 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 = self.convert(RegistryDocument, raw) + if current != raw: + changed += 1 + if apply: + registry._write_unlocked(json.dumps(current, indent=2)) + rewritten.extend(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, apply=apply) + changed += count + if count and apply: + rewritten.append(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 plan.contains_records(self.selects_record): + continue + with _store_lock(path): + raw = json.loads(path.read_bytes()) + 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)) + changed += count + if count and apply: + rewritten.append(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: + AtomicFile(HistoryArchive.path_in(self.root)).write_text( + json.dumps(FieldCodec.encode(rebound))) + rewritten.append(Path(HistoryArchive.filename)) + return Carried(changed, rewritten=tuple(rewritten)) + + @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 = [] + 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 = cls.retired_columns(row, stored) + 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)) + 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 = field.value.rewrite_records(raw, cls.rename_record, cls.selects_record) + 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 diff --git a/tools/cutover/carry_declared_failures.py b/tools/cutover/carry_declared_failures.py index 6a4e3495a3..9081ad69c3 100644 --- a/tools/cutover/carry_declared_failures.py +++ b/tools/cutover/carry_declared_failures.py @@ -22,9 +22,15 @@ from agent_comms.typed_table import HeldBusFile, sql_literal from .carry_bus_history import CarryCatalogSource +from .carry_claim_fields import ClaimFields +from .retired_claim_fields import RetiredClaimFields -class DeclaredFailures(Carry): +class DeclaredFailures(RetiredClaimFields, 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, @@ -73,7 +79,7 @@ def carry(self, *, apply: bool) -> Carried: def carry_root(self, root: Path, *, apply: bool) -> Carried: changed = 0 - rewritten: list[str] = [] + rewritten: list[Path] = [] store = GoalAttemptStore.directory_in(root) path = GoalAttemptStore.path_in(store) if path.exists(): @@ -81,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}"' @@ -91,7 +98,9 @@ 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" + if replacements: + FailedTurnEvidence.rebuild(prepared, replacements) changed += 1 if AttemptRecord.stored_columns(prepared): for old, kind in self.old_names.items(): @@ -108,7 +117,7 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: with closing(SQLiteFile.writer(path)) as destination: prepared.backup(destination) GoalAttemptStore.publish_change(path) - rewritten.extend(str(file.relative_to(self.root)) for file in ( + rewritten.extend(file.relative_to(self.root) for file in ( path, *GoalAttemptStore.source_paths(store), )) for diagnostic in DiagnosticDocument.matching_in(root): @@ -121,6 +130,6 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: document["failure"] = self.old_names[old].declared_name if apply: AtomicFile(diagnostic).write_text(json.dumps(document, indent=2) + "\n") - rewritten.append(str(diagnostic.relative_to(self.root))) + rewritten.append(diagnostic.relative_to(self.root)) changed += 1 return Carried(changed, rewritten=tuple(rewritten)) diff --git a/tools/cutover/retired_claim_fields.py b/tools/cutover/retired_claim_fields.py new file mode 100644 index 0000000000..e4baee179e --- /dev/null +++ b/tools/cutover/retired_claim_fields.py @@ -0,0 +1,94 @@ +"""Retired field-name conversion shared by the carry and physical JSONL read-back. + +This capability owns only the predecessor names. FieldCodec owns wire traversal. +Delete with the claim-field carry after all retained roots cross the format change. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from functools import cache + +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, compile_decoder +from agent_comms.goal_failure_observation import FailedTurnEvidence +from agent_comms.input_attempt import StoredInput +from agent_comms.input_origin import HumanInputOrigin +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.selected_write_plan import SelectedWriteBinding, SelectedWriteIntent +from agent_comms.thread_identity import ClaimedIdentity, TurnFence +from agent_comms.turn_lease import ActiveTurn +from agent_comms.typed_table import TypedTable, _identifier + + +@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"), +) + + +class RetiredClaimFields: + @classmethod + @cache + 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 retired_columns(cls, record: type[TypedTable], stored: tuple[str, ...]) -> dict[str, str]: + 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]: + 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"{record.__name__} contains both {old} and {entry.key}") + result = {entry.key if key == old else key: value + for key, value in result.items()} + return result + + @classmethod + def convert(cls, declared: object, raw: object) -> object: + return compile_decoder(declared).rewrite_records(raw, cls.rename_record, cls.selects_record)