Correct Unicode command keys and keyed admission receipts

rust/src/runtime/store.rs

@@ -1048,7 +1048,7 @@ where
             ));
         }
         if body.get("request_id").is_some_and(|value| {
-            !value.is_null() && value.as_str().is_none_or(|text| text.len() > 255)
+            !value.is_null() && value.as_str().is_none_or(|text| text.chars().count() > 255)
         }) {
             return Err(refuse(
                 StatusCode::UNPROCESSABLE_ENTITY,
@@ -1079,7 +1079,9 @@ where
         if let (Some(key), Some(digest)) = (key, digest.as_deref())
             && let Some((command, retained)) = Self::retained_command(&mut tx, workflow_id, "signal", key).await?
         {
-            return Self::retained_signal(&mut tx, workflow_id, name, run_id, digest, &command, retained.as_deref()).await;
+            let mut response = Self::retained_signal(&mut tx, workflow_id, name, run_id, digest, &command, retained.as_deref()).await?;
+            response["request_id"] = json!(key);
+            return Ok(response);
         }
         if run_id.is_some_and(|selected| selected != current_run) {
             return Err(refuse(
@@ -1251,9 +1253,13 @@ where
         }
         tx.commit().await?;
         self.wake.notify_waiters();
-        Ok(
-            json!({"workflow_id":workflow_id,"run_id":current_run,"resolved_run_id":current_run,"requested_run_id":null,"signal_name":name,"command_id":command_id,"command_sequence":command_sequence,"workflow_type":DB::string(&run, "workflow_type")?,"command_status":"accepted","command_source":"control_plane","target_scope":"instance","outcome":"signal_received","reason":null,"rejection_reason":null,"validation_errors":[]}),
-        )
+        let mut response = json!({"workflow_id":workflow_id,"run_id":current_run,"resolved_run_id":current_run,"requested_run_id":null,"signal_name":name,"command_id":command_id,"command_sequence":command_sequence,"workflow_type":DB::string(&run, "workflow_type")?,"command_status":"accepted","command_source":"control_plane","target_scope":"instance","outcome":"signal_received","reason":null,"rejection_reason":null,"validation_errors":[]});
+        if let Some(key) = key {
+            response["request_id"] = json!(key);
+            response["retained"] = json!(false);
+            response["intent_verified"] = json!(true);
+        }
+        Ok(response)
     }
 
     /// A refused signal's durable admission record: its rejected command,

rust/src/runtime/updates.rs

@@ -770,9 +770,10 @@ where
         tx.commit().await?;
         self.wake.notify_waiters();
         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());
+        if let Some(key) = key {
+            response["request_id"] = json!(key);
+            response["retained"] = json!(keyed.is_some());
+            response["intent_verified"] = json!(keyed.as_ref().is_none_or(|(_, digest)| digest.is_some()));
         }
         Ok((status, response))
     }

rust/tests/execution/command_keys.rs

@@ -180,6 +180,57 @@ async fn retention_prunes_request_key_digests_with_their_commands() {
     assert_eq!(task_retry::rows(&database, KEPT).await, 1, "the other run keeps its digest");
 }
 
+#[tokio::test]
+async fn unicode_request_keys_use_the_character_limit_and_keep_retry_identity() {
+    for (kind, action) in [("signal", "signal/finish"), ("update", "update/set")] {
+        let database = TestDatabase::new().await;
+        let runtime = database.open().await.unwrap();
+        let app = router(runtime.clone());
+        update_worker(&app, "unicode-key-worker").await;
+        let started = start(&app, "unicode-keys").await;
+        if kind == "update" {
+            open_update_wait(&app, "unicode-key-worker").await;
+        }
+        let path = format!("/api/workflows/unicode-keys/{action}");
+        for character in ["é", "🦀"] {
+            let key = character.repeat(255);
+            assert_eq!(key.chars().count(), 255);
+            assert!(key.len() > 255);
+            let original = request(&app, "POST", &path, input(1, Some(&key))).await;
+            assert_eq!(original.0, StatusCode::ACCEPTED, "{kind}: {}", original.1);
+            assert_eq!(original.1["request_id"], key);
+            assert_eq!((original.1["retained"].clone(), original.1["intent_verified"].clone()),
+                (json!(false), json!(true)), "{}", original.1);
+            let before = table_counts(&database).await;
+            let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await;
+            let replies = futures_util::future::join_all((0..4).map(|_| {
+                request(&app, "POST", &path, input(1, Some(&key)))
+            })).await;
+            for (status, reply) in replies {
+                assert_eq!(status, StatusCode::ACCEPTED, "{reply}");
+                assert_eq!(reply["command_id"], original.1["command_id"]);
+                assert_eq!(reply["run_id"], original.1["run_id"]);
+                assert_eq!(reply["request_id"], key);
+                assert_eq!((reply["retained"].clone(), reply["intent_verified"].clone()),
+                    (json!(true), json!(true)), "{reply}");
+                if kind == "update" {
+                    assert_eq!(reply["update_id"], original.1["update_id"]);
+                }
+            }
+            assert_conflict(&request(&app, "POST", &path, input(2, Some(&key))).await,
+                &original.1, "arguments");
+            let too_long = character.repeat(256);
+            let refused = request(&app, "POST", &path, input(1, Some(&too_long))).await;
+            assert_eq!(refused.0, StatusCode::UNPROCESSABLE_ENTITY, "{}", refused.1);
+            assert_eq!(refused.1["reason"], "validation_failed", "{}", 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;
+    }
+}
+
 /// 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) {
@@ -229,6 +280,7 @@ async fn legacy_updates_read_back_unverified_intent_after_restart_close_and_cont
             }
             assert_eq!((reply["retained"].clone(), reply["intent_verified"].clone()),
                 (json!(true), json!(false)), "{reply}");
+            assert_eq!(reply["request_id"], "legacy-update-key");
             if close.is_some() {
                 assert_eq!(reply["update_status"], "completed", "{reply}");
                 assert_eq!(reply["result"], Value::Null, "completed null remains an outcome");
@@ -272,6 +324,7 @@ async fn legacy_signals_do_not_honor_a_changed_target_or_fabricate_intent_equali
     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)));
+    assert_eq!(retry.1["request_id"], "legacy-signal-key");
     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);

rust/tests/execution/external_payloads/commands.rs

@@ -105,15 +105,16 @@ async fn retained_update_retries_never_claim_a_distinct_external_input_reference
         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",
+        let (status, 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");
+        let mut expected = original.clone();
+        expected["retained"] = json!(true);
+        expected["intent_verified"] = json!(!legacy);
+        assert_eq!(repeated, expected, "every original receipt field is preserved");
         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;

rust/tests/execution/structural_limits.rs

@@ -143,6 +143,7 @@ async fn updates_over_a_structural_limit_are_rejected_durably_like_php() {
     let (status, retried) = request(&app, "POST", path, json!({"input":oversized(),"request_id":"big"})).await;
     assert_eq!(status, StatusCode::CONFLICT, "{retried}");
     let mut retained = big.clone();
+    retained["request_id"] = json!("big");
     retained["retained"] = json!(true);
     retained["intent_verified"] = json!(false);
     assert_eq!(retried, retained, "the original raw-key rejection is explicitly unverified");

scripts/conformance/server-parity/contract.mjs

@@ -250,7 +250,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts, ta
   const workflowStreamWrites = fixture.workflow_stream_writes ? checkWorkflowStreamWrites(fixture, observation, workflowId) : null;
   const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null;
   const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId, target) : null;
-  const signalsBeforeWorker = fixture.signals_before_worker ? checkSignalsBeforeWorker(fixture, observation, workflowId) : null;
+  const signalsBeforeWorker = fixture.signals_before_worker ? checkSignalsBeforeWorker(fixture, observation, workflowId, target) : null;
   const workerListing = fixture.worker_listing ? checkWorkerListing(fixture, observation, workflowId) : null;
   const workerReplacement = fixture.worker_replacement ? checkWorkerReplacement(fixture, observation, workflowId) : null;
   const operatorMetrics = fixture.operator_metrics ? checkOperatorMetrics(fixture, observation, workflowId) : null;

scripts/conformance/server-parity/native-divergences.json

@@ -1,6 +1,18 @@
 {
   "schema": "durable-workflow.server-parity-native-divergences/v1",
   "divergences": [
+    {
+      "id": "command-key-receipt-metadata",
+      "issue": "https://github.com/durable-workflow/server/issues/934",
+      "since": "native 3.9.2",
+      "applies_to": [
+        "signals-before-worker",
+        "update-validation"
+      ],
+      "published_php": "Keyed signal and update receipts do not echo the caller request_id or identify fresh admission with retained false and intent_verified true.",
+      "native": "Fresh keyed admissions echo the literal caller key with retained false and intent_verified true. Retained receipts echo the same key and retain their existing intent-proof markers, including false for legacy receipts without a digest.",
+      "harness": "The signals-before-worker comparator requires exact metadata only on its keyed admission. Update validation requires exact caller keys and fresh markers on approved/plain/completed-wait admissions and preserves true/false intent evidence on retained approval/raw-key rejection. Every original receipt field and all other responses remain exact. Frozen PHP fixtures and observations are unchanged."
+    },
     {
       "id": "retained-command-intent-readback",
       "issue": "https://github.com/durable-workflow/server/issues/934",

scripts/conformance/server-parity/signals-before-worker-contract.mjs

@@ -1,5 +1,6 @@
 import assert from 'node:assert/strict';
 import {readFileSync} from 'node:fs';
+import {nativeDivergence} from './native-divergences.mjs';
 
 // Published PHP's reviewed, normalized responses to every step.
 const EXPECTED = JSON.parse(readFileSync(new URL('../../../tests/Fixtures/ServerParityExpected/signals-before-worker.json', import.meta.url)));
@@ -24,7 +25,7 @@ export const STATUSES = [201, 202, 202, 202, 202, 422, 200, 409, 422, 404, 200];
 const EVENTS = ['StartAccepted', 'WorkflowStarted', 'SignalReceived', 'SignalReceived', 'SignalReceived', 'SignalReceived',
   'TerminateRequested', 'WorkflowTerminated', 'SignalRejected', 'SignalRejected', 'SignalRejected', 'SignalRejected'];
 
-export function checkSignalsBeforeWorker(fixture, observation, workflowId) {
+export function checkSignalsBeforeWorker(fixture, observation, workflowId, target) {
   const state = observation.signals_before_worker;
   assert.ok(state && typeof state === 'object', 'actual signals-before-worker observation');
   if (observation.mode === 'embedded') {
@@ -49,6 +50,12 @@ export function checkSignalsBeforeWorker(fixture, observation, workflowId) {
   // This fixture covers signal admission: the start response and history
   // payloads are compared by their own fixtures.
   const compared = steps.slice(1, 10);
-  assert.deepStrictEqual(compared, EXPECTED, 'every response matches the reviewed published PHP responses');
+  const expected = structuredClone(EXPECTED);
+  if (nativeDivergence(target, 'command-key-receipt-metadata')) {
+    // Only the first signal carries a caller key. Keep every original field
+    // and all unkeyed/refused responses equal to the frozen PHP recording.
+    Object.assign(expected[0].response, {request_id: 'before-worker-1', retained: false, intent_verified: true});
+  }
+  assert.deepStrictEqual(compared, expected, 'every response matches the reviewed target contract');
   return {scope: 'signals_before_worker'};
 }

scripts/conformance/server-parity/update-validation-contract.mjs

@@ -58,13 +58,21 @@ export function checkUpdateValidation(fixture, observation, workflowId, target)
   }
   const actual = normalizeUpdateValidation(state, workflowId);
   const retainedIntent = nativeDivergence(target, 'retained-command-intent-readback');
+  const receiptMetadata = nativeDivergence(target, 'command-key-receipt-metadata');
+  const keyedReceipts = {approved_update: 'r1', plain: 'p1', completed_wait: 'r4',
+    retained_approval: 'r1', retained_rejection: 'r2'};
   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
     // whether an intent proof exists. Every original receipt field stays exact.
-    const wanted = retainedIntent && ['retained_approval', 'retained_rejection'].includes(step)
+    let wanted = retainedIntent && ['retained_approval', 'retained_rejection'].includes(step)
       ? {...expected[step], response: {...expected[step].response, retained: true, intent_verified: step === 'retained_approval'}}
       : expected[step];
+    if (receiptMetadata && Object.hasOwn(keyedReceipts, step)) {
+      const retained = step.startsWith('retained_');
+      wanted = {...wanted, response: {...wanted.response, request_id: keyedReceipts[step],
+        retained, intent_verified: step !== 'retained_rejection'}};
+    }
     assert.deepStrictEqual(actual[step], wanted, `update validation ${step}`);
   }
   return {scope: 'update_validation', steps: Object.keys(expected).length};

tests/Unit/ServerParityRunnerTest.mjs

@@ -45,13 +45,14 @@ test('the recorder forwards its native target to retained update validation chec
   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';
+  for (const [step, request_id] of Object.entries({approved_update: 'r1', plain: 'p1', completed_wait: 'r4',
+    retained_approval: 'r1', retained_rejection: 'r2'})) {
+    Object.assign(raw.update_validation[step].response, {request_id, retained: step.startsWith('retained_'),
+      intent_verified: step !== 'retained_rejection'});
   }
   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/);
+  assert.throws(() => checkObservation(combined, raw, raw.workflow_id, undefined, 'php'), /approved_update/);
 });
 
 test('comparison binds identical recordings to the selected source artifact manifest', () => {

tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs

@@ -26,6 +26,25 @@ function model() {
 const projection = {scope: 'signals_before_worker'};
 
 test('published PHP responses pass', () => assert.deepStrictEqual(checkSignalsBeforeWorker(fixture, model(), workflowId), projection));
+test('native requires exact fresh keyed metadata while preserving all original receipt fields', () => {
+  const native = model();
+  const receipt = native.signals_before_worker.steps[1].response;
+  Object.assign(receipt, {request_id: 'before-worker-1', retained: false, intent_verified: true});
+  assert.deepStrictEqual(checkSignalsBeforeWorker(fixture, native, workflowId, 'rust'), projection);
+  assert.throws(() => checkSignalsBeforeWorker(fixture, native, workflowId, 'php'));
+  assert.throws(() => checkSignalsBeforeWorker(fixture, model(), workflowId, 'rust'));
+  for (const [field, wrong] of [['request_id', 'another-key'], ['retained', true], ['intent_verified', false],
+    ['command_status', 'rejected'], ['command_id', 'missing']]) {
+    const original = receipt[field];
+    receipt[field] = wrong;
+    assert.throws(() => checkSignalsBeforeWorker(fixture, native, workflowId, 'rust'), field);
+    delete receipt[field];
+    assert.throws(() => checkSignalsBeforeWorker(fixture, native, workflowId, 'rust'), field);
+    receipt[field] = original;
+  }
+  native.signals_before_worker.steps[2].response.retained = false;
+  assert.throws(() => checkSignalsBeforeWorker(fixture, native, workflowId, 'rust'), 'unkeyed responses stay exact');
+});
 test('embedded explicitly declares the case inapplicable', () => assert.deepStrictEqual(checkSignalsBeforeWorker(fixture,
   {mode: 'embedded', signals_before_worker: {applicable: false, reason: 'embedded_runs_have_a_loadable_contract'}}, workflowId),
   projection));