Report Claude subagent activity and keep automatic session replies out of channels

src/agent_comms/activity.py

@@ -50,8 +50,7 @@ class ThreadOutput(ABC):
     def activity_thread(self) -> str: ...
 
     def enter(self, phase: TurnPhase) -> None:
-        self.agents.set_activity(self.activity_thread, phase.activity_state,
-                                 " ".join(phase.activity_detail.split()))
+        self.agents.enter_phase(self.activity_thread, phase)
 
 
 class ActivityState(Enum):

src/agent_comms/agent_activity.py

@@ -116,6 +116,10 @@ def set_activity(self, thread: str, state: ActivityState, detail: str = "") -> N
             raise RelationViolationError("Only permanent thread deletion retires its activity.")
         self._emit_activity(Activity.create(self.registry.require(thread).name, state, detail))
 
+    def enter_phase(self, thread: str, phase: TurnPhase) -> None:
+        """Native turns, bridges and seated subagents publish the phase's own activity."""
+        self.set_activity(thread, phase.activity_state, " ".join(phase.activity_detail.split()))
+
     def activity_of(self, thread: str, *, snapshot: RegistrySnapshot) -> ObservedActivity:
         participant = snapshot.require(thread)
         return self.all_activity((participant,), snapshot=snapshot)[participant.name]
@@ -153,7 +157,8 @@ def captured(self, threads: Iterable[Thread], *,
             return ActivityCapture.empty()
         events = self.activity.captured_for(
             requested,
-            active=frozenset(name for name, thread in requested.items() if thread.executing),
+            active=frozenset(name for name, thread in requested.items()
+                             if thread.execution.holds_activity(thread, snapshot.status(name))),
         )
         claims = self.registry.storage_rule.room_ledger.claims(
             self.registry.store.path.parent, requested.values())

src/agent_comms/channel_targets.py

@@ -292,6 +292,10 @@ def validate(self, message: Message) -> None: ...
     def reply_target(self, sender: str) -> str:
         return self.route.target(self.name, sender)
 
+    def automatic_reply_targets(self, sender: str) -> tuple[str, ...]:
+        """Shared destinations require an explicit send, never a session's final text."""
+        return ()
+
     def answered_sql(
         self, target: str, sender: str, snapshot: RegistryProvenance,
     ) -> tuple[str, tuple[str, ...]]:
@@ -353,6 +357,9 @@ def require_frozen_recipients(self, message: AudiencedMessage):
 
 
 class ThreadTarget(NamedTarget):
+    def automatic_reply_targets(self, sender: str) -> tuple[str, ...]:
+        return (self.reply_target(sender),)
+
     sigil = "@"
     route: ClassVar[ReplyRoute] = SenderReply()
     history_result_type = DmHistoryPageResult

src/agent_comms/harness_delivery_states.py

@@ -1,3 +1,3 @@
 """What one harness delivery attempt established: its outcome family.
 
 Kept below the coordinator tables in the import graph so the table that records
@@ -11,6 +11,7 @@
 from functools import partial
 from typing import TYPE_CHECKING
 
+from .channel_targets import CommandTarget
 from .declared_family import DeclaredFamily
 from .field_codec import TranscriptRecordId
 
@@ -115,7 +116,9 @@
             await bridge.record_mismatch(delivery, 'Observed parts do not equal the original message')
             return ()
         await bridge.finish(delivery, partial(replace, outcome=DeliveredOutcome, monitor=record))
-        return (message.exact_reply_target,)
+        return (CommandTarget.named(message.target).automatic_reply_targets(message.sender)
+                if message.role.response_policy(message).starts_turn_for(message, bridge.binding.bus)
+                else ())
 
 
 @dataclass(frozen=True)

src/agent_comms/thread_execution.py

@@ -133,6 +133,11 @@ def observe_recipient(cls, agents, snapshot: RegistrySnapshot, thread: Thread):
 
         return UnavailableRecipientActivity()
 
+    @classmethod
+    def holds_activity(cls, thread: Thread, status: ThreadStatus) -> bool:
+        """An executing native turn keeps its last activity current."""
+        return thread.executing
+
     @classmethod
     def project_activity(cls, thread: Thread, activity):
         """External participants report their own activity events."""
@@ -197,6 +202,11 @@ class SubagentExecution(ExternalThreadExecution):
     """A child working inside its parent session's turn: it reads its CLI inbox, holds no
     order and has no unit card of its own, and starts with its parent's audience."""
 
+    @classmethod
+    def holds_activity(cls, thread: Thread, status: ThreadStatus) -> bool:
+        """The occupied registration lasts until SubagentStop or SessionEnd retires it."""
+        return status.receiving
+
     @classmethod
     def registered_status(cls, thread: Thread, requested: ThreadStatus) -> ThreadStatus:
         return requested

src/agent_comms/tool_caller.py

@@ -15,6 +15,7 @@
 from .thread_execution import CommanderForkExecution
 from .thread_identity import ThreadIncarnation
 from .tool_result import ToolResult
+from .turn_phase import IdlePhase
 
 if TYPE_CHECKING:
     from .comms import Comms
@@ -200,6 +201,7 @@ def end(self, comms: Comms) -> None:
             return
         comms.registry.require(self.name).execution.require_subagent()
         self.leaving(comms)
+        comms.agents.enter_phase(self.name, IdlePhase())
         comms.registry.unregister(self.name)
         comms.threads.archive(self.name)
 

src/agent_comms_hooks/forks.py

@@ -24,8 +24,10 @@
 from typing import ClassVar
 
 from agent_comms.cli_commands import RegisteredResult
+from agent_comms.comms import client_wire
 from agent_comms.field_codec import FieldCodec
 from agent_comms.store_files import AtomicFile
+from agent_comms.turn_phase import ModelWaitPhase, TurnPhase
 from agent_comms_hooks import ClaudeHookProgram, SubagentId
 from agent_comms_hooks.watch import CommanderSession, WatchClaim, WatchRecord
 
@@ -59,24 +61,44 @@ def vacate(self, record: WatchRecord, seat: Path) -> tuple[str, ...]:
 
 
 @dataclass(frozen=True)
-class SubagentHook(SubagentSeats):
-    """What Claude Code adds for a subagent event; `transcript_path` stays the parent's."""
+class SeatedSubagent(SubagentSeats):
+    """The existing seat or commander-fork name identifies whose activity a hook reports."""
 
     agent_id: SubagentId
-    agent_type: str
 
     @property
     def seat(self) -> Path:
         return self.seats / self.agent_id
 
+    def enter(self, phase: TurnPhase) -> None:
+        state = WatchClaim(self.session_id).state()
+        state.by_role(
+            lambda: None,
+            lambda record: self.record_phase(record, self.seat.read_text(), phase),
+            lambda record: self.record_phase(record, self.agent_id.fork, phase),
+        )
+
+    @staticmethod
+    def record_phase(record: WatchRecord, name: str, phase: TurnPhase) -> None:
+        with client_wire(Path(record.root)) as comms:
+            comms.agents.enter_phase(name, phase)
+
+
+@dataclass(frozen=True)
+class SubagentHook(SeatedSubagent):
+    """What Claude Code adds for a subagent event; `transcript_path` stays the parent's."""
+
+    agent_type: str
+
     def commanding(self, verb: str) -> subprocess.CompletedProcess[str]:
         return self.run(*self.command, verb, "--fork", self.agent_id.fork)
 
     def register(self) -> None:
         state = WatchClaim(self.session_id).state()
-        state.by_role(lambda: None, lambda: self.claim(*state.bindings()), self.join)
+        state.by_role(lambda: None, self.claim, self.join)
+        self.enter(ModelWaitPhase())
 
-    def join(self) -> None:
+    def join(self, record: WatchRecord) -> None:
         run = self.commanding("join")
         if run.returncode:
             raise SystemExit(run.stderr.strip() or f"exit status {run.returncode}")
@@ -105,7 +127,7 @@ class SubagentStopHook(SubagentHook):
 
     def retire(self) -> None:
         state = WatchClaim(self.session_id).state()
-        state.by_role(lambda: None, lambda: self.leave(*state.bindings()), self.retire_fork)
+        state.by_role(lambda: None, self.leave, self.retire_fork)
 
     def leave(self, record: WatchRecord) -> None:
         if not self.seat.exists():
@@ -114,7 +136,7 @@ def leave(self, record: WatchRecord) -> None:
         for refusal in self.vacate(record, self.seat):
             raise SystemExit(refusal)
 
-    def retire_fork(self) -> None:
+    def retire_fork(self, record: WatchRecord) -> None:
         if self.stop_hook_active:
             self.end()
             return
@@ -142,15 +164,14 @@ def subagents(self) -> tuple[SubagentId, ...]:
     def end_subagents(self) -> tuple[str, ...]:
         """Core's refusals to end this session's children, or none."""
         state = WatchClaim(self.session_id).state()
-        return state.by_role(lambda: (), lambda: self.vacate_seats(*state.bindings()),
-                             self.end_forks)
+        return state.by_role(lambda: (), self.vacate_seats, self.end_forks)
 
     def vacate_seats(self, record: WatchRecord) -> tuple[str, ...]:
         """Every slot an occupant still holds: one with no stop left it seated."""
         seats = sorted(self.seats.glob("*")) if self.seats.is_dir() else []
         return tuple(refusal for seat in seats for refusal in self.vacate(record, seat))
 
-    def end_forks(self) -> tuple[str, ...]:
+    def end_forks(self, record: WatchRecord) -> tuple[str, ...]:
         """Core's one run for this session's forks, or none for a session with none."""
         subagents = self.subagents()
         if not subagents:

src/agent_comms_hooks/read_gate.py

@@ -5,9 +5,9 @@
 holds its own paraphrase of the rules and treats the order as optional. This gate makes the
 order the only path forward. PreToolUse only judges: it refuses any tool but Read while a part
 is unread. PostToolUse, which Claude Code sends only after a tool completes, records a Read that
-covered a whole dealt part; a failed or partial Read leaves the gate closed. Both read two files
-beside the transcript and nothing else, so no tool call pays a registry or bus read. A subagent
-runs on its parent's transcript and is never gated: the parent reads the parts.
+covered a whole dealt part; a failed or partial Read leaves the gate closed. Parent calls read
+only the dealt-part files. A subagent runs on its parent's transcript and is never gated: the
+parent reads the parts, while the subagent reports tool activity through its existing seat.
 """
 
 from __future__ import annotations
@@ -20,7 +20,9 @@
 from pathlib import Path
 
 from agent_comms.declared_family import DeclaredFamily
-from agent_comms_hooks import ClaudeHookProgram, ClaudeToolHook, HookRecord, SubagentId
+from agent_comms.turn_phase import ToolRunningPhase
+from agent_comms_hooks import ClaudeHookProgram, ClaudeToolHook, HookRecord
+from agent_comms_hooks.forks import SeatedSubagent
 
 
 @dataclass(frozen=True)
@@ -102,13 +104,11 @@ def judge(self) -> None:
 
 
 @dataclass(frozen=True)
-class SubagentCall(GatedCall):
-    """A subagent's call: its parent reads the parts, so it is never refused or recorded."""
-
-    agent_id: SubagentId
+class SubagentCall(GatedCall, SeatedSubagent):
+    """A subagent bypasses its parent's read gate but reports its own tool activity."""
 
     def judge(self) -> None:
-        return
+        self.enter(ToolRunningPhase(self.tool_name))
 
 
 class ReadGateProgram(ClaudeHookProgram):

src/agent_comms_hooks/skill.py

@@ -209,8 +209,8 @@ def take(self) -> tuple[str, ...]:
         """What this hook says: nothing to a session no bus binding names, the fleet's deck to
         a bound session, and the commander's deck to the one the registry names commander now."""
         return WatchClaim(self.session_id).state().by_role(
-            unbound=tuple, fleet=lambda: self.taken(FleetReader()),
-            commander=lambda: self.taken(CommanderReader()))
+            unbound=tuple, fleet=lambda record: self.taken(FleetReader()),
+            commander=lambda record: self.taken(CommanderReader()))
 
     def taken(self, reader: DeckReader) -> tuple[str, ...]:
         """A session start deals the deck."""

