Keep condition predicate fingerprints stable across formatting changes

CHANGELOG.md

@@ -20,6 +20,25 @@ project follows [Semantic Versioning](https://semver.org/).
   and group APIs are unchanged, and `childWorkflow()` refuses an `await`
   option.
 
+### Changed
+
+- New condition waits use versioned `dwphp2:sha256:` predicate fingerprints
+  (#293). Callable tokens exclude whitespace, comments and neighboring call-site
+  arguments, while preserving literals and operator boundaries. A predicate
+  change still fails replay. Indistinguishable same-line closures share a
+  combined fingerprint; put predicates on separate lines before starting runs
+  when they need independent identities. Unsupported source shapes or a PHP
+  build without the tokenizer retain the existing v1 computation.
+  Replaying v2 waits requires the tokenizer in every worker.
+- Replay accepts a recorded v1 fingerprint only when its original source-line
+  hash matches exactly. Physical reopens preserve their first recorded
+  fingerprint, including its absence. V2 formatting tolerance therefore applies
+  to waits first opened with v2, not retroactively to legacy runs.
+- Upgrade the queue's workers together before opening v2 waits. Older workers
+  reject v2 fingerprints, so rolling back below this SDK after v2 waits exist
+  can fail replay. Deploy sources atomically and restart workers, including
+  workers using OPcache.
+
 ## [3.9.4] - 2026-11-04
 
 ### Fixed

docs/portal/build/workflows-activities.md

@@ -405,7 +405,28 @@ A later deployment can raise the maximum while keeping the recorded versions in
 
 For a boolean rollout, use `patched('require-reservation-review')`. It returns `false` for legacy histories and `true` for new histories. After the legacy branch is no longer supported, replace the conditional call with `deprecatePatch()` for the same ID before eventually removing the marker. Compatible numeric and patch helpers share the same frozen version for a change ID. Every call must support that version, and `patched()` returns whether it equals `1`. Do not reuse a change ID for an unrelated rollout.
 
-A `waitCondition()` predicate is part of replay identity too. The worker fingerprints the source lines the closure occupies, so editing those lines while open runs have recorded that wait (even one already satisfied) raises `NonDeterministicWorkflow`. That includes re-indenting or editing an inner comment. Any other code on the predicate's lines, such as the variable it assigns, is hashed too. Moving the file or editing code above or below the predicate does not change the fingerprint. The key and timeout are part of replay identity in their own right; change them only behind a version marker, as with a changed predicate. [Issue #293](https://github.com/durable-workflow/sdk-php/issues/293) tracks a whitespace-insensitive fingerprint.
+A `waitCondition()` predicate is part of replay identity too. New waits use a
+`dwphp2:sha256:` fingerprint of the callable's tokens. Whitespace and comments
+outside literals, file locations and neighboring call-site arguments are
+excluded. Changing the predicate's tokens still raises
+`NonDeterministicWorkflow`, even for a wait that has already been satisfied.
+The key and timeout remain separate replay checks; change them behind a version
+marker when an open run can reach the call.
+
+Reflection can leave several same-line closures indistinguishable. Those
+closures share a combined candidate fingerprint, so changing either can affect
+both. Formatting that resolves that ambiguity can also change the candidate
+set. Put predicates on separate lines before starting runs when they need
+independent identities. If the callable cannot be isolated, or PHP lacks the
+tokenizer, the worker retains the v1 source-line fingerprint.
+Every worker that replays v2 waits needs PHP's tokenizer enabled.
+
+Waits first opened by older SDKs retain their exact v1 source-line check.
+Re-indenting those lines, changing a comment or editing neighboring code on
+the same lines can still fail legacy replay. A physical reopen echoes the
+first recorded fingerprint, so upgrading does not change an existing wait's
+identity. Upgrade the queue's workers together: older workers reject v2
+fingerprints, and cannot safely take over runs that already contain them.
 
 Deploy workflow code atomically and restart workers. The fingerprint reads the deployed source file. If files are replaced in place under any long-running worker that still has the old code loaded, the loaded code and the file on disk can disagree, and replay reports a mismatch until the worker restarts. `opcache.validate_timestamps=0` widens this window. Ship the PHP sources with the worker image; the fingerprint cannot be computed without them.
 

regression-corpus-policy.json

@@ -38,6 +38,12 @@
         {
           "glob": "src/Worker/Replayer.php"
         },
+        {
+          "glob": "src/Worker/ConditionWaitDefinition.php"
+        },
+        {
+          "glob": "src/Worker/ConditionPredicateSource.php"
+        },
         {
           "glob": "src/Worker/WorkflowEntryArguments.php"
         },

src/Worker/ConditionPredicateSource.php

@@ -0,0 +1,233 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Worker;
+
+use Closure;
+use PhpToken;
+use ReflectionFunction;
+use Throwable;
+
+/** @internal Normalizes callable source without evaluating the predicate or using compiler identities. */
+final class ConditionPredicateSource
+{
+    /** @var array<string, array{hash: string, tokens: list<PhpToken>, starts: array<int, list<int>>}> */
+    private static array $files = [];
+
+    public static function canonical(Closure $predicate): ?string
+    {
+        if (!class_exists(PhpToken::class, false)) {
+            return null;
+        }
+        try {
+            $reflection = new ReflectionFunction($predicate);
+            $file = $reflection->getFileName();
+            if (!is_string($file) || !is_file($file)) {
+                return null;
+            }
+            $source = self::fileTokens($file);
+            $tokens = $source['tokens'];
+            $candidates = [];
+            foreach ($source['starts'][$reflection->getStartLine()] ?? [] as $index) {
+                $candidate = self::candidate($tokens, $index, $reflection);
+                if ($candidate !== null) {
+                    $candidates[] = $candidate;
+                }
+            }
+            if ($candidates === []) {
+                return null;
+            }
+            // Reflection cannot always distinguish same-line closures. Preserve
+            // every possible candidate rather than guess or invoke user code.
+            $candidates = array_values(array_unique($candidates));
+            sort($candidates, SORT_STRING);
+
+            return self::frame($candidates);
+        } catch (Throwable) {
+            return null;
+        }
+    }
+
+    /** @return array{hash: string, tokens: list<PhpToken>, starts: array<int, list<int>>} */
+    private static function fileTokens(string $file): array
+    {
+        $source = @file_get_contents($file);
+        if (!is_string($source)) {
+            return ['hash' => '', 'tokens' => [], 'starts' => []];
+        }
+        $hash = hash('sha256', $source);
+        if ((self::$files[$file]['hash'] ?? null) === $hash) {
+            return self::$files[$file];
+        }
+        $tokens = array_values(array_filter(PhpToken::tokenize($source),
+            static fn (PhpToken $token): bool => !$token->is([T_WHITESPACE, T_COMMENT, T_DOC_COMMENT])));
+        $starts = [];
+        foreach ($tokens as $index => $token) {
+            if ($token->is([T_FUNCTION, T_FN])) {
+                $starts[$token->line][] = $index;
+            }
+        }
+        // Bound process memory and replace a file's entry when its bytes change.
+        // File paths and content hashes are cache keys, never fingerprint input.
+        if (count(self::$files) >= 8 && !isset(self::$files[$file])) {
+            array_shift(self::$files);
+        }
+        self::$files[$file] = ['hash' => $hash, 'tokens' => $tokens, 'starts' => $starts];
+
+        return self::$files[$file];
+    }
+
+    /** @param list<PhpToken> $tokens */
+    private static function candidate(array $tokens, int $index, ReflectionFunction $reflection): ?string
+    {
+        $cursor = $index + 1;
+        if (($tokens[$cursor]->text ?? null) === '&') {
+            ++$cursor;
+        }
+        $name = null;
+        if (($tokens[$cursor] ?? null)?->is(T_STRING)) {
+            $name = $tokens[$cursor++]->text;
+        }
+        // isAnonymous() starts at PHP 8.2. Anonymous reflection names contain
+        // this marker on 8.1 too; the rest of the name never enters the hash.
+        $anonymous = str_contains($reflection->getName(), '{closure');
+        if (($anonymous && $name !== null)
+            || (!$anonymous && $name !== $reflection->getShortName())
+            || ($tokens[$cursor]->text ?? null) !== '(') {
+            return null;
+        }
+        $parametersEnd = self::balancedEnd($tokens, $cursor);
+        if ($parametersEnd === null || !self::parametersMatch($tokens, $cursor, $parametersEnd, $reflection)) {
+            return null;
+        }
+        $start = self::declarationStart($tokens, $index);
+        $static = false;
+        foreach (array_slice($tokens, $start, $index - $start) as $modifier) {
+            $static = $static || $modifier->is(T_STATIC);
+        }
+        if ($static !== $reflection->isStatic()) {
+            return null;
+        }
+        $end = self::bodyEnd($tokens, $parametersEnd + 1, $tokens[$index]->is(T_FN));
+        if ($end === null || self::endLine($tokens[$end]) !== $reflection->getEndLine()) {
+            return null;
+        }
+
+        return self::frame(array_map(static fn (PhpToken $token): string => $token->text,
+            array_slice($tokens, $start, $end - $start + 1)));
+    }
+
+    /** @param list<PhpToken> $tokens */
+    private static function declarationStart(array $tokens, int $index): int
+    {
+        while ($index > 0 && $tokens[$index - 1]->is([T_STATIC, T_PUBLIC, T_PROTECTED, T_PRIVATE, T_FINAL, T_ABSTRACT])) {
+            --$index;
+        }
+
+        return $index;
+    }
+
+    /** @param list<PhpToken> $tokens */
+    private static function parametersMatch(array $tokens, int $start, int $end, ReflectionFunction $reflection): bool
+    {
+        $names = [];
+        for ($index = $start + 1; $index < $end; ++$index) {
+            $token = $tokens[$index];
+            if (self::opening($token) !== null) {
+                $index = self::balancedEnd($tokens, $index) ?? $end;
+            } elseif ($token->is(T_VARIABLE)) {
+                $names[] = substr($token->text, 1);
+            }
+        }
+        $parameters = array_map(static fn (\ReflectionParameter $parameter): string => $parameter->getName(), $reflection->getParameters());
+
+        return $names === $parameters;
+    }
+
+    /** @param list<PhpToken> $tokens */
+    private static function bodyEnd(array $tokens, int $start, bool $arrow): ?int
+    {
+        for ($index = $start, $count = count($tokens); $index < $count; ++$index) {
+            if ($arrow && $tokens[$index]->is(T_DOUBLE_ARROW)) {
+                return self::expressionEnd($tokens, $index + 1);
+            }
+            if (!$arrow && $tokens[$index]->text === '{') {
+                return self::balancedEnd($tokens, $index);
+            }
+            if (self::opening($tokens[$index]) !== null) {
+                $index = self::balancedEnd($tokens, $index) ?? $count;
+            }
+        }
+
+        return null;
+    }
+
+    /** @param list<PhpToken> $tokens */
+    private static function expressionEnd(array $tokens, int $start): ?int
+    {
+        $ternaries = 0;
+        for ($index = $start, $count = count($tokens); $index < $count; ++$index) {
+            $text = $tokens[$index]->text;
+            if (in_array($text, [',', ';', ')', ']', '}'], true) || ($text === ':' && $ternaries === 0)) {
+                return $index > $start ? $index - 1 : null;
+            }
+            if (self::opening($tokens[$index]) !== null) {
+                $end = self::balancedEnd($tokens, $index);
+                if ($end === null) {
+                    return null;
+                }
+                $index = $end;
+            } elseif ($text === '?') {
+                ++$ternaries;
+            } elseif ($text === ':') {
+                --$ternaries;
+            }
+        }
+
+        return $index > $start ? $index - 1 : null;
+    }
+
+    /** @param list<PhpToken> $tokens */
+    private static function balancedEnd(array $tokens, int $start): ?int
+    {
+        $stack = [];
+        for ($index = $start, $count = count($tokens); $index < $count; ++$index) {
+            $opening = self::opening($tokens[$index]);
+            if ($opening !== null) {
+                $stack[] = $opening;
+            } elseif (in_array($tokens[$index]->text, [')', ']', '}'], true)) {
+                if (array_pop($stack) !== $tokens[$index]->text) {
+                    return null;
+                }
+                if ($stack === []) {
+                    return $index;
+                }
+            }
+        }
+
+        return null;
+    }
+
+    private static function opening(PhpToken $token): ?string
+    {
+        return match ($token->text) {
+            '(' => ')',
+            '[', '#[' => ']',
+            '{', '${' => '}',
+            default => null,
+        };
+    }
+
+    private static function endLine(PhpToken $token): int
+    {
+        return $token->line + substr_count($token->text, "\n");
+    }
+
+    /** @param list<string> $parts */
+    private static function frame(array $parts): string
+    {
+        // Length framing preserves token boundaries, including + + versus ++.
+        return implode('', array_map(static fn (string $part): string => strlen($part).':'.$part, $parts));
+    }
+}

src/Worker/ConditionWaitDefinition.php

@@ -12,6 +12,16 @@
 final class ConditionWaitDefinition
 {
     public static function fingerprint(Closure $predicate): ?string
+    {
+        $tokens = ConditionPredicateSource::canonical($predicate);
+
+        return $tokens === null
+            ? self::legacyFingerprint($predicate)
+            : 'dwphp2:sha256:'.hash('sha256', $tokens);
+    }
+
+    /** The published v1 computation remains exact for histories opened by older workers. */
+    public static function legacyFingerprint(Closure $predicate): ?string
     {
         $source = self::source($predicate);
 

src/Worker/Replayer.php

@@ -420,6 +420,7 @@ public function replay(
                     $this->assertParallelPathMatches($step, $path);
                     if ($command->type === 'open_condition_wait') {
                         $this->assertConditionWaitCompatible($step, $command);
+                        $command = $command->withRecordedConditionDefinition($step['condition_definition_fingerprint']);
                     }
                     $nextSequence = max($nextSequence, $step['sequence'] + 1);
 
@@ -1318,6 +1319,7 @@ private function bindConditionReopens(array $steps): array
      */
     private function conditionReopen(WorkflowCommand $command, array $step, array $history): WorkflowCommand
     {
+        $command = $command->withRecordedConditionDefinition($step['condition_definition_fingerprint']);
         $timeout = $step['timeout_seconds'] ?? null;
         $openedAt = $step['opened_at'] ?? null;
         if (!is_int($timeout) || !$openedAt instanceof \DateTimeImmutable) {
@@ -3494,7 +3496,7 @@ private function assertConditionWaitCompatible(array $step, WorkflowCommand $com
             );
         }
         if ($step['condition_definition_fingerprint'] !== null
-            && $step['condition_definition_fingerprint'] !== $currentFingerprint) {
+            && !$command->conditionDefinitionMatches($step['condition_definition_fingerprint'])) {
             throw new NonDeterministicWorkflow(
                 'Condition wait predicate fingerprint changed during replay.',
                 $step['sequence'],

src/Worker/WorkflowCommand.php

@@ -217,6 +217,27 @@ public function conditionSatisfied(): bool
         return ($this->conditionPredicate)() === true;
     }
 
+    /** @internal V1 remains strict even when a new command would write v2. */
+    public function conditionDefinitionMatches(string $recorded): bool
+    {
+        return $recorded === ($this->attributes['condition_definition_fingerprint'] ?? null)
+            || ($this->conditionPredicate !== null
+                && preg_match('/\Asha256:[a-f0-9]{64}\z/', $recorded) === 1
+                && ConditionWaitDefinition::legacyFingerprint($this->conditionPredicate) === $recorded);
+    }
+
+    /** @internal A physical reopen keeps its logical wait's first fingerprint, including absence. */
+    public function withRecordedConditionDefinition(?string $fingerprint): self
+    {
+        $attributes = $this->attributes;
+        unset($attributes['condition_definition_fingerprint']);
+        if ($fingerprint !== null) {
+            $attributes['condition_definition_fingerprint'] = $fingerprint;
+        }
+
+        return $this->copyWithAttributes($attributes);
+    }
+
     /**
      * @param list<mixed> $arguments
      * @param array<string, mixed> $options
@@ -700,11 +721,17 @@ public static function workflowStream(array $directive): self
 
     /** @param array<string, mixed> $attributes */
     public function withAttributes(array $attributes): self
+    {
+        return $this->copyWithAttributes(array_merge($this->attributes, $attributes));
+    }
+
+    /** @param array<string, mixed> $attributes */
+    private function copyWithAttributes(array $attributes): self
     {
         return new self(
             $this->type,
             $this->historyShape,
-            array_merge($this->attributes, $attributes),
+            $attributes,
             $this->localResult,
             $this->sideEffect,
             $this->conditionPredicate,