I3eka commented on code in PR #43133: URL: https://github.com/apache/superset/pull/43133#discussion_r4002581205
########## superset/ai/orchestrator.py: ########## @@ -0,0 +1,647 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Runs one assistant turn end to end. + +Sits between the HTTP layer and the runtime: loads the conversation, assembles +the prompt, resolves the tools the chosen profile allows, drives the runtime, +publishes every event to the bus, and records the outcome on the assistant +message. + +Deliberately independent of *where* it runs. The same function body serves the +inline path and the Celery path, which is what makes the execution mode a +configuration choice rather than two implementations that drift apart. +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid as uuid_module +from collections.abc import AsyncIterator, Iterator +from dataclasses import dataclass +from typing import Any + +from superset.ai.events import ( + cancelled_event, + done_event, + error_event, + GENERIC_ERROR_MESSAGE, + session_event, + StreamEvent, +) +from superset.ai.llm.base import Message +from superset.ai.telemetry import bind_run, current_run, start_run +from superset.ai.types import MessageRole, MessageStatus, RunOutcome, StreamEventType +from superset.utils.decorators import transaction + +logger = logging.getLogger(__name__) + +#: Cache key prefix for a run's cancellation flag. A flag rather than a signal +#: because a worker cannot be interrupted mid-call reliably; the runtime checks +#: this between steps. +_CANCEL_PREFIX = "ai-cancel-" + +#: How long a cancellation request stays meaningful. +_CANCEL_TTL_SECONDS = 900 + +#: Stored when a run is stopped before it produced any answer, so the +#: transcript still records that the turn happened. +_STOPPED_WITHOUT_ANSWER = "_Stopped before an answer was produced._" + +#: Stored when a run exhausted its time budget without saying anything. Phrased +#: as something the user can act on, because retrying is usually the right move. +_TIMED_OUT_WITHOUT_ANSWER = ( + "The assistant ran out of time before it could answer. Please try again." +) + +#: Ceiling on the page context recorded on a message. Well below the prompt's own +#: limit: this is stored per turn and read back with the whole transcript. +_RECORDED_CONTEXT_LIMIT = 4_000 + + +@dataclass +class TurnRequest: + """One unit of work: answer the latest message on a thread.""" + + thread_uuid: str + user_id: int + run_id: str + #: Assistant message row to fill in. Created before the run starts so a + #: client that reconnects has something to attach to. + assistant_message_uuid: str + profile_key: str | None = None + #: Concrete model to pin, overriding the profile's tier. + model: str | None = None + #: What the user had on screen when they asked. Supplied by the client, + #: which is the only party that knows which tab is open, what is typed in + #: the editor and which filters are applied. + page_context: dict[str, Any] | None = None + + def to_payload(self) -> dict[str, Any]: + """Serialise for the task broker.""" + return { + "thread_uuid": self.thread_uuid, + "user_id": self.user_id, + "run_id": self.run_id, + "assistant_message_uuid": self.assistant_message_uuid, + "profile_key": self.profile_key, + "model": self.model, + "page_context": self.page_context, + } + + @classmethod + def from_payload(cls, payload: dict[str, Any]) -> TurnRequest: + """Rebuild from a broker payload.""" + return cls(**payload) + + +def new_run_id() -> str: + """Identifier for one run, used as the event-stream key.""" + return str(uuid_module.uuid4()) + + +#: Runs cancelled in this process. +#: +#: Held alongside the cache rather than instead of it. Superset's default cache +#: is a null cache, which accepts a write and discards it — so a cache-only +#: implementation would leave cancellation silently broken on a default install, +#: with the button appearing to work and nothing stopping. This set makes inline +#: execution correct with no cache at all; the cache is what carries a +#: cancellation across processes for worker execution. +_CANCELLED_LOCALLY: set[str] = set() + + +def request_cancel(run_id: str) -> None: + """ + Ask a run to stop. + + Cooperative by design: the flag is recorded here and observed by the runtime + between steps. A run blocked inside a single long model call or query will + not notice until that call returns, which is a real limit worth documenting + rather than hiding. + """ + from superset.extensions import cache_manager + + _CANCELLED_LOCALLY.add(run_id) + try: + cache_manager.cache.set( + f"{_CANCEL_PREFIX}{run_id}", True, timeout=_CANCEL_TTL_SECONDS + ) + except Exception: # pylint: disable=broad-except + logger.warning("Could not record cancellation for AI run %s", run_id) + + +def is_cancelled(run_id: str) -> bool: + """Whether a stop has been requested for this run.""" + from superset.extensions import cache_manager + + if run_id in _CANCELLED_LOCALLY: + return True + try: + return bool(cache_manager.cache.get(f"{_CANCEL_PREFIX}{run_id}")) + except Exception: # pylint: disable=broad-except + # A cache that cannot be read must not make every run appear cancelled; + # that would stop all inference the moment the cache went away. + return False + + +def clear_cancel(run_id: str) -> None: + """Drop a run's cancellation flag.""" + from superset.extensions import cache_manager + + _CANCELLED_LOCALLY.discard(run_id) + try: + cache_manager.cache.delete(f"{_CANCEL_PREFIX}{run_id}") + except Exception: # pylint: disable=broad-except + logger.debug("Could not clear cancellation flag for AI run %s", run_id) + + +def stream_turn(request: TurnRequest) -> Iterator[StreamEvent]: + """ + Answer a turn, yielding events as they happen. + + This is the primary entry point. Inline execution consumes it directly from + inside the streaming response, which means the producer and the reader are + the same process by construction — important because Superset runs several + web workers, and a turn that published to one process's in-memory queue + while the browser's stream landed on another would appear to hang forever. + + Never raises for an operational failure: a failure is an ``error`` event and + an ``error`` message status, because the caller may already have flushed + response headers or may be a worker with no one to report to. + """ + recorder = start_run( + run_id=request.run_id, + thread_uuid=request.thread_uuid, + user_id=request.user_id, + ) + # Shared with ``_run`` so the ``finally`` below can see the runtime's + # partial result and whether the message was already written. + state: dict[str, Any] = {} + try: + # Bound here rather than inside ``_run`` so that a run which fails before + # it has resolved a profile still produces a start and an end, and so + # that the runtime can report its own spans without the runtime contract + # growing a telemetry parameter. + with bind_run(recorder): + recorder.run_started() + yield from _run(request, state) + except Exception as ex: # pylint: disable=broad-except + logger.exception("AI turn failed for run %s", request.run_id) + recorder.error(ex) + recorder.run_ended(outcome=RunOutcome.ERROR) + answer, extra = _partial_from_state(state) + extra["outcome"] = RunOutcome.ERROR.value + _finalise_message( + request.assistant_message_uuid, + # The generic text rather than the exception: this is persisted and + # served back to the browser, so it must not carry internals. The + # detail is in the log line above, keyed by run id. + content=answer or GENERIC_ERROR_MESSAGE, + status=MessageStatus.ERROR, + extra=extra, + ) + state["finalised"] = True + yield error_event() + yield done_event(ok=False) + finally: + clear_cancel(request.run_id) + # A client that stops the run, or simply navigates away, abandons this + # generator part-way through. Nothing above will have written the + # message, so it would otherwise sit in ``streaming`` with no content + # for ever — the user loses both the partial answer and any record that + # the turn happened. Persist whatever was produced. + _abandon_message(request.assistant_message_uuid, state) + # Idempotent, so the ordinary paths above win. + recorder.run_ended(outcome=RunOutcome.CANCELLED) + + +def execute_turn(request: TurnRequest) -> RunOutcome: + """ + Answer a turn, publishing events to the event bus. + + Used by worker execution, where the reader is in another process. Shares its + whole body with :func:`stream_turn` so the two execution modes cannot drift + apart in behaviour. + """ + from superset.ai.eventbus import get_event_bus + + bus = get_event_bus() + outcome = RunOutcome.SUCCESS + + for event in stream_turn(request): + bus.publish(request.run_id, event) + if event.type is StreamEventType.ERROR: + outcome = RunOutcome.ERROR + elif event.type is StreamEventType.CANCELLED: + outcome = RunOutcome.CANCELLED + elif event.type is StreamEventType.DONE and not event.payload.get("ok"): + # A run that ended un-ok without an explicit error frame timed out. + if outcome is RunOutcome.SUCCESS: + outcome = RunOutcome.TIMEOUT + + return outcome + + +def _run(request: TurnRequest, state: dict[str, Any]) -> Iterator[StreamEvent]: + """Assemble and drive the run. See :func:`stream_turn` for error policy.""" + from superset.ai.factories import ( + get_profiles, + get_provider, + get_runtime, + get_tools_for_profile, + ) + from superset.ai.policy import load_policy_chain + from superset.ai.runtime.base import RunRequest + from superset.daos.ai import AIChatMessageDAO, AIChatThreadDAO + + recorder = current_run() + + thread = AIChatThreadDAO.find_by_uuid_for_user(request.thread_uuid, request.user_id) + if thread is None: + # The thread vanished between accepting the message and running it. + recorder.run_ended(outcome=RunOutcome.ERROR) + yield error_event("That conversation is no longer available.") + yield done_event(ok=False) + return + + profile = get_profiles().get(request.profile_key) + tools = get_tools_for_profile(profile) + provider = get_provider() + runtime = get_runtime(provider) + state["runtime"] = runtime + + yield session_event(request.thread_uuid, request.assistant_message_uuid) + + from superset.ai.page_context import render_page_context + + history = _build_history(AIChatMessageDAO.find_for_thread(thread)) + # Recorded as well as prompted with, so the transcript can show what the + # assistant was told about the user's screen. An answer that looks wrong is + # usually an answer to a different question than the reader assumed, and the + # page context is where that difference lives. + rendered_context = render_page_context(request.page_context) + state["page_context"] = rendered_context + system_prompt = _build_system_prompt(tools, rendered_context) + model = _resolved_model(provider, request.model, profile) + + recorder.describe( + agent_key=profile.key, + model=model, + question=_latest_question(history), + ) + + run_request = RunRequest( + messages=history, + system_prompt=system_prompt, + tools=tools, + policies=load_policy_chain(), + model_alias=profile.model_alias, + max_turns=profile.max_turns or _config("AI_AGENT_MAX_TURNS", 20), + timeout_seconds=profile.timeout_seconds + or _config("AI_AGENT_TIMEOUT_SECONDS", 300), + should_cancel=lambda: is_cancelled(request.run_id), + ) + + _mark_streaming(request.assistant_message_uuid) + + # The runtime is async and this is a synchronous generator, so the async + # events are drained into a list per batch rather than bridged with a + # thread. Collecting the whole run before yielding would defeat streaming, + # so the loop pulls one event at a time from a dedicated event loop. + yield from _drain(runtime.run(run_request)) + + result = runtime.result + outcome = _outcome_of(result) + if result.error is not None: + # The only place the provider's own words are recorded. They do not go on + # the message: that is served back to the browser, and a transport error + # can name internal hosts. + logger.warning( + "AI run %s failed: %s", + request.run_id, + result.error, + ) + _finalise_message( + request.assistant_message_uuid, + content=_terminal_content(result, outcome), + status=_status_of(outcome), + extra={ + "outcome": outcome.value, + "agent_key": profile.key, + "model": model, + "tool_calls": result.tool_calls, + "turns": result.turns, + **_recorded_context(rendered_context), + }, + ) + + recorder.run_ended( + outcome=outcome, + turns=result.turns, + answer=result.answer, + ) + + state["finalised"] = True + + if outcome is RunOutcome.CANCELLED: + yield cancelled_event() + yield done_event(ok=outcome is RunOutcome.SUCCESS) + + +def _drain(source: AsyncIterator[StreamEvent]) -> Iterator[StreamEvent]: + """ + Pull an async iterator one item at a time from a synchronous caller. + + A single event loop is kept for the whole run and stepped with + ``__anext__``, so each event reaches the client as it is produced rather + than after the run completes. + """ + loop = asyncio.new_event_loop() + try: + iterator = source.__aiter__() + while True: + try: + yield loop.run_until_complete(iterator.__anext__()) + except StopAsyncIteration: + return + finally: Review Comment: The `_drain` cleanup still closes the loop without explicitly awaiting the iterator's `aclose()` in this branch. This is a base provider-resource-lifecycle fix, not addressed by the current refresh. Leaving the disconnect/cleanup regression requirement open. ########## superset-frontend/src/features/ai/hooks/useChatBot.ts: ########## @@ -0,0 +1,1323 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +/** + * @fileoverview Conversation state and the send loop. + * + * Runs are tracked per conversation, not globally. That is the point of the + * structure: a user can start something slow in one conversation, switch to + * another and keep working, and come back to find the first still going. A single + * `isLoading` flag would have made switching away cancel or corrupt the run. + * + * The server owns the transcript. A finished run is re-read from it rather than + * assembled from the frames, so the tool calls persisted on the message are what + * the user sees, and what they see survives a reload. + */ + +import { useCallback, useEffect, useRef, useState } from 'react'; +import type { TextAreaRef } from 'antd/es/input/TextArea'; +import { logging } from '@apache-superset/core/utils'; +import { t } from '@apache-superset/core/translation'; +import { + type AiAgent, + type AiToolCall, + type ChatMessageWithMeta, + type ChatTab, + type CheckpointPayload, +} from '../types'; +import { + AGENT_STORAGE_KEY, + ChatRequestAbortedError, + ChatStreamEventError, + ChatStreamTimeoutError, + DEFAULT_AGENT_KEY, + DEFAULT_CHAT_AGENT, + cancelChatRun, + describeRequestError, + fetchAgents, + fetchSuggestedPrompts, + loadStoredAgentKey, + normalizeChatAgents, + startRun, + streamRun, + submitFeedback, +} from './chatRequest'; +import { + NEW_CHAT_NAME, + createThread, + deleteThread as deleteThreadApi, + getThread, + listThreads, + threadToTab, + updateThread, +} from './chatThreadsApi'; +import { buildQuickPrompts } from './quickPrompts'; +import { + buildPageContextPayload, + usePageContext, + type PageContext, +} from './usePageContext'; + +/** Cache of the conversation list, so the menu renders before the list arrives. */ +export const CHAT_TABS_STORAGE_KEY = 'superset-chat-tabs'; + +/** Which conversation was last open. */ +export const ACTIVE_TAB_STORAGE_KEY = 'superset-chat-active-tab'; + +/** Recent inputs, recalled with the arrow keys. */ +export const HISTORY_STORAGE_KEY = 'superset-chat-history'; + +export { AGENT_STORAGE_KEY } from './chatRequest'; + +/** How many inputs the arrow-key history keeps. */ +const MAX_INPUT_HISTORY = 50; + +/** A conversation title derived from a message is clipped to this. */ +const MAX_TAB_NAME_LENGTH = 30; + +export type ChatRunStatus = 'running' | 'cancelling'; + +/** Shared empty list, so a render with no steps yet keeps a stable identity. */ +const EMPTY_TOOL_CALLS: AiToolCall[] = []; + +interface ActiveChatRun { + requestId: string; + tabId: string; + threadId: string; + runId?: string; + controller: AbortController; + isStreaming: boolean; + liveThoughts: string; + liveToolLog: string; + /** + * Steps taken so far, as structured records rather than log lines. + * + * Carried alongside `liveToolLog` so a run in flight can be rendered the same + * way a finished one is — expandable per step, with the SQL and the rows it + * returned — instead of as a wall of text that only becomes legible once the + * transcript is re-read from the server. + */ + liveToolCalls: AiToolCall[]; + /** The page context this run was given, so the live view can show it too. */ + livePageContext?: string; + /** + * The answer so far, as the model produces it. + * + * Rendered directly: the deltas used to be folded into `liveThinking`, which + * nothing displayed, so an answer appeared in one piece the moment the run + * ended however long it had taken to generate. + */ + liveAnswer: string; + liveThinking: string; + status: ChatRunStatus; + startedAt: number; + checkpoint: CheckpointPayload | null; +} + +/** + * An identifier for a turn. + * + * Drawn from `crypto`, not `Math.random`. These become the idempotency key on a + * turn and the handle used to cancel one, so a value another session could guess + * is a correctness and a security problem rather than merely a collision risk. + */ +const generateId = (): string => { + if (typeof crypto.randomUUID === 'function') { + return crypto.randomUUID(); + } + // Older engines expose the entropy source without the convenience wrapper. + const bytes = new Uint8Array(16); + crypto.getRandomValues(bytes); + return Array.from(bytes, byte => byte.toString(16).padStart(2, '0')).join(''); +}; + +const createNewTab = (name: string = NEW_CHAT_NAME): ChatTab => ({ + id: generateId(), + name, + messages: [], + createdAt: Date.now(), +}); + +const truncateTabName = ( + name: string, + maxLength: number = MAX_TAB_NAME_LENGTH, +): string => + name.length <= maxLength ? name : `${name.substring(0, maxLength)}...`; + +const readJson = <T>(key: string, fallback: T): T => { + try { + const stored = localStorage.getItem(key); + return stored ? (JSON.parse(stored) as T) : fallback; + } catch (caught) { + logging.warn(`[ai] could not read ${key}`, caught); + return fallback; + } +}; + +const writeJson = (key: string, value: unknown): void => { + try { + localStorage.setItem(key, JSON.stringify(value)); + } catch (caught) { + logging.warn(`[ai] could not write ${key}`, caught); + } +}; + +/** + * Reconciles the server's transcript with what is already on screen. + * + * The server's copy is authoritative — it carries the tool calls — but it is not + * necessarily complete the moment a run ends, and replacing outright would then + * erase an answer the user has just read. So anything local that the server has + * not accounted for is kept, matched by identity first and by role and content + * second, which is how a locally-appended turn is recognised once the server + * returns its own copy of it under a real uuid. + */ +export const mergeMessages = ( + fromServer: ChatMessageWithMeta[], + local: ChatMessageWithMeta[], +): ChatMessageWithMeta[] => { + const serverIds = new Set(fromServer.map(message => message.id)); + const serverTurns = new Set( + fromServer.map(message => `${message.role}:${message.content}`), + ); + const unaccounted = local.filter( + message => + !serverIds.has(message.id) && + !serverTurns.has(`${message.role}:${message.content}`), + ); + return [...fromServer, ...unaccounted]; +}; + +/** + * The `page_context` body for one turn. + * + * Returns undefined when there is nothing to send, so an omitted field is + * distinguishable from an empty one. + */ +export const buildRequestPageContext = ( + context: PageContext | undefined, + directive?: string, +): Record<string, unknown> | undefined => { + const payload = context ? buildPageContextPayload(context) : undefined; + if (!directive) { + return payload; + } + const existing = payload?.helper_directives; + return { + ...payload, + helper_directives: [ + directive, + ...(Array.isArray(existing) ? existing : []), + ], + }; +}; + +export interface UseChatBotReturn { + // Conversations + chatTabs: ChatTab[]; + activeTabId: string; + activeTab: ChatTab | undefined; + threadsLoaded: boolean; + handleNewChat: () => Promise<string>; + handleSelectTab: (tabId: string) => Promise<void>; + handleDeleteTab: (tabId: string) => Promise<void>; + handleRenameTab: (tabId: string, newName: string) => void; + // Messages of the active conversation + messages: ChatMessageWithMeta[]; + // Input + inputValue: string; + setInputValue: (value: string) => void; + handleKeyDown: (event: React.KeyboardEvent) => void; + inputRef: React.RefObject<TextAreaRef>; + messagesEndRef: React.RefObject<HTMLDivElement>; + // The run in flight, if any, for the active conversation + isLoading: boolean; + isStreamingResponse: boolean; + liveThoughts: string; + liveToolLog: string; + /** Steps taken so far in the run in flight, for the structured live view. */ + liveToolCalls: AiToolCall[]; + /** The page context the run in flight was given. */ + livePageContext?: string; + /** The answer so far for the run in flight. */ + liveAnswer: string; + checkpoint: CheckpointPayload | null; + activeRunStatus: ChatRunStatus | null; + error?: string; + // Actions + sendMessage: ( + messageOverride?: string, + systemPromptOverride?: string, + ) => Promise<void>; + handleCancelRun: () => Promise<void>; + handleCheckpointContinue: () => void; + handleFeedback: (messageId: string, feedback: 'like' | 'dislike') => void; + messageFeedback: Record<string, 'like' | 'dislike'>; + // Suggestions + /** The message whose run just ended; its thought process stays open. */ + justCompletedId?: string; + quickPrompts: string[]; + loadQuickPrompts: () => void; + applyQuickPrompt: (prompt: string) => Promise<void>; + // Agent profiles + agents: AiAgent[]; + selectedAgent: string; + setSelectedAgent: (key: string) => void; + // Page context + pageContext: PageContext; + includePageContext: boolean; + toggleIncludePageContext: () => void; +} + +export const useChatBot = (): UseChatBotReturn => { + const [chatTabs, setChatTabs] = useState<ChatTab[]>(() => + readJson<ChatTab[]>(CHAT_TABS_STORAGE_KEY, []).map(tab => ({ + // The cache is a placeholder for the menu; message bodies are re-read from + // the server so a stale cache cannot show a conversation that has moved on. + ...tab, + messages: [], + })), + ); + const [activeTabId, setActiveTabId] = useState<string>(() => { + try { + return localStorage.getItem(ACTIVE_TAB_STORAGE_KEY) ?? ''; + } catch { + return ''; + } + }); + const [threadsLoaded, setThreadsLoaded] = useState(false); + const [error, setError] = useState<string | undefined>(undefined); + + const [inputValue, setInputValue] = useState(''); + const [activeRunsByTab, setActiveRunsByTab] = useState< + Record<string, ActiveChatRun> + >({}); + const [quickPrompts, setQuickPrompts] = useState<string[]>([]); + const [messageFeedback, setMessageFeedback] = useState< + Record<string, 'like' | 'dislike'> + >({}); + const [includePageContext, setIncludePageContext] = useState(true); + /** + * The assistant message whose run has only just ended. + * + * Its thought process stays open, because collapsing it the instant the answer + * lands moves everything below it — the answer the user is mid-sentence through + * jumps up the panel. Older messages start closed. + */ + const [justCompletedId, setJustCompletedId] = useState<string | undefined>(); + const [agents, setAgents] = useState<AiAgent[]>([DEFAULT_CHAT_AGENT]); + const [selectedAgent, setSelectedAgent] = useState<string>(() => + loadStoredAgentKey(AGENT_STORAGE_KEY), + ); + + const [messageHistory, setMessageHistory] = useState<string[]>(() => + readJson<string[]>(HISTORY_STORAGE_KEY, []), + ); + const [historyIndex, setHistoryIndex] = useState(-1); + const [currentDraft, setCurrentDraft] = useState(''); + + const messagesEndRef = useRef<HTMLDivElement>(null); + const inputRef = useRef<TextAreaRef>(null); + + /** + * The run map and the conversation list are also held in refs, and the refs are + * the authority. + * + * The send loop has to ask "is this still my run?" between awaits, and it cannot + * ask React: a run that starts and fails inside one batch never causes a render, + * so a ref synced at render time would still be empty and the loop would discard + * its own result as stale. Writing the ref at the point of mutation removes that + * window. The callbacks read the refs rather than the state so their identities + * do not churn on every streamed frame, which would restart effects mid-run. + */ + const activeRunsByTabRef = useRef<Record<string, ActiveChatRun>>({}); + const chatTabsRef = useRef<ChatTab[]>(chatTabs); + const activeTabIdRef = useRef(activeTabId); + activeTabIdRef.current = activeTabId; + + const updateRuns = useCallback( + ( + updater: ( + previous: Record<string, ActiveChatRun>, + ) => Record<string, ActiveChatRun>, + ) => { + activeRunsByTabRef.current = updater(activeRunsByTabRef.current); + setActiveRunsByTab(activeRunsByTabRef.current); + }, + [], + ); + + const updateTabs = useCallback( + (updater: (previous: ChatTab[]) => ChatTab[]) => { + chatTabsRef.current = updater(chatTabsRef.current); + setChatTabs(chatTabsRef.current); + }, + [], + ); + + /** Resolved when the user answers a checkpoint; see `streamRun`. */ + const checkpointGateRef = useRef<{ resolve: () => void } | null>(null); + const mountedRef = useRef(true); + useEffect( + () => () => { + mountedRef.current = false; + }, + [], + ); + + const activeTab = chatTabs.find(tab => tab.id === activeTabId); + const messages = activeTab?.messages ?? []; + + const activeRun = activeRunsByTab[activeTabId]; + const isLoading = Boolean(activeRun); + const isStreamingResponse = activeRun?.isStreaming ?? false; + const liveThoughts = activeRun?.liveThoughts ?? ''; + const liveToolLog = activeRun?.liveToolLog ?? ''; + const liveToolCalls = activeRun?.liveToolCalls ?? EMPTY_TOOL_CALLS; + const livePageContext = activeRun?.livePageContext; + const liveAnswer = activeRun?.liveAnswer ?? ''; + const checkpoint = activeRun?.checkpoint ?? null; + const activeRunStatus = activeRun?.status ?? null; + + const pageContext = usePageContext(); + const pageContextRef = useRef(pageContext); + pageContextRef.current = pageContext; + + const fail = useCallback(async (caught: unknown, fallback: string) => { + const message = await describeRequestError(caught, fallback); + logging.error('[ai] assistant request failed', caught); + if (mountedRef.current) { + setError(message); + } + }, []); + + // ----------------------------------------------------------------------- + // Persistence of the small things + // ----------------------------------------------------------------------- + + useEffect(() => { + // Only the shell of each conversation is cached; see the initialiser. + writeJson( + CHAT_TABS_STORAGE_KEY, + chatTabs.map(tab => ({ ...tab, messages: [] })), + ); + }, [chatTabs]); + + useEffect(() => { + try { + localStorage.setItem(ACTIVE_TAB_STORAGE_KEY, activeTabId); + } catch (caught) { + logging.warn('[ai] could not remember the active conversation', caught); + } + }, [activeTabId]); + + useEffect(() => { + if (messageHistory.length > 0) { + writeJson(HISTORY_STORAGE_KEY, messageHistory); + } + }, [messageHistory]); + + useEffect(() => { + try { + localStorage.setItem(AGENT_STORAGE_KEY, selectedAgent); + } catch (caught) { + logging.warn('[ai] could not remember the selected agent', caught); + } + }, [selectedAgent]); + + // Follows the transcript as it grows, including while a run streams. Guarded + // because `scrollIntoView` is absent in environments without a layout engine, + // and failing to scroll must not take the panel down. + useEffect(() => { + messagesEndRef.current?.scrollIntoView?.({ behavior: 'smooth' }); + }, [messages, liveToolLog, liveThoughts]); + + // ----------------------------------------------------------------------- + // Conversation management + // ----------------------------------------------------------------------- + + const setMessagesForTab = useCallback( + ( + tabId: string, + updater: (previous: ChatMessageWithMeta[]) => ChatMessageWithMeta[], + ) => { + updateTabs(previous => + previous.map(tab => + tab.id === tabId ? { ...tab, messages: updater(tab.messages) } : tab, + ), + ); + }, + [updateTabs], + ); + + const refreshThreadMessages = useCallback( + async (threadId: string) => { + const { thread, messages: threadMessages } = await getThread(threadId); + if (!mountedRef.current) { + return; + } + const refreshed = threadToTab(thread, threadMessages); + // A locally set title wins: the user may have renamed the conversation + // while the request was in flight. + updateTabs(previous => + previous.map(tab => + tab.threadId === threadId + ? { + ...refreshed, + name: tab.name || refreshed.name, + messages: mergeMessages(refreshed.messages, tab.messages), + } + : tab, + ), + ); + }, + [updateTabs], + ); + + const handleNewChat = useCallback(async (): Promise<string> => { + try { + const thread = await createThread( + undefined, + selectedAgent === DEFAULT_AGENT_KEY ? undefined : selectedAgent, + ); + const tab = threadToTab(thread); + updateTabs(previous => [tab, ...previous]); + setActiveTabId(tab.id); + activeTabIdRef.current = tab.id; + setError(undefined); + return tab.id; + } catch (caught) { + await fail(caught, t('The conversation could not be created.')); + // A local tab still lets the user type; the thread is created on send. + const tab = createNewTab(); + updateTabs(previous => [tab, ...previous]); + setActiveTabId(tab.id); + activeTabIdRef.current = tab.id; + return tab.id; + } + }, [fail, selectedAgent, updateTabs]); + + const handleSelectTab = useCallback( + async (tabId: string) => { + setActiveTabId(tabId); + const tab = chatTabsRef.current.find(candidate => candidate.id === tabId); + // Messages are fetched on first view, not up front: a user with fifty + // conversations should not pay for forty-nine of them. + if (tab?.threadId && tab.messages.length === 0) { + try { + await refreshThreadMessages(tab.threadId); + } catch (caught) { + await fail(caught, t('The conversation could not be loaded.')); + } + } + }, + [fail, refreshThreadMessages], + ); + + const handleDeleteTab = useCallback( + async (tabId: string) => { + const tab = chatTabsRef.current.find(candidate => candidate.id === tabId); + if (tab?.threadId) { + try { + await deleteThreadApi(tab.threadId); Review Comment: Deletion versus an active mutating run remains unresolved; a deleted transcript is not a cancellation acknowledgement. I have not added a delete-and-forget workaround. The base needs a terminal run/deletion contract before this authoring PR can consider the issue closed. ########## superset/ai/eventbus.py: ########## @@ -0,0 +1,314 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Carries streamed events from whatever produced them to the HTTP response. + +Two implementations, matching the two execution modes. Inline execution needs +nothing more than an in-process queue. Worker execution needs a shared, +*replayable* channel — replayable because a browser that loses its connection +must be able to rejoin a run already in progress, which rules out +publish/subscribe: a subscriber that was absent when an event was published +never sees it. + +The Redis implementation therefore uses streams, and reuses the cache backend +that Superset's async-query channel already configures rather than introducing +a second Redis client to operate. +""" + +from __future__ import annotations + +import logging +import queue +from abc import ABC, abstractmethod +from collections.abc import Iterator +from typing import Any + +from superset.ai.events import StreamEvent +from superset.ai.types import StreamEventType +from superset.utils import json + +logger = logging.getLogger(__name__) + +#: Yielded by :meth:`BaseEventBus.consume` when nothing arrived within the poll +#: interval, so a caller can emit a keep-alive rather than block indefinitely. +IDLE = None + +#: Terminal event types. Seeing one ends consumption, so a reader does not hang +#: waiting for a producer that has already finished. +_TERMINAL = frozenset( + {StreamEventType.DONE, StreamEventType.ERROR, StreamEventType.CANCELLED} +) + + +class BaseEventBus(ABC): + """A per-run channel of events.""" + + @abstractmethod + def publish(self, run_id: str, event: StreamEvent) -> None: + """Append an event to a run's channel.""" + + @abstractmethod + def consume( + self, + run_id: str, + timeout_seconds: float, + poll_seconds: float = 1.0, + ) -> Iterator[StreamEvent | None]: + """ + Yield a run's events until a terminal one arrives or time runs out. + + Yields :data:`IDLE` when a poll interval passes with nothing new, which + is the caller's cue to send a keep-alive frame. + """ + + @abstractmethod + def close(self, run_id: str) -> None: + """Release any resources held for a run.""" + + +class MemoryEventBus(BaseEventBus): + """ + An in-process queue per run. + + Correct only when the producer and the streaming request share a process. + Selecting this alongside worker execution would leave every stream silent, + which :func:`get_event_bus` refuses to allow. + """ + + def __init__(self) -> None: + self._queues: dict[str, queue.SimpleQueue[StreamEvent]] = {} + + def _queue_for(self, run_id: str) -> queue.SimpleQueue[StreamEvent]: + return self._queues.setdefault(run_id, queue.SimpleQueue()) + + def publish(self, run_id: str, event: StreamEvent) -> None: + self._queue_for(run_id).put(event) + + def consume( + self, + run_id: str, + timeout_seconds: float, + poll_seconds: float = 1.0, + ) -> Iterator[StreamEvent | None]: + import time + + # Deliberately not ``_queue_for``: reading must not create a channel. + # This bus lives for the life of the process, so a client polling + # unknown run identifiers would otherwise grow the dict without bound. + channel = self._queues.get(run_id) + deadline = time.monotonic() + timeout_seconds + + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return + if channel is None: + # The producer may not have published yet; look again rather + # than deciding the run does not exist. Only report idle if it + # is still absent, so a channel that appeared during the wait + # is drained on this pass instead of costing an extra tick. + channel = self._queues.get(run_id) + if channel is None: + yield IDLE + time.sleep(min(poll_seconds, remaining)) + continue + try: + # Bounded by whichever is sooner, so a generous poll interval + # cannot overshoot the caller's deadline. + event = channel.get(timeout=min(poll_seconds, remaining)) + except queue.Empty: + yield IDLE + continue + yield event + if event.type in _TERMINAL: + return + + def close(self, run_id: str) -> None: + self._queues.pop(run_id, None) + + +class RedisStreamEventBus(BaseEventBus): + """ + A Redis stream per run. + + Replayable by construction: a reconnecting reader starts from the beginning + of the stream and catches up, which is what makes worker execution usable + from a browser on a flaky connection. + """ + + def __init__( + self, + cache: Any, + prefix: str = "ai-events-", + ttl_seconds: int = 900, + ) -> None: + self._cache = cache + self._prefix = prefix + self._ttl = ttl_seconds + + def _stream(self, run_id: str) -> str: + return f"{self._prefix}{run_id}" + + def publish(self, run_id: str, event: StreamEvent) -> None: + payload = { + "data": json.dumps({"type": event.type.value, "payload": event.payload}) + } + # A failure to publish must not kill the run that is producing useful + # work; the reader will time out and the answer is still persisted. + try: + self._cache.xadd(self._stream(run_id), payload, "*", 10_000) + except Exception: # pylint: disable=broad-except + logger.warning("Could not publish AI event for run %s", run_id) + + def consume( + self, + run_id: str, + timeout_seconds: float, + poll_seconds: float = 1.0, + ) -> Iterator[StreamEvent | None]: + import time + + stream = self._stream(run_id) + deadline = time.monotonic() + timeout_seconds + last_id = "-" + + while time.monotonic() < deadline: + try: + entries = self._cache.xrange(stream, last_id, "+", 100) + except Exception: # pylint: disable=broad-except + logger.warning("Could not read AI events for run %s", run_id) + yield IDLE + time.sleep(poll_seconds) + continue + + fresh = [entry for entry in entries if _entry_id(entry) != last_id] + if not fresh: + yield IDLE + time.sleep(poll_seconds) + continue + + for entry in fresh: + last_id = _entry_id(entry) + event = _decode(entry) + if event is None: + continue + yield event + if event.type in _TERMINAL: Review Comment: The terminal-frame mismatch remains in this branch: consuming `error`/`cancelled` as the stopping frame is not the promised final `done` sequence. This refresh does not change the producer/consumer protocol. Leaving it open for one base-owned contract fix and worker-mode regression. ########## superset-frontend/src/features/ai/components/ChatChartEmbed.tsx: ########## @@ -0,0 +1,542 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +/** + * @fileoverview A chart rendered inside a chat message. + * + * The assistant does not send a chart, it sends a `form_data_key` it stored — so + * what arrives in the transcript is a reference the client resolves, and the + * rendered chart is the real thing, with the real permissions, rather than an + * image of one. + * + * The awkward part is timing: the key can exist before the query behind it has + * finished. Rather than show the chart's own "No data" state (which reads as a + * broken answer) the component keeps a spinner up and re-renders on a growing + * backoff until rows appear, giving up after a bounded number of attempts. + */ + +import { useCallback, useEffect, useRef, useState } from 'react'; +import { + type QueryFormData, + StatefulChart, + SupersetClient, +} from '@superset-ui/core'; +import { styled } from '@apache-superset/core/theme'; +import { t } from '@apache-superset/core/translation'; +import { Loading } from '@superset-ui/core/components'; +import { Icons } from '@superset-ui/core/components/Icons'; +import { ErrorBoundary } from 'src/components/ErrorBoundary'; + +const VALID_KEY_PATTERN = /^[a-zA-Z0-9_-]+$/; +const MIN_HEIGHT = 100; +const MAX_HEIGHT = 800; +const DEFAULT_HEIGHT = 300; +const FETCH_TIMEOUT_MS = 30_000; +const MAX_RETRIES = 3; +const RETRY_DELAYS_MS = [500, 1500, 3000]; + +/** Width used until the container has been measured. */ +const FALLBACK_CHART_WIDTH = 600; + +// Backoff for the "waiting for chart data" poll. The delay grows exponentially +// per attempt up to a ceiling, so a slow query is waited out without hammering +// the backend. +const POLL_BASE_DELAY_MS = 1000; +const POLL_MAX_DELAY_MS = 30_000; + +/** + * Polling stops after this many attempts. + * + * A retry re-issues the chart's data request, which will use the results cache if + * the query has landed but will otherwise execute it. Polling forever would keep + * re-issuing it, so an unfinished query surfaces the retry control instead. + */ +const MAX_POLL_ATTEMPTS = 6; + +const getPollDelayMs = (attempt: number): number => + Math.min(POLL_BASE_DELAY_MS * 2 ** attempt, POLL_MAX_DELAY_MS); + +export interface ChartEmbedParams { + formDataKey: string | null; + height: number; + title: string | null; +} + +/** + * Parse key=value lines from the content of a ```superset-chart fenced block. + * + * Rules are strict on purpose: the block is model output, so `form_data_key` must + * match `/^[a-zA-Z0-9_-]+$/` before it reaches a URL, and `height` is clamped. + * Unknown keys are ignored so a newer backend can add some without breaking an + * older client. + */ +export function parseChartEmbedParams(codeText: string): ChartEmbedParams { + const result: ChartEmbedParams = { + formDataKey: null, + height: DEFAULT_HEIGHT, + title: null, + }; + + const lines = codeText + .split('\n') + .map(line => line.trim()) + .filter(Boolean); + + lines.forEach(line => { + const eqIndex = line.indexOf('='); + if (eqIndex <= 0) { + return; + } + + const key = line.slice(0, eqIndex).trim().toLowerCase(); + const value = line.slice(eqIndex + 1).trim(); + + if (key === 'form_data_key') { + if (value && VALID_KEY_PATTERN.test(value)) { + result.formDataKey = value; + } + return; + } + if (key === 'height') { + const parsed = parseInt(value, 10); + if (!Number.isNaN(parsed)) { + result.height = Math.max(MIN_HEIGHT, Math.min(MAX_HEIGHT, parsed)); + } + return; + } + if (key === 'title' && value) { + result.title = value; + } + }); + + return result; +} + +const ChartContainer = styled.div` + border: 1px solid ${({ theme }) => theme.colorBorderSecondary}; + border-radius: ${({ theme }) => theme.borderRadius}px; + overflow: hidden; + margin: ${({ theme }) => theme.sizeUnit * 2}px 0; + background: ${({ theme }) => theme.colorBgContainer}; +`; + +const ChartHeader = styled.div` + display: flex; + justify-content: space-between; + align-items: center; + padding: ${({ theme }) => theme.sizeUnit * 2}px + ${({ theme }) => theme.sizeUnit * 3}px; + border-bottom: 1px solid ${({ theme }) => theme.colorBorderSecondary}; + background: ${({ theme }) => theme.colorBgLayout}; + font-size: ${({ theme }) => theme.fontSizeSM}px; +`; + +const ChartTitle = styled.span` + font-weight: ${({ theme }) => theme.fontWeightStrong}; + color: ${({ theme }) => theme.colorText}; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + flex: 1; + min-width: 0; +`; + +const ChartActions = styled.div` + display: flex; + gap: ${({ theme }) => theme.sizeUnit * 2}px; + align-items: center; + flex-shrink: 0; + margin-left: ${({ theme }) => theme.sizeUnit * 2}px; +`; + +const ActionLink = styled.a` + display: inline-flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit / 2}px; + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorPrimary}; + cursor: pointer; + text-decoration: none; + + &:hover { + text-decoration: underline; + } +`; + +const ActionButton = styled.button` + display: inline-flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit / 2}px; + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorTextSecondary}; + cursor: pointer; + background: none; + border: none; + padding: 2px ${({ theme }) => theme.sizeUnit / 2}px; + border-radius: ${({ theme }) => theme.borderRadius}px; + + &:hover { + color: ${({ theme }) => theme.colorPrimary}; + background: ${({ theme }) => theme.colorFillTertiary}; + } +`; + +const ChartBody = styled.div<{ height: number }>` + height: ${({ height }) => height}px; + position: relative; +`; + +// Covers the chart while it reports no data, hiding the underlying "No data" +// state (which looks broken) behind a spinner while refreshing continues. +const ChartDataOverlay = styled.div` + position: absolute; + inset: 0; + display: flex; + flex-direction: column; + align-items: center; + justify-content: center; + gap: ${({ theme }) => theme.sizeUnit * 3}px; + background: ${({ theme }) => theme.colorBgContainer}; + color: ${({ theme }) => theme.colorTextSecondary}; + font-size: ${({ theme }) => theme.fontSizeSM}px; + z-index: 2; +`; + +const CenteredMessage = styled.div<{ height: number }>` + display: flex; + flex-direction: column; + align-items: center; + justify-content: center; + height: ${({ height }) => height}px; + color: ${({ theme }) => theme.colorTextSecondary}; + font-size: ${({ theme }) => theme.fontSizeSM}px; + text-align: center; + padding: ${({ theme }) => theme.sizeUnit * 4}px; + gap: ${({ theme }) => theme.sizeUnit * 2}px; +`; + +interface ChatChartEmbedProps { + formDataKey: string; + height?: number; + title?: string; +} + +type FetchState = + | { status: 'loading' } + | { status: 'loaded'; formData: QueryFormData } + | { status: 'error'; message: string }; + +const exploreUrlFor = (formDataKey: string): string => + `/explore/?form_data_key=${encodeURIComponent(formDataKey)}`; + +export function ChatChartEmbedInner({ + formDataKey, + height = DEFAULT_HEIGHT, + title, +}: ChatChartEmbedProps) { + const [fetchState, setFetchState] = useState<FetchState>({ + status: 'loading', + }); + const [chartWidth, setChartWidth] = useState(0); + const [chartRenderKey, setChartRenderKey] = useState(0); + // True while the chart's query results are still unavailable (either an empty + // result or a cache miss), which is what keeps the overlay up. + const [isAwaitingData, setIsAwaitingData] = useState(false); + const chartBodyRef = useRef<HTMLDivElement>(null); + const pollTimeoutRef = useRef<ReturnType<typeof setTimeout>>(); + + // Bumping `chartRenderKey` remounts the chart, which re-issues its data + // request. `force` is left off so the results cache is preferred. + const scheduleNextPoll = useCallback(() => { + if (chartRenderKey >= MAX_POLL_ATTEMPTS) { + setIsAwaitingData(false); + return; + } + setIsAwaitingData(true); + const delay = getPollDelayMs(chartRenderKey); + if (pollTimeoutRef.current) { + clearTimeout(pollTimeoutRef.current); + } + pollTimeoutRef.current = setTimeout( + () => setChartRenderKey(key => key + 1), + delay, + ); + }, [chartRenderKey]); + + useEffect( + () => () => { + if (pollTimeoutRef.current) { + clearTimeout(pollTimeoutRef.current); + } + }, + [], + ); + + useEffect(() => { + const element = chartBodyRef.current; + if (!element) { + return undefined; + } + + const initialWidth = Math.floor(element.getBoundingClientRect().width); + if (initialWidth > 0) { + setChartWidth(initialWidth); + } + + // The panel is resizable by the host, so the chart is measured rather than + // given a fixed width. + const observer = new ResizeObserver(entries => { + const [entry] = entries; + if (entry) { + const width = Math.floor(entry.contentRect.width); + if (width > 0) { + setChartWidth(width); + } + } + }); + observer.observe(element); + return () => observer.disconnect(); + }, [fetchState.status]); + + const exploreUrl = exploreUrlFor(formDataKey); + + const fetchFormData = useCallback( + async (attempt: number = 0) => { + setFetchState({ status: 'loading' }); + setIsAwaitingData(false); + setChartRenderKey(0); + + const retry = (): boolean => { + if (attempt >= MAX_RETRIES) { + return false; + } + const delay = RETRY_DELAYS_MS[attempt] ?? 1000; + setTimeout(() => { + fetchFormData(attempt + 1); + }, delay); + return true; + }; + + try { + const controller = new AbortController(); + const timeoutId = setTimeout( + () => controller.abort(), + FETCH_TIMEOUT_MS, + ); + + const response = await SupersetClient.get({ + endpoint: `/api/v1/explore/form_data/${encodeURIComponent(formDataKey)}`, + signal: controller.signal, + }); + + clearTimeout(timeoutId); + + const raw = response.json; + const formDataStr = + typeof raw === 'string' + ? raw + : ((raw as { form_data?: string })?.form_data ?? undefined); + + if (!formDataStr) { + // The key can be written before the payload is readable, so an empty + // body is treated as "not yet" rather than "gone". + if (retry()) { + return; + } + setFetchState({ + status: 'error', + message: t('Chart configuration not found or expired.'), + }); + return; + } + + const parsed: QueryFormData = + typeof formDataStr === 'string' + ? JSON.parse(formDataStr) + : formDataStr; + + if (!parsed.viz_type) { + setFetchState({ + status: 'error', + message: t( + 'Invalid chart configuration: missing visualization type.', + ), + }); + return; + } + + // Defaults the chart pipeline requires but a stored form_data may omit. + if (!parsed.time_range) { + parsed.time_range = 'No filter'; + } + if (!parsed.result_format) { + parsed.result_format = 'json'; + } + if (!parsed.result_type) { + parsed.result_type = 'full'; + } + + setFetchState({ status: 'loaded', formData: parsed }); + } catch (caught) { + if (retry()) { + return; + } + const errorMessage = + caught instanceof DOMException && caught.name === 'AbortError' + ? t('Chart loading timed out.') + : t('Unable to load chart preview.'); + setFetchState({ status: 'error', message: errorMessage }); + } + }, + [formDataKey], + ); + + useEffect(() => { + fetchFormData(); + }, [fetchFormData]); + + return ( + <ChartContainer data-test="chat-chart-embed"> + <ChartHeader> + <ChartTitle>{title ?? t('Chart Preview')}</ChartTitle> + <ChartActions> + {fetchState.status === 'error' && ( + <ActionButton + type="button" + onClick={() => { + fetchFormData(0); + }} + > + <Icons.ReloadOutlined iconSize="s" /> {t('Retry')} + </ActionButton> + )} + <ActionLink + href={exploreUrl} + target="_blank" + rel="noopener noreferrer" + > + <Icons.ExpandOutlined iconSize="s" /> {t('Open in Explore')} + </ActionLink> + </ChartActions> + </ChartHeader> + + {fetchState.status === 'loading' && ( + <CenteredMessage height={height}> + <Loading position="inline-centered" size="s" /> + {t('Loading chart preview...')} + </CenteredMessage> + )} + + {fetchState.status === 'error' && ( + <CenteredMessage height={Math.min(height, 120)}> + <span>{fetchState.message}</span> + <ActionLink + href={exploreUrl} + target="_blank" + rel="noopener noreferrer" + > + <Icons.ExpandOutlined iconSize="s" /> {t('View in Explore instead')} + </ActionLink> + </CenteredMessage> + )} + + {fetchState.status === 'loaded' && ( + <ChartBody height={height} ref={chartBodyRef}> + <StatefulChart + key={chartRenderKey} + formData={fetchState.formData} + width={chartWidth || FALLBACK_CHART_WIDTH} + height={height} + onLoad={queryData => { + const isEmpty = queryData.every( Review Comment: Empty-success versus not-ready is still an open embed issue. Retrying a zero-row result can repeat a completed query, and the upstream refresh does not change that classifier. The base UI needs a readiness signal distinct from row count. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
