PMFlow/backend/app/services/pmflow_os_projection_service.py

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

from __future__ import annotations

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

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified

from ..models import (
    FeedbackIngest,
    IterationRecommendation,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowEvidenceLink,
    PMFlowThreadMessage,
    PMFlowThread as PMFlowThreadModel,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RepoBinding,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Task,
    TruthNode,
    User,
    UserProject,
)
from ..schemas.pmflow_os import (
    ExecutionSummaryProjection,
    PMFlowDecision,
    PMFlowExecutionHandoffConsumeRequest,
    PMFlowExecutionHandoffRequest,
    PMFlowExecutionProgressEventRequest,
    PMFlowExecutionContext,
    PMFlowFeedEvent,
    PMFlowIterationRecommendationChangeSetDraftRequest,
    PMFlowIterationRecommendationGroupChangeSetDraftRequest,
    PMFlowMetricSignal,
    PMFlowModelDecision,
    PMFlowProjectMemory,
    PMFlowProjectSummary,
    PMFlowRelease,
    PMFlowReleaseFeedbackIngestRequest,
    PMFlowReleaseValidationRunCreateRequest,
    PMFlowRequirement,
    PMFlowRisk,
    PMFlowTask,
    PMFlowTestRun,
    PMFlowThread,
    ProjectTruthStateProjection,
)
from .ai_project_control_service import AIProjectControlService
from .pmflow_change_review_service import PMFlowChangeReviewService
from ..utils.timezone import now_cn_naive


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

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

    async def get_truth_state(
        self,
        *,
        project_id: int,
        current_user_id: Optional[int] = None,
    ) -> ProjectTruthStateProjection:
        project = await self._load_project(project_id=project_id)
        truth_nodes = await self._load_truth_nodes(project_id=project_id)
        tasks = await self._load_tasks(project_id=project_id)
        task_items = self._project_tasks(tasks)
        execution_summary = await self.get_execution_summary(
            project_id=project_id,
            current_user_id=current_user_id,
        )
        project_stage = self._infer_project_stage(truth_nodes=truth_nodes, tasks=tasks, execution=execution_summary.execution)
        launch_thread, launch_messages = await self._load_project_thread(project_id=project_id)
        thread_id = str(launch_thread.id) if launch_thread is not None else f"thread-project-{project.id}"
        requirements = self._project_requirements(truth_nodes=truth_nodes, tasks=task_items)
        releases = await self._project_releases(truth_nodes=truth_nodes, project_id=project_id)
        test_runs = await self._project_test_runs(project_id=project_id)
        metric_signals = await self._project_metric_signals(project_id=project_id)
        memory = await self._project_memory(
            project=project,
            truth_nodes=truth_nodes,
            task_ids=[int(task.id) for task, _ in tasks],
        )
        review_projection = await self._load_review_projection(project_id=project_id)
        active_change_set_id = self._active_change_set_id(review_projection["change_sets"])
        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.")

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

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

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

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

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

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

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

    async def _load_project_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]

        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 recommendations:
            warnings.append("iteration_recommendations_empty: no iteration recommendation has been generated.")

        serialized_times = [
            str(item.get("updatedAt") or item.get("createdAt"))
            for item in [*feedback_items, *recommendations, *revenue_events]
            if item.get("updatedAt") or item.get("createdAt")
        ]
        return {
            "status": "active" if feedback_items or revenue_events 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),
            },
            "summary": {
                "feedbackCount": len(feedback_items),
                "revenueEventCount": len(revenue_events),
                "iterationRecommendationCount": len(recommendations),
                "latestSignalAt": max(serialized_times) if serialized_times else None,
            },
            "feedback": feedback_items,
            "revenueEvents": revenue_events,
            "recommendations": recommendations,
            "recommendationGroups": recommendation_groups,
            "nextActions": self._release_signal_next_actions(
                feedback_items=feedback_items,
                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 _create_execution_prompt_package(
        self,
        *,
        project: Project,
        task: Task,
        repo: Optional[RepoBinding],
        runtime_target: str,
        note: Optional[str],
    ) -> PromptPackage:
        truth_nodes = await self._load_truth_nodes(project_id=int(project.id))
        memory = await self._project_memory(project=project, truth_nodes=truth_nodes, task_ids=[int(task.id)])
        payload = self._execution_prompt_payload(
            project=project,
            task=task,
            repo=repo,
            runtime_target=runtime_target,
            note=note,
            memory=memory,
        )
        package = PromptPackage(
            tenant_id=self.tenant_id,
            project_id=int(project.id),
            charter_id=int(task.truth_node_id),
            task_id=int(task.id),
            package_kind="execution_handoff",
            status="active",
            package_text=self._execution_prompt_text(payload),
            package_payload=payload,
            generated_from={
                "source": "pmflow_os_execution_handoff",
                "project_id": str(project.id),
                "task_id": str(task.id),
                "repo_binding_id": None if repo is None else str(repo.id),
                "runtime_target": runtime_target,
                "generated_at": now_cn_naive().isoformat(),
            },
        )
        self.db.add(package)
        await self.db.flush()
        package.package_payload = {
            **payload,
            "latest_handoff": self._handoff_payload(
                task=task,
                repo=repo,
                runtime_target=runtime_target,
                note=note,
                prompt_package_id=int(package.id),
            ),
        }
        return package

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

    def _project_messages(self, rows: list[PMFlowThreadMessage]) -> list[dict[str, Any]]:
        return [
            {
                "id": str(row.id),
                "role": row.role,
                "author": "你" if row.role == "user" else "PMFlow",
                "content": row.content,
                "createdAt": row.created_at.isoformat() if row.created_at else "",
                "status": "sent" if row.status == "sent" else ("analyzing" if row.status == "running" else "resolved"),
                "source": row.source,
                "metadata": row.message_metadata or {},
            }
            for row in rows
        ]

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

    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]]:
        items_by_set: dict[str, list[dict[str, Any]]] = {}
        for item in change_items:
            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 "")
            pending_items = items_by_set.get(change_set_id, [])
            if not change_set_id or not pending_items:
                continue
            change_set_evidence = evidence_by_set.get(change_set_id, [])
            risk_levels = [str(item.get("riskLevel") or "medium") for item in pending_items]
            object_types = sorted({str(item.get("objectType") or "Unknown") for item in pending_items})
            risk_groups = sorted({str(item.get("group") or "unknown") for item in pending_items})
            high_risk_signal = any(str(item.get("objectType") or "").lower() == "risk" or str(item.get("group") or "").lower() == "risk" for item in pending_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,
            )
            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 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 pending_items),
                    "pendingItemCount": len(pending_items),
                    "evidenceCount": len(change_set_evidence),
                    "suggestedReviewer": suggested_reviewer,
                    "fallbackReviewerRole": fallback_role,
                    "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 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"),
                },
            },
            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,
    ) -> list[str]:
        reasons = [f"{pending_count} 个变更项仍处于待审状态。"]
        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("包含决策类变更，审查时需要确认拍板依据。")
        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" 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
        if any(item.get("suggestedReviewer") is None for item in items):
            return "reviewer_missing"
        if any(int(item.get("evidenceCount") or 0) == 0 for item in items):
            return "evidence_missing"
        return None

    @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 _active_change_set_id(change_sets: list[dict[str, Any]]) -> Optional[str]:
        for row in change_sets:
            if row.get("rawStatus") == "open" or row.get("status") == "open":
                return str(row.get("id"))
        return None

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    def _build_release_gate(self, *, events: list[dict[str, Any]], deliveries: list[dict[str, Any]]) -> dict[str, Any]:
        evidence_events = [event for event in events if self._is_release_evidence_event(event)]
        if not evidence_events:
            return {
                "status": "empty",
                "title": "缺少执行证据",
                "summary": "还没有 branch、commit、PR、test 或 failed 回写，发布门禁不能生成判断。",
                "blockingReasons": [],
                "nextActions": ["先从 Execution Bridge 回写 branch / commit / PR / test 事件。"],
                "checks": [
                    {"key": "execution_evidence", "label": "执行证据", "status": "missing", "detail": "progress_events_empty"},
                ],
                "evidence": [],
                "updatedAt": None,
            }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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