diff --git a/src/agent_comms/activity.py b/src/agent_comms/activity.py index 76f0a4b6e2..845b2b305f 100644 --- a/src/agent_comms/activity.py +++ b/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): diff --git a/src/agent_comms/agent_activity.py b/src/agent_comms/agent_activity.py index 45e1eecf49..92daf5aa68 100644 --- a/src/agent_comms/agent_activity.py +++ b/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()) diff --git a/src/agent_comms/channel_targets.py b/src/agent_comms/channel_targets.py index d07fd1600f..fdea37d61a 100644 --- a/src/agent_comms/channel_targets.py +++ b/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 diff --git a/src/agent_comms/harness_delivery_states.py b/src/agent_comms/harness_delivery_states.py index b5732647c4..079388dcac 100644 --- a/src/agent_comms/harness_delivery_states.py +++ b/src/agent_comms/harness_delivery_states.py @@ -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 @@ async def confirm(self, bridge: LineMonitorBridge, delivery: HarnessDelivery, 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) diff --git a/src/agent_comms/thread_execution.py b/src/agent_comms/thread_execution.py index fb3afb6155..941b02f546 100644 --- a/src/agent_comms/thread_execution.py +++ b/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 diff --git a/src/agent_comms/tool_caller.py b/src/agent_comms/tool_caller.py index f212f06647..451c08fa7d 100644 --- a/src/agent_comms/tool_caller.py +++ b/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) diff --git a/src/agent_comms_hooks/forks.py b/src/agent_comms_hooks/forks.py index 188e4ac1f6..c8b50490eb 100644 --- a/src/agent_comms_hooks/forks.py +++ b/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: diff --git a/src/agent_comms_hooks/read_gate.py b/src/agent_comms_hooks/read_gate.py index 2c34bdeb2c..29e9ebbac8 100644 --- a/src/agent_comms_hooks/read_gate.py +++ b/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): diff --git a/src/agent_comms_hooks/skill.py b/src/agent_comms_hooks/skill.py index 606a10fd85..5904bbcc69 100644 --- a/src/agent_comms_hooks/skill.py +++ b/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.""" diff --git a/src/agent_comms_hooks/watch.py b/src/agent_comms_hooks/watch.py index f4ffaec9a1..efdc0c3b1a 100644 --- a/src/agent_comms_hooks/watch.py +++ b/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: diff --git a/tests/test_claude_automatic_reply.py b/tests/test_claude_automatic_reply.py new file mode 100644 index 0000000000..62d20f7be5 --- /dev/null +++ b/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') diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py index 156764a1aa..b89e4bd36f 100644 --- a/tests/test_commander_forks.py +++ b/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 diff --git a/tests/test_commander_skill_hook.py b/tests/test_commander_skill_hook.py index b60ab91274..bc4740d5a0 100644 --- a/tests/test_commander_skill_hook.py +++ b/tests/test_commander_skill_hook.py @@ -485,6 +485,11 @@ def test_a_subagent_and_an_unbound_session_are_never_gated( transcript = tmp_path / "session.jsonl" transcript.write_text("") assert DeckStart.decode(hook_record(transcript, "SessionStart")).take() + # Claude's SubagentStart registers the child before PreToolUse reports its activity. + from agent_comms.thread_execution import CommanderForkExecution + + commanding_session.registry.register_child( + "claude-a9", "claude-code", CommanderForkExecution, commanding_session.goals) assert not refused(capsys, gated(transcript, "Bash", {"command": "ls"}, agent_id="a9")) elsewhere = tmp_path / "other.jsonl" elsewhere.write_text("")