Echo the worker protocol version in every worker-plane body (#1001)

rust/src/runtime/activity_heartbeats.rs

@@ -154,7 +154,7 @@ where
             "last_heartbeat_at":DB::optional_instant(&attempt,"last_heartbeat_at")?.or(DB::optional_instant(&activity,"last_heartbeat_at")?).map(super::cooperative::timestamp),
             "run_status":run_status,"run_closed_reason":DB::optional_string(&run,"closed_reason")?,"run_closed_at":DB::optional_instant(&run,"closed_at")?.map(super::cooperative::timestamp),
             "activity_status":activity_status,"attempt_status":attempt_status,"task_status":task_status,
-            "deadlines":super::activity_deadlines::deadlines_view::<DB>(&activity)?,"worker_session":worker_session,"protocol_version":protocol,"server_capabilities":super::store::capabilities()});
+            "deadlines":super::activity_deadlines::deadlines_view::<DB>(&activity)?,"worker_session":worker_session,"protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()});
         // A cooperative request that reached this attempt shows its receipt.
         if cancel_requested
             && let Some(receipt) = Self::activity_cancellation_receipt(tx, &run, &DB::string(&activity, "id")?, attempt_id).await?
@@ -233,7 +233,7 @@ where
                 "run_closed_reason":observed["run_closed_reason"],"run_closed_at":observed["run_closed_at"],
                 "heartbeat_recorded":false,"lease_expires_at":observed["lease_expires_at"],
                 "last_heartbeat_at":observed["last_heartbeat_at"],"worker_session":observed["worker_session"],
-                "protocol_version":protocol,"server_capabilities":super::store::capabilities()});
+                "protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()});
             tx.commit().await?;
             return Ok(response);
         }
@@ -288,7 +288,7 @@ where
                             "run_closed_reason":observed["run_closed_reason"],"run_closed_at":observed["run_closed_at"],
                             "heartbeat_recorded":false,"lease_expires_at":observed["lease_expires_at"],
                             "last_heartbeat_at":observed["last_heartbeat_at"],"error":renewal["error"],"worker_session":snapshot,
-                            "worker_session_renewal":renewal,"protocol_version":protocol,"server_capabilities":super::store::capabilities()}) });
+                            "worker_session_renewal":renewal,"protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}) });
                 }
             }
         }
@@ -353,7 +353,7 @@ where
             // PHP formats each time in UTC with microseconds, as status reads do.
             "heartbeat_recorded":true,"lease_expires_at":super::cooperative::timestamp(expires),
             "last_heartbeat_at":super::cooperative::timestamp(at),"worker_session":worker_session,
-            "protocol_version":protocol,"server_capabilities":super::store::capabilities()}),
+            "protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}),
         )
     }
 }

rust/src/runtime/http.rs

@@ -692,12 +692,75 @@ async fn guard(State(runtime): State<Runtime>, mut request: Request, next: Next)
     if response.extensions().get::<super::external_transport::TypedRefusal>().is_some() && (worker || route.is_some()) {
         response = typed_refusal_envelope(response, route, worker).await;
     }
+    if worker {
+        response = worker_protocol_envelope(response).await;
+    }
     response
         .headers_mut()
         .insert(header, HeaderValue::from_static(version));
     response
 }
 
