Tighten local-to-premium review handoff

docs/evaluation-protocol.md

@@ -135,8 +135,11 @@ and `orchestrate.py`:
 3. **Apply gate** requires a premium verdict (`accept` / `rewrite` /
    `reject`). A local layer-pass is not approval. `local_review` is a
    cheap SLM tool and cannot approve. Reject drops the patch. Rewrite
-   applies the premium text, not the raw local output. Accept applies
-   local text only when the four layers already passed.
+   applies the premium text, not the raw local output, **and is scored
+   on the same four layers** before apply. Accept applies local text
+   only when the four layers already passed. `.env` / credential files
+   are never delegated. The reviewer sees a `ReviewPacket` (text, layer
+   statuses, optional `local_review` notes), not only the raw MCP string.
 
 CI uses a scripted reviewer. These tests do **not** call Cursor, GPT, or
 Claude, and they do not prove that a live IDE agent followed the rule
@@ -192,6 +195,7 @@ After repeated live runs, a paper may claim:
   executed tests (`test_add_execute`; A6 now uses this checker).
 - That keep-vs-delegate and accept/rewrite/reject are enforceable as a
   state machine on stub workers plus real stdio MCP.
+- That a premium rewrite which fails the same oracles is not applied.
 
 It still may not claim:
 

src/local_coding_slm/eval/harness.py

@@ -165,7 +165,8 @@ async def _run_orchestrated_job(
 ) -> JobResult:
     from local_coding_slm.eval.routing import route
 
-    decision = route(job.signals)
+    files = job.eval_case.files if job.eval_case is not None else ()
+    decision = route(job.signals, files)
     if decision.action == "keep":
         return run_job(job)
     if job.eval_case is None:

src/local_coding_slm/eval/orchestrate.py

@@ -17,6 +17,7 @@
 from local_coding_slm.eval.record import AttemptRecord
 from local_coding_slm.eval.review import (
     LOCAL_REVIEW_TOOL,
+    ReviewPacket,
     ReviewVerdict,
     ScriptedReviewer,
     is_premium_review,
@@ -25,7 +26,7 @@
 from local_coding_slm.eval.score import EvalCase, EvalResult, score_candidate
 
 GenerateFn = Callable[[AttemptPlan], str]
-ReviewFn = Callable[[str, str, bool], ReviewVerdict | None]
+ReviewFn = Callable[..., ReviewVerdict | None]
 
 
 @dataclass(frozen=True)
@@ -56,6 +57,7 @@ class OrchestrationJob:
     local_replies: tuple[str, ...] = ()
     review: ReviewVerdict | None = None
     premium_keep_text: str = ""
+    local_review_notes: str = ""
 
 
 @dataclass(frozen=True)
@@ -97,8 +99,9 @@ def decide_apply(
     last: LocalAttempt | None,
     verdict: ReviewVerdict | None,
     keep_text: str = "",
+    eval_case: EvalCase | None = None,
 ) -> ApplyDecision:
-    """Apply gate. Local layer-pass is not approval."""
+    """Apply gate. Local layer-pass is not approval. Rewrites are re-scored."""
     if not delegated:
         return ApplyDecision(
             outcome="kept_on_premium",
@@ -134,6 +137,18 @@ def decide_apply(
                 text=None,
                 blocked="empty_rewrite",
             )
+        if eval_case is not None:
+            scored = score_candidate(verdict.text, eval_case)
+            if not scored.passed:
+                fail = scored.first_failure
+                layer = fail.name if fail is not None else "behavior"
+                return ApplyDecision(
+                    outcome="blocked",
+                    applied=False,
+                    source=None,
+                    text=None,
+                    blocked=f"rewrite_unproven:{layer}",
+                )
         return ApplyDecision(
             outcome="applied_rewrite",
             applied=True,
@@ -198,11 +213,17 @@ def finish_delegated_job(
     verdict: ReviewVerdict | None,
 ) -> JobResult:
     """Apply gate after local attempts (scripted or MCP) already ran."""
-    decision = route(job.signals)
+    files = job.eval_case.files if job.eval_case is not None else ()
+    decision = route(job.signals, files)
     if decision.action != "delegate":
         raise ValueError(f"{job.id}: finish_delegated_job requires a delegated route")
     last = attempts[-1] if attempts else None
-    apply = decide_apply(delegated=True, last=last, verdict=verdict)
+    apply = decide_apply(
+        delegated=True,
+        last=last,
+        verdict=verdict,
+        eval_case=job.eval_case,
+    )
     return _result(job, decision, apply, attempts, verdict)
 
 
@@ -212,7 +233,8 @@ def run_job(
     generate: GenerateFn | None = None,
     reviewer: ReviewFn | ScriptedReviewer | None = None,
 ) -> JobResult:
-    decision: RouteDecision = route(job.signals)
+    files = job.eval_case.files if job.eval_case is not None else ()
+    decision: RouteDecision = route(job.signals, files)
     if decision.action == "keep":
         apply = decide_apply(
             delegated=False,
@@ -232,24 +254,48 @@ def run_job(
         raise ValueError(f"{job.id}: delegated jobs need an eval_case")
 
     local = generate if generate is not None else ScriptedLocal(job.local_replies)
-    review_fn: ReviewFn
+    scripted: ScriptedReviewer | None = None
     if reviewer is None:
         scripted = ScriptedReviewer(job.review)
-        review_fn = scripted.review
+        review_fn: ReviewFn = scripted.review_packet
     elif isinstance(reviewer, ScriptedReviewer):
-        review_fn = reviewer.review
+        review_fn = reviewer.review_packet
     else:
         review_fn = reviewer
 
     attempts = run_local_loop(job.eval_case, local, job=job.id)
     last = attempts[-1] if attempts else None
-    local_text = last.text if last else ""
-    local_passed = last.passed if last else False
-    verdict = review_fn(job.id, local_text, local_passed)
-    apply = decide_apply(delegated=True, last=last, verdict=verdict)
+    packet = _packet_for(job, last)
+    verdict = _invoke_review(review_fn, packet)
+    apply = decide_apply(
+        delegated=True,
+        last=last,
+        verdict=verdict,
+        eval_case=job.eval_case,
+    )
     return _result(job, decision, apply, attempts, verdict)
 
 
+def _packet_for(job: OrchestrationJob, last: LocalAttempt | None) -> ReviewPacket:
+    layers = dict(last.record.layers) if last is not None else {}
+    return ReviewPacket(
+        job_id=job.id,
+        local_text=last.text if last is not None else "",
+        passed=last.passed if last is not None else False,
+        layers=layers,
+        first_failure=None if last is None else last.record.first_failure,
+        local_review_notes=job.local_review_notes,
+        case_id=job.eval_case.id if job.eval_case is not None else job.id,
+    )
+
+
+def _invoke_review(review_fn: ReviewFn, packet: ReviewPacket) -> ReviewVerdict | None:
+    try:
+        return review_fn(packet)
+    except TypeError:
+        return review_fn(packet.job_id, packet.local_text, packet.passed)
+
+
 def _result(
     job: OrchestrationJob,
     decision: RouteDecision,

src/local_coding_slm/eval/review.py

@@ -4,17 +4,34 @@
 Apply still requires a verdict from the premium orchestrator
 (accept / rewrite / reject). CI uses a scripted stand-in so the contract
 is testable without calling Cursor, GPT, or Claude.
+
+The handoff into review is a ``ReviewPacket``: local text, layer scores,
+and optional local_review notes. The premium model is supposed to read
+that packet, not only the raw tool string.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 PREMIUM_REVIEWER = "premium"
 LOCAL_REVIEW_TOOL = "local_review"
 DECISIONS = frozenset({"accept", "rewrite", "reject"})
 
 
+@dataclass(frozen=True)
+class ReviewPacket:
+    """What the premium reviewer sees after local_* returns."""
+
+    job_id: str
+    local_text: str
+    passed: bool
+    layers: dict[str, str] = field(default_factory=dict)
+    first_failure: str | None = None
+    local_review_notes: str = ""
+    case_id: str = ""
+
+
 @dataclass(frozen=True)
 class ReviewVerdict:
     decision: str
@@ -49,8 +66,13 @@ class ScriptedReviewer:
     def __init__(self, verdict: ReviewVerdict | None) -> None:
         self.verdict = verdict
         self.calls = 0
+        self.last_packet: ReviewPacket | None = None
 
     def review(self, job_id: str, local_text: str, passed: bool) -> ReviewVerdict | None:
         del job_id, local_text, passed
         self.calls += 1
         return self.verdict
+
+    def review_packet(self, packet: ReviewPacket) -> ReviewVerdict | None:
+        self.last_packet = packet
+        return self.review(packet.job_id, packet.local_text, packet.passed)

src/local_coding_slm/eval/routing.py

@@ -3,11 +3,13 @@
 The premium agent still chooses the signals. This module is the contract
 those signals must satisfy: mechanical work may be delegated; incident,
 architectural, and live-tool work stays on the premium model. Delegated
-work always requires a later premium review before apply.
+work always requires a later premium review before apply. Secret files
+never go to local_* even when the task looks mechanical.
 """
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 
@@ -42,8 +44,28 @@ def mechanical_signals(**overrides: bool) -> RouteSignals:
     return RouteSignals(**values)
 
 
-def route(signals: RouteSignals) -> RouteDecision:
-    """Return keep vs delegate. Do-not-delegate flags win over mechanical ones."""
+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
+
+
+def route(
+    signals: RouteSignals,
+    files: Sequence[dict[str, str]] | None = None,
+) -> RouteDecision:
+    """Return keep vs delegate. Do-not-delegate flags and secret files win."""
+    blocked = payload_block_reason(files)
+    if blocked:
+        return RouteDecision("keep", blocked, False)
     if signals.incident_debug:
         return RouteDecision("keep", "incident_debug", False)
     if signals.architectural: