feat(06-02): move pipeline core files to kb_workflow.pipeline

- Move pipeline.py, schemas.py, config.py, state.py, errors.py
- Move retry.py, idempotency.py, link_parser.py
- Create pipeline/__init__.py with updated imports
- Update all internal imports to use kb_workflow.pipeline prefix

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

kb_workflow/pipeline/__init__.py

@@ -0,0 +1,64 @@
+"""KB Pipeline - 工作流管道基础设施
+
+提供内容处理工作流的核心 Schema 和工具。
+"""
+
+from kb_workflow.pipeline.schemas import (
+    StepInput,
+    StepOutput,
+    StepResult,
+    StepStatus,
+    StepRecord,
+    WorkflowState,
+    StepDefinition,
+)
+from kb_workflow.pipeline.state import WorkflowStateManager
+from kb_workflow.pipeline.errors import (
+    KBError,
+    NetworkError,
+    ValidationError,
+    SystemError,
+    AuthError,
+    ErrorType,
+)
+from kb_workflow.pipeline.config import KBSettings, get_settings
+from kb_workflow.pipeline.retry import retry_on_network_error
+from kb_workflow.pipeline.link_parser import LinkParser, ParsedLink
+from kb_workflow.pipeline.idempotency import IdempotencyChecker
+from kb_workflow.pipeline.platforms import PlatformAdapter, BilibiliAdapter, ZhihuAdapter
+
+__all__ = [
+    # Schemas
+    "StepInput",
+    "StepOutput",
+    "StepResult",
+    "StepStatus",
+    "StepRecord",
+    "WorkflowState",
+    "StepDefinition",
+    # State
+    "WorkflowStateManager",
+    # Errors
+    "KBError",
+    "NetworkError",
+    "ValidationError",
+    "SystemError",
+    "AuthError",
+    "ErrorType",
+    # Config
+    "KBSettings",
+    "get_settings",
+    # Retry
+    "retry_on_network_error",
+    # Link Parser
+    "LinkParser",
+    "ParsedLink",
+    # Idempotency
+    "IdempotencyChecker",
+    # Platform Adapters
+    "PlatformAdapter",
+    "BilibiliAdapter",
+    "ZhihuAdapter",
+]
+
+__version__ = "0.1.0"

kb_workflow/pipeline/config.py

@@ -0,0 +1,72 @@
+"""Configuration management for KB Pipeline
+
+Provides centralized configuration using pydantic-settings.
+Configuration is loaded from ~/.kb_env file and environment variables.
+"""
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Optional
+
+from pydantic import Field, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class KBSettings(BaseSettings):
+    """KB 工作流配置
+
+    使用 pydantic-settings 从环境变量和 ~/.kb_env 文件加载配置。
+
+    Attributes:
+        base_path: 知识库根目录（必需）
+        archive_path: 归档目录（必需）
+        git_root: Git 仓库根目录（可选）
+        git_user: Git 用户名（可选）
+        git_email: Git 邮箱（可选）
+        hf_token: HuggingFace Token（敏感信息）
+        zhihu_cookie: 知乎 Cookie（敏感信息）
+    """
+
+    model_config = SettingsConfigDict(
+        env_file=str(Path.home() / ".kb_env"),
+        env_file_encoding="utf-8",
+        env_prefix="KB_",
+        extra="ignore",
+    )
+
+    # 必需配置
+    base_path: str = Field(..., description="知识库根目录")
+    archive_path: str = Field(..., description="归档目录")
+    intermediate_path: str = Field(
+        default_factory=lambda: str(Path.home() / "work" / "intermediate" / "analysis"),
+        description="中间文件存储目录"
+    )
+
+    # 可选配置
+    git_root: str = Field(default="", description="Git 仓库根目录")
+    git_user: str = Field(default="", description="Git 用户名")
+    git_email: str = Field(default="", description="Git 邮箱")
+
+    # 敏感信息（从环境变量读取）
+    hf_token: SecretStr = Field(
+        default=SecretStr(""), description="HuggingFace Token"
+    )
+    zhihu_cookie: SecretStr = Field(
+        default=SecretStr(""), description="知乎 Cookie"
+    )
+    bilibili_cookie: SecretStr = Field(
+        default=SecretStr(""), description="B站 Cookie (for video download)"
+    )
+
+
+@lru_cache
+def get_settings() -> KBSettings:
+    """获取配置单例
+
+    使用 lru_cache 确保配置只加载一次。
+    如需重新加载配置，调用 get_settings.cache_clear()。
+
+    Returns:
+        KBSettings 实例
+    """
+    return KBSettings()

kb_workflow/pipeline/errors.py

@@ -0,0 +1,174 @@
+"""Structured error handling for KB Pipeline
+
+Provides structured error types with context information for debugging.
+All errors include step_name, input_summary, error_type, message, and suggestion.
+"""
+
+from enum import Enum
+from typing import Optional, List, Any
+
+
+class ErrorType(str, Enum):
+    """错误类型分类"""
+    NETWORK = "network"
+    VALIDATION = "validation"
+    SYSTEM = "system"
+    AUTH = "auth"
+
+
+class KBError(Exception):
+    """KB 工作流基础错误
+
+    所有工作流错误的基类，包含丰富的上下文信息便于调试。
+
+    Attributes:
+        message: 错误信息
+        error_type: 错误类型
+        step_name: 发生错误的步骤名称
+        input_summary: 输入数据摘要
+        suggestion: 可操作的建议
+        cause: 原始异常（可选）
+    """
+
+    def __init__(
+        self,
+        message: str,
+        error_type: ErrorType,
+        step_name: str,
+        input_summary: str,
+        suggestion: str,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(message)
+        self.message = message
+        self.error_type = error_type
+        self.step_name = step_name
+        self.input_summary = input_summary
+        self.suggestion = suggestion
+        self.cause = cause
+
+    def to_dict(self) -> dict:
+        """将错误转换为字典格式用于日志和状态记录
+
+        Returns:
+            包含所有错误上下文信息的字典
+        """
+        return {
+            "error_type": self.error_type.value,
+            "step_name": self.step_name,
+            "input_summary": self.input_summary,
+            "message": self.message,
+            "suggestion": self.suggestion
+        }
+
+
+class NetworkError(KBError):
+    """网络错误 (403/404/500/超时)
+
+    用于表示网络请求失败，包括 HTTP 状态码错误和连接超时。
+
+    Attributes:
+        status_code: HTTP 状态码（可选）
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        status_code: Optional[int] = None,
+        cause: Optional[Exception] = None
+    ):
+        if status_code:
+            suggestion = f"检查网络连接，状态码: {status_code}"
+        else:
+            suggestion = "检查网络连接"
+        super().__init__(
+            message=f"网络请求失败: {status_code or 'unknown'}",
+            error_type=ErrorType.NETWORK,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion=suggestion,
+            cause=cause
+        )
+        self.status_code = status_code
+
+
+class ValidationError(KBError):
+    """验证错误 (Schema 验证失败)
+
+    用于表示数据验证失败，包含详细的验证错误列表。
+
+    Attributes:
+        validation_errors: 验证错误列表
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        validation_errors: Optional[List[str]] = None,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(
+            message="Schema 验证失败",
+            error_type=ErrorType.VALIDATION,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion="检查输入数据格式是否符合要求",
+            cause=cause
+        )
+        self.validation_errors = validation_errors or []
+
+
+class SystemError(KBError):
+    """系统错误 (文件系统/内存/依赖缺失)
+
+    用于表示系统级错误，如文件不存在、内存不足、依赖缺失等。
+
+    Attributes:
+        system_message: 系统错误消息
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        system_message: Optional[str] = None,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(
+            message=f"系统错误: {system_message or 'unknown'}",
+            error_type=ErrorType.SYSTEM,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion="检查系统资源和依赖是否满足要求",
+            cause=cause
+        )
+        self.system_message = system_message
+
+
+class AuthError(KBError):
+    """认证/配额错误 (API配额/Cookie过期/Token失效)
+
+    用于表示认证失败或 API 配额超限。
+
+    Attributes:
+        auth_type: 认证类型（如 API Token, Cookie 等）
+    """
+
+    def __init__(
+        self,
+        step_name: str,
+        input_summary: str,
+        auth_type: Optional[str] = None,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(
+            message=f"认证失败: {auth_type or 'unknown'}",
+            error_type=ErrorType.AUTH,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion="检查 API Token 或 Cookie 是否有效",
+            cause=cause
+        )
+        self.auth_type = auth_type

kb_workflow/pipeline/idempotency.py

@@ -0,0 +1,80 @@
+"""Idempotency Checker module
+
+Provides file existence checking to avoid reprocessing content.
+Checks for metadata JSON and content Markdown files.
+"""
+
+from pathlib import Path
+from typing import Tuple, List
+
+
+class IdempotencyChecker:
+    """Checks if content has already been processed
+
+    Determines if the output files for a given content already exist,
+    enabling idempotent processing to avoid duplicate work.
+
+    File naming convention:
+        - Metadata: {platform}_{content_id}_metadata.json
+        - Content: {platform}_{content_id}_content.md
+    """
+
+    def check_exists(
+        self,
+        output_dir: Path,
+        content_id: str,
+        platform: str
+    ) -> Tuple[bool, List[Path]]:
+        """Check if output files already exist for this content
+
+        Args:
+            output_dir: Directory where output files are stored
+            content_id: Unique content identifier (e.g., BV1GJ411x7Qw)
+            platform: Platform name (e.g., bilibili, zhihu)
+
+        Returns:
+            Tuple of (all_exist, existing_files):
+                - all_exist: True if both metadata and content files exist
+                - existing_files: List of Path objects for existing files
+        """
+        existing_files: List[Path] = []
+
+        if not content_id or not platform:
+            return (False, existing_files)
+
+        # Define expected file paths
+        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
+        content_file = output_dir / f"{platform}_{content_id}_content.md"
+
+        # Check if files exist
+        if metadata_file.exists():
+            existing_files.append(metadata_file)
+
+        if content_file.exists():
+            existing_files.append(content_file)
+
+        # Return True only if both files exist
+        all_exist = len(existing_files) == 2
+
+        return (all_exist, existing_files)
+
+    def get_expected_files(
+        self,
+        output_dir: Path,
+        content_id: str,
+        platform: str
+    ) -> Tuple[Path, Path]:
+        """Get the expected file paths for a content
+
+        Args:
+            output_dir: Directory where output files are stored
+            content_id: Unique content identifier
+            platform: Platform name
+
+        Returns:
+            Tuple of (metadata_path, content_path)
+        """
+        metadata_path = output_dir / f"{platform}_{content_id}_metadata.json"
+        content_path = output_dir / f"{platform}_{content_id}_content.md"
+
+        return (metadata_path, content_path)

kb_workflow/pipeline/link_parser.py

@@ -0,0 +1,181 @@
+"""Link Parser module
+
+Provides unified URL parsing for various content platforms.
+Supports Bilibili (video, article) and Zhihu (answer, zhuanlan) URLs.
+"""
+
+import re
+from dataclasses import dataclass, field
+from typing import Optional
+
+import requests
+
+from kb_workflow.pipeline.retry import retry_on_network_error
+
+
+@dataclass
+class ParsedLink:
+    """Represents a parsed link with platform and content information
+
+    Attributes:
+        platform: Platform name (bilibili, zhihu)
+        content_type: Content type (video, article, answer, zhuanlan)
+        content_id: Unique content identifier
+        original_url: Original URL provided by user
+        resolved_url: Resolved URL after short link expansion (defaults to original_url)
+    """
+    platform: str
+    content_type: str
+    content_id: str
+    original_url: str
+    resolved_url: str = field(default="")
+
+    def __post_init__(self):
+        """Set resolved_url to original_url if not provided"""
+        if not self.resolved_url:
+            self.resolved_url = self.original_url
+
+
+class LinkParser:
+    """Unified URL parser for multiple platforms
+
+    Parses URLs from Bilibili and Zhihu, extracting platform, content type,
+    and content ID. Supports short link resolution for b23.tv URLs.
+    """
+
+    # Bilibili patterns
+    BILIBILI_VIDEO_PATTERN = re.compile(
+        r'https?://(?:www\.)?bilibili\.com/video/(BV[a-zA-Z0-9]+)',
+        re.IGNORECASE
+    )
+    BILIBILI_ARTICLE_PATTERN = re.compile(
+        r'https?://(?:www\.)?bilibili\.com/(?:read|article)/(cv[a-zA-Z0-9]+)',
+        re.IGNORECASE
+    )
+
+    # Zhihu patterns
+    ZHIHU_ANSWER_PATTERN = re.compile(
+        r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)',
+        re.IGNORECASE
+    )
+    ZHIHU_ZHUANLAN_PATTERN = re.compile(
+        r'https?://zhuanlan\.zhihu\.com/p/(\d+)',
+        re.IGNORECASE
+    )
+
+    # Short link patterns
+    B23_TV_PATTERN = re.compile(
+        r'https?://(?:www\.)?b23\.tv/([a-zA-Z0-9]+)',
+        re.IGNORECASE
+    )
+
+    # Timeout for short link resolution
+    SHORT_LINK_TIMEOUT = 10
+
+    @retry_on_network_error
+    def _resolve_shortlink(self, url: str) -> Optional[str]:
+        """Resolve short link to its final URL
+
+        Uses HTTP HEAD request with redirects to get the final URL.
+
+        Args:
+            url: Short link URL to resolve
+
+        Returns:
+            Final URL after redirects, or None if resolution failed
+        """
+        try:
+            response = requests.head(
+                url,
+                allow_redirects=True,
+                timeout=self.SHORT_LINK_TIMEOUT
+            )
+            if response.status_code == 200:
+                return response.url
+            return None
+        except (requests.RequestException, OSError):
+            return None
+
+    def parse(self, url: str) -> Optional[ParsedLink]:
+        """Parse a URL and extract platform, content type, and ID
+
+        Args:
+            url: URL to parse
+
+        Returns:
+            ParsedLink with extracted information, or None if URL is not recognized
+        """
+        if not url or not isinstance(url, str):
+            return None
+
+        url = url.strip()
+
+        # Try b23.tv short link first
+        if self.B23_TV_PATTERN.match(url):
+            resolved_url = self._resolve_shortlink(url)
+            if resolved_url:
+                # Recursively parse the resolved URL
+                result = self._parse_standard_url(resolved_url)
+                if result:
+                    result.original_url = url
+                    result.resolved_url = resolved_url
+                return result
+            return None
+
+        return self._parse_standard_url(url)
+
+    def _parse_standard_url(self, url: str) -> Optional[ParsedLink]:
+        """Parse a standard (non-short) URL
+
+        Args:
+            url: URL to parse
+
+        Returns:
+            ParsedLink with extracted information, or None if URL is not recognized
+        """
+        # Try Bilibili video
+        match = self.BILIBILI_VIDEO_PATTERN.match(url)
+        if match:
+            return ParsedLink(
+                platform="bilibili",
+                content_type="video",
+                content_id=match.group(1),
+                original_url=url,
+                resolved_url=url
+            )
+
+        # Try Bilibili article
+        match = self.BILIBILI_ARTICLE_PATTERN.match(url)
+        if match:
+            return ParsedLink(
+                platform="bilibili",
+                content_type="article",
+                content_id=match.group(1),
+                original_url=url,
+                resolved_url=url
+            )
+
+        # Try Zhihu answer
+        match = self.ZHIHU_ANSWER_PATTERN.match(url)
+        if match:
+            return ParsedLink(
+                platform="zhihu",
+                content_type="answer",
+                content_id=match.group(1),
+                original_url=url,
+                resolved_url=url
+            )
+
+        # Try Zhihu zhuanlan
+        match = self.ZHIHU_ZHUANLAN_PATTERN.match(url)
+        if match:
+            return ParsedLink(
+                platform="zhihu",
+                content_type="zhuanlan",
+                content_id=match.group(1),
+                original_url=url,
+                resolved_url=url
+            )
+
+        # URL not recognized
+        return None

kb_workflow/pipeline/pipeline.py

@@ -0,0 +1,399 @@
+"""Pipeline engine - orchestrates multiple processors
+
+The Pipeline class coordinates the workflow:
+1. Parse URL to extract platform and content ID
+2. Run processors in order (parse, download, asr, analyze)
+3. Handle errors and state persistence
+4. Return results
+
+Key design decisions:
+- Sequential execution (simpler, easier to debug)
+- State persistence after each step (via WorkflowStateManager)
+- Simple progress logging
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from kb_workflow.pipeline.config import get_settings
+from kb_workflow.pipeline.link_parser import LinkParser, ParsedLink
+from kb_workflow.pipeline.processors.base import BaseProcessor
+from kb_workflow.pipeline.schemas import PipelineConfig, PipelineInput, PipelineOutput
+
+logger = logging.getLogger(__name__)
+
+
+class Pipeline:
+    """Pipeline engine - orchestrates multiple processors
+
+    Orchestrates the workflow:
+    1. Parse URL to extract platform and content ID
+    2. Run processors in order (parse, download, asr, analyze)
+    3. Handle errors and state persistence
+    4. Return results
+
+    Key design decisions:
+    - Sequential execution (simpler, easier to debug)
+    - State persistence after each step (via WorkflowStateManager)
+    - Simple progress logging
+    """
+
+    def __init__(
+        self,
+        processors: List[BaseProcessor],
+        config: Optional[PipelineConfig] = None,
+        output_dir: Optional[Path] = None,
+    ):
+        """Initialize the Pipeline.
+
+        Args:
+            processors: List of processor instances to use
+            config: Optional PipelineConfig for pipeline settings
+            output_dir: Optional output directory for content
+
+        Raises:
+            ValueError: If a processor's STEP_NAME is not in config.steps
+        """
+        self.processors: Dict[str, BaseProcessor] = {p.STEP_NAME: p for p in processors}
+        self.config = config or PipelineConfig()
+        self.settings = get_settings()
+        self.output_dir = output_dir or Path(self.settings.intermediate_path)
+
+        # Validate processors - each processor's STEP_NAME must be in config.steps
+        for step_name in self.processors:
+            if step_name not in self.config.steps:
+                raise ValueError(f"Unknown processor: {step_name}")
+
+    def _parse_url(self, url: str) -> Optional[ParsedLink]:
+        """Parse URL using LinkParser.
+
+        Args:
+            url: URL to parse
+
+        Returns:
+            ParsedLink with extracted information, or None if URL is not recognized
+        """
+        if not url:
+            return None
+
+        parser = LinkParser()
+        parsed = parser.parse(url)
+
+        if not parsed:
+            logger.warning(f"Failed to parse URL: {url}")
+            return None
+
+        return parsed
+
+    def _get_content_dir(self, parsed: ParsedLink) -> Path:
+        """Get content directory from parsed URL.
+
+        Creates the directory if it doesn't exist.
+
+        Args:
+            parsed: ParsedLink with platform and content_id
+
+        Returns:
+            Path to the content directory
+        """
+        content_dir = self.output_dir / f"{parsed.platform}_{parsed.content_id}"
+        content_dir.mkdir(parents=True, exist_ok=True)
+        return content_dir
+
+    async def run(self, url: str) -> PipelineOutput:
+        """Run the pipeline for a given URL.
+
+        Executes all configured processors in order:
+        1. Parse URL to get platform and content_id
+        2. Create content directory
+        3. Run each processor sequentially
+        4. Handle errors and state persistence
+        5. Return results
+
+        Args:
+            url: URL to process
+
+        Returns:
+            PipelineOutput with success status, results, and content_dir
+        """
+        # 1. Parse URL
+        parsed = self._parse_url(url)
+        if not parsed:
+            return PipelineOutput(
+                success=False,
+                error_message=f"Failed to parse URL: {url}",
+                results={},
+                content_dir=None,
+            )
+
+        # 2. Create content directory
+        content_dir = self._get_content_dir(parsed)
+
+        # 3. Initialize state manager
+        from kb_workflow.pipeline.state import WorkflowStateManager
+        state_manager = WorkflowStateManager(content_dir)
+
+        # Get or create workflow state
+        state = state_manager.get_or_create(
+            content_id=parsed.content_id,
+            platform=parsed.platform,
+            source_url=url,
+        )
+
+        # 4. Run processors in order
+        results: Dict = {}
+
+        for step_name in self.config.steps:
+            if step_name not in self.processors:
+                logger.warning(f"Step {step_name} not found in processors, skipping")
+                continue
+
+            # Check if step should be skipped
+            if self.config.skip_completed and state.is_step_completed(step_name):
+                # Ask user whether to re-run the completed step
+                should_rerun = self._ask_rerun_step(step_name, content_dir)
+
+                if should_rerun:
+                    logger.info(f"Re-running completed step: {step_name}")
+                    # Clear the step's state record so it can be re-executed
+                    if step_name in state.steps:
+                        del state.steps[step_name]
+                        state_manager.save(state)
+                else:
+                    logger.info(f"Skipping completed step: {step_name}")
+                    # Scan existing files and populate results for skipped steps
+                    self._populate_skipped_step_results(step_name, results, content_dir, parsed)
+                    continue
+
+            # Skip ASR for text-based content types
+            text_content_types = {"article", "answer", "zhuanlan", "question"}
+            if step_name == "asr" and parsed.content_type in text_content_types:
+                logger.info(f"Skipping ASR for text content type: {parsed.content_type}")
+                continue
+
+            processor = self.processors[step_name]
+
+            # Mark step as started
+            state.mark_step_started(step_name)
+            state_manager.save(state)
+
+            try:
+                # Create step input based on step type
+                from kb_workflow.pipeline.schemas import (
+                    ParseInput, DownloadInput, ASRInput, AnalyzeInput, ArchiveInput
+                )
+
+                if step_name == "parse":
+                    step_input = ParseInput(
+                        url=url,
+                        content_id=parsed.content_id,
+                        platform=parsed.platform,
+                    )
+                elif step_name == "download":
+                    step_input = DownloadInput(
+                        content_id=parsed.content_id,
+                        platform=parsed.platform,
+                        content_dir=str(content_dir),
+                        content_type=parsed.content_type,
+                        source_url=url,
+                    )
+                elif step_name == "asr":
+                    # ASR needs video path from download step
+                    download_result = results.get("download", {})
+                    video_path = download_result.get("content_file", "")
+
+                    step_input = ASRInput(
+                        content_id=parsed.content_id,
+                        platform=parsed.platform,
+                        video_path=video_path,
+                        content_type=parsed.content_type,
+                        content_dir=str(content_dir),
+                    )
+                elif step_name == "analyze":
+                    # Analyze needs metadata and content files
+                    download_result = results.get("download", {})
+                    asr_result = results.get("asr", {})
+                    metadata_file = download_result.get("metadata_file", "")
+                    # Use subtitle JSON if available, otherwise content file
+                    content_file = asr_result.get("output_files", {}).get("json", "") or download_result.get("content_file", "")
+
+                    step_input = AnalyzeInput(
+                        content_id=parsed.content_id,
+                        platform=parsed.platform,
+                        metadata_file=metadata_file,
+                        content_file=content_file,
+                        content_type=parsed.content_type,
+                        output_file=str(content_dir / "analysis.md"),
+                    )
+                elif step_name == "archive":
+                    # Archive needs analysis file from analyze step
+                    analyze_result = results.get("analyze", {})
+                    analysis_file = analyze_result.get("analysis_file", "")
+
+                    step_input = ArchiveInput(
+                        content_id=parsed.content_id,
+                        platform=parsed.platform,
+                        content_dir=str(content_dir),
+                        analysis_file=analysis_file,
+                        content_type=parsed.content_type,
+                    )
+                else:
+                    # Generic step input
+                    from kb_workflow.pipeline.schemas import StepInput
+                    step_input = StepInput(
+                        content_id=parsed.content_id,
+                        platform=parsed.platform,
+                    )
+
+                # Run processor
+                step_output = await processor.process(step_input)
+
+                # Record result
+                results[step_name] = {
+                    "success": step_output.success,
+                    "error_message": step_output.error_message,
+                }
+
+                if step_output.success:
+                    # Add output fields to results
+                    for field_name in ["metadata_file", "content_file", "title", "author",
+                                        "analysis_file", "summary", "output_files",
+                                        "archive_file", "skipped"]:
+                        if hasattr(step_output, field_name):
+                            results[step_name][field_name] = getattr(step_output, field_name)
+
+                    state.mark_step_completed(step_name)
+                else:
+                    state.mark_step_failed(step_name, step_output.error_message or "Unknown error")
+                    # Stop pipeline on error
+                    state_manager.save(state)
+                    return PipelineOutput(
+                        success=False,
+                        error_message=f"Step {step_name} failed: {step_output.error_message}",
+                        results=results,
+                        content_dir=content_dir,
+                    )
+
+            except Exception as e:
+                logger.error(f"Error in step {step_name}: {e}")
+                state.mark_step_failed(step_name, str(e))
+                state_manager.save(state)
+                return PipelineOutput(
+                    success=False,
+                    error_message=f"Step {step_name} raised exception: {e}",
+                    results=results,
+                    content_dir=content_dir,
+                )
+
+            # Save state after each successful step
+            state_manager.save(state)
+
+        # 5. Return success
+        return PipelineOutput(
+            success=True,
+            error_message=None,
+            results=results,
+            content_dir=content_dir,
+        )
+
+    def _ask_rerun_step(self, step_name: str, content_dir: Path) -> bool:
+        """Ask user whether to re-run a completed step.
+
+        Args:
+            step_name: Name of the step to potentially re-run
+            content_dir: Content directory path
+
+        Returns:
+            True if user wants to re-run, False to skip
+        """
+        try:
+            from rich.console import Console
+            from rich.prompt import Confirm
+
+            console = Console()
+            return Confirm.ask(
+                f"Step '{step_name}' is already completed. Re-run it?",
+                default=False
+            )
+        except ImportError:
+            # Fallback to basic input if rich is not available
+            response = input(f"Step '{step_name}' is already completed. Re-run it? [y/N] ")
+            return response.lower() in ('y', 'yes')
+
+    def _populate_skipped_step_results(
+        self,
+        step_name: str,
+        results: Dict,
+        content_dir: Path,
+        parsed: ParsedLink
+    ) -> None:
+        """Populate results for skipped steps by scanning existing files.
+
+        This ensures downstream steps can find the outputs they need,
+        even when a step is skipped.
+
+        Args:
+            step_name: Name of the skipped step
+            results: Results dictionary to populate
+            content_dir: Content directory path
+            parsed: Parsed link with platform and content_id
+        """
+        if step_name == "asr":
+            # Scan for subtitle files generated by ASR
+            # Try both new naming convention (with _subtitle suffix) and old convention (without suffix)
+            base_name_new = f"{parsed.platform}_{parsed.content_id}_subtitle"
+            base_name_old = f"{parsed.platform}_{parsed.content_id}"
+
+            json_path = content_dir / f"{base_name_new}.json"
+            if not json_path.exists():
+                json_path = content_dir / f"{base_name_old}.json"
+
+            srt_path = content_dir / f"{base_name_new}.srt"
+            if not srt_path.exists():
+                srt_path = content_dir / f"{base_name_old}.srt"
+
+            output_files = {}
+            if json_path.exists():
+                output_files["json"] = str(json_path)
+            if srt_path.exists():
+                output_files["srt"] = str(srt_path)
+
+            if output_files:
+                results["asr"] = {
+                    "success": True,
+                    "error_message": None,
+                    "output_files": output_files,
+                }
+                logger.info(f"Found existing ASR outputs: {output_files}")
+            else:
+                logger.warning(f"No existing ASR outputs found in {content_dir}")
+                results["asr"] = {
+                    "success": False,
+                    "error_message": "No existing subtitle files found",
+                    "output_files": {},
+                }
+
+        elif step_name == "download":
+            # Scan for metadata file
+            metadata_path = content_dir / "metadata.json"
+            if metadata_path.exists():
+                results["download"] = {
+                    "success": True,
+                    "error_message": None,
+                    "metadata_file": str(metadata_path),
+                }
+                logger.info(f"Found existing download metadata: {metadata_path}")
+
+        elif step_name == "analyze":
+            # Scan for analysis output
+            analysis_path = content_dir / "analysis.md"
+            if analysis_path.exists():
+                results["analyze"] = {
+                    "success": True,
+                    "error_message": None,
+                    "analysis_file": str(analysis_path),
+                }
+                logger.info(f"Found existing analysis: {analysis_path}")

kb_workflow/pipeline/retry.py

@@ -0,0 +1,51 @@
+"""Retry mechanism for KB Pipeline
+
+Provides automatic retry for network errors using tenacity library.
+"""
+
+import logging
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_exponential,
+    retry_if_exception_type,
+    before_sleep_log,
+)
+
+from .errors import NetworkError
+
+logger = logging.getLogger(__name__)
+
+retry_on_network_error = retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=10),
+    retry=retry_if_exception_type(NetworkError),
+    before_sleep=before_sleep_log(logger, logging.WARNING),
+    reraise=True,
+)
+"""Decorator for automatic retry on network errors
+
+Automatically retries functions that raise NetworkError up to 3 times
+with exponential backoff (2s, 4s, up to max 10s).
+
+Example:
+    @retry_on_network_error
+    def fetch_data(url):
+        response = requests.get(url)
+        if response.status_code >= 500:
+            raise NetworkError(
+                step_name="fetch",
+                input_summary=f"url={url}",
+                status_code=response.status_code
+            )
+        return response.json()
+
+Args:
+    func: Function to wrap with retry logic
+
+Returns:
+    Wrapped function with retry logic
+
+Raises:
+    NetworkError: After 3 failed attempts
+"""

kb_workflow/pipeline/schemas.py

@@ -0,0 +1,446 @@
+"""Core Pipeline Schemas
+
+定义工作流管道的核心 Schema，包括 StepInput、StepOutput、StepResult、
+StepStatus、StepRecord、WorkflowState 和 StepDefinition。
+使用 Pydantic v2 确保类型安全和自动验证。
+"""
+
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, Literal, Any, List
+from datetime import datetime
+from enum import Enum
+from pathlib import Path
+
+
+class StepStatus(str, Enum):
+    """步骤执行状态枚举"""
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+
+class StepInput(BaseModel):
+    """所有步骤输入的基类
+
+    Attributes:
+        content_id: 内容唯一标识，如 B站 BV号、知乎回答ID等
+        platform: 平台名称，限定为 bilibili 或 zhihu
+    """
+    content_id: str = Field(..., min_length=1, description="内容唯一标识")
+    platform: Literal["bilibili", "zhihu"] = Field(..., description="平台名称")
+
+
+class StepOutput(BaseModel):
+    """所有步骤输出的基类
+
+    Attributes:
+        success: 步骤是否成功执行
+        error_message: 错误信息，成功时为 None
+        completed_at: 完成时间，默认为当前时间
+    """
+    success: bool = Field(..., description="步骤是否成功")
+    error_message: Optional[str] = Field(None, description="错误信息")
+    completed_at: datetime = Field(default_factory=datetime.now, description="完成时间")
+
+
+class StepResult(BaseModel):
+    """步骤执行结果
+
+    包含步骤的输入、输出和执行状态。
+
+    Attributes:
+        step_name: 步骤名称
+        input_data: 步骤输入数据
+        output_data: 步骤输出数据，可能为 None
+        status: 步骤执行状态
+    """
+    step_name: str = Field(..., description="步骤名称")
+    input_data: StepInput = Field(..., description="步骤输入数据")
+    output_data: Optional[StepOutput] = Field(None, description="步骤输出数据")
+    status: StepStatus = Field(default=StepStatus.PENDING, description="步骤执行状态")
+
+
+class StepRecord(BaseModel):
+    """单个步骤的执行记录
+
+    记录步骤执行的状态、时间和结果摘要。
+
+    Attributes:
+        step_name: 步骤名称
+        status: 步骤执行状态
+        started_at: 开始时间
+        completed_at: 完成时间
+        error_message: 错误信息
+        output_summary: 输出摘要
+    """
+    step_name: str = Field(..., description="步骤名称")
+    status: StepStatus = Field(..., description="步骤执行状态")
+    started_at: Optional[datetime] = Field(None, description="开始时间")
+    completed_at: Optional[datetime] = Field(None, description="完成时间")
+    error_message: Optional[str] = Field(None, description="错误信息")
+    output_summary: Optional[str] = Field(None, description="输出摘要")
+
+
+class WorkflowState(BaseModel):
+    """工作流状态 - 存储在 workflow_state.json
+
+    管理整个工作流的状态，包括内容信息、执行历史和步骤状态。
+
+    Attributes:
+        schema_version: Schema 版本号
+        content_id: 内容唯一标识
+        platform: 平台名称
+        source_url: 原始链接
+        created_at: 创建时间
+        updated_at: 更新时间
+        steps: 步骤执行记录字典
+    """
+    schema_version: str = Field(default="1.0", description="Schema 版本号")
+    content_id: str = Field(..., description="内容唯一标识")
+    platform: str = Field(..., description="平台名称")
+    source_url: str = Field(..., description="原始链接")
+    created_at: datetime = Field(default_factory=datetime.now, description="创建时间")
+    updated_at: datetime = Field(default_factory=datetime.now, description="更新时间")
+    steps: Dict[str, StepRecord] = Field(default_factory=dict, description="步骤执行记录")
+
+    def is_step_completed(self, step_name: str) -> bool:
+        """检查指定步骤是否已完成
+
+        Args:
+            step_name: 步骤名称
+
+        Returns:
+            如果步骤存在且状态为 COMPLETED 返回 True，否则返回 False
+        """
+        record = self.steps.get(step_name)
+        return record is not None and record.status == StepStatus.COMPLETED
+
+    def mark_step_started(self, step_name: str) -> None:
+        """标记步骤开始执行
+
+        Args:
+            step_name: 步骤名称
+        """
+        self.steps[step_name] = StepRecord(
+            step_name=step_name,
+            status=StepStatus.RUNNING,
+            started_at=datetime.now()
+        )
+        self.updated_at = datetime.now()
+
+    def mark_step_completed(self, step_name: str, output_summary: Optional[str] = None) -> None:
+        """标记步骤完成
+
+        Args:
+            step_name: 步骤名称
+            output_summary: 输出摘要（可选）
+        """
+        if step_name in self.steps:
+            self.steps[step_name].status = StepStatus.COMPLETED
+            self.steps[step_name].completed_at = datetime.now()
+            self.steps[step_name].output_summary = output_summary
+        else:
+            self.steps[step_name] = StepRecord(
+                step_name=step_name,
+                status=StepStatus.COMPLETED,
+                completed_at=datetime.now(),
+                output_summary=output_summary
+            )
+        self.updated_at = datetime.now()
+
+    def mark_step_failed(self, step_name: str, error_message: str) -> None:
+        """标记步骤失败
+
+        Args:
+            step_name: 步骤名称
+            error_message: 错误信息
+        """
+        if step_name in self.steps:
+            self.steps[step_name].status = StepStatus.FAILED
+            self.steps[step_name].error_message = error_message
+            self.steps[step_name].completed_at = datetime.now()
+        else:
+            self.steps[step_name] = StepRecord(
+                step_name=step_name,
+                status=StepStatus.FAILED,
+                error_message=error_message,
+                completed_at=datetime.now()
+            )
+        self.updated_at = datetime.now()
+
+
+class StepDefinition(BaseModel):
+    """步骤定义 - 描述步骤的前置条件、输出格式、成功标准
+
+    Attributes:
+        step_name: 步骤名称
+        description: 步骤描述
+        input_schema: 输入 Schema 类型
+        output_schema: 输出 Schema 类型
+        dependencies: 依赖的步骤列表
+    """
+    step_name: str = Field(..., description="步骤名称")
+    description: str = Field(..., description="步骤描述")
+    input_schema: type[StepInput] = Field(..., description="输入 Schema 类型")
+    output_schema: type[StepOutput] = Field(..., description="输出 Schema 类型")
+    dependencies: list[str] = Field(default_factory=list, description="依赖的步骤列表")
+
+
+# =============================================================================
+# ASR Schemas (Phase 3)
+# =============================================================================
+
+class ASRInput(StepInput):
+    """ASR 步骤输入
+
+    继承 StepInput 的基础字段，添加视频路径和内容类型。
+
+    Attributes:
+        video_path: 视频文件路径（可能为空，用于 Whisper 降级）
+        content_type: 内容类型，默认为 'video'
+        content_dir: 内容目录路径，用于输出字幕文件
+    """
+    video_path: str = Field(default="", description="视频文件路径（可能为空）")
+    content_type: str = Field(default="video", description="内容类型")
+    content_dir: str = Field(..., description="内容目录路径")
+
+
+class ASROutput(StepOutput):
+    """ASR 步骤输出
+
+    继承 StepOutput 的基础字段，添加 ASR 特定字段。
+
+    Attributes:
+        tool: 使用的工具 (subtitle, whisper, whisper-diarization)
+        has_subtitle: 是否有官方字幕
+        output_files: 输出文件路径字典 {txt, srt, json}
+        segments: 带时间戳的分段列表
+        text: 完整文本
+        duration_seconds: 音频时长
+        speakers: 说话者列表（仅说话者分离时有值）
+    """
+    tool: Literal["subtitle", "whisper", "whisper-diarization"] = Field(
+        ..., description="使用的工具"
+    )
+    has_subtitle: bool = Field(..., description="是否有官方字幕")
+    output_files: Dict[str, str] = Field(
+        default_factory=dict, description="输出文件路径 {txt, srt, json}"
+    )
+    segments: List[Dict[str, Any]] = Field(
+        default_factory=list, description="带时间戳的分段列表"
+    )
+    text: str = Field(default="", description="完整文本")
+    duration_seconds: float = Field(default=0.0, description="音频时长")
+    speakers: Optional[List[str]] = Field(
+        None, description="说话者列表（仅说话者分离时有值）"
+    )
+
+
+# =============================================================================
+# Parse Schemas (Phase 04)
+# =============================================================================
+
+
+class ParseInput(StepInput):
+    """Parse 步骤输入
+
+    继承 StepInput 的基础字段，添加 URL 字段。
+    content_id 和 platform 在解析前是空的，解析后会被填充。
+
+    Attributes:
+        url: 要解析的 URL
+        content_id: 解析后填充的内容 ID（初始为空）
+        platform: 解析后填充的平台名称（初始为空）
+    """
+    url: str = Field(..., min_length=1, description="URL to parse")
+    content_id: str = Field(default="", description="Will be populated after parsing")
+    platform: str = Field(default="", description="Will be populated after parsing")
+
+
+class ParseOutput(StepOutput):
+    """Parse 步骤输出
+
+    继承 StepOutput 的基础字段，添加解析特定字段。
+
+    Attributes:
+        platform: 检测到的平台 (bilibili, zhihu)
+        content_type: 内容类型 (video, article, answer, zhuanlan)
+        content_id: 提取的内容 ID
+        resolved_url: 短链解析后的完整 URL
+        original_url: 用户提供的原始 URL
+    """
+    platform: str = Field(..., description="Detected platform")
+    content_type: str = Field(..., description="Content type (video, article, etc.)")
+    content_id: str = Field(..., description="Extracted content ID")
+    resolved_url: str = Field(..., description="Resolved URL after short link expansion")
+    original_url: str = Field(..., description="Original URL provided by user")
+
+
+# =============================================================================
+# Download Schemas (Phase 04)
+# =============================================================================
+
+
+class DownloadInput(StepInput):
+    """Download 步骤输入
+
+    继承 StepInput 的基础字段，添加下载特定字段。
+
+    Attributes:
+        content_dir: 保存下载内容的目录路径
+        content_type: 内容类型 (video, article, answer, zhuanlan)
+        source_url: 原始 URL（可选，用于知乎回答等需要完整 URL 的情况）
+    """
+    content_dir: str = Field(..., min_length=1, description="Directory to save downloaded content")
+    content_type: str = Field(..., min_length=1, description="Content type (video, article, answer, zhuanlan)")
+    source_url: Optional[str] = Field(None, description="Original source URL")
+
+
+class DownloadOutput(StepOutput):
+    """Download 步骤输出
+
+    继承 StepOutput 的基础字段，添加下载特定字段。
+
+    Attributes:
+        metadata_file: 元数据 JSON 文件路径
+        content_file: 内容文件路径 (视频/音频或文本)
+        title: 内容标题
+        author: 内容作者
+        duration: 视频内容的时长（秒），仅视频有值
+    """
+    metadata_file: str = Field(..., description="Path to metadata JSON file")
+    content_file: str = Field(..., description="Path to content file (video/audio or text)")
+    title: str = Field(..., description="Content title")
+    author: str = Field(..., description="Content author")
+    duration: Optional[int] = Field(None, description="Duration for video content in seconds")
+
+
+# =============================================================================
+# Analyze Schemas (Phase 04)
+# =============================================================================
+
+
+class AnalyzeInput(StepInput):
+    """Analyze 步骤输入
+
+    继承 StepInput 的基础字段，添加分析特定字段。
+
+    Attributes:
+        metadata_file: 元数据 JSON 文件路径
+        content_file: 内容文件路径（字幕 JSON 或文章内容）
+        content_type: 内容类型（video 或 article）
+        output_file: 输出分析 Markdown 文件路径
+    """
+    metadata_file: str = Field(..., min_length=1, description="Path to metadata JSON file")
+    content_file: str = Field(..., min_length=1, description="Path to content file (subtitle JSON or article content)")
+    content_type: str = Field(..., min_length=1, description="Content type (video or article)")
+    output_file: str = Field(..., min_length=1, description="Path to output analysis Markdown file")
+
+
+class AnalyzeOutput(StepOutput):
+    """Analyze 步骤输出
+
+    继承 StepOutput 的基础字段，添加分析特定字段。
+
+    Attributes:
+        analysis_file: 生成的分析报告文件路径
+        summary: 内容摘要
+        main_topics: 提取的主要主题列表
+        key_insights: 关键见解列表
+        tags: 推荐标签列表
+        stats: 分析统计信息字典
+    """
+    analysis_file: str = Field(..., description="Path to generated analysis report")
+    summary: str = Field(default="", description="Content summary")
+    main_topics: List[str] = Field(default_factory=list, description="Main topics extracted")
+    key_insights: List[str] = Field(default_factory=list, description="Key insights")
+    tags: List[str] = Field(default_factory=list, description="Recommended tags")
+    stats: Dict[str, Any] = Field(default_factory=dict, description="Statistics about analysis")
+
+
+# =============================================================================
+# Pipeline Schemas (Phase 04-04)
+# =============================================================================
+
+
+class PipelineInput(BaseModel):
+    """Pipeline input - URL to process
+
+    Attributes:
+        url: URL to process (required)
+        output_dir: Optional output directory for content
+        skip_completed: Skip already completed steps (default True)
+    """
+    url: str = Field(..., min_length=1, description="URL to process")
+    output_dir: Optional[Path] = Field(None, description="Output directory for content")
+    skip_completed: bool = Field(default=True, description="Skip already completed steps")
+
+
+class PipelineOutput(BaseModel):
+    """Pipeline output - results from all steps
+
+    Attributes:
+        success: Whether the pipeline completed successfully
+        error_message: Error message if pipeline failed
+        results: Dictionary of results from each step
+        content_dir: Path to the content directory
+    """
+    success: bool = Field(..., description="Whether the pipeline completed successfully")
+    error_message: Optional[str] = Field(None, description="Error message if pipeline failed")
+    results: Dict[str, Any] = Field(default_factory=dict, description="Results from each step")
+    content_dir: Optional[Path] = Field(None, description="Path to the content directory")
+
+
+class PipelineConfig(BaseModel):
+    """Pipeline configuration
+
+    Attributes:
+        steps: Steps to execute in order (default: parse, download, asr, analyze, archive)
+        output_dir: Base output directory
+        skip_completed: Skip completed steps (default True)
+        max_retries: Max retries per step (default 3)
+    """
+    steps: List[str] = Field(
+        default_factory=lambda: ["parse", "download", "asr", "analyze", "archive"],
+        description="Steps to execute in order"
+    )
+    output_dir: Optional[Path] = Field(None, description="Base output directory")
+    skip_completed: bool = Field(default=True, description="Skip completed steps")
+    max_retries: int = Field(default=3, description="Max retries per step")
+
+
+# =============================================================================
+# Archive Schemas (Phase 5)
+# =============================================================================
+
+
+class ArchiveInput(StepInput):
+    """Archive 步骤输入
+
+    继承 StepInput 的基础字段，添加归档特定字段。
+
+    Attributes:
+        content_dir: 内容目录路径（中间目录）
+        analysis_file: 分析报告 md 文件路径
+        content_type: 内容类型
+    """
+    content_dir: str = Field(..., min_length=1, description="Content directory path (intermediate)")
+    analysis_file: str = Field(..., min_length=1, description="Path to analysis Markdown file")
+    content_type: str = Field(..., min_length=1, description="Content type (video, article, etc.)")
+
+
+class ArchiveOutput(StepOutput):
+    """Archive 步骤输出
+
+    继承 StepOutput 的基础字段，添加归档特定字段。
+
+    Attributes:
+        archive_file: 归档后的文件路径
+        skipped: 是否跳过（已有相同文件）
+        platform: 平台名称
+    """
+    archive_file: str = Field(default="", description="Path to archived file")
+    skipped: bool = Field(default=False, description="Whether archiving was skipped")
+    platform: str = Field(default="", description="Platform name")

