From 04e0194c68ab368b66efec9af4e628fc6e95e9a0 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:54:14 -0400 Subject: [PATCH] Fold backend relocations into their behavior owners --- src/agent_comms/historical_views.py | 2 +- src/agent_comms/pi_events.py | 25 +++- src/agent_comms/pi_turn_phase.py | 33 ----- src/agent_comms/plan_phases.py | 10 +- src/agent_comms/presentation.py | 2 +- src/agent_comms/private_registry_guard.py | 73 ++++++++++ src/agent_comms/registry_provenance.py | 2 +- src/agent_comms/thread_management.py | 25 +++- src/agent_comms/thread_provenance.py | 68 --------- src/agent_comms/thread_rename.py | 154 -------------------- src/agent_comms/threads.py | 82 ++++++++--- src/agent_comms/tools.py | 3 +- src/agent_comms/transcripts.py | 3 +- tests/guards/test_source_proof_ownership.py | 2 +- tests/test_historical_views.py | 2 +- tests/test_importing.py | 2 +- tests/test_phase_evidence.py | 6 +- 17 files changed, 198 insertions(+), 296 deletions(-) delete mode 100644 src/agent_comms/pi_turn_phase.py delete mode 100644 src/agent_comms/thread_provenance.py delete mode 100644 src/agent_comms/thread_rename.py diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py index 8c175b58cd..aa9ba46a46 100644 --- a/src/agent_comms/historical_views.py +++ b/src/agent_comms/historical_views.py @@ -49,7 +49,7 @@ from .revision_views import BusSourceView from .store_files import _atomic_write_text, _store_lock from .thread_identity import UnrecordedThreadIncarnation -from .thread_provenance import ThreadProvenance +from .threads import ThreadProvenance if TYPE_CHECKING: from .catalog_document import CatalogDocument diff --git a/src/agent_comms/pi_events.py b/src/agent_comms/pi_events.py index fbe3953b28..c1f08e013a 100644 --- a/src/agent_comms/pi_events.py +++ b/src/agent_comms/pi_events.py @@ -39,7 +39,6 @@ UnknownPayload, UnreportedUsage, ) -from .pi_turn_phase import PiTurnPhase from .pi_vocabulary import CompactionReason, UnknownCompactionReason from .request_progress import RequestProgress from .tool_declaration import ToolDeclaration @@ -57,9 +56,31 @@ def retry_progress(self) -> bool: @dataclass(frozen=True) -class PiEvent(PiTurnPhase, OpenPayload, affix=""): +class PiEvent(OpenPayload, affix=""): """One declaration-owned record received by the Pi RPC channel.""" + _start_phases: ClassVar[dict[type[object], tuple[type[phases.TurnPhase], ...]]] = {} + _end_events: ClassVar[dict[type[object], frozenset[type[object]]]] = {} + + def on(self, current: phases.TurnPhase) -> phases.TurnPhase: + for event_type in type(self).__mro__: + for member in self._start_phases.get(event_type, ()): + return self.enter(member, current) + return self.leave(current) if self.matches_end(current) else current + + def enter(self, member: type[phases.TurnPhase], current: phases.TurnPhase) -> phases.TurnPhase: + return member() + + def matches_end(self, current: phases.TurnPhase) -> bool: + return any( + event_type in self._end_events.get(member, frozenset()) + for member in type(current).__mro__ + for event_type in type(self).__mro__ + ) + + def leave(self, current: phases.TurnPhase) -> phases.TurnPhase: + return phases.ModelWaitPhase() + def __init_subclass__( cls, *, starts: type[phases.TurnPhase] | None = None, ends: type[phases.TurnPhase] | None = None, **kwargs, diff --git a/src/agent_comms/pi_turn_phase.py b/src/agent_comms/pi_turn_phase.py deleted file mode 100644 index 91984b4cf7..0000000000 --- a/src/agent_comms/pi_turn_phase.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Pi event declarations own their transitions into shared turn phases.""" - -from __future__ import annotations - -from typing import ClassVar - -from . import turn_phase as phases - - -class PiTurnPhase: - """Pi declarations own event entry/end matching; shared phases own stage behavior.""" - - _start_phases: ClassVar[dict[type[object], tuple[type[phases.TurnPhase], ...]]] = {} - _end_events: ClassVar[dict[type[object], frozenset[type[object]]]] = {} - - def on(self, current: phases.TurnPhase) -> phases.TurnPhase: - for event_type in type(self).__mro__: - for member in self._start_phases.get(event_type, ()): - return self.enter(member, current) - return self.leave(current) if self.matches_end(current) else current - - def enter(self, member: type[phases.TurnPhase], current: phases.TurnPhase) -> phases.TurnPhase: - return member() - - def matches_end(self, current: phases.TurnPhase) -> bool: - return any( - event_type in self._end_events.get(member, frozenset()) - for member in type(current).__mro__ - for event_type in type(self).__mro__ - ) - - def leave(self, current: phases.TurnPhase) -> phases.TurnPhase: - return phases.ModelWaitPhase() diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 24fa581c95..980e33401e 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -13,9 +13,9 @@ from agent_comms.declared_family import DeclaredFamily from agent_comms.installer.carry import Carry -from agent_comms.pi_turn_phase import PiTurnPhase +from agent_comms.pi_events import PiEvent from agent_comms.thread_identity import BirthlessIncarnation -from agent_comms.thread_rename import ThreadRename +from agent_comms.private_registry_guard import RegistryPrivacy from agent_comms.tool_calls import CliToolArguments from agent_comms.phase_evidence import ( AllEvidence, @@ -667,11 +667,11 @@ class RostersPhase(RemainingSlopPhase): ClassFlagsAbsent(r"^\s*" + attribute + r"\s*[:=]", owners=(owner,)) for owner, attribute in ( (Carry, "declared"), - (ThreadRename, "kinds"), + (RegistryPrivacy, "kinds"), (BirthlessIncarnation, "members"), (CliToolArguments, "types"), - (PiTurnPhase, "_start_phases"), - (PiTurnPhase, "_end_events"), + (PiEvent, "_start_phases"), + (PiEvent, "_end_events"), ) ), ) diff --git a/src/agent_comms/presentation.py b/src/agent_comms/presentation.py index 259db557c1..dc41e9ba31 100644 --- a/src/agent_comms/presentation.py +++ b/src/agent_comms/presentation.py @@ -75,7 +75,7 @@ from .runtime_info import RuntimeInfo from .storage_room import RoomLedger from .thread_identity import ThreadIncarnation -from .thread_provenance import ThreadProvenance +from .threads import ThreadProvenance from .thread_presentation import ( ThreadOwnerBinding, ThreadPresentation, diff --git a/src/agent_comms/private_registry_guard.py b/src/agent_comms/private_registry_guard.py index 4e8025d103..58941c1130 100644 --- a/src/agent_comms/private_registry_guard.py +++ b/src/agent_comms/private_registry_guard.py @@ -29,7 +29,9 @@ from typing import TYPE_CHECKING, Annotated, NoReturn from . import store_files +from .coordination_contracts import MAX_IDENTIFIER_CHARS from .coordination_database import CoordinationStore +from .coordination_errors import ParticipantNotRegistered from .coordinator import Coordination from .errors import RelationViolationError from .field_codec import FieldCodec, Id128Text, MalformedRecord, WireName @@ -44,6 +46,7 @@ from .notification_assignment import NotificationAssignmentRead from .private_nk_entrypoint import RootLaunch from .registry_document import RegistrySnapshot + from .thread_management import ThreadManagement from .threads import Thread from .typed_table import TableSchema @@ -118,6 +121,15 @@ def _decode(data: bytes, root_id: bytes) -> tuple[int, int, bytes]: class RegistryPrivacy(ABC): """What a registry root's protocol marker decides, selected by the registry store.""" + @abstractmethod + def rename( + self, threads: ThreadManagement, before: Thread, new_name: str + ) -> tuple[str, str]: + """Rename under this root's original registry and coordinator authority.""" + + def rename_applied(self, root: Path) -> None: + """Public roots have no private owner intent to complete.""" + @staticmethod def capture(root: Path) -> RegistryPrivacy: """Read original privacy authority under its registry's existing lock.""" @@ -174,6 +186,11 @@ def coordinator_schema(self) -> type[TableSchema]: class PublicRegistry(RegistryPrivacy): + def rename( + self, threads: ThreadManagement, before: Thread, new_name: str + ) -> tuple[str, str]: + return threads.registry.rename(before.name, new_name) + def coordinator_schema(self) -> type[TableSchema]: from .coordination_schema import CoordinatorTable @@ -231,6 +248,62 @@ def require_root(self, root_id: str, error: Exception) -> None: class PrivateRegistryGuard(DurableFile, WireWakeFile, RegistryPrivacy): + def rename( + self, threads: ThreadManagement, before: Thread, new_name: str + ) -> tuple[str, str]: + root, registry = threads.root, threads.registry + _require_no_private_owner_rename(root) + coordinator = CoordinationStore.path_in(root) + if not coordinator.exists() or before.name == new_name: + return registry.rename(before.name, new_name) + # Validate registry/SQL name refusals before renaming the private owner. + # Registry readers cannot select a different owner while both stores change. + registry.status(before.name).require_self_rename() + if registry.snapshot().taken(new_name, owner=before.name): + raise RelationViolationError(f"Thread name {new_name!r} is already in use.") + before.require_name(new_name) + if len(new_name) > MAX_IDENTIFIER_CHARS: + raise ValueError("Private coordinator owner name exceeds its bound.") + with Coordination(str(coordinator)) as store: + try: + person = store.participants.get(before.incarnation.require_recorded().lookup) + except ParticipantNotRegistered: + return registry.rename(before.name, new_name) + if not person.committed or person.owner_identity.owner_thread != before.name: + raise RelationViolationError( + "Private coordinator owner differs from the registered owner." + ) + generation = person.owner_identity.generation + PrivateOwnerRenameIntent( + lookup=person.lookup, + old=before.name, + new=new_name, + generation=generation, + wire_root_id=self.root_id.hex(), + ).write(root) + store.participants.rename_owner(person.lookup, new_name, expected_generation=generation) + try: + return registry.rename(before.name, new_name) + except BaseException as error: + # Registry persistence can fail after the SQL CAS. If its atomic snapshot still + # names the old owner, restore that name without revoking its active attempt. + uncertain = RelationViolationError( + "Private owner rename is uncertain; inspect both authorities." + ) + try: + if registry.require(before.name).name == before.name: + store.participants.rename_owner( + person.lookup, before.name, expected_generation=generation + ) + except BaseException as compensation: + # Ambiguous dual-store failure: the inspection needs both failures. + raise uncertain from compensation + raise uncertain from error + + def rename_applied(self, root: Path) -> None: + if PrivateOwnerRenameIntent.path_in(root).exists(): + PrivateOwnerRenameIntent.complete(root) + filename = ".registry-owner-guard" def coordinator_schema(self) -> type[TableSchema]: diff --git a/src/agent_comms/registry_provenance.py b/src/agent_comms/registry_provenance.py index 86a6373275..c0c6333630 100644 --- a/src/agent_comms/registry_provenance.py +++ b/src/agent_comms/registry_provenance.py @@ -15,7 +15,7 @@ from .errors import RelationViolationError, UnregisteredThreadError from .routing import DeliveryScope from .thread_identity import ThreadIncarnation -from .thread_provenance import ThreadProvenance +from .threads import ThreadProvenance if TYPE_CHECKING: from .audience_manifest import FrozenRecipient diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py index d1d18b589d..27f6b32125 100644 --- a/src/agent_comms/thread_management.py +++ b/src/agent_comms/thread_management.py @@ -49,10 +49,17 @@ from .session_file import SavedSessionFile from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel from .thread_owned_state import ThreadOwnedState -from .thread_rename import RenameThreadResult, ThreadRename +from .tool_result import ToolResult from .threads import KeptModel, ModelChoice, NextTurnConfiguration, Thread, current_thread +@dataclass(frozen=True, slots=True) +class RenameThreadResult(ToolResult): + previous: str + current: str + changed: bool + + class TagChoice(SubclassUnion, affix="Tags"): """Select explicit, inherited or changed tags from an existing set.""" @@ -380,7 +387,21 @@ def _rename_thread( def _rename_thread_unlocked( self, name: str, new_name: str, *, title: ThreadText = NoThreadText() ) -> RenameThreadResult: - return ThreadRename.chosen(self, self.registry.require(name), new_name).applied(title) + privacy = self._privacy() + previous, current = privacy.rename(self, self.registry.require(name), new_name) + thread = self.registry.require(current) + titled = replace(thread, auto_title_pending=False, title=title.applied(thread.title)) + if titled != thread: + self.registry.register(titled, self.registry.status(current)) + changed = previous != current + if changed: + self.agents.activity.rename_thread(previous, current) + self.agents.runtime_info.rename_thread(previous, current) + self.ledger.rename_thread(previous, current) + with self.catalog.editing() as document: + document.rename_thread(previous, current) + privacy.rename_applied(self.root) + return RenameThreadResult(previous, current, changed) def heartbeat(self, name: str) -> None: self.registry.heartbeat(name) diff --git a/src/agent_comms/thread_provenance.py b/src/agent_comms/thread_provenance.py deleted file mode 100644 index a2c21d6f4e..0000000000 --- a/src/agent_comms/thread_provenance.py +++ /dev/null @@ -1,68 +0,0 @@ -"""A recorded thread declaration, independent of live execution authority.""" - -from __future__ import annotations - -from dataclasses import dataclass, field, fields -from functools import cached_property -from typing import Annotated - -from .agent_backend import AgentBackend, NewThreadBackend, RuntimeLocation -from .channel_targets import Tag -from .field_codec import WireOmitDefault, WireRequired -from .thread_identity import ThreadIncarnation, ThreadIncarnationRecord -from .thread_text import NoThreadText, ThreadText -from .session_file import NoSessionFile, SessionFile - - -@dataclass(frozen=True, kw_only=True) -class ThreadProvenance: - """Recorded thread identity, without permission to own a process or start a turn.""" - - name: str - tags: frozenset[str] - worktree: str - created_at: Annotated[float, WireRequired()] - session_file: SessionFile = field(default_factory=NoSessionFile) - title: ThreadText = field(default_factory=NoThreadText) - backend: Annotated[type[AgentBackend], WireOmitDefault()] = field( - default_factory=NewThreadBackend.selected - ) - - @cached_property - def runtime(self) -> RuntimeLocation: - """The codec decoded both stored fields; their record owns this binding.""" - return self.session_file.location(self.backend) - - @property - def display_title(self) -> str: - return self.title.otherwise(self.name) - - @staticmethod - def require_name(name: str) -> None: - allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_") - if not name or not set(name) <= allowed: - raise ValueError(f"Thread name {name!r} must be alphanumeric with hyphens/underscores.") - - def __post_init__(self) -> None: - self.incarnation - self.require_name(self.name) - if not self.worktree: - raise ValueError("Thread worktree cannot be empty.") - for tag in self.tags: - Tag.validate(tag) - - @staticmethod - def capture(thread: ThreadProvenance) -> ThreadProvenance: - """Freeze only determining provenance from the original typed declaration.""" - return ThreadProvenance( - **{ - item.name: getattr(thread, item.name) - for item in fields(ThreadProvenance) - if item.init - } - ) - - @cached_property - def incarnation(self) -> ThreadIncarnation: - """The declaration's birth member, created once and shared by its readers.""" - return ThreadIncarnationRecord(self.name, self.created_at).incarnation() diff --git a/src/agent_comms/thread_rename.py b/src/agent_comms/thread_rename.py deleted file mode 100644 index b89cbadd96..0000000000 --- a/src/agent_comms/thread_rename.py +++ /dev/null @@ -1,154 +0,0 @@ -"""The rename operation and its privacy-specific ownership transaction.""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from dataclasses import dataclass, replace -from typing import TYPE_CHECKING, ClassVar - -from .coordination_contracts import MAX_IDENTIFIER_CHARS -from .coordination_database import CoordinationStore -from .coordination_errors import ParticipantNotRegistered -from .coordinator import Coordination -from .errors import RelationViolationError -from .private_registry_guard import ( - PrivateOwnerRenameIntent, - PrivateRegistryGuard, - PublicRegistry, - RegistryPrivacy, - _require_no_private_owner_rename, -) -from .thread_provenance import ThreadProvenance -from .thread_text import ThreadText -from .tool_result import ToolResult - -if TYPE_CHECKING: - from .thread_management import ThreadManagement - from .threads import Thread - - -@dataclass(frozen=True, slots=True) -class RenameThreadResult(ToolResult): - previous: str - current: str - changed: bool - - -@dataclass(frozen=True, slots=True) -class ThreadRename(ABC): - """One rename, its kind derived from the registry's privacy.""" - - privacy: ClassVar[type[RegistryPrivacy]] - kinds: ClassVar[dict[type[RegistryPrivacy], type[ThreadRename]]] = {} - threads: ThreadManagement - before: Thread - new_name: str - registry_privacy: RegistryPrivacy - - def __init_subclass__(cls, **kwargs): - # slots=True rebuilds the class; its final declaration replaces the first. - super(ThreadRename, cls).__init_subclass__(**kwargs) - cls.kinds[cls.privacy] = cls - - @classmethod - def chosen(cls, threads: ThreadManagement, before: Thread, new_name: str) -> ThreadRename: - privacy = threads._privacy() - return cls.kinds[type(privacy)](threads, before, new_name, privacy) - - @abstractmethod - def renamed(self) -> tuple[str, str]: - """Move the registry name, after whatever this kind's owner stores need first.""" - - def registry_renamed(self) -> tuple[str, str]: - return self.threads.registry.rename(self.before.name, self.new_name) - - def applied(self, title: ThreadText) -> RenameThreadResult: - previous, current = self.renamed() - threads = self.threads - thread = threads.registry.require(current) - titled = replace(thread, auto_title_pending=False, title=title.applied(thread.title)) - if titled != thread: - threads.registry.register(titled, threads.registry.status(current)) - if previous == current: - return RenameThreadResult(previous, current, False) - threads.agents.activity.rename_thread(previous, current) - threads.agents.runtime_info.rename_thread(previous, current) - threads.ledger.rename_thread(previous, current) - with threads.catalog.editing() as document: - document.rename_thread(previous, current) - return RenameThreadResult(previous, current, True) - - -class PlainRename(ThreadRename): - privacy = PublicRegistry - - def renamed(self) -> tuple[str, str]: - return self.registry_renamed() - - -@dataclass(frozen=True, slots=True) -class PrivateOwnerRename(ThreadRename): - """Moves a committed coordinator owner under a durable intent, so a crash between the two - stores fails closed instead of turning an old attempt into a new send.""" - - privacy = PrivateRegistryGuard - registry_privacy: PrivateRegistryGuard - - def renamed(self) -> tuple[str, str]: - root, registry = self.threads.root, self.threads.registry - before, new_name = self.before, self.new_name - _require_no_private_owner_rename(root) - coordinator = CoordinationStore.path_in(root) - if not coordinator.exists() or before.name == new_name: - return self.registry_renamed() - # Validate registry/SQL name refusals before renaming the private owner. - # Registry readers cannot select a different owner while both stores change. - registry.status(before.name).require_self_rename() - if registry.snapshot().taken(new_name, owner=before.name): - raise RelationViolationError(f"Thread name {new_name!r} is already in use.") - ThreadProvenance.require_name(new_name) - if len(new_name) > MAX_IDENTIFIER_CHARS: - raise ValueError("Private coordinator owner name exceeds its bound.") - with Coordination(str(coordinator)) as store: - try: - person = store.participants.get(before.incarnation.require_recorded().lookup) - except ParticipantNotRegistered: - return self.registry_renamed() - if not person.committed or person.owner_identity.owner_thread != before.name: - raise RelationViolationError( - "Private coordinator owner differs from the registered owner." - ) - generation = person.owner_identity.generation - PrivateOwnerRenameIntent( - lookup=person.lookup, - old=before.name, - new=new_name, - generation=generation, - wire_root_id=self.registry_privacy.root_id.hex(), - ).write(root) - store.participants.rename_owner(person.lookup, new_name, expected_generation=generation) - try: - return self.registry_renamed() - except BaseException as error: - # Registry persistence can fail after the SQL CAS. If its atomic snapshot still - # names the old owner, restore that name without revoking its active attempt. - uncertain = RelationViolationError( - "Private owner rename is uncertain; inspect both authorities." - ) - try: - if registry.require(before.name).name == before.name: - store.participants.rename_owner( - person.lookup, before.name, expected_generation=generation - ) - except BaseException as compensation: - # Ambiguous dual-store failure: the inspection needs both failures. - raise uncertain from compensation - raise uncertain from error - - def applied(self, title: ThreadText) -> RenameThreadResult: - result = ThreadRename.applied(self, title) - # The registry lock and renamed() bind a present intent to this rename; - # it completes only after both authorities and the ancillary stores agree. - if PrivateOwnerRenameIntent.path_in(self.threads.root).exists(): - PrivateOwnerRenameIntent.complete(self.threads.root) - return result diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py index e01e1f3499..672a679c43 100644 --- a/src/agent_comms/threads.py +++ b/src/agent_comms/threads.py @@ -6,12 +6,14 @@ import os import time from abc import ABC, abstractmethod -from dataclasses import dataclass, field, replace +from dataclasses import dataclass, field, fields, replace +from functools import cached_property from pathlib import Path from typing import TYPE_CHECKING, ClassVar, Annotated -from .agent_backend import AgentBackend, NewThreadBackend +from .agent_backend import AgentBackend, NewThreadBackend, RuntimeLocation +from .channel_targets import Tag from .child_process import NoProcess, ProcessIdentity, ProcessState from .cli_result import ThreadToolCatalogResult from .context_segments.wake_values import WakeWorkContext @@ -47,9 +49,10 @@ from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel from .thread_text import NoParent, NoThreadText, ThreadParent, ThreadText from .session_file import NoSessionFile, SessionFile -from .thread_provenance import ThreadProvenance from .thread_identity import ( AgentThreadRole, + ThreadIncarnation, + ThreadIncarnationRecord, ThreadRole, TurnFence, TurnId, @@ -139,23 +142,72 @@ def configured(self, thread: Thread) -> Thread: return self.service_tier.configured(self.model.configured(thought)) +@dataclass(frozen=True, kw_only=True) +class ThreadProvenance: + """Recorded thread identity, without permission to own a process or start a turn.""" + + name: str = field(kw_only=False) + tags: Annotated[frozenset[str], RegistrationPolicy(InheritEmpty)] = field(kw_only=False) + worktree: str = field(kw_only=False) + created_at: Annotated[float, WireRequired()] + session_file: Annotated[SessionFile, RegistrationPolicy(InheritApplied)] = field( + default_factory=NoSessionFile + ) + title: Annotated[ThreadText, RegistrationPolicy(InheritApplied)] = field( + default_factory=NoThreadText + ) + backend: Annotated[ + type[AgentBackend], WireOmitDefault(), RegistrationPolicy(InheritPrevious) + ] = field(default_factory=NewThreadBackend.selected) + + @cached_property + def runtime(self) -> RuntimeLocation: + """The codec decoded both stored fields; their record owns this binding.""" + return self.session_file.location(self.backend) + + @property + def display_title(self) -> str: + return self.title.otherwise(self.name) + + @staticmethod + def require_name(name: str) -> None: + allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_") + if not name or not set(name) <= allowed: + raise ValueError(f"Thread name {name!r} must be alphanumeric with hyphens/underscores.") + + def __post_init__(self) -> None: + self.require_name(self.incarnation.name) + if not self.worktree: + raise ValueError("Thread worktree cannot be empty.") + for tag in self.tags: + Tag.validate(tag) + + @staticmethod + def capture(thread: ThreadProvenance) -> ThreadProvenance: + """Freeze only determining provenance from the original typed declaration.""" + return ThreadProvenance( + **{ + item.name: getattr(thread, item.name) + for item in fields(ThreadProvenance) + if item.init + } + ) + + @cached_property + def incarnation(self) -> ThreadIncarnation: + """The declaration's birth member, created once and shared by its readers.""" + return ThreadIncarnationRecord(self.name, self.created_at).incarnation() + + @dataclass(frozen=True, slots=True) class Thread(DurableFile, PatternFile, ThreadProvenance): filename: ClassVar[str] = "native-sessions/{lookup}/{name}" catalog_filename: ClassVar[str] = "comms-catalog.json" """Declares one agent thread's identity and provenance.""" - name: str = field(kw_only=False) - tags: Annotated[frozenset[str], RegistrationPolicy(InheritEmpty)] = field( - kw_only=False - ) - worktree: str = field(kw_only=False) parent: ThreadParent = field(default_factory=NoParent) task: ThreadText = field(default_factory=NoThreadText) process_identity: ProcessState = field(default_factory=NoProcess) - session_file: Annotated[SessionFile, RegistrationPolicy(InheritApplied)] = ( - field(default_factory=NoSessionFile, kw_only=True) - ) model: Annotated[ThreadModel, RegistrationPolicy(InheritApplied)] = field( default_factory=UndeclaredModel ) @@ -190,20 +242,12 @@ class Thread(DurableFile, PatternFile, ThreadProvenance): project_resume_due: Annotated[ bool, RegistrationPolicy(InheritPrevious), WireOmitDefault() ] = False - title: Annotated[ThreadText, RegistrationPolicy(InheritApplied)] = field( - default_factory=NoThreadText, kw_only=True - ) role: Annotated[type[ThreadRole], RegistrationPolicy(InheritPrevious)] = ( AgentThreadRole ) execution: Annotated[type[ThreadExecution], WireOmitDefault()] = field( default=NativeThreadExecution, kw_only=True ) - backend: Annotated[ - type[AgentBackend], - WireOmitDefault(), - RegistrationPolicy(InheritPrevious), - ] = field(default_factory=NewThreadBackend.selected, kw_only=True) turn_record: TurnRecord = field(default_factory=NoTurnRecord) channel_scope_generation: int = 0 turn_generation: int = 0 diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 7f055133eb..85ea877ab9 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -156,8 +156,7 @@ ThreadCommandAvailability, ) from .thread_identity import ThreadIncarnation -from .thread_management import InheritedTags, TagChoice -from .thread_rename import RenameThreadResult +from .thread_management import InheritedTags, RenameThreadResult, TagChoice from .thread_model import ThreadModel from .threads import NextTurnConfiguration from .tool_caller import NoThreadCaller, ThreadCaller, ThreadListing, ToolCaller diff --git a/src/agent_comms/transcripts.py b/src/agent_comms/transcripts.py index 37757e260e..5ca82a8a05 100644 --- a/src/agent_comms/transcripts.py +++ b/src/agent_comms/transcripts.py @@ -33,8 +33,7 @@ from .recorded_bus import RecordedBusIdentity from .registration import Registration from .routing import TurnRouting -from .thread_provenance import ThreadProvenance -from .threads import Thread +from .threads import Thread, ThreadProvenance from .transcript_events import ForkOriginTranscript, ForkTaskTranscript, TranscriptEvent from .transcript_routes import TranscriptRoutes from .transcript_sources import ( diff --git a/tests/guards/test_source_proof_ownership.py b/tests/guards/test_source_proof_ownership.py index 4b899cbdf0..281e18c8e2 100644 --- a/tests/guards/test_source_proof_ownership.py +++ b/tests/guards/test_source_proof_ownership.py @@ -79,7 +79,7 @@ def test_archived_identity_cannot_reacquire_a_live_registry_or_thread(): from dataclasses import MISSING, fields from typing import get_type_hints from agent_comms.historical_views import HistorySource, HistoricalThread - from agent_comms.thread_provenance import ThreadProvenance + from agent_comms.threads import ThreadProvenance provenance = next(item for item in fields(HistorySource) if item.name == "provenance") assert provenance.default is MISSING and provenance.default_factory is MISSING diff --git a/tests/test_historical_views.py b/tests/test_historical_views.py index b7156d66ed..b074cc2498 100644 --- a/tests/test_historical_views.py +++ b/tests/test_historical_views.py @@ -358,7 +358,7 @@ def test_historical_namespace_has_recorded_provenance_and_no_live_authority(migr from agent_comms.field_codec import FieldCodec from agent_comms.registry_provenance import RegistryProvenance - from agent_comms.thread_provenance import ThreadProvenance + from agent_comms.threads import ThreadProvenance old, _, live, source = migrated recorded = source.provenance.require("alice") diff --git a/tests/test_importing.py b/tests/test_importing.py index bd8a135dcd..90f6fe301a 100644 --- a/tests/test_importing.py +++ b/tests/test_importing.py @@ -26,7 +26,7 @@ from agent_comms.harness_install import ClaudeCodeHarness, CodexHarness, OpenCodeHarness from agent_comms.importing import ImportBuffer, ThreadImported, ImportSnapshot, ImportedSessionMetadata from agent_comms.native_entries import NativeEntry, ImportedMetadataEntry -from agent_comms.thread_provenance import ThreadProvenance +from agent_comms.threads import ThreadProvenance from agent_comms.thread_status import StoppedThreadStatus from agent_comms.runtime_updates import FinishedItem from agent_comms.transcript_events import ImportedReply, ImportedRequest, TextTranscript diff --git a/tests/test_phase_evidence.py b/tests/test_phase_evidence.py index 3b498c2c5c..9c8ac0194b 100644 --- a/tests/test_phase_evidence.py +++ b/tests/test_phase_evidence.py @@ -113,11 +113,11 @@ def test_roster_evidence_keeps_external_abi_and_each_remaining_internal_registry "class Backend:\n serves_threads = True\n", "class Backend:\n serves_new_threads: ClassVar[bool] = True\n", "class Carry:\n declared: ClassVar[dict] = {}\n", - "class ThreadRename:\n kinds = {}\n", + "class RegistryPrivacy:\n kinds = {}\n", "class BirthlessIncarnation:\n members = {}\n", "class CliToolArguments:\n types = {}\n", - "class PiTurnPhase:\n _start_phases = {}\n", - "class PiTurnPhase:\n _end_events = {}\n", + "class PiEvent:\n _start_phases = {}\n", + "class PiEvent:\n _end_events = {}\n", ): (tmp_path / "src/remaining.py").write_text(remaining) git(tmp_path, "add", ".")