Clamp spawn_tree.list limit to avoid unbounded payloads

plugins/platforms/telegram/adapter.py

@@ -2161,7 +2161,9 @@ def _record_polling_progress(self, generation: int) -> None:
             return
         self._polling_progress_event.set()
         self._polling_network_error_count = 0
-        if generation == self._polling_conflict_recovery_generation:
+        # Bare/test adapters may not have run ``__init__``; treat missing as
+        # "no conflict recovery in flight" (same defensive shape as teardown).
+        if generation == getattr(self, "_polling_conflict_recovery_generation", None):
             self._polling_conflict_recovery_generation = None
         else:
             self._polling_conflict_count = 0

tests/gateway/test_telegram_start_polling_timeout.py

@@ -59,6 +59,7 @@ def _bare_adapter():
     a._fatal_error_retryable = True
     a._polling_network_error_count = 0
     a._polling_conflict_count = 0
+    a._polling_conflict_recovery_generation = None
     a._polling_error_callback_ref = None
     a._background_tasks = set()
     a._send_path_degraded = False

tests/tui_gateway/test_spawn_tree_list_limit_clamp.py

@@ -0,0 +1,72 @@
+"""Regression: spawn_tree.list must clamp limit before slicing results."""
+
+from __future__ import annotations
+
+import json
+
+from tui_gateway import server
+
+
+def _seed_index(tmp_path, monkeypatch, n: int) -> None:
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    session_dir = tmp_path / "spawn-trees" / "sess-test"
+    session_dir.mkdir(parents=True)
+    lines = []
+    for i in range(n):
+        snap = session_dir / f"20260101T{i:06d}.json"
+        snap.write_text("{}", encoding="utf-8")
+        lines.append(
+            json.dumps(
+                {
+                    "path": str(snap),
+                    "session_id": "sess-test",
+                    "started_at": float(i),
+                    "finished_at": float(1000 + i),
+                    "label": f"run-{i}",
+                    "count": 1,
+                }
+            )
+        )
+    (session_dir / "_index.jsonl").write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def _call(**params):
+    return server.handle_request(
+        {
+            "id": "1",
+            "method": "spawn_tree.list",
+            "params": {"session_id": "sess-test", **params},
+        }
+    )
+
+
+def test_spawn_tree_list_clamps_excessive_limit(tmp_path, monkeypatch):
+    _seed_index(tmp_path, monkeypatch, n=520)
+    resp = _call(limit=10_000_000)
+    assert "result" in resp
+    assert len(resp["result"]["entries"]) == 500
+
+
+def test_spawn_tree_list_clamps_negative_limit(tmp_path, monkeypatch):
+    """Negative limits must not flip ``entries[:limit]`` into nearly-all rows."""
+    _seed_index(tmp_path, monkeypatch, n=10)
+    resp = _call(limit=-5)
+    assert len(resp["result"]["entries"]) == 1
+
+
+def test_spawn_tree_list_clamps_zero_limit(tmp_path, monkeypatch):
+    _seed_index(tmp_path, monkeypatch, n=5)
+    resp = _call(limit=0)
+    assert len(resp["result"]["entries"]) == 1
+
+
+def test_spawn_tree_list_default_limit(tmp_path, monkeypatch):
+    _seed_index(tmp_path, monkeypatch, n=60)
+    resp = _call()
+    assert len(resp["result"]["entries"]) == 50
+
+
+def test_spawn_tree_list_invalid_limit_falls_back_to_default(tmp_path, monkeypatch):
+    _seed_index(tmp_path, monkeypatch, n=60)
+    resp = _call(limit="nope")
+    assert len(resp["result"]["entries"]) == 50

tui_gateway/methods_session.py

@@ -2906,7 +2906,19 @@ def _(rid, params: dict) -> dict:
 @method("spawn_tree.list")
 def _(rid, params: dict) -> dict:
     session_id = str(params.get("session_id") or "").strip()
-    limit = int(params.get("limit") or 50)
+    # Don't use ``or 50`` — falsy ``0`` must clamp to 1, not silently jump
+    # back to the default page size. Cap so a hostile/buggy client can't
+    # force returning huge payload lists, and so negative limits can't flip
+    # Python ``entries[:limit]`` into "return almost everything" semantics.
+    try:
+        raw_limit = params.get("limit", 50)
+        if raw_limit is None or raw_limit == "":
+            limit = 50
+        else:
+            limit = int(raw_limit)
+    except (TypeError, ValueError):
+        limit = 50
+    limit = max(1, min(limit, 500))
     cross_session = bool(params.get("cross_session"))
 
     if cross_session: