docs(01): create phase 1 plans - State Manager, Schema Registry, Error Handler & Config

.planning/ROADMAP.md

@@ -29,12 +29,12 @@ Decimal phases appear between their surrounding integers in numeric order.
   3. 每个步骤输出通过 Pydantic Schema 验证，格式错误阻止进入下一步
   4. 所有环境变量集中在 ~/.kb_env，启动时验证必需变量
   5. 错误信息包含步骤名、输入摘要、失败原因、可操作建议
-**Plans**: TBD
+**Plans:** 3 plans
 
 Plans:
-- [ ] 01-01: State Manager - workflow_state.json 管理
-- [ ] 01-02: Schema Registry - 统一输入输出格式定义
-- [ ] 01-03: Error Handler & Config - 结构化错误处理和统一配置
+- [ ] 01-01-PLAN.md — State Manager: workflow_state.json 管理 (PIPE-02, REL-01, REL-03)
+- [ ] 01-02-PLAN.md — Schema Registry: 统一输入输出格式定义 (PIPE-01, PIPE-03, VAL-01, REL-03)
+- [ ] 01-03-PLAN.md — Error Handler & Config: 结构化错误处理和统一配置 (REL-02, MNT-02, MNT-03)
 
 ### Phase 2: Platform Skills
 **Goal**: 集成B站和知乎平台的内容获取能力，形成从链接到原始内容的完整流程
@@ -93,11 +93,12 @@ Phases execute in numeric order: 1 → 2 → 3 → 4
 
 | Phase | Plans Complete | Status | Completed |
 |-------|----------------|--------|-----------|
-| 1. Foundation | 0/3 | Not started | - |
+| 1. Foundation | 0/3 | Ready to execute | - |
 | 2. Platform Skills | 0/3 | Not started | - |
 | 3. Processing Skills | 0/2 | Not started | - |
 | 4. Analysis & Orchestration | 0/3 | Not started | - |
 
 ---
 *Roadmap created: 2026-03-13*
+*Last updated: 2026-03-13 - Phase 1 plans created*
 *Granularity: coarse (4 phases)*

.planning/phases/01-foundation/01-01-PLAN.md

@@ -0,0 +1,224 @@
+---
+phase: 01-foundation
+plan: 01
+type: tdd
+wave: 2
+depends_on: [01-02]
+files_modified: [kb_pipeline/state.py, tests/test_pipeline/test_state.py]
+autonomous: true
+requirements: [PIPE-02, REL-01, REL-03]
+user_setup: []
+
+must_haves:
+  truths:
+    - "执行任意步骤后，workflow_state.json 正确记录该步骤的完成状态"
+    - "中断后重新执行，已完成的步骤被跳过，从中断点继续"
+    - "状态文件崩溃后可从备份恢复"
+  artifacts:
+    - path: "kb_pipeline/state.py"
+      provides: "WorkflowState 和 WorkflowStateManager"
+      min_lines: 80
+      exports: ["WorkflowState", "StepRecord", "StepStatus", "WorkflowStateManager"]
+    - path: "tests/test_pipeline/test_state.py"
+      provides: "状态管理单元测试"
+      min_lines: 100
+  key_links:
+    - from: "WorkflowStateManager.save()"
+      to: "workflow_state.json"
+      via: "原子写入 + 备份轮转"
+      pattern: "tempfile.mkstemp.*os.replace"
+    - from: "WorkflowState.is_step_completed()"
+      to: "幂等执行检查"
+      via: "步骤状态查询"
+      pattern: "status == StepStatus.COMPLETED"
+---
+
+<objective>
+实现 WorkflowState 和 WorkflowStateManager，提供步骤级状态持久化、幂等执行检查和原子写入保障。
+
+Purpose: 为工作流提供可靠的状态管理，支持断点续传和崩溃恢复
+Output: kb_pipeline/state.py + 完整的单元测试
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+
+<interfaces>
+<!-- Key types from 01-02 Schema Registry that this plan depends on -->
+<!-- These will be created by 01-02 before this plan runs -->
+
+From kb_pipeline/schemas.py (created by 01-02):
+```python
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, Literal
+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)
+```
+
+Reference implementation from gsd-openclaw:
+- Atomic write pattern: write-to-temp + os.replace
+- Backup rotation: state.json -> state.json.bak
+- Load fallback: main file corrupted -> try backup
+</interfaces>
+</context>
+
+<feature>
+  <name>WorkflowStateManager</name>
+  <files>kb_pipeline/state.py, tests/test_pipeline/test_state.py</files>
+  <behavior>
+    - WorkflowState.is_step_completed(step_name) 返回 True 当且仅当步骤状态为 COMPLETED
+    - WorkflowState.mark_step_completed() 正确更新步骤状态和时间戳
+    - WorkflowStateManager.save() 使用原子写入，崩溃不损坏文件
+    - WorkflowStateManager.load() 主文件损坏时从备份恢复
+    - WorkflowStateManager.get_or_create() 返回现有状态或创建新状态
+  </behavior>
+  <implementation>
+    1. 在 state.py 中导入 schemas 模块的 WorkflowState, StepRecord, StepStatus
+    2. 实现 WorkflowStateManager 类:
+       - __init__(content_dir: Path): 初始化内容目录和状态文件路径
+       - load() -> Optional[WorkflowState]: 加载状态，失败时尝试备份
+       - save(state: WorkflowState): 原子保存，备份轮转
+       - get_or_create(content_id, platform, source_url): 获取或创建状态
+    3. 参考 gsd-openclaw/state_manager.py 的原子写入模式
+  </implementation>
+</feature>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: Write failing tests for WorkflowStateManager</name>
+  <files>tests/test_pipeline/test_state.py</files>
+  <behavior>
+    - Test is_step_completed returns True for completed step
+    - Test is_step_completed returns False for pending/running/failed step
+    - Test mark_step_completed updates status and timestamp
+    - Test save creates file atomically with backup rotation
+    - Test load falls back to backup on corruption
+    - Test get_or_create returns existing or creates new
+  </behavior>
+  <action>
+    创建 tests/test_pipeline/test_state.py 测试文件，包含以下测试类:
+    - TestWorkflowStateIsStepCompleted: 测试 is_step_completed 方法
+    - TestWorkflowStateMarkStep: 测试 mark_step_started/completed/failed 方法
+    - TestAtomicWrite: 测试原子写入和备份轮转
+    - TestBackupFallback: 测试主文件损坏时的备份恢复
+    - TestStateManagerGetOrCreate: 测试 get_or_create 方法
+
+    参考 gsd-openclaw/tests/test_state_manager.py 的测试模式。
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_state.py -v --collect-only 2>/dev/null | grep "test_" | wc -l | xargs -I {} test {} -ge 10</automated>
+  </verify>
+  <done>测试文件包含至少 10 个测试用例，所有测试当前失败（RED 阶段）</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 2: Implement WorkflowStateManager</name>
+  <files>kb_pipeline/state.py</files>
+  <behavior>
+    - WorkflowStateManager.__init__ 正确设置 content_dir 和 state_path
+    - WorkflowStateManager.load() 返回 WorkflowState 或 None
+    - WorkflowStateManager.save() 使用原子写入模式
+    - WorkflowStateManager.get_or_create() 幂等创建状态
+  </behavior>
+  <action>
+    实现 kb_pipeline/state.py:
+    1. 导入必要的模块 (json, os, tempfile, Path)
+    2. 从 kb_pipeline.schemas 导入 WorkflowState
+    3. 实现 WorkflowStateManager 类:
+       - STATE_FILENAME = "workflow_state.json"
+       - load(): 尝试加载主文件，失败时尝试 .bak 文件
+       - save(state): 原子写入 - 创建父目录 -> 轮转备份 -> 写临时文件 -> os.replace
+       - get_or_create(content_id, platform, source_url): 加载现有或创建新状态
+
+    原子写入模式 (参考 gsd-openclaw):
+    ```python
+    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
+    ```
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_state.py -v --tb=short</automated>
+  </verify>
+  <done>所有测试通过（GREEN 阶段）</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 3: Refactor and verify integration</name>
+  <files>kb_pipeline/state.py, tests/test_pipeline/test_state.py</files>
+  <behavior>
+    - 代码结构清晰，无重复
+    - 测试覆盖所有边界情况
+    - 与 schemas.py 正确集成
+  </behavior>
+  <action>
+    1. 检查代码质量: 移除重复代码，添加必要的类型注解
+    2. 添加文档字符串（如果缺失）
+    3. 确保与 kb_pipeline/schemas.py 的 WorkflowState 正确集成
+    4. 运行完整测试套件验证
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_state.py -v --tb=short -x</automated>
+  </verify>
+  <done>测试通过，代码结构清晰，与 schemas 模块正确集成</done>
+</task>
+
+</tasks>
+
+<verification>
+1. 运行 `pytest tests/test_pipeline/test_state.py -v` 全部通过
+2. 验证原子写入: 创建状态文件后，确认存在 workflow_state.json
+3. 验证备份轮转: 保存两次后，确认存在 .bak 文件
+4. 验证幂等检查: is_step_completed 对已完成步骤返回 True
+</verification>
+
+<success_criteria>
+- WorkflowStateManager 可以原子方式保存和加载状态
+- 状态文件损坏时可从备份恢复
+- is_step_completed 支持幂等执行检查
+- 所有测试通过，覆盖率 > 80%
+</success_criteria>
+
+<output>
+After completion, create `.planning/phases/01-foundation/01-01-SUMMARY.md`
+</output>

.planning/phases/01-foundation/01-02-PLAN.md

@@ -0,0 +1,319 @@
+---
+phase: 01-foundation
+plan: 02
+type: tdd
+wave: 1
+depends_on: []
+files_modified: [kb_pipeline/__init__.py, kb_pipeline/schemas.py, tests/test_pipeline/__init__.py, tests/test_pipeline/conftest.py, tests/test_pipeline/test_schemas.py]
+autonomous: true
+requirements: [PIPE-01, PIPE-03, VAL-01, REL-03]
+user_setup: []
+
+must_haves:
+  truths:
+    - "每个步骤输出通过 Pydantic Schema 验证，格式错误阻止进入下一步"
+    - "所有步骤的 Schema 集中在一个模块，便于维护"
+    - "步骤执行前验证输入，执行后验证输出"
+  artifacts:
+    - path: "kb_pipeline/schemas.py"
+      provides: "核心 Schema 定义"
+      min_lines: 100
+      exports: ["StepInput", "StepOutput", "StepResult", "StepStatus", "StepRecord", "WorkflowState", "StepDefinition"]
+    - path: "kb_pipeline/__init__.py"
+      provides: "包初始化和公共接口导出"
+    - path: "tests/test_pipeline/test_schemas.py"
+      provides: "Schema 验证测试"
+      min_lines: 80
+  key_links:
+    - from: "StepOutput"
+      to: "Pydantic BaseModel"
+      via: "继承和验证"
+      pattern: "class StepOutput.*BaseModel"
+    - from: "model_validate()"
+      to: "Schema 验证"
+      via: "Pydantic 验证方法"
+      pattern: "model_validate|ValidationError"
+---
+
+<objective>
+定义工作流管道的核心 Schema，包括 StepInput、StepOutput、StepResult、WorkflowState 等，使用 Pydantic v2 确保类型安全和自动验证。
+
+Purpose: 建立统一的数据格式规范，所有后续步骤和状态管理都基于这些 Schema
+Output: kb_pipeline/schemas.py + 完整的单元测试
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+
+<interfaces>
+<!-- No external dependencies - this is the foundation layer -->
+<!-- Following patterns from RESEARCH.md and CONTEXT.md decisions -->
+
+From CONTEXT.md decisions:
+- Schema 格式: Pydantic 模型 — 自带验证和序列化，类型安全
+- 验证范围: 验证输入和输出 — 步骤执行前验证输入，执行后验证输出
+- Schema 定义位置: 集中定义 — 所有步骤的 Schema 集中在一个模块
+
+From RESEARCH.md Pattern 2: Pydantic Schema Registry:
+```python
+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)
+```
+</interfaces>
+</context>
+
+<feature>
+  <name>Core Pipeline Schemas</name>
+  <files>kb_pipeline/schemas.py, tests/test_pipeline/test_schemas.py</files>
+  <behavior>
+    - StepInput 定义所有步骤的公共输入字段 (content_id, platform)
+    - StepOutput 定义所有步骤的公共输出字段 (success, error_message, completed_at)
+    - StepStatus 枚举定义步骤状态 (PENDING, RUNNING, COMPLETED, FAILED, SKIPPED)
+    - StepRecord 记录单个步骤的执行状态
+    - WorkflowState 管理整个工作流的状态
+    - StepDefinition 定义步骤的前置条件、输出格式、成功标准
+    - 所有 Schema 支持 Pydantic model_validate() 验证
+  </behavior>
+  <implementation>
+    1. 创建 kb_pipeline/ 目录和 __init__.py
+    2. 创建 tests/test_pipeline/ 目录和 conftest.py
+    3. 在 schemas.py 中定义所有 Schema:
+       - StepStatus (Enum)
+       - StepInput (BaseModel)
+       - StepOutput (BaseModel)
+       - StepResult (BaseModel) - 包含输入输出和状态
+       - StepRecord (BaseModel) - 步骤执行记录
+       - WorkflowState (BaseModel) - 工作流状态
+       - StepDefinition (BaseModel) - 步骤定义
+  </implementation>
+</feature>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: Create package structure and write failing tests</name>
+  <files>kb_pipeline/__init__.py, tests/test_pipeline/__init__.py, tests/test_pipeline/conftest.py, tests/test_pipeline/test_schemas.py</files>
+  <behavior>
+    - 测试 StepInput 必须包含 content_id 和 platform
+    - 测试 StepOutput 必须包含 success 字段
+    - 测试 StepStatus 包含所有必要状态
+    - 测试 WorkflowState 包含 steps 字典
+    - 测试 Pydantic 验证失败时抛出 ValidationError
+  </behavior>
+  <action>
+    1. 创建目录结构:
+       - mkdir -p kb_pipeline tests/test_pipeline
+
+    2. 创建 kb_pipeline/__init__.py:
+       ```python
+       """KB Pipeline - 工作流管道基础设施"""
+       from .schemas import (
+           StepInput, StepOutput, StepResult,
+           StepStatus, StepRecord, WorkflowState, StepDefinition
+       )
+       __all__ = [...]
+       ```
+
+    3. 创建 tests/test_pipeline/__init__.py (空文件)
+
+    4. 创建 tests/test_pipeline/conftest.py:
+       - 添加通用 fixtures (tmp_content_dir, sample_workflow_state)
+
+    5. 创建 tests/test_pipeline/test_schemas.py，包含测试类:
+       - TestStepInput: 测试必填字段和验证
+       - TestStepOutput: 测试输出字段和默认值
+       - TestStepStatus: 测试枚举值
+       - TestStepRecord: 测试步骤记录
+       - TestWorkflowState: 测试工作流状态
+       - TestSchemaValidation: 测试 Pydantic 验证
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_schemas.py -v --collect-only 2>/dev/null | grep "test_" | wc -l | xargs -I {} test {} -ge 12</automated>
+  </verify>
+  <done>测试文件包含至少 12 个测试用例，所有测试当前失败（RED 阶段）</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 2: Implement core schemas</name>
+  <files>kb_pipeline/schemas.py</files>
+  <behavior>
+    - StepInput 验证 content_id 非空
+    - StepOutput 默认 completed_at 为当前时间
+    - StepStatus 包含 PENDING, RUNNING, COMPLETED, FAILED, SKIPPED
+    - WorkflowState.steps 是 Dict[str, StepRecord]
+    - 所有模型支持 model_dump_json() 序列化
+  </behavior>
+  <action>
+    实现 kb_pipeline/schemas.py:
+
+    ```python
+    from pydantic import BaseModel, Field
+    from typing import Optional, Dict, Literal, Any
+    from datetime import datetime
+    from enum import Enum
+
+    class StepStatus(str, Enum):
+        """步骤执行状态"""
+        PENDING = "pending"
+        RUNNING = "running"
+        COMPLETED = "completed"
+        FAILED = "failed"
+        SKIPPED = "skipped"
+
+    class StepInput(BaseModel):
+        """所有步骤输入的基类"""
+        content_id: str = Field(..., min_length=1, description="内容唯一标识")
+        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 StepResult(BaseModel):
+        """步骤执行结果"""
+        step_name: str
+        input_data: StepInput
+        output_data: Optional[StepOutput] = None
+        status: StepStatus = StepStatus.PENDING
+
+    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()
+
+    class StepDefinition(BaseModel):
+        """步骤定义 - 描述步骤的前置条件、输出格式、成功标准"""
+        step_name: str
+        description: str
+        input_schema: type[StepInput]
+        output_schema: type[StepOutput]
+        dependencies: list[str] = Field(default_factory=list)
+    ```
+
+    注意: WorkflowState 的方法实现先写在 schemas.py 中，后续会考虑移到 state.py
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_schemas.py -v --tb=short</automated>
+  </verify>
+  <done>所有测试通过（GREEN 阶段）</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 3: Refactor and verify JSON serialization</name>
+  <files>kb_pipeline/schemas.py, tests/test_pipeline/test_schemas.py</files>
+  <behavior>
+    - model_dump_json() 生成有效的 JSON
+    - model_validate() 从 JSON 正确反序列化
+    - JSON Schema 可导出 (用于文档)
+  </behavior>
+  <action>
+    1. 添加 JSON 序列化测试:
+       - TestJSONSerialization: 测试 model_dump_json 和 model_validate
+
+    2. 验证所有 Schema 支持:
+       - model_dump() -> dict
+       - model_dump_json() -> str
+       - model_validate(data) -> instance
+
+    3. 确保 WorkflowState 的 steps 字典正确序列化/反序列化
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_schemas.py -v --tb=short -x</automated>
+  </verify>
+  <done>所有测试通过，JSON 序列化正常工作</done>
+</task>
+
+</tasks>
+
+<verification>
+1. 运行 `pytest tests/test_pipeline/test_schemas.py -v` 全部通过
+2. 验证 StepInput 缺少 content_id 时抛出 ValidationError
+3. 验证 WorkflowState.model_dump_json() 生成有效 JSON
+4. 验证 WorkflowState.model_validate() 可从 JSON 恢复
+</verification>
+
+<success_criteria>
+- 所有核心 Schema 定义完成
+- Pydantic 验证正常工作
+- JSON 序列化/反序列化正常
+- 测试覆盖率 > 80%
+</success_criteria>
+
+<output>
+After completion, create `.planning/phases/01-foundation/01-02-SUMMARY.md`
+</output>

