diff --git a/.planning/ROADMAP.md b/.planning/ROADMAP.md
index 08c43c0..3bcbebb 100644
--- a/.planning/ROADMAP.md
+++ b/.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)*
diff --git a/.planning/phases/01-foundation/01-01-PLAN.md b/.planning/phases/01-foundation/01-01-PLAN.md
new file mode 100644
index 0000000..2e6c4a6
--- /dev/null
+++ b/.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"
+---
+
+
+实现 WorkflowState 和 WorkflowStateManager，提供步骤级状态持久化、幂等执行检查和原子写入保障。
+
+Purpose: 为工作流提供可靠的状态管理，支持断点续传和崩溃恢复
+Output: kb_pipeline/state.py + 完整的单元测试
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+
+
+
+
+
+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
+
+
+
+
+ WorkflowStateManager
+ kb_pipeline/state.py, tests/test_pipeline/test_state.py
+
+ - WorkflowState.is_step_completed(step_name) 返回 True 当且仅当步骤状态为 COMPLETED
+ - WorkflowState.mark_step_completed() 正确更新步骤状态和时间戳
+ - WorkflowStateManager.save() 使用原子写入，崩溃不损坏文件
+ - WorkflowStateManager.load() 主文件损坏时从备份恢复
+ - WorkflowStateManager.get_or_create() 返回现有状态或创建新状态
+
+
+ 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 的原子写入模式
+
+
+
+
+
+
+ Task 1: Write failing tests for WorkflowStateManager
+ tests/test_pipeline/test_state.py
+
+ - 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
+
+
+ 创建 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 的测试模式。
+
+
+ pytest tests/test_pipeline/test_state.py -v --collect-only 2>/dev/null | grep "test_" | wc -l | xargs -I {} test {} -ge 10
+
+ 测试文件包含至少 10 个测试用例，所有测试当前失败（RED 阶段）
+
+
+
+ Task 2: Implement WorkflowStateManager
+ kb_pipeline/state.py
+
+ - WorkflowStateManager.__init__ 正确设置 content_dir 和 state_path
+ - WorkflowStateManager.load() 返回 WorkflowState 或 None
+ - WorkflowStateManager.save() 使用原子写入模式
+ - WorkflowStateManager.get_or_create() 幂等创建状态
+
+
+ 实现 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
+ ```
+
+
+ pytest tests/test_pipeline/test_state.py -v --tb=short
+
+ 所有测试通过（GREEN 阶段）
+
+
+
+ Task 3: Refactor and verify integration
+ kb_pipeline/state.py, tests/test_pipeline/test_state.py
+
+ - 代码结构清晰，无重复
+ - 测试覆盖所有边界情况
+ - 与 schemas.py 正确集成
+
+
+ 1. 检查代码质量: 移除重复代码，添加必要的类型注解
+ 2. 添加文档字符串（如果缺失）
+ 3. 确保与 kb_pipeline/schemas.py 的 WorkflowState 正确集成
+ 4. 运行完整测试套件验证
+
+
+ pytest tests/test_pipeline/test_state.py -v --tb=short -x
+
+ 测试通过，代码结构清晰，与 schemas 模块正确集成
+
+
+
+
+
+1. 运行 `pytest tests/test_pipeline/test_state.py -v` 全部通过
+2. 验证原子写入: 创建状态文件后，确认存在 workflow_state.json
+3. 验证备份轮转: 保存两次后，确认存在 .bak 文件
+4. 验证幂等检查: is_step_completed 对已完成步骤返回 True
+
+
+
+- WorkflowStateManager 可以原子方式保存和加载状态
+- 状态文件损坏时可从备份恢复
+- is_step_completed 支持幂等执行检查
+- 所有测试通过，覆盖率 > 80%
+
+
+
diff --git a/.planning/phases/01-foundation/01-02-PLAN.md b/.planning/phases/01-foundation/01-02-PLAN.md
new file mode 100644
index 0000000..3b2755b
--- /dev/null
+++ b/.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"
+---
+
+
+定义工作流管道的核心 Schema，包括 StepInput、StepOutput、StepResult、WorkflowState 等，使用 Pydantic v2 确保类型安全和自动验证。
+
+Purpose: 建立统一的数据格式规范，所有后续步骤和状态管理都基于这些 Schema
+Output: kb_pipeline/schemas.py + 完整的单元测试
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+
+
+
+
+
+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)
+```
+
+
+
+
+ Core Pipeline Schemas
+ kb_pipeline/schemas.py, tests/test_pipeline/test_schemas.py
+
+ - StepInput 定义所有步骤的公共输入字段 (content_id, platform)
+ - StepOutput 定义所有步骤的公共输出字段 (success, error_message, completed_at)
+ - StepStatus 枚举定义步骤状态 (PENDING, RUNNING, COMPLETED, FAILED, SKIPPED)
+ - StepRecord 记录单个步骤的执行状态
+ - WorkflowState 管理整个工作流的状态
+ - StepDefinition 定义步骤的前置条件、输出格式、成功标准
+ - 所有 Schema 支持 Pydantic model_validate() 验证
+
+
+ 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) - 步骤定义
+
+
+
+
+
+
+ Task 1: Create package structure and write failing tests
+ kb_pipeline/__init__.py, tests/test_pipeline/__init__.py, tests/test_pipeline/conftest.py, tests/test_pipeline/test_schemas.py
+
+ - 测试 StepInput 必须包含 content_id 和 platform
+ - 测试 StepOutput 必须包含 success 字段
+ - 测试 StepStatus 包含所有必要状态
+ - 测试 WorkflowState 包含 steps 字典
+ - 测试 Pydantic 验证失败时抛出 ValidationError
+
+
+ 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 验证
+
+
+ pytest tests/test_pipeline/test_schemas.py -v --collect-only 2>/dev/null | grep "test_" | wc -l | xargs -I {} test {} -ge 12
+
+ 测试文件包含至少 12 个测试用例，所有测试当前失败（RED 阶段）
+
+
+
+ Task 2: Implement core schemas
+ kb_pipeline/schemas.py
+
+ - StepInput 验证 content_id 非空
+ - StepOutput 默认 completed_at 为当前时间
+ - StepStatus 包含 PENDING, RUNNING, COMPLETED, FAILED, SKIPPED
+ - WorkflowState.steps 是 Dict[str, StepRecord]
+ - 所有模型支持 model_dump_json() 序列化
+
+
+ 实现 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
+
+
+ pytest tests/test_pipeline/test_schemas.py -v --tb=short
+
+ 所有测试通过（GREEN 阶段）
+
+
+
+ Task 3: Refactor and verify JSON serialization
+ kb_pipeline/schemas.py, tests/test_pipeline/test_schemas.py
+
+ - model_dump_json() 生成有效的 JSON
+ - model_validate() 从 JSON 正确反序列化
+ - JSON Schema 可导出 (用于文档)
+
+
+ 1. 添加 JSON 序列化测试:
+ - TestJSONSerialization: 测试 model_dump_json 和 model_validate
+
+ 2. 验证所有 Schema 支持:
+ - model_dump() -> dict
+ - model_dump_json() -> str
+ - model_validate(data) -> instance
+
+ 3. 确保 WorkflowState 的 steps 字典正确序列化/反序列化
+
+
+ pytest tests/test_pipeline/test_schemas.py -v --tb=short -x
+
+ 所有测试通过，JSON 序列化正常工作
+
+
+
+
+
+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 恢复
+
+
+
+- 所有核心 Schema 定义完成
+- Pydantic 验证正常工作
+- JSON 序列化/反序列化正常
+- 测试覆盖率 > 80%
+
+
+
diff --git a/.planning/phases/01-foundation/01-03-PLAN.md b/.planning/phases/01-foundation/01-03-PLAN.md
new file mode 100644
index 0000000..eb75eef
--- /dev/null
+++ b/.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"
+---
+
+
+实现结构化错误处理、统一配置管理和网络重试机制，为工作流提供可靠的错误恢复和配置验证能力。
+
+Purpose: 确保错误信息包含足够上下文便于调试，配置集中管理便于维护，网络错误自动恢复
+Output: kb_pipeline/errors.py, config.py, retry.py + 完整的单元测试
+
+
+
+@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
+@/home/starmen1265/.claude/get-shit-done/templates/summary.md
+
+
+
+@.planning/PROJECT.md
+@.planning/ROADMAP.md
+@.planning/STATE.md
+
+
+
+
+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):
+ ...
+```
+
+
+
+
+ Error Handler & Config Manager
+ 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
+
+ - KBError 包含 step_name, input_summary, error_type, message, suggestion
+ - NetworkError, ValidationError, SystemError, AuthError 继承自 KBError
+ - KBSettings 从 ~/.kb_env 加载配置，缺失必需变量时报错
+ - retry_on_network_error 装饰器自动重试网络错误3次
+ - 错误可序列化为 dict 用于日志和状态记录
+
+
+ 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)
+
+
+
+
+
+
+ Task 1: Write failing tests for error handling
+ tests/test_pipeline/test_errors.py
+
+ - 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
+
+
+ 创建 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)
+ ```
+
+
+ pytest tests/test_pipeline/test_errors.py -v --collect-only 2>/dev/null | grep "test_" | wc -l | xargs -I {} test {} -ge 8
+
+ 测试文件包含至少 8 个测试用例，所有测试当前失败（RED 阶段）
+
+
+
+ Task 2: Implement error handling
+ kb_pipeline/errors.py
+
+ - 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
+
+
+ 实现 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 内置同名，考虑命名空间或使用别名
+
+
+ pytest tests/test_pipeline/test_errors.py -v --tb=short
+
+ 所有测试通过（GREEN 阶段）
+
+
+
+ Task 3: Write and implement configuration management
+ kb_pipeline/config.py, tests/test_pipeline/test_config.py
+
+ - 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
+
+
+ 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()
+ ```
+
+
+ pytest tests/test_pipeline/test_config.py -v --tb=short
+
+ 所有测试通过，配置管理正常工作
+
+
+
+ Task 4: Write and implement retry mechanism
+ kb_pipeline/retry.py, tests/test_pipeline/test_retry.py
+
+ - 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
+
+
+ 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
+ )
+ ```
+
+
+ pytest tests/test_pipeline/test_retry.py -v --tb=short
+
+ 所有测试通过，重试机制正常工作
+
+
+
+ Task 5: Update package exports
+ kb_pipeline/__init__.py
+
+ - All public APIs exported from package
+ - Clean import interface
+
+
+ 更新 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",
+ ]
+ ```
+
+
+ python -c "from kb_pipeline import WorkflowStateManager, KBSettings, retry_on_network_error" && echo "OK"
+
+ 所有公共 API 可从包根导入
+
+
+
+
+
+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
+
+
+
+- 结构化错误类型完整，包含上下文信息
+- 配置从 ~/.kb_env 加载，缺失必需字段报错
+- 网络错误自动重试，指数退避
+- 所有测试通过，覆盖率 > 80%
+
+
+