[C7] UI streaming renderer (typed sequential ghost text; no乱序)

Author: inliver233

issues/2026-02-10_18-23-04-ui-autosave-streaming-v1.csv

@@ -5,4 +5,4 @@ C3,Bind EditorPanel documents to ProjectManager (autosave works),Add EditorPanel
 C4,Restore last session (reopen recent project & last document),"Persist and restore last-open project/document when enabled: wire SettingsDialog restore_session/auto_save prefs into Config; on startup, if enabled, auto-open most recent project and re-open last document (fallback to first scene).",Restart restores the previous project and last document content instead of showing a blank scratch doc (when restore_session enabled).,manual,none,DONE,DONE,TODO,src/main.py | src/core/config_schema.py | src/gui/dialogs/settings_dialog.py | src/core/config.py | src/gui/main_window_parts/ui.py | src/gui/main_window_parts/integrations.py | src/gui/main_window_parts/dialogs.py | src/gui/editor/text_editor.py,C3,"manual_checklist: 1) Create/open a project, edit a scene, wait for autosave, close app. 2) Reopen app; expect project auto-open and edited content present. 3) Disable restore_session in settings; restart; expect no auto-open. | validation_limited:manual app restart not executed here | evidence:python -m compileall -q src PASS | risk:medium session-restore depends on runtime restart flow | done_at:2026-02-10"
 C5,Streaming dispatch respects stream_response toggle,Plumb config ai.stream_response (default on) through AICompletionService/AIRequestDispatcher to use QtAIClient.complete_stream_async; include context/request_id in chunk signals; ensure cancellation prevents stale chunks from updating UI; add unit test for dispatch selection.,"When stream_response enabled, streaming client method is used and chunk signals are emitted; when disabled, non-stream method used; unit test passes.",python -m pytest -q tests/test_streaming_dispatcher.py,none,DONE,DONE,TODO,src/application/ai_completion_service.py | src/core/ai_qt_client.py | src/gui/ai/enhanced_ai_manager.py | src/gui/main_window_parts/integrations.py | tests/test_streaming_dispatcher.py | src/core/config_schema.py,none,test:pytest -q tests/test_streaming_dispatcher.py PASS | done_at:2026-02-10
 C6,Provider streaming adaptation (Gemini endpoint + chunk parsing),Implement provider-specific streaming endpoint/headers/parsing: Gemini uses streamGenerateContent + SSE; add Gemini extract_stream_content; improve streaming parser robustness; ensure OpenAI/Claude continue working; add unit tests with fixture chunks.,Gemini streaming endpoint/path is correct and chunk parsing yields incremental text; unit tests pass.,python -m pytest -q tests/test_provider_streaming_gemini.py,none,DONE,DONE,TODO,src/core/ai_client.py | src/core/ai_providers/gemini.py | src/core/ai_providers/openai.py | src/core/ai_providers/claude.py | tests/test_provider_streaming_gemini.py,C5,test:pytest -q tests/test_provider_streaming_gemini.py PASS | done_at:2026-02-10
-C7,UI streaming renderer (typed sequential ghost text; no乱序),"Connect streaming chunk updates to SmartCompletionManager and render as ordered, typing-like ghost text; buffer and flush to UI to avoid乱序/抖动; ignore chunks from cancelled/old requests; finalize cleanly on completion; add unit/Qt test.","During streaming completion, suggestion grows in order like typing; cancel stops updates; no stale chunks applied; test passes.",python -m pytest -q tests/test_streaming_typed_renderer.py,none,TODO,TODO,TODO,src/gui/editor/smart_completion_manager.py | src/gui/main_window_parts/integrations.py | src/gui/ai/enhanced_ai_manager.py | tests/test_streaming_typed_renderer.py,C5|C6,none
+C7,UI streaming renderer (typed sequential ghost text; no乱序),"Connect streaming chunk updates to SmartCompletionManager and render as ordered, typing-like ghost text; buffer and flush to UI to avoid乱序/抖动; ignore chunks from cancelled/old requests; finalize cleanly on completion; add unit/Qt test.","During streaming completion, suggestion grows in order like typing; cancel stops updates; no stale chunks applied; test passes.",python -m pytest -q tests/test_streaming_typed_renderer.py,none,DONE,DONE,TODO,src/gui/editor/smart_completion_manager.py | src/gui/main_window_parts/integrations.py | src/gui/ai/enhanced_ai_manager.py | tests/test_streaming_typed_renderer.py,C5|C6,test:pytest -q tests/test_streaming_typed_renderer.py PASS | done_at:2026-02-10

