diff --git a/CHANGELOG.md b/CHANGELOG.md index 51d1d12a..4c3096de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 another thread. Adds `WorkerState` and `WorkerRoleStatus`, a runnable probe example and a Kubernetes guide. Existing worker behavior is unchanged. +### Fixed + +- History-bundle verification and bundle simulation now fail a + retention-pruned export (`workflow.details_state: "pruned"`) with the + `history.details_pruned` error, matching PHP Core, instead of reporting it + as `ok`. Retained bundles and older exports without the marker are + unchanged. + ## [2.12.4] - 2026-11-02 ### Added diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 6e6564c3..9825973c 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -709,6 +709,15 @@ Unicode escapes and finite floating-point notation. A producer that loses a memo's numeric kind during JSON export still produces a projection/integrity failure; the verifier does not repair or silently coerce those bytes. +A bundle exported after retention cleanup pruned the run's details carries +`workflow.details_state: "pruned"`, `history_complete: false` and no history +events. Starting with unreleased SDK source, verification fails it with the +`history.details_pruned` error, matching PHP Core: such a bundle has no history +to replay or import, even though its workflow status stays terminal. Bundles +marked `retained`, and older exports without the field, are unaffected. Pruned +exports produced before the marker existed cannot be recognized from the +bundle alone. + Verify the original exported bytes. Do not rewrite `schema_version`, remove integrity metadata or recompute a checksum to force acceptance. A clean bundle simulation still records `replay_checked_count: 0` and `replay_skipped: true`; diff --git a/src/durable_workflow/history_bundle_verify.py b/src/durable_workflow/history_bundle_verify.py index 54787f29..64fa8014 100644 --- a/src/durable_workflow/history_bundle_verify.py +++ b/src/durable_workflow/history_bundle_verify.py @@ -100,6 +100,7 @@ def verify_bundle(bundle: Any, signing_key: str | None = None) -> dict[str, Any] _check_envelope(bundle, findings) _check_workflow(bundle, findings) + _check_retained_details(bundle, findings) _check_history_events(bundle, findings) _check_commands(bundle, findings) _check_payload_manifest(bundle, findings) @@ -342,6 +343,37 @@ def _php_integer_key(key: str) -> bool: return -(2**63) <= int(key) <= 2**63 - 1 +def _check_retained_details(bundle: Mapping[str, Any], findings: list[dict[str, Any]]) -> None: + """Fail a run whose details retention cleanup pruned. + + Such a bundle has no history to replay or import, whatever its other + fields say. Bundles without ``workflow.details_state`` (older exports) and + bundles marked ``retained`` are unaffected. + """ + + workflow = bundle.get("workflow") + if not isinstance(workflow, Mapping) or workflow.get("details_state") != "pruned": + return + + pruned_at = _string_or_none(workflow.get("details_pruned_at")) + findings.append( + _finding( + "history.details_pruned", + SEVERITY_ERROR, + ( + "Retention cleanup pruned this run's details" + f"{'' if pruned_at is None else ' at ' + pruned_at}" + "; the bundle has no history to replay or import." + ), + { + "details_pruned_at": pruned_at, + "history_complete": bundle.get("history_complete"), + }, + path="workflow.details_state", + ) + ) + + def _check_history_events(bundle: Mapping[str, Any], findings: list[dict[str, Any]]) -> None: events = bundle.get("history_events") if not isinstance(events, list): diff --git a/tests/fixtures/history_bundles/retention_pruned/provenance.json b/tests/fixtures/history_bundles/retention_pruned/provenance.json new file mode 100644 index 00000000..b4cfdf9d --- /dev/null +++ b/tests/fixtures/history_bundles/retention_pruned/provenance.json @@ -0,0 +1,31 @@ +{ + "fixture": "pruned-bundle.json", + "file_sha256": "57f12ffb97c8d3fd60648cb70378f024a3a66c841cb222713da534cb552ddbbf", + "canonical_checksum": "8fa57d07aa2835e1210f3d5a840a2d46e63b4d918e15689dd8818f9c7b1101ac", + "producer": "PHP Core HistoryExport::forRun after WorkflowRunRetentionCleanup::pruneRun", + "producer_source": "durable-workflow/workflow PR #980 head 2e562997; merged as 05ffc5794f8d554c4fb10c29a62da60e1a9e946c with no file changes from that head", + "runtime": "PHP 8.4, PostgreSQL 17, Redis 7; a real TestGreetingWorkflow run completed (workflow, activity, workflow), was pruned, then exported", + "original_record": "https://github.com/durable-workflow/sdk-python/issues/228", + "php_corpus_scenario": { + "path": "tests/Fixtures/V2/ReplayRegression/retention-pruned-run-export.json", + "commit": "05ffc5794f8d554c4fb10c29a62da60e1a9e946c", + "sha256": "7016af05eaa0b23ea1733645da25298260cb6370f9d9a166e34678e4d0a24380", + "note": "The PHP corpus entry is a scenario descriptor that generates pruned exports at test time; it carries no bundle bytes. This file holds the exact bundle that scenario's exporter produced." + }, + "expected_php_verification": { + "status": "failed", + "findings": [ + { + "rule": "history.details_pruned", + "severity": "error", + "message": "Retention cleanup pruned this run's details at 2026-11-02T21:48:14.952676Z; the bundle has no history to replay or import.", + "path": "workflow.details_state", + "context": { + "details_pruned_at": "2026-11-02T21:48:14.952676Z", + "history_complete": false + } + } + ] + }, + "note": "Test data only. Never regenerate or normalize this file in place; tests mutate parsed copies." +} diff --git a/tests/fixtures/history_bundles/retention_pruned/pruned-bundle.json b/tests/fixtures/history_bundles/retention_pruned/pruned-bundle.json new file mode 100644 index 00000000..c5b49810 --- /dev/null +++ b/tests/fixtures/history_bundles/retention_pruned/pruned-bundle.json @@ -0,0 +1,147 @@ +{ + "schema": "durable-workflow.v2.history-export", + "schema_version": 2, + "exported_at": "2026-11-02T21:48:14.974261Z", + "dedupe_key": "01m6f3qvzpvz4jth4ag0aczfv8:6:2026-11-02T21:48:14.702328Z:details-pruned:2026-11-02T21:48:14.952676Z", + "history_complete": false, + "workflow": { + "instance_id": "pruned-parity-repro", + "run_id": "01m6f3qvzpvz4jth4ag0aczfv8", + "run_number": 1, + "is_current_run": true, + "current_run_id": "01m6f3qvzpvz4jth4ag0aczfv8", + "current_run_source": "run_order_fallback", + "workflow_type": "test-greeting-workflow", + "workflow_class": "Tests\\Fixtures\\V2\\TestGreetingWorkflow", + "source_runtime": "embedded", + "source_process_class": "application_process", + "business_key": null, + "visibility_labels": [], + "memo": [], + "memo_payload": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4A" + }, + "search_attributes": [], + "status": "completed", + "status_bucket": "completed", + "closed_reason": "completed", + "archived_at": null, + "archive_command_id": null, + "archive_reason": null, + "compatibility": null, + "connection": null, + "queue": "default", + "last_history_sequence": 6, + "started_at": "2026-11-02T21:48:10.415185Z", + "closed_at": "2026-11-02T21:48:14.696276Z", + "last_progress_at": "2026-11-02T21:48:14.702328Z", + "details_state": "pruned", + "details_pruned_at": "2026-11-02T21:48:14.952676Z" + }, + "payloads": { + "codec": "avro", + "arguments": { + "available": true, + "data": "wwHioz3/VYAiNwwCCgZBZGEA" + }, + "output": { + "available": true, + "codec": "avro", + "data": "wwHioz3/VYAiNw4GEGdyZWV0aW5nChZIZWxsbywgQWRhIRZ3b3JrZmxvd19pZAomcHJ1bmVkLXBhcml0eS1yZXBybwxydW5faWQKNDAxbTZmM3F2enB2ejRqdGg0YWcwYWN6ZnY4AA==" + } + }, + "summary": null, + "selected_run": { + "waits_projection_source": "workflow_run_waits", + "timeline_projection_source": "workflow_run_timeline_entries", + "timers_projection_source": "workflow_run_timer_entries", + "timers_projection_rebuild_reasons": [], + "lineage_projection_source": "workflow_run_lineage_entries" + }, + "history_events": [], + "waits": [], + "timeline": [], + "linked_intakes_scope": "selected_run", + "linked_intakes": [], + "commands": [], + "signals": [], + "updates": [], + "tasks": [], + "activities": [], + "timers": [], + "failures": [], + "links": { + "projection_source": "workflow_run_lineage_entries", + "parents": [], + "children": [] + }, + "redaction": { + "applied": false, + "policy": null, + "paths": [] + }, + "codec_schemas": { + "avro": { + "encoding": "base64-avro-single-object", + "framing": "single_object", + "magic_hex": "c301", + "current_fingerprint": "e2a33dff55802237", + "writer_schemas": { + "e2a33dff55802237": { + "schema": "{\"type\":\"record\",\"name\":\"Value\",\"namespace\":\"durable_workflow.protocol\",\"fields\":[{\"name\":\"value\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"BooleanValue\",\"fields\":[{\"name\":\"boolean\",\"type\":\"boolean\"}]},{\"type\":\"record\",\"name\":\"LongValue\",\"fields\":[{\"name\":\"long\",\"type\":\"long\"}]},{\"type\":\"record\",\"name\":\"DoubleValue\",\"fields\":[{\"name\":\"double\",\"type\":\"double\"}]},{\"type\":\"record\",\"name\":\"BytesValue\",\"fields\":[{\"name\":\"bytes\",\"type\":\"bytes\"}]},{\"type\":\"record\",\"name\":\"StringValue\",\"fields\":[{\"name\":\"string\",\"type\":\"string\"}]},{\"type\":\"record\",\"name\":\"ArrayValue\",\"fields\":[{\"name\":\"items\",\"type\":{\"type\":\"array\",\"items\":\"Value\"}}]},{\"type\":\"record\",\"name\":\"MapValue\",\"fields\":[{\"name\":\"entries\",\"type\":{\"type\":\"map\",\"values\":\"Value\"}}]}]}]}", + "fingerprint": "crc64-avro:e2a33dff55802237" + } + } + } + }, + "payload_manifest": { + "version": 1, + "entries": [ + { + "path": "workflow.memo_payload", + "codec": "avro", + "available": true, + "redacted": false, + "encoding": "base64-avro-single-object", + "avro_framing": "single_object", + "avro_prefix_hex": "c301", + "writer_schema": "{\"type\":\"record\",\"name\":\"Value\",\"namespace\":\"durable_workflow.protocol\",\"fields\":[{\"name\":\"value\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"BooleanValue\",\"fields\":[{\"name\":\"boolean\",\"type\":\"boolean\"}]},{\"type\":\"record\",\"name\":\"LongValue\",\"fields\":[{\"name\":\"long\",\"type\":\"long\"}]},{\"type\":\"record\",\"name\":\"DoubleValue\",\"fields\":[{\"name\":\"double\",\"type\":\"double\"}]},{\"type\":\"record\",\"name\":\"BytesValue\",\"fields\":[{\"name\":\"bytes\",\"type\":\"bytes\"}]},{\"type\":\"record\",\"name\":\"StringValue\",\"fields\":[{\"name\":\"string\",\"type\":\"string\"}]},{\"type\":\"record\",\"name\":\"ArrayValue\",\"fields\":[{\"name\":\"items\",\"type\":{\"type\":\"array\",\"items\":\"Value\"}}]},{\"type\":\"record\",\"name\":\"MapValue\",\"fields\":[{\"name\":\"entries\",\"type\":{\"type\":\"map\",\"values\":\"Value\"}}]}]}]}", + "writer_schema_fingerprint": "crc64-avro:e2a33dff55802237", + "diagnostic": null + }, + { + "path": "payloads.arguments.data", + "codec": "avro", + "available": true, + "redacted": false, + "encoding": "base64-avro-single-object", + "avro_framing": "single_object", + "avro_prefix_hex": "c301", + "writer_schema": "{\"type\":\"record\",\"name\":\"Value\",\"namespace\":\"durable_workflow.protocol\",\"fields\":[{\"name\":\"value\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"BooleanValue\",\"fields\":[{\"name\":\"boolean\",\"type\":\"boolean\"}]},{\"type\":\"record\",\"name\":\"LongValue\",\"fields\":[{\"name\":\"long\",\"type\":\"long\"}]},{\"type\":\"record\",\"name\":\"DoubleValue\",\"fields\":[{\"name\":\"double\",\"type\":\"double\"}]},{\"type\":\"record\",\"name\":\"BytesValue\",\"fields\":[{\"name\":\"bytes\",\"type\":\"bytes\"}]},{\"type\":\"record\",\"name\":\"StringValue\",\"fields\":[{\"name\":\"string\",\"type\":\"string\"}]},{\"type\":\"record\",\"name\":\"ArrayValue\",\"fields\":[{\"name\":\"items\",\"type\":{\"type\":\"array\",\"items\":\"Value\"}}]},{\"type\":\"record\",\"name\":\"MapValue\",\"fields\":[{\"name\":\"entries\",\"type\":{\"type\":\"map\",\"values\":\"Value\"}}]}]}]}", + "writer_schema_fingerprint": "crc64-avro:e2a33dff55802237", + "diagnostic": null + }, + { + "path": "payloads.output.data", + "codec": "avro", + "available": true, + "redacted": false, + "encoding": "base64-avro-single-object", + "avro_framing": "single_object", + "avro_prefix_hex": "c301", + "writer_schema": "{\"type\":\"record\",\"name\":\"Value\",\"namespace\":\"durable_workflow.protocol\",\"fields\":[{\"name\":\"value\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"BooleanValue\",\"fields\":[{\"name\":\"boolean\",\"type\":\"boolean\"}]},{\"type\":\"record\",\"name\":\"LongValue\",\"fields\":[{\"name\":\"long\",\"type\":\"long\"}]},{\"type\":\"record\",\"name\":\"DoubleValue\",\"fields\":[{\"name\":\"double\",\"type\":\"double\"}]},{\"type\":\"record\",\"name\":\"BytesValue\",\"fields\":[{\"name\":\"bytes\",\"type\":\"bytes\"}]},{\"type\":\"record\",\"name\":\"StringValue\",\"fields\":[{\"name\":\"string\",\"type\":\"string\"}]},{\"type\":\"record\",\"name\":\"ArrayValue\",\"fields\":[{\"name\":\"items\",\"type\":{\"type\":\"array\",\"items\":\"Value\"}}]},{\"type\":\"record\",\"name\":\"MapValue\",\"fields\":[{\"name\":\"entries\",\"type\":{\"type\":\"map\",\"values\":\"Value\"}}]}]}]}", + "writer_schema_fingerprint": "crc64-avro:e2a33dff55802237", + "diagnostic": null + } + ] + }, + "integrity": { + "canonicalization": "json-recursive-ksort-v1", + "checksum_algorithm": "sha256", + "checksum": "8fa57d07aa2835e1210f3d5a840a2d46e63b4d918e15689dd8818f9c7b1101ac", + "signature_algorithm": null, + "signature": null, + "key_id": null + } +} diff --git a/tests/test_history_bundle_pruned.py b/tests/test_history_bundle_pruned.py new file mode 100644 index 00000000..ca6ab398 --- /dev/null +++ b/tests/test_history_bundle_pruned.py @@ -0,0 +1,132 @@ +"""Retention-pruned history bundles fail verification, matching PHP (#228).""" + +from __future__ import annotations + +import hashlib +import json +import shutil +from pathlib import Path +from typing import Any + +import pytest + +from durable_workflow.history_bundle_verify import main, verify_bundle, verify_bundle_json +from durable_workflow.replay_verify import main as replay_main +from durable_workflow.replay_verify import simulate_bundles + +FIXTURES = Path(__file__).parent / "fixtures" / "history_bundles" / "retention_pruned" +BUNDLE = FIXTURES / "pruned-bundle.json" + + +def provenance() -> dict[str, Any]: + return json.loads((FIXTURES / "provenance.json").read_text()) + + +def bundle() -> dict[str, Any]: + return json.loads(BUNDLE.read_text()) + + +def rechecksum(value: dict[str, Any]) -> dict[str, Any]: + body = {key: item for key, item in value.items() if key != "integrity"} + canonical = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + value["integrity"] = { + "canonicalization": "json-recursive-ksort-v1", + "checksum_algorithm": "sha256", + "checksum": hashlib.sha256(canonical).hexdigest(), + } + return value + + +def rules(report: dict[str, Any]) -> list[str]: + return [finding["rule"] for finding in report["findings"]] + + +def test_fixture_is_the_untouched_php_export() -> None: + record = provenance() + assert hashlib.sha256(BUNDLE.read_bytes()).hexdigest() == record["file_sha256"] + + +def test_pruned_php_export_fails_exactly_like_php() -> None: + record = provenance() + + report = verify_bundle_json(BUNDLE.read_text()) + + assert report["status"] == record["expected_php_verification"]["status"] == "failed" + assert report["findings"] == record["expected_php_verification"]["findings"] + assert report["integrity"]["checksum_matches"] is True + assert report["integrity"]["recomputed_checksum"] == record["canonical_checksum"] + assert report["summary"]["errors"] == 1 + + +def test_pruned_dedupe_key_is_accepted_as_exported() -> None: + value = bundle() + assert ":details-pruned:" in value["dedupe_key"] + + report = verify_bundle(value) + + assert rules(report) == ["history.details_pruned"] + + +@pytest.mark.parametrize("pruned_at", [None, "", 7, "absent"]) +def test_pruned_marker_without_a_timestamp_still_fails(pruned_at: object) -> None: + value = bundle() + if pruned_at == "absent": + del value["workflow"]["details_pruned_at"] + else: + value["workflow"]["details_pruned_at"] = pruned_at + + report = verify_bundle(rechecksum(value)) + + assert report["status"] == "failed" + assert report["integrity"]["checksum_matches"] is True + [finding] = report["findings"] + assert finding["rule"] == "history.details_pruned" + assert finding["message"] == ( + "Retention cleanup pruned this run's details; the bundle has no history to replay or import." + ) + assert finding["context"] == {"details_pruned_at": None, "history_complete": False} + + +@pytest.mark.parametrize("state", ["retained", "absent"]) +def test_retained_and_older_exports_are_unchanged(state: str) -> None: + value = bundle() + value["workflow"]["details_pruned_at"] = None + if state == "absent": + del value["workflow"]["details_state"] + del value["workflow"]["details_pruned_at"] + else: + value["workflow"]["details_state"] = state + + report = verify_bundle(rechecksum(value)) + + assert "history.details_pruned" not in rules(report) + assert report["status"] == "ok", report["findings"] + + +@pytest.mark.parametrize("state", ["PRUNED", "pruned ", True, None]) +def test_only_the_exact_pruned_marker_fails(state: object) -> None: + value = bundle() + value["workflow"]["details_state"] = state + + report = verify_bundle(rechecksum(value)) + + assert "history.details_pruned" not in rules(report) + + +def test_simulation_and_both_cli_entry_points_block_a_pruned_bundle(tmp_path: Path) -> None: + directory = tmp_path / "bundles" + directory.mkdir() + shutil.copyfile(BUNDLE, directory / "pruned-bundle.json") + + simulation = simulate_bundles(directory).to_dict() + assert simulation["verdict"] == "failed" + assert simulation["promotion_decision"] == "block_and_investigate" + [entry] = simulation["bundles"] + assert entry["evidence"]["integrity_status"] == "failed" + + single = tmp_path / "single.json" + replayed = tmp_path / "simulation.json" + assert main([str(BUNDLE), "--output", str(single)]) != 0 + assert replay_main([str(directory), "--simulate-bundles", "--output", str(replayed)]) != 0 + assert rules(json.loads(single.read_text())) == ["history.details_pruned"] + assert json.loads(replayed.read_text())["verdict"] == "failed"