diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index 054370bfd..816028756 100644
--- a/rust/src/runtime/store.rs
+++ b/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,9 +3092,20 @@ 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.
@@ -3096,7 +3113,9 @@ where
             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
diff --git a/rust/src/runtime/workflow_task_guard.rs b/rust/src/runtime/workflow_task_guard.rs
index b1b24b9f3..71fe3e2b2 100644
--- a/rust/src/runtime/workflow_task_guard.rs
+++ b/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));
+        }
+    }
+}
diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
index 327e4a27c..c7b547bd4 100644
--- a/rust/tests/execution.rs
+++ b/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"]
diff --git a/rust/tests/execution/late_workflow_claims.rs b/rust/tests/execution/late_workflow_claims.rs
index 3e34d0d06..782b28c03 100644
--- a/rust/tests/execution/late_workflow_claims.rs
+++ b/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,
diff --git a/rust/tests/execution/not_recorded_receipts.rs b/rust/tests/execution/not_recorded_receipts.rs
new file mode 100644
index 000000000..b33ea7a31
--- /dev/null
+++ b/rust/tests/execution/not_recorded_receipts.rs
@@ -0,0 +1,298 @@
+use super::*;
+
+// Server #994, to Security's contract (issue comment 6613395001): a workflow
+// task report (completion, failure or cancellation scope checkpoint) refused
+// `task_not_leased` or `run_closed` has recorded nothing, and its 409 carries
+// the full not-recorded receipt in the worker envelope: `recorded: false`,
+// the request's exact `task_id` and `workflow_task_attempt`, and the task
+// row's own `run_id`, beside the refusal's fields. Every other refusal stays
+// fatal to a worker and gains none of it.
+
+const ROUTES: [&str; 3] = ["complete", "fail", "checkpoint"];
+
+/// A late report on `route` for `task`, claimed by `owner` at `attempt`.
+async fn report(app: &Router, task: &Value, route: &str, owner: &str, attempt: i64) -> (StatusCode, Value) {
+    let task_id = task["task_id"].as_str().unwrap();
+    let claim = json!({"lease_owner":owner,"workflow_task_attempt":attempt});
+    let with = |fields: Value| {
+        let mut body = claim.clone();
+        body.as_object_mut().unwrap().extend(fields.as_object().unwrap().clone());
+        body
+    };
+    match route {
+        "complete" => {
+            let body = with(json!({"commands":[{"type":"complete_workflow","result":envelope(Payload::Long(7))}]}));
+            request(app, "POST", &format!("/api/worker/workflow-tasks/{task_id}/complete"), body).await
+        }
+        "fail" => {
+            let body = with(json!({"failure":{"type":"RuntimeException","message":"late report"}}));
+            request(app, "POST", &format!("/api/worker/workflow-tasks/{task_id}/fail"), body).await
+        }
+        "waiting" => {
+            let body = with(json!({"failure":{"type":"WorkflowTaskWaitingForHistory","message":"history is not ready yet"}}));
+            request(app, "POST", &format!("/api/worker/workflow-tasks/{task_id}/fail"), body).await
+        }
+        _ => {
+            let body = with(json!({"checkpoint_id":"late","start_sequence":1,
+                "commands":[{"type":"start_timer","delay_seconds":60}]}));
+            let path = format!("/api/worker/workflow-tasks/{task_id}/cancellation-scopes/checkpoint");
+            request_protocol(app, "POST", &path, body, "1.20").await
+        }
+    }
+}
+
+/// The refusal's public worker body without its capability manifest, which
+/// must be present.
+fn public_body(refused: &(StatusCode, Value), label: &str) -> Value {
+    let (status, body) = refused;
+    assert_eq!(*status, StatusCode::CONFLICT, "{label}: {body}");
+    let mut body = body.clone();
+    let fields = body.as_object_mut().unwrap();
+    assert!(fields.remove("server_capabilities").is_some_and(|manifest| manifest.is_object()), "{label}: {body}");
+    body
+}
+
+/// The exact receipt a not-recorded refusal must carry.
+fn receipt(task: &Value, attempt: i64) -> Value {
+    json!({"task_id":task["task_id"],"workflow_task_attempt":attempt,"recorded":false,"run_id":task["run_id"]})
+}
+
+fn merged(base: Value, extra: Value) -> Value {
+    let mut base = base;
+    base.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone());
+    base
+}
+
+async fn claimed(app: &Router, workflow: &str) -> (Value, Value) {
+    register(app, "nr-worker", json!(["echo"]), json!([])).await;
+    let started = start(app, workflow).await;
+    let task = poll(app, "nr-worker", "workflow").await;
+    assert_eq!(task["run_id"], started["run_id"], "{task}");
+    assert!(task["run_id"].as_str().is_some_and(|run| !run.is_empty()));
+    (started, task)
+}
+
+/// Durable state a late report must leave unchanged: every table's row
+/// count, the run's history and the stored completion receipts.
+async fn durable(database: &TestDatabase, app: &Router, started: &Value) -> Value {
+    json!({"tables":table_counts(database).await,
+        "history":run_history(app, &started["workflow_id"], &started["run_id"]).await,
+        "receipts":text_rows(database, "SELECT task_id, receipt FROM dw_task_completions ORDER BY task_id",
+            "SELECT task_id, receipt::text FROM dw_task_completions ORDER BY task_id",
+            "SELECT task_id, CAST(receipt AS CHAR) FROM dw_task_completions ORDER BY task_id").await})
+}
+
+/// `run_closed` for every terminal status the guard recognizes, on every
+/// report route: the guard's closed-run body plus the receipt, nothing
+/// recorded.
+#[tokio::test]
+async fn a_report_on_every_closed_run_status_is_run_closed_with_the_full_receipt() {
+    for close in ["completed", "failed", "timed_out", "cancelled", "terminated"] {
+        let database = TestDatabase::new().await;
+        let runtime = database.open().await.unwrap();
+        let app = router(runtime.clone());
+        let workflow = format!("nr-closed-{close}");
+        let (started, task) = claimed(&app, &workflow).await;
+        let run_id = started["run_id"].as_str().unwrap();
+        // How the run closes while the worker's report is late.
+        let (run_status, stop_reason, closed_reason, task_status, cancel_requested) = match close {
+            "completed" | "failed" => {
+                let command = if close == "completed" {
+                    json!({"type":"complete_workflow","result":envelope(Payload::Long(1))})
+                } else {
+                    json!({"type":"fail_workflow","message":"boom"})
+                };
+                let done = finish_task(&app, &task, json!([command])).await;
+                assert_eq!(done.0, StatusCode::OK, "{close}: {}", done.1);
+                // A terminal fail_workflow leaves its task failed.
+                let (stop, task_status) = if close == "completed" { ("run_completed", "completed") } else { ("run_failed", "failed") };
+                (close, stop, close, task_status, false)
+            }
+            "timed_out" => {
+                database.execute("UPDATE workflow_runs SET run_deadline_at='2000-01-01 00:00:00'").await;
+                let timed_out = report(&app, &task, "complete", "nr-worker", 1).await;
+                assert_eq!(
+                    (timed_out.0, &timed_out.1["reason"], &timed_out.1["recorded"]),
+                    (StatusCode::CONFLICT, &json!("run_timed_out"), &json!(false)),
+                    "{}",
+                    timed_out.1
+                );
+                ("failed", "run_failed", "timed_out", "completed", false)
+            }
+            _ => {
+                let path = format!("/api/workflows/{workflow}/runs/{run_id}/{}", if close == "cancelled" { "cancel" } else { "terminate" });
+                let closed = request(&app, "POST", &path, json!({"reason":"operator"})).await;
+                assert_eq!(closed.0, StatusCode::OK, "{close}: {}", closed.1);
+                let stop = if close == "cancelled" { "run_cancelled" } else { "run_terminated" };
+                (close, stop, close, "cancelled", true)
+            }
+        };
+        let before = durable(&database, &app, &started).await;
+        for route in ROUTES {
+            let label = format!("{close} {route}");
+            let mut body = public_body(&report(&app, &task, route, "nr-worker", 1).await, &label);
+            let closed_at = body.as_object_mut().unwrap().remove("run_closed_at").unwrap_or_default();
+            assert!(closed_at.as_str().is_some_and(|at| at.ends_with('Z') && at.len() == 27), "{label}: {closed_at}");
+            assert_eq!(
+                body,
+                merged(
+                    receipt(&task, 1),
+                    json!({"error":"Workflow run is already closed.","reason":"run_closed","protocol_version":"1.20",
+                        "stop_reason":stop_reason,"cancel_requested":cancel_requested,"can_continue":false,
+                        "run_status":run_status,"run_closed_reason":closed_reason,"task_status":task_status,
+                        "lease_owner":"nr-worker","lease_expires_at":null})
+                ),
+                "{label}"
+            );
+        }
+        assert_eq!(durable(&database, &app, &started).await, before, "{close}: the late reports recorded nothing");
+        runtime.close().await;
+        database.remove().await;
+    }
+}
+
+/// `task_not_leased` for a task completed on a run that is still open, and
+/// for the waiting acknowledgement's duplicate path, whose task holds no
+/// completion receipt.
+#[tokio::test]
+async fn a_report_for_a_finished_task_on_an_open_run_is_task_not_leased_with_the_full_receipt() {
+    for finish in ["completed", "waiting"] {
+        let database = TestDatabase::new().await;
+        let runtime = database.open().await.unwrap();
+        let app = router(runtime.clone());
+        let (started, task) = claimed(&app, &format!("nr-not-leased-{finish}")).await;
+        if finish == "completed" {
+            let done = finish_task(&app, &task, json!([{"type":"start_timer","delay_seconds":3600}])).await;
+            assert_eq!(done.0, StatusCode::OK, "{}", done.1);
+        } else {
+            let acknowledged = report(&app, &task, "waiting", "nr-worker", 1).await;
+            assert_eq!(
+                (acknowledged.0, &acknowledged.1["outcome"]),
+                (StatusCode::OK, &json!("waiting_for_history")),
+                "{}",
+                acknowledged.1
+            );
+        }
+        let before = durable(&database, &app, &started).await;
+        for route in ["complete", "fail", "waiting", "checkpoint"] {
+            let label = format!("{finish} {route}");
+            assert_eq!(
+                public_body(&report(&app, &task, route, "nr-worker", 1).await, &label),
+                merged(
+                    receipt(&task, 1),
+                    json!({"error":"Workflow task is not currently leased.","reason":"task_not_leased",
+                        "protocol_version":"1.20","task_status":"completed"})
+                ),
+                "{label}"
+            );
+        }
+        assert_eq!(durable(&database, &app, &started).await, before, "{finish}: the late reports recorded nothing");
+        runtime.close().await;
+        database.remove().await;
+    }
+}
+
+/// A run closed by another path while its task kept the lease passes the
+/// ownership guard and is refused by the open-run check: the bare refusal
+/// becomes the same receipt in the worker envelope.
+#[tokio::test]
+async fn a_report_on_a_closed_run_whose_task_is_still_leased_carries_the_full_receipt() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let (started, task) = claimed(&app, "nr-leased-closed").await;
+    database.execute("UPDATE workflow_runs SET status='failed'").await;
+    let before = durable(&database, &app, &started).await;
+    for route in ["complete", "fail"] {
+        assert_eq!(
+            public_body(&report(&app, &task, route, "nr-worker", 1).await, route),
+            merged(
+                receipt(&task, 1),
+                json!({"reason":"run_closed","message":"run_closed","protocol_version":"1.20","task_status":"leased"})
+            ),
+            "{route}"
+        );
+    }
+    assert_eq!(durable(&database, &app, &started).await, before);
+    runtime.close().await;
+    database.remove().await;
+}
+
+/// The refusals that are not settlement authority keep their published
+/// bodies: no `recorded`, no `run_id`.
+#[tokio::test]
+async fn fatal_report_refusals_gain_no_receipt() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let (started, replaced) = claimed(&app, "nr-fatal").await;
+    // Registering the same worker again releases its lease; the task is
+    // claimed again as attempt 2.
+    register(&app, "nr-worker", json!(["echo"]), json!([])).await;
+    let current = poll(&app, "nr-worker", "workflow").await;
+    assert_eq!((&current["task_id"], &current["workflow_task_attempt"]), (&replaced["task_id"], &json!(2)));
+    let before = durable(&database, &app, &started).await;
+    let fence = |attempt: i64, fields: Value| {
+        merged(json!({"task_id":replaced["task_id"],"workflow_task_attempt":attempt,"protocol_version":"1.20"}), fields)
+    };
+    for route in ROUTES {
+        assert_eq!(
+            public_body(&report(&app, &replaced, route, "nr-worker", 1).await, route),
+            fence(1, json!({"error":"Workflow task lease attempt does not match the current claim.",
+                "reason":"workflow_task_attempt_mismatch","current_attempt":2})),
+            "{route}: attempt mismatch"
+        );
+        assert_eq!(
+            public_body(&report(&app, &replaced, route, "nr-other-worker", 2).await, route),
+            fence(2, json!({"error":"Workflow task lease is owned by another worker.",
+                "reason":"lease_owner_mismatch","lease_owner":"nr-worker"})),
+            "{route}: owner mismatch"
+        );
+    }
+    database.set_time(false, "2000-01-01T00:00:00Z").await;
+    for route in ROUTES {
+        let body = public_body(&report(&app, &replaced, route, "nr-worker", 2).await, route);
+        assert_eq!(body["reason"], "lease_expired", "{route}: {body}");
+        assert!(body.get("recorded").is_none() && body.get("run_id").is_none(), "{route}: {body}");
+    }
+    assert_eq!(durable(&database, &app, &started).await, before);
+    runtime.close().await;
+    database.remove().await;
+}
+
+/// The shared checkpoint wrapper: a cooperative claim still leased whose
+/// run closes after the ownership guard reaches Core's checkpoint, whose
+/// refused `run_closed` answer carries the same receipt.
+#[tokio::test]
+async fn a_checkpoint_whose_run_closed_after_the_guard_carries_the_full_receipt() {
+    for status in ["completed", "failed", "cancelled", "terminated"] {
+        let database = TestDatabase::new().await;
+        let runtime = database.open().await.unwrap();
+        let app = router(runtime.clone());
+        let registered = request_protocol(&app, "POST", "/api/worker/register",
+            json!({"worker_id":"nr-worker","task_queue":"test","runtime":"php","supported_workflow_types":["echo"],
+                "supported_activity_types":[],"capabilities":["cooperative_cancellation"]}), "1.20").await;
+        assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1);
+        let started = start(&app, &format!("nr-checkpoint-{status}")).await;
+        let claimed = request_protocol(&app, "POST", "/api/worker/workflow-tasks/poll",
+            json!({"worker_id":"nr-worker","task_queue":"test","timeout_seconds":2}), "1.20").await;
+        let task = claimed.1["task"].clone();
+        assert_eq!(task["run_id"], started["run_id"], "{}", claimed.1);
+        let sql: &'static str = Box::leak(format!("UPDATE workflow_runs SET status='{status}'").into_boxed_str());
+        database.execute(sql).await;
+        let before = durable(&database, &app, &started).await;
+        let label = format!("{status} checkpoint");
+        assert_eq!(
+            public_body(&report(&app, &task, "checkpoint", "nr-worker", 1).await, &label),
+            merged(
+                receipt(&task, 1),
+                json!({"checkpointed":false,"duplicate":false,"reason":"run_closed","task_status":"leased",
+                    "history_refresh_page_token":"MA==","protocol_version":"1.20"})
+            ),
+            "{label}"
+        );
+        assert_eq!(durable(&database, &app, &started).await, before, "{label}: nothing recorded");
+        runtime.close().await;
+        database.remove().await;
+    }
+}
diff --git a/scripts/conformance/server-parity/late-workflow-completion-contract.mjs b/scripts/conformance/server-parity/late-workflow-completion-contract.mjs
index e332d56ee..c4fc62bc7 100644
--- a/scripts/conformance/server-parity/late-workflow-completion-contract.mjs
+++ b/scripts/conformance/server-parity/late-workflow-completion-contract.mjs
@@ -40,10 +40,23 @@ export function checkLateWorkflowCompletion(fixture, observation, workflowId) {
   equal(state.heartbeat_while_leased.status, 409, 'late lease renewal refused while the new claim holds the lease');
   equal(body(state.heartbeat_while_leased), body(state.while_leased), 'renewal refusal matches the completion refusal');
 
-  for (const key of ['after_completion', 'heartbeat_after_completion', 'failure_after_completion']) {
+  // Published PHP's ownership guard (lines 4737-4775 at tag 2.11.8) sends
+  // neither `recorded` nor `run_id` here; it reports them only from its rarer
+  // post-guard engine outcome. Native adds the not-recorded receipt to every
+  // refused report (server#994), naming the run from the task row. Lease
+  // renewals record nothing either way and keep the published body.
+  const {implementation} = state;
+  assert.ok(implementation === null || implementation === 'rust', `published PHP or native, not ${implementation}`);
+  if (implementation === 'rust') {
+    assert.ok(typeof state.peer_run_id === 'string' && state.peer_run_id !== '', 'authoritative run id');
+  }
+  const receipt = implementation === 'rust' ? {recorded: false, run_id: state.peer_run_id} : {};
+  const notLeased = {...fence, error: 'Workflow task is not currently leased.', reason: 'task_not_leased',
+    task_status: 'completed'};
+  for (const [key, expected] of [['after_completion', {...notLeased, ...receipt}],
+    ['heartbeat_after_completion', notLeased], ['failure_after_completion', {...notLeased, ...receipt}]]) {
     equal(state[key].status, 409, `${key} refused`);
-    equal(body(state[key]), {...fence, error: 'Workflow task is not currently leased.', reason: 'task_not_leased',
-      task_status: 'completed'}, `${key}: the completed task is not leased`);
+    equal(body(state[key]), expected, `${key}: the completed task is not leased (${implementation ?? 'php'})`);
   }
   equal(state.history_while_open.map(event => event.event_type), definition.expected_events.slice(0, 3),
     'late requests recorded nothing while the run was open');
@@ -56,7 +69,7 @@ export function checkLateWorkflowCompletion(fixture, observation, workflowId) {
   equal(closedFields, {task_id: lastReplaced.task_id, workflow_task_attempt: 1, 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', task_status: 'completed', lease_owner: worker,
-    lease_expires_at: null}, 'run closed refusal: the completed task keeps its owner and has no lease expiry');
+    lease_expires_at: null, ...receipt}, 'run closed refusal: the completed task keeps its owner and has no lease expiry');
   equal(state.history.map(event => event.event_type), definition.expected_events, 'one completion recorded');
   return projection;
 }
diff --git a/scripts/conformance/server-parity/late-workflow-completion-probe.php b/scripts/conformance/server-parity/late-workflow-completion-probe.php
index de65b956d..95bcf8f6d 100644
--- a/scripts/conformance/server-parity/late-workflow-completion-probe.php
+++ b/scripts/conformance/server-parity/late-workflow-completion-probe.php
@@ -41,7 +41,11 @@ function finishHttpLateWorkflowCompletion(Client $client, array $fixture, string
     $complete = [['type' => 'complete_workflow', 'result' => $client->payloadCodec()->envelope($fixture['input'])]];
     $late = static fn (array $task, array $commands): array => $receipt(static fn (): array => $client->completeWorkflowTask(
         $task['task_id'], $worker, $task['workflow_task_attempt'], $commands));
-    $state = ['applicable' => true, 'worker_id' => $worker, 'task_queue' => $queue, 'peer_workflow_id' => $peerId];
+    // Native adds the not-recorded receipt (server#994); published PHP omits
+    // `implementation` from cluster info. A failed read throws, so the
+    // recording never guesses which bodies to expect.
+    $state = ['applicable' => true, 'implementation' => $client->clusterInfo()->raw['implementation'] ?? null,
+        'worker_id' => $worker, 'task_queue' => $queue, 'peer_workflow_id' => $peerId];
     $register();
     $handle = $client->startWorkflow($definition['workflow_type'], $peerId, $queue, [$fixture['input']]);
     $state['peer_run_id'] = $handle->selectedRunId;
diff --git a/tests/Fixtures/ServerParity/late-workflow-completion.json b/tests/Fixtures/ServerParity/late-workflow-completion.json
index 42006e41d..28d51b1e2 100644
--- a/tests/Fixtures/ServerParity/late-workflow-completion.json
+++ b/tests/Fixtures/ServerParity/late-workflow-completion.json
@@ -43,5 +43,5 @@
       "WorkflowCompleted"
     ]
   },
-  "contract": "A worker process holding a workflow task claim is replaced by a new registration of the same worker_id, which releases the lease; the task is claimed again as attempt 2. Requests from the replaced attempt 1 are refused with the published ownership guard bodies a worker reads to recognize a lost claim. While attempt 2 holds the lease, completion and lease renewal are refused with 409 workflow_task_attempt_mismatch and the current attempt. After attempt 2 completes the task while the run is open, completion, renewal and failure are refused with 409 task_not_leased and task_status completed. After the run closes, a late completion is refused with 409 run_closed, can_continue false, the run status and closure, and the completed task, which keeps its lease owner and has no lease expiry. No late request records history; each task records one completion. Embedded applications have no HTTP worker registration."
+  "contract": "A worker process holding a workflow task claim is replaced by a new registration of the same worker_id, which releases the lease; the task is claimed again as attempt 2. Requests from the replaced attempt 1 are refused with the published ownership guard bodies a worker reads to recognize a lost claim. While attempt 2 holds the lease, completion and lease renewal are refused with 409 workflow_task_attempt_mismatch and the current attempt. After attempt 2 completes the task while the run is open, completion, renewal and failure are refused with 409 task_not_leased and task_status completed. After the run closes, a late completion is refused with 409 run_closed, can_continue false, the run status and closure, and the completed task, which keeps its lease owner and has no lease expiry. No late request records history; each task records one completion. Published PHP's ownership guard sends these refusals without recorded or run_id (it reports both only from its rarer post-guard engine outcome, which a deterministic late request does not reach); native adds the not-recorded receipt, exactly recorded false and the task's run_id, to the refused completion, failure and closed-run reports, while renewal refusals keep the published body. The recording names the implementation from cluster info, and any other implementation is refused. Embedded applications have no HTTP worker registration."
 }
diff --git a/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs b/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
index 848d689f0..3119e3d1c 100644
--- a/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
+++ b/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs
@@ -13,7 +13,7 @@ function model() {
     reason: 'workflow_task_attempt_mismatch', current_attempt: 2});
   const notLeased = refusal({error: 'Workflow task is not currently leased.', reason: 'task_not_leased', task_status: 'completed'});
   const events = fixture.late_workflow_completion.expected_events.map(event_type => ({event_type}));
-  return structuredClone({mode: 'http', late_workflow_completion: {applicable: true, worker_id: worker,
+  return structuredClone({mode: 'http', late_workflow_completion: {applicable: true, implementation: null, worker_id: worker,
     task_queue: 'test-late', peer_workflow_id: 'test-late-peer', peer_run_id: 'peer-run',
     first_claims: claims('task-1'), while_leased: mismatch, heartbeat_while_leased: mismatch,
     scheduled: {recorded: true}, after_completion: notLeased, heartbeat_after_completion: notLeased,
@@ -24,6 +24,48 @@ function model() {
       run_closed_at: '2026-10-20T17:33:21.305246Z', task_status: 'completed', lease_owner: worker, lease_expires_at: null}},
     history: events}});
 }
+// Native's bodies: the published ones plus the not-recorded receipt on each
+// refused report (server#994). Renewal and attempt-mismatch bodies are unchanged.
+function nativeModel() {
+  const observation = model();
+  const state = observation.late_workflow_completion;
+  state.implementation = 'rust';
+  for (const key of ['after_completion', 'failure_after_completion', 'after_close']) {
+    Object.assign(state[key].response, {recorded: false, run_id: state.peer_run_id});
+  }
+  state.heartbeat_after_completion = structuredClone(model().late_workflow_completion.heartbeat_after_completion);
+  return observation;
+}
+test('native lost-claim refusals with the not-recorded receipt pass', () =>
+  assert.equal(checkLateWorkflowCompletion(fixture, nativeModel(), 'test').scope, 'late_workflow_completion'));
+for (const [label, build, mutate] of [
+  ['published PHP refusals carrying native receipt fields', model,
+    s => Object.assign(s.after_completion.response, {recorded: false, run_id: s.peer_run_id})],
+  ['published PHP closed-run refusal carrying recorded', model, s => s.after_close.response.recorded = false],
+  ['native refusals without the receipt (3.0.4 bodies)', model, s => s.implementation = 'rust'],
+  ['native completion refusal without run_id', nativeModel, s => delete s.after_completion.response.run_id],
+  ['native failure refusal without recorded', nativeModel, s => delete s.failure_after_completion.response.recorded],
+  ['native closed-run refusal without the receipt', nativeModel,
+    s => { delete s.after_close.response.recorded; delete s.after_close.response.run_id; }],
+  ['native receipt that claims a record', nativeModel, s => s.after_completion.response.recorded = true],
+  ['native receipt naming another run', nativeModel, s => s.after_close.response.run_id = 'other-run'],
+  ['native renewal refusal carrying the receipt', nativeModel,
+    s => Object.assign(s.heartbeat_after_completion.response, {recorded: false, run_id: s.peer_run_id})],
+  ['native attempt mismatch carrying the receipt', nativeModel,
+    s => { Object.assign(s.while_leased.response, {recorded: false, run_id: s.peer_run_id});
+      Object.assign(s.heartbeat_while_leased.response, {recorded: false, run_id: s.peer_run_id}); }],
+  ['native without an authoritative run id', nativeModel, s => s.peer_run_id = ''],
+  // Published bodies under an unrecognized name: only the implementation check refuses them.
+  ['an unknown implementation with published bodies', model, s => s.implementation = 'php'],
+  ['an unknown implementation with native bodies', nativeModel, s => s.implementation = 'native'],
+  ['an implementation the recording did not name', model, s => delete s.implementation],
+]) {
+  test(`rejects ${label}`, () => {
+    const observation = build();
+    mutate(observation.late_workflow_completion);
+    assert.throws(() => checkLateWorkflowCompletion(fixture, observation, 'test'));
+  });
+}
 test('published lost-claim refusals pass', () =>
   assert.equal(checkLateWorkflowCompletion(fixture, model(), 'test').scope, 'late_workflow_completion'));
 test('embedded explicitly declares the HTTP worker lifecycle inapplicable', () => checkLateWorkflowCompletion(fixture,