From 434c0760bd6e909a93295d3ac80675221c17db49 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Sun, 25 Oct 2026 22:32:32 +0000
Subject: [PATCH 01/13] Derive workflow service authority from authenticated
 requests

---
 app/Http/Controllers/Api/WorkerController.php | 33 ++++++++----
 app/Providers/AppServiceProvider.php          |  5 ++
 app/Support/WorkflowPackageApiFloor.php       |  6 +++
 .../WorkflowServiceOperationAuthorizer.php    | 53 +++++++++++++++++++
 docs/contracts/nexus.md                       | 12 +++++
 tests/Feature/WorkerServiceOperationTest.php  | 28 +++++++++-
 6 files changed, 124 insertions(+), 13 deletions(-)
 create mode 100644 app/Support/WorkflowServiceOperationAuthorizer.php

diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php
index d1b25c6b2..c19c53c01 100644
--- a/app/Http/Controllers/Api/WorkerController.php
+++ b/app/Http/Controllers/Api/WorkerController.php
@@ -3212,15 +3212,11 @@ private function authorizeServiceOperationCommands(Request $request, string $nam
                 );
             }
 
-            // A worker owns command intent, not the authenticated caller identity.
+            // Principal authority travels through the request-bound authorizer,
+            // never through worker command fields.
             $commands[$index] = array_replace($command, [
                 'namespace' => $targetNamespace,
                 'caller_namespace' => $namespace,
-                'principal_subject' => $principal->subject(),
-                'principal_method' => $principal->method(),
-                'principal_roles' => $principal->roles(),
-                'principal_tenant' => $principal->tenant(),
-                'principal_claims' => $principal->claims(),
             ]);
         }
 
@@ -3253,23 +3249,38 @@ private function authorizeServiceOperationReplays(Request $request, string $name
                     ->where('idempotency_key', trim($key))
                     ->oldest('created_at')->oldest('id');
             }
-            $call = $query->first();
+            $call = $query->lockForUpdate()->first();
             if ($call === null && $callId === null) {
                 continue;
             }
 
             $runId = WorkflowTask::query()->whereKey($taskId)->value('workflow_run_id');
+            $instanceId = WorkflowRun::query()->whereKey($runId)->value('workflow_instance_id');
             $principal = Authenticate::principal($request);
             abort_unless($call !== null && $principal !== null
                 && $call->caller_namespace === $namespace
                 && $call->caller_workflow_run_id === $runId
+                && $call->caller_workflow_instance_id === $instanceId
                 && $call->endpoint_name === strtolower(trim($command['endpoint_name']))
                 && $call->service_name === strtolower(trim($command['service_name']))
                 && $call->operation_name === strtolower(trim($command['operation_name'])), 403);
 
             // Existing call IDs and idempotency keys must not skip current boundary policy.
-            $operation = $call->operation;
-            abort_unless($operation !== null && $call->endpoint !== null && $call->service !== null, 403);
+            // Hold the catalog rows through Core replay so a concurrent policy
+            // change cannot race between this decision and the early return.
+            $endpoint = $call->endpoint()->lockForUpdate()->first();
+            $service = $call->service()->lockForUpdate()->first();
+            $operation = $call->operation()->lockForUpdate()->first();
+            abort_unless($operation !== null && $endpoint !== null && $service !== null, 403);
+            abort_unless($endpoint->namespace === $call->target_namespace
+                && $service->namespace === $call->target_namespace
+                && $operation->namespace === $call->target_namespace
+                && $endpoint->endpoint_name === $call->endpoint_name
+                && $service->service_name === $call->service_name
+                && $operation->operation_name === $call->operation_name
+                && $service->workflow_service_endpoint_id === $endpoint->id
+                && $operation->workflow_service_endpoint_id === $endpoint->id
+                && $operation->workflow_service_id === $service->id, 403);
             $rejection = app(ServiceCallBoundary::class)->replayRejectionFor(
                 principal: $principal,
                 callerNamespace: $namespace,
@@ -3280,8 +3291,8 @@ private function authorizeServiceOperationReplays(Request $request, string $name
                 callerWorkflowRunId: $runId,
                 idempotencyKey: $key,
                 operationModeOverride: $command['mode_override'] ?? null,
-                endpointBoundaryPolicy: $call->endpoint->boundary_policy ?? [],
-                serviceBoundaryPolicy: $call->service->boundary_policy ?? [],
+                endpointBoundaryPolicy: $endpoint->boundary_policy ?? [],
+                serviceBoundaryPolicy: $service->boundary_policy ?? [],
                 operationBoundaryPolicy: $operation->boundary_policy ?? [],
             );
             abort_if($rejection !== null, 403);
diff --git a/app/Providers/AppServiceProvider.php b/app/Providers/AppServiceProvider.php
index 8a940502d..e7d204744 100644
--- a/app/Providers/AppServiceProvider.php
+++ b/app/Providers/AppServiceProvider.php
@@ -35,6 +35,7 @@
 use App\Support\WorkflowMemoRollingCompatibility;
 use App\Support\WorkflowObservability;
 use App\Support\WorkflowPackageApiFloor;
+use App\Support\WorkflowServiceOperationAuthorizer;
 use App\Support\WorkflowTaskLeaseConfiguration;
 use Illuminate\Console\Events\CommandStarting;
 use Illuminate\Contracts\Bus\Dispatcher as BusDispatcher;
@@ -181,6 +182,10 @@ public function register(): void
         $this->app->singleton(ServiceBoundaryAuditRecorder::class);
 
         $this->app->singleton(ServiceCallBoundary::class);
+        $this->app->bind(
+            \Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer::class,
+            WorkflowServiceOperationAuthorizer::class,
+        );
     }
 
     private function normalizeRedisPorts(): void
diff --git a/app/Support/WorkflowPackageApiFloor.php b/app/Support/WorkflowPackageApiFloor.php
index d094f473e..488d7b2be 100644
--- a/app/Support/WorkflowPackageApiFloor.php
+++ b/app/Support/WorkflowPackageApiFloor.php
@@ -12,8 +12,10 @@
 use Workflow\V2\Contracts\ExternalPayloadStoragePolicy;
 use Workflow\V2\Contracts\MatchingRole;
 use Workflow\V2\Contracts\ServiceControlPlane;
+use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
 use Workflow\V2\Exceptions\ExternalPayloadIntegrityException;
+use Workflow\V2\Exceptions\ServiceOperationAuthorizationException;
 use Workflow\V2\Exceptions\WorkflowOutputCodecUnavailableException;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowSearchAttribute;
@@ -35,6 +37,7 @@
 use Workflow\V2\Support\WorkflowCommandNormalizer;
 use Workflow\V2\Support\WorkflowDefinition;
 use Workflow\V2\Support\WorkflowQueryContract;
+use Workflow\V2\Support\WorkflowServiceOperationAuthority;
 use Workflow\V2\Support\WorkflowTaskLease;
 
 /**
@@ -62,6 +65,7 @@ final class WorkflowPackageApiFloor
      * remediation hint that points at the upgrade.
      */
     private const REQUIRED_APIS = [
+        [WorkflowServiceOperationAuthority::class, 'options'],
         // CodecRegistry::universal() is the public Avro-only codec authority
         // used by /api/cluster/info and the embedded control-plane request
         // contract.
@@ -157,6 +161,7 @@ final class WorkflowPackageApiFloor
      * Concrete classes the server instantiates or catches directly.
      */
     private const REQUIRED_CLASSES = [
+        ServiceOperationAuthorizationException::class,
         ExternalPayloadIntegrityException::class,
         LocalFilesystemExternalPayloadStorage::class,
         WorkflowOutputCodecUnavailableException::class,
@@ -189,6 +194,7 @@ final class WorkflowPackageApiFloor
      * Interface methods the server type-hints directly.
      */
     private const REQUIRED_INTERFACE_APIS = [
+        [WorkflowServiceOperationAuthorizer::class, 'authorize'],
         [MatchingRole::class, 'wake'],
         [MatchingRole::class, 'runPass'],
         [ServiceControlPlane::class, 'execute'],
diff --git a/app/Support/WorkflowServiceOperationAuthorizer.php b/app/Support/WorkflowServiceOperationAuthorizer.php
new file mode 100644
index 000000000..b2766da81
--- /dev/null
+++ b/app/Support/WorkflowServiceOperationAuthorizer.php
@@ -0,0 +1,53 @@
+<?php
+
+namespace App\Support;
+
+use App\Contracts\AuthProvider;
+use App\Http\Middleware\Authenticate;
+use Illuminate\Http\Request;
+use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer as Authorizer;
+use Workflow\V2\Exceptions\ServiceOperationAuthorizationException;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Support\ServiceCallPrincipal;
+
+/** The request's authenticated principal authorizes intent from a leased run. */
+final class WorkflowServiceOperationAuthorizer implements Authorizer
+{
+    public function __construct(
+        private readonly Request $request,
+        private readonly AuthProvider $auth,
+        private readonly RouteAuthorizationResource $resources,
+    ) {}
+
+    public function authorize(WorkflowRun $run, array $command): ServiceCallPrincipal
+    {
+        $principal = Authenticate::principal($this->request);
+        if ($principal === null || ! is_string($run->namespace) || $run->namespace === '') {
+            throw new ServiceOperationAuthorizationException;
+        }
+        $target = $command['namespace'] ?? $run->namespace;
+        $resource = array_replace($this->resources->make($this->request, ['worker']), [
+            'operation_family' => 'service',
+            'operation_name' => 'execute_operation',
+            'target_namespace' => $target,
+            'namespace_name' => $target,
+            'caller_namespace' => $run->namespace,
+            'caller_workflow_instance_id' => $run->workflow_instance_id,
+            'caller_workflow_run_id' => $run->id,
+            'service_endpoint_name' => strtolower(trim($command['endpoint_name'])),
+            'service_name' => strtolower(trim($command['service_name'])),
+            'service_operation_name' => strtolower(trim($command['operation_name'])),
+        ]);
+        if (! $this->auth->authorize($principal, 'execute_operation', $resource)) {
+            throw new ServiceOperationAuthorizationException;
+        }
+
+        return new ServiceCallPrincipal(
+            subject: $principal->subject(),
+            method: $principal->method(),
+            roles: $principal->roles(),
+            tenant: $principal->tenant(),
+            claims: $principal->claims(),
+        );
+    }
+}
diff --git a/docs/contracts/nexus.md b/docs/contracts/nexus.md
index f68eed77a..f7613e0fc 100644
--- a/docs/contracts/nexus.md
+++ b/docs/contracts/nexus.md
@@ -174,6 +174,18 @@ operator and is read-only.
 
 ## Durable record and retry semantics
 
+Workflow task completion derives caller namespace and run/instance identity from
+the leased run. A request-bound Core service-operation authorizer supplies the
+authenticated worker principal and checks requested target scope through the
+server's current authorization policy. Principal command fields are prohibited;
+they cannot supply roles, tenant or claims. An existing call or idempotency-key
+match must also belong to that caller run and instance and the exact catalog
+operation. The server retains current boundary-policy checks on replay.
+
+The server requires the corresponding Core authorizer contract at boot. An
+older package is refused with the package-floor upgrade diagnostic, rather than
+falling back to a synthetic system principal.
+
 Every accepted Nexus call writes one row in `workflow_service_calls`
 keyed by the durable service-call id (a ULID). The row carries:
 
diff --git a/tests/Feature/WorkerServiceOperationTest.php b/tests/Feature/WorkerServiceOperationTest.php
index dc2a4ad6a..ebae9a8fe 100644
--- a/tests/Feature/WorkerServiceOperationTest.php
+++ b/tests/Feature/WorkerServiceOperationTest.php
@@ -97,8 +97,10 @@ public function test_supported_options_reach_the_service_control_plane(): void
                 }
                 $this->assertSame([['invoice' => 42]], $actual['arguments']);
                 $this->assertSame('avro', $actual['payload_codec']);
-                $this->assertSame('nexus-test-principal', $actual['principal_subject']);
-                $this->assertSame(['worker'], $actual['principal_roles']);
+                $this->assertSame('nexus-test-principal', $actual['principal']->subject);
+                $this->assertSame(['worker'], $actual['principal']->roles);
+                $this->assertArrayNotHasKey('principal_subject', $actual);
+                $this->assertArrayNotHasKey('principal_roles', $actual);
                 $this->assertSame($task['run_id'], $actual['caller_workflow_run_id']);
 
                 return true;
@@ -309,6 +311,28 @@ public function test_service_call_replays_are_authorized_against_current_catalog
         $this->assertDatabaseCount('workflow_runs', 2);
     }
 
+    #[DataProvider('replaySelectors')]
+    public function test_service_call_replay_requires_the_exact_caller_instance(string $selector): void
+    {
+        $this->seedCatalog();
+        $task = $this->leaseWorkflow();
+        $this->authenticateWorker();
+        $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
+        $call = WorkflowServiceCall::query()->sole();
+        $call->update(['caller_workflow_instance_id' => 'another-instance']);
+        $task = $this->pollWorkflow();
+        $before = $call->getRawOriginal();
+        $history = WorkflowHistoryEvent::query()->count();
+
+        $this->complete($task, $this->command([$selector => $selector === 'service_call_id' ? $call->id : 'invoice-42']))
+            ->assertForbidden();
+
+        $this->assertSame($before, $call->fresh()->getRawOriginal());
+        $this->assertSame($history, WorkflowHistoryEvent::query()->count());
+        $this->assertSame('leased', WorkflowTask::query()->findOrFail($task['task_id'])->status->value);
+        $this->assertDatabaseCount('workflow_runs', 2);
+    }
+
     public static function replaySelectors(): array
     {
         return [['service_call_id'], ['idempotency_key']];

From 0ba2cba11abf9eabec57e0a6d314442c84651773 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 11:43:35 +0000
Subject: [PATCH 02/13] Pin durable-workflow/workflow 2.14.0 (b814c6c4)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 composer.json |  2 +-
 composer.lock | 16 ++++++++--------
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/composer.json b/composer.json
index ec912de86..cf9c609f8 100644
--- a/composer.json
+++ b/composer.json
@@ -6,7 +6,7 @@
     "require": {
         "php": "^8.2",
         "apache/avro": "^1.12",
-        "durable-workflow/workflow": "2.13.1",
+        "durable-workflow/workflow": "2.14.0",
         "laravel/framework": "^13.30",
         "laravel/tinker": "^3.0",
         "league/flysystem-aws-s3-v3": "^3.35.3"
diff --git a/composer.lock b/composer.lock
index aae870640..03ff13b20 100644
--- a/composer.lock
+++ b/composer.lock
@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "2e60ebd4e4219304b0287bb1c5215a43",
+    "content-hash": "48ed4385adfa0d41547d07c0008fbf0d",
     "packages": [
         {
             "name": "apache/avro",
@@ -655,16 +655,16 @@
         },
         {
             "name": "durable-workflow/workflow",
-            "version": "2.13.1",
+            "version": "2.14.0",
             "source": {
                 "type": "git",
                 "url": "https://github.com/durable-workflow/workflow.git",
-                "reference": "b2a3369110b8cbdf060802ff91e475110e66e333"
+                "reference": "b814c6c485b1c072f4092d3f478c7deba9c8c241"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/b2a3369110b8cbdf060802ff91e475110e66e333",
-                "reference": "b2a3369110b8cbdf060802ff91e475110e66e333",
+                "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/b814c6c485b1c072f4092d3f478c7deba9c8c241",
+                "reference": "b814c6c485b1c072f4092d3f478c7deba9c8c241",
                 "shasum": ""
             },
             "require": {
@@ -705,7 +705,7 @@
                     "dev-main": "2.0.x-dev"
                 },
                 "durable-workflow": {
-                    "product-train": "2.13.1",
+                    "product-train": "2.14.0",
                     "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
                     "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
                 }
@@ -732,9 +732,9 @@
             "description": "Embedded durable workflow runtime and orchestration engine for Laravel applications.",
             "support": {
                 "issues": "https://github.com/durable-workflow/workflow/issues",
-                "source": "https://github.com/durable-workflow/workflow/tree/2.13.1"
+                "source": "https://github.com/durable-workflow/workflow/tree/2.14.0"
             },
-            "time": "2026-10-24T14:18:09+00:00"
+            "time": "2026-10-26T10:33:13+00:00"
         },
         {
             "name": "egulias/email-validator",

From cf17dea477ca16a78dfc60a278afe686fbd1da73 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 12:52:03 +0000
Subject: [PATCH 03/13] Pin durable-workflow/workflow 2.14.1 (af9dbadb)

2.14.1 is 2.14.0 plus the external-worker pending-count limits (Workflow
#933, #939) and its release manifest. It replaces the held 2.14.0 pin.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 composer.json |  2 +-
 composer.lock | 16 ++++++++--------
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/composer.json b/composer.json
index cf9c609f8..01265f8d8 100644
--- a/composer.json
+++ b/composer.json
@@ -6,7 +6,7 @@
     "require": {
         "php": "^8.2",
         "apache/avro": "^1.12",
-        "durable-workflow/workflow": "2.14.0",
+        "durable-workflow/workflow": "2.14.1",
         "laravel/framework": "^13.30",
         "laravel/tinker": "^3.0",
         "league/flysystem-aws-s3-v3": "^3.35.3"
diff --git a/composer.lock b/composer.lock
index 03ff13b20..30ceaee8c 100644
--- a/composer.lock
+++ b/composer.lock
@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "48ed4385adfa0d41547d07c0008fbf0d",
+    "content-hash": "bf409453a4d519b4f3745e9fe27c3e0f",
     "packages": [
         {
             "name": "apache/avro",
@@ -655,16 +655,16 @@
         },
         {
             "name": "durable-workflow/workflow",
-            "version": "2.14.0",
+            "version": "2.14.1",
             "source": {
                 "type": "git",
                 "url": "https://github.com/durable-workflow/workflow.git",
-                "reference": "b814c6c485b1c072f4092d3f478c7deba9c8c241"
+                "reference": "af9dbadb41bf96f51ad78047022c392887e3c7a6"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/b814c6c485b1c072f4092d3f478c7deba9c8c241",
-                "reference": "b814c6c485b1c072f4092d3f478c7deba9c8c241",
+                "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/af9dbadb41bf96f51ad78047022c392887e3c7a6",
+                "reference": "af9dbadb41bf96f51ad78047022c392887e3c7a6",
                 "shasum": ""
             },
             "require": {
@@ -705,7 +705,7 @@
                     "dev-main": "2.0.x-dev"
                 },
                 "durable-workflow": {
-                    "product-train": "2.14.0",
+                    "product-train": "2.14.1",
                     "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
                     "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
                 }
@@ -732,9 +732,9 @@
             "description": "Embedded durable workflow runtime and orchestration engine for Laravel applications.",
             "support": {
                 "issues": "https://github.com/durable-workflow/workflow/issues",
-                "source": "https://github.com/durable-workflow/workflow/tree/2.14.0"
+                "source": "https://github.com/durable-workflow/workflow/tree/2.14.1"
             },
-            "time": "2026-10-26T10:33:13+00:00"
+            "time": "2026-10-26T12:23:08+00:00"
         },
         {
             "name": "egulias/email-validator",

From 9bfc79a2e8c023437a26e092b5b57a7641bbc73e Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 13:17:44 +0000
Subject: [PATCH 04/13] Adopt Workflow 2.14.1 service-operation scheduling and
 authority (WIP)

Workflow 2.14 re-authorizes an admitted service call against its stored
caller scope and current catalog policy, schedules a workflow caller's
handler after its completion commits, and bounds every service-operation
wait at 60 seconds.

- The execute route passes the admitted caller namespace and the request
  principal into the control plane, so the package checks the identity
  the boundary admitted; a package authority refusal is a plain 403.
- The execute route and workflow-task completion validation refuse a
  service-operation wait above 60 seconds with the Server's own message.
- Tests run the scheduled RunServiceCall effects the queue worker would,
  and assert the scheduled admission (ServiceOperationScheduled, no
  synthesized ServiceCallStarted) before them.

Still open: protocol 1.21 and its gate, and the execute route's contract
for callers that name a workflow (scheduled, not inline).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../Api/ServiceCatalogController.php          | 61 ++++++++++++++++---
 app/Support/ServiceOperationUpdateWait.php    | 12 ++--
 .../WorkflowTaskCompletionValidation.php      |  2 +-
 .../WorkerServiceOperationEnvelopeTest.php    |  9 ++-
 tests/Feature/Concerns/ServerTestHelpers.php  | 20 ++++++
 .../ServiceOperationUpdateWaitTest.php        | 22 ++++---
 tests/Feature/WorkerServiceOperationTest.php  | 23 ++++++-
 7 files changed, 122 insertions(+), 27 deletions(-)

diff --git a/app/Http/Controllers/Api/ServiceCatalogController.php b/app/Http/Controllers/Api/ServiceCatalogController.php
index 2f927b297..cd4510e82 100644
--- a/app/Http/Controllers/Api/ServiceCatalogController.php
+++ b/app/Http/Controllers/Api/ServiceCatalogController.php
@@ -16,11 +16,14 @@
 use Illuminate\Http\Request;
 use Illuminate\Validation\ValidationException;
 use Workflow\V2\Contracts\ServiceControlPlane;
+use Workflow\V2\Exceptions\ServiceOperationAuthorizationException;
 use Workflow\V2\Models\WorkflowService;
 use Workflow\V2\Models\WorkflowServiceCall;
 use Workflow\V2\Models\WorkflowServiceEndpoint;
 use Workflow\V2\Models\WorkflowServiceOperation;
 use Workflow\V2\Support\ServiceCallDeadlinePolicy;
+use Workflow\V2\Support\ServiceCallPrincipal;
+use Workflow\V2\Support\UpdateWaitPolicy;
 
 class ServiceCatalogController
 {
@@ -637,7 +640,8 @@ public function executeOperation(
             'payload_codec' => ['nullable', 'string', 'max:191'],
             'mode_override' => ['nullable', 'string', 'in:'.implode(',', self::OPERATION_MODES)],
             'wait_for' => ['nullable', 'string', 'in:accepted,completed'],
-            'wait_timeout_seconds' => ['nullable', 'integer', 'min:0', 'max:86400'],
+            // Workflow 2.14 bounds every service-operation wait at 60 seconds.
+            'wait_timeout_seconds' => ['nullable', 'integer', 'min:0', 'max:'.UpdateWaitPolicy::MAX_COMPLETION_TIMEOUT_SECONDS],
             'idempotency_key' => ['nullable', 'string', 'max:191'],
             'caller_namespace' => ['nullable', 'string', 'max:255'],
             'caller_workflow_instance_id' => ['nullable', 'string', 'max:191'],
@@ -760,16 +764,20 @@ public function executeOperation(
 
             if ((string) $idempotentCall->status === 'accepted'
                 && ! $this->acceptedCallHasIssuedHandler($idempotentCall)) {
-                $options = $executionOptions;
+                $options = $this->admittedExecutionOptions($request, $executionOptions, $callerNamespace);
                 $options['service_call_id'] = $idempotentCall->id;
                 $options['boundary_policy_outcome'] = 'accepted';
 
-                $result = $this->serviceControlPlane()->execute(
-                    $endpoint->endpoint_name,
-                    $service->service_name,
-                    $operation->operation_name,
-                    $options,
-                );
+                try {
+                    $result = $this->serviceControlPlane()->execute(
+                        $endpoint->endpoint_name,
+                        $service->service_name,
+                        $operation->operation_name,
+                        $options,
+                    );
+                } catch (ServiceOperationAuthorizationException) {
+                    return $this->serviceOperationForbidden();
+                }
                 $result['idempotent_replay'] = true;
             } else {
                 $result = array_replace(
@@ -815,7 +823,7 @@ public function executeOperation(
             );
         }
 
-        $options = $executionOptions;
+        $options = $this->admittedExecutionOptions($request, $executionOptions, $callerNamespace);
         $options['service_call_id'] = $admission->call->id;
         $options['boundary_policy_outcome'] = $admission->decision->outcome->value;
 
@@ -826,6 +834,8 @@ public function executeOperation(
                 $operation->operation_name,
                 $options,
             );
+        } catch (ServiceOperationAuthorizationException) {
+            return $this->serviceOperationForbidden();
         } finally {
             if ($admission->request->operationMode->value === 'sync') {
                 $this->serviceCallBoundary()->release($admission->request);
@@ -1172,6 +1182,39 @@ private function principal(Request $request): Principal
             ?? Principal::role('operator', 'none', subject: 'unknown');
     }
 
+    /**
+     * Workflow 2.14 re-authorizes an admitted call against its stored caller
+     * scope and current catalog policy before any effect. Give it the caller
+     * namespace and principal the boundary admitted, so the package checks
+     * the same identity instead of its own configured defaults.
+     *
+     * @param  array<string, mixed>  $options
+     * @return array<string, mixed>
+     */
+    private function admittedExecutionOptions(Request $request, array $options, string $callerNamespace): array
+    {
+        $principal = $this->principal($request);
+        $options['caller_namespace'] = $callerNamespace;
+        $options['principal'] = new ServiceCallPrincipal(
+            subject: $principal->subject(),
+            method: $principal->method(),
+            roles: $principal->roles(),
+            tenant: $principal->tenant(),
+            claims: $principal->claims(),
+        );
+
+        return $options;
+    }
+
+    /** A package authority refusal: forbidden, with no ownership or policy detail. */
+    private function serviceOperationForbidden(): JsonResponse
+    {
+        return ControlPlaneProtocol::json([
+            'reason' => 'service_operation_forbidden',
+            'message' => 'The service operation is not authorized for this caller.',
+        ], 403);
+    }
+
     /**
      * @return array<string, mixed>
      */
diff --git a/app/Support/ServiceOperationUpdateWait.php b/app/Support/ServiceOperationUpdateWait.php
index 5873a2c1d..72bbf849d 100644
--- a/app/Support/ServiceOperationUpdateWait.php
+++ b/app/Support/ServiceOperationUpdateWait.php
@@ -11,11 +11,13 @@
  * The update wait ceiling on service operations bound to a workflow update
  * (Server #775).
  *
- * A service operation's own `wait_timeout_seconds` contract (up to 86400 on
- * the execute route) is unchanged. When the operation's binding is a workflow
- * update, though, the service control plane hands that wait to the update
- * admission, so it is refused above the update ceiling before the operation
- * runs, as the update route refuses it.
+ * When the operation's binding is a workflow update, the service control
+ * plane hands that wait to the update admission, so it is refused above the
+ * update ceiling before the operation runs, as the update route refuses it.
+ * Since Workflow 2.14 every service-operation wait has the same 60-second
+ * bound, which the execute route and workflow-task completion validation
+ * also enforce; this check keeps the update route's message for a binding
+ * that changes after those rules ran.
  */
 final class ServiceOperationUpdateWait
 {
diff --git a/app/Support/WorkflowTaskCompletionValidation.php b/app/Support/WorkflowTaskCompletionValidation.php
index 15067ca14..8373303f1 100644
--- a/app/Support/WorkflowTaskCompletionValidation.php
+++ b/app/Support/WorkflowTaskCompletionValidation.php
@@ -115,7 +115,7 @@ public static function rules(): array
             'commands.*.idempotency_key' => ['nullable', 'string', 'max:191'],
             'commands.*.mode_override' => ['nullable', 'string', 'in:sync,async'],
             'commands.*.wait_for' => ['nullable', 'string', 'in:accepted,completed'],
-            'commands.*.wait_timeout_seconds' => ['nullable', 'integer', 'min:0'],
+            'commands.*.wait_timeout_seconds' => ['nullable', 'integer', 'min:0', 'max:'.\Workflow\V2\Support\UpdateWaitPolicy::MAX_COMPLETION_TIMEOUT_SECONDS],
             'commands.*.target_workflow_instance_id' => ['nullable', 'string', 'max:191'],
             'commands.*.target_workflow_run_id' => ['nullable', 'string', 'max:191'],
             'commands.*.business_key' => ['nullable', 'string', 'max:191'],
diff --git a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php
index d4d7a3197..c5d6e8cdc 100644
--- a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php
+++ b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php
@@ -50,12 +50,15 @@ public function test_service_operation_request_survives_worker_validation(): voi
                     'request_payload' => ['codec' => 'avro', 'blob' => 'wwHioz3/VYAiNwwCChRpbnZvaWNlLTQyAA=='],
                 ]],
             ], $this->workerHeaders())->assertOk();
+            // Workflow 2.14 admits the call in the completion and starts its
+            // target when the scheduled effect runs after commit.
+            $this->assertDatabaseHas('workflow_history_events', [
+                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled',
+            ]);
+            $this->assertSame(1, $this->runScheduledServiceCalls());
             $call = WorkflowServiceCall::query()->sole();
             $this->assertSame('started', $call->status);
             $this->assertNotNull($call->linked_workflow_run_id);
-            $this->assertDatabaseHas('workflow_history_events', [
-                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted',
-            ]);
         });
     }
 }
diff --git a/tests/Feature/Concerns/ServerTestHelpers.php b/tests/Feature/Concerns/ServerTestHelpers.php
index c4d7115ba..d188c7aa5 100644
--- a/tests/Feature/Concerns/ServerTestHelpers.php
+++ b/tests/Feature/Concerns/ServerTestHelpers.php
@@ -8,7 +8,9 @@
 use App\Models\WorkflowNamespace;
 use App\Support\WorkerProtocol;
 use App\Support\WorkflowMetadataCapabilityPolicy;
+use Workflow\V2\Jobs\RunServiceCall;
 use Workflow\V2\Jobs\RunWorkflowTask;
+use Workflow\V2\Models\WorkflowServiceCallDispatch;
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Support\WorkflowExecutor;
 
@@ -112,4 +114,22 @@ protected function runReadyWorkflowTask(string $runId): void
         $job = new RunWorkflowTask($taskId);
         $job->handle(app(WorkflowExecutor::class));
     }
+
+    /**
+     * Runs the service calls committed completions scheduled, as the queue
+     * worker runs their RunServiceCall jobs: Workflow 2.14 dispatches a
+     * workflow-authored call's effect after its completion commits.
+     */
+    protected function runScheduledServiceCalls(): int
+    {
+        $callIds = WorkflowServiceCallDispatch::query()
+            ->where('state', 'scheduled')
+            ->orderBy('available_at')
+            ->pluck('service_call_id');
+        foreach ($callIds as $callId) {
+            $this->app->call([new RunServiceCall((string) $callId), 'handle']);
+        }
+
+        return $callIds->count();
+    }
 }
diff --git a/tests/Feature/ServiceOperationUpdateWaitTest.php b/tests/Feature/ServiceOperationUpdateWaitTest.php
index 3d2a3f99b..3f4a055a4 100644
--- a/tests/Feature/ServiceOperationUpdateWaitTest.php
+++ b/tests/Feature/ServiceOperationUpdateWaitTest.php
@@ -87,11 +87,15 @@ public function test_the_workflow_task_command_bounds_an_update_bound_operations
 
         $this->complete($task, 60)->assertOk();
         $this->assertSame(1, WorkflowServiceCall::query()->count());
+        // The update runs when the scheduled effect does, after commit.
+        $this->assertSame(1, $this->runScheduledServiceCalls());
         $this->assertSame(1, $this->targetUpdates());
     }
 
-    public function test_other_bindings_keep_the_service_operation_wait_contract(): void
+    public function test_every_binding_takes_the_service_operation_wait_ceiling(): void
     {
+        // Workflow 2.14 bounds every service-operation wait at 60 seconds, so
+        // a start binding is refused above it before anything is admitted.
         $this->seedOperation('start_workflow');
 
         $this->withHeaders($this->apiHeaders())
@@ -99,17 +103,21 @@ public function test_other_bindings_keep_the_service_operation_wait_contract():
                 'arguments' => Avro::envelope(['approved' => true, 'source' => 'service']),
                 'mode_override' => 'async',
                 'wait_for' => 'accepted',
-                'wait_timeout_seconds' => 600,
+                'wait_timeout_seconds' => 61,
             ])
-            ->assertSuccessful();
-        $this->assertSame(1, WorkflowServiceCall::query()->count());
+            ->assertStatus(422)
+            ->assertJsonPath('errors.wait_timeout_seconds.0', 'The wait timeout seconds field must not be greater than 60.');
+        $this->assertSame(0, WorkflowServiceCall::query()->count());
 
         $this->withHeaders($this->apiHeaders())
             ->postJson('/api/service-endpoints/billing/services/approvals/operations/approve/execute', [
-                'wait_timeout_seconds' => 86401,
+                'arguments' => Avro::envelope(['approved' => true, 'source' => 'service']),
+                'mode_override' => 'async',
+                'wait_for' => 'accepted',
+                'wait_timeout_seconds' => 60,
             ])
-            ->assertStatus(422)
-            ->assertJsonPath('errors.wait_timeout_seconds.0', 'The wait timeout seconds field must not be greater than 86400.');
+            ->assertSuccessful();
+        $this->assertSame(1, WorkflowServiceCall::query()->count());
     }
 
     public function test_a_binding_changed_after_the_check_still_cannot_carry_a_longer_wait(): void
diff --git a/tests/Feature/WorkerServiceOperationTest.php b/tests/Feature/WorkerServiceOperationTest.php
index ebae9a8fe..8d3b1a84c 100644
--- a/tests/Feature/WorkerServiceOperationTest.php
+++ b/tests/Feature/WorkerServiceOperationTest.php
@@ -44,7 +44,16 @@ public function test_worker_command_dispatches_a_service_operation_and_persists_
 
         $this->complete($task, $this->command())->assertOk();
 
+        // Workflow 2.14 admits the call in the completion and runs its effect
+        // after commit: the call is accepted and scheduled, with no target yet.
         $call = WorkflowServiceCall::query()->sole();
+        $this->assertSame('accepted', $call->status);
+        $this->assertNull($call->linked_workflow_run_id);
+        $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled']);
+        $this->assertDatabaseMissing('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']);
+        $this->assertSame(1, $this->runScheduledServiceCalls());
+
+        $call->refresh();
         $this->assertSame('started', $call->status);
         $this->assertSame('billing', $call->endpoint_name);
         $this->assertSame('invoicing', $call->service_name);
@@ -57,7 +66,10 @@ public function test_worker_command_dispatches_a_service_operation_and_persists_
         $target = WorkflowRun::query()->findOrFail($call->linked_workflow_run_id);
         $this->assertSame([['invoice' => 42]], Avro::unserialize($target->arguments));
         $this->assertDatabaseHas('workflow_tasks', ['workflow_run_id' => $target->id, 'status' => 'ready']);
-        $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']);
+        // The caller's admission event stays ServiceOperationScheduled; its
+        // terminal outcome is delivered later, and no ServiceCallStarted is
+        // synthesized for it.
+        $this->assertDatabaseMissing('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']);
         $this->assertSame('completed', WorkflowTask::query()->findOrFail($task['task_id'])->status->value);
 
         $this->registerWorker('invoice-worker', 'invoices', supportedWorkflowTypes: ['tests.nexus-target']);
@@ -150,12 +162,13 @@ public function test_runtime_payload_reference_is_resolved_before_service_comman
                 return;
             }
             $response->assertOk();
+            $this->assertSame(1, $this->runScheduledServiceCalls());
             $call = WorkflowServiceCall::query()->sole();
             $target = WorkflowRun::query()->findOrFail($call->linked_workflow_run_id);
             $this->assertSame([['invoice' => 42]], Avro::unserialize($target->arguments));
             $this->assertNotNull(RuntimeExternalPayload::query()->findOrFail($reference['reference_id'])->retained_at);
             $this->assertDatabaseHas('workflow_history_events', [
-                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted',
+                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled',
             ]);
         } finally {
             File::deleteDirectory($directory);
@@ -227,6 +240,7 @@ public function test_an_unscoped_worker_can_use_an_authorized_cross_namespace_op
         $this->authenticateWorker(null);
 
         $this->complete($task, $this->command(['namespace' => 'other-tenant']))->assertOk();
+        $this->assertSame(1, $this->runScheduledServiceCalls());
         $call = WorkflowServiceCall::query()->sole();
         $this->assertSame('other-tenant', $call->target_namespace);
         $this->assertSame('default', $call->caller_namespace);
@@ -265,6 +279,7 @@ public function test_reusing_an_owned_service_call_does_not_dispatch_a_second_ta
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
+        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $task = $this->pollWorkflow();
 
@@ -281,6 +296,7 @@ public function test_service_call_replays_cannot_reuse_another_caller_namespace(
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
+        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $call->update(['caller_namespace' => 'other-tenant']);
         $task = $this->pollWorkflow();
@@ -300,6 +316,7 @@ public function test_service_call_replays_are_authorized_against_current_catalog
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
+        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $operation->update(['boundary_policy' => ['required_roles' => ['admin']]]);
         $task = $this->pollWorkflow();
@@ -318,6 +335,7 @@ public function test_service_call_replay_requires_the_exact_caller_instance(stri
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
+        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $call->update(['caller_workflow_instance_id' => 'another-instance']);
         $task = $this->pollWorkflow();
@@ -345,6 +363,7 @@ public function test_service_call_replays_cannot_reuse_another_workflow_run(stri
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
+        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $call->update(['caller_workflow_run_id' => $call->linked_workflow_run_id]);
         $task = $this->pollWorkflow();

From 20746ea395dbcdafe4a717e110981fb135792c44 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 18:50:02 +0000
Subject: [PATCH 05/13] Adopt published Workflow 2.14.2 with worker protocol
 context

---
 app/Http/Controllers/Api/WorkerController.php |  9 +++-
 app/Support/WorkflowPackageApiFloor.php       |  2 +
 .../WorkflowTaskCompletionValidation.php      |  3 +-
 composer.json                                 |  2 +-
 composer.lock                                 | 16 +++----
 .../WorkerServiceOperationEnvelopeTest.php    |  9 ++--
 tests/Feature/Concerns/ServerTestHelpers.php  | 20 --------
 .../ServiceOperationUpdateWaitTest.php        |  2 -
 tests/Feature/WorkerServiceOperationTest.php  | 47 ++++++++++---------
 .../Support/WorkflowPackageApiFloorTest.php   |  2 +-
 10 files changed, 51 insertions(+), 61 deletions(-)

diff --git a/app/Http/Controllers/Api/WorkerController.php b/app/Http/Controllers/Api/WorkerController.php
index c19c53c01..1833314cc 100644
--- a/app/Http/Controllers/Api/WorkerController.php
+++ b/app/Http/Controllers/Api/WorkerController.php
@@ -68,6 +68,7 @@
 use Workflow\V2\Contracts\HistoryProjectionRole;
 use Workflow\V2\Contracts\PreparedLocalActivityGroupTaskBridge;
 use Workflow\V2\Contracts\PreparedLocalActivityTaskBridge;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
 use Workflow\V2\Enums\ActivityAttemptStatus;
 use Workflow\V2\Enums\HistoryEventType;
@@ -1860,6 +1861,12 @@ private function mutateWorkflowTaskCommands(Request $request, string $taskId, bo
         $commands = $this->promoteWorkflowFailureExceptionPayload($commands);
         /** @var WorkflowTaskBridge $bridge */
         $bridge = app(WorkflowTaskBridge::class);
+        if (! $bridge instanceof ProtocolAwareWorkflowTaskBridge) {
+            return WorkerProtocol::json([
+                'error' => 'workflow_bridge_protocol_unavailable',
+                'message' => 'The workflow task bridge must support authenticated worker protocol completion.',
+            ], 503);
+        }
         $workflowTaskQueue = WorkflowTask::query()
             ->whereKey($taskId)
             ->value('queue');
@@ -2007,7 +2014,7 @@ function () use (
                             $outcome['completed'] = false;
                         } else {
                             $outcome = WorkflowObservability::completion($request, $claimedTask,
-                                fn (): array => $bridge->complete($taskId, $commands));
+                                fn (): array => $bridge->completeForProtocol($taskId, $commands, WorkerProtocol::requestVersion($request)));
                         }
                         if (($outcome[$checkpoint ? 'checkpointed' : 'completed'] ?? false) === true) {
                             app(WorkflowStreamCommandProcessor::class)->process($taskId, (string) $namespace, $streamCommands);
diff --git a/app/Support/WorkflowPackageApiFloor.php b/app/Support/WorkflowPackageApiFloor.php
index 488d7b2be..0d08f43af 100644
--- a/app/Support/WorkflowPackageApiFloor.php
+++ b/app/Support/WorkflowPackageApiFloor.php
@@ -11,6 +11,7 @@
 use Workflow\V2\Contracts\ExternalPayloadStorageDriver;
 use Workflow\V2\Contracts\ExternalPayloadStoragePolicy;
 use Workflow\V2\Contracts\MatchingRole;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\ServiceControlPlane;
 use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
@@ -194,6 +195,7 @@ final class WorkflowPackageApiFloor
      * Interface methods the server type-hints directly.
      */
     private const REQUIRED_INTERFACE_APIS = [
+        [ProtocolAwareWorkflowTaskBridge::class, 'completeForProtocol'],
         [WorkflowServiceOperationAuthorizer::class, 'authorize'],
         [MatchingRole::class, 'wake'],
         [MatchingRole::class, 'runPass'],
diff --git a/app/Support/WorkflowTaskCompletionValidation.php b/app/Support/WorkflowTaskCompletionValidation.php
index 8373303f1..9f400a1be 100644
--- a/app/Support/WorkflowTaskCompletionValidation.php
+++ b/app/Support/WorkflowTaskCompletionValidation.php
@@ -4,6 +4,7 @@
 
 namespace App\Support;
 
+use Workflow\V2\Support\UpdateWaitPolicy;
 use Workflow\V2\Support\WorkflowCommandNormalizer;
 
 /** Transport rules retain the package command grammar before semantic normalization. */
@@ -115,7 +116,7 @@ public static function rules(): array
             'commands.*.idempotency_key' => ['nullable', 'string', 'max:191'],
             'commands.*.mode_override' => ['nullable', 'string', 'in:sync,async'],
             'commands.*.wait_for' => ['nullable', 'string', 'in:accepted,completed'],
-            'commands.*.wait_timeout_seconds' => ['nullable', 'integer', 'min:0', 'max:'.\Workflow\V2\Support\UpdateWaitPolicy::MAX_COMPLETION_TIMEOUT_SECONDS],
+            'commands.*.wait_timeout_seconds' => ['nullable', 'integer', 'min:0', 'max:'.UpdateWaitPolicy::MAX_COMPLETION_TIMEOUT_SECONDS],
             'commands.*.target_workflow_instance_id' => ['nullable', 'string', 'max:191'],
             'commands.*.target_workflow_run_id' => ['nullable', 'string', 'max:191'],
             'commands.*.business_key' => ['nullable', 'string', 'max:191'],
diff --git a/composer.json b/composer.json
index 01265f8d8..4307a706e 100644
--- a/composer.json
+++ b/composer.json
@@ -6,7 +6,7 @@
     "require": {
         "php": "^8.2",
         "apache/avro": "^1.12",
-        "durable-workflow/workflow": "2.14.1",
+        "durable-workflow/workflow": "2.14.2",
         "laravel/framework": "^13.30",
         "laravel/tinker": "^3.0",
         "league/flysystem-aws-s3-v3": "^3.35.3"
diff --git a/composer.lock b/composer.lock
index 30ceaee8c..ad4ba9318 100644
--- a/composer.lock
+++ b/composer.lock
@@ -4,7 +4,7 @@
         "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
         "This file is @generated automatically"
     ],
-    "content-hash": "bf409453a4d519b4f3745e9fe27c3e0f",
+    "content-hash": "0e7c09e84880eb2236a4b2067e06605a",
     "packages": [
         {
             "name": "apache/avro",
@@ -655,16 +655,16 @@
         },
         {
             "name": "durable-workflow/workflow",
-            "version": "2.14.1",
+            "version": "2.14.2",
             "source": {
                 "type": "git",
                 "url": "https://github.com/durable-workflow/workflow.git",
-                "reference": "af9dbadb41bf96f51ad78047022c392887e3c7a6"
+                "reference": "d60ce5f60e70b834c5d4ef142c3d5dc415cd39fe"
             },
             "dist": {
                 "type": "zip",
-                "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/af9dbadb41bf96f51ad78047022c392887e3c7a6",
-                "reference": "af9dbadb41bf96f51ad78047022c392887e3c7a6",
+                "url": "https://api.github.com/repos/durable-workflow/workflow/zipball/d60ce5f60e70b834c5d4ef142c3d5dc415cd39fe",
+                "reference": "d60ce5f60e70b834c5d4ef142c3d5dc415cd39fe",
                 "shasum": ""
             },
             "require": {
@@ -705,7 +705,7 @@
                     "dev-main": "2.0.x-dev"
                 },
                 "durable-workflow": {
-                    "product-train": "2.14.1",
+                    "product-train": "2.14.2",
                     "laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
                     "laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
                 }
@@ -732,9 +732,9 @@
             "description": "Embedded durable workflow runtime and orchestration engine for Laravel applications.",
             "support": {
                 "issues": "https://github.com/durable-workflow/workflow/issues",
-                "source": "https://github.com/durable-workflow/workflow/tree/2.14.1"
+                "source": "https://github.com/durable-workflow/workflow/tree/2.14.2"
             },
-            "time": "2026-10-26T12:23:08+00:00"
+            "time": "2026-10-26T16:41:36+00:00"
         },
         {
             "name": "egulias/email-validator",
diff --git a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php
index c5d6e8cdc..d4d7a3197 100644
--- a/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php
+++ b/tests/Feature/CodecRegression/WorkerServiceOperationEnvelopeTest.php
@@ -50,15 +50,12 @@ public function test_service_operation_request_survives_worker_validation(): voi
                     'request_payload' => ['codec' => 'avro', 'blob' => 'wwHioz3/VYAiNwwCChRpbnZvaWNlLTQyAA=='],
                 ]],
             ], $this->workerHeaders())->assertOk();
-            // Workflow 2.14 admits the call in the completion and starts its
-            // target when the scheduled effect runs after commit.
-            $this->assertDatabaseHas('workflow_history_events', [
-                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled',
-            ]);
-            $this->assertSame(1, $this->runScheduledServiceCalls());
             $call = WorkflowServiceCall::query()->sole();
             $this->assertSame('started', $call->status);
             $this->assertNotNull($call->linked_workflow_run_id);
+            $this->assertDatabaseHas('workflow_history_events', [
+                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted',
+            ]);
         });
     }
 }
