From f38eaa4cd9f819a44987939c240a90fbbf9af90f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 30 Oct 2026 16:47:52 -0400 Subject: [PATCH] Type the remaining thread lookups and delete their constructor re-checks Closes plan phase id128_nominal/root_ids (RootIdsPhase). On main only one of its grep hits was left: pi_rule_gate.py:102. Thirteen more lookup annotations sit under names the phase pattern cannot see (owner_lookup, sender_lookups); they carry the same kind, so they change with it. Retyped to ThreadLookup by one codemod (14 annotations in 7 modules). None of them changes what goes on the wire: every one is a parameter, or AttemptStart (never encoded), or the NativeInputRecord parent annotation, which all three stored subclasses already override with ThreadLookup. Deleted by hand, each already a typed value at its boundary: - CurrentExecutions.__post_init__ owner_lookup checks. Field typed at coordination_tables/executions.py:231; written at participant_store.py:76 from `lookup: ThreadLookup`; stored rows decode through FieldCodec. - ExecutionRecord.__post_init__ owner_lookup loop. Field typed at coordination_tables/executions.py:346; written at execution_store.py:106 from the parameter typed here, fed by selected_participant.py:57. - AttemptRecord.__post_init__ owner_lookup entry. Field typed at coordination_tables/attempts.py:183; written at attempt_start.py:99 from ExecutionRecord.owner_lookup. - OwnerFence.__post_init__ owner_lookup entry. Field typed at owner_fence.py:33; sole constructor attempt_start.py:41, from AttemptStart.owner_lookup, whose sole constructor is selected_turn.py:125 passing participant.lookup (selected_participant.py:57). Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/attempt_start.py | 3 ++- src/agent_comms/bus.py | 4 ++-- src/agent_comms/bus_tables.py | 4 ++-- src/agent_comms/coordination_response.py | 2 +- src/agent_comms/coordination_tables/attempts.py | 8 ++------ src/agent_comms/coordination_tables/executions.py | 10 ---------- src/agent_comms/execution_store.py | 5 +++-- src/agent_comms/native_input_record.py | 11 ++++++----- src/agent_comms/owner_fence.py | 5 ++--- src/agent_comms/pi_rule_gate.py | 3 ++- 10 files changed, 22 insertions(+), 33 deletions(-) diff --git a/src/agent_comms/attempt_start.py b/src/agent_comms/attempt_start.py index 964f2fe8c2..58bf1f8514 100644 --- a/src/agent_comms/attempt_start.py +++ b/src/agent_comms/attempt_start.py @@ -12,6 +12,7 @@ ) from .coordination_snapshot import RecoverySnapshot from .coordination_tables.attempts import AttemptRecord +from .field_codec import ThreadLookup from .owner_fence import OwnerFence, _digest from .participant_store import ParticipantSnapshot @@ -27,7 +28,7 @@ class AttemptStart: execution_id: str attempt_ordinal: int - owner_lookup: str + owner_lookup: ThreadLookup owner_generation: int token: str expected_execution_revision: int diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index d905a17a2c..dbfa254f39 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -397,7 +397,7 @@ def sent( def received( self, lookup: ThreadLookup, *, after: int = 0, before: int | None = None, limit: int = 100, - sender_lookups: tuple[str, ...] = (), descending: bool = True, + sender_lookups: tuple[ThreadLookup, ...] = (), descending: bool = True, sequences: tuple[int, ...] = (), ) -> tuple[Delivered, ...]: with self.reading() as source, closing(source._source_sequences( @@ -408,7 +408,7 @@ def received( def direct_replies( self, lookup: ThreadLookup, *, after: int = 0, limit: int = 100, - sender_lookups: tuple[str, ...] = (), descending: bool = True, + sender_lookups: tuple[ThreadLookup, ...] = (), descending: bool = True, sequences: tuple[int, ...] = (), ) -> tuple[Delivered, ...]: """Read the bounded append-time policy index before loading deliveries.""" diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index 8186493cbf..acbb717b4a 100644 --- a/src/agent_comms/bus_tables.py +++ b/src/agent_comms/bus_tables.py @@ -62,7 +62,7 @@ class ReceivedSource(ParticipantSource): @classmethod def query( - cls, lookup: ThreadLookup, sender_lookups: tuple[str, ...] = (), + cls, lookup: ThreadLookup, sender_lookups: tuple[ThreadLookup, ...] = (), ) -> tuple[str, tuple[object, ...]]: query, parameters = super().query(lookup) if sender_lookups: @@ -101,7 +101,7 @@ class DirectReplySource(AwaitingAnswerSource, declared_name=None): @classmethod def query( - cls, lookup: ThreadLookup, sender_lookups: tuple[str, ...] = (), + cls, lookup: ThreadLookup, sender_lookups: tuple[ThreadLookup, ...] = (), ) -> tuple[str, tuple[object, ...]]: query, parameters = AwaitingAnswerSource.query(lookup, sender_lookups) return query + " AND starts_direct_turn=?", (*parameters, True) diff --git a/src/agent_comms/coordination_response.py b/src/agent_comms/coordination_response.py index f7dacba55c..cacfe5b0f2 100644 --- a/src/agent_comms/coordination_response.py +++ b/src/agent_comms/coordination_response.py @@ -106,7 +106,7 @@ def matches_attempt( @classmethod def resolve_retained( cls, store: Coordination, bus: MessageBus, boundary, *, - owner_lookup: str, execution_id: str | None = None, + owner_lookup: ThreadLookup, execution_id: str | None = None, ) -> tuple[RecoverySnapshot, ...]: """Finish only frozen responses whose exact keyed bus rows already exist. diff --git a/src/agent_comms/coordination_tables/attempts.py b/src/agent_comms/coordination_tables/attempts.py index 01be67d962..67f2de0ecc 100644 --- a/src/agent_comms/coordination_tables/attempts.py +++ b/src/agent_comms/coordination_tables/attempts.py @@ -216,12 +216,8 @@ def snapshot_process_dead(self): def __post_init__(self) -> None: validate_execution_id(self.execution_id) - for field, value in ( - ("owner_lookup", self.owner_lookup), - ("owner_token_digest", self.owner_token_digest), - ): - require_nonempty(value, field) - require_bounded(value, field, MAX_IDENTIFIER_CHARS) + require_nonempty(self.owner_token_digest, "owner_token_digest") + require_bounded(self.owner_token_digest, "owner_token_digest", MAX_IDENTIFIER_CHARS) if min(self.attempt_ordinal, self.owner_generation, self.revision) <= 0: raise ValueError("attempt identity/generation/revision must be positive") if self.created_at_ms < 0 or self.updated_at_ms < self.created_at_ms: diff --git a/src/agent_comms/coordination_tables/executions.py b/src/agent_comms/coordination_tables/executions.py index fd9998d44b..45bc1f6d2a 100644 --- a/src/agent_comms/coordination_tables/executions.py +++ b/src/agent_comms/coordination_tables/executions.py @@ -8,9 +8,6 @@ from typing import TYPE_CHECKING, Annotated, ClassVar from agent_comms.coordination_contracts import ( - MAX_IDENTIFIER_CHARS, - require_bounded, - require_nonempty, validate_execution_id, ) from agent_comms.coordination_errors import ( @@ -234,8 +231,6 @@ def require_idle(self) -> None: pointer_revision: Annotated[int, Column(check="pointer_revision >= 0")] def __post_init__(self) -> None: - require_nonempty(self.owner_lookup, "owner_lookup") - require_bounded(self.owner_lookup, "owner_lookup", MAX_IDENTIFIER_CHARS) if self.execution_id.pointed != self.attempt_ordinal.pointed: raise IntegrityViolationError("pointer execution and ordinal must pair") if self.pointer_revision < 0: @@ -364,11 +359,6 @@ def snapshot_current_attempt_ordinal(self): def __post_init__(self) -> None: validate_execution_id(self.execution_id) - for field, value in ( - ("owner_lookup", self.owner_lookup), - ): - require_nonempty(value, field) - require_bounded(value, field, MAX_IDENTIFIER_CHARS) if self.revision <= 0 or self.max_attempts <= 0: raise ValueError("revision and max_attempts must be positive") self.lifecycle.validate_budget(self.max_attempts) diff --git a/src/agent_comms/execution_store.py b/src/agent_comms/execution_store.py index c18916e05a..34d70c5ca7 100644 --- a/src/agent_comms/execution_store.py +++ b/src/agent_comms/execution_store.py @@ -23,6 +23,7 @@ PendingExecution, QueuedExecution, ) +from agent_comms.field_codec import ThreadLookup from agent_comms.obligation_states import ( FailedResponse, PendingResponse, @@ -38,7 +39,7 @@ def create( self, execution_id: str, origin: type[ExecutionProvenance], - owner_lookup: str, + owner_lookup: ThreadLookup, max_attempts: int, *, sources: tuple[SelectedSource, ...] = (), @@ -51,7 +52,7 @@ def create_after_triage( self, execution_id: str, origin: type[ExecutionProvenance], - owner_lookup: str, + owner_lookup: ThreadLookup, max_attempts: int, *, sources: tuple[SelectedSource, ...], diff --git a/src/agent_comms/native_input_record.py b/src/agent_comms/native_input_record.py index 4ee1181f07..5c7a83ecff 100644 --- a/src/agent_comms/native_input_record.py +++ b/src/agent_comms/native_input_record.py @@ -34,6 +34,7 @@ JsonShapeMember, MalformedRecord, NativeContext, + ThreadLookup, WireName, ) from .native_session_reopen import PiSessionId @@ -200,11 +201,11 @@ def transcript_events( def require_attempt(self) -> FullNativeExecution: raise IdentityConflict("Native triage input has no execution attempt") - def published_replies(self, db: sqlite3.Connection, owner_lookup: str) -> tuple[MessageReference, ...]: + def published_replies(self, db: sqlite3.Connection, owner_lookup: ThreadLookup) -> tuple[MessageReference, ...]: """TRIAGE has no execution publication; FULL owns that relation.""" return () - def matches_execution(self, execution: ExecutionRecord, owner_lookup: str) -> bool: + def matches_execution(self, execution: ExecutionRecord, owner_lookup: ThreadLookup) -> bool: return False def retried_in(self, db: sqlite3.Connection) -> bool: @@ -341,7 +342,7 @@ def binding_fields(self) -> dict[str, NativeColumn]: def require_attempt(self) -> FullNativeExecution: return self - def matches_execution(self, execution: ExecutionRecord, owner_lookup: str) -> bool: + def matches_execution(self, execution: ExecutionRecord, owner_lookup: ThreadLookup) -> bool: return ( execution.execution_id == self.execution_id and execution.owner_lookup == owner_lookup @@ -355,7 +356,7 @@ def retried_in(self, db: sqlite3.Connection) -> bool: """A transient provider failure leaves this input behind a later attempt.""" return self._execution(db).lifecycle.current_attempt_ordinal != self.attempt_ordinal - def published_replies(self, db: sqlite3.Connection, owner_lookup: str) -> tuple[MessageReference, ...]: + def published_replies(self, db: sqlite3.Connection, owner_lookup: ThreadLookup) -> tuple[MessageReference, ...]: execution = self._execution(db) if not self.matches_execution(execution, owner_lookup): return () @@ -449,7 +450,7 @@ def require_recorded_input(self, owner: CursorOwner, db: sqlite3.Connection) -> class NativeInputRecord(OwnerRecord, ABC): input_id: InputId execution: NativeInputExecution - owner_lookup: str + owner_lookup: ThreadLookup owner_generation: int @property diff --git a/src/agent_comms/owner_fence.py b/src/agent_comms/owner_fence.py index aa4f0a033b..10516cc80b 100644 --- a/src/agent_comms/owner_fence.py +++ b/src/agent_comms/owner_fence.py @@ -44,9 +44,8 @@ def authority(self) -> AttemptAuthority: def __post_init__(self) -> None: validate_execution_id(self.execution_id) - for field, value in (("owner_lookup", self.owner_lookup), ("token", self.token)): - require_nonempty(value, field) - require_bounded(value, field, MAX_IDENTIFIER_CHARS) + require_nonempty(self.token, "token") + require_bounded(self.token, "token", MAX_IDENTIFIER_CHARS) if min(self.attempt_ordinal, self.owner_generation, self.revision) <= 0: raise ValueError("fence ordinal, generation and revision must be positive") diff --git a/src/agent_comms/pi_rule_gate.py b/src/agent_comms/pi_rule_gate.py index d67a976c6e..ff05be6ae6 100644 --- a/src/agent_comms/pi_rule_gate.py +++ b/src/agent_comms/pi_rule_gate.py @@ -23,6 +23,7 @@ ) from .bus_publication import Delivered +from .field_codec import ThreadLookup from .rule_deal import DealTrigger, DealtRuleGate, Redeals from .tool_denial import SelectedToolDenied @@ -99,7 +100,7 @@ def owner_words(self, snapshot: RegistrySnapshot) -> tuple[str, ...]: for delivery in self.typed_by(user, lookup) )[:self.owner_prompts])) - def typed_by(self, user: Thread, lookup: str) -> Iterator[Delivered]: + def typed_by(self, user: Thread, lookup: ThreadLookup) -> Iterator[Delivered]: """The human's own messages to this thread, newest first, read a page at a time until the deal has its prompts or his history ends.""" sender = (user.incarnation.require_recorded().lookup,)