Adopt Core 2.14.2 and make PHP stream and payload writes atomic

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,

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,
         );
     }
 

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>
      */

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);
@@ -3212,15 +3219,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 +3256,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 +3298,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);

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');

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;
@@ -35,11 +36,15 @@
 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;
 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 +71,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 ([
@@ -181,6 +197,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

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'],