Pipeline workflow task completions after local activities (opt-in)

External performance report, topic 19. The reported timings and proposed fixes have not been independently reproduced. The supplied patches are reference drafts for review and current-source qualification.

Problem

In five-local-steps the worker, not the Server, bounds the race. Each
workflow costs a worker about 120 ms (p50):

replay plus five 20 ms local activities: about 108 ms;
the completion round trip: 10–11 ms;
the query poll: about 1.5 ms.

The proposed optimization overlaps completion I/O with useful work on one already-leased task. The ordering and failure semantics need independent qualification.

Proposal

Behind a Worker option, off by default at first, and only in run() on the
default transport:

The transport can let go of a sent request. The default curl_multi
transport suspends the armed Fiber once its request body is sent in full.
It finishes reading the answer when the Fiber resumes. Only the armed
Fiber's first request suspends. Retries, injected PSR-18 clients, streams
and progress callbacks never do.
Lease the next task first. Before a completion that follows a local
activity, the Worker polls with a 0-second timeout. If that poll leases a
task:
- the Worker sends the completion from a Fiber;
- it runs the leased task while the Server answers;
- it takes the answer, with the lease its carried poll took (16b),
  before that task completes.
What stays synchronous. These completions are still sent at once:
- one that schedules an activity;
- one under a cancellation request;
- one after a kept failure;
- every completion from tick(), and every completion with telemetry or
  metrics on.
Failures keep their meaning. A lost claim is let go as before. A
refusal that stops the Worker, or a failure of the poll that leased the
next task, is thrown once the next task is done.
A Worker holds at most one workflow task ahead. While it does, an
activity completion carries no workflow poll.
Read a missing workflow_poll key as "leased nothing" (11a). The
first cut read it as "unknown", counted the poll as still held, and
stopped carrying workflow polls. Five-steps then sent 3,036 workflow
polls instead of 36.

Reported benchmark evidence

Reporter-supplied cumulative measurements, not an isolated-effect or independently verified performance claim: 200 sequentially started workflows, five sequential 20ms activities or five local activities each, six PHP worker processes, mean of three repetitions, one machine, fresh SQLite/MySQL8.0/PostgreSQL16 databases with default durability. "Five" and "local" below mean last-close wall time in seconds; S/M/P means SQLite/MySQL/PostgreSQL. Preserve regressions and noise in the reported results.

Five-local-steps: 4.28 → 3.98 on the slowest backend. S / M / P went
4.22 / 4.22 / 4.28 → 3.92 / 3.93 / 3.98, which is 6–7% on every backend.
That's what the ~10 ms round trip per ~120 ms workflow predicts.
Five-steps pipelines nothing and sends the same requests as before.

Acceptance

Fourteen tests, in tests/WorkerCompletionPipeliningTest.php and
tests/Psr18TransportSuspensionTest.php, cover:
- the order of sends and answers;
- each synchronous case;
- the failure paths;
- the transport against a real socket;
- the missing-key answer.
With the option off, the Worker sends exactly what it sends today.

Implementation constraints

The initial implementation must be opt-in. It changes when the
worker learns a completion failed: after the next task, not before it.

Ship it opt-in and qualify it under realistic load. Any default change needs separate compatibility and rollout evidence.
The suspending transport is new code on every request, so it needs an
owner.

Dependencies

Related existing work: #66, #137.

Draft patches and validation boundary

Current main was independently read as afa5a95e3116cd69c731c821f00e8223d42f7ff6. The supplied draft series targets afa5a95e3116cd69c731c821f00e8223d42f7ff6; patches are cumulative reference material, not a current-main application or test result. Review and rebase on maintained source, preserve all existing tests and published compatibility, then establish causal controls and actual conformance before shipping. No patch was applied or executed during intake.

Known draft gap / required control: The supplied draft does not yet implement the requested opt-in option. Do not enable pipelining by default. Retain synchronous tick(), cancellation/failure/telemetry paths, lease/fencing rules and one-task-ahead bounds, and define transport lifecycle ownership before adoption.

The following sanitized unified diffs identify proposed reference material:

0009-t19-Pipeline-workflow-task-completions-after-local-activ.patch — 68,098 bytes, SHA-256 e218bcb68d8d32a36234986205d994c9ecc2279482d1110649515b2b40e93f8e.

Next action: the owning maintainer reviews the candidate against current source and the acceptance criteria above. Preserve durability, lease/fencing, idempotency, fairness and whole-completion/history budgets. New producer capabilities need versioned schemas and real producer/consumer compatibility proof; historical/frozen evidence is not rewritten.

Sanitized reference diff 0009-t19-Pipeline-workflow-task-completions-after-local-activ.patch, part 1/2.

This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

Complete sanitized patch: 68,098 bytes, SHA-256 e218bcb68d8d32a36234986205d994c9ecc2279482d1110649515b2b40e93f8e. This part: 46,737 bytes, SHA-256 2cc2591459fb443b6707ca53edc67ddf25b53be95283c698517155e016c8b6f2. Parts concatenate in order; a part boundary is between complete file diffs.

diff --git a/src/Client.php b/src/Client.php
index c77ebb8..e9a16b1 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -300,6 +300,19 @@ final class Client implements WorkflowClientInterface
         return $copy;
     }
 
+    /**
+     * @internal The next request $fiber sends suspends it once the request is
+     * sent in full, and finishes when the Fiber resumes; null clears that.
+     * False when this Client's transport cannot send that way.
+     *
+     * @param \Fiber<mixed, mixed, mixed, mixed>|null $fiber
+     */
+    public function suspendSendsOf(?\Fiber $fiber): bool
+    {
+        return $this->transport instanceof \DurableWorkflow\Transport\SuspendingTransport
+            && $this->transport->suspendSendsOf($fiber);
+    }
+
     /** @return array<string, mixed>|null The carried poll answer this request delivers instead of sending. */
     private function carriedPollAnswerFor(string $method, string $path): ?array
     {
diff --git a/src/Transport/Psr18Transport.php b/src/Transport/Psr18Transport.php
index 8cc679e..355b6b7 100644
--- a/src/Transport/Psr18Transport.php
+++ b/src/Transport/Psr18Transport.php
@@ -23,7 +23,7 @@ use GuzzleHttp\HandlerStack;
 use Throwable;
 
 /** Default PSR-18 JSON and bounded runtime-payload transport. */
-final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUploadTransport, BoundedTransport, BoundedProgressPayloadTransport, ResponseStatusTransport
+final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUploadTransport, BoundedTransport, BoundedProgressPayloadTransport, ResponseStatusTransport, SuspendingTransport
 {
     public const DEFAULT_REQUEST_TIMEOUT_SECONDS = 60.0;
     /** Worker lifecycle requests (register, heartbeat, complete, fail, deregister); long polls keep their own budget. */
@@ -36,6 +36,10 @@ final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUpl
     private readonly ?float $requestTimeoutSeconds;
     private readonly ?float $workerRequestTimeoutSeconds;
     private ?GuzzleClient $nativeControlClient = null;
+    /** @var \Fiber<mixed, mixed, mixed, mixed>|null The Fiber whose next request suspends once sent (`suspendSendsOf`). */
+    private ?\Fiber $suspending = null;
+    private ?SuspendingCurlHandler $suspendingHandler = null;
+    private ?GuzzleClient $suspendingClient = null;
     private ?int $responseStatus = null;
 
     public function lastResponseStatus(): ?int
@@ -87,6 +91,24 @@ final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUpl
         return $this->client instanceof GuzzleClient && extension_loaded('curl');
     }
 
+    /**
+     * @internal The next request $fiber sends suspends it once the request is
+     * sent in full, and finishes when it resumes; null clears that. Only the
+     * owned Guzzle client with cURL sends this way, so false means no request
+     * will suspend.
+     *
+     * @param \Fiber<mixed, mixed, mixed, mixed>|null $fiber
+     */
+    public function suspendSendsOf(?\Fiber $fiber): bool
+    {
+        if (!$this->defaultProcessConfiguration || !function_exists('curl_multi_exec')) {
+            return false;
+        }
+        $this->suspending = $fiber;
+
+        return true;
+    }
+
     /** @internal Custom HTTP clients require an application bootstrap in a fresh process. */
     public function supportsDefaultActivityLeaseProcess(): bool
     {
@@ -450,13 +472,31 @@ final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUpl
         ] + $this->client->getConfig());
     }
 
+    private function suspendingHttpClient(): GuzzleClient
+    {
+        if (!$this->client instanceof GuzzleClient || $this->suspending === null) {
+            throw new \LogicException('Only the owned Guzzle client can suspend a request.');
+        }
+        $this->suspendingHandler ??= new SuspendingCurlHandler();
+        $this->suspendingHandler->arm($this->suspending);
+        $this->suspending = null;
+
+        return $this->suspendingClient ??= new GuzzleClient([
+            'handler' => HandlerStack::create($this->suspendingHandler),
+        ] + $this->client->getConfig());
+    }
+
     private function sendRequest(RequestInterface $request, bool $stream = false, int|float $timeoutSeconds = 30, bool $bounded = false, ?StreamInterface $sink = null, ?\Closure $progress = null, ?float $controlTimeout = null, bool $control = true): ResponseInterface
     {
         $connectionFailure = false;
         $timedOut = false;
         try {
-            $http = $controlTimeout !== null && !function_exists('curl_exec') && !function_exists('curl_multi_exec')
-                ? $this->nativeControlHttpClient() : $this->client;
+            // A streamed or progress-observed transfer is never suspended.
+            $suspend = $this->suspending !== null && $this->suspending === \Fiber::getCurrent()
+                && !$stream && $sink === null && $progress === null;
+            $http = $suspend ? $this->suspendingHttpClient()
+                : ($controlTimeout !== null && !function_exists('curl_exec') && !function_exists('curl_multi_exec')
+                    ? $this->nativeControlHttpClient() : $this->client);
             if ($http instanceof GuzzleClient) {
                 $onStats = $http->getConfig('on_stats');
 
@@ -473,7 +513,14 @@ final class Psr18Transport implements BoundedPayloadTransport, BoundedPayloadUpl
                     }
                 };
 
-                return $http->send($request, $options);
+                try {
+                    return $http->send($request, $options);
+                } finally {
+                    if ($suspend) {
+                        // Requests sent during the suspension set the status.
+                        $this->responseStatus = null;
+                    }
+                }
             }
 
             return $this->client->sendRequest($request);
diff --git a/src/Transport/SuspendingCurlHandler.php b/src/Transport/SuspendingCurlHandler.php
new file mode 100644
index 0000000..f304ac0
--- /dev/null
+++ b/src/Transport/SuspendingCurlHandler.php
@@ -0,0 +1,109 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Transport;
+
+use GuzzleHttp\Handler\CurlFactory;
+use GuzzleHttp\Handler\HostValidator;
+use GuzzleHttp\Promise\PromiseInterface;
+use Psr\Http\Message\RequestInterface;
+
+/**
+ * @internal Guzzle's synchronous cURL handler, except that the Fiber armed
+ * for the next request suspends once that request is sent in full and the
+ * transfer finishes when the Fiber resumes. The Worker sends a workflow task
+ * completion this way and runs its next task while the Server answers.
+ *
+ * The handles, their options and the results are the cURL handler's: the
+ * transfer runs on this handler's own multi handle, which keeps its
+ * connections between requests, and a deadline the request set still
+ * elapses during the suspension.
+ */
+final class SuspendingCurlHandler
+{
+    private readonly CurlFactory $factory;
+    private ?\CurlMultiHandle $multi = null;
+    /** @var \Fiber<mixed, mixed, mixed, mixed>|null */
+    private ?\Fiber $armed = null;
+
+    public function __construct()
+    {
+        $this->factory = new CurlFactory(3);
+    }
+
+    /**
+     * Suspends $fiber once, when the next request it sends is sent in full.
+     *
+     * @param \Fiber<mixed, mixed, mixed, mixed> $fiber
+     */
+    public function arm(\Fiber $fiber): void
+    {
+        $this->armed = $fiber;
+    }
+
+    /**
+     * @param array<string, mixed> $options
+     * @return PromiseInterface<\Psr\Http\Message\ResponseInterface, mixed>
+     */
+    public function __invoke(RequestInterface $request, array $options): PromiseInterface
+    {
+        // Only the first request is suspended; a retry is sent synchronously.
+        $suspend = $this->armed !== null && $this->armed === \Fiber::getCurrent();
+        $this->armed = null;
+        if (class_exists(HostValidator::class)) {
+            HostValidator::assertRequestHost($request);
+        }
+        if (isset($options['delay']) && is_numeric($options['delay'])) {
+            usleep((int) ((float) $options['delay'] * 1000));
+        }
+        $easy = $this->factory->create($request, $options);
+        // A body of unknown size never counts as sent in full.
+        $easy->errno = $this->transfer($easy->handle, $suspend ? $request->getBody()->getSize() : null);
+
+        return CurlFactory::finish($this, $easy, $this->factory);
+    }
+
+    /**
+     * Runs the transfer to its end, suspending the current Fiber once its
+     * request line, headers and $suspendAfter body bytes are sent.
+     *
+     * @return int The transfer's cURL result code.
+     */
+    private function transfer(\CurlHandle $handle, ?int $suspendAfter): int
+    {
+        $multi = $this->multi ??= curl_multi_init();
+        $added = curl_multi_add_handle($multi, $handle);
+        if ($added !== CURLM_OK) {
+            throw new \RuntimeException('cURL could not start the transfer: '.curl_multi_strerror($added));
+        }
+        try {
+            while (true) {
+                do {
+                    $status = curl_multi_exec($multi, $running);
+                } while ($status === CURLM_CALL_MULTI_PERFORM);
+                if ($status !== CURLM_OK) {
+                    throw new \RuntimeException('cURL could not run the transfer: '.curl_multi_strerror($status));
+                }
+                while (($done = curl_multi_info_read($multi)) !== false) {
+                    if ($done['msg'] === CURLMSG_DONE && $done['handle'] === $handle) {
+                        return $done['result'];
+                    }
+                }
+                if ($suspendAfter !== null && curl_getinfo($handle, CURLINFO_REQUEST_SIZE) > 0
+                    && curl_getinfo($handle, CURLINFO_SIZE_UPLOAD_T) >= $suspendAfter) {
+                    $suspendAfter = null;
+                    \Fiber::suspend();
+
+                    continue;
+                }
+                if (curl_multi_select($multi, 1.0) === -1) {
+                    // No descriptors to wait on yet, as the multi handler does.
+                    usleep(250);
+                }
+            }
+        } finally {
+            curl_multi_remove_handle($multi, $handle);
+        }
+    }
+}
diff --git a/src/Transport/SuspendingTransport.php b/src/Transport/SuspendingTransport.php
new file mode 100644
index 0000000..91ad9d4
--- /dev/null
+++ b/src/Transport/SuspendingTransport.php
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Transport;
+
+/**
+ * @internal A transport that can return to the caller once a request is sent
+ * and finish it later, so the Worker runs its next task while the Server
+ * answers a completion.
+ */
+interface SuspendingTransport
+{
+    /**
+     * The next request $fiber sends suspends it once the request is sent in
+     * full, and finishes when the Fiber resumes; null clears that. False when
+     * no request will suspend.
+     *
+     * @param \Fiber<mixed, mixed, mixed, mixed>|null $fiber
+     */
+    public function suspendSendsOf(?\Fiber $fiber): bool;
+}
diff --git a/src/Worker.php b/src/Worker.php
index f6dcd4d..0f21226 100644
--- a/src/Worker.php
+++ b/src/Worker.php
@@ -135,6 +135,23 @@ final class Worker
     /** @var array{answer: array<string, mixed>, received: float, captured: float}|null The poll answer that completion returned with its lease. */
     private ?array $carriedWorkflowPollAnswer = null;
     private ?string $claimDeliveredCancellationId = null;
+    /** Whether a workflow task completion may be sent while the next task runs (`run()`). */
+    private bool $pipelining = false;
+    /**
+     * @var array{fiber: \Fiber<mixed, mixed, array<string, mixed>, mixed>, task: array<string, mixed>, poll_request_id: string, sent: float, sent_at: float}|null
+     * The completion sent while the next task runs, and the ID of the workflow poll it carries.
+     */
+    private ?array $pendingWorkflowCompletion = null;
+    /** @var array{poll: array<string, mixed>, received: float}|null The workflow poll that leased the next task before a completion was sent. */
+    private ?array $prefetchedWorkflowPoll = null;
+    /** Whether the workflow task being run ran a local activity. */
+    private bool $localActivityRan = false;
+    /**
+     * A failure that stops the Worker from a completion sent while the next
+     * task ran, or from the poll that leased that task, thrown once the task
+     * that saw it is done.
+     */
+    private ?Throwable $deferredFailure = null;
     private bool $preparedLocalActivityGroupsSupported = false;
     /** @var list<string> */
     private array $preparedLocalActivityCancellationPolicies = [];
@@ -418,13 +435,18 @@ final class Worker
             $this->registered = true;
             $this->lastHeartbeatAt = $this->now();
             $this->prepareActivityLeaseMaintenance();
+            // Only a Worker that runs its own loop holds the next task while a
+            // completion is answered, since it alone comes back to finish it.
+            $this->pipelining = $this->telemetry === null && !$this->metricsObserver->enabled()
+                && $this->client->suspendSendsOf(null);
             $this->diagnostic('worker.registered', [
                 'worker_id' => $this->workerId,
                 'task_queue' => $this->taskQueue,
                 'heartbeat_interval_seconds' => $this->heartbeatIntervalSeconds,
             ]);
 
