"""PMFlow OS change review state machine service.""" from __future__ import annotations from typing import Any, Optional from sqlalchemy import and_, func, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import ( PMFlowApproval, PMFlowChangeItem, PMFlowChangeSet, PMFlowEvidenceLink, PMFlowThread, PMFlowThreadMessage, SemanticFact, Task, TruthNode, ) from ..schemas.pmflow_os import ( PMFlowChangeItemEditRequest, PMFlowChangeSetCommitRequest, PMFlowExtractionCreateRequest, PMFlowThreadMessageCreateRequest, ) 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"} 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]: 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 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" 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"} ] 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, "committedItemIds": [item["changeItemId"] for item in committed_items], "skippedItemIds": [item["changeItemId"] for item in skipped_items], "failedItemIds": [], }, "committedItems": [ {key: value for key, value in item.items() if key != "changeItemIdRaw"} for item in 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 _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_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 _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 != "open": raise PMFlowChangeReviewError("changeset_not_open", "changeset_not_open", status_code=409) if item.status != "pending_review": 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, } @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 "", }