Prove published ordinary activity retry identity

.github/workflows/published-worker-metrics.yml

@@ -2,10 +2,10 @@ name: Published worker metrics
 
 on:
   pull_request:
-    paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py']
+    paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py', 'tests/smoke/activity-retry-identity.php', 'tests/smoke/activity-retry-identity-assertions.php', 'tests/PublishedActivityRetryIdentityTest.php', 'tests/ci/test_published_activity_identity_tuple.py']
   push:
     branches: [main]
-    paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py']
+    paths: ['.github/workflows/published-worker-metrics.yml', 'tests/smoke/worker-metrics.php', 'tests/smoke/update-outcome.php', 'tests/smoke/workflow-completion-limits.php', 'tests/smoke/query-results.php', 'tests/ci/test_published_query_tuple.py', 'tests/ci/check_published_laravel.py', 'tests/ci/test_published_laravel.py', 'tests/smoke/activity-retry-identity.php', 'tests/smoke/activity-retry-identity-assertions.php', 'tests/PublishedActivityRetryIdentityTest.php', 'tests/ci/test_published_activity_identity_tuple.py']
   workflow_dispatch:
     inputs:
       sdk_version:
@@ -20,6 +20,10 @@ on:
         description: Qualify binary and JSON query results with the installed Worker
         type: boolean
         default: false
+      activity_retry_identity:
+        description: Qualify ordinary retry IDs on published SDK 3.18.2/native 3.8.0
+        type: boolean
+        default: false
       laravel_bridges:
         description: Qualify registry SDK installation on both patched Laravel floors
         type: boolean
@@ -53,9 +57,22 @@ jobs:
       - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
         with:
           persist-credentials: false
+      - name: Require the published activity identity tuple
+        if: inputs.activity_retry_identity
+        env:
+          CANDIDATE_SOURCE: ${{ inputs.candidate_source }}
+          COMPLETION_LIMITS: ${{ inputs.completion_limits }}
+          QUERY_RESULTS: ${{ inputs.query_results }}
+          LARAVEL_BRIDGES: ${{ inputs.laravel_bridges }}
+        run: |
+          [[ "$GITHUB_EVENT_NAME" = workflow_dispatch && "$GITHUB_REF" = refs/heads/main ]]
+          [[ -z "$CANDIDATE_SOURCE" && "$COMPLETION_LIMITS" != true && "$QUERY_RESULTS" != true && "$LARAVEL_BRIDGES" != true ]]
+          [[ "$SDK_VERSION" = 3.18.2 && "$SERVER_VERSION" = 3.8.0 ]]
+          [[ "$SERVER_IMAGE" = durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e ]]
       - name: Require the exact published query tuple selectors
         run: |
           python3 -I -S -B tests/ci/test_published_query_tuple.py
+          python3 -I -S -B tests/ci/test_published_activity_identity_tuple.py
           python3 -S -B -m unittest discover -s tests/ci -p test_published_laravel.py
       - name: Create isolated Composer state
         run: |
@@ -170,6 +187,11 @@ jobs:
             if (($r["sdk_version"]??null)!==getenv("SDK_VERSION") || ($r["server_version"]??null)!==getenv("SERVER_VERSION")) {
               throw new RuntimeException("Query evidence tuple mismatch.");
             }'
+      - name: Assert ordinary activity retry identity over real HTTP
+        if: inputs.activity_retry_identity
+        run: |
+          php -l tests/smoke/activity-retry-identity.php
+          timeout 60s php tests/smoke/activity-retry-identity.php "$CONSUMER/vendor/autoload.php" | tee activity-retry-identity.json
       - name: Retain the tuple, observations and installed dependency graph
         if: always()
         uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2
@@ -180,6 +202,7 @@ jobs:
             update-outcome.json
             workflow-completion-limits.json
             query-results.json
+            activity-retry-identity.json
             candidate-package.json
             ${{ env.CONSUMER }}/candidate.zip
             ${{ env.CONSUMER }}/composer.lock
@@ -192,7 +215,7 @@ jobs:
         run: docker rm --force --volumes sdk-php-metrics-runtime
 
   installed-laravel-bridges:
-    if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges
+    if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity
     runs-on: ubuntu-24.04
     timeout-minutes: 15
     strategy:

tests/PublishedActivityRetryIdentityTest.php

@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Tests;
+
+use PHPUnit\Framework\TestCase;
+use RuntimeException;
+
+use function DurableWorkflow\Tests\Smoke\verifyActivityRetryIdentity;
+
+require_once __DIR__.'/smoke/activity-retry-identity-assertions.php';
+
+final class PublishedActivityRetryIdentityTest extends TestCase
+{
+    /** @return array{list<array<string, mixed>>, list<array<string, mixed>>} */
+    private static function evidence(): array
+    {
+        $callbacks = [];
+        foreach ([1, 2] as $attempt) {
+            $callbacks[] = ['activity_execution_id' => 'execution', 'workflow_id' => 'workflow', 'run_id' => 'run',
+                'task_id' => 'task-'.$attempt, 'activity_attempt_id' => 'attempt-'.$attempt, 'attempt_number' => $attempt];
+        }
+        $started = static fn (array $callback): array => ['event_type' => 'ActivityStarted', 'payload' => [
+            'activity_execution_id' => $callback['activity_execution_id'], 'task' => ['id' => $callback['task_id']],
+            'activity_attempt_id' => $callback['activity_attempt_id'], 'attempt_number' => $callback['attempt_number'],
+        ]];
+        $history = [
+            ['event_type' => 'ActivityScheduled'], $started($callbacks[0]),
+            ['event_type' => 'ActivityRetryScheduled', 'payload' => [
+                'activity_execution_id' => 'execution', 'retry_of_task_id' => 'task-1', 'retry_task_id' => 'task-2',
+                'retry_after_attempt_id' => 'attempt-1', 'retry_after_attempt' => 1, 'max_attempts' => 2, 'retry_backoff_seconds' => 0,
+            ]],
+            $started($callbacks[1]), ['event_type' => 'ActivityCompleted'], ['event_type' => 'WorkflowCompleted'],
+        ];
+
+        return [$callbacks, $history];
+    }
+
+    public function testBoundCallbackPairAndDurableRetryReceipt(): void
+    {
+        [$callbacks, $history] = self::evidence();
+        verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history);
+        self::assertCount(2, $callbacks);
+    }
+
+    public function testMissingLogicalIdentityCannotBorrowTaskIdentity(): void
+    {
+        [$callbacks, $history] = self::evidence();
+        $callbacks[0]['activity_execution_id'] = null;
+        $this->expectException(RuntimeException::class);
+        verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history);
+    }
+
+    public function testChangedLogicalIdentityAndReusedAttemptIdentitiesRefuse(): void
+    {
+        foreach (['activity_execution_id', 'workflow_id', 'run_id', 'task_id', 'activity_attempt_id', 'attempt_number'] as $field) {
+            [$callbacks, $history] = self::evidence();
+            $callbacks[1][$field] = in_array($field, ['task_id', 'activity_attempt_id', 'attempt_number'], true)
+                ? $callbacks[0][$field] : 'foreign';
+            try {
+                verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history);
+                self::fail('Accepted changed or reused '.$field);
+            } catch (RuntimeException $error) {
+                self::assertNotSame('', $error->getMessage());
+            }
+        }
+    }
+
+    public function testHistoryFromAnotherAttemptCannotQualifyTheCallbackPair(): void
+    {
+        [$callbacks, $history] = self::evidence();
+        $history[2]['payload']['retry_task_id'] = 'foreign-task';
+        $this->expectException(RuntimeException::class);
+        verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history);
+    }
+
+    public function testInventedSecondAttemptIdentityRefuses(): void
+    {
+        [$callbacks, $history] = self::evidence();
+        $callbacks[1]['activity_attempt_id'] = 'unrelated-nonempty-attempt';
+        $this->expectException(RuntimeException::class);
+        verifyActivityRetryIdentity($callbacks, 'workflow', 'run', $history);
+    }
+}

tests/ci/test_published_activity_identity_tuple.py

@@ -0,0 +1,60 @@
+"""Exercise the real pre-install selector guard without SDK or Docker execution."""
+
+import os
+from pathlib import Path
+import subprocess
+import unittest
+
+
+def guard_source():
+    workflow = (Path(__file__).resolve().parents[2] / ".github/workflows/published-worker-metrics.yml").read_text()
+    step = workflow.split("      - name: Require the published activity identity tuple\n", 1)[1].split("      - ", 1)[0]
+    return "\n".join(line[10:] for line in step.split("        run: |\n", 1)[1].splitlines())
+
+
+def require_disjoint_jobs(source):
+    expected = "    if: github.event_name == 'workflow_dispatch' && inputs.laravel_bridges && !inputs.activity_retry_identity"
+    header = source.split("  installed-laravel-bridges:\n", 1)[1].split("    runs-on:", 1)[0]
+    if header.strip() != expected.strip():
+        raise ValueError("The sibling Laravel job must not start for activity identity qualification")
+
+
+class PublishedActivityIdentityTupleTest(unittest.TestCase):
+    def setUp(self):
+        self.env = {
+            "PATH": os.defpath, "GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_REF": "refs/heads/main",
+            "SDK_VERSION": "3.18.2", "SERVER_VERSION": "3.8.0",
+            "SERVER_IMAGE": "durableworkflow/server@sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e",
+            "CANDIDATE_SOURCE": "", "COMPLETION_LIMITS": "false", "QUERY_RESULTS": "false", "LARAVEL_BRIDGES": "false",
+        }
+
+    def accepts(self, changes=None):
+        result = subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", guard_source()],
+                                env=self.env | (changes or {}), capture_output=True, check=False)
+        return result.returncode == 0
+
+    def test_exact_published_tuple(self):
+        self.assertTrue(self.accepts())
+
+    def test_source_selector_and_unrelated_qualification_flags_refuse(self):
+        for variable, value in [("CANDIDATE_SOURCE", "a" * 40), ("COMPLETION_LIMITS", "true"),
+                                ("QUERY_RESULTS", "true"), ("LARAVEL_BRIDGES", "true")]:
+            with self.subTest(variable=variable):
+                self.assertFalse(self.accepts({variable: value}))
+
+    def test_wrong_package_producer_and_dispatch_scope_refuse(self):
+        for variable, value in [("SDK_VERSION", "3.18.3"), ("SERVER_VERSION", "3.9.1"),
+                                ("SERVER_IMAGE", "durableworkflow/server:3.8.0"),
+                                ("GITHUB_EVENT_NAME", "pull_request"), ("GITHUB_REF", "refs/heads/topic")]:
+            with self.subTest(variable=variable):
+                self.assertFalse(self.accepts({variable: value}))
+
+    def test_sibling_job_cannot_start_on_conflicting_flags(self):
+        source = (Path(__file__).resolve().parents[2] / ".github/workflows/published-worker-metrics.yml").read_text()
+        require_disjoint_jobs(source)
+        with self.assertRaises(ValueError):
+            require_disjoint_jobs(source.replace(" && !inputs.activity_retry_identity", ""))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/smoke/activity-retry-identity-assertions.php

@@ -0,0 +1,66 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Tests\Smoke;
+
+use RuntimeException;
+
+/**
+ * @param list<array<string, mixed>> $callbacks
+ * @param list<array<string, mixed>> $history
+ */
+function verifyActivityRetryIdentity(array $callbacks, string $workflowId, string $runId, array $history): void
+{
+    if (count($callbacks) !== 2 || array_column($callbacks, 'attempt_number') !== [1, 2]) {
+        throw new RuntimeException('Require exactly the first and second real activity callbacks.');
+    }
+    foreach ($callbacks as $callback) {
+        foreach (['activity_execution_id', 'workflow_id', 'run_id', 'task_id', 'activity_attempt_id'] as $field) {
+            if (!is_string($callback[$field] ?? null) || $callback[$field] === '') {
+                throw new RuntimeException('Missing received callback identity: '.$field);
+            }
+        }
+        if ($callback['workflow_id'] !== $workflowId || $callback['run_id'] !== $runId) {
+            throw new RuntimeException('The callback must belong to the exact started workflow and run.');
+        }
+    }
+    [$first, $second] = $callbacks;
+    if ($first['activity_execution_id'] !== $second['activity_execution_id']
+        || $first['task_id'] === $second['task_id']
+        || $first['activity_attempt_id'] === $second['activity_attempt_id']) {
+        throw new RuntimeException('A retry must keep its logical execution and change task and attempt identity.');
+    }
+    foreach (['ActivityScheduled' => 1, 'ActivityStarted' => 2, 'ActivityRetryScheduled' => 1,
+        'ActivityCompleted' => 1, 'WorkflowCompleted' => 1, 'ActivityFailed' => 0, 'WorkflowFailed' => 0] as $type => $count) {
+        $events = array_values(array_filter($history, static fn (array $event): bool => ($event['event_type'] ?? null) === $type));
+        if (count($events) !== $count) {
+            throw new RuntimeException('Unexpected real history count for '.$type);
+        }
+        if ($type !== 'ActivityRetryScheduled') {
+            continue;
+        }
+        $payload = $events[0]['payload'] ?? [];
+        foreach ([
+            'activity_execution_id' => $first['activity_execution_id'],
+            'retry_of_task_id' => $first['task_id'], 'retry_task_id' => $second['task_id'],
+            'retry_after_attempt_id' => $first['activity_attempt_id'],
+            'retry_after_attempt' => 1, 'max_attempts' => 2, 'retry_backoff_seconds' => 0,
+        ] as $field => $expected) {
+            if (($payload[$field] ?? null) !== $expected) {
+                throw new RuntimeException('The durable retry receipt must bind the actual callbacks: '.$field);
+            }
+        }
+    }
+    $started = array_values(array_filter($history,
+        static fn (array $event): bool => ($event['event_type'] ?? null) === 'ActivityStarted'));
+    foreach ($callbacks as $index => $callback) {
+        $payload = $started[$index]['payload'] ?? [];
+        if (($payload['activity_execution_id'] ?? null) !== $callback['activity_execution_id']
+            || ($payload['task']['id'] ?? null) !== $callback['task_id']
+            || ($payload['activity_attempt_id'] ?? null) !== $callback['activity_attempt_id']
+            || ($payload['attempt_number'] ?? null) !== $callback['attempt_number']) {
+            throw new RuntimeException('Each callback must match its durable ActivityStarted identities.');
+        }
+    }
+}