feat: add executor stream capture and run summaries

This commit is contained in:
eust-w 2026-03-27 02:16:17 +08:00
parent 22efdbcf3b
commit 526317681d
21 changed files with 461 additions and 36 deletions

View File

@ -66,7 +66,7 @@ The local validation path currently used for embodied data testing is:
You can register that directory from the Assets page or via `POST /api/assets/register`. You can register that directory from the Assets page or via `POST /api/assets/register`.
The workflow editor currently requires selecting at least one registered asset before a run can be created. The workflow editor currently requires selecting at least one registered asset before a run can be created.
The Runs workspace now shows project-scoped run history, and each run detail view surfaces persisted task summaries, log lines, result previews, and artifact links into Explore. The Runs workspace now shows project-scoped run history, run-level aggregated summaries, and run detail views with persisted task summaries, stdout/stderr sections, result previews, and artifact links into Explore.
## Repository Structure ## Repository Structure

View File

@ -12,6 +12,8 @@ export const runTaskSchemaDefinition = {
upstreamNodeIds: { type: "array", required: true, default: [] }, upstreamNodeIds: { type: "array", required: true, default: [] },
outputArtifactIds: { type: "array", required: true, default: [] }, outputArtifactIds: { type: "array", required: true, default: [] },
logLines: { type: "array", required: true, default: [] }, logLines: { type: "array", required: true, default: [] },
stdoutLines: { type: "array", required: false, default: [] },
stderrLines: { type: "array", required: false, default: [] },
errorMessage: { type: "string", required: false, default: null }, errorMessage: { type: "string", required: false, default: null },
summary: { type: "object", required: false, default: null }, summary: { type: "object", required: false, default: null },
lastResultPreview: { type: "object", required: false, default: null }, lastResultPreview: { type: "object", required: false, default: null },

View File

@ -5,7 +5,11 @@ export const workflowRunSchemaDefinition = {
workflowVersionId: { type: "string", required: true }, workflowVersionId: { type: "string", required: true },
status: { type: "string", required: true }, status: { type: "string", required: true },
triggeredBy: { type: "string", required: true }, triggeredBy: { type: "string", required: true },
assetIds: { type: "array", required: true, default: [] },
summary: { type: "object", required: false, default: null },
startedAt: { type: "date", required: false, default: null }, startedAt: { type: "date", required: false, default: null },
finishedAt: { type: "date", required: false, default: null }, finishedAt: { type: "date", required: false, default: null },
durationMs: { type: "number", required: false, default: null },
createdAt: { type: "date", required: true }, createdAt: { type: "date", required: true },
updatedAt: { type: "date", required: true },
} as const; } as const;

View File

@ -7,7 +7,9 @@ import { DELIVERY_NODE_DEFINITIONS } from "../modules/plugins/builtin/delivery-n
import { probeLocalSourcePath } from "./local-source-probe.ts"; import { probeLocalSourcePath } from "./local-source-probe.ts";
import type { import type {
ExecutorType, ExecutorType,
RunExecutionSummary,
TaskExecutionSummary, TaskExecutionSummary,
TaskStatusCounts,
} from "../../../worker/src/contracts/execution-context.ts"; } from "../../../worker/src/contracts/execution-context.ts";
type Timestamped = { type Timestamped = {
@ -97,9 +99,13 @@ type WorkflowRunDocument = Timestamped & {
workflowVersionId: string; workflowVersionId: string;
workspaceId: string; workspaceId: string;
projectId: string; projectId: string;
status: "queued"; status: "queued" | "running" | "success" | "failed";
triggeredBy: string; triggeredBy: string;
assetIds: string[]; assetIds: string[];
startedAt?: string;
finishedAt?: string;
durationMs?: number;
summary?: RunExecutionSummary;
}; };
type RunTaskDocument = Timestamped & { type RunTaskDocument = Timestamped & {
@ -109,7 +115,7 @@ type RunTaskDocument = Timestamped & {
nodeId: string; nodeId: string;
nodeType: string; nodeType: string;
executorType: ExecutorType; executorType: ExecutorType;
status: "queued" | "pending"; status: "queued" | "pending" | "running" | "success" | "failed";
attempt: number; attempt: number;
assetIds: string[]; assetIds: string[];
upstreamNodeIds: string[]; upstreamNodeIds: string[];
@ -118,6 +124,8 @@ type RunTaskDocument = Timestamped & {
finishedAt?: string; finishedAt?: string;
durationMs?: number; durationMs?: number;
logLines?: string[]; logLines?: string[];
stdoutLines?: string[];
stderrLines?: string[];
summary?: TaskExecutionSummary; summary?: TaskExecutionSummary;
lastResultPreview?: Record<string, unknown>; lastResultPreview?: Record<string, unknown>;
errorMessage?: string; errorMessage?: string;
@ -148,6 +156,35 @@ function mapDoc<T extends Document>(document: WithId<T>): T {
return document as unknown as T; return document as unknown as T;
} }
function buildTaskStatusCounts(tasks: RunTaskDocument[]): TaskStatusCounts {
const counts: TaskStatusCounts = {
pending: 0,
queued: 0,
running: 0,
success: 0,
failed: 0,
};
for (const task of tasks) {
counts[task.status] += 1;
}
return counts;
}
function buildRunExecutionSummary(tasks: RunTaskDocument[]): RunExecutionSummary {
const taskCounts = buildTaskStatusCounts(tasks);
return {
totalTaskCount: tasks.length,
completedTaskCount: taskCounts.success + taskCounts.failed,
artifactCount: tasks.reduce((total, task) => total + task.outputArtifactIds.length, 0),
stdoutLineCount: tasks.reduce((total, task) => total + (task.stdoutLines?.length ?? 0), 0),
stderrLineCount: tasks.reduce((total, task) => total + (task.stderrLines?.length ?? 0), 0),
failedTaskIds: tasks.filter((task) => task.status === "failed").map((task) => task._id),
taskCounts,
};
}
export class MongoAppStore { export class MongoAppStore {
constructor(private readonly db: Db) {} constructor(private readonly db: Db) {}
@ -484,7 +521,6 @@ export class MongoAppStore {
createdAt: nowIso(), createdAt: nowIso(),
updatedAt: nowIso(), updatedAt: nowIso(),
}; };
await this.db.collection("workflow_runs").insertOne(run);
const targetNodes = new Set(version.logicGraph.edges.map((edge) => edge.to)); const targetNodes = new Set(version.logicGraph.edges.map((edge) => edge.to));
const tasks = version.logicGraph.nodes.map<RunTaskDocument>((node) => ({ const tasks = version.logicGraph.nodes.map<RunTaskDocument>((node) => ({
@ -506,6 +542,9 @@ export class MongoAppStore {
updatedAt: nowIso(), updatedAt: nowIso(),
})); }));
run.summary = buildRunExecutionSummary(tasks);
await this.db.collection("workflow_runs").insertOne(run);
if (tasks.length > 0) { if (tasks.length > 0) {
await this.db.collection("run_tasks").insertMany(tasks); await this.db.collection("run_tasks").insertMany(tasks);
} }

View File

@ -537,11 +537,15 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs",
finishedAt: "2026-03-27T10:00:02.500Z", finishedAt: "2026-03-27T10:00:02.500Z",
durationMs: 2500, durationMs: 2500,
logLines: ["Task claimed by worker", "Executor completed successfully"], logLines: ["Task claimed by worker", "Executor completed successfully"],
stdoutLines: ["python executor processed source-asset"],
stderrLines: [],
summary: { summary: {
outcome: "success", outcome: "success",
executorType: "python", executorType: "python",
assetCount: 1, assetCount: 1,
artifactIds: ["artifact-1"], artifactIds: ["artifact-1"],
stdoutLineCount: 1,
stderrLineCount: 0,
}, },
lastResultPreview: { lastResultPreview: {
taskId: task?._id, taskId: task?._id,
@ -550,6 +554,49 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs",
}, },
}, },
); );
await db.collection("workflow_runs").updateOne(
{ _id: run._id },
{
$set: {
status: "success",
startedAt: "2026-03-27T10:00:00.000Z",
finishedAt: "2026-03-27T10:00:02.500Z",
durationMs: 2500,
summary: {
totalTaskCount: 1,
completedTaskCount: 1,
artifactCount: 1,
stdoutLineCount: 1,
stderrLineCount: 0,
failedTaskIds: [],
taskCounts: {
pending: 0,
queued: 0,
running: 0,
success: 1,
failed: 0,
},
},
},
},
);
const runDetail = await readJson<{
_id: string;
status: string;
startedAt?: string;
finishedAt?: string;
durationMs?: number;
summary?: {
totalTaskCount?: number;
artifactCount?: number;
stdoutLineCount?: number;
stderrLineCount?: number;
taskCounts?: {
success?: number;
};
};
}>(await fetch(`${server.baseUrl}/api/runs/${run._id}`));
const tasks = await readJson< const tasks = await readJson<
Array<{ Array<{
@ -559,11 +606,15 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs",
finishedAt?: string; finishedAt?: string;
durationMs?: number; durationMs?: number;
logLines?: string[]; logLines?: string[];
stdoutLines?: string[];
stderrLines?: string[];
summary?: { summary?: {
outcome?: string; outcome?: string;
executorType?: string; executorType?: string;
assetCount?: number; assetCount?: number;
artifactIds?: string[]; artifactIds?: string[];
stdoutLineCount?: number;
stderrLineCount?: number;
}; };
lastResultPreview?: { lastResultPreview?: {
taskId?: string; taskId?: string;
@ -572,6 +623,13 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs",
}> }>
>(await fetch(`${server.baseUrl}/api/runs/${run._id}/tasks`)); >(await fetch(`${server.baseUrl}/api/runs/${run._id}/tasks`));
assert.equal(runDetail.status, "success");
assert.equal(runDetail.durationMs, 2500);
assert.equal(runDetail.summary?.totalTaskCount, 1);
assert.equal(runDetail.summary?.artifactCount, 1);
assert.equal(runDetail.summary?.stdoutLineCount, 1);
assert.equal(runDetail.summary?.stderrLineCount, 0);
assert.equal(runDetail.summary?.taskCounts?.success, 1);
assert.equal(tasks.length, 1); assert.equal(tasks.length, 1);
assert.equal(tasks[0]?._id, task?._id); assert.equal(tasks[0]?._id, task?._id);
assert.equal(tasks[0]?.status, "success"); assert.equal(tasks[0]?.status, "success");
@ -580,10 +638,14 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs",
"Task claimed by worker", "Task claimed by worker",
"Executor completed successfully", "Executor completed successfully",
]); ]);
assert.deepEqual(tasks[0]?.stdoutLines, ["python executor processed source-asset"]);
assert.deepEqual(tasks[0]?.stderrLines, []);
assert.equal(tasks[0]?.summary?.outcome, "success"); assert.equal(tasks[0]?.summary?.outcome, "success");
assert.equal(tasks[0]?.summary?.executorType, "python"); assert.equal(tasks[0]?.summary?.executorType, "python");
assert.equal(tasks[0]?.summary?.assetCount, 1); assert.equal(tasks[0]?.summary?.assetCount, 1);
assert.deepEqual(tasks[0]?.summary?.artifactIds, ["artifact-1"]); assert.deepEqual(tasks[0]?.summary?.artifactIds, ["artifact-1"]);
assert.equal(tasks[0]?.summary?.stdoutLineCount, 1);
assert.equal(tasks[0]?.summary?.stderrLineCount, 0);
assert.deepEqual(tasks[0]?.lastResultPreview, { assert.deepEqual(tasks[0]?.lastResultPreview, {
taskId: task?._id, taskId: task?._id,
executor: "python", executor: "python",

View File

@ -8,6 +8,8 @@ export type RunTaskView = {
durationMs?: number; durationMs?: number;
summaryLabel?: string; summaryLabel?: string;
errorMessage?: string; errorMessage?: string;
stdoutLines?: string[];
stderrLines?: string[];
logLines: string[]; logLines: string[];
}; };

View File

@ -26,6 +26,11 @@ export function renderTaskLogPanel(
.join("")}</ul>` .join("")}</ul>`
: "" : ""
} }
<h3>Stdout</h3>
<ul>${(task.stdoutLines ?? []).map((line) => `<li>${line}</li>`).join("") || "<li>No stdout</li>"}</ul>
<h3>Stderr</h3>
<ul>${(task.stderrLines ?? []).map((line) => `<li>${line}</li>`).join("") || "<li>No stderr</li>"}</ul>
<h3>Execution Log</h3>
<ul>${lines || "<li>No logs</li>"}</ul> <ul>${lines || "<li>No logs</li>"}</ul>
</aside> </aside>
`; `;

View File

@ -13,6 +13,8 @@ export type RunDetailPageInput = {
workflowName: string; workflowName: string;
status: string; status: string;
assetIds?: string[]; assetIds?: string[];
durationMs?: number;
summaryLabel?: string;
}; };
tasks: RunTaskView[]; tasks: RunTaskView[];
selectedTaskId?: string; selectedTaskId?: string;
@ -30,6 +32,8 @@ export function renderRunDetailPage(input: RunDetailPageInput): string {
<p>Run ${input.run.id}</p> <p>Run ${input.run.id}</p>
<p>Status: ${input.run.status}</p> <p>Status: ${input.run.status}</p>
<p>Input assets: ${(input.run.assetIds ?? []).join(", ") || "none"}</p> <p>Input assets: ${(input.run.assetIds ?? []).join(", ") || "none"}</p>
<p>Run duration: ${typeof input.run.durationMs === "number" ? `${input.run.durationMs} ms` : "n/a"}</p>
${input.run.summaryLabel ? `<p>Run summary: ${input.run.summaryLabel}</p>` : ""}
</header> </header>
${renderRunGraphView(input.tasks)} ${renderRunGraphView(input.tasks)}
${renderTaskLogPanel(input.tasks, input.selectedTaskId)} ${renderTaskLogPanel(input.tasks, input.selectedTaskId)}

View File

@ -37,6 +37,8 @@ test("run detail view shows node status badges from run data", () => {
workflowName: "Delivery Normalize", workflowName: "Delivery Normalize",
status: "running", status: "running",
assetIds: ["asset-1"], assetIds: ["asset-1"],
durationMs: 2450,
summaryLabel: "2 tasks complete, 1 running, 1 stdout line",
}, },
tasks: [ tasks: [
{ {
@ -48,6 +50,8 @@ test("run detail view shows node status badges from run data", () => {
artifactIds: ["artifact-1"], artifactIds: ["artifact-1"],
durationMs: 1200, durationMs: 1200,
summaryLabel: "Processed 1 asset", summaryLabel: "Processed 1 asset",
stdoutLines: ["asset ready"],
stderrLines: [],
logLines: ["Asset loaded"], logLines: ["Asset loaded"],
}, },
{ {
@ -59,6 +63,8 @@ test("run detail view shows node status badges from run data", () => {
artifactIds: ["artifact-2"], artifactIds: ["artifact-2"],
durationMs: 2450, durationMs: 2450,
summaryLabel: "Validated delivery package structure", summaryLabel: "Validated delivery package structure",
stdoutLines: ["Checking metadata"],
stderrLines: ["Minor warning"],
logLines: ["Checking metadata"], logLines: ["Checking metadata"],
}, },
], ],
@ -71,8 +77,12 @@ test("run detail view shows node status badges from run data", () => {
assert.match(html, /running/); assert.match(html, /running/);
assert.match(html, /Checking metadata/); assert.match(html, /Checking metadata/);
assert.match(html, /Input assets: asset-1/); assert.match(html, /Input assets: asset-1/);
assert.match(html, /Run duration: 2450 ms/);
assert.match(html, /2 tasks complete, 1 running, 1 stdout line/);
assert.match(html, /Duration: 2450 ms/); assert.match(html, /Duration: 2450 ms/);
assert.match(html, /Validated delivery package structure/); assert.match(html, /Validated delivery package structure/);
assert.match(html, /Stdout/);
assert.match(html, /Minor warning/);
assert.match(html, /\/explore\/artifact-2/); assert.match(html, /\/explore\/artifact-2/);
}); });

View File

@ -34,6 +34,16 @@ function formatTaskSummary(task: any) {
return `${outcome} via ${executor}; assets ${assetCount}; artifacts ${artifactCount}`; return `${outcome} via ${executor}; assets ${assetCount}; artifacts ${artifactCount}`;
} }
function formatRunSummary(run: any) {
const totalTaskCount = run?.summary?.totalTaskCount ?? 0;
const successCount = run?.summary?.taskCounts?.success ?? 0;
const failedCount = run?.summary?.taskCounts?.failed ?? 0;
const runningCount = run?.summary?.taskCounts?.running ?? 0;
const stdoutLineCount = run?.summary?.stdoutLineCount ?? 0;
const stderrLineCount = run?.summary?.stderrLineCount ?? 0;
return `${successCount} success, ${failedCount} failed, ${runningCount} running, ${stdoutLineCount} stdout lines, ${stderrLineCount} stderr lines, ${totalTaskCount} total tasks`;
}
function usePathname() { function usePathname() {
const [pathname, setPathname] = useState( const [pathname, setPathname] = useState(
typeof window === "undefined" ? "/assets" : window.location.pathname || "/assets", typeof window === "undefined" ? "/assets" : window.location.pathname || "/assets",
@ -709,6 +719,10 @@ function RunDetailPage(props: {
? run.assetIds.map((assetId: string) => assetId).join(", ") ? run.assetIds.map((assetId: string) => assetId).join(", ")
: "none"} : "none"}
</p> </p>
<p>Started at: {run.startedAt ?? "n/a"}</p>
<p>Finished at: {run.finishedAt ?? "n/a"}</p>
<p>Run duration: {typeof run.durationMs === "number" ? `${run.durationMs} ms` : "n/a"}</p>
<p>Run summary: {formatRunSummary(run)}</p>
</section> </section>
<section className="two-column"> <section className="two-column">
<div className="panel"> <div className="panel">
@ -765,6 +779,30 @@ function RunDetailPage(props: {
<p className="empty-state">No task artifacts yet.</p> <p className="empty-state">No task artifacts yet.</p>
)} )}
<div className="page-stack"> <div className="page-stack">
<div>
<strong>Stdout</strong>
{(selectedTask.stdoutLines ?? []).length > 0 ? (
<ul>
{(selectedTask.stdoutLines ?? []).map((line: string, index: number) => (
<li key={`${selectedTask._id}-stdout-${index}`}>{line}</li>
))}
</ul>
) : (
<p className="empty-state">No stdout lines.</p>
)}
</div>
<div>
<strong>Stderr</strong>
{(selectedTask.stderrLines ?? []).length > 0 ? (
<ul>
{(selectedTask.stderrLines ?? []).map((line: string, index: number) => (
<li key={`${selectedTask._id}-stderr-${index}`}>{line}</li>
))}
</ul>
) : (
<p className="empty-state">No stderr lines.</p>
)}
</div>
<div> <div>
<strong>Execution Log</strong> <strong>Execution Log</strong>
{(selectedTask.logLines ?? []).length > 0 ? ( {(selectedTask.logLines ?? []).length > 0 ? (

View File

@ -1,14 +1,39 @@
export type ExecutorType = "python" | "docker" | "http"; export type ExecutorType = "python" | "docker" | "http";
export type TaskStatus = "pending" | "queued" | "running" | "success" | "failed"; export type TaskStatus = "pending" | "queued" | "running" | "success" | "failed";
export type TaskStatusCounts = {
pending: number;
queued: number;
running: number;
success: number;
failed: number;
};
export type TaskExecutionSummary = { export type TaskExecutionSummary = {
outcome: "success" | "failed"; outcome: "success" | "failed";
executorType: ExecutorType; executorType: ExecutorType;
assetCount: number; assetCount: number;
artifactIds: string[]; artifactIds: string[];
stdoutLineCount: number;
stderrLineCount: number;
errorMessage?: string; errorMessage?: string;
}; };
export type RunExecutionSummary = {
totalTaskCount: number;
completedTaskCount: number;
artifactCount: number;
stdoutLineCount: number;
stderrLineCount: number;
failedTaskIds: string[];
taskCounts: TaskStatusCounts;
};
export type ExecutorExecutionResult = {
result: unknown;
stdoutLines?: string[];
stderrLines?: string[];
};
export type TaskRecord = { export type TaskRecord = {
id: string; id: string;
workflowRunId?: string; workflowRunId?: string;
@ -26,6 +51,8 @@ export type TaskRecord = {
finishedAt?: string; finishedAt?: string;
durationMs?: number; durationMs?: number;
logLines?: string[]; logLines?: string[];
stdoutLines?: string[];
stderrLines?: string[];
summary?: TaskExecutionSummary; summary?: TaskExecutionSummary;
lastResultPreview?: Record<string, unknown>; lastResultPreview?: Record<string, unknown>;
}; };

View File

@ -1,10 +1,18 @@
import type { ExecutionContext, TaskRecord } from "../contracts/execution-context.ts"; import type {
ExecutionContext,
ExecutorExecutionResult,
TaskRecord,
} from "../contracts/execution-context.ts";
export class DockerExecutor { export class DockerExecutor {
executionCount = 0; executionCount = 0;
async execute(task: TaskRecord, _context: ExecutionContext) { async execute(task: TaskRecord, _context: ExecutionContext): Promise<ExecutorExecutionResult> {
this.executionCount += 1; this.executionCount += 1;
return { taskId: task.id, executor: "docker" as const }; return {
result: { taskId: task.id, executor: "docker" as const },
stdoutLines: [`docker executor processed ${task.nodeId}`],
stderrLines: [],
};
} }
} }

View File

@ -1,10 +1,18 @@
import type { ExecutionContext, TaskRecord } from "../contracts/execution-context.ts"; import type {
ExecutionContext,
ExecutorExecutionResult,
TaskRecord,
} from "../contracts/execution-context.ts";
export class HttpExecutor { export class HttpExecutor {
executionCount = 0; executionCount = 0;
async execute(task: TaskRecord, _context: ExecutionContext) { async execute(task: TaskRecord, _context: ExecutionContext): Promise<ExecutorExecutionResult> {
this.executionCount += 1; this.executionCount += 1;
return { taskId: task.id, executor: "http" as const }; return {
result: { taskId: task.id, executor: "http" as const },
stdoutLines: [`http executor processed ${task.nodeId}`],
stderrLines: [],
};
} }
} }

View File

@ -1,10 +1,18 @@
import type { ExecutionContext, TaskRecord } from "../contracts/execution-context.ts"; import type {
ExecutionContext,
ExecutorExecutionResult,
TaskRecord,
} from "../contracts/execution-context.ts";
export class PythonExecutor { export class PythonExecutor {
executionCount = 0; executionCount = 0;
async execute(task: TaskRecord, _context: ExecutionContext) { async execute(task: TaskRecord, _context: ExecutionContext): Promise<ExecutorExecutionResult> {
this.executionCount += 1; this.executionCount += 1;
return { taskId: task.id, executor: "python" as const }; return {
result: { taskId: task.id, executor: "python" as const },
stdoutLines: [`python executor processed ${task.nodeId}`],
stderrLines: [],
};
} }
} }

View File

@ -4,8 +4,10 @@ import type { Db } from "mongodb";
import type { import type {
ExecutorType, ExecutorType,
RunExecutionSummary,
TaskExecutionSummary, TaskExecutionSummary,
TaskRecord, TaskRecord,
TaskStatusCounts,
TaskStatus, TaskStatus,
} from "../contracts/execution-context.ts"; } from "../contracts/execution-context.ts";
@ -16,6 +18,10 @@ type WorkflowRunDocument = {
status: "queued" | "running" | "success" | "failed"; status: "queued" | "running" | "success" | "failed";
triggeredBy: string; triggeredBy: string;
assetIds: string[]; assetIds: string[];
startedAt?: string;
finishedAt?: string;
durationMs?: number;
summary?: RunExecutionSummary;
createdAt: string; createdAt: string;
updatedAt: string; updatedAt: string;
}; };
@ -47,6 +53,8 @@ type RunTaskDocument = {
finishedAt?: string; finishedAt?: string;
durationMs?: number; durationMs?: number;
logLines?: string[]; logLines?: string[];
stdoutLines?: string[];
stderrLines?: string[];
summary?: TaskExecutionSummary; summary?: TaskExecutionSummary;
lastResultPreview?: Record<string, unknown>; lastResultPreview?: Record<string, unknown>;
createdAt: string; createdAt: string;
@ -75,11 +83,58 @@ function toTaskRecord(task: RunTaskDocument): TaskRecord {
finishedAt: task.finishedAt, finishedAt: task.finishedAt,
durationMs: task.durationMs, durationMs: task.durationMs,
logLines: task.logLines ?? [], logLines: task.logLines ?? [],
stdoutLines: task.stdoutLines ?? [],
stderrLines: task.stderrLines ?? [],
summary: task.summary, summary: task.summary,
lastResultPreview: task.lastResultPreview, lastResultPreview: task.lastResultPreview,
}; };
} }
function buildTaskStatusCounts(tasks: TaskRecord[]): TaskStatusCounts {
const counts: TaskStatusCounts = {
pending: 0,
queued: 0,
running: 0,
success: 0,
failed: 0,
};
for (const task of tasks) {
counts[task.status] += 1;
}
return counts;
}
function buildRunExecutionSummary(tasks: TaskRecord[]): RunExecutionSummary {
const taskCounts = buildTaskStatusCounts(tasks);
return {
totalTaskCount: tasks.length,
completedTaskCount: taskCounts.success + taskCounts.failed,
artifactCount: tasks.reduce((total, task) => total + (task.outputArtifactIds?.length ?? 0), 0),
stdoutLineCount: tasks.reduce((total, task) => total + (task.stdoutLines?.length ?? 0), 0),
stderrLineCount: tasks.reduce((total, task) => total + (task.stderrLines?.length ?? 0), 0),
failedTaskIds: tasks.filter((task) => task.status === "failed").map((task) => task.id),
taskCounts,
};
}
function minIso(values: Array<string | undefined>) {
const filtered = values.filter((value): value is string => Boolean(value));
if (filtered.length === 0) {
return undefined;
}
return filtered.reduce((current, value) => (value < current ? value : current));
}
function maxIso(values: Array<string | undefined>) {
const filtered = values.filter((value): value is string => Boolean(value));
if (filtered.length === 0) {
return undefined;
}
return filtered.reduce((current, value) => (value > current ? value : current));
}
export class MongoWorkerStore { export class MongoWorkerStore {
private readonly db: Db; private readonly db: Db;
@ -173,7 +228,9 @@ export class MongoWorkerStore {
finishedAt: string; finishedAt: string;
durationMs: number; durationMs: number;
summary: TaskExecutionSummary; summary: TaskExecutionSummary;
logLine: string; stdoutLines: string[];
stderrLines: string[];
logLines: string[];
lastResultPreview?: Record<string, unknown>; lastResultPreview?: Record<string, unknown>;
}, },
) { ) {
@ -184,13 +241,15 @@ export class MongoWorkerStore {
status: "success", status: "success",
finishedAt: input.finishedAt, finishedAt: input.finishedAt,
durationMs: input.durationMs, durationMs: input.durationMs,
stdoutLines: input.stdoutLines,
stderrLines: input.stderrLines,
summary: input.summary, summary: input.summary,
lastResultPreview: input.lastResultPreview, lastResultPreview: input.lastResultPreview,
updatedAt: input.finishedAt, updatedAt: input.finishedAt,
}, },
$push: { $push: {
logLines: { logLines: {
$each: [input.logLine], $each: input.logLines,
}, },
}, },
}, },
@ -204,7 +263,9 @@ export class MongoWorkerStore {
finishedAt: string; finishedAt: string;
durationMs: number; durationMs: number;
summary: TaskExecutionSummary; summary: TaskExecutionSummary;
logLine: string; stdoutLines: string[];
stderrLines: string[];
logLines: string[];
}, },
) { ) {
await this.db.collection<RunTaskDocument>("run_tasks").updateOne( await this.db.collection<RunTaskDocument>("run_tasks").updateOne(
@ -215,12 +276,14 @@ export class MongoWorkerStore {
errorMessage, errorMessage,
finishedAt: input.finishedAt, finishedAt: input.finishedAt,
durationMs: input.durationMs, durationMs: input.durationMs,
stdoutLines: input.stdoutLines,
stderrLines: input.stderrLines,
summary: input.summary, summary: input.summary,
updatedAt: input.finishedAt, updatedAt: input.finishedAt,
}, },
$push: { $push: {
logLines: { logLines: {
$each: [input.logLine], $each: input.logLines,
}, },
}, },
}, },
@ -273,14 +336,28 @@ export class MongoWorkerStore {
status = "running"; status = "running";
} }
const startedAt = minIso(tasks.map((task) => task.startedAt));
const finishedAt =
status === "success" || status === "failed"
? maxIso(tasks.map((task) => task.finishedAt))
: undefined;
const summary = buildRunExecutionSummary(tasks);
const updateSet: Partial<WorkflowRunDocument> = {
status,
summary,
updatedAt: nowIso(),
};
if (startedAt) {
updateSet.startedAt = startedAt;
}
if (finishedAt) {
updateSet.finishedAt = finishedAt;
updateSet.durationMs = Math.max(Date.parse(finishedAt) - Date.parse(startedAt ?? finishedAt), 0);
}
await this.db.collection<WorkflowRunDocument>("workflow_runs").updateOne( await this.db.collection<WorkflowRunDocument>("workflow_runs").updateOne(
{ _id: runId }, { _id: runId },
{ { $set: updateSet },
$set: {
status,
updatedAt: nowIso(),
},
},
); );
} }
} }

View File

@ -3,6 +3,7 @@ import { HttpExecutor } from "../executors/http-executor.ts";
import { PythonExecutor } from "../executors/python-executor.ts"; import { PythonExecutor } from "../executors/python-executor.ts";
import type { import type {
ExecutionContext, ExecutionContext,
ExecutorExecutionResult,
ExecutorType, ExecutorType,
TaskExecutionSummary, TaskExecutionSummary,
TaskRecord, TaskRecord,
@ -47,13 +48,15 @@ export class WorkerRuntime {
}; };
try { try {
const result = await this.executors[task.executorType as ExecutorType].execute(task, context); const execution = this.normalizeExecutionResult(
await this.executors[task.executorType as ExecutorType].execute(task, context),
);
const artifact = await this.store.createTaskArtifact(task, { const artifact = await this.store.createTaskArtifact(task, {
nodeId: task.nodeId, nodeId: task.nodeId,
nodeType: task.nodeType, nodeType: task.nodeType,
executorType: task.executorType, executorType: task.executorType,
assetIds: task.assetIds, assetIds: task.assetIds,
result, result: execution.result,
}); });
const finishedAt = new Date().toISOString(); const finishedAt = new Date().toISOString();
const summary: TaskExecutionSummary = { const summary: TaskExecutionSummary = {
@ -61,13 +64,21 @@ export class WorkerRuntime {
executorType: task.executorType, executorType: task.executorType,
assetCount: task.assetIds?.length ?? 0, assetCount: task.assetIds?.length ?? 0,
artifactIds: [artifact._id], artifactIds: [artifact._id],
stdoutLineCount: execution.stdoutLines.length,
stderrLineCount: execution.stderrLines.length,
}; };
await this.store.markTaskSuccess(task.id, { await this.store.markTaskSuccess(task.id, {
finishedAt, finishedAt,
durationMs: this.computeDurationMs(startedAt, finishedAt), durationMs: this.computeDurationMs(startedAt, finishedAt),
summary, summary,
logLine: "Executor completed successfully", stdoutLines: execution.stdoutLines,
lastResultPreview: this.createResultPreview(result), stderrLines: execution.stderrLines,
logLines: this.createTaskLogLines(
execution.stdoutLines,
execution.stderrLines,
"Executor completed successfully",
),
lastResultPreview: this.createResultPreview(execution.result),
}); });
if (task.workflowRunId) { if (task.workflowRunId) {
await this.store.queueReadyDependents(task.workflowRunId); await this.store.queueReadyDependents(task.workflowRunId);
@ -75,20 +86,28 @@ export class WorkerRuntime {
} }
return this.store.getRunTask(task.id) ?? { ...task, status: "success" }; return this.store.getRunTask(task.id) ?? { ...task, status: "success" };
} catch (error) { } catch (error) {
const message = error instanceof Error ? error.message : "worker execution failed"; const executionError = this.normalizeExecutionError(error);
const finishedAt = new Date().toISOString(); const finishedAt = new Date().toISOString();
const summary: TaskExecutionSummary = { const summary: TaskExecutionSummary = {
outcome: "failed", outcome: "failed",
executorType: task.executorType, executorType: task.executorType,
assetCount: task.assetIds?.length ?? 0, assetCount: task.assetIds?.length ?? 0,
artifactIds: [], artifactIds: [],
errorMessage: message, stdoutLineCount: executionError.stdoutLines.length,
stderrLineCount: executionError.stderrLines.length,
errorMessage: executionError.message,
}; };
await this.store.markTaskFailed(task.id, message, { await this.store.markTaskFailed(task.id, executionError.message, {
finishedAt, finishedAt,
durationMs: this.computeDurationMs(startedAt, finishedAt), durationMs: this.computeDurationMs(startedAt, finishedAt),
summary, summary,
logLine: `Execution failed: ${message}`, stdoutLines: executionError.stdoutLines,
stderrLines: executionError.stderrLines,
logLines: this.createTaskLogLines(
executionError.stdoutLines,
executionError.stderrLines,
`Execution failed: ${executionError.message}`,
),
}); });
if (task.workflowRunId) { if (task.workflowRunId) {
await this.store.refreshRunStatus(task.workflowRunId); await this.store.refreshRunStatus(task.workflowRunId);
@ -111,4 +130,47 @@ export class WorkerRuntime {
} }
return undefined; return undefined;
} }
private normalizeExecutionResult(result: unknown): ExecutorExecutionResult {
if (result && typeof result === "object" && "result" in result) {
const execution = result as ExecutorExecutionResult;
return {
result: execution.result,
stdoutLines: execution.stdoutLines ?? [],
stderrLines: execution.stderrLines ?? [],
};
}
return {
result,
stdoutLines: [],
stderrLines: [],
};
}
private normalizeExecutionError(error: unknown) {
const message = error instanceof Error ? error.message : "worker execution failed";
const stdoutLines =
error && typeof error === "object" && "stdoutLines" in error && Array.isArray(error.stdoutLines)
? error.stdoutLines.filter((line): line is string => typeof line === "string")
: [];
const stderrLines =
error && typeof error === "object" && "stderrLines" in error && Array.isArray(error.stderrLines)
? error.stderrLines.filter((line): line is string => typeof line === "string")
: [];
return {
message,
stdoutLines,
stderrLines,
};
}
private createTaskLogLines(stdoutLines: string[], stderrLines: string[], terminalLine: string) {
return [
...stdoutLines.map((line) => `stdout: ${line}`),
...stderrLines.map((line) => `stderr: ${line}`),
terminalLine,
];
}
} }

View File

@ -189,14 +189,27 @@ test("worker marks the run successful after the final queued task completes", as
assert.ok(typeof task?.startedAt === "string"); assert.ok(typeof task?.startedAt === "string");
assert.ok(typeof task?.finishedAt === "string"); assert.ok(typeof task?.finishedAt === "string");
assert.ok(typeof task?.durationMs === "number"); assert.ok(typeof task?.durationMs === "number");
assert.deepEqual(task?.stdoutLines, ["python executor processed export-delivery-package"]);
assert.deepEqual(task?.stderrLines, []);
assert.equal(task?.summary?.outcome, "success"); assert.equal(task?.summary?.outcome, "success");
assert.equal(task?.summary?.executorType, "python"); assert.equal(task?.summary?.executorType, "python");
assert.equal(task?.summary?.stdoutLineCount, 1);
assert.equal(task?.summary?.stderrLineCount, 0);
assert.match(task?.logLines?.[0] ?? "", /claimed/i); assert.match(task?.logLines?.[0] ?? "", /claimed/i);
assert.match(task?.logLines?.[1] ?? "", /stdout:/i);
assert.match(task?.logLines?.at(-1) ?? "", /completed/i); assert.match(task?.logLines?.at(-1) ?? "", /completed/i);
assert.deepEqual(task?.lastResultPreview, { assert.deepEqual(task?.lastResultPreview, {
taskId: "task-export", taskId: "task-export",
executor: "python", executor: "python",
}); });
assert.ok(typeof run?.startedAt === "string");
assert.ok(typeof run?.finishedAt === "string");
assert.ok(typeof run?.durationMs === "number");
assert.equal(run?.summary?.totalTaskCount, 1);
assert.equal(run?.summary?.artifactCount, 1);
assert.equal(run?.summary?.stdoutLineCount, 1);
assert.equal(run?.summary?.stderrLineCount, 0);
assert.equal(run?.summary?.taskCounts?.success, 1);
}); });
test("worker passes bound asset ids into the execution context and task artifacts", async (t) => { test("worker passes bound asset ids into the execution context and task artifacts", async (t) => {
@ -209,8 +222,12 @@ test("worker passes bound asset ids into the execution context and task artifact
capturedTask = task; capturedTask = task;
capturedContext = context; capturedContext = context;
return { return {
result: {
taskId: task.id, taskId: task.id,
assetIds: context.assetIds, assetIds: context.assetIds,
},
stdoutLines: ["custom executor started"],
stderrLines: [],
}; };
}, },
}, },
@ -262,6 +279,8 @@ test("worker passes bound asset ids into the execution context and task artifact
assert.equal(storedTask?.summary?.outcome, "success"); assert.equal(storedTask?.summary?.outcome, "success");
assert.equal(storedTask?.summary?.assetCount, 1); assert.equal(storedTask?.summary?.assetCount, 1);
assert.deepEqual(storedTask?.summary?.artifactIds, [artifact?._id]); assert.deepEqual(storedTask?.summary?.artifactIds, [artifact?._id]);
assert.deepEqual(storedTask?.stdoutLines, ["custom executor started"]);
assert.deepEqual(storedTask?.stderrLines, []);
}); });
test("worker persists failure summaries and task log lines when execution throws", async (t) => { test("worker persists failure summaries and task log lines when execution throws", async (t) => {
@ -269,7 +288,10 @@ test("worker persists failure summaries and task log lines when execution throws
executors: { executors: {
python: { python: {
async execute() { async execute() {
throw new Error("intentional executor failure"); throw Object.assign(new Error("intentional executor failure"), {
stdoutLines: ["failure path entered"],
stderrLines: ["stack trace line"],
});
}, },
}, },
}, },
@ -315,10 +337,21 @@ test("worker persists failure summaries and task log lines when execution throws
assert.equal(task?.errorMessage, "intentional executor failure"); assert.equal(task?.errorMessage, "intentional executor failure");
assert.equal(task?.summary?.outcome, "failed"); assert.equal(task?.summary?.outcome, "failed");
assert.equal(task?.summary?.executorType, "python"); assert.equal(task?.summary?.executorType, "python");
assert.equal(task?.summary?.stdoutLineCount, 1);
assert.equal(task?.summary?.stderrLineCount, 1);
assert.equal(task?.summary?.artifactIds?.length ?? 0, 0); assert.equal(task?.summary?.artifactIds?.length ?? 0, 0);
assert.ok(typeof task?.startedAt === "string"); assert.ok(typeof task?.startedAt === "string");
assert.ok(typeof task?.finishedAt === "string"); assert.ok(typeof task?.finishedAt === "string");
assert.ok(typeof task?.durationMs === "number"); assert.ok(typeof task?.durationMs === "number");
assert.deepEqual(task?.stdoutLines, ["failure path entered"]);
assert.deepEqual(task?.stderrLines, ["stack trace line"]);
assert.match(task?.logLines?.[0] ?? "", /claimed/i); assert.match(task?.logLines?.[0] ?? "", /claimed/i);
assert.match(task?.logLines?.[1] ?? "", /stdout:/i);
assert.match(task?.logLines?.[2] ?? "", /stderr:/i);
assert.match(task?.logLines?.at(-1) ?? "", /intentional executor failure/i); assert.match(task?.logLines?.at(-1) ?? "", /intentional executor failure/i);
assert.equal(run?.summary?.totalTaskCount, 1);
assert.equal(run?.summary?.taskCounts?.failed, 1);
assert.equal(run?.summary?.stdoutLineCount, 1);
assert.equal(run?.summary?.stderrLineCount, 1);
assert.deepEqual(run?.summary?.failedTaskIds, ["task-failure"]);
}); });

View File

@ -288,11 +288,26 @@ The worker-backed runtime now persists task execution summaries directly on `run
- `startedAt` and `finishedAt` - `startedAt` and `finishedAt`
- `durationMs` - `durationMs`
- appended `logLines` - appended `logLines`
- captured `stdoutLines` and `stderrLines`
- structured `summary` with outcome, executor, asset count, artifact ids, and failure text when present - structured `summary` with outcome, executor, asset count, artifact ids, and failure text when present
- `lastResultPreview` for a lightweight selected-task preview in the Runs workspace - `lastResultPreview` for a lightweight selected-task preview in the Runs workspace
This makes the run detail view stable even when artifacts are large or delayed and keeps task-level observability queryable without reopening every artifact payload. This makes the run detail view stable even when artifacts are large or delayed and keeps task-level observability queryable without reopening every artifact payload.
The current runtime also aggregates execution state back onto `workflow_runs`. Each refresh computes:
- run-level `startedAt` and `finishedAt`
- run-level `durationMs`
- `summary.totalTaskCount`
- `summary.completedTaskCount`
- `summary.taskCounts`
- `summary.artifactCount`
- `summary.stdoutLineCount`
- `summary.stderrLineCount`
- `summary.failedTaskIds`
This allows the Runs workspace to render a stable top-level run summary without client-side recomputation across every task document.
The API and worker runtimes now both have direct integration coverage against a real Mongo runtime through `mongodb-memory-server`, in addition to the older in-memory contract tests. The API and worker runtimes now both have direct integration coverage against a real Mongo runtime through `mongodb-memory-server`, in addition to the older in-memory contract tests.
The first web authoring surface already follows the three-pane layout contract with: The first web authoring surface already follows the three-pane layout contract with:

View File

@ -187,10 +187,18 @@ V1 run detail should render the selected task as a stable operations panel, not
- task status and executor - task status and executor
- task duration and timestamps - task duration and timestamps
- a concise execution summary - a concise execution summary
- separated stdout and stderr sections
- appended log lines - appended log lines
- artifact links into Explore - artifact links into Explore
- a lightweight result preview for quick inspection - a lightweight result preview for quick inspection
The run header itself should also show an aggregated summary:
- run duration
- run-level task counts
- total stdout/stderr line counts
- failed task ids when present
## Screen 6: Explore Workspace ## Screen 6: Explore Workspace
Purpose: Purpose:

View File

@ -299,6 +299,7 @@ Core fields:
- `summary` - `summary`
- `startedAt` - `startedAt`
- `finishedAt` - `finishedAt`
- `durationMs`
- `createdAt` - `createdAt`
### run_tasks ### run_tasks
@ -325,6 +326,8 @@ Core fields:
- `cacheKey` - `cacheKey`
- `cacheHit` - `cacheHit`
- `logLines` - `logLines`
- `stdoutLines`
- `stderrLines`
- `errorMessage` - `errorMessage`
- `summary` - `summary`
- `lastResultPreview` - `lastResultPreview`
@ -342,8 +345,17 @@ The current executable worker path expects `run_tasks` to be self-sufficient eno
- upstream node dependencies - upstream node dependencies
- produced artifact ids - produced artifact ids
- per-task status and error message - per-task status and error message
- task log lines and result preview - task log lines, stdout/stderr streams, and result preview
- structured task summaries with executor, outcome, asset count, and artifact ids - structured task summaries with executor, outcome, asset count, artifact ids, and stdout/stderr counters
The current runtime also aggregates task execution back onto `workflow_runs`, so run documents now carry:
- task counts by status
- completed task count
- artifact count
- total stdout/stderr line counts
- failed task ids
- derived run duration
### artifacts ### artifacts

View File

@ -21,6 +21,7 @@
- `2026-03-26`: The current runtime pass binds workflow runs to registered project assets so run snapshots, run tasks, worker execution context, and the editor all agree on the concrete input asset being processed. - `2026-03-26`: The current runtime pass binds workflow runs to registered project assets so run snapshots, run tasks, worker execution context, and the editor all agree on the concrete input asset being processed.
- `2026-03-26`: The current UI/runtime pass turns Runs into a real project-scoped workspace with run history queries, active-run polling, and task artifact links into Explore. - `2026-03-26`: The current UI/runtime pass turns Runs into a real project-scoped workspace with run history queries, active-run polling, and task artifact links into Explore.
- `2026-03-27`: The current observability pass persists task execution summaries, timestamps, log lines, and result previews on Mongo-backed `run_tasks`, and surfaces those fields in the React run detail view. - `2026-03-27`: The current observability pass persists task execution summaries, timestamps, log lines, and result previews on Mongo-backed `run_tasks`, and surfaces those fields in the React run detail view.
- `2026-03-27`: The current follow-up observability pass adds persisted stdout/stderr fields on `run_tasks` plus aggregated run summaries, durations, and task counts on `workflow_runs`.
--- ---