fix: clean citation leaks and duplicate agent output in thinking process

src/App/src/hooks/usePlanWebSocket.tsx

@@ -29,7 +29,7 @@ import {
     addAgentMessage,
 } from '@/store/slices/chatSlice';
 import {
-    appendToStreamingBuffer,
+    setStreamingMessageBuffer,
     setShowBufferingText,
     addStreamingMessage,
     selectStreamingMessageBuffer,
@@ -158,7 +158,8 @@ export function usePlanWebSocket({
             if (chunks.length === 0) return;
             streamingChunkQueueRef.current = [];
             dispatch(setShowBufferingText(true));
-            dispatch(appendToStreamingBuffer(chunks.join('')));
+            // Backend sends full grouped snapshots (each agent once, latest round); replace with the newest.
+            dispatch(setStreamingMessageBuffer(chunks[chunks.length - 1]));
         };
 
         const unsub = webSocketService.on(
@@ -179,9 +180,10 @@ export function usePlanWebSocket({
                 streamingFlushHandleRef.current = null;
             }
             if (streamingChunkQueueRef.current.length > 0) {
-                const remaining = streamingChunkQueueRef.current.join('');
+                const remaining = streamingChunkQueueRef.current;
                 streamingChunkQueueRef.current = [];
-                dispatch(appendToStreamingBuffer(remaining));
+                // Snapshots are full buffers — keep only the latest, don't concatenate.
+                dispatch(setStreamingMessageBuffer(remaining[remaining.length - 1]));
             }
         };
     }, [dispatch]);

src/backend/callbacks/response_handlers.py

@@ -17,6 +17,54 @@
 
 logger = logging.getLogger(__name__)
 
+# Per-(user_id, agent_id) buffer holding a trailing, not-yet-closed citation marker until its closing "】" streams in a later chunk.
+_stream_citation_buffers: dict[tuple[str, str], str] = {}
+
+# Per user_id grouped "AI thinking" snapshot: ordered {agent_display_name: latest-round text}, keeping first-appearance order.
+_thinking_sections: dict[str, dict[str, str]] = {}
+# Per user_id currently-streaming agent display name (None = turn boundary; next delta starts fresh).
+_thinking_current: dict[str, str | None] = {}
+
+
+def reset_thinking_state(user_id: str) -> None:
+    """Clear grouped thinking-process snapshot + citation state for a user at the start of a new run."""
+    _thinking_sections.pop(user_id, None)
+    _thinking_current.pop(user_id, None)
+    for key in [k for k in _stream_citation_buffers if k[0] == user_id]:
+        _stream_citation_buffers.pop(key, None)
+
+
+def mark_streaming_turn_complete(user_id: str) -> None:
+    """Mark an agent turn boundary so the next streamed delta starts a fresh section (keeps only the latest turn, even for consecutive same-agent rounds)."""
+    _thinking_current[user_id] = None
+    # Drop any trailing partial-citation buffer so an unclosed "【" can't leak
+    # into the next turn (the callback is always called with is_final=False, so
+    # the final-chunk flush path never runs at a turn boundary).
+    for key in [k for k in _stream_citation_buffers if k[0] == user_id]:
+        _stream_citation_buffers.pop(key, None)
+
+
+def _build_thinking_snapshot(user_id: str) -> str:
+    """Render the grouped snapshot: each agent once (first-appearance order) with its latest-round text."""
+    sections = _thinking_sections.get(user_id)
+    if not sections:
+        return ""
+    parts = []
+    for name, text in sections.items():
+        body = (text or "").strip()
+        parts.append(f"---\n### {name}\n\n{body}" if body else f"---\n### {name}")
+    return "\n\n".join(parts)
+
+
+def _split_trailing_partial_citation(text: str) -> tuple[str, str]:
+    """Split off a trailing, not-yet-closed full-width citation marker (【...); returns (emittable, held_back)."""
+    if not text:
+        return text, ""
+    open_idx = text.rfind('【')
+    if open_idx != -1 and text.find('】', open_idx) == -1:
+        return text[:open_idx], text[open_idx:]
+    return text, ""
+
 
 def format_agent_display_name(raw_name: str) -> str:
     """Convert raw agent IDs (e.g. 'HRHelperAgent', 'hr_helper_agent') to
@@ -164,7 +212,19 @@ async def streaming_agent_response_callback(
                     collected.append(str(txt))
             chunk_text = "".join(collected) if collected else ""
 
-        cleaned = clean_citations(chunk_text or "")
+        # Prepend any buffered partial marker so markers split across chunks are reassembled before stripping.
+        buffer_key = (user_id, agent_id)
+        combined = _stream_citation_buffers.pop(buffer_key, "") + (chunk_text or "")
+
+        if is_final:
+            # Final chunk: strip complete markers, then drop any trailing unclosed "【..." tail.
+            cleaned = clean_citations(combined)
+            cleaned = re.sub(r'【[^】]*$', '', cleaned)
+        else:
+            emittable, held_back = _split_trailing_partial_citation(combined)
+            if held_back:
+                _stream_citation_buffers[buffer_key] = held_back
+            cleaned = clean_citations(emittable)
 
         contents = getattr(update, "contents", []) or []
         tool_calls = _extract_tool_calls_from_contents(contents)
@@ -178,17 +238,28 @@ async def streaming_agent_response_callback(
             )
             logger.info("Tool calls streamed from %s: %d", agent_id, len(tool_calls))
 
+        # Group by agent, keeping only each agent's latest round: on a switch to this agent
+        # (a re-invocation across Magentic rounds), reset its text but preserve first-appearance order.
+        sections = _thinking_sections.setdefault(user_id, {})
+        if _thinking_current.get(user_id) != display_name:
+            sections[display_name] = ""
+            _thinking_current[user_id] = display_name
         if cleaned:
+            sections[display_name] = sections.get(display_name, "") + cleaned
+
+        snapshot = _build_thinking_snapshot(user_id)
+        if snapshot:
+            # Backend emits the full grouped snapshot; the frontend replaces (not appends) its buffer.
             streaming_payload = AgentMessageStreaming(
                 agent_name=display_name,
-                content=cleaned,
+                content=snapshot,
                 is_final=is_final,
             )
             await connection_config.send_status_update_async(
                 streaming_payload,
                 user_id,
                 message_type=WebsocketMessageType.AGENT_MESSAGE_STREAMING,
             )
-            logger.debug("Streaming chunk (agent=%s final=%s len=%d)", agent_id, is_final, len(cleaned))
+            logger.debug("Streaming snapshot (agent=%s final=%s len=%d)", agent_id, is_final, len(snapshot))
     except Exception as e:
         logger.error("streaming_agent_response_callback error: %s", e)

src/backend/orchestration/orchestration_manager.py

@@ -16,14 +16,16 @@
                                             MagenticPlanReviewRequest)
 from agents.agent_factory import AgentFactory
 from callbacks.response_handlers import (agent_response_callback,
-                                         format_agent_display_name,
+                                         clean_citations,
+                                         mark_streaming_turn_complete,
+                                         reset_thinking_state,
                                          streaming_agent_response_callback)
 from common.config.app_config import config
 from common.database.database_base import DatabaseBase
 from common.models.messages import TeamConfiguration
 from common.utils.markdown_utils import \
     normalize_markdown_tables as _normalize_markdown_tables
-from models.messages import AgentMessageStreaming, WebsocketMessageType
+from models.messages import WebsocketMessageType
 from orchestration.connection_config import (connection_config,
                                              orchestration_config)
 from orchestration.plan_review_helpers import (convert_plan_review_to_mplan,
@@ -367,7 +369,9 @@ async def run_orchestration(self, user_id: str, input_task) -> None:
         try:
             final_output_ref: list = [None]
             orchestrator_chunks: list[str] = []
-            current_streaming_agent_ref: list = [None]
+
+            # Reset grouped thinking-process snapshot state for this user before a fresh run.
+            reset_thinking_state(user_id)
 
             # Collect participant names for plan conversion
             participant_names = [
@@ -385,7 +389,6 @@ async def run_orchestration(self, user_id: str, input_task) -> None:
                 user_id=user_id,
                 final_output_ref=final_output_ref,
                 orchestrator_chunks=orchestrator_chunks,
-                current_streaming_agent_ref=current_streaming_agent_ref,
             )
 
             # Resume loop — handle plan reviews and tool approvals until workflow completes
@@ -445,13 +448,17 @@ async def run_orchestration(self, user_id: str, input_task) -> None:
                     user_id=user_id,
                     final_output_ref=final_output_ref,
                     orchestrator_chunks=orchestrator_chunks,
-                    current_streaming_agent_ref=current_streaming_agent_ref,
                 )
 
             # Use executor_completed Message if available; otherwise fall back to
             # accumulated orchestrator streaming chunks.
             final_text = final_output_ref[0] or "".join(orchestrator_chunks)
 
+            # Strip citation markers (e.g. 【5:0†source】) leaked by the manager.
+            # The streaming agent callback cleans work-agent output, but the
+            # Group Chat Manager's own final text bypasses that path.
+            final_text = clean_citations(final_text)
+
             # Repair collapsed markdown tables before rendering (Bug 47810).
             final_text = _normalize_markdown_tables(final_text)
 
@@ -521,6 +528,10 @@ async def run_orchestration(self, user_id: str, input_task) -> None:
             raise
 
         finally:
+            # Free this user's grouped-thinking + citation buffer state so it
+            # doesn't linger in module-level dicts after the run (incl. aborted
+            # / single-run users).
+            reset_thinking_state(user_id)
             # Clean up MCP connections to avoid noisy cross-task
             # RuntimeError from anyio when async generators are GC'd.
             await self._cleanup_workflow_mcp(user_id)
@@ -704,7 +715,6 @@ async def _process_event_stream(
         user_id: str,
         final_output_ref: list,
         orchestrator_chunks: list[str],
-        current_streaming_agent_ref: list,
     ) -> dict | None:
         """Process a workflow event stream, collecting pending requests.
 
@@ -720,6 +730,7 @@ async def _process_event_stream(
         """
         plan_requests: dict[str, MagenticPlanReviewRequest] = {}
         tool_approvals: dict[str, object] = {}  # request_id -> event.data (Content)
+        round_no = 0  # incremented each progress-ledger round (agent-selection turn)
 
         async for event in stream:
             try:
@@ -777,9 +788,27 @@ async def _process_event_stream(
                 # Magentic orchestrator events (plan created, replanned, progress ledger)
                 elif event.type == "magentic_orchestrator":
                     orch_event: MagenticOrchestratorEvent = event.data
-                    self.logger.info(
-                        "[ORCHESTRATOR:%s]", orch_event.event_type.value
-                    )
+                    ledger = getattr(orch_event, "content", None)
+                    next_speaker = getattr(ledger, "next_speaker", None)
+                    if next_speaker is not None:
+                        # One line per round: which agent the manager selected + why
+                        # it hasn't stopped yet (satisfied/loop/progress decision flags).
+                        round_no += 1
+                        satisfied = getattr(getattr(ledger, "is_request_satisfied", None), "answer", "?")
+                        in_loop = getattr(getattr(ledger, "is_in_loop", None), "answer", "?")
+                        progress = getattr(getattr(ledger, "is_progress_being_made", None), "answer", "?")
+                        self.logger.info(
+                            "[ROUND %d] next_speaker=%s satisfied=%s in_loop=%s "
+                            "progress=%s | reason=%s",
+                            round_no,
+                            getattr(next_speaker, "answer", "?"),
+                            satisfied, in_loop, progress,
+                            getattr(next_speaker, "reason", ""),
+                        )
+                    else:
+                        self.logger.info(
+                            "[ORCHESTRATOR:%s]", orch_event.event_type.value
+                        )
 
                 # Streaming output
                 elif event.type == "output":
@@ -790,30 +819,9 @@ async def _process_event_stream(
                         if executor == "magentic_orchestrator" and output_data.text:
                             orchestrator_chunks.append(output_data.text)
 
-                        if (
-                            executor != "magentic_orchestrator"
-                            and executor != current_streaming_agent_ref[0]
-                        ):
-                            current_streaming_agent_ref[0] = executor
-                            display_name = format_agent_display_name(executor)
-                            header_text = f"\n\n---\n### {display_name}\n\n"
-                            try:
-                                await connection_config.send_status_update_async(
-                                    AgentMessageStreaming(
-                                        agent_name=display_name,
-                                        content=header_text,
-                                        is_final=False,
-                                    ),
-                                    user_id,
-                                    message_type=WebsocketMessageType.AGENT_MESSAGE_STREAMING,
-                                )
-                            except Exception as cb_err:
-                                self.logger.error(
-                                    "Error sending agent header for %s: %s",
-                                    executor, cb_err,
-                                )
-
                         if executor != "magentic_orchestrator":
+                            # The streaming callback groups by agent and emits the full snapshot;
+                            # headers are rendered inside the snapshot (no separate header send).
                             try:
                                 await streaming_agent_response_callback(
                                     executor, output_data, False, user_id,
@@ -836,6 +844,8 @@ async def _process_event_stream(
                             if isinstance(msg, Message) and msg.text:
                                 final_output_ref[0] = msg.text
                     else:
+                        # Turn boundary: next streamed delta (even same agent, new round) starts fresh.
+                        mark_streaming_turn_complete(user_id)
                         for msg in event.data:
                             if isinstance(msg, Message) and msg.text:
                                 try: