PMFlow/backend/app/services/pmflow_os_projection_service.py

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

from __future__ import annotations

import hashlib
import hmac
import secrets
from datetime import datetime, timedelta
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 (
    ComplianceCheckRun,
    FeedbackIngest,
    IterationRecommendation,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowAIFeedbackEvent,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowEvidenceLink,
    PMFlowThreadMessage,
    PMFlowThread as PMFlowThreadModel,
    PMFlowExecutionRuntimeCredential,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RepoBinding,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Task,
    TruthNode,
    User,
    UserProject,
)
from ..schemas.pmflow_os import (
    ExecutionSummaryProjection,
    PMFlowDecision,
    PMFlowExecutionAcceptanceDecisionRequest,
    PMFlowArtifactReviewActionRequest,
    PMFlowArtifactReworkPackageRequest,
    PMFlowExecutionHandoffConsumeRequest,
    PMFlowExecutionHandoffRequest,
    PMFlowExecutionRuntimeCredentialCreateRequest,
    PMFlowExecutionRuntimeCredentialResponse,
    PMFlowExecutionMemoryWritebackRequest,
    PMFlowExecutionProgressEventRequest,
    PMFlowRuntimeExecutionPackageRequest,
    PMFlowRuntimeExecutionPackageResponse,
    PMFlowRuntimeExecutionResultRequest,
    PMFlowExecutionContext,
    PMFlowFeedEvent,
    PMFlowIterationRecommendationChangeSetDraftRequest,
    PMFlowIterationRecommendationGroupChangeSetDraftRequest,
    PMFlowMetricSignal,
    PMFlowModelDecision,
    PMFlowProjectMemory,
    PMFlowProjectSummary,
    PMFlowRelease,
    PMFlowReleaseFeedbackIngestRequest,
    PMFlowReleaseValidationRunCreateRequest,
    PMFlowRequirement,
    PMFlowRisk,
    PMFlowTask,
    PMFlowTestRun,
    PMFlowThread,
    ProjectTruthStateProjection,
)
from ..core.config import settings
from .ai_project_control_service import AIProjectControlService
from .pmflow_ai_feedback_service import PMFlowAIFeedbackService
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"])
        latest_writeback = self._latest_writeback_meta(review_projection["change_sets"])
        project_members = await self._load_project_members(project=project)
        reviewer_routing, reviewer_warnings = self._project_reviewer_routing(
            project_id=project_id,
            change_sets=review_projection["change_sets"],
            change_items=review_projection["change_items"],
            evidence=review_projection["evidence"],
            members=project_members,
        )
        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.")
        ai_feedback = await self._load_ai_feedback(project_id=project_id, current_user_id=current_user_id)

        return ProjectTruthStateProjection(
            loopStage=loop_stage,
            lastWritebackAt=latest_writeback.get("at")
            or self._latest_iso([node.updated_at for node in truth_nodes] + [task.updated_at for task, _ in tasks]),
            lastWritebackBy=latest_writeback.get("by"),
            lastWritebackById=latest_writeback.get("by_id"),
            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,
            aiFeedback=ai_feedback,
            feed=self._project_feed(
                review_projection=review_projection,
                reviewer_routing=reviewer_routing,
                execution_summary=execution_summary,
                test_runs=test_runs,
                releases=releases,
                metric_signals=metric_signals,
            ),
            approvals=review_projection["approvals"],
            reviewerRouting=reviewer_routing,
            activeChangeSetId=active_change_set_id,
            selectedTaskId=execution_summary.execution.taskId,
            execution=execution_summary.execution,
            projectMemory=memory,
            projectionWarnings=warnings + reviewer_warnings + execution_summary.warnings,
        )

    async def get_project_memory_overview(
        self,
        *,
        project_id: int,
        current_user_id: Optional[int] = None,
    ) -> dict[str, Any]:
        project = await self._load_project(project_id=project_id)
        items_payload = await self.get_project_memory_items(project_id=project_id, limit=500, offset=0)
        assets_payload = await self.get_project_memory_assets(project_id=project_id, limit=200, offset=0)
        context_payload = await self.get_project_memory_context_packs(project_id=project_id)
        writebacks_payload = await self.get_project_memory_writebacks(project_id=project_id, limit=8, offset=0)
        items = list(items_payload.get("items") or [])
        assets = list(assets_payload.get("assets") or [])
        context_packs = list(context_payload.get("contextPacks") or [])
        writebacks = list(writebacks_payload.get("writebacks") or [])
        metric_counts = self._memory_metric_counts(items=items, assets=assets, context_packs=context_packs)
        return {
            "projectId": str(project.id),
            "projectName": project.title or project.name,
            "generatedAt": now_cn_naive().isoformat(),
            "metrics": [
                {"key": "confirmed_facts", "label": "已确认事实", "value": metric_counts["confirmed_facts"]},
                {"key": "requirements", "label": "需求与范围", "value": metric_counts["requirements"]},
                {"key": "decisions", "label": "决策", "value": metric_counts["decisions"]},
                {"key": "constraints", "label": "约束条件", "value": metric_counts["constraints"]},
                {"key": "evidence", "label": "证据来源", "value": metric_counts["evidence"]},
                {"key": "context_packs", "label": "上下文包", "value": metric_counts["context_packs"]},
                {"key": "assets", "label": "项目资产", "value": metric_counts["assets"]},
            ],
            "facts": [item for item in items if item.get("category") == "fact"][:5],
            "requirements": [item for item in items if item.get("category") == "requirement"][:5],
            "decisions": [item for item in items if item.get("category") == "decision"][:5],
            "constraints": [item for item in items if item.get("category") in {"constraint", "risk"}][:5],
            "evidence": assets[:5],
            "contextPacks": context_packs[:5],
            "writebacks": writebacks[:5],
            "aiSuggestions": self._memory_ai_suggestions(
                items=items,
                assets=assets,
                context_packs=context_packs,
                writebacks=writebacks,
            ),
            "warnings": [],
        }

    async def get_project_memory_items(
        self,
        *,
        project_id: int,
        category: Optional[str] = None,
        stage: Optional[str] = None,
        source_type: Optional[str] = None,
        status: Optional[str] = None,
        query: Optional[str] = None,
        limit: int = 50,
        offset: int = 0,
    ) -> dict[str, Any]:
        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)
        semantic_facts = await self._load_project_memory_semantic_facts(project_id=project_id, truth_nodes=truth_nodes, tasks=tasks)
        evidence_by_item = await self._memory_evidence_by_item(project_id=project_id)
        items: list[dict[str, Any]] = []
        for node in truth_nodes:
            items.append(self._memory_item_from_truth_node(node, evidence_by_item=evidence_by_item))
        for task, user in tasks:
            items.append(self._memory_item_from_task(task, user=user))
        for fact in semantic_facts:
            items.append(self._memory_item_from_semantic_fact(fact))
        filtered = self._filter_memory_rows(items, category=category, stage=stage, source_type=source_type, status=status, query=query)
        filtered.sort(key=lambda item: item.get("updatedAt") or item.get("createdAt") or "", reverse=True)
        safe_limit = max(1, min(int(limit or 50), 200))
        safe_offset = max(0, int(offset or 0))
        return {
            "projectId": str(project.id),
            "total": len(filtered),
            "limit": safe_limit,
            "offset": safe_offset,
            "items": filtered[safe_offset : safe_offset + safe_limit],
        }

    async def get_project_memory_item(self, *, project_id: int, item_id: str) -> dict[str, Any]:
        items_payload = await self.get_project_memory_items(project_id=project_id, limit=500, offset=0)
        item_id_value = str(item_id or "").strip()
        for item in items_payload.get("items") or []:
            if str(item.get("id")) == item_id_value:
                return {"projectId": str(project_id), "item": item}
        raise ValueError("project_memory_item_not_found")

    async def get_project_memory_assets(
        self,
        *,
        project_id: int,
        asset_type: Optional[str] = None,
        source_type: Optional[str] = None,
        status: Optional[str] = None,
        query: Optional[str] = None,
        limit: int = 50,
        offset: int = 0,
    ) -> dict[str, Any]:
        project = await self._load_project(project_id=project_id)
        assets: list[dict[str, Any]] = []
        evidence_result = await self.db.execute(
            select(PMFlowEvidenceLink)
            .where(and_(PMFlowEvidenceLink.tenant_id == self.tenant_id, PMFlowEvidenceLink.project_id == int(project_id)))
            .order_by(PMFlowEvidenceLink.updated_at.desc(), PMFlowEvidenceLink.created_at.desc(), PMFlowEvidenceLink.id.desc())
            .limit(500)
        )
        for evidence in evidence_result.scalars().all():
            assets.append(self._memory_asset_from_evidence(evidence))
        message_result = await self.db.execute(
            select(PMFlowThreadMessage)
            .where(and_(PMFlowThreadMessage.tenant_id == self.tenant_id, PMFlowThreadMessage.project_id == int(project_id)))
            .order_by(PMFlowThreadMessage.updated_at.desc(), PMFlowThreadMessage.created_at.desc(), PMFlowThreadMessage.id.desc())
            .limit(300)
        )
        for message in message_result.scalars().all():
            assets.append(self._memory_asset_from_thread_message(message))
        filtered = [asset for asset in assets if not asset_type or str(asset.get("assetType")) == str(asset_type)]
        filtered = self._filter_memory_rows(filtered, category=None, stage=None, source_type=source_type, status=status, query=query)
        filtered.sort(key=lambda item: item.get("updatedAt") or item.get("createdAt") or "", reverse=True)
        safe_limit = max(1, min(int(limit or 50), 200))
        safe_offset = max(0, int(offset or 0))
        return {
            "projectId": str(project.id),
            "total": len(filtered),
            "limit": safe_limit,
            "offset": safe_offset,
            "assets": filtered[safe_offset : safe_offset + safe_limit],
        }

    async def get_project_memory_context_packs(self, *, project_id: int) -> dict[str, Any]:
        project = await self._load_project(project_id=project_id)
        items_payload = await self.get_project_memory_items(project_id=project_id, limit=500, offset=0)
        assets_payload = await self.get_project_memory_assets(project_id=project_id, limit=200, offset=0)
        items = list(items_payload.get("items") or [])
        assets = list(assets_payload.get("assets") or [])
        packs = [
            self._memory_context_pack(
                key="project-truth",
                title="项目可信真相包",
                summary="已确认事实、需求、决策与约束，供项目问答和执行主链优先引用。",
                usage="项目问答 / 需求拆解 / 执行前上下文",
                item_ids=[item["id"] for item in items if item.get("category") in {"fact", "requirement", "decision", "constraint", "risk"}],
                asset_ids=[asset["id"] for asset in assets[:20]],
            ),
            self._memory_context_pack(
                key="development",
                title="开发执行上下文包",
                summary="面向开发 Agent 的需求、决策、风险和任务上下文。",
                usage="开发任务交接 / Prompt Package / PR 前检查",
                item_ids=[item["id"] for item in items if item.get("category") in {"requirement", "decision", "constraint", "risk", "task"}],
                asset_ids=[asset["id"] for asset in assets if asset.get("assetType") in {"conversation", "evidence", "code"}][:20],
            ),
            self._memory_context_pack(
                key="testing",
                title="测试与验收上下文包",
                summary="面向测试计划、验收标准和风险检查的上下文。",
                usage="测试计划 / 验收用例 / 发布门禁",
                item_ids=[item["id"] for item in items if item.get("category") in {"requirement", "constraint", "risk", "task", "test"}],
                asset_ids=[asset["id"] for asset in assets if asset.get("assetType") in {"test", "evidence", "conversation"}][:20],
            ),
            self._memory_context_pack(
                key="release-growth",
                title="发布与增长上下文包",
                summary="面向发布判断、反馈回流和下一轮迭代的上下文。",
                usage="发布检查 / 增长复盘 / 迭代建议",
                item_ids=[item["id"] for item in items if item.get("category") in {"release", "metric", "decision", "risk"}],
                asset_ids=[asset["id"] for asset in assets if asset.get("assetType") in {"release", "metric", "feedback", "evidence"}][:20],
            ),
        ]
        return {"projectId": str(project.id), "total": len(packs), "contextPacks": packs}

    async def get_project_memory_writebacks(
        self,
        *,
        project_id: int,
        limit: int = 50,
        offset: int = 0,
    ) -> dict[str, Any]:
        project = await self._load_project(project_id=project_id)
        result = await self.db.execute(
            select(PMFlowChangeSet)
            .where(and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.project_id == int(project_id)))
            .order_by(PMFlowChangeSet.committed_at.desc(), PMFlowChangeSet.updated_at.desc(), PMFlowChangeSet.created_at.desc())
            .limit(300)
        )
        change_sets = list(result.scalars().all())
        change_set_ids = [int(change_set.id) for change_set in change_sets]
        committed_user_ids = [int(change_set.committed_by) for change_set in change_sets if change_set.committed_by]
        committed_users = await self._load_user_display_map(user_ids=committed_user_ids)
        by_change_set: dict[int, list[PMFlowChangeItem]] = {}
        if change_set_ids:
            items_result = await self.db.execute(
                select(PMFlowChangeItem)
                .where(and_(PMFlowChangeItem.tenant_id == self.tenant_id, PMFlowChangeItem.change_set_id.in_(change_set_ids)))
                .order_by(PMFlowChangeItem.updated_at.desc(), PMFlowChangeItem.created_at.desc(), PMFlowChangeItem.id.desc())
            )
            for item in items_result.scalars().all():
                by_change_set.setdefault(int(item.change_set_id), []).append(item)
        execution_result = await self.db.execute(
            select(ProgressEvent, Task, User)
            .join(Task, Task.id == ProgressEvent.task_id)
            .outerjoin(User, User.id == ProgressEvent.user_id)
            .where(
                and_(
                    ProgressEvent.tenant_id == self.tenant_id,
                    ProgressEvent.project_id == int(project_id),
                    ProgressEvent.event_type == "execution_memory_writeback",
                )
            )
            .order_by(ProgressEvent.occurred_at.desc(), ProgressEvent.id.desc())
            .limit(300)
        )
        rows = [
            self._memory_writeback_from_change_set(
                change_set,
                items=by_change_set.get(int(change_set.id), []),
                committed_by=committed_users.get(int(change_set.committed_by)) if change_set.committed_by else None,
            )
            for change_set in change_sets
        ]
        rows.extend(
            self._memory_writeback_from_execution_event(event=event, task=task, user=user)
            for event, task, user in execution_result.all()
        )
        rows.sort(key=lambda item: str(item.get("committedAt") or item.get("updatedAt") or item.get("createdAt") or ""), reverse=True)
        safe_limit = max(1, min(int(limit or 50), 200))
        safe_offset = max(0, int(offset or 0))
        return {
            "projectId": str(project.id),
            "total": len(rows),
            "limit": safe_limit,
            "offset": safe_offset,
            "writebacks": rows[safe_offset : safe_offset + safe_limit],
        }

    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)
        latest_release_validation = await self._load_latest_release_validation_run(project_id=project_id)
        quality_contract = self._execution_quality_contract_for_summary(
            project_id=project_id,
            task_id=task_id,
            active_tasks=active_tasks,
            latest_package=latest_package,
            repo_bindings=repos,
            events=events,
        )
        delivery_artifacts = self._delivery_artifacts_projection(events)
        artifact_completeness = self._artifact_completeness(
            delivery_artifacts=delivery_artifacts,
            events=events,
            quality_contract=quality_contract,
        )
        artifact_review = self._artifact_review_projection(
            delivery_artifacts=delivery_artifacts,
            artifact_completeness=artifact_completeness,
            events=events,
        )
        acceptance = self._build_execution_acceptance(
            events=events,
            quality_contract=quality_contract,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        release_gate = self._build_release_gate(
            events=events,
            deliveries=deliveries,
            acceptance=acceptance,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        if latest_release_validation is not None:
            release_gate["latestValidationRun"] = self._serialize_release_validation_run(latest_release_validation)
        release_signals = await self._build_release_signals(
            project_id=project_id,
            release_gate=release_gate,
            latest_release_validation=latest_release_validation,
        )

        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,
            ),
            qualityContract=quality_contract,
            acceptance=acceptance,
            releaseGate=release_gate,
            releaseSignals=release_signals,
            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,
            deliveryArtifacts=delivery_artifacts,
            artifactCompleteness=artifact_completeness,
            artifactReview=artifact_review,
            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))
            if str(payload.get("handoff_type") or "") == "artifact_rework":
                prompt_package = await self._create_execution_prompt_package(
                    project=project,
                    task=task,
                    repo=repo,
                    runtime_target=body.runtime_target,
                    note=body.note,
                )
                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 create_execution_artifact_rework_package(
        self,
        *,
        project_id: int,
        body: PMFlowArtifactReworkPackageRequest,
        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)
        current_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        artifact_review = dict(current_summary.artifactReview or {})
        items = [item for item in list(artifact_review.get("items") or []) if isinstance(item, dict)]
        available_ids = {str(item.get("id") or "") for item in items}
        selected_ids = [str(item).strip() for item in list(body.review_item_ids or []) if str(item).strip()]
        if selected_ids and not set(selected_ids).issubset(available_ids):
            raise ValueError("invalid_artifact_review_item_ids")
        selected_items = [item for item in items if not selected_ids or str(item.get("id") or "") in set(selected_ids)]
        if not selected_items:
            raise ValueError("artifact_review_empty")
        prompt_package = await self._create_artifact_rework_prompt_package(
            project=project,
            task=task,
            repo=repo,
            runtime_target=body.runtime_target,
            note=body.note,
            current_summary=current_summary,
            review_items=selected_items,
        )
        event_payload = self._handoff_payload(
            task=task,
            repo=repo,
            runtime_target=body.runtime_target,
            note=body.note,
            prompt_package_id=int(prompt_package.id),
        )
        event_payload.update(
            {
                "protocol_version": "artifact_rework_package_v1",
                "handoff_type": "artifact_rework",
                "review_item_ids": [str(item.get("id") or "") for item in selected_items],
                "review_items": selected_items[:20],
                "artifact_review_status": artifact_review.get("status"),
                "boundary": "PMFlow creates a rework handoff package only; execution, branch operations, deployment, and Project Memory writes remain outside this action.",
            }
        )
        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="artifact_rework_package_created",
            source="pmflow_os",
            derived_task_status="pushed",
            summary=self._artifact_rework_package_summary(task=task, items=selected_items, note=body.note),
            payload=event_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_acceptance_decision(
        self,
        *,
        project_id: int,
        body: PMFlowExecutionAcceptanceDecisionRequest,
        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)
        current_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        task_events = [
            event
            for event in list(current_summary.recentProgressEvents or [])
            if str((event.get("task") or {}).get("id") or current_summary.execution.taskId or "") == str(task.id)
        ]
        quality_contract = dict(current_summary.qualityContract or {})
        acceptance = self._build_execution_acceptance(
            events=task_events,
            quality_contract=quality_contract,
            artifact_completeness=current_summary.artifactCompleteness,
            artifact_review=current_summary.artifactReview,
        )
        if body.decision in {"accept", "approve_technical_review"}:
            if not self._acceptance_has_required_test_evidence(task_events):
                raise ValueError("acceptance_test_evidence_required")
            if body.decision == "accept" and bool((quality_contract.get("acceptanceGate") or {}).get("technicalReviewRequired")):
                if acceptance.get("status") != "technical_review_approved":
                    raise ValueError("technical_review_required")

        event_type = {
            "accept": "accepted",
            "request_changes": "changes_requested",
            "reject": "acceptance_rejected",
            "request_technical_review": "technical_review_requested",
            "approve_technical_review": "technical_review_approved",
        }[body.decision]
        workflow_state = {
            "accept": "accepted",
            "request_changes": "changes_requested",
            "reject": "failed",
            "request_technical_review": "pending_technical_review",
            "approve_technical_review": "technical_review_approved",
        }[body.decision]
        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=None,
            event_type=event_type,
            source="pmflow_os",
            derived_task_status="ready_for_acceptance" if body.decision in {"accept", "approve_technical_review"} else workflow_state,
            summary=self._execution_acceptance_summary(decision=body.decision, task=task, note=body.note),
            payload={
                "protocol_version": "execution_acceptance_decision_v1",
                "workflow_state": workflow_state,
                "decision": body.decision,
                "note": body.note,
                "evidence_event_ids": list(body.evidence_event_ids or []),
                "quality_contract_id": quality_contract.get("qualityContractId"),
                "boundary": "PMFlow records human acceptance/review decision only; it does not merge, deploy, or write Project Memory.",
            },
            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_memory_writeback(
        self,
        *,
        project_id: int,
        body: PMFlowExecutionMemoryWritebackRequest,
        current_user_id: Optional[int] = None,
    ) -> dict[str, Any]:
        await self._load_project(project_id=project_id)
        task = await self._load_project_task(project_id=project_id, task_id=body.task_id)
        current_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        task_events = [
            event
            for event in list(current_summary.recentProgressEvents or [])
            if str((event.get("task") or {}).get("id") or current_summary.execution.taskId or "") == str(task.id)
        ]
        if any(self._event_type(event) == "execution_memory_writeback" for event in task_events):
            raise ValueError("writeback_already_exists")
        quality_contract = dict(current_summary.qualityContract or {})
        acceptance = self._build_execution_acceptance(
            events=task_events,
            quality_contract=quality_contract,
            artifact_completeness=current_summary.artifactCompleteness,
            artifact_review=current_summary.artifactReview,
        )
        if acceptance.get("status") != "accepted" or not bool(acceptance.get("userAccepted")):
            raise ValueError("execution_acceptance_required")

        execution_evidence_events = [event for event in task_events if self._is_execution_result_evidence_event(event)]
        if not execution_evidence_events:
            raise ValueError("execution_evidence_required")
        selected_event_ids = {str(item).strip() for item in list(body.progress_event_ids or []) if str(item).strip()}
        available_event_ids = {str(event.get("id")) for event in execution_evidence_events if event.get("id") is not None}
        if selected_event_ids and not selected_event_ids.issubset(available_event_ids):
            raise ValueError("invalid_progress_event_ids")
        evidence_event_ids = sorted(selected_event_ids or available_event_ids)
        failed_events = self._unresolved_failed_evidence_events(
            execution_evidence_events,
            artifact_completeness=current_summary.artifactCompleteness,
            artifact_review=current_summary.artifactReview,
        )
        if body.result == "completed" and failed_events:
            raise ValueError("unresolved_blocker_exists")
        writeback_proof = self._execution_evidence_proof(
            events=execution_evidence_events,
            artifact_completeness=current_summary.artifactCompleteness,
            artifact_review=current_summary.artifactReview,
            evidence_event_ids=evidence_event_ids,
        )

        now = now_cn_naive()
        next_task_status = {
            "completed": "done",
            "blocked": "blocked",
            "changes_requested": "blocked",
        }.get(body.result, "done")
        task.status = next_task_status
        if next_task_status == "done":
            task.completed_at = now
        task.manual_overrides = self._execution_memory_writeback_task_payload(
            task=task,
            existing=task.manual_overrides,
            result=body.result,
            evidence_event_ids=evidence_event_ids,
            delivery_artifacts=dict(current_summary.deliveryArtifacts or {}),
            artifact_completeness=dict(current_summary.artifactCompleteness or {}),
            artifact_review=dict(current_summary.artifactReview or {}),
            release_gate=dict(current_summary.releaseGate or {}),
            writeback_proof=writeback_proof,
            note=body.note,
            current_user_id=current_user_id,
            occurred_at=now,
        )
        flag_modified(task, "manual_overrides")

        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=self._latest_repo_binding_id(execution_evidence_events),
            event_type="execution_memory_writeback",
            source="pmflow_os",
            derived_task_status=next_task_status,
            summary=self._execution_memory_writeback_summary(task=task, result=body.result, note=body.note),
            payload={
                "protocol_version": "execution_memory_writeback_v1",
                "workflow_state": "project_memory_written" if body.result == "completed" else body.result,
                "result": body.result,
                "note": body.note,
                "task_id": str(task.id),
                "task_status": next_task_status,
                "evidence_event_ids": evidence_event_ids,
                "acceptance": acceptance,
                "delivery_artifacts": current_summary.deliveryArtifacts,
                "artifact_completeness": current_summary.artifactCompleteness,
                "artifact_review": current_summary.artifactReview,
                "release_gate_status": (current_summary.releaseGate or {}).get("status"),
                "release_readiness_proof": (current_summary.releaseGate or {}).get("releaseReadinessProof") or {},
                "writeback_proof": writeback_proof,
                "quality_contract_id": quality_contract.get("qualityContractId"),
                "boundary": "PMFlow records accepted execution facts into Project Memory; it does not merge, deploy, publish, or execute commands.",
            },
            occurred_at=now,
        )
        self.db.add(event)
        await self.db.commit()
        execution_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        truth_state = await self.get_truth_state(project_id=project_id, current_user_id=current_user_id)
        return {
            "writeback": self._memory_writeback_from_execution_event(event=event, task=task, user=None),
            "executionSummary": execution_summary.model_dump(mode="json"),
            "projectTruthState": truth_state.model_dump(mode="json"),
        }

    async def create_release_validation_run(
        self,
        *,
        project_id: int,
        body: PMFlowReleaseValidationRunCreateRequest,
        current_user_id: Optional[int] = None,
    ) -> dict[str, Any]:
        await self._load_project(project_id=project_id)
        summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        release_gate = dict(summary.releaseGate or {})
        evidence = list(release_gate.get("evidence") or [])
        if str(release_gate.get("status") or "empty") == "empty" or not evidence:
            raise ValueError("release_validation_evidence_empty")

        status = self._release_validation_status(gate_status=str(release_gate.get("status") or ""), decision=body.decision)
        run = ReleaseValidationRun(
            tenant_id=self.tenant_id,
            project_id=int(project_id),
            status=status,
            score=self._release_validation_score(status),
            key_metrics=self._release_validation_metrics(release_gate),
            risk_items=self._release_validation_risks(release_gate, decision=body.decision),
            rollback_plan=self._release_validation_rollback_plan(release_gate),
            validation_summary={
                "source": "pmflow_os_release_gate",
                "projectId": str(project_id),
                "status": status,
                "releaseGateStatus": release_gate.get("status"),
                "humanDecision": body.decision,
                "note": body.note,
                "createdBy": None if current_user_id is None else str(current_user_id),
                "createdAt": now_cn_naive().isoformat(),
                "deploymentExecuted": False,
                "evidence": evidence,
                "blockingReasons": list(release_gate.get("blockingReasons") or []),
                "nextActions": list(release_gate.get("nextActions") or []),
                "checks": list(release_gate.get("checks") or []),
            },
        )
        self.db.add(run)
        await self.db.commit()
        await self.db.refresh(run)
        updated_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        return {
            "releaseValidationRun": self._serialize_release_validation_run(run),
            "executionSummary": updated_summary.model_dump(mode="json"),
            "warnings": [] if status != "blocked" else list(release_gate.get("blockingReasons") or []),
        }

    async def create_release_feedback_ingest(
        self,
        *,
        project_id: int,
        body: PMFlowReleaseFeedbackIngestRequest,
        current_user_id: Optional[int] = None,
    ) -> dict[str, Any]:
        await self._load_project(project_id=project_id)
        charter_id = self._optional_int(body.charter_id, error_code="charter_not_found")
        control = AIProjectControlService(self.db, tenant_id=self.tenant_id)
        try:
            payload = await control.ingest_feedback(
                project_id=int(project_id),
                actor_user_id=current_user_id,
                source_type=body.source_type,
                title=body.title,
                content=body.content,
                charter_id=charter_id,
                metrics=dict(body.metrics or {}),
                samples=list(body.samples or []),
            )
        except ValueError as exc:
            if str(exc) == "反馈内容不能为空":
                raise ValueError("feedback_content_empty") from exc
            raise
        updated_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        release_signals = dict(updated_summary.releaseSignals or {})
        raw_ingest_id = str((payload.get("ingest") or {}).get("id") or "")
        projected_ingest = next(
            (item for item in list(release_signals.get("feedback") or []) if str(item.get("id")) == raw_ingest_id),
            None,
        )
        projected_recommendation = next(
            (
                item
                for item in list(release_signals.get("recommendations") or [])
                if str(item.get("feedbackIngestId") or "") == raw_ingest_id
            ),
            None,
        )
        return {
            "ingest": projected_ingest or payload.get("ingest"),
            "recommendation": projected_recommendation or payload.get("recommendation"),
            "releaseSignals": release_signals,
            "executionSummary": updated_summary.model_dump(mode="json"),
            "warnings": list(release_signals.get("warnings") or []),
        }

    async def create_iteration_recommendation_change_set_draft(
        self,
        *,
        project_id: int,
        recommendation_id: int,
        body: PMFlowIterationRecommendationChangeSetDraftRequest,
        current_user_id: Optional[int] = None,
    ) -> dict[str, Any]:
        project = await self._load_project(project_id=project_id)
        recommendation = await self._load_iteration_recommendation(
            project_id=project_id,
            recommendation_id=recommendation_id,
        )
        summary_payload = dict(recommendation.recommendation_summary or {})
        existing_change_set_id = self._optional_int(
            summary_payload.get("changeSetId"),
            error_code="iteration_recommendation_not_found",
        )
        if existing_change_set_id is not None:
            existing = await self._load_recommendation_change_set(
                project_id=project_id,
                change_set_id=existing_change_set_id,
            )
            if existing is not None:
                detail = await PMFlowChangeReviewService(self.db, tenant_id=self.tenant_id).get_change_set_detail(
                    project_id=project_id,
                    change_set_id=int(existing.id),
                )
                updated_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
                return {
                    "changeSetDetail": detail,
                    "executionSummary": updated_summary.model_dump(mode="json"),
                    "warnings": ["iteration_recommendation_changeset_reused"],
                }

        feedback = await self._load_feedback_ingest(
            project_id=project_id,
            feedback_ingest_id=recommendation.feedback_ingest_id,
        )
        thread = await self._resolve_or_create_iteration_thread(project=project)
        message = PMFlowThreadMessage(
            tenant_id=self.tenant_id,
            thread_id=int(thread.id),
            project_id=int(project_id),
            role="assistant",
            content=self._iteration_recommendation_message(
                recommendation=recommendation,
                feedback=feedback,
                note=body.note,
            ),
            source="pmflow_os",
            attachments=[],
            message_metadata={
                "source": "iteration_recommendation",
                "recommendation_id": str(recommendation.id),
                "feedback_ingest_id": None if feedback is None else str(feedback.id),
                "created_by": None if current_user_id is None else int(current_user_id),
                "auto_commit": False,
                "note": body.note,
            },
            status="sent",
        )
        self.db.add(message)
        await self.db.flush()

        items = self._iteration_recommendation_change_items(
            recommendation=recommendation,
            feedback=feedback,
            project_id=project_id,
        )
        change_set = PMFlowChangeSet(
            tenant_id=self.tenant_id,
            project_id=int(project_id),
            thread_id=int(thread.id),
            source_message_id=int(message.id),
            status="open",
            summary=self._iteration_change_set_summary(recommendation=recommendation, feedback=feedback),
            headline=self._iteration_change_set_headline(recommendation=recommendation, feedback=feedback),
            extraction_scope="iteration_recommendation",
            routing_meta={
                "source": "release_feedback_iteration_loop",
                "recommendationId": str(recommendation.id),
                "feedbackIngestId": None if feedback is None else str(feedback.id),
                "releaseFeedbackLoop": True,
                "autoCommit": False,
                "note": body.note,
            },
            llm_call_id=None,
            created_by=int(current_user_id or project.owner_id),
        )
        self.db.add(change_set)
        await self.db.flush()

        for item in items:
            item.change_set_id = int(change_set.id)
            self.db.add(item)
        await self.db.flush()

        evidence = PMFlowEvidenceLink(
            tenant_id=self.tenant_id,
            project_id=int(project_id),
            change_set_id=int(change_set.id),
            change_item_id=None,
            source_type="iteration_recommendation",
            source_id=str(recommendation.id),
            title="发布后反馈迭代建议",
            excerpt=self._iteration_recommendation_excerpt(recommendation=recommendation, feedback=feedback),
            display_label="Release Feedback",
            evidence_metadata={
                "recommendation_id": str(recommendation.id),
                "feedback_ingest_id": None if feedback is None else str(feedback.id),
                "feedback_title": None if feedback is None else feedback.title,
                "priority": recommendation.priority,
                "change_set_created": True,
            },
        )
        self.db.add(evidence)

        summary_payload.update(
            {
                "changeSetId": str(change_set.id),
                "changeSetCreated": True,
                "changeSetCreatedAt": now_cn_naive().isoformat(),
                "changeSetCreatedBy": None if current_user_id is None else str(current_user_id),
            }
        )
        recommendation.recommendation_summary = summary_payload
        recommendation.status = "changeset_drafted"
        flag_modified(recommendation, "recommendation_summary")
        await self.db.commit()
        await self.db.refresh(change_set)

        detail = await PMFlowChangeReviewService(self.db, tenant_id=self.tenant_id).get_change_set_detail(
            project_id=project_id,
            change_set_id=int(change_set.id),
        )
        updated_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        return {
            "changeSetDetail": detail,
            "executionSummary": updated_summary.model_dump(mode="json"),
            "warnings": [],
        }

    async def create_iteration_recommendation_group_change_set_draft(
        self,
        *,
        project_id: int,
        group_key: str,
        body: PMFlowIterationRecommendationGroupChangeSetDraftRequest,
        current_user_id: Optional[int] = None,
    ) -> dict[str, Any]:
        project = await self._load_project(project_id=project_id)
        rows = await self._load_iteration_recommendations(project_id=project_id)
        feedback_by_id = await self._load_feedback_ingests_for_recommendations(project_id=project_id, recommendations=rows)
        quality_by_id = {
            int(row.id): self._recommendation_quality_profile(
                recommendation=row,
                feedback=feedback_by_id.get(int(row.feedback_ingest_id)) if row.feedback_ingest_id is not None else None,
            )
            for row in rows
        }
        requested_ids = self._parse_recommendation_ids(body.recommendation_ids)
        normalized_group_key = self._normalize_group_key(group_key)
        selected = [
            row
            for row in rows
            if (requested_ids and int(row.id) in requested_ids) or (not requested_ids and quality_by_id[int(row.id)]["groupKey"] == normalized_group_key)
        ]
        if requested_ids and len({int(row.id) for row in selected}) != len(requested_ids):
            raise ValueError("iteration_recommendation_not_found")
        if not selected:
            raise ValueError("iteration_recommendation_group_not_found")
        selected_quality = self._finalize_recommendation_quality(
            recommendations=selected,
            quality_by_id={int(row.id): quality_by_id[int(row.id)] for row in selected},
        )
        canonical_group_key = str(next(iter(selected_quality.values())).get("groupKey") or normalized_group_key)
        existing_change_set = await self._find_existing_recommendation_group_change_set(
            project_id=project_id,
            recommendations=selected,
        )
        if existing_change_set is not None:
            await self._mark_recommendations_for_group_change_set(
                recommendations=selected,
                change_set_id=int(existing_change_set.id),
                group_key=canonical_group_key,
                current_user_id=current_user_id,
                reused=True,
            )
            await self.db.commit()
            detail = await PMFlowChangeReviewService(self.db, tenant_id=self.tenant_id).get_change_set_detail(
                project_id=project_id,
                change_set_id=int(existing_change_set.id),
            )
            updated_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
            return {
                "changeSetDetail": detail,
                "executionSummary": updated_summary.model_dump(mode="json"),
                "warnings": ["iteration_recommendation_group_changeset_reused"],
            }

        thread = await self._resolve_or_create_iteration_thread(project=project)
        message = PMFlowThreadMessage(
            tenant_id=self.tenant_id,
            thread_id=int(thread.id),
            project_id=int(project_id),
            role="assistant",
            content=self._iteration_group_message(
                recommendations=selected,
                feedback_by_id=feedback_by_id,
                quality_by_id=selected_quality,
                note=body.note,
            ),
            source="pmflow_os",
            attachments=[],
            message_metadata={
                "source": "iteration_recommendation_group",
                "group_key": canonical_group_key,
                "recommendation_ids": [str(row.id) for row in selected],
                "created_by": None if current_user_id is None else int(current_user_id),
                "auto_commit": False,
                "note": body.note,
            },
            status="sent",
        )
        self.db.add(message)
        await self.db.flush()

        items = self._iteration_recommendation_group_change_items(
            recommendations=selected,
            feedback_by_id=feedback_by_id,
            quality_by_id=selected_quality,
            project_id=project_id,
        )
        change_set = PMFlowChangeSet(
            tenant_id=self.tenant_id,
            project_id=int(project_id),
            thread_id=int(thread.id),
            source_message_id=int(message.id),
            status="open",
            summary=self._iteration_group_change_set_summary(recommendations=selected, quality_by_id=selected_quality),
            headline=self._iteration_group_change_set_headline(recommendations=selected, quality_by_id=selected_quality),
            extraction_scope="iteration_recommendation_group",
            routing_meta={
                "source": "release_feedback_iteration_group",
                "groupKey": canonical_group_key,
                "recommendationIds": [str(row.id) for row in selected],
                "releaseFeedbackLoop": True,
                "autoCommit": False,
                "note": body.note,
            },
            llm_call_id=None,
            created_by=int(current_user_id or project.owner_id),
        )
        self.db.add(change_set)
        await self.db.flush()

        for item in items:
            item.change_set_id = int(change_set.id)
            self.db.add(item)
        await self.db.flush()

        evidence = PMFlowEvidenceLink(
            tenant_id=self.tenant_id,
            project_id=int(project_id),
            change_set_id=int(change_set.id),
            change_item_id=None,
            source_type="iteration_recommendation_group",
            source_id=canonical_group_key,
            title="发布后反馈迭代建议组",
            excerpt=self._iteration_group_excerpt(recommendations=selected, feedback_by_id=feedback_by_id),
            display_label="Release Feedback Group",
            evidence_metadata={
                "groupKey": canonical_group_key,
                "recommendationIds": [str(row.id) for row in selected],
                "feedbackIngestIds": [
                    str(row.feedback_ingest_id) for row in selected if row.feedback_ingest_id is not None
                ],
                "quality": next(iter(selected_quality.values()), {}),
                "change_set_created": True,
            },
        )
        self.db.add(evidence)

        await self._mark_recommendations_for_group_change_set(
            recommendations=selected,
            change_set_id=int(change_set.id),
            group_key=canonical_group_key,
            current_user_id=current_user_id,
            reused=False,
        )
        await self.db.commit()
        await self.db.refresh(change_set)

        detail = await PMFlowChangeReviewService(self.db, tenant_id=self.tenant_id).get_change_set_detail(
            project_id=project_id,
            change_set_id=int(change_set.id),
        )
        updated_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        return {
            "changeSetDetail": detail,
            "executionSummary": updated_summary.model_dump(mode="json"),
            "warnings": [],
        }

    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 create_execution_runtime_credential(
        self,
        *,
        project_id: int,
        prompt_package_id: int,
        body: PMFlowExecutionRuntimeCredentialCreateRequest,
        current_user_id: Optional[int] = None,
    ) -> PMFlowExecutionRuntimeCredentialResponse:
        await self._load_project(project_id=project_id)
        prompt_package = await self._load_execution_prompt_package(
            project_id=project_id,
            prompt_package_id=int(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))
        package_payload = self._payload(prompt_package.package_payload)
        latest_handoff = self._payload(package_payload.get("latest_handoff"))
        runtime_target = body.runtime_target or str(
            latest_handoff.get("runtime_target") or package_payload.get("runtime_target") or "local_bridge"
        )
        if runtime_target not in {"local_bridge", "codex", "github_agent", "manual"}:
            raise ValueError("runtime_target_invalid")
        repo_binding_id = str(latest_handoff.get("repo_binding_id") or package_payload.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)
        allowed_sources = self._runtime_allowed_sources(body.allowed_sources, runtime_target=runtime_target)
        expires_at = now_cn_naive() + timedelta(seconds=int(body.expires_in_seconds))
        key_id = f"erc_{secrets.token_urlsafe(18)}"[:80]
        secret = self._runtime_secret_for_key(key_id)
        credential = PMFlowExecutionRuntimeCredential(
            tenant_id=self.tenant_id,
            key_id=key_id,
            secret_hash=self._runtime_secret_hash(secret),
            project_id=int(project_id),
            task_id=int(task.id),
            prompt_package_id=int(prompt_package.id),
            repo_binding_id=int(repo.id) if repo is not None else None,
            runtime_target=runtime_target,
            allowed_sources=allowed_sources,
            status="active",
            expires_at=expires_at,
            created_by=int(current_user_id) if current_user_id is not None else None,
            use_count=0,
        )
        self.db.add(credential)
        await self.db.commit()
        return self._runtime_credential_response(
            credential=credential,
            secret=secret,
            task=task,
            repo=repo,
            note=body.note,
        )

    async def record_runtime_execution_result(
        self,
        *,
        body: PMFlowRuntimeExecutionResultRequest,
        key_id: str,
        timestamp: str,
        signature: str,
        delivery_id: str,
        raw_body: bytes,
    ) -> ExecutionSummaryProjection:
        credential = await self._load_runtime_credential(key_id=key_id)
        self.tenant_id = int(credential.tenant_id)
        self._validate_runtime_credential(
            credential=credential,
            body=body,
            timestamp=timestamp,
            signature=signature,
            raw_body=raw_body,
        )
        body.delivery_id = body.delivery_id or delivery_id
        if credential.repo_binding_id is not None:
            if body.repo_binding_id and str(body.repo_binding_id) != str(credential.repo_binding_id):
                raise ValueError("runtime_handoff_mismatch")
            body.repo_binding_id = str(credential.repo_binding_id)
        credential.last_used_at = now_cn_naive()
        credential.use_count = int(credential.use_count or 0) + 1
        summary = await self.record_execution_progress_event(
            project_id=int(credential.project_id),
            body=body,
            current_user_id=None,
        )
        return summary

    async def record_runtime_execution_package_pull(
        self,
        *,
        body: PMFlowRuntimeExecutionPackageRequest,
        key_id: str,
        timestamp: str,
        signature: str,
        delivery_id: str,
        raw_body: bytes,
    ) -> PMFlowRuntimeExecutionPackageResponse:
        credential = await self._load_runtime_credential(key_id=key_id)
        self.tenant_id = int(credential.tenant_id)
        self._validate_runtime_credential(
            credential=credential,
            body=body,
            timestamp=timestamp,
            signature=signature,
            raw_body=raw_body,
        )
        body.delivery_id = body.delivery_id or delivery_id
        project = await self._load_project(project_id=int(credential.project_id))
        prompt_package = await self._load_execution_prompt_package(
            project_id=int(credential.project_id),
            prompt_package_id=int(credential.prompt_package_id),
        )
        task = await self._load_project_task(project_id=int(credential.project_id), task_id=str(credential.task_id))
        repo = await self._load_project_repo_binding(
            project_id=int(credential.project_id),
            repo_binding_id=None if credential.repo_binding_id is None else str(credential.repo_binding_id),
        )
        truth_nodes = await self._load_truth_nodes(project_id=int(credential.project_id))
        memory = await self._project_memory(project=project, truth_nodes=truth_nodes[:50], task_ids=[int(task.id)])
        pull_event = await self._record_execution_package_pull_event(
            credential=credential,
            project=project,
            task=task,
            repo=repo,
            body=body,
        )
        credential.last_used_at = now_cn_naive()
        credential.use_count = int(credential.use_count or 0) + 1
        await self.db.commit()
        return self._runtime_execution_package_response(
            credential=credential,
            project=project,
            task=task,
            repo=repo,
            prompt_package=prompt_package,
            memory=memory,
            pull_event=pull_event,
        )

    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,
            delivery_id=body.delivery_id,
            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 record_execution_artifact_review_action(
        self,
        *,
        project_id: int,
        body: PMFlowArtifactReviewActionRequest,
        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)
        current_summary = await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id)
        artifact_review = dict(current_summary.artifactReview or {})
        items = [item for item in list(artifact_review.get("items") or []) if isinstance(item, dict)]
        available_ids = {str(item.get("id") or "") for item in items}
        selected_ids = [str(item).strip() for item in list(body.review_item_ids or []) if str(item).strip()]
        if selected_ids and not set(selected_ids).issubset(available_ids):
            raise ValueError("invalid_artifact_review_item_ids")
        selected_items = [item for item in items if not selected_ids or str(item.get("id") or "") in set(selected_ids)]
        event_type = {
            "request_fix": "artifact_review_requested",
            "record_fix": "artifact_fix_recorded",
            "mark_recheck_ready": "artifact_recheck_ready",
        }.get(body.action, "artifact_review_requested")
        payload = {
            "protocol_version": "artifact_review_action_v1",
            "workflow_state": event_type,
            "action": body.action,
            "task_id": str(task.id),
            "review_item_ids": selected_ids,
            "review_items": selected_items[:20],
            "artifact_review_status": artifact_review.get("status"),
            "artifact_completeness": current_summary.artifactCompleteness,
            "note": body.note,
            "boundary": "PMFlow records artifact review and rework intent only; command execution, branch operations, deployment, and Project Memory writes stay outside this action.",
        }
        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=self._latest_repo_binding_id(list(current_summary.recentProgressEvents or [])),
            event_type=event_type,
            source="pmflow_os",
            derived_task_status="blocked" if body.action == "request_fix" else "testing",
            summary=self._artifact_review_action_summary(action=body.action, items=selected_items, note=body.note),
            payload=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 _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_members(self, *, project: Project) -> list[User]:
        result = await self.db.execute(
            select(User)
            .outerjoin(UserProject, and_(UserProject.user_id == User.id, UserProject.project_id == int(project.id)))
            .where(
                and_(
                    User.tenant_id == self.tenant_id,
                    or_(UserProject.project_id == int(project.id), User.id == project.owner_id),
                    User.is_active.is_(True),
                )
            )
            .order_by(User.id.asc())
        )
        members: dict[int, User] = {}
        for user in result.scalars().all():
            members[int(user.id)] = user
        return list(members.values())

    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_latest_release_validation_run(self, *, project_id: int) -> Optional[ReleaseValidationRun]:
        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(), ReleaseValidationRun.id.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def _load_iteration_recommendation(self, *, project_id: int, recommendation_id: int) -> IterationRecommendation:
        result = await self.db.execute(
            select(IterationRecommendation).where(
                and_(
                    IterationRecommendation.tenant_id == self.tenant_id,
                    IterationRecommendation.project_id == int(project_id),
                    IterationRecommendation.id == int(recommendation_id),
                )
            )
        )
        row = result.scalar_one_or_none()
        if row is None:
            raise ValueError("iteration_recommendation_not_found")
        return row

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

    async def _load_recommendation_change_set(self, *, project_id: int, change_set_id: int) -> Optional[PMFlowChangeSet]:
        result = await self.db.execute(
            select(PMFlowChangeSet).where(
                and_(
                    PMFlowChangeSet.tenant_id == self.tenant_id,
                    PMFlowChangeSet.project_id == int(project_id),
                    PMFlowChangeSet.id == int(change_set_id),
                )
            )
        )
        return result.scalar_one_or_none()

    async def _find_existing_recommendation_group_change_set(
        self,
        *,
        project_id: int,
        recommendations: list[IterationRecommendation],
    ) -> Optional[PMFlowChangeSet]:
        candidate_ids: list[int] = []
        for recommendation in recommendations:
            summary = dict(recommendation.recommendation_summary or {})
            change_set_id = self._optional_int(summary.get("changeSetId"), error_code="iteration_recommendation_not_found")
            if change_set_id is not None:
                candidate_ids.append(change_set_id)
        for change_set_id in candidate_ids:
            row = await self._load_recommendation_change_set(project_id=project_id, change_set_id=change_set_id)
            if row is not None:
                return row
        return None

    async def _load_feedback_ingest(self, *, project_id: int, feedback_ingest_id: Optional[int]) -> Optional[FeedbackIngest]:
        if feedback_ingest_id is None:
            return None
        result = await self.db.execute(
            select(FeedbackIngest).where(
                and_(
                    FeedbackIngest.tenant_id == self.tenant_id,
                    FeedbackIngest.project_id == int(project_id),
                    FeedbackIngest.id == int(feedback_ingest_id),
                )
            )
        )
        return result.scalar_one_or_none()

    async def _load_feedback_ingests_for_recommendations(
        self,
        *,
        project_id: int,
        recommendations: list[IterationRecommendation],
    ) -> dict[int, FeedbackIngest]:
        ids = sorted({int(row.feedback_ingest_id) for row in recommendations if row.feedback_ingest_id is not None})
        if not ids:
            return {}
        result = await self.db.execute(
            select(FeedbackIngest).where(
                and_(
                    FeedbackIngest.tenant_id == self.tenant_id,
                    FeedbackIngest.project_id == int(project_id),
                    FeedbackIngest.id.in_(ids),
                )
            )
        )
        return {int(row.id): row for row in result.scalars().all()}

    async def _resolve_or_create_iteration_thread(self, *, project: Project) -> PMFlowThreadModel:
        result = await self.db.execute(
            select(PMFlowThreadModel)
            .where(and_(PMFlowThreadModel.tenant_id == self.tenant_id, PMFlowThreadModel.project_id == int(project.id)))
            .order_by(PMFlowThreadModel.created_at.asc(), PMFlowThreadModel.id.asc())
            .limit(1)
        )
        row = result.scalar_one_or_none()
        if row is not None:
            return row
        row = PMFlowThreadModel(
            tenant_id=self.tenant_id,
            project_id=int(project.id),
            title=f"{project.title or project.name} 项目线程",
            objective=self._project_goal(project=project, truth_nodes=[]),
            status="active",
            source="iteration_loop",
        )
        self.db.add(row)
        await self.db.flush()
        return row

    async def _build_release_signals(
        self,
        *,
        project_id: int,
        release_gate: dict[str, Any],
        latest_release_validation: Optional[ReleaseValidationRun],
    ) -> dict[str, Any]:
        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(), FeedbackIngest.id.desc())
            .limit(10)
        )
        feedback_rows = list(feedback_result.scalars().all())
        feedback_items = [self._serialize_feedback_ingest(item) for item in feedback_rows]

        recommendation_result = await self.db.execute(
            select(IterationRecommendation)
            .where(and_(IterationRecommendation.tenant_id == self.tenant_id, IterationRecommendation.project_id == int(project_id)))
            .order_by(IterationRecommendation.updated_at.desc(), IterationRecommendation.created_at.desc(), IterationRecommendation.id.desc())
            .limit(10)
        )
        recommendation_rows = list(recommendation_result.scalars().all())
        feedback_by_id_for_recommendations = await self._load_feedback_ingests_for_recommendations(
            project_id=project_id,
            recommendations=recommendation_rows,
        )
        base_quality_by_id = {
            int(item.id): self._recommendation_quality_profile(
                recommendation=item,
                feedback=feedback_by_id_for_recommendations.get(int(item.feedback_ingest_id)) if item.feedback_ingest_id is not None else None,
            )
            for item in recommendation_rows
        }
        quality_by_id = self._finalize_recommendation_quality(
            recommendations=recommendation_rows,
            quality_by_id=base_quality_by_id,
        )
        recommendations = [
            self._serialize_iteration_recommendation(item, quality=quality_by_id.get(int(item.id)))
            for item in recommendation_rows
        ]
        recommendation_groups = self._serialize_recommendation_groups(recommendations)

        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(), RevenueEvent.id.desc())
            .limit(10)
        )
        revenue_rows = list(revenue_result.scalars().all())
        revenue_events = [self._serialize_revenue_event(item) for item in revenue_rows]

        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(), ROISnapshot.id.desc())
            .limit(10)
        )
        roi_rows = list(roi_result.scalars().all())
        roi_snapshots = [self._serialize_roi_snapshot(item) for item in roi_rows]

        compliance_result = await self.db.execute(
            select(ComplianceCheckRun)
            .where(and_(ComplianceCheckRun.tenant_id == self.tenant_id, ComplianceCheckRun.project_id == int(project_id)))
            .order_by(ComplianceCheckRun.updated_at.desc(), ComplianceCheckRun.created_at.desc(), ComplianceCheckRun.id.desc())
            .limit(10)
        )
        compliance_rows = list(compliance_result.scalars().all())
        compliance_checks = [self._serialize_compliance_check(item) for item in compliance_rows]

        warnings: list[str] = []
        if not feedback_items:
            warnings.append("release_feedback_empty: no post-release feedback has been ingested.")
        if not revenue_events:
            warnings.append("release_revenue_empty: no revenue event has been recorded.")
        if not roi_snapshots:
            warnings.append("release_roi_empty: no ROI snapshot has been recorded.")
        if not compliance_checks:
            warnings.append("release_compliance_empty: no compliance check has been recorded.")
        if not recommendations:
            warnings.append("iteration_recommendations_empty: no iteration recommendation has been generated.")
        if any(item.get("riskLevel") == "high" for item in roi_snapshots):
            warnings.append("release_roi_risk: latest ROI snapshot indicates negative or missing return.")
        if any(item.get("riskLevel") == "high" for item in compliance_checks):
            warnings.append("release_compliance_risk: compliance check is blocked or has open issues.")

        serialized_times = [
            str(item.get("updatedAt") or item.get("createdAt"))
            for item in [*feedback_items, *recommendations, *revenue_events, *roi_snapshots, *compliance_checks]
            if item.get("updatedAt") or item.get("createdAt")
        ]
        return {
            "status": "active" if feedback_items or revenue_events or roi_snapshots or compliance_checks or recommendations else "empty",
            "relationship": {
                "releaseGateStatus": release_gate.get("status") or "empty",
                "latestValidationRunId": None if latest_release_validation is None else str(latest_release_validation.id),
                "latestValidationRunStatus": None if latest_release_validation is None else latest_release_validation.status,
                "source": "pmflow_os_release_growth",
                "changeSetCreated": any(bool(item.get("changeSetCreated")) for item in recommendations),
                "businessSignalTypes": self._unique_strings(
                    [
                        *("feedback" for _ in feedback_items),
                        *("revenue" for _ in revenue_events),
                        *("roi" for _ in roi_snapshots),
                        *("compliance" for _ in compliance_checks),
                    ]
                ),
            },
            "summary": {
                "feedbackCount": len(feedback_items),
                "revenueEventCount": len(revenue_events),
                "roiSnapshotCount": len(roi_snapshots),
                "complianceCheckCount": len(compliance_checks),
                "iterationRecommendationCount": len(recommendations),
                "riskSignalCount": sum(
                    1
                    for item in [*roi_snapshots, *compliance_checks]
                    if str(item.get("riskLevel") or "low") in {"medium", "high"}
                ),
                "latestSignalAt": max(serialized_times) if serialized_times else None,
            },
            "feedback": feedback_items,
            "revenueEvents": revenue_events,
            "roiSnapshots": roi_snapshots,
            "complianceChecks": compliance_checks,
            "recommendations": recommendations,
            "recommendationGroups": recommendation_groups,
            "nextActions": self._release_signal_next_actions(
                feedback_items=feedback_items,
                revenue_events=revenue_events,
                roi_snapshots=roi_snapshots,
                compliance_checks=compliance_checks,
                recommendations=recommendations,
                release_gate=release_gate,
            ),
            "warnings": warnings,
        }

    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 _load_runtime_credential(self, *, key_id: str) -> PMFlowExecutionRuntimeCredential:
        normalized = str(key_id or "").strip()
        if not normalized:
            raise ValueError("runtime_credential_missing")
        result = await self.db.execute(
            select(PMFlowExecutionRuntimeCredential).where(PMFlowExecutionRuntimeCredential.key_id == normalized)
        )
        credential = result.scalar_one_or_none()
        if credential is None:
            raise ValueError("runtime_credential_missing")
        return credential

    def _runtime_allowed_sources(self, values: list[str], *, runtime_target: str) -> list[str]:
        allowed: list[str] = []
        for value in list(values or []):
            normalized = str(value or "").strip()
            if normalized in {"local_bridge", "codex", "github_agent", "manual"} and normalized not in allowed:
                allowed.append(normalized)
        if not allowed:
            allowed.append(runtime_target if runtime_target in {"local_bridge", "codex", "github_agent", "manual"} else "manual")
        return allowed

    def _runtime_secret_for_key(self, key_id: str) -> str:
        material = f"pmflow-runtime-execution:{key_id}".encode("utf-8")
        return hmac.new(str(settings.secret_key).encode("utf-8"), material, hashlib.sha256).hexdigest()

    @staticmethod
    def _runtime_secret_hash(secret: str) -> str:
        return hashlib.sha256(str(secret).encode("utf-8")).hexdigest()

    def _validate_runtime_credential(
        self,
        *,
        credential: PMFlowExecutionRuntimeCredential,
        body: Any,
        timestamp: str,
        signature: str,
        raw_body: bytes,
    ) -> None:
        if str(credential.status or "") != "active":
            raise ValueError("runtime_credential_revoked")
        if credential.expires_at is None or credential.expires_at < now_cn_naive():
            raise ValueError("runtime_credential_expired")
        secret = self._runtime_secret_for_key(str(credential.key_id))
        if not hmac.compare_digest(self._runtime_secret_hash(secret), str(credential.secret_hash or "")):
            raise ValueError("runtime_credential_invalid")
        if not self._runtime_timestamp_is_fresh(timestamp):
            raise ValueError("runtime_timestamp_invalid")
        expected = hmac.new(
            secret.encode("utf-8"),
            f"{timestamp}.".encode("utf-8") + raw_body,
            hashlib.sha256,
        ).hexdigest()
        normalized_signature = str(signature or "").strip()
        if normalized_signature.startswith("sha256="):
            normalized_signature = normalized_signature[len("sha256=") :]
        if not hmac.compare_digest(expected, normalized_signature):
            raise ValueError("runtime_signature_invalid")
        if str(body.prompt_package_id) != str(credential.prompt_package_id) or str(body.task_id) != str(credential.task_id):
            raise ValueError("runtime_handoff_mismatch")
        if str(body.source or "") not in set(credential.allowed_sources or []):
            raise ValueError("runtime_source_not_allowed")

    @staticmethod
    def _runtime_timestamp_is_fresh(value: str) -> bool:
        raw = str(value or "").strip()
        if not raw:
            return False
        try:
            occurred = datetime.fromtimestamp(float(raw))
        except ValueError:
            try:
                occurred = datetime.fromisoformat(raw.replace("Z", "+00:00")).replace(tzinfo=None)
            except ValueError:
                return False
        return abs((now_cn_naive() - occurred).total_seconds()) <= 300

    def _runtime_credential_response(
        self,
        *,
        credential: PMFlowExecutionRuntimeCredential,
        secret: str,
        task: Task,
        repo: Optional[RepoBinding],
        note: Optional[str],
    ) -> PMFlowExecutionRuntimeCredentialResponse:
        timestamp = str(int(now_cn_naive().timestamp()))
        example_package_payload: dict[str, Any] = {
            "prompt_package_id": str(credential.prompt_package_id),
            "task_id": str(credential.task_id),
            "source": (credential.allowed_sources or [credential.runtime_target])[0],
            "delivery_id": "runtime-package-pull-unique-id",
            "event_source_id": f"{credential.runtime_target}:run-id:package-pulled",
        }
        example_payload: dict[str, Any] = {
            "prompt_package_id": str(credential.prompt_package_id),
            "task_id": str(credential.task_id),
            "repo_binding_id": None if credential.repo_binding_id is None else str(credential.repo_binding_id),
            "event_type": "test_passed",
            "source": (credential.allowed_sources or [credential.runtime_target])[0],
            "branch_name": "codex/example-runtime-bridge",
            "commit_sha": "replace-with-real-commit",
            "test_summary": "unit / contract / smoke passed",
            "execution_status": "testing",
            "delivery_id": "runtime-delivery-unique-id",
            "event_source_id": f"{credential.runtime_target}:run-id:event-id",
            "delivery_artifacts": {
                "changed_files": [
                    {
                        "path": "frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx",
                        "change_type": "modified",
                        "summary": "展示交付证据包",
                        "risk_level": "medium",
                    }
                ],
                "commands_run": [{"command": "cd frontend && npm run build", "status": "passed", "summary": "构建通过"}],
                "test_outputs": [
                    {
                        "name": "pmflow-os focused tests",
                        "status": "passed",
                        "command": "cd frontend && npm run test -- src/features/pmflow-os",
                        "summary": "相关测试通过",
                    }
                ],
                "screenshots": [],
                "logs": [{"level": "info", "summary": "运行端完成交付并回写结构化证据。"}],
                "risk_notes": [],
                "review_checklist": [{"item": "测试证据可追溯", "status": "passed"}],
            },
        }
        if note:
            example_payload["summary"] = note
        if repo is not None:
            example_payload["repository_full_name"] = f"{repo.owner}/{repo.repo_name}"
        return PMFlowExecutionRuntimeCredentialResponse(
            keyId=str(credential.key_id),
            secret=secret,
            secretPreview=f"{secret[:8]}...{secret[-6:]}",
            projectId=str(credential.project_id),
            taskId=str(task.id),
            promptPackageId=str(credential.prompt_package_id),
            repoBindingId=None if credential.repo_binding_id is None else str(credential.repo_binding_id),
            runtimeTarget=credential.runtime_target,  # type: ignore[arg-type]
            allowedSources=list(credential.allowed_sources or []),
            expiresAt=credential.expires_at.isoformat(),
            status=str(credential.status),
            exampleHeaders={
                "X-PMFlow-Key-Id": credential.key_id,
                "X-PMFlow-Timestamp": timestamp,
                "X-PMFlow-Signature": "sha256=<hmac_sha256(secret, timestamp + '.' + raw_body)>",
                "X-PMFlow-Delivery-Id": "runtime-delivery-unique-id",
            },
            examplePackagePayload=example_package_payload,
            examplePayload=example_payload,
        )

    async def _record_execution_package_pull_event(
        self,
        *,
        credential: PMFlowExecutionRuntimeCredential,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        body: PMFlowRuntimeExecutionPackageRequest,
    ) -> dict[str, Any]:
        payload = {
            "protocol_version": "runtime_execution_package_pull_v1",
            "workflow_state": "package_pulled",
            "prompt_package_id": str(credential.prompt_package_id),
            "task_id": str(task.id),
            "source": body.source,
            "runtime_target": credential.runtime_target,
            "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}",
            "delivery_id": body.delivery_id,
            "event_source_id": body.event_source_id,
            "boundary": "PMFlow returns a machine-readable execution package only; command execution stays outside backend runtime.",
        }
        control = AIProjectControlService(self.db, tenant_id=self.tenant_id)
        return await control.create_or_apply_progress_event(
            task_id=int(task.id),
            project_id=int(project.id),
            event_type="execution_package_pulled",
            source=body.source,
            payload=payload,
            summary=f"运行端已领取执行包：{task.title} / {body.source}",
            repo_binding_id=int(repo.id) if repo is not None else None,
            delivery_id=body.delivery_id,
            event_source_id=body.event_source_id,
        )

    def _runtime_execution_package_response(
        self,
        *,
        credential: PMFlowExecutionRuntimeCredential,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        prompt_package: PromptPackage,
        memory: PMFlowProjectMemory,
        pull_event: dict[str, Any],
    ) -> PMFlowRuntimeExecutionPackageResponse:
        package_payload = self._payload(prompt_package.package_payload)
        task_payload = self._payload(task.ai_breakdown_payload)
        return PMFlowRuntimeExecutionPackageResponse(
            package={
                "project": {
                    "id": str(project.id),
                    "name": project.name,
                    "title": project.title,
                    "description": project.description,
                },
                "task": {
                    "id": str(task.id),
                    "title": task.title,
                    "description": task.description,
                    "status": task.status,
                    "priority": task.priority,
                    "truthNodeId": None if task.truth_node_id is None else str(task.truth_node_id),
                    "acceptanceCriteria": list(task_payload.get("acceptance_criteria") or []),
                    "breakdownPayload": task_payload,
                },
                "repo": None
                if repo is None
                else {
                    "id": str(repo.id),
                    "provider": repo.provider,
                    "owner": repo.owner,
                    "repoName": repo.repo_name,
                    "defaultBranch": repo.default_branch,
                    "status": repo.status,
                },
                "promptPackage": self._serialize_prompt_package(prompt_package),
                "reworkContract": package_payload.get("rework_contract") or {},
                "artifactReview": package_payload.get("artifact_review") or {},
                "projectMemoryContext": memory.model_dump(mode="json"),
                "qualityContract": package_payload.get("quality_contract") or package_payload.get("qualityContract") or {},
                "constraints": [
                    "Use Project Memory and task acceptance criteria as the execution source of truth.",
                    "Return factual branch / commit / PR / test / blocker metadata and structured delivery_artifacts through the signed callback.",
                    *(
                        ["Resolve every item in reworkContract.reviewItems and return updated delivery_artifacts."]
                        if package_payload.get("rework_contract")
                        else []
                    ),
                    "Do not treat package pull, test pass, or technical review as Project Memory writeback.",
                ],
                "forbiddenActions": [
                    "Do not execute commands inside PMFlow backend.",
                    "Do not write host configuration such as ~/.codex/config.toml.",
                    "Do not merge, deploy, publish, or commit Project Memory through this package pull endpoint.",
                ],
            },
            callback={
                "ingestEndpoint": "/api/pmflow-os/runtime/execution-results",
                "headers": {
                    "X-PMFlow-Key-Id": credential.key_id,
                    "X-PMFlow-Timestamp": "<unix_seconds_or_iso8601>",
                    "X-PMFlow-Signature": "sha256=<hmac_sha256(secret, timestamp + '.' + raw_body)>",
                    "X-PMFlow-Delivery-Id": "<stable_delivery_id>",
                },
                "resultProtocol": "execution_result_protocol_v1",
                "requiredEvidence": ["branch_or_commit", "test_summary_or_blocker", "execution_status", "delivery_artifacts"],
                "deliveryArtifactSchema": {
                    "changed_files": ["path", "change_type", "summary", "risk_level"],
                    "commands_run": ["command", "status", "summary", "duration_ms"],
                    "test_outputs": ["name", "status", "command", "summary", "output_ref"],
                    "screenshots": ["label", "artifact_ref", "artifact_url", "summary"],
                    "logs": ["level", "summary", "artifact_ref"],
                    "risk_notes": ["risk", "mitigation", "severity"],
                    "review_checklist": ["item", "status", "summary"],
                },
                "idempotency": ["delivery_id", "event_source_id"],
                "allowedSources": list(credential.allowed_sources or []),
                "boundary": "Only execution result metadata is accepted; PMFlow backend does not run commands, push code, merge, deploy, release, or write Project Memory.",
            },
            audit={
                "credentialId": str(credential.key_id),
                "packagePulledEventId": str(pull_event.get("id") or ""),
                "idempotentReplay": bool(pull_event.get("idempotent_replay")),
                "expiresAt": credential.expires_at.isoformat() if credential.expires_at else None,
            },
        )

    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 _create_artifact_rework_prompt_package(
        self,
        *,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        runtime_target: str,
        note: Optional[str],
        current_summary: ExecutionSummaryProjection,
        review_items: list[dict[str, Any]],
    ) -> 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,
        )
        rework_contract = self._artifact_rework_contract(
            current_summary=current_summary,
            review_items=review_items,
            note=note,
        )
        payload.update(
            {
                "handoff_type": "artifact_rework",
                "rework_contract": rework_contract,
                "artifact_review": current_summary.artifactReview,
                "artifact_completeness": current_summary.artifactCompleteness,
                "delivery_artifacts": current_summary.deliveryArtifacts,
            }
        )
        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_artifact_rework_package",
                "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,
                "review_item_ids": [str(item.get("id") or "") for item in review_items],
                "generated_at": now_cn_naive().isoformat(),
            },
        )
        self.db.add(package)
        await self.db.flush()
        latest_handoff = self._handoff_payload(
            task=task,
            repo=repo,
            runtime_target=runtime_target,
            note=note,
            prompt_package_id=int(package.id),
        )
        latest_handoff.update(
            {
                "handoff_type": "artifact_rework",
                "rework_contract": rework_contract,
                "review_item_ids": [str(item.get("id") or "") for item in review_items],
            }
        )
        package.package_payload = {**payload, "latest_handoff": latest_handoff}
        return package

    def _artifact_rework_contract(
        self,
        *,
        current_summary: ExecutionSummaryProjection,
        review_items: list[dict[str, Any]],
        note: Optional[str],
    ) -> dict[str, Any]:
        expected_evidence: list[str] = []
        commands: list[str] = []
        for item in review_items:
            expected_evidence.extend(str(value) for value in list(item.get("expectedEvidence") or []) if str(value))
            suggested = str(item.get("suggestedEventType") or "").strip()
            if suggested:
                commands.append(f"Return event_type={suggested} when evidence is ready")
        return {
            "protocol_version": "artifact_rework_contract_v1",
            "status": "ready",
            "reviewItems": review_items[:30],
            "expectedEvidence": self._unique_strings(expected_evidence),
            "suggestedReturnEvents": self._unique_strings(commands),
            "artifactCompleteness": current_summary.artifactCompleteness,
            "previousDeliveryArtifacts": current_summary.deliveryArtifacts,
            "operatorNote": note,
            "requiredCallback": {
                "endpoint": "/api/pmflow-os/runtime/execution-results",
                "resultProtocol": "execution_result_protocol_v1",
                "mustReturn": ["delivery_artifacts", "test_summary_or_blocker", "execution_status"],
            },
            "boundary": "This rework package instructs an external runtime what evidence to fix or add. PMFlow backend does not execute commands, create branches, merge, deploy, publish, or write Project Memory.",
        }

    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_ai_feedback(self, *, project_id: int, current_user_id: Optional[int]) -> list[Any]:
        if current_user_id is None:
            return []
        result = await self.db.execute(
            select(PMFlowAIFeedbackEvent)
            .where(
                and_(
                    PMFlowAIFeedbackEvent.tenant_id == self.tenant_id,
                    PMFlowAIFeedbackEvent.project_id == int(project_id),
                    PMFlowAIFeedbackEvent.actor_user_id == int(current_user_id),
                )
            )
            .order_by(PMFlowAIFeedbackEvent.updated_at.desc(), PMFlowAIFeedbackEvent.id.desc())
        )
        return [PMFlowAIFeedbackService.serialize(row) for row in result.scalars().all()]

    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())
        committed_user_ids = [int(row.committed_by) for row in change_sets if row.committed_by]
        committed_users = await self._load_user_display_map(user_ids=committed_user_ids)
        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": [
                self._serialize_change_set_projection(
                    row,
                    items=items_by_set.get(int(row.id), []),
                    committed_by=committed_users.get(int(row.committed_by)) if row.committed_by else None,
                )
                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],
        }

    async def _load_user_display_map(self, *, user_ids: list[int]) -> dict[int, str]:
        normalized_ids = sorted({int(user_id) for user_id in user_ids if user_id})
        if not normalized_ids:
            return {}
        result = await self.db.execute(
            select(User).where(and_(User.tenant_id == self.tenant_id, User.id.in_(normalized_ids)))
        )
        return {int(user.id): self._user_display_name(user) for user in result.scalars().all()}

    def _serialize_change_set_projection(
        self,
        row: PMFlowChangeSet,
        *,
        items: list[PMFlowChangeItem],
        committed_by: Optional[str] = None,
    ) -> dict[str, Any]:
        payload = PMFlowChangeReviewService.serialize_change_set(row, items)
        payload["committedBy"] = committed_by
        payload["committedById"] = str(row.committed_by) if row.committed_by else None
        return payload

    @staticmethod
    def _latest_writeback_meta(change_sets: list[dict[str, Any]]) -> dict[str, Optional[str]]:
        committed = [
            row
            for row in change_sets
            if row.get("committedAt") and (row.get("committedBy") or row.get("committedById"))
        ]
        if not committed:
            return {"at": None, "by": None, "by_id": None}
        latest = max(committed, key=lambda row: str(row.get("committedAt") or ""))
        return {
            "at": str(latest.get("committedAt") or "") or None,
            "by": str(latest.get("committedBy") or "") or None,
            "by_id": str(latest.get("committedById") or "") or None,
        }

    @staticmethod
    def _user_display_name(user: User) -> str:
        return str(user.name or user.username or user.email or user.id)

    def _project_reviewer_routing(
        self,
        *,
        project_id: int,
        change_sets: list[dict[str, Any]],
        change_items: list[dict[str, Any]],
        evidence: list[dict[str, Any]],
        members: list[User],
    ) -> tuple[dict[str, Any], list[str]]:
        all_items_by_set: dict[str, list[dict[str, Any]]] = {}
        items_by_set: dict[str, list[dict[str, Any]]] = {}
        for item in change_items:
            all_items_by_set.setdefault(str(item.get("changeSetId")), []).append(item)
            if item.get("status") != "pending_review":
                continue
            items_by_set.setdefault(str(item.get("changeSetId")), []).append(item)

        evidence_by_set: dict[str, list[dict[str, Any]]] = {}
        for row in evidence:
            evidence_by_set.setdefault(str(row.get("changeSetId") or ""), []).append(row)

        routing_items: list[dict[str, Any]] = []
        warnings: list[str] = []
        for change_set in change_sets:
            change_set_id = str(change_set.get("id") or "")
            review_meta = self._change_set_review_decision_meta(change_set)
            review_action = self._review_decision_action(review_meta)
            approval_state = self._approval_state(review_meta)
            pending_items = items_by_set.get(change_set_id, [])
            all_items = all_items_by_set.get(change_set_id, [])
            decision_visible = bool(review_action and approval_state != "not_requested")
            if not change_set_id or (not pending_items and not decision_visible):
                continue
            change_set_evidence = evidence_by_set.get(change_set_id, [])
            review_items = pending_items or all_items
            risk_levels = [str(item.get("riskLevel") or "medium") for item in review_items]
            object_types = sorted({str(item.get("objectType") or "Unknown") for item in review_items})
            risk_groups = sorted({str(item.get("group") or "unknown") for item in review_items})
            high_risk_signal = any(str(item.get("objectType") or "").lower() == "risk" or str(item.get("group") or "").lower() == "risk" for item in review_items)
            highest_risk = "high" if high_risk_signal else self._highest_risk_level(risk_levels)
            cross_domain = len(object_types) > 1 or len(risk_groups) > 1
            review_reasons = self._review_routing_reasons(
                pending_count=len(pending_items),
                highest_risk=highest_risk,
                object_types=object_types,
                risk_groups=risk_groups,
                cross_domain=cross_domain,
                review_action=review_action,
                approval_state=approval_state,
            )
            suggested_reviewer = self._suggest_reviewer(
                members=members,
                highest_risk=highest_risk,
                cross_domain=cross_domain,
            )
            warning = None
            fallback_role = None
            if not members:
                warning = "reviewer_missing"
                fallback_role = "OWNER"
            elif suggested_reviewer is None:
                warning = "reviewer_missing"
                fallback_role = "OWNER"
            elif not change_set_evidence:
                warning = "evidence_missing"
            if decision_visible and not pending_items and approval_state in {"approved", "changes_requested", "rejected"}:
                warning = None
                fallback_role = None
            if warning:
                warnings.append(f"reviewer_routing_warning:{change_set_id}:{warning}")
            source_type = self._reviewer_routing_source_type(change_set_evidence)
            priority = self._reviewer_routing_priority(highest_risk=highest_risk, cross_domain=cross_domain)
            routing_items.append(
                {
                    "projectId": str(project_id),
                    "changeSetId": change_set_id,
                    "changeSetTitle": str(change_set.get("headline") or "待审 ChangeSet"),
                    "changeSetStatus": str(change_set.get("status") or change_set.get("rawStatus") or "open"),
                    "sourceType": source_type,
                    "reviewReason": "；".join(review_reasons),
                    "riskLevel": highest_risk,
                    "priority": priority,
                    "changeTypes": object_types,
                    "itemCount": len(change_set.get("itemIds") or review_items),
                    "pendingItemCount": len(pending_items),
                    "evidenceCount": len(change_set_evidence),
                    "suggestedReviewer": suggested_reviewer,
                    "fallbackReviewerRole": fallback_role,
                    "reviewDecision": review_action,
                    "approvalState": approval_state,
                    "requestedAt": review_meta.get("requestedAt"),
                    "decidedAt": review_meta.get("decidedAt"),
                    "decisionReason": review_meta.get("decisionReason"),
                    "targetUrl": f"/change-review?projectId={int(project_id)}&changeSetId={change_set_id}",
                    "createdAt": change_set.get("createdAt"),
                    "updatedAt": change_set.get("updatedAt") or change_set.get("createdAt"),
                }
            )
        missing_reviewer_count = sum(
            1
            for item in routing_items
            if self._reviewer_routing_requires_reviewer(item) and item.get("suggestedReviewer") is None
        )
        warning_code = self._reviewer_routing_warning_code(routing_items)
        status = "empty" if not routing_items else ("warning" if warning_code else "ready")
        return (
            {
                "status": status,
                "warningCode": warning_code,
                "items": routing_items,
                "summary": {
                    "totalPendingReview": sum(int(item.get("pendingItemCount") or 0) for item in routing_items),
                    "highRiskCount": sum(1 for item in routing_items if item.get("riskLevel") == "high"),
                    "missingReviewerCount": missing_reviewer_count,
                    "iterationDraftCount": sum(1 for item in routing_items if item.get("sourceType") == "iteration_recommendation"),
                    "requestedReviewCount": sum(1 for item in routing_items if item.get("approvalState") == "requested"),
                    "approvedCount": sum(1 for item in routing_items if item.get("approvalState") == "approved"),
                    "changesRequestedCount": sum(1 for item in routing_items if item.get("approvalState") == "changes_requested"),
                    "rejectedCount": sum(1 for item in routing_items if item.get("approvalState") == "rejected"),
                },
            },
            warnings,
        )

    @staticmethod
    def _highest_risk_level(levels: list[str]) -> str:
        rank = {"low": 0, "medium": 1, "high": 2}
        highest = "low"
        for level in levels:
            normalized = str(level or "medium").lower()
            if rank.get(normalized, 1) > rank[highest]:
                highest = normalized
        return highest

    @staticmethod
    def _review_routing_reasons(
        *,
        pending_count: int,
        highest_risk: str,
        object_types: list[str],
        risk_groups: list[str],
        cross_domain: bool,
        review_action: Optional[str] = None,
        approval_state: str = "not_requested",
    ) -> list[str]:
        reasons = [f"{pending_count} 个变更项仍处于待审状态。"] if pending_count else ["当前 ChangeSet 已形成审查决策，等待下一步处理。"]
        if approval_state == "requested":
            reasons.append("已请求 reviewer 审查，等待审查意见。")
        elif approval_state == "approved":
            reasons.append("reviewer 已通过，仍需人工决定是否 commit。")
        elif approval_state == "changes_requested":
            reasons.append("reviewer 要求修改，不能直接写入 Project Memory。")
        elif approval_state == "rejected":
            reasons.append("reviewer 已拒绝，本轮不得写入 Project Memory。")
        if highest_risk == "high":
            reasons.append("包含 high risk 变更，需要 REVIEWER 或 OWNER 复核。")
        if cross_domain:
            reasons.append(f"变更跨越 {' / '.join(object_types)}，需要项目 owner 统一审查口径。")
        if "risk" in {group.lower() for group in risk_groups}:
            reasons.append("包含风险类变更，审查时需要确认缓解策略是否足够。")
        if "decision" in {group.lower() for group in risk_groups}:
            reasons.append("包含决策类变更，审查时需要确认拍板依据。")
        if review_action == "request_changes":
            reasons.append("下一步应回到 Change Review 修改内容后再重新请求审查。")
        return reasons

    @staticmethod
    def _reviewer_routing_priority(*, highest_risk: str, cross_domain: bool) -> str:
        if highest_risk == "high":
            return "urgent"
        if cross_domain:
            return "elevated"
        return "normal"

    @staticmethod
    def _reviewer_routing_source_type(evidence: list[dict[str, Any]]) -> str:
        source_types = {str(row.get("sourceType") or "").strip() for row in evidence}
        if "iteration_recommendation_group" in source_types or "iteration_recommendation" in source_types:
            return "iteration_recommendation"
        if "thread_message" in source_types:
            return "ai_extraction"
        if source_types:
            return "manual"
        return "unknown"

    @staticmethod
    def _reviewer_routing_warning_code(items: list[dict[str, Any]]) -> Optional[str]:
        if not items:
            return None
        active_items = [item for item in items if PMFlowOSProjectionService._reviewer_routing_requires_reviewer(item)]
        if any(item.get("suggestedReviewer") is None for item in active_items):
            return "reviewer_missing"
        if any(int(item.get("evidenceCount") or 0) == 0 for item in active_items):
            return "evidence_missing"
        return None

    @staticmethod
    def _reviewer_routing_requires_reviewer(item: dict[str, Any]) -> bool:
        approval_state = str(item.get("approvalState") or "not_requested")
        if approval_state in {"approved", "changes_requested", "rejected"}:
            return int(item.get("pendingItemCount") or 0) > 0
        return True

    @staticmethod
    def _suggest_reviewer(
        *,
        members: list[User],
        highest_risk: str,
        cross_domain: bool,
    ) -> Optional[dict[str, str]]:
        reviewers = [user for user in members if str(user.role or "").upper() == "REVIEWER"]
        owners = [user for user in members if str(user.role or "").upper() == "OWNER"]
        if reviewers:
            return PMFlowOSProjectionService._reviewer_payload(
                reviewers[0],
                "REVIEWER 角色，优先复核 high risk / 跨类型 ChangeSet。",
                "high" if highest_risk == "high" else "medium",
            )
        if owners:
            reason = "项目 OWNER，适合统一跨对象变更审查口径。" if cross_domain else "项目 OWNER，可作为本轮变更审查负责人。"
            return PMFlowOSProjectionService._reviewer_payload(owners[0], reason, "medium")
        return None

    @staticmethod
    def _reviewer_payload(user: User, reason: str, confidence: str) -> dict[str, str]:
        return {
            "userId": str(user.id),
            "displayName": user.name or user.username or f"User#{user.id}",
            "role": str(user.role or "MEMBER"),
            "reason": reason,
            "confidence": confidence,
        }

    @staticmethod
    def _change_set_review_decision_meta(change_set: dict[str, Any]) -> dict[str, Any]:
        routing_meta = change_set.get("routingMeta")
        if not isinstance(routing_meta, dict):
            return {}
        review_meta = routing_meta.get("reviewDecision")
        return dict(review_meta) if isinstance(review_meta, dict) else {}

    @staticmethod
    def _review_decision_action(review_meta: dict[str, Any]) -> Optional[str]:
        action = str(review_meta.get("action") or "").strip()
        return action if action in {"request_review", "approve", "request_changes", "reject"} else None

    @staticmethod
    def _approval_state(review_meta: dict[str, Any]) -> str:
        state = str(review_meta.get("approvalState") or "").strip()
        if state in {"requested", "approved", "changes_requested", "rejected"}:
            return state
        return "not_requested"

    @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 _load_project_memory_semantic_facts(
        self,
        *,
        project_id: int,
        truth_nodes: list[TruthNode],
        tasks: list[tuple[Task, Optional[User]]],
    ) -> list[SemanticFact]:
        truth_ids = [int(node.id) for node in truth_nodes]
        task_ids = [int(task.id) for task, _ in tasks]
        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(), SemanticFact.id.desc())
            .limit(500)
        )
        return list(result.scalars().all())

    async def _memory_evidence_by_item(self, *, project_id: int) -> dict[str, list[dict[str, Any]]]:
        result = await self.db.execute(
            select(PMFlowEvidenceLink)
            .where(and_(PMFlowEvidenceLink.tenant_id == self.tenant_id, PMFlowEvidenceLink.project_id == int(project_id)))
            .order_by(PMFlowEvidenceLink.updated_at.desc(), PMFlowEvidenceLink.created_at.desc(), PMFlowEvidenceLink.id.desc())
            .limit(500)
        )
        evidence_by_item: dict[str, list[dict[str, Any]]] = {}
        evidence_links = list(result.scalars().all())
        change_item_ids = [int(evidence.change_item_id) for evidence in evidence_links if evidence.change_item_id is not None]
        change_items_by_id: dict[int, PMFlowChangeItem] = {}
        if change_item_ids:
            items_result = await self.db.execute(
                select(PMFlowChangeItem).where(
                    and_(
                        PMFlowChangeItem.tenant_id == self.tenant_id,
                        PMFlowChangeItem.project_id == int(project_id),
                        PMFlowChangeItem.id.in_(change_item_ids),
                    )
                )
            )
            change_items_by_id = {int(item.id): item for item in items_result.scalars().all()}
        for evidence in evidence_links:
            if evidence.change_item_id is None:
                continue
            evidence_ref = {
                "id": f"evidence-{evidence.id}",
                "sourceType": evidence.source_type,
                "sourceId": evidence.source_id,
                "title": evidence.title,
                "summary": evidence.excerpt,
                "displayLabel": evidence.display_label,
            }
            evidence_by_item.setdefault(str(evidence.change_item_id), []).append(evidence_ref)
            change_item = change_items_by_id.get(int(evidence.change_item_id))
            if change_item is not None and change_item.object_id:
                evidence_by_item.setdefault(str(change_item.object_id), []).append(evidence_ref)
        return evidence_by_item

    def _memory_item_from_truth_node(self, node: TruthNode, *, evidence_by_item: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
        content = self._payload(node.content)
        node_type = self._node_type(node)
        category = self._memory_category_from_truth_type(node_type)
        return {
            "id": f"truth-node-{node.id}",
            "objectId": str(node.id),
            "objectType": node_type or "truth_node",
            "category": category,
            "title": node.title,
            "summary": self._node_summary(node),
            "status": self._truth_status(node.status),
            "stage": self._stage_from_payload(content, default="planning"),
            "riskLevel": self._risk_level(content.get("level") or content.get("risk_level") or node.status) if category in {"risk", "constraint"} else None,
            "sourceType": "truth_nodes",
            "sourceId": str(node.id),
            "confidence": content.get("confidence"),
            "tags": [node_type] + self._list_from_payload(content, "tags", "labels", "keywords"),
            "evidenceRefs": evidence_by_item.get(str(node.id), []),
            "impactRefs": self._memory_impact_refs(category=category),
            "createdAt": node.created_at.isoformat() if node.created_at else None,
            "updatedAt": node.updated_at.isoformat() if node.updated_at else None,
            "metadata": {
                "version": node.version,
                "rawStatus": node.status,
                "truthNodeType": node.node_type,
                "content": self._jsonable(content),
            },
        }

    def _memory_item_from_task(self, task: Task, *, user: Optional[User]) -> dict[str, Any]:
        return {
            "id": f"task-{task.id}",
            "objectId": str(task.id),
            "objectType": "task",
            "category": "task",
            "title": task.title,
            "summary": task.description or "",
            "status": self._object_status(task.status),
            "stage": self._task_stage(task.status),
            "riskLevel": None,
            "sourceType": "tasks",
            "sourceId": str(task.id),
            "confidence": self._jsonable(task.quality_score) if task.quality_score is not None else None,
            "tags": ["task", task.priority],
            "evidenceRefs": [],
            "impactRefs": [{"type": "truth_node", "id": str(task.truth_node_id), "label": "关联真相节点"}],
            "createdAt": task.created_at.isoformat() if task.created_at else None,
            "updatedAt": task.updated_at.isoformat() if task.updated_at else None,
            "metadata": {
                "assignee": user.name if user is not None else "未分配",
                "priority": task.priority,
                "dueDate": task.due_date.isoformat() if task.due_date else None,
                "completedAt": task.completed_at.isoformat() if task.completed_at else None,
            },
        }

    def _memory_item_from_semantic_fact(self, fact: SemanticFact) -> dict[str, Any]:
        title = f"{fact.fact_namespace}.{fact.fact_key}" if fact.fact_namespace else fact.fact_key
        return {
            "id": f"semantic-fact-{fact.id}",
            "objectId": str(fact.id),
            "objectType": "semantic_fact",
            "category": "fact",
            "title": title,
            "summary": fact.fact_value_text,
            "status": "confirmed" if fact.confidence >= 0.7 else "draft",
            "stage": "planning",
            "riskLevel": None,
            "sourceType": "semantic_facts",
            "sourceId": str(fact.id),
            "confidence": fact.confidence,
            "tags": [fact.fact_namespace, fact.fact_key, fact.entity_type],
            "evidenceRefs": [
                {
                    "id": f"{fact.evidence_source_type or 'source'}-{fact.evidence_source_id}",
                    "sourceType": fact.evidence_source_type,
                    "sourceId": None if fact.evidence_source_id is None else str(fact.evidence_source_id),
                    "title": fact.evidence_source_type or "语义证据",
                    "summary": fact.evidence_excerpt or "",
                    "displayLabel": fact.evidence_source_type or "语义证据",
                }
            ] if fact.evidence_source_type or fact.evidence_excerpt else [],
            "impactRefs": [{"type": fact.entity_type, "id": str(fact.entity_id), "label": "语义绑定对象"}],
            "createdAt": fact.created_at.isoformat() if fact.created_at else None,
            "updatedAt": fact.updated_at.isoformat() if fact.updated_at else None,
            "metadata": {
                "entityType": fact.entity_type,
                "entityId": str(fact.entity_id),
                "valueType": fact.value_type,
                "extractVersion": fact.extract_version,
                "freshnessAt": fact.freshness_at,
            },
        }

    def _memory_asset_from_evidence(self, evidence: PMFlowEvidenceLink) -> dict[str, Any]:
        metadata = self._payload(evidence.evidence_metadata)
        asset_type = self._memory_asset_type(evidence.source_type)
        return {
            "id": f"evidence-{evidence.id}",
            "assetType": asset_type,
            "sourceType": evidence.source_type,
            "sourceId": evidence.source_id,
            "title": evidence.title,
            "summary": evidence.excerpt,
            "status": "indexed",
            "displayLabel": evidence.display_label,
            "linkedItemId": f"change-item-{evidence.change_item_id}" if evidence.change_item_id is not None else None,
            "changeSetId": str(evidence.change_set_id),
            "createdAt": evidence.created_at.isoformat() if evidence.created_at else None,
            "updatedAt": evidence.updated_at.isoformat() if evidence.updated_at else None,
            "metadata": self._jsonable(metadata),
        }

    def _memory_asset_from_thread_message(self, message: PMFlowThreadMessage) -> dict[str, Any]:
        return {
            "id": f"thread-message-{message.id}",
            "assetType": "conversation",
            "sourceType": message.source,
            "sourceId": str(message.id),
            "title": "项目线程消息",
            "summary": self._short_text(message.content),
            "status": message.status,
            "displayLabel": "对话",
            "linkedItemId": None,
            "changeSetId": None,
            "createdAt": message.created_at.isoformat() if message.created_at else None,
            "updatedAt": message.updated_at.isoformat() if message.updated_at else None,
            "metadata": {
                "threadId": str(message.thread_id),
                "role": message.role,
                "attachments": self._jsonable(message.attachments),
                "messageMetadata": self._jsonable(message.message_metadata),
            },
        }

    def _memory_context_pack(
        self,
        *,
        key: str,
        title: str,
        summary: str,
        usage: str,
        item_ids: list[str],
        asset_ids: list[str],
    ) -> dict[str, Any]:
        return {
            "id": f"context-pack-{key}",
            "key": key,
            "title": title,
            "summary": summary,
            "usage": usage,
            "status": "ready" if item_ids or asset_ids else "empty",
            "itemCount": len(item_ids),
            "assetCount": len(asset_ids),
            "sourceItemIds": item_ids[:80],
            "sourceAssetIds": asset_ids[:80],
            "updatedAt": now_cn_naive().isoformat() if item_ids or asset_ids else None,
            "warnings": [] if item_ids or asset_ids else ["context_pack_empty"],
        }

    def _memory_writeback_from_change_set(
        self,
        change_set: PMFlowChangeSet,
        *,
        items: list[PMFlowChangeItem],
        committed_by: Optional[str] = None,
    ) -> dict[str, Any]:
        committed = [item for item in items if item.status == "committed"]
        skipped = [item for item in items if item.status in {"deferred", "skipped"}]
        pending = [item for item in items if item.status in {"pending_review", "accepted", "edited"}]
        return {
            "id": f"writeback-{change_set.id}",
            "changeSetId": str(change_set.id),
            "title": change_set.headline,
            "summary": change_set.summary,
            "status": change_set.status,
            "committedAt": change_set.committed_at.isoformat() if change_set.committed_at else None,
            "committedBy": committed_by,
            "committedById": str(change_set.committed_by) if change_set.committed_by else None,
            "createdAt": change_set.created_at.isoformat() if change_set.created_at else None,
            "updatedAt": change_set.updated_at.isoformat() if change_set.updated_at else None,
            "itemCounts": {
                "total": len(items),
                "committed": len(committed),
                "skipped": len(skipped),
                "pending": len(pending),
            },
            "items": [
                {
                    "id": f"change-item-{item.id}",
                    "objectType": item.object_type,
                    "title": item.title,
                    "summary": item.summary,
                    "status": item.status,
                    "riskLevel": item.risk_level,
                }
                for item in items[:20]
            ],
            "metadata": {
                "extractionScope": change_set.extraction_scope,
                "routingMeta": self._jsonable(change_set.routing_meta),
            },
        }

    def _memory_writeback_from_execution_event(
        self,
        *,
        event: ProgressEvent,
        task: Task,
        user: Optional[User],
    ) -> dict[str, Any]:
        payload = self._payload(event.payload)
        evidence_event_ids = [str(item) for item in payload.get("evidence_event_ids") or []]
        committed_at = event.occurred_at.isoformat() if event.occurred_at else None
        result = str(payload.get("result") or "completed")
        return {
            "id": f"execution-writeback-{event.id}",
            "changeSetId": None,
            "sourceType": "execution_memory_writeback",
            "progressEventId": str(event.id),
            "taskId": str(task.id),
            "title": f"执行结果写回：{task.title}",
            "summary": event.summary or "执行结果已写回项目记忆。",
            "status": "committed" if result == "completed" else result,
            "committedAt": committed_at,
            "committedBy": user.name if user is not None else None,
            "committedById": str(event.user_id) if event.user_id else None,
            "createdAt": committed_at,
            "updatedAt": committed_at,
            "itemCounts": {
                "total": 1,
                "committed": 1 if result == "completed" else 0,
                "skipped": 0,
                "pending": 0,
            },
            "items": [
                {
                    "id": f"task-{task.id}",
                    "objectType": "Task",
                    "title": task.title,
                    "summary": task.description,
                    "status": task.status,
                    "riskLevel": None,
                }
            ],
            "metadata": {
                "protocolVersion": payload.get("protocol_version"),
                "result": result,
                "note": payload.get("note"),
                "taskStatus": payload.get("task_status"),
                "evidenceEventIds": evidence_event_ids,
                "deliveryArtifacts": payload.get("delivery_artifacts") or {},
                "artifactCompleteness": payload.get("artifact_completeness") or {},
                "artifactReview": payload.get("artifact_review") or {},
                "releaseGateStatus": payload.get("release_gate_status"),
                "releaseReadinessProof": payload.get("release_readiness_proof") or {},
                "writebackProof": payload.get("writeback_proof") or {},
                "qualityContractId": payload.get("quality_contract_id"),
            },
        }

    def _filter_memory_rows(
        self,
        rows: list[dict[str, Any]],
        *,
        category: Optional[str],
        stage: Optional[str],
        source_type: Optional[str],
        status: Optional[str],
        query: Optional[str],
    ) -> list[dict[str, Any]]:
        normalized_category = str(category or "").strip()
        normalized_stage = str(stage or "").strip()
        normalized_source_type = str(source_type or "").strip()
        normalized_status = str(status or "").strip()
        normalized_query = str(query or "").strip().lower()
        filtered = [
            row
            for row in rows
            if not normalized_category or str(row.get("category") or row.get("assetType") or "") == normalized_category
        ]
        if normalized_stage:
            filtered = [row for row in filtered if str(row.get("stage") or "") == normalized_stage]
        if normalized_status:
            filtered = [row for row in filtered if str(row.get("status") or "") == normalized_status]
        if normalized_source_type:
            filtered = [row for row in filtered if self._memory_row_matches_source(row, normalized_source_type)]
        if not normalized_query:
            return filtered
        return [
            row
            for row in filtered
            if normalized_query in str(row.get("title") or "").lower()
            or normalized_query in str(row.get("summary") or "").lower()
            or normalized_query in " ".join(str(tag) for tag in row.get("tags") or []).lower()
        ]

    def _memory_row_matches_source(self, row: dict[str, Any], source_type: str) -> bool:
        expected = self._memory_source_bucket(source_type)
        source_values = {str(row.get("sourceType") or "").strip().lower()}
        for evidence in row.get("evidenceRefs") or []:
            if isinstance(evidence, dict):
                source_values.add(str(evidence.get("sourceType") or "").strip().lower())
                source_values.add(str(evidence.get("displayLabel") or "").strip().lower())
        return bool(source_values.intersection(expected))

    def _memory_source_bucket(self, source_type: str) -> set[str]:
        value = str(source_type or "").strip().lower()
        buckets = {
            "conversation": {"conversation", "thread", "message", "web", "feishu", "对话"},
            "document": {"document", "doc", "file", "upload", "文档"},
            "pr": {"pr", "pull_request", "commit", "github", "code", "代码"},
            "test": {"test", "test_log", "validation", "测试"},
            "external": {"external", "research", "web_search", "url", "外部资料"},
            "ai_extraction": {"ai_extraction", "change_review", "semantic_facts", "truth_nodes", "ai 提取", "ai提取"},
        }
        return buckets.get(value, {value})

    def _memory_metric_counts(
        self,
        *,
        items: list[dict[str, Any]],
        assets: list[dict[str, Any]],
        context_packs: list[dict[str, Any]],
    ) -> dict[str, int]:
        return {
            "confirmed_facts": len([item for item in items if item.get("category") == "fact" and item.get("status") == "confirmed"]),
            "requirements": len([item for item in items if item.get("category") == "requirement"]),
            "decisions": len([item for item in items if item.get("category") == "decision"]),
            "constraints": len([item for item in items if item.get("category") in {"constraint", "risk"}]),
            "evidence": len(assets),
            "context_packs": len([pack for pack in context_packs if pack.get("status") == "ready"]),
            "assets": len(assets),
        }

    def _memory_ai_suggestions(
        self,
        *,
        items: list[dict[str, Any]],
        assets: list[dict[str, Any]],
        context_packs: list[dict[str, Any]],
        writebacks: list[dict[str, Any]],
    ) -> list[dict[str, Any]]:
        suggestions: list[dict[str, Any]] = []
        if any(item.get("category") == "requirement" for item in items):
            suggestions.append({"key": "generate_tasks", "title": "生成开发任务与验收标准", "target": "execution-bridge"})
        if any(item.get("category") in {"risk", "constraint"} for item in items):
            suggestions.append({"key": "generate_test_plan", "title": "生成测试计划和风险检查", "target": "release-growth"})
        if assets and context_packs:
            suggestions.append({"key": "refresh_context_pack", "title": "组织下一轮项目上下文包", "target": "context-packs"})
        if writebacks:
            suggestions.append({"key": "inspect_writebacks", "title": "查看最近写入影响", "target": "writebacks"})
        return suggestions[:4]

    def _memory_category_from_truth_type(self, node_type: str) -> str:
        if node_type in {"requirement", "charter", "scope"}:
            return "requirement"
        if node_type in {"decision", "model_decision"}:
            return "decision"
        if node_type == "risk":
            return "risk"
        if node_type in {"constraint", "compliance", "policy"}:
            return "constraint"
        if node_type in {"release"}:
            return "release"
        if node_type in {"metric", "signal"}:
            return "metric"
        if node_type in {"test", "test_run"}:
            return "test"
        return "fact"

    def _memory_asset_type(self, source_type: str) -> str:
        value = str(source_type or "").strip().lower()
        if value in {"conversation", "thread", "message", "feishu"}:
            return "conversation"
        if value in {"pr", "pull_request", "commit", "github", "code"}:
            return "code"
        if value in {"test", "test_log", "validation"}:
            return "test"
        if value in {"feedback", "user_feedback"}:
            return "feedback"
        if value in {"metric", "analytics", "revenue", "roi"}:
            return "metric"
        if value in {"release", "delivery"}:
            return "release"
        if value in {"document", "doc", "file", "upload"}:
            return "document"
        return "evidence"

    def _memory_impact_refs(self, *, category: str) -> list[dict[str, str]]:
        mapping: dict[str, list[dict[str, str]]] = {
            "requirement": [
                {"type": "task", "id": "derived", "label": "任务拆解"},
                {"type": "test", "id": "derived", "label": "测试计划"},
            ],
            "decision": [{"type": "execution", "id": "derived", "label": "执行判断"}],
            "risk": [{"type": "release_gate", "id": "derived", "label": "发布门禁"}],
            "constraint": [{"type": "release_gate", "id": "derived", "label": "约束检查"}],
            "fact": [{"type": "query_answer", "id": "derived", "label": "项目问答"}],
        }
        return mapping.get(category, [])

    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,
        *,
        review_projection: dict[str, list[dict[str, Any]]],
        reviewer_routing: dict[str, Any],
        execution_summary: ExecutionSummaryProjection,
        test_runs: list[PMFlowTestRun],
        releases: list[PMFlowRelease],
        metric_signals: list[PMFlowMetricSignal],
    ) -> list[PMFlowFeedEvent]:
        feed: list[PMFlowFeedEvent] = []
        feed.extend(self._review_activity_feed(review_projection=review_projection, reviewer_routing=reviewer_routing))
        for event in execution_summary.recentProgressEvents[:5]:
            feed.append(
                PMFlowFeedEvent(
                    id=f"progress-{event.get('id')}",
                    kind="execution",
                    title=self._progress_event_title(event),
                    summary=event.get("summary") or event.get("workflow_state") or "执行事件已回写",
                    timestamp=str(event.get("occurred_at") or ""),
                )
            )
        release_gate_event = self._release_gate_feed_event(execution_summary.releaseGate)
        if release_gate_event is not None:
            feed.append(release_gate_event)
        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 sorted(feed, key=lambda item: item.timestamp or "", reverse=True)[:18]

    def _review_activity_feed(
        self,
        *,
        review_projection: dict[str, list[dict[str, Any]]],
        reviewer_routing: dict[str, Any],
    ) -> list[PMFlowFeedEvent]:
        feed: list[PMFlowFeedEvent] = []
        routing_items = list(reviewer_routing.get("items") or [])
        change_sets = list(review_projection.get("change_sets") or [])
        change_items = list(review_projection.get("change_items") or [])
        evidence = list(review_projection.get("evidence") or [])
        pending_by_set: dict[str, int] = {}
        evidence_by_set: dict[str, int] = {}
        for item in change_items:
            if item.get("status") == "pending_review":
                pending_by_set[str(item.get("changeSetId") or "")] = pending_by_set.get(str(item.get("changeSetId") or ""), 0) + 1
        for row in evidence:
            change_set_id = str(row.get("changeSetId") or "")
            evidence_by_set[change_set_id] = evidence_by_set.get(change_set_id, 0) + 1

        for item in routing_items:
            change_set_id = str(item.get("changeSetId") or "")
            approval_state = str(item.get("approvalState") or "not_requested")
            title = str(item.get("changeSetTitle") or "待审 ChangeSet")
            reviewer = self._reviewer_display(item)
            if approval_state == "requested":
                feed.append(
                    PMFlowFeedEvent(
                        id=f"review-request-{change_set_id}",
                        kind="review",
                        title="已请求 ChangeSet 审查",
                        summary=f"{title} 已请求 {reviewer} 审查；{item.get('reviewReason') or '等待 reviewer 给出意见。'}",
                        timestamp=str(item.get("requestedAt") or item.get("updatedAt") or item.get("createdAt") or ""),
                    )
                )
            elif approval_state in {"approved", "changes_requested", "rejected"}:
                feed.append(
                    PMFlowFeedEvent(
                        id=f"review-decision-{change_set_id}-{approval_state}",
                        kind="review",
                        title=self._review_decision_title(approval_state),
                        summary=f"{title}：{item.get('decisionReason') or self._review_decision_summary(approval_state)}",
                        timestamp=str(item.get("decidedAt") or item.get("updatedAt") or item.get("createdAt") or ""),
                    )
                )

        for change_set in change_sets[:10]:
            change_set_id = str(change_set.get("id") or "")
            if not change_set_id:
                continue
            status = str(change_set.get("rawStatus") or change_set.get("status") or "open")
            feed.append(
                PMFlowFeedEvent(
                    id=f"changeset-{change_set_id}-{status}",
                    kind="review",
                    title=self._change_set_status_title(status),
                    summary=self._change_set_status_summary(
                        change_set=change_set,
                        pending_count=pending_by_set.get(change_set_id, 0),
                        evidence_count=evidence_by_set.get(change_set_id, 0),
                    ),
                    timestamp=str(change_set.get("committedAt") or change_set.get("createdAt") or ""),
                )
            )
        return feed

    @staticmethod
    def _reviewer_display(item: dict[str, Any]) -> str:
        reviewer = item.get("suggestedReviewer")
        if isinstance(reviewer, dict):
            display = str(reviewer.get("displayName") or reviewer.get("role") or "").strip()
            if display:
                return display
        return str(item.get("fallbackReviewerRole") or "OWNER")

    @staticmethod
    def _review_decision_title(approval_state: str) -> str:
        return {
            "approved": "审查已通过",
            "changes_requested": "审查要求修改",
            "rejected": "审查已拒绝",
        }.get(approval_state, "审查决策已记录")

    @staticmethod
    def _review_decision_summary(approval_state: str) -> str:
        return {
            "approved": "reviewer 已通过，仍需人工决定是否 commit。",
            "changes_requested": "reviewer 要求修改，不能直接写入 Project Memory。",
            "rejected": "reviewer 已拒绝，本轮不得写入 Project Memory。",
        }.get(approval_state, "审查状态已更新。")

    @staticmethod
    def _change_set_status_title(status: str) -> str:
        return {
            "open": "ChangeSet 已创建",
            "reviewed": "ChangeSet 已完成审查",
            "committed": "ChangeSet 已写入项目真相层",
            "partially_committed": "ChangeSet 已部分写入",
        }.get(status, "ChangeSet 状态已更新")

    @staticmethod
    def _change_set_status_summary(*, change_set: dict[str, Any], pending_count: int, evidence_count: int) -> str:
        title = str(change_set.get("headline") or change_set.get("summary") or "ChangeSet")
        status = str(change_set.get("rawStatus") or change_set.get("status") or "open")
        if status == "open":
            return f"{title} 进入待确认审查；待审 {pending_count} 项，证据 {evidence_count} 条。"
        if status == "reviewed":
            return f"{title} 已完成审查；仍需人工决定是否进入 commit。"
        if status == "committed":
            return f"{title} 已写入 Project Memory；动态流仅展示结果，不触发写入。"
        if status == "partially_committed":
            return f"{title} 已部分写入 Project Memory；未写入项仍需回到 Change Review。"
        return f"{title} 状态更新为 {status}。"

    @staticmethod
    def _progress_event_title(event: dict[str, Any]) -> str:
        event_type = str(event.get("event_type") or "")
        return {
            "handoff_created": "执行交接已创建",
            "handoff_consumed": "执行交接已领取",
            "branch_created": "执行分支已创建",
            "commit": "执行 commit 已回写",
            "pr_opened": "PR 已打开",
            "pr_synced": "PR 状态已同步",
            "test_passed": "测试通过",
            "test_failed": "测试失败",
            "checks_passed": "检查通过",
            "failed": "执行失败",
            "blocked": "执行阻塞",
            "merged": "PR 已合并",
            "accepted": "执行结果已接受",
            "execution_memory_writeback": "执行结果已写回项目记忆",
            "manual_progress": "执行进展已记录",
        }.get(event_type, event_type or "执行进展")

    @staticmethod
    def _release_gate_feed_event(release_gate: dict[str, Any]) -> Optional[PMFlowFeedEvent]:
        status = str((release_gate or {}).get("status") or "empty")
        if status == "empty":
            return None
        return PMFlowFeedEvent(
            id=f"release-gate-{status}",
            kind="release",
            title={
                "ready": "发布门禁已放行",
                "blocked": "发布门禁已阻塞",
                "needs_confirmation": "发布门禁需要确认",
            }.get(status, "发布门禁已更新"),
            summary=str((release_gate or {}).get("summary") or (release_gate or {}).get("title") or "发布门禁状态已更新。"),
            timestamp=str((release_gate or {}).get("updatedAt") or ""),
        )

    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 _list_from_payload(self, payload: dict[str, Any], *keys: str) -> list[str]:
        for key in keys:
            value = payload.get(key)
            if isinstance(value, list):
                return [str(item) for item in value if str(item).strip()]
            if isinstance(value, str) and value.strip():
                return [value.strip()]
        return []

    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"
        event_type = self._event_type(event)
        if event_type == "execution_memory_writeback":
            return "project_memory_written"
        if event_type in {"accepted", "technical_review_approved", "merged"}:
            return "pending_memory_writeback"
        return str(event.get("workflow_state") or event.get("summary") or event.get("event_type") or "progress event recorded")

    def _build_execution_acceptance(
        self,
        *,
        events: list[dict[str, Any]],
        quality_contract: dict[str, Any],
        artifact_completeness: Optional[dict[str, Any]] = None,
        artifact_review: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        failed_events = self._unresolved_failed_evidence_events(
            events,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        has_test_passed = self._acceptance_has_required_test_evidence(events)
        has_accept = any(self._event_type(event) == "accepted" for event in events)
        has_changes_requested = any(self._event_type(event) == "changes_requested" for event in events)
        has_rejected = any(self._event_type(event) == "acceptance_rejected" for event in events)
        has_technical_requested = any(self._event_type(event) == "technical_review_requested" for event in events)
        has_technical_approved = any(self._event_type(event) == "technical_review_approved" for event in events)
        technical_required = bool((quality_contract.get("acceptanceGate") or {}).get("technicalReviewRequired"))
        blocking_reasons: list[str] = []
        status = "waiting_execution"
        required_action = "wait_for_execution_result"
        if failed_events:
            status = "execution_blocked"
            required_action = "fix_and_retest"
            blocking_reasons = [self._event_reason(event) for event in failed_events[:5]]
        elif has_rejected:
            status = "rejected"
            required_action = "replan_or_create_fix_handoff"
        elif has_changes_requested:
            status = "changes_requested"
            required_action = "create_fix_handoff"
        elif has_technical_approved and not has_accept:
            status = "technical_review_approved"
            required_action = "accept_or_release_decision"
        elif has_accept:
            status = "accepted"
            required_action = "enter_release_gate"
        elif technical_required and has_test_passed and not has_technical_approved:
            status = "pending_technical_review"
            required_action = "approve_technical_review_or_request_changes"
            if not has_technical_requested:
                required_action = "request_technical_review"
        elif has_test_passed:
            status = "pending_user_acceptance"
            required_action = "accept_or_request_changes"
        elif events:
            status = "waiting_test_evidence"
            required_action = "run_required_tests"
            blocking_reasons = ["缺少 test_passed 或 checks_passed 证据。"]
        quality_summary = quality_contract.get("summary") if isinstance(quality_contract.get("summary"), dict) else {}
        if int(quality_summary.get("releaseBlockingMissing") or 0) > 0 and status not in {"execution_blocked", "accepted"}:
            blocking_reasons.extend(list(quality_contract.get("blockingReasons") or [])[:5])
        return {
            "status": status,
            "requiredAction": required_action,
            "reviewMode": "technical_review" if technical_required else "user_acceptance",
            "reviewerRequired": technical_required,
            "technicalReviewRequested": has_technical_requested,
            "technicalReviewApproved": has_technical_approved,
            "userAccepted": has_accept,
            "canEnterReleaseGate": status == "accepted" or (status == "technical_review_approved" and not technical_required),
            "blockingReasons": self._unique_strings(blocking_reasons),
            "latestDecision": self._latest_acceptance_decision(events),
            "evidenceProof": self._execution_evidence_proof(
                events=events,
                artifact_completeness=artifact_completeness,
                artifact_review=artifact_review,
            ),
        }

    @staticmethod
    def _acceptance_has_required_test_evidence(events: list[dict[str, Any]]) -> bool:
        return any(PMFlowOSProjectionService._event_type(event) in {"test_passed", "checks_passed"} for event in events)

    def _latest_acceptance_decision(self, events: list[dict[str, Any]]) -> Optional[dict[str, Any]]:
        decision_types = {"accepted", "changes_requested", "acceptance_rejected", "technical_review_requested", "technical_review_approved"}
        for event in events:
            if self._event_type(event) in decision_types:
                payload = self._payload(event.get("payload"))
                return {
                    "eventId": str(event.get("id") or ""),
                    "eventType": self._event_type(event),
                    "decision": payload.get("decision") or self._event_type(event),
                    "summary": self._event_reason(event),
                    "occurredAt": event.get("occurred_at"),
                }
        return None

    def _execution_acceptance_summary(self, *, decision: str, task: Task, note: Optional[str]) -> str:
        label = {
            "accept": "用户验收通过",
            "request_changes": "要求修改",
            "reject": "验收拒绝",
            "request_technical_review": "请求技术审查",
            "approve_technical_review": "技术审查通过",
        }.get(decision, "验收决策")
        suffix = f"：{note}" if note else ""
        return f"{label}：{task.title}{suffix}"

    def _execution_memory_writeback_summary(self, *, task: Task, result: str, note: Optional[str]) -> str:
        label = {
            "completed": "执行结果已写回项目记忆",
            "blocked": "执行阻塞已写回项目记忆",
            "changes_requested": "执行返工要求已写回项目记忆",
        }.get(result, "执行结果已写回项目记忆")
        suffix = f"：{note}" if note else ""
        return f"{label}：{task.title}{suffix}"

    def _execution_memory_writeback_task_payload(
        self,
        *,
        task: Task,
        existing: Optional[dict[str, Any]],
        result: str,
        evidence_event_ids: list[str],
        delivery_artifacts: dict[str, Any],
        artifact_completeness: dict[str, Any],
        artifact_review: dict[str, Any],
        release_gate: dict[str, Any],
        writeback_proof: dict[str, Any],
        note: Optional[str],
        current_user_id: Optional[int],
        occurred_at: datetime,
    ) -> dict[str, Any]:
        payload = dict(existing or {})
        history = list(payload.get("execution_memory_writebacks") or [])
        history.append(
            {
                "protocol_version": "execution_memory_writeback_v1",
                "result": result,
                "task_id": str(task.id),
                "task_status": task.status,
                "evidence_event_ids": list(evidence_event_ids),
                "delivery_artifacts": delivery_artifacts,
                "artifact_completeness": artifact_completeness,
                "artifact_review": artifact_review,
                "release_gate_status": release_gate.get("status"),
                "release_readiness_proof": release_gate.get("releaseReadinessProof") or {},
                "writeback_proof": writeback_proof,
                "note": note,
                "written_by": None if current_user_id is None else int(current_user_id),
                "written_at": occurred_at.isoformat(),
            }
        )
        payload["execution_memory_writebacks"] = history[-20:]
        payload["latest_execution_memory_writeback"] = history[-1]
        return payload

    @staticmethod
    def _latest_repo_binding_id(events: list[dict[str, Any]]) -> Optional[int]:
        for event in events:
            repo = event.get("repo")
            if isinstance(repo, dict) and repo.get("id"):
                try:
                    return int(str(repo.get("id")))
                except ValueError:
                    return None
        return None

    def _delivery_artifacts_projection(self, events: list[dict[str, Any]]) -> dict[str, Any]:
        artifact_events: list[dict[str, Any]] = []
        merged = {
            "changed_files": [],
            "commands_run": [],
            "test_outputs": [],
            "screenshots": [],
            "logs": [],
            "risk_notes": [],
            "review_checklist": [],
        }
        for event in events:
            artifacts = self._event_delivery_artifacts(event)
            if not artifacts:
                continue
            artifact_events.append(
                {
                    "eventId": str(event.get("id") or ""),
                    "eventType": self._event_type(event),
                    "source": event.get("source"),
                    "summary": event.get("summary"),
                    "occurredAt": event.get("occurred_at"),
                    "deliveryId": event.get("delivery_id"),
                    "eventSourceId": event.get("event_source_id"),
                }
            )
            for key in merged:
                values = artifacts.get(key)
                if isinstance(values, list):
                    merged[key].extend([item for item in values if isinstance(item, dict)])
        capped = {key: values[:50] for key, values in merged.items()}
        return {
            "status": "active" if artifact_events else "empty",
            "latestEvent": artifact_events[0] if artifact_events else None,
            "events": artifact_events[:12],
            "counts": {key: len(values) for key, values in merged.items()},
            **capped,
        }

    def _artifact_completeness(
        self,
        *,
        delivery_artifacts: dict[str, Any],
        events: list[dict[str, Any]],
        quality_contract: dict[str, Any],
    ) -> dict[str, Any]:
        if not delivery_artifacts or delivery_artifacts.get("status") == "empty":
            return {
                "status": "empty",
                "summary": "还没有结构化交付证据包。",
                "missing": ["delivery_artifacts"],
                "blockingReasons": ["缺少 changed files、commands run、test outputs 等结构化交付证据。"],
                "releaseBlockingMissing": 1,
            }
        changed_files = list(delivery_artifacts.get("changed_files") or [])
        commands = list(delivery_artifacts.get("commands_run") or [])
        tests = list(delivery_artifacts.get("test_outputs") or [])
        screenshots = list(delivery_artifacts.get("screenshots") or [])
        checklist = list(delivery_artifacts.get("review_checklist") or [])
        effective_commands = self._latest_artifact_items_by_key(commands, keys=("command",))
        effective_tests = self._latest_artifact_items_by_key(tests, keys=("name", "command"))
        effective_checklist = self._latest_artifact_items_by_key(checklist, keys=("item",))
        missing: list[str] = []
        if not changed_files:
            missing.append("changed_files")
        if not commands:
            missing.append("commands_run")
        if not tests:
            missing.append("test_outputs")
        ui_changed = any(self._artifact_path_is_ui(str(item.get("path") or "")) for item in changed_files if isinstance(item, dict))
        if ui_changed and not screenshots:
            missing.append("screenshots_for_ui_change")
        required_review = bool((quality_contract.get("acceptanceGate") or {}).get("technicalReviewRequired"))
        if required_review and not checklist:
            missing.append("review_checklist")
        failed_items = [
            str(item.get("command") or item.get("name") or item.get("item") or "artifact")
            for item in [*effective_commands, *effective_tests, *effective_checklist]
            if isinstance(item, dict) and str(item.get("status") or "").lower() in {"failed", "blocked"}
        ]
        has_test_passed = any(self._event_type(event) in {"test_passed", "checks_passed"} for event in events)
        if tests and not any(isinstance(item, dict) and str(item.get("status") or "").lower() == "passed" for item in effective_tests) and not has_test_passed:
            missing.append("passed_test_output")
        status = "blocked" if failed_items else ("ready" if not missing else "incomplete")
        blocking = [f"交付证据失败：{item}" for item in failed_items[:5]]
        if missing:
            blocking.extend([self._artifact_missing_label(item) for item in missing])
        return {
            "status": status,
            "summary": {
                "ready": "结构化交付证据已满足发布前审查要求。",
                "blocked": "结构化交付证据中存在失败或阻塞项。",
                "incomplete": "结构化交付证据还不完整。",
            }.get(status, "结构化交付证据待补齐。"),
            "missing": self._unique_strings(missing),
            "blockingReasons": self._unique_strings(blocking),
            "releaseBlockingMissing": len(failed_items) + len(missing),
            "counts": dict(delivery_artifacts.get("counts") or {}),
            "uiChanged": ui_changed,
        }

    def _artifact_review_projection(
        self,
        *,
        delivery_artifacts: dict[str, Any],
        artifact_completeness: dict[str, Any],
        events: list[dict[str, Any]],
    ) -> dict[str, Any]:
        status = str((artifact_completeness or {}).get("status") or "empty")
        missing = [str(item) for item in list((artifact_completeness or {}).get("missing") or []) if str(item)]
        items: list[dict[str, Any]] = []
        for key in missing:
            items.append(self._artifact_review_missing_item(key))
        commands = self._latest_artifact_items_by_key(
            [item for item in list((delivery_artifacts or {}).get("commands_run") or []) if isinstance(item, dict)],
            keys=("command",),
        )
        tests = self._latest_artifact_items_by_key(
            [item for item in list((delivery_artifacts or {}).get("test_outputs") or []) if isinstance(item, dict)],
            keys=("name", "command"),
        )
        checklist = self._latest_artifact_items_by_key(
            [item for item in list((delivery_artifacts or {}).get("review_checklist") or []) if isinstance(item, dict)],
            keys=("item",),
        )
        for index, item in enumerate(commands):
            item_status = str(item.get("status") or "").lower()
            if item_status in {"failed", "blocked"}:
                items.append(
                    self._artifact_review_failed_item(
                        item_id=f"failed_command_{index}",
                        kind="fix_failed_command",
                        title="修复失败命令",
                        subject=str(item.get("command") or "command"),
                        summary=str(item.get("summary") or "命令执行失败，需要修复后重新回写。"),
                    )
                )
        for index, item in enumerate(tests):
            item_status = str(item.get("status") or "").lower()
            if item_status in {"failed", "blocked"}:
                items.append(
                    self._artifact_review_failed_item(
                        item_id=f"failed_test_{index}",
                        kind="fix_failed_test",
                        title="修复失败测试",
                        subject=str(item.get("name") or "test"),
                        summary=str(item.get("summary") or item.get("command") or "测试输出失败，需要修复后重新回写。"),
                    )
                )
        for index, item in enumerate(checklist):
            item_status = str(item.get("status") or "").lower()
            if item_status in {"failed", "blocked", "needs_review"}:
                items.append(
                    self._artifact_review_failed_item(
                        item_id=f"review_checklist_{index}",
                        kind="resolve_review_checklist",
                        title="处理审查清单",
                        subject=str(item.get("item") or "review checklist"),
                        summary=str(item.get("summary") or "审查清单未通过或待复核，需要处理后重新回写。"),
                        severity="medium" if item_status == "needs_review" else "high",
                    )
                )
        unique_items: list[dict[str, Any]] = []
        seen: set[str] = set()
        for item in items:
            item_id = str(item.get("id") or "")
            if item_id in seen:
                continue
            seen.add(item_id)
            unique_items.append(item)
        history = self._artifact_review_action_history(events)
        closure = self._artifact_review_closure(
            delivery_artifacts=delivery_artifacts,
            events=events,
            open_items=unique_items,
        )
        review_status = "clear" if status == "ready" else ("blocked" if status == "blocked" else ("open" if unique_items else status))
        return {
            "status": review_status,
            "summary": self._artifact_review_summary(status=review_status, item_count=len(unique_items)),
            "items": unique_items[:30],
            "openCount": len(unique_items),
            "actionHistory": history,
            "closure": closure,
            "nextActions": self._artifact_review_next_actions(status=review_status, items=unique_items),
            "boundary": "Artifact review only records review/rework intent and expected evidence. PMFlow does not execute commands or create formal tasks from these items.",
        }

    @staticmethod
    def _latest_artifact_items_by_key(items: list[dict[str, Any]], *, keys: tuple[str, ...]) -> list[dict[str, Any]]:
        latest: list[dict[str, Any]] = []
        seen: set[str] = set()
        for index, item in enumerate(items):
            identity = ""
            for key in keys:
                raw = str(item.get(key) or "").strip().lower()
                if raw:
                    identity = f"{key}:{raw}"
                    break
            if not identity:
                identity = f"artifact_index:{index}"
            if identity in seen:
                continue
            seen.add(identity)
            latest.append(item)
        return latest

    def _artifact_review_missing_item(self, key: str) -> dict[str, Any]:
        mapping = {
            "delivery_artifacts": ("missing_delivery_artifacts", "补齐交付证据包", "运行端需要回写 changed files、commands run、test outputs。"),
            "changed_files": ("missing_changed_files", "补 changed files", "需要列出本次交付涉及的新增、修改、删除或重命名文件。"),
            "commands_run": ("missing_commands_run", "补 commands run", "需要回写实际运行过的检查、测试或构建命令。"),
            "test_outputs": ("missing_test_outputs", "补 test outputs", "需要回写测试名称、通过/失败状态和摘要。"),
            "screenshots_for_ui_change": ("missing_ui_screenshots", "补 UI 截图", "检测到前端/UI 改动，需要补充截图 artifact ref 或截图摘要。"),
            "review_checklist": ("missing_review_checklist", "补 review checklist", "当前质量合同需要技术审查清单，必须补齐审查项状态。"),
            "passed_test_output": ("missing_passed_test_output", "补通过测试输出", "已有测试输出但缺少 passed 状态，需要重新回写通过证据或保留失败并进入返工。"),
        }
        item_id, title, summary = mapping.get(key, (f"missing_{key}", self._artifact_missing_label(key), self._artifact_missing_label(key)))
        return {
            "id": item_id,
            "kind": key,
            "status": "open",
            "severity": "medium" if key in {"review_checklist", "screenshots_for_ui_change"} else "high",
            "title": title,
            "summary": summary,
            "recommendedAction": "重新回写结构化交付证据",
            "suggestedEventType": "manual_progress",
            "expectedEvidence": self._artifact_expected_evidence(key),
        }

    @staticmethod
    def _artifact_review_failed_item(
        *,
        item_id: str,
        kind: str,
        title: str,
        subject: str,
        summary: str,
        severity: str = "high",
    ) -> dict[str, Any]:
        return {
            "id": item_id,
            "kind": kind,
            "status": "blocked",
            "severity": severity,
            "title": title,
            "summary": f"{subject}：{summary}",
            "recommendedAction": "修复后重新回写 test_passed/checks_passed 与 delivery_artifacts",
            "suggestedEventType": "test_passed",
            "expectedEvidence": ["commands_run.status=passed", "test_outputs.status=passed"],
        }

    @staticmethod
    def _artifact_expected_evidence(key: str) -> list[str]:
        return {
            "delivery_artifacts": ["changed_files", "commands_run", "test_outputs"],
            "changed_files": ["delivery_artifacts.changed_files[].path", "delivery_artifacts.changed_files[].summary"],
            "commands_run": ["delivery_artifacts.commands_run[].command", "delivery_artifacts.commands_run[].status"],
            "test_outputs": ["delivery_artifacts.test_outputs[].name", "delivery_artifacts.test_outputs[].status"],
            "screenshots_for_ui_change": ["delivery_artifacts.screenshots[].artifact_ref 或 artifact_url"],
            "review_checklist": ["delivery_artifacts.review_checklist[].item", "delivery_artifacts.review_checklist[].status"],
            "passed_test_output": ["delivery_artifacts.test_outputs[].status=passed"],
        }.get(key, [key])

    def _artifact_review_action_history(self, events: list[dict[str, Any]]) -> list[dict[str, Any]]:
        history: list[dict[str, Any]] = []
        for event in events:
            event_type = self._event_type(event)
            if event_type not in {"artifact_review_requested", "artifact_fix_recorded", "artifact_recheck_ready"}:
                continue
            payload = self._payload(event.get("artifact_review_action")) or self._payload(event.get("payload"))
            history.append(
                {
                    "eventId": str(event.get("id") or ""),
                    "eventType": event_type,
                    "action": payload.get("action"),
                    "summary": event.get("summary"),
                    "reviewItemIds": list(payload.get("review_item_ids") or []),
                    "occurredAt": event.get("occurred_at"),
                }
            )
        return history[:12]

    def _artifact_review_closure(
        self,
        *,
        delivery_artifacts: dict[str, Any],
        events: list[dict[str, Any]],
        open_items: list[dict[str, Any]],
    ) -> dict[str, Any]:
        open_by_id = {str(item.get("id") or ""): item for item in open_items if str(item.get("id") or "")}
        tracked: dict[str, dict[str, Any]] = {}
        package_ids: set[str] = set()
        for event in events:
            if self._event_type(event) != "artifact_rework_package_created":
                continue
            payload = self._payload(event.get("artifact_rework_package")) or self._payload(event.get("payload"))
            package_id = str(payload.get("prompt_package_id") or event.get("prompt_package_id") or "")
            if package_id:
                package_ids.add(package_id)
            review_items = [item for item in list(payload.get("review_items") or []) if isinstance(item, dict)]
            review_item_ids = [str(item) for item in list(payload.get("review_item_ids") or []) if str(item)]
            for item_id in review_item_ids:
                item = next((candidate for candidate in review_items if str(candidate.get("id") or "") == item_id), None)
                if item is None:
                    item = {"id": item_id, "title": item_id, "summary": ""}
                record = tracked.setdefault(
                    item_id,
                    {
                        "id": item_id,
                        "title": item.get("title") or item_id,
                        "kind": item.get("kind"),
                        "severity": item.get("severity"),
                        "expectedEvidence": list(item.get("expectedEvidence") or []),
                        "packageIds": [],
                        "firstRequestedAt": event.get("occurred_at"),
                    },
                )
                if package_id and package_id not in record["packageIds"]:
                    record["packageIds"].append(package_id)

        closure_items: list[dict[str, Any]] = []
        for item_id, record in tracked.items():
            current_item = open_by_id.get(item_id)
            resolved = current_item is None
            closure_items.append(
                {
                    **record,
                    "status": "resolved" if resolved else "open",
                    "summary": "已被最新交付证据覆盖。" if resolved else str(current_item.get("summary") or "仍未关闭。"),
                    "currentReviewItem": current_item,
                    "evidenceRefs": self._artifact_review_evidence_refs(record, delivery_artifacts=delivery_artifacts),
                }
            )

        resolved_count = len([item for item in closure_items if item.get("status") == "resolved"])
        open_count = len(closure_items) - resolved_count
        if not closure_items:
            status = "none"
            summary = "尚未生成返工包。"
        elif open_count == 0:
            status = "all_resolved"
            summary = f"返工包中的 {resolved_count} 个审查项已被最新证据覆盖。"
        elif resolved_count:
            status = "partial"
            summary = f"返工审查项已关闭 {resolved_count} 个，仍有 {open_count} 个待处理。"
        else:
            status = "open"
            summary = f"返工包中的 {open_count} 个审查项仍未关闭。"
        return {
            "status": status,
            "summary": summary,
            "resolvedCount": resolved_count,
            "openCount": open_count,
            "items": closure_items[:30],
            "timeline": self._artifact_rework_timeline(events=events, package_ids=package_ids),
        }

    def _artifact_rework_timeline(self, *, events: list[dict[str, Any]], package_ids: set[str]) -> list[dict[str, Any]]:
        timeline: list[dict[str, Any]] = []
        for event in events:
            event_type = self._event_type(event)
            payload = (
                self._payload(event.get("artifact_rework_package"))
                or self._payload(event.get("runtime_execution_package_pull"))
                or self._payload(event.get("payload"))
            )
            package_id = str(payload.get("prompt_package_id") or event.get("prompt_package_id") or "")
            if event_type == "artifact_rework_package_created":
                kind = "package_created"
            elif event_type == "execution_package_pulled" and package_id in package_ids:
                kind = "package_pulled"
            elif package_id in package_ids and self._event_delivery_artifacts(event):
                kind = "result_returned"
            else:
                continue
            timeline.append(
                {
                    "eventId": str(event.get("id") or ""),
                    "eventType": event_type,
                    "kind": kind,
                    "promptPackageId": package_id or None,
                    "summary": event.get("summary"),
                    "source": event.get("source"),
                    "occurredAt": event.get("occurred_at"),
                }
            )
        return timeline[:12]

    def _artifact_review_evidence_refs(
        self,
        item: dict[str, Any],
        *,
        delivery_artifacts: dict[str, Any],
    ) -> list[dict[str, Any]]:
        item_id = str(item.get("id") or "")
        kind = str(item.get("kind") or "")
        evidence: list[dict[str, Any]] = []
        changed_files = [candidate for candidate in list(delivery_artifacts.get("changed_files") or []) if isinstance(candidate, dict)]
        commands = self._latest_artifact_items_by_key(
            [candidate for candidate in list(delivery_artifacts.get("commands_run") or []) if isinstance(candidate, dict)],
            keys=("command",),
        )
        tests = self._latest_artifact_items_by_key(
            [candidate for candidate in list(delivery_artifacts.get("test_outputs") or []) if isinstance(candidate, dict)],
            keys=("name", "command"),
        )
        screenshots = [candidate for candidate in list(delivery_artifacts.get("screenshots") or []) if isinstance(candidate, dict)]
        checklist = self._latest_artifact_items_by_key(
            [candidate for candidate in list(delivery_artifacts.get("review_checklist") or []) if isinstance(candidate, dict)],
            keys=("item",),
        )

        if item_id == "missing_changed_files" or kind == "changed_files":
            evidence.extend(
                {
                    "kind": "changed_file",
                    "label": str(candidate.get("path") or "changed file"),
                    "summary": candidate.get("summary"),
                    "status": candidate.get("change_type") or "changed",
                }
                for candidate in changed_files[:6]
            )
        if item_id == "missing_ui_screenshots" or kind == "screenshots_for_ui_change":
            evidence.extend(
                {
                    "kind": "screenshot",
                    "label": str(candidate.get("label") or candidate.get("artifact_ref") or candidate.get("artifact_url") or "UI screenshot"),
                    "summary": candidate.get("summary") or candidate.get("artifact_url"),
                    "status": "attached",
                    "artifactRef": candidate.get("artifact_ref"),
                    "artifactUrl": candidate.get("artifact_url"),
                }
                for candidate in screenshots[:6]
            )
        if item_id in {"missing_commands_run", "failed_command_0"} or "command" in item_id or kind in {"commands_run", "fix_failed_command"}:
            evidence.extend(
                {
                    "kind": "command",
                    "label": str(candidate.get("command") or "command"),
                    "summary": candidate.get("summary"),
                    "status": candidate.get("status") or "unknown",
                }
                for candidate in commands[:6]
                if str(candidate.get("status") or "").lower() == "passed"
            )
        if item_id in {"missing_test_outputs", "missing_passed_test_output"} or "test" in item_id or kind in {"test_outputs", "fix_failed_test"}:
            evidence.extend(
                {
                    "kind": "test",
                    "label": str(candidate.get("name") or candidate.get("command") or "test"),
                    "summary": candidate.get("summary") or candidate.get("command"),
                    "status": candidate.get("status") or "unknown",
                }
                for candidate in tests[:6]
                if str(candidate.get("status") or "").lower() == "passed"
            )
        if item_id.startswith("review_checklist") or kind in {"review_checklist", "resolve_review_checklist"}:
            evidence.extend(
                {
                    "kind": "review_checklist",
                    "label": str(candidate.get("item") or "review checklist"),
                    "summary": candidate.get("summary"),
                    "status": candidate.get("status") or "unknown",
                }
                for candidate in checklist[:6]
                if str(candidate.get("status") or "").lower() == "passed"
            )
        return evidence[:10]

    @staticmethod
    def _artifact_review_summary(*, status: str, item_count: int) -> str:
        if status == "clear":
            return "交付证据审查已清空，当前证据满足发布前检查。"
        if status == "blocked":
            return f"交付证据存在 {item_count} 个阻塞项，需要修复后重新回写。"
        if status == "open":
            return f"交付证据还有 {item_count} 个待补齐项。"
        if status == "empty":
            return "尚未形成交付证据审查项。"
        return "交付证据审查待处理。"

    @staticmethod
    def _artifact_review_next_actions(*, status: str, items: list[dict[str, Any]]) -> list[str]:
        if status == "clear":
            return ["继续进入验收、Project Memory 写回或发布验证。"]
        if not items:
            return ["等待运行端回写结构化交付证据。"]
        actions = [str(item.get("recommendedAction") or "") for item in items if item.get("recommendedAction")]
        return list(dict.fromkeys(actions))[:4]

    @staticmethod
    def _artifact_review_action_summary(*, action: str, items: list[dict[str, Any]], note: Optional[str]) -> str:
        labels = {
            "request_fix": "已要求补齐/修复交付证据",
            "record_fix": "已记录交付证据修复说明",
            "mark_recheck_ready": "交付证据已标记为待重新检查",
        }
        prefix = labels.get(action, "已记录交付证据审查动作")
        item_label = "、".join(str(item.get("title") or item.get("id") or "") for item in items[:3] if item)
        suffix = f"：{item_label}" if item_label else ""
        note_suffix = f"；{note}" if note else ""
        return f"{prefix}{suffix}{note_suffix}"

    @staticmethod
    def _artifact_rework_package_summary(*, task: Task, items: list[dict[str, Any]], note: Optional[str]) -> str:
        titles = "、".join(str(item.get("title") or item.get("id") or "") for item in items[:3] if item)
        suffix = f"：{titles}" if titles else ""
        note_suffix = f"；{note}" if note else ""
        return f"交付证据返工执行包已生成：{task.title}{suffix}{note_suffix}"

    @staticmethod
    def _artifact_path_is_ui(path: str) -> bool:
        lowered = path.lower()
        return any(token in lowered for token in ("frontend/", ".tsx", ".jsx", ".css", "tailwind", "ui/", "components/", "pages/"))

    @staticmethod
    def _artifact_missing_label(key: str) -> str:
        return {
            "delivery_artifacts": "缺少结构化交付证据包。",
            "changed_files": "缺少 changed files。",
            "commands_run": "缺少 commands run。",
            "test_outputs": "缺少 test outputs。",
            "screenshots_for_ui_change": "检测到前端/UI 改动但缺少截图证据。",
            "review_checklist": "高风险执行缺少 review checklist。",
            "passed_test_output": "缺少通过状态的测试输出。",
        }.get(key, key)

    def _unresolved_failed_evidence_events(
        self,
        events: list[dict[str, Any]],
        *,
        artifact_completeness: Optional[dict[str, Any]] = None,
        artifact_review: Optional[dict[str, Any]] = None,
    ) -> list[dict[str, Any]]:
        failed_events = [event for event in events if self._is_failed_evidence_event(event)]
        if not failed_events:
            return []
        artifact_status = str((artifact_completeness or {}).get("status") or "")
        closure = self._payload((artifact_review or {}).get("closure") if artifact_review else {})
        closure_status = str(closure.get("status") or "")
        if artifact_status == "ready" and closure_status in {"", "none", "all_resolved"}:
            return [event for event in failed_events if self._event_type(event) in {"failed", "blocked", "test_failed"}]
        return failed_events

    def _execution_evidence_proof(
        self,
        *,
        events: list[dict[str, Any]],
        artifact_completeness: Optional[dict[str, Any]] = None,
        artifact_review: Optional[dict[str, Any]] = None,
        evidence_event_ids: Optional[list[str]] = None,
    ) -> dict[str, Any]:
        selected_ids = {str(item) for item in list(evidence_event_ids or []) if str(item)}
        scoped_events = [event for event in events if not selected_ids or str(event.get("id") or "") in selected_ids]
        closure = self._payload((artifact_review or {}).get("closure") if artifact_review else {})
        closure_items = [item for item in list(closure.get("items") or []) if isinstance(item, dict)]
        closure_evidence_refs = [
            ref
            for item in closure_items
            for ref in list(item.get("evidenceRefs") or [])
            if isinstance(ref, dict)
        ]
        unresolved = self._unresolved_failed_evidence_events(
            scoped_events,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        artifact_status = str((artifact_completeness or {}).get("status") or "empty")
        closure_status = str(closure.get("status") or "none")
        has_tests = any(self._event_type(event) in {"test_passed", "checks_passed"} for event in scoped_events)
        status = "ready" if artifact_status == "ready" and has_tests and not unresolved else "incomplete"
        if unresolved:
            status = "blocked"
        return {
            "status": status,
            "summary": {
                "ready": "执行证据、交付证据和返工闭环已满足复检输入。",
                "blocked": "仍存在未关闭的失败或阻塞证据。",
                "incomplete": "复检证据链尚未完整。",
            }.get(status, "复检证据待确认。"),
            "selectedEvidenceEventIds": sorted(selected_ids),
            "testEvidenceCount": len([event for event in scoped_events if self._event_type(event) in {"test_passed", "checks_passed"}]),
            "artifactStatus": artifact_status,
            "artifactClosureStatus": closure_status,
            "resolvedReworkItems": int(closure.get("resolvedCount") or 0),
            "openReworkItems": int(closure.get("openCount") or 0),
            "unresolvedFailureEventIds": [str(event.get("id") or "") for event in unresolved[:10]],
            "evidenceRefs": closure_evidence_refs[:20],
            "timeline": list(closure.get("timeline") or [])[:12],
        }

    def _release_readiness_proof(
        self,
        *,
        events: list[dict[str, Any]],
        acceptance: Optional[dict[str, Any]],
        artifact_completeness: Optional[dict[str, Any]],
        artifact_review: Optional[dict[str, Any]],
    ) -> dict[str, Any]:
        base = self._execution_evidence_proof(
            events=events,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        has_memory_writeback = any(self._event_type(event) == "execution_memory_writeback" for event in events)
        can_enter_release_gate = bool((acceptance or {}).get("canEnterReleaseGate"))
        if base.get("status") == "blocked":
            status = "blocked"
        elif base.get("status") == "ready" and can_enter_release_gate and has_memory_writeback:
            status = "ready"
        elif base.get("status") == "ready" and can_enter_release_gate:
            status = "awaiting_memory_writeback"
        elif base.get("status") == "ready":
            status = "awaiting_acceptance"
        else:
            status = str(base.get("status") or "incomplete")
        return {
            **base,
            "status": status,
            "acceptanceStatus": (acceptance or {}).get("status"),
            "canEnterReleaseGate": can_enter_release_gate,
            "memoryWritebackDone": has_memory_writeback,
            "nextRequiredHumanAction": {
                "ready": "负责人可进行最终发布复检；PMFlow 不执行发布。",
                "awaiting_memory_writeback": "先在 Execution Bridge 写回项目记忆。",
                "awaiting_acceptance": "先完成用户验收或技术审查。",
                "blocked": "先修复未关闭失败证据。",
            }.get(status, "补齐执行、测试和交付证据。"),
        }

    def _build_release_gate(
        self,
        *,
        events: list[dict[str, Any]],
        deliveries: list[dict[str, Any]],
        acceptance: Optional[dict[str, Any]] = None,
        artifact_completeness: Optional[dict[str, Any]] = None,
        artifact_review: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        evidence_events = [event for event in events if self._is_release_evidence_event(event)]
        release_readiness_proof = self._release_readiness_proof(
            events=evidence_events,
            acceptance=acceptance,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        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": [],
                "releaseReadinessProof": release_readiness_proof,
                "updatedAt": None,
            }

        failed_events = self._unresolved_failed_evidence_events(
            evidence_events,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        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, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }
        artifact_status = str((artifact_completeness or {}).get("status") or "empty")
        artifact_reasons = list((artifact_completeness or {}).get("blockingReasons") or [])
        if artifact_status == "blocked":
            return {
                "status": "blocked",
                "title": "发布阻塞",
                "summary": "结构化交付证据中存在失败或阻塞项，发布门禁已关闭。",
                "blockingReasons": artifact_reasons,
                "nextActions": ["先修复失败的命令、测试或审查项，再重新回写执行结果。"],
                "checks": self._release_gate_checks(evidence_events, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "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)
        has_memory_writeback = any(self._event_type(event) == "execution_memory_writeback" for event in evidence_events)
        acceptance_status = str((acceptance or {}).get("status") or "")
        can_enter_release_gate = bool((acceptance or {}).get("canEnterReleaseGate"))
        if has_test_passed and not can_enter_release_gate:
            return {
                "status": "needs_confirmation",
                "title": "等待验收或技术审查",
                "summary": "测试证据已回流，但执行结果尚未通过用户验收或必要技术审查，发布门禁保持待确认。",
                "blockingReasons": list((acceptance or {}).get("blockingReasons") or []),
                "nextActions": ["先在 Execution Bridge 完成用户验收或技术审查。"],
                "checks": self._release_gate_checks(evidence_events, artifact_completeness=artifact_completeness, artifact_review=artifact_review)
                + [
                    {
                        "key": "acceptance",
                        "label": "验收/审查",
                        "status": "missing",
                        "detail": acceptance_status or "pending_acceptance",
                    }
                ],
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }
        if has_test_passed and can_enter_release_gate and not has_memory_writeback:
            return {
                "status": "needs_confirmation",
                "title": "等待写回项目记忆",
                "summary": "执行结果已通过验收，但尚未把最终证据写回项目记忆，发布门禁保持待确认。",
                "blockingReasons": [],
                "nextActions": ["先在 Execution Bridge 执行“写回项目记忆”。"],
                "checks": self._release_gate_checks(evidence_events, artifact_completeness=artifact_completeness, artifact_review=artifact_review)
                + [
                    {
                        "key": "memory_writeback",
                        "label": "项目记忆写回",
                        "status": "missing",
                        "detail": "execution_memory_writeback_required",
                    }
                ],
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }
        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, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }

        if has_test_passed and has_synced and has_memory_writeback and artifact_status != "ready":
            return {
                "status": "needs_confirmation",
                "title": "等待补齐交付证据",
                "summary": "执行结果已通过验收并写回项目记忆，但结构化交付证据仍不完整，发布门禁保持待确认。",
                "blockingReasons": artifact_reasons,
                "nextActions": ["补齐 changed files、commands run、test outputs；如有 UI 改动补充截图证据。"],
                "checks": self._release_gate_checks(evidence_events, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
            }

        if has_test_passed and has_synced and has_memory_writeback:
            return {
                "status": "ready",
                "title": "可发布",
                "summary": "已有 PR/合并、通过测试、用户验收与项目记忆写回证据，发布门禁给出可发布判断。",
                "blockingReasons": [],
                "nextActions": ["由负责人进行最终人工确认；PMFlow 不自动执行发布命令。"],
                "checks": self._release_gate_checks(evidence_events, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "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, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
                "evidence": self._release_gate_evidence(evidence_events),
                "releaseReadinessProof": release_readiness_proof,
                "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, artifact_completeness=artifact_completeness, artifact_review=artifact_review),
            "evidence": self._release_gate_evidence(evidence_events),
            "releaseReadinessProof": release_readiness_proof,
            "updatedAt": str(evidence_events[0].get("occurred_at") or "") or None,
        }

    def _release_gate_checks(
        self,
        events: list[dict[str, Any]],
        *,
        artifact_completeness: Optional[dict[str, Any]] = None,
        artifact_review: Optional[dict[str, Any]] = None,
    ) -> list[dict[str, Any]]:
        unresolved_failures = self._unresolved_failed_evidence_events(
            events,
            artifact_completeness=artifact_completeness,
            artifact_review=artifact_review,
        )
        checks = [
            {
                "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 unresolved_failures else "passed",
                "detail": self._first_event_detail(events, {"failed", "blocked", "test_failed"}) or "未发现失败回写",
            },
            {
                "key": "memory_writeback",
                "label": "项目记忆写回",
                "status": "passed" if any(self._event_type(event) == "execution_memory_writeback" for event in events) else "missing",
                "detail": self._first_event_detail(events, {"execution_memory_writeback"}) or "等待执行结果写回项目记忆",
            },
        ]
        if artifact_completeness is not None:
            status = str(artifact_completeness.get("status") or "empty")
            checks.append(
                {
                    "key": "delivery_artifacts",
                    "label": "交付证据包",
                    "status": "passed" if status == "ready" else ("failed" if status == "blocked" else "missing"),
                    "detail": artifact_completeness.get("summary") or "等待结构化交付证据",
                }
            )
        return checks

    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_delivery_artifacts(self, event: dict[str, Any]) -> dict[str, Any]:
        direct = event.get("delivery_artifacts")
        if isinstance(direct, dict):
            return direct
        payload = self._payload(event.get("payload"))
        payload_artifacts = payload.get("delivery_artifacts")
        if isinstance(payload_artifacts, dict):
            return payload_artifacts
        execution_result = self._payload(payload.get("execution_result"))
        result_artifacts = execution_result.get("delivery_artifacts")
        return result_artifacts if isinstance(result_artifacts, dict) else {}

    def _event_text(self, event: dict[str, Any]) -> str:
        payload = self._payload(event.get("payload"))
        execution_result = self._payload(payload.get("execution_result"))
        artifacts = self._event_delivery_artifacts(event)
        return " ".join(
            [
                self._event_type(event),
                str(event.get("summary") or ""),
                str(event.get("workflow_state") or ""),
                str(event.get("derived_task_status") or ""),
                str(payload.get("test_summary") or ""),
                str(payload.get("blocker_summary") or ""),
                str(execution_result.get("testSummary") or ""),
                str(execution_result.get("blockerSummary") or ""),
                self._short_text(artifacts),
            ]
        ).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",
            "execution_memory_writeback",
        } or bool(event.get("branch_name") or event.get("commit_sha") or event.get("pr_number"))

    def _is_execution_result_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",
        } 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 _release_validation_status(self, *, gate_status: str, decision: str) -> str:
        if decision == "keep_blocked" or gate_status == "blocked":
            return "blocked"
        if gate_status == "ready":
            return "ready"
        return "needs_confirmation"

    def _release_validation_score(self, status: str) -> float:
        if status == "ready":
            return 1.0
        if status == "needs_confirmation":
            return 0.6
        return 0.0

    def _release_validation_metrics(self, release_gate: dict[str, Any]) -> dict[str, Any]:
        checks = list(release_gate.get("checks") or [])
        evidence = list(release_gate.get("evidence") or [])
        return {
            "releaseGateStatus": release_gate.get("status"),
            "evidenceCount": len(evidence),
            "passedCheckCount": len([item for item in checks if isinstance(item, dict) and item.get("status") == "passed"]),
            "failedCheckCount": len([item for item in checks if isinstance(item, dict) and item.get("status") == "failed"]),
            "missingCheckCount": len([item for item in checks if isinstance(item, dict) and item.get("status") == "missing"]),
        }

    def _release_validation_risks(self, release_gate: dict[str, Any], *, decision: str) -> list[dict[str, Any]]:
        risks = [
            {"label": reason, "status": "blocked", "source": "release_gate_blocking_reason"}
            for reason in list(release_gate.get("blockingReasons") or [])
        ]
        for check in list(release_gate.get("checks") or []):
            if not isinstance(check, dict) or check.get("status") == "passed":
                continue
            risks.append(
                {
                    "label": check.get("label") or check.get("key") or "发布检查项",
                    "status": check.get("status"),
                    "detail": check.get("detail"),
                    "source": "release_gate_check",
                }
            )
        if decision == "keep_blocked" and not risks:
            risks.append({"label": "负责人选择保持阻塞，等待补充发布前确认。", "status": "blocked", "source": "human_decision"})
        return risks

    def _release_validation_rollback_plan(self, release_gate: dict[str, Any]) -> dict[str, Any]:
        return {
            "deploymentExecuted": False,
            "requiredBeforeDeploy": [
                "完成人工发布确认。",
                "确认回滚方案和发布范围。",
            ],
            "nextActions": list(release_gate.get("nextActions") or []),
        }

    def _serialize_release_validation_run(self, run: ReleaseValidationRun) -> dict[str, Any]:
        summary = self._payload(run.validation_summary)
        return {
            "id": str(run.id),
            "projectId": str(run.project_id),
            "charterId": None if run.charter_id is None else str(run.charter_id),
            "releaseNodeId": None if run.release_node_id is None else str(run.release_node_id),
            "status": run.status,
            "score": self._jsonable(run.score),
            "keyMetrics": run.key_metrics or {},
            "riskItems": run.risk_items or [],
            "rollbackPlan": run.rollback_plan or {},
            "validationSummary": summary,
            "createdBy": summary.get("createdBy"),
            "createdAt": run.created_at.isoformat() if run.created_at else summary.get("createdAt"),
            "updatedAt": run.updated_at.isoformat() if run.updated_at else None,
        }

    def _serialize_feedback_ingest(self, ingest: FeedbackIngest) -> dict[str, Any]:
        return {
            "id": str(ingest.id),
            "projectId": str(ingest.project_id),
            "charterId": None if ingest.charter_id is None else str(ingest.charter_id),
            "sourceType": ingest.source_type,
            "title": ingest.title,
            "content": ingest.content,
            "metrics": dict(ingest.metrics or {}),
            "samples": list(ingest.samples or []),
            "status": ingest.status,
            "createdAt": ingest.created_at.isoformat() if ingest.created_at else None,
            "updatedAt": ingest.updated_at.isoformat() if ingest.updated_at else None,
        }

    def _serialize_iteration_recommendation(
        self,
        recommendation: IterationRecommendation,
        *,
        quality: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        summary = dict(recommendation.recommendation_summary or {})
        change_set_id = summary.get("changeSetId")
        normalized_change_set_id = None if change_set_id is None or str(change_set_id).strip() == "" else str(change_set_id)
        quality_payload = dict(quality or {})
        return {
            "id": str(recommendation.id),
            "projectId": str(recommendation.project_id),
            "charterId": None if recommendation.charter_id is None else str(recommendation.charter_id),
            "feedbackIngestId": None if recommendation.feedback_ingest_id is None else str(recommendation.feedback_ingest_id),
            "status": recommendation.status,
            "priority": recommendation.priority,
            "recommendationSummary": summary,
            "proposedCharterPatch": dict(recommendation.proposed_charter_patch or {}),
            "proposedTasks": list(recommendation.proposed_tasks or []),
            "changeSetCreated": bool(summary.get("changeSetCreated") or normalized_change_set_id),
            "changeSetId": normalized_change_set_id,
            "groupKey": quality_payload.get("groupKey") or summary.get("groupKey"),
            "duplicateOf": quality_payload.get("duplicateOf"),
            "mergedCount": int(quality_payload.get("mergedCount") or 1),
            "impactLevel": quality_payload.get("impactLevel") or "medium",
            "riskLevel": quality_payload.get("riskLevel") or "medium",
            "suggestedChangeTypes": list(quality_payload.get("suggestedChangeTypes") or ["Requirement", "Decision"]),
            "reviewPriority": quality_payload.get("reviewPriority") or "p2",
            "quality": quality_payload,
            "createdAt": recommendation.created_at.isoformat() if recommendation.created_at else None,
            "updatedAt": recommendation.updated_at.isoformat() if recommendation.updated_at else None,
        }

    def _recommendation_quality_profile(
        self,
        *,
        recommendation: IterationRecommendation,
        feedback: Optional[FeedbackIngest],
    ) -> dict[str, Any]:
        summary = dict(recommendation.recommendation_summary or {})
        patch = dict(recommendation.proposed_charter_patch or {})
        focus_areas = patch.get("focus_areas") or patch.get("focusAreas") or summary.get("focus_areas") or summary.get("focusAreas") or []
        if not isinstance(focus_areas, list):
            focus_areas = [str(focus_areas)]
        focus_areas = [str(item).strip() for item in focus_areas if str(item).strip()]
        group_seed = focus_areas[0] if focus_areas else (feedback.title if feedback is not None and feedback.title else f"recommendation-{recommendation.id}")
        group_key = self._normalize_group_key(group_seed)
        priority = str(recommendation.priority or "medium").lower()
        proposed_tasks = [item for item in list(recommendation.proposed_tasks or []) if isinstance(item, dict)]
        metrics = dict(feedback.metrics or {}) if feedback is not None else {}
        negative_feedback = self._safe_number(metrics.get("negative_feedback_count") or metrics.get("negativeFeedbackCount"))
        suggested_change_types = ["Requirement", "Decision"]
        if proposed_tasks:
            suggested_change_types.append("Task")
        impact_level = "high" if priority in {"high", "critical"} or negative_feedback >= 3 or len(proposed_tasks) >= 3 else "medium"
        if negative_feedback <= 0 and not proposed_tasks and priority == "low":
            impact_level = "low"
        risk_level = "high" if priority in {"high", "critical"} else ("medium" if proposed_tasks or impact_level == "medium" else "low")
        if risk_level == "high":
            suggested_change_types.append("Risk")
        review_priority = "p0" if risk_level == "high" and impact_level == "high" else ("p1" if impact_level == "high" or risk_level == "medium" else "p2")
        return {
            "groupKey": group_key,
            "groupLabel": group_seed,
            "duplicateOf": None,
            "mergedCount": 1,
            "impactLevel": impact_level,
            "riskLevel": risk_level,
            "suggestedChangeTypes": self._unique_strings(suggested_change_types),
            "reviewPriority": review_priority,
            "rationale": self._recommendation_quality_rationale(
                focus_areas=focus_areas,
                negative_feedback=negative_feedback,
                task_count=len(proposed_tasks),
                priority=priority,
            ),
        }

    def _finalize_recommendation_quality(
        self,
        *,
        recommendations: list[IterationRecommendation],
        quality_by_id: dict[int, dict[str, Any]],
    ) -> dict[int, dict[str, Any]]:
        groups: dict[str, list[int]] = {}
        for recommendation in recommendations:
            quality = quality_by_id.get(int(recommendation.id)) or {}
            groups.setdefault(str(quality.get("groupKey") or f"recommendation-{recommendation.id}"), []).append(int(recommendation.id))
        finalized: dict[int, dict[str, Any]] = {}
        for group_key, ids in groups.items():
            first_id = str(ids[0])
            merged_count = len(ids)
            for rec_id in ids:
                quality = dict(quality_by_id.get(rec_id) or {})
                quality["groupKey"] = group_key
                quality["mergedCount"] = merged_count
                quality["duplicateOf"] = None if str(rec_id) == first_id else first_id
                finalized[rec_id] = quality
        return finalized

    def _serialize_recommendation_groups(self, recommendations: list[dict[str, Any]]) -> list[dict[str, Any]]:
        groups: dict[str, list[dict[str, Any]]] = {}
        for recommendation in recommendations:
            group_key = str(recommendation.get("groupKey") or f"recommendation-{recommendation.get('id')}")
            groups.setdefault(group_key, []).append(recommendation)
        serialized: list[dict[str, Any]] = []
        priority_order = {"p0": 0, "p1": 1, "p2": 2}
        level_order = {"high": 0, "medium": 1, "low": 2}
        for group_key, rows in groups.items():
            quality = dict(rows[0].get("quality") or {})
            change_set_ids = self._unique_strings([row.get("changeSetId") for row in rows if row.get("changeSetId")])
            suggested_types = self._unique_strings(
                item for row in rows for item in list(row.get("suggestedChangeTypes") or [])
            )
            serialized.append(
                {
                    "groupKey": group_key,
                    "title": f"{quality.get('groupLabel') or group_key} 反馈组",
                    "recommendationIds": [str(row.get("id")) for row in rows],
                    "mergedCount": len(rows),
                    "impactLevel": quality.get("impactLevel") or "medium",
                    "riskLevel": quality.get("riskLevel") or "medium",
                    "suggestedChangeTypes": suggested_types,
                    "reviewPriority": quality.get("reviewPriority") or "p2",
                    "changeSetCreated": bool(change_set_ids),
                    "changeSetId": change_set_ids[0] if change_set_ids else None,
                    "rationale": quality.get("rationale") or "基于同组发布后反馈聚合生成。",
                }
            )
        return sorted(
            serialized,
            key=lambda row: (
                priority_order.get(str(row.get("reviewPriority") or "p2"), 9),
                level_order.get(str(row.get("riskLevel") or "medium"), 9),
                str(row.get("groupKey") or ""),
            ),
        )

    def _recommendation_quality_rationale(
        self,
        *,
        focus_areas: list[str],
        negative_feedback: float,
        task_count: int,
        priority: str,
    ) -> str:
        parts = []
        if focus_areas:
            parts.append("聚焦 " + "、".join(focus_areas[:4]))
        if negative_feedback >= 1:
            parts.append(f"负向反馈 {int(negative_feedback)} 条")
        if task_count:
            parts.append(f"建议任务 {task_count} 个")
        parts.append(f"原始优先级 {priority or 'medium'}")
        return "；".join(parts) + "。"

    def _iteration_recommendation_message(
        self,
        *,
        recommendation: IterationRecommendation,
        feedback: Optional[FeedbackIngest],
        note: Optional[str],
    ) -> str:
        summary = self._recommendation_summary_text(recommendation)
        task_titles = [
            str(item.get("title") or item.get("summary") or "").strip()
            for item in list(recommendation.proposed_tasks or [])
            if isinstance(item, dict) and str(item.get("title") or item.get("summary") or "").strip()
        ]
        lines = [
            "Release/Growth 反馈已生成下一轮待确认变更草稿。",
            f"反馈：{feedback.title or feedback.content[:80] if feedback is not None else '未绑定 feedback ingest'}",
            f"建议：{summary}",
        ]
        if task_titles:
            lines.append("建议任务：" + "；".join(task_titles[:5]))
        if note:
            lines.append(f"人工备注：{note.strip()}")
        lines.append("此消息只作为 Change Review 证据，不自动写入 Project Memory。")
        return "\n".join(lines)

    def _iteration_change_set_headline(
        self,
        *,
        recommendation: IterationRecommendation,
        feedback: Optional[FeedbackIngest],
    ) -> str:
        source = feedback.title if feedback is not None and feedback.title else f"Recommendation {recommendation.id}"
        return f"下一轮反馈变更草稿：{source}"[:255]

    def _iteration_change_set_summary(
        self,
        *,
        recommendation: IterationRecommendation,
        feedback: Optional[FeedbackIngest],
    ) -> str:
        source = feedback.title or feedback.content[:80] if feedback is not None else "发布后反馈"
        return f"基于 {source} 生成下一轮 ChangeSet draft，等待 Change Review accept/edit/defer。"

    def _iteration_recommendation_excerpt(
        self,
        *,
        recommendation: IterationRecommendation,
        feedback: Optional[FeedbackIngest],
    ) -> str:
        feedback_text = "" if feedback is None else feedback.content.strip()
        return (feedback_text or self._recommendation_summary_text(recommendation))[:1200]

    def _recommendation_summary_text(self, recommendation: IterationRecommendation) -> str:
        summary = dict(recommendation.recommendation_summary or {})
        value = summary.get("summary") or summary.get("recommendation") or summary.get("title")
        if value:
            return str(value).strip()
        focus = summary.get("focus_areas") or summary.get("focusAreas")
        if isinstance(focus, list) and focus:
            return "围绕 " + "、".join(str(item) for item in focus[:4]) + " 推进下一轮迭代。"
        return "基于发布后反馈生成下一轮迭代建议。"

    def _iteration_recommendation_change_items(
        self,
        *,
        recommendation: IterationRecommendation,
        feedback: Optional[FeedbackIngest],
        project_id: int,
    ) -> list[PMFlowChangeItem]:
        patch = dict(recommendation.proposed_charter_patch or {})
        summary_text = self._recommendation_summary_text(recommendation)
        focus_areas = patch.get("focus_areas") or patch.get("focusAreas") or dict(recommendation.recommendation_summary or {}).get("focus_areas") or []
        success_criteria = patch.get("updated_success_criteria") or patch.get("successCriteria") or []
        risk_level = "high" if str(recommendation.priority or "").lower() in {"high", "critical"} else "medium"
        items: list[PMFlowChangeItem] = [
            PMFlowChangeItem(
                tenant_id=self.tenant_id,
                change_set_id=0,
                project_id=int(project_id),
                group="new_object",
                object_type="Requirement",
                object_id=None,
                title="下一轮迭代方向",
                summary=summary_text,
                diff="根据发布后反馈补充下一轮需求方向与成功标准。",
                payload={
                    "title": "下一轮迭代方向",
                    "summary": summary_text,
                    "stage": "planning",
                    "status": "draft",
                    "source": "iteration_recommendation",
                    "recommendationId": str(recommendation.id),
                    "feedbackIngestId": None if feedback is None else str(feedback.id),
                    "focusAreas": list(focus_areas) if isinstance(focus_areas, list) else [str(focus_areas)],
                    "acceptanceCriteria": list(success_criteria) if isinstance(success_criteria, list) else [str(success_criteria)],
                },
                edited_payload=None,
                confidence=0.76,
                risk_level=risk_level,
                status="pending_review",
            ),
            PMFlowChangeItem(
                tenant_id=self.tenant_id,
                change_set_id=0,
                project_id=int(project_id),
                group="decision",
                object_type="Decision",
                object_id=None,
                title="确认下一轮反馈驱动迭代方向",
                summary=f"确认是否将 recommendation {recommendation.id} 纳入下一轮 Change Review。",
                diff="将发布后反馈从 Release/Growth 回流到 Change Review，而不是直接写入项目记忆。",
                payload={
                    "title": "确认下一轮反馈驱动迭代方向",
                    "summary": summary_text,
                    "stage": "growth",
                    "owner": "PMFlow",
                    "source": "iteration_recommendation",
                    "recommendationId": str(recommendation.id),
                },
                edited_payload=None,
                confidence=0.72,
                risk_level="medium",
                status="pending_review",
            ),
        ]

        for raw_task in list(recommendation.proposed_tasks or [])[:4]:
            if not isinstance(raw_task, dict):
                continue
            title = str(raw_task.get("title") or raw_task.get("summary") or "").strip()
            if not title:
                continue
            task_summary = str(raw_task.get("summary") or raw_task.get("description") or title).strip()
            items.append(
                PMFlowChangeItem(
                    tenant_id=self.tenant_id,
                    change_set_id=0,
                    project_id=int(project_id),
                    group="new_object",
                    object_type="Task",
                    object_id=None,
                    title=title[:255],
                    summary=task_summary,
                    diff="新增下一轮可审查任务草稿，待 Change Review 决定是否接受。",
                    payload={
                        "title": title,
                        "description": task_summary,
                        "summary": task_summary,
                        "stage": "development",
                        "status": "draft",
                        "priority": raw_task.get("priority") or recommendation.priority,
                        "source": "iteration_recommendation",
                        "recommendationId": str(recommendation.id),
                    },
                    edited_payload=None,
                    confidence=0.7,
                    risk_level="medium",
                    status="pending_review",
                )
            )

        if risk_level == "high":
            items.append(
                PMFlowChangeItem(
                    tenant_id=self.tenant_id,
                    change_set_id=0,
                    project_id=int(project_id),
                    group="risk",
                    object_type="Risk",
                    object_id=None,
                    title="发布后反馈高优先级风险",
                    summary=summary_text,
                    diff="将高优先级反馈沉淀为风险项，避免直接跳过人工审查。",
                    payload={
                        "title": "发布后反馈高优先级风险",
                        "summary": summary_text,
                        "stage": "planning",
                        "level": "high",
                        "mitigation": "在 Change Review 中确认处理策略后再进入 Project Memory。",
                        "source": "iteration_recommendation",
                        "recommendationId": str(recommendation.id),
                    },
                    edited_payload=None,
                    confidence=0.74,
                    risk_level="high",
                    status="pending_review",
                )
            )
        return items

    def _iteration_group_message(
        self,
        *,
        recommendations: list[IterationRecommendation],
        feedback_by_id: dict[int, FeedbackIngest],
        quality_by_id: dict[int, dict[str, Any]],
        note: Optional[str],
    ) -> str:
        first_quality = next(iter(quality_by_id.values()), {})
        summaries = [self._recommendation_summary_text(item) for item in recommendations]
        feedback_titles = [
            feedback_by_id[int(item.feedback_ingest_id)].title or feedback_by_id[int(item.feedback_ingest_id)].content[:60]
            for item in recommendations
            if item.feedback_ingest_id is not None and int(item.feedback_ingest_id) in feedback_by_id
        ]
        lines = [
            "Release/Growth 同组反馈已生成下一轮待确认变更草稿。",
            f"建议组：{first_quality.get('groupLabel') or first_quality.get('groupKey') or 'feedback'}",
            f"合并建议：{len(recommendations)} 条；风险：{first_quality.get('riskLevel') or 'medium'}；影响：{first_quality.get('impactLevel') or 'medium'}。",
        ]
        if feedback_titles:
            lines.append("反馈来源：" + "；".join(feedback_titles[:5]))
        lines.append("建议摘要：" + "；".join(summaries[:4]))
        if note:
            lines.append(f"人工备注：{note.strip()}")
        lines.append("此消息只作为 Change Review 证据，不自动写入 Project Memory。")
        return "\n".join(lines)

    def _iteration_group_change_set_headline(
        self,
        *,
        recommendations: list[IterationRecommendation],
        quality_by_id: dict[int, dict[str, Any]],
    ) -> str:
        quality = next(iter(quality_by_id.values()), {})
        label = quality.get("groupLabel") or quality.get("groupKey") or "feedback"
        return f"下一轮反馈变更草稿组：{label}（{len(recommendations)} 条建议）"[:255]

    def _iteration_group_change_set_summary(
        self,
        *,
        recommendations: list[IterationRecommendation],
        quality_by_id: dict[int, dict[str, Any]],
    ) -> str:
        quality = next(iter(quality_by_id.values()), {})
        return (
            f"基于 {len(recommendations)} 条同组 iteration recommendation 生成下一轮 ChangeSet draft；"
            f"reviewPriority={quality.get('reviewPriority') or 'p2'}，risk={quality.get('riskLevel') or 'medium'}，"
            "等待 Change Review accept/edit/defer。"
        )

    def _iteration_group_excerpt(
        self,
        *,
        recommendations: list[IterationRecommendation],
        feedback_by_id: dict[int, FeedbackIngest],
    ) -> str:
        excerpts: list[str] = []
        for recommendation in recommendations:
            if recommendation.feedback_ingest_id is not None and int(recommendation.feedback_ingest_id) in feedback_by_id:
                excerpts.append(feedback_by_id[int(recommendation.feedback_ingest_id)].content.strip())
            else:
                excerpts.append(self._recommendation_summary_text(recommendation))
        return "\n".join(excerpts)[:1200]

    def _iteration_recommendation_group_change_items(
        self,
        *,
        recommendations: list[IterationRecommendation],
        feedback_by_id: dict[int, FeedbackIngest],
        quality_by_id: dict[int, dict[str, Any]],
        project_id: int,
    ) -> list[PMFlowChangeItem]:
        first_quality = next(iter(quality_by_id.values()), {})
        group_key = str(first_quality.get("groupKey") or "general-feedback")
        recommendation_ids = [str(item.id) for item in recommendations]
        feedback_ids = [str(item.feedback_ingest_id) for item in recommendations if item.feedback_ingest_id is not None]
        focus_areas = self._unique_strings(
            area
            for recommendation in recommendations
            for area in list(dict(recommendation.proposed_charter_patch or {}).get("focus_areas") or dict(recommendation.recommendation_summary or {}).get("focus_areas") or [])
        )
        summaries = [self._recommendation_summary_text(item) for item in recommendations]
        summary_text = "；".join(summaries[:4])
        risk_level = "high" if any(str(quality.get("riskLevel")) == "high" for quality in quality_by_id.values()) else "medium"
        impact_level = "high" if any(str(quality.get("impactLevel")) == "high" for quality in quality_by_id.values()) else "medium"
        base_payload = {
            "source": "iteration_recommendation_group",
            "groupKey": group_key,
            "recommendationIds": recommendation_ids,
            "feedbackIngestIds": feedback_ids,
            "quality": first_quality,
        }
        items: list[PMFlowChangeItem] = [
            PMFlowChangeItem(
                tenant_id=self.tenant_id,
                change_set_id=0,
                project_id=int(project_id),
                group="new_object",
                object_type="Requirement",
                object_id=None,
                title=f"下一轮迭代方向：{first_quality.get('groupLabel') or group_key}"[:255],
                summary=summary_text or "基于发布后反馈建议组生成下一轮迭代方向。",
                diff="合并同组发布后反馈，形成下一轮需求方向与成功标准草稿。",
                payload={
                    **base_payload,
                    "title": f"下一轮迭代方向：{first_quality.get('groupLabel') or group_key}",
                    "summary": summary_text,
                    "stage": "planning",
                    "status": "draft",
                    "focusAreas": focus_areas,
                    "acceptanceCriteria": [
                        f"本组 {len(recommendations)} 条 recommendation 在 Change Review 中完成 accept/edit/defer。",
                        "接受后再进入 Project Memory，不自动写入。",
                    ],
                },
                edited_payload=None,
                confidence=0.78,
                risk_level=risk_level,
                status="pending_review",
            ),
            PMFlowChangeItem(
                tenant_id=self.tenant_id,
                change_set_id=0,
                project_id=int(project_id),
                group="decision",
                object_type="Decision",
                object_id=None,
                title="确认同组反馈进入下一轮迭代",
                summary=f"确认 {group_key} 反馈组是否作为下一轮迭代输入；影响级别 {impact_level}，风险级别 {risk_level}。",
                diff="批量生成只进入待确认中心，不直接提交项目记忆。",
                payload={
                    **base_payload,
                    "title": "确认同组反馈进入下一轮迭代",
                    "summary": summary_text,
                    "stage": "growth",
                    "owner": "PMFlow",
                },
                edited_payload=None,
                confidence=0.74,
                risk_level=risk_level,
                status="pending_review",
            ),
        ]

        seen_task_titles: set[str] = set()
        for recommendation in recommendations:
            for raw_task in list(recommendation.proposed_tasks or []):
                if not isinstance(raw_task, dict):
                    continue
                title = str(raw_task.get("title") or raw_task.get("summary") or "").strip()
                if not title or title in seen_task_titles:
                    continue
                seen_task_titles.add(title)
                task_summary = str(raw_task.get("summary") or raw_task.get("description") or title).strip()
                items.append(
                    PMFlowChangeItem(
                        tenant_id=self.tenant_id,
                        change_set_id=0,
                        project_id=int(project_id),
                        group="new_object",
                        object_type="Task",
                        object_id=None,
                        title=title[:255],
                        summary=task_summary,
                        diff="从同组发布后反馈批量生成下一轮任务草稿，待 Change Review 决定是否接受。",
                        payload={
                            **base_payload,
                            "title": title,
                            "description": task_summary,
                            "summary": task_summary,
                            "stage": "development",
                            "status": "draft",
                            "priority": raw_task.get("priority") or first_quality.get("reviewPriority") or "p2",
                        },
                        edited_payload=None,
                        confidence=0.72,
                        risk_level="medium",
                        status="pending_review",
                    )
                )

        if risk_level == "high":
            items.append(
                PMFlowChangeItem(
                    tenant_id=self.tenant_id,
                    change_set_id=0,
                    project_id=int(project_id),
                    group="risk",
                    object_type="Risk",
                    object_id=None,
                    title="同组发布后反馈高风险",
                    summary=f"{group_key} 反馈组存在高风险或高影响建议，需先完成人工审查。",
                    diff="风险项只进入 Change Review，不直接写入项目记忆。",
                    payload={
                        **base_payload,
                        "title": "同组发布后反馈高风险",
                        "summary": summary_text,
                        "stage": "planning",
                        "level": "high",
                        "mitigation": "在 Change Review 中确认分组范围、owner 和处理顺序后再提交。",
                    },
                    edited_payload=None,
                    confidence=0.76,
                    risk_level="high",
                    status="pending_review",
                )
            )
        return items

    async def _mark_recommendations_for_group_change_set(
        self,
        *,
        recommendations: list[IterationRecommendation],
        change_set_id: int,
        group_key: str,
        current_user_id: Optional[int],
        reused: bool,
    ) -> None:
        stamped_at = now_cn_naive().isoformat()
        for recommendation in recommendations:
            summary_payload = dict(recommendation.recommendation_summary or {})
            summary_payload.update(
                {
                    "changeSetId": str(change_set_id),
                    "changeSetCreated": True,
                    "changeSetCreatedAt": summary_payload.get("changeSetCreatedAt") or stamped_at,
                    "changeSetCreatedBy": summary_payload.get("changeSetCreatedBy")
                    or (None if current_user_id is None else str(current_user_id)),
                    "groupKey": group_key,
                    "groupChangeSetId": str(change_set_id),
                    "groupDraftReused": bool(reused),
                }
            )
            recommendation.recommendation_summary = summary_payload
            recommendation.status = "changeset_drafted"
            flag_modified(recommendation, "recommendation_summary")

    def _serialize_revenue_event(self, event: RevenueEvent) -> dict[str, Any]:
        return {
            "id": str(event.id),
            "projectId": str(event.project_id),
            "sourceType": event.source_type,
            "category": event.category,
            "currency": event.currency,
            "amountUsd": self._jsonable(event.amount_usd),
            "externalRef": event.external_ref,
            "eventPayload": dict(event.event_payload or {}),
            "occurredAt": event.occurred_at.isoformat() if event.occurred_at else None,
            "createdAt": event.created_at.isoformat() if event.created_at else None,
            "updatedAt": event.updated_at.isoformat() if event.updated_at else None,
        }

    def _serialize_roi_snapshot(self, snapshot: ROISnapshot) -> dict[str, Any]:
        roi_ratio = self._safe_number(snapshot.roi_ratio)
        total_cost = self._safe_number(snapshot.total_cost_usd)
        risk_level = "high" if roi_ratio < 0 else ("medium" if roi_ratio == 0 and total_cost > 0 else "low")
        return {
            "id": str(snapshot.id),
            "projectId": str(snapshot.project_id),
            "rangeKey": snapshot.range_key,
            "windowStart": snapshot.window_start.isoformat() if snapshot.window_start else None,
            "windowEnd": snapshot.window_end.isoformat() if snapshot.window_end else None,
            "totalRevenueUsd": self._jsonable(snapshot.total_revenue_usd),
            "totalCostUsd": self._jsonable(snapshot.total_cost_usd),
            "roiRatio": self._jsonable(snapshot.roi_ratio),
            "snapshotPayload": dict(snapshot.snapshot_payload or {}),
            "riskLevel": risk_level,
            "status": "negative" if risk_level == "high" else ("watch" if risk_level == "medium" else "healthy"),
            "createdAt": snapshot.created_at.isoformat() if snapshot.created_at else None,
            "updatedAt": snapshot.updated_at.isoformat() if snapshot.updated_at else None,
        }

    def _serialize_compliance_check(self, run: ComplianceCheckRun) -> dict[str, Any]:
        issues = list(run.issues or [])
        status_text = str(run.status or "pending").lower()
        risk_level = (
            "high"
            if status_text in {"blocked", "failed", "rejected"} or issues
            else ("medium" if status_text in {"pending", "review_required"} else "low")
        )
        return {
            "id": str(run.id),
            "projectId": str(run.project_id),
            "charterId": None if run.charter_id is None else str(run.charter_id),
            "releaseNodeId": None if run.release_node_id is None else str(run.release_node_id),
            "projectModelBindingId": None if run.project_model_binding_id is None else str(run.project_model_binding_id),
            "status": run.status,
            "score": self._jsonable(run.score),
            "checklist": list(run.checklist or []),
            "issues": issues,
            "summary": dict(run.summary or {}),
            "riskLevel": risk_level,
            "createdAt": run.created_at.isoformat() if run.created_at else None,
            "updatedAt": run.updated_at.isoformat() if run.updated_at else None,
        }

    def _release_signal_next_actions(
        self,
        *,
        feedback_items: list[dict[str, Any]],
        revenue_events: list[dict[str, Any]],
        roi_snapshots: list[dict[str, Any]],
        compliance_checks: list[dict[str, Any]],
        recommendations: list[dict[str, Any]],
        release_gate: dict[str, Any],
    ) -> list[str]:
        actions: list[str] = []
        if not feedback_items:
            actions.append("发布后先摄取用户反馈、运营记录或人工观察，避免用 mock 增长数据判断下一轮。")
        if not revenue_events:
            actions.append("缺少真实 revenue event，暂不能用收入增长判断发布效果。")
        if not roi_snapshots:
            actions.append("缺少 ROI snapshot，暂不能判断投入产出是否健康。")
        if not compliance_checks:
            actions.append("缺少 compliance check，发布增长判断不能替代合规审查。")
        if any(item.get("riskLevel") == "high" for item in roi_snapshots):
            actions.append("ROI 信号处于风险态，下一轮建议应优先解释成本、收入或转化缺口。")
        if any(item.get("riskLevel") == "high" for item in compliance_checks):
            actions.append("存在合规风险信号，先进入 Change Review 明确缓解动作，不自动推进发布。")
        if feedback_items and recommendations:
            if any(bool(item.get("changeSetCreated")) for item in recommendations):
                actions.append("下一轮 ChangeSet draft 已进入 Change Review，继续 accept / edit / defer 后再决定是否写入 Project Memory。")
            else:
                actions.append("基于真实反馈选择 iteration recommendation，生成下一轮待确认 ChangeSet draft。")
        if str(release_gate.get("status") or "") == "blocked":
            actions.append("发布门禁仍阻塞时，先处理 release gate，再把反馈作为下一轮规划输入。")
        if not actions:
            actions.append("继续沉淀 feedback / revenue / ROI / compliance 信号，为下一轮 iteration recommendation 做准备。")
        return actions

    def _normalize_group_key(self, value: Any) -> str:
        text = str(value or "").strip().lower()
        chars: list[str] = []
        previous_dash = False
        for char in text:
            if char.isascii() and char.isalnum():
                chars.append(char)
                previous_dash = False
            elif not previous_dash:
                chars.append("-")
                previous_dash = True
        normalized = "".join(chars).strip("-")
        return normalized[:64] or "general-feedback"

    def _safe_number(self, value: Any) -> float:
        if value is None or value == "":
            return 0.0
        try:
            return float(value)
        except (TypeError, ValueError):
            return 0.0

    def _unique_strings(self, values: Any) -> list[str]:
        result: list[str] = []
        for value in values or []:
            text = str(value or "").strip()
            if text and text not in result:
                result.append(text)
        return result

    def _parse_recommendation_ids(self, raw_ids: list[str]) -> set[int]:
        parsed: set[int] = set()
        for raw in raw_ids or []:
            text = str(raw or "").strip()
            if not text.isdigit():
                raise ValueError("iteration_recommendation_not_found")
            parsed.add(int(text))
        return parsed

    def _optional_int(self, value: Optional[Any], *, error_code: str) -> Optional[int]:
        if value is None or str(value).strip() == "":
            return None
        try:
            return int(str(value).strip())
        except (TypeError, ValueError) as exc:
            raise ValueError(error_code) from exc

    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 _build_execution_quality_contract(
        self,
        *,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        acceptance: list[str],
        memory: PMFlowProjectMemory,
    ) -> dict[str, Any]:
        task_payload = self._payload(task.ai_breakdown_payload)
        title_text = f"{task.title} {task.description or ''}".lower()
        high_risk = str(task.priority or "").lower() in {"high", "urgent"} or any(
            token in title_text for token in ("auth", "permission", "tenant", "权限", "租户", "鉴权", "路由", "发布", "迁移")
        )
        required_levels = ["unit", "contract", "smoke"]
        if repo is not None:
            required_levels.append("integration")
        if high_risk:
            required_levels.append("review")
        test_cases = [
            {
                "id": f"tc-unit-{task.id}",
                "level": "unit",
                "title": "核心服务/状态推导单元测试",
                "why": "验证本次任务的业务规则不会只停留在 UI 或 prompt 文案。",
                "command": "PYTHONPATH=backend /opt/homebrew/bin/python3.11 -m pytest backend/tests/test_pmflow_os_projection_api.py -q",
                "expected": "相关 service / projection / helper 断言通过",
                "required": True,
                "blocksRelease": True,
                "status": "missing",
            },
            {
                "id": f"tc-contract-{task.id}",
                "level": "contract",
                "title": "API 合同、权限与租户边界测试",
                "why": "新增或消费 PMFlow OS API 时必须保证统一包络、权限和跨租户 fail-closed。",
                "command": "PYTHONPATH=backend /opt/homebrew/bin/python3.11 -m pytest backend/tests/test_pmflow_os_projection_api.py -q",
                "expected": "成功、失败、权限和租户边界返回产品语义错误",
                "required": True,
                "blocksRelease": True,
                "status": "missing",
            },
            {
                "id": f"tc-frontend-{task.id}",
                "level": "unit",
                "title": "前端执行桥 focused 测试",
                "why": "确保执行包、测试合同、验收和发布门禁状态在页面中可见且不回退 mock。",
                "command": "cd frontend && npm run test -- src/features/pmflow-os",
                "expected": "PMFlow OS focused tests 通过",
                "required": True,
                "blocksRelease": True,
                "status": "missing",
            },
            {
                "id": f"tc-smoke-{task.id}",
                "level": "smoke",
                "title": "Execution Bridge 到 Release/Growth 冒烟路径",
                "why": "验证正式任务、handoff、回写、验收和 release gate 可以形成用户可理解的闭环。",
                "command": "Playwright: /execution-bridge?projectId=<id> 与 /release-growth?projectId=<id>",
                "expected": "无横向溢出；执行状态、测试证据、验收状态和 release gate 正确展示",
                "required": True,
                "blocksRelease": True,
                "status": "missing",
            },
        ]
        if repo is not None:
            test_cases.append(
                {
                    "id": f"tc-integration-{task.id}",
                    "level": "integration",
                    "title": "代码仓库连接与执行回写联调",
                    "why": "Execution Bridge 必须证明 branch / commit / PR / test 证据能绑定到真实任务与仓库。",
                    "command": "回放或真实写入 execution-progress-events，覆盖 branch/commit/PR/test_passed/test_failed",
                    "expected": "execution-summary、releaseGate、Space next action 消费同一组证据",
                    "required": True,
                    "blocksRelease": True,
                    "status": "missing",
                }
            )
        environment = {
            "services": ["backend", "frontend", "postgres", "redis"],
            "envVars": ["DATABASE_URL"],
            "seedData": ["tenant", "project", "formal_task", "prompt_package", "progress_events"],
            "externalSystems": ["github_app_optional" if repo is not None else "none"],
            "canUseMock": ["webhook_payload", "browser_seed_project"],
            "mustUseReal": ["tenant/project/task ownership", "Project Memory committed task"],
        }
        if repo is not None:
            environment["envVars"].extend(["GITHUB_APP_ID", "GITHUB_APP_PRIVATE_KEY_PATH"])
            environment["seedData"].append("repo_binding")
        return {
            "qualityContractId": f"qc-task-{task.id}",
            "version": "execution_quality_contract_v1",
            "scope": {
                "projectId": str(project.id),
                "taskId": str(task.id),
                "requirementIds": [str(task.truth_node_id)] if task.truth_node_id is not None else [],
                "riskLevel": "high" if high_risk else "medium",
                "affectedSurfaces": self._unique_strings(
                    [
                        "ExecutionBridgePage",
                        "ReleaseGrowthPage",
                        *list(task_payload.get("affected_surfaces") or []),
                    ]
                ),
            },
            "testStrategy": {
                "requiredLevels": self._unique_strings(required_levels),
                "optionalLevels": ["e2e", "visual", "provider_sandbox"],
                "releaseBlockingLevels": ["unit", "contract", "smoke"] + (["integration"] if repo is not None else []),
            },
            "testCases": test_cases,
            "environment": environment,
            "evidenceRequired": {
                "commands": True,
                "testSummary": True,
                "changedFiles": True,
                "branchCommitPr": repo is not None,
                "screenshots": ["when_ui_changed"],
                "notRunReasons": True,
            },
            "acceptanceGate": {
                "userAcceptanceRequired": True,
                "technicalReviewRequired": high_risk,
                "releaseGateAllowedWhen": ["required_tests_passed", "no_blocker", "accepted"],
            },
            "acceptanceCriteria": acceptance,
        }

    def _execution_quality_contract_for_summary(
        self,
        *,
        project_id: int,
        task_id: Optional[str],
        active_tasks: list[dict[str, Any]],
        latest_package: Optional[PromptPackage],
        repo_bindings: list[dict[str, Any]],
        events: list[dict[str, Any]],
    ) -> dict[str, Any]:
        payload = self._payload(latest_package.package_payload if latest_package is not None else {})
        quality = self._payload(payload.get("quality_contract"))
        task = next((item for item in active_tasks if task_id is None or str(item.get("id")) == str(task_id)), None)
        if not quality and task is not None:
            task_title = f"{task.get('title') or ''} {task.get('description') or ''}".lower()
            risk_level = self._risk_level(task.get("priority"))
            technical_review_required = risk_level == "high" or any(
                token in task_title for token in ("auth", "permission", "tenant", "权限", "租户", "鉴权", "路由", "发布", "迁移")
            )
            required_levels = ["unit", "contract", "smoke"] + (["integration"] if repo_bindings else []) + (["review"] if technical_review_required else [])
            quality = {
                "qualityContractId": f"qc-task-{task.get('id')}",
                "version": "execution_quality_contract_v1",
                "scope": {
                    "projectId": str(project_id),
                    "taskId": str(task.get("id")),
                    "requirementIds": [str(task.get("charter_id"))] if task.get("charter_id") else [],
                    "riskLevel": risk_level,
                    "affectedSurfaces": ["ExecutionBridgePage", "ReleaseGrowthPage"],
                },
                "testStrategy": {
                    "requiredLevels": required_levels,
                    "optionalLevels": ["e2e", "visual", "provider_sandbox"],
                    "releaseBlockingLevels": required_levels,
                },
                "testCases": [
                    {
                        "id": f"tc-unit-{task.get('id')}",
                        "level": "unit",
                        "title": "核心服务/状态推导单元测试",
                        "command": "PYTHONPATH=backend /opt/homebrew/bin/python3.11 -m pytest backend/tests/test_pmflow_os_projection_api.py -q",
                        "required": True,
                        "blocksRelease": True,
                        "status": "missing",
                    },
                    {
                        "id": f"tc-contract-{task.get('id')}",
                        "level": "contract",
                        "title": "API 合同、权限与租户边界测试",
                        "command": "PYTHONPATH=backend /opt/homebrew/bin/python3.11 -m pytest backend/tests/test_pmflow_os_projection_api.py -q",
                        "required": True,
                        "blocksRelease": True,
                        "status": "missing",
                    },
                    {
                        "id": f"tc-smoke-{task.get('id')}",
                        "level": "smoke",
                        "title": "Execution Bridge 到 Release/Growth 冒烟路径",
                        "command": "Playwright: /execution-bridge?projectId=<id> 与 /release-growth?projectId=<id>",
                        "required": True,
                        "blocksRelease": True,
                        "status": "missing",
                    },
                    *(
                        [
                            {
                                "id": f"tc-integration-{task.get('id')}",
                                "level": "integration",
                                "title": "代码仓库联调与 PR 证据",
                                "command": "回写 branch / commit / PR / checks 证据",
                                "required": True,
                                "blocksRelease": True,
                                "status": "missing",
                            }
                        ]
                        if repo_bindings
                        else []
                    ),
                    *(
                        [
                            {
                                "id": f"tc-review-{task.get('id')}",
                                "level": "review",
                                "title": "高风险任务技术审查",
                                "command": "在 Execution Bridge 记录技术审查通过或要求修改",
                                "required": True,
                                "blocksRelease": True,
                                "status": "missing",
                            }
                        ]
                        if technical_review_required
                        else []
                    ),
                ],
                "environment": {
                    "services": ["backend", "frontend", "postgres", "redis"],
                    "envVars": ["DATABASE_URL"],
                    "seedData": ["tenant", "project", "formal_task", "progress_events"],
                    "externalSystems": ["github_app_optional" if repo_bindings else "none"],
                    "canUseMock": ["webhook_payload", "browser_seed_project"],
                    "mustUseReal": ["tenant/project/task ownership", "Project Memory committed task"],
                },
                "evidenceRequired": {"commands": True, "testSummary": True, "notRunReasons": True},
                "acceptanceGate": {
                    "userAcceptanceRequired": True,
                    "technicalReviewRequired": technical_review_required,
                    "releaseGateAllowedWhen": ["required_tests_passed", "no_blocker", "accepted"],
                },
            }
        return self._quality_contract_with_status(quality, events)

    def _quality_contract_with_status(self, quality: dict[str, Any], events: list[dict[str, Any]]) -> dict[str, Any]:
        if not quality:
            return {}
        cases: list[dict[str, Any]] = []
        for item in list(quality.get("testCases") or []):
            if not isinstance(item, dict):
                continue
            next_item = dict(item)
            next_item["status"] = self._quality_case_status(next_item, events)
            next_item["evidenceEventIds"] = [
                str(event.get("id"))
                for event in events
                if self._quality_event_matches_case(next_item, event)
            ][:5]
            cases.append(next_item)
        required = [item for item in cases if bool(item.get("required"))]
        passed = [item for item in required if item.get("status") == "passed"]
        failed = [item for item in required if item.get("status") == "failed"]
        blocked = [item for item in required if bool(item.get("blocksRelease")) and item.get("status") != "passed"]
        quality = dict(quality)
        quality["testCases"] = cases
        quality["status"] = "blocked" if failed else ("ready" if required and len(passed) == len(required) else "incomplete")
        quality["summary"] = {
            "required": len(required),
            "passed": len(passed),
            "failed": len(failed),
            "releaseBlockingMissing": len(blocked),
        }
        quality["blockingReasons"] = [str(item.get("title") or item.get("id")) for item in blocked[:6]]
        return quality

    def _quality_case_status(self, case: dict[str, Any], events: list[dict[str, Any]]) -> str:
        matched = [event for event in events if self._quality_event_matches_case(case, event)]
        if any(self._event_type(event) in {"test_failed", "failed", "blocked"} for event in matched):
            return "failed"
        if any(self._event_type(event) in {"test_passed", "checks_passed"} for event in matched):
            return "passed"
        return "missing"

    def _quality_event_matches_case(self, case: dict[str, Any], event: dict[str, Any]) -> bool:
        event_type = self._event_type(event)
        event_text = self._event_text(event)
        level = str(case.get("level") or "").lower()
        if event_type == "checks_passed":
            return True
        if level == "unit":
            return event_type in {"test_passed", "test_failed"} and any(token in event_text for token in ("unit", "pytest", "单元", "单测", "test"))
        if level == "contract":
            return event_type in {"test_passed", "test_failed"} and any(token in event_text for token in ("contract", "api", "合同", "权限", "tenant"))
        if level == "integration":
            return event_type in {"test_passed", "test_failed"} and any(token in event_text for token in ("integration", "联调", "repo", "pr", "github"))
        if level == "smoke":
            return event_type in {"test_passed", "test_failed", "checks_passed"} and any(token in event_text for token in ("smoke", "playwright", "冒烟", "browser", "浏览器"))
        return event_type in {"test_passed", "test_failed", "checks_passed"}

    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]
        quality_contract = self._build_execution_quality_contract(
            project=project,
            task=task,
            repo=repo,
            acceptance=acceptance,
            memory=memory,
        )
        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,
            },
            "quality_contract": quality_contract,
            "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 [])
        quality_contract = self._payload(payload.get("quality_contract"))
        test_strategy = self._payload(quality_contract.get("testStrategy"))
        environment = self._payload(quality_contract.get("environment"))
        test_cases = [item for item in list(quality_contract.get("testCases") or []) if isinstance(item, dict)]
        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(
            [
                "",
                "## Quality Contract",
                f"- Required levels: {', '.join(str(item) for item in list(test_strategy.get('requiredLevels') or [])) or 'unit, smoke'}",
                f"- Release blocking levels: {', '.join(str(item) for item in list(test_strategy.get('releaseBlockingLevels') or [])) or 'unit, smoke'}",
                "",
                "### Required Test Cases",
            ]
        )
        if test_cases:
            for item in test_cases[:8]:
                command = str(item.get("command") or "按测试环境执行").strip()
                lines.append(
                    f"- [{item.get('level')}] {item.get('title')} | required={bool(item.get('required'))} | blocks_release={bool(item.get('blocksRelease'))} | command: {command}"
                )
        else:
            lines.append("- 未生成测试合同；执行前必须人工补齐单测、冒烟和联调要求。")
        lines.extend(
            [
                "",
                "### Test Environment",
                f"- Services: {', '.join(str(item) for item in list(environment.get('services') or [])) or 'backend, frontend'}",
                f"- Required env vars: {', '.join(str(item) for item in list(environment.get('envVars') or [])) or 'none'}",
                f"- Mock allowed: {', '.join(str(item) for item in list(environment.get('canUseMock') or [])) or 'none'}",
                f"- Must use real data: {', '.join(str(item) for item in list(environment.get('mustUseReal') or [])) or 'tenant/project/task ownership'}",
                "",
                "### Evidence Return Format",
                "- Return changed files, commands run, pass/fail status, skipped tests with reason, branch/commit/PR, and screenshots when UI changed.",
                "- Missing required test evidence must be reported as blocked; do not claim acceptance or release readiness.",
            ]
        )
        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"))])
        rework_contract = self._payload(payload.get("rework_contract"))
        rework_items = [item for item in list(rework_contract.get("reviewItems") or []) if isinstance(item, dict)]
        if rework_contract:
            lines.extend(
                [
                    "",
                    "## Artifact Rework Contract",
                    "- This is a rework handoff package. Resolve the listed delivery evidence gaps before returning results.",
                    f"- Required callback: {self._payload(rework_contract.get('requiredCallback')).get('endpoint') or '/api/pmflow-os/runtime/execution-results'}",
                    "",
                    "### Review Items",
                ]
            )
            if rework_items:
                for item in rework_items[:12]:
                    expected = ", ".join(str(value) for value in list(item.get("expectedEvidence") or [])) or "updated delivery_artifacts"
                    lines.append(
                        f"- {item.get('id')}: {item.get('title')} | severity={item.get('severity')} | expected: {expected} | action: {item.get('recommendedAction')}"
                    )
            else:
                lines.append("- No review items were selected; stop and ask PMFlow for a refreshed package.")
            expected_evidence = list(rework_contract.get("expectedEvidence") or [])
            if expected_evidence:
                lines.extend(["", "### Expected Evidence"])
                lines.extend(f"- {item}" for item in expected_evidence[:20])
        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.",
                "- Rework packages still only return execution metadata; they do not write Project Memory or publish releases.",
            ]
        )
        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]:
        workflow_state = body.workflow_state or body.execution_status or self._default_workflow_state(body.event_type)
        repo_full_name = f"{repo.owner}/{repo.repo_name}" if repo is not None else None
        execution_result = {
            "protocol": "execution_result_protocol_v1",
            "eventType": body.event_type,
            "source": body.source,
            "status": body.execution_status or workflow_state,
            "branchName": body.branch_name,
            "commitSha": body.commit_sha,
            "prNumber": body.pr_number,
            "testSummary": body.test_summary,
            "blockerSummary": body.blocker_summary,
            "repositoryFullName": repo_full_name,
        }
        artifacts = body.delivery_artifacts.model_dump(mode="json") if body.delivery_artifacts is not None else None
        if artifacts is not None:
            execution_result["delivery_artifacts"] = artifacts
        payload: dict[str, Any] = {
            "protocol_version": "execution_result_protocol_v1",
            "workflow_state": workflow_state,
            "source": body.source,
            "execution_result": {
                key: value
                for key, value in execution_result.items()
                if value is not None and (not isinstance(value, str) or value.strip())
            },
            "boundary": "PMFlow records execution result metadata only; command execution stays in Local Bridge, Codex, GitHub agent, or manual runtime.",
        }
        for key, value in {
            "prompt_package_id": getattr(body, "prompt_package_id", None),
            "branch_name": body.branch_name,
            "commit_sha": body.commit_sha,
            "pr_number": body.pr_number,
            "test_summary": body.test_summary,
            "blocker_summary": body.blocker_summary,
            "execution_status": body.execution_status,
            "delivery_id": body.delivery_id,
            "event_source_id": body.event_source_id,
        }.items():
            if value is not None and str(value).strip():
                payload[key] = value
        if artifacts is not None:
            payload["delivery_artifacts"] = artifacts
        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.blocker_summary:
            return body.blocker_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