I3eka commented on code in PR #43133: URL: https://github.com/apache/superset/pull/43133#discussion_r4385435184
########## superset-frontend/src/features/ai/components/ChatTabsMenu.tsx: ########## @@ -0,0 +1,356 @@ +/** + * 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 The conversation list. + * + * Conversations live behind one menu rather than a tab strip: the panel is narrow + * enough in floating mode that a strip would truncate every name, and the list + * doubles as the history of past conversations, which a strip cannot be. + */ + +import { useCallback, useState } from 'react'; +import type { MouseEvent as ReactMouseEvent } from 'react'; +import { styled } from '@apache-superset/core/theme'; +import { t } from '@apache-superset/core/translation'; +import { Button, Dropdown, Popconfirm } from '@superset-ui/core/components'; +import { Icons } from '@superset-ui/core/components/Icons'; +import type { ChatTab } from '../types'; + +const MenuContainer = styled.div` + background: ${({ theme }) => theme.colorBgElevated}; + border-radius: ${({ theme }) => theme.borderRadius}px; + box-shadow: ${({ theme }) => theme.boxShadowSecondary}; + min-width: ${({ theme }) => theme.sizeUnit * 65}px; + max-height: ${({ theme }) => theme.sizeUnit * 100}px; + overflow-y: auto; + border: 1px solid ${({ theme }) => theme.colorBorderSecondary}; +`; + +const MenuHeader = styled.div` + padding: ${({ theme }) => theme.sizeUnit * 3}px + ${({ theme }) => theme.sizeUnit * 4}px; + border-bottom: 1px solid ${({ theme }) => theme.colorBorderSecondary}; + font-weight: ${({ theme }) => theme.fontWeightStrong}; + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorTextSecondary}; + text-transform: uppercase; + letter-spacing: 0.5px; +`; + +const NewChatButton = styled.button` + display: flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit * 2}px; + width: 100%; + padding: ${({ theme }) => theme.sizeUnit * 2.5}px + ${({ theme }) => theme.sizeUnit * 4}px; + cursor: pointer; + color: ${({ theme }) => theme.colorPrimary}; + font-weight: ${({ theme }) => theme.fontWeightStrong}; + background: none; + border: none; + text-align: left; + transition: background ${({ theme }) => theme.motionDurationMid}; + + &:hover { + background: ${({ theme }) => theme.colorFillTertiary}; + } +`; + +const TabItem = styled.div<{ isActive: boolean }>` + display: flex; + align-items: center; + justify-content: space-between; + padding: ${({ theme }) => theme.sizeUnit * 2.5}px + ${({ theme }) => theme.sizeUnit * 4}px; + cursor: pointer; + background: ${({ theme, isActive }) => + isActive ? theme.colorFillSecondary : 'transparent'}; + border-left: 3px solid + ${({ theme, isActive }) => (isActive ? theme.colorPrimary : 'transparent')}; + transition: background ${({ theme }) => theme.motionDurationMid}; + + &:hover { + background: ${({ theme }) => theme.colorFillTertiary}; + + .action-btn { + opacity: 1; + } + } +`; + +const TabInfo = styled.div` + display: flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit * 2}px; + flex: 1; + overflow: hidden; +`; + +const TabName = styled.span` + font-size: ${({ theme }) => theme.fontSize}px; + color: ${({ theme }) => theme.colorText}; + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; + max-width: ${({ theme }) => theme.sizeUnit * 35}px; +`; + +const TabNameInput = styled.input` + width: 100%; + max-width: ${({ theme }) => theme.sizeUnit * 40}px; + font-size: ${({ theme }) => theme.fontSize}px; + color: ${({ theme }) => theme.colorText}; + background: ${({ theme }) => theme.colorBgContainer}; + border: 1px solid ${({ theme }) => theme.colorBorder}; + border-radius: ${({ theme }) => theme.borderRadius}px; + padding: 2px ${({ theme }) => theme.sizeUnit * 1.5}px; +`; + +const TabTimestamp = styled.span` + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorTextQuaternary}; + white-space: nowrap; + flex-shrink: 0; +`; + +const ActionButtons = styled.div` + display: flex; + align-items: center; + gap: 2px; +`; + +const ActionButton = styled.button` + background: none; + border: none; + padding: ${({ theme }) => theme.sizeUnit}px; + cursor: pointer; + color: ${({ theme }) => theme.colorTextSecondary}; + opacity: 0; + transition: all ${({ theme }) => theme.motionDurationMid}; + display: flex; + align-items: center; + justify-content: center; + border-radius: ${({ theme }) => theme.borderRadius}px; + + &:hover, + &:focus-visible { + opacity: 1; + color: ${({ theme }) => theme.colorError}; + background: ${({ theme }) => theme.colorErrorBg}; + } +`; + +const Divider = styled.div` + height: 1px; + background: ${({ theme }) => theme.colorBorderSecondary}; + margin: ${({ theme }) => theme.sizeUnit}px 0; +`; + +const EmptyState = styled.div` + padding: ${({ theme }) => theme.sizeUnit * 5}px + ${({ theme }) => theme.sizeUnit * 4}px; + text-align: center; + color: ${({ theme }) => theme.colorTextSecondary}; + font-size: ${({ theme }) => theme.fontSizeSM}px; +`; + +const MINUTE_SECONDS = 60; +const HOUR_MINUTES = 60; +const DAY_HOURS = 24; +const WEEK_DAYS = 7; + +export const formatRelativeTime = (timestamp: number): string => { + const seconds = Math.floor((Date.now() - timestamp) / 1000); + if (seconds < MINUTE_SECONDS) { + return t('just now'); + } + const minutes = Math.floor(seconds / MINUTE_SECONDS); + if (minutes < HOUR_MINUTES) { + return t('%sm', String(minutes)); + } + const hours = Math.floor(minutes / HOUR_MINUTES); + if (hours < DAY_HOURS) { + return t('%sh', String(hours)); + } + const days = Math.floor(hours / DAY_HOURS); + if (days < WEEK_DAYS) { + return t('%sd', String(days)); + } + return new Date(timestamp).toLocaleDateString(undefined, { + month: 'short', + day: 'numeric', + }); +}; + +interface ChatTabsMenuProps { + tabs: ChatTab[]; + activeTabId: string; + onSelectTab: (tabId: string) => void; + onNewChat: () => void; + onDeleteTab: (tabId: string) => void; + onRenameTab: (tabId: string, name: string) => void; +} + +export const ChatTabsMenu = ({ + tabs, + activeTabId, + onSelectTab, + onNewChat, + onDeleteTab, + onRenameTab, +}: ChatTabsMenuProps) => { + const [editingTabId, setEditingTabId] = useState<string | null>(null); + const [editingName, setEditingName] = useState(''); + + const startEditing = useCallback((event: ReactMouseEvent, tab: ChatTab) => { + event.stopPropagation(); + setEditingTabId(tab.id); + setEditingName(tab.name); + }, []); + + const cancelEditing = useCallback(() => { + setEditingTabId(null); + setEditingName(''); + }, []); + + const commitRename = useCallback( + (tabId: string) => { + const trimmedName = editingName.trim(); + if (trimmedName) { + onRenameTab(tabId, trimmedName); + } + cancelEditing(); + }, + [cancelEditing, editingName, onRenameTab], + ); + + const menuContent = ( + <MenuContainer data-test="chat-tabs-menu"> + <MenuHeader>{t('Conversations')}</MenuHeader> + <NewChatButton type="button" onClick={onNewChat}> + <Icons.PlusOutlined iconSize="s" /> + <span>{t('New Chat')}</span> + </NewChatButton> + <Divider /> + {tabs.length === 0 ? ( + <EmptyState>{t('No conversations yet')}</EmptyState> + ) : ( + tabs.map(tab => ( + <TabItem + key={tab.id} + isActive={tab.id === activeTabId} + onClick={() => onSelectTab(tab.id)} + > + <TabInfo> + <Icons.MessageOutlined iconSize="s" /> + {editingTabId === tab.id ? ( + <TabNameInput + autoFocus + value={editingName} + onChange={event => setEditingName(event.target.value)} + onClick={event => event.stopPropagation()} + onBlur={() => commitRename(tab.id)} + onKeyDown={event => { + event.stopPropagation(); + if (event.key === 'Enter') { + commitRename(tab.id); + } else if (event.key === 'Escape') { + cancelEditing(); + } + }} + aria-label={t('Conversation name')} + /> + ) : ( + <TabName>{tab.name}</TabName> + )} + {tab.updatedAt !== undefined && ( + <TabTimestamp>{formatRelativeTime(tab.updatedAt)}</TabTimestamp> + )} + </TabInfo> + <ActionButtons> + <ActionButton + type="button" + className="action-btn" + onClick={event => startEditing(event, tab)} + title={t('Rename conversation')} + aria-label={t('Rename conversation')} + > + <Icons.EditOutlined iconSize="s" /> + </ActionButton> + {/* A conversation with messages is confirmed before deletion; an + empty one is discarded without a prompt. */} + {tab.messages.length > 0 ? ( Review Comment: Fixed on this branch in 89a0b8e33f. Confirmation uses the server `messageCount` as well as loaded messages, so an unopened conversation with stored history is not mistaken for an empty one. The unopened/nonempty and empty-conversation UI cases pass. ########## superset-frontend/src/features/ai/AiAssistantPanel.tsx: ########## @@ -0,0 +1,1150 @@ +/** + * 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 The assistant panel. + * + * The host owns where this sits and, when docked, how wide it is, so there is no + * positioning and no resize handle here. What is here is the conversation: the + * header, the transcript, what the assistant is doing while it works, and the + * composer. + * + * The centre of the design is that a run is legible while it happens. An answer + * can take a minute of tool calls, and a spinner for a minute is indistinguishable + * from a hang, so reasoning streams into a preview, each step appends to a tool + * log, and a checkpoint stops the run with a countdown the user can act on. + */ + +import { + memo, + useCallback, + useEffect, + useRef, + useState, + useSyncExternalStore, +} from 'react'; +import type { Dispatch, SetStateAction } from 'react'; +import ReactMarkdown from 'react-markdown'; +import type { Components } from 'react-markdown'; +import { css, keyframes, styled, useTheme } from '@apache-superset/core/theme'; +import { t } from '@apache-superset/core/translation'; +import type { chat as chatApi } from '@apache-superset/core'; +import { + Button, + Input, + Loading, + Tooltip, + Typography, +} from '@superset-ui/core/components'; +import { Icons } from '@superset-ui/core/components/Icons'; +import { chat } from 'src/core/chat'; +import ChatAgentSelect from './components/ChatAgentSelect'; +import ChatTabsMenu from './components/ChatTabsMenu'; +import { REMARK_PLUGINS, useChatMarkdown } from './components/chatMarkdown'; +import { ThoughtProcess } from './components/ThoughtProcess'; +import { useChatBot } from './hooks/useChatBot'; +import { AI_ACTION_EVENT, type AiActionEvent } from './hooks/useAIAction'; +import type { PageContext } from './hooks/usePageContext'; +import type { ChatMessageWithMeta, CheckpointPayload } from './types'; + +/** + * How long a checkpoint waits before continuing on its own. A pause that blocks + * forever is worse than one that resolves optimistically: the user may have + * walked away, and the run should not be stranded. + */ +export const CHECKPOINT_TIMEOUT_SECONDS = 30; + +/** + * Closes a code fence the model has not finished writing. + * + * A streamed answer is parsed on every delta, so a fence arrives in pieces — + * "```", then "sql", then the query. Markdown with an odd number of fences + * renders the opening backticks literally and then reflows once the closing pair + * lands, which reads as the answer glitching. Balancing the count keeps each + * intermediate state a valid document. + */ +export const balanceCodeFences = (text: string): string => { + const fences = text.match(/^```/gm)?.length ?? 0; + return fences % 2 === 0 ? text : `${text}\n\`\`\``; +}; + +/** + * Whether a message carries the structured record of how it was answered, as + * opposed to only the flat log assembled from stream frames. + */ +const hasStructuredThinking = (message: ChatMessageWithMeta): boolean => + Boolean(message.toolCalls?.length || message.thoughts || message.pageContext); + +/** Milliseconds between typewriter frames, and characters per frame. */ +const TYPEWRITER_INTERVAL_MS = 18; +const TYPEWRITER_STEP = 3; + +/** + * The panel's own size as a floating overlay. + * + * Docked width belongs to the host and is not set here. Floating does need a size + * from somewhere, though — the floating host only stacks its children in a corner + * and gives them no dimensions — so these clamp the overlay to the viewport. + */ +const FLOATING_WIDTH_PX = 440; +const FLOATING_MAX_HEIGHT_VH = 70; + +/** + * The panel surface. + * + * Positioning is deliberately absent: the host places this, in both modes. What is + * here is the surface itself — a column that fills whatever box it is given, with a + * floating size for the mode where the host provides no box. + */ +const ChatPanelContainer = styled.div<{ floating: boolean }>` + display: flex; + flex-direction: column; + min-height: 0; + overflow: hidden; + background: ${({ theme }) => theme.colorBgElevated}; + ${({ floating, theme }) => + floating + ? css` + width: min( + ${FLOATING_WIDTH_PX}px, + calc(100vw - ${theme.sizeUnit * 12}px) + ); + height: ${FLOATING_MAX_HEIGHT_VH}vh; + border: 1px solid ${theme.colorBorderSecondary}; + border-radius: ${theme.borderRadiusLG}px; + box-shadow: ${theme.boxShadow}; + ` + : css` + width: 100%; + height: 100%; + `} +`; + +const ChatHeader = styled.div` + display: flex; + justify-content: space-between; + align-items: center; + padding: ${({ theme }) => theme.sizeUnit * 3}px + ${({ theme }) => theme.sizeUnit * 4}px; + background: ${({ theme }) => theme.colorBgContainer}; + border-bottom: 1px solid ${({ theme }) => theme.colorBorderSecondary}; + font-weight: ${({ theme }) => theme.fontWeightStrong}; + font-size: ${({ theme }) => theme.fontSizeLG}px; + color: ${({ theme }) => theme.colorTextHeading}; + flex-shrink: 0; + gap: ${({ theme }) => theme.sizeUnit * 2}px; +`; + +const HeaderGroup = styled.div` + display: flex; + align-items: center; + min-width: 0; +`; + +const HeaderTitle = styled.span` + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +`; + +const ChatMessages = styled.div` + flex: 1; + min-height: 0; + padding: ${({ theme }) => theme.sizeUnit * 4}px; + overflow-y: auto; + + &::-webkit-scrollbar { + width: 4px; + } + + &::-webkit-scrollbar-track { + background: ${({ theme }) => theme.colorBgContainer}; + border-radius: 2px; + } + + &::-webkit-scrollbar-thumb { + background: ${({ theme }) => theme.colorFillSecondary}; + border-radius: 2px; + } +`; + +const MessageBubble = styled.div<{ variant: 'user' | 'assistant' }>` + margin-bottom: ${({ theme }) => theme.sizeUnit * 3}px; + display: flex; + flex-direction: column; + align-items: ${({ variant }) => + variant === 'user' ? 'flex-end' : 'flex-start'}; +`; + +const MessageContent = styled.div<{ variant: 'user' | 'assistant' }>` + max-width: 85%; + padding: ${({ theme }) => theme.sizeUnit * 3}px + ${({ theme }) => theme.sizeUnit * 4}px; + border-radius: ${({ theme }) => theme.borderRadiusLG * 2}px; + background: ${({ theme, variant }) => + variant === 'user' ? theme.colorPrimary : theme.colorBgContainer}; + color: ${({ theme, variant }) => + variant === 'user' ? theme.colorTextLightSolid : theme.colorText}; + font-size: ${({ theme }) => theme.fontSize}px; + line-height: 1.5; + border: ${({ theme, variant }) => + variant === 'assistant' + ? `1px solid ${theme.colorBorderSecondary}` + : 'none'}; + box-shadow: ${({ theme }) => theme.boxShadowTertiary}; + overflow-wrap: anywhere; + + p { + margin: 0; + } + + p:not(:last-child) { + margin-bottom: ${({ theme }) => theme.sizeUnit * 2}px; + } + + a { + color: ${({ theme, variant }) => + variant === 'user' ? theme.colorTextLightSolid : theme.colorPrimary}; + text-decoration: underline; + } + + code { + background: ${({ theme, variant }) => + variant === 'user' ? theme.colorPrimaryActive : theme.colorFillTertiary}; + padding: 2px ${({ theme }) => theme.sizeUnit * 1.5}px; + border-radius: ${({ theme }) => theme.borderRadius}px; + font-size: ${({ theme }) => theme.fontSizeSM}px; + font-family: ${({ theme }) => theme.fontFamilyCode}; + } + + pre { + background: ${({ theme }) => theme.colorFillQuaternary}; + padding: ${({ theme }) => theme.sizeUnit * 3}px; + border-radius: ${({ theme }) => theme.borderRadius}px; + overflow-x: auto; + margin: ${({ theme }) => theme.sizeUnit * 2}px 0; + border: 1px solid ${({ theme }) => theme.colorBorderSecondary}; + } + + pre code { + background: none; + padding: 0; + } +`; + +const MessageActions = styled.div` + display: flex; + gap: ${({ theme }) => theme.sizeUnit}px; + margin-top: ${({ theme }) => theme.sizeUnit}px; +`; + +const ActionButton = styled(Button)` + &&& { + padding: 2px ${({ theme }) => theme.sizeUnit * 1.5}px; + height: ${({ theme }) => theme.sizeUnit * 6}px; + font-size: ${({ theme }) => theme.fontSizeSM}px; + } + + /* The recorded verdict keeps its colour while disabled. Both thumbs lock once + a rating exists, and the default disabled grey would hide which one the + user picked — the state matters more here than the affordance. */ + &&&.is-active, + &&&.is-active:disabled, + &&&.is-active[disabled] { + color: ${({ theme }) => theme.colorPrimary}; + } +`; + +const LiveAnswer = styled.div` + margin-top: ${({ theme }) => theme.sizeUnit * 2}px; + color: ${({ theme }) => theme.colorText}; + font-size: ${({ theme }) => theme.fontSize}px; + line-height: 1.5; + overflow-wrap: anywhere; + + p { + margin: 0; + } + + p:not(:last-child) { + margin-bottom: ${({ theme }) => theme.sizeUnit * 2}px; + } +`; + +const thinkingPulse = keyframes` + 0% { + opacity: 0.45; + } + 50% { + opacity: 1; + } + 100% { + opacity: 0.45; + } +`; + +const ThinkingPreview = styled.div<{ isLive?: boolean }>` + color: ${({ theme }) => theme.colorTextTertiary}; + font-size: ${({ theme }) => theme.fontSizeSM}px; + white-space: pre-wrap; + ${({ isLive }) => + isLive && + css` + animation: ${thinkingPulse} 1.8s ease-in-out infinite; + `} +`; + +const ThinkingDetails = styled.details` + margin-bottom: ${({ theme }) => theme.sizeUnit * 2}px; + color: ${({ theme }) => theme.colorTextTertiary}; + font-size: ${({ theme }) => theme.fontSizeSM}px; + + summary { + cursor: pointer; + user-select: none; + color: ${({ theme }) => theme.colorTextTertiary}; + margin-bottom: ${({ theme }) => theme.sizeUnit}px; + } +`; + +const CheckpointDivider = styled.div` + display: flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit * 3}px; + margin: ${({ theme }) => theme.sizeUnit * 4}px 0 + ${({ theme }) => theme.sizeUnit * 3}px; + + &::before, + &::after { + content: ''; + flex: 1; + height: 1px; + background: ${({ theme }) => theme.colorBorderSecondary}; + } +`; + +const CountdownBadge = styled.span` + font-size: ${({ theme }) => theme.fontSizeSM}px; + font-weight: ${({ theme }) => theme.fontWeightStrong}; + font-variant-numeric: tabular-nums; + color: ${({ theme }) => theme.colorTextSecondary}; + white-space: nowrap; +`; + +const CheckpointContent = styled.div` + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorText}; + line-height: 1.5; +`; + +const CheckpointTaskList = styled.ul` + margin: ${({ theme }) => theme.sizeUnit * 1.5}px 0; + padding-left: ${({ theme }) => theme.sizeUnit * 4.5}px; + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorTextSecondary}; + + li { + margin-bottom: 2px; + } +`; + +const CheckpointEstimate = styled.div` + font-size: ${({ theme }) => theme.fontSizeSM}px; + color: ${({ theme }) => theme.colorTextTertiary}; + margin-top: ${({ theme }) => theme.sizeUnit}px; +`; + +const CheckpointActions = styled.div` + display: flex; + gap: ${({ theme }) => theme.sizeUnit * 2}px; + margin-top: ${({ theme }) => theme.sizeUnit * 2.5}px; +`; + +const ChatInput = styled.div` + padding: ${({ theme }) => theme.sizeUnit * 4}px; + border-top: 1px solid ${({ theme }) => theme.colorBorderSecondary}; + background: ${({ theme }) => theme.colorBgContainer}; + flex-shrink: 0; +`; + +const InputContainer = styled.div` + display: flex; + gap: ${({ theme }) => theme.sizeUnit * 2}px; + align-items: flex-end; +`; + +const QuickPromptsRow = styled.div<{ hasContent: boolean }>` + display: flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit * 2}px; + flex-wrap: wrap; + margin-bottom: ${({ theme, hasContent }) => + hasContent ? `${theme.sizeUnit * 2.5}px` : '0'}; + min-height: ${({ theme, hasContent }) => + hasContent ? `${theme.sizeUnit * 6}px` : '0'}; +`; + +const QuickPromptChip = styled(Button)` + &&& { + width: fit-content; + max-width: 100%; + height: auto; + white-space: normal; + text-align: left; + line-height: 1.35; + word-break: break-word; + } +`; + +const PageContextRow = styled.div` + display: flex; + align-items: center; + margin-bottom: ${({ theme }) => theme.sizeUnit * 1.5}px; +`; + +const PageContextPill = styled.button<{ isActive: boolean }>` + display: inline-flex; + align-items: center; + gap: ${({ theme }) => theme.sizeUnit * 1.5}px; + padding: 3px ${({ theme }) => theme.sizeUnit * 2}px; + border-radius: ${({ theme }) => theme.borderRadiusLG}px; + border: 1px solid + ${({ theme, isActive }) => + isActive ? theme.colorPrimary : theme.colorBorderSecondary}; + background: ${({ theme, isActive }) => + isActive ? theme.colorPrimaryBg : theme.colorFillQuaternary}; + color: ${({ theme, isActive }) => + isActive ? theme.colorPrimary : theme.colorTextTertiary}; + font-size: ${({ theme }) => theme.fontSizeSM}px; + cursor: pointer; + transition: all ${({ theme }) => theme.motionDurationMid}; + max-width: ${({ theme }) => theme.sizeUnit * 62}px; + white-space: nowrap; + + &:hover { + border-color: ${({ theme }) => theme.colorPrimary}; + } +`; + +const PillLabel = styled.span` + overflow: hidden; + text-overflow: ellipsis; Review Comment: Fixed in 428484e846: `liveAnswer` participates in the transcript scroll effect, so answer-only deltas trigger the same scrolling path as tool/thought updates. All 136 AI frontend tests pass on the current head; no manual browser scroll check is claimed. ########## 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); Review Comment: Fixed on this branch in af39e01c8e. Each active run owns its checkpoint resolver. Continue/Stop selects that run's gate, and unmount cleanup releases all remaining gates. The two-tab checkpoint/cancellation and unmount regressions pass. ########## superset/ai/tools/authoring.py: ########## @@ -0,0 +1,307 @@ +# 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. +"""Native AI adapters for Superset's existing MCP authoring tools.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from importlib import import_module +from threading import Thread +from typing import Any, ClassVar, TypeVar + +from pydantic import BaseModel, ValidationError + +from superset.ai.tools.base import AITool, ToolError, ToolOutput +from superset.mcp_service.chart.schemas import GenerateChartRequest +from superset.mcp_service.dashboard.schemas import GenerateDashboardRequest +from superset.mcp_service.dataset.schemas import CreateVirtualDatasetRequest +from superset.utils import json + +ModelT = TypeVar("ModelT", bound=BaseModel) +ToolCaller = Callable[[BaseModel], Any] + +_MCP_TOOL_MODULES = { + "create_virtual_dataset": ( + "superset.mcp_service.dataset.tool.create_virtual_dataset" + ), + "generate_chart": "superset.mcp_service.chart.tool.generate_chart", + "generate_dashboard": ("superset.mcp_service.dashboard.tool.generate_dashboard"), +} + + +def _tool_schema(model: type[BaseModel]) -> dict[str, Any]: + """Expose a request model without its server-only warning field.""" + schema = model.model_json_schema() + properties = dict(schema.get("properties", {})) + properties.pop("sanitization_warnings", None) + schema["properties"] = properties + if required := schema.get("required"): + schema["required"] = [ + name for name in required if name != "sanitization_warnings" + ] + return schema + + +def _validate(model: type[ModelT], payload: dict[str, Any], label: str) -> ModelT: + """Turn Pydantic errors into a correction the model can act on.""" + try: + return model.model_validate(payload) + except ValidationError as ex: + issues = [] + for error in ex.errors(include_url=False)[:3]: + location = ".".join(str(part) for part in error["loc"]) + issues.append(f"{location}: {error['msg']}") + raise ToolError(f"Invalid {label} request: {'; '.join(issues)}.") from ex + + +def _payload(response: Any) -> dict[str, Any]: + if isinstance(response, BaseModel): + return response.model_dump(mode="json", exclude_none=True) + if isinstance(response, dict): + return response + raise ToolError("Superset returned an unexpected authoring response.") + + +async def _call_mcp_tool(tool_name: str, request: BaseModel) -> Any: + """Call the registered tool through FastMCP so it gets a real context.""" + import_module(_MCP_TOOL_MODULES[tool_name]) + + from fastmcp import Client + + from superset.mcp_service.app import mcp + + arguments = { + "request": request.model_dump( + mode="json", + exclude={"sanitization_warnings"}, + exclude_none=True, + ) + } + async with Client(mcp) as client: + result = await client.call_tool(tool_name, arguments) + + if result.is_error: + raise ToolError(f"Superset could not run {tool_name}.") + return ( + result.structured_content + if result.structured_content is not None + else result.data + ) + + +def _run_mcp_tool(tool_name: str, request: BaseModel) -> dict[str, Any]: + """Run FastMCP off the agent loop with isolated Flask request state.""" + from flask import current_app, g + + try: + app = current_app._get_current_object() + user = getattr(g, "user", None) + except RuntimeError as ex: + raise ToolError("Authoring requires an authenticated request.") from ex + + username = getattr(user, "username", None) + email = getattr(user, "email", None) + if not username and not email: + raise ToolError("Authoring requires an authenticated user.") + + outcome: dict[str, Any] = {} + + def run() -> None: + try: + from flask import g as worker_g + + from superset.mcp_service.auth import load_user_with_relationships + + with app.test_request_context(): + worker_g.user = load_user_with_relationships( + username=str(username) if username else None, + email=str(email) if email else None, + ) + if worker_g.user is None: + raise ToolError("The authenticated user could not be reloaded.") + outcome["value"] = asyncio.run(_call_mcp_tool(tool_name, request)) + except BaseException as ex: # noqa: BLE001 + outcome["error"] = ex + + worker = Thread(target=run, name="superset-ai-authoring", daemon=True) + worker.start() + worker.join() Review Comment: Rechecked the authoring adapter and its stalled-worker regression: the join is bounded and returns a controlled ToolError. The error explicitly says the operation may still complete and asks the caller to check for the asset before retrying. This bounds the caller's wait; it does not forcibly terminate an in-flight write. ########## superset/ai/api.py: ########## @@ -0,0 +1,989 @@ +# 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. +""" +REST API for the AI assistant. + +Every route carries ``@protect()`` and is reached through ``@expose`` on a +``BaseSupersetApi`` subclass, which is what makes Flask-AppBuilder's +authorization actually run. Ownership is enforced a second time in the command +and DAO layers, so a conversation identifier is never on its own a capability. +""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Generator +from typing import Any, cast + +from flask import current_app, request, Response, stream_with_context +from flask_appbuilder.api import expose, permission_name, protect, safe +from marshmallow import ValidationError + +from superset.ai.events import ( + error_event, + KEEPALIVE_FRAME, + KEEPALIVE_INTERVAL_SECONDS, +) +from superset.ai.schemas import ( + AgentResponseSchema, + CancelPostSchema, + FeedbackPostSchema, + MessagePostSchema, + RunAcceptedResponseSchema, + SuggestedPromptsPostSchema, + ThreadDetailResponseSchema, + ThreadPostSchema, + ThreadPutSchema, + ThreadResponseSchema, +) +from superset.ai.types import MessageRole, MessageStatus +from superset.commands.ai.exceptions import ( + AIChatMessageInvalidError, + AIChatMessageNotFoundError, + AIChatThreadInvalidError, + AIChatThreadNotFoundError, +) +from superset.extensions import event_logger +from superset.utils.core import get_user_id +from superset.utils.decorators import transaction +from superset.views.base_api import BaseSupersetApi, statsd_metrics + +logger = logging.getLogger(__name__) + +#: Upper bound on how long a client may hold a stream open, so an abandoned +#: browser tab cannot pin a worker indefinitely. +_STREAM_TIMEOUT_SECONDS = 900 + +#: How often a reader checks the event bus for new frames. +#: +#: Deliberately separate from ``KEEPALIVE_INTERVAL_SECONDS``. Passing the +#: keep-alive interval as the poll interval made the reader sleep fifteen seconds +#: between checks and then deliver everything that had accumulated in one batch — +#: so a worker-mode run showed no streaming at all: the answer and every tool call +#: appeared in fifteen-second lumps. One controls responsiveness, the other how +#: often an idle connection is reassured; they are not the same number. +_EVENT_POLL_SECONDS = 0.1 + + +class AIRestApi(BaseSupersetApi): + """Conversations with the AI assistant.""" + + resource_name = "ai" + openapi_spec_tag = "AI Assistant" + allow_browser_login = True + class_permission_name = "AIAssistant" + + openapi_spec_component_schemas = ( + AgentResponseSchema, + CancelPostSchema, + FeedbackPostSchema, + MessagePostSchema, + RunAcceptedResponseSchema, + SuggestedPromptsPostSchema, + ThreadDetailResponseSchema, + ThreadPostSchema, + ThreadPutSchema, + ThreadResponseSchema, + ) + + @expose("/agent/", methods=("GET",)) + @protect() + @safe + @statsd_metrics + @permission_name("read") + def agents(self) -> Response: + """List agent profiles the current user may select. + --- + get: + summary: List available agent profiles + responses: + 200: + description: Available profiles + content: + application/json: + schema: + type: object + properties: + result: + type: array + items: + $ref: '#/components/schemas/AgentResponseSchema' + 401: + $ref: '#/components/responses/401' + 403: + $ref: '#/components/responses/403' + 404: + $ref: '#/components/responses/404' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.ai.factories import get_profiles + + profiles = get_profiles().visible_to_current_user() + return self.response(200, result=[p.to_public_dict() for p in profiles]) + + @expose("/model/", methods=("GET",)) + @protect() + @safe + @statsd_metrics + @permission_name("read") + def models(self) -> Response: + """List models this deployment has configured. + --- + get: + summary: List selectable models + responses: + 200: + description: Configured model identifiers + content: + application/json: + schema: + type: object + properties: + result: + type: array + items: + type: string + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.ai.factories import get_provider + + return self.response(200, result=get_provider().available_models()) + + @expose("/thread/", methods=("POST",)) + @protect() + @safe + @statsd_metrics + @permission_name("write") + @event_logger.log_this_with_context( + action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.post_thread", + log_to_statsd=False, + ) + def post_thread(self) -> Response: + """Create a conversation. + --- + post: + summary: Create a conversation + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/ThreadPostSchema' + responses: + 201: + description: Conversation created + content: + application/json: + schema: + type: object + properties: + result: + $ref: '#/components/schemas/ThreadResponseSchema' + 400: + $ref: '#/components/responses/400' + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.commands.ai import CreateAIChatThreadCommand + + try: + payload = ThreadPostSchema().load(request.json or {}) + except ValidationError as error: + return self.response_400(message=error.messages) + try: + thread = CreateAIChatThreadCommand( + user_id=self._user_id(), + title=payload.get("title"), + agent_key=payload.get("agent_key"), + ).run() + except AIChatThreadInvalidError as ex: + return self.response_422(message=str(ex)) + return self.response(201, result=_thread_dict(thread)) + + @expose("/thread/", methods=("GET",)) + @protect() + @safe + @statsd_metrics + @permission_name("read") + def get_threads(self) -> Response: + """List the current user's conversations. + --- + get: + summary: List conversations + parameters: + - in: query + name: limit + schema: + type: integer + - in: query + name: offset + schema: + type: integer + responses: + 200: + description: Conversations + content: + application/json: + schema: + type: object + properties: + count: + type: integer + result: + type: array + items: + $ref: '#/components/schemas/ThreadResponseSchema' + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.daos.ai import AIChatThreadDAO + + limit = request.args.get("limit", type=int) or 50 + offset = request.args.get("offset", type=int) or 0 + threads = AIChatThreadDAO.find_all_for_user( + self._user_id(), limit=limit, offset=offset + ) + return self.response( + 200, + count=len(threads), + result=[_thread_dict(thread) for thread in threads], + ) + + @expose("/thread/<thread_uuid>", methods=("GET",)) + @protect() + @safe + @statsd_metrics + @permission_name("read") + def get_thread(self, thread_uuid: str) -> Response: + """Fetch a conversation and its messages. + --- + get: + summary: Get a conversation + parameters: + - in: path + name: thread_uuid + required: true + schema: + type: string + format: uuid + responses: + 200: + description: Conversation with messages + content: + application/json: + schema: + type: object + properties: + result: + $ref: '#/components/schemas/ThreadDetailResponseSchema' + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.daos.ai import ( + AIChatFeedbackDAO, + AIChatMessageDAO, + AIChatThreadDAO, + ) + + user_id = self._user_id() + thread = AIChatThreadDAO.find_by_uuid_for_user(thread_uuid, user_id) + if thread is None: + return self.response_404() + + messages = AIChatMessageDAO.find_for_thread(thread) + # Resolved for the whole transcript at once so the panel can show which + # replies this user already rated; without it a reload loses the verdict + # and the message looks unrated. + verdicts = AIChatFeedbackDAO.find_verdicts_for_user( + [message.id for message in messages], user_id + ) + detail = _thread_dict(thread) + detail["messages"] = [ + _message_dict(message, liked=verdicts.get(message.id)) + for message in messages + ] + return self.response(200, result=detail) + + @expose("/thread/<thread_uuid>", methods=("PUT",)) + @protect() + @safe + @statsd_metrics + @permission_name("write") + @event_logger.log_this_with_context( + action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.put_thread", + log_to_statsd=False, + ) + def put_thread(self, thread_uuid: str) -> Response: + """Rename or archive a conversation. + --- + put: + summary: Update a conversation + parameters: + - in: path + name: thread_uuid + required: true + schema: + type: string + format: uuid + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/ThreadPutSchema' + responses: + 200: + description: Conversation updated + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + 422: + $ref: '#/components/responses/422' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.commands.ai import UpdateAIChatThreadCommand + + try: + payload = ThreadPutSchema().load(request.json or {}) + except ValidationError as error: + return self.response_400(message=error.messages) + try: + thread = UpdateAIChatThreadCommand( + thread_uuid, + self._user_id(), + title=payload.get("title"), + status=payload.get("status"), + ).run() + except AIChatThreadNotFoundError: + return self.response_404() + except AIChatThreadInvalidError as ex: + return self.response_422(message=str(ex)) + return self.response(200, result=_thread_dict(thread)) + + @expose("/thread/<thread_uuid>", methods=("DELETE",)) + @protect() + @safe + @statsd_metrics + @permission_name("write") + @event_logger.log_this_with_context( + action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.delete_thread", + log_to_statsd=False, + ) + def delete_thread(self, thread_uuid: str) -> Response: + """Delete a conversation and its messages. + --- + delete: + summary: Delete a conversation + parameters: + - in: path + name: thread_uuid + required: true + schema: + type: string + format: uuid + responses: + 200: + description: Conversation deleted + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.commands.ai import DeleteAIChatThreadCommand + + try: + DeleteAIChatThreadCommand(thread_uuid, self._user_id()).run() + except AIChatThreadNotFoundError: + return self.response_404() + return self.response(200, message="OK") + + @expose("/thread/<thread_uuid>/message", methods=("POST",)) + @protect() + @safe + @statsd_metrics + @permission_name("write") + @event_logger.log_this_with_context( + action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.post_message", + log_to_statsd=False, + ) + def post_message(self, thread_uuid: str) -> Response: + """Post a user message and start a run. + --- + post: + summary: Post a message + description: > + Stores the user's message, creates a placeholder assistant message, + and starts a run. Returns immediately; consume the answer from the + stream endpoint using the returned run identifier. + parameters: + - in: path + name: thread_uuid + required: true + schema: + type: string + format: uuid + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/MessagePostSchema' + responses: + 202: + description: Run accepted + content: + application/json: + schema: + type: object + properties: + result: + $ref: '#/components/schemas/RunAcceptedResponseSchema' + 400: + $ref: '#/components/responses/400' + 401: + $ref: '#/components/responses/401' + 404: + $ref: '#/components/responses/404' + 422: + $ref: '#/components/responses/422' + """ + if (unavailable := self._reject_if_unconfigured()) is not None: + return unavailable + + from superset.ai.orchestrator import new_run_id + from superset.commands.ai import AppendAIChatMessageCommand + + try: + payload = MessagePostSchema().load(request.json or {}) + except ValidationError as error: + return self.response_400(message=error.messages) + user_id = self._user_id() + + try: + user_message = AppendAIChatMessageCommand( + thread_uuid, + user_id, + MessageRole.USER, + payload["content"], + request_id=payload.get("request_id"), + ).run() + # Created up front so a client that reconnects before any token + # arrives still has a row to attach its stream to. + assistant_message = AppendAIChatMessageCommand( + thread_uuid, + user_id, + MessageRole.ASSISTANT, + "", + request_id=payload.get("request_id"), + status=MessageStatus.PENDING, + ).run() + except AIChatThreadNotFoundError: + return self.response_404() + except (AIChatMessageInvalidError, AIChatThreadInvalidError) as ex: + return self.response_422(message=str(ex)) + + run_id = new_run_id() Review Comment: Fixed on this branch in ca93d9b91b. Replays retain the accepted run identity/context, and an atomic pending-to-streaming claim permits only one executor. An uncertain broker acknowledgement can cause another delivery of the same run, not another authorized execution. API/command/worker idempotency regressions pass. ########## 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) Review Comment: Fixed in 83a8a4b617. Redis expiry is refreshed by every publish, including the first event, independently of whether a reader ever connects or closes. The first-publish and progress/expiry regressions pass. -- 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]