src/gui/ai/enhanced_ai_manager.py

@@ -54,7 +54,7 @@ class EnhancedAIManager(QObject):
     completionReady = pyqtSignal(str, str)      # (completion_text, context)
     completionReceived = pyqtSignal(str, dict)  # (response, metadata)
     completionError = pyqtSignal(str)           # (error_message)
-    streamUpdate = pyqtSignal(str)              # (partial_text)
+    streamUpdate = pyqtSignal(str, dict)        # (partial_text, context)
     configChanged = pyqtSignal()                # 配置更改信号
 
     def __init__(self, config: Config, shared=None, parent: QWidget = None):
@@ -64,6 +64,7 @@ def __init__(self, config: Config, shared=None, parent: QWidget = None):
         self._parent = parent
         self._task_manager = getattr(shared, "task_manager", None) if shared else None
         self._cancelled_task_keys: set[str] = set()
+        self._active_request_ids: dict[str, str] = {}
         self._ai_service = AICompletionService(
             config,
             shared,
@@ -156,6 +157,10 @@ def _init_ai_client(self):
                 # 连接信号
                 self._ai_client.responseReceived.connect(self._on_completion_ready)
                 self._ai_client.errorOccurred.connect(self._on_completion_error)
+                if hasattr(self._ai_client, "requestStarted"):
+                    self._ai_client.requestStarted.connect(self._on_request_started)
+                if hasattr(self._ai_client, "requestCompleted"):
+                    self._ai_client.requestCompleted.connect(self._on_request_completed)
                 if hasattr(self._ai_client, 'streamChunkReceived'):
                     self._ai_client.streamChunkReceived.connect(self._on_stream_update)
 
@@ -433,6 +438,23 @@ def _on_completion_error(self, error: str, context: dict):
             }
             self._task_manager.fail_external(task_key, error, details)
         self.completionError.emit(error)
+
+    @pyqtSlot(dict)
+    def _on_request_started(self, context: dict) -> None:
+        task_key = context.get("task_key") if isinstance(context, dict) else None
+        request_id = context.get("request_id") if isinstance(context, dict) else None
+        if task_key and request_id:
+            self._active_request_ids[str(task_key)] = str(request_id)
+
+    @pyqtSlot(dict)
+    def _on_request_completed(self, context: dict) -> None:
+        task_key = context.get("task_key") if isinstance(context, dict) else None
+        request_id = context.get("request_id") if isinstance(context, dict) else None
+        if not task_key:
+            return
+        active = self._active_request_ids.get(str(task_key))
+        if (not request_id) or (active == str(request_id)):
+            self._active_request_ids.pop(str(task_key), None)
 
     @pyqtSlot(str, dict)
     def _on_stream_update(self, partial_text: str, context: dict):
@@ -443,7 +465,21 @@ def _on_stream_update(self, partial_text: str, context: dict):
         task_key = context.get("task_key")
         if task_key and task_key in self._cancelled_task_keys:
             return
-        self.streamUpdate.emit(partial_text)
+        request_id = context.get("request_id")
+        if task_key and request_id:
+            active = self._active_request_ids.get(str(task_key))
+            if active and active != str(request_id):
+                return
+
+        try:
+            editor = getattr(self, "_current_editor", None)
+            smart_completion = getattr(editor, "_smart_completion", None) if editor else None
+            if smart_completion and hasattr(smart_completion, "update_streaming_ai_completion"):
+                smart_completion.update_streaming_ai_completion(partial_text, context)
+        except Exception:
+            logger.debug("Failed to forward stream update to SmartCompletionManager", exc_info=True)
+
+        self.streamUpdate.emit(partial_text, context)
 
     # 兼容性方法 - 保持与SimpleAIManager的接口兼容
     def set_completion_enabled(self, enabled: bool):
@@ -947,7 +983,7 @@ def start_stream_response(self, text: str):
         try:
             logger.debug(f"开始流式响应: {text[:50]}...")
             # 发送流式更新信号
-            self.streamUpdate.emit(text)
+            self.streamUpdate.emit(text, {})
         except Exception as e:
             logger.error(f"流式响应失败: {e}")
 

src/gui/editor/smart_completion_manager.py

@@ -6,7 +6,7 @@
 import logging
 import re
 import time
-from typing import Optional, List, Dict, Any, Tuple
+from typing import Optional, List, Dict, Any, Tuple, Callable
 from PyQt6.QtWidgets import QWidget, QLabel
 from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QObject
 from PyQt6.QtGui import QFont, QTextCursor, QKeyEvent
@@ -35,6 +35,92 @@
 DEFAULT_AI_RENDERER_ORDER = ("ghost_text", "inline", "direct_insert")
 
 
+class TypedStreamingGhostTextController(QObject):
+    """Buffer streaming chunks and render as typing-like ghost text.
+
+    - Appends incoming chunks in-order.
+    - Flushes to UI at a steady cadence to avoid jitter.
+    - Ignores chunks from stale request_id/task_key.
+    """
+
+    def __init__(
+        self,
+        *,
+        render: Callable[[str], None],
+        hide: Callable[[], None],
+        flush_interval_ms: int = 15,
+        max_chars_per_tick: int = 3,
+        parent: QObject | None = None,
+    ) -> None:
+        super().__init__(parent)
+        self._render = render
+        self._hide = hide
+        self._flush_interval_ms = max(1, int(flush_interval_ms))
+        self._max_chars_per_tick = max(1, int(max_chars_per_tick))
+
+        self._active_request_id: str | None = None
+        self._active_task_key: str | None = None
+        self._pending: str = ""
+        self._displayed: str = ""
+
+        self._timer = QTimer(self)
+        self._timer.setInterval(self._flush_interval_ms)
+        self._timer.timeout.connect(self._on_tick)
+
+    def cancel(self) -> None:
+        self._timer.stop()
+        self._pending = ""
+        self._displayed = ""
+        self._active_request_id = None
+        self._active_task_key = None
+        try:
+            self._hide()
+        except Exception:
+            return
+
+    def on_chunk(self, chunk_text: str, context: dict) -> None:
+        chunk = str(chunk_text or "")
+        if not chunk:
+            return
+
+        request_id = context.get("request_id") if isinstance(context, dict) else None
+        task_key = context.get("task_key") if isinstance(context, dict) else None
+
+        if self._active_task_key and task_key and task_key != self._active_task_key:
+            return
+
+        if self._active_request_id and request_id and request_id != self._active_request_id:
+            return
+
+        if self._active_request_id is None and request_id:
+            self._active_request_id = str(request_id)
+        if self._active_task_key is None and task_key:
+            self._active_task_key = str(task_key)
+
+        self._pending += chunk
+        if not self._timer.isActive():
+            self._timer.start()
+
+    def _on_tick(self) -> None:
+        if not self._pending:
+            self._timer.stop()
+            return
+
+        take = self._pending[: self._max_chars_per_tick]
+        self._pending = self._pending[self._max_chars_per_tick :]
+        self._displayed += take
+
+        try:
+            self._render(self._displayed)
+        except Exception:
+            # Rendering should never break the stream loop.
+            self.cancel()
+            return
+
+        if not self._pending:
+            self._timer.stop()
+
+
 class SmartCompletionManager(QObject):
     """智能补全管理器 - 统一所有补全功能"""
 
@@ -91,6 +177,14 @@ def __init__(self, text_editor, completion_engine: CompletionEngine):
         # 纯状态机（domain/application 层）用于描述 completion 生命周期
         self._completion_state_machine = CompletionStateMachine()
 
+        self._streaming_typed_controller = TypedStreamingGhostTextController(
+            render=self._render_streaming_ghost_text,
+            hide=self._hide_streaming_ghost_text,
+            flush_interval_ms=15,
+            max_chars_per_tick=3,
+            parent=self,
+        )
+
         # 连接 Ghost Text 接受/拒绝信号到状态机，确保 Esc/Tab/输入 等路径一致收口
         self._bind_ghost_signals()
         # 移除FloatingStatusIndicator以避免状态指示器冲突
@@ -205,6 +299,7 @@ def cancel_current_completion(self):
             self._ai_timeout_timer.stop()
         if hasattr(self, "_auto_completion_timer") and self._auto_completion_timer.isActive():
             self._auto_completion_timer.stop()
+        self.stop_streaming_ai_completion()
         self.hide_all_completions()
         self._is_completing = False
         self._last_completion_pos = -1
@@ -463,6 +558,9 @@ def _ai_complete(self, text: str, position: int, trigger_type: str = 'auto'):
             self._is_completing = False
             return
 
