From 2da58560437dacaa71bd5700d9b540a0faa37f12 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 03:59:15 -0400 Subject: [PATCH 1/4] Checkpoint kind families and migrate their current consumers --- src/agent_comms/agent_backend.py | 35 ++++++++++--- src/agent_comms/channel_management.py | 65 +++++++++++++++++------- src/agent_comms/channels.py | 30 +++++------ src/agent_comms/delivery_policy.py | 4 +- src/agent_comms/messages.py | 39 +++++++++----- src/agent_comms/pi_commands.py | 3 +- src/agent_comms/plan_phases.py | 2 +- src/agent_comms/private_sidecar.py | 13 +++-- src/agent_comms/publisher.py | 6 +-- src/agent_comms/response_conversation.py | 6 +-- src/agent_comms/tools.py | 21 +++----- src/agent_comms/turn_input_source.py | 4 +- src/agent_comms/turn_inputs.py | 7 +-- src/agent_comms/wake.py | 37 +++++++------- tests/bus_record_fixture.py | 6 +-- tests/native_backend_fixture.py | 4 +- tests/test_agent_backend.py | 2 +- tests/test_backend.py | 4 +- tests/test_bus.py | 4 +- tests/test_channel_display.py | 8 ++- tests/test_channel_retirement.py | 19 ++----- tests/test_channels.py | 51 ++++++------------- tests/test_exporting.py | 8 +-- tests/test_field_codec.py | 15 ++---- tests/test_message_categories.py | 10 ++-- tests/test_s4_ownership.py | 9 +--- tests/test_tracked_native_retry.py | 6 +-- tests/test_turn_inbox.py | 4 +- tests/test_wake.py | 61 ++++++++++------------ tests/verify_connector_tools.py | 6 +-- 30 files changed, 239 insertions(+), 250 deletions(-) diff --git a/src/agent_comms/agent_backend.py b/src/agent_comms/agent_backend.py index 743f8a25de..23e5bf9407 100644 --- a/src/agent_comms/agent_backend.py +++ b/src/agent_comms/agent_backend.py @@ -21,7 +21,6 @@ from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import dataclass -from enum import Enum from types import MappingProxyType from typing import TYPE_CHECKING, Any, ClassVar, Protocol, Self @@ -119,12 +118,34 @@ def __post_init__(self) -> None: raise ValueError("An input ID must be a nonempty string") -class WhenBusy(Enum): +class BusyInputPolicy(DeclaredFamily, affix="InputPolicy"): """What the backend does with an input that arrives while a run is going.""" - FOLLOW_UP = "follow_up" # run it after the current run ends - STEER = "steer" # place it at the next boundary of the current run - REJECT = "reject" # refuse it + @classmethod + @abstractmethod + def streaming(cls) -> type[StreamingInputPolicy]: ... + + +class StreamingInputPolicy(BusyInputPolicy, declared_name=None): + """The busy policies Pi's streamingBehavior protocol accepts.""" + + @classmethod + def streaming(cls) -> type[StreamingInputPolicy]: + return cls + + +class FollowUpInputPolicy(StreamingInputPolicy, declared_name="followUp"): + """Run after the current run ends; preserve Pi's exact followUp spelling.""" + + +class SteerInputPolicy(StreamingInputPolicy): + """Place the input at the next boundary of the current run.""" + + +class RejectInputPolicy(BusyInputPolicy): + @classmethod + def streaming(cls) -> type[StreamingInputPolicy]: + raise ValueError("Pi cannot reject a prompt through streamingBehavior.") @dataclass(frozen=True, kw_only=True) @@ -198,7 +219,7 @@ class InputRequest(TurnInboxItem): input_id: InputId content: InputContent - when_busy: WhenBusy + when_busy: type[BusyInputPolicy] @property def carried_inputs(self) -> tuple[InputId, ...]: @@ -328,7 +349,7 @@ async def close_idle(self) -> None: class Steers(RunsTurns): - """Accepts :class:`InputRequest` with ``WhenBusy.STEER`` while a run is going.""" + """Accepts :class:`InputRequest` with ``SteerInputPolicy`` while a run is going.""" class Interrupts(Steers): diff --git a/src/agent_comms/channel_management.py b/src/agent_comms/channel_management.py index 3d61be57de..5c30819624 100644 --- a/src/agent_comms/channel_management.py +++ b/src/agent_comms/channel_management.py @@ -5,7 +5,6 @@ from abc import abstractmethod from collections.abc import Sequence from dataclasses import dataclass, replace -from enum import Enum from pathlib import Path from typing import TYPE_CHECKING, ClassVar @@ -108,25 +107,53 @@ def deletion_cohort(self, channels, tag, cohort): for member in channels.registry.status(thread.name).deletable_threads(thread)) -class TagAction(Enum): - LIST = "list" - CREATE = "create" - RENAME = "rename" - DELETE = "delete" - - def apply( - self, channels: ChannelManagement, name: str = "", new_name: str = "", *, - disposition: TagDisposition = KeepThreadsTagDisposition(), confirmed: bool = False - ) -> frozenset[str]: - match self: - case self.CREATE: - channels.create_tag(name) - case self.RENAME: - channels.rename_tag(name, new_name) - case self.DELETE: - channels.delete_tag(name, disposition=disposition, confirmed=confirmed) +@dataclass(frozen=True, kw_only=True) +class TagEdit(DeclaredFamily, affix="TagEdit"): + """A tag operation carries only the arguments that operation uses.""" + + def apply(self, channels: ChannelManagement) -> frozenset[str]: + self.change(channels) return channels.catalog.read().tags + @abstractmethod + def change(self, channels: ChannelManagement) -> None: ... + + +class ListTagEdit(TagEdit): + label = "List tags" + def change(self, channels: ChannelManagement) -> None: + pass + + +@dataclass(frozen=True, kw_only=True) +class NamedTagEdit(TagEdit, declared_name=None): + name: str + + +class CreateTagEdit(NamedTagEdit): + label = "Create tag" + def change(self, channels: ChannelManagement) -> None: + channels.create_tag(self.name) + + +@dataclass(frozen=True, kw_only=True) +class RenameTagEdit(NamedTagEdit): + label: ClassVar[str] = "Rename tag" + new_name: str + + def change(self, channels: ChannelManagement) -> None: + channels.rename_tag(self.name, self.new_name) + + +@dataclass(frozen=True, kw_only=True) +class DeleteTagEdit(NamedTagEdit): + label: ClassVar[str] = "Delete tag" + disposition: TagDisposition = KeepThreadsTagDisposition() + confirmed: bool = False + + def change(self, channels: ChannelManagement) -> None: + channels.delete_tag(self.name, disposition=self.disposition, confirmed=self.confirmed) + class ChannelManagement: def __init__(self, root: Path, registry: Registration, bus: MessageBus, threads: ThreadManagement): @@ -181,7 +208,7 @@ def _update_tags(self, name: str, choice: TagChoice) -> Thread: Message( announcer.name, channel.name, - f"{announcer.name} {change.value} {channel.name}", + f"{announcer.name} {change.declared_name} {channel.name}", InfoKind, role=MembershipConversationRole(change), ), diff --git a/src/agent_comms/channels.py b/src/agent_comms/channels.py index b3a6f82cdd..935a9776e2 100644 --- a/src/agent_comms/channels.py +++ b/src/agent_comms/channels.py @@ -7,7 +7,6 @@ from collections import defaultdict from collections.abc import Iterable, Iterator from dataclasses import dataclass, field -from enum import StrEnum from typing import TYPE_CHECKING, ClassVar, Annotated @@ -18,7 +17,7 @@ from .errors import RelationViolationError from .field_codec import FieldCodec, CatalogExclude from .history_result import ChannelHistoryPageResult, EverythingHistoryPageResult, HistoryPageResult -from .messages import MembershipChange +from .messages import MembershipTransition from .read_basis import AllChannelDisplayScope, ChannelDisplayScope, MemberActivityDisplayScope from .tool_result import ToolResult @@ -128,7 +127,7 @@ def delete_view_bindings(self, command) -> tuple: return () @abstractmethod - def membership_changes(self, before: frozenset[str], after: frozenset[str]) -> tuple[MembershipChange, ...]: ... + def membership_changes(self, before: frozenset[str], after: frozenset[str]) -> tuple[type[MembershipTransition], ...]: ... def pending_count(self, counts: dict[str, int]) -> int: return sum(self.history_scope().target_values(counts)) @@ -190,11 +189,8 @@ class MembershipChannel: def read_only(self) -> bool: return False - def membership_changes(self, before: frozenset[str], after: frozenset[str]) -> tuple[MembershipChange, ...]: - previous, current = self.matches(before), self.matches(after) - if previous == current: - return () - return (MembershipChange.JOINED if current else MembershipChange.LEFT,) + def membership_changes(self, before: frozenset[str], after: frozenset[str]) -> tuple[type[MembershipTransition], ...]: + return MembershipTransition.between(self.matches(before), self.matches(after)) class BuiltinChannelView(PredicateChannel, MembershipChannel, Channel, declared_name=None): @@ -286,7 +282,7 @@ class ReadOnlyChannel(NonRoutableChannel): def read_only(self) -> bool: return True - def membership_changes(self, before: frozenset[str], after: frozenset[str]) -> tuple[MembershipChange, ...]: + def membership_changes(self, before: frozenset[str], after: frozenset[str]) -> tuple[type[MembershipTransition], ...]: return () @@ -360,11 +356,6 @@ def matches(self, tags: frozenset[str]) -> bool: return False -class ViewKind(StrEnum): - PARTICIPANTS = "participants" - ACTIVITY = "activity" - - class ViewMatch(DeclaredFamily, affix="Match"): """A tag relation, selected once at the external boundary.""" @@ -404,11 +395,10 @@ def matches(self, tags: frozenset[str]) -> bool: @dataclass(frozen=True, slots=True) -class SavedView(ToolResult): +class SavedView(DeclaredFamily, ToolResult, affix="View"): """A named non-routable projection over authoritative channels or activity.""" name: Annotated[str, CatalogExclude()] - kind: ViewKind predicate: ViewPredicate created_at: float = field(default_factory=time.time) original_targets: frozenset[str] = frozenset() @@ -422,3 +412,11 @@ def history_targets(self) -> frozenset[str]: def __post_init__(self) -> None: Tag.validate(self.name) + + +class ParticipantsView(SavedView): + """A saved participant projection over its declared tag predicate.""" + + +class ActivityView(SavedView): + """A saved activity projection over its declared tag predicate.""" diff --git a/src/agent_comms/delivery_policy.py b/src/agent_comms/delivery_policy.py index dc6ea373d8..f06afcda67 100644 --- a/src/agent_comms/delivery_policy.py +++ b/src/agent_comms/delivery_policy.py @@ -12,7 +12,7 @@ from agent_comms.response_acknowledgement import ( KeyedResponseAcknowledgement, NoResponseAcknowledgement, ResponseAcknowledgement, ) -from agent_comms.wake import ControlClassification, resolve_wake_cohort +from agent_comms.wake import MessageControl, resolve_wake_cohort from agent_comms.wake_policy import MessageAudience, WakeDecision @@ -22,7 +22,7 @@ class DeliveryManifest: manifest_codec: str resolver_version: str policy_version: str - control: ControlClassification + control: type[MessageControl] audience: FrozenAudience # Exact existing canonical decision projection. Its digest is already bound # by immutable cohort evidences; changing this spelling would change proofs. diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py index b7722c205d..bcf45c3c1f 100644 --- a/src/agent_comms/messages.py +++ b/src/agent_comms/messages.py @@ -10,7 +10,6 @@ from abc import abstractmethod from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass, field, fields, replace -from enum import StrEnum from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar, Self, Type, cast, dataclass_transform @@ -162,12 +161,30 @@ def require_publishable(cls, message: Message, worktree: Path) -> None: message.report.require_references_exist(worktree) -class MembershipChange(StrEnum): - JOINED = "joined" - LEFT = "left" +class MembershipTransition(DeclaredFamily, affix="Membership"): + @classmethod + def conversation_role(cls): + return MembershipConversationRole(cls) - def conversation_role(self): - return MembershipConversationRole(self) + @classmethod + def between(cls, previous: bool, current: bool) -> tuple[type[MembershipTransition], ...]: + return tuple(member for member in cls.members_with(cls) if member.applies(previous, current)) + + @staticmethod + @abstractmethod + def applies(previous: bool, current: bool) -> bool: ... + + +class JoinedMembership(MembershipTransition): + @staticmethod + def applies(previous: bool, current: bool) -> bool: + return not previous and current + + +class LeftMembership(MembershipTransition): + @staticmethod + def applies(previous: bool, current: bool) -> bool: + return previous and not current @dataclass(frozen=True, slots=True) @@ -181,7 +198,7 @@ class MessageRoleWire: """The unchanged external fields, decoded once into a message role.""" change: Annotated[ - MembershipChange | NoMembershipChange, + type[MembershipTransition] | NoMembershipChange, WireName("membership"), WireOmitDefault(), ] = field(default_factory=NoMembershipChange) @@ -275,9 +292,7 @@ def notification_references(self, message): return (message.reference,) if message.type.conversation_message else () def wake_mode(self, target, recipient, mentions, control): - if control.passive: - return PassiveWake() - return target.wake_policy(recipient, mentions) + return control.wake_mode(target, recipient, mentions) @dataclass(frozen=True, slots=True) @@ -303,13 +318,13 @@ def require_recipients(self, target, audience): @dataclass(frozen=True) class Membership(InformationalMessage): - change: MembershipChange + change: type[MembershipTransition] def wire_fields(self): return replace(super().wire_fields(), change=self.change) def export_attributes(self): - return (*super().export_attributes(), f"membership={self.change.value}") + return (*super().export_attributes(), f"membership={self.change.declared_name}") def as_notice(self): return MembershipNoticeRole(self.change) diff --git a/src/agent_comms/pi_commands.py b/src/agent_comms/pi_commands.py index 867b3f2d0b..edce90680a 100644 --- a/src/agent_comms/pi_commands.py +++ b/src/agent_comms/pi_commands.py @@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Literal from uuid import uuid4 +from .agent_backend import StreamingInputPolicy from . import agent_events as events from . import turn_phase as phases from .declared_family import DeclaredFamily, UnknownFamilyMember @@ -183,7 +184,7 @@ class Prompt(PiCommand): ] = () owner_tools: Annotated[OwnerToolConnection, WireName("ownerTools")] streaming_behavior: Annotated[ - Literal[None, "steer", "followUp"], + type[StreamingInputPolicy] | None, WireOmitDefault(), WireName("streamingBehavior"), ] = None diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 24fa581c95..594d66bdbd 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -651,7 +651,7 @@ class KindFamiliesPhase(RemainingSlopPhase): """6.3: kinds are families whose members carry their behaviour and spelling.""" evidence = PatternAbsent( - r"class (TagAction|ExecutionOrigin|WhenBusy|ControlClassification|MembershipChange" + r"class (TagAction|ExecutionOrigin|WhenBusy|MessageControl|MembershipTransition" r"|ViewKind)\b" ) diff --git a/src/agent_comms/private_sidecar.py b/src/agent_comms/private_sidecar.py index 5c7c195f60..0d24f362aa 100644 --- a/src/agent_comms/private_sidecar.py +++ b/src/agent_comms/private_sidecar.py @@ -25,11 +25,16 @@ from dataclasses import dataclass from functools import partial from pathlib import Path -from typing import Annotated, Literal +from typing import Annotated, Literal, TYPE_CHECKING from agent_comms.locked_store import SQLiteFile from agent_comms.coordination_errors import IdentityConflict +from .field_codec import FieldCodec + +if TYPE_CHECKING: + from .agent_backend import StreamingInputPolicy + from .private_path import ObservedPath from .store_files import StoreOwner from .typed_table import ( @@ -365,14 +370,14 @@ def sidecar_connection( connection.close() -def encode_request(text: str, *, streaming_behavior: str | None = None) -> str: +def encode_request(text: str, *, streaming_behavior: type[StreamingInputPolicy] | None = None) -> str: """JSON-encode like the pinned native ``_claimNativeInput`` request.""" return json.dumps( { "kind": "prompt", "text": text, "images": None, - "streamingBehavior": streaming_behavior, + "streamingBehavior": FieldCodec.encode(streaming_behavior), "expandPromptTemplates": True, "source": "rpc", }, @@ -381,7 +386,7 @@ def encode_request(text: str, *, streaming_behavior: str | None = None) -> str: ) -def native_request_digest(text: str, *, streaming_behavior: str | None = None) -> str: +def native_request_digest(text: str, *, streaming_behavior: type[StreamingInputPolicy] | None = None) -> str: """sha256 over the pinned native request envelope, not the bare text.""" return hashlib.sha256( ("pi-input-request-v1\n" + encode_request(text, streaming_behavior=streaming_behavior)).encode("utf-8") diff --git a/src/agent_comms/publisher.py b/src/agent_comms/publisher.py index 2f767e8c47..3fa5b22cc6 100644 --- a/src/agent_comms/publisher.py +++ b/src/agent_comms/publisher.py @@ -43,7 +43,7 @@ from .mentions import ThreadMention from .private_path import FileRevision from .registry_document import DEFAULT_ROSTER, RegistrySnapshot -from .wake import ControlClassification, resolve_wake_cohort +from .wake import MessageControl, resolve_wake_cohort, OrdinaryControl if TYPE_CHECKING: from .coordination_session import NativeBoundary @@ -241,11 +241,11 @@ def build(seq: int, db: sqlite3.Connection) -> PrivateRecord: revision, sender_lookup=sender_lookup, sender_name=sender, ) decisions = resolve_wake_cohort( - stored, frozen_audience=audience, control=ControlClassification.ORDINARY, + stored, frozen_audience=audience, control=OrdinaryControl, ) return PrivateRecord.from_message( stored, audience=audience, decisions=decisions, - control=ControlClassification.ORDINARY.value, + control=OrdinaryControl.declared_name, ) return origin.append(self.bus, build) diff --git a/src/agent_comms/response_conversation.py b/src/agent_comms/response_conversation.py index d68bff572d..9211725174 100644 --- a/src/agent_comms/response_conversation.py +++ b/src/agent_comms/response_conversation.py @@ -103,16 +103,16 @@ def audience(self, message: Message): def record(self, seq: int, message: Message, intent) -> KeyedPrivateRecord: from agent_comms.bus import KeyedPrivateRecord from agent_comms.delivery_policy import ResponseDeliveryPolicy - from agent_comms.wake import ControlClassification + from agent_comms.wake import MessageControl, OrdinaryControl stored = replace(message, seq=seq) audience = self.audience(stored) decisions = ResponseDeliveryPolicy.resolve( - stored, audience, ControlClassification.ORDINARY, + stored, audience, OrdinaryControl, ) return KeyedPrivateRecord.from_message( stored, audience=audience, decisions=decisions, - control=ControlClassification.ORDINARY.value, + control=OrdinaryControl.declared_name, acknowledgement=KeyedResponseAcknowledgement( self.privacy.namespace, intent.execution_id, intent.publication_key, public_envelope_digest(stored.to_wire()), diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 7f055133eb..48ad07477f 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -37,9 +37,9 @@ from . import cli as comms_cli from .change_report import NoChangeReport, ReportEvidence -from .channel_management import KeepThreadsTagDisposition, TagAction, TagDisposition +from .channel_management import KeepThreadsTagDisposition, ListTagEdit, TagDisposition, TagEdit from agent_comms.channel_targets import ChannelTarget, CommandTarget, NamedTarget -from .channels import AnyChannel, Channel, SavedView, ViewKind, ViewMatch, ViewPredicate +from .channels import AnyChannel, Channel, SavedView, ViewMatch, ViewPredicate from .cli_commands import ( ArchiveCliCommand, CliCommand, @@ -629,19 +629,12 @@ class CommsTagsTool(ToolRequest): "List, create, rename, or delete tags. Renaming/deleting updates thread " "assignments and channel filters. Uncovered tags automatically have a channel." ) - action: Annotated[TagAction, *tool_field("Operation")] = TagAction.LIST - name: Annotated[str, *tool_field("Tag name")] = "" - new_name: Annotated[str, *tool_field("Replacement tag for rename")] = "" - disposition: Annotated[TagDisposition, *tool_field("Tagged-thread operation")] = ( - dataclasses.field(default_factory=KeepThreadsTagDisposition) - ) - confirmed: Annotated[bool, *tool_field("Confirm thread archiving or deletion")] = ( - False + edit: Annotated[TagEdit, *tool_field("Tag operation and its arguments")] = ( + dataclasses.field(default_factory=ListTagEdit) ) def apply(self, comms: Comms, caller: ToolCaller) -> TagList: - tags = self.action.apply(comms.channels, self.name, self.new_name, - disposition=self.disposition, confirmed=self.confirmed) + tags = self.edit.apply(comms.channels) return TagList(tuple(sorted(tags))) @@ -703,13 +696,13 @@ class CommsSetViewTool(ToolRequest): " messages and never own channel history." ) name: Annotated[str, *tool_field("View name without #")] - kind: Annotated[ViewKind, *tool_field("Projection kind")] + kind: Annotated[type[SavedView], *tool_field("Projection kind")] match: Annotated[type[ViewMatch], *tool_field("Tag predicate")] tags: Annotated[str, *tool_field("One or more comma-separated tags")] def apply(self, comms: Comms, caller: ToolCaller) -> SavedView: return comms.channels.set_saved_view( - SavedView(self.name, self.kind, ViewPredicate(self.match, _tag_set(self.tags)))) + self.kind(self.name, ViewPredicate(self.match, _tag_set(self.tags)))) @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/turn_input_source.py b/src/agent_comms/turn_input_source.py index 6daeb31c86..f92e6b0574 100644 --- a/src/agent_comms/turn_input_source.py +++ b/src/agent_comms/turn_input_source.py @@ -7,7 +7,7 @@ from dataclasses import dataclass, replace from typing import TYPE_CHECKING, ClassVar -from .agent_backend import WhenBusy +from .agent_backend import BusyInputPolicy, SteerInputPolicy from .channel_input_batch import InputBatch, SingleInputBatch from .messages import Message from .started_input import StartedInput @@ -209,7 +209,7 @@ class ScheduledOriginalInput(DependencyOriginalInput): @dataclass(frozen=True, kw_only=True) class FollowingTurnInput(TurnInputSource): keys: tuple[str, ...] - when_busy: ClassVar[WhenBusy] = WhenBusy.STEER + when_busy: ClassVar[type[BusyInputPolicy]] = SteerInputPolicy def continue_response(self, response, parts): """An actual following request may revise the turn's automatic answer.""" diff --git a/src/agent_comms/turn_inputs.py b/src/agent_comms/turn_inputs.py index 31c32485c7..8b4f7076de 100644 --- a/src/agent_comms/turn_inputs.py +++ b/src/agent_comms/turn_inputs.py @@ -22,7 +22,6 @@ TurnInbox, TurnInboxItem, UseSettings, - WhenBusy, ) from .field_codec import Id128Text from .locked_store import RegistryFile @@ -33,10 +32,6 @@ from .pi_events import MessageStart, Response -# Pi's spelling of each busy policy it supports; Pi has no way to refuse an input. -_PI_STREAMING_BEHAVIOR = {WhenBusy.STEER: "steer", WhenBusy.FOLLOW_UP: "followUp"} - - @dataclass class InputForwarding: queue: TurnInbox = field(default_factory=TurnInbox) @@ -167,7 +162,7 @@ async def place(self, session: TurnSession, item: InputRequest) -> bool: images=item.content.images, context_contributions=item.content.contributions, owner_tools=session.tool_socket.listener.connection, - streaming_behavior=_PI_STREAMING_BEHAVIOR[item.when_busy], + streaming_behavior=item.when_busy.streaming(), ) def write(): if command.images: diff --git a/src/agent_comms/wake.py b/src/agent_comms/wake.py index 4c639ec01d..e72397589d 100644 --- a/src/agent_comms/wake.py +++ b/src/agent_comms/wake.py @@ -7,28 +7,29 @@ from __future__ import annotations -from enum import StrEnum - - from .audience_manifest import FrozenAudience, FrozenRecipient, freeze_audience from .channel_targets import CommandTarget, NamedTarget +from .declared_family import DeclaredFamily from .messages import Message -from .wake_policy import NoWakeDecision, RecipientDecision +from .wake_policy import NoWakeDecision, PassiveWake, RecipientDecision -class ControlClassification(StrEnum): +class MessageControl(DeclaredFamily, affix="Control"): """Explicit system-control classification; not inferred from text or ACK type.""" - ORDINARY = "ordinary" - SYSTEM_CONTROL = "system_control" + @classmethod + def wake_mode(cls, target, recipient, mentions): + return target.wake_policy(recipient, mentions) + + +class OrdinaryControl(MessageControl): + pass - @property - def passive(self) -> bool: - return self is self.SYSTEM_CONTROL - @property - def supports_initial(self) -> bool: - return self is self.ORDINARY +class SystemControl(MessageControl, declared_name="system_control"): + @classmethod + def wake_mode(cls, target, recipient, mentions): + return PassiveWake() def _require_stored(message: Message) -> None: @@ -37,12 +38,10 @@ def _require_stored(message: Message) -> None: def _validated_context( - message: Message, frozen_audience: FrozenAudience, control: ControlClassification + message: Message, frozen_audience: FrozenAudience, control: type[MessageControl] ) -> tuple[NamedTarget, frozenset[str]]: """Check the full envelope exactly once before interpreting recipient rows.""" _require_stored(message) - if type(control) is not ControlClassification: - raise TypeError("control must be an explicit ControlClassification.") if type(frozen_audience) is not FrozenAudience: raise TypeError("wake requires a FrozenAudience and FrozenRecipient.") if ( @@ -70,7 +69,7 @@ def _validated_context( def _member_decision( message: Message, recipient: FrozenRecipient, - control: ControlClassification, + control: type[MessageControl], target: NamedTarget, effective_mentions: frozenset[str], ) -> RecipientDecision: @@ -85,7 +84,7 @@ def resolve_wake( *, recipient: FrozenRecipient, frozen_audience: FrozenAudience, - control: ControlClassification, + control: type[MessageControl], ) -> RecipientDecision | None: """Resolve one member; prefer resolve_wake_cohort for the full frozen N. @@ -101,7 +100,7 @@ def resolve_wake( def resolve_wake_cohort( - message: Message, *, frozen_audience: FrozenAudience, control: ControlClassification + message: Message, *, frozen_audience: FrozenAudience, control: type[MessageControl] ) -> tuple[RecipientDecision, ...]: """Resolve ordered full N with one pure full-envelope check and no live lookup. diff --git a/tests/bus_record_fixture.py b/tests/bus_record_fixture.py index adf03269ff..9c9c7f006b 100644 --- a/tests/bus_record_fixture.py +++ b/tests/bus_record_fixture.py @@ -14,7 +14,7 @@ from agent_comms.response_acknowledgement import ( KeyedResponseAcknowledgement, canonical_publication_key, public_envelope_digest, ) -from agent_comms.wake import ControlClassification, resolve_wake_cohort +from agent_comms.wake import MessageControl, resolve_wake_cohort, OrdinaryControl publication_processes = ForkServerContext() @@ -58,9 +58,9 @@ def record_from(declaration, message, *, sender_lookup="a" * 32, recipient_looku capabilities.update( audience=audience, decisions=resolve_wake_cohort( - message, frozen_audience=audience, control=ControlClassification.ORDINARY + message, frozen_audience=audience, control=OrdinaryControl ), - control=ControlClassification.ORDINARY.value, + control=OrdinaryControl.declared_name, ) if issubclass(declaration, KeyedPublication): capabilities["acknowledgement"] = KeyedResponseAcknowledgement( diff --git a/tests/native_backend_fixture.py b/tests/native_backend_fixture.py index d796e10452..3737fc4496 100644 --- a/tests/native_backend_fixture.py +++ b/tests/native_backend_fixture.py @@ -14,7 +14,7 @@ import agent_comms.agent_events as events import agent_comms.backend as backend -from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, WhenBusy +from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, BusyInputPolicy, SteerInputPolicy from agent_comms.comms import Comms from agent_comms.fresh_private_session import create_fresh_private_session from agent_comms.child_process import BoundedRun, join_retirement @@ -186,7 +186,7 @@ async def run(self, text, *, followup=None, queue=None, **options): queue.put_nowait(InputRequest( input_id=InputId("queued"), content=InputContent(text=followup), - when_busy=WhenBusy.STEER, + when_busy=SteerInputPolicy, )) followup = None return result diff --git a/tests/test_agent_backend.py b/tests/test_agent_backend.py index 2f32efdff2..bdc0671b49 100644 --- a/tests/test_agent_backend.py +++ b/tests/test_agent_backend.py @@ -312,7 +312,7 @@ def _session_lines(path: Path) -> None: def test_inbox_records_say_which_inputs_they_carry(): request = ab.InputRequest( - input_id=ab.InputId("q1"), content=ab.InputContent(text="t"), when_busy=ab.WhenBusy.STEER, + input_id=ab.InputId("q1"), content=ab.InputContent(text="t"), when_busy=ab.SteerInputPolicy, ) assert request.carried_inputs == (ab.InputId("q1"),) assert ab.SendNow((ab.InputId("q1"),)).carried_inputs == () diff --git a/tests/test_backend.py b/tests/test_backend.py index c295094e8f..515905c4e8 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -10,7 +10,7 @@ import agent_comms.agent_events as ae import agent_comms.backend as backend -from agent_comms.agent_backend import InputContent, InputId, InputRequest, WhenBusy +from agent_comms.agent_backend import InputContent, InputId, InputRequest, BusyInputPolicy, SteerInputPolicy from agent_comms.native_pi import CAPABILITY from agent_comms.pi_rpc import PiRpcChannel from agent_comms.turn_phase import ( @@ -27,7 +27,7 @@ def steer(text: str, input_id: str | None = None) -> InputRequest: return InputRequest( input_id=InputId(input_id or f"agent-comms-steer-{uuid4().hex}"), content=InputContent(text=text), - when_busy=WhenBusy.STEER, + when_busy=SteerInputPolicy, ) diff --git a/tests/test_bus.py b/tests/test_bus.py index db63dc0118..6f4333ba10 100644 --- a/tests/test_bus.py +++ b/tests/test_bus.py @@ -33,7 +33,7 @@ Claimed, ClaimHistoryKind, InfoKind, - MembershipChange, + MembershipTransition, Message, MessageRoleWire, NoticeRole, @@ -132,7 +132,7 @@ def test_projection_freshness_follows_the_installed_core_commit(bus): "role", tuple( MessageRoleWire(change=change, notice=notice).role() - for change in (MessageRoleWire().change, *MembershipChange) + for change in (MessageRoleWire().change, *MembershipTransition.members_with(MembershipTransition)) for notice in (False, True) ), ) diff --git a/tests/test_channel_display.py b/tests/test_channel_display.py index b3ba2ed55b..7a0987364a 100644 --- a/tests/test_channel_display.py +++ b/tests/test_channel_display.py @@ -8,7 +8,7 @@ import pytest from agent_comms.channel_targets import CommandTarget -from agent_comms.channels import AnyOfMatch, SavedView, ViewKind, ViewPredicate +from agent_comms.channels import AnyOfMatch, ViewPredicate, ActivityView from agent_comms.comms import wire from agent_comms.coordination_errors import DMPaintedBoundChanged, DMScopeChanged from agent_comms.message_page import PageBounds @@ -86,12 +86,10 @@ def test_default_off_and_opt_in_only_changes_display(tmp_path): def test_mode_rejects_builtins_union_view_and_unknown_without_sidecar_mutation(tmp_path): comms = populated(tmp_path) comms.channels.set_saved_view( - SavedView( - "engineering", ViewKind.ACTIVITY, ViewPredicate(AnyOfMatch, frozenset({"api", "ui"})) - ) + ActivityView('engineering', ViewPredicate(AnyOfMatch, frozenset({'api', 'ui'}))) ) comms.channels.set_saved_view( - SavedView("saved", ViewKind.ACTIVITY, ViewPredicate(AnyOfMatch, frozenset({"api"}))) + ActivityView('saved', ViewPredicate(AnyOfMatch, frozenset({'api'}))) ) metadata = comms.channels.catalog.path before = metadata.read_bytes() if metadata.exists() else None diff --git a/tests/test_channel_retirement.py b/tests/test_channel_retirement.py index a285dffcef..39b4c2fb59 100644 --- a/tests/test_channel_retirement.py +++ b/tests/test_channel_retirement.py @@ -7,7 +7,7 @@ from agent_comms.bus import Bus, PublicRecord from agent_comms.catalog_store import ChannelCatalog -from agent_comms.channels import AnyOfMatch, SavedView, ViewKind, ViewPredicate +from agent_comms.channels import AnyOfMatch, ViewPredicate, ActivityView, ParticipantsView from agent_comms.comms import wire from agent_comms.field_codec import FieldCodec from agent_comms.message_page import PageBounds @@ -35,13 +35,7 @@ def saved_catalog(root): "#ui": ChannelPreferences(created_at=17), }, saved_views={ - "engineering": SavedView( - "engineering", - ViewKind.ACTIVITY, - ViewPredicate(AnyOfMatch, frozenset({"api", "ui"})), - 17, - frozenset({"#engineering"}), - ) + "engineering": ActivityView('engineering', ViewPredicate(AnyOfMatch, frozenset({'api', 'ui'})), 17, frozenset({'#engineering'})) }, list_order=LastUserInputChannelSort, ) @@ -111,7 +105,7 @@ def test_saved_projection_preserves_preferences_history_and_has_no_writable_alia def test_private_initial_rejects_view_but_routes_exact_tags_and_membership(tmp_path): comms, _store, _identity, _lookups = _root(tmp_path) comms.channels.set_saved_view( - SavedView("team-view", ViewKind.ACTIVITY, ViewPredicate(AnyOfMatch, frozenset({"team"}))) + ActivityView('team-view', ViewPredicate(AnyOfMatch, frozenset({'team'}))) ) before = comms.bus.bus.revision() with pytest.raises(ValueError): @@ -133,12 +127,7 @@ def test_restoring_same_named_view_keeps_current_predicate_and_original_target_h Bus.install(Bus.path_in(tmp_path / "destination")) comms = wire(tmp_path / "destination") comms.channels.create_tag("api") - current = SavedView( - "engineering", - ViewKind.PARTICIPANTS, - ViewPredicate(AnyOfMatch, frozenset({"api"})), - created_at=19, - ) + current = ParticipantsView('engineering', ViewPredicate(AnyOfMatch, frozenset({'api'})), created_at=19) comms.channels.set_saved_view(current) with comms.bus.opened_activity() as activity, comms.channels.catalog.editing() as document: document.restore_missing(source, activity, existing=True) diff --git a/tests/test_channels.py b/tests/test_channels.py index 5a441b5b99..e312a0f276 100644 --- a/tests/test_channels.py +++ b/tests/test_channels.py @@ -7,7 +7,7 @@ from agent_comms.tools import invoke_tool from agent_comms.tool_caller import ThreadCaller, NoThreadCaller -from agent_comms.channels import AllOfMatch, AnyOfMatch, BuiltinChannelView, ChannelMemberIndex, RetiredTagChannel, TagChannel, SavedView, ViewKind, ViewPredicate +from agent_comms.channels import AllOfMatch, AnyOfMatch, BuiltinChannelView, ChannelMemberIndex, RetiredTagChannel, TagChannel, ViewPredicate, ParticipantsView from agent_comms.channel_targets import CommandTarget, CommsTag, DeclaredTag from agent_comms.child_process import NoProcess, ProcessIdentity from agent_comms.channel_management import ( @@ -236,7 +236,9 @@ async def test_new_tag_disposition_is_declared_in_original_editor_and_tool_catal assert edited.bound[0].disposition == disposition assert 'only inactive' in edited.confirmation tag_tool = next(tool for tool in NativeRequestTools.catalog() if tool['name'] == CommsTagsTool.declared_name) - schema = tag_tool['parameters']['properties']['disposition'] + delete_edit = next(case for case in tag_tool['parameters']['properties']['edit']['oneOf'] + if case['properties']['kind']['const'] == 'delete') + schema = delete_edit['properties']['disposition'] assert schema['oneOf'] == FieldCodec.value_schema(TagDisposition)['oneOf'] assert any(member['properties'][TagDisposition.family_discriminator]['const'] == disposition.declared_name for member in schema['oneOf']) @@ -244,12 +246,7 @@ async def test_new_tag_disposition_is_declared_in_original_editor_and_tool_catal def test_saved_views_are_typed_persistent_and_non_routable(tmp_path): comms = setup_wire(tmp_path) - view = SavedView( - "api-and-ui", - ViewKind.PARTICIPANTS, - ViewPredicate(AllOfMatch, frozenset({"api", "ui"})), - created_at=123, - ) + view = ParticipantsView('api-and-ui', ViewPredicate(AllOfMatch, frozenset({'api', 'ui'})), created_at=123) assert comms.channels.set_saved_view(view) == view assert wire(tmp_path).channels.catalog.read().saved_views == {"api-and-ui": view} assert [ @@ -281,11 +278,7 @@ def test_tag_lifecycle_is_transactional_for_metadata_views_and_history(tmp_path) comms.channels.set_channel_metadata("api", parent="#ui", archived=True) comms.channels.set_channel_metadata("child", parent="#api", archived=False) comms.channels.set_saved_view( - SavedView( - "cross-team", - ViewKind.PARTICIPANTS, - ViewPredicate(AllOfMatch, frozenset({"api", "ui"})), - ) + ParticipantsView('cross-team', ViewPredicate(AllOfMatch, frozenset({'api', 'ui'}))) ) comms.messaging.send("other", "#api", "target-owned historical row").message.message_id @@ -320,11 +313,7 @@ def test_tag_lifecycle_is_transactional_for_metadata_views_and_history(tmp_path) assert observer.bus.require_channel("#child").parent is None observer.channels.set_saved_view( - SavedView( - "cross-team", - ViewKind.PARTICIPANTS, - ViewPredicate(AllOfMatch, frozenset({"docs"})), - ) + ParticipantsView('cross-team', ViewPredicate(AllOfMatch, frozenset({'docs'}))) ) restarted = wire(tmp_path) assert restarted.bus.require_channel("#child").parent is None @@ -345,14 +334,10 @@ def test_saved_view_names_are_reserved_for_tags_and_legacy_channels(tmp_path): comms = setup_wire(tmp_path) predicate = ViewPredicate(AnyOfMatch, frozenset({"api"})) with pytest.raises(ValueError): - comms.channels.set_saved_view(SavedView("api", ViewKind.PARTICIPANTS, predicate)) + comms.channels.set_saved_view(ParticipantsView('api', predicate)) comms.channels.set_saved_view( - SavedView( - "reserved", - ViewKind.PARTICIPANTS, - predicate, - ) + ParticipantsView('reserved', predicate) ) before = comms.registry.snapshot() with pytest.raises(ValueError): @@ -373,7 +358,7 @@ def test_all_implicit_tag_introduction_paths_honor_name_reservations(tmp_path): comms = setup_wire(tmp_path) predicate = ViewPredicate(AnyOfMatch, frozenset({"api"})) - comms.channels.set_saved_view(SavedView("reserved", ViewKind.PARTICIPANTS, predicate)) + comms.channels.set_saved_view(ParticipantsView('reserved', predicate)) before = comms.registry.snapshot() with pytest.raises(ValueError): @@ -441,7 +426,7 @@ def test_tag_operations_preserve_owner_and_update_all_views(tmp_path, monkeypatc comms = setup_wire(tmp_path) original = comms.registry.require("other") monkeypatch.setenv("AGENT_COMMS_THREAD", "other") - invoke_tool(comms, "comms_tags", {"action": "create", "name": "new"}, caller=NoThreadCaller()) + invoke_tool(comms, "comms_tags", {"edit": {"kind": "create", "name": "new"}}, caller=NoThreadCaller()) assert "#new" in comms.channels.channels() invoke_tool(comms, "comms_thread_tags", {"add": "new,api", "remove": "ui"}, caller=ThreadCaller("other")) assert comms.registry.require("other") == replace( @@ -452,13 +437,13 @@ def test_tag_operations_preserve_owner_and_update_all_views(tmp_path, monkeypatc "comms_set_view", {"name": "team", "kind": "participants", "match": "any_of", "tags": "api,ui"}, caller=NoThreadCaller(), ) - invoke_tool(comms, "comms_tags", {"action": "rename", "name": "api", "new_name": "backend"}, caller=NoThreadCaller()) + invoke_tool(comms, "comms_tags", {"edit": {"kind": "rename", "name": "api", "new_name": "backend"}}, caller=NoThreadCaller()) assert "backend" in comms.registry.require("a").tags assert wire(tmp_path).bus.require_channel("#team").tags == frozenset( {"backend", "ui"} ) invoke_tool(comms, "comms_delete_view", {"name": "team"}, caller=NoThreadCaller()) - invoke_tool(comms, "comms_tags", {"action": "delete", "name": "backend"}, caller=NoThreadCaller()) + invoke_tool(comms, "comms_tags", {"edit": {"kind": "delete", "name": "backend"}}, caller=NoThreadCaller()) assert not comms.registry.require("a").tags assert "#ui" in comms.channels.channels() assert "#team" not in comms.channels.channels() @@ -553,9 +538,7 @@ def test_channel_pins_persist_and_partition_existing_order(tmp_path, order): def test_member_pins_are_persistent_channel_scoped_and_preserve_sort(tmp_path, order): comms = setup_wire(tmp_path) comms.channels.set_saved_view( - SavedView( - "team", ViewKind.PARTICIPANTS, ViewPredicate(AnyOfMatch, frozenset({"api", "ui"})) - ) + ParticipantsView('team', ViewPredicate(AnyOfMatch, frozenset({'api', 'ui'}))) ) comms.channels.set_channel_sort("#team", order) original = {view.channel.name: view.members for view in comms.views.channel_views()} @@ -576,9 +559,7 @@ def test_member_pins_are_persistent_channel_scoped_and_preserve_sort(tmp_path, o def test_pin_lifecycle_tracks_identity_and_membership(tmp_path): comms = setup_wire(tmp_path) comms.channels.set_saved_view( - SavedView( - "team", ViewKind.PARTICIPANTS, ViewPredicate(AnyOfMatch, frozenset({"api", "ui"})) - ) + ParticipantsView('team', ViewPredicate(AnyOfMatch, frozenset({'api', 'ui'}))) ) comms.channels.set_channel_pinned("team", True) comms.channels.set_thread_pinned("team", "a", True) @@ -598,7 +579,7 @@ def test_pin_lifecycle_tracks_identity_and_membership(tmp_path): assert team.pinned_members == {"renamed"} and team.members[0] == "renamed" comms.channels.delete_saved_view("team") comms.channels.set_saved_view( - SavedView("team", ViewKind.PARTICIPANTS, ViewPredicate(AnyOfMatch, frozenset({"api"}))) + ParticipantsView('team', ViewPredicate(AnyOfMatch, frozenset({'api'}))) ) assert not observer.bus.require_channel("#team").pinned diff --git a/tests/test_exporting.py b/tests/test_exporting.py index 15eaa936e8..8153214b0e 100644 --- a/tests/test_exporting.py +++ b/tests/test_exporting.py @@ -7,7 +7,7 @@ import pytest -from agent_comms.channels import AnyOfMatch, SavedView, ViewKind, ViewPredicate +from agent_comms.channels import AnyOfMatch, ViewPredicate, ParticipantsView from agent_comms.cli import main from agent_comms.comms import Comms from agent_comms.errors import RelationViolationError @@ -325,11 +325,7 @@ def test_comms_export_uses_target_owned_channel_and_alias_aware_dm_scopes(tmp_pa assert dm_receipt.scope.participants == ("renamed", "bob") comms.channels.set_saved_view( - SavedView( - "projection", - ViewKind.PARTICIPANTS, - ViewPredicate(AnyOfMatch, frozenset({"team"})), - ) + ParticipantsView('projection', ViewPredicate(AnyOfMatch, frozenset({'team'}))) ) with pytest.raises(RelationViolationError): comms.views.export_wire( diff --git a/tests/test_field_codec.py b/tests/test_field_codec.py index afa0337fa6..9b3e98af9d 100644 --- a/tests/test_field_codec.py +++ b/tests/test_field_codec.py @@ -29,16 +29,7 @@ WireNonNull, WireTransient, ) -from agent_comms.messages import ( - ConversationRole, - InfoKind, - MembershipChange, - MembershipConversationRole, - MembershipNoticeRole, - Message, - MessageRole, - NoticeRole, -) +from agent_comms.messages import ConversationRole, InfoKind, MembershipTransition, MembershipConversationRole, MembershipNoticeRole, Message, MessageRole, NoticeRole, JoinedMembership, LeftMembership def test_transient_plan_excludes_fields_before_decoder_or_value_access(): @@ -840,8 +831,8 @@ def test_missing_and_null_membership_decode_to_the_same_role(wire, expected): [ (ConversationRole(), {}), (NoticeRole(), {"notice": True}), - (MembershipConversationRole(MembershipChange.JOINED), {"membership": "joined"}), - (MembershipNoticeRole(MembershipChange.LEFT), {"membership": "left", "notice": True}), + (MembershipConversationRole(JoinedMembership), {"membership": "joined"}), + (MembershipNoticeRole(LeftMembership), {"membership": "left", "notice": True}), ], ) def test_message_role_keeps_the_exact_external_envelope(role, external): diff --git a/tests/test_message_categories.py b/tests/test_message_categories.py index 502330c898..e5cf51f8d4 100644 --- a/tests/test_message_categories.py +++ b/tests/test_message_categories.py @@ -6,11 +6,7 @@ from agent_comms.bus_publication import Delivered from agent_comms.field_codec import FieldCodec -from agent_comms.messages import ( - Message, MessageRole, InfoKind, ConversationRole, NoticeRole, - HistoricalConversationRole, HistoricalNoticeRole, - MembershipConversationRole, MembershipNoticeRole, MembershipChange, -) +from agent_comms.messages import Message, MessageRole, InfoKind, ConversationRole, NoticeRole, HistoricalConversationRole, HistoricalNoticeRole, MembershipConversationRole, MembershipNoticeRole, MembershipTransition, JoinedMembership from agent_comms.thread_identity import AgentThreadRole, UserThreadRole from agent_comms.transcript_events import IncomingTranscript from agent_comms.transcript_sources import AssignedTranscriptSource @@ -26,8 +22,8 @@ (NoticeRole(), OtherCategory), (HistoricalConversationRole(), InboundCategory), (HistoricalNoticeRole(), OtherCategory), - (MembershipConversationRole(MembershipChange.JOINED), InboundCategory), - (MembershipNoticeRole(MembershipChange.JOINED), OtherCategory), + (MembershipConversationRole(JoinedMembership), InboundCategory), + (MembershipNoticeRole(JoinedMembership), OtherCategory), ]) @pytest.mark.parametrize("sender_role", [AgentThreadRole, UserThreadRole]) def test_role_owns_wire_filter_category(role, category, sender_role): diff --git a/tests/test_s4_ownership.py b/tests/test_s4_ownership.py index bb5c831a1b..7bf0084691 100644 --- a/tests/test_s4_ownership.py +++ b/tests/test_s4_ownership.py @@ -2,18 +2,13 @@ -from agent_comms.channels import AnyOfMatch, SavedView, ViewKind, ViewPredicate +from agent_comms.channels import AnyOfMatch, ViewPredicate, ActivityView from agent_comms.field_codec import FieldCodec from agent_comms.messages import Message, InfoKind def test_wire_formats_derive_field_names_optional_values_and_sorted_tags(): - view = SavedView( - "work", - ViewKind.ACTIVITY, - ViewPredicate(AnyOfMatch, frozenset({"z", "a"})), - created_at=7.0, - ) + view = ActivityView('work', ViewPredicate(AnyOfMatch, frozenset({'z', 'a'})), created_at=7.0) expected = { "name": "work", "kind": "activity", diff --git a/tests/test_tracked_native_retry.py b/tests/test_tracked_native_retry.py index 2891a5a390..0b0cae2f36 100644 --- a/tests/test_tracked_native_retry.py +++ b/tests/test_tracked_native_retry.py @@ -3,7 +3,7 @@ from agent_comms.native_arguments import SelectedModel from agent_comms.native_startup import NativeStartupPolicy -from agent_comms.agent_backend import WhenBusy +from agent_comms.agent_backend import BusyInputPolicy, SteerInputPolicy from agent_comms.pi_events import NativeRetryEnd from agent_comms.pi_payloads import FailedProviderRetry, PiMessage, SucceededProviderRetry from agent_comms.tracked_turn import CompletedTrackedTerminal, PendingTrackedTerminal, TrackedTurnSession @@ -96,8 +96,8 @@ def following(kind): def test_delivered_notice_preserves_original_but_real_steering_can_revise(text): notice = following(DeliveredFollowingInput) steering = following(AcceptedFollowingInput) - assert notice.when_busy is WhenBusy.STEER - assert steering.when_busy is WhenBusy.STEER + assert notice.when_busy is SteerInputPolicy + assert steering.when_busy is SteerInputPolicy response = notice.continue_response(CompletedTrackedTerminal("Requested answer"), ["Requested answer"]) response = response.tool_round().append(text) assert response.require_response([text]) == "Requested answer" diff --git a/tests/test_turn_inbox.py b/tests/test_turn_inbox.py index 7fb0842675..57f71e9241 100644 --- a/tests/test_turn_inbox.py +++ b/tests/test_turn_inbox.py @@ -2,7 +2,7 @@ -from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, WhenBusy +from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, BusyInputPolicy, FollowUpInputPolicy from agent_comms.turn_inputs import InputForwarding @@ -10,7 +10,7 @@ def test_inbox_owns_ready_work_and_retains_parked_input(): inbox = TurnInbox() inputs = InputForwarding(inbox) item = InputRequest(input_id=InputId("parked"), content=InputContent(text="later"), - when_busy=WhenBusy.FOLLOW_UP) + when_busy=FollowUpInputPolicy) assert not inbox.has_ready_work and inputs.settled inbox.put_nowait(item) assert inbox.has_ready_work and inputs.unresolved diff --git a/tests/test_wake.py b/tests/test_wake.py index 757a005dd4..1cdbc85462 100644 --- a/tests/test_wake.py +++ b/tests/test_wake.py @@ -10,18 +10,11 @@ from conftest import installed_comms from agent_comms.wake_policy import MessageAudience from agent_comms.mentions import ThreadMention -from agent_comms.messages import ( - AckKind, HandoffKind, InfoKind, MembershipChange, MembershipConversationRole, Message, MessageKind, - NoticeRole, -) +from agent_comms.messages import AckKind, HandoffKind, InfoKind, MembershipTransition, MembershipConversationRole, Message, MessageKind, NoticeRole, JoinedMembership from agent_comms.obligation_states import SilentResponse from agent_comms.response_policy import CollectivePolicy, InformationalPolicy, MentionedOnlyPolicy from agent_comms.thread_identity import AgentThreadRole, ThreadRole, UserThreadRole -from agent_comms.wake import ( - ControlClassification, - resolve_wake, - resolve_wake_cohort, -) +from agent_comms.wake import MessageControl, resolve_wake, resolve_wake_cohort, OrdinaryControl, SystemControl from agent_comms.wake_policy import BoundedTriageWake, FullWake, PassiveWake, NoWakeDecision, NoWakeReason, RecipientDecision, WakeDecision MEMBERS = (FrozenRecipient("lookup-alpha", "alpha"), FrozenRecipient("lookup-beta", "beta")) @@ -56,7 +49,7 @@ def decide( message: Message, lookup: str = "lookup-alpha", members: tuple[FrozenRecipient, ...] = MEMBERS, - control: ControlClassification = ControlClassification.ORDINARY, + control: MessageControl = OrdinaryControl, ) -> RecipientDecision | None: recipient = next( (member for member in members if member.recipient_lookup == lookup), @@ -159,10 +152,10 @@ def test_mixed_in_and_out_of_scope_mentions_only_name_valid_frozen_member() -> N assert MEMBERS[0] in audience(message).recipients -@pytest.mark.parametrize("control", list(ControlClassification)) -@pytest.mark.parametrize("role", [NoticeRole(), MembershipConversationRole(MembershipChange.JOINED)]) +@pytest.mark.parametrize("control", MessageControl.members_with(MessageControl)) +@pytest.mark.parametrize("role", [NoticeRole(), MembershipConversationRole(JoinedMembership)]) def test_notice_and_membership_are_passive_regardless_of_control_or_sender( - role, control: ControlClassification + role, control: MessageControl ) -> None: message = replace(stored(sender_role=AgentThreadRole), role=role) assert decide(message, control=control) == WakeDecision( @@ -174,7 +167,7 @@ def test_explicit_control_on_mentioned_row_is_passive_not_unmentioned_observer() message = stored("#team", "@beta investigate", mentions=(ThreadMention("beta", 0, 5),)) for member in MEMBERS: assert decide( - message, member.recipient_lookup, control=ControlClassification.SYSTEM_CONTROL + message, member.recipient_lookup, control=SystemControl ) == WakeDecision(member.recipient_lookup, MessageAudience.MENTIONED, PassiveWake()) @@ -183,20 +176,18 @@ def test_system_control_must_be_explicit_not_guessed_from_body_or_ack_type() -> assert decide(ordinary) == WakeDecision( "lookup-alpha", MessageAudience.COLLECTIVE, BoundedTriageWake() ) - assert decide(ordinary, control=ControlClassification.SYSTEM_CONTROL) == WakeDecision( + assert decide(ordinary, control=SystemControl) == WakeDecision( "lookup-alpha", MessageAudience.COLLECTIVE, PassiveWake() ) direct = stored("alpha", sender_role=UserThreadRole) - assert decide(direct, members=MEMBERS[:1], control=ControlClassification.SYSTEM_CONTROL) == ( + assert decide(direct, members=MEMBERS[:1], control=SystemControl) == ( WakeDecision("lookup-alpha", MessageAudience.DIRECT, PassiveWake()) ) - with pytest.raises(TypeError): - resolve_wake( - ordinary, - recipient=MEMBERS[0], - frozen_audience=audience(ordinary), - control="system_control", # type: ignore[arg-type] - ) + from agent_comms.field_codec import FieldCodec, MalformedRecord + + assert FieldCodec.decode(type[MessageControl], "system_control") is SystemControl + with pytest.raises(MalformedRecord): + FieldCodec.decode(type[MessageControl], "invented_control") def test_no_wake_is_not_a_passive_claim() -> None: @@ -223,7 +214,7 @@ def test_excluded_recipient_and_bad_identity_fail_closed() -> None: message, recipient=FrozenRecipient("lookup-alpha", "someone_else"), frozen_audience=manifest, - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) is None ) # Lookup alone cannot steal another frozen name's decision. @@ -232,14 +223,14 @@ def test_excluded_recipient_and_bad_identity_fail_closed() -> None: message, recipient="alpha", # type: ignore[arg-type] frozen_audience=manifest, - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) with pytest.raises(TypeError): resolve_wake( message, recipient=MEMBERS[0], frozen_audience=object(), # type: ignore[arg-type] - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) with pytest.raises(ValueError): decide(stored("alpha"), members=MEMBERS) @@ -250,7 +241,7 @@ def test_excluded_recipient_and_bad_identity_fail_closed() -> None: replace(message, seq=0), recipient=MEMBERS[0], frozen_audience=manifest, - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) @@ -271,14 +262,14 @@ def test_prepared_message_must_match_frozen_manifest_identity(field: str) -> Non mismatched, recipient=MEMBERS[0], frozen_audience=manifest, - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) @pytest.mark.parametrize("changes", [ {"mentions": ()}, pytest.param({"role": NoticeRole()}, id="notice"), - pytest.param({"role": MembershipConversationRole(MembershipChange.JOINED)}, id="membership"), + pytest.param({"role": MembershipConversationRole(JoinedMembership)}, id="membership"), {"type": HandoffKind}, {"sender_role": UserThreadRole}, ]) @@ -293,7 +284,7 @@ def test_same_message_id_does_not_override_frozen_full_envelope(changes) -> None changed, recipient=MEMBERS[0], frozen_audience=manifest, - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) @@ -302,13 +293,13 @@ def test_full_cohort_preserves_direct_control_and_fail_closed_envelope() -> None assert resolve_wake_cohort( direct, frozen_audience=audience(direct, MEMBERS[:1]), - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) == (WakeDecision("lookup-alpha", MessageAudience.DIRECT, FullWake()),) collective = stored() assert resolve_wake_cohort( collective, frozen_audience=audience(collective), - control=ControlClassification.SYSTEM_CONTROL, + control=SystemControl, ) == tuple( WakeDecision(member.recipient_lookup, MessageAudience.COLLECTIVE, PassiveWake()) for member in MEMBERS @@ -317,13 +308,13 @@ def test_full_cohort_preserves_direct_control_and_fail_closed_envelope() -> None resolve_wake_cohort( direct, frozen_audience=audience(direct, (MEMBERS[1],)), - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) with pytest.raises(ValueError): resolve_wake_cohort( replace(collective, role=NoticeRole()), frozen_audience=audience(collective), - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) @@ -365,7 +356,7 @@ def test_pure_shadow_does_not_mutate_envelope_state_or_live_cursor(tmp_path: Pat message, recipient=MEMBERS[0], frozen_audience=manifest, - control=ControlClassification.ORDINARY, + control=OrdinaryControl, ) == WakeDecision("lookup-alpha", MessageAudience.COLLECTIVE, BoundedTriageWake()) assert message.exact_reply_target == "#team" assert message.to_wire() == before diff --git a/tests/verify_connector_tools.py b/tests/verify_connector_tools.py index 749282c364..dd70826e2d 100644 --- a/tests/verify_connector_tools.py +++ b/tests/verify_connector_tools.py @@ -24,7 +24,7 @@ from pathlib import Path from agent_comms.change_report import FileLine -from agent_comms.channels import AnyOfMatch, SavedView, ViewKind, ViewPredicate +from agent_comms.channels import AnyOfMatch, ViewPredicate, ParticipantsView from agent_comms.child_process import ProcessIdentity from agent_comms.comms import Comms from agent_comms.connector import Connector @@ -54,9 +54,7 @@ def seed(root: Path, project: Path) -> None: ) Bus.install(Bus.path_in(comms.root)) comms.channels.set_saved_view( - SavedView( - "probe-view", ViewKind.PARTICIPANTS, ViewPredicate(AnyOfMatch, frozenset({"probe"})) - ) + ParticipantsView('probe-view', ViewPredicate(AnyOfMatch, frozenset({'probe'}))) ) comms.goals.update_goal( "claude-ai", SetGoalAction(text="Measure the connector calls.", goal_id="probe-goal") From e65d869166bb612bd24ab1883cddc15e09c02106 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:04:56 -0400 Subject: [PATCH 2/4] Move execution source and retry policy onto provenance cases --- src/agent_comms/cohort_schema.py | 4 +- src/agent_comms/coordination_snapshot.py | 7 +- .../coordination_tables/executions.py | 119 ++++++++++++++---- .../coordination_tables/responses.py | 12 +- src/agent_comms/execution_store.py | 12 +- src/agent_comms/plan_phases.py | 12 +- src/agent_comms/selected_participant.py | 4 +- src/agent_comms/selected_triage.py | 4 +- src/agent_comms/selected_turn.py | 4 +- tests/test_acp_private_nk_delivery.py | 4 +- tests/test_claim_admission_verifier.py | 4 +- tests/test_claim_bus_read.py | 4 +- tests/test_coordination.py | 20 +-- tests/test_coordination_nominal.py | 6 +- tests/test_coordination_response.py | 4 +- tests/test_coordination_store.py | 18 +-- tests/test_lifecycle_subclasses.py | 6 +- tests/test_native_failure_recovery.py | 4 +- tests/test_native_unknown_recovery.py | 4 +- tests/test_optional_awareness_projection.py | 10 +- tests/test_s3_legality_behavior.py | 6 +- 21 files changed, 171 insertions(+), 97 deletions(-) diff --git a/src/agent_comms/cohort_schema.py b/src/agent_comms/cohort_schema.py index e4954330f9..5988a0f6c2 100644 --- a/src/agent_comms/cohort_schema.py +++ b/src/agent_comms/cohort_schema.py @@ -336,11 +336,9 @@ def require_selection(self, assignment: WakeAssignment, acknowledgement: ClaimBa def require_obligation(self, execution, link, assignment, obligation) -> None: """Captured ownership and the reply's original relational key agree.""" - from .coordination_tables.executions import ExecutionOrigin from .wake_candidate_index import ProjectionUnavailableError - if execution.origin is not ExecutionOrigin.WIRE: - raise ProjectionUnavailableError("open obligation has no wire origin") + execution.origin.require_obligation() original_owner = OwnerGenerations( owner_lookup=execution.owner_lookup, owner_thread=execution.owner_thread, generation=self.owner_generation, diff --git a/src/agent_comms/coordination_snapshot.py b/src/agent_comms/coordination_snapshot.py index a9a8ca174e..2131aa551f 100644 --- a/src/agent_comms/coordination_snapshot.py +++ b/src/agent_comms/coordination_snapshot.py @@ -23,7 +23,6 @@ ) from agent_comms.coordination_tables.executions import ( CurrentExecutions, - ExecutionOrigin, ExecutionRecord, NoPointedAttempt, NoPointedExecution, @@ -208,11 +207,7 @@ def validate_current_pointer(self) -> None: raise IntegrityViolationError("snapshot current pointer is inconsistent") def validate_response_route(self) -> None: - if self.execution.origin is ExecutionOrigin.WIRE: - if not self.assignments or not self.obligations: - raise IntegrityViolationError("wire snapshots require original obligations") - elif self.obligations or self.assignments: - raise IntegrityViolationError("claimless snapshots cannot have claims or obligations") + self.execution.origin.validate_routes(self.assignments, self.obligations) expected = {row.lifecycle.exact_target for row in self.assignments} targets = tuple(row.exact_target for row in self.obligations) if len(set(targets)) != len(targets) or set(targets) != expected: diff --git a/src/agent_comms/coordination_tables/executions.py b/src/agent_comms/coordination_tables/executions.py index 6b51f79a58..8bdc95267f 100644 --- a/src/agent_comms/coordination_tables/executions.py +++ b/src/agent_comms/coordination_tables/executions.py @@ -5,7 +5,6 @@ import sqlite3 from abc import ABC, abstractmethod from dataclasses import dataclass, replace -from enum import StrEnum from typing import Annotated, ClassVar from agent_comms.coordination_contracts import ( @@ -16,6 +15,7 @@ ) from agent_comms.coordination_errors import IdentityConflict, IntegrityViolationError, StaleRevision from agent_comms.coordination_schema import CoordinatorRow +from agent_comms.declared_family import DeclaredFamily from agent_comms.coordination_tables.attempts import ( APPROVED_REPLAY_FACT_MASK, AttemptRecord, @@ -52,14 +52,95 @@ NonIncreasing, OrdinalTransitions, TextStorage, + sql_literal, ) -class ExecutionOrigin(StrEnum): - WIRE = "wire" - ACP = "acp" - GOAL = "goal" - SYSTEM = "system" +class ExecutionProvenance(DeclaredFamily, affix="Origin"): + """The origin owns source membership, route eligibility and retry policy.""" + + @classmethod + @abstractmethod + def require_sources(cls, sources) -> None: ... + + @classmethod + @abstractmethod + def validate_routes(cls, assignments, obligations) -> None: ... + + @classmethod + @abstractmethod + def retry_authorized(cls, obligations) -> bool: ... + + @classmethod + @abstractmethod + def require_obligation(cls) -> None: ... + + +class ClaimlessOrigin(ExecutionProvenance, declared_name=None): + @classmethod + def require_sources(cls, sources) -> None: + if sources: + raise IdentityConflict("claimless execution cannot bind claims") + + @classmethod + def validate_routes(cls, assignments, obligations) -> None: + if assignments or obligations: + raise IntegrityViolationError("claimless snapshots cannot have claims or obligations") + + @classmethod + def retry_authorized(cls, obligations) -> bool: + return True + + @classmethod + def require_obligation(cls) -> None: + from agent_comms.wake_candidate_index import ProjectionUnavailableError + + raise ProjectionUnavailableError("open obligation has no wire origin") + + +class WireOrigin(ExecutionProvenance): + @classmethod + def require_sources(cls, sources) -> None: + if not sources: + raise IdentityConflict("wire execution requires ordered original sources") + + @classmethod + def validate_routes(cls, assignments, obligations) -> None: + if not assignments or not obligations: + raise IntegrityViolationError("wire snapshots require original obligations") + + @classmethod + def retry_authorized(cls, obligations) -> bool: + return bool(obligations) and all(row.lifecycle.retryable for row in obligations) + + @classmethod + def require_obligation(cls) -> None: + pass + + @classmethod + def matches_sql(cls, column: str) -> str: + return f"{column} = {sql_literal(cls)}" + + @classmethod + def retry_sql(cls, execution: str) -> str: + return ( + f"({execution}.origin != {sql_literal(cls)} OR (EXISTS (SELECT 1 FROM obligations o\n" + f" WHERE o.execution_id={execution}.execution_id) AND NOT EXISTS " + "(SELECT 1 FROM obligations o\n" + f" WHERE o.execution_id={execution}.execution_id AND o.retryable != 1)))" + ) + + +class AcpOrigin(ClaimlessOrigin): + pass + + +class GoalOrigin(ClaimlessOrigin): + pass + + +class SystemOrigin(ClaimlessOrigin): + pass class PointerExecution(SubclassUnion, TextChoice, affix="Execution"): @@ -174,14 +255,12 @@ def references(cls): ), ) -_RETRY_BASIS = """SELECT e.execution_id, +_RETRY_BASIS = f"""SELECT e.execution_id, CASE WHEN e.current_attempt_ordinal IS NOT NULL AND e.current_attempt_ordinal < e.max_attempts AND EXISTS (SELECT 1 FROM replay_assessments r WHERE r.execution_id = e.execution_id AND r.retry_authorized = 1) - AND (e.origin != 'wire' OR (EXISTS (SELECT 1 FROM obligations o - WHERE o.execution_id=e.execution_id) AND NOT EXISTS (SELECT 1 FROM obligations o - WHERE o.execution_id=e.execution_id AND o.retryable != 1))) + AND {WireOrigin.retry_sql("e")} THEN 1 ELSE 0 END AS authorized FROM executions e""" @@ -199,12 +278,12 @@ class ExecutionRecord( ): column_orders = (NonDecreasing(("updated_at_ms",)),) column_guards = ( - ColumnGuard(("lifecycle",), """(json_extract(NEW.lifecycle, '$.kind') - IN ({optional_attempt_execution_names}) AND json_extract(NEW.lifecycle, '$.ordinal') + ColumnGuard(("lifecycle",), f"""(json_extract(NEW.lifecycle, '$.kind') + IN ({{optional_attempt_execution_names}}) AND json_extract(NEW.lifecycle, '$.ordinal') IS NOT NULL AND NOT EXISTS (SELECT 1 FROM attempts a WHERE a.execution_id = NEW.execution_id AND a.attempt_ordinal = json_extract(NEW.lifecycle, '$.ordinal') - AND a.owner_lookup = NEW.owner_lookup AND a.phase IN ({failed_attempt_phases}))) - OR ((NEW.origin = 'wire') AND ((NOT EXISTS (SELECT 1 FROM obligations + AND a.owner_lookup = NEW.owner_lookup AND a.phase IN ({{failed_attempt_phases}}))) + OR (({WireOrigin.matches_sql("NEW.origin")}) AND ((NOT EXISTS (SELECT 1 FROM obligations WHERE execution_id = NEW.execution_id)) OR (json_extract(NEW.lifecycle, '$.kind') = 'completed' AND EXISTS (SELECT 1 FROM obligations WHERE execution_id = NEW.execution_id @@ -232,10 +311,7 @@ def retry_authorized(self, replay: ReplayState, obligations) -> bool: return ( self.lifecycle.has_retry_budget(self.max_attempts) and replay.allows_retry - and ( - self.origin is not ExecutionOrigin.WIRE - or (bool(obligations) and all(row.lifecycle.retryable for row in obligations)) - ) + and self.origin.retry_authorized(obligations) ) execution_id: Annotated[ @@ -250,7 +326,7 @@ def retry_authorized(self, replay: ReplayState, obligations) -> bool: ), ), ] - origin: Annotated[ExecutionOrigin, FrozenColumn()] + origin: Annotated[type[ExecutionProvenance], FrozenColumn()] lifecycle: Annotated[ExecutionState, SnapshotExclude()] owner_thread: Annotated[str, FrozenColumn(check="length(owner_thread) BETWEEN 1 AND 256")] owner_lookup: Annotated[str, FrozenColumn()] @@ -274,7 +350,6 @@ def snapshot_current_attempt_ordinal(self): def __post_init__(self) -> None: validate_execution_id(self.execution_id) - object.__setattr__(self, "origin", ExecutionOrigin(self.origin)) for field, value in ( ("owner_thread", self.owner_thread), ("owner_lookup", self.owner_lookup), @@ -332,11 +407,11 @@ def __post_init__(self) -> None: ) wire_execution_id: ClassVar[Generated] = Generated( str, - Column(generated="CASE WHEN origin = 'wire' THEN execution_id END", nullable=True), + Column(generated=f"CASE WHEN {WireOrigin.matches_sql('origin')} THEN execution_id END", nullable=True), ) wire_claim_ordinal: ClassVar[Generated] = Generated( int, - Column(generated="CASE WHEN origin = 'wire' THEN 0 END", nullable=True), + Column(generated=f"CASE WHEN {WireOrigin.matches_sql('origin')} THEN 0 END", nullable=True), ) claim_status_kind: ClassVar[Generated] = Generated( str, diff --git a/src/agent_comms/coordination_tables/responses.py b/src/agent_comms/coordination_tables/responses.py index cfb9e0d96a..6ee63d8c12 100644 --- a/src/agent_comms/coordination_tables/responses.py +++ b/src/agent_comms/coordination_tables/responses.py @@ -12,7 +12,7 @@ validate_execution_id, ) from agent_comms.coordination_schema import CoordinatorRow -from agent_comms.coordination_tables.executions import ExecutionReference, RetryPartition +from agent_comms.coordination_tables.executions import ExecutionReference, RetryPartition, WireOrigin from agent_comms.field_codec import SnapshotExclude, WireName, projected from agent_comms.obligation_states import ResponseState from agent_comms.reason_codes import ReasonCode @@ -60,10 +60,10 @@ class ResponseObligation( AND exact_target = NEW.exact_target AND message_id = json_extract(NEW.lifecycle, '$.message_id') AND seq = json_extract(NEW.lifecycle, '$.seq'))))""" - insert_condition = """(NOT EXISTS (SELECT 1 FROM executions e + insert_condition = f"""(NOT EXISTS (SELECT 1 FROM executions e JOIN wake_claims c ON c.execution_id=e.execution_id WHERE e.execution_id=NEW.execution_id - AND e.origin='wire' AND c.exact_target=NEW.exact_target)) - OR (json_extract(NEW.lifecycle, '$.kind') IN ({required_intent_response_names}))""" + AND {WireOrigin.matches_sql("e.origin")} AND c.exact_target=NEW.exact_target)) + OR (json_extract(NEW.lifecycle, '$.kind') IN ({{required_intent_response_names}}))""" column_guards = ( ColumnGuard(("lifecycle",), """(EXISTS (SELECT 1 FROM publication_receipts WHERE execution_id = OLD.execution_id AND exact_target = OLD.exact_target) @@ -71,9 +71,9 @@ class ResponseObligation( SELECT 1 FROM publication_intents WHERE execution_id = OLD.execution_id AND exact_target = OLD.exact_target) AND json_extract(NEW.lifecycle, '$.kind') NOT IN ({intent_response_names}))"""), - ColumnGuard(("exact_target",), """(NOT EXISTS (SELECT 1 FROM executions e + ColumnGuard(("exact_target",), f"""(NOT EXISTS (SELECT 1 FROM executions e JOIN wake_claims c ON c.execution_id=e.execution_id WHERE e.execution_id=NEW.execution_id - AND e.origin='wire' AND c.exact_target=NEW.exact_target)) OR (EXISTS ( + AND {WireOrigin.matches_sql("e.origin")} AND c.exact_target=NEW.exact_target)) OR (EXISTS ( SELECT 1 FROM publication_intents WHERE execution_id = OLD.execution_id AND exact_target = OLD.exact_target))"""), ) diff --git a/src/agent_comms/execution_store.py b/src/agent_comms/execution_store.py index 5039a23f9b..c8462c5a03 100644 --- a/src/agent_comms/execution_store.py +++ b/src/agent_comms/execution_store.py @@ -14,7 +14,7 @@ from agent_comms.coordination_snapshot import RecoverySnapshot from agent_comms.coordination_tables.assignments import ExecutionAssignmentLink, WakeAssignment from agent_comms.coordination_tables.executions import ( - ExecutionOrigin, + ExecutionProvenance, ExecutionRecord, ) from agent_comms.coordination_tables.responses import ResponseObligation @@ -37,7 +37,7 @@ class ExecutionStore(RecoveryReader): def create( self, execution_id: str, - origin: ExecutionOrigin, + origin: type[ExecutionProvenance], owner_lookup: str, owner_thread: str, max_attempts: int, @@ -51,7 +51,7 @@ def create( def create_after_triage( self, execution_id: str, - origin: ExecutionOrigin, + origin: type[ExecutionProvenance], owner_lookup: str, owner_thread: str, max_attempts: int, @@ -69,15 +69,11 @@ def create_after_triage( def _create(self, db, execution_id, origin, owner_lookup, owner_thread, max_attempts, sources, assignments): """The single creation algorithm, under its caller's owning transaction.""" - origin = ExecutionOrigin(origin) for source in sources: if source.store is not self.assignments: raise IdentityConflict("execution source belongs to another assignment store") assignment_ids = tuple(source.assignment_id for source in sources) - if origin is ExecutionOrigin.WIRE and not sources: - raise IdentityConflict("wire execution requires ordered original sources") - if origin is not ExecutionOrigin.WIRE and sources: - raise IdentityConflict("claimless execution cannot bind claims") + origin.require_sources(sources) row = ExecutionRecord.one(self.session._connection, execution_id=execution_id) if row is not None: snapshot = self.get(execution_id) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 594d66bdbd..3994a1e2f2 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -650,9 +650,15 @@ class OwnedDispatchPhase(RemainingSlopPhase): class KindFamiliesPhase(RemainingSlopPhase): """6.3: kinds are families whose members carry their behaviour and spelling.""" - evidence = PatternAbsent( - r"class (TagAction|ExecutionOrigin|WhenBusy|MessageControl|MembershipTransition" - r"|ViewKind)\b" + evidence = AllEvidence( + ( + PatternAbsent( + r"class (TagAction|ExecutionOrigin|WhenBusy|ControlClassification" + r"|MembershipChange)\b" + ), + # SQLite's ViewKind(CatalogKind) is already a behavior-owning member. + PatternAbsent(r"class ViewKind\((StrEnum|Enum)\)"), + ) ) diff --git a/src/agent_comms/selected_participant.py b/src/agent_comms/selected_participant.py index a689fe7257..5990e03411 100644 --- a/src/agent_comms/selected_participant.py +++ b/src/agent_comms/selected_participant.py @@ -20,7 +20,7 @@ from .coordination_response import LiveResponseOwner, ResponseTable from .coordination_snapshot import RecoverySnapshot from .coordination_tables.assignments import WakeAssignment -from .coordination_tables.executions import ExecutionOrigin +from .coordination_tables.executions import ExecutionProvenance, WireOrigin from .coordinator import Coordination from .cursor_publication import CursorDelivery from .errors import RelationViolationError @@ -89,7 +89,7 @@ async def fail_before_send(self, max_attempts: int) -> None: """Settle captured originals when acquisition fails before any native send.""" await self.require_current() snapshot = self.store.create( - self.batch.execution_id, ExecutionOrigin.WIRE, + self.batch.execution_id, WireOrigin, self.lookup, self.owner.thread.name, max_attempts, sources=self.batch.sources, ).value diff --git a/src/agent_comms/selected_triage.py b/src/agent_comms/selected_triage.py index cc216f73fc..3f81eb9fb7 100644 --- a/src/agent_comms/selected_triage.py +++ b/src/agent_comms/selected_triage.py @@ -108,11 +108,11 @@ def settle(self, store, db, current): pass async def continue_turn(self, participant, session, input_id, execution, settled): - from .coordination_tables.executions import ExecutionOrigin + from .coordination_tables.executions import ExecutionProvenance, WireOrigin await participant.require_current() created = participant.store.create_after_triage( - participant.batch.execution_id, ExecutionOrigin.WIRE, + participant.batch.execution_id, WireOrigin, participant.lookup, participant.owner.thread.name, execution.max_attempts, sources=participant.batch.sources, settled=settled, diff --git a/src/agent_comms/selected_turn.py b/src/agent_comms/selected_turn.py index e2b7cf0949..14fe68b524 100644 --- a/src/agent_comms/selected_turn.py +++ b/src/agent_comms/selected_turn.py @@ -11,7 +11,7 @@ from .channel_coding_tools import SelectedCodingToolOwner from .coordination_errors import CoordinationError, IdentityConflict, StaleFence from .coordinator import Coordination -from .coordination_tables.executions import ExecutionOrigin +from .coordination_tables.executions import ExecutionProvenance, WireOrigin from .coordination_snapshot import RecoverySnapshot from .coordination_tables.responses import ResponseObligation from .durable_turn import DurableTurn @@ -256,7 +256,7 @@ async def run(self, execution, session): if not participant.batch.requires_triage: await participant.require_current() created = participant.store.create( - participant.batch.execution_id, ExecutionOrigin.WIRE, + participant.batch.execution_id, WireOrigin, participant.lookup, participant.owner.thread.name, execution.max_attempts, sources=participant.batch.sources, diff --git a/tests/test_acp_private_nk_delivery.py b/tests/test_acp_private_nk_delivery.py index 0fd3e2b835..a3458ce83f 100644 --- a/tests/test_acp_private_nk_delivery.py +++ b/tests/test_acp_private_nk_delivery.py @@ -74,7 +74,7 @@ def cursor_envelope(metadata): def test_private_owner_rename_during_active_attempt_can_complete(tmp_path, managed): from agent_comms.attempt_start import AttemptStart from agent_comms.attempt_states import SucceededAttempt - from agent_comms.coordination_tables.executions import ExecutionOrigin + from agent_comms.coordination_tables.executions import ExecutionProvenance, AcpOrigin from agent_comms.owner_fence import prepare_fence_token from test_coordination_store import final_evidence @@ -82,7 +82,7 @@ def test_private_owner_rename_during_active_attempt_can_complete(tmp_path, manag lookup = RecordedThreadIncarnation("fixture", comms.registry.require("beta").created_at).lookup with Coordination(str(comms.root / "coordination.sqlite3")) as store: before = store.participants.get(lookup) - store.create("rename-active", ExecutionOrigin.ACP, lookup, "beta", 1) + store.create("rename-active", AcpOrigin, lookup, "beta", 1) store.mark_pending("rename-active", expected_revision=1) fence = store.start(AttemptStart( "rename-active", 1, "beta", 1, prepare_fence_token(), diff --git a/tests/test_claim_admission_verifier.py b/tests/test_claim_admission_verifier.py index 7510471afd..1998334398 100644 --- a/tests/test_claim_admission_verifier.py +++ b/tests/test_claim_admission_verifier.py @@ -27,7 +27,7 @@ from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort, sealed_cohort_assignments from agent_comms.coordination_errors import IdentityConflict -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, WireOrigin from agent_comms.coordinator import Coordination from agent_comms.envelope_claim_transitions import ExistingFileClaim, WakeClaim from agent_comms.errors import RelationViolationError @@ -83,7 +83,7 @@ def test_selected_wake_verifier_refuses_no_wake_and_stale_authority( execution_id = "verifier-execution" store.create( execution_id, - ExecutionOrigin.WIRE, + WireOrigin, alice_lookup, owner.name, 1, diff --git a/tests/test_claim_bus_read.py b/tests/test_claim_bus_read.py index 0a44fbc6e5..071aa50ce4 100644 --- a/tests/test_claim_bus_read.py +++ b/tests/test_claim_bus_read.py @@ -31,7 +31,7 @@ from agent_comms.comms import Comms from agent_comms.coordination_cohort import accept_delivery_cohort, sealed_cohort_assignments from agent_comms.coordination_errors import IdentityConflict -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, WireOrigin from agent_comms.coordinator import Coordination from agent_comms.envelope_claim_transitions import ExistingFileClaim, NativeClaim, WakeClaim from agent_comms.coordination_session import NativeBoundary @@ -104,7 +104,7 @@ def selected_turn(private_comms, private_directory): ) person = store.participants.get(lookup) store.create( - "execution", ExecutionOrigin.WIRE, lookup, owner.name, 1, + "execution", WireOrigin, lookup, owner.name, 1, sources=(SelectedSource(assignment.assignment_id, store.assignments, initial),), ) snapshot = store.get("execution") diff --git a/tests/test_coordination.py b/tests/test_coordination.py index 9aa5cece55..baed606b71 100644 --- a/tests/test_coordination.py +++ b/tests/test_coordination.py @@ -43,7 +43,11 @@ from agent_comms.coordination_tables.executions import NoReplayAssessment, ReplayAssessments from agent_comms.coordination_tables.executions import ( CurrentExecutions, - ExecutionOrigin, + ExecutionProvenance, + WireOrigin, + AcpOrigin, + GoalOrigin, + SystemOrigin, ExecutionRecord, ) from agent_comms.coordination_tables.participants import OwnerGenerations, Participants @@ -71,7 +75,7 @@ from agent_comms.wake_policy import FullWake -def execution(*, status=PendingExecution, ordinal=None, origin=ExecutionOrigin.ACP): +def execution(*, status=PendingExecution, ordinal=None, origin=AcpOrigin): return ExecutionRecord( execution_id="execution-1", origin=origin, @@ -134,7 +138,7 @@ def snapshot( *, execution_record=None, attempt_record=None, - origin=ExecutionOrigin.ACP, + origin=AcpOrigin, current=False, response=None, replay=None, @@ -145,7 +149,7 @@ def snapshot( attempt=attempt_record, links=( (ExecutionAssignmentLink(record.execution_id, "claim-1", 0),) - if record.origin is ExecutionOrigin.WIRE + if record.origin is WireOrigin else () ), assignments=( @@ -161,11 +165,11 @@ def snapshot( ) ), ) - if record.origin is ExecutionOrigin.WIRE + if record.origin is WireOrigin else () ), replay=replay or ReplayAssessments(record.execution_id, ReplayFact.NONE, True, False, 1), - obligations=(response,) if record.origin is ExecutionOrigin.WIRE and response else (), + obligations=(response,) if record.origin is WireOrigin and response else (), publication_intents=(), publication_evidence=(), connectivity=None, @@ -189,7 +193,7 @@ def sql_execution( thread="worker", ): ExecutionRecord( - execution_id=name, origin=ExecutionOrigin(origin), + execution_id=name, origin=ExecutionProvenance.member_named(origin), lifecycle=ExecutionState.member_named(status).load(ordinal), owner_thread=thread, owner_lookup=owner, revision=1, max_attempts=max_attempts, reason_code=NoReasonCode(), created_at_ms=0, updated_at_ms=0, @@ -360,7 +364,7 @@ def test_snapshot_projection_retry_and_inert_secrets(): def test_wire_claim_obligation_and_publication_contract(): - record = execution(origin=ExecutionOrigin.WIRE) + record = execution(origin=WireOrigin) valid = snapshot(execution_record=record, response=obligation()) assert FieldCodec.project(valid, "snapshot")["claims"][0]["claim_id"] == "claim-1" primitive = FieldCodec.project(valid, "snapshot") diff --git a/tests/test_coordination_nominal.py b/tests/test_coordination_nominal.py index 7fc8d29cd7..6aaecad815 100644 --- a/tests/test_coordination_nominal.py +++ b/tests/test_coordination_nominal.py @@ -10,7 +10,7 @@ from agent_comms.attempt_states import SucceededAttempt from agent_comms.coordination_errors import IntegrityViolationError from agent_comms.coordination_tables.assignments import WakeAssignment -from agent_comms.coordination_tables.executions import ExecutionOrigin, ExecutionRecord +from agent_comms.coordination_tables.executions import ExecutionProvenance, ExecutionRecord, WireOrigin, AcpOrigin from agent_comms.execution_states import ( ActiveExecution, ExecutionState, @@ -43,7 +43,7 @@ def test_state_data_cannot_be_attached_to_wrong_variant(): def test_record_replacement_uses_only_nominal_state(): record = ExecutionRecord( execution_id="e", - origin=ExecutionOrigin.ACP, + origin=AcpOrigin, lifecycle=PendingExecution(), owner_thread="t", owner_lookup="o", @@ -120,7 +120,7 @@ def successors(cls): store.participants.register("p", "owner", "owner", committed=True) with store.session.writing() as db: ExecutionRecord( - execution_id="e", origin=ExecutionOrigin.WIRE, + execution_id="e", origin=WireOrigin, lifecycle=QueuedExecution(), owner_thread="owner", owner_lookup="p", revision=1, max_attempts=2, reason_code=NoReasonCode(), created_at_ms=0, updated_at_ms=0, diff --git a/tests/test_coordination_response.py b/tests/test_coordination_response.py index e9fd6520c0..cc2e3636b8 100644 --- a/tests/test_coordination_response.py +++ b/tests/test_coordination_response.py @@ -26,7 +26,7 @@ ) from agent_comms.coordination_response import LiveResponseOwner, prepare_fenced_response, publish_fenced_response, resolve_existing_response from agent_comms.coordination_results import AlreadyApplied -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, WireOrigin from agent_comms.coordinator import Coordination from agent_comms.coordination_tables.responses import ResponseObligation from agent_comms.execution_states import CompletedExecution @@ -97,7 +97,7 @@ def _ready(tmp_path: Path, *, direct: bool = False, extra_targets: tuple[str, .. assert reply_target is not None store.create( "exec", - ExecutionOrigin.WIRE, + WireOrigin, recipient.recipient_lookup, "owner", 1, diff --git a/tests/test_coordination_store.py b/tests/test_coordination_store.py index a94891ccbe..d0538b2c18 100644 --- a/tests/test_coordination_store.py +++ b/tests/test_coordination_store.py @@ -51,7 +51,7 @@ ReplayAssessment, ReplayAssessments, ) -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, WireOrigin, AcpOrigin, GoalOrigin from agent_comms.coordination_tables.publications import ( PublicationIntents, ) @@ -99,12 +99,12 @@ def wire_source(db, assignment_id: str, original: Message, mode: WakePolicy): def ready( - path: Path, *, origin: ExecutionOrigin = ExecutionOrigin.WIRE, max_attempts: int = 2 + path: Path, *, origin: type[ExecutionProvenance] = WireOrigin, max_attempts: int = 2 ) -> StandaloneCoordination: db = store(path) sources = () assert isinstance(db.participants.register("owner", "Owner", "thread", committed=True), Applied) - if origin is ExecutionOrigin.WIRE: + if origin is WireOrigin: original = Message("owner", "Owner", "Original request", InfoKind, timestamp=1.0, seq=1) db.assignments.accept( WakeAssignment( @@ -280,7 +280,7 @@ def test_claim_execution_idempotency_and_rollback(db_path: Path) -> None: original = Message("owner", "Owner", "Original request", InfoKind, timestamp=1.0, seq=1) replay = db.create( "exec", - ExecutionOrigin.WIRE, + WireOrigin, "owner", "thread", 2, @@ -292,7 +292,7 @@ def test_claim_execution_idempotency_and_rollback(db_path: Path) -> None: with pytest.raises(IdentityConflict): db.create( "exec", - ExecutionOrigin.WIRE, + WireOrigin, "owner", "thread", 2, @@ -400,7 +400,7 @@ def test_claim_preengagement_and_mixed_engagement_rollback(db_path: Path) -> Non with pytest.raises((IdentityConflict, sqlite3.IntegrityError)): db.create( "exec", - ExecutionOrigin.WIRE, + WireOrigin, "owner", "thread", 2, @@ -503,7 +503,7 @@ def test_token_digest_is_global_and_only_digest_persists(db_path: Path) -> None: with ready(db_path) as db: token, _ = started(db) db.participants.register("other", "Other", "other-thread", committed=True) - db.create("other-exec", ExecutionOrigin.GOAL, "other", "other-thread", 1) + db.create("other-exec", GoalOrigin, "other", "other-thread", 1) db.mark_pending("other-exec", expected_revision=1) with pytest.raises(IdentityConflict): db.start( @@ -529,7 +529,7 @@ def test_settled_snapshot_encodes_recovery_kind_separately_from_table(db_path: P with store(db_path) as db: db.participants.register("owner", "Owner", "thread", committed=True) - db.create("exec", ExecutionOrigin.ACP, "owner", "thread", 1) + db.create("exec", AcpOrigin, "owner", "thread", 1) db.mark_pending("exec", expected_revision=1) _, fence = started(db) fence = final_evidence(db, fence) @@ -551,7 +551,7 @@ def test_settled_snapshot_encodes_recovery_kind_separately_from_table(db_path: P def test_nonpublication_claimless_atomic_settlement(db_path: Path) -> None: - with ready(db_path, origin=ExecutionOrigin.ACP) as db: + with ready(db_path, origin=AcpOrigin) as db: _, fence = started(db) fence = final_evidence(db, fence) settled = db.settle_nonpublication( diff --git a/tests/test_lifecycle_subclasses.py b/tests/test_lifecycle_subclasses.py index 11f552ab9e..1372919cd3 100644 --- a/tests/test_lifecycle_subclasses.py +++ b/tests/test_lifecycle_subclasses.py @@ -27,7 +27,7 @@ ) from agent_comms.attempt_recovery import MonitorEvidence from agent_comms.coordination_tables.attempts import AttemptRecord, ReplayFact -from agent_comms.coordination_tables.executions import CurrentExecutions, ExecutionOrigin, ExecutionRecord +from agent_comms.coordination_tables.executions import CurrentExecutions, ExecutionProvenance, ExecutionRecord, AcpOrigin from agent_comms.execution_states import DeferredExecution, FailedExecution from agent_comms.owner_fence import prepare_fence_token from test_coordination_store import final_evidence, ready, started @@ -64,7 +64,7 @@ def test_attempt_transition_table_expands_declared_failure_subclasses(): @pytest.mark.parametrize("authorized", [False, True]) def test_unknown_settles_failed_in_current_store(tmp_path, authorized): - with ready(tmp_path / "coordination.sqlite3", origin=ExecutionOrigin.ACP) as store: + with ready(tmp_path / "coordination.sqlite3", origin=AcpOrigin) as store: _, fence = started(store) if authorized: store.observe_replay( @@ -101,7 +101,7 @@ def test_monitor_evidence_preserves_its_outcome_and_owned_replay_contribution(ou @pytest.mark.parametrize("predecessor", [AttemptFailedAttempt, UnknownOutcomeAttempt]) def test_sql_retry_predecessor_is_not_every_unsuccessful_outcome(tmp_path, predecessor): - with ready(tmp_path / "coordination.sqlite3", origin=ExecutionOrigin.ACP) as store: + with ready(tmp_path / "coordination.sqlite3", origin=AcpOrigin) as store: _, fence = started(store) store.observe_replay( fence, expected_pointer_revision=1, expected_replay_revision=None, diff --git a/tests/test_native_failure_recovery.py b/tests/test_native_failure_recovery.py index f24abd6dcf..e50236a9ad 100644 --- a/tests/test_native_failure_recovery.py +++ b/tests/test_native_failure_recovery.py @@ -19,7 +19,7 @@ from agent_comms.attempt_recovery import RecoveryMonitorCapability from agent_comms.comms import Comms from agent_comms.coordination_errors import IdentityConflict, RecoveryBlocked -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, WireOrigin from agent_comms.coordinator import Coordination from agent_comms.cursor_publication import CursorDelivery from agent_comms.declared_family import DeclaredFamily @@ -130,7 +130,7 @@ async def release_native_input(directory: Path, history: type[NativeSendHistory] ) as participant: assert participant is not None created = store.create( - participant.batch.execution_id, ExecutionOrigin.WIRE, + participant.batch.execution_id, WireOrigin, participant.lookup, participant.owner.thread.name, 1, sources=participant.batch.sources, ).value diff --git a/tests/test_native_unknown_recovery.py b/tests/test_native_unknown_recovery.py index 694c45c7c7..230aed43dc 100644 --- a/tests/test_native_unknown_recovery.py +++ b/tests/test_native_unknown_recovery.py @@ -30,7 +30,7 @@ from agent_comms.comms import Comms from agent_comms.coordination_errors import RecoveryBlocked from agent_comms.coordination_tables.attempts import ReplayFact -from agent_comms.coordination_tables.executions import CurrentExecutions, ExecutionOrigin +from agent_comms.coordination_tables.executions import CurrentExecutions, ExecutionProvenance, WireOrigin from agent_comms.cursor_publication import CursorDelivery from agent_comms.native_input_record import FullNativeExecution, NumberNativeColumn, TextNativeColumn from agent_comms.selected_participant import SelectedParticipant @@ -55,7 +55,7 @@ async def release_starting_attempt(directory): ) as participant: assert participant is not None created = store.create( - participant.batch.execution_id, ExecutionOrigin.WIRE, + participant.batch.execution_id, WireOrigin, participant.lookup, participant.owner.thread.name, 1, sources=participant.batch.sources, ).value diff --git a/tests/test_optional_awareness_projection.py b/tests/test_optional_awareness_projection.py index d159ca7171..d2a822cbd9 100644 --- a/tests/test_optional_awareness_projection.py +++ b/tests/test_optional_awareness_projection.py @@ -24,7 +24,7 @@ from agent_comms.coordination_tables.assignments import WakeAssignment from agent_comms.assignment_states import EngagedAssignment from agent_comms.wake_policy import FullWake -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, WireOrigin from agent_comms.coordinator import Coordination from agent_comms.optional_awareness_projection import OptionalAwarenessProjection from agent_comms.registration import Registration @@ -115,7 +115,7 @@ def test_selected_decision_and_open_obligation_are_both_source_cited(tmp_path: P owner = _owner(comms, "member000") store.create( "reply000", - ExecutionOrigin.WIRE, + WireOrigin, assignment.recipient_lookup, owner.name, 1, @@ -169,7 +169,7 @@ def test_open_obligation_budget_cannot_be_hidden_by_selected_cursor(tmp_path: Pa for number, (assignment, delivery) in enumerate(((old_claim, older), (new_claim, newer))): store.create( f"reply{number}", - ExecutionOrigin.WIRE, + WireOrigin, assignment.recipient_lookup, owner.name, 1, @@ -229,7 +229,7 @@ def test_fresh_gen2_selected_and_old_pending_obligation_are_scoped(tmp_path: Pat old, old_claim = _accepted(comms, store, root_id, "member000", "old pending") store.create( "old-reply", - ExecutionOrigin.WIRE, + WireOrigin, old_claim.recipient_lookup, "member000", 1, @@ -415,7 +415,7 @@ def test_saved_wire_awareness_preserves_passive_authority_and_rejects_incomplete owner = _owner(comms, "member000") store.create( "wire-awareness-reply", - ExecutionOrigin.WIRE, + WireOrigin, assignment.recipient_lookup, owner.name, 1, diff --git a/tests/test_s3_legality_behavior.py b/tests/test_s3_legality_behavior.py index f526d001ba..313f00a4fe 100644 --- a/tests/test_s3_legality_behavior.py +++ b/tests/test_s3_legality_behavior.py @@ -13,7 +13,7 @@ from agent_comms.coordination_snapshot import RecoverySnapshot from agent_comms.coordination_tables.attempts import ReplayFact from agent_comms.coordination_tables.executions import ReplayAssessments -from agent_comms.coordination_tables.executions import ExecutionOrigin +from agent_comms.coordination_tables.executions import ExecutionProvenance, AcpOrigin from agent_comms.coordination_tables.publications import ( PublicationIntents, PublicationEvidence, @@ -119,7 +119,7 @@ def test_execution_numeric_time_and_route_boundaries(pending): rejected(row, "must be positive", **{field: 0}) rejected(row, "timestamps", created_at_ms=-1) rejected(row, "timestamps", updated_at_ms=row.created_at_ms - 1) - for origin in ExecutionOrigin: + for origin in ExecutionProvenance.members_with(ExecutionProvenance): assert replace(row, origin=origin).origin is origin route = pending.obligations[0] rejected(route, "exact_target", exact_target=None) @@ -239,7 +239,7 @@ def test_snapshot_wire_claimless_and_exact_routes(pending): rejected( pending, "claimless snapshots", - execution=replace(pending.execution, origin=ExecutionOrigin.ACP), + execution=replace(pending.execution, origin=AcpOrigin), ) rejected( pending, From 9927295a85386f1a3c205b298a354c9338f611c1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:14:07 -0400 Subject: [PATCH 3/4] Finish typed policy contracts and derive sourced-origin SQL membership --- .../coordination_tables/executions.py | 44 +++++++++++++------ src/agent_comms/messages.py | 2 +- src/agent_comms/publisher.py | 2 +- src/agent_comms/response_conversation.py | 2 +- src/agent_comms/tools.py | 2 +- src/agent_comms/wake.py | 10 +++-- tests/bus_record_fixture.py | 2 +- tests/native_backend_fixture.py | 2 +- tests/test_backend.py | 2 +- tests/test_field_codec.py | 2 +- tests/test_message_categories.py | 2 +- tests/test_tracked_native_retry.py | 2 +- tests/test_turn_inbox.py | 2 +- tests/test_wake.py | 4 +- 14 files changed, 51 insertions(+), 29 deletions(-) diff --git a/src/agent_comms/coordination_tables/executions.py b/src/agent_comms/coordination_tables/executions.py index 8bdc95267f..9bccbfa1ba 100644 --- a/src/agent_comms/coordination_tables/executions.py +++ b/src/agent_comms/coordination_tables/executions.py @@ -5,7 +5,7 @@ import sqlite3 from abc import ABC, abstractmethod from dataclasses import dataclass, replace -from typing import Annotated, ClassVar +from typing import TYPE_CHECKING, Annotated, ClassVar from agent_comms.coordination_contracts import ( MAX_IDENTIFIER_CHARS, @@ -56,20 +56,28 @@ ) +if TYPE_CHECKING: + from agent_comms.coordination_tables.assignments import WakeAssignment + from agent_comms.coordination_tables.responses import ResponseObligation + from agent_comms.selected_source_batch import SelectedSource + + class ExecutionProvenance(DeclaredFamily, affix="Origin"): """The origin owns source membership, route eligibility and retry policy.""" @classmethod @abstractmethod - def require_sources(cls, sources) -> None: ... + def require_sources(cls, sources: tuple[SelectedSource, ...]) -> None: ... @classmethod @abstractmethod - def validate_routes(cls, assignments, obligations) -> None: ... + def validate_routes( + cls, assignments: tuple[WakeAssignment, ...], obligations: tuple[ResponseObligation, ...], + ) -> None: ... @classmethod @abstractmethod - def retry_authorized(cls, obligations) -> bool: ... + def retry_authorized(cls, obligations: tuple[ResponseObligation, ...]) -> bool: ... @classmethod @abstractmethod @@ -78,17 +86,19 @@ def require_obligation(cls) -> None: ... class ClaimlessOrigin(ExecutionProvenance, declared_name=None): @classmethod - def require_sources(cls, sources) -> None: + def require_sources(cls, sources: tuple[SelectedSource, ...]) -> None: if sources: raise IdentityConflict("claimless execution cannot bind claims") @classmethod - def validate_routes(cls, assignments, obligations) -> None: + def validate_routes( + cls, assignments: tuple[WakeAssignment, ...], obligations: tuple[ResponseObligation, ...], + ) -> None: if assignments or obligations: raise IntegrityViolationError("claimless snapshots cannot have claims or obligations") @classmethod - def retry_authorized(cls, obligations) -> bool: + def retry_authorized(cls, obligations: tuple[ResponseObligation, ...]) -> bool: return True @classmethod @@ -100,17 +110,19 @@ def require_obligation(cls) -> None: class WireOrigin(ExecutionProvenance): @classmethod - def require_sources(cls, sources) -> None: + def require_sources(cls, sources: tuple[SelectedSource, ...]) -> None: if not sources: raise IdentityConflict("wire execution requires ordered original sources") @classmethod - def validate_routes(cls, assignments, obligations) -> None: + def validate_routes( + cls, assignments: tuple[WakeAssignment, ...], obligations: tuple[ResponseObligation, ...], + ) -> None: if not assignments or not obligations: raise IntegrityViolationError("wire snapshots require original obligations") @classmethod - def retry_authorized(cls, obligations) -> bool: + def retry_authorized(cls, obligations: tuple[ResponseObligation, ...]) -> bool: return bool(obligations) and all(row.lifecycle.retryable for row in obligations) @classmethod @@ -119,12 +131,16 @@ def require_obligation(cls) -> None: @classmethod def matches_sql(cls, column: str) -> str: - return f"{column} = {sql_literal(cls)}" + members = cls.members_with(cls) + # Preserve the existing scalar DDL; additional sourced cases derive a set. + if len(members) == 1: + return f"{column} = {sql_literal(members[0])}" + return f"{column} IN {cls.sql_choices()}" @classmethod def retry_sql(cls, execution: str) -> str: return ( - f"({execution}.origin != {sql_literal(cls)} OR (EXISTS (SELECT 1 FROM obligations o\n" + f"(NOT ({cls.matches_sql(execution + '.origin')}) OR (EXISTS (SELECT 1 FROM obligations o\n" f" WHERE o.execution_id={execution}.execution_id) AND NOT EXISTS " "(SELECT 1 FROM obligations o\n" f" WHERE o.execution_id={execution}.execution_id AND o.retryable != 1)))" @@ -307,7 +323,9 @@ def schema_objects(cls, *, if_not_exists: bool = False): return super().schema_objects(if_not_exists=if_not_exists) | { cls.retry_view: f"CREATE VIEW {existence}{cls.retry_view} AS\n" + _RETRY_BASIS} - def retry_authorized(self, replay: ReplayState, obligations) -> bool: + def retry_authorized( + self, replay: ReplayState, obligations: tuple[ResponseObligation, ...], + ) -> bool: return ( self.lifecycle.has_retry_budget(self.max_attempts) and replay.allows_retry diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py index bcf45c3c1f..8d813a000e 100644 --- a/src/agent_comms/messages.py +++ b/src/agent_comms/messages.py @@ -163,7 +163,7 @@ def require_publishable(cls, message: Message, worktree: Path) -> None: class MembershipTransition(DeclaredFamily, affix="Membership"): @classmethod - def conversation_role(cls): + def conversation_role(cls) -> MembershipConversationRole: return MembershipConversationRole(cls) @classmethod diff --git a/src/agent_comms/publisher.py b/src/agent_comms/publisher.py index 3fa5b22cc6..5893ec6130 100644 --- a/src/agent_comms/publisher.py +++ b/src/agent_comms/publisher.py @@ -43,7 +43,7 @@ from .mentions import ThreadMention from .private_path import FileRevision from .registry_document import DEFAULT_ROSTER, RegistrySnapshot -from .wake import MessageControl, resolve_wake_cohort, OrdinaryControl +from .wake import OrdinaryControl, resolve_wake_cohort if TYPE_CHECKING: from .coordination_session import NativeBoundary diff --git a/src/agent_comms/response_conversation.py b/src/agent_comms/response_conversation.py index 9211725174..9d71c1183e 100644 --- a/src/agent_comms/response_conversation.py +++ b/src/agent_comms/response_conversation.py @@ -103,7 +103,7 @@ def audience(self, message: Message): def record(self, seq: int, message: Message, intent) -> KeyedPrivateRecord: from agent_comms.bus import KeyedPrivateRecord from agent_comms.delivery_policy import ResponseDeliveryPolicy - from agent_comms.wake import MessageControl, OrdinaryControl + from agent_comms.wake import OrdinaryControl stored = replace(message, seq=seq) audience = self.audience(stored) diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 48ad07477f..2e4e30d133 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -37,7 +37,7 @@ from . import cli as comms_cli from .change_report import NoChangeReport, ReportEvidence -from .channel_management import KeepThreadsTagDisposition, ListTagEdit, TagDisposition, TagEdit +from .channel_management import ListTagEdit, TagEdit from agent_comms.channel_targets import ChannelTarget, CommandTarget, NamedTarget from .channels import AnyChannel, Channel, SavedView, ViewMatch, ViewPredicate from .cli_commands import ( diff --git a/src/agent_comms/wake.py b/src/agent_comms/wake.py index e72397589d..1a71979ca1 100644 --- a/src/agent_comms/wake.py +++ b/src/agent_comms/wake.py @@ -11,14 +11,16 @@ from .channel_targets import CommandTarget, NamedTarget from .declared_family import DeclaredFamily from .messages import Message -from .wake_policy import NoWakeDecision, PassiveWake, RecipientDecision +from .wake_policy import NoWakeDecision, PassiveWake, RecipientDecision, WakePolicy class MessageControl(DeclaredFamily, affix="Control"): """Explicit system-control classification; not inferred from text or ACK type.""" @classmethod - def wake_mode(cls, target, recipient, mentions): + def wake_mode( + cls, target: NamedTarget, recipient: FrozenRecipient, mentions: frozenset[str], + ) -> WakePolicy: return target.wake_policy(recipient, mentions) @@ -28,7 +30,9 @@ class OrdinaryControl(MessageControl): class SystemControl(MessageControl, declared_name="system_control"): @classmethod - def wake_mode(cls, target, recipient, mentions): + def wake_mode( + cls, target: NamedTarget, recipient: FrozenRecipient, mentions: frozenset[str], + ) -> WakePolicy: return PassiveWake() diff --git a/tests/bus_record_fixture.py b/tests/bus_record_fixture.py index 9c9c7f006b..0215a59c7f 100644 --- a/tests/bus_record_fixture.py +++ b/tests/bus_record_fixture.py @@ -14,7 +14,7 @@ from agent_comms.response_acknowledgement import ( KeyedResponseAcknowledgement, canonical_publication_key, public_envelope_digest, ) -from agent_comms.wake import MessageControl, resolve_wake_cohort, OrdinaryControl +from agent_comms.wake import OrdinaryControl, resolve_wake_cohort publication_processes = ForkServerContext() diff --git a/tests/native_backend_fixture.py b/tests/native_backend_fixture.py index 3737fc4496..e30f6c8ea1 100644 --- a/tests/native_backend_fixture.py +++ b/tests/native_backend_fixture.py @@ -14,7 +14,7 @@ import agent_comms.agent_events as events import agent_comms.backend as backend -from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, BusyInputPolicy, SteerInputPolicy +from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, SteerInputPolicy from agent_comms.comms import Comms from agent_comms.fresh_private_session import create_fresh_private_session from agent_comms.child_process import BoundedRun, join_retirement diff --git a/tests/test_backend.py b/tests/test_backend.py index 515905c4e8..25961266a3 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -10,7 +10,7 @@ import agent_comms.agent_events as ae import agent_comms.backend as backend -from agent_comms.agent_backend import InputContent, InputId, InputRequest, BusyInputPolicy, SteerInputPolicy +from agent_comms.agent_backend import InputContent, InputId, InputRequest, SteerInputPolicy from agent_comms.native_pi import CAPABILITY from agent_comms.pi_rpc import PiRpcChannel from agent_comms.turn_phase import ( diff --git a/tests/test_field_codec.py b/tests/test_field_codec.py index 9b3e98af9d..410e262d66 100644 --- a/tests/test_field_codec.py +++ b/tests/test_field_codec.py @@ -29,7 +29,7 @@ WireNonNull, WireTransient, ) -from agent_comms.messages import ConversationRole, InfoKind, MembershipTransition, MembershipConversationRole, MembershipNoticeRole, Message, MessageRole, NoticeRole, JoinedMembership, LeftMembership +from agent_comms.messages import ConversationRole, InfoKind, MembershipConversationRole, MembershipNoticeRole, Message, MessageRole, NoticeRole, JoinedMembership, LeftMembership def test_transient_plan_excludes_fields_before_decoder_or_value_access(): diff --git a/tests/test_message_categories.py b/tests/test_message_categories.py index e5cf51f8d4..9db8e60af8 100644 --- a/tests/test_message_categories.py +++ b/tests/test_message_categories.py @@ -6,7 +6,7 @@ from agent_comms.bus_publication import Delivered from agent_comms.field_codec import FieldCodec -from agent_comms.messages import Message, MessageRole, InfoKind, ConversationRole, NoticeRole, HistoricalConversationRole, HistoricalNoticeRole, MembershipConversationRole, MembershipNoticeRole, MembershipTransition, JoinedMembership +from agent_comms.messages import Message, MessageRole, InfoKind, ConversationRole, NoticeRole, HistoricalConversationRole, HistoricalNoticeRole, MembershipConversationRole, MembershipNoticeRole, JoinedMembership from agent_comms.thread_identity import AgentThreadRole, UserThreadRole from agent_comms.transcript_events import IncomingTranscript from agent_comms.transcript_sources import AssignedTranscriptSource diff --git a/tests/test_tracked_native_retry.py b/tests/test_tracked_native_retry.py index 0b0cae2f36..8bc7629bd7 100644 --- a/tests/test_tracked_native_retry.py +++ b/tests/test_tracked_native_retry.py @@ -3,7 +3,7 @@ from agent_comms.native_arguments import SelectedModel from agent_comms.native_startup import NativeStartupPolicy -from agent_comms.agent_backend import BusyInputPolicy, SteerInputPolicy +from agent_comms.agent_backend import SteerInputPolicy from agent_comms.pi_events import NativeRetryEnd from agent_comms.pi_payloads import FailedProviderRetry, PiMessage, SucceededProviderRetry from agent_comms.tracked_turn import CompletedTrackedTerminal, PendingTrackedTerminal, TrackedTurnSession diff --git a/tests/test_turn_inbox.py b/tests/test_turn_inbox.py index 57f71e9241..006a22bbe5 100644 --- a/tests/test_turn_inbox.py +++ b/tests/test_turn_inbox.py @@ -2,7 +2,7 @@ -from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, BusyInputPolicy, FollowUpInputPolicy +from agent_comms.agent_backend import InputContent, InputId, InputRequest, TurnInbox, FollowUpInputPolicy from agent_comms.turn_inputs import InputForwarding diff --git a/tests/test_wake.py b/tests/test_wake.py index 1cdbc85462..fffd3183e2 100644 --- a/tests/test_wake.py +++ b/tests/test_wake.py @@ -49,7 +49,7 @@ def decide( message: Message, lookup: str = "lookup-alpha", members: tuple[FrozenRecipient, ...] = MEMBERS, - control: MessageControl = OrdinaryControl, + control: type[MessageControl] = OrdinaryControl, ) -> RecipientDecision | None: recipient = next( (member for member in members if member.recipient_lookup == lookup), @@ -155,7 +155,7 @@ def test_mixed_in_and_out_of_scope_mentions_only_name_valid_frozen_member() -> N @pytest.mark.parametrize("control", MessageControl.members_with(MessageControl)) @pytest.mark.parametrize("role", [NoticeRole(), MembershipConversationRole(JoinedMembership)]) def test_notice_and_membership_are_passive_regardless_of_control_or_sender( - role, control: MessageControl + role, control: type[MessageControl] ) -> None: message = replace(stored(sender_role=AgentThreadRole), role=role) assert decide(message, control=control) == WakeDecision( From fa87d28cd99337e5411d4796edba61adebdfaa1e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:17:21 -0400 Subject: [PATCH 4/4] Migrate membership assertions to their declaration-owned spelling --- tests/test_channels.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_channels.py b/tests/test_channels.py index e312a0f276..1f3204e02b 100644 --- a/tests/test_channels.py +++ b/tests/test_channels.py @@ -492,7 +492,7 @@ def test_membership_notices_follow_exact_membership_and_do_not_wake(tmp_path): ) comms.channels.update_tags("moving", add=frozenset({"api"})) joined = list(comms.views.history_page(CommandTarget.named("#api")).messages) - assert len(joined) == 1 and joined[0].role.change.value == "joined" + assert len(joined) == 1 and joined[0].role.change.declared_name == "joined" assert joined[0].body == "moving joined #api" and not joined[0].role.response_policy(joined[0]).starts_turn with comms.bus.opened_activity() as clocks: assert "moving" not in comms.views.last_sent_timestamps(clocks.sent.items(), registry=comms.registry.snapshot()) @@ -501,7 +501,7 @@ def test_membership_notices_follow_exact_membership_and_do_not_wake(tmp_path): assert len(list(comms.views.history_page(CommandTarget.named("#api")).messages)) == 2 comms.channels.update_tags("moving", remove=frozenset({"ui"})) history = list(wire(tmp_path).views.history_page(CommandTarget.named("#api")).messages) - assert [message.role.change.value for message in history] == ["joined", "left"] + assert [message.role.change.declared_name for message in history] == ["joined", "left"] assert all(not message.role.response_policy(message).starts_turn for message in history) assert not comms.views.viewer_snapshot(str(tmp_path)).participants("#api") comms.agents.begin_turn("peer", "active-turn")