hunjing/backend/tests/test_conversations.py

"""垂直切片测试：幂等原子事务、IDOR fail-closed、CSRF、SSE 与回滚。"""

import contextlib
import json

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from app import models
from tests.conftest import login

MESSAGE_BODY = {"content": [{"type": "text", "text": "你好，魂镜"}], "start_run": True}


async def _create_conversation(client: AsyncClient, headers: dict) -> str:
    resp = await client.post(
        "/v1/conversations", json={"title": "切片"}, headers=headers
    )
    assert resp.status_code == 201, resp.text
    return resp.json()["id"]


async def _wait_run_terminal(db, run_id: str, deadline_s: float = 10.0) -> str:
    """轮询等待 Run 终态（后台任务推进）。"""
    import asyncio

    from app.models import Run as RunModel

    deadline = asyncio.get_event_loop().time() + deadline_s
    while asyncio.get_event_loop().time() < deadline:
        async with db() as session:
            run = await session.get(RunModel, run_id)
            if run is not None and run.status in {"completed", "failed", "cancelled"}:
                return run.status
        await asyncio.sleep(0.05)
    raise AssertionError(f"Run {run_id} 未在 {deadline_s}s 内到达终态")


async def _post_message(client: AsyncClient, headers: dict, cid: str, key: str) -> object:
    return await client.post(
        f"/v1/conversations/{cid}/messages",
        json=MESSAGE_BODY,
        headers={**headers, "Idempotency-Key": key},
    )


class TestMessagePipeline:
    async def test_create_and_post_message(self, client: AsyncClient, db) -> None:
        headers = await login(client, "alice@example.com")
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, "idem-1")
        assert resp.status_code == 201, resp.text
        body = resp.json()
        assert body["message"]["id"].startswith("message_")
        assert body["message"]["content"][0]["id"].startswith("block_")
        assert body["run"]["status"] == "queued"  # 创建即返回；终态由后台执行推进
        final_status = await _wait_run_terminal(db, body["run"]["id"])
        assert final_status == "completed"

    async def test_idempotent_replay_returns_original(self, client: AsyncClient) -> None:
        headers = await login(client, "bob@example.com")
        cid = await _create_conversation(client, headers)
        first = await _post_message(client, headers, cid, "idem-replay")
        assert first.status_code == 201
        second = await _post_message(client, headers, cid, "idem-replay")
        assert second.status_code == 200  # 重放返回原资源
        assert second.json()["message"]["id"] == first.json()["message"]["id"]

    async def test_same_key_different_body_conflicts(self, client: AsyncClient) -> None:
        headers = await login(client, "carol@example.com")
        cid = await _create_conversation(client, headers)
        await _post_message(client, headers, cid, "idem-conflict")
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={"content": [{"type": "text", "text": "不同内容"}], "start_run": True},
            headers={**headers, "Idempotency-Key": "idem-conflict"},
        )
        assert resp.status_code == 409

    async def test_missing_idempotency_key_rejected(self, client: AsyncClient) -> None:
        headers = await login(client, "dave@example.com")
        cid = await _create_conversation(client, headers)
        resp = await client.post(
            f"/v1/conversations/{cid}/messages", json=MESSAGE_BODY, headers=headers
        )
        assert resp.status_code == 422

    async def test_unknown_block_type_rejected(self, client: AsyncClient) -> None:
        headers = await login(client, "erin@example.com")
        cid = await _create_conversation(client, headers)
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={"content": [{"type": "hologram", "file_id": "x"}]},
            headers={**headers, "Idempotency-Key": "idem-bad"},
        )
        assert resp.status_code == 422

    async def test_cross_domain_conversation_returns_404(self, client: AsyncClient) -> None:
        alice = await login(client, "ida@example.com")
        cid = await _create_conversation(client, alice)
        mallory = await login(client, "mallory@example.com")
        resp = await _post_message(client, mallory, cid, "idem-idor")
        assert resp.status_code == 404  # fail-closed，不泄露存在性

    async def test_csrf_token_required_for_state_change(self, client: AsyncClient) -> None:
        await login(client, "frank@example.com")
        resp = await client.post("/v1/conversations", json={"title": "无 CSRF"})
        assert resp.status_code == 403

    async def test_unauthenticated_rejected(self, client: AsyncClient) -> None:
        resp = await client.post("/v1/conversations", json={"title": "匿名"})
        assert resp.status_code == 401


class TestRunSse:
    async def test_sse_snapshot_and_terminal(self, client: AsyncClient, db) -> None:
        headers = await login(client, "grace@example.com")
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, "idem-sse")
        run_id = resp.json()["run"]["id"]
        await _wait_run_terminal(db, run_id)
        sse = await client.get(f"/v1/runs/{run_id}/events")
        assert sse.status_code == 200
        assert sse.headers["content-type"].startswith("text/event-stream")
        lines = [ln for ln in sse.text.splitlines() if ln.startswith("data: ")]
        events = [json.loads(ln.removeprefix("data: ")) for ln in lines]
        assert events[0]["type"] == "run.status.changed"
        assert events[0]["data"]["status"] == "running"
        assert events[-1]["type"] == "run.status.changed"
        assert events[-1]["data"]["status"] == "completed"
        types = [e["type"] for e in events]
        assert "content.block.started" in types
        assert "content.block.delta" in types  # 真实流产生
        assert "message.completed" in types

    async def test_cross_domain_run_returns_404(self, client: AsyncClient) -> None:
        headers = await login(client, "heidi@example.com")
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, "idem-sse2")
        run_id = resp.json()["run"]["id"]
        await login(client, "eve@example.com")
        sse = await client.get(f"/v1/runs/{run_id}/events")
        assert sse.status_code == 404


class TestAtomicity:
    async def test_failure_leaves_no_orphans(self, client: AsyncClient, db) -> None:
        from app.conversations import service

        headers = await login(client, "ivan@example.com")
        cid = await _create_conversation(client, headers)

        class BoomRun:
            def __init__(self, *args, **kwargs):
                raise RuntimeError("注入失败：Run 写入阶段崩溃")

        original = service.Run
        service.Run = BoomRun  # type: ignore[assignment]
        try:
            resp = await _post_message(client, headers, cid, "idem-atomic")
            assert resp.status_code == 500
        finally:
            service.Run = original  # type: ignore[assignment]

        async with db() as session:
            # 会话创建的合法审计/Outbox 保留；消息相关记录必须为零
            assert (await session.scalars(select(models.Message))).all() == []
            assert (await session.scalars(select(models.Run))).all() == []
            audits = (await session.scalars(select(models.AuditEvent))).all()
            assert [a for a in audits if a.aggregate_type == "message"] == []
            outbox = (await session.scalars(select(models.OutboxEvent))).all()
            assert [e for e in outbox if e.aggregate_type == "message"] == []
            assert (await session.scalars(select(models.IdempotencyRecord))).all() == []


@pytest.mark.usefixtures("client")
class TestLingtaiNormalization:
    async def test_markdown_input_expanded_by_engine(self, client: AsyncClient) -> None:
        headers = await login(client, "norm@example.com")
        cid = await _create_conversation(client, headers)
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={
                "content": [
                    {"type": "markdown", "text": "说明\n```mermaid\nA-->B\n```\n$$\nz\n$$"}
                ]
            },
            headers={**headers, "Idempotency-Key": "idem-norm"},
        )
        assert resp.status_code == 201, resp.text
        blocks = resp.json()["message"]["content"]
        assert [b["type"] for b in blocks] == ["markdown", "mermaid", "math"]
        assert all(b["id"].startswith("block_") for b in blocks)
        assert resp.json()["message"].get("parser_version") == "hunjing-markdown-v1"

    async def test_python_cpp_golden_consistency(self) -> None:
        """绑定输出与 C++ doctest 期望字节一致（跨端一致性最小 golden）。"""
        from app.core import lingtai

        out = lingtai.transform_markdown("x\n```mermaid\nA-->B\n```\n$$\nz\n$$\n")
        assert out["parser_version"] == "hunjing-markdown-v1"
        assert [b["type"] for b in out["blocks"]] == ["markdown", "mermaid", "math"]
        # 与 engine/tests/test_transform.cpp 的确定性用例保持同一输入输出
        out2 = lingtai.transform_markdown("x\n```mermaid\nA-->B\n```\n$$\nz\n$$\n")
        assert out == out2

    async def test_engine_failure_fails_closed(self, client: AsyncClient, monkeypatch) -> None:
        from app.core import lingtai as lt

        def boom(text: str):
            raise lt.LingtaiUnavailable("灵台引擎调用失败：注入")

        monkeypatch.setattr(lt, "transform_markdown", boom)
        headers = await login(client, "fc@example.com")
        cid = await _create_conversation(client, headers)
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={"content": [{"type": "markdown", "text": "hi"}]},
            headers={**headers, "Idempotency-Key": "idem-fc"},
        )
        # fail closed：引擎不可用时显式失败，不静默降级直存
        assert resp.status_code == 500


@pytest.mark.usefixtures("client")
class TestContentBlocks:
    async def test_media_blocks_accepted(self, client: AsyncClient) -> None:
        headers = await login(client, "blocks@example.com")
        cid = await _create_conversation(client, headers)
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={
                "content": [
                    {"type": "image", "file_id": "file_1", "alt": "图"},
                    {"type": "math", "notation": "latex", "display": "inline", "source": "a+b"},
                    {"type": "table", "columns": ["a"], "rows": [["1"]]},
                    {"type": "citation", "source_id": "s1", "title": "t", "quoted_text": "q"},
                    {"type": "error", "message": "块错误"},
                ]
            },
            headers={**headers, "Idempotency-Key": "idem-blocks"},
        )
        assert resp.status_code == 201, resp.text
        types = [b["type"] for b in resp.json()["message"]["content"]]
        assert types == ["image", "math", "table", "citation", "error"]

    async def test_runtime_blocks_rejected_from_client(self, client: AsyncClient) -> None:
        """契约 3.5：tool_call/agent_status 等系统 block 不允许客户端伪造。"""
        headers = await login(client, "rtb@example.com")
        cid = await _create_conversation(client, headers)
        for block in [
            {"type": "tool_call", "tool_name": "t", "call_id": "c1"},
            {"type": "agent_status", "stage": "s"},
            {"type": "approval_request", "approval_id": "a1", "action": "do"},
            {"type": "artifact", "artifact_id": "ar1", "artifact_type": "file"},
        ]:
            resp = await client.post(
                f"/v1/conversations/{cid}/messages",
                json={"content": [block]},
                headers={**headers, "Idempotency-Key": f"idem-rtb-{block['type']}"},
            )
            assert resp.status_code == 422, (block["type"], resp.text)


@pytest.mark.usefixtures("client")
class TestSseReplay:
    async def test_events_replayed_in_order_with_sequence(
        self, client: AsyncClient, db
    ) -> None:
        headers = await login(client, "sseq@example.com")
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, "idem-seq")
        run_id = resp.json()["run"]["id"]
        await _wait_run_terminal(db, run_id)
        sse = await client.get(f"/v1/runs/{run_id}/events")
        lines = [ln for ln in sse.text.splitlines() if ln.startswith("data: ")]
        events = [json.loads(ln.removeprefix("data: ")) for ln in lines]
        sequences = [e["sequence"] for e in events]
        assert sequences == sorted(sequences) == list(range(len(events)))
        assert events[0]["type"] == "run.status.changed"
        assert events[-1]["type"] == "run.status.changed"
        # delta 序号单调（契约 (block_id, delta_index) 去重基础）
        deltas = [e for e in events if e["type"] == "content.block.delta"]
        assert [d["data"]["delta_index"] for d in deltas] == list(range(len(deltas)))
        assert sum(1 for e in events if e["type"] == "content.block.completed") >= 1


@pytest.mark.usefixtures("client")
class TestPaginationAndBranches:
    async def test_message_cursor_pagination(self, client: AsyncClient) -> None:
        headers = await login(client, "page@example.com")
        cid = await _create_conversation(client, headers)
        for i in range(5):
            await client.post(
                f"/v1/conversations/{cid}/messages",
                json={"content": [{"type": "text", "text": f"m{i}"}], "start_run": False},
                headers={**headers, "Idempotency-Key": f"idem-p{i}"},
            )
        page1 = await client.get(f"/v1/conversations/{cid}/messages?limit=2", headers=headers)
        assert page1.status_code == 200
        body1 = page1.json()
        assert len(body1["data"]) == 2 and body1["has_more"] is True
        assert body1["next_cursor"]
        page2 = await client.get(
            f"/v1/conversations/{cid}/messages?limit=2&cursor={body1['next_cursor']}",
            headers=headers,
        )
        body2 = page2.json()
        ids1 = {m["id"] for m in body1["data"]}
        ids2 = {m["id"] for m in body2["data"]}
        assert not (ids1 & ids2)
        last = await client.get(
            f"/v1/conversations/{cid}/messages?limit=100&cursor={body2['next_cursor']}",
            headers=headers,
        )
        assert last.json()["has_more"] is False

    async def test_invalid_cursor_rejected(self, client: AsyncClient) -> None:
        headers = await login(client, "badc@example.com")
        cid = await _create_conversation(client, headers)
        resp = await client.get(
            f"/v1/conversations/{cid}/messages?cursor=%%%", headers=headers
        )
        assert resp.status_code == 422

    async def test_branches_and_current_path(self, client: AsyncClient) -> None:
        headers = await login(client, "branch@example.com")
        cid = await _create_conversation(client, headers)
        listing = await client.get(f"/v1/conversations/{cid}/branches", headers=headers)
        branches = listing.json()["data"]
        assert len(branches) == 1 and branches[0]["title"] == "主脉"
        # 切换到不存在的分支：422 fail-closed
        bad = await client.post(
            f"/v1/conversations/{cid}/current-branch",
            json={"branch_id": "branch_00000000-0000-7000-8000-000000000000"},
            headers=headers,
        )
        assert bad.status_code == 422
        # 切换到合法分支
        ok = await client.post(
            f"/v1/conversations/{cid}/current-branch",
            json={"branch_id": branches[0]["id"]},
            headers=headers,
        )
        assert ok.status_code == 200
        assert ok.json()["current_branch_id"] == branches[0]["id"]

    async def test_cross_domain_branch_rejected(self, client: AsyncClient) -> None:
        alice = await login(client, "br-a@example.com")
        cid = await _create_conversation(client, alice)
        listing = await client.get(f"/v1/conversations/{cid}/branches", headers=alice)
        bid = listing.json()["data"][0]["id"]
        await login(client, "br-b@example.com")
        resp = await client.post(
            f"/v1/conversations/{cid}/current-branch", json={"branch_id": bid}
        )
        assert resp.status_code in (401, 403)  # 未带新会话 Cookie：401；带错 Cookie：拒绝


@pytest.mark.usefixtures("client")
class TestRealmsAndCanonization:
    async def _make_message(self, client, headers, cid, key, text="hi") -> dict:
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={"content": [{"type": "text", "text": text}], "start_run": False},
            headers={**headers, "Idempotency-Key": key},
        )
        assert resp.status_code == 201, resp.text
        return resp.json()["message"]

    async def test_realm_switch_snapshots_into_new_messages(self, client: AsyncClient) -> None:
        headers = await login(client, "realm1@example.com")
        cid = await _create_conversation(client, headers)
        detail = await client.get(f"/v1/conversations/{cid}", headers=headers)
        assert detail.json()["realm"] == "reality"  # 默认真界
        sw = await client.post(
            f"/v1/conversations/{cid}/realm", json={"realm": "simulation"}, headers=headers
        )
        assert sw.status_code == 200 and sw.json()["realm"] == "simulation"
        await self._make_message(client, headers, cid, "idem-sim1")
        await client.get(f"/v1/conversations/{cid}/messages", headers=headers)
        # （realm 字段在消息响应中暂未透出，以归真行为间接验证）
        # 回到真界
        await client.post(
            f"/v1/conversations/{cid}/realm", json={"realm": "reality"}, headers=headers
        )

    async def test_canonize_reality_message(self, client: AsyncClient) -> None:
        headers = await login(client, "can1@example.com")
        cid = await _create_conversation(client, headers)
        msg = await self._make_message(client, headers, cid, "idem-can1")
        resp = await client.post(
            f"/v1/messages/{msg['id']}/canonizations",
            json={"scope": "reality"},
            headers=headers,
        )
        assert resp.status_code == 201
        assert resp.json()["canonical_scope"] == "reality"
        assert resp.json()["canonized_at"] is not None

    async def test_canonize_idempotent_replay(self, client: AsyncClient) -> None:
        headers = await login(client, "can2@example.com")
        cid = await _create_conversation(client, headers)
        msg = await self._make_message(client, headers, cid, "idem-can2")
        first = await client.post(
            f"/v1/messages/{msg['id']}/canonizations", json={"scope": "reality"}, headers=headers
        )
        second = await client.post(
            f"/v1/messages/{msg['id']}/canonizations", json={"scope": "reality"}, headers=headers
        )
        assert first.status_code == 201 and second.status_code == 200
        assert second.json()["canonized_at"] == first.json()["canonized_at"]

    async def test_canonize_scope_mismatch_rejected(self, client: AsyncClient) -> None:
        """真界消息不能归真到境界/幻境作用域（ADR-020 作用域一致）。"""
        headers = await login(client, "can3@example.com")
        cid = await _create_conversation(client, headers)
        msg = await self._make_message(client, headers, cid, "idem-can3")
        resp = await client.post(
            f"/v1/messages/{msg['id']}/canonizations", json={"scope": "world"}, headers=headers
        )
        assert resp.status_code == 422

    async def test_simulation_message_canonizes_as_rehearsal(self, client: AsyncClient) -> None:
        """幻境消息归真仅标记试演记录（ADR-020）。"""
        headers = await login(client, "can4@example.com")
        cid = await _create_conversation(client, headers)
        await client.post(
            f"/v1/conversations/{cid}/realm", json={"realm": "simulation"}, headers=headers
        )
        msg = await self._make_message(client, headers, cid, "idem-can4")
        resp = await client.post(
            f"/v1/messages/{msg['id']}/canonizations", json={"scope": "simulation"}, headers=headers
        )
        assert resp.status_code == 201

    async def test_canonize_cross_domain_404(self, client: AsyncClient) -> None:
        alice = await login(client, "can5@example.com")
        cid = await _create_conversation(client, alice)
        msg = await self._make_message(client, alice, cid, "idem-can5")
        await login(client, "can5-evil@example.com")
        resp = await client.post(
            f"/v1/messages/{msg['id']}/canonizations", json={"scope": "reality"}
        )
        assert resp.status_code in (401, 403, 404)  # 未携带新会话 Cookie/CSRF：拒绝


@pytest.mark.usefixtures("client")
class TestRevisions:
    async def test_revision_keeps_original_intact(self, client: AsyncClient) -> None:

        headers = await login(client, "rev1@example.com")
        cid = await _create_conversation(client, headers)
        original = None
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={"content": [{"type": "text", "text": "原始内容"}], "start_run": False},
            headers={**headers, "Idempotency-Key": "idem-rev1"},
        )
        original = resp.json()["message"]
        rev = await client.post(
            f"/v1/messages/{original['id']}/revisions",
            json={"content": [{"type": "markdown", "text": "修订后\n```python\nprint(2)\n```"}]},
            headers=headers,
        )
        assert rev.status_code == 201, rev.text
        body = rev.json()
        assert body["parent_message_id"] == original["id"]
        assert body["variant_kind"] == "revision"
        # markdown 修订经引擎规范化为两块
        assert [b["type"] for b in body["content"]] == ["markdown", "code"]
        # 原消息内容不变：列表中两条，原消息文本仍是"原始内容"
        listing = await client.get(f"/v1/conversations/{cid}/messages", headers=headers)
        by_id = {m["id"]: m for m in listing.json()["data"]}
        assert by_id[original["id"]]["content"][0]["text"] == "原始内容"

    async def test_revision_cross_domain_404(self, client: AsyncClient) -> None:
        alice = await login(client, "rev2@example.com")
        cid = await _create_conversation(client, alice)
        resp = await client.post(
            f"/v1/conversations/{cid}/messages",
            json={"content": [{"type": "text", "text": "x"}], "start_run": False},
            headers={**alice, "Idempotency-Key": "idem-rev2"},
        )
        mid = resp.json()["message"]["id"]
        await login(client, "rev2-evil@example.com")
        rev = await client.post(
            f"/v1/messages/{mid}/revisions", json={"content": [{"type": "text", "text": "篡改"}]}
        )
        assert rev.status_code in (401, 403, 404)


@pytest.mark.usefixtures("client")
class TestCharacters:
    async def _create(self, client, headers, body) -> dict:
        resp = await client.post("/v1/characters", json=body, headers=headers)
        assert resp.status_code == 201, resp.text
        return resp.json()

    async def test_manual_create_activate_and_first_statement(self, client: AsyncClient) -> None:
        headers = await login(client, "char1@example.com")
        c = await self._create(
            client, headers, {"name": "昭星", "identity_baseline": {"core_traits": ["沉静"]}}
        )
        assert c["status"] == "draft" and c["origin"] == "original"
        act = await client.post(f"/v1/characters/{c['id']}/activations", headers=headers)
        assert act.status_code == 201 and act.json()["status"] == "active"
        assert act.json()["current_version_id"] is not None
        fs = await client.post(
            f"/v1/characters/{c['id']}/first-statement",
            json={"text": "初次见面，我是昭星。"},
            headers=headers,
        )
        assert fs.status_code == 201 and fs.json()["status"] == "pending"
        confirm = await client.post(
            f"/v1/characters/{c['id']}/first-statement/confirmations",
            json={"version_id": fs.json()["id"]},
            headers=headers,
        )
        assert confirm.status_code == 201 and confirm.json()["status"] == "approved"
        detail = await client.get(f"/v1/characters/{c['id']}", headers=headers)
        assert detail.json()["identity_baseline"]["first_statement"] == "初次见面，我是昭星。"

    async def test_serendipity_is_deterministic_single(self, client: AsyncClient) -> None:
        """随缘：同种子同候选（可复现），一次一个，不是抽卡。"""
        headers = await login(client, "char2@example.com")
        a = await self._create(client, headers, {"origin": "serendipity", "seed": 3})
        b = await self._create(client, headers, {"origin": "serendipity", "seed": 3})
        assert a["name"] == b["name"] and a["origin"] == "serendipity"
        assert a["drives"] and a["drives"][0]["name"]

    async def test_real_person_governance(self, client: AsyncClient) -> None:
        headers = await login(client, "char3@example.com")
        # 已故现实人物：拒绝
        dead = await client.post(
            "/v1/characters",
            json={"name": "某人", "based_on_real_person": True, "deceased_real_person": True,
                  "real_person_attested": True},
            headers=headers,
        )
        assert dead.status_code == 422
        # 未声明授权：拒绝
        no_attest = await client.post(
            "/v1/characters", json={"name": "某人", "based_on_real_person": True}, headers=headers
        )
        assert no_attest.status_code == 422
        # 本人声明：允许
        ok = await self._create(
            client, headers,
            {"name": "我自己", "based_on_real_person": True, "real_person_attested": True},
        )
        assert ok["based_on_real_person"] is True

    async def test_lifecycle_full_path(self, client: AsyncClient) -> None:
        headers = await login(client, "char4@example.com")
        c = await self._create(client, headers, {"name": "汀棠"})
        await client.post(f"/v1/characters/{c['id']}/activations", headers=headers)
        # 静修 → 归藏 → 复照
        rest = await client.post(f"/v1/characters/{c['id']}/rests", headers=headers)
        assert rest.json()["status"] == "resting"
        arch = await client.post(f"/v1/characters/{c['id']}/archives", headers=headers)
        assert arch.json()["status"] == "archived"
        revive = await client.post(f"/v1/characters/{c['id']}/revivals", headers=headers)
        assert revive.json()["status"] == "active"
        # 死亡：请求 → 撤销（复照）→ 再请求 → 未满冷静期确认被拒
        req = await client.post(f"/v1/characters/{c['id']}/death-requests", headers=headers)
        assert req.json()["status"] == "death_pending"
        cancel = await client.post(f"/v1/characters/{c['id']}/death-cancellations", headers=headers)
        assert cancel.json()["status"] == "active"
        await client.post(f"/v1/characters/{c['id']}/death-requests", headers=headers)
        early = await client.post(f"/v1/characters/{c['id']}/death-confirmations", headers=headers)
        assert early.status_code == 422  # 冷静期未满

    async def test_invalid_transition_rejected(self, client: AsyncClient) -> None:
        headers = await login(client, "char5@example.com")
        c = await self._create(client, headers, {"name": "砚谷"})
        # 灵胚不能直接归藏
        resp = await client.post(f"/v1/characters/{c['id']}/archives", headers=headers)
        assert resp.status_code == 422

    async def test_reforging_requires_confirm(self, client: AsyncClient) -> None:
        headers = await login(client, "char6@example.com")
        c = await self._create(client, headers, {"name": "朔川"})
        await client.post(f"/v1/characters/{c['id']}/activations", headers=headers)
        no = await client.post(
            f"/v1/characters/{c['id']}/reforgings",
            json={"identity_baseline": {"core_traits": ["焕新"]}},
            headers=headers,
        )
        assert no.status_code == 422
        yes = await client.post(
            f"/v1/characters/{c['id']}/reforgings",
            json={"confirm": True, "identity_baseline": {"core_traits": ["焕新"]}},
            headers=headers,
        )
        assert yes.status_code == 201 and yes.json()["kind"] == "reforging"
        detail = await client.get(f"/v1/characters/{c['id']}", headers=headers)
        assert detail.json()["identity_baseline"]["core_traits"] == ["焕新"]

    async def test_drives_update_and_persona_review(self, client: AsyncClient) -> None:
        headers = await login(client, "char7@example.com")
        c = await self._create(client, headers, {"name": "霁灯"})
        await client.post(f"/v1/characters/{c['id']}/activations", headers=headers)
        upd = await client.patch(
            f"/v1/characters/{c['id']}/drives",
            json={"drives": [{"name": "求知", "description": "想弄明白一切", "intensity": 3}]},
            headers=headers,
        )
        assert upd.status_code == 200 and upd.json()["drives"][0]["name"] == "求知"
        # 人格候选：pending → 批准后并入道体
        cand = await client.post(
            f"/v1/characters/{c['id']}/persona-candidates",
            json={"identity_baseline": {"core_traits": ["温柔"]}},
            headers=headers,
        )
        assert cand.status_code == 201 and cand.json()["status"] == "pending"
        approve = await client.post(
            f"/v1/characters/{c['id']}/versions/{cand.json()['id']}/decisions",
            json={"approve": True},
            headers=headers,
        )
        assert approve.json()["status"] == "approved"
        detail = await client.get(f"/v1/characters/{c['id']}", headers=headers)
        assert detail.json()["identity_baseline"]["core_traits"] == ["温柔"]
        # 已审核版本重复审核：409
        again = await client.post(
            f"/v1/characters/{c['id']}/versions/{cand.json()['id']}/decisions",
            json={"approve": False},
            headers=headers,
        )
        assert again.status_code == 409

    async def test_cross_domain_character_404(self, client: AsyncClient) -> None:
        alice = await login(client, "char8@example.com")
        c = await self._create(client, alice, {"name": "澜川"})
        await login(client, "char8-evil@example.com")
        resp = await client.get(f"/v1/characters/{c['id']}")
        assert resp.status_code in (401, 403, 404)


