Optimize log buffer slicing for live logs

app/log_buffer.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from typing import List, Literal
 from collections import deque
+from itertools import islice
 
 LogLevel = Literal["init", "llm", "start", "tick", "action", "sl_tp", "stop", "error", "warn", "supervisor"]
 
@@ -31,11 +32,16 @@ def clear() -> None:
 
 
 def get_logs(limit: int = 500) -> List[dict]:
-    """Return last `limit` log entries, newest last."""
-    items = list(_buffer)
-    if limit and len(items) > limit:
-        items = items[-limit:]
-    return items
+    """Return last `limit` log entries, newest last.
+
+    Avoid materializing the full ring buffer when callers only need a small
+    tail slice for the live logs endpoint.
+    """
+    if not limit or limit >= len(_buffer):
+        return list(_buffer)
+
+    start = max(len(_buffer) - max(limit, 0), 0)
+    return list(islice(_buffer, start, None))
 
 
 def get_logs_response(running: bool) -> dict:

tests/test_log_buffer.py

@@ -0,0 +1,23 @@
+from app import log_buffer
+
+
+def setup_function():
+    log_buffer.clear()
+
+
+def test_get_logs_returns_only_requested_tail():
+    for index in range(10):
+        log_buffer.emit("tick", f"tick-{index}")
+
+    logs = log_buffer.get_logs(limit=3)
+
+    assert [entry["message"] for entry in logs] == ["tick-7", "tick-8", "tick-9"]
+
+
+def test_get_logs_keeps_existing_unlimited_behavior():
+    for index in range(3):
+        log_buffer.emit("tick", f"tick-{index}")
+
+    logs = log_buffer.get_logs(limit=0)
+
+    assert [entry["message"] for entry in logs] == ["tick-0", "tick-1", "tick-2"]