PMFlow/backend/tests/test_pmflow_os_projection_api.py

from __future__ import annotations

import os
import hashlib
import hmac
import json
from datetime import datetime, timedelta
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
    ComplianceCheckRun,
    DevWorkspaceBinding,
    FeedbackIngest,
    FeishuItem,
    GitHubWebhookDelivery,
    HumanCapacityOverride,
    IterationRecommendation,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowAIFeedbackEvent,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowExecutionRuntimeCredential,
    PMFlowEvidenceLink,
    PMFlowLaunchDraft,
    PMFlowThread,
    PMFlowThreadMessage,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RepoBinding,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Signal,
    Task,
    TaskCodeLink,
    TruthNode,
    User,
    UserProject,
    SkillTag,
    UserSkillBinding,
    UserWorkload,
)
from app.models.base import Base  # noqa: E402


TABLES = [
    User.__table__,
    SkillTag.__table__,
    UserSkillBinding.__table__,
    UserWorkload.__table__,
    HumanCapacityOverride.__table__,
    FeishuItem.__table__,
    Signal.__table__,
    Project.__table__,
    UserProject.__table__,
    PMFlowLaunchDraft.__table__,
    PMFlowThread.__table__,
    PMFlowThreadMessage.__table__,
    PMFlowChangeSet.__table__,
    PMFlowChangeItem.__table__,
    PMFlowApproval.__table__,
    PMFlowAIFeedbackEvent.__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__,
    PMFlowExecutionRuntimeCredential.__table__,
    ReleaseValidationRun.__table__,
    FeedbackIngest.__table__,
    IterationRecommendation.__table__,
    RevenueEvent.__table__,
    ROISnapshot.__table__,
    ComplianceCheckRun.__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


async def _seed_open_change_set(session, *, user, project, risk_level: str = "high", item_count: int = 1) -> None:
    thread = PMFlowThread(
        tenant_id=1,
        project_id=project.id,
        launch_draft_id=None,
        title="容量计划测试线程",
        objective="构造真实 ChangeSet 人工触点",
        status="active",
        source=f"capacity-test-{risk_level}-{item_count}",
    )
    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="容量计划 ChangeSet",
        headline="容量计划 ChangeSet",
        extraction_scope="latest_message",
        routing_meta={},
        created_by=user.id,
    )
    session.add(change_set)
    await session.flush()
    for index in range(item_count):
        session.add(
            PMFlowChangeItem(
                tenant_id=1,
                change_set_id=change_set.id,
                project_id=project.id,
                group="risk" if risk_level == "high" else "field_update",
                object_type="Risk" if risk_level == "high" else "Task",
                object_id=None,
                title=f"容量触点 {index + 1}",
                summary="用于验证动态容量基线。",
                diff="pending review",
                payload={},
                confidence=0.9,
                risk_level=risk_level,
                status="pending_review",
            )
        )
    await session.commit()


@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_project_copilot_ask_returns_truth_bound_context_packet(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    await _seed_open_change_set(session, user=user, project=project, risk_level="high", item_count=1)
    item_result = await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.project_id == project.id))
    change_item = item_result.scalars().first()
    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}/project-copilot/ask",
            json={
                "question": "这个高风险变更为什么不能直接执行？",
                "surface": "change_review",
                "intent": "ask",
                "object_anchor": {"type": "change_item", "id": str(change_item.id)},
            },
        )

    assert response.status_code == 200
    payload = response.json()["data"]
    packet = payload["project_copilot_context_packet"]
    assert packet["packet_version"] == "project_copilot_context_packet.v1"
    assert packet["project_ref"]["id"] == str(project.id)
    assert packet["surface"] == "change_review"
    assert packet["object_anchor"]["type"] == "change_item"
    assert packet["object_anchor"]["id"] == str(change_item.id)
    assert packet["action_ladder_level"] == "human_review"
    assert "review_explain" in packet["allowed_actions"]
    assert packet["automation_policy"]["mode"] == "suggest_only"
    assert packet["proactive_suggestions"][0]["requires_user_confirm"] is True
    assert packet["proactive_suggestions"][0]["automation_level"] == "suggest_only"
    assert packet["proactive_suggestions"][0]["why_now"]
    assert packet["proactive_suggestions"][0]["required_confirmation"] == "human_confirm_on_target_surface"
    assert packet["proactive_suggestions"][0]["audit_event"] == "project_copilot.proactive_suggestion.created"
    assert packet["change_context"]["pending_count"] == 1
    assert packet["change_context"]["commit_gate"] == "blocked_pending_items"
    blocked_actions = {item["action"] for item in payload["blocked_actions"]}
    assert "changeset_commit" in blocked_actions
    assert "project_memory_write" in blocked_actions
    assert "external_command_execution" in blocked_actions
    assert payload["answer_process_packet"]["model_calls"] == 0
    assert payload["proactive_suggestions"][0]["requires_user_confirm"] is True
    assert "待确认" in payload["answer_canvas"]["direct_answer"]
    assert payload["answer_canvas"]["next_actions"][0]["requires_user_confirm"] is True
    assert payload["customer_interaction_context_packet"]["current_focus_object"]["object_ref"] == f"change_item:{change_item.id}"


@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_project_memory_read_models_expose_items_assets_context_and_writebacks(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": "支持批量导入 PDF、Word 和 Markdown。", "stage": "planning", "tags": ["import"]},
        version=1,
        status="aligned",
    )
    decision = TruthNode(
        tenant_id=1,
        node_type="decision",
        project_id=project.id,
        created_by=user.id,
        title="首版限制导入数量",
        content={"summary": "首版单次最多导入 50 个文件。", "owner": "产品负责人"},
        version=1,
        status="confirmed",
    )
    session.add_all([requirement, decision])
    await session.flush()
    task = Task(
        tenant_id=1,
        truth_node_id=requirement.id,
        project_id=project.id,
        assignee_id=user.id,
        ai_recommended_assignees={},
        ai_breakdown_payload={},
        manual_overrides={},
        title="实现批量导入入口",
        description="把已确认需求拆成开发任务。",
        required_skill_tag_ids=[],
        status="todo",
        priority="high",
    )
    thread = PMFlowThread(
        tenant_id=1,
        project_id=project.id,
        launch_draft_id=None,
        title="项目线程",
        objective="沉淀项目记忆",
        status="active",
        source="project_memory_test",
    )
    session.add_all([task, 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="committed",
        summary="批量导入写入项目记忆",
        headline="批量导入写入",
        extraction_scope="latest_message",
        routing_meta={"scene": "project_memory_test"},
        created_by=user.id,
        committed_at=datetime.utcnow(),
    )
    session.add(change_set)
    await session.flush()
    change_item = PMFlowChangeItem(
        tenant_id=1,
        change_set_id=change_set.id,
        project_id=project.id,
        group="new_object",
        object_type="Requirement",
        object_id=str(requirement.id),
        title="支持批量导入文件",
        summary="支持批量导入 PDF、Word 和 Markdown。",
        diff="新增需求",
        payload={},
        confidence=0.94,
        risk_level="medium",
        status="committed",
    )
    session.add(change_item)
    await session.flush()
    session.add_all(
        [
            PMFlowEvidenceLink(
                tenant_id=1,
                project_id=project.id,
                change_set_id=change_set.id,
                change_item_id=change_item.id,
                source_type="conversation",
                source_id=str(message.id),
                title="对话证据",
                excerpt="需要支持批量导入文件。",
                display_label="对话",
                evidence_metadata={},
            ),
            SemanticFact(
                tenant_id=1,
                entity_type="truth_node",
                entity_id=requirement.id,
                fact_namespace="project_memory",
                fact_key="import_scope",
                fact_value_text="PDF、Word、Markdown",
                fact_value_norm="pdf_word_markdown",
                value_type="keyword",
                confidence=0.91,
                evidence_source_type="conversation",
                evidence_source_id=message.id,
                evidence_excerpt="支持批量导入文件。",
                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:
        overview = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/overview")
        items = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/items?category=requirement")
        filtered_items = await client.get(
            f"/pmflow-os/projects/{project.id}/project-memory/items?category=requirement&stage=planning&source_type=conversation&status=confirmed"
        )
        assets = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/assets")
        filtered_assets = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/assets?source_type=conversation&status=indexed")
        context = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/context-packs")
        writebacks = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/writebacks")
        item_detail = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/items/truth-node-{requirement.id}")

    assert overview.status_code == 200
    overview_payload = overview.json()["data"]
    assert overview_payload["metrics"][1]["key"] == "requirements"
    assert overview_payload["requirements"][0]["title"] == "支持批量导入文件"
    assert items.json()["data"]["items"][0]["category"] == "requirement"
    assert filtered_items.json()["data"]["items"][0]["evidenceRefs"][0]["sourceType"] == "conversation"
    assert assets.json()["data"]["assets"][0]["assetType"] == "conversation"
    assert filtered_assets.json()["data"]["assets"][0]["status"] == "indexed"
    assert context.json()["data"]["contextPacks"][0]["status"] == "ready"
    assert writebacks.json()["data"]["writebacks"][0]["itemCounts"]["committed"] == 1
    assert item_detail.status_code == 200
    assert item_detail.json()["data"]["item"]["id"] == f"truth-node-{requirement.id}"


@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_capabilities_returns_read_only_snapshot_with_warnings(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    await _seed_open_change_set(session, user=user, project=project, risk_level="medium", item_count=1)
    skill = SkillTag(
        tenant_id=1,
        name="产品判断",
        slug="product-judgement",
        category="product",
        description="判断需求范围与验收结果",
        is_system=True,
        sort_order=1,
    )
    session.add(skill)
    await session.flush()
    session.add_all(
        [
            UserSkillBinding(
                tenant_id=1,
                user_id=user.id,
                skill_tag_id=skill.id,
                proficiency_level="advanced",
                proficiency=4,
                level="advanced",
                years_experience=2,
            ),
            UserWorkload(
                tenant_id=1,
                user_id=user.id,
                current_assigned_hours=18,
                max_available_hours_per_week=40,
                workload_ratio=45,
            ),
            DevWorkspaceBinding(
                tenant_id=1,
                project_id=project.id,
                user_id=user.id,
                workspace_key="cursor-local",
                repo_root="/tmp/pmflow",
                editor_type="cursor",
                status="active",
            ),
        ]
    )
    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-capabilities")

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["projectId"] == str(project.id)
    assert {item["mode"] for item in payload["executionModes"]} >= {
        "ai_direct",
        "ai_with_user_acceptance",
        "ai_with_technical_review",
        "needs_human_judgement",
        "blocked",
    }
    assert payload["personalReadiness"]["waitingForUser"]["pendingConfirmations"] == 1
    assert payload["teamSignals"][0]["displayName"] == "Owner"
    assert payload["teamSignals"][0]["toolStatus"]["connectedTools"] == ["Cursor"]
    assert payload["teamSignals"][0]["referenceSkills"] == ["产品判断"]
    assert "repo_missing" in " ".join(payload["warnings"])


@pytest.mark.asyncio
async def test_execution_capabilities_requires_project_membership(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session, user_role="MEMBER", member=False)
    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-capabilities")

    assert response.status_code == 403


@pytest.mark.asyncio
async def test_capacity_plan_uses_current_touch_pool_for_dynamic_baseline(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session, user_role="OWNER")
    await _seed_open_change_set(session, user=user, project=project, risk_level="high", item_count=3)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="OWNER"))

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

    assert response.status_code == 200
    payload = response.json()["data"]
    member_plan = payload["memberPlans"][0]
    assert payload["teamPlannedTouchMinutes"]["focus"] > 0
    assert member_plan["suggestedBaseline"]["focus"] > member_plan["suggestedBaseline"]["acceptance"]
    assert member_plan["plannedTouchMinutes"]["total"] > 0
    assert "动态基线来自当前任务池" in " ".join(member_plan["explanations"])


@pytest.mark.asyncio
async def test_capacity_override_simulate_apply_and_permission_boundary(db_env) -> None:
    session = db_env
    owner, project = await _seed_user_project(session, user_role="OWNER")
    member = User(
        tenant_id=1,
        username="capacity-member",
        email="capacity-member@example.com",
        password_hash="hashed",
        name="普通成员",
        role="MEMBER",
        assigned_domains=[],
        workload=0,
        status="active",
        is_active=True,
    )
    session.add(member)
    await session.flush()
    session.add(UserProject(tenant_id=1, user_id=member.id, project_id=project.id))
    await session.commit()
    payload = {
        "userId": str(owner.id),
        "overrideType": "overtime",
        "addedMinutes": 120,
        "overtimeType": "focus",
        "maxPlannedUtilization": 0.95,
        "reason": "今天需要负责人兜住高风险判断",
    }

    owner_app = _build_app(session, SimpleNamespace(id=owner.id, tenant_id=1, role="OWNER"))
    async with AsyncClient(transport=ASGITransport(app=owner_app), base_url="http://test") as client:
        simulated = await client.post(f"/pmflow-os/projects/{project.id}/capacity-overrides/simulate", json=payload)
        applied = await client.post(f"/pmflow-os/projects/{project.id}/capacity-overrides", json=payload)
        plan = await client.get(f"/pmflow-os/projects/{project.id}/capacity-plan")

    assert simulated.status_code == 200
    assert simulated.json()["data"]["afterPlan"]["effectiveMinutes"] == 600
    assert "加班只增加时间预算" in " ".join(simulated.json()["data"]["impactSummary"])
    assert applied.status_code == 200
    assert applied.json()["data"]["reason"] == payload["reason"]
    member_plan = next(item for item in plan.json()["data"]["memberPlans"] if item["userId"] == str(owner.id))
    assert member_plan["effectiveMinutes"] == 600
    assert member_plan["manualOverride"]["overrideType"] == "overtime"

    member_app = _build_app(session, SimpleNamespace(id=member.id, tenant_id=1, role="MEMBER"))
    async with AsyncClient(transport=ASGITransport(app=member_app), base_url="http://test") as client:
        forbidden = await client.post(f"/pmflow-os/projects/{project.id}/capacity-overrides", json=payload)

    assert forbidden.status_code == 403


@pytest.mark.asyncio
async def test_expired_capacity_override_is_not_used_in_today_plan(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session, user_role="OWNER")
    yesterday = datetime.now() - timedelta(days=1)
    session.add(
        HumanCapacityOverride(
            tenant_id=1,
            project_id=project.id,
            user_id=user.id,
            date=datetime.now().date(),
            override_type="overtime",
            added_minutes=180,
            overtime_type="general",
            max_planned_utilization=0.95,
            work_type_bias={},
            high_risk_limit=None,
            reason="已过期的加班安排",
            expires_at=yesterday,
            created_by=user.id,
        )
    )
    await session.commit()
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="OWNER"))

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

    assert response.status_code == 200
    member_plan = response.json()["data"]["memberPlans"][0]
    assert member_plan["effectiveMinutes"] == 480
    assert member_plan["manualOverride"] is None
    assert overrides.status_code == 200
    assert overrides.json()["data"][0]["reason"] == "已过期的加班安排"


@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 "## Quality Contract" in payload["promptPackage"]["package_text"]
    assert payload["qualityContract"]["scope"]["taskId"] == str(task.id)
    assert {"unit", "contract", "smoke", "integration", "review"}.issubset(set(payload["qualityContract"]["testStrategy"]["requiredLevels"]))
    assert any(item["level"] == "contract" for item in payload["qualityContract"]["testCases"])
    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).order_by(ProgressEvent.id.asc()))
    ).scalars().all()
    assert [event.event_type for event in events] == ["commit", "pr_opened", "test_failed"]
    assert events[0].payload["protocol_version"] == "execution_result_protocol_v1"
    assert events[0].payload["execution_result"]["branchName"] == "codex/main-flow-d3"
    assert events[0].payload["execution_result"]["commitSha"] == "abc123def456"
    assert events[1].payload["execution_result"]["prNumber"] == 77
    assert events[2].payload["execution_result"]["testSummary"] == "单测失败：ExecutionBridgePage 回写表单缺少断言"


@pytest.mark.asyncio
async def test_execution_progress_runtime_result_contract_is_idempotent_by_delivery_or_source(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": "外部运行端重复 delivery 不得生成重复证据"},
        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 / Local Bridge / GitHub Agent 统一 result contract",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="runtime-contract",
        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"))

    payload = {
        "task_id": str(task.id),
        "repo_binding_id": str(repo.id),
        "event_type": "test_passed",
        "source": "codex",
        "branch_name": "codex/d8-runtime-contract",
        "commit_sha": "d8abcdef123456",
        "test_summary": "runtime result contract smoke 通过",
        "delivery_id": "delivery-d8-001",
        "event_source_id": "codex-run:d8-runtime-contract:test-passed",
    }
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        first = await client.post(f"/pmflow-os/projects/{project.id}/execution-progress-events", json=payload)
        second = await client.post(f"/pmflow-os/projects/{project.id}/execution-progress-events", json=payload)
        by_source = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={**payload, "delivery_id": "delivery-d8-002"},
        )
        summary = await client.get(f"/pmflow-os/projects/{project.id}/execution-summary")

    assert first.status_code == 200
    assert second.status_code == 200
    assert by_source.status_code == 200
    events = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalars().all()
    assert len(events) == 1
    event = events[0]
    assert event.delivery_id == "delivery-d8-001"
    assert event.event_source_id == "codex-run:d8-runtime-contract:test-passed"
    assert event.payload["protocol_version"] == "execution_result_protocol_v1"
    assert event.payload["execution_result"]["repositoryFullName"] == "pmflow/runtime-contract"
    assert event.payload["delivery_id"] == "delivery-d8-001"
    assert event.payload["event_source_id"] == "codex-run:d8-runtime-contract:test-passed"
    summary_payload = summary.json()["data"]
    assert len(summary_payload["recentProgressEvents"]) == 1
    assert summary_payload["recentProgressEvents"][0]["delivery_id"] == "delivery-d8-001"
    assert summary_payload["recentProgressEvents"][0]["event_source_id"] == "codex-run:d8-runtime-contract:test-passed"
    assert summary_payload["releaseGate"]["status"] == "needs_confirmation"


def _runtime_signature(secret: str, timestamp: str, raw_body: bytes) -> str:
    return hmac.new(secret.encode("utf-8"), f"{timestamp}.".encode("utf-8") + raw_body, hashlib.sha256).hexdigest()


@pytest.mark.asyncio
async def test_runtime_execution_result_uses_handoff_bound_signature_credential(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": "外部运行端必须用 handoff 绑定凭据回写"},
        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 runtime 通过签名入口回写结果",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="runtime-ingest",
        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:
        handoff = 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": "codex"},
        )
        prompt_package_id = handoff.json()["data"]["promptPackage"]["id"]
        credential_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs/{prompt_package_id}/runtime-credential",
            json={"runtime_target": "codex", "allowed_sources": ["codex"], "expires_in_seconds": 3600},
        )
        credential = credential_response.json()["data"]
        package_payload = {
            "prompt_package_id": prompt_package_id,
            "task_id": str(task.id),
            "source": "codex",
            "delivery_id": "package-d10-001",
            "event_source_id": "codex-run:d10:package",
        }
        package_raw_body = json.dumps(package_payload, separators=(",", ":")).encode("utf-8")
        package_timestamp = str(int(datetime.now().timestamp()))
        package_headers = {
            "Content-Type": "application/json",
            "X-PMFlow-Key-Id": credential["keyId"],
            "X-PMFlow-Timestamp": package_timestamp,
            "X-PMFlow-Signature": f"sha256={_runtime_signature(credential['secret'], package_timestamp, package_raw_body)}",
            "X-PMFlow-Delivery-Id": "package-d10-001",
        }
        package_first = await client.post("/pmflow-os/runtime/execution-packages", content=package_raw_body, headers=package_headers)
        package_second = await client.post("/pmflow-os/runtime/execution-packages", content=package_raw_body, headers=package_headers)
        runtime_payload = {
            "prompt_package_id": prompt_package_id,
            "task_id": str(task.id),
            "repo_binding_id": str(repo.id),
            "event_type": "test_passed",
            "source": "codex",
            "branch_name": "codex/d9-runtime-ingest",
            "commit_sha": "d9abcdef123456",
            "test_summary": "runtime credential signed ingest passed",
            "delivery_id": "delivery-d9-001",
            "event_source_id": "codex-run:d9:passed",
        }
        raw_body = json.dumps(runtime_payload, separators=(",", ":")).encode("utf-8")
        timestamp = str(int(datetime.now().timestamp()))
        headers = {
            "Content-Type": "application/json",
            "X-PMFlow-Key-Id": credential["keyId"],
            "X-PMFlow-Timestamp": timestamp,
            "X-PMFlow-Signature": f"sha256={_runtime_signature(credential['secret'], timestamp, raw_body)}",
            "X-PMFlow-Delivery-Id": "delivery-d9-001",
        }
        first = await client.post("/pmflow-os/runtime/execution-results", content=raw_body, headers=headers)
        second = await client.post("/pmflow-os/runtime/execution-results", content=raw_body, headers=headers)

    assert credential_response.status_code == 200
    assert credential["secret"]
    assert credential["secretPreview"] != credential["secret"]
    assert credential["packageEndpoint"] == "/api/pmflow-os/runtime/execution-packages"
    assert credential["examplePackagePayload"]["prompt_package_id"] == prompt_package_id
    assert package_first.status_code == 200
    assert package_second.status_code == 200
    package_data = package_first.json()["data"]
    assert package_data["package"]["task"]["title"] == "运行端签名回写任务"
    assert package_data["package"]["promptPackage"]["id"] == prompt_package_id
    assert package_data["callback"]["ingestEndpoint"] == "/api/pmflow-os/runtime/execution-results"
    assert package_data["audit"]["packagePulledEventId"]
    assert package_second.json()["data"]["audit"]["idempotentReplay"] is True
    assert first.status_code == 200
    assert second.status_code == 200
    events = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalars().all()
    package_events = [event for event in events if event.event_type == "execution_package_pulled"]
    assert len(package_events) == 1
    assert package_events[0].delivery_id == "package-d10-001"
    result_events = [event for event in events if event.event_type == "test_passed"]
    assert len(result_events) == 1
    assert result_events[0].delivery_id == "delivery-d9-001"
    assert result_events[0].payload["execution_result"]["commitSha"] == "d9abcdef123456"
    stored_credential = (
        await session.execute(select(PMFlowExecutionRuntimeCredential).where(PMFlowExecutionRuntimeCredential.key_id == credential["keyId"]))
    ).scalar_one()
    assert stored_credential.secret_hash == hashlib.sha256(credential["secret"].encode("utf-8")).hexdigest()
    assert stored_credential.last_used_at is not None
    assert stored_credential.use_count >= 1
    assert first.json()["data"]["releaseGate"]["status"] == "needs_confirmation"


@pytest.mark.asyncio
async def test_runtime_execution_result_rejects_invalid_signature_and_handoff_mismatch(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": "错误签名和越权 task 不得写入进展"},
        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="high",
    )
    other_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="high",
    )
    session.add_all([task, 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:
        handoff = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs",
            json={"task_id": str(task.id), "runtime_target": "codex"},
        )
        prompt_package_id = handoff.json()["data"]["promptPackage"]["id"]
        credential_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs/{prompt_package_id}/runtime-credential",
            json={"runtime_target": "codex", "allowed_sources": ["codex"], "expires_in_seconds": 3600},
        )
        credential = credential_response.json()["data"]
        payload = {
            "prompt_package_id": prompt_package_id,
            "task_id": str(task.id),
            "event_type": "test_passed",
            "source": "codex",
            "delivery_id": "delivery-d9-invalid",
        }
        raw_body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        timestamp = str(int(datetime.now().timestamp()))
        bad_signature = await client.post(
            "/pmflow-os/runtime/execution-results",
            content=raw_body,
            headers={
                "Content-Type": "application/json",
                "X-PMFlow-Key-Id": credential["keyId"],
                "X-PMFlow-Timestamp": timestamp,
                "X-PMFlow-Signature": "sha256=bad",
                "X-PMFlow-Delivery-Id": "delivery-d9-invalid",
            },
        )
        bad_package_signature = await client.post(
            "/pmflow-os/runtime/execution-packages",
            content=raw_body,
            headers={
                "Content-Type": "application/json",
                "X-PMFlow-Key-Id": credential["keyId"],
                "X-PMFlow-Timestamp": timestamp,
                "X-PMFlow-Signature": "sha256=bad",
                "X-PMFlow-Delivery-Id": "package-d10-invalid",
            },
        )
        mismatch_payload = {**payload, "task_id": str(other_task.id), "delivery_id": "delivery-d9-mismatch"}
        mismatch_raw = json.dumps(mismatch_payload, separators=(",", ":")).encode("utf-8")
        mismatch_timestamp = str(int(datetime.now().timestamp()))
        mismatch = await client.post(
            "/pmflow-os/runtime/execution-results",
            content=mismatch_raw,
            headers={
                "Content-Type": "application/json",
                "X-PMFlow-Key-Id": credential["keyId"],
                "X-PMFlow-Timestamp": mismatch_timestamp,
                "X-PMFlow-Signature": f"sha256={_runtime_signature(credential['secret'], mismatch_timestamp, mismatch_raw)}",
                "X-PMFlow-Delivery-Id": "delivery-d9-mismatch",
            },
        )

    assert bad_signature.status_code == 401
    assert bad_signature.json()["message"] == "runtime_signature_invalid"
    assert bad_package_signature.status_code == 401
    assert bad_package_signature.json()["message"] == "runtime_signature_invalid"
    assert mismatch.status_code == 401
    assert mismatch.json()["message"] == "runtime_handoff_mismatch"
    result_events = (
        await session.execute(select(ProgressEvent).where(ProgressEvent.event_type == "test_passed"))
    ).scalars().all()
    assert result_events == []


@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": "pytest unit contract smoke integration 全部通过",
                "delivery_artifacts": {
                    "changed_files": [
                        {"path": "backend/app/services/pmflow_os_projection_service.py", "change_type": "modified", "summary": "执行证据进入发布门禁"}
                    ],
                    "commands_run": [
                        {"command": "pytest backend/tests/test_pmflow_os_projection_api.py -q", "status": "passed", "summary": "相关投影测试通过"}
                    ],
                    "test_outputs": [
                        {"name": "pmflow-os projection tests", "status": "passed", "summary": "pytest unit contract smoke integration 全部通过"}
                    ],
                    "review_checklist": [
                        {"item": "边界检查", "status": "passed", "summary": "只记录证据，不执行 merge/deploy"}
                    ],
                },
            },
        )
        premature_accept = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "accept",
                "note": "尝试跳过技术审查",
            },
        )
        review_request = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "request_technical_review",
                "note": "测试已通过，发起技术审查",
            },
        )
        review_approved = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "approve_technical_review",
                "note": "技术审查通过",
            },
        )
        accepted = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "accept",
                "note": "用户验收通过",
            },
        )
        writeback = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-memory-writebacks",
            json={
                "task_id": str(task.id),
                "result": "completed",
                "note": "验收通过后写回项目记忆",
            },
        )
        writebacks = await client.get(f"/pmflow-os/projects/{project.id}/project-memory/writebacks")

    assert response.status_code == 200
    payload = response.json()["data"]
    assert payload["releaseGate"]["status"] == "needs_confirmation"
    assert "等待验收" in payload["releaseGate"]["title"]
    assert payload["acceptance"]["status"] == "pending_technical_review"
    assert payload["acceptance"]["requiredAction"] == "request_technical_review"
    assert payload["releaseGate"]["blockingReasons"] == []
    assert payload["artifactCompleteness"]["status"] == "ready"
    assert payload["releaseGate"]["releaseReadinessProof"]["status"] == "awaiting_acceptance"
    assert payload["acceptance"]["evidenceProof"]["status"] == "ready"
    assert any(item["key"] == "delivery_artifacts" and item["status"] == "passed" for item in payload["releaseGate"]["checks"])
    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"])
    assert premature_accept.status_code == 409
    assert premature_accept.json()["message"] == "technical_review_required"
    assert review_request.status_code == 200
    assert review_request.json()["data"]["acceptance"]["status"] == "pending_technical_review"
    assert review_approved.status_code == 200
    assert review_approved.json()["data"]["acceptance"]["status"] == "technical_review_approved"
    assert accepted.status_code == 200
    accepted_payload = accepted.json()["data"]
    assert accepted_payload["acceptance"]["status"] == "accepted"
    assert accepted_payload["acceptance"]["canEnterReleaseGate"] is True
    assert accepted_payload["releaseGate"]["status"] == "needs_confirmation"
    assert "等待写回项目记忆" in accepted_payload["releaseGate"]["title"]
    assert accepted_payload["releaseGate"]["releaseReadinessProof"]["status"] == "awaiting_memory_writeback"
    assert writeback.status_code == 200
    writeback_payload = writeback.json()["data"]
    assert writeback_payload["executionSummary"]["releaseGate"]["status"] == "ready"
    assert writeback_payload["executionSummary"]["releaseGate"]["releaseReadinessProof"]["status"] == "ready"
    assert writeback_payload["writeback"]["sourceType"] == "execution_memory_writeback"
    assert writeback_payload["writeback"]["taskId"] == str(task.id)
    assert writeback_payload["writeback"]["metadata"]["artifactCompleteness"]["status"] == "ready"
    assert writeback_payload["writeback"]["metadata"]["writebackProof"]["status"] == "ready"
    assert writebacks.status_code == 200
    assert writebacks.json()["data"]["writebacks"][0]["sourceType"] == "execution_memory_writeback"
    await session.refresh(task)
    assert task.status == "done"


@pytest.mark.asyncio
async def test_artifact_review_actions_track_missing_and_failed_delivery_evidence(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": ["UI 改动需要截图", "失败命令需要返工"]},
        manual_overrides=None,
        title="交付证据审查任务",
        description="验证 artifact review projection",
        required_skill_tag_ids=[],
        status="confirmed",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="artifact-review",
        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:
        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": "codex",
                "branch_name": "codex/d12-artifact-review",
                "test_summary": "部分测试输出已回写，但 UI 截图和失败命令需要处理",
                "delivery_artifacts": {
                    "changed_files": [
                        {"path": "frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx", "change_type": "modified", "summary": "新增交付证据审查区"}
                    ],
                    "commands_run": [
                        {"command": "npm run build", "status": "failed", "summary": "构建失败：类型缺少 artifactReview 字段"}
                    ],
                    "test_outputs": [
                        {"name": "artifact review projection", "status": "passed", "summary": "后端 projection 用例通过"}
                    ],
                },
            },
        )
        payload = response.json()["data"]
        item_ids = [item["id"] for item in payload["artifactReview"]["items"]]
        action = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-artifact-review-actions",
            json={
                "task_id": str(task.id),
                "action": "request_fix",
                "review_item_ids": item_ids,
                "note": "请补 UI 截图并修复 build 失败后重新回写 delivery_artifacts。",
            },
        )
        invalid = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-artifact-review-actions",
            json={
                "task_id": str(task.id),
                "action": "mark_recheck_ready",
                "review_item_ids": ["missing_unknown"],
            },
        )
        rework = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-artifact-rework-packages",
            json={
                "task_id": str(task.id),
                "repo_binding_id": str(repo.id),
                "runtime_target": "codex",
                "review_item_ids": item_ids,
                "note": "把审查项重新包装成运行端返工执行包。",
            },
        )
        rework_payload = rework.json()["data"]
        rework_prompt_id = rework_payload["promptPackage"]["id"]
        credential_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs/{rework_prompt_id}/runtime-credential",
            json={"runtime_target": "codex", "allowed_sources": ["codex"], "expires_in_seconds": 3600},
        )
        credential = credential_response.json()["data"]
        package_body = {
            "prompt_package_id": rework_prompt_id,
            "task_id": str(task.id),
            "source": "codex",
            "delivery_id": "d13-rework-package-pull",
            "event_source_id": "codex:d13-rework:package-pulled",
        }
        raw_body = json.dumps(package_body, separators=(",", ":")).encode("utf-8")
        timestamp = str(int(datetime.now().timestamp()))
        runtime_package = await client.post(
            "/pmflow-os/runtime/execution-packages",
            content=raw_body,
            headers={
                "Content-Type": "application/json",
                "X-PMFlow-Key-Id": credential["keyId"],
                "X-PMFlow-Timestamp": timestamp,
                "X-PMFlow-Signature": f"sha256={_runtime_signature(credential['secret'], timestamp, raw_body)}",
                "X-PMFlow-Delivery-Id": "d13-rework-package-pull",
            },
        )
        result_body = {
            "prompt_package_id": rework_prompt_id,
            "task_id": str(task.id),
            "repo_binding_id": str(repo.id),
            "event_type": "test_passed",
            "source": "codex",
            "branch_name": "codex/d14-rework-closure",
            "commit_sha": "abc123d14",
            "test_summary": "返工后构建通过，UI 截图已补齐",
            "execution_status": "testing",
            "delivery_id": "d14-rework-result",
            "event_source_id": "codex:d14-rework:result-returned",
            "delivery_artifacts": {
                "changed_files": [
                    {
                        "path": "frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx",
                        "change_type": "modified",
                        "summary": "补充返工闭环和截图证据展示",
                    }
                ],
                "commands_run": [
                    {"command": "npm run build", "status": "passed", "summary": "构建通过"}
                ],
                "test_outputs": [
                    {"name": "artifact review projection", "status": "passed", "summary": "后端 projection 用例通过"}
                ],
                "screenshots": [
                    {"label": "返工闭环截图", "artifact_ref": "local://screenshots/execution-bridge-rework.png"}
                ],
                "review_checklist": [
                    {"item": "交付证据可追溯", "status": "passed", "summary": "返工包、运行端回写和证据关闭均可追溯"}
                ],
            },
        }
        result_raw_body = json.dumps(result_body, separators=(",", ":")).encode("utf-8")
        result_timestamp = str(int(datetime.now().timestamp()))
        runtime_result = await client.post(
            "/pmflow-os/runtime/execution-results",
            content=result_raw_body,
            headers={
                "Content-Type": "application/json",
                "X-PMFlow-Key-Id": credential["keyId"],
                "X-PMFlow-Timestamp": result_timestamp,
                "X-PMFlow-Signature": f"sha256={_runtime_signature(credential['secret'], result_timestamp, result_raw_body)}",
                "X-PMFlow-Delivery-Id": "d14-rework-result",
            },
        )

    assert response.status_code == 200
    assert payload["artifactCompleteness"]["status"] == "blocked"
    assert payload["releaseGate"]["status"] == "blocked"
    assert payload["artifactReview"]["status"] == "blocked"
    assert {"missing_ui_screenshots", "failed_command_0"}.issubset(set(item_ids))
    assert action.status_code == 200
    action_payload = action.json()["data"]
    assert action_payload["recentProgressEvents"][0]["event_type"] == "artifact_review_requested"
    assert action_payload["artifactReview"]["actionHistory"][0]["action"] == "request_fix"
    assert action_payload["artifactReview"]["actionHistory"][0]["reviewItemIds"] == item_ids
    assert invalid.status_code == 422
    assert invalid.json()["message"] == "invalid_artifact_review_item_ids"
    assert rework.status_code == 200
    assert rework_payload["promptPackage"]["package_kind"] == "execution_handoff"
    assert rework_payload["promptPackage"]["package_payload"]["handoff_type"] == "artifact_rework"
    assert "## Artifact Rework Contract" in rework_payload["promptPackage"]["package_text"]
    assert rework_payload["recentProgressEvents"][0]["event_type"] == "artifact_rework_package_created"
    assert credential_response.status_code == 200
    assert runtime_package.status_code == 200
    package_payload = runtime_package.json()["data"]["package"]
    assert package_payload["reworkContract"]["protocol_version"] == "artifact_rework_contract_v1"
    assert {"missing_ui_screenshots", "failed_command_0"}.issubset(
        {item["id"] for item in package_payload["reworkContract"]["reviewItems"]}
    )
    assert runtime_result.status_code == 200
    result_payload = runtime_result.json()["data"]
    assert result_payload["artifactCompleteness"]["status"] == "ready"
    assert result_payload["releaseGate"]["status"] == "needs_confirmation"
    assert result_payload["releaseGate"]["releaseReadinessProof"]["status"] == "awaiting_acceptance"
    assert result_payload["releaseGate"]["releaseReadinessProof"]["unresolvedFailureEventIds"] == []
    assert result_payload["artifactReview"]["status"] == "clear"
    assert result_payload["artifactReview"]["openCount"] == 0
    closure = result_payload["artifactReview"]["closure"]
    assert closure["status"] == "all_resolved"
    assert {"missing_ui_screenshots", "failed_command_0"}.issubset(
        {item["id"] for item in closure["items"] if item["status"] == "resolved"}
    )
    assert {"package_created", "package_pulled", "result_returned"}.issubset(
        {item["kind"] for item in closure["timeline"]}
    )


@pytest.mark.asyncio
async def test_execution_acceptance_requires_test_evidence(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，没有测试通过事件",
        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:
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "event_type": "commit",
                "source": "codex",
                "commit_sha": "abc123def456",
            },
        )
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "accept",
                "note": "没有测试但尝试验收",
            },
        )

    assert response.status_code == 409
    assert response.json()["message"] == "acceptance_test_evidence_required"


@pytest.mark.asyncio
async def test_execution_memory_writeback_requires_acceptance(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="已有测试证据，但尚未验收",
        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:
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": str(task.id),
                "event_type": "test_passed",
                "source": "codex",
                "test_summary": "测试通过但还没有验收决策",
            },
        )
        response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-memory-writebacks",
            json={
                "task_id": str(task.id),
                "result": "completed",
            },
        )

    assert response.status_code == 409
    assert response.json()["message"] == "execution_acceptance_required"


@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",
                "execution_status": "failed",
                "blocker_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"
    event = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalar_one()
    assert event.payload["execution_result"]["status"] == "failed"
    assert event.payload["execution_result"]["blockerSummary"] == "Codex 执行失败：typecheck 未通过"


@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 全部通过",
                "delivery_artifacts": {
                    "changed_files": [
                        {"path": "backend/app/services/pmflow_os_projection_service.py", "change_type": "modified", "summary": "发布验证读取交付证据摘要"}
                    ],
                    "commands_run": [
                        {"command": "python -m pytest backend/tests/test_pmflow_os_projection_api.py -q", "status": "passed", "summary": "发布验证前置测试通过"}
                    ],
                    "test_outputs": [
                        {"name": "release validation projection", "status": "passed", "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 已合并",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "request_technical_review",
                "note": "Ready 发布任务需要技术审查",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "approve_technical_review",
                "note": "技术审查通过",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={
                "task_id": str(task.id),
                "decision": "accept",
                "note": "用户确认进入发布候选",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-memory-writebacks",
            json={
                "task_id": str(task.id),
                "result": "completed",
                "note": "验收后写回项目记忆，作为发布验证前置证据",
            },
        )
        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_release_signals_aggregate_revenue_roi_and_compliance_risks(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    now = datetime.utcnow()
    session.add(
        RevenueEvent(
            tenant_id=1,
            project_id=project.id,
            source_type="manual",
            category="subscription",
            currency="USD",
            amount_usd=1200.0,
            external_ref="rev_001",
            event_payload={"plan": "team"},
            occurred_at=now,
        )
    )
    session.add(
        ROISnapshot(
            tenant_id=1,
            project_id=project.id,
            range_key="monthly",
            window_start=now - timedelta(days=30),
            window_end=now,
            total_revenue_usd=1200.0,
            total_cost_usd=1800.0,
            roi_ratio=-0.3333,
            snapshot_payload={"reason": "paid acquisition cost spike"},
        )
    )
    session.add(
        ComplianceCheckRun(
            tenant_id=1,
            project_id=project.id,
            status="blocked",
            score=58.0,
            checklist=[{"key": "privacy", "status": "failed"}],
            issues=[{"title": "缺少隐私授权记录", "severity": "high"}],
            summary={"owner": "legal"},
        )
    )
    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
    release_signals = response.json()["data"]["releaseSignals"]
    assert release_signals["status"] == "active"
    assert release_signals["summary"]["revenueEventCount"] == 1
    assert release_signals["summary"]["roiSnapshotCount"] == 1
    assert release_signals["summary"]["complianceCheckCount"] == 1
    assert release_signals["summary"]["riskSignalCount"] == 2
    assert release_signals["revenueEvents"][0]["amountUsd"] == 1200.0
    assert release_signals["roiSnapshots"][0]["riskLevel"] == "high"
    assert release_signals["complianceChecks"][0]["riskLevel"] == "high"
    assert "revenue" in release_signals["relationship"]["businessSignalTypes"]
    assert "roi" in release_signals["relationship"]["businessSignalTypes"]
    assert "compliance" in release_signals["relationship"]["businessSignalTypes"]
    assert any("release_roi_risk" in item for item in release_signals["warnings"])
    assert any("release_compliance_risk" in item for item in release_signals["warnings"])


@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_v1_productized_loop_smoke_keeps_boundaries_until_next_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:
        message_response = await client.post(
            f"/pmflow-os/projects/{project.id}/threads/default/messages",
            json={
                "content": "把新手引导上传入口做成首个可验证闭环，完成后进入发布观察。",
                "source": "web",
            },
        )
        message = message_response.json()["data"]
        extraction_response = await client.post(
            f"/pmflow-os/projects/{project.id}/extractions",
            json={
                "thread_id": message["threadId"],
                "message_id": message["id"],
                "scope": "latest_message",
                "summary": "生成首轮需求与开发任务",
                "proposed_items": [
                    {
                        "group": "new_object",
                        "object_type": "Requirement",
                        "title": "优化新手上传入口",
                        "summary": "首轮闭环必须让用户在新手引导里找到上传入口。",
                        "diff": "新增正式需求候选。",
                        "payload": {
                            "summary": "首轮闭环必须让用户在新手引导里找到上传入口。",
                            "stage": "development",
                        },
                        "confidence": 0.9,
                        "risk_level": "medium",
                    },
                    {
                        "group": "new_object",
                        "object_type": "Task",
                        "title": "实现上传入口引导",
                        "summary": "开发新手引导里的上传入口和基础验证。",
                        "diff": "新增开发任务候选。",
                        "payload": {
                            "summary": "开发新手引导里的上传入口和基础验证。",
                            "description": "完成入口、错误提示、埋点和基础测试。",
                            "priority": "high",
                            "acceptanceCriteria": ["入口可见", "上传失败有错误提示", "单测和冒烟通过"],
                        },
                        "confidence": 0.88,
                        "risk_level": "medium",
                    },
                ],
            },
        )
        detail = extraction_response.json()["data"]
        change_set_id = detail["changeSet"]["id"]
        for item in detail["items"]:
            accepted = await client.post(f"/pmflow-os/change-items/{item['id']}/accept", json={})
            assert accepted.status_code == 200

        commit_response = await client.post(
            f"/pmflow-os/change-sets/{change_set_id}/commit",
            json={"mode": "atomic", "note": "D7 happy path 写入项目记忆"},
        )
        committed = commit_response.json()["data"]
        task_id = next(item["targetId"] for item in committed["committedItems"] if item["targetType"] == "task")

        repo = RepoBinding(
            tenant_id=1,
            project_id=project.id,
            provider="github",
            owner="pmflow",
            repo_name="v1-loop",
            default_branch="main",
            status="active",
        )
        session.add(repo)
        await session.commit()
        await session.refresh(repo)

        handoff_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs",
            json={
                "task_id": task_id,
                "repo_binding_id": str(repo.id),
                "runtime_target": "codex",
                "note": "生成 Codex 执行上下文包",
            },
        )
        prompt_package_id = handoff_response.json()["data"]["promptPackage"]["id"]
        consumed_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-handoffs/{prompt_package_id}/consume",
            json={"runtime_target": "codex", "consumer_label": "Codex", "note": "运行端已领取"},
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": task_id,
                "repo_binding_id": str(repo.id),
                "event_type": "commit",
                "source": "codex",
                "branch_name": "codex/v1-loop",
                "commit_sha": "d7abcdef123456",
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": task_id,
                "repo_binding_id": str(repo.id),
                "event_type": "pr_opened",
                "source": "github_agent",
                "branch_name": "codex/v1-loop",
                "pr_number": 207,
            },
        )
        test_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-progress-events",
            json={
                "task_id": task_id,
                "repo_binding_id": str(repo.id),
                "event_type": "test_passed",
                "source": "local_bridge",
                "test_summary": "unit / contract / smoke 全部通过",
                "delivery_artifacts": {
                    "changed_files": [
                        {"path": "backend/app/api/pmflow_os_api/__init__.py", "change_type": "modified", "summary": "端到端主链消费执行证据"}
                    ],
                    "commands_run": [
                        {"command": "pytest backend/tests/test_pmflow_os_projection_api.py -q", "status": "passed", "summary": "D11 主链 smoke 通过"}
                    ],
                    "test_outputs": [
                        {"name": "D11 e2e smoke", "status": "passed", "summary": "unit / contract / smoke 全部通过"}
                    ],
                    "review_checklist": [
                        {"item": "端到端边界检查", "status": "passed", "summary": "执行证据写回后仍由用户验收与 Project Memory 写回 gate 控制"}
                    ],
                },
            },
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={"task_id": task_id, "decision": "request_technical_review", "note": "进入技术审查"},
        )
        await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={"task_id": task_id, "decision": "approve_technical_review", "note": "技术审查通过"},
        )
        accepted_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-acceptance-decisions",
            json={"task_id": task_id, "decision": "accept", "note": "用户验收通过"},
        )
        writeback_response = await client.post(
            f"/pmflow-os/projects/{project.id}/execution-memory-writebacks",
            json={"task_id": task_id, "result": "completed", "note": "D7 smoke 写回项目记忆"},
        )
        validation_response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-validation-runs",
            json={"decision": "enter_release_candidate", "note": "进入发布候选，但不执行部署"},
        )
        feedback_response = await client.post(
            f"/pmflow-os/projects/{project.id}/release-feedback-ingests",
            json={
                "source_type": "manual",
                "title": "发布后首轮反馈",
                "content": "用户反馈上传入口已能找到，但错误提示还不够明确，建议下一轮优化失败提示。",
                "metrics": {"negative_feedback_count": 2},
            },
        )
        recommendation_id = feedback_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": "发布反馈进入下一轮待确认变更"},
        )
        truth_response = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert message_response.status_code == 200
    assert extraction_response.status_code == 200
    assert commit_response.status_code == 200
    assert {item["objectType"] for item in committed["committedItems"]} == {"Requirement", "Task"}
    assert handoff_response.status_code == 200
    assert consumed_response.json()["data"]["recentProgressEvents"][0]["event_type"] == "handoff_consumed"
    assert test_response.json()["data"]["releaseGate"]["status"] == "needs_confirmation"
    assert "等待验收" in test_response.json()["data"]["releaseGate"]["title"]
    assert accepted_response.json()["data"]["releaseGate"]["status"] == "needs_confirmation"
    assert "等待写回项目记忆" in accepted_response.json()["data"]["releaseGate"]["title"]
    assert writeback_response.json()["data"]["executionSummary"]["releaseGate"]["status"] == "ready"
    assert validation_response.json()["data"]["releaseValidationRun"]["validationSummary"]["deploymentExecuted"] is False
    assert validation_response.json()["data"]["executionSummary"]["releaseGate"]["latestValidationRun"]["status"] == "ready"
    assert feedback_response.json()["data"]["recommendation"]["changeSetCreated"] is False

    draft_payload = draft_response.json()["data"]
    assert draft_payload["changeSetDetail"]["changeSet"]["status"] == "open"
    assert all(item["status"] == "pending_review" for item in draft_payload["changeSetDetail"]["items"])
    assert draft_payload["executionSummary"]["releaseSignals"]["relationship"]["changeSetCreated"] is True
    truth_payload = truth_response.json()["data"]
    assert truth_payload["activeChangeSetId"] == draft_payload["changeSetDetail"]["changeSet"]["id"]
    assert any(item["title"] == "优化新手上传入口" for item in truth_payload["requirements"])
    assert any(item["title"] == "实现上传入口引导" and item["status"] in {"done", "confirmed", "draft"} for item in truth_payload["tasks"])

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


@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_truth_state_team_feed_aggregates_review_execution_and_release_events(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    reviewer = User(
        tenant_id=1,
        username="team-feed-reviewer",
        email="team-feed-reviewer@example.com",
        password_hash="hashed",
        name="审查负责人",
        role="REVIEWER",
        assigned_domains=[],
        workload=0,
        status="active",
        is_active=True,
    )
    session.add(reviewer)
    await session.flush()
    session.add(UserProject(tenant_id=1, user_id=reviewer.id, project_id=project.id))
    thread = PMFlowThread(
        tenant_id=1,
        project_id=project.id,
        title="团队动态线程",
        objective="验证 Team Feed 只读聚合",
        status="active",
        source="project_launch",
    )
    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_all([thread, charter])
    await session.flush()
    message = PMFlowThreadMessage(
        tenant_id=1,
        thread_id=thread.id,
        project_id=project.id,
        role="user",
        content="生成 Team Feed 动态流。",
        source="web",
        attachments=[],
        message_metadata={},
        status="sent",
        created_at=datetime(2026, 4, 26, 9, 55, 0),
    )
    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": ["Team Feed 只读展示执行进展"]},
        manual_overrides=None,
        title="回写 Team Feed 执行事件",
        description="执行进展进入团队动态",
        required_skill_tag_ids=[],
        status="in_progress",
        priority="high",
    )
    repo = RepoBinding(
        tenant_id=1,
        project_id=project.id,
        provider="github",
        owner="pmflow",
        repo_name="team-feed",
        default_branch="main",
        status="active",
    )
    session.add_all([message, task, repo])
    await session.flush()
    requested_set = PMFlowChangeSet(
        tenant_id=1,
        project_id=project.id,
        thread_id=thread.id,
        source_message_id=message.id,
        status="open",
        summary="待审变更已请求 reviewer",
        headline="请求审查的 ChangeSet",
        extraction_scope="latest_message",
        routing_meta={
            "reviewDecision": {
                "action": "request_review",
                "approvalState": "requested",
                "requestedAt": "2026-04-26T10:00:00",
                "reviewerId": str(reviewer.id),
                "decisionReason": "请复核高风险变更。",
            }
        },
        created_by=user.id,
        created_at=datetime(2026, 4, 26, 9, 58, 0),
    )
    approved_set = PMFlowChangeSet(
        tenant_id=1,
        project_id=project.id,
        thread_id=thread.id,
        source_message_id=message.id,
        status="reviewed",
        summary="审查已通过但未自动提交",
        headline="已通过审查的 ChangeSet",
        extraction_scope="latest_message",
        routing_meta={
            "reviewDecision": {
                "action": "approve",
                "approvalState": "approved",
                "decidedAt": "2026-04-26T10:20:00",
                "reviewerId": str(reviewer.id),
                "decisionReason": "reviewer 通过，不自动 commit。",
            }
        },
        created_by=user.id,
        reviewed_at=datetime(2026, 4, 26, 10, 20, 0),
        created_at=datetime(2026, 4, 26, 10, 5, 0),
    )
    session.add_all([requested_set, approved_set])
    await session.flush()
    requested_item = PMFlowChangeItem(
        tenant_id=1,
        change_set_id=requested_set.id,
        project_id=project.id,
        group="risk",
        object_type="Risk",
        object_id=None,
        title="新增风险：审查动态",
        summary="用于 Team Feed 的 review request。",
        diff="新增风险。",
        payload={},
        confidence=0.9,
        risk_level="high",
        status="pending_review",
    )
    approved_item = PMFlowChangeItem(
        tenant_id=1,
        change_set_id=approved_set.id,
        project_id=project.id,
        group="field_update",
        object_type="Task",
        object_id=None,
        title="更新任务：审查通过",
        summary="用于 Team Feed 的 review decision。",
        diff="更新任务。",
        payload={},
        confidence=0.86,
        risk_level="medium",
        status="accepted",
        reviewed_at=datetime(2026, 4, 26, 10, 20, 0),
    )
    session.add_all([requested_item, approved_item])
    await session.flush()
    session.add_all(
        [
            PMFlowEvidenceLink(
                tenant_id=1,
                project_id=project.id,
                change_set_id=requested_set.id,
                change_item_id=requested_item.id,
                source_type="thread_message",
                source_id=str(message.id),
                title="审查请求证据",
                excerpt="高风险变更需要审查。",
                display_label="Thread",
                evidence_metadata={},
            ),
            PMFlowEvidenceLink(
                tenant_id=1,
                project_id=project.id,
                change_set_id=approved_set.id,
                change_item_id=approved_item.id,
                source_type="thread_message",
                source_id=str(message.id),
                title="审查决策证据",
                excerpt="reviewer 已通过。",
                display_label="Thread",
                evidence_metadata={},
            ),
            PMFlowApproval(
                tenant_id=1,
                project_id=project.id,
                change_set_id=approved_set.id,
                change_item_id=approved_item.id,
                actor_user_id=reviewer.id,
                action="approve",
                decision="approved",
                before_status="pending_review",
                after_status="accepted",
                before_payload={},
                after_payload={"approvalState": "approved"},
                reason="reviewer 通过，不自动 commit。",
                created_at=datetime(2026, 4, 26, 10, 20, 0),
            ),
            ProgressEvent(
                tenant_id=1,
                task_id=task.id,
                project_id=project.id,
                user_id=user.id,
                repo_binding_id=repo.id,
                event_type="test_failed",
                source="codex",
                derived_task_status="failed",
                summary="Team Feed 单测失败，发布门禁阻塞。",
                payload={"test_summary": "Team Feed 单测失败", "pr_number": 42},
                occurred_at=datetime(2026, 4, 26, 11, 0, 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}/truth-state")

    assert response.status_code == 200
    feed = response.json()["data"]["feed"]
    titles = [item["title"] for item in feed]
    assert "已请求 ChangeSet 审查" in titles
    assert "审查已通过" in titles
    assert "ChangeSet 已完成审查" in titles
    assert "测试失败" in titles
    assert "发布门禁已阻塞" in titles
    assert any("请求审查的 ChangeSet" in item["summary"] and "证据 1 条" in item["summary"] for item in feed)
    assert any("reviewer 通过，不自动 commit" in item["summary"] for item in feed)
    assert titles.index("发布门禁已阻塞") < titles.index("已请求 ChangeSet 审查")


@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