Validate typed original-row snapshots without asserting phase completion

tests/fixtures/worker_startup_reconstruction/sql-proposal/README.md

@@ -17,25 +17,58 @@ namespace scope uses the supplied namespace directly.
 
 The row cap is two to expose an unexpected duplicate, not permission to use the
 first result. Zero rows means missing; two rows or any inconsistent identity,
-type or field means invalid. Exact column/identity validation, driver-specific
-bound parameters and explicit complete/missing/truncated/invalid records remain
-unimplemented. Do not interpolate parameters or execute these texts directly
+type or field means invalid. The pure `tests/worker_startup_snapshot.py` component
+now validates supplied normalized row projections against the closed
+column/identity/type contract. Driver-specific bound parameters, normalization
+and actual collection remain unimplemented. Do not interpolate parameters or execute these texts directly
 from this proposal. The field `lease_owner_present` says only that the column
 is non-null. Error-present booleans also mean non-null, including an empty
 string; they do not establish an error's contents, severity or cause.
 
 Each returned row includes that query's database UTC sample. Zero rows also
 means no clock sample from that query: mark it missing rather than borrowing the
 other query's timestamp. The future collector must bracket each read with
-observer monotonic times, retain the session timezone used to
-decode timestamp columns, and keep before/after-teardown reads distinct.
+observer monotonic times and keep before/after-teardown reads distinct. Returned
+rows now also report `@@session.time_zone` and `@@system_time_zone`, without
+changing either variable. These labels retain the timestamp-decoding context
+described in the [MySQL timezone reference](https://dev.mysql.com/doc/refman/8.0/en/time-zone-support.html).
 `UTC_TIMESTAMP(6)` does not make those views atomic or establish a shared clock.
 The row cap does not bound database work or wall time, particularly with binary
 casts: the future fixture-only owner must enforce the accepted five-second
 snapshot budget inside the unchanged case window, cancel/close overdue reads,
 and retain the ten-second overall cleanup boundary. No schema/index change,
 timeout extension, lock, transaction or worker mutation is proposed here.
 
+## Pure normalized-row validation
+
+The decoder performs no SQL, driver/process/credential access, raw-output capture,
+clock collection or timezone conversion. Its input is an explicitly supplied
+source state plus a list containing zero or one exact row dictionary. Empty input
+is missing, multiple rows are invalid, and missing/truncated/invalid upstream
+states cannot be promoted by an otherwise valid-looking row. The future collector
+must independently prove its source-state declaration and normalization; this
+component cannot certify either. Capture size/time/cancellation bounds remain
+pending with that collector.
+
+The fixture contract requires printable ASCII labels and original IDs, exact
+original namespace/run/task or instance matching, the historical status vocabulary
+and workflow task family. Counters accept unsigned 32-bit integers or canonical
+decimal strings, preserving zero. Presence fields accept only integer or textual
+`0`/`1`; native bool, float, null and generic truthiness are refused. Nullable text
+preserves null, empty and literal `NULL` distinctly. Timestamp normalization must
+supply the declared six-fractional-digit string form or null; native datetime
+objects and implicit string conversions are refused until the driver's
+normalization is reviewed.
+
+Every required column must exist, unknown columns are refused, and failed
+projections expose only a fixed code and optional known field name. No raw row,
+unknown column/value or exception text is echoed. Successful data is a fresh
+primitive-valued projection. `projection_state=complete` means this supplied
+projection has the expected fields/types/identity; `phase_verdict` stays
+`unknown`, including for valid-looking but inconsistent lease/attempt states.
+It proves no raw capture, shared clock, row consistency, claim, hydration,
+publication or causal conclusion.
+
 Private poll/claim/lease-owner binding, registration routing/current-incarnation
 and freshness evidence, API/history linkage, the snapshot collector and terminal
 classification are intentionally still missing. They must take the explicit

tests/fixtures/worker_startup_reconstruction/sql-proposal/original-run.sql

@@ -2,6 +2,8 @@
 -- This is an unexecuted proposal, not an installed comparator or driver.
 SELECT
     UTC_TIMESTAMP(6) AS database_time_utc,
+    @@session.time_zone AS session_time_zone,
+    @@system_time_zone AS system_time_zone,
     r.id AS original_run_id,
     r.workflow_instance_id,
     r.namespace,

tests/fixtures/worker_startup_reconstruction/sql-proposal/original-task.sql

@@ -1,7 +1,9 @@
 -- Unexecuted MySQL 8.0 source proposal; fixture namespace and original IDs only.
--- Parameter binding, query deadline and row validation are not implemented here.
+-- Parameter binding, query deadline and actual row collection remain unimplemented.
 SELECT
     UTC_TIMESTAMP(6) AS database_time_utc,
+    @@session.time_zone AS session_time_zone,
+    @@system_time_zone AS system_time_zone,
     t.id AS original_task_id,
     t.workflow_run_id AS original_run_id,
     t.namespace,

tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json

@@ -2,7 +2,7 @@
   "accepted_contract": "https://github.com/durable-workflow/sdk-python/blob/78ff7a1857961f92e4b9ac277b83ca60ca53e3b7/tests/fixtures/worker_startup_reconstruction/proposal.json",
   "actual_queries_executed": 0,
   "collector_source_selected": null,
-  "database_dialect": "MySQL8.0, proposed only; no SQL engine/parser execution",
+  "database_dialect": "MySQL8.0 proposed SQL only; pure normalized-row controls are not SQL execution",
   "execution_admitted": false,
   "historical_schema_sources": [
     {
@@ -28,9 +28,45 @@
       "path": "app/Support/NamespaceWorkflowScope.php",
       "repository": "durable-workflow/server",
       "sha256": "3a093652cc41ed4ecb4d8454c38ef1abb1086159698e77a685a55e89cc0f6b84"
+    },
+    {
+      "blob": "17b3a5b1505765b431b448f9fb46ccf9f45718eb",
+      "bytes": 179,
+      "commit": "19a7a1cf33604c66f1f838e8b3db22563041f6b2",
+      "path": "src/V2/Enums/TaskType.php",
+      "repository": "durable-workflow/workflow",
+      "sha256": "979c36c386433a5cc6180e7db183f13dd6cb953e509adbf25d42ffefaeb350c5"
+    },
+    {
+      "blob": "e9d33f72d8ae0b575763e2702cb25c191bc989f7",
+      "bytes": 241,
+      "commit": "19a7a1cf33604c66f1f838e8b3db22563041f6b2",
+      "path": "src/V2/Enums/TaskStatus.php",
+      "repository": "durable-workflow/workflow",
+      "sha256": "a3ca4b86cd224814ab375cbd50b5a9846329d8e57ff934b3d9f74fca92ee882b"
+    },
+    {
+      "blob": "87822f31949dfabf9c46d4d45c23079360abd215",
+      "bytes": 813,
+      "commit": "19a7a1cf33604c66f1f838e8b3db22563041f6b2",
+      "path": "src/V2/Enums/RunStatus.php",
+      "repository": "durable-workflow/workflow",
+      "sha256": "008fc2e5f2f1c1c668d9ab57f10a9bd77781b3989b6657925cc45c4d9112fb2a"
     }
   ],
   "issue": "https://github.com/durable-workflow/sdk-python/issues/123",
+  "mysql_reference_documents": [
+    "https://dev.mysql.com/doc/refman/8.0/en/time-zone-support.html"
+  ],
+  "normalized_row_decoder": {
+    "bytes": 7588,
+    "capture_size_time_and_cancellation_enforced": false,
+    "input": "explicit upstream state and exact normalized row dictionaries; no raw driver/transport",
+    "normalization_proven": false,
+    "path": "tests/worker_startup_snapshot.py",
+    "phase_verdict": "unknown",
+    "sha256": "dd02a5dae1fe2a777a802efbc431844d3e9ca8c646303776db9f7e134640c0e8"
+  },
   "parameter_driver_selected": null,
   "private_binding_comparator_selected": null,
   "registration_comparator_selected": null,
@@ -39,14 +75,14 @@
   "source_review_owner": "server",
   "sql_files": [
     {
-      "bytes": 585,
+      "bytes": 675,
       "file": "original-run.sql",
-      "sha256": "46f291d0cd698f693845aa999139342fb2b0e314b7eaf5aa1cd68231fe40382f"
+      "sha256": "025db37ab5311650d694f139941e206c2ae085d9b862426efa409a53ddd2ae13"
     },
     {
-      "bytes": 1047,
+      "bytes": 1140,
       "file": "original-task.sql",
-      "sha256": "41284331ea4ad06904cbe4312afc9aec5eff789e560e7910c50fe48397202725"
+      "sha256": "4385b4d0f08e22be9f053f282283111282f7215d12e9247d03ca0642e3e2b825"
     }
   ],
   "status": "unexecuted-source-proposal-not-a-complete-comparator"

tests/test_worker_startup_snapshot.py

@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+import datetime as dt
+import json
+import re
+from dataclasses import asdict, replace
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tests.worker_startup_snapshot import OriginalIdentity, decode_snapshot
+
+IDENTITY = OriginalIdentity(
+    "default", "task-family-original", "01BX5ZZKBKACTAV9WEVGEMMVRZ", "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+)
+NOW = "2026-11-05 20:00:00.123456"
+BITS = ("lease_owner_present", "dispatch_error_present", "claim_error_present", "task_error_present")
+COUNTERS = ("attempt_count", "repair_count")
+
+
+def row(kind: str) -> dict[str, Any]:
+    value: dict[str, Any] = {
+        "database_time_utc": NOW,
+        "session_time_zone": "SYSTEM",
+        "system_time_zone": "UTC",
+        "original_run_id": IDENTITY.original_run_id,
+        "namespace": IDENTITY.namespace,
+        "status": "ready" if kind == "task" else "running",
+    }
+    if kind == "task":
+        value.update(
+            {
+                "original_task_id": IDENTITY.original_task_id,
+                "task_type": "workflow",
+                "queue": "task-family-original",
+                "compatibility": None,
+                "available_at": None,
+                "leased_at": None,
+                "lease_expires_at": None,
+                "last_dispatch_attempt_at": None,
+                "last_dispatched_at": None,
+                "last_claim_failed_at": None,
+                "repair_available_at": None,
+                "attempt_count": 0,
+                "repair_count": 0,
+                "lease_owner_present": "0",
+                "dispatch_error_present": "0",
+                "claim_error_present": "0",
+                "task_error_present": "0",
+            }
+        )
+    else:
+        value["workflow_instance_id"] = IDENTITY.workflow_instance_id
+        value["execution_deadline_at"] = None
+        value["run_deadline_at"] = None
+    return value
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_exact_original_projection_is_not_a_phase_verdict(kind: Any) -> None:
+    supplied = row(kind)
+    result = decode_snapshot(kind, IDENTITY, [supplied])
+    assert result.projection_state == "complete" and result.phase_verdict == "unknown"
+    assert result.row is not supplied and result.row is not None
+    assert set(result.row) == set(supplied)
+    assert result.row["database_time_utc"] == NOW and result.row["session_time_zone"] == "SYSTEM"
+    supplied["status"] = "SECRET changed later"
+    assert result.row["status"] != supplied["status"]
+
+
+@pytest.mark.parametrize("value", [0, "0", 1, "1"])
+def test_presence_bits_have_exact_numeric_semantics(value: Any) -> None:
+    supplied = row("task")
+    supplied.update(dict.fromkeys(BITS, value))
+    result = decode_snapshot("task", IDENTITY, [supplied])
+    assert result.projection_state == "complete" and result.row is not None
+    for field in BITS:
+        assert result.row[field] is (value in (1, "1"))
+    assert result.row["attempt_count"] == 0
+
+
+@pytest.mark.parametrize("value", [None, False, True, "false", "true", "", "NULL", 0.0, 2, "01", [], {}])
+def test_truthiness_does_not_admit_unproved_presence_bits(value: Any) -> None:
+    supplied = row("task")
+    supplied["lease_owner_present"] = value
+    result = decode_snapshot("task", IDENTITY, [supplied])
+    assert result.projection_state == "invalid" and result.refusal_code == "PRESENCE_BIT_INVALID"
+    assert result.row is None and result.phase_verdict == "unknown"
+
+
+@pytest.mark.parametrize("value", [0, "0", 4_294_967_295, "4294967295"])
+def test_unsigned_counters_preserve_zero_and_the_historical_schema_limit(value: Any) -> None:
+    supplied = row("task")
+    supplied.update(dict.fromkeys(COUNTERS, value))
+    result = decode_snapshot("task", IDENTITY, [supplied])
+    assert result.projection_state == "complete" and result.row is not None
+    assert result.row["attempt_count"] == int(value) and type(result.row["attempt_count"]) is int
+
+
+@pytest.mark.parametrize("value", [None, False, True, -1, 4_294_967_296, "4294967296", "00", " 1", "1.0", 1.0, "-1"])
+def test_counter_coercion_cannot_create_an_attempt_or_repair_count(value: Any) -> None:
+    supplied = row("task")
+    supplied["attempt_count"] = value
+    result = decode_snapshot("task", IDENTITY, [supplied])
+    assert result.projection_state == "invalid" and result.refusal_code == "COUNTER_INVALID"
+
+
+@pytest.mark.parametrize(
+    "kind,field",
+    [
+        ("task", "namespace"),
+        ("task", "original_run_id"),
+        ("task", "original_task_id"),
+        ("run", "namespace"),
+        ("run", "original_run_id"),
+        ("run", "workflow_instance_id"),
+    ],
+)
+def test_latest_replacement_or_case_changed_identity_is_refused(kind: Any, field: str) -> None:
+    supplied = row(kind)
+    supplied[field] = str(supplied[field]).swapcase()
+    result = decode_snapshot(kind, IDENTITY, [supplied])
+    assert result.projection_state == "invalid" and result.refusal_code == "IDENTITY_MISMATCH"
+    assert result.row is None
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_missing_duplicate_and_extra_rows_cannot_become_one_original_row(kind: Any) -> None:
+    assert decode_snapshot(kind, IDENTITY, []).projection_state == "missing"
+    assert decode_snapshot(kind, IDENTITY, []).row is None  # No borrowed DB clock sample.
+    for rows in ([row(kind), row(kind)], (row(kind),), {"row": row(kind)}, "SECRET", [None]):
+        result = decode_snapshot(kind, IDENTITY, rows)
+        assert result.projection_state == "invalid" and result.row is None
+
+
+@pytest.mark.parametrize("state", ["missing", "truncated", "invalid"])
+def test_incomplete_transport_state_never_promotes_a_valid_row(state: Any) -> None:
+    result = decode_snapshot("task", IDENTITY, [row("task")], source_state=state)
+    assert result.projection_state == state and result.row is None and result.phase_verdict == "unknown"
+
+
+@pytest.mark.parametrize("kind,field", [(kind, field) for kind in ("task", "run") for field in sorted(row(kind))])
+def test_every_proposed_projection_field_is_required(kind: Any, field: str) -> None:
+    supplied = row(kind)
+    del supplied[field]
+    assert decode_snapshot(kind, IDENTITY, [supplied]).refusal_code == "FIELD_SET_INVALID"
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_independent_fixtures_match_the_sql_column_inventory(kind: str) -> None:
+    source = (
+        Path(__file__).parent / "fixtures/worker_startup_reconstruction/sql-proposal" / f"original-{kind}.sql"
+    ).read_text()
+    selected = source.split("SELECT", 1)[1].split("FROM", 1)[0]
+    names = []
+    for expression in selected.split(","):
+        match = re.search(r"(?:\bAS\s+([a-z_]+)|\b[tr]\.([a-z_]+))\s*$", expression)
+        assert match is not None, "lexical field inventory changed; this is not SQL syntax validation"
+        names.append(next(value for value in match.groups() if value is not None))
+    assert len(names) == len(set(names)) and set(names) == set(row(kind))
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_unknown_payload_or_error_columns_are_not_copied_or_echoed(kind: Any) -> None:
+    supplied = row(kind)
+    supplied["SECRET unknown payload"] = "SECRET"
+    result = decode_snapshot(kind, IDENTITY, [supplied])
+    assert result.projection_state == "invalid" and result.field is None and result.row is None
+    assert "SECRET" not in json.dumps(asdict(result))
+
+
+@pytest.mark.parametrize("value", [None, "", "NULL"])
+def test_nullable_text_keeps_null_empty_and_literal_null_distinct(value: Any) -> None:
+    supplied = row("task")
+    supplied["compatibility"] = value
+    result = decode_snapshot("task", IDENTITY, [supplied])
+    assert result.projection_state == "complete" and result.row is not None
+    assert result.row["compatibility"] == value and type(result.row["compatibility"]) is type(value)
+
+
+@pytest.mark.parametrize(
+    "value", ["", "NULL", "2026-02-30 20:00:00.123456", "2026-11-05T20:00:00Z", dt.datetime(2026, 11, 5)]
+)
+def test_timestamp_normalization_must_be_proven_by_a_future_driver(value: Any) -> None:
+    supplied = row("task")
+    supplied["leased_at"] = value
+    assert decode_snapshot("task", IDENTITY, [supplied]).refusal_code == "TIMESTAMP_INVALID"
+
+
+@pytest.mark.parametrize("field", ["database_time_utc", "session_time_zone", "system_time_zone"])
+def test_clock_and_timezone_evidence_cannot_be_omitted(field: str) -> None:
+    supplied = row("run")
+    supplied[field] = None
+    result = decode_snapshot("run", IDENTITY, [supplied])
+    assert result.projection_state == "invalid" and result.field == field
+
+
+@pytest.mark.parametrize(
+    "kind,statuses",
+    [
+        ("task", {"ready", "leased", "cancelled", "completed", "failed"}),
+        ("run", {"pending", "running", "waiting", "cancelled", "terminated", "completed", "failed"}),
+    ],
+)
+def test_historical_status_vocabulary_is_retained_without_causal_interpretation(kind: Any, statuses: set[str]) -> None:
+    for status in statuses:
+        supplied = row(kind)
+        supplied["status"] = status
+        result = decode_snapshot(kind, IDENTITY, [supplied])
+        assert result.projection_state == "complete" and result.phase_verdict == "unknown"
+    supplied = row(kind)
+    supplied["status"] = "future_unknown_status"
+    assert decode_snapshot(kind, IDENTITY, [supplied]).refusal_code == "STATUS_INVALID"
+
+
+@pytest.mark.parametrize("family", ["activity", "timer", ""])
+def test_other_task_families_do_not_substitute_for_the_original_workflow_task(family: str) -> None:
+    supplied = row("task")
+    supplied["task_type"] = family
+    assert decode_snapshot("task", IDENTITY, [supplied]).refusal_code == "WRONG_TASK_FAMILY"
+
+
+@pytest.mark.parametrize(
+    "identity", [replace(IDENTITY, namespace=""), replace(IDENTITY, original_task_id="latest"), None]
+)
+def test_invalid_expected_identity_is_not_a_missing_original_row(identity: Any) -> None:
+    assert decode_snapshot("task", identity, []).refusal_code == "IDENTITY_INVALID"
+
+
+def test_missing_input_and_unknown_source_flags_are_explicit() -> None:
+    assert decode_snapshot("task", IDENTITY, None).projection_state == "missing"
+    assert (
+        decode_snapshot("task", IDENTITY, [row("task")], source_state="unknown").refusal_code == "SOURCE_STATE_INVALID"
+    )