Add bounded activity lease recovery and runtime renewal

docs/api-stability.md

@@ -261,6 +261,46 @@ application heartbeat. Application heartbeat records progress and updates
 only its heartbeat timeout. Neither can move the original total or cleanup
 deadline, revive expired authority or establish physical callback stop.
 
+## Activity lease renewal
+
+`Workflow\V2\Contracts\ActivityLeaseRenewal` is an optional runtime integration
+role. Check the actual bound activity bridge before using `renewLease()`;
+existing custom `ActivityTaskBridge` implementations gain no required method.
+The caller must supply the original attempt ID and lease owner. Successful
+receipts preserve those identities and include `lease_renewed`, `server_time`
+and the current `lease_expires_at`. A refusal does not renew ownership, restore
+an expired attempt, or acknowledge that its callback stopped.
+
+Runtime renewal changes only the original attempt and task lease expiry. It
+does not add history, update application heartbeat progress, or advance the
+authored heartbeat, start-to-close or schedule-to-close deadlines. Read-only
+status remains read-only. Adapters must preserve their original run, namespace,
+task, owner and attempt-generation authority checks before invoking the role.
+This Core role alone does not advertise a service-worker capability.
+
+`ActivityLease::seconds()` resolves `workflows.v2.activity_task_lease_seconds`,
+published as `DW_V2_ACTIVITY_TASK_LEASE_SECONDS` for embedded applications.
+Integer values from 1 through 86400 seconds are accepted; invalid, fractional,
+boolean and out-of-range values fall back to the existing 300-second default.
+The matching-role snapshot reports the effective value. Runtime renewal allows
+a longer activity to retain a short worker-loss recovery interval without
+requiring business progress heartbeats. Actual worker integration is required
+before an adapter can claim automatic renewal.
+
+An adapter holding the original worker-registration incarnation fence may call
+`ActivityTimeoutEnforcer::revokeWorkerLease(taskId, namespace, leaseOwner)` at
+deregistration. The runtime rechecks the original task, attempt generation,
+owner and namespace under canonical locks. It settles that attempt through its
+recorded retry/backoff policy instead of silently redispatching it. Contested
+locks fail without committing the registration receipt or activity settlement.
+
+An explicit revocation before the grant expires records
+`failure_reason=worker_deregistered`, the original `lease_expires_at` and optional
+`lease_revoked_at` in `ActivityRetryScheduled` or `ActivityFailed`. An already
+expired lease retains `lease_expired`; an elapsed business timeout keeps its
+own timeout outcome. These fields do not prove physical callback stop. SDK
+shutdown must retain its physical callback ownership and late-result fences.
+
 ## Workflow service operation caller API
 
 PHP workflow code can initiate a durable Nexus service operation from inside
@@ -591,9 +631,9 @@ class, source file path, or stack trace to replay or handle a failure.
 | `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` |
 | `ActivityHeartbeatRecorded` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `heartbeat_at`, `lease_expires_at`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `activity_attempt`, `progress` | `ActivitySnapshot`, `ActivityAttemptSnapshots`, `HistoryTimeline`, `RunActivityView` |
-| `ActivityRetryScheduled` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `retry_task_id`, `retry_of_task_id`, `retry_available_at`, `retry_backoff_seconds`, `retry_after_attempt_id`, `retry_after_attempt`, `retry_reason`, `max_attempts`, `retry_policy`, `timeout_kind`, `execution_mode`, `local_activity`, `workflow_task_id`, `exception_type`, `exception_class`, `message`, `code`, `exception`, `activity`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `ActivitySnapshot`, `ActivityAttemptSnapshots`, `HistoryTimeline`, `RunTaskView`, `ParallelChildGroup` |
+| `ActivityRetryScheduled` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `retry_task_id`, `retry_of_task_id`, `retry_available_at`, `retry_backoff_seconds`, `retry_after_attempt_id`, `retry_after_attempt`, `retry_reason`, `max_attempts`, `retry_policy`, `timeout_kind`, `failure_reason`, `lease_expires_at`, `lease_revoked_at`, `lease_owner`, `execution_mode`, `local_activity`, `workflow_task_id`, `exception_type`, `exception_class`, `message`, `code`, `exception`, `activity`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `ActivitySnapshot`, `ActivityAttemptSnapshots`, `HistoryTimeline`, `RunTaskView`, `ParallelChildGroup` |
 | `ActivityCompleted` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `result`, `payload_codec`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `activity_attempt`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `WorkflowExecutor`, `QueryStateReplayer`, `ParallelChildGroup`, `ActivityRecovery` |
-| `ActivityFailed` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `failure_id`, `failure_category`, `non_retryable`, `exception_type`, `exception_class`, `message`, `code`, `exception`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `activity_attempt`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path`, `structural_limit_kind`, `structural_limit_value`, `structural_limit_configured` | `ActivitySnapshot`, `FailureSnapshots`, `HistoryTimeline`, `ParallelFailureSelector`, `ParallelChildGroup` |
+| `ActivityFailed` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `failure_id`, `failure_category`, `non_retryable`, `exception_type`, `exception_class`, `message`, `code`, `exception`, `failure_reason`, `lease_expires_at`, `lease_revoked_at`, `lease_owner`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `activity_attempt`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path`, `structural_limit_kind`, `structural_limit_value`, `structural_limit_configured` | `ActivitySnapshot`, `FailureSnapshots`, `HistoryTimeline`, `ParallelFailureSelector`, `ParallelChildGroup` |
 | `ActivityCancelled` | `workflow_command_id`, `activity_execution_id`, `activity_attempt_id`, `worker_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `cancelled_at`, `execution_mode`, `local_activity`, `workflow_task_id`, `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`, cancellation repair projections |
 | `ActivityTimedOut` | `activity_execution_id`, `activity_attempt_id`, `activity_class`, `activity_type`, `sequence`, `attempt_number`, `failure_id`, `failure_category`, `timeout_kind`, `message`, `exception_class`, `schedule_deadline_at`, `close_deadline_at`, `schedule_to_close_deadline_at`, `heartbeat_deadline_at`, `execution_mode`, `local_activity`, `workflow_task_id`, `activity`, `activity_attempt`, `parallel_group_id`, `parallel_group_kind`, `parallel_group_base_sequence`, `parallel_group_size`, `parallel_group_index`, `parallel_group_path` | `ActivitySnapshot`, `FailureSnapshots`, `HistoryTimeline`, `ParallelChildGroup`, timeout repair projections |
 | `TimerScheduled` | `timer_id`, `sequence`, `delay_seconds`, `fire_at`, `timer_kind`, `condition_wait_id`, `condition_key`, `condition_definition_fingerprint`, `signal_wait_id`, `signal_name` | `WorkflowStepHistory`, `QueryStateReplayer`, `RunTimerView`, `ConditionWaits`, `SignalWaits` |

src/V2/Activity.php

@@ -147,7 +147,7 @@ public function heartbeat(array $progress = []): void
             }
 
             $expired = $task instanceof WorkflowTask
-                ? ActivityAttemptAuthority::expiredReason($task, $execution, now()) !== null
+                ? ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run) !== null
                 : ActivityTimeoutEnforcer::hasExpiredDeadline($execution, now());
             if ($expired) {
                 return;

src/V2/Contracts/ActivityLeaseRenewal.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Contracts;
+
+/**
+ * Optional runtime liveness, independent of application progress heartbeats.
+ *
+ * Adapters must discover this contract rather than assuming that every custom
+ * ActivityTaskBridge implementation supports lease renewal.
+ */
+interface ActivityLeaseRenewal
+{
+    /**
+     * Renew only the live original owner/attempt, without advancing business
+     * deadlines or adding history. An expired lease cannot be revived.
+     *
+     * A successful acknowledgement includes the original task, run, execution,
+     * attempt, attempt number and lease owner. server_time and lease_expires_at
+     * use UTC ISO 8601 with six fractional digits. Consumers capture their
+     * monotonic request clock before calling so response latency consumes the
+     * acknowledged lease window.
+     *
+     * last_heartbeat_at and heartbeat_deadline_at report stored application
+     * progress. Renewal does not modify either field. A consumer may observe a
+     * newer committed heartbeat within its original live claim and recorded
+     * heartbeat timeout, while preserving all other business deadlines.
+     *
+     * @return array<string, mixed>
+     */
+    public function renewLease(string $attemptId, string $leaseOwner): array;
+}

src/V2/Exceptions/ActivityLeaseExpiredException.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Exceptions;
+
+use RuntimeException;
+
+/** The original activity attempt lost its task lease without settling. */
+final class ActivityLeaseExpiredException extends RuntimeException
+{
+}

src/V2/Support/ActivityAttemptAuthority.php

@@ -6,6 +6,7 @@
 
 use Carbon\CarbonInterface;
 use Workflow\V2\Models\ActivityExecution;
+use Workflow\V2\Models\WorkflowRun;
 use Workflow\V2\Models\WorkflowTask;
 
 final class ActivityAttemptAuthority
@@ -14,13 +15,35 @@ public static function expiredReason(
         WorkflowTask $task,
         ActivityExecution $execution,
         CarbonInterface $at,
+        ?WorkflowRun $run = null,
     ): ?string {
         if ($task->lease_expires_at === null || $at->gte($task->lease_expires_at)) {
             return 'lease_expired';
         }
 
+        return self::expiredBusinessReason($execution, $at, $run);
+    }
+
+    private static function expiredBusinessReason(
+        ActivityExecution $execution,
+        CarbonInterface $at,
+        ?WorkflowRun $run,
+    ): ?string {
         $kind = ActivityTimeoutEnforcer::expiredDeadlineKind($execution, $at);
 
-        return $kind === null ? null : $kind . '_timeout';
+        if ($kind !== null) {
+            return $kind . '_timeout';
+        }
+
+        foreach ([
+            'run_timeout' => $run?->run_deadline_at,
+            'execution_timeout' => $run?->execution_deadline_at,
+        ] as $reason => $deadline) {
+            if ($deadline !== null && $at->gte($deadline)) {
+                return $reason;
+            }
+        }
+
+        return null;
     }
 }

src/V2/Support/ActivityAttemptSnapshots.php

@@ -61,6 +61,22 @@ public static function forRun(WorkflowRun $run, bool $useDurableHistory = true):
                 $states[$execution->id][$attempt->id] = isset($states[$execution->id][$attempt->id])
                     ? self::mergeMissing($states[$execution->id][$attempt->id], $snapshot)
                     : $snapshot;
+
+                $state = $states[$execution->id][$attempt->id];
+                if ($execution->status === \Workflow\V2\Enums\ActivityStatus::Running
+                    && $execution->current_attempt_id === $attempt->id
+                    && $attempt->status === ActivityAttemptStatus::Running
+                    && ($state['status'] ?? null) === ActivityAttemptStatus::Running->value
+                    && ($state['workflow_task_id'] ?? null) === $attempt->workflow_task_id
+                    && ($state['attempt_number'] ?? null) === $attempt->attempt_number
+                    && ($state['lease_owner'] ?? null) === $attempt->lease_owner) {
+                    // Lease liveness can change without a business history event.
+                    // Only the same still-open original attempt may supply it;
+                    // durable terminal outcomes retain their history authority.
+                    $states[$execution->id][$attempt->id]['lease_expires_at'] = self::timestamp(
+                        $attempt->lease_expires_at
+                    );
+                }
             }
         }
 

src/V2/Support/ActivityLease.php

@@ -5,15 +5,35 @@
 namespace Workflow\V2\Support;
 
 use Carbon\CarbonInterface;
+use Illuminate\Container\Container;
 
 final class ActivityLease
 {
     public const DURATION_MINUTES = 5;
 
     public const DURATION_SECONDS = self::DURATION_MINUTES * 60;
 
+    public const CONFIG_KEY = 'workflows.v2.activity_task_lease_seconds';
+
+    public const MAX_SECONDS = 86_400;
+
+    public static function seconds(): int
+    {
+        $value = Container::getInstance()->bound('config')
+            ? config(self::CONFIG_KEY, self::DURATION_SECONDS)
+            : self::DURATION_SECONDS;
+        $seconds = is_int($value) || is_string($value) ? filter_var($value, FILTER_VALIDATE_INT, [
+            'options' => [
+                'min_range' => 1,
+                'max_range' => self::MAX_SECONDS,
+            ],
+        ]) : false;
+
+        return $seconds === false ? self::DURATION_SECONDS : $seconds;
+    }
+
     public static function expiresAt(): CarbonInterface
     {
-        return now()->addMinutes(self::DURATION_MINUTES);
+        return now()->addSeconds(self::seconds());
     }
 }

src/V2/Support/ActivityOutcomeRecorder.php

@@ -118,7 +118,7 @@ public static function record(
                 return self::ignored('activity_deadline_elapsed');
             }
 
-            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now());
+            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run);
             if ($expiredReason !== null) {
                 return self::ignored($expiredReason);
             }
@@ -168,7 +168,7 @@ public static function record(
             }
 
             // Payload preparation can outlive a claim even while its rows stay locked.
-            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now());
+            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run);
             if ($expiredReason !== null) {
                 return self::ignored($expiredReason);
             }
@@ -210,7 +210,7 @@ public static function record(
             $parallelMetadata = ParallelChildGroup::payloadForPath($parallelMetadataPath);
             $resolutionEvent = null;
 
-            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now());
+            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run);
             if ($expiredReason !== null) {
                 return self::ignored($expiredReason);
             }

src/V2/Support/ActivityRowLockOrder.php

@@ -24,24 +24,32 @@ final class ActivityRowLockOrder
      *     snapshot_attempt_id: string|null
      * }
      */
-    public static function lockForExecution(string $executionId, bool $includeClosedAttempt = false): array
-    {
+    public static function lockForExecution(
+        string $executionId,
+        bool $includeClosedAttempt = false,
+        bool $waitForLocks = true
+    ): array {
         /** @var ActivityExecution|null $snapshot */
         $snapshot = ActivityExecution::query()->find($executionId);
-        $snapshotAttemptId = $snapshot instanceof ActivityExecution
-            ? ($includeClosedAttempt ? $snapshot->current_attempt_id : self::runningAttemptId($snapshot))
-            : null;
+        $snapshotAttemptId = null;
+        if ($snapshot instanceof ActivityExecution) {
+            $snapshotAttemptId = $includeClosedAttempt ? $snapshot->current_attempt_id : self::runningAttemptId(
+                $snapshot
+            );
+        }
+
+        $lockMode = $waitForLocks ? true : 'for update nowait';
 
         /** @var ActivityAttempt|null $attempt */
         $attempt = $snapshotAttemptId === null
             ? null
             : ActivityAttempt::query()
-                ->lockForUpdate()
+                ->lock($lockMode)
                 ->find($snapshotAttemptId);
 
         /** @var ActivityExecution|null $execution */
         $execution = ActivityExecution::query()
-            ->lockForUpdate()
+            ->lock($lockMode)
             ->find($executionId);
 
         return [