src/agent_comms_hooks/watch.py

@@ -242,8 +242,8 @@ def commands(self) -> bool:
         a binding can point at it."""
         return False
 
-    def by_role[T](self, unbound: Callable[[], T], fleet: Callable[[], T],
-                   commander: Callable[[], T]) -> T:
+    def by_role[T](self, unbound: Callable[[], T], fleet: Callable[[WatchRecord], T],
+                   commander: Callable[[WatchRecord], T]) -> T:
         """An unbound session has no role on any bus: a subagent, or a session never enrolled."""
         return unbound()
 
@@ -297,11 +297,10 @@ def commands(self) -> bool:
         return any(thread.incarnation == bound.incarnation
                    for thread in snapshot.commander_project.commanders(snapshot))
 
-    def by_role[T](self, unbound: Callable[[], T], fleet: Callable[[], T],
-                   commander: Callable[[], T]) -> T:
-        """A bound session follows the fleet rules; the one the registry names commander now
-        also follows the commander's."""
-        return commander() if self.commands() else fleet()
+    def by_role[T](self, unbound: Callable[[], T], fleet: Callable[[WatchRecord], T],
+                   commander: Callable[[WatchRecord], T]) -> T:
+        """A bound session supplies its own record to the selected role's operation."""
+        return commander(self.record) if self.commands() else fleet(self.record)
 
     def connect(self, serve: Callable[[], Serve],
                 request: Callable[[Serve], SessionRequest]) -> None:

tests/test_claude_automatic_reply.py

@@ -0,0 +1,50 @@
+"""A session's final answers a DM, never an outstanding channel mention."""
+
+import pytest
+
+from agent_comms.cli_bridges.claude_protocol import ClaudeSession, KnownTurn
+from agent_comms.harness_delivery_states import DeliveredOutcome
+from agent_comms.turn_item import NativeTurnId
+from transcript_record_fixture import transcript_record
+from test_monitor_delivery import attached_monitor, bridge, decoded_record, notification
+
+
+@pytest.mark.asyncio
+async def test_channel_mention_and_dm_share_a_turn_without_broadcasting_the_final(bridge):
+    comms = bridge.comms
+    comms.channels.create_tag('comms')
+    for name in ('claude-code', 'viewer', 'peer'):
+        comms.channels.update_tags(name, add=frozenset({'comms'}))
+    channel = comms.messaging.send('peer', '#comms', '@viewer please investigate').message
+    direct = comms.messaging.send('claude-code', 'viewer', 'please answer me privately').message
+    output = bridge.output_type.for_comms(comms.transcripts.external('viewer'), comms)
+    session = ClaudeSession(turn=KnownTurn(NativeTurnId('reply-turn')))
+    async with attached_monitor(bridge) as reader:
+        for message in (channel, direct):
+            await bridge.accept(message)
+            line = (await reader.readline()).decode().rstrip('\n')
+            output.publish_batch([
+                await update.observed(bridge) for update in notification(
+                    line, record_id=transcript_record(str(message.seq))).updates(session)
+            ])
+            delivered, = await bridge.attempts(message.message_id)
+            assert delivered.outcome is DeliveredOutcome
+
+    def replies():
+        return tuple((message.target, message.body) for message in comms.bus.bus.message_history()
+                     if message.sender == 'viewer' and message.seq > channel.seq)
+
+    final = decoded_record(dict(
+        type='assistant', uuid=transcript_record('final'), timestamp='2026-10-29T01:00:00Z',
+        message=dict(content=[dict(type='text', text='Private answer for Tristan')]),
+    ))
+    output.publish_batch(list(final.updates(session)))
+    assert replies() == ()
+    ended = decoded_record(dict(type='system', subtype='turn_duration'))
+    output.publish_batch(list(ended.updates(session)))
+    assert replies() == (('claude-code', 'Private answer for Tristan'),)
+    output.publish_batch(list(ended.updates(session)))
+    assert replies() == (('claude-code', 'Private answer for Tristan'),)
+    # A channel remains writable through the explicit send path.
+    comms.messaging.send('viewer', '#comms', 'explicit channel update')
+    assert replies()[-1] == ('#comms', 'explicit channel update')

