diff --git a/src/agent_comms/cli_verbs/mute.py b/src/agent_comms/cli_verbs/mute.py new file mode 100644 index 0000000000..e01291d3e9 --- /dev/null +++ b/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) diff --git a/src/agent_comms/messaging/publisher.py b/src/agent_comms/messaging/publisher.py index 6c3ea4a2b7..610a1326dc 100644 --- a/src/agent_comms/messaging/publisher.py +++ b/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, diff --git a/src/agent_comms/messaging/wake.py b/src/agent_comms/messaging/wake.py index a777b6ce3e..df4bd5d0e1 100644 --- a/src/agent_comms/messaging/wake.py +++ b/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 ) diff --git a/src/agent_comms/waking/mute_ledger.py b/src/agent_comms/waking/mute_ledger.py new file mode 100644 index 0000000000..72eb02da6a --- /dev/null +++ b/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))) diff --git a/src/agent_comms/waking/wake_policy.py b/src/agent_comms/waking/wake_policy.py index d4c9959506..5fb6efa991 100644 --- a/src/agent_comms/waking/wake_policy.py +++ b/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) diff --git a/tests/test_mute_ledger.py b/tests/test_mute_ledger.py new file mode 100644 index 0000000000..3c2238d3c6 --- /dev/null +++ b/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) diff --git a/tests/test_wake.py b/tests/test_wake.py index 9ef3acaed0..0014ba3c66 100644 --- a/tests/test_wake.py +++ b/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: