Preserve legacy command readback and refuse ambiguous request keys

docs/rust-server-upgrade-takeover.md

@@ -724,6 +724,13 @@ named. These are the ones a client moving from 2.x can observe:
   (workflow#877) behave like native. The parity harness records this as the
   named divergence `request-id-reuse-conflict`
   (`scripts/conformance/server-parity/native-divergences.json`).
+  Pre-upgrade receipts keep their raw keys and have no intent digest. A
+  compatible retry reads back the original command with `retained: true` and
+  `intent_verified: false`; its arguments are not asserted equal. A changed
+  name or explicit target is refused. Ambiguous raw/hash identities, including
+  collation collisions, get nonretryable `409 request_id_ambiguous`, select no
+  receipt and write nothing. Inspect the retained command before deciding to
+  send a new command under a new key.
 - **Open work when a run's own code closes it (native 3.9.0 and later, Server
   #1193).** PHP 2.x completes, fails or continues a run as new and leaves its
   open activities and timers as they are. An unawaited activity, such as a

docs/server-reference.md

@@ -698,7 +698,7 @@ a key is the only command it admits:
   The intent is Core's CommandIntent v1 digest over the command type, name,
   typed arguments and explicit run target, so `1`, `1.0` and `"1"` differ and
   map key order does not matter. The receipt is found before any run gate, so a
-  retry after the run closed or continued as new still gets it. A signal
+  retry after the run closed or continued as new still gets it. A retained
   receipt carries `retained: true` and `intent_verified: true`.
 - The same key with a different name, arguments or run target is refused with
   `409`, `reason: "request_id_conflict"` and `outcome:
@@ -713,8 +713,17 @@ a key is the only command it admits:
   Both live as long as the command does. When retention prunes a run's details,
   its commands' keys and digests go with them, and a later request under that
   key is a new command.
-- An update receipt written before 3.9.0 has no recorded intent. It is read back
-  only when the update name matches, and its arguments are not compared.
+- A receipt written before intent digests were introduced has no recorded
+  intent. An exact raw-key retry reads back the original command only when
+  the name matches and any explicit run target is the original run. It carries
+  `retained: true` and `intent_verified: false`. Arguments are not compared;
+  the readback does not establish that the retry's input equals the original.
+- An ambiguous raw-key or hashed-key match is refused with
+  `409 request_id_ambiguous`, `intent_verified: false` and `retryable: false`,
+  without selecting a receipt or writing anything. This includes a legacy
+  raw key that looks like another key's hash token and database collation
+  collisions. Inspect the original command before using a new key; do not
+  retry an ambiguous key automatically.
 
 Before 3.9.0, a repeated signal `request_id` recorded a second signal, and an
 update retried with changed arguments returned the original receipt.

rust/README.md

@@ -416,7 +416,11 @@ signal or update `request_id` becomes a durable command key on its workflow
 instance: a retry with the same intent returns the original receipt, and a
 different intent is refused with `409 request_id_conflict`. The table records
 each keyed command's CommandIntent v1 digest. The upgrade creates one empty
-table; commands admitted earlier have no digest. Stop the native nodes, back up
+table; commands admitted earlier have no digest. Their compatible retries read
+back the original with `retained: true` and `intent_verified: false`. A changed
+name or explicit run target is refused, and an ambiguous raw/hash identity gets
+nonretryable `409 request_id_ambiguous` without selecting a receipt or writing
+anything. Stop the native nodes, back up
 the database and run `schema-upgrade` before starting this binary on an
 earlier schema; rolling back means restoring the backup.
 

rust/src/runtime/command_intent.rs

@@ -109,6 +109,18 @@ pub(super) fn compare(retained: Option<&str>, candidate: &str) -> Intent {
     }
 }
 
+/// A raw legacy identity and a hashed identity cannot be selected safely.
+/// Refuse without exposing either key or inventing an intent comparison.
+pub(super) fn ambiguous_key(workflow_id: &str) -> super::RuntimeError {
+    super::RuntimeError::Protocol {
+        status: axum::http::StatusCode::CONFLICT,
+        response: serde_json::json!({"workflow_id":workflow_id,"reason":"request_id_ambiguous",
+            "outcome":"rejected_request_id_ambiguous","command_status":"rejected",
+            "rejection_reason":"request_id_ambiguous","retained":true,"intent_verified":false,"retryable":false,
+            "message":"The request_id matches an ambiguous retained command identity. Inspect the original command before using a new request_id."}),
+    }
+}
+
 /// The refusal for a key reused with a different command intent. Nothing is
 /// written; the key keeps its original receipt.
 pub(super) fn conflict(workflow_id: &str, command_type: &str, retained_command_id: &str, retained_outcome: Option<&str>,

rust/src/runtime/mysql.rs

@@ -3154,6 +3154,10 @@ mod tests {
             let from = command_intents_step(Flavor::detect(&mut connection).await.unwrap());
             connection.close().await.unwrap();
             let flavor = retained_version_database(&db, from).await;
+            let mut connection = db.connection().await;
+            sqlx::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,command_type,source,status,request_id) VALUES ('legacy-signal','legacy-workflow','legacy-run','signal','control_plane','accepted','RawSignalKey'),('legacy-update','legacy-workflow','legacy-run','update','control_plane','accepted','RawUpdateKey')")
+                .execute(&mut connection).await.unwrap();
+            connection.close().await.unwrap();
             if stage == "before" {
                 refused(MySqlStorage::open(db.options.clone()).await, "native_schema_upgrade_required");
             } else {
@@ -3168,6 +3172,11 @@ mod tests {
             let mut connection = db.connection().await;
             let kept: String = sqlx::query_scalar("SELECT description FROM workflow_namespaces WHERE name='kept'").fetch_one(&mut connection).await.unwrap();
             assert_eq!(kept, format!("seeded at version {from}"));
+            let keys: Vec<(String, String, String)> = sqlx::query_as("SELECT command_type,request_id,workflow_run_id FROM workflow_commands WHERE workflow_instance_id='legacy-workflow' ORDER BY command_type")
+                .fetch_all(&mut connection).await.unwrap();
+            assert_eq!(keys, vec![("signal".into(),"RawSignalKey".into(),"legacy-run".into()),("update".into(),"RawUpdateKey".into(),"legacy-run".into())]);
+            let intents: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM dw_command_intents").fetch_one(&mut connection).await.unwrap();
+            assert_eq!(intents, 0);
             assert_eq!(catalog(&mut connection).await.unwrap(), flavor.expected().unwrap());
             connection.execute("ALTER TABLE dw_command_intents ADD COLUMN extra TEXT NULL").await.unwrap();
             connection.close().await.unwrap();

rust/src/runtime/postgres.rs

@@ -1996,12 +1996,19 @@ mod tests {
     async fn qualification_version_13_command_intents_upgrade_preserves_rows() {
         let db = Database::new().await;
         version_database(&db, 13).await;
+        let mut connection = db.connection().await;
+        sqlx::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,command_type,source,status,request_id) VALUES ('legacy-signal','legacy-workflow','legacy-run','signal','control_plane','accepted','RawSignalKey'),('legacy-update','legacy-workflow','legacy-run','update','control_plane','accepted','RawUpdateKey')")
+            .execute(&mut connection).await.unwrap();
+        connection.close().await.unwrap();
         refused(PostgresStorage::open(db.options.clone()).await, "native_schema_upgrade_required");
         assert_eq!(PostgresStorage::upgrade(db.options.clone()).await.unwrap(), 13);
         PostgresStorage::check(db.options.clone()).await.unwrap();
         let mut connection = db.connection().await;
         let kept: String = sqlx::query_scalar("SELECT description FROM workflow_namespaces WHERE name='kept'").fetch_one(&mut connection).await.unwrap();
         assert_eq!(kept, "seeded at version 13");
+        let keys: Vec<(String, String, String)> = sqlx::query_as("SELECT command_type,request_id,workflow_run_id FROM workflow_commands WHERE workflow_instance_id='legacy-workflow' ORDER BY command_type")
+            .fetch_all(&mut connection).await.unwrap();
+        assert_eq!(keys, vec![("signal".into(),"RawSignalKey".into(),"legacy-run".into()),("update".into(),"RawUpdateKey".into(),"legacy-run".into())]);
         let intents: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM dw_command_intents").fetch_one(&mut connection).await.unwrap();
         assert_eq!(intents, 0);
         sqlx::query("ALTER TABLE dw_command_intents ADD COLUMN extra text").execute(&mut connection).await.unwrap();

rust/src/runtime/schema.rs

@@ -832,13 +832,18 @@ mod tests {
         let mut connection = SqliteConnection::connect_with(&SqliteConnectOptions::new().filename(&path).create_if_missing(true)).await.unwrap();
         Migrator::with_migrations(vec![earlier_migration(12).unwrap().clone()]).run(&mut connection).await.unwrap();
         sqlx::query("INSERT INTO dw_query_cache VALUES ('kept','original',1)").execute(&mut connection).await.unwrap();
+        sqlx::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,command_type,source,status,request_id) VALUES ('legacy-signal','legacy-workflow','legacy-run','signal','control_plane','accepted','RawSignalKey'),('legacy-update','legacy-workflow','legacy-run','update','control_plane','accepted','RawUpdateKey')")
+            .execute(&mut connection).await.unwrap();
         connection.close().await.unwrap();
         assert_eq!(reason(super::super::sqlite::open(path.to_str().unwrap()).await).await, "native_schema_upgrade_required");
         let report = super::super::schema_upgrade::upgrade_sqlite(path.to_str().unwrap()).await.unwrap();
         assert_eq!(report["from_schema_version"], 12);
         let pool = super::super::sqlite::open(path.to_str().unwrap()).await.unwrap();
         let kept: String = sqlx::query_scalar("SELECT value FROM dw_query_cache WHERE key='kept'").fetch_one(&pool).await.unwrap();
         assert_eq!(kept, "original");
+        let keys: Vec<(String, String, String)> = sqlx::query_as("SELECT command_type,request_id,workflow_run_id FROM workflow_commands WHERE workflow_instance_id='legacy-workflow' ORDER BY command_type")
+            .fetch_all(&pool).await.unwrap();
+        assert_eq!(keys, vec![("signal".into(),"RawSignalKey".into(),"legacy-run".into()),("update".into(),"RawUpdateKey".into(),"legacy-run".into())]);
         let intents: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM dw_command_intents").fetch_one(&pool).await.unwrap();
         assert_eq!(intents, 0);
         sqlx::query("ALTER TABLE dw_command_intents ADD COLUMN extra TEXT").execute(&pool).await.unwrap();

rust/src/runtime/store.rs

@@ -1329,19 +1329,27 @@ where
         key: &str,
     ) -> Result<Option<(DB::Row, Option<String>)>> {
         let token = super::command_intent::key_token(command_type, key);
-        let Some(row) = Self::query("SELECT * FROM workflow_commands WHERE workflow_instance_id=$1 AND command_type=$2 AND (request_id=$3 OR request_id=$4) ORDER BY CASE WHEN request_id=$5 THEN 0 ELSE 1 END LIMIT 1")
-            .bind(workflow_id).bind(command_type).bind(&token).bind(key).bind(&token)
-            .fetch_optional(&mut **tx).await?
-        else {
-            return Ok(None);
-        };
-        // A case-insensitive collation can match another raw key.
-        if DB::optional_string(&row, "request_id")?.is_none_or(|stored| stored != token && stored != key) {
-            return Ok(None);
-        }
+        // The unique request lookup index bounds this to at most two rows.
+        // A raw legacy key can itself look like another key's token. Never
+        // choose one representation, or another key under a DB collation.
+        let mut rows = Self::query("SELECT * FROM workflow_commands WHERE workflow_instance_id=$1 AND command_type=$2 AND (request_id=$3 OR request_id=$4)")
+            .bind(workflow_id).bind(command_type).bind(&token).bind(key)
+            .fetch_all(&mut **tx).await?;
+        if rows.len() > 1 {
+            return Err(super::command_intent::ambiguous_key(workflow_id));
+        }
+        let Some(row) = rows.pop() else { return Ok(None) };
+        let stored = DB::optional_string(&row, "request_id")?;
         let digest = Self::query("SELECT intent_digest FROM dw_command_intents WHERE command_id=$1")
             .bind(DB::string(&row, "id")?).fetch_optional(&mut **tx).await?
             .map(|intent| DB::string(&intent, "intent_digest")).transpose()?;
+        // A digest identifies a token-era row. Without one only the exact
+        // raw key can identify a legacy receipt; a token-shaped raw key is
+        // otherwise indistinguishable from an incomplete modern receipt.
+        let expected = if digest.is_some() { token.as_str() } else { key };
+        if stored.as_deref() != Some(expected) {
+            return Err(super::command_intent::ambiguous_key(workflow_id));
+        }
         Ok(Some((row, digest)))
     }
 
@@ -1372,7 +1380,14 @@ where
         let signal = Self::query("SELECT signal_name FROM workflow_signal_records WHERE workflow_command_id=$1")
             .bind(&command_id).fetch_optional(&mut **tx).await?;
         let retained_name = signal.as_ref().map(|row| DB::string(row, "signal_name")).transpose()?;
+        let run = DB::string(command, "workflow_run_id")?;
         let intent = super::command_intent::compare(retained, digest);
+        if intent == super::command_intent::Intent::Unverified
+            && run_id.is_some_and(|requested| requested != run)
+        {
+            return Err(super::command_intent::conflict(workflow_id, "signal", &command_id,
+                DB::optional_string(command, "outcome")?.as_deref(), vec!["target"]));
+        }
         if intent == super::command_intent::Intent::Different
             || intent == super::command_intent::Intent::Unverified && retained_name.as_deref() != Some(name)
         {
@@ -1386,7 +1401,6 @@ where
             return Err(super::command_intent::conflict(workflow_id, "signal", &command_id,
                 DB::optional_string(command, "outcome")?.as_deref(), parts));
         }
-        let run = DB::string(command, "workflow_run_id")?;
         Ok(json!({"workflow_id":workflow_id,"run_id":run,"resolved_run_id":run,"requested_run_id":null,"signal_name":retained_name.unwrap_or_else(|| name.to_owned()),
             "command_id":command_id,"command_sequence":DB::optional_number(command, "command_sequence")?,"workflow_type":DB::optional_string(command, "workflow_type")?,
             "command_status":DB::string(command, "status")?,"command_source":"control_plane","target_scope":"instance",

rust/src/runtime/updates.rs

@@ -402,11 +402,6 @@ where
             .begin()
             .await
             .map_err(|error| at_update_lookup(error, UpdateLookup::AdmissionOwnership))?;
-        // The referenced object now belongs to the update; a refusal rolls
-        // the claim back with everything else.
-        if let Some(prepared) = external {
-            self.claim_external_in(&mut tx, &prepared.reference, &prepared.uri).await?;
-        }
         let instance = Self::query("SELECT current_run_id,last_message_sequence FROM workflow_instances WHERE id=$1 AND namespace=$2")
             .bind(workflow_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?
             .ok_or_else(|| refuse(StatusCode::NOT_FOUND, "instance_not_found"))?;
@@ -418,13 +413,26 @@ where
             Some(key) => Self::retained_command(&mut tx, workflow_id, "update", key).await?,
             None => None,
         };
+        // A retained answer must not acquire the retry's payload reference.
+        if keyed.is_none()
+            && let Some(prepared) = external
+        {
+            self.claim_external_in(&mut tx, &prepared.reference, &prepared.uri).await?;
+        }
         if let (Some((command, retained)), Some(digest)) = (&keyed, digest.as_deref())
             && super::command_intent::compare(retained.as_deref(), digest) == super::command_intent::Intent::Different
         {
             return Err(super::command_intent::conflict(workflow_id, "update", &DB::string(command, "id")?,
                 DB::optional_string(command, "outcome")?.as_deref(),
                 vec![if selected_run.is_some() { "target_or_arguments" } else { "arguments" }]));
         }
+        if let Some((command, None)) = &keyed
+            && let Some(selected) = selected_run
+            && selected != DB::string(command, "workflow_run_id")?
+        {
+            return Err(super::command_intent::conflict(workflow_id, "update", &DB::string(command, "id")?,
+                DB::optional_string(command, "outcome")?.as_deref(), vec!["target"]));
+        }
         if keyed.is_none() && selected_run.is_some_and(|selected| selected != run_id) {
             return Err(refuse(
                 StatusCode::CONFLICT,
@@ -761,7 +769,12 @@ where
         // A completion wait owns neither the transaction nor a pool connection.
         tx.commit().await?;
         self.wake.notify_waiters();
-        self.wait_update(&update_id, wait_for, timeout, codec).await
+        let (status, mut response) = self.wait_update(&update_id, wait_for, timeout, codec).await?;
+        if let Some((_, digest)) = keyed {
+            response["retained"] = json!(true);
+            response["intent_verified"] = json!(digest.is_some());
+        }
+        Ok((status, response))
     }
 
     /// Reads a retained update by id without admitting anything (#1131).