KindFamilies: put kind behavior and spelling on declared cases

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

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

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

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,

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: