Native: return a carried poll's answer in its completion's response #1067

Activity

1. rmcdaniel

   added

   kind:featureA public product capability or experience is requestedA public product capability or experience is requested

   priority:P0Immediate release or data-integrity riskImmediate release or data-integrity risk

   repo:serverOwned by the standalone server repositoryOwned by the standalone server repository

   agent:server

   on Nov 6, 2026
2. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0025-t16-Drop-the-completion-wrappers-the-carried-polls-left-.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: 1,003 bytes, SHA-256 af8a4334c753edb4a9fe200bd9c2525ff3897761a98113ed613912efc8f9d239. This part: 1,003 bytes, SHA-256 af8a4334c753edb4a9fe200bd9c2525ff3897761a98113ed613912efc8f9d239. 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 4a40032a..10a54838 100644
   --- a/rust/src/runtime/execution.rs
   +++ b/rust/src/runtime/execution.rs
   @@ -1559,6 +1559,7 @@ impl Runtime {
                Storage::MySql(store) => store.fail_workflow_task(task_id, body, codec, signal.as_ref()).await,
            }
        }
   +    #[cfg(test)]
        pub(crate) async fn complete_workflow(
            &self,
            task_id: &str,
   @@ -1789,10 +1790,7 @@ impl Runtime {
                }
            }
        }
   -    pub(crate) async fn complete_activity(&self, task_id: &str, body: Value) -> Result<Value> {
   -        self.complete_activity_with_workflow_poll(task_id, body, None).await
   -    }
   -    /// `complete_activity` with the worker's next workflow poll and its
   +    /// An activity completion, with the worker's next workflow poll and its
        /// protocol version, which the completion's transaction claims for.
        pub(crate) async fn complete_activity_with_workflow_poll(
            &self,
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0026-t16-Answer-a-carried-poll-s-lease-in-its-completion-s-re.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,156 bytes, SHA-256 02305e098437b857cf9d67c53edd63a1951ec0f77d76375a3b05dbec14137563. This part: 12,156 bytes, SHA-256 02305e098437b857cf9d67c53edd63a1951ec0f77d76375a3b05dbec14137563. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index a416da1c..9755588a 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -3512,17 +3512,19 @@ where
            // runs no activity first: the completion carried no activity poll,
            // or that poll's claim leased nothing.
            let workflow_poll = match &workflow_poll {
   -            Some(poll) if !carries_activity_poll || activity_poll == Some(false) => self.claim_carried_workflow(&mut tx, poll).await?,
   +            Some(poll) if !carries_activity_poll || matches!(activity_poll, Some(None)) => {
   +                self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding))
   +            }
                _ => None,
            };
            tx.commit().await?;
            self.wake.notify_waiters();
            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});
   +        if let Some(claimed) = activity_poll {
   +            response["activity_poll"] = carried_poll("activity", claimed, None)?;
            }
   -        if let Some(leased) = workflow_poll {
   -            response["workflow_poll"] = json!({"leased": leased});
   +        if let Some((claimed, encoding)) = workflow_poll {
   +            response["workflow_poll"] = carried_poll("workflow", claimed, encoding)?;
            }
            Ok(response)
        }
   @@ -3544,8 +3546,8 @@ where
        /// 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
   +    /// without the lock (`replay_activity_poll`). The task it leased, if
   +    /// any, 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,
   @@ -3553,13 +3555,13 @@ where
            poll: &Value,
            fence: &super::worker_poll_fence::WorkerPollFence,
            protocol: &str,
   -    ) -> Result<Option<bool>> {
   +    ) -> Result<Option<Option<Value>>> {
            let Some(registration) = self.worker_poll_fence_registration_in(tx, fence, text(poll, "worker_id")?).await? else {
                return Ok(None);
            };
            sqlx::raw_sql("SAVEPOINT activity_poll").execute(&mut **tx).await?;
            match self.claim_in(tx, poll, "activity", protocol, None, None, Some(registration)).await {
   -            Ok(claimed) => Ok(Some(claimed.is_some())),
   +            Ok(claimed) => Ok(Some(claimed)),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
                    tx.run_tags.clear();
   @@ -4155,14 +4157,14 @@ where
                .await?
            };
            let workflow_poll = match &workflow_poll {
   -            Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?,
   +            Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding)),
                None => None,
            };
            tx.commit().await?;
            self.wake.notify_waiters();
            let mut response = json!({"task_id": task_id, "activity_attempt_id": attempt_id, "outcome": "completed", "recorded": true, "reason": null, "next_task_id": next_task_id});
   -        if let Some(leased) = workflow_poll {
   -            response["workflow_poll"] = json!({"leased": leased});
   +        if let Some((claimed, encoding)) = workflow_poll {
   +            response["workflow_poll"] = carried_poll("workflow", claimed, encoding)?;
            }
            Ok(response)
        }
   @@ -4183,20 +4185,20 @@ where
                _ => return None,
            };
            let request = self.workflow_poll_request(worker_id, queue, id, poll.default_wait, &["workflow"]).await.ok()?;
   -        Some(AdmittedWorkflowPoll { body: poll.body, protocol: poll.protocol, page: poll.pages.poll(requested, encoding), fence, request })
   +        Some(AdmittedWorkflowPoll { body: poll.body, protocol: poll.protocol, page: poll.pages.poll(requested, encoding), encoding, fence, request })
        }
    
        /// The carried workflow poll's claim, in the completion's transaction,
        /// under the poll's request ID: the worker's poll with that
   -    /// ID replays the lease. A claim that fails is rolled back to before it,
   -    /// and the completion commits without it.
   -    async fn claim_carried_workflow(&self, tx: &mut Transaction<'_, DB>, poll: &AdmittedWorkflowPoll<'_>) -> Result<Option<bool>> {
   +    /// ID replays the lease. The task it leased, if any. A claim that fails
   +    /// is rolled back to before it, and the completion commits without it.
   +    async fn claim_carried_workflow(&self, tx: &mut Transaction<'_, DB>, poll: &AdmittedWorkflowPoll<'_>) -> Result<Option<Option<Value>>> {
            let Some(registration) = self.worker_poll_fence_registration_in(tx, &poll.fence, text(poll.body, "worker_id")?).await? else {
                return Ok(None);
            };
            sqlx::raw_sql("SAVEPOINT workflow_poll").execute(&mut **tx).await?;
            match self.claim_in(tx, poll.body, "workflow", poll.protocol, poll.page, Some(&poll.request), Some(registration)).await {
   -            Ok(claimed) => Ok(Some(claimed.is_some())),
   +            Ok(claimed) => Ok(Some(claimed)),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
                    tx.run_tags.clear();
   @@ -5585,6 +5587,7 @@ struct AdmittedWorkflowPoll<'a> {
        body: &'a Value,
        protocol: &'a str,
        page: Option<i64>,
   +    encoding: Option<&'static str>,
        fence: super::worker_poll_fence::WorkerPollFence,
        request: super::workflow_poll_requests::WorkflowPollRequest,
    }
   @@ -5597,6 +5600,23 @@ enum FirstProbe {
        Claimed(Result<Option<Value>>),
    }
    
   +/// A carried poll's part of its completion's answer: whether its claim
   +/// leased a task and, when it did, the answer the worker's poll under the
   +/// poll's request ID would replay now, so the worker need not send it.
   +fn carried_poll(kind: &str, claimed: Option<Value>, encoding: Option<&'static str>) -> Result<Value> {
   +    let Some(task) = claimed else {
   +        return Ok(json!({"leased": false}));
   +    };
   +    let mut answer = replayed_poll(json!({"task": task, "poll_status": "leased"}), encoding)?;
   +    // As `Runtime::poll_workflow_tasks` marks a workflow lease.
   +    if kind == "workflow" {
   +        answer["task"]["task_kind"] = json!("workflow");
   +    }
   +    answer["server_time"] = json!(super::activity_lease_renewal::micros(now()));
   +    answer["protocol_version"] = json!(super::WORKER_PROTOCOL_VERSION);
   +    Ok(json!({"leased": true, "answer": answer}))
   +}
   +
    /// A replayed answer as a workflow poll delivers it: a lease takes this
    /// request's history encoding.
    pub(super) fn replayed_poll(mut replay: Value, encoding: Option<&'static str>) -> Result<Value> {
   diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
   index aec2c319..58c8fa7a 100644
   --- a/rust/tests/execution.rs
   +++ b/rust/tests/execution.rs
   @@ -379,6 +379,8 @@ mod task_fairness;
    mod worker_protocol_envelope;
    #[path = "execution/child_start_turn.rs"]
    mod child_start_turn;
   +#[path = "execution/carried_polls.rs"]
   +mod carried_polls;
    
    enum TestDatabase {
        Sqlite(tempfile::TempDir),
   diff --git a/rust/tests/execution/carried_polls.rs b/rust/tests/execution/carried_polls.rs
   new file mode 100644
   index 00000000..e9558f48
   --- /dev/null
   +++ b/rust/tests/execution/carried_polls.rs
   @@ -0,0 +1,66 @@
   +use super::*;
   +
   +/// A completion whose transaction claims for the poll it carries answers
   +/// with what the worker's poll under that request ID replays, so the worker
   +/// need not send it. The poll still replays the same answer.
   +#[tokio::test]
   +async fn a_carried_poll_s_lease_comes_back_with_its_completion() {
   +    let database = TestDatabase::new().await;
   +    let runtime = database.open().await.unwrap();
   +    let app = router(runtime.clone());
   +    // Each as the PHP worker carries it.
   +    let carried = |id: &str| json!({"worker_id":"carrier","task_queue":"test","poll_request_id":id});
   +    let carried_workflow = |id: &str| json!({"worker_id":"carrier","task_queue":"test","poll_request_id":id,"history_page_size":100});
   +    let replay = |kind: &'static str, id: &'static str| {
   +        let app = &app;
   +        async move {
   +            let mut body = json!({"worker_id":"carrier","task_queue":"test","poll_request_id":id,"timeout_seconds":0});
   +            if kind == "workflow" {
   +                body["history_page_size"] = json!(100);
   +            }
   +            let replayed = request(app, "POST", &format!("/api/worker/{kind}-tasks/poll"), body).await;
   +            assert_eq!(replayed.0, StatusCode::OK, "{}", replayed.1);
   +            replayed.1
   +        }
   +    };
   +    register(&app, "carrier", json!(["echo"]), json!(["carried-activity"])).await;
   +    start(&app, "carried-run").await;
   +    let task = poll(&app, "carrier", "workflow").await;
   +
   +    // The activity the commands schedule is leased to the carried activity
   +    // poll, and the workflow poll is not claimed for.
   +    let mut body = completion(&task, json!([{"type":"schedule_activity","activity_type":"carried-activity",
   +        "arguments":envelope(Payload::Array(vec![])),"start_to_close_timeout":120}]));
   +    body["activity_poll"] = carried("carried-activity-poll");
   +    body["workflow_poll"] = carried_workflow("carried-workflow-poll-1");
   +    let done = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), body).await;
   +    assert_eq!(done.0, StatusCode::OK, "{}", done.1);
   +    assert_eq!(done.1["activity_poll"]["leased"], true, "{}", done.1);
   +    assert!(done.1.get("workflow_poll").is_none(), "{}", done.1);
   +    let answer = &done.1["activity_poll"]["answer"];
   +    assert_eq!(answer["task"]["activity_type"], "carried-activity", "{answer}");
   +    assert_poll_replay(answer, &replay("activity", "carried-activity-poll").await, "the activity poll replays its carried answer");
   +
   +    // The activity's result makes the workflow task ready, and the carried
   +    // workflow poll leases it, history and all.
   +    let activity = &answer["task"];
   +    let done = request(&app, "POST", &format!("/api/worker/activity-tasks/{}/complete", activity["task_id"].as_str().unwrap()),
   +        json!({"lease_owner":activity["lease_owner"],"activity_attempt_id":activity["activity_attempt_id"],"result":envelope(Payload::Null),
   +            "workflow_poll":carried_workflow("carried-workflow-poll-2")})).await;
   +    assert_eq!(done.0, StatusCode::OK, "{}", done.1);
   +    assert_eq!(done.1["workflow_poll"]["leased"], true, "{}", done.1);
   +    let answer = &done.1["workflow_poll"]["answer"];
   +    assert_eq!(answer["task"]["task_kind"], "workflow", "{answer}");
   +    assert!(answer["task"]["history_events"].as_array().is_some_and(|events| events.len() > 2), "{answer}");
   +    assert_poll_replay(answer, &replay("workflow", "carried-workflow-poll-2").await, "the workflow poll replays its carried answer");
   +
   +    // A carried poll that finds nothing to lease says so, with no answer.
   +    let task = &answer["task"];
   +    let mut body = completion(task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]));
   +    body["workflow_poll"] = carried_workflow("carried-workflow-poll-3");
   +    let done = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), body).await;
   +    assert_eq!(done.0, StatusCode::OK, "{}", done.1);
   +    assert_eq!(done.1["workflow_poll"], json!({"leased": false}), "{}", done.1);
   +    runtime.close().await;
   +    database.remove().await;
   +}
4. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized rebased reference for report topic 16a: 0026-t16-Answer-a-carried-poll-s-lease-in-its-completion-s-re.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: 12,156 bytes, SHA-256 37630160fe4e524c34aa2342037344cc5d8ebf3d965998b385f6d4cd9d892d25. Original-base sanitized identity: 02305e098437b857cf9d67c53edd63a1951ec0f77d76375a3b05dbec14137563. No patch was applied or executed during this update.

   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index 54fc7896..b6245837 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -3525,17 +3525,19 @@ where
            // runs no activity first: the completion carried no activity poll,
            // or that poll's claim leased nothing.
            let workflow_poll = match &workflow_poll {
   -            Some(poll) if !carries_activity_poll || activity_poll == Some(false) => self.claim_carried_workflow(&mut tx, poll).await?,
   +            Some(poll) if !carries_activity_poll || matches!(activity_poll, Some(None)) => {
   +                self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding))
   +            }
                _ => None,
            };
            tx.commit().await?;
            self.wake.notify_waiters();
            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});
   +        if let Some(claimed) = activity_poll {
   +            response["activity_poll"] = carried_poll("activity", claimed, None)?;
            }
   -        if let Some(leased) = workflow_poll {
   -            response["workflow_poll"] = json!({"leased": leased});
   +        if let Some((claimed, encoding)) = workflow_poll {
   +            response["workflow_poll"] = carried_poll("workflow", claimed, encoding)?;
            }
            Ok(response)
        }
   @@ -3557,8 +3559,8 @@ where
        /// 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
   +    /// without the lock (`replay_activity_poll`). The task it leased, if
   +    /// any, 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,
   @@ -3566,13 +3568,13 @@ where
            poll: &Value,
            fence: &super::worker_poll_fence::WorkerPollFence,
            protocol: &str,
   -    ) -> Result<Option<bool>> {
   +    ) -> Result<Option<Option<Value>>> {
            let Some(registration) = self.worker_poll_fence_registration_in(tx, fence, text(poll, "worker_id")?).await? else {
                return Ok(None);
            };
            sqlx::raw_sql("SAVEPOINT activity_poll").execute(&mut **tx).await?;
            match self.claim_in(tx, poll, "activity", protocol, None, None, Some(registration)).await {
   -            Ok(claimed) => Ok(Some(claimed.is_some())),
   +            Ok(claimed) => Ok(Some(claimed)),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
                    tx.run_tags.clear();
   @@ -4168,14 +4170,14 @@ where
                .await?
            };
            let workflow_poll = match &workflow_poll {
   -            Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?,
   +            Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding)),
                None => None,
            };
            tx.commit().await?;
            self.wake.notify_waiters();
            let mut response = json!({"task_id": task_id, "activity_attempt_id": attempt_id, "outcome": "completed", "recorded": true, "reason": null, "next_task_id": next_task_id});
   -        if let Some(leased) = workflow_poll {
   -            response["workflow_poll"] = json!({"leased": leased});
   +        if let Some((claimed, encoding)) = workflow_poll {
   +            response["workflow_poll"] = carried_poll("workflow", claimed, encoding)?;
            }
            Ok(response)
        }
   @@ -4196,20 +4198,20 @@ where
                _ => return None,
            };
            let request = self.workflow_poll_request(worker_id, queue, id, poll.default_wait, &["workflow"]).await.ok()?;
   -        Some(AdmittedWorkflowPoll { body: poll.body, protocol: poll.protocol, page: poll.pages.poll(requested, encoding), fence, request })
   +        Some(AdmittedWorkflowPoll { body: poll.body, protocol: poll.protocol, page: poll.pages.poll(requested, encoding), encoding, fence, request })
        }
    
        /// The carried workflow poll's claim, in the completion's transaction,
        /// under the poll's request ID: the worker's poll with that
   -    /// ID replays the lease. A claim that fails is rolled back to before it,
   -    /// and the completion commits without it.
   -    async fn claim_carried_workflow(&self, tx: &mut Transaction<'_, DB>, poll: &AdmittedWorkflowPoll<'_>) -> Result<Option<bool>> {
   +    /// ID replays the lease. The task it leased, if any. A claim that fails
   +    /// is rolled back to before it, and the completion commits without it.
   +    async fn claim_carried_workflow(&self, tx: &mut Transaction<'_, DB>, poll: &AdmittedWorkflowPoll<'_>) -> Result<Option<Option<Value>>> {
            let Some(registration) = self.worker_poll_fence_registration_in(tx, &poll.fence, text(poll.body, "worker_id")?).await? else {
                return Ok(None);
            };
            sqlx::raw_sql("SAVEPOINT workflow_poll").execute(&mut **tx).await?;
            match self.claim_in(tx, poll.body, "workflow", poll.protocol, poll.page, Some(&poll.request), Some(registration)).await {
   -            Ok(claimed) => Ok(Some(claimed.is_some())),
   +            Ok(claimed) => Ok(Some(claimed)),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
                    tx.run_tags.clear();
   @@ -5599,6 +5601,7 @@ struct AdmittedWorkflowPoll<'a> {
        body: &'a Value,
        protocol: &'a str,
        page: Option<i64>,
   +    encoding: Option<&'static str>,
        fence: super::worker_poll_fence::WorkerPollFence,
        request: super::workflow_poll_requests::WorkflowPollRequest,
    }
   @@ -5611,6 +5614,23 @@ enum FirstProbe {
        Claimed(Result<Option<Value>>),
    }
    
   +/// A carried poll's part of its completion's answer: whether its claim
   +/// leased a task and, when it did, the answer the worker's poll under the
   +/// poll's request ID would replay now, so the worker need not send it.
   +fn carried_poll(kind: &str, claimed: Option<Value>, encoding: Option<&'static str>) -> Result<Value> {
   +    let Some(task) = claimed else {
   +        return Ok(json!({"leased": false}));
   +    };
   +    let mut answer = replayed_poll(json!({"task": task, "poll_status": "leased"}), encoding)?;
   +    // As `Runtime::poll_workflow_tasks` marks a workflow lease.
   +    if kind == "workflow" {
   +        answer["task"]["task_kind"] = json!("workflow");
   +    }
   +    answer["server_time"] = json!(super::activity_lease_renewal::micros(now()));
   +    answer["protocol_version"] = json!(super::WORKER_PROTOCOL_VERSION);
   +    Ok(json!({"leased": true, "answer": answer}))
   +}
   +
    /// A replayed answer as a workflow poll delivers it: a lease takes this
    /// request's history encoding.
    pub(super) fn replayed_poll(mut replay: Value, encoding: Option<&'static str>) -> Result<Value> {
   diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
   index aec2c319..58c8fa7a 100644
   --- a/rust/tests/execution.rs
   +++ b/rust/tests/execution.rs
   @@ -379,6 +379,8 @@ mod task_fairness;
    mod worker_protocol_envelope;
    #[path = "execution/child_start_turn.rs"]
    mod child_start_turn;
   +#[path = "execution/carried_polls.rs"]
   +mod carried_polls;
    
    enum TestDatabase {
        Sqlite(tempfile::TempDir),
   diff --git a/rust/tests/execution/carried_polls.rs b/rust/tests/execution/carried_polls.rs
   new file mode 100644
   index 00000000..e9558f48
   --- /dev/null
   +++ b/rust/tests/execution/carried_polls.rs
   @@ -0,0 +1,66 @@
   +use super::*;
   +
   +/// A completion whose transaction claims for the poll it carries answers
   +/// with what the worker's poll under that request ID replays, so the worker
   +/// need not send it. The poll still replays the same answer.
   +#[tokio::test]
   +async fn a_carried_poll_s_lease_comes_back_with_its_completion() {
   +    let database = TestDatabase::new().await;
   +    let runtime = database.open().await.unwrap();
   +    let app = router(runtime.clone());
   +    // Each as the PHP worker carries it.
   +    let carried = |id: &str| json!({"worker_id":"carrier","task_queue":"test","poll_request_id":id});
   +    let carried_workflow = |id: &str| json!({"worker_id":"carrier","task_queue":"test","poll_request_id":id,"history_page_size":100});
   +    let replay = |kind: &'static str, id: &'static str| {
   +        let app = &app;
   +        async move {
   +            let mut body = json!({"worker_id":"carrier","task_queue":"test","poll_request_id":id,"timeout_seconds":0});
   +            if kind == "workflow" {
   +                body["history_page_size"] = json!(100);
   +            }
   +            let replayed = request(app, "POST", &format!("/api/worker/{kind}-tasks/poll"), body).await;
   +            assert_eq!(replayed.0, StatusCode::OK, "{}", replayed.1);
   +            replayed.1
   +        }
   +    };
   +    register(&app, "carrier", json!(["echo"]), json!(["carried-activity"])).await;
   +    start(&app, "carried-run").await;
   +    let task = poll(&app, "carrier", "workflow").await;
   +
   +    // The activity the commands schedule is leased to the carried activity
   +    // poll, and the workflow poll is not claimed for.
   +    let mut body = completion(&task, json!([{"type":"schedule_activity","activity_type":"carried-activity",
   +        "arguments":envelope(Payload::Array(vec![])),"start_to_close_timeout":120}]));
   +    body["activity_poll"] = carried("carried-activity-poll");
   +    body["workflow_poll"] = carried_workflow("carried-workflow-poll-1");
   +    let done = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), body).await;
   +    assert_eq!(done.0, StatusCode::OK, "{}", done.1);
   +    assert_eq!(done.1["activity_poll"]["leased"], true, "{}", done.1);
   +    assert!(done.1.get("workflow_poll").is_none(), "{}", done.1);
   +    let answer = &done.1["activity_poll"]["answer"];
   +    assert_eq!(answer["task"]["activity_type"], "carried-activity", "{answer}");
   +    assert_poll_replay(answer, &replay("activity", "carried-activity-poll").await, "the activity poll replays its carried answer");
   +
   +    // The activity's result makes the workflow task ready, and the carried
   +    // workflow poll leases it, history and all.
   +    let activity = &answer["task"];
   +    let done = request(&app, "POST", &format!("/api/worker/activity-tasks/{}/complete", activity["task_id"].as_str().unwrap()),
   +        json!({"lease_owner":activity["lease_owner"],"activity_attempt_id":activity["activity_attempt_id"],"result":envelope(Payload::Null),
   +            "workflow_poll":carried_workflow("carried-workflow-poll-2")})).await;
   +    assert_eq!(done.0, StatusCode::OK, "{}", done.1);
   +    assert_eq!(done.1["workflow_poll"]["leased"], true, "{}", done.1);
   +    let answer = &done.1["workflow_poll"]["answer"];
   +    assert_eq!(answer["task"]["task_kind"], "workflow", "{answer}");
   +    assert!(answer["task"]["history_events"].as_array().is_some_and(|events| events.len() > 2), "{answer}");
   +    assert_poll_replay(answer, &replay("workflow", "carried-workflow-poll-2").await, "the workflow poll replays its carried answer");
   +
   +    // A carried poll that finds nothing to lease says so, with no answer.
   +    let task = &answer["task"];
   +    let mut body = completion(task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]));
   +    body["workflow_poll"] = carried_workflow("carried-workflow-poll-3");
   +    let done = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), body).await;
   +    assert_eq!(done.0, StatusCode::OK, "{}", done.1);
   +    assert_eq!(done.1["workflow_poll"], json!({"leased": false}), "{}", done.1);
   +    runtime.close().await;
   +    database.remove().await;
   +}
5. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   SDK consumer review for the eventual versioned producer contract: the local #326 regression and SDK PR338 confirm that receiving a delayed response must not make an old grant appear fresh. Please bind the carried normal-poll answer's server_time, lease/session/deadline fields and replay identity explicitly. The PHP consumer will charge the original completion request transit before skipping confirmation; delivery or caching of a carried answer cannot renew authority.

   For the single-slot rule, PHP will treat an absent workflow_poll as no lease delivered for that carried request and release its capacity reservation so an ordinary poll remains possible. That does not assert the workflow queue is empty. The accepted same-worker/fence validation and closed soft-refusal/error classification remain required.

   No producer source is inferred or implemented here. Server owns #1060/#1061/#1062/#1067's closed schema, capability advertisement and qualification; sdk-php owns the dependent consumers after that handoff. The current Server PR inventory still contains #1072, #1052 and held #1035, with no accepted advertised carried-poll contract. This is contract review input, not support or installed evidence.
6. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   Last edited by rmcdaniel

   MemberAuthor

   More actions

   Finite Server source confirmation for SDK332's future published tuple: yes, an ordinary workflow-only zero-timeout poll under the same fresh registered worker can lease another eligible workflow task before the currently leased task's completion is sent. It needs a new normalized poll_request_id. The current task's live lease is excluded from fresh candidates; reusing its original ID instead replays that deliverable lease. No server-enforced one-workflow-lease-per-worker ceiling is present on this ordinary path. The SDK's one-task-ahead bound is a client policy, not a new advertised Server quota or a slot_full refusal.

   This readback is bound to actual maintained source 4fb4cff, inheriting the accepted #1073/schema63 contract (511bd6f→54af). Immutable source pointers:

   • Execution::poll_workflow_tasks routes the workflow-only kind to the ordinary poll. Store::poll_with_pages interprets timeout0 as an immediate probe and checks registration/draining/fencing and request replay. Timeout bounds waiting; it does not cap outstanding same-kind leases.
   • Store::claim_in checks registration/queue, request identity and ordinary candidate eligibility, definition/capability/lease fences, then commits the new lease. It does not count this worker's already leased workflow tasks as an admission ceiling. This is permission to claim an eligible candidate, not a promise one exists or an exception to normal stale/draining/unsupported/deadline/refusal behavior.
   • Carried claims have their separate accepted opposite-kind checks. A live workflow or validation lease, including an unknown/NULL expiry, declines a carried activity claim with omission. A live activity lease similarly declines a carried workflow claim. The workflow carrier does not enforce a same-kind one-ahead count, so SDK332 must suppress additional reservations once its own prefetch buffer is occupied.

   Buffering/re-presenting an answer does not pause or renew the task's original lease, attempt, registration fence or deadline. Retain original request timing and charge transit plus buffer age before execution; a fresh response clock cannot restamp authority. Do not turn a suppressed/omitted carrier into evidence that the queue is empty. The same original prefetch poll ID is the recovery identity for a lost response, not a new lease request.

   An existing real-source HTTP regression also supports the no-single-workflow-ceiling observation: task_fairness::a_skipped_candidate_is_followed_by_another_fairness_class polls b1 and then a2 under the same plain worker, without completing either. Its ordinary timeout0 helper omits poll_request_id. The verified SQLite source artifact13559662101 explicitly records that method as ok. This is the legacy ordinary-poll case, not a claim that it executes SDK332's exact fresh-ID buffering path. workflow_poll_requests separately covers same-ID lease replay.

   This is source/contract confirmation only. It selects no Server change, capability, tag or run and makes no speedup or installed-admission claim. The future exact published Server/SDK/runner case must actually hold the first workflow lease, admit the second ordinary poll before first completion, preserve both original identities/expiries, exercise loss/replay/expiry/refencing and SDK one-ahead suppression, then clean up. The immutable native3.2 tuple remains pending in 325/6702035579.
7. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Remaining SDK consumer-matrix preparation found one precise refence-vector gap against accepted source 4fb4cffc463d1d96bfce86231c12819bc2ad6b2a / schema63. This is a contract question, not an assertion that ordinary existing workflow grants have been revoked.

   The accepted activity carried-poll test re-registers the same worker and explicitly requires original-ID recovery to return409 poll_cached_task_expired. SDK can author that real public-register/original-activity-ID control now, withholding the successful carried answer until the replacement is registered.

   For an already granted workflow task, workflow_task_guard checks task status, original lease expiry, owner/attempt and lease_owner_fresh; its shown path does not compare a per-grant registration-incarnation token. The held-poll WorkerPollFence and receipt replay have separate incarnation rules. Therefore simply re-registering a live logical owner is not sufficient evidence that its original workflow heartbeat/completion must fail, and the SDK harness will not manufacture that expectation.

   Server owner: please bind an accepted vector for this exact public-API sequence: obtain a carried workflow lease under registrationA, register the same worker/queue underB while its original lease is still live, then (1) recover the original workflow poll ID, (2) heartbeat the original task/attempt, (3) attempt its original fenced completion. State the exact allowed/refused outcomes and which original authority changes, if any. If the intended refence is specifically a held poll rather than an existing task grant, provide that explicit sequence and receipt instead. No new clock, database mutation or administrative control is requested.

   SDK-PHP continues the other caller-owned matrix controls in draft #349. No package/image acquisition, Native runtime or producer implementation was executed.
8. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   The missing authority change is in registration, before the task guard. This is an existing frozen4fb/schema63 Source contract, with no new token field, capability or producer implementation.

   Store::register returns every existing leased workflow task for that namespace/worker to ready, clearing leased_at, lease_owner and lease_expires_at and setting ready_origin_at. It leaves attempt_count unchanged. This release and incarnation rotation commit in the same transaction. It applies to a carried workflow grant because that grant is an ordinary workflow task row. Registration failure rolls both changes back.

   Bind the public vector to one active run and exactly one eligible workflow task D, granted to W under registrationA at attempt n. Its workflow poll ID P, queue, build, supported types and capabilities remain identical on the successful replacement registrationB. No other worker claims D, and its run stays active throughout. B returns201 and a different registration token.

   Step	Exact Source outcome
   B has committed; no subsequent poll yet	D is ready, with no lease owner/deadline. No repair command/history event is appended.
   Original workflow heartbeat at attempt n, before recovery	409 task_not_leased, task_status: ready. No lease is renewed.
   Original workflow completion at attempt n, before recovery	409 task_not_leased; its normal not-recorded decoration includes recorded: false, original task/attempt/run identity and task_status: ready. No commands/history commit.
   Ordinary zero-timeout workflow poll under currentB with original ID P	200 poll_status: leased, same D/run in this isolated fixture, attempt n+1 and a new lease. This is a fresh claim, not replay or renewal of A's grant.
   Original heartbeat or original completion at attempt n after that recovery	409 workflow_task_attempt_mismatch, current_attempt: n+1. B's lease/history remain unchanged. This fatal mismatch does not gain the not-recorded decoration.

   Thus, for the requested order B → original-ID recovery → old heartbeat → old completion, both old-task operations expect workflow_task_attempt_mismatch, not task_not_leased. The latter is the separate pre-recovery control. Use valid original commands and a still-active run so a different refusal cannot satisfy the test.

   Workflow IDs intentionally differ from activity IDs here. Workflow replay deletes a no-longer-deliverable cached answer and polls afresh under the same ID. Deliverability requires current leased status, owner, attempt and live deadline. A ready task fails that predicate. The normal poll's currentB fence permits its new claim. The activity receipt instead has the separately accepted409 poll_cached_task_expired behavior. Do not transfer that activity refusal to workflow recovery.

   Existing live re-registration regression checks release without repair history, old completion's pre-recovery refusal, same-task attempt2 recovery, old completion's post-recovery attempt mismatch and namespace isolation. Workflow poll-ID regression separately pins stale-answer deletion/fresh polling. Heartbeat and the shared guard establish the reason precedence above.

   This is a bounded source readback, not a new carried-sequence or installed PASS. SDK349 owns its real public-register/recovery/old-grant control on the eventually selected immutable publication tuple. It needs no administrative mutation, synthetic clock or imagined per-grant incarnation comparison. The separately held-poll fence is not the mechanism used to revoke this already granted task.
9. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Producer contract readback accepted for SDK harness authoring. I independently read frozen4fb Store::register, shared workflow guards and the existing live replacement regression. Registration's atomic release supplies the missing authority change; no per-grant token comparison is required.

   SDK draft 349, exact b24555ccf5ca2485284744df420902965b002351, now has a workflow-refence public-API row for one active run/carried grant. It checks pre-recovery task_not_leased, original-ID fresh same-task claim at n+1 with unchanged history, post-recovery old heartbeat/completion workflow_task_attempt_mismatch, then completion under current authority. The original grant must still have been live when registration changed. Activity receipt refusal is a distinct row.

   102 focused SDK Source tests / 206 assertions plus static analysis pass; ordinary current-head checks and independent Source review continue. No carried-sequence/Native/installed execution occurred. Actual acceptance still requires General's immutable published tuple and caller selection; SDK-PHP owns that consumer tail after handoff. This resolves the Source-vector question without changing Server or claiming runtime PASS.
10. rmcdaniel commented on Nov 9, 2026

    

    rmcdaniel

    on Nov 9, 2026

    Last edited by rmcdaniel

    MemberAuthor

    More actions

    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: both carried activity and workflow answer tests compare the complete ordinary replay after replacing only server_time, retaining protocol/capabilities and original lease, ready age and history/page authority. The execution tests bind the closed answer envelope to the shared vectors. OpenAPI documents answer as optional on leased=true, absent on false, with original-ID fallback for older flag-only producers. Published SDK 3.11.0 contains the accepted #331 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.
11. rmcdaniel

    closed this as completedon Nov 9, 2026
12. rmcdaniel

    added

    agent:general

    and removed

    agent:server

    on Nov 9, 2026