fix(specialist): stop rejecting our own proposals on a format rule we ship broken

src/hyperloom/inference_optimizer/tests/test_critic_agent_backend.py

@@ -32,6 +32,7 @@
     _reviewed_msg_ids_from_bundle,
     _verdict_references_kb,
 )
+from hyperloom.orchestrator.specialists.patch_safety import FORBIDDEN_PROPOSAL_FIELDS
 from hyperloom.inference_optimizer.protocol.intent import IntentType
 
 
@@ -922,6 +923,54 @@ async def test_user_prompt_includes_judge_bundle_and_instructions(
     assert '"abc"' in user_text  # proposal msg_id from judge bundle
 
 
+@pytest.mark.asyncio
+async def test_the_reviewed_bundle_carries_the_quantitative_claim_rule(
+    fake_critic_root: Path,
+    fake_session_dir: Path,
+):
+    """Delivered as data so the Critic's field list stays identical to the one
+    the runner strips, and so a format slip is advisory rather than a reject
+    that costs the round every proposal in the set."""
+    judge_bundle = {
+        "kind": "coordinator_inbox",
+        "merged_context": {"model": "m", "framework": "sglang"},
+        "proposals": [
+            {
+                "msg_id": "abc",
+                "from_agent": "orchestration",
+                "action_name": "baseline",
+                "payload": {},
+                "predicted_gain_pct": 0.0,
+            }
+        ],
+        "kb_priors_by_proposal": {"abc": []},
+        "kb_read_skipped_reason": None,
+        "review_constraints": {},
+        "notes": [],
+        "missing_context": [],
+        "required_context": [],
+    }
+    reply = '{"review_verdicts": [{"target_proposal_msg_id": "abc", "verdict": "approve"}]}'
+    backend, client = _make_backend(
+        fake_critic_root,
+        fake_session_dir,
+        codex_replies=[reply],
+        judge_bundle=judge_bundle,
+    )
+    await backend.run("ignored", system_prompt="you are critic")
+
+    user_text = client.completions.calls[0]["messages"][1]["content"]
+    match = re.search(
+        r"==== JUDGE BUNDLE ====\s*(\{.*?\})\s*==== END JUDGE BUNDLE ====",
+        user_text,
+        re.DOTALL,
+    )
+    assert match
+    rule = json.loads(match.group(1))["review_constraints"]["quantitative_claim_rule"]
+    assert rule["failure_verdict"] == "advise"
+    assert set(rule["forbidden_proposal_fields"]) == set(FORBIDDEN_PROPOSAL_FIELDS)
+
+
 # Static context propagation — backend sources model/framework from manifest.json or explicit static_context.
 def _write_manifest(session_dir: Path, payload: dict[str, Any]) -> Path:
     """Write a minimal manifest.json the backend can ingest."""

src/hyperloom/inference_optimizer/tests/test_critic_verdict_map.py

@@ -29,6 +29,10 @@
     PolicyGate,
     REVIEW_VERDICTS,
 )
+from hyperloom.orchestrator.specialists.patch_safety import (
+    QUANTITATIVE_CLAIM_REASON_CODE,
+    cross_domain_rule_descriptors,
+)
 
 
 # 1. intent_parser — envelope schema accepts verdict OR verdict_map
@@ -506,6 +510,152 @@
     assert "advice=" not in line
 
 
+# 3b. A reject on a rule that asked for advice is held to that rule
+@pytest.mark.asyncio
+async def test_reject_on_an_advisory_only_rule_is_held_to_advise(coord):
+    """The quantitative-claim rule declares ``advise``; a reject citing it must not end the proposal."""
+    pending = PendingProposal(
+        proposal_msg_id="msg-held",
+        from_agent="orchestration",
+        action_name="specialist",
+        predicted_gain_pct=0.0,
+        payload={"action_name": "specialist", "params": {"task_id": "t-1"}},
+    )
+    coord.state.pending_proposals["msg-held"] = pending
+    intent = Intent(
+        type=IntentType.REVIEW_VERDICT,
+        payload={
+            "target_proposal_msg_id": "msg-held",
+            "verdict": "reject",
+            "failure_reason_code": QUANTITATIVE_CLAIM_REASON_CODE,
+            "reasoning": "proposal carried confidence",
+        },
+    )
+    await coord._handle_review_verdict("critic", intent)
+    assert pending.verdict == "advise"
+    # advise materialises, so the round keeps the proposal.
+    assert len(coord._materialise_calls) == 1
+    assert [m for m in coord.bus.messages if m.topic == "review_verdict"][0].payload["verdict"] == "advise"
+
+
+@pytest.mark.asyncio
+async def test_a_held_reject_is_recorded_not_silently_corrected(coord, caplog):
+    """The downgrade leaves both a log line and an observation, so prompt drift stays visible."""
+    import logging
+
+    pending = _seed_explore_proposal(coord, msg_id="msg-audit")
+    intent = Intent(
+        type=IntentType.REVIEW_VERDICT,
+        payload={
+            "target_proposal_msg_id": "msg-audit",
+            "verdict": "reject",
+            "failure_reason_code": QUANTITATIVE_CLAIM_REASON_CODE,
+        },
+    )
+    with caplog.at_level(logging.WARNING, logger="hyperloom.orchestrator.loop.intent_router"):
+        await coord._handle_review_verdict("critic", intent)
+    assert any("held to its rule" in r.getMessage() for r in caplog.records)
+    kinds = [call.args[2].get("kind") for call in coord._record_observation.await_args_list]
+    assert "verdict_downgraded_to_rule_verdict" in kinds
+
+
+@pytest.mark.asyncio
+async def test_a_cross_domain_hint_reject_is_held_too(coord):
+    """Every rule declaring ``advise`` is covered, not just the quantitative-claim one."""
+    reason_code = cross_domain_rule_descriptors()[0]["failure_reason_code"]
+    pending = PendingProposal(
+        proposal_msg_id="msg-xd",
+        from_agent="orchestration",
+        action_name="specialist",
+        predicted_gain_pct=0.0,
+        payload={"action_name": "specialist", "params": {"task_id": "t-2"}},
+    )
+    coord.state.pending_proposals["msg-xd"] = pending
+    intent = Intent(
+        type=IntentType.REVIEW_VERDICT,
+        payload={
+            "target_proposal_msg_id": "msg-xd",
+            "verdict": "reject",
+            "failure_reason_code": reason_code,
+        },
+    )
+    await coord._handle_review_verdict("critic", intent)
+    assert pending.verdict == "advise"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "payload_extra",
+    [
+        pytest.param({}, id="no_reason_code"),
+        # A code no rule declares as advisory — e.g. the safety hard guard,
+        # which critic.md keeps at ``reject``.
+        pytest.param({"failure_reason_code": "specialist_patch_not_grounded"}, id="code_outside_the_advisory_set"),
+    ],
+)
+async def test_a_substantive_reject_still_rejects(coord, payload_extra):
+    """The backstop is scoped to rules that declared ``advise``; every other reject stands."""
+    pending = PendingProposal(
+        proposal_msg_id="msg-real",
+        from_agent="orchestration",
+        action_name="specialist",
+        predicted_gain_pct=0.0,
+        payload={"action_name": "specialist", "params": {"task_id": "t-3"}},
+    )
+    coord.state.pending_proposals["msg-real"] = pending
+    intent = Intent(
+        type=IntentType.REVIEW_VERDICT,
+        payload={
+            "target_proposal_msg_id": "msg-real",
+            "verdict": "reject",
+            **payload_extra,
+        },
+    )
+    await coord._handle_review_verdict("critic", intent)
+    assert pending.verdict == "reject"
+    assert coord._materialise_calls == []
+
+
+@pytest.mark.asyncio
+async def test_one_variant_held_to_advise_does_not_out_rank_its_siblings(coord):
+    """The hold runs per variant before the collapse, so an advisory-only reject cannot discard the set."""
+    pending = _seed_explore_proposal(coord, msg_id="msg-grid", variants=["v_a", "v_b"])
+    intent = Intent(
+        type=IntentType.REVIEW_VERDICT,
+        payload={
+            "target_proposal_msg_id": "msg-grid",
+            "verdict_map": {
+                "v_a": {"verdict": "advise", "rationale": "worth a look"},
+                "v_b": {
+                    "verdict": "reject",
+                    "failure_reason_code": QUANTITATIVE_CLAIM_REASON_CODE,
+                },
+            },
+        },
+    )
+    await coord._handle_review_verdict("critic", intent)
+    # Without the per-entry hold, reject out-ranks advise and the set is lost.
+    assert pending.verdict == "advise"
+    assert len(coord._materialise_calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_a_grid_rejected_only_on_advisory_rules_survives(coord):
+    """A whole map rejected on advisory-only grounds collapses to advise, not reject."""
+    pending = _seed_explore_proposal(coord, msg_id="msg-grid-all", variants=["v_a", "v_b"])
+    entry = {"verdict": "reject", "failure_reason_code": QUANTITATIVE_CLAIM_REASON_CODE}
+    intent = Intent(
+        type=IntentType.REVIEW_VERDICT,
+        payload={
+            "target_proposal_msg_id": "msg-grid-all",
+            "verdict_map": {"v_a": dict(entry), "v_b": dict(entry)},
+        },
+    )
+    await coord._handle_review_verdict("critic", intent)
+    assert pending.verdict == "advise"
+    assert len(coord._materialise_calls) == 1
+
+
 # 4. _materialize_approved_proposal — filter semantics (unit)
 @pytest.mark.asyncio
 async def test_materialize_filter_drops_rejected_variants(tmp_path: Path):

src/hyperloom/inference_optimizer/tests/test_specialist_patch_safety_unit.py

@@ -7,6 +7,8 @@
 from __future__ import annotations
 
 
+import pytest
+
 from hyperloom.orchestrator.specialists import patch_safety as ps
 
 
@@ -166,16 +168,17 @@ def test_patch_safety_report_notes_empty():
 # ---- scan_quantitative_claims ---------------------------------------------
 def test_scan_quantitative_claims():
     payload = {
-        "confidence": 0.9,
+        "expected_gain": 12.0,
         "summary": "gives 20% boost",
         "proposal_set": [
-            {"score": 1, "expected_qualitative_argument": "3x faster"},
+            {"score": 1, "confidence": 0.9, "expected_qualitative_argument": "3x faster"},
             "not-a-dict",
         ],
     }
     forbidden, warnings = ps.scan_quantitative_claims(payload)
-    assert "confidence" in forbidden
+    assert "expected_gain" in forbidden
     assert "score" in forbidden
+    assert "confidence" in forbidden
     assert any("%" in w for w in warnings)
     assert any("x" in w.lower() for w in warnings)
 
@@ -184,6 +187,94 @@ def test_scan_quantitative_claims_empty():
     assert ps.scan_quantitative_claims({}) == ([], [])
 
 
+def test_round_level_confidence_is_not_a_per_proposal_gain_claim():
+    """The output schema asks for it and the round audit records it, so flagging
+    it at the top level only made our own template a violation."""
+    forbidden, _ = ps.scan_quantitative_claims({"confidence": 0.6})
+
+    assert forbidden == []
+
+
+# ---- strip_forbidden_proposal_fields --------------------------------------
+def test_forbidden_fields_are_stripped_so_the_critic_cannot_reject_on_format():
+    payload = {
+        "expected_gain": 9.0,
+        "confidence": 0.6,
+        "summary": "keep me",
+        "proposal_set": [
+            {"name": "v1", "confidence": 0.4, "score": 3, "reason": "keep me too"},
+            "not-a-dict",
+        ],
+    }
+
+    removed = ps.strip_forbidden_proposal_fields(payload)
+
+    assert set(removed) == {"expected_gain", "confidence", "score"}
+    assert "expected_gain" not in payload
+    assert payload["confidence"] == 0.6  # round-level self-assessment survives
+    assert payload["summary"] == "keep me"
+    assert payload["proposal_set"][0] == {"name": "v1", "reason": "keep me too"}
+    assert payload["proposal_set"][1] == "not-a-dict"
+
+
+def test_stripping_a_clean_payload_changes_nothing():
+    payload = {"proposal_set": [{"name": "v1", "reason": "why"}]}
+
+    assert ps.strip_forbidden_proposal_fields(payload) == []
+    assert payload == {"proposal_set": [{"name": "v1", "reason": "why"}]}
+
+
+@pytest.mark.parametrize("payload", [None, [], "", 0])
+def test_stripping_tolerates_a_payload_that_is_not_a_dict(payload):
+    assert ps.strip_forbidden_proposal_fields(payload) == []
+
+
+# ---- quantitative_claim_rule_descriptor ------------------------------------
+def test_the_rule_the_critic_gets_lists_exactly_what_the_runner_strips():
+    """A hand-copied field list in the prompt is how the Critic came to reject
+    over a field the runner never enforced."""
+    rule = ps.quantitative_claim_rule_descriptor()
+
+    assert set(rule["forbidden_proposal_fields"]) == set(ps.FORBIDDEN_PROPOSAL_FIELDS)
+
+
+def test_a_format_slip_is_advisory_not_a_reject():
+    rule = ps.quantitative_claim_rule_descriptor()
+
+    assert rule["failure_verdict"] == "advise"
+    assert rule["failure_reason_code"] == ps.QUANTITATIVE_CLAIM_REASON_CODE
+
+
+# ---- advisory_only_reason_codes --------------------------------------------
+def test_every_rule_that_asked_for_advice_is_enforceable():
+    codes = ps.advisory_only_reason_codes()
+
+    assert ps.QUANTITATIVE_CLAIM_REASON_CODE in codes
+    for rule in ps.cross_domain_rule_descriptors():
+        if rule["failure_verdict"] == ps.ADVISE_VERDICT:
+            assert rule["failure_reason_code"] in codes
+
+
+def test_a_rule_asking_for_a_reject_is_left_alone(monkeypatch):
+    """The set is derived from the descriptors, so a rule keeping ``reject`` stays out of it."""
+    hard = ps.CrossDomainRule(
+        rule_id="hard_guard",
+        description="a violation here is grounds for refusal",
+        failure_verdict="reject",
+        failure_reason_code="cross_domain_hard_guard",
+    )
+    monkeypatch.setattr(ps, "CROSS_DOMAIN_RULES", ps.CROSS_DOMAIN_RULES + (hard,))
+
+    codes = ps.advisory_only_reason_codes()
+
+    assert "cross_domain_hard_guard" not in codes
+    assert ps.QUANTITATIVE_CLAIM_REASON_CODE in codes
+
+
+def test_the_codes_carry_no_blank_entry():
+    assert "" not in ps.advisory_only_reason_codes()
+
+
 # ---- vet_patches ----------------------------------------------------------
 def test_vet_patches(tmp_path, monkeypatch):
     good = tmp_path / "good.patch"

src/hyperloom/inference_optimizer/tests/test_specialist_prompt_builder_coverage_unit.py

@@ -173,6 +173,21 @@ def test_cold_start_directive():
     assert "COLD-START MODE" in user_p
 
 
+def test_cold_start_does_not_ask_for_a_field_the_safety_gate_forbids():
+    """It used to direct the fallback proposals to carry ``confidence: low`` --
+    a field in FORBIDDEN_PROPOSAL_FIELDS -- so a compliant specialist tripped
+    the guard on exactly the round where it was the only source of ideas."""
+    inp = SpecialistPromptInputs(
+        task_id="t",
+        domain=get_domain("serving_specialist"),
+    )
+    _, user_p = build_specialist_prompts(inp)
+
+    cold_start = user_p[user_p.index("COLD-START MODE") :]
+    assert "confidence: low" not in cold_start
+    assert "provenance: domain_focus_default" in cold_start
+
+
 def test_research_hints_fallback():
     inp = SpecialistPromptInputs(
         task_id="t",