Rust #325: answer worker-plane errors as published PHP does

.github/workflows/server-parity.yml

@@ -63,6 +63,7 @@ on:
       - tests/Unit/ServerParityContinueTransfersContractTest.mjs
       - tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs
       - tests/Unit/ServerParityManyUpdatesContractTest.mjs
+      - tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs
       - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs
       - tests/Unit/ServerParityQueryArgumentsContractTest.mjs
       - tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs
@@ -1989,7 +1990,7 @@ jobs:
           bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \
             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/ServerParityRunnerBlockedExcerptTest.mjs tests/Unit/ServerParityProbeNotesTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.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/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.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/ServerParityConcurrentWaitsContractTest.mjs tests/Unit/ServerParityContinueTransfersContractTest.mjs tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs tests/Unit/ServerParityManyUpdatesContractTest.mjs tests/Unit/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.mjs tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs tests/Unit/ServerParityParentCloseCancellationContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityServiceEndpointsContractTest.mjs tests/Unit/ServerParityServiceCallsContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.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 tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityAdapterCacheTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs tests/Unit/ServerParityProbeNotesTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.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/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.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/ServerParityConcurrentWaitsContractTest.mjs tests/Unit/ServerParityContinueTransfersContractTest.mjs tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs tests/Unit/ServerParityManyUpdatesContractTest.mjs tests/Unit/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.mjs tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs tests/Unit/ServerParityParentCloseCancellationContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityServiceEndpointsContractTest.mjs tests/Unit/ServerParityServiceCallsContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.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 tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityAdapterCacheTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.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-29-worker-error-bodies.md

