Remove rank-based authority and share the rule reader

src/agent_comms/channel_targets.py

@@ -704,10 +704,4 @@ class CommsTag(DeclaredTag):
 
 
 class CommandersTag(DeclaredTag):
-    """The commanders of this bus: a member acts under its own name, holds the commander role
-    in its hooks and receives the heartbeat board. A role, not an audience: a thread carries
-    it only when it was itself declared a commander, never because its parent was."""
-
-    @classmethod
-    def inherited(cls) -> bool:
-        return False
+    """A coordination channel identity, never a permission or execution role."""

src/agent_comms/cli_bridges/hook_deferral.py

@@ -174,7 +174,7 @@ class ReadIdle:
 
     @classmethod
     def read(cls, comms: Comms, binding: Binding) -> Self:
-        return cls(tuple(unit.name for unit in UnownedWheelSpin.read_idle(comms, binding.owner)))
+        return cls(tuple(unit.name for unit in UnownedWheelSpin.read_idle(comms)))
 
     def reasons(self, deferred: tuple[str, ...]) -> Iterator[str]:
         if self.units:
@@ -213,15 +213,13 @@ def guard(self) -> None:
         if self.stop_hook_active:
             return
         snapshot = self.comms.registry.snapshot()
-        selected = snapshot.commander_project.by_role(
-            snapshot.born(self.binding.owner.require_recorded().lookup),
-            commander=self.deferred, commanded=lambda: ())
+        snapshot.born(self.binding.owner.require_recorded().lookup)
         deaf = tuple(watch.line() for watch in WatchClaim(self.session_id).deaf())
-        if reasons := (*deaf, *selected()):
+        if reasons := (*deaf, *self.deferred()):
             self.output.line(json.dumps({"decision": "block", "reason": "\n".join(reasons)}))
 
     def deferred(self) -> tuple[str, ...]:
-        """Only a commander answers for deferring work; every bound role must keep its Monitor."""
+        """Every bound collaborator answers for its own deferred work."""
         reply = self.reply()
         return (*(
             reason

src/agent_comms/cli_bridges/hook_forks.py

@@ -1,4 +1,4 @@
-"""Claude subagents operate on their resident bridge's bound bus and actor."""
+"""Genuine in-turn subagents act through their resident session's bound parent."""
 from __future__ import annotations
 
 import json
@@ -7,100 +7,45 @@
 
 from agent_comms_hooks import ClaudeSubagentHook, ClaudeTranscriptHook, SubagentId, ToolRefusedError
 
-from .session_watch import WatchClaim
 from .shared import Binding
 from ..comms import Comms
-from ..store_files import AtomicFile
-from ..thread_identity import ThreadIncarnation
-from ..thread_text import SubagentRole
-from ..tool_caller import CommanderForkCaller, SubagentCaller
+from ..tool_caller import CommanderForkCaller
 from ..turn_phase import ModelWaitPhase, TurnPhase
 
 
-class SubagentSeats(ClaudeTranscriptHook):
-    """One occupant-to-slot pairing per seat; registry/Caller owners perform the effects."""
-
-    @property
-    def seats(self) -> Path:
-        return WatchClaim(self.session_id).path.with_suffix(".slots")
-
-    def vacate(self, comms: Comms, seat: Path) -> tuple[str, ...]:
-        try:
-            SubagentCaller(seat.read_text()).end(comms)
-        except ToolRefusedError as refused:
-            return (str(refused),)
-        seat.unlink()
-        return ()
-
-
 @dataclass(frozen=True)
-class SeatedSubagent(ClaudeSubagentHook, SubagentSeats):
-    """The declared role chooses the existing seat or commander-fork activity identity."""
-
-    @property
-    def seat(self) -> Path:
-        return self.seats / self.agent_id
+class SubagentActivity(ClaudeSubagentHook):
+    """The actual SDK subagent id owns its activity; parent fork provenance is not a rank."""
 
     def enter(self, comms: Comms, binding: Binding, phase: TurnPhase) -> None:
-        snapshot = comms.registry.snapshot()
-        selected = snapshot.commander_project.by_role(
-            snapshot.born(binding.owner.require_recorded().lookup),
-            commander=lambda: self.agent_id.fork, commanded=self.seat.read_text)
-        comms.agents.enter_phase(selected(), phase)
+        comms.registry.snapshot().born(binding.owner.require_recorded().lookup)
+        comms.agents.enter_phase(self.agent_id.fork, phase)
 
 
 @dataclass(frozen=True)
-class SubagentHook(SeatedSubagent):
-    """Claude's subagent event retains its parent's transcript and actual binding."""
+class SubagentHook(SubagentActivity):
+    """An SDK subagent retains its parent's transcript and original bound incarnation."""
 
     agent_type: str
 
     def register(self, comms: Comms, binding: Binding) -> None:
-        snapshot = comms.registry.snapshot()
-        actor = snapshot.born(binding.owner.require_recorded().lookup)
-        selected = snapshot.commander_project.by_role(
-            actor, commander=self.join, commanded=self.claim)
-        selected(comms, actor.incarnation)
-        self.enter(comms, binding, ModelWaitPhase())
-
-    def join(self, comms: Comms, actor: ThreadIncarnation) -> None:
+        actor = comms.registry.snapshot().born(binding.owner.require_recorded().lookup)
         fork = self.agent_id.fork
         CommanderForkCaller(fork).join(comms, actor.name)
-        self.warn(f"Core registered you as commander fork {fork}: pass `--fork {fork}`"
-                  " on every agent-comms-commander verb; your stop is refused until Core holds"
-                  " a record of what you found.")
-
-    def claim(self, comms: Comms, actor: ThreadIncarnation) -> None:
-        slot = comms.registry.claim_slot(
-            actor.name, SubagentRole(self.agent_type), comms.goals).name
-        self.seats.mkdir(parents=True, exist_ok=True, mode=0o700)
-        AtomicFile(self.seat).write_text(slot)
-        self.warn(f"Core seated you in slot {slot}, a child of {actor.name}: messages to {slot}"
-                  f" are for this role (`agent-comms inbox --name {slot}`), and its history is"
-                  " what earlier occupants of the role did.")
+        self.warn(f"Core registered you as in-turn fork {fork}: pass `--fork {fork}`"
+                  " on every agent-comms-commander verb; the action keeps your parent authority"
+                  " and records your own incarnation. Record what you found before stopping.")
+        self.enter(comms, binding, ModelWaitPhase())
 
 
 @dataclass(frozen=True)
 class SubagentStopHook(SubagentHook):
-    """The first empty stop is refused; an already refused stop always ends its fork."""
+    """The first empty stop is refused; an already refused stop ends the same actual fork."""
 
     stop_hook_active: bool
 
     def retire(self, comms: Comms, binding: Binding) -> None:
-        snapshot = comms.registry.snapshot()
-        selected = snapshot.commander_project.by_role(
-            snapshot.born(binding.owner.require_recorded().lookup),
-            commander=self.retire_fork, commanded=self.leave)
-        selected(comms)
-
-    def leave(self, comms: Comms) -> None:
-        if not self.seat.exists():
-            raise ToolRefusedError(f"Subagent {self.agent_id} stopped without a seat: SubagentStart"
-                                   " never ran for it under this session's bus binding.")
-        for refusal in self.vacate(comms, self.seat):
-            raise ToolRefusedError(refusal)
-
-    def retire_fork(self, comms: Comms) -> None:
+        comms.registry.snapshot().born(binding.owner.require_recorded().lookup)
         actor = CommanderForkCaller(self.agent_id.fork)
         if self.stop_hook_active:
             actor.end(comms)
@@ -111,26 +56,16 @@ def retire_fork(self, comms: Comms) -> None:
             self.output.line(json.dumps({"decision": "block", "reason": str(refused)}))
 
 
-class SessionSubagents(SubagentSeats):
-    """A session's remaining children end through their existing lifecycle owners."""
+class SessionSubagents(ClaudeTranscriptHook):
+    """Remaining SDK children retire through their existing caller/lifecycle owner."""
 
     def subagents(self) -> tuple[SubagentId, ...]:
         directory = Path(self.transcript_path).with_suffix("") / "subagents"
         return tuple(SubagentId(path.stem.removeprefix("agent-"))
                      for path in sorted(directory.glob("agent-*.jsonl")))
 
     def end_subagents(self, comms: Comms, binding: Binding) -> tuple[str, ...]:
-        snapshot = comms.registry.snapshot()
-        selected = snapshot.commander_project.by_role(
-            snapshot.born(binding.owner.require_recorded().lookup),
-            commander=self.end_forks, commanded=self.vacate_seats)
-        return selected(comms)
-
-    def vacate_seats(self, comms: Comms) -> tuple[str, ...]:
-        seats = sorted(self.seats.glob("*")) if self.seats.is_dir() else []
-        return tuple(refusal for seat in seats for refusal in self.vacate(comms, seat))
-
-    def end_forks(self, comms: Comms) -> tuple[str, ...]:
+        comms.registry.snapshot().born(binding.owner.require_recorded().lookup)
         try:
             CommanderForkCaller.end_session(
                 comms, tuple(CommanderForkCaller(agent.fork) for agent in self.subagents()))

src/agent_comms/cli_bridges/hook_prime.py

@@ -30,7 +30,7 @@ def prime_threads(prime: type[Prime], names: Sequence[str]) -> None:
         for name in names:
             thread = comms.registry.require(name)
             thread.execution.require_native()
-            PiRedeal.of(comms, thread).deal_from(prime.commander, prime.fleet, prime.trigger())
+            PiRedeal.of(comms, thread).deal_from(prime.reader, prime.trigger())
             print(f"{name}: dealt {prime.__name__}")
 
 

src/agent_comms/cli_bridges/hook_read_gate.py

@@ -6,7 +6,7 @@
 only after a tool completes, records a Read that covered a whole dealt file; a failed or partial
 Read leaves the gate closed.
 A subagent is gated as its own session: its deal, reads and audit sit beside its own transcript,
-it may not write the main agent's audit, and it reports its tool activity through its seat.
+it may not write the main agent's audit, and its actual SDK id owns its tool activity.
 """
 
 from __future__ import annotations
@@ -31,7 +31,7 @@
 from agent_comms.channel_targets import ThreadTarget
 from agent_comms.task_attachments.owner_notices import MeantNotice
 from .claude_transcript import ClaudeJsonLines
-from .hook_forks import SeatedSubagent
+from .hook_forks import SubagentActivity
 
 
 @dataclass(frozen=True)
@@ -213,7 +213,7 @@ def say(self, line: str, deal: RuleDeal) -> Asked:
 
 
 @dataclass(frozen=True)
-class SubagentCall(GatedCall, SeatedSubagent):
+class SubagentCall(GatedCall, SubagentActivity):
     """A subagent shares its session's id and transcript path, but its deal is its own: it
     lives beside the subagent's own transcript, so its reads, its audit and its calls never
     touch the main agent's, and it reports its own tool activity."""

src/agent_comms/cli_bridges/session_watch.py

@@ -231,12 +231,9 @@ def bindings(self) -> tuple[WatchRecord, ...]:
         return (self.record,)
 
     def commands(self) -> bool:
-        """A saved enrollment may predate registration; only a registered recipient has a role."""
+        """Enrollment may predate registration; tags never decide a bound peer's authority."""
         snapshot = RegistryStore.at(Path(self.record.root)).snapshot()
-        if not snapshot.taken(self.record.bus):
-            return False
-        return snapshot.commander_project.by_role(
-            snapshot.require(self.record.bus), commander=True, commanded=False)
+        return snapshot.taken(self.record.bus)
 
     def connect(self, serve: Callable[[], agent_comms.cli_bridges.shared.Serve],
                 request: Callable[[agent_comms.cli_bridges.shared.Serve], agent_comms.cli_bridges.monitor.SessionRequest]) -> None:

src/agent_comms/cli_bridges/shared.py

@@ -531,28 +531,28 @@ async def execute(self, bridge: OperationBridge) -> None:
         await bridge.registered_watch(self)
 
 
-class BindingCommander(SubclassUnion, TextChoice, affix="BindingCommander"):
-    """Who commands a newly bound session, or an independent user's session."""
+class BindingChannels(SubclassUnion, TextChoice, affix="BindingChannels"):
+    """Optional initial channel source for a new independent external endpoint."""
 
     @abstractmethod
     async def register(self, bridge: CliBridge) -> None: ...
 
 
 @dataclass(frozen=True, slots=True)
-class CommandedBy(GivenText, BindingCommander):
+class ChannelsFrom(GivenText, BindingChannels):
     text: str
 
     async def register(self, bridge: CliBridge) -> None:
         worktree = await bridge.session_worktree()
         await bridge.comms.bus.when_open(partial(
-            bridge.comms.registry.register_commanded_unit,
+            bridge.comms.registry.register_bound_thread,
             bridge.address.bus, self.text, worktree, bridge.session_process(),
         ))
 
 
 @dataclass(frozen=True, slots=True)
-class Uncommanded(AbsentText, BindingCommander):
-    """A user's own session has no commanded audience or fork parent."""
+class NoChannelSource(AbsentText, BindingChannels):
+    """An independent endpoint starts without copied channel membership."""
 
     async def register(self, bridge: CliBridge) -> None:
         thread = Thread(
@@ -566,9 +566,10 @@ async def register(self, bridge: CliBridge) -> None:
 class Watch(Inbox):
     help = "Deliver this binding's bus messages and publish its harness transcript to the bus"
 
+    # Retained external CLI/record spelling; this supplies channels, never authority.
     commander: Annotated[
-        BindingCommander, field_codec.Argument(flags=("--commander",))
-    ] = Uncommanded()
+        BindingChannels, field_codec.Argument(flags=("--commander",))
+    ] = NoChannelSource()
 
     @classmethod
     def serves(cls, bridge: type[CliBridge]) -> bool:

src/agent_comms/cli_verbs/commander.py

@@ -1,49 +1,31 @@
 from __future__ import annotations
 
-from agent_comms.channel_targets import CommandersTag
 from agent_comms.cli_command import CliCommand
 from agent_comms.cli_commands import CommandersResult
-from agent_comms.cli_commands import _tags
 from agent_comms.commander.config import CommanderChoice
 from agent_comms.commander.config import KeptCommander
 from agent_comms.commander.config import KeptRepos
 from agent_comms.comms import Comms
-from agent_comms.errors import RelationViolationError
-from agent_comms.nominal.field_codec import Argument, Normalize
+from agent_comms.nominal.field_codec import Argument
 from dataclasses import dataclass
 from typing import Annotated
 
 
 @dataclass(frozen=True, kw_only=True)
 class CommanderCliCommand(CliCommand):
-    """Assign this bus's commanders at runtime, through the same owner install's --commander
-    uses: the lead is named on the registry's commander project, keeping its repositories,
-    and a non-lead commander leaves by dropping its #commanders tag."""
+    """Select the board/repository storage lead, not a rank over collaborators."""
 
-    help = "Name the lead commander or remove a commander at runtime"
+    help = "Name the storage lead and list independent collaborators"
     lead: Annotated[
         CommanderChoice,
         Argument(flags=("--lead",), options={"help": "Record this thread as the lead commander"}),
     ] = KeptCommander()
-    remove: Annotated[
-        frozenset[str],
-        Normalize(_tags),
-        Argument(flags=("--remove",), options={"help": "Remove these threads as commanders, separated by commas"}),
-    ] = frozenset()
 
     def apply(self, ctx: Comms) -> CommandersResult:
         snapshot = ctx.registry.snapshot()
         project = self.lead.configured(snapshot.commander_project, KeptRepos(), snapshot)
-        leads = frozenset(thread.incarnation for thread in project.leads(snapshot))
-        removed = tuple(snapshot.require(name) for name in self.remove)
-        for thread in removed:
-            if thread.incarnation in leads:
-                raise RelationViolationError(
-                    f"{thread.name} is the lead commander; name another lead with --lead")
         if project != snapshot.commander_project:
             ctx.registry.set_commander_project(project)
-        for thread in removed:
-            ctx.channels.update_tags(thread.name, remove=frozenset({CommandersTag.declared_name}))
         snapshot = ctx.registry.snapshot()
         project = snapshot.commander_project
         return CommandersResult(

src/agent_comms/cli_verbs/merge_review.py

@@ -7,7 +7,7 @@
 from agent_comms.cli_command import CliCommand
 from agent_comms.cli_commands import MergeReviewResult
 from agent_comms.comms import Comms
-from agent_comms.nominal.field_codec import Argument
+from agent_comms.nominal.field_codec import Argument, MalformedRecord
 from agent_comms.head_review import PullRequestHead
 from agent_comms.repository_name import RepositoryName
 from agent_comms.registry_document import RegistrySnapshot
@@ -22,22 +22,23 @@
 class ReviewedMerge:
     """The CLI and resident hook share exact-head authorization, not a subprocess protocol."""
 
-    def require_actor(self, snapshot: RegistrySnapshot, actor: RecordedThreadIncarnation) -> None:
-        if not snapshot.commander_project.by_role(
-                snapshot.born(actor.lookup), commander=True, commanded=False):
-            raise ToolRefusedError(
-                "Only a bound commander session may use a reviewed merge bypass.")
-
     def reviewed_merges(self, ctx: Comms, snapshot: RegistrySnapshot,
                         actor: RecordedThreadIncarnation,
                         bypasses: tuple[QueueBypass, ...]) -> MergeReviewResult:
-        self.require_actor(snapshot, actor)
+        project = snapshot.commander_project
+        project.require_actor(snapshot, actor)
         heads = []
         for bypass in bypasses:
-            for owed in bypass.owes_verdict(bypass.target()):
-                request = PullRequestHead(RepositoryName.from_wire(owed.repository), owed.number, owed.head)
-                StoredHeadReview.current_verdict(ctx.goals.todos, request).require_ok(request)
-                heads.append(request)
+            target = bypass.target()
+            try:
+                request = PullRequestHead(
+                    RepositoryName.from_wire(target.repository), target.number, target.head)
+            except MalformedRecord as error:
+                raise ToolRefusedError(f"Invalid merge head: {error}") from error
+            project.require_scope(snapshot, actor, request.repo)
+            for owed in bypass.owes_verdict(request):
+                StoredHeadReview.current_verdict(ctx.goals.todos, owed).require_ok(owed)
+                heads.append(owed)
         return MergeReviewResult(tuple(heads))
 
 
@@ -58,5 +59,5 @@ def apply(self, ctx: Comms) -> MergeReviewResult:
         actor, *others = (snapshot.require(binding.bus).incarnation.require_recorded()
                           for binding in bindings)
         for other in others:
-            self.require_actor(snapshot, other)
+            snapshot.commander_project.require_actor(snapshot, other)
         return self.reviewed_merges(ctx, snapshot, actor, QueueBypassGuard.bypasses(self.bash))