poll() must not mark completions as consumed

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50224 fix(process): keep poll read-only so notify_on_complete watcher fires (#10156)

  merged to main

  files: gateway/run.py, scripts/release.py, tests/gateway/test_internal_event_bypass_pairing.py, tests/tools/test_notify_on_complete.py, tools/process_registry.py

  Show the diff

  • gateway/run.py

    @@ -13136,7 +13136,9 @@ async def _run_process_watcher(self, watcher: dict) -> None:
     
                 if session.exited:
                     # --- Agent-triggered completion: inject synthetic message ---
    -                # Skip if the agent already consumed the result via wait/poll/log
    +                # Skip if the agent already consumed the result via wait/log.
    +                # poll() is read-only and intentionally does NOT mark consumed
    +                # (#10156) — a status check must not suppress this delivery turn.
                     from tools.process_registry import format_process_notification, process_registry as _pr_check
                     if agent_notify and not _pr_check.is_completion_consumed(session_id):
                         from tools.ansi_strip import strip_ansi

  • scripts/release.py

    @@ -45,6 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    +    "[redacted:email]": "de1tydev",  # PR #10158 salvage (poll read-only for notify_on_complete watcher; #10156)
         "[redacted:email]": "szzhoujiarui-sketch",  # cron model.default salvage co-author (#45550)
         "[redacted:email]": "rayjun",  # cron model.default salvage co-author (#43952)
         "[redacted:email]": "sweetcornna",  # cron ticker-liveness salvage co-author (#33849)

  • tests/gateway/test_internal_event_bypass_pairing.py

    @@ -17,6 +17,7 @@
     from gateway.platforms.base import MessageEvent
     from gateway.run import GatewayRunner
     from gateway.session import SessionSource
    +from tools.process_registry import ProcessRegistry, ProcessSession
     
     
     # ---------------------------------------------------------------------------
    @@ -99,6 +100,46 @@ async def _instant_sleep(*_a, **_kw):
         assert event.internal is True, "Synthetic completion event must be marked internal"
     
     
    [redacted:email]
    +async def test_poll_does_not_suppress_notify_on_complete_watcher(monkeypatch, tmp_path):
    +    """Regression: polling an exited process must not suppress watcher injection."""
    +    import tools.process_registry as pr_module
    +
    +    registry = ProcessRegistry()
    +    session = ProcessSession(
    +        id="proc_polled_completion",
    +        command="echo done",
    +        output_buffer="done\n",
    +        exited=True,
    +        exit_code=0,
    +        notify_on_complete=True,
    +    )
    +    registry._finished[session.id] = session
    +
    +    poll_result = registry.poll(session.id)
    +    assert poll_result["status"] == "exited"
    +    assert not registry.is_completion_consumed(session.id)
    +
    +    monkeypatch.setattr(pr_module, "process_registry", registry)
    +
    +    async def _instant_sleep(*_a, **_kw):
    +        pass
    +    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
    +
    +    runner = _build_runner(monkeypatch, tmp_path)
    +    adapter = runner.adapters[Platform.DISCORD]
    +
    +    watcher = _watcher_dict_with_notify()
    +    watcher["session_id"] = session.id
    +
    +    await runner._run_process_watcher(watcher)
    +
    +    assert adapter.handle_message.await_count == 1
    +    event = adapter.handle_message.await_args.args[0]
    +    assert session.id in event.text
    +    assert event.internal is True
    +
    +
     @pytest.mark.asyncio
     async def test_internal_event_bypasses_authorization(monkeypatch, tmp_path):
         """An internal event should skip _is_user_authorized entirely."""

  • tests/tools/test_notify_on_complete.py

    @@ -325,7 +325,7 @@ def test_notify_on_complete_blocked_in_sandbox(self):
     # =========================================================================
     
     class TestCompletionConsumed:
    -    """Test that wait/poll/log suppress redundant completion notifications."""
    +    """Test that wait/log consume completion notifications while poll stays read-only."""
     
         def test_wait_marks_completion_consumed(self, registry):
             """wait() returning exited status marks session as consumed."""
    @@ -347,8 +347,8 @@ def test_wait_marks_completion_consumed(self, registry):
             # Now the completion is marked as consumed
             assert registry.is_completion_consumed("proc_wait")
     
    -    def test_poll_marks_completion_consumed(self, registry):
    -        """poll() returning exited status marks session as consumed."""
    +    def test_poll_does_not_mark_completion_consumed(self, registry):
    +        """poll() is a read-only status check and must not suppress notify_on_complete."""
             s = _make_session(sid="proc_poll", notify_on_complete=True, output="done")
             s.exited = True
             s.exit_code = 0
    @@ -356,7 +356,7 @@ def test_poll_marks_completion_consumed(self, registry):
     
             result = registry.poll("proc_poll")
             assert result["status"] == "exited"
    -        assert registry.is_completion_consumed("proc_poll")
    +        assert not registry.is_completion_consumed("proc_poll")
     
         def test_log_marks_completion_consumed(self, registry):
             """read_log() on exited session marks as consumed."""
    @@ -378,6 +378,72 @@ def test_running_process_not_consumed(self, registry):
             assert result["status"] == "running"
             assert not registry.is_completion_consumed("proc_running")
     
    +    def test_poll_marks_poll_observed_for_cli_drain(self, registry):
    +        """poll() on an exited process records _poll_observed so the CLI drain
    +        dedups (the agent already saw the exit inline) without marking the
    +        session _completion_consumed (which would suppress the gateway watcher)."""
    +        s = _make_session(sid="proc_pobs", notify_on_complete=True, output="done")
    +        s.exited = True
    +        s.exit_code = 0
    +        registry._running[s.id] = s
    +        with patch.object(registry, "_write_checkpoint"):
    +            registry._move_to_finished(s)
    +
    +        # Completion is queued, nothing consumed/observed yet.
    +        assert not registry.completion_queue.empty()
    +        assert "proc_pobs" not in registry._poll_observed
    +        assert not registry.is_completion_consumed("proc_pobs")
    +
    +        # Agent polls inline — read-only, so NOT _completion_consumed, but the
    +        # exit was observed so the CLI drain must skip the queued completion.
    +        assert registry.poll("proc_pobs")["status"] == "exited"
    +        assert "proc_pobs" in registry._poll_observed
    +        assert not registry.is_completion_consumed("proc_pobs")
    +
    +        # CLI drain skips it → no duplicate [SYSTEM: ...] injection (#8228).
    +        drained = registry.drain_notifications()
    +        assert drained == []
    +
    +    def test_poll_observed_does_not_suppress_gateway_watcher(self, registry):
    +        """The gateway/tui watcher gate (is_completion_consumed) must stay False
    +        after a read-only poll, so the autonomous delivery turn still fires
    +        even though the CLI drain was deduped (#10156)."""
    +        s = _make_session(sid="proc_gw", notify_on_complete=True, output="done")
    +        s.exited = True
    +        s.exit_code = 0
    +        registry._finished[s.id] = s
    +
    +        registry.poll("proc_gw")
    +        # CLI-side dedup signal present...
    +        assert "proc_gw" in registry._poll_observed
    +        # ...but the gateway watcher gate is untouched, so it still delivers.
    +        assert not registry.is_completion_consumed("proc_gw")
    +
    +    def test_running_poll_does_not_mark_poll_observed(self, registry):
    +        """poll() on a still-running process must not record _poll_observed."""
    +        s = _make_session(sid="proc_run2", notify_on_complete=True, output="partial")
    +        registry._running[s.id] = s
    +
    +        registry.poll("proc_run2")
    +        assert "proc_run2" not in registry._poll_observed
    +
    +    def test_wait_and_log_still_skip_cli_drain(self, registry):
    +        """wait()/read_log() consume the output, so the CLI drain skips their
    +        completions via _completion_consumed (the original #8228 contract)."""
    +        for sid, action in (("proc_w", "wait"), ("proc_l", "log")):
    +            s = _make_session(sid=sid, notify_on_complete=True, output="done")
    +            s.exited = True
    +            s.exit_code = 0
    +            registry._running[s.id] = s
    +            with patch.object(registry, "_write_checkpoint"):
    +                registry._move_to_finished(s)
    +            if action == "wait":
    +                registry.wait(sid, timeout=1)
    +            else:
    +                registry.read_log(sid)
    +            assert registry.is_completion_consumed(sid)
    +        assert registry.drain_notifications() == []
    +
     
     # ---------------------------------------------------------------------------
     # Silent-background-process hint

  • tools/process_registry.py

    @@ -171,9 +171,21 @@ def __init__(self):
             self.completion_queue: _queue_mod.Queue = _queue_mod.Queue()
     
             # Track sessions whose completion was already consumed by the agent
    -        # via wait/poll/log.  Drain loops skip notifications for these.
    +        # via wait/log.  Drain loops AND gateway/tui watchers skip notifications
    +        # for these — a blocking wait() or a full read_log() means the agent
    +        # has the output in hand and is acting on it this turn.
             self._completion_consumed: set = set()
     
    +        # Track sessions the agent merely *observed* exited via poll().  poll()
    +        # is a read-only status check, so it does NOT mark _completion_consumed
    +        # (that would let a status check suppress the gateway/tui watcher's
    +        # autonomous delivery turn — #10156).  But on the CLI the poll result
    +        # is returned inline in the same turn, so the idle/post-turn drain must
    +        # still skip the queued completion to avoid a duplicate [SYSTEM: ...]
    +        # injection (the bug #8228 originally fixed).  drain_notifications()
    +        # consults this set; the gateway/tui watchers deliberately do NOT.
    +        self._poll_observed: set = set()
    +
             # Global watch-match circuit breaker — across all sessions.
             # Prevents sibling processes from collectively flooding the user even
             # when each stays under its own per-session cap.
    @@ -908,14 +920,28 @@ def _move_to_finished(self, session: ProcessSession):
         # ----- Query Methods -----
     
         def is_completion_consumed(self, session_id: str) -> bool:
    -        """Check if a completion notification was already consumed via wait/poll/log."""
    +        """Check if a completion notification was already consumed via wait/log."""
             return session_id in self._completion_consumed
     
    +    def _drain_should_skip(self, session_id: str) -> bool:
    +        """Whether the CLI drain should skip a completion event for this session.
    +
    +        Skips when the agent has either truly consumed the output (wait/log →
    +        ``_completion_consumed``) or observed the exit inline via poll()
    +        (``_poll_observed``).  In both cases the CLI agent already has the
    +        result this turn, so injecting a [SYSTEM: ...] completion would be a
    +        duplicate (#8228).  The gateway/tui watchers do NOT use this — they
    +        check only ``is_completion_consumed`` so a read-only poll never
    +        suppresses their autonomous delivery turn (#10156).
    +        """
    +        return session_id in self._completion_consumed or session_id in self._poll_observed
    +
         def drain_notifications(self) -> "list[tuple[dict, str]]":
             """Pop all pending notification events and return formatted pairs.
     
             Returns a list of (raw_event, formatted_text) tuples.
    -        Skips completion events that were already consumed via wait/poll/log.
    +        Skips completion events the agent already consumed via wait/log or
    +        observed inline via poll() (see ``_drain_should_skip``).
             """
             results = []
             while not self.completion_queue.empty():
    @@ -924,7 +950,7 @@ def drain_notifications(self) -> "list[tuple[dict, str]]":
                 except Exception:
                     break
                 _evt_sid = evt.get("session_id", "")
    -            if evt.get("type") == "completion" and self.is_completion_consumed(_evt_sid):
    +            if evt.get("type") == "completion" and self._drain_should_skip(_evt_sid):
                     continue
                 text = format_process_notification(evt)
                 if text:
    @@ -1038,7 +1064,17 @@ def poll(self, session_id: str) -> dict:
                 result["exit_code"] = session.exit_code
                 result["completion_reason"] = session.completion_reason
                 result["termination_source"] = session.termination_source
    -            self._completion_consumed.add(session_id)
    +            # NOTE: poll() is a read-only status query and deliberately does
    +            # NOT mark the session _completion_consumed. wait()/read_log()
    +            # represent actual output consumption and do mark it. Marking
    +            # consumed here would let a status check silently suppress the
    +            # notify_on_complete watcher's autonomous delivery turn (#10156).
    +            #
    +            # We DO record it in _poll_observed so the CLI's inline drain still
    +            # dedups (the agent already saw the exit in this turn's poll result)
    +            # without affecting the gateway/tui watchers, which only consult
    +            # _completion_consumed.
    +            self._poll_observed.add(session_id)
             if session.detached:
                 result["detached"] = True
                 result["note"] = "Process recovered after restart -- output history unavailable"
    @@ -1394,6 +1430,7 @@ def _prune_if_needed(self):
             for sid in expired:
                 del self._finished[sid]
                 self._completion_consumed.discard(sid)
    +            self._poll_observed.discard(sid)
     
             # If still over limit, remove oldest finished
             total = len(self._running) + len(self._finished)
    @@ -1401,14 +1438,19 @@ def _prune_if_needed(self):
                 oldest_id = min(self._finished, key=lambda sid: self._finished[sid].started_at)
                 del self._finished[oldest_id]
                 self._completion_consumed.discard(oldest_id)
    +            self._poll_observed.discard(oldest_id)
     
    -        # Drop any _completion_consumed entries whose sessions are no longer
    -        # tracked at all — belt-and-suspenders against module-lifetime growth
    -        # on process-registry lookup paths that don't reach the dict prunes.
    +        # Drop any _completion_consumed / _poll_observed entries whose sessions
    +        # are no longer tracked at all — belt-and-suspenders against
    +        # module-lifetime growth on registry lookup paths that don't reach the
    +        # dict prunes.
             tracked = self._running.keys() | self._finished.keys()
             stale = self._completion_consumed - tracked
             if stale:
                 self._completion_consumed -= stale
    +        stale_polls = self._poll_observed - tracked
    +        if stale_polls:
    +            self._poll_observed -= stale_polls
     
         # ----- Checkpoint (crash recovery) -----