comms mute: a member can silence their own wake without leaving the group chat

src/agent_comms/cli_verbs/mute.py

@@ -0,0 +1,45 @@
+"""Silence or restore the caller's own wake in the group chat for a bounded time.
+
+Wake-all stays the default: a post pings every member. This verb only lets a member
+opt their OWN wake out for a while; they still receive every message, they are just not
+woken to triage it until the mute expires or they clear it.
+"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Annotated
+
+from agent_comms.cli_verbs.cli_command import CliCommand
+from agent_comms.cli_verbs.cli_result import CliResult
+from agent_comms.comms import Comms
+from agent_comms.nominal.field_codec import Argument, FieldCodec, Normalize
+from agent_comms.transport.tool_caller import ThreadCaller
+from agent_comms.waking.mute_ledger import MuteLedger
+
+
+@dataclass(frozen=True)
+class MutedResult(CliResult):
+    muter: str
+    muted_until: float
+
+
+@dataclass(frozen=True, kw_only=True)
+class MuteCliCommand(CliCommand, declared_name="mute"):
+    help = ("Silence your own wake in the group chat for --minutes; you still receive every "
+            "message, you are just not woken to triage it. --clear restores it.")
+    caller: Annotated[ThreadCaller, Normalize(FieldCodec.encode),
+                      Argument(flags=("--from",), options={"type": ThreadCaller})]
+    minutes: Annotated[float, Argument(flags=("--minutes",))] = 60.0
+    clear: Annotated[bool, Argument(flags=("--clear",))] = False
+
+    def apply(self, ctx: Comms) -> MutedResult:
+        thread = ctx.registry.require(self.caller.own_thread())
+        lookup = thread.incarnation.lookup
+        ledger = MuteLedger.at(ctx.root)
+        if self.clear:
+            ledger.clear(lookup)
+            return MutedResult(thread.name, 0.0)
+        until = time.time() + self.minutes * 60
+        ledger.set(lookup, until)
+        return MutedResult(thread.name, until)

src/agent_comms/messaging/publisher.py

@@ -4,6 +4,7 @@
 
 import hashlib
 import sqlite3
+import time
 from collections.abc import Callable, Sequence
 from contextlib import AbstractContextManager
 from dataclasses import dataclass, replace
@@ -45,6 +46,7 @@
 from agent_comms.registry.registration import Registration
 from agent_comms.registry.registry_document import RegistrySnapshot
 from agent_comms.storage.private_path import FileRevision
+from agent_comms.waking.mute_ledger import MuteLedger
 
 
 class PublicationSnapshot(InputPublicationSource, BusSnapshot):
@@ -204,6 +206,7 @@ def publish_ordinary(self, message: Message, *, origin: InitialOrigin,
             selected = [snapshot.threads[name] for name in names]
             lookups = [thread.incarnation.lookup for thread in selected]
             sender_lookup = snapshot.threads[sender].incarnation.lookup
+            muted = MuteLedger.at(self.registry.store.path.parent).muted(time.time())
             def build(seq: int, db: sqlite3.Connection) -> PrivateRecord:
                 source = PublicationSnapshot.from_connection(db)
                 stored = self._prepare_message_unlocked(
@@ -224,7 +227,7 @@ def build(seq: int, db: sqlite3.Connection) -> PrivateRecord:
                     revision, sender_lookup=sender_lookup, sender_name=sender,
                 )
                 decisions = resolve_wake_cohort(
-                    audience, control=OrdinaryControl,
+                    audience, control=OrdinaryControl, muted=muted,
                 )
                 return PrivateRecord.from_message(
                     audience, decisions=decisions,

src/agent_comms/messaging/wake.py

@@ -12,8 +12,10 @@
 from agent_comms.messaging.frozen_recipient import FrozenRecipient
 from agent_comms.messaging.messages import Message
 from agent_comms.nominal.declared_family import DeclaredFamily
+from agent_comms.nominal.field_codec import ThreadLookup
 from agent_comms.waking.wake_policy import (
     NoWakeDecision,
+    NoWakeReason,
     PassiveWake,
     RecipientDecision,
     WakePolicy,
@@ -63,9 +65,12 @@ def _member_decision(
     control: type[MessageControl],
     target: NamedTarget,
     effective_mentions: frozenset[str],
+    muted: frozenset[ThreadLookup],
 ) -> RecipientDecision:
     if not message.type.is_conversation_message():
         return NoWakeDecision(recipient.recipient_lookup)
+    if recipient.recipient_lookup in muted:
+        return NoWakeDecision(recipient.recipient_lookup, NoWakeReason.MUTED)
     mode = message.role.wake_mode(target, recipient, effective_mentions, control)
     audience = target.wake_audience(effective_mentions)
     return mode.recipient_decision(recipient.recipient_lookup, audience)
@@ -76,28 +81,35 @@ def resolve_wake(
     *,
     recipient: FrozenRecipient,
     control: type[MessageControl],
+    muted: frozenset[ThreadLookup] = frozenset(),
 ) -> RecipientDecision | None:
     """Resolve one member; prefer resolve_wake_cohort for the full frozen N.
 
     ``None`` means the lookup/name pair is absent; NoWakeDecision is a frozen
     unmentioned observer, never a claim. Neither pure API proves bus commit.
+    ``muted`` carries the lookups whose wake the caller has resolved to suppress;
+    they still receive the message, only their wake is dropped.
     """
     target, effective_mentions = _validated_context(message, control)
     if recipient not in message.recipients:
         return None
-    return _member_decision(message, recipient, control, target, effective_mentions)
+    return _member_decision(message, recipient, control, target, effective_mentions, muted)
 
 
 def resolve_wake_cohort(
-    message: AudiencedMessage, *, control: type[MessageControl]
+    message: AudiencedMessage, *, control: type[MessageControl],
+    muted: frozenset[ThreadLookup] = frozenset(),
 ) -> tuple[RecipientDecision, ...]:
     """Resolve ordered full N with one pure full-envelope check and no live lookup.
 
     Stage3 must separately attest the committed row, recipient capture, and
     control/policy version at the original send boundary before persisting N.
+    ``muted`` is the caller's resolved set of suppressed-wake lookups (read from the
+    mute ledger at publish): every member is still a recipient, a muted one gets a
+    MUTED no-wake decision instead of a triage wake.
     """
     target, effective_mentions = _validated_context(message, control)
     return tuple(
-        _member_decision(message, recipient, control, target, effective_mentions)
+        _member_decision(message, recipient, control, target, effective_mentions, muted)
         for recipient in message.recipients
     )

src/agent_comms/waking/mute_ledger.py

@@ -0,0 +1,51 @@
+"""Per-member wake mutes: a member silences their own wake for a bounded time.
+
+Wake-all is the default: every channel member is woken to triage a post (#3478). A mute
+is the member's own opt-out — while it is active the member still RECEIVES every message,
+only their wake is suppressed, so the group chat still pings everybody unless someone has
+chosen quiet. One small durable file, read once per publish.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import ClassVar
+
+from agent_comms.nominal.field_codec import ThreadLookup
+from agent_comms.storage.locked_store import DocumentChange, DurableFile, LockedStore
+from agent_comms.storage.store_files import DependentStateTier, SingleFileStore
+
+
+@dataclass(frozen=True, slots=True)
+class Mute:
+    """One member's wake silenced until ``until`` (epoch seconds)."""
+
+    muter: ThreadLookup
+    until: float
+
+    def active(self, now: float) -> bool:
+        return self.until > now
+
+
+@dataclass(frozen=True, slots=True)
+class MuteLedger(SingleFileStore, DurableFile, LockedStore[tuple[Mute, ...]], DependentStateTier,
+                 declared_name="mute_ledger"):
+    """Every member's active wake mute; the declarations supply the stored shape via FieldCodec."""
+
+    filename: ClassVar[str] = "mutes.json"
+    record_type = tuple[Mute, ...]
+
+    def empty(self) -> tuple[Mute, ...]:
+        return ()
+
+    def muted(self, now: float) -> frozenset[ThreadLookup]:
+        """The lookups whose wake is suppressed right now; expired mutes fall out by time alone."""
+        return frozenset(mute.muter for mute in self.read() if mute.active(now))
+
+    def set(self, muter: ThreadLookup, until: float) -> None:
+        self.update(lambda mutes: DocumentChange.to(
+            (*(mute for mute in mutes if mute.muter != muter), Mute(muter, until))))
+
+    def clear(self, muter: ThreadLookup) -> None:
+        self.update(lambda mutes: DocumentChange.to(
+            tuple(mute for mute in mutes if mute.muter != muter)))

src/agent_comms/waking/wake_policy.py

@@ -29,6 +29,7 @@ class MessageAudience(StrEnum):
 
 class NoWakeReason(StrEnum):
     UNMENTIONED_OBSERVER = "unmentioned_observer"
+    MUTED = "muted"
 
 
 @dataclass(frozen=True, slots=True)

tests/test_mute_ledger.py

@@ -0,0 +1,29 @@
+"""The mute ledger: set, active-by-time, replace-not-duplicate, and clear."""
+
+from pathlib import Path
+
+from agent_comms.waking.mute_ledger import Mute, MuteLedger
+from bus_record_fixture import RECIPIENT_LOOKUP, SENDER_LOOKUP
+
+
+def test_mute_ledger_sets_expires_by_time_and_clears(tmp_path: Path) -> None:
+    ledger = MuteLedger.at(tmp_path)
+    assert ledger.muted(100.0) == frozenset()  # an absent file is no mutes.
+
+    ledger.set(SENDER_LOOKUP, until=200.0)
+    assert ledger.muted(100.0) == frozenset({SENDER_LOOKUP})
+    assert ledger.muted(200.0) == frozenset()  # active is strict: until > now.
+    assert ledger.muted(250.0) == frozenset()
+
+    ledger.set(SENDER_LOOKUP, until=500.0)  # re-muting replaces, never duplicates.
+    ledger.set(RECIPIENT_LOOKUP, until=500.0)
+    assert ledger.muted(100.0) == frozenset({SENDER_LOOKUP, RECIPIENT_LOOKUP})
+    assert len(ledger.read()) == 2
+
+    ledger.clear(SENDER_LOOKUP)
+    assert ledger.muted(100.0) == frozenset({RECIPIENT_LOOKUP})
+
+
+def test_mute_active_is_strict_by_until() -> None:
+    assert Mute(SENDER_LOOKUP, 10.0).active(9.0)
+    assert not Mute(SENDER_LOOKUP, 10.0).active(10.0)

tests/test_wake.py

@@ -302,6 +302,34 @@ def test_full_cohort_preserves_direct_control_and_fail_closed_envelope() -> None
         )
 
 
