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 }}
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..4297e7380
--- /dev/null
+++ b/docs/rust-server-port-checkpoints/2026-10-22-definition-routing.md
@@ -0,0 +1,65 @@
+# 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.
+- **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
+    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.
+  - 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 bd7b02079..ab3ab1fbd 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -182,15 +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 (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.
 - **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..2406206e6
--- /dev/null
+++ b/rust/src/runtime/definition_routing.rs
@@ -0,0 +1,309 @@
+//! 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, json};
+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)
+    }
+
+    /// `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/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/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/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/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/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/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/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_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)> =
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..0e93a1fea
--- /dev/null
+++ b/rust/tests/execution/definition_routing.rs
@@ -0,0 +1,133 @@
+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}");
+}
+
+// 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);
+}
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.
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);
 }
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];},