lzf_0626
790518114b
#23：413 是上传超限的标准语义，前端文档（风控业务演示文档 17）也已按 413 做提示
映射，所以不把代码降成 422，而是在 docs/05 §3.5 状态码表补登 413 —— 契约以"补齐"
而不是"改动"的方式对齐。
#24：/api/v1/risk/daily-report/stream 此前既不校验 Accept，又把鉴权留在 async
generator 内部。后者更隐蔽：StreamingResponse 已经返回、响应头已经发出，403 只能
变成"200 + 半截流"。现在 controller 先 await service.authorize(context) 再判定
Accept，顺序与 §6.4 一致（鉴权先行，不用状态码差异做探测）。SSE 协商逻辑抽到
app/api/dependencies/negotiation.py，与 /agent-runs/{run_id}/events 共用同一口径，
避免同一种客户端在一个端点上 200、另一个端点上 406。
#25：复核后确认前半段不成立 —— §19 末尾写明业务域接口由各自业务文档登记，风控 15 条
端点已在 06-模块接口与字段映射.md 逐条登记。真问题是 §12 表里写的
/api/v1/risk-scans/**、/api/v1/risk-alerts/** 与实际实现 /api/v1/risk/** 不符，
按实际实现更新 §12 并加说明；顺带把风控文档里 /daily-report/mail 的权限从
"按主项目邮件策略执行"改为实际的 risk:report:mail。
新增 tests/unit/api/test_risk_stream_negotiation.py（7 例）。
106 lines
4.6 KiB
Python
106 lines
4.6 KiB
Python
from collections.abc import AsyncIterator
|
||
from dataclasses import asdict
|
||
|
||
from fastapi import APIRouter, Depends, Request, status
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from starlette.responses import StreamingResponse
|
||
|
||
from app.api.dependencies.auth import build_request_context
|
||
from app.api.dependencies.database import get_session
|
||
from app.api.dependencies.negotiation import accepts_event_stream
|
||
from app.api.dependencies.rate_limit import enforce_rate_limit
|
||
from app.api.schemas.agent_runs import (
|
||
AgentRunAcceptedEnvelope,
|
||
AgentRunAcceptedResponse,
|
||
AgentRunCreateRequest,
|
||
AgentRunStatusEnvelope,
|
||
AgentRunStatusResponse,
|
||
)
|
||
from app.api.views.agent_run_sse import encode_events, recovery_events
|
||
from app.core.config import get_settings
|
||
from app.core.contracts import AgentRequest, RequestContext
|
||
from app.core.errors import SseNotAcceptableError
|
||
from app.service.agent_run_application_service import AgentRunApplicationService
|
||
from app.service.run_query_service import RunQueryService
|
||
|
||
router = APIRouter(prefix="/api/v1/agent-runs", tags=["agent-runs"],
|
||
dependencies=[Depends(enforce_rate_limit)])
|
||
|
||
@router.post(
|
||
"",
|
||
response_model=AgentRunAcceptedEnvelope,
|
||
status_code=status.HTTP_202_ACCEPTED,
|
||
)
|
||
async def create_agent_run(
|
||
payload: AgentRunCreateRequest,
|
||
request: Request,
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||
) -> AgentRunAcceptedEnvelope:
|
||
request.state.request_context = context
|
||
accepted = await AgentRunApplicationService(session).accept(
|
||
AgentRequest(**payload.model_dump()), context)
|
||
return AgentRunAcceptedEnvelope(
|
||
data=AgentRunAcceptedResponse(
|
||
run_id=accepted.run_id, trace_id=accepted.trace_id, status=accepted.status,
|
||
status_url=f"/api/v1/agent-runs/{accepted.run_id}",
|
||
events_url=f"/api/v1/agent-runs/{accepted.run_id}/events",
|
||
),
|
||
meta={"trace_id": context.trace_id},
|
||
)
|
||
|
||
|
||
@router.get("/{run_id}", response_model=AgentRunStatusEnvelope)
|
||
async def get_agent_run(
|
||
run_id: str, context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
) -> AgentRunStatusEnvelope:
|
||
"""查询运行（文档 §6.3）。
|
||
|
||
文档 §3.3 与 §6.3 都把成功响应定义为 `{data, meta:{trace_id}}` 信封；此前这里
|
||
直接返回资源对象，客户端必须为这一个接口特判。**只改包装结构**：`data` 内的字段名
|
||
与语义保持原样，`meta.trace_id` 用本次请求的 trace（`data.trace_id` 仍是运行自身的
|
||
追踪标识，两者语义不同，不能互相替代）。
|
||
"""
|
||
snapshot = await RunQueryService().get(run_id, context)
|
||
return AgentRunStatusEnvelope(
|
||
data=AgentRunStatusResponse(**asdict(snapshot)),
|
||
meta={"trace_id": context.trace_id},
|
||
)
|
||
|
||
|
||
@router.get("/{run_id}/events")
|
||
async def stream_agent_run_events(
|
||
run_id: str,
|
||
request: Request,
|
||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||
) -> StreamingResponse:
|
||
query = RunQueryService()
|
||
# 顺序按文档 §6.4 的主要错误列举：RUN_NOT_FOUND（含 AGENT_PERMISSION_DENIED 同级的
|
||
# 可见性判定）在前、SSE_NOT_ACCEPTABLE 在后。可见性先行（auth 依赖已先于本函数执行）
|
||
# 才能保证"运行是否存在"不因 Accept 头而异：否则用任意 run_id + 非法 Accept 探测，
|
||
# 406 与 404 的差异就等价于一次存在性枚举。
|
||
initial = await query.get(run_id, context)
|
||
if not accepts_event_stream(request.headers.get("Accept")):
|
||
raise SseNotAcceptableError("Accept 必须接受 text/event-stream")
|
||
|
||
async def generate() -> AsyncIterator[str]:
|
||
start_sent = False
|
||
async for snapshot in query.watch(initial, context):
|
||
if snapshot is None:
|
||
yield ": heartbeat\n\n"
|
||
continue
|
||
result = snapshot.result or {}
|
||
events = recovery_events(
|
||
run_id=snapshot.run_id, trace_id=snapshot.trace_id, status=snapshot.status,
|
||
error_code=snapshot.error_code, content=result.get("content"),
|
||
tool_calls=result.get("tool_calls"),
|
||
replay=initial.status in {"succeeded", "failed", "cancelled"},
|
||
chunk_size=get_settings().sse_chunk_characters,
|
||
)
|
||
for encoded in encode_events(run_id, events[1:] if start_sent else events):
|
||
yield encoded
|
||
start_sent = True
|
||
|
||
return StreamingResponse(generate(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache"})
|