+        # Reset any in-flight streaming render before starting a new request.
+        self.stop_streaming_ai_completion()
+
         try:
             self._completion_state_machine.reset()
             self._completion_state_machine.set_state(CompletionState.REQUESTING)
@@ -615,9 +713,41 @@ def _show_popup_completion(self, suggestions: List[CompletionSuggestion]):
                 self.suggestionRendered.emit(suggestions[0].text)
             except Exception:
                 self.suggestionRendered.emit("")
+
+    def update_streaming_ai_completion(self, chunk_text: str, context: dict) -> None:
+        """Render streaming chunks as typing-like Ghost Text."""
+        if not chunk_text:
+            return
+        if not self._ghost_completion:
+            self._redetect_ghost_text_system()
+        if not self._ghost_completion:
+            return
+        try:
+            self._streaming_typed_controller.on_chunk(chunk_text, context or {})
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("Streaming typed render failed: %s", exc)
+
+    def stop_streaming_ai_completion(self) -> None:
+        try:
+            self._streaming_typed_controller.cancel()
+        except Exception:
+            return
+
+    def _render_streaming_ghost_text(self, suggestion: str) -> None:
+        if not suggestion:
+            return
+        if not self._ghost_completion:
+            self._redetect_ghost_text_system()
+        renderer = GhostTextRenderer(self._ghost_completion, self._ghost_state_manager)
+        renderer.render(suggestion)
+
+    def _hide_streaming_ghost_text(self) -> None:
+        renderer = GhostTextRenderer(self._ghost_completion, self._ghost_state_manager)
+        renderer.hide()
 
     def show_ai_completion(self, suggestion: str):
         """显示AI补全建议 - 增强版本，支持多种显示模式和状态同步"""
+        self.stop_streaming_ai_completion()
         # 🔧 修复：标记请求已完成，防止超时处理冲突
         self._ai_request_completed = True
 

tests/test_streaming_typed_renderer.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import os
+
+import pytest
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+
+pytest.importorskip("PyQt6")
+pytest.importorskip("pytestqt")
+
+from gui.editor.smart_completion_manager import TypedStreamingGhostTextController
+
+
+def test_typed_streaming_renderer_types_in_order(qtbot) -> None:
+    rendered: list[str] = []
+
+    def _render(text: str) -> None:
+        rendered.append(text)
+
+    def _hide() -> None:
+        return None
+
+    controller = TypedStreamingGhostTextController(
+        render=_render,
+        hide=_hide,
+        flush_interval_ms=1,
+        max_chars_per_tick=1,
+    )
+
+    controller.on_chunk("abc", {"request_id": "r1", "task_key": "t1"})
+
+    qtbot.waitUntil(lambda: bool(rendered) and rendered[-1] == "abc", timeout=1000)
+    assert rendered == ["a", "ab", "abc"]
+
+
+def test_typed_streaming_renderer_cancel_stops_updates(qtbot) -> None:
+    rendered: list[str] = []
+
+    def _render(text: str) -> None:
+        rendered.append(text)
+
+    def _hide() -> None:
+        rendered.append("<hide>")
+
+    controller = TypedStreamingGhostTextController(
+        render=_render,
+        hide=_hide,
+        flush_interval_ms=10,
+        max_chars_per_tick=1,
+    )
+
+    controller.on_chunk("ab", {"request_id": "r1"})
+    qtbot.waitUntil(lambda: bool(rendered) and rendered[-1] == "a", timeout=1000)
+
+    controller.cancel()
+    qtbot.wait(40)
+
+    assert rendered.count("ab") == 0
+
+
+def test_typed_streaming_renderer_ignores_stale_request_id(qtbot) -> None:
+    rendered: list[str] = []
+
+    def _render(text: str) -> None:
+        rendered.append(text)
+
+    def _hide() -> None:
+        return None
+
+    controller = TypedStreamingGhostTextController(
+        render=_render,
+        hide=_hide,
+        flush_interval_ms=1,
+        max_chars_per_tick=10,
+    )
+
+    controller.on_chunk("X", {"request_id": "r2"})
+    qtbot.waitUntil(lambda: bool(rendered) and rendered[-1] == "X", timeout=1000)
+
+    controller.on_chunk("Y", {"request_id": "r1"})
+    qtbot.wait(20)
+
+    assert rendered[-1] == "X"