feat: 新增图片处理工具链与imagetool CLI命令

新增完整的图片处理能力：
1. 新增imagetool CLI工具，支持resize/crop/rotate/flip/convert/watermark/compress等基础操作，以及info/exif/histogram/colors元数据查看功能
2. 新增image_pipeline流水线构造器，支持链式编排多步图片处理DAG
3. 注册CLI别名image/img到imagetool，导出image_pipeline到顶层API
4. 配套新增完整单元测试与文档

.trae/documents/p10-image-processing-plan.md

@@ -0,0 +1,174 @@
+# P10 图片处理开发计划
+
+## Context
+
+PyFlowX 现有 `ops/files/` 工具集已覆盖 PDF（pdftool）、截图（screenshot）、文件日期（filedate）等场景，但缺少图片处理能力。`pyproject.toml` 的 `office` extra 已声明 `pillow>=10.4.0`，却无对应工具消费它。
+
+本计划新增 `imagetool` 工具，提供基础操作（resize/crop/rotate/flip/convert/watermark/compress）、元数据与信息（info/exif/histogram/colors）、批量 DAG 编排（`image_pipeline` 便捷构造器）三类能力。用户已确认范围：基础操作 + 元数据 + 批量 DAG（不含 OCR），工具形态为单一 `imagetool` + office extra。
+
+## 实现步骤
+
+### P10.1 基础操作子命令
+
+**新建 `src/pyflowx/ops/files/imagetool.py`**，参照 [pdftool.py](file:///home/zhou/pyflowx/src/pyflowx/ops/files/pdftool.py) 模式：
+
+- 顶部 `try: from PIL import Image; HAS_PIL = True; except ImportError: HAS_PIL = False`
+- `_require_pil() -> bool` 检查并打印 `pip install pyflowx[office]` 提示
+- 7 个 `@px.tool("imagetool", subcommand=..., help=...)` 子命令：
+  - `resize` (`r`) — `image_resize(input: Path, output: Path, width: int, height: int | None = None, keep_ratio: bool = True)`
+  - `crop` (`c`) — `image_crop(input: Path, output: Path, left: int, top: int, right: int, bottom: int)`
+  - `rotate` (`ro`) — `image_rotate(input: Path, output: Path, degrees: float, expand: bool = False)`
+  - `flip` (`fl`) — `image_flip(input: Path, output: Path, direction: str = "horizontal")`
+  - `convert` (`cv`) — `image_convert(input: Path, output: Path, format: str | None = None, quality: int = 85)`
+  - `watermark` (`wm`) — `image_watermark(input: Path, output: Path, text: str, position: str = "bottom-right", opacity: float = 0.5, font_size: int = 32)`
+  - `compress` (`cp`) — `image_compress(input: Path, output: Path, quality: int = 85)`
+
+实现要点：
+- 用 `Image.open(input)` 打开，操作后 `img.save(output, format=..., quality=...)` 保存
+- `resize` + `keep_ratio=True` 时用 `Image.thumbnail((width, height))` 保比缩放；`keep_ratio=False` 用 `Image.resize((width, height))`
+- `watermark` 用 `ImageDraw.Draw(img).text(...)` + `ImageFont.truetype()`（字体缺失回退 `load_default()`）
+- `convert` 根据 `output.suffix` 推断格式；JPEG 系列需先 `img.convert("RGB")` 去 alpha 通道
+- `__all__` 列出所有 `image_*` 函数名
+
+### P10.2 元数据与信息子命令
+
+继续在 `imagetool.py` 追加 4 个子命令：
+
+- `info` (`i`) — `image_info(input: Path, json: bool = False)`：打印尺寸/格式/模式/色彩空间/EXIF 摘要。`json=True` 输出 JSON 格式
+- `exif` (`e`) — `image_exif(input: Path, output: Path | None = None, show: bool = True, set: list[str] | None = None, clear: bool = False)`：读写 EXIF。`--set KEY=VAL` 修改，`--clear` 清空，`output` 另存
+- `histogram` (`hi`) — `image_histogram(input: Path, channel: str = "rgb")`：打印 RGB/Luminance 直方图统计（每通道 8 桶，纯文本表格输出）
+- `colors` (`co`) — `image_colors(input: Path, count: int = 5)`：主色调提取，用 `Image.quantize(colors=count).getpalette()` 提取并打印十六进制色值
+
+实现要点：
+- `info` 用 `img.size` / `img.format` / `img.mode` / `img.info.get("exif")`
+- `exif` 用 `img.getexif()` / `img.save(exif=bytes(exif))`
+- `histogram` 用 `img.histogram()` 分桶统计
+- `colors` 用 `img.quantize(colors=count).getpalette()` 取前 N 色
+
+### P10.3 批量 DAG 编排
+
+**新建 `src/pyflowx/imaging.py`**（类似 [compose.py](file:///home/zhou/pyflowx/src/pyflowx/compose.py) 的独立模块），提供链式 DAG 构造器：
+
+```python
+def image_pipeline(
+    source: str | Path,
+    steps: list[tuple[str, dict[str, Any]]],
+    *,
+    output_dir: str | Path | None = None,
+    naming: str = "{stem}_{step}{ext}",
+) -> Graph:
+    """图片处理流水线 DAG 构造器。
+
+    每个 step 是 (操作名, 参数字典)，操作名对应 imagetool 子命令
+    （resize/crop/rotate/flip/convert/watermark/compress）。
+    前一步输出作为后一步输入，形成链式 DAG。
+    """
+```
+
+示例：
+```python
+graph = px.image_pipeline(
+    "input.jpg",
+    steps=[
+        ("resize", {"width": 800}),
+        ("watermark", {"text": "© 2026"}),
+        ("convert", {"format": "webp", "quality": 85}),
+    ],
+)
+report = px.run(graph, strategy="sequential")
+```
+
+实现要点：
+- 每个 step 生成一个 `TaskSpec`，`fn` 调用对应的 `image_*` 函数
+- 任务名按 `naming` 模板生成（默认 `{stem}_{step}{ext}`，如 `input_resize.jpg`）
+- 任务间用 `depends_on` 链式依赖，上游输出路径 = 下游输入路径
+- `output_dir` 默认为源文件所在目录
+- 在 `__init__.py` 导出 `image_pipeline`
+
+### P10.4 验证与文档
+
+**测试 `tests/cli/test_imagetool.py`**：
+- 用 `pytest.importorskip("PIL")` 跳过未安装 Pillow 的环境
+- 用 `tmp_path` 创建真实测试图片（`Image.new("RGB", (100, 100), "red").save(tmp_path / "test.png")`）
+- 每个子命令至少 1 个测试（基础操作验证输出文件存在 + 尺寸/格式正确）
+- 元数据测试验证输出内容（info 的 JSON 结构、exif 读写往返、histogram 桶数、colors 色值数）
+- `image_pipeline` 测试验证 DAG 拓扑（3 步链式，每步输出存在且最终输出为 webp）
+- 测试 mock 优先级遵循 `python-standards.md`：优先用真实 Pillow 操作（非 subprocess），仅字体加载等不可控部分用 `monkeypatch`
+
+**新增 `tests/imaging/test_image_pipeline.py`**（或在 `tests/test_imaging.py`）：
+- 测试 `image_pipeline` 生成的 Graph 拓扑正确（任务数、依赖链）
+- 测试实际执行后输出文件链完整
+
+**注册到 CLI**：
+- 编辑 [src/pyflowx/cli/pf.py](file:///home/zhou/pyflowx/src/pyflowx/cli/pf.py) 的 `_TOOL_MODULES`：添加 `"imagetool": "pyflowx.ops.files.imagetool"`
+- 编辑 `_TOOL_ALIASES`：添加 `"imagetool": "imagetool"`、`"image": "imagetool"`、`"img": "imagetool"`
+- 编辑 [src/pyflowx/__init__.py](file:///home/zhou/pyflowx/src/pyflowx/__init__.py) 导出 `image_pipeline`
+
+**文档同步**：
+- `.trae/docs/iter-22-image-processing.md` 迭代记录
+- `.trae/skills/pyflowx-development/SKILL.md` 同步行为变更（新增"十五、图片处理工具"小节，或在十二章 CLI 工具下追加）
+- `project_memory.md` 追加 P10 工程约定
+
+## 关键文件
+
+| 操作 | 文件 |
+|------|------|
+| 新建 | `src/pyflowx/ops/files/imagetool.py`（11 子命令） |
+| 新建 | `src/pyflowx/imaging.py`（`image_pipeline` 构造器） |
+| 新建 | `tests/cli/test_imagetool.py`（子命令测试） |
+| 新建 | `tests/test_imaging.py`（DAG 构造器测试） |
+| 修改 | `src/pyflowx/cli/pf.py`（注册 imagetool + 别名） |
+| 修改 | `src/pyflowx/__init__.py`（导出 `image_pipeline`） |
+| 新建 | `.trae/docs/iter-22-image-processing.md` |
+| 修改 | `.trae/skills/pyflowx-development/SKILL.md`（行为变更同步） |
+
+## 复用的现有模式
+
+- **`@px.tool` 装饰器 + ToolSpec**：[tools.py L116-L180](file:///home/zhou/pyflowx/src/pyflowx/tools.py#L116-L180)，每个子命令一个装饰器调用
+- **可选依赖 try/except + `_require_*`**：[pdftool.py L31-L59](file:///home/zhou/pyflowx/src/pyflowx/ops/files/pdftool.py#L31-L59)
+- **`_TOOL_MODULES` 注册**：[pf.py L112-L137](file:///home/zhou/pyflowx/src/pyflowx/cli/pf.py#L112-L137)
+- **`pytest.importorskip` 跳过缺依赖**：[test_pdftool.py L22](file:///home/zhou/pyflowx/tests/cli/test_pdftool.py#L22)
+- **`image_pipeline` 委托模式**：参照 [compose.py](file:///home/zhou/pyflowx/src/pyflowx/compose.py) 的 `compose()` 函数式构造器
+- **TaskSpec 链式依赖**：用 `Graph.from_specs` + `depends_on` 构建链式 DAG（P9.2 `pipeline()` 同模式）
+
+## 验证方法
+
+```bash
+# 1. 单元测试
+uv run pytest tests/cli/test_imagetool.py tests/test_imaging.py -v
+
+# 2. 全套门禁
+uv run ruff check .
+uv run ruff format --check .
+uv run pyrefly check .
+uv run pytest --cov=pyflowx --cov-branch
+
+# 3. CLI 实测（需 pip install pyflowx[office]）
+pf imagetool info test.png
+pf imagetool resize test.png out.png --width 50
+pf imagetool watermark test.png wm.png --text "TEST"
+pf imagetool convert test.png out.webp --quality 80
+
+# 4. DAG 编排实测
+python -c "
+import pyflowx as px
+g = px.image_pipeline('test.png', steps=[
+    ('resize', {'width': 50}),
+    ('watermark', {'text': 'X'}),
+    ('convert', {'format': 'webp', 'quality': 80}),
+])
+r = px.run(g, strategy='sequential')
+print(r.success)
+"
+
+# 5. 覆盖率门槛 ≥ 95%（branch）
+```
+
+## 验收标准
+
+- ruff/pyrefly 0 错误，pytest 全绿，覆盖率 ≥ 95%（branch）
+- `pf imagetool --help` 列出全部 11 个子命令
+- `pf imagetool <sub> --help` 显示参数说明
+- `image_pipeline()` 生成的 DAG 可执行，链式输出完整
+- 未安装 office extra 时 `pf imagetool <sub>` 打印友好提示而非崩溃
+- `tests/cli/test_tool_modules.py` 自动覆盖新工具注册（无需额外修改）

src/pyflowx/__init__.py

@@ -77,6 +77,7 @@ from .errors import (
 from .executors import Strategy, run, run_iter
 from .graph import Graph, GraphDefaults
 from .history import RunHistory
+from .imaging import image_pipeline
 from .monitoring import MetricsCollector, health_check, start_metrics_server
 from .notification import (
     ALL_LEVELS,
@@ -180,6 +181,7 @@ __all__ = [
     "describe_injection",
     "diagnose",
     "health_check",
+    "image_pipeline",
     "list_subcommands",
     "list_tools",
     "load_yaml",

src/pyflowx/cli/pf.py

@@ -63,6 +63,9 @@ class PfApp:
         "gitt": "gittool",
         "gittool": "gittool",
         "gt": "gittool",
+        "image": "imagetool",
+        "imagetool": "imagetool",
+        "img": "imagetool",
         "ls": "lscalc",
         "lscalc": "lscalc",
         "msdown": "msdownload",
@@ -120,6 +123,7 @@ class PfApp:
         "folderback": "pyflowx.ops.files.folderback",
         "folderzip": "pyflowx.ops.files.folderzip",
         "gittool": "pyflowx.ops.dev.gittool",
+        "imagetool": "pyflowx.ops.files.imagetool",
         "lscalc": "pyflowx.ops.dev.lscalc",
         "msdownload": "pyflowx.ops.infra.msdownload",
         "packtool": "pyflowx.ops.dev.packtool",

src/pyflowx/imaging.py

@@ -0,0 +1,130 @@
+"""图片处理流水线 DAG 构造器.
+
+将 :mod:`pyflowx.ops.files.imagetool` 的操作封装为链式 :class:`Graph`,
+便于在 DAG 中组合多步图片处理 (如 resize → watermark → convert).
+
+设计参照 :mod:`pyflowx.compose` 的函数式构造器模式.
+"""
+
+from __future__ import annotations
+
+__all__ = ["image_pipeline"]
+
+from pathlib import Path
+from typing import Any
+
+from .graph import Graph
+from .task import TaskSpec
+
+
+def image_pipeline(
+    source: str | Path,
+    steps: list[tuple[str, dict[str, Any]]],
+    *,
+    output_dir: str | Path | None = None,
+    naming: str = "{stem}_{step}{ext}",
+) -> Graph:
+    """图片处理流水线 DAG 构造器.
+
+    每个 step 是 ``(操作名, 参数字典)``, 操作名对应 imagetool 子命令
+    (resize/crop/rotate/flip/convert/watermark/compress).
+    前一步输出作为后一步输入, 形成链式 DAG.
+
+    Parameters
+    ----------
+    source : str | Path
+        源图片路径
+    steps : list[tuple[str, dict[str, Any]]]
+        操作步骤列表, 如 ``[("resize", {"width": 800}), ("watermark", {"text": "X"})]``
+    output_dir : str | Path | None
+        输出目录 (None 时用源文件所在目录)
+    naming : str
+        输出文件名模板, 可用占位符 ``{stem}`` (累积 stem) / ``{step}`` (操作名) /
+        ``{ext}`` (当前扩展名, convert 步骤可能改变). 默认 ``"{stem}_{step}{ext}"``
+
+    Returns
+    -------
+    Graph
+        链式 DAG, 每步为一个 :class:`TaskSpec`, 通过 ``depends_on`` 链接
+
+    Raises
+    ------
+    ValueError
+        操作名不在支持列表中时
+
+    Example
+    -------
+    >>> graph = px.image_pipeline(
+    ...     "input.jpg",
+    ...     steps=[
+    ...         ("resize", {"width": 800}),
+    ...         ("watermark", {"text": "© 2026"}),
+    ...         ("convert", {"format": "webp", "quality": 85}),
+    ...     ],
+    ... )
+    >>> report = px.run(graph, strategy="sequential")
+    """
+    from .ops.files import imagetool
+
+    operations = {
+        "resize": imagetool.image_resize,
+        "crop": imagetool.image_crop,
+        "rotate": imagetool.image_rotate,
+        "flip": imagetool.image_flip,
+        "convert": imagetool.image_convert,
+        "watermark": imagetool.image_watermark,
+        "compress": imagetool.image_compress,
+    }
+
+    source_path = Path(source)
+    out_dir = Path(output_dir) if output_dir else source_path.parent
+
+    specs: list[TaskSpec[Any]] = []
+    prev_name: str | None = None
+    prev_output: Path | None = None
+    current_stem = source_path.stem
+    current_ext = source_path.suffix
+
+    for i, (op_name, params) in enumerate(steps):
+        if op_name not in operations:
+            raise ValueError(f"未知图片操作: {op_name!r}, 支持: {sorted(operations)}")
+
+        if op_name == "convert" and "format" in params:
+            current_ext = f".{params['format'].lower()}"
+
+        output_name = naming.format(stem=current_stem, step=op_name, ext=current_ext)
+        output_path = out_dir / output_name
+        input_path = source_path if prev_output is None else prev_output
+
+        task_name = f"step_{i:02d}_{op_name}"
+        base_fn = operations[op_name]
+        fn = _make_step_fn(base_fn, input_path, output_path, params)
+        fn.__name__ = task_name
+
+        depends_on = (prev_name,) if prev_name else ()
+        spec = TaskSpec(task_name, fn=fn, depends_on=depends_on)
+        specs.append(spec)
+
+        prev_name = task_name
+        prev_output = output_path
+        current_stem = output_path.stem
+
+    return Graph.from_specs(specs)
+
+
+def _make_step_fn(
+    base_fn: Any,
+    input_path: Path,
+    output_path: Path,
+    params: dict[str, Any],
+) -> Any:
+    """创建步骤执行闭包, 绑定输入/输出路径与参数.
+
+    闭包捕获的是 :func:`_make_step_fn` 的参数 (每次调用独立作用域),
+    不会受外层循环变量重绑定影响.
+    """
+
+    def _run() -> None:
+        base_fn(input_path, output_path, **params)
+
+    return _run

src/pyflowx/ops/files/imagetool.py

@@ -0,0 +1,520 @@
+"""imagetool - 图片处理工具集.
+
+提供 resize/crop/rotate/flip/convert/watermark/compress/info/exif/
+histogram/colors 子命令, 基于 Pillow 实现.
+
+依赖 ``pyflowx[office]`` extra 中的 ``pillow``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pyflowx as px
+
+__all__ = [
+    "image_colors",
+    "image_compress",
+    "image_convert",
+    "image_crop",
+    "image_exif",
+    "image_flip",
+    "image_histogram",
+    "image_info",
+    "image_resize",
+    "image_rotate",
+    "image_watermark",
+]
+
+try:
+    from PIL import Image, ImageDraw, ImageFont
+
+    HAS_PIL = True
+except ImportError:
+    HAS_PIL = False
+
+
+def _require_pil() -> bool:
+    """Pillow 未安装时打印提示, 返回是否可用."""
+    if not HAS_PIL:
+        print("未安装 Pillow 库, 请安装: pip install pyflowx[office]")
+        return False
+    return True
+
+
+def _save_image(img: Any, output: Path, fmt: str | None = None, quality: int = 85) -> None:
+    """保存图片, 自动处理 JPEG 不支持 alpha 通道的情况."""
+    output.parent.mkdir(parents=True, exist_ok=True)
+    save_fmt = fmt or output.suffix.lstrip(".").upper()
+    if save_fmt == "JPG":
+        save_fmt = "JPEG"
+    if save_fmt == "JPEG":
+        img = img.convert("RGB")
+    if save_fmt in ("JPEG", "WEBP"):
+        img.save(output, format=save_fmt, quality=quality)
+    else:
+        img.save(output, format=save_fmt)
+
+
+# ---------------------------------------------------------------------- #
+# 基础操作
+# ---------------------------------------------------------------------- #
+@px.tool("imagetool", subcommand="r", help="调整尺寸")
+def image_resize(
+    input_path: Path,
+    output_path: Path,
+    width: int,
+    height: int | None = None,
+    keep_ratio: bool = True,
+) -> None:
+    """调整图片尺寸.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片
+    width : int
+        目标宽度
+    height : int | None
+        目标高度 (keep_ratio=True 时仅作上限, 默认 None 表示按宽度等比)
+    keep_ratio : bool
+        是否保持宽高比 (默认 True)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    if keep_ratio:
+        target_height = height if height is not None else width
+        img.thumbnail((width, target_height))
+    else:
+        if height is None:
+            height = width
+        img = img.resize((width, height))
+    _save_image(img, output_path)
+    print(f"调整尺寸完成: {output_path} ({img.size[0]}x{img.size[1]})")
+
+
+@px.tool("imagetool", subcommand="c", help="裁剪图片")
+def image_crop(
+    input_path: Path,
+    output_path: Path,
+    left: int,
+    top: int,
+    right: int,
+    bottom: int,
+) -> None:
+    """裁剪图片到指定矩形.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片
+    left, top, right, bottom : int
+        裁剪矩形坐标 (左上为原点)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    cropped = img.crop((left, top, right, bottom))
+    _save_image(cropped, output_path)
+    print(f"裁剪完成: {output_path} ({right - left}x{bottom - top})")
+
+
+@px.tool("imagetool", subcommand="ro", help="旋转图片")
+def image_rotate(
+    input_path: Path,
+    output_path: Path,
+    degrees: float,
+    expand: bool = False,
+) -> None:
+    """旋转图片.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片
+    degrees : float
+        旋转角度 (正值逆时针)
+    expand : bool
+        是否扩展画布以容纳整个旋转后的图片 (默认 False)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    rotated = img.rotate(degrees, expand=expand)
+    _save_image(rotated, output_path)
+    print(f"旋转完成: {output_path} ({degrees}度)")
+
+
+@px.tool("imagetool", subcommand="fl", help="翻转图片")
+def image_flip(
+    input_path: Path,
+    output_path: Path,
+    direction: str = "horizontal",
+) -> None:
+    """翻转图片.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片
+    direction : str
+        翻转方向: "horizontal" (水平镜像) / "vertical" (垂直镜像)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    method = Image.Transpose.FLIP_LEFT_RIGHT if direction == "horizontal" else Image.Transpose.FLIP_TOP_BOTTOM
+    flipped = img.transpose(method)
+    _save_image(flipped, output_path)
+    print(f"翻转完成: {output_path} ({direction})")
+
+
+@px.tool("imagetool", subcommand="cv", help="格式转换")
+def image_convert(
+    input_path: Path,
+    output_path: Path,
+    format: str | None = None,
+    quality: int = 85,
+) -> None:
+    """转换图片格式.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片 (后缀决定格式, 除非 format 显式指定)
+    format : str | None
+        目标格式 (如 "PNG"/"JPEG"/"WEBP"), None 时按 output_path 后缀推断
+    quality : int
+        压缩质量 (1-100, 仅对 JPEG/WEBP 有效)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    _save_image(img, output_path, fmt=format, quality=quality)
+    actual_fmt = format or output_path.suffix.lstrip(".").upper()
+    print(f"格式转换完成: {output_path} ({actual_fmt})")
+
+
+@px.tool("imagetool", subcommand="wm", help="添加文字水印")
+def image_watermark(
+    input_path: Path,
+    output_path: Path,
+    text: str,
+    position: str = "bottom-right",
+    opacity: float = 0.5,
+    font_size: int = 32,
+) -> None:
+    """添加文字水印.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片
+    text : str
+        水印文字
+    position : str
+        水印位置: top-left/top-right/bottom-left/bottom-right/center
+    opacity : float
+        不透明度 (0.0-1.0)
+    font_size : int
+        字体大小
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path).convert("RGBA")
+    overlay = Image.new("RGBA", img.size, (0, 0, 0, 0))
+    draw = ImageDraw.Draw(overlay)
+
+    font = _load_font(font_size)
+    bbox = draw.textbbox((0, 0), text, font=font)
+    text_w = int(bbox[2] - bbox[0])
+    text_h = int(bbox[3] - bbox[1])
+    margin = 10
+    x, y = _resolve_position(position, img.size, text_w, text_h, margin)
+
+    alpha = int(255 * max(0.0, min(1.0, opacity)))
+    draw.text((x, y), text, font=font, fill=(255, 255, 255, alpha))
+    result = Image.alpha_composite(img, overlay)
+    _save_image(result, output_path)
+    print(f"水印添加完成: {output_path}")
+
+
+def _load_font(size: int) -> Any:
+    """加载字体, 优先 truetype, 失败回退默认字体."""
+    candidates = ("DejaVuSans.ttf", "Arial.ttf", "LiberationSans-Regular.ttf")
+    for name in candidates:
+        try:
+            return ImageFont.truetype(name, size)
+        except OSError:
+            continue
+    return ImageFont.load_default()
+
+
+def _resolve_position(
+    position: str,
+    img_size: tuple[int, int],
+    text_w: int,
+    text_h: int,
+    margin: int,
+) -> tuple[int, int]:
+    """根据位置描述符计算水印坐标."""
+    w, h = img_size
+    pos_map = {
+        "top-left": (margin, margin),
+        "top-right": (w - text_w - margin, margin),
+        "bottom-left": (margin, h - text_h - margin),
+        "bottom-right": (w - text_w - margin, h - text_h - margin),
+        "center": ((w - text_w) // 2, (h - text_h) // 2),
+    }
+    return pos_map.get(position, pos_map["bottom-right"])
+
+
+@px.tool("imagetool", subcommand="cp", help="压缩图片")
+def image_compress(
+    input_path: Path,
+    output_path: Path,
+    quality: int = 85,
+) -> None:
+    """压缩图片 (重新编码以减小体积).
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path
+        输出图片
+    quality : int
+        压缩质量 (1-100)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    fmt = input_path.suffix.lstrip(".").upper()
+    _save_image(img, output_path, fmt=fmt, quality=quality)
+    in_size = input_path.stat().st_size
+    out_size = output_path.stat().st_size
+    ratio = (1 - out_size / in_size) * 100 if in_size > 0 else 0.0
+    print(f"压缩完成: {output_path} (原 {in_size}B → 新 {out_size}B, 节省 {ratio:.1f}%)")
+
+
+# ---------------------------------------------------------------------- #
+# 元数据与信息
+# ---------------------------------------------------------------------- #
+@px.tool("imagetool", subcommand="i", help="查看图片信息")
+def image_info(input_path: Path, json: bool = False) -> None:
+    """打印图片信息 (尺寸/格式/模式/EXIF 摘要).
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    json : bool
+        是否以 JSON 格式输出 (默认纯文本表格)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    exif = img.getexif()
+    exif_count = len(exif) if exif else 0
+
+    import json as json_mod
+
+    data = {
+        "path": str(input_path),
+        "format": img.format,
+        "mode": img.mode,
+        "width": img.size[0],
+        "height": img.size[1],
+        "exif_tags": exif_count,
+    }
+    if json:
+        print(json_mod.dumps(data, ensure_ascii=False, indent=2))
+    else:
+        print(f"文件: {data['path']}")
+        print(f"格式: {data['format']}")
+        print(f"模式: {data['mode']}")
+        print(f"尺寸: {data['width']}x{data['height']}")
+        print(f"EXIF 标签数: {data['exif_tags']}")
+
+
+@px.tool("imagetool", subcommand="e", help="读取/修改 EXIF")
+def image_exif(
+    input_path: Path,
+    output_path: Path | None = None,
+    show: bool = True,
+    set: list[str] | None = None,
+    clear: bool = False,
+) -> None:
+    """读取或修改 EXIF 元数据.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    output_path : Path | None
+        输出路径 (None 时原地覆盖; 仅 show=True 时可省略)
+    show : bool
+        打印全部 EXIF 标签 (默认 True)
+    set : list[str] | None
+        设置标签, 格式 ["KEY=VALUE", ...] (KEY 为数字标签号)
+    clear : bool
+        清空所有 EXIF 标签 (在 set 之前执行)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    exif = img.getexif()
+
+    if show:
+        _print_exif(exif)
+
+    modified = _apply_exif_modifications(exif, set, clear)
+    if modified:
+        _save_exif(img, exif, output_path if output_path is not None else input_path)
+
+
+def _print_exif(exif: Any) -> None:
+    """打印 EXIF 标签."""
+    if exif:
+        for tag, value in exif.items():
+            print(f"  {tag}: {value}")
+    else:
+        print("  (无 EXIF 数据)")
+
+
+def _apply_exif_modifications(exif: Any, set_items: list[str] | None, clear: bool) -> bool:
+    """应用 EXIF 修改 (clear + set), 返回是否有改动."""
+    if clear:
+        for tag in list(exif.keys()):
+            del exif[tag]
+    if set_items:
+        for item in set_items:
+            _apply_single_exif_set(exif, item)
+    return bool(set_items or clear)
+
+
+def _apply_single_exif_set(exif: Any, item: str) -> None:
+    """解析并应用单个 KEY=VALUE 设置项."""
+    if "=" not in item:
+        print(f"跳过无效项 (缺少 =): {item}")
+        return
+    key_str, value = item.split("=", 1)
+    try:
+        tag = int(key_str)
+    except ValueError:
+        print(f"跳过无效标签号: {key_str}")
+        return
+    exif[tag] = value
+
+
+def _save_exif(img: Any, exif: Any, output_path: Path) -> None:
+    """保存图片与 EXIF."""
+    exif_bytes = exif.tobytes() if exif else b""
+    img.save(output_path, exif=exif_bytes)
+    print(f"EXIF 已保存: {output_path}")
+
+
+@px.tool("imagetool", subcommand="hi", help="颜色直方图")
+def image_histogram(
+    input_path: Path,
+    channel: str = "rgb",
+) -> None:
+    """打印颜色直方图统计 (每通道 8 桶).
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    channel : str
+        通道: "rgb" (R/G/B 三通道) / "luminance" (亮度单通道)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path)
+    hist = img.histogram()
+    buckets = 8
+
+    if channel == "luminance":
+        gray = img.convert("L")
+        gray_hist = gray.histogram()
+        print("亮度直方图 (8 桶):")
+        _print_histogram_buckets(gray_hist, buckets, "L")
+    else:
+        print("RGB 直方图 (8 桶):")
+        if len(hist) == 256:
+            _print_histogram_buckets(hist, buckets, "L")
+        else:
+            for idx, name in enumerate(("R", "G", "B")):
+                start = idx * 256
+                _print_histogram_buckets(hist[start : start + 256], buckets, name)
+
+
+def _print_histogram_buckets(channel_hist: list[int], buckets: int, name: str) -> None:
+    """将单通道 256 桶直方图聚合为指定桶数并打印."""
+    bucket_size = max(1, len(channel_hist) // buckets)
+    print(f"  {name}:")
+    for i in range(buckets):
+        start = i * bucket_size
+        end = min((i + 1) * bucket_size, len(channel_hist))
+        count = sum(channel_hist[start:end])
+        slice_max = max(channel_hist[start:end] or [1])
+        bar = "#" * min(40, count * 40 // max(1, slice_max))
+        print(f"    [{start:3d}-{end:3d}] {count:>8d} {bar}")
+
+
+@px.tool("imagetool", subcommand="co", help="提取主色调")
+def image_colors(
+    input_path: Path,
+    count: int = 5,
+) -> None:
+    """提取并打印主色调.
+
+    Parameters
+    ----------
+    input_path : Path
+        输入图片
+    count : int
+        提取的颜色数 (默认 5)
+    """
+    if not _require_pil():
+        return
+
+    img = Image.open(input_path).convert("RGB")
+    quantized = img.quantize(colors=count)
+    palette = quantized.getpalette()
+    if palette is None:
+        print("无法提取调色板")
+        return
+
+    actual_count = len(palette) // 3
+    print(f"主色调 (前 {min(count, actual_count)} 色):")
+    for i in range(min(count, actual_count)):
+        r = palette[i * 3]
+        g = palette[i * 3 + 1]
+        b = palette[i * 3 + 2]
+        hex_color = f"#{r:02X}{g:02X}{b:02X}"
+        print(f"  {i + 1}. {hex_color}  rgb({r}, {g}, {b})")

tests/cli/test_imagetool.py

@@ -0,0 +1,316 @@
+"""imagetool 子命令测试.
+
+用 ``pytest.importorskip("PIL")`` 跳过未安装 Pillow 的环境.
+用真实 Pillow 操作 (非 subprocess), 创建测试图片验证.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from pyflowx.ops.files import imagetool
+
+pytest.importorskip("PIL")
+from PIL import Image
+
+
+def _make_test_image(path: Path, size: tuple[int, int] = (100, 80), color: str = "red") -> None:
+    """创建测试图片."""
+    Image.new("RGB", size, color).save(path)
+
+
+def _make_test_image_rgba(path: Path, size: tuple[int, int] = (100, 80)) -> None:
+    """创建带 alpha 通道的测试图片."""
+    Image.new("RGBA", size, (255, 0, 0, 128)).save(path)
+
+
+class TestImageResize:
+    """image_resize 测试."""
+
+    def test_resize_with_keep_ratio(self, tmp_path: Path) -> None:
+        """keep_ratio=True 时按宽度等比缩放."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.png"
+        imagetool.image_resize(src, out, width=50)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.size[0] == 50
+
+    def test_resize_without_keep_ratio(self, tmp_path: Path) -> None:
+        """keep_ratio=False 时强制缩放到指定尺寸."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.png"
+        imagetool.image_resize(src, out, width=50, height=40, keep_ratio=False)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.size == (50, 40)
+
+
+class TestImageCrop:
+    """image_crop 测试."""
+
+    def test_crop(self, tmp_path: Path) -> None:
+        """裁剪到指定矩形."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.png"
+        imagetool.image_crop(src, out, left=10, top=10, right=60, bottom=50)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.size == (50, 40)
+
+
+class TestImageRotate:
+    """image_rotate 测试."""
+
+    def test_rotate_90(self, tmp_path: Path) -> None:
+        """旋转 90 度."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.png"
+        imagetool.image_rotate(src, out, degrees=90, expand=True)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.size[0] == 80
+            assert img.size[1] == 100
+
+
+class TestImageFlip:
+    """image_flip 测试."""
+
+    def test_flip_horizontal(self, tmp_path: Path) -> None:
+        """水平翻转."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.png"
+        imagetool.image_flip(src, out, direction="horizontal")
+        assert out.exists()
+
+    def test_flip_vertical(self, tmp_path: Path) -> None:
+        """垂直翻转."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.png"
+        imagetool.image_flip(src, out, direction="vertical")
+        assert out.exists()
+
+
+class TestImageConvert:
+    """image_convert 测试."""
+
+    def test_convert_to_jpeg(self, tmp_path: Path) -> None:
+        """PNG → JPEG 转换."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        out = tmp_path / "out.jpg"
+        imagetool.image_convert(src, out)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.format == "JPEG"
+
+    def test_convert_to_webp(self, tmp_path: Path) -> None:
+        """PNG → WEBP 转换."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        out = tmp_path / "out.webp"
+        imagetool.image_convert(src, out, quality=80)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.format == "WEBP"
+
+    def test_convert_rgba_to_jpeg(self, tmp_path: Path) -> None:
+        """RGBA → JPEG 自动去 alpha 通道."""
+        src = tmp_path / "src.png"
+        _make_test_image_rgba(src, (50, 50))
+        out = tmp_path / "out.jpg"
+        imagetool.image_convert(src, out)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.format == "JPEG"
+            assert img.mode == "RGB"
+
+
+class TestImageWatermark:
+    """image_watermark 测试."""
+
+    def test_watermark_text(self, tmp_path: Path) -> None:
+        """添加文字水印."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (200, 150))
+        out = tmp_path / "out.png"
+        imagetool.image_watermark(src, out, text="TEST", position="bottom-right", opacity=0.7)
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.size == (200, 150)
+
+    def test_watermark_center_position(self, tmp_path: Path) -> None:
+        """center 位置水印."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (200, 150))
+        out = tmp_path / "out.png"
+        imagetool.image_watermark(src, out, text="CENTER", position="center")
+        assert out.exists()
+
+
+class TestImageCompress:
+    """image_compress 测试."""
+
+    def test_compress_jpeg(self, tmp_path: Path) -> None:
+        """JPEG 压缩后文件存在."""
+        src = tmp_path / "src.jpg"
+        _make_test_image(src, (100, 80))
+        out = tmp_path / "out.jpg"
+        imagetool.image_compress(src, out, quality=50)
+        assert out.exists()
+        assert out.stat().st_size > 0
+
+
+class TestImageInfo:
+    """image_info 测试."""
+
+    def test_info_text(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """文本格式信息输出."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        imagetool.image_info(src, json=False)
+        captured = capsys.readouterr()
+        assert "格式: PNG" in captured.out
+        assert "100x80" in captured.out
+
+    def test_info_json(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """JSON 格式信息输出."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        imagetool.image_info(src, json=True)
+        captured = capsys.readouterr()
+        import json
+
+        data = json.loads(captured.out)
+        assert data["format"] == "PNG"
+        assert data["width"] == 100
+        assert data["height"] == 80
+
+
+class TestImageExif:
+    """image_exif 测试."""
+
+    def test_exif_show_empty(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """无 EXIF 时打印提示."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        imagetool.image_exif(src, show=True)
+        captured = capsys.readouterr()
+        assert "无 EXIF 数据" in captured.out
+
+    def test_exif_set_and_read(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """设置 EXIF 后能读取."""
+        src = tmp_path / "src.jpg"
+        _make_test_image(src, (50, 50))
+        out = tmp_path / "out.jpg"
+        imagetool.image_exif(src, output_path=out, show=False, set=["271=TestCamera"])
+        assert out.exists()
+
+        capsys.readouterr()
+        imagetool.image_exif(out, show=True)
+        captured = capsys.readouterr()
+        assert "271: TestCamera" in captured.out
+
+    def test_exif_clear(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """清空 EXIF."""
+        src = tmp_path / "src.jpg"
+        _make_test_image(src, (50, 50))
+        out_set = tmp_path / "set.jpg"
+        imagetool.image_exif(src, output_path=out_set, show=False, set=["271=TestCamera"])
+
+        out_clear = tmp_path / "clear.jpg"
+        imagetool.image_exif(out_set, output_path=out_clear, show=False, clear=True)
+
+        capsys.readouterr()
+        imagetool.image_exif(out_clear, show=True)
+        captured = capsys.readouterr()
+        assert "无 EXIF 数据" in captured.out
+
+    def test_exif_invalid_set_item(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """无效设置项被跳过."""
+        src = tmp_path / "src.jpg"
+        _make_test_image(src, (50, 50))
+        out = tmp_path / "out.jpg"
+        imagetool.image_exif(src, output_path=out, show=False, set=["invalid", "abc=val"])
+        captured = capsys.readouterr()
+        assert "跳过" in captured.out
+
+
+class TestImageHistogram:
+    """image_histogram 测试."""
+
+    def test_histogram_rgb(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """RGB 直方图输出 3 通道."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50), color="red")
+        imagetool.image_histogram(src, channel="rgb")
+        captured = capsys.readouterr()
+        assert "R:" in captured.out
+        assert "G:" in captured.out
+        assert "B:" in captured.out
+
+    def test_histogram_luminance(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """亮度直方图输出 L 通道."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50), color="red")
+        imagetool.image_histogram(src, channel="luminance")
+        captured = capsys.readouterr()
+        assert "L:" in captured.out
+
+
+class TestImageColors:
+    """image_colors 测试."""
+
+    def test_colors_extraction(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """主色调提取输出指定数量颜色."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50), color="red")
+        imagetool.image_colors(src, count=3)
+        captured = capsys.readouterr()
+        assert "主色调" in captured.out
+        assert "#FF0000" in captured.out or "#" in captured.out
+
+    def test_colors_count(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        """主色调数量符合请求."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50), color="blue")
+        imagetool.image_colors(src, count=5)
+        captured = capsys.readouterr()
+        lines = [line for line in captured.out.splitlines() if line.strip().startswith(tuple("12345"))]
+        assert len(lines) <= 5
+
+
+class TestRequirePil:
+    """_require_pil 未安装时的行为."""
+
+    def test_require_pil_returns_false_when_missing(
+        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        """HAS_PIL=False 时打印提示并返回 False."""
+        monkeypatch.setattr(imagetool, "HAS_PIL", False)
+        result = imagetool._require_pil()
+        assert result is False
+        captured = capsys.readouterr()
+        assert "pip install pyflowx[office]" in captured.out
+
+    def test_require_pil_returns_true_when_installed(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """HAS_PIL=True 时返回 True."""
+        monkeypatch.setattr(imagetool, "HAS_PIL", True)
+        assert imagetool._require_pil() is True
+
+    def test_resize_no_pil_noop(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        """未安装 PIL 时 resize 不创建输出文件."""
+        monkeypatch.setattr(imagetool, "HAS_PIL", False)
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        out = tmp_path / "out.png"
+        imagetool.image_resize(src, out, width=25)
+        assert not out.exists()

tests/test_imaging.py

@@ -0,0 +1,158 @@
+"""image_pipeline DAG 构造器测试.
+
+验证:
+* 生成的 Graph 拓扑正确 (任务数/依赖链).
+* 实际执行后输出文件链完整.
+* 未知操作抛 ValueError.
+* convert 步骤改变扩展名.
+* 自定义 output_dir 与 naming 模板.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import pyflowx as px
+
+pytest.importorskip("PIL")
+from PIL import Image
+
+
+def _make_test_image(path: Path, size: tuple[int, int] = (100, 80), color: str = "red") -> None:
+    """创建测试图片."""
+    Image.new("RGB", size, color).save(path)
+
+
+class TestImagePipelineTopology:
+    """image_pipeline 生成的 Graph 拓扑测试."""
+
+    def test_single_step(self, tmp_path: Path) -> None:
+        """单步流水线生成 1 个任务无依赖."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        graph = px.image_pipeline(src, steps=[("resize", {"width": 50})])
+        assert len(graph.all_specs()) == 1
+        spec = next(iter(graph.all_specs().values()))
+        assert spec.depends_on == ()
+
+    def test_multi_step_chain(self, tmp_path: Path) -> None:
+        """多步流水线形成链式依赖."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        graph = px.image_pipeline(
+            src,
+            steps=[
+                ("resize", {"width": 50}),
+                ("watermark", {"text": "X"}),
+                ("compress", {"quality": 80}),
+            ],
+        )
+        specs = graph.all_specs()
+        assert len(specs) == 3
+        names = list(specs.keys())
+        assert specs[names[0]].depends_on == ()
+        assert specs[names[1]].depends_on == (names[0],)
+        assert specs[names[2]].depends_on == (names[1],)
+
+    def test_unknown_operation_raises(self, tmp_path: Path) -> None:
+        """未知操作抛 ValueError."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        with pytest.raises(ValueError, match="未知图片操作"):
+            px.image_pipeline(src, steps=[("unknown_op", {})])
+
+
+class TestImagePipelineExecution:
+    """image_pipeline 实际执行测试."""
+
+    def test_execute_resize_only(self, tmp_path: Path) -> None:
+        """单步 resize 执行后输出文件存在且尺寸正确."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        graph = px.image_pipeline(src, steps=[("resize", {"width": 50})])
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        out = tmp_path / "src_resize.png"
+        assert out.exists()
+        with Image.open(out) as img:
+            assert img.size[0] == 50
+
+    def test_execute_chain(self, tmp_path: Path) -> None:
+        """链式执行: resize → watermark → convert."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (200, 150))
+        graph = px.image_pipeline(
+            src,
+            steps=[
+                ("resize", {"width": 100}),
+                ("watermark", {"text": "TEST"}),
+                ("convert", {"format": "webp", "quality": 80}),
+            ],
+        )
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        final_out = tmp_path / "src_resize_watermark_convert.webp"
+        assert final_out.exists()
+        with Image.open(final_out) as img:
+            assert img.format == "WEBP"
+
+    def test_execute_with_output_dir(self, tmp_path: Path) -> None:
+        """自定义 output_dir 输出到指定目录."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        out_dir = tmp_path / "output"
+        graph = px.image_pipeline(
+            src,
+            steps=[("resize", {"width": 50})],
+            output_dir=out_dir,
+        )
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        assert (out_dir / "src_resize.png").exists()
+
+    def test_execute_convert_changes_extension(self, tmp_path: Path) -> None:
+        """convert 步骤改变输出扩展名."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        graph = px.image_pipeline(
+            src,
+            steps=[
+                ("convert", {"format": "jpeg"}),
+            ],
+        )
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        out = tmp_path / "src_convert.jpeg"
+        assert out.exists()
+
+    def test_execute_custom_naming(self, tmp_path: Path) -> None:
+        """自定义 naming 模板."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (50, 50))
+        graph = px.image_pipeline(
+            src,
+            steps=[("resize", {"width": 25})],
+            naming="processed_{step}_{stem}{ext}",
+        )
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        out = tmp_path / "processed_resize_src.png"
+        assert out.exists()
+
+    def test_execute_flip_and_rotate(self, tmp_path: Path) -> None:
+        """flip + rotate 组合执行."""
+        src = tmp_path / "src.png"
+        _make_test_image(src, (100, 80))
+        graph = px.image_pipeline(
+            src,
+            steps=[
+                ("flip", {"direction": "horizontal"}),
+                ("rotate", {"degrees": 90, "expand": True}),
+            ],
+        )
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        out = tmp_path / "src_flip_rotate.png"
+        assert out.exists()