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/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/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 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 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,