Bind explicitly typed search attributes to replay identity

CHANGELOG.md

@@ -8,6 +8,12 @@ project follows [Semantic Versioning](https://semver.org/).
 
 ### Added
 
+- `WorkflowContext::upsertSearchAttributes()` accepts an optional complete map
+  of canonical type declarations (#244). Explicitly typed calls bind replay to
+  both values and recorded type identity. Existing one-argument calls retain
+  value-only replay, and histories without type metadata retain legacy
+  compatibility. Malformed recorded type identities are refused.
+
 - `WorkflowContext::startChildWorkflow()` starts a child and returns a
   `ChildWorkflowHandle` once the runtime has committed the start (#185). The
   handle's `identity()` gives the child call ID, the child workflow ID and the

docs/portal/build/workflows-activities.md

@@ -605,6 +605,31 @@ The context provides these replay-aware commands:
 
 Do not use PHP's process sleep, random-number functions, current wall-clock time, or mutable global state directly to make workflow decisions. `sideEffect()` is for small values whose result belongs in history, not for network or database work.
 
+## Declare search-attribute types
+
+Pass a type map when the workflow must preserve search-attribute type identity
+across replay and redrive:
+
+```php
+$context->upsertSearchAttributes(
+    ['Stage' => 'approved', 'Attempts' => 2],
+    ['Stage' => 'keyword', 'Attempts' => 'int'],
+);
+```
+
+Declare every non-deleted attribute. Canonical types are `string`, `keyword`,
+`keyword_list`, `int`, `float`, `bool`, and `datetime`. Use `null` to delete a
+key. The runtime also checks declarations against the namespace's registered
+types. Search metadata uses the existing `typed_search_attributes` worker
+capability and protocol contract.
+
+An explicitly typed call refuses a changed value or a changed recorded type.
+Malformed recorded type maps also refuse replay. Histories written before type
+metadata existed still compare values, since their type identity is unknown.
+Existing calls that omit the type map keep their value-only behavior. That
+legacy path does not establish typed replay equivalence; signal-wait redrive
+qualification is also separate.
+
 ## Bound long histories
 
 Use `continueAsNew()` for recurring or event-heavy workflows. Pass only the state needed by the next run:

regression-corpus-policy.json

@@ -44,6 +44,9 @@
         {
           "glob": "src/Worker/ConditionPredicateSource.php"
         },
+        {
+          "glob": "src/Worker/SearchAttributeTypes.php"
+        },
         {
           "glob": "src/Worker/WorkflowEntryArguments.php"
         },

scripts/ci/run-replay-regression-fixture.php

@@ -943,6 +943,12 @@ private static function workflow(string $workflowType): callable
 
                 return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]);
             },
+            'golden.redrive-reused-search-attribute-types' => static function (WorkflowContext $context, mixed $name): mixed {
+                $context->upsertMemo(['stage' => 'greeting', 'customer' => $name]);
+                $context->upsertSearchAttributes(['Stage' => 'greeting'], ['Stage' => 'keyword']);
+
+                return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]);
+            },
             'golden.redrive-reused-search-attributes-missing' => static function (WorkflowContext $context, mixed $name): mixed {
                 $context->upsertMemo(['stage' => 'greeting', 'customer' => $name]);
                 $context->upsertSearchAttributes(['Stage' => 'greeting']);

src/Worker/Replayer.php

@@ -1857,6 +1857,9 @@ private function recordedSteps(array $history, bool $preserveConditionReopens, C
                         resolutionOrder: $resolutionOrder,
                     ),
                     'search_attributes' => $this->canonicalSearchAttributes($attributes, $sequence, 'recorded'),
+                    'search_attribute_types' => array_key_exists('attribute_types', $payload)
+                        ? $this->recordedSearchAttributeTypes($attributes, $payload['attribute_types'], $sequence)
+                        : null,
                 ];
             } elseif ($type === 'ConditionWaitOpened') {
                 $conditionKey = $this->stringValue($payload['condition_key'] ?? null);
@@ -2699,6 +2702,14 @@ private function assertCommandMatchesStep(WorkflowCommand $command, array $step)
                     'search_attribute_value_mismatch',
                 );
             }
+            $recordedTypes = $step['search_attribute_types'] ?? null;
+            if ($recordedTypes !== null && array_key_exists('attribute_types', $command->attributes)
+                && $recordedTypes !== $command->attributes['attribute_types']) {
+                throw new NonDeterministicWorkflow(
+                    'Declared search-attribute types changed during replay.',
+                    $step['sequence'], reason: 'search_attribute_type_mismatch',
+                );
+            }
         }
 
         $actualDetail = $this->commandDetail($command);
@@ -2987,6 +2998,21 @@ private function commandDetail(WorkflowCommand $command): ?string
 
     private const SEARCH_ATTRIBUTE_MAX_KEYWORD_LENGTH = 255;
 
+    /**
+     * @param array<string, mixed> $attributes
+     * @return array<string, string>
+     */
+    private function recordedSearchAttributeTypes(array $attributes, mixed $types, int $sequence): array
+    {
+        try {
+            return SearchAttributeTypes::canonical($attributes, $types);
+        } catch (\InvalidArgumentException $error) {
+            throw new NonDeterministicWorkflow(
+                $error->getMessage(), $sequence, reason: 'search_attribute_types_malformed',
+            );
+        }
+    }
+
     /**
      * The canonical replay identity of a search-attribute upsert's values.
      *
@@ -2998,10 +3024,10 @@ private function commandDetail(WorkflowCommand $command): ?string
      * sorted. The authored call is normalized the same way. Anything outside
      * that set, recorded or authored, refuses replay instead of being coerced.
      *
-     * Recorded attribute_types are not part of this identity: this SDK's
-     * upsert command carries no declared types, and the effective type the
-     * Server records may come from a namespace registration the worker cannot
-     * see. This is value-only binding.
+     * Explicitly authored types bind separately against known recorded types.
+     * Legacy one-argument calls remain value-only because effective types may
+     * come from a namespace registration the worker cannot see. Histories
+     * without type metadata also retain value-only compatibility.
      *
      * @return array<string, null|bool|int|float|string|list<string>>
      */

src/Worker/SearchAttributeTypes.php

@@ -0,0 +1,120 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Worker;
+
+use InvalidArgumentException;
+
+/** @internal Canonical, explicitly authored search-attribute type identity. */
+final class SearchAttributeTypes
+{
+    private const TYPES = ['string', 'keyword', 'keyword_list', 'int', 'float', 'bool', 'datetime'];
+
+    /**
+     * Null values are deletions and need no declaration. A declaration for a
+     * deletion is retained because the Server records explicitly supplied types.
+     *
+     * @param array<array-key, mixed> $attributes
+     * @return array<string, string>
+     */
+    public static function canonical(array $attributes, mixed $types): array
+    {
+        if ($attributes === [] || count($attributes) > 100) {
+            throw new InvalidArgumentException('A typed search-attribute update requires between one and 100 attributes.');
+        }
+        foreach ($attributes as $key => $_) {
+            if (!is_string($key) || preg_match('/\A[A-Za-z0-9_.:-]{1,64}\z/', $key) !== 1) {
+                throw new InvalidArgumentException('Search attributes require portable string keys.');
+            }
+        }
+        $canonical = self::typeMap($attributes, $types);
+        foreach ($attributes as $key => $value) {
+            if ($value !== null && !array_key_exists($key, $canonical)) {
+                throw new InvalidArgumentException("Search attribute [{$key}] needs an explicit type; use null to delete it.");
+            }
+        }
+        ksort($canonical, SORT_STRING);
+
+        return $canonical;
+    }
+
+    /**
+     * @param array<array-key, mixed> $attributes
+     * @return array<string, string>
+     */
+    private static function typeMap(array $attributes, mixed $types): array
+    {
+        if (!is_array($types)) {
+            throw new InvalidArgumentException('Search-attribute types must be a canonical type map.');
+        }
+        $canonical = [];
+        foreach ($types as $key => $type) {
+            if (!is_string($key) || !array_key_exists($key, $attributes)) {
+                throw new InvalidArgumentException('A declared search-attribute type has no corresponding attribute.');
+            }
+            if (!is_string($type) || !in_array($type, self::TYPES, true)) {
+                throw new InvalidArgumentException("Search attribute [{$key}] must declare a canonical type name.");
+            }
+            if (!self::accepts($type, $attributes[$key])) {
+                throw new InvalidArgumentException("Search attribute [{$key}] has a value incompatible with its declared type.");
+            }
+            $canonical[$key] = $type;
+        }
+        return $canonical;
+    }
+
+    private static function accepts(string $type, mixed $value): bool
+    {
+        if ($value === null) {
+            return true;
+        }
+
+        return match ($type) {
+            'string' => self::stringValue($value, 2048),
+            'keyword' => self::stringValue($value, 255),
+            'keyword_list' => self::keywordList($value),
+            'int' => is_int($value),
+            'float' => is_int($value) || (is_float($value) && is_finite($value)),
+            'bool' => is_bool($value),
+            'datetime' => self::dateTime($value),
+            default => false,
+        };
+    }
+
+    private static function stringValue(mixed $value, int $limit, bool $allowBlank = false): bool
+    {
+        if (!is_string($value) || (!$allowBlank && trim($value) === '')) {
+            return false;
+        }
+        $length = preg_match_all('/./su', trim($value));
+
+        return $length !== false && $length <= $limit;
+    }
+
+    private static function keywordList(mixed $value): bool
+    {
+        if (!is_array($value) || !array_is_list($value)) {
+            return false;
+        }
+        foreach ($value as $item) {
+            // Core allows blank list items, unlike standalone string attributes.
+            if (!self::stringValue($item, 255, allowBlank: true)) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    private static function dateTime(mixed $value): bool
+    {
+        if (!is_string($value) || !self::stringValue($value, 2048)
+            || preg_match('/\A\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)\z/i', trim($value)) !== 1) {
+            return false;
+        }
+        $parsed = date_parse(trim($value));
+
+        return $parsed['warning_count'] === 0 && $parsed['error_count'] === 0;
+    }
+}

src/Worker/WorkflowCommand.php

@@ -786,10 +786,18 @@ public static function continueAsNew(array $arguments, ?string $workflowType = n
         ]);
     }
 
