feat(flow): each step records when it started, and a node can write log lines

lib/Service/Flow/FlowEngine.php

@@ -38,6 +38,8 @@
 
 namespace OCA\OpenRegister\Service\Flow;
 
+use DateTimeImmutable;
+use DateTimeZone;
 use InvalidArgumentException;
 use OCA\OpenRegister\Db\FlowRun;
 use Psr\Log\LoggerInterface;
@@ -483,6 +485,7 @@ public function run(
 				$items = $pinned;
 				$log[] = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'pinned',
 					'itemsIn' => count($itemsIn),
@@ -541,6 +544,7 @@ public function run(
 				$items = FlowItems::normalise(value: $produced);
 				$entry = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'completed',
 					'itemsIn' => count($itemsIn),
@@ -572,6 +576,7 @@ public function run(
 				// to end, and it ends with their message and their outcome.
 				$log[] = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'stopped',
 					'reason' => $stop->getMessage(),
@@ -604,6 +609,7 @@ public function run(
 				// transition, re-entering the step that asked to wait.
 				$log[] = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'status' => 'suspended',
 					'reason' => $suspension->getMessage(),
 				];
@@ -618,6 +624,7 @@ public function run(
 			} catch (Throwable $e) {
 				$entry = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'failed',
 					'error' => $e->getMessage(),
@@ -852,6 +859,7 @@ private function walkStreams(
 				$items = $pinned;
 				$entry = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'pinned',
 					'itemsIn' => count($itemsIn),
@@ -887,6 +895,7 @@ private function walkStreams(
 				$items = FlowItems::normalise(value: $produced);
 				$entry = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'completed',
 					'itemsIn' => count($itemsIn),
@@ -903,6 +912,7 @@ private function walkStreams(
 				$streams->release(places: $claimed);
 				$log[] = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'stopped',
 					'reason' => $stop->getMessage(),
@@ -928,6 +938,7 @@ private function walkStreams(
 				// resumes ON this transition — and its siblings keep going.
 				$log[] = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'status' => 'suspended',
 					'reason' => $suspension->getMessage(),
 					'streamId' => $streamId,
@@ -952,6 +963,7 @@ private function walkStreams(
 			} catch (Throwable $e) {
 				$entry = [
 					'transition' => $name,
+					'startedAt' => $this->moment(microtime: $startedAt),
 					'type' => $stepType,
 					'status' => 'failed',
 					'error' => $e->getMessage(),
@@ -1292,6 +1304,27 @@ private function stepReport(array $context): array {
 		return $report->take();
 	}//end stepReport()
 
+	/**
+	 * A `microtime(true)` moment as ISO 8601 with milliseconds, for the run log.
+	 *
+	 * A step's own start, so a run view can say when each step ran rather
+	 * than adding up durations from the run's start.
+	 *
+	 * @param float $microtime The moment, as `microtime(true)` returns it.
+	 *
+	 * @return string The moment, UTC, e.g. `2026-10-07T08:41:02.380+00:00`.
+	 *
+	 * @spec openspec/changes/flow-step-log-lines/specs/flow-step-log-lines/spec.md
+	 */
+	private function moment(float $microtime): string {
+		$moment = DateTimeImmutable::createFromFormat('U.u', sprintf('%.6F', $microtime), new DateTimeZone('UTC'));
+		if ($moment === false) {
+			return gmdate('Y-m-d\TH:i:s+00:00', (int)$microtime);
+		}
+
+		return $moment->format('Y-m-d\TH:i:s.vP');
+	}//end moment()
+
 	/**
 	 * A bounded, honest sample of an item list, for the run log.
 	 *

lib/Service/Flow/FlowStepReport.php

@@ -32,6 +32,9 @@
 
 namespace OCA\OpenRegister\Service\Flow;
 
+use DateTimeImmutable;
+use DateTimeZone;
+
 /**
  * Collects a node's own log detail for the current hop.
  */
@@ -42,13 +45,30 @@ class FlowStepReport {
 	 */
 	public const CONTEXT_KEY = '_stepReport';
 
+	/**
+	 * The most lines one step keeps in its report.
+	 */
+	public const MAX_LINES = 50;
+
+	/**
+	 * The levels a line may carry.
+	 */
+	public const LEVELS = ['info', 'success', 'warning', 'error'];
+
 	/**
 	 * The detail written by the current hop's node.
 	 *
 	 * @var array<string, mixed>
 	 */
 	private array $detail = [];
 
+	/**
+	 * Lines left out of the current hop past {@see self::MAX_LINES}.
+	 *
+	 * @var integer
+	 */
+	private int $dropped = 0;
+
 	/**
 	 * Record one piece of detail for the current step's log entry.
 	 *
@@ -66,6 +86,68 @@ public function put(string $key, mixed $value): void {
 		$this->detail[$key] = $value;
 	}//end put()
 
+	/**
+	 * Record one timestamped line for the current step, under `report.lines`.
+	 *
+	 * What a person reads in a run view: "looked up the account", "no
+	 * account, posted nothing". `level` is `info`, `success`, `warning` or
+	 * `error`; anything else is recorded as `info`. At most
+	 * {@see self::MAX_LINES} lines are kept per step, and the last one then
+	 * says how many were left out, so a node writing in a loop cannot grow
+	 * the log without bound.
+	 *
+	 * @param string $level   The line's level.
+	 * @param string $message What happened, in plain words.
+	 *
+	 * @return void
+	 *
+	 * @spec openspec/changes/flow-step-log-lines/specs/flow-step-log-lines/spec.md
+	 */
+	public function line(string $level, string $message): void {
+		$lines = ($this->detail['lines'] ?? []);
+		if (is_array($lines) === false) {
+			$lines = [];
+		}
+
+		if (count($lines) >= self::MAX_LINES) {
+			$this->dropped++;
+			$lines[self::MAX_LINES - 1] = [
+				'at' => $this->now(),
+				'level' => 'info',
+				'message' => sprintf('%d more lines left out.', $this->dropped + 1),
+			];
+			$this->detail['lines'] = $lines;
+			return;
+		}
+
+		if (in_array($level, self::LEVELS, true) === false) {
+			$level = 'info';
+		}
+
+		$lines[] = [
+			'at' => $this->now(),
+			'level' => $level,
+			'message' => $message,
+		];
+		$this->detail['lines'] = $lines;
+	}//end line()
+
+	/**
+	 * The current moment as ISO 8601 with milliseconds, UTC.
+	 *
+	 * @return string The moment.
+	 *
+	 * @spec openspec/changes/flow-step-log-lines/specs/flow-step-log-lines/spec.md
+	 */
+	private function now(): string {
+		$moment = DateTimeImmutable::createFromFormat('U.u', sprintf('%.6F', microtime(true)), new DateTimeZone('UTC'));
+		if ($moment === false) {
+			return gmdate('Y-m-d\TH:i:s+00:00');
+		}
+
+		return $moment->format('Y-m-d\TH:i:s.vP');
+	}//end now()
+
 	/**
 	 * Return and clear the current hop's detail.
 	 *
@@ -79,6 +161,7 @@ public function put(string $key, mixed $value): void {
 	public function take(): array {
 		$detail = $this->detail;
 		$this->detail = [];
+		$this->dropped = 0;
 
 		return $detail;
 	}//end take()

openspec/changes/flow-step-log-lines/proposal.md

@@ -0,0 +1,31 @@
+---
+kind: code
+---
+
+# Proposal: flow-step-log-lines
+
+## Summary
+
+A run view shows, per step, when it started and what it said: "08:41:02.380
+looked up the account", "08:41:02.441 no account on the portal". The run log
+has neither today: one entry per step with a duration, and a free `report`
+map. This change adds `startedAt` (ISO 8601, milliseconds, UTC) to every step
+entry the engine writes, and `FlowStepReport::line(level, message)`, which a
+node calls to append a timestamped line to its entry under `report.lines`.
+
+## Why
+
+The shared run view in `@conduction/nextcloud-vue` (boards PqFlowRun and
+PqFlow) draws a step's log lines with an outcome pill and a time. Without a
+start per step the client can only add durations up from the run's start,
+and without lines it can only restate the step's status.
+
+## Scope
+
+- `FlowEngine`: `startedAt` on every step entry (completed, failed, stopped,
+  suspended, pinned), on both walks.
+- `FlowStepReport::line()`: levels `info`, `success`, `warning`, `error`
+  (anything else is `info`), at most 50 lines per step, the last saying how
+  many were left out.
+- Nodes adopt `line()` in follow-up changes per app; nothing breaks for a node
+  that does not.

openspec/changes/flow-step-log-lines/specs/flow-step-log-lines/spec.md

@@ -0,0 +1,29 @@
+## Purpose
+
+Each step of a run says when it started and, in its own words, what it did.
+
+## ADDED Requirements
+
+### Requirement: Every step entry records when the step started
+
+Every log entry the engine writes for a step SHALL carry `startedAt`, the
+moment the step began, as ISO 8601 with milliseconds in UTC. This SHALL hold
+for every step status, including a pinned step and a suspended one.
+
+#### Scenario: a failed step still says when it started
+- GIVEN a two-step flow whose second step throws
+- WHEN the flow runs
+- THEN both log entries carry `startedAt`, the first no later than the second
+
+### Requirement: A node can write timestamped lines into its step's entry
+
+`FlowStepReport::line(level, message)` SHALL append `{at, level, message}` to
+the current step's `report.lines`. `level` SHALL be one of `info`, `success`,
+`warning`, `error`; any other value SHALL be recorded as `info`. A step SHALL
+keep at most 50 lines; past that the last line SHALL say how many lines were
+left out. Lines SHALL clear with the rest of the report after each hop.
+
+#### Scenario: the cap is stated, not silent
+- GIVEN a node that writes 53 lines in one step
+- WHEN the step's report is taken
+- THEN it holds 50 lines and the last reads "4 more lines left out."

openspec/changes/flow-step-log-lines/tasks.md

@@ -0,0 +1,6 @@
+# Tasks: flow-step-log-lines
+
+- [x] `FlowEngine`: `startedAt` on every step entry, both walks
+- [x] `FlowStepReport::line()` with levels, cap and per-hop reset
+- [x] Unit tests: engine entries carry `startedAt`; report lines, levels, cap, reset
+- [ ] Nodes write lines (send, portal post, external call), per app, in follow-ups

tests/Unit/Service/Flow/FlowEngineTest.php

@@ -11,6 +11,9 @@
  * SPDX-License-Identifier: EUPL-1.2
  */
 
+// phpcs:disable PEAR.Commenting.FunctionComment.Missing -- arrange/act/assert PHPUnit conventions.
+// phpcs:disable CustomSniffs.Functions.NamedParameters.RequireNamedParameters -- PHPUnit assertion helpers use positional args.
+
 namespace Unit\Service\Flow;
 
 use OCA\OpenRegister\Service\Flow\FlowDefinitionBuilder;
@@ -157,6 +160,19 @@ public function testEachStepIsTracedInTheRunLog(): void {
 		}
 	}//end testEachStepIsTracedInTheRunLog()
 
+	public function testEveryStepEntryCarriesWhenItStarted(): void {
+		$before = microtime(true);
+		$result = $this->runFlow($this->linearFlow(), new RecordingDispatcher(failOn: 'second'));
+
+		$this->assertCount(2, $result['log']);
+		foreach ($result['log'] as $entry) {
+			$this->assertMatchesRegularExpression('/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}\+00:00$/', $entry['startedAt']);
+			$this->assertGreaterThanOrEqual((int)floor($before), (new \DateTimeImmutable($entry['startedAt']))->getTimestamp());
+		}
+
+		$this->assertLessThanOrEqual($result['log'][1]['startedAt'], $result['log'][0]['startedAt']);
+	}//end testEveryStepEntryCarriesWhenItStarted()
+
 	public function testItemsAreThreadedFromEachStepIntoTheNext(): void {
 		$dispatcher = new RecordingDispatcher();
 		$result = $this->runFlow($this->linearFlow(), $dispatcher);
@@ -401,6 +417,7 @@ public function testAPinnedStepIsNotExecutedAndItsOutputIsUsed(): void {
 		// Only 'second' ran — 'first' was served from the pin.
 		$this->assertSame(['second'], $dispatcher->dispatched);
 		$this->assertSame('pinned', $result['log'][0]['status']);
+		$this->assertArrayHasKey('startedAt', $result['log'][0]);
 		// The next step saw the pinned output, not a re-execution.
 		$this->assertTrue($dispatcher->seenItems['second'][0]['json']['pinned']);
 	}//end testAPinnedStepIsNotExecutedAndItsOutputIsUsed()