暂停继续

offline-backend-20260512/backend/app/gateway/routers/ai_writing.py

@@ -130,6 +130,12 @@ _session_error_flags: dict[str, bool] = {}
 # session_id → valid_article_types（list[dict] with key/label）
 _session_article_types: dict[str, list[dict]] = {}
 
+# session_id → pending await_user event tuple for SSE reconnect re-delivery
+# asyncio.Queue is consume-once; if the old SSE generator dequeued the await_user event
+# before the client disconnected, the event is lost from the queue. We save a copy here
+# so that a new SSE connection (e.g., after user clicks "继续") can re-inject it.
+_session_pending_pause: dict[str, tuple] = {}
+
 
 # ── 工具函数 ──────────────────────────────────────────────────────────────────
 
@@ -403,6 +409,9 @@ async def resume_writing(session_id: str, body: ResumeRequest, request: Request)
             except Exception:
                 logger.exception("ai_writing: failed to save intervention: session=%s", session_id)
 
+    # Clear pending pause so SSE reconnect won't re-inject the stale await_user event
+    _session_pending_pause.pop(session_id, None)
+
     asyncio.create_task(
         _stream_graph_segment(session_id, Command(resume=intervention), config, AI_WRITING_GRAPH)
     )
@@ -450,7 +459,11 @@ async def _stream_graph_segment(
                 interrupt_list = node_output if isinstance(node_output, (list, tuple)) else [node_output]
                 for intr in interrupt_list:
                     interrupt_data = intr.value if hasattr(intr, "value") else intr
-                    await queue.put(("await_user", interrupt_data))
+                    event_tuple = ("await_user", interrupt_data)
+                    await queue.put(event_tuple)
+                    # Save for SSE reconnect re-delivery: if the old generator consumed this
+                    # event before the client disconnected, a new connection would block forever.
+                    _session_pending_pause[session_id] = event_tuple
                 return  # 等待 resume，不关闭队列
 
             if not isinstance(node_output, dict):
@@ -552,6 +565,7 @@ async def _stream_graph_segment(
             _session_interventions.pop(session_id, None)
             _session_error_flags.pop(session_id, None)
             _session_article_types.pop(session_id, None)
+            _session_pending_pause.pop(session_id, None)
 
 
 def _event_type(evt: dict) -> str:
@@ -577,6 +591,12 @@ async def _sse_generator(session_id: str) -> AsyncGenerator[str, None]:
     queue = _event_queues[session_id]
     yield _sse("connected", {"session_id": session_id})
 
+    # Re-inject the saved await_user event if the queue is empty and a pending pause exists.
+    # This covers the case where the old SSE generator dequeued the await_user event and
+    # yielded it to a disconnecting client — the event is then lost from the consume-once queue.
+    if queue.empty() and session_id in _session_pending_pause:
+        await queue.put(_session_pending_pause[session_id])
+
     while True:
         item = await queue.get()
         if item is None: