246 lines
8.7 KiB
Python
246 lines
8.7 KiB
Python
"""Run 与 SSE API。"""
|
|
|
|
import asyncio
|
|
import json
|
|
from collections.abc import AsyncGenerator
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Header, Query, Request
|
|
from sse_starlette.sse import EventSourceResponse
|
|
|
|
from app.application.agent.confirmation import ConfirmationRegistry
|
|
from app.application.agent.run_registry import AgentRunRegistry
|
|
from app.core.config import settings
|
|
from app.core.errors import AppError
|
|
from app.core.time import to_iso, utc_now
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.repositories import runtime as runtime_repo
|
|
from app.schemas.common import Page
|
|
from app.schemas.runtime import (
|
|
AgentControlRequest,
|
|
AgentControlResponse,
|
|
ConfirmationResolutionResponse,
|
|
EventResponse,
|
|
RunResponse,
|
|
)
|
|
from app.services.event_bus import bus
|
|
from app.services.runtime import (
|
|
cancel_run,
|
|
confirmation_exists_in_run,
|
|
get_run_response,
|
|
list_run_event_responses,
|
|
list_runs_response,
|
|
)
|
|
|
|
router = APIRouter(prefix="/runs", tags=["runs"])
|
|
|
|
|
|
@router.get("", response_model=Page[RunResponse])
|
|
async def list_runs(
|
|
character_id: str | None = Query(default=None, description="按人物过滤。"),
|
|
status: str | None = Query(default=None, description="按状态过滤。"),
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=20, ge=1, le=100),
|
|
) -> Page[RunResponse]:
|
|
"""Run 列表（按创建时间倒序）；归档页运行时间河数据源。"""
|
|
return await list_runs_response(
|
|
character_id=character_id,
|
|
status=status,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.get("/{run_id}", response_model=RunResponse)
|
|
async def get_run(run_id: str) -> RunResponse:
|
|
return await get_run_response(run_id)
|
|
|
|
|
|
@router.get("/{run_id}/events", response_model=list[EventResponse])
|
|
async def list_run_events(
|
|
run_id: str,
|
|
limit: int = Query(default=200, ge=1, le=500),
|
|
offset: int = Query(default=0, ge=0),
|
|
) -> list[EventResponse]:
|
|
"""Run 的事件时间线（按 sequence 升序）。"""
|
|
return await list_run_event_responses(run_id, limit=limit, offset=offset)
|
|
|
|
|
|
def _get_agent_run_registry(request: Request) -> AgentRunRegistry:
|
|
registry: AgentRunRegistry | None = getattr(request.app.state, "agent_run_registry", None)
|
|
if registry is None:
|
|
raise AppError("service_unavailable", "Agent 运行控制未启用。", status_code=503)
|
|
return registry
|
|
|
|
|
|
@router.post("/{run_id}/cancel", response_model=RunResponse)
|
|
async def cancel_run_api(run_id: str, request: Request) -> RunResponse:
|
|
registry = getattr(request.app.state, "agent_run_registry", None)
|
|
if registry is not None:
|
|
await registry.abort(run_id)
|
|
return await cancel_run(run_id)
|
|
|
|
|
|
@router.post("/{run_id}/steer", response_model=AgentControlResponse)
|
|
async def steer_run(
|
|
run_id: str,
|
|
payload: AgentControlRequest,
|
|
request: Request,
|
|
) -> AgentControlResponse:
|
|
if not _get_agent_run_registry(request).steer(run_id, payload.content):
|
|
raise AppError("conflict", "Run 当前不在本进程执行，无法 steer。", status_code=409)
|
|
return AgentControlResponse()
|
|
|
|
|
|
@router.post("/{run_id}/follow-up", response_model=AgentControlResponse)
|
|
async def follow_up_run(
|
|
run_id: str,
|
|
payload: AgentControlRequest,
|
|
request: Request,
|
|
) -> AgentControlResponse:
|
|
if not _get_agent_run_registry(request).follow_up(run_id, payload.content):
|
|
raise AppError("conflict", "Run 当前不在本进程执行，无法追加 follow-up。", status_code=409)
|
|
return AgentControlResponse()
|
|
|
|
|
|
def _get_confirmation_registry(request: Request) -> ConfirmationRegistry:
|
|
"""从 app.state 取 ConfirmationRegistry；未挂载时 503。"""
|
|
registry: ConfirmationRegistry | None = getattr(
|
|
request.app.state, "confirmation_registry", None
|
|
)
|
|
if registry is None:
|
|
raise AppError(
|
|
"service_unavailable",
|
|
"确认机制未启用（进程内 Future 注册表未挂载）",
|
|
status_code=503,
|
|
)
|
|
return registry
|
|
|
|
|
|
async def _resolve_confirmation(
|
|
request: Request,
|
|
run_id: str,
|
|
confirmation_id: str,
|
|
approved: bool,
|
|
) -> ConfirmationResolutionResponse:
|
|
"""approve/deny 共用逻辑：通过 ConfirmationRegistry 更新确认状态。"""
|
|
registry = _get_confirmation_registry(request)
|
|
if await registry.resolve(run_id, approved, confirmation_id=confirmation_id):
|
|
return ConfirmationResolutionResponse(approved=approved)
|
|
|
|
# 记录已处理/不存在——查 events 表区分 410 与 404
|
|
event_store = getattr(request.app.state, "event_store", None)
|
|
if event_store is not None and await confirmation_exists_in_run(
|
|
event_store, run_id, confirmation_id
|
|
):
|
|
raise AppError(
|
|
"confirmation_expired",
|
|
"确认已失效（服务可能已重启或已处理），请重新发起对话",
|
|
status_code=410,
|
|
)
|
|
raise AppError("confirmation_not_found", "确认项不存在", status_code=404)
|
|
|
|
|
|
@router.post(
|
|
"/{run_id}/confirmations/{confirmation_id}/approve",
|
|
response_model=ConfirmationResolutionResponse,
|
|
)
|
|
async def approve_confirmation(
|
|
run_id: str,
|
|
confirmation_id: str,
|
|
request: Request,
|
|
) -> ConfirmationResolutionResponse:
|
|
"""批准高风险工具执行：触发 Agent 循环恢复（confirmed=True 重试）。"""
|
|
return await _resolve_confirmation(request, run_id, confirmation_id, approved=True)
|
|
|
|
|
|
@router.post(
|
|
"/{run_id}/confirmations/{confirmation_id}/deny",
|
|
response_model=ConfirmationResolutionResponse,
|
|
)
|
|
async def deny_confirmation(
|
|
run_id: str,
|
|
confirmation_id: str,
|
|
request: Request,
|
|
) -> ConfirmationResolutionResponse:
|
|
"""拒绝高风险工具执行：Agent 循环回填拒绝 ToolMessage。"""
|
|
return await _resolve_confirmation(request, run_id, confirmation_id, approved=False)
|
|
|
|
|
|
@router.get("/{run_id}/stream")
|
|
async def stream_run(
|
|
run_id: str,
|
|
last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
|
|
) -> EventSourceResponse:
|
|
"""Run 事件流（SSE）：按 run_id 实时订阅事件增量。
|
|
|
|
事件通过进程内 ``bus`` 实时推送（``append_event`` 自动 publish），
|
|
DB 持久化事件供断线重连（``Last-Event-ID`` 回放）。
|
|
"""
|
|
if not settings.enable_sse:
|
|
raise AppError("invalid_request", "SSE 当前未启用。")
|
|
async with AsyncSessionLocal() as session:
|
|
run = await runtime_repo.get_run(session, run_id)
|
|
if run is None:
|
|
raise AppError("not_found", "Run 不存在。")
|
|
return EventSourceResponse(_event_generator(run_id, last_event_id))
|
|
|
|
|
|
async def _event_generator(
|
|
run_id: str, last_event_id: str | None
|
|
) -> AsyncGenerator[dict[str, Any], None]:
|
|
# 断线重连：解析 Last-Event-ID 对应的 sequence 作为回放起点
|
|
after_sequence = 0
|
|
if last_event_id:
|
|
async with AsyncSessionLocal() as session:
|
|
previous_event = await runtime_repo.get_event(session, last_event_id)
|
|
if previous_event and previous_event.run_id == run_id:
|
|
after_sequence = previous_event.sequence
|
|
|
|
last_sequence = after_sequence
|
|
# 先订阅再回放，避免回放与订阅之间的事件落入间隙
|
|
queue = bus.subscribe(run_id)
|
|
try:
|
|
for event in bus.list_after(run_id, after_sequence=last_sequence):
|
|
last_sequence = event.sequence
|
|
yield {
|
|
"id": event.event_id,
|
|
"event": event.type,
|
|
"data": event.model_dump_json(),
|
|
}
|
|
if bus.is_terminal(event):
|
|
return
|
|
|
|
while True:
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout=settings.sse_heartbeat_seconds)
|
|
except TimeoutError:
|
|
yield {
|
|
"event": "heartbeat",
|
|
"data": json.dumps(
|
|
{
|
|
"event_id": "heartbeat",
|
|
"run_id": run_id,
|
|
"type": "heartbeat",
|
|
"sequence": last_sequence,
|
|
"visibility": "public",
|
|
"payload": {},
|
|
"created_at": to_iso(utc_now()),
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
}
|
|
continue
|
|
if event.sequence <= last_sequence:
|
|
continue # 回放阶段已转发，去重
|
|
last_sequence = event.sequence
|
|
yield {
|
|
"id": event.event_id,
|
|
"event": event.type,
|
|
"data": event.model_dump_json(),
|
|
}
|
|
if bus.is_terminal(event):
|
|
return
|
|
finally:
|
|
bus.unsubscribe(run_id, queue)
|