From 2cfffed8a893359a04739a4bb4cad95898e79711 Mon Sep 17 00:00:00 2001 From: Conduction Release Bot Date: Wed, 2 Sep 2026 05:31:10 +0200 Subject: [PATCH 1/3] fix(flow): scope the resume payload to the node whose slot it answers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A resume signal wakes the RUN but answers one NODE. The payload landed in the shared context key 'signal', which every wait node re-entered in the same walk could read — so after any wait-state resolution, every LATER wait node consumed the same payload as its own answer and completed in 0ms instead of suspending, and the run raced to its end. Observed live on 2.0.15-unstable (runs f8996ccc and ca50c56c): a decision node's outcome held an applicant task's completion payload, and a second decision node inherited the first decision's decisionRef because it never suspended. The dispatcher now withholds the signal from every node except the one whose held FlowNodeResumeState slot marks it as the node that suspended — the same enforcement point that already scopes the resume slots. The answered node reads its payload; every other wait node the resumed walk enters suspends fresh, creating its own task or decision. WaitNode rode the same seam through the run-wide 'resuming' flag: a second wait in a resumed walk read the flag as 'my wait is over' and passed through in zero seconds. It now writes its own slot when it suspends and passes through only when that slot is held. Regressions: FlowEngineResumeScopeTest (engine-level, two sequential waits + the decision-correlation variant), FlowRunServiceTest:: testAnsweringOneWaitDoesNotAnswerTheNext (the worker path) and testAFreshRunsFirstAskSeesNoSignal, RegistryStepDispatcherResumeTest signal-scoping quartet, WaitNodeTest second-wait pair. --- lib/Service/Flow/Nodes/WaitNode.php | 31 +- lib/Service/Flow/RegistryStepDispatcher.php | 57 ++++ .../Flow/FlowEngineResumeScopeTest.php | 301 ++++++++++++++++++ .../Unit/Service/Flow/FlowRunServiceTest.php | 252 +++++++++++++++ .../Flow/RegistryStepDispatcherResumeTest.php | 117 +++++++ tests/Unit/Service/Flow/WaitNodeTest.php | 67 ++++ 6 files changed, 821 insertions(+), 4 deletions(-) create mode 100644 tests/Unit/Service/Flow/FlowEngineResumeScopeTest.php diff --git a/lib/Service/Flow/Nodes/WaitNode.php b/lib/Service/Flow/Nodes/WaitNode.php index 6eb37ad053..e393354160 100644 --- a/lib/Service/Flow/Nodes/WaitNode.php +++ b/lib/Service/Flow/Nodes/WaitNode.php @@ -9,7 +9,11 @@ * * Because the marking does not advance, this node runs a SECOND time when the * run resumes. That is what makes it correct rather than a one-shot: on the - * way back in it sees `context.resuming` and lets the items straight through. + * way back in it sees its OWN resume slot held and lets the items straight + * through. Its own slot, not `context.resuming` — the run-wide flag is true + * for every node of a resumed walk, so a SECOND wait node reached later in + * the same walk would read it as "my wait is over" and pass through without + * ever having waited. The slot marks exactly the node that suspended. * * SPDX-FileCopyrightText: 2026 Conduction B.V. * SPDX-License-Identifier: EUPL-1.2 @@ -31,6 +35,7 @@ namespace OCA\OpenRegister\Service\Flow\Nodes; use DateTime; +use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\IFlowNode; use OCA\OpenRegister\Service\Flow\IFlowNodeConfigKeys; @@ -175,9 +180,19 @@ public function execute(array $items, array $config, array $context): array { return $items; } - if (($context['resuming'] ?? false) === true) { - // Woken by the worker: the wait is over by construction, because - // the run was only eligible once `resumeAt` had passed. + $resume = ($context[FlowNodeResumeState::CONTEXT_KEY] ?? null); + if ($resume instanceof FlowNodeResumeState === true) { + if ($resume->isResuming() === true) { + // THIS node's wait is over: it wrote its slot when it + // suspended, and the run was only eligible again once its + // `resumeAt` had passed. The dispatcher clears the slot on + // return, so a loop back into this node waits again. + return $items; + } + } elseif (($context['resuming'] ?? false) === true) { + // No slot machinery at all — a context built outside a real run + // (the flow tester, a node unit test). There is only one wait in + // such a walk, so the run-wide flag is unambiguous there. return $items; } @@ -189,6 +204,14 @@ public function execute(array $items, array $config, array $context): array { return $items; } + // The slot is the addressee mark the re-entry above reads. Without it, + // a resumed walk cannot tell the wait that is OVER from a wait it has + // only just reached — `context.resuming` is true for both, and reading + // that flag made a second wait node pass through in zero seconds. + if ($resume instanceof FlowNodeResumeState === true) { + $resume->set(key: 'waitingUntil', value: $resumeAt->format('c')); + } + throw new FlowSuspension( resumeAt: $resumeAt, reason: sprintf('waiting until %s', $resumeAt->format('c')) diff --git a/lib/Service/Flow/RegistryStepDispatcher.php b/lib/Service/Flow/RegistryStepDispatcher.php index 3dc05b299a..f1ec654184 100644 --- a/lib/Service/Flow/RegistryStepDispatcher.php +++ b/lib/Service/Flow/RegistryStepDispatcher.php @@ -110,6 +110,12 @@ public function dispatch(array $step, array $items, array $context): array { // and cannot reach another node's slot even by accident. $scoped = $this->scopeResumeState(step: $step, context: $context); + // Scope the resume SIGNAL the same way. Without this, the payload that + // answered ONE node's question is readable by every wait node the + // resumed walk goes on to enter, and each of them completes on somebody + // else's answer instead of suspending with a question of its own. + $this->scopeSignal(context: $context, scoped: $scoped); + $startedAt = microtime(true); $out = $node->execute(items: $items, config: $config, context: $context); $tookMs = (int)round((microtime(true) - $startedAt) * 1000); @@ -159,6 +165,57 @@ private function scopeResumeState(array $step, array &$context): ?FlowNodeResume return $scoped; }//end scopeResumeState() + /** + * Withhold the resume signal from every node except the one it answers. + * + * A signal wakes a RUN, but it answers one NODE: the one that suspended and + * whose resume slot is still held. The walk's context is shared, so without + * this gate every wait node the resumed walk re-enters AFTER the answered + * one reads the same payload as its own answer and completes instead of + * suspending — a flow with two approval steps auto-approves the second the + * moment the first is granted, and a decision step downstream of an answered + * task adopts the task's payload as its decision outcome. Observed live on + * dossiq case flows (runs f8996ccc and ca50c56c, 2026-09-01): a DECISION + * node's outcome held `{decision, node: "ask-indiener", taskId}` — an + * applicant task's completion — and a second decision node inherited the + * first decision's reference because it never suspended at all. + * + * The slot is the addressee test, not `$context['resuming']`: the run-wide + * flag is true for every node of a resumed walk, while a held slot marks + * exactly the node that suspended and has not yet been given its answer. + * The dispatcher clears the slot when a node returns, so a wait node + * re-entered later in the SAME walk (a loop) asks fresh rather than + * re-reading a consumed answer. With no slot machinery at all (a + * container-built dispatcher walking a tester context) the signal is + * withheld too: with no slots there is no addressee, and withholding makes + * the node suspend visibly where delivering would answer the wrong + * question silently. + * + * The strip is LOCAL to this node's context copy — `dispatch()` receives + * `$context` by value — so the walk keeps carrying the signal to the node + * whose slot it answers, wherever in the round-robin that node is visited. + * + * @param array $context The node context, modified in place. + * @param FlowNodeResumeState|null $scoped This node's resume slot, when it has one. + * + * @return void + * + * @spec openspec/specs/flow-engine/spec.md#requirement-a-run-suspended-on-an-external-signal-must-be-reachable + */ + private function scopeSignal(array &$context, ?FlowNodeResumeState $scoped): void { + if (array_key_exists(FlowRunService::SIGNAL_CONTEXT_KEY, $context) === false) { + return; + } + + if ($scoped !== null && $scoped->isResuming() === true) { + // This node is the one that suspended: the answer is its to read. + return; + } + + unset($context[FlowRunService::SIGNAL_CONTEXT_KEY]); + + }//end scopeSignal() + /** * Stop a step that took longer than its own `maxRuntimeSeconds`. * diff --git a/tests/Unit/Service/Flow/FlowEngineResumeScopeTest.php b/tests/Unit/Service/Flow/FlowEngineResumeScopeTest.php new file mode 100644 index 0000000000..9789c24cb5 --- /dev/null +++ b/tests/Unit/Service/Flow/FlowEngineResumeScopeTest.php @@ -0,0 +1,301 @@ + + * SPDX-License-Identifier: EUPL-1.2 + * + * @category Test + * @package OCA\OpenRegister\Tests\Unit\Service\Flow + * + * @author Conduction Development Team + * @license EUPL-1.2 https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12 + */ + +declare(strict_types=1); + +namespace Unit\Service\Flow; + +use OCA\OpenRegister\Service\Flow\FlowDefinitionBuilder; +use OCA\OpenRegister\Service\Flow\FlowEngine; +use OCA\OpenRegister\Service\Flow\FlowNodeRegistry; +use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; +use OCA\OpenRegister\Service\Flow\FlowResumeState; +use OCA\OpenRegister\Service\Flow\FlowRunService; +use OCA\OpenRegister\Service\Flow\FlowSuspension; +use OCA\OpenRegister\Service\Flow\IFlowNode; +use OCA\OpenRegister\Service\Flow\RegistryStepDispatcher; +use PHPUnit\Framework\TestCase; +use Psr\Log\LoggerInterface; +use Symfony\Component\Workflow\MarkingStore\MethodMarkingStore; + +/** + * An await-signal-shaped node: consumes the signal when one is visible, + * otherwise records its ask in its own slot and suspends. + * + * Mirrors what `AwaitSignalNode`, dossiq's askPerson and dossiq's + * requestDecision all do at this seam, including the per-ask reference a + * decision node mints (the correlation the second node must NOT inherit). + */ +class ScopedAskStub implements IFlowNode { + + /** + * The signal each node id saw on each entry, in order. + * + * @var array> + */ + public array $sawSignal = []; + + /** + * How many asks have been minted, so each ref is distinct. + * + * @var int + */ + private int $asks = 0; + + public function getId(): string { + return 'test.scoped-ask'; + } + + public function getDisplayName(): string { + return 'Ask'; + } + + public function getDescription(): string { + return 'Waits for an answer.'; + } + + public function getIcon(): string { + return 'i.svg'; + } + + public function isAvailableForScope(int $scope): bool { + return true; + } + + public function validateConfig(array $config): void { + } + + public function execute(array $items, array $config, array $context): array { + $slot = $context[FlowNodeResumeState::CONTEXT_KEY]; + $nodeId = $slot->nodeId(); + + $signal = ($context[FlowRunService::SIGNAL_CONTEXT_KEY] ?? null); + $this->sawSignal[$nodeId][] = $signal; + + if (is_array($signal) === true && trim((string)($signal['decision'] ?? '')) !== '') { + foreach ($items as $index => $item) { + $item['json'][(string)($config['signalKey'] ?? 'signal')] = $signal; + $items[$index] = $item; + } + + return $items; + } + + if ($slot->has(key: 'ref') === false) { + $this->asks++; + $slot->merge( + values: [ + 'askedAt' => '2026-09-01T00:00:00+00:00', + 'ref' => sprintf('ask-%d', $this->asks), + ] + ); + } + + throw new FlowSuspension(resumeAt: null, reason: 'waiting for an answer'); + } +}//end class + +/** + * A subject whose marking is a plain property, as the engine tests use. + */ +class ResumeScopeSubject { + + public $marking = []; +}//end class + +class FlowEngineResumeScopeTest extends TestCase { + + /** + * The stub both wait steps resolve to. + * + * @var ScopedAskStub + */ + private ScopedAskStub $ask; + + /** + * The engine under test. + * + * @var FlowEngine + */ + private FlowEngine $engine; + + /** + * The dispatcher — the REAL one, because the scoping under test lives in it. + * + * @var RegistryStepDispatcher + */ + private RegistryStepDispatcher $dispatcher; + + protected function setUp(): void { + $this->ask = new ScopedAskStub(); + + $registry = $this->createMock(FlowNodeRegistry::class); + $registry->method('get')->willReturn($this->ask); + + $this->engine = new FlowEngine( + new FlowDefinitionBuilder(), + $this->createMock(LoggerInterface::class) + ); + $this->dispatcher = new RegistryStepDispatcher(registry: $registry); + }//end setUp() + + /** + * Two sequential wait nodes, one edge between them. + */ + private function twoAsksFlow(): array { + return [ + 'id' => 'f-scope', + 'nodes' => [ + ['id' => 'first-ask', 'type' => 'test.scoped-ask', 'config' => ['signalKey' => 'firstAnswer']], + ['id' => 'second-ask', 'type' => 'test.scoped-ask', 'config' => ['signalKey' => 'secondAnswer']], + ], + 'edges' => [ + ['id' => 'e1', 'from' => 'first-ask', 'to' => 'second-ask'], + ], + ]; + }//end twoAsksFlow() + + /** + * Answering the first wait leaves the second SUSPENDED, asking fresh. + * + * The walk that consumed the first answer must not hand the same payload + * to the second wait: the second suspends with an ask of its own, and the + * run does not race to the end on one answer. + */ + public function testAnsweringTheFirstWaitLeavesTheSecondSuspendedWithItsOwnAsk(): void { + $flow = $this->twoAsksFlow(); + $subject = new ResumeScopeSubject(); + $store = new MethodMarkingStore(false, 'marking'); + $state = new FlowResumeState(); + + // Walk 1: the first ask suspends the run. + $first = $this->engine->run( + $flow, + $store, + $subject, + $this->dispatcher, + [FlowResumeState::CONTEXT_KEY => $state] + ); + $this->assertSame(FlowEngine::STATUS_SUSPENDED, $first['status']); + $this->assertTrue($state->forNode(nodeId: 'first-ask')->has(key: 'askedAt')); + + // Walk 2: the answer arrives, exactly as FlowRunService seeds it — the + // stored slots and the signal, on a context whose run-wide `resuming` + // flag is true for EVERY node. + $payload = ['decision' => 'approved', 'node' => 'first-ask', 'taskId' => 'task-1']; + $second = $this->engine->run( + $flow, + $store, + $subject, + $this->dispatcher, + [ + 'resuming' => true, + FlowResumeState::CONTEXT_KEY => FlowResumeState::fromArray($state->all()), + FlowRunService::SIGNAL_CONTEXT_KEY => $payload, + ] + ); + + $this->assertSame( + FlowEngine::STATUS_SUSPENDED, + $second['status'], + 'one answer must advance the run to the NEXT question, not to the end' + ); + + $statuses = []; + foreach ($second['log'] as $entry) { + $statuses[$entry['transition']] = $entry['status']; + } + + $this->assertSame('completed', $statuses['first-ask'], 'the answered node completes'); + $this->assertSame('suspended', $statuses['second-ask'], 'the next wait suspends fresh'); + + $this->assertSame( + [null, $payload], + $this->ask->sawSignal['first-ask'], + 'the answered node reads the payload on its resume, and only then' + ); + $this->assertSame( + [null], + $this->ask->sawSignal['second-ask'], + 'the second node never sees the first answer' + ); + + $fresh = ($second['context'][FlowResumeState::CONTEXT_KEY] ?? null); + $this->assertInstanceOf(FlowResumeState::class, $fresh); + $this->assertTrue( + $fresh->forNode(nodeId: 'second-ask')->has(key: 'askedAt'), + 'the second node recorded an ask of its own' + ); + $this->assertSame( + [], + $fresh->forNode(nodeId: 'first-ask')->all(), + 'the answered node keeps no slot' + ); + }//end testAnsweringTheFirstWaitLeavesTheSecondSuspendedWithItsOwnAsk() + + /** + * The decision-node variant: the second ask mints its OWN reference. + * + * Run ca50c56c's defect in miniature — the second decision node carried + * the FIRST decision's `decisionRef`, because it consumed the first's + * outcome instead of suspending and creating a decision of its own. + */ + public function testTheSecondAskNeverInheritsTheFirstsCorrelation(): void { + $flow = $this->twoAsksFlow(); + $subject = new ResumeScopeSubject(); + $store = new MethodMarkingStore(false, 'marking'); + $state = new FlowResumeState(); + + $this->engine->run($flow, $store, $subject, $this->dispatcher, [FlowResumeState::CONTEXT_KEY => $state]); + $firstRef = $state->forNode(nodeId: 'first-ask')->get(key: 'ref'); + $this->assertNotNull($firstRef); + + $second = $this->engine->run( + $flow, + $store, + $subject, + $this->dispatcher, + [ + 'resuming' => true, + FlowResumeState::CONTEXT_KEY => FlowResumeState::fromArray($state->all()), + FlowRunService::SIGNAL_CONTEXT_KEY => ['decision' => 'completed', 'decisionRef' => $firstRef], + ] + ); + + $fresh = $second['context'][FlowResumeState::CONTEXT_KEY]; + $secondRef = $fresh->forNode(nodeId: 'second-ask')->get(key: 'ref'); + + $this->assertNotNull($secondRef, 'the second ask creates its own reference'); + $this->assertNotSame($firstRef, $secondRef, 'the second ask must not ride the first decision'); + + // And the second node's slot holds nothing of the first's payload. + $this->assertNull($fresh->forNode(nodeId: 'second-ask')->get(key: 'decisionRef')); + }//end testTheSecondAskNeverInheritsTheFirstsCorrelation() +}//end class diff --git a/tests/Unit/Service/Flow/FlowRunServiceTest.php b/tests/Unit/Service/Flow/FlowRunServiceTest.php index 48bdc95557..ac278104df 100644 --- a/tests/Unit/Service/Flow/FlowRunServiceTest.php +++ b/tests/Unit/Service/Flow/FlowRunServiceTest.php @@ -16,13 +16,18 @@ use OCA\OpenRegister\Service\Flow\FlowEngine; use OCA\OpenRegister\Service\Flow\FlowItems; use OCA\OpenRegister\Service\Flow\FlowNodeRegistry; +use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; +use OCA\OpenRegister\Service\Flow\FlowOversightRegistry; +use OCA\OpenRegister\Service\Flow\FlowResumeState; use OCA\OpenRegister\Service\Flow\FlowRunMarkingStore; use OCA\OpenRegister\Service\Flow\FlowRunService; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\IFlowNode; +use OCA\OpenRegister\Service\Flow\Oversight\KillSwitchCheck; use OCA\OpenRegister\Service\Flow\RegisterFlowNodesEvent; use OCP\EventDispatcher\Event; use OCP\EventDispatcher\IEventDispatcher; +use OCP\IAppConfig; use PHPUnit\Framework\TestCase; use Psr\Container\ContainerInterface; use Psr\Log\LoggerInterface; @@ -126,6 +131,62 @@ public function execute(array $items, array $config, array $context): array { } } +/** + * An await-signal-shaped node: consumes a visible signal as its answer, + * otherwise records the ask in its OWN resume slot and suspends — the shape + * AwaitSignalNode and dossiq's askPerson/requestDecision nodes share. + */ +class AskingNode implements IFlowNode { + /** + * The signal each node id saw on each entry, in order. + */ + public array $sawSignal = []; + + public function getId(): string { + return 'test.ask'; + } + + public function getDisplayName(): string { + return 'Ask'; + } + + public function getDescription(): string { + return 'Waits for an answer.'; + } + + public function getIcon(): string { + return 'i.svg'; + } + + public function isAvailableForScope(int $scope): bool { + return true; + } + + public function validateConfig(array $config): void { + } + + public function execute(array $items, array $config, array $context): array { + $slot = $context[FlowNodeResumeState::CONTEXT_KEY]; + $signal = ($context[FlowRunService::SIGNAL_CONTEXT_KEY] ?? null); + $this->sawSignal[$slot->nodeId()][] = $signal; + + if (is_array($signal) === true && trim((string)($signal['decision'] ?? '')) !== '') { + foreach ($items as $index => $item) { + $item['json'][$slot->nodeId()] = $signal; + $items[$index] = $item; + } + + return $items; + } + + if ($slot->has(key: 'askedAt') === false) { + $slot->set(key: 'askedAt', value: '2026-09-01T00:00:00+00:00'); + } + + throw new FlowSuspension(resumeAt: null, reason: 'waiting for an answer'); + } +} + /** Records the context it was handed, so attribution can be asserted. */ class ContextCapturingNode implements IFlowNode { public array $seenContext = []; @@ -173,6 +234,8 @@ class FlowRunServiceTest extends TestCase { private BranchRecordingNode $retryBranch; + private AskingNode $asker; + protected function setUp(): void { $this->mapper = $this->createMock(FlowRunMapper::class); // insert/update echo the entity back, so assertions read the real state. @@ -183,6 +246,7 @@ protected function setUp(): void { $this->capturer = new ContextCapturingNode(); $this->doneBranch = new BranchRecordingNode(id: 'test.done'); $this->retryBranch = new BranchRecordingNode(id: 'test.retry'); + $this->asker = new AskingNode(); $dispatcher = $this->createMock(IEventDispatcher::class); $dispatcher->method('dispatchTyped')->willReturnCallback( @@ -192,6 +256,7 @@ function (Event $event): void { $event->registerNode($this->capturer); $event->registerNode($this->doneBranch); $event->registerNode($this->retryBranch); + $event->registerNode($this->asker); } } ); @@ -410,6 +475,193 @@ public function testAReAskLoopThatReceivesTheAnswerTerminates(): void { $this->assertSame(1, $this->doneBranch->calls); }//end testAReAskLoopThatReceivesTheAnswerTerminates() + /** + * THE RESUME ANSWER IS SCOPED TO THE NODE THAT SUSPENDED — through the + * worker path. `signal()` then `execute()` is exactly what the resume + * endpoint plus the next FlowRunWorker pass do, so this is the defect the + * 2026-09-01 acceptance run caught (runs f8996ccc / ca50c56c) driven + * through the same seam: answering the FIRST wait must leave the SECOND + * suspended on a fresh ask of its own, not complete it in zero + * milliseconds on somebody else's answer. + * + * @return void + */ + public function testAnsweringOneWaitDoesNotAnswerTheNext(): void { + $flow = [ + 'id' => 'f1', + 'nodes' => [ + ['id' => 'first', 'type' => 'test.ask'], + ['id' => 'second', 'type' => 'test.ask'], + ], + 'edges' => [ + ['id' => 'e1', 'from' => 'first', 'to' => 'second'], + ], + ]; + + $run = $this->service->queue('f1', user: 'alice'); + $run = $this->service->execute($run, $flow, new RunSubject()); + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus()); + $this->assertArrayHasKey( + 'first', + ($run->getContext()[FlowResumeState::CONTEXT_KEY] ?? []), + 'the first ask holds the run' + ); + + // The answer to the FIRST question arrives; the worker picks the run up. + $answer = ['decision' => 'approved', 'node' => 'first', 'taskId' => 'task-1']; + $run = $this->service->signal($run, $answer); + $this->assertNotNull($run, 'a suspended run accepts its signal'); + $run = $this->service->execute($run, $flow, new RunSubject()); + + $this->assertSame( + FlowRun::STATUS_SUSPENDED, + $run->getStatus(), + 'one answer advances the run to the NEXT question, never to the end' + ); + + $slots = ($run->getContext()[FlowResumeState::CONTEXT_KEY] ?? []); + $this->assertArrayHasKey('second', $slots, 'the second ask recorded a question of its own'); + $this->assertArrayNotHasKey('first', $slots, 'the answered ask keeps no slot'); + $this->assertArrayNotHasKey( + FlowRunService::SIGNAL_CONTEXT_KEY, + ($run->getContext() ?? []), + 'the consumed signal does not linger in the stored context' + ); + + $this->assertSame( + [null, $answer], + $this->asker->sawSignal['first'] ?? [], + 'the answered node read the payload on its resume, and only then' + ); + $this->assertSame( + [null], + $this->asker->sawSignal['second'] ?? [], + 'the second node never saw the first answer' + ); + + // The second answer is the one that finishes the run, and each item + // carries each answer under its OWN node. + $secondAnswer = ['decision' => 'approved', 'node' => 'second']; + $run = $this->service->signal($run, $secondAnswer); + $run = $this->service->execute($run, $flow, new RunSubject()); + + $this->assertSame(FlowRun::STATUS_COMPLETED, $run->getStatus()); + $item = ($run->getItems()[0]['json'] ?? []); + $this->assertSame($answer, ($item['first'] ?? null)); + $this->assertSame($secondAnswer, ($item['second'] ?? null)); + }//end testAnsweringOneWaitDoesNotAnswerTheNext() + + /** + * On the first walk — no suspension yet, nothing resuming — an ask node + * asks rather than answering itself: the negative control for the + * scoping, proving the signal gate does not leak on fresh runs either. + * + * @return void + */ + public function testAFreshRunsFirstAskSeesNoSignal(): void { + $flow = [ + 'id' => 'f1', + 'nodes' => [['id' => 'only', 'type' => 'test.ask']], + 'edges' => [], + ]; + + $run = $this->service->queue('f1', user: 'alice'); + $run = $this->service->execute($run, $flow, new RunSubject()); + + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus()); + $this->assertSame([null], $this->asker->sawSignal['only'] ?? []); + }//end testAFreshRunsFirstAskSeesNoSignal() + + /** + * THE OPERATOR'S STOP LANDS ON THE NEXT OBSERVATION. The kill switch is + * thrown while a run is suspended; the operator nudges it (`resume` with + * an empty body is `signal([])`) and the next worker pass must end the run + * `stopped` — the oversight veto travels an author's Stop-step path — not + * leave it suspended forever. The run's terminal write is what triggers + * task termination (FlowRunTerminalEvent → terminateForRun), so a stop + * that never lands is also an inbox that never empties. + * + * @return void + */ + public function testAKillSwitchVetoLandsTheStopOnTheNextObservation(): void { + $thrown = false; + $appConfig = $this->createMock(IAppConfig::class); + $appConfig->method('getValueBool')->willReturnCallback( + static function (string $app, string $key, bool $default = false) use (&$thrown): bool { + return $thrown; + } + ); + + $oversight = new FlowOversightRegistry(logger: $this->createMock(LoggerInterface::class)); + $oversight->register(check: new KillSwitchCheck(appConfig: $appConfig)); + + $dispatcher = $this->createMock(IEventDispatcher::class); + $dispatcher->method('dispatchTyped')->willReturnCallback( + function (Event $event): void { + if ($event instanceof RegisterFlowNodesEvent) { + $event->registerNode($this->waiter); + } + } + ); + $registry = new FlowNodeRegistry($dispatcher, $this->createMock(LoggerInterface::class)); + $engine = new FlowEngine( + new FlowDefinitionBuilder(), + $this->createMock(LoggerInterface::class), + $oversight + ); + + $container = $this->createMock(ContainerInterface::class); + $versions = $this->publishedVersionMapper(); + $pin = $this->pinReturning(); + $container->method('get')->willReturnCallback( + function (string $id) use ($versions, $pin): object { + if ($id === \OCA\OpenRegister\Db\FlowVersionMapper::class) { + return $versions; + } + + if ($id === \OCA\OpenRegister\Service\Flow\FlowDefinitionPin::class) { + return $pin; + } + + throw new \RuntimeException('not available'); + } + ); + + $service = new FlowRunService( + $this->mapper, + $this->createMock(\OCA\OpenRegister\Db\FlowStateMapper::class), + $engine, + $registry, + $this->createMock(LoggerInterface::class), + $container + ); + + $run = $service->queue('f1', user: 'alice'); + $run = $service->execute($run, $this->waitFlow(), new RunSubject()); + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus()); + + // The operator throws the switch and nudges the parked run. + $thrown = true; + $run = $service->signal($run, []); + $this->assertNotNull($run); + $run = $service->execute($run, $this->waitFlow(), new RunSubject()); + + $this->assertSame( + FlowRun::STATUS_STOPPED, + $run->getStatus(), + 'the veto must END the run; a stop that leaves it suspended never lands' + ); + + $last = $run->getLog()[count($run->getLog()) - 1]; + $this->assertSame('stopped', $last['status']); + $this->assertSame( + 'openregister.kill-switch', + ($last['checkId'] ?? null), + 'the history records WHICH gate closed' + ); + $this->assertNull($run->getResumeAt(), 'a stopped run is never due again'); + }//end testAKillSwitchVetoLandsTheStopOnTheNextObservation() + /** * The refresh reaches the PER-PLACE buffers too: with a stream layer the * resumed walk reads its input place's buffer, not the flat list, so a diff --git a/tests/Unit/Service/Flow/RegistryStepDispatcherResumeTest.php b/tests/Unit/Service/Flow/RegistryStepDispatcherResumeTest.php index 7e85237819..825c5c6c5f 100644 --- a/tests/Unit/Service/Flow/RegistryStepDispatcherResumeTest.php +++ b/tests/Unit/Service/Flow/RegistryStepDispatcherResumeTest.php @@ -12,6 +12,7 @@ use OCA\OpenRegister\Service\Flow\FlowNodeRegistry; use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; use OCA\OpenRegister\Service\Flow\FlowResumeState; +use OCA\OpenRegister\Service\Flow\FlowRunService; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\IFlowNode; use OCA\OpenRegister\Service\Flow\RegistryStepDispatcher; @@ -192,4 +193,120 @@ public function testAStepWithNoIdGetsNoSlot(): void { $this->assertNull($seen); } + + /** + * A resume signal answers ONE node: the one whose held slot marks it as + * the node that suspended. That node reads the payload. + */ + public function testTheSignalReachesTheNodeWhoseSlotItAnswers(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'first-ask')->set(key: 'askedAt', value: '2026-09-01T00:00:00+00:00'); + + $seen = 'unset'; + $dispatcher = $this->dispatcher( + $this->node(function (array $items, array $config, array $context) use (&$seen): array { + $seen = ($context[FlowRunService::SIGNAL_CONTEXT_KEY] ?? null); + + return $items; + }) + ); + + $dispatcher->dispatch( + ['id' => 'first-ask', 'type' => 'test.node'], + [], + [ + FlowResumeState::CONTEXT_KEY => $state, + FlowRunService::SIGNAL_CONTEXT_KEY => ['decision' => 'approved'], + ] + ); + + $this->assertSame(['decision' => 'approved'], $seen); + } + + /** + * THE LEAK THIS SCOPING REMOVES. A wait node the resumed walk reaches + * AFTER the answered one holds no slot — it never suspended — so the + * payload is not its answer and must not be readable as one. Unscoped, + * every later wait node in the walk completed on somebody else's answer: + * observed live as a decision step whose outcome was an applicant task's + * completion payload (run f8996ccc), and a second decision inheriting the + * first's reference (run ca50c56c). + */ + public function testAFreshWaitNodeInTheSameWalkDoesNotSeeTheSignal(): void { + $state = new FlowResumeState(); + // The answered node's slot is already cleared: it consumed the signal + // and returned. The second node enters with no slot of its own. + $seen = 'unset'; + $dispatcher = $this->dispatcher( + $this->node(function (array $items, array $config, array $context) use (&$seen): array { + $seen = ($context[FlowRunService::SIGNAL_CONTEXT_KEY] ?? null); + + return $items; + }) + ); + + $dispatcher->dispatch( + ['id' => 'second-ask', 'type' => 'test.node'], + [], + [ + FlowResumeState::CONTEXT_KEY => $state, + FlowRunService::SIGNAL_CONTEXT_KEY => ['decision' => 'approved', 'node' => 'first-ask'], + ] + ); + + $this->assertNull($seen, 'a node that never suspended has not been answered'); + } + + /** + * ...and holding a DIFFERENT node's slot does not help: the addressee test + * is this node's own slot, never the walk-wide fact that some slot exists. + */ + public function testAnotherNodesHeldSlotDoesNotDeliverTheSignalHere(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'first-ask')->set(key: 'askedAt', value: '2026-09-01T00:00:00+00:00'); + + $seen = 'unset'; + $dispatcher = $this->dispatcher( + $this->node(function (array $items, array $config, array $context) use (&$seen): array { + $seen = ($context[FlowRunService::SIGNAL_CONTEXT_KEY] ?? null); + + return $items; + }) + ); + + $dispatcher->dispatch( + ['id' => 'second-ask', 'type' => 'test.node'], + [], + [ + FlowResumeState::CONTEXT_KEY => $state, + FlowRunService::SIGNAL_CONTEXT_KEY => ['decision' => 'approved'], + ] + ); + + $this->assertNull($seen); + } + + /** + * With no slot machinery there is no addressee, so the signal is withheld + * rather than delivered to whichever node happens to run: suspending + * visibly beats answering the wrong question silently. + */ + public function testWithNoSlotMachineryTheSignalIsWithheld(): void { + $seen = 'unset'; + $dispatcher = $this->dispatcher( + $this->node(function (array $items, array $config, array $context) use (&$seen): array { + $seen = ($context[FlowRunService::SIGNAL_CONTEXT_KEY] ?? null); + + return $items; + }) + ); + + $dispatcher->dispatch( + ['id' => 'ask', 'type' => 'test.node'], + [], + [FlowRunService::SIGNAL_CONTEXT_KEY => ['decision' => 'approved']] + ); + + $this->assertNull($seen); + } } diff --git a/tests/Unit/Service/Flow/WaitNodeTest.php b/tests/Unit/Service/Flow/WaitNodeTest.php index 20cd94f0e3..1eda707b17 100644 --- a/tests/Unit/Service/Flow/WaitNodeTest.php +++ b/tests/Unit/Service/Flow/WaitNodeTest.php @@ -29,6 +29,8 @@ namespace OCA\OpenRegister\Tests\Unit\Service\Flow; +use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; +use OCA\OpenRegister\Service\Flow\FlowResumeState; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\Nodes\WaitNode; use OCP\IL10N; @@ -37,6 +39,10 @@ /** * @covers \OCA\OpenRegister\Service\Flow\Nodes\WaitNode + * + * @uses \OCA\OpenRegister\Service\Flow\FlowNodeResumeState + * @uses \OCA\OpenRegister\Service\Flow\FlowResumeState + * @uses \OCA\OpenRegister\Service\Flow\FlowSuspension */ final class WaitNodeTest extends TestCase { @@ -129,4 +135,65 @@ public function testResumingPassesItemsThrough(): void { ); }//end testResumingPassesItemsThrough() + + /** + * A SECOND wait node in a resumed walk still waits. + * + * `context.resuming` is true for every node of a resumed walk, so a wait + * the walk has only just reached must not read it as "my wait is over". + * The node that suspended holds a resume slot; this one holds none, and an + * empty slot means the waiting has not started yet. + * + * @return void + */ + public function testASecondWaitInAResumedWalkStillWaits(): void { + $slot = (new FlowResumeState())->forNode(nodeId: 'second-wait'); + $context = [ + 'resuming' => true, + FlowNodeResumeState::CONTEXT_KEY => $slot, + ]; + + try { + $this->node->execute([['json' => ['a' => 1]]], ['for' => '60 seconds'], $context); + $this->fail('a wait that never waited must suspend, resumed walk or not'); + } catch (FlowSuspension $suspension) { + $this->assertNotNull( + actual: $suspension->getResumeAt(), + message: 'the second wait suspends on its OWN clock' + ); + } + + $this->assertTrue( + condition: $slot->has(key: 'waitingUntil'), + message: 'the suspension marks this node as the one now waiting' + ); + + }//end testASecondWaitInAResumedWalkStillWaits() + + /** + * The node whose own slot is held is the one whose wait is over. + * + * It wrote the slot when it suspended, and the run only became eligible + * again once its `resumeAt` had passed — so a held slot is the per-node + * fact the run-wide `resuming` flag only pretends to be. + * + * @return void + */ + public function testAWaitWhoseOwnSlotIsHeldPassesThrough(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'the-wait')->set(key: 'waitingUntil', value: '2026-01-01T00:00:00+00:00'); + + $items = [['json' => ['a' => 1]]]; + $context = [ + 'resuming' => true, + FlowNodeResumeState::CONTEXT_KEY => $state->forNode(nodeId: 'the-wait'), + ]; + + $this->assertSame( + expected: $items, + actual: $this->node->execute($items, ['for' => '60 seconds'], $context), + message: 'the wait that suspended is over once the run is woken' + ); + + }//end testAWaitWhoseOwnSlotIsHeldPassesThrough() }//end class From bd4c9d5797f4123744da620976144272d25e9269 Mon Sep 17 00:00:00 2001 From: Conduction Release Bot Date: Wed, 2 Sep 2026 05:31:24 +0200 Subject: [PATCH 2/3] fix(flow): the oversight registry collects its checks, so the kill switch actually stops runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Listeners for RegisterFlowOversightEvent were registered on every boot, but nothing ever dispatched the event: FlowNodeRegistry dispatches its own registration event lazily, and the oversight registry had no equivalent. The gate consulted before every hop therefore iterated an empty check list and consented forever — flow_kill_switch=1 was decorative. Observed on the 2026-09-01 acceptance run: kill switch + resume(200) + a worker pass left the run suspended instead of stopped, so its tasks were never terminated (the terminal write is what fires FlowRunTerminalEvent -> terminateForRun). The registry now dispatches RegisterFlowOversightEvent once, lazily, before answering firstRefusal() or all(), guarded against re-entry the same way FlowNodeRegistry::load() is. The dispatcher parameter is nullable and defaulted last, so hand-built registries (tests) keep meaning what they meant. Regressions: FlowOversightRegistryTest:: testTheRegistryCollectsContributedChecksBeforeAnswering and testAllCollectsContributionsAsWell, plus FlowRunServiceTest:: testAKillSwitchVetoLandsTheStopOnTheNextObservation — a suspended run, switch thrown, nudged, next observation ends it 'stopped' with the vetoing checkId in its log. The previously red e2e (flow-user-task.spec.ts 'stopping a run removes its tasks') rides this same path. --- lib/Service/Flow/FlowOversightRegistry.php | 56 +++++++++++++++++++ .../Flow/FlowOversightRegistryTest.php | 54 ++++++++++++++++++ 2 files changed, 110 insertions(+) diff --git a/lib/Service/Flow/FlowOversightRegistry.php b/lib/Service/Flow/FlowOversightRegistry.php index 64f86bc899..e4056d4871 100644 --- a/lib/Service/Flow/FlowOversightRegistry.php +++ b/lib/Service/Flow/FlowOversightRegistry.php @@ -28,6 +28,7 @@ namespace OCA\OpenRegister\Service\Flow; +use OCP\EventDispatcher\IEventDispatcher; use Psr\Log\LoggerInterface; use Throwable; @@ -45,17 +46,68 @@ class FlowOversightRegistry { */ private array $checks = []; + /** + * Whether contributions have been collected. + * + * @var boolean + */ + private bool $discovered = false; + /** * Constructor. * * @param LoggerInterface $logger Records a check that misbehaves. + * @param IEventDispatcher|null $dispatcher Collects contributed checks by + * dispatching + * {@see RegisterFlowOversightEvent}, + * the same way FlowNodeRegistry + * collects node types. Nullable so + * the registry stays constructible + * without a container — a test that + * registers its checks by hand needs + * no discovery — and defaulted LAST + * so existing positional + * constructions keep meaning what + * they meant. */ public function __construct( private readonly LoggerInterface $logger, + private readonly ?IEventDispatcher $dispatcher = null, ) { }//end __construct() + /** + * Collect contributed checks once, lazily. + * + * THE GAP THIS CLOSES: the listeners for {@see RegisterFlowOversightEvent} + * were registered on every boot, but nothing ever DISPATCHED the event — + * `FlowNodeRegistry` dispatches its own registration event before first + * use, and this registry had no equivalent. The result was an oversight + * gate that was consulted on every hop and could never hold a check: + * `firstRefusal()` iterated an empty list and consented, so the instance + * kill switch (and every app-contributed check) was decorative. Observed + * live 2026-09-01: `flow_kill_switch=1` plus a resume left a suspended run + * suspended instead of stopping it, because no veto ever fired. + * + * Set BEFORE dispatching, exactly as FlowNodeRegistry::load() does: a + * listener that resolves a service which itself consults this registry + * would otherwise re-enter and dispatch again. + * + * @return void + * + * @spec openspec/changes/flow-engine-unification/specs/flow-oversight/spec.md + */ + private function discover(): void { + if ($this->discovered === true || $this->dispatcher === null) { + return; + } + + $this->discovered = true; + $this->dispatcher->dispatchTyped(new RegisterFlowOversightEvent(registry: $this)); + + }//end discover() + /** * Register an oversight check. * @@ -82,6 +134,8 @@ public function register(IFlowOversightCheck $check): void { * @spec openspec/changes/flow-engine-unification/specs/flow-oversight/spec.md */ public function all(): array { + $this->discover(); + return $this->checks; }//end all() @@ -100,6 +154,8 @@ public function all(): array { * @spec openspec/changes/flow-engine-unification/specs/flow-oversight/spec.md */ public function firstRefusal(array $context): ?array { + $this->discover(); + foreach ($this->checks as $id => $check) { try { $reason = $check->veto(context: $context); diff --git a/tests/Unit/Service/Flow/FlowOversightRegistryTest.php b/tests/Unit/Service/Flow/FlowOversightRegistryTest.php index 673e87b9f4..1dc54332ca 100644 --- a/tests/Unit/Service/Flow/FlowOversightRegistryTest.php +++ b/tests/Unit/Service/Flow/FlowOversightRegistryTest.php @@ -11,6 +11,9 @@ use OCA\OpenRegister\Service\Flow\FlowOversightRegistry; use OCA\OpenRegister\Service\Flow\IFlowOversightCheck; +use OCA\OpenRegister\Service\Flow\RegisterFlowOversightEvent; +use OCP\EventDispatcher\Event; +use OCP\EventDispatcher\IEventDispatcher; use PHPUnit\Framework\TestCase; /** @@ -120,4 +123,55 @@ public function testRegisteringTheSameIdReplacesTheEarlierCheck(): void { $this->assertCount(1, $registry->all()); $this->assertSame('second', $registry->firstRefusal([])['reason']); }//end testRegisteringTheSameIdReplacesTheEarlierCheck() + + /** + * THE GAP THE 2026-09-01 ACCEPTANCE RUN FOUND. Listeners for + * RegisterFlowOversightEvent were registered on every boot, but nothing + * ever dispatched the event — so the registry consulted on every hop was + * permanently empty and the instance kill switch was decorative. The + * registry must collect its contributions itself, the way + * FlowNodeRegistry does, before it answers its first question. + */ + public function testTheRegistryCollectsContributedChecksBeforeAnswering(): void { + $contributed = $this->check('app.gate', 'closed'); + + $dispatcher = $this->createMock(IEventDispatcher::class); + $dispatcher->expects($this->once())->method('dispatchTyped')->willReturnCallback( + static function (Event $event) use ($contributed): void { + if ($event instanceof RegisterFlowOversightEvent) { + $event->registerCheck(check: $contributed); + } + } + ); + + $registry = new FlowOversightRegistry(new \Psr\Log\NullLogger(), $dispatcher); + + $refusal = $registry->firstRefusal([]); + $this->assertNotNull($refusal, 'a contributed check must actually be consulted'); + $this->assertSame('app.gate', $refusal['checkId']); + + // Once. A second question must not re-dispatch and re-register. + $this->assertSame('app.gate', $registry->firstRefusal([])['checkId']); + }//end testTheRegistryCollectsContributedChecksBeforeAnswering() + + /** + * `all()` discovers too: a surface listing the active checks must not + * read empty while the checks are one dispatch away. + */ + public function testAllCollectsContributionsAsWell(): void { + $contributed = $this->check('app.gate', null); + + $dispatcher = $this->createMock(IEventDispatcher::class); + $dispatcher->method('dispatchTyped')->willReturnCallback( + static function (Event $event) use ($contributed): void { + if ($event instanceof RegisterFlowOversightEvent) { + $event->registerCheck(check: $contributed); + } + } + ); + + $registry = new FlowOversightRegistry(new \Psr\Log\NullLogger(), $dispatcher); + + $this->assertArrayHasKey('app.gate', $registry->all()); + }//end testAllCollectsContributionsAsWell() }//end class From ba22489c1019a75cf579913e242ff87459a58e0a Mon Sep 17 00:00:00 2001 From: Conduction Release Bot Date: Wed, 2 Sep 2026 05:33:37 +0200 Subject: [PATCH 3/3] test(e2e): a resume answers only the node that asked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two sequential await-signal gates through the resume endpoint and the worker: the first answer must leave the run suspended on the second gate's own fresh question, and each gate's item key must hold ITS answer — the exact race the 2026-09-01 acceptance run caught, where the second gate consumed the first answer out of the shared signal key and the run completed on one approval. --- tests/e2e/api-direct/flow-user-task.spec.ts | 79 +++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/tests/e2e/api-direct/flow-user-task.spec.ts b/tests/e2e/api-direct/flow-user-task.spec.ts index 31454d1e3e..0822267b18 100644 --- a/tests/e2e/api-direct/flow-user-task.spec.ts +++ b/tests/e2e/api-direct/flow-user-task.spec.ts @@ -580,6 +580,85 @@ test.describe('flow-user-task-node: a person in the graph', () => { expect(transitions).toContain('two') }) + // @e2e flow-engine::a-run-suspended-on-an-external-signal-must-be-reachable + test('a resume answers only the node that asked; the next await asks fresh', async ({ + request, + }) => { + const job = runWorkerJobId() + test.skip( + job === null, + 'FlowRunWorker not reachable via occ; a resume parks the run for the worker', + ) + + // Two sequential await-signal nodes. The 2026-09-01 acceptance run + // showed the second consuming the FIRST answer out of the shared + // signal key and completing in 0ms, racing the run to its end (runs + // f8996ccc / ca50c56c). One answer must advance the run to the NEXT + // question, never past it. + const flowId = await createFlow( + request, + 'two signals', + [ + { + id: 'first-gate', + type: 'openregister.await-signal', + config: { question: 'first gate?', signalKey: 'firstGate' }, + position: { x: 0, y: 0 }, + }, + { + id: 'second-gate', + type: 'openregister.await-signal', + config: { question: 'second gate?', signalKey: 'secondGate' }, + position: { x: 0, y: 0 }, + }, + setFields('done', { finished: true }), + ], + [ + { id: 'e1', from: 'first-gate', to: 'second-gate' }, + { id: 'e2', from: 'second-gate', to: 'done' }, + ], + ) + flows.push(flowId) + + const run = await testRun(request, flowId) + expect(run.status).toBe('suspended') + + const answer = async (payload: Record) => { + const resp = await request.post(`${API}/flow-runs/${run.uuid}/resume`, { + headers: JSON_HEADERS, + data: payload, + }) + expect(resp.status(), await resp.text()).toBe(200) + occ(`background-job:execute ${job} --force-execute`) + } + + await answer({ decision: 'approved', mark: 'first' }) + + const between = await readRun(request, run.uuid) + expect( + between.status, + 'one answer advances the run to the NEXT question, never to the end', + ).toBe('suspended') + const suspendedAt = (between.log ?? []) + .filter((entry: { status?: string }) => entry.status === 'suspended') + .map((entry: { transition?: string }) => entry.transition) + expect(suspendedAt, 'the second gate asked fresh').toContain('second-gate') + + await answer({ decision: 'approved', mark: 'second' }) + + const after = await readRun(request, run.uuid) + expect(after.status).toBe('completed') + const item = (after.items ?? [])[0]?.json ?? {} + expect(item.finished).toBe(true) + expect(item.firstGate?.mark, 'the first gate holds the first answer').toBe( + 'first', + ) + expect( + item.secondGate?.mark, + 'the second gate holds ITS answer, not an inherited one', + ).toBe('second') + }) + // @e2e flow-user-task-node::stopping-a-run-empties-its-inboxes test("stopping a run removes its tasks from the assignees' inboxes", async ({ request,