feat(fiber): coroutine scope — Coroutine::spawn()/await()/concurrently()

bridge/rector/FEATURE_PARITY.md

@@ -31,7 +31,7 @@ Conversion coverage across the three directions supported by `testo/bridge-recto
 | **Mocks** (`createMock`/`getMockBuilder`/`prophesize`) | ➖ | ⛔ *Testo has no built-in mocking* | ➖ |
 | **Memory-leak expectations** | ⛔ *no PHPUnit equivalent* | ➖ | ➖ |
 | **Retry / Repeat** (`#[Retry]`/`#[Repeat]`) | ⛔ *no PHPUnit equivalent* | ➖ | ➖ |
-| **Fiber** (`#[RunInFiber]`) | ⛔ *no PHPUnit/Pest equivalent — neither has a fiber/coroutine test attribute* | ➖ | ➖ |
+| **Fiber** (`#[RunInFiber]`, `Coroutine::spawn/await/concurrently`) | ⛔ *no PHPUnit/Pest equivalent — neither has a fiber/coroutine test attribute or an in-test coroutine scope* | ➖ | ➖ |
 | **`uses()`** (Pest) | ➖ | ➖ | ⛔ *a converted function has no base class, traits or `$this` to attach to; closures that capture `$this`-shared state are left untouched* |
 | **`arch()` tests** (Pest) | ➖ | ➖ | ⛔ *Testo has no arch-assertion subsystem* |
 

core/Pipeline/Attribute/FallbackInterceptor.php

@@ -15,11 +15,14 @@
  *  final class RetryPolicy {}
  * ```
  *
+ * Repeatable: an attribute may wire several interceptors, each with its own pipeline position
+ * ({@see InterceptorOptions}); every one of them is instantiated with the attribute instance.
+ *
  * Makes sense only for interceptors that are executed during tests execution.
  *
  * @api
  */
-#[\Attribute(\Attribute::TARGET_CLASS)]
+#[\Attribute(\Attribute::TARGET_CLASS | \Attribute::IS_REPEATABLE)]
 final class FallbackInterceptor
 {
     public function __construct(

core/Pipeline/InterceptorProvider.php

@@ -90,12 +90,15 @@ public function fromAttributes(string $class, Interceptable ...$attributes): arr
         $result = [];
 
         foreach ($attributes as $attribute) {
-            # Get alias interceptor
-            $iClass = Cache::resolveAlias($attribute::class) ?? throw new \RuntimeException(
+            # Get alias interceptors
+            $iClasses = Cache::resolveAliases($attribute::class);
+            $iClasses === [] and throw new \RuntimeException(
                 \sprintf('No interceptor found for attribute %s.', $attribute::class),
             );
 
-            \is_a($iClass, $class, true) and $result[] = $this->createInstance($iClass, [$attribute]);
+            foreach ($iClasses as $iClass) {
+                \is_a($iClass, $class, true) and $result[] = $this->createInstance($iClass, [$attribute]);
+            }
         }
 
         return $result;

core/Pipeline/Internal/Cache.php

@@ -18,17 +18,17 @@
 final class Cache
 {
     /**
-     * @var array<class-string<Interceptable>, null|class-string<Interceptor>>
+     * @var array<class-string<Interceptable>, list<class-string<Interceptor>>>
      */
     private static array $map = [];
 
     /**
-     * Resolve alias interceptor for the given attribute class.
+     * Resolve alias interceptors for the given attribute class.
      *
      * @param class-string<Interceptable> $class The attribute class.
-     * @return class-string<Interceptor>|null The interceptor class or null if not found.
+     * @return list<class-string<Interceptor>> The interceptor classes; empty if none found.
      */
-    public static function resolveAlias(string $class): ?string
+    public static function resolveAliases(string $class): array
     {
         $c = $class;
         do {
@@ -40,11 +40,14 @@ public static function resolveAlias(string $class): ?string
         } while ($c);
 
         /**
-         * Resolve fallback handler from the {@see FallbackInterceptor} attribute
+         * Resolve fallback handlers from the repeatable {@see FallbackInterceptor} attribute
          * @var list<\ReflectionAttribute<FallbackInterceptor>> $attrs
          */
         $attrs = Reflection::fetchClassAttributes($class, attributeClass: FallbackInterceptor::class);
 
-        return self::$map[$class] ??= $attrs === [] ? null : $attrs[0]->newInstance()->class;
+        return self::$map[$class] ??= \array_map(
+            static fn(\ReflectionAttribute $attr): string => $attr->newInstance()->class,
+            $attrs,
+        );
     }
 }

plugin/fiber/README.md

@@ -28,10 +28,25 @@
 Runs tests as plain PHP fibers driven by Testo's own cooperative scheduler, so a test (or the code it exercises) may suspend with `\Fiber::suspend()` and be resumed, and a case's tests may be interleaved to shake out order-dependent races.
 
 - `#[RunInFiber]` — run a test (method) or a whole case (class) inside fibers, scheduled by `Schedule::Solo` (default), `RoundRobin` or `Random`.
+- `Coroutine::spawn()` / `->await()` / `Coroutine::concurrently()` — add coroutines to the running test's schedule and wait for them; they interleave with the test body (and, under a class-level `#[RunInFiber]`, with the case's other tests) at every suspension point.
+
+```php
+#[RunInFiber]
+public function pingPong(): void
+{
+    $server = Coroutine::spawn(fn(): string => $this->acceptAndEcho());
+    $client = Coroutine::spawn(fn(): string => $this->connectAndSend('ping'));
+
+    Assert::same($client->await(), 'pong');
+    Assert::same($server->await(), 'ping');
+}
+```
+
+The scope is structured: the test is not finished until every coroutine it spawned is. Coroutine failures always surface wrapped in a `CompositeException` — even a single one; if the test body fails, pending coroutines are cancelled with a `CancelledException` thrown into them, and an await cycle is broken with a `DeadlockException` at the guilty `await()`.
 
 Switching is cooperative and happens only at suspension points — there is no event loop and no preemption. This is for fiber-based/cooperative code and race hunting, **not** for real async I/O: awaiting a timer, socket or `Future` needs the Revolt event loop — use the `testo/bridge-revolt` `#[RunInRevolt]` attribute for that.
 
-The attribute lives under the `Testo\Fiber\` namespace.
+Everything lives under the `Testo\Fiber\` namespace.
 
 ## Install
 

plugin/fiber/src/Coroutine.php

@@ -0,0 +1,144 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Testo\Fiber;
+
+use Testo\Fiber\Exception\CancelledException;
+use Testo\Fiber\Exception\CompositeException;
+use Testo\Fiber\Internal\Scheduler;
+use Testo\Fiber\Internal\Task;
+
+/**
+ * A handle to a coroutine spawned into the running test's cooperative scope.
+ *
+ * Every {@see RunInFiber} test runs inside its own coroutine scope: the test body is the scope's
+ * first coroutine, and {@see spawn()} adds more to the same schedule. Coroutines interleave with the
+ * test body (and each other) wherever a fiber calls `\Fiber::suspend()`, and — under a class-level
+ * `#[RunInFiber]` — the whole scope keeps interleaving with the case's other tests. Assertions and
+ * messages inside a coroutine are attributed to the test that spawned it.
+ *
+ * ```php
+ *  #[RunInFiber]
+ *  public function pingPong(): void
+ *  {
+ *      $server = Coroutine::spawn(fn(): string => $this->acceptAndEcho());
+ *      $client = Coroutine::spawn(fn(): string => $this->connectAndSend('ping'));
+ *
+ *      Assert::same($client->await(), 'pong');
+ *      Assert::same($server->await(), 'ping');
+ *  }
+ * ```
+ *
+ * The scope is structured: the test is not finished until every coroutine it spawned is. A coroutine
+ * still pending when the test body returns keeps being driven; if the body fails, pending coroutines
+ * are cancelled ({@see CancelledException} is thrown into them). Coroutine failures are always
+ * surfaced wrapped in a {@see CompositeException} — even a single one — whether rethrown by
+ * {@see await()} / {@see concurrently()} or reported by the scope for a coroutine nobody awaited.
+ *
+ * @api
+ */
+final readonly class Coroutine
+{
+    private function __construct(
+        private Task $task,
+    ) {}
+
+    /**
+     * Schedule a closure (or an unstarted fiber) as a coroutine of the running test's scope.
+     *
+     * The coroutine gets its first step in the current scheduling round; from there it runs
+     * cooperatively — it holds the floor until it suspends, finishes, or awaits.
+     *
+     * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]` —
+     *         or when the scope is already closing (spawning from a cancelled coroutine's `finally`).
+     */
+    public static function spawn(\Closure|\Fiber $body): self
+    {
+        $scheduler = Scheduler::current() ?? throw new \LogicException(
+            'No active coroutine scope — run the test with #[RunInFiber] to use Coroutine::spawn().',
+        );
+
+        return new self($scheduler->spawn($body));
+    }
+
+    /**
+     * Run the given closures/fibers concurrently and wait for all of them.
+     *
+     * Sugar over {@see spawn()} + {@see await()}: schedules everything into the running scope, parks
+     * the caller until every coroutine finished, and returns the results keyed like the arguments
+     * (named arguments give string keys). Failures are collected until all coroutines settle, then
+     * bundled into one {@see CompositeException} — its errors keyed like the arguments too,
+     * symmetric to the results. A coroutine that itself died with a `CompositeException` appears
+     * nested: that whole exception sits under the argument's key, its own structure intact.
+     *
+     * @return array<array-key, mixed> Results keyed like the arguments.
+     *
+     * @throws CompositeException When any of the coroutines threw — errors keyed like the arguments.
+     * @throws \LogicException When no coroutine scope is active — run the test with `#[RunInFiber]`.
+     */
+    public static function concurrently(\Closure|\Fiber ...$bodies): array
+    {
+        $handles = \array_map(self::spawn(...), $bodies);
+
+        $results = $errors = [];
+        foreach ($handles as $key => $handle) {
+            try {
+                $results[$key] = $handle->await();
+            } catch (CompositeException $e) {
+                // await() wraps exactly one task's error — unwrap and re-key it by the argument,
+                // so callers never see the scheduler's internal task ids.
+                $errors[$key] = $e->errors[\array_key_first($e->errors)];
+            }
+        }
+
+        $errors === [] or throw new CompositeException($errors);
+
+        return $results;
+    }
+
+    /**
+     * Whether the coroutine has settled — returned, thrown, or been cancelled.
+     */
+    public function isFinished(): bool
+    {
+        return $this->task->finished;
+    }
+
+    /**
+     * Park the calling coroutine until this one finishes, and return its result.
+     *
+     * Other coroutines keep running while the caller is parked. Rethrowing a failure here marks it
+     * as observed, so the scope will not report it again. A cancellation is rethrown unwrapped — it
+     * is the scope's control signal, not a failure of the coroutine.
+     *
+     * @throws CompositeException When the awaited coroutine threw.
+     * @throws CancelledException When the awaited coroutine was cancelled with its scope.
+     * @throws \LogicException When called outside a coroutine scope, or when a coroutine awaits itself.
+     */
+    public function await(): mixed
+    {
+        while (!$this->task->finished) {
+            $caller = Scheduler::current()?->runningTask() ?? throw new \LogicException(
+                'Coroutine::await() on a pending coroutine must be called from inside a coroutine scope.',
+            );
+            $caller === $this->task and throw new \LogicException('A coroutine cannot await itself.');
+
+            $caller->awaiting = $this->task;
+            try {
+                \Fiber::suspend();
+            } finally {
+                $caller->awaiting = null;
+            }
+        }
+
+        if ($this->task->error !== null) {
+            $this->task->errorObserved = true;
+            throw new CompositeException([$this->task->id => $this->task->error]);
+        }
+
+        $this->task->cancelled and throw new CancelledException('The awaited coroutine was cancelled with its scope.');
+
+        return $this->task->result;
+    }
+}

plugin/fiber/src/Exception/CancelledException.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Testo\Fiber\Exception;
+
+/**
+ * Thrown into a pending coroutine's fiber when its scope is torn down — the test body failed, so the
+ * coroutine will not be driven any further.
+ *
+ * Raised at the coroutine's current suspension point, so `finally` blocks run as the fiber unwinds.
+ * Don't swallow it: a coroutine that catches the cancellation and suspends again is resumed until it
+ * terminates, but it has no schedule to cooperate with anymore.
+ *
+ * Also rethrown by {@see \Testo\Fiber\Coroutine::await()} on a cancelled coroutine — it has no
+ * result to report.
+ *
+ * @api
+ */
+final class CancelledException extends \RuntimeException {}

plugin/fiber/src/Exception/CompositeException.php

@@ -5,41 +5,48 @@
 namespace Testo\Fiber\Exception;
 
 /**
- * Aggregates every throwable raised by the fibers of a single {@see \Testo\Fiber\RunInFiber} case batch.
+ * Aggregates throwables raised by scheduled fibers — a {@see \Testo\Fiber\RunInFiber} case batch, or
+ * coroutines of a test's scope ({@see \Testo\Fiber\Coroutine}).
  *
- * Testo's per-test pipeline never throws (a failure becomes a result), so the fiber scheduler surfacing
- * even one throwable means something broke below the pipeline. When more than one fiber fails in an
- * interleaved run this bundles them all, instead of dropping every failure but the first — the
- * individual throwables stay reachable via {@see self::$errors}, and the earliest is chained as
- * {@see \Throwable::getPrevious()} so ordinary renderers still show a root cause.
+ * Coroutine failures are **always** surfaced through this wrapper, even a single one — whether
+ * rethrown by `await()` / `concurrently()` or reported by the scope for a coroutine nobody awaited —
+ * so handling code is uniform. The individual throwables stay reachable via {@see self::$errors},
+ * and the earliest is chained as {@see \Throwable::getPrevious()} so ordinary renderers still show
+ * a root cause.
  *
  * @api
  */
 final class CompositeException extends \RuntimeException
 {
     /**
-     * The collected throwables, keyed by the fiber (test) index that raised each one.
+     * The collected throwables, keyed by whatever names each fiber to the producer: the task id for
+     * scope/batch failures, or the argument key for {@see \Testo\Fiber\Coroutine::concurrently()}.
      *
-     * @var non-empty-array<int, \Throwable>
+     * @var non-empty-array<array-key, \Throwable>
      */
     public readonly array $errors;
 
     /**
-     * @param non-empty-array<int, \Throwable> $errors
+     * @param non-empty-array<array-key, \Throwable> $errors
      */
     public function __construct(array $errors)
     {
         $this->errors = $errors;
 
         $lines = \array_map(
-            static fn(int $i, \Throwable $e): string => \sprintf('  #%d %s: %s', $i, $e::class, $e->getMessage()),
+            static fn(int|string $key, \Throwable $e): string => \sprintf(
+                '  %s %s: %s',
+                \is_int($key) ? "#$key" : $key,
+                $e::class,
+                $e->getMessage(),
+            ),
             \array_keys($errors),
             \array_values($errors),
         );
 
         parent::__construct(
             \sprintf(
-                "%d test fiber(s) failed while running the case batch:\n%s",
+                "%d fiber(s) failed:\n%s",
                 \count($errors),
                 \implode("\n", $lines),
             ),

plugin/fiber/src/Exception/DeadlockException.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Testo\Fiber\Exception;
+
+/**
+ * A coroutine is parked on an {@see \Testo\Fiber\Coroutine::await()} that can never complete — an
+ * await cycle, including one spanning several tests' scopes when handles are shared under a
+ * class-level `#[RunInFiber]`.
+ *
+ * The scheduler breaks the cycle by raising this at the first doomed coroutine's `await()` call, so
+ * the stack trace points at the guilty wait; the failure then cascades to the coroutines awaiting it.
+ * A bare `\Fiber::suspend()` loop waiting for something that never happens is **not** detected — only
+ * `await()` parks a coroutine in a way the scheduler can reason about.
+ *
+ * @api
+ */
+final class DeadlockException extends \RuntimeException {}