From 2c1511a487da8bbdb1c666912c43e55f0ac565a3 Mon Sep 17 00:00:00 2001 From: olegbrok Date: Sun, 28 Jun 2026 04:17:37 -0700 Subject: [PATCH 1/3] fix(broker): make agent-reply auto-route a true fallback with dedup (#280) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit requester's inbox. But agents also reply explicitly via send_to_agent (comms.send -> inbox directly), so a turn that did both delivered the reply TWICE. The dup hadn't surfaced yet only because no agent-injected turn had completed since #279 deployed; the next Murzik review verdict would have doubled. Make auto-route a fallback: suppress it when the responder already sent an explicit (non-auto_routed) message to the requester since the turn began. The turn boundary is captured at inject_agent_message time ((responder, requester) -> start ts), consumed by route_agent_reply. - Backend-agnostic: no reliance on tool_uses (codex turns omit them). - Never content-matches, so it can't drop a genuinely new reply. - Missing marker (e.g. lost across a restart) or a probe error falls through to delivery — a stray dup is acceptable, a dropped verdict is not. - Orphan markers swept by TTL on insert (bounded memory). - Kill-switch PINKY_AGENT_REPLY_DEDUP (default on) restores #279. New AgentComms.has_message_since() backs the probe. Counted via the existing broker `deduped` stat. Tests cover suppress-on-explicit-send, deliver-on-no-send, deliver-on-missing-marker (safety), killswitch, the inject marker, and the comms probe (incl. legacy/empty metadata). Co-Authored-By: Claude Opus 4.8 --- src/pinky_daemon/agent_comms.py | 32 +++++++ src/pinky_daemon/broker.py | 89 ++++++++++++++++++++ tests/test_agent_comms.py | 51 +++++++++++ tests/test_broker.py | 145 ++++++++++++++++++++++++++++++++ 4 files changed, 317 insertions(+) diff --git a/src/pinky_daemon/agent_comms.py b/src/pinky_daemon/agent_comms.py index ef0c9eba..34d451bd 100644 --- a/src/pinky_daemon/agent_comms.py +++ b/src/pinky_daemon/agent_comms.py @@ -456,6 +456,38 @@ def unread_count(self, session_id: str) -> int: ).fetchone() return row[0] + def has_message_since( + self, + from_session: str, + to_session: str, + since_ts: float, + *, + exclude_auto_routed: bool = True, + ) -> bool: + """Whether a direct message from ``from_session`` to ``to_session`` was + stored at or after ``since_ts``. + + Backs the #280 agent-reply dedup: the broker asks "did the responder + already send the requester an explicit reply during this turn?" before + auto-routing the same turn's text. ``exclude_auto_routed`` skips the + broker's own auto-routed copies (metadata ``{"auto_routed": true}``) so + only deliberate ``send_to_agent`` sends count. Existence probe — returns + as soon as one match is found. + """ + query = ( + "SELECT 1 FROM messages " + "WHERE from_session = ? AND to_session = ? AND timestamp >= ?" + ) + if exclude_auto_routed: + # COALESCE so rows with empty/legacy metadata (NULL json_extract) + # are treated as explicit (not auto-routed) and DO count. + query += " AND COALESCE(json_extract(metadata, '$.auto_routed'), 0) = 0" + query += " LIMIT 1" + row = self._conn.execute( + query, (from_session, to_session, since_ts) + ).fetchone() + return row is not None + # ── Groups ─────────────────────────────────────────────── def create_group(self, name: str, session_ids: list[str]) -> dict: diff --git a/src/pinky_daemon/broker.py b/src/pinky_daemon/broker.py index 246f3063..d36aba5e 100644 --- a/src/pinky_daemon/broker.py +++ b/src/pinky_daemon/broker.py @@ -69,6 +69,28 @@ except (TypeError, ValueError): _AGENT_MSG_NUDGE_BACKOFF_SEC = 15.0 +# #280: auto-route is a FALLBACK, not a second delivery path. When the +# recipient already replied explicitly via ``send_to_agent`` (which writes +# straight to the requester's inbox through ``comms.send``), auto-routing the +# same turn's final text would land the reply TWICE. ``route_agent_reply`` +# therefore suppresses the auto-route when an explicit (non-``auto_routed``) +# message from the responder to the requester was recorded since the turn +# began — the turn boundary captured at ``inject_agent_message`` time. This is +# backend-agnostic (no reliance on ``tool_uses``, which codex turns omit) and +# never content-matches, so it can't drop a genuinely new reply. If the turn +# boundary is unknown (e.g. the marker was lost across a restart), it falls +# back to delivering — a stray dup is acceptable; a dropped review verdict is +# not. Kill-switch defaults ON; set ``PINKY_AGENT_REPLY_DEDUP=0`` to restore +# the unconditional #279 delivery. +_AGENT_REPLY_DEDUP_ENABLED = os.environ.get( + "PINKY_AGENT_REPLY_DEDUP", "1" +).strip().lower() not in ("0", "false", "no", "off") +# How long an unconsumed turn-start marker lives before it's swept as an +# orphan. Generous — an agent-reply turn (e.g. a code review) can run for +# minutes — but bounded so a turn that dies before its callback can't leak a +# marker forever. Only affects the dedup window, never delivery correctness. +_AGENT_REPLY_MARKER_TTL = 3600.0 + def _log(msg: str) -> None: print(msg, file=sys.stderr, flush=True) @@ -309,6 +331,14 @@ def __init__( # agent_name -> {label -> StreamingSession} self._streaming: dict[str, dict[str, object]] = {} + # #280: agent-reply turn boundaries — (responder, requester) -> start ts. + # Stamped when ``inject_agent_message`` delivers a route-back turn; + # consumed (popped) by ``route_agent_reply`` to bound the "did the + # responder already send an explicit reply this turn?" dedup window. + # Bounded by consume-on-use plus a TTL sweep on insert so orphaned + # markers (turns that error before the callback) can't accumulate. + self._agent_reply_turn_start: dict[tuple[str, str], float] = {} + # Track voice-pending chats: (agent_name, chat_id) -> True when last inbound was voice self._voice_pending: dict[tuple[str, str], bool] = {} self._message_contexts: dict[tuple[str, str], MessageContext] = {} @@ -1434,6 +1464,11 @@ async def inject_agent_message( # (see ``route_agent_reply``). The ``platform="agent"`` sentinel is # intercepted in the response callback before normal platform # routing; ``chat_id`` carries the requester agent name. + # #280: stamp the turn's start so route_agent_reply can tell whether + # the responder ALSO sent an explicit reply during this turn (→ a + # dup to suppress). Key is (responder, requester) = (to_agent, + # from_agent), matching how route_agent_reply reads it back. + self._mark_agent_reply_turn_start(to_agent, from_agent) handoff = await streaming.send( prompt, platform=AGENT_REPLY_PLATFORM, chat_id=from_agent ) @@ -1575,6 +1610,24 @@ async def route_response( else: await self._send_message(agent_name, platform, chat_id, stripped) + def _mark_agent_reply_turn_start(self, responder: str, requester: str) -> None: + """#280: record when an agent-reply turn began, so ``route_agent_reply`` + can scope its dup check to "did the responder send an explicit reply to + the requester *during this turn*?". Sweeps orphaned markers (turns that + died before their response callback) older than the TTL on each insert + so the dict can't grow unbounded. Only affects the dedup window — never + delivery correctness (a lost marker just means we deliver).""" + now = time.time() + if self._agent_reply_turn_start: + cutoff = now - _AGENT_REPLY_MARKER_TTL + stale = [ + k for k, started in self._agent_reply_turn_start.items() + if started < cutoff + ] + for k in stale: + self._agent_reply_turn_start.pop(k, None) + self._agent_reply_turn_start[(responder, requester)] = now + async def route_agent_reply(self, comms, turn_result) -> bool: """Auto-deliver a completed agent-to-agent turn to the requester's inbox. @@ -1585,6 +1638,12 @@ async def route_agent_reply(self, comms, turn_result) -> bool: return leg that left review verdicts (and every other agent reply) stranded in the recipient's transcript with nowhere to go. + #280: this is a FALLBACK for replies the responder produced but did not + send explicitly (e.g. an interrupted turn). When the responder already + called ``send_to_agent`` to the requester this turn, that message is in + the inbox; the auto-route is suppressed so the reply isn't duplicated + (see the dedup block + ``_mark_agent_reply_turn_start``). + Loop-safe by construction: this writes to the inbox only and never injects a live turn, so an auto-routed reply cannot itself trigger another routed turn — the exchange terminates. @@ -1608,6 +1667,36 @@ async def route_agent_reply(self, comms, turn_result) -> bool: # A pure tool-call turn with no final text — nothing to relay. _log(f"broker: agent reply {responder} -> {requester} had no text") return True + # #280: auto-route is a FALLBACK, not a second delivery path. If the + # responder already replied to the requester explicitly this turn + # (send_to_agent -> comms.send, which lands straight in the inbox), + # auto-routing the same turn's final text would deliver it twice. + # Suppress when an explicit (non-auto_routed) responder->requester + # message exists since this turn began. Backend-agnostic (no reliance on + # tool_uses, which codex turns omit) and never content-matches, so a + # genuinely new reply can't be dropped. A missing marker (e.g. lost + # across a restart) falls through to delivery — a stray dup is + # acceptable, a dropped verdict is not. + since = self._agent_reply_turn_start.pop((responder, requester), None) + if _AGENT_REPLY_DEDUP_ENABLED and since is not None: + try: + already_sent = comms.has_message_since( + responder, requester, since, exclude_auto_routed=True + ) + except Exception as e: + # A dedup-probe failure must never drop a reply — deliver. + already_sent = False + _log( + f"broker: agent-reply dedup probe failed " + f"({responder} -> {requester}): {e}; delivering" + ) + if already_sent: + self._stats["deduped"] += 1 + _log( + f"broker: suppressed dup auto-route {responder} -> " + f"{requester} (explicit reply already sent this turn)" + ) + return True try: comms.send(responder, requester, text, metadata={"auto_routed": True}) self._stats["routed"] += 1 diff --git a/tests/test_agent_comms.py b/tests/test_agent_comms.py index c0702638..f671d5e8 100644 --- a/tests/test_agent_comms.py +++ b/tests/test_agent_comms.py @@ -52,6 +52,57 @@ def test_direct_not_in_sender_inbox(self): assert len(inbox) == 0 self._cleanup(comms, path) + # ── has_message_since (#280 agent-reply dedup probe) ───── + + def test_has_message_since_detects_explicit_send(self): + comms, path = self._make_comms() + import time + + t0 = time.time() + comms.send("murzik", "barsik", "LGTM verdict") + assert comms.has_message_since("murzik", "barsik", t0) is True + # Direction matters: barsik->murzik is a different pair. + assert comms.has_message_since("barsik", "murzik", t0) is False + self._cleanup(comms, path) + + def test_has_message_since_excludes_auto_routed_by_default(self): + comms, path = self._make_comms() + import time + + t0 = time.time() + # Only an auto-routed copy exists — must NOT count as an explicit send. + comms.send("murzik", "barsik", "auto copy", metadata={"auto_routed": True}) + assert comms.has_message_since("murzik", "barsik", t0) is False + # ...but it IS visible when auto-routed rows are included. + assert ( + comms.has_message_since( + "murzik", "barsik", t0, exclude_auto_routed=False + ) + is True + ) + self._cleanup(comms, path) + + def test_has_message_since_respects_window(self): + comms, path = self._make_comms() + import time + + comms.send("murzik", "barsik", "earlier reply") + future = time.time() + 100 + # Nothing was sent at/after a future cutoff. + assert comms.has_message_since("murzik", "barsik", future) is False + self._cleanup(comms, path) + + def test_has_message_since_legacy_empty_metadata_counts(self): + comms, path = self._make_comms() + import time + + t0 = time.time() + # Default metadata is "{}" — json_extract returns NULL, COALESCE -> 0, + # so a normal send (no auto_routed key) is correctly treated as explicit. + comms.send("murzik", "barsik", "plain reply") + assert comms.has_message_since("murzik", "barsik", t0) is True + self._cleanup(comms, path) + def test_mark_read(self): comms, path = self._make_comms() msg = comms.send("alice", "bob", "Hey!") diff --git a/tests/test_broker.py b/tests/test_broker.py index 2411dbd8..21c8a348 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -652,6 +652,151 @@ async def test_route_agent_reply_real_comms_lands_in_inbox(self): finally: tmpdir.cleanup() + @pytest.mark.asyncio + async def test_route_agent_reply_suppresses_dup_when_explicit_send_this_turn(self): + """#280: if the responder already replied explicitly (send_to_agent -> + comms.send) during the turn, the auto-route is suppressed — the inbox + keeps the single explicit copy, not a duplicate.""" + from pinky_daemon.agent_comms import AgentComms + from pinky_daemon.turn_response import TurnResponse + + tmpdir, registry, broker, _, _ = self._make_broker() + try: + comms = AgentComms(db_path=f"{tmpdir.name}/comms.db") + # Turn begins (inject_agent_message would stamp this marker). + broker._mark_agent_reply_turn_start("murzik", "barsik") + # Responder sends the verdict explicitly mid-turn. + comms.send("murzik", "barsik", "PR #277 LGTM") + before = broker._stats["deduped"] + + tr = TurnResponse( + agent_name="murzik", + platform="agent", + chat_id="barsik", + text="PR #277 LGTM", + ) + handled = await broker.route_agent_reply(comms, tr) + assert handled is True + inbox = comms.get_inbox("barsik") + # Only the explicit send is present — no auto_routed duplicate. + assert len(inbox) == 1 + assert inbox[0].metadata.get("auto_routed") is None + assert broker._stats["deduped"] == before + 1 + # Marker is consumed. + assert ("murzik", "barsik") not in broker._agent_reply_turn_start + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_route_agent_reply_delivers_when_marker_but_no_explicit_send(self): + """#280: turn produced text but never sent it explicitly (e.g. an + interrupted reply) — the auto-route fallback still delivers.""" + from pinky_daemon.agent_comms import AgentComms + from pinky_daemon.turn_response import TurnResponse + + tmpdir, registry, broker, _, _ = self._make_broker() + try: + comms = AgentComms(db_path=f"{tmpdir.name}/comms.db") + broker._mark_agent_reply_turn_start("murzik", "barsik") + + tr = TurnResponse( + agent_name="murzik", + platform="agent", + chat_id="barsik", + text="LGTM but I never called send_to_agent", + ) + handled = await broker.route_agent_reply(comms, tr) + assert handled is True + inbox = comms.get_inbox("barsik") + assert len(inbox) == 1 + assert inbox[0].metadata.get("auto_routed") is True + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_route_agent_reply_delivers_when_marker_missing(self): + """#280 safety: with no turn marker (e.g. lost across a restart) the + reply is DELIVERED even if an explicit send happens to exist — a stray + dup is acceptable, dropping a real reply is not.""" + from pinky_daemon.agent_comms import AgentComms + from pinky_daemon.turn_response import TurnResponse + + tmpdir, registry, broker, _, _ = self._make_broker() + try: + comms = AgentComms(db_path=f"{tmpdir.name}/comms.db") + comms.send("murzik", "barsik", "explicit reply") + # No _mark_agent_reply_turn_start — marker is absent. + before = broker._stats["deduped"] + + tr = TurnResponse( + agent_name="murzik", + platform="agent", + chat_id="barsik", + text="explicit reply", + ) + handled = await broker.route_agent_reply(comms, tr) + assert handled is True + inbox = comms.get_inbox("barsik") + # Both the explicit send and the (safe) auto-route are present. + assert len(inbox) == 2 + assert broker._stats["deduped"] == before + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_route_agent_reply_dedup_killswitch_off_always_delivers( + self, monkeypatch + ): + """With PINKY_AGENT_REPLY_DEDUP off, the #279 unconditional delivery is + restored: even a marker + explicit send doesn't suppress the auto-route.""" + import pinky_daemon.broker as broker_mod + from pinky_daemon.agent_comms import AgentComms + from pinky_daemon.turn_response import TurnResponse + + monkeypatch.setattr(broker_mod, "_AGENT_REPLY_DEDUP_ENABLED", False) + tmpdir, registry, broker, _, _ = self._make_broker() + try: + comms = AgentComms(db_path=f"{tmpdir.name}/comms.db") + broker._mark_agent_reply_turn_start("murzik", "barsik") + comms.send("murzik", "barsik", "verdict") + + tr = TurnResponse( + agent_name="murzik", + platform="agent", + chat_id="barsik", + text="verdict", + ) + handled = await broker.route_agent_reply(comms, tr) + assert handled is True + inbox = comms.get_inbox("barsik") + assert len(inbox) == 2 # explicit + auto-route (no dedup) + # Marker is still consumed even with dedup disabled (no leak). + assert ("murzik", "barsik") not in broker._agent_reply_turn_start + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_inject_agent_message_marks_turn_start(self): + """#280: a routed inject stamps the (responder, requester) turn-start + marker that route_agent_reply later consumes for dedup.""" + tmpdir, registry, broker, _, _ = self._make_broker() + try: + from pinky_daemon.transport_state import SessionState + + class _FakeStreaming: + state = SessionState.CONNECTED + + async def send(self, prompt, *, platform="", chat_id="", message_id=""): + pass + + broker.register_streaming("barsik", _FakeStreaming(), label="main") + delivered, _ = await broker.inject_agent_message("pushok", "barsik", "review please") + assert delivered is True + # Key is (responder, requester) = (to_agent, from_agent). + assert ("barsik", "pushok") in broker._agent_reply_turn_start + finally: + tmpdir.cleanup() + @pytest.mark.asyncio async def test_first_inbound_claims_primary_on_fresh_install(self, monkeypatch): """Fresh install: no primary user configured. The first person to From 33c6e1d464946d4141bb6639d1c34079beace08e Mon Sep 17 00:00:00 2001 From: olegbrok Date: Fri, 10 Jul 2026 12:52:54 -0700 Subject: [PATCH 2/3] fix(broker): remove turn-start marker on failed/raised handoff (#831 P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failed per-call handoff produces no route-back turn, but the #280 dedup marker stamped before the send survived it. The orphan stayed live for up to the marker TTL and any unrelated explicit responder->requester message would then suppress a DIFFERENT injection's distinct route-back — a dropped verdict, not a stray dup. The unmark is identity-guarded (timestamp token) so an older failed call can never pop a newer concurrent injection's marker, and dedup still fails open to delivery when no marker is present. Regressions: failed handoff leaves no marker and cannot suppress a distinct later route-back; raising sends unmark and propagate; True-handoff behavior unchanged. Co-Authored-By: Claude Fable 5 --- src/pinky_daemon/broker.py | 47 +++++++++++++--- tests/test_broker.py | 106 +++++++++++++++++++++++++++++++++++-- 2 files changed, 143 insertions(+), 10 deletions(-) diff --git a/src/pinky_daemon/broker.py b/src/pinky_daemon/broker.py index d36aba5e..8c6e9808 100644 --- a/src/pinky_daemon/broker.py +++ b/src/pinky_daemon/broker.py @@ -1467,11 +1467,24 @@ async def inject_agent_message( # #280: stamp the turn's start so route_agent_reply can tell whether # the responder ALSO sent an explicit reply during this turn (→ a # dup to suppress). Key is (responder, requester) = (to_agent, - # from_agent), matching how route_agent_reply reads it back. - self._mark_agent_reply_turn_start(to_agent, from_agent) - handoff = await streaming.send( - prompt, platform=AGENT_REPLY_PLATFORM, chat_id=from_agent - ) + # from_agent), matching how route_agent_reply reads it back. The + # stamp precedes the await so the window covers the whole turn, but + # it must not outlive a failed handoff — no turn routes back for + # this injection, and an orphaned marker would suppress a distinct + # LATER route-back for the same pair once any unrelated explicit + # message lands (Murzik #831 P1). Unmark is identity-guarded, and + # dedup fails open to delivery, so an unconfirmed-but-live turn at + # worst duplicates; it can never drop. + marker = self._mark_agent_reply_turn_start(to_agent, from_agent) + try: + handoff = await streaming.send( + prompt, platform=AGENT_REPLY_PLATFORM, chat_id=from_agent + ) + except BaseException: + self._unmark_agent_reply_turn_start(to_agent, from_agent, marker) + raise + if not handoff: + self._unmark_agent_reply_turn_start(to_agent, from_agent, marker) else: handoff = await streaming.send(prompt) confirmed = bool(handoff) and bool( @@ -1610,13 +1623,16 @@ async def route_response( else: await self._send_message(agent_name, platform, chat_id, stripped) - def _mark_agent_reply_turn_start(self, responder: str, requester: str) -> None: + def _mark_agent_reply_turn_start(self, responder: str, requester: str) -> float: """#280: record when an agent-reply turn began, so ``route_agent_reply`` can scope its dup check to "did the responder send an explicit reply to the requester *during this turn*?". Sweeps orphaned markers (turns that died before their response callback) older than the TTL on each insert so the dict can't grow unbounded. Only affects the dedup window — never - delivery correctness (a lost marker just means we deliver).""" + delivery correctness (a lost marker just means we deliver). + + Returns the stamped timestamp: the caller's identity token for + ``_unmark_agent_reply_turn_start`` when its handoff fails.""" now = time.time() if self._agent_reply_turn_start: cutoff = now - _AGENT_REPLY_MARKER_TTL @@ -1627,6 +1643,23 @@ def _mark_agent_reply_turn_start(self, responder: str, requester: str) -> None: for k in stale: self._agent_reply_turn_start.pop(k, None) self._agent_reply_turn_start[(responder, requester)] = now + return now + + def _unmark_agent_reply_turn_start( + self, responder: str, requester: str, marker: float + ) -> None: + """Remove ONLY the marker instance stamped by the caller's injection. + + A failed/raised handoff produces no route-back turn, so its retained + stamp would let any unrelated explicit responder→requester message + suppress a DIFFERENT injection's distinct route-back for up to the + marker TTL (Murzik #831 P1 — a dropped verdict, not a stray dup). The + equality guard keeps this safe under concurrency: a newer injection + for the same pair owns a newer timestamp, which an older failed call + must not pop.""" + key = (responder, requester) + if self._agent_reply_turn_start.get(key) == marker: + self._agent_reply_turn_start.pop(key, None) async def route_agent_reply(self, comms, turn_result) -> bool: """Auto-deliver a completed agent-to-agent turn to the requester's inbox. diff --git a/tests/test_broker.py b/tests/test_broker.py index 21c8a348..b89a149f 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -777,8 +777,9 @@ async def test_route_agent_reply_dedup_killswitch_off_always_delivers( @pytest.mark.asyncio async def test_inject_agent_message_marks_turn_start(self): - """#280: a routed inject stamps the (responder, requester) turn-start - marker that route_agent_reply later consumes for dedup.""" + """#280: a routed inject with a SUCCESSFUL handoff stamps the + (responder, requester) turn-start marker that route_agent_reply later + consumes for dedup.""" tmpdir, registry, broker, _, _ = self._make_broker() try: from pinky_daemon.transport_state import SessionState @@ -787,7 +788,7 @@ class _FakeStreaming: state = SessionState.CONNECTED async def send(self, prompt, *, platform="", chat_id="", message_id=""): - pass + return True broker.register_streaming("barsik", _FakeStreaming(), label="main") delivered, _ = await broker.inject_agent_message("pushok", "barsik", "review please") @@ -797,6 +798,105 @@ async def send(self, prompt, *, platform="", chat_id="", message_id=""): finally: tmpdir.cleanup() + @pytest.mark.asyncio + async def test_inject_failed_handoff_leaves_no_marker(self): + """Murzik #831 P1: a FAILED handoff produces no route-back turn, so its + marker must not survive — an orphan would suppress a distinct later + route-back for the same pair once any unrelated explicit message lands.""" + tmpdir, registry, broker, _, _ = self._make_broker() + try: + from pinky_daemon.transport_state import SessionState + + class _FakeStreaming: + state = SessionState.CONNECTED + + async def send(self, prompt, *, platform="", chat_id="", message_id=""): + return False + + broker.register_streaming("barsik", _FakeStreaming(), label="main") + delivered, confirmed = await broker.inject_agent_message( + "pushok", "barsik", "review please" + ) + assert delivered is True and confirmed is False + assert ("barsik", "pushok") not in broker._agent_reply_turn_start + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_inject_raising_send_removes_marker_and_propagates(self): + """A send that raises has no route-back turn either — the marker is + removed in the except path and the exception still propagates.""" + tmpdir, registry, broker, _, _ = self._make_broker() + try: + from pinky_daemon.transport_state import SessionState + + class _FakeStreaming: + state = SessionState.CONNECTED + + async def send(self, prompt, *, platform="", chat_id="", message_id=""): + raise RuntimeError("transport exploded") + + broker.register_streaming("barsik", _FakeStreaming(), label="main") + with pytest.raises(RuntimeError, match="transport exploded"): + await broker.inject_agent_message("pushok", "barsik", "review please") + assert ("barsik", "pushok") not in broker._agent_reply_turn_start + finally: + tmpdir.cleanup() + + @pytest.mark.asyncio + async def test_failed_handoff_cannot_suppress_distinct_later_route_back(self): + """Murzik #831 P1 end-to-end repro, inverted: failed-handoff inject, + then an unrelated explicit responder->requester message, then a DISTINCT + routed turn text — the route-back must DELIVER, not dedup-suppress.""" + from pinky_daemon.agent_comms import AgentComms + from pinky_daemon.turn_response import TurnResponse + + tmpdir, registry, broker, _, _ = self._make_broker() + try: + from pinky_daemon.transport_state import SessionState + + class _FakeStreaming: + state = SessionState.CONNECTED + + async def send(self, prompt, *, platform="", chat_id="", message_id=""): + return False + + broker.register_streaming("barsik", _FakeStreaming(), label="main") + await broker.inject_agent_message("pushok", "barsik", "review please") + comms = AgentComms(db_path=f"{tmpdir.name}/comms.db") + # Unrelated explicit message AFTER the failed stamp would have + # satisfied has-message-since against an orphaned marker. + comms.send("barsik", "pushok", "unrelated status note") + before = broker._stats["deduped"] + + tr = TurnResponse( + agent_name="barsik", + platform="agent", + chat_id="pushok", + text="distinct in-flight verdict", + ) + handled = await broker.route_agent_reply(comms, tr) + assert handled is True + inbox = comms.get_inbox("pushok") + contents = [m.content for m in inbox] + assert "distinct in-flight verdict" in contents + assert broker._stats["deduped"] == before + finally: + tmpdir.cleanup() + + def test_unmark_is_identity_guarded(self): + """An older failed injection's unmark must not pop a NEWER concurrent + injection's marker for the same pair.""" + tmpdir, registry, broker, _, _ = self._make_broker() + try: + broker._agent_reply_turn_start[("barsik", "pushok")] = 123.0 + broker._unmark_agent_reply_turn_start("barsik", "pushok", 100.0) + assert broker._agent_reply_turn_start[("barsik", "pushok")] == 123.0 + broker._unmark_agent_reply_turn_start("barsik", "pushok", 123.0) + assert ("barsik", "pushok") not in broker._agent_reply_turn_start + finally: + tmpdir.cleanup() + @pytest.mark.asyncio async def test_first_inbound_claims_primary_on_fresh_install(self, monkeypatch): """Fresh install: no primary user configured. The first person to From 668895c5732e3d212dcb50288d6365c0c1cfbe6e Mon Sep 17 00:00:00 2001 From: olegbrok Date: Fri, 10 Jul 2026 13:06:54 -0700 Subject: [PATCH 3/3] fix(broker): marker ownership by object identity, not timestamp equality (#831 P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit time.time() repeats at clock resolution, so two overlapping injections for the same pair can stamp EQUAL floats — an equality-guarded unmark then lets the older failed inject pop the newer successful marker. Markers are now _ReplyTurnMarker records: started_at carries the dedup window boundary consumed by route_agent_reply, while ownership is the record's object identity compared with `is`. Adds the overlapping same-timestamp regression (frozen clock, older send blocks and fails while newer succeeds — newer marker must survive). Co-Authored-By: Claude Fable 5 --- src/pinky_daemon/broker.py | 52 ++++++++++++++++++++-------- tests/test_broker.py | 71 +++++++++++++++++++++++++++++++++++--- 2 files changed, 103 insertions(+), 20 deletions(-) diff --git a/src/pinky_daemon/broker.py b/src/pinky_daemon/broker.py index 8c6e9808..57a28a65 100644 --- a/src/pinky_daemon/broker.py +++ b/src/pinky_daemon/broker.py @@ -96,6 +96,23 @@ def _log(msg: str) -> None: print(msg, file=sys.stderr, flush=True) +class _ReplyTurnMarker: + """#280 dedup marker: the turn-start boundary plus its ownership identity. + + ``started_at`` is the dedup window boundary route_agent_reply compares + messages against. OWNERSHIP is the record's object identity — never the + timestamp: ``time.time()`` repeats at clock resolution, so two overlapping + injections for the same pair can stamp EQUAL floats, and an equality-based + unmark would let the older failed call pop the newer successful marker + (Murzik #831 P1 follow-up). Each ``_mark`` allocates a fresh instance; + ``_unmark`` compares with ``is``.""" + + __slots__ = ("started_at",) + + def __init__(self, started_at: float) -> None: + self.started_at = started_at + + class InjectResult(NamedTuple): """Outcome of a live agent-message injection attempt. @@ -1623,7 +1640,9 @@ async def route_response( else: await self._send_message(agent_name, platform, chat_id, stripped) - def _mark_agent_reply_turn_start(self, responder: str, requester: str) -> float: + def _mark_agent_reply_turn_start( + self, responder: str, requester: str + ) -> _ReplyTurnMarker: """#280: record when an agent-reply turn began, so ``route_agent_reply`` can scope its dup check to "did the responder send an explicit reply to the requester *during this turn*?". Sweeps orphaned markers (turns that @@ -1631,22 +1650,24 @@ def _mark_agent_reply_turn_start(self, responder: str, requester: str) -> float: so the dict can't grow unbounded. Only affects the dedup window — never delivery correctness (a lost marker just means we deliver). - Returns the stamped timestamp: the caller's identity token for - ``_unmark_agent_reply_turn_start`` when its handoff fails.""" + Returns the freshly allocated marker record: the caller's ownership + handle for ``_unmark_agent_reply_turn_start`` when its handoff fails. + Identity lives in the OBJECT, not the timestamp — see _ReplyTurnMarker.""" now = time.time() if self._agent_reply_turn_start: cutoff = now - _AGENT_REPLY_MARKER_TTL stale = [ - k for k, started in self._agent_reply_turn_start.items() - if started < cutoff + k for k, marker in self._agent_reply_turn_start.items() + if marker.started_at < cutoff ] for k in stale: self._agent_reply_turn_start.pop(k, None) - self._agent_reply_turn_start[(responder, requester)] = now - return now + marker = _ReplyTurnMarker(now) + self._agent_reply_turn_start[(responder, requester)] = marker + return marker def _unmark_agent_reply_turn_start( - self, responder: str, requester: str, marker: float + self, responder: str, requester: str, marker: _ReplyTurnMarker ) -> None: """Remove ONLY the marker instance stamped by the caller's injection. @@ -1654,11 +1675,12 @@ def _unmark_agent_reply_turn_start( stamp would let any unrelated explicit responder→requester message suppress a DIFFERENT injection's distinct route-back for up to the marker TTL (Murzik #831 P1 — a dropped verdict, not a stray dup). The - equality guard keeps this safe under concurrency: a newer injection - for the same pair owns a newer timestamp, which an older failed call - must not pop.""" + ``is`` guard keeps this safe under concurrency even when two markers + carry EQUAL wall-clock timestamps: a newer injection for the same pair + owns a distinct record object, which an older failed call can never + pop.""" key = (responder, requester) - if self._agent_reply_turn_start.get(key) == marker: + if self._agent_reply_turn_start.get(key) is marker: self._agent_reply_turn_start.pop(key, None) async def route_agent_reply(self, comms, turn_result) -> bool: @@ -1710,11 +1732,11 @@ async def route_agent_reply(self, comms, turn_result) -> bool: # genuinely new reply can't be dropped. A missing marker (e.g. lost # across a restart) falls through to delivery — a stray dup is # acceptable, a dropped verdict is not. - since = self._agent_reply_turn_start.pop((responder, requester), None) - if _AGENT_REPLY_DEDUP_ENABLED and since is not None: + marker = self._agent_reply_turn_start.pop((responder, requester), None) + if _AGENT_REPLY_DEDUP_ENABLED and marker is not None: try: already_sent = comms.has_message_since( - responder, requester, since, exclude_auto_routed=True + responder, requester, marker.started_at, exclude_auto_routed=True ) except Exception as e: # A dedup-probe failure must never drop a reply — deliver. diff --git a/tests/test_broker.py b/tests/test_broker.py index b89a149f..2a5556e5 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -886,17 +886,78 @@ async def send(self, prompt, *, platform="", chat_id="", message_id=""): def test_unmark_is_identity_guarded(self): """An older failed injection's unmark must not pop a NEWER concurrent - injection's marker for the same pair.""" + injection's marker — even when both carry the SAME wall-clock stamp + (time.time() repeats at clock resolution; equality is not ownership).""" + from pinky_daemon.broker import _ReplyTurnMarker + tmpdir, registry, broker, _, _ = self._make_broker() try: - broker._agent_reply_turn_start[("barsik", "pushok")] = 123.0 - broker._unmark_agent_reply_turn_start("barsik", "pushok", 100.0) - assert broker._agent_reply_turn_start[("barsik", "pushok")] == 123.0 - broker._unmark_agent_reply_turn_start("barsik", "pushok", 123.0) + old = _ReplyTurnMarker(123.0) + new = _ReplyTurnMarker(123.0) # equal timestamp, distinct identity + broker._agent_reply_turn_start[("barsik", "pushok")] = new + broker._unmark_agent_reply_turn_start("barsik", "pushok", old) + assert broker._agent_reply_turn_start[("barsik", "pushok")] is new + broker._unmark_agent_reply_turn_start("barsik", "pushok", new) assert ("barsik", "pushok") not in broker._agent_reply_turn_start finally: tmpdir.cleanup() + @pytest.mark.asyncio + async def test_overlapping_injects_same_timestamp_newer_marker_survives( + self, monkeypatch + ): + """Murzik #831 P1 follow-up, end-to-end: wall clock frozen so both + injects stamp EQUAL timestamps; the older inject's handoff fails while + the newer succeeds — the newer marker must survive the older cleanup.""" + import asyncio + + import pinky_daemon.broker as broker_mod + + tmpdir, registry, broker, _, _ = self._make_broker() + try: + from pinky_daemon.transport_state import SessionState + + monkeypatch.setattr(broker_mod.time, "time", lambda: 123.0) + release_a = asyncio.Event() + + class _FakeStreaming: + state = SessionState.CONNECTED + calls = 0 + + async def send(self, prompt, *, platform="", chat_id="", message_id=""): + _FakeStreaming.calls += 1 + if _FakeStreaming.calls == 1: + await release_a.wait() + return False # older inject A: failed handoff + return True # newer inject B: success + + broker.register_streaming("barsik", _FakeStreaming(), label="main") + task_a = asyncio.create_task( + broker.inject_agent_message("pushok", "barsik", "first") + ) + for _ in range(1000): + if _FakeStreaming.calls: + break + await asyncio.sleep(0) + assert _FakeStreaming.calls == 1, "inject A never reached its send" + + delivered_b, _ = await broker.inject_agent_message( + "pushok", "barsik", "second" + ) + assert delivered_b is True + marker_after_b = broker._agent_reply_turn_start.get(("barsik", "pushok")) + assert marker_after_b is not None + + release_a.set() + await task_a + # A's failed-handoff cleanup must not have popped B's marker. + assert ( + broker._agent_reply_turn_start.get(("barsik", "pushok")) + is marker_after_b + ) + finally: + tmpdir.cleanup() + @pytest.mark.asyncio async def test_first_inbound_claims_primary_on_fresh_install(self, monkeypatch): """Fresh install: no primary user configured. The first person to