Deadline-triggered timeout cleanup, embedded (#863)

.github/workflows/php.yml

@@ -174,7 +174,9 @@ jobs:
   regression-corpus:
     needs: preflight
     runs-on: ubuntu-latest
-    timeout-minutes: 6
+    # Each new fixture adds two full consumer runs after the baseline.
+    # Four fixtures already exceed six minutes on the hosted runner.
+    timeout-minutes: 12
     env:
       # The runner's system PHP loads Xdebug. Each official consumer run takes
       # several times longer with it, and the validator runs the consumer

CHANGELOG.md

@@ -2,6 +2,26 @@
 
 ## Unreleased
 
+- **Run migrations.** Opt-in deadline-triggered cleanup (#863, experimental,
+  embedded). A start can reserve the last part of its hard execution or run
+  timeout for cleanup with `StartOptions::withTimeoutCleanup($seconds)` (or
+  `timeoutCleanupSeconds:`). When the work deadline passes (the hard deadline
+  minus the allowance), the run enters timeout cleanup once: workflow code
+  receives `WorkflowTimeoutCleanupRequested`, a cooperative cancellation
+  request whose context has `cause: timeout`, so existing compensation
+  handlers run. An idle run is woken at that time. The run always ends with
+  `WorkflowTimedOut`, which now carries a `timeout_cleanup` outcome:
+  `completed`, `failed` (the cleanup failure is linked), `expired` (still
+  running at the hard deadline) or `preempted` (terminated or hard-cancelled
+  first). The hard deadline never moves, and nothing changes without the
+  option. Continue-as-new and redrive carry the allowance like the hard
+  deadlines. New history event `WorkflowTimeoutCleanupStarted`; the cancellation
+  context gains the optional top-level `cause`. The migration adds the nullable
+  `workflow_runs.timeout_cleanup_work_deadline_at` with no backfill and builds
+  its index without blocking writes. Poll-mode and service starts refuse this
+  option until their capability admission is implemented (server#1201). Bridge
+  workers cannot claim an opted-in embedded run. Service mode follows separately.
+
 ## 2.28.0 - 2026-11-13
 
 **Behavior change:** an un-awaited `select()` loser is cancelled when the run

docs/api-stability.md

@@ -765,9 +765,10 @@ class, source file path, or stack trace to replay or handle a failure.
 | --- | --- | --- |
 | `StartAccepted` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `workflow_class`, `workflow_type`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `outcome`, `rejection_reason` | `HistoryTimeline`, `HistoryExport`, `RunCommandContract`, operator command projections |
 | `StartRejected` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `workflow_class`, `workflow_type`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `outcome`, `rejection_reason` | `HistoryTimeline`, `HistoryExport`, `RunCommandContract`, operator command projections |
-| `WorkflowStarted` | `workflow_class`, `workflow_type`, `workflow_instance_id`, `workflow_run_id`, `workflow_command_id`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `execution_timeout_seconds`, `run_timeout_seconds`, `execution_deadline_at`, `run_deadline_at`, `workflow_definition_fingerprint`, `definition_changed_from`, `declared_queries`, `declared_query_contracts`, `declared_signals`, `declared_signal_contracts`, `declared_updates`, `declared_update_contracts`, `declared_entry_method`, `declared_entry_mode`, `declared_entry_declaring_class`, `parent_workflow_instance_id`, `parent_workflow_run_id`, `parent_sequence`, `workflow_link_id`, `child_call_id`, `retry_policy`, `timeout_policy`, `continued_from_run_id`, `retry_attempt`, `retry_of_child_workflow_run_id` | `WorkflowDefinitionFingerprint`, `RunLineageView`, worker history payload consumers |
+| `WorkflowStarted` | `workflow_class`, `workflow_type`, `workflow_instance_id`, `workflow_run_id`, `workflow_command_id`, `business_key`, `visibility_labels`, `memo`, `search_attributes`, `execution_timeout_seconds`, `run_timeout_seconds`, `execution_deadline_at`, `run_deadline_at`, `workflow_definition_fingerprint`, `definition_changed_from`, `declared_queries`, `declared_query_contracts`, `declared_signals`, `declared_signal_contracts`, `declared_updates`, `declared_update_contracts`, `declared_entry_method`, `declared_entry_mode`, `declared_entry_declaring_class`, `parent_workflow_instance_id`, `parent_workflow_run_id`, `parent_sequence`, `workflow_link_id`, `child_call_id`, `retry_policy`, `timeout_policy`, `continued_from_run_id`, `retry_attempt`, `retry_of_child_workflow_run_id`, `timeout_cleanup` (experimental: `allowance_seconds` and the derived `work_deadlines` per timeout kind) | `WorkflowDefinitionFingerprint`, `RunLineageView`, worker history payload consumers |
 | `WorkflowContinuedAsNew` | `sequence`, `continued_to_run_id`, `continued_to_run_number`, `workflow_link_id`, `closed_reason`, `abandoned_activities` | `WorkflowStepHistory`, `RunLineageView`, `HistoryTimeline`, operator detail projections |
 | `CooperativeCancellationRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason`, `cleanup_deadline_at` | workflow cancellation delivery, cleanup deadline enforcement, history export |
+| `WorkflowTimeoutCleanupStarted` | `workflow_command_id`, `timeout_kind` (`execution` or `run`), `work_deadline_at`, `cleanup_deadline_at`, `hard_deadline_at`, `detected_at` | `HistoryTimeline`, timeout cleanup outcome (experimental) |
 | `CooperativeCancellationDelivered` | `workflow_command_id`, `workflow_run_id`, `sequence`, `call_kind`, `sequence_span`, `operation_sequence`, `operation_sequence_span` | `WorkflowFiberRunner`, `WorkflowStepHistory`, cooperative task bridges, history export |
 | `ActivityScheduled` | `activity_execution_id`, `activity_class`, `activity_type`, `sequence`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `ActivityRecovery`, `ParallelChildGroup` |
 | `ActivityStarted` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `execution_mode`, `local_activity`, `workflow_task_id`, `lease_expires_at`, `activity`, `activity_attempt`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `ActivitySnapshot`, `ActivityAttemptSnapshots`, `HistoryTimeline`, `RunActivityView`, `ParallelChildGroup` |
@@ -809,12 +810,12 @@ class, source file path, or stack trace to replay or handle a failure.
 | `MemoUpserted` | `sequence`, `entries`, `merged`, `reused_from_run_id`, `reused_recorded_at` | `WorkflowStepHistory`, `WorkflowExecutor`, `QueryStateReplayer`, `WorkflowFiberRunner`, `HistoryTimeline`, run detail projections, history export, `RedriveReplayClock` |
 | `RepairRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `outcome`, `liveness_state`, `wait_kind`, `task_id`, `task_type` | `HistoryTimeline`, `RunCommandContract`, repair diagnostics, operator detail projections |
 | `CancelRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason` | `HistoryTimeline`, `RunCommandContract`, cancellation projections |
-| `WorkflowCancelled` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `cancellation_cleanup` | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, cancellation projections |
+| `WorkflowCancelled` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `cancellation_cleanup`, `timeout_cleanup` (experimental; the deadline-triggered cleanup outcome) | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, cancellation projections |
 | `TerminateRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `reason` | `HistoryTimeline`, `RunCommandContract`, termination projections |
-| `WorkflowTerminated` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason` | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, termination projections |
+| `WorkflowTerminated` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `failure_id`, `failure_category`, `closed_reason`, `exception_class`, `message`, `reason`, `timeout_cleanup` (experimental; the deadline-triggered cleanup outcome) | `HistoryTimeline`, `FailureSnapshots`, `ChildRunHistory`, termination projections |
 | `ArchiveRequested` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `command_type`, `outcome`, `reason` | `HistoryTimeline`, `RunCommandContract`, archive/export projections |
 | `WorkflowArchived` | `workflow_command_id`, `workflow_instance_id`, `workflow_run_id`, `archive_command_id`, `reason` | `HistoryTimeline`, archive/export projections, operator detail projections |
-| `WorkflowTimedOut` | `failure_id`, `timeout_kind`, `failure_category`, `message`, `exception_class`, `execution_deadline_at`, `run_deadline_at` | `FailureSnapshots`, `HistoryTimeline`, `ChildRunHistory`, timeout repair projections |
+| `WorkflowTimedOut` | `failure_id`, `timeout_kind`, `failure_category`, `message`, `exception_class`, `execution_deadline_at`, `run_deadline_at`, `timeout_cleanup` (experimental; the deadline-triggered cleanup outcome), `abandoned_activities`, `cancellation_cleanup` (preempted by the hard timeout) | `FailureSnapshots`, `HistoryTimeline`, `ChildRunHistory`, timeout repair projections |
 | `WorkflowCompleted` | `output`, `payload_codec`, `abandoned_activities` | `ChildRunHistory`, `HistoryTimeline`, history export, parent resume projections |
 | `WorkflowFailed` | `failure_id`, `source_kind`, `source_id`, `failure_category`, `non_retryable`, `exception_type`, `exception_class`, `message`, `exception`, `structural_limit_kind`, `structural_limit_value`, `structural_limit_configured`, `structural_limit_name`, `failed_step_sequence`, `failed_step_kind` (`activity`, `workflow_code` or `structural_limit`), `abandoned_activities` | `FailureSnapshots`, `HistoryTimeline`, `ChildRunHistory`, failure repair projections, `FailedRunRedrivePlan` |
 | `FailureHandled` | `failure_id`, `sequence`, `failure_category`, `source_kind`, `source_id`, `propagation_kind`, `exception_class`, `exception_type`, `message`, `handled` | `FailureSnapshots`, `HistoryTimeline`, `QueryStateReplayer`, operator detail projections |

docs/architecture/timeout-cleanup.md

@@ -46,9 +46,13 @@ Validation at admission, before any write:
   work deadline is in the future.
 
 The allowance is recorded in `WorkflowStarted` (`timeout_cleanup.allowance_seconds`,
-plus the derived `work_deadline_at` per timeout kind), which is immutable. There
-is no new `workflow_runs` column: the engine and native read the admitted values
-from `WorkflowStarted` (server review on [workflow pull request 1122](https://github.com/durable-workflow/workflow/pull/1122)). Describe and run detail show them.
+plus the derived `work_deadline_at` per timeout kind), which is immutable. Native
+reads the admitted values from `WorkflowStarted` and adds no column (server review
+on [workflow pull request 1122](https://github.com/durable-workflow/workflow/pull/1122)).
+Embedded also stores the earliest derived work deadline in
+`workflow_runs.timeout_cleanup_work_deadline_at`, so the watchdog can wake an idle
+run at that time; the column is nullable and set only for opted-in runs (General,
+2026-11-13). Describe and run detail show them.
 
 ### Inheritance
 
@@ -80,15 +84,20 @@ notices first) does the following in one transaction under the run lock:
 3. Delivers the interruption through the existing cooperative-cancellation
    path. Workflow code receives a `WorkflowTimeoutCleanupRequested` exception,
    which extends `WorkflowCancellationRequestedException`, with a
-   `CancellationContext` whose `cause` is `timeout` and whose `timeoutKind`,
-   `workDeadlineAt` and `cleanupDeadlineAt` are set. Existing
+   `CancellationContext` whose optional top-level `cause` is `timeout` and
+   whose existing `cleanup_deadline_at` is the cleanup deadline. The timeout
+   kind and work deadline are recorded on `WorkflowTimeoutCleanupStarted`, not
+   in the context. `cause` is never inherited by descendants, scoped or not,
+   and an explicit `cause: cancellation` round-trips unchanged. Existing
    `catch (WorkflowCancellationRequestedException)` compensation code
    therefore runs without changes. `cancellationShield()` scopes and the
    operation cancellation policies behave as they do for a cooperative
    cancellation request.
 
 A user cancellation request that arrives during timeout cleanup is recorded
-but doesn't restart cleanup or move its deadline. A timeout that arrives
+as an additional accepted RequestCancellation command referencing the original
+engine request. It does not create another interruption, restart cleanup or move
+its deadline. A timeout that arrives
 during an earlier cooperative cancellation doesn't enter timeout cleanup:
 the earlier request already owns cleanup, its deadline is capped by the hard
 deadline, and the final outcome is the timeout if the hard deadline wins.
@@ -133,6 +142,10 @@ with `cause: timeout`. After that, they show the recorded outcome.
   new event type, so readers that don't know it refuse it as an unknown event.
   Service-mode SDKs therefore need the capability below before Server sends
   it.
+- This embedded increment admits the option only with queue dispatch. Poll-mode
+  starts and service starts refuse timeout cleanup, and bridge claims of an
+  opted-in embedded run return `timeout_cleanup_not_supported` without taking
+  the lease. Service support follows in [server issue 1201](https://github.com/durable-workflow/server/issues/1201).
 - Service mode: Server advertises `timeout_cleanup` (`{supported: true,
   version: 1}`) only when it implements this contract, and a service worker
   advertises it only when its SDK delivers the typed timeout interruption.

scripts/ci/validate-regression-corpus.py

@@ -177,6 +177,7 @@
         "tests/Fixtures/V2/TestNexusWaitingCallerWorkflow.php",
         "tests/Fixtures/V2/TestNexusCancellationCleanupCallerWorkflow.php",
         "tests/Fixtures/V2/TestNexusDeadlineCallerWorkflow.php",
+        "tests/Fixtures/V2/TestTimeoutCleanupWorkflow.php",
         "tests/Support/NexusCallerDeliveryScenario.php",
         "tests/Unit/V2/ReplayRegressionCorpusTest.php",
     ),

src/V2/CancellationContext.php

@@ -15,6 +15,12 @@
 /** Immutable request metadata recorded in canonical workflow history. */
 final class CancellationContext
 {
+    /**
+     * Cause of a run's own interruption: `timeout` when deadline-triggered
+     * cleanup (#863) requested it. Absent means an ordinary cancellation.
+     */
+    public const CAUSE_TIMEOUT = 'timeout';
+
     /**
      * @var WeakMap<self, WeakReference<Fiber>>|null
      */
@@ -37,9 +43,15 @@ private function __construct(
         private readonly CarbonImmutable $cleanupDeadline,
         public readonly array $lineage,
         public readonly ?ScopedCancellationContext $scopeOrigin = null,
+        public readonly ?string $cause = null,
     ) {
     }
 
+    public function isTimeout(): bool
+    {
+        return $this->cause === self::CAUSE_TIMEOUT;
+    }
+
     /**
      * @param array<string, mixed> $snapshot
      */
@@ -100,6 +112,14 @@ public static function fromArray(array $snapshot): self
         $rootRunId = self::text($snapshot, 'root_workflow_run_id');
         $parentRequestId = $snapshot['parent_request_id'] ?? null;
         $reason = $snapshot['reason'] ?? null;
+        // Top-level and optional; absent or `cancellation` is an ordinary
+        // cancellation (#863).
+        $cause = $snapshot['cause'] ?? null;
+        if ($cause !== null && ! in_array($cause, ['cancellation', self::CAUSE_TIMEOUT], true)) {
+            throw new InvalidArgumentException('Cancellation cause must be cancellation or timeout.');
+        }
+        // An explicit `cancellation` is preserved, so a context round-trips
+        // with exactly its own keys.
         if (count(array_unique(array_column($normalizedLineage, 'request_id'))) !== count($normalizedLineage)
             || count(array_unique(array_column($normalizedLineage, 'workflow_run_id'))) !== count($normalizedLineage)) {
             throw new InvalidArgumentException('Cancellation lineage cannot contain a cycle.');
@@ -168,6 +188,7 @@ public static function fromArray(array $snapshot): self
             $deadline,
             $normalizedLineage,
             $scopeOrigin,
+            $cause,
         );
     }
 
@@ -195,6 +216,9 @@ public function forDescendant(string $requestId, string $workflowInstanceId, str
             );
         }
         $snapshot = $this->toArray();
+        // A descendant is cancelled because its ancestor was; it was not
+        // itself timed out.
+        unset($snapshot['cause']);
         $snapshot['request_id'] = $requestId;
         $snapshot['parent_request_id'] = $this->requestId;
         $snapshot['lineage'][] = [
@@ -271,6 +295,9 @@ public function toArray(): array
             'requested_at' => $this->originalRequestedAt->toISOString(),
             'cleanup_deadline_at' => $this->cleanupDeadline->toISOString(),
             'lineage' => $this->lineage,
+            ...($this->cause === null ? [] : [
+                'cause' => $this->cause,
+            ]),
             ...($this->scopeOrigin === null ? [] : [
                 'scope_origin' => $this->scopeOrigin->toArray(),
                 'scope_authority_deadline_at' => $this->cleanupDeadline->toISOString(),
@@ -313,6 +340,10 @@ private static function scopedDescendantSnapshot(
                 $lineage[] = $address;
             }
         }
+        // A descendant is cancelled because its ancestor was; it was not
+        // itself timed out (#863).
+        unset($root['cause']);
+
         return [
             ...$root,
             'schema' => 'durable-workflow.cancellation-context/v2',

src/V2/CommandContext.php

@@ -31,6 +31,26 @@ public static function phpApi(): self
         ]);
     }
 
+    /**
+     * A command the engine issues itself, such as entering timeout cleanup at
+     * a run's work deadline (#863).
+     *
+     * @internal Engine attribution, not an application command context.
+     */
+    public static function system(string $id): self
+    {
+        return new self('system', [
+            'caller' => [
+                'type' => 'system',
+                'label' => $id,
+            ],
+            'auth' => [
+                'status' => 'not_applicable',
+                'method' => 'none',
+            ],
+        ]);
+    }
+
     public static function controlPlane(): self
     {
         return new self('control_plane', [

src/V2/Enums/HistoryEventType.php

@@ -71,6 +71,8 @@ enum HistoryEventType: string
     case SearchAttributesUpserted = 'SearchAttributesUpserted';
     case MemoUpserted = 'MemoUpserted';
     case WorkflowTimedOut = 'WorkflowTimedOut';
+
+    case WorkflowTimeoutCleanupStarted = 'WorkflowTimeoutCleanupStarted';
     case WorkflowCompleted = 'WorkflowCompleted';
     case WorkflowFailed = 'WorkflowFailed';
     case ParentClosePolicyApplied = 'ParentClosePolicyApplied';

src/V2/Exceptions/WorkflowCancellationRequestedException.php

@@ -8,7 +8,7 @@
 use Throwable;
 use Workflow\V2\CancellationContext;
 
-final class WorkflowCancellationRequestedException extends Error
+class WorkflowCancellationRequestedException extends Error
 {
     public function __construct(
         string $message = 'Cooperative cancellation requested.',
@@ -18,4 +18,15 @@ public function __construct(
     ) {
         parent::__construct($message, $code, $previous);
     }
+
+    /**
+     * The exception for a delivered request: the timeout cleanup subtype when
+     * deadline-triggered cleanup (#863) requested it.
+     */
+    public static function forContext(?CancellationContext $cancellation): self
+    {
+        return $cancellation?->isTimeout() === true
+            ? new WorkflowTimeoutCleanupRequested(cancellation: $cancellation)
+            : new self(cancellation: $cancellation);
+    }
 }