""" Stream chat service implemented with Agno SDK (Agent + tools + DB). """ from __future__ import annotations import ast import asyncio import json import os import re import time from collections.abc import AsyncGenerator from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from agno.agent import Agent, RunEvent from agno.models.message import Message from agno.run.agent import RunOutput, ToolCallCompletedEvent, ToolCallStartedEvent from agno.run.team import TeamRunEvent from agno.utils.log import logger from ..models.stream_chat import ( AgentStatusEvent, DoneEvent, ErrorEvent, FormRequestEvent, # New: HITL form request event SourceEvent, StreamChatRequest, TextEvent, ThoughtEvent, ToolCallEvent, ToolResultEvent, ) from .agent_registry import get_agent_for_provider, build_team, resolve_agent_config from .hitl_storage import get_hitl_storage from .summary_service import update_session_summary from .tool_registry import resolve_tool_name MEMORY_OPTIMIZE_THRESHOLD = 50 MEMORY_OPTIMIZE_INTERVAL_SECONDS = 60 * 60 * 12 THINK_TAG_REGEX = re.compile(r"", re.IGNORECASE) PROTOCOL_TAG_REGEX = re.compile( r"(?:<\s*[|｜]\s*(?P[a-zA-Z0-9_]+)\s*[|｜]\s*>)" r"|(?:<\s*(?P/?)\s*[|｜]\s*DSML\s*[|｜]\s*(?P[^>]+)>)", re.IGNORECASE, ) TOOL_TRACE_BEGIN_TAGS = { "tool_calls_begin", "tool_calls_section_begin", "tool_call_begin", "tool_argument_begin", "tool_call_argument_begin", } TOOL_TRACE_END_TAGS = { "tool_argument_end", "tool_call_argument_end", "tool_call_end", "tool_calls_end", "tool_calls_section_end", } def _strip_internal_tool_trace(text: str) -> str: """Remove explicit protocol marker tokens without truncating normal text.""" if not text: return "" cleaned = str(text) cleaned = re.sub(r"", "", cleaned, flags=re.IGNORECASE) # Remove protocol markers like <|tool_calls_begin|> and spaced variants. cleaned = re.sub(r"<\s*[|｜]\s*[^|>|｜]*\s*[|｜]\s*>", "", cleaned, flags=re.IGNORECASE) cleaned = re.sub(r"]*>", "", cleaned, flags=re.IGNORECASE) cleaned = re.sub(r"", "", cleaned, flags=re.IGNORECASE) cleaned = re.sub(r"\[SYSTEM INJECTED CONTEXT\]", "", cleaned, flags=re.IGNORECASE) # Remove inline tool-call-like snippets leaked by some models. cleaned = re.sub( r"([:：]\s*)?[a-zA-Z_][a-zA-Z0-9_]{1,80}\s*\{[^{}\n]{0,1200}\}", "", cleaned, flags=re.IGNORECASE, ) return cleaned def _split_content_by_think_tags(text: str, in_think: bool) -> tuple[list[tuple[str, str]], bool]: """Split a content chunk into ordered thought/text segments by / tags.""" if not text: return [], in_think segments: list[tuple[str, str]] = [] cursor = 0 current_in_think = in_think for match in THINK_TAG_REGEX.finditer(text): start, end = match.span() if start > cursor: piece = text[cursor:start] if piece: segments.append(("thought" if current_in_think else "text", piece)) tag = match.group(0).lower() current_in_think = not tag.startswith(" tuple[str, int, str, bool]: """ Strip inline tool-protocol payloads from content chunks safely across chunk boundaries. Returns: cleaned_text, next_tool_trace_depth, next_protocol_tail, had_protocol_tokens """ combined = f"{protocol_tail}{text or ''}" if not combined: return "", tool_trace_depth, "", False # Keep trailing incomplete protocol marker for next chunk. tail = "" tail_start = -1 for m in re.finditer(r"<\s*/?\s*[|｜]", combined): tail_start = m.start() if tail_start != -1 and ">" not in combined[tail_start:]: tail = combined[tail_start:] combined = combined[:tail_start] if not combined: return "", tool_trace_depth, tail, bool(tail) # Non-destructive stripping: remove marker tokens only. matches = list(PROTOCOL_TAG_REGEX.finditer(combined)) had_protocol = bool(matches) if not had_protocol: return combined, 0, tail, bool(tail) cleaned = PROTOCOL_TAG_REGEX.sub("", combined) return cleaned, 0, tail, True def _squash_whitespace(text: Any) -> str: return re.sub(r"\s+", "", str(text or "")) def _extract_agent_info_from_event( run_event: Any, leader_id: str | None = None, leader_name: str | None = None, leader_emoji: str | None = None, agent_metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: """ Extract agent identification from a run event. For Team mode, member events have agent_id and agent_name directly on the event. If these are set, it's a member event; otherwise it's from the leader. Returns: dict with 'agent_id', 'agent_name', 'agent_role', 'agent_emoji', 'model', 'provider' keys """ agent_id = getattr(run_event, "agent_id", None) agent_name = getattr(run_event, "agent_name", None) agent_emoji = getattr(run_event, "agent_emoji", None) # Check if this matches the leader. # Important: some providers might use slightly different names, but if IDs match it's definitely leader. is_leader = False if leader_id and agent_id == leader_id: is_leader = True elif not agent_id and not agent_name: # Default to leader if no info is present is_leader = True elif not leader_id and agent_name == leader_name: is_leader = True # If the ID starts with 'qurio-' (default Agno IDs often follow this pattern) # and we are in team mode, and it's not explicitly a member ID in our metadata, # it's highly likely the leader's initialization event. elif str(agent_id or "").startswith("qurio-") and agent_metadata: is_leader = agent_id not in agent_metadata if is_leader: res = { "agent_id": leader_id, "agent_name": leader_name, "agent_role": "leader", "agent_emoji": leader_emoji, "model": None, "provider": None, } if agent_metadata and leader_id in agent_metadata: res.update(agent_metadata[leader_id]) return res # Otherwise treat as member if agent_id or agent_name: # Log at DEBUG level to reduce main stream noise, as switch logs will provide context. logger.debug(f"[TEAM] Member event detected: agent_id={agent_id}, agent_name={agent_name}") res = { "agent_id": agent_id, "agent_name": agent_name, "agent_role": "member", "agent_emoji": agent_emoji, "model": None, "provider": None, } # Enrich from metadata if possible if agent_metadata: if agent_id in agent_metadata: res.update(agent_metadata[agent_id]) elif agent_name in agent_metadata: res.update(agent_metadata[agent_name]) elif not agent_id and agent_name: # Fallback lookup by name if ID missing on event for meta_id, meta in agent_metadata.items(): if meta.get("name") == agent_name: res.update(meta) res["agent_id"] = meta_id break return res # Fallback to leader if totally ambiguous res = { "agent_id": leader_id, "agent_name": leader_name, "agent_role": "leader", "agent_emoji": leader_emoji, "model": None, "provider": None, } if agent_metadata and leader_id in agent_metadata: res.update(agent_metadata[leader_id]) return res def _is_reasoning_duplicate_of_content(reasoning: str, content: str) -> bool: """ Detect provider chunks where answer text is mirrored in reasoning_content. This prevents answer paragraphs from being rendered as a second thought block. """ reasoning_flat = _squash_whitespace(reasoning) content_flat = _squash_whitespace(content) if not reasoning_flat or not content_flat: return False if reasoning_flat == content_flat: return True shorter, longer = ( (reasoning_flat, content_flat) if len(reasoning_flat) <= len(content_flat) else (content_flat, reasoning_flat) ) if len(shorter) < 12: return False if shorter in longer and len(shorter) >= int(len(longer) * 0.75): return True return False def _is_stream_trace_enabled() -> bool: value = str(os.getenv("QURIO_STREAM_TRACE", "")).strip().lower() return value in {"1", "true", "yes", "on", "debug"} def _is_verbose_logs_enabled() -> bool: value = str(os.getenv("QURIO_VERBOSE_LOGS", "0")).strip().lower() return value in {"1", "true", "yes", "on", "debug"} def _log_verbose_info(message: str) -> None: if _is_verbose_logs_enabled(): logger.info(message) else: logger.debug(message) def _preview(text: Any, limit: int = 140) -> str: raw = str(text or "").replace("\n", "\\n") return raw[:limit] + ("..." if len(raw) > limit else "") def _extract_message_from_payload(payload: Any) -> str | None: if payload is None: return None if hasattr(payload, "model_dump"): try: payload = payload.model_dump() except Exception: payload = str(payload) if isinstance(payload, dict): for key in ("message", "error", "detail", "msg"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() nested = _extract_message_from_payload(value) if nested: return nested for value in payload.values(): nested = _extract_message_from_payload(value) if nested: return nested return None if isinstance(payload, (list, tuple)): for item in payload: nested = _extract_message_from_payload(item) if nested: return nested return None text = str(payload).strip() if not text: return None # Agno RunErrorEvent repr: RunErrorEvent(..., content='Unknown model error', ...) run_error_content_match = re.search( r"""content\s*=\s*(['"])(.*?)\1""", text, re.IGNORECASE | re.DOTALL, ) if run_error_content_match and run_error_content_match.group(2).strip(): return run_error_content_match.group(2).strip() for parser in (json.loads, ast.literal_eval): try: parsed = parser(text) nested = _extract_message_from_payload(parsed) if nested: return nested except Exception: pass json_like = re.search(r"(\{[\s\S]*\})", text) if json_like: snippet = json_like.group(1) for parser in (json.loads, ast.literal_eval): try: parsed = parser(snippet) nested = _extract_message_from_payload(parsed) if nested: return nested except Exception: pass message_match = re.search(r"""['"]message['"]\s*:\s*['"](.+?)['"]""", text, re.IGNORECASE) if message_match and message_match.group(1).strip(): return message_match.group(1).strip() return text def _extract_best_error_message(exc: Exception | Any) -> str: """Extract the most actionable provider message from nested exceptions.""" generic_markers = ("unknown model error", "unknown error", "model provider error") def _is_generic(text: str) -> bool: lowered = text.strip().lower() return any(marker in lowered for marker in generic_markers) candidates: list[str] = [] queue: list[Any] = [exc] seen: set[int] = set() while queue: current = queue.pop(0) if current is None: continue marker = id(current) if marker in seen: continue seen.add(marker) extracted = _extract_message_from_payload(current) if extracted and extracted.strip(): candidates.append(extracted.strip()) if isinstance(current, BaseException): queue.append(getattr(current, "__cause__", None)) queue.append(getattr(current, "__context__", None)) args = getattr(current, "args", None) if isinstance(args, tuple): queue.extend(args) for attr in ("content", "error", "message", "detail", "model_provider_data"): if hasattr(current, attr): queue.append(getattr(current, attr, None)) for msg in candidates: if not _is_generic(msg): return msg if candidates: # Avoid dumping full event repr like "RunErrorEvent(...)" to UI. filtered = [msg for msg in candidates if not msg.strip().lower().startswith("runerrorevent(")] if filtered: return min(filtered, key=len) return min(candidates, key=len) return str(exc or "Unknown error") def _extract_text_chunk(run_event: Any) -> str: """Extract assistant text only from explicit content fields. Shared between stream_chat() and _continue_hitl_run() to avoid duplication. """ provider_data = getattr(run_event, "model_provider_data", None) if isinstance(provider_data, dict): choices = provider_data.get("choices") or [] if choices and isinstance(choices[0], dict): delta = choices[0].get("delta") or {} raw_content = delta.get("content") if isinstance(raw_content, str) and raw_content: return raw_content if isinstance(raw_content, list): parts: list[str] = [] for item in raw_content: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if text_part: parts.append(str(text_part)) elif isinstance(item, str) and item: parts.append(item) if parts: return "".join(parts) raw_reasoning = delta.get("reasoning_content") if isinstance(raw_reasoning, str) and raw_reasoning: return "" if isinstance(raw_reasoning, list): reasoning_parts: list[str] = [] for item in raw_reasoning: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if text_part: reasoning_parts.append(str(text_part)) elif isinstance(item, str) and item: reasoning_parts.append(item) if reasoning_parts: return "" content = getattr(run_event, "content", None) if isinstance(content, str) and content: return content if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if isinstance(text_part, str) and text_part: parts.append(text_part) elif isinstance(item, str) and item: parts.append(item) if parts: return "".join(parts) if isinstance(provider_data, dict): choices = provider_data.get("choices") or [] if choices and isinstance(choices[0], dict): delta = choices[0].get("delta") or {} raw_content = delta.get("content") if isinstance(raw_content, str) and raw_content: return raw_content if isinstance(raw_content, list): parts: list[str] = [] for item in raw_content: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if text_part: parts.append(str(text_part)) elif isinstance(item, str) and item: parts.append(item) if parts: return "".join(parts) return "" def _extract_reasoning_chunk( run_event: Any, trace_fn: Any = None, ) -> str: """Extract reasoning/thought content from a stream event. Shared between stream_chat() and _continue_hitl_run() to avoid duplication. ``trace_fn`` is an optional callable(stage, **kwargs) for trace logging. """ def _trace(stage: str, **kwargs: Any) -> None: if trace_fn: trace_fn(stage, **kwargs) provider_data = getattr(run_event, "model_provider_data", None) if isinstance(provider_data, dict): choices = provider_data.get("choices") or [] if choices and isinstance(choices[0], dict): delta = choices[0].get("delta") or {} raw_reasoning = delta.get("reasoning_content") if isinstance(raw_reasoning, str) and raw_reasoning: _trace("reasoning_source", source="provider_data.delta.reasoning_content") return raw_reasoning if isinstance(raw_reasoning, list): parts: list[str] = [] for item in raw_reasoning: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if text_part: parts.append(str(text_part)) elif isinstance(item, str) and item: parts.append(item) if parts: _trace("reasoning_source", source="provider_data.delta.reasoning_content[]") return "".join(parts) reasoning = getattr(run_event, "reasoning_content", None) if isinstance(reasoning, str) and reasoning: _trace("reasoning_source", source="run_event.reasoning_content") return reasoning if isinstance(reasoning, list): parts: list[str] = [] for item in reasoning: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if text_part: parts.append(str(text_part)) elif isinstance(item, str) and item: parts.append(item) if parts: _trace("reasoning_source", source="run_event.reasoning_content[]") return "".join(parts) if isinstance(provider_data, dict): choices = provider_data.get("choices") or [] if choices and isinstance(choices[0], dict): delta = choices[0].get("delta") or {} raw_reasoning = delta.get("reasoning_content") if isinstance(raw_reasoning, str) and raw_reasoning: _trace("reasoning_source", source="provider_data.delta.reasoning_content") return raw_reasoning if isinstance(raw_reasoning, list): parts: list[str] = [] for item in raw_reasoning: if isinstance(item, dict): text_part = item.get("text") or item.get("content") if text_part: parts.append(str(text_part)) if parts: _trace("reasoning_source", source="provider_data.delta.reasoning_content[]") return "".join(parts) # NOTE: # Some providers (e.g. DeepSeek-compatible streams) may place # assistant answer tokens under `delta.reasoning`. # Treating that field as reasoning can misclassify answer text as thought. # Keep reasoning extraction strict to `reasoning_content` only. return "" def _is_raw_events_log_enabled() -> bool: value = str(os.getenv("QURIO_RAW_EVENTS_LOG", "0")).strip().lower() return value not in {"0", "false", "off", "no"} def _raw_events_log_path() -> Path: configured = str(os.getenv("QURIO_RAW_EVENTS_LOG_PATH", "")).strip() if configured: return Path(configured) logs_dir = Path(__file__).resolve().parents[2] / "logs" logs_dir.mkdir(parents=True, exist_ok=True) date_tag = datetime.utcnow().strftime("%Y%m%d") return logs_dir / f"agno_raw_events_{date_tag}.jsonl" def _append_raw_event_log( *, phase: str, request: StreamChatRequest, run_id: str | None, run_event: Any, ) -> None: if not _is_raw_events_log_enabled(): return try: event_name = str(getattr(run_event, "event", "") or "") content_chunk = _extract_text_chunk(run_event) reasoning_chunk = _extract_reasoning_chunk(run_event) payload = { "timestamp": datetime.utcnow().isoformat() + "Z", "phase": phase, "provider": request.provider, "model": request.model, "conversation_id": request.conversation_id, "run_id": run_id or getattr(run_event, "run_id", None), "event_name": event_name, "raw_event_type": type(run_event).__name__, "has_content": bool(str(content_chunk or "").strip()), "has_reasoning_content": bool(str(reasoning_chunk or "").strip()), "content_preview": _preview(content_chunk), "reasoning_preview": _preview(reasoning_chunk), "raw_event": repr(run_event), } log_path = _raw_events_log_path() with log_path.open("a", encoding="utf-8") as f: f.write(json.dumps(payload, ensure_ascii=False) + "\n") except Exception: # Never break stream flow due to diagnostics logging failures. return def _extract_completed_content_and_output( run_event: Any, streamed_content: str = "", ) -> tuple[str, Any]: """ Extract final assistant content/output from RunCompleted-style events. Shared between normal stream_chat() and HITL continuation to keep behavior aligned. """ agn_content = getattr(run_event, "content", None) run_response = getattr(run_event, "run_response", None) if not agn_content and run_response is not None: agn_content = getattr(run_response, "content", None) final_content = streamed_content or "" output = None # Structured output should override streamed text to preserve canonical payload. if agn_content and hasattr(agn_content, "model_dump"): output = agn_content final_content = json.dumps(agn_content.model_dump(), ensure_ascii=False) elif isinstance(agn_content, (dict, list)): output = agn_content final_content = json.dumps(agn_content, ensure_ascii=False) elif isinstance(agn_content, str) and agn_content.strip() and not final_content: final_content = agn_content # Fallback for providers that only keep final assistant text in run_response.messages. if not final_content and run_response is not None: try: rr_messages = getattr(run_response, "messages", None) or [] for rr_msg in reversed(rr_messages): rr_role = getattr(rr_msg, "role", None) rr_content = getattr(rr_msg, "content", None) if rr_role != "assistant": continue extracted = _extract_text_from_message_content(rr_content).strip() if extracted: final_content = extracted break except Exception: pass return final_content, output def _extract_text_from_message_content(content: Any) -> str: """ Best-effort text extraction for provider-specific assistant message payloads. """ if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, str): parts.append(item) continue if isinstance(item, dict): text_part = item.get("text") if isinstance(text_part, str): parts.append(text_part) continue content_part = item.get("content") if isinstance(content_part, str): parts.append(content_part) continue if isinstance(content_part, (list, dict)): nested = _extract_text_from_message_content(content_part) if nested: parts.append(nested) parts_part = item.get("parts") if isinstance(parts_part, (list, dict)): nested = _extract_text_from_message_content(parts_part) if nested: parts.append(nested) return "".join(parts) if isinstance(content, dict): text_part = content.get("text") if isinstance(text_part, str): return text_part content_part = content.get("content") if isinstance(content_part, str): return content_part if isinstance(content_part, (list, dict)): nested = _extract_text_from_message_content(content_part) if nested: return nested parts_part = content.get("parts") if isinstance(parts_part, (list, dict)): nested = _extract_text_from_message_content(parts_part) if nested: return nested return "" def _coerce_tool_result_payload(output: Any) -> Any: """ Normalize tool output payload into JSON-friendly objects when possible. """ normalized = output if normalized and isinstance(normalized, str): try: normalized = json.loads(normalized) except json.JSONDecodeError: pass if isinstance(normalized, str): try: parsed = ast.literal_eval(normalized) if isinstance(parsed, dict): normalized = parsed except (ValueError, SyntaxError): pass return normalized def _build_tool_result_event( tool: Any, duration_ms: int | None, normalize_tool_output_fn: Any, agent_info: dict[str, str | None] | None = None, ) -> tuple[dict[str, Any], Any]: """ Build frontend ToolResultEvent payload and return parsed tool output. """ output = _coerce_tool_result_payload(normalize_tool_output_fn(getattr(tool, "result", None))) event = ToolResultEvent( id=getattr(tool, "tool_call_id", None), name=getattr(tool, "tool_name", "") or "", status="done" if not getattr(tool, "tool_call_error", None) else "error", output=output, durationMs=duration_ms, agent_id=agent_info.get("agent_id") if agent_info else None, agent_name=agent_info.get("agent_name") if agent_info else None, agent_role=agent_info.get("agent_role") if agent_info else None, agent_emoji=agent_info.get("agent_emoji") if agent_info else None, ).model_dump(by_alias=True, exclude_none=True) return event, output def _build_tool_call_event( tool: Any, text_index: int, include_none: bool = False, agent_info: dict[str, str | None] | None = None, ) -> dict[str, Any]: payload = ToolCallEvent( id=getattr(tool, "tool_call_id", None), name=getattr(tool, "tool_name", "") or "", arguments=json.dumps(getattr(tool, "tool_args", None) or {}), text_index=text_index, agent_id=agent_info.get("agent_id") if agent_info else None, agent_name=agent_info.get("agent_name") if agent_info else None, agent_role=agent_info.get("agent_role") if agent_info else None, agent_emoji=agent_info.get("agent_emoji") if agent_info else None, ) if include_none: return payload.model_dump(by_alias=True, exclude_none=False) return payload.model_dump(by_alias=True, exclude_none=True) def _normalize_interactive_form_fields(raw_fields: Any) -> list[dict[str, Any]]: """ Normalize interactive_form fields to a strict list[dict]. Some providers/tool runtimes return fields as a JSON string; this helper parses and sanitizes that shape so FormRequestEvent validation won't fail. """ parsed = raw_fields if isinstance(parsed, str): text = parsed.strip() if not text: return [] try: parsed = json.loads(text) except Exception: try: parsed = ast.literal_eval(text) except Exception: logger.warning("interactive_form fields is invalid string, fallback to empty list") return [] if isinstance(parsed, dict): maybe_fields = parsed.get("fields") if isinstance(maybe_fields, list): parsed = maybe_fields else: return [] if not isinstance(parsed, list): return [] normalized: list[dict[str, Any]] = [] used_names: set[str] = set() def _slugify_name(value: Any, fallback_index: int) -> str: base = re.sub(r"[^a-zA-Z0-9_]+", "_", str(value or "").strip().lower()).strip("_") if not base: base = f"field_{fallback_index}" candidate = base suffix = 2 while candidate in used_names: candidate = f"{base}_{suffix}" suffix += 1 used_names.add(candidate) return candidate def _normalize_field_type(value: Any) -> str: candidate = str(value or "").strip().lower() if candidate in {"text", "number", "select", "checkbox", "range"}: return candidate return "text" for idx, item in enumerate(parsed, start=1): if isinstance(item, str): label = item.strip() if not label: continue normalized.append( { "name": _slugify_name(label, idx), "label": label, "type": "text", "required": False, } ) continue if not isinstance(item, dict): logger.warning("interactive_form field item is not dict, skipped: %s", type(item).__name__) continue raw_name = item.get("name") raw_label = item.get("label") label = str(raw_label or raw_name or f"Field {idx}").strip() or f"Field {idx}" field_name = _slugify_name(raw_name or label, idx) field_type = _normalize_field_type(item.get("type")) normalized_item = dict(item) normalized_item["name"] = field_name normalized_item["label"] = label normalized_item["type"] = field_type normalized_item["required"] = bool(item.get("required", False)) normalized.append(normalized_item) return normalized def _extract_interactive_form_payload(req: Any, default_title: str) -> tuple[str | None, str, list[dict[str, Any]]]: """Extract interactive form payload from requirement in a validation-safe way.""" tool_args = req.tool_execution.tool_args if getattr(req, "tool_execution", None) else {} if not isinstance(tool_args, dict): tool_args = {} form_id = tool_args.get("id") title = str(tool_args.get("title") or default_title) fields = _normalize_interactive_form_fields(tool_args.get("fields", [])) return form_id, title, fields def _is_interactive_form_requirement(req: Any) -> bool: tool_exec = getattr(req, "tool_execution", None) tool_name = getattr(tool_exec, "tool_name", None) if tool_exec else None return tool_name == "interactive_form" class StreamChatService: """Stream chat service implemented using Agno Agent streaming events.""" def __init__(self) -> None: self._last_memory_optimization: dict[str, float] = {} async def stream_chat( self, request: StreamChatRequest, ) -> AsyncGenerator[dict[str, Any], None]: """ Stream chat completion with HITL support. If request.run_id is present, this is a resumption request after form submission. Otherwise, this is a normal chat request. """ # ================================================================ # HITL: Check if this is a resumption request # ================================================================ if request.run_id and request.field_values: _log_verbose_info(f"Detected HITL resumption request (run_id: {request.run_id})") async for event in self._continue_hitl_run(request): yield event return # Debug: Log request fields for expert mode if request.expert_mode: logger.info(f"[DEBUG] Expert mode request - leader_agent_id: {getattr(request, 'leader_agent_id', 'NOT_FOUND')}, team_agent_ids: {getattr(request, 'team_agent_ids', [])}") # ================================================================ # Normal chat flow # ================================================================ try: if not request.provider: raise ValueError("Missing required field: provider") if not request.messages: raise ValueError("Missing required field: messages") # Enable skills for the definitive user-facing chat agent request.enable_skills = True # Build standard agent or Team agent_metadata: dict[str, Any] = {} if request.expert_mode and getattr(request, "team_agent_ids", []): # Log leader configuration for debugging logger.info(f"[TEAM] Building team - leader_agent_id: {getattr(request, 'leader_agent_id', None)}, team_agent_ids: {request.team_agent_ids}") # 1. Resolve Leader Configuration if ID is provided if getattr(request, "leader_agent_id", None): request = resolve_agent_config(request.leader_agent_id, request) logger.info(f"[TEAM] Leader resolved - agent_id: {getattr(request, 'agent_id', None)}, agent_name: {getattr(request, 'agent_name', None)}") if request.agent_id: agent_metadata[request.agent_id] = { "model": request.model, "provider": request.provider, } # 2. Resolve Member Agents members = [] for a_id in request.team_agent_ids: import copy sub_req = copy.deepcopy(request) sub_req.expert_mode = False # Fetch actual member config from DB member_req = resolve_agent_config(a_id, sub_req) members.append(get_agent_for_provider(member_req)) # Capture member metadata if member_req.agent_id: agent_metadata[member_req.agent_id] = { "model": member_req.model, "provider": member_req.provider, } if member_req.agent_name: agent_metadata[member_req.agent_name] = { "model": member_req.model, "provider": member_req.provider, } # 3. Build the Team with resolved leader (request) and members agent = build_team(request, members) is_team_mode = True else: agent = get_agent_for_provider(request) is_team_mode = False sources_map: dict[str, Any] = {} full_content = "" full_thought = "" tool_start_times: dict[str, float] = {} should_break_next_thought = False in_reasoning_phase = False reasoning_closed_for_current_cycle = False in_content_think_block = False inline_tool_trace_depth = 0 inline_protocol_tail = "" stream_trace = _is_stream_trace_enabled() # Current agent info for Team mode (updated per event) current_agent_info: dict[str, Any] = {"agent_id": None, "agent_name": None} last_active_agent_id = None def trace_stream(stage: str, **kwargs: Any) -> None: if not stream_trace: return payload = ", ".join([f"{k}={v}" for k, v in kwargs.items()]) logger.info(f"[STREAM_TRACE][main] {stage} | {payload}") def emit_thought_part(part: str): nonlocal full_thought, full_content, should_break_next_thought, in_reasoning_phase, reasoning_closed_for_current_cycle text = _strip_internal_tool_trace(str(part or "")) if not text or not text.strip(): return should_break_next_thought = False in_reasoning_phase = True full_thought += text trace_stream("emit_reasoning", reasoning_preview=_preview(text)) current_text_index = len(full_content) yield ThoughtEvent( content=text, text_index=current_text_index, agent_id=current_agent_info.get("agent_id"), agent_name=current_agent_info.get("agent_name"), agent_status=current_agent_info.get("status"), ).model_dump(by_alias=True, exclude_none=True) def process_text(text: str): nonlocal full_content, in_reasoning_phase, should_break_next_thought, reasoning_closed_for_current_cycle clean_text = _strip_internal_tool_trace(text) if clean_text: has_visible_text = bool(clean_text.strip()) if has_visible_text: in_reasoning_phase = False should_break_next_thought = True reasoning_closed_for_current_cycle = True full_content += clean_text yield TextEvent( content=clean_text, agent_id=current_agent_info.get("agent_id"), agent_name=current_agent_info.get("agent_name"), agent_status=current_agent_info.get("status"), ).model_dump(by_alias=True, exclude_none=True) # Agent status tracking for Team mode agent_statuses: dict[str, str] = {} def set_agent_status(agent_id: str | None, status: str): if not agent_id: return if agent_statuses.get(agent_id) == status: return agent_statuses[agent_id] = status return AgentStatusEvent(agentId=agent_id, status=status).model_dump(by_alias=True) async def update_status_and_yield(agent_id: str | None, status: str): event = set_agent_status(agent_id, status) if event: yield event # Context management now handled by Agno's num_history_runs parameter messages = request.messages pre_events: list[dict[str, Any]] = [] messages = self._inject_local_time_context(messages, request, pre_events) enabled_tool_names = self._collect_enabled_tool_names(request) messages = self._inject_tool_guidance(messages, enabled_tool_names, request) for event in pre_events: yield event # ================================================================ # MANUAL CONTEXT MANAGEMENT (Rolling Summary + Fixed Window) # ================================================================ # 1. Fetch Session Summary from DB session_summary_text = None old_summary_json = None if request.conversation_id: try: from ..models.db import DbFilter, DbQueryRequest from .db_service import execute_db_async, get_db_adapter adapter = get_db_adapter(request.database_provider) if adapter: req = DbQueryRequest( providerId=adapter.config.id, action="select", table="conversations", columns=["session_summary"], filters=[DbFilter(op="eq", column="id", value=request.conversation_id)], maybeSingle=True, ) result = await execute_db_async(adapter, req) if result.data and isinstance(result.data, dict): row = result.data raw_summary = row.get("session_summary") if raw_summary: # Parsing handled by adapter often, but double check if isinstance(raw_summary, str): try: old_summary_json = json.loads(raw_summary) except (ValueError, json.JSONDecodeError): pass elif isinstance(raw_summary, dict): old_summary_json = raw_summary if old_summary_json: session_summary_text = old_summary_json.get("summary") except Exception as e: logger.warning(f"Failed to fetch session summary: {e}") logger.error(f"Failed to fetch session summary: {e}") # 2. Slice History (Turn-Based Window) # Strategy: Keep all System messages + Last N User turns (User + AI + Tools) # N comes from frontend context setting: contextTurns. raw_turn_limit = request.context_turn_limit turn_limit = ( max(1, min(50, int(raw_turn_limit))) if isinstance(raw_turn_limit, int) and raw_turn_limit > 0 else 2 ) # Separate System and Non-System system_messages = [m for m in messages if m.get("role") == "system"] chat_messages = [m for m in messages if m.get("role") != "system"] # Find the indices of User messages to determine run boundaries user_indices = [i for i, m in enumerate(chat_messages) if m.get("role") == "user"] user_turn_count = len(user_indices) if user_turn_count > turn_limit: cutoff_index = user_indices[-turn_limit] recent_history = chat_messages[cutoff_index:] else: recent_history = chat_messages # For single-turn requests (common during first-turn regenerate), # using persisted summary can re-introduce stale assistant text. # In this case, use fresh request messages only and rebuild summary from this turn. is_single_user_turn = user_turn_count <= 1 # Force rebuild if it's the first turn OR if the user is editing/regenerating should_rebuild_summary = bool(is_single_user_turn or request.is_editing) # Inject summary only when history exceeds turn window and request is not rebuild flow. should_inject_summary = bool(session_summary_text) and (user_turn_count > turn_limit) and (not should_rebuild_summary) if not should_inject_summary and session_summary_text: _log_verbose_info( "Skipping session summary injection (within turn window or single-turn rebuild context)." ) session_summary_text = None old_summary_json = None # 3. Inject Summary into System Prompt if session_summary_text: summary_prompt = ( "\n\nSession memory summary:\n" "Here is a summary of the conversation so far. Use this to understand long-term context, " "but prioritize the details in the recent messages below.\n" f"{session_summary_text}\n" ) # Inject into the LAST system message, or create a new one if none exist if system_messages: last_sys = system_messages[-1] # Avoid appending if already present (defensive) if "Session memory summary:" not in str(last_sys.get("content", "")): new_content = str(last_sys.get("content", "")) + summary_prompt # Update the dict (need to be careful not to mutate original request list in place if reused, but here it's fine) last_sys["content"] = new_content else: system_messages.append({"role": "system", "content": summary_prompt}) # Final Agent Input agent_input = system_messages + recent_history stream = agent.arun( input=agent_input, stream=True, stream_events=True, user_id=request.user_id, session_id=request.conversation_id, # Only pass explicit structured-output schema. # Do not fallback to response_format, otherwise {"type":"json_object"} # may be treated as grammar and trigger provider-side grammar cache errors. output_schema=request.output_schema, ) # ================================================================ # Stream processing with HITL support # ================================================================ async for run_event in stream: _append_raw_event_log( phase="main", request=request, run_id=getattr(run_event, "run_id", None), run_event=run_event, ) # ============================================================ # HITL: Check if agent paused for user input # ============================================================ if hasattr(run_event, 'is_paused') and run_event.is_paused: logger.info(f"Agent paused for HITL (run_id: {run_event.run_id})") # Extract requirements requirements = getattr(run_event, 'active_requirements', None) or getattr(run_event, 'requirements', None) if requirements: form_requirements = [req for req in requirements if _is_interactive_form_requirement(req)] if not form_requirements: logger.info("Agent paused without interactive_form; skipping HITL form handling") yield DoneEvent( content=full_content or "", thought=full_thought.strip() or None, sources=list(sources_map.values()) or None, ).model_dump() return # Save to Supabase try: paused_tools = getattr(run_event, "tools", None) or [] serialized_tools = [ tool.to_dict() if hasattr(tool, "to_dict") else tool for tool in paused_tools if tool is not None ] serialized_requirements = [ req.to_dict() if hasattr(req, "to_dict") else req for req in form_requirements ] paused_run_output = { "run_id": getattr(run_event, "run_id", None), "session_id": getattr(run_event, "session_id", None) or request.conversation_id, "user_id": request.user_id, "messages": messages or [], "tools": serialized_tools, "requirements": serialized_requirements, "status": "PAUSED", } logger.info( f"[HITL] Saving pending run_id={run_event.run_id} " f"with database_provider={request.database_provider}" ) hitl_storage = get_hitl_storage(request.database_provider) saved = await hitl_storage.save_pending_run( run_id=run_event.run_id, requirements=form_requirements, conversation_id=request.conversation_id, user_id=request.user_id, agent_model=request.model, messages=messages, run_output=paused_run_output, ) if not saved: raise RuntimeError("Failed to persist HITL pending run") # Extract form fields for frontend for req in form_requirements: # Handle external execution (e.g., interactive_form with external_execution=True) if (hasattr(req, 'needs_external_execution') and req.needs_external_execution) or \ (req.tool_execution and req.tool_execution.tool_name == "interactive_form"): form_id, title, fields = _extract_interactive_form_payload( req, default_title="Please provide the following information", ) # Send form_request event to frontend yield FormRequestEvent( run_id=run_event.run_id, form_id=form_id, title=title, fields=fields ).model_dump() # Fallback handle traditional user input (e.g., get_user_input) elif req.needs_user_input and req.user_input_schema: # Convert from user_input_schema form_id = None title = "Please provide the following information" fields = [ { "name": field.name, "type": self._map_field_type_to_frontend(field.field_type), "label": field.description or field.name, "required": True, "value": field.value } for field in req.user_input_schema ] # Send form_request event to frontend yield FormRequestEvent( run_id=run_event.run_id, form_id=form_id, title=title, fields=fields ).model_dump() # Send done event to indicate pause yield DoneEvent( content=full_content or "", thought=full_thought.strip() or None, sources=list(sources_map.values()) or None, ).model_dump() _log_verbose_info(f"HITL pause successful, waiting for user submission (run_id: {run_event.run_id})") return # Exit stream, wait for user to submit form except Exception as e: logger.error(f"Failed to save HITL state: {e}") yield ErrorEvent(error=f"Failed to pause for form: {str(e)}").model_dump() return else: logger.warning("Agent paused but no requirements found") yield DoneEvent( content=full_content or "", thought=full_thought.strip() or None, sources=list(sources_map.values()) or None, ).model_dump() return # ============================================================ # Normal streaming events (use stream_events for details) # ============================================================ # Check if this is a detailed event (from stream_events=True) if hasattr(run_event, 'event'): # Extract agent info for Team mode (member vs leader identification) current_agent_info = _extract_agent_info_from_event( run_event, leader_id=request.agent_id, leader_name=request.agent_name, leader_emoji=request.agent_emoji, agent_metadata=agent_metadata, ) # Log active agent switch in Team mode if is_team_mode: current_id = current_agent_info.get("agent_id") if current_id != last_active_agent_id: last_active_agent_id = current_id active_name = current_agent_info.get("agent_name") active_role = current_agent_info.get("agent_role") active_model = current_agent_info.get("model") active_provider = current_agent_info.get("provider") logger.info( f"[TEAM] >>> Active Agent Switch: {active_name} ({active_role}) " f"| Model: {active_model} | Provider: {active_provider}" ) # Apply current tracked status to info for text/thought events current_agent_info["status"] = agent_statuses.get(current_id, "active") if current_agent_info.get("agent_role") == "member": trace_stream( "member_event", agent_id=current_agent_info.get("agent_id"), agent_name=current_agent_info.get("agent_name"), ) match run_event.event: case RunEvent.run_started.value | TeamRunEvent.run_started: if is_team_mode: active_id = current_agent_info.get("agent_id") active_name = current_agent_info.get("agent_name") active_role = current_agent_info.get("agent_role") active_model = current_agent_info.get("model") active_provider = current_agent_info.get("provider") logger.info( f"[TEAM] >>> run_started: {active_name} ({active_role}) " f"| Model: {active_model} | Provider: {active_provider}" ) # Member starts -> Leader waits, Member active if active_role == "member": # Ensure leader is set to waiting when member starts async for e in update_status_and_yield(request.agent_id, "waiting"): yield e async for e in update_status_and_yield(active_id, "active"): yield e else: # Leader starts -> Leader active async for e in update_status_and_yield(active_id, "active"): yield e continue case TeamRunEvent.run_completed: if is_team_mode: active_id = current_agent_info.get("agent_id") active_role = current_agent_info.get("agent_role") if active_role == "member": # Member finished -> Leader still waiting (until it resumes), Member ready async for e in update_status_and_yield(active_id, "ready"): yield e continue # Handle both Agent RunEvent and Team TeamRunEvent for content streaming case RunEvent.run_content.value | TeamRunEvent.run_content: raw_content_chunk = _extract_text_chunk(run_event) raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, had_protocol = _strip_inline_tool_protocol( raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, ) if had_protocol: trace_stream( "strip_tool_protocol", depth=inline_tool_trace_depth, tail_len=len(inline_protocol_tail), cleaned_preview=_preview(raw_content_chunk), ) raw_reasoning = _extract_reasoning_chunk(run_event, trace_fn=trace_stream) content_segments, in_content_think_block = _split_content_by_think_tags( raw_content_chunk, in_content_think_block, ) content_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "text" ) inline_thought_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "thought" ) reasoning = raw_reasoning if reasoning and content_chunk and _is_reasoning_duplicate_of_content( str(reasoning), str(content_chunk), ): trace_stream( "suppress_reasoning_overlap", event="run_content", reasoning_preview=_preview(reasoning), content_preview=_preview(content_chunk), ) reasoning = "" has_content_chunk = bool(content_chunk) has_inline_thought = bool(inline_thought_chunk) trace_stream( "run_content", has_content=has_content_chunk, has_reasoning=bool(reasoning) or has_inline_thought, reasoning_closed=reasoning_closed_for_current_cycle, content_preview=_preview(content_chunk), reasoning_preview=_preview((reasoning or "") + (inline_thought_chunk or "")), ) has_any_thought = bool(reasoning) or has_inline_thought if has_any_thought and reasoning_closed_for_current_cycle: # Re-open reasoning phase (e.g. after tool call or interleaved model output). reasoning_closed_for_current_cycle = False in_reasoning_phase = False should_break_next_thought = True if reasoning: for event in emit_thought_part(str(reasoning)): yield event if has_inline_thought: for event in emit_thought_part(str(inline_thought_chunk)): yield event if content_chunk: for e in process_text(content_chunk): yield e case RunEvent.reasoning_content_delta.value | TeamRunEvent.reasoning_content_delta: raw_content_chunk = _extract_text_chunk(run_event) raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, had_protocol = _strip_inline_tool_protocol( raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, ) if had_protocol: trace_stream( "strip_tool_protocol", depth=inline_tool_trace_depth, tail_len=len(inline_protocol_tail), cleaned_preview=_preview(raw_content_chunk), ) raw_reasoning = _extract_reasoning_chunk(run_event) content_segments, in_content_think_block = _split_content_by_think_tags( raw_content_chunk, in_content_think_block, ) content_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "text" ) inline_thought_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "thought" ) reasoning = raw_reasoning if reasoning and content_chunk and _is_reasoning_duplicate_of_content( str(reasoning), str(content_chunk), ): trace_stream( "suppress_reasoning_overlap", event="reasoning_content_delta", reasoning_preview=_preview(reasoning), content_preview=_preview(content_chunk), ) reasoning = "" has_content_chunk = bool(content_chunk) has_inline_thought = bool(inline_thought_chunk) trace_stream( "reasoning_delta", has_content=has_content_chunk, has_reasoning=bool(reasoning) or has_inline_thought, reasoning_closed=reasoning_closed_for_current_cycle, content_preview=_preview(content_chunk), reasoning_preview=_preview((reasoning or "") + (inline_thought_chunk or "")), ) has_any_thought = bool(reasoning) or has_inline_thought if has_any_thought and reasoning_closed_for_current_cycle: reasoning_closed_for_current_cycle = False in_reasoning_phase = False should_break_next_thought = True if reasoning: for event in emit_thought_part(str(reasoning)): yield event if has_inline_thought: for event in emit_thought_part(str(inline_thought_chunk)): yield event if content_chunk: for e in process_text(content_chunk): yield e case RunEvent.tool_call_started.value | TeamRunEvent.tool_call_started: tool_event: ToolCallStartedEvent = run_event # type: ignore[assignment] tool = tool_event.tool if tool: in_reasoning_phase = False should_break_next_thought = True reasoning_closed_for_current_cycle = False in_content_think_block = False inline_tool_trace_depth = 0 inline_protocol_tail = "" if getattr(tool, "tool_call_id", None): tool_start_times[tool.tool_call_id] = time.time() current_id = current_agent_info.get("agent_id") # If leader calls a tool, it's either an internal tool (code, etc) or delegation. # During the tool call itself, the agent is "active". async for e in update_status_and_yield(current_id, "active"): yield e trace_stream( "tool_call_started", tool_name=getattr(tool, "tool_name", ""), tool_call_id=getattr(tool, "tool_call_id", None), ) current_text_index = len(full_content) yield _build_tool_call_event( tool, current_text_index, agent_info=current_agent_info ) case RunEvent.tool_call_completed.value | TeamRunEvent.tool_call_completed: tool_event: ToolCallCompletedEvent = run_event # type: ignore[assignment] tool = tool_event.tool if tool: in_reasoning_phase = False should_break_next_thought = True reasoning_closed_for_current_cycle = False in_content_think_block = False inline_tool_trace_depth = 0 inline_protocol_tail = "" duration_ms = None if tool.tool_call_id and tool.tool_call_id in tool_start_times: duration_ms = int((time.time() - tool_start_times[tool.tool_call_id]) * 1000) trace_stream( "tool_call_completed", tool_name=tool.tool_name or "", tool_call_id=tool.tool_call_id, is_error=bool(tool.tool_call_error), ) tool_result_event, output = _build_tool_result_event( tool, duration_ms, self._normalize_tool_output, agent_info=current_agent_info, ) yield tool_result_event self._collect_search_sources(output, sources_map) case RunEvent.run_completed.value | TeamRunEvent.run_completed: # Extract agent info to check if this is a member or leader event_agent_info = _extract_agent_info_from_event( run_event, leader_id=request.agent_id, leader_name=request.agent_name, leader_emoji=request.agent_emoji, agent_metadata=agent_metadata, ) is_member_completion = event_agent_info.get("agent_role") == "member" # In Team Mode, only terminate when the LEADER (no agent_id on event) completes. # Member completions should just let the main loop continue. if is_team_mode and is_member_completion: active_id = event_agent_info.get("agent_id") active_name = event_agent_info.get("agent_name") active_model = event_agent_info.get("model") active_provider = event_agent_info.get("provider") logger.info( f"[TEAM] Member {active_name} completed " f"(Model: {active_model} | Provider: {active_provider}). " "Continuing stream..." ) # Ensure member is marked as ready if not already handled by TeamRunEvent.run_completed async for e in update_status_and_yield(active_id, "ready"): yield e continue # Leader completed if is_team_mode: async for e in update_status_and_yield(request.agent_id, "idle"): yield e final_content, output = _extract_completed_content_and_output( run_event, full_content, ) yield DoneEvent( content=final_content or "", output=output, thought=full_thought.strip() or None, sources=list(sources_map.values()) or None, ).model_dump() if request: asyncio.create_task(self._maybe_optimize_memories(agent, request)) # 4. Trigger Async Session Summary Update # Only if conversation_id exists (Main Chat Flow) AND summary is enabled if request.conversation_id and request.enable_session_summary: # Prepare summary lines: # - Normal flow: incremental update with last user + new assistant # - Regenerate/Edit (or single-turn rebuild): rebuild from current request context + new assistant if should_rebuild_summary: new_lines = [ m for m in messages if m.get("role") in ("user", "assistant") ] new_lines.append({"role": "assistant", "content": final_content}) else: new_lines = [] last_user = next((m for m in reversed(messages) if m.get("role") == "user"), None) if last_user: new_lines.append(last_user) new_lines.append({"role": "assistant", "content": final_content}) _log_verbose_info(f"Triggering async summary update for {request.conversation_id} with {len(new_lines)} messages (rebuild: {should_rebuild_summary}, is_editing: {request.is_editing})") asyncio.create_task(update_session_summary( conversation_id=request.conversation_id, old_summary=old_summary_json, new_messages=new_lines, database_provider=request.database_provider, memory_provider=request.memory_provider, memory_model=request.memory_model, memory_api_key=request.memory_api_key, memory_base_url=request.memory_base_url, summary_provider=request.summary_provider, summary_model=request.summary_model, summary_api_key=request.summary_api_key, summary_base_url=request.summary_base_url, rebuild_from_scratch=should_rebuild_summary, )) return case RunEvent.run_error.value | TeamRunEvent.run_error: error_msg = _extract_best_error_message(run_event) active_id = current_agent_info.get("agent_id") if active_id: async for e in update_status_and_yield(active_id, "error"): yield e yield ErrorEvent(error=error_msg).model_dump() return else: # Simple event Fallback (no detailed event type), just check for content raw_content_chunk = _extract_text_chunk(run_event) raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, _ = _strip_inline_tool_protocol( raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, ) raw_reasoning = _extract_reasoning_chunk(run_event) content_segments, in_content_think_block = _split_content_by_think_tags( raw_content_chunk, in_content_think_block, ) content_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "text" ) inline_thought_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "thought" ) if raw_reasoning: for event in emit_thought_part(str(raw_reasoning)): yield event if inline_thought_chunk: for event in emit_thought_part(str(inline_thought_chunk)): yield event if content_chunk: for e in process_text(content_chunk): yield e except Exception as exc: logger.error(f"Stream chat error: {exc}") yield ErrorEvent(error=_extract_best_error_message(exc)).model_dump() async def _continue_hitl_run( self, request: StreamChatRequest, ) -> AsyncGenerator[dict[str, Any], None]: """ Continue a paused HITL run after user submits form. This method: 1. Retrieves requirements from storage 2. Rebuilds continuation messages with submitted form values 3. Runs agent.arun() and streams completion 4. Cleans up storage record """ try: run_id = request.run_id field_values = request.field_values or {} _log_verbose_info( f"[HITL] Continuing run_id={run_id!r} " f"with database_provider={request.database_provider} " f"and field_values={list(field_values.keys())}" ) # 1. Fetch Session Summary from DB session_summary_text = None old_summary_json = None if request.conversation_id: try: from ..models.db import DbFilter, DbQueryRequest from .db_service import execute_db_async, get_db_adapter adapter = get_db_adapter(request.database_provider) if adapter: req = DbQueryRequest( providerId=adapter.config.id, action="select", table="conversations", columns=["session_summary"], filters=[DbFilter(op="eq", column="id", value=request.conversation_id)], maybeSingle=True, ) result = await execute_db_async(adapter, req) if result.data and isinstance(result.data, dict): row = result.data raw_summary = row.get("session_summary") if raw_summary: if isinstance(raw_summary, str): try: old_summary_json = json.loads(raw_summary) except (ValueError, json.JSONDecodeError): pass elif isinstance(raw_summary, dict): old_summary_json = raw_summary if old_summary_json: session_summary_text = old_summary_json.get("summary") except Exception as e: logger.warning(f"Failed to fetch session summary in HITL flow: {e}") # Retrieve requirements from Supabase hitl_storage = get_hitl_storage(request.database_provider) pending = await hitl_storage.get_pending_run(run_id) requirements = None saved_messages = None saved_run_output = None if isinstance(pending, dict): requirements = pending.get("requirements") saved_messages = pending.get("messages") saved_run_output = pending.get("run_output") else: requirements = pending _log_verbose_info( "[HITL] loaded pending payload: " f"has_requirements={bool(requirements)}, " f"messages_type={type(saved_messages).__name__ if saved_messages is not None else 'None'}, " f"has_run_output={saved_run_output is not None}, " f"run_output_type={type(saved_run_output).__name__ if saved_run_output is not None else 'None'}" ) if not requirements: logger.error( f"[HITL] No pending run found for run_id={run_id!r} " f"(database_provider={request.database_provider})" ) yield ErrorEvent(error="Form session expired or not found").model_dump() return # Enable skills for the definitive user-facing chat agent request.enable_skills = True # Get agent (same provider as original request) agent = get_agent_for_provider(request) _log_verbose_info(f"[HITL Continue] Agent instructions: {getattr(agent, 'instructions', None)}") full_content = "" full_thought = "" sources_map: dict[str, Any] = {} tool_start_times: dict[str, float] = {} should_break_next_thought = False in_reasoning_phase = False reasoning_closed_for_current_cycle = False in_content_think_block = False inline_tool_trace_depth = 0 inline_protocol_tail = "" stream_trace = _is_stream_trace_enabled() paused_again = False # Flag to prevent cleanup when multi-form chaining occurs stream_had_error = False completed_content_fallback = "" saw_terminal_completion = False continuation_event_count = 0 last_event_name: str | None = None last_event_type: str | None = None last_event_run_id: str | None = None # Current agent info for Team mode (updated per event) current_agent_info: dict[str, Any] = {"agent_id": request.agent_id, "agent_name": request.agent_name} def trace_stream(stage: str, **kwargs: Any) -> None: if not stream_trace: return payload = ", ".join([f"{k}={v}" for k, v in kwargs.items()]) logger.info(f"[STREAM_TRACE][hitl] {stage} | {payload}") def emit_thought_part(part: str): nonlocal full_thought, full_content, should_break_next_thought, in_reasoning_phase, reasoning_closed_for_current_cycle text = _strip_internal_tool_trace(str(part or "")) if not text or not text.strip(): return should_break_next_thought = False in_reasoning_phase = True full_thought += text trace_stream("emit_reasoning", reasoning_preview=_preview(text)) current_text_index = len(full_content) yield ThoughtEvent( content=text, text_index=current_text_index, agent_id=current_agent_info.get("agent_id"), agent_name=current_agent_info.get("agent_name"), ).model_dump(by_alias=True, exclude_none=True) def process_text(text: str): nonlocal full_content, in_reasoning_phase, should_break_next_thought, reasoning_closed_for_current_cycle clean_text = _strip_internal_tool_trace(text) if clean_text: has_visible_text = bool(clean_text.strip()) if has_visible_text: in_reasoning_phase = False should_break_next_thought = True reasoning_closed_for_current_cycle = True full_content += clean_text yield TextEvent( content=clean_text, agent_id=current_agent_info.get("agent_id"), agent_name=current_agent_info.get("agent_name"), ).model_dump(by_alias=True, exclude_none=True) async def _iterate_run_stream(stream: Any): """ Normalize both async and sync Agno run streams into an async iterator. """ if hasattr(stream, "__aiter__"): async for item in stream: yield item return iterator = iter(stream) sentinel = object() while True: item = await asyncio.to_thread(lambda: next(iterator, sentinel)) if item is sentinel: break yield item async def _stream_events(stream): nonlocal full_content, full_thought, sources_map, tool_start_times, paused_again, stream_had_error nonlocal in_reasoning_phase, should_break_next_thought, reasoning_closed_for_current_cycle nonlocal in_content_think_block, inline_tool_trace_depth, inline_protocol_tail nonlocal completed_content_fallback, saw_terminal_completion nonlocal continuation_event_count, last_event_name, last_event_type, last_event_run_id async for run_event in _iterate_run_stream(stream): _append_raw_event_log( phase="hitl_continuation", request=request, run_id=run_id, run_event=run_event, ) continuation_event_count += 1 last_event_type = type(run_event).__name__ last_event_name = str(getattr(run_event, "event", None) or last_event_type) last_event_run_id = str(raw_event_run_id) if raw_event_run_id else None # Extract agent info for Team mode (though Team HITL is currently disabled) current_agent_info = _extract_agent_info_from_event( run_event, leader_id=request.agent_id, leader_name=request.agent_name, leader_emoji=request.agent_emoji, ) # When yield_run_output=True, acontinue_run may yield the final RunOutput object. # Capture its canonical content as a robust fallback for providers that emit sparse events. # IMPORTANT: Do NOT re-emit this content via process_text() if we already received # streaming chunks (run_content events). Doing so would cause the full answer to be # appended a second time, resulting in visible duplication in the UI. # Only use RunOutput.content as a fallback when the stream produced nothing. if isinstance(run_event, RunOutput): saw_terminal_completion = True completed_content_fallback, _ = _extract_completed_content_and_output( run_event, completed_content_fallback or full_content, ) # Only emit if no content was streamed yet (sparse-event provider fallback). if not full_content: text_from_output = _extract_text_from_message_content( getattr(run_event, "content", None) ).strip() if text_from_output: for e in process_text(text_from_output): yield e continue # HITL Pause Check if hasattr(run_event, 'is_paused') and run_event.is_paused: logger.info(f"Agent paused again during continuation (multi-form chain, run_id: {run_id})") # Extract new requirements new_requirements = getattr(run_event, 'active_requirements', None) or getattr(run_event, 'requirements', None) if new_requirements: form_requirements = [req for req in new_requirements if _is_interactive_form_requirement(req)] if form_requirements: # Save new form requirements (overwrites previous in memory) hitl_storage_multi = get_hitl_storage(request.database_provider) saved = await hitl_storage_multi.save_pending_run( run_id=run_id, requirements=form_requirements, conversation_id=request.conversation_id, user_id=request.user_id, agent_model=request.model, messages=saved_messages, # Reuse saved messages run_output=( { "run_id": getattr(run_event, "run_id", None) or run_id, "session_id": getattr(run_event, "session_id", None) or request.conversation_id, "user_id": request.user_id, "messages": saved_messages or [], "tools": [ tool.to_dict() if hasattr(tool, "to_dict") else tool for tool in (getattr(run_event, "tools", None) or []) if tool is not None ], "requirements": [ req.to_dict() if hasattr(req, "to_dict") else req for req in form_requirements ], "status": "PAUSED", } ), ) if not saved: raise RuntimeError("Failed to persist chained HITL pending run") # Extract form fields and notify frontend for req in form_requirements: if (hasattr(req, 'needs_external_execution') and req.needs_external_execution) or \ (req.tool_execution and req.tool_execution.tool_name == "interactive_form"): form_id, title, fields = _extract_interactive_form_payload( req, default_title="Please provide additional information", ) yield FormRequestEvent( run_id=run_id, form_id=form_id, title=title, fields=fields ).model_dump() # Send partial done event yield DoneEvent( content=full_content or "", thought=full_thought.strip() or None, sources=list(sources_map.values()) or None, ).model_dump() logger.info(f"Multi-form: saved second form, waiting for user (run_id: {run_id})") paused_again = True # Mark as paused again to skip cleanup return # If no form requirements, just continue logger.warning(f"Agent paused again but no interactive_form found (run_id: {run_id})") yield ErrorEvent(error="Agent paused unexpectedly").model_dump() return # Check if this is a detailed event (from stream_events=True or implicit) if hasattr(run_event, 'event'): match run_event.event: case RunEvent.run_content.value: raw_content_chunk = _extract_text_chunk(run_event) raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, had_protocol = _strip_inline_tool_protocol( raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, ) if had_protocol: trace_stream( "strip_tool_protocol", depth=inline_tool_trace_depth, tail_len=len(inline_protocol_tail), cleaned_preview=_preview(raw_content_chunk), ) raw_reasoning = _extract_reasoning_chunk(run_event, trace_fn=trace_stream) content_segments, in_content_think_block = _split_content_by_think_tags( raw_content_chunk, in_content_think_block, ) content_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "text" ) inline_thought_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "thought" ) reasoning = raw_reasoning if reasoning and content_chunk and _is_reasoning_duplicate_of_content( str(reasoning), str(content_chunk), ): trace_stream( "suppress_reasoning_overlap", event="run_content", reasoning_preview=_preview(reasoning), content_preview=_preview(content_chunk), ) reasoning = "" has_content_chunk = bool(content_chunk) has_inline_thought = bool(inline_thought_chunk) trace_stream( "run_content", has_content=has_content_chunk, has_reasoning=bool(reasoning) or has_inline_thought, reasoning_closed=reasoning_closed_for_current_cycle, content_preview=_preview(content_chunk), reasoning_preview=_preview((reasoning or "") + (inline_thought_chunk or "")), ) has_any_thought = bool(reasoning) or has_inline_thought if has_any_thought and reasoning_closed_for_current_cycle: reasoning_closed_for_current_cycle = False in_reasoning_phase = False should_break_next_thought = True if reasoning: for event in emit_thought_part(str(reasoning)): yield event if has_inline_thought: for event in emit_thought_part(str(inline_thought_chunk)): yield event if content_chunk: for e in process_text(content_chunk): yield e trace_stream( "emit_content", reasoning_closed=reasoning_closed_for_current_cycle, content_preview=_preview(content_chunk), ) case RunEvent.reasoning_content_delta.value: raw_content_chunk = _extract_text_chunk(run_event) raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, had_protocol = _strip_inline_tool_protocol( raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, ) if had_protocol: trace_stream( "strip_tool_protocol", depth=inline_tool_trace_depth, tail_len=len(inline_protocol_tail), cleaned_preview=_preview(raw_content_chunk), ) raw_reasoning = _extract_reasoning_chunk(run_event) content_segments, in_content_think_block = _split_content_by_think_tags( raw_content_chunk, in_content_think_block, ) content_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "text" ) inline_thought_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "thought" ) reasoning = raw_reasoning if reasoning and content_chunk and _is_reasoning_duplicate_of_content( str(reasoning), str(content_chunk), ): trace_stream( "suppress_reasoning_overlap", event="reasoning_content_delta", reasoning_preview=_preview(reasoning), content_preview=_preview(content_chunk), ) reasoning = "" has_content_chunk = bool(content_chunk) has_inline_thought = bool(inline_thought_chunk) trace_stream( "reasoning_delta", has_content=has_content_chunk, has_reasoning=bool(reasoning) or has_inline_thought, reasoning_closed=reasoning_closed_for_current_cycle, content_preview=_preview(content_chunk), reasoning_preview=_preview((reasoning or "") + (inline_thought_chunk or "")), ) has_any_thought = bool(reasoning) or has_inline_thought if has_any_thought and reasoning_closed_for_current_cycle: reasoning_closed_for_current_cycle = False in_reasoning_phase = False should_break_next_thought = True if reasoning: for event in emit_thought_part(str(reasoning)): yield event if has_inline_thought: for event in emit_thought_part(str(inline_thought_chunk)): yield event if content_chunk: for e in process_text(content_chunk): yield e trace_stream( "emit_content", reasoning_closed=reasoning_closed_for_current_cycle, content_preview=_preview(content_chunk), ) case RunEvent.tool_call_started.value: tool_event: ToolCallStartedEvent = run_event # type: ignore[assignment] tool = tool_event.tool if tool: in_reasoning_phase = False should_break_next_thought = True reasoning_closed_for_current_cycle = False in_content_think_block = False inline_tool_trace_depth = 0 inline_protocol_tail = "" if tool.tool_call_id: tool_start_times[tool.tool_call_id] = time.time() trace_stream( "tool_call_started", tool_name=tool.tool_name or "", tool_call_id=tool.tool_call_id, ) current_text_index = len(full_content) yield _build_tool_call_event(tool, current_text_index) case RunEvent.tool_call_completed.value: tool_event: ToolCallCompletedEvent = run_event # type: ignore[assignment] tool = tool_event.tool if tool: in_reasoning_phase = False should_break_next_thought = True reasoning_closed_for_current_cycle = False in_content_think_block = False inline_tool_trace_depth = 0 inline_protocol_tail = "" duration_ms = None if tool.tool_call_id and tool.tool_call_id in tool_start_times: duration_ms = int((time.time() - tool_start_times[tool.tool_call_id]) * 1000) trace_stream( "tool_call_completed", tool_name=tool.tool_name or "", tool_call_id=tool.tool_call_id, is_error=bool(tool.tool_call_error), ) tool_result_event, output = _build_tool_result_event( tool, duration_ms, self._normalize_tool_output, ) yield tool_result_event self._collect_search_sources(output, sources_map) case RunEvent.run_completed.value: saw_terminal_completion = True completed_content_fallback, _ = _extract_completed_content_and_output( run_event, completed_content_fallback or full_content, ) case RunEvent.run_error.value: error_msg = _extract_best_error_message(run_event) stream_had_error = True yield ErrorEvent(error=error_msg).model_dump() return else: # Simple event Fallback raw_content_chunk = _extract_text_chunk(run_event) raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, _ = _strip_inline_tool_protocol( raw_content_chunk, inline_tool_trace_depth, inline_protocol_tail, ) raw_reasoning = _extract_reasoning_chunk(run_event) content_segments, in_content_think_block = _split_content_by_think_tags( raw_content_chunk, in_content_think_block, ) content_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "text" ) inline_thought_chunk = "".join( seg_text for seg_type, seg_text in content_segments if seg_type == "thought" ) if raw_reasoning: for event in emit_thought_part(str(raw_reasoning)): yield event if inline_thought_chunk: for event in emit_thought_part(str(inline_thought_chunk)): yield event if content_chunk: for e in process_text(content_chunk): yield e def _build_fallback_messages(): if not saved_messages: return None updated_messages = list(saved_messages) for req in requirements: tool_exec = getattr(req, 'tool_execution', None) tool_name = getattr(tool_exec, 'tool_name', None) if tool_exec else None if tool_name != "interactive_form": continue tool_args = getattr(tool_exec, 'tool_args', {}) if tool_exec else {} tool_call_id = getattr(req, "id", None) or tool_args.get("id") or f"form-{int(time.time() * 1000)}" updated_messages.append({ "role": "assistant", "content": None, "tool_calls": [{ "id": tool_call_id, "type": "function", "function": { "name": "interactive_form", "arguments": json.dumps(tool_args or {}), } }], }) updated_messages.append({ "role": "tool", "content": json.dumps(field_values), "tool_call_id": tool_call_id, }) return updated_messages def _apply_field_values_to_requirements() -> list[Any]: """ Resolve pending HITL requirements with the submitted form payload. Prefer Agno-native requirement resolution so we can use acontinue_run(). """ resolved_requirements: list[Any] = [] serialized_values = json.dumps(field_values, ensure_ascii=False) for req in requirements or []: try: if hasattr(req, "needs_external_execution") and req.needs_external_execution: req.set_external_execution_result(serialized_values) tool_exec = getattr(req, "tool_execution", None) if tool_exec is not None and getattr(tool_exec, "result", None) is None: tool_exec.result = serialized_values elif hasattr(req, "needs_user_input") and req.needs_user_input: req.provide_user_input(field_values) elif hasattr(req, "needs_confirmation") and req.needs_confirmation: req.confirm() except Exception as req_err: logger.warning( f"[HITL] Failed to resolve requirement {getattr(req, 'id', None)} for run_id={run_id}: {req_err}" ) resolved_requirements.append(req) return resolved_requirements def _build_continuation_agent_input(base_messages: list[dict[str, Any]]) -> list[dict[str, Any]]: messages = self._inject_local_time_context(list(base_messages), request, []) system_messages = [m for m in messages if m.get("role") == "system"] chat_messages = [m for m in messages if m.get("role") != "system"] raw_turn_limit = request.context_turn_limit turn_limit = ( max(1, min(50, int(raw_turn_limit))) if isinstance(raw_turn_limit, int) and raw_turn_limit > 0 else 2 ) user_indices = [i for i, m in enumerate(chat_messages) if m.get("role") == "user"] user_turn_count = len(user_indices) if user_turn_count > turn_limit: cutoff_index = user_indices[-turn_limit] recent_history = chat_messages[cutoff_index:] else: recent_history = chat_messages should_inject_summary = bool(session_summary_text) and (user_turn_count > turn_limit) if should_inject_summary: summary_prompt = ( "\n\nSession memory summary:\n" "Here is a summary of the conversation so far. Use this to understand long-term context, " "but prioritize the details in the recent messages below.\n" f"{session_summary_text}\n" ) if system_messages: last_sys = system_messages[-1] if "Session memory summary:" not in str(last_sys.get("content", "")): last_sys["content"] = str(last_sys.get("content", "")) + summary_prompt else: system_messages.append({"role": "system", "content": summary_prompt}) return system_messages + recent_history resolved_requirements = _apply_field_values_to_requirements() stream = None try: restored_run_output = None if isinstance(saved_run_output, dict): try: restored_run_output = RunOutput.from_dict(dict(saved_run_output)) # Ensure external-execution tool results are concretely attached to # run_response.tools before acontinue_run() processes updates. if restored_run_output and isinstance(restored_run_output.tools, list): serialized_values = json.dumps(field_values, ensure_ascii=False) tool_result_by_id: dict[str, Any] = {} for req in resolved_requirements or []: tool_exec = getattr(req, "tool_execution", None) if not tool_exec: continue tcid = getattr(tool_exec, "tool_call_id", None) if tcid: tool_result_by_id[str(tcid)] = getattr(tool_exec, "result", None) for tool in restored_run_output.tools: if getattr(tool, "result", None) is not None: continue tcid = getattr(tool, "tool_call_id", None) if tcid and str(tcid) in tool_result_by_id: tool.result = tool_result_by_id[str(tcid)] continue if getattr(tool, "tool_name", None) == "interactive_form": tool.result = serialized_values # OpenAI-compatible tool flow requires an assistant message # with matching tool_calls before any tool message can appear. if restored_run_output and isinstance(restored_run_output.tools, list): existing_messages = ( list(restored_run_output.messages) if isinstance(restored_run_output.messages, list) else [] ) existing_tool_call_ids: set[str] = set() for msg in existing_messages: msg_tool_calls = getattr(msg, "tool_calls", None) or [] for tc in msg_tool_calls: if isinstance(tc, dict) and tc.get("id"): existing_tool_call_ids.add(str(tc.get("id"))) for tool in restored_run_output.tools: tool_call_id = getattr(tool, "tool_call_id", None) tool_name = getattr(tool, "tool_name", None) or "interactive_form" if not tool_call_id or str(tool_call_id) in existing_tool_call_ids: continue tool_args = getattr(tool, "tool_args", None) or {} existing_messages.append( Message( role="assistant", content="", tool_calls=[ { "id": str(tool_call_id), "type": "function", "function": { "name": str(tool_name), "arguments": json.dumps(tool_args, ensure_ascii=False), }, } ], ) ) existing_tool_call_ids.add(str(tool_call_id)) restored_run_output.messages = existing_messages if resolved_requirements: restored_run_output.requirements = resolved_requirements except Exception as restore_err: logger.warning( f"[HITL] Failed to restore RunOutput for run_id={run_id}, fallback to run_id path: {restore_err}" ) _log_verbose_info( f"[HITL] continuation restore mode: {'run_response' if restored_run_output is not None else 'run_id'} " f"(has_saved_run_output={isinstance(saved_run_output, dict)})" ) _log_verbose_info( f"Running HITL continuation via continue_run (run_id: {run_id}, session_id: {request.conversation_id})" ) if restored_run_output is not None: stream = agent.continue_run( run_response=restored_run_output, stream=True, stream_events=True, yield_run_output=True, user_id=request.user_id, session_id=request.conversation_id, output_schema=request.output_schema, ) else: stream = agent.continue_run( run_id=run_id, requirements=resolved_requirements, stream=True, stream_events=True, yield_run_output=True, user_id=request.user_id, session_id=request.conversation_id, output_schema=request.output_schema, ) except Exception as continue_err: logger.warning( f"[HITL] continue_run failed for run_id={run_id}, fallback to rebuilt arun: {continue_err}" ) fallback_messages = _build_fallback_messages() if not fallback_messages: yield ErrorEvent(error="Form session cannot be resumed (missing state)").model_dump() return agent_input = _build_continuation_agent_input(fallback_messages) stream = agent.arun( input=agent_input, stream=True, stream_events=True, user_id=request.user_id, session_id=request.conversation_id, output_schema=request.output_schema, ) async for event in _stream_events(stream): yield event _log_verbose_info( "[HITL] continuation stream summary: " f"run_id={run_id}, " f"events={continuation_event_count}, " f"last_event={last_event_name}, " f"last_event_type={last_event_type}, " f"last_event_run_id={last_event_run_id}, " f"saw_terminal_completion={saw_terminal_completion}, " f"stream_had_error={stream_had_error}, " f"paused_again={paused_again}" ) if stream_had_error: logger.warning(f"HITL run {run_id} ended with stream error; skipping done/cleanup") return if not saw_terminal_completion: logger.warning( f"HITL run {run_id} stream ended without terminal completion event; skipping done/cleanup" ) yield ErrorEvent(error="HITL continuation ended before completion").model_dump() return # Stream completed, send done event yield DoneEvent( content=full_content or completed_content_fallback, thought=full_thought.strip() or None, sources=list(sources_map.values()) or None, ).model_dump() # Clean up Supabase (skip if paused again for multi-form) if not paused_again: await hitl_storage.delete_pending_run(run_id) logger.info(f"HITL run {run_id} completed and cleaned up") else: logger.info(f"HITL run {run_id} paused again (multi-form), skipping cleanup") # 6. Trigger Async Session Summary Update if request.conversation_id and not paused_again: # For HITL resumption, extract the last turn's context from saved_messages # (matching normal flow: only last user + assistant, not full history) summary_messages = [] # Extract only the last user-assistant turn from saved_messages if saved_messages: last_user_idx = -1 for i in range(len(saved_messages) - 1, -1, -1): if saved_messages[i].get("role") == "user": last_user_idx = i break if last_user_idx >= 0: # Include from last user message to end of saved_messages # This captures: user question -> assistant form(s) -> any intermediate interactions for msg in saved_messages[last_user_idx:]: role = msg.get("role") content = msg.get("content") # Only include user/assistant messages with content for summary if role in ("user", "assistant") and content: summary_messages.append({"role": role, "content": content}) # Add the form submission as user input (provides structured data context) form_submission_text = f"[Form Submitted] Values: {json.dumps(field_values)}" summary_messages.append({"role": "user", "content": form_submission_text}) # Add the new assistant response (based on form data) summary_messages.append({"role": "assistant", "content": full_content}) _log_verbose_info(f"Triggering async summary update for {request.conversation_id} (Resumed HITL flow, {len(summary_messages)} messages)") asyncio.create_task(update_session_summary( conversation_id=request.conversation_id, old_summary=old_summary_json, new_messages=summary_messages, database_provider=request.database_provider, memory_provider=request.memory_provider, memory_model=request.memory_model, memory_api_key=request.memory_api_key, memory_base_url=request.memory_base_url, summary_provider=request.summary_provider, summary_model=request.summary_model, summary_api_key=request.summary_api_key, summary_base_url=request.summary_base_url, rebuild_from_scratch=False, # HITL resumption is usually incremental )) except Exception as exc: import traceback error_details = traceback.format_exc() logger.error(f"HITL continuation error: {exc}\n{error_details}") yield ErrorEvent(error=_extract_best_error_message(exc)).model_dump() def _collect_enabled_tool_names(self, request: StreamChatRequest) -> set[str]: names: list[str] = [] if request.provider != "gemini": for tool_id in request.tool_ids or []: names.append(resolve_tool_name(str(tool_id))) for tool_def in request.tools or []: if hasattr(tool_def, "model_dump"): tool_def = tool_def.model_dump() if not isinstance(tool_def, dict): continue name = tool_def.get("function", {}).get("name") or tool_def.get("name") if name: names.append(resolve_tool_name(str(name))) for user_tool in request.user_tools or []: if hasattr(user_tool, "name") and user_tool.name: names.append(str(user_tool.name)) return set(names) def _inject_local_time_context( self, messages: list[dict[str, Any]], request: StreamChatRequest, pre_events: list[dict[str, Any]], ) -> list[dict[str, Any]]: if not messages: return messages timezone = request.user_timezone or "UTC" locale = request.user_locale or "en-US" time_result = self._compute_local_time(timezone, locale) try: local_date = datetime.fromisoformat(str(time_result.get("iso"))).strftime("%Y-%m-%d") except Exception: local_date = str(time_result.get("formatted", "")).split(" ")[0] or datetime.now().strftime( "%Y-%m-%d" ) tz_label = str(time_result.get("timezone") or timezone) note = ( f"\n\n[Time note for this query]\n" f"Note: local date is {local_date} ({tz_label}). " "Interpret relative time terms using this date." ) updated: list[dict[str, Any]] = [] for msg in messages: if not isinstance(msg, dict) or msg.get("role") != "user": updated.append(msg) continue content = msg.get("content") if isinstance(content, str): if "[Time note for this query]" in content: updated.append(msg) else: updated.append({**msg, "content": f"{content}{note}"}) continue if isinstance(content, list): has_note = False for part in content: if isinstance(part, dict): if "[Time note for this query]" in str(part.get("text", "")) or "[Time note for this query]" in str(part.get("content", "")): has_note = True break if has_note: updated.append(msg) else: updated.append({**msg, "content": [*content, {"type": "text", "text": note}]}) continue updated.append(msg) return updated def _compute_local_time(self, timezone: str, locale: str) -> dict[str, Any]: try: tzinfo = ZoneInfo(timezone) now = datetime.now(tzinfo) except Exception: now = datetime.now() return { "timezone": timezone, "locale": locale, "formatted": now.strftime("%Y-%m-%d %H:%M:%S"), "iso": now.isoformat(), "now": now, } def _inject_tool_guidance( self, messages: list[dict[str, Any]], enabled_tools: set[str], request: Any | None = None, ) -> list[dict[str, Any]]: if not enabled_tools: return messages updated = list(messages) system_index = next((i for i, m in enumerate(updated) if m.get("role") == "system"), -1) no_tool_narration_guidance = ( "\n\n[OUTPUT DIRECTIVES]\n" "1. The main text (Answer) must contain ONLY the final helpful content and necessary explanations for the user.\n" "2. In the main text, NEVER describe that you are going to, are currently, or have already called any tools, searched, browsed, retrieved memory, or queried databases. These are internal traces (Trace).\n" "3. In the main text, do NOT refer to yourself performing actions (e.g., \"Let me check\", \"I will search\", \"I have retrieved\").\n" " Instead, directly present results as established information.\n" "4. If citing sources, use neutral phrasing such as: \"According to available data\", \"Based on public information\", \"According to the returned data\".\n" " Never mention tool names or the calling process.\n" "5. If information is insufficient, directly state the missing gap and ask clarifying questions.\n" " Do NOT say \"Let me check again\" or similar transitional action phrases.\n" "6. Once you start presenting the final answer, do not switch back to planning, searching, or tool-calling language.\n" "7. The final answer should begin naturally with the content itself, without meta commentary or transitional phrases.\n" ) updated = self._append_system_message(updated, no_tool_narration_guidance, system_index) system_index = next((i for i, m in enumerate(updated) if m.get("role") == "system"), -1) if "interactive_form" in enabled_tools: form_guidance = ( "\n[TOOL USE GUIDANCE]\n" "When you need to collect structured information from the user (e.g. preferences, requirements, " "booking details), use the 'interactive_form' tool.\n" "CRITICAL: DO NOT list questions in text or markdown. YOU MUST USE the 'interactive_form' tool to " "display fields.\n" "CRITICAL: If the user explicitly asks you to confirm something via a form, interactive form, or form tool, " "you MUST call 'interactive_form' in this response instead of asking in plain text.\n" "CRITICAL: If you need approval before installing a dependency, you MUST use 'interactive_form'. " "Do NOT ask yes/no approval questions in normal prose when the tool is available.\n" "CRITICAL: For approval forms, ask only for information the model does NOT already know. " "If skill_id or package_name are already known from the current tool result or failure context, " "do NOT include text inputs for them in the form.\n" "CRITICAL: For dependency installation approval, prefer a single required field such as " "'approve_install', and place the package name in the form title or description.\n" "Keep forms concise (3-6 fields).\n\n" "[SIMPLIFIED PAYLOAD]\n" "You may use a minimal payload to reduce tool-call size.\n" "- 'id' and 'title' are optional.\n" "- Each field may be minimal (e.g., {'name':'budget'}) or even a short string label.\n" "- Backend will auto-fill missing label/type defaults.\n\n" "[MANDATORY TEXT-FIRST RULE]\n" "CRITICAL: You MUST output meaningful introductory text BEFORE calling 'interactive_form'.\n" "- NEVER call 'interactive_form' as the very first thing in your response\n" "- ALWAYS explain the context, acknowledge the user's request, or provide guidance BEFORE the form\n" "- Minimum: Output at least 1-2 sentences before the form call\n" '- Example: "I can help you with that. To provide the best recommendation, please share some ' 'details below:"\n\n' "[SINGLE FORM PER RESPONSE]\n" "CRITICAL: You may call 'interactive_form' ONLY ONCE per response. Do NOT call it multiple times in " "the same answer.\n" "If you need to collect information, design ONE comprehensive form that gathers all necessary " "details at once.\n\n" "[MULTI-TURN INTERACTIONS]\n" "1. If the information from a submitted form is insufficient, you MAY present another " "'interactive_form' in your NEXT response (after the user submits the first form).\n" "2. LIMIT: Use at most 2-3 forms total across the entire conversation. Excessive questioning " "frustrates users.\n" "3. INTERLEAVING: You can place the form anywhere in your response. Output introductory text FIRST " "(e.g., \"I can help with that. Please provide some details below:\"), then call 'interactive_form' " "once.\n" "4. If the user has provided enough context through previous forms, proceed directly to the final " "answer without requesting more information." ) updated = self._append_system_message(updated, form_guidance, system_index) system_index = next((i for i, m in enumerate(updated) if m.get("role") == "system"), -1) search_tools_requiring_citations = { "Tavily_web_search", "Tavily_academic_search", "web_search_using_tavily", "web_search", "search_news", "search_exa", "search_arxiv_and_return_articles", "search_wikipedia", } if enabled_tools.intersection(search_tools_requiring_citations): citation_prompt = ( "\n\n[IMPORTANT] You have access to search tools. When you use them to answer a question, " "you MUST cite the search results in your answer using the format [1], [2], etc., " "corresponding to the index of the search result provided in the tool output. Do not fabricate " "citations." ) updated = self._append_system_message(updated, citation_prompt, system_index) if "local_time" in enabled_tools: local_time_guidance = ( "\n\n[TIME CONTEXT GUIDANCE]\n" "A local-date note is already appended to each user query before model execution.\n" "Do not call local_time again unless the user explicitly asks to refresh/recheck time." ) updated = self._append_system_message(updated, local_time_guidance, system_index) # [REINFORCED] System-level long-term memory guidance for the internal agent-memory skill if getattr(request, "enable_long_term_memory", False): memory_guidance = ( "\n\n[SYSTEM-LEVEL META-RULE: LONG-TERM MEMORY]\n" "You are equipped with a persistent memory layer ('agent-memory'). This is a core infrastructure skill that overrides situational constraints.\n" "1) MANDATORY PREREQUISITE: If a query involves personal history, preferences, or prior facts, searching memory is a mandatory PREREQUISITE. " "This MUST be executed BEFORE any other skill logic (e.g., Roleplay, Image Search) to avoid factual inconsistency.\n" "2) PERSISTENCE AGAINST OVERRIDES: This meta-rule remains in effect even if other skills (like roleplay) demand immediate responses. " "Retrieving ground-truth user context is the first step of all processing.\n" "3) USAGE: Call 'execute_skill_script' with skill_id='agent-memory'. Factual alignment via memory retrieval is non-negotiable." ) updated = self._append_system_message(updated, memory_guidance, system_index) return updated def _append_system_message( self, messages: list[dict[str, Any]], addition: str, system_index: int, ) -> list[dict[str, Any]]: updated = list(messages) if system_index != -1: updated[system_index] = { **updated[system_index], "content": f"{updated[system_index].get('content', '')}{addition}", } else: updated.insert(0, {"role": "system", "content": addition}) return updated def _normalize_tool_output(self, output: Any) -> Any: if hasattr(output, "model_dump"): try: return output.model_dump() except Exception: return str(output) if isinstance(output, dict): return output if isinstance(output, list): return [self._normalize_tool_output(item) for item in output] return output def _collect_search_sources(self, result: Any, sources_map: dict[str, Any]) -> None: def _extract_results(payload: Any) -> list[dict[str, Any]]: if isinstance(payload, list): return [item for item in payload if isinstance(item, dict)] if isinstance(payload, dict): for key in ("results", "items", "data", "sources", "articles", "news", "papers"): value = payload.get(key) if isinstance(value, list): return [item for item in value if isinstance(item, dict)] return [] results = _extract_results(result) if not results: return for item in results: url = ( item.get("url") or item.get("link") or item.get("uri") or item.get("source") or item.get("href") ) if not url or url in sources_map: continue title = ( item.get("title") or item.get("name") or item.get("headline") or item.get("paper_title") or "Unknown Source" ) snippet = ( item.get("content") or item.get("snippet") or item.get("summary") or item.get("abstract") or "" ) sources_map[url] = SourceEvent( uri=url, title=title, snippet=str(snippet)[:200], ).model_dump() async def _maybe_optimize_memories(self, agent: Agent, request: StreamChatRequest) -> None: return def _map_field_type_to_frontend(self, field_type: Any) -> str: """ Map Python/Agno field types to frontend form types. Args: field_type: Python type (class or string) Returns: Frontend form field type (text, number, checkbox, etc.) """ # Handle cases where field_type is a class/type instead of a string field_type_str = "" if isinstance(field_type, type): field_type_str = field_type.__name__ elif not isinstance(field_type, str): field_type_str = str(field_type) else: field_type_str = field_type type_mapping = { "str": "text", "int": "number", "float": "number", "bool": "checkbox", "date": "date", "time": "time", "datetime": "datetime", "list": "text", "dict": "textarea", } return type_mapping.get(field_type_str.lower(), "text") _stream_chat_service: StreamChatService | None = None def get_stream_chat_service() -> StreamChatService: global _stream_chat_service if _stream_chat_service is None: _stream_chat_service = StreamChatService() return _stream_chat_service