Harden system-prompt builder + fix memory search concurrency race

prompt.py (runs every turn): malformed MCP server entries (missing
toolCount/name/status, or non-dict), None elements in permission_summary, and
malformed skill dicts each crashed build_system_prompt_bundle — guarded all
section builders with .get() defaults + per-entry isinstance checks so one bad
entry can't take down the whole prompt build.

memory.py MemoryFile.search: snapshot entries once at the start instead of
iterating self.entries twice. Previously a concurrent add_entry could append
between the entry_tokens-building loop and the scoring loop, making
entry_tokens[i] index out of range (test_concurrent_add_and_search crashed
deterministically). The two loops now iterate the same snapshot.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: QUSETIONS

minicode/memory.py

@@ -821,6 +821,12 @@ def search(self, query: str, active_domains: list[str] | None = None) -> list[Me
         if not self.entries:
             return []
 
+        # Snapshot entries so concurrent add_entry/_enforce_limits can't shift
+        # indices between the two loops below (was: "list index out of range"
+        # when another thread appended between building entry_tokens and the
+        # scoring loop).
+        entries = list(self.entries)
+
         query_tokens = _tokenize(query)
         query_tokens = _expand_query_terms(query_tokens, active_domains=active_domains)
         if not query_tokens:
@@ -830,15 +836,15 @@ def search(self, query: str, active_domains: list[str] | None = None) -> list[Me
         query_terms = query_lower.split()
 
         entry_tokens = []
-        for entry in self.entries:
+        for entry in entries:
             text = f"{entry.content} {entry.category} {' '.join(entry.tags)}"
             entry_tokens.append(_tokenize(text))
 
         idf = _compute_idf(entry_tokens)
         avgdl = _compute_avgdl(entry_tokens)
 
         scored: list[tuple[float, MemoryEntry]] = []
-        for i, entry in enumerate(self.entries):
+        for i, entry in enumerate(entries):
             bm25 = _bm25_score(query_tokens, entry_tokens[i], idf, avgdl)
 
             substring_score = 0.0

minicode/prompt.py

@@ -156,7 +156,10 @@ def build_system_prompt_bundle(
     # --- Dynamic Suffix (Per-turn) ---
     # Permission context
     if permission_summary:
-        perm_text = "Permission context:\n" + "\n".join(permission_summary)
+        # Coerce/filter so a None element in the summary can't crash join().
+        perm_text = "Permission context:\n" + "\n".join(
+            str(p) for p in permission_summary if p is not None
+        )
         pipeline.register_dynamic("permissions", lambda: perm_text)
 
     # Skills section with conditional injection
@@ -165,7 +168,10 @@ def build_system_prompt_bundle(
         def _build_skills():
             lines = ["Available skills:"]
             lines.extend(
-                f"- {skill['name']}: {skill['description']}" for skill in skills
+                f"- {skill.get('name', '?')}: {skill.get('description', '')}"
+                if isinstance(skill, dict)
+                else f"- {skill}"
+                for skill in skills
             )
             lines.extend([
                 "",
@@ -194,31 +200,44 @@ def _build_skills():
     # MCP servers section
     mcp_servers = extras.get("mcpServers", [])
     if mcp_servers:
+        def _server_line(server: Any) -> str:
+            # Guard each entry so one malformed server (missing keys / non-dict)
+            # can't crash the whole system-prompt build.
+            if not isinstance(server, dict):
+                return f"- (malformed MCP server entry: {server!r})"
+            parts = [
+                f"- {server.get('name', '(unnamed)')}: "
+                f"{server.get('status', 'unknown')}, tools={server.get('toolCount', 0)}"
+            ]
+            if server.get("resourceCount") is not None:
+                parts.append(f", resources={server['resourceCount']}")
+            if server.get("promptCount") is not None:
+                parts.append(f", prompts={server['promptCount']}")
+            if server.get("protocol"):
+                parts.append(f", protocol={server['protocol']}")
+            if server.get("error"):
+                parts.append(f" ({server['error']})")
+            return "".join(parts)
+
         def _build_mcp():
             lines = ["Configured MCP servers:"]
-            lines.extend(
-                "- "
-                + server["name"]
-                + f": {server['status']}, tools={server['toolCount']}"
-                + (f", resources={server['resourceCount']}" if server.get("resourceCount") is not None else "")
-                + (f", prompts={server['promptCount']}" if server.get("promptCount") is not None else "")
-                + (f", protocol={server['protocol']}" if server.get("protocol") else "")
-                + (f" ({server['error']})" if server.get("error") else "")
-                for server in mcp_servers
-            )
-            if any(server.get("status") == "connected" for server in mcp_servers):
+            lines.extend(_server_line(server) for server in mcp_servers)
+            if any((server.get("status") if isinstance(server, dict) else None) == "connected" for server in mcp_servers):
                 lines.append(
                     "Connected MCP tools are already exposed in the tool list with names prefixed like mcp__server__tool. "
                     "Use list_mcp_resources/read_mcp_resource and list_mcp_prompts/get_mcp_prompt when a server exposes those capabilities."
                 )
             # Sequential thinking server detection
             sequential_servers = [
                 server for server in mcp_servers
-                if "sequential" in server.get("name", "").lower()
-                or "branch-thinking" in server.get("name", "").lower()
-                or "think" in server.get("name", "").lower()
+                if isinstance(server, dict)
+                and (
+                    "sequential" in server.get("name", "").lower()
+                    or "branch-thinking" in server.get("name", "").lower()
+                    or "think" in server.get("name", "").lower()
+                )
             ]
-            if any(server.get("status") == "connected" for server in sequential_servers):
+            if any(isinstance(s, dict) and s.get("status") == "connected" for s in sequential_servers):
                 lines.extend([
                     "",
                     "SEQUENTIAL THINKING MCP SERVER IS CONNECTED!",

tests/test_prompt.py

@@ -43,3 +43,50 @@ def test_build_system_prompt_includes_memory_context(tmp_path: Path) -> None:
 
     assert "Project Memory & Context" in prompt
     assert "Always run pytest before release." in prompt
+
+
+# ---------------------------------------------------------------------------
+# Robustness: the system-prompt builder runs every turn; malformed MCP/skill/
+# permission inputs must not crash it.
+# ---------------------------------------------------------------------------
+
+
+def test_build_system_prompt_bundle_handles_malformed_mcp_entry():
+    """A partial MCP server dict (missing toolCount/name/status) or a non-dict
+    entry must not KeyError/AttributeError the prompt build."""
+    from minicode.prompt import build_system_prompt_bundle
+
+    extras = {
+        "mcpServers": [
+            {"name": "broken", "status": "error"},  # missing toolCount
+            "not-a-dict",  # wholly malformed
+            {"name": "sequential-thinking", "status": "connected", "toolCount": 2},
+        ],
+        "skills": [],
+        "memory_context": "",
+        "runtime": {},
+    }
+    bundle = build_system_prompt_bundle(".", ["cwd: ."], extras)
+    assert isinstance(bundle.prompt, str) and bundle.prompt
+    assert "sequential-thinking" in bundle.prompt
+
+
+def test_build_system_prompt_bundle_handles_none_in_permission_summary():
+    """A None element in permission_summary must not crash join()."""
+    from minicode.prompt import build_system_prompt_bundle
+
+    bundle = build_system_prompt_bundle(
+        ".", ["ok", None, "x"], {"mcpServers": [], "skills": [], "memory_context": "", "runtime": {}}
+    )
+    assert "Permission context" in bundle.prompt
+    assert "ok" in bundle.prompt and "x" in bundle.prompt
+
+
+def test_build_system_prompt_bundle_handles_malformed_skill():
+    """A skill dict missing name/description must not KeyError the build."""
+    from minicode.prompt import build_system_prompt_bundle
+
+    bundle = build_system_prompt_bundle(
+        ".", [], {"mcpServers": [], "skills": [{"name": "s"}], "memory_context": "", "runtime": {}}
+    )
+    assert isinstance(bundle.prompt, str) and bundle.prompt