fix(chat): harden WebUI prefill scripts

CHANGELOG.md

@@ -5,7 +5,7 @@
 ### Added
 
 - Add non-sensitive SSE stream runtime diagnostics to deep health checks, including active stream count, subscriber totals, and offline buffered-event counts for stuck or slow WebUI chat investigations.
-- Add WebUI session prefill parity: browser-originated chat turns now load configured prefill context from `prefill_messages_file` or `prefill_messages_script`, pass it to Hermes Agent as ephemeral model context, and surface a compact context status event in the chat UI without exposing prefill message bodies.
+- Add WebUI session prefill parity: browser-originated chat turns now load configured prefill context from `prefill_messages_file` or `prefill_messages_script`, pass it to Hermes Agent as ephemeral model context, and surface a compact context status event in the chat UI without exposing prefill message bodies. File prefill is cheap to read each turn; script prefill runs before SSE output starts, inherits only a minimal `PATH`/`HOME` environment, and uses a short path/mtime TTL cache to avoid re-shelling during rapid browser turns.
 
 ### Changed
 

api/streaming.py

@@ -231,9 +231,14 @@ def _webui_ephemeral_system_prompt(
 
 _SECRET_SHAPED_RE = re.compile(
     r"(?i)(api[_-]?key|token|password|secret)\s*[:=]\s*[^\s]+|"
+    r"\b(?:sk-[A-Za-z0-9_-]{16,}|ghp_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b|"
     r"[A-Za-z0-9_-]{24,}\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{16,}"
 )
 
+_PREFILL_SCRIPT_CACHE_TTL_SECONDS = 10.0
+_PREFILL_SCRIPT_CACHE_LOCK = threading.Lock()
+_PREFILL_SCRIPT_CACHE: dict[tuple[str, int], tuple[float, dict]] = {}
+
 
 def _redact_prefill_status_text(text: str) -> str:
     """Return a short, non-secret diagnostic string for prefill status."""
@@ -241,6 +246,36 @@ def _redact_prefill_status_text(text: str) -> str:
     return " ".join(clean.split())[:240]
 
 
+def _prefill_script_env() -> dict[str, str]:
+    """Return the minimal environment inherited by configured prefill scripts."""
+    return {
+        "PATH": os.environ.get("PATH", ""),
+        "HOME": os.environ.get("HOME", ""),
+    }
+
+
+def _prefill_script_cache_get(cache_key: tuple[str, int], ttl: float) -> dict | None:
+    if ttl <= 0:
+        return None
+    now = time.monotonic()
+    with _PREFILL_SCRIPT_CACHE_LOCK:
+        cached = _PREFILL_SCRIPT_CACHE.get(cache_key)
+        if not cached:
+            return None
+        cached_at, result = cached
+        if now - cached_at > ttl:
+            _PREFILL_SCRIPT_CACHE.pop(cache_key, None)
+            return None
+        return copy.deepcopy(result)
+
+
+def _prefill_script_cache_put(cache_key: tuple[str, int], result: dict, ttl: float) -> None:
+    if ttl <= 0:
+        return
+    with _PREFILL_SCRIPT_CACHE_LOCK:
+        _PREFILL_SCRIPT_CACHE[cache_key] = (time.monotonic(), copy.deepcopy(result))
+
+
 def _valid_prefill_messages(value) -> list[dict]:
     """Normalize a prefill payload to role/content messages."""
     if not isinstance(value, list):
@@ -274,6 +309,7 @@ def _load_webui_prefill_context(
     python_exe: Optional[str] = None,
     env: Optional[dict] = None,
     timeout: float = 20.0,
+    script_cache_ttl: float = _PREFILL_SCRIPT_CACHE_TTL_SECONDS,
 ) -> dict:
     """Load configured WebUI session prefill messages.
 
@@ -281,6 +317,10 @@ def _load_webui_prefill_context(
     ``prefill_messages_script`` hook whose stdout becomes one user-role recall
     message. Script output is intentionally ephemeral: it is passed to the
     agent as prefill context and is not written into the session transcript.
+
+    File prefill is read directly each turn. Script prefill executes before SSE
+    output starts, so successful script results are cached briefly by path/mtime
+    to avoid re-shelling on every rapid browser turn.
     """
     cfg = config_data if isinstance(config_data, dict) else get_config()
     file_raw = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "") or str(cfg.get("prefill_messages_file") or "")
