Refuse unsafe MCP payloads and attach local_review notes

.github/workflows/tests.yml

@@ -26,3 +26,5 @@ jobs:
         run: python scripts/run_harness.py --backend stub --profile golden --fast-ms 1 --strong-ms 1
       - name: Stub MCP + apply gate
         run: python scripts/run_harness.py --backend stub --profile golden --orchestrate --fast-ms 1 --strong-ms 1
+      - name: Stub MCP observed failover
+        run: python scripts/run_harness.py --backend stub --profile observed --fast-ms 1 --strong-ms 1

docs/evaluation-protocol.md

@@ -148,7 +148,14 @@ file. They prove the state machine the agent is supposed to follow.
 The same gate runs after **real stdio MCP** calls when you pass
 `--orchestrate`. Stub Ollama still supplies the worker text. Keep jobs
 never call `local_*`. Accept / rewrite / reject then run on the scored
-candidate.
+candidate. Security-sensitive delegated jobs first call `local_review`
+and attach those notes to the premium packet; that still cannot approve.
+
+The MCP server also refuses secret filenames, private-key / token
+blobs, oversized file sets, and `max_tokens` above 8192 **before**
+calling Ollama. That is defense in depth if a client skips the eval
+router. A function that only mentions `password` is not treated as a
+secret blob.
 
 ```bash
 PYTHONPATH=src .venv/bin/python -m unittest tests.test_eval_orchestrate tests.test_eval_mcp_orchestrate -v

scripts/run_harness.py

@@ -67,6 +67,7 @@ def main() -> None:
                     "source": item.apply_source,
                     "models": list(item.local_models),
                     "review": item.review_decision,
+                    "local_review_notes_chars": len(item.local_review_notes),
                 }
                 for item in results
             ]

spec.md

@@ -238,7 +238,7 @@ premium agent decides whether to apply edits.
 | Field | Type | Required | Notes |
 | --- | --- | --- | --- |
 | `task` | string | yes | What to produce. Bounded input. |
-| `files` | array of `{ path, content }` | no | Snippets the premium agent chooses to send. Not a full-repo dump. |
+| `files` | array of `{ path, content }` | no | Snippets the premium agent chooses to send. Not a full-repo dump. The server refuses `.env` / private-key / token blobs, more than 12 files, or `max_tokens` above 8192 before calling Ollama. |
 | `language` | string | no | Hint, e.g. `java`, `typescript`. |
 | `style` | string | no | Short conventions: test framework, naming, etc. |
 | `model` | `fast` \| `strong` | no | Default `fast`. |

src/local_coding_slm/eval/harness.py

@@ -5,6 +5,7 @@
 import os
 import sys
 import time
+from dataclasses import replace
 from pathlib import Path
 
 from local_coding_slm.eval.cases import CASES, EvalCase
@@ -171,6 +172,10 @@ async def _run_orchestrated_job(
         return run_job(job)
     if job.eval_case is None:
         raise ValueError(f"{job.id}: delegated jobs need an eval_case")
+    notes = job.local_review_notes
+    if job.signals.security_sensitive and not notes:
+        notes = await _call_local_review(session, job.eval_case)
+        job = replace(job, local_review_notes=notes)
     attempts = await _run_local_mcp(
         session,
         job.eval_case,
@@ -202,6 +207,25 @@ async def _run_local_mcp(
         records.append(item.record)
 
 
+async def _call_local_review(session: object, case: EvalCase) -> str:
+    """Cheap first-pass notes for the premium packet. Not an apply."""
+    payload = {
+        "task": (
+            "First-pass review only. Flag obvious null, auth, secret, and "
+            "error-handling gaps. Do not rewrite.\n\nOriginal task:\n"
+            + case.task
+        ),
+        "files": list(case.files),
+        "language": case.language,
+        "model": "fast",
+        "max_tokens": 400,
+    }
+    tool = await session.call_tool("local_review", payload)  # type: ignore[attr-defined]
+    return "".join(
+        block.text for block in tool.content if getattr(block, "text", None)
+    )
+
+
 async def _one_attempt(
     session: object,
     case: EvalCase,

src/local_coding_slm/eval/orchestrate.py

@@ -77,6 +77,7 @@ class JobResult:
     review_decision: str | None
     review_reviewer: str | None
     local_passed: bool | None
+    local_review_notes: str = ""
 
 
 class ScriptedLocal:
@@ -321,4 +322,5 @@ def _result(
         review_decision=None if verdict is None else verdict.decision,
         review_reviewer=None if verdict is None else verdict.reviewer,
         local_passed=last_passed,
+        local_review_notes=job.local_review_notes,
     )

src/local_coding_slm/eval/routing.py

@@ -12,6 +12,8 @@
 from collections.abc import Sequence
 from dataclasses import dataclass
 
+from local_coding_slm.payload import inspect_payload
+
 
 @dataclass(frozen=True)
 class RouteSignals:
@@ -45,17 +47,8 @@ def mechanical_signals(**overrides: bool) -> RouteSignals:
 
 
 def payload_block_reason(files: Sequence[dict[str, str]] | None) -> str | None:
-    """Refuse to send secrets or credential files to local_*."""
-    for item in files or ():
-        path = (item.get("path") or "").replace("\\", "/").lower()
-        name = path.rsplit("/", 1)[-1]
-        if name == ".env.example":
-            continue
-        if name == ".env" or name.startswith(".env."):
-            return "secrets_file"
-        if name in {"credentials.json", "id_rsa", "id_rsa.pub"}:
-            return "secrets_file"
-    return None
+    """Refuse secrets, credential files, and oversized snippets."""
+    return inspect_payload(files)
 
 
 def route(

src/local_coding_slm/payload.py

@@ -0,0 +1,76 @@
+"""Refuse unsafe or oversized snippets before they reach Ollama.
+
+Used by the MCP server (defense in depth) and by eval routing. This is
+not a classifier: it only looks at file names, size, and a few secret
+shapes. Code that merely mentions ``password`` is allowed.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+MAX_FILES = 12
+MAX_BYTES = 120_000
+MAX_TOKENS = 8192
+
+_PRIVATE_KEY_MARKERS = (
+    "-----BEGIN OPENSSH PRIVATE KEY-----",
+    "-----BEGIN RSA PRIVATE KEY-----",
+    "-----BEGIN EC PRIVATE KEY-----",
+    "-----BEGIN PRIVATE KEY-----",
+)
+_TOKEN_PREFIXES = ("ghp_", "github_pat_", "sk-proj-", "sk-ant-")
+
+
+def inspect_payload(
+    files: Sequence[dict[str, str]] | None,
+    *,
+    max_tokens: int | None = None,
+) -> str | None:
+    """Return a stable reason string, or None if the payload may be sent."""
+    if max_tokens is not None and max_tokens > MAX_TOKENS:
+        return "max_tokens_too_large"
+    items = list(files or ())
+    if len(items) > MAX_FILES:
+        return "too_many_files"
+    total = 0
+    for item in items:
+        path = str(item.get("path") or "")
+        content = str(item.get("content") or "")
+        total += len(content)
+        name_reason = _secret_path(path)
+        if name_reason:
+            return name_reason
+        content_reason = _secret_content(content)
+        if content_reason:
+            return content_reason
+    if total > MAX_BYTES:
+        return "payload_too_large"
+    return None
+
+
+def refusal_message(reason: str) -> str:
+    return f"ERROR: refused to send payload to local model ({reason})"
+
+
+def _secret_path(path: str) -> str | None:
+    name = path.replace("\\", "/").rsplit("/", 1)[-1].lower()
+    if name == ".env.example":
+        return None
+    if name == ".env" or name.startswith(".env."):
+        return "secrets_file"
+    if name in {"credentials.json", "id_rsa", "id_rsa.pub"}:
+        return "secrets_file"
+    return None
+
+
+def _secret_content(content: str) -> str | None:
+    if any(marker in content for marker in _PRIVATE_KEY_MARKERS):
+        return "secret_content"
+    lowered = content.lower()
+    if "aws_secret_access_key=" in lowered:
+        return "secret_content"
+    for prefix in _TOKEN_PREFIXES:
+        if prefix in content:
+            return "secret_content"
+    return None

src/local_coding_slm/server.py

@@ -19,6 +19,7 @@
     format_user_task,
     status_report,
 )
+from local_coding_slm.payload import inspect_payload, refusal_message  # noqa: E402
 from local_coding_slm.prompts import SYSTEM_PROMPTS  # noqa: E402
 
 
@@ -64,6 +65,9 @@ def _run_tool(
 ) -> str:
     if not task or not task.strip():
         return "ERROR: task is required"
+    blocked = inspect_payload(files, max_tokens=max_tokens)
+    if blocked:
+        return refusal_message(blocked)
     user = format_user_task(task, files=files, language=language, style=style)
     try:
         return chat(

tests/test_eval_mcp_orchestrate.py

@@ -78,6 +78,19 @@ async def test_mcp_accept_rewrite_reject(self) -> None:
         self.assertEqual(notes.case_id, "review_login")
         self.assertEqual(notes.outcome, "rejected")
         self.assertFalse(notes.applied)
+        self.assertTrue(notes.local_review_notes)
+
+    async def test_security_job_runs_local_review_prelude(self) -> None:
+        results = await run_orchestrated_campaign(
+            backend="stub",
+            profile="golden",
+            job_ids=["mcp_reject_security"],
+            fast_ms=1,
+            strong_ms=1,
+        )
+        item = results[0]
+        self.assertGreater(len(item.local_review_notes), 0)
+        self.assertEqual(item.outcome, "rejected")
 
     async def test_observed_move_repairs_then_accepts(self) -> None:
         results = await run_orchestrated_campaign(

tests/test_eval_score.py

@@ -73,6 +73,14 @@ def test_prose_explain_missing_phrase_is_structure(self) -> None:
         self.assertEqual(result.layer("behavior").status, "skip")
         self.assertIn("hi", result.layer("structure").message)
 
+    def test_explain_and_review_are_prose_cases(self) -> None:
+        explain = CASES_BY_ID["explain_clamp"]
+        review = CASES_BY_ID["review_login"]
+        self.assertFalse(explain.expect_fences)
+        self.assertFalse(review.expect_fences)
+        self.assertEqual(explain.tool, "local_explain")
+        self.assertEqual(review.tool, "local_review")
+
     def test_move_partial_is_format_not_behavior(self) -> None:
         case = CASES_BY_ID["move_function_imports"]
         partial = next(item for item in FIXTURES if item.name == "move_function_partial")

tests/test_payload.py

@@ -0,0 +1,68 @@
+"""Payload guards. No Ollama."""
+
+from __future__ import annotations
+
+import unittest
+
+from local_coding_slm.eval.routing import mechanical_signals, route
+from local_coding_slm.payload import MAX_FILES, inspect_payload, refusal_message
+
+
+class InspectPayloadTests(unittest.TestCase):
+    def test_clean_snippet_ok(self) -> None:
+        self.assertIsNone(
+            inspect_payload([{"path": "add.py", "content": "def add(a, b): return a + b\n"}])
+        )
+
+    def test_env_file(self) -> None:
+        self.assertEqual(
+            inspect_payload([{"path": ".env", "content": "K=v"}]),
+            "secrets_file",
+        )
+
+    def test_env_example_ok(self) -> None:
+        self.assertIsNone(
+            inspect_payload([{"path": ".env.example", "content": "OLLAMA_BASE_URL="}])
+        )
+
+    def test_private_key_content(self) -> None:
+        self.assertEqual(
+            inspect_payload(
+                [
+                    {
+                        "path": "app.py",
+                        "content": "-----BEGIN OPENSSH PRIVATE KEY-----\nabc\n",
+                    }
+                ]
+            ),
+            "secret_content",
+        )
+
+    def test_password_in_code_is_not_a_secret_blob(self) -> None:
+        self.assertIsNone(
+            inspect_payload(
+                [{"path": "auth.py", "content": "def login(user, password):\n    return user.name\n"}]
+            )
+        )
+
+    def test_too_many_files(self) -> None:
+        files = [{"path": f"f{i}.py", "content": "x"} for i in range(MAX_FILES + 1)]
+        self.assertEqual(inspect_payload(files), "too_many_files")
+
+    def test_max_tokens(self) -> None:
+        self.assertEqual(inspect_payload([], max_tokens=9000), "max_tokens_too_large")
+
+    def test_route_uses_payload_guard(self) -> None:
+        decision = route(
+            mechanical_signals(),
+            files=[{"path": "id_rsa", "content": "not-a-real-key"}],
+        )
+        self.assertEqual(decision.action, "keep")
+        self.assertEqual(decision.reason, "secrets_file")
+
+    def test_refusal_message_is_transport_shaped(self) -> None:
+        self.assertTrue(refusal_message("secrets_file").startswith("ERROR:"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_server_payload.py

@@ -0,0 +1,57 @@
+"""MCP server refuses unsafe payloads before calling Ollama."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from local_coding_slm.server import _run_tool
+
+
+class ServerPayloadTests(unittest.TestCase):
+    @patch("local_coding_slm.server.chat")
+    def test_env_file_never_calls_ollama(self, chat: object) -> None:
+        text = _run_tool(
+            "local_code",
+            "rename a helper",
+            [{"path": ".env", "content": "OLLAMA_BASE_URL=http://127.0.0.1:11434"}],
+            None,
+            None,
+            "fast",
+            None,
+        )
+        self.assertIn("secrets_file", text)
+        self.assertTrue(text.startswith("ERROR:"))
+        chat.assert_not_called()  # type: ignore[attr-defined]
+
+    @patch("local_coding_slm.server.chat")
+    def test_private_key_never_calls_ollama(self, chat: object) -> None:
+        text = _run_tool(
+            "local_refactor",
+            "extract a helper",
+            [{"path": "key.py", "content": "-----BEGIN RSA PRIVATE KEY-----\nxx\n"}],
+            None,
+            None,
+            "fast",
+            700,
+        )
+        self.assertIn("secret_content", text)
+        chat.assert_not_called()  # type: ignore[attr-defined]
+
+    @patch("local_coding_slm.server.chat", return_value="ok")
+    def test_clean_payload_reaches_chat(self, chat: object) -> None:
+        text = _run_tool(
+            "local_code",
+            "write clamp",
+            [{"path": "spec.md", "content": "clamp value between lo and hi"}],
+            "python",
+            None,
+            "fast",
+            400,
+        )
+        self.assertEqual(text, "ok")
+        chat.assert_called_once()  # type: ignore[attr-defined]
+
+
+if __name__ == "__main__":
+    unittest.main()