From f5444e933adf6d8ce3d8a5cb067a61be5d0701ca Mon Sep 17 00:00:00 2001
From: Tom Qiao <zqiao@microsoft.com>
Date: Sun, 12 Apr 2026 09:02:44 +0800
Subject: [PATCH] fix: update activity tracker during Codex Responses streaming

The Codex Responses stream paths (_run_codex_stream and
_run_codex_create_stream_fallback) never called _touch_activity(),
leaving the cron inactivity tracker stale at "starting API call #N"
even while stream events were actively flowing. This caused cron jobs
to be killed by the inactivity timeout during long Codex responses.

Add _touch_activity() calls on stream entry, text deltas, reasoning
deltas, and in the fallback stream loop.

Fixes #7794

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 run_agent.py                                  |  4 ++
 .../test_run_agent_codex_responses.py         | 50 +++++++++++++++++++
 2 files changed, 54 insertions(+)

diff --git a/run_agent.py b/run_agent.py
index 3956f89048805..7118e35d35df5 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -4240,6 +4240,7 @@ def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta
             collected_output_items: list = []
             try:
                 with active_client.responses.stream(**api_kwargs) as stream:
+                    self._touch_activity("receiving Codex stream response")
                     for event in stream:
                         if self._interrupt_requested:
                             break
@@ -4249,6 +4250,7 @@ def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta
                             delta_text = getattr(event, "delta", "")
                             if delta_text:
                                 self._codex_streamed_text_parts.append(delta_text)
+                                self._touch_activity("receiving Codex stream response")
                             if delta_text and not has_tool_calls:
                                 if not first_delta_fired:
                                     first_delta_fired = True
@@ -4266,6 +4268,7 @@ def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta
                             reasoning_text = getattr(event, "delta", "")
                             if reasoning_text:
                                 self._fire_reasoning_delta(reasoning_text)
+                                self._touch_activity("receiving Codex reasoning")
                         # Collect completed output items — some backends
                         # (chatgpt.com/backend-api/codex) stream valid items
                         # via response.output_item.done but the SDK's
@@ -4368,6 +4371,7 @@ def _run_codex_create_stream_fallback(self, api_kwargs: dict, client: Any = None
                 event_type = getattr(event, "type", None)
                 if not event_type and isinstance(event, dict):
                     event_type = event.get("type")
+                self._touch_activity("receiving Codex fallback stream response")
 
                 # Collect output items and text deltas for backfill
                 if event_type == "response.output_item.done":
diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py
index 533a85ac83511..1a848916d0310 100644
--- a/tests/run_agent/test_run_agent_codex_responses.py
+++ b/tests/run_agent/test_run_agent_codex_responses.py
@@ -1197,3 +1197,53 @@ def test_preflight_codex_input_deduplicates_reasoning_ids(monkeypatch):
     assert reasoning_ids.count("rs_xyz") == 1
     assert reasoning_ids.count("rs_zzz") == 1
     assert len(reasoning_items) == 2
+
+
+def test_codex_stream_updates_activity_tracker(monkeypatch):
+    """Regression for #7794: Codex stream events must update _touch_activity."""
+    agent = _build_agent(monkeypatch)
+
+    text_event = SimpleNamespace(type="response.output_text.delta", delta="hello")
+    reasoning_event = SimpleNamespace(type="response.reasoning.delta", delta="thinking")
+
+    class _StreamWithEvents:
+        def __enter__(self):
+            return self
+        def __exit__(self, *a):
+            return False
+        def __iter__(self):
+            yield text_event
+            yield reasoning_event
+        def get_final_response(self):
+            return _codex_message_response("done")
+
+    agent.client = SimpleNamespace(
+        responses=SimpleNamespace(stream=lambda **kw: _StreamWithEvents())
+    )
+
+    initial_ts = agent._last_activity_ts
+    agent._run_codex_stream(_codex_request_kwargs())
+    assert agent._last_activity_ts > initial_ts
+    assert "Codex" in agent._last_activity_desc
+
+
+def test_codex_fallback_stream_updates_activity_tracker(monkeypatch):
+    """Regression for #7794: Codex fallback stream must also update activity."""
+    agent = _build_agent(monkeypatch)
+
+    events = [
+        SimpleNamespace(type="response.output_text.delta", delta="hi"),
+        SimpleNamespace(
+            type="response.completed",
+            response=_codex_message_response("result"),
+        ),
+    ]
+
+    agent.client = SimpleNamespace(
+        responses=SimpleNamespace(create=lambda **kw: iter(events))
+    )
+
+    initial_ts = agent._last_activity_ts
+    agent._run_codex_create_stream_fallback(_codex_request_kwargs())
+    assert agent._last_activity_ts > initial_ts
+    assert "Codex" in agent._last_activity_desc