From a14c429e900a6588b97999e7a6115f1ff6ad5760 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 16:50:46 +0800 Subject: [PATCH] docs(01): create phase 1 plans - State Manager, Schema Registry, Error Handler & Config --- .planning/ROADMAP.md | 11 +- .planning/phases/01-foundation/01-01-PLAN.md | 224 +++++++++ .planning/phases/01-foundation/01-02-PLAN.md | 319 ++++++++++++ .planning/phases/01-foundation/01-03-PLAN.md | 498 +++++++++++++++++++ 4 files changed, 1047 insertions(+), 5 deletions(-) create mode 100644 .planning/phases/01-foundation/01-01-PLAN.md create mode 100644 .planning/phases/01-foundation/01-02-PLAN.md create mode 100644 .planning/phases/01-foundation/01-03-PLAN.md 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% + + + +After completion, create `.planning/phases/01-foundation/01-01-SUMMARY.md` + 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% + + + +After completion, create `.planning/phases/01-foundation/01-02-SUMMARY.md` + 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% + + + +After completion, create `.planning/phases/01-foundation/01-03-SUMMARY.md` +