Type the remaining thread lookups and delete their constructor re-checks

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

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."""

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)

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.
 

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:

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)

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, ...],

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

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")
 

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,)