fix(flow): resume reads the live subject, routed items reach their branch, portal rows carry case context, and one bad flow cannot silence an event (#3310)

* fix(flow): a resumed run reads the subject as it stands, not the trigger-time snapshot

A run's items begin as a snapshot of the subject, and resume replayed that
snapshot verbatim: the applicant supplied the missing description and
completed the task, and the re-check still read description: null off the
frozen item — the shipped re-ask loop could never succeed and the case
stranded. The live subject was already resolved on every resume path and
then used only as the marking holder.

Now both resume paths (worker execute() and the sync advanceStream())
refresh the subject's own fields on every stored item that IS the subject,
matched by identity, on the flat items list and the per-place buffers
alike. Live wins on the subject's keys; step-produced keys survive, so
flow-runs REQ-FR-003 (items survive the pause) still holds — the refresh
never re-seeds and never touches an item carrying another object's
identity.

Regression: FlowRunServiceTest::testAResumedRunBranchesOnTheLiveSubjectNotTheTriggerTimeSnapshot
and ::testAReAskLoopThatReceivesTheAnswerTerminates — both red before this
change (the loop one strands at the transition ceiling, the field symptom).

* fix(flow): a routing tag names an exit; resolve it to the exit's place before delivery

The two halves of per-item routing spoke different vocabularies. The route
node tags each item with the output its rule matched — an EXIT id ('no') —
while FlowItemPlacement::itemsForOutput() matches tags against PLACE names
('onRejected'). Unresolved, every routed item was discarded as the token
landed: the branch fired with zero items, SetFields iterated over nothing,
and the run persisted items: [], losing the seed and the earlier steps'
output. That is the sync task-completion repro (advance: "all") where
ask, route and onRejected all fired correctly and the items were empty
afterwards — and it would strand identically under the worker; the sync
path was just the only one an e2e drove through a router.

FlowTokenRouter::resolveOutputTags() now rewrites exit-id tags to the
exit's place at all three firing sites, before the exit choice and the
per-place delivery. takenExits() additionally honours the routing tag
among UNCONDITIONED exits: the tagged exit beats declaration order, so the
token follows the node's own routing decision instead of landing on the
first declared exit for the wrong reason. Conditioned exits (Switch) are
unchanged and still win.

Regression: FlowLogicTest::testAnItemTaggedWithAnExitIdLandsOnTheExitsPlace
and ::testTheTaggedExitBeatsDeclarationOrder (both red before), plus
FlowRunServiceAdvanceStreamTest::testTheSyncAdvanceCarriesTheStoredItemsInAndPersistsTheWalkItemsOut
pinning the sync persistence seam; tests/e2e/api-direct/flow-user-task.spec.ts:429
is the end-to-end regression on the next rig reset.

* fix(task): the inbox reads the subject's real serialised shape, so rows carry their case context

ObjectEntity::jsonSerialize() puts the object's own data at top level and
its identity under @self (uuid keyed id, mirrored as a top-level id); there
is no flat uuid/register/schema/name key. TaskInboxService::subjectContexts()
read exactly those flat keys, so the uuid check failed for every real object
and every inbox and portal row shipped subject: null — flow-portal-task's
seam contract says the row carries its case context, and
flow-portal-task.spec.ts:383 pinned the gap deterministically. The unit
test's hand-rolled stub returned the flat shape the entity never emits, so
it agreed with the exact bug it should have caught.

The context now reads @self first (uuid via @self.id / top-level id), with
the flat keys kept as fallbacks for stores that serialise flat. The stub is
replaced with a real ObjectEntity, and a portal-level test builds the inbox
WITH an object store — the portal unit tests used to omit it entirely,
which made subject structurally null in every one of them.

Regression: TaskInboxServiceTest::testSubjectContextIsBatchedForThePage
(now over a real ObjectEntity; red before) and
PortalTaskServiceTest::testAPortalRowCarriesItsCaseContext; the e2e at
flow-portal-task.spec.ts:383 goes green on the next rig reset.

* fix(flow): one unpublished flow no longer silences every flow on its event

Two defects with one blast radius. FlowTriggerService::fire() held its
try/catch OUTSIDE the per-flow loop, so the queue path's refusal of one
enabled-but-unpublished flow ('This flow has no published version') aborted
queuing for every healthy flow wired to the same event — the poisoned case
got no run from the flow that would have served it. The loop now isolates
per flow: a queue refusal logs against THAT flow and the rest queue; a sync
flow's inline-execution failure gets its own catch too, because by then the
run exists and the worker drains it (the count now says so — reporting zero
for a queued run was the instrument lying, and one execution-mode test pin
is updated accordingly).

And the delivery mechanism: FlowLocator's column fallback treated 'has no
index rows' as 'unconverted, columns decide', which is indistinguishable
from 'unpublished, subscribed to nothing'. flow-definition-versioning is
explicit — only a flow's published version contributes trigger records, a
draft's trigger nodes match nothing, and enabled is orthogonal to lifecycle
(so enabled:true pre-publish is neither indexed nor refused; the create and
enable paths already both index nothing pre-publish, which
FlowTriggerIndexPublishedTest pins). The fallback now refuses, out loud, a
flow with no published version; legacy flows all hold a backfilled
published v1, so the only thing filtered is a flow that could never have
run. Unreadable version rows fail open — the queue path still decides.

Regression: FlowTriggerServiceTest::testOneUnpublishedFlowDoesNotAbortTheFanOutForItsSiblings
and FlowLocatorTriggerCutoverTest::testAnUnpublishedFlowIsFilteredFromTheColumnFallback
(both red before this change).

* refactor(flow): keep the analyzers' thresholds after the trigger and routing fixes

The fixes pushed four methods and one class exactly onto their PHPMD
thresholds. Extract rather than suppress: the column-fallback loop moves to
FlowLocator::columnFallbackMatches() with the two-rule docblock, the
per-object context read moves to TaskInboxService::contextRow(), the
router's tag set and tag match get named helpers, and one over-long log
line wraps. Behaviour is pinned unchanged by the existing suites.

* test(flow): list FlowVersion in the uses rosters the locator seam now exercises

* fix(quality): wrap the second over-length log line and cover the new seams' branches

CI's phpcs names line 137 of FlowTriggerService (151 chars): the inline
failure message gained one indent when it got its own catch, after the
local phpcs pass. Wrapped like its sibling.

The pgsql cell's coverage guard measures the code this change keeps or
adds, and the fixes added statements whose failure branches no test
walked. Covered now, each in the suite that owns the seam: the
subject-projection refresh's identity fallbacks and no-ops
(FlowItemsRefreshTest), the per-place buffer rewrite on resume
(FlowRunServiceTest), the locator's fail-open on an unreadable version
table (FlowLocatorTriggerCutoverTest), the router's tag resolution
across multi-place, unroutable and already-resolved tags plus the
tag-over-declaration-order choice (FlowTokenRouterEndpointsTest), and
the inbox's flat-serialisation fallback (TaskInboxServiceTest).

* test(flow): count the trigger and locator branch coverage the strict cell was discarding

The coverage guard still read the fan-out isolation and fallback-filter
branches as uncovered: their pins live in files with docblock rosters, and
beStrictAboutCoverageMetadata marks a roster'd test risky the moment it
executes an unlisted class, discarding that test's WHOLE coverage. So the
cutover suite's refusal branch and the execution-mode suite's inline paths
counted for nothing (CI's coverage run shows Risky: 314).

Two-sided fix. The rosters are completed where they were short (Db\Flow on
the cutover suite; Db\FlowRun and FlowItems on the execution-mode suite),
and the branches are additionally exercised from roster-free suites, which
the cell always counts: the locator's unpublished-flow refusal and version-
table fail-open in FlowLocatorTest, the locator-failure and inline-failure
catches plus the subjectless payload seed and unresolvable-subject paths of
runInline in FlowTriggerServiceTest, the router's malformed/unwired exit
skips, and the getObject() subject shape in FlowItemsRefreshTest.

Author: rubenvdlinde

lib/Service/Flow/FlowEngine.php

@@ -490,6 +490,9 @@ public function run(
 					'durationMs' => 0,
 				];
 
+				// An exit-id tag becomes its place BEFORE routing, so the
+				// exit choice and the per-place delivery read the same name.
+				$items = $this->router()->resolveOutputTags(flow: $flow, transition: $transition, items: $items);
 				$taken = $this->router()->takenExits(flow: $flow, transition: $transition, items: $items, context: $context);
 				$placeItems = $this->placement()->advanceItems(
 					transition: $transition,
@@ -657,6 +660,12 @@ public function run(
 				$this->runContext?->pop();
 			}//end try
 
+			// A routing step tags items with the exit id its rule named;
+			// resolve those to the exit's PLACE first, so the choice below and
+			// the per-place delivery in advanceItems() read the same name —
+			// unresolved, a routed branch fired with zero items.
+			$items = $this->router()->resolveOutputTags(flow: $flow, transition: $transition, items: $items);
+
 			// Which single exit this firing takes. A token is unique and
 			// exclusive, so a branching node hands it to exactly one successor.
 			$taken = $this->router()->takenExits(flow: $flow, transition: $transition, items: $items, context: $context);
@@ -1075,6 +1084,10 @@ private function fireOnStream(
 		?string $streamError = null,
 	): array {
 		$name = $transition->getName();
+
+		// Same tag resolution as the single-stream walk: an exit-id tag
+		// becomes its place before the exit choice and the item delivery.
+		$items = $this->router()->resolveOutputTags(flow: $flow, transition: $transition, items: $items);
 		$taken = $this->router()->takenExits(flow: $flow, transition: $transition, items: $items, context: $context);
 		$placeItems = $this->placement()->advanceItems(transition: $transition, placeItems: $placeItems, items: $items, taken: $taken);
 		$workflow->apply(subject: $subject, transitionName: $name);

lib/Service/Flow/FlowItems.php

@@ -209,6 +209,93 @@ public static function fromSubject(object $subject): array {
 		return [self::item(json: $json)];
 	}//end fromSubject()
 
+	/**
+	 * Refresh the subject's own fields on every item that IS the subject.
+	 *
+	 * The seam between two requirements that are compatible but easy to
+	 * conflate. Items survive the pause (flow-runs REQ-FR-003): resuming
+	 * continues from the stored list, never a fresh seed, so everything the
+	 * earlier steps produced is kept. But an item that is the run's SUBJECT
+	 * began as a trigger-time snapshot of it, and a step that branches on a
+	 * subject field after a human answered — "is the description filled in
+	 * now?" — must read the subject as it stands, not as it stood when the
+	 * run started. Without this, the shipped re-ask loop can never succeed:
+	 * the answer lands on the object, the re-check reads the frozen snapshot,
+	 * and the run strands.
+	 *
+	 * So: only items carrying the subject's identity are touched, and on
+	 * those the live subject's serialised fields are merged OVER the stale
+	 * snapshot. A key the live subject does not serialise — a task outcome
+	 * bag, a step's computed field — survives untouched, which is what keeps
+	 * REQ-FR-003 true. An item that is some OTHER object (a fan-out read)
+	 * carries its own identity and is never smeared with the subject's data.
+	 *
+	 * @param array<int, array<string, mixed>> $items The stored items.
+	 * @param object $subject The live subject.
+	 * @param string $subjectUuid The run's subject uuid, matching items by identity.
+	 *
+	 * @return array<int, array<string, mixed>> The items, subject fields refreshed.
+	 *
+	 * @spec openspec/changes/or-flow-runs/specs/flow-runs/spec.md#requirement-resuming-carries-the-runs-own-items-req-fr-003
+	 */
+	public static function refreshSubjectProjection(array $items, object $subject, string $subjectUuid): array {
+		$subjectUuid = trim($subjectUuid);
+		if ($subjectUuid === '' || $items === []) {
+			return $items;
+		}
+
+		$live = (array)(self::fromSubject(subject: $subject)[0][self::JSON] ?? []);
+		if ($live === []) {
+			return $items;
+		}
+
+		foreach ($items as $index => $item) {
+			if (is_array($item) === false) {
+				continue;
+			}
+
+			$json = (array)($item[self::JSON] ?? []);
+			if (self::identityOf(json: $json) !== $subjectUuid) {
+				continue;
+			}
+
+			// Live wins on the subject's own keys; step-produced keys the
+			// subject does not serialise are kept as the run carried them.
+			$items[$index][self::JSON] = array_merge($json, $live);
+		}
+
+		return $items;
+	}//end refreshSubjectProjection()
+
+	/**
+	 * The object identity an item's record carries, if any.
+	 *
+	 * A subject-seeded item serialises its uuid as `@self.id` mirrored at the
+	 * top-level `id`; older shapes carried a flat `uuid`. An item with none of
+	 * these has no object identity and can never be "the subject".
+	 *
+	 * @param array $json The item's record.
+	 *
+	 * @return string The identity, or '' when the item carries none.
+	 *
+	 * @spec openspec/changes/or-flow-runs/specs/flow-runs/spec.md#requirement-resuming-carries-the-runs-own-items-req-fr-003
+	 */
+	private static function identityOf(array $json): string {
+		$self = ($json['@self'] ?? null);
+		if (is_array($self) === true && trim((string)($self['id'] ?? '')) !== '') {
+			return trim((string)$self['id']);
+		}
+
+		foreach (['id', 'uuid'] as $key) {
+			$value = ($json[$key] ?? null);
+			if (is_string($value) === true && trim($value) !== '') {
+				return trim($value);
+			}
+		}
+
+		return '';
+	}//end identityOf()
+
 	/**
 	 * Read an already-set `pairedItem` index, falling back when absent.
 	 *

lib/Service/Flow/FlowLocator.php

@@ -36,6 +36,7 @@
 use OCA\OpenRegister\Db\Flow;
 use OCA\OpenRegister\Db\FlowMapper;
 use OCA\OpenRegister\Db\FlowTriggerMapper;
+use OCA\OpenRegister\Db\FlowVersionMapper;
 use OCA\OpenRegister\Db\ObjectEntity;
 use OCA\OpenRegister\Service\ObjectService;
 use Psr\Log\LoggerInterface;
@@ -66,12 +67,14 @@ class FlowLocator {
 	 * @param FlowTriggerMapper $triggerMapper The node-derived trigger index.
 	 * @param ObjectService $objectService Loads the object a run is about.
 	 * @param LoggerInterface $logger Records refusals and failures.
+	 * @param FlowVersionMapper $versions Answers whether a fallback-matched flow has a published version.
 	 */
 	public function __construct(
 		private readonly FlowMapper $mapper,
 		private readonly FlowTriggerMapper $triggerMapper,
 		private readonly ObjectService $objectService,
 		private readonly LoggerInterface $logger,
+		private readonly FlowVersionMapper $versions,
 	) {
 
 	}//end __construct()
@@ -221,19 +224,10 @@ public function flowsForTrigger(string $event, string $register, string $schema)
 			return [];
 		}
 
-		// The column match is kept ONLY for flows the index does not represent.
-		// A converted flow's columns are stale by construction — they hold one
-		// trigger where its nodes may hold several — so consulting them for a
-		// converted flow would resurrect a subscription its author removed.
-		$matched = [];
-		foreach ($candidates as $flow) {
-			$uuid = (string)$flow->getUuid();
-			if (isset($converted[$uuid]) === true) {
-				continue;
-			}
-
-			$matched[$uuid] = $flow;
-		}
+		// The column match is kept ONLY for unconverted flows, and among those
+		// only for flows a published version can back — see the docblock of
+		// {@see self::columnFallbackMatches()} for both halves of that rule.
+		$matched = $this->columnFallbackMatches(candidates: $candidates, converted: $converted, event: $event);
 
 		// The index answers in UUIDs; the dispatch check needs the row. These
 		// are the flows whose NODES want this event.
@@ -259,6 +253,85 @@ public function flowsForTrigger(string $event, string $register, string $schema)
 		return $this->dispatchableUuids(matched: $matched, event: $event);
 	}//end flowsForTrigger()
 
+	/**
+	 * The column-fallback matches: unconverted flows a published version can back.
+	 *
+	 * Two rules, one per failure mode this fallback has produced:
+	 *
+	 * - ONLY flows the index does not represent. A converted flow's columns
+	 *   are stale by construction — they hold one trigger where its nodes may
+	 *   hold several — so consulting them would resurrect a subscription its
+	 *   author removed.
+	 * - 🔴 ONLY flows with a published version (flow-definition-versioning:
+	 *   a draft's trigger nodes SHALL NOT match anything). The index enforces
+	 *   that by derivation from the published graph, but the fallback did not:
+	 *   an enabled flow that was never published has zero index rows, so it is
+	 *   "unrepresented", its columns match, and `queue()` then refuses it with
+	 *   "no published version" — a refusal that used to abort the whole
+	 *   fan-out for every healthy flow on the same event. Refused OUT LOUD
+	 *   here instead. Legacy flows all hold a backfilled published version 1,
+	 *   so the only thing filtered is a flow that could never have run.
+	 *
+	 * @param array<int, Flow> $candidates The flows whose columns match the event.
+	 * @param array<string, int> $converted Uuids the index represents, flipped for lookup.
+	 * @param string $event The trigger being resolved, for the log line.
+	 *
+	 * @return array<string, Flow> The matches, by uuid.
+	 *
+	 * @spec openspec/changes/flow-definition-versioning/specs/flow-definition-versioning/spec.md#requirement-trigger-matching-answers-which-flow-the-queue-path-answers-which-version
+	 */
+	private function columnFallbackMatches(array $candidates, array $converted, string $event): array {
+		$matched = [];
+		foreach ($candidates as $flow) {
+			$uuid = (string)$flow->getUuid();
+			if (isset($converted[$uuid]) === true) {
+				continue;
+			}
+
+			if ($this->hasPublishedVersion(uuid: $uuid) === false) {
+				$this->logger->warning(
+					message: '[FlowLocator] Flow "' . $uuid . '" matched trigger "' . $event
+						. '" through its columns but has no published version, so it cannot back a run.'
+						. ' Publish a version to make it fire.',
+					context: ['file' => __FILE__, 'line' => __LINE__, 'flow' => $uuid]
+				);
+				continue;
+			}
+
+			$matched[$uuid] = $flow;
+		}//end foreach
+
+		return $matched;
+	}//end columnFallbackMatches()
+
+	/**
+	 * Whether a flow has a published version to back a run.
+	 *
+	 * FAIL OPEN on an unreadable version table: silencing every fallback flow
+	 * because one lookup failed would stop the engine without a word, and
+	 * `FlowRunVersionPin` re-checks the same fact at queue time anyway — this
+	 * filter only decides whether a flow is worth OFFERING to the queue.
+	 *
+	 * @param string $uuid The flow's uuid.
+	 *
+	 * @return bool True when a published version exists, or when the answer is unknowable.
+	 *
+	 * @spec openspec/changes/flow-definition-versioning/specs/flow-definition-versioning/spec.md#requirement-trigger-matching-answers-which-flow-the-queue-path-answers-which-version
+	 */
+	private function hasPublishedVersion(string $uuid): bool {
+		try {
+			return $this->versions->findPublished(flowUuid: $uuid) !== null;
+		} catch (Throwable $e) {
+			$this->logger->warning(
+				message: '[FlowLocator] Could not read the published version of flow "' . $uuid . '": '
+					. $e->getMessage() . '. Keeping it in the match; the queue path decides.',
+				context: ['file' => __FILE__, 'line' => __LINE__, 'flow' => $uuid]
+			);
+
+			return true;
+		}
+	}//end hasPublishedVersion()
+
 	/**
 	 * The uuids of the matched flows that can actually be dispatched.
 	 *

lib/Service/Flow/FlowRunService.php

@@ -269,6 +269,11 @@ public function advanceStream(FlowRun $run, array $flow, object $subject, string
 
 		$flow = $pinned;
 
+		// Resume-after-human-answer must see the subject as it stands: the
+		// answer the task asked for landed on the OBJECT, and the stored items
+		// still hold the trigger-time snapshot of it.
+		$this->refreshSubjectItems(run: $run, subject: $subject);
+
 		$walk = $this->streamWalkFor(run: $run, flow: $flow, onlyStream: $streamId, budget: $firings);
 		if ($walk === null) {
 			// Without the stream layer there is no branch to scope to; the
@@ -823,6 +828,12 @@ public function execute(FlowRun $run, array $flow, object $subject, ?array $seed
 		$flow = $pinned;
 
 		$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()}
+			$this->refreshSubjectItems(run: $run, subject: $subject);
+		}
+
 		$run->setStatus(FlowRun::STATUS_RUNNING);
 		$run->setUpdated(new DateTime());
 		$this->mapper->update($run);
@@ -895,6 +906,60 @@ public function execute(FlowRun $run, array $flow, object $subject, ?array $seed
 		return $this->persistResult(run: $run, result: $result);
 	}//end execute()
 
+	/**
+	 * Refresh the subject's fields on a resumed run's stored items, in place.
+	 *
+	 * Both stores are refreshed because both are read on resume: the flat
+	 * `items` list feeds a legacy walk, and the per-place buffers are what
+	 * {@see FlowItemPlacement::seedPlaceItems()} prefers — the check node that
+	 * re-enters after a human answered reads ITS input place's buffer, and a
+	 * refresh that missed it would leave the stale snapshot exactly where the
+	 * branch decision is made.
+	 *
+	 * In place and before the walk's first persist, so the commit path — which
+	 * re-reads the run row under its lock — sees the refreshed buffers too.
+	 *
+	 * A subjectless run, or one whose stored items never carried the subject,
+	 * is untouched: {@see FlowItems::refreshSubjectProjection()} matches by
+	 * identity and only ever touches the item that IS the subject.
+	 *
+	 * @param FlowRun $run The resumed run, mutated in place.
+	 * @param object $subject The live subject, as the advancer just resolved it.
+	 *
+	 * @return void
+	 *
+	 * @SuppressWarnings(PHPMD.StaticAccess) FlowItems::refreshSubjectProjection
+	 * is a pure function on value shapes; a factory to call it would add a
+	 * dependency to say the same thing.
+	 *
+	 * @spec openspec/changes/or-flow-runs/specs/flow-runs/spec.md#requirement-resuming-carries-the-runs-own-items-req-fr-003
+	 */
+	private function refreshSubjectItems(FlowRun $run, object $subject): void {
+		$subjectUuid = trim((string)$run->getSubjectUuid());
+		if ($subjectUuid === '') {
+			return;
+		}
+
+		$run->setItems(FlowItems::refreshSubjectProjection(
+			items: ($run->getItems() ?? []),
+			subject: $subject,
+			subjectUuid: $subjectUuid
+		));
+
+		$stored = $run->getPlaceItems();
+		if (is_array($stored) === true && $stored !== []) {
+			foreach ($stored as $place => $bucket) {
+				$stored[$place] = FlowItems::refreshSubjectProjection(
+					items: (array)$bucket,
+					subject: $subject,
+					subjectUuid: $subjectUuid
+				);
+			}
+
+			$run->setPlaceItems($stored);
+		}
+	}//end refreshSubjectItems()
+
 	/**
 	 * Fail a run whose pinned version cannot be resolved.
 	 *