mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-08-28 05:27:41 +00:00
Show live status for parallel branches
This commit is contained in:
Release Repro
parent
6efba896f4
commit
c812274db8
13 changed files with 497 additions and 97 deletions
|
|
@ -1,4 +1,5 @@
|
|||
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
|
||||
import { StageState } from "@qltysh/fabro-api-client";
|
||||
import type { EventEnvelope } from "@qltysh/fabro-api-client";
|
||||
import TestRenderer, { act } from "react-test-renderer";
|
||||
import { MemoryRouter } from "react-router";
|
||||
|
|
@ -13,35 +14,98 @@ beforeEach(() => {
|
|||
});
|
||||
afterEach(() => teardown());
|
||||
|
||||
const parallelStage: Stage = {
|
||||
function makeStage(overrides: Partial<Stage> = {}): Stage {
|
||||
return {
|
||||
id: "stage@1",
|
||||
name: "stage",
|
||||
handler: "agent",
|
||||
status: StageState.RUNNING,
|
||||
duration: "--",
|
||||
nodeId: "stage",
|
||||
visit: 1,
|
||||
graphVisit: 1,
|
||||
resumedFromStageId: null,
|
||||
parallelGroupId: null,
|
||||
parallelBranchIndex: null,
|
||||
startedAt: "2026-04-09T12:00:00Z",
|
||||
providerUsed: null,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
const parallelStage = makeStage({
|
||||
id: "fork@1",
|
||||
name: "fork",
|
||||
handler: "parallel",
|
||||
status: "succeeded",
|
||||
status: StageState.RUNNING,
|
||||
duration: "12s",
|
||||
nodeId: "fork",
|
||||
visit: 1,
|
||||
startedAt: "2026-04-09T12:00:00Z",
|
||||
providerUsed: null,
|
||||
};
|
||||
});
|
||||
|
||||
function event(partial: Partial<EventEnvelope>): EventEnvelope {
|
||||
return makeEventEnvelope(partial.seq ?? 1, { event: "parallel.completed", ...partial });
|
||||
function branchStage(
|
||||
name: string,
|
||||
index: number,
|
||||
status: StageState,
|
||||
groupId = "fork@1",
|
||||
visit = 1,
|
||||
): Stage {
|
||||
return makeStage({
|
||||
id: `${name}@${visit}`,
|
||||
name,
|
||||
nodeId: name,
|
||||
visit,
|
||||
status,
|
||||
parallelGroupId: groupId,
|
||||
parallelBranchIndex: index,
|
||||
});
|
||||
}
|
||||
|
||||
function renderParallel(events: EventEnvelope[]): TestRenderer.ReactTestRenderer {
|
||||
function event(partial: Partial<EventEnvelope>): EventEnvelope {
|
||||
return makeEventEnvelope(partial.seq ?? 1, {
|
||||
event: "parallel.completed",
|
||||
stage_id: "fork@1",
|
||||
...partial,
|
||||
});
|
||||
}
|
||||
|
||||
function startedEvent(branchCount: number): EventEnvelope {
|
||||
return event({
|
||||
event: "parallel.started",
|
||||
properties: { branch_count: branchCount },
|
||||
});
|
||||
}
|
||||
|
||||
function completedEvent(
|
||||
results: Array<{ id: string; status: string }>,
|
||||
successCount: number,
|
||||
failureCount: number,
|
||||
): EventEnvelope {
|
||||
return event({
|
||||
seq: 2,
|
||||
event: "parallel.completed",
|
||||
properties: {
|
||||
duration_ms: 12000,
|
||||
success_count: successCount,
|
||||
failure_count: failureCount,
|
||||
results: results.map((result) => ({ ...result, context_updates: {} })),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function renderParallel(
|
||||
events: EventEnvelope[],
|
||||
allStages: Stage[],
|
||||
stage = parallelStage,
|
||||
): TestRenderer.ReactTestRenderer {
|
||||
let renderer!: TestRenderer.ReactTestRenderer;
|
||||
act(() => {
|
||||
renderer = TestRenderer.create(
|
||||
<MemoryRouter>
|
||||
<ParallelChildren
|
||||
stage={parallelStage}
|
||||
stage={stage}
|
||||
events={events}
|
||||
runId="run-1"
|
||||
allStages={[
|
||||
{ ...parallelStage, id: "branch-a@1", name: "branch-a", nodeId: "branch-a", handler: "agent" },
|
||||
{ ...parallelStage, id: "branch-b@1", name: "branch-b", nodeId: "branch-b", handler: "agent", status: "failed" },
|
||||
]}
|
||||
allStages={allStages}
|
||||
/>
|
||||
</MemoryRouter>,
|
||||
);
|
||||
|
|
@ -49,37 +113,132 @@ function renderParallel(events: EventEnvelope[]): TestRenderer.ReactTestRenderer
|
|||
return renderer;
|
||||
}
|
||||
|
||||
describe("ParallelChildren", () => {
|
||||
test("renders branch status and stage links without checkout metadata", () => {
|
||||
const renderer = renderParallel([
|
||||
event({
|
||||
event: "parallel.started",
|
||||
properties: { branch_count: 2 },
|
||||
}),
|
||||
event({
|
||||
seq: 2,
|
||||
event: "parallel.completed",
|
||||
properties: {
|
||||
duration_ms: 12000,
|
||||
success_count: 1,
|
||||
failure_count: 1,
|
||||
results: [
|
||||
{ id: "branch-a", status: "succeeded", context_updates: {} },
|
||||
{ id: "branch-b", status: "failed", context_updates: {} },
|
||||
],
|
||||
},
|
||||
}),
|
||||
]);
|
||||
function textContent(node: TestRenderer.ReactTestInstance): string {
|
||||
return node.children
|
||||
.map((child) => typeof child === "string" ? child : textContent(child))
|
||||
.join("");
|
||||
}
|
||||
|
||||
const rendered = JSON.stringify(renderer.toJSON());
|
||||
expect(rendered).toContain("branch-a");
|
||||
expect(rendered).toContain("Succeeded");
|
||||
expect(rendered).toContain("branch-b");
|
||||
expect(rendered).toContain("Failed");
|
||||
const hrefs = renderer.root.findAllByType("a").map((link) => link.props.href);
|
||||
expect(hrefs).toEqual([
|
||||
"/runs/run-1/stages/branch-a@1",
|
||||
"/runs/run-1/stages/branch-b@1",
|
||||
function branchRowText(renderer: TestRenderer.ReactTestRenderer): string[] {
|
||||
return renderer.root.findAllByType("li").map(textContent);
|
||||
}
|
||||
|
||||
function hrefs(renderer: TestRenderer.ReactTestRenderer): string[] {
|
||||
return renderer.root.findAllByType("a").map((link) => link.props.href);
|
||||
}
|
||||
|
||||
function statValue(renderer: TestRenderer.ReactTestRenderer, label: string): string {
|
||||
const stat = renderer.root
|
||||
.findAllByProps({ className: "flex flex-col gap-0.5" })
|
||||
.find((item) => textContent(item).startsWith(label));
|
||||
if (!stat) throw new Error(`stat ${label} not found`);
|
||||
return textContent(stat.findAllByType("span")[1]);
|
||||
}
|
||||
|
||||
describe("ParallelChildren", () => {
|
||||
test("renders live branch names, statuses, counts, and stage links", () => {
|
||||
const renderer = renderParallel(
|
||||
[startedEvent(2)],
|
||||
[
|
||||
branchStage("review_glm", 0, StageState.SUCCEEDED),
|
||||
branchStage("review_opus", 1, StageState.RUNNING),
|
||||
],
|
||||
);
|
||||
|
||||
const rows = branchRowText(renderer);
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(rows[0]).toContain("Succeeded");
|
||||
expect(rows[0]).toContain("review_glm");
|
||||
expect(rows[1]).toContain("Running");
|
||||
expect(rows[1]).toContain("review_opus");
|
||||
expect(hrefs(renderer)).toEqual([
|
||||
"/runs/run-1/stages/review_glm@1",
|
||||
"/runs/run-1/stages/review_opus@1",
|
||||
]);
|
||||
expect(statValue(renderer, "Succeeded")).toBe("1");
|
||||
expect(statValue(renderer, "Failed")).toBe("0");
|
||||
});
|
||||
|
||||
test("keeps looped fork links scoped to the selected fork visit", () => {
|
||||
const renderer = renderParallel(
|
||||
[startedEvent(1)],
|
||||
[
|
||||
branchStage("review_glm", 0, StageState.SUCCEEDED, "fork@1", 1),
|
||||
branchStage("review_glm", 0, StageState.RUNNING, "fork@2", 2),
|
||||
],
|
||||
);
|
||||
|
||||
expect(hrefs(renderer)).toEqual(["/runs/run-1/stages/review_glm@1"]);
|
||||
});
|
||||
|
||||
test("keeps duplicate branch targets in index order and only links recorded stages", () => {
|
||||
const renderer = renderParallel(
|
||||
[
|
||||
startedEvent(2),
|
||||
completedEvent(
|
||||
[
|
||||
{ id: "review", status: "failed" },
|
||||
{ id: "review", status: "failed" },
|
||||
],
|
||||
1,
|
||||
1,
|
||||
),
|
||||
],
|
||||
[branchStage("review", 0, StageState.SUCCEEDED)],
|
||||
);
|
||||
|
||||
const rows = branchRowText(renderer);
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(rows[0]).toContain("Succeeded");
|
||||
expect(rows[1]).toContain("Failed");
|
||||
expect(hrefs(renderer)).toEqual(["/runs/run-1/stages/review@1"]);
|
||||
});
|
||||
|
||||
test("renders a completed result without a matching stage as an unlinked row", () => {
|
||||
const renderer = renderParallel(
|
||||
[
|
||||
startedEvent(1),
|
||||
completedEvent(
|
||||
[{ id: "legacy_branch", status: "succeeded" }],
|
||||
1,
|
||||
0,
|
||||
),
|
||||
],
|
||||
[],
|
||||
);
|
||||
|
||||
expect(branchRowText(renderer)).toEqual(["Succeededlegacy_branch"]);
|
||||
expect(hrefs(renderer)).toEqual([]);
|
||||
});
|
||||
|
||||
test("counts partial and skipped branches as neither succeeded nor failed", () => {
|
||||
const allStages = [
|
||||
branchStage("partial", 0, StageState.PARTIALLY_SUCCEEDED),
|
||||
branchStage("skipped", 1, StageState.SKIPPED),
|
||||
];
|
||||
const running = renderParallel([startedEvent(2)], allStages);
|
||||
const completed = renderParallel(
|
||||
[
|
||||
startedEvent(2),
|
||||
completedEvent(
|
||||
[
|
||||
{ id: "partial", status: "partially_succeeded" },
|
||||
{ id: "skipped", status: "skipped" },
|
||||
],
|
||||
0,
|
||||
0,
|
||||
),
|
||||
],
|
||||
allStages,
|
||||
);
|
||||
|
||||
expect([
|
||||
statValue(running, "Succeeded"),
|
||||
statValue(running, "Failed"),
|
||||
]).toEqual(["0", "0"]);
|
||||
expect([
|
||||
statValue(completed, "Succeeded"),
|
||||
statValue(completed, "Failed"),
|
||||
]).toEqual(["0", "0"]);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -10,10 +10,12 @@ import { formatDurationMs } from "../../lib/format";
|
|||
import { StageMetaBar } from "./meta-bar";
|
||||
import { parseParallelOverview } from "./helpers";
|
||||
|
||||
/** Branch row view state: completed outcomes plus a synthesized in-flight row. */
|
||||
/** Branch row view state sourced from a live branch stage or completed result. */
|
||||
interface BranchRow {
|
||||
branchIndex: number;
|
||||
id: string;
|
||||
status: StageState;
|
||||
stageHref: string | null;
|
||||
}
|
||||
|
||||
function StatItem({
|
||||
|
|
@ -38,25 +40,23 @@ function StatItem({
|
|||
}
|
||||
|
||||
function ChildRow({
|
||||
result,
|
||||
stageHref,
|
||||
row,
|
||||
}: {
|
||||
result: BranchRow;
|
||||
stageHref: string | null;
|
||||
row: BranchRow;
|
||||
}) {
|
||||
const tone = stageStatusTone(result.status);
|
||||
const tone = stageStatusTone(row.status);
|
||||
|
||||
const inner = (
|
||||
<>
|
||||
<span
|
||||
className={`inline-flex w-24 shrink-0 justify-center rounded-full px-2 py-0.5 text-[10px] font-medium uppercase tracking-wider ${tone}`}
|
||||
>
|
||||
{stageStatusLabel(result.status)}
|
||||
{stageStatusLabel(row.status)}
|
||||
</span>
|
||||
<span className="min-w-0 flex-1 truncate font-mono text-sm text-fg-3">
|
||||
{result.id}
|
||||
{row.id}
|
||||
</span>
|
||||
{stageHref && (
|
||||
{row.stageHref && (
|
||||
<ArrowTopRightOnSquareIcon
|
||||
className="size-3.5 shrink-0 text-fg-muted transition-colors group-hover:text-fg-2"
|
||||
aria-hidden="true"
|
||||
|
|
@ -67,9 +67,9 @@ function ChildRow({
|
|||
|
||||
return (
|
||||
<li className="flex items-center gap-3 px-4 py-2.5">
|
||||
{stageHref ? (
|
||||
{row.stageHref ? (
|
||||
<Link
|
||||
to={stageHref}
|
||||
to={row.stageHref}
|
||||
className="group flex flex-1 items-center gap-3 rounded -m-1 p-1 transition-colors hover:bg-overlay focus-visible:bg-overlay focus-visible:outline-2 focus-visible:-outline-offset-2 focus-visible:outline-teal-500"
|
||||
>
|
||||
{inner}
|
||||
|
|
@ -94,24 +94,63 @@ export function ParallelChildren({
|
|||
}) {
|
||||
const overview = useMemo(() => parseParallelOverview(events), [events]);
|
||||
|
||||
// Map node_id -> latest stage_id so we can deep-link branches.
|
||||
const latestStageByNode = useMemo(() => {
|
||||
const latest = new Map<string, Stage>();
|
||||
for (const s of allStages) {
|
||||
const prev = latest.get(s.nodeId);
|
||||
if (!prev || s.visit > prev.visit) latest.set(s.nodeId, s);
|
||||
const stagesByBranchIndex = useMemo(() => {
|
||||
const byIndex = new Map<number, Stage>();
|
||||
for (const candidate of allStages) {
|
||||
if (
|
||||
candidate.parallelGroupId === stage.id
|
||||
&& candidate.parallelBranchIndex != null
|
||||
) {
|
||||
byIndex.set(candidate.parallelBranchIndex, candidate);
|
||||
}
|
||||
}
|
||||
return new Map(Array.from(latest.entries()).map(([nodeId, s]) => [nodeId, s.id]));
|
||||
}, [allStages]);
|
||||
return byIndex;
|
||||
}, [allStages, stage.id]);
|
||||
|
||||
const items: BranchRow[] = overview.results.length > 0
|
||||
? overview.results
|
||||
: overview.branchCount && overview.branchCount > 0
|
||||
? Array.from({ length: overview.branchCount }, (_, i) => ({
|
||||
id: `branch ${i + 1}`,
|
||||
status: StageState.RUNNING,
|
||||
}))
|
||||
: [];
|
||||
const branchCount = overview.branchCount ?? stagesByBranchIndex.size;
|
||||
const slots = Array.from({ length: branchCount }, (_, index) => ({
|
||||
index,
|
||||
stage: stagesByBranchIndex.get(index) ?? null,
|
||||
result: overview.results[index] ?? null,
|
||||
}));
|
||||
const rows = slots.map<BranchRow>(({ index, stage: branchStage, result }) => {
|
||||
if (branchStage) {
|
||||
return {
|
||||
branchIndex: index,
|
||||
id: branchStage.name,
|
||||
status: branchStage.status,
|
||||
stageHref: `/runs/${runId}/stages/${branchStage.id}`,
|
||||
};
|
||||
}
|
||||
if (result) {
|
||||
return {
|
||||
branchIndex: index,
|
||||
id: result.id,
|
||||
status: result.status,
|
||||
stageHref: null,
|
||||
};
|
||||
}
|
||||
return {
|
||||
branchIndex: index,
|
||||
id: `branch ${index + 1}`,
|
||||
status: StageState.PENDING,
|
||||
stageHref: null,
|
||||
};
|
||||
});
|
||||
|
||||
const liveBranchStages = Array.from(stagesByBranchIndex.values());
|
||||
const liveSuccessCount = liveBranchStages
|
||||
.filter((branchStage) => branchStage.status === StageState.SUCCEEDED)
|
||||
.length;
|
||||
const liveFailureCount = liveBranchStages
|
||||
.filter((branchStage) => branchStage.status === StageState.FAILED)
|
||||
.length;
|
||||
const successCount = overview.isComplete
|
||||
? overview.successCount ?? 0
|
||||
: liveSuccessCount;
|
||||
const failureCount = overview.isComplete
|
||||
? overview.failureCount ?? 0
|
||||
: liveFailureCount;
|
||||
|
||||
return (
|
||||
<div className="space-y-6 pl-3 pr-4 sm:pr-6 lg:pr-8">
|
||||
|
|
@ -121,13 +160,13 @@ export function ParallelChildren({
|
|||
<StatItem label="Branches" value={overview.branchCount ?? "—"} />
|
||||
<StatItem
|
||||
label="Succeeded"
|
||||
value={overview.successCount ?? (overview.isComplete ? 0 : "—")}
|
||||
value={successCount}
|
||||
tone="success"
|
||||
/>
|
||||
<StatItem
|
||||
label="Failed"
|
||||
value={overview.failureCount ?? (overview.isComplete ? 0 : "—")}
|
||||
tone={overview.failureCount && overview.failureCount > 0 ? "danger" : "default"}
|
||||
value={failureCount}
|
||||
tone={failureCount > 0 ? "danger" : "default"}
|
||||
/>
|
||||
<StatItem
|
||||
label="Duration"
|
||||
|
|
@ -139,21 +178,13 @@ export function ParallelChildren({
|
|||
<h3 className="mb-2 text-xs font-medium uppercase tracking-wider text-fg-muted">
|
||||
Branches
|
||||
</h3>
|
||||
{items.length === 0 ? (
|
||||
{rows.length === 0 ? (
|
||||
<p className="text-sm text-fg-muted">No branches recorded yet.</p>
|
||||
) : (
|
||||
<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);
|
||||
const href = stageId ? `/runs/${runId}/stages/${stageId}` : null;
|
||||
return (
|
||||
<ChildRow
|
||||
key={`${result.id}-${i}`}
|
||||
result={result}
|
||||
stageHref={href}
|
||||
/>
|
||||
);
|
||||
})}
|
||||
{rows.map((row) => (
|
||||
<ChildRow key={row.branchIndex} row={row} />
|
||||
))}
|
||||
</ul>
|
||||
)}
|
||||
</section>
|
||||
|
|
|
|||
|
|
@ -13,6 +13,8 @@ function makeStage(nodeId: string, visit: number, status: StageState): Stage {
|
|||
visit,
|
||||
graphVisit: null,
|
||||
resumedFromStageId: null,
|
||||
parallelGroupId: null,
|
||||
parallelBranchIndex: null,
|
||||
status,
|
||||
duration: "--",
|
||||
startedAt: null,
|
||||
|
|
@ -182,6 +184,28 @@ describe("mapRunStagesToSidebarStages", () => {
|
|||
expect(result[0].resumedFromStageId).toBeNull();
|
||||
});
|
||||
|
||||
test("maps parallel branch identity without parsing it in the client", () => {
|
||||
const stages: PaginatedRunStageList = {
|
||||
data: [
|
||||
{
|
||||
id: "review_opus@2",
|
||||
name: "review_opus",
|
||||
handler: "agent",
|
||||
status: "running",
|
||||
node_id: "review_opus",
|
||||
visit: 2,
|
||||
parallel_group_id: "review_fork@1",
|
||||
parallel_branch_index: 3,
|
||||
},
|
||||
],
|
||||
meta: { has_more: false },
|
||||
};
|
||||
|
||||
const result = mapRunStagesToSidebarStages(stages);
|
||||
expect(result[0].parallelGroupId).toBe("review_fork@1");
|
||||
expect(result[0].parallelBranchIndex).toBe(3);
|
||||
});
|
||||
|
||||
test("preserves the authoritative handler for renderer dispatch", () => {
|
||||
const stages: PaginatedRunStageList = {
|
||||
data: [
|
||||
|
|
|
|||
|
|
@ -25,6 +25,10 @@ export interface Stage {
|
|||
graphVisit: number | null;
|
||||
/** StageId of the prior execution superseded by this resumed replay, if any. */
|
||||
resumedFromStageId: string | null;
|
||||
/** Exact StageId of the parent parallel execution, if this is a branch. */
|
||||
parallelGroupId: string | null;
|
||||
/** Zero-based outgoing-edge index within the parent parallel execution. */
|
||||
parallelBranchIndex: number | null;
|
||||
startedAt: string | null;
|
||||
providerUsed: StageModelUsage | null;
|
||||
}
|
||||
|
|
@ -96,6 +100,8 @@ export function mapRunStagesToSidebarStages(
|
|||
visit: stage.visit,
|
||||
graphVisit: stage.graph_visit ?? null,
|
||||
resumedFromStageId: stage.resumed_from_stage_id ?? null,
|
||||
parallelGroupId: stage.parallel_group_id ?? null,
|
||||
parallelBranchIndex: stage.parallel_branch_index ?? null,
|
||||
status: stage.status,
|
||||
duration: stage.wall_time_ms != null
|
||||
? formatDurationMs(stage.wall_time_ms)
|
||||
|
|
|
|||
|
|
@ -10652,6 +10652,11 @@ components:
|
|||
items:
|
||||
$ref: "#/components/schemas/ParallelBranchResult"
|
||||
description: Ordered per-branch results produced by a parallel stage.
|
||||
parallel_branch_id:
|
||||
type: ["string", "null"]
|
||||
description: >
|
||||
Durable identity of this branch within a parallel execution,
|
||||
formatted as "{parallel_group_id}:{index}".
|
||||
output:
|
||||
type: ["string", "null"]
|
||||
output_bytes:
|
||||
|
|
@ -12658,6 +12663,23 @@ components:
|
|||
StageId of the prior post-checkpoint execution superseded by this
|
||||
replay after the run was resumed.
|
||||
example: verify@1
|
||||
parallel_group_id:
|
||||
oneOf:
|
||||
- $ref: "#/components/schemas/StageId"
|
||||
- type: "null"
|
||||
description: >-
|
||||
Exact StageId of the parent parallel execution. Clients can compare
|
||||
this directly with a parallel stage's `id`. Null for stages that
|
||||
are not parallel branches.
|
||||
example: review_fork@1
|
||||
parallel_branch_index:
|
||||
type: ["integer", "null"]
|
||||
format: uint32
|
||||
minimum: 0
|
||||
description: >-
|
||||
Zero-based outgoing-edge index within the parent parallel
|
||||
execution. Null for stages that are not parallel branches.
|
||||
example: 1
|
||||
provider_used:
|
||||
oneOf:
|
||||
- $ref: "#/components/schemas/StageModelUsage"
|
||||
|
|
|
|||
|
|
@ -1143,6 +1143,8 @@ mod runs {
|
|||
started_at: None,
|
||||
graph_visit: None,
|
||||
resumed_from_stage_id: None,
|
||||
parallel_group_id: None,
|
||||
parallel_branch_index: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -32,6 +32,11 @@ fn run_stage_from_projection(
|
|||
.and_then(|node| node.handler_type()),
|
||||
)
|
||||
});
|
||||
let (parallel_group_id, parallel_branch_index) = stage
|
||||
.parallel_branch_id
|
||||
.as_ref()
|
||||
.map(|branch_id| (branch_id.group().clone(), branch_id.index()))
|
||||
.unzip();
|
||||
RunStage {
|
||||
id: stage_id.clone(),
|
||||
name: stage_id.node_id().to_owned(),
|
||||
|
|
@ -45,6 +50,8 @@ fn run_stage_from_projection(
|
|||
started_at: stage.started_at,
|
||||
graph_visit: stage.graph_visit.and_then(std::num::NonZeroU32::new),
|
||||
resumed_from_stage_id: stage.resumed_from_stage_id.clone(),
|
||||
parallel_group_id,
|
||||
parallel_branch_index,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -26,8 +26,8 @@ use fabro_types::settings::ServerAuthMethod;
|
|||
use fabro_types::settings::run::EnvironmentProvider;
|
||||
use fabro_types::{
|
||||
AgentBackend, AttrValue, AuthMethod, CommandTermination, FailureCategory, FailureDetail, Graph,
|
||||
InterviewQuestionRecord, Node, Outcome, QuestionType, RunBlobId, RunId, RunSpec,
|
||||
SandboxProviderKind, StageContextWindowBreakdownItem, StageContextWindowCategory,
|
||||
InterviewQuestionRecord, Node, Outcome, ParallelBranchId, QuestionType, RunBlobId, RunId,
|
||||
RunSpec, SandboxProviderKind, StageContextWindowBreakdownItem, StageContextWindowCategory,
|
||||
StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness,
|
||||
StageContextWindowWarning, StageModelUsage, StageTiming, SuccessReason, SystemActorKind,
|
||||
WorkflowSettings, fixtures, test_support,
|
||||
|
|
@ -5503,6 +5503,81 @@ async fn list_run_stages_exposes_execution_identity_for_resumed_stage() {
|
|||
assert_eq!(second["resumed_from_stage_id"], "work@1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_run_stages_exposes_parallel_branch_identity() {
|
||||
let state = test_app_state_with_isolated_storage();
|
||||
let app = crate::test_support::build_test_router(Arc::clone(&state));
|
||||
let run_id = RunId::new();
|
||||
|
||||
create_durable_run_with_events(&state, run_id, &[
|
||||
workflow_event::Event::RunSubmitted {
|
||||
definition_blob: None,
|
||||
},
|
||||
workflow_event::Event::RunStarting,
|
||||
workflow_event::Event::RunRunning,
|
||||
])
|
||||
.await;
|
||||
append_scoped_stage_event(
|
||||
&state,
|
||||
run_id,
|
||||
"ordinary",
|
||||
1,
|
||||
&workflow_event::Event::StageStarted {
|
||||
graph_visit: Some(1),
|
||||
resumed_from_stage_id: None,
|
||||
node_id: "ordinary".to_string(),
|
||||
name: "Ordinary".to_string(),
|
||||
index: 0,
|
||||
handler_type: "agent".to_string(),
|
||||
attempt: 1,
|
||||
max_attempts: 1,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
let parallel_group_id = StageId::new("review_fork", 2);
|
||||
let parallel_branch_id = ParallelBranchId::new(parallel_group_id.clone(), 4);
|
||||
let branch_event = workflow_event::Event::ParallelBranchStarted {
|
||||
parallel_group_id: parallel_group_id.clone(),
|
||||
parallel_branch_id: parallel_branch_id.clone(),
|
||||
branch: "review_glm".to_string(),
|
||||
index: 4,
|
||||
graph_visit: Some(3),
|
||||
resumed_from_stage_id: None,
|
||||
};
|
||||
let branch_scope = workflow_event::StageScope::for_parallel_branch(
|
||||
"review_glm",
|
||||
3,
|
||||
parallel_group_id,
|
||||
parallel_branch_id,
|
||||
);
|
||||
let stored =
|
||||
workflow_event::to_run_event_at(&run_id, &branch_event, Utc::now(), Some(&branch_scope));
|
||||
let payload = workflow_event::build_redacted_event_payload(&stored, &run_id).unwrap();
|
||||
let run_store = state.stores.runs.open_run(&run_id).await.unwrap();
|
||||
run_store.append_event(&payload).await.unwrap();
|
||||
|
||||
let response = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("GET")
|
||||
.uri(api(&format!("/runs/{run_id}/stages")))
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = response_json!(response, StatusCode::OK).await;
|
||||
|
||||
let branch = stage_entry(&body, "review_glm@3");
|
||||
assert_eq!(branch["parallel_group_id"], "review_fork@2");
|
||||
assert_eq!(branch["parallel_branch_index"], 4);
|
||||
|
||||
let ordinary = stage_entry(&body, "ordinary@1");
|
||||
assert!(ordinary.get("parallel_group_id").is_none());
|
||||
assert!(ordinary.get("parallel_branch_index").is_none());
|
||||
}
|
||||
|
||||
/// `checkpoint.completed_nodes` records every visit, so a looped node appears
|
||||
/// once per re-entry. Billing must dedup so a retried node renders as one row
|
||||
/// and `runtime_secs` is summed across all visits exactly once.
|
||||
|
|
|
|||
|
|
@ -609,6 +609,9 @@ impl RunProjectionReducer for RunProjection {
|
|||
stage
|
||||
.resumed_from_stage_id
|
||||
.clone_from(&props.resumed_from_stage_id);
|
||||
stage
|
||||
.parallel_branch_id
|
||||
.clone_from(&stored.parallel_branch_id);
|
||||
}
|
||||
stage.state = StageState::Running;
|
||||
}
|
||||
|
|
@ -1548,9 +1551,9 @@ mod tests {
|
|||
AgentBackend, AgentControlState, AutomationRef, BilledModelUsage, BilledTokenCounts,
|
||||
BlockedReason, Checkpoint, CheckpointRecord, CommandTermination, EventBody,
|
||||
FailureCategory, FailureDetail, FailureReason, Graph, McpServerStatus, Outcome,
|
||||
PendingReason, PermissionLevel, PullRequestLink, QuestionType, ReasoningEffort,
|
||||
RunApprovalState, RunBlobId, RunControlAction, RunDiff, RunEvent, RunSize, RunSpec,
|
||||
RunStatus, Speed, StageContextWindowBreakdownItem, StageContextWindowCategory,
|
||||
ParallelBranchId, PendingReason, PermissionLevel, PullRequestLink, QuestionType,
|
||||
ReasoningEffort, RunApprovalState, RunBlobId, RunControlAction, RunDiff, RunEvent, RunSize,
|
||||
RunSpec, RunStatus, Speed, StageContextWindowBreakdownItem, StageContextWindowCategory,
|
||||
StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness,
|
||||
StageContextWindowWarning, StageModelUsage, StageOutcome, StageState, SubAgentStatus,
|
||||
SuccessReason, WorkflowSettings, first_event_seq, fixtures, test_support,
|
||||
|
|
@ -2277,6 +2280,56 @@ mod tests {
|
|||
assert_eq!(stage.prompt.as_deref(), Some("prompt"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parallel_branch_started_projects_identity_without_churning_it() {
|
||||
let mut state = initialized_projection();
|
||||
let group_id = StageId::new("review_fork", 1);
|
||||
let branch_stage_id = StageId::new("review_glm", 1);
|
||||
let branch_id = ParallelBranchId::new(group_id.clone(), 0);
|
||||
let mut started = test_stage_event(
|
||||
3,
|
||||
EventBody::ParallelBranchStarted(ParallelBranchStartedProps {
|
||||
index: 0,
|
||||
graph_visit: Some(1),
|
||||
resumed_from_stage_id: None,
|
||||
}),
|
||||
branch_stage_id.clone(),
|
||||
);
|
||||
started.event.parallel_group_id = Some(group_id.clone());
|
||||
started.event.parallel_branch_id = Some(branch_id.clone());
|
||||
|
||||
state.apply_event(&started).unwrap();
|
||||
|
||||
assert_eq!(
|
||||
state
|
||||
.stage(&branch_stage_id)
|
||||
.unwrap()
|
||||
.parallel_branch_id
|
||||
.as_ref(),
|
||||
Some(&branch_id)
|
||||
);
|
||||
|
||||
let replacement_branch_id = ParallelBranchId::new(group_id.clone(), 1);
|
||||
let mut reobserved = test_stage_event(
|
||||
4,
|
||||
EventBody::ParallelBranchStarted(ParallelBranchStartedProps {
|
||||
index: 1,
|
||||
graph_visit: Some(2),
|
||||
resumed_from_stage_id: Some(StageId::new("review_glm", 2)),
|
||||
}),
|
||||
branch_stage_id.clone(),
|
||||
);
|
||||
reobserved.event.parallel_group_id = Some(group_id);
|
||||
reobserved.event.parallel_branch_id = Some(replacement_branch_id);
|
||||
|
||||
state.apply_event(&reobserved).unwrap();
|
||||
|
||||
let stage = state.stage(&branch_stage_id).unwrap();
|
||||
assert_eq!(stage.parallel_branch_id.as_ref(), Some(&branch_id));
|
||||
assert_eq!(stage.graph_visit, Some(1));
|
||||
assert!(stage.resumed_from_stage_id.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parallel_branch_completed_finalizes_branch_stage() {
|
||||
// A parallel branch never runs through the engine's StageStarted/
|
||||
|
|
|
|||
|
|
@ -26,11 +26,11 @@ use fabro_types::{
|
|||
ActivatedSkill, AgentControlState, AgentMcpToolSummary, AgentSkillActivationSource,
|
||||
AgentSkillSummary, AgentToolCategory, AgentToolSource, AgentToolSummary,
|
||||
AgentToolsAvailableProps, LlmOutputKind, McpServerProjection, McpServerStatus,
|
||||
ParallelBranchResult, PermissionLevel, SkillsProjection, StageContextWindow,
|
||||
ParallelBranchId, ParallelBranchResult, PermissionLevel, SkillsProjection, StageContextWindow,
|
||||
StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
|
||||
StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowUnavailableReason,
|
||||
StageContextWindowWarning, StageInferenceProjection, StageProjection, SubAgentProjection,
|
||||
SubAgentStatus, TodoListKind, TodoListProjection,
|
||||
StageContextWindowWarning, StageId, StageInferenceProjection, StageProjection,
|
||||
SubAgentProjection, SubAgentStatus, TodoListKind, TodoListProjection,
|
||||
};
|
||||
use serde_json::json;
|
||||
|
||||
|
|
@ -188,6 +188,7 @@ fn stage_projection_round_trips_representative_json() {
|
|||
"context_updates": {}
|
||||
}
|
||||
],
|
||||
"parallel_branch_id": "review_fork@3:1",
|
||||
"output": "ok",
|
||||
"termination": "exited",
|
||||
"started_at": "2026-04-29T12:34:00Z",
|
||||
|
|
@ -316,6 +317,10 @@ fn stage_projection_round_trips_representative_json() {
|
|||
});
|
||||
|
||||
let state: StageProjection = serde_json::from_value(value.clone()).unwrap();
|
||||
assert_eq!(
|
||||
state.parallel_branch_id,
|
||||
Some(ParallelBranchId::new(StageId::new("review_fork", 3), 1))
|
||||
);
|
||||
assert_eq!(serde_json::to_value(state).unwrap(), value);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,9 +10,10 @@ use crate::run_event::{AgentSessionActivatedProps, StagePromptProps};
|
|||
use crate::{
|
||||
AgentBackend, AgentMcpToolSummary, AgentSkillActivationSource, AgentSkillSummary,
|
||||
AgentToolSummary, BilledTokenCounts, Checkpoint, Conclusion, InterviewQuestionRecord,
|
||||
InvalidTransition, LlmOutputKind, ModelRef, PermissionLevel, PullRequestLink, RunApproval,
|
||||
RunControlAction, RunDiff, RunId, RunSandbox, RunSpec, RunStatus, RunTiming, StageCompletion,
|
||||
StageHandler, StageId, StageState, StageTiming, StartRecord, TodoListProjection,
|
||||
InvalidTransition, LlmOutputKind, ModelRef, ParallelBranchId, PermissionLevel, PullRequestLink,
|
||||
RunApproval, RunControlAction, RunDiff, RunId, RunSandbox, RunSpec, RunStatus, RunTiming,
|
||||
StageCompletion, StageHandler, StageId, StageState, StageTiming, StartRecord,
|
||||
TodoListProjection,
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
|
|
@ -328,6 +329,8 @@ pub struct StageProjection {
|
|||
pub script_invocation: Option<serde_json::Value>,
|
||||
pub script_timing: Option<serde_json::Value>,
|
||||
pub parallel_results: Option<Vec<crate::ParallelBranchResult>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub parallel_branch_id: Option<ParallelBranchId>,
|
||||
pub output: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub output_bytes: Option<u64>,
|
||||
|
|
@ -522,6 +525,7 @@ impl StageProjection {
|
|||
script_invocation: None,
|
||||
script_timing: None,
|
||||
parallel_results: None,
|
||||
parallel_branch_id: None,
|
||||
output: None,
|
||||
output_bytes: None,
|
||||
live_streaming: None,
|
||||
|
|
|
|||
|
|
@ -57,6 +57,14 @@ export interface RunStage {
|
|||
* Canonical stage execution identifier in `node_id@visit` form.
|
||||
*/
|
||||
'resumed_from_stage_id'?: string | null;
|
||||
/**
|
||||
* Canonical stage execution identifier in `node_id@visit` form.
|
||||
*/
|
||||
'parallel_group_id'?: string | null;
|
||||
/**
|
||||
* Zero-based outgoing-edge index within the parent parallel execution. Null for stages that are not parallel branches.
|
||||
*/
|
||||
'parallel_branch_index'?: number | null;
|
||||
'provider_used'?: StageModelUsage | null;
|
||||
/**
|
||||
* Wall-clock time the latest attempt of this stage started, if known.
|
||||
|
|
|
|||
|
|
@ -87,6 +87,10 @@ export interface StageProjection {
|
|||
* Ordered per-branch results produced by a parallel stage.
|
||||
*/
|
||||
'parallel_results'?: Array<ParallelBranchResult> | null;
|
||||
/**
|
||||
* Durable identity of this branch within a parallel execution, formatted as \"{parallel_group_id}:{index}\".
|
||||
*/
|
||||
'parallel_branch_id'?: string | null;
|
||||
'output'?: string | null;
|
||||
'output_bytes'?: number | null;
|
||||
'live_streaming'?: boolean | null;
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue