diff --git a/agent/context_compressor.py b/agent/context_compressor.py index d66bee1cad6f..a5fe72d73e03 100644 --- a/agent/context_compressor.py +++ b/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( diff --git a/agent/title_generator.py b/agent/title_generator.py index 5323f8bd047a..9f889dbf6621 100644 --- a/agent/title_generator.py +++ b/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( diff --git a/gateway/run.py b/gateway/run.py index 6952198e200e..bf52bb8e7a8b 100644 --- a/gateway/run.py +++ b/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 "", 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 diff --git a/tests/agent/test_context_compressor_zero_user_provenance.py b/tests/agent/test_context_compressor_zero_user_provenance.py index 7a5bd1256869..a574280f16cc 100644 --- a/tests/agent/test_context_compressor_zero_user_provenance.py +++ b/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", [ diff --git a/tests/agent/test_title_generator.py b/tests/agent/test_title_generator.py index e4ae76b8aed1..98ab20ff7ad9 100644 --- a/tests/agent/test_title_generator.py +++ b/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() diff --git a/tests/gateway/test_background_process_notifications.py b/tests/gateway/test_background_process_notifications.py index 76941bb71064..a573817da4a7 100644 --- a/tests/gateway/test_background_process_notifications.py +++ b/tests/gateway/test_background_process_notifications.py @@ -613,3 +613,108 @@ def test_gateway_drain_retains_and_formats_overflow_events(): out_released = _format_gateway_process_notification(released) assert "notifications resumed" in out_released assert "exit code" not in out_released + + +# --------------------------------------------------------------------------- +# watch_match output/command get the same forced redaction floor as +# completion notifications, even when security.redact_secrets is off. +# --------------------------------------------------------------------------- + +def test_gateway_watch_match_force_redacts_output_when_redaction_disabled(monkeypatch): + """A user setting cannot disable the gateway's outbound secret floor for + watch_match — it must match completion notifications' force=True floor.""" + import agent.redact as redact_module + from gateway.run import _format_gateway_process_notification + + secret = "abc123randomopaquetokenvalue999" + monkeypatch.setattr(redact_module, "_REDACT_ENABLED", False) + + evt = { + "type": "watch_match", + "session_id": "proc_secret", + "pattern": "TOKEN", + "command": f"echo MY_SERVICE_TOKEN={secret}", + "output": f"MY_SERVICE_TOKEN={secret}\nHOME=/home/user", + "suppressed": 0, + } + + text = _format_gateway_process_notification(evt) + + assert secret not in text + assert "HOME=/home/user" in text + + +def test_process_registry_watch_match_force_redacts_output_when_redaction_disabled( + monkeypatch, +): + """Same floor for the shared tools/process_registry.py formatter used by + the TUI gateway surface. + + Uses a recognized credential prefix (not a bare KEY=value pair) because + the command here ("echo ...") is not an env-dump command, so + redact_terminal_output's code_file=True heuristic intentionally skips the + generic ENV-assignment pass regardless of the force floor (pre-existing, + unrelated to this fix) — prefix-matched secrets are unaffected by that. + """ + import agent.redact as redact_module + from tools.process_registry import format_process_notification + + secret = "sk-ant-api03-" + "x" * 40 + monkeypatch.setattr(redact_module, "_REDACT_ENABLED", False) + + evt = { + "type": "watch_match", + "session_id": "proc_secret", + "pattern": "TOKEN", + "command": "echo hello", + "output": f"Authorization token: {secret}\nHOME=/home/user", + "suppressed": 0, + } + + text = format_process_notification(evt) + + assert secret not in text + assert "HOME=/home/user" in text + + +def test_gateway_watch_match_and_watch_disabled_carry_subagent_attribution(): + """_format_gateway_process_notification used to hand-roll watch_match/ + watch_disabled formatting with no delegation attribution at all — a + SEPARATE implementation from tools.process_registry.format_process_ + notification's (which already attributes watch_match), so a subagent's + background process notification reached a gateway session anonymously + even when the CLI/TUI equivalent already attributed it correctly.""" + from tools.delegate_tool import _register_subagent, _unregister_subagent + from gateway.run import _format_gateway_process_notification + + sid = "sa-0-gwattr1" + _register_subagent({ + "subagent_id": sid, + "goal": "watch the build log for failures", + "delegation_id": "deleg_gwattr1", + }) + try: + match_evt = { + "type": "watch_match", + "session_id": "proc_gw1", + "task_id": sid, + "command": "tail -f build.log", + "pattern": "FAIL", + "output": "FAIL: build step 3", + "suppressed": 0, + } + disabled_evt = { + "type": "watch_disabled", + "session_id": "proc_gw1", + "task_id": sid, + "message": "Watch patterns disabled for process proc_gw1 — 3 consecutive rate-limit windows triggered.", + } + out_match = _format_gateway_process_notification(match_evt) + out_disabled = _format_gateway_process_notification(disabled_evt) + finally: + _unregister_subagent(sid) + + assert f"Started by subagent {sid}" in out_match + assert "watch the build log for failures" in out_match + assert f"Started by subagent {sid}" in out_disabled + assert "watch the build log for failures" in out_disabled diff --git a/tests/gateway/test_completion_session_boundary.py b/tests/gateway/test_completion_session_boundary.py index b01b0f6c1eeb..b5ea069cc480 100644 --- a/tests/gateway/test_completion_session_boundary.py +++ b/tests/gateway/test_completion_session_boundary.py @@ -16,6 +16,7 @@ import asyncio import json +import queue from collections import OrderedDict from types import SimpleNamespace from unittest.mock import AsyncMock @@ -299,6 +300,117 @@ def test_async_delegation_gate_unchanged(): adapter.handle_message.assert_not_awaited() +# --------------------------------------------------------------------------- +# Watch events (watch_match / watch_disabled) get the same boundary gate +# --------------------------------------------------------------------------- +# +# watch_match/watch_disabled historically bypassed _classify_completion_target +# entirely — _drain_watch_notifications called _inject_watch_notification +# directly, so a watch pattern match from a process spawned in session A +# could still land in session B's chat after /new. This mirrors the +# completion-type fix above for the watch-event family. + +def _watch_match_evt(parent_session_id=None, session_id="proc_watch"): + evt = { + "type": "watch_match", + "session_id": session_id, + "session_key": "agent:main:telegram:dm:123", + "platform": "telegram", + "chat_type": "dm", + "chat_id": "123", + "command": "tail -f app.log", + "pattern": "ERROR", + "output": "ERROR: something broke", + "suppressed": 0, + } + if parent_session_id is not None: + evt["parent_session_id"] = parent_session_id + return evt + + +def _drain(monkeypatch, runner, evt): + q = queue.Queue() + q.put(evt) + monkeypatch.setattr(runner, "_load_background_notifications_mode", lambda: "concise") + asyncio.run(runner._drain_watch_notifications(q)) + + +def test_watch_match_from_user_closed_session_is_dropped(monkeypatch): + """/new closed the spawning session -> the stamped watch_match must NOT + land in the chat's new session.""" + adapter = SimpleNamespace(handle_message=AsyncMock()) + runner = _runner( + adapter, + session_db=_SessionDB( + {"ended_at": 1786288000.0, "end_reason": "session_reset"} + ), + ) + + _drain(monkeypatch, runner, _watch_match_evt("sess-closed")) + + adapter.handle_message.assert_not_awaited() + + +def test_watch_match_from_live_session_delivers(monkeypatch): + adapter = SimpleNamespace(handle_message=AsyncMock()) + runner = _runner(adapter, session_db=_SessionDB({"ended_at": None})) + + _drain(monkeypatch, runner, _watch_match_evt("sess-live")) + + adapter.handle_message.assert_awaited_once() + + +def test_watch_disabled_from_user_closed_session_is_dropped(monkeypatch): + """The watch_disabled summary event carries the same stamp and must be + dropped by the same boundary check as watch_match.""" + adapter = SimpleNamespace(handle_message=AsyncMock()) + runner = _runner( + adapter, + session_db=_SessionDB( + {"ended_at": 1786288000.0, "end_reason": "session_reset"} + ), + ) + evt = { + "type": "watch_disabled", + "session_id": "proc_watch", + "session_key": "agent:main:telegram:dm:123", + "platform": "telegram", + "chat_type": "dm", + "chat_id": "123", + "suppressed": 3, + "message": "Watch patterns disabled for process proc_watch — 5 consecutive rate-limit windows triggered.", + "parent_session_id": "sess-closed", + } + + _drain(monkeypatch, runner, evt) + + adapter.handle_message.assert_not_awaited() + + +def test_unstamped_watch_match_delivers(monkeypatch): + """Legacy/global events without parent_session_id (e.g. the overflow + events, which have no single owning session) keep delivering + unconditionally, exactly like unstamped completion events do.""" + adapter = SimpleNamespace(handle_message=AsyncMock()) + runner = _runner(adapter, session_db=_SessionDB(None)) + + _drain(monkeypatch, runner, _watch_match_evt(None)) + + adapter.handle_message.assert_awaited_once() + + +def test_watch_match_retry_verdict_still_delivers(monkeypatch): + """Watch events have no watcher to re-poll them, unlike completions — + a transient retry verdict must fail open and deliver rather than lose + the match outright.""" + adapter = SimpleNamespace(handle_message=AsyncMock()) + runner = _runner(adapter, session_db=None) # no session_db -> "retry" + + _drain(monkeypatch, runner, _watch_match_evt("sess-uncertain")) + + adapter.handle_message.assert_awaited_once() + + # --------------------------------------------------------------------------- # The stamp survives checkpoint/restore # --------------------------------------------------------------------------- diff --git a/tests/tools/test_watch_patterns.py b/tests/tools/test_watch_patterns.py index 760e3ede6aa9..2f38fe3db91f 100644 --- a/tests/tools/test_watch_patterns.py +++ b/tests/tools/test_watch_patterns.py @@ -365,3 +365,68 @@ def test_overflow_released_formats_message(self): out = format_process_notification(evt) assert "notifications resumed" in out assert "exit code" not in out + + +class TestWatchDisabledSubagentAttribution: + """watch_disabled is a single-session, single-process event — same shape + as its watch_match sibling (session.id, session.command) — that fires + when a background process a subagent started trips WATCH_STRIKE_LIMIT. + Without task_id (and the formatter using it), the parent conversation + sees "Watch patterns disabled for process ..." with no hint it came from + a delegation, unlike watch_match, which already carries this.""" + + def test_watch_disabled_event_carries_task_id(self, registry): + s = _make_session(task_id="sa-0-watchdis1", watch_patterns=["E"]) + # Simulate WATCH_STRIKE_LIMIT - 1 strikes already accumulated across + # prior cooldown windows, so this one drop trips the limit and emits + # watch_disabled without needing to wait out multiple real windows. + s._watch_consecutive_strikes = WATCH_STRIKE_LIMIT - 1 + s._watch_cooldown_until = time.time() + 100 + registry._check_watch_patterns(s, "E hit\n") + + evt = None + while not registry.completion_queue.empty(): + candidate = registry.completion_queue.get_nowait() + if candidate.get("type") == "watch_disabled": + evt = candidate + assert evt is not None + assert evt["task_id"] == "sa-0-watchdis1" + + def test_watch_disabled_formats_without_attribution_for_parent_owned_process(self): + from tools.process_registry import format_process_notification + + evt = { + "type": "watch_disabled", + "session_id": "proc_parent1", + "task_id": "t1", + "message": "Watch patterns disabled for process proc_parent1 — 3 consecutive rate-limit windows triggered.", + } + out = format_process_notification(evt) + assert "Watch patterns disabled" in out + assert "Started by subagent" not in out + + def test_watch_disabled_formats_with_subagent_attribution(self): + from tools.delegate_tool import _register_subagent, _unregister_subagent + from tools.process_registry import format_process_notification + + sid = "sa-0-watchdisfmt1" + _register_subagent({ + "subagent_id": sid, + "goal": "tail the deploy log for errors", + "delegation_id": "deleg_watchdis1", + }) + try: + evt = { + "type": "watch_disabled", + "session_id": "proc_child1", + "task_id": sid, + "message": "Watch patterns disabled for process proc_child1 — 3 consecutive rate-limit windows triggered.", + } + out = format_process_notification(evt) + finally: + _unregister_subagent(sid) + + assert "Watch patterns disabled" in out + assert f"Started by subagent {sid}" in out + assert "of delegation deleg_watchdis1" in out + assert "tail the deploy log for errors" in out diff --git a/tools/process_registry.py b/tools/process_registry.py index b5fdf7704da6..738dcd8ada42 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -598,6 +598,7 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: self.completion_queue.put({ "session_id": session.id, "session_key": session.session_key, + "task_id": session.task_id, "command": session.command, "type": "watch_disabled", "suppressed": session._watch_suppressed, @@ -607,6 +608,12 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, + # Spawning-session boundary stamp — same field the + # "completion" type carries (#70300) so the gateway's + # session-boundary pre-flight can drop this if the + # spawning conversation was closed by an explicit user + # boundary (/new) before this event drained. + "parent_session_id": session.parent_session_id, "message": ( f"Watch patterns disabled for process {session.id} — " f"{WATCH_STRIKE_LIMIT} consecutive rate-limit windows triggered " @@ -641,6 +648,8 @@ def _check_watch_patterns(self, session: ProcessSession, new_text: str) -> None: "user_name": session.watcher_user_name, "thread_id": session.watcher_thread_id, "message_id": session.watcher_message_id, + # See the watch_disabled event above for why this is stamped. + "parent_session_id": session.parent_session_id, } _redact_process_result(notification) self.completion_queue.put(notification) @@ -2925,7 +2934,11 @@ def format_process_notification(evt: dict) -> "str | None": _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` — # without this case they fall through to the completion formatter and @@ -2935,7 +2948,16 @@ def format_process_notification(evt: dict) -> "str | None": if evt_type == "watch_match": _pat = evt.get("pattern", "?") - _out = evt.get("output", "") + # _check_watch_patterns already ran the producer-side, non-forced + # _redact_process_result pass, but that respects the configurable + # security.redact_secrets. This text is delivered straight to a chat + # surface (gateway platform adapter or TUI), so apply the forced, + # unconditional redactor here too — same defence-in-depth reasoning + # as the completion-notification path in gateway/run.py. + from agent.redact import redact_terminal_output + + _match_cmd = redact_terminal_output(_cmd, _cmd, force=True) + _out = redact_terminal_output(evt.get("output", ""), _cmd, force=True) _sup = evt.get("suppressed", 0) text = ( f"[IMPORTANT: Background process {_sid} matched " @@ -2944,7 +2966,7 @@ def format_process_notification(evt: dict) -> "str | None": if _attribution: text += f"{_attribution}\n" text += ( - f"Command: {_cmd}\n" + f"Command: {_match_cmd}\n" f"Matched output:\n{_out}" ) if _sup: