Skip to content

Commit 3306d2f

Browse files
xuyushun441-sysos-zhuangclaude
authored
feat(automation): surface structured-region body steps in run observability (#1505) (#1580)
loop / parallel / try_catch ran their body, branch and handler regions against a region-local step log that was discarded, so listRuns/getRun showed the container as a single opaque step. runRegion() now RETURNS its body steps and the container folds them into the parent run log via a new NodeExecutionResult.childSteps field, appended right after the container's own step. Each surfaced step is tagged with its immediate container via three new optional fields on ExecutionStepLogSchema (and the engine's StepLogEntry): - parentNodeId: enclosing loop/parallel/try_catch node - iteration: zero-based loop iteration or parallel branch index - regionKind: loop-body | parallel-branch | try | catch Tagging fills only undefined fields, so nested regions keep each step's innermost container. Failed try-attempt partial steps stay unsurfaced (preserves try_catch retry semantics). Fully additive. +4 tests across loop/parallel/try_catch. Resolves the ADR-0031 #1505 / TODO(#1479) follow-up. Co-authored-by: Jack Zhuang <277994282+os-zhuang@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 8183e0a commit 3306d2f

10 files changed

Lines changed: 191 additions & 17 deletions

File tree

.changeset/region-step-logs.md

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
---
2+
'@objectstack/spec': patch
3+
'@objectstack/service-automation': patch
4+
---
5+
6+
feat(automation): surface structured-region body steps in run observability (#1505)
7+
8+
`loop` / `parallel` / `try_catch` previously ran their body, branch, and handler
9+
regions against a region-local step log that was **discarded** — run logs
10+
(`listRuns` / `getRun`) showed the container as a single opaque step, hiding the
11+
per-iteration / per-branch steps that actually executed.
12+
13+
`AutomationEngine.runRegion()` now **returns** its body steps, and the container
14+
node folds them into the parent run log via a new `NodeExecutionResult.childSteps`
15+
field. Each surfaced step is tagged with its **immediate** container via three new
16+
optional fields on `ExecutionStepLogSchema` (and the engine's `StepLogEntry`):
17+
18+
- `parentNodeId` — the enclosing `loop` / `parallel` / `try_catch` node
19+
- `iteration` — zero-based loop iteration or parallel branch index
20+
- `regionKind``loop-body` | `parallel-branch` | `try` | `catch`
21+
22+
Tagging fills only fields left undefined, so nested regions keep each step's
23+
innermost container. A failed try-region attempt's partial steps are still not
24+
surfaced (preserving `try_catch` retry semantics). Fully additive — existing run
25+
logs and consumers are unaffected.

docs/adr/0031-advanced-flow-node-executors-and-dag.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# ADR-0031: Structured control-flow for flows (loop / parallel / try-catch) — native + AI-authored, BPMN as interop
22

3-
**Status**: Accepted (2026-06-02) — implemented in #1482 (spec + loop), #1489 (parallel), #1499 (try/catch/retry), #1500 (BPMN mapping); docs #1497. Follow-ups: #1504 (BPMN XML plugin), #1505 (region step logs).
3+
**Status**: Accepted (2026-06-02) — implemented in #1482 (spec + loop), #1489 (parallel), #1499 (try/catch/retry), #1500 (BPMN mapping); docs #1497. Follow-ups: #1504 (BPMN XML plugin); ~~#1505 (region step logs)~~ **done**`runRegion` now returns its body steps and the container folds them into the run log via `NodeExecutionResult.childSteps`, tagged with `parentNodeId` / `iteration` / `regionKind` (`ExecutionStepLogSchema`).
44
**Deciders**: ObjectStack Protocol Architects
55
**Builds on**: [ADR-0018](./0018-unified-node-action-registry.md) (open action registry — node types are an open vocabulary, executors are the source of truth), [ADR-0019](./0019-approval-as-flow-node.md) (durable-pause node via suspend/resume), [ADR-0010](./0010-nl-to-flow-authoring.md) + [ADR-0011](./0011-actions-as-ai-tools.md) (AI flow authoring — **the design center**)
66
**Consumers**: `@objectstack/services/service-automation` (engine + builtin executors), `@objectstack/spec` (`automation/flow.zod.ts`, `automation/bpmn-interop.zod.ts`, `studio/flow-builder.zod.ts`), `../objectui` (flow designer)

packages/services/service-automation/src/builtin/loop-node.test.ts

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,34 @@ describe('loop container executor (ADR-0031)', () => {
9292
]);
9393
});
9494

95+
it('surfaces each iteration\'s body steps in the run log, tagged with parentNodeId + index (#1479)', async () => {
96+
engine.registerFlow('loop_flow', loopFlow(
97+
{
98+
collection: '{items}',
99+
iteratorVariable: 'item',
100+
indexVariable: 'i',
101+
body: { nodes: [{ id: 'b', type: 'collect', label: 'Body', config: { itemVar: 'item', idxVar: 'i' } }], edges: [] },
102+
},
103+
{ items: ['a', 'b', 'c'] },
104+
));
105+
106+
await engine.execute('loop_flow');
107+
const runs = await engine.listRuns('loop_flow');
108+
const bodySteps = runs[0].steps.filter(s => s.nodeId === 'b');
109+
110+
// One body step per iteration, each tagged with the loop container + index.
111+
expect(bodySteps).toHaveLength(3);
112+
expect(bodySteps.map(s => s.iteration)).toEqual([0, 1, 2]);
113+
expect(bodySteps.every(s => s.parentNodeId === 'loop')).toBe(true);
114+
expect(bodySteps.every(s => s.regionKind === 'loop-body')).toBe(true);
115+
116+
// The container step and the after-loop step stay un-grouped (top level).
117+
const loopStep = runs[0].steps.find(s => s.nodeId === 'loop');
118+
const afterStep = runs[0].steps.find(s => s.nodeId === 'after');
119+
expect(loopStep?.parentNodeId).toBeUndefined();
120+
expect(afterStep?.parentNodeId).toBeUndefined();
121+
});
122+
95123
it('runs a multi-node body region in order each iteration', async () => {
96124
engine.registerFlow('loop_flow', loopFlow(
97125
{

packages/services/service-automation/src/builtin/loop-node.ts

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import type { PluginContext } from '@objectstack/core';
44
import { defineActionDescriptor, LOOP_MAX_ITERATIONS_CEILING } from '@objectstack/spec/automation';
55
import type { FlowRegionParsed } from '@objectstack/spec/automation';
66
import type { AutomationContext } from '@objectstack/spec/contracts';
7-
import type { AutomationEngine } from '../engine.js';
7+
import type { AutomationEngine, StepLogEntry } from '../engine.js';
88
import { interpolate } from './template.js';
99

1010
/**
@@ -110,15 +110,22 @@ export function registerLoopNode(engine: AutomationEngine, ctx: PluginContext):
110110
}
111111

112112
let iterations = 0;
113+
const childSteps: StepLogEntry[] = [];
113114
for (let i = 0; i < collection.length; i++) {
114115
variables.set(iteratorVariable, collection[i]);
115116
if (indexVariable) variables.set(indexVariable, i);
116117
// Body runs in the shared scope; iterator var + mutations are visible.
117-
await engine.runRegion(body, variables, context ?? ({} as AutomationContext));
118+
// #1479: collect each iteration's body steps, tagged with the index.
119+
const iterSteps = await engine.runRegion(body, variables, context ?? ({} as AutomationContext), {
120+
parentNodeId: node.id,
121+
iteration: i,
122+
regionKind: 'loop-body',
123+
});
124+
childSteps.push(...iterSteps);
118125
iterations++;
119126
}
120127

121-
return { success: true, output: { iterations } };
128+
return { success: true, output: { iterations }, childSteps };
122129
},
123130
});
124131

packages/services/service-automation/src/builtin/parallel-node.test.ts

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,25 @@ describe('parallel block executor (ADR-0031)', () => {
8484
expect(order.filter(o => o === 'after')).toHaveLength(1);
8585
});
8686

87+
it('surfaces each branch\'s steps in the run log, tagged with parentNodeId + branch index (#1479)', async () => {
88+
engine.registerFlow('par_flow', parallelFlow([
89+
{ name: 'A', nodes: [{ id: 'a', type: 'setvar', label: 'A', config: { key: 'a', value: 1 } }], edges: [] },
90+
{ name: 'B', nodes: [{ id: 'b', type: 'setvar', label: 'B', config: { key: 'b', value: 2 } }], edges: [] },
91+
]));
92+
93+
await engine.execute('par_flow');
94+
const runs = await engine.listRuns('par_flow');
95+
96+
const stepA = runs[0].steps.find(s => s.nodeId === 'a');
97+
const stepB = runs[0].steps.find(s => s.nodeId === 'b');
98+
expect(stepA?.parentNodeId).toBe('par');
99+
expect(stepA?.iteration).toBe(0);
100+
expect(stepA?.regionKind).toBe('parallel-branch');
101+
expect(stepB?.parentNodeId).toBe('par');
102+
expect(stepB?.iteration).toBe(1);
103+
expect(stepB?.regionKind).toBe('parallel-branch');
104+
});
105+
87106
it('joins only after the slowest branch completes', async () => {
88107
// Branch "slow" awaits several microtasks; "fast" resolves immediately.
89108
// The join ('after') must still be last.

packages/services/service-automation/src/builtin/parallel-node.ts

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import type { PluginContext } from '@objectstack/core';
44
import { defineActionDescriptor } from '@objectstack/spec/automation';
55
import type { FlowRegionParsed } from '@objectstack/spec/automation';
66
import type { AutomationContext } from '@objectstack/spec/contracts';
7-
import type { AutomationEngine } from '../engine.js';
7+
import type { AutomationEngine, StepLogEntry } from '../engine.js';
88

99
/** One branch of a parallel block — a region plus an optional label. */
1010
interface ParallelBranch extends FlowRegionParsed {
@@ -75,17 +75,25 @@ export function registerParallelNode(engine: AutomationEngine, ctx: PluginContex
7575
};
7676
}
7777

78+
let branchSteps: StepLogEntry[][];
7879
try {
7980
// Implicit join: continue once when ALL branches have completed.
80-
await Promise.all(
81-
branches.map(branch => engine.runRegion(branch, variables, context ?? ({} as AutomationContext))),
81+
// #1479: each branch returns its body steps, tagged with the branch index.
82+
branchSteps = await Promise.all(
83+
branches.map((branch, i) =>
84+
engine.runRegion(branch, variables, context ?? ({} as AutomationContext), {
85+
parentNodeId: node.id,
86+
iteration: i,
87+
regionKind: 'parallel-branch',
88+
}),
89+
),
8290
);
8391
} catch (err) {
8492
const message = err instanceof Error ? err.message : String(err);
8593
return { success: false, error: `parallel '${node.id}': branch failed — ${message}` };
8694
}
8795

88-
return { success: true, output: { branches: branches.length } };
96+
return { success: true, output: { branches: branches.length }, childSteps: branchSteps.flat() };
8997
},
9098
});
9199

packages/services/service-automation/src/builtin/try-catch-node.test.ts

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,31 @@ describe('try/catch/retry executor (ADR-0031)', () => {
9898
expect(ran[ran.length - 1]).toBe('after'); // downstream continued after catch
9999
});
100100

101+
it('surfaces the try region steps tagged regionKind=try (#1479)', async () => {
102+
engine.registerFlow('tc_flow', tcFlow({
103+
try: { nodes: [{ id: 't', type: 'ok', label: 'T', config: { tag: 'try' } }], edges: [] },
104+
}));
105+
106+
await engine.execute('tc_flow');
107+
const runs = await engine.listRuns('tc_flow');
108+
const tryStep = runs[0].steps.find(s => s.nodeId === 't');
109+
expect(tryStep?.parentNodeId).toBe('tc');
110+
expect(tryStep?.regionKind).toBe('try');
111+
});
112+
113+
it('surfaces the catch region steps tagged regionKind=catch when the try fails (#1479)', async () => {
114+
engine.registerFlow('tc_flow', tcFlow({
115+
try: { nodes: [{ id: 't', type: 'boom', label: 'T' }], edges: [] },
116+
catch: { nodes: [{ id: 'c', type: 'handler', label: 'C' }], edges: [] },
117+
}));
118+
119+
await engine.execute('tc_flow');
120+
const runs = await engine.listRuns('tc_flow');
121+
const catchStep = runs[0].steps.find(s => s.nodeId === 'c');
122+
expect(catchStep?.parentNodeId).toBe('tc');
123+
expect(catchStep?.regionKind).toBe('catch');
124+
});
125+
101126
it('retries the try region with backoff and succeeds without running catch', async () => {
102127
engine.registerFlow('tc_flow', tcFlow({
103128
try: { nodes: [{ id: 't', type: 'flaky', label: 'T', config: { failTimes: 2 } }], edges: [] },

packages/services/service-automation/src/builtin/try-catch-node.ts

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -104,8 +104,12 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex
104104
if (delay > 0) await new Promise(r => setTimeout(r, delay));
105105
}
106106
try {
107-
await engine.runRegion(tryRegion, variables, ctxOrEmpty);
108-
return { success: true, output: { attempts: attempt + 1, caught: false } };
107+
// #1479: surface the successful try region's steps.
108+
const trySteps = await engine.runRegion(tryRegion, variables, ctxOrEmpty, {
109+
parentNodeId: node.id,
110+
regionKind: 'try',
111+
});
112+
return { success: true, output: { attempts: attempt + 1, caught: false }, childSteps: trySteps };
109113
} catch (err) {
110114
lastError = err instanceof Error ? err.message : String(err);
111115
}
@@ -115,8 +119,16 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex
115119
if (catchRegion != null) {
116120
variables.set(errorVariable, { nodeId: node.id, message: lastError });
117121
try {
118-
await engine.runRegion(catchRegion, variables, ctxOrEmpty);
119-
return { success: true, output: { attempts: maxRetries + 1, caught: true, error: lastError } };
122+
// #1479: surface the catch handler region's steps.
123+
const catchSteps = await engine.runRegion(catchRegion, variables, ctxOrEmpty, {
124+
parentNodeId: node.id,
125+
regionKind: 'catch',
126+
});
127+
return {
128+
success: true,
129+
output: { attempts: maxRetries + 1, caught: true, error: lastError },
130+
childSteps: catchSteps,
131+
};
120132
} catch (catchErr) {
121133
const catchMsg = catchErr instanceof Error ? catchErr.message : String(catchErr);
122134
return { success: false, error: `try_catch '${node.id}': catch region failed — ${catchMsg}` };

packages/services/service-automation/src/engine.ts

Lines changed: 48 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,14 @@ export interface NodeExecutionResult {
7676
* the form and `resume()` with the values.
7777
*/
7878
screen?: ScreenSpec;
79+
/**
80+
* #1479: step logs produced inside the node's structured region(s). A
81+
* container node (`loop` / `parallel` / `try_catch`) collects the
82+
* {@link AutomationEngine.runRegion} return value(s) here; {@link AutomationEngine.executeNode}
83+
* appends them to the parent run log right after the container's own step,
84+
* so per-iteration / per-branch body steps surface in run observability.
85+
*/
86+
childSteps?: StepLogEntry[];
7987
}
8088

8189
// ─── Trigger Interface (Plugin Extension Point) ─────────────────────
@@ -202,6 +210,18 @@ export interface StepLogEntry {
202210
completedAt?: string;
203211
durationMs?: number;
204212
error?: { code: string; message: string; stack?: string };
213+
/**
214+
* #1479: structured-region grouping. When a step ran inside a `loop` /
215+
* `parallel` / `try_catch` body region, these tag it with its **immediate**
216+
* container so run observability can distinguish per-iteration / per-branch
217+
* body steps from top-level ones. Set by {@link AutomationEngine.runRegion}
218+
* (innermost wins — never overwritten as steps bubble through nested regions).
219+
*/
220+
parentNodeId?: string;
221+
/** Zero-based loop iteration or parallel branch index of the enclosing region. */
222+
iteration?: number;
223+
/** Which region kind the step ran in: `loop-body` | `parallel-branch` | `try` | `catch`. */
224+
regionKind?: string;
205225
}
206226

207227
/**
@@ -1471,6 +1491,12 @@ export class AutomationEngine implements IAutomationService {
14711491
durationMs: Date.now() - stepStart,
14721492
});
14731493

1494+
// #1479: fold a structured-region container's body/branch/handler
1495+
// steps into the run log, right after the container's own step.
1496+
if (result.childSteps?.length) {
1497+
steps.push(...result.childSteps);
1498+
}
1499+
14741500
// Write back output variables
14751501
if (result.output) {
14761502
for (const [key, value] of Object.entries(result.output)) {
@@ -1564,8 +1590,13 @@ export class AutomationEngine implements IAutomationService {
15641590
* nodes/edges, so the main DAG traversal (`traverseNext`) is never aware of
15651591
* scope markers — keeping the shared traversal untouched.
15661592
*
1567-
* Body step logs are kept in a region-local array (not yet merged into the
1568-
* parent run log); surfacing per-iteration steps is a follow-up.
1593+
* #1479: the executed body steps are **returned** (tagged with `grouping`)
1594+
* so the calling container node can fold them into the parent run log via
1595+
* `NodeExecutionResult.childSteps`. Tagging only fills fields left undefined,
1596+
* so when regions nest, each step keeps its **innermost** container's
1597+
* `parentNodeId` / `iteration` / `regionKind`. On failure the region throws
1598+
* as before (preserving `try_catch` retry semantics); a failed attempt's
1599+
* partial steps are not surfaced.
15691600
*
15701601
* Durable pause (`suspend`) inside a region is not supported in this
15711602
* iteration — it is converted into a clear error (mirrors the `subflow`
@@ -1575,16 +1606,15 @@ export class AutomationEngine implements IAutomationService {
15751606
region: FlowRegionParsed,
15761607
variables: Map<string, unknown>,
15771608
context: AutomationContext,
1578-
): Promise<void> {
1609+
grouping?: { parentNodeId: string; iteration?: number; regionKind?: string },
1610+
): Promise<StepLogEntry[]> {
15791611
const entryId = findRegionEntry(region);
15801612
const entry = region.nodes.find(n => n.id === entryId);
15811613
if (!entry) {
15821614
throw new Error(`region entry node '${entryId}' not found`);
15831615
}
15841616
// A synthetic flow view — executeNode/traverseNext only read `nodes`/`edges`.
15851617
const subFlow = { nodes: region.nodes, edges: region.edges ?? [] } as unknown as FlowParsed;
1586-
// TODO(#1479): merge region step logs into the parent run log so
1587-
// per-iteration body steps surface in run observability.
15881618
const regionSteps: StepLogEntry[] = [];
15891619
try {
15901620
await this.executeNode(entry, subFlow, variables, context, regionSteps);
@@ -1596,6 +1626,19 @@ export class AutomationEngine implements IAutomationService {
15961626
}
15971627
throw err;
15981628
}
1629+
// Tag this region's steps with their immediate container. Innermost wins:
1630+
// a step that already carries a `parentNodeId` (set by a nested region)
1631+
// is left untouched.
1632+
if (grouping) {
1633+
for (const step of regionSteps) {
1634+
if (step.parentNodeId === undefined) {
1635+
step.parentNodeId = grouping.parentNodeId;
1636+
if (grouping.iteration !== undefined) step.iteration = grouping.iteration;
1637+
if (grouping.regionKind !== undefined) step.regionKind = grouping.regionKind;
1638+
}
1639+
}
1640+
}
1641+
return regionSteps;
15991642
}
16001643

16011644
/**

packages/spec/src/automation/execution.zod.ts

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,13 @@ export const ExecutionStepLogSchema = lazySchema(() => z.object({
5858
stack: z.string().optional().describe('Stack trace'),
5959
}).optional().describe('Error details if step failed'),
6060
retryAttempt: z.number().int().min(0).optional().describe('Retry attempt number (0 = first try)'),
61+
// #1479: structured-region grouping. Tag a step that ran inside a
62+
// `loop` / `parallel` / `try_catch` body region with its immediate container,
63+
// so run observability can nest per-iteration / per-branch body steps under
64+
// the container instead of showing it as a single opaque step.
65+
parentNodeId: z.string().optional().describe('Enclosing structured-region container node ID (loop/parallel/try_catch)'),
66+
iteration: z.number().int().min(0).optional().describe('Zero-based loop iteration or parallel branch index of the enclosing region'),
67+
regionKind: z.string().optional().describe('Region kind the step ran in: loop-body | parallel-branch | try | catch'),
6168
}));
6269
export type ExecutionStepLog = z.infer<typeof ExecutionStepLogSchema>;
6370

0 commit comments

Comments
 (0)