fix: enforce canonical ownership for legacy namespace rows

app/Http/Controllers/Api/WorkerController.php

@@ -1164,16 +1164,14 @@ private function workerProcessIdentity(mixed $processMetrics): array
 
     private function releaseLeasedWorkflowTasksForReplacedWorker(string $namespace, string $workerId): void
     {
-        WorkflowRun::query()
-            ->where('namespace', $namespace)
+        NamespaceWorkflowScope::ownedRunQuery(WorkflowRun::query(), $namespace)
             ->where('sticky_worker_id', $workerId)
             ->update([
                 'sticky_worker_id' => null,
                 'sticky_until' => null,
             ]);
 
-        WorkflowTask::query()
-            ->where('namespace', $namespace)
+        NamespaceWorkflowScope::taskQuery($namespace)
             ->where('sticky_worker_id', $workerId)
             ->update([
                 'sticky_worker_id' => null,
@@ -1182,8 +1180,7 @@ private function releaseLeasedWorkflowTasksForReplacedWorker(string $namespace,
                 'sticky_claimed_at' => now(),
             ]);
 
-        WorkflowTask::query()
-            ->where('namespace', $namespace)
+        NamespaceWorkflowScope::taskQuery($namespace)
             ->where('task_type', TaskType::Workflow->value)
             ->where('status', TaskStatus::Leased->value)
             ->where('lease_owner', $workerId)
@@ -1201,8 +1198,7 @@ private function releaseLeasedWorkflowTasksForReplacedWorker(string $namespace,
 
     private function releaseLeasedActivityTasksForReplacedWorker(string $namespace, string $workerId): void
     {
-        WorkflowTask::query()
-            ->where('namespace', $namespace)
+        NamespaceWorkflowScope::taskQuery($namespace)
             ->where('task_type', TaskType::Activity->value)
             ->where('status', TaskStatus::Leased->value)
             ->where('lease_owner', $workerId)
@@ -1232,7 +1228,9 @@ private function expireLeasedActivityAttemptForReplacedWorker(WorkflowTask $task
         }
 
         /** @var ActivityExecution|null $execution */
-        $execution = ActivityExecution::query()->find($executionId);
+        $execution = ActivityExecution::query()
+            ->where('workflow_run_id', $task->workflow_run_id)
+            ->find($executionId);
 
         if (! $execution instanceof ActivityExecution) {
             return;
@@ -1242,6 +1240,7 @@ private function expireLeasedActivityAttemptForReplacedWorker(WorkflowTask $task
         $attempt = ActivityAttempt::query()
             ->where('workflow_task_id', $task->id)
             ->where('activity_execution_id', $execution->id)
+            ->where('workflow_run_id', $task->workflow_run_id)
             ->where('lease_owner', $workerId)
             ->where('status', ActivityAttemptStatus::Running->value)
             ->latest('attempt_number')

app/Support/ActivityTaskPoller.php

@@ -12,6 +12,7 @@
 use Workflow\V2\Models\ActivityExecution;
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowTask;
+use Workflow\V2\Support\DefaultActivityTaskBridge;
 use Workflow\V2\Support\TaskFairnessKey;
 use Workflow\V2\Support\TaskFairnessScheduler;
 use Workflow\V2\Support\TaskFairnessState;
@@ -897,6 +898,14 @@ private function claimReadyTask(
                         throw new ActivityTaskClaimRolledBack;
                     }
 
+                    // Revalidate legacy task ownership before the ID-only
+                    // package claim can write a lease or activity attempt.
+                    // Custom bridges may keep their tasks outside this table.
+                    if (($this->bridge instanceof DefaultActivityTaskBridge || WorkflowTask::query()->whereKey($taskId)->exists())
+                        && NamespaceWorkflowScope::lockTaskForMutation($namespace, $taskId) === null) {
+                        return null;
+                    }
+
                     $this->pollLeaseBindings->ensureUnbound(
                         $namespace,
                         TaskType::Activity,

app/Support/NamespaceWorkflowScope.php

@@ -7,6 +7,7 @@
 use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowRunSummary;
 use Workflow\V2\Models\WorkflowTask;
+use Workflow\V2\Support\NamespaceIdentity;
 
 class NamespaceWorkflowScope
 {
@@ -29,9 +30,8 @@ public static function bind(string $namespace, string $workflowId, ?string $work
 
     public static function workflowBound(string $namespace, string $workflowId): bool
     {
-        return WorkflowInstance::query()
+        return NamespaceIdentity::constrain(WorkflowInstance::query(), $namespace)
             ->whereKey($workflowId)
-            ->where('namespace', $namespace)
             ->exists();
     }
 
@@ -48,8 +48,7 @@ public static function namespaceForWorkflow(string $workflowId): ?string
 
     public static function runQuery(string $namespace, string $workflowId): Builder
     {
-        return WorkflowRun::query()
-            ->where('workflow_runs.namespace', $namespace)
+        return self::ownedRunQuery(WorkflowRun::query(), $namespace)
             ->where('workflow_runs.workflow_instance_id', $workflowId);
     }
 
@@ -69,20 +68,28 @@ public static function run(string $namespace, string $workflowId, string $runId)
 
     public static function runSummaryQuery(string $namespace): Builder
     {
-        return WorkflowRunSummary::query()
-            ->where('workflow_run_summaries.namespace', $namespace);
+        // Cached metadata is not authority for a legacy import's owner.
+        // Keep filtering and pagination inside the canonical run boundary.
+        return NamespaceIdentity::constrain(WorkflowRunSummary::query(), $namespace)
+            ->whereHas('run', static fn (Builder $run) => self::ownedRunQuery($run, $namespace)
+                ->whereColumn('workflow_runs.workflow_instance_id', 'workflow_run_summaries.workflow_instance_id'));
+    }
+
+    public static function ownedRunQuery(Builder $query, string $namespace): Builder
+    {
+        return NamespaceIdentity::constrain($query, $namespace)
+            ->whereHas('instance', static fn (Builder $instance) => NamespaceIdentity::constrain($instance, $namespace));
     }
 
     public static function taskQuery(string $namespace): Builder
     {
-        return WorkflowTask::query()
-            ->where('workflow_tasks.namespace', $namespace);
+        return NamespaceIdentity::constrain(WorkflowTask::query(), $namespace)
+            ->whereHas('run', static fn (Builder $run) => self::ownedRunQuery($run, $namespace));
     }
 
     public static function task(string $namespace, string $taskId): ?WorkflowTask
     {
-        return WorkflowTask::query()
-            ->where('namespace', $namespace)
+        return self::taskQuery($namespace)
             ->whereKey($taskId)
             ->first();
     }
@@ -98,7 +105,7 @@ public static function lockTaskForMutation(string $namespace, string $taskId): ?
             return null;
         }
 
-        $run = WorkflowRun::query()->where('namespace', $namespace)
+        $run = self::ownedRunQuery(WorkflowRun::query(), $namespace)
             ->whereKey($runId)->lockForUpdate()->first();
         if (! $run instanceof WorkflowRun) {
             return null;

app/Support/WorkflowTaskPoller.php

@@ -1073,7 +1073,7 @@ private function claimReadyTask(
                 // Request admission takes the same run lock before inspecting
                 // leases. Either it observes this claim's immutable proof, or
                 // this claim observes the newly accepted request.
-                $run = WorkflowRun::query()->where('namespace', $namespace)
+                $run = NamespaceWorkflowScope::ownedRunQuery(WorkflowRun::query(), $namespace)
                     ->lockForUpdate()->find($runId);
 
                 if (! $run instanceof WorkflowRun) {
@@ -1627,9 +1627,8 @@ private function backfillReadyTaskCompatibility(string $namespace, array &$ready
             return;
         }
 
-        $updated = WorkflowTask::query()
+        $updated = NamespaceWorkflowScope::taskQuery($namespace)
             ->whereKey($taskId)
-            ->where('namespace', $namespace)
             ->where('task_type', TaskType::Workflow->value)
             ->where(function ($query): void {
                 $query->whereNull('compatibility')
@@ -2339,7 +2338,7 @@ private function workerCanExecuteTask(
             ->where('worker_id', $workerId)
             ->first();
 
-        $run = WorkflowRun::query()->where('namespace', $namespace)->find($runId);
+        $run = NamespaceWorkflowScope::ownedRunQuery(WorkflowRun::query(), $namespace)->find($runId);
         if (! $worker instanceof WorkerRegistration || ! $run instanceof WorkflowRun
             || ! WorkflowReplayEligibility::allows($worker, $run, $protocolVersion)) {
             return false;

composer.json

@@ -6,7 +6,7 @@
     "require": {
         "php": "^8.2",
         "apache/avro": "^1.12",
-        "durable-workflow/workflow": "2.15.10",
+        "durable-workflow/workflow": "2.15.12",
         "laravel/framework": "^13.30",
         "laravel/tinker": "^3.0",
         "league/flysystem-aws-s3-v3": "^3.35.3"

tests/Feature/ActivityWorkerDeregistrationTest.php

@@ -190,7 +190,7 @@ public function test_contended_attempt_refuses_without_committing_the_registrati
 
     private function leaseActivity(int $maxAttempts): array
     {
-        $workflow = WorkflowStub::make(ExternalGreetingWorkflow::class, 'deregister-original-activity');
+        $workflow = WorkflowStub::make(ExternalGreetingWorkflow::class, 'deregister-original-activity', 'default');
         $start = $workflow->start('Ada');
         $this->runReadyWorkflowTask($start->runId());
         $execution = ActivityExecution::query()->where('workflow_run_id', $start->runId())->sole();

tests/Feature/LegacyActivityPollWindowTest.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace Tests\Feature;
+
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Queue;
+use Tests\Feature\Concerns\ServerTestHelpers;
+use Tests\Fixtures\ExternalGreetingWorkflow;
+use Tests\TestCase;
+use Workflow\V2\Models\WorkflowTask;
+
+final class LegacyActivityPollWindowTest extends TestCase
+{
+    use RefreshDatabase;
+    use ServerTestHelpers;
+
+    public function test_legacy_foreign_tasks_cannot_fill_the_activity_poll_window(): void
+    {
+        Queue::fake();
+        $this->createNamespace('default');
+        $this->configureWorkflowTypes(['tests.external-greeting-workflow' => ExternalGreetingWorkflow::class]);
+        $tasks = [];
+        foreach (['foreign', 'owned'] as $owner) {
+            $runId = $this->postJson('/api/workflows', [
+                'workflow_id' => $owner, 'workflow_type' => 'tests.external-greeting-workflow',
+                'task_queue' => 'workflow', 'input' => ['Ada'],
+            ], $this->apiHeaders())->assertCreated()->json('run_id');
+            $this->runReadyWorkflowTask($runId);
+            $tasks[$owner] = WorkflowTask::query()->where('workflow_run_id', $runId)->where('task_type', 'activity')->sole();
+        }
+        DB::table('workflow_instances')->where('id', 'foreign')->update(['namespace' => 'other']);
+        $tasks['foreign']->forceFill(['priority' => 0])->save();
+        for ($i = 1; $i < 10; $i++) {
+            $tasks['foreign']->replicate()->save();
+        }
+        $this->registerWorker('window-worker', 'external-activities');
+        $this->postJson('/api/worker/activity-tasks/poll', [
+            'worker_id' => 'window-worker', 'task_queue' => 'external-activities',
+        ], $this->workerHeaders())->assertOk()->assertJsonPath('task.task_id', $tasks['owned']->id);
+    }
+}