Return a query worker's plain result like PHP

docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md

@@ -0,0 +1,26 @@
+# October 22, query worker results
+
+- **Found by.** The Shared Server fixtures workflow on main. After #626, its
+  `mysql-execution` and `postgres-storage` jobs still failed in the SDK query
+  directions profile (main run 39704552600). The PHP-worker cases differed in
+  `control_plane.result`:
+  - native's wire response had `null`;
+  - the published Python client's hydrated response had the query state.
+- **Cause.** Published PHP SDK workers complete a query task with both the
+  plain `result` and its `result_envelope`. Published PHP Server's
+  `completeQueryTask` accepts both and stores both. Its answer returns the
+  plain `result` beside the envelope, and the control-plane envelope copies
+  it. Native required the envelope, ignored the plain `result`, and always
+  answered `result: null`.
+- **Source.**
+  - **Storage.** A completed query task keeps the worker's plain `result`.
+  - **Answer.** The answer returns it, and its control-plane envelope copies
+    it, as PHP does.
+  - **Plain-only completions.** A completion with a plain `result` and no
+    envelope is accepted, as PHP's rules allow. A completion with neither is
+    still refused.
+  - **Budget.** The plain result counts toward the query result budget.
+- **Verification.** A new integration test completes one query with a plain
+  result and an envelope, and one with only a plain result. Both answers
+  carry the plain result at the top level and in `control_plane`. The query
+  tests pass on SQLite, PostgreSQL and MySQL.

rust/src/runtime/execution.rs

@@ -750,20 +750,25 @@ impl Runtime {
         if let Some(receipt) = receipt {
             return Ok(receipt);
         }
-        let result = super::envelope(&body, "result_envelope")?;
-        if result.len() > super::queries::RESULT_BYTES {
-            return Err(refuse(
-                StatusCode::PAYLOAD_TOO_LARGE,
-                "query_result_exceeds_budget",
-            ));
+        // A plain `result` without an envelope needs no decoding.
+        let plain_only = body.get("result").is_some_and(|value| !value.is_null())
+            && body.get("result_envelope").is_none_or(Value::is_null);
+        if !plain_only {
+            let result = super::envelope(&body, "result_envelope")?;
+            if result.len() > super::queries::RESULT_BYTES {
+                return Err(refuse(
+                    StatusCode::PAYLOAD_TOO_LARGE,
+                    "query_result_exceeds_budget",
+                ));
+            }
+            super::queries::decode(
+                self.signal_codec.clone(),
+                result,
+                false,
+                "invalid_query_result",
+            )
+            .await?;
         }
-        super::queries::decode(
-            self.signal_codec.clone(),
-            result,
-            false,
-            "invalid_query_result",
-        )
-        .await?;
         self.finish_query(task_id, body, false).await
     }
     pub(crate) async fn finish_query(

rust/src/runtime/queries.rs

@@ -458,7 +458,7 @@ where
             match task["status"].as_str() {
                 Some("completed") => {
                     let mut answer = json!({"workflow_id":workflow_id,"run_id":run_id,"target_scope":"instance",
-                        "query_name":name,"result":null,"result_envelope":task["result_envelope"],"reason":null});
+                        "query_name":name,"result":task["result"],"result_envelope":task["result_envelope"],"reason":null});
                     if task["principal"].is_object() {
                         answer["principal"] = task["principal"].clone();
                     }
@@ -1153,6 +1153,7 @@ where
         let reason: Option<String>;
         let failure;
         let result;
+        let mut plain = None;
         if failed {
             let supplied = &body["failure"];
             if !supplied.is_object() || serde_json::to_vec(supplied)?.len() > 32 * 1024 {
@@ -1171,14 +1172,22 @@ where
             failure = Some(supplied.clone());
             result = None;
         } else {
-            let supplied = envelope(body, "result_envelope")?;
-            if supplied.len() > RESULT_BYTES {
+            // `completeQueryTask`: a worker may also send the plain `result`,
+            // which published PHP returns to the caller beside the envelope.
+            plain = body.get("result").filter(|value| !value.is_null()).cloned();
+            let supplied = if plain.is_none() || body.get("result_envelope").is_some_and(|value| !value.is_null()) {
+                Some(envelope(body, "result_envelope")?)
+            } else {
+                None
+            };
+            let plain_bytes = plain.as_ref().map(serde_json::to_vec).transpose()?.map_or(0, |bytes| bytes.len());
+            if supplied.as_ref().map_or(0, String::len) + plain_bytes > RESULT_BYTES {
                 return Err(refuse(
                     StatusCode::PAYLOAD_TOO_LARGE,
                     "query_result_exceeds_budget",
                 ));
             }
-            result = Some(wire(&supplied));
+            result = supplied.map(|supplied| wire(&supplied));
             failure = None;
             reason = None;
         }
@@ -1196,6 +1205,9 @@ where
         if let Some(result) = result {
             task["result_envelope"] = result;
         }
+        if let Some(plain) = plain {
+            task["result"] = plain;
+        }
         task["terminal_receipt"] = json!({"lease_owner":owner,"attempt":attempt,
             "fingerprint":fingerprint,"decision":decision,"decided_at":authority::wire(decided_at)});
         // Results reserve their slot at admission, and discard the large replay

rust/tests/execution.rs

@@ -79,6 +79,8 @@ mod worker_deregistration;
 mod worker_listing;
 #[path = "execution/activity_failure_fields.rs"]
 mod activity_failure_fields;
+#[path = "execution/query_worker_results.rs"]
+mod query_worker_results;
 #[path = "execution/query_open_tasks.rs"]
 mod query_open_tasks;
 #[path = "execution/retry_message_tasks.rs"]

rust/tests/execution/query_worker_results.rs

@@ -0,0 +1,40 @@
+//! A query worker's plain `result`, which published PHP stores beside the
+//! result envelope and returns to the caller (`completeQueryTask`).
+use super::*;
+
+async fn ask(app: &Router, workflow_id: &str) -> tokio::task::JoinHandle<(StatusCode, Value)> {
+    let app = app.clone();
+    let path = format!("/api/workflows/{workflow_id}/query/state");
+    tokio::spawn(async move {
+        request(&app, "POST", &path, json!({"input":envelope(Payload::Array(vec![Payload::Long(1)]))})).await
+    })
+}
+
+#[tokio::test]
+async fn query_answers_carry_the_worker_s_plain_result() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    query_worker(&app, "plain-query-worker").await;
+    let plain = json!({"received":0,"last":null});
+    for (workflow_id, with_envelope) in [("plain-and-envelope", true), ("plain-only", false)] {
+        start(&app, workflow_id).await;
+        let asked = ask(&app, workflow_id).await;
+        let claim = await_query(&app, "plain-query-worker").await;
+        let mut completion = json!({"lease_owner":"plain-query-worker","query_task_attempt":claim["query_task_attempt"],
+            "result":plain});
+        if with_envelope {
+            completion["result_envelope"] = envelope(Payload::Long(7));
+        }
+        let (status, body) = request(&app, "POST",
+            &format!("/api/worker/query-tasks/{}/complete", claim["query_task_id"].as_str().unwrap()), completion).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        let (status, answer) = asked.await.unwrap();
+        assert_eq!(status, StatusCode::OK, "{answer}");
+        assert_eq!(answer["result"], plain, "{answer}");
+        assert_eq!(answer["control_plane"]["result"], plain, "{answer}");
+        assert_eq!(answer["result_envelope"].is_object(), with_envelope, "{answer}");
+    }
+    runtime.close().await;
+    database.remove().await;
+}