From 06f632ec5987ffb152c6f91700a7fd7ee20e81f6 Mon Sep 17 00:00:00 2001 From: jubaoliang Date: Mon, 21 Sep 2026 03:02:52 +0000 Subject: [PATCH] fix(hitl): stop dashboard ask cards from reappearing (#782) Merged from #877. --- CHANGELOG.md | 1 + dashboard/src/pages/Chat/hooks/sseHelpers.ts | 1 + .../utils/injectPendingHitlMessage.test.ts | 58 ++++++++++++++++ .../src/utils/injectPendingHitlMessage.ts | 18 +++-- src/octop/api/routers/chat/routes.py | 12 ++-- src/octop/infra/gateway/hitl/coordinator.py | 14 ++++ src/octop/infra/gateway/hitl/store.py | 18 +++++ src/octop/infra/gateway/process/processor.py | 2 + tests/unit/api/test_chat_hitl_resume.py | 43 ++++++++++++ tests/unit/gateway/test_dashboard_ws.py | 66 +++++++++++++++++++ tests/unit/gateway/test_hitl_channel.py | 18 +++++ 11 files changed, 242 insertions(+), 9 deletions(-) create mode 100644 dashboard/src/utils/injectPendingHitlMessage.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index caf84341..5d5022da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ - 对话中可隐藏不常用的共享专家（浏览器本地偏好）(#589) ### 修复 +- Dashboard `ask_user_question` 卡片不再在刷新/切会话后重复弹出：HITL pending 在恢复开始即标记完成，新回合会作废遗留提问，历史重注入按 `pending_id` 幂等去重 (#782) - 远程 OCR 返回「未收到图片 / 请上传图片」等拒绝提示时不再当作正文入库（此前会被向量化、文档仍标记 ready），现按无可提取文本处理 - 环境变量值含换行（如粘贴的多行密钥）时，保存后再读取会被截断并残留引号；现按引号跨行读回，保证「写入 → 读取」往返保真 - 定时任务与系统任务（TLS 自动续期、自动备份）此前按宿主系统时区触发，现按 `default_timezone` 配置时区执行 diff --git a/dashboard/src/pages/Chat/hooks/sseHelpers.ts b/dashboard/src/pages/Chat/hooks/sseHelpers.ts index 9f5a842f..06303411 100644 --- a/dashboard/src/pages/Chat/hooks/sseHelpers.ts +++ b/dashboard/src/pages/Chat/hooks/sseHelpers.ts @@ -34,6 +34,7 @@ export interface HitlRequestData { action_requests: HitlActionRequest[]; review_configs?: Array<{ action_name: string; allowed_decisions: string[] }>; status?: "pending" | "approved" | "rejected"; + pending_id?: string; } export interface ChatAttachment { diff --git a/dashboard/src/utils/injectPendingHitlMessage.test.ts b/dashboard/src/utils/injectPendingHitlMessage.test.ts new file mode 100644 index 00000000..89be9fcb --- /dev/null +++ b/dashboard/src/utils/injectPendingHitlMessage.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, it } from "vitest"; +import type { ChatMessage } from "../pages/Chat/hooks/useChat"; +import { injectPendingHitlMessage } from "./injectPendingHitlMessage"; + +function msg( + partial: Partial & Pick, +): ChatMessage { + return { + content: "", + status: "done", + timestamp: 0, + ...partial, + }; +} + +const pending = { + pending_id: "ab12", + action_requests: [{ name: "ask_user_question", args: { questions: [] } }], +}; + +describe("injectPendingHitlMessage", () => { + it("appends one pending card when history has none", () => { + const out = injectPendingHitlMessage([], pending); + expect(out).toHaveLength(1); + expect(out[0].id).toBe("hitl-ab12"); + expect(out[0].hitlData?.status).toBe("pending"); + expect(out[0].hitlData?.pending_id).toBe("ab12"); + }); + + it("does not append a second card after the first was answered", () => { + const answered = [ + msg({ + id: "hitl-ab12", + role: "assistant", + hitlData: { + action_requests: pending.action_requests, + status: "approved", + pending_id: "ab12", + }, + }), + ]; + expect(injectPendingHitlMessage(answered, pending)).toEqual(answered); + }); + + it("does not append when a pending card is already in the list", () => { + const existing = [ + msg({ + id: "other", + role: "assistant", + hitlData: { + action_requests: pending.action_requests, + status: "pending", + }, + }), + ]; + expect(injectPendingHitlMessage(existing, pending)).toEqual(existing); + }); +}); diff --git a/dashboard/src/utils/injectPendingHitlMessage.ts b/dashboard/src/utils/injectPendingHitlMessage.ts index 72b0be03..c30025e8 100644 --- a/dashboard/src/utils/injectPendingHitlMessage.ts +++ b/dashboard/src/utils/injectPendingHitlMessage.ts @@ -6,19 +6,29 @@ export function injectPendingHitlMessage( pending: HitlPendingPayload | null | undefined, ): ChatMessage[] { if (!pending?.action_requests?.length) return messages; - if (messages.some((m) => m.hitlData?.status === "pending")) return messages; + const pendingId = pending.pending_id?.trim() || ""; + const injectedId = pendingId ? `hitl-${pendingId}` : ""; + if ( + messages.some( + (m) => + m.hitlData?.status === "pending" || + (pendingId && + (m.id === injectedId || m.hitlData?.pending_id === pendingId)), + ) + ) { + return messages; + } return [ ...messages, { - id: pending.pending_id - ? `hitl-${pending.pending_id}` - : `hitl-${Date.now()}`, + id: injectedId || `hitl-${Date.now()}`, role: "assistant", content: "", hitlData: { action_requests: pending.action_requests, review_configs: pending.review_configs, status: "pending", + ...(pendingId ? { pending_id: pendingId } : {}), }, status: "done", timestamp: Date.now(), diff --git a/src/octop/api/routers/chat/routes.py b/src/octop/api/routers/chat/routes.py index d4713e00..34fcba2c 100644 --- a/src/octop/api/routers/chat/routes.py +++ b/src/octop/api/routers/chat/routes.py @@ -131,6 +131,13 @@ async def iter_dashboard_hitl_resume_sse( channel_type=channel_type, ) disconnected = False + # Clear the in-memory pause before the (possibly long) resume stream so + # history reload cannot reinject the same card while the turn continues. + if pending is not None: + hitl_coordinator.store.mark_resolved( + pending.pending_id, + "rejected" if rejected else "approved", + ) try: async for chunk in processor.iter_hitl_resume_chunks( agent_id=agent_id, @@ -146,11 +153,6 @@ async def iter_dashboard_hitl_resume_sse( hitl_coordinator.register_from_request(request_payload, ctx=hitl_ctx) if not disconnected: yield format_sse("chunk", chunk) - if pending is not None: - hitl_coordinator.store.mark_resolved( - pending.pending_id, - "rejected" if rejected else "approved", - ) if not disconnected: yield format_sse("chunk", {"type": "done"}) except Exception as exc: diff --git a/src/octop/infra/gateway/hitl/coordinator.py b/src/octop/infra/gateway/hitl/coordinator.py index ec351f0f..240c748f 100644 --- a/src/octop/infra/gateway/hitl/coordinator.py +++ b/src/octop/infra/gateway/hitl/coordinator.py @@ -146,6 +146,20 @@ class HitlChannelCoordinator: reject_message = message or "Rejected by user" return [{"type": "reject", "message": reject_message} for _ in range(count)] + def expire_pending_for_thread( + self, + thread_id: str, + *, + agent_id: str, + user_id: int | None = None, + ) -> None: + """Expire in-memory pauses for a dashboard thread (new turn / interrupt).""" + self._store.expire_pending_for_thread( + thread_id, + agent_id=agent_id, + user_id=user_id, + ) + def resolve_ask_pending( self, session_key: str, diff --git a/src/octop/infra/gateway/hitl/store.py b/src/octop/infra/gateway/hitl/store.py index ed65fc8f..f9da2345 100644 --- a/src/octop/infra/gateway/hitl/store.py +++ b/src/octop/infra/gateway/hitl/store.py @@ -162,6 +162,24 @@ class HitlPendingStore: rows.sort(key=lambda r: r.created_at, reverse=True) return rows + def expire_pending_for_thread( + self, + thread_id: str, + *, + agent_id: str, + user_id: int | None = None, + ) -> None: + """Drop leftover dashboard pauses so history reload cannot reinject them.""" + self._gc() + for record in self._records.values(): + if record.thread_id != thread_id or record.status != "pending": + continue + if record.agent_id != agent_id: + continue + if user_id is not None and record.user_id != user_id: + continue + record.status = "expired" + def mark_resolved(self, pending_id: str, status: Literal["approved", "rejected"]) -> None: record = self._records.get(pending_id) if record is not None: diff --git a/src/octop/infra/gateway/process/processor.py b/src/octop/infra/gateway/process/processor.py index 2774e634..ddbb97b1 100644 --- a/src/octop/infra/gateway/process/processor.py +++ b/src/octop/infra/gateway/process/processor.py @@ -915,6 +915,7 @@ class GlobalProcessor: command=msg.text, response_lines=slash_lines, ) + self._hitl.expire_pending_for_thread(thread_id, agent_id=agent_id, user_id=user_id) self._touch_thread_after_turn(thread_id, msg.text) for line in slash_lines: yield _maybe_stamp_team_host( @@ -948,6 +949,7 @@ class GlobalProcessor: channel_metadata=im_meta, ) + self._hitl.expire_pending_for_thread(thread_id, agent_id=agent_id, user_id=user_id) request = await self._build_dashboard_request( msg, agent_id=agent_id, diff --git a/tests/unit/api/test_chat_hitl_resume.py b/tests/unit/api/test_chat_hitl_resume.py index 8101d787..0d4e102e 100644 --- a/tests/unit/api/test_chat_hitl_resume.py +++ b/tests/unit/api/test_chat_hitl_resume.py @@ -158,3 +158,46 @@ async def test_dashboard_hitl_resume_finishes_after_client_disconnect() -> None: assert completed is True assert frames == [] + + +@pytest.mark.asyncio +async def test_dashboard_hitl_resume_marks_pending_resolved_when_stream_errors() -> None: + async def _resume(*_args: object, **_kwargs: object): + raise RuntimeError("stream died") + yield {} # pragma: no cover + + processor = MagicMock() + processor.iter_hitl_resume_chunks = _resume + hitl = HitlChannelCoordinator() + first = hitl.store.register( + thread_id="thr-err", + agent_id="agent-1", + user_id=1, + session_key="sk-err", + channel_type="dashboard", + action_requests=[{"name": "ask_user_question", "args": {"questions": []}}], + review_configs=None, + ) + + frames: list[str] = [] + async for frame in iter_dashboard_hitl_resume_sse( + processor=processor, + hitl_coordinator=hitl, + agent_id="agent-1", + thread_id="thr-err", + user_id=1, + decisions=[{"type": "respond", "message": "ok"}], + pending=first, + session_key="sk-err", + channel_type="dashboard", + locale="zh", + is_disconnected=AsyncMock(return_value=False), + ): + frames.append(frame) + + chunks = _parse_sse_chunks("".join(frames)) + assert any(c.get("type") == "error" for c in chunks) + resolved = hitl.store.get(first.pending_id) + assert resolved is not None + assert resolved.status == "approved" + assert hitl.store.resolve_pending_for_thread("thr-err", agent_id="agent-1", user_id=1) is None diff --git a/tests/unit/gateway/test_dashboard_ws.py b/tests/unit/gateway/test_dashboard_ws.py index 7ef80b7d..e40f776c 100644 --- a/tests/unit/gateway/test_dashboard_ws.py +++ b/tests/unit/gateway/test_dashboard_ws.py @@ -584,6 +584,72 @@ async def test_global_processor_iter_turn_chunks_registers_hitl() -> None: assert pending.action_requests[0]["name"] == "execute" +@pytest.mark.asyncio +async def test_global_processor_iter_turn_chunks_expires_stale_hitl() -> None: + from unittest.mock import AsyncMock, MagicMock + + from octop.infra.gateway.hitl.coordinator import HitlChannelCoordinator + from octop.infra.gateway.process.processor import GlobalProcessor + from octop.infra.gateway.slash.dispatcher import SlashDispatcher + + async def _stream(*_args: object, **_kwargs: object): + yield {"type": "token", "content": "ok"} + + agent_manager = MagicMock() + agent_manager.stream = _stream + agent_manager.merge_turn_mcp_servers = MagicMock(return_value=None) + agent_manager.prepare_chat_mcp = AsyncMock(return_value=[]) + + thread_registry = MagicMock() + thread_registry.get_or_create_by_key = AsyncMock(return_value="thread-hitl") + + hitl = HitlChannelCoordinator() + stale = hitl.store.register( + thread_id="thread-hitl", + agent_id="agent-1", + user_id=1, + session_key="sk", + channel_type="dashboard", + action_requests=[{"name": "ask_user_question", "args": {"questions": []}}], + review_configs=None, + ) + processor = GlobalProcessor( + agent_manager=agent_manager, + thread_registry=thread_registry, + audit_repo=MagicMock(), + agent_repo=MagicMock(), + user_repo=MagicMock(), + connector_repo=MagicMock(), + dispatcher=SlashDispatcher(), + usage_repo=None, + gateway=None, + hitl=hitl, + ) + + msg = InboundMessage( + channel_id=WS_CHANNEL_ID, + channel_type="dashboard", + tenant_id="agent-1", + channel_subject=ChannelSubject(subject_id="1"), + content=[TextContent(text="never mind, do this instead")], + metadata={"session_key": "sk", "thread_id": "thread-hitl"}, + ) + + chunks = [c async for c in processor.iter_turn_chunks(msg)] + assert any(c.get("type") == "token" for c in chunks) + assert ( + hitl.store.resolve_pending_for_thread( + "thread-hitl", + agent_id="agent-1", + user_id=1, + ) + is None + ) + leftover = hitl.store.get(stale.pending_id) + assert leftover is not None + assert leftover.status == "expired" + + @pytest.mark.asyncio async def test_global_processor_iter_turn_chunks_slash(tmp_path: Path) -> None: from unittest.mock import AsyncMock, MagicMock diff --git a/tests/unit/gateway/test_hitl_channel.py b/tests/unit/gateway/test_hitl_channel.py index f7b71c47..c441a37b 100644 --- a/tests/unit/gateway/test_hitl_channel.py +++ b/tests/unit/gateway/test_hitl_channel.py @@ -356,6 +356,24 @@ def test_hitl_store_get_pending_agent_mismatch() -> None: assert store.get_pending(record.pending_id, session_key="sk1", agent_id="other") is None +def test_expire_pending_for_thread_drops_history_reinject() -> None: + store = HitlPendingStore() + record = store.register( + thread_id="thr-ask", + agent_id="agent1", + user_id=7, + session_key="sk-ask", + channel_type="dashboard", + action_requests=[{"name": "ask_user_question", "args": {"questions": []}}], + review_configs=None, + ) + store.expire_pending_for_thread("thr-ask", agent_id="agent1", user_id=7) + assert store.resolve_pending_for_thread("thr-ask", agent_id="agent1", user_id=7) is None + expired = store.get(record.pending_id) + assert expired is not None + assert expired.status == "expired" + + @pytest.mark.asyncio async def test_slash_outcome_completed_turn() -> None: store = HitlPendingStore()