Reject TOTEM scores that contradict fingerprint metrics

src/totem_bridge.py

@@ -138,6 +138,7 @@ def dashboard_output_contract() -> dict[str, Any]:
             field: {
                 "score": "integer 0..100 matching scores.<field>",
                 "fingerprint_metrics_used": ["VM metric ids used, e.g. VM-001, VM-010, VM-025"],
+                "fingerprint_metric_values": {"VM-010": "numeric/string value used from manuscript_fingerprint"},
                 "workbook_rule": "target/threshold/rubric rule used from workbook matrix",
                 "manuscript_evidence": "short paraphrased manuscript evidence; no long excerpts",
                 "reason": "why this score follows from the workbook rule and fingerprint metrics",
@@ -199,8 +200,11 @@ Use the workbook matrix as scoring authority, manuscript fingerprint VM metrics
             "Return exactly one JSON object matching required_output_contract. "
             "Scores must be 0..100 dashboard values for the six dimensions. "
             "Every score_evidence entry must cite at least one VM metric id from manuscript_fingerprint "
-            "and one workbook rule/target from rubric_matrix. If the fingerprint is missing or insufficient, "
-            "return gate REVISE and explain the missing evidence in dashboard_message. "
+            "with its observed value in fingerprint_metric_values, plus one workbook rule/target from rubric_matrix. "
+            "If the fingerprint is missing or insufficient, return gate REVISE and explain the missing evidence in dashboard_message. "
+            "Hard caps for picture-book scoring: if VM-010 sentence length mean is above 30 or VM-025 reading age "
+            "is above 10, clarity and read_aloud_flow must not exceed 75. If VM-025 reading age is above 10, "
+            "commercial_viability must not exceed 75. If VM-002 syllable variance is above 5, rhythm must not exceed 75. "
             "Do not write a report and do not include manuscript excerpts."
         ),
     }
@@ -210,7 +214,7 @@ Use the workbook matrix as scoring authority, manuscript fingerprint VM metrics
         if os.getenv("SPACE_ID") and not _env_truthy("TOTEM_ALLOW_FAKE_API_IN_SPACE"):
             raise BridgeError("TOTEM_FAKE_API is disabled on Hugging Face Spaces.")
         raw_payload = _fake_dashboard_payload(manuscript)
-        validated = validate_dashboard_payload(raw_payload)
+        validated = validate_dashboard_payload(raw_payload, manuscript.fingerprint)
         raw_payload_json = json.dumps(raw_payload, ensure_ascii=False)
         audit_path = _write_token_audit(
             transaction_id=transaction_id,
@@ -319,7 +323,7 @@ Use the workbook matrix as scoring authority, manuscript fingerprint VM metrics
         raise BridgeValidationError(f"Model returned invalid JSON: {exc}") from exc
 
     try:
-        validated = validate_dashboard_payload(parsed)
+        validated = validate_dashboard_payload(parsed, manuscript.fingerprint)
     except BridgeValidationError as exc:
         audit_path = _write_token_audit(
             transaction_id=transaction_id,
@@ -548,7 +552,7 @@ def _first_env_value(names: tuple[str, ...]) -> tuple[str | None, str | None]:
     return None, None
 
 
-def validate_dashboard_payload(payload: dict[str, Any]) -> dict[str, Any]:
+def validate_dashboard_payload(payload: dict[str, Any], fingerprint: dict[str, Any] | None = None) -> dict[str, Any]:
     if not isinstance(payload, dict):
         raise BridgeValidationError("Dashboard payload must be a JSON object.")
 
@@ -595,7 +599,8 @@ def validate_dashboard_payload(payload: dict[str, Any]) -> dict[str, Any]:
     score_evidence = payload.get("score_evidence")
     if not isinstance(score_evidence, dict):
         raise BridgeValidationError("score_evidence must be a JSON object keyed by scoring dimension.")
-    out["score_evidence"] = _normalize_score_evidence(score_evidence, out["scores"])
+    out["score_evidence"] = _normalize_score_evidence(score_evidence, out["scores"], fingerprint)
+    _validate_fingerprint_score_alignment(out, fingerprint)
 
     rpq = payload.get("revision_priority_queue")
     if not isinstance(rpq, list):
@@ -609,7 +614,11 @@ def validate_dashboard_payload(payload: dict[str, Any]) -> dict[str, Any]:
     return out
 
 
-def _normalize_score_evidence(evidence: dict[str, Any], scores: dict[str, int]) -> dict[str, dict[str, Any]]:
+def _normalize_score_evidence(
+    evidence: dict[str, Any],
+    scores: dict[str, int],
+    fingerprint: dict[str, Any] | None = None,
+) -> dict[str, dict[str, Any]]:
     normalized: dict[str, dict[str, Any]] = {}
     missing: list[str] = []
     weak: list[str] = []
@@ -626,6 +635,10 @@ def _normalize_score_evidence(evidence: dict[str, Any], scores: dict[str, int])
         if not metric_ids or not any(m.upper().startswith("VM-") for m in metric_ids):
             weak.append(field)
 
+        required_metric = _required_bottleneck_metric(field, fingerprint)
+        if required_metric and required_metric not in {m.upper().split("_", 1)[0] for m in metric_ids}:
+            weak.append(field)
+
         workbook_rule = str(item.get("workbook_rule") or "").strip()[:260]
         manuscript_evidence = str(item.get("manuscript_evidence") or "").strip()[:260]
         reason = str(item.get("reason") or "").strip()[:320]
@@ -635,6 +648,10 @@ def _normalize_score_evidence(evidence: dict[str, Any], scores: dict[str, int])
         normalized[field] = {
             "score": scores[field],
             "fingerprint_metrics_used": metric_ids[:8],
+            "fingerprint_metric_values": {
+                metric_id.upper().split("_", 1)[0]: _fingerprint_value(fingerprint, metric_id)
+                for metric_id in metric_ids[:8]
+            },
             "workbook_rule": workbook_rule,
             "manuscript_evidence": manuscript_evidence,
             "reason": reason,
@@ -648,6 +665,55 @@ def _normalize_score_evidence(evidence: dict[str, Any], scores: dict[str, int])
     return normalized
 
 
+def _required_bottleneck_metric(field: str, fingerprint: dict[str, Any] | None) -> str | None:
+    sentence_mean = _fingerprint_float(fingerprint, "VM-010")
+    reading_age = _fingerprint_float(fingerprint, "VM-025")
+    syllable_variance = _fingerprint_float(fingerprint, "VM-002")
+
+    if field == "rhythm" and syllable_variance is not None and syllable_variance > 5:
+        return "VM-002"
+    if field in {"clarity", "read_aloud_flow"}:
+        if sentence_mean is not None and sentence_mean > 30:
+            return "VM-010"
+        if reading_age is not None and reading_age > 10:
+            return "VM-025"
+    if field == "commercial_viability" and reading_age is not None and reading_age > 10:
+        return "VM-025"
+    return None
+
+
+def _validate_fingerprint_score_alignment(out: dict[str, Any], fingerprint: dict[str, Any] | None) -> None:
+    if not isinstance(fingerprint, dict) or not fingerprint:
+        return
+
+    sentence_mean = _fingerprint_float(fingerprint, "VM-010")
+    reading_age = _fingerprint_float(fingerprint, "VM-025")
+    syllable_variance = _fingerprint_float(fingerprint, "VM-002")
+    problems: list[str] = []
+
+    if sentence_mean is not None and sentence_mean > 30:
+        if out["scores"].get("clarity", 0) > 75:
+            problems.append(f"clarity={out['scores']['clarity']} exceeds cap 75 while VM-010 sentence mean is {sentence_mean}")
+        if out["scores"].get("read_aloud_flow", 0) > 75:
+            problems.append(
+                f"read_aloud_flow={out['scores']['read_aloud_flow']} exceeds cap 75 while VM-010 sentence mean is {sentence_mean}"
+            )
+    if reading_age is not None and reading_age > 10:
+        if out["scores"].get("read_aloud_flow", 0) > 75:
+            problems.append(
+                f"read_aloud_flow={out['scores']['read_aloud_flow']} exceeds cap 75 while VM-025 reading age is {reading_age}"
+            )
+        if out["scores"].get("commercial_viability", 0) > 75:
+            problems.append(
+                f"commercial_viability={out['scores']['commercial_viability']} exceeds cap 75 while VM-025 reading age is {reading_age}"
+            )
+    if syllable_variance is not None and syllable_variance > 5 and out["scores"].get("rhythm", 0) > 75:
+        problems.append(f"rhythm={out['scores']['rhythm']} exceeds cap 75 while VM-002 syllable variance is {syllable_variance}")
+
+    if problems:
+        raise BridgeValidationError("Fingerprint/score contradiction: " + "; ".join(problems))
+
+
 def recompute_gate(metrics: dict[str, int], rubric_matrix: dict[str, Any]) -> str:
     weights = rubric_matrix.get("weights", {}) if isinstance(rubric_matrix, dict) else {}
     if not isinstance(weights, dict) or not weights:
@@ -802,3 +868,23 @@ def _fingerprint_metric_count(fingerprint: dict[str, Any] | None) -> int:
     if not isinstance(fingerprint, dict):
         return 0
     return sum(1 for key in fingerprint if str(key).upper().startswith("VM-"))
+
+
+def _fingerprint_value(fingerprint: dict[str, Any] | None, metric_id: str) -> Any:
+    if not isinstance(fingerprint, dict):
+        return None
+    wanted = str(metric_id).upper().split("_", 1)[0]
+    for key, value in fingerprint.items():
+        if str(key).upper().startswith(wanted):
+            return value
+    return None
+
+
+def _fingerprint_float(fingerprint: dict[str, Any] | None, metric_id: str) -> float | None:
+    value = _fingerprint_value(fingerprint, metric_id)
+    try:
+        if value is None or value == "":
+            return None
+        return float(value)
+    except Exception:
+        return None