Handle blocked queue workers and preserve job outcomes

AGENTS.md

@@ -395,8 +395,13 @@ Decide where state lives before writing code:
 Classes that use static caching need a `flushState()` method for test cleanup — see "Static state and test cleanup" under Writing Tests.
 
 - **Release idle pooled connections before unrelated long waits** — Holding connections while waiting for external services or child coroutines can exhaust pools and limit concurrency. Release them at the appropriate shared boundary, preserving active operations and session ownership; consider reacquisition costs rather than adding release calls indiscriminately. For database sessions, reuse `ConnectionResolver::releaseIdleConnections()`, which preserves transactions and explicit pins. See `Mailer::sendSymfonyMessage()` and `ConcurrentImportRunner::wait()` for examples.
-- **Treat coroutine cancellation separately from errors** — Code that owns coroutine cancellation must pass `CanceledException` through unchanged and clean up the children or resources it owns. Add this only where cancellation can actually happen and mishandling it causes a real bug—not to every `catch (Throwable)`.
-- **Contain cancellation in direct engine coroutines** — A callback passed directly to `Engine\Coroutine` must catch cancellation after cleanup so it cannot terminate the worker. The higher-level `Coroutine` class already provides this protection.
+
+### Coroutine cancellation
+
+- **Treat coroutine cancellation separately from errors** — Where cancellation needs different handling and the operation runs callbacks or listeners—including `releaseIdleConnections()`—use `Coroutine::causedByCancellation($exception)`. Cleanup errors can carry cancellation in their exception chain. Preserve the original exception, perform owned cleanup, and propagate cancellation instead of retrying or continuing normal work. Report genuine cleanup errors without swallowing cancellation. Add this only where cancellation can actually happen and mishandling it causes a real bug. Keep plain checks for native-only operations.
+- **Preserve cancellation ownership** — If shared work can safely be retried, allow another waiter to retry after its owner is canceled. Otherwise, share the failure behind `ForeignCancellationException`, or `ChildCancellationException` for an independently canceled child, preserving the original cause. The canceled owner propagates its own exception unchanged; waiters must not mistake it for their own cancellation.
+- **Contain cancellation in direct engine coroutines** — Callbacks passed directly to `Engine\Coroutine` must catch cancellation after cleanup so it cannot terminate the worker. The higher-level `Coroutine` class already provides this protection.
+- **Test cancellation through cleanup** — When adding or changing cancellation-sensitive retry, child-management or shared-result handling, extend relevant regression tests with throwing cleanup during cancellation. Verify the resulting behavior, not just the exception type.
 
 ## When to Stop and Report
 

docs/todo.md

@@ -30,6 +30,7 @@
 - Audit unmatched PHPStan inline ignores and global patterns with `reportUnmatchedIgnoredErrors` enabled — currently 196 unmatched inline ignores across 99 files plus 5 unmatched global patterns. Remove only suppressions that no longer match after tracing the underlying code; do not replace correct source with runtime branches or wider types merely to keep static analysis green. Decide as part of the work whether `phpstan.neon.dist` should then set `reportUnmatchedIgnoredErrors: true` permanently, since leaving it `false` lets the suppressions rot again.
 - Add PHPStan Eloquent extensions that preserve `Eloquent\Builder<TModel>` for non-passthrough methods forwarded to `Query\Builder`, and expose model named scopes on Eloquent builders and relations. The query-builder mixin currently gives fluent calls the wrong builder type, while named scopes are treated as nonexistent methods; these gaps force scopes to split mutation from return and leave `HasDatabaseNotifications` with `method.notFound` suppressions. This will be the repository's first PHPStan extension, so use Larastan as prior art and wire the extensions into `phpstan.neon.dist` without maintaining duplicate query-method or scope lists in `@method` annotations.
 - Investigate static analysis that turns the coroutine-safety and worker-lifetime rules in `AGENTS.md` into automated checks, so mistakes no longer depend on agent steering and review alone. Decide which tool fits each check: PHPStan rules for detection with type information, Rector in dry-run mode, or another linter. Start with property writes outside the constructor on container-resolved classes (auto-singletons and singletons) that are not `Transient`, scoped, or execution-scoped, since per-request data stored there leaks to concurrent requests in the same worker. Intentional worker-lifetime state such as manager driver caches, registries, and resolved configuration needs an explicit way to opt out, so measure false positives across `src/` before choosing that mechanism. Then evaluate other patterns: static property writes outside boot code and `flushState()`; static caches keyed by request-derived values; request state captured in constructors of classes that are not execution-scoped; `Config::set()` during request handling; child coroutines started inside `DB::transaction()`, which borrow their own pooled connection and therefore run outside the transaction; and raw `sleep()`/`usleep()` in source instead of `Sleep`. Ship the useful rules so applications and private packages can enable the same checks.
+- Investigate a focused PHPStan rule for cancellation checks that miss cleanup exception chains. Detect bare `CanceledException` checks around callbacks or listeners, including `releaseIdleConnections()`, where `Coroutine::causedByCancellation()` is needed to prevent retries or continued work after cancellation. Use the cancellation regression cases to assess detection and false positives; preserve valid native-only checks, cleanup-error reporting and foreign-cancellation boundaries. Integrate useful checks with the coroutine-safety analysis above rather than attempting to infer every ownership decision automatically.
 - Audit typed input accessors across `InteractsWithData`, Support `DataObject`, and Hypervel Data. Define the accepted integer, float, and boolean forms for each public contract. Extract a neutral Support conversion primitive only if `InteractsWithData` and `DataObject` converge on exactly the same strict semantics, and decide separately whether Hypervel Data should adopt those semantics rather than changing its tested permissive casts as a side effect.
 
 ## Testing

src/ai/src/Gateway/TextGenerationLoop.php

@@ -65,6 +65,7 @@
 use Hypervel\Ai\Tools\Request;
 use Hypervel\Ai\Tools\ToolNameResolver;
 use Hypervel\Context\CoroutineContext;
+use Hypervel\Coroutine\Coroutine;
 use Hypervel\Coroutine\WaitConcurrent;
 use Hypervel\Database\ConnectionResolver;
 use Hypervel\Engine\Channel;
@@ -1163,9 +1164,11 @@ protected function resolveApprovalResults(array $approval, array $messages, arra
 
             try {
                 $result = $this->executeTool($tool, $arguments, $toolCall->id, $context);
-            } catch (CanceledException $exception) {
-                throw $exception;
             } catch (Throwable $exception) {
+                if (Coroutine::causedByCancellation($exception)) {
+                    throw $exception;
+                }
+
                 $failed = true;
                 $result = 'The tool call failed: ' . $exception->getMessage();
             }

src/ai/src/Middleware/RememberConversation.php

@@ -22,10 +22,10 @@
 use Hypervel\Ai\Storage\StoredTurn;
 use Hypervel\Ai\Support\ConversationSelection;
 use Hypervel\Ai\Support\PendingConversationTitle;
+use Hypervel\Coroutine\Coroutine;
 use Hypervel\Support\ClassMetadataCache;
 use Hypervel\Support\Facades\DB;
 use Hypervel\Support\Str;
-use Swoole\Coroutine\CanceledException;
 use Throwable;
 use WeakReference;
 
@@ -207,10 +207,14 @@ protected function rememberTurn(AgentPrompt $prompt, AgentResponse $response, ?s
         $cancellation = null;
 
         if ($newConversation) {
-            if (! $exception instanceof CanceledException && ! $exception instanceof StreamAbandonedException) {
+            if (($exception === null || ! Coroutine::causedByCancellation($exception)) && ! $exception instanceof StreamAbandonedException) {
                 try {
                     $title = $prompt->pendingConversationTitle()?->value() ?? $this->generateTitle($prompt->prompt);
-                } catch (CanceledException $caught) {
+                } catch (Throwable $caught) {
+                    if (! Coroutine::causedByCancellation($caught)) {
+                        throw $caught;
+                    }
+
                     $cancellation = $caught;
                 }
             }
@@ -367,9 +371,11 @@ protected function generateTitle(string $prompt): string
             );
 
             return Str::limit($response->text, 100);
-        } catch (CanceledException $exception) {
-            throw $exception;
-        } catch (Throwable) {
+        } catch (Throwable $exception) {
+            if (Coroutine::causedByCancellation($exception)) {
+                throw $exception;
+            }
+
             return Str::limit($prompt, 100, preserveWords: true);
         }
     }

src/ai/src/Responses/StreamableAgentResponse.php

@@ -25,7 +25,6 @@
 use Hypervel\Http\Request;
 use Hypervel\Support\Collection;
 use IteratorAggregate;
-use Swoole\Coroutine\CanceledException;
 use Symfony\Component\HttpFoundation\Response;
 use Throwable;
 use Traversable;
@@ -146,16 +145,20 @@ public function broadcast(Channel|array $channels, bool $now = true, array $exce
                     $send($event);
                 }
             });
-        } catch (CanceledException $exception) {
-            $batcher?->cancel();
-
-            throw $exception;
         } catch (Throwable $exception) {
+            if (Coroutine::causedByCancellation($exception)) {
+                $batcher?->cancel();
+
+                throw $exception;
+            }
+
             try {
                 $batcher?->finish();
-            } catch (CanceledException $cancellation) {
-                throw $cancellation;
             } catch (Throwable $senderFailure) {
+                if (Coroutine::causedByCancellation($senderFailure)) {
+                    throw $senderFailure;
+                }
+
                 if ($senderFailure !== $exception) {
                     report($senderFailure);
                 }

src/ai/src/Streaming/BroadcastBatcher.php

@@ -8,6 +8,7 @@
 use Hypervel\Ai\Streaming\Events\ReasoningDelta;
 use Hypervel\Ai\Streaming\Events\StreamEvent;
 use Hypervel\Ai\Streaming\Events\TextDelta;
+use Hypervel\Coroutine\Coroutine;
 use Hypervel\Coroutine\Exceptions\ChildCancellationException;
 use Hypervel\Coroutine\WaitConcurrent;
 use Hypervel\Database\ConnectionResolver;
@@ -52,10 +53,10 @@ public function __construct(
             try {
                 $work = fn () => $this->drain($send);
                 $runner === null ? $work() : $runner($work);
-            } catch (CanceledException $exception) {
-                $this->failure = new ChildCancellationException('The broadcast sender was canceled while its owner remained active.', previous: $exception);
             } catch (Throwable $exception) {
-                $this->failure = $exception;
+                $this->failure = Coroutine::causedByCancellation($exception)
+                    ? new ChildCancellationException('The broadcast sender was canceled while its owner remained active.', previous: $exception)
+                    : $exception;
             } finally {
                 // Wake a producer blocked on a full channel if the sender fails.
                 $this->channel->close();
@@ -98,12 +99,12 @@ public function finish(): void
         try {
             // Completion callbacks may have borrowed the slot needed to send the final batches.
             ConnectionResolver::releaseIdleConnections();
-        } catch (CanceledException $exception) {
-            $this->cancel();
-
-            throw $exception;
         } catch (Throwable $exception) {
-            $this->concurrent->wait();
+            if (Coroutine::causedByCancellation($exception)) {
+                $this->cancel();
+            } else {
+                $this->concurrent->wait();
+            }
 
             throw $exception;
         }

src/ai/src/Tools/AgentTool.php

@@ -15,8 +15,8 @@
 use Hypervel\Ai\Streaming\Events\StreamEvent;
 use Hypervel\Contracts\Container\Transient;
 use Hypervel\Contracts\JsonSchema\JsonSchema;
+use Hypervel\Coroutine\Coroutine;
 use Stringable;
-use Swoole\Coroutine\CanceledException;
 use Throwable;
 
 class AgentTool implements Approvable, Tool, Transient
@@ -60,9 +60,11 @@ public function handle(Request $request): string
     {
         try {
             return $this->agent->prompt((string) $request['task'])->text;
-        } catch (CanceledException $exception) {
-            throw $exception;
         } catch (Throwable $throwable) {
+            if (Coroutine::causedByCancellation($throwable)) {
+                throw $throwable;
+            }
+
             // Degraded to a result so the parent run survives; the failure surfaces as the sub-agent's own AgentFailed...
             return 'Agent failed: ' . $throwable->getMessage();
         }
@@ -85,9 +87,11 @@ public function stream(Request $request): Generator
             yield from $stream;
 
             return (string) $stream->text;
-        } catch (CanceledException $exception) {
-            throw $exception;
         } catch (Throwable $throwable) {
+            if (Coroutine::causedByCancellation($throwable)) {
+                throw $throwable;
+            }
+
             return 'Agent failed: ' . $throwable->getMessage();
         }
     }

src/ai/src/Tools/Filesystem/CopyFile.php

@@ -7,7 +7,7 @@
 use Hypervel\Ai\Attributes\Strict;
 use Hypervel\Ai\Tools\Request;
 use Hypervel\Contracts\JsonSchema\JsonSchema;
-use Swoole\Coroutine\CanceledException;
+use Hypervel\Coroutine\Coroutine;
 use Throwable;
 
 #[Strict]
@@ -37,9 +37,11 @@ public function handle(Request $request): string
             }
 
             $copied = $disk->copy($from, $to);
-        } catch (CanceledException $exception) {
-            throw $exception;
         } catch (Throwable $throwable) {
+            if (Coroutine::causedByCancellation($throwable)) {
+                throw $throwable;
+            }
+
             return "Unable to copy [{$from}] to [{$to}]: {$throwable->getMessage()}";
         }
 

src/ai/src/Tools/Filesystem/GetFileMetadata.php

@@ -7,7 +7,7 @@
 use Hypervel\Ai\Attributes\Strict;
 use Hypervel\Ai\Tools\Request;
 use Hypervel\Contracts\JsonSchema\JsonSchema;
-use Swoole\Coroutine\CanceledException;
+use Hypervel\Coroutine\Coroutine;
 use Throwable;
 
 #[Strict]
@@ -32,9 +32,11 @@ public function handle(Request $request): string
 
         try {
             $size = $disk->size($path);
-        } catch (CanceledException $exception) {
-            throw $exception;
-        } catch (Throwable) {
+        } catch (Throwable $exception) {
+            if (Coroutine::causedByCancellation($exception)) {
+                throw $exception;
+            }
+
             return "File [{$path}] does not exist.";
         }
 

src/ai/src/Tools/Filesystem/GetFileUrl.php

@@ -7,7 +7,7 @@
 use Hypervel\Ai\Attributes\Strict;
 use Hypervel\Ai\Tools\Request;
 use Hypervel\Contracts\JsonSchema\JsonSchema;
-use Swoole\Coroutine\CanceledException;
+use Hypervel\Coroutine\Coroutine;
 use Throwable;
 
 #[Strict]
@@ -43,9 +43,11 @@ public function handle(Request $request): string
                 ? $disk->temporaryUrl($path, now()->addMinutes($minutes))
                 // @phpstan-ignore method.notFound
                 : $disk->url($path);
-        } catch (CanceledException $exception) {
-            throw $exception;
         } catch (Throwable $throwable) {
+            if (Coroutine::causedByCancellation($throwable)) {
+                throw $throwable;
+            }
+
             return "Unable to generate a URL for [{$path}]: {$throwable->getMessage()}";
         }
     }

src/ai/src/Tools/Filesystem/MoveFile.php

@@ -7,7 +7,7 @@
 use Hypervel\Ai\Attributes\Strict;
 use Hypervel\Ai\Tools\Request;
 use Hypervel\Contracts\JsonSchema\JsonSchema;
-use Swoole\Coroutine\CanceledException;
+use Hypervel\Coroutine\Coroutine;
 use Throwable;
 
 #[Strict]
@@ -37,9 +37,11 @@ public function handle(Request $request): string
             }
 
             $moved = $disk->move($from, $to);
-        } catch (CanceledException $exception) {
-            throw $exception;
         } catch (Throwable $throwable) {
+            if (Coroutine::causedByCancellation($throwable)) {
+                throw $throwable;
+            }
+
             return "Unable to move [{$from}] to [{$to}]: {$throwable->getMessage()}";
         }
 

src/ai/src/Tools/Filesystem/ReadFile.php

@@ -7,7 +7,7 @@
 use Hypervel\Ai\Attributes\Strict;
 use Hypervel\Ai\Tools\Request;
 use Hypervel\Contracts\JsonSchema\JsonSchema;
-use Swoole\Coroutine\CanceledException;
+use Hypervel\Coroutine\Coroutine;
 use Throwable;
 
 #[Strict]
@@ -37,9 +37,11 @@ public function handle(Request $request): string
 
         try {
             $size = $disk->size($path);
-        } catch (CanceledException $exception) {
-            throw $exception;
-        } catch (Throwable) {
+        } catch (Throwable $exception) {
+            if (Coroutine::causedByCancellation($exception)) {
+                throw $exception;
+            }
+
             return "File [{$path}] does not exist.";
         }
 

src/cache/src/FailoverStore.php

@@ -15,8 +15,8 @@
 use Hypervel\Contracts\Cache\RawReadable;
 use Hypervel\Contracts\Cache\Repository as RepositoryContract;
 use Hypervel\Contracts\Events\Dispatcher;
+use Hypervel\Coroutine\Coroutine;
 use RuntimeException;
-use Swoole\Coroutine\CanceledException;
 use Throwable;
 use UnitEnum;
 use WeakReference;
@@ -296,9 +296,11 @@ public function flushLocks(): bool
                 if (! $store->flushLocks()) {
                     $result = false;
                 }
-            } catch (CanceledException $exception) {
-                throw $exception;
             } catch (Throwable $throwable) {
+                if (Coroutine::causedByCancellation($throwable)) {
+                    throw $throwable;
+                }
+
                 $exception ??= $throwable;
             }
         }
@@ -361,9 +363,11 @@ protected function attemptOnAllStores(Closure|string $method, array $arguments =
                     return $method instanceof Closure
                         ? $method($repository)
                         : $repository->{$method}(...$arguments);
-                } catch (CanceledException $exception) {
-                    throw $exception;
                 } catch (Throwable $exception) {
+                    if (Coroutine::causedByCancellation($exception)) {
+                        throw $exception;
+                    }
+
                     $lastException = $exception;
 
                     $failedCaches[] = $store;
@@ -405,9 +409,11 @@ protected function attemptOnEveryStore(string $method, array $arguments): array
             foreach ($this->stores as $position => $store) {
                 try {
                     $results[] = $this->store($store)->{$method}(...$arguments);
-                } catch (CanceledException $exception) {
-                    throw $exception;
                 } catch (Throwable $exception) {
+                    if (Coroutine::causedByCancellation($exception)) {
+                        throw $exception;
+                    }
+
                     $lastException = $exception;
                     $failedCaches[] = $store;