diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index 93c2cfcf82816..03d7a1c9dd27d 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -473,6 +473,38 @@ def test_read_with_offset(self, registry): result = registry.read_log(s.id, offset=10, limit=5) assert "5 lines" in result["showing"] + def test_read_log_clamps_negative_limit(self, registry): + """Negative limit must not become lines[0:-N] (nearly the full buffer).""" + lines = "\n".join([f"line {i}" for i in range(50)]) + s = _make_session(output=lines) + registry._running[s.id] = s + result = registry.read_log(s.id, offset=0, limit=-5) + assert result["showing"] == "1 lines" + assert result["output"] == "line 49" + + def test_read_log_clamps_negative_offset(self, registry): + lines = "\n".join([f"line {i}" for i in range(20)]) + s = _make_session(output=lines) + registry._running[s.id] = s + result = registry.read_log(s.id, offset=-10, limit=5) + assert result["showing"] == "5 lines" + assert result["output"].splitlines() == [f"line {i}" for i in range(15, 20)] + + def test_read_log_clamps_huge_limit(self, registry): + lines = "\n".join([f"line {i}" for i in range(50)]) + s = _make_session(output=lines) + registry._running[s.id] = s + result = registry.read_log(s.id, offset=0, limit=10**9) + assert result["total_lines"] == 50 + assert result["showing"] == "50 lines" + + def test_read_log_clamps_zero_limit(self, registry): + lines = "\n".join([f"line {i}" for i in range(10)]) + s = _make_session(output=lines) + registry._running[s.id] = s + result = registry.read_log(s.id, offset=0, limit=0) + assert result["showing"] == "1 lines" + assert result["output"] == "line 9" # ========================================================================= # Stdin helpers diff --git a/tools/process_registry.py b/tools/process_registry.py index 29af51f000680..08f2992668bad 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1485,6 +1485,20 @@ def read_log(self, session_id: str, offset: int = 0, limit: int = 200) -> dict: """Read the full output log with optional pagination by lines.""" from tools.ansi_strip import strip_ansi + try: + offset = int(offset) + except (TypeError, ValueError): + offset = 0 + try: + limit = int(limit) + except (TypeError, ValueError): + limit = 200 + # Negative limit must not fall into ``lines[offset:offset+limit]`` + # with a negative stop (returns nearly the whole buffer). Huge + # limits dump the entire log into the tool result. + offset = max(0, offset) + limit = max(1, min(limit, 5000)) + session = self.get(session_id) if session is None: return {"status": "not_found", "error": f"No process with ID {session_id}"}