diff --git a/docs/cli-reference.md b/docs/cli-reference.md index daaa95a..8a7dca5 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -486,6 +486,7 @@ dw watch workflow order-123 --run-id=01HXYZ --interval=5 --max-polls=60 # Send a signal dw workflow:signal order-123 payment-received --input='{"amount":99.99}' +dw workflow:signal order-123 payment-received --input='{"amount":99.99}' --request-id=payment-order-123-2026-10-20 dw workflow:signal counter-1 increment --input='["not-an-int"]' --output=json # {"error":"Server error: Signal argument validation failed.","exit_code":2,"status_code":422,"reason":"invalid_signal_arguments",...} @@ -496,6 +497,7 @@ dw workflow:query counter-1 current-at --input='["not-an-int"]' --output=json # Send an update dw workflow:update order-123 approve --input='{"approver":"admin"}' +dw workflow:update order-123 approve --input='{"approver":"admin"}' --wait=completed --request-id=approve-order-123-2026-10-20 # Send an integration event through a bounded bridge adapter dw bridge:webhook stripe --action=start_workflow --idempotency-key=stripe-event-1001 --target='{"workflow_type":"orders.fulfillment","task_queue":"external-workflows","business_key":"order-1001"}' --input='{"order_id":"order-1001"}' @@ -513,6 +515,22 @@ dw workflow:history order-123 01HXYZ dw workflow:history order-123 01HXYZ --follow ``` +`workflow:signal` and `workflow:update` send a durable command key as the +`request_id` body field. The CLI generates `cli-signal-` or +`cli-update-` once per invocation unless `--request-id` (1-255 bytes, +opaque, case-sensitive) is given, and prints the key it used as `Request ID` +in human output and as `request_id` in `--json` output, including in error +envelopes after a timeout or network failure. Retrying the same command with +that key returns the original receipt instead of recording a second signal or +update; the response then carries `retained: true`, and `intent_verified` +reports whether the server could compare the retry against the retained +intent. A retry that reuses the key with a different name, arguments or run +target fails with HTTP 409 and `reason: request_id_conflict`; the error names +the key, the retained command (`retained_command_id`, `retained_command_type`, +`retained_outcome`) and the mismatched `conflict` parts. Servers that predate +keyed signals accept the field and omit the `retained`/`intent_verified` +flags. + ### Nexus service calls Inspect one durable call in its endpoint namespace: diff --git a/schemas/output/manifest.json b/schemas/output/manifest.json index e3fd4fc..723113b 100644 --- a/schemas/output/manifest.json +++ b/schemas/output/manifest.json @@ -550,7 +550,7 @@ "schema": "schemas/output/workflow-signal.schema.json", "schema_id": "https://durable-workflow.github.io/cli-json-envelopes/v3/schemas/workflow-signal.schema.json", "resolver_url": "https://durable-workflow.github.io/cli-json-envelopes/v3/schemas/workflow-signal.schema.json", - "sha256": "sha256:c42c02f7a54faa082982028482bc231ca29c495a2a00fabce5b0ae580a2505bf" + "sha256": "sha256:5a5fe23d6a1bead8508887b57b1dfb58448b92ed4292ce0da88bcdd1abf37e81" }, "workflow:start": { "output": "--json", @@ -571,7 +571,7 @@ "schema": "schemas/output/workflow-update.schema.json", "schema_id": "https://durable-workflow.github.io/cli-json-envelopes/v3/schemas/workflow-update.schema.json", "resolver_url": "https://durable-workflow.github.io/cli-json-envelopes/v3/schemas/workflow-update.schema.json", - "sha256": "sha256:a3997a4b623d84b166ea3a124b0a33e5c400acc51694332186813b96aa389eb1" + "sha256": "sha256:8888b18ff8d8ef7cac35f2aae362192b04617dc28760c25ddccf085880c3e25c" } }, "jsonl_commands": { diff --git a/schemas/output/workflow-signal.schema.json b/schemas/output/workflow-signal.schema.json index 1e1e3c9..33804b1 100644 --- a/schemas/output/workflow-signal.schema.json +++ b/schemas/output/workflow-signal.schema.json @@ -11,7 +11,10 @@ "signal_name": { "type": "string" }, "outcome": { "type": "string" }, "command_status": { "type": ["string", "null"] }, - "command_id": { "type": ["string", "null"] } + "command_id": { "type": ["string", "null"] }, + "request_id": { "type": "string" }, + "retained": { "type": ["boolean", "null"] }, + "intent_verified": { "type": ["boolean", "null"] } }, "additionalProperties": true } diff --git a/schemas/output/workflow-update.schema.json b/schemas/output/workflow-update.schema.json index a863656..99b22c1 100644 --- a/schemas/output/workflow-update.schema.json +++ b/schemas/output/workflow-update.schema.json @@ -13,6 +13,8 @@ "update_name": { "type": "string" }, "update_id": { "type": ["string", "null"] }, "request_id": { "type": ["string", "null"] }, + "retained": { "type": ["boolean", "null"] }, + "intent_verified": { "type": ["boolean", "null"] }, "outcome": { "type": "string" }, "state": { "type": ["string", "null"] }, "update_state": { "type": ["string", "null"] }, @@ -119,6 +121,8 @@ "update_name": { "type": ["string", "null"] }, "update_id": { "type": ["string", "null"] }, "request_id": { "type": ["string", "null"] }, + "retained": { "type": ["boolean", "null"] }, + "intent_verified": { "type": ["boolean", "null"] }, "outcome": { "type": ["string", "null"] }, "reason": { "type": ["string", "null"] }, "rejection_reason": { "type": ["string", "null"] }, diff --git a/src/Commands/BaseCommand.php b/src/Commands/BaseCommand.php index 4f4762b..bd47465 100644 --- a/src/Commands/BaseCommand.php +++ b/src/Commands/BaseCommand.php @@ -29,6 +29,10 @@ abstract class BaseCommand extends Command { private const INPUT_ENCODINGS = ['json', 'raw', 'base64']; + private const COMMAND_REQUEST_ID_MAX_CHARACTERS = 255; + + private const REQUEST_ID_CONFLICT_REASON = 'request_id_conflict'; + private const NAMESPACE_SCOPED_COMMAND_NAMES = [ 'activity:complete', 'activity:describe', @@ -122,6 +126,12 @@ abstract class BaseCommand extends Command private ?ProfileStore $profileStore = null; + /** + * The durable command key sent with the current signal or update request, + * remembered so error output can name it when the response is lost. + */ + private ?string $commandRequestId = null; + protected function configure(): void { $this->addOption('server', 's', InputOption::VALUE_OPTIONAL, 'Server URL', null); @@ -663,6 +673,64 @@ protected function withNamespaceContext(string $namespace, array $payload, ?stri return $payload; } + /** + * Resolve the durable command key for a signal or update request: the + * explicit `--request-id` value when one was given, otherwise one key + * generated for this invocation. The key is remembered so a lost or + * conflicting response can still name it for a retry. + * + * @param string $commandKind `signal` or `update` + */ + protected function commandRequestId(InputInterface $input, string $commandKind): string + { + $this->commandRequestId = null; + + $explicit = $input->hasOption('request-id') ? $input->getOption('request-id') : null; + + if (is_scalar($explicit) && trim((string) $explicit) !== '') { + $explicit = (string) $explicit; + + $characters = preg_match_all('/./us', $explicit); + if ($characters === false) { + throw new InvalidOptionException('--request-id must be valid UTF-8.'); + } + if ($characters > self::COMMAND_REQUEST_ID_MAX_CHARACTERS) { + throw new InvalidOptionException(sprintf( + '--request-id must be 1 to %d Unicode characters; got %d characters.', + self::COMMAND_REQUEST_ID_MAX_CHARACTERS, + $characters, + )); + } + + return $this->commandRequestId = $explicit; + } + + return $this->commandRequestId = sprintf('cli-%s-%s', $commandKind, bin2hex(random_bytes(16))); + } + + /** + * Print the command key and, when the server reported them, the retained + * receipt flags so an operator can retry with the same key. + * + * @param array $result + */ + protected function writeCommandReceiptLines(OutputInterface $output, array $result): void + { + if ($this->hasNonEmptyString($result['request_id'] ?? null)) { + $output->writeln(' Request ID: '.$result['request_id']); + } + + if (is_bool($result['retained'] ?? null)) { + $output->writeln(' Retained: '.($result['retained'] + ? 'yes (original receipt returned for this request id)' + : 'no')); + } + + if (is_bool($result['intent_verified'] ?? null)) { + $output->writeln(' Intent Verified: '.($result['intent_verified'] ? 'yes' : 'no')); + } + } + /** * @param array $payload * @param array $request @@ -871,6 +939,8 @@ private function workflowUpdateDiagnosticEnvelope(array $payload): array 'update_name', 'update_id', 'request_id', + 'retained', + 'intent_verified', 'outcome', 'reason', 'rejection_reason', @@ -926,6 +996,8 @@ private function workflowUpdateCliFields(array $payload): array 'update_name', 'update_id', 'request_id', + 'retained', + 'intent_verified', 'state', 'update_state', 'outcome', @@ -1143,6 +1215,7 @@ private function emitError(InputInterface $input, OutputInterface $output, \Thro $this->writeHumanError($output, $e->getMessage()); $this->writeHumanErrorNamespaceContext($output, $input); + $this->writeHumanErrorRequestIdContext($output, $input); $this->writeHumanHttpErrorDetails($output, $e); $recommendations = $this->errorRecommendations($e, $input); @@ -1247,6 +1320,12 @@ private function copyServerResponseDiagnostics(array &$envelope, array $body): v 'rejection_reason', 'rejection_category', 'request_id', + 'retained', + 'intent_verified', + 'retained_command_id', + 'retained_command_type', + 'retained_outcome', + 'conflict', 'workflow_sequence', 'result', 'result_envelope', @@ -1309,6 +1388,17 @@ private function writeHumanErrorNamespaceContext(OutputInterface $output, InputI $target->writeln('Namespace: '.$namespace); } + private function writeHumanErrorRequestIdContext(OutputInterface $output, InputInterface $input): void + { + $requestId = $this->errorRequestIdContext($input); + if ($requestId === null) { + return; + } + + $target = $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output; + $target->writeln('Request ID: '.$requestId); + } + private function writeHumanHttpErrorDetails(OutputInterface $output, \Throwable $e): void { if (! $e instanceof ServerHttpException) { @@ -1328,6 +1418,12 @@ private function writeHumanHttpErrorDetails(OutputInterface $output, \Throwable $target->writeln(str_replace('_', ' ', ucfirst($key)).': '.$value); } } + + if ($e->reason() === self::REQUEST_ID_CONFLICT_REASON) { + foreach ($this->requestIdConflictLines($e->body) as $line) { + $target->writeln($line); + } + } } $validationErrors = $e->validationErrors(); @@ -1382,6 +1478,10 @@ private function errorNamespaceContext(InputInterface $input): ?string private function errorRequestIdContext(InputInterface $input): ?string { + if ($this->commandRequestId !== null) { + return $this->commandRequestId; + } + if (! $input->hasOption('request-id')) { return null; } @@ -1420,7 +1520,7 @@ private function withWorkflowUpdateErrorDiagnostics( $request = array_filter([ 'wait_for' => $this->stringOptionContext($input, 'wait') ?? (is_scalar($envelope['wait_for'] ?? null) ? (string) $envelope['wait_for'] : 'accepted'), - 'request_id' => $this->stringOptionContext($input, 'request-id') ?? (is_scalar($envelope['request_id'] ?? null) ? (string) $envelope['request_id'] : null), + 'request_id' => $this->errorRequestIdContext($input) ?? (is_scalar($envelope['request_id'] ?? null) ? (string) $envelope['request_id'] : null), 'input' => $this->workflowUpdateInputContext($input), ], static fn (mixed $value): bool => $value !== null); @@ -1574,6 +1674,104 @@ private function writeHumanWarning(OutputInterface $output, string $message): vo * @return list */ private function errorRecommendations(\Throwable $e, ?InputInterface $input): array + { + $recommendations = $this->baseErrorRecommendations($e, $input); + + $retry = $this->requestIdRetryRecommendation($e); + if ($retry !== null) { + $recommendations[] = $retry; + } + + return $recommendations; + } + + /** + * When the outcome of a keyed signal or update is unknown (transport + * failure or a server-side error), tell the operator which key to reuse so + * the retry reads the original receipt instead of recording a new command. + * + * @return array{id: string, severity: string, message: string}|null + */ + private function requestIdRetryRecommendation(\Throwable $e): ?array + { + if ($this->commandRequestId === null) { + return null; + } + + $uncertain = $e instanceof NetworkException + || $e instanceof TimeoutException + || ($e instanceof ServerHttpException && $e->statusCode >= 500 && $e->statusCode <= 599); + + if (! $uncertain) { + return null; + } + + return [ + 'id' => 'request_id.retry', + 'severity' => 'error', + 'message' => sprintf( + 'The outcome of this command is unknown. Retry it unchanged with --request-id=%s so the server returns the original receipt instead of recording a second command.', + $this->commandRequestId, + ), + ]; + } + + /** + * @param array $body + * @return list + */ + private function requestIdConflictLines(array $body): array + { + $lines = []; + + $retained = array_filter([ + 'id' => $this->nonEmptyScalarString($body['retained_command_id'] ?? null), + 'type' => $this->nonEmptyScalarString($body['retained_command_type'] ?? null), + 'outcome' => $this->nonEmptyScalarString($body['retained_outcome'] ?? null), + ], static fn (?string $value): bool => $value !== null); + + if ($retained !== []) { + $lines[] = 'Retained command: '.implode(', ', array_map( + static fn (string $field, string $value): string => $field.' '.$value, + array_keys($retained), + $retained, + )); + } + + $conflict = $this->requestIdConflictParts($body); + if ($conflict !== []) { + $lines[] = 'Conflicting parts: '.implode(', ', $conflict); + } + + return $lines; + } + + /** + * @param array $body + * @return list + */ + private function requestIdConflictParts(array $body): array + { + $conflict = $body['conflict'] ?? null; + if (! is_array($conflict)) { + return []; + } + + $parts = []; + foreach ($conflict as $part) { + $value = $this->nonEmptyScalarString($part); + if ($value !== null) { + $parts[] = $value; + } + } + + return $parts; + } + + /** + * @return list + */ + private function baseErrorRecommendations(\Throwable $e, ?InputInterface $input): array { if ($e instanceof NetworkException) { return [[ @@ -1641,6 +1839,25 @@ private function httpErrorRecommendations(ServerHttpException $e, ?InputInterfac ]]; } + if ($e->statusCode === 409 && $e->reason() === self::REQUEST_ID_CONFLICT_REASON) { + $body = $e->body ?? []; + $requestId = $this->nonEmptyScalarString($body['request_id'] ?? null) + ?? ($input !== null ? $this->errorRequestIdContext($input) : null) + ?? 'given'; + $parts = $this->requestIdConflictParts($body); + + return [[ + 'id' => 'request_id.conflict', + 'severity' => 'error', + 'message' => sprintf( + 'Request id [%s] already identifies a %s command with a different %s. Repeat that original command unchanged to read its receipt, or pass a new --request-id to record a new command.', + $requestId, + $this->nonEmptyScalarString($body['retained_command_type'] ?? null) ?? 'retained', + $parts === [] ? 'intent' : implode('/', $parts), + ), + ]]; + } + if ($e->statusCode >= 400 && $e->statusCode <= 499) { return [[ 'id' => 'request.invalid', diff --git a/src/Commands/WorkflowCommand/SignalCommand.php b/src/Commands/WorkflowCommand/SignalCommand.php index fe37f02..2363426 100644 --- a/src/Commands/WorkflowCommand/SignalCommand.php +++ b/src/Commands/WorkflowCommand/SignalCommand.php @@ -23,14 +23,24 @@ protected function configure(): void durably recorded in history even if the workflow is currently idle or waiting. +Every signal carries a durable command key (request_id). +The CLI generates one per invocation unless --request-id +is given, and prints the key it used. After a lost response, retry the same +command with that key: the server returns the original receipt instead of +recording a second signal, and refuses the key with HTTP 409 +request_id_conflict if the retry carries a different +signal name, arguments or run target. + Examples: dw workflow:signal chk-42 approve dw workflow:signal chk-42 payment_received -i '{"amount":99.95}' + dw workflow:signal chk-42 approve --request-id=approve-chk-42-2026-10-20 --json HELP) ->addArgument('workflow-id', InputArgument::REQUIRED, 'Workflow ID') ->addArgument('signal-name', InputArgument::REQUIRED, 'Signal name') ->addOption('run-id', null, InputOption::VALUE_OPTIONAL, 'Target a specific run ID') + ->addOption('request-id', null, InputOption::VALUE_OPTIONAL, 'Durable command key reused on retry (1-255 Unicode characters); generated when omitted') ->addOption('json', null, InputOption::VALUE_NONE, 'Output the command response as JSON'); $this->addInputOptions('Signal input payload'); } @@ -40,8 +50,11 @@ protected function execute(InputInterface $input, OutputInterface $output): int $workflowId = $input->getArgument('workflow-id'); $signalName = $input->getArgument('signal-name'); $runId = $input->getOption('run-id'); + $requestId = $this->commandRequestId($input, 'signal'); - $body = []; + $body = [ + 'request_id' => $requestId, + ]; $parsedInput = $this->parseInputArgumentsOption($input); if ($parsedInput !== null) { $body['input'] = $parsedInput; @@ -53,6 +66,12 @@ protected function execute(InputInterface $input, OutputInterface $output): int $result = $this->addNamespaceContext($input, $this->client($input)->post($path, $body)); + // Servers that predate keyed signals do not echo the key; the operator + // still needs it to retry after a lost response. + if (! is_string($result['request_id'] ?? null) || $result['request_id'] === '') { + $result['request_id'] = $requestId; + } + if ($this->wantsJson($input)) { return $this->renderJson($output, $result); } @@ -68,6 +87,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int if (isset($result['command_id'])) { $output->writeln(' Command ID: '.$result['command_id']); } + $this->writeCommandReceiptLines($output, $result); return Command::SUCCESS; } diff --git a/src/Commands/WorkflowCommand/UpdateCommand.php b/src/Commands/WorkflowCommand/UpdateCommand.php index 5845c1f..8dc1be3 100644 --- a/src/Commands/WorkflowCommand/UpdateCommand.php +++ b/src/Commands/WorkflowCommand/UpdateCommand.php @@ -25,16 +25,25 @@ protected function configure(): void validated the update; --wait=completed returns when the handler has finished and a result is available. +Every update carries a durable command key (request_id). +The CLI generates one per invocation unless --request-id +is given, and prints the key it used. After a lost response, retry the same +command with that key: the server returns the original receipt (same update +id, status and result) instead of recording a second update, and refuses the +key with HTTP 409 request_id_conflict if the retry carries +a different update name, arguments or run target. + Examples: dw workflow:update chk-42 increase_quota -i '{"by":10}' dw workflow:update chk-42 increase_quota -i '{"by":10}' --wait=completed + dw workflow:update chk-42 increase_quota -i '{"by":10}' --request-id=quota-chk-42-2026-10-20 --json HELP) ->addArgument('workflow-id', InputArgument::REQUIRED, 'Workflow ID') ->addArgument('update-name', InputArgument::REQUIRED, 'Update name') ->addOption('wait', null, InputOption::VALUE_OPTIONAL, 'Wait policy (accepted, completed)', 'accepted', CompletionValues::UPDATE_WAIT_POLICIES) ->addOption('run-id', null, InputOption::VALUE_OPTIONAL, 'Target a specific run ID') - ->addOption('request-id', null, InputOption::VALUE_OPTIONAL, 'Durable request identifier for retry/idempotency diagnostics') + ->addOption('request-id', null, InputOption::VALUE_OPTIONAL, 'Durable command key reused on retry (1-255 Unicode characters); generated when omitted') ->addOption('json', null, InputOption::VALUE_NONE, 'Output the command response as JSON'); $this->addInputOptions('Update input payload'); } @@ -60,11 +69,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int $body = [ 'wait_for' => $waitFor, + 'request_id' => $this->commandRequestId($input, 'update'), ]; - $requestId = $this->optionalString($input->getOption('request-id')); - if ($requestId !== null) { - $body['request_id'] = $requestId; - } $parsedInput = $this->parseInputArgumentsOption($input); if ($parsedInput !== null) { @@ -103,6 +109,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int if (isset($result['wait_for'])) { $output->writeln(' Wait For: '.$result['wait_for']); } + $this->writeCommandReceiptLines($output, $result); return Command::SUCCESS; } diff --git a/tests/Commands/WorkflowCommandRequestIdTest.php b/tests/Commands/WorkflowCommandRequestIdTest.php new file mode 100644 index 0000000..96fbfb8 --- /dev/null +++ b/tests/Commands/WorkflowCommandRequestIdTest.php @@ -0,0 +1,765 @@ +runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertSame(1, $client->postCalls); + + $key = $client->lastPostBody['request_id'] ?? null; + self::assertIsString($key); + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $key); + self::assertSame(['request_id'], array_keys($client->lastPostBody)); + self::assertStringContainsString('Request ID: '.$key, $tester->getDisplay()); + + $second = new CommandRequestIdFakeClient(self::signalReceipt()); + $this->runSignal($second, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + ]); + + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $second->lastPostBody['request_id']); + self::assertNotSame($key, $second->lastPostBody['request_id'], 'Each invocation must get its own key.'); + } + + public function test_update_generates_one_cli_update_key_when_request_id_is_omitted(): void + { + $client = new CommandRequestIdFakeClient(self::updateReceipt(), contract: self::requestContract()); + + $tester = $this->runUpdate($client, [ + 'workflow-id' => 'wf-1', + 'update-name' => 'approve', + '--json' => true, + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + + $key = $client->lastPostBody['request_id'] ?? null; + self::assertIsString($key); + self::assertMatchesRegularExpression(self::UPDATE_KEY_PATTERN, $key); + self::assertSame(['wait_for' => 'accepted', 'request_id' => $key], $client->lastPostBody); + + $decoded = $this->json($tester); + + self::assertSame($key, $decoded['request_id']); + self::assertSame($key, $decoded['request']['request_id']); + self::assertSame($key, $decoded['update_diagnostics']['request_id']); + self::assertSame($key, $decoded['cli_fields']['request_id']); + } + + public function test_signal_sends_an_explicit_request_id_verbatim_and_renders_receipt_flags(): void + { + $key = 'Approve-Order.42:Attempt/1'; + $client = new CommandRequestIdFakeClient(self::signalReceipt([ + 'request_id' => $key, + 'retained' => true, + 'intent_verified' => true, + ])); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + '--request-id' => $key, + '--input' => '[{"approved":true}]', + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertSame([ + 'request_id' => $key, + 'input' => [['approved' => true]], + ], $client->lastPostBody); + + $display = $tester->getDisplay(); + + self::assertStringContainsString('Signal sent', $display); + self::assertStringContainsString('Request ID: '.$key, $display); + self::assertStringContainsString('Retained: yes', $display); + self::assertStringContainsString('Intent Verified: yes', $display); + } + + public function test_update_sends_an_explicit_request_id_verbatim(): void + { + $key = 'Quota-Adjust.7:Attempt/2'; + $client = new CommandRequestIdFakeClient(self::updateReceipt(), contract: self::requestContract()); + + $tester = $this->runUpdate($client, [ + 'workflow-id' => 'wf-1', + 'update-name' => 'approve', + '--wait' => 'completed', + '--request-id' => $key, + '--input' => '[{"by":10}]', + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertSame([ + 'wait_for' => 'completed', + 'request_id' => $key, + 'input' => [['by' => 10]], + ], $client->lastPostBody); + self::assertStringContainsString('Request ID: '.$key, $tester->getDisplay()); + } + + public function test_signal_json_output_includes_request_id_and_passes_through_retained_and_intent_verified(): void + { + $client = new CommandRequestIdFakeClient(self::signalReceipt([ + 'request_id' => 'retry-1', + 'retained' => false, + 'intent_verified' => true, + ])); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + '--request-id' => 'retry-1', + '--json' => true, + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + + $decoded = $this->json($tester); + + self::assertSame('retry-1', $decoded['request_id']); + self::assertFalse($decoded['retained']); + self::assertTrue($decoded['intent_verified']); + self::assertSame('default', $decoded['namespace']); + self::assertSame('signal_received', $decoded['outcome']); + } + + public function test_signal_json_output_includes_the_generated_key_when_an_older_server_omits_it(): void + { + $client = new CommandRequestIdFakeClient(self::signalReceipt()); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + '--json' => true, + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + + $decoded = $this->json($tester); + + self::assertSame($client->lastPostBody['request_id'], $decoded['request_id']); + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $decoded['request_id']); + self::assertArrayNotHasKey('retained', $decoded, 'The CLI must not fabricate receipt flags an older server did not report.'); + self::assertArrayNotHasKey('intent_verified', $decoded); + } + + public function test_update_output_passes_through_retained_and_intent_verified(): void + { + $receipt = self::updateReceipt([ + 'request_id' => 'quota-7', + 'retained' => true, + 'intent_verified' => true, + ]); + + $human = $this->runUpdate( + new CommandRequestIdFakeClient($receipt, contract: self::requestContract()), + ['workflow-id' => 'wf-1', 'update-name' => 'approve', '--request-id' => 'quota-7'], + ); + + self::assertSame(Command::SUCCESS, $human->getStatusCode()); + self::assertStringContainsString('Request ID: quota-7', $human->getDisplay()); + self::assertStringContainsString('Retained: yes', $human->getDisplay()); + self::assertStringContainsString('Intent Verified: yes', $human->getDisplay()); + + $json = $this->runUpdate( + new CommandRequestIdFakeClient($receipt, contract: self::requestContract()), + ['workflow-id' => 'wf-1', 'update-name' => 'approve', '--request-id' => 'quota-7', '--json' => true], + ); + + self::assertSame(Command::SUCCESS, $json->getStatusCode()); + + $decoded = $this->json($json); + + self::assertSame('quota-7', $decoded['request_id']); + self::assertTrue($decoded['retained']); + self::assertTrue($decoded['intent_verified']); + self::assertTrue($decoded['update_diagnostics']['retained']); + self::assertTrue($decoded['update_diagnostics']['intent_verified']); + self::assertContains('workflow:update.retained', $decoded['cli_fields']['fields_present']); + self::assertContains('workflow:update.intent_verified', $decoded['cli_fields']['fields_present']); + } + + public function test_signal_request_id_conflict_names_the_key_and_the_retained_command_in_human_output(): void + { + $body = self::conflictBody('signal', 'approve-1'); + $client = new CommandRequestIdFakeClient([], exception: new ServerHttpException( + 'Server error: Request id already identifies a different signal.', + 409, + body: $body, + )); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'reject', + '--request-id' => 'approve-1', + '--input' => '[{"approved":false}]', + ]); + + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + self::assertNotSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertSame('approve-1', $client->lastPostBody['request_id']); + + $display = $tester->getDisplay(); + + self::assertStringContainsString('Request id already identifies a different signal.', $display); + self::assertStringContainsString('Request ID: approve-1', $display); + self::assertStringContainsString('Reason: request_id_conflict', $display); + self::assertStringContainsString('Retained command: id cmd-retained-1, type signal, outcome signal_received', $display); + self::assertStringContainsString('Conflicting parts: name, arguments', $display); + self::assertStringContainsString('Next steps:', $display); + self::assertStringContainsString('Request id [approve-1] already identifies a signal command with a different name/arguments.', $display); + self::assertStringContainsString('pass a new --request-id', $display); + } + + public function test_signal_request_id_conflict_json_error_is_structured(): void + { + $body = self::conflictBody('signal', 'approve-1'); + $client = new CommandRequestIdFakeClient([], exception: new ServerHttpException( + 'Server error: Request id already identifies a different signal.', + 409, + body: $body, + )); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'reject', + '--request-id' => 'approve-1', + '--json' => true, + ]); + + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + + $decoded = $this->json($tester); + + self::assertSame('Server error: Request id already identifies a different signal.', $decoded['error']); + self::assertSame(ExitCode::INVALID, $decoded['exit_code']); + self::assertSame('workflow:signal', $decoded['command']); + self::assertSame(409, $decoded['status_code']); + self::assertSame('request_id_conflict', $decoded['reason']); + self::assertSame('approve-1', $decoded['request_id']); + self::assertSame('cmd-retained-1', $decoded['retained_command_id']); + self::assertSame('signal', $decoded['retained_command_type']); + self::assertSame('signal_received', $decoded['retained_outcome']); + self::assertSame(['name', 'arguments'], $decoded['conflict']); + self::assertSame('wf-1', $decoded['workflow_id']); + self::assertSame('reject', $decoded['signal_name']); + self::assertSame($body, $decoded['server_response']); + self::assertSame(['request_id.conflict'], array_column($decoded['recommendations'], 'id')); + } + + public function test_update_request_id_conflict_json_error_is_structured(): void + { + $body = self::conflictBody('update', 'quota-7', ['arguments', 'target']); + $client = new CommandRequestIdFakeClient([], exception: new ServerHttpException( + 'Server error: Request id already identifies a different update.', + 409, + body: $body, + ), contract: self::requestContract()); + + $tester = $this->runUpdate($client, [ + 'workflow-id' => 'wf-1', + 'update-name' => 'approve', + '--run-id' => 'run-2', + '--request-id' => 'quota-7', + '--input' => '[{"by":20}]', + '--json' => true, + ]); + + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + self::assertSame('/workflows/wf-1/runs/run-2/update/approve', $client->lastPostPath); + + $decoded = $this->json($tester); + + self::assertSame(409, $decoded['status_code']); + self::assertSame('request_id_conflict', $decoded['reason']); + self::assertSame('quota-7', $decoded['request_id']); + self::assertSame('cmd-retained-1', $decoded['retained_command_id']); + self::assertSame('update', $decoded['retained_command_type']); + self::assertSame('update_completed', $decoded['retained_outcome']); + self::assertSame(['arguments', 'target'], $decoded['conflict']); + self::assertSame('wf-1', $decoded['workflow_id']); + self::assertSame('run-2', $decoded['run_id']); + self::assertSame('approve', $decoded['update_name']); + self::assertSame('refused', $decoded['state']); + self::assertSame('update_refused', $decoded['outcome']); + self::assertSame('quota-7', $decoded['request']['request_id']); + self::assertSame([['by' => 20]], $decoded['request']['input']); + self::assertSame(409, $decoded['update_diagnostics']['error']['status_code']); + self::assertSame('request_id_conflict', $decoded['update_diagnostics']['error']['reason']); + self::assertSame('quota-7', $decoded['cli_fields']['request_id']); + self::assertSame($body, $decoded['server_response']); + self::assertSame(['request_id.conflict'], array_column($decoded['recommendations'], 'id')); + } + + public function test_signal_transport_failure_prints_the_key_that_was_sent(): void + { + $client = new CommandRequestIdFakeClient([], exception: new NetworkException('Server unreachable: connection reset')); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + ]); + + self::assertSame(ExitCode::NETWORK, $tester->getStatusCode()); + + $key = $client->lastPostBody['request_id'] ?? null; + self::assertIsString($key); + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $key); + + $display = $tester->getDisplay(); + + self::assertStringContainsString('Server unreachable: connection reset', $display); + self::assertStringContainsString('Request ID: '.$key, $display); + self::assertStringContainsString('--request-id='.$key, $display); + } + + public function test_update_timeout_json_error_includes_the_key_that_was_sent(): void + { + $client = new CommandRequestIdFakeClient( + [], + exception: new TimeoutException('Request timed out: idle timeout reached'), + contract: self::requestContract(), + ); + + $tester = $this->runUpdate($client, [ + 'workflow-id' => 'wf-1', + 'update-name' => 'approve', + '--wait' => 'completed', + '--json' => true, + ]); + + self::assertSame(ExitCode::TIMEOUT, $tester->getStatusCode()); + + $key = $client->lastPostBody['request_id'] ?? null; + self::assertIsString($key); + self::assertMatchesRegularExpression(self::UPDATE_KEY_PATTERN, $key); + + $decoded = $this->json($tester); + + self::assertSame(ExitCode::TIMEOUT, $decoded['exit_code']); + self::assertSame($key, $decoded['request_id']); + self::assertSame($key, $decoded['request']['request_id']); + self::assertSame($key, $decoded['cli_fields']['request_id']); + + $retry = array_values(array_filter( + $decoded['recommendations'], + static fn (array $recommendation): bool => $recommendation['id'] === 'request_id.retry', + )); + self::assertCount(1, $retry); + self::assertStringContainsString('--request-id='.$key, $retry[0]['message']); + } + + public function test_update_server_error_json_includes_the_retry_key(): void + { + $client = new CommandRequestIdFakeClient( + [], + exception: new ServerHttpException('Server error: HTTP 502', 502, body: null), + contract: self::requestContract(), + ); + + $tester = $this->runUpdate($client, [ + 'workflow-id' => 'wf-1', + 'update-name' => 'approve', + '--request-id' => 'quota-7', + '--json' => true, + ]); + + self::assertSame(ExitCode::SERVER, $tester->getStatusCode()); + + $decoded = $this->json($tester); + + self::assertSame('quota-7', $decoded['request_id']); + self::assertContains('request_id.retry', array_column($decoded['recommendations'], 'id')); + } + + public function test_request_id_longer_than_255_characters_is_rejected_before_sending(): void + { + $client = new CommandRequestIdFakeClient(self::signalReceipt()); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + '--request-id' => str_repeat('k', 256), + ]); + + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + self::assertSame(0, $client->postCalls); + self::assertStringContainsString('--request-id must be 1 to 255 Unicode characters; got 256 characters.', $tester->getDisplay()); + + $longest = str_repeat('k', 255); + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + '--request-id' => $longest, + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertSame($longest, $client->lastPostBody['request_id']); + } + + public function test_the_published_server_accepts_255_unicode_characters_as_one_key(): void + { + $key = str_repeat('é', 255); + $client = new CommandRequestIdFakeClient(self::signalReceipt()); + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', 'signal-name' => 'approve', '--request-id' => $key, + ]); + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertSame($key, $client->lastPostBody['request_id']); + } + + public function test_blank_request_id_is_treated_as_omitted(): void + { + $client = new CommandRequestIdFakeClient(self::signalReceipt()); + + $tester = $this->runSignal($client, [ + 'workflow-id' => 'wf-1', + 'signal-name' => 'approve', + '--request-id' => ' ', + ]); + + self::assertSame(Command::SUCCESS, $tester->getStatusCode()); + self::assertMatchesRegularExpression(self::SIGNAL_KEY_PATTERN, $client->lastPostBody['request_id']); + } + + public function test_signal_and_update_send_the_key_in_the_body_and_never_as_an_x_request_id_header(): void + { + /** @var list}> $requests */ + $requests = []; + + $http = new MockHttpClient(static function (string $method, string $url, array $options) use (&$requests): MockResponse { + if (str_ends_with($url, '/api/cluster/info')) { + return new MockResponse(json_encode([ + 'version' => '0.2.210', + 'control_plane' => [ + 'version' => ServerClient::CONTROL_PLANE_VERSION, + 'request_contract' => [ + 'schema' => ControlPlaneRequestContract::SCHEMA, + 'version' => ControlPlaneRequestContract::VERSION, + 'operations' => [ + 'update' => [ + 'fields' => [ + 'wait_for' => ['canonical_values' => ['accepted', 'completed']], + ], + ], + ], + ], + ], + ], JSON_THROW_ON_ERROR), ['http_code' => 200]); + } + + $requests[] = ['url' => $url, 'options' => $options]; + + $operation = str_contains($url, '/update/') ? 'update' : 'signal'; + + return new MockResponse(json_encode(self::controlPlaneEnvelope($operation), JSON_THROW_ON_ERROR), [ + 'http_code' => 202, + 'response_headers' => [ + ServerClient::CONTROL_PLANE_HEADER.': '.ServerClient::CONTROL_PLANE_VERSION, + ], + ]); + }); + + $client = new ServerClient( + baseUrl: 'http://example.test', + namespace: 'default', + http: $http, + ); + + $signal = $this->runSignal($client, [ + 'workflow-id' => 'wf-123', + 'signal-name' => 'advance', + '--request-id' => 'signal-key-1', + '--json' => true, + ]); + self::assertSame(Command::SUCCESS, $signal->getStatusCode(), $signal->getDisplay()); + + $update = $this->runUpdate($client, [ + 'workflow-id' => 'wf-123', + 'update-name' => 'advance', + '--json' => true, + ]); + self::assertSame(Command::SUCCESS, $update->getStatusCode(), $update->getDisplay()); + + self::assertCount(2, $requests); + self::assertStringEndsWith('/api/workflows/wf-123/signal/advance', $requests[0]['url']); + self::assertStringEndsWith('/api/workflows/wf-123/update/advance', $requests[1]['url']); + + $updateKey = $this->json($update)['request_id']; + self::assertMatchesRegularExpression(self::UPDATE_KEY_PATTERN, $updateKey); + + foreach ([['signal-key-1', $requests[0]], [$updateKey, $requests[1]]] as [$expectedKey, $request]) { + self::assertSame($expectedKey, self::requestBody($request['options'])['request_id'] ?? null); + self::assertNull( + self::headerValue($request['options'], 'X-Request-Id'), + 'The command key travels only in the request body; X-Request-Id stays correlation-only.', + ); + self::assertSame('default', self::headerValue($request['options'], 'X-Namespace')); + } + + self::assertSame('signal-key-1', $this->json($signal)['request_id']); + } + + /** + * @param array $arguments + */ + private function runSignal(ServerClient $client, array $arguments): CommandTester + { + $command = new SignalCommand(); + $command->setServerClient($client); + + $tester = new CommandTester($command); + $tester->execute($arguments); + + return $tester; + } + + /** + * @param array $arguments + */ + private function runUpdate(ServerClient $client, array $arguments): CommandTester + { + $command = new UpdateCommand(); + $command->setServerClient($client); + + $tester = new CommandTester($command); + $tester->execute($arguments); + + return $tester; + } + + /** + * @return array + */ + private function json(CommandTester $tester): array + { + $decoded = json_decode(trim($tester->getDisplay()), true, flags: JSON_THROW_ON_ERROR); + self::assertIsArray($decoded); + + return $decoded; + } + + /** + * @param array $overrides + * @return array + */ + private static function signalReceipt(array $overrides = []): array + { + return array_replace([ + 'workflow_id' => 'wf-1', + 'signal_name' => 'approve', + 'outcome' => 'signal_received', + 'command_status' => 'accepted', + 'command_id' => 'cmd-1', + ], $overrides); + } + + /** + * @param array $overrides + * @return array + */ + private static function updateReceipt(array $overrides = []): array + { + return array_replace([ + 'workflow_id' => 'wf-1', + 'update_name' => 'approve', + 'update_id' => 'upd-1', + 'outcome' => 'update_accepted', + 'command_status' => 'accepted', + 'update_status' => 'accepted', + 'wait_for' => 'accepted', + ], $overrides); + } + + /** + * @param list $conflict + * @return array + */ + private static function conflictBody(string $commandType, string $requestId, array $conflict = ['name', 'arguments']): array + { + return [ + 'message' => sprintf('Request id already identifies a different %s.', $commandType), + 'reason' => 'request_id_conflict', + 'request_id' => $requestId, + 'retained_command_id' => 'cmd-retained-1', + 'retained_command_type' => $commandType, + 'retained_outcome' => $commandType === 'signal' ? 'signal_received' : 'update_completed', + 'conflict' => $conflict, + ]; + } + + private static function requestContract(): ControlPlaneRequestContract + { + return new ControlPlaneRequestContract([ + 'update' => [ + 'fields' => [ + 'wait_for' => [ + 'canonical_values' => ['accepted', 'completed'], + ], + ], + ], + ]); + } + + /** + * @return array + */ + private static function controlPlaneEnvelope(string $operation): array + { + $body = [ + 'control_plane' => [ + 'schema' => 'durable-workflow.v2.control-plane-response', + 'version' => 1, + 'operation' => $operation, + 'operation_name' => 'advance', + 'operation_name_field' => $operation.'_name', + 'workflow_id' => 'wf-123', + 'outcome' => $operation.'_accepted', + 'contract' => [ + 'schema' => 'durable-workflow.v2.control-plane-response.contract', + 'version' => 1, + 'legacy_field_policy' => 'reject_non_canonical', + 'legacy_fields' => [ + 'query' => 'query_name', + 'signal' => 'signal_name', + 'update' => 'update_name', + 'wait_policy' => 'wait_for', + ], + 'required_fields' => ['workflow_id', 'operation_name', 'operation_name_field'], + 'success_fields' => ['outcome'], + ], + ], + ]; + + if ($operation === 'update') { + $body['update_id'] = 'upd-123'; + $body['update_status'] = 'accepted'; + } + + return $body; + } + + /** + * @param array $options + * @return array + */ + private static function requestBody(array $options): array + { + $body = $options['body'] ?? null; + + if (is_string($body)) { + $decoded = json_decode($body, true, flags: JSON_THROW_ON_ERROR); + self::assertIsArray($decoded); + + return $decoded; + } + + $json = $options['json'] ?? null; + self::assertIsArray($json, 'The mocked request carried neither a JSON body string nor a json option.'); + + return $json; + } + + /** + * @param array $options + */ + private static function headerValue(array $options, string $wantedHeader): ?string + { + foreach (['headers', 'normalized_headers'] as $optionKey) { + foreach (($options[$optionKey] ?? []) as $name => $value) { + if (is_string($name) && strcasecmp($name, $wantedHeader) === 0) { + $value = is_array($value) ? ($value[0] ?? null) : $value; + + return is_string($value) ? trim((string) preg_replace('/^[^:]+:\s*/', '', $value)) : null; + } + + foreach (is_array($value) ? $value : [$value] as $line) { + if (is_string($line) && stripos($line, $wantedHeader.':') === 0) { + return trim(substr($line, strlen($wantedHeader) + 1)); + } + } + } + } + + return null; + } +} + +final class CommandRequestIdFakeClient extends ServerClient +{ + /** + * @var array + */ + public array $lastPostBody = []; + + public string $lastPostPath = ''; + + public int $postCalls = 0; + + /** + * @param array $payload + */ + public function __construct( + private readonly array $payload, + private readonly ?\Throwable $exception = null, + private readonly ?ControlPlaneRequestContract $contract = null, + ) { + } + + public function post(string $path, array $body = []): array + { + $this->postCalls++; + $this->lastPostPath = $path; + $this->lastPostBody = $body; + + if ($this->exception !== null) { + throw $this->exception; + } + + return $this->payload; + } + + public function controlPlaneRequestContract(): ?ControlPlaneRequestContract + { + return $this->contract; + } +} diff --git a/tests/Commands/WorkflowControlPlaneCommandTest.php b/tests/Commands/WorkflowControlPlaneCommandTest.php index 6f82bc7..a41e975 100644 --- a/tests/Commands/WorkflowControlPlaneCommandTest.php +++ b/tests/Commands/WorkflowControlPlaneCommandTest.php @@ -66,14 +66,19 @@ public function test_update_command_sends_wait_for_and_renders_the_canonical_upd '--wait' => 'completed', ])); - self::assertSame([ - 'wait_for' => 'completed', - ], $client->lastPostBody); + self::assertSame('completed', $client->lastPostBody['wait_for'] ?? null); + self::assertMatchesRegularExpression( + '/\Acli-update-[0-9a-f]{32}\z/', + $client->lastPostBody['request_id'] ?? '', + 'workflow:update must send a generated durable command key when --request-id is omitted.', + ); + self::assertSame(['wait_for', 'request_id'], array_keys($client->lastPostBody)); $display = $tester->getDisplay(); self::assertStringContainsString('Update: approve', $display); self::assertStringContainsString('Wait For: completed', $display); + self::assertStringContainsString('Request ID: '.$client->lastPostBody['request_id'], $display); } public function test_update_command_rejects_invalid_wait_values_from_the_server_contract(): void diff --git a/tests/fixtures/control-plane/workflow-signal-parity.json b/tests/fixtures/control-plane/workflow-signal-parity.json index 9c83ee9..83eb702 100644 --- a/tests/fixtures/control-plane/workflow-signal-parity.json +++ b/tests/fixtures/control-plane/workflow-signal-parity.json @@ -21,9 +21,11 @@ "argv": { "workflow-id": "wf-polyglot-231", "signal-name": "approval.received", + "--request-id": "cli-signal-polyglot-231", "--input": "[{\"approved_by\":\"ada\",\"approved\":true,\"notes\":\"ship it\"}]" }, "expected_body": { + "request_id": "cli-signal-polyglot-231", "input": [ { "approved_by": "ada", diff --git a/tests/fixtures/control-plane/workflow-update-parity.json b/tests/fixtures/control-plane/workflow-update-parity.json index 136a98c..d00d444 100644 --- a/tests/fixtures/control-plane/workflow-update-parity.json +++ b/tests/fixtures/control-plane/workflow-update-parity.json @@ -22,10 +22,12 @@ "workflow-id": "wf-polyglot-231", "update-name": "quota.adjust", "--wait": "completed", + "--request-id": "cli-update-polyglot-231", "--input": "[{\"quota\":50,\"reason\":\"gold-tier\"}]" }, "expected_body": { "wait_for": "completed", + "request_id": "cli-update-polyglot-231", "input": [ { "quota": 50,