.planning/phases/01-foundation/01-03-PLAN.md

@@ -0,0 +1,498 @@
+---
+phase: 01-foundation
+plan: 03
+type: tdd
+wave: 2
+depends_on: [01-02]
+files_modified: [kb_pipeline/errors.py, kb_pipeline/config.py, kb_pipeline/retry.py, tests/test_pipeline/test_errors.py, tests/test_pipeline/test_config.py, tests/test_pipeline/test_retry.py]
+autonomous: true
+requirements: [REL-02, MNT-02, MNT-03]
+user_setup:
+  - service: pydantic-settings
+    why: "配置管理需要读取 ~/.kb_env 文件"
+    env_vars:
+      - name: KB_BASE_PATH
+        source: "~/.kb_env 文件中定义"
+      - name: KB_ARCHIVE_PATH
+        source: "~/.kb_env 文件中定义"
+    dashboard_config: []
+
+must_haves:
+  truths:
+    - "所有环境变量集中在 ~/.kb_env，启动时验证必需变量"
+    - "错误信息包含步骤名、输入摘要、失败原因、可操作建议"
+    - "网络错误自动重试3次，指数退避"
+  artifacts:
+    - path: "kb_pipeline/errors.py"
+      provides: "结构化错误类型"
+      min_lines: 60
+      exports: ["KBError", "NetworkError", "ValidationError", "SystemError", "AuthError", "ErrorType"]
+    - path: "kb_pipeline/config.py"
+      provides: "统一配置管理"
+      min_lines: 40
+      exports: ["KBSettings", "get_settings"]
+    - path: "kb_pipeline/retry.py"
+      provides: "重试装饰器"
+      min_lines: 30
+      exports: ["retry_on_network_error"]
+    - path: "tests/test_pipeline/test_errors.py"
+      provides: "错误处理测试"
+    - path: "tests/test_pipeline/test_config.py"
+      provides: "配置管理测试"
+    - path: "tests/test_pipeline/test_retry.py"
+      provides: "重试逻辑测试"
+  key_links:
+    - from: "KBSettings"
+      to: "~/.kb_env"
+      via: "pydantic-settings env_file"
+      pattern: "SettingsConfigDict.*env_file"
+    - from: "KBError.to_dict()"
+      to: "错误信息输出"
+      via: "结构化序列化"
+      pattern: "step_name.*input_summary.*suggestion"
+    - from: "retry_on_network_error"
+      to: "网络请求"
+      via: "装饰器"
+      pattern: "@retry.*stop_after_attempt.*3"
+---
+
+<objective>
+实现结构化错误处理、统一配置管理和网络重试机制，为工作流提供可靠的错误恢复和配置验证能力。
+
+Purpose: 确保错误信息包含足够上下文便于调试，配置集中管理便于维护，网络错误自动恢复
+Output: kb_pipeline/errors.py, config.py, retry.py + 完整的单元测试
+</objective>
+
+<execution_context>
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+</execution_context>
+
+<context>
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+
+<interfaces>
+<!-- Dependencies from 01-02 schemas -->
+
+From kb_pipeline/schemas.py:
+```python
+class StepStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+```
+
+From CONTEXT.md decisions:
+- 错误分类: 网络错误 (403/404/500/超时), 验证错误, 系统错误, 认证/配额错误
+- 重试策略: 网络错误自动重试3次，指数退避
+- 错误信息: 步骤名、输入摘要、错误类型、失败原因、可操作建议
+- 配置文件位置: ~/.kb_env
+- 配置库: pydantic-settings
+
+From RESEARCH.md Pattern 3 & 4:
+```python
+# pydantic-settings configuration
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+class KBSettings(BaseSettings):
+    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="归档目录")
+```
+
+```python
+# Structured error handling
+class ErrorType(str, Enum):
+    NETWORK = "network"
+    VALIDATION = "validation"
+    SYSTEM = "system"
+    AUTH = "auth"
+
+class KBError(Exception):
+    def __init__(self, message, error_type, step_name, input_summary, suggestion, cause=None):
+        ...
+```
+</interfaces>
+</context>
+
+<feature>
+  <name>Error Handler & Config Manager</name>
+  <files>kb_pipeline/errors.py, kb_pipeline/config.py, kb_pipeline/retry.py, tests/test_pipeline/test_errors.py, tests/test_pipeline/test_config.py, tests/test_pipeline/test_retry.py</files>
+  <behavior>
+    - KBError 包含 step_name, input_summary, error_type, message, suggestion
+    - NetworkError, ValidationError, SystemError, AuthError 继承自 KBError
+    - KBSettings 从 ~/.kb_env 加载配置，缺失必需变量时报错
+    - retry_on_network_error 装饰器自动重试网络错误3次
+    - 错误可序列化为 dict 用于日志和状态记录
+  </behavior>
+  <implementation>
+    1. 实现 kb_pipeline/errors.py:
+       - ErrorType 枚举
+       - KBError 基类 (包含 to_dict 方法)
+       - NetworkError, ValidationError, SystemError, AuthError 子类
+
+    2. 实现 kb_pipeline/config.py:
+       - KBSettings (pydantic-settings BaseSettings)
+       - get_settings() 单例函数
+
+    3. 实现 kb_pipeline/retry.py:
+       - retry_on_network_error 装饰器 (使用 tenacity)
+  </implementation>
+</feature>
+
+<tasks>
+
+<task type="auto" tdd="true">
+  <name>Task 1: Write failing tests for error handling</name>
+  <files>tests/test_pipeline/test_errors.py</files>
+  <behavior>
+    - Test KBError contains all required fields
+    - Test KBError.to_dict() returns correct dict
+    - Test NetworkError includes status_code
+    - Test each error type has appropriate suggestion
+  </behavior>
+  <action>
+    创建 tests/test_pipeline/test_errors.py:
+
+    ```python
+    import pytest
+    from kb_pipeline.errors import (
+        KBError, NetworkError, ValidationError, SystemError, AuthError, ErrorType
+    )
+
+    class TestKBError:
+        def test_kberror_contains_all_fields(self):
+            error = KBError(
+                message="Test error",
+                error_type=ErrorType.NETWORK,
+                step_name="test_step",
+                input_summary="input_id=123",
+                suggestion="Check your network"
+            )
+            assert error.step_name == "test_step"
+            assert error.input_summary == "input_id=123"
+            assert error.suggestion == "Check your network"
+
+        def test_kberror_to_dict(self):
+            error = KBError(...)
+            result = error.to_dict()
+            assert "step_name" in result
+            assert "error_type" in result
+            assert "suggestion" in result
+
+    class TestNetworkError:
+        def test_network_error_includes_status_code(self):
+            error = NetworkError(
+                step_name="fetch",
+                input_summary="url=http://example.com",
+                status_code=403
+            )
+            assert error.status_code == 403
+            assert "403" in error.suggestion
+
+    class TestErrorHierarchy:
+        def test_network_error_is_kberror(self):
+            assert issubclass(NetworkError, KBError)
+    ```
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_errors.py -v --collect-only 2>/dev/null | grep "test_" | wc -l | xargs -I {} test {} -ge 8</automated>
+  </verify>
+  <done>测试文件包含至少 8 个测试用例，所有测试当前失败（RED 阶段）</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 2: Implement error handling</name>
+  <files>kb_pipeline/errors.py</files>
+  <behavior>
+    - ErrorType enum with NETWORK, VALIDATION, SYSTEM, AUTH
+    - KBError base class with to_dict method
+    - NetworkError with status_code and auto-suggestion
+    - ValidationError for schema failures
+    - SystemError for file/memory/dependency issues
+    - AuthError for API quota/token issues
+  </behavior>
+  <action>
+    实现 kb_pipeline/errors.py:
+
+    ```python
+    from enum import Enum
+    from typing import Optional, Any
+
+    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, cause: Exception = None):
+            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,
+                cause=cause
+            )
+            self.status_code = status_code
+
+    class ValidationError(KBError):
+        """验证错误 (Schema 验证失败)"""
+        def __init__(self, step_name: str, input_summary: str, validation_errors: list = None, cause: 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):
+        """系统错误 (文件系统/内存/依赖缺失)"""
+        def __init__(self, step_name: str, input_summary: str, system_message: str = None, cause: 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
+            )
+
+    class AuthError(KBError):
+        """认证/配额错误 (API配额/Cookie过期/Token失效)"""
+        def __init__(self, step_name: str, input_summary: str, auth_type: str = None, cause: 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
+    ```
+
+    注意: SystemError 与 Python 内置同名，考虑命名空间或使用别名
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_errors.py -v --tb=short</automated>
+  </verify>
+  <done>所有测试通过（GREEN 阶段）</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 3: Write and implement configuration management</name>
+  <files>kb_pipeline/config.py, tests/test_pipeline/test_config.py</files>
+  <behavior>
+    - KBSettings loads from ~/.kb_env
+    - Required fields: base_path, archive_path
+    - Optional fields: git_root, git_user, git_email
+    - Secret fields: hf_token, zhihu_cookie (from env vars)
+    - get_settings() returns singleton
+    - Missing required field raises ValidationError
+  </behavior>
+  <action>
+    1. 创建 tests/test_pipeline/test_config.py:
+       - Test KBSettings loads from env file
+       - Test required fields validation
+       - Test get_settings returns singleton
+
+    2. 实现 kb_pipeline/config.py:
+
+    ```python
+    from pydantic import Field, SecretStr
+    from pydantic_settings import BaseSettings, SettingsConfigDict
+    from functools import lru_cache
+
+    class KBSettings(BaseSettings):
+        """KB 工作流配置"""
+        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="归档目录")
+
+        # 可选配置
+        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")
+
+    @lru_cache
+    def get_settings() -> KBSettings:
+        """获取配置单例"""
+        return KBSettings()
+    ```
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_config.py -v --tb=short</automated>
+  </verify>
+  <done>所有测试通过，配置管理正常工作</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 4: Write and implement retry mechanism</name>
+  <files>kb_pipeline/retry.py, tests/test_pipeline/test_retry.py</files>
+  <behavior>
+    - retry_on_network_error decorator retries 3 times
+    - Uses exponential backoff (2s, 4s, 8s max)
+    - Only retries on NetworkError
+    - Logs retry attempts
+    - Reraises after max attempts
+  </behavior>
+  <action>
+    1. 创建 tests/test_pipeline/test_retry.py:
+       - Test retry succeeds on second attempt
+       - Test retry exhausts attempts and raises
+       - Test non-NetworkError is not retried
+       - Test exponential backoff timing
+
+    2. 实现 kb_pipeline/retry.py:
+
+    ```python
+    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
+    )
+    ```
+  </action>
+  <verify>
+    <automated>pytest tests/test_pipeline/test_retry.py -v --tb=short</automated>
+  </verify>
+  <done>所有测试通过，重试机制正常工作</done>
+</task>
+
+<task type="auto" tdd="true">
+  <name>Task 5: Update package exports</name>
+  <files>kb_pipeline/__init__.py</files>
+  <behavior>
+    - All public APIs exported from package
+    - Clean import interface
+  </behavior>
+  <action>
+    更新 kb_pipeline/__init__.py 导出:
+
+    ```python
+    """KB Pipeline - 工作流管道基础设施"""
+
+    from .schemas import (
+        StepInput, StepOutput, StepResult,
+        StepStatus, StepRecord, WorkflowState, StepDefinition
+    )
+    from .state import WorkflowStateManager
+    from .errors import (
+        KBError, NetworkError, ValidationError,
+        SystemError, AuthError, ErrorType
+    )
+    from .config import KBSettings, get_settings
+    from .retry import retry_on_network_error
+
+    __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",
+    ]
+    ```
+  </action>
+  <verify>
+    <automated>python -c "from kb_pipeline import WorkflowStateManager, KBSettings, retry_on_network_error" && echo "OK"</automated>
+  </verify>
+  <done>所有公共 API 可从包根导入</done>
+</task>
+
+</tasks>
+
+<verification>
+1. 运行 `pytest tests/test_pipeline/test_errors.py -v` 全部通过
+2. 运行 `pytest tests/test_pipeline/test_config.py -v` 全部通过
+3. 运行 `pytest tests/test_pipeline/test_retry.py -v` 全部通过
+4. 验证错误 to_dict() 包含所有必要字段
+5. 验证配置缺失必需字段时抛出 ValidationError
+</verification>
+
+<success_criteria>
+- 结构化错误类型完整，包含上下文信息
+- 配置从 ~/.kb_env 加载，缺失必需字段报错
+- 网络错误自动重试，指数退避
+- 所有测试通过，覆盖率 > 80%
+</success_criteria>
+
+<output>
+After completion, create `.planning/phases/01-foundation/01-03-SUMMARY.md`
+</output>