fix(gateway): close 3 watch-event gaps left by the completion-notification hardening

agent/context_compressor.py

@@ -192,6 +192,19 @@ def _is_summary_access_or_quota_error(exc: Exception) -> bool:
     "without calling any more tools."
 )
 _BACKGROUND_PROCESS_NOTIFICATION_PREFIX = "[IMPORTANT: Background process "
+# Sibling notification prefixes the block above did not cover: watch_disabled
+# and watch_overflow_* share the "[IMPORTANT: ...]" wrapper but not the
+# "Background process " text (watch_overflow is a cross-session summary with
+# no single owning process), and async-delegation completions use their own
+# "[ASYNC DELEGATION ...]" wrapper entirely. See
+# tools/process_registry.py's format_process_notification /
+# _format_async_delegation and gateway/run.py's
+# _format_gateway_process_notification for the exact producer text.
+_WATCH_DISABLED_NOTIFICATION_PREFIX = "[IMPORTANT: Watch patterns disabled"
+_WATCH_OVERFLOW_TRIPPED_PREFIX = "[IMPORTANT: Watch-pattern overflow:"
+_WATCH_OVERFLOW_RELEASED_PREFIX = "[IMPORTANT: Watch-pattern notifications resumed"
+_ASYNC_DELEGATION_COMPLETE_PREFIX = "[ASYNC DELEGATION COMPLETE"
+_ASYNC_DELEGATION_BATCH_COMPLETE_PREFIX = "[ASYNC DELEGATION BATCH COMPLETE"
 
 
 def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]:
@@ -5041,9 +5054,32 @@ def _is_synthetic_compression_user_turn(cls, message: Any) -> bool:
         if cls._has_compressed_summary_metadata(message):
             return True
         content = message.get("content")
+        text = _content_text_for_contains(content).strip()
+        # Async-delegation completions get the same internal=True /
+        # display_kind="internal_notification" stamping as every other
+        # background notification (gateway/run.py stamps it generically), but
+        # unlike watch/background-process bookkeeping they are NOT synthetic
+        # for compaction purposes: _format_async_delegation's own docstring
+        # says the block carries "the complete result summary" — genuine
+        # actionable content a real user turn would also carry. Excluding
+        # them here would let compaction blank out a delegation's actual
+        # result (test_completion_survives_compaction_verbatim_after_blank_echo,
+        # bc4824167d). They're still excluded from *titling*
+        # (title_generator.py), where the boilerplate wrapper text would make
+        # a bad title regardless of the payload.
+        is_async_delegation_notification = text.startswith(
+            _ASYNC_DELEGATION_COMPLETE_PREFIX
+        ) or text.startswith(_ASYNC_DELEGATION_BATCH_COMPLETE_PREFIX)
+        # display_kind survives SessionDB projection (it is a real column,
+        # not stripped underscore-prefixed metadata) and is the authoritative
+        # marker for background-process notifications persisted via
+        # gateway/run.py's internal-turn stamping. Checked narrowly for this
+        # one kind so model_switch/personality_switch markers (handled by
+        # their own dedicated recognizers) are unaffected.
+        if message.get("display_kind") == "internal_notification" and not is_async_delegation_notification:
+            return True
         if cls._is_context_summary_content(content):
             return True
-        text = _content_text_for_contains(content).strip()
         # Sibling recovery nudges from agent.conversation_loop's retry loop:
         # same "ephemeral scaffolding, not a real human turn" class as the
         # markers above (see _CODEX_INCOMPLETE_NUDGE's own docstring there),
@@ -5071,6 +5107,12 @@ def _is_synthetic_compression_user_turn(cls, message: Any) -> bool:
             _LENGTH_CONTINUATION_OUTPUT_LIMIT,
         } or text.startswith(
             _BACKGROUND_PROCESS_NOTIFICATION_PREFIX
+        ) or text.startswith(
+            _WATCH_DISABLED_NOTIFICATION_PREFIX
+        ) or text.startswith(
+            _WATCH_OVERFLOW_TRIPPED_PREFIX
+        ) or text.startswith(
+            _WATCH_OVERFLOW_RELEASED_PREFIX
         ) or text.startswith(
             TODO_INJECTION_HEADER + "\n"
         ) or text.startswith(

agent/title_generator.py

@@ -150,6 +150,20 @@
     # actual question. Keep in sync with
     # tui_gateway.server._MODEL_SWITCH_MARKER_PREFIX.
     "[System: The active model for this chat has changed to ",
+    # Background-process and async-delegation notifications injected via
+    # gateway/run.py's _inject_watch_notification (display_kind is the
+    # primary recognizer for these — see _is_real_user_turn — but callers
+    # here only have the opening message's raw text, e.g. a session whose
+    # very first persisted turn is a notification with no prior human ask).
+    # Text formats: tools/process_registry.py's format_process_notification /
+    # gateway/run.py's _format_gateway_process_notification and
+    # _format_async_delegation.
+    "[IMPORTANT: Background process ",
+    "[IMPORTANT: Watch patterns disabled",
+    "[IMPORTANT: Watch-pattern overflow:",
+    "[IMPORTANT: Watch-pattern notifications resumed",
+    "[ASYNC DELEGATION COMPLETE",
+    "[ASYNC DELEGATION BATCH COMPLETE",
 )
 
 
@@ -670,6 +684,14 @@ def _is_real_user_turn(message: Any) -> bool:
     """
     if not isinstance(message, dict) or message.get("role") != "user":
         return False
+    # display_kind is the authoritative, structural marker for background-
+    # process/async-delegation notifications persisted via gateway/run.py's
+    # internal-turn stamping (#82888) — checked directly here since this
+    # function (unlike is_titleable_user_message) has the full message dict.
+    # Narrow to this one kind so model_switch's own dedicated _MACHINE_PREFIXES
+    # entry is unaffected.
+    if message.get("display_kind") == "internal_notification":
+        return False
     content = message.get("content")
 
     return is_titleable_user_message(

gateway/run.py

@@ -3803,9 +3803,22 @@ def _format_gateway_process_notification(evt: dict) -> "str | None":
     evt_type = evt.get("type", "completion")
     _sid = evt.get("session_id", "unknown")
     _cmd = evt.get("command", "unknown")
+    # Subagent-owned watch events reach a gateway session anonymously without
+    # this — same "Started by subagent ..." provenance line the shared
+    # formatter in tools/process_registry.py already attaches. Computed here
+    # too (not delegated wholesale) so this branch keeps its own gateway-
+    # specific _redact_gateway_user_facing_secrets floor rather than silently
+    # swapping to the shared formatter's redactor.
+    from tools.process_registry import _delegation_attribution_line
+
+    _attribution = _delegation_attribution_line(evt)
 
     if evt_type == "watch_disabled":
-        return f"[IMPORTANT: {evt.get('message', '')}]"
+        text = f"[IMPORTANT: {evt.get('message', '')}"
+        if _attribution:
+            text += f"\n{_attribution}"
+        text += "]"
+        return text
 
     # Overflow events carry their human-readable summary in `message`,
     # like watch_disabled — see the shared formatter in
@@ -3815,12 +3828,23 @@ def _format_gateway_process_notification(evt: dict) -> "str | None":
 
     if evt_type == "watch_match":
         _pat = evt.get("pattern", "?")
-        _out = evt.get("output", "")
+        # The producer-side redaction in _check_watch_patterns respects
+        # security.redact_secrets (deliberately configurable). This event
+        # goes straight to the platform adapter like a completion
+        # notification does, so apply the same unconditional gateway floor
+        # here as defence in depth (matches the reasoning in
+        # _format_coalesced_process_completions / _run_process_watcher).
+        _out = _redact_gateway_user_facing_secrets(str(evt.get("output", "")))
+        _match_cmd = _redact_gateway_user_facing_secrets(str(_cmd))
         _sup = evt.get("suppressed", 0)
         text = (
             f"[IMPORTANT: Background process {_sid} matched "
             f"watch pattern \"{_pat}\".\n"
-            f"Command: {_cmd}\n"
+        )
+        if _attribution:
+            text += f"{_attribution}\n"
+        text += (
+            f"Command: {_match_cmd}\n"
             f"Matched output:\n{_out}"
         )
         if _sup:
@@ -24746,6 +24770,33 @@ async def _drain_watch_notifications(self, completion_queue) -> None:
             return
 
         for evt in watch_events:
+            # Same spawning-session-boundary pre-flight completion/
+            # async_delegation events get in _deliver_completion_notification
+            # (#70300): a stamped parent_session_id that resolves to a
+            # permanently-gone session (explicit /new boundary) means this
+            # event's own conversation is dead, so route the notification
+            # nowhere rather than injecting it into whatever session now
+            # occupies the same session_key. Unstamped (legacy/global
+            # overflow) events have no single owning session and keep
+            # delivering unconditionally, matching completion's own
+            # unstamped-event fallback.
+            parent_session_id = str(evt.get("parent_session_id") or "").strip()
+            if parent_session_id:
+                verdict = await self._classify_completion_target(parent_session_id)
+                if verdict == "terminal":
+                    logger.warning(
+                        "Watch notification for process %s targets "
+                        "permanently-gone session %s (user boundary such as "
+                        "/new); dropping notification.",
+                        evt.get("session_id") or "<unknown>", parent_session_id,
+                    )
+                    continue
+                # "retry" is a narrow, transient uncertainty (session DB
+                # unavailable, or a compression rotation caught mid-flight).
+                # Watch events have no watcher to re-poll them later — unlike
+                # completion notifications, this is the only chance to
+                # deliver — so fail open and inject rather than losing the
+                # match outright.
             synth_text = _format_gateway_process_notification(evt)
             if not synth_text:
                 continue

tests/agent/test_context_compressor_zero_user_provenance.py

@@ -261,6 +261,41 @@ def test_real_task_wins_over_trailing_max_iterations_nudge(compressor):
             },
             id="watch_match",
         ),
+        pytest.param(
+            {
+                "type": "watch_disabled",
+                "session_id": "proc_server",
+                "message": (
+                    "Watch patterns disabled for process proc_server — 5 "
+                    "consecutive rate-limit windows triggered (min spacing "
+                    "30s). Falling back to notify_on_complete semantics; "
+                    "you'll get exactly one notification when the process "
+                    "exits."
+                ),
+            },
+            id="watch_disabled",
+        ),
+        pytest.param(
+            {
+                "type": "watch_overflow_tripped",
+                "message": (
+                    "Watch-pattern overflow: >40 notifications in 60s "
+                    "across all processes. Suppressing further watch_match "
+                    "events for 120s."
+                ),
+            },
+            id="watch_overflow_tripped",
+        ),
+        pytest.param(
+            {
+                "type": "watch_overflow_released",
+                "message": (
+                    "Watch-pattern notifications resumed. 12 match "
+                    "event(s) were suppressed during the flood."
+                ),
+            },
+            id="watch_overflow_released",
+        ),
     ],
 )
 def test_background_process_notifications_do_not_become_compaction_anchors(
@@ -284,6 +319,57 @@ def test_background_process_notifications_do_not_become_compaction_anchors(
     assert compressor._find_last_user_message_idx(messages, head_end=0) == 0
 
 
+def test_internal_notification_display_kind_is_synthetic_regardless_of_text():
+    """gateway/run.py stamps display_kind='internal_notification' on these
+    rows at persist time (#82888). That structural marker must be recognized
+    on its own, independent of the text-prefix matching above — proven with
+    unrelated-looking content."""
+    tagged = {
+        "role": "user",
+        "content": "some future notification wording we haven't seen yet",
+        "display_kind": "internal_notification",
+    }
+    assert ContextCompressor._is_synthetic_compression_user_turn(tagged) is True
+
+    untagged = {
+        "role": "user",
+        "content": "some future notification wording we haven't seen yet",
+    }
+    assert ContextCompressor._is_synthetic_compression_user_turn(untagged) is False
+
+
+@pytest.mark.parametrize(
+    "notification_text",
+    [
+        pytest.param(
+            "[ASYNC DELEGATION COMPLETE — deleg-1]\nOriginal goal: fix the flaky "
+            "test\n\nResult: found the race condition in session setup.",
+            id="single",
+        ),
+        pytest.param(
+            "[ASYNC DELEGATION BATCH COMPLETE — deleg-2]\nA background fan-out "
+            "finished.\n\n--- ✓ TASK 1/1: fix the flaky test "
+            "(status=completed) ---\nfound the race condition in session setup.",
+            id="batch",
+        ),
+    ],
+)
+def test_async_delegation_completion_is_not_synthetic(notification_text):
+    """Unlike watch/background-process bookkeeping, an async-delegation
+    completion carries a real result (format_process_notification's own
+    docstring: "the complete result summary") and must remain eligible as a
+    compaction anchor — regression guard for the conflict this exact prefix
+    check caused with test_completion_survives_compaction_verbatim_after_blank_echo
+    (bc4824167d) the first time it was added."""
+    # Real production rows get display_kind stamped alongside the text
+    # (gateway/run.py, #82888); both must be tolerated.
+    for message in (
+        {"role": "user", "content": notification_text},
+        {"role": "user", "content": notification_text, "display_kind": "internal_notification"},
+    ):
+        assert ContextCompressor._is_synthetic_compression_user_turn(message) is False
+
+
 @pytest.mark.parametrize(
     "content",
     [

tests/agent/test_title_generator.py

@@ -597,3 +597,74 @@ def test_instant_title_skips_marker_uses_real_message(self):
         assert apply_instant_title(db, "sess-1", "南京市秦淮区 小时级天气预报") == (
             "南京市秦淮区 小时级天气预报"
         )
+
+
+class TestInternalNotificationNotTitleable:
+    """Background-process/async-delegation notifications (display_kind=
+    'internal_notification', #82888) must not title a session either — same
+    "machinery persisted under role=user" class as the model-switch marker
+    above, previously unrecognized by both the display_kind check in
+    _is_real_user_turn and the text-prefix list in _MACHINE_PREFIXES.
+    """
+
+    def test_is_real_user_turn_false_for_internal_notification_display_kind(self):
+        """The structural display_kind marker is authoritative, independent
+        of the message text — proven with unrelated-looking content."""
+        from agent.title_generator import _is_real_user_turn
+
+        assert not _is_real_user_turn(
+            {
+                "role": "user",
+                "content": "fix the login button",
+                "display_kind": "internal_notification",
+            }
+        )
+
+    def test_ordinary_user_turn_with_no_display_kind_still_counts(self):
+        from agent.title_generator import _is_real_user_turn
+
+        assert _is_real_user_turn(
+            {"role": "user", "content": "fix the login button"}
+        )
+
+    @pytest.mark.parametrize(
+        "opener",
+        [
+            "[IMPORTANT: Background process proc_1 matched watch pattern "
+            '"ERROR".\nCommand: tail -f app.log\nMatched output:\nERROR: boom]',
+            "[IMPORTANT: Watch patterns disabled for process proc_1 — 5 "
+            "consecutive rate-limit windows triggered (min spacing 30s). "
+            "Falling back to notify_on_complete semantics; you'll get "
+            "exactly one notification when the process exits.]",
+            "[IMPORTANT: Watch-pattern overflow: >40 notifications in 60s "
+            "across all processes. Suppressing further watch_match events "
+            "for 120s.]",
+            "[IMPORTANT: Watch-pattern notifications resumed. 12 match "
+            "event(s) were suppressed during the flood.]",
+            "[ASYNC DELEGATION COMPLETE — deleg-1]\nA background subagent "
+            "you dispatched earlier has finished.",
+            "[ASYNC DELEGATION BATCH COMPLETE — deleg-2]\nA background "
+            "fan-out of 3 subagent(s) you dispatched earlier has finished.",
+        ],
+    )
+    def test_notification_openers_are_not_titleable(self, opener):
+        """A session whose very first persisted turn is one of these
+        notifications (no display_kind available to the raw-string callers)
+        must still not be titled from it."""
+        from agent.title_generator import is_titleable_user_message
+
+        assert is_titleable_user_message(opener) is False
+
+    def test_skips_watch_match_opening_message(self, tmp_path):
+        """End-to-end: maybe_auto_title must not name a session after a
+        watch_match notification that happens to be its opening turn."""
+        db = SessionDB(tmp_path / "state.db")
+        db.create_session(session_id="sess-1", source="cli")
+        opener = (
+            '[IMPORTANT: Background process proc_1 matched watch pattern "ERROR".\n'
+            "Command: tail -f app.log\nMatched output:\nERROR: boom]"
+        )
+        with patch("agent.title_generator.auto_title_session") as mock_auto:
+            maybe_auto_title(db, "sess-1", opener, [])
+        assert db.get_session_title("sess-1") is None
+        mock_auto.assert_not_called()