tests/test_commander_forks.py

@@ -76,6 +76,8 @@
     UnlocatedProblemFinding,
 )
 from agent_comms.tool_caller import CommanderForkCaller, SubagentCaller
+from agent_comms.activity import ActivityState
+from agent_comms_hooks import read_gate
 from agent_comms.tool_declaration import ToolDeclaration
 from agent_comms.transcript_events import ToolStartTranscript
 from agent_comms.turn_item import NativeItemId, NativeTurnId
@@ -978,10 +980,15 @@ def test_a_subagent_start_registers_its_fork_and_a_resume_reuses_the_archived_th
     assert "--fork claude-a1" in told["additionalContext"]
     fork = comms.registry.require("claude-a1")
     assert fork.parent.named("lead") and comms.registry.status("claude-a1").receiving
+    assert comms.agents.activity.current(fork.name).state is ActivityState.THINKING
+    assert run(read_gate.main, "PreToolUse", tool_name="Bash", tool_input={"command": "pwd"}) == ""
+    activity = comms.agents.activity.current(fork.name)
+    assert (activity.state, activity.detail) == (ActivityState.WORKING, "Bash")
     found = tmp_path / "found.txt"
     found.write_text("The fork recorded its finding.")
     assert command(comms, "order", "say", "#comms", str(found), "--fork", "claude-a1").returncode == 0
     assert run(forks.stopped, "SubagentStop", stop_hook_active=False) == ""
+    assert comms.agents.activity.current(fork.name).state is ActivityState.IDLE
     run(forks.started, "SubagentStart")
     assert comms.registry.require("claude-a1").incarnation == fork.incarnation
     assert comms.registry.status("claude-a1").receiving
@@ -1039,9 +1046,9 @@ def test_a_unit_sessions_subagents_take_role_slots_and_an_unbound_sessions_take_
     monkeypatch.setenv("PYTHONPATH", str(Path(__file__).parents[1] / "src"))
     transcript = tmp_path / "unit.jsonl"
 
-    def run(program, event, agent_id):
+    def run(program, event, agent_id, **tool):
         record = {"session_id": "unit", "hook_event_name": event, "transcript_path": str(transcript),
-                  "agent_id": agent_id, "agent_type": "Explore", "stop_hook_active": False}
+                  "agent_id": agent_id, "agent_type": "Explore", "stop_hook_active": False, **tool}
         monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(record)))
         program()
         return capsys.readouterr().out
@@ -1055,6 +1062,15 @@ def run(program, event, agent_id):
     slot = comms.registry.require("worker-explore")
     assert slot.parent.named("worker") and comms.registry.status("worker-explore").receiving
     slot.execution.require_subagent()
+    assert comms.agents.activity.current(slot.name).state is ActivityState.THINKING
+    assert run(read_gate.main, "PreToolUse", "a2", tool_name="Read",
+               tool_input={"file_path": str(transcript)}) == ""
+    activity = comms.agents.activity.current(slot.name)
+    assert (activity.state, activity.detail) == (ActivityState.WORKING, "Read")
+    # A long-running tool stays working beyond the activity store's ordinary stale budget.
+    comms.agents.activity.emit(replace(activity, timestamp=1.))
+    snapshot = comms.registry.snapshot()
+    assert comms.agents.activity_of(slot.name, snapshot=snapshot).state is ActivityState.WORKING
     with pytest.raises(RelationViolationError, match="not a declared commander fork"):
         slot.execution.require_commander_fork()
     assert "claude-a2" not in comms.registry
@@ -1063,6 +1079,7 @@ def run(program, event, agent_id):
 
     assert run(forks.stopped, "SubagentStop", "a2") == ""
     assert isinstance(comms.registry.status("worker-explore"), ArchivedThreadStatus)
+    assert comms.agents.activity.current(slot.name).state is ActivityState.IDLE
     assert comms.registry.status("worker-explore-2").receiving
     run(forks.started, "SubagentStart", "a4")
     assert comms.registry.require("worker-explore").incarnation == slot.incarnation