"""PMFlow Project Launch draft and confirm service.""" from __future__ import annotations import hashlib import json from datetime import timedelta from typing import Any, Optional from uuid import uuid4 from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession from ..config.ai_prompts import PROMPT_PMFLOW_LAUNCH_DRAFT from ..models import ( PMFlowLaunchDraft, PMFlowThread, PMFlowThreadMessage, Project, SemanticFact, TruthNode, UserProject, ) from ..schemas.pmflow_launch import ( PMFlowLaunchConfirmResponse, PMFlowLaunchDraftPayload, PMFlowLaunchDraftRequest, PMFlowLaunchDraftResponse, ) from ..services.ai_pmo.ai_pmo_kernel import AIPMOKernelService from ..services.context_cache_service import context_cache_service from ..services.llm_service import llm_service from ..services.operation_log_service import log_operation from ..services.task_persistence_core import build_task_create_draft, persist_task from ..utils.sequence import ensure_project_id_sequence from ..utils.timezone import now_cn_naive from ..utils.title import compact_title class PMFlowLaunchError(RuntimeError): def __init__(self, code: str, message: str, *, status_code: int = 400): super().__init__(message) self.code = code self.message = message self.status_code = int(status_code) class PMFlowLaunchService: """Owns the two-step Project Launch write chain.""" DRAFT_TTL_HOURS = 24 def __init__(self, db: AsyncSession, *, tenant_id: int): self.db = db self.tenant_id = int(tenant_id) async def create_draft( self, *, body: PMFlowLaunchDraftRequest, user_id: int, ) -> PMFlowLaunchDraftResponse: prompt = PROMPT_PMFLOW_LAUNCH_DRAFT.format( input_text=str(body.input or "")[:4000], attachments_json=json.dumps([item.model_dump(mode="json") for item in body.attachments], ensure_ascii=False), ) try: structured = await llm_service.generate_structured_object( prompt, db=self.db, tenant_id=self.tenant_id, function_name="pmflow_launch_draft_v1", temperature=0.0, max_tokens=2200, response_mode="fast", routing_meta={ "scene": "pmflow_project_launch", "tier_alias": "low", "response_contract": "structured", }, ) except RuntimeError as exc: raise PMFlowLaunchError("draft_generation_failed", str(exc), status_code=502) from exc except Exception as exc: raise PMFlowLaunchError("draft_generation_failed", f"项目草案生成失败: {exc}", status_code=502) from exc payload = structured.payload if structured is not None and isinstance(structured.payload, dict) else {} draft_payload = self._normalize_draft_payload(payload) self._validate_draft_payload(draft_payload, require_confirmable=False) warnings = self._draft_count_warnings(draft_payload) merged_warnings = self._merge_warnings(draft_payload.warnings, warnings) draft_payload.warnings = merged_warnings expires_at = now_cn_naive() + timedelta(hours=self.DRAFT_TTL_HOURS) draft_id = f"draft_{uuid4().hex[:16]}" row = PMFlowLaunchDraft( tenant_id=self.tenant_id, user_id=int(user_id), draft_id=draft_id, input_text=str(body.input or "").strip(), input_type=str(body.input_type or "text"), source=str(body.source or "web")[:32], attachments=[item.model_dump(mode="json") for item in body.attachments], draft_payload=draft_payload.model_dump(mode="json"), edited_payload=None, status="draft", confidence=float(draft_payload.confidence), warnings=merged_warnings, expires_at=expires_at, ) self.db.add(row) await log_operation( self.db, tenant_id=self.tenant_id, user_id=int(user_id), project_id=None, entity_type="pmflow_launch_draft", entity_id=draft_id, action="pmflow_launch.draft.create", before_data=None, after_data={ "draft_id": draft_id, "title": draft_payload.title, "warnings": merged_warnings, }, interface_path="/pmflow-launch/drafts", caller="user", ) try: await self.db.commit() except Exception: await self.db.rollback() raise return PMFlowLaunchDraftResponse( **draft_payload.model_dump(mode="json"), draft_id=draft_id, status="draft", expires_at=expires_at.isoformat(), ) async def confirm_draft( self, *, draft_id: str, edited_draft: dict[str, Any], user_id: int, ) -> PMFlowLaunchConfirmResponse: row = await self._load_draft_for_update(draft_id=draft_id, user_id=user_id) if row.status == "confirmed" and row.confirmed_project_id is not None: return PMFlowLaunchConfirmResponse( project_id=int(row.confirmed_project_id), redirect_url=f"/pmflow-os?projectId={int(row.confirmed_project_id)}", warnings=self._safe_str_list(row.warnings), ) self._assert_draft_confirmable(row) payload = self._normalize_draft_payload(self._merge_draft(row.draft_payload, edited_draft)) self._validate_draft_payload(payload, require_confirmable=True) warnings = self._merge_warnings(payload.warnings, self._draft_count_warnings(payload)) request_hash = self._confirm_hash(payload.model_dump(mode="json")) created_truth_node_ids: list[int] = [] created_task_ids: list[int] = [] thread_id: Optional[int] = None thread_message_id: Optional[int] = None try: await ensure_project_id_sequence(self.db) title = compact_title(payload.title, max_len=24) or "新项目" project = Project( tenant_id=self.tenant_id, name=title, title=title, description=self._project_description(payload), owner_id=int(user_id), status="active", assigned_role="BUILDER", signal_count=0, ) self.db.add(project) await self.db.flush() self.db.add(UserProject(tenant_id=self.tenant_id, user_id=int(user_id), project_id=int(project.id))) await self.db.flush() thread = PMFlowThread( tenant_id=self.tenant_id, project_id=int(project.id), launch_draft_id=int(row.id), title=f"{title} 项目线程", objective=payload.one_liner, status="active", source="project_launch", ) self.db.add(thread) await self.db.flush() thread_id = int(thread.id) message = PMFlowThreadMessage( tenant_id=self.tenant_id, thread_id=int(thread.id), project_id=int(project.id), launch_draft_id=int(row.id), role="user", content=row.input_text, source=row.source, attachments=row.attachments or [], message_metadata={ "launch_draft_id": row.draft_id, "input_type": row.input_type, "source": row.source, }, status="sent", ) self.db.add(message) await self.db.flush() thread_message_id = int(message.id) metadata = self._source_metadata( launch_draft=row, thread_id=thread_id, thread_message_id=thread_message_id, payload=payload, ) requirement_nodes = await self._create_truth_nodes( project_id=int(project.id), user_id=int(user_id), payload=payload, metadata=metadata, created_truth_node_ids=created_truth_node_ids, ) await self._create_tasks( project_id=int(project.id), user_id=int(user_id), payload=payload, metadata=metadata, requirement_nodes=requirement_nodes, created_task_ids=created_task_ids, ) self._create_semantic_facts(project_id=int(project.id), payload=payload, metadata=metadata) row.status = "confirmed" row.edited_payload = payload.model_dump(mode="json") row.warnings = warnings row.confidence = float(payload.confidence) row.confirmed_project_id = int(project.id) row.confirmed_at = now_cn_naive() row.confirm_request_hash = request_hash await log_operation( self.db, tenant_id=self.tenant_id, user_id=int(user_id), project_id=int(project.id), entity_type="pmflow_launch_draft", entity_id=row.draft_id, action="pmflow_launch.confirm", before_data={"status": "draft"}, after_data={ "status": "confirmed", "project_id": int(project.id), "thread_id": thread_id, "thread_message_id": thread_message_id, "created_truth_node_ids": created_truth_node_ids, "created_task_ids": created_task_ids, "confirm_request_hash": request_hash, }, interface_path=f"/pmflow-launch/drafts/{row.draft_id}/confirm", caller="user", ) await self.db.commit() except Exception: await self.db.rollback() raise await context_cache_service.invalidate_system_context(tenant_id=self.tenant_id) await AIPMOKernelService.invalidate_world_model_cache(tenant_id=self.tenant_id, project_id=int(row.confirmed_project_id)) return PMFlowLaunchConfirmResponse( project_id=int(row.confirmed_project_id), redirect_url=f"/pmflow-os?projectId={int(row.confirmed_project_id)}", created_truth_node_ids=created_truth_node_ids, created_task_ids=created_task_ids, thread_id=thread_id, thread_message_id=thread_message_id, warnings=warnings, ) async def _load_draft_for_update(self, *, draft_id: str, user_id: int) -> PMFlowLaunchDraft: result = await self.db.execute( select(PMFlowLaunchDraft) .where( and_( PMFlowLaunchDraft.tenant_id == self.tenant_id, PMFlowLaunchDraft.user_id == int(user_id), PMFlowLaunchDraft.draft_id == str(draft_id), ) ) .with_for_update() ) row = result.scalar_one_or_none() if row is None: raise PMFlowLaunchError("draft_not_found", "项目草案不存在或不属于当前用户", status_code=404) return row def _assert_draft_confirmable(self, row: PMFlowLaunchDraft) -> None: if row.status in {"expired", "cancelled", "failed"}: raise PMFlowLaunchError("draft_not_confirmable", f"项目草案状态不可确认: {row.status}", status_code=409) if row.status != "draft": raise PMFlowLaunchError("draft_not_confirmable", f"项目草案状态不可确认: {row.status}", status_code=409) expires_at = row.expires_at if expires_at is not None and expires_at <= now_cn_naive(): row.status = "expired" raise PMFlowLaunchError("draft_expired", "项目草案已过期，请重新生成", status_code=409) def _normalize_draft_payload(self, raw: dict[str, Any]) -> PMFlowLaunchDraftPayload: payload = dict(raw or {}) payload["requirements"] = [ item for item in (self._normalize_requirement(row) for row in self._safe_dict_list(payload.get("requirements"))) if item["title"] ] payload["tasks"] = [ item for item in (self._normalize_task(row) for row in self._safe_dict_list(payload.get("tasks"))) if item["title"] ] payload["risks"] = [ item for item in (self._normalize_risk(row) for row in self._safe_dict_list(payload.get("risks"))) if item["title"] ] payload["decisions"] = [ item for item in (self._normalize_decision(row) for row in self._safe_dict_list(payload.get("decisions"))) if item["title"] ] payload["evidence"] = [ item for item in (self._normalize_evidence(row) for row in self._safe_dict_list(payload.get("evidence"))) if item["excerpt"] ] payload["warnings"] = self._safe_str_list(payload.get("warnings")) try: payload["confidence"] = max(0.0, min(float(payload.get("confidence") or 0), 1.0)) except (TypeError, ValueError): payload["confidence"] = 0.0 return PMFlowLaunchDraftPayload.model_validate(payload) def _validate_draft_payload(self, payload: PMFlowLaunchDraftPayload, *, require_confirmable: bool) -> None: missing = [] for field in ("title", "one_liner", "target_user", "problem", "solution"): if not str(getattr(payload, field) or "").strip(): missing.append(field) if not payload.evidence: missing.append("evidence") if require_confirmable: if not payload.requirements: missing.append("requirements") if not payload.tasks: missing.append("tasks") if not payload.risks: missing.append("risks") if not payload.decisions: missing.append("decisions") if missing: raise PMFlowLaunchError( "draft_structured_failed", f"项目草案结构化字段缺失: {', '.join(missing)}", status_code=422, ) def _draft_count_warnings(self, payload: PMFlowLaunchDraftPayload) -> list[str]: warnings: list[str] = [] if not 3 <= len(payload.requirements) <= 5: warnings.append(f"requirements_count_out_of_range: expected 3-5, got {len(payload.requirements)}") if not 3 <= len(payload.tasks) <= 7: warnings.append(f"tasks_count_out_of_range: expected 3-7, got {len(payload.tasks)}") if not 2 <= len(payload.risks) <= 4: warnings.append(f"risks_count_out_of_range: expected 2-4, got {len(payload.risks)}") if not 1 <= len(payload.decisions) <= 3: warnings.append(f"decisions_count_out_of_range: expected 1-3, got {len(payload.decisions)}") return warnings async def _create_truth_nodes( self, *, project_id: int, user_id: int, payload: PMFlowLaunchDraftPayload, metadata: dict[str, Any], created_truth_node_ids: list[int], ) -> list[TruthNode]: requirements: list[TruthNode] = [] all_nodes: list[TruthNode] = [] for item in payload.requirements: node = TruthNode( tenant_id=self.tenant_id, node_type="requirement", project_id=int(project_id), created_by=int(user_id), title=compact_title(item.title, max_len=120) or "初始需求", content={ "summary": item.summary, "acceptance_criteria": item.acceptance_criteria, "stage": "planning", "metadata": metadata, }, version=1, status="confirmed", ) self.db.add(node) requirements.append(node) all_nodes.append(node) for item in payload.risks: node = TruthNode( tenant_id=self.tenant_id, node_type="risk", project_id=int(project_id), created_by=int(user_id), title=compact_title(item.title, max_len=120) or "初始风险", content={ "summary": item.summary, "risk_level": item.level, "mitigation": item.mitigation, "stage": "planning", "metadata": metadata, }, version=1, status="confirmed", ) self.db.add(node) all_nodes.append(node) for item in payload.decisions: node = TruthNode( tenant_id=self.tenant_id, node_type="decision", project_id=int(project_id), created_by=int(user_id), title=compact_title(item.title, max_len=120) or "初始决策", content={ "summary": item.reason, "owner": item.owner or "项目负责人", "stage": "planning", "metadata": metadata, }, version=1, status="confirmed", ) self.db.add(node) all_nodes.append(node) await self.db.flush() created_truth_node_ids.extend([int(node.id) for node in all_nodes]) return requirements async def _create_tasks( self, *, project_id: int, user_id: int, payload: PMFlowLaunchDraftPayload, metadata: dict[str, Any], requirement_nodes: list[TruthNode], created_task_ids: list[int], ) -> None: if not requirement_nodes: raise PMFlowLaunchError("draft_structured_failed", "项目草案缺少可绑定任务的需求", status_code=422) for index, item in enumerate(payload.tasks): req_index = item.requirement_index if item.requirement_index is not None else index % len(requirement_nodes) if req_index is None or req_index < 0 or req_index >= len(requirement_nodes): req_index = 0 task_payload = { "source": "pmflow_launch_confirmed_draft", "launch_metadata": metadata, "task": item.model_dump(mode="json"), "acceptance_criteria": item.acceptance_criteria, } task = await persist_task( self.db, tenant_id=self.tenant_id, truth_node_id=int(requirement_nodes[int(req_index)].id), project_id=int(project_id), draft=build_task_create_draft( title=item.title, description=item.summary, required_skill_tag_ids=[], ai_breakdown_payload=task_payload, manual_overrides={"launch_metadata": metadata, "acceptance_criteria": item.acceptance_criteria}, priority=item.priority, status="todo", ), actor_user_id=int(user_id), action="pmflow_launch.task.create", remark="pmflow_launch.confirm", commit=False, refresh_workload=False, ) created_task_ids.append(int(task.id)) def _create_semantic_facts( self, *, project_id: int, payload: PMFlowLaunchDraftPayload, metadata: dict[str, Any], ) -> None: facts = { "one_liner": payload.one_liner, "target_user": payload.target_user, "problem": payload.problem, "solution": payload.solution, } for key, value in facts.items(): text = str(value or "").strip() if not text: continue self.db.add( SemanticFact( tenant_id=self.tenant_id, entity_type="project", entity_id=int(project_id), fact_namespace="pmflow_launch_seed", fact_key=key, fact_value_text=text, fact_value_norm=text[:255], value_type="text", confidence=float(payload.confidence), evidence_source_type="pmflow_thread_message", evidence_source_id=metadata.get("thread_message_id"), evidence_excerpt=self._evidence_excerpt(payload), extract_version="pmflow_launch_v1", freshness_at=now_cn_naive().isoformat(), is_active=True, ) ) @staticmethod def _merge_draft(base: Any, edited: dict[str, Any]) -> dict[str, Any]: merged = dict(base or {}) for key, value in dict(edited or {}).items(): if value is not None: merged[key] = value return merged @staticmethod def _safe_dict_list(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] return [dict(item) for item in value if isinstance(item, dict)] @staticmethod def _safe_str_list(value: Any) -> list[str]: if not isinstance(value, list): return [] return [str(item).strip() for item in value if str(item or "").strip()] @staticmethod def _merge_warnings(*groups: list[str]) -> list[str]: seen: set[str] = set() merged: list[str] = [] for group in groups: for item in group or []: text = str(item or "").strip() if text and text not in seen: seen.add(text) merged.append(text) return merged @staticmethod def _normalize_requirement(item: dict[str, Any]) -> dict[str, Any]: return { "title": str(item.get("title") or "").strip(), "summary": str(item.get("summary") or item.get("description") or "").strip(), "acceptance_criteria": PMFlowLaunchService._safe_str_list( item.get("acceptance_criteria") or item.get("acceptanceCriteria") or item.get("criteria") ), } @staticmethod def _normalize_task(item: dict[str, Any]) -> dict[str, Any]: group = str(item.get("group") or "product").strip().lower() group_map = {"产品任务": "product", "开发任务": "engineering", "测试任务": "qa"} group = group_map.get(group, group) if group not in {"product", "engineering", "qa"}: group = "product" priority = str(item.get("priority") or "medium").strip().lower() if priority not in {"low", "medium", "high", "urgent"}: priority = "medium" req_index = item.get("requirement_index") try: req_index = int(req_index) if req_index is not None else None except (TypeError, ValueError): req_index = None return { "title": str(item.get("title") or "").strip(), "summary": str(item.get("summary") or item.get("description") or "").strip(), "group": group, "priority": priority, "acceptance_criteria": PMFlowLaunchService._safe_str_list( item.get("acceptance_criteria") or item.get("acceptanceCriteria") or item.get("criteria") ), "requirement_index": req_index, } @staticmethod def _normalize_risk(item: dict[str, Any]) -> dict[str, Any]: level = str(item.get("level") or item.get("risk_level") or "medium").strip().lower() level_map = {"高": "high", "中": "medium", "低": "low"} level = level_map.get(level, level) if level not in {"low", "medium", "high"}: level = "medium" return { "title": str(item.get("title") or "").strip(), "summary": str(item.get("summary") or item.get("description") or "").strip(), "level": level, "mitigation": str(item.get("mitigation") or item.get("mitigation_plan") or "").strip(), } @staticmethod def _normalize_decision(item: dict[str, Any]) -> dict[str, Any]: return { "title": str(item.get("title") or "").strip(), "reason": str(item.get("reason") or item.get("summary") or item.get("rationale") or "").strip(), "owner": str(item.get("owner") or "").strip(), } @staticmethod def _normalize_evidence(item: dict[str, Any]) -> dict[str, Any]: return { "source": str(item.get("source") or "user_input").strip() or "user_input", "excerpt": str(item.get("excerpt") or item.get("summary") or "").strip(), } @staticmethod def _project_description(payload: PMFlowLaunchDraftPayload) -> str: return "\n".join( [ payload.one_liner, f"目标用户：{payload.target_user}", f"核心问题：{payload.problem}", f"解决方案：{payload.solution}", ] ) @staticmethod def _evidence_excerpt(payload: PMFlowLaunchDraftPayload) -> str: if payload.evidence: return payload.evidence[0].excerpt[:500] return payload.one_liner[:500] def _source_metadata( self, *, launch_draft: PMFlowLaunchDraft, thread_id: int, thread_message_id: int, payload: PMFlowLaunchDraftPayload, ) -> dict[str, Any]: return { "source": "pmflow_launch_confirmed_draft", "launch_draft_id": launch_draft.draft_id, "launch_draft_row_id": int(launch_draft.id), "thread_id": int(thread_id), "thread_message_id": int(thread_message_id), "confidence": float(payload.confidence), "evidence": [item.model_dump(mode="json") for item in payload.evidence], } @staticmethod def _confirm_hash(payload: dict[str, Any]) -> str: raw = json.dumps(payload, ensure_ascii=False, sort_keys=True) return hashlib.sha256(raw.encode("utf-8")).hexdigest() __all__ = ["PMFlowLaunchError", "PMFlowLaunchService"]