@pytest.mark.usefixtures("client")
class TestGatewayExecution:
    async def test_model_call_usage_and_assistant_message(self, client, db) -> None:
        """端到端：mock 流 → ModelCall usage/TTFT + assistant 权威消息。"""
        from sqlalchemy import select as sa

        from app.models import Message as M
        from app.models import ModelCall

        headers = await login(client, "gw1@example.com")
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, "idem-gw1")
        run_id = resp.json()["run"]["id"]
        assert await _wait_run_terminal(db, run_id) == "completed"
        async with db() as session:
            calls = list((await session.execute(sa(ModelCall))).scalars().all())
            assert len(calls) == 1
            assert calls[0].status == "succeeded"
            assert calls[0].usage == {
                "prompt_tokens": 12, "completion_tokens": 6, "total_tokens": 18,
            }
            assert calls[0].ttft_ms is not None
            asst = list(
                (
                    await session.execute(
                        sa(M).where(M.run_id == run_id, M.role == "assistant")
                    )
                ).scalars().all()
            )
            assert len(asst) == 1
            assert asst[0].content[0]["type"] == "text"
            assert asst[0].content[0]["status"] == "completed"

    async def test_no_endpoint_fails_closed(self, client, db) -> None:
        """无可用端点：Run 显式失败并留 error 事件，不静默降级（ADR-016）。"""
        from sqlalchemy import update

        from app.models import ProviderEndpoint

        async with db() as session:
            await session.execute(update(ProviderEndpoint).values(enabled=False))
            await session.commit()
        try:
            headers = await login(client, "gw2@example.com")
            cid = await _create_conversation(client, headers)
            resp = await _post_message(client, headers, cid, "idem-gw2")
            run_id = resp.json()["run"]["id"]
            assert await _wait_run_terminal(db, run_id) == "failed"
        finally:
            async with db() as session:
                await session.execute(update(ProviderEndpoint).values(enabled=True))
                await session.commit()
        from sqlalchemy import select as sa_select

        from app.models import RunEvent as RunEventModel

        async with db() as session:
            evs = list(
                (
                    await session.execute(
                        sa_select(RunEventModel)
                        .where(RunEventModel.run_id == run_id)
                        .order_by(RunEventModel.sequence)
                    )
                ).scalars().all()
            )
        types = [e.type for e in evs]
        assert "content.block.failed" in types
        failed = next(e for e in evs if e.type == "content.block.failed")
        assert failed.data["error"] == "no_endpoint_available"


@contextlib.asynccontextmanager
async def _test_db():
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    from tests.conftest import TEST_DATABASE_URL

    engine = create_async_engine(TEST_DATABASE_URL)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    try:
        yield factory
    finally:
        await engine.dispose()


@pytest.mark.usefixtures("client")
class TestMemories:
    async def _submit(self, client, headers, body) -> dict:
        resp = await client.post("/v1/memories/candidates", json=body, headers=headers)
        assert resp.status_code == 201, resp.text
        return resp.json()

    async def test_manual_candidate_review_flow(self, client: AsyncClient) -> None:
        headers = await login(client, "mem1@example.com")
        c = await self._submit(client, headers, {
            "kind": "relationship", "sensitivity": "high",
            "payload": {"text": "用户偏好深夜交谈", "note": "高影响关系结论"},
        })
        assert c["status"] == "pending"  # 敏感默认人工
        approve = await client.post(
            f"/v1/memories/candidates/{c['id']}/decisions",
            json={"approve": True}, headers=headers,
        )
        assert approve.status_code == 201
        assert approve.json()["status"] == "approved"
        assert approve.json()["created_memory_id"] is not None

    async def test_whitelist_auto_approve_only_low_sensitivity_fact(self, client) -> None:
        headers = await login(client, "mem2@example.com")
        auto = await self._submit(client, headers, {
            "kind": "fact", "sensitivity": "low", "source": "auto",
            "payload": {"text": "用户住在杭州"},
        })
        assert auto["status"] == "approved"  # 白名单自动写入
        assert auto["created_memory_id"] is not None
        # 高敏感自动候选不自动批准
        sensitive = await self._submit(client, headers, {
            "kind": "fact", "sensitivity": "high", "source": "auto",
            "payload": {"text": "敏感事实"},
        })
        assert sensitive["status"] == "pending"
        # 非 fact 也不自动
        pref = await self._submit(client, headers, {
            "kind": "preference", "sensitivity": "low", "source": "auto",
            "payload": {"text": "喜欢猫"},
        })
        assert pref["status"] == "pending"

    async def test_correction_with_etag_and_conflict(self, client) -> None:
        headers = await login(client, "mem3@example.com")
        c = await self._submit(client, headers, {
            "kind": "fact", "sensitivity": "low", "source": "auto",
            "payload": {"text": "原记录"},
        })
        mid = c["created_memory_id"]
        # 缺 If-Match：412
        no_header = await client.patch(
            f"/v1/memories/{mid}", json={"payload": {"text": "纠正"}}, headers=headers
        )
        assert no_header.status_code == 412
        # 错误版本：412
        stale = await client.patch(
            f"/v1/memories/{mid}", json={"payload": {"text": "纠正"}},
            headers={**headers, "If-Match": '"99"'},
        )
        assert stale.status_code == 412
        # 正确 ETag → v2
        ok = await client.patch(
            f"/v1/memories/{mid}", json={"payload": {"text": "纠正后"}},
            headers={**headers, "If-Match": '"1"'},
        )
        assert ok.status_code == 200
        body = ok.json()
        assert body["version"] == 2 and body["payload"]["text"] == "纠正后"
        assert body["resource_version"] == 2

    async def test_search_filters_and_isolation(self, client) -> None:
        headers = await login(client, "mem4@example.com")
        for text in ("杭州西湖", "喜欢跑步", "工作在软件园"):
            await self._submit(client, headers, {
                "kind": "fact", "sensitivity": "low", "source": "auto",
                "payload": {"text": text},
            })
        hits = await client.get("/v1/memories?q=西湖", headers=headers)
        data = hits.json()["data"]
        assert len(data) == 1 and "西湖" in data[0]["payload"]["text"]
        assert data[0]["recall"]["channel"] == "text"  # 召回证据
        # 跨域隔离
        other = await login(client, "mem4-evil@example.com")
        leak = await client.get("/v1/memories?q=西湖", headers=other)
        assert leak.json()["data"] == []

    async def test_delete_purges_content_and_blocks_recall(self, client) -> None:
        from sqlalchemy import select as sa_select

        from app.models import MemoryVersion as MV

        headers = await login(client, "mem5@example.com")
        c = await self._submit(client, headers, {
            "kind": "fact", "sensitivity": "low", "source": "auto",
            "payload": {"text": "将被删除的秘密"},
        })
        mid = c["created_memory_id"]
        resp = await client.delete(f"/v1/memories/{mid}", headers=headers)
        assert resp.status_code == 204
        # 不再召回
        after = await client.get("/v1/memories?q=秘密", headers=headers)
        assert after.json()["data"] == []
        # 墓碑：版本原文已清除（契约：墓碑不存被删原文）
        async with _test_db() as factory:
            async with factory() as session:
                versions = list(
                    (
                        await session.execute(sa_select(MV).where(MV.memory_id == mid))
                    ).scalars().all()
                )
            assert versions and all(v.payload == {} for v in versions)


@pytest.mark.usefixtures("client")
class TestBigCycle:
    async def test_manual_trigger_produces_pending_candidate(self, client, db) -> None:

        from app.models import BigCycle, CharacterVersion

        headers = await login(client, "bc1@example.com")
        c = await client.post("/v1/characters", json={"name": "砚舟"}, headers=headers)
        cid = c.json()["id"]
        await client.post(f"/v1/characters/{cid}/activations", headers=headers)
        # 先有一条对话消息供回顾
        conv = await _create_conversation(client, headers)
        await client.post(
            f"/v1/conversations/{conv}/messages",
            json={"content": [{"type": "text", "text": "最近在想什么呢"}], "start_run": False},
            headers={**headers, "Idempotency-Key": "idem-bc1"},
        )
        trigger = await client.post(f"/v1/characters/{cid}/big-cycles", headers=headers)
        assert trigger.status_code == 201, trigger.text
        cycle_id = trigger.json()["id"]
        # 轮询等待 cycle 终态
        import asyncio

        deadline = asyncio.get_event_loop().time() + 10
        cycle = None
        while asyncio.get_event_loop().time() < deadline:
            async with db() as session:
                cycle = await session.get(BigCycle, cycle_id)
                if cycle and cycle.status in {"completed", "failed"}:
                    break
            await asyncio.sleep(0.05)
        assert cycle is not None and cycle.status == "completed", (
            cycle.status if cycle else "missing"
        )
        assert cycle.produced_version_ids
        async with db() as session:
            version = await session.get(CharacterVersion, cycle.produced_version_ids[0])
            assert version is not None
            assert version.kind == "persona_candidate"
            assert version.status == "pending"  # 候选先于写入
            assert version.payload["source"] == "big_cycle"
            assert version.payload["proposal"]  # 模型真实输出


@pytest.mark.usefixtures("client")
class TestTools:
    async def _make_run(self, client, headers) -> str:
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, f"idem-tool-{_uuid4().split(chr(45))[0]}")
        await _wait_run_terminal_headers(client, headers)
        return resp.json()["run"]["id"]

    async def test_rule_bound_tool_executes_immediately(self, client: AsyncClient) -> None:
        headers = await login(client, "tool1@example.com")
        await client.get("/v1/tools", headers=headers)  # 触发内置登记
        run_id = await self._make_run(client, headers)
        resp = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "calculate", "input": {"expression": "(1+2)*3"}},
            headers={**headers, "Idempotency-Key": "tc-1"},
        )
        assert resp.status_code == 201, resp.text
        body = resp.json()
        assert body["status"] == "succeeded"
        assert body["output"]["result"] == 9.0

    async def test_side_effect_tool_requires_approval(self, client: AsyncClient) -> None:
        headers = await login(client, "tool2@example.com")
        await client.get("/v1/tools", headers=headers)
        run_id = await self._make_run(client, headers)
        created = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "record-note", "input": {"text": "重要笔记"}},
            headers=headers,
        )
        assert created.json()["status"] == "pending_approval"
        call_id = created.json()["id"]
        # 待审批列表可见
        pending = await client.get("/v1/tool-calls", headers=headers)
        assert any(c["id"] == call_id for c in pending.json()["data"])
        # 朱批 → 执行 → 真实副作用（记忆候选创建）
        approved = await client.post(
            f"/v1/tool-calls/{call_id}/approvals", json={"approve": True}, headers=headers
        )
        assert approved.json()["status"] == "succeeded"
        assert approved.json()["output"]["memory_candidate_id"]

    async def test_denied_tool_never_executes(self, client: AsyncClient) -> None:
        headers = await login(client, "tool3@example.com")
        await client.get("/v1/tools", headers=headers)
        run_id = await self._make_run(client, headers)
        created = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "record-note", "input": {"text": "被拒绝的笔记"}},
            headers=headers,
        )
        call_id = created.json()["id"]
        denied = await client.post(
            f"/v1/tool-calls/{call_id}/approvals", json={"approve": False}, headers=headers
        )
        assert denied.json()["status"] == "denied"
        assert denied.json()["output"] is None  # 未执行
        again = await client.post(
            f"/v1/tool-calls/{call_id}/approvals", json={"approve": True}, headers=headers
        )
        assert again.status_code == 409  # 拒绝不可逆

    async def test_schema_validation_rejects_bad_input(self, client) -> None:
        headers = await login(client, "tool4@example.com")
        await client.get("/v1/tools", headers=headers)
        run_id = await self._make_run(client, headers)
        missing = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "calculate", "input": {}}, headers=headers,
        )
        assert missing.status_code == 422
        wrong_type = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "calculate", "input": {"expression": 42}}, headers=headers,
        )
        assert wrong_type.status_code == 422

    async def test_unknown_tool_404(self, client) -> None:
        headers = await login(client, "tool5@example.com")
        run_id = await self._make_run(client, headers)
        resp = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "rm-rf", "input": {}}, headers=headers,
        )
        assert resp.status_code == 404

    async def test_idempotent_replay_returns_original(self, client) -> None:
        from fastapi.responses import JSONResponse  # noqa: F401

        headers = await login(client, "tool6@example.com")
        await client.get("/v1/tools", headers=headers)
        run_id = await self._make_run(client, headers)
        first = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "get-current-time", "input": {}},
            headers={**headers, "Idempotency-Key": "tc-replay"},
        )
        assert first.status_code == 201
        second = await client.post(
            f"/v1/runs/{run_id}/tool-calls",
            json={"tool_name": "get-current-time", "input": {}},
            headers={**headers, "Idempotency-Key": "tc-replay"},
        )
        assert second.status_code == 200
        assert second.json()["id"] == first.json()["id"]

    async def test_run_budget_hard_limit(self, client) -> None:
        headers = await login(client, "tool7@example.com")
        await client.get("/v1/tools", headers=headers)
        run_id = await self._make_run(client, headers)
        # max_tool_calls 默认 10：连续调用到上限后第 11 次被拒
        last_status = None
        for i in range(11):
            resp = await client.post(
                f"/v1/runs/{run_id}/tool-calls",
                json={"tool_name": "get-current-time", "input": {}},
                headers={**headers, "Idempotency-Key": f"tc-budget-{i}"},
            )
            last_status = resp.status_code
        assert last_status == 422


def _uuid4() -> str:
    from uuid import uuid4

    return str(uuid4())


async def _wait_run_terminal_headers(client, headers, limit_s: float = 10.0) -> None:
    import asyncio

    await asyncio.sleep(0)


@pytest.mark.usefixtures("client")
class TestMarketplace:
    async def _make_listing(self, client, headers, price=1000) -> str:
        resp = await client.post(
            "/v1/marketplace/listings",
            json={"title": "角色包·昭星", "price_minor": price}, headers=headers,
        )
        assert resp.status_code == 201, resp.text
        return resp.json()["id"]

    async def _make_order(self, client, headers, listing_id, key) -> dict:
        resp = await client.post(
            "/v1/marketplace/orders",
            json={"listing_ids": [listing_id]},
            headers={**headers, "Idempotency-Key": key},
        )
        assert resp.status_code in (200, 201), resp.text
        return resp.json()

    async def test_purchase_ledger_double_entry(self, client, db) -> None:
        headers = await login(client, "mkt1@example.com")
        lid = await self._make_listing(client, headers, price=1000)
        order = await self._make_order(client, headers, lid, "ord-1")
        assert order["status"] == "pending_payment" and order["total_minor"] == 1000
        paid = await client.post(
            f"/v1/marketplace/orders/{order['id']}/payment-confirmations", headers=headers
        )
        assert paid.json()["status"] == "paid"
        summary = await client.get("/v1/finance/summary", headers=headers)
        totals = summary.json()["totals_minor"]
        assert totals["charge"] == 1000
        assert totals["platform_fee"] == 100  # 10%
        assert totals["creator_payable"] == 900
        assert summary.json()["balanced"] is True

    async def test_order_idempotent_replay(self, client) -> None:
        headers = await login(client, "mkt2@example.com")
        lid = await self._make_listing(client, headers)
        first = await self._make_order(client, headers, lid, "ord-replay")
        second = await client.post(
            "/v1/marketplace/orders", json={"listing_ids": [lid]},
            headers={**headers, "Idempotency-Key": "ord-replay"},
        )
        assert second.status_code == 200
        assert second.json()["id"] == first["id"]

    async def test_refund_writes_reversal_entries(self, client) -> None:
        headers = await login(client, "mkt3@example.com")
        lid = await self._make_listing(client, headers, price=2000)
        order = await self._make_order(client, headers, lid, "ord-refund")
        await client.post(
            f"/v1/marketplace/orders/{order['id']}/payment-confirmations", headers=headers
        )
        refunded = await client.post(
            f"/v1/marketplace/orders/{order['id']}/refunds", headers=headers
        )
        assert refunded.json()["status"] == "refunded"
        # 重复退款幂等
        again = await client.post(
            f"/v1/marketplace/orders/{order['id']}/refunds", headers=headers
        )
        assert again.json()["status"] == "refunded"
        summary = await client.get("/v1/finance/summary", headers=headers)
        totals = summary.json()["totals_minor"]
        # charge=2000, reversal=-2000 → 净 0；fee/payable 各自正负抵消
        assert totals.get("charge", 0) + totals.get("reversal", 0) == 0
        assert summary.json()["balanced"] is True

    async def test_webhook_signature_and_idempotency(self, client) -> None:
        import hashlib
        import hmac as hmac_mod
        import os

        headers = await login(client, "mkt4@example.com")
        lid = await self._make_listing(client, headers, price=500)
        order = await self._make_order(client, headers, lid, "ord-wh")
        payload = json.dumps({
            "provider_event_id": "evt-001", "type": "payment.succeeded",
            "order_id": order["id"],
        })
        secret = os.environ.get("HUNJING_WEBHOOK_SECRET", "whsec_test")
        os.environ["HUNJING_WEBHOOK_SECRET"] = "whsec_test"
        sig = hmac_mod.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
        # 错误签名：401
        bad = await client.post(
            "/v1/webhooks/v1/payments/mock", content=payload,
            headers={"X-Signature": "deadbeef", "Content-Type": "application/json"},
        )
        assert bad.status_code == 401
        # 正确签名：200 处理
        ok = await client.post(
            "/v1/webhooks/v1/payments/mock", content=payload,
            headers={"X-Signature": sig, "Content-Type": "application/json"},
        )
        assert ok.status_code == 200 and ok.json()["result"] == "processed"
        # 重复投递：幂等 duplicate
        dup = await client.post(
            "/v1/webhooks/v1/payments/mock", content=payload,
            headers={"X-Signature": sig, "Content-Type": "application/json"},
        )
        assert dup.json()["result"] == "duplicate"

    async def test_preservation_readonly_and_recovery(self, client) -> None:
        headers = await login(client, "mkt5@example.com")
        await client.post(
            "/v1/finance/preservation", json={"preserved": True}, headers=headers
        )
        # 状态变更被拒绝（保全只读）
        blocked = await client.post(
            "/v1/conversations", json={"title": "保全期"}, headers=headers
        )
        assert blocked.status_code == 403
        assert "保全" in blocked.json()["detail"]
        # 读取不受影响：数据仍在
        listing_resp = await client.post(  # 验证写拦截一致
            "/v1/marketplace/listings", json={"title": "x"}, headers=headers
        )
        assert listing_resp.status_code == 403
        # 恢复：写恢复
        await client.post(
            "/v1/finance/preservation", json={"preserved": False}, headers=headers
        )
        recovered = await client.post(
            "/v1/conversations", json={"title": "恢复"}, headers=headers
        )
        assert recovered.status_code == 201


@pytest.mark.usefixtures("client")
class TestCommunity:
    async def _publish_post(self, client, headers, title="昭星使用心得") -> str:
        resp = await client.post(
            "/v1/community/contents",
            json={"kind": "post", "title": title, "text": "很好用"}, headers=headers,
        )
        assert resp.status_code == 201, resp.text
        return resp.json()["id"]

    async def test_publish_and_public_listing(self, client) -> None:
        headers = await login(client, "cm1@example.com")
        cid = await self._publish_post(client, headers)
        listing = await client.get("/v1/community/contents", headers=headers)
        assert any(c["id"] == cid for c in listing.json()["data"])

    async def test_character_release_snapshot_has_no_private_data(self, client) -> None:
        """Release 快照仅公开字段：结构上不含记忆/关系/对话（契约）。"""
        headers = await login(client, "cm2@example.com")
        c = await client.post(
            "/v1/characters", json={"name": "霁月"}, headers=headers
        )
        cid = c.json()["id"]
        await client.post(f"/v1/characters/{cid}/activations", headers=headers)
        # 给角色添加一条私密记忆（不应出现在发布物）
        await client.post(
            "/v1/memories/candidates",
            json={"kind": "fact", "sensitivity": "high", "character_id": cid,
                  "payload": {"text": "绝对私密的事实"}},
            headers=headers,
        )
        release = await client.post(
            "/v1/community/contents",
            json={"kind": "character_release", "title": "霁月 · 发布",
                  "character_id": cid},
            headers=headers,
        )
        assert release.status_code == 201, release.text
        body = release.json()["body"]
        assert body["character_name"] == "霁月"
        assert "public_baseline" in body
        # 私密字段结构性缺席
        import json as _json

        assert "绝对私密的事实" not in _json.dumps(body, ensure_ascii=False)
        assert "memory" not in _json.dumps(body)

    async def test_comments_likes_saves(self, client) -> None:
        headers = await login(client, "cm3@example.com")
        cid = await self._publish_post(client, headers, "互动测试")
        c1 = await client.post(
            f"/v1/community/contents/{cid}/comments",
            json={"body": "第一条"}, headers=headers,
        )
        assert c1.status_code == 201
        c2 = await client.post(
            f"/v1/community/contents/{cid}/comments",
            json={"body": "回复", "parent_comment_id": c1.json()["id"]}, headers=headers,
        )
        assert c2.status_code == 201 and c2.json()["parent_comment_id"] == c1.json()["id"]
        like = await client.post(
            f"/v1/community/contents/{cid}/reactions", json={"kind": "like"}, headers=headers
        )
        assert like.json()["like"] == "1"
        dup = await client.post(
            f"/v1/community/contents/{cid}/reactions", json={"kind": "like"}, headers=headers
        )
        assert dup.status_code == 409  # uk 防重复
        save = await client.post(
            f"/v1/community/contents/{cid}/reactions", json={"kind": "save"}, headers=headers
        )
        assert save.json()["save"] == "1"
        await client.request(
            "DELETE",
            f"/v1/community/contents/{cid}/reactions",
            json={"kind": "like"}, headers=headers,
        )
        detail = await client.get(f"/v1/community/contents/{cid}", headers=headers)
        assert detail.json()["reactions"] == {"save": 1}  # like 已取消

    async def test_follow_counts(self, client) -> None:
        a = await login(client, "cm4a@example.com")
        b = await login(client, "cm4b@example.com")
        # b 的域 ID：发布一条后从列表拿 author_domain_id
        cid = await self._publish_post(client, b)
        listing = await client.get("/v1/community/contents", headers=b)
        b_domain = next(
            c["author_domain_id"] for c in listing.json()["data"] if c["id"] == cid
        )
        follow = await client.post(
            "/v1/community/follows", json={"followee_domain_id": b_domain}, headers=a
        )
        assert follow.status_code == 201
        assert follow.json()["following"] == 1
        counts_b = await client.get("/v1/community/follows/counts", headers=b)
        assert counts_b.json()["followers"] == 1

    async def test_report_creates_moderation_case(self, client) -> None:
        a = await login(client, "cm5a@example.com")
        cid = await self._publish_post(client, a)
        b = await login(client, "cm5b@example.com")
        report = await client.post(
            "/v1/community/reports",
            json={"target_type": "content", "target_id": cid, "reason": "违规内容"},
            headers=b,
        )
        assert report.status_code == 201
        assert report.json()["case_id"] and report.json()["status"] == "pending"

    async def test_withdraw_propagates_invisibility(self, client) -> None:
        a = await login(client, "cm6a@example.com")
        cid = await self._publish_post(client, a)
        b = await login(client, "cm6b@example.com")
        await client.post(
            f"/v1/community/contents/{cid}/comments", json={"body": "x"}, headers=b
        )
        withdrawn = await client.post(
            f"/v1/community/contents/{cid}/withdrawals", headers=a
        )
        assert withdrawn.json()["status"] == "removed"
        # 列表不可见
        listing = await client.get("/v1/community/contents", headers=b)
        assert all(c["id"] != cid for c in listing.json()["data"])
        # 详情 404
        detail = await client.get(f"/v1/community/contents/{cid}", headers=b)
        assert detail.status_code == 404
        # 评论不可见
        comments = await client.get(f"/v1/community/contents/{cid}/comments", headers=b)
        assert comments.status_code == 404

    async def test_non_author_cannot_withdraw(self, client) -> None:
        a = await login(client, "cm7a@example.com")
        cid = await self._publish_post(client, a)
        b = await login(client, "cm7b@example.com")
        resp = await client.post(f"/v1/community/contents/{cid}/withdrawals", headers=b)
        assert resp.status_code == 404  # 非作者：fail-closed


@pytest.mark.usefixtures("client")
class TestWorkspaces:
    async def _domain_id_of(self, client, headers) -> str:
        """经发布列表拿自己的域 ID（稳定取法）。"""
        resp = await client.post(
            "/v1/community/contents",
            json={"kind": "post", "title": f"定位-{headers['Cookie'][-8:]}"},
            headers=headers,
        )
        cid = resp.json()["id"]
        listing = await client.get("/v1/community/contents", headers=headers)
        return next(
            c["author_domain_id"] for c in listing.json()["data"] if c["id"] == cid
        )

    async def test_full_collaboration_flow(self, client) -> None:
        owner = await login(client, "ws-owner@example.com")
        editor = await login(client, "ws-editor@example.com")
        reviewer = await login(client, "ws-reviewer@example.com")
        viewer = await login(client, "ws-viewer@example.com")
        ws = await client.post(
            "/v1/workspaces", json={"name": "同契台"}, headers=owner
        )
        assert ws.status_code == 201, ws.text
        ws_id = ws.json()["id"]
        # 邀请三角色
        invs = {}
        for name, hdrs, role in [
            ("editor", editor, "editor"), ("reviewer", reviewer, "reviewer"),
            ("viewer", viewer, "viewer"),
        ]:
            did = await self._domain_id_of(client, hdrs)
            inv = await client.post(
                f"/v1/workspaces/{ws_id}/invitations",
                json={"invitee_domain_id": did, "role": role}, headers=owner,
            )
            assert inv.status_code == 201, inv.text
            acc = await client.post(
                f"/v1/workspaces/invitations/{inv.json()['id']}/acceptances", headers=hdrs
            )
            assert acc.status_code == 201 and acc.json()["role"] == role
            invs[name] = acc.json()
        members = await client.get(f"/v1/workspaces/{ws_id}/members", headers=owner)
        assert len(members.json()["data"]) == 4

        # 项目与版本评审：editor 提交 → reviewer 批准 → owner 发布
        project = await client.post(
            f"/v1/workspaces/{ws_id}/projects", json={"name": "世界观设定"}, headers=editor
        )
        assert project.status_code == 201
        pid = project.json()["id"]
        v1 = await client.post(
            f"/v1/workspaces/projects/{pid}/versions",
            json={"payload": {"text": "第一版"}}, headers=editor,
        )
        assert v1.status_code == 201 and v1.json()["version"] == 1
        assert v1.json()["status"] == "submitted"
        # viewer 不能提交
        denied = await client.post(
            f"/v1/workspaces/projects/{pid}/versions",
            json={"payload": {"text": "越权"}}, headers=viewer,
        )
        assert denied.status_code == 403
        # editor 不能评审（三权分立）
        self_review = await client.post(
            f"/v1/workspaces/project-versions/{v1.json()['id']}/reviews",
            json={"approve": True}, headers=editor,
        )
        assert self_review.status_code == 403
        # 未批准前 owner 不能发布
        early = await client.post(
            f"/v1/workspaces/projects/{pid}/publications", headers=owner
        )
        assert early.status_code == 422
        # reviewer 批准 → 设为当前版本
        approved = await client.post(
            f"/v1/workspaces/project-versions/{v1.json()['id']}/reviews",
            json={"approve": True}, headers=reviewer,
        )
        assert approved.json()["status"] == "approved"
        # owner 发布
        published = await client.post(
            f"/v1/workspaces/projects/{pid}/publications", headers=owner
        )
        assert published.json()["status"] == "published"

    async def test_rejection_path(self, client) -> None:
        owner = await login(client, "ws2-owner@example.com")
        editor = await login(client, "ws2-editor@example.com")
        reviewer = await login(client, "ws2-reviewer@example.com")
        ws = await client.post("/v1/workspaces", json={"name": "拒绝路径"}, headers=owner)
        ws_id = ws.json()["id"]
        for hdrs, role in [(editor, "editor"), (reviewer, "reviewer")]:
            did = await self._domain_id_of(client, hdrs)
            inv = await client.post(
                f"/v1/workspaces/{ws_id}/invitations",
                json={"invitee_domain_id": did, "role": role}, headers=owner,
            )
            await client.post(
                f"/v1/workspaces/invitations/{inv.json()['id']}/acceptances", headers=hdrs
            )
        project = await client.post(
            f"/v1/workspaces/{ws_id}/projects", json={"name": "P"}, headers=editor
        )
        pid = project.json()["id"]
        v1 = await client.post(
            f"/v1/workspaces/projects/{pid}/versions",
            json={"payload": {"text": "v1"}}, headers=editor,
        )
        rejected = await client.post(
            f"/v1/workspaces/project-versions/{v1.json()['id']}/reviews",
            json={"approve": False}, headers=reviewer,
        )
        assert rejected.json()["status"] == "rejected"
        # 拒绝后重审 409
        again = await client.post(
            f"/v1/workspaces/project-versions/{v1.json()['id']}/reviews",
            json={"approve": True}, headers=reviewer,
        )
        assert again.status_code == 409
        # 被拒版本不能作为当前版本发布
        pub = await client.post(
            f"/v1/workspaces/projects/{pid}/publications", headers=owner
        )
        assert pub.status_code == 422

    async def test_member_removal_and_isolation(self, client) -> None:
        owner = await login(client, "ws3-owner@example.com")
        editor = await login(client, "ws3-editor@example.com")
        outsider = await login(client, "ws3-outsider@example.com")
        ws = await client.post("/v1/workspaces", json={"name": "隔离"}, headers=owner)
        ws_id = ws.json()["id"]
        did = await self._domain_id_of(client, editor)
        inv = await client.post(
            f"/v1/workspaces/{ws_id}/invitations",
            json={"invitee_domain_id": did, "role": "editor"}, headers=owner,
        )
        member = await client.post(
            f"/v1/workspaces/invitations/{inv.json()['id']}/acceptances", headers=editor
        )
        # 外人访问 404
        leak = await client.get(f"/v1/workspaces/{ws_id}/members", headers=outsider)
        assert leak.status_code == 404
        # owner 移除成员
        removed = await client.request(
            "DELETE",
            f"/v1/workspaces/{ws_id}/members/{member.json()['id']}",
            headers=owner,
        )
        assert removed.status_code == 204
        # 被移除后访问失效
        after = await client.get(f"/v1/workspaces/{ws_id}/members", headers=editor)
        assert after.status_code == 404


@pytest.mark.usefixtures("client")
class TestPlugins:
    MANIFEST_V1 = {
        "permissions": {"network": ["api.example.com"], "data": ["public_web"]},
        "response_template": "已按声明式规则处理",
    }
    MANIFEST_V2 = {
        "permissions": {
            "network": ["api.example.com"],
            "data": ["public_web", "local_files"],  # 新增权限
        },
        "response_template": "v2 处理",
    }

    async def _publish(self, client, headers, version, manifest) -> None:
        resp = await client.post(
            "/v1/plugins/versions",
            json={"name": "example.fetch", "kind": "declarative",
                  "version": version, "manifest": manifest},
            headers=headers,
        )
        assert resp.status_code == 201, resp.text

    async def test_three_state_separation(self, client) -> None:
        """购买≠安装≠启用≠授权：安装默认禁用、零授权；启用显式授权。"""
        publisher = await login(client, "pl-pub@example.com")
        await self._publish(client, publisher, "1.0.0", self.MANIFEST_V1)
        user = await login(client, "pl-user@example.com")
        # 未安装直接调用：403
        before = await client.post(
            "/v1/plugins/example.fetch/invocations", json={"input": {}}, headers=user
        )
        assert before.status_code == 403
        # 安装：默认 disabled
        installation = await client.post(
            "/v1/plugins/example.fetch/installations",
            json={"version": "1.0.0"}, headers=user,
        )
        assert installation.status_code == 201
        assert installation.json()["enabled"] is False
        assert installation.json()["granted_permissions"] == []
        # 已安装未启用：调用仍拒
        disabled = await client.post(
            "/v1/plugins/example.fetch/invocations", json={"input": {}}, headers=user
        )
        assert disabled.status_code == 403
        # 启用：显式授权子集
        enabled = await client.post(
            f"/v1/plugin-installations/{installation.json()['id']}/enablings",
            json={"granted_permissions": ["network:api.example.com"]}, headers=user,
        )
        assert enabled.status_code == 201 and enabled.json()["enabled"] is True
        # 调用成功（manifest 模板回填）
        invoked = await client.post(
            "/v1/plugins/example.fetch/invocations", json={"input": {}}, headers=user
        )
        assert invoked.status_code == 200
        assert invoked.json()["response"] == "已按声明式规则处理"
        assert invoked.json()["runtime"] == "declarative"

    async def test_grant_outside_manifest_rejected(self, client) -> None:
        publisher = await login(client, "pl2-pub@example.com")
        await self._publish(client, publisher, "1.0.0", self.MANIFEST_V1)
        user = await login(client, "pl2-user@example.com")
        installation = await client.post(
            "/v1/plugins/example.fetch/installations",
            json={"version": "1.0.0"}, headers=user,
        )
        overreach = await client.post(
            f"/v1/plugin-installations/{installation.json()['id']}/enablings",
            json={"granted_permissions": ["device:camera"]}, headers=user,
        )
        assert overreach.status_code == 422  # 授权越界

    async def test_upgrade_requires_reconfirmation_of_new_permissions(self, client) -> None:
        publisher = await login(client, "pl3-pub@example.com")
        await self._publish(client, publisher, "1.0.0", self.MANIFEST_V1)
        await self._publish(client, publisher, "2.0.0", self.MANIFEST_V2)
        user = await login(client, "pl3-user@example.com")
        installation = await client.post(
            "/v1/plugins/example.fetch/installations",
            json={"version": "1.0.0"}, headers=user,
        )
        iid = installation.json()["id"]
        await client.post(
            f"/v1/plugin-installations/{iid}/enablings",
            json={"granted_permissions": ["network:api.example.com"]}, headers=user,
        )
        # 升级不确认新增权限：422
        blocked = await client.post(
            f"/v1/plugin-installations/{iid}/upgrades",
            json={"version": "2.0.0"}, headers=user,
        )
        assert blocked.status_code == 422
        assert "重新确认" in blocked.json()["detail"]
        # 带新增授权：成功
        ok = await client.post(
            f"/v1/plugin-installations/{iid}/upgrades",
            json={"version": "2.0.0",
                  "granted_permissions": ["network:api.example.com", "data:local_files"]},
            headers=user,
        )
        assert ok.status_code == 201
        assert ok.json()["granted_permissions"] == [
            "data:local_files", "network:api.example.com",
        ]

    async def test_withdrawal_blocks_new_invocations_keeps_history(self, client, db) -> None:
        from sqlalchemy import select as sa_select

        from app.models import AuditEvent

        publisher = await login(client, "pl4-pub@example.com")
        await self._publish(client, publisher, "1.0.0", self.MANIFEST_V1)
        user = await login(client, "pl4-user@example.com")
        installation = await client.post(
            "/v1/plugins/example.fetch/installations",
            json={"version": "1.0.0"}, headers=user,
        )
        await client.post(
            f"/v1/plugin-installations/{installation.json()['id']}/enablings",
            json={"granted_permissions": ["network:api.example.com"]}, headers=user,
        )
        await client.post(
            "/v1/plugins/example.fetch/invocations", json={"input": {}}, headers=user
        )
        # 撤架（发布者）
        withdrawn = await client.post(
            "/v1/plugins/example.fetch/withdrawals", headers=publisher
        )
        assert withdrawn.json()["status"] == "removed"
        # 新调用阻断
        blocked = await client.post(
            "/v1/plugins/example.fetch/invocations", json={"input": {}}, headers=user
        )
        assert blocked.status_code == 403
        # 新安装阻断
        new_install = await client.post(
            "/v1/plugins/example.fetch/installations",
            json={"version": "1.0.0"}, headers=await login(client, "pl4-new@example.com"),
        )
        assert new_install.status_code == 404
        # 历史审计保留
        async with db() as session:
            audits = list(
                (
                    await session.execute(
                        sa_select(AuditEvent).where(
                            AuditEvent.aggregate_id == "example.fetch",
                            AuditEvent.action == "plugin.invoked",
                        )
                    )
                ).scalars().all()
            )
            assert len(audits) == 1  # 撤架前那次调用仍在审计链


@pytest.mark.usefixtures("client")
class TestMedia:
    async def _upload_audio(self, client, headers) -> str:
        up = await client.post(
            "/v1/uploads",
            json={"filename": "memo.wav", "media_type": "audio/wav"}, headers=headers,
        )
        assert up.status_code == 201, up.text
        data = await client.request(
            "PUT", f"/v1/uploads/{up.json()['id']}/data",
            content=b"RIFF....fake-wav-bytes",
            headers={**headers, "Content-Type": "application/octet-stream"},
        )
        assert data.status_code == 200, data.text
        done = await client.post(
            f"/v1/uploads/{up.json()['id']}/completions", headers=headers
        )
        assert done.status_code == 201 and done.json()["status"] == "ready"
        return done.json()["id"]

    async def test_upload_three_step_flow(self, client) -> None:
        headers = await login(client, "md1@example.com")
        file_id = await self._upload_audio(client, headers)
        meta = await client.get(f"/v1/files/{file_id}", headers=headers)
        assert meta.json()["size_bytes"] == len(b"RIFF....fake-wav-bytes")
        assert meta.json()["sha256"].startswith("sha256:")
        content = await client.get(f"/v1/files/{file_id}/content", headers=headers)
        assert content.status_code == 200 and content.content == b"RIFF....fake-wav-bytes"

    async def test_upload_rejects_bad_mime(self, client) -> None:
        headers = await login(client, "md2@example.com")
        resp = await client.post(
            "/v1/uploads",
            json={"filename": "x.exe", "media_type": "application/x-executable"},
            headers=headers,
        )
        assert resp.status_code == 422

    async def test_asr_job_with_mock_provider(self, client, db) -> None:
        import asyncio

        from app.models import MediaJob as MJ

        headers = await login(client, "md3@example.com")
        file_id = await self._upload_audio(client, headers)
        job = await client.post(
            "/v1/media/jobs",
            json={"kind": "asr", "input": {"file_id": file_id}}, headers=headers,
        )
        assert job.status_code == 201, job.text
        job_id = job.json()["id"]
        deadline = asyncio.get_event_loop().time() + 10
        final = None
        while asyncio.get_event_loop().time() < deadline:
            async with db() as session:
                row = await session.get(MJ, job_id)
                if row and row.status in {"succeeded", "failed", "cancelled"}:
                    final = row
                    break
            await asyncio.sleep(0.05)
        assert final is not None and final.status == "succeeded", (
            final.error if final else "missing"
        )
        assert final.output["transcript"]

    async def test_asr_requires_ready_audio(self, client) -> None:
        headers = await login(client, "md4@example.com")
        up = await client.post(
            "/v1/uploads",
            json={"filename": "t.txt", "media_type": "text/plain"}, headers=headers,
        )
        await client.request(
            "PUT", f"/v1/uploads/{up.json()['id']}/data",
            content=b"hello", headers=headers,
        )
        job = await client.post(
            "/v1/media/jobs",
            json={"kind": "asr", "input": {"file_id": "file_nonexistent"}}, headers=headers,
        )
        assert job.status_code == 404  # 未就绪/不存在文件不可被引用

    async def test_tts_requires_voice_consent(self, client) -> None:
        headers = await login(client, "md5@example.com")
        denied = await client.post(
            "/v1/media/jobs",
            json={"kind": "tts", "input": {"text": "你好"}}, headers=headers,
        )
        assert denied.status_code == 403  # 无声契拒绝
        consent = await client.post(
            "/v1/voice-consents", json={"purpose": "voice_clone"}, headers=headers
        )
        assert consent.status_code == 201
        ok = await client.post(
            "/v1/media/jobs",
            json={"kind": "tts", "input": {"text": "你好"}}, headers=headers
        )
        assert ok.status_code == 201
        # 撤回后再次拒绝
        revoke = await client.post(
            f"/v1/voice-consents/{consent.json()['id']}/revocations", headers=headers
        )
        assert revoke.json()["revoked_at"] is not None
        after = await client.post(
            "/v1/media/jobs",
            json={"kind": "tts", "input": {"text": "再见"}}, headers=headers
        )
        assert after.status_code == 403


@pytest.mark.usefixtures("client")
class TestOps:
    async def test_diagnostic_dual_auth_flow(self, client, db) -> None:
        headers = await login(client, "ops1@example.com")
        cid = await _create_conversation(client, headers)
        resp = await _post_message(client, headers, cid, "idem-ops1")
        run_id = resp.json()["run"]["id"]
        await _wait_run_terminal(db, run_id)
        # 签发：明文只此一次
        grant = await client.post(
            f"/v1/runs/{run_id}/diagnostic-grants",
            json={"purpose": "user_inspection"}, headers=headers,
        )
        assert grant.status_code == 201, grant.text
        token = grant.json()["token"]
        # 无令牌：401
        no_token = await client.get(f"/v1/runs/{run_id}/diagnostics", headers=headers)
        assert no_token.status_code == 401
        # 用途不符：403
        wrong = await client.get(
            f"/v1/runs/{run_id}/diagnostics",
            headers={**headers, "Hunjing-Diagnostic-Token": token,
                     "Hunjing-Diagnostic-Purpose": "other"},
        )
        assert wrong.status_code == 403
        # 正确读取：观测摘要 + no-store + 单次消耗
        ok = await client.get(
            f"/v1/runs/{run_id}/diagnostics",
            headers={**headers, "Hunjing-Diagnostic-Token": token,
                     "Hunjing-Diagnostic-Purpose": "user_inspection"},
        )
        assert ok.status_code == 200
        assert ok.headers["cache-control"] == "no-store, private"
        body = ok.json()
        assert body["run"]["id"] == run_id and body["turns"]
        # 单次：重复使用 401
        replay = await client.get(
            f"/v1/runs/{run_id}/diagnostics",
            headers={**headers, "Hunjing-Diagnostic-Token": token,
                     "Hunjing-Diagnostic-Purpose": "user_inspection"},
        )
        assert replay.status_code == 401

    async def test_admin_moderation_resolution(self, client, db) -> None:
        from app.models import User as UserModel

        # 提权管理员（开发档直改）
        admin_headers = await login(client, "ops-admin@example.com")
        ordinary = await login(client, "ops-ordinary@example.com")
        from sqlalchemy import select as sa_sel

        async with db() as session:
            admin_user = (
                await session.execute(
                    sa_sel(UserModel).where(UserModel.email == "ops-admin@example.com")
                )
            ).scalar_one()
            admin_user.role = "admin"
            await session.commit()
        # 普通用户访问管理面：403
        forbidden = await client.get("/v1/admin/moderation-cases", headers=ordinary)
        assert forbidden.status_code == 403
        # 举报→处置→联动下架
        author = await login(client, "ops-author@example.com")
        pub = await client.post(
            "/v1/community/contents",
            json={"kind": "post", "title": "待处置"}, headers=author,
        )
        content_id = pub.json()["id"]
        report = await client.post(
            "/v1/community/reports",
            json={"target_type": "content", "target_id": content_id, "reason": "违规"},
            headers=ordinary,
        )
        case_id = report.json()["case_id"]
        pending = await client.get(
            "/v1/admin/moderation-cases", headers=admin_headers
        )
        assert any(c["case_id"] == case_id for c in pending.json()["data"])
        resolved = await client.post(
            f"/v1/admin/moderation-cases/{case_id}/resolutions",
            json={"action": "content_removed"}, headers=admin_headers,
        )
        assert resolved.json()["status"] == "resolved"
        # 下架传播：内容 404
        gone = await client.get(f"/v1/community/contents/{content_id}", headers=ordinary)
        assert gone.status_code == 404

    async def test_support_bundle_and_metrics(self, client) -> None:
        headers = await login(client, "ops3@example.com")
        bundle = await client.post("/v1/support/bundles", headers=headers)
        assert bundle.status_code == 200
        body = bundle.json()
        assert "runs_by_status" in body and "messages_total" in body
        assert "detail" not in body
        metrics = await client.get("/v1/metrics")
        assert metrics.status_code == 200
        assert "hunjing_runs_total" in metrics.text
        ready = await client.get("/v1/readyz")
        assert ready.json()["status"] == "ready"