Replay histories whose run close settled open activities (server#1193)

CHANGELOG.md

@@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Replay supports histories in which a run's own close settled the work it never awaited (server#1193, workflow#1117/#1121, Workflow 2.28.0). When a run completes, fails or continues as new, each still-open activity, such as a `select()` loser, gets an `ActivityCancelled` with `cancellation_reason: run_closed` just before the terminal event. Replay, query and update replay now skip that cancellation and leave the activity open, as the workflow saw it. It never resolves a selection member, never supplies replay time, and is never taken as a list group's first positional failure. Before this fix, a close's cancellation of a lower-index member could replace the failure the workflow handled. Replay also accepts the new `abandoned_activities` key on `WorkflowCompleted`, `WorkflowFailed` and `WorkflowContinuedAsNew`, which lists abandon-policy activities left running at close.
+- Activity workers treat a `run_closed` refusal as final. A completion refused with `recorded: false, reason: run_closed` is split on its `activity_status`, as in the PHP worker. When it is `completed` or `failed`, the activity was an abandon-policy activity that kept running after its run closed and its own outcome closed it: the task counts as completed and logs at info level. Otherwise the close cancelled the activity: its outcome is dropped, the task is reported as aborted (never completed), and a warning is logged. A failure or cancellation report refused with `run_closed` is not reported again and logs at info level. A claim refused with `run_closed` counts as an answered poll, with no failure backoff. A heartbeat refused with `run_closed` raises `ActivityLeaseAuthorityLost` and publishes no outcome, as before.
+
 ## [2.17.0] - 2026-11-13
 
 ### Added

src/durable_workflow/worker.py

@@ -166,6 +166,36 @@
 _MULTIPLEXED_WORKFLOW_POLL_ENDPOINT = "/worker/workflow-tasks/poll"
 
 _TERMINAL_WORKFLOW_STATUSES = {"completed", "failed", "terminated", "canceled", "cancelled"}
+
+# The typed refusal for late activity work on a run whose own code closed it
+# (server#1193, workflow#1117). The run settled the activity at close, or left
+# it running under the abandon policy, and nothing is ever appended after the
+# terminal event, so the refusal is final: never retried or reported again.
+_RUN_CLOSED_REFUSAL = "run_closed"
+
+
+def _run_closed_refusal(error: BaseException) -> bool:
+    """Whether a claim or outcome request was refused because its run closed."""
+    if isinstance(error, ActivityLeaseAuthorityLost):
+        return error.reason == _RUN_CLOSED_REFUSAL
+    return isinstance(error, ServerError) and error.reason() == _RUN_CLOSED_REFUSAL
+
+
+def _run_closed_completion_kept(error: BaseException) -> bool | None:
+    """Classify a completion refused with ``run_closed``, as the PHP worker does.
+
+    ``True``: an abandon-policy activity its run left running at close, whose
+    own outcome closed it (``activity_status`` completed or failed); nothing
+    is recorded on the closed run. ``False``: the close cancelled the activity
+    and its outcome was dropped. ``None``: not a ``run_closed`` refusal with
+    ``recorded: false``.
+    """
+    if not isinstance(error, ServerError) or error.status != 409 or error.reason() != _RUN_CLOSED_REFUSAL:
+        return None
+    body = error.body
+    if not isinstance(body, dict) or body.get("recorded") is not False:
+        return None
+    return body.get("activity_status") in ("completed", "failed")
 _QUERY_TASK_FINAL_REJECTION_REASONS = {
     "lease_expired",
     "query_task_not_found",
@@ -3359,7 +3389,7 @@
             log.warning("activity claim abandoned: %s", error)
             return "claim_aborted"
 
     async def _run_activity_task_body(self, task: dict[str, Any]) -> str:
         self._track_worker_session_from_task(task)
         task_id: str = task["task_id"]
         attempt_id: str = task.get("activity_attempt_id") or task.get("attempt_id", "")
@@ -3450,6 +3480,23 @@
                 **self._external_storage_completion_kwargs(),
             )
         except Exception as e:
+            kept = _run_closed_completion_kept(e)
+            if kept is True:
+                # The run left this activity running under the abandon policy:
+                # its outcome closed it, and nothing is appended to the run.
+                log.info(
+                    "activity task %s completed; its workflow run closed, so the outcome is not recorded on the run",
+                    task_id,
+                )
+                return "completed"
+            if kept is False:
+                # The run's close cancelled this activity: the outcome is
+                # dropped, and the refusal is final, never retried.
+                log.warning(
+                    "activity task %s outcome not recorded: its workflow run closed and cancelled the activity",
+                    task_id,
+                )
+                return "claim_aborted"
             log.warning("failed to complete activity task %s: %s", task_id, e)
             return "complete_error"
         return "completed"
@@ -3505,6 +3552,9 @@
                 task_id=task_id, activity_attempt_id=attempt_id, lease_owner=self.worker_id, **failure,
             )
         except Exception as error:
+            if _run_closed_refusal(error):
+                log.info("%s for activity task %s not recorded: its workflow run closed", context, task_id)
+                return
             log.warning("failed to report %s: %s", context, error)
 
     async def _invoke_remote_activity(self, task: dict[str, Any], activity_type: str,
@@ -4115,6 +4165,14 @@
                 if self._stop.is_set():
                     self._status.request_abandoned("activity_poll")
                     return
+                if _run_closed_refusal(e):
+                    # The claim's run closed before it started: the Server
+                    # settled that task, so this answered poll is no failure.
+                    log.info("activity claim refused because its workflow run closed: %s", e)
+                    failures.reset()
+                    self._status.request_succeeded("activity_poll")
+                    self._record_poll_metrics("activity", "empty", time.perf_counter() - poll_start)
+                    continue
                 delay = _poll_capacity_delay(e, "activity_task", self.task_queue)
                 self._record_poll_status("activity_poll", e, answered=delay is not None)
                 if _is_storage_admission_error(e):

src/durable_workflow/workflow.py

@@ -3955,6 +3955,23 @@ def _history_event_type(event: Mapping[str, Any]) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+# ``cancellation_reason`` of an ActivityCancelled that the run's own close
+# recorded (Workflow 2.28.0, workflow#1117 and server#1193). Completing,
+# failing or continuing as new cancels each activity the workflow never
+# awaited, such as a select() loser, in the closing transaction just before the
+# terminal event. The workflow had already moved past those operations, so the
+# cancellation is never an outcome replay can observe: the activity stays open.
+RUN_CLOSED_CANCELLATION_REASON = "run_closed"
+
+
+def _settled_at_close(event: Any) -> bool:
+    """Whether this is an activity cancellation recorded by its run's close."""
+    if not isinstance(event, Mapping) or _history_event_type(event) != "ActivityCancelled":
+        return False
+    payload = event.get("payload")
+    return isinstance(payload, Mapping) and payload.get("cancellation_reason") == RUN_CLOSED_CANCELLATION_REASON
+
+
 def _workflow_sequence(payload: Mapping[str, Any]) -> int | None:
     value = payload.get("sequence") or payload.get("workflow_sequence")
     if isinstance(value, int):
@@ -4865,7 +4882,10 @@ def _replay_state(
                 exception_type=type(exception).__name__,
             ) from exception
 
-    events = list(history_events)
+    # A run-close cancellation never resolves a select() member, never fails
+    # an all() group and never supplies replay time: the workflow had closed
+    # past it. Every history reader below sees the activity as still open.
+    events = [event for event in history_events if not _settled_at_close(event)]
     # A malformed delivery span refuses replay before any history parser can
     # read it more permissively.
     _assert_delivery_spans(events)