From d7abee65b0c59a7833d158618c477cd8c6941c96 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 00:42:31 +0000
Subject: [PATCH 1/8] Release message streams at retention as PHP does

Native's retention pass held every run whose instance had inbound message
streams, because PHP's MessageStreamRetentionCleanup was never ported. A
namespace that used message streams therefore kept those runs forever.

The prune now plans and applies the stream cleanup in the run's own
transaction. The instance's last retained run, once terminal, takes its
streams and items with it. An earlier run's consumed items lose their
payloads once a retained later run's cursor checkpoint covers them. Tested
on SQLite, PostgreSQL, MySQL and MariaDB.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../2026-10-27-message-stream-retention.md    |  36 +++++
 docs/rust-server-port.md                      |  16 +--
 rust/src/runtime/operator_metrics.rs          | 123 +++++++++++++++++-
 rust/src/runtime/retention.rs                 |  32 +++--
 rust/tests/execution/message_streams.rs       |  93 +++++++++++++
 5 files changed, 272 insertions(+), 28 deletions(-)
 create mode 100644 docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md

diff --git a/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
new file mode 100644
index 000000000..a8994e890
--- /dev/null
+++ b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
@@ -0,0 +1,36 @@
+# October 27, message-stream retention
+
+- **Before.** Native's retention pass held any run that had
+  `workflow_messages` rows or whose instance had inbound message streams, with
+  `message_stream_cleanup_unavailable`. Published PHP's
+  `MessageStreamRetentionCleanup` was never ported. So a namespace that used
+  message streams kept those runs' details forever, and takeover refused any
+  PHP database with stream rows.
+- **Change.** `prune_run` plans and applies the stream cleanup in the run's
+  own prune transaction, under the transition lock. The plan follows PHP's
+  `planForRun`, read before the run's summary goes:
+  - **No inbound streams on the instance:** nothing to do.
+  - **The instance's current run is terminal and no other run summary
+    remains:** every item of the instance's streams is deleted, then the
+    streams. Reported as `message_stream_items_deleted` and
+    `message_streams_deleted`.
+  - **Otherwise:** on each stream whose `cursor_checkpoint_run_id` is a
+    retained later run (another run summary of the instance, or its current
+    run), the items this run consumed lose their payloads. That covers items
+    at or before the stream's cursor that are not released yet: `payload_blob`
+    becomes `''` and `payload_released_at` is set. The item rows stay as
+    identity tombstones. Reported as `message_stream_items_compacted`.
+    `cleanup_blocked_*` markers naming this run are cleared.
+  - The run's `workflow_messages` rows go with the run's other details, as
+    before.
+- **Not ported.** PHP's `markBlocked`, which marks a stream when
+  external-payload cleanup blocks a prune. Native never deletes external
+  objects: a run holding an external reference stays held as
+  `external_payload_cleanup_unavailable`, and native refuses external
+  references as message input.
+- **Tests** on SQLite, PostgreSQL, MySQL and MariaDB:
+  - `retention_releases_consumed_messages_and_removes_the_last_runs_streams`:
+    pruning a continued run compacts the two messages it consumed and keeps the
+    third. Pruning the last run deletes the stream and its three items.
+  - `retention_keeps_consumed_messages_no_retained_checkpoint_covers`: without
+    a retained checkpoint, nothing is compacted and the stream stays.
diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index 5c83a58a6..5f32478dd 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -412,14 +412,14 @@ contract fixture before its correction.
   workflow task guard does not check registration freshness yet, and the
   local activity and cancellation scope checkpoints still use the older fence.
   Add a fixture for each before changing them.
-- **Retention of runs with external payloads or message streams.**
-  Published PHP deletes a pruned run's external payload objects and releases
-  its message stream items. It also compacts its instance's consumed inbound
-  stream items, and removes the instance's streams with its last run. Native
-  does not clean any of these up yet. It skips such a run, including any run
-  whose instance has inbound streams, with
-  `external_payload_cleanup_unavailable` or
-  `message_stream_cleanup_unavailable`, and keeps its details.
+- **Retention of runs with external payloads.** Published PHP deletes a
+  pruned run's external payload objects. Native does not delete them yet: it
+  skips such a run with `external_payload_cleanup_unavailable` and keeps its
+  details. A run whose instance has inbound message streams is pruned as PHP's
+  `MessageStreamRetentionCleanup` prunes it (October 27): the messages it
+  consumed lose their payloads once a retained later run's cursor checkpoint
+  covers them, and the instance's last retained run, once terminal, takes the
+  instance's streams with it.
 - **Deeply nested JSON bodies.** Published PHP decodes JSON up to 512 levels
   deep. Native checks every body as PHP does before any route, but its
   handlers parse with serde_json's 128-level limit. A body nested 129 to 512
diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs
index 815835b3c..e39c76c2d 100644
--- a/rust/src/runtime/operator_metrics.rs
+++ b/rust/src/runtime/operator_metrics.rs
@@ -1353,7 +1353,7 @@ where
             return Ok(Skipped(reason));
         }
         let mut tx = self.begin().await?;
-        let run = Self::query("SELECT status,closed_at FROM workflow_runs WHERE id=$1")
+        let run = Self::query("SELECT status,closed_at,workflow_instance_id FROM workflow_runs WHERE id=$1")
             .bind(run_id)
             .fetch_one(&mut *tx)
             .await?;
@@ -1366,10 +1366,8 @@ where
         }
         let mut deleted = serde_json::Map::new();
         deleted.insert("run_deleted".into(), json!(0));
-        // A run with message streams is held above, so none are released.
-        for key in retention::STREAM_COUNTS {
-            deleted.insert(key.into(), json!(0));
-        }
+        // Planned before the run's own summary goes, as PHP plans it.
+        let streams = Self::stream_retention_plan(&mut tx, self.namespace.as_ref(), run_id, &DB::string(&run, "workflow_instance_id")?).await?;
         for (key, sql, binds) in retention::PRUNED {
             let mut query = Self::query(sql);
             for _ in 0..*binds {
@@ -1383,6 +1381,10 @@ where
                 deleted.insert((*key).into(), json!(affected));
             }
         }
+        let [streams_deleted, items_deleted, items_compacted] = Self::apply_stream_retention(&mut tx, run_id, &streams).await?;
+        for (key, value) in retention::STREAM_COUNTS.into_iter().zip([streams_deleted, items_deleted, items_compacted]) {
+            deleted.insert(key.into(), json!(value));
+        }
         // The run's history is gone; no cached read of it may survive.
         tx.scope_history = None;
         Self::query("UPDATE workflow_runs SET details_pruned_at=$1 WHERE id=$2 AND details_pruned_at IS NULL")
@@ -1394,9 +1396,116 @@ where
         Ok(Pruned(deleted))
     }
 
+    /// `MessageStreamRetentionCleanup::planForRun`: what pruning `run_id`
+    /// releases of its instance's inbound message streams.
+    ///
+    /// An active or continued instance keeps its streams, cursor checkpoint,
+    /// pending items and identity rows. An item the run consumed loses its
+    /// payload only once a retained later run's cursor checkpoint covers it,
+    /// so replay never needs it again. A terminal instance keeps all stream
+    /// state until its last retained run goes, which removes every stream.
+    async fn stream_retention_plan(
+        tx: &mut DB::Connection,
+        namespace: &str,
+        run_id: &str,
+        instance_id: &str,
+    ) -> Result<retention::StreamPlan> {
+        let streams = Self::query("SELECT id,cursor_position,cursor_checkpoint_run_id FROM workflow_inbound_streams WHERE namespace=$1 AND workflow_instance_id=$2 ORDER BY id")
+            .bind(namespace)
+            .bind(instance_id)
+            .fetch_all(&mut *tx)
+            .await?;
+        let mut plan = retention::StreamPlan::default();
+        if streams.is_empty() {
+            return Ok(plan);
+        }
+        let current = Self::query("SELECT current_run_id FROM workflow_instances WHERE id=$1 AND namespace=$2")
+            .bind(instance_id)
+            .bind(namespace)
+            .fetch_optional(&mut *tx)
+            .await?
+            .map(|row| DB::optional_string(&row, "current_run_id"))
+            .transpose()?
+            .flatten();
+        let current_terminal = match &current {
+            Some(current) => Self::query("SELECT status FROM workflow_runs WHERE id=$1")
+                .bind(current)
+                .fetch_optional(&mut *tx)
+                .await?
+                .map(|row| DB::string(&row, "status"))
+                .transpose()?
+                .is_some_and(|status| retention::terminal(&status)),
+            None => false,
+        };
+        let mut retained = Vec::new();
+        for row in Self::query("SELECT id FROM workflow_run_summaries WHERE namespace=$1 AND workflow_instance_id=$2 AND id<>$3")
+            .bind(namespace)
+            .bind(instance_id)
+            .bind(run_id)
+            .fetch_all(&mut *tx)
+            .await?
+        {
+            retained.push(DB::string(&row, "id")?);
+        }
+        for row in &streams {
+            plan.stream_ids.push(DB::number(row, "id")?);
+        }
+        if current_terminal && retained.is_empty() {
+            plan.delete_instance_state = true;
+            return Ok(plan);
+        }
+        if let Some(current) = current.filter(|current| current != run_id && !retained.contains(current)) {
+            retained.push(current);
+        }
+        for row in &streams {
+            if DB::optional_string(row, "cursor_checkpoint_run_id")?.is_some_and(|checkpoint| retained.contains(&checkpoint)) {
+                plan.checkpointed.push((DB::number(row, "id")?, DB::number(row, "cursor_position")?));
+            }
+        }
+        Ok(plan)
+    }
+
+    /// `MessageStreamRetentionCleanup::apply`, in the prune's transaction:
+    /// streams deleted, items deleted and items compacted.
+    async fn apply_stream_retention(tx: &mut DB::Connection, run_id: &str, plan: &retention::StreamPlan) -> Result<[u64; 3]> {
+        if plan.stream_ids.is_empty() {
+            return Ok([0, 0, 0]);
+        }
+        let mut counts = [0, 0, 0];
+        if plan.delete_instance_state {
+            for stream_id in &plan.stream_ids {
+                counts[1] += DB::affected(Self::query("DELETE FROM workflow_inbound_stream_items WHERE stream_id=$1").bind(*stream_id).execute(&mut *tx).await?);
+                counts[0] += DB::affected(Self::query("DELETE FROM workflow_inbound_streams WHERE id=$1").bind(*stream_id).execute(&mut *tx).await?);
+            }
+            return Ok(counts);
+        }
+        let at = now();
+        for (stream_id, cursor) in &plan.checkpointed {
+            counts[2] += DB::affected(
+                Self::query("UPDATE workflow_inbound_stream_items SET payload_blob='',payload_released_at=$1,updated_at=$2 WHERE stream_id=$3 AND consumed_run_id=$4 AND consumed_at IS NOT NULL AND payload_released_at IS NULL AND position<=$5")
+                    .bind(DB::bind_time(at))
+                    .bind(DB::bind_time(at))
+                    .bind(*stream_id)
+                    .bind(run_id)
+                    .bind(*cursor)
+                    .execute(&mut *tx)
+                    .await?,
+            );
+        }
+        for stream_id in &plan.stream_ids {
+            Self::query("UPDATE workflow_inbound_streams SET cleanup_blocked_at=NULL,cleanup_blocked_reason=NULL,cleanup_blocked_run_id=NULL,updated_at=$1 WHERE id=$2 AND cleanup_blocked_run_id=$3")
+                .bind(DB::bind_time(at))
+                .bind(*stream_id)
+                .bind(run_id)
+                .execute(&mut *tx)
+                .await?;
+        }
+        Ok(counts)
+    }
+
     /// What keeps a closed run's details: an activity it abandoned that is
-    /// still open, a message stream, or a stored external payload reference,
-    /// whose cleanup native does not perform yet.
+    /// still open, or a stored external payload reference, whose cleanup
+    /// native does not perform yet.
     async fn retention_hold(&self, run_id: &str) -> Result<Option<&'static str>> {
         if self.abandoned_activity_open(run_id).await? {
             return Ok(Some("detached_activity_still_open"));
diff --git a/rust/src/runtime/retention.rs b/rust/src/runtime/retention.rs
index 3053e0c8b..bd1844a2c 100644
--- a/rust/src/runtime/retention.rs
+++ b/rust/src/runtime/retention.rs
@@ -3,9 +3,10 @@
 //! with `details_pruned_at`.
 //!
 //! Native never deletes stored external payload objects. A run that holds a
-//! reference to one, an open abandoned activity or a message stream is
-//! skipped with its reason, so a prune never strands or loses data PHP would
-//! have handled.
+//! reference to one, or an open abandoned activity, is skipped with its
+//! reason, so a prune never strands or loses data PHP would have handled.
+//! Inbound message streams are released as PHP's
+//! `MessageStreamRetentionCleanup` releases them.
 use super::laravel_validation::{Rule, failure_messages, rule_errors};
 use chrono::{DateTime, Duration, Utc};
 use serde_json::{Map, Value, json};
@@ -87,16 +88,21 @@ pub(super) const STREAM_COUNTS: [&str; 3] =
     ["message_streams_deleted", "message_stream_items_deleted", "message_stream_items_compacted"];
 
 /// Counts that hold a run's details, the run-id binds of each, and the
-/// skip reason. An abandoned activity's hold is read from history. A run
-/// with messages or whose instance has inbound message streams is held too:
-/// native does not compact or remove that state as PHP's
-/// MessageStreamRetentionCleanup does yet, so it keeps the run rather than
-/// orphan it. Durable workflow streams go with the run (`DURABLE_STREAMS`)
-/// unless they hold an external reference (`Backend::RUN_STREAM_REFERENCES`).
-pub(super) const HOLDS: &[(&str, usize, &str)] = &[
-    ("SELECT (SELECT COUNT(*) FROM workflow_messages WHERE workflow_run_id=$1)+(SELECT COUNT(*) FROM workflow_inbound_streams WHERE workflow_instance_id IN (SELECT workflow_instance_id FROM workflow_runs WHERE id=$2))", 2, "message_stream_cleanup_unavailable"),
-    (REFERENCES_SQL, 5, "external_payload_cleanup_unavailable"),
-];
+/// skip reason. An abandoned activity's hold is read from history. Durable
+/// workflow streams go with the run (`DURABLE_STREAMS`) unless they hold an
+/// external reference (`Backend::RUN_STREAM_REFERENCES`).
+pub(super) const HOLDS: &[(&str, usize, &str)] = &[(REFERENCES_SQL, 5, "external_payload_cleanup_unavailable")];
+
+/// `MessageStreamRetentionCleanup::planForRun`'s plan for one run: the
+/// instance's inbound streams, whether they all go with the instance's last
+/// retained run, and otherwise each stream whose cursor checkpoint belongs
+/// to a retained later run, with that cursor.
+#[derive(Default)]
+pub(super) struct StreamPlan {
+    pub(super) stream_ids: Vec<i64>,
+    pub(super) delete_instance_state: bool,
+    pub(super) checkpointed: Vec<(i64, i64)>,
+}
 
 /// `server.history.retention_days`.
 fn default_days() -> i64 {
diff --git a/rust/tests/execution/message_streams.rs b/rust/tests/execution/message_streams.rs
index 254cfb6f8..2bff785b5 100644
--- a/rust/tests/execution/message_streams.rs
+++ b/rust/tests/execution/message_streams.rs
@@ -176,3 +176,96 @@ async fn cluster_info_and_workers_see_the_contract() {
     assert_eq!(capability["supported"], true, "{capability}");
     assert_eq!(capability["minimum_worker_protocol_version"], "1.15");
 }
+
+/// Counts rows with a query built for this test.
+async fn rows(database: &TestDatabase, sql: String) -> i64 {
+    task_retry::rows(database, Box::leak(sql.into_boxed_str())).await
+}
+
+// Published PHP's MessageStreamRetentionCleanup. Pruning a continued run
+// releases the payloads of the messages it consumed once the next run's
+// cursor checkpoint covers them, and keeps their identity rows. Pruning the
+// instance's last retained run, once it is terminal, removes its streams.
+#[tokio::test]
+async fn retention_releases_consumed_messages_and_removes_the_last_runs_streams() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    assert_eq!(request(&app, "POST", "/api/worker/register", stream_worker()).await.0, StatusCode::CREATED);
+    let started = start(&app, "streams-kept").await;
+    let first = started["run_id"].as_str().unwrap().to_owned();
+    for (id, n) in [("m1", 1), ("m2", 2)] {
+        assert_eq!(append(&app, "streams-kept", "orders", json!({"message_id":id,"input":[{"n":n}]})).await.0, StatusCode::ACCEPTED);
+    }
+    // The first run consumes both messages and continues as new.
+    let task = poll(&app, "streams", "workflow").await;
+    let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":2}]});
+    let (status, body) = complete(&app, &task, json!([{"type":"continue_as_new","arguments":envelope(Payload::Array(vec![]))}]), cursor, "1.20").await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    assert_eq!(append(&app, "streams-kept", "orders", json!({"message_id":"m3","input":[{"n":3}]})).await.0, StatusCode::ACCEPTED);
+    let task = poll(&app, "streams", "workflow").await;
+    let second = task["run_id"].as_str().unwrap().to_owned();
+    assert_ne!(second, first);
+    let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":3}]});
+    let (status, body) = complete(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::String("done".into()))}]), cursor, "1.20").await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    let expire = |run: &str| format!("UPDATE {{table}} SET closed_at='2020-01-01 00:00:00.000000' WHERE id='{run}'");
+    for table in ["workflow_runs", "workflow_run_summaries"] {
+        database.execute(sqlx::AssertSqlSafe(expire(&first).replace("{table}", table))).await;
+    }
+
+    // The first run goes; the second run's checkpoint covers what it consumed.
+    let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[first]})).await;
+    assert_eq!(status, StatusCode::OK, "{pruned}");
+    let deleted = &pruned["results"][0]["deleted"];
+    assert_eq!(pruned["results"][0]["outcome"], "pruned", "{pruned}");
+    assert_eq!(
+        [&deleted["message_stream_items_compacted"], &deleted["message_stream_items_deleted"], &deleted["message_streams_deleted"]],
+        [&json!(2), &json!(0), &json!(0)],
+        "{pruned}"
+    );
+    assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE payload_released_at IS NOT NULL AND payload_blob=''".into()).await, 2);
+    assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE message_id='m3' AND payload_released_at IS NULL AND payload_blob<>''".into()).await, 1);
+    let orders = &streams(&app, "streams-kept").await["streams"][0];
+    assert_eq!((&orders["last_position"], &orders["cursor_position"]), (&json!(3), &json!(3)), "{orders}");
+
+    // The last run goes too, and the instance's stream with it.
+    for table in ["workflow_runs", "workflow_run_summaries"] {
+        database.execute(sqlx::AssertSqlSafe(expire(&second).replace("{table}", table))).await;
+    }
+    let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[second]})).await;
+    assert_eq!(status, StatusCode::OK, "{pruned}");
+    let deleted = &pruned["results"][0]["deleted"];
+    assert_eq!(
+        [&deleted["message_stream_items_compacted"], &deleted["message_stream_items_deleted"], &deleted["message_streams_deleted"]],
+        [&json!(0), &json!(3), &json!(1)],
+        "{pruned}"
+    );
+    assert_eq!(rows(&database, "SELECT (SELECT COUNT(*) FROM workflow_inbound_streams)+(SELECT COUNT(*) FROM workflow_inbound_stream_items)".into()).await, 0);
+}
+
+// Without a retained checkpoint past them, a run's consumed messages keep
+// their payloads: replaying the instance's current run still needs them.
+#[tokio::test]
+async fn retention_keeps_consumed_messages_no_retained_checkpoint_covers() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    assert_eq!(request(&app, "POST", "/api/worker/register", stream_worker()).await.0, StatusCode::CREATED);
+    let started = start(&app, "streams-open").await;
+    let first = started["run_id"].as_str().unwrap().to_owned();
+    assert_eq!(append(&app, "streams-open", "orders", json!({"message_id":"m1","input":[{"n":1}]})).await.0, StatusCode::ACCEPTED);
+    let task = poll(&app, "streams", "workflow").await;
+    let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":1}]});
+    let (status, body) = complete(&app, &task, json!([{"type":"continue_as_new","arguments":envelope(Payload::Array(vec![]))}]), cursor, "1.20").await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    // The checkpoint names a run that is no longer retained.
+    database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_checkpoint_run_id='gone'".to_owned())).await;
+    for table in ["workflow_runs", "workflow_run_summaries"] {
+        database.execute(sqlx::AssertSqlSafe(format!("UPDATE {table} SET closed_at='2020-01-01 00:00:00.000000' WHERE id='{first}'"))).await;
+    }
+    let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[first]})).await;
+    assert_eq!(status, StatusCode::OK, "{pruned}");
+    assert_eq!(pruned["results"][0]["outcome"], "pruned", "{pruned}");
+    assert_eq!(pruned["results"][0]["deleted"]["message_stream_items_compacted"], 0, "{pruned}");
+    assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE payload_released_at IS NULL AND payload_blob<>''".into()).await, 1);
+    assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_streams".into()).await, 1, "the open instance keeps its stream");
+}

From 4e9012c58a5ba8a86844c1417bf4cdfb28944159 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 00:55:35 +0000
Subject: [PATCH 2/8] Keep PHP message and workflow streams at takeover

Native serves inbound message streams and durable workflow streams from
PHP's own tables, and now releases them at retention as PHP does, so
takeover carries their rows over instead of refusing the source.

Each closed or errored durable stream gets the reclaim deadline native's
retention pass reads: PHP 2.11.3's reclaim_at, or closed_at plus its
retention for earlier releases, as 2.11.3's migration backfills it.
Inspection now counts stream rows that hold an external payload reference,
which takeover still refuses. Only workflow_messages, which native does
not deliver yet, stays refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../2026-10-27-message-stream-retention.md    | 22 ++++++++
 docs/rust-server-upgrade-takeover.md          | 14 +++++
 rust/src/runtime/backend.rs                   | 18 ++++++
 rust/src/runtime/upgrade.rs                   | 12 ++++
 rust/src/runtime/upgrade/stream_scan.rs       | 19 +++++++
 rust/src/runtime/upgrade/takeover.rs          | 56 +++++++++++++++----
 .../runtime/upgrade/takeover/mysql_journal.rs |  5 ++
 rust/src/runtime/upgrade_tests.rs             | 48 +++++++++++++++-
 8 files changed, 180 insertions(+), 14 deletions(-)
 create mode 100644 rust/src/runtime/upgrade/stream_scan.rs

diff --git a/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
index a8994e890..30b5e83da 100644
--- a/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
+++ b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
@@ -34,3 +34,25 @@
     third. Pruning the last run deletes the stream and its three items.
   - `retention_keeps_consumed_messages_no_retained_checkpoint_covers`: without
     a retained checkpoint, nothing is compacted and the stream stays.
+
+## Takeover keeps message and workflow streams
+
+- **Before.** Takeover refused any source with rows in
+  `workflow_messages`, either kind of stream, or their items.
+- **Change.** Stream rows carry over unchanged.
+  - Each closed or errored durable stream gets its `dw_stream_reclaims`
+    deadline: 2.11.3's `reclaim_at`, or `closed_at` plus `retention_seconds`
+    (3600 unless set) for earlier layouts. Reported as
+    `stream_reclaims_recorded`; MySQL and MariaDB do it in journal step 10.
+  - Inspection reports `stream_reference_inspection`. A stream row holding an
+    external payload reference refuses with
+    `upgrade_source_external_payloads_unsupported`.
+  - Only `workflow_messages` stays refused, now as
+    `upgrade_source_workflow_messages_unsupported`.
+- **Tests.**
+  - `takeover_keeps_php_streams_and_records_their_reclaim_deadlines` (SQLite).
+  - The 2.11.3 takeover test now copies a recorded deadline.
+  - The refusal list covers a durable stream item holding an external
+    reference.
+  - Takeover qualification passes on PostgreSQL, MySQL and MariaDB,
+    including a MySQL takeover killed after step 9.
diff --git a/docs/rust-server-upgrade-takeover.md b/docs/rust-server-upgrade-takeover.md
index 7439f085a..a121504f5 100644
--- a/docs/rust-server-upgrade-takeover.md
+++ b/docs/rust-server-upgrade-takeover.md
@@ -77,6 +77,20 @@ inspection is `unqualified` and takeover refuses with
 `upgrade_source_external_payload_storage_unsupported` before any change. A
 disabled policy does not block takeover.
 
+Inbound message streams and durable workflow streams carry over unchanged:
+native serves them from PHP's own tables. Each closed or errored durable
+stream gets the reclaim deadline native's retention pass reads,
+`dw_stream_reclaims`: PHP 2.11.3's recorded `reclaim_at`, or for earlier
+releases its `closed_at` plus its retention (an hour unless set), as 2.11.3's
+migration backfills it. Every inspection reports
+`stream_reference_inspection`: the number of stream items, durable streams
+and inbound messages holding an external payload reference. Native does not
+resolve those, so any such row makes the inspection `unqualified` and takeover
+refuses with `upgrade_source_external_payloads_unsupported`.
+Workflow-to-workflow messages (`workflow_messages`) are not delivered by
+native yet, so a source holding any refuses with
+`upgrade_source_workflow_messages_unsupported`.
+
 PHP 2.11.3 (Core 2.14) schedules a service call's handler in the
 `workflow_service_call_dispatches` outbox and dispatches it in a later pass.
 Native runs the handler when it accepts the call, so it cannot resume a
diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
index 179490386..761a4dd8b 100644
--- a/rust/src/runtime/backend.rs
+++ b/rust/src/runtime/backend.rs
@@ -94,6 +94,14 @@ pub(super) trait Backend: Database {
     // reference, which native does not settle (Server #774). PostgreSQL
     // stores these payloads as `json`, matched through its text.
     const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), ")");
+    // Every stream item, durable stream and inbound message that holds an
+    // external payload reference, which takeover refuses as it refuses any
+    // other external payload.
+    const STREAM_REFERENCES: &str = concat!(
+        "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), ")",
+        "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", super::retention::reference!("metadata::text"), ")",
+        "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", super::retention::reference!("payload_blob"), ") AS n"
+    );
     const RUN_STREAM_REFERENCES: &str = concat!(
         "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), "))",
         "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata::text"), ")"
@@ -217,6 +225,11 @@ ORDER BY t.available_order,t.id LIMIT 100
 "#;
     const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND json_extract(payload,'$.selection_group_id')=$2 ORDER BY sequence LIMIT 1";
     const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")");
+    const STREAM_REFERENCES: &str = concat!(
+        "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")",
+        "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", super::retention::reference!("metadata"), ")",
+        "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", super::retention::reference!("payload_blob"), ") AS n"
+    );
     const RUN_STREAM_REFERENCES: &str = concat!(
         "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), "))",
         "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata"), ")"
@@ -477,6 +490,11 @@ impl Backend for MySql {
     const STREAM_SWEEP_BOUNDARY: &str = "SELECT `offset` FROM workflow_durable_stream_items WHERE stream_id=$1 ORDER BY `offset` LIMIT 1 OFFSET $2";
     const STREAM_SWEEP_THROUGH: &str = "DELETE FROM workflow_durable_stream_items WHERE stream_id=$1 AND `offset`<=$2";
     const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")");
+    const STREAM_REFERENCES: &str = concat!(
+        "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")",
+        "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", super::retention::reference!("metadata"), ")",
+        "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", super::retention::reference!("payload_blob"), ") AS n"
+    );
     const RUN_STREAM_REFERENCES: &str = concat!(
         "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), "))",
         "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata"), ")"
diff --git a/rust/src/runtime/upgrade.rs b/rust/src/runtime/upgrade.rs
index cbd0ffa98..310bd7b34 100644
--- a/rust/src/runtime/upgrade.rs
+++ b/rust/src/runtime/upgrade.rs
@@ -3,6 +3,7 @@ pub mod external_reference;
 pub mod fingerprint;
 pub mod takeover;
 mod dispatch_scan;
+mod stream_scan;
 mod payload_scan;
 mod schedule_scan;
 mod storage_scan;
@@ -255,6 +256,7 @@ fn report(
     schedule_inspection: Value,
     storage_inspection: Value,
     dispatch_inspection: Value,
+    stream_inspection: Value,
 ) -> Result<Value> {
     let digest = Sha256::digest(serde_json::to_vec(catalog)?)
         .iter()
@@ -291,6 +293,10 @@ fn report(
         report["status"] = json!("unqualified");
     }
     report["service_call_dispatch_inspection"] = dispatch_inspection;
+    if stream_inspection["references"] != 0 {
+        report["status"] = json!("unqualified");
+    }
+    report["stream_reference_inspection"] = stream_inspection;
     Ok(report)
 }
 
@@ -466,6 +472,7 @@ pub(super) async fn sqlite_inventory(
     let schedule_inspection = schedule_scan::inspect::<sqlx::Sqlite>(&mut *tx).await?;
     let storage_inspection = storage_scan::inspect::<sqlx::Sqlite>(&mut *tx).await?;
     let dispatch_inspection = dispatch_scan::inspect::<sqlx::Sqlite>(&mut *tx, &counts).await?;
+    let stream_inspection = stream_scan::inspect::<sqlx::Sqlite>(&mut *tx).await?;
     report(
         "sqlite",
         &catalog,
@@ -476,6 +483,7 @@ pub(super) async fn sqlite_inventory(
         schedule_inspection,
         storage_inspection,
         dispatch_inspection,
+        stream_inspection,
     )
 }
 
@@ -741,6 +749,7 @@ pub(super) async fn mysql_inventory(
     let schedule_inspection = schedule_scan::inspect::<sqlx::MySql>(&mut *tx).await?;
     let storage_inspection = storage_scan::inspect::<sqlx::MySql>(&mut *tx).await?;
     let dispatch_inspection = dispatch_scan::inspect::<sqlx::MySql>(&mut *tx, &counts).await?;
+    let stream_inspection = stream_scan::inspect::<sqlx::MySql>(&mut *tx).await?;
     // MySQL's metadata is not an MVCC snapshot. Refuse a changed visible
     // catalog rather than joining counts to a different schema.
     if mysql_source_catalog(&mut *tx, fenced).await?.0 != catalog {
@@ -759,6 +768,7 @@ pub(super) async fn mysql_inventory(
         schedule_inspection,
         storage_inspection,
         dispatch_inspection,
+        stream_inspection,
     )
 }
 
@@ -828,6 +838,7 @@ pub(super) async fn postgres_inventory(
     let schedule_inspection = schedule_scan::inspect::<sqlx::Postgres>(&mut *tx).await?;
     let storage_inspection = storage_scan::inspect::<sqlx::Postgres>(&mut *tx).await?;
     let dispatch_inspection = dispatch_scan::inspect::<sqlx::Postgres>(&mut *tx, &counts).await?;
+    let stream_inspection = stream_scan::inspect::<sqlx::Postgres>(&mut *tx).await?;
     report(
         "pgsql",
         &catalog,
@@ -838,6 +849,7 @@ pub(super) async fn postgres_inventory(
         schedule_inspection,
         storage_inspection,
         dispatch_inspection,
+        stream_inspection,
     )
 }
 
diff --git a/rust/src/runtime/upgrade/stream_scan.rs b/rust/src/runtime/upgrade/stream_scan.rs
new file mode 100644
index 000000000..7f9b5d8da
--- /dev/null
+++ b/rust/src/runtime/upgrade/stream_scan.rs
@@ -0,0 +1,19 @@
+//! Message and workflow stream rows that hold an external payload
+//! reference. Native carries streams over unchanged, but it neither resolves
+//! nor deletes stored external payload objects, so takeover refuses such a
+//! row as it refuses any other external payload. The report counts rows and
+//! never names streams.
+use crate::runtime::{Result, backend::Backend};
+use serde_json::{Value, json};
+use sqlx::{Executor, IntoArguments};
+
+pub(super) async fn inspect<DB: Backend>(connection: &mut DB::Connection) -> Result<Value>
+where
+    for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>,
+    DB::Arguments: IntoArguments<DB>,
+{
+    let row = sqlx::query(sqlx::AssertSqlSafe(DB::statement(DB::STREAM_REFERENCES)))
+        .fetch_one(&mut *connection)
+        .await?;
+    Ok(json!({"references": DB::number(&row, "n")?}))
+}
diff --git a/rust/src/runtime/upgrade/takeover.rs b/rust/src/runtime/upgrade/takeover.rs
index 7c0c251ab..1ec5748bb 100644
--- a/rust/src/runtime/upgrade/takeover.rs
+++ b/rust/src/runtime/upgrade/takeover.rs
@@ -221,16 +221,9 @@ fn report(backend: &str, layout: &str, digest: &Value, inventory: &Value, conver
 /// Published PHP features native does not implement yet. A source holding any
 /// of their rows is refused, rather than taken over without them. Derived
 /// projections and PHP's own queue and poll bookkeeping are left in place.
-const UNSUPPORTED_FEATURES: [(&str, &[&str]); 1] = [    (
-        "upgrade_source_message_streams_unsupported",
-        &[
-            "workflow_messages",
-            "workflow_durable_streams",
-            "workflow_durable_stream_items",
-            "workflow_inbound_streams",
-            "workflow_inbound_stream_items",
-        ],
-    ),
+const UNSUPPORTED_FEATURES: [(&str, &[&str]); 1] = [
+    // Workflow-to-workflow messages, which native does not deliver yet.
+    ("upgrade_source_workflow_messages_unsupported", &["workflow_messages"]),
 ];
 
 /// Takeover needs every inspection result it relies on, and refuses what
@@ -294,6 +287,7 @@ fn qualified(inventory: &Value) -> Result<&str> {
         || total("unqualified_references") != 0
         || total("other_unvalidated") != 0
         || inventory["table_rows"]["runtime_external_payloads"] != 0
+        || inventory["stream_reference_inspection"]["references"] != 0
     {
         return Err(refuse(
             StatusCode::CONFLICT,
@@ -406,6 +400,7 @@ where
     } else {
         0
     };
+    let reclaims = adopt_stream_reclaims::<DB>(tx, super::has_release(layout, "php-2.11.3")).await?;
     let release_2113 = if super::has_release(layout, "php-2.11.3") {
         let messages = count::<DB>(tx, FULL_FAILURE_MESSAGES).await?;
         let dispatches = count::<DB>(tx, COMPLETED_DISPATCHES).await?;
@@ -419,7 +414,8 @@ where
         "session_holders_recorded":holders,
         "service_call_deadlines_copied":deadlines,
         "queue_delivery_references_removed":delivery_references,
-        "observability_contexts_removed":observability_contexts
+        "observability_contexts_removed":observability_contexts,
+        "stream_reclaims_recorded":reclaims
     });
     if let (Value::Object(converted), Value::Object(release)) = (&mut converted, release_2113) {
         converted.extend(release);
@@ -427,6 +423,44 @@ where
     Ok(converted)
 }
 
+/// Native keeps each closed or errored durable stream's reclaim deadline in
+/// `dw_stream_reclaims`, which its retention pass reads. PHP 2.11.3 records it
+/// as `workflow_durable_streams.reclaim_at` (null once the stream's items are
+/// gone). Earlier releases recorded none, so the deadline is `closed_at` plus
+/// the stream's retention, an hour unless set, as 2.11.3's own migration
+/// backfills it. A repeated call records only streams not recorded yet.
+/// Returns how many it recorded.
+pub(super) async fn adopt_stream_reclaims<DB: Backend>(tx: &mut DB::Connection, recorded: bool) -> Result<usize>
+where
+    for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>,
+    DB::Arguments: IntoArguments<DB>,
+    for<'q> String: Encode<'q, DB> + Type<DB>,
+{
+    let rows = query::<DB>(if recorded {
+        "SELECT id,namespace,workflow_run_id,reclaim_at FROM workflow_durable_streams WHERE reclaim_at IS NOT NULL AND id NOT IN (SELECT stream_id FROM dw_stream_reclaims) ORDER BY id"
+    } else {
+        "SELECT id,namespace,workflow_run_id,closed_at,retention_seconds FROM workflow_durable_streams WHERE status IN ('closed','errored') AND closed_at IS NOT NULL AND id NOT IN (SELECT stream_id FROM dw_stream_reclaims) ORDER BY id"
+    })
+    .fetch_all(&mut *tx)
+    .await?;
+    for row in &rows {
+        let reclaim_at = if recorded {
+            DB::instant(row, "reclaim_at")?
+        } else {
+            DB::instant(row, "closed_at")?
+                + chrono::Duration::seconds(DB::optional_number(row, "retention_seconds")?.unwrap_or(3600))
+        };
+        query::<DB>("INSERT INTO dw_stream_reclaims(stream_id,namespace,workflow_run_id,reclaim_at) VALUES ($1,$2,$3,$4)")
+            .bind(DB::optional_integer(Some(DB::number(row, "id")?)))
+            .bind(DB::string(row, "namespace")?)
+            .bind(DB::string(row, "workflow_run_id")?)
+            .bind(DB::bind_time(reclaim_at))
+            .execute(&mut *tx)
+            .await?;
+    }
+    Ok(rows.len())
+}
+
 /// Core 2.14's full failure messages: a message beyond the old TEXT capacity,
 /// or with a NUL, keeps a labelled preview in `message` and its full bytes in
 /// `message_full`. Native has no second column; it reads the preview, as
diff --git a/rust/src/runtime/upgrade/takeover/mysql_journal.rs b/rust/src/runtime/upgrade/takeover/mysql_journal.rs
index fad95eab0..90c186ef0 100644
--- a/rust/src/runtime/upgrade/takeover/mysql_journal.rs
+++ b/rust/src/runtime/upgrade/takeover/mysql_journal.rs
@@ -390,6 +390,11 @@ async fn perform(connection: &mut MySqlConnection, run: &mut Run<'_>, step: i64)
             }
         }
         10 => {
+            // Durable stream reclaim deadlines, from 2.11.3's column while it
+            // is still there, else from each closed stream's retention.
+            let recorded = column_exists(connection, "workflow_durable_streams", "reclaim_at").await?;
+            let reclaims = super::adopt_stream_reclaims::<MySql>(connection, recorded).await?;
+            run.converted.insert("stream_reclaims_recorded".into(), json!(reclaims));
             // PHP 2.11.3's additions. Count before dropping, as step 9 does.
             if column_exists(connection, "workflow_failures", "message_full").await? {
                 let messages = count::<MySql>(connection, super::FULL_FAILURE_MESSAGES).await?;
diff --git a/rust/src/runtime/upgrade_tests.rs b/rust/src/runtime/upgrade_tests.rs
index b9bcdb4d7..db7309eb5 100644
--- a/rust/src/runtime/upgrade_tests.rs
+++ b/rust/src/runtime/upgrade_tests.rs
@@ -1108,6 +1108,7 @@ async fn takeover_converts_a_php_2113_database(release: &str, changes: &[&[&'sta
         "INSERT INTO workflow_failures(id,workflow_run_id,source_kind,source_id,propagation_kind,exception_class,message,file,message_full,created_at,updated_at) VALUES ('01m5wvy8d1j3dtr13es05kfail','01jrun0000000000000000000a','activity_execution','a','activity','RuntimeException','preview\n[full message stored separately]','','base64:ZnVsbA==','2026-10-26 10:00:00','2026-10-26 10:00:00')".to_owned(),
         "INSERT INTO workflow_service_calls(id,namespace,endpoint_name,service_name,operation_name,caller_namespace,target_namespace,status,outcome,operation_mode,idempotency_key,idempotency_scope_hash,accepted_at,completed_at,created_at,updated_at) VALUES ('01m5wvy8d1j3dtr13es05kcall','default','pay','billing','charge','default','default','completed','succeeded','async','order-1','0000000000000000000000000000000000000000000000000000000000000001','2026-10-26 10:00:00','2026-10-26 10:00:01','2026-10-26 10:00:00','2026-10-26 10:00:01')".to_owned(),
         "INSERT INTO workflow_service_call_dispatches(service_call_id,state,dispatch_options,available_at,completed_at,created_at,updated_at) VALUES ('01m5wvy8d1j3dtr13es05kcall','completed','{}','2026-10-26 10:00:00','2026-10-26 10:00:01','2026-10-26 10:00:00','2026-10-26 10:00:01')".to_owned(),
+        "INSERT INTO workflow_durable_streams(namespace,workflow_instance_id,workflow_run_id,stream_name,status,retention_seconds,closed_at,reclaim_at) VALUES ('default','i','01jrun0000000000000000000a','progress','closed',600,'2026-10-26 10:00:00','2026-10-26 10:10:00')".to_owned(),
     ];
     for statement in &statements {
         mutate(&path, Box::leak(statement.clone().into_boxed_str())).await;
@@ -1117,8 +1118,8 @@ async fn takeover_converts_a_php_2113_database(release: &str, changes: &[&[&'sta
     assert_eq!(report["status"], "taken_over", "{report}");
     assert_eq!(report["source_layout"], release);
     assert_eq!(
-        [&report["converted"]["session_holders_recorded"], &report["converted"]["full_failure_messages_removed"], &report["converted"]["service_call_dispatches_removed"]],
-        [&json!(2), &json!(1), &json!(1)],
+        [&report["converted"]["session_holders_recorded"], &report["converted"]["full_failure_messages_removed"], &report["converted"]["service_call_dispatches_removed"], &report["converted"]["stream_reclaims_recorded"]],
+        [&json!(2), &json!(1), &json!(1), &json!(1)],
         "{report}"
     );
     let mut connection = SqliteConnection::connect(&format!("sqlite://{path}")).await.unwrap();
@@ -1278,7 +1279,12 @@ async fn takeover_refusals_leave_the_php_database_unchanged() {
         // Features native lacks: their rows would be left behind unread.
         (
             "INSERT INTO workflow_messages(workflow_instance_id,direction,channel,stream_key,sequence) VALUES ('w','inbound','c','k',1)",
-            "upgrade_source_message_streams_unsupported",
+            "upgrade_source_workflow_messages_unsupported",
+        ),
+        // A stream item native would carry over but cannot resolve.
+        (
+            "INSERT INTO workflow_durable_stream_items(stream_id,namespace,workflow_run_id,stream_name,\"offset\",payload_reference,emitted_at) VALUES (1,'default','r','s',0,'dw-external-payload:v1:x','2026-10-26 10:00:00')",
+            "upgrade_source_external_payloads_unsupported",
         ),
         // A namespace whose uploads would all be refused: native serves
         // only the local driver.
@@ -1402,6 +1408,42 @@ async fn takeover_keeps_php_service_calls_and_their_deadlines() {
     runtime.close().await;
 }
 
+/// Message streams and durable workflow streams carry over: native serves
+/// them from PHP's own tables. A closed durable stream gets the reclaim
+/// deadline native's retention pass reads, its `closed_at` plus its
+/// retention, as PHP 2.11.3's migration backfills it; an open one gets none.
+#[tokio::test]
+async fn takeover_keeps_php_streams_and_records_their_reclaim_deadlines() {
+    let (_directory, path) = php_release_layout(&[PHP_280_CHANGES, TRACE_CARRIERS, QUEUE_DELIVERY]).await;
+    for statement in [
+        "INSERT INTO workflow_durable_streams(id,namespace,workflow_instance_id,workflow_run_id,stream_name,status,last_offset,total_items,retention_seconds,closed_at) VALUES (7,'default','i','01jrun0000000000000000000a','progress','closed',0,1,600,'2026-10-26 10:00:00')",
+        "INSERT INTO workflow_durable_streams(id,namespace,workflow_instance_id,workflow_run_id,stream_name,status) VALUES (8,'default','i','01jrun0000000000000000000a','log','open')",
+        "INSERT INTO workflow_durable_stream_items(stream_id,namespace,workflow_run_id,stream_name,\"offset\",payload,emitted_at) VALUES (7,'default','01jrun0000000000000000000a','progress',0,'{\"step\":1}','2026-10-26 09:59:00')",
+        "INSERT INTO workflow_inbound_streams(id,namespace,workflow_instance_id,stream_name,last_position) VALUES (3,'default','i','orders',1)",
+        "INSERT INTO workflow_inbound_stream_items(stream_id,namespace,workflow_instance_id,stream_name,message_id,position,payload_codec,payload_blob,payload_hash) VALUES (3,'default','i','orders','m1',1,'avro','AA==','h')",
+    ] {
+        mutate(&path, statement).await;
+    }
+    let inspection = inspect_sqlite(&path).await.unwrap();
+    assert_eq!(inspection["stream_reference_inspection"], json!({"references": 0}), "{inspection}");
+    let fingerprint = backup_fingerprint(&path).await;
+    let report = takeover::take_over_sqlite(&path, &fingerprint, &queue_config()).await.unwrap();
+    assert_eq!(report["status"], "taken_over", "{report}");
+    assert_eq!(report["converted"]["stream_reclaims_recorded"], 1, "{report}");
+    let mut connection = SqliteConnection::connect(&format!("sqlite://{path}")).await.unwrap();
+    let reclaims: Vec<(i64, String)> = sqlx::query_as("SELECT stream_id,reclaim_at FROM dw_stream_reclaims")
+        .fetch_all(&mut connection).await.unwrap();
+    assert_eq!(reclaims.len(), 1);
+    assert_eq!(reclaims[0].0, 7);
+    assert!(reclaims[0].1.starts_with("2026-10-26 10:10:00"), "{reclaims:?}");
+    let kept: i64 = sqlx::query_scalar("SELECT (SELECT COUNT(*) FROM workflow_durable_streams)+(SELECT COUNT(*) FROM workflow_durable_stream_items)+(SELECT COUNT(*) FROM workflow_inbound_streams)+(SELECT COUNT(*) FROM workflow_inbound_stream_items)")
+        .fetch_one(&mut connection).await.unwrap();
+    assert_eq!(kept, 5, "every stream row carries over");
+    connection.close().await.unwrap();
+    let runtime = crate::runtime::Runtime::open(&path, "test-token".into()).await.unwrap();
+    runtime.close().await;
+}
+
 /// PHP's update validation tasks carry over unchanged: native validates
 /// updates in the same `workflow_update_validation_tasks` rows, with no state
 /// of its own. A pending task whose caller was waiting on PHP is then what a

From 8b2a19df18bad80593e62baf4b2b15d337ff34c2 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 00:57:06 +0000
Subject: [PATCH 3/8] Refresh the Rust port's current state for October 27

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/rust-server-port.md | 120 +++++++++++++--------------------------
 1 file changed, 41 insertions(+), 79 deletions(-)

diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index 5f32478dd..c3e4a5f37 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -14,92 +14,54 @@ qualify a later tree.
 
 ## Current state
 
-Updated October 24, 2026.
-
-- **Merged October 24.** Train
-  [#729](https://github.com/durable-workflow/server/pull/729) (merge
-  `61219609b`):
-  - MySQL statement closes are sent with the next command
-    ([#717](https://github.com/durable-workflow/server/pull/717)), ending the
-    MySQL-family stalls of [#714](https://github.com/durable-workflow/server/issues/714);
-  - policy maps that are not JSON objects are refused, like PHP #705, and a
-    namespace quota map that is not UTF-8 is refused under either name
-    ([#719](https://github.com/durable-workflow/server/pull/719));
-  - worker history shows consumed signal sequences, hydrating only each page's
-    own signals, plus full task snapshots
-    ([#710](https://github.com/durable-workflow/server/pull/710));
-  - a newer push cancels a superseded Shared Server fixtures run
-    ([#724](https://github.com/durable-workflow/server/pull/724)).
-- **Merged October 20-23.** These are summarized in the checkpoint history:
-  - backup-first takeover on every database;
-  - multi-node kill and restart;
-  - ownership guards, search attributes, namespace update, history export,
-    retention and activity timeouts;
-  - the October 21-23 trains;
-  - the move to the published 2.11.1 reference;
-  - the GHSA-6q48-6c7w-9gw2 fixes.
-
-  Each train's PR holds its evidence.
-- **In review or qualifying, in merge order.**
-  1. CI speed and the merge queue:
-     - [#741](https://github.com/durable-workflow/server/pull/741) runs the
-       SDK direction profiles in parallel lanes, about 14 minutes off every
-       database job, and fixes a timestamp-comparison test flake;
-     - [#742](https://github.com/durable-workflow/server/pull/742) adds the
-       merge-queue prerequisites (`merge_group` triggers, one fixtures
-       result, and fixture cells skipped when a change cannot affect them).
-  2. This train: operator metrics read across PHP's redispatch steps
-     ([#708](https://github.com/durable-workflow/server/pull/708)),
-     query-authority clients that wait for the acknowledgement
-     ([#720](https://github.com/durable-workflow/server/pull/720)), and the
-     machine-checked route list below.
-  3. Condition waits without a key, task snapshots, the SQLite patch cohort and
-     message streams (#713, #718, #722, #709).
-  4. Update validation with PHP's poll request ID contract
-     ([#712](https://github.com/durable-workflow/server/pull/712)), then the
-     held-poll registration fence
-     ([#739](https://github.com/durable-workflow/server/pull/739), #734) and
-     worker sessions
-     ([#740](https://github.com/durable-workflow/server/pull/740), #736),
-     stacked in that order.
-  5. Activity cancellation acknowledgement, and pip, Composer and Cargo trust
-     in CI ([#673](https://github.com/durable-workflow/server/pull/673)).
+Updated October 27, 2026.
+
+- **Merged October 26-27, through the merge queue.**
+  - [#790](https://github.com/durable-workflow/server/pull/790): prepared
+    local-activity refusals.
+  - [#793](https://github.com/durable-workflow/server/pull/793): Nexus catalog
+    administration, service-call execute, show and cancel, shared admission
+    limits, and catalog call reads.
+  - [#795](https://github.com/durable-workflow/server/pull/795): takeover of
+    published 2.11.1 databases (fresh and upgraded from 2.9.x), keeping search
+    attribute definitions, worker sessions and service calls, plus the repair
+    pass's service-call deadline sweep.
+- **In the queue or next, in order.**
+  1. [#797](https://github.com/durable-workflow/server/pull/797):
+     - native's own maintenance repair loop;
+     - takeover keeps runtime credentials and update validation tasks;
+     - takeover of published 2.11.3, fresh and upgraded from 2.9.x.
+  2. Message-stream retention (PHP's `MessageStreamRetentionCleanup`), and
+     takeover that keeps message and workflow streams.
+  3. The parity reference moves to published 2.11.3. Native already matches it
+     on 62 of 65 fixtures. The move waits on two published 2.11.3 defects
+     General owns: the storage diagnostic (#789) and cross-caller idempotent
+     replay (Workflow #931).
+- **Still refused at takeover:** workflow-to-workflow messages
+  (`workflow_messages`) and external payloads.
 - **Shared Server fixtures.** Native, embedded and published PHP 2.11.1 are
   compared on SQLite, MySQL 8.0, MariaDB 10.11 and PostgreSQL 17. Where the
   published reference itself misbehaves, an explicit boundary separates it
-  from native, and the PHP owner has an issue for each case:
-  - `operator-metrics` across PHP's redispatch window (#708);
-  - the published-SDK patch cohort's SQLite poll refusal
-    ([#721](https://github.com/durable-workflow/server/issues/721), #722);
-  - update validation under SQLite write locks
-    ([#726](https://github.com/durable-workflow/server/issues/726), fixed in
-    PHP by #730);
-  - `worker-sdk-persistent-pressure`'s shutdown window
-    ([#728](https://github.com/durable-workflow/server/issues/728));
-  - a retried activity's cooperative cancellation
-    ([Workflow #923](https://github.com/durable-workflow/workflow/issues/923));
-  - a reacquired worker session keeping its old failure reason
-    ([#737](https://github.com/durable-workflow/server/issues/737)).
+  from native, and the PHP owner has an issue for each case. Recent ones:
+  - #796, `service-calls` under full-corpus SQLite load;
+  - #794, workflow-bound Nexus handlers.
 
   No expectation is edited to make native pass.
 - **Parity surface.** Native answers 174 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, service catalog
-  administration, Nexus service-call execution, show and cancel for the
-  handler kinds that stay accepted, caller-indexed Nexus operation reads,
-  bridge adapter webhooks, and embedded v2 and Waterline v1 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
+  published 2.11.1's route table. Eight of those are the prepared
+  local-activity routes, which it answers with PHP's refusal for a bridge
+  without the prepared role. Nexus workflow-bound handler kinds answer 501
+  until #794 settles their contract. An execution test sends each route
+  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
-  the signing key only from the request body
-  ([#765](https://github.com/durable-workflow/server/issues/765)).
-- **Next action.** Land the items above in order, keep porting the unserved
-  routes fixture first, and package the native binary into the server image.
+- **Remaining for the release candidate:**
+  - takeover of workflow messages and external payloads;
+  - the 2.11.3 reference move;
+  - prepared local activities and workflow-bound Nexus handlers (scope
+    question with the operator);
+  - native image packaging and publication, with a cutover runbook;
+  - differential and performance qualification.
 
 ## API surface still missing
 

From 01ec49bf3b4d8b8ae3a109685ffc7dbaa59e8ce2 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 02:00:38 +0000
Subject: [PATCH 4/8] Register DW_RUST_MAINTENANCE_LOOP as a native-only
 setting

The opt-in for native's maintenance loop (#797) was missing from the
config audit's native-only names, so setting it logged "Unknown DW_
variables set in environment". The native image smoke test showed it.
The shipped-environment test now sets it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rust/src/config_audit.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs
index a23e2ada0..1e583efed 100644
--- a/rust/src/config_audit.rs
+++ b/rust/src/config_audit.rs
@@ -502,7 +502,10 @@ pub fn audit(environment: &BTreeMap<String, String>) -> Report {
 
 /// Names native reads that PHP's contract does not list.
 fn native_only(name: &str) -> bool {
-    matches!(name, "DW_BIND_ADDRESS" | "DW_ENV" | "DW_RUST_EXPERIMENTAL" | "DW_UPGRADE_BACKUP_FINGERPRINT")
+    matches!(
+        name,
+        "DW_BIND_ADDRESS" | "DW_ENV" | "DW_RUST_EXPERIMENTAL" | "DW_RUST_MAINTENANCE_LOOP" | "DW_UPGRADE_BACKUP_FINGERPRINT"
+    )
         || ["DW_RUST_UPGRADE_QUEUE_", "DW_SCHEMA_TEST_", "DW_TAKEOVER_TEST_", "DW_TEST_"]
             .iter()
             .any(|prefix| name.starts_with(prefix))
@@ -820,6 +823,7 @@ mod tests {
             ("DW_WORKFLOW_MEMO_MIGRATION_RECOVERY", ""),
             ("DW_ENV_AUDIT_STRICT", "1"),
             ("DW_RUST_EXPERIMENTAL", "1"),
+            ("DW_RUST_MAINTENANCE_LOOP", "1"),
             ("DW_BIND_ADDRESS", "0.0.0.0:8080"),
         ];
         let report = audit(&environment(&shipped));

From 3da8f54fd9b7a97a68ece457fe79197270c6983d Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 02:06:06 +0000
Subject: [PATCH 5/8] Refresh the Rust port's current state after #797

#797 merged; #799, this PR and the native image are next. External payload
takeover stays refused: PHP keeps references in about 30 locations in
their URI form, so it is its own project, and the release-candidate scope
question now includes it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/rust-server-port.md | 28 +++++++++++++++++-----------
 1 file changed, 17 insertions(+), 11 deletions(-)

diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index c3e4a5f37..917de2887 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -26,19 +26,26 @@ Updated October 27, 2026.
     published 2.11.1 databases (fresh and upgraded from 2.9.x), keeping search
     attribute definitions, worker sessions and service calls, plus the repair
     pass's service-call deadline sweep.
+  - [#797](https://github.com/durable-workflow/server/pull/797): native's own
+    maintenance repair loop (opt-in, `DW_RUST_MAINTENANCE_LOOP=1`); takeover
+    keeps runtime credentials and update validation tasks; takeover of
+    published 2.11.3, fresh and upgraded from 2.9.x.
 - **In the queue or next, in order.**
-  1. [#797](https://github.com/durable-workflow/server/pull/797):
-     - native's own maintenance repair loop;
-     - takeover keeps runtime credentials and update validation tasks;
-     - takeover of published 2.11.3, fresh and upgraded from 2.9.x.
+  1. [#799](https://github.com/durable-workflow/server/pull/799): the fixture
+     adapters' Cargo cache, saved only by pushes to main.
   2. Message-stream retention (PHP's `MessageStreamRetentionCleanup`), and
      takeover that keeps message and workflow streams.
-  3. The parity reference moves to published 2.11.3. Native already matches it
+  3. The native release-candidate image (`Dockerfile.native`), its release
+     workflow and the takeover runbook's image commands.
+  4. The parity reference moves to published 2.11.3. Native already matches it
      on 62 of 65 fixtures. The move waits on two published 2.11.3 defects
      General owns: the storage diagnostic (#789) and cross-caller idempotent
      replay (Workflow #931).
 - **Still refused at takeover:** workflow-to-workflow messages
-  (`workflow_messages`) and external payloads.
+  (`workflow_messages`, written only by embedded PHP workflows) and external
+  payload references. PHP keeps references in about 30 table and column
+  locations in their URI form, so taking them over is its own project; the
+  recommendation is that the release candidate keeps refusing them.
 - **Shared Server fixtures.** Native, embedded and published PHP 2.11.1 are
   compared on SQLite, MySQL 8.0, MariaDB 10.11 and PostgreSQL 17. Where the
   published reference itself misbehaves, an explicit boundary separates it
@@ -56,12 +63,11 @@ Updated October 27, 2026.
   `501 rust_capability_not_implemented` must match
   [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt).
 - **Remaining for the release candidate:**
-  - takeover of workflow messages and external payloads;
   - the 2.11.3 reference move;
-  - prepared local activities and workflow-bound Nexus handlers (scope
-    question with the operator);
-  - native image packaging and publication, with a cutover runbook;
-  - differential and performance qualification.
+  - the scope question with the operator: prepared local activities,
+    workflow-bound Nexus handlers and external payload takeover;
+  - publishing the native image (General runs the release);
+  - differential and performance qualification of the published image.
 
 ## API surface still missing
 

From ec23a5ad7851cdf31b57e8aa06ac05ea5e4049cc Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 02:30:49 +0000
Subject: [PATCH 6/8] Tighten stream retention and takeover after review

An independent review of the stream retention and stream takeover
commits found no serious defects. This fixes what it did find:
- A resumed MySQL takeover now picks step 10's deadline source by layout,
  as SQLite and PostgreSQL do, not by whether reclaim_at still exists. A
  2.11.3 takeover killed after that column's drop had copied every
  deadline already, so it adopts nothing more instead of adding
  closed_at + retention rows for streams PHP had emptied.
- The current run counts as retained only when its run row exists, as in
  PHP.
- A run whose instance has an inbound stream item holding an external
  reference stays held (external_payload_cleanup_unavailable). Nothing
  writes one today; this keeps a later change from stranding objects.
- The payload scan's comment no longer says stream item rows are refused.

The retention test now puts the cursor exactly on the pruned run's last
consumed message, and points the instance at a missing current run. Each
new assertion fails against the code it guards (checked by mutation).
SQLite: lib 338, execution 560.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../2026-10-27-message-stream-retention.md    | 22 +++++++++++++++++--
 rust/src/runtime/operator_metrics.rs          | 11 ++++++----
 rust/src/runtime/retention.rs                 |  9 ++++++--
 rust/src/runtime/upgrade/payload_scan.rs      |  6 +++--
 .../runtime/upgrade/takeover/mysql_journal.rs | 16 +++++++++-----
 rust/tests/execution/message_streams.rs       |  9 +++++++-
 6 files changed, 57 insertions(+), 16 deletions(-)

diff --git a/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
index 30b5e83da..aa41777f1 100644
--- a/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
+++ b/docs/rust-server-port-checkpoints/2026-10-27-message-stream-retention.md
@@ -21,8 +21,13 @@
     becomes `''` and `payload_released_at` is set. The item rows stay as
     identity tombstones. Reported as `message_stream_items_compacted`.
     `cleanup_blocked_*` markers naming this run are cleared.
-  - The run's `workflow_messages` rows go with the run's other details, as
-    before.
+  - The run's `workflow_messages` rows go with the run's other details
+    (`messages_deleted`), as in PHP. Before, such a run was held.
+  - The current run counts as retained only when its run row exists, as in
+    PHP.
+  - A run stays held with `external_payload_cleanup_unavailable` when an
+    inbound item on its instance holds an external reference. Nothing writes
+    one today; the hold keeps a later change from stranding stored objects.
 - **Not ported.** PHP's `markBlocked`, which marks a stream when
   external-payload cleanup blocks a prune. Native never deletes external
   objects: a run holding an external reference stays held as
@@ -56,3 +61,16 @@
     reference.
   - Takeover qualification passes on PostgreSQL, MySQL and MariaDB,
     including a MySQL takeover killed after step 9.
+
+## Review fixes before the PR
+
+An independent review found no serious defects. Fixed:
+- A resumed MySQL takeover decides step 10's deadline source by layout, not
+  by whether `reclaim_at` still exists. A 2.11.3 takeover killed after that
+  column's drop had already copied every deadline, so it adopts nothing more
+  instead of adding `closed_at + retention` rows for streams PHP had emptied.
+- The current run's missing row and the inbound-item hold above.
+- The payload scan's comment no longer says stream item rows are refused.
+- The retention test now puts the cursor exactly on the pruned run's last
+  consumed message (so `<=` matters) and points the instance at a missing
+  current run.
diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs
index e39c76c2d..3409aede0 100644
--- a/rust/src/runtime/operator_metrics.rs
+++ b/rust/src/runtime/operator_metrics.rs
@@ -1427,16 +1427,19 @@ where
             .map(|row| DB::optional_string(&row, "current_run_id"))
             .transpose()?
             .flatten();
-        let current_terminal = match &current {
+        // The current run counts only when its row exists, as in PHP: an
+        // instance pointing at a missing run retains nothing through it.
+        let current_status = match &current {
             Some(current) => Self::query("SELECT status FROM workflow_runs WHERE id=$1")
                 .bind(current)
                 .fetch_optional(&mut *tx)
                 .await?
                 .map(|row| DB::string(&row, "status"))
-                .transpose()?
-                .is_some_and(|status| retention::terminal(&status)),
-            None => false,
+                .transpose()?,
+            None => None,
         };
+        let current_terminal = current_status.as_deref().is_some_and(retention::terminal);
+        let current = current.filter(|_| current_status.is_some());
         let mut retained = Vec::new();
         for row in Self::query("SELECT id FROM workflow_run_summaries WHERE namespace=$1 AND workflow_instance_id=$2 AND id<>$3")
             .bind(namespace)
diff --git a/rust/src/runtime/retention.rs b/rust/src/runtime/retention.rs
index bd1844a2c..1db35af65 100644
--- a/rust/src/runtime/retention.rs
+++ b/rust/src/runtime/retention.rs
@@ -65,13 +65,18 @@ macro_rules! reference {
 pub(super) use reference;
 
 /// The run's payload columns that can hold an external reference. History
-/// events repeat these payloads, so they are not scanned again.
+/// events repeat these payloads, so they are not scanned again. Inbound
+/// message stream items on the run's instance count too: retention blanks or
+/// deletes them, and native never deletes the stored object behind one.
+/// Nothing writes such an item today (append and takeover refuse external
+/// references), so this only keeps a later change from stranding objects.
 const REFERENCES_SQL: &str = concat!(
     "SELECT (SELECT COUNT(*) FROM workflow_runs WHERE id=$1 AND (", reference!("arguments"), " OR ", reference!("output"), "))",
     "+(SELECT COUNT(*) FROM workflow_commands WHERE workflow_run_id=$2 AND ", reference!("payload"), ")",
     "+(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$3 AND ", reference!("arguments"), ")",
     "+(SELECT COUNT(*) FROM workflow_updates WHERE workflow_run_id=$4 AND (", reference!("arguments"), " OR ", reference!("result"), "))",
     "+(SELECT COUNT(*) FROM activity_executions WHERE workflow_run_id=$5 AND (", reference!("arguments"), " OR ", reference!("result"), "))",
+    "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE stream_id IN (SELECT id FROM workflow_inbound_streams WHERE workflow_instance_id IN (SELECT workflow_instance_id FROM workflow_runs WHERE id=$6)) AND ", reference!("payload_blob"), ")",
 );
 
 /// The run's durable workflow streams, deleted with it as PHP's
@@ -91,7 +96,7 @@ pub(super) const STREAM_COUNTS: [&str; 3] =
 /// skip reason. An abandoned activity's hold is read from history. Durable
 /// workflow streams go with the run (`DURABLE_STREAMS`) unless they hold an
 /// external reference (`Backend::RUN_STREAM_REFERENCES`).
-pub(super) const HOLDS: &[(&str, usize, &str)] = &[(REFERENCES_SQL, 5, "external_payload_cleanup_unavailable")];
+pub(super) const HOLDS: &[(&str, usize, &str)] = &[(REFERENCES_SQL, 6, "external_payload_cleanup_unavailable")];
 
 /// `MessageStreamRetentionCleanup::planForRun`'s plan for one run: the
 /// instance's inbound streams, whether they all go with the instance's last
diff --git a/rust/src/runtime/upgrade/payload_scan.rs b/rust/src/runtime/upgrade/payload_scan.rs
index 412c2b91e..f3cf64122 100644
--- a/rust/src/runtime/upgrade/payload_scan.rs
+++ b/rust/src/runtime/upgrade/payload_scan.rs
@@ -222,8 +222,10 @@ where
         "references_metadata_supported":references_supported,
         "diagnostics":diagnostics, "diagnostics_omitted":omitted,
         "inline_values_verified":inline_supported, "nested_payloads_verified":false,
-        // These tables also hold codec-tagged payloads; takeover refuses a
-        // source with any of their rows, since native lacks the features.
+        // These tables also hold codec-tagged payloads. Takeover carries
+        // their rows over unchanged without validating each inline value;
+        // a stream row holding an external reference is refused through
+        // `stream_reference_inspection`.
         "other_root_payload_columns_verified":false,
         "unscanned_payload_codec_tables":["workflow_durable_stream_items","workflow_inbound_stream_items","workflow_update_validation_tasks"],
         "inline_validation_limits":{"encoded_frame_bytes":65536,"traversal_depth":64,
diff --git a/rust/src/runtime/upgrade/takeover/mysql_journal.rs b/rust/src/runtime/upgrade/takeover/mysql_journal.rs
index 90c186ef0..e55b82596 100644
--- a/rust/src/runtime/upgrade/takeover/mysql_journal.rs
+++ b/rust/src/runtime/upgrade/takeover/mysql_journal.rs
@@ -390,11 +390,17 @@ async fn perform(connection: &mut MySqlConnection, run: &mut Run<'_>, step: i64)
             }
         }
         10 => {
-            // Durable stream reclaim deadlines, from 2.11.3's column while it
-            // is still there, else from each closed stream's retention.
-            let recorded = column_exists(connection, "workflow_durable_streams", "reclaim_at").await?;
-            let reclaims = super::adopt_stream_reclaims::<MySql>(connection, recorded).await?;
-            run.converted.insert("stream_reclaims_recorded".into(), json!(reclaims));
+            // Durable stream reclaim deadlines: 2.11.3 recorded them in
+            // `reclaim_at`; earlier layouts get each closed stream's
+            // retention. The layout decides, not the column: a resumed
+            // 2.11.3 takeover whose column is already dropped copied every
+            // deadline before the drop (each insert commits), so it adopts
+            // nothing more.
+            let recorded = super::super::has_release(&run.layout, "php-2.11.3");
+            if !recorded || column_exists(connection, "workflow_durable_streams", "reclaim_at").await? {
+                let reclaims = super::adopt_stream_reclaims::<MySql>(connection, recorded).await?;
+                run.converted.insert("stream_reclaims_recorded".into(), json!(reclaims));
+            }
             // PHP 2.11.3's additions. Count before dropping, as step 9 does.
             if column_exists(connection, "workflow_failures", "message_full").await? {
                 let messages = count::<MySql>(connection, super::FULL_FAILURE_MESSAGES).await?;
diff --git a/rust/tests/execution/message_streams.rs b/rust/tests/execution/message_streams.rs
index 2bff785b5..f1fe2b763 100644
--- a/rust/tests/execution/message_streams.rs
+++ b/rust/tests/execution/message_streams.rs
@@ -214,6 +214,9 @@ async fn retention_releases_consumed_messages_and_removes_the_last_runs_streams(
     }
 
     // The first run goes; the second run's checkpoint covers what it consumed.
+    // Put the cursor exactly on the last message the first run consumed: that
+    // message is covered too (PHP compares with <=).
+    database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_position=2".to_owned())).await;
     let (status, pruned) = request(&app, "POST", "/api/system/retention/pass", json!({"run_ids":[first]})).await;
     assert_eq!(status, StatusCode::OK, "{pruned}");
     let deleted = &pruned["results"][0]["deleted"];
@@ -225,6 +228,7 @@ async fn retention_releases_consumed_messages_and_removes_the_last_runs_streams(
     );
     assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE payload_released_at IS NOT NULL AND payload_blob=''".into()).await, 2);
     assert_eq!(rows(&database, "SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE message_id='m3' AND payload_released_at IS NULL AND payload_blob<>''".into()).await, 1);
+    database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_position=3".to_owned())).await;
     let orders = &streams(&app, "streams-kept").await["streams"][0];
     assert_eq!((&orders["last_position"], &orders["cursor_position"]), (&json!(3), &json!(3)), "{orders}");
 
@@ -257,8 +261,11 @@ async fn retention_keeps_consumed_messages_no_retained_checkpoint_covers() {
     let cursor = json!({"message_stream_cursors":[{"stream_name":"orders","through_position":1}]});
     let (status, body) = complete(&app, &task, json!([{"type":"continue_as_new","arguments":envelope(Payload::Array(vec![]))}]), cursor, "1.20").await;
     assert_eq!(status, StatusCode::OK, "{body}");
-    // The checkpoint names a run that is no longer retained.
+    // The checkpoint names a run that is no longer retained. The instance
+    // names it as its current run too, but without a run row it retains
+    // nothing, as in PHP.
     database.execute(sqlx::AssertSqlSafe("UPDATE workflow_inbound_streams SET cursor_checkpoint_run_id='gone'".to_owned())).await;
+    database.execute(sqlx::AssertSqlSafe("UPDATE workflow_instances SET current_run_id='gone'".to_owned())).await;
     for table in ["workflow_runs", "workflow_run_summaries"] {
         database.execute(sqlx::AssertSqlSafe(format!("UPDATE {table} SET closed_at='2020-01-01 00:00:00.000000' WHERE id='{first}'"))).await;
     }

From 63de188037d3d9cfac7db2ebd61f8c97fd27a681 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 03:05:18 +0000
Subject: [PATCH 7/8] Generate the stream reference count from one macro

SonarCloud failed this PR's quality gate on duplication (4.5% of new
lines against a 3% limit): STREAM_REFERENCES repeated the same 6-line
query for PostgreSQL, SQLite and MySQL, differing only in the ::text
casts. One macro now builds it from the payload and metadata column
expressions; SQLite and MySQL share one constant. The SQL each backend
runs is unchanged. SQLite lib: 338 passed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rust/src/runtime/backend.rs | 38 ++++++++++++++++++++++---------------
 1 file changed, 23 insertions(+), 15 deletions(-)

diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
index 761a4dd8b..75340c6a7 100644
--- a/rust/src/runtime/backend.rs
+++ b/rust/src/runtime/backend.rs
@@ -6,6 +6,26 @@ use std::borrow::Cow;
 
 use super::{Result, mysql, postgres, schema};
 
+/// Counts every durable stream item, durable stream and inbound message that
+/// holds an external payload reference. The arguments name the JSON payload
+/// and metadata columns as the backend reads them as text.
+macro_rules! stream_references {
+    ($payload:literal, $metadata:literal) => {
+        concat!(
+            "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ",
+            super::retention::reference!($payload),
+            ")+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ",
+            super::retention::reference!($metadata),
+            ")+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ",
+            super::retention::reference!("payload_blob"),
+            ") AS n"
+        )
+    };
+}
+
+/// `STREAM_REFERENCES` for backends that store JSON as text.
+const PLAIN_STREAM_REFERENCES: &str = stream_references!("payload", "metadata");
+
 pub(super) trait Backend: Database {
     const PHP_CONNECTION: &str;
     type EncodedTime: for<'q> Encode<'q, Self> + Type<Self> + Send + Sync + 'static;
@@ -97,11 +117,7 @@ pub(super) trait Backend: Database {
     // Every stream item, durable stream and inbound message that holds an
     // external payload reference, which takeover refuses as it refuses any
     // other external payload.
-    const STREAM_REFERENCES: &str = concat!(
-        "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), ")",
-        "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", super::retention::reference!("metadata::text"), ")",
-        "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", super::retention::reference!("payload_blob"), ") AS n"
-    );
+    const STREAM_REFERENCES: &str = stream_references!("payload::text", "metadata::text");
     const RUN_STREAM_REFERENCES: &str = concat!(
         "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload::text"), "))",
         "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata::text"), ")"
@@ -225,11 +241,7 @@ ORDER BY t.available_order,t.id LIMIT 100
 "#;
     const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND json_extract(payload,'$.selection_group_id')=$2 ORDER BY sequence LIMIT 1";
     const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")");
-    const STREAM_REFERENCES: &str = concat!(
-        "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")",
-        "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", super::retention::reference!("metadata"), ")",
-        "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", super::retention::reference!("payload_blob"), ") AS n"
-    );
+    const STREAM_REFERENCES: &str = PLAIN_STREAM_REFERENCES;
     const RUN_STREAM_REFERENCES: &str = concat!(
         "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), "))",
         "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata"), ")"
@@ -490,11 +502,7 @@ impl Backend for MySql {
     const STREAM_SWEEP_BOUNDARY: &str = "SELECT `offset` FROM workflow_durable_stream_items WHERE stream_id=$1 ORDER BY `offset` LIMIT 1 OFFSET $2";
     const STREAM_SWEEP_THROUGH: &str = "DELETE FROM workflow_durable_stream_items WHERE stream_id=$1 AND `offset`<=$2";
     const STREAM_ITEM_REFERENCES: &str = concat!("SELECT COUNT(*) FROM workflow_durable_stream_items WHERE stream_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")");
-    const STREAM_REFERENCES: &str = concat!(
-        "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), ")",
-        "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE ", super::retention::reference!("metadata"), ")",
-        "+(SELECT COUNT(*) FROM workflow_inbound_stream_items WHERE ", super::retention::reference!("payload_blob"), ") AS n"
-    );
+    const STREAM_REFERENCES: &str = PLAIN_STREAM_REFERENCES;
     const RUN_STREAM_REFERENCES: &str = concat!(
         "SELECT (SELECT COUNT(*) FROM workflow_durable_stream_items WHERE workflow_run_id=$1 AND (payload_reference IS NOT NULL OR ", super::retention::reference!("payload"), "))",
         "+(SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$2 AND ", super::retention::reference!("metadata"), ")"

From f9610d4292168a4dc4079395a0533a2b09fb2afd Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Tue, 27 Oct 2026 03:32:42 +0000
Subject: [PATCH 8/8] Give the Rust test builds 8 GB

This PR's first full run lost execution-and-codec (native-tests): rustc
was killed with signal 9 compiling the library tests inside the 6 GB
container. A cold cargo test --all-targets --no-run with two jobs peaks
at 6.8 GB (cgroup memory.peak, measured locally on this tree); main
passed only narrowly. The three containers that compile test targets
(the all-targets run and the MySQL and PostgreSQL native test runs) now
get 8 GB. The binary-only build keeps 6 GB. A hosted runner has 16 GB,
which leaves room for the database container beside them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/workflows/server-parity.yml | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml
index 57db9b6a9..33ca9f07b 100644
--- a/.github/workflows/server-parity.yml
+++ b/.github/workflows/server-parity.yml
@@ -523,8 +523,8 @@ jobs:
             -e DB_USERNAME=root -e DB_PASSWORD=parity-disposable-password \
             -e CACHE_STORE=file -e QUEUE_CONNECTION=database -e DW_AUTH_TOKEN=parity-test-token \
             "$layout" artisan server:bootstrap --force > parity-evidence/published-bootstrap.txt
-          # Compiling the library tests with two jobs peaks near 4.2 GB; a 4 GB limit OOM-killed rustc (exit 101).
-          docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=6g --memory-swap=6g \
+          # Compiling the test targets with two jobs peaked at 6.8 GB cold (cgroup memory.peak, October 27); a 6 GB limit OOM-killed rustc (signal 9, PR #801).
+          docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=8g --memory-swap=8g \
             -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target -e CARGO_BUILD_JOBS=2 -e DW_PARITY_PART -e DW_EXECUTION_SHARDS \
             -e DW_TEST_MYSQL_URL=mysql://root:parity-disposable-password@parity-mysql:3306/php_ref \
             -e DW_EXECUTION_MYSQL_URL=mysql://root:parity-disposable-password@parity-mysql:3306/php_ref \
@@ -1183,8 +1183,8 @@ jobs:
             -e DB_USERNAME=parity -e DB_PASSWORD=parity-disposable-password \
             -e CACHE_STORE=file -e QUEUE_CONNECTION=database -e DW_AUTH_TOKEN=parity-test-token \
             "$layout" artisan server:bootstrap --force > parity-evidence/published-bootstrap.txt
-          # Compiling the library tests with two jobs peaks near 4.2 GB; a 4 GB limit OOM-killed rustc (exit 101).
-          docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=6g --memory-swap=6g \
+          # Compiling the test targets with two jobs peaked at 6.8 GB cold (cgroup memory.peak, October 27); a 6 GB limit OOM-killed rustc (signal 9, PR #801).
+          docker run --rm --user="$(id -u):$(id -g)" --network "$RESOURCE_SCOPE" --cpus=2 --memory=8g --memory-swap=8g \
             -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target -e CARGO_BUILD_JOBS=2 -e DW_PARITY_PART \
             -e DW_TEST_POSTGRES_URL=postgres://parity:parity-disposable-password@parity-pg:5432/php_ref \
             -e DW_EXECUTION_POSTGRES_URL=postgres://parity:parity-disposable-password@parity-pg:5432/php_ref \
@@ -1862,8 +1862,8 @@ jobs:
           }
           trap rust_test_failures ERR
           mkdir -p cargo-home rust-target parity-evidence
-          # Compiling the library tests with two jobs peaks near 4.2 GB; a 4 GB limit OOM-killed rustc (exit 101).
-          docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=6g --memory-swap=6g \
+          # Compiling the test targets with two jobs peaked at 6.8 GB cold (cgroup memory.peak, October 27); a 6 GB limit OOM-killed rustc (signal 9, PR #801).
+          docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=8g --memory-swap=8g \
             -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target -e CARGO_BUILD_JOBS=2 \
             -e DW_UPGRADE_TEST_WRITABLE=/evidence -e DW_UPGRADE_TEST_READONLY=/source/parity-evidence \
             -v "$PWD:/source:ro" -v "$PWD/cargo-home:/cargo" -v "$PWD/rust-target:/target" \