Guard prepared retry admission with original claim authority

.github/workflows/phpunit-feature.yml

@@ -582,7 +582,7 @@ jobs:
                 composer install --no-interaction --no-progress --prefer-dist
                 mv vendor/durable-workflow/workflow /tmp/locked-workflow-package
                 cp -a prepared-workflow-source vendor/durable-workflow/workflow
-                php vendor/bin/phpunit tests/Feature/PreparedLocalActivityProtocolTest.php tests/Feature/ActivityAttemptStatusTest.php tests/Feature/HistoryRetentionTest.php tests/Unit/WorkerPollFenceTest.php tests/Source/CancellationCascadeDiagnosticsTest.php tests/Source/PreparedLocalCancellationPolicyTest.php tests/Source/CancellationScopeAdmissionTest.php tests/Source/ScopedActivityCancellationReceiptTest.php tests/Source/CancellationScopeDeliveryTest.php tests/Source/NexusTargetCancellationAdmissionTest.php tests/Feature/NexusTargetCancellationPolicyTest.php tests/Feature/WorkflowStreamsTest.php tests/Feature/WorkflowDebugTest.php --fail-on-skipped --log-junit "/evidence/$DW_PREPARED_DATABASE.xml" --colors=never
+                php vendor/bin/phpunit tests/Feature/PreparedLocalActivityProtocolTest.php tests/Feature/ActivityAttemptStatusTest.php tests/Feature/HistoryRetentionTest.php tests/Unit/WorkerPollFenceTest.php tests/Source/CancellationCascadeDiagnosticsTest.php tests/Source/PreparedLocalCancellationPolicyTest.php tests/Source/PreparedLocalRetryAdmissionSourceTest.php tests/Source/CancellationScopeAdmissionTest.php tests/Source/ScopedActivityCancellationReceiptTest.php tests/Source/CancellationScopeDeliveryTest.php tests/Source/NexusTargetCancellationAdmissionTest.php tests/Feature/NexusTargetCancellationPolicyTest.php tests/Feature/WorkflowStreamsTest.php tests/Feature/WorkflowDebugTest.php --fail-on-skipped --log-junit "/evidence/$DW_PREPARED_DATABASE.xml" --colors=never
                 if [ "$DW_PREPARED_DATABASE" = sqlite ]; then
                   php vendor/bin/phpunit tests/Feature/CooperativeCancellationProtocolTest.php tests/Feature/WorkflowWorkerProtocolTest.php tests/Feature/ActivityWorkerProtocolTest.php tests/Feature/SearchAttributeValueValidationTest.php tests/Feature/NamespaceDurableStateQuotaTest.php tests/Feature/RuntimeExternalPayloadTransportTest.php tests/Feature/RuntimePayloadCompletionUploadsTest.php tests/Unit/RuntimePayloadCompletionContextTest.php tests/Unit/WorkerProtocolOpenApiContractTest.php --log-junit /evidence/affected-regression.xml --colors=never
                 fi

app/Support/PreparedLocalActivityAdmissionRefused.php

@@ -2,5 +2,11 @@
 
 namespace App\Support;
 
-/** Roll back preparation if its original Server authority cannot be retained. */
-final class PreparedLocalActivityAdmissionRefused extends \RuntimeException {}
+/** Roll back preparation or replay when original authority cannot be retained. */
+final class PreparedLocalActivityAdmissionRefused extends \RuntimeException
+{
+    public function __construct(public readonly ?array $reply = null)
+    {
+        parent::__construct('Original prepared authority could not be retained.');
+    }
+}

app/Support/RuntimePayloadCompletionLease.php

@@ -7,6 +7,7 @@
 use Workflow\V2\Contracts\ActivityTaskBridge;
 use Workflow\V2\Contracts\CancellationScopeTaskBridge;
 use Workflow\V2\Contracts\PreparedLocalActivityTaskBridge;
+use Workflow\V2\Contracts\PreparedLocalActivityRetryTaskBridge;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
 use Workflow\V2\Enums\TaskType;
 use Workflow\V2\Models\ActivityAttempt;
@@ -139,14 +140,21 @@
             }
             if ($context->operation === 'local_activity_outcome') {
                 $attempt = ActivityAttempt::query()->where('workflow_task_id', $task->id)->find($context->activityAttemptId);
-                if (! $attempt instanceof ActivityAttempt
-                    || PreparedLocalActivityPolicy::originalClaim($task, $attempt, $context->leaseOwner, $context->attempt) !== $claim) {
+                $retry = $attempt instanceof ActivityAttempt && PreparedLocalRetryAdmission::retainedIntent($task, $attempt);
+                $originalClaim = $attempt instanceof ActivityAttempt
+                    ? ($retry ? PreparedLocalRetryAdmission::originalClaim($task, $attempt, $context->leaseOwner, $context->attempt)
+                        : PreparedLocalActivityPolicy::originalClaim($task, $attempt, $context->leaseOwner, $context->attempt))
+                    : null;
+                if (! $attempt instanceof ActivityAttempt || $originalClaim !== $claim) {
                     throw self::rejected();
                 }
-                /** @var PreparedLocalActivityTaskBridge $bridge */
+                /** @var PreparedLocalActivityTaskBridge|PreparedLocalActivityRetryTaskBridge $bridge */
                 $bridge = app(WorkflowTaskBridge::class);
-                $control = $bridge->controlLocalActivity($attempt->id, $context->leaseOwner, $context->attempt, false,
-                    PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION);
+                $control = $retry
+                    ? $bridge->controlLocalActivityWithRetryPolicy($attempt->id, $context->leaseOwner, $context->attempt, false,
+                        PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION)
+                    : $bridge->controlLocalActivity($attempt->id, $context->leaseOwner, $context->attempt, false,
+                        PreparedLocalActivityPolicy::MINIMUM_PROTOCOL_VERSION);
                 if (($control['active'] ?? null) !== true
                     || ($control['activity_attempt_id'] ?? null) !== $attempt->id
                     || ($control['workflow_task_id'] ?? null) !== $task->id

docs/contracts/prepared-local-retry-v2.md

@@ -157,3 +157,60 @@ canonical original mode/snapshot to the retained claim under the existing outer
 transaction. The current bound bridge does not expose that role. Native
 implementation, schema, installed/backend qualification and release remain
 separate work.
+
+## Guarded Server adapter
+
+The source adapter targets Core's optional interface and pure helpers at
+`ce24207ef86856e64f97b939c9aeef9d1fb387c8`. The locked package and default
+bridge remain unchanged. No retry capability is advertised by this adapter.
+Policy-member or override presence selects only untrusted intent. A real bound
+`PreparedLocalActivityRetryTaskBridge`, the original issued capability and
+canonical contract/snapshot authority are required independently.
+
+A successful version-2 preparation receives a distinct private
+`durable-workflow.server.prepared-local-authority/v2` receipt. Its exact keys
+are `schema`, `claim`, `activity_execution_id`, `activity_attempt_id`,
+`worker_attempt_id`, `local_activity_contract_version`, `retry_policy_snapshot`,
+`preparation_history_event_id` and `descriptor_fingerprint`. The contract value
+comes from canonical Core preparation. The complete normalized snapshot and
+unique correlated Started event must agree on every read. A legacy receipt is
+never rewritten or promoted. The hosting task's later current claim cannot
+replace the original callback's issued claim.
+
+A new private preparation receipt also requires the bound role's literal
+`prepared: true` and `duplicate: false` result. A duplicate can only read an
+already retained matching receipt. It cannot recreate missing Server authority
+from an existing Core graph.
+
+A fresh recovery also retains a separate closed private
+`durable-workflow.server.prepared-local-recovery-authority/v2` receipt on the
+recovering task. Its keys are `schema`, `claim`, `activity_execution_id`,
+`activity_attempt_id`, `local_activity_contract_version`, `retry_policy_snapshot`,
+`preparation_history_event_id`, `descriptor_fingerprint` and `engine_event`.
+The copied engine projection contains the event ID/type/history sequence,
+strict integer activity sequence and insertion time,
+retry-task identity, original recovery envelope and scalar retry decision.
+The decision's retained failure sample and history insertion time are distinct
+clocks. Their values are preserved independently rather than equated.
+
+Recovery reads the original callback authority and the recovering claim
+independently. An exact duplicate binds the old attempt even after a following
+attempt exists. A complete unchanged private recovery receipt permits Core's
+matching committed duplicate before fresh registration, quota or clock work.
+It grants no fresh writer authority. Missing receipts are never backfilled from
+an existing engine event, and changed graph, event or output identities roll
+back the transaction. Fresh recovery still requires the current recovering
+claim and registration fence. Preparation and fresh recovery retain their
+Server authority atomically with the Core writer. Physical callback stop remains
+unknown.
+
+Completed external payload uploads retain their existing completion-lease,
+namespace, storage and reference checks. The original receipt and supervisor
+control dispatch use the same proven mode as the outcome route. This adds no
+fetch-free completed replay guarantee. Failed and timed-out pure report checks
+perform no payload resolution.
+
+Normal locked-package tests cover unavailable-role precedence and unchanged
+legacy authority. The exact candidate writer/transaction/transport comparison
+and all semantic cases remain required qualification work. PHP syntax or pure
+intent controls alone do not qualify a prepared writer or enable a runtime role.

resources/platform-protocol-specs/worker-protocol-api.openapi.yaml

@@ -1,7 +1,7 @@
 openapi: 3.1.0
 info:
   title: durable-workflow.v2.worker-protocol-api
-  version: "57"
+  version: "58"
   summary: Durable Workflow worker-plane HTTP+JSON API
   description: >
     Normative OpenAPI specification for the worker-plane HTTP+JSON API:
@@ -2373,6 +2373,22 @@ components:
       type: object
       additionalProperties: false
       required: [type, activity_type, arguments]
+      allOf:
+        - if:
+            required: [retry_policy]
+            properties:
+              retry_policy:
+                type: object
+                anyOf:
+                  - required: [contract_version]
+                  - required: [snapshot_version]
+                  - required: [mode]
+                  - required: [maximum_interval_seconds]
+          then:
+            properties:
+              start_to_close_timeout: { type: [integer, "null"], minimum: 1, maximum: 2147483647 }
+              schedule_to_close_timeout: { type: [integer, "null"], minimum: 1, maximum: 2147483647 }
+              heartbeat_timeout: { type: [integer, "null"], minimum: 1, maximum: 2147483647 }
       properties:
         type: { type: string, const: record_local_activity }
         activity_type: { type: string, minLength: 1 }
@@ -2410,10 +2426,38 @@ components:
           items: { type: object }
         retry_policy:
           type: [object, "null"]
+          description: Versioned members select only untrusted contract-2 intent. Admission requires the actual optional retry bridge and the original activity_retry_delay_override_v1 claim capability. Legacy policies retain their existing path.
+          allOf:
+            - if:
+                type: object
+                anyOf:
+                  - required: [contract_version]
+                  - required: [snapshot_version]
+                  - required: [mode]
+                  - required: [maximum_interval_seconds]
+              then: { $ref: "#/components/schemas/PreparedLocalRetryPolicyV2" }
           properties:
             max_attempts: { type: [integer, "null"], minimum: 1 }
             backoff_seconds: { type: array, items: { type: integer, minimum: 0 } }
             non_retryable_error_types: { type: array, items: { type: string } }
+    PreparedLocalRetryPolicyV2:
+      type: object
+      additionalProperties: false
+      required: [contract_version, mode, backoff_seconds]
+      description: Flattened prepared-per-attempt policy. This schema does not grant mode or callback admission. The complete retained ten-key snapshot must be version 2 and match original Started authority. New-route timeouts are null or integers 1..2147483647; schedule-to-start remains null. Backend timestamp checks remain independent of the portable delay bound.
+      properties:
+        contract_version: { type: integer, const: 2 }
+        mode: { type: string, enum: [explicit, generated] }
+        max_attempts: { type: integer, minimum: 1, maximum: 1024 }
+        backoff_seconds:
+          type: array
+          maxItems: 1024
+          items: { type: integer, minimum: 0, maximum: 9007199254740991 }
+        maximum_interval_seconds: { type: integer, minimum: 0, maximum: 9007199254740991 }
+        non_retryable_error_types:
+          type: array
+          maxItems: 128
+          items: { type: string, minLength: 1, maxLength: 255, x-durable-workflow-max-utf8-bytes: 255 }
     PreparedLocalRecoverRequest:
       allOf:
         - $ref: "#/components/schemas/PreparedLocalClaimRequest"
@@ -2543,6 +2587,11 @@ components:
                     message: { type: string }
                     exception_type: { type: string, minLength: 1, maxLength: 255 }
                     non_retryable: { type: boolean, default: false }
+                    next_retry_delay_seconds:
+                      type: integer
+                      minimum: 0
+                      maximum: 9007199254740991
+                      description: Failed-only optional supplied delay. Absence differs from zero. Requires the installed retry role plus original claim capability, local contract 2 and retained snapshot 2; never upgrades a legacy report. Strict override failures are 422; other closed report failures are 409 invalid_local_activity_outcome.
                 - type: object
                   additionalProperties: false
                   required: [outcome]

tests/Feature/ActivityAttemptStatusTest.php

@@ -254,7 +254,7 @@ public function test_workflow_deadline_crossing_during_payload_resolution_cannot
 
         $this->withHeaders($this->headers())->postJson($this->path($task, 'complete'), $this->fence($task) + ['result' => null])
             ->assertConflict()->assertJsonPath('recorded', false)->assertJsonPath('reason', $reason);
-        $this->assertSame($before, $this->snapshot($task));
+        $this->assertRefusedOutcomeSnapshot($before, $task, 'completed', $reason);
     }
 
     #[DataProvider('expiredOutcomeProvider')]
@@ -277,7 +277,7 @@ public function test_outcome_crossing_expiry_during_payload_resolution_cannot_re
             ->postJson($this->path($task, $action), $this->fence($task) + $payload)
             ->assertStatus(409)->assertJsonPath('recorded', false)->assertJsonPath('reason', 'lease_expired');
 
-        $this->assertSame($before, $this->snapshot($task));
+        $this->assertRefusedOutcomeSnapshot($before, $task, $action === 'complete' ? 'completed' : 'failed', 'lease_expired');
     }
 
     public static function deadlineProvider(): array
@@ -665,6 +665,32 @@ private function observe(array $task, array $overrides = [], string $namespace =
         return $this->withHeaders($this->headers($namespace))->postJson($this->path($task), array_replace($this->fence($task), $overrides));
     }
 
+    private function assertRefusedOutcomeSnapshot(array $before, array $task, string $outcome, string $reason): void
+    {
+        $after = $this->snapshot($task);
+        // Maintained Core records a bounded rejection audit. The locked older
+        // package has no recorder and must retain its original history count.
+        $audited = class_exists(\Workflow\V2\Support\ActivityOutcomeRejection::class);
+        $this->assertSame($before[4] + ($audited ? 1 : 0), $after[4]);
+        unset($before[4], $after[4]);
+        $this->assertSame($before, $after);
+        if (! $audited) {
+            return;
+        }
+        $event = WorkflowHistoryEvent::query()->where('workflow_run_id', $task['run_id'])
+            ->where('event_type', 'ActivityOutcomeRejected')->sole();
+        $this->assertSame($task['task_id'], $event->workflow_task_id);
+        $expected = ['activity_execution_id' => $task['activity_execution_id'],
+            'activity_attempt_id' => $task['activity_attempt_id'],
+            'attempt_number' => ActivityAttempt::query()->findOrFail($task['activity_attempt_id'])->attempt_number,
+            'lease_owner' => $task['lease_owner'], 'outcome' => $outcome, 'reason' => $reason,
+            'observed_at' => now()->toJSON()];
+        $actual = $event->payload;
+        ksort($expected);
+        ksort($actual);
+        $this->assertSame($expected, $actual);
+    }
+
     private function snapshot(array $task): array
     {
         return [