From e5dee738b48d6c14d5f2146d688024172ac6312f Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Thu, 22 Oct 2026 13:58:21 +0000
Subject: [PATCH 1/6] Route pinned runs by their recorded definition like PHP

Server 2.10.0's WorkflowDefinitionRouting decides at claim which worker may
take a workflow task of a run whose start recorded its worker cohort's
definition fingerprint. Native offered such tasks to any worker.

Now a worker advertising the recorded fingerprint takes the task. A worker
with another definition waits while an eligible peer has the recorded one,
then takes it only if it advertises workflow_entry_arguments_validation.
Workers without a fingerprint, or runs without a recorded one, get nothing.
Eligible peers follow WorkflowReplayEligibility: fresh, on the run's queue
and build, supporting the type, and able to replay its cooperative
cancellation and metadata history.

Registrations and heartbeats now record their protocol version under
_worker_protocol_version in the process metrics, as PHP 2.10.0 does; worker
listings show it.
---
 .../2026-10-22-definition-routing.md          |  52 +++++
 docs/rust-server-port.md                      |  15 +-
 rust/src/runtime/definition_routing.rs        | 204 ++++++++++++++++++
 rust/src/runtime/execution.rs                 |  20 +-
 rust/src/runtime/http.rs                      |   5 +-
 rust/src/runtime/mod.rs                       |   1 +
 rust/src/runtime/query_codec_tests.rs         |   2 +-
 rust/src/runtime/store.rs                     |  15 +-
 rust/src/runtime/transition_codec_tests.rs    |   4 +-
 rust/src/runtime/worker_management.rs         |  38 +++-
 rust/tests/execution.rs                       |   4 +
 rust/tests/execution/definition_routing.rs    |  80 +++++++
 rust/tests/execution/worker_listing.rs        |   8 +-
 13 files changed, 416 insertions(+), 32 deletions(-)
 create mode 100644 docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
 create mode 100644 rust/src/runtime/definition_routing.rs
 create mode 100644 rust/tests/execution/definition_routing.rs

diff --git a/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md b/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
new file mode 100644
index 000000000..66116fb8b
--- /dev/null
+++ b/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
@@ -0,0 +1,52 @@
+# October 22, workflow definition routing
+
+- **Gap.** Native records the definition fingerprint a start's worker cohort
+  agrees on, with `workflow_definition_fingerprint_source: worker`, as PHP
+  does. It then offered that run's workflow tasks to any worker, so a worker
+  with another definition could take a pinned run while the recorded one was
+  live.
+- **Source.** Server 2.10.0's `WorkflowDefinitionRouting::forWorker`,
+  which `WorkflowTaskPoller::matchesRecordedWorkflowDefinition` applies at
+  claim, and `WorkflowReplayEligibility`.
+  - **Decision.** For a workflow task of a worker-pinned run:
+    - a worker advertising the recorded fingerprint for the type takes it
+      (`matching_definition`);
+    - without a recorded fingerprint (`recorded_definition_missing`), or
+      without one from the worker (`worker_definition_missing`), nobody
+      does;
+    - a worker with another definition waits while an eligible peer
+      advertises the recorded one (`matching_definition_preferred`);
+    - otherwise it takes the task only if it advertises
+      `workflow_entry_arguments_validation` (`replay_required`, else
+      `entry_validation_unavailable`).
+
+    An unpinned run goes to any worker (`not_worker_pinned`). Native
+    refuses to start with `DW_V2_PIN_TO_RECORDED_FINGERPRINT` off, so
+    pinning is always on.
+  - **Eligible peers** (`eligibleWorkers`). Active registrations on the
+    run's namespace and queue that:
+    - match the build the task or run requires;
+    - support the type, with workflow capacity above zero;
+    - replay the run (`WorkflowReplayEligibility::allows`). That means a
+      heartbeat inside the stale window and cooperative-cancellation support
+      when the run has a request, at the protocol version the worker last
+      reported. It also means the memo and typed search attribute
+      capabilities its history needs.
+  - **Protocol version.** Each registration and heartbeat records its
+    protocol version under `_worker_protocol_version` in the worker's process
+    metrics, after the metrics it reported. A heartbeat without metrics
+    keeps the stored ones. Worker listings therefore show the key, as PHP
+    2.10.0's do; 2.5.14 had no such key.
+- **Remaining gap** (the port's gap list): the `definition_routing` reports in
+  describe, repair, debug and task queues.
+- **Tests** (`definition_routing.rs`).
+  - A pinned run goes to the worker with its recorded definition. Workers
+    with another definition, or without one, wait even when they validate
+    entry arguments.
+  - Once the recorded definition leaves, only an entry-validating worker
+    takes the run.
+  - An unpinned run goes to any worker.
+  - Registrations and heartbeats record the protocol version with the
+    reported metrics.
+  - The worker listing test now expects that key.
+  - Full SQLite suite: 397 passed.
diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index bd7b02079..432f03082 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -182,15 +182,12 @@ contract fixture before its correction.
   then did not stop within the 5-second shutdown budget. Forty local
   repetitions of that case alone passed. Find what holds the query thread
   after the fence before changing either side.
-- **Workflow definition routing (Server 2.10.0).** Native records the
-  definition fingerprint a start's worker cohort agrees on, as PHP does.
-  Published PHP 2.10.0 also routes a task for such a run by that
-  fingerprint. It prefers a live worker advertising the recorded definition.
-  Otherwise it offers the task only to a worker that advertises
-  `workflow_entry_arguments_validation`. It reports `definition_routing` in
-  describe, repair, debug and task queues, and stores each worker's protocol
-  version under `_worker_protocol_version` in its process metrics. Native
-  offers such a task to any worker and reports none of this.
+- **Workflow definition routing reports (Server 2.10.0).** Native routes a
+  pinned run's workflow task by its recorded definition, as PHP's
+  `WorkflowDefinitionRouting` does
+  ([checkpoint](rust-server-port-checkpoints/2026-10-22-definition-routing.md)).
+  Published PHP also reports that decision as `definition_routing` in
+  describe, repair, debug and task queues. Native does not report it yet.
 - **Start build-id pin.** Published PHP pins a new run to its start cohort's
   build id: a promoted build-id rollout, else the single build id among the
   live workers for the type, else among all the queue's live workers. Native
diff --git a/rust/src/runtime/definition_routing.rs b/rust/src/runtime/definition_routing.rs
new file mode 100644
index 000000000..9b926ca8f
--- /dev/null
+++ b/rust/src/runtime/definition_routing.rs
@@ -0,0 +1,204 @@
+//! Server 2.10.0's `WorkflowDefinitionRouting` and
+//! `WorkflowReplayEligibility`. A run whose start recorded the definition
+//! fingerprint its worker cohort agreed on (`workflow_definition_fingerprint_source`
+//! `worker`) is pinned to that definition: its workflow tasks go to a worker
+//! advertising the recorded fingerprint. A worker with another definition gets
+//! them only when no live peer advertises the recorded one, and only if it
+//! validates entry arguments (`workflow_entry_arguments_validation`), so it can
+//! safely refuse incompatible input on replay.
+//!
+//! Native refuses to start with `DW_V2_PIN_TO_RECORDED_FINGERPRINT` turned off,
+//! so pinning is always enabled here.
+use super::transaction::Transaction;
+use super::{
+    Result,
+    backend::Backend,
+    store::{Store, now},
+};
+use serde_json::Value;
+use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
+
+/// The worker capability that lets a changed definition replay a pinned run.
+pub(super) const ENTRY_VALIDATION_CAPABILITY: &str = "workflow_entry_arguments_validation";
+
+/// PHP's `stringValue`: a trimmed, non-empty string.
+fn trimmed(value: Option<&str>) -> Option<String> {
+    value.map(str::trim).filter(|value| !value.is_empty()).map(str::to_owned)
+}
+
+fn has_capability(capabilities: &Value, capability: &str) -> bool {
+    capabilities.as_array().is_some_and(|items| items.iter().any(|item| item == capability))
+}
+
+/// `context`: the run's recorded definition and where a decision applies.
+pub(super) struct DefinitionContext {
+    pub(super) worker_pinned: bool,
+    pub(super) recorded: Option<String>,
+    pub(super) namespace: String,
+    pub(super) workflow_type: String,
+    pub(super) task_queue: String,
+    pub(super) required_build_id: Option<String>,
+}
+
+impl<DB: Backend> Store<DB>
+where
+    for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>,
+    DB::Arguments: IntoArguments<DB>,
+    for<'q> &'q str: Encode<'q, DB> + Type<DB>,
+    for<'q> Option<&'q str>: Encode<'q, DB>,
+    for<'q> Option<DB::EncodedDocument>: Encode<'q, DB>,
+    for<'q> String: Encode<'q, DB> + Type<DB>,
+    for<'q> i64: Encode<'q, DB> + Type<DB>,
+    for<'r> i64: Decode<'r, DB>,
+    usize: ColumnIndex<DB::Row>,
+{
+    pub(super) async fn definition_context(
+        tx: &mut Transaction<'_, DB>,
+        run: &DB::Row,
+        task_queue: Option<&str>,
+        required_build_id: Option<String>,
+    ) -> Result<DefinitionContext> {
+        let started = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowStarted' ORDER BY sequence LIMIT 1")
+            .bind(DB::string(run, "id")?)
+            .fetch_optional(&mut **tx)
+            .await?
+            .map(|row| DB::document_row(&row, "payload"))
+            .transpose()?
+            .unwrap_or(Value::Null);
+        Ok(DefinitionContext {
+            worker_pinned: started["workflow_definition_fingerprint_source"] == "worker",
+            recorded: trimmed(started["workflow_definition_fingerprint"].as_str()),
+            namespace: DB::string(run, "namespace")?,
+            workflow_type: DB::string(run, "workflow_type")?,
+            task_queue: match task_queue {
+                Some(queue) => queue.to_owned(),
+                None => DB::string(run, "queue")?,
+            },
+            required_build_id: required_build_id.or(trimmed(DB::optional_string(run, "compatibility")?.as_deref())),
+        })
+    }
+
+    /// `forWorker`: whether this worker may take the run's workflow task, and
+    /// PHP's status for the decision.
+    pub(super) async fn definition_allows(
+        &self,
+        tx: &mut Transaction<'_, DB>,
+        run: &DB::Row,
+        worker: &DB::Row,
+        context: &DefinitionContext,
+    ) -> Result<(bool, &'static str)> {
+        if !context.worker_pinned {
+            return Ok((true, "not_worker_pinned"));
+        }
+        let Some(recorded) = &context.recorded else {
+            return Ok((false, "recorded_definition_missing"));
+        };
+        let advertised = DB::optional_document_row(worker, "workflow_definition_fingerprints")?
+            .and_then(|fingerprints| trimmed(fingerprints[&context.workflow_type].as_str()));
+        let Some(advertised) = advertised else {
+            return Ok((false, "worker_definition_missing"));
+        };
+        if &advertised == recorded {
+            return Ok((true, "matching_definition"));
+        }
+        for (_, fingerprint, _) in self.definition_peers(tx, run, context).await? {
+            if fingerprint.as_deref() == Some(recorded.as_str()) {
+                return Ok((false, "matching_definition_preferred"));
+            }
+        }
+        // Match the embedded resolver: prefer the recorded source when it is
+        // available, otherwise let the current definition validate replay.
+        let capabilities = DB::optional_document_row(worker, "capabilities")?.unwrap_or(Value::Null);
+        Ok(if has_capability(&capabilities, ENTRY_VALIDATION_CAPABILITY) {
+            (true, "replay_required")
+        } else {
+            (false, "entry_validation_unavailable")
+        })
+    }
+
+    /// `eligibleWorkers`: the live workers on the run's queue (and build) that
+    /// handle its type, take workflow tasks and can replay its history. Each
+    /// is its id, advertised fingerprint and capabilities.
+    pub(super) async fn definition_peers(
+        &self,
+        tx: &mut Transaction<'_, DB>,
+        run: &DB::Row,
+        context: &DefinitionContext,
+    ) -> Result<Vec<(String, Option<String>, Value)>> {
+        let rows = Self::query("SELECT worker_id,build_id,status,last_heartbeat_at,workflow_definition_fingerprints,supported_workflow_types,max_concurrent_workflow_tasks,capabilities,process_metrics FROM workflow_worker_registrations WHERE namespace=$1 AND task_queue=$2 AND status='active' ORDER BY worker_id")
+            .bind(&context.namespace)
+            .bind(&context.task_queue)
+            .fetch_all(&mut **tx)
+            .await?;
+        let mut peers = Vec::new();
+        for row in rows {
+            if context
+                .required_build_id
+                .as_deref()
+                .is_some_and(|build| DB::optional_string(&row, "build_id").ok().flatten().as_deref() != Some(build))
+            {
+                continue;
+            }
+            let types = DB::optional_document_row(&row, "supported_workflow_types")?.unwrap_or(Value::Null);
+            if !types.as_array().is_some_and(|types| types.iter().any(|kind| kind == context.workflow_type.as_str()))
+                || DB::optional_number(&row, "max_concurrent_workflow_tasks")?.is_some_and(|max| max < 1)
+            {
+                continue;
+            }
+            if !self.replay_eligible(tx, run, &row).await? {
+                continue;
+            }
+            let fingerprint = DB::optional_document_row(&row, "workflow_definition_fingerprints")?
+                .and_then(|fingerprints| trimmed(fingerprints[&context.workflow_type].as_str()));
+            let capabilities = DB::optional_document_row(&row, "capabilities")?.unwrap_or(Value::Null);
+            peers.push((DB::string(&row, "worker_id")?, fingerprint, capabilities));
+        }
+        Ok(peers)
+    }
+
+    /// `WorkflowReplayEligibility::allows` for a registered worker, with the
+    /// protocol version it last registered or heartbeated with.
+    async fn replay_eligible(&self, tx: &mut Transaction<'_, DB>, run: &DB::Row, worker: &DB::Row) -> Result<bool> {
+        // `WorkerPollFence::isFresh`: active, with a heartbeat inside the
+        // stale window.
+        let fresh = DB::string(worker, "status")? == "active"
+            && DB::optional_instant(worker, "last_heartbeat_at")?.is_some_and(|heartbeat| {
+                heartbeat >= now() - chrono::Duration::seconds(self.worker_stale_after_seconds)
+            });
+        if !fresh {
+            return Ok(false);
+        }
+        let capabilities = DB::optional_document_row(worker, "capabilities")?.unwrap_or(Value::Null);
+        let protocol = DB::optional_document_row(worker, "process_metrics")?
+            .and_then(|metrics| trimmed(metrics[super::worker_management::PROTOCOL_METRIC].as_str()));
+        if Self::cancellation_request(run)?.is_some()
+            && !super::cooperative::supports(&capabilities, protocol.as_deref().unwrap_or(""))
+        {
+            return Ok(false);
+        }
+        // `WorkflowMetadataCapabilityPolicy::canReplayRun`: memo history needs
+        // `memo_upserts`, typed search attribute history needs typed upserts.
+        let run_id = DB::string(run, "id")?;
+        if !has_capability(&capabilities, "memo_upserts")
+            && Self::scalar("SELECT COUNT(*) FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='MemoUpserted'")
+                .bind(&run_id)
+                .fetch_one(&mut **tx)
+                .await?
+                > 0
+        {
+            return Ok(false);
+        }
+        if !has_capability(&capabilities, super::search_attribute_upserts::CAPABILITY) {
+            let typed = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SearchAttributesUpserted' ORDER BY sequence LIMIT 1001")
+                .bind(&run_id)
+                .fetch_all(&mut **tx)
+                .await?;
+            for event in &typed {
+                if DB::document_row(event, "payload")?.get("attribute_types").is_some() {
+                    return Ok(false);
+                }
+            }
+        }
+        Ok(true)
+    }
+}
diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
index 7fd36531f..10fa16ce8 100644
--- a/rust/src/runtime/execution.rs
+++ b/rust/src/runtime/execution.rs
@@ -883,18 +883,22 @@ impl Runtime {
             page_size: i64,
         ) -> Result<Value>
     );
