Preserve inline service calls and gate durable scheduling by protocol

docs/api-stability.md

@@ -220,6 +220,27 @@ and keep workers below that floor away from histories that require the event.
 The original `start_service_operation` command remains available from its 1.13
 floor; the scheduled execution semantics require 1.21.
 
+The stable two-argument `WorkflowTaskBridge::complete()` method remains
+unchanged. External hosts that offer workflow-task completion must require an
+authenticated, supported worker protocol and use the optional
+`ProtocolAwareWorkflowTaskBridge::completeForProtocol()` role. They must check
+that the configured bridge implements that role before advertising completion
+support. Missing, malformed, noncanonical, different-major, and future-minor versions fail
+closed. Supported protocols below 1.21 retain the pre-1.21 inline
+service-operation path and therefore cannot create `ServiceOperationScheduled`.
+The built-in `WorkflowExecutor` runs at the package's current protocol and
+supplies scheduler authority explicitly.
+
+Terminal service-call result envelopes remain intact in workflow history so
+inline results replay through the same Avro payload contract as scheduled results.
+
+Caller workflow identity on a direct `ServiceControlPlane::execute()` request
+does not activate the workflow-task scheduler. In particular, an idempotency
+transaction plus `caller_workflow_run_id` must continue to honor the requested
+HTTP wait surface: `wait_for=completed` returns the terminal result or typed
+failure. Scheduling requires private evidence supplied by the version-gated
+workflow-task completion path.
+
 `WorkflowServiceCallDispatch` is a documented persisted model. Its table and
 state/attempt/claim/revision/repair fields are additive schema and observability
 surface. `ServiceCallScheduler`, `ServiceCallAttemptClaim`, and the

docs/architecture/workflow-service-calls-architecture.md

@@ -215,6 +215,17 @@ immutable accepted handler-binding snapshot, a
 `workflow_service_call_dispatches` row, and a `ServiceOperationScheduled`
 history event. It MUST NOT invoke a workflow/activity/carrier handler, enqueue
 work before commit, wait for an update result, or sleep for retry backoff.
+This rule applies to protocol 1.21 and later completions. A host passes the
+authenticated protocol through the optional
+`ProtocolAwareWorkflowTaskBridge::completeForProtocol()` role; the stable
+two-argument `WorkflowTaskBridge::complete()` contract remains unchanged.
+Missing, malformed, different-major, and future-minor versions fail closed.
+Supported older protocols retain the original inline behavior and do not write
+the scheduled event.
+The built-in `WorkflowExecutor` explicitly selects the package's current
+protocol for the same decision.
+Direct service-control-plane calls never infer scheduler authority from caller
+workflow identity or from an unrelated idempotency transaction.
 
 `workflow_service_call_dispatches` is the durable scheduler record. Its state is
 `scheduled`, `claimed`, or `completed`; `attempt`, `max_attempts`, and

src/Providers/WorkflowServiceProvider.php

@@ -37,6 +37,7 @@
 use Workflow\V2\Contracts\OperatorObservabilityRepository;
 use Workflow\V2\Contracts\PreparedLocalActivityGroupTaskBridge;
 use Workflow\V2\Contracts\PreparedLocalActivityTaskBridge;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\RuntimeSignalControlPlane;
 use Workflow\V2\Contracts\SchedulerRole;
 use Workflow\V2\Contracts\ScheduleWorkflowStarter;
@@ -118,6 +119,7 @@ static function ($app): HistoryProjectionMaintenanceRole {
         $this->app->alias(WorkflowTaskBridge::class, CooperativeWorkflowTaskBridge::class);
         $this->app->alias(WorkflowTaskBridge::class, PreparedLocalActivityTaskBridge::class);
         $this->app->alias(WorkflowTaskBridge::class, PreparedLocalActivityGroupTaskBridge::class);
+        $this->app->alias(WorkflowTaskBridge::class, ProtocolAwareWorkflowTaskBridge::class);
 
         $this->app->singleton(ActivityTaskBridge::class, DefaultActivityTaskBridge::class);
 

src/V2/Contracts/ProtocolAwareWorkflowTaskBridge.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Contracts;
+
+/**
+ * Optional external-worker completion role.
+ *
+ * Existing WorkflowTaskBridge implementations need not provide this role.
+ * Hosts must check the bound bridge before advertising external worker
+ * completion support.
+ */
+interface ProtocolAwareWorkflowTaskBridge extends WorkflowTaskBridge
+{
+    /**
+     * Complete a workflow task under the authenticated worker protocol.
+     *
+     * @param list<array{type: string, ...}> $commands
+     * @return array{
+     *     completed: bool,
+     *     task_id: string,
+     *     workflow_run_id: string|null,
+     *     run_status: string|null,
+     *     created_task_ids: list<string>,
+     *     reason: string|null,
+     * }
+     */
+    public function completeForProtocol(string $taskId, array $commands, string $protocolVersion): array;
+}

src/V2/Support/DefaultServiceControlPlane.php

@@ -293,7 +293,7 @@ public function execute(
             ];
         }
 
-        $scheduled = $this->shouldScheduleHandler($call);
+        $scheduled = $this->shouldScheduleHandler($call, $options);
         $handler = $scheduled
             ? $this->scheduleHandler($call, $operation, $options)
             : $this->dispatchHandler($call, $operation, $boundaryRequest, $options);
@@ -1033,9 +1033,13 @@ private function scheduleHandler(
         ];
     }
 
-    private function shouldScheduleHandler(WorkflowServiceCall $call): bool
+    /**
+     * @param array<string, mixed> $options
+     */
+    private function shouldScheduleHandler(WorkflowServiceCall $call, array $options): bool
     {
-        return is_string($call->caller_workflow_run_id)
+        return ServiceCallScheduler::hasAfterCommitAuthority($options)
+            && is_string($call->caller_workflow_run_id)
             && $call->caller_workflow_run_id !== ''
             && $call->getConnection()
                 ->transactionLevel() > 0;
@@ -1383,10 +1387,6 @@ private function dispatchWorkflowUpdate(
         $commandOptions['wait_for'] ??= $call->operation_mode === ServiceCallOperationMode::Sync->value
             ? 'completed'
             : 'accepted';
-        if ($call->getConnection()->transactionLevel() > 0) {
-            $commandOptions['wait_for'] = UpdateWaitPolicy::WAIT_FOR_ACCEPTED;
-            unset($commandOptions['wait_timeout_seconds']);
-        }
 
         try {
             $result = $this->workflows()

src/V2/Support/DefaultWorkflowTaskBridge.php

@@ -21,6 +21,7 @@
 use Workflow\V2\Contracts\HistoryProjectionRole;
 use Workflow\V2\Contracts\PreparedCancellationScopeTaskBridge;
 use Workflow\V2\Contracts\PreparedLocalActivityGroupTaskBridge;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\ServiceControlPlane;
 use Workflow\V2\Contracts\WorkflowControlPlane;
 use Workflow\V2\Enums\ActivityAttemptStatus;
@@ -57,7 +58,7 @@
 use Workflow\V2\Models\WorkflowTimer;
 use Workflow\V2\Models\WorkflowUpdate;
 
-final class DefaultWorkflowTaskBridge implements CooperativeWorkflowTaskBridge, PreparedLocalActivityGroupTaskBridge, CancellationScopeAdmission, PreparedCancellationScopeTaskBridge
+final class DefaultWorkflowTaskBridge implements CooperativeWorkflowTaskBridge, PreparedLocalActivityGroupTaskBridge, CancellationScopeAdmission, PreparedCancellationScopeTaskBridge, ProtocolAwareWorkflowTaskBridge
 {
     public const POLL_BATCH_CAP = 100;
 
@@ -1229,6 +1230,32 @@
 
     public function complete(string $taskId, array $commands): array
     {
+        return $this->completeForProtocol($taskId, $commands, WorkerProtocolVersion::VERSION);
+    }
+
+    public function completeForProtocol(string $taskId, array $commands, string $protocolVersion): array
+    {
+        if (! WorkerProtocolVersion::isWellFormed($protocolVersion)) {
+            return [
+                'completed' => false,
+                'task_id' => $taskId,
+                'workflow_run_id' => null,
+                'run_status' => null,
+                'created_task_ids' => [],
+                'reason' => 'invalid_protocol_version',
+            ];
+        }
+        if (! WorkerProtocolVersion::isSupportedRequestVersion($protocolVersion)) {
+            return [
+                'completed' => false,
+                'task_id' => $taskId,
+                'workflow_run_id' => null,
+                'run_status' => null,
+                'created_task_ids' => [],
+                'reason' => 'unsupported_protocol_version',
+            ];
+        }
+
         $parsed = self::parseCommands($commands);
 
         if ($parsed === null && $commands !== []) {
@@ -1247,7 +1274,7 @@
         // charged as it is written, and a refusal rolls everything back,
         // leaving the task's lease as it was.
         try {
-            return self::meteredTransaction(function () use ($taskId, $parsed, $commands): array {
+            return self::meteredTransaction(function () use ($taskId, $parsed, $commands, $protocolVersion): array {
                 /** @var WorkflowTask|null $task */
                 $task = ConfiguredV2Models::query('task_model', WorkflowTask::class)
                     ->lockForUpdate()
@@ -1432,7 +1459,14 @@
                 );
 
                 foreach ($parsed['non_terminal'] as $command) {
-                    $sequence = $this->applyNonTerminalCommand($run, $task, $command, $sequence, $createdTaskIds);
+                    $sequence = $this->applyNonTerminalCommand(
+                        $run,
+                        $task,
+                        $command,
+                        $sequence,
+                        $createdTaskIds,
+                        $protocolVersion,
+                    );
                 }
 
                 $terminal = $parsed['terminal'];
@@ -1649,6 +1683,7 @@
                 $group,
                 $scopePrefix,
                 $checkpointName,
+                $protocolVersion,
             ): array {
                 /** @var WorkflowRun|null $run */
                 $run = ConfiguredV2Models::query('run_model', WorkflowRun::class)
@@ -1803,7 +1838,14 @@
                         ++$sequence;
                         continue;
                     }
-                    $sequence = $this->applyNonTerminalCommand($run, $task, $command, $sequence, $createdTaskIds);
+                    $sequence = $this->applyNonTerminalCommand(
+                        $run,
+                        $task,
+                        $command,
+                        $sequence,
+                        $createdTaskIds,
+                        $protocolVersion,
+                    );
                 }
                 // Command application can update task payload (for example signal
                 // consumption). Preserve that state rather than the old snapshot.
@@ -2529,6 +2571,7 @@
         array $command,
         int $sequence,
         array &$createdTaskIds,
+        string $protocolVersion,
     ): int {
         return match ($command['type']) {
             'cancel_selection_operation' => $this->applyCancelSelectionOperation($run, $task, $command, $sequence),
@@ -2541,6 +2584,7 @@
                 $command,
                 $sequence,
                 $createdTaskIds,
+                $protocolVersion,
             ),
             'complete_update' => $this->applyCompleteUpdate($run, $task, $command, $sequence),
             'fail_update' => $this->applyFailUpdate($run, $task, $command, $sequence),
@@ -3633,11 +3677,18 @@
         array $command,
         int $sequence,
         array &$createdTaskIds,
+        string $protocolVersion,
     ): int {
         $payloadCodec = is_string($command['payload_codec'] ?? null) && $command['payload_codec'] !== ''
             ? $command['payload_codec']
             : ($run->payload_codec ?? CodecRegistry::defaultCodec());
-        $options = $this->serviceOperationControlPlaneOptions($run, $command, $sequence, $payloadCodec);
+        $options = $this->serviceOperationControlPlaneOptions(
+            $run,
+            $command,
+            $sequence,
+            $payloadCodec,
+            $protocolVersion,
+        );
         $namespace = is_string($run->namespace) ? $run->namespace : null;
         $requestPayload = isset($command['request_payload']) && is_string($command['request_payload'])
             ? ExternalPayloads::externalizeForNamespace($command['request_payload'], $payloadCodec, $namespace)
@@ -3691,6 +3742,7 @@
         array $command,
         int $sequence,
         string $payloadCodec,
+        string $protocolVersion,
     ): array {
         if (! isset($command['service_call_id'])) {
             $command['idempotency_key'] ??= $this->defaultServiceOperationIdempotencyKey($run, $sequence);
@@ -3703,7 +3755,7 @@
             'workflow_sequence' => $sequence,
         ] + $metadata;
 
-        return WorkflowServiceOperationAuthority::options($run, $command) + array_filter([
+        $options = WorkflowServiceOperationAuthority::options($run, $command) + array_filter([
             'arguments' => $this->serviceOperationRequestArguments($command, $payloadCodec),
             'payload_blob' => self::nonEmptyString($command['request_payload'] ?? null),
             'payload_codec' => $payloadCodec,
@@ -3730,6 +3782,12 @@
             'metadata' => $metadata,
             'request_payload_reference' => self::nonEmptyString($command['request_payload_reference'] ?? null),
         ], static fn (mixed $value): bool => $value !== null);
+
+        if (WorkerProtocolVersion::supportsDurableServiceOperationScheduler($protocolVersion)) {
+            $options = ServiceCallScheduler::withAfterCommitAuthority($options);
+        }
+
+        return $options;
     }
 
     /**
@@ -3842,7 +3900,7 @@
         ?string $requestPayload,
         string $payloadCodec,
     ): array {
-        $responsePayload = $surface['response_payload'] ?? $surface['result'] ?? null;
+        $responsePayload = $surface['response_payload'] ?? $surface['result_envelope'] ?? $surface['result'] ?? null;
         $payload = [
             'sequence' => $sequence,
             'service_call_id' => self::nonEmptyString($surface['service_call_id'] ?? null)

src/V2/Support/ServiceCallScheduler.php

@@ -30,12 +30,35 @@ final class ServiceCallScheduler
 
     public const STATE_COMPLETED = 'completed';
 
+    private const AFTER_COMMIT_OPTION = '__schedule_service_call_after_commit';
+
     private const MIN_ORPHAN_GRACE_SECONDS = 3_600;
 
     private const ORPHAN_CLEANUP_BATCH_CAP = 100;
 
     private const EFFECT_TOKEN_OPTION = '__service_call_effect_token';
 
+    private static ?object $afterCommitAuthority = null;
+
+    /**
+     * @param array<string, mixed> $options
+     * @return array<string, mixed>
+     */
+    public static function withAfterCommitAuthority(array $options): array
+    {
+        $options[self::AFTER_COMMIT_OPTION] = self::afterCommitAuthority();
+
+        return $options;
+    }
+
+    /**
+     * @param array<string, mixed> $options
+     */
+    public static function hasAfterCommitAuthority(array $options): bool
+    {
+        return ($options[self::AFTER_COMMIT_OPTION] ?? null) === self::afterCommitAuthority();
+    }
+
     /**
      * @param array<string, mixed> $options
      */
@@ -45,6 +68,7 @@ public static function schedule(
         array $options,
         int $maxAttempts,
     ): void {
+        unset($options[self::AFTER_COMMIT_OPTION]);
         self::assertSharedConnection($call);
         $dispatch = self::dispatchQuery()
             ->lockForUpdate()
@@ -470,6 +494,11 @@ public static function attemptIdentity(string $serviceCallId, int $attempt): str
         return sprintf('workflow-service-call:%s:attempt:%d', $serviceCallId, $attempt);
     }
 
+    private static function afterCommitAuthority(): object
+    {
+        return self::$afterCommitAuthority ??= new \stdClass();
+    }
+
     private static function lockedCallerRun(WorkflowServiceCall $call): ?WorkflowRun
     {
         if (! is_string($call->caller_workflow_run_id) || $call->caller_workflow_run_id === '') {

src/V2/Support/WorkerProtocolVersion.php

@@ -338,6 +338,30 @@ public static function supportsLocalActivityAttemptReports(string $protocolVersi
         );
     }
 
+    public static function supportsDurableServiceOperationScheduler(string $protocolVersion): bool
+    {
+        return self::supportsFeatureVersion(
+            $protocolVersion,
+            self::DURABLE_SERVICE_OPERATION_SCHEDULER_MINIMUM_PROTOCOL_VERSION,
+        );
+    }
+
+    public static function isWellFormed(string $protocolVersion): bool
+    {
+        return preg_match('/\A(0|[1-9]\d*)\.(0|[1-9]\d*)\z/D', $protocolVersion) === 1;
+    }
+
+    public static function isSupportedRequestVersion(string $protocolVersion): bool
+    {
+        if (! self::isWellFormed($protocolVersion)) {
+            return false;
+        }
+        [$requestedMajor, $requestedMinor] = explode('.', $protocolVersion);
+        [$currentMajor, $currentMinor] = explode('.', self::VERSION);
+
+        return $requestedMajor === $currentMajor && (int) $requestedMinor <= (int) $currentMinor;
+    }
+
     /**
      * Version-gated workflow-task completion fields a worker may submit.
      *