diff --git a/tests/Feature/Concerns/ServerTestHelpers.php b/tests/Feature/Concerns/ServerTestHelpers.php
index d188c7aa5..c4d7115ba 100644
--- a/tests/Feature/Concerns/ServerTestHelpers.php
+++ b/tests/Feature/Concerns/ServerTestHelpers.php
@@ -8,9 +8,7 @@
 use App\Models\WorkflowNamespace;
 use App\Support\WorkerProtocol;
 use App\Support\WorkflowMetadataCapabilityPolicy;
-use Workflow\V2\Jobs\RunServiceCall;
 use Workflow\V2\Jobs\RunWorkflowTask;
-use Workflow\V2\Models\WorkflowServiceCallDispatch;
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Support\WorkflowExecutor;
 
@@ -114,22 +112,4 @@ protected function runReadyWorkflowTask(string $runId): void
         $job = new RunWorkflowTask($taskId);
         $job->handle(app(WorkflowExecutor::class));
     }
-
-    /**
-     * Runs the service calls committed completions scheduled, as the queue
-     * worker runs their RunServiceCall jobs: Workflow 2.14 dispatches a
-     * workflow-authored call's effect after its completion commits.
-     */
-    protected function runScheduledServiceCalls(): int
-    {
-        $callIds = WorkflowServiceCallDispatch::query()
-            ->where('state', 'scheduled')
-            ->orderBy('available_at')
-            ->pluck('service_call_id');
-        foreach ($callIds as $callId) {
-            $this->app->call([new RunServiceCall((string) $callId), 'handle']);
-        }
-
-        return $callIds->count();
-    }
 }
diff --git a/tests/Feature/ServiceOperationUpdateWaitTest.php b/tests/Feature/ServiceOperationUpdateWaitTest.php
index 3f4a055a4..2c87d964f 100644
--- a/tests/Feature/ServiceOperationUpdateWaitTest.php
+++ b/tests/Feature/ServiceOperationUpdateWaitTest.php
@@ -87,8 +87,6 @@ public function test_the_workflow_task_command_bounds_an_update_bound_operations
 
         $this->complete($task, 60)->assertOk();
         $this->assertSame(1, WorkflowServiceCall::query()->count());
-        // The update runs when the scheduled effect does, after commit.
-        $this->assertSame(1, $this->runScheduledServiceCalls());
         $this->assertSame(1, $this->targetUpdates());
     }
 
diff --git a/tests/Feature/WorkerServiceOperationTest.php b/tests/Feature/WorkerServiceOperationTest.php
index 8d3b1a84c..19e9126b6 100644
--- a/tests/Feature/WorkerServiceOperationTest.php
+++ b/tests/Feature/WorkerServiceOperationTest.php
@@ -10,11 +10,13 @@
 use Illuminate\Support\Facades\File;
 use Illuminate\Support\Facades\Queue;
 use Illuminate\Testing\TestResponse;
+use Mockery;
 use PHPUnit\Framework\Attributes\DataProvider;
 use Tests\Feature\Concerns\ServerTestHelpers;
 use Tests\TestCase;
 use Workflow\Serializers\Avro;
 use Workflow\V2\Contracts\ServiceControlPlane;
+use Workflow\V2\Contracts\WorkflowTaskBridge;
 use Workflow\V2\Models\WorkflowHistoryEvent;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowService;
@@ -44,16 +46,7 @@ public function test_worker_command_dispatches_a_service_operation_and_persists_
 
         $this->complete($task, $this->command())->assertOk();
 
-        // Workflow 2.14 admits the call in the completion and runs its effect
-        // after commit: the call is accepted and scheduled, with no target yet.
         $call = WorkflowServiceCall::query()->sole();
-        $this->assertSame('accepted', $call->status);
-        $this->assertNull($call->linked_workflow_run_id);
-        $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled']);
-        $this->assertDatabaseMissing('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']);
-        $this->assertSame(1, $this->runScheduledServiceCalls());
-
-        $call->refresh();
         $this->assertSame('started', $call->status);
         $this->assertSame('billing', $call->endpoint_name);
         $this->assertSame('invoicing', $call->service_name);
@@ -66,10 +59,8 @@ public function test_worker_command_dispatches_a_service_operation_and_persists_
         $target = WorkflowRun::query()->findOrFail($call->linked_workflow_run_id);
         $this->assertSame([['invoice' => 42]], Avro::unserialize($target->arguments));
         $this->assertDatabaseHas('workflow_tasks', ['workflow_run_id' => $target->id, 'status' => 'ready']);
-        // The caller's admission event stays ServiceOperationScheduled; its
-        // terminal outcome is delivered later, and no ServiceCallStarted is
-        // synthesized for it.
-        $this->assertDatabaseMissing('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']);
+        $this->assertDatabaseHas('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted']);
+        $this->assertDatabaseMissing('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled']);
         $this->assertSame('completed', WorkflowTask::query()->findOrFail($task['task_id'])->status->value);
 
         $this->registerWorker('invoice-worker', 'invoices', supportedWorkflowTypes: ['tests.nexus-target']);
@@ -84,6 +75,27 @@ public function test_worker_command_dispatches_a_service_operation_and_persists_
         $this->assertSame(['invoice' => 42, 'accepted' => true], Avro::unserialize($target->fresh()->output));
     }
 
+    public function test_a_bridge_without_protocol_context_cannot_complete_the_task(): void
+    {
+        $this->seedCatalog();
+        $task = $this->leaseWorkflow();
+        $this->authenticateWorker();
+        $status = app(WorkflowTaskBridge::class)->status($task['task_id']);
+        $legacyBridge = Mockery::mock(WorkflowTaskBridge::class);
+        $legacyBridge->shouldReceive('status')->with($task['task_id'])->andReturn($status);
+        $legacyBridge->shouldNotReceive('complete');
+        $this->app->instance(WorkflowTaskBridge::class, $legacyBridge);
+
+        $this->complete($task, $this->command())->assertStatus(503)
+            ->assertJsonPath('error', 'workflow_bridge_protocol_unavailable');
+
+        $this->assertSame('leased', WorkflowTask::query()->findOrFail($task['task_id'])->status->value);
+        $this->assertDatabaseCount('workflow_service_calls', 0);
+        $this->assertDatabaseMissing('workflow_history_events', [
+            'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted',
+        ]);
+    }
+
     public function test_supported_options_reach_the_service_control_plane(): void
     {
         $task = $this->leaseWorkflow();
@@ -162,13 +174,12 @@ public function test_runtime_payload_reference_is_resolved_before_service_comman
                 return;
             }
             $response->assertOk();
-            $this->assertSame(1, $this->runScheduledServiceCalls());
             $call = WorkflowServiceCall::query()->sole();
             $target = WorkflowRun::query()->findOrFail($call->linked_workflow_run_id);
             $this->assertSame([['invoice' => 42]], Avro::unserialize($target->arguments));
             $this->assertNotNull(RuntimeExternalPayload::query()->findOrFail($reference['reference_id'])->retained_at);
             $this->assertDatabaseHas('workflow_history_events', [
-                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceOperationScheduled',
+                'workflow_run_id' => $task['run_id'], 'event_type' => 'ServiceCallStarted',
             ]);
         } finally {
             File::deleteDirectory($directory);
@@ -240,7 +251,6 @@ public function test_an_unscoped_worker_can_use_an_authorized_cross_namespace_op
         $this->authenticateWorker(null);
 
         $this->complete($task, $this->command(['namespace' => 'other-tenant']))->assertOk();
-        $this->assertSame(1, $this->runScheduledServiceCalls());
         $call = WorkflowServiceCall::query()->sole();
         $this->assertSame('other-tenant', $call->target_namespace);
         $this->assertSame('default', $call->caller_namespace);
@@ -279,7 +289,6 @@ public function test_reusing_an_owned_service_call_does_not_dispatch_a_second_ta
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
-        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $task = $this->pollWorkflow();
 
@@ -296,7 +305,6 @@ public function test_service_call_replays_cannot_reuse_another_caller_namespace(
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
-        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $call->update(['caller_namespace' => 'other-tenant']);
         $task = $this->pollWorkflow();
@@ -316,7 +324,6 @@ public function test_service_call_replays_are_authorized_against_current_catalog
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
-        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $operation->update(['boundary_policy' => ['required_roles' => ['admin']]]);
         $task = $this->pollWorkflow();
@@ -335,7 +342,6 @@ public function test_service_call_replay_requires_the_exact_caller_instance(stri
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
-        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $call->update(['caller_workflow_instance_id' => 'another-instance']);
         $task = $this->pollWorkflow();
@@ -363,7 +369,6 @@ public function test_service_call_replays_cannot_reuse_another_workflow_run(stri
         $task = $this->leaseWorkflow();
         $this->authenticateWorker();
         $this->complete($task, $this->command(['idempotency_key' => 'invoice-42']))->assertOk();
-        $this->runScheduledServiceCalls();
         $call = WorkflowServiceCall::query()->sole();
         $call->update(['caller_workflow_run_id' => $call->linked_workflow_run_id]);
         $task = $this->pollWorkflow();
diff --git a/tests/Unit/Support/WorkflowPackageApiFloorTest.php b/tests/Unit/Support/WorkflowPackageApiFloorTest.php
index 70f962e6b..ce2be8bbf 100644
--- a/tests/Unit/Support/WorkflowPackageApiFloorTest.php
+++ b/tests/Unit/Support/WorkflowPackageApiFloorTest.php
@@ -199,7 +199,7 @@ public function test_worker_query_task_protocol_contract_is_public_static(): voi
         $this->assertNotFalse($capability);
         $this->assertTrue($capability->isPublic());
 
-        $this->assertSame(WorkerProtocolVersion::VERSION, WorkerProtocol::VERSION);
+        $this->assertTrue(WorkerProtocolVersion::isSupportedRequestVersion(WorkerProtocol::VERSION));
         $this->assertSame('query_tasks', WorkerProtocolVersion::CAPABILITY_QUERY_TASKS);
         $this->assertSame(['poll', 'complete', 'fail'], WorkerProtocolVersion::queryTaskVerbs());
         $this->assertContains(WorkerProtocolVersion::CAPABILITY_QUERY_TASKS, WorkerProtocolVersion::workerCapabilities());

From e3595d8928e8cceb619df59d59992887959109c0 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 19:05:28 +0000
Subject: [PATCH 06/13] Compare committed Nexus envelope fields across database
 engines

---
 tests/Feature/ServiceCatalogReadControllerTest.php | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/tests/Feature/ServiceCatalogReadControllerTest.php b/tests/Feature/ServiceCatalogReadControllerTest.php
index 25b5170b6..05b1bb0e9 100644
--- a/tests/Feature/ServiceCatalogReadControllerTest.php
+++ b/tests/Feature/ServiceCatalogReadControllerTest.php
@@ -287,7 +287,9 @@ public function test_committed_inline_service_results_are_inspectable_only_by_th
                 'metadata' => ['nexus_result' => $envelope, 'retained_diagnostic' => 'original'],
             ]);
             $this->withHeaders($this->apiHeaders('target'))->getJson('/api/service-catalog/calls/'.$call->id)
-                ->assertOk()->assertJsonPath('result_envelope', $envelope)
+                ->assertOk()->assertJsonCount(2, 'result_envelope')
+                ->assertJsonPath('result_envelope.codec', $envelope['codec'])
+                ->assertJsonPath('result_envelope.blob', $envelope['blob'])
                 ->assertJsonPath('payloads_in_observer_namespace', true);
             $foreign = $this->withHeaders($this->apiHeaders('caller'))->getJson('/api/service-catalog/calls/'.$call->id)
                 ->assertOk()->assertJsonPath('result_envelope', null)

From b7f906d103f079117074232ffc63ce83a835339e Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 19:37:58 +0000
Subject: [PATCH 07/13] Avoid missing-stream gap locks during concurrent opens

---
 app/Support/WorkflowStreamService.php        |  59 ++++++----
 tests/Feature/WorkflowStreamOpenRaceTest.php | 114 +++++++++++++++++++
 2 files changed, 154 insertions(+), 19 deletions(-)
 create mode 100644 tests/Feature/WorkflowStreamOpenRaceTest.php

diff --git a/app/Support/WorkflowStreamService.php b/app/Support/WorkflowStreamService.php
index 07b77b50d..005612bc7 100644
--- a/app/Support/WorkflowStreamService.php
+++ b/app/Support/WorkflowStreamService.php
@@ -5,6 +5,7 @@
 use App\Models\WorkflowDurableStream;
 use App\Models\WorkflowDurableStreamItem;
 use Illuminate\Database\Eloquent\Collection;
+use Illuminate\Database\UniqueConstraintViolationException;
 use Illuminate\Support\Carbon;
 use Illuminate\Support\Facades\DB;
 use InvalidArgumentException;
@@ -431,30 +432,50 @@ private function lockStream(
             throw new StreamNotFoundException($run, $streamName);
         }
 
-        $stream = WorkflowDurableStream::query()
+        $query = WorkflowDurableStream::query()
             ->where('namespace', $namespace)
             ->where('workflow_run_id', $run->id)
-            ->where('stream_name', $streamName)
-            ->lockForUpdate()
-            ->first();
+            ->where('stream_name', $streamName);
+
+        if (! $openIfMissing) {
+            return $query->lockForUpdate()->first();
+        }
 
-        if ($stream !== null || ! $openIfMissing) {
-            return $stream;
+        // A missing-key locking read takes an InnoDB gap lock. Two openers
+        // on different runs can share that gap, then deadlock on insertion.
+        // The run lock already serializes same-run writers: discover without
+        // locking, then take a current record lock only on a row we found.
+        $streamId = (clone $query)->value('id');
+        if ($streamId !== null) {
+            return WorkflowDurableStream::query()->whereKey($streamId)->lockForUpdate()->firstOrFail();
         }
 
-        // The run lock admits one opener at a time, so the stream is created
-        // through its model and the namespace stream quota applies.
-        return WorkflowDurableStream::query()->create([
-            'namespace' => $namespace,
-            'workflow_instance_id' => $run->workflow_instance_id,
-            'workflow_run_id' => $run->id,
-            'stream_name' => $streamName,
-            'status' => WorkflowDurableStream::STATUS_OPEN,
-            'last_offset' => -1,
-            'total_items' => 0,
-            'pending_items' => 0,
-            'opened_at' => Carbon::now(),
-        ]);
+        try {
+            // Keep model quota admission. A savepoint rolls a failed insert
+            // back without aborting a surrounding command transaction.
+            return DB::transaction(fn () => WorkflowDurableStream::query()->create([
+                'namespace' => $namespace,
+                'workflow_instance_id' => $run->workflow_instance_id,
+                'workflow_run_id' => $run->id,
+                'stream_name' => $streamName,
+                'status' => WorkflowDurableStream::STATUS_OPEN,
+                'last_offset' => -1,
+                'total_items' => 0,
+                'pending_items' => 0,
+                'opened_at' => Carbon::now(),
+            ]));
+        } catch (UniqueConstraintViolationException|NamespaceDurableStateException $exception) {
+            if ($exception instanceof NamespaceDurableStateException
+                && $exception->reason !== 'namespace_workflow_streams_exhausted') {
+                throw $exception;
+            }
+
+            // An outer command may hold a repeatable snapshot older than the
+            // run lock. Its stream can already exist even if discovery missed
+            // it. Reuse that committed row; it needs no new quota slot, even
+            // when the namespace is now full or over its configured limit.
+            return $query->lockForUpdate()->first() ?? throw $exception;
+        }
     }
 
     private function assertValidStreamName(string $streamName): void
diff --git a/tests/Feature/WorkflowStreamOpenRaceTest.php b/tests/Feature/WorkflowStreamOpenRaceTest.php
new file mode 100644
index 000000000..b4b8e0955
--- /dev/null
+++ b/tests/Feature/WorkflowStreamOpenRaceTest.php
@@ -0,0 +1,114 @@
+<?php
+
+namespace Tests\Feature;
+
+use App\Models\WorkflowDurableStream;
+use App\Models\WorkflowDurableStreamItem;
+use App\Support\WorkflowStreamService;
+use Illuminate\Support\Facades\DB;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Tests\Feature\Concerns\ForkedDatabaseActors;
+use Tests\TestCase;
+use Workflow\V2\Models\WorkflowRun;
+
+class WorkflowStreamOpenRaceTest extends TestCase
+{
+    use ForkedDatabaseActors;
+
+    public static function innodbDrivers(): array
+    {
+        return ['mysql' => ['mysql'], 'mariadb' => ['mariadb']];
+    }
+
+    #[DataProvider('innodbDrivers')]
+    public function test_first_appends_to_different_runs_both_commit_after_both_missing_reads(string $driver): void
+    {
+        $this->initializeActorDatabase($driver);
+        $firstRun = $this->createRun('wf-open-first');
+        $secondRun = $this->createRun('wf-open-second');
+
+        try {
+            $first = $this->startActor(fn (): int => $this->append($firstRun), self::missingRead(...));
+            $this->runUntilPaused($first);
+            $second = $this->startActor(fn (): int => $this->append($secondRun), self::missingRead(...));
+            $this->runUntilPaused($second);
+
+            // Both reads finish before either insert. Missing-key FOR UPDATE
+            // reads hold compatible gap locks; the two inserts then deadlock.
+            $this->resume($first);
+            $this->resume($second);
+            $this->assertSame(1, $this->finishActor($first));
+            $this->assertSame(1, $this->finishActor($second));
+        } finally {
+            DB::purge();
+            DB::reconnect();
+        }
+
+        $this->assertSame(2, WorkflowDurableStream::query()->count());
+        $this->assertSame(2, WorkflowDurableStreamItem::query()->count());
+    }
+
+    public static function staleSnapshots(): array
+    {
+        $cases = [];
+        foreach (['mysql', 'mariadb'] as $driver) {
+            foreach (['unlimited' => null, 'full' => 1, 'over-limit' => 0] as $name => $limit) {
+                $cases[$driver.' '.$name] = [$driver, $limit];
+            }
+        }
+
+        return $cases;
+    }
+
+    #[DataProvider('staleSnapshots')]
+    public function test_an_older_command_snapshot_uses_the_stream_committed_before_its_run_lock(string $driver, ?int $limit): void
+    {
+        $this->initializeActorDatabase($driver);
+        $runId = $this->createRun('wf-open-snapshot');
+        config(['server.namespace_durable_state.overrides' => [
+            'default' => ['max_workflow_streams' => $limit],
+        ]]);
+
+        try {
+            $older = $this->startActor(fn (): int => DB::transaction(function () use ($runId): int {
+                // A command transaction may have established its repeatable
+                // snapshot before it takes the stream's run lock.
+                $this->assertSame(0, WorkflowDurableStream::query()->count());
+
+                return $this->append($runId);
+            }), self::missingRead(...));
+            $this->runUntilPaused($older);
+            $opener = $this->startActor(function () use ($runId): int {
+                // A lower quota may be deployed while an existing stream is
+                // still writable. Its append must not require another slot.
+                config(['server.namespace_durable_state.overrides' => []]);
+
+                return $this->append($runId);
+            });
+            $this->release($opener);
+            $this->assertSame(1, $this->finishActor($opener));
+            $this->resume($older);
+            $this->assertSame(1, $this->finishActor($older));
+        } finally {
+            DB::purge();
+            DB::reconnect();
+        }
+
+        $this->assertSame(1, WorkflowDurableStream::query()->count());
+        $this->assertSame([0, 1], WorkflowDurableStreamItem::query()->orderBy('offset')->pluck('offset')->map(intval(...))->all());
+        $this->assertSame(2, (int) WorkflowDurableStream::query()->value('pending_items'));
+    }
+
+    private static function missingRead(string $sql, string $driver, int $level): bool
+    {
+        return str_starts_with($sql, 'select') && str_contains($sql, 'workflow_durable_streams')
+            && ! str_contains($sql, 'workflow_durable_stream_items');
+    }
+
+    private function append(string $runId): int
+    {
+        return app(WorkflowStreamService::class)->append(
+            WorkflowRun::query()->findOrFail($runId), 'default', 'tokens', [['payload' => 'item']],
+        )['accepted'];
+    }
+}

From ef9432af3bf8a48428b0bcb5f0b055c815201ab1 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 17:53:24 +0000
Subject: [PATCH 08/13] Reproduce: a workflow start's input claim does not hold
 its object until the run commits

A failing race test, not a fix. On MySQL, MariaDB, PostgreSQL and SQLite,
run retention deletes an external payload object between a workflow
start's committed claim and the transaction that writes the run, and the
start then commits a run whose arguments reference the deleted object.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../Feature/WorkflowStreamPayloadRaceTest.php | 50 +++++++++++++++++++
 1 file changed, 50 insertions(+)

diff --git a/tests/Feature/WorkflowStreamPayloadRaceTest.php b/tests/Feature/WorkflowStreamPayloadRaceTest.php
index bf80a2bac..fba4651a5 100644
--- a/tests/Feature/WorkflowStreamPayloadRaceTest.php
+++ b/tests/Feature/WorkflowStreamPayloadRaceTest.php
@@ -26,6 +26,7 @@
 use Workflow\V2\Enums\TaskType;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowTask;
+use Workflow\V2\Support\ExternalPayloads;
 use Workflow\V2\Support\RunSummaryProjector;
 
 /**
@@ -247,6 +248,55 @@ public function test_appends_listing_shared_objects_in_opposite_orders_both_comm
         $this->assertObjectKept($b);
     }
 
+    /**
+     * REPRODUCTION (not a fix): a workflow start claims its external input
+     * in its own committed step, before the transaction that writes the run
+     * referencing it. Run retention that runs between the two finds no
+     * retained reference and deletes the object, and the start then commits
+     * a run whose arguments reference a deleted object. Expected to fail
+     * until non-stream writers claim inside the transaction that records the
+     * reference.
+     */
+    #[DataProvider('drivers')]
+    public function test_a_workflow_start_and_run_pruning_decide_a_shared_object(string $driver): void
+    {
+        $this->initializeDatabase($driver);
+        $object = $this->registerObject('start-input');
+        $oldRunId = $this->summarizedRun('wf-claim-race-old', RunStatus::Completed);
+        $this->addStream($oldRunId, 'old', [$object['uri']]);
+
+        // The start has verified and claimed the object under its lock and
+        // committed that step; it has not yet written its run.
+        $start = $this->startActor(fn (): array => $this->httpStart('wf-claim-race-new', $object['reference']), self::afterVerification());
+        $this->runUntilPaused($start);
+        $pruned = HistoryRetentionEnforcer::pruneRun('default', $oldRunId);
+        $this->assertTrue($pruned['pruned'], json_encode($pruned));
+        $this->resume($start);
+        $started = $this->finishActor($start);
+
+        // Either the start is refused, or the object its run references
+        // still exists.
+        if ($started['status'] === 201) {
+            $arguments = (string) WorkflowRun::query()->where('workflow_instance_id', 'wf-claim-race-new')->value('arguments');
+            $this->assertSame($object['uri'], ExternalPayloads::storedEnvelope($arguments)['external_storage']['uri']);
+            $this->assertObjectKept($object);
+        } else {
+            $this->assertObjectDeleted($object);
+        }
+    }
+
+    private function httpStart(string $workflowId, array $reference): array
+    {
+        $response = $this->withHeaders($this->apiHeaders())->postJson('/api/workflows', [
+            'workflow_id' => $workflowId,
+            'workflow_type' => 'remote.claim-race',
+            'task_queue' => 'claim-race',
+            'input' => ['codec' => 'avro', 'external_payload' => $reference],
+        ]);
+
+        return ['status' => $response->status(), 'reason' => $response->json('reason')];
+    }
+
     private function initializeDatabase(string $driver): void
     {
         $this->initializeActorDatabase($driver);

From 0ac8206ccf5a3852c275a637af336b53284f5348 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 20:28:58 +0000
Subject: [PATCH 09/13] Keep runtime payload claims through the reference
 writer commit

---
 .../Api/MessageStreamController.php           |   2 +-
 .../Controllers/Api/ScheduleController.php    |   1 +
 .../RuntimeExternalPayloadTransport.php       |   3 +-
 app/Providers/AppServiceProvider.php          |  15 +++
 .../RuntimeExternalPayloadWriteClaims.php     | 115 +++++++++++++++++
 .../RuntimeTrackedExternalPayloadStorage.php  |  22 ++--
 tests/Feature/NexusPayloadTransportTest.php   |   6 +-
 .../RuntimeExternalPayloadTransportTest.php   |   8 +-
 .../RuntimeExternalPayloadWriteClaimsTest.php | 116 ++++++++++++++++++
 .../Feature/WorkflowStreamPayloadRaceTest.php |  75 ++++++++++-
 10 files changed, 346 insertions(+), 17 deletions(-)
 create mode 100644 app/Support/RuntimeExternalPayloadWriteClaims.php
 create mode 100644 tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php

diff --git a/app/Http/Controllers/Api/MessageStreamController.php b/app/Http/Controllers/Api/MessageStreamController.php
index 174561250..003ad47d1 100644
--- a/app/Http/Controllers/Api/MessageStreamController.php
+++ b/app/Http/Controllers/Api/MessageStreamController.php
@@ -62,7 +62,7 @@ public function append(Request $request, string $workflowId, string $streamName)
 
         $storage = $this->externalPayloadStorage->driverFor($namespace);
         try {
-            $envelope = AvroPayloadEnvelopeResolver::resolve($validated['input'] ?? null, 'input', $storage);
+            $envelope = AvroPayloadEnvelopeResolver::resolve($validated['input'] ?? null, 'input', $storage, retainExternal: true);
         } catch (ValidationException $exception) {
             $this->streams->recordMalformed(
                 $namespace,
diff --git a/app/Http/Controllers/Api/ScheduleController.php b/app/Http/Controllers/Api/ScheduleController.php
index 6680cd0a2..bbe7ce159 100644
--- a/app/Http/Controllers/Api/ScheduleController.php
+++ b/app/Http/Controllers/Api/ScheduleController.php
@@ -1021,6 +1021,7 @@ private function validateActionInput(?array $action, ?string $namespace): void
             $action['input'],
             'action.input',
             $this->externalPayloadStorage->driverFor($namespace),
+            retainExternal: true,
         );
     }
 
diff --git a/app/Http/Middleware/RuntimeExternalPayloadTransport.php b/app/Http/Middleware/RuntimeExternalPayloadTransport.php
index 20812ba13..b5da06ec2 100644
--- a/app/Http/Middleware/RuntimeExternalPayloadTransport.php
+++ b/app/Http/Middleware/RuntimeExternalPayloadTransport.php
@@ -6,6 +6,7 @@
 use App\Support\RuntimeExternalPayloadAudit;
 use App\Support\RuntimeExternalPayloadException;
 use App\Support\RuntimeExternalPayloadRegistry;
+use App\Support\RuntimeExternalPayloadWriteClaims;
 use App\Support\WorkerProtocol;
 use App\Support\WorkflowQueryTaskBroker;
 use Closure;
@@ -220,7 +221,7 @@ public function handle(Request $request, Closure $next): Response
         ));
 
         try {
-            $response = $next($request);
+            $response = app(RuntimeExternalPayloadWriteClaims::class)->transport(fn (): Response => $next($request));
 
             if ($request->attributes->getBoolean(self::ATTRIBUTE_CLAIMED) && $response->getStatusCode() < 400) {
                 $this->audit->record($request, 'external_payload.claimed');
diff --git a/app/Providers/AppServiceProvider.php b/app/Providers/AppServiceProvider.php
index 8a940502d..e28306f1d 100644
--- a/app/Providers/AppServiceProvider.php
+++ b/app/Providers/AppServiceProvider.php
@@ -24,6 +24,7 @@
 use App\Support\ExternalWorkflowUpdateAdmission;
 use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\RemoteScheduleStarter;
+use App\Support\RuntimeExternalPayloadWriteClaims;
 use App\Support\ServerPollingCache;
 use App\Support\ServerWorkflowControlPlane;
 use App\Support\ServiceCallBoundary;
@@ -40,6 +41,9 @@
 use Illuminate\Contracts\Bus\Dispatcher as BusDispatcher;
 use Illuminate\Contracts\Console\Kernel as ConsoleKernel;
 use Illuminate\Database\Connection;
+use Illuminate\Database\Events\TransactionCommitted;
+use Illuminate\Database\Events\TransactionCommitting;
+use Illuminate\Database\Events\TransactionRolledBack;
 use Illuminate\Queue\Events\Looping;
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\ServiceProvider;
@@ -66,6 +70,17 @@ class AppServiceProvider extends ServiceProvider
 {
     public function register(): void
     {
+        $this->app->scoped(RuntimeExternalPayloadWriteClaims::class);
+        $this->app->make('events')->listen(TransactionCommitting::class, static function (TransactionCommitting $event): void {
+            app(RuntimeExternalPayloadWriteClaims::class)->beforeCommit($event->connection);
+        });
+        $this->app->make('events')->listen(TransactionCommitted::class, static function (TransactionCommitted $event): void {
+            app(RuntimeExternalPayloadWriteClaims::class)->afterCommit($event->connection);
+        });
+        $this->app->make('events')->listen(TransactionRolledBack::class, static function (TransactionRolledBack $event): void {
+            app(RuntimeExternalPayloadWriteClaims::class)->afterRollback($event->connection);
+        });
+
         Connection::resolverFor('sqlite', static fn ($pdo, $database, $prefix, $config) => new GuardedSqliteConnection($pdo, $database, $prefix, $config));
         // Install before package boot callbacks or commands can open a database.
         foreach ([
diff --git a/app/Support/RuntimeExternalPayloadWriteClaims.php b/app/Support/RuntimeExternalPayloadWriteClaims.php
new file mode 100644
index 000000000..9bcd5a852
--- /dev/null
+++ b/app/Support/RuntimeExternalPayloadWriteClaims.php
@@ -0,0 +1,115 @@
+<?php
+
+namespace App\Support;
+
+use Closure;
+use Illuminate\Database\Connection;
+use Illuminate\Support\Facades\DB;
+use Throwable;
+
+/** Keep HTTP ingress's verified objects until its reference writer commits. */
+final class RuntimeExternalPayloadWriteClaims
+{
+    private bool $active = false;
+
+    private bool $committing = false;
+
+    private int $preparing = 0;
+
+    /** @var array<string, array<string, array<string, array{size_bytes: int, sha256: string, level: int}>>> */
+    private array $verified = [];
+
+    public function transport(Closure $operation): mixed
+    {
+        $previous = [$this->active, $this->verified];
+        $this->active = true;
+        $this->verified = [];
+        try {
+            return $operation();
+        } finally {
+            [$this->active, $this->verified] = $previous;
+        }
+    }
+
+    /** Storage preparation may commit metadata without writing a reference. */
+    public function prepare(Closure $operation): mixed
+    {
+        $this->preparing++;
+        try {
+            return $operation();
+        } finally {
+            $this->preparing--;
+        }
+    }
+
+    public function verified(string $namespace, string $uri, int $sizeBytes, string $sha256): void
+    {
+        if ($this->active) {
+            $connection = DB::connection();
+            $name = $connection->getName();
+            $this->verified[$name][$uri][$namespace] = [
+                'size_bytes' => $sizeBytes, 'sha256' => $sha256,
+                'level' => min($connection->transactionLevel(), $this->verified[$name][$uri][$namespace]['level'] ?? PHP_INT_MAX),
+            ];
+        }
+    }
+
+    public function beforeCommit(Connection $connection): void
+    {
+        $verified = $this->verified[$connection->getName()] ?? [];
+        if (! $this->active || $this->preparing > 0 || $this->committing || $verified === []) {
+            return;
+        }
+
+        // This runs before the actual outer commit, after the writer's run,
+        // task and attempt locks. All objects are claimed in one order, and
+        // those locks stay held until the references become visible. Storage
+        // preparation's own metadata commits are deliberately excluded.
+        $this->committing = true;
+        try {
+            $registry = app(RuntimeExternalPayloadRegistry::class);
+            foreach (RuntimeExternalPayloadObjectLock::inLockOrder(array_keys($verified)) as $uri) {
+                foreach ($verified[$uri] as $namespace => $bytes) {
+                    $registry->claimVerified($namespace, $uri, $bytes['size_bytes'], $bytes['sha256']);
+                }
+            }
+        } catch (Throwable $exception) {
+            // Laravel decrements its transaction level on a commit-listener
+            // exception but does not roll back a failed final commit attempt.
+            // Roll back here, while its transaction and callbacks still exist.
+            if ($connection->transactionLevel() > 0) {
+                $connection->rollBack();
+            }
+            throw $exception;
+        } finally {
+            $this->committing = false;
+        }
+    }
+
+    public function afterCommit(Connection $connection): void
+    {
+        $name = $connection->getName();
+        foreach ($this->verified[$name] ?? [] as $uri => $namespaces) {
+            foreach ($namespaces as $namespace => $bytes) {
+                // A released savepoint's registration now belongs to its
+                // parent. A later sibling rollback must not discard it.
+                $this->verified[$name][$uri][$namespace]['level'] = min($bytes['level'], $connection->transactionLevel());
+            }
+        }
+    }
+
+    public function afterRollback(Connection $connection): void
+    {
+        $name = $connection->getName();
+        foreach ($this->verified[$name] ?? [] as $uri => $namespaces) {
+            foreach ($namespaces as $namespace => $bytes) {
+                if ($bytes['level'] > $connection->transactionLevel()) {
+                    unset($this->verified[$name][$uri][$namespace]);
+                }
+            }
+            if ($this->verified[$name][$uri] === []) {
+                unset($this->verified[$name][$uri]);
+            }
+        }
+    }
+}
diff --git a/app/Support/RuntimeTrackedExternalPayloadStorage.php b/app/Support/RuntimeTrackedExternalPayloadStorage.php
index b0dc4d3bb..d1b0523be 100644
--- a/app/Support/RuntimeTrackedExternalPayloadStorage.php
+++ b/app/Support/RuntimeTrackedExternalPayloadStorage.php
@@ -14,13 +14,17 @@ public function __construct(
 
     public function put(string $data, string $sha256, string $codec): string
     {
-        return app(RuntimeExternalPayloadRegistry::class)->storeRetained(
+        $claims = app(RuntimeExternalPayloadWriteClaims::class);
+        $uri = $claims->prepare(fn (): string => app(RuntimeExternalPayloadRegistry::class)->storeRetained(
             $this->namespace,
             $this->inner,
             $data,
             $codec,
             strtolower($sha256),
-        );
+        ));
+        $claims->verified($this->namespace, $uri, strlen($data), strtolower($sha256));
+
+        return $uri;
     }
 
     public function uriFor(string $sha256, string $codec): string
@@ -45,8 +49,11 @@ public function retainEnvelope(array $envelope, ?string $validationField = null)
             if ($validationField !== null) {
                 AvroExternalPayloadValidator::validate($result['data'], $validationField);
             }
-            $registry->verifyFetchedBytesAndClaim(
+            app(RuntimeExternalPayloadWriteClaims::class)->prepare(fn () => $registry->verifyFetchedBytesAndClaim(
                 $this->namespace, $envelope['external_storage']['uri'], $result['data'],
+            ));
+            app(RuntimeExternalPayloadWriteClaims::class)->verified(
+                $this->namespace, $envelope['external_storage']['uri'], $result['data']->sizeBytes, $result['data']->sha256,
             );
 
             return ExternalPayloads::encodeStoredEnvelope($envelope);
@@ -59,11 +66,12 @@ public function get(string $uri): string
     {
         $data = $this->fetch($uri);
 
-        app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytesAndClaim(
+        $claims = app(RuntimeExternalPayloadWriteClaims::class);
+        $claims->prepare(fn () => app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytesAndClaim(
             $this->namespace,
             $uri,
             $data,
-        );
+        ));
 
         return $data;
     }
@@ -76,11 +84,11 @@ public function get(string $uri): string
      */
     public function verify(string $uri): array
     {
-        return app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytes(
+        return app(RuntimeExternalPayloadWriteClaims::class)->prepare(fn (): array => app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytes(
             $this->namespace,
             $uri,
             $this->fetch($uri),
-        );
+        ));
     }
 
     public function claimVerified(string $uri, int $sizeBytes, string $sha256): void
diff --git a/tests/Feature/NexusPayloadTransportTest.php b/tests/Feature/NexusPayloadTransportTest.php
index 41c85051a..05ed360a1 100644
--- a/tests/Feature/NexusPayloadTransportTest.php
+++ b/tests/Feature/NexusPayloadTransportTest.php
@@ -5,7 +5,8 @@
 use App\Models\RuntimeExternalPayload;
 use App\Models\WorkflowNamespace;
 use App\Support\RuntimeExternalPayloadRegistry;
-use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
+use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\File;
 use PHPUnit\Framework\Attributes\DataProvider;
 use Tests\Feature\Concerns\ServerTestHelpers;
@@ -19,7 +20,7 @@
 
 class NexusPayloadTransportTest extends TestCase
 {
-    use RefreshDatabase;
+    use DatabaseMigrations;
     use ServerTestHelpers;
 
     private const EXECUTE_PATH = '/api/service-endpoints/greeter/services/shared/operations/greet/execute';
@@ -27,6 +28,7 @@ class NexusPayloadTransportTest extends TestCase
     protected function setUp(): void
     {
         parent::setUp();
+        $this->assertSame(0, DB::connection()->transactionLevel(), 'HTTP writers must reach real commits.');
         $this->createNamespace('default');
         $this->createNamespace('other');
         foreach (['default', 'other'] as $namespace) {
diff --git a/tests/Feature/RuntimeExternalPayloadTransportTest.php b/tests/Feature/RuntimeExternalPayloadTransportTest.php
index eb83a8928..f22c2672c 100644
--- a/tests/Feature/RuntimeExternalPayloadTransportTest.php
+++ b/tests/Feature/RuntimeExternalPayloadTransportTest.php
@@ -13,8 +13,9 @@
 use App\Support\WorkerProtocol;
 use App\Support\WorkflowHistoryPageToken;
 use Illuminate\Contracts\Http\Kernel as HttpKernel;
-use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
 use Illuminate\Http\Request;
+use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\File;
 use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Facades\Queue;
@@ -33,7 +34,9 @@
 
 class RuntimeExternalPayloadTransportTest extends TestCase
 {
-    use RefreshDatabase;
+    // Reference lifetime reaches the outer commit; a test transaction would
+    // hide that boundary behind savepoints for every HTTP writer.
+    use DatabaseMigrations;
     use StoragePressureFixture;
 
     private string $storageDirectory;
@@ -41,6 +44,7 @@ class RuntimeExternalPayloadTransportTest extends TestCase
     protected function setUp(): void
     {
         parent::setUp();
+        $this->assertSame(0, DB::connection()->transactionLevel(), 'HTTP writers must reach real commits.');
 
         $this->storageDirectory = storage_path('framework/testing/runtime-external-payloads');
         File::deleteDirectory($this->storageDirectory);
diff --git a/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php b/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php
new file mode 100644
index 000000000..1375d6387
--- /dev/null
+++ b/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php
@@ -0,0 +1,116 @@
+<?php
+
+namespace Tests\Feature;
+
+use App\Models\WorkflowNamespace;
+use App\Support\RuntimeExternalPayloadException;
+use App\Support\RuntimeExternalPayloadRegistry;
+use App\Support\RuntimeExternalPayloadWriteClaims;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
+use Illuminate\Support\Facades\DB;
+use Mockery;
+use RuntimeException;
+use Tests\TestCase;
+
+class RuntimeExternalPayloadWriteClaimsTest extends TestCase
+{
+    use DatabaseMigrations;
+
+    public function test_a_failed_final_claim_rolls_back_the_writer_and_discards_its_commit_callbacks(): void
+    {
+        $claims = app(RuntimeExternalPayloadWriteClaims::class);
+        $committed = false;
+        try {
+            $claims->transport(function () use ($claims, &$committed): void {
+                $claims->verified('default', 'fixture://already-pruned', 1, hash('sha256', 'x'));
+                DB::transaction(function () use (&$committed): void {
+                    WorkflowNamespace::query()->create(['name' => 'uncommitted-reference', 'retention_days' => 30]);
+                    DB::afterCommit(static function () use (&$committed): void {
+                        $committed = true;
+                    });
+                });
+            });
+            $this->fail('A writer must not commit after cleanup removed its reference.');
+        } catch (RuntimeExternalPayloadException) {
+            $this->assertFalse($committed);
+            $this->assertSame(0, DB::connection()->transactionLevel());
+            $this->assertFalse(DB::connection()->getPdo()->inTransaction());
+            $this->assertFalse(WorkflowNamespace::query()->where('name', 'uncommitted-reference')->exists());
+        }
+
+        // Failed-request state cannot poison the next request or transaction.
+        DB::transaction(fn () => WorkflowNamespace::query()->create(['name' => 'next-request', 'retention_days' => 30]));
+        $this->assertTrue(WorkflowNamespace::query()->where('name', 'next-request')->exists());
+    }
+
+    public function test_preparation_commits_do_not_take_claims_but_the_reference_writer_does(): void
+    {
+        $registry = Mockery::mock(RuntimeExternalPayloadRegistry::class);
+        $registry->shouldReceive('claimVerified')->once()->with('default', 'fixture://verified', 1, hash('sha256', 'x'))
+            ->andReturnUsing(function (): void {
+                $this->assertSame(1, DB::connection()->transactionLevel());
+                $this->assertTrue(WorkflowNamespace::query()->where('name', 'reference-writer')->exists());
+            });
+        $this->app->instance(RuntimeExternalPayloadRegistry::class, $registry);
+        $claims = app(RuntimeExternalPayloadWriteClaims::class);
+        $claims->transport(function () use ($claims): void {
+            $claims->verified('default', 'fixture://verified', 1, hash('sha256', 'x'));
+            $claims->prepare(fn () => DB::transaction(fn () => WorkflowNamespace::query()->count()));
+            DB::transaction(fn () => WorkflowNamespace::query()->create(['name' => 'reference-writer', 'retention_days' => 30]));
+        });
+        $this->assertTrue(WorkflowNamespace::query()->where('name', 'reference-writer')->exists());
+    }
+
+    public function test_an_earlier_commit_does_not_consume_a_later_writers_reference_claim(): void
+    {
+        $registry = Mockery::mock(RuntimeExternalPayloadRegistry::class);
+        $registry->shouldReceive('claimVerified')->once()->ordered()->andReturnNull();
+        $registry->shouldReceive('claimVerified')->once()->ordered()->andThrow(new RuntimeExternalPayloadException(
+            'external_payload_not_found', 404, false, 'Cleanup won before the actual reference writer.',
+        ));
+        $this->app->instance(RuntimeExternalPayloadRegistry::class, $registry);
+        $claims = app(RuntimeExternalPayloadWriteClaims::class);
+        try {
+            $claims->transport(function () use ($claims): void {
+                $claims->verified('default', 'fixture://verified', 1, hash('sha256', 'x'));
+                DB::transaction(fn () => WorkflowNamespace::query()->create(['name' => 'earlier-metadata', 'retention_days' => 30]));
+                DB::transaction(fn () => WorkflowNamespace::query()->create(['name' => 'later-reference', 'retention_days' => 30]));
+            });
+            $this->fail('Every later writer must recheck the prepared reference.');
+        } catch (RuntimeExternalPayloadException) {
+            $this->assertTrue(WorkflowNamespace::query()->where('name', 'earlier-metadata')->exists());
+            $this->assertFalse(WorkflowNamespace::query()->where('name', 'later-reference')->exists());
+            $this->assertSame(0, DB::connection()->transactionLevel());
+            $this->assertFalse(DB::connection()->getPdo()->inTransaction());
+        }
+    }
+
+    public function test_savepoint_rollback_drops_only_its_own_claims_and_the_final_batch_uses_bucket_order(): void
+    {
+        $seen = [];
+        $registry = Mockery::mock(RuntimeExternalPayloadRegistry::class);
+        $registry->shouldReceive('claimVerified')->twice()->andReturnUsing(static function ($namespace, $uri) use (&$seen): void {
+            $seen[] = $uri;
+        });
+        $this->app->instance(RuntimeExternalPayloadRegistry::class, $registry);
+        $claims = app(RuntimeExternalPayloadWriteClaims::class);
+        $claims->transport(function () use ($claims): void {
+            // The first URI hashes to lock bucket255; the second to bucket0.
+            $claims->verified('default', 'fixture://guard-163', 1, hash('sha256', 'x'));
+            DB::transaction(function () use ($claims): void {
+                DB::transaction(fn () => $claims->verified('default', 'fixture://guard-78', 1, hash('sha256', 'x')));
+                try {
+                    DB::transaction(function () use ($claims): void {
+                        $claims->verified('default', 'fixture://rolled-back-registration', 1, hash('sha256', 'x'));
+                        throw new RuntimeException('Rollback only this savepoint.');
+                    });
+                } catch (RuntimeException) {
+                    // The previous sibling's committed registration is now
+                    // owned by the outer transaction and must still be claimed.
+                }
+                WorkflowNamespace::query()->create(['name' => 'savepoint-writer', 'retention_days' => 30]);
+            });
+        });
+        $this->assertSame(['fixture://guard-78', 'fixture://guard-163'], $seen);
+    }
+}
diff --git a/tests/Feature/WorkflowStreamPayloadRaceTest.php b/tests/Feature/WorkflowStreamPayloadRaceTest.php
index fba4651a5..39b22e494 100644
--- a/tests/Feature/WorkflowStreamPayloadRaceTest.php
+++ b/tests/Feature/WorkflowStreamPayloadRaceTest.php
@@ -7,9 +7,11 @@
 use App\Models\WorkflowDurableStreamItem;
 use App\Models\WorkflowNamespace;
 use App\Support\HistoryRetentionEnforcer;
+use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\RuntimeExternalPayloadException;
 use App\Support\RuntimeExternalPayloadObjectLock;
 use App\Support\RuntimeExternalPayloadRegistry;
+use App\Support\RuntimeExternalPayloadWriteClaims;
 use App\Support\WorkflowRunStreamLock;
 use App\Support\WorkflowStreamCommandProcessor;
 use App\Support\WorkflowStreamRetention;
@@ -24,6 +26,7 @@
 use Workflow\V2\Enums\RunStatus;
 use Workflow\V2\Enums\TaskStatus;
 use Workflow\V2\Enums\TaskType;
+use Workflow\V2\Models\WorkflowInstance;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Support\ExternalPayloads;
@@ -249,13 +252,12 @@ public function test_appends_listing_shared_objects_in_opposite_orders_both_comm
     }
 
     /**
-     * REPRODUCTION (not a fix): a workflow start claims its external input
+     * A workflow start previously claimed its external input
      * in its own committed step, before the transaction that writes the run
      * referencing it. Run retention that runs between the two finds no
      * retained reference and deletes the object, and the start then commits
-     * a run whose arguments reference a deleted object. Expected to fail
-     * until non-stream writers claim inside the transaction that records the
-     * reference.
+     * a run whose arguments reference a deleted object. The final claim must
+     * now refuse that start and roll back all its reference-writing state.
      */
     #[DataProvider('drivers')]
     public function test_a_workflow_start_and_run_pruning_decide_a_shared_object(string $driver): void
@@ -281,10 +283,48 @@ public function test_a_workflow_start_and_run_pruning_decide_a_shared_object(str
             $this->assertSame($object['uri'], ExternalPayloads::storedEnvelope($arguments)['external_storage']['uri']);
             $this->assertObjectKept($object);
         } else {
+            $this->assertContains($started['status'], [404, 410, 422], json_encode($started));
+            $this->assertContains($started['reason'], [
+                'external_payload_not_found', 'external_payload_expired', 'external_payload_unsupported',
+            ]);
+            $this->assertFalse(WorkflowRun::query()->where('workflow_instance_id', 'wf-claim-race-new')->exists());
+            $this->assertFalse(WorkflowInstance::query()->whereKey('wf-claim-race-new')->exists());
             $this->assertObjectDeleted($object);
         }
     }
 
+    #[DataProvider('drivers')]
+    public function test_a_workflow_start_holds_its_final_claim_until_its_reference_is_visible(string $driver): void
+    {
+        $this->initializeDatabase($driver);
+        $object = $this->registerObject('start-wins');
+        $oldRunId = $this->summarizedRun('wf-claim-wins-old', RunStatus::Completed);
+        $this->addStream($oldRunId, 'old', [$object['uri']]);
+        $written = false;
+        $finalClaim = static function (string $sql, string $driver, int $level) use (&$written): bool {
+            if (str_starts_with($sql, 'insert') && str_contains($sql, 'workflow_runs')) {
+                $written = true;
+            }
+
+            return $written && $level >= 1 && str_starts_with($sql, 'select')
+                && str_contains($sql, 'runtime_external_payloads');
+        };
+
+        $start = $this->startActor(fn (): array => $this->httpStart('wf-claim-wins-new', $object['reference']), $finalClaim);
+        $this->runUntilPaused($start);
+        $cleanup = $this->startActor(fn (): array => HistoryRetentionEnforcer::pruneRun('default', $oldRunId));
+        $this->release($cleanup);
+        $this->awaitWait($cleanup, 'runtime_external_payload_object_locks');
+        $this->resume($start);
+        $this->assertSame(201, $this->finishActor($start)['status']);
+        $this->assertTrue($this->finishActor($cleanup)['pruned']);
+        DB::purge();
+        DB::reconnect();
+        $arguments = (string) WorkflowRun::query()->where('workflow_instance_id', 'wf-claim-wins-new')->value('arguments');
+        $this->assertSame($object['uri'], ExternalPayloads::storedEnvelope($arguments)['external_storage']['uri']);
+        $this->assertObjectKept($object);
+    }
+
     private function httpStart(string $workflowId, array $reference): array
     {
         $response = $this->withHeaders($this->apiHeaders())->postJson('/api/workflows', [
@@ -297,6 +337,33 @@ private function httpStart(string $workflowId, array $reference): array
         return ['status' => $response->status(), 'reason' => $response->json('reason')];
     }
 
+    #[DataProvider('drivers')]
+    public function test_materialized_input_can_commit_inline_after_its_source_object_is_pruned(string $driver): void
+    {
+        $this->initializeDatabase($driver);
+        $object = $this->registerObject('materialized-input');
+        $oldRunId = $this->summarizedRun('wf-inline-old', RunStatus::Completed);
+        $this->addStream($oldRunId, 'old', [$object['uri']]);
+
+        $writer = $this->startActor(fn (): string => app(RuntimeExternalPayloadWriteClaims::class)->transport(function () use ($object): string {
+            $bytes = app(NamespaceExternalPayloadStorage::class)->driverFor('default')->get($object['uri']);
+
+            return DB::transaction(function () use ($bytes): string {
+                $runId = $this->createRun('wf-inline-new');
+                WorkflowRun::query()->whereKey($runId)->update(['arguments' => $bytes, 'payload_codec' => 'avro']);
+
+                return $runId;
+            });
+        }), self::afterVerification());
+        $this->runUntilPaused($writer);
+        $this->assertTrue(HistoryRetentionEnforcer::pruneRun('default', $oldRunId)['pruned']);
+        $this->resume($writer);
+        $runId = $this->finishActor($writer);
+        $arguments = WorkflowRun::query()->whereKey($runId)->value('arguments');
+        $this->assertSame(['materialized-input'], Serializer::unserializeWithCodec('avro', $arguments));
+        $this->assertObjectDeleted($object);
+    }
+
     private function initializeDatabase(string $driver): void
     {
         $this->initializeActorDatabase($driver);

From 8ddf53d1a2ac3d0b2e7a658a70da251d44fb8ba6 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 20:49:18 +0000
Subject: [PATCH 10/13] Assert payload refusal against the stored instance
 primary key

---
 tests/Feature/RuntimeExternalPayloadTransportTest.php | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/Feature/RuntimeExternalPayloadTransportTest.php b/tests/Feature/RuntimeExternalPayloadTransportTest.php
index f22c2672c..458a3f351 100644
--- a/tests/Feature/RuntimeExternalPayloadTransportTest.php
+++ b/tests/Feature/RuntimeExternalPayloadTransportTest.php
@@ -624,7 +624,7 @@ public function test_state_bearing_request_rejects_oversized_backing_object_with
             ->assertJsonPath('retryable', false);
 
         $this->assertDatabaseMissing('workflow_instances', [
-            'workflow_id' => 'oversized-runtime-reference',
+            'id' => 'oversized-runtime-reference',
         ]);
         $this->assertNull($row->fresh()?->last_fetched_at);
     }
@@ -676,7 +676,7 @@ public function test_state_bearing_request_claims_reference_and_obsolete_provide
             ])
             ->assertStatus(422)
             ->assertJsonPath('reason', 'external_payload_integrity_mismatch');
-        $this->assertDatabaseMissing('workflow_instances', ['workflow_id' => 'tampered-runtime-reference']);
+        $this->assertDatabaseMissing('workflow_instances', ['id' => 'tampered-runtime-reference']);
 
         $this->withHeaders($this->controlHeaders())
             ->postJson('/api/workflows', [

From 265fa72b938fd5190d0bea7bd014bc0d8ccab32a Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 20:50:43 +0000
Subject: [PATCH 11/13] Correct the provider reference refusal assertion key

---
 tests/Feature/RuntimeExternalPayloadTransportTest.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/Feature/RuntimeExternalPayloadTransportTest.php b/tests/Feature/RuntimeExternalPayloadTransportTest.php
index 458a3f351..8eb4c3af5 100644
--- a/tests/Feature/RuntimeExternalPayloadTransportTest.php
+++ b/tests/Feature/RuntimeExternalPayloadTransportTest.php
@@ -695,7 +695,7 @@ public function test_state_bearing_request_claims_reference_and_obsolete_provide
             ])
             ->assertStatus(422)
             ->assertJsonPath('reason', 'external_payload_unsupported');
-        $this->assertDatabaseMissing('workflow_instances', ['workflow_id' => 'provider-reference-rejected']);
+        $this->assertDatabaseMissing('workflow_instances', ['id' => 'provider-reference-rejected']);
     }
 
     public function test_large_workflow_input_keeps_the_verified_reference_without_materializing_bytes(): void

From e90a244c9f9b5505a6d79a2aa68aac65a599b310 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 21:22:31 +0000
Subject: [PATCH 12/13] Claim HTTP inputs at commit and preserve scheduled
 payloads

---
 .../ExternalPayloadRetentionCleanup.php       |  2 +
 .../RuntimeExternalPayloadWriteClaims.php     | 19 +++++++
 .../RuntimeTrackedExternalPayloadStorage.php  | 12 +----
 docs/contracts/external-payload-storage.md    |  7 +++
 tests/Feature/NexusPayloadTransportTest.php   |  6 ++-
 .../PayloadEnvelopeIntegrationTest.php        |  5 +-
 .../RuntimeExternalPayloadWriteClaimsTest.php | 40 +++++++++++++++
 .../RuntimePayloadCompletionUploadsTest.php   |  5 +-
 tests/Feature/WorkerServiceOperationTest.php  |  6 ++-
 .../Feature/WorkflowStreamPayloadRaceTest.php | 51 ++++++++++++++-----
 10 files changed, 124 insertions(+), 29 deletions(-)

diff --git a/app/Support/ExternalPayloadRetentionCleanup.php b/app/Support/ExternalPayloadRetentionCleanup.php
index fd0036942..a3703d9c9 100644
--- a/app/Support/ExternalPayloadRetentionCleanup.php
+++ b/app/Support/ExternalPayloadRetentionCleanup.php
@@ -21,6 +21,7 @@
 use Workflow\V2\Models\WorkflowRunSummary;
 use Workflow\V2\Models\WorkflowRunTimerEntry;
 use Workflow\V2\Models\WorkflowRunWait;
+use Workflow\V2\Models\WorkflowSchedule;
 use Workflow\V2\Models\WorkflowScheduleHistoryEvent;
 use Workflow\V2\Models\WorkflowSearchAttribute;
 use Workflow\V2\Models\WorkflowServiceCall;
@@ -686,6 +687,7 @@ private function retainedPayloadColumns(
         }
 
         return [
+            [WorkflowSchedule::query(), 'action'],
             [WorkflowRunSummary::query()->whereIn('id', $this->retainedRunIdsQuery($deletedRunIds)), 'visibility_labels'],
             [ActivityExecution::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'arguments'],
             [ActivityExecution::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'result'],
diff --git a/app/Support/RuntimeExternalPayloadWriteClaims.php b/app/Support/RuntimeExternalPayloadWriteClaims.php
index 9bcd5a852..e12acef01 100644
--- a/app/Support/RuntimeExternalPayloadWriteClaims.php
+++ b/app/Support/RuntimeExternalPayloadWriteClaims.php
@@ -54,6 +54,25 @@ public function verified(string $namespace, string $uri, int $sizeBytes, string
         }
     }
 
+    public function verifyInput(string $namespace, string $uri, string|ExternalPayloadStream $data): void
+    {
+        $bytes = $this->prepare(function () use ($namespace, $uri, $data): ?array {
+            $registry = app(RuntimeExternalPayloadRegistry::class);
+            if ($this->active) {
+                return $registry->verifyFetchedBytes($namespace, $uri, $data);
+            }
+
+            // Non-transport callers retain their existing storage-driver
+            // contract. HTTP input claims belong to the reference writer.
+            $registry->verifyFetchedBytesAndClaim($namespace, $uri, $data);
+
+            return null;
+        });
+        if ($bytes !== null) {
+            $this->verified($namespace, $uri, $bytes['size_bytes'], $bytes['sha256']);
+        }
+    }
+
     public function beforeCommit(Connection $connection): void
     {
         $verified = $this->verified[$connection->getName()] ?? [];
diff --git a/app/Support/RuntimeTrackedExternalPayloadStorage.php b/app/Support/RuntimeTrackedExternalPayloadStorage.php
index d1b0523be..a55eb5046 100644
--- a/app/Support/RuntimeTrackedExternalPayloadStorage.php
+++ b/app/Support/RuntimeTrackedExternalPayloadStorage.php
@@ -49,11 +49,8 @@ public function retainEnvelope(array $envelope, ?string $validationField = null)
             if ($validationField !== null) {
                 AvroExternalPayloadValidator::validate($result['data'], $validationField);
             }
-            app(RuntimeExternalPayloadWriteClaims::class)->prepare(fn () => $registry->verifyFetchedBytesAndClaim(
+            app(RuntimeExternalPayloadWriteClaims::class)->verifyInput(
                 $this->namespace, $envelope['external_storage']['uri'], $result['data'],
-            ));
-            app(RuntimeExternalPayloadWriteClaims::class)->verified(
-                $this->namespace, $envelope['external_storage']['uri'], $result['data']->sizeBytes, $result['data']->sha256,
             );
 
             return ExternalPayloads::encodeStoredEnvelope($envelope);
@@ -66,12 +63,7 @@ public function get(string $uri): string
     {
         $data = $this->fetch($uri);
 
-        $claims = app(RuntimeExternalPayloadWriteClaims::class);
-        $claims->prepare(fn () => app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytesAndClaim(
-            $this->namespace,
-            $uri,
-            $data,
-        ));
+        app(RuntimeExternalPayloadWriteClaims::class)->verifyInput($this->namespace, $uri, $data);
 
         return $data;
     }
diff --git a/docs/contracts/external-payload-storage.md b/docs/contracts/external-payload-storage.md
index 1b93286c5..6c9dfb0db 100644
--- a/docs/contracts/external-payload-storage.md
+++ b/docs/contracts/external-payload-storage.md
@@ -144,6 +144,13 @@ retention authority. SDKs have no delete endpoint, and namespace/run cleanup
 does not remove a backing object while retained state in any namespace still
 owns it.
 
+Opaque-reference preflight verifies bytes without committing a claim. The
+state-bearing database transaction claims verified inputs before its outer
+commit, after ownership locks and in canonical object-lock order. A failed
+writer rolls back its claim updates and commit callbacks. If cleanup wins
+first, the request receives a typed payload refusal and commits no reference
+to the removed object.
+
 The singleton maintenance runner executes
 `external-payloads:cleanup --limit=100` on every maintenance pass. The Docker
 topology runs a pass every 10 seconds and the default Kubernetes CronJob runs a
diff --git a/tests/Feature/NexusPayloadTransportTest.php b/tests/Feature/NexusPayloadTransportTest.php
index 05ed360a1..abe1b4add 100644
--- a/tests/Feature/NexusPayloadTransportTest.php
+++ b/tests/Feature/NexusPayloadTransportTest.php
@@ -137,6 +137,10 @@ private function expectDispatch(string $blob): void
                 $this->assertSame('avro', $options['payload_codec']);
 
                 return true;
-            })->andReturn(['accepted' => true, 'status' => 'started']);
+            })->andReturnUsing(static fn (): array => DB::transaction(
+                // The dispatch double models the real admission transaction,
+                // including its commit-time input claim.
+                static fn (): array => ['accepted' => true, 'status' => 'started'],
+            ));
     }
 }
diff --git a/tests/Feature/PayloadEnvelopeIntegrationTest.php b/tests/Feature/PayloadEnvelopeIntegrationTest.php
index e0caca1ad..e9942493c 100644
--- a/tests/Feature/PayloadEnvelopeIntegrationTest.php
+++ b/tests/Feature/PayloadEnvelopeIntegrationTest.php
@@ -11,7 +11,7 @@
 use App\Support\ExternalPayloadEnvelopeService;
 use App\Support\RuntimeExternalPayloadRegistry;
 use App\Support\WorkerProtocol;
-use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\File;
 use Illuminate\Support\Facades\Queue;
@@ -38,13 +38,14 @@
 
 class PayloadEnvelopeIntegrationTest extends TestCase
 {
-    use RefreshDatabase;
+    use DatabaseMigrations;
 
     private string $externalStorageDirectory;
 
     protected function setUp(): void
     {
         parent::setUp();
+        $this->assertSame(0, DB::connection()->transactionLevel(), 'Payload writers must reach real commits.');
 
         $this->externalStorageDirectory = storage_path('framework/testing/payload-envelope-external-storage');
         File::deleteDirectory($this->externalStorageDirectory);
diff --git a/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php b/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php
index 1375d6387..89e5dae86 100644
--- a/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php
+++ b/tests/Feature/RuntimeExternalPayloadWriteClaimsTest.php
@@ -2,15 +2,19 @@
 
 namespace Tests\Feature;
 
+use App\Models\RuntimeExternalPayload;
 use App\Models\WorkflowNamespace;
+use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\RuntimeExternalPayloadException;
 use App\Support\RuntimeExternalPayloadRegistry;
 use App\Support\RuntimeExternalPayloadWriteClaims;
 use Illuminate\Foundation\Testing\DatabaseMigrations;
 use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\File;
 use Mockery;
 use RuntimeException;
 use Tests\TestCase;
+use Workflow\Serializers\Serializer;
 
 class RuntimeExternalPayloadWriteClaimsTest extends TestCase
 {
@@ -113,4 +117,40 @@ public function test_savepoint_rollback_drops_only_its_own_claims_and_the_final_
         });
         $this->assertSame(['fixture://guard-78', 'fixture://guard-163'], $seen);
     }
+
+    public function test_preflight_and_a_failed_writer_leave_the_verified_upload_unclaimed(): void
+    {
+        $directory = storage_path('framework/testing/write-claims-'.bin2hex(random_bytes(4)));
+        try {
+            WorkflowNamespace::query()->create([
+                'name' => 'claim-preflight', 'retention_days' => 30,
+                'external_payload_storage' => [
+                    'enabled' => true, 'driver' => 'local', 'threshold_bytes' => 1,
+                    'config' => ['uri' => 'file://'.$directory],
+                ],
+            ]);
+            $bytes = Serializer::serializeWithCodec('avro', ['verified before mutation']);
+            $registry = app(RuntimeExternalPayloadRegistry::class);
+            $reference = $registry->upload('claim-preflight', $bytes, 'avro', hash('sha256', $bytes));
+            $row = RuntimeExternalPayload::query()->findOrFail($reference['reference_id']);
+            $envelope = $registry->resolveAndClaim('claim-preflight', $reference);
+            $storage = app(NamespaceExternalPayloadStorage::class)->driverFor('claim-preflight');
+            app(RuntimeExternalPayloadWriteClaims::class)->transport(function () use ($storage, $row, $envelope, $bytes): void {
+                $this->assertSame($bytes, $storage->get($row->storage_uri));
+                $storage->retainEnvelope($envelope);
+                $this->assertNull($row->fresh()->retained_at, 'Verification must not commit a claim.');
+                try {
+                    DB::transaction(function (): void {
+                        WorkflowNamespace::query()->create(['name' => 'failed-preflight-writer', 'retention_days' => 30]);
+                        throw new RuntimeException('The reference writer rolled back.');
+                    });
+                } catch (RuntimeException) {
+                    $this->assertNull($row->fresh()->retained_at);
+                    $this->assertNotNull($row->fresh()->expires_at);
+                }
+            });
+        } finally {
+            File::deleteDirectory($directory);
+        }
+    }
 }
diff --git a/tests/Feature/RuntimePayloadCompletionUploadsTest.php b/tests/Feature/RuntimePayloadCompletionUploadsTest.php
index 6871c385b..e6cdc280d 100644
--- a/tests/Feature/RuntimePayloadCompletionUploadsTest.php
+++ b/tests/Feature/RuntimePayloadCompletionUploadsTest.php
@@ -10,7 +10,7 @@
 use App\Support\RuntimeExternalPayloadRegistry;
 use App\Support\RuntimePayloadCompletionContext;
 use App\Support\WorkflowQueryTaskBroker;
-use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\File;
 use Illuminate\Support\Facades\Queue;
@@ -25,7 +25,7 @@
 
 class RuntimePayloadCompletionUploadsTest extends TestCase
 {
-    use RefreshDatabase;
+    use DatabaseMigrations;
     use ServerTestHelpers;
     use StoragePressureFixture;
 
@@ -34,6 +34,7 @@ class RuntimePayloadCompletionUploadsTest extends TestCase
     protected function setUp(): void
     {
         parent::setUp();
+        $this->assertSame(0, DB::connection()->transactionLevel(), 'Completion uploads must reach real commits.');
         Queue::fake();
         $this->directory = storage_path('framework/testing/completion-uploads');
         File::deleteDirectory($this->directory);
diff --git a/tests/Feature/WorkerServiceOperationTest.php b/tests/Feature/WorkerServiceOperationTest.php
index 19e9126b6..b90d987e8 100644
--- a/tests/Feature/WorkerServiceOperationTest.php
+++ b/tests/Feature/WorkerServiceOperationTest.php
@@ -6,7 +6,8 @@
 
 use App\Models\RuntimeExternalPayload;
 use App\Models\WorkflowNamespace;
-use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Foundation\Testing\DatabaseMigrations;
+use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\File;
 use Illuminate\Support\Facades\Queue;
 use Illuminate\Testing\TestResponse;
@@ -27,12 +28,13 @@
 
 final class WorkerServiceOperationTest extends TestCase
 {
-    use RefreshDatabase;
+    use DatabaseMigrations;
     use ServerTestHelpers;
 
     protected function setUp(): void
     {
         parent::setUp();
+        $this->assertSame(0, DB::connection()->transactionLevel(), 'Service admissions must reach real commits.');
         Queue::fake();
         $this->createNamespace('default');
         $this->registerWorker('nexus-worker', 'nexus', supportedWorkflowTypes: ['tests.nexus-caller', 'tests.nexus-target']);
diff --git a/tests/Feature/WorkflowStreamPayloadRaceTest.php b/tests/Feature/WorkflowStreamPayloadRaceTest.php
index 39b22e494..8f432d455 100644
--- a/tests/Feature/WorkflowStreamPayloadRaceTest.php
+++ b/tests/Feature/WorkflowStreamPayloadRaceTest.php
@@ -338,32 +338,59 @@ private function httpStart(string $workflowId, array $reference): array
     }
 
     #[DataProvider('drivers')]
-    public function test_materialized_input_can_commit_inline_after_its_source_object_is_pruned(string $driver): void
+    public function test_materialized_input_cannot_bypass_its_commit_time_reference_claim(string $driver): void
     {
         $this->initializeDatabase($driver);
         $object = $this->registerObject('materialized-input');
         $oldRunId = $this->summarizedRun('wf-inline-old', RunStatus::Completed);
         $this->addStream($oldRunId, 'old', [$object['uri']]);
 
-        $writer = $this->startActor(fn (): string => app(RuntimeExternalPayloadWriteClaims::class)->transport(function () use ($object): string {
-            $bytes = app(NamespaceExternalPayloadStorage::class)->driverFor('default')->get($object['uri']);
+        $writer = $this->startActor(function () use ($object): array {
+            try {
+                $runId = app(RuntimeExternalPayloadWriteClaims::class)->transport(function () use ($object): string {
+                    $bytes = app(NamespaceExternalPayloadStorage::class)->driverFor('default')->get($object['uri']);
 
-            return DB::transaction(function () use ($bytes): string {
-                $runId = $this->createRun('wf-inline-new');
-                WorkflowRun::query()->whereKey($runId)->update(['arguments' => $bytes, 'payload_codec' => 'avro']);
+                    return DB::transaction(function () use ($bytes): string {
+                        $runId = $this->createRun('wf-inline-new');
+                        WorkflowRun::query()->whereKey($runId)->update(['arguments' => $bytes, 'payload_codec' => 'avro']);
 
-                return $runId;
-            });
-        }), self::afterVerification());
+                        return $runId;
+                    });
+                });
+
+                return ['status' => 201, 'run_id' => $runId];
+            } catch (RuntimeExternalPayloadException $exception) {
+                return ['status' => $exception->status, 'reason' => $exception->reason];
+            }
+        }, self::afterVerification());
         $this->runUntilPaused($writer);
         $this->assertTrue(HistoryRetentionEnforcer::pruneRun('default', $oldRunId)['pruned']);
         $this->resume($writer);
-        $runId = $this->finishActor($writer);
-        $arguments = WorkflowRun::query()->whereKey($runId)->value('arguments');
-        $this->assertSame(['materialized-input'], Serializer::unserializeWithCodec('avro', $arguments));
+        $this->assertSame(['status' => 422, 'reason' => 'external_payload_unsupported'], $this->finishActor($writer));
+        $this->assertFalse(WorkflowRun::query()->where('workflow_instance_id', 'wf-inline-new')->exists());
         $this->assertObjectDeleted($object);
     }
 
+    #[DataProvider('drivers')]
+    public function test_a_retained_schedule_protects_input_shared_with_a_pruned_run(string $driver): void
+    {
+        $this->initializeDatabase($driver);
+        $object = $this->registerObject('scheduled-input');
+        $oldRunId = $this->summarizedRun('wf-scheduled-old', RunStatus::Completed);
+        $this->addStream($oldRunId, 'old', [$object['uri']]);
+
+        $this->withHeaders($this->apiHeaders())->postJson('/api/schedules', [
+            'schedule_id' => 'shared-input-schedule',
+            'spec' => ['cron_expressions' => ['0 * * * *']],
+            'action' => [
+                'workflow_type' => 'remote.claim-race',
+                'input' => ['codec' => 'avro', 'external_payload' => $object['reference']],
+            ],
+        ])->assertCreated();
+        $this->assertTrue(HistoryRetentionEnforcer::pruneRun('default', $oldRunId)['pruned']);
+        $this->assertObjectKept($object);
+    }
+
     private function initializeDatabase(string $driver): void
     {
         $this->initializeActorDatabase($driver);

From 9d7cd8fcba3dda8bcec984e7048d0ffa27566d00 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Mon, 26 Oct 2026 21:39:41 +0000
Subject: [PATCH 13/13] Align compatibility tests with protocol-aware Core
 adoption

---
 tests/Feature/ClusterInfoCompatibilityTest.php   |  3 ++-
 .../RuntimePayloadCompletionUploadsTest.php      | 16 ++++++++++++----
 tests/Feature/WorkflowWorkerProtocolTest.php     |  7 +++++--
 3 files changed, 19 insertions(+), 7 deletions(-)

diff --git a/tests/Feature/ClusterInfoCompatibilityTest.php b/tests/Feature/ClusterInfoCompatibilityTest.php
index 051d1f398..d6bfa5d5b 100644
--- a/tests/Feature/ClusterInfoCompatibilityTest.php
+++ b/tests/Feature/ClusterInfoCompatibilityTest.php
@@ -1087,7 +1087,8 @@ public function test_worker_protocol_manifest_uses_server_protocol_and_package_w
         ));
         $response = $this->getJson('/api/cluster/info')->assertOk();
 
-        $this->assertSame(WorkerProtocolVersion::VERSION, WorkerProtocol::VERSION);
+        $this->assertSame('1.20', WorkerProtocol::VERSION);
+        $this->assertTrue(version_compare(WorkerProtocolVersion::VERSION, WorkerProtocol::VERSION, '>='));
         $this->assertSame(WorkerProtocol::VERSION, (string) config('server.worker_protocol.version'));
         $this->assertSame($expectedCommands, WorkerProtocol::supportedWorkflowTaskCommands());
         $this->assertContains('cancel_selection_operation', WorkerProtocol::supportedWorkflowTaskCommands());
diff --git a/tests/Feature/RuntimePayloadCompletionUploadsTest.php b/tests/Feature/RuntimePayloadCompletionUploadsTest.php
index e6cdc280d..74d7b261d 100644
--- a/tests/Feature/RuntimePayloadCompletionUploadsTest.php
+++ b/tests/Feature/RuntimePayloadCompletionUploadsTest.php
@@ -67,14 +67,14 @@ public function test_owned_activity_result_can_upload_complete_and_reconcile_aft
         $payload = Serializer::serializeWithCodec('avro', str_repeat('result', 200));
         $this->observeStoragePressure('draining');
         $reference = $this->upload($payload, $context)->assertCreated()->json('reference');
-        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+        $this->assertSameReference($this->upload($payload, $context)->assertCreated(), $reference);
         $this->postJson('/api/worker/activity-tasks/'.$context['task_id'].'/complete', [
             'activity_attempt_id' => $context['attempt'], 'lease_owner' => 'worker',
             'result' => ['codec' => 'avro', 'external_payload' => $reference],
         ], $this->workerHeaders())->assertOk();
         $before = RuntimeExternalPayload::query()->sole()->getRawOriginal();
         $budgetBefore = RuntimePayloadCompletionBudget::query()->sole()->getRawOriginal();
-        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+        $this->assertSameReference($this->upload($payload, $context)->assertCreated(), $reference);
         $this->assertSame($before, RuntimeExternalPayload::query()->sole()->getRawOriginal());
         $this->assertSame($budgetBefore, RuntimePayloadCompletionBudget::query()->sole()->getRawOriginal());
         $this->getJson('/api/activities/activity', $this->apiHeaders())->assertOk()
@@ -109,7 +109,7 @@ public function test_workflow_completion_upload_can_be_committed_during_draining
         ], $this->workerHeaders())->assertOk();
         $this->assertNotNull(RuntimeExternalPayload::query()->sole()->retained_at);
         $this->assertSame('completed', WorkflowRun::query()->sole()->status->value);
-        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+        $this->assertSameReference($this->upload($payload, $context)->assertCreated(), $reference);
     }
 
     #[DataProvider('completionRosterStates')]
@@ -147,7 +147,7 @@ public function test_query_completion_upload_uses_its_own_current_lease(string $
             'lease_owner' => 'worker', 'query_task_attempt' => $context['attempt'],
             'result_envelope' => ['codec' => 'avro', 'external_payload' => $reference],
         ], $this->workerHeaders())->assertOk()->assertJsonPath('outcome', 'completed');
-        $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference);
+        $this->assertSameReference($this->upload($payload, $context)->assertCreated(), $reference);
     }
 
     public function test_activity_cannot_claim_another_budget_as_a_workflow_task(): void
@@ -394,6 +394,14 @@ private function workflow(): array
             'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['commands', 0, 'result']];
     }
 
+    private function assertSameReference(TestResponse $response, array $reference): void
+    {
+        $response->assertJsonCount(count($reference), 'reference');
+        foreach ($reference as $field => $value) {
+            $response->assertJsonPath('reference.'.$field, $value);
+        }
+    }
+
     private function upload(string $bytes, ?array $context, string $namespace = 'default', ?string $token = null): TestResponse
     {
         return $this->call('POST', '/api/external-payloads/v1', [], [], [], [
diff --git a/tests/Feature/WorkflowWorkerProtocolTest.php b/tests/Feature/WorkflowWorkerProtocolTest.php
index cb0f5761b..17974a144 100644
--- a/tests/Feature/WorkflowWorkerProtocolTest.php
+++ b/tests/Feature/WorkflowWorkerProtocolTest.php
@@ -20,6 +20,7 @@
 use Tests\Fixtures\InteractiveCommandWorkflow;
 use Tests\TestCase;
 use Workflow\Serializers\Serializer;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
 use Workflow\V2\Enums\ActivityAttemptStatus;
 use Workflow\V2\Enums\HistoryEventType;
@@ -2433,8 +2434,10 @@ public function test_completion_returns_422_when_structural_limit_exceeded(): vo
 
         $this->instance(
             WorkflowTaskBridge::class,
-            \Mockery::mock(WorkflowTaskBridge::class, static function (MockInterface $mock) {
-                $mock->shouldReceive('complete')
+            \Mockery::mock(ProtocolAwareWorkflowTaskBridge::class, static function (MockInterface $mock) use ($taskId) {
+                $mock->shouldReceive('completeForProtocol')
+                    ->once()
+                    ->with($taskId, \Mockery::type('array'), WorkerProtocol::VERSION)
                     ->andThrow(
                         StructuralLimitExceededException::pendingActivityCount(2000, 2000),
                     );