blank/stories/epic-1.1-state-machine.md

Epic 1.1: State Machine 状态管理

优先级: P0 (Phase 1a 核心功能) 估算: 18 故事点 依赖: 无

---

Epic 目标

实现 Pipeline 状态机，支持状态转换、持久化和恢复，确保翻译任务在各种异常情况下能够正确管理和恢复。

---

用户价值

As a 系统, I want 使用状态机管理翻译任务的生命周期, So that 可以追踪任务状态并支持状态转换验证。

---

技术栈

• 状态机库: transitions==0.9.0
• 测试框架: pytest==7.4.0
• 代码覆盖率: pytest-cov==4.1.0

---

Story 列表

Story 1.1.1: 定义 PipelineState 枚举和转换规则

估算: 3 SP

描述: 定义任务的所有可能状态以及状态之间的合法转换路径。

验收标准:

# 状态定义
class PipelineState(Enum):
    IDLE = "idle"           # 初始状态，任务未开始
    PREPARING = "preparing" # 准备阶段（文件解析、术语提取）
    CLEANING = "cleaning"   # 清洗阶段
    TRANSLATING = "translating"  # 翻译阶段
    UPLOADING = "uploading"      # 上传阶段
    PAUSED = "paused"       # 暂停状态
    COMPLETED = "completed" # 完成状态
    FAILED = "failed"       # 失败状态

# 合法转换路径
TRANSITIONS = {
    'IDLE': ['PREPARING'],
    'PREPARING': ['CLEANING', 'FAILED', 'PAUSED'],
    'CLEANING': ['TRANSLATING', 'FAILED', 'PAUSED'],
    'TRANSLATING': ['UPLOADING', 'FAILED', 'PAUSED'],
    'UPLOADING': ['COMPLETED', 'FAILED', 'PAUSED'],
    'PAUSED': ['IDLE', 'PREPARING', 'CLEANING', 'TRANSLATING', 'UPLOADING'],
    'FAILED': ['IDLE'],
    'COMPLETED': ['IDLE']
}

技术任务:

1. 创建 src/pipeline/state_machine.py
2. 定义 PipelineState 枚举
3. 定义转换规则字典
4. 编写单元测试验证状态定义

---

Story 1.1.2: 实现状态转换引擎

估算: 5 SP

描述: 使用 transitions 库实现状态机引擎，支持状态转换和回调。

验收标准:

class PipelineStateMachine:
    def __init__(self):
        self.machine = Machine(...)
        self.state = PipelineState.IDLE
        self.state_history = []

    def transition_to(self, new_state: PipelineState) -> bool:
        """尝试转换到新状态"""
        pass

    def can_transition_to(self, new_state: PipelineState) -> bool:
        """检查是否可以转换到新状态"""
        pass

    def get_current_state(self) -> PipelineState:
        """获取当前状态"""
        pass

    def get_state_history(self) -> List[Dict]:
        """获取状态历史记录"""
        pass

回调机制:

• on_enter_PREPARING(): 进入准备阶段时的回调
• on_exit_PREPARING(): 退出准备阶段时的回调
• 每个状态转换都记录时间戳和原因

技术任务:

1. 集成 transitions 库
2. 实现状态转换逻辑
3. 实现回调机制
4. 编写单元测试验证所有转换路径

---

Story 1.1.3: 实现状态持久化

估算: 4 SP

描述: 将状态机状态持久化到文件，支持崩溃后恢复。

验收标准:

# 持久化格式
{
    "work_id": "abc123",
    "current_state": "TRANSLATING",
    "state_history": [
        {"state": "IDLE", "entered_at": "2026-03-15T10:00:00"},
        {"state": "PREPARING", "entered_at": "2026-03-15T10:00:05"},
        {"state": "CLEANING", "entered_at": "2026-03-15T10:01:00"},
        {"state": "TRANSLATING", "entered_at": "2026-03-15T10:05:00"}
    ],
    "progress": {
        "current_chapter": 15,
        "total_chapters": 100
    },
    "metadata": {
        "file_path": "/path/to/novel.txt",
        "last_updated": "2026-03-15T10:30:00"
    }
}

class StatePersistence:
    def save_state(self, work_id: str, machine: PipelineStateMachine) -> None:
        """保存状态到文件"""
        pass

    def load_state(self, work_id: str) -> Optional[Dict]:
        """从文件加载状态"""
        pass

    def get_state_file_path(self, work_id: str) -> Path:
        """获取状态文件路径"""
        pass

技术任务:

1. 创建 src/utils/persistence.py
2. 实现状态序列化（JSON 格式）
3. 实现状态反序列化
4. 使用原子写入（.tmp + rename）确保数据安全
5. 编写测试验证持久化功能

---

Story 1.1.4: 实现状态恢复和验证

估算: 3 SP

描述: 从持久化状态恢复状态机，并验证状态一致性。

验收标准:

class StateRecovery:
    def recover_state_machine(self, work_id: str) -> Optional[PipelineStateMachine]:
        """恢复状态机"""
        pass

    def validate_state(self, state_data: Dict) -> bool:
        """验证状态数据完整性"""
        pass

    def get_resume_point(self, state_data: Dict) -> Optional[str]:
        """获取恢复点（应该从哪个阶段继续）"""
        pass

验证规则:

1. 状态文件格式正确
2. 当前状态是有效状态
3. 进度数据完整（章节索引在有效范围内）
4. 文件路径存在

技术任务:

1. 实现状态恢复逻辑
2. 实现状态验证规则
3. 处理损坏的状态文件
4. 编写测试验证恢复逻辑

---

Story 1.1.5: 单元测试覆盖所有转换路径

估算: 3 SP

描述: 编写完整的单元测试，覆盖所有状态转换路径。

验收标准:

• 代码覆盖率 >= 90%
• 所有状态转换路径测试
• 边界条件测试
• 异常情况测试

测试用例:

class TestPipelineStateMachine:
    def test_initial_state_is_idle(self):
        pass

    def test_valid_transitions(self):
        """测试所有合法转换"""
        pass

    def test_invalid_transitions_rejected(self):
        """测试非法转换被拒绝"""
        pass

    def test_state_from_idle_to_translating(self):
        """测试完整流程"""
        pass

    def test_pause_from_any_state(self):
        """测试从任何状态暂停"""
        pass

    def test_resume_from_pause(self):
        """测试从暂停恢复"""
        pass

    def test_failed_state_only_goes_to_idle(self):
        """测试失败状态只能回到空闲"""
        pass

    def test_state_history_tracking(self):
        """测试状态历史记录"""
        pass

class TestStatePersistence:
    def test_save_and_load_state(self):
        pass

    def test_atomic_write(self):
        pass

    def test_corrupted_state_handling(self):
        pass

class TestStateRecovery:
    def test_recover_to_last_state(self):
        pass

    def test_recover_with_missing_file(self):
        pass

    def test_recover_with_corrupted_data(self):
        pass

技术任务:

1. 创建 tests/test_state_machine.py
2. 实现所有测试用例
3. 运行覆盖率报告
4. 确保覆盖率 >= 90%

---

文件结构

src/
└── pipeline/
    ├── __init__.py
    ├── state_machine.py      # PipelineStateMachine 类
    └── models.py              # PipelineState 枚举

src/utils/
└── persistence.py             # StatePersistence 类

tests/
└── test_state_machine.py      # 所有状态机测试

---

依赖关系

• Epic 1.1 无外部依赖，可独立开发
• Epic 1.2 (Crash-Safe) 依赖 Epic 1.1 的状态持久化功能
• Epic 7a (任务调度) 将使用 Epic 1.1 的状态机

---

完成标准

• 所有 5 个 Story 完成
• 单元测试覆盖率 >= 90%
• 所有验收标准通过
• 代码审查通过

---

下一步

完成 Epic 1.1 后，开始 Epic 1.2 (Crash-Safe 机制) 开发。