[cross-repo from workflow#395] server + workflow + sdk-python: make replay verification a first-class platform contract (#22)

Author: durable-workflow-ops

src/durable_workflow/history_bundle_verify.py

@@ -700,8 +700,19 @@ def _finding(
     severity: str,
     message: str,
     context: Mapping[str, Any] | None = None,
+    path: str | None = None,
 ) -> dict[str, Any]:
-    finding: dict[str, Any] = {"rule": rule, "severity": severity, "message": message}
+    if path is None and context is not None:
+        context_path = context.get("path")
+        if isinstance(context_path, str) and context_path:
+            path = context_path
+
+    finding: dict[str, Any] = {
+        "rule": rule,
+        "severity": severity,
+        "message": message,
+        "path": path,
+    }
     if context:
         finding["context"] = dict(context)
     return finding

src/durable_workflow/replay_verify.py

@@ -191,6 +191,8 @@ class GoldenHistoryReport:
     fixture_schema: str = FIXTURE_SCHEMA
     cases: list[CaseReport] = field(default_factory=list)
     missing_families: list[str] = field(default_factory=list)
+    required_families: list[str] = field(default_factory=list)
+    covered_families: list[str] = field(default_factory=list)
     summary: dict[str, int] = field(default_factory=dict)
 
     @property
@@ -212,13 +214,25 @@ def verdict(self) -> str:
     def promotion_decision(self) -> str:
         return promotion_decision_for(self.verdict)
 
+    @property
+    def evidence(self) -> dict[str, Any]:
+        return {
+            "fixture_count": int(self.summary.get("fixtures", 0)),
+            "case_count": int(self.summary.get("cases", 0)),
+            "required_families": list(self.required_families),
+            "covered_families": list(self.covered_families),
+            "missing_family_count": len(self.missing_families),
+            "missing_families": list(self.missing_families),
+        }
+
     def to_dict(self) -> dict[str, Any]:
         return {
             "schema": self.schema,
             "schema_version": self.schema_version,
             "status": self.status,
             "verdict": self.verdict,
             "promotion_decision": self.promotion_decision,
+            "evidence": self.evidence,
             "fixture_schema": self.fixture_schema,
             "summary": dict(self.summary),
             "missing_families": list(self.missing_families),
@@ -234,6 +248,7 @@ class BundleEntry:
     verdict: str
     promotion_decision: str
     integrity: Mapping[str, Any] | None = None
+    evidence: Mapping[str, Any] | None = None
     reason: str | None = None
 
     def to_dict(self) -> dict[str, Any]:
@@ -242,6 +257,7 @@ def to_dict(self) -> dict[str, Any]:
             "verdict": self.verdict,
             "promotion_decision": self.promotion_decision,
             "reason": self.reason,
+            "evidence": dict(self.evidence) if self.evidence is not None else None,
             "integrity": dict(self.integrity) if self.integrity is not None else None,
         }
 
@@ -263,12 +279,34 @@ class SimulationReport:
     missing_bundles: list[str] = field(default_factory=list)
     error: str | None = None
 
+    @property
+    def evidence(self) -> dict[str, Any]:
+        bundle_count = int(self.summary.get("total", len(self.bundles)))
+        integrity_checked = 0
+        for entry in self.bundles:
+            evidence = entry.evidence or {}
+            if evidence.get("integrity_checked") is True:
+                integrity_checked += 1
+
+        return {
+            "bundle_count": bundle_count,
+            "missing_bundle_count": len(self.missing_bundles),
+            "integrity_checked_count": integrity_checked,
+            "replay_checked_count": 0,
+            "replay_skipped": True,
+            "strict_warnings": any(
+                (entry.evidence or {}).get("strict_warnings") is True
+                for entry in self.bundles
+            ),
+        }
+
     def to_dict(self) -> dict[str, Any]:
         payload: dict[str, Any] = {
             "schema": self.schema,
             "schema_version": self.schema_version,
             "verdict": self.verdict,
             "promotion_decision": self.promotion_decision,
+            "evidence": self.evidence,
             "summary": dict(self.summary),
             "bundles": [entry.to_dict() for entry in self.bundles],
             "missing_bundles": list(self.missing_bundles),
@@ -325,10 +363,16 @@ def verify_golden_history(
     caller decides whether to gate promotion on them.
     """
 
+    required = sorted(set(required_families))
     fixtures = sorted(Path(fixture_dir).glob("*.json"))
 
     if not fixtures:
-        report = GoldenHistoryReport(status=STATUS_FAILED)
+        report = GoldenHistoryReport(
+            status=STATUS_FAILED,
+            missing_families=list(required),
+            required_families=list(required),
+            covered_families=[],
+        )
         report.summary = {
             "fixtures": 0,
             "cases": 0,
@@ -386,7 +430,7 @@ def verify_golden_history(
                 )
             )
 
-    missing = sorted(set(required_families) - covered_families)
+    missing = sorted(set(required) - covered_families)
     summary = _summarize(cases)
     summary["fixtures"] = len(fixtures)
 
@@ -401,6 +445,8 @@ def verify_golden_history(
         status=overall,
         cases=cases,
         missing_families=missing,
+        required_families=list(required),
+        covered_families=sorted(covered_families),
         summary=summary,
     )
 
@@ -695,6 +741,15 @@ def simulate_bundles(
                 path=str(path),
                 verdict=VERDICT_FAILED,
                 promotion_decision=PROMOTION_BLOCK_AND_INVESTIGATE,
+                evidence={
+                    "integrity_checked": False,
+                    "integrity_status": None,
+                    "integrity_finding_count": 0,
+                    "replay_checked": False,
+                    "replay_status": None,
+                    "replay_skipped": True,
+                    "strict_warnings": strict_warnings,
+                },
                 reason=f"bundle_unreadable: {exc}",
             )
             bundles.append(entry)
@@ -713,6 +768,19 @@ def simulate_bundles(
                 verdict=verdict,
                 promotion_decision=decision,
                 integrity=integrity,
+                evidence={
+                    "integrity_checked": True,
+                    "integrity_status": integrity.get("status"),
+                    "integrity_finding_count": int(
+                        (integrity.get("summary") or {}).get(
+                            "findings", len(integrity.get("findings") or [])
+                        )
+                    ),
+                    "replay_checked": False,
+                    "replay_status": None,
+                    "replay_skipped": True,
+                    "strict_warnings": strict_warnings,
+                },
             )
         )
         verdicts.append(verdict)

tests/test_history_bundle_verify.py

@@ -262,6 +262,12 @@ def test_writer_schema_fingerprint_mismatch_in_payload_manifest() -> None:
 
     report = verify_bundle(bundle, signing_key=signing_key)
     assert "payload_manifest.writer_schema_fingerprint_mismatch" in _rule_names(report)
+    finding = next(
+        finding
+        for finding in report["findings"]
+        if finding["rule"] == "payload_manifest.writer_schema_fingerprint_mismatch"
+    )
+    assert finding["path"] == "payloads.output.data"
     assert report["status"] == STATUS_FAILED
 
 
@@ -295,6 +301,12 @@ def test_payload_marked_available_but_missing_is_failed() -> None:
     report = verify_bundle(bundle, signing_key=signing_key)
 
     assert "payload_manifest.payload_missing" in _rule_names(report)
+    finding = next(
+        finding
+        for finding in report["findings"]
+        if finding["rule"] == "payload_manifest.payload_missing"
+    )
+    assert finding["path"] == "payloads.arguments.data"
     assert report["status"] == STATUS_FAILED
 
 

tests/test_replay_verify.py

@@ -136,6 +136,9 @@ def test_verify_golden_history_replays_clean_fixture(tmp_path: Path) -> None:
         "failed": 0,
     }
     assert report.missing_families == []
+    assert report.evidence["required_families"] == ["activity"]
+    assert report.evidence["covered_families"] == ["activity"]
+    assert report.evidence["missing_family_count"] == 0
     assert report.cases[0].status == STATUS_REPLAYED
     assert report.cases[0].family == "activity"
 
@@ -237,6 +240,8 @@ def test_report_to_dict_uses_published_schema(tmp_path: Path) -> None:
     payload = report.to_dict()
     assert payload["schema"] == REPORT_SCHEMA
     assert payload["schema_version"] == REPORT_SCHEMA_VERSION
+    assert payload["evidence"]["fixture_count"] == 1
+    assert payload["evidence"]["case_count"] == 0
 
 
 def _greet_workflows() -> list[type]:
@@ -386,6 +391,7 @@ def test_golden_history_report_to_dict_includes_promotion_decision(tmp_path: Pat
     payload = report.to_dict()
     assert payload["verdict"] == VERDICT_OK
     assert payload["promotion_decision"] == PROMOTION_SAFE_TO_PROMOTE
+    assert payload["evidence"]["missing_family_count"] == 0
     assert payload["cases"][0]["promotion_decision"] == PROMOTION_SAFE_TO_PROMOTE
 
 
@@ -461,9 +467,15 @@ def test_simulate_bundles_aggregates_per_bundle_verdicts(tmp_path: Path) -> None
     assert payload["promotion_decision"] == PROMOTION_BLOCK_AND_INVESTIGATE
     assert payload["summary"]["total"] == 2
     assert payload["summary"][VERDICT_FAILED] == 2
+    assert payload["evidence"]["bundle_count"] == 2
+    assert payload["evidence"]["missing_bundle_count"] == 0
+    assert payload["evidence"]["integrity_checked_count"] == 2
+    assert payload["evidence"]["replay_checked_count"] == 0
+    assert payload["evidence"]["replay_skipped"] is True
     for entry in payload["bundles"]:
         assert entry["verdict"] == VERDICT_FAILED
         assert entry["promotion_decision"] == PROMOTION_BLOCK_AND_INVESTIGATE
+        assert entry["evidence"]["integrity_checked"] is True
 
 
 def test_simulate_bundles_cli(tmp_path: Path) -> None:
@@ -486,6 +498,7 @@ def test_simulate_bundles_cli(tmp_path: Path) -> None:
     assert payload["schema"] == SIMULATION_REPORT_SCHEMA
     assert payload["verdict"] == VERDICT_FAILED
     assert payload["promotion_decision"] == PROMOTION_BLOCK_AND_INVESTIGATE
+    assert payload["evidence"]["bundle_count"] == 1
 
 
 def test_cli_requires_workflows_when_not_simulating(tmp_path: Path) -> None: