feat: wire diagnostic costs into profile triage pairs

PILOT.md

@@ -1849,3 +1849,10 @@ Six API tests, including changed-cost cache invalidation, three new CLI integrat
 ## VCR407 — Coding-skill guidance for diagnostic costs
 
 The bundled English coding workflow explains optional caller-verified relative check costs, omission of unverified estimates, direct execution of decisive local checks and separation from causal ranking. It preserves required test execution and local fallback. This documents the merged API/CLI without claiming measured benefit or changing the published release. Nine skill-pack tests and three coding-workflow tests pass. Hosted CI remains required before merge.
+
+
+## VCR408 — Diagnostic costs wired into profile triage trials
+
+Future paired trials may declare caller-verified relative diagnostic costs in a case profile for supplied hypothesis IDs only. Both arms receive the same cost disclosure; the treatment arm is told costs may guide next-check ordering only, are never causal-likelihood evidence, and never waive required checks. Costs reach the remote decision only as fixed low/medium/high/unknown enum tokens through the validated bridge spec, the exact triage command, the intercepted shim request, and the production triage call, keeping transport bridge-counted. Cost-bearing requests are rejected by the bridge and shim unless they match the supervisor-validated profile. Profiles without costs keep identical prompts, commands, requests, and decision state. The supervisor triage command now orders observation flags before `--rank-hypotheses` to match the bridge command and shim, closing a silent-fallback path for exact rank-plus-observation requests in future trials; no historical trial is rescored or rerun.
+
+This is prospective trial instrumentation; no pair was run. Ten new wiring tests plus updated doubles pass, including parity and interception for rank-plus-observation commands, and the full 858-test suite passes locally with `git diff --check` clean. Hosted CI remains required before merge. No efficacy, delivery, or release claim is made.

TASKS.md

@@ -1253,3 +1253,10 @@ Six API tests, including changed-cost cache invalidation, three new CLI integrat
 ## VCR407 — Coding-skill guidance for diagnostic costs
 
 The bundled English coding workflow explains optional caller-verified relative check costs, omission of unverified estimates, direct execution of decisive local checks and separation from causal ranking. It preserves required test execution and local fallback. This documents the merged API/CLI without claiming measured benefit or changing the published release. Nine skill-pack tests and three coding-workflow tests pass. Hosted CI remains required before merge.
+
+
+## VCR408 — Diagnostic costs wired into profile triage trials
+
+Case profiles may declare caller-verified relative diagnostic costs for supplied hypothesis IDs. Both arms receive the same cost disclosure and the treatment arm is told costs may guide next-check ordering only, are never causal-likelihood evidence, and never waive required checks. The validated plain-string costs flow through the bridge spec, the exact triage command, the intercepted shim request, and the supervisor bridge into the production triage call, so cost-bearing requests stay bridge-counted instead of silently falling back to the real CLI. Profiles without costs keep identical prompts, commands, requests, and decision state. The supervisor command now places observation flags before `--rank-hypotheses`, matching the bridge command and generated shim, so exact rank-plus-observation triage requests are intercepted instead of silently falling back to the real interpreter.
+
+Ten new wiring tests plus updated doubles cover loader accept/reject, equal-arm disclosure, supervisor argv construction, spec validation, bridge request matching and forwarding, shim interception without fallback, and rank-plus-observation command order. The full 858-test suite passes locally (43.1 s) with `git diff --check` clean. Hosted CI remains required before merge. No trial was run and no speed or quality benefit is claimed.

scripts/pilot_contract_triage_pair.py

@@ -144,6 +144,7 @@ class CaseProfile:
     outcome_mode: str
     oracle_script: Path
     oracle_sha256: str
+    triage_diagnostic_costs: dict[Any, Any] | None = None
 
 
 def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
@@ -274,12 +275,12 @@ def load_case_profile(path: Path) -> CaseProfile:
                    for value in failure_markers)):
         raise ValueError("invalid case-profile failure signature")
     triage = raw["triage"]
-    if not isinstance(triage, dict) or set(triage) != {
-        "kinds", "hypotheses", "accepted_ids", "accepted_statuses", "observations",
-        "rank_hypotheses",
-    }:
+    if not isinstance(triage, dict) or set(triage) not in (
+        {"kinds", "hypotheses", "accepted_ids", "accepted_statuses", "observations", "rank_hypotheses"},
+        {"kinds", "hypotheses", "accepted_ids", "accepted_statuses", "observations", "rank_hypotheses", "diagnostic_costs"},
+    ):
         raise ValueError("invalid case-profile triage")
-    from jevcompass.triage import FailureKind, HypothesisId
+    from jevcompass.triage import DiagnosticCost, FailureKind, HypothesisId
     kinds, hypotheses, accepted = (triage[key] for key in ("kinds", "hypotheses", "accepted_ids"))
     statuses = triage["accepted_statuses"]
     observations = triage["observations"]
@@ -319,6 +320,16 @@ def load_case_profile(path: Path) -> CaseProfile:
     rank_hypotheses = triage["rank_hypotheses"]
     if type(rank_hypotheses) is not bool:
         raise ValueError("invalid case-profile ranking opt-in")
+    raw_costs = triage.get("diagnostic_costs", {})
+    if (not isinstance(raw_costs, dict) or len(raw_costs) > len(hypotheses)
+            or any(not isinstance(key, str) or key not in hypotheses for key in raw_costs)):
+        raise ValueError("invalid case-profile diagnostic costs")
+    try:
+        diagnostic_costs = {
+            HypothesisId(key): DiagnosticCost(value) for key, value in raw_costs.items()
+        }
+    except (TypeError, ValueError) as exc:
+        raise ValueError("invalid case-profile diagnostic costs") from exc
     allow_remote = False
     outcome = raw["outcome_mode"]
     if outcome not in {"contract_triage", "repair"}:
@@ -345,14 +356,25 @@ def load_case_profile(path: Path) -> CaseProfile:
         case_id, fixture, prompt, source_file, test_file, focused_command,
         safe_evidence, safe_markers, tuple(failure_markers), tuple(kinds),
         tuple(hypotheses), tuple(accepted), tuple(statuses), safe_observations,
-        rank_hypotheses, outcome, oracle_script, expected_hash,
+        rank_hypotheses, outcome, oracle_script, expected_hash, diagnostic_costs,
     )
 
 def _case_base_prompt(profile: CaseProfile | None) -> str:
     """Build the shared profiled prompt with exact mandatory validation commands."""
     if profile is None:
         return BASE_PROMPT
     prompt = profile.task_prompt
+    costs = profile.triage_diagnostic_costs or {}
+    if costs:
+        declarations = ", ".join(
+            f"{hypothesis.value}={cost.value}"
+            for hypothesis, cost in costs.items()
+        )
+        prompt += (
+            "\n\nCaller-declared relative diagnostic check effort (shared with both arms): "
+            f"{declarations}. These declarations describe next-check effort only; "
+            "they are not evidence of causal likelihood."
+        )
     if profile.outcome_mode != "repair":
         return prompt
 
@@ -385,6 +407,10 @@ def _profile_bridge_spec(
             profile.triage_observations if include_observations else {}
         ),
         rank_hypotheses=profile.rank_hypotheses,
+        diagnostic_costs={
+            hypothesis.value: cost.value
+            for hypothesis, cost in (profile.triage_diagnostic_costs or {}).items()
+        },
     )
 
 
@@ -421,6 +447,11 @@ def _case_treatment_prompt(
             "Triage is optional: skip it when local evidence resolves the choice. "
             f"If you invoke triage, use this exact command: {triage_command}. "
         )
+    if profile.triage_diagnostic_costs:
+        guidance += (
+            "Caller-declared relative check effort may guide diagnostic next-step ordering only; "
+            "it is not evidence of causal likelihood and must not be used to skip required checks. "
+        )
     if profile.outcome_mode == "repair":
         guidance += (
             f"Complete the requested behavior change in {profile.source_file}. "
@@ -487,8 +518,9 @@ def _triage_argv(
             candidates.append("confirm_behavior_contract")
         for hypothesis in candidates:
             suffix.extend(("--hypothesis", hypothesis))
-        if profile.rank_hypotheses:
-            suffix.append("--rank-hypotheses")
+        for hypothesis, cost in (profile.triage_diagnostic_costs or {}).items():
+            suffix.extend(("--diagnostic-cost", f"{hypothesis.value}={cost.value}"))
+        # Observation flags precede --rank-hypotheses to match command_for() and the shim.
         if include_observations:
             observation_flags = {
                 "import": "--import-observation",
@@ -498,6 +530,8 @@ def _triage_argv(
             for category, values in profile.triage_observations.items():
                 for value in values:
                     suffix.extend((observation_flags[category], value))
+        if profile.rank_hypotheses:
+            suffix.append("--rank-hypotheses")
         suffix.append("--json")
     return [*TRIAGE_PREFIX, str(exit_code), *suffix]
 

scripts/pilot_profile_triage_bridge.py

@@ -9,7 +9,7 @@
 """
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 import json
 import math
 import os
@@ -63,9 +63,10 @@ class ProfileTriageSpec:
     hypotheses: tuple[str, ...]
     allowed_observations: Mapping[str, tuple[str, ...]]
     rank_hypotheses: bool = False
+    diagnostic_costs: Mapping[str, str] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
-        from jevcompass.triage import FailureKind, HypothesisId
+        from jevcompass.triage import DiagnosticCost, FailureKind, HypothesisId
 
         if (not self.failure_kinds or not self.hypotheses
                 or any(not isinstance(value, str) for value in self.failure_kinds)
@@ -80,6 +81,15 @@ def __post_init__(self) -> None:
             raise ValueError("invalid profile triage enum contract")
         if type(self.rank_hypotheses) is not bool:
             raise ValueError("rank_hypotheses must be boolean")
+        # Plain strings only: these values are embedded verbatim in the shim.
+        if not isinstance(self.diagnostic_costs, Mapping):
+            raise ValueError("invalid profile diagnostic costs")
+        allowed_costs = {item.value for item in DiagnosticCost}
+        for hypothesis, cost in self.diagnostic_costs.items():
+            if (type(hypothesis) is not str or type(cost) is not str
+                    or hypothesis not in self.hypotheses
+                    or cost not in allowed_costs):
+                raise ValueError("invalid profile diagnostic costs")
         if not isinstance(self.allowed_observations, Mapping):
             raise ValueError("invalid profile observation allowlist")
         for group, values in self.allowed_observations.items():
@@ -322,7 +332,7 @@ def _read_request(self, handler: BaseHTTPRequestHandler) -> Any:
     def _validated_request(self, request: Any, observed_exit: int) -> dict[str, Any]:
         if not isinstance(request, dict) or set(request) != {
             "observed_exit_status", "failure_kinds", "hypotheses", "observations",
-            "rank_hypotheses",
+            "rank_hypotheses", "diagnostic_costs",
         }:
             raise ValueError("invalid enum-only request")
         if (type(request["observed_exit_status"]) is not int
@@ -343,12 +353,21 @@ def _validated_request(self, request: Any, observed_exit: int) -> dict[str, Any]
                     or len(set(values)) != len(values)):
                 raise ValueError("invalid observation enum")
             observations[group] = tuple(values)
-        return {"observations": observations}
+        raw_costs = request["diagnostic_costs"]
+        if (not isinstance(raw_costs, dict)
+                or any(type(key) is not str or type(value) is not str
+                       for key, value in raw_costs.items())
+                or raw_costs != dict(self.spec.diagnostic_costs)):
+            raise ValueError("invalid diagnostic cost request")
+        return {
+            "observations": observations,
+            "diagnostic_costs": dict(raw_costs),
+        }
 
     def _decide(self, args: dict[str, Any], observed_exit: int):
         from jevcompass.decisions import DecisionsClient
         from jevcompass.triage import (
-            CATALOG, AssertionObservation, FailureKind, HypothesisId,
+            CATALOG, AssertionObservation, DiagnosticCost, FailureKind, HypothesisId,
             ImportObservation, TriageDecisionReason, TimeoutObservation,
             TriageResult, triage_failure,
         )
@@ -367,13 +386,20 @@ def factory():
             enum_groups[group][1]: tuple(enum_groups[group][0](value) for value in values)
             for group, values in args["observations"].items()
         }
+        diagnostic_costs = {
+            HypothesisId(key): DiagnosticCost(value)
+            for key, value in args["diagnostic_costs"].items()
+        }
+        triage_options = dict(triage_observations)
+        if diagnostic_costs:
+            triage_options["diagnostic_costs"] = diagnostic_costs
         result = triage_failure(
             tuple(FailureKind(value) for value in self.spec.failure_kinds),
             tuple(HypothesisId(value) for value in self.spec.hypotheses),
             observed_exit,
             client=client,
             rank_hypotheses=self.spec.rank_hypotheses,
-            **triage_observations,
+            **triage_options,
         )
         if not isinstance(result, TriageResult) or result.observed_exit_status != observed_exit:
             raise ValueError("invalid production triage result")
@@ -510,6 +536,8 @@ def command_for(spec: ProfileTriageSpec, exit_status: int,
         args.extend(("--kind", kind))
     for hypothesis in spec.hypotheses:
         args.extend(("--hypothesis", hypothesis))
+    for hypothesis, cost in spec.diagnostic_costs.items():
+        args.extend(("--diagnostic-cost", f"{hypothesis}={cost}"))
     for group in ("import", "assertion", "timeout"):
         values = tuple(observed.get(group, ()))
         allowed = set(spec.allowed_observations.get(group, ()))
@@ -534,6 +562,7 @@ def write_python_shim(directory: Path, bridge: ProfileTriageBridge,
     bindir.mkdir(mode=0o700, parents=True, exist_ok=True)
     shim = bindir / "python"
     groups = {key: list(value) for key, value in bridge.spec.allowed_observations.items()}
+    costs = dict(bridge.spec.diagnostic_costs)
     source = f'''#!/usr/bin/env python3
 import json, os, sys
 from urllib.request import Request, urlopen
@@ -543,6 +572,7 @@ def write_python_shim(directory: Path, bridge: ProfileTriageBridge,
 HYPOTHESES = {list(bridge.spec.hypotheses)!r}
 OBSERVATIONS = {groups!r}
 RANK = {bridge.spec.rank_hypotheses!r}
+COSTS = {costs!r}
 FLAGS = {dict(_OBSERVATION_FLAGS)!r}
 ARGS = sys.argv[1:]
 def fallback():
@@ -564,6 +594,18 @@ def pairs(flag, values):
     if tail[:len(expected)] != expected:
         fallback()
     tail = tail[len(expected):]
+    supplied_costs = {{}}
+    while tail and tail[0] == "--diagnostic-cost":
+        if len(tail) < 2:
+            fallback()
+        name, separator, level = tail[1].partition("=")
+        if (not separator or name not in COSTS or level != COSTS[name]
+                or name in supplied_costs):
+            fallback()
+        supplied_costs[name] = level
+        tail = tail[2:]
+    if supplied_costs != COSTS:
+        fallback()
     observations = {{group: [] for group in OBSERVATIONS}}
     while tail and tail[0] != "--json" and tail[0] != "--rank-hypotheses":
         if len(tail) < 2:
@@ -585,7 +627,7 @@ def pairs(flag, values):
     body = json.dumps({{
         "observed_exit_status": code, "failure_kinds": KINDS,
         "hypotheses": HYPOTHESES, "observations": observations,
-        "rank_hypotheses": RANK,
+        "rank_hypotheses": RANK, "diagnostic_costs": supplied_costs,
     }}, separators=(",", ":")).encode()
     req = Request(URL, data=body, headers={{"Content-Type": "application/json"}}, method="POST")
     with urlopen(req, timeout=2.0) as response:

tests/test_pilot_mandatory_command_visibility.py

@@ -37,6 +37,7 @@ def setUp(self):
             triage_accepted_ids=(),
             rank_hypotheses=False,
             triage_observations={},
+            triage_diagnostic_costs={},
         )
 
     def test_exact_standalone_commands_are_shared_before_treatment_extras(self):
@@ -64,6 +65,7 @@ def test_legacy_and_nonrepair_prompt_behavior_is_unchanged(self):
         contract_profile = SimpleNamespace(
             task_prompt="Inspect and preserve the original failure.",
             outcome_mode="contract_triage",
+            triage_diagnostic_costs={},
         )
         self.assertEqual(runner._case_base_prompt(contract_profile), contract_profile.task_prompt)