Preserve explicit worker command declarations in native registration

rust/src/runtime/children.rs

@@ -155,6 +155,11 @@ where
             {
                 let contracts = DB::document_row(&worker, "workflow_command_contracts")?;
                 if let Some(contract) = contracts.get(workflow_type) {
+                    // Older development registrations may predate validation.
+                    // Never turn a malformed entry into authoritative empties.
+                    let Ok(contract) = super::worker_contracts::declaration(contract) else {
+                        continue;
+                    };
                     for (source, target) in [
                         ("signals", "declared_signals"),
                         ("signal_contracts", "declared_signal_contracts"),
@@ -164,7 +169,9 @@ where
                         ("update_contracts", "declared_update_contracts"),
                         ("update_validators", "declared_update_validators"),
                     ] {
-                        started[target] = contract[source].clone();
+                        if let Some(value) = contract.get(source) {
+                            started[target] = value.clone();
+                        }
                     }
                     break;
                 }

rust/src/runtime/mod.rs

@@ -63,6 +63,7 @@ pub mod upgrade;
 mod visibility;
 mod wait_projection;
 mod wait_timeouts;
+mod worker_contracts;
 mod worker_incarnations;
 mod workflow_failures;
 

rust/src/runtime/store.rs

@@ -813,27 +813,29 @@ where
             }
         }
         let runtime = text(&body, "runtime")?;
-        let contracts = body
+        let supplied_contracts = body
             .get("workflow_command_contracts")
-            .cloned()
-            .unwrap_or_else(|| json!({}));
-        // PHP encodes an empty associative map as []; activity-only published
-        // workers send this shape. Preserve a map internally for lookups.
-        let contracts = if contracts.as_array().is_some_and(|values| values.is_empty()) {
-            json!({})
-        } else {
-            contracts
-        };
-        if !contracts.is_object() {
-            return Err(refuse(
-                StatusCode::UNPROCESSABLE_ENTITY,
-                "invalid_worker_definition",
-            ));
-        }
+            .map(|contracts| {
+                super::worker_contracts::normalize(contracts, &body["supported_workflow_types"])
+            })
+            .transpose()?;
         let token = super::worker_incarnations::token().await?;
         let mut tx = self.begin().await?;
-        let previous = Self::query("SELECT worker_id FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2")
+        let previous = Self::query("SELECT worker_id,workflow_command_contracts FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2")
             .bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut *tx).await?;
+        let contracts = match supplied_contracts {
+            Some(contracts) => contracts,
+            None => {
+                let existing = match &previous {
+                    Some(worker) => {
+                        DB::optional_document_row(worker, "workflow_command_contracts")?
+                            .unwrap_or(Value::Null)
+                    }
+                    None => Value::Null,
+                };
+                super::worker_contracts::normalize(&existing, &body["supported_workflow_types"])?
+            }
+        };
         if previous.is_some() {
             // Match PHP's registration boundary: return original workflow leases
             // to ready without a repair command/history event. Token rotation

rust/src/runtime/worker_contracts.rs

@@ -0,0 +1,93 @@
+//! Registration declarations become immutable admission snapshots. A missing
+//! workflow entry stays unknown; a valid empty entry declares empty lists.
+use super::{Result, refuse};
+use axum::http::StatusCode;
+use serde_json::{Value, json};
+use std::collections::BTreeSet;
+
+fn invalid() -> super::RuntimeError {
+    refuse(
+        StatusCode::UNPROCESSABLE_ENTITY,
+        "invalid_worker_definition",
+    )
+}
+
+fn name(value: &Value) -> Result<&str> {
+    let text = value.as_str().ok_or_else(invalid)?;
+    let text = text.trim_matches([' ', '\t', '\n', '\r', '\0', '\u{b}']);
+    if text.is_empty() || text.chars().count() > 255 {
+        return Err(invalid());
+    }
+    Ok(text)
+}
+
+fn list<'a>(entry: &'a Value, field: &str) -> Result<&'a [Value]> {
+    match entry.get(field) {
+        None => Ok(&[]),
+        Some(Value::Array(values)) => Ok(values),
+        Some(_) => Err(invalid()),
+    }
+}
+
+pub(super) fn declaration(entry: &Value) -> Result<Value> {
+    if !entry.is_object() && !entry.as_array().is_some_and(Vec::is_empty) {
+        return Err(invalid());
+    }
+    let mut result = json!({});
+    for (names, handlers) in [
+        ("queries", "query_contracts"),
+        ("signals", "signal_contracts"),
+        ("updates", "update_contracts"),
+    ] {
+        let mut declared = BTreeSet::new();
+        for value in list(entry, names)? {
+            declared.insert(name(value)?.to_owned());
+        }
+        let mut seen = BTreeSet::new();
+        let mut contracts = Vec::new();
+        for handler in list(entry, handlers)? {
+            let handler_name = name(&handler["name"])?;
+            declared.insert(handler_name.to_owned());
+            if seen.insert(handler_name.to_owned()) {
+                contracts.push(
+                    json!({"name":handler_name,"parameters":handler.get("parameters")
+                    .filter(|value| value.is_array()).cloned().unwrap_or_else(|| json!([]))}),
+                );
+            }
+        }
+        result[names] = json!(declared);
+        result[handlers] = json!(contracts);
+    }
+    if entry.get("update_validators").is_some() {
+        let mut validators = BTreeSet::new();
+        for value in list(entry, "update_validators")? {
+            validators.insert(name(value)?.to_owned());
+        }
+        result["update_validators"] = json!(validators);
+    }
+    Ok(result)
+}
+
+pub(super) fn normalize(value: &Value, supported: &Value) -> Result<Value> {
+    if value.is_null() || value.as_array().is_some_and(Vec::is_empty) {
+        return Ok(json!({}));
+    }
+    let entries = value.as_object().ok_or_else(invalid)?;
+    let mut result = json!({});
+    for (workflow_type, entry) in entries {
+        // Validate even an unadvertised declaration before filtering it out.
+        // Otherwise malformed metadata could rotate an existing registration.
+        let entry = declaration(entry)?;
+        let workflow_type = workflow_type.trim_matches([' ', '\t', '\n', '\r', '\0', '\u{b}']);
+        if !workflow_type.is_empty()
+            && supported.as_array().is_some_and(|types| {
+                types
+                    .iter()
+                    .any(|value| value.as_str() == Some(workflow_type))
+            })
+        {
+            result[workflow_type] = entry;
+        }
+    }
+    Ok(result)
+}

rust/tests/execution.rs

@@ -49,6 +49,9 @@ mod selected_run_signals;
 #[path = "execution/worker_incarnations.rs"]
 mod worker_incarnations;
 
+#[path = "execution/worker_contracts.rs"]
+mod worker_contracts;
+
 #[path = "execution/worker_deregistration.rs"]
 mod worker_deregistration;
 

rust/tests/execution/worker_contracts.rs

@@ -0,0 +1,173 @@
+use super::*;
+
+fn definition() -> Value {
+    json!({"worker_id":"contract-worker","task_queue":"test","runtime":"python",
+        "supported_workflow_types":["echo"],"supported_activity_types":[]})
+}
+
+async fn register_contract(app: &Router, declaration: Option<Value>) -> (StatusCode, Value) {
+    let mut body = definition();
+    if let Some(declaration) = declaration {
+        body["workflow_command_contracts"] = declaration;
+    }
+    request(app, "POST", "/api/worker/register", body).await
+}
+
+async fn snapshot(app: &Router, started: &Value) -> Value {
+    let history = run_history(app, &started["workflow_id"], &started["run_id"]).await;
+    history
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|event| event["event_type"] == "WorkflowStarted")
+        .unwrap()["payload"]
+        .clone()
+}
+
+fn assert_empty(payload: &Value) {
+    for field in [
+        "declared_queries",
+        "declared_signals",
+        "declared_updates",
+        "declared_query_contracts",
+        "declared_signal_contracts",
+        "declared_update_contracts",
+    ] {
+        assert_eq!(payload.get(field), Some(&json!([])), "{field}: {payload}");
+    }
+}
+
+#[tokio::test]
+async fn explicit_empty_declarations_survive_cold_reads_and_later_registration() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let mut originals = Vec::new();
+    for (index, declaration) in [
+        json!({}),
+        json!([]),
+        json!({"queries":[],"signals":[],
+        "updates":[],"query_contracts":[],"signal_contracts":[],"update_contracts":[]}),
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let registered = register_contract(&app, Some(json!({"echo":declaration}))).await;
+        assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1);
+        let started = start(&app, &format!("empty-contract-{index}")).await;
+        let before = snapshot(&app, &started).await;
+        assert_empty(&before);
+        originals.push((started, before));
+    }
+    assert_eq!(
+        register_contract(
+            &app,
+            Some(json!({"echo":{"query_contracts":[
+        {"name":"state","parameters":[]}],"signals":["z","a","z"]}}))
+        )
+        .await
+        .0,
+        StatusCode::CREATED
+    );
+    let later = start(&app, "named-contract").await;
+    let named = snapshot(&app, &later).await;
+    assert_eq!(named["declared_queries"], json!(["state"]));
+    assert_eq!(named["declared_signals"], json!(["a", "z"]));
+    assert_eq!(register_contract(&app, None).await.0, StatusCode::CREATED);
+    let omitted = start(&app, "reregister-omitted").await;
+    assert_eq!(
+        snapshot(&app, &omitted).await["declared_queries"],
+        json!(["state"])
+    );
+    drop(app);
+    drop(runtime);
+    let cold = router(database.open().await.unwrap());
+    for (started, before) in originals {
+        assert_eq!(snapshot(&cold, &started).await, before);
+    }
+    drop(cold);
+    database.remove().await;
+}
+
+#[tokio::test]
+async fn legacy_missing_entries_remain_unknown_instead_of_becoming_empty() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    for (index, declaration) in [
+        None,
+        Some(Value::Null),
+        Some(json!({})),
+        Some(json!([])),
+        Some(json!({"other":{}})),
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let registered = register_contract(&app, declaration).await;
+        assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1);
+        let original = start(&app, &format!("legacy-contract-{index}")).await;
+        let before = snapshot(&app, &original).await;
+        assert!(before.get("declared_queries").is_none(), "{before}");
+        assert_eq!(
+            register_contract(&app, Some(json!({"echo":{}}))).await.0,
+            StatusCode::CREATED
+        );
+        assert_eq!(snapshot(&app, &original).await, before);
+        assert_eq!(
+            register_contract(&app, Some(json!({}))).await.0,
+            StatusCode::CREATED
+        );
+    }
+    drop(app);
+    database.remove().await;
+}
+
+#[tokio::test]
+async fn invalid_declarations_refuse_before_registration_rotation_or_lease_release() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    let accepted = register_contract(&app, Some(json!({"echo":{}}))).await;
+    assert_eq!(accepted.0, StatusCode::CREATED);
+    let started = start(&app, "contract-lease").await;
+    let task = poll(&app, "contract-worker", "workflow").await;
+    let before = snapshot(&app, &started).await;
+    for declaration in [
+        Value::Null,
+        json!("bad"),
+        json!({"queries":null}),
+        json!({"queries":"state"}),
+        json!({"queries":[17]}),
+        json!({"signals":[" "]}),
+        json!({"updates":{"key":"set"}}),
+        json!({"update_validators":null}),
+        json!({"query_contracts":"bad"}),
+        json!({"signal_contracts":[null]}),
+        json!({"query_contracts":[{}]}),
+        json!({"update_contracts":[{"name":17}]}),
+    ] {
+        let refused = register_contract(&app, Some(json!({"echo":declaration}))).await;
+        assert_eq!(refused.0, StatusCode::UNPROCESSABLE_ENTITY, "{}", refused.1);
+        assert_eq!(snapshot(&app, &started).await, before);
+    }
+    let completed = finish_task(
+        &app,
+        &task,
+        json!([{"type":"complete_workflow",
+        "result":envelope(Payload::Long(9007199254740993))}]),
+    )
+    .await;
+    assert_eq!(completed.0, StatusCode::OK, "{}", completed.1);
+    assert_eq!(completed.1["recorded"], true);
+    let retired = request_protocol(
+        &app,
+        "POST",
+        "/api/worker/registrations/contract-worker/deregister",
+        json!({"registration_token":accepted.1["registration_token"]}),
+        "1.20",
+    )
+    .await;
+    assert_eq!(retired.0, StatusCode::OK, "{}", retired.1);
+    assert_eq!(retired.1["recovered_workflow_task_count"], 0);
+    drop(app);
+    database.remove().await;
+}