Record bounded diagnostic history for rejected late activity outcomes

docs/activity-outcome-rejection-audit.md

@@ -0,0 +1,56 @@
+# Rejected activity outcome audit
+
+This is the proposed contract for the remaining history requirement in
+[Workflow issue 878](https://github.com/durable-workflow/workflow/issues/878).
+It is not a published capability. Existing completion and failure responses
+continue to decide whether the business outcome was accepted.
+
+## Durable fact
+
+Record `ActivityOutcomeRejected` once for the first refused late outcome of a
+recognized original attempt after it loses lease or deadline authority. Bind
+the record to its original run, activity, attempt, task, attempt number and
+lease owner. Include the attempted outcome (`completed` or `failed`), the
+server's refusal reason and the observation timestamp. The first record stays
+immutable. Repeated requests do not create additional records.
+
+Do not store the rejected result, failure message, stack trace, external object
+reference, authentication material or an unrestricted request snapshot.
+An already committed successful or failed outcome is not a new rejected
+outcome merely because its response was retried.
+
+## Authority and ordering
+
+- Authenticate and enforce namespace, task, attempt and original owner binding
+  before any audit write. Unrelated or unknown identities create no record.
+  Attempt identifiers require exact stored spelling even when the database
+  collation considers a case alias equal. A missing stored original lease owner
+  is unavailable evidence; do not manufacture an owner or append a fact without
+  that binding. These internal checks do not replace HTTP authentication.
+- Recheck lost authority under the existing storage transaction and locks.
+  Select the reason from stored state and original deadlines. A request cannot
+  nominate its own audit reason or make a valid claim stale.
+- Refuse lost authority before resolving or retaining an external payload.
+  Audit recording never accepts the result, renews a lease, reopens a run,
+  dispatches work or creates a success receipt. `recorded=false` retains its
+  existing business meaning.
+- Keep prior business history and progress timestamps unchanged. The diagnostic
+  record advances history sequence/count metadata only. It does not become a
+  replay step or manufacture business progress.
+- Keep retention boundaries. Do not recreate deleted or pruned history, and
+  serialize the append against pruning. Closed retained runs may acquire this
+  diagnostic fact without acquiring new execution authority.
+
+## Required qualification
+
+Prove original identity, one-record behavior across repeated and concurrent
+requests, first-record immutability, transaction rollback, no payload/provider
+access, namespace/owner refusals, pruning races and unchanged business state.
+Exercise active and closed runs, expired leases, declared retry exhaustion and
+authored deadlines on SQLite, MySQL and PostgreSQL. Replay the resulting history
+with the published PHP, Python and Rust readers and verify real Waterline
+visibility. Existing failed originals remain failed.
+
+No schema, SDK protocol floor or native/frozen adoption is implied. The event
+and payload contract must be coordinated before publication; a source draft or
+passing local test does not deliver this requirement.

src/V2/Models/WorkflowHistoryEvent.php

@@ -239,18 +239,19 @@ private static function recordEvent(
                 ]);
                 ExecutionTracing::observeOutcome($lockedRun, $taskModel, $event);
 
-                $lockedRun->forceFill([
+                $updates = [
                     'last_history_sequence' => $sequence,
-                    'last_progress_at' => $now,
-                ])->save();
+                ];
+                if ($eventType !== HistoryEventType::ActivityOutcomeRejected) {
+                    $updates['last_progress_at'] = $now;
+                }
+                $lockedRun->forceFill($updates)
+                    ->save();
 
                 PendingSignalCloser::closeForTerminalEvent($lockedRun, $event, $taskModel);
 
-                $run->forceFill([
-                    'last_history_sequence' => $lockedRun->last_history_sequence,
-                    'last_progress_at' => $lockedRun->last_progress_at,
-                ]);
-                $run->syncOriginalAttributes(['last_history_sequence', 'last_progress_at']);
+                $run->forceFill($updates);
+                $run->syncOriginalAttributes(array_keys($updates));
 
                 LinkedServiceCallCloser::closeForTerminalEvent($lockedRun, $event);
                 ServiceCallWorkflowDelivery::forAdmission($lockedRun, $event);

src/V2/Support/ActivityOutcomeRecorder.php

@@ -43,8 +43,15 @@
     ): array {
         $sqlite = (new ActivityAttempt())->getConnection()
             ->getDriverName() === 'sqlite';
+        $attemptedOutcome = $throwable === null ? 'completed' : 'failed';
         if (! $sqlite) {
-            $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline);
+            $refused = self::refusedExpiredAttempt(
+                $taskId,
+                $attemptId,
+                $attemptCount,
+                $attemptedOutcome,
+                $admissionDeadline,
+            );
             if ($refused !== null) {
                 return $refused;
             }
@@ -60,12 +67,19 @@
             $backoffSeconds,
             $codec,
             $sqlite,
+            $attemptedOutcome,
             $admissionDeadline,
         ): array {
             // Even an unlocked SQLite read can wait behind an exclusive writer.
             // Keep the refusal metadata behind the original bounded admission.
             if ($sqlite) {
-                $refused = self::refusedExpiredAttempt($taskId, $attemptId, $attemptCount, $admissionDeadline);
+                $refused = self::refusedExpiredAttempt(
+                    $taskId,
+                    $attemptId,
+                    $attemptCount,
+                    $attemptedOutcome,
+                    $admissionDeadline,
+                );
                 if ($refused !== null) {
                     return $refused;
                 }
@@ -143,6 +157,15 @@
                 || $lockedExecution->current_attempt_id !== $attemptId
             ) {
                 self::closeAttemptIfStale($run, $attemptId);
+                ActivityOutcomeRejection::record(
+                    $run,
+                    $lockedExecution,
+                    $task,
+                    $attemptId,
+                    $attemptCount,
+                    $attemptedOutcome,
+                    'stale_attempt',
+                );
 
                 return self::ignored('stale_attempt');
             }
@@ -151,9 +174,16 @@
                 return self::ignored('activity_deadline_elapsed');
             }
 
-            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run);
-            if ($expiredReason !== null) {
-                return self::ignored($expiredReason);
+            $refusal = self::expiredOutcome(
+                $run,
+                $lockedExecution,
+                $task,
+                $attemptId,
+                $attemptCount,
+                $attemptedOutcome
+            );
+            if ($refusal !== null) {
+                return $refusal;
             }
 
             $runCodec = is_string($run->payload_codec) && $run->payload_codec !== ''
@@ -201,9 +231,16 @@
             }
 
             // Payload preparation can outlive a claim even while its rows stay locked.
-            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run);
-            if ($expiredReason !== null) {
-                return self::ignored($expiredReason);
+            $refusal = self::expiredOutcome(
+                $run,
+                $lockedExecution,
+                $task,
+                $attemptId,
+                $attemptCount,
+                $attemptedOutcome
+            );
+            if ($refusal !== null) {
+                return $refusal;
             }
 
             if (in_array($run->status, [RunStatus::Completed, RunStatus::Failed], true)) {
@@ -243,9 +280,16 @@
             $parallelMetadata = ParallelChildGroup::payloadForPath($parallelMetadataPath);
             $resolutionEvent = null;
 
-            $expiredReason = ActivityAttemptAuthority::expiredReason($task, $lockedExecution, now(), $run);
-            if ($expiredReason !== null) {
-                return self::ignored($expiredReason);
+            $refusal = self::expiredOutcome(
+                $run,
+                $lockedExecution,
+                $task,
+                $attemptId,
+                $attemptCount,
+                $attemptedOutcome
+            );
+            if ($refusal !== null) {
+                return $refusal;
             }
 
             if ($throwable === null) {
@@ -551,6 +595,7 @@
         string $taskId,
         string $attemptId,
         int $attemptNumber,
+        string $attemptedOutcome,
         ?int $admissionDeadline,
     ): ?array {
         $original = ActivityAttempt::query()->find($attemptId);
@@ -567,8 +612,8 @@
         $connection = $original->getConnection();
         $runId = (string) $original->workflow_run_id;
 
-        return SqliteTransaction::run(static function () use ($original): ?array {
-            return self::refusedExpiredAttemptUnderRunLock($original);
+        return SqliteTransaction::run(static function () use ($original, $attemptedOutcome): ?array {
+            return self::refusedExpiredAttemptUnderRunLock($original, $attemptedOutcome);
         }, admission: static function () use ($connection, $runId): void {
             WorkflowRun::on($connection->getName())->whereKey($runId)->toBase()->update([
                 'id' => $runId,
@@ -579,8 +624,10 @@
     /**
      * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null
      */
-    private static function refusedExpiredAttemptUnderRunLock(ActivityAttempt $original): ?array
-    {
+    private static function refusedExpiredAttemptUnderRunLock(
+        ActivityAttempt $original,
+        string $attemptedOutcome,
+    ): ?array {
         $connection = $original->getConnectionName();
         $run = WorkflowRun::on($connection)->whereKey($original->workflow_run_id)->lockForUpdate()->first();
         if (! $run instanceof WorkflowRun || $run->details_pruned_at !== null) {
@@ -600,8 +647,23 @@
         }
 
         $reason = self::expiredAttemptReason($run, $execution, $task, $attempt);
+        if ($reason === null) {
+            return null;
+        }
+
+        // The audit append reads the attempt without a lock, so it adds no lock
+        // after this run lock (renewal locks the attempt before the run).
+        ActivityOutcomeRejection::record(
+            $run,
+            $execution,
+            $task,
+            $attempt->id,
+            $attempt->attempt_number,
+            $attemptedOutcome,
+            $reason,
+        );
 
-        return $reason === null ? null : self::ignored($reason);
+        return self::ignored($reason);
     }
 
     private static function expiredAttemptReason(
@@ -736,6 +798,35 @@
         ];
     }
 
+    /**
+     * @return array{recorded: bool, reason: string|null, next_task: WorkflowTask|null}|null
+     */
+    private static function expiredOutcome(
+        WorkflowRun $run,
+        ActivityExecution $execution,
+        WorkflowTask $task,
+        string $attemptId,
+        int $attemptCount,
+        string $attemptedOutcome,
+    ): ?array {
+        $reason = ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run);
+        if ($reason === null) {
+            return null;
+        }
+
+        ActivityOutcomeRejection::record(
+            $run,
+            $execution,
+            $task,
+            $attemptId,
+            $attemptCount,
+            $attemptedOutcome,
+            $reason
+        );
+
+        return self::ignored($reason);
+    }
+
     /**
      * @return array<string, mixed>
      */

src/V2/Support/ActivityOutcomeRejection.php

@@ -0,0 +1,96 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Support;
+
+use Workflow\V2\Enums\ActivityAttemptStatus;
+use Workflow\V2\Enums\HistoryEventType;
+use Workflow\V2\Models\ActivityAttempt;
+use Workflow\V2\Models\ActivityExecution;
+use Workflow\V2\Models\WorkflowHistoryEvent;
+use Workflow\V2\Models\WorkflowRun;
+use Workflow\V2\Models\WorkflowTask;
+
+final class ActivityOutcomeRejection
+{
+    /**
+     * Record only while holding the run lock. Every caller holds it, so audit
+     * appends and their duplicate check are serialized per run.
+     *
+     * The attempt row is read without a lock. Lease renewal locks the attempt
+     * before the run, so asking for the attempt lock here, after the run lock,
+     * could deadlock with a renewal. A renewal that has not committed cannot
+     * change what this refusal observed.
+     */
+    public static function record(
+        WorkflowRun $run,
+        ActivityExecution $execution,
+        WorkflowTask $task,
+        string $attemptId,
+        int $attemptNumber,
+        string $outcome,
+        string $reason,
+    ): ?WorkflowHistoryEvent {
+        if ($run->details_pruned_at !== null || ! in_array($outcome, ['completed', 'failed'], true)) {
+            return null;
+        }
+        $attempt = ActivityAttempt::query()->find($attemptId);
+        if (! $attempt instanceof ActivityAttempt
+            || $attempt->id !== $attemptId
+            || ! is_string($attempt->lease_owner)
+            || $attempt->lease_owner === ''
+            || $attempt->workflow_task_id !== $task->id
+            || $attempt->activity_execution_id !== $execution->id
+            || $attempt->workflow_run_id !== $run->id
+            || $attempt->attempt_number !== $attemptNumber
+            || $execution->workflow_run_id !== $run->id
+            || $task->workflow_run_id !== $run->id
+            || $task->namespace !== $run->namespace
+            || ($task->payload['activity_execution_id'] ?? null) !== $execution->id
+            || ! self::lostAuthority($attempt, $task, $execution, $run, $reason)) {
+            return null;
+        }
+
+        $existing = WorkflowHistoryEvent::query()->where('workflow_run_id', $run->id)
+            ->where('event_type', HistoryEventType::ActivityOutcomeRejected)
+            ->where('workflow_task_id', $task->id)
+            ->where('payload->activity_attempt_id', $attemptId)
+            ->first();
+        if ($existing instanceof WorkflowHistoryEvent) {
+            return $existing;
+        }
+
+        // Passing the stored ID, rather than a task model, deliberately omits
+        // the generic task snapshot and its unrestricted dispatch diagnostics.
+        $event = WorkflowHistoryEvent::record($run, HistoryEventType::ActivityOutcomeRejected, [
+            'activity_execution_id' => $execution->id,
+            'activity_attempt_id' => $attempt->id,
+            'attempt_number' => $attempt->attempt_number,
+            'lease_owner' => $attempt->lease_owner,
+            'outcome' => $outcome,
+            'reason' => $reason,
+            'observed_at' => now()
+                ->toJSON(),
+        ], (string) $task->id);
+        RunSummaryProjector::projectActivityOutcomeRejection($run, $event);
+        RunTimelineProjector::projectActivityOutcomeRejection($run, $event);
+
+        return $event;
+    }
+
+    private static function lostAuthority(
+        ActivityAttempt $attempt,
+        WorkflowTask $task,
+        ActivityExecution $execution,
+        WorkflowRun $run,
+        string $reason,
+    ): bool {
+        if ($reason === 'stale_attempt' && $attempt->status === ActivityAttemptStatus::Expired) {
+            return true;
+        }
+
+        return $attempt->status === ActivityAttemptStatus::Running
+            && ActivityAttemptAuthority::expiredReason($task, $execution, now(), $run) === $reason;
+    }
+}