feat(refiner): the refine loop now gates on blast radius too (G3)

.env.example

@@ -24,6 +24,7 @@ GITHUB_TOKEN=        # repo-scoped PAT for opening PRs (or use `gh auth login`)
 # VPCOPILOT_REQUIRE_PROBE=1            # fail-closed: don't validate a finding that has no derived probe
 # VPCOPILOT_ACTOR=jane.doe             # audit-log attribution — default: the OS user; set it in CI
 # VPCOPILOT_SIM_THRESHOLD=0.01         # would-block rate that flags a policy in `vpcopilot simulate`
+# VPCOPILOT_SIM_LOGS=traffic.jsonl     # traffic sample the refiner also checks each candidate against
 # --- Validation auth (auth-protected targets: let the probe log in so it can demonstrate the exploit) ---
 # VPCOPILOT_PROBE_USER=                # username the validation probe logs in with (cookie or token app)
 # VPCOPILOT_PROBE_PASS=                # its password

ROADMAP.md

@@ -150,12 +150,21 @@ Live validation on the LB stays. This runs before it, not instead of it.
     convenience: it needs its own test file, a redaction count in `simulation.json`, and a
     `caveats` line in `export.build_manifest` stating what was stripped and what was not.
 
-- [ ] **G3** Simulation feeds the refiner. (S, P2) Depends on G2 (landed).
-  `refiner.py` retries until the band-aid blocks the exploit. Add the second half: retry
-  until it blocks the exploit **and** stays under the false-positive threshold. Today a
-  refinement widening the policy to catch the exploit has nothing telling it it went too far.
+- [x] **G3** Simulation feeds the refiner. (S, P2) — **DONE:** the refine loop now accepts a
+  candidate only when it blocks the exploit **and** stays under the blast-radius threshold. An
+  over-broad candidate is fed back as the `over_block` diagnosis the refine agent already
+  understands, with the top blocked paths attached so it knows *what* it over-blocked.
   - Acceptance: a refinement pass increasing would-block rate past the threshold gets
-    rejected and the refiner tries again or gives up honestly, matching existing behavior.
+    rejected and the refiner tries again or gives up honestly, matching existing behavior ✅
+  - **Measured where it is already attached.** The refiner has just confirmed the candidate is
+    enforcing, so `simulate.measure_attached` replays through that LB rather than paying a second
+    attach and propagation wait elsewhere — slower-but-correct per the decision of 2026-07-27, but
+    only one round trip per attempt rather than two.
+  - `_score` is shared by `simulate_policies` and `measure_attached`, so the standalone run and the
+    refiner cannot drift on what "too broad" means.
+  - **Sample comes from `VPCOPILOT_SIM_LOGS`** (or an explicit `records=`). With neither, the second
+    gate is skipped and the loop behaves exactly as before — pinned by a test that fails if anything
+    is replayed without a sample. An unreadable sample logs a warning and refines without the gate.
 
 - [ ] **G4** Published model scorecard. (M, P1)
   `D3` proved a config-only swap on `gpt-4o`. Turn that into a committed table across four

docs/USAGE.md

@@ -81,6 +81,11 @@ vpcopilot report --open   # standalone shareable HTML dashboard of the results
 vpcopilot retire --finding <id>   # C2: when the cure PR merges, detach the band-aid + mark retired
 vpcopilot retire --all            # retire every mitigated finding whose cure PR merged (--force to skip the check)
 ```
+**Refine with a blast-radius gate.** Set `VPCOPILOT_SIM_LOGS` to a traffic sample and the
+refiner stops at the first policy that blocks the exploit *and* stays under the threshold, instead
+of the first that merely blocks it — a refinement that widened the rule too far is fed back as
+`over_block` and retried. With the variable unset the refine loop behaves exactly as before.
+
 `export` writes `<out>/audit-bundle.zip` (`--all` → `<root>/audit-bundle-all.zip` with a top-level
 `index.json`). Inside: `manifest.json` (bundle identity, the run manifest, a SHA-256 per member, and
 an explicit `caveats` list), `audit.csv` + `audit-events.json` (one normalized row per change, joined

src/vpcopilot/console/static/index.html

@@ -502,6 +502,7 @@
   if(res.attempts>1) bits.push(`<span class="badge">self-healed in ${res.attempts} attempts</span>`);
   else if(res.attempts) bits.push(`${res.attempts} attempt`);
   if(res.kept) bits.push("· KEPT LIVE"); else if(res.rolled_back) bits.push("(rolled back)");
+  if(res.blast_radius) bits.push(`<span class="muted">· blast radius ${esc(res.blast_radius.would_block)}/${esc(res.blast_radius.evaluated)} recorded (${(res.blast_radius.block_rate*100).toFixed(1)}%)</span>`);
   if(res.unfixable) bits.push('<span class="nob">unfixable — ship the code fix</span>');
   const _before=((res.before_after||{}).before)||{}, _bst=_before.exploit_status;
   if(_before.auth_failed)

src/vpcopilot/refiner.py

@@ -38,14 +38,51 @@ def _load_finding(out_dir: str, finding_id: str | None) -> Finding | None:
     return None
 
 
+def _sim_threshold_default() -> float:
+    from .simulate import DEFAULT_THRESHOLD
+    try:
+        return float(os.environ.get("VPCOPILOT_SIM_THRESHOLD", DEFAULT_THRESHOLD))
+    except ValueError:
+        return DEFAULT_THRESHOLD
+
+
+def _resolve_records(records, log: Callable):
+    """The sample the second gate measures against: an explicit list, else `VPCOPILOT_SIM_LOGS`.
+    With neither there is no gate — fail-soft, because a missing or unreadable sample must never
+    turn a working refine loop into a failure."""
+    if records:
+        return records
+    path = (os.environ.get("VPCOPILOT_SIM_LOGS") or "").strip()
+    if not path:
+        return None
+    try:
+        from .traffic import load
+        recs, _ = load(path)
+        log(f"blast-radius gate: {len(recs)} record(s) from {path}")
+        return recs or None
+    except Exception as e:  # noqa: BLE001
+        log(f"  ⚠ could not read the traffic sample at {path}: {e} — refining without the gate")
+        return None
+
+
 def refine_apply_service_policy(artifact_path: str, lb: str, target_url: str, *,
                                 finding_id: str | None = None, name: str | None = None,
                                 max_refine: int | None = None, keep: bool = False,
                                 allow_protected: bool = False, config_path: str | None = None,
                                 retries: int = 6, wait_seconds: int = 8,
+                                records: list | None = None, sim_threshold: float | None = None,
                                 out_dir: str = "out", log: Callable = print) -> dict:
     """Create/attach a service policy, validate it live, and refine-until-it-works (or give up
-    honestly). Returns passed + attempts + before/after; persists the WORKING spec to the artifact."""
+    honestly). Returns passed + attempts + before/after; persists the WORKING spec to the artifact.
+
+    G3 — with a traffic sample, "works" means two things, not one: the policy blocks the exploit
+    **and** stays under the blast-radius threshold. Without that second gate a refinement that
+    widens the policy far enough to catch the exploit has nothing telling it it went too far, and
+    the loop happily accepts a rule that would deny production. An over-broad result is fed back as
+    the `over_block` diagnosis the refine agent already understands.
+
+    `records` (or `VPCOPILOT_SIM_LOGS`) supplies the sample. With neither, the second gate is
+    skipped entirely and this behaves exactly as it did before G3."""
     xc = XC()
     if lb in _protected_lbs() and not allow_protected:
         raise RuntimeError(f"refusing to mutate protected LB '{lb}'. Pass allow_protected=True to override.")
@@ -80,6 +117,9 @@ def attach():
     def detach():
         _put_lb(copy.deepcopy(orig_spec))
 
+    records = _resolve_records(records, log)
+    threshold = sim_threshold if sim_threshold is not None else _sim_threshold_default()
+
     h = Harness(config_path) if finding else None
     exploit = (probe or {}).get("exploit")
     before = _run_validation(target_url, finding_id, out_dir, probe_negative_pay, log)
@@ -142,7 +182,17 @@ def detach():
             if result["exploit_blocked"] and result["legit_ok"]:
                 break
 
-        if result["exploit_blocked"] and result["legit_ok"]:
+        blast = None
+        if result["exploit_blocked"] and result["legit_ok"] and records:
+            # The policy is attached and confirmed enforcing right now — measure it here rather
+            # than paying a second attach + propagation wait on another LB.
+            from .simulate import measure_attached
+            blast = measure_attached(records, target_url, policy_name=policy_name,
+                                     finding_id=finding_id or "", threshold=threshold, log=log)
+            log(f"  blast radius: would block {blast['would_block']}/{blast['evaluated']} "
+                f"recorded request(s) ({blast['block_rate']:.1%})")
+
+        if result["exploit_blocked"] and result["legit_ok"] and not (blast and blast["blocked_promotion"]):
             log(f"validation PASS on attempt {attempt} ✓")
             Path(artifact_path).write_text(json.dumps({"metadata": {"name": policy_name}, "spec": spec}, indent=2))
             rolled = False
@@ -153,18 +203,27 @@ def detach():
                 rolled = True
             audit.record(out_dir, "refine_apply", finding_id=finding_id, namespace=xc.ns, control="service_policy", policy=policy_name, lb=lb,
                          passed=True, attempts=attempt, rolled_back=rolled,
-                         before_after={"before": before, "after": result})
-            return {"mode": "refine_apply", "control": "service_policy", "policy": policy_name,
-                    "passed": True, "attempts": attempt, "rolled_back": rolled, "kept": keep,
-                    "before_after": {"before": before, "after": result}}
+                         before_after={"before": before, "after": result}, blast_radius=blast)
+            out = {"mode": "refine_apply", "control": "service_policy", "policy": policy_name,
+                   "passed": True, "attempts": attempt, "rolled_back": rolled, "kept": keep,
+                   "before_after": {"before": before, "after": result}}
+            if blast:
+                out["blast_radius"] = blast
+            return out
 
         # failed this attempt — detach, diagnose, refine
         detach()
-        diagnosis = "exploit_not_blocked" if not result["exploit_blocked"] else "over_block"
-        log(f"attempt {attempt}: FAIL ({diagnosis}, exploit_status={result['exploit_status']})")
+        agent_result = result
+        if blast and blast["blocked_promotion"]:
+            diagnosis = "over_block"
+            agent_result = {**result, "blast_radius": blast}
+            log(f"attempt {attempt}: FAIL (over_block — {blast['reason']})")
+        else:
+            diagnosis = "exploit_not_blocked" if not result["exploit_blocked"] else "over_block"
+            log(f"attempt {attempt}: FAIL ({diagnosis}, exploit_status={result['exploit_status']})")
         if attempt == max_refine or not (h and finding):
             break
-        refined = refine_agent.run(h, finding, "service_policy", spec, probe, result, diagnosis)
+        refined = refine_agent.run(h, finding, "service_policy", spec, probe, agent_result, diagnosis)
         log(f"  refined: {refined.rationale}" + (" [UNFIXABLE]" if refined.unfixable else ""))
         if refined.unfixable:
             audit.record(out_dir, "refine_apply", finding_id=finding_id, namespace=xc.ns, control="service_policy", policy=policy_name, lb=lb,

src/vpcopilot/simulate.py

@@ -140,6 +140,44 @@ def _replay(records: list[RequestRecord], url: str,
     return len(records) - errored, blocked, errored
 
 
+def _score(sim: PolicySimulation, total: int, blocked: list, errored: int, threshold: float) -> PolicySimulation:
+    """Fill a verdict from a replay. Shared by the standalone run and the refiner's per-attempt
+    check so the two can never drift on what "too broad" means."""
+    sim.errored = errored
+    sim.evaluated = total
+    sim.would_block = len(blocked)
+    sim.block_rate = round(len(blocked) / total, 4) if total else 0.0
+    sim.threshold = threshold
+    sim.samples = [b.model_dump(exclude={"headers"}) for b in blocked[:MAX_SAMPLES]]
+    sim.top_paths = [list(x) for x in Counter(b.path for b in blocked).most_common(5)]
+    sim.top_user_agents = [list(x) for x in
+                           Counter(b.user_agent for b in blocked if b.user_agent).most_common(5)]
+    if not total:
+        sim.reason = ("nothing measured — every replayed request failed in transit, so this is "
+                      "zero evidence, not a clean result")
+    elif sim.block_rate > threshold:
+        sim.blocked_promotion = True
+        sim.reason = (f"would block {sim.would_block}/{total} recorded requests "
+                      f"({sim.block_rate:.1%}), over the {threshold:.1%} threshold")
+    return sim
+
+
+def measure_attached(records: list[RequestRecord], url: str, *, policy_name: str = "",
+                     finding_id: str = "", threshold: float = DEFAULT_THRESHOLD,
+                     max_records: int | None = None, log: Callable = print) -> dict:
+    """G3: blast radius of the policy that is ALREADY attached and validating.
+
+    The refiner has just attached a candidate and confirmed it blocks the exploit, so there is no
+    second attach, no second propagation wait, and no second LB — we measure the exact policy that
+    just passed, where it passed. Returns a plain dict so it drops straight into the refine agent's
+    result payload, the audit record and the apply result."""
+    sample = records[:max_records] if max_records else records
+    total, blocked, errored = _replay(sample, url, log)
+    sim = _score(PolicySimulation(policy_name=policy_name, finding_id=finding_id), total, blocked,
+                 errored, threshold)
+    return sim.model_dump(exclude={"samples", "top_user_agents", "enforcement_confirmed", "error"})
+
+
 def simulate_policies(candidates: list[dict], records: list[RequestRecord], *, lb: str, url: str,
                       out_dir: str = "out", threshold: float = DEFAULT_THRESHOLD,
                       max_records: int | None = None, source: str = "", window: str = "",
@@ -186,26 +224,11 @@ def simulate_policies(candidates: list[dict], records: list[RequestRecord], *, l
                 wait_seconds=wait_seconds, retries=retries, sleep=ctx.sleep)
             log(f"  replaying {len(sample)} record(s)")
             total, blocked, errored = _replay(sample, url, log)
-            sim.errored = errored
             if errored:
                 log(f"  {errored} record(s) failed in transit — excluded from the rate")
-            sim.evaluated = total
-            sim.would_block = len(blocked)
-            sim.block_rate = round(len(blocked) / total, 4) if total else 0.0
-            sim.samples = [b.model_dump(exclude={"headers"}) for b in blocked[:MAX_SAMPLES]]
-            sim.top_paths = [list(x) for x in Counter(b.path for b in blocked).most_common(5)]
-            sim.top_user_agents = [list(x) for x in
-                                   Counter(b.user_agent for b in blocked if b.user_agent).most_common(5)]
-            if not total:
-                sim.reason = ("nothing measured — every replayed request failed in transit, so this "
-                              "is zero evidence, not a clean result")
-            if total and sim.block_rate > threshold:
-                sim.blocked_promotion = True
-                sim.reason = (f"would block {sim.would_block}/{total} recorded requests "
-                              f"({sim.block_rate:.1%}), over the {threshold:.1%} threshold")
-                log(f"  ⚠ {name}: {sim.reason}")
-            else:
-                log(f"  {name}: would block {sim.would_block}/{total}")
+            _score(sim, total, blocked, errored, threshold)
+            log(f"  ⚠ {name}: {sim.reason}" if sim.blocked_promotion
+                else f"  {name}: would block {sim.would_block}/{total}")
         except Exception as e:  # noqa: BLE001 — a replay failure is reported, never raised past restore
             sim.error = str(e)
             log(f"  ⚠ {name}: simulation failed — {e}")