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..57a28a65 100644
--- a/src/pinky_daemon/broker.py
+++ b/src/pinky_daemon/broker.py
@@ -69,11 +69,50 @@
 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)
 
 
+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.
 
@@ -309,6 +348,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,9 +1481,27 @@ 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.
-            handoff = await streaming.send(
-                prompt, platform=AGENT_REPLY_PLATFORM, chat_id=from_agent
-            )
+            # #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. 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(
@@ -1575,6 +1640,49 @@ 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
+    ) -> _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
+        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).
+
+        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, 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)
+        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: _ReplyTurnMarker
+    ) -> 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
+        ``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) is 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.
 
@@ -1585,6 +1693,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 +1722,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.
+        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, marker.started_at, 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..2a5556e5 100644
--- a/tests/test_broker.py
+++ b/tests/test_broker.py
@@ -652,6 +652,312 @@ 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 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
+
+            class _FakeStreaming:
+                state = SessionState.CONNECTED
+
+                async def send(self, prompt, *, platform="", chat_id="", message_id=""):
+                    return True
+
+            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_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 — 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:
+            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