diff --git a/backend/api/v1/chat.py b/backend/api/v1/chat.py index ee4f6368..c4a27e96 100644 --- a/backend/api/v1/chat.py +++ b/backend/api/v1/chat.py @@ -11,18 +11,23 @@ v1 薄闭环: 唯一写动作 = 启停 source。验证通后按同模式扩 trigger_task / update_schedule。 """ +import asyncio import json import logging import re +from contextvars import ContextVar +from collections.abc import Awaitable, Callable from typing import Any, Literal, Optional from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import StreamingResponse from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from backend.control.agent_control import ACTION_REGISTRY, agent_control_service -from backend.database import get_db +from backend.database import AsyncSessionLocal, get_db +from backend.models.agent_run import AgentRun, AgentRunEvent, AgentSession from backend.models.provider import ModelProvider from backend.schemas.common import ApiResponse from backend.security.identity import RequestIdentity, get_request_identity @@ -35,6 +40,44 @@ MAX_TOOL_STEPS = 5 +ActivitySink = Callable[[dict[str, Any]], Awaitable[None]] +_activity_sink: ContextVar[ActivitySink | None] = ContextVar("chat_activity_sink", default=None) +_background_runs: set[asyncio.Task] = set() + +_PUBLIC_TOOL_LABELS = { + "list_sources": ("检查数据源", "数据源"), + "list_schedules": ("检查调度计划", "调度计划"), + "list_tasks": ("检查最近任务", "采集任务"), + "list_providers": ("检查模型连接", "模型提供商"), + "toggle_source": ("变更数据源状态", "数据源"), + "trigger_task": ("启动采集任务", "数据源"), + "update_schedule": ("更新调度计划", "调度计划"), + "update_provider": ("更新模型配置", "模型提供商"), +} + + +async def _emit_activity(event_type: str, label: str, detail: str, **extra: Any) -> None: + sink = _activity_sink.get() + if sink is not None: + await sink({"type": event_type, "label": label, "detail": detail, **extra}) + + +def _tool_public_description(name: str, args: dict[str, Any]) -> tuple[str, str, str | None]: + label, target_type = _PUBLIC_TOOL_LABELS.get(name, ("执行操作", "系统对象")) + target_id = next( + (str(args[key]) for key in ("source_id", "schedule_id", "provider_id") if args.get(key)), + None, + ) + return label, target_type, target_id + + +def _result_public_summary(result: Any) -> str: + if isinstance(result, list): + return f"找到 {len(result)} 项可用信息" + if isinstance(result, dict) and result.get("error"): + return "未能读取目标信息" + return "已读取目标信息" + SYSTEM_PROMPT = """你是 opencli-admin 的全局操作助手。用户可能位于任意产品页面。\ 你的职责: 根据当前页面和对象上下文解释系统状态，并在已有工具覆盖范围内按用户意图查询或修改后端配置。 @@ -171,6 +214,7 @@ class ChatRequest(BaseModel): provider_id: Optional[str] = None # 当前页面、项目或选中对象上下文，注入给 agent 当指代背景 context: Optional[dict[str, Any]] = None + session_id: str | None = None class Proposal(BaseModel): @@ -193,6 +237,73 @@ class ConfirmRequest(BaseModel): proposal: Proposal +async def _create_durable_run(body: ChatRequest, identity: RequestIdentity | None) -> AgentRun: + """Create a durable run before work begins so clients can reconnect immediately.""" + async with AsyncSessionLocal() as session: + agent_session: AgentSession | None = None + if body.session_id: + agent_session = await session.get(AgentSession, body.session_id) + if agent_session is None: + agent_session = AgentSession( + workspace_id=_workspace_id(body.context), + actor_subject=identity.subject if identity else None, + context=body.context or {}, + ) + session.add(agent_session) + await session.flush() + goal = next( + (message.content for message in reversed(body.messages) if message.role == "user"), + "", + ) + run = AgentRun( + session_id=agent_session.id, + status="queued", + goal=goal, + request_payload={ + "messages": [message.model_dump() for message in body.messages], + "context": body.context or {}, + }, + ) + session.add(run) + await session.commit() + await session.refresh(run) + return run + + +async def _record_durable_event(run_id: str, event: dict[str, Any]) -> dict[str, Any]: + """Append one public event atomically. Event payloads are deliberately already redacted.""" + async with AsyncSessionLocal() as session: + run = await session.get(AgentRun, run_id, with_for_update=True) + if run is None: + raise RuntimeError("Agent run disappeared") + sequence = run.next_event_sequence + run.next_event_sequence += 1 + payload = {"sequence": sequence, **event} + session.add( + AgentRunEvent( + run_id=run.id, + sequence=sequence, + event_type=event["type"], + payload=payload, + ) + ) + await session.commit() + return payload + + +async def _finish_durable_run( + run_id: str, *, reply: dict[str, Any] | None = None, error: str | None = None +) -> None: + async with AsyncSessionLocal() as session: + run = await session.get(AgentRun, run_id) + if run is None: + return + run.status = "failed" if error else "completed" + run.reply_payload = reply + run.error_message = error + await session.commit() + + # ── provider → AsyncOpenAI client ─────────────────────────────────────────── async def _pick_provider(db: AsyncSession, provider_id: Optional[str]) -> ModelProvider: if provider_id: @@ -347,9 +458,21 @@ async def chat( identity: RequestIdentity | None = Depends(_optional_request_identity), db: AsyncSession = Depends(get_db), ) -> ApiResponse: + await _emit_activity( + "phase.changed", + "理解目标", + "正在结合当前页面、工作区和选中对象理解请求。", + state="completed", + ) provider = await _pick_provider(db, body.provider_id) client = await _build_client(provider) model = provider.default_model or "gpt-4o-mini" + await _emit_activity( + "phase.changed", + "制定执行路径", + "已选择可用模型，正在判断需要读取的信息和可能的操作。", + state="active", + ) system = SYSTEM_PROMPT if body.context: @@ -362,6 +485,12 @@ async def chat( messages += [{"role": m.role, "content": m.content} for m in body.messages] for _step in range(MAX_TOOL_STEPS): + await _emit_activity( + "phase.changed", + "分析当前状态", + "正在根据已获得的信息决定下一步。", + state="active", + ) try: response = await client.chat.completions.create( model=model, messages=messages, tools=TOOLS, tool_choice="auto" @@ -374,12 +503,26 @@ async def chat( tool_calls = msg.tool_calls or [] if not tool_calls: + await _emit_activity( + "run.completed", + "处理完成", + "已生成基于本次执行信息的结果摘要。", + state="completed", + ) return ApiResponse.ok(ChatReply(type="message", content=msg.content or "")) # 写工具命中 → 立即返回 proposal (不执行, 不继续推理) for tc in tool_calls: if tc.function.name in WRITE_TOOLS: args = _safe_json(tc.function.arguments) + label, target_type, target_id = _tool_public_description(tc.function.name, args) + await _emit_activity( + "tool.completed", + label, + "已定位目标并准备变更方案。", + state="completed", + target={"type": target_type, "id": target_id}, + ) proposal = await _build_proposal( db, tc.function.name, @@ -387,6 +530,13 @@ async def chat( identity=_require_write_identity(identity), workspace_id=_workspace_id(body.context), ) + await _emit_activity( + "approval.required", + "等待确认", + proposal.summary, + state="attention", + target={"type": target_type, "id": target_id}, + ) return ApiResponse.ok(ChatReply(type="proposal", proposal=proposal)) # 只读工具 → 执行, 喂回结果, 继续循环 @@ -405,7 +555,23 @@ async def chat( } ) for tc in tool_calls: - result = await _run_read_tool(db, tc.function.name, _safe_json(tc.function.arguments)) + args = _safe_json(tc.function.arguments) + label, target_type, target_id = _tool_public_description(tc.function.name, args) + await _emit_activity( + "tool.started", + label, + f"正在读取{target_type}的当前状态。", + state="active", + target={"type": target_type, "id": target_id}, + ) + result = await _run_read_tool(db, tc.function.name, args) + await _emit_activity( + "tool.completed", + label, + _result_public_summary(result), + state="completed", + target={"type": target_type, "id": target_id}, + ) messages.append( {"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False)} ) @@ -413,6 +579,147 @@ async def chat( return ApiResponse.ok(ChatReply(type="message", content="(达到工具调用步数上限, 请换个说法再试)")) +@router.post("/stream") +async def chat_stream( + body: ChatRequest, + identity: RequestIdentity | None = Depends(_optional_request_identity), + db: AsyncSession = Depends(get_db), +) -> StreamingResponse: + """Stream public execution facts as NDJSON while the existing chat run executes. + + Events deliberately contain no model reasoning, raw tool arguments, credentials, or + unbounded tool results. The terminal ``reply`` event preserves the established ChatReply + contract so confirmation continues through the governed endpoint. + """ + + durable_run = await _create_durable_run(body, identity) + + async def event_source(): + queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue() + + async def emit(event: dict[str, Any]) -> None: + await queue.put(await _record_durable_event(durable_run.id, event)) + + async def produce() -> None: + token = _activity_sink.set(emit) + try: + async with AsyncSessionLocal() as run_db: + run = await run_db.get(AgentRun, durable_run.id) + if run: + run.status = "running" + await run_db.commit() + await emit( + { + "type": "run.started", + "label": "开始处理", + "detail": "已接收请求，正在建立执行上下文。", + "state": "active", + } + ) + async with AsyncSessionLocal() as run_db: + response = await chat(body, identity, run_db) + await emit( + { + "type": "reply", + "label": "结果已就绪", + "detail": "本次处理已返回结果。", + "state": "completed", + "reply": response.data.model_dump(mode="json"), + } + ) + await _finish_durable_run( + durable_run.id, reply=response.data.model_dump(mode="json") + ) + except HTTPException as exc: + await emit( + { + "type": "run.failed", + "label": "处理未完成", + "detail": str(exc.detail), + "state": "failed", + "status": exc.status_code, + "recovery": "检查连接或目标状态后重试。", + } + ) + await _finish_durable_run(durable_run.id, error=str(exc.detail)) + except Exception: + logger.exception("chat stream failed") + await emit( + { + "type": "run.failed", + "label": "处理未完成", + "detail": "Agent 暂时无法完成这项任务。", + "state": "failed", + "status": 500, + "recovery": "稍后重试，或调整请求后继续。", + } + ) + await _finish_durable_run(durable_run.id, error="Agent run failed") + finally: + _activity_sink.reset(token) + await queue.put(None) + + task = asyncio.create_task(produce()) + _background_runs.add(task) + task.add_done_callback(_background_runs.discard) + try: + while True: + event = await queue.get() + if event is None: + break + yield json.dumps(event, ensure_ascii=False) + "\n" + finally: + # A disconnected client can replay the persisted events; do not cancel work. + pass + + return StreamingResponse( + event_source(), + media_type="application/x-ndjson", + headers={ + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "X-Agent-Run-Id": durable_run.id, + "X-Agent-Session-Id": durable_run.session_id, + }, + ) + + +@router.get("/runs/{run_id}", response_model=ApiResponse[dict[str, Any]]) +async def get_chat_run(run_id: str, db: AsyncSession = Depends(get_db)) -> ApiResponse: + run = await db.get(AgentRun, run_id) + if run is None: + raise HTTPException(status_code=404, detail="Agent run not found") + return ApiResponse.ok( + { + "id": run.id, + "session_id": run.session_id, + "status": run.status, + "goal": run.goal, + "reply": run.reply_payload, + "error": run.error_message, + "created_at": run.created_at, + "updated_at": run.updated_at, + } + ) + + +@router.get("/runs/{run_id}/events", response_model=ApiResponse[list[dict[str, Any]]]) +async def get_chat_run_events( + run_id: str, after_sequence: int = 0, db: AsyncSession = Depends(get_db) +) -> ApiResponse: + if await db.get(AgentRun, run_id) is None: + raise HTTPException(status_code=404, detail="Agent run not found") + events = ( + await db.scalars( + select(AgentRunEvent) + .where(AgentRunEvent.run_id == run_id) + .where(AgentRunEvent.sequence > after_sequence) + .order_by(AgentRunEvent.sequence) + ) + ).all() + return ApiResponse.ok([event.payload for event in events]) + + @router.post("/confirm", response_model=ApiResponse[dict]) async def confirm( body: ConfirmRequest, diff --git a/backend/migrations/versions/a8b9c0d1e2f3_add_durable_agent_runs.py b/backend/migrations/versions/a8b9c0d1e2f3_add_durable_agent_runs.py new file mode 100644 index 00000000..f26b22a8 --- /dev/null +++ b/backend/migrations/versions/a8b9c0d1e2f3_add_durable_agent_runs.py @@ -0,0 +1,72 @@ +"""add durable Agent sessions, runs, and public events + +Revision ID: a8b9c0d1e2f3 +Revises: k8l9m0n1o2p3 +Create Date: 2026-08-06 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "a8b9c0d1e2f3" +down_revision = "k8l9m0n1o2p3" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "agent_sessions", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("workspace_id", sa.String(36), nullable=True), + sa.Column("actor_subject", sa.String(255), nullable=True), + sa.Column("context", sa.JSON(), nullable=False), + ) + op.create_index("ix_agent_sessions_workspace_id", "agent_sessions", ["workspace_id"]) + op.create_index("ix_agent_sessions_actor_subject", "agent_sessions", ["actor_subject"]) + op.create_table( + "agent_runs", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.Column( + "session_id", + sa.String(36), + sa.ForeignKey("agent_sessions.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("kind", sa.String(32), nullable=False), + sa.Column("status", sa.String(32), nullable=False), + sa.Column("goal", sa.Text(), nullable=False), + sa.Column("request_payload", sa.JSON(), nullable=False), + sa.Column("reply_payload", sa.JSON(), nullable=True), + sa.Column("error_message", sa.Text(), nullable=True), + sa.Column("next_event_sequence", sa.Integer(), nullable=False), + ) + op.create_index("ix_agent_runs_session_id", "agent_runs", ["session_id"]) + op.create_index("ix_agent_runs_status", "agent_runs", ["status"]) + op.create_table( + "agent_run_events", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.Column( + "run_id", + sa.String(36), + sa.ForeignKey("agent_runs.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("event_type", sa.String(64), nullable=False), + sa.Column("payload", sa.JSON(), nullable=False), + sa.UniqueConstraint("run_id", "sequence", name="ux_agent_run_events_run_id_sequence"), + ) + op.create_index("ix_agent_run_events_run_id", "agent_run_events", ["run_id"]) + + +def downgrade() -> None: + op.drop_table("agent_run_events") + op.drop_table("agent_runs") + op.drop_table("agent_sessions") diff --git a/backend/models/__init__.py b/backend/models/__init__.py index 0da337e8..8d586656 100644 --- a/backend/models/__init__.py +++ b/backend/models/__init__.py @@ -75,6 +75,7 @@ from backend.models.worker import WorkerNode from backend.models.workflow import Project, Workflow, WorkflowDraft, WorkflowVersion from backend.models.workflow_run import WorkflowRun, WorkflowRunEvent +from backend.models.agent_run import AgentRun, AgentRunEvent, AgentSession __all__ = [ "TimestampMixin", @@ -152,4 +153,7 @@ "WorkflowVersion", "WorkflowRun", "WorkflowRunEvent", + "AgentSession", + "AgentRun", + "AgentRunEvent", ] diff --git a/backend/models/agent_run.py b/backend/models/agent_run.py new file mode 100644 index 00000000..2e03f841 --- /dev/null +++ b/backend/models/agent_run.py @@ -0,0 +1,57 @@ +"""Durable public execution records for interactive Agent runs.""" + +from sqlalchemy import JSON, ForeignKey, Index, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from backend.models.base import TimestampMixin + + +class AgentSession(TimestampMixin): + __tablename__ = "agent_sessions" + + workspace_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + actor_subject: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + context: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + + runs: Mapped[list["AgentRun"]] = relationship( + "AgentRun", back_populates="session", cascade="all, delete-orphan" + ) + + +class AgentRun(TimestampMixin): + __tablename__ = "agent_runs" + + session_id: Mapped[str] = mapped_column( + String(36), ForeignKey("agent_sessions.id", ondelete="CASCADE"), nullable=False, index=True + ) + kind: Mapped[str] = mapped_column(String(32), nullable=False, default="chat") + status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", index=True) + goal: Mapped[str] = mapped_column(Text, nullable=False, default="") + request_payload: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + reply_payload: Mapped[dict | None] = mapped_column(JSON, nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + next_event_sequence: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + + session: Mapped[AgentSession] = relationship("AgentSession", back_populates="runs") + events: Mapped[list["AgentRunEvent"]] = relationship( + "AgentRunEvent", + back_populates="run", + cascade="all, delete-orphan", + order_by="AgentRunEvent.sequence", + ) + + +class AgentRunEvent(TimestampMixin): + __tablename__ = "agent_run_events" + __table_args__ = ( + Index("ux_agent_run_events_run_id_sequence", "run_id", "sequence", unique=True), + ) + + run_id: Mapped[str] = mapped_column( + String(36), ForeignKey("agent_runs.id", ondelete="CASCADE"), nullable=False, index=True + ) + sequence: Mapped[int] = mapped_column(Integer, nullable=False) + event_type: Mapped[str] = mapped_column(String(64), nullable=False) + payload: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + + run: Mapped[AgentRun] = relationship("AgentRun", back_populates="events") diff --git a/docs/backend-capability-exposure-matrix.yaml b/docs/backend-capability-exposure-matrix.yaml index 5a4a3961..9337db25 100644 --- a/docs/backend-capability-exposure-matrix.yaml +++ b/docs/backend-capability-exposure-matrix.yaml @@ -1,6 +1,6 @@ version: 1 source: backend.main.app.openapi -openapi_operation_count: 231 +openapi_operation_count: 234 allowed_dispositions: - operator_ui - studio_binding @@ -237,6 +237,14 @@ capability_groups: owner: opencli-admin wrapper_names: [] workflow_node_ids: [] +- capability_id: agent.runs + label: Agent 运行与可观测 + projection: operator_resource + distribution: builtin + lifecycle: active + owner: opencli-admin + wrapper_names: [] + workflow_node_ids: [] operations: - method: GET path: /api/v1/agents @@ -2317,6 +2325,33 @@ operations: decision: Start only published project workflows through the Studio runtime boundary. target_epic: Epic 8 capability_id: studio.workflow +- method: GET + path: /api/v1/chat/runs/{run_id} + operation_id: get_chat_run_api_v1_chat_runs__run_id__get + disposition: studio_binding + frontend_route: /operations-agents + wrapper: null + decision: Agent run detail for the Operations/Agents page (durable agent runs, ex-PR #61 T2). + target_epic: Epic 8 + capability_id: agent.runs +- method: GET + path: /api/v1/chat/runs/{run_id}/events + operation_id: get_chat_run_events_api_v1_chat_runs__run_id__events_get + disposition: studio_binding + frontend_route: /operations-agents + wrapper: null + decision: Agent run event stream consumed by the global agent dock live view. + target_epic: Epic 8 + capability_id: agent.runs +- method: POST + path: /api/v1/chat/stream + operation_id: chat_stream_api_v1_chat_stream_post + disposition: studio_binding + frontend_route: /operations-agents + wrapper: null + decision: Streaming chat endpoint for observable agent execution. + target_epic: Epic 8 + capability_id: agent.runs unreferenced_wrappers: - wrapper: getOperationsAgentVersion operation_id: get_agent_version_api_v1_workspaces__workspace_id__operations_agents__agent_id__versions__version_number__get diff --git a/frontend/app/(app)/operations-agents/page.tsx b/frontend/app/(app)/operations-agents/page.tsx index 2476c970..07cb5fac 100644 --- a/frontend/app/(app)/operations-agents/page.tsx +++ b/frontend/app/(app)/operations-agents/page.tsx @@ -51,6 +51,15 @@ function parseJsonObject(value: string, label: string) { return parsed as Record } +function publicRunSummary(payload: Record | null) { + if (!payload) return null + const values = Object.entries(payload) + .filter(([, value]) => ['string', 'number', 'boolean'].includes(typeof value)) + .slice(0, 4) + .map(([key, value]) => `${key}: ${String(value)}`) + return values.length ? values.join(' · ') : '已生成结构化执行结果，可在运行记录中审计。' +} + function ContractEditor({ workspaceId, agent }: { workspaceId: string; agent: OperationsAgent }) { const draft = useOperationsAgentDraft(workspaceId, agent.id) const versions = useOperationsAgentVersions(workspaceId, agent.id) @@ -346,7 +355,7 @@ export default function OperationsAgentsPage() {
SESSION OUTPUT
- {(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).length ?
{(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).map((run) =>
[{run.status}] {new Date(run.updated_at).toLocaleString()}
{run.trigger_type} → {run.target_resource_type}/{run.target_resource_id}
profile v{run.profile_version} · agent v{run.published_version}
{run.error_message ?
{run.error_message}
: null}{run.output_payload ? 
{JSON.stringify(run.output_payload, null, 2)}
: null}
)}
:

还没有会话输出

智能体收到任务后，这里会显示真实的 CLI 活动和运行状态。

} + {(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).length ?
{(activity.data ?? []).filter((run) => run.operations_agent_id === selectedAgent.id).map((run) =>
{run.status === 'queued' ? '等待执行' : run.status === 'running' ? '正在执行' : run.status === 'completed' ? '已完成' : run.status === 'paused' ? '等待确认' : run.status === 'cancelled' ? '已取消' : '执行失败'}

目标：{run.target_resource_type} · {run.target_resource_id}

{run.error_message ?
{run.error_message}

检查目标状态后可以重新启动。

: null}{publicRunSummary(run.output_payload) ?
结果：{publicRunSummary(run.output_payload)}
: null}
)}
:

还没有执行活动

智能体收到任务后，这里会显示目标、当前状态和结果摘要。

}
diff --git a/frontend/components/shell/global-agent-dock.tsx b/frontend/components/shell/global-agent-dock.tsx index 148599db..17fc2c77 100644 --- a/frontend/components/shell/global-agent-dock.tsx +++ b/frontend/components/shell/global-agent-dock.tsx @@ -1,7 +1,7 @@ 'use client' import { useQueryClient } from '@tanstack/react-query' -import { Bot, Check, Loader2, Send, ShieldCheck, X } from 'lucide-react' +import { Bot, Check, CircleAlert, CircleCheck, Clock3, Loader2, Monitor, RotateCcw, Send, ShieldCheck, Sparkles, X } from 'lucide-react' import { usePathname } from 'next/navigation' import { FormEvent, KeyboardEvent, useState } from 'react' @@ -16,7 +16,7 @@ import { } from '@/components/ui/sheet' import { Textarea } from '@/components/ui/textarea' import { apiClient } from '@/lib/api/client' -import type { ApiResponse } from '@/lib/api/types' +import { getApiAuthHeaders } from '@/lib/api/auth-headers' import { ROUTE_LABELS } from '@/lib/navigation' type AgentMessage = { @@ -40,6 +40,45 @@ type AgentReply = { proposal?: AgentProposal | null } +type ActivityState = 'active' | 'complete' | 'attention' + +type Activity = { + label: string + detail: string + state: ActivityState + target?: { type?: string; id?: string | null } +} + +type AgentRunEvent = { + sequence: number + type: string + label: string + detail: string + state?: 'active' | 'completed' | 'attention' | 'failed' + target?: { type?: string; id?: string | null } + recovery?: string + reply?: AgentReply +} + +function activityFromEvent(event: AgentRunEvent): Activity { + return { + label: event.label, + detail: event.recovery ? `${event.detail} ${event.recovery}` : event.detail, + state: event.state === 'completed' ? 'complete' : event.state === 'failed' || event.state === 'attention' ? 'attention' : 'active', + target: event.target, + } +} + +function activityForReply(reply: AgentReply): Activity[] { + if (reply.type === 'proposal' && reply.proposal) { + return [ + { label: '已定位操作对象', detail: reply.proposal.summary, state: 'complete' }, + { label: '等待你的确认', detail: '这是一次会改变软件状态的操作。确认后才会执行。', state: 'attention' }, + ] + } + return [{ label: '已完成处理', detail: '已基于当前可访问的数据生成结果。', state: 'complete' }] +} + export function GlobalAgentDock({ open, onOpenChange, @@ -55,6 +94,12 @@ export function GlobalAgentDock({ const [error, setError] = useState(null) const [sending, setSending] = useState(false) const [confirming, setConfirming] = useState(false) + const [goal, setGoal] = useState(null) + const [activities, setActivities] = useState([]) + const [lastFailedProposal, setLastFailedProposal] = useState(null) + const [showLiveSurface, setShowLiveSurface] = useState(false) + const [agentSessionId, setAgentSessionId] = useState(null) + const [agentRunId, setAgentRunId] = useState(null) async function sendMessage(event?: FormEvent) { event?.preventDefault() @@ -66,6 +111,12 @@ export function GlobalAgentDock({ setInput('') setError(null) setSending(true) + setGoal(content) + setActivities([ + { label: '理解你的目标', detail: '正在结合当前页面和选中对象梳理任务。', state: 'complete' }, + { label: '检查可用信息', detail: '正在判断是否需要读取数据或准备操作。', state: 'active' }, + ]) + let activeRunId: string | null = null try { const searchParams = new URLSearchParams(window.location.search) const workspaceId = searchParams.get('workspace') @@ -76,8 +127,12 @@ export function GlobalAgentDock({ const sourceId = searchParams.get('source') ?? pathname.match(/^\/sources\/([^/]+)/)?.[1] ?? null - const response = await apiClient.post>('/chat', { + const response = await fetch('/api/v1/chat/stream', { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getApiAuthHeaders() }, + body: JSON.stringify({ messages: nextMessages, + session_id: agentSessionId, context: { surface: ROUTE_LABELS[pathname] ?? pathname, pathname, @@ -87,8 +142,44 @@ export function GlobalAgentDock({ workflow_id: workflowId, source_id: sourceId, }, + }), }) - const reply = response.data.data + if (!response.ok || !response.body) throw new Error(`Agent 请求失败（${response.status}）`) + + const receivedRunId = response.headers.get('X-Agent-Run-Id') + const receivedSessionId = response.headers.get('X-Agent-Session-Id') + if (receivedRunId) { + activeRunId = receivedRunId + setAgentRunId(receivedRunId) + } + if (receivedSessionId) setAgentSessionId(receivedSessionId) + const reader = response.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + let reply: AgentReply | null = null + let streamError: string | null = null + while (true) { + const { value, done } = await reader.read() + buffer += decoder.decode(value, { stream: !done }) + const lines = buffer.split('\n') + buffer = lines.pop() ?? '' + for (const line of lines) { + if (!line.trim()) continue + const runEvent = JSON.parse(line) as AgentRunEvent + if (runEvent.type === 'reply' && runEvent.reply) { + reply = runEvent.reply + } else { + setActivities((current) => { + const next = [...current.filter((item) => item.state !== 'active'), activityFromEvent(runEvent)] + return next.slice(-8) + }) + } + if (runEvent.type === 'run.failed') streamError = runEvent.detail + } + if (done) break + } + if (streamError) throw new Error(streamError) + if (!reply) throw new Error('Agent 执行结束但没有返回结果') if (reply.type === 'proposal' && reply.proposal) { setProposal(reply.proposal) } else { @@ -97,25 +188,54 @@ export function GlobalAgentDock({ { role: 'assistant', content: reply.content?.trim() || '没有返回内容。' }, ]) } + setActivities((current) => [...current.filter((item) => item.state !== 'active'), ...activityForReply(reply)].slice(-8)) } catch (reason) { - setError(reason instanceof Error ? reason.message : 'Agent 暂时不可用') + const recoverableRunId = activeRunId ?? agentRunId + if (recoverableRunId) { + try { + const recovery = await apiClient.get(`/chat/runs/${recoverableRunId}/events`) + for (const replayed of recovery.data ?? []) { + if (replayed.type === 'reply' && replayed.reply) { + if (replayed.reply.type === 'proposal' && replayed.reply.proposal) setProposal(replayed.reply.proposal) + else setMessages((current) => [...current, { role: 'assistant', content: replayed.reply?.content?.trim() || '' }]) + } + } + } catch { + // Keep the stream error as the primary recovery signal. + } + } + const message = reason instanceof Error ? reason.message : 'Agent 暂时不可用' + setError(message) + setActivities([ + { label: '暂时无法完成理解', detail: message, state: 'attention' }, + { label: '恢复方式', detail: '请检查模型连接后重试，或换一种说法继续。', state: 'attention' }, + ]) } finally { setSending(false) } } - async function confirmProposal() { - if (!proposal || confirming) return + async function confirmProposal(proposalToConfirm = proposal) { + if (!proposalToConfirm || confirming) return setError(null) setConfirming(true) + setLastFailedProposal(null) + setActivities([ + { label: '已获得你的确认', detail: proposalToConfirm.summary, state: 'complete' }, + { label: '正在执行操作', detail: '系统正在应用这项变更。', state: 'active' }, + ]) try { - await apiClient.post('/chat/confirm', { proposal }) + await apiClient.post('/chat/confirm', { proposal: proposalToConfirm }) setMessages((current) => [ ...current, - { role: 'assistant', content: `已执行：${proposal.summary}` }, + { role: 'assistant', content: `已完成：${proposalToConfirm.summary}` }, ]) setProposal(null) await queryClient.invalidateQueries() + setActivities([ + { label: '操作已完成', detail: proposalToConfirm.summary, state: 'complete' }, + { label: '界面已同步', detail: '已刷新相关数据；你现在看到的是最新状态。', state: 'complete' }, + ]) } catch (reason) { const status = reason instanceof Error && 'status' in reason ? reason.status : undefined const message = reason instanceof Error ? reason.message : '操作执行失败' @@ -124,6 +244,11 @@ export function GlobalAgentDock({ ? `提案已失效或目标已变化：${message}。请拒绝后重新发起。` : message, ) + setLastFailedProposal(proposalToConfirm) + setActivities([ + { label: '操作未完成', detail: message, state: 'attention' }, + { label: '可恢复', detail: '检查目标状态后，可重新执行或回到对话调整请求。', state: 'attention' }, + ]) } finally { setConfirming(false) } @@ -178,14 +303,66 @@ export function GlobalAgentDock({ Agent 正在处理
) : null} + {goal ? ( +
+
+ + 正在处理 +
+

目标：{goal}

+
    + {activities.map((activity, index) => { + const Icon = activity.state === 'complete' ? CircleCheck : activity.state === 'attention' ? CircleAlert : Clock3 + return ( +
  1. + +
    +

    {activity.label}

    +

    {activity.detail}

    + {activity.target?.type ? ( +

    + 对象：{activity.target.type}{activity.target.id ? ` · ${activity.target.id}` : ''} +

    + ) : null} +
    +
    2. + ) + })} +
+
+ ) : null} + {goal ? ( +
+
+
+

+ + 软件现场 +

+

查看内置浏览器正在发生的实际变化。

+
+ +
+ {showLiveSurface ? ( +