@@ -300,6 +340,14 @@ def _load_webui_prefill_context(
         label = path.name or "prefill script"
         if not path.exists():
             return {"status": "error", "source": "script", "label": label, "messages": [], "message_count": 0, "error": "prefill script not found"}
+        try:
+            mtime_ns = path.stat().st_mtime_ns
+        except OSError as exc:
+            return {"status": "error", "source": "script", "label": label, "messages": [], "message_count": 0, "error": _redact_prefill_status_text(str(exc))}
+        cache_key = (str(path), mtime_ns)
+        cached = _prefill_script_cache_get(cache_key, script_cache_ttl)
+        if cached is not None:
+            return cached
         exe = python_exe or sys.executable
         cmd = [exe, str(path)] if path.suffix == ".py" else [str(path)]
         try:
@@ -320,7 +368,9 @@ def _load_webui_prefill_context(
         output = (proc.stdout or "").strip()
         messages = [{"role": "user", "content": output}] if output else []
         status = "loaded" if messages else "empty"
-        return {"status": status, "source": "script", "label": label, "messages": messages, "message_count": len(messages)}
+        result = {"status": status, "source": "script", "label": label, "messages": messages, "message_count": len(messages)}
+        _prefill_script_cache_put(cache_key, result, script_cache_ttl)
+        return result
     return {"status": "not_configured", "source": "none", "label": "", "messages": [], "message_count": 0}
 
 
@@ -3653,7 +3703,7 @@ def _run_agent_streaming(
             # Read per-profile config at call time (not module-level snapshot)
             from api.config import get_config as _get_config
             _cfg = _get_config()
-            _prefill_context = _load_webui_prefill_context(_cfg)
+            _prefill_context = _load_webui_prefill_context(_cfg, env=_prefill_script_env())
             _prefill_messages = _prefill_context.get('messages') or []
             put('context_status', {
                 'session_id': session_id,

tests/test_webui_prefill_context.py

@@ -33,30 +33,132 @@ def test_prefill_script_output_becomes_safe_user_message(tmp_path):
     ]
 
 
-def test_prefill_json_file_is_loaded_without_script_execution(tmp_path):
+def test_prefill_script_uses_short_ttl_cache_keyed_by_path_and_mtime(tmp_path):
+    from api.streaming import _load_webui_prefill_context
+
+    script = tmp_path / "recall.py"
+    counter = tmp_path / "counter.txt"
+    script.write_text(
+        "from pathlib import Path\n"
+        f"p=Path({str(counter)!r})\n"
+        "n=int(p.read_text() or '0') if p.exists() else 0\n"
+        "p.write_text(str(n+1))\n"
+        "print(f'run {n+1}')\n",
+        encoding="utf-8",
+    )
+
+    first = _load_webui_prefill_context(
+        {"prefill_messages_script": str(script)},
+        python_exe=sys.executable,
+        env={"PATH": os.environ.get("PATH", "")},
+        script_cache_ttl=10.0,
+    )
+    second = _load_webui_prefill_context(
+        {"prefill_messages_script": str(script)},
+        python_exe=sys.executable,
+        env={"PATH": os.environ.get("PATH", "")},
+        script_cache_ttl=10.0,
+    )
+
+    assert first["messages"] == [{"role": "user", "content": "run 1"}]
+    assert second["messages"] == [{"role": "user", "content": "run 1"}]
+    assert counter.read_text(encoding="utf-8") == "1"
+
+    script.write_text(script.read_text(encoding="utf-8") + "# invalidate\n", encoding="utf-8")
+    third = _load_webui_prefill_context(
+        {"prefill_messages_script": str(script)},
+        python_exe=sys.executable,
+        env={"PATH": os.environ.get("PATH", "")},
+        script_cache_ttl=10.0,
+    )
+
+    assert third["messages"] == [{"role": "user", "content": "run 2"}]
+    assert counter.read_text(encoding="utf-8") == "2"
+
+
+def test_prefill_script_timeout_returns_error_without_hanging(tmp_path):
+    from api.streaming import _load_webui_prefill_context
+
+    script = tmp_path / "slow.py"
+    script.write_text("import time\ntime.sleep(1)\nprint('late')\n", encoding="utf-8")
+
+    result = _load_webui_prefill_context(
+        {"prefill_messages_script": str(script)},
+        python_exe=sys.executable,
+        env={"PATH": os.environ.get("PATH", "")},
+        timeout=0.05,
+        script_cache_ttl=0,
+    )
+
+    assert result["status"] == "error"
+    assert result["source"] == "script"
+    assert result["messages"] == []
+    assert "timed out" in result["error"].lower()
+
+
+def test_prefill_json_file_keeps_valid_roles_and_drops_invalid_items(tmp_path):
     from api.streaming import _load_webui_prefill_context
 
     prefill = tmp_path / "prefill.json"
-    prefill.write_text(json.dumps([{"role": "user", "content": "Pinned context"}]), encoding="utf-8")
+    prefill.write_text(
+        json.dumps(
+            [
+                {"role": "user", "content": "Pinned context"},
+                {"role": "tool", "content": "drop invalid role"},
+                {"role": "assistant", "content": "Useful assistant context"},
+                {"role": "system", "content": "   "},
+                "not a message",
+            ]
+        ),
+        encoding="utf-8",
+    )
 
     result = _load_webui_prefill_context({"prefill_messages_file": str(prefill)}, python_exe=sys.executable)
 
     assert result["status"] == "loaded"
     assert result["source"] == "file"
     assert result["label"] == "prefill.json"
-    assert result["messages"] == [{"role": "user", "content": "Pinned context"}]
+    assert result["messages"] == [
+        {"role": "user", "content": "Pinned context"},
+        {"role": "assistant", "content": "Useful assistant context"},
+    ]
+
+
+def test_public_prefill_status_strips_message_bodies():
+    from api.streaming import _public_prefill_context_status
+
+    public = _public_prefill_context_status(
+        {
+            "status": "loaded",
+            "source": "script",
+            "label": "recall.py",
+            "message_count": 1,
+            "messages": [{"role": "user", "content": "private recall payload"}],
+        }
+    )
+
+    assert public == {
+        "status": "loaded",
+        "source": "script",
+        "label": "recall.py",
+        "message_count": 1,
+    }
+    assert "messages" not in public
 
 
 def test_prefill_context_redacts_secret_shaped_errors(tmp_path):
     from api.streaming import _load_webui_prefill_context
 
-    missing = tmp_path / "missing.py"
+    script = tmp_path / "leaky.py"
+    script.write_text("import sys\nsys.stderr.write('sk-proj-abcdefghijklmnopqrstuvwxyz123456 leaked')\nsys.exit(2)\n", encoding="utf-8")
     result = _load_webui_prefill_context(
-        {"prefill_messages_script": str(missing)},
+        {"prefill_messages_script": str(script)},
         python_exe=sys.executable,
+        env={"PATH": os.environ.get("PATH", "")},
+        script_cache_ttl=0,
     )
 
     assert result["status"] == "error"
     assert result["messages"] == []
-    assert "missing.py" in result["label"]
-    assert "token" not in result.get("error", "").lower()
+    assert "sk-proj" not in result.get("error", "")
+    assert "[REDACTED]" in result.get("error", "")