Opt in to workflow continuation carried with an activity completion #341

Activity

1. rmcdaniel

   added

   kind:featureA public product capability or experience is requestedA public product capability or experience is requested

   kind:cross-repositoryWork spans more than one public repositoryWork spans more than one public repository

   priority:P0Immediate release or data-integrity riskImmediate release or data-integrity risk

   repo:sdk-phpOwned by the PHP SDK repositoryOwned by the PHP SDK repository

   completion:evidence-requiredClose only after all explicit acceptance and operational evidence is publicClose only after all explicit acceptance and operational evidence is public

   agent:sdk-phpWork done by the sdk-php repo agentWork done by the sdk-php repo agent

   on Nov 6, 2026
2. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized external reference for topic 24-sdk: 0001-Continue-a-run-s-workflow-task-from-its-activity-s-c.patch, part 1/1. Complete sanitized diff: 39,147 bytes / SHA-256 37d05531ec36a7a9f90356018faae6bd6ce8e4f47d6e831caec38a7348308327. Ordered parts preserve the complete diff bytes; none was applied or executed. The patch is a cumulative older-source draft, not maintained implementation or qualification. Mail headers/author/date/message/application instructions were removed.

   diff --git a/src/Client.php b/src/Client.php
   index e9a16b1..1dc82e6 100644
   --- a/src/Client.php
   +++ b/src/Client.php
   @@ -2441,6 +2441,9 @@ final class Client implements WorkflowClientInterface
        /**
         * @param array{worker_id: string, task_queue: string, poll_request_id: string, history_page_size: int}|null $workflowPoll
         *     The Worker's next workflow poll, which the Server claims in the completion's transaction.
   +     * @param array<string, mixed>|null $continuation
   +     *     The Worker's completion of the workflow task the result makes ready, which the Server
   +     *     records in the same transaction when the run's history is still the one it was computed from.
         * @return array<string, mixed>
         */
        public function completeActivityTask(
   @@ -2449,12 +2452,14 @@ final class Client implements WorkflowClientInterface
            string $leaseOwner,
            mixed $result,
            ?array $workflowPoll = null,
   +        ?array $continuation = null,
        ): array {
            return $this->worker('POST', self::ACTIVITY_TASK_PATH.$this->segment($taskId).'/complete', [
                'activity_attempt_id' => $activityAttemptId,
                'lease_owner' => $leaseOwner,
                'result' => $this->codec->envelope($result),
   -        ] + ($workflowPoll === null ? [] : ['workflow_poll' => $workflowPoll]));
   +        ] + ($workflowPoll === null ? [] : ['workflow_poll' => $workflowPoll])
   +            + ($continuation === null ? [] : ['workflow_continuation' => $continuation]));
        }
    
        /**
   diff --git a/src/Worker.php b/src/Worker.php
   index 0f21226..d58162f 100644
   --- a/src/Worker.php
   +++ b/src/Worker.php
   @@ -128,6 +128,14 @@ final class Worker
        private ?array $carriedActivityPollAnswer = null;
        private bool $activityCompletionWorkflowPollSupported = false;
        private bool $completionWorkflowPollSupported = false;
   +    private bool $workflowContinuationSupported = false;
   +    /**
   +     * @var array{activity_task_id: string, history: list<array<string, mixed>>, task: array<string, mixed>}|null
   +     * The run whose next workflow task the completion of the activity this
   +     * Worker leased may complete (`workflowContinuation`): the run's whole
   +     * history through that activity's start, and its workflow task.
   +     */
   +    private ?array $continuable = null;
        /** The request ID of the workflow poll the last completion carried. */
        private ?string $carriedWorkflowPollRequestId = null;
        /** Whether that completion's claim leased a workflow task, or null when it did not answer. */
   @@ -859,7 +867,12 @@ final class Worker
                $this->carriedWorkflowPollLeased = null;
                $this->carriedWorkflowPollAnswer = null;
            }
   -        if ($this->shutdownRequested && !$carriedWorkflowLeased && $prefetched === null) {
   +        // The activity a continuation's carried poll leased (`workflowContinuation`)
   +        // runs first, as it would after the workflow task's own completion,
   +        // even during shutdown; that claim took no workflow task for the
   +        // completion's carried workflow poll.
   +        $continued = $prefetched === null && !$carriedWorkflowLeased && $this->carriedActivityPollLeased === true;
   +        if ($this->shutdownRequested && !$carriedWorkflowLeased && $prefetched === null && !$continued) {
                return false;
            }
    
   @@ -869,51 +882,57 @@ final class Worker
            }
    
            $handled = false;
   -        $workflowPoll = $prefetched['poll'] ?? $this->pollWithRetry(
   -            'workflow',
   -            fn (string $requestId, int $timeoutSeconds): array => $this->pollClient($requestId, $carriedWorkflowRequestId, $carriedWorkflowAnswer)->pollWorkflowTaskResponse(
   -                $this->workerId,
   -                $this->taskQueue,
   -                $timeoutSeconds,
   -                $requestId,
   -                self::WORKFLOW_HISTORY_PAGE_SIZE,
   -            ),
   -            $pollTimeoutSeconds,
   -            $carriedWorkflowRequestId,
   -            $carriedWorkflowLeased,
   -        );
   -        if ($workflowPoll === null) {
   -            return false;
   -        }
   -        if ($this->stopForTerminalPoll($workflowPoll)) {
   -            return false;
   -        }
   -        $this->workflowGrant = is_array($workflowPoll['task'] ?? null) ? [
   -            'task_id' => (string) ($workflowPoll['task']['task_id'] ?? ''),
   -            'received' => $prefetched['received'] ?? $this->workflowGrantReceived($workflowPoll, $carriedWorkflowAnswer),
   -            'server_time' => $workflowPoll['server_time'] ?? null,
   -        ] : null;
   -        $this->rememberWorkflowMemoCapability($workflowPoll);
   -        $capabilities = $workflowPoll['server_capabilities'] ?? null;
   -        $supportedCommands = is_array($capabilities) ? ($capabilities['supported_workflow_task_commands'] ?? null) : null;
   -        $this->serviceOperationsSupported = is_array($supportedCommands)
   -            && in_array('start_service_operation', $supportedCommands, true);
   -        $this->childStartTurnSupported = is_array($capabilities)
   -            && ($capabilities['child_workflow_start_turn'] ?? null) === true;
   -        $this->completionActivityPollSupported = is_array($capabilities)
   -            && ($capabilities['workflow_task_completion_activity_poll'] ?? null) === true;
   -        $this->activityCompletionWorkflowPollSupported = is_array($capabilities)
   -            && ($capabilities['activity_completion_workflow_poll'] ?? null) === true;
   -        $this->completionWorkflowPollSupported = is_array($capabilities)
   -            && ($capabilities['workflow_task_completion_workflow_poll'] ?? null) === true;
   -        $this->heartbeatIfDue();
   -        if ($this->crossKindPollWakeReceived($workflowPoll)) {
   -            $pollTimeoutSeconds = 0;
   -        }
   -        $workflowTask = $this->taskFromPoll($workflowPoll);
   -        if ($workflowTask !== null) {
   -            $this->executePolledTask('workflow', $workflowTask);
   -            $handled = true;
   +        if (!$continued) {
   +            $workflowPoll = $prefetched['poll'] ?? $this->pollWithRetry(
   +                'workflow',
   +                fn (string $requestId, int $timeoutSeconds): array => $this->pollClient($requestId, $carriedWorkflowRequestId, $carriedWorkflowAnswer)->pollWorkflowTaskResponse(
   +                    $this->workerId,
   +                    $this->taskQueue,
   +                    $timeoutSeconds,
   +                    $requestId,
   +                    self::WORKFLOW_HISTORY_PAGE_SIZE,
   +                ),
   +                $pollTimeoutSeconds,
   +                $carriedWorkflowRequestId,
   +                $carriedWorkflowLeased,
   +            );
   +            if ($workflowPoll === null) {
   +                return false;
   +            }
   +            if ($this->stopForTerminalPoll($workflowPoll)) {
   +                return false;
   +            }
   +            $this->workflowGrant = is_array($workflowPoll['task'] ?? null) ? [
   +                'task_id' => (string) ($workflowPoll['task']['task_id'] ?? ''),
   +                'received' => $prefetched['received'] ?? $this->workflowGrantReceived($workflowPoll, $carriedWorkflowAnswer),
   +                'server_time' => $workflowPoll['server_time'] ?? null,
   +            ] : null;
   +            $this->rememberWorkflowMemoCapability($workflowPoll);
   +            $capabilities = $workflowPoll['server_capabilities'] ?? null;
   +            $supportedCommands = is_array($capabilities) ? ($capabilities['supported_workflow_task_commands'] ?? null) : null;
   +            $this->serviceOperationsSupported = is_array($supportedCommands)
   +                && in_array('start_service_operation', $supportedCommands, true);
   +            $this->childStartTurnSupported = is_array($capabilities)
   +                && ($capabilities['child_workflow_start_turn'] ?? null) === true;
   +            $this->completionActivityPollSupported = is_array($capabilities)
   +                && ($capabilities['workflow_task_completion_activity_poll'] ?? null) === true;
   +            $this->activityCompletionWorkflowPollSupported = is_array($capabilities)
   +                && ($capabilities['activity_completion_workflow_poll'] ?? null) === true;
   +            $this->completionWorkflowPollSupported = is_array($capabilities)
   +                && ($capabilities['workflow_task_completion_workflow_poll'] ?? null) === true;
   +            $this->workflowContinuationSupported = is_array($capabilities)
   +                && ($capabilities['activity_completion_workflow_continuation'] ?? null) === true
   +                && ($capabilities['carried_activity_poll_eager_dispatch'] ?? null) === true
   +                && ($capabilities['carried_activity_poll_history_tail'] ?? null) === true;
   +            $this->heartbeatIfDue();
   +            if ($this->crossKindPollWakeReceived($workflowPoll)) {
   +                $pollTimeoutSeconds = 0;
   +            }
   +            $workflowTask = $this->taskFromPoll($workflowPoll);
   +            if ($workflowTask !== null) {
   +                $this->executePolledTask('workflow', $workflowTask);
   +                $handled = true;
   +            }
            }
            // The workflow task's completion carried this activity poll. A lease
            // its claim took is delivered by the poll under the same request ID,
   @@ -1893,10 +1912,22 @@ final class Worker
                // Worker that stops once this task is done leases nothing more.
                $stopping = $this->shutdownRequested || $this->deferredFailure !== null;
                $activityPoll = null;
   +            // Commands that came from the run's history alone may continue:
   +            // the poll leases the activity they schedule first, with the
   +            // history this task added, so that activity's completion can
   +            // complete the run's next workflow task (`workflowContinuation`).
   +            $continuable = $this->workflowContinuationSupported && ($updateId === null || $updateId === '')
   +                && $messageStreamCursors === [] && $messageStreamWaits === [] && $this->claimCancellation === null
   +                && !$this->localActivityRan && $history !== [] && self::continuableCommands($commands)
   +                && self::schedulesActivity($commands);
                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];
   +                if ($continuable) {
   +                    $activityPoll['eager'] = true;
   +                    $activityPoll['history_after_sequence'] = $history[array_key_last($history)]['sequence'] ?? null;
   +                }
                }
                // It carries the next workflow poll too, which the Server claims
                // for only when that activity poll leases nothing, so this Worker
   @@ -1913,6 +1944,8 @@ final class Worker
                $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);
   +            $this->continuable = $continuable && $activityPoll !== null
   +                ? self::continuableRun($task, $history, $completion['activity_poll'] ?? null) : null;
                // A lease held already stays for the next tick.
                if ($workflowPoll !== null) {
                    $this->carriedWorkflowPollLeased = self::carriedWorkflowPollLeased($completion);
   @@ -2996,6 +3029,8 @@ final class Worker
            $leaseOwner = (string) ($task['lease_owner'] ?? $this->workerId);
            $activityType = (string) ($task['activity_type'] ?? '');
            $callbackStopped = false;
   +        $continuable = ($this->continuable['activity_task_id'] ?? null) === $taskId ? $this->continuable : null;
   +        $this->continuable = null;
            try {
                $this->trackWorkerSessionFromTask($task);
                $handler = $this->activities[$activityType] ?? null;
   @@ -3067,9 +3102,15 @@ final class Worker
                    $result = $this->invokeActivity($handler, $context, $arguments,
                        ExecutionTelemetry::taskFields($task, $this->client->namespace, $this->taskQueue), authorizedTask: $task);
                }
   -            // The completion carries the next workflow poll, so the Server
   -            // claims the workflow task the result makes ready in the same
   -            // transaction, unless this Worker holds or awaits one already.
   +            // The completion may carry this Worker's completion of the
   +            // workflow task the result makes ready, which the Server records
   +            // with the result when the run's history is still the one it held.
   +            $continuation = $continuable === null || $this->shutdownRequested
   +                ? null : $this->workflowContinuation($continuable, $result);
   +            // It carries the next workflow poll, so the Server claims the
   +            // workflow task the result makes ready in the same transaction
   +            // when there is no continuation or it is refused, unless this
   +            // Worker holds or awaits one already.
                $workflowPoll = null;
                if ($this->activityCompletionWorkflowPollSupported && !$this->shutdownRequested && !$this->workflowPollHeld()) {
                    $this->carriedWorkflowPollRequestId = 'php-workflow-poll-'.bin2hex(random_bytes(16));
   @@ -3079,7 +3120,7 @@ final class Worker
                }
                $completion = $this->retryStorageAdmission(
                    'activity_complete',
   -                fn (): array => $this->client->completeActivityTask($taskId, $attemptId, $leaseOwner, $result, $workflowPoll),
   +                fn (): array => $this->client->completeActivityTask($taskId, $attemptId, $leaseOwner, $result, $workflowPoll, $continuation),
                    activityTaskLease: [
                        'task_id' => $taskId,
                        'activity_attempt_id' => $attemptId,
   @@ -3090,6 +3131,9 @@ final class Worker
                    $this->carriedWorkflowPollLeased = self::carriedWorkflowPollLeased($completion);
                    $this->carriedWorkflowPollAnswer = $this->carriedPollAnswer($completion['workflow_poll'] ?? null);
                }
   +            if ($continuation !== null) {
   +                $this->continued($continuable, $continuation, $completion);
   +            }
            } catch (Throwable $exception) {
                if ($exception instanceof ServerException && $this->discardClosedActivityAttempt($task, $exception)) {
                    return;
   @@ -3134,6 +3178,185 @@ final class Worker
            }
        }
    
   +    /**
   +     * Whether a workflow task's commands may be a continuation's: only
   +     * commands the Server records in one (schedule_activity, start_timer,
   +     * complete_workflow and record_version_marker), with at most one
   +     * activity, outside any parallel group, selection or cancellation scope.
   +     *
   +     * @param list<array<string, mixed>> $commands
   +     */
   +    private static function continuableCommands(array $commands): bool
   +    {
   +        $activities = 0;
   +        foreach ($commands as $command) {
   +            $type = $command['type'] ?? null;
   +            if (!in_array($type, ['schedule_activity', 'start_timer', 'complete_workflow', 'record_version_marker'], true)) {
   +                return false;
   +            }
   +            foreach (array_keys($command) as $field) {
   +                if (str_starts_with((string) $field, 'parallel_') || str_starts_with((string) $field, 'selection_')
   +                    || $field === 'cancellation_scope_id') {
   +                    return false;
   +                }
   +            }
   +            $activities += $type === 'schedule_activity' ? 1 : 0;
   +        }
   +
   +        return $commands !== [] && $activities <= 1;
   +    }
   +
   +    /** @param list<array<string, mixed>> $commands */
   +    private static function schedulesActivity(array $commands): bool
   +    {
   +        foreach ($commands as $command) {
   +            if (($command['type'] ?? null) === 'schedule_activity') {
   +                return true;
   +            }
   +        }
   +
   +        return false;
   +    }
   +
   +    /**
   +     * The run a carried activity poll's lease may continue: when it leased an
   +     * activity of the workflow task's run with the history after the one
   +     * this Worker held, the run's whole history, numbered from one without a
   +     * gap and ending with that activity attempt's start. Null otherwise, or
   +     * when the history holds a cancellation, whose replay this Worker does
   +     * not speculate on.
   +     *
   +     * @param array<string, mixed> $task
   +     * @param list<array<string, mixed>> $held
   +     * @return array{activity_task_id: string, history: list<array<string, mixed>>, task: array<string, mixed>}|null
   +     */
   +    private static function continuableRun(array $task, array $held, mixed $poll): ?array
   +    {
   +        $activity = is_array($poll) && ($poll['leased'] ?? null) === true && is_array($poll['answer'] ?? null)
   +            ? ($poll['answer']['task'] ?? null) : null;
   +        $tail = is_array($activity) ? ($activity['history_tail'] ?? null) : null;
   +        if (!is_array($tail) || !is_array($tail['events'] ?? null) || !array_is_list($tail['events']) || $held === []
   +            || ($activity['workflow_run_id'] ?? null) !== ($task['run_id'] ?? null)
   +            || ($tail['after_sequence'] ?? null) !== ($held[array_key_last($held)]['sequence'] ?? null)) {
   +            return null;
   +        }
   +        $history = [...$held, ...$tail['events']];
   +        foreach ($history as $index => $event) {
   +            if (!is_array($event) || ($event['sequence'] ?? null) !== $index + 1
   +                || !is_array($event['payload'] ?? null) || str_contains((string) ($event['event_type'] ?? ''), 'Cancel')) {
   +                return null;
   +            }
   +        }
   +        $started = $history[array_key_last($history)];
   +        if (($started['event_type'] ?? null) !== 'ActivityStarted'
   +            || ($started['payload']['activity_attempt_id'] ?? null) !== ($activity['activity_attempt_id'] ?? null)) {
   +            return null;
   +        }
   +        unset($task['history_events']);
   +
   +        return ['activity_task_id' => (string) ($activity['task_id'] ?? ''), 'history' => $history, 'task' => $task];
   +    }
   +
   +    /**
   +     * This Worker's completion of the workflow task an activity's result
   +     * makes ready: the run replayed on the history it holds through the
   +     * activity's start, then the result as the Server records it. Null
   +     * when the commands could depend on anything that replay does not
   +     * show, such as a local activity, a cancellation or a message stream,
   +     * or are not ones a continuation carries.
   +     *
   +     * @param array{activity_task_id: string, history: list<array<string, mixed>>, task: array<string, mixed>} $continuable
   +     * @return array<string, mixed>|null
   +     */
   +    private function workflowContinuation(array $continuable, mixed $result): ?array
   +    {
   +        $history = $continuable['history'];
   +        $task = $continuable['task'];
   +        $handler = $this->workflows[(string) ($task['workflow_type'] ?? '')] ?? null;
   +        if ($handler === null || ($task['cancel_requested'] ?? false) === true) {
   +            return null;
   +        }
   +        $task['cancellation_request'] = null;
   +        $started = $history[array_key_last($history)];
   +        $basis = (int) $started['sequence'];
   +        // The ActivityCompleted the Server appends for this result: the
   +        // started snapshot, closed, with the result as this Worker sends it.
   +        $completed = $started;
   +        $completed['sequence'] = $basis + 1;
   +        $completed['event_type'] = 'ActivityCompleted';
   +        $completed['payload']['result'] = $this->client->payloadCodec()->envelope($result);
   +        $completed['payload']['payload_codec'] = 'avro';
   +        if (is_array($completed['payload']['activity'] ?? null)) {
   +            $completed['payload']['activity']['status'] = 'completed';
   +        }
   +        $localActivityCalled = false;
   +        try {
   +            $input = $this->decodeArguments($task['arguments'] ?? $task['input'] ?? null);
   +            WorkflowEntryArguments::validate($handler, $input, (string) $task['workflow_type']);
   +            $replay = $this->replayer->replay($handler, [...$history, $completed], $input, $this->taskQueue, $task,
   +                function () use (&$localActivityCalled): array {
   +                    $localActivityCalled = true;
   +                    throw new WorkflowClaimAborted('A workflow continuation runs no local activity.');
   +                }, $this->enablePreparedLocalActivities, $this->preparedLocalActivityGroupsSupported,
   +                $this->preparedLocalActivityCancellationPolicies, allowCancellationScopeAuthoring: $this->enableCooperativeCancellation,
   +                replayCommittedCancellationScopes: $this->enableCancellationScopes,
   +                prepareCancellationScopeDelivery: $this->enableCancellationScopes,
   +                allowScopedPreparedLocalActivities: $this->enableCancellationScopes && $this->enablePreparedLocalActivities);
   +            $commands = $replay->commands;
   +            $this->assertWorkflowMemoUpdatesAvailable($commands);
   +            $this->assertServiceOperationsAvailable($commands);
   +            $this->assertChildStartTurnAvailable($commands);
   +            $this->assertCancellationPoliciesAvailable($commands);
   +        } catch (Throwable) {
   +            return null;
   +        }
   +        if ($localActivityCalled || $replay->terminalFailure !== null || $replay->messageStreamCursors !== []
   +            || $replay->messageStreamWaits !== [] || $replay->cancellationDelivery !== null
   +            || $replay->preparedLocalActivity !== null || $replay->preparedLocalActivityGroup !== null
   +            || $replay->cancellationScopeOpening !== null || $replay->cancellationScopeDelivery !== null
   +            || !self::continuableCommands($commands)) {
   +            return null;
   +        }
   +        $continuation = ['basis_sequence' => $basis, 'task_queue' => $this->taskQueue, 'commands' => $commands];
   +        // Its poll leases the activity the commands schedule as a workflow
   +        // task completion's does, with the history the continuation adds.
   +        if (self::schedulesActivity($commands) && $this->completionActivityPollSupported) {
   +            $this->carriedActivityPollRequestId = 'php-activity-poll-'.bin2hex(random_bytes(16));
   +            $continuation['activity_poll'] = ['worker_id' => $this->workerId, 'task_queue' => $this->taskQueue,
   +                'poll_request_id' => $this->carriedActivityPollRequestId, 'eager' => true,
   +                'history_after_sequence' => $basis];
   +        }
   +
   +        return $continuation;
   +    }
   +
   +    /**
   +     * Records what the Server answered a continuation: a recorded one's
   +     * carried activity poll, whose lease may continue the run again; a
   +     * refused one claimed nothing for its poll, and the workflow task stays
   +     * for this Worker's carried workflow poll or its next one.
   +     *
   +     * @param array{activity_task_id: string, history: list<array<string, mixed>>, task: array<string, mixed>} $continuable
   +     * @param array<string, mixed> $continuation
   +     * @param array<string, mixed> $completion
   +     */
   +    private function continued(array $continuable, array $continuation, array $completion): void
   +    {
   +        $answer = $completion['workflow_continuation'] ?? null;
   +        if (!array_key_exists('activity_poll', $continuation)) {
   +            return;
   +        }
   +        if (!is_array($answer) || ($answer['recorded'] ?? null) !== true) {
   +            $this->carriedActivityPollRequestId = null;
   +
   +            return;
   +        }
   +        $leased = $answer['activity_poll']['leased'] ?? null;
   +        $this->carriedActivityPollLeased = is_bool($leased) ? $leased : null;
   +        $this->carriedActivityPollAnswer = $this->carriedPollAnswer($answer['activity_poll'] ?? null);
   +        $this->continuable = self::continuableRun($continuable['task'], $continuable['history'], $answer['activity_poll'] ?? null);
   +    }
   +
        /** This is reached only after the isolated callback stopped and its relay joined. */
        private function acknowledgeStoppedRemoteActivity(string $taskId, string $attemptId, string $leaseOwner): void
        {
   diff --git a/tests/WorkerContinuationTest.php b/tests/WorkerContinuationTest.php
   new file mode 100644
   index 0000000..8179150
   --- /dev/null
   +++ b/tests/WorkerContinuationTest.php
   @@ -0,0 +1,296 @@
   +<?php
   +
   +declare(strict_types=1);
   +
   +namespace DurableWorkflow\Tests;
   +
   +use DurableWorkflow\Client;
   +use DurableWorkflow\Codec\AvroPayloadCodec;
   +use DurableWorkflow\Tests\Support\FakeTransport;
   +use DurableWorkflow\Worker;
   +use DurableWorkflow\Worker\ActivityContext;
   +use DurableWorkflow\Worker\WorkflowContext;
   +use PHPUnit\Framework\TestCase;
   +
   +/**
   + * An activity completion can carry the Worker's completion of the workflow
   + * task the result makes ready, computed from the run's history the Worker
   + * holds through the activity's start. The Server records both when nothing
   + * else was written since, and leases the activity the commands schedule to
   + * the continuation's carried poll.
   + */
   +final class WorkerContinuationTest extends TestCase
   +{
   +    public function testAnActivityCompletionCarriesTheRunsNextWorkflowTaskCompletion(): void
   +    {
   +        $workflowPolls = 0;
   +        $activityPolls = 0;
   +        $completions = [];
   +        $transport = $this->transport(static function (string $uri, ?array $body) use (&$workflowPolls, &$activityPolls, &$completions): ?array {
   +            if (str_ends_with($uri, '/workflow-tasks/poll')) {
   +                return ++$workflowPolls === 1 ? self::workflowLease() : ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            if (str_ends_with($uri, '/activity-tasks/poll')) {
   +                $activityPolls++;
   +
   +                return ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            if (str_ends_with($uri, '/workflow-tasks/chain-task-1/complete')) {
   +                $completions[] = ['chain-task-1', $body];
   +
   +                return ['outcome' => 'completed', 'recorded' => true, 'activity_poll' => self::activityLease(1, 1)];
   +            }
   +            if (preg_match('#/activity-tasks/(chain-activity-\d)/complete$#', $uri, $match) === 1) {
   +                $completions[] = [$match[1], $body];
   +                $continued = ['recorded' => true, 'outcome' => 'completed', 'task_id' => 'chain-task-next', 'run_status' => 'waiting'];
   +                if (isset($body['workflow_continuation']['activity_poll'])) {
   +                    $continued['activity_poll'] = self::activityLease(2, (int) $body['workflow_continuation']['basis_sequence']);
   +                } else {
   +                    $continued['run_status'] = 'completed';
   +                }
   +
   +                return ['outcome' => 'completed', 'recorded' => true, 'next_task_id' => 'chain-task-next', 'workflow_continuation' => $continued];
   +            }
   +
   +            return null;
   +        });
   +        $worker = $this->worker($transport);
   +
   +        self::assertTrue($worker->tick(0));
   +        self::assertTrue($worker->tick(0));
   +
   +        self::assertSame(1, $workflowPolls);
   +        self::assertSame(0, $activityPolls);
   +        self::assertSame(['chain-task-1', 'chain-activity-1', 'chain-activity-2'], array_column($completions, 0));
   +        self::assertTrue($completions[0][1]['activity_poll']['eager'] ?? null);
   +        self::assertSame(1, $completions[0][1]['activity_poll']['history_after_sequence'] ?? null);
   +        $first = $completions[1][1]['workflow_continuation'] ?? null;
   +        self::assertIsArray($first);
   +        self::assertSame(3, $first['basis_sequence'] ?? null);
   +        self::assertSame('chain-workers', $first['task_queue'] ?? null);
   +        self::assertSame(['schedule_activity'], array_column($first['commands'], 'type'));
   +        self::assertSame('chain.step', $first['commands'][0]['activity_type'] ?? null);
   +        self::assertSame(['worker_id' => 'worker-1', 'task_queue' => 'chain-workers',
   +            'poll_request_id' => $first['activity_poll']['poll_request_id'] ?? null, 'eager' => true,
   +            'history_after_sequence' => 3], $first['activity_poll'] ?? null);
   +        $last = $completions[2][1]['workflow_continuation'] ?? null;
   +        self::assertIsArray($last);
   +        self::assertSame(6, $last['basis_sequence'] ?? null);
   +        self::assertArrayNotHasKey('activity_poll', $last);
   +        self::assertSame(['complete_workflow'], array_column($last['commands'], 'type'));
   +        self::assertSame([1, 2], (new AvroPayloadCodec())->decodeEnvelope($last['commands'][0]['result']));
   +    }
   +
   +    public function testARefusedContinuationLeavesTheWorkflowTaskForTheCarriedPoll(): void
   +    {
   +        $workflowPolls = 0;
   +        $completions = [];
   +        $transport = $this->transport(static function (string $uri, ?array $body) use (&$workflowPolls, &$completions): ?array {
   +            if (str_ends_with($uri, '/workflow-tasks/poll')) {
   +                return ++$workflowPolls === 1 ? self::workflowLease() : ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            if (str_ends_with($uri, '/activity-tasks/poll')) {
   +                return ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            if (preg_match('#/workflow-tasks/(chain-task-\d)/complete$#', $uri, $match) === 1) {
   +                $completions[] = [$match[1], $body];
   +
   +                return ['outcome' => 'completed', 'recorded' => true,
   +                    'activity_poll' => $match[1] === 'chain-task-1' ? self::activityLease(1, 1) : ['leased' => false]];
   +            }
   +            if (str_ends_with($uri, '/activity-tasks/chain-activity-1/complete')) {
   +                $completions[] = ['chain-activity-1', $body];
   +                $task = self::workflowLease('chain-task-2')['task'];
   +                $task['history_events'] = [...self::history(1), self::completed(1, 4, 1)];
   +
   +                return ['outcome' => 'completed', 'recorded' => true, 'next_task_id' => 'chain-task-2',
   +                    'workflow_continuation' => ['recorded' => false, 'reason' => 'history_moved'],
   +                    'workflow_poll' => ['leased' => true, 'answer' => ['poll_status' => 'leased', 'task' => $task,
   +                        'server_capabilities' => self::capabilities()]]];
   +            }
   +
   +            return null;
   +        });
   +        $worker = $this->worker($transport);
   +
   +        self::assertTrue($worker->tick(0));
   +        self::assertTrue($worker->tick(0));
   +
   +        self::assertSame(1, $workflowPolls);
   +        self::assertSame(['chain-task-1', 'chain-activity-1', 'chain-task-2'], array_column($completions, 0));
   +        self::assertIsArray($completions[1][1]['workflow_continuation'] ?? null);
   +        self::assertIsString($completions[1][1]['workflow_poll']['poll_request_id'] ?? null);
   +        // The Worker's own completion of the task the refused one would have.
   +        self::assertSame($completions[1][1]['workflow_continuation']['commands'], $completions[2][1]['commands']);
   +    }
   +
   +    public function testAWorkflowFailureIsNotAContinuation(): void
   +    {
   +        $workflowPolls = 0;
   +        $completions = [];
   +        $transport = $this->transport(static function (string $uri, ?array $body) use (&$workflowPolls, &$completions): ?array {
   +            if (str_ends_with($uri, '/workflow-tasks/poll')) {
   +                return ++$workflowPolls === 1 ? self::workflowLease('chain-task-1', 'chain.failing') : ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            if (str_ends_with($uri, '/activity-tasks/poll')) {
   +                return ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            if (str_ends_with($uri, '/workflow-tasks/chain-task-1/complete')) {
   +                $completions[] = ['chain-task-1', $body];
   +
   +                return ['outcome' => 'completed', 'recorded' => true, 'activity_poll' => self::activityLease(1, 1)];
   +            }
   +            if (str_ends_with($uri, '/activity-tasks/chain-activity-1/complete')) {
   +                $completions[] = ['chain-activity-1', $body];
   +
   +                return ['outcome' => 'completed', 'recorded' => true, 'workflow_poll' => ['leased' => false]];
   +            }
   +
   +            return null;
   +        });
   +        $worker = $this->worker($transport);
   +
   +        self::assertTrue($worker->tick(0));
   +
   +        self::assertSame(['chain-task-1', 'chain-activity-1'], array_column($completions, 0));
   +        self::assertTrue($completions[0][1]['activity_poll']['eager'] ?? null);
   +        self::assertArrayNotHasKey('workflow_continuation', $completions[1][1]);
   +        self::assertIsString($completions[1][1]['workflow_poll']['poll_request_id'] ?? null);
   +    }
   +
   +    /**
   +     * @param \Closure(string, ?array<string, mixed>): ?array<string, mixed> $handle
   +     */
   +    private function transport(\Closure $handle): FakeTransport
   +    {
   +        return new FakeTransport(handler: static function (string $method, string $uri, array $headers, ?array $body) use ($handle): array {
   +            if (str_ends_with($uri, '/query-tasks/poll')) {
   +                return ['task' => null, 'poll_status' => 'empty'];
   +            }
   +            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];
   +            }
   +
   +            return $handle($uri, $body) ?? self::fail("Unexpected worker request: {$method} {$uri}");
   +        });
   +    }
   +
   +    private function worker(FakeTransport $transport): Worker
   +    {
   +        $worker = new Worker(new Client('https://server.example', transport: $transport), 'chain-workers', workerId: 'worker-1');
   +        $worker->registerWorkflow('chain.steps', static fn (WorkflowContext $context): array => [
   +            $context->activity('chain.step'),
   +            $context->activity('chain.step'),
   +        ]);
   +        $worker->registerWorkflow('chain.failing', static function (WorkflowContext $context): never {
   +            $context->activity('chain.step');
   +
   +            throw new \RuntimeException('The step was not enough.');
   +        });
   +        $steps = 0;
   +        $worker->registerActivity('chain.step', static function (ActivityContext $context) use (&$steps): int {
   +            return ++$steps;
   +        });
   +
   +        return $worker;
   +    }
   +
   +    /** @return array<string, mixed> A workflow poll's answer leasing the run's first workflow task. */
   +    private static function workflowLease(string $taskId = 'chain-task-1', string $type = 'chain.steps'): array
   +    {
   +        return [
   +            'poll_status' => 'leased',
   +            'task' => [
   +                'task_id' => $taskId,
   +                'workflow_task_attempt' => 1,
   +                'lease_owner' => 'worker-1',
   +                'workflow_id' => 'chain-workflow',
   +                'run_id' => 'chain-run',
   +                'workflow_type' => $type,
   +                'payload_codec' => 'avro',
   +                'history_events' => [self::event(1, 'WorkflowStarted', [])],
   +                'task_kind' => 'workflow',
   +            ],
   +            'server_capabilities' => self::capabilities(),
   +        ];
   +    }
   +
   +    /**
   +     * A carried activity poll's lease of the run's $step-th activity, with
   +     * the history the workflow task scheduling it added after $after.
   +     *
   +     * @return array<string, mixed>
   +     */
   +    private static function activityLease(int $step, int $after): array
   +    {
   +        $events = array_values(array_filter(self::history($step), static fn (array $event): bool => $event['sequence'] > $after));
   +
   +        return ['leased' => true, 'answer' => [
   +            'poll_status' => 'leased',
   +            'task' => ['task_id' => "chain-activity-{$step}", 'activity_attempt_id' => "chain-attempt-{$step}",
   +                'lease_owner' => 'worker-1', 'activity_type' => 'chain.step', 'payload_codec' => 'avro',
   +                'workflow_id' => 'chain-workflow', 'workflow_run_id' => 'chain-run',
   +                'history_tail' => ['after_sequence' => $after, 'events' => $events]],
   +            'server_capabilities' => self::capabilities(),
   +        ]];
   +    }
   +
   +    /**
   +     * The run's history through the start of its $through-th activity, each
   +     * earlier one completed with its step number.
   +     *
   +     * @return list<array<string, mixed>>
   +     */
   +    private static function history(int $through): array
   +    {
   +        $events = [self::event(1, 'WorkflowStarted', [])];
   +        for ($step = 1; $step <= $through; $step++) {
   +            if ($step > 1) {
   +                $events[] = self::completed($step - 1, count($events) + 1, $step - 1);
   +            }
   +            $events[] = self::event(count($events) + 1, 'ActivityScheduled', self::activity($step));
   +            $events[] = self::event(count($events) + 1, 'ActivityStarted', self::activity($step) + ['activity_attempt_id' => "chain-attempt-{$step}"]);
   +        }
   +
   +        return $events;
   +    }
   +
   +    /** @return array<string, mixed> */
   +    private static function completed(int $step, int $sequence, mixed $result): array
   +    {
   +        return self::event($sequence, 'ActivityCompleted', self::activity($step) + [
   +            'activity_attempt_id' => "chain-attempt-{$step}", 'result' => (new AvroPayloadCodec())->envelope($result),
   +            'payload_codec' => 'avro',
   +        ]);
   +    }
   +
   +    /** @return array<string, mixed> */
   +    private static function activity(int $step): array
   +    {
   +        return ['sequence' => $step, 'activity_type' => 'chain.step', 'activity' => ['type' => 'chain.step']];
   +    }
   +
   +    /**
   +     * @param array<string, mixed> $payload
   +     * @return array<string, mixed>
   +     */
   +    private static function event(int $sequence, string $type, array $payload): array
   +    {
   +        return ['id' => "chain-event-{$sequence}", 'sequence' => $sequence, 'event_type' => $type, 'payload' => $payload];
   +    }
   +
   +    /** @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,
   +            'activity_completion_workflow_continuation' => true,
   +            'carried_activity_poll_eager_dispatch' => true,
   +            'carried_activity_poll_history_tail' => true,
   +        ];
   +    }
   +}
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   sdk-php owns this P0 through source, release and applicable published producer/SDK verification. I read maintained SDK main c87eae373325d0a5bc8529058984becf17c82793, the Server1078 acceptance, and the sanitized SDK reference as data only. Its 39,147 bytes / SHA-256 37d05531ec36a7a9f90356018faae6bd6ce8e4f47d6e831caec38a7348308327 match the filed reference. No patch was applied or executed, and no consumer implementation or runtime result is claimed.

   The finite current-source review identifies these required implementation boundaries:

   1. Explicit default OFF before any speculative work. Current Worker constructor/create have no continuation option. The reference adds capability-derived predicates but supplies no new Worker opt-in/default-off evidence. The eventual implementation must gate preparation, retained history and additional replay as well as request emission. A constructor/create option can be appended without disturbing existing positional/named calls; bridge factories currently use Worker::create and must retain OFF unless explicitly configured. Capability advertisement alone cannot enable it. Exact option naming is SDK-owned; no field spelling is inferred for the producer.
   2. Refuse unsupported effects before execution. Current Replayer resolves a new record_side_effect callback before returning commands (Replayer.php/WorkflowCommand.php); the reference filters the returned command list afterward. It blocks local callbacks but that is not a general speculative execution boundary. An eligibility/command-admission mode must refuse unsupported callback work before it runs, preserve the already-computed remote activity result and leave ordinary replay available. Allowed returned commands alone also do not prove the admitted prefix lacks the excluded group/selection/local/cancellation/update/message state. ReplayResult exposes several prepared-local/scope/cancellation intents, but the complete producer-defined eligibility trace/prefix checks remain required.
   3. A proven activity commit is not an array continuation reply. Maintained retryStorageAdmission intentionally returns null after an uncertain activity completion is proven committed by exact task/attempt/owner/status evidence. The reference routes its completion into continued(..., array $completion). A port must explicitly handle the maintained null/commit-proof case rather than letting it become an application activity failure or rerun. The proof establishes the activity result, not whether proposed workflow commands committed. Server1078 must define continuation receipt/recovery for that lost-response case; it cannot be relabeled an ordinary eligibility decline.
   4. History/result/time must be producer authority. The reference constructs ActivityCompleted by copying ActivityStarted and changing sequence/type/result. Current replay includes a business clock, so copied start metadata is not evidence of the actual terminal/entry timestamp or canonical codec/result identity. Server must define the exact correlated prefix/tail and canonical result/event recipe, including clock-dependent replay; SDK must not guess a terminal event or fetch an unbounded replacement history under a closed prior workflow claim. Event/byte/command/external-payload limits and unsupported history states need closed vectors.
   5. Keep authority and publication paths intact. Preserve original activity/run/execution/attempt/owner/registration fences, session/business/lease deadlines and monotonic completion-request transit. Use the accepted ordinary response/lease path for any eager grant. Keep tick/shutdown/telemetry, ordinary completion retry, computed-result retention, duplicate/lost replies and old-Server fallback explicit. Only a documented decline may take the ordinary path; backend/decoder/budget/invariant and unknown-commit outcomes remain truthful.

   Next action / owners: I requested Server's versioned accepted+advertised #1078 request/reply/capability, canonical result/history/clock, command/eligibility, decline/error/recovery, capacity and history-tail budget checkpoint, coordinated with #1060/#1061/#1062/#1067. Server owns producer implementation and vectors; sdk-php owns its own default-off consumer and real worker base/candidate controls once that handoff exists. No published producer/SDK tuple is accepted yet. The immediate source task is this pre-execution eligibility/default-off/result-retention review, not applying the cumulative reference or inventing a producer shape. #340 Source/effect and #339/#337 maintenance-quality/release holds remain separate.
4. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Intake is filed as P0, owner sdk-php. Shared index, allocation and static review findings.

   Complete sanitized diff reference(s): 0001 part 1/1.

   Shared Lockbox bundle: lab4/general/dw-performance-intake/2026-11-06/sanitized-tricks-20-25.zip (101,306 bytes; SHA-256 3e5b593ed89536451be1c8d74abf2ee6e4782cd3f7a75fe5fc9a645d648a52da). Its roundtrip and member hashes are verified. The supplied patches and benchmark results remain external reference data; no patch was applied or executed during intake. Existing issue acceptance and subsequent owner review remain in force.
5. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   I reread Server1078 and the completed intake: no accepted advertised continuation wire/result/history/closed-decline/budget contract has been returned. The five concrete SDK findings at6697722902 and producer request6697723363 remain the source plan, including pre-execution eligibility, explicit default-OFF Worker option, canonical completion authority and unknown-commit handling. Actual3.9.5 delivers only separate helper/grant changes and enables no continuation. Next owner/action: Server supplies the versioned producer contract and truthful error disposition; sdk-php then owns the consumer and causal old-Server/default-off/fence/deadline/replay controls. No Server implementation or reference patch was executed.
6. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Handoff audit after the separate #328–#332 source cohort: Server #1078 still has no accepted, advertised continuation wire/result/history/closed-decline/budget handoff. Its broad error-softening finding remains material; the carried-poll and pipeline changes do not implement continuation. I re-read the primary producer record and retained the five SDK findings.

   Next owner/action: Server returns that versioned contract and truthful unknown-commit/error authority; General retains producer ordering. sdk-php then implements pre-execution eligibility and an explicit default-OFF Worker option, with old-Server, lease/fence/deadline/transit and replay controls. The producer owner has been asked; no reference patch or Server implementation was run.
7. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Recheck 2026-11-07 19:40 UTC: server#1078 still has no accepted, advertised continuation contract (last activity 2026-11-06). Nothing in the 3.10 cohort implements continuation.

   Next owner / action: Server returns the versioned wire/result/history/decline/budget contract and truthful unknown-commit error handling. sdk-php then adds the default-OFF Worker option, pre-execution eligibility, and old-Server/lease/replay controls. I asked Server in chat today for a date or the specific blocker.
8. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Server posted contract draft v0 on server#1078. My consumer-fit review is server#1078/6712366270: D1 (a), include version markers (D2), no eager dispatch (D3), and one blocking gap (F1: the SDK can't learn the events its own workflow-task completion recorded). Next: Server answers F1–F6, then General selects D1–D3. No SDK code until the contract is accepted and advertised.
9. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Server accepted all six consumer findings in v1 amendments:

   • F1: history tail plus last sequence on the workflow-task completion;
   • F2: verifiable accept reply;
   • F3: closed recovery receipt;
   • F5: empty commands are refused;
   • F6: the capacity rule counts this worker only.

   On F4, faults and unknown commits share the same 503; re-sending the identical completion is safe for both, which works for the SDK. A distinct rollback code is listed as optional D4. Next: Security review, then General selects D1–D4. sdk-php starts implementation only once the contract is accepted and advertised.
10. rmcdaniel commented on Nov 8, 2026

    

    rmcdaniel

    on Nov 8, 2026

    MemberAuthor

    More actions

    Operator nudge: this benchmax ticket (P0) has had no linked PR and no activity for about 8 h. It is the SDK side of server#1078 (opt-in workflow continuation with an activity completion): please coordinate with that server ticket and start a draft PR as soon as the shape is agreed. If it is blocked on the server side, add the status:blocked label with a one-line reason.
11. rmcdaniel

    added

    agent:general

    and removed

    agent:sdk-phpWork done by the sdk-php repo agentWork done by the sdk-php repo agent

    on Nov 9, 2026
12. rmcdaniel commented on Nov 9, 2026

    

    rmcdaniel

    on Nov 9, 2026

    MemberAuthor

    More actions

    Opt-in consumer merged in #373 (14:46 UTC): the Worker option enableActivityContinuation defaults off and is used only when the server advertises activity_workflow_continuation v1. The producer (durable-workflow/server#1116) is in the merge queue. Validated end to end against the producer branch (9/9 accepted, results correct, full-history replay clean). Still open: a Native release carrying the capability, an SDK release, installed producer/consumer qualification of the published pair, and a measured A/B before any speedup claim.
13. rmcdaniel commented on Nov 9, 2026

    

    rmcdaniel

    on Nov 9, 2026

    MemberAuthor

    More actions

    General is holding the next continuation SDK/Native publication while repairing the producer atomic-transition and exact retry-receipt gaps found in the release audit: durable-workflow/server#1078 . The merged opt-in consumer and its prior source checks remain source evidence only. Next: producer repair and three-backend HTTP controls, then the matching published pair and installed qualification; no default enablement or speedup claim.