refactor(gateway): TurnContext/TurnRunner seam — extract _run_agent_inner nested closures (byte-identical bodies)

gateway/turn_context.py

@@ -0,0 +1,66 @@
+"""Per-turn context shared between ``GatewayRunner._run_agent_inner`` and the
+``TurnRunner`` collaborator (gateway/run.py).
+
+``_run_agent_inner`` historically defined its tool-progress plumbing as nested
+closures (``progress_callback`` ~250 LOC, ``send_progress_messages`` ~353 LOC)
+that closed over ~20 enclosing locals.  ``TurnContext`` is the extraction seam:
+each closed-over local becomes a field on this dataclass, so the closure bodies
+can move onto ``TurnRunner`` methods unchanged modulo ``name`` -> ``ctx.name``
+rewrites.
+
+Field notes:
+
+- All fields are written once by ``_run_agent_inner`` while wiring up the turn
+  (a few — ``_progress_metadata``, ``_progress_reply_to``, ``agent_holder`` —
+  are computed slightly later than construction and assigned onto the ctx as
+  soon as the original locals were bound).  None of the original closures
+  *rebound* their captured names (no ``nonlocal``); mutable state uses the
+  same single-element-list containers as before (``last_progress_msg``,
+  ``repeat_count``, ...), so mutation stays visible to the outer body through
+  the shared objects exactly as it did through the shared closure cells.
+- ``_run_still_current`` stays a callable (it captures ``self``/
+  ``session_key``/``run_generation``); carrying the callable keeps the
+  extracted bodies byte-identical.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable, List, Optional
+
+
+@dataclass
+class TurnContext:
+    """Closed-over locals of ``_run_agent_inner`` needed by ``TurnRunner``."""
+
+    # --- read-only turn identity / wiring -------------------------------
+    source: Any = None
+    _run_still_current: Callable[[], bool] = None  # type: ignore[assignment]
+    _live_status_adapter: Any = None
+    _live_status_mode: str = "off"
+    _thinking_enabled: bool = False
+    progress_mode: str = "off"
+    progress_grouping: str = "grouped"
+    tool_progress_enabled: bool = False
+
+    # --- queues ----------------------------------------------------------
+    progress_queue: Any = None
+    log_queue: Any = None
+
+    # --- mutable single-element containers (shared with the outer body) --
+    last_progress_msg: list = field(default_factory=lambda: [None])
+    last_tool: list = field(default_factory=lambda: [None])
+    last_was_terminal_block: list = field(default_factory=lambda: [False])
+    repeat_count: list = field(default_factory=lambda: [0])
+    long_tool_hint_fired: list = field(default_factory=lambda: [False])
+    agent_holder: list = field(default_factory=lambda: [None])
+
+    # --- constants / cleanup bookkeeping ---------------------------------
+    _LONG_TOOL_THRESHOLD_S: float = 30.0
+    _cleanup_progress: bool = False
+    _cleanup_msg_ids: List[str] = field(default_factory=list)
+
+    # --- progress threading metadata (assigned after construction, before
+    #     send_progress_messages is scheduled) ----------------------------
+    _progress_metadata: Optional[dict] = None
+    _progress_reply_to: Optional[Any] = None

tests/gateway/test_turn_context.py

@@ -0,0 +1,66 @@
+"""Unit tests for the TurnContext/TurnRunner seam extracted from
+``GatewayRunner._run_agent_inner`` (gateway/turn_context.py + gateway/run.py).
+
+The extraction contract: the closure bodies moved onto ``TurnRunner`` methods
+byte-identically (modulo local -> ctx.field rewrites), with every closed-over
+local carried as a ``TurnContext`` field. These tests pin the seam's wiring —
+shared mutable containers, no-queue early returns — not the progress behavior
+itself (that's covered by test_run_progress_topics.py et al.).
+"""
+
+import asyncio
+import queue as queue_mod
+
+import pytest
+
+from gateway.turn_context import TurnContext
+
+
+def _make_runner(ctx):
+    from gateway.run import TurnRunner
+
+    class _StubGatewayRunner:
+        def _adapter_for_source(self, source):
+            return None
+
+    return TurnRunner(_StubGatewayRunner(), ctx)
+
+
+class TestTurnContext:
+    def test_defaults_are_independent_containers(self):
+        a, b = TurnContext(), TurnContext()
+        a.last_progress_msg[0] = "x"
+        a.repeat_count[0] = 3
+        a._cleanup_msg_ids.append("1")
+        assert b.last_progress_msg == [None]
+        assert b.repeat_count == [0]
+        assert b._cleanup_msg_ids == []
+
+    def test_shared_containers_visible_to_outer_scope(self):
+        # The outer body and the runner share the SAME list objects, so
+        # mutation through the ctx is visible to locals captured elsewhere.
+        last_progress_msg = [None]
+        ctx = TurnContext(last_progress_msg=last_progress_msg)
+        ctx.last_progress_msg[0] = "🔍 web_search"
+        assert last_progress_msg[0] == "🔍 web_search"
+
+
+class TestTurnRunner:
+    def test_methods_exist_and_bind(self):
+        from gateway.run import TurnRunner
+
+        ctx = TurnContext()
+        runner = _make_runner(ctx)
+        assert callable(runner.progress_callback)
+        assert asyncio.iscoroutinefunction(TurnRunner.send_progress_messages)
+        assert runner._ctx is ctx
+
+    def test_send_progress_messages_no_queue_returns(self):
+        ctx = TurnContext(progress_queue=None)
+        runner = _make_runner(ctx)
+        assert asyncio.run(runner.send_progress_messages()) is None
+
+    def test_send_progress_messages_no_adapter_returns(self):
+        ctx = TurnContext(progress_queue=queue_mod.Queue())
+        runner = _make_runner(ctx)  # stub adapter resolver returns None
+        assert asyncio.run(runner.send_progress_messages()) is None