diff --git a/src/App/src/hooks/usePlanWebSocket.tsx b/src/App/src/hooks/usePlanWebSocket.tsx index c7256fc13..d09bdb5a6 100644 --- a/src/App/src/hooks/usePlanWebSocket.tsx +++ b/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]); diff --git a/src/backend/callbacks/response_handlers.py b/src/backend/callbacks/response_handlers.py index c2bf654ff..cfb3ee4c9 100644 --- a/src/backend/callbacks/response_handlers.py +++ b/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,10 +238,21 @@ 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( @@ -189,6 +260,6 @@ async def streaming_agent_response_callback( 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) diff --git a/src/backend/orchestration/orchestration_manager.py b/src/backend/orchestration/orchestration_manager.py index df43f8d47..ec0831814 100644 --- a/src/backend/orchestration/orchestration_manager.py +++ b/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: