Allow replay-compatible workflow definition upgrades and explain routing

app/Http/Controllers/Api/TaskQueueController.php

@@ -10,6 +10,7 @@
 use App\Support\TaskQueueAdmission;
 use App\Support\TaskQueueBuildIdRolloutSnapshot;
 use App\Support\TaskQueuePriorityFairnessSurface;
+use App\Support\WorkflowDefinitionRouting;
 use App\Support\WorkflowQueryTaskBroker;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
@@ -28,6 +29,7 @@ public function __construct(
         private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
         private readonly TaskQueuePriorityFairnessSurface $priorityFairness,
         private readonly DeploymentLifecycleService $deployments,
+        private readonly WorkflowDefinitionRouting $definitionRouting,
     ) {}
 
     public function index(Request $request): JsonResponse
@@ -67,22 +69,26 @@ public function show(Request $request, string $taskQueue): JsonResponse
 
         $namespace = (string) $request->attributes->get('namespace');
 
-        return ControlPlaneProtocol::json(
-            $this->withAdmission(
+        $payload = $this->withAdmission(
+            $namespace,
+            $this->withRecentTaskFlow(
                 $namespace,
-                $this->withRecentTaskFlow(
+                $taskQueue,
+                StandaloneWorkerVisibility::queueDetail(
                     $namespace,
                     $taskQueue,
-                    StandaloneWorkerVisibility::queueDetail(
-                        $namespace,
-                        $taskQueue,
-                        WorkerRegistration::class,
-                        now(),
-                        $this->workerStaleAfterSeconds(),
-                    )->toArray(),
-                ),
+                    WorkerRegistration::class,
+                    now(),
+                    $this->workerStaleAfterSeconds(),
+                )->toArray(),
             ),
         );
+        $payload['definition_routing'] = $this->definitionRouting->forQueue($namespace, $taskQueue);
+        if ($payload['definition_routing']['blocked_sample_count'] > 0) {
+            $payload['repair']['needs_attention'] = true;
+        }
+
+        return ControlPlaneProtocol::json($payload);
     }
 
     /**
@@ -263,7 +269,7 @@ private function setBuildIdDrainIntent(Request $request, string $taskQueue, stri
     }
 
     /**
-     * @param array<string, mixed> $deployment
+     * @param  array<string, mixed>  $deployment
      * @return array<string, mixed>
      */
     private function buildIdLifecyclePayload(
@@ -351,7 +357,7 @@ private function normalizeDeploymentLifecycleTimestamps(array $deployment): arra
     }
 
     /**
-     * @param list<array<string, mixed>> $blockages
+     * @param  list<array<string, mixed>>  $blockages
      */
     private function orderBlockages(array &$blockages): void
     {
@@ -630,6 +636,7 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue
             if ($attemptStatus === ActivityAttemptStatus::Running->value) {
                 $counts['running_count'] += $count;
                 $counts['open_count'] += $count;
+
                 continue;
             }
 
@@ -639,13 +646,15 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue
             ) {
                 $counts['retrying_count'] += $count;
                 $counts['open_count'] += $count;
+
                 continue;
             }
 
             if ($attemptStatus === ActivityAttemptStatus::Expired->value) {
                 $counts['timed_out_count'] += $count;
                 $counts['expired_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 
@@ -656,18 +665,21 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue
             ) {
                 $counts['timed_out_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 
             if ($attemptStatus === ActivityAttemptStatus::Failed->value) {
                 $counts['failed_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 
             if ($attemptStatus === ActivityAttemptStatus::Completed->value) {
                 $counts['completed_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 

app/Http/Controllers/Api/WorkerController.php

@@ -46,6 +46,7 @@
 use App\Support\WorkflowHistoryPageToken;
 use App\Support\WorkflowMetadataCapabilityPolicy;
 use App\Support\WorkflowQueryTaskBroker;
+use App\Support\WorkflowReplayEligibility;
 use App\Support\WorkflowStreamCommandProcessor;
 use App\Support\WorkflowStreamService;
 use App\Support\WorkflowTaskLeaseRecovery;
@@ -160,6 +161,7 @@ class WorkerController
         'version_marker_kind_mismatch',
         'version_marker_sequence_invalid',
         'version_range_invalid',
+        'workflow_entry_arguments_mismatch',
         'workflow_nondeterministic',
     ];
 
@@ -407,7 +409,10 @@ public function register(Request $request): JsonResponse
         $this->validateWorkerTaskCapacity($maxWorkflowTasks, $maxActivityTasks);
         $maxWorkerSessions = $validated['max_concurrent_worker_sessions'] ?? 10;
         $taskSlots = is_array($validated['task_slots'] ?? null) ? $validated['task_slots'] : [];
-        $processMetrics = $this->normalizeProcessMetrics($validated['process_metrics'] ?? null);
+        $processMetrics = [
+            ...($this->normalizeProcessMetrics($validated['process_metrics'] ?? null) ?? []),
+            WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request),
+        ];
         $releaseLeasesForRegistration = $this->shouldReleaseLeasesForWorkerRegistration($existing, $processMetrics);
 
         try {
@@ -935,6 +940,10 @@ public function heartbeat(Request $request): JsonResponse
         if (array_key_exists('process_metrics', $validated)) {
             $update['process_metrics'] = $this->normalizeProcessMetrics($validated['process_metrics']);
         }
+        $update['process_metrics'] = [
+            ...(array_key_exists('process_metrics', $update) ? ($update['process_metrics'] ?? []) : ($worker->process_metrics ?? [])),
+            WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request),
+        ];
 
         if (array_key_exists('heartbeat_interval_seconds', $validated)
             && $validated['heartbeat_interval_seconds'] !== null) {

app/Http/Controllers/Api/WorkflowController.php

@@ -1250,6 +1250,7 @@ private function formatRun(WorkflowRun $run, string $namespace, array $descripti
             'compatibility_supported_in_fleet' => $this->compatibilitySupportedInFleet($namespace, $run),
             'compatibility_fleet_reason' => $this->compatibilityFleetReason($namespace, $run),
             'payload_codec' => $run->payload_codec,
+            'definition_routing' => $description['definition_routing'] ?? null,
             'execution_timeout_seconds' => $description['execution_timeout_seconds'] ?? null,
             'run_timeout_seconds' => $runDescription['run_timeout_seconds'] ?? null,
             'execution_deadline_at' => $runDescription['execution_deadline_at'] ?? null,

app/Support/ServerWorkflowControlPlane.php

@@ -26,6 +26,7 @@ public function __construct(
         private readonly ControlPlaneFailureDiagnostics $failureDiagnostics,
         private readonly NamespaceDurableStateQuota $durableStateQuota,
         private readonly WorkflowStartVersionPin $versionPin,
+        private readonly WorkflowDefinitionRouting $definitionRouting,
     ) {}
 
     public function start(string $workflowType, ?string $instanceId = null, array $options = []): array
@@ -148,10 +149,12 @@ public function terminate(string $instanceId, array $options = []): array
 
     public function repair(string $instanceId, array $options = []): array
     {
-        return $this->mutateCommand(
+        $result = $this->mutateCommand(
             $instanceId, 'repair', HistoryEventType::RepairRequested, $options,
             fn (): array => $this->inner->repair($instanceId, $options),
         );
+
+        return $this->withDefinitionRouting($instanceId, $options, $result);
     }
 
     public function redrive(string $instanceId, string $failedRunId, array $options = []): array
@@ -309,7 +312,27 @@ private function recoveredCommandResult(Throwable $exception, WorkflowCommand $c
 
     public function describe(string $instanceId, array $options = []): array
     {
-        return $this->inner->describe($instanceId, $options);
+        return $this->withDefinitionRouting($instanceId, $options, $this->inner->describe($instanceId, $options));
+    }
+
+    private function withDefinitionRouting(string $instanceId, array $options, array $result): array
+    {
+        $namespace = $this->namespace($options);
+        if ($namespace === null || ($result['found'] ?? true) === false
+            || in_array($result['reason'] ?? null, ['instance_not_found', 'run_not_found'], true)) {
+            return $result;
+        }
+
+        $runId = data_get($result, 'run.run_id') ?? $result['resolved_run_id'] ?? $result['run_id']
+            ?? $result['workflow_run_id'] ?? $options['run_id'] ?? null;
+        $run = is_string($runId)
+            ? NamespaceWorkflowScope::run($namespace, $instanceId, $runId)
+            : NamespaceWorkflowScope::currentRun($namespace, $instanceId);
+        if ($run instanceof WorkflowRun) {
+            $result['definition_routing'] = $this->definitionRouting->inspect($run);
+        }
+
+        return $result;
     }
 
     /**