Native: eager activity dispatch for a poll carried on the workflow task completion

Delivered in Native 3.3.0 and PHP SDK 3.11.0

General reconciled the original acceptance against maintained backend controls and the exact published producer/consumer tuple. Acceptance, publication and scope limits. No benchmark speedup or installed SQL-count measurement is claimed. Earlier NONPASSs remain attributed.

Original requirement and historical evidence

External performance report, topic 9a. The reported timings and proposed fixes have not been independently reproduced. The supplied patches are reference drafts for review and current-source qualification.

Problem

A workflow task completion that schedules an activity is followed by the
same worker's activity poll. That's a second locked transaction, just to
claim the activity the completion created. Temporal calls avoiding it
eager activity dispatch.

Proposed contract

POST /api/worker/workflow-tasks/{task_id}/complete accepts an optional
activity_poll object. It holds the worker's next activity poll request,
with the same fields and rules as POST /api/worker/activity-tasks/poll,
including poll_request_id.
The completion's transaction claims the scheduled activity for that
request ID, inside a savepoint, so a failed claim leaves the completion
intact.
The response carries activity_poll: {"leased": true|false}.
The worker's poll under the same request ID replays the receipt without
the lock.
A poll the rules refuse, or one from a draining or fenced-out worker, is
left to the worker's own poll, which answers the refusal.
Advertised as the workflow_task_completion_activity_poll server
capability.

Reported benchmark evidence

Reporter-supplied cumulative measurements, not an isolated-effect or independently verified performance claim: 200 sequentially started workflows, five sequential 20ms activities or five local activities each, six PHP worker processes, mean of three repetitions, one machine, fresh SQLite/MySQL8.0/PostgreSQL16 databases with default durability. "Five" and "local" below mean last-close wall time in seconds; S/M/P means SQLite/MySQL/PostgreSQL. Preserve regressions and noise in the reported results.

	five	local
slowest backend	21.48 → 19.16	5.71 → 4.87

Acceptance

With a carried poll, the scheduled activity is leased in the
completion's transaction, and the replayed poll answers that lease.
A refused, draining or fenced poll leaves the completion recorded and
the activity ready.
Old workers that send no activity_poll behave as before.
worker-protocol-api.openapi.yaml documents the capability, the
activity_poll request field and the response field, as main documents
claimed_task_ready_age. The reference commit doesn't yet.

Implementation constraints

Carry the poll only when an activity slot is available; eager dispatch must not claim work the worker cannot start.

Dependencies

No new producer dependency is required for source investigation.

Related existing work: #325, #1030.

Draft patches and validation boundary

Current main was independently read as b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. The supplied draft series targets 3d4d4afaf2b24aaa5443eac154e253b011924738; patches are cumulative reference material, not a current-main application or test result. Review and rebase on maintained source, preserve all existing tests and published compatibility, then establish causal controls and actual conformance before shipping. No patch was applied or executed during intake.

Known draft gap / required control: The original supplied draft was reported not to compile against the ready-age claim shape. The updated report now supplies a cumulative rebase onto b8f4dd1 and reports that every commit builds and the tip passes SQLite/MySQL/PostgreSQL suites, with carried claims using claim_in for ready ages. This is reporter-attributed evidence, not an independently reproduced maintainer result. Preserve ready_origin_at/ready_age_ms and qualify the maintained-source integration. The complete versioned OpenAPI/capability update and the separately recorded same-worker/error-classification controls remain required.

The following sanitized unified diffs identify proposed reference material:

0010-t09-Carry-the-next-activity-poll-on-the-workflow-task-co.patch — 12,120 bytes, SHA-256 e9b5a930108a4437694184338d3a93b995f8110c114d66b5b2fc9f6e0c4db8cf.

Current implementation: PR1073 is normally merged as 54af464004d85759e16e939124059449c8052758 / tree 784e5506945c4a6f2072d5005e806e8d8d68f4dd. Its accepted source has62 successful PR checks/one declared skip and48 successful merge-group jobs, with OpenAPI revision63 and the shared carried-poll vectors. The accepted SDK328 source handoff replaces the withdrawn b631 handoff. The automatic main attempts remain NONPASS/runner-blocked after two locked-archive HTTP504 setup failures; this is not a product assertion or published result. No further same-head retry is selected. Next owner/action: Server continues the ordered P0 source train, then owns the coherent published producer and consumer/runner qualification tuple; SDK-PHP owns its accepted consumer source. This issue remains open for published-artifact acceptance. The reported benchmark measurements remain unverified.

Updated reference, November 6

The revised source report changes only its shared introduction; all 24 issue sections are unchanged. The archive adds a Server series rebased onto b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. The sanitized rebased topic9a diff is 13,109 bytes / SHA-256 6da0d07ac1d9d50851f98e930978470a4b0e634df81af343c81bda4d2adaaa54. Earlier attached diffs and their review findings remain historical evidence; this rebase alone does not close this issue or establish a build, performance, security or compatibility result.

Sanitized reference diff 0010-t09-Carry-the-next-activity-poll-on-the-workflow-task-co.patch, part 1/1.

This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

Complete sanitized patch: 12,120 bytes, SHA-256 e9b5a930108a4437694184338d3a93b995f8110c114d66b5b2fc9f6e0c4db8cf. This part: 12,120 bytes, SHA-256 e9b5a930108a4437694184338d3a93b995f8110c114d66b5b2fc9f6e0c4db8cf. Parts concatenate in order; a part boundary is between complete file diffs.

diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
index be36228b..39affb00 100644
--- a/rust/src/runtime/execution.rs
+++ b/rust/src/runtime/execution.rs
@@ -1566,7 +1566,20 @@ impl Runtime {
         worker: Option<super::runtime_credentials::Principal>,
         protocol: Option<&str>,
     ) -> Result<Value> {
-        self.complete_or_checkpoint(task_id, body, worker, protocol, None).await
+        self.complete_or_checkpoint(task_id, body, worker, protocol, None, None).await
+    }
+
+    /// A completion that carries the worker's next activity poll
+    /// (`Store::claim_piggybacked_activity`).
+    pub(crate) async fn complete_workflow_with_activity_poll(
+        &self,
+        task_id: &str,
+        body: Value,
+        worker: Option<super::runtime_credentials::Principal>,
+        protocol: Option<&str>,
+        activity_poll: Option<Value>,
+    ) -> Result<Value> {
+        self.complete_or_checkpoint(task_id, body, worker, protocol, None, activity_poll.as_ref()).await
     }
 
     /// POST …/cancellation-scopes/checkpoint, after its route checks: the
@@ -1579,7 +1592,7 @@ impl Runtime {
         protocol: Option<&str>,
         checkpoint: &super::scope_checkpoint::Checkpoint,
     ) -> Result<Value> {
-        self.complete_or_checkpoint(task_id, body, worker, protocol, Some(checkpoint)).await
+        self.complete_or_checkpoint(task_id, body, worker, protocol, Some(checkpoint), None).await
     }
 
     async fn complete_or_checkpoint(
@@ -1589,6 +1602,7 @@ impl Runtime {
         worker: Option<super::runtime_credentials::Principal>,
         protocol: Option<&str>,
         checkpoint: Option<&super::scope_checkpoint::Checkpoint>,
+        activity_poll: Option<&Value>,
     ) -> Result<Value> {
         // A task the namespace does not hold is answered before any payload
         // is read or fetched, as PHP's guards run before its normalizer.
@@ -1647,17 +1661,17 @@ impl Runtime {
         match &*self.storage {
             Storage::Sqlite(store) => {
                 store
-                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref())
+                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref(), activity_poll)
                     .await
             }
             Storage::Postgres(store) => {
                 store
-                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref())
+                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref(), activity_poll)
                     .await
             }
             Storage::MySql(store) => {
                 store
-                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref())
+                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref(), activity_poll)
                     .await
             }
         }
diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
index 628cc2bb..1259f739 100644
--- a/rust/src/runtime/http.rs
+++ b/rust/src/runtime/http.rs
@@ -2394,12 +2394,20 @@ async fn complete_workflow(
     request: CommandRequest,
     Path(task_id): Path<String>,
     headers: HeaderMap,
-    Json(body): Json<Value>,
+    Json(mut body): Json<Value>,
 ) -> Result<Json<Value>> {
+    // The worker's next activity poll, carried by the completion so its
+    // transaction claims the activity the commands schedule. It is not part
+    // of the completion's request: a poll the rules refuse is left to the
+    // worker's own poll, which answers the refusal.
+    let activity_poll = body
+        .as_object_mut()
+        .and_then(|body| body.remove("activity_poll"))
+        .filter(|poll| poll.is_object() && poll_rules(poll, ACTIVITY_POLL_RULES).is_ok());
     let body = completion_checks(&task_id, &headers, body)?;
     let attempt = body["workflow_task_attempt"].clone();
     runtime
-        .complete_workflow(&task_id, body, Some(request.principal().clone()), headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok()))
+        .complete_workflow_with_activity_poll(&task_id, body, Some(request.principal().clone()), headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok()), activity_poll)
         .await
         .map_err(|error| {
             completion_refusal(
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index a5c4d037..fa00a0d4 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -1947,6 +1947,18 @@ where
             super::worker_poll_fence::PollAdmission::Fenced(fence) => Some(fence),
             super::worker_poll_fence::PollAdmission::Unregistered => None,
         };
+        // An activity poll that a workflow task completion claimed for, or a
+        // retry, answers from its receipt before the lock.
+        if kind == "activity"
+            && fence.is_some()
+            && let Some(request_id) = body["poll_request_id"].as_str()
+            && let Some(task) = self.replay_activity_poll(text(&body, "worker_id")?, request_id).await?
+        {
+            return Ok(activity_server_time(
+                kind,
+                json!({"task": task, "poll_status": "leased", "server_capabilities": capabilities()}),
+            ));
+        }
         // A workflow poll's request ID replays its last deliverable answer,
         // after the registration checks that come first in PHP. The first
         // probe replays it (`replay_or_claim`).
@@ -2945,6 +2957,7 @@ where
         limits: super::memo_limits::StructuralLimits,
         checkpoint: Option<&super::scope_checkpoint::Checkpoint>,
         signal: Option<&super::signal_payloads::ResolvedSignal>,
+        activity_poll: Option<&Value>,
     ) -> Result<Value> {
         let transition_codec_for_streams = transition_codec.clone();
         // `commands` is required and non-empty, with no count limit of its
@@ -3099,6 +3112,10 @@ where
                 }
             }
         }
+        let activity_poll = match (activity_poll, protocol) {
+            (Some(poll), Some(protocol)) => self.admit_piggybacked_activity_poll(poll).await.map(|fence| (poll, fence, protocol)),
+            _ => None,
+        };
         let mut tx = self.begin().await?;
         let task = self.task_row(&mut tx, task_id).await?;
         if DB::string(&task, "task_type")? != "workflow" {
@@ -3414,11 +3431,77 @@ where
                 .await?;
         }
         tx.summary_changed(&run_id);
+        let activity_poll = match activity_poll {
+            Some((poll, fence, protocol)) => self.claim_piggybacked_activity(&mut tx, poll, &fence, protocol).await?,
+            None => None,
+        };
         tx.commit().await?;
         self.wake.notify_waiters();
-        Ok(
-            json!({"task_id": task_id, "run_id": run_id, "run_status": run_status, "workflow_task_attempt": body["workflow_task_attempt"], "outcome": "completed", "recorded": true, "reason": null, "created_task_ids": created}),
-        )
+        let mut response = json!({"task_id": task_id, "run_id": run_id, "run_status": run_status, "workflow_task_attempt": body["workflow_task_attempt"], "outcome": "completed", "recorded": true, "reason": null, "created_task_ids": created});
+        if let Some(leased) = activity_poll {
+            response["activity_poll"] = json!({"leased": leased});
+        }
+        Ok(response)
+    }
+
+    /// The poll checks a carried activity poll takes before the completion's
+    /// transaction, as `poll_with_pages` takes them: its registration's
+    /// fence, or `None` when the worker's own poll is to answer it instead.
+    async fn admit_piggybacked_activity_poll(&self, poll: &Value) -> Option<super::worker_poll_fence::WorkerPollFence> {
+        let worker_id = text(poll, "worker_id").ok()?;
+        let queue = text(poll, "task_queue").ok()?;
+        self.refuse_draining_worker(worker_id, queue).await.ok()?;
+        match self.admit_worker_poll(worker_id, queue).await.ok()? {
+            super::worker_poll_fence::PollAdmission::Fenced(fence) => Some(fence),
+            _ => None,
+        }
+    }
+
+    /// A worker's next activity poll, carried by its workflow task
+    /// completion and claimed in the completion's transaction, which then
+    /// takes the transition lock once for both. The poll's receipt records
+    /// the lease, which the worker's poll under the same request ID replays
+    /// without the lock (`replay_activity_poll`). Whether it leased a task,
+    /// or `None` when the poll did not run or was refused: the claim's
+    /// writes roll back to the savepoint, and the worker's own poll answers.
+    async fn claim_piggybacked_activity(
+        &self,
+        tx: &mut Transaction<'_, DB>,
+        poll: &Value,
+        fence: &super::worker_poll_fence::WorkerPollFence,
+        protocol: &str,
+    ) -> Result<Option<bool>> {
+        if !self.worker_poll_fence_holds_in(tx, fence, text(poll, "worker_id")?).await? {
+            return Ok(None);
+        }
+        sqlx::raw_sql("SAVEPOINT activity_poll").execute(&mut **tx).await?;
+        match self.claim_in(tx, poll, "activity", protocol, None, None).await {
+            Ok(claimed) => Ok(Some(claimed.is_some())),
+            Err(_) => {
+                sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
+                Ok(None)
+            }
+        }
+    }
+
+    /// An activity poll's receipt for a lease the worker still holds, read
+    /// without the transition lock: a committed lease that the claim would
+    /// replay under the lock (`claim_in`). A receipt in any other state
+    /// takes the claim's locked path, which expires or refuses it.
+    async fn replay_activity_poll(&self, worker_id: &str, request_id: &str) -> Result<Option<Value>> {
+        let Some(receipt) = Self::query("SELECT p.*,t.status,t.attempt_count,t.lease_owner,t.lease_expires_at FROM dw_poll_receipts p JOIN workflow_tasks t ON t.id=p.task_id WHERE p.worker_id=$1 AND p.kind=$2 AND p.request_id=$3 AND p.namespace=$4")
+            .bind(worker_id).bind("activity").bind(request_id).bind(self.namespace.as_ref()).fetch_optional(&self.pool).await?
+        else {
+            return Ok(None);
+        };
+        if DB::string(&receipt, "status")? == "leased"
+            && DB::number(&receipt, "attempt")? == DB::number(&receipt, "attempt_count")?
+            && DB::optional_string(&receipt, "lease_owner")?.as_deref() == Some(worker_id)
+            && DB::optional_instant(&receipt, "lease_expires_at")?.is_some_and(|expires| expires > now())
+        {
+            return Ok(Some(DB::document_row(&receipt, "response")?));
+        }
+        Ok(None)
     }
 
     /// One non-terminal command of a workflow task completion or of a
