Admit signals before any worker declares a contract, as PHP does

.github/workflows/server-parity.yml

@@ -26,7 +26,7 @@ on:
       - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
       - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeContractTest.mjs
-      - tests/Unit/ServerParityStorageTestContractTest.mjs
+      - tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
@@ -68,7 +68,7 @@ on:
       - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
       - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeContractTest.mjs
-      - tests/Unit/ServerParityStorageTestContractTest.mjs
+      - tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
@@ -1460,7 +1460,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 > 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/ServerParitySignalsBeforeWorkerContractTest.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.md

@@ -229,6 +229,38 @@ support is not evidence of multi-node safety.
 Newest checkpoints first. Each entry is exact to the source, tree and runs it
 names, including unsuccessful observations.
 
+October 21, signals before any worker registers:
+
+- **Found by.** Waterline 2.12.0 service mode was compared against
+  published PHP 2.5.14 and against native with identical seeds. A signal
+  sent to a run that no worker had picked up was accepted by PHP and
+  rejected by native with 404 `unknown_signal`, so the signal was lost.
+- **Cause.** Native records a run's command contract at start only when a
+  worker for the type has registered. It then treated a missing contract as
+  one that declares nothing, and its recorded contracts lacked PHP's
+  `declared_entry_*` fields.
+- **Source.** Admission now follows PHP's `signalAdmissionForRun`:
+  - A signal is unknown only under a complete durable contract.
+  - A run without one admits every signal (`external_contract_unavailable`).
+  - A declared signal without a parameter contract takes its arguments as
+    given.
+  - The runtime message-stream signal name is refused with 422
+    `runtime_reserved_signal` before admission, whatever the run's state,
+    and no command is recorded.
+  - A signal without `input`, or with null input, has no arguments.
+- **Contracts.** Recorded contracts now carry PHP's entry defaults (`handle`,
+  `canonical`, `external:<type>`), so they are complete.
+- **Responses.** Signal responses carry `workflow_type`. The `control_plane`
+  envelope projects exactly PHP's `PROJECTED_FIELDS`, and cancel and
+  terminate responses now carry it.
+- **Evidence.** The new `signals-before-worker` fixture sends five signals,
+  the reserved name before and after termination, and a missing workflow to
+  a run on an unserved queue. Its responses match a reviewed copy of
+  published PHP's, and its history holds the same events. Published PHP,
+  embedded and native passed it locally; native from main failed it.
+- **Next.** Durable rejected updates and PHP's query routing refusals, which
+  the same comparison found.
+
 October 21, takeover refuses schedules native cannot run:
 
 - **Found by.** A review of takeover against native's schedule support.

rust/src/runtime/children.rs

@@ -173,6 +173,11 @@ where
                             started[target] = value.clone();
                         }
                     }
+                    // WorkflowStartService's defaults for an external worker's
+                    // contract, which make it a complete durable contract.
+                    started["declared_entry_method"] = json!("handle");
+                    started["declared_entry_mode"] = json!("canonical");
+                    started["declared_entry_declaring_class"] = json!(format!("external:{workflow_type}"));
                     break;
                 }
             }

rust/src/runtime/control_plane.rs

@@ -226,12 +226,10 @@ impl Operation<'_> {
                 "backend_lock_pressure",
                 "control_plane_internal_error"
             ]);
+            // PHP's ControlPlaneResponseContract::PROJECTED_FIELDS for a signal.
             for field in [
                 "command_id",
                 "target_scope",
-                "resolved_run_id",
-                "requested_run_id",
-                "signal_name",
                 "command_contract_source",
                 "command_contract_backfill_needed",
                 "command_contract_backfill_available",
@@ -243,35 +241,60 @@ impl Operation<'_> {
                 }
             }
         }
-        // Copy scalar read identities/state, never input/output/event payloads.
+        // ControlPlaneResponseContract::PROJECTED_FIELDS: each present field.
         for field in [
             "run_id",
             "workflow_type",
             "namespace",
             "business_key",
             "status",
             "status_bucket",
+            "run_status",
             "is_terminal",
             "task_queue",
             "run_number",
             "run_count",
+            "workflow_count",
             "is_current_run",
             "compatibility",
             "started_at",
             "closed_at",
             "last_progress_at",
             "wait_kind",
             "wait_reason",
-            "workflow_count",
             "next_page_token",
+            "command_id",
             "command_status",
             "command_source",
+            "target_scope",
+            "query_name",
             "outcome",
             "reason",
             "rejection_reason",
+            "rejection_category",
+            "validation_errors",
+            "result",
+            "result_envelope",
+            "update_id",
+            "update_status",
+            "principal",
+            "wait_for",
+            "wait_timed_out",
+            "wait_timeout_seconds",
+            "blocked_reason",
             "message",
             "remediation",
-            "validation_errors",
+            "execution_owner",
+            "identity",
+            "command_contract_source",
+            "command_contract_backfill_needed",
+            "command_contract_backfill_available",
+            "declared_signals",
+            "signal_admission",
+            "diagnostic_status",
+            "retryable",
+            "error_id",
+            "exception",
         ] {
             if let Some(value) = body.get(field) {
                 metadata[field] = value.clone();

rust/src/runtime/execution.rs

@@ -388,7 +388,12 @@ impl Runtime {
         name: &str,
         body: Value,
     ) -> Result<Value> {
-        let arguments = super::envelope(&body, "input")?;
+        // PHP's `input` is nullable: an absent or null input is no arguments.
+        let arguments = if body.get("input").is_none_or(Value::is_null) {
+            super::queries::empty_arguments(self.signal_codec.clone()).await?
+        } else {
+            super::envelope(&body, "input")?
+        };
         let prepared = super::store::prepare_control_arguments(
             self.signal_codec.clone(),
             name.to_owned(),

rust/src/runtime/http.rs

@@ -701,7 +701,7 @@ async fn close(
         .close_workflow(command, workflow_id, run_id, body)
         .await
         .map_err(|error| operation().refusal(error, workflow_id, run_id))?;
-    Ok((status, Json(response)))
+    Ok((status, Json(operation().response(response))))
 }
 
 async fn cancel_current(

rust/src/runtime/store.rs

@@ -15,6 +15,8 @@ pub(super) struct PreparedControlArguments {
     pub(super) value: String,
     pub(super) command: String,
 }
+/// The runtime-owned message-stream delivery signal (`WorkflowStub::MESSAGE_STREAM_RUNTIME_SIGNAL`).
+const MESSAGE_STREAM_SIGNAL: &str = "__durable_workflow_message_stream";
 pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) ORDER BY t.available_at,t.id LIMIT 100";
 pub(super) const POLL_RECEIPT_CLEANUP_SQL: &str = "DELETE FROM dw_poll_receipts WHERE expires_at<=$1 AND task_id IN (SELECT id FROM workflow_tasks WHERE status!='leased' OR lease_expires_at<=$2)";
 pub(super) const WORKER_REGISTRATION_SQL: &str = "INSERT INTO workflow_worker_registrations(namespace,worker_id,task_queue,runtime,sdk_version,build_id,supported_workflow_types,supported_activity_types,capabilities,capability_manifest,last_heartbeat_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT(worker_id,namespace) DO UPDATE SET task_queue=excluded.task_queue,runtime=excluded.runtime,sdk_version=excluded.sdk_version,build_id=excluded.build_id,supported_workflow_types=excluded.supported_workflow_types,supported_activity_types=excluded.supported_activity_types,capabilities=excluded.capabilities,capability_manifest=excluded.capability_manifest,last_heartbeat_at=excluded.last_heartbeat_at,updated_at=excluded.updated_at";
@@ -453,6 +455,16 @@ where
                 "historical_run_command_rejected",
             ));
         }
+        // PHP's control plane refuses the runtime's own message-stream signal
+        // before admission, whatever the run's state, and records nothing.
+        if name == MESSAGE_STREAM_SIGNAL {
+            let message = "This signal name is reserved for the durable runtime transport.";
+            return Err(super::RuntimeError::Protocol {
+                status: StatusCode::UNPROCESSABLE_ENTITY,
+                response: json!({"workflow_id":workflow_id,"signal_name":name,"outcome":"rejected",
+                    "reason":"runtime_reserved_signal","message":message,"accepted":false}),
+            });
+        }
         let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1")
             .bind(&current_run)
             .fetch_one(&mut *tx)
@@ -472,19 +484,30 @@ where
         let declared = contract["declared_signals"]
             .as_array()
             .is_some_and(|names| names.iter().any(|value| value == name));
+        // PHP's signalAdmissionForRun: only a complete durable contract can
+        // make a signal unknown. A run started before any worker declared its
+        // contract admits every signal (external_contract_unavailable), and a
+        // declared signal without a parameter contract takes its arguments
+        // as given.
+        let durable = super::run_observation::command_contract(Some(&contract))["source"] == "durable_history";
+        let has_parameters = contract["declared_signal_contracts"]
+            .as_array()
+            .is_some_and(|contracts| contracts.iter().any(|contract| contract["name"] == name));
         let mut validation_errors = json!([]);
         let rejection = if !active {
             Some((
                 StatusCode::CONFLICT,
                 "run_not_active",
                 "rejected_not_active",
             ))
-        } else if !declared {
+        } else if !declared && durable {
             Some((
                 StatusCode::NOT_FOUND,
                 "unknown_signal",
                 "rejected_unknown_signal",
             ))
+        } else if !has_parameters {
+            None
         } else {
             match validate_signal_arguments(&contract, name, &prepared.values) {
                 Ok(()) => None,
@@ -601,7 +624,7 @@ where
         tx.commit().await?;
         self.wake.notify_waiters();
         Ok(
-            json!({"workflow_id":workflow_id,"run_id":current_run,"resolved_run_id":current_run,"requested_run_id":null,"signal_name":name,"command_id":command_id,"command_sequence":command_sequence,"command_status":"accepted","command_source":"control_plane","target_scope":"instance","outcome":"signal_received","reason":null,"rejection_reason":null,"validation_errors":[]}),
+            json!({"workflow_id":workflow_id,"run_id":current_run,"resolved_run_id":current_run,"requested_run_id":null,"signal_name":name,"command_id":command_id,"command_sequence":command_sequence,"workflow_type":DB::string(&run, "workflow_type")?,"command_status":"accepted","command_source":"control_plane","target_scope":"instance","outcome":"signal_received","reason":null,"rejection_reason":null,"validation_errors":[]}),
         )
     }
 

rust/tests/execution/namespaces.rs

@@ -114,10 +114,9 @@ async fn named_signal_waits_keep_both_original_runs_and_foreign_polls_isolated()
                 accepted.1["control_plane"]["command_id"],
                 accepted.1["command_id"]
             );
-            assert_eq!(
-                accepted.1["control_plane"]["resolved_run_id"],
-                starts[index]["run_id"]
-            );
+            // PHP projects only its PROJECTED_FIELDS into the envelope.
+            assert_eq!(accepted.1["resolved_run_id"], starts[index]["run_id"]);
+            assert!(accepted.1["control_plane"].get("resolved_run_id").is_none());
             assert_eq!(accepted.1["control_plane"]["operation_name"], "data");
             command_ids.push(accepted.1["command_id"].as_str().unwrap().to_owned());
             let ready = scoped(&app, namespaces[index], "GET", &history_path, Value::Null).await;

rust/tests/execution/selected_run_signals.rs

@@ -244,7 +244,8 @@ async fn selected_run_guards_and_durable_rejections_preserve_original_execution(
     let accepted = scoped(&app, "alpha", "POST", &path, json!({"input":input})).await;
     assert_eq!(accepted.0, StatusCode::ACCEPTED, "{}", accepted.1);
     assert_eq!(accepted.1["command_sequence"], 4);
-    assert_eq!(accepted.1["control_plane"]["resolved_run_id"], run);
+    assert_eq!(accepted.1["resolved_run_id"], run);
+    assert!(accepted.1["control_plane"].get("resolved_run_id").is_none());
     let resumed = scoped(
         &app,
         "alpha",

rust/tests/execution/worker_contracts.rs

@@ -171,3 +171,45 @@ async fn invalid_declarations_refuse_before_registration_rotation_or_lease_relea
     drop(app);
     database.remove().await;
 }
+
+async fn signal(app: &Router, workflow: &Value, name: &str, body: Value) -> (StatusCode, Value) {
+    let path = format!("/api/workflows/{}/signal/{name}", workflow["workflow_id"].as_str().unwrap());
+    request(app, "POST", &path, body).await
+}
+
+#[tokio::test]
+async fn signals_follow_php_admission_with_and_without_a_recorded_contract() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    let input = json!({"input":envelope(Payload::Array(vec![Payload::String("x".into())]))});
+    // No worker has declared a contract: PHP admits every signal.
+    let unregistered = start(&app, "before-any-worker").await;
+    for body in [input.clone(), json!({}), json!({"input":null})] {
+        let admitted = signal(&app, &unregistered, "anything", body).await;
+        assert_eq!((admitted.0, admitted.1["outcome"].clone()), (StatusCode::ACCEPTED, json!("signal_received")), "{}", admitted.1);
+        assert_eq!(admitted.1["workflow_type"], "echo");
+    }
+    let reserved = signal(&app, &unregistered, "__durable_workflow_message_stream", input.clone()).await;
+    assert_eq!(reserved.0, StatusCode::UNPROCESSABLE_ENTITY);
+    assert_eq!((reserved.1["reason"].clone(), reserved.1["accepted"].clone()), (json!("runtime_reserved_signal"), json!(false)));
+    assert!(reserved.1.get("command_id").is_none(), "{}", reserved.1);
+    let history = run_history(&app, &unregistered["workflow_id"], &unregistered["run_id"]).await;
+    let received = history.as_array().unwrap().iter().filter(|event| event["event_type"] == "SignalReceived").count();
+    assert_eq!(received, 3);
+    // A registered contract is complete and durable, as PHP records it.
+    let registered = register_contract(&app, Some(json!({"echo":{"signals":["a"]}}))).await;
+    assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1);
+    let declared = start(&app, "after-a-worker").await;
+    let started = snapshot(&app, &declared).await;
+    assert_eq!(
+        (started["declared_entry_method"].clone(), started["declared_entry_mode"].clone(), started["declared_entry_declaring_class"].clone()),
+        (json!("handle"), json!("canonical"), json!("external:echo"))
+    );
+    let unknown = signal(&app, &declared, "b", input.clone()).await;
+    assert_eq!((unknown.0, unknown.1["reason"].clone()), (StatusCode::NOT_FOUND, json!("unknown_signal")), "{}", unknown.1);
+    // A declared signal without a parameter contract takes its arguments as given.
+    let untyped = signal(&app, &declared, "a", input).await;
+    assert_eq!(untyped.0, StatusCode::ACCEPTED, "{}", untyped.1);
+    drop(app);
+    database.remove().await;
+}