Add optional PHP Server trace context propagation

.env.example

@@ -85,3 +85,7 @@ DW_METRICS_WORKFLOW_TASK_FAILURE_TYPE_LIMIT=20
 # with the latest published `:8.0` / `:7-alpine` tag (TD-S058).
 MYSQL_VERSION=8.0
 REDIS_VERSION=7-alpine
+
+# Optional trace context: requires the application-owned OpenTelemetry API.
+DW_OBSERVABILITY_CONTEXT_ENABLED=false
+DW_OBSERVABILITY_BAGGAGE_ALLOWLIST='[]'

README.md

@@ -120,6 +120,13 @@ generic "workflows per second" number cannot hide the workload being measured.
 - [External payload storage contract](docs/contracts/external-payload-storage.md)
 - [Helm upgrade guide](k8s/helm/durable-workflow/docs/UPGRADING.md)
 
+## Optional trace context
+
+A configured application image can persist bounded W3C context across compatible
+SDK workers. Enable the optional API and discover the actual capability as
+explained in [PHP Server trace context](docs/observability-context.md). The guide
+covers nullable schema migrations, baggage controls and runtime boundaries.
+
 ## Development
 
 ```bash

app/Http/Controllers/Api/ActivityTaskController.php

@@ -23,6 +23,7 @@
 use App\Support\WorkerProtocol;
 use App\Support\WorkerProtocolMutationRetrier;
 use App\Support\WorkerSessionRegistry;
+use App\Support\WorkflowObservabilityPoll;
 use Carbon\CarbonImmutable;
 use Carbon\CarbonInterface;
 use Illuminate\Http\JsonResponse;
@@ -187,7 +188,7 @@ public function poll(Request $request): JsonResponse
         );
 
         return WorkerProtocol::json([
-            'task' => $claim === null ? null : array_filter([
+            'task' => $claim === null ? null : WorkflowObservabilityPoll::filterDelivered(array_filter([
                 'task_id' => $claim['task_id'],
                 'workflow_id' => $claim['workflow_instance_id'],
                 'run_id' => $claim['workflow_run_id'],
@@ -216,7 +217,8 @@ public function poll(Request $request): JsonResponse
                         : null,
                 ),
                 'external_executor' => $externalExecutor,
-            ], static fn (mixed $v): bool => $v !== null),
+                ...WorkflowObservabilityPoll::forActivityClaim((string) $namespace, $claim),
+            ], static fn (mixed $v): bool => $v !== null)),
             'poll_status' => is_string($poll['poll_status'] ?? null)
                 ? $poll['poll_status']
                 : ($claim === null ? 'empty' : 'leased'),

app/Http/Controllers/Api/WorkerController.php

@@ -45,6 +45,9 @@
 use App\Support\WorkerTerminalEventAttribution;
 use App\Support\WorkflowHistoryPageToken;
 use App\Support\WorkflowMetadataCapabilityPolicy;
+use App\Support\WorkflowObservability;
+use App\Support\WorkflowObservabilityPoll;
+use App\Support\WorkflowObservabilityTransitions;
 use App\Support\WorkflowQueryTaskBroker;
 use App\Support\WorkflowReplayEligibility;
 use App\Support\WorkflowStreamCommandProcessor;
@@ -1460,6 +1463,10 @@ public function pollWorkflowTasks(Request $request): JsonResponse
             $task['sticky_replay_mode'] = null;
         }
 
+        if (is_array($task)) {
+            $task = WorkflowObservabilityPoll::filterDelivered($task);
+        }
+
         return WorkerProtocol::json([
             'task' => $task,
             'poll_status' => is_string($poll['poll_status'] ?? null)
@@ -1805,6 +1812,9 @@ private function mutateWorkflowTaskCommands(Request $request, string $taskId, bo
         }
 
         $input = $request->all();
+        if (is_array($input['commands'] ?? null)) {
+            $input['commands'] = WorkflowObservabilityTransitions::businessCommands($input['commands']);
+        }
         $topLevelValidator = Validator::make($input, $topLevelRules);
         $validationErrors = $topLevelValidator->fails()
             ? $topLevelValidator->errors()->messages()
@@ -2132,7 +2142,8 @@ function () use (
                             // or attribute a terminal event to a prefix receipt.
                             $outcome['completed'] = false;
                         } else {
-                            $outcome = $bridge->complete($taskId, $commands);
+                            $outcome = WorkflowObservability::completion($request, $claimedTask,
+                                fn (): array => $bridge->complete($taskId, $commands));
                         }
                         if (($outcome[$checkpoint ? 'checkpointed' : 'completed'] ?? false) === true) {
                             app(WorkflowStreamCommandProcessor::class)->process($taskId, (string) $namespace, $streamCommands);
@@ -4399,7 +4410,7 @@ public function pollQueryTasks(Request $request): JsonResponse
         }
 
         return WorkerProtocol::json([
-            'task' => $poll['task'],
+            'task' => WorkflowObservabilityPoll::filterOptionalDelivered($poll['task']),
             'poll_status' => $poll['poll_status'],
         ]);
     }

app/Http/Controllers/Api/WorkflowController.php

@@ -18,6 +18,7 @@
 use App\Support\SearchAttributeValueValidator;
 use App\Support\TaskQueueRoutingGate;
 use App\Support\WorkflowCommandContextFactory;
+use App\Support\WorkflowObservability;
 use App\Support\WorkflowQueryTaskBroker;
 use App\Support\WorkflowRunDiagnostics;
 use App\Support\WorkflowStartService;
@@ -288,7 +289,7 @@ public function start(Request $request): JsonResponse
         }
 
         try {
-            $start = $this->workflowStartService->start(
+            $start = WorkflowObservability::admission($request, $workflowId, 'start', fn () => $this->workflowStartService->start(
                 $validated,
                 $namespace,
                 $this->commandContexts->make(
@@ -301,7 +302,7 @@ public function start(Request $request): JsonResponse
                         'duplicate_policy' => $validated['duplicate_policy'] ?? null,
                     ], static fn (mixed $value): bool => $value !== null),
                 ),
-            );
+            ));
         } catch (ExternalPayloadStorageUnavailable $exception) {
             return ControlPlaneProtocol::jsonForRequest(
                 $request,
@@ -556,7 +557,7 @@ public function signal(Request $request, string $workflowId, string $signalName)
             retainExternal: true,
         );
 
-        $result = $this->workflowControlPlane->signal(
+        $result = WorkflowObservability::admission($request, $workflowId, 'signal', fn () => $this->workflowControlPlane->signal(
             $workflowId,
             $signalName,
             [
@@ -575,7 +576,7 @@ public function signal(Request $request, string $workflowId, string $signalName)
                 ),
                 'strict_configured_type_validation' => true,
             ],
-        );
+        ));
 
         return $this->resultMapper->signal(
             $workflowId,
@@ -767,15 +768,15 @@ public function update(Request $request, string $workflowId, string $updateName)
             ], static fn (mixed $value): bool => $value !== null),
         );
 
-        $externalResult = $this->externalUpdates->admit(
+        $externalResult = WorkflowObservability::admission($request, $workflowId, 'update', fn () => $this->externalUpdates->admit(
             namespace: (string) $namespace,
             workflowId: $workflowId,
             updateName: $updateName,
             arguments: $arguments,
             commandContext: $commandContext,
             waitFor: $waitFor,
             waitTimeoutSeconds: $validated['wait_timeout_seconds'] ?? null,
-        );
+        ));
 
         if ($externalResult !== null) {
             return $this->resultMapper->update(
@@ -787,7 +788,7 @@ public function update(Request $request, string $workflowId, string $updateName)
             );
         }
 
-        $result = $this->workflowControlPlane->update(
+        $result = WorkflowObservability::admission($request, $workflowId, 'update', fn () => $this->workflowControlPlane->update(
             $workflowId,
             $updateName,
             [
@@ -798,7 +799,7 @@ public function update(Request $request, string $workflowId, string $updateName)
                 'wait_timeout_seconds' => $validated['wait_timeout_seconds'] ?? null,
                 'strict_configured_type_validation' => true,
             ],
-        );
+        ));
 
         return $this->resultMapper->update(
             workflowId: $workflowId,
@@ -833,7 +834,7 @@ public function cancel(Request $request, string $workflowId): JsonResponse
             'request_id' => ['nullable', 'string', 'max:255'],
         ]);
 
-        $result = $this->workflowControlPlane->cancel(
+        $result = WorkflowObservability::admission($request, $workflowId, 'cancel', fn () => $this->workflowControlPlane->cancel(
             $workflowId,
             [
                 'namespace' => $namespace,
@@ -850,7 +851,7 @@ public function cancel(Request $request, string $workflowId): JsonResponse
                 ),
                 'strict_configured_type_validation' => true,
             ],
-        );
+        ));
 
         return $this->resultMapper->cancel(
             $workflowId,

app/Providers/AppServiceProvider.php

@@ -19,6 +19,7 @@
 use App\Observers\WorkflowHistoryEventObserver;
 use App\Observers\WorkflowTaskObserver;
 use App\Observers\WorkflowUpdateValidationObserver;
+use App\Support\DispatchObservability;
 use App\Support\ExternalWorkflowUpdateAdmission;
 use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\RemoteScheduleStarter;
@@ -30,6 +31,7 @@
 use App\Support\StoragePressure;
 use App\Support\ValidatedExternalWorkflowUpdateAdmission;
 use App\Support\WorkflowMemoRollingCompatibility;
+use App\Support\WorkflowObservability;
 use App\Support\WorkflowPackageApiFloor;
 use App\Support\WorkflowTaskLeaseConfiguration;
 use Illuminate\Console\Events\CommandStarting;
@@ -233,6 +235,10 @@ public function boot(): void
         }
 
         WorkflowTask::observe(WorkflowTaskObserver::class);
+        WorkflowRun::creating(WorkflowObservability::creatingRun(...));
+        WorkflowRun::created(WorkflowObservability::createdRun(...));
+        WorkflowCommand::creating(WorkflowObservability::creatingCommand(...));
+        DispatchObservability::install();
         WorkflowHistoryEvent::observe(WorkflowHistoryEventObserver::class);
         WorkflowUpdate::observe(WorkflowUpdateValidationObserver::class);
         WorkflowInstance::observe(NamespaceDurableStateObserver::class);

app/Support/DispatchObservability.php

@@ -0,0 +1,103 @@
+<?php
+
+namespace App\Support;
+
+use Illuminate\Queue\DatabaseQueue;
+use Illuminate\Queue\QueueManager;
+use Throwable;
+use Workflow\V2\Enums\TaskType;
+use Workflow\V2\Jobs\RunActivityTask;
+use Workflow\V2\Jobs\RunTimerTask;
+use Workflow\V2\Jobs\RunWorkflowTask;
+use Workflow\V2\Models\WorkflowTask;
+use Workflow\V2\Support\TaskTraceCarrier;
+
+/** Observe an actual queue producer, independently of business task authority. */
+final class DispatchObservability extends DatabaseQueue
+{
+    public static function install(): void
+    {
+        $callback = [self::class, 'payload'];
+        if (! in_array($callback, self::$createPayloadCallbacks, true)) {
+            self::createPayloadUsing($callback);
+        }
+    }
+
+    public static function payload(mixed $connection, mixed $queue, array $payload): array
+    {
+        if (! WorkflowObservability::enabled()) {
+            return [];
+        }
+        try {
+            $job = $payload['data']['command'] ?? null;
+            if ((! $job instanceof RunWorkflowTask && ! $job instanceof RunActivityTask && ! $job instanceof RunTimerTask)
+                || ! is_string($job->taskId) || ! property_exists($job, 'dispatchObservabilityContext')) {
+                return [];
+            }
+            $value = $job->dispatchObservabilityContext;
+            $context = TaskTraceCarrier::fromJob($value, $job->taskId);
+
+            return $context === null ? [] : ['dispatch_observability_context' => array_replace($value, ['context' => $context])];
+        } catch (Throwable) {
+            return [];
+        }
+    }
+
+    /** Called only after the poller's original task/attempt/lease fences. */
+    public static function forTask(WorkflowTask $task): ?array
+    {
+        if (! WorkflowObservability::enabled()) {
+            return null;
+        }
+        try {
+            $raw = $task->getRawOriginal('queue_delivery_reference');
+            if (! is_string($raw) || strlen($raw) > 2048) {
+                return null;
+            }
+            $reference = json_decode($raw, true, flags: JSON_THROW_ON_ERROR);
+            $manager = app('queue');
+            if ($manager::class !== QueueManager::class) {
+                return null;
+            }
+            $queue = $manager->connection($task->connection);
+            if (! $queue instanceof DatabaseQueue || ! is_array($reference) || ! self::matches($reference, $task, $queue)) {
+                return null;
+            }
+            $row = $queue->database->table($queue->table)->useWritePdo()
+                ->where('id', $reference['job_id'])->where('queue', $reference['queue'])->first(['payload']);
+            if ($row === null || ! is_string($row->payload) || strlen($row->payload) > 65536) {
+                return null;
+            }
+            $payload = json_decode($row->payload, true, flags: JSON_THROW_ON_ERROR);
+            $jobClass = match ($task->task_type) {
+                TaskType::Workflow => RunWorkflowTask::class,
+                TaskType::Activity => RunActivityTask::class,
+                TaskType::Timer => RunTimerTask::class,
+            };
+            if (($payload['uuid'] ?? null) !== $reference['payload_uuid'] || ($payload['data']['commandName'] ?? null) !== $jobClass) {
+                return null;
+            }
+            // The command remains an opaque serialized string. No job is unserialized.
+            $value = $payload['dispatch_observability_context'] ?? null;
+            $context = TaskTraceCarrier::fromJob($value, $task->id);
+
+            return $context === null ? null : array_replace($value, ['context' => $context]);
+        } catch (Throwable) {
+            return null;
+        }
+    }
+
+    private static function matches(array $reference, WorkflowTask $task, DatabaseQueue $queue): bool
+    {
+        return ($reference['version'] ?? null) === 1 && ($reference['task_id'] ?? null) === $task->id
+            && ($reference['run_id'] ?? null) === $task->workflow_run_id && ($reference['namespace'] ?? null) === $task->namespace
+            && ($reference['connection'] ?? null) === $queue->getConnectionName()
+            && ($reference['queue'] ?? null) === $queue->getQueue($task->queue)
+            && ($reference['database'] ?? null) === $queue->database->getName()
+            && ($reference['table'] ?? null) === $queue->table
+            && is_string($reference['payload_uuid'] ?? null) && $reference['payload_uuid'] !== '' && strlen($reference['payload_uuid']) <= 128
+            && is_string($reference['job_id'] ?? null)
+            && filter_var($reference['job_id'], FILTER_VALIDATE_INT, ['options' => ['min_range' => 1]]) !== false
+            && preg_match('/^[1-9]\d{0,19}$/D', $reference['job_id']) === 1;
+    }
+}

app/Support/WorkerProtocol.php

@@ -446,6 +446,7 @@ public static function serverCapabilities(): array
         $workerSessionSupported = self::workerSessionsSupported();
 
         return [
+            ...WorkflowObservability::capabilities(),
             'long_poll_timeout' => (int) config(
                 'server.polling.timeout',
                 WorkerProtocolVersion::DEFAULT_LONG_POLL_TIMEOUT,

app/Support/WorkflowCommandContextFactory.php

@@ -214,7 +214,7 @@ private function requestMetadata(Request $request, string $commandName): array
         $fingerprintPayload = [
             'method' => $request->method(),
             'path' => $path,
-            'payload' => $this->normalize($request->all()),
+            'payload' => $this->normalize($this->businessRequestPayload($request, $commandName)),
             'headers' => $headers,
         ];
 
@@ -242,6 +242,22 @@ private function requestMetadata(Request $request, string $commandName): array
         ], static fn (mixed $value): bool => $value !== null && $value !== '' && $value !== []);
     }
 
+    /**
+     * Exclude only the optional carrier at its admitted structural location.
+     * Same-named fields inside business input retain their ordinary identity.
+     *
+     * @return array<string, mixed>
+     */
+    private function businessRequestPayload(Request $request, string $commandName): array
+    {
+        $payload = $request->all();
+        if (in_array($commandName, ['start', 'signal_with_start', 'signal', 'update', 'cancel'], true)) {
+            unset($payload['observability_context']);
+        }
+
+        return $payload;
+    }
+
     private function bodyRequestId(Request $request): ?string
     {
         $value = $request->input('request_id');