Add utility smoke coverage and response cache

Author: staradigm

config/includes.chroot/usr/local/bin/neuros-draw

@@ -349,7 +349,6 @@ def main():
 
     sub = parser.add_subparsers(dest="subcommand", help="Subcommands")
 
-    sub.add_parser("gallery", aliases=["list", "ls"], help="Browse generated images")
     p_gal = sub.add_parser("gallery", aliases=["list", "ls"], help="Browse generated images")
     p_gal.add_argument("--open", "-o", action="store_true", help="Open latest image")
 

config/includes.chroot/usr/local/bin/neuros-edit

@@ -378,7 +378,7 @@ def main():
     parser.add_argument("--diff", action="store_true", help="Show diff from backup")
     parser.add_argument("--batch", help="Batch edit with glob pattern")
     parser.add_argument("--history", "-H", action="store_true", help="Show edit history")
-    parser.add_argument("-n", type=int, default=10, help="Number of history entries (default: 10)")
+    parser.add_argument("--number", type=int, default=10, help="Number of history entries (default: 10)")
 
     args = parser.parse_args()
 

config/includes.chroot/usr/local/bin/neuros-firstboot

@@ -5,6 +5,11 @@
 
 set -e
 
+if [ "$#" -gt 0 ] && { [ "$1" = "--help" ] || [ "$1" = "-h" ]; }; then
+    sed -n '2,/^$/p' "$0"
+    exit 0
+fi
+
 MARKER="$HOME/.config/neuros/.firstboot-done"
 CONFIG_DIR="$HOME/.config/neuros"
 SKEL_CONTINUE="/etc/skel/.continue/config.json"

config/includes.chroot/usr/local/bin/neuros-tray

@@ -6,15 +6,21 @@ Shows model status, RAM usage, and provides quick AI access from the GNOME panel
 Requires: python3-gi, gir1.2-gtk-3.0, gir1.2-appindicator3-0.1
 """
 
-import gi
-gi.require_version('Gtk', '3.0')
-gi.require_version('AppIndicator3', '0.1')
-from gi.repository import Gtk, AppIndicator3, GLib
 import subprocess
 import json
 import os
+import sys
 import threading
 
+if any(arg in ("-h", "--help") for arg in sys.argv[1:]):
+    print(__doc__.strip())
+    sys.exit(0)
+
+import gi
+gi.require_version('Gtk', '3.0')
+gi.require_version('AppIndicator3', '0.1')
+from gi.repository import Gtk, AppIndicator3, GLib
+
 APP_ID = "neuros-tray"
 CONFIG_PATH = os.path.expanduser("~/.config/neuros/llm.conf")
 OLLAMA_URL = "http://localhost:11434"

config/includes.chroot/usr/local/bin/neuros-welcome

@@ -8,6 +8,10 @@ import sys
 import os
 import subprocess
 
+if any(arg in ("-h", "--help") for arg in sys.argv[1:]):
+    print(__doc__.strip())
+    sys.exit(0)
+
 try:
     import gi
     gi.require_version('Gtk', '3.0')

config/includes.chroot/usr/local/bin/neuroslib.py

@@ -17,12 +17,17 @@
 import shlex
 import subprocess
 import sys
+import tempfile
 import time
 from datetime import datetime
 
 # === Paths ===
 CONFIG_DIR = os.path.expanduser("~/.config/neuros")
 LLM_CONF = os.path.join(CONFIG_DIR, "llm.conf")
+RESPONSE_CACHE_PATH = os.path.expanduser(
+    os.getenv("NEUROS_RESPONSE_CACHE", "~/.cache/neuros/responses.json")
+)
+RESPONSE_CACHE_LIMIT = 256
 
 
 # ═══════════════════════════════════════════════════════════════════════
@@ -50,24 +55,74 @@ def load_config():
 # ═══════════════════════════════════════════════════════════════════════
 
 def query_llm(prompt, system="You are a helpful AI assistant.", timeout=120):
-    """Query the local Ollama instance. Returns response string or None."""
+    """Query Ollama, reusing successful responses for the same model/input."""
     config = load_config()
+    model = config["model"]
+    cache_key = _response_cache_key(model, prompt, system)
+    cached = _read_response_cache().get(cache_key)
+    if cached is not None:
+        return cached
     try:
         import urllib.request
         url = f"http://{config['host']}:{config['port']}/api/generate"
         data = json.dumps({
-            "model": config["model"],
+            "model": model,
             "prompt": prompt,
             "system": system,
             "stream": False,
         }).encode()
         req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
         with urllib.request.urlopen(req, timeout=timeout) as resp:
-            return json.loads(resp.read()).get("response", "").strip()
+            response = json.loads(resp.read()).get("response", "").strip()
+        if response:
+            _write_response_cache(cache_key, response)
+        return response
     except Exception:
         return None
 
 
+def _response_cache_key(model, prompt, system):
+    """Return a stable key; system is included to prevent prompt collisions."""
+    payload = json.dumps(
+        {"model": model, "prompt": prompt, "system": system},
+        sort_keys=True,
+        ensure_ascii=False,
+    ).encode("utf-8")
+    return hashlib.sha256(payload).hexdigest()
+
+
+def _read_response_cache():
+    try:
+        with open(RESPONSE_CACHE_PATH, encoding="utf-8") as f:
+            data = json.load(f)
+        return data if isinstance(data, dict) else {}
+    except (OSError, json.JSONDecodeError, TypeError):
+        return {}
+
+
+def _write_response_cache(key, response):
+    cache = _read_response_cache()
+    cache[key] = response
+    if len(cache) > RESPONSE_CACHE_LIMIT:
+        cache = dict(list(cache.items())[-RESPONSE_CACHE_LIMIT:])
+    try:
+        parent = os.path.dirname(RESPONSE_CACHE_PATH) or "."
+        os.makedirs(parent, exist_ok=True)
+        fd, tmp_path = tempfile.mkstemp(prefix="responses-", dir=parent, text=True)
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                json.dump(cache, f, ensure_ascii=False)
+            os.replace(tmp_path, RESPONSE_CACHE_PATH)
+        except Exception:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+            raise
+    except OSError:
+        pass
+
+
 def query_llm_stream(prompt, system="You are a helpful AI assistant.", timeout=120):
     """Query Ollama with streaming response. Yields tokens."""
     config = load_config()

tests/benchmark_cache.py

@@ -0,0 +1,50 @@
+"""Reproducible in-process benchmark for the response cache."""
+
+import importlib.util
+import os
+import pathlib
+import tempfile
+import time
+from importlib.machinery import SourceFileLoader
+from unittest.mock import patch
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+LIBRARY = ROOT / "config/includes.chroot/usr/local/bin/neuroslib.py"
+
+
+class Response:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        return False
+
+    def read(self):
+        return b'{"response":"benchmark response"}'
+
+
+def main():
+    loader = SourceFileLoader("neuroslib_benchmark", str(LIBRARY))
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    module = importlib.util.module_from_spec(spec)
+    loader.exec_module(module)
+    with tempfile.TemporaryDirectory() as tmp:
+        module.RESPONSE_CACHE_PATH = os.path.join(tmp, "responses.json")
+        config = {"model": "mistral", "host": "localhost", "port": "11434"}
+        with patch.object(module, "load_config", return_value=config), \
+             patch("urllib.request.urlopen", return_value=Response()) as request:
+            start = time.perf_counter()
+            module.query_llm("benchmark prompt")
+            cold = time.perf_counter() - start
+            start = time.perf_counter()
+            module.query_llm("benchmark prompt")
+            warm = time.perf_counter() - start
+        print(f"cold_seconds={cold:.9f}")
+        print(f"cached_seconds={warm:.9f}")
+        print(f"network_calls={request.call_count}")
+        print(f"speedup={cold / warm:.2f}x")
+
+
+if __name__ == "__main__":
+    main()

tests/test_cache.py

@@ -0,0 +1,76 @@
+"""Tests for the shared NeurOS response cache."""
+
+import importlib.util
+import json
+import os
+import pathlib
+import tempfile
+import unittest
+from importlib.machinery import SourceFileLoader
+from unittest.mock import patch
+
+
+LIBRARY_PATH = pathlib.Path(__file__).resolve().parents[1] / "config/includes.chroot/usr/local/bin/neuroslib.py"
+
+
+def load_library():
+    loader = SourceFileLoader("neuroslib_cache_test", str(LIBRARY_PATH))
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    module = importlib.util.module_from_spec(spec)
+    loader.exec_module(module)
+    return module
+
+
+class FakeResponse:
+    def __init__(self, payload):
+        self.payload = json.dumps(payload).encode()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        return False
+
+    def read(self):
+        return self.payload
+
+
+class ResponseCacheTests(unittest.TestCase):
+    def test_same_prompt_and_model_uses_cached_response(self):
+        module = load_library()
+        with tempfile.TemporaryDirectory() as tmp:
+            module.RESPONSE_CACHE_PATH = os.path.join(tmp, "responses.json")
+            with patch.object(module, "load_config", return_value={"model": "mistral", "host": "localhost", "port": "11434"}), \
+                 patch("urllib.request.urlopen", return_value=FakeResponse({"response": "cached"})) as request:
+                self.assertEqual(module.query_llm("same"), "cached")
+                self.assertEqual(module.query_llm("same"), "cached")
+            self.assertEqual(request.call_count, 1)
+
+    def test_model_is_part_of_cache_key(self):
+        module = load_library()
+        with tempfile.TemporaryDirectory() as tmp:
+            module.RESPONSE_CACHE_PATH = os.path.join(tmp, "responses.json")
+            configs = iter([
+                {"model": "mistral", "host": "localhost", "port": "11434"},
+                {"model": "llama3", "host": "localhost", "port": "11434"},
+            ])
+            with patch.object(module, "load_config", side_effect=lambda: next(configs)), \
+                 patch("urllib.request.urlopen", side_effect=[FakeResponse({"response": "one"}), FakeResponse({"response": "two"})]) as request:
+                self.assertEqual(module.query_llm("same"), "one")
+                self.assertEqual(module.query_llm("same"), "two")
+            self.assertEqual(request.call_count, 2)
+
+    def test_cache_write_is_json_and_failure_is_not_cached(self):
+        module = load_library()
+        with tempfile.TemporaryDirectory() as tmp:
+            module.RESPONSE_CACHE_PATH = os.path.join(tmp, "responses.json")
+            with patch.object(module, "load_config", return_value={"model": "mistral", "host": "localhost", "port": "11434"}), \
+                 patch("urllib.request.urlopen", return_value=FakeResponse({"response": ""})) as request:
+                self.assertEqual(module.query_llm("empty"), "")
+                self.assertEqual(module.query_llm("empty"), "")
+            self.assertEqual(request.call_count, 2)
+            self.assertFalse(os.path.exists(module.RESPONSE_CACHE_PATH))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_utilities.py

@@ -0,0 +1,51 @@
+"""Smoke-test every executable installed by the NeurOS tools hook."""
+
+import ast
+import os
+import pathlib
+import subprocess
+import unittest
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+TOOLS = ROOT / "config" / "includes.chroot" / "usr" / "local" / "bin"
+ENTRYPOINTS = sorted(
+    path for path in TOOLS.iterdir()
+    if path.is_file() and os.access(path, os.X_OK)
+)
+
+
+class UtilitySmokeTests(unittest.TestCase):
+    def test_entrypoint_inventory_is_complete(self):
+        self.assertEqual(len(ENTRYPOINTS), 80)
+
+    def test_python_and_shell_sources_parse(self):
+        for path in ENTRYPOINTS:
+            with self.subTest(path=path.name):
+                first_line = path.read_bytes().splitlines()[0]
+                if b"python" in first_line:
+                    ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+                elif b"sh" in first_line or b"bash" in first_line:
+                    result = subprocess.run(["bash", "-n", str(path)], capture_output=True, text=True)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                else:
+                    self.fail(f"missing recognized shebang: {path.name}")
+
+    def test_help_for_every_entrypoint(self):
+        failures = []
+        for path in ENTRYPOINTS:
+            with self.subTest(path=path.name):
+                result = subprocess.run(
+                    [str(path), "--help"],
+                    cwd=ROOT,
+                    capture_output=True,
+                    text=True,
+                    timeout=10,
+                )
+                if result.returncode != 0:
+                    failures.append(f"{path.name}: rc={result.returncode}: {(result.stderr or result.stdout).strip()}")
+        self.assertFalse(failures, "\n".join(failures))
+
+
+if __name__ == "__main__":
+    unittest.main()