mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-08-28 05:27:41 +00:00
Add runtime for_each item injection
This commit is contained in:
Bryan Helmkamp
parent
6bb6b5efcc
commit
b53045a1ac
32 changed files with 2348 additions and 233 deletions
|
|
@ -172,14 +172,48 @@ describe("parseParallelOverview", () => {
|
|||
failureCount: 1,
|
||||
durationMs: 12000,
|
||||
results: [
|
||||
{ id: "branch-a", status: "succeeded" },
|
||||
{ id: "branch-b", status: "succeeded" },
|
||||
{ id: "branch-c", status: "failed" },
|
||||
{ id: "branch-a", index: null, itemLabel: null, status: "succeeded" },
|
||||
{ id: "branch-b", index: null, itemLabel: null, status: "succeeded" },
|
||||
{ id: "branch-c", index: null, itemLabel: null, status: "failed" },
|
||||
],
|
||||
isComplete: true,
|
||||
});
|
||||
});
|
||||
|
||||
test("parses dynamic item identity from results", () => {
|
||||
const events: EventEnvelope[] = [
|
||||
makeEventEnvelope(1, {
|
||||
event: "parallel.completed",
|
||||
properties: {
|
||||
duration_ms: 20,
|
||||
success_count: 2,
|
||||
failure_count: 0,
|
||||
results: [
|
||||
{
|
||||
id: "reviewer",
|
||||
index: 0,
|
||||
item_label: "auth",
|
||||
status: "succeeded",
|
||||
context_updates: {},
|
||||
},
|
||||
{
|
||||
id: "reviewer",
|
||||
index: 1,
|
||||
item_label: "api",
|
||||
status: "succeeded",
|
||||
context_updates: {},
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
];
|
||||
|
||||
expect(parseParallelOverview(events).results).toEqual([
|
||||
{ id: "reviewer", index: 0, itemLabel: "auth", status: "succeeded" },
|
||||
{ id: "reviewer", index: 1, itemLabel: "api", status: "succeeded" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("reports in-flight when only the started event is present", () => {
|
||||
const events: EventEnvelope[] = [
|
||||
makeEventEnvelope(1, {
|
||||
|
|
|
|||
|
|
@ -147,6 +147,8 @@ export function parseHumanInterviewPairs(events: EventEnvelope[]): HumanIntervie
|
|||
/** Identity and outcome of one branch, parsed from `parallel.completed`. */
|
||||
export interface ParallelBranchSummary {
|
||||
id: string;
|
||||
index: number | null;
|
||||
itemLabel: string | null;
|
||||
status: StageOutcome;
|
||||
}
|
||||
|
||||
|
|
@ -187,9 +189,11 @@ export function parseParallelOverview(events: EventEnvelope[]): ParallelOverview
|
|||
const record = entry && typeof entry === "object" ? (entry as UnknownRecord) : null;
|
||||
if (!record) return null;
|
||||
const id = getString(record, "id");
|
||||
const index = getNumber(record, "index") ?? null;
|
||||
const itemLabel = getString(record, "item_label") ?? null;
|
||||
const status = asStageOutcome(getString(record, "status"));
|
||||
if (!id || !status) return null;
|
||||
return { id, status } satisfies ParallelBranchSummary;
|
||||
return { id, index, itemLabel, status } satisfies ParallelBranchSummary;
|
||||
})
|
||||
.filter((r): r is ParallelBranchSummary => r != null);
|
||||
if (branchCount == null) branchCount = results.length;
|
||||
|
|
|
|||
|
|
@ -82,4 +82,65 @@ describe("ParallelChildren", () => {
|
|||
"/runs/run-1/stages/branch-b@1",
|
||||
]);
|
||||
});
|
||||
|
||||
test("uses item labels and avoids ambiguous id-only links", () => {
|
||||
let renderer!: TestRenderer.ReactTestRenderer;
|
||||
act(() => {
|
||||
renderer = TestRenderer.create(
|
||||
<MemoryRouter>
|
||||
<ParallelChildren
|
||||
stage={parallelStage}
|
||||
events={[
|
||||
event({
|
||||
properties: {
|
||||
duration_ms: 100,
|
||||
success_count: 2,
|
||||
failure_count: 0,
|
||||
results: [
|
||||
{
|
||||
id: "reviewer",
|
||||
index: 0,
|
||||
item_label: "auth",
|
||||
status: "succeeded",
|
||||
context_updates: {},
|
||||
},
|
||||
{
|
||||
id: "reviewer",
|
||||
index: 1,
|
||||
item_label: "api",
|
||||
status: "succeeded",
|
||||
context_updates: {},
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
]}
|
||||
runId="run-1"
|
||||
allStages={[
|
||||
{
|
||||
...parallelStage,
|
||||
id: "reviewer@1",
|
||||
name: "reviewer",
|
||||
nodeId: "reviewer",
|
||||
handler: "agent",
|
||||
},
|
||||
{
|
||||
...parallelStage,
|
||||
id: "reviewer@2",
|
||||
name: "reviewer",
|
||||
nodeId: "reviewer",
|
||||
handler: "agent",
|
||||
},
|
||||
]}
|
||||
/>
|
||||
</MemoryRouter>,
|
||||
);
|
||||
});
|
||||
|
||||
const rendered = JSON.stringify(renderer.toJSON());
|
||||
expect(rendered).toContain("auth");
|
||||
expect(rendered).toContain("api");
|
||||
expect(rendered).toContain("reviewer");
|
||||
expect(renderer.root.findAllByType("a")).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -13,6 +13,8 @@ import { parseParallelOverview } from "./helpers";
|
|||
/** Branch row view state: completed outcomes plus a synthesized in-flight row. */
|
||||
interface BranchRow {
|
||||
id: string;
|
||||
index: number | null;
|
||||
itemLabel: string | null;
|
||||
status: StageState;
|
||||
}
|
||||
|
||||
|
|
@ -53,8 +55,15 @@ function ChildRow({
|
|||
>
|
||||
{stageStatusLabel(result.status)}
|
||||
</span>
|
||||
<span className="min-w-0 flex-1 truncate font-mono text-sm text-fg-3">
|
||||
{result.id}
|
||||
<span className="min-w-0 flex flex-1 items-baseline gap-2">
|
||||
<span className="truncate font-mono text-sm text-fg-3">
|
||||
{result.itemLabel ?? result.id}
|
||||
</span>
|
||||
{result.itemLabel && (
|
||||
<span className="shrink-0 font-mono text-[11px] text-fg-muted">
|
||||
{result.id}
|
||||
</span>
|
||||
)}
|
||||
</span>
|
||||
{stageHref && (
|
||||
<ArrowTopRightOnSquareIcon
|
||||
|
|
@ -104,11 +113,21 @@ export function ParallelChildren({
|
|||
return new Map(Array.from(latest.entries()).map(([nodeId, s]) => [nodeId, s.id]));
|
||||
}, [allStages]);
|
||||
|
||||
const resultCountByNode = useMemo(() => {
|
||||
const counts = new Map<string, number>();
|
||||
for (const result of overview.results) {
|
||||
counts.set(result.id, (counts.get(result.id) ?? 0) + 1);
|
||||
}
|
||||
return counts;
|
||||
}, [overview.results]);
|
||||
|
||||
const items: BranchRow[] = overview.results.length > 0
|
||||
? overview.results
|
||||
: overview.branchCount && overview.branchCount > 0
|
||||
? Array.from({ length: overview.branchCount }, (_, i) => ({
|
||||
id: `branch ${i + 1}`,
|
||||
index: i,
|
||||
itemLabel: null,
|
||||
status: StageState.RUNNING,
|
||||
}))
|
||||
: [];
|
||||
|
|
@ -144,11 +163,16 @@ export function ParallelChildren({
|
|||
) : (
|
||||
<ul className="divide-y divide-line rounded-lg bg-panel outline-1 -outline-offset-1 outline-line">
|
||||
{items.map((result, i) => {
|
||||
const stageId = latestStageByNode.get(result.id);
|
||||
// A node id alone cannot identify one dynamic item when several
|
||||
// results share the template target. Avoid linking every row to
|
||||
// whichever execution happened to finish last.
|
||||
const stageId = resultCountByNode.get(result.id) === 1
|
||||
? latestStageByNode.get(result.id)
|
||||
: null;
|
||||
const href = stageId ? `/runs/${runId}/stages/${stageId}` : null;
|
||||
return (
|
||||
<ChildRow
|
||||
key={`${result.id}-${i}`}
|
||||
key={`${result.id}-${result.index ?? i}`}
|
||||
result={result}
|
||||
stageHref={href}
|
||||
/>
|
||||
|
|
|
|||
|
|
@ -7,8 +7,10 @@ This document defines Fabro's parallel fan-out (`shape=component`) and fan-in
|
|||
|
||||
## 1. Execution model
|
||||
|
||||
A parallel node dispatches one branch for each outgoing edge. A branch executes
|
||||
the single target node on that edge; parallel branches are not subgraph walks.
|
||||
A static parallel node dispatches one branch for each outgoing edge. A
|
||||
`for_each` parallel node has one outgoing template edge and dispatches one
|
||||
branch for each item in a runtime JSON array. A branch executes the single
|
||||
target node on that edge; parallel branches are not subgraph walks.
|
||||
Every branch:
|
||||
|
||||
- receives an independent fork of the parent workflow context;
|
||||
|
|
@ -22,6 +24,11 @@ once and defaults to 4. The parallel node always waits for every branch task,
|
|||
even when a branch fails or run cancellation begins. There is no early-success
|
||||
join mode.
|
||||
|
||||
`for_each` sources use flat context lookup: try the declared key, then strip a
|
||||
leading `context.` and try again. Inline arrays and managed `blob://` or
|
||||
`file://` JSON references are accepted. The template target is limited to an
|
||||
agent or prompt node, and nested `for_each` is rejected.
|
||||
|
||||
The parent context is not used as shared mutable branch state. A branch can
|
||||
change its context fork without exposing those changes as top-level values to
|
||||
other branches or to the parent.
|
||||
|
|
@ -55,15 +62,19 @@ The shared result type is:
|
|||
```rust
|
||||
ParallelBranchResult {
|
||||
id: String,
|
||||
index: Option<usize>,
|
||||
item_label: Option<String>,
|
||||
status: StageOutcome,
|
||||
context_updates: BTreeMap<String, serde_json::Value>,
|
||||
}
|
||||
```
|
||||
|
||||
The parallel handler stores one result per outgoing edge in
|
||||
`parallel.results`. Results preserve outgoing-edge order, independent of branch
|
||||
completion order. `parallel.branch_count` stores the number of dispatched
|
||||
branches.
|
||||
The parallel handler stores one result per outgoing edge or runtime item in
|
||||
`parallel.results`. Results preserve outgoing-edge or input order, independent
|
||||
of branch completion order. New results always contain `index`; it is optional
|
||||
only so records written before indexed identity still deserialize.
|
||||
`item_label` is set for `for_each` from item `name`, then `label`, then index.
|
||||
`parallel.branch_count` stores the number of dispatched branches.
|
||||
|
||||
`context_updates` includes changes made in the branch context and updates
|
||||
returned by the branch outcome. This applies to successful and failed branches,
|
||||
|
|
@ -79,7 +90,13 @@ The parallel stage outcome is:
|
|||
|
||||
- `succeeded` when every branch succeeds;
|
||||
- `failed` when every branch fails;
|
||||
- `partially_succeeded` for mixed outcomes, partial outcomes, and zero branches.
|
||||
- `partially_succeeded` for mixed outcomes, partial outcomes, and a static
|
||||
fan-out with zero branches;
|
||||
- `succeeded` for a valid `for_each` source with zero items.
|
||||
|
||||
For `for_each`, a missing key, missing blob, invalid JSON, or non-array fails
|
||||
before `parallel.started`. A valid empty array emits paired parallel events
|
||||
with count zero and jumps directly to the template target's fan-in.
|
||||
|
||||
## 4. Artifacts and downstream context
|
||||
|
||||
|
|
@ -123,14 +140,23 @@ winner, restore files, or choose workspace state.
|
|||
Parallel execution emits:
|
||||
|
||||
- `parallel.started` with `visit` and `branch_count`;
|
||||
- `parallel.branch.started` with stable branch identity and index;
|
||||
- `parallel.branch.completed` with index, duration, and status;
|
||||
- `parallel.branch.started` with stable branch identity, index, and optional
|
||||
item label;
|
||||
- `parallel.branch.completed` with index, optional item label, duration, and
|
||||
status;
|
||||
- `parallel.completed` with counts and the ordered typed result array.
|
||||
|
||||
Every branch task emits one terminal branch completion event, including handler
|
||||
failure, cancellation before semaphore acquisition, panic, or join failure.
|
||||
The final typed array is also projected into
|
||||
`StageProjection.parallel_results`.
|
||||
`StageProjection.parallel_results`. Raw runtime items are recorded once in the
|
||||
existing `stage.prompt` event and are not duplicated in branch events or
|
||||
results.
|
||||
|
||||
Branch attempts use the same artifact, panic, and executor-timeout envelope as
|
||||
ordinary nodes, wrapped in a branch-local retry loop. A retry preserves its
|
||||
branch identity, stage scope, item label, context fork, and result index. It
|
||||
releases the concurrency permit during backoff and reacquires it before the
|
||||
next attempt. Generic graph lifecycle callbacks, edge selection, thread reuse,
|
||||
and per-item checkpoints are intentionally excluded.
|
||||
|
||||
## 7. Cancellation
|
||||
|
||||
|
|
|
|||
|
|
@ -10607,6 +10607,17 @@ components:
|
|||
properties:
|
||||
id:
|
||||
type: string
|
||||
index:
|
||||
type: integer
|
||||
minimum: 0
|
||||
description: >-
|
||||
Zero-based input item or outgoing-edge position. Absent only on
|
||||
parallel results written before indexed branch identity was added.
|
||||
item_label:
|
||||
type: string
|
||||
description: >-
|
||||
Human-readable for_each item identity, derived from name, then
|
||||
label, then the zero-based input index.
|
||||
status:
|
||||
$ref: "#/components/schemas/StageOutcome"
|
||||
context_updates:
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ Start → Plan → Implement → Test → Exit
|
|||
└─ sets response.plan, last_response
|
||||
```
|
||||
|
||||
Context is thread-safe and shared across the entire run. Parallel branches receive an isolated **deep copy** of the context at the point of fan-out, so branches can't interfere with each other. When branches merge, the fan-in handler records the results under `parallel.fan_in.*` keys.
|
||||
Context is thread-safe and shared across the entire run. Parallel branches receive an isolated **deep copy** of the context at the point of fan-out, so branches can't interfere with each other. The parallel handler gathers their results under `parallel.results`.
|
||||
|
||||
## How agents access context
|
||||
|
||||
|
|
@ -61,11 +61,30 @@ Agents can also emit arbitrary context updates by including a JSON object with a
|
|||
|
||||
| Key | Value |
|
||||
|---|---|
|
||||
| `parallel.results` | Ordered branch results. Each entry contains `id`, `status`, and the branch's isolated `context_updates`. |
|
||||
| `parallel.branch_count` | Number of outgoing branches dispatched by the parallel node. |
|
||||
| `parallel.results` | Ordered branch results. Each entry contains `id`, `index`, optional `item_label`, `status`, and the branch's isolated `context_updates`. Legacy results may omit `index`. |
|
||||
| `parallel.branch_count` | Number of branches dispatched. For `for_each`, this is the runtime array length. |
|
||||
|
||||
Branch updates remain nested inside `parallel.results`; they are not merged into top-level context. Prompted fan-in nodes can synthesize the complete result array, while promptless fan-in nodes act as barriers.
|
||||
|
||||
### Runtime arrays with `for_each`
|
||||
|
||||
A parallel node can read a flat context key and run one agent or prompt target
|
||||
per array item:
|
||||
|
||||
```dot
|
||||
batch [shape=component, for_each="context.candidates"]
|
||||
batch -> reviewer
|
||||
```
|
||||
|
||||
`context.candidates` first checks the exact key and then falls back to
|
||||
`candidates`. The source must be a JSON array, either inline or stored behind a
|
||||
Fabro-managed `blob://` or `file://` reference. General nested lookup such as
|
||||
`output.scan.candidates` is not supported; have the producing node write the
|
||||
array to a flat context key such as `candidates`.
|
||||
|
||||
Each item receives a separate context fork, while `(id, index)` identifies its
|
||||
result. The item itself is not copied into `parallel.results`.
|
||||
|
||||
### Engine-managed keys
|
||||
|
||||
The engine sets several keys automatically. These are prefixed with `internal.` and are excluded from preambles:
|
||||
|
|
|
|||
|
|
@ -66,6 +66,13 @@ Envelope fields:
|
|||
|
||||
Only `id`, `ts`, `run_id`, and `event` are always present. Optional fields are omitted when they do not apply.
|
||||
|
||||
For runtime `for_each` branches, `parallel.branch.started` and
|
||||
`parallel.branch.completed` include the zero-based `index` and an optional
|
||||
`item_label`. They do not include the raw item. The final prompt is recorded by
|
||||
the existing `stage.prompt` event, including the fenced item data, so event
|
||||
streams, run dumps, and retained logs are source-bearing data. Apply the same
|
||||
access controls and retention policy you use for workflow inputs.
|
||||
|
||||
## Reading the event stream
|
||||
|
||||
Because event payload lives in `properties`, most shell queries should look there.
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ Each node type has its own rules for which outcomes it can return:
|
|||
|---|---|---|
|
||||
| **Command** | `succeeded`, `failed` | `succeeded` when exit code is 0; `failed` otherwise |
|
||||
| **Agent / Prompt** | `succeeded`, `failed`, `partially_succeeded`, `skipped` | Defaults to `succeeded`. The LLM can set any outcome via a [routing directive](/agents/outputs#routing-directives) JSON object in its response. Backend errors request retry when retryable or finish as `failed`. |
|
||||
| **Parallel** | `succeeded`, `partially_succeeded`, `failed` | Waits for every branch. `succeeded` when all branches succeed, `failed` when all branches fail, and `partially_succeeded` for mixed, partial, or zero-branch results. |
|
||||
| **Parallel** | `succeeded`, `partially_succeeded`, `failed` | Waits for every branch. `succeeded` when all branches succeed, `failed` when all branches fail, and `partially_succeeded` for mixed or partial results. A static fan-out with no branches remains partial; a valid `for_each` source with zero items succeeds. |
|
||||
| **Human** | `succeeded` | Always succeeds — the user's selection becomes a routing signal via `preferred_label` |
|
||||
| **Conditional** | `succeeded` | Always succeeds — routing is handled by the engine's edge selection |
|
||||
| **Start / Exit / Wait** | `succeeded` | Always succeed |
|
||||
|
|
|
|||
|
|
@ -253,9 +253,27 @@ audit [
|
|||
| Attribute | Type | Description |
|
||||
|---|---|---|
|
||||
| `max_parallel` | Integer | Maximum concurrent branches (default: 4). The node always waits for every branch. |
|
||||
| `for_each` | String | Flat runtime context key containing a JSON array. Runs the node's single agent or prompt target once per item. `context.items` first checks that exact key, then falls back to `items`. |
|
||||
|
||||
For the first node in each branch, `fidelity` resolves from the fork-to-branch edge, then the branch node; without either, the fork preamble is inherited unchanged. Branch-specific preambles are rendered before fan-out from the fork's context snapshot. Concurrent branches cannot share sessions, so explicit branch `full` becomes `summary:high`, and branch-level `thread_id` is inert.
|
||||
|
||||
When `for_each` is set, the parallel node must have exactly one outgoing edge,
|
||||
and its target must be an agent or prompt node. Fabro resolves the source as an
|
||||
inline array or a managed `blob://` or `file://` JSON artifact, then clones the
|
||||
target once per item. Nested `for_each` is not supported.
|
||||
|
||||
Each clone receives the target's normal prompt followed by the item as pretty
|
||||
JSON inside a fresh random fence and a fixed notice that the content is data,
|
||||
not instructions. There is no item interpolation syntax. The fence prevents an
|
||||
item from closing its own data block, but it does not restrict an agent's tools;
|
||||
workflow authors must give the target only the tool access appropriate for
|
||||
untrusted item data.
|
||||
|
||||
Dynamic results retain input order. Each result uses the template target ID and
|
||||
adds a zero-based `index` plus `item_label`, derived from the item's `name`,
|
||||
then `label`, then its index. An empty array succeeds and proceeds directly to
|
||||
the target's fan-in node without executing the unparameterized target.
|
||||
|
||||
### Wait nodes
|
||||
|
||||
| Attribute | Type | Description |
|
||||
|
|
|
|||
|
|
@ -91,9 +91,72 @@ fork [shape=component, max_parallel=2]
|
|||
|
||||
This is useful when branches are resource-intensive (e.g., each running a full agent session with tool calls) and you want to limit concurrency.
|
||||
|
||||
## Review a runtime candidate list
|
||||
|
||||
Static branches work when the review perspectives are known while writing the
|
||||
graph. A security scan often discovers its review candidates at runtime. Write
|
||||
that array to a flat context key, then use `for_each` to run one reviewer per
|
||||
candidate:
|
||||
|
||||
```dot
|
||||
discover [
|
||||
shape=tab,
|
||||
output_schema="routing",
|
||||
prompt="Identify security review candidates. Return only JSON with \
|
||||
context_updates.candidates as an array of objects. Give every object a \
|
||||
name, path, and reason."
|
||||
]
|
||||
|
||||
review_batch [
|
||||
shape=component,
|
||||
for_each="context.candidates",
|
||||
max_parallel=4
|
||||
]
|
||||
|
||||
reviewer [
|
||||
label="Candidate Reviewer",
|
||||
prompt="Inspect this candidate for exploitable security problems. Report \
|
||||
evidence, severity, and a concrete remediation."
|
||||
]
|
||||
|
||||
aggregate [
|
||||
shape=tripleoctagon,
|
||||
prompt="Synthesize all candidate reviews. Call out candidates whose branch failed."
|
||||
]
|
||||
|
||||
discover -> review_batch
|
||||
review_batch -> reviewer
|
||||
reviewer -> aggregate
|
||||
```
|
||||
|
||||
The `routing` output schema merges `context_updates.candidates` into the flat
|
||||
`candidates` context key. `review_batch` accepts that inline array or its
|
||||
automatically offloaded managed artifact reference. It clones `reviewer` for
|
||||
each item, appends the item as pretty JSON inside a fresh matching fence, and
|
||||
keeps `parallel.results` in candidate order.
|
||||
|
||||
Each result has `id="reviewer"`, a zero-based `index`, and an `item_label`
|
||||
chosen from the item's `name`, then `label`, then index. An empty candidate
|
||||
array succeeds and proceeds directly to `aggregate`. Mixed success and failure
|
||||
also proceeds; only an all-failed batch fails the parallel stage.
|
||||
|
||||
Retries and executor-enforced timeouts apply independently to each candidate.
|
||||
A retry keeps the same result index and branch identity, and releases its
|
||||
`max_parallel` slot while waiting for backoff. If a run stops partway through
|
||||
the batch, resuming it reruns every item because per-item checkpoints are not
|
||||
created.
|
||||
|
||||
<Warning>
|
||||
The randomized fence prevents candidate text from closing its own data block,
|
||||
but it does not sandbox a tool-enabled reviewer. Candidate data can still try
|
||||
to influence the model. Limit the target agent's tools and permissions to the
|
||||
minimum needed for the review.
|
||||
</Warning>
|
||||
|
||||
## What you've learned
|
||||
|
||||
- **Fan-out nodes** (`shape=component`) spawn concurrent branches and wait for all of them
|
||||
- `for_each` runs one agent or prompt template for every item in a runtime array
|
||||
- Parallel branches share one checkout, so workflows must prevent or tolerate file races
|
||||
- **Fan-in nodes** (`shape=tripleoctagon`) can synthesize `parallel.results` with a prompt
|
||||
- Fan-in never selects or restores workspace state, and no results file is created
|
||||
|
|
|
|||
|
|
@ -168,6 +168,28 @@ fork -> quality
|
|||
| Attribute | Description |
|
||||
|---|---|
|
||||
| `max_parallel` | Maximum concurrent branches (default: 4) |
|
||||
| `for_each` | Flat context key containing a runtime JSON array. Requires one outgoing agent or prompt template target. |
|
||||
|
||||
To run one template node for a runtime array, add `for_each`:
|
||||
|
||||
```dot
|
||||
review_batch [shape=component, for_each="context.candidates", max_parallel=8]
|
||||
reviewer [prompt="Review this candidate for security issues."]
|
||||
aggregate [shape=tripleoctagon, prompt="Synthesize every candidate review."]
|
||||
|
||||
review_batch -> reviewer -> aggregate
|
||||
```
|
||||
|
||||
Fabro accepts an inline array or a managed JSON artifact reference. It runs
|
||||
`reviewer` once per item, appends the item to the prompt as fenced data, and
|
||||
keeps the results in input order. The source lookup is flat:
|
||||
`context.candidates` checks that exact key and then `candidates`; it does not
|
||||
traverse nested objects.
|
||||
|
||||
The template target must be an agent or prompt node, and nested `for_each` is
|
||||
rejected. An empty source array succeeds with `parallel.results=[]` and skips
|
||||
straight to `aggregate`. Missing, invalid, or non-array sources fail the
|
||||
parallel stage before any branches start.
|
||||
|
||||
Because the checkout is shared, file changes from one branch are immediately visible to the others. Concurrent writes can race or overwrite each other. Fabro does not isolate branch files, lock paths, detect conflicts, or warn about overlapping writes. Design branches to be read-only or assign each branch disjoint files and directories when deterministic workspace changes matter.
|
||||
|
||||
|
|
|
|||
|
|
@ -330,11 +330,15 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
|
|||
delay_ms: props.delay_ms,
|
||||
}),
|
||||
EventBody::ParallelStarted(_) => Some(ProgressEvent::ParallelStarted),
|
||||
EventBody::ParallelBranchStarted(_) => {
|
||||
Some(ProgressEvent::ParallelBranchStarted { branch: node_id })
|
||||
}
|
||||
EventBody::ParallelBranchStarted(props) => Some(ProgressEvent::ParallelBranchStarted {
|
||||
branch: parallel_branch_display(&node_id, props.index, props.item_label.as_deref()),
|
||||
}),
|
||||
EventBody::ParallelBranchCompleted(props) => Some(ProgressEvent::ParallelBranchCompleted {
|
||||
branch: node_id,
|
||||
branch: parallel_branch_display(
|
||||
&node_id,
|
||||
props.index,
|
||||
props.item_label.as_deref(),
|
||||
),
|
||||
duration_ms: props.duration_ms,
|
||||
status: props.status,
|
||||
}),
|
||||
|
|
@ -464,6 +468,13 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
|
|||
}
|
||||
}
|
||||
|
||||
fn parallel_branch_display(node_id: &str, index: usize, item_label: Option<&str>) -> String {
|
||||
item_label.map_or_else(
|
||||
|| node_id.to_string(),
|
||||
|label| format!("{label} ({node_id} #{index})"),
|
||||
)
|
||||
}
|
||||
|
||||
pub(super) fn from_json_line(line: &str) -> Option<ProgressEvent> {
|
||||
let stored = RunEvent::from_json_str(line).ok()?;
|
||||
from_run_event(&stored)
|
||||
|
|
|
|||
|
|
@ -660,10 +660,11 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
|
||||
branch: "security".into(),
|
||||
index: 0,
|
||||
item_label: Some("auth".into()),
|
||||
});
|
||||
let stage = &ui.stage.active_stages["fork1"];
|
||||
assert_eq!(stage.tool_calls.len(), 1);
|
||||
assert_eq!(stage.tool_calls[0].tool_call_id, "security");
|
||||
assert_eq!(stage.tool_calls[0].tool_call_id, "auth (security #0)");
|
||||
assert!(matches!(
|
||||
stage.tool_calls[0].status,
|
||||
ToolCallStatus::Running
|
||||
|
|
@ -674,6 +675,7 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
|
||||
branch: "security".into(),
|
||||
index: 0,
|
||||
item_label: Some("auth".into()),
|
||||
duration_ms: 2000,
|
||||
status: fabro_workflow::outcome::StageOutcome::Succeeded,
|
||||
});
|
||||
|
|
@ -701,6 +703,7 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
|
||||
branch: "security".into(),
|
||||
index: 0,
|
||||
item_label: None,
|
||||
});
|
||||
|
||||
let stage = &ui.stage.active_stages["fork1"];
|
||||
|
|
@ -1477,12 +1480,14 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
|
||||
branch: "security".into(),
|
||||
index: 0,
|
||||
item_label: None,
|
||||
});
|
||||
emit(&mut ui, Event::ParallelBranchCompleted {
|
||||
parallel_group_id: StageId::new("fork1", 1),
|
||||
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
|
||||
branch: "security".into(),
|
||||
index: 0,
|
||||
item_label: None,
|
||||
duration_ms: 500,
|
||||
status: fabro_workflow::outcome::StageOutcome::Succeeded,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -607,6 +607,8 @@ mod tests {
|
|||
stage.script_timing = Some(serde_json::json!({ "duration_ms": 10 }));
|
||||
stage.parallel_results = Some(vec![fabro_types::ParallelBranchResult {
|
||||
id: "review".to_string(),
|
||||
index: Some(0),
|
||||
item_label: None,
|
||||
status: fabro_types::StageOutcome::Succeeded,
|
||||
context_updates: std::collections::BTreeMap::from([(
|
||||
"response.review".to_string(),
|
||||
|
|
|
|||
|
|
@ -2293,6 +2293,7 @@ mod tests {
|
|||
"2026-04-07T12:00:00Z",
|
||||
EventBody::ParallelBranchStarted(ParallelBranchStartedProps {
|
||||
index: 0,
|
||||
item_label: None,
|
||||
graph_visit: None,
|
||||
resumed_from_stage_id: None,
|
||||
}),
|
||||
|
|
@ -2308,6 +2309,7 @@ mod tests {
|
|||
4,
|
||||
EventBody::ParallelBranchCompleted(ParallelBranchCompletedProps {
|
||||
index: 0,
|
||||
item_label: None,
|
||||
duration_ms: 1234,
|
||||
status: StageOutcome::Succeeded,
|
||||
}),
|
||||
|
|
@ -2334,6 +2336,7 @@ mod tests {
|
|||
3,
|
||||
EventBody::ParallelBranchStarted(ParallelBranchStartedProps {
|
||||
index: 0,
|
||||
item_label: None,
|
||||
graph_visit: None,
|
||||
resumed_from_stage_id: None,
|
||||
}),
|
||||
|
|
@ -2345,6 +2348,7 @@ mod tests {
|
|||
4,
|
||||
EventBody::ParallelBranchCompleted(ParallelBranchCompletedProps {
|
||||
index: 0,
|
||||
item_label: None,
|
||||
duration_ms: 500,
|
||||
status: StageOutcome::Failed {
|
||||
retry_requested: false,
|
||||
|
|
|
|||
|
|
@ -145,6 +145,8 @@ fn serializable_projection_round_trips_and_trims_bulky_node_fields() {
|
|||
stage.script_timing = Some(json!({ "duration_ms": 10 }));
|
||||
let parallel_results = vec![ParallelBranchResult {
|
||||
id: "review".to_string(),
|
||||
index: Some(0),
|
||||
item_label: None,
|
||||
status: StageOutcome::Succeeded,
|
||||
context_updates: BTreeMap::from([("response.review".to_string(), json!("looks good"))]),
|
||||
}];
|
||||
|
|
|
|||
226
lib/components/fabro-validate/src/rules/for_each_contract.rs
Normal file
226
lib/components/fabro-validate/src/rules/for_each_contract.rs
Normal file
|
|
@ -0,0 +1,226 @@
|
|||
use fabro_graphviz::graph::{Graph, is_llm_handler_type};
|
||||
|
||||
use crate::{Diagnostic, LintRule, Severity};
|
||||
|
||||
pub(super) fn rule() -> Box<dyn LintRule> {
|
||||
Box::new(Rule)
|
||||
}
|
||||
|
||||
struct Rule;
|
||||
|
||||
fn diagnostic(node_id: &str, message: String, fix: impl Into<String>) -> Diagnostic {
|
||||
Diagnostic {
|
||||
rule: "for_each_contract".to_string(),
|
||||
severity: Severity::Error,
|
||||
message,
|
||||
node_id: Some(node_id.to_string()),
|
||||
edge: None,
|
||||
fix: Some(fix.into()),
|
||||
..Diagnostic::default()
|
||||
}
|
||||
}
|
||||
|
||||
impl LintRule for Rule {
|
||||
fn name(&self) -> &'static str {
|
||||
"for_each_contract"
|
||||
}
|
||||
|
||||
fn apply(&self, graph: &Graph) -> Vec<Diagnostic> {
|
||||
let mut diagnostics = Vec::new();
|
||||
|
||||
for node in graph.nodes.values() {
|
||||
let Some(raw_source) = node.attrs.get("for_each") else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let source = raw_source.as_str();
|
||||
if source.is_none_or(|source| source.trim().is_empty()) {
|
||||
diagnostics.push(diagnostic(
|
||||
&node.id,
|
||||
format!(
|
||||
"Node '{}' has an empty or non-string 'for_each' source",
|
||||
node.id
|
||||
),
|
||||
"Set 'for_each' to a context key such as \"context.candidates\"",
|
||||
));
|
||||
}
|
||||
|
||||
if node.handler_type() != Some("parallel") {
|
||||
diagnostics.push(diagnostic(
|
||||
&node.id,
|
||||
format!(
|
||||
"Node '{}' sets 'for_each', but only parallel nodes can fan out over runtime items",
|
||||
node.id
|
||||
),
|
||||
"Remove 'for_each' or change the node to type=\"parallel\"",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
|
||||
let outgoing = graph.outgoing_edges(&node.id);
|
||||
if outgoing.len() != 1 {
|
||||
diagnostics.push(diagnostic(
|
||||
&node.id,
|
||||
format!(
|
||||
"Parallel node '{}' sets 'for_each' and must have exactly one outgoing template edge, but has {}",
|
||||
node.id,
|
||||
outgoing.len()
|
||||
),
|
||||
"Keep one outgoing edge whose target is the agent or prompt to run for each item",
|
||||
));
|
||||
continue;
|
||||
}
|
||||
|
||||
let target_id = &outgoing[0].to;
|
||||
let Some(target) = graph.nodes.get(target_id) else {
|
||||
continue;
|
||||
};
|
||||
if target.attrs.contains_key("for_each") {
|
||||
diagnostics.push(diagnostic(
|
||||
&node.id,
|
||||
format!(
|
||||
"Parallel node '{}' targets '{}', which also sets 'for_each'; nested for_each is not supported",
|
||||
node.id, target.id
|
||||
),
|
||||
"Remove the nested 'for_each' and use a single runtime fan-out",
|
||||
));
|
||||
}
|
||||
if !is_llm_handler_type(target.handler_type()) {
|
||||
diagnostics.push(diagnostic(
|
||||
&node.id,
|
||||
format!(
|
||||
"Parallel node '{}' sets 'for_each', but template target '{}' is not an agent or prompt node",
|
||||
node.id, target.id
|
||||
),
|
||||
"Target one agent or prompt node from the for_each parallel node",
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
diagnostics
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use fabro_graphviz::graph::{AttrValue, Edge, Node};
|
||||
|
||||
use super::Rule;
|
||||
use crate::rules::test_support::minimal_graph;
|
||||
use crate::{LintRule, Severity};
|
||||
|
||||
fn for_each_node(id: &str) -> Node {
|
||||
let mut node = Node::new(id);
|
||||
node.attrs.insert(
|
||||
"type".to_string(),
|
||||
AttrValue::String("parallel".to_string()),
|
||||
);
|
||||
node.attrs.insert(
|
||||
"for_each".to_string(),
|
||||
AttrValue::String("context.items".to_string()),
|
||||
);
|
||||
node
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_one_agent_template_target() {
|
||||
let mut graph = minimal_graph();
|
||||
graph
|
||||
.nodes
|
||||
.insert("fanout".to_string(), for_each_node("fanout"));
|
||||
graph
|
||||
.nodes
|
||||
.insert("worker".to_string(), Node::new("worker"));
|
||||
graph.edges.push(Edge::new("fanout", "worker"));
|
||||
|
||||
assert!(Rule.apply(&graph).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_for_each_on_non_parallel_node() {
|
||||
let mut graph = minimal_graph();
|
||||
let mut worker = Node::new("worker");
|
||||
worker.attrs.insert(
|
||||
"for_each".to_string(),
|
||||
AttrValue::String("items".to_string()),
|
||||
);
|
||||
graph.nodes.insert("worker".to_string(), worker);
|
||||
|
||||
let diagnostics = Rule.apply(&graph);
|
||||
|
||||
assert_eq!(diagnostics.len(), 1);
|
||||
assert_eq!(diagnostics[0].severity, Severity::Error);
|
||||
assert!(diagnostics[0].message.contains("only parallel"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_multiple_template_edges() {
|
||||
let mut graph = minimal_graph();
|
||||
graph
|
||||
.nodes
|
||||
.insert("fanout".to_string(), for_each_node("fanout"));
|
||||
graph.nodes.insert("one".to_string(), Node::new("one"));
|
||||
graph.nodes.insert("two".to_string(), Node::new("two"));
|
||||
graph.edges.push(Edge::new("fanout", "one"));
|
||||
graph.edges.push(Edge::new("fanout", "two"));
|
||||
|
||||
let diagnostics = Rule.apply(&graph);
|
||||
|
||||
assert_eq!(diagnostics.len(), 1);
|
||||
assert!(diagnostics[0].message.contains("exactly one"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_non_llm_and_nested_template_targets() {
|
||||
let mut graph = minimal_graph();
|
||||
graph
|
||||
.nodes
|
||||
.insert("outer".to_string(), for_each_node("outer"));
|
||||
graph
|
||||
.nodes
|
||||
.insert("inner".to_string(), for_each_node("inner"));
|
||||
graph.edges.push(Edge::new("outer", "inner"));
|
||||
|
||||
let diagnostics = Rule.apply(&graph);
|
||||
let outer_diagnostics = diagnostics
|
||||
.iter()
|
||||
.filter(|diagnostic| diagnostic.node_id.as_deref() == Some("outer"))
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(outer_diagnostics.len(), 2);
|
||||
assert!(diagnostics.iter().all(|d| d.severity == Severity::Error));
|
||||
assert!(
|
||||
outer_diagnostics
|
||||
.iter()
|
||||
.any(|d| d.message.contains("nested"))
|
||||
);
|
||||
assert!(
|
||||
outer_diagnostics
|
||||
.iter()
|
||||
.any(|d| d.message.contains("not an agent or prompt"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_empty_or_non_string_source() {
|
||||
for value in [
|
||||
AttrValue::String(String::new()),
|
||||
AttrValue::String(" ".to_string()),
|
||||
AttrValue::Integer(3),
|
||||
] {
|
||||
let mut graph = minimal_graph();
|
||||
let mut fanout = for_each_node("fanout");
|
||||
fanout.attrs.insert("for_each".to_string(), value);
|
||||
graph.nodes.insert("fanout".to_string(), fanout);
|
||||
graph
|
||||
.nodes
|
||||
.insert("worker".to_string(), Node::new("worker"));
|
||||
graph.edges.push(Edge::new("fanout", "worker"));
|
||||
|
||||
let diagnostics = Rule.apply(&graph);
|
||||
|
||||
assert_eq!(diagnostics.len(), 1);
|
||||
assert!(diagnostics[0].message.contains("empty or non-string"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -5,6 +5,7 @@ mod direction_valid;
|
|||
mod edge_target_exists;
|
||||
mod exit_no_outgoing;
|
||||
mod fidelity_valid;
|
||||
mod for_each_contract;
|
||||
mod freeform_edge_count;
|
||||
mod goal_gate_has_retry;
|
||||
mod import_error;
|
||||
|
|
@ -54,6 +55,7 @@ pub fn built_in_rules() -> Vec<Box<dyn LintRule>> {
|
|||
goal_gate_has_retry::rule(),
|
||||
prompt_on_llm_nodes::rule(),
|
||||
freeform_edge_count::rule(),
|
||||
for_each_contract::rule(),
|
||||
direction_valid::rule(),
|
||||
reserved_keyword_node_id::rule(),
|
||||
all_conditional_edges::rule(),
|
||||
|
|
|
|||
|
|
@ -228,6 +228,27 @@ pub async fn resolve_text_or_blob_ref(value: &Value, run_store: &RunStoreHandle)
|
|||
}
|
||||
}
|
||||
|
||||
/// Resolve a structured JSON value from inline context or a Fabro-managed
|
||||
/// blob reference.
|
||||
///
|
||||
/// Managed `file://` references are normalized through their content-addressed
|
||||
/// blob id instead of reading an execution-local path. Ordinary strings and
|
||||
/// ordinary file references remain unchanged for the caller to validate.
|
||||
pub(crate) async fn resolve_json_value(value: &Value, run_store: &RunStoreHandle) -> Result<Value> {
|
||||
let Some(reference) = value.as_str() else {
|
||||
return Ok(value.clone());
|
||||
};
|
||||
let Some(blob_id) =
|
||||
parse_blob_ref(reference).or_else(|| parse_managed_blob_file_ref(reference))
|
||||
else {
|
||||
return Ok(value.clone());
|
||||
};
|
||||
|
||||
let bytes = read_required_blob(&blob_id, run_store).await?;
|
||||
serde_json::from_slice(&bytes)
|
||||
.map_err(|err| Error::engine_with_source("artifact blob was not valid JSON", err))
|
||||
}
|
||||
|
||||
pub async fn resolve_text_or_blob_ref_str(
|
||||
current: &str,
|
||||
run_store: &RunStoreHandle,
|
||||
|
|
@ -552,6 +573,44 @@ mod tests {
|
|||
assert_eq!(updates.get("small_key").unwrap(), &small_value);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolve_json_value_hydrates_blob_and_managed_file_references() {
|
||||
let run_store = make_run_store("structured-json-resolution").await;
|
||||
let value = serde_json::json!([{"name": "api"}, {"name": "web"}]);
|
||||
let blob_id = run_store
|
||||
.write_blob(&serde_json::to_vec(&value).unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
let handle = run_store.clone().into();
|
||||
|
||||
assert_eq!(
|
||||
resolve_json_value(&serde_json::json!(format_blob_ref(&blob_id)), &handle)
|
||||
.await
|
||||
.unwrap(),
|
||||
value
|
||||
);
|
||||
assert_eq!(
|
||||
resolve_json_value(
|
||||
&serde_json::json!(format!("file:///sandbox/.fabro/blobs/{blob_id}.json")),
|
||||
&handle,
|
||||
)
|
||||
.await
|
||||
.unwrap(),
|
||||
value
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolve_json_value_preserves_inline_json() {
|
||||
let run_store = make_run_store("inline-json-resolution").await;
|
||||
let value = serde_json::json!([1, 2, 3]);
|
||||
|
||||
assert_eq!(
|
||||
resolve_json_value(&value, &run_store.into()).await.unwrap(),
|
||||
value
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn offload_preserves_parallel_results_and_replaces_large_context_updates() {
|
||||
let run_store = make_run_store("parallel-result-artifact-offload").await;
|
||||
|
|
@ -564,6 +623,8 @@ mod tests {
|
|||
let expected_report_blob = RunBlobId::new(&serde_json::to_vec(&large_report).unwrap());
|
||||
let mut typed_results = vec![ParallelBranchResult {
|
||||
id: "branch_a".to_string(),
|
||||
index: Some(0),
|
||||
item_label: None,
|
||||
status: fabro_types::StageOutcome::Succeeded,
|
||||
context_updates: std::collections::BTreeMap::from([
|
||||
(
|
||||
|
|
|
|||
|
|
@ -382,21 +382,25 @@ fn event_body_from_event(event: &Event) -> EventBody {
|
|||
}),
|
||||
Event::ParallelBranchStarted {
|
||||
index,
|
||||
item_label,
|
||||
graph_visit,
|
||||
resumed_from_stage_id,
|
||||
..
|
||||
} => EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps {
|
||||
index: *index,
|
||||
item_label: item_label.clone(),
|
||||
graph_visit: *graph_visit,
|
||||
resumed_from_stage_id: resumed_from_stage_id.clone(),
|
||||
}),
|
||||
Event::ParallelBranchCompleted {
|
||||
index,
|
||||
item_label,
|
||||
duration_ms,
|
||||
status,
|
||||
..
|
||||
} => EventBody::ParallelBranchCompleted(fabro_types::ParallelBranchCompletedProps {
|
||||
index: *index,
|
||||
item_label: item_label.clone(),
|
||||
duration_ms: *duration_ms,
|
||||
status: *status,
|
||||
}),
|
||||
|
|
@ -1811,6 +1815,7 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(group_id, 1),
|
||||
branch: "review".to_string(),
|
||||
index: 1,
|
||||
item_label: Some("api".to_string()),
|
||||
duration_ms: 42,
|
||||
status: StageOutcome::Succeeded,
|
||||
});
|
||||
|
|
@ -1819,6 +1824,7 @@ mod tests {
|
|||
stored.properties().unwrap(),
|
||||
serde_json::json!({
|
||||
"index": 1,
|
||||
"item_label": "api",
|
||||
"duration_ms": 42,
|
||||
"status": "succeeded",
|
||||
})
|
||||
|
|
@ -1836,6 +1842,8 @@ mod tests {
|
|||
results: vec![
|
||||
::fabro_types::ParallelBranchResult {
|
||||
id: "review_api".to_string(),
|
||||
index: Some(0),
|
||||
item_label: Some("api".to_string()),
|
||||
status: StageOutcome::Succeeded,
|
||||
context_updates: BTreeMap::from([(
|
||||
"response.review_api".to_string(),
|
||||
|
|
@ -1844,6 +1852,8 @@ mod tests {
|
|||
},
|
||||
::fabro_types::ParallelBranchResult {
|
||||
id: "review_ux".to_string(),
|
||||
index: Some(1),
|
||||
item_label: Some("ux".to_string()),
|
||||
status: StageOutcome::Failed {
|
||||
retry_requested: false,
|
||||
},
|
||||
|
|
@ -1865,11 +1875,15 @@ mod tests {
|
|||
"results": [
|
||||
{
|
||||
"id": "review_api",
|
||||
"index": 0,
|
||||
"item_label": "api",
|
||||
"status": "succeeded",
|
||||
"context_updates": {"response.review_api": "looks good"},
|
||||
},
|
||||
{
|
||||
"id": "review_ux",
|
||||
"index": 1,
|
||||
"item_label": "ux",
|
||||
"status": "failed",
|
||||
"context_updates": {"response.review_ux": "needs work"},
|
||||
},
|
||||
|
|
@ -1887,6 +1901,7 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1),
|
||||
branch: "review".to_string(),
|
||||
index: 1,
|
||||
item_label: Some("api".to_string()),
|
||||
});
|
||||
assert_eq!(stored.parallel_group_id, Some(StageId::new("fanout", 2)));
|
||||
assert_eq!(
|
||||
|
|
|
|||
|
|
@ -321,6 +321,8 @@ pub enum Event {
|
|||
branch: String,
|
||||
index: usize,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
item_label: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
graph_visit: Option<u32>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
resumed_from_stage_id: Option<StageId>,
|
||||
|
|
@ -330,6 +332,8 @@ pub enum Event {
|
|||
parallel_branch_id: ParallelBranchId,
|
||||
branch: String,
|
||||
index: usize,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
item_label: Option<String>,
|
||||
duration_ms: u64,
|
||||
status: StageOutcome,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -175,6 +175,7 @@ mod tests {
|
|||
parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0),
|
||||
branch: "fork".to_string(),
|
||||
index: 0,
|
||||
item_label: None,
|
||||
}),
|
||||
"parallel.branch.started"
|
||||
);
|
||||
|
|
|
|||
|
|
@ -448,6 +448,8 @@ mod tests {
|
|||
failure_count: 0,
|
||||
results: vec![fabro_types::ParallelBranchResult {
|
||||
id: "a".to_string(),
|
||||
index: Some(0),
|
||||
item_label: None,
|
||||
status: fabro_types::StageOutcome::Succeeded,
|
||||
context_updates: std::collections::BTreeMap::new(),
|
||||
}],
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -1,5 +1,5 @@
|
|||
use std::panic::AssertUnwindSafe;
|
||||
use std::path::PathBuf;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
|
@ -7,7 +7,7 @@ use fabro_core::error::{Error as CoreError, HandlerErrorDetail, Result as CoreRe
|
|||
use fabro_core::handler::NodeHandler;
|
||||
use fabro_core::outcome::FailureCategory;
|
||||
use fabro_core::retry::RetryPolicy as CoreRetryPolicy;
|
||||
use fabro_graphviz::graph::types::Graph as GvGraph;
|
||||
use fabro_graphviz::graph::types::{Graph as GvGraph, Node as GvNode};
|
||||
use fabro_types::SystemActorKind;
|
||||
use futures::FutureExt;
|
||||
use tokio::time::timeout;
|
||||
|
|
@ -31,6 +31,106 @@ pub(crate) struct WorkflowNodeHandler {
|
|||
pub graph: Arc<GvGraph>,
|
||||
}
|
||||
|
||||
/// Execute one handler attempt through the workflow-owned artifact, panic, and
|
||||
/// timeout envelope.
|
||||
///
|
||||
/// The core executor and direct parallel branch runner deliberately own their
|
||||
/// retry loops separately, but both attempts must receive identical handler
|
||||
/// semantics.
|
||||
pub(crate) async fn execute_single_attempt(
|
||||
node: &GvNode,
|
||||
context: &Context,
|
||||
graph: &GvGraph,
|
||||
run_dir: &Path,
|
||||
services: &EngineServices,
|
||||
) -> CoreResult<Outcome> {
|
||||
let handler = services.registry.resolve(node);
|
||||
|
||||
let wf_context = artifact::resolve_context_for_execution(
|
||||
context,
|
||||
&services.run.run_store,
|
||||
&*services.run.sandbox,
|
||||
run_dir,
|
||||
)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
CoreError::handler(HandlerErrorDetail {
|
||||
retryable: true,
|
||||
failure: err.to_failure_detail(),
|
||||
})
|
||||
})?;
|
||||
let execution_snapshot = wf_context.snapshot();
|
||||
|
||||
let node_timeout = match handler.node_timeout_policy(node) {
|
||||
NodeTimeoutPolicy::ExecutorEnforced => node.timeout(),
|
||||
NodeTimeoutPolicy::HandlerManaged => None,
|
||||
};
|
||||
|
||||
let future = dispatch_handler(handler, node, &wf_context, graph, run_dir, services);
|
||||
let panic_safe = AssertUnwindSafe(future).catch_unwind();
|
||||
let timed_result = if let Some(duration) = node_timeout {
|
||||
match timeout(duration, panic_safe).await {
|
||||
Ok(inner) => inner,
|
||||
Err(_elapsed) => {
|
||||
let mut failure = FailureDetail::new(
|
||||
format!("handler timed out after {}ms", duration.as_millis()),
|
||||
FailureCategory::TransientInfra,
|
||||
);
|
||||
failure.system_actor = Some(SystemActorKind::Timeout);
|
||||
return Err(CoreError::handler(HandlerErrorDetail {
|
||||
retryable: true,
|
||||
failure,
|
||||
}));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
panic_safe.await
|
||||
};
|
||||
|
||||
let mut new_values = wf_context.snapshot();
|
||||
artifact::normalize_durable_updates(&mut new_values);
|
||||
for (key, value) in &new_values {
|
||||
if execution_snapshot.get(key) != Some(value) {
|
||||
context.set(key.clone(), value.clone());
|
||||
}
|
||||
}
|
||||
|
||||
match timed_result {
|
||||
Ok(Ok(wf_outcome)) => Ok(wf_outcome),
|
||||
Ok(Err(Error::Cancelled)) => Err(CoreError::Cancelled),
|
||||
Ok(Err(fabro_err)) => {
|
||||
let retryable = handler.should_retry(&fabro_err);
|
||||
Err(CoreError::handler(HandlerErrorDetail {
|
||||
retryable,
|
||||
failure: fabro_err.to_failure_detail(),
|
||||
}))
|
||||
}
|
||||
Err(panic_payload) => {
|
||||
let msg = format_panic_message(&panic_payload);
|
||||
Err(CoreError::handler(HandlerErrorDetail {
|
||||
retryable: false,
|
||||
failure: FailureDetail::new(msg, FailureCategory::Deterministic),
|
||||
}))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn finalize_retries_exhausted(node: &GvNode, last_outcome: Outcome) -> Outcome {
|
||||
if node.allow_partial() {
|
||||
Outcome {
|
||||
status: StageOutcome::PartiallySucceeded,
|
||||
..last_outcome
|
||||
}
|
||||
} else {
|
||||
Outcome {
|
||||
status: StageOutcome::Failed {
|
||||
retry_requested: false,
|
||||
},
|
||||
..last_outcome
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
|
||||
async fn execute(
|
||||
|
|
@ -39,89 +139,14 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
|
|||
context: &Context,
|
||||
_graph: &WorkflowGraph,
|
||||
) -> CoreResult<Outcome> {
|
||||
let gv_node = node.inner();
|
||||
let handler = self.services.registry.resolve(gv_node);
|
||||
|
||||
let wf_context = artifact::resolve_context_for_execution(
|
||||
execute_single_attempt(
|
||||
node.inner(),
|
||||
context,
|
||||
&self.services.run.run_store,
|
||||
&*self.services.run.sandbox,
|
||||
&self.graph,
|
||||
&self.run_dir,
|
||||
&self.services,
|
||||
)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
CoreError::handler(HandlerErrorDetail {
|
||||
retryable: true,
|
||||
failure: err.to_failure_detail(),
|
||||
})
|
||||
})?;
|
||||
let execution_snapshot = wf_context.snapshot();
|
||||
|
||||
// Timeout from the node
|
||||
let node_timeout = match handler.node_timeout_policy(gv_node) {
|
||||
NodeTimeoutPolicy::ExecutorEnforced => gv_node.timeout(),
|
||||
NodeTimeoutPolicy::HandlerManaged => None,
|
||||
};
|
||||
|
||||
// Wrap with panic catch + timeout
|
||||
let run_dir = self.run_dir.clone();
|
||||
let future = dispatch_handler(
|
||||
handler,
|
||||
gv_node,
|
||||
&wf_context,
|
||||
&self.graph,
|
||||
&run_dir,
|
||||
&self.services,
|
||||
);
|
||||
let panic_safe = AssertUnwindSafe(future).catch_unwind();
|
||||
|
||||
let timed_result = if let Some(duration) = node_timeout {
|
||||
match timeout(duration, panic_safe).await {
|
||||
Ok(inner) => inner,
|
||||
Err(_elapsed) => {
|
||||
let mut failure = FailureDetail::new(
|
||||
format!("handler timed out after {}ms", duration.as_millis()),
|
||||
FailureCategory::TransientInfra,
|
||||
);
|
||||
failure.system_actor = Some(SystemActorKind::Timeout);
|
||||
return Err(CoreError::handler(HandlerErrorDetail {
|
||||
retryable: true,
|
||||
failure,
|
||||
}));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
panic_safe.await
|
||||
};
|
||||
|
||||
// 2. After handler returns, diff the forked context against the snapshot and
|
||||
// apply changes back to the original context
|
||||
let mut new_values = wf_context.snapshot();
|
||||
artifact::normalize_durable_updates(&mut new_values);
|
||||
for (k, v) in &new_values {
|
||||
if execution_snapshot.get(k) != Some(v) {
|
||||
context.set(k.clone(), v.clone());
|
||||
}
|
||||
}
|
||||
|
||||
match timed_result {
|
||||
Ok(Ok(wf_outcome)) => Ok(wf_outcome),
|
||||
Ok(Err(Error::Cancelled)) => Err(CoreError::Cancelled),
|
||||
Ok(Err(fabro_err)) => {
|
||||
let retryable = handler.should_retry(&fabro_err);
|
||||
Err(CoreError::handler(HandlerErrorDetail {
|
||||
retryable,
|
||||
failure: fabro_err.to_failure_detail(),
|
||||
}))
|
||||
}
|
||||
Err(panic_payload) => {
|
||||
let msg = format_panic_message(&panic_payload);
|
||||
Err(CoreError::handler(HandlerErrorDetail {
|
||||
retryable: false,
|
||||
failure: FailureDetail::new(msg, FailureCategory::Deterministic),
|
||||
}))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn context_for_edge_selection(
|
||||
|
|
@ -145,20 +170,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
|
|||
}
|
||||
|
||||
fn on_retries_exhausted(&self, node: &WorkflowNode, last_outcome: Outcome) -> Outcome {
|
||||
let gv_node = node.inner();
|
||||
if gv_node.allow_partial() {
|
||||
Outcome {
|
||||
status: StageOutcome::PartiallySucceeded,
|
||||
..last_outcome
|
||||
}
|
||||
} else {
|
||||
Outcome {
|
||||
status: StageOutcome::Failed {
|
||||
retry_requested: false,
|
||||
},
|
||||
..last_outcome
|
||||
}
|
||||
}
|
||||
finalize_retries_exhausted(node.inner(), last_outcome)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -155,6 +155,23 @@ impl StageExecutionTracker {
|
|||
Self::reserve_locked(&mut state, node_id, graph_visit)
|
||||
}
|
||||
|
||||
/// Allocate an execution ordinal without changing the node's active
|
||||
/// lifecycle scope or consuming resume provenance.
|
||||
///
|
||||
/// Parallel branch dispatches use detached reservations because several
|
||||
/// executions of one template node may run concurrently, while the parent
|
||||
/// parallel stage remains the owner of resume provenance.
|
||||
pub(crate) fn reserve_detached(&self, node_id: &str, graph_visit: u32) -> Arc<StageExecution> {
|
||||
let mut state = self.lock();
|
||||
let node = state.entry(node_id.to_owned()).or_default();
|
||||
node.high_water = node.high_water.saturating_add(1);
|
||||
Arc::new(StageExecution {
|
||||
stage_id: StageId::new(node_id, node.high_water),
|
||||
graph_visit,
|
||||
resumed_from: None,
|
||||
})
|
||||
}
|
||||
|
||||
/// The active scope for the node, reserving one only when none exists.
|
||||
/// Later attempts within one execution and checkpoint pre-steps reuse the
|
||||
/// first attempt's reservation.
|
||||
|
|
@ -303,6 +320,34 @@ mod tests {
|
|||
assert_eq!(second.resumed_from, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn detached_reservation_preserves_active_scope_and_resume_provenance() {
|
||||
let projection = projection_with_stages(&[("work", 1, 6)]);
|
||||
let seed = StageExecutionSeed::from_projection(&projection, 5);
|
||||
let tracker = StageExecutionTracker::seeded(seed);
|
||||
|
||||
let detached = tracker.reserve_detached("work", 1);
|
||||
assert_eq!(detached.stage_id, StageId::new("work", 2));
|
||||
assert_eq!(detached.resumed_from, None);
|
||||
assert_eq!(tracker.active("work"), None);
|
||||
|
||||
let normal = tracker.reserve("work", 1);
|
||||
assert_eq!(normal.stage_id, StageId::new("work", 3));
|
||||
assert_eq!(normal.resumed_from, Some(StageId::new("work", 1)));
|
||||
assert_eq!(tracker.active("work"), Some(normal));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn detached_reservation_does_not_replace_existing_active_scope() {
|
||||
let tracker = StageExecutionTracker::default();
|
||||
let active = tracker.reserve("work", 1);
|
||||
|
||||
let detached = tracker.reserve_detached("work", 1);
|
||||
|
||||
assert_eq!(detached.stage_id, StageId::new("work", 2));
|
||||
assert_eq!(tracker.active("work"), Some(active));
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn concurrent_reservations_stay_unique_per_node() {
|
||||
let tracker = StageExecutionTracker::default();
|
||||
|
|
@ -321,6 +366,25 @@ mod tests {
|
|||
assert_eq!(ordinals, (1..=8).collect::<Vec<_>>());
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn concurrent_detached_reservations_stay_unique_without_becoming_active() {
|
||||
let tracker = StageExecutionTracker::default();
|
||||
let handles: Vec<_> = (0..8)
|
||||
.map(|_| {
|
||||
let tracker = tracker.clone();
|
||||
tokio::spawn(async move { tracker.reserve_detached("branch", 1).stage_id.visit() })
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut ordinals = Vec::new();
|
||||
for handle in handles {
|
||||
ordinals.push(handle.await.expect("reservation task panicked"));
|
||||
}
|
||||
ordinals.sort_unstable();
|
||||
assert_eq!(ordinals, (1..=8).collect::<Vec<_>>());
|
||||
assert_eq!(tracker.active("branch"), None);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn concurrent_ensure_calls_reuse_one_reservation() {
|
||||
let tracker = StageExecutionTracker::default();
|
||||
|
|
|
|||
|
|
@ -7148,6 +7148,194 @@ mod real_llm {
|
|||
);
|
||||
}
|
||||
|
||||
#[fabro_macros::e2e_test(twin)]
|
||||
async fn twin_structured_array_flows_through_for_each_agents_to_fan_in() {
|
||||
use fabro_test::{TwinScenario, TwinScenarios};
|
||||
use fabro_workflow::handler::fan_in::FanInHandler;
|
||||
use fabro_workflow::handler::parallel::ParallelHandler;
|
||||
use fabro_workflow::handler::prompt::PromptHandler;
|
||||
|
||||
let twin = fabro_test::twin_openai().await;
|
||||
let namespace = format!("{}::for-each", module_path!());
|
||||
TwinScenarios::new(namespace.clone())
|
||||
.scenario(TwinScenario::responses("gpt-5.4-mini").text(
|
||||
r#"{"context_updates":{"candidates":[{"name":"auth","path":"src/auth.rs"},{"label":"api","path":"src/api.rs"}]}}"#,
|
||||
))
|
||||
.scenario(
|
||||
TwinScenario::responses("gpt-5.4-mini")
|
||||
.text("Reviewed the first security candidate."),
|
||||
)
|
||||
.scenario(
|
||||
TwinScenario::responses("gpt-5.4-mini")
|
||||
.text("Reviewed the second security candidate."),
|
||||
)
|
||||
.scenario(
|
||||
TwinScenario::responses("gpt-5.4-mini")
|
||||
.text("Combined both security reviews."),
|
||||
)
|
||||
.load(twin)
|
||||
.await;
|
||||
|
||||
let adapter: Arc<dyn fabro_llm::provider::ProviderAdapter> =
|
||||
Arc::new(OpenAiAdapter::new(namespace.clone()).with_base_url(twin.base_url.clone()));
|
||||
let providers = HashMap::from([("openai".to_string(), adapter)]);
|
||||
let client = Arc::new(Client::new(
|
||||
providers,
|
||||
Some("openai".to_string()),
|
||||
Vec::new(),
|
||||
));
|
||||
|
||||
let mut graph = Graph::new("ForEachSecurityReview");
|
||||
graph.attrs.insert(
|
||||
"goal".to_string(),
|
||||
AttrValue::String("Review runtime security candidates".to_string()),
|
||||
);
|
||||
|
||||
let mut start = Node::new("start");
|
||||
start.attrs.insert(
|
||||
"shape".to_string(),
|
||||
AttrValue::String("Mdiamond".to_string()),
|
||||
);
|
||||
let mut discover = Node::new("discover");
|
||||
discover
|
||||
.attrs
|
||||
.insert("shape".to_string(), AttrValue::String("tab".to_string()));
|
||||
discover.attrs.insert(
|
||||
"prompt".to_string(),
|
||||
AttrValue::String("Return the candidate array as routing context updates.".to_string()),
|
||||
);
|
||||
discover.attrs.insert(
|
||||
"output_schema".to_string(),
|
||||
AttrValue::String("routing".to_string()),
|
||||
);
|
||||
let mut fanout = Node::new("review_batch");
|
||||
fanout.attrs.insert(
|
||||
"shape".to_string(),
|
||||
AttrValue::String("component".to_string()),
|
||||
);
|
||||
fanout.attrs.insert(
|
||||
"for_each".to_string(),
|
||||
AttrValue::String("context.candidates".to_string()),
|
||||
);
|
||||
fanout
|
||||
.attrs
|
||||
.insert("max_parallel".to_string(), AttrValue::Integer(2));
|
||||
let mut reviewer = Node::new("reviewer");
|
||||
reviewer.attrs.insert(
|
||||
"prompt".to_string(),
|
||||
AttrValue::String("Review this security candidate.".to_string()),
|
||||
);
|
||||
let mut aggregate = Node::new("aggregate");
|
||||
aggregate.attrs.insert(
|
||||
"shape".to_string(),
|
||||
AttrValue::String("tripleoctagon".to_string()),
|
||||
);
|
||||
aggregate.attrs.insert(
|
||||
"prompt".to_string(),
|
||||
AttrValue::String("Synthesize every candidate review.".to_string()),
|
||||
);
|
||||
let mut exit = Node::new("exit");
|
||||
exit.attrs.insert(
|
||||
"shape".to_string(),
|
||||
AttrValue::String("Msquare".to_string()),
|
||||
);
|
||||
for node in [start, discover, fanout, reviewer, aggregate, exit] {
|
||||
graph.nodes.insert(node.id.clone(), node);
|
||||
}
|
||||
graph.edges.push(Edge::new("start", "discover"));
|
||||
graph.edges.push(Edge::new("discover", "review_batch"));
|
||||
graph.edges.push(Edge::new("review_batch", "reviewer"));
|
||||
graph.edges.push(Edge::new("reviewer", "aggregate"));
|
||||
graph.edges.push(Edge::new("aggregate", "exit"));
|
||||
|
||||
let emitter = Emitter::default();
|
||||
let events = super::collect_events(&emitter);
|
||||
let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(Some(
|
||||
make_llm_backend(Arc::clone(&client)),
|
||||
))));
|
||||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
registry.register(
|
||||
"prompt",
|
||||
Box::new(PromptHandler::new(Some(make_llm_backend(Arc::clone(
|
||||
&client,
|
||||
))))),
|
||||
);
|
||||
registry.register(
|
||||
"agent",
|
||||
Box::new(AgentHandler::new(Some(make_llm_backend(Arc::clone(
|
||||
&client,
|
||||
))))),
|
||||
);
|
||||
registry.register("parallel", Box::new(ParallelHandler));
|
||||
registry.register(
|
||||
"parallel.fan_in",
|
||||
Box::new(FanInHandler::new(Some(make_llm_backend(client)))),
|
||||
);
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(emitter), local_env());
|
||||
let run_options = RunOptions {
|
||||
settings: WorkflowSettings::default(),
|
||||
run_dir: dir.path().to_path_buf(),
|
||||
cancel_token: CancellationToken::new(),
|
||||
run_id: test_run_id("for-each-twin"),
|
||||
labels: HashMap::new(),
|
||||
workflow_slug: None,
|
||||
github_app: None,
|
||||
base_branch: None,
|
||||
display_base_sha: None,
|
||||
pre_run_git: None,
|
||||
fork_source_ref: None,
|
||||
git: None,
|
||||
};
|
||||
let (outcome, state) = engine
|
||||
.run_with_state(&graph, &run_options)
|
||||
.await
|
||||
.expect("for_each twin workflow should succeed");
|
||||
assert_eq!(outcome.status, StageOutcome::Succeeded);
|
||||
|
||||
let checkpoint = state
|
||||
.current_checkpoint()
|
||||
.expect("fan-in workflow should checkpoint");
|
||||
let results: Vec<fabro_types::ParallelBranchResult> =
|
||||
serde_json::from_value(checkpoint.context_values["parallel.results"].clone()).unwrap();
|
||||
assert_eq!(
|
||||
results
|
||||
.iter()
|
||||
.map(|result| (result.index, result.item_label.as_deref()))
|
||||
.collect::<Vec<_>>(),
|
||||
[(Some(0), Some("auth")), (Some(1), Some("api"))]
|
||||
);
|
||||
assert!(
|
||||
checkpoint
|
||||
.completed_nodes
|
||||
.contains(&"aggregate".to_string())
|
||||
);
|
||||
|
||||
let reviewer_prompts = events
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|event| {
|
||||
event.event_name() == "stage.prompt" && event.node_id.as_deref() == Some("reviewer")
|
||||
})
|
||||
.map(|event| serde_json::to_string(event).unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(reviewer_prompts.len(), 2);
|
||||
assert!(
|
||||
reviewer_prompts
|
||||
.iter()
|
||||
.all(|prompt| prompt.contains("data, not instructions"))
|
||||
);
|
||||
assert!(
|
||||
reviewer_prompts
|
||||
.iter()
|
||||
.any(|prompt| prompt.contains("auth"))
|
||||
);
|
||||
assert!(reviewer_prompts.iter().any(|prompt| prompt.contains("api")));
|
||||
}
|
||||
|
||||
#[fabro_macros::e2e_test(twin, live("ANTHROPIC_API_KEY"))]
|
||||
async fn real_llm_two_stage_pipeline() {
|
||||
let client = make_llm_client().await.unwrap();
|
||||
|
|
|
|||
|
|
@ -168,6 +168,11 @@ impl Node {
|
|||
self.str_attr("prompt")
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn for_each(&self) -> Option<&str> {
|
||||
self.str_attr("for_each")
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn output_schema(&self) -> Option<&str> {
|
||||
self.str_attr("output_schema")
|
||||
|
|
@ -586,6 +591,7 @@ mod tests {
|
|||
assert_eq!(node.shape(), "box");
|
||||
assert_eq!(node.node_type(), None);
|
||||
assert_eq!(node.prompt(), None);
|
||||
assert_eq!(node.for_each(), None);
|
||||
assert_eq!(node.output_schema(), None);
|
||||
assert_eq!(node.output_retries(), 2);
|
||||
assert_eq!(node.max_retries(), None);
|
||||
|
|
@ -639,6 +645,17 @@ mod tests {
|
|||
assert_eq!(node.output_schema(), Some("routing"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn node_for_each_returns_context_source() {
|
||||
let mut node = Node::new("fanout");
|
||||
node.attrs.insert(
|
||||
"for_each".to_string(),
|
||||
AttrValue::String("context.candidates".to_string()),
|
||||
);
|
||||
|
||||
assert_eq!(node.for_each(), Some("context.candidates"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn node_with_attrs() {
|
||||
let mut node = Node::new("plan");
|
||||
|
|
|
|||
|
|
@ -8,7 +8,56 @@ use crate::StageOutcome;
|
|||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct ParallelBranchResult {
|
||||
pub id: String,
|
||||
/// Zero-based input or outgoing-edge position. Absent only on records
|
||||
/// written before branch indexes became part of result identity.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub index: Option<usize>,
|
||||
/// Human-readable runtime item identity for `for_each` results.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub item_label: Option<String>,
|
||||
pub status: StageOutcome,
|
||||
#[serde(default)]
|
||||
pub context_updates: BTreeMap<String, Value>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use serde_json::json;
|
||||
|
||||
use super::ParallelBranchResult;
|
||||
use crate::StageOutcome;
|
||||
|
||||
#[test]
|
||||
fn indexed_result_round_trips_with_item_label() {
|
||||
let result = ParallelBranchResult {
|
||||
id: "review".to_string(),
|
||||
index: Some(3),
|
||||
item_label: Some("api".to_string()),
|
||||
status: StageOutcome::Succeeded,
|
||||
context_updates: BTreeMap::default(),
|
||||
};
|
||||
|
||||
let value = serde_json::to_value(&result).unwrap();
|
||||
assert_eq!(value["index"], 3);
|
||||
assert_eq!(value["item_label"], "api");
|
||||
assert_eq!(
|
||||
serde_json::from_value::<ParallelBranchResult>(value).unwrap(),
|
||||
result
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_result_without_index_or_label_still_deserializes() {
|
||||
let result: ParallelBranchResult = serde_json::from_value(json!({
|
||||
"id": "review",
|
||||
"status": "succeeded",
|
||||
"context_updates": {}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.index, None);
|
||||
assert_eq!(result.item_label, None);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,8 @@ pub struct ParallelStartedProps {
|
|||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct ParallelBranchStartedProps {
|
||||
pub index: usize,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub item_label: Option<String>,
|
||||
/// Graph visit of the branch target for this dispatch. The envelope
|
||||
/// `stage_id` ordinal counts executions, so a resumed fan-out's branches
|
||||
/// keep visit metadata even though their ordinals advanced.
|
||||
|
|
@ -35,6 +37,8 @@ pub struct ParallelBranchStartedProps {
|
|||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct ParallelBranchCompletedProps {
|
||||
pub index: usize,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub item_label: Option<String>,
|
||||
pub duration_ms: u64,
|
||||
pub status: StageOutcome,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,14 @@ import type { StageOutcome } from './stage-outcome';
|
|||
*/
|
||||
export interface ParallelBranchResult {
|
||||
'id': string;
|
||||
/**
|
||||
* Zero-based input item or outgoing-edge position. Absent only on parallel results written before indexed branch identity was added.
|
||||
*/
|
||||
'index'?: number;
|
||||
/**
|
||||
* Human-readable for_each item identity, derived from name, then label, then the zero-based input index.
|
||||
*/
|
||||
'item_label'?: string;
|
||||
'status': StageOutcome;
|
||||
'context_updates': { [key: string]: any; };
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue