diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index edb2c4e8a..3d7621c45 100644 --- a/rust/src/runtime/store.rs +++ b/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, diff --git a/rust/src/runtime/updates.rs b/rust/src/runtime/updates.rs index 658edb4e4..63fcad3ee 100644 --- a/rust/src/runtime/updates.rs +++ b/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)) } diff --git a/rust/tests/execution/command_keys.rs b/rust/tests/execution/command_keys.rs index 49b3ae679..4156513d1 100644 --- a/rust/tests/execution/command_keys.rs +++ b/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); diff --git a/rust/tests/execution/external_payloads/commands.rs b/rust/tests/execution/external_payloads/commands.rs index f84d5026e..fcd08ba7b 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/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; diff --git a/rust/tests/execution/structural_limits.rs b/rust/tests/execution/structural_limits.rs index 088ce98e2..5242404d8 100644 --- a/rust/tests/execution/structural_limits.rs +++ b/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"); diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index a988b0325..ddd9c370a 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/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; diff --git a/scripts/conformance/server-parity/native-divergences.json b/scripts/conformance/server-parity/native-divergences.json index f7d94b6e4..abf3f99eb 100644 --- a/scripts/conformance/server-parity/native-divergences.json +++ b/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", diff --git a/scripts/conformance/server-parity/signals-before-worker-contract.mjs b/scripts/conformance/server-parity/signals-before-worker-contract.mjs index 957b5d84d..5ce67700f 100644 --- a/scripts/conformance/server-parity/signals-before-worker-contract.mjs +++ b/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'}; } diff --git a/scripts/conformance/server-parity/update-validation-contract.mjs b/scripts/conformance/server-parity/update-validation-contract.mjs index 88000ce3a..e61dcb56a 100644 --- a/scripts/conformance/server-parity/update-validation-contract.mjs +++ b/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}; diff --git a/tests/Unit/ServerParityRunnerTest.mjs b/tests/Unit/ServerParityRunnerTest.mjs index 69f7f0bae..f7ff068bd 100644 --- a/tests/Unit/ServerParityRunnerTest.mjs +++ b/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', () => { diff --git a/tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs b/tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs index da497a13d..073e87409 100644 --- a/tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs +++ b/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)); diff --git a/tests/Unit/ServerParityUpdateValidationContractTest.mjs b/tests/Unit/ServerParityUpdateValidationContractTest.mjs index f047bb45a..ffd6fd006 100644 --- a/tests/Unit/ServerParityUpdateValidationContractTest.mjs +++ b/tests/Unit/ServerParityUpdateValidationContractTest.mjs @@ -8,6 +8,15 @@ const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/update const recorded = JSON.parse(readFileSync(new URL('../Fixtures/ServerParityExpected/update-validation-observation.json', import.meta.url))); const workflowId = 'uv-probe-3'; const observation = () => ({mode: 'http', update_validation: structuredClone(recorded)}); +function nativeObservation() { + const actual = observation(); + for (const [step, request_id] of Object.entries({approved_update: 'r1', plain: 'p1', completed_wait: 'r4', + retained_approval: 'r1', retained_rejection: 'r2'})) { + Object.assign(actual.update_validation[step].response, {request_id, retained: step.startsWith('retained_'), + intent_verified: step !== 'retained_rejection'}); + } + return actual; +} const projection = {scope: 'update_validation', steps: Object.keys(fixture.update_validation.expected).length}; test('published PHP passes', () => { @@ -15,7 +24,7 @@ test('published PHP passes', () => { }); test('native retained receipts explicitly report intent proof under their named divergence', () => { - const actual = observation(); + const actual = nativeObservation(); const response = actual.update_validation.retained_approval.response; response.retained = true; response.intent_verified = true; @@ -36,6 +45,26 @@ test('native retained receipts explicitly report intent proof under their named assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), /approved_update/); }); +test('native keyed receipts require exact keys and fresh markers without weakening original fields', () => { + const steps = ['approved_update', 'plain', 'completed_wait', 'retained_approval', 'retained_rejection']; + for (const step of steps) { + for (const field of ['request_id', 'retained', 'intent_verified']) { + for (const mutation of ['missing', 'wrong']) { + const actual = nativeObservation(); + const response = actual.update_validation[step].response; + if (mutation === 'missing') delete response[field]; + else response[field] = field === 'request_id' ? 'another-key' : !response[field]; + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), new RegExp(step)); + } + } + const actual = nativeObservation(); + delete actual.update_validation[step].response.command_id; + assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), new RegExp(step)); + } + assert.throws(() => checkUpdateValidation(fixture, observation(), workflowId, 'rust'), /approved_update/); + assert.throws(() => checkUpdateValidation(fixture, nativeObservation(), workflowId, 'php'), /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); @@ -81,7 +110,7 @@ for (const [label, mutate, message] of [ ['a capability that is not advertised', state => { state.capability[0] = false; }], ]) { test(`rejects ${label}`, () => { - const actual = observation(); + const actual = nativeObservation(); mutate(actual.update_validation); assert.throws(() => checkUpdateValidation(fixture, actual, workflowId, 'rust'), message); });