from datetime import datetime from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.agents.learning.store import LearningArtifactStore, SessionRetrospectiveStore from app.agents.registry import load_builtin_registry_indexes from app.agents.runtime_metrics import coerce_cost_thresholds, estimate_token_cost, is_cost_budget_warning from app.models.agent import Agent from app.models.conversation import Conversation from app.models.skill import Skill from app.models.user import User from app.routers.auth import get_current_user from app.schemas.agent import ( AgentConfigOut, AgentConfigUpdate, AgentCreate, AgentOut, AgentStats, AgentVisibilityCostByAgentOut, AgentVisibilityCostOut, AgentVisibilityCostSummaryOut, AgentVisibilityEvidenceOut, AgentVisibilityEventsResponse, AgentVisibilityEventOut, AgentVisibilityIsolationOut, AgentVisibilityRuntimeSummaryOut, AgentVisibilityTaskSummaryOut, AgentVisibilityThreadMessageOut, AgentVisibilityThreadOut, AgentVisibilityTopologyNodeOut, AgentVisibilityTopologyOut, AgentVisibilityToolGovernanceItemOut, AgentVisibilityToolGovernanceOut, AgentVisibilityVerifierOut, ) from app.services.agent_service import _extract_continuity_snapshot from app.services.runtime_observability import build_runtime_observability_report router = APIRouter(prefix="/api/agents", tags=["Agent"]) _agent_call_counts: dict[str, int] = {} _agent_current_tasks: dict[str, str | None] = {} _agent_statuses: dict[str, str] = {} DEFAULT_AGENT_ROLES = ["master", "schedule_planner", "executor", "librarian", "analyst"] SUB_COMMANDERS_BY_ROLE = { "schedule_planner": ["schedule_analysis", "schedule_planning"], "executor": ["executor_tasks", "executor_forum"], "librarian": ["librarian_retrieval", "librarian_graph"], "analyst": ["analyst_progress", "analyst_insights"], } ALLOWED_AGENT_ROLES = set(DEFAULT_AGENT_ROLES) | { role for sub_roles in SUB_COMMANDERS_BY_ROLE.values() for role in sub_roles } def _parse_visibility_datetime(value: str | None) -> datetime | None: if value is None: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise HTTPException(status_code=400, detail="时间参数必须是 ISO 8601 格式") from exc async def _get_visibility_state( conversation_id: str, *, current_user: User, db: AsyncSession, ) -> dict[str, Any]: result = await db.execute( select(Conversation).where( Conversation.id == conversation_id, Conversation.user_id == current_user.id, ) ) conversation = result.scalar_one_or_none() if conversation is None: raise HTTPException(status_code=404, detail="对话不存在") snapshot = _extract_continuity_snapshot(conversation.agent_state) if snapshot is None: raise HTTPException(status_code=404, detail="当前会话暂无可视化运行时数据") return snapshot def _coerce_event_payload(event: dict[str, Any]) -> AgentVisibilityEventOut: return AgentVisibilityEventOut.model_validate(event) def _filter_events( events: list[dict[str, Any]], *, agent_id: str | None, thread_id: str | None, event_type: str | None, started_after: datetime | None, ended_before: datetime | None, ) -> list[dict[str, Any]]: filtered: list[dict[str, Any]] = [] for event in events: if agent_id and event.get("agent_id") != agent_id: continue if thread_id and event.get("thread_id") != thread_id: continue if event_type and event.get("event_type") != event_type: continue timestamp_raw = event.get("timestamp") timestamp = None if isinstance(timestamp_raw, str): try: timestamp = datetime.fromisoformat(timestamp_raw.replace("Z", "+00:00")) except ValueError: timestamp = None if started_after and timestamp and timestamp < started_after: continue if ended_before and timestamp and timestamp > ended_before: continue filtered.append(event) return filtered def _summarize_tasks(tasks: list[dict[str, Any]], task_results: list[dict[str, Any]]) -> list[AgentVisibilityTaskSummaryOut]: result_by_task_id = {item.get("task_id"): item for item in task_results} summaries: list[AgentVisibilityTaskSummaryOut] = [] for task in tasks: task_id = str(task.get("task_id") or "") result = result_by_task_id.get(task_id) or {} evidence = result.get("evidence") or task.get("evidence") or [] summaries.append( AgentVisibilityTaskSummaryOut( task_id=task_id, role=task.get("role"), owner_agent_id=task.get("owner_agent_id") or result.get("owner_agent_id"), status=result.get("status") or task.get("status"), summary=result.get("summary") or task.get("result_summary"), evidence_count=len(evidence), ) ) return summaries def _build_topology_nodes( state: dict[str, Any], tasks: list[dict[str, Any]], task_results: list[dict[str, Any]], ) -> list[AgentVisibilityTopologyNodeOut]: task_counts: dict[str, int] = {} completed_counts: dict[str, int] = {} for task in tasks: owner = str(task.get("owner_agent_id") or "") if owner: task_counts[owner] = task_counts.get(owner, 0) + 1 for result in task_results: owner = str(result.get("owner_agent_id") or "") if owner and result.get("status") == "completed": completed_counts[owner] = completed_counts.get(owner, 0) + 1 root_agent_id = str(state.get("root_agent_id") or state.get("agent_id") or "") or None current_agent = str(state.get("current_agent") or "") or None parent_agent_id = str(state.get("parent_agent_id") or "") or None nodes: dict[str, AgentVisibilityTopologyNodeOut] = {} if root_agent_id: nodes[root_agent_id] = AgentVisibilityTopologyNodeOut( agent_id=root_agent_id, role=root_agent_id.split("-")[0], parent_agent_id=parent_agent_id if root_agent_id != state.get("agent_id") else None, source="root", task_count=task_counts.get(root_agent_id, 0), completed_task_count=completed_counts.get(root_agent_id, 0), ) for agent_id in state.get("spawned_agent_ids") or []: agent_id = str(agent_id) nodes[agent_id] = AgentVisibilityTopologyNodeOut( agent_id=agent_id, role=agent_id.split("-")[0], parent_agent_id=root_agent_id, source="spawned", task_count=task_counts.get(agent_id, 0), completed_task_count=completed_counts.get(agent_id, 0), ) if current_agent and current_agent not in nodes: nodes[current_agent] = AgentVisibilityTopologyNodeOut( agent_id=current_agent, role=current_agent.split("-")[0], parent_agent_id=None if current_agent == root_agent_id else root_agent_id, source="current", task_count=task_counts.get(current_agent, 0), completed_task_count=completed_counts.get(current_agent, 0), ) return list(nodes.values()) def _estimate_runtime_cost(input_tokens: int, output_tokens: int) -> float | None: return estimate_token_cost(input_tokens, output_tokens) def _build_cost_summary( state: dict[str, Any], *, conversation_id: str, ) -> AgentVisibilityCostSummaryOut: input_tokens = int(state.get("input_tokens") or 0) output_tokens = int(state.get("output_tokens") or 0) estimated_cost = _estimate_runtime_cost(input_tokens, output_tokens) thresholds = coerce_cost_thresholds(state.get("cost_thresholds")) total_budget_warning = bool(state.get("budget_warning") or False) or is_cost_budget_warning( input_tokens, output_tokens, estimated_cost, thresholds, ) by_agent_items: list[AgentVisibilityCostByAgentOut] = [] for agent_id, payload in dict(state.get("cost_by_agent") or {}).items(): payload_dict = dict(payload or {}) agent_input_tokens = int(payload_dict.get("input_tokens") or 0) agent_output_tokens = int(payload_dict.get("output_tokens") or 0) agent_estimated_cost = payload_dict.get("estimated_cost") if agent_estimated_cost is None: agent_estimated_cost = _estimate_runtime_cost(agent_input_tokens, agent_output_tokens) by_agent_items.append( AgentVisibilityCostByAgentOut( agent_id=str(payload_dict.get("agent_id") or agent_id), input_tokens=agent_input_tokens, output_tokens=agent_output_tokens, total_tokens=int(payload_dict.get("total_tokens") or (agent_input_tokens + agent_output_tokens)), estimated_cost=agent_estimated_cost, budget_warning=bool(payload_dict.get("budget_warning") or False), ) ) by_agent_items.sort(key=lambda item: item.total_tokens, reverse=True) return AgentVisibilityCostSummaryOut( conversation_id=conversation_id, total=AgentVisibilityCostOut( input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, estimated_cost=estimated_cost, budget_warning=total_budget_warning, ), thresholds=thresholds, by_agent=by_agent_items, ) def _build_tool_governance( state: dict[str, Any], *, conversation_id: str, ) -> AgentVisibilityToolGovernanceOut: indexes = load_builtin_registry_indexes() tool_outcomes = [dict(item) for item in state.get("tool_outcomes") or [] if isinstance(item, dict)] usage_count_by_tool: dict[str, int] = {} last_result_preview_by_tool: dict[str, str | None] = {} for item in tool_outcomes: tool_name = str(item.get("tool_name") or "") if tool_name == "search_web": tool_name = "web_search" if not tool_name: continue usage_count_by_tool[tool_name] = usage_count_by_tool.get(tool_name, 0) + 1 preview = item.get("result_preview") if isinstance(preview, str) and preview: last_result_preview_by_tool[tool_name] = preview items = [ AgentVisibilityToolGovernanceItemOut( capability_id=capability.capability_id, tool_name=capability.tool_name, permission_class=capability.permission_class.value, side_effect_scope=capability.side_effect_scope.value, supports_retry=capability.supports_retry, idempotent=capability.idempotent, safe_for_parallel_use=capability.safe_for_parallel_use, requires_confirmation=capability.requires_confirmation, usage_count=usage_count_by_tool.get(capability.tool_name, 0), last_result_preview=last_result_preview_by_tool.get(capability.tool_name), ) for capability in indexes.capability_by_id.values() ] items.sort(key=lambda item: (-item.usage_count, item.tool_name)) return AgentVisibilityToolGovernanceOut( conversation_id=conversation_id, total_tools=len(items), used_tools=sum(1 for item in items if item.usage_count > 0), items=items, upgrade_candidates=[ "worktree_manager", "cost_inspector", "runtime_event_drilldown", "tool_policy_explorer", ], ) def _build_runtime_summary( state: dict[str, Any], *, conversation_id: str, ) -> AgentVisibilityRuntimeSummaryOut: tasks = [dict(item) for item in state.get("active_tasks") or []] task_results = [dict(item) for item in state.get("task_results") or []] topology_nodes = _build_topology_nodes(state, tasks, task_results) cost_summary = _build_cost_summary(state, conversation_id=conversation_id) input_tokens = cost_summary.total.input_tokens output_tokens = cost_summary.total.output_tokens recent_events_raw = [dict(item) for item in (state.get("event_trace") or [])[-10:]] isolation_mode = str(state.get("isolation_mode") or "none") return AgentVisibilityRuntimeSummaryOut( conversation_id=conversation_id, execution_mode=state.get("execution_mode"), current_phase=state.get("current_phase"), current_checkpoint=state.get("current_checkpoint"), phase_history=list(state.get("phase_history") or []), checkpoint_history=list(state.get("checkpoint_history") or []), verifier=AgentVisibilityVerifierOut( conversation_id=conversation_id, status=state.get("verification_status"), summary=state.get("verification_summary"), evidence=list(state.get("verification_evidence") or []), ), isolation=AgentVisibilityIsolationOut( mode=isolation_mode, isolation_id=state.get("isolation_id"), workspace_path=state.get("isolation_workspace_path"), parent_conversation_id=state.get("isolation_parent_conversation_id") or state.get("parent_conversation_id"), metadata=dict(state.get("isolation_metadata") or {}), ), cost=cost_summary.total, topology_node_count=len(topology_nodes), active_task_count=len(tasks), completed_task_count=sum(1 for item in task_results if item.get("status") == "completed"), recent_events=[_coerce_event_payload(item) for item in recent_events_raw], ) def record_agent_call(agent_id: str): _agent_call_counts[agent_id] = _agent_call_counts.get(agent_id, 0) + 1 def set_agent_task(agent_id: str, task: str | None): _agent_current_tasks[agent_id] = task _agent_statuses[agent_id] = "active" if task else "idle" def set_agent_status(agent_id: str, status: str): _agent_statuses[agent_id] = status def _build_agent_stats(agent_id: str) -> AgentStats: return AgentStats( agent_id=agent_id, call_count=_agent_call_counts.get(agent_id, 0), current_task=_agent_current_tasks.get(agent_id), status=_agent_statuses.get(agent_id, "idle"), ) @router.get("", response_model=list[AgentOut]) async def list_agents( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute( select(Agent).where(Agent.is_active == True).order_by(Agent.role) ) return result.scalars().all() @router.get("/stats", response_model=list[AgentStats]) async def get_agent_stats( current_user: User = Depends(get_current_user), ): return [_build_agent_stats(role) for role in DEFAULT_AGENT_ROLES] @router.get("/stats/hierarchy") async def get_agent_hierarchy_stats( current_user: User = Depends(get_current_user), ): main_agents = [] for role in DEFAULT_AGENT_ROLES: if role == "master": continue node = _build_agent_stats(role).model_dump() node["sub_commanders"] = [ _build_agent_stats(sub_id).model_dump() for sub_id in SUB_COMMANDERS_BY_ROLE.get(role, []) ] main_agents.append(node) return {"main_agents": main_agents} @router.get("/config/{agent_id}", response_model=AgentConfigOut) async def get_agent_config( agent_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute(select(Agent).where(Agent.role == agent_id)) agent = result.scalar_one_or_none() if not agent: from app.agents.prompts import MASTER_SYSTEM_PROMPT, SCHEDULE_PLANNER_SYSTEM_PROMPT, EXECUTOR_SYSTEM_PROMPT, LIBRARIAN_SYSTEM_PROMPT, ANALYST_SYSTEM_PROMPT defaults = { "master": ("JARVIS", "主控制核心", MASTER_SYSTEM_PROMPT), "schedule_planner": ("SCHEDULE PLANNER", "日程规划师", SCHEDULE_PLANNER_SYSTEM_PROMPT), "executor": ("EXECUTOR", "执行专家", EXECUTOR_SYSTEM_PROMPT), "librarian": ("LIBRARIAN", "知识管理员", LIBRARIAN_SYSTEM_PROMPT), "analyst": ("ANALYST", "数据分析师", ANALYST_SYSTEM_PROMPT), } if agent_id not in defaults: raise HTTPException(status_code=404, detail="Agent 不存在") name, desc, prompt = defaults[agent_id] return AgentConfigOut( id=agent_id, name=name, role=agent_id, description=desc, system_prompt=prompt, enabled=True, is_active=True, selected_skill_ids=[], ) return AgentConfigOut( id=agent.role, name=agent.name, role=agent.role, description=agent.description, system_prompt=agent.system_prompt, enabled=agent.is_active, is_active=agent.is_active, selected_skill_ids=agent.selected_skill_ids or [], ) @router.put("/config/{agent_id}", response_model=AgentConfigOut) async def update_agent_config( agent_id: str, data: AgentConfigUpdate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute(select(Agent).where(Agent.role == agent_id)) agent = result.scalar_one_or_none() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") if data.name is not None: agent.name = data.name if data.description is not None: agent.description = data.description if data.system_prompt is not None: agent.system_prompt = data.system_prompt if data.enabled is not None: agent.is_active = data.enabled _agent_statuses[agent_id] = "disabled" if not data.enabled else "idle" if data.selected_skill_ids is not None: if data.selected_skill_ids: result = await db.execute( select(Skill.id).where( Skill.id.in_(data.selected_skill_ids), Skill.owner_id == current_user.id, ) ) allowed_skill_ids = set(result.scalars().all()) invalid_skill_ids = [skill_id for skill_id in data.selected_skill_ids if skill_id not in allowed_skill_ids] if invalid_skill_ids: raise HTTPException(status_code=400, detail="存在无效的技能绑定") agent.selected_skill_ids = data.selected_skill_ids await db.commit() await db.refresh(agent) return AgentConfigOut( id=agent.role, name=agent.name, role=agent.role, description=agent.description, system_prompt=agent.system_prompt, enabled=agent.is_active, is_active=agent.is_active, selected_skill_ids=agent.selected_skill_ids or [], ) @router.get("/visibility/events", response_model=AgentVisibilityEventsResponse) async def get_visibility_events( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), agent_id: str | None = None, thread_id: str | None = None, event_type: str | None = None, started_after: str | None = None, ended_before: str | None = None, limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) events = [dict(item) for item in state.get("event_trace") or []] filtered = _filter_events( events, agent_id=agent_id, thread_id=thread_id, event_type=event_type, started_after=_parse_visibility_datetime(started_after), ended_before=_parse_visibility_datetime(ended_before), ) paged = filtered[offset:offset + limit] return AgentVisibilityEventsResponse( conversation_id=conversation_id, total=len(filtered), limit=limit, offset=offset, items=[_coerce_event_payload(item) for item in paged], ) @router.get("/visibility/topology", response_model=AgentVisibilityTopologyOut) async def get_visibility_topology( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) tasks = [dict(item) for item in state.get("active_tasks") or []] task_results = [dict(item) for item in state.get("task_results") or []] nodes = _build_topology_nodes(state, tasks, task_results) root_agent_id = str(state.get("root_agent_id") or state.get("agent_id") or "") or None edges = [ {"parent_agent_id": root_agent_id, "child_agent_id": node.agent_id} for node in nodes if node.parent_agent_id and root_agent_id and node.agent_id != root_agent_id ] return AgentVisibilityTopologyOut( conversation_id=conversation_id, root_agent_id=root_agent_id, current_agent=str(state.get("current_agent") or "") or None, nodes=nodes, edges=edges, tasks=_summarize_tasks(tasks, task_results), task_hierarchy=dict(state.get("task_hierarchy") or {}), ) @router.get("/visibility/tasks/{task_id}/evidence", response_model=AgentVisibilityEvidenceOut) async def get_visibility_task_evidence( task_id: str, conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) tasks = [dict(item) for item in state.get("active_tasks") or []] task = next((item for item in tasks if item.get("task_id") == task_id), None) task_results = [dict(item) for item in state.get("task_results") or []] result = next((item for item in task_results if item.get("task_id") == task_id), None) if task is None and result is None: raise HTTPException(status_code=404, detail="任务不存在") tool_outcomes = [ dict(evidence) for evidence in (result or {}).get("evidence") or [] if isinstance(evidence, dict) and evidence.get("tool_name") ] verification_entry = next( ( dict(evidence) for evidence in (result or {}).get("evidence") or [] if isinstance(evidence, dict) and evidence.get("type") == "verification" ), None, ) verifier = { "status": (verification_entry or {}).get("status"), "summary": (verification_entry or {}).get("summary"), "evidence": [dict(item) for item in state.get("verification_evidence") or [] if item.get("task_id") == task_id], } return AgentVisibilityEvidenceOut( conversation_id=conversation_id, task_id=task_id, task=task, result=result, tool_outcomes=tool_outcomes, verifier=verifier, ) @router.get("/visibility/threads/{thread_id}/messages", response_model=AgentVisibilityThreadOut) async def get_visibility_thread_messages( thread_id: str, conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) items = [ AgentVisibilityThreadMessageOut.model_validate(item) for item in state.get("message_trace") or [] if isinstance(item, dict) and item.get("thread_id") == thread_id ] if not items: raise HTTPException(status_code=404, detail="线程不存在") return AgentVisibilityThreadOut( conversation_id=conversation_id, thread_id=thread_id, total=len(items), items=items, ) @router.get("/visibility/verifier", response_model=AgentVisibilityVerifierOut) async def get_visibility_verifier( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) return AgentVisibilityVerifierOut( conversation_id=conversation_id, status=state.get("verification_status"), summary=state.get("verification_summary"), evidence=list(state.get("verification_evidence") or []), ) @router.get("/visibility/runtime-summary", response_model=AgentVisibilityRuntimeSummaryOut) async def get_visibility_runtime_summary( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) return _build_runtime_summary(state, conversation_id=conversation_id) @router.get("/visibility/cost", response_model=AgentVisibilityCostSummaryOut) async def get_visibility_cost( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) return _build_cost_summary(state, conversation_id=conversation_id) @router.get("/visibility/tools", response_model=AgentVisibilityToolGovernanceOut) async def get_visibility_tools( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) return _build_tool_governance(state, conversation_id=conversation_id) @router.get("/visibility/debug") async def get_visibility_debug( conversation_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): state = await _get_visibility_state(conversation_id, current_user=current_user, db=db) observability = build_runtime_observability_report( state=state, feature_flags=dict(state.get("feature_flags") or {}), ) retrospective_store = SessionRetrospectiveStore(db) artifact_store = LearningArtifactStore(db) recent_retrospectives = await retrospective_store.list_recent( user_id=current_user.id, limit=5, ) recent_artifacts = await artifact_store.list_recent( user_id=current_user.id, limit=10, ) return { "conversation_id": conversation_id, "observability": observability, "skill_shortlist": list(state.get("skill_shortlist") or []), "retrospective_shortlist": list(state.get("retrospective_shortlist") or []), "merge_report": state.get("merge_report"), "verification_report": state.get("verification_report"), "recent_retrospectives": [ { "id": item.id, "task_type": item.task_type, "summary": item.summary_text, "execution_mode": item.execution_mode, "verification_status": item.verification_status, "recorded_at": item.recorded_at.isoformat() if item.recorded_at else None, } for item in recent_retrospectives ], "recent_learning_artifacts": [ { "id": item.id, "artifact_type": item.artifact_type, "artifact_key": item.artifact_key, "summary": item.summary_text, "recorded_at": item.recorded_at.isoformat() if item.recorded_at else None, } for item in recent_artifacts ], } @router.post("", response_model=AgentOut, status_code=201) async def create_agent( data: AgentCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not current_user.is_superuser: raise HTTPException(status_code=403, detail="仅管理员可创建 Agent") if not data.spawn_permission: raise HTTPException(status_code=400, detail="缺少 spawn_permission,禁止直接创建 runtime agent") if data.role not in ALLOWED_AGENT_ROLES: raise HTTPException(status_code=400, detail="不支持的 Agent 角色") agent = Agent( name=data.name, role=data.role, description=data.description, system_prompt=data.system_prompt, ) db.add(agent) await db.commit() await db.refresh(agent) return agent @router.get("/{agent_id}", response_model=AgentOut) async def get_agent( agent_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute(select(Agent).where(Agent.id == agent_id)) agent = result.scalar_one_or_none() if not agent: raise HTTPException(status_code=404, detail="Agent 不存在") return agent