@@ -0,0 +1,56 @@
+# October 29, worker-plane error bodies
+
+- **Before.** Sending the same malformed worker requests to published
+  2.11.5 and native showed three differences:
+  - a workflow or activity poll that failed its request rules was answered
+    with native's own reasons (`422 invalid_request`, `invalid_poll_timeout`
+    and the like), not Laravel's `validation_failed`;
+  - failing or heartbeating an activity task the namespace does not hold was
+    answered with a bare `{message, reason}` outside the worker envelope;
+  - completing a workflow task the namespace does not hold with a
+    `complete_workflow` command without a result was answered
+    `422 unsupported_payload_codec`, where PHP answers `404 task_not_found`.
+
+  Validation failures on other worker routes, and unknown routes and wrong
+  methods in the default namespace, already matched.
+- **Contract** (published 2.11.5; `bootstrap/app.php` exception rendering
+  with `WorkerProtocol::json`, `WorkerController::pollWorkflowTasks`,
+  `ActivityTaskController::poll`, `guardAttemptOwnership`):
+  - a worker-plane validation failure answers `422 validation_failed` with
+    Laravel's messages in `errors` and `validation_errors`, plus
+    `protocol_version` and `server_capabilities`;
+  - the workflow poll's rules are `worker_id` and `task_queue` (required
+    strings), `build_id`, `poll_request_id` (at most 255),
+    `timeout_seconds` (0 to 60), `history_page_size` (1 to 1000),
+    `accept_history_encoding` (at most 64) and `task_kinds` (1 or 2 of
+    `workflow` and `update_validation`, distinct). The activity poll takes
+    the first five;
+  - completing, failing or heartbeating an activity task the namespace does
+    not hold answers `404` with `task_id`, `activity_attempt_id`,
+    `error: "Activity task not found."` and `reason: task_not_found`, in the
+    envelope;
+  - completing a workflow task the namespace does not hold answers `404`
+    with `task_id`, `workflow_task_attempt` and "Workflow task not found."
+    The ownership guard runs before the command normalizer reads payloads.
+- **Fixture.** `worker-error-bodies`, raw worker HTTP on protocol 1.20 in a
+  fresh namespace: twenty requests. Ten of them probe `distinct` on
+  `task_kinds.*` (from review). It runs in the same pass as the other rules,
+  and compares candidates as Laravel takes them (`Arr::dot`: list or
+  associative items, scalars and empty arrays, not non-empty nested arrays)
+  with PHP 8's loose `==`. That includes `zendi_smart_str_equals` for numeric
+  strings: integers that fit compare exactly, and integers overflowing to the
+  same side, or equal infinities, compare as text. The probe records each status and body,
+  with `server_capabilities` reduced to whether it is present.
+- **Native.** `http.rs` validates both polls with the published rules before
+  reading them (native's task-kinds reader still checks distinct values),
+  and answers an unknown activity task on the fail and heartbeat routes as
+  the complete route already did. In a named namespace, a request PHP has
+  no route for used to be refused as an unqualified named-namespace
+  operation (`422 development_namespace_operation_unqualified`); it now gets
+  PHP's 404 or 405, as in the default namespace. Only operations PHP serves
+  are still refused that way. A workflow task completion or checkpoint now
+  looks the task up before checking or fetching any command payload.
+- **Tests.** `worker_error_bodies.rs` fails on the previous source and
+  passes now. Its last test replays every fixture request against native
+  and compares PHP's recorded answers. `php_value::loose_eq` has its own
+  unit test.

docs/rust-server-port.md

@@ -666,11 +666,6 @@ contract fixture before its correction.
 - **A repeated identical activity result.** Published PHP answers a repeat
   of the result that finished an activity task with `409 stale_attempt`.
   Native returns its earlier idempotent receipt (`200`, `already_completed`).
-- **Worker-plane error bodies.** Published PHP adds `protocol_version` to
-  every worker-plane response, including validation failures, and answers an
-  unknown activity task with `task_id`, `activity_attempt_id`, `error` and
-  `reason`. Native's worker-plane errors carry only Laravel's validation
-  fields or `message` and `reason`.
 - **Authentication beyond role tokens and runtime credentials.** Native
   refuses principal tokens (`DW_PRINCIPAL_TOKENS`), custom providers and the
   signature driver at startup. It still requires `DW_AUTH_TOKEN` with the

rust/src/runtime/execution.rs

@@ -1546,6 +1546,16 @@ impl Runtime {
         protocol: Option<&str>,
         checkpoint: Option<&super::scope_checkpoint::Checkpoint>,
     ) -> Result<Value> {
+        // A task the namespace does not hold is answered before any payload
+        // is read or fetched, as PHP's guards run before its normalizer.
+        let held = match &*self.storage {
+            Storage::Sqlite(store) => store.task_held(task_id).await?,
+            Storage::Postgres(store) => store.task_held(task_id).await?,
+            Storage::MySql(store) => store.task_held(task_id).await?,
+        };
+        if !held {
+            return Err(refuse(StatusCode::NOT_FOUND, "task_not_found"));
+        }
         self.check_command_payloads(&body).await?;
         let external = self.prepare_command_payloads(&body).await?;
         let body = super::completion_rules::canonical_payloads(body)?;

rust/src/runtime/http.rs

@@ -1,3 +1,4 @@
+use super::laravel_rules::Rule;
 use axum::{
     Extension, Json, Router,
     extract::{DefaultBodyLimit, Path, Query, Request, State},
@@ -655,6 +656,13 @@ async fn guard(State(runtime): State<Runtime>, mut request: Request, next: Next)
             return admission_error(request, worker, StatusCode::NOT_FOUND, body).await;
         }
         if namespace != "default" && !named_namespace_operation(request.method().as_str(), &path) {
+            // A request PHP has no route for gets PHP's 404 or 405 in any
+            // namespace; only an operation PHP serves is refused here.
+            if !matches!(super::route_fallback::answer(request.method().as_str(), &path),
+                super::route_fallback::Answer::NotImplemented)
+            {
+                return route_fallback(request.method().clone(), request.uri().clone()).await;
+            }
             return admission_error(request,worker,StatusCode::UNPROCESSABLE_ENTITY,
                 json!({"reason":"development_namespace_operation_unqualified","namespace":namespace,
                     "message":"This operation is not implemented for named namespaces in the unpublished execution slice."})).await;
@@ -2088,13 +2096,43 @@ async fn deregister_fenced(
         }
     }
 }
+/// `WorkerController::pollWorkflowTasks`' request rules.
+const WORKFLOW_POLL_RULES: &super::laravel_rules::Rules = &[
+    ("worker_id", &[Rule::Required, Rule::String]),
+    ("task_queue", &[Rule::Required, Rule::String]),
+    ("build_id", &[Rule::Nullable, Rule::String]),
+    ("poll_request_id", &[Rule::Nullable, Rule::String, Rule::Max(255)]),
+    ("timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(0), Rule::Max(60)]),
+    ("history_page_size", &[Rule::Nullable, Rule::Integer, Rule::Min(1), Rule::Max(1000)]),
+    ("accept_history_encoding", &[Rule::Nullable, Rule::String, Rule::Max(64)]),
+    ("task_kinds", &[Rule::Nullable, Rule::Array, Rule::Min(1), Rule::Max(2)]),
+    ("task_kinds.*", &[Rule::String, Rule::Distinct, Rule::In(&["workflow", "update_validation"])]),
+];
+
+/// `ActivityTaskController::poll`'s request rules.
+const ACTIVITY_POLL_RULES: &super::laravel_rules::Rules = &[
+    ("worker_id", &[Rule::Required, Rule::String]),
+    ("task_queue", &[Rule::Required, Rule::String]),
+    ("build_id", &[Rule::Nullable, Rule::String]),
+    ("poll_request_id", &[Rule::Nullable, Rule::String, Rule::Max(255)]),
+    ("timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(0), Rule::Max(60)]),
+];
+
+/// A poll's request rules, answered as PHP's worker plane answers a
+/// validation failure: in the worker envelope.
+fn poll_rules(body: &Value, rules: &super::laravel_rules::Rules) -> Result<()> {
+    super::laravel_rules::validate(&super::laravel_rules::normalize(body), rules).map_err(worker_refusal)
+}
+
 async fn poll_workflow(
     Extension(runtime): Extension<Runtime>,
     headers: HeaderMap,
     Json(body): Json<Value>,
 ) -> Result<Json<Value>> {
+    let protocol = request_protocol(&headers)?;
+    poll_rules(&body, WORKFLOW_POLL_RULES)?;
     runtime
-        .poll_workflow_tasks(body, request_protocol(&headers)?)
+        .poll_workflow_tasks(body, protocol)
         .await
         .map(Json)
         .map_err(worker_refusal)
@@ -2139,9 +2177,11 @@ async fn poll_activity(
     headers: HeaderMap,
     Json(body): Json<Value>,
 ) -> Result<Json<Value>> {
+    let protocol = request_protocol(&headers)?;
+    poll_rules(&body, ACTIVITY_POLL_RULES)?;
     let wait = runtime.worker_poll_wait();
     runtime
-        .poll_with_pages(body, "activity", request_protocol(&headers)?, Default::default(), wait)
+        .poll_with_pages(body, "activity", protocol, Default::default(), wait)
         .await
         .map(Json)
 }
@@ -2380,12 +2420,26 @@ async fn heartbeat_activity(
     headers: HeaderMap,
     Json(body): Json<Value>,
 ) -> Result<Json<Value>> {
+    let attempt = body["activity_attempt_id"].clone();
     runtime
         .heartbeat_activity(&task_id, body, request_protocol(&headers)?)
         .await
+        .map_err(|error| activity_not_found(error, &task_id, attempt))
         .map(Json)
 }
 
+/// `guardAttemptOwnership`'s answer for a task the namespace does not hold,
+/// in the worker envelope, as the fail and heartbeat routes give it.
+fn activity_not_found(error: super::RuntimeError, task_id: &str, attempt: Value) -> super::RuntimeError {
+    match error {
+        super::RuntimeError::Refused { status, reason: "task_not_found" } if status == StatusCode::NOT_FOUND => {
+            super::RuntimeError::Protocol { status, response: worker_response(json!({"task_id":task_id,
+                "activity_attempt_id":attempt,"error":"Activity task not found.","reason":"task_not_found"})) }
+        }
+        other => other,
+    }
+}
+
 async fn renew_activity_lease(
     Extension(runtime): Extension<Runtime>,
     Path(task_id): Path<String>,
@@ -2470,7 +2524,12 @@ async fn fail_activity(
     Path(task_id): Path<String>,
     Json(body): Json<Value>,
 ) -> Result<Json<Value>> {
-    runtime.fail_activity(&task_id, body).await.map(|body| Json(worker_response(body)))
+    let attempt = body["activity_attempt_id"].clone();
+    runtime
+        .fail_activity(&task_id, body)
+        .await
+        .map_err(|error| activity_not_found(error, &task_id, attempt))
+        .map(|body| Json(worker_response(body)))
 }
 /// The query's command context, as PHP's controller builds it.
 fn query_context(request: &CommandRequest, body: &Value, workflow_id: &str, name: &str) -> Value {