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 sqlalchemy.orm.attributes import flag_modified

from ..models import (
    FeedbackIngest,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowEvidenceLink,
    PMFlowThreadMessage,
    PMFlowThread as PMFlowThreadModel,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RepoBinding,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Task,
    TruthNode,
    User,
)
from ..schemas.pmflow_os import (
    ExecutionSummaryProjection,
    PMFlowDecision,
    PMFlowExecutionHandoffConsumeRequest,
    PMFlowExecutionHandoffRequest,
    PMFlowExecutionProgressEventRequest,
    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
from ..utils.timezone import now_cn_naive


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,
            ),
            releaseGate=self._build_release_gate(events=events, deliveries=deliveries),
            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 create_execution_handoff(
        self,
        *,
        project_id: int,
        body: PMFlowExecutionHandoffRequest,
        current_user_id: Optional[int] = None,
    ) -> ExecutionSummaryProjection:
        project = await self._load_project(project_id=project_id)
        task = await self._load_project_task(project_id=project_id, task_id=body.task_id)
        repo = await self._load_project_repo_binding(project_id=project_id, repo_binding_id=body.repo_binding_id)
        prompt_package = await self._load_active_prompt_package_for_task(project_id=project_id, task_id=int(task.id))
        if prompt_package is None:
            prompt_package = await self._create_execution_prompt_package(
                project=project,
                task=task,
                repo=repo,
                runtime_target=body.runtime_target,
                note=body.note,
            )
        else:
            payload = dict(self._payload(prompt_package.package_payload))
            payload["latest_handoff"] = self._handoff_payload(
                task=task,
                repo=repo,
                runtime_target=body.runtime_target,
                note=body.note,
                prompt_package_id=int(prompt_package.id),
            )
            prompt_package.package_payload = payload
            flag_modified(prompt_package, "package_payload")

        event = ProgressEvent(
            tenant_id=self.tenant_id,
            task_id=int(task.id),
            project_id=int(project_id),
            user_id=int(current_user_id) if current_user_id is not None else None,
            repo_binding_id=int(repo.id) if repo is not None else None,
            event_type="handoff_created",
            source="pmflow_os",
            derived_task_status="pushed",
            summary=self._handoff_summary(task=task, repo=repo, runtime_target=body.runtime_target),
            payload=self._handoff_payload(
                task=task,
                repo=repo,
                runtime_target=body.runtime_target,
                note=body.note,
                prompt_package_id=int(prompt_package.id),
            ),
            occurred_at=now_cn_naive(),
        )
        self.db.add(event)
        await self.db.commit()
        return await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)

    async def consume_execution_handoff(
        self,
        *,
        project_id: int,
        prompt_package_id: int,
        body: PMFlowExecutionHandoffConsumeRequest,
        current_user_id: Optional[int] = None,
    ) -> ExecutionSummaryProjection:
        await self._load_project(project_id=project_id)
        prompt_package = await self._load_execution_prompt_package(
            project_id=project_id,
            prompt_package_id=prompt_package_id,
        )
        if prompt_package.task_id is None:
            raise ValueError("task_not_found")
        task = await self._load_project_task(project_id=project_id, task_id=str(prompt_package.task_id))
        payload = dict(self._payload(prompt_package.package_payload))
        latest_handoff = dict(self._payload(payload.get("latest_handoff")))
        runtime_target = body.runtime_target or str(latest_handoff.get("runtime_target") or payload.get("runtime_target") or "manual")
        repo_binding_id = str(latest_handoff.get("repo_binding_id") or "").strip() or None
        repo = await self._load_project_repo_binding(project_id=project_id, repo_binding_id=repo_binding_id)
        consumer_label = (body.consumer_label or runtime_target or "runtime").strip()
        consumed_at = now_cn_naive().isoformat()
        consumed_payload = self._handoff_consumed_payload(
            task=task,
            repo=repo,
            runtime_target=runtime_target,
            note=body.note,
            prompt_package_id=int(prompt_package.id),
            consumer_label=consumer_label,
            consumed_at=consumed_at,
        )
        latest_handoff.update(consumed_payload)
        payload["latest_handoff"] = latest_handoff
        payload.setdefault("handoff_consumptions", [])
        if isinstance(payload["handoff_consumptions"], list):
            payload["handoff_consumptions"] = [consumed_payload, *payload["handoff_consumptions"][:9]]
        prompt_package.package_payload = payload
        flag_modified(prompt_package, "package_payload")

        event = ProgressEvent(
            tenant_id=self.tenant_id,
            task_id=int(task.id),
            project_id=int(project_id),
            user_id=int(current_user_id) if current_user_id is not None else None,
            repo_binding_id=int(repo.id) if repo is not None else None,
            event_type="handoff_consumed",
            source="pmflow_os",
            derived_task_status="coding",
            summary=self._handoff_consumed_summary(task=task, runtime_target=runtime_target, consumer_label=consumer_label),
            payload=consumed_payload,
            occurred_at=now_cn_naive(),
        )
        self.db.add(event)
        await self.db.commit()
        return await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)

    async def record_execution_progress_event(
        self,
        *,
        project_id: int,
        body: PMFlowExecutionProgressEventRequest,
        current_user_id: Optional[int] = None,
    ) -> ExecutionSummaryProjection:
        await self._load_project(project_id=project_id)
        task = await self._load_project_task(project_id=project_id, task_id=body.task_id)
        repo = await self._load_project_repo_binding(project_id=project_id, repo_binding_id=body.repo_binding_id)
        payload = self._progress_event_payload(body=body, repo=repo)
        summary = self._progress_event_summary(body=body, task=task, repo=repo)
        control = AIProjectControlService(self.db, tenant_id=self.tenant_id)
        await control.create_or_apply_progress_event(
            task_id=int(task.id),
            project_id=int(project_id),
            event_type=body.event_type,
            source=body.source,
            payload=payload,
            summary=summary,
            user_id=int(current_user_id) if current_user_id is not None else None,
            repo_binding_id=int(repo.id) if repo is not None else None,
            event_source_id=body.event_source_id,
            occurred_at=body.occurred_at,
        )
        return await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)

    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_project_task(self, *, project_id: int, task_id: str) -> Task:
        try:
            normalized_task_id = int(str(task_id).strip())
        except (TypeError, ValueError) as exc:
            raise ValueError("task_not_found") from exc
        result = await self.db.execute(
            select(Task).where(
                and_(
                    Task.tenant_id == self.tenant_id,
                    Task.project_id == int(project_id),
                    Task.id == normalized_task_id,
                )
            )
        )
        task = result.scalar_one_or_none()
        if task is None:
            raise ValueError("task_not_found")
        return task

    async def _load_project_repo_binding(self, *, project_id: int, repo_binding_id: Optional[str]) -> Optional[RepoBinding]:
        if repo_binding_id is None or str(repo_binding_id).strip() == "":
            return None
        try:
            normalized_repo_id = int(str(repo_binding_id).strip())
        except (TypeError, ValueError) as exc:
            raise ValueError("repo_binding_not_found") from exc
        result = await self.db.execute(
            select(RepoBinding).where(
                and_(
                    RepoBinding.tenant_id == self.tenant_id,
                    RepoBinding.project_id == int(project_id),
                    RepoBinding.id == normalized_repo_id,
                    RepoBinding.status == "active",
                )
            )
        )
        repo = result.scalar_one_or_none()
        if repo is None:
            raise ValueError("repo_binding_not_found")
        return repo

    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_active_prompt_package_for_task(self, *, project_id: int, task_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),
                    PromptPackage.task_id == int(task_id),
                    PromptPackage.status == "active",
                )
            )
            .order_by(PromptPackage.updated_at.desc(), PromptPackage.created_at.desc(), PromptPackage.id.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def _load_execution_prompt_package(self, *, project_id: int, prompt_package_id: int) -> PromptPackage:
        result = await self.db.execute(
            select(PromptPackage).where(
                and_(
                    PromptPackage.tenant_id == self.tenant_id,
                    PromptPackage.project_id == int(project_id),
                    PromptPackage.id == int(prompt_package_id),
                )
            )
        )
        prompt_package = result.scalar_one_or_none()
        if prompt_package is None:
            raise ValueError("prompt_package_not_found")
        if prompt_package.package_kind != "execution_handoff":
            raise ValueError("execution_handoff_not_found")
        return prompt_package

    async def _create_execution_prompt_package(
        self,
        *,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        runtime_target: str,
        note: Optional[str],
    ) -> PromptPackage:
        truth_nodes = await self._load_truth_nodes(project_id=int(project.id))
        memory = await self._project_memory(project=project, truth_nodes=truth_nodes, task_ids=[int(task.id)])
        payload = self._execution_prompt_payload(
            project=project,
            task=task,
            repo=repo,
            runtime_target=runtime_target,
            note=note,
            memory=memory,
        )
        package = PromptPackage(
            tenant_id=self.tenant_id,
            project_id=int(project.id),
            charter_id=int(task.truth_node_id),
            task_id=int(task.id),
            package_kind="execution_handoff",
            status="active",
            package_text=self._execution_prompt_text(payload),
            package_payload=payload,
            generated_from={
                "source": "pmflow_os_execution_handoff",
                "project_id": str(project.id),
                "task_id": str(task.id),
                "repo_binding_id": None if repo is None else str(repo.id),
                "runtime_target": runtime_target,
                "generated_at": now_cn_naive().isoformat(),
            },
        )
        self.db.add(package)
        await self.db.flush()
        package.package_payload = {
            **payload,
            "latest_handoff": self._handoff_payload(
                task=task,
                repo=repo,
                runtime_target=runtime_target,
                note=note,
                prompt_package_id=int(package.id),
            ),
        }
        return package

    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 "failed" in combined or "blocked" in combined:
            return "failed"
        if "merged" in combined or "accepted" in combined or "done" in combined or "completed" in combined or "synced" in combined:
            return "synced"
        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 "branch" in combined or "pushed" in combined or "queued" in combined:
            return "pushed"
        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:
            event_type = str(event.get("event_type") or "").lower()
            workflow_state = str(event.get("workflow_state") or "").lower()
            if "test" in event_type or "check" in event_type or workflow_state == "testing":
                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 _build_release_gate(self, *, events: list[dict[str, Any]], deliveries: list[dict[str, Any]]) -> dict[str, Any]:
        evidence_events = [event for event in events if self._is_release_evidence_event(event)]
        if not evidence_events:
            return {
                "status": "empty",
                "title": "缺少执行证据",
                "summary": "还没有 branch、commit、PR、test 或 failed 回写，发布门禁不能生成判断。",
                "blockingReasons": [],
                "nextActions": ["先从 Execution Bridge 回写 branch / commit / PR / test 事件。"],
                "checks": [
                    {"key": "execution_evidence", "label": "执行证据", "status": "missing", "detail": "progress_events_empty"},
                ],
                "evidence": [],
                "updatedAt": None,
            }

        failed_events = [event for event in evidence_events if self._is_failed_evidence_event(event)]
        if failed_events:
            return {
                "status": "blocked",
                "title": "发布阻塞",
                "summary": "执行回写中存在失败或阻塞事件，发布门禁已关闭。",
                "blockingReasons": [self._event_reason(event) for event in failed_events[:5]],
                "nextActions": [
                    "先修复失败事件对应的任务或测试。",
                    "修复后重新回写 test_passed 或 checks_passed 证据。",
                ],
                "checks": self._release_gate_checks(evidence_events),
                "evidence": self._release_gate_evidence(evidence_events),
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }

        has_test_passed = any(self._event_type(event) in {"test_passed", "checks_passed"} for event in evidence_events)
        has_pr = any(self._event_type(event) in {"pr_opened", "pr_synced", "merged", "accepted"} or event.get("pr_number") for event in evidence_events)
        has_synced = any(self._event_type(event) in {"merged", "accepted"} or "synced" in self._event_text(event) for event in evidence_events)
        failed_deliveries = [item for item in deliveries if str(item.get("status") or "").lower() == "failed"]
        if failed_deliveries:
            return {
                "status": "blocked",
                "title": "发布阻塞",
                "summary": "GitHub delivery 存在失败记录，发布前需要先确认回写链路。",
                "blockingReasons": [
                    str(item.get("failure_reason") or item.get("delivery_id") or "GitHub delivery failed")
                    for item in failed_deliveries[:5]
                ],
                "nextActions": ["检查 GitHub webhook delivery 失败原因。", "恢复回写链路后重新生成发布判断。"],
                "checks": self._release_gate_checks(evidence_events),
                "evidence": self._release_gate_evidence(evidence_events),
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }

        if has_test_passed and has_synced:
            return {
                "status": "ready",
                "title": "可发布",
                "summary": "已有 PR/合并与通过测试证据，发布门禁给出可发布判断。",
                "blockingReasons": [],
                "nextActions": ["由负责人进行最终人工确认；PMFlow 不自动执行发布命令。"],
                "checks": self._release_gate_checks(evidence_events),
                "evidence": self._release_gate_evidence(evidence_events),
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }

        if has_test_passed:
            return {
                "status": "needs_confirmation",
                "title": "发布候选，需人工确认",
                "summary": "测试已通过，但仍需要负责人确认 PR 状态、发布范围和回滚预案。",
                "blockingReasons": [],
                "nextActions": [
                    "确认 PR 是否已审查或合并。",
                    "确认本次发布范围与回滚方案后再进入发布。",
                ],
                "checks": self._release_gate_checks(evidence_events),
                "evidence": self._release_gate_evidence(evidence_events),
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }

        return {
            "status": "needs_confirmation",
            "title": "需要人工确认",
            "summary": "已有执行回写，但缺少通过测试证据，不能直接给出可发布判断。",
            "blockingReasons": ["缺少 test_passed 或 checks_passed 证据。"],
            "nextActions": ["补充测试回写；若确需跳过测试，需要负责人显式确认。"],
            "checks": self._release_gate_checks(evidence_events),
            "evidence": self._release_gate_evidence(evidence_events),
            "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
        }

    def _release_gate_checks(self, events: list[dict[str, Any]]) -> list[dict[str, Any]]:
        return [
            {
                "key": "branch",
                "label": "分支证据",
                "status": "passed" if any(event.get("branch_name") or self._event_type(event) == "branch_created" for event in events) else "missing",
                "detail": self._first_event_detail(events, {"branch_created", "commit", "pr_opened", "pr_synced"}),
            },
            {
                "key": "commit",
                "label": "提交证据",
                "status": "passed" if any(event.get("commit_sha") or self._event_type(event) == "commit" for event in events) else "missing",
                "detail": self._first_event_detail(events, {"commit"}),
            },
            {
                "key": "pr",
                "label": "PR 证据",
                "status": "passed" if any(event.get("pr_number") or self._event_type(event) in {"pr_opened", "pr_synced", "merged", "accepted"} for event in events) else "missing",
                "detail": self._first_event_detail(events, {"pr_opened", "pr_synced", "merged", "accepted"}),
            },
            {
                "key": "test",
                "label": "测试证据",
                "status": "failed" if any(self._event_type(event) == "test_failed" for event in events) else ("passed" if any(self._event_type(event) in {"test_passed", "checks_passed"} for event in events) else "missing"),
                "detail": self._first_event_detail(events, {"test_passed", "checks_passed", "test_failed"}),
            },
            {
                "key": "failure",
                "label": "失败事件",
                "status": "failed" if any(self._is_failed_evidence_event(event) for event in events) else "passed",
                "detail": self._first_event_detail(events, {"failed", "blocked", "test_failed"}) or "未发现失败回写",
            },
        ]

    def _release_gate_evidence(self, events: list[dict[str, Any]]) -> list[dict[str, Any]]:
        return [
            {
                "id": str(event.get("id") or ""),
                "type": self._event_type(event),
                "summary": self._event_reason(event),
                "taskTitle": (event.get("task") or {}).get("title") if isinstance(event.get("task"), dict) else None,
                "repo": (event.get("repo") or {}).get("full_name") if isinstance(event.get("repo"), dict) else None,
                "branchName": event.get("branch_name"),
                "commitSha": event.get("commit_sha"),
                "prNumber": event.get("pr_number"),
                "occurredAt": event.get("occurred_at"),
            }
            for event in events[:12]
        ]

    @staticmethod
    def _event_type(event: dict[str, Any]) -> str:
        return str(event.get("event_type") or "").strip().lower()

    def _event_text(self, event: dict[str, Any]) -> str:
        return " ".join(
            [
                self._event_type(event),
                str(event.get("summary") or ""),
                str(event.get("workflow_state") or ""),
                str(event.get("derived_task_status") or ""),
            ]
        ).lower()

    def _is_release_evidence_event(self, event: dict[str, Any]) -> bool:
        event_type = self._event_type(event)
        return event_type in {
            "branch_created",
            "commit",
            "pr_opened",
            "pr_synced",
            "test_passed",
            "test_failed",
            "checks_passed",
            "failed",
            "blocked",
            "merged",
            "accepted",
        } or bool(event.get("branch_name") or event.get("commit_sha") or event.get("pr_number"))

    def _is_failed_evidence_event(self, event: dict[str, Any]) -> bool:
        text = self._event_text(event)
        return any(token in text for token in ("test_failed", "failed", "blocked", "失败", "阻塞", "error"))

    def _event_reason(self, event: dict[str, Any]) -> str:
        return str(event.get("summary") or event.get("workflow_state") or event.get("event_type") or "执行证据").strip()

    def _first_event_detail(self, events: list[dict[str, Any]], event_types: set[str]) -> str:
        for event in events:
            if self._event_type(event) in event_types:
                return self._event_reason(event)
        return ""

    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 _execution_prompt_payload(
        self,
        *,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        runtime_target: str,
        note: Optional[str],
        memory: PMFlowProjectMemory,
    ) -> dict[str, Any]:
        acceptance = self._task_acceptance_criteria(task)
        truth_facts = memory.truthNodeFacts[:12]
        semantic_facts = memory.semanticFacts[:12]
        return {
            "project": {
                "id": str(project.id),
                "name": project.title or project.name,
                "goal": project.description or project.title or project.name,
            },
            "task": {
                "id": str(task.id),
                "title": task.title,
                "description": task.description or "",
                "status": task.status,
                "priority": task.priority,
                "charter_id": str(task.truth_node_id),
                "acceptance_criteria": acceptance,
            },
            "repo": None
            if repo is None
            else {
                "id": str(repo.id),
                "provider": repo.provider,
                "full_name": f"{repo.owner}/{repo.repo_name}",
                "default_branch": repo.default_branch,
                "status": repo.status,
            },
            "runtime_target": runtime_target,
            "note": note,
            "project_memory": {
                "truth_node_facts": truth_facts,
                "semantic_facts": semantic_facts,
                "source_summary": memory.sourceSummary,
            },
            "boundary": {
                "execution": "PMFlow only creates the handoff contract and progress event. Local commands must run through Local Bridge or a compatible MCP host.",
                "host_config": "PMFlow backend must not write host ~/.codex/config.toml or execute local stdio commands directly.",
            },
        }

    def _execution_prompt_text(self, payload: dict[str, Any]) -> str:
        project = self._payload(payload.get("project"))
        task = self._payload(payload.get("task"))
        repo = payload.get("repo") if isinstance(payload.get("repo"), dict) else None
        memory = self._payload(payload.get("project_memory"))
        truth_facts = list(memory.get("truth_node_facts") or [])[:6]
        semantic_facts = list(memory.get("semantic_facts") or [])[:6]
        acceptance = list(task.get("acceptance_criteria") or [])
        lines = [
            "# PMFlow Execution Handoff",
            "",
            f"Project: {project.get('name') or project.get('id')}",
            f"Goal: {project.get('goal') or '未填写项目目标'}",
            "",
            "## Task",
            f"- ID: {task.get('id')}",
            f"- Title: {task.get('title')}",
            f"- Description: {task.get('description') or '未填写任务描述'}",
            f"- Status: {task.get('status')}",
            f"- Priority: {task.get('priority')}",
            "",
            "## Acceptance Criteria",
        ]
        lines.extend([f"- {item}" for item in acceptance] or ["- 未提供验收标准；执行前需要人工补齐。"])
        lines.extend(["", "## Repository"])
        if repo is None:
            lines.append("- No repo binding is available. Create a handoff draft only; do not infer repository, branch, or local path.")
        else:
            lines.extend(
                [
                    f"- Provider: {repo.get('provider')}",
                    f"- Repository: {repo.get('full_name')}",
                    f"- Default branch: {repo.get('default_branch')}",
                    f"- Binding status: {repo.get('status')}",
                ]
            )
        lines.extend(
            [
                "",
                "## Runtime Target",
                f"- {payload.get('runtime_target')}",
            ]
        )
        if payload.get("note"):
            lines.extend(["", "## Operator Note", str(payload.get("note"))])
        lines.extend(["", "## Project Memory"])
        if truth_facts:
            lines.append("Truth Nodes:")
            lines.extend(
                f"- {str(item.get('title') or item.get('id'))}: {str(item.get('summary') or item.get('status') or '')}"
                for item in truth_facts
                if isinstance(item, dict)
            )
        if semantic_facts:
            lines.append("Semantic Facts:")
            lines.extend(
                f"- {str(item.get('namespace') or 'fact')}.{str(item.get('key') or item.get('id'))}: {str(item.get('value') or '')}"
                for item in semantic_facts
                if isinstance(item, dict)
            )
        if not truth_facts and not semantic_facts:
            lines.append("- Project Memory has no truth/semantic facts for this task yet.")
        lines.extend(
            [
                "",
                "## Boundary",
                "- Do not execute local commands from PMFlow backend.",
                "- Use Local Bridge, Codex, GitHub agent, or manual runtime according to the selected target.",
            ]
        )
        return "\n".join(lines)

    def _handoff_payload(
        self,
        *,
        task: Task,
        repo: Optional[RepoBinding],
        runtime_target: str,
        note: Optional[str],
        prompt_package_id: int,
    ) -> dict[str, Any]:
        return {
            "workflow_state": "pushed",
            "handoff_state": "draft" if repo is None else "ready",
            "runtime_target": runtime_target,
            "prompt_package_id": str(prompt_package_id),
            "task_id": str(task.id),
            "repo_binding_id": None if repo is None else str(repo.id),
            "repository_full_name": None if repo is None else f"{repo.owner}/{repo.repo_name}",
            "note": note,
            "boundary": "PMFlow backend creates metadata only; execution must be picked up by Local Bridge, Codex, GitHub agent, or manual operator.",
        }

    def _handoff_summary(self, *, task: Task, repo: Optional[RepoBinding], runtime_target: str) -> str:
        target = f"{repo.owner}/{repo.repo_name}" if repo is not None else "未绑定仓库"
        return f"执行交接已创建：{task.title} -> {runtime_target} / {target}"

    def _handoff_consumed_payload(
        self,
        *,
        task: Task,
        repo: Optional[RepoBinding],
        runtime_target: str,
        note: Optional[str],
        prompt_package_id: int,
        consumer_label: str,
        consumed_at: str,
    ) -> dict[str, Any]:
        return {
            "workflow_state": "coding",
            "handoff_state": "consumed",
            "runtime_target": runtime_target,
            "consumer_label": consumer_label,
            "consumed_at": consumed_at,
            "prompt_package_id": str(prompt_package_id),
            "task_id": str(task.id),
            "repo_binding_id": None if repo is None else str(repo.id),
            "repository_full_name": None if repo is None else f"{repo.owner}/{repo.repo_name}",
            "note": note,
            "boundary": "PMFlow records handoff pickup only; command execution remains outside backend runtime.",
        }

    def _handoff_consumed_summary(self, *, task: Task, runtime_target: str, consumer_label: str) -> str:
        return f"执行交接已领取：{task.title} -> {runtime_target} / {consumer_label}"

    def _progress_event_payload(
        self,
        *,
        body: PMFlowExecutionProgressEventRequest,
        repo: Optional[RepoBinding],
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "workflow_state": body.workflow_state or self._default_workflow_state(body.event_type),
            "source": body.source,
            "boundary": "PMFlow records execution result metadata only; command execution stays in Local Bridge, Codex, GitHub agent, or manual runtime.",
        }
        for key, value in {
            "branch_name": body.branch_name,
            "commit_sha": body.commit_sha,
            "pr_number": body.pr_number,
            "test_summary": body.test_summary,
            "event_source_id": body.event_source_id,
        }.items():
            if value is not None and str(value).strip():
                payload[key] = value
        if repo is not None:
            payload["repo_binding_id"] = str(repo.id)
            payload["repository_full_name"] = f"{repo.owner}/{repo.repo_name}"
        return payload

    def _progress_event_summary(
        self,
        *,
        body: PMFlowExecutionProgressEventRequest,
        task: Task,
        repo: Optional[RepoBinding],
    ) -> str:
        if body.summary:
            return body.summary
        if body.test_summary:
            return body.test_summary
        repo_label = f"{repo.owner}/{repo.repo_name}" if repo is not None else "未绑定仓库"
        detail = body.commit_sha or (f"PR #{body.pr_number}" if body.pr_number else body.branch_name) or self._default_workflow_state(body.event_type)
        return f"执行结果回写：{task.title} / {body.event_type} / {repo_label} / {detail}"

    def _default_workflow_state(self, event_type: str) -> str:
        value = str(event_type or "").strip().lower()
        if value == "branch_created":
            return "pushed"
        if value == "commit":
            return "coding"
        if value in {"pr_opened", "pr_synced"}:
            return "in_review"
        if value in {"test_passed", "test_failed", "checks_passed"}:
            return "testing"
        if value in {"merged", "accepted"}:
            return "synced"
        if value in {"blocked", "failed"}:
            return "failed"
        return "manual_progress"

    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