diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index a82e4a1dc..c4ffc514a 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -82,14 +82,15 @@ Updated October 24, 2026.
     ([#737](https://github.com/durable-workflow/server/issues/737)).
 
   No expectation is edited to make native pass.
-- **Parity surface.** Native serves 148 of the 175 API route methods in
+- **Parity surface.** Native answers 156 of the 175 API route methods in
   published 2.11.1's route table, with message streams, durable workflow
   stream writes, update validation, activity cancellation acknowledgement,
   worker sessions, workflow-task cancellation scopes, caller-indexed Nexus
   operation reads, bridge adapter webhooks, and embedded v2 and Waterline v1
-  workflow import. An execution test sends each one through the native
-  router, and the ones still answering `501 rust_capability_not_implemented`
-  must match
+  workflow import. Eight of those are the prepared local-activity routes,
+  which it answers with PHP's refusal for a bridge without the prepared role.
+  An execution test sends each one through the native router, and the ones
+  still answering `501 rust_capability_not_implemented` must match
   [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt).
   Unlike published PHP, native's embedded v2 import refuses a bundle whose
   rows reference rows outside it, links only runs of one namespace, and reads
@@ -139,8 +140,20 @@ any other unknown path or method answers PHP's own 404 or 405
   `repair-pass` fixture; a recovery native does not perform yet is
   reported as its candidate's failure, with 207.
 - **Worker protocol features.**
-  - Local activities (prepare, checkpoint, recover, heartbeat, control,
-    outcome and cancellation acknowledgement).
+  - Prepared local activities (prepare, checkpoint, group checkpoint,
+    recover, heartbeat, control, outcome and cancellation acknowledgement)
+    are answered as a PHP server whose bridge lacks the prepared role answers
+    them. Native advertises `prepared_local_activities: false`, no groups and
+    no policies, so SDKs record local activities with `record_local_activity`.
+    A worker that calls one anyway gets PHP's 409
+    `prepared_local_activity[_groups]_not_supported`, listing
+    `request_protocol` below 1.20 and `installed_runtime_prepared_local_activity`
+    (plus `…_groups` for the group checkpoint), before any task is read.
+    No recorded fixture: published 2.11.1 has the role, so its answers can't
+    be recorded. `prepared_local_routes_are_refused_without_the_prepared_role`
+    pins the answers to `PreparedLocalActivityController::refused`. Serving
+    them is planned in
+    [#325 comment 6423707058](https://github.com/durable-workflow/server/issues/325#issuecomment-6423707058).
   - Cancellation scope checkpoints. (Open, prepare and deliver are served,
     with the `cancellation-scopes` fixture.)
   - (Worker sessions are served: create, heartbeat, close, list and detail,
diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
index 6cd8306a9..179490386 100644
--- a/rust/src/runtime/backend.rs
+++ b/rust/src/runtime/backend.rs
@@ -19,7 +19,9 @@ pub(super) trait Backend: Database {
     // deferred read transaction already keeps one WAL snapshot.
     const READ_SNAPSHOT_BEGIN: &str = "BEGIN";
     const ACTIVITY_METADATA_HISTORY: &str = "SELECT sequence,event_type,CASE WHEN projection_bytes<=8388608 THEN payload ELSE NULL END AS payload,workflow_task_id,recorded_at,projection_count FROM (SELECT sequence,event_type,payload,workflow_task_id,recorded_at,COUNT(*) OVER () AS projection_count,SUM(octet_length(CAST(payload AS TEXT))) OVER (ORDER BY sequence) AS projection_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') ORDER BY sequence LIMIT 10001) AS page ORDER BY sequence";
-    const OPEN_WAIT_ORDER: &str = "WITH clocks AS (SELECT id,wait_id,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open') SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id LIMIT 100 OFFSET $2";
+    // Waits whose stored position differs from their chronological ordinal
+    // over every retained wait of the run, as Workflow #941's projector stores.
+    const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1), ordinals AS (SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE position IS DISTINCT FROM ordinal ORDER BY ordinal LIMIT 100";
     const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS TEXT) AS payload_json FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16";
     const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND payload->>'failure_id'=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1";
     const OBSERVATION_ATTRIBUTES: &str = "SELECT \"key\" AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND \"key\" IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY \"key\" LIMIT 20";
@@ -171,20 +173,23 @@ impl Backend for Sqlite {
     // whole UTC seconds with SQLite's clock functions and round integer
     // microseconds explicitly; SQLite's fractional-date rounding differs at ties.
     // Preserve Carbon's signed U.u decimal key for pre-epoch dates as well.
-    const OPEN_WAIT_ORDER: &str = r#"
+    const WAIT_POSITION_MOVES: &str = r#"
 WITH utc_waits AS (
-    SELECT id,wait_id,opened_at,
+    SELECT id,wait_id,position,opened_at,
         replace(replace(replace(opened_at,'T',' '),'Z',''),'+00:00','') AS utc_opened
-    FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open'
+    FROM workflow_run_waits WHERE workflow_run_id=$1
 ), clocks AS (
     SELECT *,unixepoch(substr(utc_opened,1,19))*1000000
         +(CASE WHEN unixepoch(substr(utc_opened,1,19))<0 THEN -1 ELSE 1 END)
         *CAST(CASE WHEN substr(utc_opened,20,1)='.'
             THEN substr(substr(utc_opened,21)||'000000',1,6) ELSE '000000' END AS INTEGER) AS micros
     FROM utc_waits
+), ordinals AS (
+    SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,
+        CASE WHEN micros<0 THEN (micros-500)/1000 ELSE (micros+500)/1000 END,wait_id)-1 AS ordinal
+    FROM clocks
 )
-SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,
-    CASE WHEN micros<0 THEN (micros-500)/1000 ELSE (micros+500)/1000 END,wait_id LIMIT 100 OFFSET $2
+SELECT id,ordinal FROM ordinals WHERE position IS NOT ordinal ORDER BY ordinal LIMIT 100
 "#;
     const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND json_extract(payload,'$.failure_id') COLLATE BINARY=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1";
     const ACTIVITY_RESULT_EVENT_SQL: &str = "SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') AND (json_extract(payload,'$.activity_execution_id') COLLATE BINARY=$2 OR json_extract(payload,'$.activity.id') COLLATE BINARY=$3) ORDER BY sequence DESC LIMIT 1";
@@ -479,7 +484,7 @@ impl Backend for MySql {
     const READ_SNAPSHOT_BEGIN: &str = "START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY";
     const ACTIVITY_METADATA_HISTORY: &str = "SELECT h.sequence,h.event_type,CASE WHEN page.projection_bytes<=8388608 THEN h.payload ELSE NULL END AS payload,h.workflow_task_id,h.recorded_at,page.projection_count FROM (SELECT id,sequence,COUNT(*) OVER () AS projection_count,SUM(payload_bytes) OVER (ORDER BY sequence) AS projection_bytes FROM (SELECT id,sequence,octet_length(CAST(payload AS CHAR CHARACTER SET utf8mb4)) AS payload_bytes FROM workflow_history_events FORCE INDEX (workflow_history_events_workflow_run_id_sequence_unique) WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled')) AS sizes ORDER BY sequence LIMIT 10001) AS page JOIN workflow_history_events h ON h.id=page.id ORDER BY page.sequence";
     const PHP_CONNECTION: &str = "mysql";
-    const OPEN_WAIT_ORDER: &str = "WITH clocks AS (SELECT id,wait_id,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=? AND status='open') SELECT id FROM clocks ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id LIMIT 100 OFFSET ?";
+    const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,TIMESTAMPDIFF(MICROSECOND,'1970-01-01 00:00:00',opened_at) AS micros FROM workflow_run_waits WHERE workflow_run_id=?), ordinals AS (SELECT id,position,CAST(ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN micros<0 THEN 2*FLOOR(micros/1000000)*1000000-micros ELSE micros END)/1000),wait_id) AS SIGNED)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE NOT (position <=> ordinal) ORDER BY ordinal LIMIT 100";
     // The run-and-sequence index keeps MySQL from sorting whole rows: a
     // filesort of one large payload exceeds the default sort buffer and
     // fails the transaction ("Out of sort memory").
diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
index ad36fc574..82bf465cb 100644
--- a/rust/src/runtime/http.rs
+++ b/rust/src/runtime/http.rs
@@ -281,6 +281,19 @@ pub fn router(runtime: Runtime) -> Router {
             "/api/worker/workflow-tasks/{task_id}/cancellation-scopes/deliver",
             post(deliver_cancellation_scope),
         )
+        // Native lacks the prepared local-activity role: refused as PHP
+        // refuses them without it (prepared_local.rs).
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/prepare", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/recover", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/checkpoint", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/checkpoint-group", post(prepared_local_group_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/control", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/heartbeat", post(prepared_local_refused))
+        .route("/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/outcome", post(prepared_local_refused))
+        .route(
+            "/api/worker/workflow-tasks/{task_id}/local-activities/{attempt_id}/acknowledge-cancellation",
+            post(prepared_local_refused),
+        )
         .route(
             "/api/worker/workflow-tasks/{task_id}/complete",
             post(complete_workflow),
@@ -817,6 +830,8 @@ fn worker_route(rest: &str) -> Option<Allowed> {
             | ["update-validation-tasks", _, "approve" | "reject"]
             | ["workflow-tasks", _, "complete" | "fail" | "history" | "heartbeat" | "deliver-cancellation"]
             | ["workflow-tasks", _, "cancellation-scopes", "open" | "prepare" | "deliver"]
+            | ["workflow-tasks", _, "local-activities", "prepare" | "recover" | "checkpoint" | "checkpoint-group"]
+            | ["workflow-tasks", _, "local-activities", _, "control" | "heartbeat" | "outcome" | "acknowledge-cancellation"]
             | ["activity-tasks", _, "complete" | "fail" | "heartbeat" | "status" | "renew-lease" | "acknowledge-cancellation"]
     )
     .then_some(Allowed::Any)
@@ -2332,6 +2347,22 @@ async fn deliver_cancellation_scope(
     Ok((status, Json(body)))
 }
 
+/// POST …/local-activities/*: refused before any task is read, as a PHP
+/// server whose bridge lacks the prepared role refuses them.
+async fn prepared_local_refused(headers: HeaderMap, body: axum::body::Bytes) -> Result<(StatusCode, Json<Value>)> {
+    prepared_local(&headers, &body, super::prepared_local::Capability::Prepared)
+}
+
+async fn prepared_local_group_refused(headers: HeaderMap, body: axum::body::Bytes) -> Result<(StatusCode, Json<Value>)> {
+    prepared_local(&headers, &body, super::prepared_local::Capability::Groups)
+}
+
+fn prepared_local(headers: &HeaderMap, body: &[u8], capability: super::prepared_local::Capability) -> Result<(StatusCode, Json<Value>)> {
+    let protocol = headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok());
+    let (status, response) = super::prepared_local::refusal(&request_json(body)?, protocol, capability);
+    Ok((status, Json(response)))
+}
+
 async fn acknowledge_activity_cancellation(
     Extension(runtime): Extension<Runtime>,
     Path(task_id): Path<String>,
diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs
index 9f05a7dcf..7d8dbdfe3 100644
--- a/rust/src/runtime/mod.rs
+++ b/rust/src/runtime/mod.rs
@@ -66,6 +66,7 @@ mod history_presentation;
 mod http;
 mod laravel_validation;
 mod local_activities;
+mod prepared_local;
 mod memo_json;
 mod memo_limits;
 mod history_meter;
diff --git a/rust/src/runtime/prepared_local.rs b/rust/src/runtime/prepared_local.rs
new file mode 100644
index 000000000..5b4f04eec
--- /dev/null
+++ b/rust/src/runtime/prepared_local.rs
@@ -0,0 +1,115 @@
+//! The prepared local-activity worker routes, as a published PHP server
+//! answers them when its installed Workflow bridge lacks the prepared role
+//! (`PreparedLocalActivityController::unavailable` and `refused`, and
+//! `WorkerController::checkpointLocalActivityPrefix` and
+//! `checkpointLocalActivityGroup`).
+//!
+//! Native advertises `prepared_local_activities: false`, no groups and no
+//! cancellation policies, so SDKs record local activities with
+//! `record_local_activity` instead. A worker that calls one of these routes
+//! anyway is refused before any task is read, with PHP's 409 body:
+//! - `request_protocol` when its protocol version is below 1.20;
+//! - `installed_runtime_prepared_local_activity`, always;
+//! - for `checkpoint-group`, also `installed_runtime_prepared_local_activity_groups`.
+//!
+//! A version header native does not support at all is refused earlier, by the
+//! worker protocol gate, with 400 as in PHP. PHP's own server-protocol reason
+//! never applies: native serves cooperative cancellation.
+//!
+//! PHP reports `(string) $request->input('lease_owner', '')` as `worker_id`. An
+//! array or object there is an array-to-string conversion, which Laravel turns
+//! into a 500. Native reports an empty `worker_id` instead.
+use super::laravel_rules;
+use axum::http::StatusCode;
+use serde_json::{Value, json};
+
+/// `PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION`.
+const MINIMUM_PROTOCOL_VERSION: &str = "1.20";
+
+/// Which prepared capability the refused route needs.
+#[derive(Clone, Copy)]
+pub(crate) enum Capability {
+    /// Prepare, recover, control, heartbeat, outcome, acknowledge-cancellation
+    /// and the local prefix checkpoint.
+    Prepared,
+    /// The group checkpoint.
+    Groups,
+}
+
+/// PHP's 409 for a route whose prepared role the installed runtime lacks.
+pub(crate) fn refusal(body: &Value, protocol: Option<&str>, capability: Capability) -> (StatusCode, Value) {
+    let mut unavailable = Vec::new();
+    if !super::worker_admission::at_least(protocol, MINIMUM_PROTOCOL_VERSION) {
+        unavailable.push("request_protocol");
+    }
+    unavailable.push("installed_runtime_prepared_local_activity");
+    let (reason, required) = match capability {
+        Capability::Prepared => ("prepared_local_activity_not_supported", "prepared_local_activities"),
+        Capability::Groups => {
+            unavailable.push("installed_runtime_prepared_local_activity_groups");
+            ("prepared_local_activity_groups_not_supported", "prepared_local_activity_groups")
+        }
+    };
+    let body = laravel_rules::normalize(body);
+    let worker_id = match &body["lease_owner"] {
+        Value::String(text) => text.clone(),
+        Value::Number(number) => php_number(number),
+        Value::Bool(true) => "1".into(),
+        _ => String::new(),
+    };
+    let response = json!({
+        "reason": reason,
+        "required_capability": required,
+        "minimum_protocol_version": MINIMUM_PROTOCOL_VERSION,
+        "worker_id": worker_id,
+        "unavailable": unavailable,
+        "remediation": "Use a prepared-local backend and a claim issued to a capable protocol 1.20 worker.",
+    });
+    (StatusCode::CONFLICT, super::cancellation_scopes::enveloped(response))
+}
+
+/// PHP's `(string)` of a JSON number.
+fn php_number(number: &serde_json::Number) -> String {
+    match number.as_i64() {
+        Some(whole) => whole.to_string(),
+        None => {
+            let float = number.as_f64().unwrap_or_default();
+            if float.fract() == 0.0 && float.abs() < 1e15 { format!("{float:.0}") } else { float.to_string() }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_current_worker_is_refused_for_the_missing_runtime_role_only() {
+        let (status, body) = refusal(&json!({"lease_owner": " worker-1 "}), Some("1.20"), Capability::Prepared);
+        assert_eq!(status, StatusCode::CONFLICT);
+        assert_eq!(body["reason"], "prepared_local_activity_not_supported");
+        assert_eq!(body["required_capability"], "prepared_local_activities");
+        assert_eq!(body["worker_id"], "worker-1");
+        assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity"]));
+    }
+
+    #[test]
+    fn an_older_worker_is_also_refused_for_its_protocol() {
+        let (_, body) = refusal(&json!({}), Some("1.19"), Capability::Groups);
+        assert_eq!(body["reason"], "prepared_local_activity_groups_not_supported");
+        assert_eq!(body["required_capability"], "prepared_local_activity_groups");
+        assert_eq!(body["worker_id"], "");
+        assert_eq!(
+            body["unavailable"],
+            json!(["request_protocol", "installed_runtime_prepared_local_activity", "installed_runtime_prepared_local_activity_groups"])
+        );
+    }
+
+    #[test]
+    fn worker_id_is_php_string_cast_of_lease_owner() {
+        for (owner, expected) in [(json!(7), "7"), (json!(true), "1"), (json!(false), ""), (json!(null), ""), (json!(["x"]), "")] {
+            let (_, body) = refusal(&json!({"lease_owner": owner}), Some("1.20"), Capability::Prepared);
+            assert_eq!(body["worker_id"], expected, "{owner}");
+        }
+    }
+}
diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs
index 21bc4c851..a0bd59035 100644
--- a/rust/src/runtime/transaction.rs
+++ b/rust/src/runtime/transaction.rs
@@ -96,7 +96,7 @@ where
             return Err(refusal);
         }
         for run_id in std::mem::take(&mut self.summaries) {
-            Store::<DB>::order_open_waits(&mut self, &run_id).await?;
+            Store::<DB>::order_waits(&mut self, &run_id).await?;
             Store::<DB>::project_run_summary(&mut self, &run_id).await?;
         }
         self.inner.commit().await?;
diff --git a/rust/src/runtime/wait_projection.rs b/rust/src/runtime/wait_projection.rs
index 15077a2c0..a41e772d3 100644
--- a/rust/src/runtime/wait_projection.rs
+++ b/rust/src/runtime/wait_projection.rs
@@ -18,32 +18,41 @@ where
     for<'r> i64: Decode<'r, DB>,
     usize: ColumnIndex<DB::Row>,
 {
-    pub(super) async fn order_open_waits(tx: &mut Transaction<'_, DB>, run: &str) -> Result<()> {
-        let mut position = 0_i64;
+    /// Store each wait's chronological ordinal over all of the run's retained
+    /// waits, as Workflow #941's projector does, writing only waits that moved.
+    /// Resolving a wait moves nothing, so most transitions write no positions.
+    pub(super) async fn order_waits(tx: &mut Transaction<'_, DB>, run: &str) -> Result<()> {
+        let mut moved = std::collections::HashSet::new();
         loop {
-            // Match the published projector's open-wait order without
-            // rebuilding missing rows or hydrating the complete collection.
-            let rows = Self::query(DB::OPEN_WAIT_ORDER)
+            // The database ranks the run's waits; only rows whose stored
+            // position differs come back, at most a page at a time.
+            let rows = Self::query(DB::WAIT_POSITION_MOVES)
                 .bind(run)
-                .bind(position)
                 .fetch_all(&mut **tx)
                 .await?;
             if rows.is_empty() {
-                break;
+                return Ok(());
             }
             for row in rows {
+                let id = DB::string(&row, "id")?;
+                // A wait already written in this pass that still differs
+                // means the write did not apply; stop instead of looping.
+                if !moved.insert(id.clone()) {
+                    return Err(super::refuse(
+                        axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+                        "wait_position_not_applied",
+                    ));
+                }
                 Self::query(
                     "UPDATE workflow_run_waits SET position=$1 WHERE id=$2 AND workflow_run_id=$3",
                 )
-                .bind(position)
-                .bind(DB::string(&row, "id")?)
+                .bind(DB::number(&row, "ordinal")?)
+                .bind(id)
                 .bind(run)
                 .execute(&mut **tx)
                 .await?;
-                position += 1;
             }
         }
-        Ok(())
     }
 
     pub(super) async fn project_update_wait_event(
diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
index de8866abf..26241677f 100644
--- a/rust/tests/execution.rs
+++ b/rust/tests/execution.rs
@@ -246,6 +246,8 @@ mod run_observation_signals;
 mod run_observation_summaries;
 #[path = "execution/run_observation_timers.rs"]
 mod run_observation_timers;
+#[path = "execution/wait_positions.rs"]
+mod wait_positions;
 #[path = "execution/run_observation_updates.rs"]
 mod run_observation_updates;
 
diff --git a/rust/tests/execution/local_activities.rs b/rust/tests/execution/local_activities.rs
index 41490976b..36aecc763 100644
--- a/rust/tests/execution/local_activities.rs
+++ b/rust/tests/execution/local_activities.rs
@@ -174,3 +174,44 @@ async fn a_recorded_local_activity_writes_its_attempts_history_and_failure() {
     assert_eq!(status, StatusCode::OK, "{described}");
     assert_eq!((&described["status"], &described["wait_kind"]), (&json!("waiting"), &json!("signal")), "{described}");
 }
+
+/// The prepared local-activity routes answer as a PHP server whose bridge lacks
+/// the prepared role: refused with 409 before any task is read, whether or not
+/// the task exists, in the worker envelope.
+#[tokio::test]
+async fn prepared_local_routes_are_refused_without_the_prepared_role() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    let base = "/api/worker/workflow-tasks/no-such-task/local-activities";
+    let prepared = ["prepare", "recover", "checkpoint", "attempt-1/control", "attempt-1/heartbeat", "attempt-1/outcome",
+        "attempt-1/acknowledge-cancellation"];
+    for route in prepared {
+        let (status, body) = request_protocol(&app, "POST", &format!("{base}/{route}"), json!({"lease_owner": " worker-1 "}), "1.20").await;
+        assert_eq!(status, StatusCode::CONFLICT, "{route}: {body}");
+        assert_eq!(body["reason"], "prepared_local_activity_not_supported", "{route}: {body}");
+        assert_eq!(body["required_capability"], "prepared_local_activities", "{route}");
+        assert_eq!(body["minimum_protocol_version"], "1.20", "{route}");
+        assert_eq!(body["worker_id"], "worker-1", "{route}");
+        assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity"]), "{route}");
+        assert_eq!(body["remediation"], "Use a prepared-local backend and a claim issued to a capable protocol 1.20 worker.");
+        assert_eq!(body["protocol_version"], "1.20", "{route}");
+        assert_eq!(body["server_capabilities"]["prepared_local_activities"], false, "{route}");
+
+        // An older worker is refused for its protocol too.
+        let (status, body) = request_protocol(&app, "POST", &format!("{base}/{route}"), Value::Null, "1.19").await;
+        assert_eq!(status, StatusCode::CONFLICT, "{route}: {body}");
+        assert_eq!(body["worker_id"], "", "{route}");
+        assert_eq!(body["unavailable"], json!(["request_protocol", "installed_runtime_prepared_local_activity"]), "{route}");
+    }
+
+    let (status, body) = request_protocol(&app, "POST", &format!("{base}/checkpoint-group"), json!({"lease_owner": 7}), "1.20").await;
+    assert_eq!(status, StatusCode::CONFLICT, "{body}");
+    assert_eq!(body["reason"], "prepared_local_activity_groups_not_supported", "{body}");
+    assert_eq!(body["required_capability"], "prepared_local_activity_groups");
+    assert_eq!(body["worker_id"], "7");
+    assert_eq!(body["unavailable"], json!(["installed_runtime_prepared_local_activity", "installed_runtime_prepared_local_activity_groups"]));
+
+    // A version native does not support at all is the protocol gate's 400.
+    let (status, body) = request_protocol(&app, "POST", &format!("{base}/prepare"), json!({}), "2.0").await;
+    assert_eq!((status, &body["reason"]), (StatusCode::BAD_REQUEST, &json!("unsupported_protocol_version")), "{body}");
+}
diff --git a/rust/tests/execution/wait_positions.rs b/rust/tests/execution/wait_positions.rs
new file mode 100644
index 000000000..d19f41ce3
--- /dev/null
+++ b/rust/tests/execution/wait_positions.rs
@@ -0,0 +1,139 @@
+//! Stored wait positions are chronological over every retained wait of a run,
+//! as Workflow #941's projector stores them, and a transition writes only the
+//! waits whose position moved (Server #783).
+use super::run_observation_signals::{claim, finish, register_signal_worker, signal};
+use super::*;
+
+const RUN: &str = "wait-positions";
+
+/// Log position writes. SQLite and PostgreSQL log every write of the column,
+/// including a rewrite of the same value; MySQL has no column-scoped update
+/// trigger, so it logs only writes that change the value.
+async fn log_position_writes(database: &TestDatabase) {
+    database.execute("CREATE TABLE wait_position_writes (wait_id VARCHAR(191) NOT NULL, old_position BIGINT NOT NULL, new_position BIGINT NOT NULL)").await;
+    match database {
+        TestDatabase::Sqlite(_) => {
+            database.execute("CREATE TRIGGER log_wait_position AFTER UPDATE OF position ON workflow_run_waits BEGIN INSERT INTO wait_position_writes VALUES (new.wait_id, old.position, new.position); END").await;
+        }
+        TestDatabase::Postgres { .. } => {
+            database.execute("CREATE FUNCTION log_wait_position() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN INSERT INTO wait_position_writes VALUES (NEW.wait_id, OLD.position, NEW.position); RETURN NEW; END $$").await;
+            database.execute("CREATE TRIGGER log_wait_position AFTER UPDATE OF position ON workflow_run_waits FOR EACH ROW EXECUTE FUNCTION log_wait_position()").await;
+        }
+        TestDatabase::MySql { .. } => {
+            database.execute("CREATE TRIGGER log_wait_position AFTER UPDATE ON workflow_run_waits FOR EACH ROW INSERT INTO wait_position_writes SELECT NEW.wait_id, OLD.position, NEW.position FROM DUAL WHERE NOT (OLD.position <=> NEW.position)").await;
+        }
+    }
+}
+
+/// Logged position writes as `(wait_id, "old>new")`, sorted by wait id.
+async fn position_writes(database: &TestDatabase) -> Vec<(String, String)> {
+    sorted(text_rows(database,
+        "SELECT wait_id, old_position||'>'||new_position FROM wait_position_writes",
+        "SELECT wait_id, old_position::text||'>'||new_position::text FROM wait_position_writes",
+        "SELECT CAST(wait_id AS CHAR), CAST(CONCAT(old_position,'>',new_position) AS CHAR) FROM wait_position_writes").await)
+}
+
+/// The run's waits as `(wait_id, "position:status")`, sorted by wait id.
+async fn waits(database: &TestDatabase) -> Vec<(String, String)> {
+    sorted(text_rows(database,
+        "SELECT wait_id, position||':'||status FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'",
+        "SELECT wait_id, position::text||':'||status FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'",
+        "SELECT CAST(wait_id AS CHAR), CAST(CONCAT(position,':',status) AS CHAR) FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'").await)
+}
+
+/// The run's waits as `(wait_id, kind)`, sorted by wait id.
+async fn kinds(database: &TestDatabase) -> Vec<(String, String)> {
+    sorted(text_rows(database,
+        "SELECT wait_id, kind FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'",
+        "SELECT wait_id, kind FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'",
+        "SELECT CAST(wait_id AS CHAR), CAST(kind AS CHAR) FROM workflow_run_waits WHERE workflow_instance_id='wait-positions'").await)
+}
+
+fn sorted(mut rows: Vec<(String, String)>) -> Vec<(String, String)> {
+    rows.sort();
+    rows
+}
+
+fn expected(rows: &[(&str, &str)]) -> Vec<(String, String)> {
+    sorted(rows.iter().map(|(id, value)| ((*id).to_owned(), (*value).to_owned())).collect())
+}
+
+async fn set_wait(database: &TestDatabase, wait_id: &str, opened_at: &str, position: i64) {
+    database.execute(sqlx::AssertSqlSafe(format!(
+        "UPDATE workflow_run_waits SET opened_at='{opened_at}',position={position} WHERE workflow_instance_id='{RUN}' AND wait_id='{wait_id}'"
+    ))).await;
+}
+
+#[tokio::test]
+async fn wait_positions_are_chronological_and_written_only_when_a_wait_moves() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    register_signal_worker(&app, "default").await;
+    let run = namespaces::start_scoped(&app, "default", RUN).await;
+    let task = claim(&app, "default").await;
+    let opened = finish(&app, "default", &task, json!([
+        {"type":"start_timer","delay_seconds":3600},
+        {"type":"start_timer","delay_seconds":7200},
+        {"type":"open_signal_wait","signal_name":"payload"}])).await;
+    assert_eq!(opened.0, StatusCode::OK, "{}", opened.1);
+    let opened = kinds(&database).await;
+    let timers: Vec<&str> = opened.iter().filter(|(_, kind)| kind == "timer").map(|(id, _)| id.as_str()).collect();
+    let signals: Vec<&str> = opened.iter().filter(|(_, kind)| kind == "signal").map(|(id, _)| id.as_str()).collect();
+    assert_eq!((timers.len(), signals.len(), opened.len()), (2, 1, 3), "{opened:?}");
+    let (first_signal, early_timer, late_timer) = (signals[0].to_owned(), timers[0].to_owned(), timers[1].to_owned());
+    // Retained clocks that run against wait-id order, stored in chronological
+    // order: the later timer id opened first and the earlier one last.
+    set_wait(&database, &late_timer, "2026-01-01 00:00:01", 0).await;
+    set_wait(&database, &first_signal, "2026-01-01 00:00:02", 1).await;
+    set_wait(&database, &early_timer, "2026-01-01 00:00:03", 2).await;
+    log_position_writes(&database).await;
+
+    // Resolving a wait moves no wait: the open-wait ordinal would have moved
+    // the last timer from 2 to 1 and rewritten the first in place.
+    let received = signal(&app, "default", &run).await;
+    assert_eq!(received.0, StatusCode::ACCEPTED, "{}", received.1);
+    let task = claim(&app, "default").await;
+    assert_eq!(
+        waits(&database).await,
+        expected(&[(&late_timer, "0:open"), (&first_signal, "1:resolved"), (&early_timer, "2:open")])
+    );
+    assert_eq!(position_writes(&database).await, Vec::new());
+
+    // A new wait opens after every retained wait and is the only one that
+    // may be written.
+    let reopened = finish(&app, "default", &task, json!([{"type":"open_signal_wait","signal_name":"payload"}])).await;
+    assert_eq!(reopened.0, StatusCode::OK, "{}", reopened.1);
+    let current = kinds(&database).await;
+    let second_signal = current.iter().find(|(id, kind)| kind == "signal" && *id != first_signal).map(|(id, _)| id.clone()).unwrap();
+    assert_eq!(
+        waits(&database).await,
+        expected(&[(&late_timer, "0:open"), (&first_signal, "1:resolved"), (&early_timer, "2:open"), (&second_signal, "3:open")])
+    );
+    let writes = position_writes(&database).await;
+    assert!(writes.iter().all(|(id, _)| *id == second_signal), "{writes:?}");
+
+    // Positions stored by another projector converge on the chronological
+    // ordinal, and only those rows are written.
+    set_wait(&database, &late_timer, "2026-01-01 00:00:01", 9).await;
+    set_wait(&database, &early_timer, "2026-01-01 00:00:03", 7).await;
+    database.execute("DELETE FROM wait_position_writes").await;
+    let received = signal(&app, "default", &run).await;
+    assert_eq!(received.0, StatusCode::ACCEPTED, "{}", received.1);
+    let task = claim(&app, "default").await;
+    let closed = finish(&app, "default", &task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await;
+    assert_eq!(closed.0, StatusCode::OK, "{}", closed.1);
+    let positions: Vec<(String, String)> = waits(&database)
+        .await
+        .into_iter()
+        .map(|(id, value)| (id, value.split(':').next().unwrap().to_owned()))
+        .collect();
+    assert_eq!(
+        positions,
+        expected(&[(&late_timer, "0"), (&first_signal, "1"), (&early_timer, "2"), (&second_signal, "3")])
+    );
+    assert_eq!(
+        position_writes(&database).await,
+        expected(&[(&early_timer, "7>2"), (&late_timer, "9>0")])
+    );
+}
diff --git a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt
index eea755f16..d22e43f90 100644
--- a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt
+++ b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt
@@ -20,11 +20,3 @@ POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{op
 GET /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/service-calls/{serviceCallId}
 POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/service-calls/{serviceCallId}/cancel
 POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint
-POST /api/worker/workflow-tasks/{taskId}/local-activities/checkpoint
-POST /api/worker/workflow-tasks/{taskId}/local-activities/checkpoint-group
-POST /api/worker/workflow-tasks/{taskId}/local-activities/prepare
-POST /api/worker/workflow-tasks/{taskId}/local-activities/recover
-POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/acknowledge-cancellation
-POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/control
-POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/heartbeat
-POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/outcome