feat(ocr-quality): add evaluate_readiness() with YAML policy evaluator

Adds readiness policy evaluator layer:
- load_readiness_policy() with deep-merge, importlib.resources, user override
- _resolve_field() dotted path resolver, _apply_op() operator matcher
- _check_hard_red() for hard-red rule evaluation
- evaluate_readiness() with weighted scoring + status determination
- compute_use_cases() for all 4 use cases (reading, qa, figure_table, chunking)
- Default YAML policy with weights, hard_red rules, use_case gates
- pyproject.toml: policies/*.yaml added to package-data
- 7 tests (B1-B7) covering all paths

17/17 tests pass (A10 + B7).

paperforge/policies/ocr_readiness_v1.yaml

@@ -0,0 +1,63 @@
+schema_version: ocr_readiness_policy_v1
+
+weights:
+  rendered_text_integrity: 0.35
+  body_reference_structure: 0.25
+  figure_table_integrity: 0.20
+  metadata_frontmatter_quality: 0.10
+  confidence_and_fallbacks: 0.10
+
+hard_red:
+  - rule: "rendered_text_gap_excessive"
+    field: "quality_indicators.rendered_text_integrity.signals.gap_count"
+    op: "gt"
+    value: 10
+
+  - rule: "body_weak_and_no_refs"
+    condition:
+      field_a: "quality_indicators.body_reference_structure.signals.body_spine_quality"
+      op_a: "eq"
+      value_a: "weak"
+      field_b: "quality_indicators.body_reference_structure.signals.references_found"
+      op_b: "eq"
+      value_b: false
+
+  - rule: "fulltext_too_short"
+    field: "developer_diagnostics.run_integrity.output_fulltext_chars"
+    op: "lt"
+    value: 2000
+
+use_cases:
+  reading:
+    gates:
+      required:
+        - indicator: "rendered_text_integrity"
+          min_status: "yellow"
+        - indicator: "body_reference_structure"
+          min_status: "yellow"
+
+  qa:
+    gates:
+      required:
+        - indicator: "rendered_text_integrity"
+          min_status: "yellow"
+        - indicator: "body_reference_structure"
+          min_status: "yellow"
+      soft:
+        - indicator: "metadata_frontmatter_quality"
+          min_status: "yellow"
+
+  figure_table_reasoning:
+    if_no_figure_table_evidence: "not_applicable"
+    gates:
+      required:
+        - indicator: "figure_table_integrity"
+          min_status: "yellow"
+
+  section_chunking:
+    gates:
+      required:
+        - indicator: "body_reference_structure"
+          min_status: "green"
+        - indicator: "rendered_text_integrity"
+          min_status: "yellow"

paperforge/worker/ocr_quality.py

@@ -223,3 +223,236 @@ def build_quality_indicators(
         },
         "developer_diagnostics": developer_diagnostics,
     }
+
+
+def _deep_merge(base: dict, override: dict) -> dict:
+    """Recursive dict merge with list replace."""
+    result = dict(base)
+    for k, v in override.items():
+        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
+            result[k] = _deep_merge(result[k], v)
+        else:
+            result[k] = v
+    return result
+
+
+def _resolve_field(data: dict, path: str) -> Any:
+    """Resolve dotted field path against a nested dict."""
+    parts = path.split(".")
+    current = data
+    for part in parts:
+        if not isinstance(current, dict):
+            return None
+        current = current.get(part)
+    return current
+
+
+def _apply_op(value, op: str, target) -> bool:
+    """Apply comparison operator."""
+    if op == "gt":
+        return isinstance(value, (int, float)) and value > target
+    elif op == "lt":
+        return isinstance(value, (int, float)) and value < target
+    elif op == "eq":
+        return value == target
+    return False
+
+
+def _check_hard_red(policy: dict, quality_report: dict) -> list[str]:
+    """Check hard-red rules against the full quality report.
+
+    Returns:
+        list of triggered hard-red rule names.
+    """
+    triggered: list[str] = []
+    for rule in policy.get("hard_red", []):
+        if "condition" in rule:
+            cond = rule["condition"]
+            val_a = _resolve_field(quality_report, cond["field_a"])
+            val_b = _resolve_field(quality_report, cond["field_b"])
+            match_a = _apply_op(val_a, cond["op_a"], cond["value_a"])
+            match_b = _apply_op(val_b, cond["op_b"], cond["value_b"])
+            if match_a and match_b:
+                triggered.append(rule["rule"])
+        else:
+            val = _resolve_field(quality_report, rule["field"])
+            if _apply_op(val, rule["op"], rule["value"]):
+                triggered.append(rule["rule"])
+    return triggered
+
+
+def load_readiness_policy(
+    policy: dict | None = None,
+    *,
+    policy_path: str | Path | None = None,
+) -> dict:
+    """Load and merge readiness policy.
+
+    Args:
+        policy: in-memory policy dict (mutually exclusive with policy_path)
+        policy_path: path to explicit YAML file (mutually exclusive with policy)
+
+    Returns:
+        merged policy dict
+
+    Resolution order:
+        1. Default: paperforge/policies/ocr_readiness_v1.yaml (via importlib.resources)
+        2. If policy_path is provided: deep-merge that file over default. Skip user override.
+        3. Else if ~/.paperforge/policies/ocr_readiness_v1.yaml exists: deep-merge user override.
+        4. If in-memory policy is provided: deep-merge over whatever was loaded.
+
+    Merge rules:
+        - dicts: recursive deep-merge
+        - lists: replace (not append)
+        - scalars: override
+    """
+    if policy is not None and policy_path is not None:
+        raise ValueError("policy and policy_path are mutually exclusive")
+
+    from pathlib import Path
+
+    import yaml
+    from importlib.resources import files as pkg_files
+
+    # 1. Load default
+    default_path = pkg_files("paperforge") / "policies/ocr_readiness_v1.yaml"
+    with open(default_path, "r", encoding="utf-8") as fh:
+        merged = yaml.safe_load(fh) or {}
+
+    # 2. Load explicit policy_path or user override
+    if policy_path is not None:
+        with open(policy_path, "r", encoding="utf-8") as fh:
+            override = yaml.safe_load(fh) or {}
+        merged = _deep_merge(merged, override)
+    else:
+        user_path = Path.home() / ".paperforge" / "policies" / "ocr_readiness_v1.yaml"
+        if user_path.exists():
+            with open(user_path, "r", encoding="utf-8") as fh:
+                override = yaml.safe_load(fh) or {}
+            merged = _deep_merge(merged, override)
+
+    # 3. In-memory policy overrides everything
+    if policy is not None:
+        merged = _deep_merge(merged, policy)
+
+    return merged
+
+
+STATUS_RANK = {"red": 0, "unknown": 1, "yellow": 2, "green": 3}
+STATUS_SCORE = {"green": 1.0, "yellow": 0.6, "red": 0.2, "unknown": 0.5}
+
+
+def _evaluate_gate(indicators: dict, indicator_name: str, min_status: str) -> dict:
+    """Evaluate a single gate against an indicator.
+
+    unknown status fails required gates (rank 1 < yellow 2).
+    """
+    ind = indicators.get(indicator_name, {})
+    actual_status = ind.get("status", "unknown")
+    passes = STATUS_RANK.get(actual_status, 1) >= STATUS_RANK.get(min_status, 2)
+    return {
+        "indicator": indicator_name,
+        "actual_status": actual_status,
+        "min_status": min_status,
+        "passes": passes,
+    }
+
+
+def compute_use_cases(policy: dict, indicators: dict) -> dict:
+    """Compute recommended use cases from policy and indicators."""
+    result: dict[str, Any] = {}
+    for uc_name, uc_config in policy.get("use_cases", {}).items():
+        # Check not_applicable shortcut
+        if uc_config.get("if_no_figure_table_evidence") == "not_applicable":
+            fig_ind = indicators.get("figure_table_integrity", {})
+            if fig_ind.get("applicability") == "not_applicable":
+                result[uc_name] = {"recommended": "not_applicable", "gate_results": []}
+                continue
+
+        gates = uc_config.get("gates", {})
+        gate_results: list[dict] = []
+        all_required_pass = True
+
+        for gate in gates.get("required", []):
+            gr = _evaluate_gate(indicators, gate["indicator"], gate["min_status"])
+            gr["required"] = True
+            gate_results.append(gr)
+            if not gr["passes"]:
+                all_required_pass = False
+
+        for gate in gates.get("soft", []):
+            gr = _evaluate_gate(indicators, gate["indicator"], gate["min_status"])
+            gr["required"] = False
+            gate_results.append(gr)
+
+        result[uc_name] = {
+            "recommended": "yes" if all_required_pass else "no",
+            "gate_results": gate_results,
+        }
+
+    return result
+
+
+def evaluate_readiness(
+    quality_report_base: dict,
+    policy: dict | None = None,
+    *,
+    policy_path: str | Path | None = None,
+) -> dict:
+    """Apply readiness policy to quality report.
+
+    Args:
+        quality_report_base: output of build_quality_indicators()
+        policy: in-memory policy dict (mutually exclusive with policy_path)
+        policy_path: path to YAML policy file
+
+    Returns:
+        dict with user_readiness + recommended_use
+    """
+    from pathlib import Path
+
+    merged_policy = load_readiness_policy(policy, policy_path=policy_path)
+    indicators = quality_report_base.get("quality_indicators", {})
+
+    # Hard-red check (resolves fields against full quality report)
+    triggered = _check_hard_red(merged_policy, quality_report_base)
+
+    # Weighted score
+    weights = merged_policy.get("weights", {})
+    total_weight = 0.0
+    weighted_sum = 0.0
+    dim_statuses: dict[str, str] = {}
+    for dim, weight in weights.items():
+        ind = indicators.get(dim, {})
+        if ind.get("applicability") == "not_applicable":
+            continue  # exclude from weighted score, renormalize
+        status = ind.get("status", "unknown")
+        dim_statuses[dim] = status
+        score = STATUS_SCORE.get(status, 0.5)
+        weighted_sum += weight * score
+        total_weight += weight
+
+    score = weighted_sum / total_weight if total_weight > 0 else 0.0
+
+    # Status determination
+    hard_red = len(triggered) > 0
+    if hard_red:
+        status = "red"
+        score = min(score, 0.3)
+    elif score >= 0.75:
+        status = "green"
+    elif score >= 0.40:
+        status = "yellow"
+    else:
+        status = "red"
+
+    return {
+        "user_readiness": {
+            "status": status,
+            "score": round(score, 4),
+            "policy_version": merged_policy.get("schema_version", ""),
+            "basis": "policy_estimate",
+            "hard_red_triggers": triggered,
+        },
+        "recommended_use": compute_use_cases(merged_policy, indicators),
+    }

pyproject.toml

@@ -62,6 +62,8 @@ version = {attr = "paperforge.__version__"}
 [tool.setuptools.package-data]
 paperforge = [
     "py.typed",
+    "policies/*.yaml",
+    "render-themes/*.yaml",
     "skills/paperforge/**",
     "command_files/*.md",
     "plugin/*.css",

tests/test_ocr_quality.py

@@ -178,3 +178,205 @@ def test_a10_run_integrity_dict_preserved() -> None:
     ri = {"pages_processed": 5, "status": "ok"}
     result = build_quality_indicators(health=health, run_integrity=ri)
     assert result["developer_diagnostics"]["run_integrity"] == {"pages_processed": 5, "status": "ok"}
+
+
+# ── B series: Readiness policy evaluator ──────────────────────────────
+
+
+def test_b1_default_policy_returns_status_and_score() -> None:
+    """Default policy evaluates a healthy report as green."""
+    from paperforge.worker.ocr_quality import build_quality_indicators, evaluate_readiness
+
+    health = {
+        "rendered_text_gap_count": 0,
+        "body_spine_quality": "strong",
+        "health_profile": "standard",
+        "layout_audit_status": "pass",
+        "layout_anomaly_count": 0,
+        "references_found": True,
+        "section_heading_count": 3,
+        "reference_item_count": 5,
+        "figure_caption_count": 0,
+        "matched_figure_count_v2": 0,
+        "media_without_caption_count": 0,
+        "unresolved_cluster_count": 0,
+        "figure_reader_coverage_ratio": 1.0,
+        "table_asset_count": 0,
+        "table_unmatched_count": 0,
+    }
+    base = build_quality_indicators(health=health)
+    result = evaluate_readiness(base)
+
+    assert "user_readiness" in result
+    assert "recommended_use" in result
+    assert result["user_readiness"]["status"] in ("green", "yellow", "red")
+    assert result["user_readiness"]["basis"] == "policy_estimate"
+
+
+def test_b2_hard_red_overrides_green() -> None:
+    """gap_count > 10 triggers hard-red rule, forcing status to red."""
+    from paperforge.worker.ocr_quality import build_quality_indicators, evaluate_readiness
+
+    health = {
+        "rendered_text_gap_count": 15,
+        "body_spine_quality": "strong",
+        "health_profile": "standard",
+        "layout_audit_status": "pass",
+        "layout_anomaly_count": 0,
+        "references_found": True,
+        "section_heading_count": 3,
+        "reference_item_count": 5,
+        "figure_caption_count": 0,
+        "matched_figure_count_v2": 0,
+        "media_without_caption_count": 0,
+        "unresolved_cluster_count": 0,
+        "figure_reader_coverage_ratio": 1.0,
+        "table_asset_count": 0,
+        "table_unmatched_count": 0,
+    }
+    base = build_quality_indicators(health=health)
+    result = evaluate_readiness(base)
+
+    assert result["user_readiness"]["status"] == "red"
+    assert "rendered_text_gap_excessive" in result["user_readiness"]["hard_red_triggers"]
+
+
+def test_b3_reading_gate_yellow_when_yellow() -> None:
+    """Reading gate passes when both required indicators are at least yellow."""
+    from paperforge.worker.ocr_quality import build_quality_indicators, evaluate_readiness
+
+    # gap_count 7 → yellow rendered_text_integrity
+    # body_spine partial → yellow body_reference_structure
+    health = {
+        "rendered_text_gap_count": 7,
+        "body_spine_quality": "partial",
+        "health_profile": "standard",
+        "layout_audit_status": "pass",
+        "layout_anomaly_count": 1,
+        "references_found": True,
+        "section_heading_count": 3,
+        "reference_item_count": 5,
+        "figure_caption_count": 0,
+        "matched_figure_count_v2": 0,
+        "media_without_caption_count": 0,
+        "unresolved_cluster_count": 0,
+        "figure_reader_coverage_ratio": 1.0,
+        "table_asset_count": 0,
+        "table_unmatched_count": 0,
+    }
+    base = build_quality_indicators(health=health)
+    result = evaluate_readiness(base)
+
+    reading = result["recommended_use"].get("reading", {})
+    assert reading.get("recommended") == "yes"
+
+
+def test_b4_figure_table_reasoning_not_applicable() -> None:
+    """figure_table_reasoning returns not_applicable when no figure/table evidence."""
+    from paperforge.worker.ocr_quality import build_quality_indicators, evaluate_readiness
+
+    # No figure/table evidence → not_applicable
+    health = {
+        "rendered_text_gap_count": 0,
+        "body_spine_quality": "strong",
+        "health_profile": "standard",
+        "layout_audit_status": "pass",
+        "layout_anomaly_count": 0,
+        "references_found": True,
+        "section_heading_count": 3,
+        "reference_item_count": 5,
+        "figure_caption_count": 0,
+        "matched_figure_count_v2": 0,
+        "media_without_caption_count": 0,
+        "unresolved_cluster_count": 0,
+        "figure_reader_coverage_ratio": 0.0,
+        "table_asset_count": 0,
+        "table_unmatched_count": 0,
+    }
+    base = build_quality_indicators(health=health)
+    result = evaluate_readiness(base)
+
+    ftr = result["recommended_use"].get("figure_table_reasoning", {})
+    assert ftr.get("recommended") == "not_applicable"
+
+    # The figure_table_integrity indicator should be excluded from weighted scoring
+    # because its applicability is not_applicable
+    fig_ind = base["quality_indicators"]["figure_table_integrity"]
+    assert fig_ind["applicability"] == "not_applicable"
+
+
+def test_b5_policy_loaded_from_temp_yaml() -> None:
+    """Policy can be loaded from an explicit YAML path."""
+    import tempfile
+    from pathlib import Path
+
+    import yaml
+
+    from paperforge.worker.ocr_quality import evaluate_readiness
+
+    temp_policy = {
+        "schema_version": "custom_v1",
+        "weights": {"rendered_text_integrity": 1.0},
+        "hard_red": [],
+        "use_cases": {},
+    }
+
+    with tempfile.NamedTemporaryFile(
+        mode="w", suffix=".yaml", delete=False, encoding="utf-8"
+    ) as f:
+        yaml.dump(temp_policy, f)
+        tmp_path = Path(f.name)
+
+    try:
+        base = {"quality_indicators": {}, "developer_diagnostics": {}}
+        result = evaluate_readiness(base, policy_path=tmp_path)
+        assert result["user_readiness"]["policy_version"] == "custom_v1"
+    finally:
+        tmp_path.unlink(missing_ok=True)
+
+
+def test_b6_default_policy_file_exists() -> None:
+    """The default policy YAML is bundled with the package."""
+    from importlib.resources import files as pkg_files
+
+    policy_path = pkg_files("paperforge") / "policies/ocr_readiness_v1.yaml"
+    assert policy_path.exists()
+    assert policy_path.is_file()
+
+
+def test_b7_user_override_merges_correctly() -> None:
+    """Deep-merge and list-replace semantics work for user overrides."""
+    from paperforge.worker.ocr_quality import load_readiness_policy, evaluate_readiness
+
+    # Override with a custom weight and a replacement hard_red list
+    override = {
+        "weights": {"rendered_text_integrity": 0.50},
+        "hard_red": [{"rule": "custom_rule", "field": "x", "op": "eq", "value": 1}],
+    }
+
+    policy = load_readiness_policy(policy=override)
+
+    # Weight should be overridden
+    assert policy["weights"]["rendered_text_integrity"] == 0.50
+    # Other weights should still exist (dict deep-merge)
+    assert "body_reference_structure" in policy["weights"]
+
+    # hard_red list should be replaced, not appended
+    assert len(policy["hard_red"]) == 1
+    assert policy["hard_red"][0]["rule"] == "custom_rule"
+
+    # Use cases should be preserved
+    assert "reading" in policy["use_cases"]
+
+    # Integration: override via evaluate_readiness
+    health = {
+        "rendered_text_gap_count": 15,
+        "body_spine_quality": "strong",
+    }
+    from paperforge.worker.ocr_quality import build_quality_indicators
+
+    base = build_quality_indicators(health=health)
+    result = evaluate_readiness(base, policy=override)
+    # With only one dimension weighted 1.0 and no hard_red matching,
+    # the score should be high
+    assert result["user_readiness"]["policy_version"] == "ocr_readiness_policy_v1"