@@ -5330,7 +5413,8 @@ pub(crate) fn capabilities() -> Value {
         "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(),
         "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true,
         "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true,
-        "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true});
+        "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true,
+        "workflow_task_completion_activity_poll": true});
     value["runtime_external_payload_transport"] = super::payload_discovery::transport_manifest();
     value["supported_workflow_task_commands"]
         .as_array_mut()

Security static review found two blocking controls missing from the reference implementation.

The carried poll identity is independent of the completed lease. complete_workflow removes activity_poll, while the completion guard later proves the task owner from body.lease_owner. admit_piggybacked_activity_poll instead trusts poll.worker_id. The authenticated principal is tenant/role scoped, not bound to a worker ID. A completion owned by worker A can therefore cause the transaction to lease an activity and write a replay receipt for registered worker B, violating the stated same-worker and available-slot contract. Require activity_poll.worker_id == completion.lease_owner before admission and revalidate it inside the transaction with the task/fence. Add a cross-worker refusal control proving the completion behavior remains defined while no foreign lease or poll receipt is written.
claim_piggybacked_activity catches every Err from claim_in, rolls back the savepoint, and returns Ok(None). claim_in can fail on backend, decoding, stored-shape, budget, and invariant errors as well as documented poll refusals. Those unexpected faults must not be converted into an apparently successful completion with an omitted poll result. Downgrade only an explicit, closed refusal class; propagate or explicitly surface all unexpected failures. Add fault-injection controls for database and malformed stored-state errors.

This is a static finding only. The supplied patch was not applied or executed.

Sanitized rebased reference for report topic 9a: 0010-t09-Carry-the-next-activity-poll-on-the-workflow-task-co.patch.

The supplied revision targets Server b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. It is cumulative external reference material, not independently qualified against today's maintained source. Mail headers and commit-message/application instructions were removed; the unified diff is retained. The reporter now claims every rebased commit builds and the tip passes SQLite/MySQL/PostgreSQL tests. Those are attributed claims, not a new maintainer test result. Existing acceptance criteria, security findings and completed implementation receipts remain authoritative. Earlier attached patches are retained as historical references.

Sanitized patch: 13,109 bytes, SHA-256 6da0d07ac1d9d50851f98e930978470a4b0e634df81af343c81bda4d2adaaa54. Original-base sanitized identity: e9b5a930108a4437694184338d3a93b995f8110c114d66b5b2fc9f6e0c4db8cf. No patch was applied or executed during this update.

diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
index be36228b..39affb00 100644
--- a/rust/src/runtime/execution.rs
+++ b/rust/src/runtime/execution.rs
@@ -1566,7 +1566,20 @@ impl Runtime {
         worker: Option<super::runtime_credentials::Principal>,
         protocol: Option<&str>,
     ) -> Result<Value> {
-        self.complete_or_checkpoint(task_id, body, worker, protocol, None).await
+        self.complete_or_checkpoint(task_id, body, worker, protocol, None, None).await
+    }
+
+    /// A completion that carries the worker's next activity poll
+    /// (`Store::claim_piggybacked_activity`).
+    pub(crate) async fn complete_workflow_with_activity_poll(
+        &self,
+        task_id: &str,
+        body: Value,
+        worker: Option<super::runtime_credentials::Principal>,
+        protocol: Option<&str>,
+        activity_poll: Option<Value>,
+    ) -> Result<Value> {
+        self.complete_or_checkpoint(task_id, body, worker, protocol, None, activity_poll.as_ref()).await
     }
 
     /// POST …/cancellation-scopes/checkpoint, after its route checks: the
@@ -1579,7 +1592,7 @@ impl Runtime {
         protocol: Option<&str>,
         checkpoint: &super::scope_checkpoint::Checkpoint,
     ) -> Result<Value> {
-        self.complete_or_checkpoint(task_id, body, worker, protocol, Some(checkpoint)).await
+        self.complete_or_checkpoint(task_id, body, worker, protocol, Some(checkpoint), None).await
     }
 
     async fn complete_or_checkpoint(
@@ -1589,6 +1602,7 @@ impl Runtime {
         worker: Option<super::runtime_credentials::Principal>,
         protocol: Option<&str>,
         checkpoint: Option<&super::scope_checkpoint::Checkpoint>,
+        activity_poll: Option<&Value>,
     ) -> Result<Value> {
         // A task the namespace does not hold is answered before any payload
         // is read or fetched, as PHP's guards run before its normalizer.
@@ -1647,17 +1661,17 @@ impl Runtime {
         match &*self.storage {
             Storage::Sqlite(store) => {
                 store
-                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref())
+                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref(), activity_poll)
                     .await
             }
             Storage::Postgres(store) => {
                 store
-                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref())
+                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref(), activity_poll)
                     .await
             }
             Storage::MySql(store) => {
                 store
-                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref())
+                    .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint, signal.as_ref(), activity_poll)
                     .await
             }
         }
diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
index 628cc2bb..1259f739 100644
--- a/rust/src/runtime/http.rs
+++ b/rust/src/runtime/http.rs
@@ -2394,12 +2394,20 @@ async fn complete_workflow(
     request: CommandRequest,
     Path(task_id): Path<String>,
     headers: HeaderMap,
-    Json(body): Json<Value>,
+    Json(mut body): Json<Value>,
 ) -> Result<Json<Value>> {
+    // The worker's next activity poll, carried by the completion so its
+    // transaction claims the activity the commands schedule. It is not part
+    // of the completion's request: a poll the rules refuse is left to the
+    // worker's own poll, which answers the refusal.
+    let activity_poll = body
+        .as_object_mut()
+        .and_then(|body| body.remove("activity_poll"))
+        .filter(|poll| poll.is_object() && poll_rules(poll, ACTIVITY_POLL_RULES).is_ok());
     let body = completion_checks(&task_id, &headers, body)?;
     let attempt = body["workflow_task_attempt"].clone();
     runtime
-        .complete_workflow(&task_id, body, Some(request.principal().clone()), headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok()))
+        .complete_workflow_with_activity_poll(&task_id, body, Some(request.principal().clone()), headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok()), activity_poll)
         .await
         .map_err(|error| {
             completion_refusal(
diff --git a/rust/src/runtime/ready_age_tests.rs b/rust/src/runtime/ready_age_tests.rs
index 090daee9..50f64437 100644
--- a/rust/src/runtime/ready_age_tests.rs
+++ b/rust/src/runtime/ready_age_tests.rs
@@ -20,7 +20,7 @@ store_methods! {
                 "commands":[{"type":"schedule_activity","activity_type":"ready-age-activity",
                     "arguments":{"codec":"avro","blob":"AA=="},
                     "retry_policy":{"max_attempts":2,"backoff_seconds":[0]}}]}),
-                Arc::new(Semaphore::new(1)), BTreeMap::new(), None, Some("1.20"), Default::default(), None, None).await.unwrap();
+                Arc::new(Semaphore::new(1)), BTreeMap::new(), None, Some("1.20"), Default::default(), None, None, None).await.unwrap();
         }
         let row = Self::query("SELECT id FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type=$2 AND status='ready'")
             .bind(run).bind(kind).fetch_one(&self.pool).await.unwrap();
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index fccbe10d..a6d3c211 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -1955,6 +1955,18 @@ where
             super::worker_poll_fence::PollAdmission::Fenced(fence) => Some(fence),
             super::worker_poll_fence::PollAdmission::Unregistered => None,
         };
+        // An activity poll that a workflow task completion claimed for, or a
+        // retry, answers from its receipt before the lock.
+        if kind == "activity"
+            && fence.is_some()
+            && let Some(request_id) = body["poll_request_id"].as_str()
+            && let Some(task) = self.replay_activity_poll(text(&body, "worker_id")?, request_id).await?
+        {
+            return Ok(activity_server_time(
+                kind,
+                json!({"task": task, "poll_status": "leased", "server_capabilities": capabilities()}),
+            ));
+        }
         // A workflow poll's request ID replays its last deliverable answer,
         // after the registration checks that come first in PHP. The first
         // probe replays it (`replay_or_claim`).
@@ -2958,6 +2970,7 @@ where
         limits: super::memo_limits::StructuralLimits,
         checkpoint: Option<&super::scope_checkpoint::Checkpoint>,
         signal: Option<&super::signal_payloads::ResolvedSignal>,
+        activity_poll: Option<&Value>,
     ) -> Result<Value> {
         let transition_codec_for_streams = transition_codec.clone();
         // `commands` is required and non-empty, with no count limit of its
@@ -3112,6 +3125,10 @@ where
                 }
             }
         }
+        let activity_poll = match (activity_poll, protocol) {
+            (Some(poll), Some(protocol)) => self.admit_piggybacked_activity_poll(poll).await.map(|fence| (poll, fence, protocol)),
+            _ => None,
+        };
         let mut tx = self.begin().await?;
         let task = self.task_row(&mut tx, task_id).await?;
         if DB::string(&task, "task_type")? != "workflow" {
@@ -3427,11 +3444,77 @@ where
                 .await?;
         }
         tx.summary_changed(&run_id);
+        let activity_poll = match activity_poll {
+            Some((poll, fence, protocol)) => self.claim_piggybacked_activity(&mut tx, poll, &fence, protocol).await?,
+            None => None,
+        };
         tx.commit().await?;
         self.wake.notify_waiters();
-        Ok(
-            json!({"task_id": task_id, "run_id": run_id, "run_status": run_status, "workflow_task_attempt": body["workflow_task_attempt"], "outcome": "completed", "recorded": true, "reason": null, "created_task_ids": created}),
-        )
+        let mut response = json!({"task_id": task_id, "run_id": run_id, "run_status": run_status, "workflow_task_attempt": body["workflow_task_attempt"], "outcome": "completed", "recorded": true, "reason": null, "created_task_ids": created});
+        if let Some(leased) = activity_poll {
+            response["activity_poll"] = json!({"leased": leased});
+        }
+        Ok(response)
+    }
+
+    /// The poll checks a carried activity poll takes before the completion's
+    /// transaction, as `poll_with_pages` takes them: its registration's
+    /// fence, or `None` when the worker's own poll is to answer it instead.
+    async fn admit_piggybacked_activity_poll(&self, poll: &Value) -> Option<super::worker_poll_fence::WorkerPollFence> {
+        let worker_id = text(poll, "worker_id").ok()?;
+        let queue = text(poll, "task_queue").ok()?;
+        self.refuse_draining_worker(worker_id, queue).await.ok()?;
+        match self.admit_worker_poll(worker_id, queue).await.ok()? {
+            super::worker_poll_fence::PollAdmission::Fenced(fence) => Some(fence),
+            _ => None,
+        }
+    }
+
+    /// A worker's next activity poll, carried by its workflow task
+    /// completion and claimed in the completion's transaction, which then
+    /// takes the transition lock once for both. The poll's receipt records
+    /// the lease, which the worker's poll under the same request ID replays
+    /// without the lock (`replay_activity_poll`). Whether it leased a task,
+    /// or `None` when the poll did not run or was refused: the claim's
+    /// writes roll back to the savepoint, and the worker's own poll answers.
+    async fn claim_piggybacked_activity(
+        &self,
+        tx: &mut Transaction<'_, DB>,
+        poll: &Value,
+        fence: &super::worker_poll_fence::WorkerPollFence,
+        protocol: &str,
+    ) -> Result<Option<bool>> {
+        if !self.worker_poll_fence_holds_in(tx, fence, text(poll, "worker_id")?).await? {
+            return Ok(None);
+        }
+        sqlx::raw_sql("SAVEPOINT activity_poll").execute(&mut **tx).await?;
+        match self.claim_in(tx, poll, "activity", protocol, None, None).await {
+            Ok(claimed) => Ok(Some(claimed.is_some())),
+            Err(_) => {
+                sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
+                Ok(None)
+            }
+        }
+    }
+
+    /// An activity poll's receipt for a lease the worker still holds, read
+    /// without the transition lock: a committed lease that the claim would
+    /// replay under the lock (`claim_in`). A receipt in any other state
+    /// takes the claim's locked path, which expires or refuses it.
+    async fn replay_activity_poll(&self, worker_id: &str, request_id: &str) -> Result<Option<Value>> {
+        let Some(receipt) = Self::query("SELECT p.*,t.status,t.attempt_count,t.lease_owner,t.lease_expires_at FROM dw_poll_receipts p JOIN workflow_tasks t ON t.id=p.task_id WHERE p.worker_id=$1 AND p.kind=$2 AND p.request_id=$3 AND p.namespace=$4")
+            .bind(worker_id).bind("activity").bind(request_id).bind(self.namespace.as_ref()).fetch_optional(&self.pool).await?
+        else {
+            return Ok(None);
+        };
+        if DB::string(&receipt, "status")? == "leased"
+            && DB::number(&receipt, "attempt")? == DB::number(&receipt, "attempt_count")?
+            && DB::optional_string(&receipt, "lease_owner")?.as_deref() == Some(worker_id)
+            && DB::optional_instant(&receipt, "lease_expires_at")?.is_some_and(|expires| expires > now())
+        {
+            return Ok(Some(DB::document_row(&receipt, "response")?));
+        }
+        Ok(None)
     }
 
     /// One non-terminal command of a workflow task completion or of a
@@ -5344,7 +5427,8 @@ pub(crate) fn capabilities() -> Value {
         "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(),
         "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true,
         "claimed_task_ready_age": true, "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true,
-        "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true});
+        "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true,
+        "workflow_task_completion_activity_poll": true});
     value["runtime_external_payload_transport"] = super::payload_discovery::transport_manifest();
     value["supported_workflow_task_commands"]
         .as_array_mut()

Producer checkpoint for the dependent carried-poll SDK work: Server still owns #1060/#1061/#1062/#1067. Current accepted main is 7fbf5124; the active source train is #1072 for #1055/#1056, now 5d3eb6a0 after its first real-backend corrections. The carried-poll capabilities and versioned reply schemas are not implemented or advertised in that source. There is no new authoritative producer/receipt tuple for a consumer to infer yet.

The following accepted constraints remain mandatory for the producer contract, not inferred support:

Every attached poll must identify the same canonical worker as the completion lease owner, checked before claim admission and again with the retained task/claim fence. If both poll kinds are supplied, they must also identify that same worker.
Only a closed set of known soft claim refusals may become a no-lease disposition. Database, decoding, malformed stored-state, budget and invariant faults must propagate or be explicitly surfaced with the completion outcome; they cannot silently become leased:false.
Under Native: carry the next workflow poll on the workflow task completion #1062's single-capacity rule, a successful activity grant prevents a workflow attempt. The missing workflow_poll in that case means it was not attempted, not that no workflow task exists. Exact omitted/member behavior for every other variant still needs the versioned producer schema.
Native: return a carried poll's answer in its completion's response #1067 requires an actual normal poll answer for a successful carried grant, preserving encoding, capabilities, history/external-payload projection and budgets. A draft leased flag is not that final receipt contract.

The next Server-owned contract checkpoint will bind the complete closed request/reply fields, limits, error dispositions and versioned schema to the actual implementation and its producer/consumer qualification. Dependent SDK source can follow those exact bytes when supplied; no duplicate producer implementation, guessed enabled field, native tuple, runtime or publication admission follows from this status record. The existing finite Security findings on these issues remain part of the implementation work.

Producer implementation is now active on draft #1073, initially e83d6f498f30be611a032192770ecdebf28d1aa1, based on accepted main c3bbfa5f. #1072 actual-main acceptance and cleanup are complete; this is the sole hosted source train.

The first activity-carrier slice binds the poll worker to the completing owner before admission and again under the retained fence, checks the opposite execution slot, uses the existing savepoint/claim path, and distinguishes a closed soft-refusal list from visible database/decoder/budget/invariant faults. It is not yet advertised or an accepted consumer contract. Versioned OpenAPI, real backend/HTTP controls and replay vectors are being completed in the user's order #1060 → #1061 → #1062 → #1067 on this draft.

Server owns the next exact contract/schema checkpoint for sdk-php #328–#331. Normal replay must preserve original lease/session/deadline authority; a fresh response server_time does not renew it, and the consumer must charge the original completion request's transit. Missing workflow_poll delivers no workflow lease and must not be interpreted as an empty queue or disabled capability. No published tuple, release, installed behavior or source PASS is inferred from the draft.

Security follow-up on the updated rebased reference 0010-t09-Carry-the-next-activity-poll-on-the-workflow-task-co.patch (13,109 B / SHA-256 6da0d07ac1d9d50851f98e930978470a4b0e634df81af343c81bda4d2adaaa54): both blockers in 6692713547 remain.

The rebase supplies the maintained ready-age call shape/capability context, but it still admits and fences the carried poll from poll.worker_id without requiring equality to the completing workflow task's lease_owner. A completion owned by worker A can still allocate/replay for registered worker B. It also still catches every Err(_) from claim_in, rolls back the savepoint and returns Ok(None), collapsing unexpected backend/decoding/stored-state/budget/invariant faults into an apparently successful completion.

The ready-age compile-context update is useful reference material but does not replace the same-worker in-transaction equality control, closed soft-refusal error class, cross-worker no-write proof or injected-fault propagation controls. Static review only; no patch was applied or executed.

A concrete maintained-source boundary matters for the draining/refencing controls. Store::workflow_task_guard calls lease_owner_fresh, which requires an active, fresh registration. A steadily draining owner therefore already receives stale_worker_registration for its workflow completion, with or without a carrier. The new same-worker constraint prevents using another active worker's poll to work around that parent guard.

The producer contract preserves the parent completion guard. Soft carrier declines leave an otherwise valid completion intact; they cannot authorize a stale, expired, mismatched or draining parent's completion. The controls will separately cover carrier admission declines, a changed retained poll fence, occupied opposite slots and unchanged parent-guard refusal. A draining poll is never a new-lease authority, and no steady-state draining completion PASS is claimed.

This is a source reconciliation on #1073, not permission to widen completion authority through an optional request member. OpenAPI and the SDK handoff will state the distinction explicitly. Database, decoder, budget and invariant faults remain visible errors, not soft declines. No capability contract or runtime tuple is accepted from this finding alone.

Accepted producer checkpoint: PR1073, accepted head 511bd6fd, normally merged 54af4640. Its contract is included in the accepted 4fb runtime and now the immutable published native 3.2.0 source/tag 3c6f1eda9c6d2ebc0ee5bec8ccf431900052be9e. The earlier reference-draft same-worker/error findings are superseded by that implementation and its actual SQLite/PostgreSQL/MySQL/MariaDB Source controls. They are not remaining implementation holds on the accepted producer.

The authoritative OpenAPI schema 63 and closed replay vectors retain protocol 1.20. Schema SHA-256 is d6c726a7f97417494d9897ae1101ac7300ba6414174041dea875ae01e5052391; vectors SHA-256 12ca5db29d96b5e0f03852b698ca1d15bd796c07486a1af85484d4e4dfc45e9b.

Capability workflow_task_completion_activity_poll admits the optional activity_poll transport carrier on workflow completion. Its worker must equal the completing lease owner before admission and again under the transaction/retained registration fence. An occupied workflow or validation slot, including unknown expiry, declines eager dispatch without creating an activity lease/attempt/receipt. The valid completion remains recorded on closed soft refusal; an unexpected claim/backend fault propagates and rolls back. Commit uncertainty retains its visible storage error and original completion/poll recovery identities. Duplicate completion does not dispatch again.

The actual publisher input is now available: owner terminal PASS, independent Security publication PASS, index ghcr.io/durable-workflow/server@sha256:6916f6f3bb1ee58d071f10f888068fd321cd1576fb70dc97ce8e9e61da493d62, exact producer/native runner 3c6. This issue stays open for actual published qualification. Server next owns the existing corpus/four-cell Compose receipt packet 325/6705614909. General next binds the actual published SDK package and SDK runner plus the bounded invocation to this producer; SDK-PHP owns its existing consumer/harness execution and result. The shared consumer handoff is SDK293/6705418888. Installed clock/transit/buffer-age, missing-key capacity, lost-answer/original-ID recovery, expiry/refence and selected load acceptance remain open. No tentative/withdrawn head, Source fixture or older tuple supplies that result.

General reconciled the carried-poll and workflow-clock delivery against the original acceptance criteria while both repository agents are Off.

The producer is published Native Server 3.3.0, source 2a15179f29801eec9f2a2d4e6a9c5f93233f104e, index sha256:cb4b11bd6235a551cf51e7dc6342504fbbcef19959618b376fb5648ca00fc388. The consumer is normal published PHP SDK 3.11.0, source fbc64ef98c27be467d75a29a29bfb0ee4e80dba7.

Feature-specific acceptance is covered by the maintained source controls. The carried-poll runtime tests, HTTP execution tests, workflow replay/history-budget tests and versioned OpenAPI are byte-identical between the qualified integrated source 5f937ec8 (run 42448138571) and the 3.3.0 cut. The source lanes execute the registered controls on SQLite, MySQL, MariaDB and PostgreSQL; the accepted #1073 source also has 62 successful checks and its declared skip.

Published compatibility is now verified with the unchanged full PHP replay matrix: 42531701411, 13 executed scenarios passed, zero skips, exact Server 3.3.0 and normal Packagist SDK 3.11.0, distinct clients, restarted worker, completed-history replay/query, in-flight signal/restart and nondeterminism refusal. Captured audit. Its runner bootstrap fix has complete ordinary main custody: signed reviewed tree, 71 successful main checks/eight expected skips and topic absence verified. The published product bytes and feature-specific controls are unchanged.

The source controls establish the transaction/receipt/answer invariants. The published matrix establishes producer/consumer compatibility; it is not an installed SQL-count or benchmark experiment. No speedup is claimed. #1073's earlier archive-acquisition NONPASSs and the first published replay's interpreter-bootstrap NONPASS remain attributed.

Acceptance: qualify_carried_activity_polls checks one completion transition, receipt replay with no new transition and equality of the embedded answer with its ordinary replay except the clock. Refused/draining/fenced/occupied-slot and fault controls preserve the stated completion boundary; the HTTP vector test covers lost completion acknowledgement and original lease recovery. The capability, optional request and response are versioned in OpenAPI. No-carrier requests retain their ordinary path. Published SDK 3.11.0 contains the accepted #328 consumer.

The separate condition-wait durable-workflow/sdk-php#269 final-audit C2/C5 follow-up remains open; it is not a carried-poll or workflow-clock acceptance claim. The original reporter benchmark remains unverified.

Navigation Menu

Repository navigation

Native: eager activity dispatch for a poll carried on the workflow task completion #1060

Delivered in Native 3.3.0 and PHP SDK 3.11.0

Original requirement and historical evidence

Problem

Proposed contract

Reported benchmark evidence

Acceptance

Implementation constraints

Dependencies

Draft patches and validation boundary

Updated reference, November 6

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Native: eager activity dispatch for a poll carried on the workflow task completion #1060

Description

Delivered in Native 3.3.0 and PHP SDK 3.11.0

Original requirement and historical evidence

Problem

Proposed contract

Reported benchmark evidence

Acceptance

Implementation constraints

Dependencies

Draft patches and validation boundary

Updated reference, November 6

Activity

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 9, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions