group_fqcd_jr/app/service/agent_persistence_service.py

from datetime import UTC, datetime
from decimal import Decimal
from uuid import uuid4

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.contracts import AgentResult, DomainEvent
from app.core.errors import RunLeaseLostError
from app.model.audit import InteractionAudit
from app.model.conversation import ConversationMessage
from app.model.platform import AgentRun, DomainEventOutbox, RequestIdempotency

#: 治理层追加免责声明时使用的分隔形状（`app/service/agent/governance.py` 里定义）。
#: 这里只用于**审计留痕**，不参与任何判定：判据是"末尾是否出现这个形状"。
_GOVERNANCE_APPEND_MARKERS: tuple[str, ...] = ("\n\n本内容仅为投资分析参考",)


def _governance_rewrote(result: AgentResult) -> bool:
    """治理层是否改写过这次输出（用于审计）。

    两条可观测痕迹（都不改协议、只读结果本身）：
    1. **追加了固定免责声明**：正文末尾出现治理层使用的分隔形状；
    2. **拦截并替换**：命中禁用词/硬规则时治理层会把回复换成安全话术并置 `transfer_required`。

    保守取值：任一条成立即记 True。它只是审计信息，判错方向的代价是"多标了一次"，
    不会影响业务行为——因此宁可宽一点，也不为了精确而改动治理协议。
    """
    text = result.result.text or ""
    appended = any(text.endswith(marker) or marker in text
                   for marker in _GOVERNANCE_APPEND_MARKERS)
    return appended or bool(result.result.transfer_required)


class AgentPersistenceService:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def complete_run(
        self, run_id: str, result: AgentResult, memory_extraction_requested: bool = True,
        *, worker_id: str | None = None,
    ) -> int:
        now = datetime.now(UTC).replace(tzinfo=None)
        async with self.session.begin():
            run = await self.session.scalar(
                select(AgentRun).where(AgentRun.run_id == run_id).with_for_update()
            )
            if run is None:
                raise ValueError("run not found")
            if result.run_id != run_id:
                raise ValueError("result belongs to another run")
            if run.status == "succeeded" and run.result_message_id is not None:
                return run.result_message_id
            if worker_id is not None and (
                run.status != "running" or run.worker_id != worker_id
                or run.locked_until is None or run.locked_until <= now
            ):
                raise RunLeaseLostError("运行租约失效或已取消")
            if run.status not in {"queued", "running"}:
                raise RunLeaseLostError("不能覆盖运行终态")
            message = ConversationMessage(
                session_id=run.session_id, customer_id=run.user_id, portal="agent",
                role="assistant", content=result.result.text,
                trace_id=run.trace_id, created_at=now,
                intent=result.result.intent.intent if result.result.intent else None,
                confidence=(Decimal(str(result.result.intent.confidence))
                            if result.result.intent else None),
                source_references=[ref.model_dump(mode="json")
                                   for ref in result.result.source_references],
                # `tool_calls` 是**唯一能承载附加信息的现成 JSON 列**（`conversation_message`
                # 没有 `transfer_required` 列，加列要迁移，而规则 4 禁止改既有字段定义）。
                # 因此把「转人工标记」作为 `calls` 的**兄弟键**放进来：
                #   {"calls": [...], "transfer_required": bool, "transfer_reason": str|None}
                # 之所以必须落库：`docs/05` §6.3 规定 `GET /agent-runs/{run_id}` 的
                # `result` 里要有 `transfer_required` / `transfer_reason`，而它此前
                # **既没落库也没出参** —— 前端只能靠"回答里是否含兜底话术开头"来猜要不要转人工
                # （`docs/24` 自己把这称为权宜之计）。落库后读写两侧才有同一份真相。
                # 读侧允许 `calls` 是裸列表（历史行），见 `RunQueryService.get`。
                tool_calls={
                    "calls": [call.model_dump(mode="json")
                              for call in result.result.tool_calls],
                    "transfer_required": bool(result.result.transfer_required),
                    "transfer_reason": result.result.transfer_reason,
                },
            )
            self.session.add(message)
            await self.session.flush()
            run.result_message_id = message.id
            run.status = "succeeded"
            run.result_version = 1
            run.completed_at = now
            run.updated_at = now
            run.locked_until = None
            self.session.add(InteractionAudit(
                actor_type="agent", actor_id=run.user_id, target_customer_id=run.user_id,
                session_id=run.session_id, portal="agent", action_type="agent.run_completed",
                # `agent_type` 必须落进审计：治理层（`PlatformGovernance.review`）会**改写对外
                # 输出**（追加固定免责声明、命中禁用词时整条替换成安全话术），事后要能回答
                # "这次改写是哪个 Agent 触发的、改写到了什么程度"。
                # `governance_rewrite` 记录治理是否动过输出：正文里出现固定话术的追加形状，
                # 或该次运行被标记为需转人工（拦截分支会置 `transfer_required`）。
                # 不改表结构：`detail` 是 JSON 列，加键不需要迁移（AGENTS.md 规则 4）。
                detail={
                    "run_id": run_id,
                    "result_message_id": message.id,
                    "agent_type": run.agent_type,
                    "governance_rewrite": _governance_rewrote(result),
                },
                created_at=now,
            ))
            await self.session.execute(
                update(RequestIdempotency)
                .where(RequestIdempotency.id == run.idempotency_id)
                .values(status="completed", result_message_id=message.id, updated_at=now)
            )
            events = [DomainEvent(
                event_id=str(uuid4()), event_type="agent.run_completed", aggregate_type="agent_run",
                aggregate_id=run_id, trace_id=run.trace_id,
                payload={"run_id": run_id}, occurred_at=now,
            )]
            if memory_extraction_requested:
                events.append(DomainEvent(
                    event_id=str(uuid4()), event_type="memory.extraction_requested",
                    aggregate_type="agent_run", aggregate_id=run_id, trace_id=run.trace_id,
                    payload={"run_id": run_id, "message_id": message.id,
                             "customer_id": run.user_id}, occurred_at=now,
                ))
            for event in events:
                self.session.add(DomainEventOutbox(
                    event_id=event.event_id, event_type=event.event_type,
                    aggregate_type=event.aggregate_type, aggregate_id=event.aggregate_id,
                    trace_id=event.trace_id, payload=event.payload, occurred_at=now,
                    created_at=now, updated_at=now,
                ))
            return message.id