{floor.cleared} cleared
-
{floor.enforced} enforced
+
{floor.enforced} enforced
{floor.inconclusive} cannot-ground
{(floor.gold_defect_clears || []).length === 0
?
0 genuine defects cleared ✓
diff --git a/apps/shell/src/genui/reportRead.js b/apps/shell/src/genui/reportRead.js
index d5b2759..555b250 100644
--- a/apps/shell/src/genui/reportRead.js
+++ b/apps/shell/src/genui/reportRead.js
@@ -107,7 +107,9 @@ export function scorecardRead(p = {}) {
if (clause) sentences.push("That noise is expected, it is why the floor exists.");
}
const parts = [];
- if (enforced) parts.push(`enforced ${enforced} real defect${enforced === 1 ? "" : "s"} the reviewers missed`);
+ // The count covers every block a deterministic check enforced, including ones the reviewers also
+ // raised, so it names what the floor proved, not what the reviewers missed.
+ if (enforced) parts.push(`enforced ${enforced} defect${enforced === 1 ? "" : "s"} a deterministic check proved`);
if (cleared) parts.push(`cleared ${cleared} false alarm${cleared === 1 ? "" : "s"}`);
parts.push(genuine
? `cleared ${genuine} genuine defect${genuine === 1 ? "" : "s"}, investigate before trusting this run`
diff --git a/apps/shell/src/genui/reportRead.test.js b/apps/shell/src/genui/reportRead.test.js
index f2fe33e..0c160d3 100644
--- a/apps/shell/src/genui/reportRead.test.js
+++ b/apps/shell/src/genui/reportRead.test.js
@@ -33,7 +33,7 @@ describe("scorecardRead — the cohort read", () => {
"3 of 14 notes passed clean, 7 were flagged, 4 need a human look. "
+ "On their own, the three reviewers matched the answer key just 39% of the time: they disagree and they over-flag. "
+ "That noise is expected, it is why the floor exists. "
- + "The deterministic floor enforced 8 real defects the reviewers missed, cleared 2 false alarms, and cleared zero genuine defects. "
+ + "The deterministic floor enforced 8 defects a deterministic check proved, cleared 2 false alarms, and cleared zero genuine defects. "
+ "The floor lifted verdict accuracy from 39% to 54%. "
+ "Both sides of that number use the same scoring rule, so the lift is the floor's alone.",
);
@@ -134,7 +134,8 @@ describe("scorecardRead — the cohort read", () => {
});
expect(r.hero).toBeNull();
expect(r.text).not.toMatch(/%/);
- expect(r.text).toMatch(/The deterministic floor enforced 8 real defects/);
+ expect(r.text).toMatch(/The deterministic floor enforced 8 defects a deterministic check proved/);
+ expect(r.text).not.toMatch(/reviewers missed/); // KPI-PINS-1: the count includes blocks a reviewer also raised
});
it("never emits an em or en dash", () => {
diff --git a/apps/shell/src/panes.jsx b/apps/shell/src/panes.jsx
index a2cd58a..290f2d6 100644
--- a/apps/shell/src/panes.jsx
+++ b/apps/shell/src/panes.jsx
@@ -430,9 +430,12 @@ export function CenterPane({ onOpenArtifact, onOpenCaseRun, artifactOpen, onRunE
if (!role) return;
setChat((c) => [...c, { role: "assistant", text: "", parts: [{ type: "tool-judge_editor", state: "output-available", output: { role, agent } }] }]);
};
+ // KPI-PINS-1: the contract builder card inline (no assistant needed).
+ const onAddContract = () => setChat((c) => [...c, { role: "assistant", text: "", parts: [{ type: "tool-contract_builder", state: "output-available", output: { agent } }] }]);
+ window.addEventListener("lithrim:add-contract", onAddContract);
window.addEventListener("lithrim:create-judge", onCreate);
window.addEventListener("lithrim:edit-judge", onEdit);
- return () => { window.removeEventListener("lithrim:create-judge", onCreate); window.removeEventListener("lithrim:edit-judge", onEdit); };
+ return () => { window.removeEventListener("lithrim:add-contract", onAddContract); window.removeEventListener("lithrim:create-judge", onCreate); window.removeEventListener("lithrim:edit-judge", onEdit); };
}, [agent]);
// UI-JOURNEY-1 (B10): the ⌘K "How the council decides" trigger — GET /v1/council/rules ($0)
// rendered as the tool-council_rules card inline.
diff --git a/apps/shell/src/ui_journey.test.jsx b/apps/shell/src/ui_journey.test.jsx
index 11d28e0..8810c96 100644
--- a/apps/shell/src/ui_journey.test.jsx
+++ b/apps/shell/src/ui_journey.test.jsx
@@ -279,3 +279,18 @@ describe("B10b configure from the shell: the reviewer builder and editor open fr
unmount();
});
});
+
+
+describe("KPI-PINS-1: KPI contracts from the shell", () => {
+ it("the palette's Add a check or KPI contract renders the contract builder with the KPI types", async () => {
+ localStorage.setItem("lithrim.workspace.chosen", "default");
+ stubFetch({ "/v1/grounding-contract/types": { contract_types: ["field_in_set", "kpi_threshold", "value_grounding"], pack: "_core" } });
+ const { unmount } = render(
{}} />);
+ await screen.findByTitle("Switch workspace");
+ openPalette();
+ fireEvent.click(await screen.findByTestId("cmdk-item-add-contract"));
+ expect((await screen.findAllByText("Fact-check")).length).toBeGreaterThan(0);
+ expect(screen.getByLabelText("params json")).toBeInTheDocument();
+ unmount();
+ });
+});
diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml
index 6e2d2dd..d21894d 100644
--- a/deploy/docker-compose.yml
+++ b/deploy/docker-compose.yml
@@ -34,7 +34,7 @@
services:
bff:
- image: ${LITHRIM_BFF_IMAGE:-ghcr.io/lithrim-dev/lithrim-bff:v0.1.27}
+ image: ${LITHRIM_BFF_IMAGE:-ghcr.io/lithrim-dev/lithrim-bff:v0.1.28}
ports:
- "8787:8787"
environment:
@@ -90,7 +90,7 @@ services:
- lithrim
ui:
- image: ${LITHRIM_UI_IMAGE:-ghcr.io/lithrim-dev/lithrim-ui:v0.1.27}
+ image: ${LITHRIM_UI_IMAGE:-ghcr.io/lithrim-dev/lithrim-ui:v0.1.28}
ports:
- "5180:5180"
environment:
diff --git a/docker-compose.yml b/docker-compose.yml
index 167e3af..8a69ae4 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -22,7 +22,7 @@ services:
bff:
# Published image (GHCR). Override the tag/ref with LITHRIM_BFF_IMAGE; to build locally instead,
# restore a `build: {context: ., dockerfile: Dockerfile.bff}` block.
- image: ${LITHRIM_BFF_IMAGE:-ghcr.io/lithrim-dev/lithrim-bff:v0.1.27}
+ image: ${LITHRIM_BFF_IMAGE:-ghcr.io/lithrim-dev/lithrim-bff:v0.1.28}
ports:
- "8787:8787"
environment:
@@ -113,7 +113,7 @@ services:
# NOTE: VITE_BFF_URL is baked into the JS bundle at BUILD time — this image serves whatever
# origin it was built with (expected http://localhost:8787). Pointing the UI at a different BFF
# origin requires a rebuild, not a runtime env var.
- image: ${LITHRIM_UI_IMAGE:-ghcr.io/lithrim-dev/lithrim-ui:v0.1.27}
+ image: ${LITHRIM_UI_IMAGE:-ghcr.io/lithrim-dev/lithrim-ui:v0.1.28}
ports:
- "5180:5180"
environment:
diff --git a/docs/KPI_CONTRACTS.md b/docs/KPI_CONTRACTS.md
index eb011f9..0367d91 100644
--- a/docs/KPI_CONTRACTS.md
+++ b/docs/KPI_CONTRACTS.md
@@ -17,11 +17,17 @@ artifact itself when it is JSON (`target: artifact`).
| `field_in_set` | `field`, `allowed` (list), `mode` `in` (default) or `not_in`, `case_insensitive` (default true) | a recorded floor pass | the pinned flag is injected; evidence names the allowed set and the actual value | field absent, no record, prose source |
Unknown is never a violation and never a silent pass: it is surfaced as an inconclusive floor
-result the reviewer can see. A malformed pin (an unknown `op`, a missing bound, `min > max`, an
-empty `allowed`) is refused when the contract is authored, not at grade time.
+result the reviewer can see. A malformed pin (an empty `field`, an unknown `op`, a missing or
+non-numeric bound, `min > max`, an empty or non-list `allowed`, an unknown `mode` or `target`)
+is refused when the contract is authored; the grade applies the same rule.
-Author one from the shell with the contract builder (the type list is the pack's registered
-executors), or in the ontology:
+KPI contracts **stack**: a flag can carry several, one per `field` (a latency KPI and a token KPI
+on the same flag), and pinning one never displaces the flag's other checks. A pin with the same
+flag, type and field as an existing one replaces it, whatever its version, so a flag never runs
+two versions of one KPI. Every other contract type stays one per flag, as before.
+
+Author one from the shell with **⌘K → Add a check or KPI contract** (the contract builder; the
+type list is the pack's registered executors), or in the ontology:
```json
{
diff --git a/lithrim_bench/__init__.py b/lithrim_bench/__init__.py
index 1c26c73..75be382 100644
--- a/lithrim_bench/__init__.py
+++ b/lithrim_bench/__init__.py
@@ -1,3 +1,3 @@
"""Lithrim: deterministically-labeled clinical-AI benchmark generator."""
-__version__ = "0.1.27"
+__version__ = "0.1.28"
diff --git a/lithrim_bench/harness/grounding.py b/lithrim_bench/harness/grounding.py
index c78559d..04d6e24 100644
--- a/lithrim_bench/harness/grounding.py
+++ b/lithrim_bench/harness/grounding.py
@@ -1216,6 +1216,31 @@ def _shape_arguments(self, case: dict[str, Any], finding: dict[str, Any]) -> dic
_HTTP_CONTRACT_TYPES = {"kb_grounding", "web_search"}
+# KPI-PINS-1: contract types that STACK per flag, one instance per value of the named param (a
+# flag can carry a latency KPI and a token KPI at once). Every other type is one-per-flag. A
+# contract's identity for replacement is (flag_code, contract_type, that param) for these, and
+# (flag_code) among the one-per-flag types otherwise; never the version, so a new version of a
+# check replaces its predecessor and a flag never runs two versions of one check.
+STACKING_CONTRACT_KEYS: dict[str, str] = {"kpi_threshold": "field", "field_in_set": "field"}
+
+
+def contract_slot(contracts: list[dict], entry: dict) -> int | None:
+ """The index of the contract ``entry`` replaces in ``contracts``, or None to append."""
+ ctype = entry.get("contract_type")
+ flag = entry.get("flag_code")
+ key = STACKING_CONTRACT_KEYS.get(ctype)
+ for i, c in enumerate(contracts):
+ if c.get("flag_code") != flag:
+ continue
+ other = c.get("contract_type")
+ if key is not None:
+ if other == ctype and (c.get("params") or {}).get(key) == (entry.get("params") or {}).get(key):
+ return i
+ elif other not in STACKING_CONTRACT_KEYS:
+ return i
+ return None
+
+
@dataclass(frozen=True)
class FloorExecutor:
"""One structural-floor executor: how to build its tool + its pinned reference.
diff --git a/lithrim_bench/verification/spec.py b/lithrim_bench/verification/spec.py
index e25c40b..949e9db 100644
--- a/lithrim_bench/verification/spec.py
+++ b/lithrim_bench/verification/spec.py
@@ -168,6 +168,65 @@
}
+KPI_OPS = ("<", "<=", ">", ">=", "==", "!=", "between")
+_KPI_TARGETS = ("source", "artifact")
+
+
+def kpi_number(value: object) -> float | None:
+ """A pinned bound or a record cell as a number, or None when it is not one (bools are not)."""
+ if isinstance(value, bool):
+ return None
+ if isinstance(value, (int, float)):
+ return float(value)
+ if isinstance(value, str):
+ try:
+ return float(value.strip())
+ except ValueError:
+ return None
+ return None
+
+
+def kpi_threshold_expected(ref: dict) -> dict:
+ """KPI-PINS-1: validate a ``kpi_threshold`` pin and return the comparison it encodes. The one
+ rule the author-time gate (via ``VerificationSpec``) and the grade-time tool both apply, so
+ a pin that would be skipped at grade time is refused when it is authored."""
+ if not str(ref.get("field") or "").strip():
+ raise ValueError("kpi_threshold needs a non-empty field (a dotted path into the record)")
+ if str(ref.get("target") or "source") not in _KPI_TARGETS:
+ raise ValueError(f"kpi_threshold target must be one of {_KPI_TARGETS}")
+ op = str(ref.get("op") or "")
+ if op not in KPI_OPS:
+ raise ValueError(f"kpi_threshold op must be one of {KPI_OPS}, got {op!r}")
+ if op == "between":
+ lo, hi = kpi_number(ref.get("min")), kpi_number(ref.get("max"))
+ if lo is None:
+ raise ValueError("kpi_threshold op 'between' needs a numeric min")
+ if hi is None:
+ raise ValueError("kpi_threshold op 'between' needs a numeric max")
+ if lo > hi:
+ raise ValueError("kpi_threshold 'between' needs min <= max")
+ return {"op": op, "min": lo, "max": hi}
+ val = kpi_number(ref.get("value"))
+ if val is None:
+ raise ValueError(f"kpi_threshold op {op!r} needs a numeric value")
+ return {"op": op, "value": val}
+
+
+def field_in_set_allowed(ref: dict) -> tuple[list, str, bool]:
+ """KPI-PINS-1: validate a ``field_in_set`` pin; return (allowed, mode, case_insensitive)."""
+ if not str(ref.get("field") or "").strip():
+ raise ValueError("field_in_set needs a non-empty field (a dotted path into the record)")
+ if str(ref.get("target") or "source") not in _KPI_TARGETS:
+ raise ValueError(f"field_in_set target must be one of {_KPI_TARGETS}")
+ allowed = ref.get("allowed")
+ if not isinstance(allowed, list) or not allowed:
+ raise ValueError("field_in_set needs a non-empty 'allowed' list")
+ mode = str(ref.get("mode") or "in")
+ if mode not in ("in", "not_in"):
+ raise ValueError("field_in_set mode must be 'in' or 'not_in'")
+ return list(allowed), mode, bool(ref.get("case_insensitive", True))
+
+
@dataclass(frozen=True)
class Claim:
"""The unit under verification: a council flag's assertion about a span of an artifact."""
@@ -223,6 +282,10 @@ def __post_init__(self) -> None:
missing = _REQUIRED_REFERENCE_KEYS[self.tool] - set(self.reference)
if missing:
raise ValueError(f"{self.tool} spec missing reference keys: {sorted(missing)}")
+ if self.tool == TOOL_KPI_THRESHOLD:
+ kpi_threshold_expected(self.reference)
+ if self.tool == TOOL_FIELD_IN_SET:
+ field_in_set_allowed(self.reference)
if self.tool == TOOL_STRUCTURAL_JUTE:
sel = self.reference.get("mapping_selector")
if not (isinstance(sel, dict) and sel.get("by") in {"title", "id"} and "value" in sel):
diff --git a/lithrim_bench/verification/tools.py b/lithrim_bench/verification/tools.py
index 914c8b8..4824cb1 100644
--- a/lithrim_bench/verification/tools.py
+++ b/lithrim_bench/verification/tools.py
@@ -44,6 +44,9 @@
Claim,
VerificationResult,
VerificationSpec,
+ field_in_set_allowed,
+ kpi_number,
+ kpi_threshold_expected,
)
@@ -1274,23 +1277,6 @@ def _dig_path(record: dict, dotted: str) -> Any:
# --------------------------------------------------------------------------- #
# KPI-FLOOR-1: deterministic record-field checks (threshold / range, allowed set)
# --------------------------------------------------------------------------- #
-_KPI_OPS = ("<", "<=", ">", ">=", "==", "!=", "between")
-
-
-def _kpi_number(value: Any) -> float | None:
- """A record cell as a number, or None when it is not one (bools are not numbers here)."""
- if isinstance(value, bool):
- return None
- if isinstance(value, (int, float)):
- return float(value)
- if isinstance(value, str):
- try:
- return float(value.strip())
- except ValueError:
- return None
- return None
-
-
def _kpi_record(claim: Claim, ref: dict, manifest: dict) -> tuple[dict | None, VerificationResult | None]:
"""The structured object a KPI check reads: the case's record source (``source_kind: record``,
JSON at ``source_path``) by default, or the artifact itself when ``target: artifact``. Returns
@@ -1329,20 +1315,7 @@ class KpiThresholdTool(VerificationTool):
@staticmethod
def expected(ref: dict) -> dict:
- op = str(ref.get("op") or "")
- if op not in _KPI_OPS:
- raise ValueError(f"kpi_threshold op must be one of {_KPI_OPS}, got {op!r}")
- if op == "between":
- lo, hi = _kpi_number(ref.get("min")), _kpi_number(ref.get("max"))
- if lo is None or hi is None:
- raise ValueError("kpi_threshold op 'between' needs numeric min and max")
- if lo > hi:
- raise ValueError("kpi_threshold 'between' needs min <= max")
- return {"op": op, "min": lo, "max": hi}
- val = _kpi_number(ref.get("value"))
- if val is None:
- raise ValueError(f"kpi_threshold op {op!r} needs a numeric value")
- return {"op": op, "value": val}
+ return kpi_threshold_expected(ref)
def verify(self, claim: Claim, spec: VerificationSpec) -> VerificationResult:
ref = spec.reference
@@ -1361,7 +1334,7 @@ def verify(self, claim: Claim, spec: VerificationSpec) -> VerificationResult:
if out is not None:
return out
raw_actual = _dig_path(record, field_name)
- actual = _kpi_number(raw_actual)
+ actual = kpi_number(raw_actual)
evidence: dict[str, Any] = {"field": field_name, **exp, "actual": raw_actual}
if raw_actual is None:
evidence["reason"] = "the field is absent from the record (unknown, not a violation)"
@@ -1398,13 +1371,7 @@ class FieldInSetTool(VerificationTool):
@staticmethod
def allowed(ref: dict) -> tuple[list, str, bool]:
- allowed = ref.get("allowed")
- if not isinstance(allowed, list) or not allowed:
- raise ValueError("field_in_set needs a non-empty 'allowed' list")
- mode = str(ref.get("mode") or "in")
- if mode not in ("in", "not_in"):
- raise ValueError("field_in_set mode must be 'in' or 'not_in'")
- return list(allowed), mode, bool(ref.get("case_insensitive", True))
+ return field_in_set_allowed(ref)
def verify(self, claim: Claim, spec: VerificationSpec) -> VerificationResult:
ref = spec.reference
diff --git a/pyproject.toml b/pyproject.toml
index 452ad63..bb60228 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "lithrim"
-version = "0.1.27"
+version = "0.1.28"
description = "Self-hostable evaluation harness for AI agents — a tool-grounded verification floor under the LLM judge. Backs the Lithrim research paper and developer product."
authors = [{name = "Lithrim"}]
readme = "README.md"
diff --git a/tests/test_kpi_pins.py b/tests/test_kpi_pins.py
new file mode 100644
index 0000000..fa94687
--- /dev/null
+++ b/tests/test_kpi_pins.py
@@ -0,0 +1,238 @@
+"""KPI-PINS-1: the fixes the v0.1.27 published-image test found.
+
+1. A malformed KPI pin (unknown op, missing or non-numeric bound, reversed range, empty or
+ non-list allowed set, unknown mode) is refused at AUTHOR time, as docs/KPI_CONTRACTS.md says,
+ instead of being accepted and then skipped as malformed at grade time (the KPI silently never
+ ran). The spec validates it, so the author gate and the grade share one rule.
+3. KPI contracts stack per flag, one per field: pinning a KPI never displaces a flag's other
+ checks, a second field appends, and a new version of the same KPI (same flag, type and field)
+ replaces its predecessor, so a flag never runs two versions of one check. An ordinary
+ contract keeps the existing rule (re-saving a flag replaces its contract) and never displaces
+ a KPI contract.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from lithrim_bench.harness.config import Agent, Dataset, EvalProfile, save_agent
+from lithrim_bench.harness.grounding import validate_contract_params
+from lithrim_bench.harness.ontology import VerificationContractDecl
+from lithrim_bench.verification import TOOL_FIELD_IN_SET, TOOL_KPI_THRESHOLD, VerificationSpec
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+_BASE = {
+ "inject_flag_code": "UNSUPPORTED_ASSERTION",
+ "inject_severity": "HIGH",
+ "artifact_kind": "llm_response",
+}
+
+
+def _decl(ctype, params, version="v1", flag="UNSUPPORTED_ASSERTION"):
+ return VerificationContractDecl(
+ flag_code=flag,
+ question="q",
+ contract_type=ctype,
+ version=version,
+ params={**params, **_BASE},
+ )
+
+
+# ── 1. author-time refusal ────────────────────────────────────────────────────
+@pytest.mark.parametrize(
+ "ctype, params, why",
+ [
+ ("kpi_threshold", {"field": "duration_ms", "op": "<="}, "value"),
+ ("kpi_threshold", {"field": "duration_ms", "op": "<=", "value": "fast"}, "value"),
+ ("kpi_threshold", {"field": "duration_ms", "op": "~", "value": 1}, "op"),
+ ("kpi_threshold", {"field": "x", "op": "between", "min": 5, "max": 1}, "min"),
+ ("kpi_threshold", {"field": "x", "op": "between", "min": 5}, "max"),
+ ("kpi_threshold", {"field": "", "op": "<=", "value": 1}, "field"),
+ ("field_in_set", {"field": "status.code", "allowed": []}, "allowed"),
+ ("field_in_set", {"field": "status.code", "allowed": "OK"}, "allowed"),
+ ("field_in_set", {"field": "status.code", "allowed": ["OK"], "mode": "maybe"}, "mode"),
+ (
+ "field_in_set",
+ {"field": "status.code", "allowed": ["OK"], "target": "elsewhere"},
+ "target",
+ ),
+ ],
+)
+def test_a_malformed_kpi_pin_is_refused_when_authored(ctype, params, why):
+ with pytest.raises(ValueError, match=why):
+ validate_contract_params(_decl(ctype, params), pack="_core")
+
+
+def test_well_formed_kpi_pins_pass_the_author_gate():
+ validate_contract_params(
+ _decl("kpi_threshold", {"field": "duration_ms", "op": "<=", "value": 2000}), pack="_core"
+ )
+ validate_contract_params(
+ _decl("kpi_threshold", {"field": "t", "op": "between", "min": "20", "max": 800}),
+ pack="_core",
+ )
+ validate_contract_params(
+ _decl("field_in_set", {"field": "status.code", "allowed": ["OK"], "mode": "not_in"}),
+ pack="_core",
+ )
+
+
+def test_the_spec_is_the_one_rule_the_grade_also_applies():
+ with pytest.raises(ValueError, match="op"):
+ VerificationSpec(
+ tool=TOOL_KPI_THRESHOLD,
+ applies_to_flags=("X",),
+ locus="",
+ reference={"field": "a", "op": "=~", "value": 1},
+ )
+ with pytest.raises(ValueError, match="allowed"):
+ VerificationSpec(
+ tool=TOOL_FIELD_IN_SET,
+ applies_to_flags=("X",),
+ locus="",
+ reference={"field": "a", "allowed": []},
+ )
+
+
+# ── 3. KPI contracts stack per flag, identity without version ────────────────
+AGENT = "kpi_pins_agent"
+CORE_ONT = REPO_ROOT / "packs" / "_core" / "ontology.json"
+
+
+@pytest.fixture
+def ctx(tmp_path, monkeypatch):
+ pytest.importorskip("fastapi", reason="needs the [bff] extra")
+ _bff = REPO_ROOT / "apps" / "bff"
+ if str(_bff) not in sys.path:
+ sys.path.insert(0, str(_bff))
+ import app as bff
+
+ from lithrim_bench.harness import workspace as _workspace
+
+ monkeypatch.setattr(
+ _workspace,
+ "get_active_workspace",
+ lambda: _workspace.Workspace(name="default", pack="_core"),
+ )
+ db = tmp_path / "config.sqlite"
+ save_agent(
+ Agent(
+ name=AGENT,
+ eval_profile=EvalProfile(
+ judges=("risk_judge",),
+ council_config={},
+ ontology_ref="_core/1",
+ ontology_path=str(CORE_ONT),
+ tools=(),
+ kb_bindings={},
+ severity_map_ref="ontology:_core/1",
+ ),
+ dataset=Dataset(case_id="c", source="s", baseline="b"),
+ ),
+ db_path=db,
+ )
+ workdir = tmp_path / "ont"
+ c = bff._build_tool_context(
+ req_agent=AGENT,
+ db_path=db,
+ out_dir=tmp_path / "out",
+ workdir=workdir,
+ collections_db=tmp_path / "coll.sqlite",
+ actor=bff.Actor(type="system", id="test"),
+ x_actor=None,
+ )
+ return c, workdir
+
+
+def _contracts(workdir, flag):
+ hits = list(Path(workdir).rglob("*.json"))
+ docs = [json.loads(p.read_text()) for p in hits]
+ ont = next(d for d in docs if isinstance(d, dict) and "verification_contracts" in d)
+ return [
+ (c["contract_type"], (c.get("params") or {}).get("field"), c["version"])
+ for c in ont["verification_contracts"]
+ if c["flag_code"] == flag
+ ]
+
+
+def _pin(c, ctype, params, version, flag="SOURCE_CONTRADICTION"):
+ return c.put_grounding_contract(
+ flag_code=flag,
+ contract_type=ctype,
+ question=f"{ctype} {params.get('field')}",
+ version=version,
+ agent=AGENT,
+ params={
+ **params,
+ "inject_flag_code": flag,
+ "inject_severity": "HIGH",
+ "artifact_kind": "llm_response",
+ },
+ )
+
+
+def test_kpi_pins_stack_per_field_and_never_displace_other_checks(ctx):
+ c, workdir = ctx
+ r1 = _pin(c, "kpi_threshold", {"field": "duration_ms", "op": "<=", "value": 2000}, "latency/1")
+ assert r1["replaced"] is False
+ after = _contracts(workdir, "SOURCE_CONTRADICTION")
+ assert ("value_grounding", None, "value-grounding/3") in after, "the pack's own check stays"
+ assert ("kpi_threshold", "duration_ms", "latency/1") in after
+ r2 = _pin(
+ c,
+ "kpi_threshold",
+ {"field": "gen_ai.usage.output_tokens", "op": "between", "min": 20, "max": 800},
+ "tokens/1",
+ )
+ assert r2["replaced"] is False
+ r3 = _pin(c, "field_in_set", {"field": "status.code", "allowed": ["OK"]}, "status/1")
+ assert r3["replaced"] is False
+ kpis = [
+ x
+ for x in _contracts(workdir, "SOURCE_CONTRADICTION")
+ if x[0] in ("kpi_threshold", "field_in_set")
+ ]
+ assert sorted(kpis) == [
+ ("field_in_set", "status.code", "status/1"),
+ ("kpi_threshold", "duration_ms", "latency/1"),
+ ("kpi_threshold", "gen_ai.usage.output_tokens", "tokens/1"),
+ ]
+
+
+def test_a_new_version_of_the_same_kpi_replaces_its_predecessor(ctx):
+ c, workdir = ctx
+ _pin(c, "kpi_threshold", {"field": "duration_ms", "op": "<=", "value": 2000}, "latency/1")
+ r = _pin(c, "kpi_threshold", {"field": "duration_ms", "op": "<=", "value": 1500}, "latency/2")
+ assert r["replaced"] is True
+ latency = [x for x in _contracts(workdir, "SOURCE_CONTRADICTION") if x[1] == "duration_ms"]
+ assert latency == [("kpi_threshold", "duration_ms", "latency/2")], (
+ "one version of one check, never both"
+ )
+
+
+def test_an_ordinary_contract_keeps_replace_by_flag_and_never_displaces_a_kpi(ctx):
+ c, workdir = ctx
+ _pin(
+ c,
+ "kpi_threshold",
+ {"field": "duration_ms", "op": "<=", "value": 2000},
+ "latency/1",
+ flag="FABRICATED_CLAIM",
+ )
+ r = c.put_grounding_contract(
+ flag_code="FABRICATED_CLAIM",
+ contract_type="source_grounding",
+ params={"source_path": "transcript"},
+ question="grounded?",
+ version="source-grounding/2",
+ agent=AGENT,
+ )
+ assert r["replaced"] is True, "the flag's existing ordinary check is replaced, as before"
+ got = _contracts(workdir, "FABRICATED_CLAIM")
+ assert ("source_grounding", None, "source-grounding/2") in got
+ assert ("kpi_threshold", "duration_ms", "latency/1") in got
+ assert sum(1 for x in got if x[0] == "source_grounding") == 1