Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions python/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions python/ethos_pdf/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
parse_pdf_json,
parse_pdf_markdown,
parse_pdf_text,
proof_summary,
verify,
)

Expand All @@ -53,5 +54,6 @@
"parse_pdf_json",
"parse_pdf_markdown",
"parse_pdf_text",
"proof_summary",
"verify",
]
92 changes: 92 additions & 0 deletions python/ethos_pdf/_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down
106 changes: 106 additions & 0 deletions python/tests/test_cli_surface.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
anchor,
crop_element,
parse_pdf_json,
proof_summary,
verify,
)

Expand Down Expand Up @@ -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,
Expand Down
Loading