fix(hitl): stop dashboard ask cards from reappearing (#782)

Merged from #877.

CHANGELOG.md

@@ -12,6 +12,7 @@
 - 对话中可隐藏不常用的共享专家（浏览器本地偏好）(#589)
 
 ### 修复
+- Dashboard `ask_user_question` 卡片不再在刷新/切会话后重复弹出：HITL pending 在恢复开始即标记完成，新回合会作废遗留提问，历史重注入按 `pending_id` 幂等去重 (#782)
 - 远程 OCR 返回「未收到图片 / 请上传图片」等拒绝提示时不再当作正文入库（此前会被向量化、文档仍标记 ready），现按无可提取文本处理
 - 环境变量值含换行（如粘贴的多行密钥）时，保存后再读取会被截断并残留引号；现按引号跨行读回，保证「写入 → 读取」往返保真
 - 定时任务与系统任务（TLS 自动续期、自动备份）此前按宿主系统时区触发，现按 `default_timezone` 配置时区执行

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 {

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<ChatMessage> & Pick<ChatMessage, "id" | "role">,
+): 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);
+  });
+});

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(),

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:

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,

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:

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,

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

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

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()