feat: improve CLI flow log streaming and job inspection (#8644)

* fix: improve CLI flow log streaming, sub-job listing, and failure handling Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * feat: add hierarchical flow status in job get and aggregated flow logs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: remove duplicate ansi color hint in job logs output Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * docs: update cli-commands skill with new job/flow features Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * test: add integration tests for flow job inspection and log aggregation Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * chore: remove internal friction discovery doc from branch Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * docs: trim cli-commands skill to reduce context bloat Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * docs: update job command descriptions and regenerate skills.ts Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * chore: commit auto-generated files from system_prompts Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: address review comments on flow streaming and test assertions - Move for-loop waiting logic outside --silent guard (Cubic #2) - Break outer loop when for-loop module fails (Cubic #3) - Strengthen test assertion: toContain("a") -> toContain("a: Generate data") (Cubic #1) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: generator regex truncating descriptions with parentheses The .command() regex used [^)]+ for the second arg, stopping at the first ')' inside description strings like "(machine-friendly)". Now matches quoted strings properly before falling back. Fixes 6 truncated descriptions across job, flow, and script commands. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-08-26 08:01:38 +00:00 · 2026-03-31 14:22:18 -04:00
parent 852c59efbb
commit 6c3c971af5
9 changed files with 531 additions and 58 deletions
@@ -340,31 +340,108 @@ async function run(
    requestBody: input,
  });

+  // Build step label map from raw_flow if available
+  const stepLabels = new Map<string, string>();
+  try {
+    const initialJob = await wmill.getJob({
+      workspace: workspace.workspaceId,
+      id,
+    });
+    const rawFlow = (initialJob as any).raw_flow;
+    if (rawFlow?.modules) {
+      for (const mod of rawFlow.modules) {
+        if (mod.id) {
+          const label = mod.summary ? `${mod.id}: ${mod.summary}` : mod.id;
+          stepLabels.set(mod.id, label);
+        }
+      }
+    }
+  } catch {
+    // Best-effort — fall back to module IDs
+  }
+
  let i = 0;
+  let lastStatus = "";
  while (true) {
    const jobInfo = await wmill.getJob({
      workspace: workspace.workspaceId,
      id,
    });
-    if (jobInfo.flow_status!.modules.length <= i) {
+
+    // Check if flow has completed (success or failure)
+    const isCompleted = (jobInfo as any).type === "CompletedJob";
+    const flowStatus = jobInfo.flow_status!;
+
+    if (flowStatus.modules.length <= i) {
      break;
    }
-    const module = jobInfo.flow_status!.modules[i];
+    const module = flowStatus.modules[i];

-    if (module.job) {
-      if (!opts.silent) {
-        log.info("====== Job " + (i + 1) + " ======");
+    // If a module has failed, track its job (to show error logs), then break
+    if (module.type === "Failure") {
+      if (module.job && !opts.silent) {
+        const label = stepLabels.get(module.id!) ?? `Step ${i + 1}`;
+        log.info("====== " + label + " ======");
        await track_job(workspace.workspaceId, module.job);
      }
+      break;
+    }
+
+    if (module.job) {
+      const label = stepLabels.get(module.id!) ?? `Step ${i + 1}`;
+      const isForLoop = (module as any).flow_jobs !== undefined;
+
+      if (isForLoop) {
+        // For-loop: track iterations as they appear, re-polling until module completes
+        let trackedIterations = 0;
+        let forLoopFailed = false;
+        while (true) {
+          const refreshed = await wmill.getJob({
+            workspace: workspace.workspaceId,
+            id,
+          });
+          const refreshedModule = refreshed.flow_status!.modules[i];
+          const flowJobs = ((refreshedModule as any).flow_jobs as string[] | undefined) ?? [];
+
+          // Track any new iterations
+          while (trackedIterations < flowJobs.length) {
+            if (!opts.silent) {
+              log.info(`====== ${label} (iteration ${trackedIterations}) ======`);
+              await track_job(workspace.workspaceId, flowJobs[trackedIterations]);
+            }
+            trackedIterations++;
+          }
+
+          if (refreshedModule.type === "Success" || refreshedModule.type === "Failure") {
+            forLoopFailed = refreshedModule.type === "Failure";
+            break;
+          }
+          await new Promise((resolve) => setTimeout(resolve, 200));
+        }
+        if (forLoopFailed) break;
+      } else {
+        if (!opts.silent) {
+          log.info("====== " + label + " ======");
+          await track_job(workspace.workspaceId, module.job);
+        }
+      }
    } else {
-      if (!opts.silent) {
-        log.info(module.type);
+      // Module not started yet — deduplicate status messages
+      const status = String(module.type);
+      if (!opts.silent && status !== lastStatus) {
+        log.info(colors.dim(status));
+        lastStatus = status;
      }
      await new Promise((resolve, _) =>
        setTimeout(() => resolve(undefined), 100)
      );
+
+      // If flow already completed while we were waiting, break out
+      if (isCompleted) break;
+
      continue;
    }
+    lastStatus = "";
    i++;
  }

@@ -379,7 +456,11 @@ async function run(
      });

      if (!opts.silent) {
-        log.info(colors.green.underline.bold("Flow ran to completion"));
+        if (jobInfo.success === false) {
+          log.info(colors.red.underline.bold("Flow failed"));
+        } else {
+          log.info(colors.green.underline.bold("Flow ran to completion"));
+        }
        log.info("\n");
      }

@@ -56,6 +56,8 @@ async function list(
    jobKinds?: string;
    label?: string;
    all?: boolean;
+    parent?: string;
+    isFlowStep?: boolean;
  }
 ) {
  if (opts.json) log.setSilent(true);
@@ -67,6 +69,12 @@ async function list(
  let successFilter = opts.success;
  if (opts.failed) successFilter = false;

+  // When --all or --parent is used, include flow sub-job kinds too
+  const showSubJobs = opts.all || opts.parent;
+  const defaultJobKinds = showSubJobs
+    ? "script,flow,singlestepflow,flowscript,flowdependencies"
+    : "script,flow,singlestepflow";
+
  const limit = Math.min(opts.limit ?? 30, 100);
  const allJobs = await wmill.listJobs({
    workspace: workspace.workspaceId,
@@ -75,9 +83,11 @@ async function list(
    running: opts.running,
    success: successFilter,
    perPage: limit,
-    jobKinds: opts.jobKinds ?? "script,flow,singlestepflow",
+    jobKinds: opts.jobKinds ?? defaultJobKinds,
    label: opts.label,
-    hasNullParent: opts.all ? undefined : true,
+    hasNullParent: showSubJobs ? undefined : true,
+    parentJob: opts.parent,
+    isFlowStep: opts.isFlowStep,
  });
  // API may return more than perPage — enforce limit client-side
  const jobs = allJobs.slice(0, limit);
@@ -108,6 +118,77 @@ async function list(
  }
 }

+function getModuleStatusIcon(type: string, success?: boolean): string {
+  switch (type) {
+    case "Success": return colors.green("✓");
+    case "Failure": return colors.red("✗");
+    case "InProgress": return colors.blue("▶");
+    case "WaitingForPriorSteps": return colors.dim("○");
+    case "WaitingForEvents": return colors.yellow("⏳");
+    default: return colors.dim("·");
+  }
+}
+
+function formatFlowSteps(
+  flowStatus: any,
+  rawFlow: any,
+) {
+  const modules = flowStatus?.modules ?? [];
+  const rawModules = rawFlow?.modules ?? [];
+
+  // Build summary map from raw_flow
+  const summaryMap = new Map<string, string>();
+  for (const mod of rawModules) {
+    if (mod.id && mod.summary) {
+      summaryMap.set(mod.id, mod.summary);
+    }
+  }
+
+  console.log(colors.bold("\nSteps:"));
+  for (const mod of modules) {
+    const icon = getModuleStatusIcon(mod.type);
+    const summary = summaryMap.get(mod.id) ?? "";
+    const label = summary ? `${mod.id}: ${summary}` : mod.id;
+    const jobId = mod.job ? colors.dim(mod.job) : "";
+    const flowJobsDuration = mod.flow_jobs_duration;
+
+    // For-loop modules: show parent line + iteration sub-lines
+    const flowJobs = mod.flow_jobs as string[] | undefined;
+    if (flowJobs && flowJobs.length > 0) {
+      // Total duration for the for-loop
+      const totalMs = flowJobsDuration?.duration_ms
+        ? (flowJobsDuration.duration_ms as number[]).reduce((a: number, b: number) => a + b, 0)
+        : undefined;
+      const durationStr = totalMs != null ? colors.dim(formatDuration(totalMs)) : "";
+      console.log(`  ${icon} ${label}  ${durationStr}`);
+
+      const flowJobsSuccess = (mod.flow_jobs_success ?? []) as boolean[];
+      const durationMs = (flowJobsDuration?.duration_ms ?? []) as number[];
+      for (let iter = 0; iter < flowJobs.length; iter++) {
+        const iterSuccess = flowJobsSuccess[iter];
+        const iterIcon = iterSuccess === true ? colors.green("✓")
+          : iterSuccess === false ? colors.red("✗")
+          : colors.dim("·");
+        const iterDur = durationMs[iter] != null ? colors.dim(formatDuration(durationMs[iter])) : "";
+        const iterJobId = colors.dim(flowJobs[iter]);
+        console.log(`    ${iterIcon} iteration ${iter}  ${iterJobId}  ${iterDur}`);
+      }
+    } else {
+      // Regular step
+      const durationStr = mod.duration_ms != null
+        ? colors.dim(formatDuration(mod.duration_ms))
+        : "";
+      console.log(`  ${icon} ${label}  ${jobId}  ${durationStr}`);
+    }
+  }
+
+  // Show hint for diving into step logs
+  const hasJobs = modules.some((m: any) => m.job);
+  if (hasJobs) {
+    console.log(colors.dim("\nUse 'wmill job logs <job-id>' for step logs"));
+  }
+}
+
 async function get(
  opts: GlobalOptions & { json?: boolean },
  id: string
@@ -141,8 +222,15 @@ async function get(
    if (j.schedule_path) {
      console.log(colors.bold("Schedule:") + " " + j.schedule_path);
    }
+
+    // Flow: show hierarchical step status
+    const isFlow = j.job_kind === "flow" || j.job_kind === "flowpreview";
+    if (isFlow && j.flow_status) {
+      formatFlowSteps(j.flow_status, j.raw_flow);
+    }
+
    if (j.result !== undefined) {
-      console.log(colors.bold("Result:"));
+      console.log(colors.bold("\nResult:"));
      console.log(JSON.stringify(j.result, null, 2));
    }
  }
@@ -173,18 +261,66 @@ async function logs(
  const workspace = await resolveWorkspace(opts);
  await requireLogin(opts);

-  // Check if this is a flow job (flows don't have top-level logs)
+  // Check if this is a flow job — if so, aggregate all step logs
  try {
    const job = await wmill.getJob({
      workspace: workspace.workspaceId,
      id,
    });
-    const jobKind = (job as any).job_kind; // job_kind not in generated types yet
-    if (jobKind === "flow" || jobKind === "flowpreview") {
-      log.info(colors.yellow(
-        "Flow jobs don't have direct logs. Each step runs as a separate job.\n" +
-        "Use 'wmill job list --all' to see sub-jobs, then 'wmill job logs <sub-job-id>' for individual step logs."
-      ));
+    const j = job as any;
+    const jobKind = j.job_kind;
+    if ((jobKind === "flow" || jobKind === "flowpreview") && j.flow_status?.modules) {
+      const modules = j.flow_status.modules;
+      const rawModules = j.raw_flow?.modules ?? [];
+      const summaryMap = new Map<string, string>();
+      for (const mod of rawModules) {
+        if (mod.id && mod.summary) summaryMap.set(mod.id, mod.summary);
+      }
+
+      // Strip the "to remove ansi colors" hint that appears in each step's logs
+      const stripHint = (text: string) =>
+        text.replace(/^to remove ansi colors.*\n?/gm, "");
+
+      let hasLogs = false;
+      for (const mod of modules) {
+        const summary = summaryMap.get(mod.id) ?? "";
+        const label = summary ? `${mod.id}: ${summary}` : mod.id;
+
+        // For-loop modules: get logs for each iteration
+        const flowJobs = mod.flow_jobs as string[] | undefined;
+        if (flowJobs && flowJobs.length > 0) {
+          for (let iter = 0; iter < flowJobs.length; iter++) {
+            try {
+              const stepLogs = await wmill.getJobLogs({
+                workspace: workspace.workspaceId,
+                id: flowJobs[iter],
+              });
+              if (stepLogs) {
+                console.log(colors.bold.cyan(`\n====== ${label} (iteration ${iter}) ======`));
+                console.log(stripHint(stepLogs));
+                hasLogs = true;
+              }
+            } catch { /* step may not exist yet */ }
+          }
+        } else if (mod.job) {
+          // Regular step
+          try {
+            const stepLogs = await wmill.getJobLogs({
+              workspace: workspace.workspaceId,
+              id: mod.job,
+            });
+            if (stepLogs) {
+              console.log(colors.bold.cyan(`\n====== ${label} ======`));
+              console.log(stripHint(stepLogs));
+              hasLogs = true;
+            }
+          } catch { /* step may not exist yet */ }
+        }
+      }
+
+      if (!hasLogs) {
+        log.info("No logs available for this flow's steps.");
+      }
      return;
    }
  } catch {
@@ -199,8 +335,10 @@ async function logs(
  if (jobLogs == null || jobLogs === "") {
    log.info("No logs available for this job.");
  } else {
+    // Strip the hint if the API already includes it, then print it once to stderr
+    const stripped = jobLogs.replace(/^to remove ansi colors.*\n?/gm, "");
    console.error("to remove ansi colors, use: | sed 's/\\x1B\\[[0-9;]\\{1,\\}[A-Za-z]//g'");
-    console.log(jobLogs);
+    console.log(stripped);
  }
 }

@@ -235,21 +373,23 @@ const listOptions = (cmd: Command) =>
    .option("--limit <limit:number>", "Number of jobs to return (default 30, max 100)")
    .option("--job-kinds <jobKinds:string>", "Filter by job kinds (default: script,flow,singlestepflow)")
    .option("--label <label:string>", "Filter by job label")
-    .option("--all", "Include sub-jobs (flow steps). By default only top-level jobs are shown");
+    .option("--all", "Include sub-jobs (flow steps). By default only top-level jobs are shown")
+    .option("--parent <parent:string>", "Filter by parent job ID (show sub-jobs of a specific flow)")
+    .option("--is-flow-step", "Show only flow step jobs");

 const command = listOptions(new Command()
  .description("Manage jobs (list, inspect, cancel)"))
  .action(list as any)
  .command("list", listOptions(new Command().description("List recent jobs")))
  .action(list as any)
-  .command("get", "Get job details and result")
+  .command("get", "Get job details. For flows: shows step tree with sub-job IDs")
  .arguments("<id:string>")
  .option("--json", "Output as JSON (for piping to jq)")
  .action(get as any)
  .command("result", "Get the result of a completed job (machine-friendly)")
  .arguments("<id:string>")
  .action(result as any)
-  .command("logs", "Get job logs")
+  .command("logs", "Get job logs. For flows: aggregates all step logs")
  .arguments("<id:string>")
  .action(logs as any)
  .command("cancel", "Cancel a running or queued job")
@@ -5074,7 +5074,7 @@ flow related commands
 - \`flow new <flow_path:string>\` - create a new empty flow
  - \`--summary <summary:string>\` - flow summary
  - \`--description <description:string>\` - flow description
- \`flow bootstrap <flow_path:string>\` - create a new empty flow (alias for new
+- \`flow bootstrap <flow_path:string>\` - create a new empty flow (alias for new)
  - \`--summary <summary:string>\` - flow summary
  - \`--description <description:string>\` - flow description
 - \`flow history <path:string>\` - Show version history for a flow
@@ -5227,10 +5227,10 @@ Manage jobs (list, inspect, cancel)
 **Subcommands:**

 - \`job list\` - List recent jobs
- \`job get <id:string>\` - Get job details and result
+- \`job get <id:string>\` - Get job details. For flows: shows step tree with sub-job IDs
  - \`--json\` - Output as JSON (for piping to jq)
- \`job result <id:string>\` - Get the result of a completed job (machine-friendly
- \`job logs <id:string>\` - Get job logs
+- \`job result <id:string>\` - Get the result of a completed job (machine-friendly)
+- \`job logs <id:string>\` - Get job logs. For flows: aggregates all step logs
 - \`job cancel <id:string>\` - Cancel a running or queued job
  - \`--reason <reason:string>\` - Reason for cancellation

@@ -5337,11 +5337,11 @@ script related commands
 - \`script list\` - list all scripts
  - \`--show-archived\` - Show archived scripts instead of active ones
  - \`--json\` - Output as JSON (for piping to jq)
- \`script push <path:file>\` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh
+- \`script push <path:file>\` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh)
  - \`--message <message:string>\` - Deployment message
 - \`script get <path:file>\` - get a script's details
  - \`--json\` - Output as JSON (for piping to jq)
- \`script show <path:file>\` - show a script's content (alias for get
+- \`script show <path:file>\` - show a script's content (alias for get)
 - \`script run <path:file>\` - run a script by path
  - \`-d --data <data:file>\` - Inputs specified as a JSON string or a file using @<filename> or stdin using @-.
  - \`-s --silent\` - Do not output anything other then the final output. Useful for scripting.
@@ -5351,10 +5351,10 @@ script related commands
 - \`script new <path:file> <language:string>\` - create a new script
  - \`--summary <summary:string>\` - script summary
  - \`--description <description:string>\` - script description
- \`script bootstrap <path:file> <language:string>\` - create a new script (alias for new
+- \`script bootstrap <path:file> <language:string>\` - create a new script (alias for new)
  - \`--summary <summary:string>\` - script summary
  - \`--description <description:string>\` - script description
- \`script generate-metadata [script:file]\` - re-generate the metadata file updating the lock and the script schema (for flows, use \`wmill flow generate-locks\`
+- \`script generate-metadata [script:file]\` - re-generate the metadata file updating the lock and the script schema (for flows, use \`wmill flow generate-locks\`)
  - \`--yes\` - Skip confirmation prompt
  - \`--dry-run\` - Perform a dry run without making changes
  - \`--lock-only\` - re-generate only the lock
@@ -9,6 +9,8 @@ import {
  setupWorkspaceProfile,
  createRemoteScript,
  createRemoteFlow,
+  createRemoteMultiStepFlow,
+  createRemoteFailingFlow,
  runRemoteScript,
  runRemoteFlow,
  waitForJob,
@@ -169,13 +171,13 @@ describe("job command", () => {
    });
  });

-  test("job logs for flow job shows helpful message", async () => {
+  test("job logs for flow job aggregates step logs", async () => {
    await withTestBackend(async (backend, tempDir) => {
      await setupWorkspaceProfile(backend);

      const uniqueId = Date.now();
      const flowPath = `f/test/flow_logs_${uniqueId}`;
-      await createRemoteFlow(backend, flowPath);
+      await createRemoteMultiStepFlow(backend, flowPath);
      const jobId = await runRemoteFlow(backend, flowPath);
      await waitForJob(backend, jobId);

@@ -185,7 +187,9 @@ describe("job command", () => {
      );

      expect(result.code).toEqual(0);
-      expect(result.stdout).toContain("Flow jobs don't have direct logs");
+      // Should show labeled step headers instead of "no direct logs"
+      expect(result.stdout).toContain("======");
+      expect(result.stdout).toContain("a: Generate data");
    });
  });

@@ -215,6 +219,130 @@ describe("job command", () => {
      expect(output).toContain("cancel");
      expect(output).toContain("--failed");
      expect(output).toContain("--running");
+      expect(output).toContain("--parent");
+      expect(output).toContain("--is-flow-step");
+    });
+  });
+
+  test("job get for flow shows hierarchical step tree", async () => {
+    await withTestBackend(async (backend, tempDir) => {
+      await setupWorkspaceProfile(backend);
+
+      const uniqueId = Date.now();
+      const flowPath = `f/test/flow_get_${uniqueId}`;
+      await createRemoteMultiStepFlow(backend, flowPath);
+      const jobId = await runRemoteFlow(backend, flowPath);
+      await waitForJob(backend, jobId);
+
+      const result = await backend.runCLICommand(
+        ["job", "get", jobId],
+        tempDir
+      );
+
+      expect(result.code).toEqual(0);
+      expect(result.stdout).toContain("Steps:");
+      // Should show step IDs from the flow definition
+      expect(result.stdout).toContain("a");
+      expect(result.stdout).toContain("b");
+      // Should show status icons (✓ for success)
+      expect(result.stdout).toContain("✓");
+    });
+  });
+
+  test("job get --json for flow includes flow_status", async () => {
+    await withTestBackend(async (backend, tempDir) => {
+      await setupWorkspaceProfile(backend);
+
+      const uniqueId = Date.now();
+      const flowPath = `f/test/flow_json_${uniqueId}`;
+      await createRemoteMultiStepFlow(backend, flowPath);
+      const jobId = await runRemoteFlow(backend, flowPath);
+      await waitForJob(backend, jobId);
+
+      const result = await backend.runCLICommand(
+        ["job", "get", jobId, "--json"],
+        tempDir
+      );
+
+      expect(result.code).toEqual(0);
+      const parsed = JSON.parse(result.stdout);
+      expect(parsed.flow_status).toBeDefined();
+      expect(parsed.flow_status.modules).toBeDefined();
+      expect(parsed.flow_status.modules.length).toBe(2);
+    });
+  });
+
+  test("job list --parent shows sub-jobs of a flow", async () => {
+    await withTestBackend(async (backend, tempDir) => {
+      await setupWorkspaceProfile(backend);
+
+      const uniqueId = Date.now();
+      const flowPath = `f/test/flow_parent_${uniqueId}`;
+      await createRemoteMultiStepFlow(backend, flowPath);
+      const jobId = await runRemoteFlow(backend, flowPath);
+      await waitForJob(backend, jobId);
+
+      const result = await backend.runCLICommand(
+        ["job", "list", "--json", "--parent", jobId],
+        tempDir
+      );
+
+      expect(result.code).toEqual(0);
+      const parsed = JSON.parse(result.stdout);
+      expect(Array.isArray(parsed)).toBe(true);
+      // A 2-step flow should have at least 2 sub-jobs
+      expect(parsed.length).toBeGreaterThanOrEqual(2);
+      // All sub-jobs should reference the parent flow
+      expect(parsed.every((j: any) => j.parent_job === jobId)).toBe(true);
+    });
+  });
+
+  test("job list --all includes sub-jobs", async () => {
+    await withTestBackend(async (backend, tempDir) => {
+      await setupWorkspaceProfile(backend);
+
+      const uniqueId = Date.now();
+      const flowPath = `f/test/flow_all_${uniqueId}`;
+      await createRemoteMultiStepFlow(backend, flowPath);
+      const jobId = await runRemoteFlow(backend, flowPath);
+      await waitForJob(backend, jobId);
+
+      const result = await backend.runCLICommand(
+        ["job", "list", "--json", "--all"],
+        tempDir
+      );
+
+      expect(result.code).toEqual(0);
+      const parsed = JSON.parse(result.stdout);
+      // Should contain both the parent flow and its sub-jobs
+      const parentJob = parsed.find((j: any) => j.id === jobId);
+      const subJobs = parsed.filter((j: any) => j.parent_job === jobId);
+      expect(parentJob).toBeDefined();
+      expect(subJobs.length).toBeGreaterThanOrEqual(2);
+    });
+  });
+
+  test("job get for failed flow shows failure status", async () => {
+    await withTestBackend(async (backend, tempDir) => {
+      await setupWorkspaceProfile(backend);
+
+      const uniqueId = Date.now();
+      const flowPath = `f/test/flow_fail_${uniqueId}`;
+      await createRemoteFailingFlow(backend, flowPath);
+      const jobId = await runRemoteFlow(backend, flowPath);
+      await waitForJob(backend, jobId);
+
+      const result = await backend.runCLICommand(
+        ["job", "get", jobId],
+        tempDir
+      );
+
+      expect(result.code).toEqual(0);
+      expect(result.stdout).toContain("failure");
+      expect(result.stdout).toContain("Steps:");
+      // Step a should succeed, step b should fail
+      expect(result.stdout).toContain("✓");
+      expect(result.stdout).toContain("✗");
    });
  });
 });
@@ -170,6 +170,130 @@ export async function runRemoteFlow(
  throw new Error(`Failed to run flow ${flowPath} after ${retries} retries`);
 }

+/**
+ * Create a multi-step flow with 2 steps (a prints, b returns result).
+ * Useful for testing hierarchical job get and aggregated logs.
+ */
+export async function createRemoteMultiStepFlow(
+  backend: TestBackend,
+  flowPath: string
+): Promise<void> {
+  const parts = flowPath.split("/");
+  if (parts[0] === "f" && parts.length > 2) {
+    await ensureFolder(backend, parts[1]);
+  }
+
+  const resp = await backend.apiRequest!(
+    `/api/w/${backend.workspace}/flows/create`,
+    {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        path: flowPath,
+        summary: "Multi-step test flow",
+        description: "A flow with two steps for testing",
+        value: {
+          modules: [
+            {
+              id: "a",
+              summary: "Generate data",
+              value: {
+                type: "rawscript",
+                content:
+                  'export async function main() { console.log("step a running"); return { value: 42 }; }',
+                language: "bun",
+                input_transforms: {},
+              },
+            },
+            {
+              id: "b",
+              summary: "Process data",
+              value: {
+                type: "rawscript",
+                content:
+                  'export async function main(data: any) { console.log("step b running"); return "done"; }',
+                language: "bun",
+                input_transforms: {
+                  data: { type: "javascript", expr: "results.a" },
+                },
+              },
+            },
+          ],
+        },
+        schema: {
+          $schema: "https://json-schema.org/draft/2020-12/schema",
+          type: "object",
+          properties: {},
+          required: [],
+        },
+      }),
+    }
+  );
+  expect(resp.status).toBeLessThan(300);
+  await resp.text();
+}
+
+/**
+ * Create a flow where step b throws an error.
+ * Useful for testing failure handling.
+ */
+export async function createRemoteFailingFlow(
+  backend: TestBackend,
+  flowPath: string
+): Promise<void> {
+  const parts = flowPath.split("/");
+  if (parts[0] === "f" && parts.length > 2) {
+    await ensureFolder(backend, parts[1]);
+  }
+
+  const resp = await backend.apiRequest!(
+    `/api/w/${backend.workspace}/flows/create`,
+    {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        path: flowPath,
+        summary: "Failing test flow",
+        description: "A flow where step b fails",
+        value: {
+          modules: [
+            {
+              id: "a",
+              summary: "Succeeding step",
+              value: {
+                type: "rawscript",
+                content:
+                  'export async function main() { return "ok"; }',
+                language: "bun",
+                input_transforms: {},
+              },
+            },
+            {
+              id: "b",
+              summary: "Failing step",
+              value: {
+                type: "rawscript",
+                content:
+                  'export async function main() { throw new Error("simulated failure"); }',
+                language: "bun",
+                input_transforms: {},
+              },
+            },
+          ],
+        },
+        schema: {
+          $schema: "https://json-schema.org/draft/2020-12/schema",
+          type: "object",
+          properties: {},
+          required: [],
+        },
+      }),
+    }
+  );
+  expect(resp.status).toBeLessThan(300);
+  await resp.text();
+}
+
 export async function createRemoteSchedule(
  backend: TestBackend,
  schedulePath: string,
@@ -116,7 +116,7 @@ flow related commands
 - `flow new <flow_path:string>` - create a new empty flow
  - `--summary <summary:string>` - flow summary
  - `--description <description:string>` - flow description
- `flow bootstrap <flow_path:string>` - create a new empty flow (alias for new
+- `flow bootstrap <flow_path:string>` - create a new empty flow (alias for new)
  - `--summary <summary:string>` - flow summary
  - `--description <description:string>` - flow description
 - `flow history <path:string>` - Show version history for a flow
@@ -269,10 +269,10 @@ Manage jobs (list, inspect, cancel)
 **Subcommands:**

 - `job list` - List recent jobs
- `job get <id:string>` - Get job details and result
+- `job get <id:string>` - Get job details. For flows: shows step tree with sub-job IDs
  - `--json` - Output as JSON (for piping to jq)
- `job result <id:string>` - Get the result of a completed job (machine-friendly
- `job logs <id:string>` - Get job logs
+- `job result <id:string>` - Get the result of a completed job (machine-friendly)
+- `job logs <id:string>` - Get job logs. For flows: aggregates all step logs
 - `job cancel <id:string>` - Cancel a running or queued job
  - `--reason <reason:string>` - Reason for cancellation

@@ -379,11 +379,11 @@ script related commands
 - `script list` - list all scripts
  - `--show-archived` - Show archived scripts instead of active ones
  - `--json` - Output as JSON (for piping to jq)
- `script push <path:file>` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh
+- `script push <path:file>` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh)
  - `--message <message:string>` - Deployment message
 - `script get <path:file>` - get a script's details
  - `--json` - Output as JSON (for piping to jq)
- `script show <path:file>` - show a script's content (alias for get
+- `script show <path:file>` - show a script's content (alias for get)
 - `script run <path:file>` - run a script by path
  - `-d --data <data:file>` - Inputs specified as a JSON string or a file using @<filename> or stdin using @-.
  - `-s --silent` - Do not output anything other then the final output. Useful for scripting.
@@ -393,10 +393,10 @@ script related commands
 - `script new <path:file> <language:string>` - create a new script
  - `--summary <summary:string>` - script summary
  - `--description <description:string>` - script description
- `script bootstrap <path:file> <language:string>` - create a new script (alias for new
+- `script bootstrap <path:file> <language:string>` - create a new script (alias for new)
  - `--summary <summary:string>` - script summary
  - `--description <description:string>` - script description
- `script generate-metadata [script:file]` - re-generate the metadata file updating the lock and the script schema (for flows, use `wmill flow generate-locks`
+- `script generate-metadata [script:file]` - re-generate the metadata file updating the lock and the script schema (for flows, use `wmill flow generate-locks`)
  - `--yes` - Skip confirmation prompt
  - `--dry-run` - Perform a dry run without making changes
  - `--lock-only` - re-generate only the lock
@@ -1643,7 +1643,7 @@ flow related commands
 - \`flow new <flow_path:string>\` - create a new empty flow
  - \`--summary <summary:string>\` - flow summary
  - \`--description <description:string>\` - flow description
- \`flow bootstrap <flow_path:string>\` - create a new empty flow (alias for new
+- \`flow bootstrap <flow_path:string>\` - create a new empty flow (alias for new)
  - \`--summary <summary:string>\` - flow summary
  - \`--description <description:string>\` - flow description
 - \`flow history <path:string>\` - Show version history for a flow
@@ -1796,10 +1796,10 @@ Manage jobs (list, inspect, cancel)
 **Subcommands:**

 - \`job list\` - List recent jobs
- \`job get <id:string>\` - Get job details and result
+- \`job get <id:string>\` - Get job details. For flows: shows step tree with sub-job IDs
  - \`--json\` - Output as JSON (for piping to jq)
- \`job result <id:string>\` - Get the result of a completed job (machine-friendly
- \`job logs <id:string>\` - Get job logs
+- \`job result <id:string>\` - Get the result of a completed job (machine-friendly)
+- \`job logs <id:string>\` - Get job logs. For flows: aggregates all step logs
 - \`job cancel <id:string>\` - Cancel a running or queued job
  - \`--reason <reason:string>\` - Reason for cancellation

@@ -1906,11 +1906,11 @@ script related commands
 - \`script list\` - list all scripts
  - \`--show-archived\` - Show archived scripts instead of active ones
  - \`--json\` - Output as JSON (for piping to jq)
- \`script push <path:file>\` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh
+- \`script push <path:file>\` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh)
  - \`--message <message:string>\` - Deployment message
 - \`script get <path:file>\` - get a script's details
  - \`--json\` - Output as JSON (for piping to jq)
- \`script show <path:file>\` - show a script's content (alias for get
+- \`script show <path:file>\` - show a script's content (alias for get)
 - \`script run <path:file>\` - run a script by path
  - \`-d --data <data:file>\` - Inputs specified as a JSON string or a file using @<filename> or stdin using @-.
  - \`-s --silent\` - Do not output anything other then the final output. Useful for scripting.
@@ -1920,10 +1920,10 @@ script related commands
 - \`script new <path:file> <language:string>\` - create a new script
  - \`--summary <summary:string>\` - script summary
  - \`--description <description:string>\` - script description
- \`script bootstrap <path:file> <language:string>\` - create a new script (alias for new
+- \`script bootstrap <path:file> <language:string>\` - create a new script (alias for new)
  - \`--summary <summary:string>\` - script summary
  - \`--description <description:string>\` - script description
- \`script generate-metadata [script:file]\` - re-generate the metadata file updating the lock and the script schema (for flows, use \`wmill flow generate-locks\`
+- \`script generate-metadata [script:file]\` - re-generate the metadata file updating the lock and the script schema (for flows, use \`wmill flow generate-locks\`)
  - \`--yes\` - Skip confirmation prompt
  - \`--dry-run\` - Perform a dry run without making changes
  - \`--lock-only\` - re-generate only the lock
@@ -121,7 +121,7 @@ flow related commands
 - `flow new <flow_path:string>` - create a new empty flow
  - `--summary <summary:string>` - flow summary
  - `--description <description:string>` - flow description
- `flow bootstrap <flow_path:string>` - create a new empty flow (alias for new
+- `flow bootstrap <flow_path:string>` - create a new empty flow (alias for new)
  - `--summary <summary:string>` - flow summary
  - `--description <description:string>` - flow description
 - `flow history <path:string>` - Show version history for a flow
@@ -274,10 +274,10 @@ Manage jobs (list, inspect, cancel)
 **Subcommands:**

 - `job list` - List recent jobs
- `job get <id:string>` - Get job details and result
+- `job get <id:string>` - Get job details. For flows: shows step tree with sub-job IDs
  - `--json` - Output as JSON (for piping to jq)
- `job result <id:string>` - Get the result of a completed job (machine-friendly
- `job logs <id:string>` - Get job logs
+- `job result <id:string>` - Get the result of a completed job (machine-friendly)
+- `job logs <id:string>` - Get job logs. For flows: aggregates all step logs
 - `job cancel <id:string>` - Cancel a running or queued job
  - `--reason <reason:string>` - Reason for cancellation

@@ -384,11 +384,11 @@ script related commands
 - `script list` - list all scripts
  - `--show-archived` - Show archived scripts instead of active ones
  - `--json` - Output as JSON (for piping to jq)
- `script push <path:file>` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh
+- `script push <path:file>` - push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh)
  - `--message <message:string>` - Deployment message
 - `script get <path:file>` - get a script's details
  - `--json` - Output as JSON (for piping to jq)
- `script show <path:file>` - show a script's content (alias for get
+- `script show <path:file>` - show a script's content (alias for get)
 - `script run <path:file>` - run a script by path
  - `-d --data <data:file>` - Inputs specified as a JSON string or a file using @<filename> or stdin using @-.
  - `-s --silent` - Do not output anything other then the final output. Useful for scripting.
@@ -398,10 +398,10 @@ script related commands
 - `script new <path:file> <language:string>` - create a new script
  - `--summary <summary:string>` - script summary
  - `--description <description:string>` - script description
- `script bootstrap <path:file> <language:string>` - create a new script (alias for new
+- `script bootstrap <path:file> <language:string>` - create a new script (alias for new)
  - `--summary <summary:string>` - script summary
  - `--description <description:string>` - script description
- `script generate-metadata [script:file]` - re-generate the metadata file updating the lock and the script schema (for flows, use `wmill flow generate-locks`
+- `script generate-metadata [script:file]` - re-generate the metadata file updating the lock and the script schema (for flows, use `wmill flow generate-locks`)
  - `--yes` - Skip confirmation prompt
  - `--dry-run` - Perform a dry run without making changes
  - `--lock-only` - re-generate only the lock
@@ -347,7 +347,7 @@ def parse_command_block(content: str, file_path: Path | None = None) -> dict:
    subcommand_sections = re.split(r'(?=\.command\()', block)

    for section in subcommand_sections:
-        cmd_match = re.match(r'\.command\(\s*["\']([^"\']+)["\']\s*(?:,\s*([^)]+))?\s*\)', section)
+        cmd_match = re.match(r'\.command\(\s*["\']([^"\']+)["\']\s*(?:,\s*("(?:[^"\\]|\\.)*"|\'(?:[^\'\\]|\\.)*\'|[^)]+))?\s*\)', section)
        if not cmd_match:
            continue