""" app/session.py - 会话状态管理 """ import uuid import time from enum import Enum from typing import Optional from app.models import RequirementAnalysis, TestCaseResult, CodeGenerationResult class SessionStatus(str, Enum): CLARIFYING = "clarifying" # 正在澄清需求 PM_READY = "pm_ready" # 需求澄清完毕,可运行 PM Agent PM_DONE = "pm_done" # PM Agent 已完成 QA_READY = "qa_ready" # 可运行 QA Agent QA_DONE = "qa_done" # QA Agent 已完成 DEV_READY = "dev_ready" # 可运行 Dev Agent DEV_DONE = "dev_done" # Dev Agent 已完成 TEST_DONE = "test_done" # 单元测试执行完成 class Session: def __init__(self, session_id: str, raw_requirement: str): self.session_id: str = session_id self.raw_requirement: str = raw_requirement # 用户最原始的需求 self.clarified_requirement: str = raw_requirement # 经过澄清补充后的完整需求 self.clarify_history: list[dict] = [] # 澄清对话历史 self.status: SessionStatus = SessionStatus.CLARIFYING self.created_at: float = time.time() self.updated_at: float = time.time() # 各 Agent 产出 self.requirement_analysis: Optional[RequirementAnalysis] = None self.test_cases: Optional[TestCaseResult] = None self.code_generation: Optional[CodeGenerationResult] = None self.test_execution: Optional[dict] = None # pytest 执行结果 def touch(self): self.updated_at = time.time() class SessionStore: _store: dict[str, Session] = {} _TTL_SECONDS = 7200 # 2小时过期 @classmethod def create(cls, raw_requirement: str) -> Session: session_id = str(uuid.uuid4()) session = Session(session_id, raw_requirement) cls._store[session_id] = session cls._evict_expired() return session @classmethod def get(cls, session_id: str) -> Optional[Session]: session = cls._store.get(session_id) if session and time.time() - session.updated_at > cls._TTL_SECONDS: del cls._store[session_id] return None return session @classmethod def _evict_expired(cls): now = time.time() expired = [sid for sid, s in cls._store.items() if now - s.updated_at > cls._TTL_SECONDS] for sid in expired: del cls._store[sid]