-    /** @param array<string, mixed> $attributes */
-    public static function upsertSearchAttributes(array $attributes): self
+    /**
+     * @param array<string, mixed> $attributes
+     * @param array<string, string>|null $attributeTypes
+     */
+    public static function upsertSearchAttributes(array $attributes, ?array $attributeTypes = null): self
     {
-        return new self('upsert_search_attributes', 'search_attributes', ['attributes' => $attributes]);
+        $identity = ['attributes' => $attributes];
+        if ($attributeTypes !== null) {
+            $identity['attribute_types'] = SearchAttributeTypes::canonical($attributes, $attributeTypes);
+        }
+
+        return new self('upsert_search_attributes', 'search_attributes', $identity);
     }
 
     /** @param array<string, mixed> $entries */
@@ -890,6 +898,11 @@ public function toWire(PayloadCodec $codec, string $defaultTaskQueue): array
     {
         $wire = ['type' => $this->type];
         foreach ($this->attributes as $key => $value) {
+            if ($key === 'attribute_types' && $value === []) {
+                // An all-deletion update has no wire type declarations, but its
+                // empty authored identity still constrains recorded type maps.
+                continue;
+            }
             if ($key === 'request_payload_value') {
                 $wire['request_payload'] = $codec->envelope($value);
                 continue;

src/Worker/WorkflowContext.php

@@ -671,10 +671,16 @@ public function continueAsNew(
         throw new LogicException('A continue-as-new command cannot resume the current workflow execution.');
     }
 
-    /** @param array<string, mixed> $attributes */
-    public function upsertSearchAttributes(array $attributes): void
+    /**
+     * Explicit types bind replay to the complete type identity of non-deleted
+     * attributes. Omit the map to retain the existing value-only API.
+     *
+     * @param array<string, mixed> $attributes
+     * @param array<string, string>|null $attributeTypes
+     */
+    public function upsertSearchAttributes(array $attributes, ?array $attributeTypes = null): void
     {
-        $this->suspend(WorkflowCommand::upsertSearchAttributes($attributes));
+        $this->suspend(WorkflowCommand::upsertSearchAttributes($attributes, $attributeTypes));
     }
 
     /**

tests/SearchAttributeReplayTest.php

@@ -22,8 +22,8 @@
  * matches its history. This matters most for redrive, where the successor
  * reuses the failed run's upsert instead of running it again.
  *
- * This is value-only binding. Recorded attribute_types are not compared,
- * because this SDK's upsert command cannot declare types.
+ * These legacy one-argument calls bind values. Explicit type declarations
+ * and their replay identity are covered in TypedSearchAttributeReplayTest.
  */
 final class SearchAttributeReplayTest extends TestCase
 {

tests/Support/RedriveFixtureRegistry.php

@@ -28,6 +28,7 @@ final class RedriveFixtureRegistry
         'redrive-reused-signal-wait-clock' => ['SignalWaitOpened', 'SignalReceived', 'SignalApplied'],
         'redrive-reused-search-attributes-changed' => ['MemoUpserted', 'SearchAttributesUpserted'],
         'redrive-reused-search-attributes-missing' => ['MemoUpserted', 'SearchAttributesUpserted'],
+        'redrive-reused-search-attribute-type-changed' => ['MemoUpserted', 'SearchAttributesUpserted'],
         'redrive-successor-reused-markers' => ['SideEffectRecorded', 'VersionMarkerRecorded', 'ActivityCompleted'],
         'redrive-successor-reused-timer' => ['ActivityCompleted', 'TimerScheduled', 'TimerFired'],
         'redrive-successor-reused-cancellation-clock' => ['ActivityCompleted', 'TimerScheduled', 'TimerFired'],

tests/Support/ReplayRegressionFixture.php

@@ -643,6 +643,12 @@ private static function workflow(string $workflowType): callable
 
                 return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]);
             },
+            'golden.redrive-reused-search-attribute-types' => static function (WorkflowContext $context, mixed $name): mixed {
+                $context->upsertMemo(['stage' => 'greeting', 'customer' => $name]);
+                $context->upsertSearchAttributes(['Stage' => 'greeting'], ['Stage' => 'keyword']);
+
+                return $context->activity('Tests\\Fixtures\\V2\\TestRedriveFirstActivity', [$name]);
+            },
             'golden.redrive-reused-search-attributes-missing' => static function (WorkflowContext $context, mixed $name): mixed {
                 $context->upsertMemo(['stage' => 'greeting', 'customer' => $name]);
                 $context->upsertSearchAttributes(['Stage' => 'greeting']);