+def test_muted_member_is_delivered_without_wake_while_others_still_triage() -> None:
+    message = stored()  # a collective #team post: wake-all is the default.
+    decisions = resolve_wake_cohort(
+        audience(message), control=OrdinaryControl, muted=frozenset({RECIPIENT_LOOKUP}),
+    )
+    assert decisions == (
+        WakeDecision(SENDER_LOOKUP, MessageAudience.COLLECTIVE, BoundedTriageWake()),
+        NoWakeDecision(RECIPIENT_LOOKUP, NoWakeReason.MUTED),
+    )
+    # The muted member is still a frozen recipient: delivery is unchanged, only the wake drops.
+    assert MEMBERS[1] in audience(message).recipients
+    # A MUTED no-wake row round-trips through the wire like any committed decision.
+    assert RecipientDecision.from_wire(decisions[1].to_wire()) == decisions[1]
+
+
+def test_mute_suppresses_wake_even_for_a_mentioned_member() -> None:
+    message = stored("#team", "@beta investigate", mentions=(ThreadMention("beta", 0, 5),))
+    decisions = resolve_wake_cohort(
+        audience(message), control=OrdinaryControl, muted=frozenset({RECIPIENT_LOOKUP}),
+    )
+    # beta is mentioned (would be FullWake) but muted itself: still delivered, not woken.
+    assert decisions[1] == NoWakeDecision(RECIPIENT_LOOKUP, NoWakeReason.MUTED)
+    # An empty mute set leaves the default wake-all untouched.
+    assert resolve_wake_cohort(audience(message), control=OrdinaryControl)[1] == WakeDecision(
+        RECIPIENT_LOOKUP, MessageAudience.MENTIONED, FullWake()
+    )
+
+
 def test_exact_stored_channel_routes_no_wildcard_or_silence_inference(
     tmp_path: Path,
 ) -> None: