Throttle and schedule buffered output delivery

tests/test_bridge.py

@@ -61,7 +61,7 @@ class FakePtyManager:
     def send(self, text: str) -> None:
         self.sent_inputs.append(text)
 
-    def read_output(self, timeout: int = 5) -> str:
+    def read_output(self, timeout: float = 5) -> str:
         return ""
 
 
@@ -297,3 +297,105 @@ def test_send_output_chunks_keeps_non_tmux_status_like_lines(monkeypatch) -> Non
         ("C1", "```\n[2,3] \"job-runner\" 05:12 17-Feb-26\n```"),
         ("C1", "```\n[2,3] \"job-runner\" 05:13 17-Feb-26\n```"),
     ]
+
+
+def test_should_flush_output_buffer_when_settled(monkeypatch) -> None:
+    bridge = _make_bridge(monkeypatch)
+    bridge.config.output_settle_seconds = 4.0
+    bridge.config.output_flush_interval_seconds = 15.0
+    bridge._last_output_at = 10.0
+    bridge._output_buffer_started_at = 2.0
+
+    assert bridge._should_flush_output_buffer(14.1) is True
+
+
+def test_should_flush_output_buffer_when_flush_interval_elapsed(monkeypatch) -> None:
+    bridge = _make_bridge(monkeypatch)
+    bridge.config.output_settle_seconds = 4.0
+    bridge.config.output_flush_interval_seconds = 15.0
+    bridge._last_output_at = 20.0
+    bridge._output_buffer_started_at = 2.0
+
+    assert bridge._should_flush_output_buffer(17.1) is True
+
+
+def test_should_flush_output_buffer_false_during_active_stream(monkeypatch) -> None:
+    bridge = _make_bridge(monkeypatch)
+    bridge.config.output_settle_seconds = 4.0
+    bridge.config.output_flush_interval_seconds = 15.0
+    bridge._last_output_at = 19.0
+    bridge._output_buffer_started_at = 10.0
+
+    assert bridge._should_flush_output_buffer(20.0) is False
+
+
+def test_poll_output_skips_final_flush_after_intentional_stop(monkeypatch) -> None:
+    bridge = _make_bridge(monkeypatch)
+    bridge._channel = "C1"
+    bridge.config.output_settle_seconds = 9999.0
+    bridge.config.output_flush_interval_seconds = 9999.0
+    bridge.config.output_buffer_interval = 0.0
+
+    pty = FakePtyManager("codex-room", cli_name="codex")
+    pty._alive = True
+    bridge.pty = pty
+    bridge._running = True
+
+    sent_buffers: list[str] = []
+    monkeypatch.setattr(bridge, "_send_output_chunks", sent_buffers.append)
+
+    def _read_output(timeout: float = 5) -> str:
+        bridge._running = False
+        return "planning update"
+
+    monkeypatch.setattr(pty, "read_output", _read_output)
+    bridge._poll_output()
+
+    assert sent_buffers == []
+
+
+def test_next_read_timeout_is_capped_by_flush_deadline(monkeypatch) -> None:
+    bridge = _make_bridge(monkeypatch)
+    bridge.config.pty_read_timeout = 5
+    bridge.config.output_settle_seconds = 4.0
+    bridge.config.output_flush_interval_seconds = 15.0
+    bridge._last_output_at = 100.0
+    bridge._output_buffer_started_at = 95.0
+
+    timeout = bridge._next_read_timeout(103.6, has_buffer=True)
+    assert 0.39 <= timeout <= 0.41
+
+
+def test_poll_output_uses_shorter_timeout_near_settle_deadline(monkeypatch) -> None:
+    bridge = _make_bridge(monkeypatch)
+    bridge._channel = "C1"
+    bridge.config.pty_read_timeout = 5
+    bridge.config.output_settle_seconds = 4.0
+    bridge.config.output_flush_interval_seconds = 15.0
+    bridge.config.output_buffer_interval = 0.0
+    bridge.config.output_idle_report_seconds = 0
+    bridge.config.input_idle_report_seconds = 0
+
+    pty = FakePtyManager("codex-room", cli_name="codex")
+    pty._alive = True
+    bridge.pty = pty
+    bridge._running = True
+
+    observed_timeouts: list[float] = []
+    call_count = 0
+
+    def _read_output(timeout: float = 5) -> str:
+        nonlocal call_count
+        observed_timeouts.append(timeout)
+        if call_count == 0:
+            call_count += 1
+            return "first chunk"
+        bridge._running = False
+        return ""
+
+    monkeypatch.setattr(pty, "read_output", _read_output)
+    bridge._poll_output()
+
+    assert len(observed_timeouts) == 2
+    assert observed_timeouts[0] == 5
+    assert 0.0 <= observed_timeouts[1] < 5