diff --git a/PixPin_2026-05-26_21-54-34.png b/PixPin_2026-05-26_21-54-34.png deleted file mode 100644 index ac6aa1de..00000000 Binary files a/PixPin_2026-05-26_21-54-34.png and /dev/null differ diff --git a/backend/package/yuxi/repositories/agent_run_repository.py b/backend/package/yuxi/repositories/agent_run_repository.py index 5fe0f791..d387c07b 100644 --- a/backend/package/yuxi/repositories/agent_run_repository.py +++ b/backend/package/yuxi/repositories/agent_run_repository.py @@ -23,12 +23,19 @@ class AgentRunRepository: result = await self.db.execute(select(AgentRun).where(AgentRun.request_id == request_id)) return result.scalar_one_or_none() - async def get_run_for_user(self, run_id: str, uid: str) -> AgentRun | None: + async def get_resume_run(self, parent_run_id: str, resume_request_id: str) -> AgentRun | None: result = await self.db.execute( - select(AgentRun).where(and_(AgentRun.id == run_id, AgentRun.uid == str(uid))) + select(AgentRun).where( + AgentRun.parent_run_id == parent_run_id, + AgentRun.resume_request_id == resume_request_id, + ) ) return result.scalar_one_or_none() + async def get_run_for_user(self, run_id: str, uid: str) -> AgentRun | None: + result = await self.db.execute(select(AgentRun).where(and_(AgentRun.id == run_id, AgentRun.uid == str(uid)))) + return result.scalar_one_or_none() + async def create_run( self, *, @@ -38,6 +45,11 @@ class AgentRunRepository: uid: str, request_id: str, input_payload: dict, + conversation_id: int | None = None, + parent_run_id: str | None = None, + run_type: str = "chat", + resume_request_id: str | None = None, + checkpoint_thread_id: str | None = None, ) -> AgentRun: run = AgentRun( id=run_id, @@ -45,6 +57,11 @@ class AgentRunRepository: agent_id=agent_id, uid=str(uid), request_id=request_id, + conversation_id=conversation_id, + parent_run_id=parent_run_id, + run_type=run_type, + resume_request_id=resume_request_id, + checkpoint_thread_id=checkpoint_thread_id or thread_id, input_payload=input_payload or {}, status="pending", ) @@ -52,6 +69,15 @@ class AgentRunRepository: await self.db.flush() return run + async def set_input_message(self, run_id: str, message_id: int) -> AgentRun | None: + run = await self.get_run(run_id) + if not run: + return None + run.input_message_id = message_id + run.updated_at = utc_now_naive() + await self.db.flush() + return run + async def mark_running(self, run_id: str) -> AgentRun | None: run = await self._lock_run(run_id) if not run: diff --git a/backend/package/yuxi/repositories/conversation_repository.py b/backend/package/yuxi/repositories/conversation_repository.py index 67b7dd2b..4b3f77f6 100644 --- a/backend/package/yuxi/repositories/conversation_repository.py +++ b/backend/package/yuxi/repositories/conversation_repository.py @@ -98,6 +98,9 @@ class ConversationRepository: message_type: str = "text", extra_metadata: dict | None = None, image_content: str | None = None, + run_id: str | None = None, + request_id: str | None = None, + delivery_status: str = "complete", ) -> Message: message = Message( conversation_id=conversation_id, @@ -106,6 +109,9 @@ class ConversationRepository: message_type=message_type, extra_metadata=extra_metadata or {}, image_content=image_content, + run_id=run_id, + request_id=request_id, + delivery_status=delivery_status, ) self.db.add(message) @@ -129,6 +135,9 @@ class ConversationRepository: message_type: str = "text", extra_metadata: dict | None = None, image_content: str | None = None, + run_id: str | None = None, + request_id: str | None = None, + delivery_status: str = "complete", ) -> Message | None: conversation = await self.get_conversation_by_thread_id(thread_id) if not conversation: @@ -142,6 +151,9 @@ class ConversationRepository: message_type=message_type, extra_metadata=extra_metadata, image_content=image_content, + run_id=run_id, + request_id=request_id, + delivery_status=delivery_status, ) async def add_tool_call( diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py index e3a59c37..169588db 100644 --- a/backend/package/yuxi/services/agent_run_service.py +++ b/backend/package/yuxi/services/agent_run_service.py @@ -24,7 +24,7 @@ from yuxi.services.run_queue_service import ( publish_cancel_signal, ) from yuxi.storage.postgres.manager import pg_manager -from yuxi.storage.postgres.models_business import User +from yuxi.storage.postgres.models_business import Message, User from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger @@ -39,31 +39,37 @@ def _build_run_response(run) -> dict: "thread_id": run.thread_id, "status": run.status, "request_id": run.request_id, - "stream_url": f"/api/agent/runs/{run.id}/events?after_seq=0-0", + "stream_url": f"/api/agent/runs/{run.id}/events", } -def _format_sse(data: dict, event: str | None = None) -> str: - lines = [] - if event: - lines.append(f"event: {event}") - lines.append(f"data: {json.dumps(data, ensure_ascii=False)}") +def _format_sse(data: dict, event: str, event_id: str | None = None) -> str: + lines = [f"event: {event}", f"data: {json.dumps(data, ensure_ascii=False)}"] + if event_id: + lines.append(f"id: {event_id}") lines.append("") return "\n".join(lines) + "\n" +def _format_heartbeat() -> str: + return ": heartbeat\n\n" + + async def create_agent_run_view( *, - query: str, + query: str | None, agent_id: str, thread_id: str, meta: dict, image_content: str | None, current_uid: str, db: AsyncSession, + resume: object | None = None, + parent_run_id: str | None = None, + resume_request_id: str | None = None, ) -> dict: - if not query: - raise HTTPException(status_code=422, detail="query 不能为空") + if not query and resume is None: + raise HTTPException(status_code=422, detail="query 或 resume 不能为空") if not thread_id: raise HTTPException(status_code=422, detail="thread_id 不能为空") @@ -87,9 +93,22 @@ async def create_agent_run_view( if not agent_manager.get_agent(agent_item.backend_id): raise HTTPException(status_code=404, detail=f"智能体后端 {agent_item.backend_id} 不存在") - request_id = str((meta or {}).get("request_id") or uuid.uuid4()) + run_type = "resume" if resume is not None else "chat" + request_id = str(resume_request_id or (meta or {}).get("request_id") or uuid.uuid4()) config = {"thread_id": thread_id, "agent_id": agent_id} run_repo = AgentRunRepository(db) + if run_type == "resume": + if not parent_run_id: + raise HTTPException(status_code=422, detail="parent_run_id 不能为空") + parent_run = await run_repo.get_run_for_user(parent_run_id, str(current_uid)) + if not parent_run or parent_run.thread_id != thread_id: + raise HTTPException(status_code=404, detail="被恢复的运行任务不存在") + if parent_run.status != "interrupted": + raise HTTPException(status_code=409, detail="只有 interrupted run 可以恢复") + if resume_request_id: + existing_resume = await run_repo.get_resume_run(parent_run_id, resume_request_id) + if existing_resume and existing_resume.uid == str(current_uid): + return _build_run_response(existing_resume) existing = await run_repo.get_run_by_request_id(request_id) if existing and existing.uid == str(current_uid): return _build_run_response(existing) @@ -98,7 +117,11 @@ async def create_agent_run_view( run_id = str(uuid.uuid4()) input_payload = { - "query": query, + "query": query or "", + "resume": resume, + "parent_run_id": parent_run_id, + "resume_request_id": resume_request_id, + "run_type": run_type, "config": config or {}, "image_content": image_content, "agent_id": agent_id, @@ -117,7 +140,38 @@ async def create_agent_run_view( uid=str(current_uid), request_id=request_id, input_payload=input_payload, + conversation_id=conversation.id, + parent_run_id=parent_run_id, + run_type=run_type, + resume_request_id=resume_request_id, + checkpoint_thread_id=thread_id, ) + input_content = query or json.dumps(resume, ensure_ascii=False) + input_metadata = { + "request_id": request_id, + "run_id": run_id, + "run_type": run_type, + "parent_run_id": parent_run_id, + "resume": resume, + "attachments": [], + } + if run_type == "resume": + input_metadata["source"] = "ask_user_question_resume" + + input_message = Message( + conversation_id=conversation.id, + role="user", + content=input_content, + message_type="resume" if run_type == "resume" else "text", + image_content=image_content, + run_id=run_id, + request_id=request_id, + delivery_status="complete", + extra_metadata=input_metadata, + ) + db.add(input_message) + await db.flush() + await run_repo.set_input_message(run_id, input_message.id) await db.commit() except IntegrityError: await db.rollback() @@ -170,7 +224,6 @@ async def stream_agent_run_events( run = await repo.get_run_for_user(run_id, str(current_uid)) if not run: yield _format_sse({"run_id": run_id, "message": "运行任务不存在"}, event="error") - yield _format_sse({"run_id": run_id, "last_seq": last_seq}, event="close") return except asyncio.CancelledError: raise @@ -184,7 +237,6 @@ async def stream_agent_run_events( }, event="error", ) - yield _format_sse({"run_id": run_id, "last_seq": last_seq}, event="close") return try: @@ -199,32 +251,38 @@ async def stream_agent_run_events( }, event="error", ) - yield _format_sse({"run_id": run_id, "last_seq": last_seq}, event="close") return + emitted_terminal = False for event in events: seq = str(event.get("seq") or "0-0") last_seq = seq + event_type = event.get("event_type") or "message" + envelope = event.get("payload") or {} + yield _format_sse(envelope, event=event_type, event_id=seq) + if event_type == "end": + emitted_terminal = True - yield _format_sse( - { - "run_id": run_id, - "seq": seq, - "event_type": event.get("event_type") or "message", - "payload": event.get("payload") or {}, - "ts": event.get("ts"), - }, - event=event.get("event_type") or "message", - ) + if emitted_terminal: + return if run.status in TERMINAL_RUN_STATUSES and not events: terminal_seq = last_seq if terminal_seq in {"", "0-0"}: terminal_seq = await get_last_run_stream_seq(run_id) - + if terminal_seq in {"", "0-0"}: + terminal_seq = None yield _format_sse( - {"run_id": run_id, "status": run.status, "last_seq": terminal_seq}, - event="close", + { + "schema_version": 1, + "run_id": run_id, + "thread_id": run.thread_id, + "event": "end", + "payload": {"status": run.status}, + "created_at": utc_now_naive().isoformat(), + }, + event="end", + event_id=terminal_seq, ) return @@ -232,11 +290,10 @@ async def stream_agent_run_events( elapsed_seconds = (now - started_at).total_seconds() heartbeat_elapsed = (now - last_heartbeat_ts).total_seconds() if heartbeat_elapsed >= SSE_HEARTBEAT_SECONDS: - yield _format_sse({"run_id": run_id, "last_seq": last_seq}, event="heartbeat") + yield _format_heartbeat() last_heartbeat_ts = now if elapsed_seconds >= SSE_MAX_CONNECTION_MINUTES * 60: - yield _format_sse({"run_id": run_id, "last_seq": last_seq}, event="close") return await asyncio.sleep(SSE_POLL_INTERVAL_SECONDS) @@ -248,15 +305,27 @@ async def get_active_run_by_thread(*, thread_id: str, current_uid: str, db: Asyn from sqlalchemy import select from yuxi.storage.postgres.models_business import AgentRun - result = await db.execute( + active_result = await db.execute( select(AgentRun) .where( AgentRun.thread_id == thread_id, AgentRun.uid == str(current_uid), - AgentRun.status.notin_(list(TERMINAL_RUN_STATUSES)), + AgentRun.status.in_(["pending", "running", "cancel_requested"]), ) .order_by(AgentRun.created_at.desc()) .limit(1) ) - run = result.scalar_one_or_none() + run = active_result.scalar_one_or_none() + if not run: + interrupted_result = await db.execute( + select(AgentRun) + .where( + AgentRun.thread_id == thread_id, + AgentRun.uid == str(current_uid), + AgentRun.status == "interrupted", + ) + .order_by(AgentRun.created_at.desc()) + .limit(1) + ) + run = interrupted_result.scalar_one_or_none() return {"run": run.to_dict() if run else None} diff --git a/backend/package/yuxi/services/chat_service.py b/backend/package/yuxi/services/chat_service.py index 7669e385..3b5ed5e8 100644 --- a/backend/package/yuxi/services/chat_service.py +++ b/backend/package/yuxi/services/chat_service.py @@ -711,6 +711,7 @@ async def stream_agent_chat( image_content: str | None, current_user, db, + save_user_message: bool = True, ) -> AsyncIterator[bytes]: start_time = asyncio.get_event_loop().time() @@ -819,21 +820,22 @@ async def stream_agent_chat( init_msg["image_content"] = image_content yield make_chunk(status="init", meta=meta, msg=init_msg) - try: - await conv_repo.add_message_by_thread_id( - thread_id=thread_id, - role="user", - content=query, - message_type=message_type, - image_content=image_content, - extra_metadata={ - "raw_message": human_message.model_dump(), - "request_id": meta.get("request_id"), - "attachments": request_attachments, - }, - ) - except Exception as e: - logger.error(f"Error saving user message: {e}") + if save_user_message: + try: + await conv_repo.add_message_by_thread_id( + thread_id=thread_id, + role="user", + content=query, + message_type=message_type, + image_content=image_content, + extra_metadata={ + "raw_message": human_message.model_dump(), + "request_id": meta.get("request_id"), + "attachments": request_attachments, + }, + ) + except Exception as e: + logger.error(f"Error saving user message: {e}") # 先构建 langgraph_config langgraph_config = {"configurable": {"thread_id": thread_id, "uid": uid}} @@ -1001,8 +1003,7 @@ async def stream_agent_resume( + b"\n" ) - init_msg = {"type": "system", "content": f"Resume with input: {resume_input}"} - yield make_resume_chunk(status="init", meta=meta, msg=init_msg) + yield make_resume_chunk(status="init", meta=meta) resume_command = Command(resume=resume_input) diff --git a/backend/package/yuxi/services/oidc_service.py b/backend/package/yuxi/services/oidc_service.py index 27a38d5a..a4972234 100644 --- a/backend/package/yuxi/services/oidc_service.py +++ b/backend/package/yuxi/services/oidc_service.py @@ -855,7 +855,7 @@ async def oidc_callback_handler(code: str, state: str, db, request: Request | No response_data = { "access_token": jwt_token, "token_type": "bearer", - "uid": user.id, + "user_id": user.id, "username": user.username, "uid": user.uid, "phone_number": user.phone_number, diff --git a/backend/package/yuxi/services/run_queue_service.py b/backend/package/yuxi/services/run_queue_service.py index c0de3bae..e6c13d35 100644 --- a/backend/package/yuxi/services/run_queue_service.py +++ b/backend/package/yuxi/services/run_queue_service.py @@ -159,13 +159,22 @@ async def clear_cancel_signal(run_id: str) -> None: logger.warning(f"Failed to clear cancel signal for run {run_id}: {e}") -async def append_run_stream_event(run_id: str, event_type: str, payload: dict) -> str: +async def append_run_stream_event(run_id: str, event_type: str, payload: dict, *, thread_id: str | None = None) -> str: redis = await get_redis_client() key = _event_stream_key(run_id) - now_ms = int(datetime.now(tz=UTC).timestamp() * 1000) + now = datetime.now(tz=UTC) + now_ms = int(now.timestamp() * 1000) + envelope = { + "schema_version": 1, + "run_id": run_id, + "thread_id": thread_id, + "event": event_type, + "payload": payload or {}, + "created_at": now.isoformat(), + } fields = { "event_type": event_type, - "payload": json.dumps(payload or {}, ensure_ascii=False), + "payload": json.dumps(envelope, ensure_ascii=False), "ts": str(now_ms), } @@ -198,11 +207,22 @@ async def list_run_stream_events( except Exception: payload = {} + event_type = fields.get("event_type") or "message" + if not isinstance(payload, dict) or payload.get("schema_version") != 1: + payload = { + "schema_version": 1, + "run_id": run_id, + "thread_id": None, + "event": event_type, + "payload": payload if isinstance(payload, dict) else {}, + "created_at": None, + } + ts_value = fields.get("ts") events.append( { "seq": str(event_id), - "event_type": fields.get("event_type") or "message", + "event_type": event_type, "payload": payload, "ts": int(ts_value) if ts_value else None, } diff --git a/backend/package/yuxi/services/run_worker.py b/backend/package/yuxi/services/run_worker.py index 78272eae..0d3d1cf6 100644 --- a/backend/package/yuxi/services/run_worker.py +++ b/backend/package/yuxi/services/run_worker.py @@ -11,15 +11,15 @@ from dataclasses import dataclass, field from sqlalchemy import select from sqlalchemy.exc import OperationalError from yuxi.repositories.agent_run_repository import TERMINAL_RUN_STATUSES, AgentRunRepository -from yuxi.services.chat_service import stream_agent_chat +from yuxi.services.chat_service import stream_agent_chat, stream_agent_resume from yuxi.services.mcp_service import ensure_builtin_mcp_servers_in_db -from yuxi.services.skill_service import init_builtin_skills from yuxi.services.run_queue_service import ( append_run_stream_event, clear_cancel_signal, has_cancel_signal, wait_for_cancel_signal, ) +from yuxi.services.skill_service import init_builtin_skills from yuxi.storage.postgres.manager import pg_manager from yuxi.storage.postgres.models_business import User from yuxi.utils.logging_config import logger @@ -77,8 +77,9 @@ class RunContext: class ChunkedEventWriter: - def __init__(self, run_id: str, interval_ms: int = 100, max_chars: int = 512): + def __init__(self, run_id: str, thread_id: str | None, interval_ms: int = 100, max_chars: int = 512): self.run_id = run_id + self.thread_id = thread_id self.interval_seconds = interval_ms / 1000 self.max_chars = max_chars self.buffer: list[dict] = [] @@ -97,7 +98,7 @@ class ChunkedEventWriter: async def flush(self): if not self.buffer: return - await append_run_event(self.run_id, "loading", {"items": self.buffer}) + await append_run_event(self.run_id, "messages", {"items": self.buffer}, thread_id=self.thread_id) self.buffer = [] self.buffer_chars = 0 self.last_flush = time.monotonic() @@ -109,8 +110,8 @@ async def _get_run(run_id: str): return await repo.get_run(run_id) -async def append_run_event(run_id: str, event_type: str, payload: dict): - await append_run_stream_event(run_id, event_type, payload) +async def append_run_event(run_id: str, event_type: str, payload: dict, *, thread_id: str | None = None): + await append_run_stream_event(run_id, event_type, payload, thread_id=thread_id) async def mark_run_running(run_id: str): @@ -169,6 +170,31 @@ def _iter_json_chunks(chunk_bytes: bytes) -> list[dict]: return chunks +def _map_chunk_to_run_event(chunk: dict) -> tuple[str, dict]: + status = chunk.get("status") or "event" + if status == "loading": + return "messages", {"chunk": chunk} + if status == "agent_state": + return "custom", {"name": "yuxi.agent_state", "chunk": chunk, "agent_state": chunk.get("agent_state") or {}} + if status in {"ask_user_question_required", "human_approval_required", "interrupted"}: + reason = "human_approval" if status == "human_approval_required" else status + return "interrupt", {"reason": reason, "chunk": chunk} + if status == "warning": + return "custom", {"name": "yuxi.warning", "chunk": chunk} + if status == "error": + return "error", {"chunk": chunk, "retryable": bool(chunk.get("retryable"))} + if status == "finished": + return "end", {"status": "completed", "chunk": chunk} + return "custom", {"name": f"yuxi.{status}", "chunk": chunk} + + +async def _append_end_event(run_id: str, status: str, *, thread_id: str | None, payload: dict | None = None): + end_payload = {"status": status} + if payload: + end_payload.update(payload) + await append_run_event(run_id, "end", end_payload, thread_id=thread_id) + + async def _consume_stream_with_cancel(agen, run_ctx: RunContext): while True: next_task = asyncio.create_task(agen.__anext__()) @@ -200,11 +226,14 @@ async def process_agent_run(ctx, run_id: str): payload = run.input_payload or {} query = payload.get("query") + resume_input = payload.get("resume") + run_type = payload.get("run_type") or "chat" config = payload.get("config") or {} agent_id = payload.get("agent_id") image_content = payload.get("image_content") uid = payload.get("uid") request_id = payload.get("request_id") + thread_id = config.get("thread_id") or payload.get("thread_id") user = await _load_user(uid) if not user: @@ -229,23 +258,45 @@ async def process_agent_run(ctx, run_id: str): run_ctx = RunContext(run_id=run_id) writer = ChunkedEventWriter( run_id=run_id, + thread_id=thread_id, interval_ms=LOADING_FLUSH_INTERVAL_MS, max_chars=LOADING_FLUSH_MAX_CHARS, ) await run_ctx.start() + await append_run_event( + run_id, + "metadata", + { + "request_id": request_id, + "agent_id": agent_id, + "backend_id": payload.get("backend_id"), + "uid": uid, + }, + thread_id=thread_id, + ) terminal_set = False try: async with pg_manager.get_async_session_context() as db: - stream = stream_agent_chat( - query=query, - agent_id=config.get("agent_id") or agent_id, - thread_id=config.get("thread_id"), - meta=meta, - image_content=image_content, - current_user=user, - db=db, - ) + if run_type == "resume": + stream = stream_agent_resume( + thread_id=thread_id, + resume_input=resume_input, + meta=meta, + current_user=user, + db=db, + ) + else: + stream = stream_agent_chat( + query=query, + agent_id=config.get("agent_id") or agent_id, + thread_id=thread_id, + meta=meta, + image_content=image_content, + current_user=user, + db=db, + save_user_message=False, + ) async for chunk_bytes in _consume_stream_with_cancel(stream, run_ctx): for chunk in _iter_json_chunks(chunk_bytes): @@ -255,10 +306,13 @@ async def process_agent_run(ctx, run_id: str): await writer.flush() status = chunk.get("status") or "event" - await append_run_event(run_id, status, {"chunk": chunk}) + event_type, event_payload = _map_chunk_to_run_event(chunk) + if event_type != "end": + await append_run_event(run_id, event_type, event_payload, thread_id=thread_id) if status == "finished": await mark_run_terminal(run_id, "completed") + await _append_end_event(run_id, "completed", thread_id=thread_id, payload={"chunk": chunk}) terminal_set = True elif status == "error": await mark_run_terminal( @@ -267,6 +321,7 @@ async def process_agent_run(ctx, run_id: str): error_type=chunk.get("error_type") or "stream_error", error_message=chunk.get("error_message") or chunk.get("message"), ) + await _append_end_event(run_id, "failed", thread_id=thread_id, payload={"chunk": chunk}) terminal_set = True elif status == "interrupted": status_value = "cancelled" if await _is_cancel_requested(run_id) else "interrupted" @@ -276,8 +331,9 @@ async def process_agent_run(ctx, run_id: str): error_type=status_value, error_message=chunk.get("message"), ) + await _append_end_event(run_id, status_value, thread_id=thread_id, payload={"chunk": chunk}) terminal_set = True - elif status == "ask_user_question_required": + elif status in {"ask_user_question_required", "human_approval_required"}: questions = chunk.get("questions") if isinstance(chunk, dict) else None first_question = "" if isinstance(questions, list) and questions: @@ -288,9 +344,10 @@ async def process_agent_run(ctx, run_id: str): await mark_run_terminal( run_id, "interrupted", - error_type="ask_user_question_required", + error_type=status, error_message=first_question or "需要用户回答问题", ) + await _append_end_event(run_id, "interrupted", thread_id=thread_id, payload={"chunk": chunk}) terminal_set = True if await run_ctx.is_cancelled(): @@ -298,36 +355,40 @@ async def process_agent_run(ctx, run_id: str): await writer.flush() if not terminal_set: + finished_chunk = {"status": "finished", "request_id": request_id} await mark_run_terminal(run_id, "completed") - await append_run_event(run_id, "finished", {"chunk": {"status": "finished", "request_id": request_id}}) + await _append_end_event(run_id, "completed", thread_id=thread_id, payload={"chunk": finished_chunk}) except asyncio.CancelledError: await writer.flush() + cancel_chunk = {"status": "interrupted", "message": "对话已取消", "request_id": request_id} await append_run_event( run_id, - "interrupted", - {"chunk": {"status": "interrupted", "message": "对话已取消", "request_id": request_id}}, + "interrupt", + {"reason": "cancelled", "chunk": cancel_chunk}, + thread_id=thread_id, ) await mark_run_terminal(run_id, "cancelled", error_type="cancelled", error_message="对话已取消") + await _append_end_event(run_id, "cancelled", thread_id=thread_id, payload={"chunk": cancel_chunk}) logger.info(f"Run cancelled: {run_id}") except Exception as e: await writer.flush() if _is_retryable_exception(e): job_try = _job_try(ctx) logger.warning(f"Run retryable failure {run_id} (try={job_try}): {e}") + retryable_error_chunk = { + "status": "error", + "error_type": "retryable_worker_error", + "error_message": str(e), + "request_id": request_id, + "retryable": True, + "job_try": job_try, + } await append_run_event( run_id, "error", - { - "chunk": { - "status": "error", - "error_type": "retryable_worker_error", - "error_message": str(e), - "request_id": request_id, - "retryable": True, - "job_try": job_try, - } - }, + {"chunk": retryable_error_chunk, "retryable": True}, + thread_id=thread_id, ) if _is_last_try(ctx): await mark_run_terminal( @@ -336,6 +397,12 @@ async def process_agent_run(ctx, run_id: str): error_type="retryable_worker_error", error_message=str(e), ) + await _append_end_event( + run_id, + "failed", + thread_id=thread_id, + payload={"chunk": retryable_error_chunk}, + ) logger.error(f"Run failed after retries exhausted {run_id}: {e}") return @@ -344,20 +411,21 @@ async def process_agent_run(ctx, run_id: str): raise RetryableRunError(str(e)) from e logger.error(f"Run failed {run_id}: {e}") + error_chunk = { + "status": "error", + "error_type": "worker_error", + "error_message": str(e), + "request_id": request_id, + "retryable": False, + } await append_run_event( run_id, "error", - { - "chunk": { - "status": "error", - "error_type": "worker_error", - "error_message": str(e), - "request_id": request_id, - "retryable": False, - } - }, + {"chunk": error_chunk, "retryable": False}, + thread_id=thread_id, ) await mark_run_terminal(run_id, "failed", error_type="worker_error", error_message=str(e)) + await _append_end_event(run_id, "failed", thread_id=thread_id, payload={"chunk": error_chunk}) return finally: await run_ctx.close() diff --git a/backend/package/yuxi/storage/postgres/manager.py b/backend/package/yuxi/storage/postgres/manager.py index f563a0f2..80e53da8 100644 --- a/backend/package/yuxi/storage/postgres/manager.py +++ b/backend/package/yuxi/storage/postgres/manager.py @@ -447,23 +447,6 @@ class PostgresManager(metaclass=SingletonMeta): updated_at TIMESTAMPTZ DEFAULT NOW() ) """, - """ - CREATE TABLE IF NOT EXISTS agent_runs ( - id VARCHAR(64) PRIMARY KEY, - thread_id VARCHAR(64) NOT NULL, - agent_id VARCHAR(64) NOT NULL, - uid VARCHAR(64) NOT NULL, - status VARCHAR(32) NOT NULL DEFAULT 'pending', - request_id VARCHAR(64) NOT NULL UNIQUE, - input_payload JSONB NOT NULL DEFAULT '{}'::jsonb, - error_type VARCHAR(64), - error_message TEXT, - started_at TIMESTAMPTZ, - finished_at TIMESTAMPTZ, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() - ) - """, "CREATE INDEX IF NOT EXISTS idx_agent_runs_uid_created ON agent_runs(uid, created_at DESC)", "CREATE INDEX IF NOT EXISTS idx_agent_runs_thread_created ON agent_runs(thread_id, created_at DESC)", "CREATE INDEX IF NOT EXISTS idx_agent_runs_status_updated ON agent_runs(status, updated_at)", diff --git a/backend/package/yuxi/storage/postgres/models_business.py b/backend/package/yuxi/storage/postgres/models_business.py index 1c9a2419..69e9cef8 100644 --- a/backend/package/yuxi/storage/postgres/models_business.py +++ b/backend/package/yuxi/storage/postgres/models_business.py @@ -300,6 +300,9 @@ class Message(Base): token_count = Column(Integer, nullable=True, comment="Token count (optional)") extra_metadata = Column(JSON, nullable=True, comment="Additional metadata (complete message dump)") image_content = Column(Text, nullable=True, comment="Base64 encoded image content for multimodal messages") + run_id = Column(String(64), ForeignKey("agent_runs.id"), nullable=True, index=True, comment="Agent run ID") + request_id = Column(String(64), nullable=True, index=True, comment="Request ID for idempotency") + delivery_status = Column(String(32), nullable=False, default="complete", comment="Message status") # Relationships conversation = relationship("Conversation", back_populates="messages") @@ -317,6 +320,9 @@ class Message(Base): "token_count": self.token_count, "metadata": self.extra_metadata or {}, "image_content": self.image_content, + "run_id": self.run_id, + "request_id": self.request_id, + "status": self.delivery_status, "tool_calls": [tc.to_dict() for tc in self.tool_calls] if self.tool_calls else [], } @@ -769,6 +775,16 @@ class AgentRun(Base): comment="Run status: pending/running/completed/failed/cancel_requested/cancelled/interrupted", ) request_id = Column(String(64), unique=True, index=True, nullable=False, comment="Idempotency request ID") + conversation_id = Column( + Integer, ForeignKey("conversations.id"), nullable=True, index=True, comment="Conversation ID" + ) + parent_run_id = Column(String(64), nullable=True, index=True, comment="Parent interrupted run ID") + run_type = Column(String(32), nullable=False, default="chat", comment="Run type: chat/resume") + resume_request_id = Column(String(64), nullable=True, index=True, comment="Resume idempotency request ID") + input_message_id = Column(Integer, nullable=True, comment="Input message ID") + output_message_id = Column(Integer, nullable=True, comment="Output message ID") + checkpoint_thread_id = Column(String(64), nullable=True, comment="LangGraph checkpoint thread ID") + last_event_id = Column(String(64), nullable=True, comment="Last Redis stream event ID") input_payload = Column(JSON, nullable=False, default=dict, comment="Original input payload") error_type = Column(String(64), nullable=True, comment="Error type") error_message = Column(Text, nullable=True, comment="Error message") @@ -785,6 +801,14 @@ class AgentRun(Base): "uid": self.uid, "status": self.status, "request_id": self.request_id, + "conversation_id": self.conversation_id, + "parent_run_id": self.parent_run_id, + "run_type": self.run_type, + "resume_request_id": self.resume_request_id, + "input_message_id": self.input_message_id, + "output_message_id": self.output_message_id, + "checkpoint_thread_id": self.checkpoint_thread_id, + "last_event_id": self.last_event_id, "input_payload": self.input_payload or {}, "error_type": self.error_type, "error_message": self.error_message, diff --git a/backend/server/routers/agent_router.py b/backend/server/routers/agent_router.py index 58d3fc40..46264162 100644 --- a/backend/server/routers/agent_router.py +++ b/backend/server/routers/agent_router.py @@ -1,6 +1,8 @@ from __future__ import annotations -from fastapi import APIRouter, Depends, HTTPException +from typing import Any + +from fastapi import APIRouter, Depends, Header, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession @@ -46,11 +48,14 @@ class AgentUpdate(BaseModel): class AgentRunCreate(BaseModel): - query: str = Field(..., description="用户输入的问题") + query: str | None = Field(None, description="用户输入的问题") agent_id: str = Field(..., description="智能体 ID") thread_id: str = Field(..., description="会话线程 ID") meta: dict = Field(default_factory=dict, description="可选，请求追踪信息，例如 request_id") image_content: str | None = Field(None, description="可选，base64 图片内容") + resume: Any | None = Field(None, description="可选，恢复 interrupted run 的输入") + parent_run_id: str | None = Field(None, description="可选，被恢复的 run ID") + resume_request_id: str | None = Field(None, description="可选，resume 幂等键") class AgentChatRequest(BaseModel): @@ -293,6 +298,9 @@ async def create_agent_run( image_content=payload.image_content, current_uid=str(current_user.uid), db=db, + resume=payload.resume, + parent_run_id=payload.parent_run_id, + resume_request_id=payload.resume_request_id, ) @@ -311,9 +319,15 @@ async def cancel_agent_run( @agent_router.get("/runs/{run_id}/events") -async def stream_run_events(run_id: str, after_seq: str = "0-0", current_user: User = Depends(get_required_user)): +async def stream_run_events( + run_id: str, + after_seq: str = "0-0", + last_event_id: str | None = Header(default=None, alias="Last-Event-ID"), + current_user: User = Depends(get_required_user), +): + cursor = last_event_id or after_seq return StreamingResponse( - stream_agent_run_events(run_id=run_id, after_seq=after_seq, current_uid=str(current_user.uid)), + stream_agent_run_events(run_id=run_id, after_seq=cursor, current_uid=str(current_user.uid)), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"}, ) diff --git a/backend/test/unit/services/test_agent_run_service.py b/backend/test/unit/services/test_agent_run_service.py index 7dce295e..e8c0a096 100644 --- a/backend/test/unit/services/test_agent_run_service.py +++ b/backend/test/unit/services/test_agent_run_service.py @@ -4,21 +4,12 @@ from contextlib import asynccontextmanager from types import SimpleNamespace import pytest -from sqlalchemy.exc import IntegrityError import yuxi.services.agent_run_service as agent_run_service -class FakeConfigRepo: - def __init__(self, db_session): - self.db = db_session - - async def get_by_id(self, config_id: int): - return SimpleNamespace(id=config_id, agent_id="ChatbotAgent", uid="user-1") - - @pytest.mark.asyncio -async def test_stream_agent_run_events_emits_error_and_close_on_db_error(monkeypatch: pytest.MonkeyPatch): +async def test_stream_agent_run_events_emits_error_on_db_error(monkeypatch: pytest.MonkeyPatch): @asynccontextmanager async def fake_session_ctx(): yield object() @@ -42,14 +33,13 @@ async def test_stream_agent_run_events_emits_error_and_close_on_db_error(monkeyp ): chunks.append(chunk) - assert len(chunks) == 2 + assert len(chunks) == 1 assert chunks[0].startswith("event: error") assert '"reason": "db_error"' in chunks[0] - assert chunks[1].startswith("event: close") @pytest.mark.asyncio -async def test_stream_agent_run_events_reads_redis_and_close_terminal(monkeypatch: pytest.MonkeyPatch): +async def test_stream_agent_run_events_reads_redis_and_ends_on_end_event(monkeypatch: pytest.MonkeyPatch): @asynccontextmanager async def fake_session_ctx(): yield object() @@ -60,7 +50,7 @@ async def test_stream_agent_run_events_reads_redis_and_close_terminal(monkeypatc async def get_run_for_user(self, run_id: str, uid: str): del run_id, uid - return SimpleNamespace(status="completed") + return SimpleNamespace(status="completed", thread_id="thread-1") calls = {"count": 0} @@ -71,21 +61,36 @@ async def test_stream_agent_run_events_reads_redis_and_close_terminal(monkeypatc return [ { "seq": "1700000000000-0", - "event_type": "loading", - "payload": {"items": [{"status": "loading", "response": "你"}]}, + "event_type": "messages", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "messages", + "payload": {"items": [{"status": "loading", "response": "你"}]}, + "created_at": "2026-05-27T00:00:00+00:00", + }, "ts": 1700000000000, - } + }, + { + "seq": "1700000000001-0", + "event_type": "end", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "end", + "payload": {"status": "completed"}, + "created_at": "2026-05-27T00:00:01+00:00", + }, + "ts": 1700000000001, + }, ] return [] - async def fake_last_seq(run_id: str): - del run_id - return "1700000000000-0" - monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx) monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo) monkeypatch.setattr(agent_run_service, "list_run_stream_events", fake_list_events) - monkeypatch.setattr(agent_run_service, "get_last_run_stream_seq", fake_last_seq) monkeypatch.setattr(agent_run_service, "SSE_POLL_INTERVAL_SECONDS", 0) chunks = [] @@ -96,17 +101,36 @@ async def test_stream_agent_run_events_reads_redis_and_close_terminal(monkeypatc ): chunks.append(chunk) - assert any(item.startswith("event: loading") for item in chunks) - assert chunks[-1].startswith("event: close") - assert '"last_seq": "1700000000000-0"' in chunks[-1] + assert chunks[0].startswith("event: messages") + assert "id: 1700000000000-0" in chunks[0] + assert chunks[-1].startswith("event: end") + assert "id: 1700000000001-0" in chunks[-1] @pytest.mark.asyncio -async def test_create_agent_run_commits_before_enqueue(monkeypatch: pytest.MonkeyPatch): +async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytest.MonkeyPatch): + class FakeResult: + def scalar_one_or_none(self): + return SimpleNamespace(uid="user-1", role="user") + class FakeDB: def __init__(self): self.order: list[str] = [] self.committed = False + self.added = [] + + async def execute(self, stmt): + del stmt + return FakeResult() + + def add(self, item): + self.added.append(item) + + async def flush(self): + self.order.append("flush") + for item in self.added: + if getattr(item, "id", None) is None: + item.id = 10 async def commit(self): self.order.append("commit") @@ -117,14 +141,14 @@ async def test_create_agent_run_commits_before_enqueue(monkeypatch: pytest.Monke db = FakeDB() created_run = SimpleNamespace( - id="run-1", + id="", thread_id="thread-1", status="pending", request_id="req-1", uid="user-1", ) - class Repo: + class RunRepo: def __init__(self, db_session): self.db = db_session @@ -134,6 +158,13 @@ async def test_create_agent_run_commits_before_enqueue(monkeypatch: pytest.Monke async def create_run(self, **kwargs): assert kwargs["request_id"] == "req-1" + assert kwargs["conversation_id"] == 1 + created_run.id = kwargs["run_id"] + return created_run + + async def set_input_message(self, run_id: str, message_id: int): + assert run_id == created_run.id + assert message_id == 10 return created_run class ConvRepo: @@ -142,33 +173,36 @@ async def test_create_agent_run_commits_before_enqueue(monkeypatch: pytest.Monke async def get_conversation_by_thread_id(self, thread_id: str): del thread_id - return SimpleNamespace( - uid="user-1", - status="active", - department_id=1, - extra_metadata={"agent_config_id": 1}, - ) + return SimpleNamespace(id=1, uid="user-1", status="active", agent_id="default") + + class AgentRepo: + def __init__(self, db_session): + self.db = db_session + + async def get_visible_by_slug(self, slug: str, user): + del user + return SimpleNamespace(slug=slug, backend_id="ChatbotAgent") class Queue: async def enqueue_job(self, job_name: str, run_id: str, _job_id: str): assert job_name == "process_agent_run" - assert run_id == "run-1" - assert _job_id == "run:run-1" + assert run_id == created_run.id + assert _job_id == f"run:{created_run.id}" db.order.append("enqueue") assert db.committed is True async def fake_get_arq_pool(): return Queue() - monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda agent_id: object()) - monkeypatch.setattr(agent_run_service, "AgentConfigRepository", FakeConfigRepo) + monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: object()) + monkeypatch.setattr(agent_run_service, "AgentRepository", AgentRepo) monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo) - monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo) + monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo) monkeypatch.setattr(agent_run_service, "get_arq_pool", fake_get_arq_pool) result = await agent_run_service.create_agent_run_view( query="hello", - agent_config_id=1, + agent_id="default", thread_id="thread-1", meta={"request_id": "req-1"}, image_content=None, @@ -176,55 +210,83 @@ async def test_create_agent_run_commits_before_enqueue(monkeypatch: pytest.Monke db=db, ) - assert db.order == ["commit", "enqueue"] - assert result["run_id"] == "run-1" + assert db.order[-2:] == ["commit", "enqueue"] + assert result["run_id"] == created_run.id assert result["request_id"] == "req-1" + assert db.added[0].run_id == created_run.id + assert db.added[0].request_id == "req-1" @pytest.mark.asyncio -async def test_create_agent_run_handles_integrity_error_with_same_user_existing(monkeypatch: pytest.MonkeyPatch): +async def test_create_resume_run_marks_input_message_source(monkeypatch: pytest.MonkeyPatch): + class FakeResult: + def scalar_one_or_none(self): + return SimpleNamespace(uid="user-1", role="user") + class FakeDB: def __init__(self): - self.commit_called = 0 - self.rollback_called = 0 + self.order: list[str] = [] + self.committed = False + self.added = [] + + async def execute(self, stmt): + del stmt + return FakeResult() + + def add(self, item): + self.added.append(item) + + async def flush(self): + self.order.append("flush") + for item in self.added: + if getattr(item, "id", None) is None: + item.id = 11 async def commit(self): - self.commit_called += 1 - raise IntegrityError("insert", {"request_id": "req-1"}, Exception("duplicate")) + self.order.append("commit") + self.committed = True async def rollback(self): - self.rollback_called += 1 + raise AssertionError("rollback should not be called") db = FakeDB() - existing_run = SimpleNamespace( - id="run-existing", + created_run = SimpleNamespace( + id="", thread_id="thread-1", - status="running", - request_id="req-1", + status="pending", + request_id="resume-req", uid="user-1", ) - state = {"lookup_count": 0} - class Repo: + class RunRepo: def __init__(self, db_session): self.db = db_session + async def get_run_for_user(self, run_id: str, uid: str): + assert run_id == "parent-run" + assert uid == "user-1" + return SimpleNamespace(id=run_id, thread_id="thread-1", status="interrupted") + + async def get_resume_run(self, parent_run_id: str, resume_request_id: str): + assert parent_run_id == "parent-run" + assert resume_request_id == "resume-req" + return None + async def get_run_by_request_id(self, request_id: str): - del request_id - state["lookup_count"] += 1 - if state["lookup_count"] == 1: - return None - return existing_run + assert request_id == "resume-req" + return None async def create_run(self, **kwargs): - del kwargs - return SimpleNamespace( - id="run-new", - thread_id="thread-1", - status="pending", - request_id="req-1", - uid="user-1", - ) + assert kwargs["run_type"] == "resume" + assert kwargs["parent_run_id"] == "parent-run" + assert kwargs["resume_request_id"] == "resume-req" + created_run.id = kwargs["run_id"] + return created_run + + async def set_input_message(self, run_id: str, message_id: int): + assert run_id == created_run.id + assert message_id == 11 + return created_run class ConvRepo: def __init__(self, db_session): @@ -232,106 +294,45 @@ async def test_create_agent_run_handles_integrity_error_with_same_user_existing( async def get_conversation_by_thread_id(self, thread_id: str): del thread_id - return SimpleNamespace( - uid="user-1", - status="active", - department_id=1, - extra_metadata={"agent_config_id": 1}, - ) + return SimpleNamespace(id=1, uid="user-1", status="active", agent_id="default") + + class AgentRepo: + def __init__(self, db_session): + self.db = db_session + + async def get_visible_by_slug(self, slug: str, user): + del user + return SimpleNamespace(slug=slug, backend_id="ChatbotAgent") + + class Queue: + async def enqueue_job(self, job_name: str, run_id: str, _job_id: str): + assert job_name == "process_agent_run" + assert run_id == created_run.id + assert _job_id == f"run:{created_run.id}" + assert db.committed is True async def fake_get_arq_pool(): - raise AssertionError("should not enqueue on integrity fallback") + return Queue() - monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda agent_id: object()) - monkeypatch.setattr(agent_run_service, "AgentConfigRepository", FakeConfigRepo) + monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda backend_id: object()) + monkeypatch.setattr(agent_run_service, "AgentRepository", AgentRepo) monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo) - monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo) + monkeypatch.setattr(agent_run_service, "AgentRunRepository", RunRepo) monkeypatch.setattr(agent_run_service, "get_arq_pool", fake_get_arq_pool) result = await agent_run_service.create_agent_run_view( - query="hello", - agent_config_id=1, + query=None, + agent_id="default", thread_id="thread-1", - meta={"request_id": "req-1"}, + meta={"request_id": "resume-req"}, image_content=None, current_uid="user-1", db=db, + resume={"language": "python"}, + parent_run_id="parent-run", + resume_request_id="resume-req", ) - assert db.commit_called == 1 - assert db.rollback_called == 1 - assert result["run_id"] == "run-existing" - assert result["status"] == "running" - - -@pytest.mark.asyncio -async def test_create_agent_run_integrity_error_returns_409_for_other_user(monkeypatch: pytest.MonkeyPatch): - class FakeDB: - async def commit(self): - raise IntegrityError("insert", {"request_id": "req-1"}, Exception("duplicate")) - - async def rollback(self): - return None - - db = FakeDB() - existing_run = SimpleNamespace( - id="run-existing", - thread_id="thread-1", - status="pending", - request_id="req-1", - uid="user-2", - ) - state = {"lookup_count": 0} - - class Repo: - def __init__(self, db_session): - self.db = db_session - - async def get_run_by_request_id(self, request_id: str): - del request_id - state["lookup_count"] += 1 - if state["lookup_count"] == 1: - return None - return existing_run - - async def create_run(self, **kwargs): - del kwargs - return SimpleNamespace( - id="run-new", - thread_id="thread-1", - status="pending", - request_id="req-1", - uid="user-1", - ) - - class ConvRepo: - def __init__(self, db_session): - self.db = db_session - - async def get_conversation_by_thread_id(self, thread_id: str): - del thread_id - return SimpleNamespace( - uid="user-1", - status="active", - department_id=1, - extra_metadata={"agent_config_id": 1}, - ) - - monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda agent_id: object()) - monkeypatch.setattr(agent_run_service, "AgentConfigRepository", FakeConfigRepo) - monkeypatch.setattr(agent_run_service, "ConversationRepository", ConvRepo) - monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo) - - with pytest.raises(agent_run_service.HTTPException) as exc: - await agent_run_service.create_agent_run_view( - query="hello", - agent_config_id=1, - thread_id="thread-1", - meta={"request_id": "req-1"}, - image_content=None, - current_uid="user-1", - db=db, - ) - - assert exc.value.status_code == 409 - assert exc.value.detail == "request_id 冲突" + assert result["run_id"] == created_run.id + assert db.added[0].message_type == "resume" + assert db.added[0].extra_metadata["source"] == "ask_user_question_resume" diff --git a/backend/test/unit/services/test_chat_stream_interrupt.py b/backend/test/unit/services/test_chat_stream_interrupt.py index bbff3df1..77ed9c3f 100644 --- a/backend/test/unit/services/test_chat_stream_interrupt.py +++ b/backend/test/unit/services/test_chat_stream_interrupt.py @@ -1,7 +1,11 @@ """测试 chat_service 中的 interrupt 相关函数""" +import json import sys import os +from types import SimpleNamespace + +import pytest sys.path.insert(0, os.getcwd()) @@ -9,6 +13,7 @@ from yuxi.services.chat_service import ( _normalize_interrupt_questions, _build_ask_user_question_payload, _coerce_interrupt_payload, + stream_agent_resume, ) from yuxi.utils.question_utils import normalize_options @@ -173,6 +178,24 @@ class TestNormalizeInterruptQuestions: assert result[0]["question"] == "有效问题" +@pytest.mark.asyncio +async def test_stream_agent_resume_init_does_not_render_resume_input(): + stream = stream_agent_resume( + thread_id="thread-1", + resume_input={"language": "python"}, + meta={"request_id": "req-1"}, + current_user=SimpleNamespace(uid="user-1"), + db=object(), + ) + + first_chunk = json.loads((await stream.__anext__()).decode("utf-8")) + await stream.aclose() + + assert first_chunk["status"] == "init" + assert "msg" not in first_chunk + assert "Resume with input" not in json.dumps(first_chunk, ensure_ascii=False) + + class TestCoerceInterruptPayload: """测试 _coerce_interrupt_payload 函数""" diff --git a/backend/test/unit/services/test_run_queue_service.py b/backend/test/unit/services/test_run_queue_service.py index f07642ad..ecccc417 100644 --- a/backend/test/unit/services/test_run_queue_service.py +++ b/backend/test/unit/services/test_run_queue_service.py @@ -103,7 +103,9 @@ async def test_run_stream_event_roundtrip(monkeypatch: pytest.MonkeyPatch): events = await run_queue_service.list_run_stream_events(run_id, after_seq="0-0", limit=100) assert [item["event_type"] for item in events] == ["loading", "finished"] - assert events[0]["payload"] == {"items": [1]} + assert events[0]["payload"]["schema_version"] == 1 + assert events[0]["payload"]["run_id"] == run_id + assert events[0]["payload"]["payload"] == {"items": [1]} next_events = await run_queue_service.list_run_stream_events(run_id, after_seq=seq1, limit=100) assert len(next_events) == 1 diff --git a/backend/test/unit/services/test_run_worker.py b/backend/test/unit/services/test_run_worker.py index 57beb783..c3e9000b 100644 --- a/backend/test/unit/services/test_run_worker.py +++ b/backend/test/unit/services/test_run_worker.py @@ -90,8 +90,8 @@ async def test_process_agent_run_non_retryable_error_marks_failed(monkeypatch: p terminal_statuses: list[str] = [] events: list[str] = [] - async def fake_append_event(run_id: str, event_type: str, payload: dict): - del run_id, payload + async def fake_append_event(run_id: str, event_type: str, payload: dict, **kwargs): + del run_id, payload, kwargs events.append(event_type) async def fake_mark_terminal(run_id: str, status: str, error_type=None, error_message=None): @@ -121,8 +121,8 @@ async def test_process_agent_run_retryable_error_retries_then_completes(monkeypa events: list[dict] = [] attempts = {"count": 0} - async def fake_append_event(run_id: str, event_type: str, payload: dict): - del run_id + async def fake_append_event(run_id: str, event_type: str, payload: dict, **kwargs): + del run_id, kwargs events.append({"event_type": event_type, "payload": payload}) async def fake_mark_terminal(run_id: str, status: str, error_type=None, error_message=None): @@ -169,10 +169,20 @@ async def test_worker_startup_ensures_builtin_mcp_servers(monkeypatch: pytest.Mo async def fake_ensure_builtin_mcp_servers_in_db(): calls.append("ensure_builtin_mcp_servers_in_db") + @asynccontextmanager + async def fake_session_ctx(): + yield object() + + async def fake_init_builtin_skills(session): + del session + calls.append("init_builtin_skills") + monkeypatch.setattr(run_worker.pg_manager, "initialize", fake_initialize) monkeypatch.setattr(run_worker.pg_manager, "create_business_tables", fake_create_business_tables) monkeypatch.setattr(run_worker.pg_manager, "ensure_business_schema", fake_ensure_business_schema) + monkeypatch.setattr(run_worker.pg_manager, "get_async_session_context", fake_session_ctx) monkeypatch.setattr(run_worker, "ensure_builtin_mcp_servers_in_db", fake_ensure_builtin_mcp_servers_in_db) + monkeypatch.setattr(run_worker, "init_builtin_skills", fake_init_builtin_skills) await run_worker._worker_startup({}) @@ -181,4 +191,5 @@ async def test_worker_startup_ensures_builtin_mcp_servers(monkeypatch: pytest.Mo "create_business_tables", "ensure_business_schema", "ensure_builtin_mcp_servers_in_db", + "init_builtin_skills", ] diff --git a/docs/develop-guides/roadmap.md b/docs/develop-guides/roadmap.md index 0fdf8a88..a1bf508f 100644 --- a/docs/develop-guides/roadmap.md +++ b/docs/develop-guides/roadmap.md @@ -66,6 +66,7 @@ - 收敛用户身份命名：原业务登录标识统一改为 `uid`，Agent/LangGraph runtime、conversation、agent_run、sandbox 路径和前端用户态均使用字符串 `uid`；`user_id` 仅保留给外部响应中的数值 `users.id` 或真实外键场景。 - 工作区知识库分类显示：知识库侧边栏按创建者分组为“我的知识库”和“共享知识库”，自己创建的知识库显示在“我的知识库”下，非自己创建的显示在“共享知识库”下；`knowledge_bases` 表新增 `created_by` 字段记录创建者 uid。 - 聊天附件新增 MinIO tmp 临时上传、可选 PDF/图片解析、确认后加入线程附件的流程；前端改为弹窗内上传、解析与确认。 +- 标准化 Agent run/SSE 执行链路：run 创建时持久化输入消息并提交后入队，worker 统一写入 Redis Stream envelope，SSE 输出 `event/data/id`、心跳注释、`Last-Event-ID` 回放和终止 `end` 事件；前端强制使用 run API 并支持 ask_user_question 中断后以 resume run 恢复。 --- diff --git a/notion_preresearch.md b/notion_preresearch.md deleted file mode 100644 index e1d2965e..00000000 --- a/notion_preresearch.md +++ /dev/null @@ -1,159 +0,0 @@ -# Notion Zotero Knowledge Base Reader - -This folder contains a small Python CLI for reading the shared Notion `Zotero` -knowledge base through the public Notion API. - -## Verified configuration - -- Connection token: provide through `NOTION_TOKEN`; do not hard-code it. -- Required capability: read content. -- Required sharing: the parent database must be shared with the connection. -- Notion API version used by the script: `2026-03-11`. -- Data source found: `Zotero` -- Data source id: `1a958b0f-ad1f-4f78-a082-cbdc66a3cd23` - -## Usage - -```bash -export NOTION_TOKEN="ntn_..." -``` - -Search shared pages/data sources by title: - -```bash -./notion_kb_reader.py search zotero --type data_source --limit 5 -./notion_kb_reader.py search "chain of thought" --type page --limit 10 -``` - -Query the Zotero data source and locally filter page properties: - -```bash -./notion_kb_reader.py query \ - --data-source-id 1a958b0f-ad1f-4f78-a082-cbdc66a3cd23 \ - "reasoning" \ - --limit 10 -``` - -Read a page into Markdown-like text: - -```bash -./notion_kb_reader.py read 12fee1d7-d69e-813b-81c8-f150b3d324af --output sample_page.md -``` - -Find text inside one page after fetching properties and block content: - -```bash -./notion_kb_reader.py find 12fee1d7-d69e-813b-81c8-f150b3d324af reasoning -``` - -Search or query, select the first matched page, then read it: - -```bash -./notion_kb_reader.py demo \ - --data-source-id 1a958b0f-ad1f-4f78-a082-cbdc66a3cd23 \ - "reasoning" \ - --output first_match.md -``` - -```python -#!/usr/bin/env python3 -"""Search and read a Notion knowledge base via the public API. - -Set NOTION_TOKEN before running: - export NOTION_TOKEN="ntn_..." -""" - -from __future__ import annotations - -import argparse -import json -import os -import sys -import textwrap -import time -import urllib.error -import urllib.parse -import urllib.request -from typing import Any - - -API_BASE = "https://api.notion.com/v1" -NOTION_VERSION = os.environ.get("NOTION_VERSION", "2026-03-11") - - -class NotionAPIError(RuntimeError): - pass - - -class NotionClient: - def __init__(self, token: str) -> None: - self.token = token - - def request( - self, - method: str, - path: str, - body: dict[str, Any] | None = None, - query: dict[str, Any] | None = None, - retries: int = 4, - ) -> dict[str, Any]: - url = f"{API_BASE}{path}" - if query: - url += "?" + urllib.parse.urlencode( - {k: v for k, v in query.items() if v is not None} - ) - data = json.dumps(body).encode("utf-8") if body is not None else None - headers = { - "Authorization": f"Bearer {self.token}", - "Notion-Version": NOTION_VERSION, - "Content-Type": "application/json", - "User-Agent": "notion-kb-reader/1.0", - } - - for attempt in range(retries + 1): - req = urllib.request.Request(url, data=data, headers=headers, method=method) - try: - with urllib.request.urlopen(req, timeout=45) as resp: - return json.loads(resp.read().decode("utf-8")) - except urllib.error.HTTPError as exc: - detail = exc.read().decode("utf-8", errors="replace") - retry_after = exc.headers.get("Retry-After") - if exc.code in {429, 500, 502, 503, 504} and attempt < retries: - sleep_s = float(retry_after or min(2**attempt, 8)) - time.sleep(sleep_s) - continue - raise NotionAPIError(f"HTTP {exc.code} {exc.reason}: {detail}") from exc - except urllib.error.URLError as exc: - if attempt < retries: - time.sleep(min(2**attempt, 8)) - continue - raise NotionAPIError(str(exc)) from exc - - raise NotionAPIError("request failed after retries") - - -def require_token() -> str: - token = os.environ.get("NOTION_TOKEN") or os.environ.get("NOTION_API_KEY") - if not token: - raise SystemExit( - "Missing token. Run: export NOTION_TOKEN='ntn_...'\n" - "Tip: do not hard-code the token in this script." - ) - return token - - -def paginate( - client: NotionClient, - method: str, - path: str, - body: dict[str, Any] | None = None, - query: dict[str, Any] | None = None, - limit: int | None = None, -) -> list[dict[str, Any]]: - results: list[dict[str, Any]] = [] - cursor: str | None = None - - while True: - request_body = dict(body or {}) - request_query = dict(query or {}) -``` \ No newline at end of file diff --git a/web/src/apis/agent_api.js b/web/src/apis/agent_api.js index f42bcfdc..1b732ec7 100644 --- a/web/src/apis/agent_api.js +++ b/web/src/apis/agent_api.js @@ -1,10 +1,4 @@ -import { - apiGet, - apiPost, - apiDelete, - apiPut, - apiRequest -} from './base' +import { apiGet, apiPost, apiDelete, apiPut, apiRequest } from './base' import { useUserStore } from '@/stores/user' /** @@ -18,30 +12,6 @@ import { useUserStore } from '@/stores/user' // ============================================================================= export const agentApi = { - /** - * 发送聊天消息到指定智能体（流式响应） - * @param {Object} data - 聊天数据 - * @returns {Promise} - 聊天响应流 - */ - sendAgentMessage: (data, options = {}) => { - const { signal, headers: extraHeaders, ...restOptions } = options || {} - const baseHeaders = { - 'Content-Type': 'application/json', - ...useUserStore().getAuthHeaders() - } - - return fetch('/api/agent/chat', { - method: 'POST', - body: JSON.stringify(data), - signal, - headers: { - ...baseHeaders, - ...(extraHeaders || {}) - }, - ...restOptions - }) - }, - /** * 简单聊天调用（非流式） * @param {string} query - 查询内容 @@ -111,39 +81,12 @@ export const agentApi = { */ getMessageFeedback: (messageId) => apiGet(`/api/chat/message/${messageId}/feedback`), - createAgent: (payload) => apiPost('/api/agent', payload), updateAgent: (agentId, payload) => apiPut(`/api/agent/${agentId}`, payload), deleteAgent: (agentId) => apiDelete(`/api/agent/${agentId}`), - /** - * 恢复被人工审批中断的对话（流式响应） - * @param {string} threadId - 会话 ID - * @param {Object} data - 恢复数据 { answer: { question_id: answer }, approved } - * @param {Object} options - 可选参数（signal, headers等） - * @returns {Promise} - 恢复响应流 - */ - resumeAgentChat: (threadId, data, options = {}) => { - const { signal, headers: extraHeaders, ...restOptions } = options || {} - const baseHeaders = { - 'Content-Type': 'application/json', - ...useUserStore().getAuthHeaders() - } - - return fetch(`/api/chat/thread/${threadId}/resume`, { - method: 'POST', - body: JSON.stringify(data), - signal, - headers: { - ...baseHeaders, - ...(extraHeaders || {}) - }, - ...restOptions - }) - }, - /** * 创建异步运行任务（Run） * @param {Object} data - run 请求体 @@ -155,7 +98,10 @@ export const agentApi = { agent_id: data.agent_id, thread_id: data.thread_id, meta: data.meta || {}, - image_content: data.image_content || null + image_content: data.image_content || null, + resume: data.resume ?? null, + parent_run_id: data.parent_run_id || null, + resume_request_id: data.resume_request_id || null }), /** @@ -188,16 +134,18 @@ export const agentApi = { */ streamAgentRunEvents: (runId, afterSeq = '0-0', options = {}) => { const { signal } = options - return fetch( - `/api/agent/runs/${runId}/events?after_seq=${encodeURIComponent(String(afterSeq))}`, - { - method: 'GET', - headers: { - ...useUserStore().getAuthHeaders() - }, - signal - } - ) + const headers = { + ...useUserStore().getAuthHeaders() + } + const cursor = String(afterSeq || '0-0') + if (cursor && cursor !== '0-0') { + headers['Last-Event-ID'] = cursor + } + return fetch(`/api/agent/runs/${runId}/events`, { + method: 'GET', + headers, + signal + }) } } diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue index 9ff9516f..d832b318 100644 --- a/web/src/components/AgentChatComponent.vue +++ b/web/src/components/AgentChatComponent.vue @@ -234,8 +234,6 @@ const { threads, currentThreadId, currentThread } = storeToRefs(chatThreadsStore const userInput = ref('') const sendCooldownActive = ref(false) let sendCooldownTimer = null -const useRunsApi = import.meta.env.VITE_USE_RUNS_API === 'true' - // 预设的打招呼文本 const greetingMessages = [ '👋 您好，有什么可以帮您？', @@ -1085,13 +1083,12 @@ const handleAttachmentRemove = async (attachment) => { } // ==================== 审批功能管理 ==================== -const { approvalState, handleApproval, processApprovalInStream } = useApproval({ +const { approvalState, processApprovalInStream } = useApproval({ getThreadState, - resetOnGoingConv, fetchThreadMessages }) -const { handleAgentResponse, handleStreamChunk } = useAgentStreamHandler({ +const { handleStreamChunk } = useAgentStreamHandler({ getThreadState, processApprovalInStream, currentAgentId, @@ -1100,10 +1097,8 @@ const { handleAgentResponse, handleStreamChunk } = useAgentStreamHandler({ }) const { startRunStream, resumeActiveRunForThread, stopRunStreamSubscription } = useAgentRunStream({ getThreadState, - useRunsApi, currentAgentId, handleStreamChunk, - processApprovalInStream, fetchThreadMessages, fetchAgentState, resetOnGoingConv, @@ -1111,45 +1106,6 @@ const { startRunStream, resumeActiveRunForThread, stopRunStreamSubscription } = streamSmoother }) -// 发送消息并处理流式响应 -const sendMessage = async ({ - agentId, - threadId, - text, - signal = undefined, - imageData = undefined, - requestId = '', - attachmentFileIds = [] -}) => { - if (!agentId || !threadId || !text) { - const error = new Error('Missing agent, thread, or message text') - handleChatError(error, 'send') - return Promise.reject(error) - } - - const requestData = { - query: text, - agent_id: agentId, - thread_id: threadId, - meta: { - request_id: requestId, - attachment_file_ids: attachmentFileIds - } - } - - // 如果有图片，添加到请求中 - if (imageData && imageData.imageContent) { - requestData.image_content = imageData.imageContent - } - - try { - return await agentApi.sendAgentMessage(requestData, signal ? { signal } : undefined) - } catch (error) { - handleChatError(error, 'send') - throw error - } -} - // ==================== CHAT ACTIONS ==================== // 获取第一个非置顶的对话 const getFirstNonPinnedChat = (chatList) => { @@ -1289,72 +1245,6 @@ const handleSendMessage = async ({ image } = {}) => { .map((attachment) => attachment.file_id) .filter(Boolean) - if (useRunsApi) { - if ((threadMessages.value[threadId] || []).length === 0) { - const autoTitle = text.replace(/\s+/g, ' ').trim().slice(0, 2000) - if (autoTitle) { - void (async () => { - try { - const generatedTitle = await agentApi.generateTitle( - autoTitle, - configStore.config?.fast_model - ) - if (generatedTitle) { - const finalTitle = generatedTitle.slice(0, 30).replace(/\s+/g, ' ').trim() - if (finalTitle) { - void chatThreadsStore.updateThread(threadId, finalTitle).catch(() => {}) - } - } - } catch (e) { - console.error('Title generation failed:', e) - // 失败时使用原始文本作为标题 - void chatThreadsStore.updateThread(threadId, autoTitle.slice(0, 30)).catch(() => {}) - } - })() - } - } - - resetOnGoingConv(threadId) - const requestId = createClientRequestId() - const previousAttachments = markAttachmentsRequestId(threadId, pendingAttachments, requestId) - insertOptimisticHumanMessage(threadState, { - requestId, - text, - imageContent, - attachments: pendingAttachments.map((attachment) => ({ - ...attachment, - request_id: requestId - })) - }) - threadState.isStreaming = true - try { - const runResp = await agentApi.createAgentRun({ - query: text, - agent_id: currentAgentId.value, - thread_id: threadId, - meta: { - request_id: requestId, - attachment_file_ids: pendingAttachmentFileIds - }, - image_content: imageContent - }) - const runId = runResp?.run_id - if (!runId) { - throw new Error('创建 run 失败：缺少 run_id') - } - await startRunStream(threadId, runId, 0) - } catch (error) { - threadState.isStreaming = false - threadState.replyLoadingVisible = false - threadState.pendingRequestId = null - rollbackAttachments(threadId, previousAttachments) - resetOnGoingConv(threadId) - handleChatError(error, 'send') - } - return - } - - // 如果是新对话，用 fast-model 异步生成标题（不阻塞消息发送） if ((threadMessages.value[threadId] || []).length === 0) { const autoTitle = text.replace(/\s+/g, ' ').trim().slice(0, 2000) if (autoTitle) { @@ -1372,14 +1262,12 @@ const handleSendMessage = async ({ image } = {}) => { } } catch (e) { console.error('Title generation failed:', e) - // 失败时使用原始文本作为标题 void chatThreadsStore.updateThread(threadId, autoTitle.slice(0, 30)).catch(() => {}) } })() } } - threadState.isStreaming = true resetOnGoingConv(threadId) const requestId = createClientRequestId() const previousAttachments = markAttachmentsRequestId(threadId, pendingAttachments, requestId) @@ -1387,40 +1275,36 @@ const handleSendMessage = async ({ image } = {}) => { requestId, text, imageContent, - attachments: pendingAttachments.map((attachment) => ({ ...attachment, request_id: requestId })) + attachments: pendingAttachments.map((attachment) => ({ + ...attachment, + request_id: requestId + })) }) - threadState.streamAbortController = new AbortController() + threadState.isStreaming = true try { - const response = await sendMessage({ - agentId: currentAgentId.value, - threadId: threadId, - text: text, - signal: threadState.streamAbortController?.signal, - imageData: image, - requestId, - attachmentFileIds: pendingAttachmentFileIds + const runResp = await agentApi.createAgentRun({ + query: text, + agent_id: currentAgentId.value, + thread_id: threadId, + meta: { + request_id: requestId, + attachment_file_ids: pendingAttachmentFileIds + }, + image_content: imageContent }) - - await handleAgentResponse(response, threadId) - } catch (error) { - if (error.name !== 'AbortError') { - console.error('Stream error:', error) - rollbackAttachments(threadId, previousAttachments) - handleChatError(error, 'send') - } else { - console.warn('[Interrupted] Catch') + const runId = runResp?.run_id + if (!runId) { + throw new Error('创建 run 失败：缺少 run_id') } + await startRunStream(threadId, runId, 0) + } catch (error) { threadState.isStreaming = false - } finally { - threadState.streamAbortController = null - // 异步加载历史记录，保持当前消息显示直到历史记录加载完成 - fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }).finally(() => { - // 历史记录加载完成后，安全地清空当前进行中的对话 - resetOnGoingConv(threadId) - handleAgentStateRefresh(threadId) - scrollController.scrollToBottom() - }) + threadState.replyLoadingVisible = false + threadState.pendingRequestId = null + rollbackAttachments(threadId, previousAttachments) + resetOnGoingConv(threadId) + handleChatError(error, 'send') } } @@ -1432,32 +1316,14 @@ const handleSendOrStop = async (payload) => { const threadId = currentChatId.value const threadState = getThreadState(threadId) - if (isProcessing.value && threadState) { - if (useRunsApi && threadState.activeRunId) { - try { - await agentApi.cancelAgentRun(threadState.activeRunId) - message.info('已发送取消请求') - } catch (error) { - handleChatError(error, 'stop') - } - return - } - - if (threadState.streamAbortController) { - // 中断生成 - threadState.streamAbortController.abort() - - // 中断后刷新消息历史，确保显示最新的状态 - try { - await fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId, delay: 500 }) - fetchAgentState(currentAgentId.value, threadId) - message.info('已中断对话生成') - } catch (error) { - console.error('刷新消息历史失败:', error) - message.info('已中断对话生成') - } - return + if (isProcessing.value && threadState?.activeRunId) { + try { + await agentApi.cancelAgentRun(threadState.activeRunId) + message.info('已发送取消请求') + } catch (error) { + handleChatError(error, 'stop') } + return } if (props.sendDisabled) return await handleSendMessage(payload) @@ -1479,30 +1345,35 @@ const handleApprovalWithStream = async (answer) => { return } + if (!approvalState.parentRunId) { + message.error('无法找到需要恢复的运行任务') + approvalState.showModal = false + return + } + try { - // 使用审批 composable 处理审批 - const response = await handleApproval(answer) - - if (!response) return // 如果 handleApproval 抛出错误，这里不会执行 - - // 处理流式响应 - await handleAgentResponse(response, threadId) - } catch (error) { - if (error.name !== 'AbortError') { - console.error('Resume approval error:', error) - } - } finally { - if (threadState) { - threadState.isStreaming = false - threadState.streamAbortController = null - } - - // 异步加载历史记录，保持当前消息显示直到历史记录加载完成 - fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }).finally(() => { - resetOnGoingConv(threadId) - fetchAgentState(currentAgentId.value, threadId) - scrollController.scrollToBottom() + approvalState.showModal = false + threadState.isStreaming = true + resetOnGoingConv(threadId) + const resumeRequestId = createClientRequestId() + const runResp = await agentApi.createAgentRun({ + query: null, + agent_id: currentAgentId.value, + thread_id: threadId, + meta: { request_id: resumeRequestId }, + resume: answer, + parent_run_id: approvalState.parentRunId, + resume_request_id: resumeRequestId }) + const runId = runResp?.run_id + if (!runId) { + throw new Error('创建 resume run 失败：缺少 run_id') + } + await startRunStream(threadId, runId, '0-0') + } catch (error) { + threadState.isStreaming = false + threadState.replyLoadingVisible = false + handleChatError(error, 'resume') } } diff --git a/web/src/components/AgentPanel.vue b/web/src/components/AgentPanel.vue index e4180914..c883c0fa 100644 --- a/web/src/components/AgentPanel.vue +++ b/web/src/components/AgentPanel.vue @@ -792,14 +792,14 @@ watch( align-items: center; border: 1px solid var(--gray-150); border-radius: 8px; - background: var(--gray-0); + background: var(--gray-25); color: var(--gray-700); overflow: hidden; flex-shrink: 0; &.active { - border-color: var(--main-200); - background: var(--main-20); + border-color: var(--main-600); + background: var(--gray-0); color: var(--main-800); } } diff --git a/web/src/components/ToolCallsGroupComponent.vue b/web/src/components/ToolCallsGroupComponent.vue index c0fc3cd0..e177e5a2 100644 --- a/web/src/components/ToolCallsGroupComponent.vue +++ b/web/src/components/ToolCallsGroupComponent.vue @@ -150,7 +150,7 @@ const toggleToolCallsExpanded = () => {