diff --git a/docs/api-stability.md b/docs/api-stability.md
index 304a8ae4..16a5e401 100644
--- a/docs/api-stability.md
+++ b/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
diff --git a/docs/architecture/workflow-service-calls-architecture.md b/docs/architecture/workflow-service-calls-architecture.md
index 9ccd9213..6a175e52 100644
--- a/docs/architecture/workflow-service-calls-architecture.md
+++ b/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
diff --git a/src/Providers/WorkflowServiceProvider.php b/src/Providers/WorkflowServiceProvider.php
index 9efa531f..834f7e9f 100644
--- a/src/Providers/WorkflowServiceProvider.php
+++ b/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);
 
diff --git a/src/V2/Contracts/ProtocolAwareWorkflowTaskBridge.php b/src/V2/Contracts/ProtocolAwareWorkflowTaskBridge.php
new file mode 100644
index 00000000..7aa76458
--- /dev/null
+++ b/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;
+}
diff --git a/src/V2/Support/DefaultServiceControlPlane.php b/src/V2/Support/DefaultServiceControlPlane.php
index ea66faee..8c9f7f5c 100644
--- a/src/V2/Support/DefaultServiceControlPlane.php
+++ b/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()
diff --git a/src/V2/Support/DefaultWorkflowTaskBridge.php b/src/V2/Support/DefaultWorkflowTaskBridge.php
index 4875a75e..e2b12bc8 100644
--- a/src/V2/Support/DefaultWorkflowTaskBridge.php
+++ b/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 checkpointCancellationScopePrefix(
 
     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 @@ public function complete(string $taskId, array $commands): array
         // 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 @@ public function complete(string $taskId, array $commands): array
                 );
 
                 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 @@ private function checkpointCommands(
                 $group,
                 $scopePrefix,
                 $checkpointName,
+                $protocolVersion,
             ): array {
                 /** @var WorkflowRun|null $run */
                 $run = ConfiguredV2Models::query('run_model', WorkflowRun::class)
@@ -1803,7 +1838,14 @@ private function checkpointCommands(
                         ++$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 @@ private function applyNonTerminalCommand(
         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 @@ private function applyNonTerminalCommand(
                 $command,
                 $sequence,
                 $createdTaskIds,
+                $protocolVersion,
             ),
             'complete_update' => $this->applyCompleteUpdate($run, $task, $command, $sequence),
             'fail_update' => $this->applyFailUpdate($run, $task, $command, $sequence),
@@ -3633,11 +3677,18 @@ private function applyStartServiceOperation(
         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 @@ private function serviceOperationControlPlaneOptions(
         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 @@ private function serviceOperationControlPlaneOptions(
             '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 @@ private function serviceOperationControlPlaneOptions(
             '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 @@ private function serviceOperationEventPayload(
         ?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)
diff --git a/src/V2/Support/ServiceCallScheduler.php b/src/V2/Support/ServiceCallScheduler.php
index 53a2536e..8870fa5f 100644
--- a/src/V2/Support/ServiceCallScheduler.php
+++ b/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 === '') {
diff --git a/src/V2/Support/WorkerProtocolVersion.php b/src/V2/Support/WorkerProtocolVersion.php
index e46d9ccc..8442c83d 100644
--- a/src/V2/Support/WorkerProtocolVersion.php
+++ b/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.
      *
diff --git a/src/V2/Support/WorkflowExecutor.php b/src/V2/Support/WorkflowExecutor.php
index 8fa42278..9a4fb683 100644
--- a/src/V2/Support/WorkflowExecutor.php
+++ b/src/V2/Support/WorkflowExecutor.php
@@ -6026,7 +6026,7 @@ private function serviceOperationControlPlaneOptions(
             'workflow_sequence' => $sequence,
         ] + $metadata;
 
-        return WorkflowServiceOperationAuthority::options($run, $options + [
+        $options = WorkflowServiceOperationAuthority::options($run, $options + [
             'endpoint_name' => $call->endpointName,
             'service_name' => $call->serviceName,
             'operation_name' => $call->operationName,
@@ -6057,6 +6057,10 @@ private function serviceOperationControlPlaneOptions(
             'metadata' => $metadata,
             'request_payload_reference' => self::nonEmptyString($options['request_payload_reference'] ?? null),
         ], static fn (mixed $value): bool => $value !== null);
+
+        return WorkerProtocolVersion::supportsDurableServiceOperationScheduler(WorkerProtocolVersion::VERSION)
+            ? ServiceCallScheduler::withAfterCommitAuthority($options)
+            : $options;
     }
 
     private function defaultServiceOperationIdempotencyKey(WorkflowRun $run, int $sequence): string
@@ -6138,7 +6142,7 @@ private function serviceOperationEventPayload(
         string $payloadCodec,
     ): array {
         $metadata = $call->options?->metadata ?? [];
-        $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)
diff --git a/src/V2/Webhooks.php b/src/V2/Webhooks.php
index af64d85d..444c4e0c 100644
--- a/src/V2/Webhooks.php
+++ b/src/V2/Webhooks.php
@@ -14,6 +14,7 @@
 use Workflow\Auth\SignatureAuthenticator;
 use Workflow\Auth\TokenAuthenticator;
 use Workflow\Auth\WebhookAuthenticator;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\WorkflowControlPlane;
 use Workflow\V2\Contracts\WorkflowTaskBridge as WorkflowTaskBridgeContract;
 use Workflow\V2\Enums\CommandOutcome;
@@ -32,6 +33,7 @@
 use Workflow\V2\Support\TaskPriority;
 use Workflow\V2\Support\TypeRegistry;
 use Workflow\V2\Support\UpdateWaitPolicy;
+use Workflow\V2\Support\WorkerProtocolVersion;
 use Workflow\V2\Support\WorkflowInstanceId;
 
 final class Webhooks
@@ -205,7 +207,7 @@ public static function routes(array $workflows, ?string $basePath = null): void
         ) {
             $request = self::validateAuth($request);
 
-            return self::workflowTaskCompleteResponse($taskId, self::resolveWorkflowTaskCommands($request->all()));
+            return self::workflowTaskCompleteResponse($taskId, $request);
         })->name('workflows.v2.workflow-tasks.complete');
 
         Route::post("{$basePath}/workflow-tasks/{taskId}/fail", static function (
@@ -1114,15 +1116,31 @@ private static function workflowTaskExecuteResponse(string $taskId)
         return response()->json($payload, $status);
     }
 
-    /**
-     * @param list<array{type: string, ...}> $commands
-     */
-    private static function workflowTaskCompleteResponse(string $taskId, array $commands)
+    private static function workflowTaskCompleteResponse(string $taskId, Request $request)
     {
-        $payload = self::workflowTaskBridge()->complete($taskId, $commands);
+        $header = $request->header('X-Durable-Workflow-Protocol-Version');
+        $protocolVersion = is_string($header) ? $header : '';
+        if (! WorkerProtocolVersion::isWellFormed($protocolVersion)) {
+            return response()->json(self::protocolCompletionRefusal($taskId, 'invalid_protocol_version'), 400);
+        }
+        if (! WorkerProtocolVersion::isSupportedRequestVersion($protocolVersion)) {
+            return response()->json(self::protocolCompletionRefusal($taskId, 'unsupported_protocol_version'), 400);
+        }
+
+        $bridge = self::workflowTaskBridge();
+        if (! $bridge instanceof ProtocolAwareWorkflowTaskBridge) {
+            return response()->json(
+                self::protocolCompletionRefusal($taskId, 'protocol_aware_completion_unavailable'),
+                501,
+            );
+        }
+
+        $commands = self::resolveWorkflowTaskCommands($request->all());
+        $payload = $bridge->completeForProtocol($taskId, $commands, $protocolVersion);
         $status = match ($payload['reason']) {
             null => 200,
             'task_not_found' => 404,
+            'invalid_protocol_version', 'unsupported_protocol_version' => 400,
             'invalid_commands' => 422,
             default => 409,
         };
@@ -1130,6 +1148,28 @@ private static function workflowTaskCompleteResponse(string $taskId, array $comm
         return response()->json($payload, $status);
     }
 
+    /**
+     * @return array{
+     *     completed: false,
+     *     task_id: string,
+     *     workflow_run_id: null,
+     *     run_status: null,
+     *     created_task_ids: list<string>,
+     *     reason: string,
+     * }
+     */
+    private static function protocolCompletionRefusal(string $taskId, string $reason): array
+    {
+        return [
+            'completed' => false,
+            'task_id' => $taskId,
+            'workflow_run_id' => null,
+            'run_status' => null,
+            'created_task_ids' => [],
+            'reason' => $reason,
+        ];
+    }
+
     /**
      * @param array<string, mixed>|string $failure
      */
diff --git a/src/V2/WorkflowTaskBridge.php b/src/V2/WorkflowTaskBridge.php
index 667d21cc..c909f905 100644
--- a/src/V2/WorkflowTaskBridge.php
+++ b/src/V2/WorkflowTaskBridge.php
@@ -4,7 +4,9 @@
 
 namespace Workflow\V2;
 
+use LogicException;
 use Throwable;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\WorkflowTaskBridge as WorkflowTaskBridgeContract;
 use Workflow\V2\Support\WorkerProtocolVersion;
 
@@ -206,6 +208,24 @@ public static function complete(string $taskId, array $commands): array
         return self::resolve()->complete($taskId, $commands);
     }
 
+    /**
+     * Complete a workflow task under the authenticated external-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 static function completeForProtocol(string $taskId, array $commands, string $protocolVersion): array
+    {
+        return self::resolveProtocolAware()->completeForProtocol($taskId, $commands, $protocolVersion);
+    }
+
     /**
      * @param Throwable|array<string, mixed>|string $failure
      * @return array{recorded: bool, task_id: string, reason: string|null, next_task_id: string|null}
@@ -257,4 +277,14 @@ private static function resolve(): WorkflowTaskBridgeContract
     {
         return app(WorkflowTaskBridgeContract::class);
     }
+
+    private static function resolveProtocolAware(): ProtocolAwareWorkflowTaskBridge
+    {
+        $bridge = self::resolve();
+        if (! $bridge instanceof ProtocolAwareWorkflowTaskBridge) {
+            throw new LogicException('The configured workflow task bridge does not support protocol-aware completion.');
+        }
+
+        return $bridge;
+    }
 }
diff --git a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
index 9e735df9..d8c948c0 100644
--- a/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
+++ b/tests/Feature/V2/V2EmbeddedReplayRegressionCorpusTest.php
@@ -11,6 +11,7 @@
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\Queue;
 use Illuminate\Validation\ValidationException;
+use Mockery;
 use PHPUnit\Framework\Attributes\PreserveGlobalState;
 use PHPUnit\Framework\Attributes\RunInSeparateProcess;
 use Symfony\Component\Uid\Ulid;
@@ -29,6 +30,10 @@
 use Workflow\V2\Activity;
 use Workflow\V2\Contracts\ActivityTaskBridge;
 use Workflow\V2\Contracts\HistoryProjectionRole;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
+use Workflow\V2\Contracts\ServiceControlPlane;
+use Workflow\V2\Contracts\WorkflowControlPlane;
+use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
 use Workflow\V2\Enums\CancellationPolicy;
 use Workflow\V2\Enums\HistoryEventType;
@@ -41,6 +46,7 @@
 use Workflow\V2\Exceptions\VersionRangeMismatchException;
 use Workflow\V2\Exceptions\WorkflowEntryArgumentsMismatchException;
 use Workflow\V2\Jobs\RunActivityTask;
+use Workflow\V2\Jobs\RunServiceCall;
 use Workflow\V2\Jobs\RunTimerTask;
 use Workflow\V2\Jobs\RunWorkflowTask;
 use Workflow\V2\Models\ActivityAttempt;
@@ -55,6 +61,11 @@
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowRunSummary;
 use Workflow\V2\Models\WorkflowSearchAttribute;
+use Workflow\V2\Models\WorkflowService;
+use Workflow\V2\Models\WorkflowServiceCall;
+use Workflow\V2\Models\WorkflowServiceCallDispatch;
+use Workflow\V2\Models\WorkflowServiceEndpoint;
+use Workflow\V2\Models\WorkflowServiceOperation;
 use Workflow\V2\Models\WorkflowSignal;
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Models\WorkflowTimelineEntry;
@@ -66,6 +77,7 @@
 use Workflow\V2\Support\CancellationScopeRequests;
 use Workflow\V2\Support\ConditionWaits;
 use Workflow\V2\Support\DefaultHistoryProjectionRole;
+use Workflow\V2\Support\DefaultServiceControlPlane;
 use Workflow\V2\Support\DefaultWorkflowTaskBridge;
 use Workflow\V2\Support\EmbeddedV2HistoryImport;
 use Workflow\V2\Support\HistoryExport;
@@ -73,6 +85,8 @@
 use Workflow\V2\Support\ReplayDiff;
 use Workflow\V2\Support\RunActivityView;
 use Workflow\V2\Support\RunTimelineProjector;
+use Workflow\V2\Support\ServiceCallPrincipal;
+use Workflow\V2\Support\ServiceCallScheduler;
 use Workflow\V2\Support\WorkflowCommandNormalizer;
 use Workflow\V2\Support\WorkflowFiberRunner;
 use Workflow\V2\Support\WorkflowReplayer;
@@ -347,6 +361,14 @@ public function testFixturesExecuteThroughDeclaredReplayConsumers(): void
                 continue;
             }
 
+            if (str_starts_with($fixture['id'], 'workflow-service-operation-protocol-')) {
+                $this->assertServiceOperationProtocolPersistence($fixture);
+                if ($fixture['protocol_version'] === '1.21') {
+                    $this->assertServiceOperationProtocolPersistence($fixture, embedded: true);
+                }
+                continue;
+            }
+
             if ($fixture['id'] === 'nexus-waiting-caller-terminal-delivery') {
                 $this->clearWorkflowState();
                 $this->assertWaitingNexusCallerReceivesResult(
@@ -630,6 +652,217 @@ private function assertExternalWorkerPendingCountLimitRollsBackCompletion(array
         }
     }
 
+    /**
+     * @param array<string, mixed> $fixture
+     */
+    private function assertServiceOperationProtocolPersistence(array $fixture, bool $embedded = false): void
+    {
+        $this->clearWorkflowState();
+        Queue::fake();
+        $previousControlPlane = $this->app->make(ServiceControlPlane::class);
+        $previousAuthorizer = $this->installNexusServiceAuthority();
+        $scheduled = $fixture['protocol_version'] === '1.21';
+        $failed = str_ends_with($fixture['id'], '-failure');
+        $targetInstance = WorkflowInstance::query()->create([
+            'id' => 'protocol-target',
+            'namespace' => 'service',
+            'workflow_class' => 'protocol-service',
+            'workflow_type' => 'protocol-service',
+            'run_count' => 1,
+        ]);
+        $targetRun = WorkflowRun::query()->create([
+            'workflow_instance_id' => $targetInstance->id,
+            'namespace' => 'service',
+            'run_number' => 1,
+            'workflow_class' => 'protocol-service',
+            'workflow_type' => 'protocol-service',
+            'status' => $failed ? RunStatus::Failed->value : RunStatus::Completed->value,
+            'payload_codec' => 'avro',
+            'connection' => 'database',
+            'queue' => 'default',
+            'started_at' => now(),
+            'closed_at' => now(),
+        ]);
+        WorkflowHistoryEvent::record(
+            $targetRun,
+            $failed ? HistoryEventType::WorkflowFailed : HistoryEventType::WorkflowCompleted,
+            $failed ? [
+                'exception_type' => 'invoice_declined',
+                'message' => 'The invoice approval was denied.',
+            ] : [
+                'output' => Serializer::serializeWithCodec('avro', $fixture['terminal_result']),
+                'payload_codec' => 'avro',
+            ],
+        );
+        $handlerCalls = 0;
+        $handler = Mockery::mock(WorkflowControlPlane::class);
+        $handler->shouldReceive('start')
+            ->once()
+            ->withArgs(
+                function (string $workflowType, ?string $instanceId, array $options) use (&$handlerCalls): bool {
+                    ++$handlerCalls;
+                    $this->assertGreaterThan(0, DB::transactionLevel());
+                    return $workflowType === 'protocol-service';
+                },
+            )->andReturn([
+                'started' => true,
+                'workflow_instance_id' => $targetInstance->id,
+                'workflow_run_id' => $targetRun->id,
+            ]);
+        $this->app->instance(ServiceControlPlane::class, new DefaultServiceControlPlane($handler));
+
+        try {
+            $endpoint = WorkflowServiceEndpoint::query()->create([
+                'namespace' => 'service',
+                'endpoint_name' => 'php-service',
+            ]);
+            $service = WorkflowService::query()->create([
+                'namespace' => 'service',
+                'workflow_service_endpoint_id' => $endpoint->id,
+                'service_name' => 'greeter',
+            ]);
+            WorkflowServiceOperation::query()->create([
+                'namespace' => 'service',
+                'workflow_service_endpoint_id' => $endpoint->id,
+                'workflow_service_id' => $service->id,
+                'operation_name' => 'greet',
+                'operation_mode' => 'sync',
+                'handler_binding_kind' => 'workflow_run',
+                'handler_target_reference' => 'protocol-service',
+            ]);
+            $workflow = $fixture['workflow'];
+            $stub = WorkflowStub::make(
+                $workflow['type'],
+                sprintf('regression-corpus-protocol-%d', ++$this->workflowNumber),
+                'caller',
+            );
+            $stub->start(...$workflow['arguments']);
+            $run = WorkflowRun::query()->findOrFail($stub->runId());
+            $task = $run->tasks()
+                ->where('task_type', TaskType::Workflow->value)
+                ->where('status', TaskStatus::Ready->value)
+                ->sole();
+            $bridge = $this->app->make(WorkflowTaskBridge::class);
+            $claim = $bridge->claimStatus($task->id, 'regression-corpus-protocol-worker');
+            $this->assertTrue($claim['claimed'], json_encode($claim, JSON_THROW_ON_ERROR));
+            $history = $bridge->historyPayload($task->id);
+            $this->assertNotNull($history);
+            $authored = WorkflowFiberRunner::forClass(
+                $workflow['type'],
+                $run->workflow_instance_id,
+                $run->id,
+                $workflow['arguments'],
+                $workflow['payload_codec'],
+                $history['history_events'],
+            )->step();
+            $this->assertSame(['start_service_operation'], array_column($authored->commands, 'type'));
+            Queue::fake();
+
+            DB::transaction(function () use ($bridge, $task, $authored, $fixture, $embedded, $scheduled): void {
+                $result = $embedded
+                    ? $bridge->execute($task->id)
+                    : ($bridge instanceof ProtocolAwareWorkflowTaskBridge
+                        ? $bridge->completeForProtocol($task->id, $authored->commands, $fixture['protocol_version'])
+                        : $bridge->complete($task->id, $authored->commands));
+                $this->assertTrue(
+                    $result[$embedded ? 'executed' : 'completed'],
+                    json_encode($result, JSON_THROW_ON_ERROR)
+                );
+                $this->assertSame($scheduled ? 1 : 0, WorkflowServiceCallDispatch::query()->count());
+                Queue::assertNothingPushed();
+            });
+
+            $call = WorkflowServiceCall::query()->where('caller_workflow_run_id', $run->id)->sole();
+            $this->assertSame($scheduled ? 'accepted' : ($failed ? 'failed' : 'completed'), $call->status);
+            $event = $run->historyEvents()
+                ->whereIn(
+                    'event_type',
+                    ['ServiceCallCompleted', 'ServiceCallFailed', 'ServiceOperationScheduled']
+                )->sole();
+            $this->assertSame($fixture['history'][1]['event_type'], $event->event_type->value);
+            $this->assertSame(
+                $scheduled ? 1 : 0,
+                $run->historyEvents()
+                    ->where('event_type', 'ServiceOperationScheduled')
+                    ->count()
+            );
+            if ($scheduled) {
+                $this->assertFalse($event->payload['workflow_resume_on_admission']);
+                $dispatch = WorkflowServiceCallDispatch::query()->sole();
+                $this->assertSame($call->id, $dispatch->service_call_id);
+                $this->assertSame('scheduled', $dispatch->state);
+                $this->assertFalse(ServiceCallScheduler::hasAfterCommitAuthority($dispatch->dispatch_options));
+                Queue::assertPushed(RunServiceCall::class, 1);
+                Queue::assertPushed(
+                    RunServiceCall::class,
+                    static fn (RunServiceCall $job): bool => $job->serviceCallId === $call->id
+                );
+            } else {
+                Queue::assertNothingPushed();
+                if ($failed) {
+                    $this->assertSame('invoice_declined', $event->payload['exception_type']);
+                    $this->assertSame('The invoice approval was denied.', $event->payload['message']);
+                }
+            }
+            $cold = WorkflowFiberRunner::forClass(
+                $workflow['type'],
+                $run->workflow_instance_id,
+                $run->id,
+                $workflow['arguments'],
+                $workflow['payload_codec'],
+                HistoryExport::forRun($run->fresh())['history_events'],
+            )->step();
+            $this->assertStepMatches($fixture['expected'], $cold, $fixture['id'] . ' persisted history');
+            $this->assertSame(TaskStatus::Completed, $task->fresh()->status);
+            $this->assertSame($scheduled ? 0 : 1, $handlerCalls);
+            if ($scheduled) {
+                if ($embedded) {
+                    // Cold embedded replay of the retained scheduled history
+                    // must wait without dispatching its handler a second time.
+                    $replayRun = $this->createRunFromFixture($fixture);
+                    Queue::fake();
+                    $this->runReadyWorkflowTask($replayRun);
+                    $this->assertSame(RunStatus::Waiting, $replayRun->fresh()->status);
+                    $this->assertSame(
+                        1,
+                        $replayRun->historyEvents()
+                            ->where('event_type', 'ServiceOperationScheduled')
+                            ->count()
+                    );
+                    $this->assertSame(1, WorkflowServiceCallDispatch::query()->count());
+                    $this->assertSame(0, $handlerCalls);
+                    Queue::assertNothingPushed();
+                }
+
+                // The same 1.21 host must still execute a direct HTTP-style
+                // idempotent call inline, despite carrying the caller identity.
+                Queue::fake();
+                $direct = $this->app->make(ServiceControlPlane::class)->execute('php-service', 'greeter', 'greet', [
+                    'namespace' => 'service',
+                    'caller_namespace' => 'caller',
+                    'caller_workflow_instance_id' => $run->workflow_instance_id,
+                    'caller_workflow_run_id' => $run->id,
+                    'principal' => new ServiceCallPrincipal(subject: 'workflow:' . $run->id, method: 'workflow'),
+                    'idempotency_key' => 'direct-protocol-call',
+                    'wait_for' => 'completed',
+                ]);
+                $this->assertSame('completed', $direct['status']);
+                $this->assertSame($fixture['terminal_result'], Serializer::unserializeWithCodec(
+                    $direct['result_envelope']['codec'],
+                    $direct['result_envelope']['blob'],
+                ));
+                $this->assertSame(1, $handlerCalls);
+                $this->assertSame(1, WorkflowServiceCallDispatch::query()->count());
+                Queue::assertNothingPushed();
+            }
+        } finally {
+            $this->app->instance(ServiceControlPlane::class, $previousControlPlane);
+            if ($previousAuthorizer !== null) {
+                $this->app->instance(WorkflowServiceOperationAuthorizer::class, $previousAuthorizer);
+            }
+        }
+    }
+
     /**
      * @return array<string, int>
      */
@@ -1897,6 +2130,7 @@ public function handle(string $name): string
     private function clearWorkflowState(): void
     {
         foreach ([
+            'workflow_service_call_dispatches',
             'workflow_service_calls',
             'workflow_service_operations',
             'workflow_services',
diff --git a/tests/Feature/V2/V2WebhookWorkflowTest.php b/tests/Feature/V2/V2WebhookWorkflowTest.php
index a7bba8f1..4eb10074 100644
--- a/tests/Feature/V2/V2WebhookWorkflowTest.php
+++ b/tests/Feature/V2/V2WebhookWorkflowTest.php
@@ -4,7 +4,10 @@
 
 namespace Tests\Feature\V2;
 
+use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\Queue;
+use Mockery;
+use PHPUnit\Framework\Attributes\DataProvider;
 use Tests\Fixtures\V2\TestAliasedUpdateWorkflow;
 use Tests\Fixtures\V2\TestConfiguredGreetingWorkflow;
 use Tests\Fixtures\V2\TestGreetingWorkflow;
@@ -17,6 +20,8 @@
 use Tests\TestCase;
 use Workflow\Serializers\CodecRegistry;
 use Workflow\Serializers\Serializer;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
+use Workflow\V2\Contracts\WorkflowTaskBridge as WorkflowTaskBridgeContract;
 use Workflow\V2\Enums\RunStatus;
 use Workflow\V2\Enums\TaskStatus;
 use Workflow\V2\Enums\TaskType;
@@ -38,6 +43,7 @@
 use Workflow\V2\Support\LocalFilesystemExternalPayloadStorage;
 use Workflow\V2\Support\RunDetailView;
 use Workflow\V2\Support\RunSummaryProjector;
+use Workflow\V2\Support\WorkerProtocolVersion;
 use Workflow\V2\Support\WorkflowInstanceId;
 use Workflow\V2\Webhooks;
 use Workflow\V2\WorkflowStub;
@@ -3062,7 +3068,10 @@ public function testWorkflowTaskCompleteWebhookAppliesNonTerminalCommands(): voi
             'lease_owner' => 'server-complete-worker',
         ])->assertOk();
 
-        $response = $this->postJson("/webhooks/workflow-tasks/{$task->id}/complete", [
+        $response = $this->withHeader(
+            'X-Durable-Workflow-Protocol-Version',
+            WorkerProtocolVersion::VERSION,
+        )->postJson("/webhooks/workflow-tasks/{$task->id}/complete", [
             'commands' => [
                 [
                     'type' => 'schedule_activity',
@@ -3114,7 +3123,10 @@ public function testWorkflowTaskCompleteWebhookAppliesMetadataCommands(): void
             'lease_owner' => 'server-metadata-worker',
         ])->assertOk();
 
-        $response = $this->postJson("/webhooks/workflow-tasks/{$task->id}/complete", [
+        $response = $this->withHeader(
+            'X-Durable-Workflow-Protocol-Version',
+            WorkerProtocolVersion::VERSION,
+        )->postJson("/webhooks/workflow-tasks/{$task->id}/complete", [
             'commands' => [
                 [
                     'type' => 'record_side_effect',
@@ -3196,7 +3208,10 @@ public function testWorkflowTaskCompleteWebhookAppliesTerminalCommand(): void
             'lease_owner' => 'server-terminal-worker',
         ])->assertOk();
 
-        $response = $this->postJson("/webhooks/workflow-tasks/{$task->id}/complete", [
+        $response = $this->withHeader(
+            'X-Durable-Workflow-Protocol-Version',
+            WorkerProtocolVersion::VERSION,
+        )->postJson("/webhooks/workflow-tasks/{$task->id}/complete", [
             'commands' => [
                 [
                     'type' => 'complete_workflow',
@@ -3227,6 +3242,8 @@ public function testWorkflowTaskCompleteWebhookReturns422ForEmptyCommands(): voi
     {
         $response = $this->postJson('/webhooks/workflow-tasks/some-task/complete', [
             'commands' => [],
+        ], [
+            'X-Durable-Workflow-Protocol-Version' => WorkerProtocolVersion::VERSION,
         ]);
 
         $response
@@ -3234,6 +3251,103 @@ public function testWorkflowTaskCompleteWebhookReturns422ForEmptyCommands(): voi
             ->assertJsonValidationErrors(['commands']);
     }
 
+    #[DataProvider('supportedWorkflowTaskCompletionProtocols')]
+    public function testWorkflowTaskCompleteWebhookForwardsTheWorkerProtocolHeader(string $protocolVersion): void
+    {
+        $commands = [[
+            'type' => 'complete_workflow',
+        ]];
+        $bridge = Mockery::mock(ProtocolAwareWorkflowTaskBridge::class);
+        $bridge->shouldReceive('completeForProtocol')
+            ->once()
+            ->with('protocol-task', $commands, $protocolVersion)
+            ->andReturn([
+                'completed' => true,
+                'task_id' => 'protocol-task',
+                'workflow_run_id' => 'protocol-run',
+                'run_status' => 'completed',
+                'created_task_ids' => [],
+                'reason' => null,
+            ]);
+        $this->app->instance(WorkflowTaskBridgeContract::class, $bridge);
+
+        $this->withHeader('X-Durable-Workflow-Protocol-Version', $protocolVersion)
+            ->postJson('/webhooks/workflow-tasks/protocol-task/complete', [
+                'commands' => $commands,
+            ])
+            ->assertOk()
+            ->assertJsonPath('completed', true);
+    }
+
+    /**
+     * @return iterable<string, array{string}>
+     */
+    public static function supportedWorkflowTaskCompletionProtocols(): iterable
+    {
+        yield 'inline protocol' => ['1.20'];
+        yield 'scheduler protocol' => ['1.21'];
+    }
+
+    #[DataProvider('unsupportedWorkflowTaskCompletionProtocols')]
+    public function testWorkflowTaskCompleteWebhookRejectsUnsupportedWorkerProtocols(
+        ?string $protocolVersion,
+        string $reason,
+    ): void {
+        $bridge = Mockery::mock(ProtocolAwareWorkflowTaskBridge::class);
+        $bridge->shouldNotReceive('completeForProtocol');
+        $this->app->instance(WorkflowTaskBridgeContract::class, $bridge);
+        if ($protocolVersion !== null) {
+            $this->withHeader('X-Durable-Workflow-Protocol-Version', $protocolVersion);
+        }
+        $writes = [];
+        DB::listen(static function ($query) use (&$writes): void {
+            if (preg_match('/\A\s*(insert|update|delete|replace|create|alter|drop)\b/i', $query->sql) === 1) {
+                $writes[] = $query->sql;
+            }
+        });
+
+        $this->postJson('/webhooks/workflow-tasks/protocol-task/complete', [
+            // Protocol refusal must precede command parsing as well as writes.
+            'commands' => 'invalid-command-document',
+        ])
+            ->assertStatus(400)
+            ->assertJsonPath('completed', false)
+            ->assertJsonPath('task_id', 'protocol-task')
+            ->assertJsonPath('reason', $reason);
+        $this->assertSame([], $writes);
+        Queue::assertNothingPushed();
+    }
+
+    /**
+     * @return iterable<string, array{?string, string}>
+     */
+    public static function unsupportedWorkflowTaskCompletionProtocols(): iterable
+    {
+        yield 'missing' => [null, 'invalid_protocol_version'];
+        yield 'empty' => ['', 'invalid_protocol_version'];
+        yield 'malformed patch version' => ['1.21.0', 'invalid_protocol_version'];
+        yield 'future minor' => ['1.22', 'unsupported_protocol_version'];
+        yield 'different major' => ['2.0', 'unsupported_protocol_version'];
+        yield 'leading zero major' => ['01.21', 'invalid_protocol_version'];
+        yield 'leading zero minor' => ['1.020', 'invalid_protocol_version'];
+    }
+
+    public function testWorkflowTaskCompleteWebhookFailsClosedWithoutTheOptionalProtocolRole(): void
+    {
+        $this->app->instance(WorkflowTaskBridgeContract::class, Mockery::mock(WorkflowTaskBridgeContract::class));
+
+        $this->withHeader('X-Durable-Workflow-Protocol-Version', WorkerProtocolVersion::VERSION)
+            ->postJson('/webhooks/workflow-tasks/protocol-task/complete', [
+                'commands' => [[
+                    'type' => 'complete_workflow',
+                ]],
+            ])
+            ->assertStatus(501)
+            ->assertJsonPath('completed', false)
+            ->assertJsonPath('task_id', 'protocol-task')
+            ->assertJsonPath('reason', 'protocol_aware_completion_unavailable');
+    }
+
     public function testWorkflowTaskCompleteWebhookReturns422ForInvalidOpenWaitTimeouts(): void
     {
         $cases = [
@@ -3249,7 +3363,10 @@ public function testWorkflowTaskCompleteWebhookReturns422ForInvalidOpenWaitTimeo
         ];
 
         foreach ($cases as $label => $command) {
-            $response = $this->postJson("/webhooks/workflow-tasks/invalid-timeout-{$label}/complete", [
+            $response = $this->withHeader(
+                'X-Durable-Workflow-Protocol-Version',
+                WorkerProtocolVersion::VERSION,
+            )->postJson("/webhooks/workflow-tasks/invalid-timeout-{$label}/complete", [
                 'commands' => [$command],
             ]);
 
@@ -3262,7 +3379,9 @@ public function testWorkflowTaskCompleteWebhookReturns422ForInvalidOpenWaitTimeo
 
     public function testWorkflowTaskCompleteWebhookReturns422ForMissingCommands(): void
     {
-        $response = $this->postJson('/webhooks/workflow-tasks/some-task/complete', []);
+        $response = $this->postJson('/webhooks/workflow-tasks/some-task/complete', [], [
+            'X-Durable-Workflow-Protocol-Version' => WorkerProtocolVersion::VERSION,
+        ]);
 
         $response
             ->assertStatus(422)
diff --git a/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-120-failure.json b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-120-failure.json
new file mode 100644
index 00000000..6d94a9f8
--- /dev/null
+++ b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-120-failure.json
@@ -0,0 +1,38 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "workflow-service-operation-protocol-120-failure",
+  "protocol_version": "1.20",
+  "bindings": ["php"],
+  "consumers": ["workflow-fiber-runner"],
+  "workflow": {
+    "type": "Tests\\Fixtures\\V2\\TestNexusDeadlineCallerWorkflow",
+    "arguments": ["world"],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {"sequence": 1, "event_type": "WorkflowStarted", "payload": {}},
+    {
+      "sequence": 2,
+      "event_type": "ServiceCallFailed",
+      "payload": {
+        "sequence": 1,
+        "service_call_id": "protocol-120-failure-call",
+        "endpoint_name": "php-service",
+        "service_name": "greeter",
+        "operation_name": "greet",
+        "status": "failed",
+        "outcome": "handler_failed",
+        "exception_class": "RuntimeException",
+        "exception_type": "invoice_declined",
+        "message": "The invoice approval was denied."
+      }
+    }
+  ],
+  "expected": {
+    "completed": true,
+    "result": {"failure_message": "The invoice approval was denied."},
+    "commands": [{"type": "complete_workflow"}]
+  },
+  "terminal_result": {"failure_message": "The invoice approval was denied."}
+}
diff --git a/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-120-inline.json b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-120-inline.json
new file mode 100644
index 00000000..bc529ee1
--- /dev/null
+++ b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-120-inline.json
@@ -0,0 +1,53 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "workflow-service-operation-protocol-120-inline",
+  "protocol_version": "1.20",
+  "bindings": ["php"],
+  "consumers": ["workflow-fiber-runner"],
+  "workflow": {
+    "type": "Tests\\Fixtures\\V2\\TestNexusWaitingCallerWorkflow",
+    "arguments": ["world"],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "sequence": 1,
+      "event_type": "WorkflowStarted",
+      "payload": {}
+    },
+    {
+      "sequence": 2,
+      "event_type": "ServiceCallCompleted",
+      "payload": {
+        "sequence": 1,
+        "service_call_id": "protocol-120-inline-call",
+        "endpoint_name": "php-service",
+        "service_name": "greeter",
+        "operation_name": "greet",
+        "operation_mode": "sync",
+        "wait_for": "completed",
+        "status": "completed",
+        "outcome": "completed",
+        "response_payload": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNw4CEGdyZWV0aW5nChZoZWxsbyB3b3JsZAA="
+        }
+      }
+    }
+  ],
+  "expected": {
+    "completed": true,
+    "result": {
+      "greeting": "hello world"
+    },
+    "commands": [
+      {
+        "type": "complete_workflow"
+      }
+    ]
+  },
+  "terminal_result": {
+    "greeting": "hello world"
+  }
+}
diff --git a/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-121-scheduled.json b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-121-scheduled.json
new file mode 100644
index 00000000..5be1652a
--- /dev/null
+++ b/tests/Fixtures/V2/ReplayRegression/workflow-service-operation-protocol-121-scheduled.json
@@ -0,0 +1,34 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "workflow-service-operation-protocol-121-scheduled",
+  "protocol_version": "1.21",
+  "bindings": ["php"],
+  "consumers": ["workflow-fiber-runner", "workflow-executor"],
+  "workflow": {
+    "type": "Tests\\Fixtures\\V2\\TestNexusWaitingCallerWorkflow",
+    "arguments": ["world"],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {"sequence": 1, "event_type": "WorkflowStarted", "payload": {}},
+    {
+      "sequence": 2,
+      "event_type": "ServiceOperationScheduled",
+      "payload": {
+        "sequence": 1,
+        "service_call_id": "protocol-121-scheduled-call",
+        "endpoint_name": "php-service",
+        "service_name": "greeter",
+        "operation_name": "greet",
+        "operation_mode": "sync",
+        "wait_for": "accepted",
+        "workflow_resume_on_admission": false,
+        "status": "accepted",
+        "outcome": "accepted"
+      }
+    }
+  ],
+  "expected": {"completed": false, "result": null, "commands": []},
+  "terminal_result": {"greeting": "hello world"}
+}
diff --git a/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php b/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php
index 9209afa7..e8765ba7 100644
--- a/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php
+++ b/tests/Unit/V2/DefaultWorkflowTaskBridgeServiceOperationTest.php
@@ -8,10 +8,13 @@
 use Illuminate\Support\Carbon;
 use Illuminate\Support\Facades\Queue;
 use LogicException;
+use Mockery;
 use PHPUnit\Framework\Attributes\DataProvider;
+use ReflectionMethod;
 use Tests\TestCase;
 use Workflow\Serializers\Serializer;
 use Workflow\V2\Attributes\Type;
+use Workflow\V2\Contracts\ProtocolAwareWorkflowTaskBridge;
 use Workflow\V2\Contracts\ServiceControlPlane;
 use Workflow\V2\Contracts\WorkflowServiceOperationAuthorizer;
 use Workflow\V2\Contracts\WorkflowTaskBridge;
@@ -26,13 +29,14 @@
 use Workflow\V2\Models\WorkflowTask;
 use Workflow\V2\Support\RunServiceOperationAuthorizer;
 use Workflow\V2\Support\ServiceCallPrincipal;
+use Workflow\V2\Support\ServiceCallScheduler;
 use Workflow\V2\Support\WorkerCompatibilityFleet;
 use Workflow\V2\Workflow;
 use Workflow\V2\WorkflowStub;
 
 final class DefaultWorkflowTaskBridgeServiceOperationTest extends TestCase
 {
-    private WorkflowTaskBridge $bridge;
+    private ProtocolAwareWorkflowTaskBridge $bridge;
 
     private WorkflowRun $run;
 
@@ -57,7 +61,7 @@ protected function setUp(): void
         $this->assertTrue($workflow->attemptStart()->accepted());
         $this->run = WorkflowRun::query()->findOrFail($workflow->runId());
         $this->task = WorkflowTask::query()->where('workflow_run_id', $this->run->id)->sole();
-        $this->bridge = $this->app->make(WorkflowTaskBridge::class);
+        $this->bridge = $this->app->make(ProtocolAwareWorkflowTaskBridge::class);
         $claim = $this->bridge->claimStatus($this->task->id, 'service-command-worker');
         $this->assertTrue($claim['claimed'], json_encode($claim, JSON_THROW_ON_ERROR));
         $this->assertSame(TaskStatus::Leased, $this->task->fresh()->status);
@@ -72,6 +76,115 @@ protected function tearDown(): void
         parent::tearDown();
     }
 
+    #[DataProvider('serviceSchedulerProtocols')]
+    public function testServiceSchedulerOptionIsVersionGated(string $protocolVersion, bool $scheduled): void
+    {
+        $service = new BridgeServiceControlPlaneProbe([
+            'service_call_id' => 'version-gated-call',
+            'accepted' => true,
+            'status' => 'completed',
+            'result' => 'done',
+        ]);
+        $this->app->instance(ServiceControlPlane::class, $service);
+        $command = [
+            'type' => 'start_service_operation',
+            'endpoint_name' => 'billing',
+            'service_name' => 'invoices',
+            'operation_name' => 'create',
+            'request_payload' => Serializer::serialize([
+                'invoice_id' => 42,
+            ]),
+            'payload_codec' => 'avro',
+        ];
+
+        $result = $this->bridge->completeForProtocol($this->task->id, [$command], $protocolVersion);
+
+        $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR));
+        $this->assertCount(1, $service->executions);
+        $options = $service->executions[0][3];
+        $this->assertSame($scheduled, ServiceCallScheduler::hasAfterCommitAuthority($options));
+    }
+
+    /**
+     * @return iterable<string, array{string, bool}>
+     */
+    public static function serviceSchedulerProtocols(): iterable
+    {
+        yield 'protocol 1.20 preserves inline execution' => ['1.20', false];
+        yield 'protocol 1.21 enables durable scheduling' => ['1.21', true];
+    }
+
+    #[DataProvider('unsupportedCompletionProtocols')]
+    public function testInvalidCompletionProtocolIsRejectedBeforeAnyWrite(
+        string $protocolVersion,
+        string $reason,
+    ): void {
+        $service = new BridgeServiceControlPlaneProbe([
+            'service_call_id' => 'must-not-execute',
+            'accepted' => true,
+            'status' => 'completed',
+        ]);
+        $this->app->instance(ServiceControlPlane::class, $service);
+        $before = $this->records();
+
+        $result = $this->bridge->completeForProtocol($this->task->id, [[
+            'type' => 'start_service_operation',
+            'endpoint_name' => 'billing',
+            'service_name' => 'invoices',
+            'operation_name' => 'create',
+            'request_payload' => Serializer::serialize([
+                'invoice_id' => 42,
+            ]),
+            'payload_codec' => 'avro',
+        ]], $protocolVersion);
+
+        $this->assertFalse($result['completed']);
+        $this->assertSame($reason, $result['reason']);
+        $this->assertSame([], $result['created_task_ids']);
+        $this->assertSame($before, $this->records());
+        $this->assertSame([], $service->executions);
+        $this->assertSame(TaskStatus::Leased, $this->task->fresh()->status);
+        Queue::assertNothingPushed();
+    }
+
+    /**
+     * @return iterable<string, array{string, string}>
+     */
+    public static function unsupportedCompletionProtocols(): iterable
+    {
+        yield 'malformed patch version' => ['1.21.0', 'invalid_protocol_version'];
+        yield 'future minor' => ['1.22', 'unsupported_protocol_version'];
+        yield 'different major' => ['2.0', 'unsupported_protocol_version'];
+        yield 'leading zero major' => ['01.21', 'invalid_protocol_version'];
+        yield 'leading zero minor' => ['1.020', 'invalid_protocol_version'];
+    }
+
+    public function testProtocolAwareCompletionIsAdditiveToTheStableBridgeContract(): void
+    {
+        $stableComplete = new ReflectionMethod(WorkflowTaskBridge::class, 'complete');
+        $protocolComplete = new ReflectionMethod(ProtocolAwareWorkflowTaskBridge::class, 'completeForProtocol');
+
+        $this->assertCount(2, $stableComplete->getParameters());
+        $this->assertCount(3, $protocolComplete->getParameters());
+    }
+
+    public function testUnchangedLegacyCompletionImplementationLoadsAndWorksThroughTheFacade(): void
+    {
+        // This class declares the original two-argument implementation. PHP
+        // rejects loading it if the base interface gains a third parameter.
+        $legacy = Mockery::mock(LegacyWorkflowTaskCompletionBridge::class)->makePartial();
+        $this->app->instance(WorkflowTaskBridge::class, $legacy);
+        $this->assertNotInstanceOf(ProtocolAwareWorkflowTaskBridge::class, $legacy);
+        $result = \Workflow\V2\WorkflowTaskBridge::complete('legacy-task', [[
+            'type' => 'complete_workflow',
+        ]]);
+        $this->assertTrue($result['completed']);
+        $this->assertSame('legacy-task', $result['task_id']);
+
+        $this->expectException(LogicException::class);
+        \Workflow\V2\WorkflowTaskBridge::completeForProtocol('legacy-task', [], '1.21');
+    }
+
     /**
      * @param array<string, mixed> $surface
      * @param array<string, mixed> $options
@@ -381,7 +494,7 @@ public function testServiceOptionsPreserveTypedValuesAndAuthoritativeCallerMetad
         $this->assertTrue($result['completed'], json_encode($result, JSON_THROW_ON_ERROR));
         $this->assertCount(1, $service->executions);
         $forwarded = $service->executions[0][3];
-        $expected = $options;
+        $expected = ServiceCallScheduler::withAfterCommitAuthority($options);
         $expected['mode_override'] = 'sync';
         $expected['wait_for'] = 'completed';
         $expected['metadata'] = [
@@ -582,6 +695,21 @@ private function records(): array
     }
 }
 
+abstract class LegacyWorkflowTaskCompletionBridge implements WorkflowTaskBridge
+{
+    public function complete(string $taskId, array $commands): array
+    {
+        return [
+            'completed' => $commands !== [],
+            'task_id' => $taskId,
+            'workflow_run_id' => null,
+            'run_status' => null,
+            'created_task_ids' => [],
+            'reason' => null,
+        ];
+    }
+}
+
 final class BridgeServiceControlPlaneProbe implements ServiceControlPlane
 {
     /**
diff --git a/tests/Unit/V2/ServiceCallSchedulerTest.php b/tests/Unit/V2/ServiceCallSchedulerTest.php
index 8c2e3efc..e88bd03a 100644
--- a/tests/Unit/V2/ServiceCallSchedulerTest.php
+++ b/tests/Unit/V2/ServiceCallSchedulerTest.php
@@ -95,11 +95,87 @@ public function testCompletionTransactionOnlyPersistsScheduleAndEnqueuesAfterCom
         $this->assertSame(ServiceCallScheduler::STATE_SCHEDULED, $dispatch->state);
         $this->assertSame(1, $dispatch->attempt);
         $this->assertSame(3, $dispatch->max_attempts);
+        $this->assertFalse(ServiceCallScheduler::hasAfterCommitAuthority($dispatch->dispatch_options));
         $this->assertLessThan(1_000, $elapsedMilliseconds);
         Queue::assertPushed(RunServiceCall::class, static fn (RunServiceCall $job): bool =>
             $job->serviceCallId === $call->id);
     }
 
+    public function testCallerContextAndIdempotencyTransactionDoNotScheduleAnHttpExecution(): void
+    {
+        $this->catalog();
+        /** @var WorkflowControlPlane&MockInterface $workflows */
+        $workflows = Mockery::mock(WorkflowControlPlane::class);
+        $workflows->shouldReceive('update')
+            ->once()
+            ->withArgs(static function (string $instanceId, string $name, array $options): bool {
+                return $instanceId === 'accepted-target'
+                    && $name === 'approve'
+                    && WorkflowServiceCall::query()->getModel()->getConnection()->transactionLevel() === 1
+                    && $options['wait_for'] === 'completed'
+                    && $options['wait_timeout_seconds'] === 60;
+            })
+            ->andReturn([
+                'accepted' => true,
+                'update_status' => 'completed',
+                'workflow_instance_id' => 'accepted-target',
+                'run_id' => 'target-run',
+                'update_id' => 'target-update',
+                'result' => [
+                    'approved' => true,
+                ],
+            ]);
+        $options = $this->callOptions(false);
+        $control = new DefaultServiceControlPlane($workflows);
+        $admitted = $control->execute('billing', 'invoices', 'approve', $options + [
+            'dispatch_handler' => false,
+        ]);
+        $options['service_call_id'] = $admitted['service_call_id'];
+        $options['boundary_policy_outcome'] = ServiceCallOutcome::Accepted->value;
+
+        $result = $control->execute('billing', 'invoices', 'approve', $options);
+
+        $this->assertTrue($result['accepted']);
+        $this->assertSame(ServiceCallStatus::Completed->value, $result['status']);
+        $this->assertSame(ServiceCallOutcome::Completed->value, $result['outcome']);
+        $this->assertArrayNotHasKey('retry_scheduled', $result);
+        $this->assertSame(0, WorkflowServiceCallDispatch::query()->count());
+        Queue::assertNothingPushed();
+    }
+
+    public function testCallerContextHttpExecutionPreservesInlineTypedFailure(): void
+    {
+        $this->catalog();
+        /** @var WorkflowControlPlane&MockInterface $workflows */
+        $workflows = Mockery::mock(WorkflowControlPlane::class);
+        $workflows->shouldReceive('update')
+            ->once()
+            ->andReturn([
+                'accepted' => false,
+                'reason' => 'approval_denied',
+                'message' => 'The invoice approval was denied.',
+                'service_error_type' => 'invoice_declined',
+            ]);
+        $options = $this->callOptions(false);
+        $control = new DefaultServiceControlPlane($workflows);
+        $admitted = $control->execute('billing', 'invoices', 'approve', $options + [
+            'dispatch_handler' => false,
+        ]);
+        $options['service_call_id'] = $admitted['service_call_id'];
+        $options['boundary_policy_outcome'] = ServiceCallOutcome::Accepted->value;
+
+        $result = $control->execute('billing', 'invoices', 'approve', $options);
+
+        $this->assertFalse($result['accepted']);
+        $this->assertSame(ServiceCallStatus::Failed->value, $result['status']);
+        $this->assertSame('approval_denied', $result['reason']);
+        $this->assertSame('invoice_declined', $result['service_error_type']);
+        $this->assertSame('The invoice approval was denied.', $result['message']);
+        $this->assertArrayNotHasKey('retry_scheduled', $result);
+        $this->assertSame(0, WorkflowServiceCallDispatch::query()->count());
+        Queue::assertNothingPushed();
+    }
+
     public function testRollbackLeavesNoCallScheduleOrRunnableJob(): void
     {
         $this->catalog();
@@ -806,9 +882,9 @@ private function catalog(?array $retryPolicy = null): WorkflowServiceOperation
     /**
      * @return array<string, mixed>
      */
-    private function callOptions(): array
+    private function callOptions(bool $scheduleAfterCommit = true): array
     {
-        return [
+        $options = [
             'namespace' => 'billing',
             'caller_namespace' => 'caller',
             'caller_workflow_instance_id' => $this->run->workflow_instance_id,
@@ -818,6 +894,10 @@ private function callOptions(): array
             'wait_for' => 'completed',
             'wait_timeout_seconds' => 60,
         ];
+
+        return $scheduleAfterCommit
+            ? ServiceCallScheduler::withAfterCommitAuthority($options)
+            : $options;
     }
 }
 
diff --git a/tests/Unit/V2/WorkerProtocolVersionTest.php b/tests/Unit/V2/WorkerProtocolVersionTest.php
index 49d1c803..8fe880b7 100644
--- a/tests/Unit/V2/WorkerProtocolVersionTest.php
+++ b/tests/Unit/V2/WorkerProtocolVersionTest.php
@@ -24,6 +24,28 @@ public function testVersionTracksServiceOperationCommandShape(): void
         $this->assertSame(0, WorkerProtocolVersion::longPollSemantics()['min_timeout_seconds']);
     }
 
+    public function testDurableServiceSchedulerIsAvailableOnlyFromProtocol121(): void
+    {
+        $this->assertTrue(WorkerProtocolVersion::isWellFormed('1.20'));
+        $this->assertTrue(WorkerProtocolVersion::isWellFormed('1.21'));
+        $this->assertFalse(WorkerProtocolVersion::isWellFormed('1.21.0'));
+        $this->assertFalse(WorkerProtocolVersion::supportsDurableServiceOperationScheduler('1.20'));
+        $this->assertTrue(WorkerProtocolVersion::supportsDurableServiceOperationScheduler('1.21'));
+        $this->assertTrue(WorkerProtocolVersion::supportsDurableServiceOperationScheduler('1.22'));
+        $this->assertFalse(WorkerProtocolVersion::supportsDurableServiceOperationScheduler('2.0'));
+    }
+
+    public function testRequestVersionNegotiationFailsClosed(): void
+    {
+        $this->assertTrue(WorkerProtocolVersion::isSupportedRequestVersion('1.20'));
+        $this->assertTrue(WorkerProtocolVersion::isSupportedRequestVersion('1.21'));
+        $this->assertFalse(WorkerProtocolVersion::isSupportedRequestVersion('1.22'));
+        $this->assertFalse(WorkerProtocolVersion::isSupportedRequestVersion('2.0'));
+        $this->assertFalse(WorkerProtocolVersion::isSupportedRequestVersion('1.21.0'));
+        $this->assertFalse(WorkerProtocolVersion::isSupportedRequestVersion('01.21'));
+        $this->assertFalse(WorkerProtocolVersion::isSupportedRequestVersion('1.021'));
+    }
+
     public function testVersionIncludesSignalWaitCommandShape(): void
     {
         $this->assertTrue(version_compare(WorkerProtocolVersion::VERSION, '1.9', '>='));