From fe9c88473a71e8e1f76d0ccd4566a336d40441e6 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 9 Nov 2026 00:33:36 +0000 Subject: [PATCH 1/2] Prepare native 3.3.1 and installed condition-wait qualification Package the seven accepted transition reductions with upgrade notes and the published Compose default. Keep the paired cache writer deferred. Select the installed SDK C1-C5 qualifier in the existing hosted replay job, retain its nine Native exports, verify the original C2 activity/signal window and CAN lineage, and check the owned MySQL rows in a read-only transaction. The original thirteen replay scenarios remain required and unchanged. Exclude only the publication-subject JSON from unrelated Shared fixture inputs; retain full selection for executable CI helpers and mixed changes. Co-Authored-By: Codex --- .github/workflows/phpunit-feature.yml | 2 +- .../published-workflow-lifecycle.yml | 28 +- .github/workflows/server-parity.yml | 3 + docker-compose.published.yml | 2 +- docs/releases/3.3.1.md | 51 ++++ .../condition-wait-anchored-check.sql | 272 ++++++++++++++++++ .../php-sdk-condition-wait-export.php | 64 +++++ .../php-sdk-condition-wait-input.py | 140 +++++++++ ...php-sdk-replay-host-published-artifacts.sh | 34 ++- scripts/conformance/server-parity/changes.mjs | 15 +- tests/Unit/ServerParityAggregateTest.mjs | 36 +++ tests/Unit/Support/php_condition_wait_test.py | 134 +++++++++ .../Unit/Support/php_replay_published_test.py | 1 + 13 files changed, 762 insertions(+), 20 deletions(-) create mode 100644 docs/releases/3.3.1.md create mode 100644 scripts/conformance/condition-wait-anchored-check.sql create mode 100644 scripts/conformance/php-sdk-condition-wait-export.php create mode 100644 scripts/conformance/php-sdk-condition-wait-input.py create mode 100644 tests/Unit/Support/php_condition_wait_test.py diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index edfd4fdd0..244ebed7b 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -190,7 +190,7 @@ jobs: run: python3 -m unittest discover -s tests/Unit/Support -p 'single_region_failover*test.py' - name: Require all PHP replay runtime cells and exact artifact identity - run: python3 -m unittest tests/Unit/Support/php_replay_published_test.py + run: python3 -m unittest tests/Unit/Support/php_replay_published_test.py tests/Unit/Support/php_condition_wait_test.py feature: name: PHPUnit feature suite diff --git a/.github/workflows/published-workflow-lifecycle.yml b/.github/workflows/published-workflow-lifecycle.yml index 457df2201..3ab8536a2 100644 --- a/.github/workflows/published-workflow-lifecycle.yml +++ b/.github/workflows/published-workflow-lifecycle.yml @@ -4,11 +4,11 @@ on: workflow_dispatch: inputs: experiment: - description: Existing lifecycle aggregate or complete PHP replay matrix + description: Lifecycle aggregate, complete PHP replay, or replay plus installed condition-wait cases required: true default: lifecycle type: choice - options: [lifecycle, php-replay] + options: [lifecycle, php-replay, php-replay-conditions] php_runner_image: description: Published tooling image digest for php-replay (no product source) required: false @@ -67,6 +67,7 @@ jobs: SERVER_PORT: '0' DW_CONFORMANCE_EXPERIMENT: ${{ inputs.experiment }} DW_REPLAY_PHP_RUNNER_IMAGE: ${{ inputs.php_runner_image }} + DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS: ${{ inputs.experiment == 'php-replay-conditions' && '1' || '0' }} DW_WORKFLOW_LIFECYCLE_MYSQL_IMAGE: mysql@sha256:679e7e924f38a3cbb62a3d7df32924b83f7321a602d3f9f967c01b3df18495d6 DW_WORKFLOW_LIFECYCLE_REDIS_IMAGE: redis@sha256:0637954999d01b7c9ce9167db2da50656e2590d3b884f1c600c5f63bb6e6773c steps: @@ -84,7 +85,7 @@ jobs: done case "$DW_CONFORMANCE_EXPERIMENT" in lifecycle) ;; - php-replay) + php-replay|php-replay-conditions) [[ "$DW_REPLAY_PHP_RUNNER_IMAGE" =~ ^ghcr.io/durable-workflow/sample-app-devcontainer@sha256:[0-9a-f]{64}$ ]] ;; *) exit 2 ;; @@ -94,13 +95,17 @@ jobs: if: ${{ always() && steps.validate.outcome == 'success' }} run: | mkdir -p "$RUNNER_TEMP/published-workflow-lifecycle" - if [[ "$DW_CONFORMANCE_EXPERIMENT" == php-replay ]]; then - bash scripts/conformance/replay-published-artifacts.sh --focus php \ - --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" - else - bash scripts/conformance/workflow-lifecycle-host-published-artifacts.sh \ - --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" - fi + case "$DW_CONFORMANCE_EXPERIMENT" in + php-replay|php-replay-conditions) + bash scripts/conformance/replay-published-artifacts.sh --focus php \ + --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" + ;; + lifecycle) + bash scripts/conformance/workflow-lifecycle-host-published-artifacts.sh \ + --result-dir "$RUNNER_TEMP/published-workflow-lifecycle" + ;; + *) exit 2 ;; + esac - name: Remove job-owned lifecycle containers and network if: always() @@ -114,7 +119,7 @@ jobs: docker compose -p "$COMPOSE_PROJECT_NAME" -f docker-compose.php.yml down --volumes --remove-orphans >/dev/null 2>&1 || true - name: Require PHP replay cleanup receipt - if: ${{ always() && steps.validate.outcome == 'success' && inputs.experiment == 'php-replay' }} + if: ${{ always() && steps.validate.outcome == 'success' && (inputs.experiment == 'php-replay' || inputs.experiment == 'php-replay-conditions') }} run: test "$(<"$RUNNER_TEMP/published-workflow-lifecycle/php-replay-cleanup.txt")" = 'cleanup=0' - name: Make lifecycle evidence readable @@ -140,6 +145,7 @@ jobs: ${{ runner.temp }}/published-workflow-lifecycle/*.txt ${{ runner.temp }}/published-workflow-lifecycle/*.csv ${{ runner.temp }}/published-workflow-lifecycle/*.tsv + ${{ runner.temp }}/published-workflow-lifecycle/*.sql ${{ runner.temp }}/published-workflow-lifecycle/*.sqlite if-no-files-found: error retention-days: 30 diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 557a0b17f..af4f34272 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -14,6 +14,9 @@ on: - Cargo.lock - scripts/conformance/** - scripts/ci/** + # Publication identity is checked by Native promotion contracts. The + # fixture jobs do not read it; every other CI helper stays selected. + - '!scripts/ci/native-qualification-subject.json' - tests/Unit/ServerParity*.mjs - tests/Support/ServerParity*.mjs - scripts/conformance/server-parity* diff --git a/docker-compose.published.yml b/docker-compose.published.yml index c6eac606f..83605822c 100644 --- a/docker-compose.published.yml +++ b/docker-compose.published.yml @@ -16,7 +16,7 @@ name: durable-workflow-server # release whose default names another image). Set DW_SERVER_IMAGE to run # any other native image, by tag or digest; pin a digest for production # change control. -x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:3.3.0} +x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:3.3.1} services: server: diff --git a/docs/releases/3.3.1.md b/docs/releases/3.3.1.md new file mode 100644 index 000000000..bbef0ff85 --- /dev/null +++ b/docs/releases/3.3.1.md @@ -0,0 +1,51 @@ +# Durable Workflow Server 3.3.1 (native) + +Server 3.3.1 is a patch release of the native (Rust) Server, published as +`durableworkflow/server:3.3.1` and `ghcr.io/durable-workflow/server:3.3.1`. +The published Compose file defaults to it. Rolling tags move in the separate +promotion step after qualification. + +## Upgrade + +There is no schema upgrade from 3.3.0. MySQL, MariaDB and PostgreSQL remain +at native schema 11; SQLite remains at schema 10. The worker protocol stays +at `1.20`, the control plane at version `2`, and the capabilities are unchanged. +Existing clients need no SDK release for these internal query reductions. + +For upgrades from 3.2.x or earlier, follow the +[3.3.0 upgrade and rollback instructions](3.3.0.md#upgrade), including the +backup, explicit `schema-upgrade` and condition-wait rollback constraints. +This patch does not relax those constraints or repair a mixed condition-wait +chain. + +## Changes + +[PR #1099](https://github.com/durable-workflow/server/pull/1099) reduces +database statements on workflow transitions while retaining the existing +answers, transaction boundaries and lease authority: + +- Upsert a run summary instead of checking its existence before writing it. +- Read a workflow poll's cached binding and result together. +- Aggregate the projection census in one query. +- Read the authored sequence counters together, retaining integer types on + MariaDB and zero for missing rows. +- Reuse the waiting-status result already read in the transition. +- Reuse fenced worker registration within the transaction. +- Keep run tags within the existing transaction context. + +The paired poll-cache writer proposed in +[issue #1063](https://github.com/durable-workflow/server/issues/1063) is +deferred. Binding and result writes retain their separate expiry-time origins. +The other items in the performance report remain tracked individually. +No benchmark speedup or latency improvement is claimed for this release. + +The qualification tooling also verifies the published PHP replay matrix +under a read-only root, dropped capabilities and `no-new-privileges`. A private, +byte-identical PHP executable copy avoids inheriting the tooling image's file +capability; Server and SDK product bytes are not patched for the experiment. + +## Known limitations + +The [3.3.0 known limitations](3.3.0.md#known-limitations) still apply. +Source/backend query-count controls and published-artifact compatibility are +separate evidence; passing them does not reproduce the reporter's benchmark. diff --git a/scripts/conformance/condition-wait-anchored-check.sql b/scripts/conformance/condition-wait-anchored-check.sql new file mode 100644 index 000000000..b9bf6e85d --- /dev/null +++ b/scripts/conformance/condition-wait-anchored-check.sql @@ -0,0 +1,272 @@ +-- Read-only anchored check v4 for sdk-php#269 occurrence-deadline cells on +-- Native with server#1098 (condition_wait_occurrence_deadline), MySQL 8.x. +-- A distinct successor. The original condition-wait-physical-check.sql +-- (a162896d), the v1 prototype (485861da), v2 (d09159ce) and v3 (f03de3f0) +-- stay unchanged. +-- v4 changes no predicate. It makes every text value independent of the +-- client connection: each output column, each session-variable or CAST/ +-- DATE_FORMAT string, and each table-column/payload comparison is pinned to +-- utf8mb4_bin (case-sensitive, PAD SPACE). v3 aborted on a latin1 client with ERROR 1267 +-- (utf8mb3 SYSCONST @@session.time_zone UNION latin1 CAST); a utf8mb4 +-- client would instead mix utf8mb4_0900_ai_ci with the schema's +-- utf8mb4_unicode_ci. Run-ID membership uses that collation; the runner admits fixed-format ULIDs. +-- v3 checks each identity's raw JSON type is STRING before unquoting it: +-- JSON null unquotes to the text 'null' and must never pass as an identity. +-- Server-owned bytes; bind them by SHA-256. +-- +-- Native keeps the policy, original budget and anchor A in history payloads +-- (ConditionWaitOpened: deadline_policy, condition_wait_timeout_budget_seconds, +-- occurrence_deadline_at; TimerScheduled: deadline_policy, occurrence_deadline_at, +-- fire_at, timer_id) and in the first timer row's fire_at. There is no schema +-- column for them. The waits projection copies TimerScheduled.fire_at into +-- workflow_run_waits.deadline_at (wait_id = condition_wait_id). +-- +-- Every predicate refuses with "(...) IS NOT TRUE", so a missing or null +-- field is a violation, never a silent pass. +-- +-- Caller sets these, then sources this file on a read-only connection: +-- SET time_zone = '+00:00'; -- required (B-1) +-- SET @run_ids = ',...'; -- every closed run of the cell +-- SET @max_dispatch_seconds = 2; -- required, exactly the selected 2 s (B-2) +-- SET @expected_budgets = NULL; -- or a JSON object mapping +-- '|' to that chain's original T. +-- Leave NULL for a heterogeneous cell whose per-chain T is checked by +-- the SDK evidence; when set, every chain needs an equal entry and no +-- key may name a chain that is not present (B2b, B2c). +-- Output: one row per violation (assertion, run_id, detail). No rows = PASS. +-- Only SELECTs; no table or row is written. + +WITH opens AS ( + SELECT h.workflow_run_id AS run_id, h.sequence AS hseq, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_id, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_wait_occurrence_id')) AS occ, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_key')) AS ckey, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_definition_fingerprint')) AS fp, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy, + JSON_EXTRACT(h.payload, '$.condition_wait_timeout_budget_seconds') AS budget_j, + JSON_EXTRACT(h.payload, '$.timeout_seconds') AS timeout_j, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_s, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.condition_wait_occurrence_id')) AS occ_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_t + FROM workflow_history_events h + WHERE FIND_IN_SET(h.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND h.event_type = 'ConditionWaitOpened' +), sched AS ( + SELECT h.workflow_run_id AS run_id, h.sequence AS hseq, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_id, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.timer_id')) AS timer_id, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.fire_at')) AS fire_s, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy, + JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_s, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.condition_wait_id')) AS wait_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.timer_id')) AS timer_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.fire_at')) AS fire_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.deadline_policy')) AS policy_t, + JSON_TYPE(JSON_EXTRACT(h.payload, '$.occurrence_deadline_at')) AS anchor_t + FROM workflow_history_events h + WHERE FIND_IN_SET(h.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND h.event_type = 'TimerScheduled' + AND JSON_UNQUOTE(JSON_EXTRACT(h.payload, '$.timer_kind')) = 'condition_timeout' +), firsts AS ( + SELECT o.* FROM opens o + WHERE o.hseq = (SELECT MIN(f.hseq) FROM opens f WHERE f.run_id = o.run_id AND f.occ <=> o.occ) +), anchors AS ( + -- A is the first open's own timer row fire_at, read by its timer ID. The + -- bijection checks B8 make any join multiplication visible on their own. + SELECT f.run_id, f.occ, f.wait_id AS first_wait, f.budget_j, f.ckey, f.fp, f.anchor_s, + t.id AS timer_id, t.fire_at AS a_at + FROM firsts f + LEFT JOIN sched s ON s.run_id = f.run_id AND s.wait_id = f.wait_id + LEFT JOIN workflow_run_timers t ON t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = f.run_id +) + +SELECT CONVERT('B-1 session time zone is not +00:00' USING utf8mb4) COLLATE utf8mb4_bin AS assertion, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin AS run_id, CONVERT(@@session.time_zone USING utf8mb4) COLLATE utf8mb4_bin AS detail +FROM DUAL WHERE (@@session.time_zone = '+00:00') IS NOT TRUE + +UNION ALL +SELECT CONVERT('B-2 dispatch bound is not the selected 2 seconds' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(COALESCE(CAST(@max_dispatch_seconds AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL') USING utf8mb4) COLLATE utf8mb4_bin +FROM DUAL WHERE (@max_dispatch_seconds <=> 2) IS NOT TRUE + +UNION ALL +SELECT CONVERT('B-3 expected budget map is not a JSON object' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(LEFT(CAST(@expected_budgets AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 200) USING utf8mb4) COLLATE utf8mb4_bin +FROM DUAL WHERE @expected_budgets IS NOT NULL + AND (JSON_VALID(@expected_budgets) AND JSON_TYPE(@expected_budgets) = 'OBJECT') IS NOT TRUE + +UNION ALL +-- Every physical open carries the complete opt-in, with integer budget and +-- remaining timeout and 0 <= timeout <= budget. +SELECT CONVERT('B1 condition wait without a complete occurrence policy' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' policy=', COALESCE(o.policy, 'NULL'), ' budget=', COALESCE(CAST(o.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), + ' timeout=', COALESCE(CAST(o.timeout_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' occ=', COALESCE(o.occ, 'NULL'), + ' anchor=', COALESCE(o.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +WHERE (o.policy = 'occurrence' AND JSON_TYPE(o.budget_j) = 'INTEGER' AND JSON_TYPE(o.timeout_j) = 'INTEGER' + AND CAST(o.timeout_j AS SIGNED) >= 0 AND CAST(o.timeout_j AS SIGNED) <= CAST(o.budget_j AS SIGNED) + AND o.wait_t = 'STRING' AND o.occ_t = 'STRING' AND o.policy_t = 'STRING' AND o.anchor_t = 'STRING' + AND o.wait_id <> '' AND o.occ <> '' AND o.anchor_s <> '') IS NOT TRUE + +UNION ALL +-- The chain's first open authors its whole budget. +SELECT CONVERT('B2 first open does not author its whole budget' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(f.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(f.wait_id, 'NULL'), ' timeout=', COALESCE(CAST(f.timeout_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), + ' budget=', COALESCE(CAST(f.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM firsts f +WHERE (JSON_TYPE(f.timeout_j) = 'INTEGER' AND JSON_TYPE(f.budget_j) = 'INTEGER' + AND CAST(f.timeout_j AS SIGNED) = CAST(f.budget_j AS SIGNED)) IS NOT TRUE + +UNION ALL +-- When a per-chain map is bound, each chain's original T equals its entry. +SELECT CONVERT('B2b chain budget differs from its expected entry' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(f.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(f.occ, 'NULL'), ' budget=', COALESCE(CAST(f.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' expected=', + COALESCE(CAST(JSON_EXTRACT(@expected_budgets, CONCAT('$."', f.run_id COLLATE utf8mb4_bin, '|', f.occ, '"')) AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'missing')) USING utf8mb4) COLLATE utf8mb4_bin +FROM firsts f +WHERE @expected_budgets IS NOT NULL + AND (CAST(JSON_EXTRACT(@expected_budgets, CONCAT('$."', f.run_id COLLATE utf8mb4_bin, '|', f.occ, '"')) AS SIGNED) + = CAST(f.budget_j AS SIGNED)) IS NOT TRUE + +UNION ALL +SELECT CONVERT('B2c expected budget names no present chain' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(NULL USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(k.chain USING utf8mb4) COLLATE utf8mb4_bin +FROM JSON_TABLE(COALESCE(JSON_KEYS(@expected_budgets), JSON_ARRAY()), '$[*]' COLUMNS (chain VARCHAR(300) PATH '$')) k +WHERE @expected_budgets IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM firsts f WHERE CONCAT(f.run_id COLLATE utf8mb4_bin, '|', f.occ) = CONVERT(k.chain USING utf8mb4) COLLATE utf8mb4_bin) + +UNION ALL +-- Every open of one run/occurrence keeps the first open's budget, key, +-- fingerprint and recorded anchor string. +SELECT CONVERT('B3 reopen changed its chain identity, budget or anchor' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' vs first ', COALESCE(a.first_wait, 'NULL'), + ' budget ', COALESCE(CAST(o.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), '/', COALESCE(CAST(a.budget_j AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin, 'NULL'), + ' anchor ', COALESCE(o.anchor_s, 'NULL'), '/', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o JOIN anchors a ON a.run_id = o.run_id AND a.occ <=> o.occ +WHERE (CAST(o.budget_j AS SIGNED) = CAST(a.budget_j AS SIGNED) AND o.ckey <=> a.ckey AND o.fp <=> a.fp + AND o.anchor_s = a.anchor_s) IS NOT TRUE + +UNION ALL +-- B8: exact one-to-one identities, so no join below can hide a duplicate. +SELECT CONVERT('B8a duplicate physical open identity' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' opens=', COUNT(*)) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o GROUP BY o.run_id, o.wait_id HAVING COUNT(*) <> 1 + +UNION ALL +SELECT CONVERT('B8b open without exactly one condition-timeout schedule' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' schedules=', + (SELECT COUNT(*) FROM sched s WHERE s.run_id = o.run_id AND s.wait_id <=> o.wait_id)) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +WHERE ((SELECT COUNT(*) FROM sched s WHERE s.run_id = o.run_id AND s.wait_id <=> o.wait_id) = 1) IS NOT TRUE + +UNION ALL +SELECT CONVERT('B8c orphan condition-timeout schedule' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(s.run_id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(CONCAT(COALESCE(s.wait_id, 'NULL'), ' timer=', COALESCE(s.timer_id, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM sched s +WHERE NOT EXISTS (SELECT 1 FROM opens o WHERE o.run_id = s.run_id AND o.wait_id <=> s.wait_id) + +UNION ALL +SELECT CONVERT('B8d schedule identity not a string, timer shared or without its row' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(s.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(s.wait_id, 'NULL'), ' timer=', COALESCE(s.timer_id, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM sched s +WHERE (s.wait_t = 'STRING' AND s.timer_t = 'STRING' AND s.fire_t = 'STRING' AND s.policy_t = 'STRING' AND s.anchor_t = 'STRING' + AND s.wait_id <> '' AND s.timer_id <> '' + AND (SELECT COUNT(*) FROM sched d WHERE d.timer_id = s.timer_id) = 1 + AND (SELECT COUNT(*) FROM workflow_run_timers t WHERE t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = s.run_id) = 1) IS NOT TRUE + +UNION ALL +-- The first open's recorded anchor is exactly its own timer row's fire_at +-- (Native writes microsecond RFC 3339 with Z). +SELECT CONVERT('B4 first open anchor is not its own timer fire_at' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(a.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(a.first_wait, 'NULL'), ' timer=', COALESCE(a.timer_id, 'missing'), ' fire_at=', + COALESCE(CONVERT(DATE_FORMAT(a.a_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' recorded=', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM anchors a +WHERE (CONVERT(DATE_FORMAT(a.a_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin = a.anchor_s) IS NOT TRUE + +UNION ALL +-- Every open, the first included, scheduled its timer at exactly A. +SELECT CONVERT('B5 chain timer not scheduled exactly at the first anchor' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' timer=', COALESCE(s.timer_id, 'missing'), ' scheduled=', COALESCE(s.fire_s, 'NULL'), + ' policy=', COALESCE(s.policy, 'NULL'), ' row_fire_at=', + COALESCE(CONVERT(DATE_FORMAT(t.fire_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'NULL'), ' A=', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +JOIN anchors a ON a.run_id = o.run_id AND a.occ <=> o.occ +LEFT JOIN sched s ON s.run_id = o.run_id AND s.wait_id = o.wait_id +LEFT JOIN workflow_run_timers t ON t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = o.run_id +WHERE (s.policy = 'occurrence' AND s.anchor_s = a.anchor_s AND s.fire_s = a.anchor_s AND t.fire_at = a.a_at) IS NOT TRUE + +UNION ALL +-- The waits projection carries A as every chain wait's deadline. +SELECT CONVERT('B6 projected wait deadline is not the first anchor' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(o.wait_id, 'NULL'), ' deadline_at=', COALESCE(CONVERT(DATE_FORMAT(w.deadline_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'missing'), + ' A=', COALESCE(a.anchor_s, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +JOIN anchors a ON a.run_id = o.run_id AND a.occ <=> o.occ +LEFT JOIN workflow_run_waits w ON w.workflow_run_id = o.run_id AND w.wait_id COLLATE utf8mb4_bin = o.wait_id AND w.kind = 'condition' +WHERE (w.deadline_at = a.a_at) IS NOT TRUE + +UNION ALL +-- A fired chain timer is never early and fires within the selected bound. +SELECT CONVERT('B7 chain timer fired early or beyond the dispatch bound' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(o.run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(t.id, ' fire_at=', CONVERT(DATE_FORMAT(t.fire_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, + ' fired_at=', COALESCE(CONVERT(DATE_FORMAT(t.fired_at, '%Y-%m-%dT%H:%i:%s.%fZ') USING utf8mb4) COLLATE utf8mb4_bin, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM opens o +JOIN sched s ON s.run_id = o.run_id AND s.wait_id = o.wait_id +JOIN workflow_run_timers t ON t.id COLLATE utf8mb4_bin = s.timer_id AND t.workflow_run_id = o.run_id +WHERE t.status = 'fired' + AND (t.fired_at >= t.fire_at + AND TIMESTAMPDIFF(MICROSECOND, t.fire_at, t.fired_at) <= @max_dispatch_seconds * 1000000) IS NOT TRUE + +UNION ALL +-- The original A1, A2, A3, A4, A5, A6 and A0 predicates, unchanged in meaning. +-- Server#1098 changed no projection status, timer settlement or task rule. +SELECT CONVERT('A1 open condition wait' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(w.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(w.wait_id, ' status=', w.status, ' source_status=', COALESCE(w.source_status, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_waits w +WHERE FIND_IN_SET(w.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND w.kind = 'condition' AND w.status = 'open' + +UNION ALL +SELECT CONVERT('A2 condition wait not settled to a native terminal pair' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(w.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(w.wait_id, ' ', w.status, '/', COALESCE(w.source_status, 'NULL'), ' resolved_at=', COALESCE(w.resolved_at, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_waits w +WHERE FIND_IN_SET(w.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND w.kind = 'condition' + AND NOT (w.resolved_at IS NOT NULL AND ( + (w.status = 'resolved' AND w.source_status = 'satisfied') + OR (w.status = 'cancelled' AND w.source_status IN ('cancelled', 'terminated', 'continued', 'closed')))) + +UNION ALL +SELECT CONVERT('A3 condition timeout timer not fired or cancelled' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(t.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(t.id, ' seq=', t.sequence, ' status=', t.status, ' fire_at=', t.fire_at) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_timers t +WHERE FIND_IN_SET(t.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND t.status NOT IN ('fired', 'cancelled') + AND EXISTS (SELECT 1 FROM workflow_run_waits w WHERE w.workflow_run_id = t.workflow_run_id + AND w.kind = 'condition' AND w.sequence = t.sequence) + +UNION ALL +SELECT CONVERT('A4 unexpected timer entry row' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(e.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(CONCAT(e.id, ' status=', COALESCE(e.status, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_timer_entries e +WHERE FIND_IN_SET(e.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) + +UNION ALL +SELECT CONVERT('A5 runnable condition timeout task after close' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(k.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(k.id, ' status=', k.status, ' available_at=', COALESCE(k.available_at, 'NULL')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_tasks k +WHERE FIND_IN_SET(k.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND k.task_type = 'timer' AND k.status IN ('ready', 'leased') + AND JSON_EXTRACT(k.payload, '$.condition_wait_id') IS NOT NULL + +UNION ALL +SELECT CONVERT('A6 overlapping condition waits for one key' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(a.workflow_run_id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT(COALESCE(a.target_name, 'NULL'), ': ', a.wait_id, ' and ', b.wait_id) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_run_waits a +JOIN workflow_run_waits b ON b.workflow_run_id = a.workflow_run_id AND b.kind = 'condition' + AND b.target_name <=> a.target_name AND b.wait_id > a.wait_id +WHERE FIND_IN_SET(a.workflow_run_id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) AND a.kind = 'condition' + AND a.opened_at < COALESCE(b.resolved_at, '9999-12-31') AND b.opened_at < COALESCE(a.resolved_at, '9999-12-31') + +UNION ALL +SELECT CONVERT('A6 condition rows differ from ConditionWaitOpened events' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(r.id USING utf8mb4) COLLATE utf8mb4_bin, + CONVERT(CONCAT('rows=', (SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id = r.id AND w.kind = 'condition'), + ' events=', (SELECT COUNT(*) FROM workflow_history_events h WHERE h.workflow_run_id = r.id AND h.event_type = 'ConditionWaitOpened')) USING utf8mb4) COLLATE utf8mb4_bin +FROM workflow_runs r +WHERE FIND_IN_SET(r.id, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) + AND (SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id = r.id AND w.kind = 'condition') + <> (SELECT COUNT(*) FROM workflow_history_events h WHERE h.workflow_run_id = r.id AND h.event_type = 'ConditionWaitOpened') + +UNION ALL +SELECT CONVERT('A0 run missing or still open' USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(ids.id USING utf8mb4) COLLATE utf8mb4_bin, CONVERT(COALESCE(r.status, 'missing') USING utf8mb4) COLLATE utf8mb4_bin +FROM JSON_TABLE(CONCAT('["', REPLACE(CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin, ',', '","'), '"]'), '$[*]' COLUMNS (id VARCHAR(26) PATH '$')) ids +LEFT JOIN workflow_runs r ON r.id COLLATE utf8mb4_bin = CONVERT(ids.id USING utf8mb4) COLLATE utf8mb4_bin +WHERE r.id IS NULL OR r.status NOT IN ('completed', 'failed', 'cancelled', 'terminated'); diff --git a/scripts/conformance/php-sdk-condition-wait-export.php b/scripts/conformance/php-sdk-condition-wait-export.php new file mode 100644 index 000000000..cf9a38a9a --- /dev/null +++ b/scripts/conformance/php-sdk-condition-wait-export.php @@ -0,0 +1,64 @@ + $workflowId) { + $export = $client->exportWorkflowHistory($workflowId, $runId); + $handle = fopen($root.'/php-condition-wait-export-'.$runId.'.json', 'xb'); + if ($handle === false) { + throw new RuntimeException('condition_export_output_exists'); + } + $bytes = json_encode($export, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT).PHP_EOL; + if (fwrite($handle, $bytes) !== strlen($bytes)) { + throw new RuntimeException('condition_export_short_write'); + } + fclose($handle); + } + echo json_encode(['exports' => count($runs), 'outcome' => 'pass'], JSON_THROW_ON_ERROR).PHP_EOL; +} catch (Throwable $error) { + $message = 'Condition export failed: '.get_class($error); + if ($error instanceof ServerException) { + $message .= ' status='.$error->status; + if (preg_match('/\A[a-z0-9_]{1,80}\z/D', (string) $error->reason) === 1) { + $message .= ' reason='.$error->reason; + } + } elseif (preg_match('/\Acondition_export_[a-z_]{1,60}\z/D', $error->getMessage()) === 1) { + $message .= ' reason='.$error->getMessage(); + } + fwrite(STDERR, $message.PHP_EOL); + exit(1); +} diff --git a/scripts/conformance/php-sdk-condition-wait-input.py b/scripts/conformance/php-sdk-condition-wait-input.py new file mode 100644 index 000000000..6b7866b6f --- /dev/null +++ b/scripts/conformance/php-sdk-condition-wait-input.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +"""Bind the installed C1-C5 runs and raw exports to the reviewed read-only SQL.""" +import hashlib +import json +from pathlib import Path +import re +import sys + +CASES = ('C1', 'C2', 'C3-terminate', 'C3-cancel', 'C3-continue', 'C4', 'C5') +SQL_SHA256 = '1f828778b952b931fd99acccc05af415aab1e279f2713cc164f737a99337b21c' +RUN_ID = re.compile(r'[0-9A-HJKMNP-TV-Z]{26}') + + +def require(condition, reason): + if not condition: + raise ValueError(reason) + + +def report_runs(report): + require(isinstance(report, dict), 'condition_report_shape') + require(isinstance(report.get('namespace'), str) + and re.fullmatch(r'php-cwn-[a-f0-9]{12}', report['namespace']) is not None, + 'condition_namespace') + cases = report.get('cases') + require(isinstance(cases, dict) and set(cases) == set(CASES), 'condition_cases_missing') + rows = [] + for case in CASES: + entry = cases[case] + require(isinstance(entry, dict) and entry.get('evidence'), 'condition_evidence_missing') + require(isinstance(entry.get('command_contract'), dict) + and entry['command_contract'].get('source') == 'durable_history', + 'condition_contract_not_durable') + runs = entry.get('runs') + require(isinstance(runs, list) and len(runs) == (2 if case in ('C3-continue', 'C5') else 1), + 'condition_run_count') + for run in runs: + require(isinstance(run, dict), 'condition_run_shape') + require(isinstance(run.get('run_id'), str) and RUN_ID.fullmatch(run['run_id']) is not None, + 'condition_run_id') + require(isinstance(run.get('workflow_id'), str) + and re.fullmatch(r'c[1-5]-[a-z0-9-]+', run['workflow_id']) is not None, + 'condition_workflow_id') + require(isinstance(run.get('events'), list) and run['events'] + and all(isinstance(event, str) and event for event in run['events']), + 'condition_events_missing') + rows.append((case, run)) + ids = [run['run_id'] for _, run in rows] + require(len(ids) == 9 and len(set(ids)) == 9 and report.get('run_ids') == ids, + 'condition_run_inventory') + require(report.get('c1_run_ids') == [rows[0][1]['run_id']] and report.get('c1_timeout') == 20, + 'condition_first_case_identity') + return rows + + +def validate_exports(report, exports): + rows = report_runs(report) + for case, run in rows: + export = exports.get(run['run_id']) + require(isinstance(export, dict) and export.get('schema') == 'durable-workflow.v2.history-export' + and export.get('schema_version') == 2 and export.get('history_complete') is True, + 'condition_export_incomplete') + workflow = export.get('workflow', {}) + require(isinstance(workflow, dict) + and workflow.get('run_id') == run['run_id'] and workflow.get('instance_id') == run['workflow_id'] + and workflow.get('source_runtime') == 'standalone_server' + and workflow.get('source_process_class') == 'server_http_node' + and 'migration_projection' not in export, 'condition_export_identity') + events = export.get('history_events') + require(isinstance(events, list) and len(events) >= 2, 'condition_export_events') + identities = set() + for sequence, event in enumerate(events, 1): + require(isinstance(event, dict) and type(event.get('sequence')) is int + and event['sequence'] == sequence and isinstance(event.get('id'), str) + and event['id'] and event['id'] not in identities, 'condition_export_prefix') + identities.add(event['id']) + require([event.get('type') for event in events] == run['events'], 'condition_export_history_mismatch') + pair = report['cases'][case]['runs'] + if len(pair) == 2 and run['run_id'] == pair[1]['run_id']: + parent = pair[0] + started = events[0].get('payload', {}) + parent_events = exports[parent['run_id']]['history_events'] + links = [event.get('payload', {}) for event in parent_events + if event.get('type') == 'WorkflowContinuedAsNew'] + require(events[0]['type'] == 'WorkflowStarted' and isinstance(started, dict) + and events[0].get('workflow_task_id') is None + and events[0].get('workflow_command_id') is None + and parent['workflow_id'] == run['workflow_id'] + and started.get('continued_from_run_id') == parent['run_id'] + and started.get('workflow_run_id') == run['run_id'] + and started.get('workflow_instance_id') == run['workflow_id'] + and isinstance(started.get('workflow_link_id'), str) and started['workflow_link_id'] + and len(links) == 1 and isinstance(links[0], dict) + and links[0].get('continued_to_run_id') == run['run_id'] + and links[0].get('workflow_link_id') == started['workflow_link_id'], + 'condition_export_continuation_authority') + else: + require(events[0]['type'] == 'StartAccepted' and events[1]['type'] == 'WorkflowStarted' + and events[0].get('workflow_task_id') is None and events[1].get('workflow_task_id') is None + and isinstance(events[0].get('workflow_command_id'), str) + and events[0]['workflow_command_id'] + and events[0]['workflow_command_id'] == events[1].get('workflow_command_id'), + 'condition_export_start_authority') + if case == 'C2': + kinds = [event['type'] for event in events] + required = ('ActivityScheduled', 'ActivityCompleted', 'ConditionWaitOpened', 'WorkflowCompleted') + require(all(kind in kinds for kind in required), 'condition_c2_activity_missing') + scheduled, completed, opened = (kinds.index(kind) for kind in required[:3]) + noise = [index for index, event in enumerate(events) if event['type'] == 'SignalReceived' + and isinstance(event.get('payload'), dict) and event['payload'].get('signal_name') == 'noise'] + require(any(scheduled < index < completed < opened for index in noise), + 'condition_c2_signal_not_during_activity') + return rows + + +def sql_input(report, exports, sql): + rows = validate_exports(report, exports) + require(hashlib.sha256(sql).hexdigest() == SQL_SHA256, 'condition_sql_identity') + ids = ','.join(run['run_id'] for _, run in rows) + return ("SET SESSION time_zone = '+00:00';\nSET @run_ids = '" + ids + "';\n" + "SET @expected_namespace = '" + report['namespace'] + "';\n" + "SET @max_dispatch_seconds = 2;\nSET @expected_budgets = NULL;\n" + "START TRANSACTION READ ONLY;\n" + "SELECT 'R0_run_inventory', COUNT(*) FROM workflow_runs " + "WHERE FIND_IN_SET(id COLLATE utf8mb4_bin, CONVERT(@run_ids USING utf8mb4) COLLATE utf8mb4_bin) " + "AND namespace COLLATE utf8mb4_bin = CONVERT(@expected_namespace USING utf8mb4) COLLATE utf8mb4_bin " + "HAVING COUNT(*) <> 9;\n" + sql.decode('utf-8') + "\nROLLBACK;\n") + + +def main(): + root = Path(sys.argv[1]) + report = json.loads((root / 'php-condition-waits.json').read_bytes()) + rows = report_runs(report) + exports = {run['run_id']: json.loads((root / ('php-condition-wait-export-' + run['run_id'] + '.json')).read_bytes()) + for _, run in rows} + sql = Path(__file__).with_name('condition-wait-anchored-check.sql').read_bytes() + sys.stdout.write(sql_input(report, exports, sql)) + + +if __name__ == '__main__': + main() diff --git a/scripts/conformance/php-sdk-replay-host-published-artifacts.sh b/scripts/conformance/php-sdk-replay-host-published-artifacts.sh index 55cb4e56b..5b06b1ca7 100755 --- a/scripts/conformance/php-sdk-replay-host-published-artifacts.sh +++ b/scripts/conformance/php-sdk-replay-host-published-artifacts.sh @@ -10,11 +10,13 @@ fi server_image="${DW_SERVER_IMAGE:-}" runner_image="${DW_REPLAY_PHP_RUNNER_IMAGE:-}" mysql_image="${DW_WORKFLOW_LIFECYCLE_MYSQL_IMAGE:-}" +condition_waits="${DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS:-0}" if [[ ! "${DW_SERVER_VERSION:-}" =~ ^3\.[0-9]+\.[0-9]+$ \ || ! "${DW_PHP_SDK_VERSION:-}" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ \ || ! "$server_image" =~ ^(docker.io/)?durableworkflow/server@sha256:[0-9a-f]{64}$ \ || ! "$runner_image" =~ ^ghcr.io/durable-workflow/sample-app-devcontainer@sha256:[0-9a-f]{64}$ \ - || ! "$mysql_image" =~ ^(docker.io/library/)?mysql@sha256:[0-9a-f]{64}$ ]]; then + || ! "$mysql_image" =~ ^(docker.io/library/)?mysql@sha256:[0-9a-f]{64}$ \ + || ! "$condition_waits" =~ ^[01]$ ]]; then echo 'PHP replay requires exact Native/SDK versions and digest-pinned Server, MySQL and published tooling images.' >&2 exit 2 fi @@ -134,7 +136,8 @@ docker create --name "$server" --label "$label" --network "$network" \ -e DB_CONNECTION=mysql -e DB_HOST="$mysql" -e DB_PORT=3306 \ -e DB_DATABASE=durable_workflow -e DB_USERNAME=workflow -e DB_PASSWORD=workflow \ -e DW_BIND_ADDRESS=0.0.0.0:8080 -e DW_AUTH_DRIVER=token \ - -e DW_AUTH_TOKEN=php-replay-control -e DW_WORKER_TOKEN=php-replay-worker \ + -e DW_AUTH_TOKEN=claimed-task-admin -e DW_OPERATOR_TOKEN=claimed-task-client \ + -e DW_WORKER_TOKEN=claimed-task-worker \ "$server_image" >"$result_dir/php-replay-server-id.txt" grep -Ex '[0-9a-f]{64}' "$result_dir/php-replay-server-id.txt" >/dev/null creating='' @@ -150,9 +153,10 @@ docker create --name "$probe" --label "$label" --network "$network" \ -e DW_SERVER_VERSION -e DW_SERVER_IMAGE -e DW_PHP_SDK_VERSION \ -e DW_PHP_SDK_CONFORMANCE_SERVER_URL="http://$server:8080" \ -e DW_PHP_SDK_CONFORMANCE_NAMESPACE=php-replay-conformance \ - -e DW_PHP_SDK_CONFORMANCE_TOKEN=php-replay-control \ - -e DW_PHP_SDK_CONFORMANCE_WORKER_TOKEN=php-replay-worker \ + -e DW_PHP_SDK_CONFORMANCE_TOKEN=claimed-task-admin \ + -e DW_PHP_SDK_CONFORMANCE_WORKER_TOKEN=claimed-task-worker \ -e DW_PHP_SDK_CONFORMANCE_REPLAY_MATRIX=1 \ + -e DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS="$condition_waits" \ --entrypoint bash "$runner_image" -euc ' mkdir -p -m 0700 "$HOME/bin" # The image PHP has a low-port file capability, which cannot execute with @@ -182,6 +186,16 @@ docker create --name "$probe" --label "$label" --network "$network" \ --data "{\"name\":\"$DW_PHP_SDK_CONFORMANCE_NAMESPACE\",\"retention_days\":7}" \ "$DW_PHP_SDK_CONFORMANCE_SERVER_URL/api/namespaces" > /result/php-replay-namespace.json bash /runner/php-sdk-published-artifacts.sh --result-dir /result + if [[ "$DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS" == 1 ]]; then + # Run the qualifier from the normal installed package, with its own + # synthetic namespace and worker. The original replay matrix stays intact. + php -l /runner/php-sdk-condition-wait-export.php > /result/php-condition-export-lint.txt + php /result/php-sdk-project/vendor/durable-workflow/sdk/scripts/qualify-condition-wait-native.php \ + /result/php-sdk-project/vendor/autoload.php "$DW_PHP_SDK_CONFORMANCE_SERVER_URL" \ + > /result/php-condition-waits.json 2> /result/php-condition-waits.log + php /runner/php-sdk-condition-wait-export.php > /result/php-condition-wait-exports.json + python3 /runner/php-sdk-condition-wait-input.py /result > /result/php-condition-wait-input.sql + fi python3 /runner/php-sdk-replay-result.py /result ' >"$result_dir/php-replay-probe-id.txt" grep -Ex '[0-9a-f]{64}' "$result_dir/php-replay-probe-id.txt" >/dev/null @@ -189,3 +203,15 @@ creating='' docker start --attach "$probe" >"$result_dir/php-replay-shard.log" 2>&1 test "$(query container inspect --format '{{.State.Status}}' "$probe")" = exited test "$(query container inspect --format '{{.State.ExitCode}}' "$probe")" = 0 +if [[ "$condition_waits" == 1 ]]; then + # The reviewed checker is SELECT-only, and the session is read-only too. + timeout 60 docker exec -i -e MYSQL_PWD=workflow "$mysql" mysql \ + --default-character-set=utf8mb4 --batch --raw --skip-column-names -uworkflow durable_workflow \ + < "$result_dir/php-condition-wait-input.sql" \ + > "$result_dir/php-condition-wait-sql.tsv" 2> "$result_dir/php-condition-wait-sql.err" + test ! -s "$result_dir/php-condition-wait-sql.err" + test ! -s "$result_dir/php-condition-wait-sql.tsv" + sha256sum "$result_dir/php-condition-wait-input.sql" > "$result_dir/php-condition-wait-sql-sha256.txt" + printf '{"outcome":"pass","violations":0,"readonly_transaction":true}\n' \ + > "$result_dir/php-condition-wait-sql-check.json" +fi diff --git a/scripts/conformance/server-parity/changes.mjs b/scripts/conformance/server-parity/changes.mjs index fdd59540c..cf4b3da87 100644 --- a/scripts/conformance/server-parity/changes.mjs +++ b/scripts/conformance/server-parity/changes.mjs @@ -28,10 +28,15 @@ export function pushPathPatterns(workflowText) { let inPaths = false; for (const line of lines.slice(push + 1)) { if (/^ {0,3}\S/.test(line)) break; + if (/^ {4,}#/.test(line)) continue; if (line === ' paths:') { inPaths = true; } else if (inPaths && line.startsWith(' - ')) { - patterns.push(line.slice(' - '.length).trim()); + // YAML requires quoting a leading ! so it is not read as a tag. + const entry = line.slice(' - '.length).trim(); + const quoted = /^'(![^']+)'$/.exec(entry); + if (!quoted && /^['"!]/.test(entry)) throw new Error(`Unsupported path filter entry: ${entry}`); + patterns.push(quoted ? quoted[1] : entry); } else if (inPaths) { break; } @@ -66,8 +71,12 @@ export function nulSeparatedPaths(bytes) { } export function matchingFiles(patterns, files) { - const expressions = patterns.map(patternRegExp); - return files.filter(file => expressions.some(expression => expression.test(file))); + // GitHub evaluates paths in order: the last matching pattern decides. + const rules = patterns.map(pattern => pattern.startsWith('!') + ? {include: false, expression: patternRegExp(pattern.slice(1))} + : {include: true, expression: patternRegExp(pattern)}); + if (rules.length > 0 && !rules[0].include) throw new Error('A path filter cannot start with a negated pattern'); + return files.filter(file => rules.reduce((selected, rule) => rule.expression.test(file) ? rule.include : selected, false)); } function classify({event, headSha, mergeBaseRef, mergeBaseSha, mergeHeadSha, workflowPath}) { diff --git a/tests/Unit/ServerParityAggregateTest.mjs b/tests/Unit/ServerParityAggregateTest.mjs index 2f39690da..718cb16d7 100644 --- a/tests/Unit/ServerParityAggregateTest.mjs +++ b/tests/Unit/ServerParityAggregateTest.mjs @@ -71,6 +71,34 @@ test('path patterns follow GitHub filter semantics for the forms in use', () => assert.deepEqual(matchingFiles(['rust/**', 'docs/x.md'], ['README.md', 'rust/Cargo.toml', 'docs/x.md']), ['rust/Cargo.toml', 'docs/x.md']); }); +test('only the publication subject is excluded from CI-helper fixture inputs', () => { + const patterns = pushPathPatterns(workflowText); + const subject = 'scripts/ci/native-qualification-subject.json'; + assert.deepEqual(matchingFiles(patterns, [subject]), []); + // These executable helpers remain selected, including the isolation check + // that the changes job now executes before the expensive fixture fan-out. + const selected = ['rust/src/runtime/store.rs', 'Cargo.lock', '.github/workflows/server-parity.yml', + 'scripts/ci/check-docker-workflow-isolation.py', 'scripts/ci/native_alias_evidence.py', + 'scripts/ci/native_qualification.py', 'scripts/ci/new-helper.sh', + 'scripts/ci/nested/native-qualification-subject.json', + 'scripts/conformance/server-parity/native-image-corpus.sh']; + for (const file of selected) { + assert.deepEqual(matchingFiles(patterns, [file]), [file], `${file} must re-run the fixtures`); + assert.deepEqual(matchingFiles(patterns, [subject, file]), [file], `a subject change must not hide ${file}`); + } +}); + +test('negated path filters preserve GitHub ordering and refuse ambiguous YAML entries', () => { + assert.deepEqual(matchingFiles(['a/**', '!a/b.py', 'a/b*'], ['a/b.py', 'a/c.py', 'x.py']), ['a/b.py', 'a/c.py']); + assert.deepEqual(matchingFiles(['a/**', '!a/*.py'], ['a/b.py', 'a/b/c.py']), ['a/b/c.py']); + assert.throws(() => matchingFiles(['!a/**', 'b/**'], ['b/c']), /cannot start with a negated pattern/); + assert.throws(() => patternRegExp('!a/**'), /Unsupported path filter pattern/); + const listed = entries => pushPathPatterns(`on:\n push:\n paths:\n${entries}\njobs:\n`); + assert.deepEqual(listed(" - a/**\n # Ordered exclusion\n - '!a/b.py'"), ['a/**', '!a/b.py']); + assert.throws(() => listed(' - a/**\n - !a/b.py'), /Unsupported path filter entry/); + assert.throws(() => listed(" - a/**\n - '!a/b.py"), /Unsupported path filter entry/); +}); + test('literal native inputs and parity imports are covered by the shared input list', () => { const sourceFiles = (directory, suffix) => readdirSync(directory, {withFileTypes: true}).flatMap(entry => { const path = join(directory, entry.name); @@ -331,6 +359,12 @@ function classify(root, env) { for (const [label, change, expected] of [ ['a documentation-only change skips the fixtures', ({write}) => write('docs/note.md', 'x\n'), 'run=false'], + ['a publication-subject-only change skips the fixtures', ({write}) => + write('scripts/ci/native-qualification-subject.json', '{}\n'), 'run=false'], + ['a publication subject cannot hide a Rust change', ({write}) => { + write('scripts/ci/native-qualification-subject.json', '{}\n'); + write('rust/src/lib.rs', 'x\n'); + }, 'run=true'], ['a Rust change runs the fixtures', ({write}) => write('rust/src/lib.rs', 'x\n'), 'run=true'], ['a configuration-audit contract edit runs the fixtures', ({write}) => write('config/dw-contract.php', " ['env' => 'FIXTURE_OPTION']];\n"), 'run=true'], @@ -471,6 +505,8 @@ test('the workflow supplies the merge-group identity to the classifier', () => { for (const [label, change, expected] of [ ['PHP-only changes skip the fixtures', ({write}) => write('app/Controller.php', ' write('docs/note.md', 'x\n'), 'run=false'], + ['a publication-subject-only change skips the fixtures', ({write}) => + write('scripts/ci/native-qualification-subject.json', '{}\n'), 'run=false'], ['native changes run the fixtures', ({write}) => write('rust/src/lib.rs', 'x\n'), 'run=true'], ['parent Cargo configuration runs the fixtures', ({write}) => write('.cargo/config.toml', '[build]\n'), 'run=true'], ['parent Cargo manifests run the fixtures', ({write}) => write('Cargo.toml', '[workspace]\n'), 'run=true'], diff --git a/tests/Unit/Support/php_condition_wait_test.py b/tests/Unit/Support/php_condition_wait_test.py new file mode 100644 index 000000000..3a4c9772a --- /dev/null +++ b/tests/Unit/Support/php_condition_wait_test.py @@ -0,0 +1,134 @@ +import importlib.util +from pathlib import Path +import unittest + +RUNNERS = Path(__file__).resolve().parents[3] / 'scripts/conformance' +spec = importlib.util.spec_from_file_location('condition_input', RUNNERS / 'php-sdk-condition-wait-input.py') +gate = importlib.util.module_from_spec(spec) +spec.loader.exec_module(gate) + + +class InstalledConditionWaitInputTest(unittest.TestCase): + def fixture(self): + report = {'namespace': 'php-cwn-0123456789ab', 'cases': {}, 'run_ids': [], 'c1_timeout': 20} + exports = {} + number = 0 + for case in gate.CASES: + rows = [] + for position in range(2 if case in ('C3-continue', 'C5') else 1): + number += 1 + run_id = str(number).zfill(26) + workflow_id = case.lower() + '-0123456789ab' + kinds = ['StartAccepted', 'WorkflowStarted', 'ConditionWaitOpened', 'WorkflowCompleted'] + if case == 'C2': + kinds = ['StartAccepted', 'WorkflowStarted', 'ActivityScheduled', 'ActivityStarted', + 'SignalReceived', 'ActivityCompleted', 'ConditionWaitOpened', 'WorkflowCompleted'] + if position == 1: + kinds = ['WorkflowStarted', 'WorkflowCompleted'] + events = [{'id': run_id + '-event-' + str(index), 'sequence': index, 'type': kind, + 'workflow_command_id': 'start-authority' if index <= 2 else None, + 'payload': {'signal_name': 'noise'} if kind == 'SignalReceived' else {}} + for index, kind in enumerate(kinds, 1)] + if position == 1: + parent = rows[0] + link = 'continue-link-' + run_id + events[0]['workflow_command_id'] = None + events[0]['payload'] = {'continued_from_run_id': parent['run_id'], 'workflow_run_id': run_id, + 'workflow_instance_id': workflow_id, 'workflow_link_id': link} + parent['events'][-1] = 'WorkflowContinuedAsNew' + terminal = exports[parent['run_id']]['history_events'][-1] + terminal['type'] = 'WorkflowContinuedAsNew' + terminal['payload'] = {'continued_to_run_id': run_id, 'workflow_link_id': link} + exports[run_id] = { + 'schema': 'durable-workflow.v2.history-export', 'schema_version': 2, 'history_complete': True, + 'workflow': {'run_id': run_id, 'instance_id': workflow_id, + 'source_runtime': 'standalone_server', 'source_process_class': 'server_http_node'}, + 'history_events': events, + } + rows.append({'run_id': run_id, 'workflow_id': workflow_id, 'events': kinds}) + report['run_ids'].append(run_id) + report['cases'][case] = {'command_contract': {'source': 'durable_history'}, + 'runs': rows, 'evidence': {'checked_by_installed_qualifier': True}} + report['c1_run_ids'] = report['run_ids'][:1] + return report, exports + + def test_complete_inventory_binds_all_nine_runs_in_a_read_only_transaction(self): + report, exports = self.fixture() + sql = (RUNNERS / 'condition-wait-anchored-check.sql').read_bytes() + result = gate.sql_input(report, exports, sql) + self.assertIn("SET @run_ids = '" + ','.join(report['run_ids']) + "';", result) + self.assertIn('START TRANSACTION READ ONLY;\n', result) + self.assertIn("SET @expected_namespace = 'php-cwn-0123456789ab';", result) + self.assertIn("SELECT 'R0_run_inventory', COUNT(*)", result) + self.assertIn('HAVING COUNT(*) <> 9;', result) + self.assertTrue(result.endswith('\nROLLBACK;\n')) + + def test_missing_duplicate_and_untrusted_run_inventories_refuse(self): + for mutation in ('missing_case', 'extra_case', 'missing_evidence', 'contract', 'duplicate', 'sql_id', 'workflow_id'): + with self.subTest(mutation=mutation): + report, _ = self.fixture() + row = report['cases']['C2']['runs'][0] + if mutation == 'missing_case': del report['cases']['C2'] + if mutation == 'extra_case': report['cases']['unexpected'] = report['cases']['C1'] + if mutation == 'missing_evidence': report['cases']['C2']['evidence'] = {} + if mutation == 'contract': report['cases']['C2']['command_contract']['source'] = 'inferred' + if mutation == 'duplicate': report['run_ids'][1] = report['run_ids'][0] + if mutation == 'sql_id': row['run_id'] = "x'; DELETE FROM workflow_runs; --" + if mutation == 'workflow_id': row['workflow_id'] = '../other-run' + with self.assertRaises(ValueError): gate.report_runs(report) + + def test_incomplete_foreign_imported_and_unbound_exports_refuse(self): + for mutation in ('missing', 'truncated', 'foreign_run', 'foreign_workflow', 'imported', 'gap', + 'duplicate_event', 'different_history', 'wrong_start'): + with self.subTest(mutation=mutation): + report, exports = self.fixture() + run_id = report['run_ids'][0] + export = exports[run_id] + if mutation == 'missing': del exports[run_id] + if mutation == 'truncated': export['history_complete'] = False + if mutation == 'foreign_run': export['workflow']['run_id'] = report['run_ids'][1] + if mutation == 'foreign_workflow': export['workflow']['instance_id'] = 'other-instance' + if mutation == 'imported': export['migration_projection'] = {} + if mutation == 'gap': export['history_events'][1]['sequence'] = 3 + if mutation == 'duplicate_event': export['history_events'][1]['id'] = export['history_events'][0]['id'] + if mutation == 'different_history': export['history_events'][-1]['type'] = 'WorkflowFailed' + if mutation == 'wrong_start': export['history_events'][1]['workflow_command_id'] = 'other-command' + with self.assertRaises(ValueError): gate.validate_exports(report, exports) + + def test_c2_must_actually_overlap_the_activity_before_opening_the_wait(self): + for mutation in ('no_noise', 'before_activity', 'after_activity'): + with self.subTest(mutation=mutation): + report, exports = self.fixture() + run_id = report['cases']['C2']['runs'][0]['run_id'] + events = exports[run_id]['history_events'] + if mutation == 'no_noise': events[4]['payload']['signal_name'] = 'other' + if mutation == 'before_activity': events[2]['type'], events[4]['type'] = events[4]['type'], events[2]['type']; events[2]['payload'] = {'signal_name': 'noise'} + if mutation == 'after_activity': events[4]['type'], events[5]['type'] = events[5]['type'], events[4]['type']; events[5]['payload'] = {'signal_name': 'noise'} + report['cases']['C2']['runs'][0]['events'] = [event['type'] for event in events] + with self.assertRaisesRegex(ValueError, 'signal_not_during_activity'): + gate.validate_exports(report, exports) + + def test_sql_drift_refuses_before_emitting_a_statement(self): + report, exports = self.fixture() + with self.assertRaisesRegex(ValueError, 'sql_identity'): + gate.sql_input(report, exports, (RUNNERS / 'condition-wait-anchored-check.sql').read_bytes() + b'\nSELECT 1;') + + def test_continue_as_new_requires_the_exact_parent_and_both_halves_of_its_link(self): + for mutation in ('parent', 'child', 'link', 'missing_predecessor'): + with self.subTest(mutation=mutation): + report, exports = self.fixture() + parent, child = report['cases']['C5']['runs'] + started = exports[child['run_id']]['history_events'][0]['payload'] + if mutation == 'parent': started['continued_from_run_id'] = report['run_ids'][0] + if mutation == 'child': started['workflow_run_id'] = report['run_ids'][0] + if mutation == 'link': started['workflow_link_id'] = 'other-link' + if mutation == 'missing_predecessor': + export = exports[parent['run_id']] + export['history_events'][-1]['type'] = 'WorkflowCompleted' + parent['events'][-1] = 'WorkflowCompleted' + with self.assertRaisesRegex(ValueError, 'continuation_authority'): + gate.validate_exports(report, exports) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/Unit/Support/php_replay_published_test.py b/tests/Unit/Support/php_replay_published_test.py index 43b60c31a..bcf0bc4c0 100644 --- a/tests/Unit/Support/php_replay_published_test.py +++ b/tests/Unit/Support/php_replay_published_test.py @@ -93,6 +93,7 @@ def test_invalid_artifact_pins_refuse_before_any_docker_command(self): ('DW_SERVER_IMAGE', 'durableworkflow/server:3.3.0'), ('DW_REPLAY_PHP_RUNNER_IMAGE', 'ghcr.io/foreign/tool@sha256:' + 'b' * 64), ('DW_WORKFLOW_LIFECYCLE_MYSQL_IMAGE', 'mysql:latest'), + ('DW_PHP_SDK_CONFORMANCE_CONDITION_WAITS', 'true'), ('DW_JOB_RESOURCE_PREFIX', 'bad/name')]: with self.subTest(key=key): env = dict(base); env[key] = value From 887cd0f15e3bcdbc8ddfae341704b54ddd685511 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 9 Nov 2026 00:49:03 +0000 Subject: [PATCH 2/2] Read the installed qualifier report after its progress output Retain the normal SDK CLI stdout verbatim, then require its progress prefix and one complete validated report before the export and SQL checks. Reject unexpected output, crashes, trailing data and incomplete reports. Keep the published package bytes and original replay scenarios unchanged. Co-Authored-By: Codex --- .../php-sdk-condition-wait-input.py | 19 ++++++++++++++++++ ...php-sdk-replay-host-published-artifacts.sh | 3 ++- tests/Unit/Support/php_condition_wait_test.py | 20 +++++++++++++++++++ 3 files changed, 41 insertions(+), 1 deletion(-) diff --git a/scripts/conformance/php-sdk-condition-wait-input.py b/scripts/conformance/php-sdk-condition-wait-input.py index 6b7866b6f..c8f8ee3d6 100644 --- a/scripts/conformance/php-sdk-condition-wait-input.py +++ b/scripts/conformance/php-sdk-condition-wait-input.py @@ -52,6 +52,20 @@ def report_runs(report): return rows +def extract_report(raw): + """Keep the published CLI's progress separate from its final JSON object.""" + lines = raw.splitlines(keepends=True) + progress = 0 + for line in lines: + if not line.startswith(b'ok: '): + break + progress += 1 + require(progress > 0, 'condition_cli_progress_missing') + report = json.loads(b''.join(lines[progress:])) + report_runs(report) + return report + + def validate_exports(report, exports): rows = report_runs(report) for case, run in rows: @@ -127,6 +141,11 @@ def sql_input(report, exports, sql): def main(): + if len(sys.argv) == 3 and sys.argv[1] == '--extract-report': + root = Path(sys.argv[2]) + report = extract_report((root / 'php-condition-waits.stdout.log').read_bytes()) + sys.stdout.write(json.dumps(report, indent=2) + '\n') + return root = Path(sys.argv[1]) report = json.loads((root / 'php-condition-waits.json').read_bytes()) rows = report_runs(report) diff --git a/scripts/conformance/php-sdk-replay-host-published-artifacts.sh b/scripts/conformance/php-sdk-replay-host-published-artifacts.sh index 5b06b1ca7..771ac6ed0 100755 --- a/scripts/conformance/php-sdk-replay-host-published-artifacts.sh +++ b/scripts/conformance/php-sdk-replay-host-published-artifacts.sh @@ -192,7 +192,8 @@ docker create --name "$probe" --label "$label" --network "$network" \ php -l /runner/php-sdk-condition-wait-export.php > /result/php-condition-export-lint.txt php /result/php-sdk-project/vendor/durable-workflow/sdk/scripts/qualify-condition-wait-native.php \ /result/php-sdk-project/vendor/autoload.php "$DW_PHP_SDK_CONFORMANCE_SERVER_URL" \ - > /result/php-condition-waits.json 2> /result/php-condition-waits.log + > /result/php-condition-waits.stdout.log 2> /result/php-condition-waits.log + python3 /runner/php-sdk-condition-wait-input.py --extract-report /result > /result/php-condition-waits.json php /runner/php-sdk-condition-wait-export.php > /result/php-condition-wait-exports.json python3 /runner/php-sdk-condition-wait-input.py /result > /result/php-condition-wait-input.sql fi diff --git a/tests/Unit/Support/php_condition_wait_test.py b/tests/Unit/Support/php_condition_wait_test.py index 3a4c9772a..d04e6dee8 100644 --- a/tests/Unit/Support/php_condition_wait_test.py +++ b/tests/Unit/Support/php_condition_wait_test.py @@ -1,4 +1,5 @@ import importlib.util +import json from pathlib import Path import unittest @@ -63,6 +64,25 @@ def test_complete_inventory_binds_all_nine_runs_in_a_read_only_transaction(self) self.assertIn('HAVING COUNT(*) <> 9;', result) self.assertTrue(result.endswith('\nROLLBACK;\n')) + def test_published_cli_progress_is_retained_separately_from_the_final_report(self): + report, _ = self.fixture() + raw = b'ok: runtime advertises condition_wait_idempotent_reopen\nok: history fits one complete page\n' + raw += json.dumps(report, indent=2).encode() + b'\n' + with self.assertRaises(json.JSONDecodeError): + json.loads(raw) + self.assertEqual(gate.extract_report(raw), report) + + def test_cli_noise_crashes_missing_progress_and_partial_reports_refuse(self): + report, _ = self.fixture() + body = json.dumps(report).encode() + for raw in (body, b'ok: started\n', b'unexpected log\n' + body, + b'ok: started\nunexpected log\n' + body, + b'ok: started\n' + body + b'\ntrailing output', + b'ok: started\n{}'): + with self.subTest(raw=raw[:60]): + with self.assertRaises(ValueError): + gate.extract_report(raw) + def test_missing_duplicate_and_untrusted_run_inventories_refuse(self): for mutation in ('missing_case', 'extra_case', 'missing_evidence', 'contract', 'duplicate', 'sql_id', 'workflow_id'): with self.subTest(mutation=mutation):