Honor start command context in control plane

Author: durable-workflow-ops

src/V2/Contracts/WorkflowControlPlane.php

@@ -36,6 +36,7 @@ interface WorkflowControlPlane
      * - business_key: string|null — caller-supplied business key
      * - labels: array<string, string>|null — visibility labels
      * - memo: array<string, mixed>|null — non-indexed metadata
+     * - command_context: \Workflow\V2\CommandContext|null — recorded command attribution/context
      * - duplicate_start_policy: 'reject_duplicate'|'return_existing_active'
      *
      * @return array{

src/V2/Support/CacheLongPollWakeStore.php

@@ -143,7 +143,7 @@ public function signalTask(WorkflowTask $task): void
             default => [],
         };
 
-        $this->signal(...$channels);
+        $this->signal(...$this->normalizeChannels($channels));
     }
 
     /**
@@ -206,7 +206,7 @@ private function signalTaskWithNamespace(WorkflowTask $task, ?string $namespace)
             default => [],
         };
 
-        $this->signal(...$channels);
+        $this->signal(...$this->normalizeChannels($channels));
     }
 
     private function namespaceForTask(WorkflowTask $task): ?string

src/V2/Support/DefaultWorkflowControlPlane.php

@@ -46,6 +46,7 @@ public function start(string $workflowType, ?string $instanceId = null, array $o
         $memo = $options['memo'] ?? null;
         $searchAttributes = $options['search_attributes'] ?? null;
         $namespace = $this->resolveNamespace($options);
+        $commandContext = $this->commandContext($options);
         $executionTimeoutSeconds = isset($options['execution_timeout_seconds']) ? (int) $options['execution_timeout_seconds'] : null;
         $runTimeoutSeconds = isset($options['run_timeout_seconds']) ? (int) $options['run_timeout_seconds'] : null;
         $duplicatePolicy = ($options['duplicate_start_policy'] ?? null) === 'return_existing_active'
@@ -72,6 +73,7 @@ public function start(string $workflowType, ?string $instanceId = null, array $o
             $memo,
             $searchAttributes,
             $namespace,
+            $commandContext,
             $executionTimeoutSeconds,
             $runTimeoutSeconds,
             $duplicatePolicy,
@@ -93,7 +95,7 @@ public function start(string $workflowType, ?string $instanceId = null, array $o
                         true
                     );
 
-                $command = WorkflowCommand::record($instance, $currentRun, $this->commandAttributes([
+                $command = WorkflowCommand::record($instance, $currentRun, $this->commandAttributes($commandContext, [
                     'command_type' => CommandType::Start->value,
                     'target_scope' => 'instance',
                     'status' => $canReturnExisting
@@ -193,7 +195,7 @@ public function start(string $workflowType, ?string $instanceId = null, array $o
                     'last_history_sequence' => 0,
                 ]);
 
-            $command = WorkflowCommand::record($instance, $run, $this->commandAttributes([
+            $command = WorkflowCommand::record($instance, $run, $this->commandAttributes($commandContext, [
                 'command_type' => CommandType::Start->value,
                 'target_scope' => 'instance',
                 'status' => CommandStatus::Accepted->value,
@@ -797,9 +799,9 @@ private function classPropertyDefault(string $workflowClass, string $property):
      * @param array<string, mixed> $attributes
      * @return array<string, mixed>
      */
-    private function commandAttributes(array $attributes): array
+    private function commandAttributes(CommandContext $commandContext, array $attributes): array
     {
-        return array_merge(CommandContext::controlPlane()->attributes(), $attributes);
+        return array_merge($commandContext->attributes(), $attributes);
     }
 
     /**

tests/Feature/V2/V2WorkflowControlPlaneTest.php

@@ -133,19 +133,69 @@ public function testStartWithAutoGeneratedInstanceId(): void
         $this->assertNotNull($result['workflow_run_id']);
     }
 
+    public function testStartSupportsCommandContext(): void
+    {
+        $result = $this->controlPlane->start('remote-workflow-type', 'ctrl-plane-start-context-1', [
+            'connection' => 'redis',
+            'queue' => 'default',
+            'command_context' => CommandContext::controlPlane()->with([
+                'caller' => [
+                    'type' => 'server',
+                    'label' => 'Standalone Server',
+                ],
+                'server' => [
+                    'namespace' => 'default',
+                    'command' => 'start',
+                ],
+            ]),
+        ]);
+
+        $this->assertTrue($result['started']);
+
+        $command = WorkflowCommand::query()
+            ->where('workflow_instance_id', 'ctrl-plane-start-context-1')
+            ->where('outcome', 'started_new')
+            ->firstOrFail();
+
+        $this->assertSame('control_plane', $command->source);
+        $this->assertSame('server', $command->commandContext()['caller']['type'] ?? null);
+        $this->assertSame('Standalone Server', $command->commandContext()['caller']['label'] ?? null);
+        $this->assertSame('default', $command->commandContext()['server']['namespace'] ?? null);
+        $this->assertSame('start', $command->commandContext()['server']['command'] ?? null);
+    }
+
     public function testStartRejectsDuplicate(): void
     {
         $this->controlPlane->start('remote-workflow-type', 'ctrl-plane-dup-1', [
             'connection' => 'redis',
             'queue' => 'default',
         ]);
 
-        $result = $this->controlPlane->start('remote-workflow-type', 'ctrl-plane-dup-1');
+        $result = $this->controlPlane->start('remote-workflow-type', 'ctrl-plane-dup-1', [
+            'command_context' => CommandContext::controlPlane()->with([
+                'caller' => [
+                    'type' => 'server',
+                    'label' => 'Standalone Server',
+                ],
+                'server' => [
+                    'namespace' => 'default',
+                    'command' => 'start',
+                ],
+            ]),
+        ]);
 
         $this->assertFalse($result['started']);
         $this->assertSame('ctrl-plane-dup-1', $result['workflow_instance_id']);
         $this->assertSame('rejected_duplicate', $result['outcome']);
         $this->assertSame('instance_already_started', $result['reason']);
+
+        $command = WorkflowCommand::query()
+            ->where('workflow_instance_id', 'ctrl-plane-dup-1')
+            ->where('outcome', 'rejected_duplicate')
+            ->firstOrFail();
+
+        $this->assertSame('server', $command->commandContext()['caller']['type'] ?? null);
+        $this->assertSame('default', $command->commandContext()['server']['namespace'] ?? null);
     }
 
     public function testStartReturnExistingActive(): void