hunjing/backend/app/services/runtime.py

"""Run/Event 运行时服务。"""

from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.errors import AppError, current_trace_id
from app.core.security import redact_value
from app.core.time import utc_now
from app.db.session import AsyncSessionLocal
from app.models.runtime import Event, Run
from app.repositories import runtime as runtime_repo
from app.repositories.json_utils import dumps
from app.schemas.common import Page
from app.schemas.runtime import EventResponse, RunResponse
from app.services.audit import record_audit_event
from app.services.event_bus import bus
from app.services.serializers import event_to_response, run_to_response

TERMINAL_RUN_STATUS = {"completed", "failed", "cancelled", "interrupted"}
TERMINAL_EVENT_TYPES = {"run.completed", "run.failed", "run.cancelled", "run.interrupted"}


async def create_run(
    session: AsyncSession,
    *,
    character_id: str,
    model_hint: str,
    goal: str = "回复用户消息",
    origin: str = "chat",
) -> Run:
    run = Run(
        character_id=character_id,
        status="running",
        goal=goal,
        origin=origin,
        model_hint=model_hint,
        started_at=utc_now(),
    )
    session.add(run)
    await session.flush()
    await record_audit_event(
        session,
        action="run.started",
        character_id=character_id,
        run_id=run.id,
        actor="system",
        decision="auto",
        resource_type="run",
        resource_id=run.id,
    )
    return run


async def append_event(
    session: AsyncSession,
    *,
    run_id: str,
    character_id: str,
    event_type: str,
    payload: dict[str, Any],
    source: str = "system",
    visibility: str = "public",
    sensitivity: str = "low",
) -> Event:
    if event_type not in TERMINAL_EVENT_TYPES:
        event_count = await runtime_repo.count_events(session, run_id)
        if event_count >= settings.run_max_events:
            raise AppError("context_overflow", "运行事件数超过限制。")

    payload_json = dumps(redact_value(payload))
    if len(payload_json.encode("utf-8")) > settings.event_payload_max_bytes:
        raise AppError("payload_too_large", "运行事件 payload 超过大小限制。")

    sequence = await runtime_repo.next_event_sequence(session, run_id)
    event = Event(
        character_id=character_id,
        run_id=run_id,
        type=event_type,
        source=source,
        payload_json=payload_json,
        visibility=visibility,
        sensitivity=sensitivity,
        trace_id=current_trace_id(),
        sequence=sequence,
    )
    session.add(event)
    await session.flush()
    # publish 到进程内事件总线，供 SSE 实时转发（不再依赖 DB 轮询）
    bus.publish(run_id, event_to_response(event))
    return event


async def complete_run(session: AsyncSession, run: Run) -> Run:
    run.status = "completed"
    run.completed_at = utc_now()
    await session.flush()
    await record_audit_event(
        session,
        action="run.completed",
        character_id=run.character_id,
        run_id=run.id,
        actor="system",
        decision="auto",
        resource_type="run",
        resource_id=run.id,
    )
    return run


async def fail_run(session: AsyncSession, run: Run, *, code: str, message: str) -> Run:
    run.status = "failed"
    run.completed_at = utc_now()
    run.error_code = code
    run.error_message = message[:500]
    await session.flush()
    await record_audit_event(
        session,
        action="run.failed",
        character_id=run.character_id,
        run_id=run.id,
        actor="system",
        decision="blocked",
        reason=code,
        output_summary=message,
        resource_type="run",
        resource_id=run.id,
    )
    return run


async def cancel_run(run_id: str) -> RunResponse:
    async with AsyncSessionLocal() as session:
        run = await runtime_repo.get_run(session, run_id)
        if run is None:
            raise AppError("not_found", "Run 不存在。")
        if run.status not in TERMINAL_RUN_STATUS:
            run.status = "cancelled"
            run.completed_at = utc_now()
            await append_event(
                session,
                run_id=run.id,
                character_id=run.character_id,
                event_type="run.cancelled",
                payload={"run_id": run.id},
            )
            await record_audit_event(
                session,
                action="run.cancelled",
                character_id=run.character_id,
                run_id=run.id,
                actor="user",
                decision="auto",
                resource_type="run",
                resource_id=run.id,
            )
            await session.commit()
            await session.refresh(run)
        return run_to_response(run)


async def get_run_response(run_id: str) -> RunResponse:
    async with AsyncSessionLocal() as session:
        run = await runtime_repo.get_run(session, run_id)
        if run is None:
            raise AppError("not_found", "Run 不存在。")
        return run_to_response(run)


async def list_runs_response(
    *,
    character_id: str | None = None,
    status: str | None = None,
    page: int = 1,
    page_size: int = 20,
) -> Page[RunResponse]:
    """Run 列表（按创建时间倒序）；可选按人物/状态过滤。归档页运行时间河数据源。"""
    async with AsyncSessionLocal() as session:
        items, total = await runtime_repo.list_runs(
            session,
            character_id=character_id,
            status=status,
            page=page,
            page_size=page_size,
        )
        return Page(
            items=[run_to_response(run) for run in items],
            page=page,
            page_size=page_size,
            total=total,
        )


async def list_public_event_responses_after(
    session: AsyncSession, run_id: str, *, after_sequence: int = 0
) -> list[EventResponse]:
    events = await runtime_repo.list_public_events_after(
        session, run_id, after_sequence=after_sequence
    )
    return [event_to_response(event) for event in events]


async def list_run_event_responses(
    run_id: str, *, limit: int = 200, offset: int = 0
) -> list[EventResponse]:
    """列出 Run 的事件时间线（按 sequence 升序）。"""
    async with AsyncSessionLocal() as session:
        run = await session.get(Run, run_id)
        if run is None:
            raise AppError("not_found", "Run 不存在。")
        events, _ = await runtime_repo.list_events_by_run(
            session, run_id, limit=limit, offset=offset
        )
        return [event_to_response(event) for event in events]


async def get_tool_call_detail(tool_call_id: str):
    """GET /tools/calls/{tool_call_id}：单条 tool.* 事件详情。"""
    from app.core.errors import AppError
    from app.core.time import to_iso
    from app.repositories.json_utils import loads_dict
    from app.schemas.safety import ToolCallDetail

    async with AsyncSessionLocal() as session:
        event = await runtime_repo.get_event(session, tool_call_id)
    if event is None or not event.type.startswith("tool."):
        raise AppError("not_found", f"工具调用 {tool_call_id} 不存在。")
    payload = loads_dict(event.payload_json)
    return ToolCallDetail(
        event_id=event.id,
        type=event.type,
        tool=payload.get("tool"),
        run_id=event.run_id,
        character_id=event.character_id,
        trace_id=event.trace_id,
        payload=payload,
        created_at=to_iso(event.created_at),
    )