PMFlow/backend/tests/test_pmflow_os_projection_api.py

from __future__ import annotations

import os
from datetime import datetime
from types import SimpleNamespace

import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy import JSON, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test")

from app.api import pmflow_os_api  # noqa: E402
from app.core.database import get_db  # noqa: E402
from app.dependencies.auth import get_current_user  # noqa: E402
from app.models import (  # noqa: E402
    DevWorkspaceBinding,
    FeedbackIngest,
    FeishuItem,
    GitHubWebhookDelivery,
    IterationRecommendation,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowEvidenceLink,
    PMFlowLaunchDraft,
    PMFlowThread,
    PMFlowThreadMessage,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RepoBinding,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Signal,
    Task,
    TaskCodeLink,
    TruthNode,
    User,
    UserProject,
)
from app.models.base import Base  # noqa: E402


TABLES = [
    User.__table__,
    FeishuItem.__table__,
    Signal.__table__,
    Project.__table__,
    UserProject.__table__,
    PMFlowLaunchDraft.__table__,
    PMFlowThread.__table__,
    PMFlowThreadMessage.__table__,
    PMFlowChangeSet.__table__,
    PMFlowChangeItem.__table__,
    PMFlowApproval.__table__,
    PMFlowEvidenceLink.__table__,
    TruthNode.__table__,
    Task.__table__,
    SemanticFact.__table__,
    ModelCatalogItem.__table__,
    ProjectModelBinding.__table__,
    ModelRecommendationSnapshot.__table__,
    ModelEvalRun.__table__,
    PromptPackage.__table__,
    DevWorkspaceBinding.__table__,
    RepoBinding.__table__,
    TaskCodeLink.__table__,
    ProgressEvent.__table__,
    ReleaseValidationRun.__table__,
    FeedbackIngest.__table__,
    IterationRecommendation.__table__,
    RevenueEvent.__table__,
    ROISnapshot.__table__,
    GitHubWebhookDelivery.__table__,
]


def _use_sqlite_json_types() -> None:
    for table in TABLES:
        for column in table.columns:
            type_name = column.type.__class__.__name__.upper()
            if type_name in {"JSONB", "ARRAY"}:
                column.type = JSON()


@pytest_asyncio.fixture
async def db_env():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    _use_sqlite_json_types()
    async with engine.begin() as conn:
        await conn.run_sync(lambda sync_conn: Base.metadata.create_all(sync_conn, tables=TABLES))
    maker = async_sessionmaker(engine, expire_on_commit=False)
    async with maker() as session:
        yield session
    await engine.dispose()


async def _seed_user_project(session, *, tenant_id: int = 1, user_role: str = "MEMBER", member: bool = True):
    user = User(
        tenant_id=tenant_id,
        username=f"user-{tenant_id}-{user_role}-{member}",
        email=f"user-{tenant_id}-{user_role}-{member}@example.com",
        password_hash="hashed",
        name="Owner",
        role=user_role,
        assigned_domains=[],
        workload=0,
        status="active",
        is_active=True,
    )
    session.add(user)
    await session.flush()
    project = Project(
        tenant_id=tenant_id,
        name="真实项目",
        title="真实项目",
        description="从真实后端读取的项目",
        owner_id=user.id,
        status="active",
        assigned_role="BUILDER",
        signal_count=0,
    )
    session.add(project)
    await session.flush()
    if member:
        session.add(UserProject(tenant_id=tenant_id, user_id=user.id, project_id=project.id))
    await session.commit()
    return user, project


def _build_app(session, current_user) -> FastAPI:
    app = FastAPI()
    app.include_router(pmflow_os_api.router)

    async def _fake_get_db():
        yield session

    app.dependency_overrides[get_db] = _fake_get_db
    app.dependency_overrides[get_current_user] = lambda: current_user
    return app


@pytest.mark.asyncio
async def test_truth_state_empty_project_returns_compatible_empty_projection(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["project"]["id"] == str(project.id)
    assert payload["requirements"] == []
    assert payload["tasks"] == []
    assert payload["changeSets"] == []
    assert "truth_nodes_empty" in " ".join(payload["projectionWarnings"])


@pytest.mark.asyncio
async def test_truth_state_projects_memory_from_truth_nodes_and_semantic_facts(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    requirement = TruthNode(
        tenant_id=1,
        node_type="requirement",
        project_id=project.id,
        created_by=user.id,
        title="真实需求",
        content={"summary": "来自 truth_nodes 的需求", "stage": "planning"},
        version=1,
        status="aligned",
    )
    risk = TruthNode(
        tenant_id=1,
        node_type="risk",
        project_id=project.id,
        created_by=user.id,
        title="真实风险",
        content={"summary": "来自 truth_nodes 的风险", "risk_level": "high", "mitigation": "人工确认"},
        version=1,
        status="confirmed",
    )
    session.add_all([requirement, risk])
    await session.flush()
    session.add(
        SemanticFact(
            tenant_id=1,
            entity_type="truth_node",
            entity_id=requirement.id,
            fact_namespace="project_memory",
            fact_key="target_user",
            fact_value_text="个人开发者",
            fact_value_norm="个人开发者",
            value_type="keyword",
            confidence=0.91,
            evidence_source_type="truth_node",
            evidence_source_id=requirement.id,
            evidence_excerpt="来自 truth node",
            is_active=True,
        )
    )
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    payload = response.json()["data"]
    assert payload["requirements"][0]["title"] == "真实需求"
    assert payload["risks"][0]["level"] == "high"
    assert payload["projectMemory"]["sourceSummary"]["truth_nodes"] == 2
    assert payload["projectMemory"]["sourceSummary"]["semantic_facts"] == 1
    assert payload["projectMemory"]["semanticFacts"][0]["source"] == "semantic_facts"


@pytest.mark.asyncio
async def test_truth_state_task_projection_comes_from_tasks_table(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="真实宪章",
        content={"business_goal": "真实任务来源"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload={"acceptance_criteria": ["必须来自 tasks 表"]},
        manual_overrides=None,
        title="真实开发任务",
        description="从 tasks 表投影出来",
        required_skill_tag_ids=[],
        status="in_progress",
        priority="high",
    )
    session.add(task)
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    payload = response.json()["data"]
    assert payload["tasks"][0]["title"] == "真实开发任务"
    assert payload["tasks"][0]["requirementId"] == str(charter.id)
    assert payload["tasks"][0]["acceptanceCriteria"] == ["必须来自 tasks 表"]
    assert all(item["title"] != "重构会议转写主链" for item in payload["tasks"])


@pytest.mark.asyncio
async def test_execution_summary_empty_state_is_pending_with_warnings(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/execution-summary")

    payload = response.json()["data"]
    assert payload["execution"]["status"] == "queued"
    assert payload["execution"]["prompt"] == ""
    assert payload["execution"]["testSummary"] == "pending"
    assert payload["releaseGate"]["status"] == "empty"
    assert "缺少执行证据" in payload["releaseGate"]["title"]
    assert payload["releaseSignals"]["status"] == "empty"
    assert "release_feedback_empty" in " ".join(payload["releaseSignals"]["warnings"])
    assert "prompt_package_empty" in " ".join(payload["warnings"])
    assert payload["activeDevTasks"] == []


@pytest.mark.asyncio
async def test_execution_summary_projects_real_prompt_repo_tasks_and_events(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="真实执行宪章",
        content={"business_goal": "进入真实开发执行桥"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload={"acceptance_criteria": ["必须投影真实 dev-sync 数据"]},
        manual_overrides=None,
        title="接入真实执行摘要",
        description="从 execution-summary 渲染执行桥",
        required_skill_tag_ids=[],
        status="in_progress",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="pmflow-app",
        default_branch="main",
        installation_id="install-1",
        status="active",
    )
    session.add_all([task, repo])
    await session.flush()
    session.add_all(
        [
            PromptPackage(
                tenant_id=1,
                project_id=project.id,
                charter_id=charter.id,
                task_id=task.id,
                package_kind="development",
                status="active",
                package_text="真实开发提示词",
                package_payload={"scope": "execution"},
                generated_from={"source": "truth_state"},
            ),
            TaskCodeLink(
                tenant_id=1,
                task_id=task.id,
                repo_binding_id=repo.id,
                branch_name="codex/main-flow-phase-d-execution",
                pr_number=128,
                last_commit_sha="abcdef1234567890",
                status="review",
            ),
            ProgressEvent(
                tenant_id=1,
                task_id=task.id,
                project_id=project.id,
                user_id=user.id,
                repo_binding_id=repo.id,
                event_type="commit",
                source="github",
                derived_task_status="in_progress",
                summary="真实 commit 已回流",
                payload={
                    "workflow_state": "coding",
                    "branch_name": "codex/main-flow-phase-d-execution",
                    "commit_sha": "abcdef1234567890",
                    "pr_number": 128,
                },
                occurred_at=datetime(2026, 4, 26, 10, 30, 0),
            ),
            GitHubWebhookDelivery(
                tenant_id=1,
                delivery_id="delivery-1",
                event_type="push",
                repo_binding_id=repo.id,
                project_id=project.id,
                installation_id="install-1",
                status="matched",
                payload={"repository": {"full_name": "pmflow/pmflow-app"}},
                received_at=datetime(2026, 4, 26, 10, 31, 0),
            ),
        ]
    )
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/execution-summary")

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["promptPackage"]["package_text"] == "真实开发提示词"
    assert payload["repoBindings"][0]["full_name"] == "pmflow/pmflow-app"
    assert payload["repoBindings"][0]["task_link_count"] == 1
    assert payload["activeDevTasks"][0]["title"] == "接入真实执行摘要"
    assert payload["activeDevTasks"][0]["code_link"]["pr_number"] == 128
    assert payload["recentProgressEvents"][0]["summary"] == "真实 commit 已回流"
    assert payload["recentDeliveries"][0]["repository_full_name"] == "pmflow/pmflow-app"
    assert payload["warnings"] == []


@pytest.mark.asyncio
async def test_execution_handoff_creates_prompt_package_and_progress_event(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="执行交接宪章",
        content={"business_goal": "把确认任务交给真实开发执行桥"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload={"acceptance_criteria": ["生成可追踪 handoff 起点"]},
        manual_overrides=None,
        title="发起真实执行交接",
        description="从任务、项目记忆和仓库生成 PromptPackage",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="execution-bridge",
        default_branch="main",
        installation_id="install-2",
        status="active",
    )
    session.add_all([task, repo])
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "runtime_target": "local_bridge",
                "note": "交给本地桥接执行",
            },
        )

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["promptPackage"]["task_id"] == str(task.id)
    assert payload["promptPackage"]["package_kind"] == "execution_handoff"
    assert "发起真实执行交接" in payload["promptPackage"]["package_text"]
    assert "pmflow/execution-bridge" in payload["promptPackage"]["package_text"]
    assert payload["recentProgressEvents"][0]["event_type"] == "handoff_created"
    assert payload["recentProgressEvents"][0]["repo"]["full_name"] == "pmflow/execution-bridge"
    assert payload["execution"]["status"] == "pushed"
    assert payload["warnings"] == []

    prompt_rows = (await session.execute(select(PromptPackage).where(PromptPackage.task_id == task.id))).scalars().all()
    event_rows = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalars().all()
    assert len(prompt_rows) == 1
    assert len(event_rows) == 1
    assert event_rows[0].payload["runtime_target"] == "local_bridge"


@pytest.mark.asyncio
async def test_execution_handoff_without_repo_creates_draft_warning(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="无仓库交接宪章",
        content={"business_goal": "允许先生成执行交接草案"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload=None,
        manual_overrides=None,
        title="缺少仓库也能建草案",
        description="仓库后续绑定",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="medium",
    )
    session.add(task)
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs",
            json={"task_id": str(task.id), "runtime_target": "manual"},
        )

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["promptPackage"]["task_id"] == str(task.id)
    assert payload["recentProgressEvents"][0]["repo"] is None
    assert "repo_bindings_empty" in " ".join(payload["warnings"])
    event = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalar_one()
    assert event.payload["handoff_state"] == "draft"


@pytest.mark.asyncio
async def test_execution_handoff_consume_records_pickup_event(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="交接领取宪章",
        content={"business_goal": "执行交接被运行端领取后可追踪"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload={"acceptance_criteria": ["领取交接包后进入开发中"]},
        manual_overrides=None,
        title="领取执行交接包",
        description="Codex 消费 PromptPackage",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="handoff-consume",
        default_branch="main",
        installation_id="install-3",
        status="active",
    )
    session.add_all([task, repo])
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        create_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs",
            json={"task_id": str(task.id), "repo_binding_id": str(repo.id), "runtime_target": "local_bridge"},
        )
        prompt_package_id = create_response.json()["data"]["promptPackage"]["id"]
        consume_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs/{prompt_package_id}/consume",
            json={"runtime_target": "codex", "consumer_label": "Codex", "note": "开始开发"},
        )

    assert consume_response.status_code == 200
    payload = consume_response.json()["data"]
    assert payload["execution"]["status"] == "coding"
    assert payload["recentProgressEvents"][0]["event_type"] == "handoff_consumed"
    assert payload["recentProgressEvents"][0]["derived_task_status"] == "coding"
    assert payload["recentProgressEvents"][0]["repo"]["full_name"] == "pmflow/handoff-consume"

    prompt = (
        await session.execute(select(PromptPackage).where(PromptPackage.id == int(prompt_package_id)))
    ).scalar_one()
    latest_handoff = prompt.package_payload["latest_handoff"]
    assert latest_handoff["handoff_state"] == "consumed"
    assert latest_handoff["consumer_label"] == "Codex"
    assert latest_handoff["runtime_target"] == "codex"
    event_types = [
        row.event_type
        for row in (
            await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id).order_by(ProgressEvent.id.asc()))
        )
        .scalars()
        .all()
    ]
    assert event_types == ["handoff_created", "handoff_consumed"]


@pytest.mark.asyncio
async def test_execution_progress_writeback_records_branch_pr_commit_and_test(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="执行回写宪章",
        content={"business_goal": "把运行端结果回流到执行摘要"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload={"acceptance_criteria": ["回流 branch/commit/PR/test"]},
        manual_overrides=None,
        title="执行结果写回",
        description="Codex 或 Local Bridge 写回进展",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="writeback",
        default_branch="main",
        installation_id="install-4",
        status="active",
    )
    session.add_all([task, repo])
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        commit_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "commit",
                "source": "codex",
                "branch_name": "codex/main-flow-d3",
                "commit_sha": "abc123def456",
                "workflow_state": "coding",
            },
        )
        pr_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "pr_opened",
                "source": "github_agent",
                "branch_name": "codex/main-flow-d3",
                "pr_number": 77,
            },
        )
        test_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "test_failed",
                "source": "local_bridge",
                "branch_name": "codex/main-flow-d3",
                "test_summary": "单测失败：ExecutionBridgePage 回写表单缺少断言",
            },
        )

    assert commit_response.status_code == 200
    assert commit_response.json()["data"]["execution"]["status"] == "coding"
    assert pr_response.status_code == 200
    pr_payload = pr_response.json()["data"]
    assert pr_payload["execution"]["status"] == "pr_open"
    assert pr_payload["execution"]["prStatus"] == "PR #77 · linked"
    assert test_response.status_code == 200
    test_payload = test_response.json()["data"]
    assert test_payload["execution"]["status"] == "failed"
    assert test_payload["execution"]["testSummary"] == "单测失败：ExecutionBridgePage 回写表单缺少断言"
    assert test_payload["releaseGate"]["status"] == "blocked"
    assert "单测失败" in " ".join(test_payload["releaseGate"]["blockingReasons"])
    assert test_payload["recentProgressEvents"][0]["event_type"] == "test_failed"
    assert test_payload["recentProgressEvents"][0]["repo"]["full_name"] == "pmflow/writeback"

    link = (await session.execute(select(TaskCodeLink).where(TaskCodeLink.task_id == task.id))).scalar_one()
    assert link.branch_name == "codex/main-flow-d3"
    assert link.pr_number == 77
    assert link.last_commit_sha == "abc123def456"
    events = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalars().all()
    assert [event.event_type for event in events] == ["commit", "pr_opened", "test_failed"]


@pytest.mark.asyncio
async def test_release_gate_marks_test_passed_execution_as_candidate(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="发布候选宪章",
        content={"business_goal": "测试通过后进入发布候选"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload={"acceptance_criteria": ["测试通过时生成发布候选"]},
        manual_overrides=None,
        title="发布候选任务",
        description="提供 commit / PR / test evidence",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="release-candidate",
        default_branch="main",
        status="active",
    )
    session.add_all([task, repo])
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "commit",
                "source": "codex",
                "branch_name": "codex/release-candidate",
                "commit_sha": "abc123def456",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "pr_opened",
                "source": "github_agent",
                "branch_name": "codex/release-candidate",
                "pr_number": 99,
            },
        )
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "test_passed",
                "source": "local_bridge",
                "branch_name": "codex/release-candidate",
                "test_summary": "单测与构建全部通过",
            },
        )

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["releaseGate"]["status"] == "needs_confirmation"
    assert "发布候选" in payload["releaseGate"]["title"]
    assert payload["releaseGate"]["blockingReasons"] == []
    assert any(item["key"] == "test" and item["status"] == "passed" for item in payload["releaseGate"]["checks"])
    assert any(item["type"] == "test_passed" for item in payload["releaseGate"]["evidence"])


@pytest.mark.asyncio
async def test_release_gate_blocks_explicit_failed_event(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="失败回写宪章",
        content={"business_goal": "failed event 必须阻塞发布"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload=None,
        manual_overrides=None,
        title="失败回写任务",
        description="显式 failed event",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    session.add(task)
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "event_type": "failed",
                "source": "codex",
                "summary": "Codex 执行失败：typecheck 未通过",
            },
        )

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["execution"]["status"] == "failed"
    assert payload["releaseGate"]["status"] == "blocked"
    assert "typecheck 未通过" in " ".join(payload["releaseGate"]["blockingReasons"])
    assert payload["releaseGate"]["evidence"][0]["type"] == "failed"


@pytest.mark.asyncio
async def test_release_validation_run_creates_ready_record_from_ready_gate(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="Ready 发布验证宪章",
        content={"business_goal": "ready evidence 可以固化为发布验证记录"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload=None,
        manual_overrides=None,
        title="Ready 发布任务",
        description="测试通过并合并",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="ready-release",
        default_branch="main",
        status="active",
    )
    session.add_all([task, repo])
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "test_passed",
                "source": "local_bridge",
                "branch_name": "codex/ready-release",
                "test_summary": "单测、typecheck、build 全部通过",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "merged",
                "source": "github_agent",
                "branch_name": "codex/ready-release",
                "commit_sha": "ready123456789",
                "pr_number": 108,
                "summary": "PR #108 已合并",
            },
        )
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-validation-runs",
            json={"decision": "enter_release_candidate", "note": "确认进入发布候选"},
        )
        summary_response = await client.get(f"/pmflow-os/projects/{project.id}/execution-summary")

    assert response.status_code == 200
    payload = response.json()["data"]
    run_payload = payload["releaseValidationRun"]
    assert run_payload["status"] == "ready"
    assert run_payload["projectId"] == str(project.id)
    assert run_payload["validationSummary"]["releaseGateStatus"] == "ready"
    assert run_payload["validationSummary"]["humanDecision"] == "enter_release_candidate"
    assert run_payload["validationSummary"]["deploymentExecuted"] is False
    assert run_payload["validationSummary"]["createdBy"] == str(user.id)
    assert run_payload["validationSummary"]["evidence"]
    assert run_payload["rollbackPlan"]["deploymentExecuted"] is False
    assert payload["executionSummary"]["releaseGate"]["latestValidationRun"]["id"] == run_payload["id"]
    assert summary_response.json()["data"]["releaseGate"]["latestValidationRun"]["status"] == "ready"
    rows = (await session.execute(select(ReleaseValidationRun).where(ReleaseValidationRun.project_id == project.id))).scalars().all()
    assert len(rows) == 1
    assert rows[0].status == "ready"


@pytest.mark.asyncio
async def test_release_validation_run_creates_blocked_record_from_failed_gate(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="Blocked 发布验证宪章",
        content={"business_goal": "失败 evidence 固化为阻塞记录"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload=None,
        manual_overrides=None,
        title="Blocked 发布任务",
        description="测试失败",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    session.add(task)
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "event_type": "test_failed",
                "source": "codex",
                "test_summary": "发布前验证失败：回归测试未通过",
            },
        )
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-validation-runs",
            json={"decision": "keep_blocked", "note": "阻塞发布"},
        )

    assert response.status_code == 200
    payload = response.json()["data"]
    run_payload = payload["releaseValidationRun"]
    assert run_payload["status"] == "blocked"
    assert run_payload["validationSummary"]["releaseGateStatus"] == "blocked"
    assert "回归测试未通过" in " ".join(run_payload["validationSummary"]["blockingReasons"])
    assert any(item["status"] == "blocked" for item in run_payload["riskItems"])
    rows = (await session.execute(select(ReleaseValidationRun).where(ReleaseValidationRun.project_id == project.id))).scalars().all()
    assert len(rows) == 1
    assert rows[0].status == "blocked"


@pytest.mark.asyncio
async def test_release_validation_run_rejects_empty_gate(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-validation-runs",
            json={"decision": "enter_release_candidate", "note": "证据为空时不能成功"},
        )

    assert response.status_code == 409
    assert response.json()["message"] == "release_validation_evidence_empty"
    rows = (await session.execute(select(ReleaseValidationRun).where(ReleaseValidationRun.project_id == project.id))).scalars().all()
    assert rows == []


@pytest.mark.asyncio
async def test_release_feedback_ingest_creates_signal_and_iteration_recommendation(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-feedback-ingests",
            json={
                "source_type": "manual",
                "title": "首批试用用户反馈",
                "content": "用户反馈上传入口难找，希望下一轮优化 onboarding。",
                "metrics": {"negative_feedback_count": 3, "activation_drop": "12%"},
                "samples": [{"channel": "访谈", "text": "入口难找"}],
            },
        )
        summary_response = await client.get(f"/pmflow-os/projects/{project.id}/execution-summary")

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["ingest"]["title"] == "首批试用用户反馈"
    assert payload["recommendation"]["feedbackIngestId"] == payload["ingest"]["id"]
    assert payload["recommendation"]["changeSetCreated"] is False
    assert payload["releaseSignals"]["status"] == "active"
    assert payload["releaseSignals"]["feedback"][0]["metrics"]["negative_feedback_count"] == 3
    assert payload["releaseSignals"]["recommendations"][0]["feedbackIngestId"] == payload["ingest"]["id"]
    assert payload["releaseSignals"]["relationship"]["changeSetCreated"] is False
    assert payload["executionSummary"]["releaseSignals"]["status"] == "active"
    assert summary_response.json()["data"]["releaseSignals"]["feedback"][0]["title"] == "首批试用用户反馈"
    feedback_rows = (await session.execute(select(FeedbackIngest).where(FeedbackIngest.project_id == project.id))).scalars().all()
    recommendation_rows = (
        await session.execute(select(IterationRecommendation).where(IterationRecommendation.project_id == project.id))
    ).scalars().all()
    assert len(feedback_rows) == 1
    assert len(recommendation_rows) == 1


@pytest.mark.asyncio
async def test_iteration_recommendation_creates_reusable_changeset_draft(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        ingest_response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-feedback-ingests",
            json={
                "source_type": "manual",
                "title": "发布后反馈回流",
                "content": "用户反馈首次配置成功率低，希望下一轮优化 onboarding 和错误提示。",
                "metrics": {"negative_feedback_count": 5},
            },
        )
        recommendation_id = ingest_response.json()["data"]["recommendation"]["id"]
        draft_response = await client.post(
            f"/pmflow-os/projects/{project.id}/iteration-recommendations/{recommendation_id}/change-set-drafts",
            json={"note": "进入下一轮待确认变更"},
        )
        reuse_response = await client.post(
            f"/pmflow-os/projects/{project.id}/iteration-recommendations/{recommendation_id}/change-set-drafts",
            json={},
        )
        truth_response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert draft_response.status_code == 200
    payload = draft_response.json()["data"]
    detail = payload["changeSetDetail"]
    assert detail["changeSet"]["status"] == "open"
    assert detail["changeSet"]["rawStatus"] == "open"
    assert detail["evidence"][0]["sourceType"] == "iteration_recommendation"
    object_types = {item["objectType"] for item in detail["items"]}
    assert object_types.issubset({"Requirement", "Task", "Risk", "Decision"})
    assert {"Requirement", "Task", "Decision"}.issubset(object_types)
    assert all(item["status"] == "pending_review" for item in detail["items"])
    assert payload["executionSummary"]["releaseSignals"]["relationship"]["changeSetCreated"] is True
    assert payload["executionSummary"]["releaseSignals"]["recommendations"][0]["changeSetCreated"] is True
    assert payload["executionSummary"]["releaseSignals"]["recommendations"][0]["changeSetId"] == detail["changeSet"]["id"]

    assert reuse_response.status_code == 200
    assert reuse_response.json()["data"]["changeSetDetail"]["changeSet"]["id"] == detail["changeSet"]["id"]
    assert "iteration_recommendation_changeset_reused" in reuse_response.json()["data"]["warnings"]
    truth_payload = truth_response.json()["data"]
    assert truth_payload["activeChangeSetId"] == detail["changeSet"]["id"]
    assert truth_payload["changeItems"]
    assert all(item["status"] == "pending_review" for item in truth_payload["changeItems"])

    change_sets = (await session.execute(select(PMFlowChangeSet).where(PMFlowChangeSet.project_id == project.id))).scalars().all()
    tasks = (await session.execute(select(Task).where(Task.project_id == project.id))).scalars().all()
    assert len(change_sets) == 1
    assert tasks == []


@pytest.mark.asyncio
async def test_iteration_recommendation_group_batches_quality_and_reuses_changeset_draft(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        await client.post(
            f"/pmflow-os/projects/{project.id}/release-feedback-ingests",
            json={
                "source_type": "manual",
                "title": "回答质量反馈 A",
                "content": "bad answer 多次不准，用户要求下一轮补齐失败样本。",
                "metrics": {"negative_feedback_count": 4},
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/release-feedback-ingests",
            json={
                "source_type": "manual",
                "title": "回答质量反馈 B",
                "content": "出现幻觉和不准回答，需要改进 prompt quality。",
                "metrics": {"negative_feedback_count": 2},
            },
        )
        summary_response = await client.get(f"/pmflow-os/projects/{project.id}/execution-summary")
        release_signals = summary_response.json()["data"]["releaseSignals"]
        group = release_signals["recommendationGroups"][0]
        draft_response = await client.post(
            f"/pmflow-os/projects/{project.id}/iteration-recommendation-groups/{group['groupKey']}/change-set-drafts",
            json={"recommendation_ids": group["recommendationIds"], "note": "批量进入下一轮待确认"},
        )
        reuse_response = await client.post(
            f"/pmflow-os/projects/{project.id}/iteration-recommendation-groups/{group['groupKey']}/change-set-drafts",
            json={"recommendation_ids": group["recommendationIds"]},
        )

    assert summary_response.status_code == 200
    recommendations = release_signals["recommendations"]
    assert group["groupKey"] == "quality"
    assert group["mergedCount"] == 2
    assert group["impactLevel"] == "high"
    assert group["riskLevel"] == "high"
    assert group["reviewPriority"] == "p0"
    assert set(group["suggestedChangeTypes"]).issuperset({"Requirement", "Decision", "Task", "Risk"})
    assert {item["mergedCount"] for item in recommendations} == {2}
    assert sum(1 for item in recommendations if item["duplicateOf"]) == 1
    assert all(item["groupKey"] == "quality" for item in recommendations)

    assert draft_response.status_code == 200
    payload = draft_response.json()["data"]
    detail = payload["changeSetDetail"]
    assert detail["changeSet"]["status"] == "open"
    assert detail["evidence"][0]["sourceType"] == "iteration_recommendation_group"
    assert detail["evidence"][0]["metadata"]["groupKey"] == "quality"
    assert set(detail["evidence"][0]["metadata"]["recommendationIds"]) == set(group["recommendationIds"])
    object_types = {item["objectType"] for item in detail["items"]}
    assert object_types.issubset({"Requirement", "Task", "Risk", "Decision"})
    assert {"Requirement", "Decision", "Task", "Risk"}.issubset(object_types)
    assert all(item["status"] == "pending_review" for item in detail["items"])
    updated_groups = payload["executionSummary"]["releaseSignals"]["recommendationGroups"]
    assert updated_groups[0]["changeSetCreated"] is True
    assert updated_groups[0]["changeSetId"] == detail["changeSet"]["id"]
    assert all(
        item["changeSetId"] == detail["changeSet"]["id"]
        for item in payload["executionSummary"]["releaseSignals"]["recommendations"]
    )

    assert reuse_response.status_code == 200
    assert reuse_response.json()["data"]["changeSetDetail"]["changeSet"]["id"] == detail["changeSet"]["id"]
    assert "iteration_recommendation_group_changeset_reused" in reuse_response.json()["data"]["warnings"]
    change_sets = (await session.execute(select(PMFlowChangeSet).where(PMFlowChangeSet.project_id == project.id))).scalars().all()
    tasks = (await session.execute(select(Task).where(Task.project_id == project.id))).scalars().all()
    assert len(change_sets) == 1
    assert tasks == []


@pytest.mark.asyncio
async def test_truth_state_projects_reviewer_routing_for_pending_changesets(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    reviewer = User(
        tenant_id=1,
        username="reviewer",
        email="reviewer@example.com",
        password_hash="hashed",
        name="审查负责人",
        role="REVIEWER",
        assigned_domains=[],
        workload=0,
        status="active",
        is_active=True,
    )
    owner = User(
        tenant_id=1,
        username="project-owner",
        email="project-owner@example.com",
        password_hash="hashed",
        name="项目 Owner",
        role="OWNER",
        assigned_domains=[],
        workload=0,
        status="active",
        is_active=True,
    )
    session.add_all([reviewer, owner])
    await session.flush()
    session.add_all(
        [
            UserProject(tenant_id=1, user_id=reviewer.id, project_id=project.id),
            UserProject(tenant_id=1, user_id=owner.id, project_id=project.id),
        ]
    )
    thread = PMFlowThread(
        tenant_id=1,
        project_id=project.id,
        title="审查线程",
        objective="验证 reviewer routing",
        status="active",
        source="project_launch",
    )
    session.add(thread)
    await session.flush()
    message = PMFlowThreadMessage(
        tenant_id=1,
        thread_id=thread.id,
        project_id=project.id,
        role="user",
        content="本轮包含高风险和跨对象变更。",
        source="web",
        attachments=[],
        message_metadata={},
        status="sent",
    )
    session.add(message)
    await session.flush()
    change_set = PMFlowChangeSet(
        tenant_id=1,
        project_id=project.id,
        thread_id=thread.id,
        source_message_id=message.id,
        status="open",
        summary="高风险跨域变更待审",
        headline="跨域高风险 ChangeSet",
        extraction_scope="latest_message",
        routing_meta={},
        created_by=user.id,
    )
    session.add(change_set)
    await session.flush()
    risk_item = PMFlowChangeItem(
        tenant_id=1,
        change_set_id=change_set.id,
        project_id=project.id,
        group="risk",
        object_type="Risk",
        object_id=None,
        title="新增风险：权限收敛失败",
        summary="需要审查权限和数据暴露风险。",
        diff="新增 high risk。",
        payload={},
        confidence=0.92,
        risk_level="high",
        status="pending_review",
    )
    task_item = PMFlowChangeItem(
        tenant_id=1,
        change_set_id=change_set.id,
        project_id=project.id,
        group="field_update",
        object_type="Task",
        object_id=None,
        title="更新任务：补审查口径",
        summary="任务验收增加 review gate。",
        diff="更新验收标准。",
        payload={},
        confidence=0.86,
        risk_level="medium",
        status="pending_review",
    )
    session.add_all([risk_item, task_item])
    await session.flush()
    session.add(
        PMFlowEvidenceLink(
            tenant_id=1,
            project_id=project.id,
            change_set_id=change_set.id,
            change_item_id=risk_item.id,
            source_type="iteration_recommendation",
            source_id="iteration-1",
            title="迭代建议证据",
            excerpt="发布后反馈建议下一轮修复权限风险。",
            display_label="Iteration recommendation",
            evidence_metadata={},
        )
    )
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert response.status_code == 200
    payload = response.json()["data"]
    routing = payload["reviewerRouting"]
    assert routing["status"] == "ready"
    assert routing["warningCode"] is None
    assert routing["summary"]["totalPendingReview"] == 2
    assert routing["summary"]["highRiskCount"] == 1
    assert routing["summary"]["missingReviewerCount"] == 0
    assert routing["summary"]["iterationDraftCount"] == 1
    item = routing["items"][0]
    assert item["projectId"] == str(project.id)
    assert item["changeSetId"] == str(change_set.id)
    assert item["sourceType"] == "iteration_recommendation"
    assert item["pendingItemCount"] == 2
    assert item["evidenceCount"] == 1
    assert item["riskLevel"] == "high"
    assert item["priority"] == "urgent"
    assert item["changeTypes"] == ["Risk", "Task"]
    assert item["suggestedReviewer"]["role"] == "REVIEWER"
    assert item["suggestedReviewer"]["displayName"] == "审查负责人"
    assert item["targetUrl"] == f"/change-review?projectId={project.id}&changeSetId={change_set.id}"
    assert "high risk" in item["reviewReason"]
    assert payload["approvals"] == []
    assert not any("reviewer_routing_warning" in warning for warning in payload["projectionWarnings"])


@pytest.mark.asyncio
async def test_truth_state_reviewer_routing_warns_without_owner_or_reviewer_role(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    thread = PMFlowThread(
        tenant_id=1,
        project_id=project.id,
        title="审查线程",
        objective="验证 reviewer routing warning",
        status="active",
        source="project_launch",
    )
    session.add(thread)
    await session.flush()
    message = PMFlowThreadMessage(
        tenant_id=1,
        thread_id=thread.id,
        project_id=project.id,
        role="user",
        content="本轮缺少审查角色。",
        source="web",
        attachments=[],
        message_metadata={},
        status="sent",
    )
    session.add(message)
    await session.flush()
    change_set = PMFlowChangeSet(
        tenant_id=1,
        project_id=project.id,
        thread_id=thread.id,
        source_message_id=message.id,
        status="open",
        summary="缺少 reviewer",
        headline="待审但无审查角色",
        extraction_scope="latest_message",
        routing_meta={},
        created_by=user.id,
    )
    session.add(change_set)
    await session.flush()
    session.add(
        PMFlowChangeItem(
            tenant_id=1,
            change_set_id=change_set.id,
            project_id=project.id,
            group="risk",
            object_type="Risk",
            object_id=None,
            title="新增风险：缺少审查角色",
            summary="应提示 warning，不虚构 reviewer。",
            diff="新增 high risk。",
            payload={},
            confidence=0.92,
            risk_level="high",
            status="pending_review",
        )
    )
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert response.status_code == 200
    payload = response.json()["data"]
    routing = payload["reviewerRouting"]
    assert routing["status"] == "warning"
    assert routing["warningCode"] == "reviewer_missing"
    assert routing["summary"]["missingReviewerCount"] == 1
    item = routing["items"][0]
    assert item["suggestedReviewer"] is None
    assert item["fallbackReviewerRole"] == "OWNER"
    assert any("reviewer_routing_warning" in warning for warning in payload["projectionWarnings"])


@pytest.mark.asyncio
async def test_truth_state_reviewer_routing_empty_without_pending_changesets(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert response.status_code == 200
    routing = response.json()["data"]["reviewerRouting"]
    assert routing == {
        "status": "empty",
        "warningCode": None,
        "items": [],
        "summary": {
            "totalPendingReview": 0,
            "highRiskCount": 0,
            "missingReviewerCount": 0,
            "iterationDraftCount": 0,
            "requestedReviewCount": 0,
            "approvedCount": 0,
            "changesRequestedCount": 0,
            "rejectedCount": 0,
        },
    }


@pytest.mark.asyncio
async def test_execution_progress_writeback_rejects_repo_outside_project(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    other_project = Project(
        tenant_id=1,
        name="其它回写项目",
        title="其它回写项目",
        description="不能被当前项目进展事件引用",
        owner_id=user.id,
        status="active",
        assigned_role="BUILDER",
        signal_count=0,
    )
    session.add(other_project)
    await session.flush()
    session.add(UserProject(tenant_id=1, user_id=user.id, project_id=other_project.id))
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=project.id,
        created_by=user.id,
        title="当前项目宪章",
        content={"business_goal": "校验 repo project 边界"},
        version=1,
        status="aligned",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=other_project.id,
        provider="github",
        owner="pmflow",
        repo_name="other-writeback",
        default_branch="main",
        status="active",
    )
    session.add_all([charter, repo])
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload=None,
        manual_overrides=None,
        title="当前项目任务",
        description="不能绑定其它项目 repo",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="medium",
    )
    session.add(task)
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "event_type": "commit",
                "source": "codex",
                "branch_name": "codex/wrong-repo",
            },
        )

    assert response.status_code == 422
    assert response.json()["message"] == "repo_binding_not_found"


@pytest.mark.asyncio
async def test_execution_handoff_rejects_task_outside_project(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    other_project = Project(
        tenant_id=1,
        name="其它真实项目",
        title="其它真实项目",
        description="不能被当前项目交接",
        owner_id=user.id,
        status="active",
        assigned_role="BUILDER",
        signal_count=0,
    )
    session.add(other_project)
    await session.flush()
    session.add(UserProject(tenant_id=1, user_id=user.id, project_id=other_project.id))
    await session.flush()
    charter = TruthNode(
        tenant_id=1,
        node_type="charter",
        project_id=other_project.id,
        created_by=user.id,
        title="其它项目宪章",
        content={"business_goal": "不能跨项目交接"},
        version=1,
        status="aligned",
    )
    session.add(charter)
    await session.flush()
    other_task = Task(
        tenant_id=1,
        truth_node_id=charter.id,
        project_id=other_project.id,
        assignee_id=user.id,
        ai_recommended_assignees=None,
        ai_breakdown_payload=None,
        manual_overrides=None,
        title="其它项目任务",
        description="不能被当前项目 handoff",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="medium",
    )
    session.add(other_task)
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs",
            json={"task_id": str(other_task.id), "runtime_target": "codex"},
        )

    assert response.status_code == 404
    assert response.json()["message"] == "task_not_found"


@pytest.mark.asyncio
async def test_project_permission_and_tenant_boundary(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session, member=True)
    _, forbidden_project = await _seed_user_project(session, tenant_id=1, member=False)
    _, other_tenant_project = await _seed_user_project(session, tenant_id=2, member=True)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        ok_response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")
        forbidden_response = await client.get(f"/pmflow-os/projects/{forbidden_project.id}/truth-state")
        not_found_response = await client.get(f"/pmflow-os/projects/{other_tenant_project.id}/truth-state")

    assert ok_response.status_code == 200
    assert forbidden_response.status_code == 403
    assert not_found_response.status_code == 404