From 48cf0b576deb922f0f94e72c5115074da8103b82 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 20:23:33 +0000 Subject: [PATCH 1/2] Validate typed original-row snapshots without asserting phase completion --- .../sql-proposal/README.md | 43 +++- .../sql-proposal/original-run.sql | 2 + .../sql-proposal/original-task.sql | 4 +- .../sql-proposal/sources.json | 46 +++- tests/test_worker_startup_snapshot.py | 235 ++++++++++++++++++ tests/worker_startup_snapshot.py | 200 +++++++++++++++ 6 files changed, 519 insertions(+), 11 deletions(-) create mode 100644 tests/test_worker_startup_snapshot.py create mode 100644 tests/worker_startup_snapshot.py diff --git a/tests/fixtures/worker_startup_reconstruction/sql-proposal/README.md b/tests/fixtures/worker_startup_reconstruction/sql-proposal/README.md index f715513..541c862 100644 --- a/tests/fixtures/worker_startup_reconstruction/sql-proposal/README.md +++ b/tests/fixtures/worker_startup_reconstruction/sql-proposal/README.md @@ -17,9 +17,10 @@ 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. @@ -27,8 +28,10 @@ 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 @@ -36,6 +39,36 @@ 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 diff --git a/tests/fixtures/worker_startup_reconstruction/sql-proposal/original-run.sql b/tests/fixtures/worker_startup_reconstruction/sql-proposal/original-run.sql index 14982a6..5717212 100644 --- a/tests/fixtures/worker_startup_reconstruction/sql-proposal/original-run.sql +++ b/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, diff --git a/tests/fixtures/worker_startup_reconstruction/sql-proposal/original-task.sql b/tests/fixtures/worker_startup_reconstruction/sql-proposal/original-task.sql index dcf2e55..5a05859 100644 --- a/tests/fixtures/worker_startup_reconstruction/sql-proposal/original-task.sql +++ b/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, diff --git a/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json b/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json index 2e8d35a..1d8cdb5 100644 --- a/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json +++ b/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": 7423, + "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": "6adc599d9025d1f8e52249b45be486f0e60c8e7e82168c62719c9750adcdb20c" + }, "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" diff --git a/tests/test_worker_startup_snapshot.py b/tests/test_worker_startup_snapshot.py new file mode 100644 index 0000000..209c9d3 --- /dev/null +++ b/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" + ) diff --git a/tests/worker_startup_snapshot.py b/tests/worker_startup_snapshot.py new file mode 100644 index 0000000..a663f9d --- /dev/null +++ b/tests/worker_startup_snapshot.py @@ -0,0 +1,200 @@ +"""Pure validation of proposed normalized, fixture-only historical SQL rows. + +No query, driver, credential, transport, capture bound or clock conversion is +implemented. Complete projection never establishes a claim or phase verdict. +""" + +from __future__ import annotations + +import datetime as dt +import re +from dataclasses import dataclass +from typing import Any, Literal, cast + +TASK_STATUSES = {"ready", "leased", "cancelled", "completed", "failed"} +RUN_STATUSES = {"pending", "running", "waiting", "cancelled", "terminated", "completed", "failed"} +COMMON_FIELDS = {"database_time_utc", "session_time_zone", "system_time_zone", "original_run_id", "namespace", "status"} +TASK_TIMES = { + "available_at", + "leased_at", + "lease_expires_at", + "last_dispatch_attempt_at", + "last_dispatched_at", + "last_claim_failed_at", + "repair_available_at", +} +PRESENCE_FIELDS = {"lease_owner_present", "dispatch_error_present", "claim_error_present", "task_error_present"} +COUNTER_FIELDS = {"attempt_count", "repair_count"} +TASK_FIELDS = ( + COMMON_FIELDS + | TASK_TIMES + | PRESENCE_FIELDS + | COUNTER_FIELDS + | { + "original_task_id", + "task_type", + "queue", + "compatibility", + } +) +RUN_TIMES = {"execution_deadline_at", "run_deadline_at"} +RUN_FIELDS = COMMON_FIELDS | RUN_TIMES | {"workflow_instance_id"} +State = Literal["complete", "missing", "truncated", "invalid"] + + +@dataclass(frozen=True) +class OriginalIdentity: + namespace: str + workflow_instance_id: str + original_run_id: str + original_task_id: str + + +@dataclass(frozen=True) +class SnapshotProjection: + projection_state: State + refusal_code: str | None = None + field: str | None = None + row: dict[str, Any] | None = None + phase_verdict: Literal["unknown"] = "unknown" + + +class ProjectionRefusal(ValueError): + def __init__(self, code: str, field: str | None = None) -> None: + super().__init__(code) + self.code = code + self.field = field + + +def require(value: bool, code: str, field: str | None = None) -> None: + if not value: + raise ProjectionRefusal(code, field) + + +def text(value: Any, limit: int) -> bool: + return type(value) is str and len(value) <= limit and all(32 <= ord(char) < 127 for char in value) + + +def identity_fields(kind: str, identity: OriginalIdentity) -> dict[str, str]: + require(type(kind) is str and kind in {"task", "run"}, "UNKNOWN_PROJECTION") + require(type(identity) is OriginalIdentity, "IDENTITY_INVALID") + require( + text(identity.namespace, 128) + and bool(identity.namespace) + and text(identity.workflow_instance_id, 191) + and bool(identity.workflow_instance_id), + "IDENTITY_INVALID", + ) + for value in (identity.original_run_id, identity.original_task_id): + require( + type(value) is str + and len(value) == 26 + and re.fullmatch(r"[0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{26}", value) is not None, + "IDENTITY_INVALID", + ) + return { + "namespace": identity.namespace, + "original_run_id": identity.original_run_id, + ("original_task_id" if kind == "task" else "workflow_instance_id"): identity.original_task_id + if kind == "task" + else identity.workflow_instance_id, + } + + +def timestamp(value: Any, field: str, *, nullable: bool = True) -> str | None: + if value is None and nullable: + return None + require( + type(value) is str + and len(value) == 26 + and re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{6}", value) is not None, + "TIMESTAMP_INVALID", + field, + ) + try: + dt.datetime.strptime(value, "%Y-%m-%d %H:%M:%S.%f") + except ValueError as error: + raise ProjectionRefusal("TIMESTAMP_INVALID", field) from error + return cast(str, value) + + +def counter(value: Any, field: str) -> int: + if type(value) is str: + require(len(value) <= 10 and re.fullmatch(r"0|[1-9][0-9]{0,9}", value) is not None, "COUNTER_INVALID", field) + value = int(value) + require(type(value) is int and 0 <= value <= 4_294_967_295, "COUNTER_INVALID", field) + return cast(int, value) + + +def presence(value: Any, field: str) -> bool: + require( + (type(value) is int and value in (0, 1)) or (type(value) is str and value in ("0", "1")), + "PRESENCE_BIT_INVALID", + field, + ) + return value in (1, "1") + + +def typed_row(kind: str, row: dict[str, Any], expected: dict[str, str]) -> dict[str, Any]: + for field, value in expected.items(): + require(type(row[field]) is str and row[field] == value, "IDENTITY_MISMATCH", field) + for field in ("session_time_zone", "system_time_zone"): + require(text(row[field], 64) and bool(row[field]), "TIMEZONE_LABEL_INVALID", field) + require( + type(row["status"]) is str and row["status"] in (TASK_STATUSES if kind == "task" else RUN_STATUSES), + "STATUS_INVALID", + "status", + ) + result = {field: row[field] for field in expected} + result.update({field: row[field] for field in ("status", "session_time_zone", "system_time_zone")}) + result["database_time_utc"] = timestamp(row["database_time_utc"], "database_time_utc", nullable=False) + if kind == "task": + require(type(row["task_type"]) is str and row["task_type"] == "workflow", "WRONG_TASK_FAMILY", "task_type") + result["task_type"] = row["task_type"] + for field in ("queue", "compatibility"): + require(row[field] is None or text(row[field], 255), "TEXT_INVALID", field) + result[field] = row[field] + for field in COUNTER_FIELDS: + result[field] = counter(row[field], field) + for field in PRESENCE_FIELDS: + result[field] = presence(row[field], field) + for field in TASK_TIMES if kind == "task" else RUN_TIMES: + result[field] = timestamp(row[field], field) + return result + + +def decode_snapshot( + kind: Literal["task", "run"], + identity: OriginalIdentity, + rows: Any, + *, + source_state: State = "complete", +) -> SnapshotProjection: + """Validate supplied normalized rows; the future driver must prove that normalization. + + Complete means one typed projection matches the supplied original identity. + Raw capture, clock interpretation and row/phase/claim consistency stay unproved. + """ + try: + expected = identity_fields(kind, identity) + require( + type(source_state) is str and source_state in {"complete", "missing", "truncated", "invalid"}, + "SOURCE_STATE_INVALID", + ) + if source_state != "complete": + return SnapshotProjection(source_state, "SOURCE_" + source_state.upper()) + if rows is None: + return SnapshotProjection("missing", "INPUT_MISSING") + require(type(rows) is list and len(rows) <= 1, "ROW_COUNT_OR_TYPE_INVALID") + if not rows: + return SnapshotProjection("missing", "ROW_MISSING") + row = rows[0] + require( + type(row) is dict + and all(type(key) is str for key in row) + and set(row) == (TASK_FIELDS if kind == "task" else RUN_FIELDS), + "FIELD_SET_INVALID", + ) + return SnapshotProjection("complete", row=typed_row(kind, row, expected)) + except ProjectionRefusal as error: + return SnapshotProjection("invalid", error.code, error.field) From 98e0eaed557ffc72fd270f75de22d6be7ea956ab Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 20:33:46 +0000 Subject: [PATCH 2/2] Bound normalized snapshot shape checks before field lookups --- .../sql-proposal/sources.json | 4 ++-- tests/worker_startup_snapshot.py | 11 +++++++---- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json b/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json index 1d8cdb5..5d45c4b 100644 --- a/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json +++ b/tests/fixtures/worker_startup_reconstruction/sql-proposal/sources.json @@ -59,13 +59,13 @@ "https://dev.mysql.com/doc/refman/8.0/en/time-zone-support.html" ], "normalized_row_decoder": { - "bytes": 7423, + "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": "6adc599d9025d1f8e52249b45be486f0e60c8e7e82168c62719c9750adcdb20c" + "sha256": "dd02a5dae1fe2a777a802efbc431844d3e9ca8c646303776db9f7e134640c0e8" }, "parameter_driver_selected": null, "private_binding_comparator_selected": null, diff --git a/tests/worker_startup_snapshot.py b/tests/worker_startup_snapshot.py index a663f9d..fe28e5f 100644 --- a/tests/worker_startup_snapshot.py +++ b/tests/worker_startup_snapshot.py @@ -76,7 +76,7 @@ def text(value: Any, limit: int) -> bool: def identity_fields(kind: str, identity: OriginalIdentity) -> dict[str, str]: - require(type(kind) is str and kind in {"task", "run"}, "UNKNOWN_PROJECTION") + require(type(kind) is str and len(kind) <= 4 and kind in {"task", "run"}, "UNKNOWN_PROJECTION") require(type(identity) is OriginalIdentity, "IDENTITY_INVALID") require( text(identity.namespace, 128) @@ -141,7 +141,7 @@ def typed_row(kind: str, row: dict[str, Any], expected: dict[str, str]) -> dict[ for field in ("session_time_zone", "system_time_zone"): require(text(row[field], 64) and bool(row[field]), "TIMEZONE_LABEL_INVALID", field) require( - type(row["status"]) is str and row["status"] in (TASK_STATUSES if kind == "task" else RUN_STATUSES), + text(row["status"], 32) and row["status"] in (TASK_STATUSES if kind == "task" else RUN_STATUSES), "STATUS_INVALID", "status", ) @@ -178,7 +178,9 @@ def decode_snapshot( try: expected = identity_fields(kind, identity) require( - type(source_state) is str and source_state in {"complete", "missing", "truncated", "invalid"}, + type(source_state) is str + and len(source_state) <= 9 + and source_state in {"complete", "missing", "truncated", "invalid"}, "SOURCE_STATE_INVALID", ) if source_state != "complete": @@ -191,7 +193,8 @@ def decode_snapshot( row = rows[0] require( type(row) is dict - and all(type(key) is str for key in row) + and len(row) == len(TASK_FIELDS if kind == "task" else RUN_FIELDS) + and all(type(key) is str and len(key) <= 64 for key in row) and set(row) == (TASK_FIELDS if kind == "task" else RUN_FIELDS), "FIELD_SET_INVALID", )