diff --git a/python/README.md b/python/README.md index 28db322..781b216 100644 --- a/python/README.md +++ b/python/README.md @@ -33,6 +33,7 @@ Public API: - `parse_pdf_text` - `crop_element` - `verify` +- `proof_summary` - `anchor` The current module is intentionally thin: it shells out to a caller-provided local `ethos` CLI @@ -111,6 +112,22 @@ Verify exit semantics: - exit `1` with JSON returns a negative verification report when `fail_on_ungrounded=True`; - exit `>=2` raises `EthosCommandError` or a more specific subclass. +Use `proof_summary(report)` when a product or API wrapper needs the same derived status as the Rust +`VerificationReport::proof_summary()` helper: + +```python +from ethos_pdf import EthosCli, proof_summary + +ethos = EthosCli(binary="/path/to/ethos") +report = ethos.verify("source.ethos.json", citations="citations.json") +summary = proof_summary(report) +print(summary["proof_status"]) +``` + +The summary is not a replacement for the canonical verification report. It deterministically +derives `proof_status`, `request_certified`, reusable grounded check ids, needs-review check ids, +and proof limitations from the report that `ethos verify` already emitted. + Run the focused tests with: ```sh diff --git a/python/ethos_pdf/__init__.py b/python/ethos_pdf/__init__.py index 9d72f93..97e23cf 100644 --- a/python/ethos_pdf/__init__.py +++ b/python/ethos_pdf/__init__.py @@ -32,6 +32,7 @@ parse_pdf_json, parse_pdf_markdown, parse_pdf_text, + proof_summary, verify, ) @@ -53,5 +54,6 @@ "parse_pdf_json", "parse_pdf_markdown", "parse_pdf_text", + "proof_summary", "verify", ] diff --git a/python/ethos_pdf/_cli.py b/python/ethos_pdf/_cli.py index b702995..39b8138 100644 --- a/python/ethos_pdf/_cli.py +++ b/python/ethos_pdf/_cli.py @@ -26,6 +26,8 @@ _DOC_PARSE_FORMATS = frozenset(("json", "markdown", "text")) _DEFAULT_CROP_CHECK_ID = "v0001" +_CAPABILITY_LIMITED = "capability_limited" +_GROUNDED = "grounded" class EthosPythonSurfaceError(Exception): @@ -621,6 +623,66 @@ def anchor( ) +def proof_summary(report: Mapping[str, Any]) -> Dict[str, Any]: + """Derive product-facing proof status from a verification report. + + This mirrors the Rust `VerificationReport::proof_summary()` helper without + changing the canonical JSON report. `request_certified` mirrors + `all_evidence_grounded`; reusable grounded checks exclude report-level stale + fingerprints and claim-level `semantic_unverified` cases. + """ + + checks = _list_value(report.get("checks")) + fingerprint_stale = bool(report.get("fingerprint_stale", False)) + reusable_grounded_check_ids = [] + needs_review_check_ids = [] + + for check in checks: + if not isinstance(check, Mapping): + continue + check_id = check.get("id") + if not isinstance(check_id, str): + continue + if _is_reusable_grounded_check(check, fingerprint_stale): + reusable_grounded_check_ids.append(check_id) + else: + needs_review_check_ids.append(check_id) + + request_certified = bool(report.get("all_evidence_grounded", False)) + if request_certified: + status = "verified" + elif reusable_grounded_check_ids: + status = "partially_verified" + else: + status = "unverified" + + limitations = [] + if _has_capability_limit(report, checks): + limitations.append(_CAPABILITY_LIMITED) + if fingerprint_stale: + limitations.append("stale_fingerprint") + if _list_value(report.get("unsupported_claim_kinds")): + limitations.append("unsupported_claim_kind") + if any( + isinstance(check, Mapping) and check.get("status") != _GROUNDED + for check in checks + ): + limitations.append("non_grounded_checks") + if any( + isinstance(check, Mapping) and bool(check.get("semantic_unverified", False)) + for check in checks + ): + limitations.append("semantic_unverified") + + return { + "proof_status": status, + "request_certified": request_certified, + "reusable_grounded_check_ids": reusable_grounded_check_ids, + "needs_review_check_ids": needs_review_check_ids, + "proof_limitations": limitations, + } + + def _validate_timeout_seconds(timeout_seconds: Optional[float]) -> None: if timeout_seconds is not None and timeout_seconds <= 0: raise ValueError("timeout_seconds must be greater than zero when provided") @@ -649,6 +711,36 @@ def _load_json_stdout(stdout: str, command: Sequence[str]) -> Any: ) from exc +def _is_reusable_grounded_check( + check: Mapping[str, Any], fingerprint_stale: bool +) -> bool: + return ( + not fingerprint_stale + and check.get("status") == _GROUNDED + and not bool(check.get("semantic_unverified", False)) + ) + + +def _has_capability_limit( + report: Mapping[str, Any], checks: Sequence[Any] +) -> bool: + if _list_value(report.get("capability_limits")): + return True + if _CAPABILITY_LIMITED in _list_value(report.get("warnings")): + return True + return any( + isinstance(check, Mapping) + and _CAPABILITY_LIMITED in _list_value(check.get("warnings")) + for check in checks + ) + + +def _list_value(value: Any) -> Sequence[Any]: + if isinstance(value, list): + return value + return () + + def _decode_utf8(data: bytes, command: Sequence[str], stream: str) -> str: try: return data.decode("utf-8") diff --git a/python/tests/test_cli_surface.py b/python/tests/test_cli_surface.py index cc84192..1495e0f 100644 --- a/python/tests/test_cli_surface.py +++ b/python/tests/test_cli_surface.py @@ -36,6 +36,7 @@ anchor, crop_element, parse_pdf_json, + proof_summary, verify, ) @@ -470,6 +471,111 @@ def test_verify_rejects_non_json_format(self) -> None: output_format="summary", ) + def test_proof_summary_keeps_capability_limit_visible_without_failing_certified_request( + self, + ) -> None: + report = { + "all_evidence_grounded": True, + "fingerprint_stale": False, + "capability_limits": ["missing_fingerprint"], + "unsupported_claim_kinds": [], + "warnings": ["capability_limited"], + "checks": [ + { + "id": "v0001", + "status": "grounded", + "semantic_unverified": False, + "warnings": [], + } + ], + } + + result = proof_summary(report) + + self.assertEqual(result["proof_status"], "verified") + self.assertTrue(result["request_certified"]) + self.assertEqual(result["reusable_grounded_check_ids"], ["v0001"]) + self.assertEqual(result["needs_review_check_ids"], []) + self.assertEqual(result["proof_limitations"], ["capability_limited"]) + + def test_proof_summary_marks_mixed_report_as_partially_verified(self) -> None: + report = { + "all_evidence_grounded": False, + "fingerprint_stale": False, + "capability_limits": [], + "unsupported_claim_kinds": ["region"], + "warnings": [], + "checks": [ + { + "id": "v0001", + "status": "grounded", + "semantic_unverified": False, + "warnings": [], + }, + { + "id": "v0002", + "status": "unsupported_claim_kind", + "semantic_unverified": False, + "warnings": [], + }, + ], + } + + result = proof_summary(report) + + self.assertEqual(result["proof_status"], "partially_verified") + self.assertFalse(result["request_certified"]) + self.assertEqual(result["reusable_grounded_check_ids"], ["v0001"]) + self.assertEqual(result["needs_review_check_ids"], ["v0002"]) + self.assertEqual( + result["proof_limitations"], + ["unsupported_claim_kind", "non_grounded_checks"], + ) + + def test_proof_summary_excludes_stale_and_semantic_grounded_checks(self) -> None: + stale_report = { + "all_evidence_grounded": False, + "fingerprint_stale": True, + "capability_limits": [], + "unsupported_claim_kinds": [], + "warnings": [], + "checks": [ + { + "id": "v0001", + "status": "grounded", + "semantic_unverified": False, + "warnings": [], + } + ], + } + semantic_report = { + "all_evidence_grounded": False, + "fingerprint_stale": False, + "capability_limits": [], + "unsupported_claim_kinds": [], + "warnings": [], + "checks": [ + { + "id": "v0001", + "status": "grounded", + "semantic_unverified": True, + "warnings": [], + } + ], + } + + stale = proof_summary(stale_report) + semantic = proof_summary(semantic_report) + + self.assertEqual(stale["proof_status"], "unverified") + self.assertEqual(stale["reusable_grounded_check_ids"], []) + self.assertEqual(stale["needs_review_check_ids"], ["v0001"]) + self.assertEqual(stale["proof_limitations"], ["stale_fingerprint"]) + self.assertEqual(semantic["proof_status"], "unverified") + self.assertEqual(semantic["reusable_grounded_check_ids"], []) + self.assertEqual(semantic["needs_review_check_ids"], ["v0001"]) + self.assertEqual(semantic["proof_limitations"], ["semantic_unverified"]) + def test_anchor_maps_source_evidence_refs_and_grounding(self) -> None: result = anchor( self.document,