diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 4d840c7..51ceea8 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -212,7 +212,9 @@ jobs: - name: Fetch the pinned 2.x reference CLI run: parity/fetch-reference.sh - name: Compare against the 2.x CLI - run: python3 parity/parity.py --verbose + run: | + python3 -m unittest discover -s parity -p 'test_*.py' + python3 parity/parity.py --verbose target-branch-qualification: name: Target branch qualification diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 568441d..baf1d75 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -487,6 +487,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",...} @@ -497,6 +498,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"}' @@ -514,6 +516,31 @@ dw workflow:history order-123 01HXYZ dw workflow:history order-123 01HXYZ --follow ``` +`workflow:signal` and `workflow:update` send a caller command key as the +`request_id` body field. The CLI generates `cli-signal-` or +`cli-update-` once per invocation only when `--request-id` is omitted. +Explicit empty or whitespace-only keys are refused before POST. Keys contain +1-255 Unicode code points (up to 1020 UTF-8 bytes), without normalization, +truncation or case conversion. The key is shown safely in human output and +preserved exactly in JSON/JSONL, including after a timeout or network failure. + +After a lost response the admission outcome is unknown. Inspect workflow +history and command receipts before retrying; a key alone does not establish +retry safety. Rely on the complete retained-admission and canonical-intent +guarantee only when the Server advertises that published capability. Ordinary +request-key storage and partial update receipt lookup do not supply the full +signal/update guarantee. No currently qualified published Server tuple +advertises that complete capability. + +When the Server reports receipt metadata, `retained: true` identifies an +original receipt and `intent_verified` reports whether its intent was +verified. Preserve its original status and exact target. A qualified +retained-admission contract may refuse a different typed intent with HTTP 409 +and `reason: request_id_conflict`; the response identifies the retained +command and conflict parts. Legacy receipts with unverified intent never +authorize a new command. Deduplicated admission does not make external effects +exactly once. + Batch cancellation with `--all-matching` cancels the exact run each listed workflow had when it was selected. If that run has continued as new, been replaced or closed before the request reaches Server, the workflow is reported diff --git a/rust/README.md b/rust/README.md index 6bfb69b..ca05bef 100644 --- a/rust/README.md +++ b/rust/README.md @@ -59,6 +59,10 @@ CLI and once with this build, each in a fresh config directory. It compares: - stderr as normalized text (the same normalizations), plus its category; - the config file each case leaves behind. +The manifest case records exact reference and candidate hashes for reviewed +additive schema changes. Only those two values at the named command's `sha256` +field are normalized. Other hashes, manifest metadata and formatting still compare. + ```bash parity/fetch-reference.sh # pinned 2.x CLI, checksum-verified, into target/ cargo build diff --git a/rust/parity/cases.json b/rust/parity/cases.json index f026a9a..fd26e40 100644 --- a/rust/parity/cases.json +++ b/rust/parity/cases.json @@ -1798,7 +1798,17 @@ "name": "schema:manifest", "argv": [ "schema:manifest" - ] + ], + "schema_sha256_updates": { + "workflow:signal": { + "reference": "sha256:c42c02f7a54faa082982028482bc231ca29c495a2a00fabce5b0ae580a2505bf", + "candidate": "sha256:5a5fe23d6a1bead8508887b57b1dfb58448b92ed4292ce0da88bcdd1abf37e81" + }, + "workflow:update": { + "reference": "sha256:a3997a4b623d84b166ea3a124b0a33e5c400acc51694332186813b96aa389eb1", + "candidate": "sha256:8888b18ff8d8ef7cac35f2aae362192b04617dc28760c25ddccf085880c3e25c" + } + } }, { "name": "schema:show a command's JSON schema", diff --git a/rust/parity/parity.py b/rust/parity/parity.py index 0d133b5..f4265cd 100755 --- a/rust/parity/parity.py +++ b/rust/parity/parity.py @@ -209,11 +209,41 @@ def prepare_project(config_dir, options, env): return project +def normalize_schema_manifest(text, case, side): + """Allow only reviewed, directional schema hash updates; retain all other bytes.""" + if case.get("argv") != ["schema:manifest"] or not isinstance(text, str): + return text + try: + manifest = json.loads(text) + except ValueError: + return text + if (not isinstance(manifest, dict) + or manifest.get("schema") != "durable-workflow.cli.output-schema-manifest" + or manifest.get("version") != 3 + or not isinstance(manifest.get("commands"), dict)): + return text + for command, update in case.get("schema_sha256_updates", {}).items(): + digest = update[side] + entry = manifest["commands"].get(command) + if (isinstance(entry, dict) and entry.get("sha256") == digest + and text.count(digest) == 1): + text = text.replace(digest, f"") + return text + + def compare(reference, candidate, case): fields = ["exit_code", "stderr_category", "stderr", "stdout_json", "config"] if case.get("compare_text", True): fields.append("stdout_text") - return [field for field in fields if reference[field] != candidate[field]] + differences = [] + for field in fields: + left, right = reference[field], candidate[field] + if field == "stdout_text" and case.get("schema_sha256_updates"): + left = normalize_schema_manifest(left, case, "reference") + right = normalize_schema_manifest(right, case, "candidate") + if left != right: + differences.append(field) + return differences def patched_cluster_info(patches): diff --git a/rust/parity/test_parity.py b/rust/parity/test_parity.py new file mode 100644 index 0000000..8178fd4 --- /dev/null +++ b/rust/parity/test_parity.py @@ -0,0 +1,85 @@ +import copy +import hashlib +import json +from pathlib import Path +import unittest + +import parity + + +class SchemaManifestParityTest(unittest.TestCase): + def setUp(self): + with open(parity.CASES, encoding="utf-8") as handle: + self.case = next(case for case in json.load(handle)["cases"] + if case["name"] == "schema:manifest") + self.root = Path(parity.CRATE).parent + self.candidate = (self.root / "schemas/output/manifest.json").read_text(encoding="utf-8") + self.reference = self.candidate + for update in self.case["schema_sha256_updates"].values(): + self.reference = self.reference.replace(update["candidate"], update["reference"]) + + def differences(self, reference=None, candidate=None, case=None): + result = {key: None for key in ("exit_code", "stderr_category", "stderr", "stdout_json", "config")} + return parity.compare( + dict(result, stdout_text=self.reference if reference is None else reference), + dict(result, stdout_text=self.candidate if candidate is None else candidate), + self.case if case is None else case, + ) + + def test_exact_reviewed_schema_hash_updates_match(self): + self.assertEqual([], self.differences()) + manifest = json.loads(self.candidate) + for command, update in self.case["schema_sha256_updates"].items(): + entry = manifest["commands"][command] + digest = hashlib.sha256((self.root / entry["schema"]).read_bytes()).hexdigest() + self.assertEqual(update["candidate"], "sha256:" + digest) + + def test_unreviewed_hash_or_old_candidate_hash_is_a_difference(self): + for update in self.case["schema_sha256_updates"].values(): + for replacement in ("sha256:" + "0" * 64, update["reference"]): + with self.subTest(replacement=replacement): + self.assertEqual(["stdout_text"], self.differences( + candidate=self.candidate.replace(update["candidate"], replacement))) + self.assertEqual(["stdout_text"], self.differences( + reference=self.reference.replace(update["reference"], update["candidate"]))) + + def test_other_manifest_bytes_remain_significant(self): + for old, new in (("patch-stable-additive", "changed"), + ('"version": 3', '"version": 4'), + ("workflow-signal.schema.json", "wrong.schema.json"), + ('"output": "--json"', '"output": "--wrong"'), + (' "commands": {', ' "commands": {')): + with self.subTest(old=old): + self.assertIn(old, self.candidate) + self.assertEqual(["stdout_text"], self.differences( + candidate=self.candidate.replace(old, new, 1))) + + def test_hash_in_wrong_field_or_repeated_elsewhere_does_not_match(self): + update = self.case["schema_sha256_updates"]["workflow:signal"] + for candidate in ( + self.candidate.replace('"sha256": "' + update["candidate"], + '"wrong": "' + update["candidate"]), + self.candidate.replace('"stability": "patch-stable-additive"', + '"stability": "' + update["candidate"] + '"'), + ): + self.assertEqual(["stdout_text"], self.differences(candidate=candidate)) + + def test_unrelated_hash_changes_still_fail(self): + digest = json.loads(self.candidate)["commands"]["workflow:list"]["sha256"] + self.assertEqual(["stdout_text"], self.differences( + candidate=self.candidate.replace(digest, "sha256:" + "0" * 64))) + + def test_other_cases_cannot_enable_manifest_hash_normalization(self): + case = copy.deepcopy(self.case) + case["argv"] = ["schema:show", "workflow:signal"] + self.assertEqual(["stdout_text"], self.differences(case=case)) + del case["schema_sha256_updates"] + self.assertEqual(["stdout_text"], self.differences(case=case)) + + def test_invalid_manifest_is_a_difference(self): + for candidate in (self.candidate + "invalid", "[]", "null"): + self.assertEqual(["stdout_text"], self.differences(candidate=candidate)) + + +if __name__ == "__main__": + unittest.main() 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..885f1d3 100644 --- a/src/Commands/BaseCommand.php +++ b/src/Commands/BaseCommand.php @@ -20,6 +20,7 @@ use DurableWorkflow\Cli\Support\TimeoutException; use Symfony\Component\Console\Command\Command; use Symfony\Component\Console\Exception\ExceptionInterface as ConsoleException; +use Symfony\Component\Console\Formatter\OutputFormatter; use Symfony\Component\Console\Input\InputInterface; use Symfony\Component\Console\Input\InputOption; use Symfony\Component\Console\Output\ConsoleOutputInterface; @@ -29,6 +30,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 +127,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); @@ -397,10 +408,10 @@ protected function validateControlPlaneOption( value: $value, optionName: $optionName, ); + } catch (ServerException|InvalidOptionException $exception) { + throw $exception; } catch (\RuntimeException $exception) { - $this->writeHumanError($output, $exception->getMessage()); - - return false; + throw new InvalidOptionException($exception->getMessage(), 0, $exception); } return true; @@ -582,7 +593,7 @@ private function inputEncoding(InputInterface $input): string */ protected function renderJson(OutputInterface $output, array $data): int { - $output->writeln(json_encode($data, JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); + $output->writeln(json_encode($data, JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR), OutputInterface::OUTPUT_RAW); return Command::SUCCESS; } @@ -607,7 +618,7 @@ protected function renderJsonList( if ($this->outputMode($input) === OutputMode::JSONL) { $items = $envelope[$itemsKey] ?? []; foreach ($items as $item) { - $output->writeln(json_encode($item, JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); + $output->writeln(json_encode($item, JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR), OutputInterface::OUTPUT_RAW); } return Command::SUCCESS; @@ -663,6 +674,75 @@ 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 ($explicit !== null || $input->hasParameterOption('--request-id', true)) { + if (! is_scalar($explicit)) { + throw new InvalidOptionException('--request-id must not be empty or whitespace-only.'); + } + $explicit = (string) $explicit; + + $characters = preg_match_all('/./us', $explicit); + if ($characters === false) { + throw new InvalidOptionException('--request-id must be valid UTF-8.'); + } + if ($this->isBlankCommandRequestId($explicit)) { + throw new InvalidOptionException('--request-id must not be empty or whitespace-only.'); + } + 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))); + } + + private function isBlankCommandRequestId(string $value): bool + { + return trim($value) === '' || preg_match('/\A[\s\p{Z}]*\z/u', $value) === 1; + } + + /** + * 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: '.$this->humanOutputText($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 +951,8 @@ private function workflowUpdateDiagnosticEnvelope(array $payload): array 'update_name', 'update_id', 'request_id', + 'retained', + 'intent_verified', 'outcome', 'reason', 'rejection_reason', @@ -926,6 +1008,8 @@ private function workflowUpdateCliFields(array $payload): array 'update_name', 'update_id', 'request_id', + 'retained', + 'intent_verified', 'state', 'update_state', 'outcome', @@ -1136,13 +1220,14 @@ private function emitError(InputInterface $input, OutputInterface $output, \Thro } if ($jsonRequested) { - $output->writeln($this->encodeErrorEnvelope($e, $exitCode, $input)); + $output->writeln($this->encodeErrorEnvelope($e, $exitCode, $input), OutputInterface::OUTPUT_RAW); return $exitCode; } $this->writeHumanError($output, $e->getMessage()); $this->writeHumanErrorNamespaceContext($output, $input); + $this->writeHumanErrorRequestIdContext($output, $input); $this->writeHumanHttpErrorDetails($output, $e); $recommendations = $this->errorRecommendations($e, $input); @@ -1247,6 +1332,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', @@ -1295,7 +1386,12 @@ private function copyServerResponseDiagnostics(array &$envelope, array $body): v private function writeHumanError(OutputInterface $output, string $message): void { $target = $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output; - $target->writeln(''.$message.''); + $target->writeln(''.$this->humanOutputText($message).''); + } + + private function humanOutputText(string $text): string + { + return OutputFormatter::escape(preg_replace('/[\x00-\x1F\x7F\x{80}-\x{9F}]/u', ' ', $text) ?? '-'); } private function writeHumanErrorNamespaceContext(OutputInterface $output, InputInterface $input): void @@ -1309,6 +1405,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: '.$this->humanOutputText($requestId)); + } + private function writeHumanHttpErrorDetails(OutputInterface $output, \Throwable $e): void { if (! $e instanceof ServerHttpException) { @@ -1318,17 +1425,10 @@ private function writeHumanHttpErrorDetails(OutputInterface $output, \Throwable $target = $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output; if ($e->reason() !== null) { - $target->writeln('Reason: '.$e->reason()); + $target->writeln('Reason: '.$this->humanOutputText($e->reason())); } - if ($e->body !== null) { - foreach (['code', 'error_code', 'type', 'rejection_reason'] as $key) { - $value = $e->body[$key] ?? null; - if (is_string($value) && $value !== '') { - $target->writeln(str_replace('_', ' ', ucfirst($key)).': '.$value); - } - } - } + $this->writeHumanHttpErrorBody($target, $e); $validationErrors = $e->validationErrors(); if ($validationErrors === null) { @@ -1336,7 +1436,27 @@ private function writeHumanHttpErrorDetails(OutputInterface $output, \Throwable } foreach ($this->flattenValidationErrors($validationErrors) as $message) { - $target->writeln('Validation: '.$message); + $target->writeln('Validation: '.$this->humanOutputText($message)); + } + } + + private function writeHumanHttpErrorBody(OutputInterface $output, ServerHttpException $e): void + { + if ($e->body === null) { + return; + } + + foreach (['code', 'error_code', 'type', 'rejection_reason'] as $key) { + $value = $e->body[$key] ?? null; + if (is_string($value) && $value !== '') { + $output->writeln(str_replace('_', ' ', ucfirst($key)).': '.$this->humanOutputText($value)); + } + } + + if ($e->reason() === self::REQUEST_ID_CONFLICT_REASON) { + foreach ($this->requestIdConflictLines($e->body) as $line) { + $output->writeln($this->humanOutputText($line)); + } } } @@ -1382,19 +1502,19 @@ private function errorNamespaceContext(InputInterface $input): ?string private function errorRequestIdContext(InputInterface $input): ?string { - if (! $input->hasOption('request-id')) { - return null; + if ($this->commandRequestId !== null) { + return $this->commandRequestId; } - $requestId = $input->getOption('request-id'); + $requestId = $input->hasOption('request-id') ? $input->getOption('request-id') : null; if (! is_scalar($requestId)) { return null; } - $requestId = trim((string) $requestId); + $requestId = (string) $requestId; - return $requestId === '' ? null : $requestId; + return $this->isBlankCommandRequestId($requestId) || preg_match('//u', $requestId) !== 1 ? null : $requestId; } /** @@ -1420,7 +1540,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); @@ -1556,10 +1676,10 @@ private function writeHumanNextSteps(OutputInterface $output, array $recommendat $target->writeln('Next steps:'); foreach ($recommendations as $recommendation) { - $target->writeln(' - '.$recommendation['message']); + $target->writeln(' - '.$this->humanOutputText($recommendation['message'])); if (isset($recommendation['command'])) { - $target->writeln(' Try: '.$recommendation['command']); + $target->writeln(' Try: '.$this->humanOutputText($recommendation['command'])); } } } @@ -1575,47 +1695,129 @@ private function writeHumanWarning(OutputInterface $output, string $message): vo */ private function errorRecommendations(\Throwable $e, ?InputInterface $input): array { - if ($e instanceof NetworkException) { - return [[ + $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), default to inspection. A transport + * failure does not establish the Server's retained-admission capability. + * + * @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' => 'The outcome of this command is unknown. Inspect workflow history and command receipts before retrying. Keep the exact request ID; use --json for an exact copy. Retry with the same key only if the Server advertises retained command admission; a request ID alone does not prevent duplicates.', + ]; + } + + /** + * @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 + { + return match (true) { + $e instanceof NetworkException => [[ 'id' => 'server.unreachable', 'severity' => 'error', 'message' => 'Check the server URL, DNS, port, and TLS settings for the selected environment.', 'command' => $this->doctorCommand($input), - ]]; - } - - if ($e instanceof TimeoutException) { - return [[ + ]], + $e instanceof TimeoutException => [[ 'id' => 'server.timeout', 'severity' => 'error', 'message' => 'Check server load and network reachability, then retry with a smaller request or narrower query.', 'command' => $this->doctorCommand($input), - ]]; - } - - if ($e instanceof ServerHttpException) { - return $this->httpErrorRecommendations($e, $input); - } - - if ($e instanceof CompatibilityException) { - return [[ + ]], + $e instanceof ServerHttpException => $this->httpErrorRecommendations($e, $input), + $e instanceof CompatibilityException => [[ 'id' => 'compatibility.unsupported', 'severity' => 'error', 'message' => $e->nextStep(), 'command' => $this->doctorCommand($input), - ]]; - } - - if ($e instanceof InvalidOptionException) { - return [[ + ]], + $e instanceof InvalidOptionException => [[ 'id' => 'input.invalid', 'severity' => 'error', 'message' => 'Review the command usage and option values before retrying.', 'command' => sprintf('dw %s --help', (string) $this->getName()), - ]]; - } - - return []; + ]], + default => [], + }; } /** @@ -1641,6 +1843,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..2c174d6 100644 --- a/src/Commands/WorkflowCommand/SignalCommand.php +++ b/src/Commands/WorkflowCommand/SignalCommand.php @@ -23,14 +23,25 @@ protected function configure(): void durably recorded in history even if the workflow is currently idle or waiting. +Every signal carries a caller command key (request_id). +The CLI generates one per invocation unless --request-id +is given, and prints the key it used. An explicit empty or whitespace-only +key is refused. After a lost response the outcome is unknown: inspect +workflow history and command receipts before retrying. A request ID alone +does not prove that a repeated signal is safe. Rely on retained admission +and intent-conflict guarantees only when the Server advertises that +capability; legacy servers may store a key without deduplicating signals. + 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, 'Caller command key (1-255 Unicode code points); generated only when omitted') ->addOption('json', null, InputOption::VALUE_NONE, 'Output the command response as JSON'); $this->addInputOptions('Signal input payload'); } @@ -40,8 +51,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 +67,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 +88,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..1f0b460 100644 --- a/src/Commands/WorkflowCommand/UpdateCommand.php +++ b/src/Commands/WorkflowCommand/UpdateCommand.php @@ -25,16 +25,26 @@ protected function configure(): void validated the update; --wait=completed returns when the handler has finished and a result is available. +Every update carries a caller command key (request_id). +The CLI generates one per invocation unless --request-id +is given, and prints the key it used. An explicit empty or whitespace-only +key is refused. After a lost response the outcome is unknown: inspect +workflow history and command receipts before retrying. A request ID alone +does not establish the complete retry guarantee. Rely on retained admission +and intent-conflict guarantees only when the Server advertises that +capability. Preserve any original receipt and its reported status. + 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, 'Caller command key (1-255 Unicode code points); generated only when omitted') ->addOption('json', null, InputOption::VALUE_NONE, 'Output the command response as JSON'); $this->addInputOptions('Update input payload'); } @@ -60,11 +70,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 +110,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/src/Support/ControlPlaneRequestContract.php b/src/Support/ControlPlaneRequestContract.php index e0fb8f1..cf32fa5 100644 --- a/src/Support/ControlPlaneRequestContract.php +++ b/src/Support/ControlPlaneRequestContract.php @@ -106,7 +106,7 @@ public function assertCanonicalValue( $rejectedAliases = $this->stringMap($definition['rejected_aliases'] ?? null); if (array_key_exists($value, $rejectedAliases)) { - throw new \RuntimeException(sprintf( + throw new InvalidOptionException(sprintf( 'Server contract rejects %s value [%s]; use [%s].', $displayField, $value, @@ -120,7 +120,7 @@ public function assertCanonicalValue( return; } - throw new \RuntimeException(sprintf( + throw new InvalidOptionException(sprintf( 'Server contract expects %s to be one of [%s]; got [%s].', $displayField, implode(', ', $canonicalValues), diff --git a/tests/Commands/OutputContractTest.php b/tests/Commands/OutputContractTest.php index 82810a3..8c185d8 100644 --- a/tests/Commands/OutputContractTest.php +++ b/tests/Commands/OutputContractTest.php @@ -29,6 +29,28 @@ */ class OutputContractTest extends TestCase { + public function test_json_and_jsonl_preserve_opaque_values_with_or_without_color(): void + { + $value = "wf-opaquelink\x1B]0;title\x07\nend"; + foreach (['json', 'jsonl'] as $mode) { + foreach ([false, true] as $decorated) { + $command = new WorkflowListCommand(); + $command->setServerClient(new OutputContractListClient([ + 'workflows' => [['workflow_id' => $value, 'status' => 'running']], + 'next_page_token' => $value, + ])); + $tester = new CommandTester($command); + self::assertSame(0, $tester->execute(['--output' => $mode], ['decorated' => $decorated])); + $display = trim($tester->getDisplay()); + self::assertSame(1, substr_count($display, "\n") + 1); + $decoded = json_decode($display, true, flags: JSON_THROW_ON_ERROR); + $item = $mode === 'json' ? $decoded['workflows'][0] : $decoded; + self::assertSame($value, $item['workflow_id']); + if ($mode === 'json') { self::assertSame($value, $decoded['next_page_token']); } + } + } + } + public function test_json_mode_emits_single_json_document_for_list_command(): void { $command = new WorkflowListCommand(); diff --git a/tests/Commands/WorkflowCommandRequestIdTest.php b/tests/Commands/WorkflowCommandRequestIdTest.php new file mode 100644 index 0000000..3615c68 --- /dev/null +++ b/tests/Commands/WorkflowCommandRequestIdTest.php @@ -0,0 +1,946 @@ + $value]) + : self::updateReceipt(['request_id' => $value]); + $body = self::conflictBody($operation, $value, ['name', $value]); + foreach (['retained_command_id', 'retained_command_type', 'retained_outcome'] as $field) { + $body[$field] = $value; + } + $message = 'Request failed for '.$value; + foreach ([false, true] as $json) { + $exception = match ($scenario) { + 'conflict' => new ServerHttpException($message, 409, body: $body), + 'uncertain' => new NetworkException($message), + default => null, + }; + $client = new CommandRequestIdFakeClient($receipt, exception: $exception, contract: self::requestContract()); + $arguments = ['workflow-id' => 'wf-1', $operation.'-name' => 'approve', '--request-id' => $value]; + if ($json) { $arguments['--json'] = true; } + $tester = $operation === 'signal' + ? $this->runSignal($client, $arguments, $decorated) + : $this->runUpdate($client, $arguments, $decorated); + self::assertSame($value, $client->lastPostBody['request_id']); + self::assertSame(match ($scenario) { + 'conflict' => ExitCode::INVALID, 'uncertain' => ExitCode::NETWORK, default => Command::SUCCESS, + }, $tester->getStatusCode()); + if ($json) { + $decoded = $this->json($tester); + self::assertSame($value, $decoded['request_id']); + if ($scenario === 'conflict') { + self::assertSame($body, $decoded['server_response']); + self::assertSame($value, $decoded['retained_command_id']); + self::assertSame($value, $decoded['retained_command_type']); + self::assertSame($value, $decoded['retained_outcome']); + } + if ($scenario !== 'success') { self::assertSame($message, $decoded['error']); } + if ($scenario === 'uncertain') { + $retry = array_values(array_filter($decoded['recommendations'], static fn (array $item): bool => $item['id'] === 'request_id.retry')); + self::assertStringNotContainsString('--request-id=', $retry[0]['message']); + } + continue; + } + // Remove the CLI's own color styling, leaving injected OSC/C1/control bytes visible. + $display = preg_replace('/\x1B\[[0-9;]*m/', '', $tester->getDisplay()); + self::assertIsString($display); + self::assertDoesNotMatchRegularExpression('/[\x00-\x09\x0B-\x1F\x7F\x{80}-\x{9F}]/u', $display); + $shown = str_replace(["\n", "\r", "\t", "\x1B", "\x07", "\u{009B}"], ' ', $value); + self::assertStringContainsString('Request ID: '.$shown, $display); + if ($scenario === 'conflict') { + self::assertStringContainsString('Retained command: id '.$shown, $display); + self::assertStringContainsString('Conflicting parts: name, '.$shown, $display); + self::assertStringContainsString('Request id ['.$shown.']', $display); + } + if ($scenario === 'uncertain') { + self::assertStringNotContainsString('--request-id=', $display); + self::assertStringContainsString('exact request ID', $display); + } + } + } + + public static function opaqueOutputCases(): iterable + { + $values = [ + 'controls' => "key-\nFORGED\r\t\x1B]0;title\x07-\u{009B}31m-end", + 'markup' => 'key-redlink-end', + 'shell' => "key-';$(id);`echo nope`\"\\-end", + ]; + foreach (['signal', 'update'] as $operation) { + foreach (['success', 'conflict', 'uncertain'] as $scenario) { + foreach ($values as $name => $value) { + foreach ([false, true] as $decorated) { + yield $operation.' '.$scenario.' '.$name.' decorated='.(int) $decorated => [$operation, $scenario, $value, $decorated]; + } + } + } + } + } + + public function test_malformed_utf8_key_is_refused_as_json_without_sending(): void + { + foreach (['signal', 'update'] as $operation) { + $client = new CommandRequestIdFakeClient([], contract: self::requestContract()); + $arguments = ['workflow-id' => 'wf-1', $operation.'-name' => 'approve', '--request-id' => "bad-\xC3\x28", '--json' => true]; + $tester = $operation === 'signal' + ? $this->runSignal($client, $arguments) + : $this->runUpdate($client, $arguments); + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + self::assertSame(0, $client->postCalls); + $decoded = $this->json($tester); + self::assertStringContainsString('valid UTF-8', $decoded['error']); + self::assertNull($decoded['request_id'] ?? null); + } + } + + public function test_a_valid_unattempted_key_keeps_its_exact_json_error_context(): void + { + $key = " \topaque \t"; + $client = new CommandRequestIdFakeClient([], contract: self::requestContract()); + $tester = $this->runUpdate($client, [ + 'workflow-id' => 'wf-1', 'update-name' => 'approve', + '--request-id' => $key, '--wait' => 'not-a-wait', '--json' => true, + ]); + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + self::assertSame(0, $client->postCalls); + self::assertSame($key, $this->json($tester)['request_id']); + } + + public function test_signal_generates_one_cli_signal_key_per_invocation_when_request_id_is_omitted(): void + { + $client = new CommandRequestIdFakeClient(self::signalReceipt()); + + $tester = $this->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('exact request ID', $display); + self::assertMatchesRegularExpression('/inspect[^.]*before retry/i', $display); + self::assertStringNotContainsString('--request-id=', $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::assertStringNotContainsString('--request-id=', $retry[0]['message']); + self::assertMatchesRegularExpression('/inspect[^.]*before retry/i', $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_correction_explicit_blank_keys_refuse_before_posting(): void + { + foreach ([null, '', ' ', "\t\r\n", "\u{00A0}\u{3000}"] as $key) { + foreach (['signal', 'update'] as $operation) { + foreach (['human', 'json', 'jsonl'] as $mode) { + $receipt = $operation === 'signal' ? self::signalReceipt() : self::updateReceipt(); + $client = new CommandRequestIdFakeClient($receipt, contract: self::requestContract()); + $arguments = ['workflow-id' => 'wf-1', $operation.'-name' => 'approve', '--request-id' => $key]; + if ($mode !== 'human') { $arguments['--output'] = $mode; } + $tester = $operation === 'signal' + ? $this->runSignal($client, $arguments) + : $this->runUpdate($client, $arguments); + self::assertSame(ExitCode::INVALID, $tester->getStatusCode()); + self::assertSame(0, $client->postCalls); + if ($mode !== 'human') { + $result = $this->json($tester); + self::assertNull($result['request_id'] ?? null); + self::assertNotContains('request_id.retry', array_column($result['recommendations'], 'id')); + } + } + } + } + } + + public function test_correction_unknown_admission_requires_inspection_before_retry(): void + { + foreach (['signal', 'update'] as $operation) { + foreach ([new NetworkException('lost response'), new TimeoutException('timeout'), new ServerHttpException('HTTP 502', 502)] as $error) { + foreach (['human', 'json', 'jsonl'] as $mode) { + $key = "opaque-key\n"; + $client = new CommandRequestIdFakeClient([], exception: $error, contract: self::requestContract()); + $arguments = ['workflow-id' => 'wf-1', $operation.'-name' => 'approve', '--request-id' => $key]; + if ($mode !== 'human') { $arguments['--output'] = $mode; } + $tester = $operation === 'signal' + ? $this->runSignal($client, $arguments) + : $this->runUpdate($client, $arguments); + self::assertSame(1, $client->postCalls); + self::assertSame($key, $client->lastPostBody['request_id']); + $advice = $tester->getDisplay(); + if ($mode !== 'human') { + $result = $this->json($tester); + self::assertSame($key, $result['request_id']); + $recommendation = array_values(array_filter($result['recommendations'], static fn (array $item): bool => $item['id'] === 'request_id.retry')); + self::assertCount(1, $recommendation); + self::assertArrayNotHasKey('command', $recommendation[0]); + $advice = $recommendation[0]['message']; + } + self::assertMatchesRegularExpression('/inspect[^.]*before retry/i', $advice); + self::assertStringNotContainsString('returns the original receipt', $advice); + self::assertStringNotContainsString('--request-id=', $advice); + } + } + } + } + + public function test_correction_unicode_code_point_limits_keep_exact_wire_values(): void + { + $cases = [ + [str_repeat('🦀', 255), true], + [str_repeat('🦀', 256), false], + [str_repeat("e\u{0301}", 127).'x', true], + [str_repeat("e\u{0301}", 128), false], + ['0', true], + [" \tkey-é \t", true], + ]; + foreach ($cases as [$key, $accepted]) { + foreach (['signal', 'update'] as $operation) { + $receipt = $operation === 'signal' ? self::signalReceipt() : self::updateReceipt(); + $client = new CommandRequestIdFakeClient($receipt, contract: self::requestContract()); + $arguments = ['workflow-id' => 'wf-1', $operation.'-name' => 'approve', '--request-id' => $key, '--json' => true]; + $tester = $operation === 'signal' + ? $this->runSignal($client, $arguments) + : $this->runUpdate($client, $arguments); + self::assertSame($accepted ? Command::SUCCESS : ExitCode::INVALID, $tester->getStatusCode()); + self::assertSame($accepted ? 1 : 0, $client->postCalls); + if ($accepted) { + self::assertSame($key, $client->lastPostBody['request_id']); + self::assertSame($key, $this->json($tester)['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, bool $decorated = false): CommandTester + { + $command = new SignalCommand(); + $command->setServerClient($client); + + $tester = new CommandTester($command); + $tester->execute($arguments, ['decorated' => $decorated]); + + return $tester; + } + + /** + * @param array $arguments + */ + private function runUpdate(ServerClient $client, array $arguments, bool $decorated = false): CommandTester + { + $command = new UpdateCommand(); + $command->setServerClient($client); + + $tester = new CommandTester($command); + $tester->execute($arguments, ['decorated' => $decorated]); + + 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 fe7a4ad..0748d5d 100644 --- a/tests/Commands/WorkflowControlPlaneCommandTest.php +++ b/tests/Commands/WorkflowControlPlaneCommandTest.php @@ -68,14 +68,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,