diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5ecbc6b74..a30227637 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -29,6 +29,15 @@
timers in a cancellation scope or parallel group are refused with
`unsupported_timer`. A timer step without both its scheduled and fired events
is refused with `incomplete_timer_step` (#864).
+- Retain keyed signal admissions and compare canonical typed intent for keyed
+ signals and updates. Conflicting key reuse refuses without durable writes;
+ undecodable legacy receipts remain explicitly unverified. This strengthens
+ the earlier same-key update receipt contract (#877).
+- Require valid UTF-8 caller keys of 1..255 code points before admission.
+ Explicit empty or non-string keys refuse; null clears an inherited key.
+ Apply the nullable `intent_digest` migration, drain and upgrade all embedded
+ database writers together, and allocate new keys for different operations.
+ Native Server and published SDK/CLI qualification remain separate.
## 2.17.1 - 2026-11-03
diff --git a/docs/api-stability.md b/docs/api-stability.md
index 6c3066dd1..8fe81be66 100644
--- a/docs/api-stability.md
+++ b/docs/api-stability.md
@@ -551,30 +551,56 @@ models should update app-owned projections at domain milestones and store
`workflow_id` plus `run_id` beside those business rows. See
[`docs/architecture/business-reporting-read-models.md`](architecture/business-reporting-read-models.md).
-## Update Request Identity
-
-An update caller key identifies a retained command within one logical workflow
-instance, across update names and runs. A retry returns that original lifecycle
-and does not replace its arguments. Omitted keys admit independent operations.
-Server uses the body `request_id`; proxy correlation headers remain audit data.
-Embedded webhook callers retain their existing `X-Request-Id` contract.
+
+
+## Signal and update command identity
+
+In unreleased Core source, a caller key identifies a retained command within
+one logical workflow instance and command type, across command names and runs.
+A same-intent retry returns the original admission and lifecycle, including
+after closure or continue-as-new while its receipt remains retained. Omitted
+keys admit independent operations. A signal and an update have separate key
+spaces. Embedded webhook callers retain their existing `X-Request-Id` contract.
+This embedded contract does not advertise retry safety for a Server release.
+Native service support requires separate published runtime and SDK/CLI
+qualification.
+
+Intent comparison includes the command name, typed arguments and target scope,
+including an explicitly requested run. Reusing a key for different intent
+returns `request_id_conflict` without recording another command, consuming a
+sequence or changing the retained outcome. Integer, float and string values
+remain distinct. A retained legacy receipt whose original intent cannot be
+decoded returns `intent_verified: false`; it never authorizes a fresh command
+or claims that the new intent matches. Once a receipt is pruned, its key no
+longer provides duplicate protection. Activity effects still require their own
+application or provider idempotency.
+
+Caller keys must be valid UTF-8 strings containing 1..255 code points, with no
+normalization or truncation. Case and canonically equivalent Unicode spellings
+remain distinct. Explicit null clears an inherited key; an omitted control-plane
+option preserves it. Explicit empty, non-string, malformed and oversized keys
+are refused before admission. Embedded calls throw
+`InvalidCommandRequestIdException`; control-plane signal/update returns a
+nonrecorded 422 result with validation errors and no command identity.
The logical key is available through `WorkflowCommand::requestId()` and command
-context. New update rows store a bounded `update-v1:` lookup token in the
-existing `workflow_commands.request_id` column. Consumers must not interpret
-that internal indexed value as caller text. Exact logical comparison preserves
-opaque case and long Unicode keys regardless of database collation. Legacy
-Server body keys remain readable from `context.server.metadata.request_id`,
-even when the old indexed column contains a proxy header. Existing rows are
-never rewritten. The earliest retained receipt wins if old code admitted more
-than one command for a caller key.
-
-Upgrade every process that writes directly to the database together before
-admitting new updates with this storage format. Drain and restart old embedded
-workers and Server processes; do not roll back to an old writer after new
-update rows are admitted. Remote SDK clients do not write this column and do
-not need an upgrade for this change. Native takeover must qualify the new
-representation separately.
+context. New signal/update rows store bounded `signal-v1:` / `update-v1:` lookup
+tokens in `workflow_commands.request_id`. Consumers must not interpret that
+internal indexed value as caller text. Exact logical comparison preserves
+opaque keys regardless of database collation. Legacy Server body keys remain
+readable from `context.server.metadata.request_id`, even when the old indexed
+column contains a proxy header. Proxy correlation is not a command key. The
+earliest retained receipt wins if old code admitted several for one caller key.
+
+Before upgrading writers, apply the additive migration that creates nullable
+`workflow_commands.intent_digest`. It performs no backfill or row rewrite;
+legacy intent is compared on read when recoverable. Drain and upgrade every
+process that writes directly to the workflow database together before admitting
+new keyed signals or updates. Mixed old/new writers and rollback to an old
+writer after admission are unsupported. Existing callers that reused a key for
+different intent must allocate a new key for a new operation; oversized keys
+must also be replaced before upgrade. Remote SDK clients do not write these
+columns. Native takeover must qualify the new representation separately.
Update intake limits are checked before requiring a local executable workflow
definition. A rejected update remains inspectable with
diff --git a/src/V2/CommandResult.php b/src/V2/CommandResult.php
index 118715338..6d65843d5 100644
--- a/src/V2/CommandResult.php
+++ b/src/V2/CommandResult.php
@@ -7,14 +7,94 @@
use Workflow\V2\Enums\CommandOutcome;
use Workflow\V2\Enums\CommandStatus;
use Workflow\V2\Models\WorkflowCommand;
+use Workflow\V2\Support\CommandRequestIdentity;
class CommandResult
{
+ protected bool $retained = false;
+
+ protected bool $retainedIntentVerified = true;
+
public function __construct(
protected readonly WorkflowCommand $command,
) {
}
+ /**
+ * The caller's command key for keyed signals and updates.
+ *
+ * @api Stable v2 command result identity API.
+ */
+ public function requestId(): ?string
+ {
+ return $this->command->requestId();
+ }
+
+ /**
+ * Whether this result replays a receipt retained from an earlier admission
+ * with the same caller key, rather than a new admission.
+ */
+ public function retained(): bool
+ {
+ return $this->retained;
+ }
+
+ /**
+ * False only for a retained legacy receipt whose intent could not be
+ * compared because its arguments are no longer decodable.
+ */
+ public function intentVerified(): bool
+ {
+ return ! $this->retained || $this->retainedIntentVerified;
+ }
+
+ public function requestIdConflict(): bool
+ {
+ return $this->command->rejection_reason === CommandRequestIdentity::CONFLICT_REASON;
+ }
+
+ /**
+ * @return array{retained_command_id: string|null, retained_command_type: string|null, retained_outcome: string|null, conflict: list}|null
+ */
+ public function conflict(): ?array
+ {
+ if (! $this->requestIdConflict()) {
+ return null;
+ }
+
+ $values = $this->command->payloadValues([
+ 'retained_command_id',
+ 'retained_command_type',
+ 'retained_outcome',
+ 'conflict',
+ ]);
+ $parts = $values['conflict'] ?? [];
+
+ return [
+ 'retained_command_id' => is_string(
+ $values['retained_command_id'] ?? null
+ ) ? $values['retained_command_id'] : null,
+ 'retained_command_type' => is_string(
+ $values['retained_command_type'] ?? null
+ ) ? $values['retained_command_type'] : null,
+ 'retained_outcome' => is_string($values['retained_outcome'] ?? null) ? $values['retained_outcome'] : null,
+ 'conflict' => is_array($parts) ? array_values(array_filter($parts, 'is_string')) : [],
+ ];
+ }
+
+ public function asRetainedReceipt(bool $intentVerified): static
+ {
+ $clone = clone $this;
+ $clone->retained = true;
+ $clone->retainedIntentVerified = $intentVerified;
+
+ return $clone;
+ }
+
+ /**
+ * For a request-key conflict this identifies the retained command that
+ * refused the new intent; no new command is recorded for that response.
+ */
public function commandId(): string
{
return $this->command->id;
diff --git a/src/V2/Contracts/WorkflowControlPlane.php b/src/V2/Contracts/WorkflowControlPlane.php
index 0479d9b56..b0315d155 100644
--- a/src/V2/Contracts/WorkflowControlPlane.php
+++ b/src/V2/Contracts/WorkflowControlPlane.php
@@ -77,6 +77,8 @@ public function redrive(string $instanceId, string $failedRunId, array $options
* Options:
* - arguments: array — positional signal arguments
* - namespace: string|null — require the workflow instance to belong to this namespace
+ * - request_id: string|null — the caller's command key; a retry with the same key and the same
+ * signal returns the retained admission, reuse with a different signal is refused (409)
* - command_context: \Workflow\V2\CommandContext|null — recorded command attribution/context
* - strict_configured_type_validation: bool — fail closed when a configured workflow type mapping is now invalid
*
@@ -136,6 +138,8 @@ public function query(string $instanceId, string $name, array $options = []): ar
* Options:
* - arguments: array — positional or named update arguments
* - namespace: string|null — require the workflow instance to belong to this namespace
+ * - request_id: string|null — the caller's command key; a retry with the same key and the same
+ * update returns the retained receipt, reuse with a different update is refused (409)
* - command_context: \Workflow\V2\CommandContext|null — recorded command attribution/context
* - wait_for: 'accepted'|'completed'|null — accepted-only submit or completion wait
* - wait_timeout_seconds: int|null — completion wait timeout
diff --git a/src/V2/Enums/CommandOutcome.php b/src/V2/Enums/CommandOutcome.php
index 62e0553d7..8fd1daeda 100644
--- a/src/V2/Enums/CommandOutcome.php
+++ b/src/V2/Enums/CommandOutcome.php
@@ -27,6 +27,7 @@ enum CommandOutcome: string
case RejectedUnknownSignal = 'rejected_unknown_signal';
case RejectedUnknownUpdate = 'rejected_unknown_update';
case RejectedInvalidArguments = 'rejected_invalid_arguments';
+ case RejectedRequestIdConflict = 'rejected_request_id_conflict';
case RejectedCompatibilityBlocked = 'rejected_compatibility_blocked';
case RejectedWorkflowDefinitionUnavailable = 'rejected_workflow_definition_unavailable';
}
diff --git a/src/V2/Exceptions/InvalidCommandRequestIdException.php b/src/V2/Exceptions/InvalidCommandRequestIdException.php
new file mode 100644
index 000000000..ffac8fb7d
--- /dev/null
+++ b/src/V2/Exceptions/InvalidCommandRequestIdException.php
@@ -0,0 +1,25 @@
+>
+ */
+ public function validationErrors(): array
+ {
+ return [
+ 'request_id' => [$this->getMessage()],
+ ];
+ }
+}
diff --git a/src/V2/Models/WorkflowCommand.php b/src/V2/Models/WorkflowCommand.php
index db389c592..31108f490 100644
--- a/src/V2/Models/WorkflowCommand.php
+++ b/src/V2/Models/WorkflowCommand.php
@@ -15,12 +15,12 @@
use Workflow\V2\Enums\CommandStatus;
use Workflow\V2\Enums\CommandType;
use Workflow\V2\Support\BridgeCommandIdentity;
+use Workflow\V2\Support\CommandRequestIdentity;
use Workflow\V2\Support\CommandSequence;
use Workflow\V2\Support\ConfiguredV2Models;
use Workflow\V2\Support\ExternalPayloads;
use Workflow\V2\Support\MessageStreamCursor;
use Workflow\V2\Support\ServiceCallEffectIdentity;
-use Workflow\V2\Support\UpdateRequestIdentity;
class WorkflowCommand extends Model
{
@@ -371,8 +371,8 @@ public function requestFingerprint(): ?string
public function requestId(): ?string
{
- if ($this->command_type === CommandType::Update) {
- return UpdateRequestIdentity::fromCommand($this);
+ if (CommandRequestIdentity::supports($this->command_type)) {
+ return CommandRequestIdentity::fromCommand($this);
}
if (is_string($this->request_id ?? null) && $this->request_id !== '') {
@@ -386,6 +386,17 @@ public function requestId(): ?string
: null;
}
+ /**
+ * Canonical intent recorded at admission for keyed signals and updates.
+ * Receipts recorded before intent digests existed return null.
+ */
+ public function intentDigest(): ?string
+ {
+ $digest = $this->intent_digest ?? null;
+
+ return is_string($digest) && $digest !== '' ? $digest : null;
+ }
+
public function requestedRunId(): ?string
{
if (is_string($this->requested_workflow_run_id ?? null) && $this->requested_workflow_run_id !== '') {
@@ -505,7 +516,10 @@ public function payloadValues(array $keys): array
private static function withRequestIdentity(WorkflowInstance $instance, array $attributes): array
{
$commandType = $attributes['command_type'] ?? null;
- $isUpdate = $commandType === CommandType::Update || $commandType === CommandType::Update->value;
+ $keyedType = $commandType;
+ if (! $keyedType instanceof CommandType) {
+ $keyedType = is_string($commandType) ? CommandType::tryFrom($commandType) : null;
+ }
$usesRequestId = $commandType === CommandType::Redrive
|| $commandType === CommandType::Redrive->value;
$context = is_array($attributes['context'] ?? null) ? $attributes['context'] : [];
@@ -528,22 +542,18 @@ private static function withRequestIdentity(WorkflowInstance $instance, array $a
}
}
- if ($isUpdate) {
- if ($serviceCallIdentity instanceof ServiceCallEffectIdentity) {
- $attributes['request_id'] = $serviceCallIdentity->storageToken();
- } else {
- $context['request'] = is_array($context['request'] ?? null) ? $context['request'] : [];
- $context['request']['request_id'] = UpdateRequestIdentity::fromContext($context);
- $attributes['context'] = $context;
- $attributes['request_id'] = UpdateRequestIdentity::storageToken($instance, $context);
- }
- } elseif ($serviceCallIdentity instanceof ServiceCallEffectIdentity && in_array($commandType, [
- CommandType::Start,
- CommandType::Start->value,
- CommandType::Signal,
- CommandType::Signal->value,
+ if ($serviceCallIdentity instanceof ServiceCallEffectIdentity && in_array($commandType, [
+ CommandType::Start, CommandType::Start->value,
+ CommandType::Signal, CommandType::Signal->value,
+ CommandType::Update, CommandType::Update->value,
], true)) {
$attributes['request_id'] = $serviceCallIdentity->storageToken();
+ } elseif ($keyedType instanceof CommandType && CommandRequestIdentity::supports($keyedType)) {
+ // Caller keys remain independent of bridge/service-call identities.
+ $context['request'] = is_array($context['request'] ?? null) ? $context['request'] : [];
+ $context['request']['request_id'] = CommandRequestIdentity::fromContext($context, $keyedType);
+ $attributes['context'] = $context;
+ $attributes['request_id'] = CommandRequestIdentity::storageToken($instance, $keyedType, $context);
} elseif ($usesRequestId) {
$request = is_array($attributes['context']['request'] ?? null)
? $attributes['context']['request']
diff --git a/src/V2/Support/CommandIntent.php b/src/V2/Support/CommandIntent.php
new file mode 100644
index 000000000..3aae12c6f
--- /dev/null
+++ b/src/V2/Support/CommandIntent.php
@@ -0,0 +1,156 @@
+ $arguments
+ */
+ public static function digest(
+ CommandType $type,
+ string $name,
+ array $arguments,
+ string $targetScope = 'instance',
+ ?string $requestedRunId = null,
+ ): string {
+ $canonical = self::canonical([
+ 'arguments' => $arguments,
+ 'name' => $name,
+ 'target' => [
+ 'run' => $targetScope === 'run' ? $requestedRunId : null,
+ 'scope' => $targetScope === 'run' ? 'run' : 'instance',
+ ],
+ 'type' => $type->value,
+ ]);
+
+ return self::VERSION . ':' . hash('sha256', $canonical);
+ }
+
+ public static function isDigest(mixed $value): bool
+ {
+ return is_string($value) && preg_match('/^v\d+:[0-9a-f]{64}$/', $value) === 1;
+ }
+
+ /**
+ * Type-tagged canonical encoding used by {@see digest()}.
+ */
+ public static function canonical(mixed $value): string
+ {
+ return match (true) {
+ $value === null => 'n',
+ is_bool($value) => $value ? 'b1' : 'b0',
+ is_int($value) => 'i' . $value,
+ is_float($value) => 'd' . self::floatToken($value),
+ is_string($value) => 's' . strlen($value) . ':' . $value,
+ $value instanceof AvroBinaryValue => 'x' . strlen($value->bytes) . ':' . $value->bytes,
+ $value instanceof AvroMapValue => self::avroMapToken($value),
+ is_array($value) => self::arrayToken($value),
+ $value instanceof \BackedEnum => self::canonical($value->value),
+ $value instanceof \DateTimeInterface => 't' . $value->format('Y-m-d\TH:i:s.uP'),
+ $value instanceof \Stringable => self::canonical((string) $value),
+ default => self::serializedToken($value),
+ };
+ }
+
+ private static function avroMapToken(AvroMapValue $value): string
+ {
+ $pairs = [];
+ foreach ($value->pairs as $pair) {
+ $pairs[] = [(string) $pair[0], $pair[1]];
+ }
+
+ return self::mapToken($pairs);
+ }
+
+ /**
+ * @param array $value
+ */
+ private static function arrayToken(array $value): string
+ {
+ if (array_is_list($value)) {
+ $items = '';
+ foreach ($value as $item) {
+ $items .= self::canonical($item);
+ }
+
+ return 'l' . count($value) . '[' . $items . ']';
+ }
+
+ $pairs = [];
+ foreach ($value as $key => $item) {
+ $pairs[] = [(string) $key, $item];
+ }
+
+ return self::mapToken($pairs);
+ }
+
+ private static function serializedToken(mixed $value): string
+ {
+ $serialized = serialize($value);
+
+ return 'o' . strlen($serialized) . ':' . $serialized;
+ }
+
+ /**
+ * @param list $pairs
+ */
+ private static function mapToken(array $pairs): string
+ {
+ if (count($pairs) === 1 && $pairs[0][0] === self::PROTECTED_INTENT_KEY && is_string($pairs[0][1])) {
+ return 'p' . strlen($pairs[0][1]) . ':' . $pairs[0][1];
+ }
+
+ usort($pairs, static fn (array $left, array $right): int => strcmp($left[0], $right[0]));
+
+ $entries = '';
+
+ foreach ($pairs as [$key, $item]) {
+ $entries .= 's' . strlen($key) . ':' . $key . self::canonical($item);
+ }
+
+ return 'm' . count($pairs) . '{' . $entries . '}';
+ }
+
+ private static function floatToken(float $value): string
+ {
+ if (is_nan($value)) {
+ return 'nan';
+ }
+
+ if (is_infinite($value)) {
+ return $value > 0 ? 'inf' : '-inf';
+ }
+
+ // Fixed-endian IEEE 754 bytes preserve every finite value without
+ // depending on the process's locale or serialize_precision setting.
+ return '0x' . bin2hex(pack('E', $value));
+ }
+}
diff --git a/src/V2/Support/CommandRequestIdentity.php b/src/V2/Support/CommandRequestIdentity.php
new file mode 100644
index 000000000..e9fc91fbb
--- /dev/null
+++ b/src/V2/Support/CommandRequestIdentity.php
@@ -0,0 +1,334 @@
+ 1020
+ || preg_match('/\A.{1,255}\z/us', $requestId) !== 1) {
+ throw new InvalidCommandRequestIdException();
+ }
+
+ return $requestId;
+ }
+
+ public static function supports(CommandType|string|null $type): bool
+ {
+ $value = $type instanceof CommandType ? $type->value : $type;
+
+ return $value === CommandType::Signal->value || $value === CommandType::Update->value;
+ }
+
+ /**
+ * @param array $context
+ */
+ public static function fromContext(array $context, CommandType $type): ?string
+ {
+ return self::nonEmptyString(self::contextValue($context, $type));
+ }
+
+ /**
+ * @param array $context
+ */
+ public static function validateContext(array $context, CommandType $type): void
+ {
+ self::validate(self::contextValue($context, $type));
+ }
+
+ public static function fromCommand(WorkflowCommand $command): ?string
+ {
+ $type = $command->command_type instanceof CommandType
+ ? $command->command_type
+ : CommandType::tryFrom((string) $command->command_type);
+ if (! $type instanceof CommandType || ! self::supports($type)) {
+ return null;
+ }
+
+ $context = is_array($command->context) ? $command->context : [];
+ $request = is_array($context['request'] ?? null) ? $context['request'] : [];
+ if (($context['server']['command'] ?? null) === $type->value || array_key_exists('request_id', $request)) {
+ return self::fromContext($context, $type);
+ }
+
+ $column = self::nonEmptyString($command->request_id);
+ return $column !== null && ! self::isToken($column) ? $column : null;
+ }
+
+ /**
+ * Called while the owning instance is locked. Legacy context lookup must
+ * not depend on a column that may contain an unrelated proxy header.
+ *
+ * @param array $context
+ */
+ public static function find(WorkflowInstance $instance, CommandType $type, array $context): ?WorkflowCommand
+ {
+ $key = self::fromContext($context, $type);
+ if ($key === null) {
+ return null;
+ }
+
+ $token = self::keyToken($type, $key);
+ $commands = self::query($instance, $type)->where(static function (Builder $query) use ($key, $token): void {
+ $query->where('request_id', $token)
+ ->orWhere('request_id', 'like', $token . ':%')
+ ->orWhere('request_id', $key)
+ ->orWhere('context->request->request_id', $key)
+ ->orWhere('context->server->metadata->request_id', $key);
+ })->orderBy('created_at')
+ ->orderBy('id')
+ ->lockForUpdate()
+ ->cursor();
+
+ foreach ($commands as $command) {
+ // SQL collations may compare opaque keys case-insensitively.
+ if (self::fromCommand($command) !== $key) {
+ continue;
+ }
+
+ // A recorded conflict never consumes the key; the receipt it
+ // refused against is always earlier in this ordering.
+ if ($command->rejection_reason === self::CONFLICT_REASON) {
+ continue;
+ }
+
+ return $command;
+ }
+
+ return null;
+ }
+
+ /**
+ * The existing indexed column holds a bounded token, not caller text.
+ * Preserve every old row, including a legacy key occupying that token.
+ * The caller already holds the instance lock used by admission.
+ *
+ * @param array $context
+ */
+ public static function storageToken(WorkflowInstance $instance, CommandType $type, array $context): string
+ {
+ $key = self::fromContext($context, $type);
+ $base = $key === null ? self::tokenPrefix($type) . Str::ulid() : self::keyToken($type, $key);
+ $candidate = $base;
+ $suffix = 0;
+ while (self::query($instance, $type)->where('request_id', $candidate)->exists()) {
+ $candidate = $base . ':' . ++$suffix;
+ }
+ return $candidate;
+ }
+
+ /**
+ * Compare a retry's intent digest with the retained receipt.
+ *
+ * Receipts recorded before intent digests exist are recomputed from their
+ * retained name and validated arguments when those can still be decoded.
+ * Legacy rejections kept no reliable arguments, so they stay unverified
+ * rather than claiming equality or refusing a retry.
+ */
+ public static function compareIntent(WorkflowCommand $retained, string $candidateDigest): string
+ {
+ $stored = self::retainedDigest($retained);
+
+ if ($stored === null) {
+ return self::INTENT_UNVERIFIED;
+ }
+
+ return hash_equals($stored, $candidateDigest) ? self::INTENT_SAME : self::INTENT_DIFFERENT;
+ }
+
+ public static function retainedDigest(WorkflowCommand $retained): ?string
+ {
+ $stored = self::nonEmptyString($retained->intent_digest ?? null);
+
+ if ($stored !== null || $retained->status !== CommandStatus::Accepted) {
+ return $stored;
+ }
+
+ try {
+ return match ($retained->command_type) {
+ CommandType::Update => self::legacyUpdateDigest($retained),
+ CommandType::Signal => self::legacySignalDigest($retained),
+ default => null,
+ };
+ } catch (Throwable) {
+ return null;
+ }
+ }
+
+ /**
+ * Parts of the retained command known to differ from a conflicting retry.
+ *
+ * @return list
+ */
+ public static function conflictingParts(
+ WorkflowCommand $retained,
+ string $name,
+ string $targetScope,
+ ?string $requestedRunId
+ ): array {
+ $parts = [];
+ $retainedName = self::retainedName($retained);
+
+ if ($retainedName !== null && $retainedName !== $name) {
+ $parts[] = 'name';
+ }
+
+ $retainedScope = $retained->target_scope === 'run' ? 'run' : 'instance';
+ $scope = $targetScope === 'run' ? 'run' : 'instance';
+ $retainedRun = $retainedScope === 'run' ? $retained->requestedRunId() : null;
+ $requestedRun = $scope === 'run' ? $requestedRunId : null;
+
+ if ($retainedScope !== $scope || $retainedRun !== $requestedRun) {
+ $parts[] = 'target';
+ }
+
+ if ($parts === []) {
+ $parts[] = 'arguments';
+ }
+
+ return $parts;
+ }
+
+ public static function keyToken(CommandType $type, string $key): string
+ {
+ return self::tokenPrefix($type) . hash('sha256', $key);
+ }
+
+ /**
+ * @param array $context
+ */
+ private static function contextValue(array $context, CommandType $type): mixed
+ {
+ // Server has always retained the caller's body key separately from
+ // proxy correlation. Its omission means an independent operation.
+ $server = $context['server'] ?? null;
+ if (is_array($server) && ($server['command'] ?? null) === $type->value) {
+ $metadata = is_array($server['metadata'] ?? null) ? $server['metadata'] : [];
+ return $metadata['request_id'] ?? null;
+ }
+
+ $request = is_array($context['request'] ?? null) ? $context['request'] : [];
+ return $request['request_id'] ?? null;
+ }
+
+ private static function tokenPrefix(CommandType $type): string
+ {
+ return $type->value . '-v1:';
+ }
+
+ private static function isToken(string $value): bool
+ {
+ return preg_match('/^(signal|update)-v1:/', $value) === 1;
+ }
+
+ private static function retainedName(WorkflowCommand $retained): ?string
+ {
+ return match ($retained->command_type) {
+ CommandType::Update => self::nonEmptyString(
+ ConfiguredV2Models::query('update_model', WorkflowUpdate::class)
+ ->where('workflow_command_id', $retained->id)
+ ->value('update_name'),
+ ),
+ CommandType::Signal => self::nonEmptyString(
+ ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
+ ->where('workflow_command_id', $retained->id)
+ ->value('signal_name'),
+ ),
+ default => null,
+ };
+ }
+
+ private static function legacyUpdateDigest(WorkflowCommand $retained): ?string
+ {
+ /** @var WorkflowUpdate|null $update */
+ $update = ConfiguredV2Models::query('update_model', WorkflowUpdate::class)
+ ->where('workflow_command_id', $retained->id)
+ ->first();
+
+ if (! $update instanceof WorkflowUpdate || ! is_string($update->update_name)) {
+ return null;
+ }
+
+ return CommandIntent::digest(
+ CommandType::Update,
+ $update->update_name,
+ $update->updateArguments(),
+ $retained->target_scope,
+ $retained->requestedRunId(),
+ );
+ }
+
+ private static function legacySignalDigest(WorkflowCommand $retained): ?string
+ {
+ /** @var WorkflowSignal|null $signal */
+ $signal = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
+ ->where('workflow_command_id', $retained->id)
+ ->first();
+
+ if (! $signal instanceof WorkflowSignal || ! is_string($signal->signal_name)) {
+ return null;
+ }
+
+ return CommandIntent::digest(
+ CommandType::Signal,
+ $signal->signal_name,
+ $signal->signalArguments(),
+ $retained->target_scope,
+ $retained->requestedRunId(),
+ );
+ }
+
+ private static function nonEmptyString(mixed $value): ?string
+ {
+ return is_string($value) && $value !== '' ? $value : null;
+ }
+
+ /**
+ * @return Builder
+ */
+ private static function query(WorkflowInstance $instance, CommandType $type): Builder
+ {
+ return ConfiguredV2Models::query('command_model', WorkflowCommand::class)
+ ->where('workflow_instance_id', $instance->id)
+ ->where('command_type', $type->value);
+ }
+}
diff --git a/src/V2/Support/CommandResponse.php b/src/V2/Support/CommandResponse.php
index 00bca9439..68fe8e630 100644
--- a/src/V2/Support/CommandResponse.php
+++ b/src/V2/Support/CommandResponse.php
@@ -21,7 +21,7 @@ public static function payload(CommandResult $result, ?string $workflowType = nu
'run_id' => $result->runId(),
'requested_run_id' => $result->requestedRunId(),
'resolved_run_id' => $result->resolvedRunId(),
- 'command_id' => $result->commandId(),
+ 'command_id' => $result->requestIdConflict() ? null : $result->commandId(),
'command_sequence' => $result->commandSequence(),
'target_scope' => $result->targetScope(),
'workflow_type' => $workflowType ?? $result->workflowType(),
@@ -59,6 +59,18 @@ public static function payload(CommandResult $result, ?string $workflowType = nu
}
}
+ if (CommandRequestIdentity::supports($result->type())) {
+ $payload['request_id'] = $result->requestId();
+ $payload['retained'] = $result->retained();
+ $payload['intent_verified'] = $result->intentVerified();
+
+ $conflict = $result->conflict();
+
+ if ($conflict !== null) {
+ $payload = array_merge($payload, $conflict);
+ }
+ }
+
if ($result instanceof SignalWithStartResult) {
$payload['start_command_id'] = $result->startCommandId();
$payload['start_command_sequence'] = $result->startCommandSequence();
diff --git a/src/V2/Support/DefaultWorkflowControlPlane.php b/src/V2/Support/DefaultWorkflowControlPlane.php
index 958d633fb..cea97598f 100644
--- a/src/V2/Support/DefaultWorkflowControlPlane.php
+++ b/src/V2/Support/DefaultWorkflowControlPlane.php
@@ -23,6 +23,7 @@
use Workflow\V2\Enums\RunStatus;
use Workflow\V2\Enums\TaskStatus;
use Workflow\V2\Enums\TaskType;
+use Workflow\V2\Exceptions\InvalidCommandRequestIdException;
use Workflow\V2\Exceptions\InvalidQueryArgumentsException;
use Workflow\V2\Exceptions\WorkflowExecutionUnavailableException;
use Workflow\V2\Models\WorkflowCommand;
@@ -805,32 +806,32 @@ public function query(string $instanceId, string $name, array $options = []): ar
public function update(string $instanceId, string $name, array $options = []): array
{
$loaded = $this->loadControlPlaneWorkflow($instanceId, $options);
-
- if (($loaded['error'] ?? null) !== null) {
- return $loaded['error'];
- }
-
$stub = $loaded['workflow'] ?? null;
+ $error = $loaded['error'] ?? null;
- if (! $stub instanceof WorkflowStub) {
- return $this->notFoundControlPlaneResult($instanceId, 'update_id');
+ if ($error !== null || ! $stub instanceof WorkflowStub) {
+ return $error ?? $this->notFoundControlPlaneResult($instanceId, 'update_id');
}
$waitFor = array_key_exists('wait_for', $options)
? UpdateWaitPolicy::requestedWaitFor($options['wait_for'])
: UpdateWaitPolicy::WAIT_FOR_ACCEPTED;
- $stub = $stub->withCommandContext($this->commandContext($options));
+ try {
+ $stub = $this->withRequestIdOption($stub->withCommandContext($this->commandContext($options)), $options);
- if (($options['wait_timeout_seconds'] ?? null) !== null && $waitFor !== UpdateWaitPolicy::WAIT_FOR_ACCEPTED) {
- $stub = $stub->withUpdateWaitTimeout(
- UpdateWaitPolicy::requestedTimeoutSeconds($options['wait_timeout_seconds'] ?? null),
- );
- }
+ if (($options['wait_timeout_seconds'] ?? null) !== null && $waitFor !== UpdateWaitPolicy::WAIT_FOR_ACCEPTED) {
+ $stub = $stub->withUpdateWaitTimeout(
+ UpdateWaitPolicy::requestedTimeoutSeconds($options['wait_timeout_seconds'] ?? null),
+ );
+ }
- $result = $waitFor === UpdateWaitPolicy::WAIT_FOR_ACCEPTED
- ? $stub->submitUpdateWithArguments($name, $this->commandArguments($options))
- : $stub->attemptUpdateWithArguments($name, $this->commandArguments($options));
+ $result = $waitFor === UpdateWaitPolicy::WAIT_FOR_ACCEPTED
+ ? $stub->submitUpdateWithArguments($name, $this->commandArguments($options))
+ : $stub->attemptUpdateWithArguments($name, $this->commandArguments($options));
+ } catch (InvalidCommandRequestIdException $exception) {
+ return $this->invalidRequestIdResult($instanceId, $exception);
+ }
return array_merge(
CommandResponse::payload($result),
@@ -1116,29 +1117,29 @@ private function deliverSignal(
: null;
$payloadBlob = $options['payload_blob'] ?? null;
$loaded = $this->loadControlPlaneWorkflow($instanceId, $options);
-
- if (($loaded['error'] ?? null) !== null) {
- return $loaded['error'];
- }
-
$stub = $loaded['workflow'] ?? null;
+ $error = $loaded['error'] ?? null;
- if (! $stub instanceof WorkflowStub) {
- return $this->notFoundControlPlaneResult($instanceId, 'workflow_command_id');
+ if ($error !== null || ! $stub instanceof WorkflowStub) {
+ return $error ?? $this->notFoundControlPlaneResult($instanceId, 'workflow_command_id');
}
$arguments = $this->commandArguments($options);
- $stub = $stub->withCommandContext($this->commandContext($options));
- $result = $runtimeReserved
- ? $stub->attemptRuntimeSignalWithArguments($name, $arguments, $payloadCodec, $payloadBlob)
- : $stub->attemptSignalWithArguments($name, $arguments, $payloadCodec, $payloadBlob);
+ try {
+ $stub = $this->withRequestIdOption($stub->withCommandContext($this->commandContext($options)), $options);
+ $result = $runtimeReserved
+ ? $stub->attemptRuntimeSignalWithArguments($name, $arguments, $payloadCodec, $payloadBlob)
+ : $stub->attemptSignalWithArguments($name, $arguments, $payloadCodec, $payloadBlob);
+ } catch (InvalidCommandRequestIdException $exception) {
+ return $this->invalidRequestIdResult($instanceId, $exception);
+ }
return array_merge(
CommandResponse::payload($result),
[
'accepted' => $result->accepted(),
'workflow_instance_id' => $instanceId,
- 'workflow_command_id' => $result->commandId(),
+ 'workflow_command_id' => $result->requestIdConflict() ? null : $result->commandId(),
'signal_name' => $name,
'command_reason' => $result->reason(),
'reason' => $result->rejected() ? $result->rejectionReason() : null,
@@ -1251,6 +1252,7 @@ private function commandAttributes(CommandContext $commandContext, array $attrib
private function loadControlPlaneWorkflow(string $instanceId, array $options): array
{
$namespace = $this->namespaceOption($options);
+ $runId = $options['run_id'] ?? null;
try {
$query = $this->instanceQuery();
@@ -1261,6 +1263,7 @@ private function loadControlPlaneWorkflow(string $instanceId, array $options): a
/** @var WorkflowInstance $instance */
$instance = $query->findOrFail($instanceId);
+ $workflow = WorkflowStub::loadSelection($instanceId, is_string($runId) ? $runId : null, $namespace);
} catch (ModelNotFoundException) {
return [
'workflow' => null,
@@ -1295,7 +1298,7 @@ private function loadControlPlaneWorkflow(string $instanceId, array $options): a
}
return [
- 'workflow' => WorkflowStub::load($instanceId, $namespace),
+ 'workflow' => $workflow,
'error' => null,
];
}
@@ -1311,6 +1314,43 @@ private function commandArguments(array $options): array
return is_array($arguments) ? $arguments : [];
}
+ /**
+ * An explicit request_id option is the caller's command key and takes
+ * precedence over any key already carried by the command context.
+ * Null clears an inherited key; empty or invalid values are refused.
+ *
+ * @param array $options
+ */
+ private function withRequestIdOption(WorkflowStub $stub, array $options): WorkflowStub
+ {
+ if (! array_key_exists('request_id', $options)) {
+ return $stub;
+ }
+
+ return $stub->withRequestId(CommandRequestIdentity::validate($options['request_id']));
+ }
+
+ /**
+ * @return array
+ */
+ private function invalidRequestIdResult(string $instanceId, InvalidCommandRequestIdException $exception): array
+ {
+ return [
+ 'success' => false,
+ 'accepted' => false,
+ 'recorded' => false,
+ 'workflow_instance_id' => $instanceId,
+ 'workflow_id' => $instanceId,
+ 'workflow_command_id' => null,
+ 'command_id' => null,
+ 'update_id' => null,
+ 'reason' => 'invalid_request_id',
+ 'message' => $exception->getMessage(),
+ 'validation_errors' => $exception->validationErrors(),
+ 'status' => 422,
+ ];
+ }
+
/**
* @param array $options
*/
diff --git a/src/V2/Support/UpdateRequestIdentity.php b/src/V2/Support/UpdateRequestIdentity.php
index 6344e5fc1..3f8fd21b1 100644
--- a/src/V2/Support/UpdateRequestIdentity.php
+++ b/src/V2/Support/UpdateRequestIdentity.php
@@ -4,14 +4,13 @@
namespace Workflow\V2\Support;
-use Illuminate\Database\Eloquent\Builder;
-use Illuminate\Support\Str;
use Workflow\V2\Enums\CommandType;
use Workflow\V2\Models\WorkflowCommand;
use Workflow\V2\Models\WorkflowInstance;
/**
- * Shared update receipt identity for embedded and Server admission.
+ * Update receipt identity, kept as the update-specific view of
+ * {@see CommandRequestIdentity} for embedded and Server admission.
*
* @api
*/
@@ -22,100 +21,29 @@ final class UpdateRequestIdentity
*/
public static function fromContext(array $context): ?string
{
- // Server has always retained the caller's body key separately from
- // proxy correlation. Its omission means an independent operation.
- $server = $context['server'] ?? null;
- if (is_array($server) && ($server['command'] ?? null) === 'update') {
- $metadata = is_array($server['metadata'] ?? null) ? $server['metadata'] : [];
- return self::nonEmptyString($metadata['request_id'] ?? null);
- }
-
- $request = is_array($context['request'] ?? null) ? $context['request'] : [];
- return self::nonEmptyString($request['request_id'] ?? null);
+ return CommandRequestIdentity::fromContext($context, CommandType::Update);
}
public static function fromCommand(WorkflowCommand $command): ?string
{
- $context = is_array($command->context) ? $command->context : [];
- $request = is_array($context['request'] ?? null) ? $context['request'] : [];
- if (($context['server']['command'] ?? null) === 'update' || array_key_exists('request_id', $request)) {
- return self::fromContext($context);
- }
-
- return self::nonEmptyString($command->request_id);
+ return CommandRequestIdentity::fromCommand($command);
}
/**
- * Called while the owning instance is locked. Legacy context lookup must
- * not depend on a column that may contain an unrelated proxy header.
+ * Called while the owning instance is locked.
*
* @param array $context
*/
public static function find(WorkflowInstance $instance, array $context): ?WorkflowCommand
{
- $key = self::fromContext($context);
- if ($key === null) {
- return null;
- }
-
- $token = self::keyToken($key);
- $commands = self::query($instance)->where(static function (Builder $query) use ($key, $token): void {
- $query->where('request_id', $token)
- ->orWhere('request_id', 'like', $token . ':%')
- ->orWhere('request_id', $key)
- ->orWhere('context->request->request_id', $key)
- ->orWhere('context->server->metadata->request_id', $key);
- })->orderBy('created_at')
- ->orderBy('id')
- ->lockForUpdate()
- ->cursor();
-
- foreach ($commands as $command) {
- // SQL collations may compare opaque keys case-insensitively.
- if (self::fromCommand($command) === $key) {
- return $command;
- }
- }
-
- return null;
+ return CommandRequestIdentity::find($instance, CommandType::Update, $context);
}
/**
- * The existing indexed column holds a bounded token, not caller text.
- * Preserve every old row, including a legacy key occupying that token.
- * The caller already holds the instance lock used by update admission.
- *
* @param array $context
*/
public static function storageToken(WorkflowInstance $instance, array $context): string
{
- $key = self::fromContext($context);
- $base = $key === null ? 'update-v1:' . Str::ulid() : self::keyToken($key);
- $candidate = $base;
- $suffix = 0;
- while (self::query($instance)->where('request_id', $candidate)->exists()) {
- $candidate = $base . ':' . ++$suffix;
- }
- return $candidate;
- }
-
- private static function keyToken(string $key): string
- {
- return 'update-v1:' . hash('sha256', $key);
- }
-
- private static function nonEmptyString(mixed $value): ?string
- {
- return is_string($value) && $value !== '' ? $value : null;
- }
-
- /**
- * @return Builder
- */
- private static function query(WorkflowInstance $instance): Builder
- {
- return ConfiguredV2Models::query('command_model', WorkflowCommand::class)
- ->where('workflow_instance_id', $instance->id)
- ->where('command_type', CommandType::Update->value);
+ return CommandRequestIdentity::storageToken($instance, CommandType::Update, $context);
}
}
diff --git a/src/V2/WorkflowStub.php b/src/V2/WorkflowStub.php
index ed5934ccb..e8af8ffc9 100644
--- a/src/V2/WorkflowStub.php
+++ b/src/V2/WorkflowStub.php
@@ -56,6 +56,8 @@
use Workflow\V2\Support\CancellationScopeHistory;
use Workflow\V2\Support\CancellationScopeRequests;
use Workflow\V2\Support\ChildRunHistory;
+use Workflow\V2\Support\CommandIntent;
+use Workflow\V2\Support\CommandRequestIdentity;
use Workflow\V2\Support\ConfiguredV2Models;
use Workflow\V2\Support\CurrentRunResolver;
use Workflow\V2\Support\ExecutionTracing;
@@ -87,7 +89,6 @@
use Workflow\V2\Support\TraceCarrier;
use Workflow\V2\Support\TypeRegistry;
use Workflow\V2\Support\UpdateCommandGate;
-use Workflow\V2\Support\UpdateRequestIdentity;
use Workflow\V2\Support\UpdateWaitPolicy;
use Workflow\V2\Support\UpsertMemosCall;
use Workflow\V2\Support\UpsertSearchAttributesCall;
@@ -912,6 +913,40 @@ public function withCommandContext(CommandContext $commandContext): self
return $clone;
}
+ /**
+ * Attach the caller's command key for the next signal or update.
+ *
+ * A retry with the same key and the same command returns the retained
+ * admission instead of recording another command; reuse of the key with a
+ * different command name, arguments or run target is refused.
+ * Keys contain 1 to 255 Unicode characters without normalization. Null
+ * clears the explicit key; invalid UTF-8 or length throws before admission.
+ *
+ * @api Stable v2 command identity API.
+ */
+ public function withRequestId(?string $requestId): self
+ {
+ CommandRequestIdentity::validate($requestId);
+
+ $context = $this->resolvedCommandContext();
+ $attributes = $context->attributes()['context'];
+ $patch = [
+ 'request' => [
+ 'request_id' => $requestId,
+ ],
+ ];
+
+ if (is_array($attributes['server'] ?? null) && isset($attributes['server']['command'])) {
+ $patch['server'] = [
+ 'metadata' => [
+ 'request_id' => $requestId,
+ ],
+ ];
+ }
+
+ return $this->withCommandContext($context->with($patch));
+ }
+
public function withUpdateWaitTimeout(?int $seconds): self
{
if ($seconds !== null && $seconds < 1) {
@@ -1155,7 +1190,7 @@ public function inspectUpdate(string $updateId): UpdateResult
*/
public function submitUpdateWithArguments(string $method, array $arguments): UpdateResult
{
- [$command, $update, $resumeTask] = $this->recordAcceptedUpdateWithArguments($method, $arguments);
+ [$command, $update, $resumeTask, $retained] = $this->recordAcceptedUpdateWithArguments($method, $arguments);
$this->refresh();
@@ -1170,6 +1205,11 @@ public function submitUpdateWithArguments(string $method, array $arguments): Upd
));
}
+ if ($retained !== null) {
+ return UpdateResult::fromCommand($command, null, null, $update, 'accepted')
+ ->asRetainedReceipt($retained['verified']);
+ }
+
if ($command->status !== CommandStatus::Rejected) {
self::recordUpdateSent($this->instance->id, $method, $arguments);
}
@@ -1182,7 +1222,7 @@ public function submitUpdateWithArguments(string $method, array $arguments): Upd
*/
public function attemptUpdateWithArguments(string $method, array $arguments): UpdateResult
{
- [$command, $update, $resumeTask] = $this->recordAcceptedUpdateWithArguments($method, $arguments);
+ [$command, $update, $resumeTask, $retained] = $this->recordAcceptedUpdateWithArguments($method, $arguments);
$this->refresh();
@@ -1197,12 +1237,12 @@ public function attemptUpdateWithArguments(string $method, array $arguments): Up
));
}
- if ($command->status !== CommandStatus::Rejected) {
+ if ($retained === null && $command->status !== CommandStatus::Rejected) {
self::recordUpdateSent($this->instance->id, $method, $arguments);
}
if (! $update instanceof WorkflowUpdate || $command->status === CommandStatus::Rejected) {
- return UpdateResult::fromCommand(
+ $result = UpdateResult::fromCommand(
$command,
null,
null,
@@ -1211,6 +1251,8 @@ public function attemptUpdateWithArguments(string $method, array $arguments): Up
false,
$this->updateWaitTimeoutSeconds(),
);
+
+ return $retained === null ? $result : $result->asRetainedReceipt($retained['verified']);
}
if ($this->shouldInlineAcceptedUpdateCompletion($update)) {
@@ -1219,13 +1261,15 @@ public function attemptUpdateWithArguments(string $method, array $arguments): Up
$waitTimedOut = $this->waitForAcceptedUpdateToClose($update->id, $this->updateWaitTimeoutSeconds());
- return $this->freshUpdateResult(
+ $result = $this->freshUpdateResult(
$command->id,
$update->id,
'completed',
$waitTimedOut,
$this->updateWaitTimeoutSeconds(),
);
+
+ return $retained === null ? $result : $result->asRetainedReceipt($retained['verified']);
}
public function repair(): CommandResult
@@ -2432,10 +2476,11 @@ private function recordCancellationRequest(
/**
* @param array $arguments
- * @return array{0: WorkflowCommand|null, 1: WorkflowUpdate|null, 2: WorkflowTask|null}
+ * @return array{0: WorkflowCommand|null, 1: WorkflowUpdate|null, 2: WorkflowTask|null, 3: array{verified: bool}|null}
*/
private function recordAcceptedUpdateWithArguments(string $method, array $arguments): array
{
+ $this->validateCommandRequestId(CommandType::Update);
$this->refresh();
$incomingContext = TraceCarrier::captureConfigured();
@@ -2445,6 +2490,8 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
$update = null;
/** @var WorkflowTask|null $resumeTask */
$resumeTask = null;
+ /** @var array{verified: bool}|null $retained */
+ $retained = null;
ConfiguredV2Models::connection()->transaction(function () use (
$method,
@@ -2452,18 +2499,43 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
&$command,
&$update,
&$resumeTask,
+ &$retained,
$incomingContext
): void {
+ $retained = null;
+
/** @var WorkflowInstance $instance */
$instance = self::instanceQuery()
->lockForUpdate()
->findOrFail($this->instance->id);
- [$existingCommand, $existingUpdate] = $this->existingUpdateForRequest($instance);
+ $internalCommand = $this->existingInternalCommandForRequest($instance, CommandType::Update);
+ if ($internalCommand instanceof WorkflowCommand) {
+ $command = $internalCommand;
+ $update = $this->updateForCommand($command);
+ return;
+ }
+
+ // A keyed retry is answered from its retained receipt before any run
+ // resolution, so closure or continuation never admits it again.
+ $retainedReceipt = $this->retainedCommandForRequest(
+ $instance,
+ CommandType::Update,
+ fn (?WorkflowRun $retainedRun): string => $this->updateIntentDigest($retainedRun, $method, $arguments),
+ );
+
+ if ($retainedReceipt !== null) {
+ if ($retainedReceipt['intent'] !== CommandRequestIdentity::INTENT_DIFFERENT) {
+ $command = $retainedReceipt['command'];
+ $update = $this->updateForCommand($command);
+ $retained = [
+ 'verified' => $retainedReceipt['intent'] === CommandRequestIdentity::INTENT_SAME,
+ ];
+
+ return;
+ }
- if ($existingCommand instanceof WorkflowCommand) {
- $command = $existingCommand;
- $update = $existingUpdate;
+ $command = $this->requestIdConflictSnapshot($instance, $retainedReceipt['command'], $method);
return;
}
@@ -2477,7 +2549,9 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
CommandType::Update,
'instance_not_started',
$this->commandTargetScope(),
- $this->updateCommandPayloadAttributes($method, $arguments),
+ array_merge($this->updateCommandPayloadAttributes($method, $arguments), [
+ 'intent_digest' => $this->updateIntentDigest(null, $method, $arguments),
+ ]),
);
$update = $this->recordRejectedUpdate($command, $method, $arguments);
@@ -2492,6 +2566,7 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
->findOrFail($this->selectedRunId);
$updateName = $method;
$updateCommandAttributes = $this->updateCommandPayloadAttributes($updateName, $arguments);
+ $updateCommandAttributes['intent_digest'] = $this->updateIntentDigest($run, $updateName, $arguments);
if ($run->id !== $currentRun->id) {
[$command, $update] = $this->rejectUpdateCommand(
@@ -2512,6 +2587,7 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
$run = $currentRun;
$updateName = $method;
$updateCommandAttributes = $this->updateCommandPayloadAttributes($updateName, $arguments);
+ $updateCommandAttributes['intent_digest'] = $this->updateIntentDigest($run, $updateName, $arguments);
}
if (in_array($run->status, [
@@ -2557,6 +2633,10 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
$arguments,
$validatedArguments['validation_errors'],
);
+ $updateCommandAttributes['intent_digest'] = $this->updateIntentDigestFromArguments(
+ $updateName,
+ $arguments
+ );
[$command, $update] = $this->rejectUpdateCommand(
$instance,
@@ -2583,10 +2663,15 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
[],
'structural_limit_exceeded',
$this->commandTargetScope(),
- $this->structuralLimitExceededCommandPayloadAttributes(
- $updateName,
- $run->payload_codec ?? CodecRegistry::defaultCodec(),
- $e,
+ array_merge(
+ $this->structuralLimitExceededCommandPayloadAttributes(
+ $updateName,
+ $run->payload_codec ?? CodecRegistry::defaultCodec(),
+ $e,
+ ),
+ [
+ 'intent_digest' => $updateCommandAttributes['intent_digest'] ?? null,
+ ],
),
[
'arguments' => [$e->getMessage()],
@@ -2619,7 +2704,9 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
[],
'structural_limit_exceeded',
$this->commandTargetScope(),
- $this->structuralLimitExceededCommandPayloadAttributes($updateName, $updateCodec, $e),
+ array_merge($this->structuralLimitExceededCommandPayloadAttributes($updateName, $updateCodec, $e), [
+ 'intent_digest' => $updateCommandAttributes['intent_digest'] ?? null,
+ ]),
[
'arguments' => [$e->getMessage()],
],
@@ -2700,7 +2787,7 @@ private function recordAcceptedUpdateWithArguments(string $method, array $argume
self::projectRun($run, self::PROJECTION_RUN_RELATIONS);
});
- return [$command, $update, $resumeTask];
+ return [$command, $update, $resumeTask, $retained];
}
private function shouldInlineAcceptedUpdateCompletion(WorkflowUpdate $update): bool
@@ -2716,38 +2803,12 @@ private function shouldInlineAcceptedUpdateCompletion(WorkflowUpdate $update): b
|| ! (UpdateCommandGate::blockingSignal($run, $update->command_sequence) instanceof WorkflowCommand);
}
- /**
- * @return array{0: WorkflowCommand|null, 1: WorkflowUpdate|null}
- */
- private function existingUpdateForRequest(WorkflowInstance $instance): array
+ private function updateForCommand(WorkflowCommand $command): ?WorkflowUpdate
{
- $context = $this->resolvedCommandContext()
- ->attributes()['context'] ?? [];
- $serviceCallIdentity = ServiceCallEffectIdentity::fromContext($context);
- $bridgeIdentity = BridgeCommandIdentity::fromContext($context);
- $command = $serviceCallIdentity !== null
- ? ConfiguredV2Models::query('command_model', WorkflowCommand::class)
- ->where('workflow_instance_id', $instance->id)
- ->where('command_type', CommandType::Update->value)
- ->where('request_id', $serviceCallIdentity->storageToken())
- ->lockForUpdate()
- ->first()
- : ($bridgeIdentity !== null ? $bridgeIdentity->find(
- $instance,
- CommandType::Update
- ) : UpdateRequestIdentity::find($instance, $context));
-
- if (! $command instanceof WorkflowCommand) {
- return [null, null];
- }
-
- /** @var WorkflowUpdate|null $update */
- $update = ConfiguredV2Models::query('update_model', WorkflowUpdate::class)
+ return ConfiguredV2Models::query('update_model', WorkflowUpdate::class)
->where('workflow_command_id', $command->id)
->lockForUpdate()
->first();
-
- return [$command, $update];
}
private function processAcceptedUpdateInline(?string $runId, string $updateId): void
@@ -2881,6 +2942,8 @@ private function attemptSignalInternal(
?string $payloadBlob = null,
bool $runtimeReserved = false,
): CommandResult {
+ $this->validateCommandRequestId(CommandType::Signal);
+
if ($name === '') {
throw new LogicException('Signal name cannot be empty.');
}
@@ -2892,6 +2955,8 @@ private function attemptSignalInternal(
/** @var WorkflowCommand|null $command */
$command = null;
$task = null;
+ /** @var array{verified: bool}|null $retained */
+ $retained = null;
$observabilityContext = $runtimeReserved ? null : TraceCarrier::captureConfigured();
SqliteTransaction::run(function () use (
@@ -2902,10 +2967,12 @@ private function attemptSignalInternal(
$runtimeReserved,
$observabilityContext,
&$command,
- &$task
+ &$task,
+ &$retained
): void {
$command = null;
$task = null;
+ $retained = null;
/** @var WorkflowInstance $instance */
$instance = self::instanceQuery()
@@ -2931,6 +2998,34 @@ private function attemptSignalInternal(
}
}
+ // A keyed retry is answered from its retained receipt before any run
+ // resolution, so closure or continuation never admits it again.
+ $retainedReceipt = $this->retainedCommandForRequest(
+ $instance,
+ CommandType::Signal,
+ fn (?WorkflowRun $retainedRun): string => $this->signalIntentDigest(
+ $retainedRun,
+ $name,
+ $arguments,
+ $runtimeReserved
+ ),
+ );
+
+ if ($retainedReceipt !== null) {
+ if ($retainedReceipt['intent'] !== CommandRequestIdentity::INTENT_DIFFERENT) {
+ $command = $retainedReceipt['command'];
+ $retained = [
+ 'verified' => $retainedReceipt['intent'] === CommandRequestIdentity::INTENT_SAME,
+ ];
+
+ return;
+ }
+
+ $command = $this->requestIdConflictSnapshot($instance, $retainedReceipt['command'], $name);
+
+ return;
+ }
+
$target = $this->signalRunOrRejection($instance, $name, $arguments, $payloadCodec);
if ($target instanceof WorkflowCommand) {
$command = $target;
@@ -3026,7 +3121,14 @@ private function attemptSignalInternal(
);
$serializedSignalArguments = $payloadBlob ?? Serializer::serializeWithCodec($signalCodec, $arguments);
- $command = $this->rejectOversizedSignal($instance, $run, $name, $signalCodec, $serializedSignalArguments);
+ $command = $this->rejectOversizedSignal(
+ $instance,
+ $run,
+ $name,
+ $signalCodec,
+ $serializedSignalArguments,
+ $arguments
+ );
if ($command instanceof WorkflowCommand) {
return;
}
@@ -3088,6 +3190,10 @@ private function attemptSignalInternal(
$result = new CommandResult($command);
+ if ($retained !== null) {
+ return $result->asRetainedReceipt($retained['verified']);
+ }
+
if ($result->accepted()) {
self::recordSignalSent($this->instance->id, $name, $arguments);
}
@@ -3199,6 +3305,7 @@ private function recordAcceptedSignalCommand(
$storedExternally = $payloadBlob !== null && ExternalPayloads::isStoredReference($payloadBlob);
$attributes = $this->commandAttributes([
'command_type' => CommandType::Signal->value,
+ 'intent_digest' => $this->signalIntentDigestFromArguments($name, $arguments),
'observability_context' => $observabilityContext,
'target_scope' => $this->commandTargetScope(),
'status' => CommandStatus::Accepted->value,
@@ -3220,12 +3327,16 @@ private function recordAcceptedSignalCommand(
return WorkflowCommand::record($instance, $run, $attributes);
}
+ /**
+ * @param array $signalArguments
+ */
private function rejectOversizedSignal(
WorkflowInstance $instance,
WorkflowRun $run,
string $name,
string $signalCodec,
string $arguments,
+ array $signalArguments,
): ?WorkflowCommand {
StructuralLimits::logWarning(StructuralLimits::warnApproachingPayloadSize($arguments), [
'workflow_run_id' => $run->id,
@@ -3243,7 +3354,9 @@ private function rejectOversizedSignal(
CommandType::Signal,
'structural_limit_exceeded',
$this->commandTargetScope(),
- $this->structuralLimitExceededCommandPayloadAttributes($name, $signalCodec, $e),
+ array_merge($this->structuralLimitExceededCommandPayloadAttributes($name, $signalCodec, $e), [
+ 'intent_digest' => $this->signalIntentDigestFromArguments($name, $signalArguments),
+ ]),
);
$this->recordRejectedSignal($command, $name, [], [
'arguments' => [$e->getMessage()],
@@ -3265,6 +3378,8 @@ private function attemptSignalWithStartInternal(
array $startArguments,
StartOptions $startOptions,
): SignalWithStartResult {
+ $this->validateCommandRequestId(CommandType::Signal);
+
if ($name === '') {
throw new LogicException('Signal name cannot be empty.');
}
@@ -3308,6 +3423,8 @@ private function admitSignalWithStart(
$signalCommand = null;
$task = null;
$startedRun = null;
+ /** @var array{verified: bool}|null $retained */
+ $retained = null;
ConfiguredV2Models::connection()->transaction(function () use (
$name,
@@ -3320,11 +3437,13 @@ private function admitSignalWithStart(
&$signalCommand,
&$task,
&$startedRun,
+ &$retained,
): void {
$task = null;
$startCommand = null;
$signalCommand = null;
$startedRun = null;
+ $retained = null;
/** @var WorkflowInstance $instance */
$instance = self::instanceQuery()
@@ -3332,6 +3451,34 @@ private function admitSignalWithStart(
->findOrFail($this->instance->id);
$commandContext = $this->signalWithStartCommandContext($intakeGroupId);
+ // A keyed retry returns the original intake instead of recording a
+ // second start/signal pair under a new intake group.
+ $retainedReceipt = $this->retainedCommandForRequest(
+ $instance,
+ CommandType::Signal,
+ fn (?WorkflowRun $retainedRun): string => $this->signalIntentDigest(
+ $retainedRun,
+ $name,
+ $signalArguments
+ ),
+ );
+
+ if ($retainedReceipt !== null) {
+ if ($retainedReceipt['intent'] !== CommandRequestIdentity::INTENT_DIFFERENT) {
+ $signalCommand = $retainedReceipt['command'];
+ $startCommand = $this->intakeStartCommand($signalCommand);
+ $retained = [
+ 'verified' => $retainedReceipt['intent'] === CommandRequestIdentity::INTENT_SAME,
+ ];
+
+ return;
+ }
+
+ $signalCommand = $this->requestIdConflictSnapshot($instance, $retainedReceipt['command'], $name);
+
+ return;
+ }
+
$currentRun = $this->currentRunForInstance($instance, true);
if ($currentRun instanceof WorkflowRun && $this->runIsActive($currentRun)) {
@@ -3376,7 +3523,16 @@ private function admitSignalWithStart(
));
}
- $signalWithStartResult = SignalWithStartResult::fromCommands($signalCommand, $startCommand, $intakeGroupId);
+ if ($retained !== null) {
+ return SignalWithStartResult::fromCommands($signalCommand, $startCommand, $signalCommand->intakeGroupId())
+ ->asRetainedReceipt($retained['verified']);
+ }
+
+ $signalWithStartResult = SignalWithStartResult::fromCommands(
+ $signalCommand,
+ $startCommand,
+ $signalCommand->rejection_reason === CommandRequestIdentity::CONFLICT_REASON ? null : $intakeGroupId,
+ );
if ($signalWithStartResult->accepted()) {
self::recordSignalSent($this->instance->id, $name, $signalArguments);
@@ -4294,6 +4450,7 @@ private function rejectionOutcome(string $reason): ?string
'unknown_update' => CommandOutcome::RejectedUnknownUpdate->value,
'invalid_signal_arguments' => CommandOutcome::RejectedInvalidArguments->value,
'invalid_update_arguments' => CommandOutcome::RejectedInvalidArguments->value,
+ CommandRequestIdentity::CONFLICT_REASON => CommandOutcome::RejectedRequestIdConflict->value,
WorkflowStartGate::BLOCKED_COMPATIBILITY => CommandOutcome::RejectedCompatibilityBlocked->value,
WorkflowExecutionGate::BLOCKED_WORKFLOW_DEFINITION_UNAVAILABLE => CommandOutcome::RejectedWorkflowDefinitionUnavailable->value,
default => null,
@@ -4326,6 +4483,7 @@ private function rejectSignalCommandForContext(
'payload_codec' => $run?->payload_codec ?? CodecRegistry::defaultCodec(),
'rejection_reason' => $reason,
'rejected_at' => now(),
+ 'intent_digest' => CommandIntent::digest(CommandType::Signal, $name, $arguments),
...$this->signalCommandPayloadAttributes(
$name,
$arguments,
@@ -4350,10 +4508,220 @@ private function rejectSignalCommandForContext(
}
/**
- * @param array $arguments
- * @param array> $validationErrors
+ * @param array $context
+ */
+ private function usesInternalCommandIdentity(array $context): bool
+ {
+ return ServiceCallEffectIdentity::fromContext($context) !== null
+ || BridgeCommandIdentity::fromContext($context) !== null;
+ }
+
+ private function existingInternalCommandForRequest(WorkflowInstance $instance, CommandType $type): ?WorkflowCommand
+ {
+ $context = $this->resolvedCommandContext()
+ ->attributes()['context'];
+ $serviceCall = ServiceCallEffectIdentity::fromContext($context);
+ if ($serviceCall !== null) {
+ return ConfiguredV2Models::query('command_model', WorkflowCommand::class)
+ ->where('workflow_instance_id', $instance->id)
+ ->where('command_type', $type->value)
+ ->where('request_id', $serviceCall->storageToken())
+ ->lockForUpdate()
+ ->first();
+ }
+
+ return BridgeCommandIdentity::fromContext($context)?->find($instance, $type);
+ }
+
+ private function validateCommandRequestId(CommandType $type): void
+ {
+ $context = $this->resolvedCommandContext()
+ ->attributes()['context'];
+ if ($this->usesInternalCommandIdentity($context)) {
+ return;
+ }
+ CommandRequestIdentity::validateContext($context, $type);
+ }
+
+ /**
+ * Find the receipt retained for the caller's command key, if any, and
+ * compare the retry's intent against it. Called under the instance lock.
+ *
+ * @param callable(WorkflowRun|null): string $digestForRun
+ * @return array{command: WorkflowCommand, run: WorkflowRun|null, intent: string, digest: string}|null
+ */
+ private function retainedCommandForRequest(
+ WorkflowInstance $instance,
+ CommandType $type,
+ callable $digestForRun,
+ ): ?array {
+ $context = $this->resolvedCommandContext()
+ ->attributes()['context'];
+ if ($this->usesInternalCommandIdentity($context)) {
+ return null;
+ }
+ $retained = CommandRequestIdentity::find($instance, $type, $context);
+
+ if (! $retained instanceof WorkflowCommand) {
+ return null;
+ }
+
+ $run = is_string($retained->workflow_run_id)
+ ? self::runQuery()->find($retained->workflow_run_id)
+ : null;
+ $run = $run instanceof WorkflowRun ? $run : null;
+ $digest = $digestForRun($run);
+
+ return [
+ 'command' => $retained,
+ 'run' => $run,
+ 'intent' => CommandRequestIdentity::compareIntent($retained, $digest),
+ 'digest' => $digest,
+ ];
+ }
+
+ /**
* @return array
*/
+ private function requestIdConflictPayload(WorkflowCommand $retained, string $name): array
+ {
+ return [
+ 'retained_command_id' => $retained->id,
+ 'retained_command_type' => $retained->command_type?->value,
+ 'retained_outcome' => $retained->outcome?->value ?? $retained->rejection_reason,
+ 'conflict' => CommandRequestIdentity::conflictingParts(
+ $retained,
+ $name,
+ $this->commandTargetScope(),
+ $this->runTargeted ? $this->selectedRunId : null,
+ ),
+ ];
+ }
+
+ /**
+ * A conflict is a response associated with the retained command, never a
+ * new durable command. The snapshot keeps that ID for the embedded result
+ * API but has no new sequence, update/signal record or admission timestamp.
+ * CommandResponse exposes the retained ID in its explicit conflict field.
+ */
+ private function requestIdConflictSnapshot(
+ WorkflowInstance $instance,
+ WorkflowCommand $retained,
+ string $name,
+ ): WorkflowCommand {
+ return $retained->newInstance($this->commandAttributes(array_merge([
+ 'id' => $retained->id,
+ 'workflow_instance_id' => $instance->id,
+ 'workflow_run_id' => $retained->workflow_run_id,
+ 'workflow_type' => $retained->workflow_type,
+ 'workflow_class' => $retained->workflow_class,
+ 'command_type' => $retained->command_type,
+ 'target_scope' => $this->commandTargetScope(),
+ 'requested_workflow_run_id' => $this->runTargeted ? $this->selectedRunId : null,
+ 'resolved_workflow_run_id' => $instance->current_run_id,
+ 'status' => CommandStatus::Rejected,
+ 'outcome' => CommandOutcome::RejectedRequestIdConflict,
+ 'rejection_reason' => CommandRequestIdentity::CONFLICT_REASON,
+ ], $this->signalCommandPayloadAttributes(
+ $name,
+ [],
+ extraPayload: $this->requestIdConflictPayload($retained, $name),
+ ))));
+ }
+
+ /**
+ * Intent of a signal as it would be admitted against the given run: the
+ * validated arguments when the signal is declared and valid, otherwise the
+ * arguments as submitted. The admission paths record the same digest.
+ *
+ * @param array $arguments
+ */
+ private function signalIntentDigest(
+ ?WorkflowRun $run,
+ string $name,
+ array $arguments,
+ bool $runtimeReserved = false,
+ ): string {
+ $normalized = $arguments;
+
+ if ($run instanceof WorkflowRun && ! $runtimeReserved) {
+ $admission = $this->signalAdmissionForRun($run, $name);
+
+ if (($admission['allowed'] ?? false) === true) {
+ $validated = $this->validatedSignalArgumentsForRun($run, $name, $arguments);
+
+ if ($validated['validation_errors'] === []) {
+ $normalized = $validated['arguments'];
+ }
+ }
+ }
+
+ return $this->signalIntentDigestFromArguments($name, $normalized);
+ }
+
+ /**
+ * @param array $arguments
+ */
+ private function signalIntentDigestFromArguments(string $name, array $arguments): string
+ {
+ return CommandIntent::digest(
+ CommandType::Signal,
+ $name,
+ $arguments,
+ $this->commandTargetScope(),
+ $this->runTargeted ? $this->selectedRunId : null,
+ );
+ }
+
+ /**
+ * @param array $arguments
+ */
+ private function updateIntentDigest(?WorkflowRun $run, string $updateName, array $arguments): string
+ {
+ $normalized = $arguments;
+
+ if ($run instanceof WorkflowRun && RunCommandContract::hasUpdateMethod($run, $updateName)) {
+ $validated = $this->validatedUpdateArgumentsForRun($run, $updateName, $arguments);
+
+ if ($validated['validation_errors'] === []) {
+ $normalized = $validated['arguments'];
+ }
+ }
+
+ return $this->updateIntentDigestFromArguments($updateName, $normalized);
+ }
+
+ /**
+ * @param array $arguments
+ */
+ private function updateIntentDigestFromArguments(string $updateName, array $arguments): string
+ {
+ return CommandIntent::digest(
+ CommandType::Update,
+ $updateName,
+ $arguments,
+ $this->commandTargetScope(),
+ $this->runTargeted ? $this->selectedRunId : null,
+ );
+ }
+
+ private function intakeStartCommand(WorkflowCommand $signalCommand): ?WorkflowCommand
+ {
+ $groupId = $signalCommand->intakeGroupId();
+
+ if ($groupId === null) {
+ return null;
+ }
+
+ return ConfiguredV2Models::query('command_model', WorkflowCommand::class)
+ ->where('workflow_instance_id', $signalCommand->workflow_instance_id)
+ ->where('command_type', CommandType::Start->value)
+ ->where('context->intake->group_id', $groupId)
+ ->orderBy('created_at')
+ ->orderBy('id')
+ ->first();
+ }
+
private function signalCommandPayloadAttributes(
string $name,
array $arguments,
@@ -4364,6 +4732,7 @@ private function signalCommandPayloadAttributes(
$codec = $payloadCodec ?? CodecRegistry::defaultCodec();
return [
+ 'intent_digest' => $this->signalIntentDigestFromArguments($name, $arguments),
'payload_codec' => $codec,
'payload' => Serializer::serializeWithCodec($codec, array_merge([
'name' => $name,
@@ -4410,14 +4779,15 @@ private function updateCommandPayloadAttributes(
string $method,
array $arguments,
array $validationErrors = [],
+ array $extraPayload = [],
): array {
return [
'payload_codec' => CodecRegistry::defaultCodec(),
- 'payload' => Serializer::serializeWithCodec(CodecRegistry::defaultCodec(), [
+ 'payload' => Serializer::serializeWithCodec(CodecRegistry::defaultCodec(), array_merge([
'name' => $method,
'arguments' => $arguments,
'validation_errors' => $validationErrors,
- ]),
+ ], $extraPayload)),
];
}
diff --git a/src/migrations/2026_10_20_000100_add_intent_digest_to_workflow_commands.php b/src/migrations/2026_10_20_000100_add_intent_digest_to_workflow_commands.php
new file mode 100644
index 000000000..4c3348ea1
--- /dev/null
+++ b/src/migrations/2026_10_20_000100_add_intent_digest_to_workflow_commands.php
@@ -0,0 +1,31 @@
+string('intent_digest', 80)
+ ->nullable()
+ ->after('request_id');
+ }
+ });
+ }
+
+ public function down(): void
+ {
+ Schema::table('workflow_commands', static function (Blueprint $table): void {
+ if (Schema::hasColumn('workflow_commands', 'intent_digest')) {
+ $table->dropColumn('intent_digest');
+ }
+ });
+ }
+};
diff --git a/tests/Feature/V2/V2WorkflowTaskBridgeTest.php b/tests/Feature/V2/V2WorkflowTaskBridgeTest.php
index a820dfc1d..a64fb44ce 100644
--- a/tests/Feature/V2/V2WorkflowTaskBridgeTest.php
+++ b/tests/Feature/V2/V2WorkflowTaskBridgeTest.php
@@ -6577,11 +6577,11 @@ public function testInitiallyLeasedExternalRunAppliesDeclaredSignalsBeforeQueued
);
}
- public function testRequestIdIdempotencyColumnIsReservedForUpdateCommands(): void
+ public function testRequestIdIdempotencyColumnIsReservedForSignalAndUpdateCommands(): void
{
$run = $this->createWaitingRun();
$instance = WorkflowInstance::query()->findOrFail($run->workflow_instance_id);
- $commandTypes = ['start', 'signal', 'repair', 'cancel', 'terminate', 'archive'];
+ $commandTypes = ['start', 'signal', 'update', 'repair', 'cancel', 'terminate', 'archive'];
foreach ($commandTypes as $commandType) {
for ($attempt = 0; $attempt < 2; $attempt++) {
@@ -6597,12 +6597,16 @@ public function testRequestIdIdempotencyColumnIsReservedForUpdateCommands(): voi
'accepted_at' => now(),
]);
- $this->assertNull($command->getAttribute('request_id'));
+ if (in_array($commandType, ['signal', 'update'], true)) {
+ $this->assertStringStartsWith($commandType . '-v1:', $command->getAttribute('request_id'));
+ } else {
+ $this->assertNull($command->getAttribute('request_id'));
+ }
$this->assertSame('shared-' . $commandType . '-request', $command->requestId());
}
}
- $this->assertSame(12, WorkflowCommand::query()
+ $this->assertSame(14, WorkflowCommand::query()
->where('workflow_instance_id', $instance->id)
->whereIn('command_type', $commandTypes)
->count());
diff --git a/tests/Unit/V2/BridgeCommandIdentityTest.php b/tests/Unit/V2/BridgeCommandIdentityTest.php
index 7bb709bec..329f10e00 100644
--- a/tests/Unit/V2/BridgeCommandIdentityTest.php
+++ b/tests/Unit/V2/BridgeCommandIdentityTest.php
@@ -111,9 +111,20 @@ public function testJsonAndOrdinaryRequestIdsCannotCreateABridgeIdentity(): void
$second = $control->signal($workflow->id(), 'finish', [
'command_context' => $context,
]);
- $this->assertNotSame($first['command_id'], $second['command_id']);
+ // A JSON object cannot become a host-created bridge identity. Its
+ // ordinary caller key still follows the signal receipt contract.
+ $this->assertSame($first['command_id'], $second['command_id']);
+ $this->assertTrue($second['retained']);
+ $command = WorkflowCommand::query()->findOrFail($first['command_id']);
+ $this->assertStringStartsWith('signal-v1:', $command->request_id);
+ $this->assertNotSame($key, $command->request_id);
+
+ $differentKey = $control->signal($workflow->id(), 'finish', [
+ 'command_context' => $context,
+ 'request_id' => 'different-caller-key',
+ ]);
+ $this->assertNotSame($first['command_id'], $differentKey['command_id']);
$this->assertSame(2, WorkflowSignal::query()->count());
- $this->assertNull(WorkflowCommand::query()->findOrFail($first['command_id'])->request_id);
}
public function testBridgeIdentityCannotOccupyThePrivilegedServiceCallNamespace(): void
@@ -145,6 +156,12 @@ public function testLegacyAcceptedReceiptIsReusedWithoutChangingItsIdentityOrEff
'arguments' => ['first'],
]);
$this->assertTrue($first['accepted']);
+ // Recreate the pre-command-key receipt shape rather than relying on
+ // today's signal admission to omit its indexed identity and digest.
+ WorkflowCommand::query()->findOrFail($first['command_id'])->forceFill([
+ 'request_id' => null,
+ 'intent_digest' => null,
+ ])->save();
$this->assertNull(WorkflowCommand::query()->findOrFail($first['command_id'])->request_id);
$before = $this->rows();
$second = $control->signal($workflow->id(), 'advance', [
diff --git a/tests/Unit/V2/CommandIntentTest.php b/tests/Unit/V2/CommandIntentTest.php
new file mode 100644
index 000000000..97b4e4391
--- /dev/null
+++ b/tests/Unit/V2/CommandIntentTest.php
@@ -0,0 +1,152 @@
+ 2,
+ 'a' => 1,
+ ]]);
+
+ $this->assertTrue(CommandIntent::isDigest($digest));
+ $this->assertStringStartsWith(CommandIntent::VERSION . ':', $digest);
+ $this->assertSame($digest, CommandIntent::digest(CommandType::Signal, 'changed', [
+ 1, 'Ünïcødé ☃ 日本', null, [
+ 'a' => 1,
+ 'b' => 2,
+ ]]));
+ $this->assertFalse(CommandIntent::isDigest('update-v1:' . hash('sha256', 'x')));
+ }
+
+ public static function differentIntents(): iterable
+ {
+ yield 'type' => [[CommandType::Signal, 'n', [1]], [CommandType::Update, 'n', [1]]];
+ yield 'name' => [[CommandType::Signal, 'n', [1]], [CommandType::Signal, 'm', [1]]];
+ yield 'int vs float' => [[CommandType::Signal, 'n', [1]], [CommandType::Signal, 'n', [1.0]]];
+ yield 'int vs string' => [[CommandType::Signal, 'n', [1]], [CommandType::Signal, 'n', ['1']]];
+ yield 'null vs false' => [[CommandType::Signal, 'n', [null]], [CommandType::Signal, 'n', [false]]];
+ yield 'null vs empty list' => [[CommandType::Signal, 'n', [null]], [CommandType::Signal, 'n', [[]]]];
+ yield 'unicode bytes' => [[CommandType::Signal, 'n', ["e\u{301}"]], [CommandType::Signal, 'n', ["\u{e9}"]]];
+ yield 'list vs map' => [[CommandType::Signal, 'n', [[1, 2]]], [
+ CommandType::Signal, 'n', [[
+ '0' => 1,
+ 'x' => 2,
+ ]]]];
+ yield 'int64 edge' => [[CommandType::Signal, 'n', [PHP_INT_MAX]],
+ [CommandType::Signal, 'n', [PHP_INT_MAX - 1]]];
+ yield 'bytes vs string' => [[CommandType::Signal, 'n', [AvroBinaryValue::fromBytes('ab')]],
+ [CommandType::Signal, 'n', ['ab']]];
+ yield 'run target' => [[CommandType::Signal, 'n', [1], 'instance', null],
+ [CommandType::Signal, 'n', [1], 'run', 'run-a'],
+ ];
+ yield 'different run' => [[CommandType::Signal, 'n', [1], 'run', 'run-a'],
+ [CommandType::Signal, 'n', [1], 'run', 'run-b'],
+ ];
+ yield 'nested map value' => [[
+ CommandType::Signal, 'n', [[
+ 'k' => [1],
+ ]]], [
+ CommandType::Signal, 'n', [[
+ 'k' => [2],
+ ]]]];
+ }
+
+ #[DataProvider('differentIntents')]
+ public function testDifferentIntentsDigestDifferently(array $left, array $right): void
+ {
+ $this->assertNotSame(CommandIntent::digest(...$left), CommandIntent::digest(...$right));
+ }
+
+ public function testMapOrderAndAvroMapWrappersDoNotChangeIntent(): void
+ {
+ $plain = CommandIntent::digest(CommandType::Update, 'rename', [[
+ 'z' => 1,
+ 'a' => [2, 3],
+ ]]);
+ $reordered = CommandIntent::digest(CommandType::Update, 'rename', [[
+ 'a' => [2, 3],
+ 'z' => 1,
+ ]]);
+ $wrapped = CommandIntent::digest(
+ CommandType::Update,
+ 'rename',
+ [AvroMapValue::fromPairs([['z', 1], ['a', [2, 3]]])]
+ );
+
+ $this->assertSame($plain, $reordered);
+ $this->assertSame($plain, $wrapped);
+ }
+
+ public function testFloatsKeepTheirExactBigEndianRepresentation(): void
+ {
+ $this->assertSame('d0x3fb999999999999a', CommandIntent::canonical(0.1));
+ $this->assertSame('d0x3ff0000000000000', CommandIntent::canonical(1.0));
+ $this->assertSame('d0x8000000000000000', CommandIntent::canonical(-0.0));
+ $this->assertSame('dnan', CommandIntent::canonical(NAN));
+ $this->assertSame('d-inf', CommandIntent::canonical(-INF));
+ $this->assertSame('i' . PHP_INT_MIN, CommandIntent::canonical(PHP_INT_MIN));
+ $this->assertSame('s5:a' . "\x00" . 'b' . "\n" . 'c', CommandIntent::canonical("a\x00b\nc"));
+ }
+
+ public function testProtectedValuesContributeTheirDeclaredIntentInsteadOfBytes(): void
+ {
+ $protected = [
+ CommandIntent::PROTECTED_INTENT_KEY => 'plain-digest',
+ ];
+
+ $this->assertSame('p12:plain-digest', CommandIntent::canonical($protected));
+ $this->assertSame(
+ CommandIntent::digest(CommandType::Signal, 'n', [$protected]),
+ CommandIntent::digest(CommandType::Signal, 'n', [[
+ CommandIntent::PROTECTED_INTENT_KEY => 'plain-digest',
+ ]]),
+ );
+ $this->assertNotSame(
+ CommandIntent::canonical($protected),
+ CommandIntent::canonical([
+ CommandIntent::PROTECTED_INTENT_KEY => 'plain-digest',
+ 'extra' => 1,
+ ]),
+ );
+ }
+
+ public function testFloatIntentDoesNotDependOnPhpSerializationPrecision(): void
+ {
+ $original = ini_get('serialize_precision');
+ try {
+ ini_set('serialize_precision', '-1');
+ $expected = CommandIntent::digest(CommandType::Signal, 'changed', [0.1234567890123456]);
+ ini_set('serialize_precision', '6');
+ $this->assertSame($expected, CommandIntent::digest(CommandType::Signal, 'changed', [0.1234567890123456]));
+ } finally {
+ ini_set('serialize_precision', $original);
+ }
+ }
+
+ public function testLowSerializationPrecisionCannotMergeDifferentFloatIntents(): void
+ {
+ $original = ini_get('serialize_precision');
+ try {
+ ini_set('serialize_precision', '6');
+ $this->assertNotSame(
+ CommandIntent::digest(CommandType::Signal, 'changed', [0.1234567890123456]),
+ CommandIntent::digest(CommandType::Signal, 'changed', [0.1234567890123457]),
+ );
+ } finally {
+ ini_set('serialize_precision', $original);
+ }
+ }
+}
diff --git a/tests/Unit/V2/CommandRequestIdentityTest.php b/tests/Unit/V2/CommandRequestIdentityTest.php
new file mode 100644
index 000000000..7103b06bd
--- /dev/null
+++ b/tests/Unit/V2/CommandRequestIdentityTest.php
@@ -0,0 +1,856 @@
+set('queue.default', 'redis');
+ Queue::fake();
+ }
+
+ public function testKeyedSignalRetryReturnsTheOriginalAdmission(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withRequestId('caller-key')
+ ->signal('changed', 1);
+ $before = $this->rows();
+
+ $this->assertTrue($first->accepted());
+ $this->assertFalse($first->retained());
+ $this->assertTrue($first->intentVerified());
+ $this->assertSame('caller-key', $first->requestId());
+
+ $retry = $workflow->withRequestId('caller-key')
+ ->signal('changed', 1);
+
+ $this->assertTrue($retry->accepted());
+ $this->assertTrue($retry->retained());
+ $this->assertTrue($retry->intentVerified());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame($first->commandSequence(), $retry->commandSequence());
+ $this->assertSame('caller-key', $retry->requestId());
+ $this->assertSame($before, $this->rows());
+ $this->assertSame(1, $this->signalReceivedCount($workflow));
+
+ $command = WorkflowCommand::query()->findOrFail($first->commandId());
+ $this->assertStringStartsWith('signal-v1:', (string) $command->request_id);
+ $this->assertLessThanOrEqual(191, strlen((string) $command->request_id));
+ $this->assertTrue(CommandIntent::isDigest($command->intentDigest()));
+ $this->assertSame('caller-key', $command->context['request']['request_id']);
+ }
+
+ public function testUnkeyedSignalsStayIndependent(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->signal('changed', 1);
+ $second = $workflow->signal('changed', 1);
+
+ $this->assertNull($first->requestId());
+ $this->assertFalse($second->retained());
+ $this->assertNotSame($first->commandId(), $second->commandId());
+ $this->assertSame(2, $this->signalReceivedCount($workflow));
+ }
+
+ #[DataProvider('invalidCallerKeys')]
+ public function testInvalidCallerKeysAreRefusedWithoutDurableWrites(string $route, mixed $key): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $before = $this->rows();
+
+ if (str_starts_with($route, 'plane-')) {
+ $result = $this->keyedAdmission($workflow, $route, $key);
+ $this->assertSame(422, $result['status']);
+ $this->assertSame('invalid_request_id', $result['reason']);
+ $this->assertFalse($result['accepted']);
+ $this->assertArrayHasKey('request_id', $result['validation_errors']);
+ $this->assertSame($before, $this->rows());
+ json_encode($result, JSON_THROW_ON_ERROR);
+
+ return;
+ }
+
+ try {
+ $this->keyedAdmission($workflow, $route, $key);
+ $this->fail('An invalid caller key must fail before admission.');
+ } catch (InvalidCommandRequestIdException $exception) {
+ $this->assertArrayHasKey('request_id', $exception->validationErrors());
+ $this->assertSame($before, $this->rows());
+ }
+ }
+
+ /**
+ * @return iterable
+ */
+ public static function invalidCallerKeys(): iterable
+ {
+ foreach (self::keyRoutes() as $route) {
+ $keys = [
+ 'empty' => '',
+ 'malformed-utf8' => "\xFF",
+ 'ascii-256' => str_repeat('a', 256),
+ 'unicode-256' => str_repeat('雪', 256),
+ 'four-byte-256' => str_repeat('🧭', 256),
+ ];
+ if (! str_starts_with($route, 'stub-')) {
+ $keys += [
+ 'not-string-int' => 0,
+ 'not-string-bool' => false,
+ 'not-string-array' => [],
+ ];
+ }
+ foreach ($keys as $case => $key) {
+ yield $route . '-' . $case => [$route, $key];
+ }
+ }
+ }
+
+ #[DataProvider('maximumCallerKeys')]
+ public function testMaximumUnicodeCallerKeysRetainTheirExactAdmission(string $route, string $key): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $this->keyedAdmission($workflow, $route, $key);
+ $before = $this->rows();
+ $retry = $this->keyedAdmission($workflow, $route, $key);
+
+ $this->assertSame($key, $first['request_id']);
+ $this->assertFalse($first['retained']);
+ $this->assertTrue($retry['retained']);
+ $this->assertSame($first['command_id'], $retry['command_id']);
+ $this->assertSame($key, $retry['request_id']);
+ $this->assertSame($before, $this->rows());
+ }
+
+ /**
+ * @return iterable
+ */
+ public static function maximumCallerKeys(): iterable
+ {
+ foreach (self::keyRoutes() as $route) {
+ foreach (['a', '雪', '🧭'] as $character) {
+ yield $route . '-' . strlen($character) . '-byte' => [$route, str_repeat($character, 255)];
+ }
+ }
+ }
+
+ #[DataProvider('callerKeyRoutes')]
+ public function testCallerKeysRemainCaseAndNormalizationDistinct(string $route): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $commandIds = [];
+ foreach (['CASE', 'case', "\u{00E9}", "e\u{0301}", '0'] as $key) {
+ $first = $this->keyedAdmission($workflow, $route, $key);
+ $retry = $this->keyedAdmission($workflow, $route, $key);
+ $this->assertSame($key, $retry['request_id']);
+ $this->assertTrue($retry['retained']);
+ $this->assertSame($first['command_id'], $retry['command_id']);
+ $commandIds[] = $first['command_id'];
+ }
+
+ $this->assertCount(5, array_unique($commandIds));
+ }
+
+ /**
+ * @return iterable
+ */
+ public static function callerKeyRoutes(): iterable
+ {
+ foreach (self::keyRoutes() as $route) {
+ yield $route => [$route];
+ }
+ }
+
+ #[DataProvider('callerKeyRoutes')]
+ public function testNullCallerKeysRemainUnkeyedAcrossAllRoutes(string $route): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $this->keyedAdmission($workflow, $route, null);
+ $second = $this->keyedAdmission($workflow, $route, null);
+
+ $this->assertNull($first['request_id']);
+ $this->assertNull($second['request_id']);
+ $this->assertFalse($second['retained']);
+ $this->assertNotSame($first['command_id'], $second['command_id']);
+ }
+
+ public function testExplicitNullOptionsClearInheritedKeysWhileOmittedOptionsRetainThem(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $control = app(WorkflowControlPlane::class);
+ $context = CommandContext::controlPlane()->with([
+ 'request' => [
+ 'request_id' => 'inherited-key',
+ ],
+ ]);
+ foreach ([
+ 'signal' => 'changed',
+ 'update' => 'rename',
+ ] as $method => $name) {
+ $options = [
+ 'arguments' => $method === 'signal' ? [1] : ['x'],
+ 'command_context' => $context,
+ ];
+ $first = $control->{$method}($workflow->workflowId(), $name, $options);
+ $retry = $control->{$method}($workflow->workflowId(), $name, $options);
+ $cleared = $control->{$method}($workflow->workflowId(), $name, $options + [
+ 'request_id' => null,
+ ]);
+
+ $this->assertSame('inherited-key', $first['request_id']);
+ $this->assertSame($first['command_id'], $retry['command_id']);
+ $this->assertTrue($retry['retained']);
+ $this->assertNull($cleared['request_id']);
+ $this->assertFalse($cleared['retained']);
+ $this->assertNotSame($first['command_id'], $cleared['command_id']);
+ }
+ }
+
+ public function testEmptyServerBodyKeysCannotFallBackToProxyCorrelation(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ foreach (['signal', 'update'] as $method) {
+ $before = $this->rows();
+ $stub = $workflow->withCommandContext(CommandContext::controlPlane()->with([
+ 'request' => [
+ 'request_id' => 'proxy-correlation',
+ ],
+ 'server' => [
+ 'command' => $method,
+ 'metadata' => [
+ 'request_id' => '',
+ ],
+ ],
+ ]));
+ try {
+ if ($method === 'signal') {
+ $stub->signal('changed', 1);
+ } else {
+ $stub->submitUpdateWithArguments('rename', ['x']);
+ }
+ $this->fail('An explicit empty body key must be refused.');
+ } catch (InvalidCommandRequestIdException) {
+ $this->assertSame($before, $this->rows());
+ }
+ }
+ }
+
+ public function testSignalWithStartRejectsAnInvalidContextKeyBeforeStarting(): void
+ {
+ $stub = WorkflowStub::make(CommandRequestIdentityWorkflow::class, 'invalid-key-start');
+ $before = $this->rows();
+
+ try {
+ $stub->withCommandContext(CommandContext::controlPlane()->with([
+ 'request' => [
+ 'request_id' => "\xFF",
+ ],
+ ]))->signalWithStart('changed', [1]);
+ $this->fail('An invalid key must not start a run.');
+ } catch (InvalidCommandRequestIdException) {
+ $this->assertSame($before, $this->rows());
+ }
+ }
+
+ public function testAnExplicitEmptyKeyIsRefusedBeforeAdmission(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $before = $this->rows();
+
+ try {
+ $workflow->withRequestId('');
+ $this->fail('An explicitly empty key is invalid.');
+ } catch (InvalidCommandRequestIdException) {
+ $this->assertSame($before, $this->rows());
+ }
+ }
+
+ public function testNullClearsTheExplicitKeyAndPreservesIndependentAdmission(): void
+ {
+ $workflow = $this->waitingWorkflow()
+ ->withRequestId('cleared-key')
+ ->withRequestId(null);
+ $first = $workflow->signal('changed', 1);
+ $second = $workflow->signal('changed', 1);
+
+ $this->assertNull($first->requestId());
+ $this->assertNull($second->requestId());
+ $this->assertFalse($second->retained());
+ $this->assertNotSame($first->commandId(), $second->commandId());
+ }
+
+ public function testKeyReuseWithDifferentIntentIsRefusedWithoutConsumingTheKey(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withRequestId('caller-key')
+ ->signal('changed', 1);
+ $before = $this->rows();
+
+ $conflict = $workflow->withRequestId('caller-key')
+ ->attemptSignal('changed', 2);
+
+ $this->assertTrue($conflict->rejected());
+ $this->assertTrue($conflict->requestIdConflict());
+ $this->assertSame('request_id_conflict', $conflict->rejectionReason());
+ $this->assertSame(CommandOutcome::RejectedRequestIdConflict->value, $conflict->outcome());
+ $this->assertSame('caller-key', $conflict->requestId());
+ $this->assertSame([
+ 'retained_command_id' => $first->commandId(),
+ 'retained_command_type' => 'signal',
+ 'retained_outcome' => 'signal_received',
+ 'conflict' => ['arguments'],
+ ], $conflict->conflict());
+ $this->assertSame(1, $this->signalReceivedCount($workflow));
+ $this->assertSame($before, $this->rows());
+ $this->assertSame($first->commandId(), $conflict->commandId());
+ $this->assertNull($conflict->commandSequence());
+ $this->assertSame(SignalStatus::Received, WorkflowSignal::query()->sole()->status);
+ $this->assertSame(1, WorkflowCommand::query()->where('command_type', 'signal')->count());
+
+ $renamed = $workflow->withRequestId('caller-key')
+ ->attemptSignal('other', 1);
+ $this->assertTrue($renamed->requestIdConflict());
+ $this->assertSame(['name'], $renamed->conflict()['conflict']);
+ $this->assertSame($before, $this->rows());
+
+ $retry = $workflow->withRequestId('caller-key')
+ ->signal('changed', 1);
+ $this->assertTrue($retry->retained());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame(1, $this->signalReceivedCount($workflow));
+ }
+
+ public function testSignalAndUpdateKeysAreIndependentScopes(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $signal = $workflow->withRequestId('shared-key')
+ ->signal('changed', 1);
+ $update = $workflow->withRequestId('shared-key')
+ ->submitUpdate('rename', 'x');
+
+ $this->assertTrue($signal->accepted());
+ $this->assertTrue($update->accepted());
+ $this->assertFalse($update->retained());
+ $this->assertNotSame($signal->commandId(), $update->commandId());
+ }
+
+ public function testRetryAfterCompletionReturnsTheRetainedReceipt(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withRequestId('caller-key')
+ ->signal('changed', 1);
+ $workflow->signal('changed', 2);
+ $this->runTask($workflow);
+ $run = WorkflowRun::query()->findOrFail($workflow->runId());
+ $this->assertSame('completed', $run->status->value);
+ $before = $this->rows();
+
+ $retry = $workflow->withRequestId('caller-key')
+ ->signal('changed', 1);
+ $this->assertTrue($retry->accepted());
+ $this->assertTrue($retry->retained());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame($before, $this->rows());
+
+ $late = $workflow->withRequestId('late-key')
+ ->attemptSignal('changed', 3);
+ $this->assertTrue($late->rejected());
+ $this->assertSame('run_not_active', $late->rejectionReason());
+ $lateRows = $this->rows();
+
+ $lateRetry = $workflow->withRequestId('late-key')
+ ->attemptSignal('changed', 3);
+ $this->assertTrue($lateRetry->retained());
+ $this->assertTrue($lateRetry->rejected());
+ $this->assertSame($late->commandId(), $lateRetry->commandId());
+ $this->assertSame($lateRows, $this->rows());
+ }
+
+ public function testRetryAfterContinueAsNewReturnsTheOriginalReceipt(): void
+ {
+ $stub = WorkflowStub::make(TestConfiguredContinueSignalWorkflow::class);
+ $stub->start(0);
+ $first = $stub->withRequestId('caller-key')
+ ->signal('name-provided', 'Taylor 雪');
+ $this->assertTrue($first->accepted());
+ $predecessor = $stub->runId();
+ $bridge = app(DefaultWorkflowTaskBridge::class);
+ $task = WorkflowTask::query()->where('workflow_run_id', $predecessor)->sole();
+ $result = $bridge->execute($task->id);
+ $this->assertTrue($result['executed']);
+ $successor = WorkflowTask::query()->findOrFail($result['next_task_id'])->workflow_run_id;
+ $this->assertNotSame($predecessor, $successor);
+ $before = $this->rows();
+
+ $retry = $stub->withRequestId('caller-key')
+ ->signal('name-provided', 'Taylor 雪');
+ $this->assertTrue($retry->retained());
+ $this->assertTrue($retry->intentVerified());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame($successor, $retry->runId());
+ $this->assertSame($before, $this->rows());
+
+ $conflict = $stub->withRequestId('caller-key')
+ ->attemptSignal('name-provided', 'Someone else');
+ $this->assertTrue($conflict->requestIdConflict());
+ $this->assertSame($first->commandId(), $conflict->conflict()['retained_command_id']);
+
+ $this->assertTrue($bridge->execute($result['next_task_id'])['executed']);
+ $this->assertSame('Taylor 雪', $stub->refresh()->output()['name']);
+ }
+
+ public function testLegacyAcceptedReceiptWithoutDigestIsComparedFromItsArguments(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withRequestId('legacy')
+ ->signal('changed', 7);
+ WorkflowCommand::query()->whereKey($first->commandId())->update([
+ 'intent_digest' => null,
+ ]);
+ $before = $this->rows();
+
+ $same = $workflow->withRequestId('legacy')
+ ->signal('changed', 7);
+ $this->assertTrue($same->retained());
+ $this->assertTrue($same->intentVerified());
+ $this->assertSame($first->commandId(), $same->commandId());
+ $this->assertSame($before, $this->rows());
+
+ $different = $workflow->withRequestId('legacy')
+ ->attemptSignal('changed', 8);
+ $this->assertTrue($different->requestIdConflict());
+ }
+
+ public function testLegacyRejectedReceiptWithoutDigestIsReturnedUnverified(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $rejection = $workflow->withRequestId('legacy')
+ ->attemptSignal('undeclared', 1);
+ $this->assertSame('unknown_signal', $rejection->rejectionReason());
+ WorkflowCommand::query()->whereKey($rejection->commandId())->update([
+ 'intent_digest' => null,
+ ]);
+ $before = $this->rows();
+
+ $retry = $workflow->withRequestId('legacy')
+ ->attemptSignal('changed', 1);
+ $this->assertTrue($retry->retained());
+ $this->assertFalse($retry->intentVerified());
+ $this->assertTrue($retry->rejected());
+ $this->assertSame($rejection->commandId(), $retry->commandId());
+ $this->assertSame($before, $this->rows());
+ }
+
+ public function testSignalWithStartRetryReturnsTheOriginalIntake(): void
+ {
+ $stub = WorkflowStub::make(CommandRequestIdentityWorkflow::class, 'signal-with-start-key');
+ $first = $stub->withRequestId('intake-key')
+ ->signalWithStart('changed', [1]);
+ $this->assertTrue($first->accepted());
+ $this->assertTrue($first->startedNew());
+ $before = $this->rows();
+
+ $retry = $stub->withRequestId('intake-key')
+ ->signalWithStart('changed', [1]);
+ $this->assertTrue($retry->accepted());
+ $this->assertTrue($retry->retained());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame($first->startCommandId(), $retry->startCommandId());
+ $this->assertSame($first->intakeGroupId(), $retry->intakeGroupId());
+ $this->assertTrue($retry->startedNew());
+ $this->assertSame($before, $this->rows());
+ $this->assertSame(1, WorkflowRun::query()->where('workflow_instance_id', $stub->workflowId())->count());
+
+ $conflict = $stub->withRequestId('intake-key')
+ ->attemptSignalWithStart('changed', [2]);
+ $this->assertTrue($conflict->requestIdConflict());
+ $this->assertNull($conflict->startCommandId());
+ $this->assertNull($conflict->intakeGroupId());
+ $this->assertSame($before, $this->rows());
+ $this->assertSame(1, $this->signalReceivedCount($stub));
+ }
+
+ public function testControlPlaneRequestIdOptionKeysSignalsAndUpdates(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $controlPlane = app(WorkflowControlPlane::class);
+
+ $first = $controlPlane->signal($workflow->workflowId(), 'changed', [
+ 'arguments' => [1],
+ 'request_id' => 'plane-key',
+ ]);
+ $rows = $this->rows();
+ $retry = $controlPlane->signal($workflow->workflowId(), 'changed', [
+ 'arguments' => [1],
+ 'request_id' => 'plane-key',
+ ]);
+ $conflict = $controlPlane->signal($workflow->workflowId(), 'changed', [
+ 'arguments' => [2],
+ 'request_id' => 'plane-key',
+ ]);
+
+ $this->assertSame(202, $first['status']);
+ $this->assertSame('plane-key', $first['request_id']);
+ $this->assertFalse($first['retained']);
+ $this->assertTrue($first['intent_verified']);
+ $this->assertSame(202, $retry['status']);
+ $this->assertTrue($retry['retained']);
+ $this->assertSame($first['command_id'], $retry['command_id']);
+ $this->assertSame(409, $conflict['status']);
+ $this->assertSame('request_id_conflict', $conflict['reason']);
+ $this->assertSame($first['command_id'], $conflict['retained_command_id']);
+ $this->assertNull($conflict['command_id']);
+ $this->assertNull($conflict['command_sequence']);
+ $this->assertNull($conflict['workflow_command_id']);
+ $this->assertSame($rows, $this->rows());
+ $this->assertSame(['arguments'], $conflict['conflict']);
+
+ $update = $controlPlane->update($workflow->workflowId(), 'rename', [
+ 'arguments' => ['x'],
+ 'request_id' => 'plane-update',
+ ]);
+ $updateRetry = $controlPlane->update($workflow->workflowId(), 'rename', [
+ 'arguments' => ['x'],
+ 'request_id' => 'plane-update',
+ ]);
+ $updateConflict = $controlPlane->update($workflow->workflowId(), 'rename', [
+ 'arguments' => ['y'],
+ 'request_id' => 'plane-update',
+ ]);
+
+ $this->assertSame(202, $update['status']);
+ $this->assertSame('plane-update', $update['request_id']);
+ $this->assertTrue($updateRetry['retained']);
+ $this->assertSame($update['update_id'], $updateRetry['update_id']);
+ $this->assertSame(409, $updateConflict['status']);
+ $this->assertSame('request_id_conflict', $updateConflict['reason']);
+ $this->assertSame($update['command_id'], $updateConflict['retained_command_id']);
+ $this->assertSame(1, WorkflowUpdate::query()->where('status', 'accepted')->count());
+ }
+
+ public function testServerContextsUseTheBodyKeyAndIgnoreTheCorrelationHeader(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withCommandContext($this->serverContext('body-key', 'proxy-a'))
+ ->signal('changed', 1);
+ $retry = $workflow->withCommandContext($this->serverContext('body-key', 'proxy-b'))
+ ->signal('changed', 1);
+ $headerOnly = $workflow->withCommandContext($this->serverContext(null, 'proxy-a'))
+ ->signal('changed', 1);
+
+ $this->assertTrue($retry->retained());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame('body-key', $retry->requestId());
+ $this->assertFalse($headerOnly->retained());
+ $this->assertNull($headerOnly->requestId());
+ $this->assertSame(2, $this->signalReceivedCount($workflow));
+ }
+
+ #[DataProvider('runTargetCommands')]
+ public function testControlPlaneRefusesASelectedPredecessorInsteadOfDeliveringToItsSuccessor(
+ string $operation,
+ string $name,
+ array $arguments,
+ ): void {
+ $stub = WorkflowStub::make(TestConfiguredContinueSignalWorkflow::class);
+ $stub->start(0);
+ $predecessor = $stub->runId();
+ $this->runTask($stub);
+ $stub->refresh();
+ $successor = $stub->runId();
+ $this->assertNotSame($predecessor, $successor);
+ $before = WorkflowHistoryEvent::query()->where('workflow_run_id', $successor)->orderBy('id')->get()->toArray();
+
+ $result = app(WorkflowControlPlane::class)->{$operation}($stub->id(), $name, [
+ 'run_id' => $predecessor,
+ 'request_id' => 'selected-original',
+ 'arguments' => $arguments,
+ ]);
+
+ $this->assertSame(409, $result['status']);
+ $this->assertSame('selected_run_not_current', $result['reason']);
+ $this->assertSame('run', $result['target_scope']);
+ $this->assertSame($predecessor, $result['run_id']);
+ $this->assertSame(
+ $before,
+ WorkflowHistoryEvent::query()->where('workflow_run_id', $successor)->orderBy('id')->get()->toArray(),
+ );
+ $rows = $this->rows();
+ $retry = app(WorkflowControlPlane::class)->{$operation}($stub->id(), $name, [
+ 'run_id' => $predecessor,
+ 'request_id' => 'selected-original',
+ 'arguments' => $arguments,
+ ]);
+ $this->assertTrue($retry['retained']);
+ $this->assertSame($result['command_id'], $retry['command_id']);
+ $this->assertSame($rows, $this->rows());
+ }
+
+ #[DataProvider('runTargetCommands')]
+ public function testControlPlaneIncludesTheSelectedRunInIntent(
+ string $operation,
+ string $name,
+ array $arguments,
+ ): void {
+ $stub = WorkflowStub::make(TestConfiguredContinueSignalWorkflow::class);
+ $stub->start(0);
+ $predecessor = $stub->runId();
+ $this->runTask($stub);
+ $stub->refresh();
+ $controlPlane = app(WorkflowControlPlane::class);
+ $options = [
+ 'run_id' => $stub->runId(),
+ 'request_id' => 'selected-current',
+ 'arguments' => $arguments,
+ ];
+ $first = $controlPlane->{$operation}($stub->id(), $name, $options);
+ $this->assertSame(202, $first['status']);
+ $this->assertSame('run', $first['target_scope']);
+ $this->assertSame($stub->runId(), $first['run_id']);
+ $rows = $this->rows();
+ $retry = $controlPlane->{$operation}($stub->id(), $name, $options);
+ $this->assertTrue($retry['retained']);
+ $this->assertSame($first['command_id'], $retry['command_id']);
+ $this->assertSame($rows, $this->rows());
+
+ $options['run_id'] = $predecessor;
+ $conflict = $controlPlane->{$operation}($stub->id(), $name, $options);
+ $this->assertSame(409, $conflict['status']);
+ $this->assertSame('request_id_conflict', $conflict['reason']);
+ $this->assertSame(['target'], $conflict['conflict']);
+ $this->assertSame($first['command_id'], $conflict['retained_command_id']);
+ $this->assertSame($rows, $this->rows());
+ }
+
+ #[DataProvider('runTargetCommands')]
+ public function testControlPlaneMissingAndForeignRunTargetsDoNotMutateEitherWorkflow(
+ string $operation,
+ string $name,
+ array $arguments,
+ ): void {
+ $stub = WorkflowStub::make(TestConfiguredContinueSignalWorkflow::class, null, 'tenant-a');
+ $stub->start(1);
+ $foreign = WorkflowStub::make(TestConfiguredContinueSignalWorkflow::class, null, 'tenant-b');
+ $foreign->start(1);
+ $rows = $this->rows();
+
+ foreach (['missing-run', $foreign->runId()] as $runId) {
+ $result = app(WorkflowControlPlane::class)->{$operation}($stub->id(), $name, [
+ 'namespace' => 'tenant-a',
+ 'run_id' => $runId,
+ 'request_id' => 'foreign-key',
+ 'arguments' => $arguments,
+ ]);
+ $this->assertSame(404, $result['status']);
+ $this->assertSame('instance_not_found', $result['reason']);
+ $this->assertSame($rows, $this->rows());
+ }
+ }
+
+ /**
+ * @return iterable
+ */
+ public static function runTargetCommands(): iterable
+ {
+ yield 'signal' => ['signal', 'name-provided', ['Taylor 雪']];
+ yield 'update' => ['update', 'mark-approved', [true]];
+ }
+
+ public function testWebhookContextsKeepTheRequestHeaderAsTheEmbeddedKey(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $request = Request::create('/hooks/changed', 'POST', [
+ 'revision' => 1,
+ ], [], [], [
+ 'HTTP_X_REQUEST_ID' => 'delivery-1',
+ ]);
+ $context = CommandContext::webhook($request, 'none');
+
+ $first = $workflow->withCommandContext($context)
+ ->signal('changed', 1);
+ $retry = $workflow->withCommandContext($context)
+ ->signal('changed', 1);
+
+ $this->assertSame('delivery-1', $first->requestId());
+ $this->assertTrue($retry->retained());
+ $this->assertSame($first->commandId(), $retry->commandId());
+ $this->assertSame(1, $this->signalReceivedCount($workflow));
+ }
+
+ /**
+ * @return list
+ */
+ private static function keyRoutes(): array
+ {
+ return ['stub-signal', 'stub-update', 'plane-signal', 'plane-update', 'context-signal', 'context-update'];
+ }
+
+ /**
+ * @return array
+ */
+ private function keyedAdmission(WorkflowStub $workflow, string $route, mixed $key): array
+ {
+ if (str_starts_with($route, 'plane-')) {
+ $method = $route === 'plane-signal' ? 'signal' : 'update';
+
+ return app(WorkflowControlPlane::class)->{$method}(
+ $workflow->workflowId(),
+ $method === 'signal' ? 'changed' : 'rename',
+ [
+ 'arguments' => $method === 'signal' ? [1] : ['x'],
+ 'request_id' => $key,
+ ],
+ );
+ }
+
+ $stub = str_starts_with($route, 'context-')
+ ? $workflow->withCommandContext(CommandContext::controlPlane()->with([
+ 'request' => [
+ 'request_id' => $key,
+ ],
+ ]))
+ : $workflow->withRequestId($key);
+ $result = str_ends_with($route, '-signal')
+ ? $stub->signal('changed', 1)
+ : $stub->submitUpdateWithArguments('rename', ['x']);
+
+ return [
+ 'request_id' => $result->requestId(),
+ 'command_id' => $result->commandId(),
+ 'retained' => $result->retained(),
+ ];
+ }
+
+ private function waitingWorkflow(): WorkflowStub
+ {
+ $workflow = WorkflowStub::make(CommandRequestIdentityWorkflow::class);
+ $workflow->start();
+ $this->runTask($workflow);
+
+ return $workflow;
+ }
+
+ private function runTask(WorkflowStub $workflow): void
+ {
+ $task = WorkflowTask::query()
+ ->where('workflow_run_id', $workflow->runId())
+ ->where('status', 'ready')
+ ->orderBy('available_at')
+ ->first();
+ $this->assertInstanceOf(WorkflowTask::class, $task);
+ $this->assertTrue(app(DefaultWorkflowTaskBridge::class)->execute($task->id)['executed']);
+ }
+
+ private function signalReceivedCount(WorkflowStub $workflow): int
+ {
+ $runIds = WorkflowRun::query()->where('workflow_instance_id', $workflow->workflowId())->pluck('id');
+
+ return WorkflowHistoryEvent::query()
+ ->whereIn('workflow_run_id', $runIds)
+ ->where('event_type', HistoryEventType::SignalReceived->value)
+ ->count();
+ }
+
+ private function serverContext(?string $key, string $header): CommandContext
+ {
+ return CommandContext::controlPlane()->with([
+ 'request' => [
+ 'request_id' => $header,
+ 'headers' => [
+ 'x_request_id' => $header,
+ ],
+ ],
+ 'server' => [
+ 'command' => 'signal',
+ 'metadata' => array_filter([
+ 'request_id' => $key,
+ ], static fn (mixed $value): bool => $value !== null),
+ ],
+ ]);
+ }
+
+ private function rows(): array
+ {
+ $rows = [];
+ foreach ([
+ WorkflowInstance::class,
+ WorkflowRun::class,
+ WorkflowCommand::class,
+ WorkflowSignal::class,
+ WorkflowUpdate::class,
+ WorkflowTask::class,
+ WorkflowHistoryEvent::class,
+ ] as $model) {
+ $rows[$model] = $model::query()->orderBy('id')->get()->map(
+ static fn ($row): array => $row->getRawOriginal()
+ )->all();
+ }
+
+ return $rows;
+ }
+}
+
+#[Signal('changed', [
+ [
+ 'name' => 'revision',
+ 'type' => 'int',
+ ],
+])]
+#[Signal('other', [
+ [
+ 'name' => 'revision',
+ 'type' => 'int',
+ ],
+])]
+final class CommandRequestIdentityWorkflow extends Workflow
+{
+ private string $name = '';
+
+ public function handle(): array
+ {
+ $first = signal('changed');
+ $second = signal('changed');
+
+ return [
+ 'revisions' => [$first, $second],
+ 'name' => $this->name,
+ ];
+ }
+
+ #[\Workflow\UpdateMethod]
+ public function rename(string $value): string
+ {
+ $this->name = $value;
+
+ return $value;
+ }
+}
diff --git a/tests/Unit/V2/ServiceCallEffectIdentityTest.php b/tests/Unit/V2/ServiceCallEffectIdentityTest.php
index eefb84fb3..3b56dee80 100644
--- a/tests/Unit/V2/ServiceCallEffectIdentityTest.php
+++ b/tests/Unit/V2/ServiceCallEffectIdentityTest.php
@@ -158,13 +158,22 @@ public function testRawDirectOptionsCannotClaimReservedStartUpdateOrSignalIdenti
'arguments' => ['second'],
]));
$this->assertNotSame($firstUpdate['update_id'], $secondUpdate['update_id']);
- $this->assertSame(2, WorkflowUpdate::query()->count());
+ $this->assertSame(409, $secondUpdate['status']);
+ $this->assertSame('request_id_conflict', $secondUpdate['reason']);
+ $this->assertSame(1, WorkflowUpdate::query()->count());
+ $updateCommand = WorkflowCommand::query()->findOrFail($firstUpdate['command_id']);
+ $this->assertStringStartsWith('update-v1:', $updateCommand->request_id);
+ $this->assertNotSame($rawOptions['request_id'], $updateCommand->request_id);
$signalWorkflow = $this->waitingWorkflow('raw-effect-signal');
$firstSignal = $control->signal($signalWorkflow->id(), 'finish', $rawOptions);
$secondSignal = $control->signal($signalWorkflow->id(), 'finish', $rawOptions);
- $this->assertNotSame($firstSignal['workflow_command_id'], $secondSignal['workflow_command_id']);
- $this->assertSame(2, WorkflowSignal::query()->count());
+ $this->assertSame($firstSignal['workflow_command_id'], $secondSignal['workflow_command_id']);
+ $this->assertTrue($secondSignal['retained']);
+ $this->assertSame(1, WorkflowSignal::query()->count());
+ $signalCommand = WorkflowCommand::query()->findOrFail($firstSignal['command_id']);
+ $this->assertStringStartsWith('signal-v1:', $signalCommand->request_id);
+ $this->assertNotSame($rawOptions['request_id'], $signalCommand->request_id);
}
public function testCraftedCommandContextCannotClaimReservedSignalIdentity(): void
@@ -191,8 +200,11 @@ public function testCraftedCommandContextCannotClaimReservedSignalIdentity(): vo
->where('command_type', CommandType::Signal->value)
->get();
$this->assertCount(2, $commands);
- $this->assertTrue($commands->every(static fn (WorkflowCommand $command): bool =>
- $command->request_id === null));
+ foreach ($commands as $command) {
+ $this->assertStringStartsWith('signal-v1:', $command->request_id);
+ $this->assertNull($command->requestId());
+ }
+ $this->assertCount(2, $commands->pluck('request_id')->unique());
}
public function testPredictableRawIdentityCannotPreseedTheScheduledUpdateReceipt(): void
@@ -278,7 +290,8 @@ public function testLowLevelCommandRequestIdCannotReserveTheOldCanonicalIdentity
$this->assertNull($start->request_id);
$this->assertStringStartsWith('update-v1:', $update->request_id);
$this->assertNotSame($guessedReservedToken, $update->request_id);
- $this->assertNull($signal->request_id);
+ $this->assertStringStartsWith('signal-v1:', $signal->request_id);
+ $this->assertNotSame($guessedReservedToken, $signal->request_id);
}
public function testLowLevelCommandRequestIdCannotReserveAnActualOpaqueEffectToken(): void
@@ -319,7 +332,8 @@ public function testLowLevelCommandRequestIdCannotReserveAnActualOpaqueEffectTok
$this->assertNull($start->request_id);
$this->assertStringStartsWith('update-v1:', $update->request_id);
$this->assertNotSame($effectToken, $update->request_id);
- $this->assertNull($signal->request_id);
+ $this->assertStringStartsWith('signal-v1:', $signal->request_id);
+ $this->assertNotSame($effectToken, $signal->request_id);
$legitimate = (new DefaultWorkflowControlPlane())->update($workflow->id(), 'change', [
'arguments' => ['legitimate'],
diff --git a/tests/Unit/V2/UpdateRequestIdentityTest.php b/tests/Unit/V2/UpdateRequestIdentityTest.php
index cdf643974..d0264d899 100644
--- a/tests/Unit/V2/UpdateRequestIdentityTest.php
+++ b/tests/Unit/V2/UpdateRequestIdentityTest.php
@@ -9,12 +9,14 @@
use Tests\TestCase;
use Workflow\UpdateMethod;
use Workflow\V2\CommandContext;
+use Workflow\V2\Enums\CommandOutcome;
use Workflow\V2\Models\WorkflowCommand;
use Workflow\V2\Models\WorkflowHistoryEvent;
use Workflow\V2\Models\WorkflowRun;
use Workflow\V2\Models\WorkflowTask;
use Workflow\V2\Models\WorkflowUpdate;
use function Workflow\V2\signal;
+use Workflow\V2\Support\CommandIntent;
use Workflow\V2\Support\DefaultWorkflowTaskBridge;
use Workflow\V2\Workflow;
use Workflow\V2\WorkflowStub;
@@ -44,23 +46,85 @@ public function testCallerKeyControlsTheReceiptIndependentlyOfCorrelation(
->submitUpdate('change', 'original');
$before = $this->rows();
$second = $workflow->withCommandContext($this->context($secondKey, $secondHeader))
- ->submitUpdate('change', 'changed');
+ ->submitUpdate('change', 'original');
$this->assertTrue($first->accepted());
$this->assertTrue($second->accepted());
+ $this->assertFalse($first->retained());
+ $this->assertSame($firstKey, $first->requestId());
if ($sameReceipt) {
$this->assertSame($first->updateId(), $second->updateId());
+ $this->assertSame($first->commandId(), $second->commandId());
+ $this->assertTrue($second->retained());
+ $this->assertTrue($second->intentVerified());
$this->assertSame($before, $this->rows());
} else {
$this->assertNotSame($first->updateId(), $second->updateId());
+ $this->assertFalse($second->retained());
$this->assertSame(2, WorkflowUpdate::query()->count());
}
$command = WorkflowCommand::query()->findOrFail($first->commandId());
$this->assertSame($firstKey, $command->requestId());
$this->assertLessThanOrEqual(191, strlen($command->request_id));
+ $this->assertTrue(CommandIntent::isDigest($command->intentDigest()));
$this->assertSame($firstHeader, $command->context['request']['headers']['x_request_id']);
}
+ public function testReusingAKeyWithDifferentArgumentsIsRefusedWithoutConsumingTheKey(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withCommandContext($this->context('caller', 'proxy-a'))
+ ->submitUpdate('change', 'original');
+ $before = $this->rows();
+ $accepted = WorkflowHistoryEvent::query()->where('event_type', 'UpdateAccepted')->count();
+
+ $conflict = $workflow->withCommandContext($this->context('caller', 'proxy-b'))
+ ->submitUpdate('change', 'changed');
+
+ $this->assertTrue($conflict->rejected());
+ $this->assertFalse($conflict->accepted());
+ $this->assertTrue($conflict->requestIdConflict());
+ $this->assertSame('request_id_conflict', $conflict->rejectionReason());
+ $this->assertSame(CommandOutcome::RejectedRequestIdConflict->value, $conflict->outcome());
+ $this->assertSame('caller', $conflict->requestId());
+ $this->assertSame($before, $this->rows());
+ $this->assertSame($first->commandId(), $conflict->commandId());
+ $this->assertNull($conflict->commandSequence());
+ $this->assertSame([
+ 'retained_command_id' => $first->commandId(),
+ 'retained_command_type' => 'update',
+ 'retained_outcome' => null,
+ 'conflict' => ['arguments'],
+ ], $conflict->conflict());
+ $this->assertSame($accepted, WorkflowHistoryEvent::query()->where('event_type', 'UpdateAccepted')->count());
+ $this->assertNull($conflict->updateId());
+
+ $retry = $workflow->withCommandContext($this->context('caller', 'proxy-c'))
+ ->submitUpdate('change', 'original');
+ $this->assertTrue($retry->retained());
+ $this->assertSame($first->updateId(), $retry->updateId());
+
+ $renamed = $workflow->withCommandContext($this->context('caller', 'proxy-d'))
+ ->submitUpdate('rename', 'original');
+ $this->assertTrue($renamed->requestIdConflict());
+ $this->assertSame(['name'], $renamed->conflict()['conflict']);
+ $this->assertSame($before, $this->rows());
+ }
+
+ public function testTypedArgumentsAndNamedFormsCompareByValue(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withCommandContext($this->context('typed', 'a'))
+ ->submitUpdate('change', 'same');
+ $named = $workflow->withCommandContext($this->context('typed', 'b'))
+ ->submitUpdateWithArguments('change', [
+ 'value' => 'same',
+ ]);
+
+ $this->assertTrue($named->retained());
+ $this->assertSame($first->updateId(), $named->updateId());
+ }
+
public function testLegacyProxyColumnsNeitherHideReceiptsNorReserveAnotherCallersKey(): void
{
$workflow = $this->waitingWorkflow();
@@ -73,8 +137,9 @@ public function testLegacyProxyColumnsNeitherHideReceiptsNorReserveAnotherCaller
$before = $this->rows();
$duplicate = $workflow->withCommandContext($this->context('original-caller', 'other-proxy'))
- ->submitUpdate('change', 'ignored');
+ ->submitUpdate('change', 'original');
$this->assertSame($first->updateId(), $duplicate->updateId());
+ $this->assertTrue($duplicate->retained());
$this->assertSame($before, $this->rows());
$next = $workflow->withCommandContext($this->context('next-caller', 'proxy'))
->submitUpdate('change', 'next');
@@ -103,7 +168,7 @@ public function testAnOccupiedLookupTokenDoesNotAliasAnotherLogicalKey(): void
$this->assertSame($original, $command->fresh()->getRawOriginal());
$before = $this->rows();
$repeat = $workflow->withCommandContext($this->context('new-caller', 'other-proxy'))
- ->submitUpdate('change', 'ignored');
+ ->submitUpdate('change', 'new');
$this->assertSame($new->updateId(), $repeat->updateId());
$this->assertSame($before, $this->rows());
}
@@ -127,11 +192,55 @@ public function testLegacyEmbeddedColumnKeyRemainsReadableWithoutContextRepair()
$this->assertSame('embedded-key', $command->requestId());
$duplicate = $workflow->withCommandContext($context)
- ->submitUpdate('change', 'ignored');
+ ->submitUpdate('change', 'first');
$this->assertSame($first->updateId(), $duplicate->updateId());
$this->assertSame($before, $this->rows());
}
+ public function testLegacyAcceptedReceiptWithoutDigestIsComparedFromItsRetainedArguments(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $first = $workflow->withCommandContext($this->context('legacy', 'a'))
+ ->submitUpdate('change', 'first');
+ WorkflowCommand::query()->whereKey($first->commandId())->update([
+ 'intent_digest' => null,
+ ]);
+ $before = $this->rows();
+
+ $same = $workflow->withCommandContext($this->context('legacy', 'b'))
+ ->submitUpdate('change', 'first');
+ $this->assertTrue($same->retained());
+ $this->assertTrue($same->intentVerified());
+ $this->assertSame($first->updateId(), $same->updateId());
+ $this->assertSame($before, $this->rows());
+
+ $different = $workflow->withCommandContext($this->context('legacy', 'c'))
+ ->submitUpdate('change', 'second');
+ $this->assertTrue($different->requestIdConflict());
+ $this->assertSame($first->commandId(), $different->conflict()['retained_command_id']);
+ }
+
+ public function testLegacyRejectedReceiptWithoutDigestIsReturnedUnverified(): void
+ {
+ $workflow = $this->waitingWorkflow();
+ $rejection = $workflow->withCommandContext($this->context('legacy-rejected', 'a'))
+ ->submitUpdate('missing', 'first');
+ $this->assertTrue($rejection->rejected());
+ $this->assertSame('unknown_update', $rejection->rejectionReason());
+ WorkflowCommand::query()->whereKey($rejection->commandId())->update([
+ 'intent_digest' => null,
+ ]);
+ $before = $this->rows();
+
+ $retry = $workflow->withCommandContext($this->context('legacy-rejected', 'b'))
+ ->submitUpdate('change', 'anything');
+ $this->assertTrue($retry->retained());
+ $this->assertFalse($retry->intentVerified());
+ $this->assertTrue($retry->rejected());
+ $this->assertSame($rejection->commandId(), $retry->commandId());
+ $this->assertSame($before, $this->rows());
+ }
+
public function testTheEarliestRetainedReceiptWinsAfterLegacyDuplicateAdmission(): void
{
$workflow = $this->waitingWorkflow();
@@ -149,7 +258,7 @@ public function testTheEarliestRetainedReceiptWinsAfterLegacyDuplicateAdmission(
$before = $this->rows();
$repeat = $workflow->withCommandContext($this->context('caller', 'c'))
- ->submitUpdate('change', 'ignored');
+ ->submitUpdate('change', 'first');
$this->assertSame($first->updateId(), $repeat->updateId());
$this->assertSame($before, $this->rows());
}
@@ -190,9 +299,16 @@ public function testIntakeLimitsPrecedeTheMissingDefinitionExecutionGate(string
$this->assertEquals($deadline, $run->fresh()->run_deadline_at);
$before = $this->rows();
$duplicate = $workflow->withCommandContext($this->context('over-limit', 'different-proxy'))
- ->submitUpdate('change', 'changed');
+ ->submitUpdate('change', 'over-limit');
$this->assertSame($rejection->commandId(), $duplicate->commandId());
+ $this->assertTrue($duplicate->retained());
+ $this->assertTrue($duplicate->intentVerified());
$this->assertSame($before, $this->rows());
+
+ $conflict = $workflow->withCommandContext($this->context('over-limit', 'another-proxy'))
+ ->submitUpdate('change', 'changed');
+ $this->assertTrue($conflict->requestIdConflict());
+ $this->assertSame($rejection->commandId(), $conflict->conflict()['retained_command_id']);
}
private function waitingWorkflow(): WorkflowStub
@@ -254,4 +370,10 @@ public function change(string $value): string
{
return $value;
}
+
+ #[UpdateMethod]
+ public function rename(string $value): string
+ {
+ return strtoupper($value);
+ }
}
diff --git a/tests/Unit/migrations/MigrationsTest.php b/tests/Unit/migrations/MigrationsTest.php
index 0f137e345..4127b503b 100644
--- a/tests/Unit/migrations/MigrationsTest.php
+++ b/tests/Unit/migrations/MigrationsTest.php
@@ -218,6 +218,7 @@ public function testDownMethodsDropTables(): void
$this->assertTrue(Schema::hasColumn('workflow_runs', 'output_payload_codec'));
$this->assertTrue(Schema::hasTable('workflow_commands'));
$this->assertTrue(Schema::hasColumn('workflow_commands', 'request_id'));
+ $this->assertTrue(Schema::hasColumn('workflow_commands', 'intent_digest'));
$this->assertTrue(Schema::hasTable('workflow_links'));
$this->assertTrue(Schema::hasTable('activity_attempts'));
$this->assertTrue(Schema::hasColumn('activity_attempts', 'worker_attempt_id'));