diff --git a/src/V2/Jobs/RunWorkflowTask.php b/src/V2/Jobs/RunWorkflowTask.php
index 60deefecb..f4f6837b2 100644
--- a/src/V2/Jobs/RunWorkflowTask.php
+++ b/src/V2/Jobs/RunWorkflowTask.php
@@ -29,6 +29,7 @@
 use Workflow\V2\Support\WorkerCompatibilityFleet;
 use Workflow\V2\Support\WorkflowContinuationCache;
 use Workflow\V2\Support\WorkflowExecutor;
+use Workflow\V2\Support\WorkflowSignalReadScope;
 use Workflow\V2\Support\WorkflowTaskLease;
 
 final class RunWorkflowTask implements ShouldQueue
@@ -130,7 +131,10 @@ private function executeClaim(WorkflowExecutor $executor, int $claimedAttempt):
                     return null;
                 }
 
-                $nextTask = $executor->run($run, $task, $continuation);
+                $nextTask = WorkflowSignalReadScope::forLockedRun(
+                    $run,
+                    static fn (): ?WorkflowTask => $executor->run($run, $task, $continuation),
+                );
                 $continuation?->offer($task);
 
                 return $nextTask;
diff --git a/src/V2/Support/ContinuedRunMessageHandoff.php b/src/V2/Support/ContinuedRunMessageHandoff.php
index e943d86f1..94ab4882e 100644
--- a/src/V2/Support/ContinuedRunMessageHandoff.php
+++ b/src/V2/Support/ContinuedRunMessageHandoff.php
@@ -47,13 +47,25 @@ public static function closePredecessor(
 
     public static function transferInstanceScoped(WorkflowRun $closingRun, WorkflowRun $continuedRun): void
     {
-        $signals = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
+        $signalQuery = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
             ->where('workflow_run_id', $closingRun->id)
             ->where('target_scope', 'instance')
             ->where('status', SignalStatus::Received->value)
-            ->whereNull('closed_at')
-            ->lockForUpdate()
-            ->get();
+            ->whereNull('closed_at');
+        if (WorkflowSignalReadScope::allowsConsistentRead($closingRun, $signalQuery->getModel()->getConnection())) {
+            // Lock exact candidates without reserving neighboring index gaps.
+            $signals = $signalQuery->get()
+                ->map(static function (WorkflowSignal $candidate) use ($signalQuery): ?WorkflowSignal {
+                    return (clone $signalQuery)->whereKey($candidate->getKey())
+                        ->lockForUpdate()
+                        ->first();
+                })->filter();
+        } else {
+            // Outside the owned task, the caller may already have an older read view.
+            // Preserve current reads for every other execution path.
+            $signals = $signalQuery->lockForUpdate()
+                ->get();
+        }
         if ($signals->isNotEmpty()) {
             $declaredSignals = RunCommandContract::forRun($closingRun)['signals'];
             $signals = $signals->filter(
diff --git a/src/V2/Support/PendingSignalCloser.php b/src/V2/Support/PendingSignalCloser.php
index dd2c85476..2cd771d9c 100644
--- a/src/V2/Support/PendingSignalCloser.php
+++ b/src/V2/Support/PendingSignalCloser.php
@@ -30,15 +30,27 @@ public static function closeForTerminalEvent(
             return;
         }
 
-        $signals = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
+        $signalQuery = ConfiguredV2Models::query('signal_model', WorkflowSignal::class)
             ->where('workflow_run_id', $run->id)
             ->where('status', SignalStatus::Received->value)
             ->orderByRaw('CASE WHEN command_sequence IS NULL THEN 1 ELSE 0 END')
             ->orderBy('command_sequence')
             ->orderBy('received_at')
-            ->orderBy('id')
-            ->lockForUpdate()
-            ->get();
+            ->orderBy('id');
+        if (WorkflowSignalReadScope::allowsConsistentRead($run, $signalQuery->getModel()->getConnection())) {
+            // Lock exact candidates without reserving neighboring index gaps.
+            $signals = $signalQuery->get()
+                ->map(static function (WorkflowSignal $candidate) use ($signalQuery): ?WorkflowSignal {
+                    return (clone $signalQuery)->whereKey($candidate->getKey())
+                        ->lockForUpdate()
+                        ->first();
+                })->filter();
+        } else {
+            // Outside the owned task, the caller may already have an older read view.
+            // Preserve current reads for every other execution path.
+            $signals = $signalQuery->lockForUpdate()
+                ->get();
+        }
 
         foreach ($signals as $signal) {
             if (! $signal instanceof WorkflowSignal) {
diff --git a/src/V2/Support/WorkflowSignalReadScope.php b/src/V2/Support/WorkflowSignalReadScope.php
new file mode 100644
index 000000000..74198b921
--- /dev/null
+++ b/src/V2/Support/WorkflowSignalReadScope.php
@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+use Closure;
+use Illuminate\Database\Connection;
+use PDO;
+use WeakMap;
+use Workflow\V2\Models\WorkflowRun;
+
+/** @internal The embedded task has locked this run before its first consistent read. */
+final class WorkflowSignalReadScope
+{
+    /**
+     * @var WeakMap<Connection, array{pdo: PDO, run_id: string}>|null
+     */
+    private static ?WeakMap $scopes = null;
+
+    /**
+     * @template T
+     *
+     * @param  Closure(): T  $operation
+     * @return T
+     */
+    public static function forLockedRun(WorkflowRun $run, Closure $operation): mixed
+    {
+        $connection = $run->getConnection();
+        if ($connection->getDriverName() !== 'mysql' || $connection->transactionLevel() !== 1) {
+            return $operation();
+        }
+
+        self::$scopes ??= new WeakMap();
+        $previous = self::$scopes[$connection] ?? null;
+        self::$scopes[$connection] = [
+            'pdo' => $connection->getPdo(),
+            'run_id' => $run->id,
+        ];
+        try {
+            return $operation();
+        } finally {
+            if ($previous === null) {
+                unset(self::$scopes[$connection]);
+            } else {
+                self::$scopes[$connection] = $previous;
+            }
+        }
+    }
+
+    public static function allowsConsistentRead(WorkflowRun $run, Connection $connection): bool
+    {
+        $scope = self::$scopes[$connection] ?? null;
+
+        return $scope !== null
+            && $connection->transactionLevel() > 0
+            && $scope['pdo'] === $connection->getRawPdo()
+            && $scope['run_id'] === $run->id;
+    }
+}
diff --git a/tests/Feature/V2/WorkflowTaskSignalGapLockTest.php b/tests/Feature/V2/WorkflowTaskSignalGapLockTest.php
new file mode 100644
index 000000000..bdc8a6af9
--- /dev/null
+++ b/tests/Feature/V2/WorkflowTaskSignalGapLockTest.php
@@ -0,0 +1,323 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Feature\V2;
+
+use Illuminate\Database\Connection;
+use Illuminate\Database\Events\QueryExecuted;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Queue;
+use PHPUnit\Framework\Attributes\DataProvider;
+use RuntimeException;
+use Tests\Fixtures\V2\TestConfiguredContinueSignalWorkflow;
+use Tests\Fixtures\V2\TestSignalContinuationWorkflow;
+use Tests\TestCase;
+use Throwable;
+use Workflow\V2\Enums\HistoryEventType;
+use Workflow\V2\Enums\RunStatus;
+use Workflow\V2\Enums\SignalStatus;
+use Workflow\V2\Enums\TaskStatus;
+use Workflow\V2\Jobs\RunWorkflowTask;
+use Workflow\V2\Models\WorkflowHistoryEvent;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Models\WorkflowSignal;
+use Workflow\V2\Models\WorkflowTask;
+use Workflow\V2\WorkflowStub;
+
+final class WorkflowTaskSignalGapLockTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        parent::setUp();
+        self::stopWorkers();
+        Queue::fake();
+        config([
+            'workflows.v2.task_dispatch_mode' => 'poll',
+            'workflows.storage.transaction_attempts' => 1,
+            'cache.default' => 'array',
+        ]);
+        if (DB::connection()->getDriverName() !== 'mysql'
+            || str_contains(strtolower((string) DB::selectOne('SELECT VERSION() AS version')->version), 'mariadb')) {
+            $this->markTestSkipped('MySQL 8 InnoDB REPEATABLE READ is required.');
+        }
+        if (! function_exists('pcntl_fork') || ! function_exists('posix_kill')
+            || ! function_exists('stream_socket_pair')) {
+            $this->markTestSkipped('Process control and local sockets are required.');
+        }
+        DB::statement('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
+        DB::statement('SET SESSION innodb_lock_wait_timeout = 5');
+    }
+
+    public static function races(): iterable
+    {
+        foreach ([false, true] as $named) {
+            foreach (['history', 'task'] as $variant) {
+                yield ($named ? 'named' : 'default') . '/' . $variant => [$variant, $named];
+            }
+        }
+    }
+
+    #[DataProvider('races')]
+    public function testContinueAsNewDoesNotDeadlockWithAnUnrelatedSignal(string $variant, bool $named): void
+    {
+        $this->configureStorage($named);
+        $runs = [];
+        // A < B < C bounds A's signal gap with C. For the successor-task
+        // variant, A < D < B bounds B's history gap with D instead.
+        foreach ($variant === 'task' ? ['a', 'd', 'b'] : ['a', 'b', 'c'] as $name) {
+            $stub = WorkflowStub::make(TestSignalContinuationWorkflow::class);
+            $stub->start(0);
+            $this->executeReadyTask($stub->runId());
+            $runs[$name] = $stub;
+        }
+        foreach ($variant === 'task' ? ['a'] : ['a', 'c'] as $name) {
+            $this->assertTrue($runs[$name]->signal('resume')->accepted());
+        }
+        $task = WorkflowTask::query()->where('workflow_run_id', $runs['a']->runId())
+            ->where('status', TaskStatus::Ready->value)->sole();
+        $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        $this->assertIsArray($sockets);
+        $connectionName = $this->storage()
+            ->getName();
+        foreach (array_keys(DB::getConnections()) as $openConnection) {
+            DB::purge($openConnection);
+        }
+        $pid = pcntl_fork();
+        $this->assertNotSame(-1, $pid);
+        if ($pid === 0) {
+            fclose($sockets[0]);
+            $this->signalActor($sockets[1], $runs['b']->id(), $runs['b']->runId());
+        }
+        fclose($sockets[1]);
+        stream_set_timeout($sockets[0], 15);
+        $reaped = false;
+        $armed = true;
+        try {
+            DB::listen(function (QueryExecuted $query) use (&$armed, $runs, $sockets): void {
+                if (! $armed || ! str_starts_with($query->sql, 'update `workflow_runs` set `status`')
+                    || ! in_array('continued', $query->bindings, true)
+                    || ! in_array($runs['a']->runId(), $query->bindings, true)) {
+                    return;
+                }
+                $armed = false;
+                fwrite($sockets[0], "continue-ready\n");
+                $this->assertSame("history-locked\n", fgets($sockets[0]));
+                fwrite($sockets[0], "proceed\n");
+            });
+            $error = null;
+            try {
+                app()->call([new RunWorkflowTask($task->id), 'handle']);
+            } catch (Throwable $exception) {
+                $error = $exception;
+            }
+            $line = fgets($sockets[0]);
+            $this->assertIsString($line, 'The signal actor did not finish.');
+            $signal = json_decode($line, true, flags: JSON_THROW_ON_ERROR);
+            pcntl_waitpid($pid, $status);
+            $reaped = true;
+            $this->assertTrue(pcntl_wifexited($status));
+            $this->assertSame(0, pcntl_wexitstatus($status));
+            $this->assertNull($error, $error === null ? '' : $this->describeError($error));
+            $this->assertTrue($signal['accepted'] ?? false, json_encode($signal, JSON_THROW_ON_ERROR));
+            $this->assertSame(1, $signal['attempts'], 'Signal admission must not hide the cycle with a retry.');
+            $this->assertFalse($armed, 'The terminal-turn barrier was not reached.');
+            $oldRun = WorkflowRun::query()->findOrFail($runs['a']->runId());
+            $this->assertSame('continued', $oldRun->closed_reason);
+            $this->assertSame(TaskStatus::Completed, $task->fresh()->status);
+            $this->assertSame(2, WorkflowRun::query()->where('workflow_instance_id', $runs['a']->id())->count());
+            $this->assertSame(1, $oldRun->historyEvents()
+                ->where('event_type', HistoryEventType::WorkflowContinuedAsNew->value)->count());
+            $this->assertSame(1, WorkflowSignal::query()->where('workflow_run_id', $runs['b']->runId())->count());
+            $this->assertSame(0, $this->storage()->transactionLevel());
+        } finally {
+            $armed = false;
+            if (! $reaped) {
+                posix_kill($pid, SIGKILL);
+                pcntl_waitpid($pid, $status);
+            }
+            fclose($sockets[0]);
+            DB::purge($connectionName);
+        }
+    }
+
+    public static function callerTransactions(): iterable
+    {
+        foreach ([false, true] as $named) {
+            foreach (['cancel', 'continue', 'record', 'failed-task'] as $mode) {
+                yield ($named ? 'named' : 'default') . '/' . $mode => [$mode, $named];
+            }
+        }
+    }
+
+    #[DataProvider('callerTransactions')]
+    public function testCallerReadViewDoesNotHideAnAcceptedSignal(string $mode, bool $named): void
+    {
+        $this->configureStorage($named);
+        if ($mode === 'continue') {
+            $stub = WorkflowStub::make(TestConfiguredContinueSignalWorkflow::class);
+            $stub->start(0);
+        } else {
+            $stub = WorkflowStub::make(TestSignalContinuationWorkflow::class);
+            $stub->start(0);
+            $this->executeReadyTask($stub->runId());
+        }
+        if ($mode === 'failed-task') {
+            $this->failContinuation($stub);
+        }
+        $runId = $stub->runId();
+        $primary = $this->storage();
+        $storage = config('workflows.storage.connection');
+        $writer = $primary->getConfig();
+        unset($writer['name']);
+        $writer['url'] = null;
+        config([
+            'database.connections.signal_gap_writer' => $writer,
+        ]);
+        $primary->beginTransaction();
+        try {
+            $primary->table('workflow_signal_records')
+                ->where('workflow_run_id', $runId)
+                ->count();
+            config([
+                'workflows.storage.connection' => 'signal_gap_writer',
+            ]);
+            $receipt = $mode === 'continue'
+                ? WorkflowStub::load($stub->id())->signal('name-provided', 'late')
+                : WorkflowStub::load($stub->id())->signal('resume');
+            $signalId = WorkflowSignal::query()->where('workflow_command_id', $receipt->commandId())->sole()->id;
+            config([
+                'workflows.storage.connection' => $storage,
+            ]);
+            if ($mode === 'continue') {
+                $this->executeReadyTask($runId);
+            } elseif ($mode === 'cancel') {
+                WorkflowStub::load($stub->id())->cancel('caller transaction');
+            } else {
+                // record() itself adds a transaction level. Depth alone is
+                // insufficient to establish ownership of the caller's read view.
+                $run = WorkflowRun::query()->whereKey($runId)->lockForUpdate()->firstOrFail();
+                $run->forceFill([
+                    'status' => RunStatus::Cancelled,
+                    'closed_reason' => 'cancelled',
+                    'closed_at' => now(),
+                ])->save();
+                WorkflowHistoryEvent::record($run, HistoryEventType::WorkflowCancelled);
+            }
+            $signal = WorkflowSignal::query()->whereKey($signalId)->lockForUpdate()->firstOrFail();
+            $this->assertTrue($receipt->accepted());
+            $this->assertSame($mode === 'continue' ? SignalStatus::Received : SignalStatus::Rejected, $signal->status);
+            if ($mode === 'continue') {
+                $this->assertNotSame($runId, $signal->workflow_run_id);
+            }
+        } finally {
+            config([
+                'workflows.storage.connection' => $storage,
+            ]);
+            if ($primary->transactionLevel() > 0) {
+                $primary->rollBack();
+            }
+            DB::purge('signal_gap_writer');
+        }
+    }
+
+    private function configureStorage(bool $named): void
+    {
+        if ($named) {
+            $config = DB::connection()->getConfig();
+            unset($config['name']);
+            $config['url'] = null;
+            config([
+                'database.connections.signal_gap_store' => $config,
+                'workflows.storage.connection' => 'signal_gap_store',
+            ]);
+        }
+        $this->storage()
+            ->statement('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
+        $this->storage()
+            ->statement('SET SESSION innodb_lock_wait_timeout = 5');
+    }
+
+    private function storage(): Connection
+    {
+        return DB::connection(config('workflows.storage.connection'));
+    }
+
+    private function executeReadyTask(string $runId): void
+    {
+        $task = WorkflowTask::query()->where('workflow_run_id', $runId)
+            ->where('status', TaskStatus::Ready->value)->sole();
+        app()
+            ->call([new RunWorkflowTask($task->id), 'handle']);
+    }
+
+    private function failContinuation(WorkflowStub $stub): void
+    {
+        $stub->signal('resume');
+        $armed = true;
+        DB::listen(static function (QueryExecuted $query) use (&$armed, $stub): void {
+            if ($armed && str_starts_with($query->sql, 'update `workflow_runs` set `status`')
+                && in_array('continued', $query->bindings, true)
+                && in_array($stub->runId(), $query->bindings, true)) {
+                throw new RuntimeException('terminal-turn rollback control');
+            }
+        });
+        try {
+            $this->executeReadyTask($stub->runId());
+            $this->fail('The rollback control did not execute.');
+        } catch (RuntimeException $error) {
+            $this->assertSame('terminal-turn rollback control', $error->getMessage());
+        } finally {
+            $armed = false;
+        }
+        $this->assertSame(0, $this->storage()->transactionLevel());
+    }
+
+    /**
+     * @param resource $socket
+     */
+    private function signalActor($socket, string $instanceId, string $runId): never
+    {
+        stream_set_timeout($socket, 15);
+        $attempts = 0;
+        try {
+            if (fgets($socket) !== "continue-ready\n") {
+                throw new RuntimeException('The continuation did not reach its terminal turn.');
+            }
+            $this->storage()
+                ->statement('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
+            $this->storage()
+                ->statement('SET SESSION innodb_lock_wait_timeout = 5');
+            DB::listen(static function (QueryExecuted $query) use ($socket, $runId, &$attempts): void {
+                if (str_contains($query->sql, 'from `workflow_history_events`')
+                    && str_contains($query->sql, 'for update') && in_array($runId, $query->bindings, true)) {
+                    $attempts++;
+                    fwrite($socket, "history-locked\n");
+                    if (fgets($socket) !== "proceed\n") {
+                        throw new RuntimeException('The continuation did not release the barrier.');
+                    }
+                }
+            });
+            $receipt = WorkflowStub::load($instanceId)->signal('resume');
+            $result = [
+                'accepted' => $receipt->accepted(),
+                'attempts' => $attempts,
+            ];
+        } catch (Throwable $error) {
+            $result = [
+                'accepted' => false,
+                'attempts' => $attempts,
+                'error' => $this->describeError($error),
+            ];
+        }
+        fwrite($socket, json_encode($result, JSON_THROW_ON_ERROR) . PHP_EOL);
+        DB::disconnect($this->storage()->getName());
+        fclose($socket);
+        exit(0);
+    }
+
+    private function describeError(Throwable $error): string
+    {
+        return $error::class . ': ' . $error->getMessage();
+    }
+}
diff --git a/tests/Fixtures/V2/TestSignalContinuationWorkflow.php b/tests/Fixtures/V2/TestSignalContinuationWorkflow.php
new file mode 100644
index 000000000..4cc0f0241
--- /dev/null
+++ b/tests/Fixtures/V2/TestSignalContinuationWorkflow.php
@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures\V2;
+
+use Workflow\V2\Attributes\Signal;
+use function Workflow\V2\continueAsNew;
+use function Workflow\V2\signal;
+use Workflow\V2\Workflow;
+
+#[Signal('resume')]
+final class TestSignalContinuationWorkflow extends Workflow
+{
+    public function handle(int $round = 0): mixed
+    {
+        signal('resume');
+
+        return continueAsNew($round + 1);
+    }
+}
diff --git a/tests/Unit/V2/WorkflowSignalReadScopeTest.php b/tests/Unit/V2/WorkflowSignalReadScopeTest.php
new file mode 100644
index 000000000..314bd3d0b
--- /dev/null
+++ b/tests/Unit/V2/WorkflowSignalReadScopeTest.php
@@ -0,0 +1,130 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Unit\V2;
+
+use Illuminate\Database\Connection;
+use PDO;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use RuntimeException;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Support\WorkflowSignalReadScope;
+
+final class WorkflowSignalReadScopeTest extends TestCase
+{
+    public static function entryConditions(): iterable
+    {
+        yield 'root MySQL' => ['mysql', 1, true];
+        yield 'no transaction' => ['mysql', 0, false];
+        yield 'caller transaction' => ['mysql', 2, false];
+        yield 'other driver' => ['sqlite', 1, false];
+    }
+
+    #[DataProvider('entryConditions')]
+    public function testOnlyTheOwnedMysqlRootTransactionOpensAScope(string $driver, int $depth, bool $allowed): void
+    {
+        $connection = $this->connection($driver);
+        $run = $this->makeRun($connection, 'one');
+        for ($i = 0; $i < $depth; $i++) {
+            $connection->beginTransaction();
+        }
+        try {
+            $this->assertFalse(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+            $result = WorkflowSignalReadScope::forLockedRun($run, function () use (
+                $run,
+                $connection,
+                $allowed
+            ): string {
+                $this->assertSame($allowed, WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+                return 'returned';
+            });
+            $this->assertSame('returned', $result);
+            $this->assertFalse(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+        } finally {
+            if ($connection->transactionLevel() > 0) {
+                $connection->rollBack(0);
+            }
+        }
+    }
+
+    public function testScopeRequiresTheSameRunConnectionAndPdo(): void
+    {
+        $connection = $this->connection();
+        $other = $this->connection();
+        $run = $this->makeRun($connection, 'one');
+        $connection->beginTransaction();
+        try {
+            WorkflowSignalReadScope::forLockedRun($run, function () use ($connection, $other, $run): void {
+                $this->assertTrue(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+                $this->assertFalse(
+                    WorkflowSignalReadScope::allowsConsistentRead($this->makeRun($connection, 'two'), $connection)
+                );
+                $this->assertFalse(WorkflowSignalReadScope::allowsConsistentRead($run, $other));
+                $pdo = $connection->getPdo();
+                try {
+                    $connection->setPdo(new PDO('sqlite::memory:'));
+                    $connection->beginTransaction();
+                    $this->assertFalse(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+                    $connection->rollBack();
+                } finally {
+                    $connection->setPdo($pdo);
+                }
+                // setPdo resets Laravel's depth even though the old PDO still
+                // has its transaction. A stale transaction must stay refused.
+                $this->assertFalse(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+            });
+        } finally {
+            if ($connection->getPdo()->inTransaction()) {
+                $connection->getPdo()
+                    ->rollBack();
+            }
+        }
+    }
+
+    public function testExceptionsClearTheScopeBeforeAnotherCallerCanUseTheConnection(): void
+    {
+        $connection = $this->connection();
+        $run = $this->makeRun($connection, 'one');
+        $connection->beginTransaction();
+        try {
+            try {
+                WorkflowSignalReadScope::forLockedRun($run, function () use ($run, $connection): never {
+                    $this->assertTrue(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+                    throw new RuntimeException('scope control');
+                });
+                $this->fail('The original exception was hidden.');
+            } catch (RuntimeException $error) {
+                $this->assertSame('scope control', $error->getMessage());
+            }
+            $this->assertFalse(WorkflowSignalReadScope::allowsConsistentRead($run, $connection));
+        } finally {
+            $connection->rollBack();
+        }
+    }
+
+    private function connection(string $driver = 'mysql'): Connection
+    {
+        // SQLite backs only scope/identity unit controls. The separate feature
+        // suite exercises actual InnoDB locking with independent processes.
+        return new Connection(new PDO('sqlite::memory:'), config: [
+            'driver' => $driver,
+        ]);
+    }
+
+    private function makeRun(Connection $connection, string $id): WorkflowRun
+    {
+        $run = new class() extends WorkflowRun {
+            public Connection $scopeConnection;
+
+            public function getConnection(): Connection
+            {
+                return $this->scopeConnection;
+            }
+        };
+        $run->scopeConnection = $connection;
+        $run->id = $id;
+        return $run;
+    }
+}