From 4ea2cd5b17272f31ed9c31d5c80c12853c7f90c4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:15:06 +0000 Subject: [PATCH 1/6] Preserve unverified legacy command receipts and refuse identity ambiguity (#934) --- rust/src/runtime/command_intent.rs | 12 +++ rust/src/runtime/store.rs | 36 ++++--- rust/src/runtime/updates.rs | 25 +++-- rust/tests/execution/command_keys.rs | 134 +++++++++++++++++++++++++++ 4 files changed, 190 insertions(+), 17 deletions(-) diff --git a/rust/src/runtime/command_intent.rs b/rust/src/runtime/command_intent.rs index 8b4c438ac..0d360e94e 100644 --- a/rust/src/runtime/command_intent.rs +++ b/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>, diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 1b09a6a07..edb2c4e8a 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1329,19 +1329,27 @@ where key: &str, ) -> Result)>> { 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", diff --git a/rust/src/runtime/updates.rs b/rust/src/runtime/updates.rs index 981761d02..658edb4e4 100644 --- a/rust/src/runtime/updates.rs +++ b/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,6 +413,12 @@ 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 { @@ -425,6 +426,13 @@ where 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). diff --git a/rust/tests/execution/command_keys.rs b/rust/tests/execution/command_keys.rs index cbd20832c..eec7c5887 100644 --- a/rust/tests/execution/command_keys.rs +++ b/rust/tests/execution/command_keys.rs @@ -179,3 +179,137 @@ async fn retention_prunes_request_key_digests_with_their_commands() { const KEPT: &str = "SELECT COUNT(*) FROM dw_command_intents i JOIN workflow_commands c ON c.id=i.command_id WHERE c.workflow_instance_id='kept-keys'"; assert_eq!(task_retry::rows(&database, KEPT).await, 1, "the other run keeps its digest"); } + +/// The pre-3.9 writer stored a raw request_id and no intent row. Preserve +/// that exact shape without rewriting the original command or its history. +async fn legacy_key(database: &TestDatabase, command: &Value, key: &str) { + let id = command["command_id"].as_str().unwrap(); + assert!(id.bytes().all(|byte| byte.is_ascii_alphanumeric())); + assert!(key.bytes().all(|byte| byte.is_ascii_alphanumeric() || b"-:_".contains(&byte))); + database.execute(sqlx::AssertSqlSafe(format!( + "DELETE FROM dw_command_intents WHERE command_id='{id}'"))).await; + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_commands SET request_id='{key}' WHERE id='{id}'"))).await; +} + +#[tokio::test] +async fn legacy_updates_read_back_unverified_intent_after_restart_close_and_continuation() { + for close in [None, Some("complete_workflow"), Some("continue_as_new")] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + update_worker(&app, "legacy-worker").await; + let started = start(&app, "legacy-update").await; + open_update_wait(&app, "legacy-worker").await; + let path = "/api/workflows/legacy-update/update/set"; + let original = request(&app, "POST", path, input(1, Some("legacy-update-key"))).await; + assert_eq!(original.0, StatusCode::ACCEPTED, "{}", original.1); + legacy_key(&database, &original.1, "legacy-update-key").await; + if let Some(close) = close { + let task = poll(&app, "legacy-worker", "workflow").await; + let done = finish_task(&app, &task, json!([ + {"type":"complete_update","update_id":original.1["update_id"],"result":envelope(Payload::Null)}, + {"type":close,"result":envelope(Payload::Null)}])).await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + } + runtime.close().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let before = table_counts(&database).await; + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + // Unknown old arguments cannot establish equality. Read back the + // original with that fact explicit, even for concurrent retries. + let replies = futures_util::future::join_all((0..4).map(|_| { + request(&app, "POST", path, input(999, Some("legacy-update-key"))) + })).await; + for (status, reply) in replies { + assert!(status.is_success(), "{reply}"); + for field in ["command_id", "update_id", "run_id", "accepted_at"] { + assert_eq!(reply[field], original.1[field], "{field}: {reply}"); + } + assert_eq!((reply["retained"].clone(), reply["intent_verified"].clone()), + (json!(true), json!(false)), "{reply}"); + if close.is_some() { + assert_eq!(reply["update_status"], "completed", "{reply}"); + assert_eq!(reply["result"], Value::Null, "completed null remains an outcome"); + } + } + let changed_target = "/api/workflows/legacy-update/runs/00000000000000000000000000/update/set"; + assert_conflict(&request(&app, "POST", changed_target, input(1, Some("legacy-update-key"))).await, + &original.1, "target"); + let changed_name = path.replace("/set", "/other"); + assert_conflict(&request(&app, "POST", &changed_name, input(1, Some("legacy-update-key"))).await, + &original.1, "name"); + assert_eq!(table_counts(&database).await, before, "all retries and refusals write nothing"); + assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn legacy_signals_do_not_honor_a_changed_target_or_fabricate_intent_equality() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + update_worker(&app, "legacy-worker").await; + let started = start(&app, "legacy-signal").await; + let original = signal(&app, "legacy-signal", "finish", input(1, Some("legacy-signal-key"))).await; + assert_eq!(original.0, StatusCode::ACCEPTED, "{}", original.1); + legacy_key(&database, &original.1, "legacy-signal-key").await; + let task = poll(&app, "legacy-worker", "workflow").await; + let done = finish_task(&app, &task, json!([{"type":"continue_as_new"}])).await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + let before = table_counts(&database).await; + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let retry = signal(&app, "legacy-signal", "finish", input(999, Some("legacy-signal-key"))).await; + assert_eq!(retry.0, StatusCode::ACCEPTED, "{}", retry.1); + assert_eq!(retry.1["command_id"], original.1["command_id"]); + assert_eq!(retry.1["run_id"], original.1["run_id"]); + assert_eq!((retry.1["retained"].clone(), retry.1["intent_verified"].clone()), (json!(true), json!(false))); + let target = "/api/workflows/legacy-signal/runs/00000000000000000000000000/signal/finish"; + assert_conflict(&request(&app, "POST", target, input(1, Some("legacy-signal-key"))).await, &original.1, "target"); + assert_eq!(table_counts(&database).await, before); + assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn raw_and_token_identity_collisions_are_refused_without_selecting_a_receipt() { + use sha2::{Digest, Sha256}; + let token = format!("signal-v1:{:x}", Sha256::digest(b"token-key")); + for shape in ["two-rows", "legacy-token", "modern-raw"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + update_worker(&app, "collision-worker").await; + let started = start(&app, "key-collision").await; + let original = signal(&app, "key-collision", "finish", input(1, Some("token-key"))).await; + assert_eq!(original.0, StatusCode::ACCEPTED, "{}", original.1); + let key = if shape == "modern-raw" { + token.as_str() + } else if shape == "legacy-token" { + // A legacy caller deliberately used the bytes of another key's + // modern token; those bytes are not proof of the modern key. + legacy_key(&database, &original.1, &token).await; + "token-key" + } else { + let legacy = signal(&app, "key-collision", "finish", input(2, Some("second-key"))).await; + assert_eq!(legacy.0, StatusCode::ACCEPTED, "{}", legacy.1); + legacy_key(&database, &legacy.1, "token-key").await; + "token-key" + }; + let before = table_counts(&database).await; + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let refused = signal(&app, "key-collision", "finish", input(1, Some(key))).await; + assert_eq!(refused.0, StatusCode::CONFLICT, "{shape}: {}", refused.1); + assert_eq!(refused.1["reason"], "request_id_ambiguous", "{shape}: {}", refused.1); + assert_eq!(refused.1["intent_verified"], false); + assert!(refused.1["command_id"].is_null(), "no candidate is selected: {}", refused.1); + assert_eq!(table_counts(&database).await, before); + assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); + runtime.close().await; + database.remove().await; + } +} From 4c02773582cd59192dd002b380a6d41b4fb3ec73 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:22:13 +0000 Subject: [PATCH 2/6] Document legacy receipt readback and cover both command kinds and key casing --- docs/rust-server-upgrade-takeover.md | 7 ++ docs/server-reference.md | 15 ++++- rust/README.md | 6 +- rust/tests/execution/command_keys.rs | 99 +++++++++++++++++++++------- 4 files changed, 98 insertions(+), 29 deletions(-) diff --git a/docs/rust-server-upgrade-takeover.md b/docs/rust-server-upgrade-takeover.md index 85c85a6cf..6293bab92 100644 --- a/docs/rust-server-upgrade-takeover.md +++ b/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 diff --git a/docs/server-reference.md b/docs/server-reference.md index 4a6be5b53..d9952aa6d 100644 --- a/docs/server-reference.md +++ b/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. diff --git a/rust/README.md b/rust/README.md index 066c88f19..095bb8e66 100644 --- a/rust/README.md +++ b/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. diff --git a/rust/tests/execution/command_keys.rs b/rust/tests/execution/command_keys.rs index eec7c5887..9e3fb155d 100644 --- a/rust/tests/execution/command_keys.rs +++ b/rust/tests/execution/command_keys.rs @@ -234,6 +234,11 @@ async fn legacy_updates_read_back_unverified_intent_after_restart_close_and_cont assert_eq!(reply["result"], Value::Null, "completed null remains an outcome"); } } + let original_target = format!("/api/workflows/legacy-update/runs/{}/update/set", started["run_id"].as_str().unwrap()); + let selected = request(&app, "POST", &original_target, input(999, Some("legacy-update-key"))).await; + assert!(selected.0.is_success(), "{}", selected.1); + assert_eq!(selected.1["update_id"], original.1["update_id"]); + assert_eq!(selected.1["intent_verified"], false); let changed_target = "/api/workflows/legacy-update/runs/00000000000000000000000000/update/set"; assert_conflict(&request(&app, "POST", changed_target, input(1, Some("legacy-update-key"))).await, &original.1, "target"); @@ -278,37 +283,81 @@ async fn legacy_signals_do_not_honor_a_changed_target_or_fabricate_intent_equali #[tokio::test] async fn raw_and_token_identity_collisions_are_refused_without_selecting_a_receipt() { use sha2::{Digest, Sha256}; - let token = format!("signal-v1:{:x}", Sha256::digest(b"token-key")); - for shape in ["two-rows", "legacy-token", "modern-raw"] { + let hash: String = Sha256::digest(b"token-key").iter().map(|byte| format!("{byte:02x}")).collect(); + for (kind, action) in [("signal", "signal/finish"), ("update", "update/set")] { + let token = format!("{kind}-v1:{hash}"); + for shape in ["two-rows", "legacy-token", "modern-raw"] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + update_worker(&app, "collision-worker").await; + let started = start(&app, "key-collision").await; + if kind == "update" { + open_update_wait(&app, "collision-worker").await; + } + let path = format!("/api/workflows/key-collision/{action}"); + let original = request(&app, "POST", &path, input(1, Some("token-key"))).await; + assert_eq!(original.0, StatusCode::ACCEPTED, "{}", original.1); + let key = if shape == "modern-raw" { + token.as_str() + } else if shape == "legacy-token" { + // A legacy caller deliberately used the bytes of another key's + // modern token; those bytes are not proof of the modern key. + legacy_key(&database, &original.1, &token).await; + "token-key" + } else { + let legacy = request(&app, "POST", &path, input(2, Some("second-key"))).await; + assert_eq!(legacy.0, StatusCode::ACCEPTED, "{}", legacy.1); + legacy_key(&database, &legacy.1, "token-key").await; + "token-key" + }; + let before = table_counts(&database).await; + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let refused = request(&app, "POST", &path, input(1, Some(key))).await; + assert_eq!(refused.0, StatusCode::CONFLICT, "{shape}: {}", refused.1); + assert_eq!(refused.1["reason"], "request_id_ambiguous", "{shape}: {}", refused.1); + assert_eq!(refused.1["intent_verified"], false); + assert!(refused.1["command_id"].is_null(), "no candidate is selected: {}", refused.1); + assert_eq!(table_counts(&database).await, before); + assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); + runtime.close().await; + database.remove().await; + } + } +} + +#[tokio::test] +async fn legacy_key_case_collisions_never_return_another_keys_receipt() { + for action in ["signal/finish", "update/set"] { let database = TestDatabase::new().await; let runtime = database.open().await.unwrap(); let app = router(runtime.clone()); - update_worker(&app, "collision-worker").await; - let started = start(&app, "key-collision").await; - let original = signal(&app, "key-collision", "finish", input(1, Some("token-key"))).await; + update_worker(&app, "case-worker").await; + let started = start(&app, "case-keys").await; + if action == "update/set" { + open_update_wait(&app, "case-worker").await; + } + let path = format!("/api/workflows/case-keys/{action}"); + let original = request(&app, "POST", &path, input(1, Some("CaseSensitive"))).await; assert_eq!(original.0, StatusCode::ACCEPTED, "{}", original.1); - let key = if shape == "modern-raw" { - token.as_str() - } else if shape == "legacy-token" { - // A legacy caller deliberately used the bytes of another key's - // modern token; those bytes are not proof of the modern key. - legacy_key(&database, &original.1, &token).await; - "token-key" - } else { - let legacy = signal(&app, "key-collision", "finish", input(2, Some("second-key"))).await; - assert_eq!(legacy.0, StatusCode::ACCEPTED, "{}", legacy.1); - legacy_key(&database, &legacy.1, "token-key").await; - "token-key" - }; + legacy_key(&database, &original.1, "CaseSensitive").await; let before = table_counts(&database).await; let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; - let refused = signal(&app, "key-collision", "finish", input(1, Some(key))).await; - assert_eq!(refused.0, StatusCode::CONFLICT, "{shape}: {}", refused.1); - assert_eq!(refused.1["reason"], "request_id_ambiguous", "{shape}: {}", refused.1); - assert_eq!(refused.1["intent_verified"], false); - assert!(refused.1["command_id"].is_null(), "no candidate is selected: {}", refused.1); - assert_eq!(table_counts(&database).await, before); - assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); + let exact = request(&app, "POST", &path, input(99, Some("CaseSensitive"))).await; + assert_eq!(exact.1["command_id"], original.1["command_id"]); + assert_eq!(exact.1["intent_verified"], false); + let other = request(&app, "POST", &path, input(1, Some("casesensitive"))).await; + if matches!(&database, TestDatabase::MySql { .. }) { + // The default MySQL/MariaDB collation sees a raw-key match. + // Refuse safely instead of reporting equality for other bytes. + assert_eq!(other.0, StatusCode::CONFLICT, "{}", other.1); + assert_eq!(other.1["reason"], "request_id_ambiguous"); + assert_eq!(table_counts(&database).await, before); + assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); + } else { + assert_eq!(other.0, StatusCode::ACCEPTED, "{}", other.1); + assert_ne!(other.1["command_id"], original.1["command_id"]); + } runtime.close().await; database.remove().await; } From 3f1cd022f57dedb6fd4f91f0038d7ac23aade19b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:32:48 +0000 Subject: [PATCH 3/6] Prove legacy key upgrades and retained retry payload ownership (#934) --- rust/src/runtime/mysql.rs | 9 +++ rust/src/runtime/postgres.rs | 7 ++ rust/src/runtime/schema.rs | 5 ++ rust/tests/execution/command_keys.rs | 2 + .../execution/external_payloads/commands.rs | 70 +++++++++++++++++++ 5 files changed, 93 insertions(+) diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 2f1d3c0af..da2f6ecea 100644 --- a/rust/src/runtime/mysql.rs +++ b/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(); diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 095d3fd63..d9ed967c4 100644 --- a/rust/src/runtime/postgres.rs +++ b/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(); diff --git a/rust/src/runtime/schema.rs b/rust/src/runtime/schema.rs index 52e004fe4..78f4fc9f8 100644 --- a/rust/src/runtime/schema.rs +++ b/rust/src/runtime/schema.rs @@ -832,6 +832,8 @@ 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(); @@ -839,6 +841,9 @@ mod tests { 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(); diff --git a/rust/tests/execution/command_keys.rs b/rust/tests/execution/command_keys.rs index 9e3fb155d..49b3ae679 100644 --- a/rust/tests/execution/command_keys.rs +++ b/rust/tests/execution/command_keys.rs @@ -317,6 +317,7 @@ async fn raw_and_token_identity_collisions_are_refused_without_selecting_a_recei assert_eq!(refused.0, StatusCode::CONFLICT, "{shape}: {}", refused.1); assert_eq!(refused.1["reason"], "request_id_ambiguous", "{shape}: {}", refused.1); assert_eq!(refused.1["intent_verified"], false); + assert_eq!(refused.1["retryable"], false, "{shape}: {}", refused.1); assert!(refused.1["command_id"].is_null(), "no candidate is selected: {}", refused.1); assert_eq!(table_counts(&database).await, before); assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); @@ -352,6 +353,7 @@ async fn legacy_key_case_collisions_never_return_another_keys_receipt() { // Refuse safely instead of reporting equality for other bytes. assert_eq!(other.0, StatusCode::CONFLICT, "{}", other.1); assert_eq!(other.1["reason"], "request_id_ambiguous"); + assert_eq!(other.1["retryable"], false); assert_eq!(table_counts(&database).await, before); assert_eq!(run_history(&app, &started["workflow_id"], &started["run_id"]).await, history); } else { diff --git a/rust/tests/execution/external_payloads/commands.rs b/rust/tests/execution/external_payloads/commands.rs index d89a04455..b0c8b96ab 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/rust/tests/execution/external_payloads/commands.rs @@ -48,6 +48,76 @@ fn typed(body: &Value, status: u16, reason: &str, message: &str, workflow_id: &s assert_eq!(body["workflow_id"], workflow_id, "{body}"); } +/// Compare complete stored rows, not just counts: retaining the retry input +/// would update an existing registry row without adding a command or event. +async fn admission_snapshot(database: &TestDatabase) -> Vec<(String, Vec<(String, String)>)> { + let mut snapshot = Vec::new(); + for (table, key) in [("workflow_commands", "id"), ("dw_command_intents", "command_id"), + ("workflow_history_events", "id"), ("workflow_tasks", "id"), ("workflow_updates", "id"), + ("workflow_runs", "id"), ("workflow_instances", "id"), ("runtime_external_payloads", "id")] { + let columns = text_rows(database, + &format!("SELECT name,name FROM pragma_table_info('{table}') ORDER BY cid"), + &format!("SELECT column_name::text,column_name::text FROM information_schema.columns WHERE table_schema='public' AND table_name='{table}' ORDER BY ordinal_position"), + &format!("SELECT CAST(column_name AS CHAR),CAST(column_name AS CHAR) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='{table}' ORDER BY ordinal_position")).await; + assert!(!columns.is_empty(), "missing snapshot table {table}"); + let mut sqlite = Vec::new(); + let mut mysql = Vec::new(); + for (column, _) in columns { + assert!(column.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')); + sqlite.push(format!("'{column}',\"{column}\"")); + mysql.push(format!("'{column}',`{column}`")); + } + let rows = text_rows(database, + &format!("SELECT \"{key}\",json_object({}) FROM \"{table}\" ORDER BY \"{key}\"", sqlite.join(",")), + &format!("SELECT {key}::text,to_jsonb(t)::text FROM {table} t ORDER BY {key}"), + &format!("SELECT CAST(`{key}` AS CHAR),CAST(JSON_OBJECT({}) AS CHAR) FROM `{table}` ORDER BY `{key}`", mysql.join(","))).await; + snapshot.push((table.to_owned(), rows)); + } + snapshot +} + +#[tokio::test] +async fn retained_update_retries_never_claim_a_distinct_external_input_reference() { + for legacy in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let (database, runtime, app) = setup(ExternalPayloadConfig::default(), &temp.path().join("objects")).await; + waiting_run(&app, "retry-reference", json!({"worker_id":"retry-worker","task_queue":"retry-reference", + "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":[],"capabilities":["workflow_updates"], + "workflow_command_contracts":{"echo":{"updates":["u"],"update_validators":[],"update_contracts":[{"name":"u","parameters":string_parameter()}]}}})).await; + let blob = argument("same typed value"); + let original_input = upload(&app, &blob).await; + let path = "/api/workflows/retry-reference/update/u"; + let (status, original) = json_request(&app, "POST", path, "operator-token", + json!({"request_id":"external-retry-key","wait_for":"accepted", + "input":{"codec":"avro","external_payload":original_input}})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{original}"); + assert_eq!(unclaimed(&database).await, 0); + if legacy { + let command = original["command_id"].as_str().unwrap(); + assert!(command.bytes().all(|byte| byte.is_ascii_alphanumeric())); + database.execute(sqlx::AssertSqlSafe(format!("DELETE FROM dw_command_intents WHERE command_id='{command}'"))).await; + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_commands SET request_id='external-retry-key' WHERE id='{command}'"))).await; + } + let retry_input = upload(&app, &blob).await; + assert_ne!(retry_input["reference_id"], original_input["reference_id"]); + assert_eq!(unclaimed(&database).await, 1); + let before = admission_snapshot(&database).await; + let (status, mut repeated) = json_request(&app, "POST", path, "operator-token", + json!({"request_id":"external-retry-key","wait_for":"accepted", + "input":{"codec":"avro","external_payload":retry_input}})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{repeated}"); + assert_eq!(repeated["retained"], true); + assert_eq!(repeated["intent_verified"], !legacy); + repeated.as_object_mut().unwrap().remove("retained"); + repeated.as_object_mut().unwrap().remove("intent_verified"); + assert_eq!(repeated, original, "the original receipt is unchanged"); + assert_eq!(unclaimed(&database).await, 1, "the retry input is still owned by its uploader"); + assert_eq!(admission_snapshot(&database).await, before, "a retained retry changes no stored row"); + runtime.close().await; + database.remove().await; + } +} + #[tokio::test] async fn signal_update_and_query_input_resolve_runtime_references_like_php() { let temp = tempfile::tempdir().unwrap(); From 2481c5007acb317be6328d98444c3a53e92784b6 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:37:28 +0000 Subject: [PATCH 4/6] Use distinct carriers for equal-intent external retry evidence --- rust/tests/execution/external_payloads/commands.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rust/tests/execution/external_payloads/commands.rs b/rust/tests/execution/external_payloads/commands.rs index b0c8b96ab..f84d5026e 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/rust/tests/execution/external_payloads/commands.rs @@ -98,7 +98,10 @@ async fn retained_update_retries_never_claim_a_distinct_external_input_reference database.execute(sqlx::AssertSqlSafe(format!("DELETE FROM dw_command_intents WHERE command_id='{command}'"))).await; database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_commands SET request_id='external-retry-key' WHERE id='{command}'"))).await; } - let retry_input = upload(&app, &blob).await; + // Identical uploads deduplicate by digest. The transport trims the + // blob carrier before Avro decoding, so a trailing newline gives a + // distinct object/reference with the same decoded command intent. + let retry_input = upload(&app, &format!("{blob}\n")).await; assert_ne!(retry_input["reference_id"], original_input["reference_id"]); assert_eq!(unclaimed(&database).await, 1); let before = admission_snapshot(&database).await; From 6d7afc6ae99cf18fe4d085803baa29947c0ff64f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:58:15 +0000 Subject: [PATCH 5/6] Require explicit retained intent markers in receipt and parity expectations --- rust/tests/execution/structural_limits.rs | 5 ++++- rust/tests/execution/terminal_updates.rs | 5 ++++- .../server-parity/native-divergences.json | 9 ++++++++ .../update-validation-contract.mjs | 9 +++++++- ...rverParityUpdateValidationContractTest.mjs | 22 +++++++++++++++++++ 5 files changed, 47 insertions(+), 3 deletions(-) diff --git a/rust/tests/execution/structural_limits.rs b/rust/tests/execution/structural_limits.rs index 41ceaa452..088ce98e2 100644 --- a/rust/tests/execution/structural_limits.rs +++ b/rust/tests/execution/structural_limits.rs @@ -142,7 +142,10 @@ async fn updates_over_a_structural_limit_are_rejected_durably_like_php() { // A retried request returns the original rejection with its limit. let (status, retried) = request(&app, "POST", path, json!({"input":oversized(),"request_id":"big"})).await; assert_eq!(status, StatusCode::CONFLICT, "{retried}"); - assert_eq!(retried, big); + let mut retained = big.clone(); + retained["retained"] = json!(true); + retained["intent_verified"] = json!(false); + assert_eq!(retried, retained, "the original raw-key rejection is explicitly unverified"); let (status, accepted) = request(&app, "POST", path, json!({"input":note("a"),"wait_for":"accepted"})).await; assert_eq!(status, StatusCode::ACCEPTED, "{accepted}"); diff --git a/rust/tests/execution/terminal_updates.rs b/rust/tests/execution/terminal_updates.rs index c95ed75f9..e22db5c02 100644 --- a/rust/tests/execution/terminal_updates.rs +++ b/rust/tests/execution/terminal_updates.rs @@ -498,6 +498,9 @@ async fn terminal_updates_preserve_completed_receipts_and_another_active_run() { request(&app, "POST", first_path, input("resolved")).await, original_receipt ); + let mut retained_other = other_update.1.clone(); + retained_other["retained"] = json!(true); + retained_other["intent_verified"] = json!(true); assert_eq!( request( &app, @@ -507,7 +510,7 @@ async fn terminal_updates_preserve_completed_receipts_and_another_active_run() { ) .await .1, - other_update.1 + retained_other ); assert_eq!(finish_task(&app,&other_task,json!([{"type":"complete_update","update_id":other_update.1["update_id"],"result":envelope(Payload::Long(17))}])).await.0,StatusCode::OK); let history = run_history(&app, &first["workflow_id"], &first["run_id"]).await; diff --git a/scripts/conformance/server-parity/native-divergences.json b/scripts/conformance/server-parity/native-divergences.json index 1cebd2b73..5f5945db9 100644 --- a/scripts/conformance/server-parity/native-divergences.json +++ b/scripts/conformance/server-parity/native-divergences.json @@ -1,6 +1,15 @@ { "schema": "durable-workflow.server-parity-native-divergences/v1", "divergences": [ + { + "id": "retained-command-intent-readback", + "issue": "https://github.com/durable-workflow/server/issues/934", + "since": "native 3.9.1", + "applies_to": ["update-validation"], + "published_php": "A retained update receipt has no explicit retained or intent_verified fields.", + "native": "Retried update receipts explicitly identify retention and whether a recorded intent proves equality. A pre-intent legacy receipt is returned with intent_verified false rather than fabricated equality.", + "harness": "The update-validation contract requires retained true and intent_verified true on native's retained approval, and retained true with intent_verified false on the raw-key rejection that has no digest. Original receipt fields, PHP recordings and frozen fixtures are unchanged. Legacy unverified readback, ambiguity and zero writes are covered by all-backend execution regressions." + }, { "id": "request-id-reuse-conflict", "issue": "https://github.com/durable-workflow/server/issues/934", diff --git a/scripts/conformance/server-parity/update-validation-contract.mjs b/scripts/conformance/server-parity/update-validation-contract.mjs index 56007ac98..6a12afd6f 100644 --- a/scripts/conformance/server-parity/update-validation-contract.mjs +++ b/scripts/conformance/server-parity/update-validation-contract.mjs @@ -1,4 +1,5 @@ import assert from 'node:assert/strict'; +import {nativeDivergence} from './native-divergences.mjs'; // Synchronous update validation, as published Server 2.11.1 serves it. Every // recorded step must equal the reviewed fixture after three normalizations: @@ -56,9 +57,15 @@ export function checkUpdateValidation(fixture, observation, workflowId) { `update validation ${step} waited for the validation timeout (${seconds} seconds)`); } const actual = normalizeUpdateValidation(state, workflowId); + const retainedIntent = nativeDivergence(observation.target, 'retained-command-intent-readback'); assert.deepStrictEqual(Object.keys(actual), Object.keys(expected), 'update validation steps'); for (const step of Object.keys(expected)) { - assert.deepStrictEqual(actual[step], expected[step], `update validation ${step}`); + // Native's duplicate approval and raw-key rejection explicitly report + // whether an intent proof exists. Every original receipt field stays exact. + const wanted = retainedIntent && ['retained_approval', 'retained_rejection'].includes(step) + ? {...expected[step], response: {...expected[step].response, retained: true, intent_verified: step === 'retained_approval'}} + : expected[step]; + assert.deepStrictEqual(actual[step], wanted, `update validation ${step}`); } return {scope: 'update_validation', steps: Object.keys(expected).length}; } diff --git a/tests/Unit/ServerParityUpdateValidationContractTest.mjs b/tests/Unit/ServerParityUpdateValidationContractTest.mjs index 8fb0a412d..043f9fb4e 100644 --- a/tests/Unit/ServerParityUpdateValidationContractTest.mjs +++ b/tests/Unit/ServerParityUpdateValidationContractTest.mjs @@ -14,6 +14,28 @@ test('published PHP passes', () => { assert.deepStrictEqual(checkUpdateValidation(fixture, observation(), workflowId), projection); }); +test('native retained receipts explicitly report intent proof under their named divergence', () => { + const actual = {...observation(), target: 'rust'}; + const response = actual.update_validation.retained_approval.response; + response.retained = true; + response.intent_verified = true; + actual.update_validation.retained_rejection.response.retained = true; + actual.update_validation.retained_rejection.response.intent_verified = false; + assert.deepStrictEqual(checkUpdateValidation(fixture, actual, workflowId), projection); + for (const field of ['retained', 'intent_verified']) { + response[field] = false; + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /retained_approval/); + delete response[field]; + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /retained_approval/); + response[field] = true; + } + actual.update_validation.retained_rejection.response.intent_verified = true; + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /retained_rejection/); + actual.update_validation.retained_rejection.response.intent_verified = false; + actual.update_validation.approved_update.response.retained = true; + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /approved_update/); +}); + test('embedded explicitly declares update validation inapplicable', () => { assert.deepStrictEqual(checkUpdateValidation(fixture, {mode: 'embedded', update_validation: {applicable: false, reason: 'embedded_has_no_http_update_validation'}}, workflowId), projection); From cb7158a02c4fab6ef09d325425e5d2ea1c1a9469 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:24:04 +0000 Subject: [PATCH 6/6] Fix native target propagation in update validation comparator --- rust/tests/execution/update_validation.rs | 2 ++ scripts/conformance/server-parity/contract.mjs | 2 +- .../server-parity/update-validation-contract.mjs | 4 ++-- tests/Unit/ServerParityRunnerTest.mjs | 16 ++++++++++++++++ .../ServerParityUpdateValidationContractTest.mjs | 14 +++++++------- 5 files changed, 28 insertions(+), 10 deletions(-) diff --git a/rust/tests/execution/update_validation.rs b/rust/tests/execution/update_validation.rs index ade028323..63542c45c 100644 --- a/rust/tests/execution/update_validation.rs +++ b/rust/tests/execution/update_validation.rs @@ -170,12 +170,14 @@ async fn validators_approve_and_reject_updates_before_acceptance() { // Retries return the recorded receipts; an undeclared validator is not asked. let (status, retained) = request(&app, "POST", "/api/workflows/validated/update/approve_me", update_body(3, Some("r1"))).await; assert_eq!((status, retained["update_id"].clone()), (StatusCode::ACCEPTED, json!(id))); + assert_eq!((retained["retained"].clone(), retained["intent_verified"].clone()), (json!(true), json!(true)), "{retained}"); // Server #934: other arguments under the same request id conflict. let (status, conflict) = request(&app, "POST", "/api/workflows/validated/update/approve_me", update_body(99, Some("r1"))).await; assert_eq!((status, conflict["reason"].clone()), (StatusCode::CONFLICT, json!("request_id_conflict")), "{conflict}"); let (status, retained) = request(&app, "POST", "/api/workflows/validated/update/reject_me", update_body(2, Some("r2"))).await; assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{retained}"); assert_eq!(retained["reason"], "update_validator_rejected"); + assert_eq!((retained["retained"].clone(), retained["intent_verified"].clone()), (json!(true), json!(false)), "{retained}"); assert_eq!(retained["validation_errors"], json!({"v":["too small"]})); let (status, plain) = request(&app, "POST", "/api/workflows/validated/update/plain", update_body(4, Some("p1"))).await; assert_eq!(status, StatusCode::ACCEPTED, "{plain}"); diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 2e6aa8f73..a988b0325 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -245,7 +245,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts, ta const repairPass = fixture.repair_pass ? checkRepairPass(fixture, observation, workflowId) : null; const namespaceDelete = fixture.namespace_delete ? checkNamespaceDelete(fixture, observation, workflowId) : null; const namespaceDeletePayloads = fixture.namespace_delete_payloads ? checkNamespaceDeletePayloads(fixture, observation, workflowId) : null; - const updateValidation = fixture.update_validation ? checkUpdateValidation(fixture, observation, workflowId) : null; + const updateValidation = fixture.update_validation ? checkUpdateValidation(fixture, observation, workflowId, target) : null; const messageStreams = fixture.message_streams ? checkMessageStreams(fixture, observation, workflowId) : null; const workflowStreamWrites = fixture.workflow_stream_writes ? checkWorkflowStreamWrites(fixture, observation, workflowId) : null; const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null; diff --git a/scripts/conformance/server-parity/update-validation-contract.mjs b/scripts/conformance/server-parity/update-validation-contract.mjs index 6a12afd6f..88000ce3a 100644 --- a/scripts/conformance/server-parity/update-validation-contract.mjs +++ b/scripts/conformance/server-parity/update-validation-contract.mjs @@ -41,7 +41,7 @@ export function normalizeUpdateValidation(state, workflowId) { return walk(steps); } -export function checkUpdateValidation(fixture, observation, workflowId) { +export function checkUpdateValidation(fixture, observation, workflowId, target) { const state = observation.update_validation; assert.ok(state && typeof state === 'object', 'actual update validation observation'); const expected = fixture.update_validation.expected; @@ -57,7 +57,7 @@ export function checkUpdateValidation(fixture, observation, workflowId) { `update validation ${step} waited for the validation timeout (${seconds} seconds)`); } const actual = normalizeUpdateValidation(state, workflowId); - const retainedIntent = nativeDivergence(observation.target, 'retained-command-intent-readback'); + const retainedIntent = nativeDivergence(target, 'retained-command-intent-readback'); assert.deepStrictEqual(Object.keys(actual), Object.keys(expected), 'update validation steps'); for (const step of Object.keys(expected)) { // Native's duplicate approval and raw-key rejection explicitly report diff --git a/tests/Unit/ServerParityRunnerTest.mjs b/tests/Unit/ServerParityRunnerTest.mjs index adf2a32fe..69f7f0bae 100644 --- a/tests/Unit/ServerParityRunnerTest.mjs +++ b/tests/Unit/ServerParityRunnerTest.mjs @@ -38,6 +38,22 @@ function record(suffix = '') { return {schema: 'durable-workflow.server-parity-record/v1', target: `target${suffix}`, outcome: 'pass', runner_revision: 'a'.repeat(40), fixture_hashes: {'one-activity.json': 'fixture-sha'}, artifacts: {sdk_php: '2.2.6'}, cases: [{fixture_id: fixture.id, fixture, observation: raw, projection: checkObservation(fixture, raw, 'test-one-activity')}]}; } +test('the recorder forwards its native target to retained update validation checks', () => { + const updateFixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/update-validation.json', import.meta.url))); + const updateState = JSON.parse(readFileSync(new URL('../Fixtures/ServerParityExpected/update-validation-observation.json', import.meta.url), 'utf8').replaceAll('uv-probe-3', 'test-one-activity')); + const combined = {...fixture, update_validation: updateFixture.update_validation}; + const raw = {...observation(), mode: 'http', update_validation: updateState}; + // The implementation target belongs to the recording, never the probe observation. + raw.update_validation.workflow_id = `${raw.workflow_id}-uv`; + for (const step of ['retained_approval', 'retained_rejection']) { + raw.update_validation[step].response.retained = true; + raw.update_validation[step].response.intent_verified = step === 'retained_approval'; + } + assert.deepEqual(checkObservation(combined, raw, raw.workflow_id, undefined, 'rust').update_validation, + {scope: 'update_validation', steps: Object.keys(updateFixture.update_validation.expected).length}); + assert.throws(() => checkObservation(combined, raw, raw.workflow_id, undefined, 'php'), /retained_approval/); +}); + test('comparison binds identical recordings to the selected source artifact manifest', () => { const records = [record(), record('-other-database')]; compareCorpusRecords(records, {'one-activity.json': 'fixture-sha'}, reviewed, {sdk_php: '2.2.6'}); diff --git a/tests/Unit/ServerParityUpdateValidationContractTest.mjs b/tests/Unit/ServerParityUpdateValidationContractTest.mjs index 043f9fb4e..f047bb45a 100644 --- a/tests/Unit/ServerParityUpdateValidationContractTest.mjs +++ b/tests/Unit/ServerParityUpdateValidationContractTest.mjs @@ -15,25 +15,25 @@ test('published PHP passes', () => { }); test('native retained receipts explicitly report intent proof under their named divergence', () => { - const actual = {...observation(), target: 'rust'}; + const actual = observation(); const response = actual.update_validation.retained_approval.response; response.retained = true; response.intent_verified = true; actual.update_validation.retained_rejection.response.retained = true; actual.update_validation.retained_rejection.response.intent_verified = false; - assert.deepStrictEqual(checkUpdateValidation(fixture, actual, workflowId), projection); + assert.deepStrictEqual(checkUpdateValidation(fixture, actual, workflowId, 'rust'), projection); for (const field of ['retained', 'intent_verified']) { response[field] = false; - assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /retained_approval/); + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), /retained_approval/); delete response[field]; - assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /retained_approval/); + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), /retained_approval/); response[field] = true; } actual.update_validation.retained_rejection.response.intent_verified = true; - assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /retained_rejection/); + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), /retained_rejection/); actual.update_validation.retained_rejection.response.intent_verified = false; actual.update_validation.approved_update.response.retained = true; - assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), /approved_update/); + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), /approved_update/); }); test('embedded explicitly declares update validation inapplicable', () => { @@ -83,6 +83,6 @@ for (const [label, mutate, message] of [ test(`rejects ${label}`, () => { const actual = observation(); mutate(actual.update_validation); - assert.throws(() => checkUpdateValidation(fixture, actual, workflowId), message); + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), message); }); }