-            while (!$this->shutdownRequested || $this->carriedWorkflowPollLeased === true) {
+            while (!$this->shutdownRequested || $this->carriedWorkflowPollLeased === true
+                || $this->pendingWorkflowCompletion !== null || $this->prefetchedWorkflowPoll !== null) {
                 $this->tick($pollTimeoutSeconds);
                 $this->heartbeatIfDue();
             }
@@ -437,6 +459,22 @@ final class Worker
             ], 'error');
             throw $exception;
         } finally {
+            if ($this->pendingWorkflowCompletion !== null || $this->deferredFailure !== null) {
+                // The loop ended on a failure; the completion still finishes,
+                // and a failure kept for after a task is reported.
+                $taskId = $this->pendingWorkflowCompletion['task']['task_id'] ?? null;
+                try {
+                    $this->settleWorkflowCompletion();
+                    $this->throwDeferredFailure();
+                } catch (Throwable $exception) {
+                    $this->diagnostic('worker.shutdown_failed', [
+                        'worker_id' => $this->workerId,
+                        'task_id' => $taskId,
+                        'exception' => $exception,
+                    ], 'error');
+                }
+            }
+            $this->pipelining = false;
             $this->stopActivityLease();
             $this->closeWorkerSessions();
             $this->stickyCache->clear();
@@ -794,19 +832,34 @@ final class Worker
      */
     public function tick(int $pollTimeoutSeconds = 1): bool
     {
-        // The last completion carried this workflow poll. A lease
-        // its claim took is delivered by the poll under the same request ID,
-        // even during shutdown, as an in-flight poll's would be. The poll
-        // keeps the ID whatever the completion answered, so a lease whose
-        // answer was lost is delivered too. A poll answer the completion
-        // returned with the lease is delivered without sending the poll.
-        $carriedWorkflowRequestId = $this->carriedWorkflowPollRequestId;
-        $carriedWorkflowLeased = $this->carriedWorkflowPollLeased === true;
-        $carriedWorkflowAnswer = $this->carriedWorkflowPollAnswer;
-        $this->carriedWorkflowPollRequestId = null;
-        $this->carriedWorkflowPollLeased = null;
-        $this->carriedWorkflowPollAnswer = null;
-        if ($this->shutdownRequested && !$carriedWorkflowLeased) {
+        // A completion sent while the last task ran is finished first, so
+        // the lease its carried poll took is delivered below, unless a poll
+        // leased the next task before it was sent. That poll's task runs
+        // now, and a lease carried meanwhile waits for the next tick.
+        if ($this->pendingWorkflowCompletion !== null && $this->prefetchedWorkflowPoll === null) {
+            $this->settleWorkflowCompletion();
+            $this->throwDeferredFailure();
+        }
+        $prefetched = $this->prefetchedWorkflowPoll;
+        $this->prefetchedWorkflowPoll = null;
+        $carriedWorkflowRequestId = null;
+        $carriedWorkflowLeased = false;
+        $carriedWorkflowAnswer = null;
+        if ($prefetched === null) {
+            // The last completion carried this workflow poll. A lease
+            // its claim took is delivered by the poll under the same request ID,
+            // even during shutdown, as an in-flight poll's would be. The poll
+            // keeps the ID whatever the completion answered, so a lease whose
+            // answer was lost is delivered too. A poll answer the completion
+            // returned with the lease is delivered without sending the poll.
+            $carriedWorkflowRequestId = $this->carriedWorkflowPollRequestId;
+            $carriedWorkflowLeased = $this->carriedWorkflowPollLeased === true;
+            $carriedWorkflowAnswer = $this->carriedWorkflowPollAnswer;
+            $this->carriedWorkflowPollRequestId = null;
+            $this->carriedWorkflowPollLeased = null;
+            $this->carriedWorkflowPollAnswer = null;
+        }
+        if ($this->shutdownRequested && !$carriedWorkflowLeased && $prefetched === null) {
             return false;
         }
 
@@ -816,7 +869,7 @@ final class Worker
         }
 
         $handled = false;
-        $workflowPoll = $this->pollWithRetry(
+        $workflowPoll = $prefetched['poll'] ?? $this->pollWithRetry(
             'workflow',
             fn (string $requestId, int $timeoutSeconds): array => $this->pollClient($requestId, $carriedWorkflowRequestId, $carriedWorkflowAnswer)->pollWorkflowTaskResponse(
                 $this->workerId,
@@ -837,10 +890,7 @@ final class Worker
         }
         $this->workflowGrant = is_array($workflowPoll['task'] ?? null) ? [
             'task_id' => (string) ($workflowPoll['task']['task_id'] ?? ''),
-            // A carried answer was received with its completion.
-            'received' => $carriedWorkflowAnswer !== null
-                && ($workflowPoll['task']['task_id'] ?? null) === ($carriedWorkflowAnswer['answer']['task']['task_id'] ?? null)
-                ? $carriedWorkflowAnswer['received'] : $this->now(),
+            'received' => $prefetched['received'] ?? $this->workflowGrantReceived($workflowPoll, $carriedWorkflowAnswer),
             'server_time' => $workflowPoll['server_time'] ?? null,
         ] : null;
         $this->rememberWorkflowMemoCapability($workflowPoll);
@@ -966,11 +1016,12 @@ final class Worker
 
     /**
      * The poll answer a completion returned with its carried poll's lease, or
-     * null when it returned none.
+     * null when it returned none. It counts as received when given, the time
+     * the completion was sent if it was answered later, else now.
      *
      * @return array{answer: array<string, mixed>, received: float, captured: float}|null
      */
-    private function carriedPollAnswer(mixed $poll): ?array
+    private function carriedPollAnswer(mixed $poll, ?float $received = null, ?float $captured = null): ?array
     {
         $answer = is_array($poll) && ($poll['leased'] ?? null) === true ? ($poll['answer'] ?? null) : null;
         if (!is_array($answer) || array_is_list($answer) || !is_array($answer['task'] ?? null)) {
@@ -978,7 +1029,225 @@ final class Worker
         }
 
         /** @var array<string, mixed> $answer */
-        return ['answer' => $answer, 'received' => $this->now(), 'captured' => hrtime(true) / 1e9];
+        return ['answer' => $answer, 'received' => $received ?? $this->now(), 'captured' => $captured ?? hrtime(true) / 1e9];
+    }
+
+    /**
+     * When the workflow task a poll delivered was granted: a carried answer
+     * was received with its completion, any other on arrival.
+     *
+     * @param array<string, mixed> $poll
+     * @param array{answer: array<string, mixed>, received: float, captured: float}|null $carried
+     */
+    private function workflowGrantReceived(array $poll, ?array $carried): float
+    {
+        return $carried !== null && is_array($poll['task'] ?? null)
+            && ($poll['task']['task_id'] ?? null) === ($carried['answer']['task']['task_id'] ?? null)
+            ? $carried['received'] : $this->now();
+    }
+
+    /**
+     * Whether a workflow task completion is sent while the next workflow
+     * task runs (`sendWhileNextTaskRuns`). That pays when this task ran a
+     * local activity, so the next one likely does too, or when the next task
+     * is leased already. A completion that schedules an activity is sent at
+     * once, so the activity poll it carries leases that activity for this
+     * Worker; so is one under a cancellation request, whose deadline its
+     * retries keep.
+     *
+     * @param list<array<string, mixed>> $commands
+     */
+    private function pipelinesCompletion(array $commands): bool
+    {
+        if (!$this->pipelining || $this->shutdownRequested || !$this->completionWorkflowPollSupported
+            || $this->claimCancellation !== null || $this->prefetchedWorkflowPoll !== null
+            || $this->deferredFailure !== null
+            || (!$this->localActivityRan && $this->carriedWorkflowPollLeased !== true)) {
+            return false;
+        }
+        foreach ($commands as $command) {
+            if (($command['type'] ?? null) === 'schedule_activity') {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    /**
+     * Leases the next workflow task before this one's completion is sent:
+     * delivers the lease the last completion's carried poll took, as the next
+     * tick would, or polls without waiting. Whether it leased one; the next
+     * tick runs it. A poll failure is kept for after this task completes,
+     * as the next tick's poll would have thrown it.
+     */
+    private function prefetchWorkflowTask(): bool
+    {
+        $requestId = $this->carriedWorkflowPollRequestId;
+        $leased = $this->carriedWorkflowPollLeased === true;
+        $answer = $this->carriedWorkflowPollAnswer;
+        $this->carriedWorkflowPollRequestId = null;
+        $this->carriedWorkflowPollLeased = null;
+        $this->carriedWorkflowPollAnswer = null;
+        try {
+            $poll = $this->pollWithRetry(
+                'workflow',
+                fn (string $id, int $timeoutSeconds): array => $this->pollClient($id, $requestId, $answer)->pollWorkflowTaskResponse(
+                    $this->workerId,
+                    $this->taskQueue,
+                    $timeoutSeconds,
+                    $id,
+                    self::WORKFLOW_HISTORY_PAGE_SIZE,
+                ),
+                0,
+                $requestId,
+                $leased,
+            );
+        } catch (Throwable $exception) {
+            $this->deferredFailure ??= $exception;
+
+            return false;
+        }
+        if ($poll === null || $this->stopForTerminalPoll($poll) || $this->taskFromPoll($poll) === null) {
+            return false;
+        }
+        $this->prefetchedWorkflowPoll = ['poll' => $poll, 'received' => $this->workflowGrantReceived($poll, $answer)];
+
+        return true;
+    }
+
+    /**
+     * Sends a workflow task completion that carries the next workflow poll,
+     * returning once it is sent while the Server answers; the next task,
+     * already leased, runs meanwhile, and settleWorkflowCompletion() takes the
+     * answer. Without a transport that can do that it is sent synchronously.
+     *
+     * @param array<string, mixed> $task
+     * @param \Closure(?array<string, mixed>, ?array<string, mixed>): array<string, mixed> $complete
+     */
+    private function sendWhileNextTaskRuns(array $task, \Closure $complete): void
+    {
+        $requestId = 'php-workflow-poll-'.bin2hex(random_bytes(16));
+        $workflowPoll = ['worker_id' => $this->workerId, 'task_queue' => $this->taskQueue,
+            'poll_request_id' => $requestId, 'history_page_size' => self::WORKFLOW_HISTORY_PAGE_SIZE];
+        // As for a synchronous send: a lease whose answer is lost is
+        // delivered by polling under the same request ID.
+        $this->carriedWorkflowPollRequestId = $requestId;
+        $fiber = new \Fiber(static fn (): array => $complete(null, $workflowPoll));
+        $sent = $this->now();
+        $sentAt = hrtime(true) / 1e9;
+        $this->client->suspendSendsOf($fiber);
+        try {
+            $fiber->start();
+        } finally {
+            $this->client->suspendSendsOf(null);
+        }
+        if ($fiber->isTerminated()) {
+            $this->publishCarriedWorkflowPoll($fiber->getReturn(), $sent, $sentAt);
+
+            return;
+        }
+        $this->carriedWorkflowPollRequestId = null;
+        $this->pendingWorkflowCompletion = ['fiber' => $fiber, 'task' => $task, 'poll_request_id' => $requestId,
+            'sent' => $sent, 'sent_at' => $sentAt];
+    }
+
+    /**
+     * Finishes the completion sent while the next task ran, and leaves the
+     * lease its carried poll took for the next tick, as a completion sent
+     * synchronously would. Its failure is handled as that task's; one that
+     * stops the Worker waits for throwDeferredFailure().
+     */
+    private function settleWorkflowCompletion(): void
+    {
+        $pending = $this->pendingWorkflowCompletion;
+        if ($pending === null) {
+            return;
+        }
+        $this->pendingWorkflowCompletion = null;
+        $this->carriedWorkflowPollRequestId = $pending['poll_request_id'];
+        // It was sent under no cancellation request, which its retries keep.
+        $claimCancellation = $this->claimCancellation;
+        $claimDeliveredCancellationId = $this->claimDeliveredCancellationId;
+        $this->claimCancellation = null;
+        $this->claimDeliveredCancellationId = null;
+        try {
+            $pending['fiber']->resume();
+            $completed = true;
+        } catch (Throwable $exception) {
+            $completed = false;
+            try {
+                $this->workflowTaskFailed($pending['task'], $exception);
+            } catch (Throwable $failure) {
+                if (!$failure instanceof ServerException || !$this->isTimedOutWorkflowCompletion($pending['task'], $failure)) {
+                    $this->deferredFailure ??= $failure;
+                }
+            }
+        } finally {
+            $this->claimCancellation = $claimCancellation;
+            $this->claimDeliveredCancellationId = $claimDeliveredCancellationId;
+        }
+        if (!$completed) {
+            return;
+        }
+        if (!$pending['fiber']->isTerminated()) {
+            throw new \LogicException('A workflow task completion suspended more than once.');
+        }
+        $this->publishCarriedWorkflowPoll($pending['fiber']->getReturn(), $pending['sent'], $pending['sent_at']);
+    }
+
+    /**
+     * Records what a completion's carried workflow poll leased, its lease
+     * counting from when the completion was sent.
+     */
+    private function publishCarriedWorkflowPoll(mixed $completion, float $sent, float $sentAt): void
+    {
+        $poll = is_array($completion) ? ($completion['workflow_poll'] ?? null) : null;
+        $this->carriedWorkflowPollLeased = self::carriedWorkflowPollLeased($completion);
+        $this->carriedWorkflowPollAnswer = $this->carriedPollAnswer($poll, $sent, $sentAt);
+    }
+
+    /**
+     * Whether the workflow poll a completion carried leased, or null when
+     * that is unknown. A completion answered without the poll claimed
+     * nothing for it (the Server claims for it only when the completion's
+     * carried activity poll leases nothing), so the poll's ID holds no lease.
+     */
+    private static function carriedWorkflowPollLeased(mixed $completion): ?bool
+    {
+        if (!is_array($completion)) {
+            return null;
+        }
+        if (!array_key_exists('workflow_poll', $completion)) {
+            return false;
+        }
+        $leased = is_array($completion['workflow_poll']) ? ($completion['workflow_poll']['leased'] ?? null) : null;
+
+        return is_bool($leased) ? $leased : null;
+    }
+
+    /** Throws the failure kept while the last task ran (`$deferredFailure`). */
+    private function throwDeferredFailure(): void
+    {
+        $failure = $this->deferredFailure;
+        $this->deferredFailure = null;
+        if ($failure !== null) {
+            throw $failure;
+        }
+    }
+
+    /**
+     * Whether this Worker holds a workflow task or awaits one, so a completion
+     * carries no workflow poll: a completion sent while the next task runs,
+     * the task a poll leased before it was sent, a carried poll's lease, or
+     * one a failed completion's carried poll may have taken, which the next
+     * tick delivers under its request ID.
+     */
+    private function workflowPollHeld(): bool
+    {
+        return $this->pendingWorkflowCompletion !== null || $this->prefetchedWorkflowPoll !== null
+            || $this->carriedWorkflowPollLeased === true
+            || ($this->carriedWorkflowPollRequestId !== null && $this->carriedWorkflowPollLeased === null);
     }
 
     /**
@@ -1509,10 +1778,23 @@ final class Worker
         );
     }
 
-    /** @param array<string, mixed> $task */
+    /**
+     * Runs a workflow task, then throws a failure kept while it ran. One the
+     * task throws comes first; run() reports the kept one as it stops.
+     *
+     * @param array<string, mixed> $task
+     */
     private function executeWorkflowTask(array $task): void
+    {
+        $this->runWorkflowTask($task);
+        $this->throwDeferredFailure();
+    }
+
+    /** @param array<string, mixed> $task */
+    private function runWorkflowTask(array $task): void
     {
         $traceScope = null;
+        $this->localActivityRan = false;
         $this->claimCancellation = null;
         $this->claimDeliveredCancellationId = null;
         $taskId = (string) ($task['task_id'] ?? '');
@@ -1583,81 +1865,61 @@ final class Worker
             $this->assertChildStartTurnAvailable($commands);
             $this->assertCancellationPoliciesAvailable($commands);
             $stickyClaim = $this->stickyCacheClaim($task);
+            $complete = fn (?array $activityPoll, ?array $workflowPoll): array => $this->retryStorageAdmission(
+                'workflow_complete',
+                fn (): array => $this->client->completeWorkflowTask(
+                    $taskId,
+                    $leaseOwner,
+                    $attempt,
+                    $commands,
+                    $messageStreamCursors,
+                    $messageStreamWaits,
+                    $stickyClaim,
+                    $activityPoll,
+                    $workflowPoll,
+                ),
+                ['task_id' => $taskId, 'lease_owner' => $leaseOwner, 'attempt' => $attempt],
+            );
+            // The completion sent while this task ran is finished first, since
+            // its carried poll may have leased the next task.
+            $this->settleWorkflowCompletion();
+            if ($this->pipelinesCompletion($commands) && $this->prefetchWorkflowTask()) {
+                $this->sendWhileNextTaskRuns($task, $complete);
+
+                return;
+            }
             // The completion carries the next activity poll, so the Server
-            // claims the activity it schedules in the same transaction.
+            // claims the activity it schedules in the same transaction. A
+            // Worker that stops once this task is done leases nothing more.
+            $stopping = $this->shutdownRequested || $this->deferredFailure !== null;
             $activityPoll = null;
-            if ($this->completionActivityPollSupported && !$this->shutdownRequested) {
+            if ($this->completionActivityPollSupported && !$stopping) {
                 $this->carriedActivityPollRequestId = 'php-activity-poll-'.bin2hex(random_bytes(16));
                 $activityPoll = ['worker_id' => $this->workerId, 'task_queue' => $this->taskQueue,
                     'poll_request_id' => $this->carriedActivityPollRequestId];
             }
             // It carries the next workflow poll too, which the Server claims
             // for only when that activity poll leases nothing, so this Worker
-            // holds no workflow task while it runs an activity.
+            // holds no workflow task while it runs an activity. A workflow
+            // task this Worker holds or awaits already comes first.
             $workflowPoll = null;
-            if ($this->completionWorkflowPollSupported && !$this->shutdownRequested) {
+            if ($this->completionWorkflowPollSupported && !$stopping && !$this->workflowPollHeld()) {
                 $this->carriedWorkflowPollRequestId = 'php-workflow-poll-'.bin2hex(random_bytes(16));
                 $workflowPoll = ['worker_id' => $this->workerId, 'task_queue' => $this->taskQueue,
                     'poll_request_id' => $this->carriedWorkflowPollRequestId,
                     'history_page_size' => self::WORKFLOW_HISTORY_PAGE_SIZE];
             }
-            $completion = $this->retryStorageAdmission('workflow_complete', fn (): array => $this->client->completeWorkflowTask(
-                $taskId,
-                $leaseOwner,
-                $attempt,
-                $commands,
-                $messageStreamCursors,
-                $messageStreamWaits,
-                $stickyClaim,
-                $activityPoll,
-                $workflowPoll,
-            ), [
-                'task_id' => $taskId,
-                'lease_owner' => $leaseOwner,
-                'attempt' => $attempt,
-            ]);
+            $completion = $complete($activityPoll, $workflowPoll);
             $leased = $activityPoll === null ? null : ($completion['activity_poll']['leased'] ?? null);
             $this->carriedActivityPollLeased = is_bool($leased) ? $leased : null;
             $this->carriedActivityPollAnswer = $activityPoll === null ? null : $this->carriedPollAnswer($completion['activity_poll'] ?? null);
-            $leased = $workflowPoll === null ? null : ($completion['workflow_poll']['leased'] ?? null);
-            $this->carriedWorkflowPollLeased = is_bool($leased) ? $leased : null;
-            $this->carriedWorkflowPollAnswer = $workflowPoll === null ? null : $this->carriedPollAnswer($completion['workflow_poll'] ?? null);
-        } catch (Throwable $exception) {
-            if ($exception instanceof WorkflowClaimDeferred) {
-                $this->diagnostic('worker.claim_deferred', [
-                    'task_id' => $taskId, 'task_kind' => 'workflow',
-                    'reason' => $this->shutdownRequested ? 'shutdown_requested' : 'claim_released',
-                    'message' => $exception->getMessage(),
-                ], 'info');
-
-                return;
-            }
-            if ($exception instanceof WorkflowClaimRevoked) {
-                $this->diagnostic('worker.claim_aborted', [
-                    'task_id' => $taskId, 'task_kind' => 'workflow',
-                    'reason' => $exception->reason, 'message' => $exception->getMessage(),
-                ], 'warning');
-
-                return;
+            // A lease held already stays for the next tick.
+            if ($workflowPoll !== null) {
+                $this->carriedWorkflowPollLeased = self::carriedWorkflowPollLeased($completion);
+                $this->carriedWorkflowPollAnswer = $this->carriedPollAnswer($completion['workflow_poll'] ?? null);
             }
-            $this->acknowledgeTaskFailure(
-                'workflow',
-                $taskId,
-                $exception,
-                function (Throwable $failure) use ($taskId, $leaseOwner, $attempt): void {
-                    $this->client->failWorkflowTask(
-                        $taskId,
-                        $leaseOwner,
-                        $attempt,
-                        'PHP workflow task execution failed: '.$failure->getMessage(),
-                        $failure::class,
-                        $failure instanceof NonDeterministicWorkflow || $failure instanceof WorkflowEntryArgumentsMismatch ? $failure->reason : null,
-                        $failure instanceof NonDeterministicWorkflow ? $failure->sequence : null,
-                        $failure->getTraceAsString(),
-                    );
-                },
-                $attempt,
-            );
+        } catch (Throwable $exception) {
+            $this->workflowTaskFailed($task, $exception);
         } finally {
             $traceScope?->__invoke();
             $this->claimCancellation = null;
@@ -1665,6 +1927,54 @@ final class Worker
         }
     }
 
+    /**
+     * Handles the failure of a workflow task or of its completion: a claim
+     * deferred or revoked is let go, any other failure is reported.
+     *
+     * @param array<string, mixed> $task
+     */
+    private function workflowTaskFailed(array $task, Throwable $exception): void
+    {
+        $taskId = (string) ($task['task_id'] ?? '');
+        $attempt = (int) ($task['workflow_task_attempt'] ?? 1);
+        $leaseOwner = (string) ($task['lease_owner'] ?? $this->workerId);
+        if ($exception instanceof WorkflowClaimDeferred) {
+            $this->diagnostic('worker.claim_deferred', [
+                'task_id' => $taskId, 'task_kind' => 'workflow',
+                'reason' => $this->shutdownRequested ? 'shutdown_requested' : 'claim_released',
+                'message' => $exception->getMessage(),
+            ], 'info');
+
+            return;
+        }
+        if ($exception instanceof WorkflowClaimRevoked) {
+            $this->diagnostic('worker.claim_aborted', [
+                'task_id' => $taskId, 'task_kind' => 'workflow',
+                'reason' => $exception->reason, 'message' => $exception->getMessage(),
+            ], 'warning');
+
+            return;
+        }
+        $this->acknowledgeTaskFailure(
+            'workflow',
+            $taskId,
+            $exception,
+            function (Throwable $failure) use ($taskId, $leaseOwner, $attempt): void {
+                $this->client->failWorkflowTask(
+                    $taskId,
+                    $leaseOwner,
+                    $attempt,
+                    'PHP workflow task execution failed: '.$failure->getMessage(),
+                    $failure::class,
+                    $failure instanceof NonDeterministicWorkflow || $failure instanceof WorkflowEntryArgumentsMismatch ? $failure->reason : null,
+                    $failure instanceof NonDeterministicWorkflow ? $failure->sequence : null,
+                    $failure->getTraceAsString(),
+                );
+            },
+            $attempt,
+        );
+    }
+
     /** @param array<string, mixed> $task */
     private function isTimedOutWorkflowCompletion(array $task, ServerException $exception): bool
     {
@@ -2759,9 +3069,9 @@ final class Worker
             }
             // The completion carries the next workflow poll, so the Server
             // claims the workflow task the result makes ready in the same
-            // transaction.
+            // transaction, unless this Worker holds or awaits one already.
             $workflowPoll = null;
-            if ($this->activityCompletionWorkflowPollSupported && !$this->shutdownRequested) {
+            if ($this->activityCompletionWorkflowPollSupported && !$this->shutdownRequested && !$this->workflowPollHeld()) {
                 $this->carriedWorkflowPollRequestId = 'php-workflow-poll-'.bin2hex(random_bytes(16));
                 $workflowPoll = ['worker_id' => $this->workerId, 'task_queue' => $this->taskQueue,
                     'poll_request_id' => $this->carriedWorkflowPollRequestId,
@@ -2776,9 +3086,10 @@ final class Worker
                     'lease_owner' => $leaseOwner,
                 ],
             );
-            $leased = $workflowPoll === null ? null : ($completion['workflow_poll']['leased'] ?? null);
-            $this->carriedWorkflowPollLeased = is_bool($leased) ? $leased : null;
-            $this->carriedWorkflowPollAnswer = $workflowPoll === null ? null : $this->carriedPollAnswer($completion['workflow_poll'] ?? null);
+            if ($workflowPoll !== null) {
+                $this->carriedWorkflowPollLeased = self::carriedWorkflowPollLeased($completion);
+                $this->carriedWorkflowPollAnswer = $this->carriedPollAnswer($completion['workflow_poll'] ?? null);
+            }
         } catch (Throwable $exception) {
             if ($exception instanceof ServerException && $this->discardClosedActivityAttempt($task, $exception)) {
                 return;
@@ -3293,6 +3604,7 @@ final class Worker
         array $arguments,
         array $options,
     ): array {
+        $this->localActivityRan = true;
         $options = WorkflowCommand::canonicalLocalActivityOptions($options);
         $handler = $this->activities[$activityType] ?? null;
         if ($handler === null) {
diff --git a/tests/Psr18TransportSuspensionTest.php b/tests/Psr18TransportSuspensionTest.php
new file mode 100644
index 0000000..54f4170
--- /dev/null
+++ b/tests/Psr18TransportSuspensionTest.php
@@ -0,0 +1,120 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Tests;
+
+use DurableWorkflow\Transport\Psr18Transport;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * The default transport can suspend the Fiber that sends a request once the
+ * request is sent in full, and finish the request when the Fiber resumes.
+ */
+final class Psr18TransportSuspensionTest extends TestCase
+{
+    private string $directory;
+    /** @var resource|null */
+    private $process = null;
+    private string $address;
+
+    protected function setUp(): void
+    {
+        if (!function_exists('curl_multi_exec') || !function_exists('proc_open')) {
+            self::markTestSkipped('Real cURL and proc_open are required.');
+        }
+        $this->directory = sys_get_temp_dir().'/dw-suspending-'.bin2hex(random_bytes(8));
+        self::assertTrue(mkdir($this->directory, 0700));
+        $process = proc_open([PHP_BINARY, __DIR__.'/Support/suspending-peer.php', $this->directory],
+            [0 => ['file', '/dev/null', 'r'], 1 => ['file', '/dev/null', 'w'], 2 => ['file', $this->directory.'/stderr', 'w']], $pipes);
+        self::assertIsResource($process);
+        $this->process = $process;
+        $this->address = $this->awaitFile('address');
+    }
+
+    protected function tearDown(): void
+    {
+        if (is_resource($this->process)) {
+            proc_terminate($this->process);
+            proc_close($this->process);
+        }
+        if (isset($this->directory)) {
+            foreach (glob($this->directory.'/*') ?: [] as $file) {
+                unlink($file);
+            }
+            rmdir($this->directory);
+        }
+    }
+
+    public function testARequestSentInFullSuspendsItsFiberUntilItResumes(): void
+    {
+        $transport = new Psr18Transport();
+        $body = ['payload' => str_repeat('x', 512 * 1024)];
+        $fiber = new \Fiber(fn (): ?array => $transport->send('POST', "http://{$this->address}/complete",
+            ['Content-Type' => 'application/json'], $body));
+
+        self::assertTrue($transport->suspendSendsOf($fiber));
+        $fiber->start();
+        $transport->suspendSendsOf(null);
+
+        self::assertTrue($fiber->isSuspended());
+        self::assertSame(json_encode($body), $this->awaitFile('received-1'));
+
+        // The answer waits while other requests, from an unarmed Fiber too,
+        // are sent and answered.
+        touch($this->directory.'/release-1');
+        touch($this->directory.'/release-2');
+        $other = new \Fiber(fn (): ?array => $transport->send('GET', "http://{$this->address}/other", []));
+        $other->start();
+        self::assertTrue($other->isTerminated());
+        self::assertSame(['request' => 2, 'path' => '/other', 'bytes' => 0], $other->getReturn());
+        self::assertTrue($fiber->isSuspended());
+
+        $fiber->resume();
+
+        self::assertTrue($fiber->isTerminated());
+        self::assertSame(['request' => 1, 'path' => '/complete', 'bytes' => strlen((string) json_encode($body))], $fiber->getReturn());
+    }
+
+    public function testOnlyTheArmedFibersFirstRequestSuspends(): void
+    {
+        $transport = new Psr18Transport();
+        touch($this->directory.'/release-1');
+        touch($this->directory.'/release-2');
+        $fiber = new \Fiber(function () use ($transport): array {
+            return [
+                $transport->send('POST', "http://{$this->address}/first", ['Content-Type' => 'application/json'], ['n' => 1]),
+                $transport->send('POST', "http://{$this->address}/second", ['Content-Type' => 'application/json'], ['n' => 2]),
+            ];
+        });
+
+        self::assertTrue($transport->suspendSendsOf($fiber));
+        $fiber->start();
+        $transport->suspendSendsOf(null);
+        self::assertTrue($fiber->isSuspended());
+        $fiber->resume();
+
+        self::assertTrue($fiber->isTerminated());
+        self::assertSame(['/first', '/second'], array_column($fiber->getReturn(), 'path'));
+    }
+
+    public function testATransportWithAnInjectedClientNeverSuspends(): void
+    {
+        self::assertFalse((new Psr18Transport(new \GuzzleHttp\Client()))->suspendSendsOf(null));
+    }
+
+    private function awaitFile(string $name): string
+    {
+        $path = $this->directory.'/'.$name;
+        $until = microtime(true) + 10;
+        while (!file_exists($path) && microtime(true) < $until) {
+            usleep(1000);
+        }
+        // The peer writes the file in one call; a reader may still see it short.
+        usleep(2000);
+        $contents = file_get_contents($path);
+        self::assertIsString($contents, "The peer never wrote {$name}.");
+
+        return $contents;
+    }
+}

Sanitized reference diff 0009-t19-Pipeline-workflow-task-completions-after-local-activ.patch, part 2/2.

This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

Complete sanitized patch: 68,098 bytes, SHA-256 e218bcb68d8d32a36234986205d994c9ecc2279482d1110649515b2b40e93f8e. This part: 21,361 bytes, SHA-256 842416fb2f3f9ab89357793cca0a33efbfebe78788b4dfc812a751af9b34c434. Parts concatenate in order; a part boundary is between complete file diffs.

diff --git a/tests/Support/suspending-peer.php b/tests/Support/suspending-peer.php
new file mode 100644
index 0000000..0f03d2c
--- /dev/null
+++ b/tests/Support/suspending-peer.php
@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+// A real HTTP peer that records each request body it reads in full, then
+// answers it only once the test creates the release file for it.
+[$script, $directory] = $argv;
+$server = stream_socket_server('tcp://127.0.0.1:0', $errno, $error);
+if ($server === false) {
+    throw new RuntimeException($error);
+}
+file_put_contents($directory.'/address', stream_socket_get_name($server, false));
+$served = 0;
+$until = microtime(true) + 30;
+while (microtime(true) < $until) {
+    $peer = @stream_socket_accept($server, 1);
+    if ($peer === false) {
+        continue;
+    }
+    stream_set_timeout($peer, 10);
+    // One request a connection, so a reply held back blocks no other.
+    if (($first = fgets($peer)) !== false) {
+        $length = 0;
+        while (($line = fgets($peer)) !== false && $line !== "\r\n") {
+            if (preg_match('/^Content-Length:\s*(\d+)/i', $line, $matches)) {
+                $length = (int) $matches[1];
+            }
+        }
+        $body = '';
+        while (strlen($body) < $length && !feof($peer)) {
+            $chunk = fread($peer, $length - strlen($body));
+            if ($chunk === false) {
+                break;
+            }
+            $body .= $chunk;
+        }
+        ++$served;
+        file_put_contents($directory.'/received-'.$served, $body);
+        while (!file_exists($directory.'/release-'.$served) && microtime(true) < $until) {
+            usleep(1000);
+        }
+        $reply = json_encode(['request' => $served, 'path' => explode(' ', $first)[1] ?? '', 'bytes' => strlen($body)]);
+        fwrite($peer, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: ".strlen($reply)."\r\nConnection: close\r\n\r\n".$reply);
+    }
+    fclose($peer);
+}
diff --git a/tests/WorkerCompletionPipeliningTest.php b/tests/WorkerCompletionPipeliningTest.php
new file mode 100644
index 0000000..a34fd7d
--- /dev/null
+++ b/tests/WorkerCompletionPipeliningTest.php
@@ -0,0 +1,432 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DurableWorkflow\Tests;
+
+use DurableWorkflow\Client;
+use DurableWorkflow\Exception\ServerException;
+use DurableWorkflow\Transport\SuspendingTransport;
+use DurableWorkflow\Transport\Transport;
+use DurableWorkflow\Worker;
+use DurableWorkflow\Worker\ActivityContext;
+use DurableWorkflow\Worker\WorkflowContext;
+use PHPUnit\Framework\TestCase;
+use Throwable;
+
+/**
+ * A Worker that runs its own loop sends a workflow task completion after a
+ * local activity without waiting for the answer: it leases the next task
+ * first, runs it while the Server answers, and takes the answer, with the
+ * lease its carried poll took, before that task completes.
+ */
+final class WorkerCompletionPipeliningTest extends TestCase
+{
+    public function testACompletionAfterALocalActivityIsAnsweredWhileTheNextTaskRuns(): void
+    {
+        $server = new PipeliningServer(['pipe-1', 'pipe-2', 'pipe-3']);
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        self::assertSame([
+            'step pipe-1', 'sent pipe-1',
+            'step pipe-2', 'answered pipe-1', 'sent pipe-2',
+            'step pipe-3', 'answered pipe-2', 'sent pipe-3', 'answered pipe-3',
+        ], $server->events);
+        // The second task was polled before the first completion was sent,
+        // the third came back with it, and the last poll found none.
+        self::assertSame(3, $server->workflowPolls);
+        $first = $server->completions['pipe-1'];
+        self::assertArrayNotHasKey('activity_poll', $first);
+        self::assertIsString($first['workflow_poll']['poll_request_id'] ?? null);
+        self::assertArrayNotHasKey('workflow_poll', $server->completions['pipe-3']);
+        self::assertSame([], $server->failures);
+    }
+
+    public function testACompletionIsSentAtOnceWithoutALocalActivity(): void
+    {
+        $server = new PipeliningServer(['plain-1', 'plain-2'], 'pipe.plain');
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        self::assertSame(['sent plain-1', 'answered plain-1', 'sent plain-2', 'answered plain-2'], $server->events);
+    }
+
+    public function testACompletionThatSchedulesAnActivityIsSentAtOnce(): void
+    {
+        $server = new PipeliningServer(['mixed-1'], 'pipe.mixed');
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        self::assertSame(['step mixed-1', 'sent mixed-1', 'answered mixed-1'], $server->events);
+        self::assertSame(['record_local_activity', 'schedule_activity'], array_column($server->completions['mixed-1']['commands'], 'type'));
+        self::assertIsString($server->completions['mixed-1']['activity_poll']['poll_request_id'] ?? null);
+    }
+
+    public function testAnActivityLeasedByACompletionCarriesTheNextWorkflowPollOnItsOwnCompletion(): void
+    {
+        $server = new PipeliningServer(['eager-1'], 'pipe.mixed');
+        $server->leasesScheduledActivities = true;
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        // The completion's workflow poll was left out of its answer, so it
+        // holds no lease, and the activity's completion carries the next.
+        self::assertSame(['step eager-1', 'sent eager-1', 'answered eager-1', 'activity eager-1-activity'], $server->events);
+        $carried = $server->completions['eager-1']['workflow_poll']['poll_request_id'] ?? null;
+        self::assertIsString($carried);
+        $next = $server->activityCompletions['eager-1-activity']['workflow_poll']['poll_request_id'] ?? null;
+        self::assertIsString($next);
+        self::assertNotSame($carried, $next);
+        // That poll leased nothing, so the next tick polls under its ID.
+        self::assertSame(2, $server->workflowPolls);
+        self::assertSame($next, $server->workflowPollRequestIds[2]);
+    }
+
+    public function testAWorkerDrivenByTickSendsEveryCompletionAtOnce(): void
+    {
+        $server = new PipeliningServer(['tick-1', 'tick-2']);
+        $worker = $this->worker($server);
+
+        self::assertTrue($worker->tick(0));
+        self::assertTrue($worker->tick(0));
+
+        self::assertSame(['step tick-1', 'sent tick-1', 'answered tick-1', 'step tick-2', 'sent tick-2', 'answered tick-2'],
+            $server->events);
+    }
+
+    public function testATransportThatCannotSuspendSendsEveryCompletionAtOnce(): void
+    {
+        $server = new PipeliningServer(['sync-1', 'sync-2']);
+        $server->suspends = false;
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        self::assertSame(['step sync-1', 'sent sync-1', 'answered sync-1', 'step sync-2', 'sent sync-2', 'answered sync-2'],
+            $server->events);
+    }
+
+    public function testACompletionWhoseClaimWasLostIsLetGoBeforeTheNextTaskCompletes(): void
+    {
+        $server = new PipeliningServer(['lost-1', 'lost-2']);
+        $server->completionFailures['lost-1'] = new ServerException('The run closed.', 409, 'run_closed',
+            ['task_id' => 'lost-1', 'can_continue' => false, 'task_status' => 'cancelled']);
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        self::assertSame(['step lost-1', 'sent lost-1', 'step lost-2', 'answered lost-1', 'sent lost-2', 'answered lost-2'],
+            $server->events);
+        self::assertSame([], $server->failures);
+        // The second completion polls under the first one's carried request
+        // ID, which a lease taken before the failure is delivered by.
+        self::assertSame($server->completions['lost-1']['workflow_poll']['poll_request_id'] ?? null,
+            $server->workflowPollRequestIds[3] ?? null);
+    }
+
+    public function testACompletionFailureThatStopsTheWorkerIsThrownOnceTheNextTaskIsDone(): void
+    {
+        $server = new PipeliningServer(['stop-1', 'stop-2']);
+        $refusal = new ServerException('The commands are invalid.', 422, 'invalid_commands');
+        $server->completionFailures['stop-1'] = $refusal;
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        try {
+            $worker->run(0);
+            self::fail('The refused completion should stop the Worker.');
+        } catch (ServerException $exception) {
+            self::assertSame($refusal, $exception);
+        }
+
+        self::assertSame(['step stop-1', 'sent stop-1', 'step stop-2', 'answered stop-1', 'sent stop-2', 'answered stop-2'],
+            $server->events);
+    }
+
+    public function testAPollFailureWhileLeasingTheNextTaskIsThrownOnceTheTaskCompletes(): void
+    {
+        $server = new PipeliningServer(['drop-1', 'drop-2']);
+        $worker = $this->worker($server);
+        $server->workflowPollFailure = new \LogicException('The poll broke.');
+        $server->failPollAfter = 2;
+
+        try {
+            $worker->run(0);
+            self::fail('The broken poll should stop the Worker.');
+        } catch (\LogicException $exception) {
+            self::assertSame('The poll broke.', $exception->getMessage());
+        }
+
+        self::assertSame(['step drop-1', 'sent drop-1', 'step drop-2', 'answered drop-1', 'sent drop-2', 'answered drop-2'],
+            $server->events);
+        self::assertArrayNotHasKey('workflow_poll', $server->completions['drop-2']);
+        self::assertSame([], $server->failures);
+    }
+
+    public function testAPendingCompletionIsAnsweredWhenTheLoopFails(): void
+    {
+        $server = new PipeliningServer(['loop-1', 'loop-2']);
+        $server->activityPollFailure = new \LogicException('The activity poll broke.');
+        $worker = $this->worker($server);
+
+        try {
+            $worker->run(0);
+            self::fail('The broken poll should stop the Worker.');
+        } catch (\LogicException $exception) {
+            self::assertSame('The activity poll broke.', $exception->getMessage());
+        }
+
+        self::assertSame(['step loop-1', 'sent loop-1', 'answered loop-1'], $server->events);
+    }
+
+    public function testAnActivityCompletionCarriesNoWorkflowPollWhileATaskIsHeld(): void
+    {
+        $server = new PipeliningServer(['held-1', 'held-2']);
+        $server->activities = ['held-activity'];
+        $worker = $this->worker($server);
+        $server->onEmptyWorkflowPoll = static fn () => $worker->requestShutdown();
+
+        $worker->run(0);
+
+        // The activity ran while the first completion was pending and the
+        // second task was held, so its completion carried no workflow poll.
+        self::assertSame(['step held-1', 'sent held-1', 'activity held-activity',
+            'step held-2', 'answered held-1', 'sent held-2', 'answered held-2'], $server->events);
+        self::assertArrayNotHasKey('workflow_poll', $server->activityCompletions['held-activity']);
+    }
+
+    private function worker(PipeliningServer $server): Worker
+    {
+        $worker = new Worker(new Client('https://server.example', transport: $server), 'pipe-workers', workerId: 'worker-1');
+        $events = &$server->events;
+        $worker->registerActivity('pipe.step', static function (ActivityContext $context) use (&$events): string {
+            $events[] = 'step '.substr($context->workflowId, 0, -strlen('-workflow'));
+
+            return 'stepped';
+        });
+        $worker->registerActivity('pipe.activity', static function (ActivityContext $context) use (&$events): string {
+            $events[] = 'activity '.$context->taskId;
+
+            return 'done';
+        });
+        $worker->registerWorkflow('pipe.local', static fn (WorkflowContext $context): mixed => $context->localActivity('pipe.step'));
+        $worker->registerWorkflow('pipe.plain', static fn (WorkflowContext $context): string => 'done');
+        $worker->registerWorkflow('pipe.mixed', static function (WorkflowContext $context): mixed {
+            $context->localActivity('pipe.step');
+
+            return $context->activity('pipe.activity');
+        });
+
+        return $worker;
+    }
+}
+
+/**
+ * A Server on a transport that suspends the armed Fiber once its request is
+ * sent: the Server handles the request then, and the answer is returned when
+ * the Fiber resumes. It serves its workflow tasks to polls in order.
+ */
+final class PipeliningServer implements Transport, SuspendingTransport
+{
+    /** @var list<string> */
+    public array $events = [];
+    /** @var array<string, array<string, mixed>> */
+    public array $completions = [];
+    /** @var array<string, array<string, mixed>> */
+    public array $activityCompletions = [];
+    /** @var list<string> */
+    public array $failures = [];
+    /** @var array<string, Throwable> */
+    public array $completionFailures = [];
+    /** @var list<string> Activity tasks served to the first activity poll. */
+    public array $activities = [];
+    /** @var array<int, ?string> The request ID of each workflow poll, from 1. */
+    public array $workflowPollRequestIds = [];
+    public int $workflowPolls = 0;
+    public bool $suspends = true;
+    /** Whether a completion's carried activity poll leases the activity it schedules. */
+    public bool $leasesScheduledActivities = false;
+    public ?\Closure $onEmptyWorkflowPoll = null;
+    public ?Throwable $workflowPollFailure = null;
+    public ?Throwable $activityPollFailure = null;
+    public int $failPollAfter = PHP_INT_MAX;
+    /** @var \Fiber<mixed, mixed, mixed, mixed>|null */
+    private ?\Fiber $suspending = null;
+
+    /** @param list<string> $tasks */
+    public function __construct(private array $tasks, private readonly string $workflowType = 'pipe.local')
+    {
+    }
+
+    public function suspendSendsOf(?\Fiber $fiber): bool
+    {
+        if (!$this->suspends) {
+            return false;
+        }
+        $this->suspending = $fiber;
+
+        return true;
+    }
+
+    public function send(string $method, string $uri, array $headers, ?array $body = null): ?array
+    {
+        $suspend = $this->suspending !== null && $this->suspending === \Fiber::getCurrent();
+        $this->suspending = null;
+        $completed = preg_match('#/workflow-tasks/([^/]+)/complete$#', $uri, $match) === 1 ? $match[1] : null;
+        try {
+            $answer = $this->answer($method, $uri, $body ?? []);
+        } catch (Throwable $failure) {
+            $answer = $failure;
+        }
+        if ($completed !== null) {
+            $this->events[] = 'sent '.$completed;
+        }
+        if ($suspend) {
+            \Fiber::suspend();
+        }
+        if ($completed !== null) {
+            $this->events[] = 'answered '.$completed;
+        }
+        if ($answer instanceof Throwable) {
+            throw $answer;
+        }
+
+        return $answer;
+    }
+
+    /**
+     * @param array<string, mixed> $body
+     * @return array<string, mixed>|Throwable
+     */
+    private function answer(string $method, string $uri, array $body): array|Throwable
+    {
+        if (str_ends_with($uri, '/worker/register')) {
+            return ['registered' => true];
+        }
+        if ($method === 'DELETE' && str_ends_with($uri, '/worker/registrations/worker-1')) {
+            return ['worker_id' => 'worker-1', 'outcome' => 'deregistered', 'recovered_workflow_task_count' => 0];
+        }
+        if (str_ends_with($uri, '/workflow-tasks/poll')) {
+            $this->workflowPollRequestIds[++$this->workflowPolls] = $body['poll_request_id'] ?? null;
+            if ($this->workflowPolls > $this->failPollAfter && $this->workflowPollFailure !== null) {
+                return $this->workflowPollFailure;
+            }
+            $task = array_shift($this->tasks);
+            if ($task === null) {
+                if ($this->onEmptyWorkflowPoll !== null) {
+                    ($this->onEmptyWorkflowPoll)();
+                }
+
+                return ['task' => null, 'poll_status' => 'empty'];
+            }
+
+            return $this->workflowLease($task);
+        }
+        if (str_ends_with($uri, '/activity-tasks/poll')) {
+            if ($this->activityPollFailure !== null) {
+                return $this->activityPollFailure;
+            }
+            $activity = array_shift($this->activities);
+
+            return $activity === null ? ['task' => null, 'poll_status' => 'empty'] : self::activityLease($activity);
+        }
+        if (str_ends_with($uri, '/query-tasks/poll')) {
+            return ['task' => null, 'poll_status' => 'empty'];
+        }
+        if (preg_match('#/workflow-tasks/([^/]+)/complete$#', $uri, $match) === 1) {
+            $this->completions[$match[1]] = $body;
+            if (isset($this->completionFailures[$match[1]])) {
+                return $this->completionFailures[$match[1]];
+            }
+            $answer = ['outcome' => 'completed', 'recorded' => true];
+            $schedules = in_array('schedule_activity', array_column($body['commands'] ?? [], 'type'), true);
+            if (isset($body['activity_poll'])) {
+                $answer['activity_poll'] = $schedules && $this->leasesScheduledActivities
+                    ? ['leased' => true, 'answer' => self::activityLease($match[1].'-activity')] : ['leased' => false];
+            }
+            // As the Server does, the workflow poll is claimed for only when
+            // the activity poll leased nothing; otherwise it is left out.
+            if (isset($body['workflow_poll']) && ($answer['activity_poll']['leased'] ?? false) !== true) {
+                $task = array_shift($this->tasks);
+                $answer['workflow_poll'] = $task === null ? ['leased' => false]
+                    : ['leased' => true, 'answer' => $this->workflowLease($task)];
+            }
+
+            return $answer;
+        }
+        if (preg_match('#/workflow-tasks/([^/]+)/fail$#', $uri, $match) === 1) {
+            $this->failures[] = $match[1];
+            $this->events[] = 'failed '.$match[1];
+
+            return ['outcome' => 'failed', 'recorded' => true];
+        }
+        if (preg_match('#/activity-tasks/([^/]+)/complete$#', $uri, $match) === 1) {
+            $this->activityCompletions[$match[1]] = $body;
+            $answer = ['outcome' => 'completed', 'recorded' => true];
+            if (isset($body['workflow_poll'])) {
+                $task = array_shift($this->tasks);
+                $answer['workflow_poll'] = $task === null ? ['leased' => false]
+                    : ['leased' => true, 'answer' => $this->workflowLease($task)];
+            }
+
+            return $answer;
+        }
+        if (preg_match('#/workflow-tasks/([^/]+)/heartbeat$#', $uri, $match) === 1) {
+            return ['task_id' => $match[1], 'workflow_task_attempt' => $body['workflow_task_attempt'] ?? null,
+                'lease_owner' => $body['lease_owner'] ?? null, 'renewed' => true];
+        }
+
+        throw new \LogicException("Unexpected worker request: {$method} {$uri}");
+    }
+
+    /** @return array<string, mixed> */
+    private function workflowLease(string $taskId): array
+    {
+        return [
+            'poll_status' => 'leased',
+            'task' => [
+                'task_id' => $taskId,
+                'workflow_task_attempt' => 1,
+                'lease_owner' => 'worker-1',
+                'workflow_id' => $taskId.'-workflow',
+                'run_id' => $taskId.'-run',
+                'workflow_type' => $this->workflowType,
+                'payload_codec' => 'avro',
+                'history_events' => [],
+                'task_kind' => 'workflow',
+            ],
+            'server_capabilities' => self::capabilities(),
+        ];
+    }
+
+    /** @return array<string, mixed> */
+    private static function activityLease(string $taskId): array
+    {
+        return ['poll_status' => 'leased',
+            'task' => ['task_id' => $taskId, 'activity_attempt_id' => $taskId.'-attempt', 'lease_owner' => 'worker-1',
+                'activity_type' => 'pipe.activity', 'payload_codec' => 'avro'],
+            'server_capabilities' => self::capabilities()];
+    }
+
+    /** @return array<string, mixed> */
+    private static function capabilities(): array
+    {
+        return [
+            'supported_workflow_task_commands' => ['schedule_activity', 'complete_workflow', 'fail_workflow'],
+            'workflow_task_completion_activity_poll' => true,
+            'activity_completion_workflow_poll' => true,
+            'workflow_task_completion_workflow_poll' => true,
+        ];
+    }
+}

sdk-php owns this opt-in P0 after the carried-poll prerequisites. I reconciled the actual dependency chain: #330/#331 still need Server #1062/#1067, with #1060/#1061's unresolved worker-binding and error controls retained. The supplied draft's missing opt-in option remains a concrete implementation gap. Nothing in the accepted helper/grant backport #335 enables completion pipelining.

Next action: after the carried-answer contracts are accepted, sdk-php implements an explicit default-off Worker option and owns the default transport's send/suspend/resume/cleanup lifecycle. Qualification must preserve synchronous tick/cancellation/kept-failure/telemetry/metrics paths, one-task-ahead and available-slot limits, original lease fencing, missing-key semantics and deferred error ordering, including a real-socket transport control. I requested the actual producer checkpoints from Server; no default change, invented capability or benchmark claim is selected.

I reconciled the actual3.9.5 package and current Server1073 successor: the maintenance release does not enable completion pipelining. Producer b985 has terminal CI but retains Security Source HOLD6699666744 for the real HTTP envelope and overly broad soft-decline shape; #330/#331 therefore have no valid successor handoff yet. Next owner/action: Server returns the accepted advertised carried-answer contracts, then sdk-php implements the explicit default-OFF Worker option and owns send/suspend/resume/cleanup, one-task-ahead/free-slot limits, original leases and deferred-error/telemetry/metrics semantics with real-socket controls. No option/default or transport was silently changed.

The carried-poll source prerequisite has landed: #344 at e929f1d, #345 at 41a3e37, and #346 at fc6309bdb77b17c91f6875984df6997750966a5b / tree 3d7d2f7f883cfcda859d5a849d364827baa0ccf3. #346 has independent Security source acceptance and 41 successful exact-head checks / one intended skip. Actual-main qualification and completed-topic cleanup are finishing.

SDK-PHP owns the next source slice. It will add an explicit Worker option defaulting OFF, restricted to managed run() and the owned default transport. Public tick(), injected HTTP clients, streams/progress, activity-scheduling, cancellation, retained-failure, telemetry and metrics paths remain synchronous. One already-leased workflow ahead is the hard bound; its slot prevents another carried workflow poll.

The implementation must own the suspended request's lifecycle, consume only the armed Fiber's first eligible request, join the prior answer before the next task completes, and report deferred failure once. Current Worker claim/cancellation/grant fields are mutable across a task: resuming a parent frame must not clear or replace the next task's state. This is a design constraint from maintained-source review, not a claimed defect in the existing synchronous worker. Real-socket request-body/answer-order and cleanup controls, default-off packet parity, causal corpus and independent source review are required before adoption.

No pipeline source, imported patch, new runtime, release or benchmark has been executed or accepted by this update. The published producer/SDK tuple and existing release gates remain separate.

The #330/#331 source prerequisite is now fully main-qualified and cleaned up: #346 main fc6309bdb77b17c91f6875984df6997750966a5b, tree 3d7d2f7f883cfcda859d5a849d364827baa0ccf3, all 42 main checks terminal (40 success / two intended skips).

SDK-PHP is starting the pipeline source on isolated feat/332-opt-in-completion-pipeline, rooted at that exact main. No pipeline code or test has run yet; the branch currently contains only the accepted baseline. The first concrete slice is the owned, one-shot default-transport suspension primitive with real-socket send/answer/cleanup controls, followed by the explicit Worker option defaulting OFF and the managed one-task-ahead path. The existing synchronous exclusions and deferred-failure/cancellation/grant constraints from the preceding design record remain required.

Normal source authoring and bounded source tests do not select a release, new bootstrap/profile execution or a published Server tuple. Those gates remain separate.

First useful increment is pushed in draft #347, exact ddce20b064b8f0a1572f703ce85e6920b92e93db / tree cf1e4d2e5167459eab89f3dac504bf925774f91c, based on main fc6309bdb77b17c91f6875984df6997750966a5b.

This increment only implements the explicitly armed owned-default-transport primitive. One Fiber's first request must match the exact completion URI; it yields outside cURL's native callback after the full JSON body is sent, then finishes the same response on resume. Another Fiber cannot consume or release its arm. Earlier ineligible I/O, payload progress and injected HTTP clients remain synchronous. Unarmed requests retain their existing path.

Eight real-loopback/socket and ownership tests pass with 45 assertions. The related transport/timeout/payload/poll set passes 226 tests / 1,998 assertions, and changed-production PHPStan passes. Credential/private-path and public-boundary checks pass; local containers are removed. Normal CI and requested finite primitive review are active.

No Worker option or pipeline loop is implemented yet. SDK-PHP is continuing those on this same draft, with the explicit option defaulting OFF, managed-run-only eligibility, one-ahead bound, synchronous exclusions and deferred failure/claim-state controls. The draft is not merge-ready and is not full #332 source, release or performance acceptance.

The complete Worker source increment is now pushed in draft #347: 1e8cb6cdb466cb1f7b99d572cdefd57469c794e6 / tree 5d67ce3714b8be1e60e2d3b2b82bb4d73719a938, based on main fc6309bdb77b17c91f6875984df6997750966a5b.

The explicit constructor/create option defaults OFF. Managed run can prefetch one ordinary zero-timeout workflow, send the local-work completion, execute that original leased workflow while the reply waits, then join before the next completion or failure. Original grant/expiry/transit/fence are preserved. Held or uncertain slots block another carried workflow poll. Public tick, scheduling, cancellation/cooperative mode, kept local failures, telemetry, metrics and unsupported transport paths stay synchronous.

Local qualification passes 106 tests / 995 assertions plus changed-production PHPStan. The official same-consumer regression passes this candidate and fails exact pre-pipeline main at the expected ordering assertion. Default-OFF real HTTP packet streams are equal against that predecessor after only generated poll-ID/duration normalization (both SHA-256 3b6d769737587a8401b2d1eff49376a9bd9aa9f189e49fd05ed412209a49787e). Lost-answer retry retains identical commands/fence without re-executing local work; original-grant refusal, matched/mismatched lost claims, fatal/transient probes, explicit/SIGTERM drain and one-ahead bounds pass. Review also found and fixed a prototype gap: a joined registration refusal now blocks the next task's failure acknowledgement as well as its successful completion.

SDK-PHP owns normal full CI and the requested independent full-source review next. Earlier primitive-only reviews do not accept this successor. Credential/private-path and public-boundary checks pass; local containers are absent. No published runtime, package, realistic-load result or performance claim is inferred; existing release/combined-minor and installed tuple gates remain separate. This issue stays open.

Draft #347 now contains the default-OFF Worker implementation and review-fix successor bf604c2533bbdb6ccbb88b32b001b091ca223702 / tree c80b22d9d16589d275b3c0dbd55a54c05eaead72. All 42 exact-head checks are terminal (41 success / one intended skip), including the full PHP 8.3 result of 8,402 tests / 85,970 assertions / 56 skips. Local causal socket controls, original-grant/fencing, retry/fairness, terminal ownership cleanup, shutdown and default-OFF packet parity pass; the unchanged corpus validator and all 83 policy tests pass.

The independent review found a real prototype prefetch-policy gap. The successor preserves identity rotation versus uncertain recovery, bounded delay and attempt counts, capacity fairness and occupied-slot heartbeat reporting. Exact reviewed-predecessor failures and successor passes are recorded in the fix handoff. SDK-PHP has requested independent closure of that Source HOLD; the draft remains unmerged until it closes. SDK-PHP then owns normal main qualification and cleanup.

Published delivery is still separate: General retains the #293 combined candidate/freeze/version decision and publication inputs; Server owns the immutable published producer handoff; SDK-PHP owns the actual installed consumer and realistic-load qualification. I asked General for that concrete release path and asked Server to bind ordinary prefetch admission/slot behavior to the actual producer revision. Source socket fixtures do not establish Server admission or speedup. No tag or narrower maintenance feature release is inferred, and this issue stays open.

The initial installed-consumer harness is now Source/main/cleanup complete in #348 at exact main 530da5f3c27d94fffe631700c5da16bbf87299ac / tree 7df8cb1b0788aa8639ef2d88fe3cf0949d56fe5c. Final receipt binds independent custody/main PASS, 42 PR contexts (41 success / one intended skip), 40 actual-main checks (39 success / one intended skip), PHP8.3 8,449 tests / 86,063 assertions, and exact topic/container cleanup. Product and workflow bytes remain the accepted 783d7ef bytes. Pipelining remains default OFF.

The maintained CLI prepares all three inline-carrier paths, actual older-producer fallback, useful controlled overlap and one successful-answer loss/retry, with authenticated single-read subject/lock selection and bounded cleanup. These are Source controls. The full driver has not executed against a published SDK or Native image, and the four initial scenarios do not close real expiry/refencing/opposite-slot/refusal or undelayed backend/load acceptance.

Concrete next inputs / owners: Server qualifies its one-line release-metadata successor and returns the actual producer publication/index/runner packet; General owns its execution selection, the actual Workflow successor caller/published2.21 support handoff and final SDK cohort/freeze under #293. I asked both owners and returned the exact source runner. sdk-php owns the selected normally installed proof and complete remaining matrix after that handoff. No release, runtime, profile or performance result is inferred; this issue stays open.

The remaining caller-owned qualification selections are implemented in draft #349, exact b24555ccf5ca2485284744df420902965b002351 / tree 101bebcee2a51d1c6e685c2d82df4026900c5c8b. This extends maintained #348 beyond its initial four rows: original-ID loss recovery, whole transit/buffer age, expiry and activity/workflow registration replacement, opposite-slot/missing-key capacity, default OFF/false, plus direct six-worker / 200-workflow / five-20ms-step local/remote cells and a three-repetition OFF/ON comparator. Timings and regressions remain descriptive.

102 focused Source tests / 206 assertions, lint and level-6 analysis pass; source container cleanup and boundary/secret checks pass. No new package/image/runtime/profile or actual load execution occurred. Workflow refence now follows the producer-owned accepted Source vector, including atomic release and a fresh n+1 claim under the original ID, rather than transferring the activity refusal to workflow recovery.

Next action/owner: SDK-PHP completes normal b245 CI and independent Security Source review, then maintained Source acceptance. General supplies the immutable published tuple and actual caller selection; SDK-PHP then owns the real installed/fallback/failure/load matrix. This issue and #328–#331 stay open for that acceptance; Source tests do not close it. Held #340 remains separate.

The remaining consumer qualification harness is now maintained at main b9525ef5f685f79acdc7da172f7cab39f8251e4b: #349 Source/main/cleanup receipt. Independent Source review passed, all 40 actual-main checks are terminal (39 success / one intended skip), and topic/container/temporary vendor-copy cleanup is verified. PHP8.3 passes 8,504 tests / 86,176 assertions.

This supplies the real-fault/authority/capacity/default/fallback and bounded-load harness source. No installed or Native row ran. Next owner/action: General returns the immutable published tuple and actual caller selection; SDK-PHP then executes and records this issue's actual required installed outcomes. The issue remains open. The current publisher prerequisite is the corrected Workflow caller's separate installer-404 NONPASS under central #297, not an accepted profile or product result.

The publisher prerequisite has advanced: the corrected actual Workflow caller is now independently accepted, with terminal/custody reconciliation. Its earlier installer-404 original remains NONPASS. SDK #340's refreshed exact head591 is Source/hosted PASS; its first merge/preview packet is prepared, unselected and under finite review.

This issue's implementation and complete maintained harness remain accepted Source; no installed/Native row has run. The real next input is the immutable published tuple under #293's current producer handoff. Server owns the frozen3c6 native publisher and actual index/platform/runner receipts; General owns distinct execution selection, published Workflow2.21 metadata/verifier and the combined SDK freeze. SDK-PHP has asked those owners and retains this issue's actual installed/fallback/fault/load qualification. A docs preview does not supply that tuple. The issue remains open.

The actual producer is published as Native Server 3.2.0, source/runner 3c6f1eda9c6d2ebc0ee5bec8ccf431900052be9e, index sha256:6916f6f3bb1ee58d071f10f888068fd321cd1576fb70dc97ce8e9e61da493d62, with the accepted schema-63 carrier handoff. Published Workflow 2.21.0 authority is also available. The remaining SDK release prerequisite is coordinated Laravel contract Source PR350, under fresh hosted qualification and independent Source review.

SDK-PHP retains this issue's implemented consumer and complete maintained harness, then the actual published-package/fallback/fencing/transit/lost-answer/load qualification. General owns the SDK freeze/publication and exact bounded tuple admission; Server owns the remaining Native parity/Compose receipts. The accepted preview and stable documentation builds supply no installed-product result. No installed row has run and this issue remains open.

The coordinated Laravel prerequisite is now actually integrated on main 6ec471d with Source/main acceptance, preview terminal/artifact/nondeployment acceptance, and owned-topic cleanup. Live documentation remains on the accepted published SDK3.9.5 tuple. The narrow Unreleased/published-history carry is draft351, under ordinary Source qualification.

This issue still requires released code and the actual installed consumer matrix. General owns the exact SDK candidate freeze/publication/tuple admission under #293; Server owns its already-prepared qualifier runner and separately selected producer receipts after the operator-required split. SDK-PHP retains package identity, default-OFF/fallback/fencing/transit/original-ID and bounded-load evidence. No SDK3.10 tag/package or installed row is claimed; this issue remains open.

The older-producer fallback candidate is now concrete at #293/6707736265, following General's read-only preparation request: published native Server3.0.5, source bc24589db02161a21febbb30211309537fd1e24e, annotated tag object 685f57ed19127a9e53ea6da28fddf310bfb16c3a, release433291416, original publisher41729986887/a1 SUCCESS.

Both DockerHub and GHCR agree on index sha256:3cb6e5bf8ec766ac2abd933369b40a77af62794b6d7b921ce21d3fb263e88b72; linux/amd64 manifest sha256:2659280f3ad58d1812ce00f44a927beed9781237143b8487ef12e1301eed551e; config sha256:0abb6928655d3a56e3ea9fb061f7edefd479a731eade94a30346dde9f7634800. Config platform, source, tag and publication-run labels agree. Static source's complete worker capability map omits all three carried-poll flags, and registration plus cluster discovery attach that map under protocol1.20. The linked proposal binds the exact source blobs and verified small metadata.

This supplies an eligible metadata/Source candidate for the real older-Server fallback row. No image layer was acquired and no runtime/install/profile or fallback invocation ran. Default-OFF, explicit-OFF, one-ahead/failure boundaries and actual load acceptance remain open under this issue's unchanged criteria. The actual fallback cell and exact SDK3.10 package/runner still require the separate complete tuple/admission after canonical3.2 qualification. SDK-PHP retains that installed proof; General retains the tuple/execution choice. No capability-masked3.2 result will replace this older producer.

The proposed Server3.0.5 older-producer binding now has independent METADATA / STATIC CONTRACT PASS for the exact candidate proposal. It confirms the release/source/tag chain, both registries' exact index/amd64 manifest/config, publisher labels, protocol1.20 and absent carried capability map. The annotated tag is unsigned and targets a valid GitHub-signed merge; no stronger signature claim is made.

The new-producer prerequisite also has an actual owned Source PR: Server1089, exact prepared head 1a5462058ac7c2c7008c372292a50b3580d3b3c7, follows accepted Server1088 main. Its ordinary Source qualification and later separately selected producer receipts remain incomplete. General retains exact SDK3.10 publication/tuple admission. SDK-PHP retains the actual installed older-producer fallback and this issue's default-OFF/one-ahead/failure/load matrix. No runtime row or release acceptance follows from this metadata PASS.

SDK source prerequisites are complete on main a969c1d4c07b5d1c1fab4e354f98a98a80e3cbd5: the coordinated Laravel contract and release-note/history carry have Source/main acceptance and exact topic cleanup. The complete maintained consumer harness remains accepted Source.

The real older-producer candidate Server3.0.5 now has metadata/static-contract PASS, not an executed fallback result. The new producer's qualifier is Server1089, current head 462ca242be0c45534702bc40989fa01786d1d5e7. Its narrow security delta is accepted and fresh Sonar passes; broad CI/full Source, main and separately selected canonical producer receipts remain pending. Old1a546's failed result stays attributed.

Next actions and owners: Server completes that qualifier Source/main and producer evidence; General freezes/publishes the exact SDK3.10 candidate and returns the bounded published tuple; SDK-PHP then verifies normal registry/archive identity and the actual installed fallback, lease/fencing/transit/original-ID, default-OFF/one-ahead/failure and bounded-load rows. Those owners have the concrete handoff requests and diagnosis. No SDK3.10 package, installed row or throughput result is claimed. This issue remains open for its released/installed acceptance.

Recheck 2026-11-07 19:40 UTC: Server's qualifier fix server#1091 is now merged and main-accepted (1c34f3b7). SDK source and the consumer harness are complete on main 69d0e408.

Next owner / action: Server returns the refreshed qualifier packet → Security effect review → General selects the effect and the 3.10.0 freeze/publication. sdk-php then runs the installed matrix for this issue (expiry/refence, opposite-slot, fallback on Server 3.0.5, default-OFF, load). Details: #293 status. I asked Server and General in chat.

SDK-owned next actions are on #293:

release source PR Prepare SDK 3.10.0 release source #353 (Security PASS, waiting on General's merge decision);
runner admission PR Admit one held consumer root in the installed completion-poll runner #354 (draft; CI and Security review next);
the execution packet in Admit one held consumer root in the installed completion-poll runner #354, which includes this issue's installed cells.

Owners: sdk-php for #353 and #354. General owns the tag and phase selections; Server owns the qualifier receipts.

Progress on this issue's installed cells:

the runner admission and execution packet are merged (Admit one held consumer root in the installed completion-poll runner #354, main 4e6c6b7b);
the runner PHP image (phase P) has passed.

Remaining: Server's Native 3.2.0 qualifier receipts, then General's 3.10.0 tag, then the selected installed phases. sdk-php executes the phases. Details are on #293 (latest comment).

State: no longer blocked externally. SDK 3.10.0 is published (tag 3.10.0 → 4e6c6b7b, release 437881850, Packagist). Its installed identity is independently accepted: lock f8142378, SDK inventory c562d378, runner seal 94a9604f. The Native 3.2.0 producer qualifier is accepted.

This issue's installed behavior is pending. Its cells are in the 18-cell phase D, prepared on #355/6713911732, plus the phase E load cells.

Next owner / action: General selects phase D, then sdk-php executes it and posts the per-cell results. The issue is not complete until those results exist. Details: #293 (latest comment).

Closing: acceptance met on the published 3.10.0 tuple.

Acceptance met: the opt-in pipeline (default OFF) passed its source tests, and installed D4 pipeline, pipeline-loss, pipeline-buffer-age, default-off, explicit-off, packet-defaults. It was qualified under realistic load: 36 load cells, explicit OFF/ON × local/remote × 3 repetitions × SQLite/MySQL/PostgreSQL. The descriptive results include regressions (MySQL local ON slower; SQLite remote ON slower) and are preserved. The option stays default OFF; any default change needs separate compatibility and rollout evidence, as the issue requires.

Evidence:

Source: accepted through Carry activity polls on workflow task completions #344–Add opt-in workflow completion pipelining #347 (source cohort at main 783d7ef, tree c80b22d9), with the consumer harness in Prepare installed native completion-poll consumer controls #348/Prepare the remaining completion-poll qualification matrix #349. The #355 opposite-slot correction was merged at main 1aebc537 (runner seal 76ab2cf5).
Published package: SDK 3.10.0, tag 3.10.0 → 4e6c6b7b, release 437881850, Packagist. Installed identity: lock f8142378, vendor 21e7ddae, SDK inventory c562d378.
Producer: Native Server 3.2.0 3c6f1eda (image 2a1c9659, canonical acceptance server#325/6713776065), with Server 3.0.5 2659280f for the no-capability fallback.
Installed qualification: all 54 declared completion-poll cells are independently accepted (audit 6715592197):
- functional D4, 18/18 (receipt);
- load SQLite 12/12 (receipt);
- load MySQL 12/12 under the 2 CPU / 2 GiB TLS profile (receipt);
- load PostgreSQL 12/12 (receipt).
Preserved: the earlier D1/D2/D3 and two MySQL NONPASS originals keep their raw classifications. Timings are descriptive only (performance_acceptance: not_assessed).

Navigation Menu

Repository navigation

Pipeline workflow task completions after local activities (opt-in) #332

Problem

Proposal

Reported benchmark evidence

Acceptance

Implementation constraints

Dependencies

Draft patches and validation boundary

4 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Pipeline workflow task completions after local activities (opt-in) #332

Description

Problem

Proposal

Reported benchmark evidence

Acceptance

Implementation constraints

Dependencies

Draft patches and validation boundary

Activity

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

4 remaining items

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 8, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions