PMFlow/backend/app/services/pmflow_os_projection_service.py

"""PMFlow OS read-only projection service."""

from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Any, Optional

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import (
    FeedbackIngest,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowEvidenceLink,
    PMFlowThreadMessage,
    PMFlowThread as PMFlowThreadModel,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Task,
    TruthNode,
    User,
)
from ..schemas.pmflow_os import (
    ExecutionSummaryProjection,
    PMFlowDecision,
    PMFlowExecutionContext,
    PMFlowFeedEvent,
    PMFlowMetricSignal,
    PMFlowModelDecision,
    PMFlowProjectMemory,
    PMFlowProjectSummary,
    PMFlowRelease,
    PMFlowRequirement,
    PMFlowRisk,
    PMFlowTask,
    PMFlowTestRun,
    PMFlowThread,
    ProjectTruthStateProjection,
)
from .ai_project_control_service import AIProjectControlService
from .pmflow_change_review_service import PMFlowChangeReviewService


class PMFlowOSProjectionService:
    """Builds PMFlow OS compatible read models from existing production tables."""

    def __init__(self, db: AsyncSession, *, tenant_id: int):
        self.db = db
        self.tenant_id = int(tenant_id)

    async def get_truth_state(
        self,
        *,
        project_id: int,
        current_user_id: Optional[int] = None,
    ) -> ProjectTruthStateProjection:
        project = await self._load_project(project_id=project_id)
        truth_nodes = await self._load_truth_nodes(project_id=project_id)
        tasks = await self._load_tasks(project_id=project_id)
        task_items = self._project_tasks(tasks)
        execution_summary = await self.get_execution_summary(
            project_id=project_id,
            current_user_id=current_user_id,
        )
        project_stage = self._infer_project_stage(truth_nodes=truth_nodes, tasks=tasks, execution=execution_summary.execution)
        launch_thread, launch_messages = await self._load_project_thread(project_id=project_id)
        thread_id = str(launch_thread.id) if launch_thread is not None else f"thread-project-{project.id}"
        requirements = self._project_requirements(truth_nodes=truth_nodes, tasks=task_items)
        releases = await self._project_releases(truth_nodes=truth_nodes, project_id=project_id)
        test_runs = await self._project_test_runs(project_id=project_id)
        metric_signals = await self._project_metric_signals(project_id=project_id)
        memory = await self._project_memory(
            project=project,
            truth_nodes=truth_nodes,
            task_ids=[int(task.id) for task, _ in tasks],
        )
        review_projection = await self._load_review_projection(project_id=project_id)
        active_change_set_id = self._active_change_set_id(review_projection["change_sets"])
        loop_stage = self._review_loop_stage(
            change_sets=review_projection["change_sets"],
            change_items=review_projection["change_items"],
        )
        warnings: list[str] = []
        if launch_thread is None:
            warnings.append("thread_projection_empty: no pmflow_threads row was found; returning an empty canonical thread.")
        if not launch_messages:
            warnings.append("thread_messages_empty: no pmflow_thread_messages rows were found for this project.")
        if not truth_nodes:
            warnings.append("truth_nodes_empty: no confirmed project truth nodes were found for this project.")
        if not tasks:
            warnings.append("tasks_empty: no task rows were found for this project.")

        return ProjectTruthStateProjection(
            loopStage=loop_stage,
            lastWritebackAt=self._latest_iso([node.updated_at for node in truth_nodes] + [task.updated_at for task, _ in tasks]),
            project=PMFlowProjectSummary(
                id=str(project.id),
                name=project.title or project.name,
                goal=self._project_goal(project=project, truth_nodes=truth_nodes),
                stage=project_stage,
                threadId=thread_id,
            ),
            thread=PMFlowThread(
                id=thread_id,
                title=launch_thread.title if launch_thread is not None else f"{project.title or project.name} 项目线程",
                projectId=str(project.id),
                objective=launch_thread.objective if launch_thread is not None else self._project_goal(project=project, truth_nodes=truth_nodes),
                extractionStatus="idle",
                messageIds=[str(message.id) for message in launch_messages],
            ),
            messages=self._project_messages(launch_messages),
            changeSets=review_projection["change_sets"],
            changeItems=review_projection["change_items"],
            evidence=review_projection["evidence"],
            requirements=requirements,
            decisions=self._project_decisions(truth_nodes),
            tasks=task_items,
            risks=self._project_risks(truth_nodes),
            modelDecisions=await self._project_model_decisions(project_id=project_id),
            testRuns=test_runs,
            releases=releases,
            metricSignals=metric_signals,
            feed=self._project_feed(
                execution_summary=execution_summary,
                test_runs=test_runs,
                releases=releases,
                metric_signals=metric_signals,
            ),
            approvals=review_projection["approvals"],
            activeChangeSetId=active_change_set_id,
            selectedTaskId=execution_summary.execution.taskId,
            execution=execution_summary.execution,
            projectMemory=memory,
            projectionWarnings=warnings + execution_summary.warnings,
        )

    async def get_execution_summary(
        self,
        *,
        project_id: int,
        current_user_id: Optional[int] = None,
    ) -> ExecutionSummaryProjection:
        await self._load_project(project_id=project_id)
        control = AIProjectControlService(self.db, tenant_id=self.tenant_id)
        dev_sync = await control.get_project_dev_sync_summary(
            project_id=int(project_id),
            current_user_id=current_user_id,
        )
        latest_package = await self._load_latest_prompt_package(project_id=project_id)
        warnings: list[str] = []
        events = list(dev_sync.get("recent_progress_events") or [])
        active_tasks = list(dev_sync.get("active_dev_tasks") or [])
        repos = list(dev_sync.get("repo_bindings") or [])
        deliveries = list(dev_sync.get("recent_deliveries") or [])

        if not repos:
            warnings.append("repo_bindings_empty: no active repo binding is available for this project.")
        if not active_tasks:
            warnings.append("active_dev_tasks_empty: no active dev task is available in dev-sync summary.")
        if not latest_package:
            warnings.append("prompt_package_empty: no prompt package has been generated for this project.")
        if not events:
            warnings.append("progress_events_empty: no execution progress events have been recorded.")

        task_id = self._pick_execution_task_id(active_tasks=active_tasks, latest_package=latest_package, events=events)
        latest_event = events[0] if events else None
        status = self._map_execution_status(
            task_status=self._pick_task_status(active_tasks, task_id),
            event=latest_event,
        )
        prompt = latest_package.package_text if latest_package is not None else ""
        commit_log = self._event_commit_log(events)
        pr_status = self._project_pr_status(active_tasks=active_tasks, task_id=task_id, events=events)
        test_summary = self._project_test_summary(events)
        writeback = self._project_writeback_status(latest_event)

        return ExecutionSummaryProjection(
            projectId=str(project_id),
            execution=PMFlowExecutionContext(
                taskId=task_id,
                status=status,
                prompt=prompt,
                commitLog=commit_log,
                prStatus=pr_status,
                testSummary=test_summary,
                writebackStatus=writeback,
            ),
            devSync=dev_sync.get("summary") or {},
            promptPackage=None if latest_package is None else self._serialize_prompt_package(latest_package),
            repoBindings=repos,
            activeDevTasks=active_tasks,
            recentProgressEvents=events,
            recentDeliveries=deliveries,
            warnings=warnings,
        )

    async def _load_project(self, *, project_id: int) -> Project:
        result = await self.db.execute(
            select(Project).where(and_(Project.id == int(project_id), Project.tenant_id == self.tenant_id))
        )
        project = result.scalar_one_or_none()
        if project is None:
            raise ValueError("project_not_found")
        return project

    async def _load_truth_nodes(self, *, project_id: int) -> list[TruthNode]:
        result = await self.db.execute(
            select(TruthNode)
            .where(and_(TruthNode.tenant_id == self.tenant_id, TruthNode.project_id == int(project_id)))
            .order_by(TruthNode.updated_at.desc(), TruthNode.created_at.desc(), TruthNode.id.desc())
        )
        return list(result.scalars().all())

    async def _load_tasks(self, *, project_id: int) -> list[tuple[Task, Optional[User]]]:
        result = await self.db.execute(
            select(Task, User)
            .outerjoin(User, User.id == Task.assignee_id)
            .where(and_(Task.tenant_id == self.tenant_id, Task.project_id == int(project_id)))
            .order_by(Task.updated_at.desc(), Task.created_at.desc(), Task.id.desc())
        )
        return [(task, user) for task, user in result.all()]

    async def _load_latest_prompt_package(self, *, project_id: int) -> Optional[PromptPackage]:
        result = await self.db.execute(
            select(PromptPackage)
            .where(and_(PromptPackage.tenant_id == self.tenant_id, PromptPackage.project_id == int(project_id)))
            .order_by(PromptPackage.updated_at.desc(), PromptPackage.created_at.desc(), PromptPackage.id.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def _load_project_thread(self, *, project_id: int) -> tuple[Optional[PMFlowThreadModel], list[PMFlowThreadMessage]]:
        thread_result = await self.db.execute(
            select(PMFlowThreadModel)
            .where(
                and_(
                    PMFlowThreadModel.tenant_id == self.tenant_id,
                    PMFlowThreadModel.project_id == int(project_id),
                    PMFlowThreadModel.status == "active",
                )
            )
            .order_by(
                (PMFlowThreadModel.source == "project_launch").desc(),
                PMFlowThreadModel.created_at.asc(),
                PMFlowThreadModel.id.asc(),
            )
            .limit(1)
        )
        thread = thread_result.scalar_one_or_none()
        if thread is None:
            return None, []
        message_result = await self.db.execute(
            select(PMFlowThreadMessage)
            .where(
                and_(
                    PMFlowThreadMessage.tenant_id == self.tenant_id,
                    PMFlowThreadMessage.project_id == int(project_id),
                    PMFlowThreadMessage.thread_id == int(thread.id),
                )
            )
            .order_by(PMFlowThreadMessage.created_at.asc(), PMFlowThreadMessage.id.asc())
        )
        return thread, list(message_result.scalars().all())

    def _project_messages(self, rows: list[PMFlowThreadMessage]) -> list[dict[str, Any]]:
        return [
            {
                "id": str(row.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" if row.status == "sent" else ("analyzing" if row.status == "running" else "resolved"),
                "source": row.source,
                "metadata": row.message_metadata or {},
            }
            for row in rows
        ]

    async def _load_review_projection(self, *, project_id: int) -> dict[str, list[dict[str, Any]]]:
        change_set_result = await self.db.execute(
            select(PMFlowChangeSet)
            .where(and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.project_id == int(project_id)))
            .order_by(PMFlowChangeSet.updated_at.desc(), PMFlowChangeSet.created_at.desc(), PMFlowChangeSet.id.desc())
        )
        change_sets = list(change_set_result.scalars().all())
        if not change_sets:
            return {"change_sets": [], "change_items": [], "evidence": [], "approvals": []}
        change_set_ids = [int(row.id) for row in change_sets]
        item_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.in_(change_set_ids),
                )
            )
            .order_by(PMFlowChangeItem.created_at.asc(), PMFlowChangeItem.id.asc())
        )
        item_rows = list(item_result.scalars().all())
        evidence_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.in_(change_set_ids),
                )
            )
            .order_by(PMFlowEvidenceLink.created_at.asc(), PMFlowEvidenceLink.id.asc())
        )
        evidence_rows = list(evidence_result.scalars().all())
        approval_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.in_(change_set_ids),
                )
            )
            .order_by(PMFlowApproval.created_at.asc(), PMFlowApproval.id.asc())
        )
        approval_rows = list(approval_result.scalars().all())
        items_by_set: dict[int, list[PMFlowChangeItem]] = {}
        evidence_by_set: dict[int, list[PMFlowEvidenceLink]] = {}
        for item in item_rows:
            items_by_set.setdefault(int(item.change_set_id), []).append(item)
        for evidence in evidence_rows:
            evidence_by_set.setdefault(int(evidence.change_set_id), []).append(evidence)
        return {
            "change_sets": [
                PMFlowChangeReviewService.serialize_change_set(row, items_by_set.get(int(row.id), []))
                for row in change_sets
            ],
            "change_items": [
                PMFlowChangeReviewService.serialize_change_item(item, evidence_by_set.get(int(item.change_set_id), []))
                for item in item_rows
            ],
            "evidence": [PMFlowChangeReviewService.serialize_evidence(row) for row in evidence_rows],
            "approvals": [PMFlowChangeReviewService.serialize_approval(row) for row in approval_rows],
        }

    @staticmethod
    def _active_change_set_id(change_sets: list[dict[str, Any]]) -> Optional[str]:
        for row in change_sets:
            if row.get("rawStatus") == "open" or row.get("status") == "open":
                return str(row.get("id"))
        return None

    @staticmethod
    def _review_loop_stage(*, change_sets: list[dict[str, Any]], change_items: list[dict[str, Any]]) -> str:
        active = PMFlowOSProjectionService._active_change_set_id(change_sets)
        if active is None:
            return "command_idle"
        if any(str(item.get("changeSetId")) == active and item.get("status") == "pending_review" for item in change_items):
            return "review_open"
        return "extraction_ready"

    def _project_tasks(self, rows: list[tuple[Task, Optional[User]]]) -> list[PMFlowTask]:
        return [
            PMFlowTask(
                id=str(task.id),
                title=task.title,
                summary=task.description or "",
                stage=self._task_stage(task.status),
                status=self._object_status(task.status),
                assignee=user.name if user is not None else "未分配",
                acceptanceCriteria=self._task_acceptance_criteria(task),
                requirementId=str(task.truth_node_id) if task.truth_node_id is not None else None,
            )
            for task, user in rows
        ]

    def _project_requirements(self, *, truth_nodes: list[TruthNode], tasks: list[PMFlowTask]) -> list[PMFlowRequirement]:
        linked_by_node: dict[str, list[str]] = {}
        for task in tasks:
            if task.requirementId:
                linked_by_node.setdefault(str(task.requirementId), []).append(task.id)
        requirements: list[PMFlowRequirement] = []
        for node in truth_nodes:
            node_type = self._node_type(node)
            if node_type not in {"requirement", "charter"}:
                continue
            requirements.append(
                PMFlowRequirement(
                    id=str(node.id),
                    title=node.title,
                    summary=self._node_summary(node),
                    stage=self._stage_from_payload(node.content, default="planning"),
                    status=self._truth_status(node.status),
                    linkedTaskIds=linked_by_node.get(str(node.id), []),
                )
            )
        return requirements

    def _project_decisions(self, truth_nodes: list[TruthNode]) -> list[PMFlowDecision]:
        decisions: list[PMFlowDecision] = []
        for node in truth_nodes:
            if self._node_type(node) != "decision":
                continue
            content = self._payload(node.content)
            decisions.append(
                PMFlowDecision(
                    id=str(node.id),
                    title=node.title,
                    summary=self._node_summary(node),
                    stage=self._stage_from_payload(content, default="planning"),
                    status=self._truth_status(node.status),
                    owner=str(content.get("owner") or content.get("decision_owner") or "未分配"),
                )
            )
        return decisions

    def _project_risks(self, truth_nodes: list[TruthNode]) -> list[PMFlowRisk]:
        risks: list[PMFlowRisk] = []
        for node in truth_nodes:
            if self._node_type(node) != "risk":
                continue
            content = self._payload(node.content)
            risks.append(
                PMFlowRisk(
                    id=str(node.id),
                    title=node.title,
                    summary=self._node_summary(node),
                    stage=self._stage_from_payload(content, default="planning"),
                    status=self._truth_status(node.status),
                    level=self._risk_level(content.get("level") or content.get("risk_level") or node.status),
                    mitigation=str(content.get("mitigation") or content.get("mitigation_plan") or "待补充缓解策略"),
                )
            )
        return risks

    async def _project_model_decisions(self, *, project_id: int) -> list[PMFlowModelDecision]:
        binding_rows = await self.db.execute(
            select(ProjectModelBinding, ModelCatalogItem)
            .join(ModelCatalogItem, ModelCatalogItem.id == ProjectModelBinding.model_catalog_item_id)
            .where(
                and_(
                    ProjectModelBinding.tenant_id == self.tenant_id,
                    ProjectModelBinding.project_id == int(project_id),
                )
            )
            .order_by(ProjectModelBinding.updated_at.desc(), ProjectModelBinding.created_at.desc())
            .limit(10)
        )
        decisions: list[PMFlowModelDecision] = []
        for binding, catalog in binding_rows.all():
            reason = self._payload(binding.selection_reason)
            decisions.append(
                PMFlowModelDecision(
                    id=f"model-binding-{binding.id}",
                    title=f"模型绑定：{catalog.display_name}",
                    summary=self._short_text(reason) or f"{catalog.provider} / {catalog.model_key}",
                    stage="planning",
                    status="confirmed" if str(binding.status).lower() == "active" else "draft",
                    rationale=self._short_text(reason) or "来自 project_model_bindings.selection_reason",
                )
            )
        snapshot_rows = await self.db.execute(
            select(ModelRecommendationSnapshot)
            .where(
                and_(
                    ModelRecommendationSnapshot.tenant_id == self.tenant_id,
                    ModelRecommendationSnapshot.project_id == int(project_id),
                )
            )
            .order_by(ModelRecommendationSnapshot.updated_at.desc(), ModelRecommendationSnapshot.created_at.desc())
            .limit(5)
        )
        for snapshot in snapshot_rows.scalars().all():
            first = next(iter(snapshot.recommendations or []), {})
            decisions.append(
                PMFlowModelDecision(
                    id=f"model-recommendation-{snapshot.id}",
                    title=f"模型推荐快照：{snapshot.recommendation_scope}",
                    summary=self._short_text(first) or "模型推荐已生成，等待人工绑定确认。",
                    stage="planning",
                    status="draft",
                    rationale="来自 model_recommendation_snapshots；A3 不把推荐快照标记为已确认事实。",
                )
            )
        return decisions

    async def _project_test_runs(self, *, project_id: int) -> list[PMFlowTestRun]:
        runs: list[PMFlowTestRun] = []
        eval_result = await self.db.execute(
            select(ModelEvalRun)
            .where(and_(ModelEvalRun.tenant_id == self.tenant_id, ModelEvalRun.project_id == int(project_id)))
            .order_by(ModelEvalRun.updated_at.desc(), ModelEvalRun.created_at.desc())
            .limit(10)
        )
        for run in eval_result.scalars().all():
            result = self._test_result(run.status, run.summary)
            if result is None:
                continue
            runs.append(
                PMFlowTestRun(
                    id=f"model-eval-{run.id}",
                    title="模型效果验证",
                    summary=self._short_text(run.summary) or str(run.status),
                    status="done" if result == "passed" else "blocked",
                    result=result,
                )
            )
        progress_result = await self.db.execute(
            select(ProgressEvent)
            .where(
                and_(
                    ProgressEvent.tenant_id == self.tenant_id,
                    ProgressEvent.project_id == int(project_id),
                    ProgressEvent.event_type.ilike("%test%"),
                )
            )
            .order_by(ProgressEvent.occurred_at.desc(), ProgressEvent.created_at.desc())
            .limit(10)
        )
        for event in progress_result.scalars().all():
            result = self._test_result(event.derived_task_status or event.event_type, event.payload)
            if result is None:
                continue
            runs.append(
                PMFlowTestRun(
                    id=f"progress-test-{event.id}",
                    title=event.summary or "执行桥测试回写",
                    summary=event.summary or self._short_text(event.payload) or event.event_type,
                    status="done" if result == "passed" else "blocked",
                    result=result,
                    linkedTaskId=str(event.task_id),
                )
            )
        return runs

    async def _project_releases(self, *, truth_nodes: list[TruthNode], project_id: int) -> list[PMFlowRelease]:
        validation_result = await self.db.execute(
            select(ReleaseValidationRun)
            .where(and_(ReleaseValidationRun.tenant_id == self.tenant_id, ReleaseValidationRun.project_id == int(project_id)))
            .order_by(ReleaseValidationRun.updated_at.desc(), ReleaseValidationRun.created_at.desc())
        )
        validation_by_release = {
            int(run.release_node_id): run
            for run in validation_result.scalars().all()
            if run.release_node_id is not None
        }
        releases: list[PMFlowRelease] = []
        for node in truth_nodes:
            if self._node_type(node) != "release":
                continue
            content = self._payload(node.content)
            validation = validation_by_release.get(int(node.id))
            releases.append(
                PMFlowRelease(
                    id=str(node.id),
                    title=node.title,
                    summary=self._node_summary(node),
                    status=self._release_status(node.status, validation.status if validation else None),
                    gateItems=self._release_gate_items(content, validation),
                )
            )
        return releases

    async def _project_metric_signals(self, *, project_id: int) -> list[PMFlowMetricSignal]:
        signals: list[PMFlowMetricSignal] = []
        feedback_result = await self.db.execute(
            select(FeedbackIngest)
            .where(and_(FeedbackIngest.tenant_id == self.tenant_id, FeedbackIngest.project_id == int(project_id)))
            .order_by(FeedbackIngest.updated_at.desc(), FeedbackIngest.created_at.desc())
            .limit(10)
        )
        for feedback in feedback_result.scalars().all():
            for key, value in self._payload(feedback.metrics).items():
                signals.append(
                    PMFlowMetricSignal(
                        id=f"feedback-{feedback.id}-{key}",
                        title=str(key),
                        summary=feedback.title or feedback.content[:120],
                        value=str(value),
                        direction="flat",
                    )
                )
        revenue_result = await self.db.execute(
            select(RevenueEvent)
            .where(and_(RevenueEvent.tenant_id == self.tenant_id, RevenueEvent.project_id == int(project_id)))
            .order_by(RevenueEvent.occurred_at.desc(), RevenueEvent.created_at.desc())
            .limit(10)
        )
        for event in revenue_result.scalars().all():
            signals.append(
                PMFlowMetricSignal(
                    id=f"revenue-{event.id}",
                    title=f"收入事件：{event.category}",
                    summary=str(event.external_ref or event.source_type or "收入回写"),
                    value=f"{event.currency} {self._jsonable(event.amount_usd)}",
                    direction="up" if float(event.amount_usd or 0) > 0 else "flat",
                )
            )
        roi_result = await self.db.execute(
            select(ROISnapshot)
            .where(and_(ROISnapshot.tenant_id == self.tenant_id, ROISnapshot.project_id == int(project_id)))
            .order_by(ROISnapshot.updated_at.desc(), ROISnapshot.created_at.desc())
            .limit(5)
        )
        for snapshot in roi_result.scalars().all():
            signals.append(
                PMFlowMetricSignal(
                    id=f"roi-{snapshot.id}",
                    title=f"ROI 观察：{snapshot.range_key}",
                    summary=self._short_text(snapshot.snapshot_payload) or "ROI 快照",
                    value=str(self._jsonable(snapshot.roi_ratio)),
                    direction="up" if float(snapshot.roi_ratio or 0) > 0 else "flat",
                )
            )
        return signals

    async def _project_memory(
        self,
        *,
        project: Project,
        truth_nodes: list[TruthNode],
        task_ids: list[int],
    ) -> PMFlowProjectMemory:
        truth_ids = [int(node.id) for node in truth_nodes]
        clauses = [
            and_(SemanticFact.entity_type == "project", SemanticFact.entity_id == int(project.id)),
        ]
        if truth_ids:
            clauses.append(and_(SemanticFact.entity_type.in_(["truth_node", "truth_nodes"]), SemanticFact.entity_id.in_(truth_ids)))
        if task_ids:
            clauses.append(and_(SemanticFact.entity_type == "task", SemanticFact.entity_id.in_(task_ids)))
        result = await self.db.execute(
            select(SemanticFact)
            .where(and_(SemanticFact.tenant_id == self.tenant_id, SemanticFact.is_active.is_(True), or_(*clauses)))
            .order_by(SemanticFact.updated_at.desc(), SemanticFact.created_at.desc())
            .limit(100)
        )
        semantic_facts = [
            {
                "id": str(fact.id),
                "source": "semantic_facts",
                "entityType": fact.entity_type,
                "entityId": str(fact.entity_id),
                "namespace": fact.fact_namespace,
                "key": fact.fact_key,
                "value": fact.fact_value_text,
                "confidence": fact.confidence,
                "evidenceSourceType": fact.evidence_source_type,
                "evidenceSourceId": None if fact.evidence_source_id is None else str(fact.evidence_source_id),
                "evidenceExcerpt": fact.evidence_excerpt,
            }
            for fact in result.scalars().all()
        ]
        truth_facts = [
            {
                "id": str(node.id),
                "source": "truth_nodes",
                "nodeType": node.node_type,
                "title": node.title,
                "summary": self._node_summary(node),
                "status": node.status,
                "updatedAt": node.updated_at.isoformat() if node.updated_at else None,
            }
            for node in truth_nodes
        ]
        return PMFlowProjectMemory(
            truthNodeFacts=truth_facts,
            semanticFacts=semantic_facts,
            sourceSummary={
                "truth_nodes": len(truth_facts),
                "semantic_facts": len(semantic_facts),
            },
        )

    def _project_feed(
        self,
        *,
        execution_summary: ExecutionSummaryProjection,
        test_runs: list[PMFlowTestRun],
        releases: list[PMFlowRelease],
        metric_signals: list[PMFlowMetricSignal],
    ) -> list[PMFlowFeedEvent]:
        feed: list[PMFlowFeedEvent] = []
        for event in execution_summary.recentProgressEvents[:5]:
            feed.append(
                PMFlowFeedEvent(
                    id=f"progress-{event.get('id')}",
                    kind="execution",
                    title=event.get("event_type") or "执行进展",
                    summary=event.get("summary") or event.get("workflow_state") or "执行事件已回写",
                    timestamp=str(event.get("occurred_at") or ""),
                )
            )
        for run in test_runs[:3]:
            feed.append(PMFlowFeedEvent(id=f"feed-{run.id}", kind="test", title=run.title, summary=run.summary, timestamp=""))
        for release in releases[:3]:
            feed.append(PMFlowFeedEvent(id=f"feed-{release.id}", kind="release", title=release.title, summary=release.summary, timestamp=""))
        for signal in metric_signals[:3]:
            feed.append(PMFlowFeedEvent(id=f"feed-{signal.id}", kind="growth", title=signal.title, summary=signal.summary, timestamp=""))
        return feed

    def _project_goal(self, *, project: Project, truth_nodes: list[TruthNode]) -> str:
        charter = next((node for node in truth_nodes if self._node_type(node) == "charter"), None)
        if charter is not None:
            content = self._payload(charter.content)
            for key in ("business_goal", "goal", "objective", "summary"):
                if content.get(key):
                    return self._short_text(content.get(key))
        return project.description or project.title or project.name

    def _infer_project_stage(
        self,
        *,
        truth_nodes: list[TruthNode],
        tasks: list[tuple[Task, Optional[User]]],
        execution: PMFlowExecutionContext,
    ) -> str:
        if any(self._node_type(node) == "release" for node in truth_nodes):
            return "release"
        if execution.status in {"coding", "pr_open", "testing", "synced"}:
            return "development"
        if any(str(task.status).lower() in {"in_progress", "review", "blocked"} for task, _ in tasks):
            return "development"
        if truth_nodes:
            return "planning"
        return "idea"

    def _task_acceptance_criteria(self, task: Task) -> list[str]:
        payloads = [self._payload(task.manual_overrides), self._payload(task.ai_breakdown_payload)]
        for payload in payloads:
            for key in ("acceptance_criteria", "acceptanceCriteria", "criteria"):
                value = payload.get(key)
                if isinstance(value, list):
                    return [str(item) for item in value if str(item).strip()]
        return []

    def _node_summary(self, node: TruthNode) -> str:
        return self._short_text(node.content) or node.title

    def _node_type(self, node: TruthNode) -> str:
        return str(node.node_type or "").strip().lower()

    def _payload(self, value: Any) -> dict[str, Any]:
        return value if isinstance(value, dict) else {}

    def _short_text(self, value: Any) -> str:
        if value is None:
            return ""
        if isinstance(value, str):
            return value.strip()
        if isinstance(value, (int, float, Decimal)):
            return str(self._jsonable(value))
        if isinstance(value, list):
            return "；".join(str(item) for item in value[:5])
        if isinstance(value, dict):
            for key in ("summary", "description", "business_goal", "goal", "rationale", "reason", "conclusion", "content"):
                if value.get(key):
                    return self._short_text(value.get(key))
            pairs = []
            for key, item in list(value.items())[:5]:
                if isinstance(item, (dict, list)):
                    continue
                pairs.append(f"{key}: {item}")
            return "；".join(pairs)
        return str(value)

    def _stage_from_payload(self, payload: Any, *, default: str) -> str:
        stage = str(self._payload(payload).get("stage") or default).strip().lower()
        if stage in {"idea", "validation", "planning", "development", "testing", "release", "growth"}:
            return stage
        return default

    def _truth_status(self, status: str) -> str:
        value = str(status or "").strip().lower()
        if value in {"aligned", "confirmed", "active", "verified", "approved"}:
            return "confirmed"
        if value in {"in_progress", "review", "aligning", "running"}:
            return "in_progress"
        if value in {"blocked", "failed"}:
            return "blocked"
        if value in {"done", "completed"}:
            return "done"
        if value in {"released", "live", "online"}:
            return "released"
        return "draft"

    def _object_status(self, status: str) -> str:
        value = str(status or "").strip().lower()
        if value in {"done", "completed", "closed"}:
            return "done"
        if value in {"in_progress", "review", "running"}:
            return "in_progress"
        if value in {"blocked", "failed"}:
            return "blocked"
        if value in {"released", "live", "online"}:
            return "released"
        if value in {"confirmed", "aligned", "active"}:
            return "confirmed"
        return "draft"

    def _task_stage(self, status: str) -> str:
        value = str(status or "").strip().lower()
        if value in {"done", "completed", "review"}:
            return "testing"
        return "development"

    def _risk_level(self, value: Any) -> str:
        text = str(value or "").strip().lower()
        if text in {"high", "高", "critical", "urgent"}:
            return "high"
        if text in {"low", "低"}:
            return "low"
        return "medium"

    def _release_status(self, node_status: str, validation_status: Optional[str]) -> str:
        value = str(validation_status or node_status or "").strip().lower()
        if value in {"live", "online", "released", "verified"}:
            return "live"
        if value in {"releasing", "shipping"}:
            return "shipping"
        if value in {"ready", "passed", "success", "done", "completed"}:
            return "ready"
        if value in {"watch", "observing"}:
            return "watch"
        return "check_pending"

    def _release_gate_items(self, content: dict[str, Any], validation: Optional[ReleaseValidationRun]) -> list[dict[str, Any]]:
        source = []
        if validation is not None:
            source = list(validation.risk_items or [])
            summary = self._payload(validation.validation_summary)
            hits = summary.get("success_criteria_hits")
            if isinstance(hits, list):
                source = hits
        if not source:
            source = list(content.get("gate_items") or content.get("verification_items") or [])
        items = []
        for item in source:
            if not isinstance(item, dict):
                continue
            label = str(item.get("label") or item.get("title") or item.get("name") or item.get("criterion") or "发布检查项")
            passed = bool(item.get("passed") or item.get("hit") or item.get("status") in {"done", "passed", "success"})
            items.append({"label": label, "passed": passed})
        return items

    def _test_result(self, status: Any, payload: Any) -> Optional[str]:
        text = f"{status} {self._short_text(payload)}".lower()
        if any(token in text for token in ("passed", "pass", "success", "done", "completed", "通过")):
            return "passed"
        if any(token in text for token in ("failed", "fail", "blocked", "error", "失败", "阻塞")):
            return "failed"
        return None

    def _map_execution_status(self, *, task_status: Optional[str], event: Optional[dict[str, Any]]) -> str:
        event_type = str((event or {}).get("event_type") or "").lower()
        workflow_state = str((event or {}).get("workflow_state") or "").lower()
        derived = str((event or {}).get("derived_task_status") or task_status or "").lower()
        combined = " ".join([event_type, workflow_state, derived])
        if "test" in combined:
            return "testing"
        if "pull_request" in combined or "pr" in combined or "review" in combined:
            return "pr_open"
        if "commit" in combined or "coding" in combined or "in_progress" in combined:
            return "coding"
        if "pushed" in combined or "queued" in combined:
            return "pushed"
        if "done" in combined or "completed" in combined or "synced" in combined:
            return "synced"
        if "failed" in combined or "blocked" in combined:
            return "failed"
        return "queued"

    def _pick_execution_task_id(
        self,
        *,
        active_tasks: list[dict[str, Any]],
        latest_package: Optional[PromptPackage],
        events: list[dict[str, Any]],
    ) -> Optional[str]:
        if latest_package is not None and latest_package.task_id is not None:
            return str(latest_package.task_id)
        if active_tasks:
            return str(active_tasks[0].get("id"))
        if events:
            task = events[0].get("task")
            if isinstance(task, dict) and task.get("id"):
                return str(task.get("id"))
        return None

    def _pick_task_status(self, active_tasks: list[dict[str, Any]], task_id: Optional[str]) -> Optional[str]:
        if task_id is None:
            return None
        for task in active_tasks:
            if str(task.get("id")) == str(task_id):
                return str(task.get("status") or "")
        return None

    def _event_commit_log(self, events: list[dict[str, Any]]) -> list[str]:
        logs = []
        for event in events[:8]:
            commit = str(event.get("commit_sha") or "").strip()
            summary = str(event.get("summary") or event.get("event_type") or "").strip()
            if commit:
                logs.append(f"{summary} · {commit[:12]}")
            elif summary:
                logs.append(summary)
        return logs

    def _project_pr_status(self, *, active_tasks: list[dict[str, Any]], task_id: Optional[str], events: list[dict[str, Any]]) -> str:
        for task in active_tasks:
            if task_id is not None and str(task.get("id")) != str(task_id):
                continue
            code_link = task.get("code_link")
            if isinstance(code_link, dict) and code_link.get("pr_number"):
                return f"PR #{code_link.get('pr_number')} · {code_link.get('status') or 'linked'}"
        for event in events:
            pr_number = event.get("pr_number")
            if pr_number:
                return f"PR #{pr_number}"
        return "pending"

    def _project_test_summary(self, events: list[dict[str, Any]]) -> str:
        for event in events:
            if "test" in str(event.get("event_type") or "").lower():
                return str(event.get("summary") or event.get("workflow_state") or "test event recorded")
        return "pending"

    def _project_writeback_status(self, event: Optional[dict[str, Any]]) -> str:
        if event is None:
            return "pending"
        return str(event.get("workflow_state") or event.get("summary") or event.get("event_type") or "progress event recorded")

    def _serialize_prompt_package(self, package: PromptPackage) -> dict[str, Any]:
        return {
            "id": str(package.id),
            "project_id": str(package.project_id),
            "charter_id": str(package.charter_id),
            "task_id": None if package.task_id is None else str(package.task_id),
            "status": package.status,
            "package_kind": package.package_kind,
            "package_text": package.package_text,
            "package_payload": package.package_payload,
            "created_at": package.created_at.isoformat() if package.created_at else None,
        }

    def _latest_iso(self, values: list[Optional[datetime]]) -> Optional[str]:
        valid = [value for value in values if value is not None]
        if not valid:
            return None
        return max(valid).isoformat()

    def _jsonable(self, value: Any) -> Any:
        if isinstance(value, Decimal):
            return float(value)
        return value