Rust Server train: condition waits (#713), task snapshots (#718), patch cohort (#722), message streams (#709)

.github/workflows/phpunit-feature.yml

@@ -71,6 +71,9 @@ jobs:
       - name: Require safe PHP image build context
         run: node --test tests/Unit/PhpImageContextTest.mjs
 
+      - name: Restore the patch cohort's maintenance loop on every path
+        run: node --test tests/Unit/PatchCohortMaintenanceTest.mjs
+
       - name: Validate embedded Workflow package authority
         run: php scripts/ci/test-workflow-package-authority.php
 

.github/workflows/server-parity.yml

@@ -43,6 +43,7 @@ on:
       - tests/Unit/ServerParityRepairPassContractTest.mjs
       - tests/Unit/ServerParityClusterInfoContractTest.mjs
       - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs
+      - tests/Unit/ServerParityMessageStreamsContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
       - tests/Unit/ServerParityPublishedPatchContractTest.mjs
@@ -99,6 +100,7 @@ on:
       - tests/Unit/ServerParityRepairPassContractTest.mjs
       - tests/Unit/ServerParityClusterInfoContractTest.mjs
       - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs
+      - tests/Unit/ServerParityMessageStreamsContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
       - tests/Unit/ServerParityPublishedPatchContractTest.mjs
@@ -1529,7 +1531,7 @@ jobs:
             -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \
             install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress
           docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \
-            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs > parity-evidence/comparator-tests.txt
           docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \
             --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \
             "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt

docs/rust-server-port-checkpoints/2026-10-23-condition-wait-fields.md

@@ -0,0 +1,21 @@
+# October 23, optional condition wait fields
+
+- **The gap.** Native refused an `open_condition_wait` command without a
+  `condition_key`, or without a `condition_definition_fingerprint` of the form
+  `sha256:` followed by 64 hex digits. Published PHP accepts both.
+- **How PHP 2.11.1 does it.** `normalizeOpenConditionWaitCommand` treats the key
+  and the fingerprint as optional strings of any form. It trims them and leaves
+  them out when empty. `applyOpenConditionWait` records only the fields that
+  remain, on `ConditionWaitOpened` and on a condition timeout's
+  `TimerScheduled`. Run summaries name a keyless wait "Waiting for condition",
+  without a key. Observed live against the published image: a bare
+  `open_condition_wait`, a key of `"  ready  "` with the fingerprint
+  `"any form"`, and blank values all behave this way.
+- **What native does now.** The same. The open wait projection stores no
+  target name for a keyless wait, and the run summary reads such a wait as
+  PHP does. A key or fingerprint that is not a string is still refused by the
+  completion transport rules.
+- **How it was checked.** A new integration test covers a bare wait, a padded
+  key with an arbitrary fingerprint, blank values with a timeout, and a key
+  that is not a string. The 63 existing condition, timed-condition and
+  observation tests still pass.

docs/rust-server-port-checkpoints/2026-10-23-message-streams.md

@@ -0,0 +1,64 @@
+# October 23, message streams
+
+- **The gap.** All three published SDKs declare `message-streams` from
+  worker protocol 1.15. Native refused a worker that asked for them, did not
+  serve the append and diagnostics routes, and ignored
+  `message_stream_cursors` and `message_stream_waits` on a completion.
+- **How PHP 2.11.1 does it.** Learned from its source, then observed live
+  against the published image:
+  - Each accepted message is delivered to the current run as the
+    runtime-reserved signal `__durable_workflow_message_stream`, in the
+    append's transaction. The signal skips the declared-signal and argument
+    checks but not the pending-signal and payload ceilings.
+  - SDK receives open a condition wait keyed by the stream cursor. A stream
+    message wakes that wait.
+  - When a signal wakes a condition wait, PHP consumes it without
+    `SignalApplied`, unless the run's durable contract declares that signal.
+    This holds for any undeclared signal, not only stream messages.
+  - The stream cursor's `MessageCursorAdvanced` is recorded after the
+    completion, with the completed task's snapshot.
+  - On continue-as-new, the next run receives a payload-free cursor
+    checkpoint for each acknowledged stream, then every unconsumed message
+    again, by append order. The old run rejects the signals it had not
+    consumed.
+- **What native does now.**
+  - `POST /api/workflows/{id}/message-streams/{stream}/messages`, with PHP's
+    validation order, refusals and diagnostics of malformed input.
+  - `GET /api/workflows/{id}/message-streams` and
+    `GET /api/workflows/{id}/message-streams/{stream}`.
+  - Completion cursors and waits: validated before anything changes,
+    recorded in the completion's transaction, handed over on continue-as-new.
+  - The worker capability is PHP's contract with `supported: true`, and
+    cluster info publishes `message_streams_contract`.
+  - Continue-as-new no longer refuses a run with message streams.
+- **Native changes found on the way.**
+  - A signal consumed by a condition wait no longer records `SignalApplied`
+    unless it is declared, and keeps the wait's sequence.
+  - The instance-level `MessageCursorAdvanced` carries the task snapshot, as
+    PHP's does.
+  - A completion reports the run's next message task in
+    `created_task_ids`, as PHP does.
+- **How it was checked.**
+  - The `message-streams` fixture: 15 appends, 8 refused completions, the
+    deliveries each worker read, both runs' full event sequences, and the
+    diagnostics at each step.
+  - Published PHP 2.11.1 and native pass it on SQLite. The probe's output
+    is the same size on both, and both runs' event sequences are identical.
+  - The contract test refuses 50 changed observations, among them a
+    `SignalApplied` for a consumed message, a missing next message task, a
+    redelivered consumed message and a missing cursor checkpoint.
+- **Not done yet** (Server #325 follow-ups):
+  - **External payload references.** Native refuses a reference as stream
+    input with 501 `message_stream_external_payload_unavailable`. PHP keeps
+    the reference and resolves it when history reaches a worker.
+  - **Cleanup and upgrade.** Retention cleanup of streams
+    (`MessageStreamRetentionCleanup`) is not ported. Takeover of PHP data
+    with streams is still refused.
+  - **Cluster `capabilities` map.** Native has none yet, so
+    `capabilities.message_streams` is absent; that map is the cluster info
+    work.
+  - **History presentation, outside streams.** Task snapshots lack PHP's
+    `last_dispatch_attempt_at` and `last_dispatched_at`. A consumed signal's
+    `SignalReceived` lacks `workflow_sequence` in poll history.
+  - **Condition fingerprint.** Native requires `condition_definition_fingerprint`
+    on `open_condition_wait`; PHP accepts a wait without one.

docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md

@@ -0,0 +1,79 @@
+# October 23, task snapshots on history events and described commands
+
+- **The gap.** Published PHP's `WorkflowHistoryEvent::record` adds a `task`
+  snapshot to every event it records with the task model in hand. Native
+  added snapshots event by event and missed several. Driving one workflow
+  through a side effect, an activity, a timer, a child, a signal wait and a
+  condition wait against both servers showed that native had no snapshot on:
+  - `TimerFired`, `ChildWorkflowScheduled`, `ChildRunStarted`;
+  - `SignalWaitOpened`, `ConditionWaitOpened`, `ConditionWaitSatisfied`;
+  - `SignalApplied` and the instance `MessageCursorAdvanced`.
+
+  Native's own snapshots on `TimerScheduled`, `ActivityStarted`,
+  `ActivityCompleted` and `ActivityFailed` also left out fields that PHP
+  keeps, such as `queue`, `connection`, `available_at` and `repair_count`.
+- **What native does now.**
+  - When an event of a type PHP records with the task model is appended with
+    a task and no snapshot of its own, native adds the task's snapshot. The
+    list of those event types comes from the Workflow package's source.
+  - `TimerScheduled` uses the full snapshot. The activity events' snapshots
+    keep PHP's fields.
+  - As `RunTimerTask` does, a timer task is leased to itself while its timer
+    fires: status leased, `leased_at`, `lease_owner` set to the task id, a
+    workflow-task lease, and one more attempt. Its lease is cleared once the
+    timer has fired. `TimerFired` therefore carries the leased timer task.
+- **How it was checked.** The same probe now finds the same events with
+  snapshots on both servers, with the same fields. A new integration test
+  checks the timer, child, signal-wait, signal and condition-wait events.
+- **Described commands.** A field-by-field diff of the same workflow's
+  description on both servers found two more gaps. Each command PHP
+  describes names the run's `workflow_class` (`WorkflowCommand::record` keeps
+  it), and says whether its payload preview is omitted
+  (`payload_preview_omitted`, true only for a stored external payload over the
+  size limit). Native now sends both. With fresh workers on both sides, the
+  descriptions then differ only in `request_route_name` and the
+  `request_fingerprint` hashed from it.
+- **Security review corrections.**
+  - The size check read only PHP's `external_storage` envelope. Native's
+    stored envelope names its reference `external_payload`, so the flag was
+    always false. It now reads both, which also corrects the run's input and
+    output previews. As PHP does, an omitted preview also omits the
+    command's `target_name`.
+  - The query read any command payload starting with `dw-`, and SQLite's
+    `LIKE` and MySQL's case-insensitive collation ignore case. It also read the
+    wrong column for signals. Native keeps a signal's stored input reference
+    in its signal record (`workflow_signal_records.arguments`) and a start's
+    in the command row (`workflow_commands.payload`). Other commands keep
+    none. The query now reads only that column for each of the two command
+    types, and only when its first 30 characters equal native's stored prefix
+    exactly: `COLLATE BINARY` on SQLite, a binary cast on MySQL and MariaDB,
+    and PostgreSQL's case-sensitive text comparison. On MySQL 8.0 and MariaDB
+    10.11 the binary cast refuses the upper-case prefix that the column's
+    collation would match.
+  - A task snapshot is read only from a task of the event's own run.
+  - Tests:
+    - a real uploaded reference on a signal, found in the signal record and
+      not in the command row;
+    - that signal record rewritten with the same reference at the 2 MiB limit
+      (not omitted) and one byte over (omitted, with no target name). The
+      described command never includes the reference;
+    - counterfactuals that are not native references: an upper-case or v2
+      prefix, the prefix with no valid reference after it, and a large inline
+      payload starting with `dw-`. An oversized reference in the signal's
+      command row is not counted either;
+    - the same sizes and counterfactuals in a start command's row;
+    - on SQLite, the query itself returns only the exact prefix in the
+      command type's own column;
+    - an event recorded on one run with another run's task carries no
+      snapshot.
+
+    Each test fails without its correction.
+- **Left as they are.**
+  - PHP's command snapshots carry `request_route_name`, a name Laravel
+    generates at random for the deployment's cached routes. Native has no
+    such name.
+  - Native stamps a dispatch on every new task, so the snapshot of a task
+    created inside a completion keeps `last_dispatch_attempt_at` and
+    `last_dispatched_at`, which PHP's does not (Server #325 follow-up).
+  - A message stream's `SignalApplied` and the condition wait it satisfies
+    follow the message streams change (#709).

docs/rust-server-port.md

@@ -144,7 +144,10 @@ any other unknown path or method answers PHP's own 404 or 405
 - **Run operations.**
   - Imports from embedded v2 and Waterline v1.
 - **Other families.**
-  - Message streams.
+  - Message streams are served
+    ([checkpoint](rust-server-port-checkpoints/2026-10-23-message-streams.md)),
+    except external payload references as input, stream retention cleanup
+    and takeover of PHP data with streams.
   - Durable workflow stream items and close.
   - Service endpoints. (The service catalog reads from Server 2.11 are served:
     native never holds catalog rows, so every list is PHP's empty page and every
@@ -271,9 +274,12 @@ contract fixture before its correction.
   Add a fixture for each before changing them.
 - **Retention of runs with external payloads or message streams.**
   Published PHP deletes a pruned run's external payload objects and releases
-  its message stream items. Native does not clean either up yet. It skips
-  such a run with `external_payload_cleanup_unavailable` or
-  `message_stream_cleanup_unavailable` and keeps its details.
+  its message stream items. It also compacts its instance's consumed inbound
+  stream items, and removes the instance's streams with its last run. Native
+  does not clean any of these up yet. It skips such a run, including any run
+  whose instance has inbound streams, with
+  `external_payload_cleanup_unavailable` or
+  `message_stream_cleanup_unavailable`, and keeps its details.
 - **Deeply nested JSON bodies.** Published PHP decodes JSON up to 512 levels
   deep. Native checks every body as PHP does before any route, but its
   handlers parse with serde_json's 128-level limit. A body nested 129 to 512

rust/src/runtime/activity_failures.rs

@@ -540,7 +540,9 @@ where
             }
             payload["task"] = json!({"id":task_id,"type":"activity","status":"leased",
                 "available_at":DB::optional_instant(&task,"available_at")?,"leased_at":DB::optional_instant(&task,"leased_at")?,
-                "lease_owner":DB::optional_string(&task,"lease_owner")?,"attempt_count":count});
+                "lease_owner":DB::optional_string(&task,"lease_owner")?,"lease_expires_at":DB::optional_instant(&task,"lease_expires_at")?,
+                "attempt_count":count,"repair_count":DB::number(&task,"repair_count")?,"queue":DB::string(&task,"queue")?,
+                "connection":DB::PHP_CONNECTION});
             let failed_event = payload.clone();
             Self::append(
                 &mut tx,