fix(gateway): skip sender prefix for internal events + NO_REPLY contract in async delegation (#66480)

gateway/run.py

@@ -11477,7 +11477,7 @@ async def _prepare_inbound_message_text(
             group_sessions_per_user=_group_sessions_per_user,
             thread_sessions_per_user=_thread_sessions_per_user,
         )
-        if _is_shared_multi_user and source.user_name:
+        if _is_shared_multi_user and source.user_name and not event.internal:
             # source.user_name is the platform display name — attacker-
             # influenceable on any platform that lets participants set their
             # own name. Neutralize embedded newlines/control chars before

tests/gateway/test_shared_group_sender_prefix.py

@@ -43,6 +43,48 @@ async def test_preprocess_prefixes_sender_for_shared_non_thread_group_session():
     assert result == "[Alice] hello"
 
 
+@pytest.mark.asyncio
+async def test_internal_event_skips_sender_prefix_in_shared_session():
+    """Internal synthetic events (async delegation completions) must not
+    receive the human sender prefix in shared multi-user sessions.
+
+    Regression for #66480: an async delegation completion injected via
+    _inject_watch_notification carries the originating user's SessionSource
+    but is marked internal=True. Without checking event.internal, the
+    shared-session prefix logic impersonates that user.
+    """
+    runner = _make_runner(
+        GatewayConfig(
+            platforms={
+                Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake"),
+            },
+            group_sessions_per_user=False,
+        )
+    )
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="-1002285219667",
+        chat_name="Test Group",
+        chat_type="group",
+        user_name="Alice",
+    )
+    event = MessageEvent(
+        text="[ASYNC DELEGATION COMPLETE — deleg_abc]",
+        source=source,
+        internal=True,
+    )
+
+    result = await runner._prepare_inbound_message_text(
+        event=event,
+        source=source,
+        history=[],
+    )
+
+    # Internal events must not be prefixed with the originating user's name.
+    assert result == "[ASYNC DELEGATION COMPLETE — deleg_abc]"
+    assert "[Alice]" not in result
+
+
 @pytest.mark.asyncio
 async def test_preprocess_keeps_plain_text_for_default_group_sessions():
     runner = _make_runner(

tests/tools/test_async_delegation.py

@@ -829,6 +829,47 @@ def test_gateway_formatter_renders_async_block():
     assert "Investigate flaky test" in txt
 
 
+def test_async_delegation_formatter_includes_no_reply_contract():
+    """The async delegation completion formatter must tell the parent agent
+    to respond exactly NO_REPLY when the result does not materially change
+    or correct an answer already delivered.
+
+    Regression for #66480: without this instruction, a no-news async
+    completion triggers a redundant second user-facing reply.
+    """
+    from tools.process_registry import format_process_notification
+
+    txt = format_process_notification(_make_async_evt())
+    assert txt is not None
+    assert "NO_REPLY" in txt
+
+
+def test_batch_async_delegation_formatter_includes_no_reply_contract():
+    """The batch (fan-out) async delegation completion formatter must also
+    tell the parent agent to respond exactly NO_REPLY when the results do not
+    materially change or correct an answer already delivered.
+
+    Regression for #66480: the issue requires the no-news suppression
+    contract in BOTH the single and the batch completion formatters. A batch
+    fan-out that finishes with no material change must not trigger a
+    redundant follow-up either.
+    """
+    from tools.process_registry import format_process_notification
+
+    batch_evt = _make_async_evt(
+        is_batch=True,
+        results=[
+            {"task_index": 0, "status": "completed", "summary": "no new findings"},
+        ],
+        goals=["Investigate flaky test"],
+        total_duration_seconds=12.0,
+    )
+    txt = format_process_notification(batch_evt)
+    assert txt is not None
+    assert "BATCH COMPLETE" in txt
+    assert "NO_REPLY" in txt
+
+
 def test_gateway_watch_drain_requeues_async_without_looping():
     from gateway.run import _drain_gateway_watch_events
 

tools/process_registry.py

@@ -2072,7 +2072,10 @@ def _format_async_delegation(evt: dict) -> str:
             f"A background fan-out of {n} subagent(s) you dispatched earlier "
             "has finished. All ran in parallel and waited on each other; their "
             "consolidated results are below. You may have moved on since "
-            "dispatching — act on these or re-dispatch if things have changed.",
+            "dispatching — act on these or re-dispatch if things have changed. "
+            "If this does not materially change or correct an answer you have "
+            "already delivered, respond exactly NO_REPLY so no redundant "
+            "follow-up is sent to the user.",
             "",
         ]
         if isinstance(dispatched_at, (int, float)):
@@ -2135,6 +2138,9 @@ def _format_async_delegation(evt: dict) -> str:
         "A background subagent you dispatched earlier has finished. You may "
         "have moved on since dispatching it; the full task source is below so "
         "you can act on the result or re-dispatch if things have changed.",
+        "If this does not materially change or correct an answer you have "
+        "already delivered, respond exactly NO_REPLY so no redundant "
+        "follow-up is sent to the user.",
         "",
     ]
     if isinstance(dispatched_at, (int, float)):