"""PMFlow OS change review state machine service.""" from __future__ import annotations import json from typing import Any, Optional from sqlalchemy import and_, func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import flag_modified from ..config.ai_prompts import PROMPT_PMFLOW_CHANGE_REVIEW_EXTRACTION from ..config.ai_function_runtime_profiles import DEFAULT_AI_FUNCTION_RUNTIME_PROFILES from ..models import ( PMFlowApproval, PMFlowChangeItem, PMFlowChangeSet, PMFlowEvidenceLink, PMFlowThread, PMFlowThreadMessage, SemanticFact, Task, TruthNode, User, UserProject, ) from ..schemas.pmflow_os import ( PMFlowChangeItemEditRequest, PMFlowChangeSetCommitRequest, PMFlowExtractionCreateRequest, PMFlowProposedChangeItem, PMFlowReviewDecisionRequest, PMFlowThreadMessageCreateRequest, ) from .ai_function_runtime_profile_service import PLATFORM_RUNTIME_TENANT_ID, AIFunctionRuntimeProfile, AIFunctionRuntimeProfileService from .llm_service import llm_service from .prompt_runtime_service import PromptRuntimeService from ..utils.timezone import now_cn_naive class PMFlowChangeReviewError(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 PMFlowChangeReviewService: COMMIT_OBJECT_TYPES = {"Requirement", "Decision", "Task", "Risk"} AI_EXTRACTION_FUNCTION_NAME = "pmflow_change_review_extraction_v1" AI_EXTRACTION_SCENE_KEY = "change_review" AI_EXTRACTION_PROMPT_KEY = "PROMPT_PMFLOW_CHANGE_REVIEW_EXTRACTION" def __init__(self, db: AsyncSession, *, tenant_id: int): self.db = db self.tenant_id = int(tenant_id) async def create_thread_message( self, *, project_id: int, thread_id: str, body: PMFlowThreadMessageCreateRequest, user_id: int, ) -> dict[str, Any]: thread = await self._resolve_or_create_thread(project_id=project_id, thread_id=thread_id) row = PMFlowThreadMessage( tenant_id=self.tenant_id, thread_id=int(thread.id), project_id=int(project_id), role="user", content=body.content.strip(), source=(body.source or "web")[:32], attachments=body.attachments, message_metadata={"created_by": int(user_id), "source": body.source or "web"}, status="sent", ) self.db.add(row) await self.db.commit() await self.db.refresh(row) return self.serialize_message(row) async def create_extraction( self, *, project_id: int, body: PMFlowExtractionCreateRequest, user_id: int, ) -> dict[str, Any]: if not body.proposed_items: return await self.create_ai_extraction(project_id=project_id, body=body, user_id=user_id) thread, message = await self._load_thread_message( project_id=project_id, thread_id=body.thread_id, message_id=body.message_id, ) headline = (body.headline or "本轮结构化提取").strip()[:255] summary = (body.summary or f"本轮对话生成 {len(body.proposed_items)} 个待审变更项。").strip() change_set = PMFlowChangeSet( tenant_id=self.tenant_id, project_id=int(project_id), thread_id=int(thread.id), source_message_id=int(message.id), status="open", summary=summary, headline=headline, extraction_scope=(body.scope or "latest_message")[:64], routing_meta=body.routing_meta, llm_call_id=body.llm_call_id, created_by=int(user_id), ) self.db.add(change_set) await self.db.flush() items: list[PMFlowChangeItem] = [] for proposed in body.proposed_items: payload = proposed.payload or {} item = PMFlowChangeItem( tenant_id=self.tenant_id, change_set_id=int(change_set.id), project_id=int(project_id), group=proposed.group, object_type=proposed.object_type, object_id=proposed.object_id, title=proposed.title.strip(), summary=proposed.summary.strip(), diff=(proposed.diff or proposed.summary).strip(), payload=payload, edited_payload=None, confidence=float(proposed.confidence or 0.0), risk_level=proposed.risk_level, status="pending_review", ) self.db.add(item) items.append(item) await self.db.flush() evidence = PMFlowEvidenceLink( tenant_id=self.tenant_id, project_id=int(project_id), change_set_id=int(change_set.id), change_item_id=None, source_type="thread_message", source_id=str(message.id), title="项目线程消息", excerpt=message.content[:1200], display_label="项目线程", evidence_metadata={ "thread_id": int(thread.id), "message_id": int(message.id), "source": message.source, }, ) self.db.add(evidence) await self.db.commit() return await self.get_change_set_detail(project_id=project_id, change_set_id=int(change_set.id)) async def create_ai_extraction( self, *, project_id: int, body: PMFlowExtractionCreateRequest, user_id: int, ) -> dict[str, Any]: thread, message = await self._load_thread_message( project_id=project_id, thread_id=body.thread_id, message_id=body.message_id, ) message_text = str(message.content or "").strip() if not message_text: raise PMFlowChangeReviewError("extraction_evidence_insufficient", "结构化失败：项目线程消息为空，待补充。", status_code=409) structured, runtime_profile, routing_meta = await self._generate_ai_extraction_structured( message_text=message_text, attachments=list(message.attachments or []), profile_stage="active", user_id=user_id, routing_meta_extra=dict(body.routing_meta or {}), ) payload = structured.payload if structured is not None and isinstance(structured.payload, dict) else {} proposed_items = self._normalize_ai_extraction_payload(payload) if not proposed_items: await self._ensure_extraction_clarification_message( thread=thread, source_message=message, user_id=user_id, code="extraction_evidence_insufficient", missing_info=payload.get("missing_info") if isinstance(payload.get("missing_info"), list) else [], warnings=payload.get("warnings") if isinstance(payload.get("warnings"), list) else [], ) raise PMFlowChangeReviewError("extraction_evidence_insufficient", "证据不足：AI 未能抽取可审查对象，待补充。", status_code=409) ai_body = PMFlowExtractionCreateRequest( thread_id=body.thread_id, message_id=body.message_id, scope=body.scope, headline=body.headline or str(payload.get("change_summary") or payload.get("title") or "AI 结构化提取待确认")[:255], summary=body.summary or self._ai_extraction_summary(payload=payload, item_count=len(proposed_items)), routing_meta={ **routing_meta, "aiExtraction": { "functionName": self.AI_EXTRACTION_FUNCTION_NAME, "confidence": payload.get("confidence"), "warnings": payload.get("warnings") if isinstance(payload.get("warnings"), list) else [], "sourceMessageId": str(message.id), "runtimeProfileVersion": runtime_profile.profile_version, "maxTokens": runtime_profile.max_tokens, }, }, llm_call_id=body.llm_call_id, proposed_items=proposed_items, ) return await self.create_extraction(project_id=project_id, body=ai_body, user_id=user_id) async def _ensure_extraction_clarification_message( self, *, thread: PMFlowThread, source_message: PMFlowThreadMessage, user_id: int, code: str, missing_info: list[Any], warnings: list[Any], ) -> None: source_message_id = str(source_message.id) existing_result = await self.db.execute( select(PMFlowThreadMessage.id).where( and_( PMFlowThreadMessage.tenant_id == self.tenant_id, PMFlowThreadMessage.project_id == int(source_message.project_id), PMFlowThreadMessage.thread_id == int(thread.id), PMFlowThreadMessage.role == "assistant", PMFlowThreadMessage.message_metadata.op("->>")("source") == "change_review_extraction_clarification", PMFlowThreadMessage.message_metadata.op("->>")("source_message_id") == source_message_id, ) ) ) if existing_result.scalar_one_or_none() is not None: return clean_missing = [str(item).strip() for item in missing_info if str(item).strip()][:3] clean_warnings = [str(item).strip() for item in warnings if str(item).strip()][:3] detail = clean_missing or clean_warnings if detail: detail_text = "还缺少：" + "、".join(detail) + "。" else: detail_text = "还缺少明确的项目目标、对象、约束或下一步动作。" row = PMFlowThreadMessage( tenant_id=self.tenant_id, thread_id=int(thread.id), project_id=int(source_message.project_id), role="assistant", content=( "我收到了。这条消息暂时没有形成可审查的需求、任务、风险或决策，所以不会写入项目记忆。\n\n" f"{detail_text}继续补充后，我会重新提取待确认变更。" ), source="pmflow_os", attachments=[], message_metadata={ "source": "change_review_extraction_clarification", "source_message_id": source_message_id, "extraction_error_code": code, "missing_info": clean_missing, "warnings": clean_warnings, "created_by": int(user_id), }, status="resolved", ) self.db.add(row) await self.db.commit() async def generate_extraction_payload_for_readiness_check(self, *, user_id: int, profile_stage: str = "draft") -> dict[str, Any]: structured, _profile, _routing_meta = await self._generate_ai_extraction_structured( message_text="增加一个场景评估功能，用于开始前先收集家长最需要解决的高频问题场景。", attachments=[], profile_stage=profile_stage, user_id=user_id, routing_meta_extra={"readiness_check": True, "source": "system_smoke"}, ) payload = structured.payload if structured is not None and isinstance(structured.payload, dict) else {} proposed_items = self._normalize_ai_extraction_payload(payload) if not proposed_items: raise PMFlowChangeReviewError("extraction_evidence_insufficient", "变更抽取 smoke 未产出可审查对象", status_code=409) return {"payload": payload, "proposed_items": proposed_items} async def _generate_ai_extraction_structured( self, *, message_text: str, attachments: list[Any], profile_stage: str = "active", user_id: int | None = None, routing_meta_extra: dict[str, Any] | None = None, ): prompt_result = await PromptRuntimeService(self.db).render( tenant_id=PLATFORM_RUNTIME_TENANT_ID, prompt_key=self.AI_EXTRACTION_PROMPT_KEY, fallback_template=PROMPT_PMFLOW_CHANGE_REVIEW_EXTRACTION, message_text=str(message_text or "")[:4000], attachments_json=json.dumps(attachments or [], ensure_ascii=False), ) runtime_profile = await self._resolve_ai_extraction_runtime_profile(stage=profile_stage) routing_meta = { "scene": "pmflow_os_thread_extraction", "scene_key": self.AI_EXTRACTION_SCENE_KEY, "source": "project_thread_message", "response_contract": "structured", "change_review_contract": "proposed_items_v1", "prompt_runtime_tenant_id": PLATFORM_RUNTIME_TENANT_ID, **(routing_meta_extra if isinstance(routing_meta_extra, dict) else {}), **runtime_profile.routing_meta(), **prompt_result.routing_meta(), } try: structured = await llm_service.generate_structured_object( prompt_result.prompt, db=self.db, tenant_id=self.tenant_id, function_name=self.AI_EXTRACTION_FUNCTION_NAME, temperature=runtime_profile.temperature, max_tokens=runtime_profile.max_tokens, response_mode=runtime_profile.response_mode, routing_meta=routing_meta, user_id=int(user_id or 0), ) except PMFlowChangeReviewError: raise except Exception as exc: raise PMFlowChangeReviewError("ai_extraction_failed", f"结构化失败：AI 提取失败，待补充。{exc}", status_code=502) from exc return structured, runtime_profile, routing_meta async def _resolve_ai_extraction_runtime_profile(self, *, stage: str = "active") -> AIFunctionRuntimeProfile: try: return await AIFunctionRuntimeProfileService(self.db).resolve_profile( function_name=self.AI_EXTRACTION_FUNCTION_NAME, scene_key=self.AI_EXTRACTION_SCENE_KEY, prompt_key=self.AI_EXTRACTION_PROMPT_KEY, stage=stage, ) except Exception: profile = dict(DEFAULT_AI_FUNCTION_RUNTIME_PROFILES[self.AI_EXTRACTION_FUNCTION_NAME]) return AIFunctionRuntimeProfile( function_name=self.AI_EXTRACTION_FUNCTION_NAME, scene_key=self.AI_EXTRACTION_SCENE_KEY, prompt_key=self.AI_EXTRACTION_PROMPT_KEY, required_slot=str(profile.get("required_slot") or "structured"), response_contract=str(profile.get("response_contract") or "structured"), max_tokens=int(profile.get("max_tokens") or 7200), repair_max_tokens=int(profile.get("repair_max_tokens") or 3600), temperature=float(profile.get("temperature") or 0.0), foreground_wait_seconds=int(profile.get("foreground_wait_seconds") or 10), provider_timeout_seconds=int(profile.get("provider_timeout_seconds") or 90), job_timeout_seconds=int(profile.get("job_timeout_seconds") or 180), max_retries=int(profile.get("max_retries") or 1), large_input_policy=str(profile.get("large_input_policy") or "preprocess_then_generate"), response_mode=str(profile.get("response_mode") or "fast"), response_format=str(profile.get("response_format") or "json_object"), runtime_mode=str(profile.get("runtime_mode") or "active"), quality_gate_mode=str(profile.get("quality_gate_mode") or "strict"), profile_version="code-default", profile_source="code_default", ) async def get_change_set_detail(self, *, project_id: int, change_set_id: int) -> dict[str, Any]: change_set = await self._load_change_set(project_id=project_id, change_set_id=change_set_id) items = await self._load_items(project_id=project_id, change_set_id=int(change_set.id)) evidence = await self._load_evidence(project_id=project_id, change_set_id=int(change_set.id)) approvals = await self._load_approvals(project_id=project_id, change_set_id=int(change_set.id)) return { "changeSet": self.serialize_change_set(change_set, items), "items": [self.serialize_change_item(item, evidence) for item in items], "evidence": [self.serialize_evidence(row) for row in evidence], "approvals": [self.serialize_approval(row) for row in approvals], } async def get_item_project_id(self, *, change_item_id: int) -> Optional[int]: result = await self.db.execute( select(PMFlowChangeItem.project_id).where( and_(PMFlowChangeItem.tenant_id == self.tenant_id, PMFlowChangeItem.id == int(change_item_id)) ) ) value = result.scalar_one_or_none() return int(value) if value is not None else None async def get_change_set_project_id(self, *, change_set_id: int) -> Optional[int]: result = await self.db.execute( select(PMFlowChangeSet.project_id).where( and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.id == int(change_set_id)) ) ) value = result.scalar_one_or_none() return int(value) if value is not None else None async def accept_item(self, *, change_item_id: int, user_id: int) -> dict[str, Any]: item, change_set = await self._load_item_with_set(change_item_id=change_item_id) await self._review_item(item=item, change_set=change_set, action="accept", decision="approved", user_id=user_id) await self.db.commit() return await self.get_change_set_detail(project_id=int(item.project_id), change_set_id=int(item.change_set_id)) async def edit_item(self, *, change_item_id: int, body: PMFlowChangeItemEditRequest, user_id: int) -> dict[str, Any]: item, change_set = await self._load_item_with_set(change_item_id=change_item_id) before_payload = dict(item.edited_payload or item.payload or {}) base_payload = dict(item.edited_payload or item.payload or {}) base_payload.update(body.payload_patch or {}) if body.summary is not None: item.summary = body.summary.strip() base_payload["summary"] = item.summary if body.diff is not None: item.diff = body.diff.strip() base_payload["diff"] = item.diff item.edited_payload = base_payload await self._review_item( item=item, change_set=change_set, action="edit", decision="edited", user_id=user_id, reason=body.reason, before_payload_override=before_payload, ) await self.db.commit() return await self.get_change_set_detail(project_id=int(item.project_id), change_set_id=int(item.change_set_id)) async def defer_item(self, *, change_item_id: int, user_id: int, reason: Optional[str] = None) -> dict[str, Any]: item, change_set = await self._load_item_with_set(change_item_id=change_item_id) await self._review_item( item=item, change_set=change_set, action="defer", decision="deferred", user_id=user_id, reason=reason, ) await self.db.commit() return await self.get_change_set_detail(project_id=int(item.project_id), change_set_id=int(item.change_set_id)) async def commit_change_set( self, *, change_set_id: int, body: PMFlowChangeSetCommitRequest, user_id: int, ) -> dict[str, Any]: try: change_set = await self._load_change_set_for_commit(change_set_id=change_set_id) items = await self._load_items(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) evidence = await self._load_evidence(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) approvals = await self._load_approvals(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) selected_ids = self._parse_commit_item_ids(body.item_ids) self._validate_commit_preflight(change_set=change_set, items=items, selected_ids=selected_ids) eligible_items = [ item for item in items if item.status in {"accepted", "edited"} and (not selected_ids or int(item.id) in selected_ids) ] if not eligible_items: raise PMFlowChangeReviewError("no_accepted_items", "no_accepted_items", status_code=409) committed_items: list[dict[str, Any]] = [] for item in eligible_items: self._validate_commit_item(item=item, evidence=evidence, approvals=approvals) for item in eligible_items: committed_items.append( await self._commit_item( item=item, change_set=change_set, evidence=evidence, user_id=user_id, ) ) now = now_cn_naive() for item in eligible_items: item.status = "committed" change_set.committed_by = int(user_id) remaining_open = [ item for item in items if item.id not in {committed["changeItemIdRaw"] for committed in committed_items} and item.status in {"pending_review", "accepted", "edited"} ] if not remaining_open: change_set.status = "committed" change_set.committed_at = now await self.db.commit() skipped_items = [ self._serialize_skipped_commit_item(item) for item in items if item.id not in {committed["changeItemIdRaw"] for committed in committed_items} and item.status in {"deferred", "skipped"} ] committed_by = await self._load_user_display_name(user_id=user_id) serialized_committed_items = [ { **{key: value for key, value in item.items() if key != "changeItemIdRaw"}, "committedAt": now.isoformat(), "committedById": str(user_id), "committedBy": committed_by, } for item in committed_items ] return { "changeSet": { "id": str(change_set.id), "status": change_set.status, "rawStatus": change_set.status, "committedAt": change_set.committed_at.isoformat() if change_set.committed_at else None, "writebackAt": now.isoformat(), "committedById": str(user_id), "committedBy": committed_by, "committedItemIds": [item["changeItemId"] for item in committed_items], "skippedItemIds": [item["changeItemId"] for item in skipped_items], "failedItemIds": [], }, "committedItems": serialized_committed_items, "skippedItems": skipped_items, "failedItems": [], } except PMFlowChangeReviewError: await self.db.rollback() raise except Exception as exc: await self.db.rollback() raise PMFlowChangeReviewError("commit_failed", "commit_failed", status_code=500) from exc async def record_review_decision( self, *, change_set_id: int, body: PMFlowReviewDecisionRequest, user_id: int, ) -> dict[str, Any]: try: change_set = await self._load_change_set_for_review_decision(change_set_id=change_set_id) items = await self._load_items(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) selected_items = self._select_review_decision_items(items=items, raw_item_ids=body.item_ids) warnings: list[str] = [] if body.action == "request_review": reviewer = await self._load_project_reviewer( project_id=int(change_set.project_id), reviewer_id=body.reviewer_id, ) if reviewer is None: warnings.append("reviewer_missing") else: self._mark_change_set_review_meta( change_set=change_set, action="request_review", approval_state="requested", user_id=user_id, reviewer_id=int(reviewer.id), reason=body.reason, selected_items=selected_items, ) await self.db.commit() return { "changeSetDetail": await self.get_change_set_detail( project_id=int(change_set.project_id), change_set_id=int(change_set.id), ), "warnings": warnings, } if not selected_items: raise PMFlowChangeReviewError("review_decision_item_empty", "review_decision_item_empty", status_code=409) before_status_by_id = {int(item.id): item.status for item in selected_items} if body.action == "approve": decision = "approved" approval_state = "approved" after_status = "accepted" for item in selected_items: item.status = "accepted" item.reviewed_at = now_cn_naive() elif body.action == "request_changes": decision = "changes_requested" approval_state = "changes_requested" after_status = "pending_review" elif body.action == "reject": decision = "rejected" approval_state = "rejected" after_status = "deferred" for item in selected_items: item.status = "deferred" item.reviewed_at = now_cn_naive() else: raise PMFlowChangeReviewError("invalid_review_decision", "invalid_review_decision", status_code=422) actor_id = int(user_id) reviewer_meta_id = int(body.reviewer_id or user_id) for item in selected_items: self.db.add( PMFlowApproval( tenant_id=self.tenant_id, project_id=int(item.project_id), change_set_id=int(item.change_set_id), change_item_id=int(item.id), actor_user_id=actor_id, action=body.action, decision=decision, before_status=before_status_by_id[int(item.id)], after_status=after_status if body.action != "approve" else item.status, before_payload=dict(item.edited_payload or item.payload or {}), after_payload={ **dict(item.edited_payload or item.payload or {}), "reviewDecision": body.action, "approvalState": approval_state, }, reason=body.reason, ) ) self._mark_change_set_review_meta( change_set=change_set, action=body.action, approval_state=approval_state, user_id=user_id, reviewer_id=reviewer_meta_id, reason=body.reason, selected_items=selected_items, ) if body.action in {"approve", "reject"}: self._mark_change_set_reviewed_if_no_pending(change_set=change_set, items=items) await self.db.commit() return { "changeSetDetail": await self.get_change_set_detail( project_id=int(change_set.project_id), change_set_id=int(change_set.id), ), "warnings": warnings, } except PMFlowChangeReviewError: await self.db.rollback() raise def _normalize_ai_extraction_payload(self, payload: dict[str, Any]) -> list[PMFlowProposedChangeItem]: direct = payload.get("proposed_items") if isinstance(direct, list): return self._coerce_proposed_items(direct) items: list[dict[str, Any]] = [] confidence = self._confidence(payload.get("confidence"), default=0.72) for raw in self._list_of_dicts(payload.get("requirements")): title = self._clean_text(raw.get("title")) summary = self._clean_text(raw.get("summary") or raw.get("description")) if not title or not summary: continue acceptance = raw.get("acceptance_criteria") or raw.get("acceptanceCriteria") or [] items.append( { "group": "new_object", "object_type": "Requirement", "title": self._prefixed_title(title, "新增需求"), "summary": summary, "diff": f"新增待确认需求：{summary}", "payload": { "summary": summary, "acceptanceCriteria": self._string_list(acceptance), "stage": "planning", }, "confidence": confidence, "risk_level": "medium", } ) for raw in self._list_of_dicts(payload.get("tasks")): title = self._clean_text(raw.get("title")) summary = self._clean_text(raw.get("summary") or raw.get("description")) if not title or not summary: continue priority = self._priority(raw.get("priority")) items.append( { "group": "new_object", "object_type": "Task", "title": self._prefixed_title(title, "新增任务"), "summary": summary, "diff": f"新增待确认任务：{summary}", "payload": { "summary": summary, "description": summary, "priority": priority, "acceptanceCriteria": self._string_list(raw.get("acceptance_criteria") or raw.get("acceptanceCriteria")), "requirementIndex": raw.get("requirement_index"), "stage": "development", }, "confidence": confidence, "risk_level": "medium" if priority != "high" else "high", } ) for raw in self._list_of_dicts(payload.get("risks")): title = self._clean_text(raw.get("title")) summary = self._clean_text(raw.get("summary") or raw.get("description")) if not title or not summary: continue level = self._risk_level(raw.get("level") or raw.get("risk_level")) mitigation = self._clean_text(raw.get("mitigation") or raw.get("mitigation_plan")) or "待补充缓解策略。" items.append( { "group": "risk", "object_type": "Risk", "title": self._prefixed_title(title, "新增风险"), "summary": summary, "diff": f"新增待确认风险与缓解策略：{mitigation}", "payload": { "summary": summary, "mitigation": mitigation, "stage": "planning", }, "confidence": confidence, "risk_level": level, } ) for raw in self._list_of_dicts(payload.get("decisions")): title = self._clean_text(raw.get("title")) summary = self._clean_text(raw.get("summary") or raw.get("reason")) if not title or not summary: continue items.append( { "group": "decision", "object_type": "Decision", "title": self._prefixed_title(title, "待确认决策"), "summary": summary, "diff": f"新增待确认决策：{summary}", "payload": { "summary": summary, "owner": self._clean_text(raw.get("owner")) or "未分配", "stage": "planning", }, "confidence": confidence, "risk_level": "medium", } ) return self._coerce_proposed_items(items) @staticmethod def _coerce_proposed_items(items: list[Any]) -> list[PMFlowProposedChangeItem]: proposed: list[PMFlowProposedChangeItem] = [] for raw in items[:24]: if not isinstance(raw, dict): continue try: proposed.append(PMFlowProposedChangeItem.model_validate(raw)) except Exception: continue return proposed @staticmethod def _ai_extraction_summary(*, payload: dict[str, Any], item_count: int) -> str: warnings = payload.get("warnings") if isinstance(payload.get("warnings"), list) else [] suffix = f"；警告：{'；'.join(str(item) for item in warnings[:3])}" if warnings else "" return f"AI 从本轮项目线程生成 {item_count} 个待审变更项，尚未写入项目记忆{suffix}。" @staticmethod def _list_of_dicts(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, dict)] @staticmethod def _clean_text(value: Any) -> str: return str(value or "").strip() @classmethod def _string_list(cls, value: Any) -> list[str]: if isinstance(value, list): return [cls._clean_text(item) for item in value if cls._clean_text(item)] text = cls._clean_text(value) return [text] if text else [] @staticmethod def _prefixed_title(title: str, prefix: str) -> str: if title.startswith(prefix): return title[:255] return f"{prefix}：{title}"[:255] @staticmethod def _confidence(value: Any, *, default: float) -> float: try: number = float(value) except Exception: number = default return max(0.0, min(1.0, number)) @staticmethod def _risk_level(value: Any) -> str: text = str(value or "medium").strip().lower() if text in {"low", "medium", "high"}: return text if text in {"urgent", "critical", "p0"}: return "high" return "medium" async def _resolve_or_create_thread(self, *, project_id: int, thread_id: str) -> PMFlowThread: cleaned = str(thread_id or "").strip() if cleaned.isdigit(): result = await self.db.execute( select(PMFlowThread).where( and_( PMFlowThread.tenant_id == self.tenant_id, PMFlowThread.project_id == int(project_id), PMFlowThread.id == int(cleaned), ) ) ) thread = result.scalar_one_or_none() if thread is None: raise PMFlowChangeReviewError("thread_not_found", "thread_not_found", status_code=404) return thread if cleaned not in {"default", f"thread-project-{int(project_id)}"}: raise PMFlowChangeReviewError("thread_not_found", "thread_not_found", status_code=404) result = await self.db.execute( select(PMFlowThread) .where(and_(PMFlowThread.tenant_id == self.tenant_id, PMFlowThread.project_id == int(project_id), PMFlowThread.status == "active")) .order_by(PMFlowThread.created_at.asc(), PMFlowThread.id.asc()) .limit(1) ) thread = result.scalar_one_or_none() if thread is not None: return thread thread = PMFlowThread( tenant_id=self.tenant_id, project_id=int(project_id), title="项目默认线程", objective="项目内 PMFlow OS 对话线程。", status="active", source="web", ) self.db.add(thread) await self.db.flush() return thread async def _load_thread_message(self, *, project_id: int, thread_id: str, message_id: str) -> tuple[PMFlowThread, PMFlowThreadMessage]: if not str(thread_id).isdigit() or not str(message_id).isdigit(): raise PMFlowChangeReviewError("message_not_found", "message_not_found", status_code=404) result = await self.db.execute( select(PMFlowThread, PMFlowThreadMessage) .join(PMFlowThreadMessage, PMFlowThreadMessage.thread_id == PMFlowThread.id) .where( and_( PMFlowThread.tenant_id == self.tenant_id, PMFlowThread.project_id == int(project_id), PMFlowThread.id == int(thread_id), PMFlowThreadMessage.tenant_id == self.tenant_id, PMFlowThreadMessage.project_id == int(project_id), PMFlowThreadMessage.id == int(message_id), ) ) ) row = result.first() if row is None: raise PMFlowChangeReviewError("message_not_found", "message_not_found", status_code=404) return row[0], row[1] async def _load_change_set(self, *, project_id: int, change_set_id: int) -> PMFlowChangeSet: result = await self.db.execute( select(PMFlowChangeSet).where( and_( PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.project_id == int(project_id), PMFlowChangeSet.id == int(change_set_id), ) ) ) row = result.scalar_one_or_none() if row is None: raise PMFlowChangeReviewError("changeset_not_found", "changeset_not_found", status_code=404) return row async def _load_change_set_for_commit(self, *, change_set_id: int) -> PMFlowChangeSet: result = await self.db.execute( select(PMFlowChangeSet) .where(and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.id == int(change_set_id))) .with_for_update() ) row = result.scalar_one_or_none() if row is None: raise PMFlowChangeReviewError("changeset_not_found", "changeset_not_found", status_code=404) return row async def _load_change_set_for_review_decision(self, *, change_set_id: int) -> PMFlowChangeSet: result = await self.db.execute( select(PMFlowChangeSet) .where(and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.id == int(change_set_id))) .with_for_update() ) row = result.scalar_one_or_none() if row is None: raise PMFlowChangeReviewError("changeset_not_found", "changeset_not_found", status_code=404) if row.status == "committed": raise PMFlowChangeReviewError("changeset_already_committed", "changeset_already_committed", status_code=409) if row.status not in {"open", "reviewed"}: raise PMFlowChangeReviewError("changeset_not_open", "changeset_not_open", status_code=409) return row async def _load_items(self, *, project_id: int, change_set_id: int) -> list[PMFlowChangeItem]: result = await self.db.execute( select(PMFlowChangeItem) .where( and_( PMFlowChangeItem.tenant_id == self.tenant_id, PMFlowChangeItem.project_id == int(project_id), PMFlowChangeItem.change_set_id == int(change_set_id), ) ) .order_by(PMFlowChangeItem.created_at.asc(), PMFlowChangeItem.id.asc()) ) return list(result.scalars().all()) async def _load_evidence(self, *, project_id: int, change_set_id: int) -> list[PMFlowEvidenceLink]: result = await self.db.execute( select(PMFlowEvidenceLink) .where( and_( PMFlowEvidenceLink.tenant_id == self.tenant_id, PMFlowEvidenceLink.project_id == int(project_id), PMFlowEvidenceLink.change_set_id == int(change_set_id), ) ) .order_by(PMFlowEvidenceLink.created_at.asc(), PMFlowEvidenceLink.id.asc()) ) return list(result.scalars().all()) async def _load_approvals(self, *, project_id: int, change_set_id: int) -> list[PMFlowApproval]: result = await self.db.execute( select(PMFlowApproval) .where( and_( PMFlowApproval.tenant_id == self.tenant_id, PMFlowApproval.project_id == int(project_id), PMFlowApproval.change_set_id == int(change_set_id), ) ) .order_by(PMFlowApproval.created_at.asc(), PMFlowApproval.id.asc()) ) return list(result.scalars().all()) async def _load_item_with_set(self, *, change_item_id: int) -> tuple[PMFlowChangeItem, PMFlowChangeSet]: result = await self.db.execute( select(PMFlowChangeItem, PMFlowChangeSet) .join(PMFlowChangeSet, PMFlowChangeSet.id == PMFlowChangeItem.change_set_id) .where(and_(PMFlowChangeItem.tenant_id == self.tenant_id, PMFlowChangeItem.id == int(change_item_id))) .with_for_update() ) row = result.first() if row is None: raise PMFlowChangeReviewError("change_item_not_found", "change_item_not_found", status_code=404) return row[0], row[1] async def _load_project_reviewer(self, *, project_id: int, reviewer_id: Optional[int]) -> Optional[User]: if reviewer_id is None: return None result = await self.db.execute( select(User) .join(UserProject, UserProject.user_id == User.id) .where( and_( User.tenant_id == self.tenant_id, User.id == int(reviewer_id), UserProject.tenant_id == self.tenant_id, UserProject.project_id == int(project_id), UserProject.user_id == int(reviewer_id), User.role.in_(["OWNER", "REVIEWER"]), ) ) ) return result.scalar_one_or_none() def _select_review_decision_items( self, *, items: list[PMFlowChangeItem], raw_item_ids: list[str], ) -> list[PMFlowChangeItem]: selected_ids = self._parse_review_decision_item_ids(raw_item_ids) if selected_ids: item_ids = {int(item.id) for item in items} if not selected_ids.issubset(item_ids): raise PMFlowChangeReviewError("invalid_review_decision_item_ids", "invalid_review_decision_item_ids", status_code=422) return [item for item in items if int(item.id) in selected_ids] return [item for item in items if item.status == "pending_review"] @staticmethod def _parse_review_decision_item_ids(raw_item_ids: list[str]) -> set[int]: parsed: set[int] = set() for raw in raw_item_ids or []: text = str(raw or "").strip() if not text.isdigit(): raise PMFlowChangeReviewError("invalid_review_decision_item_ids", "invalid_review_decision_item_ids", status_code=422) parsed.add(int(text)) return parsed def _mark_change_set_review_meta( self, *, change_set: PMFlowChangeSet, action: str, approval_state: str, user_id: int, reviewer_id: Optional[int], reason: Optional[str], selected_items: list[PMFlowChangeItem], ) -> None: now = now_cn_naive() routing_meta = dict(change_set.routing_meta or {}) review_meta = dict(routing_meta.get("reviewDecision") or {}) if action == "request_review": review_meta["requestedAt"] = review_meta.get("requestedAt") or now.isoformat() review_meta["requestedBy"] = str(user_id) else: review_meta["decidedAt"] = now.isoformat() review_meta["decidedBy"] = str(user_id) review_meta.update( { "action": action, "approvalState": approval_state, "reviewerId": None if reviewer_id is None else str(reviewer_id), "decisionReason": reason, "itemIds": [str(item.id) for item in selected_items], } ) routing_meta["reviewDecision"] = review_meta change_set.routing_meta = routing_meta flag_modified(change_set, "routing_meta") @staticmethod def _mark_change_set_reviewed_if_no_pending(*, change_set: PMFlowChangeSet, items: list[PMFlowChangeItem]) -> None: if not any(item.status == "pending_review" for item in items): change_set.status = "reviewed" change_set.reviewed_at = now_cn_naive() async def _review_item( self, *, item: PMFlowChangeItem, change_set: PMFlowChangeSet, action: str, decision: str, user_id: int, reason: Optional[str] = None, before_payload_override: Optional[dict[str, Any]] = None, ) -> None: if change_set.status not in {"open", "reviewed"}: raise PMFlowChangeReviewError("changeset_not_open", "changeset_not_open", status_code=409) if item.status in {"committed", "skipped"}: raise PMFlowChangeReviewError("item_already_reviewed", "item_already_reviewed", status_code=409) before_status = item.status before_payload = dict(before_payload_override or item.edited_payload or item.payload or {}) next_status = "accepted" if action == "accept" else ("edited" if action == "edit" else "deferred") item.status = next_status item.reviewed_at = now_cn_naive() after_payload = dict(item.edited_payload or item.payload or {}) self.db.add( PMFlowApproval( tenant_id=self.tenant_id, project_id=int(item.project_id), change_set_id=int(item.change_set_id), change_item_id=int(item.id), actor_user_id=int(user_id), action=action, decision=decision, before_status=before_status, after_status=next_status, before_payload=before_payload, after_payload=after_payload, reason=reason, ) ) await self.db.flush() pending_count = await self.db.scalar( select(func.count(PMFlowChangeItem.id)).where( and_( PMFlowChangeItem.tenant_id == self.tenant_id, PMFlowChangeItem.change_set_id == int(change_set.id), PMFlowChangeItem.status == "pending_review", ) ) ) if int(pending_count or 0) == 0: change_set.status = "reviewed" change_set.reviewed_at = now_cn_naive() @staticmethod def _parse_commit_item_ids(raw_item_ids: list[str]) -> set[int]: parsed: set[int] = set() for raw in raw_item_ids or []: text = str(raw or "").strip() if not text.isdigit(): raise PMFlowChangeReviewError("invalid_commit_item_ids", "invalid_commit_item_ids", status_code=422) parsed.add(int(text)) return parsed def _validate_commit_preflight( self, *, change_set: PMFlowChangeSet, items: list[PMFlowChangeItem], selected_ids: set[int], ) -> None: if change_set.status == "committed": raise PMFlowChangeReviewError("changeset_already_committed", "changeset_already_committed", status_code=409) if change_set.status not in {"open", "reviewed"}: raise PMFlowChangeReviewError("changeset_not_reviewed", "changeset_not_reviewed", status_code=409) item_ids = {int(item.id) for item in items} if selected_ids and not selected_ids.issubset(item_ids): raise PMFlowChangeReviewError("invalid_commit_item_ids", "invalid_commit_item_ids", status_code=422) if selected_ids: invalid_selected = [ item for item in items if int(item.id) in selected_ids and item.status not in {"accepted", "edited"} ] if invalid_selected: raise PMFlowChangeReviewError("invalid_commit_item_ids", "invalid_commit_item_ids", status_code=422) return if any(item.status == "pending_review" for item in items): raise PMFlowChangeReviewError("changeset_not_reviewed", "changeset_not_reviewed", status_code=409) def _validate_commit_item( self, *, item: PMFlowChangeItem, evidence: list[PMFlowEvidenceLink], approvals: list[PMFlowApproval], ) -> None: if item.object_type not in self.COMMIT_OBJECT_TYPES: raise PMFlowChangeReviewError("unsupported_object_type", "unsupported_object_type", status_code=422) payload = self._effective_payload(item) self._require_commit_text(item.title, field="title") self._require_commit_text(str(payload.get("summary") or item.summary), field="summary") if item.object_type == "Risk": self._require_commit_text(str(payload.get("mitigation") or payload.get("mitigation_plan") or item.summary), field="mitigation") if item.object_type == "Task": self._require_commit_text(str(payload.get("description") or payload.get("summary") or item.summary), field="description") if str(item.risk_level or "").lower() == "high": approval_ok = any( int(row.change_item_id) == int(item.id) and row.decision in {"approved", "edited"} and row.after_status in {"accepted", "edited"} for row in approvals ) if not approval_ok or not self._item_evidence(item=item, evidence=evidence): raise PMFlowChangeReviewError("high_risk_approval_required", "high_risk_approval_required", status_code=409) async def _commit_item( self, *, item: PMFlowChangeItem, change_set: PMFlowChangeSet, evidence: list[PMFlowEvidenceLink], user_id: int, ) -> dict[str, Any]: if item.object_type == "Task": target_type, target_id = await self._commit_task_item( item=item, change_set=change_set, evidence=evidence, user_id=user_id, ) else: target_type, target_id = await self._commit_truth_node_item( item=item, change_set=change_set, evidence=evidence, user_id=user_id, ) item.object_id = str(target_id) evidence_ids = [str(row.id) for row in self._item_evidence(item=item, evidence=evidence)] return { "changeItemIdRaw": int(item.id), "changeItemId": str(item.id), "objectType": item.object_type, "targetType": target_type, "targetId": str(target_id), "status": "committed", "evidenceIds": evidence_ids, "semanticFactStatus": "seeded", } async def _commit_truth_node_item( self, *, item: PMFlowChangeItem, change_set: PMFlowChangeSet, evidence: list[PMFlowEvidenceLink], user_id: int, ) -> tuple[str, int]: node_type = { "Requirement": "requirement", "Decision": "decision", "Risk": "risk", }.get(item.object_type) if node_type is None: raise PMFlowChangeReviewError("unsupported_object_type", "unsupported_object_type", status_code=422) content = self._commit_content(item=item, change_set=change_set, evidence=evidence, user_id=user_id) node = TruthNode( tenant_id=self.tenant_id, node_type=node_type, project_id=int(item.project_id), signal_id=None, created_by=int(user_id), title=item.title[:500], content=content, version=1, status="confirmed", ) self.db.add(node) await self.db.flush() self._add_semantic_fact_for_commit(node=node, item=item, evidence=evidence) return "truth_node", int(node.id) async def _commit_task_item( self, *, item: PMFlowChangeItem, change_set: PMFlowChangeSet, evidence: list[PMFlowEvidenceLink], user_id: int, ) -> tuple[str, int]: content = self._commit_content(item=item, change_set=change_set, evidence=evidence, user_id=user_id) node = TruthNode( tenant_id=self.tenant_id, node_type="task", project_id=int(item.project_id), signal_id=None, created_by=int(user_id), title=item.title[:500], content=content, version=1, status="confirmed", ) self.db.add(node) await self.db.flush() payload = self._effective_payload(item) task = Task( tenant_id=self.tenant_id, truth_node_id=int(node.id), project_id=int(item.project_id), assignee_id=None, ai_recommended_assignees=None, ai_breakdown_payload={"source": "pmflow_change_set_commit", "payload": payload}, manual_overrides={ "change_set_id": int(change_set.id), "change_item_id": int(item.id), "acceptance_criteria": self._acceptance_criteria(payload), }, title=item.title[:500], description=str(payload.get("description") or payload.get("summary") or item.summary), required_skill_tag_ids=[], status="todo", priority=self._priority(payload.get("priority")), ) self.db.add(task) await self.db.flush() self._add_semantic_fact_for_commit(node=node, item=item, evidence=evidence) self._add_semantic_fact_for_commit(node=None, item=item, evidence=evidence, task=task) return "task", int(task.id) def _commit_content( self, *, item: PMFlowChangeItem, change_set: PMFlowChangeSet, evidence: list[PMFlowEvidenceLink], user_id: int, ) -> dict[str, Any]: payload = self._effective_payload(item) evidence_ids = [str(row.id) for row in self._item_evidence(item=item, evidence=evidence)] return { "source": "pmflow_change_set_commit", "change_set_id": int(change_set.id), "change_item_id": int(item.id), "object_type": item.object_type, "summary": str(payload.get("summary") or item.summary), "diff": item.diff, "payload": payload, "stage": str(payload.get("stage") or ("development" if item.object_type == "Task" else "planning")), "owner": str(payload.get("owner") or payload.get("decision_owner") or "未分配"), "mitigation": str(payload.get("mitigation") or payload.get("mitigation_plan") or item.summary), "evidence_ids": evidence_ids, "confidence": float(item.confidence or 0.0), "risk_level": item.risk_level, "committed_by": int(user_id), "committed_at": now_cn_naive().isoformat(), } def _add_semantic_fact_for_commit( self, *, node: Optional[TruthNode], item: PMFlowChangeItem, evidence: list[PMFlowEvidenceLink], task: Optional[Task] = None, ) -> None: target_type = "task" if task is not None else "truth_node" target_id = int(task.id) if task is not None else int(node.id) # type: ignore[union-attr] source_evidence = self._item_evidence(item=item, evidence=evidence) first_evidence = source_evidence[0] if source_evidence else None text = str(self._effective_payload(item).get("summary") or item.summary).strip() if not text: return self.db.add( SemanticFact( tenant_id=self.tenant_id, entity_type=target_type, entity_id=target_id, fact_namespace="pmflow_change_commit", fact_key="summary", fact_value_text=text, fact_value_norm=text[:255], value_type="text", confidence=float(item.confidence or 0.0), evidence_source_type="pmflow_evidence_link" if first_evidence is not None else None, evidence_source_id=int(first_evidence.id) if first_evidence is not None else None, evidence_excerpt=first_evidence.excerpt if first_evidence is not None else None, extract_version="pmflow_change_commit_v1", freshness_at=now_cn_naive().isoformat(), is_active=True, ) ) @staticmethod def _effective_payload(item: PMFlowChangeItem) -> dict[str, Any]: payload = item.edited_payload if item.status == "edited" and item.edited_payload is not None else item.payload return dict(payload or {}) @staticmethod def _item_evidence(*, item: PMFlowChangeItem, evidence: list[PMFlowEvidenceLink]) -> list[PMFlowEvidenceLink]: return [row for row in evidence if row.change_item_id in {None, item.id}] @staticmethod def _require_commit_text(value: str, *, field: str) -> None: if not str(value or "").strip(): raise PMFlowChangeReviewError("invalid_change_payload", f"missing_{field}", status_code=422) @staticmethod def _priority(value: Any) -> str: text = str(value or "medium").strip().lower() if text in {"high", "medium", "low"}: return text if text in {"p0", "critical", "urgent"}: return "high" if text in {"p2", "p3"}: return "low" return "medium" @staticmethod def _acceptance_criteria(payload: dict[str, Any]) -> list[str]: raw = payload.get("acceptanceCriteria") or payload.get("acceptance_criteria") or [] if isinstance(raw, list): return [str(item).strip() for item in raw if str(item).strip()] text = str(raw or "").strip() return [text] if text else [] @staticmethod def _serialize_skipped_commit_item(item: PMFlowChangeItem) -> dict[str, Any]: return { "changeItemId": str(item.id), "objectType": item.object_type, "status": item.status, "reason": "item_not_accepted_or_edited", } @staticmethod def serialize_message(row: PMFlowThreadMessage) -> dict[str, Any]: return { "id": str(row.id), "threadId": str(row.thread_id), "projectId": str(row.project_id), "role": row.role, "author": "你" if row.role == "user" else "PMFlow", "content": row.content, "createdAt": row.created_at.isoformat() if row.created_at else "", "status": "sent", "source": row.source, "metadata": row.message_metadata or {}, } @staticmethod def serialize_change_set(row: PMFlowChangeSet, items: list[PMFlowChangeItem]) -> dict[str, Any]: status = row.status if row.status in {"open", "reviewed", "committed", "partially_committed"} else "reviewed" return { "id": str(row.id), "threadId": str(row.thread_id), "summary": row.summary, "headline": row.headline, "createdAt": row.created_at.isoformat() if row.created_at else "", "status": status, "itemIds": [str(item.id) for item in items], "sourceMessageId": str(row.source_message_id), "rawStatus": row.status, "committedAt": row.committed_at.isoformat() if row.committed_at else None, "committedById": str(row.committed_by) if row.committed_by else None, "routingMeta": row.routing_meta or {}, } async def _load_user_display_name(self, *, user_id: int) -> str: result = await self.db.execute( select(User).where(and_(User.tenant_id == self.tenant_id, User.id == int(user_id))) ) user = result.scalar_one_or_none() if user is None: return str(user_id) return str(user.name or user.username or user.email or user.id) @staticmethod def serialize_change_item(row: PMFlowChangeItem, evidence: list[PMFlowEvidenceLink]) -> dict[str, Any]: evidence_ids = [str(item.id) for item in evidence if item.change_item_id in {None, row.id}] return { "id": str(row.id), "changeSetId": str(row.change_set_id), "group": row.group, "objectType": row.object_type, "objectId": str(row.object_id or f"pending-{row.id}"), "title": row.title, "summary": row.summary, "diff": row.diff, "confidence": float(row.confidence or 0.0), "riskLevel": row.risk_level, "status": row.status if row.status in {"pending_review", "accepted", "edited", "deferred", "committed", "skipped"} else "deferred", "evidenceIds": evidence_ids, "payload": row.payload or {}, "editedPayload": row.edited_payload, } @staticmethod def serialize_evidence(row: PMFlowEvidenceLink) -> dict[str, Any]: kind = "conversation" if row.source_type == "thread_message" else "document" return { "id": str(row.id), "kind": kind, "title": row.title, "excerpt": row.excerpt, "sourceLabel": row.display_label, "sourceType": row.source_type, "sourceId": row.source_id, "changeSetId": str(row.change_set_id), "changeItemId": str(row.change_item_id) if row.change_item_id is not None else None, "metadata": row.evidence_metadata or {}, } @staticmethod def serialize_approval(row: PMFlowApproval) -> dict[str, Any]: return { "id": str(row.id), "title": f"{row.action} {row.change_item_id}", "summary": row.reason or f"{row.before_status} -> {row.after_status}", "stage": "planning", "status": "confirmed" if row.decision in {"approved", "edited"} else "draft", "owner": str(row.actor_user_id), "changeSetId": str(row.change_set_id), "changeItemId": str(row.change_item_id), "action": row.action, "decision": row.decision, "createdAt": row.created_at.isoformat() if row.created_at else "", }