Speed up REPL autocomplete with threaded TTL caches

Before cold/warm: agent 418.722/1.116ms, set 5.819/2.841ms, slash 410.299/7.811ms, cd 0.005/0.003ms, model 2.095/0.733ms, plain 0.003/0.001ms. After prewarm first/second: agent 0.172/0.027ms, set 13.474/9.531ms, slash 0.108/0.046ms, cd 0.010/0.007ms, model 3.427/0.083ms, plain 0.005/0.003ms. Completion runs off-loop via ThreadedCompleter.

Author: mpfaffenberger

code_puppy/command_line/completion_cache.py

@@ -0,0 +1,37 @@
+"""Tiny thread-safe TTL caches for interactive completion lookups."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Callable
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class TTLCache(Generic[T]):
+    """Cache one value briefly, coalescing concurrent cold loads."""
+
+    def __init__(self, ttl: float = 4.0, clock: Callable[[], float] = time.monotonic):
+        self.ttl = ttl
+        self.clock = clock
+        self._value: T | None = None
+        self._deadline = 0.0
+        self._lock = threading.Lock()
+
+    def get(self, loader: Callable[[], T]) -> T:
+        now = self.clock()
+        if self._value is not None and now < self._deadline:
+            return self._value
+        with self._lock:
+            now = self.clock()
+            if self._value is None or now >= self._deadline:
+                self._value = loader()
+                self._deadline = now + self.ttl
+            return self._value
+
+    def clear(self) -> None:
+        with self._lock:
+            self._value = None
+            self._deadline = 0.0

code_puppy/command_line/model_picker_completion.py

@@ -11,6 +11,7 @@
 from termflow.tui.menu import MenuResult
 
 from code_puppy.callbacks import on_prompt_toolkit_style
+from code_puppy.command_line.completion_cache import TTLCache
 from code_puppy.command_line.menu_session import menu_session
 from code_puppy.command_line.tui_style import themed
 from code_puppy.command_line.utils import safe_input
@@ -27,14 +28,20 @@
 logger = logging.getLogger(__name__)
 
 MODEL_PICKER_PAGE_SIZE = 15
+_models_config_cache: TTLCache[dict] = TTLCache()
 
 
-def _load_models_config() -> dict:
+def _read_models_config() -> dict:
     from code_puppy.model_factory import ModelFactory
 
     return ModelFactory.load_config()
 
 
+def _load_models_config() -> dict:
+    """Return the merged model config, refreshing it after a short TTL."""
+    return _models_config_cache.get(_read_models_config)
+
+
 def load_model_names():
     """Load model names from the config that's fetched from the endpoint."""
     models_config = _load_models_config()

code_puppy/command_line/pin_command_completion.py

@@ -4,6 +4,11 @@
 from prompt_toolkit.completion import Completer, Completion
 from prompt_toolkit.document import Document
 
+from code_puppy.command_line.completion_cache import TTLCache
+
+_agent_names_cache: TTLCache[tuple[tuple[object, object], list[str]]] = TTLCache()
+_agent_meta_caches: dict[str, TTLCache[tuple[object, str]]] = {}
+
 
 def _get_json_agents_for_model(model_name: str) -> list:
     """Get JSON agents that have this model pinned in their JSON file."""
@@ -71,15 +76,28 @@ def _get_model_display_meta(model_name: str) -> str:
     return "Model"
 
 
-def _get_agent_display_meta(agent_name: str) -> str:
-    """Get display meta for an agent showing pinned model."""
+def _read_agent_display_meta(agent_name: str) -> str:
     pinned_model = _get_pinned_model_for_agent(agent_name)
     if pinned_model:
         return f"→ {pinned_model}"
     return "default"
 
 
-def load_agent_names():
+def _get_agent_display_meta(agent_name: str) -> str:
+    """Get briefly cached display meta for an agent's pinned model."""
+    cache = _agent_meta_caches.setdefault(agent_name, TTLCache())
+    source, meta = cache.get(
+        lambda: (_get_pinned_model_for_agent, _read_agent_display_meta(agent_name))
+    )
+    if source is not _get_pinned_model_for_agent:
+        cache.clear()
+        _, meta = cache.get(
+            lambda: (_get_pinned_model_for_agent, _read_agent_display_meta(agent_name))
+        )
+    return meta
+
+
+def _read_agent_names() -> list[str]:
     """Load all available agent names (both built-in and JSON agents)."""
     agents = set()
 
@@ -104,6 +122,21 @@ def load_agent_names():
     return sorted(list(agents))
 
 
+def load_agent_names() -> list[str]:
+    """Return available agent names, refreshing after a short TTL."""
+    from code_puppy.agents.agent_manager import get_agent_descriptions
+    from code_puppy.agents.json_agent import discover_json_agents
+
+    signature = (get_agent_descriptions, discover_json_agents)
+    cached_signature, names = _agent_names_cache.get(
+        lambda: (signature, _read_agent_names())
+    )
+    if cached_signature != signature:
+        _agent_names_cache.clear()
+        _, names = _agent_names_cache.get(lambda: (signature, _read_agent_names()))
+    return names
+
+
 def load_model_names():
     """Load model names from the config."""
     try: