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

Author: durable-workflow-ops

README.md

@@ -223,7 +223,10 @@ durable-workflow-history-bundle-verify exported-history-bundles/run-001.json \
 `promotion_decision` vocabulary as the platform replay contract. Golden-history
 mode replays cross-runtime fixtures against registered workflow classes;
 `--simulate-bundles` integrity-checks every exported history bundle in a
-directory and reports missing bundle evidence as a blocking result.
+directory and reports missing bundle evidence as a blocking result. Because
+bundle simulation does not execute workflow code in Python, a clean
+integrity-only simulation recommends `review_before_promote` rather than
+`safe_to_promote`.
 
 ## External payload storage
 

src/durable_workflow/__init__.py

@@ -141,6 +141,7 @@
     SimulationReport,
     aggregate_verdicts,
     promotion_decision_for,
+    promotion_decision_for_report,
     simulate_bundles,
     verify_golden_history,
     verify_replay,
@@ -294,6 +295,7 @@
     "SimulationReport",
     "aggregate_verdicts",
     "promotion_decision_for",
+    "promotion_decision_for_report",
     "simulate_bundles",
     "verify_golden_history",
     "verify_replay",

src/durable_workflow/history_bundle_verify.py

@@ -350,17 +350,31 @@ def _check_commands(bundle: Mapping[str, Any], findings: list[dict[str, Any]]) -
             continue
 
         applied = _string_or_none(command.get("applied_at"))
+        rejected = _string_or_none(command.get("rejected_at"))
         status = _string_or_none(command.get("status"))
-        if applied is not None and event_command_ids and command_id not in event_command_ids:
+        outcome = _string_or_none(command.get("outcome"))
+        settled = (
+            status in {"applied", "rejected"}
+            or outcome in {"applied", "rejected"}
+            or applied is not None
+            or rejected is not None
+        )
+
+        if settled and command_id not in event_command_ids:
             findings.append(
                 _finding(
                     "commands.history_event_missing",
                     SEVERITY_WARNING,
                     (
-                        f"commands[{index}] (id={command_id}, status={status or 'unknown'}) was applied "
+                        f"commands[{index}] (id={command_id}, status={status or 'unknown'}) was settled "
                         "but no history event references it."
                     ),
-                    {"index": index, "command_id": command_id, "command_status": status},
+                    {
+                        "index": index,
+                        "command_id": command_id,
+                        "command_status": status,
+                        "command_outcome": outcome,
+                    },
                 )
             )
 

src/durable_workflow/replay_verify.py

@@ -103,6 +103,27 @@ def promotion_decision_for(verdict: str) -> str:
     }.get(verdict, PROMOTION_BLOCK_AND_INVESTIGATE)
 
 
+def promotion_decision_for_report(
+    verdict: str, evidence: Mapping[str, Any] | None = None
+) -> str:
+    """Map a report verdict and evidence block to a promotion decision.
+
+    A clean integrity-only report is useful, but it is not enough to
+    claim replay-safe promotion. Keep the verdict tied to the checks that
+    ran while reducing the recommendation to review when replay was
+    intentionally skipped.
+    """
+
+    decision = promotion_decision_for(verdict)
+    if (
+        decision == PROMOTION_SAFE_TO_PROMOTE
+        and evidence is not None
+        and evidence.get("replay_skipped") is True
+    ):
+        return PROMOTION_REVIEW_BEFORE_PROMOTE
+    return decision
+
+
 def aggregate_verdicts(verdicts: Sequence[str]) -> str:
     """Reduce a list of verdicts to the strictest one.
 
@@ -252,12 +273,16 @@ class BundleEntry:
     reason: str | None = None
 
     def to_dict(self) -> dict[str, Any]:
+        evidence = dict(self.evidence) if self.evidence is not None else None
         return {
             "bundle_path": self.path,
             "verdict": self.verdict,
-            "promotion_decision": self.promotion_decision,
+            "promotion_decision": promotion_decision_for_report(
+                self.verdict,
+                evidence,
+            ),
             "reason": self.reason,
-            "evidence": dict(self.evidence) if self.evidence is not None else None,
+            "evidence": evidence,
             "integrity": dict(self.integrity) if self.integrity is not None else None,
         }
 
@@ -301,12 +326,16 @@ def evidence(self) -> dict[str, Any]:
         }
 
     def to_dict(self) -> dict[str, Any]:
+        evidence = self.evidence
         payload: dict[str, Any] = {
             "schema": self.schema,
             "schema_version": self.schema_version,
             "verdict": self.verdict,
-            "promotion_decision": self.promotion_decision,
-            "evidence": self.evidence,
+            "promotion_decision": promotion_decision_for_report(
+                self.verdict,
+                evidence,
+            ),
+            "evidence": evidence,
             "summary": dict(self.summary),
             "bundles": [entry.to_dict() for entry in self.bundles],
             "missing_bundles": list(self.missing_bundles),
@@ -760,27 +789,28 @@ def simulate_bundles(
 
         integrity = history_bundle_verify.verify_bundle_json(payload, signing_key)
         verdict = _integrity_status_to_verdict(integrity, strict_warnings)
-        decision = promotion_decision_for(verdict)
+
+        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,
+        }
 
         bundles.append(
             BundleEntry(
                 path=str(path),
                 verdict=verdict,
-                promotion_decision=decision,
+                promotion_decision=promotion_decision_for_report(verdict, evidence),
                 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,
-                },
+                evidence=evidence,
             )
         )
         verdicts.append(verdict)
@@ -789,12 +819,14 @@ def simulate_bundles(
 
     overall = aggregate_verdicts(verdicts)
 
-    return SimulationReport(
+    report = SimulationReport(
         verdict=overall,
         promotion_decision=promotion_decision_for(overall),
         summary=summary,
         bundles=bundles,
     )
+    report.promotion_decision = promotion_decision_for_report(overall, report.evidence)
+    return report
 
 
 def _integrity_status_to_verdict(
@@ -974,6 +1006,7 @@ def main(argv: Sequence[str] | None = None) -> int:
     "BundleEntry",
     "SimulationReport",
     "promotion_decision_for",
+    "promotion_decision_for_report",
     "aggregate_verdicts",
     "verify_replay",
     "verify_golden_history",

tests/test_history_bundle_verify.py

@@ -283,6 +283,27 @@ def test_required_evidence_sections_are_enforced() -> None:
     assert report["status"] == STATUS_FAILED
 
 
+def test_settled_command_without_history_event_warns_even_when_no_events_reference_commands() -> None:
+    signing_key = "secret"
+    bundle = _base_bundle()
+    bundle["commands"].append(
+        {
+            "id": "cmd-orphan",
+            "sequence": 2,
+            "type": "workflow.start",
+            "status": "applied",
+            "outcome": "applied",
+            "applied_at": "2026-04-09T12:00:01.000000Z",
+        }
+    )
+    bundle = _sign(bundle, signing_key, "key-1")
+
+    report = verify_bundle(bundle, signing_key=signing_key)
+
+    assert report["status"] == STATUS_WARNING
+    assert "commands.history_event_missing" in _rule_names(report)
+
+
 def test_payload_marked_available_but_missing_is_failed() -> None:
     signing_key = "secret"
     bundle = _base_bundle()

tests/test_replay_verify.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 from pathlib import Path
+from typing import Any
 
 from durable_workflow import workflow
 from durable_workflow.replay_verify import (
@@ -28,6 +30,7 @@
     aggregate_verdicts,
     main as replay_verify_main,
     promotion_decision_for,
+    promotion_decision_for_report,
     simulate_bundles,
     verify_golden_history,
     verify_replay,
@@ -350,6 +353,17 @@ def test_promotion_decision_for_unknown_verdict_blocks() -> None:
     assert promotion_decision_for("totally-bogus") == PROMOTION_BLOCK_AND_INVESTIGATE
 
 
+def test_promotion_decision_for_report_reviews_skipped_replay() -> None:
+    assert (
+        promotion_decision_for_report(VERDICT_OK, {"replay_skipped": True})
+        == PROMOTION_REVIEW_BEFORE_PROMOTE
+    )
+    assert (
+        promotion_decision_for_report(VERDICT_OK, {"replay_skipped": False})
+        == PROMOTION_SAFE_TO_PROMOTE
+    )
+
+
 def test_aggregate_verdicts_picks_strictest() -> None:
     assert aggregate_verdicts([VERDICT_OK, VERDICT_OK]) == VERDICT_OK
     assert aggregate_verdicts([VERDICT_OK, "warning"]) == "warning"
@@ -478,6 +492,24 @@ def test_simulate_bundles_aggregates_per_bundle_verdicts(tmp_path: Path) -> None
         assert entry["evidence"]["integrity_checked"] is True
 
 
+def test_simulate_bundles_clean_integrity_only_recommends_review(tmp_path: Path) -> None:
+    bundle_dir = tmp_path / "bundles"
+    bundle_dir.mkdir()
+    (bundle_dir / "clean.json").write_text(
+        json.dumps(_history_export_bundle()),
+        encoding="utf-8",
+    )
+
+    report = simulate_bundles(bundle_dir)
+    payload = report.to_dict()
+
+    assert report.verdict == VERDICT_OK
+    assert report.promotion_decision == PROMOTION_REVIEW_BEFORE_PROMOTE
+    assert payload["promotion_decision"] == PROMOTION_REVIEW_BEFORE_PROMOTE
+    assert payload["evidence"]["replay_skipped"] is True
+    assert payload["bundles"][0]["promotion_decision"] == PROMOTION_REVIEW_BEFORE_PROMOTE
+
+
 def test_simulate_bundles_cli(tmp_path: Path) -> None:
     bundle_dir = tmp_path / "bundles"
     bundle_dir.mkdir()
@@ -511,3 +543,71 @@ def test_cli_requires_workflows_when_not_simulating(tmp_path: Path) -> None:
         replay_verify_main([str(fixture_dir)])
 
     assert excinfo.value.code == 2
+
+
+def _history_export_bundle() -> dict[str, Any]:
+    bundle: dict[str, Any] = {
+        "schema": "durable-workflow.v2.history-export",
+        "schema_version": 1,
+        "exported_at": "2026-04-09T12:00:00.000000Z",
+        "dedupe_key": "run-1:1:2026-04-09T12:00:00.000000Z",
+        "history_complete": True,
+        "workflow": {
+            "instance_id": "instance-1",
+            "run_id": "run-1",
+            "run_number": 1,
+            "workflow_type": "history.export",
+            "workflow_class": "tests.Replay",
+            "last_history_sequence": 1,
+        },
+        "payloads": {
+            "codec": "json",
+            "arguments": {"available": False, "data": None},
+            "output": {"available": False, "data": None},
+        },
+        "history_events": [
+            {
+                "id": "evt-1",
+                "sequence": 1,
+                "type": "WorkflowStarted",
+                "workflow_task_id": None,
+                "workflow_command_id": None,
+                "recorded_at": "2026-04-09T12:00:00.000000Z",
+                "payload": {},
+            }
+        ],
+        "commands": [],
+        "signals": [],
+        "updates": [],
+        "tasks": [],
+        "activities": [],
+        "timers": [],
+        "failures": [],
+        "links": {"projection_source": "rebuilt", "parents": [], "children": []},
+        "codec_schemas": {},
+        "payload_manifest": {"version": 1, "entries": []},
+        "redaction": {"applied": False, "policy": None, "paths": []},
+    }
+    cloned = {key: value for key, value in bundle.items() if key != "integrity"}
+    canonical = json.dumps(
+        _canonicalize(cloned),
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+    bundle["integrity"] = {
+        "canonicalization": "json-recursive-ksort-v1",
+        "checksum_algorithm": "sha256",
+        "checksum": hashlib.sha256(canonical.encode("utf-8")).hexdigest(),
+        "signature_algorithm": None,
+        "signature": None,
+        "key_id": None,
+    }
+    return bundle
+
+
+def _canonicalize(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {key: _canonicalize(value[key]) for key in sorted(value)}
+    if isinstance(value, list):
+        return [_canonicalize(item) for item in value]
+    return value