Refuse a lost workflow task claim with PHP's ownership guard fields

.github/workflows/server-parity.yml

@@ -24,6 +24,7 @@ on:
       - tests/Unit/ServerParityAdmissionContractTest.mjs
       - tests/Unit/ServerParityRoleTokenContractTest.mjs
       - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
+      - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeContractTest.mjs
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
@@ -64,6 +65,7 @@ on:
       - tests/Unit/ServerParityAdmissionContractTest.mjs
       - tests/Unit/ServerParityRoleTokenContractTest.mjs
       - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
+      - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeContractTest.mjs
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
@@ -1426,7 +1428,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 > 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 > 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

@@ -121,6 +121,12 @@ contract fixture before its correction.
   above the request timeout, still inside the 9-second load wave, needs its own
   reviewed profile change.
 
+- **A valid workflow task claim held by a stale worker.** Published PHP
+  refuses a request whose lease is otherwise valid when its owner's
+  registration is no longer fresh (`409 stale_worker_registration`). Native's
+  workflow task guard does not check registration freshness yet, and the
+  local activity and cancellation scope checkpoints still use the older fence.
+  Add a fixture for each before changing them.
 - **History retention.** Published PHP's worker heartbeat runs an inline
   history retention pass, and the default namespace retention is bounded at
   30 days, so PHP prunes old closed runs. Native has no retention yet. Its
@@ -223,6 +229,31 @@ 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 20, lost workflow task claims:
+
+- **Found by.** The [multi-node runs](../benchmarks/server-port/multinode/results-2026-10-20.md).
+  A paused SDK worker resumed with workflow tasks the server had already
+  given to the other worker. Native refused its late renewals and completion
+  with only a reason (`task_not_leased`, `conflicting_completion`), and the
+  worker exited 6 times. Published PHP's ownership guard also returns the task
+  status and, after the run closes, `can_continue: false`. A current worker
+  reads those fields to drop a lost claim and keep running.
+- **Source.** Workflow task completion, failure, lease renewal and task history
+  now share PHP's guard (`WorkflowTaskOwnership::guard`), in PHP's order.
+  A task that is no longer leased is refused with `run_closed` when its run
+  has closed, and otherwise with `task_not_leased` and its status. An open
+  lease is then checked for expiry, owner and attempt. Each refusal has PHP's
+  body: `error`, the reason's fields, `protocol_version` and
+  `server_capabilities`. A completion that differs from the recorded one now
+  gets this refusal instead of `conflicting_completion`. An identical replay
+  still returns the recorded `already_completed` receipt. A completed workflow
+  task now clears its lease expiry and keeps its owner, as PHP's does.
+- **Fixture.** `late-workflow-completion` replaces a worker that holds a
+  claim and sends the replaced claim's completion, renewal and failure while
+  the new claim holds the lease, after it completes, and after the run
+  closes. Published PHP 2.5.14, embedded and native passed in a local copy of
+  the SQLite parity step. Before the change, native failed the same fixture.
+
 October 20, search attribute registry:
 
 - **Source.** Native serves `GET` and `POST /api/search-attributes` and

rust/src/runtime/mod.rs

@@ -69,6 +69,7 @@ mod worker_contracts;
 mod worker_incarnations;
 mod worker_management;
 mod workflow_failures;
+mod workflow_task_guard;
 
 pub use execution::Runtime;
 pub use external_config::ExternalPayloadConfig;

rust/src/runtime/store.rs

@@ -1309,10 +1309,10 @@ where
     pub(crate) async fn heartbeat_task(&self, task_id: &str, body: Value) -> Result<Value> {
         let mut tx = self.begin().await?;
         let task = self.task_row(&mut tx, task_id).await?;
-        Self::fence(&task, &body)?;
         if DB::string(&task, "task_type")? != "workflow" {
             return Err(refuse(StatusCode::NOT_FOUND, "task_not_found"));
         }
+        Self::workflow_task_guard(&mut tx, &task, &body).await?;
         let run = Self::active_run(&mut tx, &DB::string(&task, "workflow_run_id")?).await?;
         let expires = Self::cancellation_lease_expiry(&run)?;
         tx.summary_changed(&DB::string(&run, "id")?);
@@ -1337,7 +1337,7 @@ where
         if DB::string(&task, "task_type")? != "workflow" {
             return Err(refuse(StatusCode::NOT_FOUND, "task_not_found"));
         }
-        Self::fence(&task, &body)?;
+        Self::workflow_task_guard(&mut tx, &task, &body).await?;
         let run_id = DB::string(&task, "workflow_run_id")?;
         let (events, next) = Self::history_page_connection(&mut tx, &run_id, after, 100).await?;
         let run = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
@@ -1435,9 +1435,9 @@ where
         if DB::string(&task, "task_type")? != "workflow" {
             return Err(refuse(StatusCode::NOT_FOUND, "task_not_found"));
         }
+        Self::workflow_task_guard(&mut tx, &task, &body).await?;
         let run_id = DB::string(&task, "workflow_run_id")?;
         let run = Self::active_run(&mut tx, &run_id).await?;
-        Self::fence(&task, &body)?;
         let mut payload = DB::document_row(&task, "payload")?;
         let signals = Self::scalar("SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$1 AND status='received'")
             .bind(&run_id).fetch_one(&mut *tx).await?;
@@ -1708,7 +1708,17 @@ where
         if DB::string(&task, "task_type")? != "workflow" {
             return Err(refuse(StatusCode::NOT_FOUND, "task_not_found"));
         }
-        if Self::duplicate_receipt(&task, &body)? {
+        // A completion that differs from the recorded one, or follows a
+        // waiting acknowledgement, comes from a lost claim: the guard refuses
+        // it with the fields a worker uses to recognize that.
+        let replay = match Self::duplicate_receipt(&task, &body) {
+            Err(super::RuntimeError::Refused {
+                reason: "conflicting_completion" | "task_not_leased",
+                ..
+            }) => false,
+            other => other?,
+        };
+        if replay {
             let run_id = DB::string(&task, "workflow_run_id")?;
             let current = Self::query("SELECT status FROM workflow_runs WHERE id=$1")
                 .bind(&run_id)
@@ -1719,7 +1729,7 @@ where
                 "outcome": "completed", "recorded": false, "reason": "already_completed"}),
             );
         }
-        Self::fence(&task, &body)?;
+        Self::workflow_task_guard(&mut tx, &task, &body).await?;
         let run_id = DB::string(&task, "workflow_run_id")?;
         let run = Self::active_run(&mut tx, &run_id).await?;
         let task_payload = DB::document_row(&task, "payload")?;
@@ -2050,7 +2060,8 @@ where
         } else {
             false
         };
-        Self::query("UPDATE workflow_tasks SET status=$1 WHERE id=$2")
+        // PHP clears a completed task's lease expiry and keeps its owner.
+        Self::query("UPDATE workflow_tasks SET status=$1,lease_expires_at=NULL WHERE id=$2")
             .bind(if failed { "failed" } else { "completed" })
             .bind(task_id)
             .execute(&mut *tx)
@@ -2476,6 +2487,86 @@ where
         Err(refuse(StatusCode::CONFLICT, "conflicting_completion"))
     }
 
+    /// PHP's workflow task ownership guard, in PHP's order: a task that is
+    /// no longer leased first (run_closed when its run has closed), then the
+    /// lease's expiry, owner and attempt.
+    pub(super) async fn workflow_task_guard(
+        tx: &mut Transaction<'_, DB>,
+        task: &DB::Row,
+        body: &Value,
+    ) -> Result<()> {
+        use super::query_authority::wire;
+        let lease_owner = text(body, "lease_owner")?;
+        let task_id = DB::string(task, "id")?;
+        let attempt = body["workflow_task_attempt"]
+            .as_i64()
+            .map_or(Value::Null, Value::from);
+        let refused = |error: &str, reason: &str, fields: Value| {
+            super::workflow_task_guard::refusal(&task_id, &attempt, error, reason, fields)
+        };
+        let status = DB::string(task, "status")?;
+        let owner = DB::optional_string(task, "lease_owner")?
+            .map(|owner| owner.trim().to_owned())
+            .filter(|owner| !owner.is_empty());
+        let expires = DB::optional_instant(task, "lease_expires_at")?;
+        if status != "leased" {
+            let run =
+                Self::query("SELECT status,closed_reason,closed_at FROM workflow_runs WHERE id=$1")
+                    .bind(DB::string(task, "workflow_run_id")?)
+                    .fetch_one(&mut **tx)
+                    .await?;
+            let run_status = DB::string(&run, "status")?;
+            if super::run_observation_store::terminal(&run_status) {
+                let stop_reason = match run_status.as_str() {
+                    "cancelled" => "run_cancelled",
+                    "terminated" => "run_terminated",
+                    "completed" => "run_completed",
+                    _ => "run_failed",
+                };
+                return Err(refused(
+                    "Workflow run is already closed.",
+                    "run_closed",
+                    json!({"stop_reason":stop_reason,"cancel_requested":matches!(run_status.as_str(),"cancelled"|"terminated"),
+                        "can_continue":false,"run_status":run_status,
+                        "run_closed_reason":DB::optional_string(&run,"closed_reason")?,
+                        "run_closed_at":DB::optional_instant(&run,"closed_at")?.map(wire),
+                        "task_status":status,"lease_owner":owner,"lease_expires_at":expires.map(wire)}),
+                ));
+            }
+        }
+        let (true, Some(owner), Some(expires)) = (status == "leased", owner, expires) else {
+            return Err(refused(
+                "Workflow task is not currently leased.",
+                "task_not_leased",
+                json!({"task_status":status}),
+            ));
+        };
+        // An expired lease is claimed again by the next poll.
+        if expires <= now() {
+            return Err(refused(
+                "Workflow task lease has expired and is waiting for recovery.",
+                "lease_expired",
+                json!({"task_status":"leased","lease_owner":owner,"lease_expires_at":wire(expires)}),
+            ));
+        }
+        if lease_owner != owner {
+            return Err(refused(
+                "Workflow task lease is owned by another worker.",
+                "lease_owner_mismatch",
+                json!({"lease_owner":owner}),
+            ));
+        }
+        let current = DB::number(task, "attempt_count")?;
+        if current > 0 && attempt.as_i64() != Some(current) {
+            return Err(refused(
+                "Workflow task lease attempt does not match the current claim.",
+                "workflow_task_attempt_mismatch",
+                json!({"current_attempt":current}),
+            ));
+        }
+        Ok(())
+    }
+
     pub(super) fn fence(task: &DB::Row, body: &Value) -> Result<()> {
         if DB::string(task, "status")? != "leased" {
             return Err(refuse(StatusCode::CONFLICT, "task_not_leased"));

rust/src/runtime/workflow_task_guard.rs

@@ -0,0 +1,26 @@
+//! Refusal bodies of PHP's workflow task ownership guard
+//! (WorkflowTaskOwnership::guard, rendered by WorkerController). A worker
+//! reads these fields, not just the reason, to tell that its claim was lost:
+//! the task's status, whether the run can continue, and who holds the lease
+//! now. `Store::workflow_task_guard` decides which refusal applies.
+use super::RuntimeError;
+use axum::http::StatusCode;
+use serde_json::{Value, json};
+
+pub(super) fn refusal(
+    task_id: &str,
+    attempt: &Value,
+    error: &str,
+    reason: &str,
+    fields: Value,
+) -> RuntimeError {
+    let mut response = json!({"task_id":task_id,"workflow_task_attempt":attempt,"error":error,"reason":reason,
+        "protocol_version":"1.20","server_capabilities":super::store::capabilities()});
+    if let (Some(response), Value::Object(fields)) = (response.as_object_mut(), fields) {
+        response.extend(fields);
+    }
+    RuntimeError::Protocol {
+        status: StatusCode::CONFLICT,
+        response,
+    }
+}

rust/tests/execution.rs

@@ -37,6 +37,8 @@ mod role_tokens;
 #[path = "execution/namespaces.rs"]
 mod namespaces;
 
+#[path = "execution/late_workflow_claims.rs"]
+mod late_workflow_claims;
 #[path = "execution/search_attributes.rs"]
 mod search_attributes;
 

rust/tests/execution/continuations.rs

@@ -485,7 +485,10 @@ async fn continuation_commits_one_successor_and_preserves_original_receipts_afte
     conflicting["commands"][0]["arguments"] = envelope(Payload::Array(vec![Payload::Long(8)]));
     let refused = request(&app, "POST", &path, conflicting).await;
     assert_eq!(refused.0, StatusCode::CONFLICT);
-    assert_eq!(refused.1["reason"], "conflicting_completion");
+    // PHP's ownership guard: the original run closed when it continued.
+    assert_eq!(refused.1["reason"], "run_closed", "{}", refused.1);
+    assert_eq!(refused.1["can_continue"], false);
+    assert_eq!(refused.1["task_status"], "completed");
     let runs = request(
         &app,
         "GET",