docs(01): research phase 1 domain - state, schema, error, config

2026-03-13 16:42:16 +08:00
parent 2ee0a9e41d
commit c555b78d9b
1 changed files with 630 additions and 0 deletions
@@ -0,0 +1,630 @@
+# Phase 1: Foundation - Research
+
+**Researched:** 2026-03-13
+**Domain:** Python workflow infrastructure - state management, schema validation, configuration, error handling
+**Confidence:** HIGH
+
+## Summary
+
+Phase 1 建立工作流基础设施的核心组件：状态管理器、Schema 注册表、错误处理器和配置管理器。这些组件遵循 Python 生态系统的最佳实践，使用 Pydantic v2 进行数据验证，pydantic-settings 进行配置管理，以及原子文件写入模式保证状态持久化的可靠性。
+
+**Primary recommendation:** 使用 Pydantic v2 BaseModel 定义所有 Schema，pydantic-settings 管理配置，复用 gsd-openclaw/state_manager.py 的原子写入模式，定义统一的错误类型层次结构。
+
+<user_constraints>
+## User Constraints (from CONTEXT.md)
+
+### Locked Decisions
+
+#### 状态管理
+- **状态粒度**: 步骤级状态 — 符合现有子技能映射 (metadata→content→whisper→analyzer)
+- **状态关联**: 按内容ID关联 — 如 BV123xxx，重复链接自动识别
+- **幂等执行**: 自动跳过已完成步骤 — 重新执行同一链接时从中断点继续
+- **备份机制**: 需要备份 — 原子写入 + .bak 轮转，崩溃后可恢复
+- **存储位置**: 状态文件存放在内容目录下 — 方便关联查看
+- **历史记录**: 只保留最新状态 — 不追加历史执行记录
+
+#### Schema 验证
+- **验证时机**: 立即验证 — 步骤完成后立即验证，失败则阻止进入下一步
+- **Schema 定义位置**: 集中定义 — 所有步骤的 Schema 集中在一个模块
+- **验证范围**: 验证输入和输出 — 步骤执行前验证输入，执行后验证输出
+- **Schema 格式**: Pydantic 模型 — 自带验证和序列化，类型安全
+
+#### 错误处理
+- **错误分类**: 支持四种错误类型
+  - 网络错误 (403/404/500/超时)
+  - 验证错误 (Schema 验证失败)
+  - 系统错误 (文件系统/内存/依赖缺失)
+  - 认证/配额错误 (API配额/Cookie过期/Token失效)
+- **重试策略**: 自动重试 — 网络错误自动重试3次，指数退避
+- **错误信息**: 完整上下文 — 步骤名、输入摘要、错误类型、失败原因、可操作建议
+- **降级策略**: 定义降级链 — 如官方字幕失败后尝试语音识别
+
+#### 配置管理
+- **配置文件位置**: ~/.kb_env — 延续现有习惯
+- **验证时机**: 启动时验证 — 缺失必需变量则报错退出
+- **敏感信息处理**: 环境变量存敏感 — Cookie/Token 用环境变量，配置文件只存非敏感配置
+- **配置库**: pydantic-settings — 自动验证和类型转换
+
+### Claude's Discretion
+
+无特别指定 — 在锁定决策框架内做技术实现选择。
+
+### Deferred Ideas (OUT OF SCOPE)
+
+None — discussion stayed within phase scope
+</user_constraints>
+
+<phase_requirements>
+## Phase Requirements
+
+| ID | Description | Research Support |
+|----|-------------|-----------------|
+| PIPE-01 | 统一的管道模型 - 每个步骤是独立管道，定义清晰的输入→处理→输出 | Pydantic BaseModel 定义 StepInput/StepOutput |
+| PIPE-02 | 步骤级状态管理 - 主要步骤完成后记录状态到 workflow_state.json | 原子写入 + 备份轮转模式 (参考 gsd-openclaw/state_manager.py) |
+| PIPE-03 | 统一 JSON Schema - 所有子技能遵守统一的 StepInput/StepOutput 格式规范 | Pydantic v2 model_dump() / model_validate() |
+| REL-01 | 幂等执行保证 - 重复执行同一链接不会产生重复文件或重复处理 | WorkflowState.is_step_completed() 检查 |
+| REL-02 | 错误恢复策略 - 403/超时/网络错误有明确的降级策略和重试机制 | FallbackChain + retry_with_backoff() |
+| REL-03 | 清晰的流程定义 - 每个步骤的前置条件、输出格式、成功标准明确 | StepDefinition + StepResult Pydantic 模型 |
+| VAL-01 | Schema 验证机制 - 每个步骤输出通过 Pydantic Schema 验证，失败则阻止进入下一步 | StepOutput 基类 + model_validate() |
+| MNT-02 | 统一配置管理 - 环境变量集中在 ~/.kb_env，使用 pydantic-settings 管理 | pydantic-settings BaseSettings + env_file |
+| MNT-03 | 结构化错误处理 - 错误包含上下文信息，便于调试和日志追踪 | KBError 层次结构 + StructuredErrorContext |
+</phase_requirements>
+
+## Standard Stack
+
+### Core
+| Library | Version | Purpose | Why Standard |
+|---------|---------|---------|--------------|
+| pydantic | 2.x | 数据验证和 Schema 定义 | Python 生态类型安全验证的事实标准，v2 性能提升 5-50x |
+| pydantic-settings | 2.x | 配置管理 | 官方配置库，自动环境变量映射和验证 |
+| python | 3.12+ | 运行时 | 项目约束，支持最新类型语法 |
+
+### Supporting
+| Library | Version | Purpose | When to Use |
+|---------|---------|---------|-------------|
+| pytest | 7.0+ | 测试框架 | 所有单元测试和集成测试 |
+| tenacity | 8.x | 重试机制 | 网络错误自动重试，指数退避 |
+
+### Alternatives Considered
+| Instead of | Could Use | Tradeoff |
+|------------|-----------|----------|
+| pydantic | dataclasses + 手动验证 | 失去自动验证、序列化、JSON Schema 生成 |
+| pydantic-settings | dynaconf | dynaconf 更复杂，pydantic-settings 更轻量且与 Pydantic 无缝集成 |
+| tenacity | 手写重试逻辑 | tenacity 提供成熟的指数退避和钩子机制 |
+
+**Installation:**
+```bash
+pip install pydantic>=2.0 pydantic-settings>=2.0 pytest>=7.0 tenacity>=8.0
+```
+
+## Architecture Patterns
+
+### Recommended Project Structure
+```
+kb-skills/
+├── kb_workflow/                 # 现有工作流工具
+│   └── kb_workflow.py
+├── kb_pipeline/                 # 新建：核心管道基础设施
+│   ├── __init__.py
+│   ├── state.py                 # WorkflowStateManager
+│   ├── schemas.py               # StepInput, StepOutput, StepResult
+│   ├── errors.py                # KBError 层次结构
+│   ├── config.py                # KBSettings (pydantic-settings)
+│   └── retry.py                 # retry_with_backoff
+└── tests/
+    └── test_pipeline/
+        ├── test_state.py
+        ├── test_schemas.py
+        ├── test_errors.py
+        └── test_config.py
+```
+
+### Pattern 1: Atomic State Persistence
+**What:** 原子写入状态文件，防止崩溃时数据损坏
+**When to use:** 所有 workflow_state.json 写入操作
+**Example:**
+```python
+# Source: gsd-openclaw/src/gsd_openclaw/state_manager.py (已验证)
+import json
+import os
+import tempfile
+from pathlib import Path
+
+def save_state_atomically(state: dict, state_path: Path) -> None:
+    """Atomic state save with backup rotation."""
+    backup_path = state_path.with_suffix('.json.bak')
+
+    # 1. Create parent directories if missing
+    state_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # 2. Rotate backup: if state.json exists, move it to state.json.bak
+    if state_path.exists():
+        if backup_path.exists():
+            backup_path.unlink()
+        state_path.rename(backup_path)
+
+    # 3. Write to temp file in same directory (for atomic rename)
+    fd, temp_path = tempfile.mkstemp(dir=state_path.parent, suffix='.tmp')
+    try:
+        with os.fdopen(fd, 'w', encoding='utf-8') as f:
+            json.dump(state, f, indent=2, ensure_ascii=False)
+        # 4. Atomic replace
+        os.replace(temp_path, state_path)
+    except Exception:
+        if os.path.exists(temp_path):
+            os.unlink(temp_path)
+        raise
+```
+
+### Pattern 2: Pydantic Schema Registry
+**What:** 集中定义所有步骤的输入输出 Schema
+**When to use:** 定义新步骤或验证步骤输出
+**Example:**
+```python
+# Source: https://docs.pydantic.dev/latest/concepts/models/
+from pydantic import BaseModel, Field
+from typing import Optional, Literal
+from datetime import datetime
+
+class StepInput(BaseModel):
+    """所有步骤输入的基类"""
+    content_id: str = Field(..., description="内容唯一标识，如 BV号")
+    platform: Literal["bilibili", "zhihu"] = Field(..., description="平台名称")
+
+class StepOutput(BaseModel):
+    """所有步骤输出的基类"""
+    success: bool = Field(..., description="步骤是否成功")
+    error_message: Optional[str] = Field(None, description="错误信息")
+    completed_at: datetime = Field(default_factory=datetime.now)
+
+    class Config:
+        # Pydantic v2 配置
+        json_schema_extra = {
+            "example": {
+                "success": True,
+                "error_message": None,
+                "completed_at": "2026-03-13T10:00:00"
+            }
+        }
+
+class MetadataStepOutput(StepOutput):
+    """元数据步骤输出"""
+    title: Optional[str] = None
+    author: Optional[str] = None
+    duration_seconds: Optional[int] = None
+```
+
+### Pattern 3: pydantic-settings Configuration
+**What:** 使用 pydantic-settings 管理环境变量配置
+**When to use:** 应用启动时加载配置
+**Example:**
+```python
+# Source: https://docs.pydantic.dev/latest/concepts/pydantic_settings/
+from pydantic import Field, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+class KBSettings(BaseSettings):
+    """KB 工作流配置"""
+    model_config = SettingsConfigDict(
+        env_file='~/.kb_env',
+        env_file_encoding='utf-8',
+        env_prefix='KB_',
+        extra='ignore'  # 兼容旧配置文件
+    )
+
+    # 必需配置
+    base_path: str = Field(..., description="知识库根目录")
+    archive_path: str = Field(..., 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="", description="HuggingFace Token")
+    zhihu_cookie: SecretStr = Field(default="", description="知乎 Cookie")
+
+# 使用
+settings = KBSettings()  # 自动从 ~/.kb_env 和环境变量加载
+```
+
+### Pattern 4: Structured Error Handling
+**What:** 结构化错误类型，包含完整上下文
+**When to use:** 所有步骤失败时抛出
+**Example:**
+```python
+from enum import Enum
+from typing import Optional, Any
+from pydantic import BaseModel
+
+class ErrorType(str, Enum):
+    NETWORK = "network"
+    VALIDATION = "validation"
+    SYSTEM = "system"
+    AUTH = "auth"
+
+class KBError(Exception):
+    """KB 工作流基础错误"""
+    def __init__(
+        self,
+        message: str,
+        error_type: ErrorType,
+        step_name: str,
+        input_summary: str,
+        suggestion: str,
+        cause: Optional[Exception] = None
+    ):
+        super().__init__(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:
+        return {
+            "error_type": self.error_type.value,
+            "step_name": self.step_name,
+            "input_summary": self.input_summary,
+            "message": str(self),
+            "suggestion": self.suggestion
+        }
+
+class NetworkError(KBError):
+    """网络错误 (403/404/500/超时)"""
+    def __init__(self, step_name: str, input_summary: str, status_code: int = None, **kwargs):
+        suggestion = f"检查网络连接，状态码: {status_code}" if status_code else "检查网络连接"
+        super().__init__(
+            message=f"网络请求失败: {status_code or 'unknown'}",
+            error_type=ErrorType.NETWORK,
+            step_name=step_name,
+            input_summary=input_summary,
+            suggestion=suggestion,
+            **kwargs
+        )
+        self.status_code = status_code
+```
+
+### Anti-Patterns to Avoid
+- **直接写入状态文件无原子保证**: 崩溃时可能导致文件损坏或内容丢失
+- **散落的配置读取**: `os.environ.get()` 分散在多个文件，难以追踪和测试
+- **裸 Exception 捕获**: 丢失错误类型信息，无法实现针对性恢复
+- **Schema 定义分散**: 每个子技能自己定义格式，难以统一验证
+
+## Don't Hand-Roll
+
+| Problem | Don't Build | Use Instead | Why |
+|---------|-------------|-------------|-----|
+| 配置管理 | 手写 env 文件解析 | pydantic-settings | 自动类型转换、验证、环境变量映射 |
+| 数据验证 | 手写 if-else 检查 | Pydantic BaseModel | 类型安全、自动序列化、JSON Schema 生成 |
+| 重试逻辑 | while 循环 + sleep | tenacity | 成熟的指数退避、钩子、停止条件 |
+| 原子写入 | 手写临时文件逻辑 | 参考 state_manager.py 模式 | 边界情况处理（权限、磁盘满、崩溃） |
+
+**Key insight:** 配置和验证是 Python 生态中已解决的问题，使用 Pydantic 生态可以获得类型安全、IDE 支持和自动文档生成。
+
+## Common Pitfalls
+
+### Pitfall 1: 状态文件损坏
+**What goes wrong:** 写入状态文件时崩溃，导致文件内容不完整或为空
+**Why it happens:** 直接写入目标文件，没有使用原子替换
+**How to avoid:** 使用 write-to-temp-then-rename 模式，确保写入是原子的
+**Warning signs:** workflow_state.json 为空或包含截断的 JSON
+
+### Pitfall 2: 配置分散难以调试
+**What goes wrong:** 环境变量散落在多个文件，不知道哪个变量在哪使用
+**Why it happens:** 每个模块直接调用 `os.environ.get()`
+**How to avoid:** 使用 pydantic-settings 集中定义所有配置，启动时一次性验证
+**Warning signs:** 同一变量在多处读取，默认值不一致
+
+### Pitfall 3: Schema 演进导致兼容性问题
+**What goes wrong:** 添加新字段后旧的状态文件无法加载
+**Why it happens:** Schema 变更没有向后兼容性考虑
+**How to avoid:**
+- 使用 Optional 类型和默认值
+- 使用 schema_version 字段标记版本
+- load_state 时处理旧版本数据
+
+### Pitfall 4: 错误信息缺少上下文
+**What goes wrong:** 错误只有 "Network error"，不知道是哪个步骤、哪个输入
+**Why it happens:** 只抛出简单的 Exception 或使用通用错误消息
+**How to avoid:** 使用结构化错误类，强制包含 step_name, input_summary, suggestion
+
+## Code Examples
+
+### WorkflowState 完整定义
+
+```python
+# Source: 基于 gsd-openclaw/state_manager.py 和 CONTEXT.md 决策
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, List
+from datetime import datetime
+from enum import Enum
+
+class StepStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+class StepRecord(BaseModel):
+    """单个步骤的执行记录"""
+    step_name: str
+    status: StepStatus
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+    error_message: Optional[str] = None
+    output_summary: Optional[str] = None
+
+class WorkflowState(BaseModel):
+    """工作流状态 - 存储在 workflow_state.json"""
+    schema_version: str = Field(default="1.0")
+    content_id: str = Field(..., description="内容唯一标识")
+    platform: str = Field(..., description="平台名称")
+    source_url: str = Field(..., description="原始链接")
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+    steps: Dict[str, StepRecord] = Field(default_factory=dict)
+
+    def is_step_completed(self, step_name: str) -> bool:
+        """检查步骤是否已完成"""
+        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:
+        """标记步骤开始"""
+        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: str = None) -> None:
+        """标记步骤完成"""
+        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:
+        """标记步骤失败"""
+        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()
+```
+
+### WorkflowStateManager
+
+```python
+# Source: 基于 gsd-openclaw/state_manager.py
+import json
+from pathlib import Path
+from typing import Optional
+from .schemas import WorkflowState
+
+class WorkflowStateManager:
+    """工作流状态管理器"""
+
+    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
+
+    def load(self) -> Optional[WorkflowState]:
+        """加载状态，如果不存在返回 None"""
+        if not self.state_path.exists():
+            # 尝试加载备份
+            backup_path = self.state_path.with_suffix('.json.bak')
+            if backup_path.exists():
+                try:
+                    with open(backup_path, 'r', encoding='utf-8') as f:
+                        data = json.load(f)
+                    return WorkflowState.model_validate(data)
+                except (json.JSONDecodeError, ValueError):
+                    pass
+            return None
+
+        try:
+            with open(self.state_path, 'r', encoding='utf-8') as f:
+                data = json.load(f)
+            return WorkflowState.model_validate(data)
+        except (json.JSONDecodeError, ValueError) as e:
+            # 主文件损坏，尝试备份
+            backup_path = self.state_path.with_suffix('.json.bak')
+            if backup_path.exists():
+                with open(backup_path, 'r', encoding='utf-8') as f:
+                    data = json.load(f)
+                return WorkflowState.model_validate(data)
+            raise
+
+    def save(self, state: WorkflowState) -> None:
+        """原子保存状态"""
+        import os
+        import tempfile
+
+        backup_path = self.state_path.with_suffix('.json.bak')
+
+        # 确保目录存在
+        self.content_dir.mkdir(parents=True, exist_ok=True)
+
+        # 轮转备份
+        if self.state_path.exists():
+            if backup_path.exists():
+                backup_path.unlink()
+            self.state_path.rename(backup_path)
+
+        # 原子写入
+        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))
+            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:
+        """获取现有状态或创建新状态"""
+        state = self.load()
+        if state is None:
+            state = WorkflowState(
+                content_id=content_id,
+                platform=platform,
+                source_url=source_url
+            )
+        return state
+```
+
+### Retry with Exponential Backoff
+
+```python
+# Source: tenacity 最佳实践
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_exponential,
+    retry_if_exception_type,
+    before_sleep_log
+)
+import logging
+
+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
+)
+
+# 使用
+@retry_on_network_error
+def fetch_with_retry(url: str) -> bytes:
+    """带重试的网络请求"""
+    import requests
+    try:
+        response = requests.get(url, timeout=30)
+        response.raise_for_status()
+        return response.content
+    except requests.RequestException as e:
+        raise NetworkError(
+            step_name="fetch",
+            input_summary=f"url={url}",
+            status_code=getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None,
+            cause=e
+        )
+```
+
+## State of the Art
+
+| Old Approach | Current Approach | When Changed | Impact |
+|--------------|------------------|--------------|--------|
+| Pydantic v1 Config class | Pydantic v2 model_config | Pydantic v2 (2023) | 更清晰的配置语法，性能提升 |
+| 手写 env 文件解析 | pydantic-settings | pydantic-settings 2.0 (2023) | 自动验证、类型转换、嵌套模型支持 |
+| 手写重试逻辑 | tenacity 库 | 成熟实践 | 指数退避、钩子、条件重试 |
+| dataclass + 手动验证 | Pydantic BaseModel | Pydantic v2 (2023) | 性能提升 5-50x，更丰富的验证器 |
+
+**Deprecated/outdated:**
+- Pydantic v1 `class Config:` 语法: 使用 `model_config = ConfigDict(...)` 替代
+- python-dotenv 单独使用: pydantic-settings 已内置 dotenv 支持
+
+## Open Questions
+
+1. **降级链如何配置化？**
+   - What we know: 需要支持 "官方字幕 → Whisper → FunASR" 这样的降级链
+   - What's unclear: 降级链是硬编码还是可配置？配置格式是什么？
+   - Recommendation: Phase 1 实现简单的 FallbackChain 类，降级顺序通过代码定义；Phase 2+ 考虑配置文件支持
+
+2. **状态文件的并发访问？**
+   - What we know: 当前设计假设单进程执行
+   - What's unclear: 是否需要支持多进程同时处理不同内容？
+   - Recommendation: Phase 1 不处理并发，使用文件锁是后续优化点
+
+## Validation Architecture
+
+### Test Framework
+| Property | Value |
+|----------|-------|
+| Framework | pytest 7.x |
+| Config file | pyproject.toml (参考 gsd-openclaw) |
+| Quick run command | `pytest tests/test_pipeline/ -v --tb=short` |
+| Full suite command | `pytest tests/ -v --tb=short` |
+
+### Phase Requirements → Test Map
+| Req ID | Behavior | Test Type | Automated Command | File Exists? |
+|--------|----------|-----------|-------------------|-------------|
+| PIPE-01 | StepInput/StepOutput Schema 定义 | unit | `pytest tests/test_pipeline/test_schemas.py -v` | Wave 0 |
+| PIPE-02 | 原子状态保存和备份轮转 | unit | `pytest tests/test_pipeline/test_state.py::TestAtomicWrite -v` | Wave 0 |
+| PIPE-03 | JSON Schema 导出和验证 | unit | `pytest tests/test_pipeline/test_schemas.py::TestJSONSchema -v` | Wave 0 |
+| REL-01 | is_step_completed 幂等检查 | unit | `pytest tests/test_pipeline/test_state.py::TestIdempotency -v` | Wave 0 |
+| REL-02 | 网络错误重试 | unit | `pytest tests/test_pipeline/test_retry.py -v` | Wave 0 |
+| REL-03 | StepDefinition 验证 | unit | `pytest tests/test_pipeline/test_schemas.py::TestStepDefinition -v` | Wave 0 |
+| VAL-01 | Pydantic Schema 验证 | unit | `pytest tests/test_pipeline/test_schemas.py::TestValidation -v` | Wave 0 |
+| MNT-02 | pydantic-settings 配置加载 | unit | `pytest tests/test_pipeline/test_config.py -v` | Wave 0 |
+| MNT-03 | 结构化错误上下文 | unit | `pytest tests/test_pipeline/test_errors.py -v` | Wave 0 |
+
+### Sampling Rate
+- **Per task commit:** `pytest tests/test_pipeline/ -v --tb=short -x`
+- **Per wave merge:** `pytest tests/ -v --tb=short`
+- **Phase gate:** Full suite green before `/gsd:verify-work`
+
+### Wave 0 Gaps
+- [ ] `tests/test_pipeline/test_state.py` — 测试 WorkflowStateManager
+- [ ] `tests/test_pipeline/test_schemas.py` — 测试 StepInput/StepOutput/StepResult
+- [ ] `tests/test_pipeline/test_errors.py` — 测试 KBError 层次结构
+- [ ] `tests/test_pipeline/test_config.py` — 测试 KBSettings
+- [ ] `tests/test_pipeline/test_retry.py` — 测试重试逻辑
+- [ ] `tests/test_pipeline/conftest.py` — 共享 fixtures
+- [ ] `kb_pipeline/__init__.py` — 包初始化
+
+## Sources
+
+### Primary (HIGH confidence)
+- [Pydantic Models Documentation](https://docs.pydantic.dev/latest/concepts/models/) - BaseModel, validation, serialization
+- [Pydantic Settings Documentation](https://docs.pydantic.dev/latest/concepts/pydantic_settings/) - BaseSettings, env_file, secrets
+- [gsd-openclaw/state_manager.py](/home/starmen1265/work/kb-skills/gsd-openclaw/src/gsd_openclaw/state_manager.py) - 原子写入、备份轮转参考实现
+- [gsd-openclaw/tests/test_state_manager.py](/home/starmen1265/work/kb-skills/gsd-openclaw/tests/test_state_manager.py) - 状态管理测试模式参考
+
+### Secondary (MEDIUM confidence)
+- [kb-workflow/kb_workflow.py](/home/starmen1265/work/kb-skills/kb-workflow/kb_workflow.py) - 现有目录管理和查重逻辑
+
+### Tertiary (LOW confidence)
+- 无 — 所有关键信息已从 Primary 源验证
+
+## Metadata
+
+**Confidence breakdown:**
+- Standard stack: HIGH - Pydantic 是 Python 生态成熟方案，文档完整
+- Architecture: HIGH - 复用 gsd-openclaw 已验证的模式，符合项目决策
+- Pitfalls: HIGH - 基于实际代码审查和已知问题
+
+**Research date:** 2026-03-13
+**Valid until:** 2026-04-13 (30 days - Pydantic 生态稳定)