improve review-plan preview output and docs

CHANGELOG.md

@@ -1,5 +1,36 @@
 # Changelog
 
+## 2026-04-02
+
+### Schema Validation & Deterministic Planning
+
+- Added runtime validation for `identities.json`, `analysis.json`, and `plan.json` on load/save.
+- Plan generation now records environment-derived checks in `validation_snapshot` metadata, keeping logical plan output reproducible.
+- Added targeted regression tests for plan round-trip validation and invalid-schema rejection.
+- Verified the affected workflows with `uv run pytest tests/test_io.py tests/test_planner.py`.
+
+
+### Logic Optimization & Human-in-the-Loop
+
+- **Human-in-the-Loop Workflow Completion**:
+  - Added `vlm apply-review` command to sync modifications from `plan_manual_review.csv` back to `plan.json`.
+  - Implemented `apply_review_to_plan` in `src/vlm/planner.py` to allow manual rejection/modification of high-risk operations via CSV.
+
+- **Scanner Robustness & Environment Awareness**:
+  - Added proactive `ffprobe` availability check at the start of `vlm scan`.
+  - Implemented graceful degradation: if `ffprobe` is missing, the scanner automatically falls back to file-only metadata mode with a clear warning, instead of failing per-file.
+
+- **Parser Enhancements**:
+  - Improved `src/vlm/parser.py` to better handle release group tags (`[Group]`, `(Group)`, `Group_Subs`).
+  - Added support for hyphenated episode numbering (e.g., `Name - 01`) common in Anime, defaulting to Season 1 with reduced confidence.
+  - Refined release group removal to be more robust against various bracket styles.
+
+- **CLI & UX**:
+  - `vlm review-plan` now prints a plan overview plus a structured high-risk operation preview in terminal output.
+  - Added `--preview-limit` and `--show-all` to control review-plan preview verbosity.
+  - Added shared plan rendering helpers so `review-plan` and `execute` reuse the same summary fallback behavior.
+  - Added `vlm apply-review` to the main CLI group.
+
 ## 2026-02-16
 
 ### Artifact Path Governance

README.md

@@ -2,25 +2,28 @@
 
 ## Documentation Status
 
-- Last synchronized: **2026-02-16**
-- CLI command refactor landed (`parse`, `enrich`, `execute`, `rollback` logic moved to `src/vlm/commands/`).
-- Unified JSON I/O interfaces are available in `src/vlm/io.py`.
-- Full test baseline after refactor: **477 passed**.
+- Last synchronized: **2026-04-02**
+- Human-in-the-loop workflow includes `vlm apply-review` for syncing manual plan edits from CSV back into `plan.json`.
+- Scanner now detects `ffprobe` availability and degrades gracefully.
+- JSON artifacts (`identities.json`, `analysis.json`, `plan.json`) are schema-validated on load/save.
+- Plan generation separates logical intent from environment-derived validation snapshots, improving reproducibility.
 
 A Python-based CLI tool for managing personal video collections with a safety-first, human-in-the-loop approach.
 
 ## Features
 
 - **Safety-First Design**: All file operations are reversible with rollback support
-- **Human-in-the-Loop**: Explicit confirmation required before making any changes
+- **Human-in-the-Loop**: Explicit confirmation required before making any changes. **New: Full review-apply cycle for manual plan adjustments via CSV.**
 - **Comprehensive Analysis**: Detect episode gaps and duplicate files
-- **Rich Metadata**: Extract video resolution, codec, duration, and bitrate
+- **Rich Metadata**: Extract video resolution, codec, duration, and bitrate (with graceful fallback if `ffprobe` is missing)
 - **Flexible Organization**: Customizable directory structure and naming templates
 - **Metadata Enrichment**: Add bilingual titles and reputation signals (TMDB + optional AI fallback)
 - **Incremental Performance**: SQLite-backed cache avoids repeated metadata lookups
 - **State Tracking**: Track file status throughout the workflow
-- **Detailed Reporting**: Generate inventory, completeness, and duplicate reports (reports can include plan content summary via `--plan`)
-- **Plan–Analysis Integration**: `vlm plan --analysis` applies duplicate resolution (keep by reputation, quarantine rest) and adds a Chinese human summary to the plan for quick review
+- **Detailed Reporting**: Generate inventory, completeness, and duplicate reports
+- **Plan–Analysis Integration**: `vlm plan --analysis` applies duplicate resolution and adds a Chinese human summary
+- **Artifact Validation**: JSON artifacts are validated early to catch malformed inputs before later stages run
+- **Deterministic Planning**: Plan generation records live filesystem checks as validation snapshots instead of mixing them into core plan facts
 
 ## Installation
 
@@ -159,7 +162,12 @@ Duplicate keep strategy is configurable in `~/.vlm/config.yaml` under `plan.dupl
 - `first_seen` - Keep the first file in each duplicate group.
 - `manual` - Do not generate quarantine operations; duplicates are listed in analysis only.
 
-**Review the plan** by opening `artifacts/plan.json` in your editor, or read the human summary when you run `vlm execute`. You can edit the plan JSON if needed.
+**Review the plan** in one of three ways:
+- Open `artifacts/plan.json` in your editor
+- Run `vlm review-plan` to get a terminal preview (summary + high-risk operation preview)
+- Run `vlm execute` to see the same plan summary in dry-run mode
+
+You can still edit `plan.json` directly when needed.
 
 ### 7. Execute (Dry-Run First)
 
@@ -220,13 +228,20 @@ vlm analyze
 vlm plan --analysis artifacts/analysis.json
 # Output: artifacts/plan.json with operations, human summary, and duplicate quarantine decisions
 
-# 7. Review the plan
-cat artifacts/plan.json | less
-# or open in your editor
+# 7. Review the plan in terminal (summary + high-risk preview)
+vlm review-plan
+# Optional: control preview size
+vlm review-plan --preview-limit 20
+# Optional: show every high-risk operation in terminal
+vlm review-plan --show-all
+# Edit artifacts/plan_manual_review.csv in Excel/Numbers
+# Sync your manual decisions back to artifacts/plan.json
+vlm apply-review
+# Output: Successfully updated plan saved to: artifacts/plan.json
 
 # 8. Dry-run to preview
 vlm execute
-# Shows what would happen without making changes
+# Shows what would happen without making changes (respecting your manual edits)
 
 # 9. Execute with confirmation
 vlm execute --confirm
@@ -331,6 +346,22 @@ vlm plan --input my_identities.json --output my_plan.json
 vlm plan --input my_identities.json --analysis my_analysis.json --output my_plan.json
 ```
 
+### Manual Plan Review
+
+```bash
+# Export high-risk operations and preview them in terminal
+vlm review-plan
+
+# Preview first N high-risk operations in terminal (default: 10)
+vlm review-plan --preview-limit 20
+
+# Show all high-risk operations in terminal preview
+vlm review-plan --show-all
+
+# Apply edited CSV decisions back to plan.json
+vlm apply-review
+```
+
 ### Execution
 
 ```bash

plans/2026-04-02-review-plan-output-refactor-v1.md

@@ -0,0 +1,102 @@
+# Review-Plan Output Refactor Plan
+
+## Objective
+
+在不修改 `plan.json` schema、不中断现有 CSV 手工审核流程的前提下，重构 `vlm review-plan` 的输出体验，让用户在终端中直接看到计划的核心内容和可审核的操作预览，减少必须打开 `plan.json` 才能继续操作的成本。
+
+## Validated Baseline
+
+- `review-plan` 当前只输出计划加载提示、风险统计、CSV 保存路径和前 5 条样例，没有输出完整计划预览，见 `src/vlm/cli.py:536-563`。
+- `ExecutionPlan` 已包含 `operations`、`summary`、`summary_by_reason`、`human_summary`、`metadata`，足以支撑更强的终端展示，见 `src/vlm/models.py:133-153`。
+- 计划保存时会把上述字段全部写入 JSON，因此无需改动 plan schema，见 `src/vlm/planner.py:591-625`。
+- CLI 中已经存在 fallback 计划摘要逻辑，可复用于 `review-plan`，见 `src/vlm/cli.py:976-987`。
+- `execute` 已经采用“计划概要 + 样例操作”的输出方式，可作为统一风格参考，见 `src/vlm/commands/execute.py:84-103` 和 `src/vlm/commands/execute.py:137-146`。
+- 当前测试只覆盖 summary 和 CSV 导出，未覆盖完整计划预览输出，见 `tests/test_cli_review_plan.py:31-88`。
+
+## Recommended Approach
+
+采用推荐方案：**抽离通用渲染层，并为 `review-plan` 提供受控预览输出**。
+
+原因：
+
+1. 只加 `human_summary` 无法解决“看不到计划内容”的核心问题。
+2. 直接打印全部 operations 会在大计划场景下严重刷屏。
+3. 抽离通用渲染层可以同时提升用户体验、结构清晰度和后续复用性。
+
+## Scope
+
+### In Scope
+
+- 优化 `review-plan` 的终端输出结构。
+- 复用已有 `human_summary` / fallback summary。
+- 增加受控的操作预览输出。
+- 为计划展示提取可复用 helper。
+- 补充 CLI 测试，覆盖新增展示行为。
+
+### Out of Scope
+
+- 修改 `ExecutionPlan` 数据模型。
+- 修改 `plan.json` schema。
+- 修改 review CSV 字段或 `apply-review` 工作流。
+- 引入交互式 TUI/Web 界面。
+
+## Implementation Plan
+
+- [x] Task 1. [Status: Done] 重新定义 `review-plan` 的输出顺序为“计划概览 → 风险统计 → 操作预览 → CSV 路径”，优先展示决策信息，再展示审核细节，以替代当前仅有 summary 和 5 条样例的输出方式，现状见 `src/vlm/cli.py:546-563`。
+- [x] Task 2. [Status: Done] 在 `review-plan` 中优先输出 `ExecutionPlan.human_summary`，若为空则复用现有 fallback summary，避免重复设计摘要逻辑并统一跨命令体验，相关能力见 `src/vlm/planner.py:169-184`、`src/vlm/cli.py:976-987`。
+- [x] Task 3. [Status: Done] 提取统一的计划终端渲染 helper，负责 plan header、summary、reason 分布和 operation preview 的格式化输出，避免 CLI 命令函数继续承载大量展示细节，参考现有输出风格见 `src/vlm/commands/execute.py:84-103`。
+- [x] Task 4. [Status: Done] 设计受控预览机制，默认仅展示有限条操作并提示剩余数量，同时预留完整显示模式的扩展点，以兼顾可读性和信息完整性。
+- [x] Task 5. [Status: Done] 在预览输出中优先展示审核价值最高的字段，包括 `index`、`operation_type`、`risk_flags`、`source_path`、`destination_path`、`reason`，以便用户在不打开 JSON 的情况下完成多数审核判断，字段来源见 `src/vlm/plan_review.py:75-83` 和 `src/vlm/planner.py:606-620`。
+- [x] Task 6. [Status: Done] 保持 `ExecutionPlan` 模型、plan JSON schema 和 review CSV schema 完全兼容，将改动严格限制在输出层，降低对 `execute`、`report` 和 `apply-review` 的影响，相关结构见 `src/vlm/models.py:133-153` 和 `src/vlm/planner.py:591-625`。
+- [x] Task 7. [Status: Done] 扩展 `review-plan` CLI 测试，覆盖默认摘要输出、受控预览、完整显示模式、CSV 不变性和原有 summary 输出兼容性，弥补当前测试缺口，基线见 `tests/test_cli_review_plan.py:31-88`。
+- [x] Task 8. [Status: Done] 评估是否将 `execute` 的计划摘要展示逐步迁移到同一渲染 helper，减少跨命令输出风格分叉，参考现有入口见 `src/vlm/commands/execute.py:84-103` 和 `src/vlm/commands/execute.py:137-146`。
+- [x] Task 9. [Status: Done] 在最终验收中重点验证大计划场景下的可读性，确保默认输出足够简洁、重点清晰，并且不影响后续 `apply-review` 使用链路，相关流程见 `src/vlm/cli.py:589-643`。
+
+## Verification Criteria
+
+- [x] `vlm review-plan` 默认输出中包含计划摘要，而不只是风险计数。
+- [x] 默认输出中包含可读的操作预览，且预览字段足以支持人工初步审核。
+- [x] 大计划场景下默认输出不会无上限刷屏，并会提示仍有未展示操作。
+- [x] review CSV 的字段、写入逻辑与后续 `apply-review` 流程保持兼容，相关链路见 `src/vlm/plan_review.py:89-96` 和 `src/vlm/cli.py:621-643`。
+- [x] `plan.json` 的 schema、读写行为与现有字段保持不变，见 `src/vlm/planner.py:591-625` 和 `src/vlm/planner.py:628-671`。
+- [x] CLI 测试覆盖新增预览行为，并保留现有 summary/CSV 行为验证，基线见 `tests/test_cli_review_plan.py:31-88`。
+
+## Risks and Mitigations
+
+1. **默认输出过长，降低可读性**  
+   Mitigation: 使用默认限量预览，只展示高价值字段，并明确提示剩余条目数量。
+
+2. **展示逻辑分散，后续难维护**  
+   Mitigation: 将计划渲染抽离为统一 helper，让 CLI 命令函数只负责流程编排与参数处理。
+
+3. **改动误伤 CSV 手工审核链路**  
+   Mitigation: 将 CSV 视为稳定接口，不调整字段结构与导出逻辑，保持 `src/vlm/plan_review.py:89-96` 行为不变。
+
+4. **CLI 输出测试过于脆弱**  
+   Mitigation: 测试聚焦结构性关键片段与核心字段，不对整段输出做过度刚性匹配。
+
+## Alternatives Considered
+
+1. **仅增加 `human_summary` 输出**  
+   优点：改动最小，交付最快。  
+   缺点：仍然看不到操作层内容。
+
+2. **直接打印全部 operations**  
+   优点：实现简单，信息最完整。  
+   缺点：大计划会严重刷屏。
+
+3. **抽离通用渲染层并提供受控预览**  
+   优点：用户体验、可维护性与复用性最平衡。  
+   缺点：实现成本略高于局部修补。  
+   结论：**推荐采用**。
+
+## Recommended Outcome
+
+推荐采用“仅重构展示层、不改数据层”的方案：
+
+- 保持 `ExecutionPlan`、plan JSON、review CSV 全部兼容。
+- 为 `review-plan` 增加计划摘要与受控操作预览。
+- 把计划展示逻辑抽离为可复用渲染能力。
+- 用测试确保 CLI 可见行为稳定。
+
+这样可以以最小风险解决当前“review 时看不到计划内容”的核心问题。

src/vlm/cli.py

@@ -17,6 +17,7 @@ from click.core import ParameterSource
 from vlm.config import Config, load_config, create_default_config, validate_config
 from vlm.context import CLIContext, pass_context
 from vlm.logging_config import setup_logging, get_logger
+from vlm.plan_render import fallback_plan_summary, preferred_plan_summary, render_review_preview
 from vlm.utils import format_size
 
 
@@ -76,6 +77,53 @@ def resolve_legacy_default_input_path(
     return legacy_default
 
 
+def _command_error(ctx: CLIContext, user_message: str, logger_message: str, *, exc_info: bool = False) -> None:
+    """Print a command error message, log it, and exit consistently."""
+    click.echo(user_message, err=True)
+    ctx.logger.error(logger_message, exc_info=exc_info)
+    raise SystemExit(1)
+
+
+def _load_or_create_config(config: Path) -> Config:
+    """Load configuration from disk or create a default config file."""
+    if config.exists():
+        try:
+            cfg = load_config(config)
+        except yaml.YAMLError as e:
+            click.echo(f"Error: Invalid YAML syntax in configuration file: {e}", err=True)
+            click.echo("Using default configuration values.", err=True)
+            cfg = create_default_config(config)
+        except ValueError as e:
+            click.echo(f"Error: {e}", err=True)
+            click.echo("Using default configuration values.", err=True)
+            cfg = create_default_config(config)
+    else:
+        click.echo(f"Configuration file not found at {config}", err=True)
+        click.echo("Creating default configuration...", err=True)
+        cfg = create_default_config(config)
+        click.echo(f"Default configuration created at {config}", err=True)
+
+    validation_errors = validate_config(cfg)
+    if validation_errors:
+        click.echo("Configuration validation errors:", err=True)
+        for error in validation_errors:
+            click.echo(f"  - {error}", err=True)
+        click.echo("Please fix the configuration file and try again.", err=True)
+        raise ValueError("invalid configuration")
+
+    return cfg
+
+
+def _initialize_cli_context(config: Path, log_level: Optional[str]) -> CLIContext:
+    """Load config, apply CLI overrides, and build the CLI context."""
+    cfg = _load_or_create_config(config)
+    if log_level:
+        cfg.log_level = log_level.upper()
+
+    logger = setup_logging(log_level=cfg.log_level)
+    return CLIContext(config=cfg, logger=logger)
+
+
 @click.group()
 @click.option(
     '--config',
@@ -101,10 +149,11 @@ def main(ctx, config: Path, log_level: Optional[str]):
     
       1. vlm scan              - Discover all video files
       2. vlm parse             - Extract titles, years, seasons, episodes
-      3. vlm analyze           - Detect gaps and duplicates
-      4. vlm plan              - Generate execution plan
-      5. vlm execute           - Execute plan (dry-run by default)
-      6. vlm execute --confirm - Actually execute operations
+      3. vlm enrich            - Enrich parsed identities with external metadata
+      4. vlm analyze           - Detect gaps and duplicates
+      5. vlm plan              - Generate execution plan
+      6. vlm execute           - Execute plan (dry-run by default)
+      7. vlm execute --confirm - Actually execute operations
     
     Use 'vlm COMMAND --help' for more information on a specific command.
     """
@@ -112,43 +161,7 @@ def main(ctx, config: Path, log_level: Optional[str]):
     ctx.ensure_object(dict)
     
     try:
-        # Load or create configuration
-        if config.exists():
-            try:
-                cfg = load_config(config)
-            except yaml.YAMLError as e:
-                click.echo(f"Error: Invalid YAML syntax in configuration file: {e}", err=True)
-                click.echo("Using default configuration values.", err=True)
-                cfg = create_default_config(config)
-            except ValueError as e:
-                click.echo(f"Error: {e}", err=True)
-                click.echo("Using default configuration values.", err=True)
-                cfg = create_default_config(config)
-        else:
-            click.echo(f"Configuration file not found at {config}", err=True)
-            click.echo("Creating default configuration...", err=True)
-            cfg = create_default_config(config)
-            click.echo(f"Default configuration created at {config}", err=True)
-        
-        # Validate configuration
-        validation_errors = validate_config(cfg)
-        if validation_errors:
-            click.echo("Configuration validation errors:", err=True)
-            for error in validation_errors:
-                click.echo(f"  - {error}", err=True)
-            click.echo("Please fix the configuration file and try again.", err=True)
-            sys.exit(1)
-        
-        # Override log level if specified on command line
-        if log_level:
-            cfg.log_level = log_level.upper()
-        
-        # Set up logging
-        logger = setup_logging(log_level=cfg.log_level)
-        
-        # Store context for subcommands
-        ctx.obj = CLIContext(config=cfg, logger=logger)
-        
+        ctx.obj = _initialize_cli_context(config, log_level)
     except Exception as e:
         traceback.print_exc(file=sys.stderr)
         click.echo(f"Error initializing VLM: {e}", err=True)
@@ -250,17 +263,20 @@ def parse(ctx: CLIContext, input: Path, output: Path, inventory: Optional[Path])
         from vlm.commands.parse import parse_cmd
         parse_cmd(ctx, input, output, inventory)
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        ctx.logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Input file not found: {input}",
+            f"Input file not found: {input}",
+        )
     except ValueError as e:
-        click.echo(f"Error: {e}", err=True)
-        ctx.logger.error(f"Parse failed: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Parse failed: {e}")
     except OSError as e:
-        click.echo(f"Error reading/writing files: {e}", err=True)
-        ctx.logger.error(f"Parse file I/O failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error reading/writing files: {e}",
+            f"Parse file I/O failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.command()
@@ -330,21 +346,27 @@ def enrich(
             retries,
         )
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        ctx.logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Input file not found: {input}",
+            f"Input file not found: {input}",
+        )
     except json.JSONDecodeError as e:
-        click.echo(f"Error: Failed to parse JSON file: {e}", err=True)
-        ctx.logger.error(f"JSON parsing failed during enrich: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Failed to parse JSON file: {e}",
+            f"JSON parsing failed during enrich: {e}",
+            exc_info=True,
+        )
     except ValueError as e:
-        click.echo(f"Error: {e}", err=True)
-        ctx.logger.error(f"Enrich validation failed: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Enrich validation failed: {e}")
     except OSError as e:
-        click.echo(f"Error reading/writing files: {e}", err=True)
-        ctx.logger.error(f"Enrich file I/O failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error reading/writing files: {e}",
+            f"Enrich file I/O failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.command()
@@ -385,17 +407,25 @@ def analyze(ctx: CLIContext, input: Path, output: Path, inventory: Optional[Path
         from vlm.commands.analyze import analyze_cmd
         analyze_cmd(ctx, input, output, inventory)
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        ctx.logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Input file not found: {input}",
+            f"Input file not found: {input}",
+        )
     except json.JSONDecodeError as e:
-        click.echo(f"Error: Failed to parse JSON file: {e}", err=True)
-        ctx.logger.error(f"JSON parsing failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Failed to parse JSON file: {e}",
+            f"JSON parsing failed: {e}",
+            exc_info=True,
+        )
     except Exception as e:
-        click.echo(f"Error during analysis: {e}", err=True)
-        ctx.logger.error(f"Analysis failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error during analysis: {e}",
+            f"Analysis failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.command()
@@ -436,17 +466,25 @@ def plan(ctx: CLIContext, input: Path, output: Path, analysis: Path | None):
         from vlm.commands.plan import plan_cmd
         plan_cmd(ctx, input, output, analysis)
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        ctx.logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Input file not found: {input}",
+            f"Input file not found: {input}",
+        )
     except json.JSONDecodeError as e:
-        click.echo(f"Error: Failed to parse JSON file: {e}", err=True)
-        ctx.logger.error(f"JSON parsing failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Failed to parse JSON file: {e}",
+            f"JSON parsing failed: {e}",
+            exc_info=True,
+        )
     except Exception as e:
-        click.echo(f"Error during plan generation: {e}", err=True)
-        ctx.logger.error(f"Plan generation failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error during plan generation: {e}",
+            f"Plan generation failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.command(name="review-plan")
@@ -476,6 +514,19 @@ def plan(ctx: CLIContext, input: Path, output: Path, analysis: Path | None):
     show_default=True,
     help='Flag operations with episode >= this value as high risk'
 )
+@click.option(
+    '--preview-limit',
+    type=int,
+    default=10,
+    show_default=True,
+    help='Number of high-risk operations to preview in console output'
+)
+@click.option(
+    '--show-all',
+    is_flag=True,
+    default=False,
+    help='Show all high-risk operations in console preview'
+)
 @pass_context
 def review_plan_cmd(
     ctx: CLIContext,
@@ -483,6 +534,8 @@ def review_plan_cmd(
     output: Path,
     season_threshold: int,
     episode_threshold: int,
+    preview_limit: int,
+    show_all: bool,
 ):
     """Review a plan and export high-risk operations for manual confirmation."""
     from vlm.planner import load_plan
@@ -495,6 +548,9 @@ def review_plan_cmd(
         if season_threshold < 1 or episode_threshold < 1:
             click.echo("Error: thresholds must be >= 1", err=True)
             sys.exit(1)
+        if preview_limit < 1:
+            click.echo("Error: --preview-limit must be >= 1", err=True)
+            sys.exit(1)
 
         click.echo(f"Loading plan: {input}")
         execution_plan = load_plan(input)
@@ -506,6 +562,10 @@ def review_plan_cmd(
 
         save_review_csv(rows, output)
 
+        click.echo()
+        click.echo("Plan overview:")
+        click.echo(preferred_plan_summary(execution_plan))
+
         click.echo()
         click.echo("Plan review summary:")
         click.echo(f"  Total operations: {counters['total_operations']}")
@@ -515,16 +575,28 @@ def review_plan_cmd(
         click.echo(f"  high_season: {counters['high_season']}")
         click.echo(f"  high_episode: {counters['high_episode']}")
         click.echo(f"  conflicts: {counters['conflicts']}")
+
+        click.echo()
+        click.echo("High-risk operations preview:")
+        if rows:
+            preview_lines, hidden_count = render_review_preview(
+                rows,
+                preview_limit=preview_limit,
+                show_all=show_all,
+            )
+            for line in preview_lines:
+                click.echo(line)
+            if hidden_count > 0:
+                click.echo(
+                    f"  ... and {hidden_count} more high-risk operations "
+                    "(use --show-all to display all)"
+                )
+        else:
+            click.echo("  (none)")
+
         click.echo()
         click.echo(f"Saved manual review CSV to: {output}")
 
-        if rows:
-            click.echo("Top review samples:")
-            for row in rows[:5]:
-                click.echo(
-                    f"  - [{row['index']}] {row['operation_type']} {Path(row['source_path']).name} ({row['risk_flags']})"
-                )
-
         logger.info(
             "Plan review completed: total=%s high_risk=%s output=%s",
             counters["total_operations"],
@@ -532,17 +604,86 @@ def review_plan_cmd(
             output,
         )
     except FileNotFoundError:
-        click.echo(f"Error: Plan file not found: {input}", err=True)
-        logger.error(f"Plan file not found: {input}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: Plan file not found: {input}", f"Plan file not found: {input}")
     except json.JSONDecodeError as e:
-        click.echo(f"Error: Failed to parse plan JSON: {e}", err=True)
-        logger.error(f"Plan review JSON parsing failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Failed to parse plan JSON: {e}",
+            f"Plan review JSON parsing failed: {e}",
+            exc_info=True,
+        )
     except Exception as e:
-        click.echo(f"Error during plan review: {e}", err=True)
-        logger.error(f"Plan review failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error during plan review: {e}",
+            f"Plan review failed: {e}",
+            exc_info=True,
+        )
+
+
+@main.command(name="apply-review")
+@click.option(
+    '--plan',
+    type=click.Path(exists=True, path_type=Path),
+    default=Path('plan.json'),
+    help='Path to execution plan JSON file (default: plan.json)'
+)
+@click.option(
+    '--csv',
+    type=click.Path(exists=True, path_type=Path),
+    default=Path('plan_manual_review.csv'),
+    help='Path to the modified manual review CSV (default: plan_manual_review.csv)'
+)
+@click.option(
+    '--output',
+    type=click.Path(path_type=Path),
+    default=None,
+    help='Path to save updated plan (default: overwrite input plan)'
+)
+@pass_context
+def apply_review_cmd(
+    ctx: CLIContext,
+    plan: Path,
+    csv: Path,
+    output: Optional[Path],
+):
+    """Apply modifications from a manual review CSV back to the plan JSON.
+    
+    This command reads the 'operation_type' column from the CSV and updates
+    the corresponding operations in the plan. This is the primary way to
+    manually approve or reject high-risk operations.
+    """
+    from vlm.planner import load_plan, save_plan, apply_review_to_plan
+    
+    logger = ctx.logger
+    output_path = output or plan
+    
+    try:
+        click.echo(f"Loading plan: {plan}")
+        execution_plan = load_plan(plan)
+        
+        click.echo(f"Applying review from: {csv}")
+        updated_plan = apply_review_to_plan(execution_plan, csv)
+        
+        save_plan(updated_plan, output_path)
+        click.echo(f"Successfully updated plan saved to: {output_path}")
+        
+        # Calculate changes for user feedback
+        modified = 0
+        for i, op in enumerate(updated_plan.operations):
+            if op.operation_type != execution_plan.operations[i].operation_type:
+                modified += 1
+        
+        click.echo(f"Total operations modified: {modified}")
+        logger.info("Applied review from %s to %s, modified %s ops", csv, output_path, modified)
+        
+    except Exception as e:
+        _command_error(
+            ctx,
+            f"Error applying review: {e}",
+            f"Apply review failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.command()
@@ -602,17 +743,20 @@ def execute(ctx: CLIContext, plan: Path, confirm: bool, yes: bool, verbose_ops:
         from vlm.commands.execute import execute_cmd
         execute_cmd(ctx, plan, confirm, yes, verbose_ops, preserve_directories, safe_mode)
     except FileNotFoundError:
-        click.echo(f"Error: File not found: {plan}", err=True)
-        ctx.logger.error(f"Execution file not found: {plan}")
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: File not found: {plan}",
+            f"Execution file not found: {plan}",
+        )
     except ValueError as e:
-        click.echo(f"Error: {e}", err=True)
-        ctx.logger.error(f"Execution validation failed: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Execution validation failed: {e}")
     except OSError as e:
-        click.echo(f"Error during execution: {e}", err=True)
-        ctx.logger.error(f"Execution I/O failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error during execution: {e}",
+            f"Execution I/O failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.group()
@@ -692,9 +836,12 @@ def quarantine_list(ctx: CLIContext, category: Optional[str]):
                    (f" from category '{category}'" if category else ""))
         
     except Exception as e:
-        click.echo(f"Error listing quarantined files: {e}", err=True)
-        logger.error(f"Failed to list quarantined files: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error listing quarantined files: {e}",
+            f"Failed to list quarantined files: {e}",
+            exc_info=True,
+        )
 
 
 @quarantine.command('add')
@@ -744,21 +891,25 @@ def quarantine_add(ctx: CLIContext, file: Path, reason: Optional[str]):
             click.echo("To restore this file, run:")
             click.echo(f"  vlm quarantine restore {result.operation.destination_path}")
         else:
-            click.echo(f"✗ Failed to quarantine file: {result.error_message}", err=True)
-            sys.exit(1)
-        
+            _command_error(
+                ctx,
+                f"✗ Failed to quarantine file: {result.error_message}",
+                f"Quarantine failed for {file}: {result.error_message}",
+            )
+
         logger.info(f"Quarantined file: {file}")
         
     except ValueError as e:
         # Category restriction error
-        click.echo(f"Error: {e}", err=True)
-        logger.error(f"Quarantine rejected: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Quarantine rejected: {e}")
     
     except Exception as e:
-        click.echo(f"Error quarantining file: {e}", err=True)
-        logger.error(f"Failed to quarantine file: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error quarantining file: {e}",
+            f"Failed to quarantine file: {e}",
+            exc_info=True,
+        )
 
 
 @quarantine.command('restore')
@@ -800,14 +951,21 @@ def quarantine_restore(ctx: CLIContext, file: Path):
                 click.echo(f"  Original location: {result.operation.destination_path}", err=True)
             else:
                 click.echo(f"✗ Failed to restore file: {result.error_message}", err=True)
-            sys.exit(1)
+            _command_error(
+                ctx,
+                f"Error restoring file: {result.error_message or result.operation.conflict_reason or 'unknown error'}",
+                f"Failed to restore file: {file}",
+            )
         
         logger.info(f"Restored file from quarantine: {file}")
         
     except Exception as e:
-        click.echo(f"Error restoring file: {e}", err=True)
-        logger.error(f"Failed to restore file: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error restoring file: {e}",
+            f"Failed to restore file: {e}",
+            exc_info=True,
+        )
 
 
 @main.command()
@@ -838,31 +996,21 @@ def rollback(ctx: CLIContext, log: Optional[Path]):
         from vlm.commands.execute import rollback_cmd
         rollback_cmd(ctx, log)
     except FileNotFoundError as e:
-        click.echo(f"Error: {e}", err=True)
-        ctx.logger.error(f"Rollback log not found: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Rollback log not found: {e}")
     except ValueError as e:
-        click.echo(f"Error: {e}", err=True)
-        ctx.logger.error(f"Rollback failed: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Rollback failed: {e}")
     except OSError as e:
-        click.echo(f"Error during rollback: {e}", err=True)
-        ctx.logger.error(f"Rollback failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error during rollback: {e}",
+            f"Rollback failed: {e}",
+            exc_info=True,
+        )
 
 
 def _fallback_plan_summary(execution_plan) -> str:
     """Build a short plan summary from summary and summary_by_reason when human_summary is empty."""
-    s = execution_plan.summary or {}
-    by_r = execution_plan.summary_by_reason or {}
-    total = s.get("total", len(execution_plan.operations))
-    parts = [
-        f"计划操作统计：共 {total} 条（move {s.get('move', 0)}，rename {s.get('rename', 0)}，"
-        f"quarantine {s.get('quarantine', 0)}，no-op {s.get('no-op', 0)}）"
-    ]
-    if by_r:
-        parts.append("原因分布：" + "；".join(f"{r}: {c}" for r, c in list(by_r.items())[:8]))
-    return "\n".join(parts)
+    return fallback_plan_summary(execution_plan)
 
 
 @main.group()
@@ -945,14 +1093,15 @@ def report_inventory(ctx: CLIContext, format: str, input: Path, output: Optional
         logger.info(f"Generated inventory report in {format} format with {len(video_files)} files")
         
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}")
     
     except Exception as e:
-        click.echo(f"Error generating inventory report: {e}", err=True)
-        logger.error(f"Inventory report generation failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error generating inventory report: {e}",
+            f"Inventory report generation failed: {e}",
+            exc_info=True,
+        )
 
 
 @report.command('completeness')
@@ -1053,19 +1202,23 @@ def report_completeness(ctx: CLIContext, format: str, input: Path, output: Optio
         logger.info(f"Generated completeness report in {format} format with {len(season_completeness)} series")
         
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}")
     
     except json.JSONDecodeError as e:
-        click.echo(f"Error: Failed to parse JSON file: {e}", err=True)
-        logger.error(f"JSON parsing failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Failed to parse JSON file: {e}",
+            f"JSON parsing failed: {e}",
+            exc_info=True,
+        )
     
     except Exception as e:
-        click.echo(f"Error generating completeness report: {e}", err=True)
-        logger.error(f"Completeness report generation failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error generating completeness report: {e}",
+            f"Completeness report generation failed: {e}",
+            exc_info=True,
+        )
 
 
 @report.command('duplicates')
@@ -1208,19 +1361,23 @@ def report_duplicates(ctx: CLIContext, format: str, input: Path, output: Optiona
         logger.info(f"Generated duplicate report in {format} format with {len(duplicate_groups)} groups")
         
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}")
     
     except json.JSONDecodeError as e:
-        click.echo(f"Error: Failed to parse JSON file: {e}", err=True)
-        logger.error(f"JSON parsing failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error: Failed to parse JSON file: {e}",
+            f"JSON parsing failed: {e}",
+            exc_info=True,
+        )
     
     except Exception as e:
-        click.echo(f"Error generating duplicate report: {e}", err=True)
-        logger.error(f"Duplicate report generation failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error generating duplicate report: {e}",
+            f"Duplicate report generation failed: {e}",
+            exc_info=True,
+        )
 
 
 @report.command('summary')
@@ -1283,14 +1440,15 @@ def report_summary(ctx: CLIContext, input: Path, output: Optional[Path]):
         logger.info(f"Generated summary report with {len(video_files)} files")
         
     except FileNotFoundError:
-        click.echo(f"Error: Input file not found: {input}", err=True)
-        logger.error(f"Input file not found: {input}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}")
     
     except Exception as e:
-        click.echo(f"Error generating summary report: {e}", err=True)
-        logger.error(f"Summary report generation failed: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error generating summary report: {e}",
+            f"Summary report generation failed: {e}",
+            exc_info=True,
+        )
 
 
 @main.group()
@@ -1344,9 +1502,12 @@ def state_show(ctx: CLIContext, file: Path):
         logger.info(f"Showed state for file: {file}")
         
     except Exception as e:
-        click.echo(f"Error showing file state: {e}", err=True)
-        logger.error(f"Failed to show file state: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error showing file state: {e}",
+            f"Failed to show file state: {e}",
+            exc_info=True,
+        )
 
 
 @state.command('set')
@@ -1405,14 +1566,15 @@ def state_set(ctx: CLIContext, file: Path, status: str, reason: Optional[str]):
         logger.info(f"Set state for file {file}: status={status}, reason={reason}")
         
     except ValueError as e:
-        click.echo(f"Error: {e}", err=True)
-        logger.error(f"Invalid status: {e}")
-        sys.exit(1)
+        _command_error(ctx, f"Error: {e}", f"Invalid status: {e}")
     
     except Exception as e:
-        click.echo(f"Error setting file state: {e}", err=True)
-        logger.error(f"Failed to set file state: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error setting file state: {e}",
+            f"Failed to set file state: {e}",
+            exc_info=True,
+        )
 
 
 @state.command('query')
@@ -1470,9 +1632,12 @@ def state_query(ctx: CLIContext, status: str):
         logger.info(f"Queried files with status '{status}': {len(file_states)} found")
         
     except Exception as e:
-        click.echo(f"Error querying file states: {e}", err=True)
-        logger.error(f"Failed to query file states: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error querying file states: {e}",
+            f"Failed to query file states: {e}",
+            exc_info=True,
+        )
 
 
 @state.command('clear')
@@ -1519,9 +1684,12 @@ def state_clear(ctx: CLIContext, file: Path):
         logger.info(f"Cleared state for file: {file}")
         
     except Exception as e:
-        click.echo(f"Error clearing file state: {e}", err=True)
-        logger.error(f"Failed to clear file state: {e}", exc_info=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error clearing file state: {e}",
+            f"Failed to clear file state: {e}",
+            exc_info=True,
+        )
 
 
 @main.group(name='config')
@@ -1556,8 +1724,12 @@ def config_init(ctx: CLIContext, path: Path):
         click.echo("Edit this file to customize your settings.")
         
     except Exception as e:
-        click.echo(f"Error creating configuration: {e}", err=True)
-        sys.exit(1)
+        _command_error(
+            ctx,
+            f"Error creating configuration: {e}",
+            f"Configuration creation failed: {e}",
+            exc_info=True,
+        )
 
 
 @config_cmd.command('show')
@@ -1590,7 +1762,7 @@ def config_validate(ctx: CLIContext):
         click.echo("Configuration validation errors:", err=True)
         for error in errors:
             click.echo(f"  - {error}", err=True)
-        sys.exit(1)
+        _command_error(ctx, "Configuration validation failed.", "Configuration validation failed")
 
 
 if __name__ == '__main__':

src/vlm/commands/execute.py

@@ -10,6 +10,7 @@ import click
 from vlm.context import CLIContext
 from vlm.executor import ExecutionEngine
 from vlm.planner import load_plan
+from vlm.plan_render import preferred_plan_summary
 from vlm.state import StateManager
 
 
@@ -58,7 +59,10 @@ def execute_cmd(
 
     if safe_mode:
         click.echo("Safe mode enabled - validating plan for directory preservation...")
-        emptied_dirs = execution_plan.metadata.get("emptied_directories", [])
+        validation_snapshot = execution_plan.metadata.get("validation_snapshot", {})
+        if not isinstance(validation_snapshot, dict):
+            validation_snapshot = {}
+        emptied_dirs = validation_snapshot.get("emptied_directories", execution_plan.metadata.get("emptied_directories", []))
         if emptied_dirs:
             raise ValueError(
                 "SAFE MODE VIOLATION: plan would empty directories. "
@@ -82,22 +86,10 @@ def execute_cmd(
     click.echo(f"Created at: {execution_plan.created_at}")
     click.echo(f"Total operations: {len(execution_plan.operations)}")
 
-    if execution_plan.human_summary:
+    plan_summary = preferred_plan_summary(execution_plan)
+    if plan_summary:
         click.echo()
-        click.echo(execution_plan.human_summary)
-    elif execution_plan.summary or execution_plan.summary_by_reason:
-        s = execution_plan.summary or {}
-        by_r = execution_plan.summary_by_reason or {}
-        parts = [
-            "操作统计：共 "
-            f"{s.get('total', len(execution_plan.operations))} 条"
-            f"（move {s.get('move', 0)}，rename {s.get('rename', 0)}，"
-            f"quarantine {s.get('quarantine', 0)}，no-op {s.get('no-op', 0)}）"
-        ]
-        if by_r:
-            parts.append("原因分布：" + "；".join(f"{r}: {c}" for r, c in list(by_r.items())[:5]))
-        click.echo()
-        click.echo("\n".join(parts))
+        click.echo(plan_summary)
     click.echo()
 
     if mode == "dry-run":

src/vlm/plan_render.py

@@ -0,0 +1,66 @@
+"""Utilities for rendering execution plan information in CLI output."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def fallback_plan_summary(execution_plan: Any) -> str:
+    """Build a short plan summary from summary and summary_by_reason.
+
+    This is used when execution_plan.human_summary is missing or empty.
+    """
+    summary = getattr(execution_plan, "summary", {}) or {}
+    summary_by_reason = getattr(execution_plan, "summary_by_reason", {}) or {}
+    operations = getattr(execution_plan, "operations", []) or []
+
+    total = summary.get("total", len(operations))
+    parts = [
+        f"计划操作统计：共 {total} 条（move {summary.get('move', 0)}，rename {summary.get('rename', 0)}，"
+        f"quarantine {summary.get('quarantine', 0)}，no-op {summary.get('no-op', 0)}）"
+    ]
+    if summary_by_reason:
+        parts.append(
+            "原因分布：" + "；".join(f"{reason}: {count}" for reason, count in list(summary_by_reason.items())[:8])
+        )
+    return "\n".join(parts)
+
+
+def preferred_plan_summary(execution_plan: Any) -> str:
+    """Return human_summary if available, otherwise fallback summary."""
+    human_summary = getattr(execution_plan, "human_summary", "")
+    if isinstance(human_summary, str) and human_summary.strip():
+        return human_summary
+    return fallback_plan_summary(execution_plan)
+
+
+def render_review_preview(
+    rows: list[dict[str, str]],
+    preview_limit: int = 10,
+    show_all: bool = False,
+) -> tuple[list[str], int]:
+    """Render high-risk review rows for console preview.
+
+    Returns a tuple of (lines, remaining_count).
+    """
+    if preview_limit < 1:
+        raise ValueError("preview_limit must be >= 1")
+
+    selected = rows if show_all else rows[:preview_limit]
+    remaining = 0 if show_all else max(0, len(rows) - len(selected))
+
+    lines: list[str] = []
+    for row in selected:
+        idx = row.get("index", "?")
+        operation_type = row.get("operation_type", "")
+        flags = row.get("risk_flags", "") or "none"
+        source_path = row.get("source_path", "")
+        destination_path = row.get("destination_path", "")
+        reason = row.get("reason", "")
+
+        lines.append(f"  - [{idx}] {operation_type} | flags={flags}")
+        lines.append(f"      source: {source_path}")
+        lines.append(f"      destination: {destination_path or '(none)'}")
+        lines.append(f"      reason: {reason}")
+
+    return lines, remaining

tests/test_cli_review_plan.py

@@ -24,61 +24,86 @@ def _write_config(path: Path, library_root: Path) -> None:
                 'quarantine_dir: ".quarantine"',
                 'log_level: "INFO"',
             ]
-        )
+        ),
+        encoding="utf-8",
     )
 
 
-def test_review_plan_generates_csv_and_summary(tmp_path):
-    """review-plan should export flagged operations and summary counters."""
+def _write_plan(path: Path, operations: list[dict], summary: dict, human_summary: str = "") -> None:
+    plan_data = {
+        "vlm_schema_version": "1.0",
+        "plan_id": "test-plan",
+        "created_at": "2026-02-13T00:00:00+00:00",
+        "operations": operations,
+        "summary": summary,
+        "summary_by_reason": {},
+        "human_summary": human_summary,
+        "metadata": {},
+    }
+    path.write_text(json.dumps(plan_data), encoding="utf-8")
+
+
+def _invoke_review_plan(runner: CliRunner, config_path: Path, args: list[str]):
+    return runner.invoke(main, ["--config", str(config_path), "review-plan", *args])
+
+
+def test_review_plan_generates_csv_summary_and_preview(tmp_path):
+    """review-plan should export CSV, show summary, and print preview content."""
     config_path = tmp_path / "config.yaml"
     _write_config(config_path, tmp_path / "library")
 
     plan_path = tmp_path / "plan.json"
-    plan_data = {
-        "plan_id": "test-plan",
-        "created_at": "2026-02-13T00:00:00+00:00",
-        "operations": [
-            {
-                "operation_type": "move",
-                "source_path": str(tmp_path / "Show.Sample.S01E01.mkv"),
-                "destination_path": str(tmp_path / "library/series/Show/Season 01/S01E01.mkv"),
-                "reason": "Organize series: Show S01E01",
-                "has_conflict": False,
-                "conflict_reason": None,
-            },
-            {
-                "operation_type": "no-op",
-                "source_path": str(tmp_path / "Show.S20E50.mkv"),
-                "destination_path": None,
-                "reason": "Series needs manual review (season exceeds configured threshold)",
-                "has_conflict": False,
-                "conflict_reason": None,
-            },
-        ],
-        "summary": {"total": 2, "move": 1, "rename": 0, "quarantine": 0, "no-op": 1},
-        "summary_by_reason": {},
-        "human_summary": "",
-        "metadata": {},
-    }
-    plan_path.write_text(json.dumps(plan_data), encoding="utf-8")
+    operations = [
+        {
+            "operation_type": "move",
+            "source_path": str(tmp_path / "Show.Sample.S01E01.mkv"),
+            "destination_path": str(tmp_path / "library/series/Show/Season 01/S01E01.mkv"),
+            "reason": "Organize series: Show S01E01",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Show.S20E50.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+    ]
+    _write_plan(
+        plan_path,
+        operations=operations,
+        summary={"total": 2, "move": 1, "rename": 0, "quarantine": 0, "no-op": 1},
+        human_summary="这是测试计划摘要",
+    )
 
     output_csv = tmp_path / "review.csv"
     runner = CliRunner()
-    result = runner.invoke(
-        main,
+    result = _invoke_review_plan(
+        runner,
+        config_path,
         [
-            "--config", str(config_path),
-            "review-plan",
-            "--input", str(plan_path),
-            "--output", str(output_csv),
-            "--season-threshold", "20",
-            "--episode-threshold", "40",
+            "--input",
+            str(plan_path),
+            "--output",
+            str(output_csv),
+            "--season-threshold",
+            "20",
+            "--episode-threshold",
+            "40",
         ],
     )
 
     assert result.exit_code == 0
+    assert "Plan overview:" in result.output
+    assert "这是测试计划摘要" in result.output
     assert "Plan review summary:" in result.output
     assert "High-risk operations: 2" in result.output
+    assert "High-risk operations preview:" in result.output
+    assert "source:" in result.output
+    assert "destination:" in result.output
+    assert "reason:" in result.output
     assert output_csv.exists()
 
     with open(output_csv, "r", encoding="utf-8", newline="") as f:
@@ -86,3 +111,124 @@ def test_review_plan_generates_csv_and_summary(tmp_path):
     assert len(rows) == 2
     assert any("sample_source" in row["risk_flags"] for row in rows)
     assert any("manual_review" in row["risk_flags"] for row in rows)
+
+
+def test_review_plan_preview_limit_controls_console_output(tmp_path):
+    """--preview-limit should cap displayed high-risk operations and report remaining count."""
+    config_path = tmp_path / "config.yaml"
+    _write_config(config_path, tmp_path / "library")
+
+    plan_path = tmp_path / "plan.json"
+    operations = [
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Series.S20E41.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Series.S20E42.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Series.S20E43.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+    ]
+    _write_plan(
+        plan_path,
+        operations=operations,
+        summary={"total": 3, "move": 0, "rename": 0, "quarantine": 0, "no-op": 3},
+    )
+
+    output_csv = tmp_path / "review.csv"
+    runner = CliRunner()
+    result = _invoke_review_plan(
+        runner,
+        config_path,
+        [
+            "--input",
+            str(plan_path),
+            "--output",
+            str(output_csv),
+            "--preview-limit",
+            "1",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert "  - [1] " in result.output
+    assert "  - [2] " not in result.output
+    assert "  - [3] " not in result.output
+    assert "... and 2 more high-risk operations" in result.output
+
+
+def test_review_plan_show_all_overrides_preview_limit(tmp_path):
+    """--show-all should display all high-risk operations regardless of --preview-limit."""
+    config_path = tmp_path / "config.yaml"
+    _write_config(config_path, tmp_path / "library")
+
+    plan_path = tmp_path / "plan.json"
+    operations = [
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Series.S20E51.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Series.S20E52.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+        {
+            "operation_type": "no-op",
+            "source_path": str(tmp_path / "Series.S20E53.mkv"),
+            "destination_path": None,
+            "reason": "Series needs manual review (season exceeds configured threshold)",
+            "has_conflict": False,
+            "conflict_reason": None,
+        },
+    ]
+    _write_plan(
+        plan_path,
+        operations=operations,
+        summary={"total": 3, "move": 0, "rename": 0, "quarantine": 0, "no-op": 3},
+    )
+
+    output_csv = tmp_path / "review.csv"
+    runner = CliRunner()
+    result = _invoke_review_plan(
+        runner,
+        config_path,
+        [
+            "--input",
+            str(plan_path),
+            "--output",
+            str(output_csv),
+            "--preview-limit",
+            "1",
+            "--show-all",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert "  - [1] " in result.output
+    assert "  - [2] " in result.output
+    assert "  - [3] " in result.output
+    assert "more high-risk operations" not in result.output