feat: pipeline watermark 处理完善

packages/commands/tests/e2e/pipeline.e2e.test.ts

@@ -240,4 +240,75 @@ describe("e2e: pipeline", () => {
     expect(stdout).toBe("");
     expect(stderr).toMatch(/--events must be one of: jsonl/i);
   });
+
+  test("pipeline image/generate dry-run 继承 Profile watermark=false", async () => {
+    const configDir = await mkdtemp(join(tmpdir(), "bl-pipeline-wm-"));
+    const workflowPath = join(configDir, "image-generate.json");
+    try {
+      await writeFile(
+        join(configDir, "config.json"),
+        JSON.stringify({ api_key: "sk-test-placeholder", watermark: false }, null, 2) + "\n",
+      );
+      await writeFile(
+        workflowPath,
+        JSON.stringify({
+          version: "workflow/v1",
+          steps: [{ id: "gen", type: "image/generate", input: { prompt: "A cat" } }],
+        }),
+      );
+      const { stdout, stderr, exitCode } = await runCommandE2e(
+        PIPELINE_ROUTES,
+        ["pipeline", "run", "--file", workflowPath, "--dry-run", "--output", "json"],
+        { BAILIAN_CONFIG_DIR: configDir, DASHSCOPE_API_KEY: "", DASHSCOPE_BASE_URL: "" },
+      );
+      expect(exitCode, stderr).toBe(0);
+      const report = parseStdoutJson<{
+        status?: string;
+        steps?: Array<{ type?: string; input?: { watermark?: boolean; prompt?: string } }>;
+      }>(stdout);
+      expect(report.status).toBe("planned");
+      expect(report.steps?.[0]).toMatchObject({
+        type: "image/generate",
+        input: { prompt: "A cat", watermark: false },
+      });
+    } finally {
+      await rm(configDir, { recursive: true, force: true });
+    }
+  });
+
+  test("pipeline 步骤显式 watermark=true 覆盖 Profile false", async () => {
+    const configDir = await mkdtemp(join(tmpdir(), "bl-pipeline-wm-ov-"));
+    const workflowPath = join(configDir, "image-generate.json");
+    try {
+      await writeFile(
+        join(configDir, "config.json"),
+        JSON.stringify({ api_key: "sk-test-placeholder", watermark: false }, null, 2) + "\n",
+      );
+      await writeFile(
+        workflowPath,
+        JSON.stringify({
+          version: "workflow/v1",
+          steps: [
+            {
+              id: "gen",
+              type: "image/generate",
+              input: { prompt: "A cat", watermark: true },
+            },
+          ],
+        }),
+      );
+      const { stdout, stderr, exitCode } = await runCommandE2e(
+        PIPELINE_ROUTES,
+        ["pipeline", "run", "--file", workflowPath, "--dry-run", "--output", "json"],
+        { BAILIAN_CONFIG_DIR: configDir, DASHSCOPE_API_KEY: "", DASHSCOPE_BASE_URL: "" },
+      );
+      expect(exitCode, stderr).toBe(0);
+      const report = parseStdoutJson<{
+        steps?: Array<{ input?: { watermark?: boolean } }>;
+      }>(stdout);
+      expect(report.steps?.[0]?.input?.watermark).toBe(true);
+    } finally {
+      await rm(configDir, { recursive: true, force: true });
+    }
+  });
 });

packages/runtime/src/pipeline/bl-config.ts

@@ -16,6 +16,22 @@ export interface PipelineEnv {
   settings: Settings;
 }
 
+/** Media steps that inherit Profile `watermark` when the YAML omits the field. */
+export const PIPELINE_WATERMARK_STEPS = new Set(["image/generate", "image/edit", "video/generate"]);
+
+/**
+ * Fill Profile watermark into a planned/executed step input.
+ * Explicit step `watermark` wins; otherwise use Settings (file → default true).
+ */
+export function applyProfileWatermarkToStepInput(
+  stepType: string,
+  input: Record<string, unknown>,
+  settings: Settings,
+): Record<string, unknown> {
+  if (!PIPELINE_WATERMARK_STEPS.has(stepType) || input.watermark !== undefined) return input;
+  return { ...input, watermark: settings.watermark };
+}
+
 /**
  * Build the in-process env for pipeline steps. Uses the same source resolution
  * as the CLI itself (env vars, config file; no CLI flags), but forces JSON

packages/runtime/src/pipeline/executor.ts

@@ -1,5 +1,5 @@
 import { PipelineError, toPipelineError } from "./errors.ts";
-import { buildPipelineEnv } from "./bl-config.ts";
+import { applyProfileWatermarkToStepInput, buildPipelineEnv } from "./bl-config.ts";
 import { getDefaultStepDispatcher, type StepDispatcher } from "./dispatcher.ts";
 import {
   evaluateCondition,
@@ -156,12 +156,17 @@ async function executePipelineInternal(
   if (options.dryRun) {
     for (const planStep of topologicalOrder(plan)) {
       const resolved = resolvePlannedStepInput(planStep.step, pipeline, normalizedRuntimeInput);
+      const plannedInput = applyProfileWatermarkToStepInput(
+        planStep.step.type,
+        resolved.redacted,
+        blEnv.settings,
+      );
       const report: PipelineStepReport = {
         id: planStep.step.id,
         type: planStep.step.type,
         status: "planned",
         dependencies: planStep.dependencies,
-        input: resolved.redacted,
+        input: plannedInput,
         ...(planStep.step.when !== undefined ? { condition: "pending" } : {}),
       };
       reports.push(report);
@@ -170,14 +175,14 @@ async function executePipelineInternal(
         timestamp: now(),
         status: "planned",
         step: stepEvent(planStep),
-        input: inputSummary(resolved.redacted, resolved.sensitiveKeys),
+        input: inputSummary(plannedInput, resolved.sensitiveKeys),
       });
       await emit({
         type: "step.planned",
         timestamp: now(),
         status: "planned",
         step: stepEvent(planStep),
-        input: inputSummary(resolved.redacted, resolved.sensitiveKeys),
+        input: inputSummary(plannedInput, resolved.sensitiveKeys),
         ...(planStep.step.when !== undefined ? { condition: "pending" as const } : {}),
       });
     }

packages/runtime/src/pipeline/steps/bl-api.ts

@@ -192,7 +192,8 @@ export async function imageGenerate(
     n,
     seed: input.seed,
     prompt_extend: promptExtend,
-    watermark: resolveWatermark(input.watermark),
+    // Step input overrides Profile; omit → use Profile / CLI default (true).
+    watermark: resolveWatermark(input.watermark, env.settings.watermark),
   };
 
   const body: DashScopeImageRequest =
@@ -299,7 +300,8 @@ export async function imageEdit(
     n,
     seed: input.seed,
     prompt_extend: promptExtend,
-    watermark: resolveWatermark(input.watermark),
+    // Step input overrides Profile; omit → use Profile / CLI default (true).
+    watermark: resolveWatermark(input.watermark, env.settings.watermark),
   };
 
   let body: DashScopeImageRequest;
@@ -460,7 +462,8 @@ export async function videoGenerate(
       ratio: input.ratio || undefined,
       duration: input.duration,
       prompt_extend: resolveBooleanFlag(input["prompt-extend"], undefined, "prompt-extend"),
-      watermark: resolveWatermark(input.watermark),
+      // Step input overrides Profile; omit → use Profile / CLI default (true).
+      watermark: resolveWatermark(input.watermark, env.settings.watermark),
       seed: input.seed,
     },
   };

packages/runtime/src/pipeline/steps/bl-steps.ts

@@ -1,5 +1,9 @@
 import { registerStep } from "../dispatcher.ts";
-import { buildPipelineEnv, type PipelineEnv } from "../bl-config.ts";
+import {
+  applyProfileWatermarkToStepInput,
+  buildPipelineEnv,
+  type PipelineEnv,
+} from "../bl-config.ts";
 import { isRecord } from "../utils.ts";
 import {
   textChat,
@@ -153,9 +157,13 @@ async function executeDirectBlStep(
   input: Record<string, unknown>,
   ctx: StepContext,
 ): Promise<StepResult> {
+  const env = (ctx.blEnv as PipelineEnv | undefined) ?? buildPipelineEnv();
+
   if (ctx.dryRun) {
+    // Surface the Profile-effective watermark in dry-run so plans match runtime.
+    const plannedInput = applyProfileWatermarkToStepInput(id, input, env.settings);
     return {
-      metadata: { dryRun: true, step: id, plannedInput: input },
+      metadata: { dryRun: true, step: id, plannedInput },
       warnings: [
         { code: "dry_run_skipped", message: `Step ${id} was not executed in dry-run mode` },
       ],
@@ -167,7 +175,6 @@ async function executeDirectBlStep(
     throw new Error(`No direct API handler registered for step: ${id}`);
   }
 
-  const env = (ctx.blEnv as PipelineEnv | undefined) ?? buildPipelineEnv();
   const data = await handler(env, input, ctx);
   const builder = RESULT_BUILDERS[id];
   if (builder) {

packages/runtime/tests/watermark-pipeline.test.ts

@@ -0,0 +1,129 @@
+import { expect, test } from "vite-plus/test";
+import type { Client } from "bailian-cli-core";
+import { applyProfileWatermarkToStepInput, type PipelineEnv } from "../src/pipeline/bl-config.ts";
+import { imageEdit, imageGenerate, videoGenerate } from "../src/pipeline/steps/bl-api.ts";
+import type { StepContext } from "../src/pipeline/types.ts";
+
+type CapturedRequest = {
+  path?: string;
+  method?: string;
+  body?: {
+    parameters?: { watermark?: boolean };
+  };
+  async?: boolean;
+};
+
+function makeEnv(watermark: boolean): {
+  env: PipelineEnv;
+  captured: CapturedRequest[];
+} {
+  const captured: CapturedRequest[] = [];
+  const client = {
+    uploadFile: async (source: string) => source,
+    requestJson: async (opts: CapturedRequest) => {
+      captured.push(opts);
+      if (opts.async) {
+        return { output: { task_id: "task-wm", task_status: "PENDING" } };
+      }
+      // Sync image response shape (qwen-image / wan2.7-image).
+      return {
+        request_id: "req-wm",
+        output: {
+          choices: [{ message: { content: [{ image: "https://example.com/out.png" }] } }],
+        },
+      };
+    },
+  } as unknown as Client;
+
+  return {
+    env: {
+      client,
+      settings: {
+        quiet: true,
+        output: "json",
+        outputExplicit: true,
+        timeout: 300,
+        watermark,
+        verbose: false,
+        dryRun: false,
+        telemetry: false,
+      } as PipelineEnv["settings"],
+    },
+    captured,
+  };
+}
+
+function makeCtx(): StepContext {
+  return { dryRun: false, signal: new AbortController().signal, timeoutSeconds: 1 };
+}
+
+test("pipeline imageGenerate inherits Profile watermark=false when step omits it", async () => {
+  const { env, captured } = makeEnv(false);
+  await imageGenerate(env, { prompt: "A cat" }, makeCtx());
+  expect(captured[0]?.body?.parameters?.watermark).toBe(false);
+});
+
+test("pipeline imageGenerate keeps step watermark over Profile", async () => {
+  const { env, captured } = makeEnv(false);
+  await imageGenerate(env, { prompt: "A cat", watermark: true }, makeCtx());
+  expect(captured[0]?.body?.parameters?.watermark).toBe(true);
+});
+
+test("pipeline imageEdit inherits Profile watermark=false when step omits it", async () => {
+  const { env, captured } = makeEnv(false);
+  await imageEdit(env, { prompt: "Blue sky", image: "https://example.com/in.png" }, makeCtx());
+  expect(captured[0]?.body?.parameters?.watermark).toBe(false);
+});
+
+test("pipeline videoGenerate inherits Profile watermark=false when step omits it", async () => {
+  const { env, captured } = makeEnv(false);
+  // First call submits async task; pollTaskWithOptions will keep polling — mock SUCCEEDED quickly.
+  const client = env.client as unknown as {
+    requestJson: (opts: CapturedRequest) => Promise<unknown>;
+  };
+  let calls = 0;
+  client.requestJson = async (opts: CapturedRequest) => {
+    captured.push(opts);
+    calls += 1;
+    if (calls === 1) {
+      return { output: { task_id: "task-wm", task_status: "PENDING" } };
+    }
+    return {
+      output: {
+        task_id: "task-wm",
+        task_status: "SUCCEEDED",
+        video_url: "https://example.com/out.mp4",
+      },
+    };
+  };
+
+  await videoGenerate(
+    env,
+    { prompt: "A cat walks", "poll-interval": 0 },
+    { ...makeCtx(), timeoutSeconds: 5 },
+  );
+  expect(captured[0]?.body?.parameters?.watermark).toBe(false);
+});
+
+test("pipeline defaults watermark to true when Profile leaves the compliance default", async () => {
+  const { env, captured } = makeEnv(true);
+  await imageGenerate(env, { prompt: "A cat" }, makeCtx());
+  expect(captured[0]?.body?.parameters?.watermark).toBe(true);
+});
+
+test("applyProfileWatermarkToStepInput fills omitted watermark from Settings", () => {
+  const settings = { watermark: false } as PipelineEnv["settings"];
+  expect(
+    applyProfileWatermarkToStepInput("image/generate", { prompt: "A cat" }, settings).watermark,
+  ).toBe(false);
+  expect(
+    applyProfileWatermarkToStepInput(
+      "image/generate",
+      { prompt: "A cat", watermark: true },
+      settings,
+    ).watermark,
+  ).toBe(true);
+  expect(
+    applyProfileWatermarkToStepInput("text/chat", { message: "hi" }, settings).watermark,
+  ).toBeUndefined();
+});