From 526317681dbc2383effcc773d9e0ad1e0bd7bb3d Mon Sep 17 00:00:00 2001 From: eust-w Date: Fri, 27 Mar 2026 02:16:17 +0800 Subject: [PATCH] :sparkles: feat: add executor stream capture and run summaries --- README.md | 2 +- .../common/mongo/schemas/run-task.schema.ts | 2 + .../mongo/schemas/workflow-run.schema.ts | 4 + apps/api/src/runtime/mongo-store.ts | 45 ++++++++- .../api/test/runtime-http.integration.spec.ts | 62 ++++++++++++ .../runs/components/run-graph-view.tsx | 2 + .../runs/components/task-log-panel.tsx | 5 + .../web/src/features/runs/run-detail-page.tsx | 4 + .../workflows/workflow-editor-page.test.tsx | 10 ++ apps/web/src/runtime/app.tsx | 38 ++++++++ .../worker/src/contracts/execution-context.ts | 27 ++++++ apps/worker/src/executors/docker-executor.ts | 14 ++- apps/worker/src/executors/http-executor.ts | 14 ++- apps/worker/src/executors/python-executor.ts | 14 ++- apps/worker/src/runtime/mongo-worker-store.ts | 97 +++++++++++++++++-- apps/worker/src/runtime/worker-runtime.ts | 78 +++++++++++++-- apps/worker/test/mongo-worker-runtime.spec.ts | 39 +++++++- .../03-workflows/workflow-execution-model.md | 15 +++ ...nformation-architecture-and-key-screens.md | 8 ++ design/05-data/mongodb-data-model.md | 16 ++- ...26-03-26-emboflow-v1-foundation-and-mvp.md | 1 + 21 files changed, 461 insertions(+), 36 deletions(-) diff --git a/README.md b/README.md index e203430..98e0791 100644 --- a/README.md +++ b/README.md @@ -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`. 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 diff --git a/apps/api/src/common/mongo/schemas/run-task.schema.ts b/apps/api/src/common/mongo/schemas/run-task.schema.ts index 4762c60..f13823f 100644 --- a/apps/api/src/common/mongo/schemas/run-task.schema.ts +++ b/apps/api/src/common/mongo/schemas/run-task.schema.ts @@ -12,6 +12,8 @@ export const runTaskSchemaDefinition = { upstreamNodeIds: { type: "array", required: true, default: [] }, outputArtifactIds: { 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 }, summary: { type: "object", required: false, default: null }, lastResultPreview: { type: "object", required: false, default: null }, diff --git a/apps/api/src/common/mongo/schemas/workflow-run.schema.ts b/apps/api/src/common/mongo/schemas/workflow-run.schema.ts index daf4483..4666398 100644 --- a/apps/api/src/common/mongo/schemas/workflow-run.schema.ts +++ b/apps/api/src/common/mongo/schemas/workflow-run.schema.ts @@ -5,7 +5,11 @@ export const workflowRunSchemaDefinition = { workflowVersionId: { type: "string", required: true }, status: { 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 }, finishedAt: { type: "date", required: false, default: null }, + durationMs: { type: "number", required: false, default: null }, createdAt: { type: "date", required: true }, + updatedAt: { type: "date", required: true }, } as const; diff --git a/apps/api/src/runtime/mongo-store.ts b/apps/api/src/runtime/mongo-store.ts index 7486dfb..795fb4d 100644 --- a/apps/api/src/runtime/mongo-store.ts +++ b/apps/api/src/runtime/mongo-store.ts @@ -7,7 +7,9 @@ import { DELIVERY_NODE_DEFINITIONS } from "../modules/plugins/builtin/delivery-n import { probeLocalSourcePath } from "./local-source-probe.ts"; import type { ExecutorType, + RunExecutionSummary, TaskExecutionSummary, + TaskStatusCounts, } from "../../../worker/src/contracts/execution-context.ts"; type Timestamped = { @@ -97,9 +99,13 @@ type WorkflowRunDocument = Timestamped & { workflowVersionId: string; workspaceId: string; projectId: string; - status: "queued"; + status: "queued" | "running" | "success" | "failed"; triggeredBy: string; assetIds: string[]; + startedAt?: string; + finishedAt?: string; + durationMs?: number; + summary?: RunExecutionSummary; }; type RunTaskDocument = Timestamped & { @@ -109,7 +115,7 @@ type RunTaskDocument = Timestamped & { nodeId: string; nodeType: string; executorType: ExecutorType; - status: "queued" | "pending"; + status: "queued" | "pending" | "running" | "success" | "failed"; attempt: number; assetIds: string[]; upstreamNodeIds: string[]; @@ -118,6 +124,8 @@ type RunTaskDocument = Timestamped & { finishedAt?: string; durationMs?: number; logLines?: string[]; + stdoutLines?: string[]; + stderrLines?: string[]; summary?: TaskExecutionSummary; lastResultPreview?: Record; errorMessage?: string; @@ -148,6 +156,35 @@ function mapDoc(document: WithId): 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 { constructor(private readonly db: Db) {} @@ -484,7 +521,6 @@ export class MongoAppStore { createdAt: nowIso(), updatedAt: nowIso(), }; - await this.db.collection("workflow_runs").insertOne(run); const targetNodes = new Set(version.logicGraph.edges.map((edge) => edge.to)); const tasks = version.logicGraph.nodes.map((node) => ({ @@ -506,6 +542,9 @@ export class MongoAppStore { updatedAt: nowIso(), })); + run.summary = buildRunExecutionSummary(tasks); + await this.db.collection("workflow_runs").insertOne(run); + if (tasks.length > 0) { await this.db.collection("run_tasks").insertMany(tasks); } diff --git a/apps/api/test/runtime-http.integration.spec.ts b/apps/api/test/runtime-http.integration.spec.ts index b25cae0..8467623 100644 --- a/apps/api/test/runtime-http.integration.spec.ts +++ b/apps/api/test/runtime-http.integration.spec.ts @@ -537,11 +537,15 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs", finishedAt: "2026-03-27T10:00:02.500Z", durationMs: 2500, logLines: ["Task claimed by worker", "Executor completed successfully"], + stdoutLines: ["python executor processed source-asset"], + stderrLines: [], summary: { outcome: "success", executorType: "python", assetCount: 1, artifactIds: ["artifact-1"], + stdoutLineCount: 1, + stderrLineCount: 0, }, lastResultPreview: { 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< Array<{ @@ -559,11 +606,15 @@ test("mongo-backed runtime exposes persisted task execution summaries and logs", finishedAt?: string; durationMs?: number; logLines?: string[]; + stdoutLines?: string[]; + stderrLines?: string[]; summary?: { outcome?: string; executorType?: string; assetCount?: number; artifactIds?: string[]; + stdoutLineCount?: number; + stderrLineCount?: number; }; lastResultPreview?: { 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`)); + 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[0]?._id, task?._id); 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", "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?.executorType, "python"); assert.equal(tasks[0]?.summary?.assetCount, 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, { taskId: task?._id, executor: "python", diff --git a/apps/web/src/features/runs/components/run-graph-view.tsx b/apps/web/src/features/runs/components/run-graph-view.tsx index 3466e3a..532dd47 100644 --- a/apps/web/src/features/runs/components/run-graph-view.tsx +++ b/apps/web/src/features/runs/components/run-graph-view.tsx @@ -8,6 +8,8 @@ export type RunTaskView = { durationMs?: number; summaryLabel?: string; errorMessage?: string; + stdoutLines?: string[]; + stderrLines?: string[]; logLines: string[]; }; diff --git a/apps/web/src/features/runs/components/task-log-panel.tsx b/apps/web/src/features/runs/components/task-log-panel.tsx index 1dd802d..70a0587 100644 --- a/apps/web/src/features/runs/components/task-log-panel.tsx +++ b/apps/web/src/features/runs/components/task-log-panel.tsx @@ -26,6 +26,11 @@ export function renderTaskLogPanel( .join("")}` : "" } +

Stdout

+
    ${(task.stdoutLines ?? []).map((line) => `
  • ${line}
    • `).join("") || "
  • No stdout
    • "}
+

Stderr

+
    ${(task.stderrLines ?? []).map((line) => `
  • ${line}
    • `).join("") || "
  • No stderr
    • "}
+

Execution Log

    ${lines || "
  • No logs
    • "}
`; diff --git a/apps/web/src/features/runs/run-detail-page.tsx b/apps/web/src/features/runs/run-detail-page.tsx index 89c8e43..e75d441 100644 --- a/apps/web/src/features/runs/run-detail-page.tsx +++ b/apps/web/src/features/runs/run-detail-page.tsx @@ -13,6 +13,8 @@ export type RunDetailPageInput = { workflowName: string; status: string; assetIds?: string[]; + durationMs?: number; + summaryLabel?: string; }; tasks: RunTaskView[]; selectedTaskId?: string; @@ -30,6 +32,8 @@ export function renderRunDetailPage(input: RunDetailPageInput): string {

Run ${input.run.id}

Status: ${input.run.status}

Input assets: ${(input.run.assetIds ?? []).join(", ") || "none"}

+

Run duration: ${typeof input.run.durationMs === "number" ? `${input.run.durationMs} ms` : "n/a"}

+ ${input.run.summaryLabel ? `

Run summary: ${input.run.summaryLabel}

` : ""} ${renderRunGraphView(input.tasks)} ${renderTaskLogPanel(input.tasks, input.selectedTaskId)} diff --git a/apps/web/src/features/workflows/workflow-editor-page.test.tsx b/apps/web/src/features/workflows/workflow-editor-page.test.tsx index 6881c57..db4bf58 100644 --- a/apps/web/src/features/workflows/workflow-editor-page.test.tsx +++ b/apps/web/src/features/workflows/workflow-editor-page.test.tsx @@ -37,6 +37,8 @@ test("run detail view shows node status badges from run data", () => { workflowName: "Delivery Normalize", status: "running", assetIds: ["asset-1"], + durationMs: 2450, + summaryLabel: "2 tasks complete, 1 running, 1 stdout line", }, tasks: [ { @@ -48,6 +50,8 @@ test("run detail view shows node status badges from run data", () => { artifactIds: ["artifact-1"], durationMs: 1200, summaryLabel: "Processed 1 asset", + stdoutLines: ["asset ready"], + stderrLines: [], logLines: ["Asset loaded"], }, { @@ -59,6 +63,8 @@ test("run detail view shows node status badges from run data", () => { artifactIds: ["artifact-2"], durationMs: 2450, summaryLabel: "Validated delivery package structure", + stdoutLines: ["Checking metadata"], + stderrLines: ["Minor warning"], 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, /Checking metadata/); 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, /Validated delivery package structure/); + assert.match(html, /Stdout/); + assert.match(html, /Minor warning/); assert.match(html, /\/explore\/artifact-2/); }); diff --git a/apps/web/src/runtime/app.tsx b/apps/web/src/runtime/app.tsx index f8b637e..20bbf1b 100644 --- a/apps/web/src/runtime/app.tsx +++ b/apps/web/src/runtime/app.tsx @@ -34,6 +34,16 @@ function formatTaskSummary(task: any) { 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() { const [pathname, setPathname] = useState( typeof window === "undefined" ? "/assets" : window.location.pathname || "/assets", @@ -709,6 +719,10 @@ function RunDetailPage(props: { ? run.assetIds.map((assetId: string) => assetId).join(", ") : "none"}

+

Started at: {run.startedAt ?? "n/a"}

+

Finished at: {run.finishedAt ?? "n/a"}

+

Run duration: {typeof run.durationMs === "number" ? `${run.durationMs} ms` : "n/a"}

+

Run summary: {formatRunSummary(run)}

@@ -765,6 +779,30 @@ function RunDetailPage(props: {

No task artifacts yet.

)}
+
+ Stdout + {(selectedTask.stdoutLines ?? []).length > 0 ? ( +
    + {(selectedTask.stdoutLines ?? []).map((line: string, index: number) => ( +
  • {line}
    • + ))} +
+ ) : ( +

No stdout lines.

+ )} +
+
+ Stderr + {(selectedTask.stderrLines ?? []).length > 0 ? ( +
    + {(selectedTask.stderrLines ?? []).map((line: string, index: number) => ( +
  • {line}
    • + ))} +
+ ) : ( +

No stderr lines.

+ )} +
Execution Log {(selectedTask.logLines ?? []).length > 0 ? ( diff --git a/apps/worker/src/contracts/execution-context.ts b/apps/worker/src/contracts/execution-context.ts index c0bb582..ca80862 100644 --- a/apps/worker/src/contracts/execution-context.ts +++ b/apps/worker/src/contracts/execution-context.ts @@ -1,14 +1,39 @@ export type ExecutorType = "python" | "docker" | "http"; export type TaskStatus = "pending" | "queued" | "running" | "success" | "failed"; +export type TaskStatusCounts = { + pending: number; + queued: number; + running: number; + success: number; + failed: number; +}; export type TaskExecutionSummary = { outcome: "success" | "failed"; executorType: ExecutorType; assetCount: number; artifactIds: string[]; + stdoutLineCount: number; + stderrLineCount: number; 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 = { id: string; workflowRunId?: string; @@ -26,6 +51,8 @@ export type TaskRecord = { finishedAt?: string; durationMs?: number; logLines?: string[]; + stdoutLines?: string[]; + stderrLines?: string[]; summary?: TaskExecutionSummary; lastResultPreview?: Record; }; diff --git a/apps/worker/src/executors/docker-executor.ts b/apps/worker/src/executors/docker-executor.ts index f8b4c74..c4600c7 100644 --- a/apps/worker/src/executors/docker-executor.ts +++ b/apps/worker/src/executors/docker-executor.ts @@ -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 { executionCount = 0; - async execute(task: TaskRecord, _context: ExecutionContext) { + async execute(task: TaskRecord, _context: ExecutionContext): Promise { 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: [], + }; } } diff --git a/apps/worker/src/executors/http-executor.ts b/apps/worker/src/executors/http-executor.ts index 608520a..509c829 100644 --- a/apps/worker/src/executors/http-executor.ts +++ b/apps/worker/src/executors/http-executor.ts @@ -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 { executionCount = 0; - async execute(task: TaskRecord, _context: ExecutionContext) { + async execute(task: TaskRecord, _context: ExecutionContext): Promise { 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: [], + }; } } diff --git a/apps/worker/src/executors/python-executor.ts b/apps/worker/src/executors/python-executor.ts index d19cf89..66e512b 100644 --- a/apps/worker/src/executors/python-executor.ts +++ b/apps/worker/src/executors/python-executor.ts @@ -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 { executionCount = 0; - async execute(task: TaskRecord, _context: ExecutionContext) { + async execute(task: TaskRecord, _context: ExecutionContext): Promise { 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: [], + }; } } diff --git a/apps/worker/src/runtime/mongo-worker-store.ts b/apps/worker/src/runtime/mongo-worker-store.ts index 8bdf921..3b8ee21 100644 --- a/apps/worker/src/runtime/mongo-worker-store.ts +++ b/apps/worker/src/runtime/mongo-worker-store.ts @@ -4,8 +4,10 @@ import type { Db } from "mongodb"; import type { ExecutorType, + RunExecutionSummary, TaskExecutionSummary, TaskRecord, + TaskStatusCounts, TaskStatus, } from "../contracts/execution-context.ts"; @@ -16,6 +18,10 @@ type WorkflowRunDocument = { status: "queued" | "running" | "success" | "failed"; triggeredBy: string; assetIds: string[]; + startedAt?: string; + finishedAt?: string; + durationMs?: number; + summary?: RunExecutionSummary; createdAt: string; updatedAt: string; }; @@ -47,6 +53,8 @@ type RunTaskDocument = { finishedAt?: string; durationMs?: number; logLines?: string[]; + stdoutLines?: string[]; + stderrLines?: string[]; summary?: TaskExecutionSummary; lastResultPreview?: Record; createdAt: string; @@ -75,11 +83,58 @@ function toTaskRecord(task: RunTaskDocument): TaskRecord { finishedAt: task.finishedAt, durationMs: task.durationMs, logLines: task.logLines ?? [], + stdoutLines: task.stdoutLines ?? [], + stderrLines: task.stderrLines ?? [], summary: task.summary, 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) { + 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) { + 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 { private readonly db: Db; @@ -173,7 +228,9 @@ export class MongoWorkerStore { finishedAt: string; durationMs: number; summary: TaskExecutionSummary; - logLine: string; + stdoutLines: string[]; + stderrLines: string[]; + logLines: string[]; lastResultPreview?: Record; }, ) { @@ -184,13 +241,15 @@ export class MongoWorkerStore { status: "success", finishedAt: input.finishedAt, durationMs: input.durationMs, + stdoutLines: input.stdoutLines, + stderrLines: input.stderrLines, summary: input.summary, lastResultPreview: input.lastResultPreview, updatedAt: input.finishedAt, }, $push: { logLines: { - $each: [input.logLine], + $each: input.logLines, }, }, }, @@ -204,7 +263,9 @@ export class MongoWorkerStore { finishedAt: string; durationMs: number; summary: TaskExecutionSummary; - logLine: string; + stdoutLines: string[]; + stderrLines: string[]; + logLines: string[]; }, ) { await this.db.collection("run_tasks").updateOne( @@ -215,12 +276,14 @@ export class MongoWorkerStore { errorMessage, finishedAt: input.finishedAt, durationMs: input.durationMs, + stdoutLines: input.stdoutLines, + stderrLines: input.stderrLines, summary: input.summary, updatedAt: input.finishedAt, }, $push: { logLines: { - $each: [input.logLine], + $each: input.logLines, }, }, }, @@ -273,14 +336,28 @@ export class MongoWorkerStore { 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 = { + 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("workflow_runs").updateOne( { _id: runId }, - { - $set: { - status, - updatedAt: nowIso(), - }, - }, + { $set: updateSet }, ); } } diff --git a/apps/worker/src/runtime/worker-runtime.ts b/apps/worker/src/runtime/worker-runtime.ts index 62e7c46..5ab3ecd 100644 --- a/apps/worker/src/runtime/worker-runtime.ts +++ b/apps/worker/src/runtime/worker-runtime.ts @@ -3,6 +3,7 @@ import { HttpExecutor } from "../executors/http-executor.ts"; import { PythonExecutor } from "../executors/python-executor.ts"; import type { ExecutionContext, + ExecutorExecutionResult, ExecutorType, TaskExecutionSummary, TaskRecord, @@ -47,13 +48,15 @@ export class WorkerRuntime { }; 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, { nodeId: task.nodeId, nodeType: task.nodeType, executorType: task.executorType, assetIds: task.assetIds, - result, + result: execution.result, }); const finishedAt = new Date().toISOString(); const summary: TaskExecutionSummary = { @@ -61,13 +64,21 @@ export class WorkerRuntime { executorType: task.executorType, assetCount: task.assetIds?.length ?? 0, artifactIds: [artifact._id], + stdoutLineCount: execution.stdoutLines.length, + stderrLineCount: execution.stderrLines.length, }; await this.store.markTaskSuccess(task.id, { finishedAt, durationMs: this.computeDurationMs(startedAt, finishedAt), summary, - logLine: "Executor completed successfully", - lastResultPreview: this.createResultPreview(result), + stdoutLines: execution.stdoutLines, + stderrLines: execution.stderrLines, + logLines: this.createTaskLogLines( + execution.stdoutLines, + execution.stderrLines, + "Executor completed successfully", + ), + lastResultPreview: this.createResultPreview(execution.result), }); if (task.workflowRunId) { await this.store.queueReadyDependents(task.workflowRunId); @@ -75,20 +86,28 @@ export class WorkerRuntime { } return this.store.getRunTask(task.id) ?? { ...task, status: "success" }; } catch (error) { - const message = error instanceof Error ? error.message : "worker execution failed"; + const executionError = this.normalizeExecutionError(error); const finishedAt = new Date().toISOString(); const summary: TaskExecutionSummary = { outcome: "failed", executorType: task.executorType, assetCount: task.assetIds?.length ?? 0, 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, durationMs: this.computeDurationMs(startedAt, finishedAt), 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) { await this.store.refreshRunStatus(task.workflowRunId); @@ -111,4 +130,47 @@ export class WorkerRuntime { } 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, + ]; + } } diff --git a/apps/worker/test/mongo-worker-runtime.spec.ts b/apps/worker/test/mongo-worker-runtime.spec.ts index 83dfab0..7f8a1e1 100644 --- a/apps/worker/test/mongo-worker-runtime.spec.ts +++ b/apps/worker/test/mongo-worker-runtime.spec.ts @@ -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?.finishedAt === "string"); 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?.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?.[1] ?? "", /stdout:/i); assert.match(task?.logLines?.at(-1) ?? "", /completed/i); assert.deepEqual(task?.lastResultPreview, { taskId: "task-export", 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) => { @@ -209,8 +222,12 @@ test("worker passes bound asset ids into the execution context and task artifact capturedTask = task; capturedContext = context; return { - taskId: task.id, - assetIds: context.assetIds, + result: { + taskId: task.id, + 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?.assetCount, 1); 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) => { @@ -269,7 +288,10 @@ test("worker persists failure summaries and task log lines when execution throws executors: { python: { 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?.summary?.outcome, "failed"); 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.ok(typeof task?.startedAt === "string"); assert.ok(typeof task?.finishedAt === "string"); 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?.[1] ?? "", /stdout:/i); + assert.match(task?.logLines?.[2] ?? "", /stderr:/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"]); }); diff --git a/design/03-workflows/workflow-execution-model.md b/design/03-workflows/workflow-execution-model.md index ec982e6..8acfd63 100644 --- a/design/03-workflows/workflow-execution-model.md +++ b/design/03-workflows/workflow-execution-model.md @@ -288,11 +288,26 @@ The worker-backed runtime now persists task execution summaries directly on `run - `startedAt` and `finishedAt` - `durationMs` - appended `logLines` +- captured `stdoutLines` and `stderrLines` - 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 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 first web authoring surface already follows the three-pane layout contract with: diff --git a/design/04-ui-ux/information-architecture-and-key-screens.md b/design/04-ui-ux/information-architecture-and-key-screens.md index f844adf..6bb514a 100644 --- a/design/04-ui-ux/information-architecture-and-key-screens.md +++ b/design/04-ui-ux/information-architecture-and-key-screens.md @@ -187,10 +187,18 @@ V1 run detail should render the selected task as a stable operations panel, not - task status and executor - task duration and timestamps - a concise execution summary +- separated stdout and stderr sections - appended log lines - artifact links into Explore - 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 Purpose: diff --git a/design/05-data/mongodb-data-model.md b/design/05-data/mongodb-data-model.md index 20d333f..87fad10 100644 --- a/design/05-data/mongodb-data-model.md +++ b/design/05-data/mongodb-data-model.md @@ -299,6 +299,7 @@ Core fields: - `summary` - `startedAt` - `finishedAt` +- `durationMs` - `createdAt` ### run_tasks @@ -325,6 +326,8 @@ Core fields: - `cacheKey` - `cacheHit` - `logLines` +- `stdoutLines` +- `stderrLines` - `errorMessage` - `summary` - `lastResultPreview` @@ -342,8 +345,17 @@ The current executable worker path expects `run_tasks` to be self-sufficient eno - upstream node dependencies - produced artifact ids - per-task status and error message -- task log lines and result preview -- structured task summaries with executor, outcome, asset count, and artifact ids +- task log lines, stdout/stderr streams, and result preview +- 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 diff --git a/docs/plans/2026-03-26-emboflow-v1-foundation-and-mvp.md b/docs/plans/2026-03-26-emboflow-v1-foundation-and-mvp.md index a947221..f16defb 100644 --- a/docs/plans/2026-03-26-emboflow-v1-foundation-and-mvp.md +++ b/docs/plans/2026-03-26-emboflow-v1-foundation-and-mvp.md @@ -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 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 follow-up observability pass adds persisted stdout/stderr fields on `run_tasks` plus aggregated run summaries, durations, and task counts on `workflow_runs`. ---