fix(flow): the heartbeat recovers a missed signal instead of rolling forever

lib/Service/Flow/FlowResumeState.php

@@ -207,24 +207,37 @@ public static function fromArray(mixed $stored): self {
 	/**
 	 * The storable form, or null when there is nothing worth storing.
 	 *
-	 * Only a SUSPENDED run has anywhere to continue from. A terminal one does
-	 * not, so keeping its slots would put a stale cursor in front of anyone
-	 * reading the run to find out what happened — and the dispatcher has already
-	 * cleared every node that returned, so anything still held belongs to a node
-	 * the run never came back to.
+	 * Kept for every run that can still advance, dropped only on a terminal
+	 * one. The first version of this rule said "only a SUSPENDED run has
+	 * anywhere to continue from", and that conflated NOT-SUSPENDED with
+	 * TERMINAL: a pass can end `queued` — an in-request advance whose sibling
+	 * still has enabled work, a claim refused on contention — while a node
+	 * parked in an EARLIER pass still holds live progress. Dropping the slots
+	 * there is how the heartbeat wedge happened: a user-task node lost the
+	 * uuid of the task it was waiting on, asked again on the next wake, and
+	 * the original task's completion could never address the node's slot
+	 * again — its signal was refused against the new slot's assignee, and the
+	 * run rolled its heartbeat forever.
+	 *
+	 * A terminal run still drops them: keeping its slots would put a stale
+	 * cursor in front of anyone reading the run to find out what happened —
+	 * and the dispatcher has already cleared every node that returned, so
+	 * anything still held belongs to a node the run never came back to.
 	 *
 	 * Lives here rather than in the run service because it is a question about
 	 * this value, not about persistence: the state knows when it is worth
 	 * keeping.
 	 *
-	 * @param boolean $suspended Whether the walk ended suspended.
+	 * @param boolean $live Whether the run can still advance (any non-terminal
+	 *                      status — suspended, queued, running).
 	 *
 	 * @return array<string, array<string, mixed>>|null The slots, or null to drop them.
 	 *
 	 * @spec openspec/specs/flow-engine/spec.md#requirement-a-node-must-be-able-to-resume-from-where-it-stopped
+	 * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot
 	 */
-	public function storableWhen(bool $suspended): ?array {
-		if ($suspended === false || $this->byNode === []) {
+	public function storableWhen(bool $live): ?array {
+		if ($live === false || $this->byNode === []) {
 			return null;
 		}
 

lib/Service/Flow/FlowRunService.php

@@ -907,7 +907,7 @@ public function execute(FlowRun $run, array $flow, object $subject, ?array $seed
 		$resuming = ($run->getStatus() === FlowRun::STATUS_SUSPENDED);
 		if ($resuming === true) {
 			// Stored items win on resume (below), but the subject's own fields
-			// on them are a trigger-time snapshot. {@see self::refreshSubjectItems()}
+			// on them are a trigger-time snapshot: see refreshSubjectItems().
 			$this->refreshSubjectItems(run: $run, subject: $subject);
 		}
 
@@ -1091,6 +1091,7 @@ private function failUnresolvableVersion(FlowRun $run): FlowRun {
 	 * @return FlowRun The updated run.
 	 *
 	 * @spec openspec/changes/or-flow-runs/specs/flow-runs/spec.md
+	 * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot
 	 */
 	private function persistResult(FlowRun $run, array $result): FlowRun {
 		$status = (string)($result['status'] ?? FlowRun::STATUS_FAILED);
@@ -1141,14 +1142,7 @@ private function persistResult(FlowRun $run, array $result): FlowRun {
 		// other node also reads from.
 		unset($context[FlowNodeResumeState::CONTEXT_KEY]);
 
-		$resumeState = ($context[FlowResumeState::CONTEXT_KEY] ?? null);
-		unset($context[FlowResumeState::CONTEXT_KEY]);
-		if ($resumeState instanceof FlowResumeState === true) {
-			$storable = $resumeState->storableWhen(suspended: ($status === FlowRun::STATUS_SUSPENDED));
-			if ($storable !== null) {
-				$context[FlowResumeState::CONTEXT_KEY] = $storable;
-			}
-		}
+		$this->keepResumeSlots(context: $context, status: $status);
 
 		// A signal is consumed by the walk it woke. Kept, it would still be
 		// sitting there the NEXT time this run suspends on a signal, and that
@@ -1186,6 +1180,48 @@ private function persistResult(FlowRun $run, array $result): FlowRun {
 		return $persisted;
 	}//end persistResult()
 
+	/**
+	 * Fold the walk's per-node resume slots back into the storable context.
+	 *
+	 * 🔴 LIVE, NOT SUSPENDED. The rule used to be "only a suspended run has
+	 * anywhere to continue from", which quietly conflated NOT-SUSPENDED with
+	 * TERMINAL. A pass legitimately ends `queued` while a node parked in an
+	 * EARLIER pass is still waiting: the in-request advance of one branch
+	 * finalises `queued` whenever a sibling has enabled work, and a claim
+	 * refused on contention does the same. Dropping the slots there costs a
+	 * task-waiting node the uuid of the task it is waiting on — and that loss
+	 * IS the heartbeat wedge. The node's next wake finds an empty slot, so
+	 * (correctly, by its own idempotency guard) it asks again; from then on
+	 * the ORIGINAL task's completion can never address the node's slot, its
+	 * signal is refused against the new slot's recorded assignee, and the run
+	 * re-suspends on its heartbeat forever while a duplicate task sits in
+	 * somebody's inbox.
+	 *
+	 * A terminal run still drops them, for the reason it always did: anything
+	 * still held belongs to a node the run never came back to.
+	 *
+	 * @param array<string, mixed> $context The context being persisted, modified in place.
+	 * @param string $status The status the walk ended in.
+	 *
+	 * @return void
+	 *
+	 * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot
+	 */
+	private function keepResumeSlots(array &$context, string $status): void {
+		$resumeState = ($context[FlowResumeState::CONTEXT_KEY] ?? null);
+		unset($context[FlowResumeState::CONTEXT_KEY]);
+
+		if ($resumeState instanceof FlowResumeState === false) {
+			return;
+		}
+
+		$storable = $resumeState->storableWhen(live: (in_array($status, FlowRun::TERMINAL, true) === false));
+		if ($storable !== null) {
+			$context[FlowResumeState::CONTEXT_KEY] = $storable;
+		}
+
+	}//end keepResumeSlots()
+
 	/**
 	 * The correlation key a suspended run can be addressed by, or null.
 	 *

lib/Service/Flow/FlowTaskBridge.php

@@ -186,6 +186,60 @@ public function record(string $uuid, string $action, ?string $actor, string $rea
 		$this->tasks->record(uuid: $uuid, action: $action, actor: $actor, reason: $reason);
 	}//end record()
 
+	/**
+	 * Record that a heartbeat, not the completion's signal, delivered a
+	 * terminal task's answer to its run.
+	 *
+	 * The heartbeat exists precisely to recover a missed wake — a completion
+	 * whose signal was refused (the assignee guard, a group that did not exist
+	 * yet) or lost. When it does recover one, the audit must say so: the
+	 * guarded signal seam records a refusal, and without this entry the trail
+	 * ends there, reading as though the answer never reached the run at all.
+	 * Attributed to whoever completed the task, because the fact being
+	 * recorded is THEIR answer arriving — late, by poll — not the cron job's.
+	 *
+	 * Best-effort by design: the recovery itself is the node applying the
+	 * outcome, and a failure to write the audit row must never turn a
+	 * recovered run back into a wedged one.
+	 *
+	 * @param Task $task The terminal task whose outcome the heartbeat applied.
+	 *
+	 * @return void
+	 *
+	 * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-recovered-delivery-is-recorded-on-the-tasks-audit
+	 */
+	public function recordHeartbeatRecovery(Task $task): void {
+		try {
+			$this->tasks->record(
+				uuid: (string)$task->getUuid(),
+				action: 'heartbeat-recovered',
+				actor: $task->getCompletedBy(),
+				reason: sprintf(
+					'The completion signal never reached run %s; the heartbeat re-read this task and applied its outcome.',
+					(string)$task->getRunUuid()
+				)
+			);
+
+			$this->logger->info(
+				message: '[FlowTaskBridge] Heartbeat recovered a missed completion signal',
+				context: [
+					'file' => __FILE__,
+					'line' => __LINE__,
+					'task' => (string)$task->getUuid(),
+					'run' => (string)$task->getRunUuid(),
+					'node' => (string)$task->getNodeId(),
+					'completedBy' => (string)($task->getCompletedBy() ?? ''),
+				]
+			);
+		} catch (Throwable $failure) {
+			$this->logger->warning(
+				message: '[FlowTaskBridge] Could not record a heartbeat recovery on task ' . $task->getUuid()
+					. '; the outcome itself was applied: ' . $failure->getMessage(),
+				context: ['file' => __FILE__, 'line' => __LINE__, 'run' => (string)$task->getRunUuid()]
+			);
+		}//end try
+	}//end recordHeartbeatRecovery()
+
 	/**
 	 * The task a node's resume slot points at, or null when it is gone.
 	 *

lib/Service/Flow/Nodes/PortalTaskNode.php

@@ -76,6 +76,7 @@
 use OCA\OpenRegister\Service\Flow\FlowItems;
 use OCA\OpenRegister\Service\Flow\FlowNodeResumeState;
 use OCA\OpenRegister\Service\Flow\FlowRunContext;
+use OCA\OpenRegister\Service\Flow\FlowRunService;
 use OCA\OpenRegister\Service\Flow\FlowSuspension;
 use OCA\OpenRegister\Service\Flow\FlowTaskBridge;
 use OCA\OpenRegister\Service\Flow\IFlowNode;
@@ -267,6 +268,7 @@ public function validateConfig(array $config): void {
 	 *                          the case names nobody, or a re-ask has no reason.
 	 *
 	 * @spec openspec/changes/flow-portal-task/specs/flow-portal-task/spec.md#requirement-a-portal-task-step-creates-one-external-task-and-suspends-the-run
+	 * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome
 	 */
 	public function execute(array $items, array $config, array $context): array {
 		if ($items === []) {
@@ -300,6 +302,14 @@ public function execute(array $items, array $config, array $context): array {
 		}
 
 		if ($resume->get(key: PortalTaskConfig::SLOT_PASSED_AT, default: null) === null) {
+			// A terminal read with no signal in hand means the completion's
+			// wake never arrived — refused or lost — and the heartbeat is what
+			// recovered it. Recorded before the outcome is applied, which is
+			// identical on both paths; the user-task node makes the same call.
+			if (array_key_exists(FlowRunService::SIGNAL_CONTEXT_KEY, $context) === false) {
+				$this->bridge->recordHeartbeatRecovery(task: $task);
+			}
+
 			// The first pass over a terminal task: the answer travels on. Marked
 			// ONCE, so the next firing of this node in this run is a re-entry.
 			$resume->set(key: PortalTaskConfig::SLOT_PASSED_AT, value: (new DateTime())->format('c'));

lib/Service/Flow/Nodes/UserTaskNode.php

@@ -65,6 +65,7 @@
 use OCA\OpenRegister\Service\Flow\FlowItems;
 use OCA\OpenRegister\Service\Flow\FlowNodeResumeState;
 use OCA\OpenRegister\Service\Flow\FlowRunContext;
+use OCA\OpenRegister\Service\Flow\FlowRunService;
 use OCA\OpenRegister\Service\Flow\FlowStop;
 use OCA\OpenRegister\Service\Flow\FlowSuspension;
 use OCA\OpenRegister\Service\Flow\FlowTaskBridge;
@@ -270,6 +271,7 @@ public function validateConfig(array $config): void {
 	 * @throws RuntimeException When the node has no resume slot, or its task is gone.
 	 *
 	 * @spec openspec/changes/flow-user-task-node/specs/flow-user-task-node/spec.md#requirement-a-user-task-step-creates-exactly-one-task-and-suspends-the-run
+	 * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome
 	 */
 	public function execute(array $items, array $config, array $context): array {
 		if ($items === []) {
@@ -309,6 +311,15 @@ public function execute(array $items, array $config, array $context): array {
 			throw $this->suspension(config: $config, items: $items);
 		}
 
+		// A terminal read with no signal in hand means the completion's wake
+		// never arrived — refused by the assignee guard, or lost — and the
+		// heartbeat is what recovered it. Recorded on the task's audit,
+		// attributed to its completer, BEFORE the outcome is applied below:
+		// the applying is identical on both paths, which is the contract.
+		if (array_key_exists(FlowRunService::SIGNAL_CONTEXT_KEY, $context) === false) {
+			$this->bridge->recordHeartbeatRecovery(task: $task);
+		}
+
 		$bag = FlowTaskBridge::outcomeBagFor(task: $task);
 
 		if ($bag['rejected'] === true && ($config['failOnReject'] ?? false) === true) {

openspec/changes/flow-heartbeat-recovery/design.md

@@ -0,0 +1,97 @@
+# Design: flow-heartbeat-recovery
+
+## Context
+
+Measured before deciding anything:
+
+- **The node re-read has always existed.** `UserTaskNode::execute()`
+  (`lib/Service/Flow/Nodes/UserTaskNode.php`) re-enters on every wake, reads
+  `taskOrNull()` and applies the outcome bag when the task is terminal.
+  `PortalTaskNode` mirrors it. The single-stream walk and the stream walk
+  both re-dispatch a parked node on resume (`FlowStreamWalk::begin()`:
+  "Suspended streams become eligible again").
+- **The slot is the node's only memory.** The task uuid lives in the node's
+  per-node resume slot (#3325's scoping) and nowhere else the node can
+  reach. Lose the slot and the node MUST create a new task — its own
+  idempotency guard reads the slot to decide.
+- **`persistResult()` was the only writer that lost it.**
+  `FlowResumeState::storableWhen(suspended:)` returned null for every
+  non-suspended pass end. `advanceStream()` finalising `queued` (sibling
+  work enabled) is a routine pass end for a run with parallel human
+  branches, and a claim refusal produces `queued` too.
+
+## Decisions
+
+### D-1: Fix the state, not the walk
+
+The heartbeat is made honest by making the state it reads durable, not by
+adding a recovery sweep to `FlowRunWorker`. A worker-side "re-read every
+suspended run's tasks" would be a second delivery mechanism with its own
+addressing rules, racing the node's own re-read. With the slot intact, the
+existing wake (`findDue()` → `advance()` → `execute()` → node re-entry) does
+everything the defect report asks: re-read, apply, advance.
+
+`storableWhen()` keeps slots for every status outside `FlowRun::TERMINAL`.
+Terminal runs still drop them, for the original reason: anything still held
+belongs to a node the run never came back to, and the dispatcher has already
+cleared every node that returned.
+
+### D-2: The recovery is audited on the task, attributed to the completer
+
+The guarded signal seam records a refusal
+(`FlowRunSignalService::auditRefusal()`); without a matching entry the trail
+ends there and reads as though the answer never reached the run. When a node
+reads its task terminal with NO signal in the walk's context, the wake was a
+heartbeat, not the completion's signal — `context['signal']` is set by
+`signal()` and survives into the woken walk, so its absence is the
+discriminator. The node then calls
+`FlowTaskBridge::recordHeartbeatRecovery()`, which appends a
+`heartbeat-recovered` audit entry on the task via `TaskService::record()`,
+actor = `completedBy` — the fact recorded is THAT PERSON's answer arriving
+late, not the cron job acting. Best-effort: a failure to write the entry is
+logged and swallowed, because it must never turn a recovered run back into a
+wedged one.
+
+### D-3: The symmetric cases need no new mechanism
+
+Stated explicitly, as the defect report asks:
+
+- **A task completed while the run was not yet suspended (the race).**
+  `signal()` returns null for a run that is `running` or `queued`, so the
+  completion's wake is lost. The run then parks with a non-null heartbeat
+  (`🔴 THE HEARTBEAT IS NEVER NULL`, UserTaskNode), and the next wake
+  re-reads the task — with the slot now durable, the race costs at most one
+  heartbeat period of latency. No pre-suspension re-check is added: the node
+  cannot read an answer before it has parked on the question, and the
+  heartbeat already bounds the wait.
+- **A task whose sequence concluded (`TaskSequenceService`).** A sequence
+  drives every per-task transition through `TaskService`'s verbs, so the
+  task named by the node's slot reaches its terminal state on the same row
+  the heartbeat re-reads. Terminality is a property of that row
+  (`isInTerminalState()`); the re-read covers sequence-concluded tasks with
+  no sequence-specific handling.
+
+### D-4: Wrong-slot isolation is preserved by construction
+
+Each node reads only the slot the dispatcher scoped to it
+(`FlowNodeResumeState`), so a heartbeat wake recovers exactly the nodes
+whose OWN tasks are terminal; a sibling parked on an open task re-suspends
+with its slot untouched. `FlowHeartbeatRecoveryTest::testOnlyTheNodeWhoseTaskEndedRecovers`
+pins it.
+
+## Risks
+
+- Keeping slots on `queued`/`running` stores per-node state a little longer
+  than before. That state is exactly what a parked node needs on its next
+  wake; nodes that returned were already cleared by the dispatcher, so no
+  stale cursor can leak into a later pass.
+- `recordHeartbeatRecovery()` re-dispatches `TaskTerminalEvent`, because
+  every `TaskService::record()` on a terminal task does. The re-entrancy is
+  closed by an existing guard rather than by a new one, and the chain is
+  short enough to state in full: the listener calls
+  `FlowTaskBridge::continueRun()`, which calls `FlowRunService::signal()`,
+  which returns null for any run that is not `suspended`. At the moment the
+  recovery is recorded the run row says `running` — `execute()` sets and
+  persists that before the walk begins — so the signal is refused and no
+  second walk starts. The recovery is written from inside that walk, and the
+  walk finishes normally.