Native: return the full not-recorded receipt for refused workflow-task reports (#994)

rust/src/runtime/store.rs

@@ -2699,9 +2699,13 @@ 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, self.worker_stale_after_seconds).await?;
+        Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds)
+            .await
+            .map_err(|error| Self::not_recorded(&task, &body, error))?;
         let run_id = DB::string(&task, "workflow_run_id")?;
-        let run = Self::active_run(&mut tx, &run_id).await?;
+        let run = Self::active_run(&mut tx, &run_id)
+            .await
+            .map_err(|error| Self::not_recorded(&task, &body, error))?;
         // Server #794: a failure or history wait reported after the run's
         // deadline closes the run as timed out, with the receipt a completion
         // gets. Published 2.11.8 acknowledged the wait and left the run waiting.
@@ -3038,7 +3042,9 @@ where
                 "outcome": "completed", "recorded": false, "reason": "already_completed", "created_task_ids": created}),
             );
         }
-        Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds).await?;
+        Self::workflow_task_guard(&mut tx, &task, &body, self.worker_stale_after_seconds)
+            .await
+            .map_err(|error| Self::not_recorded(&task, &body, error))?;
         // PHP admits cancellation scopes before it guards worker-session
         // commands, and guards both before validating streams.
         Self::cancellation_scope_admission(&mut tx, &task, &body, commands, protocol).await?;
@@ -3086,17 +3092,30 @@ where
             exceptions = super::local_activities::exceptions(&local, transition_codec.clone()).await?;
         }
         if let Some(checkpoint) = checkpoint {
+            // Server #994: Core's refused checkpoint for a run that closed
+            // after the guard is the same not-recorded receipt.
+            let identity = (DB::string(&task, "id")?, DB::string(&task, "workflow_run_id")?, DB::string(&task, "status")?);
+            let (receipt_task, receipt_run, receipt_status) = &identity;
+            let attempt = body["workflow_task_attempt"].clone();
             return self
                 .checkpoint_prefix(tx, task, &body, commands, checkpoint, transition_codec, &external, &local, &exceptions, limits, signal)
-                .await;
+                .await
+                .map(|answer| {
+                    super::workflow_task_guard::not_recorded_answer(answer, receipt_task, &attempt, receipt_run, receipt_status)
+                })
+                .map_err(|error| {
+                    super::workflow_task_guard::not_recorded(error, receipt_task, &attempt, receipt_run, receipt_status)
+                });
         }
         // `parseCommands`: at most one terminal command, anywhere in the
         // batch. It applies after every other command.
         if commands.iter().filter(|command| is_terminal(command)).count() > 1 {
             return Err(refuse(StatusCode::CONFLICT, "invalid_commands"));
         }
         let run_id = DB::string(&task, "workflow_run_id")?;
-        let run = Self::active_run(&mut tx, &run_id).await?;
+        let run = Self::active_run(&mut tx, &run_id)
+            .await
+            .map_err(|error| Self::not_recorded(&task, &body, error))?;
         // `timeoutIfDeadlineExpired`: published PHP applies a run's deadlines
         // when one of its workflow tasks completes, and closes the run.
         let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?;
@@ -4046,6 +4065,30 @@ where
         Err(refuse(StatusCode::CONFLICT, "conflicting_completion"))
     }
 
+    /// Server #994: a workflow task completion, failure or scope checkpoint
+    /// that the ownership guard or the open-run check refuses
+    /// `task_not_leased` or `run_closed` has written nothing (its transaction
+    /// is dropped), so its 409 carries the full not-recorded receipt:
+    /// `recorded: false` with the request's `workflow_task_attempt` and this
+    /// task row's own `task_id` and `run_id`, on top of the refusal's fields.
+    /// Workers settle on these fields alone and never infer them from the
+    /// reason. Every other refusal and error is returned as is.
+    fn not_recorded(task: &DB::Row, body: &Value, error: super::RuntimeError) -> super::RuntimeError {
+        let fields = (|| -> Result<(String, String, String)> {
+            Ok((DB::string(task, "id")?, DB::string(task, "workflow_run_id")?, DB::string(task, "status")?))
+        })();
+        match fields {
+            Ok((task_id, run_id, task_status)) => super::workflow_task_guard::not_recorded(
+                error,
+                &task_id,
+                &body["workflow_task_attempt"],
+                &run_id,
+                &task_status,
+            ),
+            Err(read) => read,
+        }
+    }
+
     /// 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, and finally the owner's

rust/src/runtime/workflow_task_guard.rs

@@ -24,3 +24,172 @@ pub(super) fn refusal(
         response,
     }
 }
+
+/// The refusals that settle a workflow task report as not recorded: the task
+/// is no longer leased, or its run has closed. Every other refusal (lease
+/// owner or attempt mismatch, expired lease, stale registration, conflicting
+/// completion, invalid commands) stays fatal to a worker and gains nothing.
+const NOT_RECORDED: [&str; 2] = ["task_not_leased", "run_closed"];
+
+/// The full not-recorded receipt (Server #994, Security's contract in issue
+/// comment 6613395001) on a workflow task completion, failure or scope
+/// checkpoint refused with 409 `task_not_leased` or `run_closed` before
+/// anything was written: `recorded: false` with the request's exact
+/// `task_id` and `workflow_task_attempt`, the task row's own `run_id`, and
+/// the task's status where the refusal has none. A guard refusal keeps every
+/// field it already has; a bare refusal keeps its `reason` and `message`. No
+/// payload or command data is added. Other errors pass through unchanged.
+pub(super) fn not_recorded(
+    error: RuntimeError,
+    task_id: &str,
+    attempt: &Value,
+    run_id: &str,
+    task_status: &str,
+) -> RuntimeError {
+    let mut response = match error {
+        RuntimeError::Protocol { status, response }
+            if status == StatusCode::CONFLICT
+                && response["reason"].as_str().is_some_and(|reason| NOT_RECORDED.contains(&reason)) =>
+        {
+            response
+        }
+        RuntimeError::Refused { status, reason } if status == StatusCode::CONFLICT && NOT_RECORDED.contains(&reason) => {
+            json!({"reason":reason,"message":reason,"protocol_version":super::WORKER_PROTOCOL_VERSION,
+                "server_capabilities":super::store::capabilities()})
+        }
+        other => return other,
+    };
+    receipt(&mut response, task_id, attempt, run_id, task_status);
+    RuntimeError::Protocol {
+        status: StatusCode::CONFLICT,
+        response,
+    }
+}
+
+/// The same receipt on a refused answer the checkpoint returns as its body
+/// (Core's refused checkpoint, `checkpointed: false`, rendered as 409): a
+/// `task_not_leased` or `run_closed` checkpoint refusal gains the receipt;
+/// any other answer is returned unchanged.
+pub(super) fn not_recorded_answer(
+    mut answer: Value,
+    task_id: &str,
+    attempt: &Value,
+    run_id: &str,
+    task_status: &str,
+) -> Value {
+    let refused = answer["checkpointed"] != json!(true)
+        && answer["reason"].as_str().is_some_and(|reason| NOT_RECORDED.contains(&reason));
+    if refused {
+        receipt(&mut answer, task_id, attempt, run_id, task_status);
+    }
+    answer
+}
+
+/// The receipt's identity is authoritative: the task row's id and run and
+/// the request's normalized attempt replace whatever the refusal carried.
+fn receipt(response: &mut Value, task_id: &str, attempt: &Value, run_id: &str, task_status: &str) {
+    if let Some(fields) = response.as_object_mut() {
+        let attempt = attempt.as_i64().map_or(Value::Null, Value::from);
+        fields.insert("task_id".to_owned(), json!(task_id));
+        fields.insert("workflow_task_attempt".to_owned(), attempt);
+        fields.insert("recorded".to_owned(), json!(false));
+        fields.insert("run_id".to_owned(), json!(run_id));
+        fields.entry("task_status").or_insert_with(|| json!(task_status));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_guard_refusal_keeps_its_fields_and_gains_the_receipt() {
+        let guarded = refusal("task-1", &json!(2), "Workflow task is not currently leased.", "task_not_leased",
+            json!({"task_status":"completed"}));
+        let RuntimeError::Protocol { status, mut response } = not_recorded(guarded, "task-1", &json!(2), "run-1", "leased") else {
+            panic!("a protocol refusal");
+        };
+        assert_eq!(status, StatusCode::CONFLICT);
+        assert!(response.as_object_mut().unwrap().remove("server_capabilities").is_some());
+        assert_eq!(
+            response,
+            json!({"task_id":"task-1","workflow_task_attempt":2,"error":"Workflow task is not currently leased.",
+                "reason":"task_not_leased","protocol_version":super::super::WORKER_PROTOCOL_VERSION,
+                "task_status":"completed","recorded":false,"run_id":"run-1"})
+        );
+    }
+
+    #[test]
+    fn the_receipt_identity_is_authoritative_over_a_conflicting_body() {
+        let conflicting = refusal("other-task", &json!(9), "Workflow run is already closed.", "run_closed",
+            json!({"run_id":"other-run","recorded":true}));
+        let RuntimeError::Protocol { response, .. } = not_recorded(conflicting, "task-1", &json!(2), "run-1", "completed") else {
+            panic!("a protocol refusal");
+        };
+        assert_eq!(
+            (&response["task_id"], &response["workflow_task_attempt"], &response["run_id"], &response["recorded"]),
+            (&json!("task-1"), &json!(2), &json!("run-1"), &json!(false))
+        );
+        let answer = not_recorded_answer(json!({"checkpointed":false,"duplicate":false,"task_id":"other-task",
+            "workflow_task_attempt":9,"reason":"run_closed"}), "task-1", &json!(2), "run-1", "leased");
+        assert_eq!(
+            answer,
+            json!({"checkpointed":false,"duplicate":false,"task_id":"task-1","workflow_task_attempt":2,
+                "reason":"run_closed","recorded":false,"run_id":"run-1","task_status":"leased"})
+        );
+    }
+
+    #[test]
+    fn other_checkpoint_answers_are_unchanged() {
+        for answer in [
+            json!({"checkpointed":false,"duplicate":false,"task_id":"task-1","reason":"workflow_claim_mismatch"}),
+            json!({"checkpointed":false,"reason":"active_claim_cancellation_not_supported"}),
+            json!({"checkpointed":true,"duplicate":true,"task_id":"task-1","reason":null}),
+        ] {
+            assert_eq!(not_recorded_answer(answer.clone(), "task-1", &json!(1), "run-1", "leased"), answer);
+        }
+    }
+
+    #[test]
+    fn a_bare_run_closed_refusal_becomes_the_full_receipt() {
+        let RuntimeError::Protocol { status, mut response } =
+            not_recorded(super::super::refuse(StatusCode::CONFLICT, "run_closed"), "task-1", &json!(1), "run-1", "leased")
+        else {
+            panic!("a protocol refusal");
+        };
+        assert_eq!(status, StatusCode::CONFLICT);
+        assert!(response.as_object_mut().unwrap().remove("server_capabilities").is_some_and(|c| c.is_object()));
+        assert_eq!(
+            response,
+            json!({"reason":"run_closed","message":"run_closed","protocol_version":super::super::WORKER_PROTOCOL_VERSION,
+                "task_id":"task-1","workflow_task_attempt":1,"recorded":false,"run_id":"run-1","task_status":"leased"})
+        );
+    }
+
+    #[test]
+    fn fatal_refusals_and_other_errors_pass_through_unchanged() {
+        for reason in ["lease_owner_mismatch", "workflow_task_attempt_mismatch", "lease_expired", "stale_worker_registration"] {
+            let guarded = refusal("task-1", &json!(1), "refused", reason, json!({}));
+            let RuntimeError::Protocol { response: expected, .. } = refusal("task-1", &json!(1), "refused", reason, json!({})) else {
+                unreachable!()
+            };
+            let RuntimeError::Protocol { status, response } = not_recorded(guarded, "task-1", &json!(1), "run-1", "leased") else {
+                panic!("{reason}: unchanged");
+            };
+            assert_eq!((status, response), (StatusCode::CONFLICT, expected), "{reason}");
+        }
+        for (status, reason) in [
+            (StatusCode::CONFLICT, "conflicting_completion"),
+            (StatusCode::CONFLICT, "cancellation_deadline_expired"),
+            (StatusCode::UNPROCESSABLE_ENTITY, "invalid_request"),
+            (StatusCode::NOT_FOUND, "task_not_found"),
+        ] {
+            let RuntimeError::Refused { status: kept, reason: kept_reason } =
+                not_recorded(super::super::refuse(status, reason), "task-1", &json!(1), "run-1", "leased")
+            else {
+                panic!("{reason}: unchanged");
+            };
+            assert_eq!((kept, kept_reason), (status, reason));
+        }
+    }
+}

rust/tests/execution.rs

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

rust/tests/execution/late_workflow_claims.rs

@@ -93,15 +93,19 @@ async fn a_replaced_workflow_claim_is_refused_with_the_published_ownership_field
             closed_at.ends_with('Z') && closed_at.len() == 27,
             "{action}: {refused}"
         );
-        assert_eq!(
-            refused,
-            with(
-                json!({"error":"Workflow run is already closed.","reason":"run_closed","stop_reason":"run_completed",
-                "cancel_requested":false,"can_continue":false,"run_status":"completed","run_closed_reason":"completed",
-                "run_closed_at":closed_at,"task_status":"completed","lease_owner":"late-worker","lease_expires_at":null})
-            ),
-            "{action}"
+        let mut expected = with(
+            json!({"error":"Workflow run is already closed.","reason":"run_closed","stop_reason":"run_completed",
+            "cancel_requested":false,"can_continue":false,"run_status":"completed","run_closed_reason":"completed",
+            "run_closed_at":closed_at,"task_status":"completed","lease_owner":"late-worker","lease_expires_at":null}),
         );
+        // Server #994: a refused report also carries the full not-recorded
+        // receipt; a renewal keeps the guard body.
+        if action != "heartbeat" {
+            let fields = expected.as_object_mut().unwrap();
+            fields.insert("recorded".into(), json!(false));
+            fields.insert("run_id".into(), run["run_id"].clone());
+        }
+        assert_eq!(refused, expected, "{action}");
     }
     assert_eq!(
         request(&app, "GET", &history_path, Value::Null).await.1,