-    pub(crate) async fn register(&self, body: Value) -> Result<Value> {
+    #[cfg(test)]
+    pub(crate) async fn register_unversioned(&self, body: Value) -> Result<Value> {
+        self.register(body, None).await
+    }
+    pub(crate) async fn register(&self, body: Value, protocol: Option<&str>) -> Result<Value> {
         match &*self.storage {
-            Storage::Sqlite(store) => store.register(body, &self.queries).await,
-            Storage::Postgres(store) => store.register(body, &self.queries).await,
-            Storage::MySql(store) => store.register(body, &self.queries).await,
+            Storage::Sqlite(store) => store.register(body, &self.queries, protocol).await,
+            Storage::Postgres(store) => store.register(body, &self.queries, protocol).await,
+            Storage::MySql(store) => store.register(body, &self.queries, protocol).await,
         }
     }
-    pub(crate) async fn worker_heartbeat(&self, body: Value) -> Result<Value> {
+    pub(crate) async fn worker_heartbeat(&self, body: Value, protocol: Option<&str>) -> Result<Value> {
         match &*self.storage {
-            Storage::Sqlite(store) => store.worker_heartbeat(body, &self.queries).await,
-            Storage::Postgres(store) => store.worker_heartbeat(body, &self.queries).await,
-            Storage::MySql(store) => store.worker_heartbeat(body, &self.queries).await,
+            Storage::Sqlite(store) => store.worker_heartbeat(body, &self.queries, protocol).await,
+            Storage::Postgres(store) => store.worker_heartbeat(body, &self.queries, protocol).await,
+            Storage::MySql(store) => store.worker_heartbeat(body, &self.queries, protocol).await,
         }
     }
     pub(crate) async fn list_workers(
diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
index b6bb73eef..268db087f 100644
--- a/rust/src/runtime/http.rs
+++ b/rust/src/runtime/http.rs
@@ -1340,7 +1340,7 @@ async fn register(
     }
     Ok((
         StatusCode::CREATED,
-        Json(worker_response(runtime.register(body).await?)),
+        Json(worker_response(runtime.register(body, Some(request_protocol(&headers)?)).await?)),
     ))
 }
 
@@ -1351,9 +1351,10 @@ fn worker_response(mut body: Value) -> Value {
 }
 async fn worker_heartbeat(
     Extension(runtime): Extension<Runtime>,
+    headers: HeaderMap,
     Json(body): Json<Value>,
 ) -> Result<Json<Value>> {
-    runtime.worker_heartbeat(body).await.map(Json)
+    runtime.worker_heartbeat(body, Some(request_protocol(&headers)?)).await.map(Json)
 }
 fn metrics_config(runtime: &Runtime) -> super::operator_metrics::MetricsConfig {
     super::operator_metrics::MetricsConfig {
diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs
index 8723ac1ff..d6dd33a1e 100644
--- a/rust/src/runtime/mod.rs
+++ b/rust/src/runtime/mod.rs
@@ -21,6 +21,7 @@ mod children;
 mod command_context;
 mod command_requests;
 mod continuations;
+mod definition_routing;
 mod control_plane;
 mod cooperative;
 mod cross_kind_poll;
diff --git a/rust/src/runtime/query_codec_tests.rs b/rust/src/runtime/query_codec_tests.rs
index 4fc813a87..c91ed219f 100644
--- a/rust/src/runtime/query_codec_tests.rs
+++ b/rust/src/runtime/query_codec_tests.rs
@@ -33,7 +33,7 @@ async fn original(timeout_seconds: i64) -> Original {
         lease_timeout_seconds: timeout_seconds,
         ..Default::default()
     });
-    runtime.register(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php",
+    runtime.register_unversioned(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php",
         "supported_workflow_types":["echo"],"supported_activity_types":[],"capabilities":["query_tasks"],
         "workflow_command_contracts":{"echo":{"queries":["state"],"query_contracts":[{"name":"state","parameters":[]}]}}})).await.unwrap();
     let started = runtime
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index 5f1a66539..08df04f26 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -1125,6 +1125,7 @@ where
         &self,
         body: Value,
         config: &super::QueryTaskConfig,
+        protocol: Option<&str>,
     ) -> Result<Value> {
         let worker_id = text(&body, "worker_id")?;
         let queue = text(&body, "task_queue")?;
@@ -1142,7 +1143,8 @@ where
             }
         }
         let runtime = text(&body, "runtime")?;
-        let registration = super::worker_management::Registration::parse(&body)?;
+        let mut registration = super::worker_management::Registration::parse(&body)?;
+        registration.record_protocol(protocol);
         let supplied_contracts = body
             .get("workflow_command_contracts")
             .map(|contracts| {
@@ -1479,6 +1481,17 @@ where
             {
                 continue;
             }
+            // `matchesRecordedWorkflowDefinition`: a run pinned to its recorded
+            // definition goes to a worker that has it, while one is live.
+            if kind == "workflow" {
+                let required = DB::optional_string(&task, "compatibility")?
+                    .map(|build| build.trim().to_owned())
+                    .filter(|build| !build.is_empty());
+                let context = Self::definition_context(&mut tx, &run, Some(queue), required).await?;
+                if !self.definition_allows(&mut tx, &run, &worker, &context).await?.0 {
+                    continue;
+                }
+            }
             if kind == "workflow"
                 && let Some(update_id) = task_payload["workflow_update_id"].as_str()
             {
diff --git a/rust/src/runtime/transition_codec_tests.rs b/rust/src/runtime/transition_codec_tests.rs
index 90a5d1f8a..7ab514a78 100644
--- a/rust/src/runtime/transition_codec_tests.rs
+++ b/rust/src/runtime/transition_codec_tests.rs
@@ -21,7 +21,7 @@ async fn runtime(directory: &tempfile::TempDir) -> Runtime {
     .await
     .unwrap();
     runtime
-        .register(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php",
+        .register_unversioned(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php",
             "supported_workflow_types":["echo"],"supported_activity_types":[],
             "workflow_command_contracts":{"echo":{"signals":["payload"],"signal_contracts":[{"name":"payload",
                 "parameters":[{"name":"value","position":0,"type":"array","required":true,"variadic":false,
@@ -135,7 +135,7 @@ async fn fenced_deregistration_repair_does_not_wait_for_read_codec_capacity() {
     .await
     .unwrap();
     let registration = runtime
-        .register(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php",
+        .register_unversioned(json!({"worker_id":"codec-worker","task_queue":"test","runtime":"php",
             "supported_workflow_types":["echo"],"supported_activity_types":[]}))
         .await
         .unwrap();
diff --git a/rust/src/runtime/worker_management.rs b/rust/src/runtime/worker_management.rs
index a80d77aa7..32d5418a1 100644
--- a/rust/src/runtime/worker_management.rs
+++ b/rust/src/runtime/worker_management.rs
@@ -91,6 +91,20 @@ fn slot(value: &Value, capacity: i64) -> Option<i64> {
     Some(count.max(0).min(capacity.max(0)))
 }
 
+/// The key under which PHP records a worker's protocol version.
+pub(super) const PROTOCOL_METRIC: &str = "_worker_protocol_version";
+
+/// Process metrics with the request's protocol version added, as PHP's
+/// `[...$metrics, PROTOCOL_METRIC => requestVersion]`.
+pub(super) fn with_protocol(metrics: Option<Value>, protocol: Option<&str>) -> Value {
+    let mut metrics = match metrics {
+        Some(Value::Object(metrics)) => metrics,
+        _ => Map::new(),
+    };
+    metrics.insert(PROTOCOL_METRIC.into(), json!(protocol));
+    Value::Object(metrics)
+}
+
 /// Validates the process metrics a worker reports and keeps PHP's normalized
 /// subset. Absent means "not reported"; the outer Option is presence.
 fn process_metrics(body: &Value) -> Result<Option<Option<Value>>> {
@@ -209,6 +223,13 @@ pub(super) struct Registration {
 }
 
 impl Registration {
+    /// `WorkflowReplayEligibility::PROTOCOL_METRIC`: PHP keeps the protocol
+    /// version of each registration and heartbeat in the worker's process
+    /// metrics, after the metrics the worker reported.
+    pub(super) fn record_protocol(&mut self, protocol: Option<&str>) {
+        self.process_metrics = Some(with_protocol(self.process_metrics.take(), protocol));
+    }
+
     pub(super) fn parse(body: &Value) -> Result<Self> {
         let fingerprints = match body.get("workflow_definition_fingerprints") {
             None => None,
@@ -439,12 +460,13 @@ where
         &self,
         body: Value,
         config: &QueryTaskConfig,
+        protocol: Option<&str>,
     ) -> Result<Value> {
         let worker_id = super::text(&body, "worker_id")?;
         let metrics = process_metrics(&body)?;
         let interval = bounded(&body, "heartbeat_interval_seconds", 1, 3600)?;
         let mut tx = self.begin().await?;
-        let Some(worker) = Self::query("SELECT max_concurrent_workflow_tasks,max_concurrent_activity_tasks,max_concurrent_worker_sessions FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2")
+        let Some(worker) = Self::query("SELECT max_concurrent_workflow_tasks,max_concurrent_activity_tasks,max_concurrent_worker_sessions,process_metrics FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2")
             .bind(worker_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?
         else {
             return Err(RuntimeError::Protocol {
@@ -471,11 +493,15 @@ where
                 }
             }
         }
-        if let Some(metrics) = metrics {
-            Self::query("UPDATE workflow_worker_registrations SET process_metrics=$1 WHERE worker_id=$2 AND namespace=$3")
-                .bind(metrics.as_ref().map(DB::document)).bind(worker_id).bind(namespace)
-                .execute(&mut *tx).await?;
-        }
+        // PHP stores the reported metrics, else keeps the stored ones, and adds
+        // this heartbeat's protocol version either way.
+        let reported = match metrics {
+            Some(metrics) => metrics,
+            None => DB::optional_document_row(&worker, "process_metrics").ok().flatten(),
+        };
+        Self::query("UPDATE workflow_worker_registrations SET process_metrics=$1 WHERE worker_id=$2 AND namespace=$3")
+            .bind(DB::document(&with_protocol(reported, protocol))).bind(worker_id).bind(namespace)
+            .execute(&mut *tx).await?;
         if interval.is_some() {
             Self::set_count(&mut tx, &INTERVAL, interval, worker_id, namespace).await?;
         }
diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
index 746707a9a..6ac4387dc 100644
--- a/rust/tests/execution.rs
+++ b/rust/tests/execution.rs
@@ -97,6 +97,10 @@ mod signal_rejection;
 mod cooperative_waits;
 #[path = "execution/child_policies.rs"]
 mod child_policies;
+#[path = "execution/definition_routing.rs"]
+mod definition_routing;
+/// `WorkflowDefinitionRouting::ENTRY_VALIDATION_CAPABILITY`.
+const ENTRY_VALIDATION: &str = "workflow_entry_arguments_validation";
 #[path = "execution/parent_close_policies.rs"]
 mod parent_close_policies;
 #[path = "execution/history_pages.rs"]
diff --git a/rust/tests/execution/definition_routing.rs b/rust/tests/execution/definition_routing.rs
new file mode 100644
index 000000000..4697e2473
--- /dev/null
+++ b/rust/tests/execution/definition_routing.rs
@@ -0,0 +1,80 @@
+use super::*;
+
+// Server 2.10.0 `WorkflowDefinitionRouting`: a run whose start recorded its
+// worker cohort's definition fingerprint goes to a worker advertising that
+// definition while one is live. A worker with another definition gets it only
+// when none is, and only if it validates entry arguments.
+
+async fn worker(app: &Router, id: &str, fingerprint: Option<&str>, capabilities: Value) {
+    let mut body = json!({"worker_id":id,"task_queue":"test","runtime":"php","supported_workflow_types":["echo"],
+        "supported_activity_types":[],"capabilities":capabilities});
+    if let Some(fingerprint) = fingerprint {
+        body["workflow_definition_fingerprints"] = json!({"echo": fingerprint});
+    }
+    let (status, registered) = request(app, "POST", "/api/worker/register", body).await;
+    assert_eq!(status, StatusCode::CREATED, "{registered}");
+}
+
+async fn deregister(app: &Router, id: &str) {
+    let (status, body) = request(app, "DELETE", &format!("/api/workers/{id}"), Value::Null).await;
+    assert!(status.is_success(), "{status} {body}");
+}
+
+#[tokio::test]
+async fn a_pinned_run_goes_to_a_worker_with_its_recorded_definition() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    worker(&app, "original", Some("sha256:original"), json!([])).await;
+    let started = start(&app, "pinned").await;
+    let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await;
+    let recorded = history.as_array().unwrap().iter().find(|event| event["event_type"] == "WorkflowStarted").unwrap();
+    assert_eq!(recorded["payload"]["workflow_definition_fingerprint"], "sha256:original");
+    assert_eq!(recorded["payload"]["workflow_definition_fingerprint_source"], "worker");
+
+    // A changed definition waits while the recorded one is live, even if it
+    // could validate entry arguments; so does a worker without a fingerprint.
+    worker(&app, "changed", Some("sha256:changed"), json!([super::ENTRY_VALIDATION])).await;
+    worker(&app, "unfingerprinted", None, json!([super::ENTRY_VALIDATION])).await;
+    assert!(poll(&app, "changed", "workflow").await.is_null(), "matching_definition_preferred");
+    assert!(poll(&app, "unfingerprinted", "workflow").await.is_null(), "worker_definition_missing");
+    let task = poll(&app, "original", "workflow").await;
+    assert_eq!(task["run_id"], started["run_id"], "{task}");
+}
+
+#[tokio::test]
+async fn without_the_recorded_definition_only_an_entry_validating_worker_replays() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    worker(&app, "original", Some("sha256:original"), json!([])).await;
+    let started = start(&app, "replacement").await;
+    deregister(&app, "original").await;
+    worker(&app, "unvalidated", Some("sha256:changed"), json!([])).await;
+    assert!(poll(&app, "unvalidated", "workflow").await.is_null(), "entry_validation_unavailable");
+    worker(&app, "validating", Some("sha256:changed"), json!([super::ENTRY_VALIDATION])).await;
+    let task = poll(&app, "validating", "workflow").await;
+    assert_eq!(task["run_id"], started["run_id"], "replay_required: {task}");
+}
+
+#[tokio::test]
+async fn an_unpinned_run_goes_to_any_worker_and_workers_record_their_protocol() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    // No fingerprint when the run starts, so it is not worker-pinned.
+    worker(&app, "plain", None, json!([])).await;
+    let started = start(&app, "unpinned").await;
+    worker(&app, "fingerprinted", Some("sha256:any"), json!([])).await;
+    let task = poll(&app, "fingerprinted", "workflow").await;
+    assert_eq!(task["run_id"], started["run_id"], "not_worker_pinned: {task}");
+
+    // `WorkflowReplayEligibility::PROTOCOL_METRIC`: each registration and
+    // heartbeat records its protocol version with the process metrics.
+    let (status, described) = request(&app, "GET", "/api/workers/plain", Value::Null).await;
+    assert_eq!(status, StatusCode::OK, "{described}");
+    assert_eq!(described["process_metrics"], json!({"_worker_protocol_version": "1.19"}), "{described}");
+    let (status, beat) = request(&app, "POST", "/api/worker/heartbeat",
+        json!({"worker_id":"plain","process_metrics":{"host":"worker-host"}})).await;
+    assert_eq!(status, StatusCode::OK, "{beat}");
+    let (_, described) = request(&app, "GET", "/api/workers/plain", Value::Null).await;
+    assert_eq!(described["process_metrics"]["host"], "worker-host", "{described}");
+    assert_eq!(described["process_metrics"]["_worker_protocol_version"], "1.19", "{described}");
+}
diff --git a/rust/tests/execution/worker_listing.rs b/rust/tests/execution/worker_listing.rs
index 1abcbc1e9..034b47201 100644
--- a/rust/tests/execution/worker_listing.rs
+++ b/rust/tests/execution/worker_listing.rs
@@ -82,7 +82,7 @@ async fn workers_are_listed_with_their_registration_and_heartbeat_facts() {
     );
     assert_eq!(
         worker["process_metrics"],
-        json!({"process_uptime_seconds": 0, "process_id": 17})
+        json!({"process_uptime_seconds": 0, "process_id": 17, "_worker_protocol_version": "1.19"})
     );
     assert_eq!(worker["status"], "active");
     assert_eq!(worker["heartbeat_interval_seconds"], Value::Null);
@@ -94,7 +94,9 @@ async fn workers_are_listed_with_their_registration_and_heartbeat_facts() {
         json!({"workflow_available": null, "activity_available": null,
         "session_available": null, "workflow_capacity": 100, "activity_capacity": 100, "session_capacity": 10})
     );
-    assert_eq!(defaults["process_metrics"], Value::Null);
+    // Server 2.10.0 records each registration's protocol version with the
+    // process metrics (`WorkflowReplayEligibility::PROTOCOL_METRIC`).
+    assert_eq!(defaults["process_metrics"], json!({"_worker_protocol_version": "1.19"}));
     assert_eq!(ours(&app, "?task_queue=queue-a").await.len(), 1);
     assert_eq!(
         ours(&app, "?task_queue=0").await.len(),
@@ -162,7 +164,7 @@ async fn workers_are_listed_with_their_registration_and_heartbeat_facts() {
         json!({"listing": fingerprint})
     );
     assert_eq!(after["task_slots"]["workflow_available"], Value::Null);
-    assert_eq!(after["process_metrics"], Value::Null);
+    assert_eq!(after["process_metrics"], json!({"_worker_protocol_version": "1.19"}));
     assert_eq!(after["registered_at"], described["registered_at"]);
 
     // A heartbeat changes only what it reports.

From adb55763facbfbbd217e501f99f6cedcb25680ea Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Thu, 22 Oct 2026 14:22:26 +0000
Subject: [PATCH 2/6] Report definition routing in describe, repair, debug and
 task queues

Server 2.10.0 reports WorkflowDefinitionRouting's decision for a run's next
due ready workflow task as definition_routing in describe, repair and debug
(inspect), and samples a queue's ready runs in task queue describe
(forQueue). Native now does the same: the recorded context, PHP's status,
eligible worker counts by definition, up to ten example workers, blocked,
runnable_task and PHP's resolution text. A blocked route is a debug warning
(workflow_definition_routing_blocked, diagnostic_status needs_attention) and
marks the queue's repair as needing attention.
---
 .../2026-10-22-definition-routing.md          |  21 +++-
 docs/rust-server-port.md                      |   6 -
 rust/src/runtime/definition_routing.rs        | 107 +++++++++++++++++-
 rust/src/runtime/run_describe.rs              |   5 +
 rust/src/runtime/run_diagnostics.rs           |  12 ++
 rust/src/runtime/task_queues.rs               |  12 +-
 rust/src/runtime/task_repair.rs               |  25 ++++
 rust/tests/execution/definition_routing.rs    |  53 +++++++++
 8 files changed, 229 insertions(+), 12 deletions(-)

diff --git a/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md b/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
index 66116fb8b..4297e7380 100644
--- a/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
+++ b/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
@@ -37,8 +37,18 @@
     metrics, after the metrics it reported. A heartbeat without metrics
     keeps the stored ones. Worker listings therefore show the key, as PHP
     2.10.0's do; 2.5.14 had no such key.
-- **Remaining gap** (the port's gap list): the `definition_routing` reports in
-  describe, repair, debug and task queues.
+- **Reports** (`inspect` and `forQueue`).
+  - Describe, repair and debug report `definition_routing` for the run's
+    next due ready task:
+    - the context (`pinning_enabled`, `worker_pinned`,
+      `recorded_fingerprint`, type, queue and required build);
+    - PHP's status, the counts of eligible workers by definition, up to ten
+      example workers, `blocked` and `runnable_task`;
+    - PHP's resolution text.
+  - A blocked route adds debug's `workflow_definition_routing_blocked`
+    warning and `diagnostic_status: needs_attention`.
+  - Task queue describe samples up to ten runs with a due ready task. A
+    blocked sample sets `repair.needs_attention`.
 - **Tests** (`definition_routing.rs`).
   - A pinned run goes to the worker with its recorded definition. Workers
     with another definition, or without one, wait even when they validate
@@ -48,5 +58,8 @@
   - An unpinned run goes to any worker.
   - Registrations and heartbeats record the protocol version with the
     reported metrics.
-  - The worker listing test now expects that key.
-  - Full SQLite suite: 397 passed.
+  - Describe, debug, task queue and repair report a blocked route, then the
+    route an entry-validating replacement unblocks.
+  - The worker listing test now expects the protocol key.
+  - Full SQLite suite: 398 passed. Routing, worker, poll, child, cooperative,
+    debug, task queue, repair and describe tests: PostgreSQL 134, MySQL 134.
diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index 432f03082..ab3ab1fbd 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -182,12 +182,6 @@ contract fixture before its correction.
   then did not stop within the 5-second shutdown budget. Forty local
   repetitions of that case alone passed. Find what holds the query thread
   after the fence before changing either side.
-- **Workflow definition routing reports (Server 2.10.0).** Native routes a
-  pinned run's workflow task by its recorded definition, as PHP's
-  `WorkflowDefinitionRouting` does
-  ([checkpoint](rust-server-port-checkpoints/2026-10-22-definition-routing.md)).
-  Published PHP also reports that decision as `definition_routing` in
-  describe, repair, debug and task queues. Native does not report it yet.
 - **Start build-id pin.** Published PHP pins a new run to its start cohort's
   build id: a promoted build-id rollout, else the single build id among the
   live workers for the type, else among all the queue's live workers. Native
diff --git a/rust/src/runtime/definition_routing.rs b/rust/src/runtime/definition_routing.rs
index 9b926ca8f..2406206e6 100644
--- a/rust/src/runtime/definition_routing.rs
+++ b/rust/src/runtime/definition_routing.rs
@@ -15,7 +15,7 @@ use super::{
     backend::Backend,
     store::{Store, now},
 };
-use serde_json::Value;
+use serde_json::{Value, json};
 use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
 
 /// The worker capability that lets a changed definition replay a pinned run.
@@ -201,4 +201,109 @@ where
         }
         Ok(true)
     }
+
+    /// `inspect`: how the run's next ready workflow task would be routed,
+    /// with counts of the eligible workers by definition, for describe,
+    /// repair and debug.
+    pub(super) async fn definition_inspect(
+        &self,
+        tx: &mut Transaction<'_, DB>,
+        run: &DB::Row,
+        task_queue: Option<&str>,
+    ) -> Result<Value> {
+        let run_id = DB::string(run, "id")?;
+        let task = Self::query("SELECT queue,compatibility FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status='ready' AND (available_at IS NULL OR available_at<=$2) ORDER BY id")
+            .bind(&run_id)
+            .bind(DB::bind_time(now()))
+            .fetch_all(&mut **tx)
+            .await?
+            .into_iter()
+            .find(|task| task_queue.is_none_or(|queue| DB::optional_string(task, "queue").ok().flatten().as_deref() == Some(queue)));
+        let (queue, required) = match &task {
+            Some(task) => (
+                task_queue.map(str::to_owned).or(DB::optional_string(task, "queue")?),
+                trimmed(DB::optional_string(task, "compatibility")?.as_deref()),
+            ),
+            None => (task_queue.map(str::to_owned), None),
+        };
+        let context = Self::definition_context(tx, run, queue.as_deref(), required).await?;
+        let (mut matching, mut changed, mut missing, mut unsafe_replacement) = (0, 0, 0, 0);
+        let mut examples = Vec::new();
+        let peers = self.definition_peers(tx, run, &context).await?;
+        for (worker_id, fingerprint, capabilities) in &peers {
+            match fingerprint {
+                None => missing += 1,
+                Some(fingerprint) if Some(fingerprint) == context.recorded.as_ref() => matching += 1,
+                Some(_) => {
+                    changed += 1;
+                    if !has_capability(capabilities, ENTRY_VALIDATION_CAPABILITY) {
+                        unsafe_replacement += 1;
+                    }
+                }
+            }
+            if examples.len() < 10 {
+                examples.push(json!({"worker_id": worker_id, "fingerprint": fingerprint}));
+            }
+        }
+        let eligible = matching + changed + missing;
+        let status = match () {
+            _ if !context.worker_pinned => "not_worker_pinned",
+            _ if context.recorded.is_none() => "recorded_definition_missing",
+            _ if matching > 0 => "matching_definition_available",
+            _ if changed > unsafe_replacement => "replay_required",
+            _ if unsafe_replacement > 0 => "entry_validation_unavailable",
+            _ if missing > 0 => "worker_definition_missing",
+            _ => "no_eligible_worker",
+        };
+        let terminal = !matches!(DB::string(run, "status")?.as_str(), "pending" | "running" | "waiting");
+        let runnable = !terminal && task.is_some();
+        let blocked = runnable
+            && context.worker_pinned
+            && matches!(status, "recorded_definition_missing" | "worker_definition_missing" | "no_eligible_worker" | "entry_validation_unavailable");
+        let resolution = match () {
+            _ if blocked && status == "entry_validation_unavailable" => "Upgrade the replacement worker SDK to an implementation advertising workflow_entry_arguments_validation, or restore the recorded definition. The current replacement cannot safely refuse incompatible entry input.",
+            _ if blocked => "Register a fresh worker for the original namespace, task queue, workflow type, build and replay capabilities with its definition fingerprint. Do not rewrite the run history.",
+            _ if status == "replay_required" => "Current workers may replay the original history. Use versioned branches for command changes; incompatible replay still refuses.",
+            _ => "Recorded build, namespace, type and replay-capability guards remain in force.",
+        };
+        Ok(json!({
+            "pinning_enabled": true, "worker_pinned": context.worker_pinned,
+            "recorded_fingerprint": context.recorded, "workflow_type": context.workflow_type,
+            "task_queue": context.task_queue, "required_build_id": context.required_build_id,
+            "status": status, "blocked": blocked, "runnable_task": runnable,
+            "eligible_worker_count": eligible, "matching_worker_count": matching,
+            "changed_worker_count": changed, "missing_fingerprint_count": missing,
+            "entry_validation_unavailable_count": unsafe_replacement,
+            "workers": examples, "workers_truncated": eligible > examples.len(),
+            "resolution": resolution,
+        }))
+    }
+
+    /// `forQueue`: bounded examples of the queue's runs with a due ready
+    /// workflow task, never a repair or a payload read.
+    pub(super) async fn definition_for_queue(&self, tx: &mut Transaction<'_, DB>, task_queue: &str) -> Result<Value> {
+        let runs = Self::query("SELECT * FROM workflow_runs WHERE namespace=$1 AND queue=$2 AND id IN (SELECT workflow_run_id FROM workflow_tasks WHERE task_type='workflow' AND status='ready' AND queue=$3 AND (available_at IS NULL OR available_at<=$4)) ORDER BY id LIMIT 11")
+            .bind(self.namespace.as_ref())
+            .bind(task_queue)
+            .bind(task_queue)
+            .bind(DB::bind_time(now()))
+            .fetch_all(&mut **tx)
+            .await?;
+        let mut examples = Vec::new();
+        for run in runs.iter().take(10) {
+            if !matches!(DB::string(run, "status")?.as_str(), "pending" | "running" | "waiting") {
+                continue;
+            }
+            let mut example = json!({"workflow_id": DB::string(run, "workflow_instance_id")?, "run_id": DB::string(run, "id")?});
+            if let (Some(example), Value::Object(inspected)) =
+                (example.as_object_mut(), self.definition_inspect(tx, run, Some(task_queue)).await?)
+            {
+                example.extend(inspected);
+            }
+            examples.push(example);
+        }
+        let blocked = examples.iter().filter(|row| row["blocked"] == true).count();
+        Ok(json!({"sample_limit": 10, "has_more_ready_runs": runs.len() > 10,
+            "blocked_sample_count": blocked, "runs": examples}))
+    }
 }
diff --git a/rust/src/runtime/run_describe.rs b/rust/src/runtime/run_describe.rs
index 51c1fe383..55aebc489 100644
--- a/rust/src/runtime/run_describe.rs
+++ b/rust/src/runtime/run_describe.rs
@@ -268,6 +268,11 @@ where
         fields.insert("compatibility_status".into(), json!(if supported { "compatible" } else { "no_compatible_worker" }));
         fields.insert("compatibility_supported_in_fleet".into(), json!(supported));
         fields.insert("compatibility_fleet_reason".into(), json!(fleet_reason));
+        // `ServerWorkflowControlPlane::withDefinitionRouting`.
+        let mut tx = self.begin_read().await?;
+        let routing = self.definition_inspect(&mut tx, run, None).await?;
+        tx.commit().await?;
+        fields.insert("definition_routing".into(), routing);
         fields.insert("payload_previews".into(), json!({"input_omitted":input_omitted,"output_omitted":output_omitted,
             "max_encoded_bytes":max_payload_bytes()}));
         fields.insert("last_progress_at".into(), iso8601(DB::optional_instant(run, "last_progress_at")?));
diff --git a/rust/src/runtime/run_diagnostics.rs b/rust/src/runtime/run_diagnostics.rs
index 6f4c1656e..03c4bca78 100644
--- a/rust/src/runtime/run_diagnostics.rs
+++ b/rust/src/runtime/run_diagnostics.rs
@@ -835,6 +835,18 @@ where
             findings.extend(self.routing_findings(&run, &rows, at, &fleet, config).await?);
         }
         payload["cancellation_cascade"] = cascade;
+        // `WorkflowRunDiagnostics`: the definition route of the next ready
+        // task, and a warning when no eligible worker can take it.
+        let mut tx = self.begin_read().await?;
+        let routing = self.definition_inspect(&mut tx, &run, None).await?;
+        tx.commit().await?;
+        if routing["blocked"] == true {
+            findings.push(json!({"severity": "warning", "code": "workflow_definition_routing_blocked",
+                "message": "A ready workflow task has no eligible definition route.",
+                "routing_status": routing["status"], "expected_resolution": routing["resolution"]}));
+            payload["diagnostic_status"] = json!("needs_attention");
+        }
+        payload["definition_routing"] = routing;
         payload["findings"] = json!(findings);
         Ok(payload)
     }
diff --git a/rust/src/runtime/task_queues.rs b/rust/src/runtime/task_queues.rs
index d4359f316..68d9129bd 100644
--- a/rust/src/runtime/task_queues.rs
+++ b/rust/src/runtime/task_queues.rs
@@ -425,7 +425,17 @@ where
 
     /// `GET /api/task-queues/{name}`.
     pub(crate) async fn describe_task_queue(&self, queue: &str, config: &QueryTaskConfig) -> Result<Value> {
-        self.queue_view(queue, config, true).await
+        let mut payload = self.queue_view(queue, config, true).await?;
+        // `TaskQueueController::show`: sampled definition routes of the
+        // queue's ready runs; a blocked one needs attention.
+        let mut tx = self.begin_read().await?;
+        let routing = self.definition_for_queue(&mut tx, queue).await?;
+        tx.commit().await?;
+        if routing["blocked_sample_count"].as_u64().unwrap_or(0) > 0 {
+            payload["repair"]["needs_attention"] = json!(true);
+        }
+        payload["definition_routing"] = routing;
+        Ok(payload)
     }
 }
 
diff --git a/rust/src/runtime/task_repair.rs b/rust/src/runtime/task_repair.rs
index a4db66df1..d82373fab 100644
--- a/rust/src/runtime/task_repair.rs
+++ b/rust/src/runtime/task_repair.rs
@@ -90,6 +90,31 @@ where
         body: Value,
         context: Value,
         codec: Arc<Semaphore>,
+    ) -> Result<(StatusCode, Value)> {
+        let (status, mut response) = self.repair_run(workflow_id, requested_run_id, body, context, codec).await?;
+        // `ServerWorkflowControlPlane::withDefinitionRouting`: the route of the
+        // repaired run's next ready task.
+        let mut tx = self.begin_read().await?;
+        let run = match response["run_id"].as_str() {
+            Some(run_id) => Self::query("SELECT * FROM workflow_runs WHERE id=$1 AND namespace=$2")
+                .bind(run_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?,
+            None => Self::query("SELECT r.* FROM workflow_instances i JOIN workflow_runs r ON r.id=i.current_run_id WHERE i.id=$1 AND i.namespace=$2")
+                .bind(workflow_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?,
+        };
+        if let Some(run) = run {
+            response["definition_routing"] = self.definition_inspect(&mut tx, &run, None).await?;
+        }
+        tx.commit().await?;
+        Ok((status, response))
+    }
+
+    async fn repair_run(
+        &self,
+        workflow_id: &str,
+        requested_run_id: Option<&str>,
+        body: Value,
+        context: Value,
+        codec: Arc<Semaphore>,
     ) -> Result<(StatusCode, Value)> {
         if !body.is_object()
             || body.get("request_id").is_some_and(|value| {
diff --git a/rust/tests/execution/definition_routing.rs b/rust/tests/execution/definition_routing.rs
index 4697e2473..0e93a1fea 100644
--- a/rust/tests/execution/definition_routing.rs
+++ b/rust/tests/execution/definition_routing.rs
@@ -78,3 +78,56 @@ async fn an_unpinned_run_goes_to_any_worker_and_workers_record_their_protocol()
     assert_eq!(described["process_metrics"]["host"], "worker-host", "{described}");
     assert_eq!(described["process_metrics"]["_worker_protocol_version"], "1.19", "{described}");
 }
+
+// Describe, repair, debug and the task queue report the route of a pinned
+// run's next ready task as `definition_routing`. A blocked route is a debug
+// finding and makes the queue need attention.
+#[tokio::test]
+async fn reads_report_a_blocked_definition_route() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    worker(&app, "original", Some("sha256:original"), json!([])).await;
+    let started = start(&app, "reported").await;
+    deregister(&app, "original").await;
+    worker(&app, "unvalidated", Some("sha256:changed"), json!([])).await;
+
+    let (status, described) = request(&app, "GET", "/api/workflows/reported", Value::Null).await;
+    assert_eq!(status, StatusCode::OK, "{described}");
+    let routing = &described["definition_routing"];
+    assert_eq!(routing["status"], "entry_validation_unavailable", "{routing}");
+    assert_eq!(routing["blocked"], true);
+    assert_eq!(routing["runnable_task"], true);
+    assert_eq!(routing["worker_pinned"], true);
+    assert_eq!(routing["pinning_enabled"], true);
+    assert_eq!(routing["recorded_fingerprint"], "sha256:original");
+    assert_eq!(routing["task_queue"], "test");
+    assert_eq!((routing["eligible_worker_count"].clone(), routing["changed_worker_count"].clone(),
+        routing["entry_validation_unavailable_count"].clone()), (json!(1), json!(1), json!(1)));
+    assert_eq!(routing["workers"], json!([{"worker_id": "unvalidated", "fingerprint": "sha256:changed"}]));
+    assert!(routing["resolution"].as_str().unwrap().starts_with("Upgrade the replacement worker SDK"));
+
+    let (status, debug) = request(&app, "GET", "/api/workflows/reported/debug", Value::Null).await;
+    assert_eq!(status, StatusCode::OK, "{debug}");
+    assert_eq!(debug["definition_routing"], *routing);
+    assert_eq!(debug["diagnostic_status"], "needs_attention");
+    let finding = debug["findings"].as_array().unwrap().iter()
+        .find(|finding| finding["code"] == "workflow_definition_routing_blocked").unwrap_or_else(|| panic!("{debug}"));
+    assert_eq!(finding["routing_status"], "entry_validation_unavailable");
+
+    let (status, queue) = request(&app, "GET", "/api/task-queues/test", Value::Null).await;
+    assert_eq!(status, StatusCode::OK, "{queue}");
+    assert_eq!(queue["definition_routing"]["blocked_sample_count"], 1, "{queue}");
+    assert_eq!(queue["definition_routing"]["runs"][0]["run_id"], started["run_id"]);
+    assert_eq!(queue["definition_routing"]["runs"][0]["workflow_id"], "reported");
+    assert_eq!(queue["repair"]["needs_attention"], true, "{queue}");
+
+    let (status, repaired) = request(&app, "POST", "/api/workflows/reported/repair", json!({})).await;
+    assert!(status.is_success(), "{status} {repaired}");
+    assert_eq!(repaired["definition_routing"]["status"], "entry_validation_unavailable", "{repaired}");
+
+    // An entry-validating replacement unblocks the route.
+    worker(&app, "validating", Some("sha256:changed"), json!([super::ENTRY_VALIDATION])).await;
+    let (_, described) = request(&app, "GET", "/api/workflows/reported", Value::Null).await;
+    assert_eq!(described["definition_routing"]["status"], "replay_required", "{described}");
+    assert_eq!(described["definition_routing"]["blocked"], false);
+}

From 06dab3e72d611e9a8f536989591c41dcae7ef208 Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Thu, 22 Oct 2026 14:25:54 +0000
Subject: [PATCH 3/6] Run every fixture check before failing on a native
 difference

The SQLite, PostgreSQL and MySQL fixture steps stopped as soon as native's
record of the frozen corpus failed, which skipped every later check in the
step: the SDK direction, patch, marker, range and restart profiles. One
expected difference (native's Server 2.10.0 SignalRejected in
signals-before-worker) therefore hid everything after it.

The native record and the comparison now keep their status, the rest of
the step runs, and the step fails at its end if either failed. Nothing
about the expectations or the frozen artifacts changes.
---
 .github/workflows/server-parity.yml | 30 +++++++++++++++++++++++------
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml
index 1c6e559c1..7abad4686 100644
--- a/.github/workflows/server-parity.yml
+++ b/.github/workflows/server-parity.yml
@@ -554,6 +554,7 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json
+          native_record_status=0
           docker run --rm --user=1000:1000 --entrypoint node --network "$RESOURCE_SCOPE" \
             -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source \
             -e DW_PARITY_TOKEN=parity-test-token -e DW_PARITY_RUNNER_REVISION -e DW_PARITY_CLI_PHAR -e MYSQL_IMAGE "$image" scripts/conformance/server-parity.mjs record \
@@ -564,7 +565,8 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/patch-created-activity-completed.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
-            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json
+            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json || native_record_status=$?
+          compare_status=0
           docker run --rm --user=1000:1000 --entrypoint node --network=none \
             -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source "$image" \
             scripts/conformance/server-parity.mjs compare /evidence/php.json /evidence/rust.json /evidence/embedded.json \
@@ -574,7 +576,7 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/patch-created-activity-completed.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
-            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json > parity-evidence/comparison.json
+            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json > parity-evidence/comparison.json || compare_status=$?
           bash scripts/conformance/server-parity/published-sdk-patch-profile.sh mysql http://parity-php:8080 http://parity-native-a:8080
           bash scripts/conformance/server-parity/repeated-patch-profile.sh mysql http://parity-php:8080 http://parity-native-a:8080
           bash scripts/conformance/server-parity/legacy-marker-profile.sh mysql http://parity-php:8080 http://parity-native-a:8080
@@ -686,6 +688,10 @@ jobs:
             catalog php_ref > parity-evidence/catalog-after.json
           cmp parity-evidence/php-before.json parity-evidence/php-after.json
           cmp parity-evidence/catalog-before.json parity-evidence/catalog-after.json
+          # A native fixture difference fails the step only after every later
+          # check has run, so one difference cannot hide the rest.
+          test "$native_record_status" -eq 0
+          test "$compare_status" -eq 0
       - name: Compare separate configured role-token cohort
         run: bash scripts/conformance/server-parity/role-profile.sh mysql
       - name: Back up, take over and resume a separate published PHP upgrade source
@@ -1051,6 +1057,7 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json
+          native_record_status=0
           docker run --rm --user=1000:1000 --entrypoint node --network "$RESOURCE_SCOPE" \
             -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source \
             -e DW_PARITY_TOKEN=parity-test-token -e DW_PARITY_RUNNER_REVISION -e DW_PARITY_CLI_PHAR "$image" scripts/conformance/server-parity.mjs record \
@@ -1061,7 +1068,8 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/patch-created-activity-completed.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
-            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json
+            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json || native_record_status=$?
+          compare_status=0
           docker run --rm --user=1000:1000 --entrypoint node --network=none \
             -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source "$image" \
             scripts/conformance/server-parity.mjs compare /evidence/php.json /evidence/rust.json /evidence/embedded.json \
@@ -1071,7 +1079,7 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/patch-created-activity-completed.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
-            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json > parity-evidence/comparison.json
+            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json > parity-evidence/comparison.json || compare_status=$?
           bash scripts/conformance/server-parity/published-sdk-patch-profile.sh pgsql http://parity-php:8080 http://parity-native-a:8080
           bash scripts/conformance/server-parity/repeated-patch-profile.sh pgsql http://parity-php:8080 http://parity-native-a:8080
           bash scripts/conformance/server-parity/legacy-marker-profile.sh pgsql http://parity-php:8080 http://parity-native-a:8080
@@ -1183,6 +1191,10 @@ jobs:
             catalog php_ref > parity-evidence/catalog-after.json
           cmp parity-evidence/php-before.json parity-evidence/php-after.json
           cmp parity-evidence/catalog-before.json parity-evidence/catalog-after.json
+          # A native fixture difference fails the step only after every later
+          # check has run, so one difference cannot hide the rest.
+          test "$native_record_status" -eq 0
+          test "$compare_status" -eq 0
       - name: Compare separate configured role-token cohort
         env:
           POSTGRES_IMAGE: ${{ matrix.image }}
@@ -1542,6 +1554,7 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json
+          native_record_status=0
           docker run --rm --user=1000:1000 --entrypoint node --network "$RESOURCE_SCOPE" \
             -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source \
             -e DW_PARITY_TOKEN=parity-test-token -e DW_PARITY_RUNNER_REVISION -e DW_PARITY_CLI_PHAR "$image" scripts/conformance/server-parity.mjs record \
@@ -1552,7 +1565,8 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/patch-created-activity-completed.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
-            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json
+            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json || native_record_status=$?
+          compare_status=0
           docker run --rm --user=1000:1000 --entrypoint node \
             -v "$PWD:/source:ro" -v "$PWD/parity-evidence:/evidence" -w /source "$image" \
             scripts/conformance/server-parity.mjs compare /evidence/php.json /evidence/rust.json /evidence/embedded.json \
@@ -1562,7 +1576,7 @@ jobs:
             --include-fixture tests/Fixtures/ServerParityPending/patch-created-activity-completed.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-type.json \
             --include-fixture tests/Fixtures/ServerParityPending/workflow-waiting-history-message.json \
-            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json > parity-evidence/comparison.json
+            --include-fixture tests/Fixtures/ServerParityPending/workflow-task-transient-retry.json > parity-evidence/comparison.json || compare_status=$?
           bash scripts/conformance/server-parity/published-sdk-patch-profile.sh sqlite http://parity-server:8080 http://parity-rust:8080
           bash scripts/conformance/server-parity/repeated-patch-profile.sh sqlite http://parity-server:8080 http://parity-rust:8080
           bash scripts/conformance/server-parity/legacy-marker-profile.sh sqlite http://parity-server:8080 http://parity-rust:8080
@@ -1673,6 +1687,10 @@ jobs:
               php scripts/conformance/server-parity/cooperative-restart.php finish http://parity-recovered:8080 /evidence/cooperative-restart-receipt.json > /evidence/cooperative-restart-finish.json
               php scripts/conformance/server-parity/restart.php finish http://parity-recovered:8080 /evidence/restart-receipt.json > /evidence/restart-finish.json
             '
+          # A native fixture difference fails the step only after every later
+          # check has run, so one difference cannot hide the rest.
+          test "$native_record_status" -eq 0
+          test "$compare_status" -eq 0
       - name: Compare separate configured role-token cohort
         env:
           RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}

From c6f5ddf71566aa1762bf7271f9251fa9d3fff945 Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Thu, 22 Oct 2026 14:52:12 +0000
Subject: [PATCH 4/6] Register unit-test workers with the new protocol argument

---
 rust/src/runtime/memos.rs               | 2 +-
 rust/src/runtime/worker_incarnations.rs | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/rust/src/runtime/memos.rs b/rust/src/runtime/memos.rs
index 27cd0119a..0418763fe 100644
--- a/rust/src/runtime/memos.rs
+++ b/rust/src/runtime/memos.rs
@@ -454,7 +454,7 @@ mod tests {
                 .unwrap(),
         );
         store.register(json!({"worker_id":"memo-unit-worker", "task_queue":"test", "runtime":"php",
-            "supported_workflow_types":["echo"], "supported_activity_types":[], "capabilities":["memo_upserts"]}), &crate::runtime::QueryTaskConfig::default()).await.unwrap();
+            "supported_workflow_types":["echo"], "supported_activity_types":[], "capabilities":["memo_upserts"]}), &crate::runtime::QueryTaskConfig::default(), None).await.unwrap();
         let original = store
             .start(
                 json!({"workflow_id":"memo-unit", "workflow_type":"echo", "task_queue":"test",
diff --git a/rust/src/runtime/worker_incarnations.rs b/rust/src/runtime/worker_incarnations.rs
index 968bdc729..9bf30e711 100644
--- a/rust/src/runtime/worker_incarnations.rs
+++ b/rust/src/runtime/worker_incarnations.rs
@@ -142,7 +142,7 @@ mod tests {
             .unwrap();
         let definition = json!({"worker_id":"original-worker","task_queue":"original-queue",
             "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":[]});
-        runtime.register(definition.clone()).await.unwrap();
+        runtime.register_unversioned(definition.clone()).await.unwrap();
         let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", path.display()))
             .await
             .unwrap();
@@ -154,7 +154,7 @@ mod tests {
         let mut replacement = definition;
         replacement["task_queue"] = json!("replacement-queue");
         assert!(matches!(
-            runtime.register(replacement).await,
+            runtime.register_unversioned(replacement).await,
             Err(RuntimeError::Database(_))
         ));
         let after: Vec<(String,String,String,String,Option<i64>,Option<String>,String,String)> =

From b4156ac14c4625384ad636878dbfa3ea9812f6ae Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Thu, 22 Oct 2026 15:27:14 +0000
Subject: [PATCH 5/6] Leave the live routing view out of run-preservation
 checks

The worker-fencing contract compares whole run reads across a live
registration replacement. Native and Server 2.10.0 describe a run with
definition_routing, which reports how the next ready workflow task would
route now; releasing the lease makes that task runnable without changing
the run or its history. The preservation checks now compare the reads
without that block, and a unit case covers both a routing-only change and
a real run change.
---
 .../worker-deregistration-contract.mjs        | 20 ++++++++++++++-----
 ...ParityWorkerDeregistrationContractTest.mjs | 11 ++++++++++
 2 files changed, 26 insertions(+), 5 deletions(-)

diff --git a/scripts/conformance/server-parity/worker-deregistration-contract.mjs b/scripts/conformance/server-parity/worker-deregistration-contract.mjs
index 0c684209e..6cddf091b 100644
--- a/scripts/conformance/server-parity/worker-deregistration-contract.mjs
+++ b/scripts/conformance/server-parity/worker-deregistration-contract.mjs
@@ -2,6 +2,16 @@ import assert from 'node:assert/strict';
 import {checkSdkDatabasePressure} from './worker-pressure-contract.mjs';
 import {checkSdkPersistentPressure} from './worker-persistent-contract.mjs';
 
+// A run read without `definition_routing`: that block (Server 2.10.0 and
+// native) reports how the next ready workflow task would route now, so a
+// lease release or a registration change alters it without touching the
+// run or its history.
+const durable = run => {
+  if (!run?.execution?.definition_routing) return run;
+  const {definition_routing: _routing, ...execution} = run.execution;
+  return {...run, execution};
+};
+
 export function checkWorkerDeregistration(fixture, observation, workflowId) {
   const definition = fixture.worker_deregistration;
   const state = observation.worker_deregistration;
@@ -131,7 +141,7 @@ export function checkWorkerDeregistration(fixture, observation, workflowId) {
     assert.equal(initial.workflow_task_attempt, 1);
     assert.equal(task.workflow_task_attempt, initial.workflow_task_attempt+1);
     assert.deepStrictEqual(live.before_replacement.events, history.slice(0,2));
-    assert.deepStrictEqual(live.after_replacement, live.before_replacement, 'registration release preserves full original run/history');
+    assert.deepStrictEqual(durable(live.after_replacement), durable(live.before_replacement), 'registration release preserves full original run/history');
     for (const key of ['before_replacement','after_replacement','after_superseded','after_stale_before_poll','before_stale_after_poll','after_stale_after_poll']) {
       const run = live[key];
       assert.equal(run.workflow_id, peerId);
@@ -141,10 +151,10 @@ export function checkWorkerDeregistration(fixture, observation, workflowId) {
       assert.equal(run.status, state.before_unknown.status);
       assert.deepStrictEqual(run.typed_input, {type:'list',value:[fixture.typed_value]});
     }
-    assert.deepStrictEqual(live.after_superseded, live.after_replacement);
-    assert.deepStrictEqual(live.after_stale_before_poll, live.after_replacement);
-    assert.deepStrictEqual(live.after_stale_after_poll, live.before_stale_after_poll);
-    assert.deepStrictEqual(live.after_stale_after_poll, state.before_unknown);
+    assert.deepStrictEqual(durable(live.after_superseded), durable(live.after_replacement));
+    assert.deepStrictEqual(durable(live.after_stale_before_poll), durable(live.after_replacement));
+    assert.deepStrictEqual(durable(live.after_stale_after_poll), durable(live.before_stale_after_poll));
+    assert.deepStrictEqual(durable(live.after_stale_after_poll), durable(state.before_unknown));
     assert.equal(live.superseded.status, 409);
     assert.equal(live.superseded.response.reason, 'worker_registration_lost_authority');
     assert.equal(live.superseded.response.worker_id, workerId);
diff --git a/tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs b/tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
index 169888729..d775df3a8 100644
--- a/tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
+++ b/tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs
@@ -116,6 +116,17 @@ test('modeled live lease replacement requires unchanged original history and fen
   assert.deepStrictEqual(checkLive(liveModel()),checkWorkerDeregistration(liveFixture,
     {mode:'embedded',worker_deregistration:{applicable:false,reason:'embedded_has_no_http_worker_registration_lifecycle'}},id));
 });
+test('a live release may change only the current definition routing of the run read', () => {
+  const s=liveModel();
+  const routing=runnable=>({status:'not_worker_pinned',blocked:false,runnable_task:runnable});
+  s.live_replacement.before_replacement={...s.live_replacement.before_replacement,execution:{definition_routing:routing(false)}};
+  for (const key of ['after_replacement','after_superseded','after_stale_before_poll']) {
+    s.live_replacement[key]={...s.live_replacement[key],execution:{definition_routing:routing(true)}};
+  }
+  assert.doesNotThrow(()=>checkLive(s));
+  s.live_replacement.after_replacement.execution.status='completed';
+  assert.throws(()=>checkLive(s));
+});
 const liveMutations={
   'missing live replacement evidence':s=>{delete s.live_replacement;},
   'live token reused':s=>{s.live_replacement.initial_registration.registration_token=tokens[0];},

From 1950f385d15559df228c39ce8383f7f6443ec8d6 Mon Sep 17 00:00:00 2001
From: Durable Workflow <support@durable-workflow.com>
Date: Thu, 22 Oct 2026 16:33:56 +0000
Subject: [PATCH 6/6] Leave the live routing view out of the query preservation
 check

The SDK query contract compares the run read before and after a query. definition_routing reports how the next ready task would route now, so a worker leaving changes its counts without touching the run or its history.
---
 .../server-parity/sdk-query-direction-contract.mjs  | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/scripts/conformance/server-parity/sdk-query-direction-contract.mjs b/scripts/conformance/server-parity/sdk-query-direction-contract.mjs
index fbb8fabe1..731b6051e 100644
--- a/scripts/conformance/server-parity/sdk-query-direction-contract.mjs
+++ b/scripts/conformance/server-parity/sdk-query-direction-contract.mjs
@@ -42,11 +42,16 @@ function unchangedDurableSnapshot(before, after, message) {
   const hasAvailability = snapshot => Object.hasOwn(snapshot.execution?.actions ?? {}, 'can_query');
   assert.equal(hasAvailability(after), hasAvailability(before), 'computed query availability field stays present');
   const durable = snapshot => {
-    if (!hasAvailability(snapshot)) return snapshot;
-    assert.equal(typeof snapshot.execution.actions.can_query, 'boolean', 'actual computed query availability');
-    const actions = {...snapshot.execution.actions};
+    // definition_routing (Server 2.10.0 and native) reports how the next ready
+    // workflow task would route now, so worker presence changes it without
+    // touching the run or its history.
+    const {definition_routing: _routing, ...execution} = snapshot.execution ?? {};
+    const stable = snapshot.execution ? {...snapshot, execution} : snapshot;
+    if (!hasAvailability(stable)) return stable;
+    assert.equal(typeof stable.execution.actions.can_query, 'boolean', 'actual computed query availability');
+    const actions = {...stable.execution.actions};
     delete actions.can_query; // Only live worker availability; complete raw value stays recorded.
-    return {...snapshot, execution: {...snapshot.execution, actions}};
+    return {...stable, execution: {...stable.execution, actions}};
   };
   unchanged(durable(before), durable(after), message);
 }