fix(flows): stop the flow migration copying non-flow objects, and remove the rows it already wrote (#3408)

MigrateRegisterFlowsToTable asked ObjectService::findAll() for
register `flows` / schema `flow` at the TOP LEVEL of the config.
prepareFindAllConfig() reads $config['filters']['register'] and
$config['filters']['schema'] and no other key, so both were inert:
setRegister()/setSchema() were never called and the read ran against
whatever $currentRegister/$currentSchema the SHARED ObjectService
instance was still carrying.

saveObject() sets that context and never restores it - find() puts it
back in a finally, saveObject() has no such block - and
ImportCredentialBrokerRegister runs four repair steps ahead of the
migration, saving credential_broker_register.json's two example objects
through it. So the migration inherited credential-broker /
brokeredcredential and copied both examples into openregister_flows:
empty nodes, empty edges, no trigger, owner __system__ (which is only
what OpenRegister stamps on a sessionless write, not a convention).

Every other findAll() caller under lib/ nests the pair under filters.
This step was the outlier, and no test looked at the argument because
every fixture mocked findAll() to answer regardless of what was asked.

Two independent guards, because they fail independently:

  - selection: the pair moves under `filters`, with _rbac/_multitenancy
    off since a repair step has no session;
  - validity: isFlowDefinition() refuses any row with no nodes, no edges
    and no trigger. It depends on nothing but the row in front of it,
    where the scoping guard depends on shared mutable service state
    staying correct for a whole occ upgrade - which is what broke.

PurgePhantomMigratedFlows removes what already shipped, on the full
conjunction only: openregister-owned, disabled, no nodes, no edges, no
trigger/triggerRegister/triggerSchema/cron, never dispatched (lastRun*
all null AND zero rows in openregister_flow_runs), AND its uuid still
resolves to a register object whose schema is not `flow`.

It spares a row with one node or one edge, a graph-less draft that names
a trigger, any enabled flow, anything with run history, any other app's
flow, and the ambiguous case - an unrunnable shell resolving to nothing
or to a real flow object is named in the output, never deleted. Erring
toward keeping an unrunnable row is recoverable; erring the other way is
not. Idempotent; post-migration only, since the fixed migration cannot
produce these rows on a fresh install.

Also completes Flow's @method roster: status, statusMessage and the four
lastRun* accessors were added as columns by Version1Date20260805100000
and never declared, so phpstan could not see them.

Author: rubenvdlinde

appinfo/info.xml

@@ -217,6 +217,27 @@ Vrij en open source onder de EUPL-licentie.
 			     Copies the register-authored ones across, disabled, leaving the
 			     register rows in place so the step stays reversible. -->
 			<step>OCA\OpenRegister\Repair\MigrateRegisterFlowsToTable</step>
+			<!-- CLEANS UP AFTER THE VERSION OF THE STEP ABOVE THAT DID NOT SCOPE
+			     ITS READ. It asked findAll() for register `flows` / schema `flow`
+			     at the TOP LEVEL of the config; ObjectService reads that pair
+			     from `filters` and nowhere else, so both keys were inert and the
+			     read ran against whatever register/schema the SHARED service was
+			     still carrying. saveObject() sets that context and never restores
+			     it, and ImportCredentialBrokerRegister saves two example objects
+			     through it four steps earlier — so the migration copied both
+			     `brokeredcredential` examples into openregister_flows: no nodes,
+			     no edges, no trigger, owner `__system__`.
+
+			     Post-migration ONLY. A fresh install runs the fixed migration,
+			     which cannot produce these rows, so listing this under <install>
+			     would be a guaranteed no-op.
+
+			     Removes a row ONLY on the full conjunction: openregister-owned,
+			     disabled, no nodes, no edges, no trigger/cron, never dispatched
+			     (lastRun* null AND zero run rows), AND its uuid still resolves to
+			     a register object whose schema is not `flow`. Anything ambiguous
+			     is REPORTED by uuid, not deleted. Idempotent. -->
+			<step>OCA\OpenRegister\Repair\PurgePhantomMigratedFlows</step>
 			<step>OCA\OpenRegister\Repair\ImportTrustConfigurationRegister</step>
 			<step>OCA\OpenRegister\Repair\SeedVocabularyRegister</step>
 			<step>OCA\OpenRegister\Repair\RegisterOpenRegisterWithDoriath</step>

lib/Db/Flow.php

@@ -96,6 +96,18 @@
  * @method void setCreated(?DateTime $created)
  * @method DateTime|null getUpdated()
  * @method void setUpdated(?DateTime $updated)
+ * @method string|null getStatus()
+ * @method void setStatus(?string $status)
+ * @method string|null getStatusMessage()
+ * @method void setStatusMessage(?string $statusMessage)
+ * @method string|null getLastRunUuid()
+ * @method void setLastRunUuid(?string $lastRunUuid)
+ * @method string|null getLastRunStatus()
+ * @method void setLastRunStatus(?string $lastRunStatus)
+ * @method string|null getLastRunMessage()
+ * @method void setLastRunMessage(?string $lastRunMessage)
+ * @method DateTime|null getLastRunAt()
+ * @method void setLastRunAt(?DateTime $lastRunAt)
  *
  * An Entity mirrors its table one field per column, so the field count below is
  * the schema's rather than a design choice. Splitting it would need a second

lib/Db/FlowRunMapper.php

@@ -674,6 +674,41 @@ public function hasActiveRun(string $flowId): bool {
 		return $row !== false;
 	}//end hasActiveRun()
 
+	/**
+	 * How many runs this flow has EVER had, terminal ones included.
+	 *
+	 * {@see hasActiveRun} answers a different question — whether a run is going
+	 * right now — and a flow that ran to completion last year has none. Deleting
+	 * a flow is irreversible, so the guard that authorises it has to ask about
+	 * the flow's whole history, not its current activity.
+	 *
+	 * Used by {@see \OCA\OpenRegister\Repair\PurgePhantomMigratedFlows} as one
+	 * conjunct of the never-dispatched proof: a row with even one run row has
+	 * been part of something that happened and is never removed.
+	 *
+	 * @param string $flowId The flow's uuid.
+	 *
+	 * @return integer The number of run rows recorded against this flow.
+	 *
+	 * @spec openspec/changes/flow-engine-unification/specs/flow-storage/spec.md
+	 */
+	public function countRunsForFlow(string $flowId): int {
+		if (trim($flowId) === '') {
+			return 0;
+		}
+
+		$qb = $this->db->getQueryBuilder();
+		$qb->select($qb->createFunction('COUNT(*) AS `total`'))
+			->from($this->getTableName())
+			->where($qb->expr()->eq('flow_id', $qb->createNamedParameter($flowId)));
+
+		$result = $qb->executeQuery();
+		$row = $result->fetch();
+		$result->closeCursor();
+
+		return (int)($row['total'] ?? 0);
+	}//end countRunsForFlow()
+
 	/**
 	 * How many runs are still going, for one organisation.
 	 *

lib/Repair/MigrateRegisterFlowsToTable.php

@@ -175,6 +175,7 @@ public function run(IOutput $output): void {
 		$moved = 0;
 		$already = 0;
 		$failed = 0;
+		$notFlows = 0;
 
 		foreach ($objects as $object) {
 			$row  = $this->normalise(row: $object);
@@ -184,6 +185,16 @@ public function run(IOutput $output): void {
 				continue;
 			}
 
+			// The second, independent guard — see isFlowDefinition().
+			if ($this->isFlowDefinition(object: $row['data']) === false) {
+				$notFlows++;
+				$this->logger->warning(
+					'[MigrateRegisterFlowsToTable] refused ' . $uuid
+					. ': no nodes, no edges and no trigger, so it is not a flow definition'
+				);
+				continue;
+			}
+
 			try {
 				$this->flowMapper->findByUuid($uuid);
 				$already++;
@@ -211,26 +222,87 @@ public function run(IOutput $output): void {
 		// the defect this repairs stayed invisible.
 		$output->info(
 			sprintf(
-				'Register-authored flows: %d migrated, %d already in the table, %d failed (register rows left in place).',
+				'Register-authored flows: %d migrated, %d already in the table, %d not a flow definition, %d failed (register rows left in place).',
 				$moved,
 				$already,
+				$notFlows,
 				$failed
 			)
 		);
 	}//end run()
 
+	/**
+	 * Whether this object is a flow definition at all.
+	 *
+	 * 🔴 BOTH GUARDS, DELIBERATELY. Scoping the read (see registerFlows()) fixes
+	 * the wrong POPULATION being selected; this fixes a wrong ROW inside the
+	 * selected one being written. They fail independently, which is the point:
+	 * the scoping guard depends on shared, mutable service state staying correct
+	 * for the whole of an `occ upgrade`, and that state is exactly what broke.
+	 * This one depends on nothing but the row in front of it.
+	 *
+	 * The test is what the flow engine needs to do anything: a graph to walk
+	 * (`nodes` or `edges`) or something to start it (`trigger`). A row with none
+	 * of the three cannot run, cannot be made to run, and is not a flow — it is
+	 * whatever else the read handed us. It is not a completeness check: an empty
+	 * graph that already names a trigger is a half-authored flow and crosses.
+	 *
+	 * @param array<string, mixed> $object The register object's payload.
+	 *
+	 * @return boolean True when the payload is a flow definition.
+	 */
+	private function isFlowDefinition(array $object): bool {
+		if (empty($object['nodes']) === false) {
+			return true;
+		}
+
+		if (empty($object['edges']) === false) {
+			return true;
+		}
+
+		$trigger = ($object['trigger'] ?? null);
+
+		return is_string($trigger) === true && trim($trigger) !== '';
+	}//end isFlowDefinition()
+
 	/**
 	 * Every object in the flows register.
 	 *
 	 * @return array<int, mixed> The rows, each an ObjectEntity (see normalise()).
 	 */
 	private function registerFlows(): array {
+		// 🔴 `register` / `schema` GO UNDER `filters`, AND NOWHERE ELSE.
+		//
+		// This step used to pass them at the top level of the config, which
+		// reads correctly and does nothing: `ObjectService::prepareFindAllConfig()`
+		// inspects `$config['filters']['register']` and `$config['filters']['schema']`
+		// and no other key, so `setRegister()` / `setSchema()` were never called
+		// and the read ran against whatever `$currentRegister` / `$currentSchema`
+		// the SHARED ObjectService instance was still carrying.
+		//
+		// MEASURED, not inferred. `saveObject()` sets that context and never
+		// restores it — `find()` does, in a `finally`; `saveObject()` has no such
+		// block. `ImportCredentialBrokerRegister` runs four repair steps ahead of
+		// this one and saves `credential_broker_register.json`'s two example
+		// objects through it. So this step inherited `credential-broker` /
+		// `brokeredcredential` and copied both examples into `openregister_flows`:
+		// empty nodes, empty edges, no trigger, `_owner` = `__system__` (which is
+		// only what OpenRegister stamps on a sessionless write, not a convention).
+		//
+		// Every other `findAll()` caller under `lib/` nests the pair. This step
+		// was the outlier, and `_rbac` / `_multitenancy` are off because a repair
+		// step has no session — `occ` runs as Anonymous, so a scoped read would
+		// return nothing at all.
 		$result = $this->objectService->findAll(
-			[
-				'register' => self::FLOW_REGISTER,
-				'schema'   => self::FLOW_SCHEMA,
-				'limit'    => 1000,
-			]
+			config: [
+				'filters' => [
+					'register' => self::FLOW_REGISTER,
+					'schema'   => self::FLOW_SCHEMA,
+				],
+				'limit'   => 1000,
+			],
+			_rbac: false,
+			_multitenancy: false
 		);
 
 		// No `is_array($result)` guard: `findAll()` is typed to return an array,