Integration: kind families, thread-name owners, disclosure mixins, Pi event phases, phase evidence, commander package

plans/commander-native-plan.md

@@ -265,6 +265,23 @@ rebuildable projections follow their producing Core build.
 
 **Done when** the skill names no script.
 
+### Phase 10: fleet awareness (after the four plans)
+
+**Why.** On 2026-10-23 the bus was down for the installer route fix (#2633). Eight Codex threads were coordinated through `codex queue` and one handoff file, and the commander kept the map by hand-run scripts: a git overlap scan and a thread rotation reader. Every finding below came from those scripts or from Tristan noticing. Each finding is a fact Core can derive from git, its order records or turn records, so a commander that has to remember it is a defect. Tristan, 2026-10-23: "i can't wait to have all the infra together so i don't need to manually remember this shit". This phase starts after the virtual bus, multiple inheritance, harness and this plan's phase 9 are done. Its `PlanPhase` declaration in `plan_phases.py` is the first edit of its first order.
+
+**Changes.** Each line names the fact, its owner, and the 2026-10-23 incident it would have caught.
+- **One holder per file, refused at order time.** `CommsOrderTool` refuses an order whose scope's files overlap a file already held by another open order or by another open branch's diff. The refusal names both holders. Incident: two threads wrote the same route fix in `installer/backup.py` until Tristan reassigned it by hand.
+- **Drift and overlap as standing rules.** Two `StandingRule` members are derived from git on read. A held branch more than 30 commits behind `main`. Two open branches whose `src` diffs share a file. Each line names the holders and the files. Incident: `bus_tables.py` was edited on four branches at once, and five branches sat 66 to 90 commits behind `main`.
+- **Spinning as a `UnitState` member.** Derived from turn records: patches without a commit, test reruns with no source change between them, and minutes since the last push while holding an order. It extends phase 7's `UnitTime` instead of adding a second clock. Incident: the commander sampled tool calls per 15 minutes by hand to tell work from wheel-spinning.
+- **Unpushed work as a standing rule.** `comms_work_branches` gains local branches with no remote alongside unlinked remote branches, and a `StandingRule` member reports both. Incident: the lock fix and the projection producer fix lived only in local worktrees.
+- **Empty PRs as a standing rule.** An open PR with no commit ahead of `main` is a line whose disposition is closing it. Incident: #2618, #2619, #2621 and #2625.
+- **Review is triggered by delivery.** A delivery recorded at a commit (`comms_order_deliver`) enqueues phase 4's review of that commit. A reviewer never reviews a head nobody delivered. Incident: #2633 was reviewed and failed at 8af8f95, a head that did not yet contain the fix, and a rehearsal was ordered against the same stale head.
+- **Lost tools are a bus fact.** Each thread's turn records the tool set it was offered. A turn offered fewer tools than the previous one raises a fact for the commander, naming the missing tools and the runtime version. Incident: the Codex 0.164.0 upgrade removed `send_message_to_thread` and `read_thread` from every thread after restart, and only Tristan noticed.
+
+**Deletes:** the hand-run overlap and rotation scripts the commander keeps until this phase lands (`fleet_check.py` in the commander's scratch, outside `src`).
+
+**Done when** replaying the 2026-10-23 fleet state on a copy raises each incident above from Core alone, once each, with no script running.
+
 ## Parity gate (before phase 8)
 
 1. **Every row of the matrix** names a native owner that is on `main`.

src/agent_comms/acp.py

@@ -384,7 +384,7 @@ def select(store):
             with store.session.read():
                 participant = store.participants.get(lookup)
                 candidate = next_sealed_assignment(
-                    store, participant.lookup, owner.name, after_seq=after_seq,
+                    store, participant.lookup, after_seq=after_seq,
                 )
                 if candidate is None:
                     return False

src/agent_comms/acp_extension.py

@@ -370,7 +370,7 @@ def validate_scope(self, scope: CursorScope) -> None:
         cursor = self.cursor
         if (
             cursor.wire_root_id != scope.wire_root_id
-            or cursor.owner_thread != scope.claim_scope.incarnation.name
+            or cursor.recipient_lookup != scope.claim_scope.incarnation.lookup
             or cursor.owner_claimed_generation != scope.claimed_generation
             or not 0 <= cursor.injected_seq <= cursor.covered_seq
         ):

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/annotation_disclosure.py

@@ -0,0 +1,57 @@
+"""Disclosure policies, and the capabilities that give a message or segment its policy."""
+
+from __future__ import annotations
+
+import re
+from abc import abstractmethod
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from .declared_family import DeclaredFamily
+
+if TYPE_CHECKING:
+    from .turn_context import ContextSpan
+
+
+class DisclosureWithheld(ValueError):  # noqa: N818 - nominal refusal state
+    """The original segment's policy does not permit this text to leave."""
+
+
+class DisclosurePolicy(DeclaredFamily, affix="Disclosure"):
+    @abstractmethod
+    def require_text(self, text: str) -> str: ...
+
+
+@dataclass(frozen=True)
+class WithheldDisclosure(DisclosurePolicy):
+    def require_text(self, text: str) -> str:
+        raise DisclosureWithheld("This original segment does not permit third-party disclosure")
+
+
+@dataclass(frozen=True)
+class PublicInstructionDisclosure(DisclosurePolicy):
+    # Credential-shaped values are withheld as a whole. Replacing them would
+    # make the classifier label prose which is not the addressed original span.
+    credential = re.compile(
+        r"-----BEGIN [A-Z ]*PRIVATE KEY-----|\b(?:sk|ghp|github_pat)[-_][A-Za-z0-9_-]+"
+        r"|\b(?:api[_ -]?key|access[_ -]?token|password|secret)\s*[:=]\s*\S+"
+        r"|\bBearer\s+\S+", re.IGNORECASE)
+
+    def require_text(self, text: str) -> str:
+        if self.credential.search(text):
+            raise DisclosureWithheld("Credential-shaped original text cannot be disclosed")
+        return text
+
+
+class WithheldAnnotations:
+    """Source text cannot be disclosed to an annotation provider."""
+
+    def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy:
+        return WithheldDisclosure()
+
+
+class PublicInstructionAnnotations:
+    """Instruction text uses the existing credential-filtering disclosure policy."""
+
+    def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy:
+        return PublicInstructionDisclosure()

src/agent_comms/attempt_recovery.py

@@ -93,7 +93,7 @@ def observe_native_release(
             if attempt is None or not snapshot.is_current:
                 raise RecoveryBlocked("native recovery requires the current attempted execution")
             try:
-                release = comms.owners.releases.get(attempt.owner_thread)
+                release = comms.owners.releases.get(registry.by_lookup[attempt.owner_lookup].name)
             except (OSError, RelationViolationError, MalformedRecord) as error:
                 raise RecoveryBlocked("native owner release acknowledgement is invalid") from error
             try:
@@ -203,7 +203,7 @@ def settle(self, recovery_cls, store: Coordination, execution_id: str, owner) ->
         self.require_native_exit()
         PublicationAppendDispatches.resolve_retained(
             store, self.boundary.bus, self.boundary,
-            owner_lookup=owner.incarnation.lookup, owner_thread=owner.name,
+            owner_lookup=owner.incarnation.lookup,
             execution_id=execution_id,
         )
         pointer = CurrentExecutions.require(store.session._connection, owner_lookup=owner.incarnation.lookup)

src/agent_comms/attempt_start.py

@@ -27,7 +27,7 @@
 class AttemptStart:
     execution_id: str
     attempt_ordinal: int
-    owner_thread: str
+    owner_lookup: str
     owner_generation: int
     token: str
     expected_execution_revision: int
@@ -41,7 +41,7 @@ def fence(self, revision: int) -> OwnerFence:
         return OwnerFence(
             self.execution_id,
             self.attempt_ordinal,
-            self.owner_thread,
+            self.owner_lookup,
             self.owner_generation,
             revision,
             self.token,
@@ -52,13 +52,11 @@ def replayed_fence(self, snapshot: RecoverySnapshot, old: AttemptRecord) -> Owne
         if (
             old.attempt_ordinal,
             old.owner_lookup,
-            old.owner_thread,
             old.owner_generation,
             old.created_at_ms >= snapshot.execution.created_at_ms,
         ) != (
             self.attempt_ordinal,
-            snapshot.execution.owner_lookup,
-            self.owner_thread,
+            self.owner_lookup,
             self.owner_generation,
             True,
         ):
@@ -69,10 +67,10 @@ def validate(self, snapshot: RecoverySnapshot, participant: ParticipantSnapshot)
         execution = snapshot.execution
         ordinal = execution.lifecycle.next_attempt_ordinal
         if not participant.committed or (
-            participant.owner_identity.owner_thread,
+            participant.owner_identity.owner_lookup,
             participant.owner_identity.generation,
-        ) != (self.owner_thread, self.owner_generation):
-            raise StaleFence("owner generation or thread changed")
+        ) != (self.owner_lookup, self.owner_generation):
+            raise StaleFence("owner lookup or generation changed")
         if (execution.revision, participant.pointer.pointer_revision) != (
             self.expected_execution_revision,
             self.expected_pointer_revision,
@@ -99,7 +97,6 @@ def activate(self, session, snapshot: RecoverySnapshot) -> None:
             execution_id=self.execution_id,
             attempt_ordinal=self.attempt_ordinal,
             owner_lookup=execution.owner_lookup,
-            owner_thread=self.owner_thread,
             owner_generation=self.owner_generation,
             owner_token_digest=self.digest,
             lifecycle=PromptStartingAttempt(lease),

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