kb_workflow/pipeline/state.py

@@ -0,0 +1,152 @@
+"""
+WorkflowStateManager - 步骤级状态持久化管理器
+
+提供工作流状态的原子写入、备份轮转和崩溃恢复功能。
+
+Purpose: 为工作流提供可靠的状态管理，支持断点续传和崩溃恢复
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from pathlib import Path
+from typing import Optional
+
+from kb_workflow.pipeline.schemas import WorkflowState
+
+
+class WorkflowStateManager:
+    """工作流状态管理器
+
+    管理工作流状态的持久化，包括：
+    - 原子写入（使用临时文件 + os.replace）
+    - 备份轮转（workflow_state.json -> workflow_state.json.bak）
+    - 主文件损坏时从备份恢复
+
+    Attributes:
+        STATE_FILENAME: 状态文件名
+        content_dir: 内容目录路径
+        state_path: 状态文件完整路径
+        backup_path: 备份文件完整路径
+    """
+
+    STATE_FILENAME = "workflow_state.json"
+
+    def __init__(self, content_dir: Path):
+        """初始化状态管理器
+
+        Args:
+            content_dir: 内容目录路径，状态文件将保存在此目录下
+        """
+        self.content_dir = Path(content_dir)
+        self.state_path = self.content_dir / self.STATE_FILENAME
+        self.backup_path = self.content_dir / f"{self.STATE_FILENAME}.bak"
+
+    def load(self) -> Optional[WorkflowState]:
+        """加载工作流状态
+
+        加载策略：
+        1. 尝试加载主文件 (workflow_state.json)
+        2. 如果主文件损坏或不存在，尝试加载备份文件
+        3. 如果都不存在，返回 None
+
+        Returns:
+            WorkflowState 实例，如果文件不存在则返回 None
+
+        Raises:
+            ValueError: 如果文件格式无效且备份也无法恢复
+        """
+        def _try_load(path: Path) -> Optional[WorkflowState]:
+            """尝试从指定路径加载状态"""
+            if not path.exists():
+                return None
+
+            try:
+                with open(path, 'r', encoding='utf-8') as f:
+                    data = json.load(f)
+                return WorkflowState.model_validate(data)
+            except (json.JSONDecodeError, ValueError):
+                return None
+
+        # 先尝试主文件
+        state = _try_load(self.state_path)
+        if state is not None:
+            return state
+
+        # 主文件失败，尝试备份
+        state = _try_load(self.backup_path)
+        return state
+
+    def save(self, state: WorkflowState) -> None:
+        """原子保存工作流状态
+
+        保存策略（原子写入 + 备份轮转）：
+        1. 创建父目录（如果不存在）
+        2. 如果状态文件存在，将其重命名为备份文件
+        3. 写入临时文件
+        4. 原子替换（os.replace）
+
+        原子写入确保：
+        - 读取者永远不会看到部分写入的文件
+        - 崩溃时要么是旧文件，要么是新文件
+
+        Args:
+            state: 要保存的工作流状态
+
+        Raises:
+            OSError: 文件操作失败（权限、磁盘空间等）
+        """
+        # 1. 创建父目录
+        self.content_dir.mkdir(parents=True, exist_ok=True)
+
+        # 2. 备份轮转：如果状态文件存在，移动到备份
+        if self.state_path.exists():
+            # 删除旧备份
+            if self.backup_path.exists():
+                self.backup_path.unlink()
+            # 将当前状态移动到备份
+            self.state_path.rename(self.backup_path)
+
+        # 3. 写入临时文件（在相同目录下，确保原子替换有效）
+        fd, temp_path = tempfile.mkstemp(dir=self.content_dir, suffix='.tmp')
+        try:
+            with os.fdopen(fd, 'w', encoding='utf-8') as f:
+                f.write(state.model_dump_json(indent=2))
+            # 4. 原子替换
+            os.replace(temp_path, self.state_path)
+        except Exception:
+            # 清理临时文件
+            if os.path.exists(temp_path):
+                os.unlink(temp_path)
+            raise
+
+    def get_or_create(
+        self,
+        content_id: str,
+        platform: str,
+        source_url: str
+    ) -> WorkflowState:
+        """获取现有状态或创建新状态
+
+        幂等操作：如果状态已存在，返回现有状态；否则创建新状态。
+
+        Args:
+            content_id: 内容唯一标识
+            platform: 平台名称
+            source_url: 原始链接
+
+        Returns:
+            WorkflowState 实例
+        """
+        existing = self.load()
+        if existing is not None:
+            return existing
+
+        # 创建新状态
+        return WorkflowState(
+            content_id=content_id,
+            platform=platform,
+            source_url=source_url
+        )