"""Run 执行器：调用模型、产生真实流事件与 assistant 消息（WP-05）。 - 后台任务执行：独立数据库会话（不复用请求会话）。 - 事件序列（契约第 5 节）：run.status.changed(running) → content.block.started → content.block.delta* → content.block.completed → message.completed → run.status.changed(completed/failed)。 - 首个用户可见 delta 前的失败允许显式 fail；无可用端点时 Run 失败并留 error 事件。 """ import time from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.core.ids import new_id from app.gateway.adapters import AdapterRequest, GatewayError, stream_chat from app.models import ( AuditEvent, Message, ModelCall, OutboxEvent, ProviderEndpoint, Run, RunEvent, Turn, ) async def _next_sequence(db: AsyncSession, run_id: str) -> int: last = await db.execute( select(RunEvent.sequence) .where(RunEvent.run_id == run_id) .order_by(RunEvent.sequence.desc()) .limit(1) ) row = last.scalar_one_or_none() return (row + 1) if row is not None else 0 def _emit(db: AsyncSession, run_id: str, seq: int, event_type: str, data: dict[str, Any]) -> int: db.add(RunEvent(run_id=run_id, sequence=seq, type=event_type, data=data)) return seq + 1 async def execute_run( factory: async_sessionmaker[AsyncSession], run_id: str, transport: Any | None = None, ) -> None: """执行一个 Run 至终态。transport 注入仅供测试。""" async with factory() as db: try: run = await db.get(Run, run_id) if run is None: return seq = await _next_sequence(db, run_id) # 状态事件：running run.status = "running" seq = _emit( db, run_id, seq, "run.status.changed", {"run_id": run_id, "status": "running"}, ) await db.commit() # 路由：按逻辑模型选择启用端点（优先级升序） endpoint = ( await db.execute( select(ProviderEndpoint) .where( ProviderEndpoint.enabled.is_(True), ProviderEndpoint.logical_model_id == "default-chat", ) .order_by(ProviderEndpoint.priority.asc()) .limit(1) ) ).scalar_one_or_none() turn = Turn(run_id=run_id, index=0) db.add(turn) await db.flush() if endpoint is None: # fail closed：无可用端点时 Run 显式失败，不静默降级 call = ModelCall( turn_id=turn.id, status="failed", request_summary={}, error="无可用 Provider 端点（default-chat）", started_at=datetime.now(UTC), completed_at=datetime.now(UTC), ) db.add(call) run.status = "failed" run.completed_at = datetime.now(UTC) seq = _emit(db, run_id, seq, "content.block.failed", { "run_id": run_id, "message_id": None, "index": 0, "block": {"id": new_id("block_"), "type": "text", "status": "failed"}, "error": "no_endpoint_available", }) seq = _emit(db, run_id, seq, "message.completed", {"run_id": run_id, "message_ids": [], "final_status": "failed"}) seq = _emit(db, run_id, seq, "run.status.changed", {"run_id": run_id, "status": "failed"}) await db.commit() return # 对话历史（最小：当前分支文本消息） messages = ( await db.execute( select(Message) .where( Message.conversation_id == run.conversation_id, Message.role.in_(["user", "assistant"]), ) .order_by(Message.created_at.asc(), Message.id.asc()) .limit(50) ) ).scalars().all() prompt = [ { "role": m.role, "content": "".join( str(b.get("text", b.get("source", ""))) for b in m.content if isinstance(b, dict) ), } for m in messages ] call = ModelCall( turn_id=turn.id, endpoint_id=endpoint.id, status="streaming", request_summary={"messages": len(prompt), "model_id": endpoint.model_id}, started_at=datetime.now(UTC), ) db.add(call) await db.commit() block_id = new_id("block_") seq = _emit(db, run_id, seq, "content.block.started", { "run_id": run_id, "index": 0, "block": {"id": block_id, "type": "text", "status": "streaming"}, }) await db.commit() # 流式消费：真实 delta 事件 started = time.monotonic() ttft_ms: int | None = None full_text: list[str] = [] usage: dict[str, int] | None = None finish_reason: str | None = None delta_index = 0 try: adapter_req = AdapterRequest(messages=prompt, model=endpoint.model_id) async for chunk in stream_chat( endpoint.base_url, endpoint.model_id, endpoint.api_key_env, adapter_req, transport=transport, ): if chunk.kind == "delta" and chunk.text: if ttft_ms is None: ttft_ms = int((time.monotonic() - started) * 1000) full_text.append(chunk.text) seq = _emit(db, run_id, seq, "content.block.delta", { "run_id": run_id, "block_id": block_id, "delta_index": delta_index, "delta": {"type": "text_append", "text": chunk.text}, }) delta_index += 1 await db.commit() elif chunk.kind == "usage" and chunk.usage: usage = chunk.usage elif chunk.kind == "done" and chunk.finish_reason: finish_reason = chunk.finish_reason except GatewayError as exc: failed_call = await db.get(ModelCall, call.id) assert failed_call is not None call.status = "failed" call.error = str(exc)[:500] call.completed_at = datetime.now(UTC) run.status = "failed" run.completed_at = datetime.now(UTC) seq = _emit(db, run_id, seq, "content.block.failed", { "run_id": run_id, "block_id": block_id, "index": 0, "error": "upstream_error", }) seq = _emit(db, run_id, seq, "message.completed", {"run_id": run_id, "message_ids": [], "final_status": "failed"}) seq = _emit(db, run_id, seq, "run.status.changed", {"run_id": run_id, "status": "failed"}) await db.commit() return text = "".join(full_text) seq = _emit(db, run_id, seq, "content.block.completed", { "run_id": run_id, "index": 0, "block": {"id": block_id, "type": "text", "status": "completed", "text": text}, }) # assistant 消息（权威快照） assistant = Message( conversation_id=run.conversation_id, branch_id=( await db.execute( select(Message.branch_id).where(Message.run_id == run_id).limit(1) ) ).scalar_one_or_none() or new_id("branch_"), role="assistant", realm="reality", content=[{"id": block_id, "type": "text", "text": text, "status": "completed"}], run_id=run_id, ) db.add(assistant) await db.flush() seq = _emit(db, run_id, seq, "message.completed", { "run_id": run_id, "message_ids": [assistant.id], "final_status": "completed", }) run.status = "completed" run.completed_at = datetime.now(UTC) seq = _emit(db, run_id, seq, "run.status.changed", {"run_id": run_id, "status": "completed"}) success_call = await db.get(ModelCall, call.id) assert success_call is not None success_call.status = "succeeded" # Provider 未返回的 usage 保持 None：不估算伪装（ADR-003） success_call.usage = usage success_call.ttft_ms = ttft_ms success_call.completed_at = datetime.now(UTC) if finish_reason: success_call.request_summary = { **success_call.request_summary, "finish_reason": finish_reason } db.add( AuditEvent( actor_type="system", actor_id="gateway", action="model_call.completed", aggregate_type="run", aggregate_id=run_id, payload={"endpoint": endpoint.name, "usage": usage, "ttft_ms": ttft_ms}, ) ) db.add( OutboxEvent( aggregate_type="message", aggregate_id=assistant.id, event_type="message.created", payload={"message_id": assistant.id, "run_id": run_id, "role": "assistant"}, ) ) await db.commit() except Exception: await db.rollback() raise