+/// The largest worker answer whose envelope is completed here. Bigger or
+/// unsized bodies pass through unchanged rather than being buffered.
+const WORKER_ENVELOPE_MAX_BYTES: u64 = 64 * 1024 * 1024;
+
+/// PHP's `WorkerProtocol::json`: every worker-plane JSON object answer
+/// carries `protocol_version` and `server_capabilities`, each added only when
+/// the handler left it out (`??=`), so a value already present is never
+/// changed (server#1001). The body's bytes are kept and the missing members
+/// are appended before its closing brace, as PHP appends them; numbers are
+/// never re-serialized.
+async fn worker_protocol_envelope(response: Response) -> Response {
+    use axum::body::HttpBody as _;
+    let json = response
+        .headers()
+        .get(header::CONTENT_TYPE)
+        .and_then(|value| value.to_str().ok())
+        .is_some_and(|value| value.starts_with("application/json"));
+    let size = response.body().size_hint().exact();
+    if !json || !size.is_some_and(|size| size <= WORKER_ENVELOPE_MAX_BYTES) {
+        return response;
+    }
+    let (mut parts, body) = response.into_parts();
+    let Ok(bytes) = axum::body::to_bytes(body, WORKER_ENVELOPE_MAX_BYTES as usize).await else {
+        // An exact-sized in-memory body cannot fail to collect.
+        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"reason":"worker_response_unavailable"}))).into_response();
+    };
+    match complete_worker_envelope(&bytes) {
+        Some(completed) => {
+            parts.headers.remove(header::CONTENT_LENGTH);
+            Response::from_parts(parts, Body::from(completed))
+        }
+        None => Response::from_parts(parts, Body::from(bytes)),
+    }
+}
+
+/// The body with any missing envelope member appended, or `None` when it is
+/// not a JSON object or already has both members.
+fn complete_worker_envelope(bytes: &[u8]) -> Option<Vec<u8>> {
+    // Only the member names are read; values are skipped unparsed.
+    let members: std::collections::HashMap<String, serde::de::IgnoredAny> = serde_json::from_slice(bytes).ok()?;
+    let mut missing = Vec::new();
+    if !members.contains_key("protocol_version") {
+        missing.push(format!("\"protocol_version\":{}", json!(super::WORKER_PROTOCOL_VERSION)));
+    }
+    if !members.contains_key("server_capabilities") {
+        missing.push(format!("\"server_capabilities\":{}", capabilities()));
+    }
+    if missing.is_empty() {
+        return None;
+    }
+    let end = bytes.iter().rposition(|byte| *byte == b'}')?;
+    let mut completed = bytes[..end].to_vec();
+    if !members.is_empty() {
+        completed.push(b',');
+    }
+    completed.extend_from_slice(missing.join(",").as_bytes());
+    completed.extend_from_slice(&bytes[end..]);
+    Some(completed)
+}
+
 /// PHP renders a typed external payload error through `WorkerProtocol` or
 /// `ControlPlaneProtocol`, which add the route's envelope fields.
 async fn typed_refusal_envelope(response: Response, route: Option<Request>, worker: bool) -> Response {
@@ -2894,3 +2957,50 @@ async fn unavailable() -> Response {
     )
     .into_response()
 }
+
+#[cfg(test)]
+mod worker_envelope_tests {
+    use super::*;
+
+    fn complete(text: &str) -> Option<String> {
+        complete_worker_envelope(text.as_bytes()).map(|bytes| String::from_utf8(bytes).unwrap())
+    }
+
+    #[test]
+    fn missing_members_are_appended_and_present_ones_untouched() {
+        let completed = complete(r#"{"task":null,"poll_status":"empty"}"#).unwrap();
+        let body: Value = serde_json::from_str(&completed).unwrap();
+        assert_eq!(body["protocol_version"], super::super::WORKER_PROTOCOL_VERSION);
+        assert_eq!(body["server_capabilities"], capabilities());
+        // The original members keep their exact bytes and order.
+        assert!(completed.starts_with(r#"{"task":null,"poll_status":"empty","protocol_version":"#));
+        // A handler's own server_capabilities is kept; only the version is added.
+        let completed = complete(r#"{"task":null,"server_capabilities":{"own":true}}"#).unwrap();
+        let body: Value = serde_json::from_str(&completed).unwrap();
+        assert_eq!(body["server_capabilities"], json!({"own": true}));
+        assert_eq!(body["protocol_version"], super::super::WORKER_PROTOCOL_VERSION);
+        // A handler's own protocol_version is never replaced.
+        let completed = complete(r#"{"protocol_version":"1.7","x":1}"#).unwrap();
+        let body: Value = serde_json::from_str(&completed).unwrap();
+        assert_eq!(body["protocol_version"], "1.7");
+    }
+
+    #[test]
+    fn complete_bodies_and_non_objects_are_left_alone() {
+        assert_eq!(complete(r#"{"protocol_version":"1.20","server_capabilities":{}}"#), None);
+        for body in [r#"[1,2]"#, r#""text""#, "42", "null", "{not json", ""] {
+            assert_eq!(complete(body), None, "{body}");
+        }
+    }
+
+    #[test]
+    fn numbers_and_whitespace_keep_their_bytes() {
+        // Re-serializing would turn these into 1.1 and a rounded float.
+        let completed = complete("{\"a\":1.10,\"big\":123456789012345678901234567890} \n").unwrap();
+        assert!(completed.starts_with("{\"a\":1.10,\"big\":123456789012345678901234567890,\"protocol_version\":"));
+        assert!(completed.ends_with("} \n"));
+        let completed = complete("{}").unwrap();
+        assert!(completed.starts_with("{\"protocol_version\":"));
+        serde_json::from_str::<Value>(&completed).unwrap();
+    }
+}

rust/tests/execution.rs

@@ -370,6 +370,8 @@
 mod activity_result_repeats;
 #[path = "execution/task_fairness.rs"]
 mod task_fairness;
+#[path = "execution/worker_protocol_envelope.rs"]
+mod worker_protocol_envelope;
 #[path = "execution/child_start_turn.rs"]
 mod child_start_turn;
 
@@ -1125,10 +1127,54 @@
         .unwrap();
     let response = app.clone().oneshot(req).await.unwrap();
     let status = response.status();
+    let header = response
+        .headers()
+        .get("x-durable-workflow-protocol-version")
+        .and_then(|value| value.to_str().ok())
+        .map(str::to_owned);
     let body = to_bytes(response.into_body(), 4 * 1024 * 1024)
         .await
         .unwrap();
-    (status, serde_json::from_slice(&body).unwrap())
+    let body: Value = serde_json::from_slice(&body).unwrap();
+    if path.starts_with("/api/worker/") {
+        worker_envelope_sweep(method, path, status, header.as_deref(), &body);
+    }
+    (status, body)
+}
+
+/// Every worker-plane JSON object answer this suite sees carries the
+/// `protocol_version` body member its header advertises, plus
+/// `server_capabilities`, as PHP's `WorkerProtocol::json` guarantees
+/// (server#1001). With `DW_WORKER_ENVELOPE_SWEEP` set to a file, each
+/// answer's shape is also recorded there, for a before/after comparison.
+fn worker_envelope_sweep(method: &str, path: &str, status: StatusCode, header: Option<&str>, body: &Value) {
+    let Some(members) = body.as_object() else {
+        return;
+    };
+    if let Ok(file) = std::env::var("DW_WORKER_ENVELOPE_SWEEP") {
+        use std::io::Write;
+        let route = path
+            .split('/')
+            .map(|segment| if segment.len() >= 20 && segment.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') { "{id}" } else { segment })
+            .collect::<Vec<_>>()
+            .join("/");
+        let route = route.split('?').next().unwrap_or_default();
+        let mut keys: Vec<&String> = members.keys().collect();
+        keys.sort();
+        let record = json!({"method": method, "route": route, "status": status.as_u16(), "keys": keys,
+            "protocol_version": members.get("protocol_version"),
+            "server_capabilities": members.get("server_capabilities").map(|value| value.to_string())});
+        // Tests run in parallel: one locked write per record keeps lines whole.
+        static SWEEP_FILE: std::sync::Mutex<()> = std::sync::Mutex::new(());
+        let _guard = SWEEP_FILE.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let mut out = std::fs::OpenOptions::new().create(true).append(true).open(file).unwrap();
+        out.write_all(format!("{record}\n").as_bytes()).unwrap();
+    }
+    assert_eq!(header, Some("1.20"), "{method} {path}: worker protocol header");
+    assert_eq!(members.get("protocol_version").and_then(Value::as_str), header,
+        "{method} {path} {status}: body protocol_version must echo the header: {body}");
+    assert!(members.get("server_capabilities").is_some_and(Value::is_object),
+        "{method} {path} {status}: body server_capabilities: {body}");
 }
 
 async fn register(app: &Router, worker: &str, workflows: Value, activities: Value) {

rust/tests/execution/activity_deadlines.rs

@@ -219,7 +219,11 @@ async fn every_authored_deadline_fences_outcomes_and_resolves_once() {
                 if outcome == "heartbeat" {
                     assert_eq!(status.1["can_continue"], false);
                     assert_eq!(status.1["heartbeat_recorded"], false);
-                    assert_eq!(status.1["protocol_version"], "1.19");
+                    // The server's advertised version, not the 1.19 request's:
+                    // PHP 2.11.8's heartbeat answers (ActivityTaskController
+                    // lines 865 and 903) set no protocol_version, so
+                    // WorkerProtocol::json supplies the server's (server#1001).
+                    assert_eq!(status.1["protocol_version"], "1.20");
                 }
             }
         }

rust/tests/execution/pending_limits.rs

@@ -34,6 +34,12 @@ async fn a_pending_count_past_its_limit_refuses_the_whole_completion() {
         // The first command meets a limit of one; the second passes it.
         let (status, body) = finish_task(&app, &task, json!([command(kind), command(kind)])).await;
         assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{kind}: {body}");
+        // The worker envelope PHP's WorkerProtocol::json adds (server#1001),
+        // then the refusal's own members exactly.
+        let mut body = body;
+        let members = body.as_object_mut().unwrap();
+        assert_eq!(members.remove("protocol_version"), Some(json!("1.20")), "{kind}");
+        assert!(members.remove("server_capabilities").is_some_and(|capabilities| capabilities.is_object()), "{kind}");
         assert_eq!(body, json!({"task_id":task["task_id"],"workflow_task_attempt":task["workflow_task_attempt"],
             "outcome":"rejected","recorded":false,"error":format!("Structural limit exceeded: 1 {noun} (limit 1)."),
             "reason":"structural_limit_exceeded","limit_kind":limit_kind,"current_value":1,"configured_limit":1}), "{kind}");

rust/tests/execution/worker_protocol_envelope.rs

@@ -0,0 +1,52 @@
+use super::*;
+
+// Server #1001: like PHP's `WorkerProtocol::json`, every worker-plane JSON
+// answer carries the `protocol_version` body member its header advertises,
+// plus `server_capabilities`. Native 3.0.5 sent only the header on successful
+// polls, so the published CLI's `dw workflow-task:poll` refused the answer.
+
+/// The raw answer to a worker poll, with its protocol header.
+async fn raw_poll(app: &Router, kind: &str, worker: &str) -> (StatusCode, Option<String>, Value) {
+    let request = Request::builder()
+        .method("POST")
+        .uri(format!("/api/worker/{kind}-tasks/poll"))
+        .header("authorization", "Bearer test-token")
+        .header("content-type", "application/json")
+        .header("x-namespace", "default")
+        .header("x-durable-workflow-protocol-version", "1.0")
+        .body(Body::from(json!({"worker_id":worker,"task_queue":"test","timeout_seconds":0}).to_string()))
+        .unwrap();
+    let response = app.clone().oneshot(request).await.unwrap();
+    let status = response.status();
+    let header = response
+        .headers()
+        .get("x-durable-workflow-protocol-version")
+        .and_then(|value| value.to_str().ok())
+        .map(str::to_owned);
+    let body = to_bytes(response.into_body(), 4 * 1024 * 1024).await.unwrap();
+    (status, header, serde_json::from_slice(&body).unwrap())
+}
+
+#[tokio::test]
+async fn leased_and_empty_polls_echo_the_protocol_version_in_the_body() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    register(&app, "envelope-worker", json!(["echo"]), json!(["activity"])).await;
+    start(&app, "envelope-poll").await;
+    for (kind, leased) in [("workflow", true), ("workflow", false), ("activity", false)] {
+        // A 1.0 worker, as the published CLI 2.3.3 sends, gets the server's
+        // advertised version back in both places.
+        let (status, header, body) = raw_poll(&app, kind, "envelope-worker").await;
+        assert_eq!(status, StatusCode::OK, "{kind}: {body}");
+        assert_eq!(body["task"].is_object(), leased, "{kind}: {body}");
+        assert_eq!(header.as_deref(), Some("1.20"), "{kind}");
+        assert_eq!(body["protocol_version"], "1.20", "{kind}: {body}");
+        assert!(body["server_capabilities"].is_object(), "{kind}: {body}");
+        if kind == "workflow" {
+            assert_eq!(body["poll_status"], if leased { "leased" } else { "empty" }, "{kind}: {body}");
+        }
+    }
+    runtime.close().await;
+    database.remove().await;
+}