diff --git a/zeppelin-web-angular/e2e/models/assistant-conversation-page.util.ts b/zeppelin-web-angular/e2e/models/assistant-conversation-page.util.ts new file mode 100644 index 00000000000..a349400d9f6 --- /dev/null +++ b/zeppelin-web-angular/e2e/models/assistant-conversation-page.util.ts @@ -0,0 +1,257 @@ +/* + * 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. + */ + +import { Page, Route, WebSocketRoute } from '@playwright/test'; + +const ZEPPELIN_WS_URL_PATTERN = /\/ws(\?|$)/; +const CREATED_AT = '2026-10-19T12:00:00Z'; +const STREAM_PREFIX = 'I checked the notebook results.'; + +interface ConversationMetadata { + id: string; + noteId: string; + ownerId: string; + title: string; + createdAt: string; + updatedAt: string; + canSendMessage: boolean; +} + +// Message.User / Message.Assistant / Message.Tool, not the panel's segments contract. +interface RawMessage { + id: string; + createdAt: string; + role: 'user' | 'assistant' | 'tool'; + content: string; + toolCalls?: Array<{ id: string; name: string; arguments: Record }>; + toolCallId?: string; +} + +interface SendMessage { + noteId: string; + conversationId: string; + content: string; +} + +interface SocketEnvelope { + op?: string; + data?: Record; +} + +const parse = (frame: string | Buffer): SocketEnvelope | undefined => { + try { + return JSON.parse(frame.toString()) as SocketEnvelope; + } catch { + return undefined; + } +}; + +const message = (id: string, role: RawMessage['role'], content: string): RawMessage => ({ + id, + role, + content, + createdAt: CREATED_AT, + ...(role === 'assistant' ? { toolCalls: [] } : {}) +}); + +const revenueHistory = (): RawMessage[] => { + // Exactly 50 raw messages follow the first assistant round, as requested by the + // production transport. Its tool results straddle the boundary without adding bubbles. + const toolCalls = Array.from({ length: 47 }, (_, index) => ({ + id: `c1-tool-${index + 1}`, + name: 'read_notebook', + arguments: { paragraphId: `regional-revenue-${index + 1}` } + })); + return [ + message('c1-user-1', 'user', 'How did revenue change by region?'), + { + ...message('c1-answer-1a', 'assistant', 'I checked the regional revenue paragraph.'), + toolCalls + }, + ...toolCalls.map(call => ({ + ...message(`${call.id}-result`, 'tool', 'Regional revenue data.'), + toolCallId: call.id + })), + message('c1-answer-1b', 'assistant', 'Revenue increased in all three regions, led by Europe.'), + message('c1-user-2', 'user', 'Which region should we investigate next?'), + message( + 'c1-answer-2', + 'assistant', + [ + 'Compare Europe’s order volume with its average order value.', + 'Start with the monthly totals in the regional revenue paragraph. Use the same reporting period for each region.', + 'Then separate the change in order count from the change in average order value. Either can explain revenue growth.', + 'Check whether the regional mix changed. A larger share of high-value orders can raise revenue without a similar rise in volume.', + 'Compare those observations with the Americas and Asia Pacific before drawing a conclusion about Europe.', + 'The earlier answer describes the initial regional comparison. Scroll to the top to read that part of this conversation.' + ].join('\n\n') + ) + ]; +}; + +/** Owns only assistant wire traffic; notebook/auth/other socket traffic still reaches Zeppelin. */ +export class AssistantConversationWireMock { + readonly sent: SendMessage[] = []; + readonly historyRequests: Array<{ conversationId: string; cursor: string | null; limit: number }> = []; + readonly restRequests: Array<{ method: string; path: string; body?: unknown }> = []; + private noteId?: string; + private readonly conversations = new Map(); + private readonly runs: Array<{ socket: WebSocketRoute; request: SendMessage; completed: boolean }> = []; + private createdCount = 0; + + static async install(page: Page): Promise { + const mock = new AssistantConversationWireMock(); + await page.route( + url => !!mock.noteId && (url.pathname === mock.restBase() || url.pathname.startsWith(`${mock.restBase()}/`)), + route => mock.handleRest(route) + ); + // Install before the initial navigation: hash navigation to the note reuses its socket. + await page.routeWebSocket(ZEPPELIN_WS_URL_PATTERN, socket => { + const server = socket.connectToServer(); + socket.onMessage(frame => { + const envelope = parse(frame); + if (envelope?.op === 'ASSISTANT_SEND_MESSAGE') { + mock.startRun(socket, envelope.data); + } else { + server.send(frame); + } + }); + server.onMessage(frame => { + if (parse(frame)?.op !== 'ASSISTANT_EVENT') socket.send(frame); + }); + }); + return mock; + } + + useNote(noteId: string): void { + this.noteId = noteId; + this.addConversation('c1', 'Revenue analysis', revenueHistory()); + this.addConversation('c2', 'Notebook summary', [ + message('c2-user-1', 'user', 'Summarize this notebook.'), + message('c2-answer-1', 'assistant', 'This notebook compares monthly revenue across regions.') + ]); + } + + // No timer can complete a stream: the test releases it after observing the busy DOM. + completeRun(index: number): void { + const run = this.runs[index]; + if (!run || run.completed) throw new Error(`No pending assistant run at index ${index}`); + run.completed = true; + const content = [ + STREAM_PREFIX, + run.request.conversationId === 'c2' + ? '| Region | Summary |\n| --- | --- |\n| Europe | Strongest regional growth |' + : '| Region | Revenue |\n| --- | --- |\n| Europe | 118 |' + ].join('\n\n'); + const answer = message(`run-${index}-answer`, 'assistant', content); + this.conversations.get(run.request.conversationId)!.messages.push(answer); + this.emit(run.socket, run.request.conversationId, 'message.done', { messageId: answer.id, content }); + this.emit(run.socket, run.request.conversationId, 'run.completed', { runId: `run-${index}` }); + } + + private restBase(): string { + return `/api/notes/${encodeURIComponent(this.noteId!)}/conversations`; + } + + private addConversation(id: string, title: string, messages: RawMessage[]): ConversationMetadata { + const metadata = { + id, + noteId: this.noteId!, + ownerId: 'e2e-user', + title, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + canSendMessage: true + }; + this.conversations.set(id, { metadata, messages }); + return metadata; + } + + private async handleRest(route: Route): Promise { + const url = new URL(route.request().url()); + const suffix = url.pathname.slice(this.restBase().length); + const method = route.request().method(); + this.restRequests.push({ + method, + path: url.pathname, + ...(method === 'POST' ? { body: route.request().postDataJSON() } : {}) + }); + const respond = (body: unknown, status = 200) => + route.fulfill({ status, json: { status: status === 201 ? 'CREATED' : 'OK', message: '', body } }); + if (!suffix && method === 'GET') { + await respond([...this.conversations.values()].map(({ metadata }) => metadata)); + return; + } + if (!suffix && method === 'POST') { + const body = route.request().postDataJSON() as { title?: string }; + await respond( + this.addConversation(`c-created-${++this.createdCount}`, body.title ?? 'New conversation', []), + 201 + ); + return; + } + const match = /^\/([^/]+)\/messages$/.exec(suffix); + const conversationId = match ? decodeURIComponent(match[1]) : ''; + const conversation = this.conversations.get(conversationId); + if (conversation && method === 'GET') { + const cursor = url.searchParams.get('cursor'); + const limit = Number(url.searchParams.get('limit') ?? 10); + this.historyRequests.push({ conversationId, cursor, limit }); + const newestFirst = [...conversation.messages].reverse(); + const cursorIndex = cursor === null ? -1 : newestFirst.findIndex(item => item.id === cursor); + if (!Number.isInteger(limit) || limit <= 0 || (cursor !== null && cursorIndex < 0)) { + await route.fulfill({ status: cursor !== null && cursorIndex < 0 ? 404 : 400 }); + return; + } + const start = cursorIndex + 1; + const messages = newestFirst.slice(start, start + limit); + const nextCursor = start + messages.length < newestFirst.length ? messages[messages.length - 1].id : null; + // Gson omits the null cursor field in MessagesResponse. + await respond({ messages, ...(nextCursor === null ? {} : { cursor: nextCursor }) }); + return; + } + throw new Error(`Unexpected assistant REST request: ${method} ${url.pathname}`); + } + + private startRun(socket: WebSocketRoute, data?: Record): void { + if ( + typeof data?.noteId !== 'string' || + data?.noteId !== this.noteId || + typeof data.conversationId !== 'string' || + typeof data.content !== 'string' || + !this.conversations.has(data.conversationId) + ) { + throw new Error(`Unexpected assistant send: ${JSON.stringify(data)}`); + } + const request = { noteId: data.noteId, conversationId: data.conversationId, content: data.content }; + const index = this.runs.length; + this.sent.push(request); + this.runs.push({ socket, request, completed: false }); + this.conversations + .get(request.conversationId)! + .messages.push(message(`run-${index}-user`, 'user', request.content)); + this.emit(socket, request.conversationId, 'run.started', { runId: `run-${index}`, createdAt: CREATED_AT }); + this.emit(socket, request.conversationId, 'message.delta', { + messageId: `run-${index}-answer`, + delta: STREAM_PREFIX + }); + } + + private emit(socket: WebSocketRoute, conversationId: string, type: string, payload: unknown): void { + socket.send(JSON.stringify({ op: 'ASSISTANT_EVENT', data: { conversationId, type, payload } })); + } +} diff --git a/zeppelin-web-angular/e2e/tests/notebook/assistant/assistant-conversation.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/assistant/assistant-conversation.spec.ts index 0d447b78420..aeb6811b533 100644 --- a/zeppelin-web-angular/e2e/tests/notebook/assistant/assistant-conversation.spec.ts +++ b/zeppelin-web-angular/e2e/tests/notebook/assistant/assistant-conversation.spec.ts @@ -11,27 +11,34 @@ */ import { expect, test } from '@playwright/test'; +import { AssistantConversationWireMock } from '../../../models/assistant-conversation-page.util'; import { NotebookAssistantPage } from '../../../models/notebook-assistant-page'; import { addPageAnnotationBeforeEach, createTestNotebook, PAGES, waitForZeppelinReady } from '../../../utils'; -test.describe('Notebook AI Assistant fixture conversations', () => { +test.describe('Notebook AI Assistant conversations over REST and WebSocket', () => { addPageAnnotationBeforeEach(PAGES.WORKSPACE.NOTEBOOK_ASSISTANT); let assistant: NotebookAssistantPage; + let wire: AssistantConversationWireMock; let noteId: string | undefined; test.beforeEach(async ({ page }) => { noteId = undefined; assistant = new NotebookAssistantPage(page); + wire = await AssistantConversationWireMock.install(page); await page.goto('/#/'); await waitForZeppelinReady(page); ({ noteId } = await createTestNotebook(page)); + wire.useNote(noteId); await page.goto(`/#/notebook/${noteId}?reactAssistant=true`); await waitForZeppelinReady(page); await assistant.toggleButton.click(); await expect(assistant.question).toBeEditable(); + await expect + .poll(() => wire.restRequests) + .toContainEqual({ method: 'GET', path: `/api/notes/${noteId}/conversations` }); }); test.afterEach(async ({ page }) => { @@ -45,12 +52,19 @@ test.describe('Notebook AI Assistant fixture conversations', () => { const question = 'Explain the revenue results for this new conversation.'; await assistant.sendQuestion(question); + await expect.poll(() => wire.sent).toEqual([{ noteId, conversationId: 'c-created-1', content: question }]); + expect(wire.restRequests).toContainEqual({ + method: 'POST', + path: `/api/notes/${noteId}/conversations`, + body: {} + }); const answer = assistant.messages.getByRole('article', { name: 'Assistant answer', exact: true }); await expect(answer).toHaveAttribute('aria-busy', 'true'); await expect(answer).toContainText('I checked'); const original = await answer.elementHandle(); expect(original).not.toBeNull(); + wire.completeRun(0); await expect(answer.getByRole('table')).toBeVisible(); await expect(answer).not.toHaveAttribute('aria-busy', 'true'); @@ -62,10 +76,13 @@ test.describe('Notebook AI Assistant fixture conversations', () => { ); await expect(answer).toHaveCount(1); + const previousHistoryReads = wire.historyRequests.length; await assistant.closeButton.click(); await expect(assistant.panel).toBeHidden(); await assistant.toggleButton.click(); await assistant.selectConversation('New conversation'); + await expect.poll(() => wire.historyRequests.length).toBeGreaterThan(previousHistoryReads); + expect(wire.historyRequests).toContainEqual({ conversationId: 'c-created-1', cursor: null, limit: 50 }); await expect(assistant.messages).not.toHaveAttribute('aria-busy', 'true'); await expect(assistant.messages.getByRole('article', { name: 'Your question', exact: true })).toContainText( @@ -87,6 +104,7 @@ test.describe('Notebook AI Assistant fixture conversations', () => { await expect(assistant.question).toBeEditable(); await assistant.sendQuestion(aQuestion); + await expect.poll(() => wire.sent).toEqual([{ noteId, conversationId: 'c1', content: aQuestion }]); await expect( assistant.messages.getByRole('article', { name: 'Assistant answer', exact: true }).last() @@ -101,7 +119,16 @@ test.describe('Notebook AI Assistant fixture conversations', () => { await assistant.sendButton.click(); const bAnswer = assistant.messages.getByRole('article', { name: 'Assistant answer', exact: true }).last(); + await expect + .poll(() => wire.sent) + .toEqual([ + { noteId, conversationId: 'c1', content: aQuestion }, + { noteId, conversationId: 'c2', content: bQuestion } + ]); + await expect(bAnswer).toHaveAttribute('aria-busy', 'true'); + wire.completeRun(1); await expect(bAnswer.getByRole('table')).toBeVisible(); + await expect(bAnswer.getByRole('table')).toContainText('Strongest regional growth'); await expect(bAnswer).not.toHaveAttribute('aria-busy', 'true'); await expect(assistant.question).toBeEditable(); await expect(assistant.messages).not.toHaveAttribute('aria-busy', 'true'); @@ -110,10 +137,14 @@ test.describe('Notebook AI Assistant fixture conversations', () => { await expect(assistant.messages).not.toContainText(aQuestion); await assistant.question.fill('Keep this next B draft.'); + wire.completeRun(0); + await expect(bAnswer.getByRole('table')).not.toContainText('118'); await assistant.selectConversation('Revenue analysis'); await expect(assistant.messages.getByText(aQuestion, { exact: true })).toHaveCount(1); await expect(assistant.messages.getByRole('table')).toHaveCount(1); + await expect(assistant.messages.getByRole('table')).toContainText('118'); + await expect(assistant.messages).not.toContainText('Strongest regional growth'); await expect(assistant.question).toBeEditable(); await expect(assistant.messages).not.toContainText(bQuestion); @@ -122,6 +153,8 @@ test.describe('Notebook AI Assistant fixture conversations', () => { await expect(assistant.question).toHaveValue('Keep this next B draft.'); await expect(assistant.messages.getByText(bQuestion, { exact: true })).toHaveCount(1); await expect(assistant.messages.getByRole('table')).toHaveCount(1); + await expect(assistant.messages.getByRole('table')).toContainText('Strongest regional growth'); + await expect(assistant.messages.getByRole('table')).not.toContainText('118'); await expect(assistant.messages).toContainText(summary); }); @@ -133,11 +166,16 @@ test.describe('Notebook AI Assistant fixture conversations', () => { await expect(laterAnswer).toBeVisible(); const original = await laterAnswer.elementHandle(); expect(original).not.toBeNull(); + expect(wire.historyRequests).toContainEqual({ conversationId: 'c1', cursor: null, limit: 50 }); + await expect(assistant.messages).not.toContainText('How did revenue change by region?'); await assistant.messages.evaluate(element => { element.scrollTop = 0; }); + await expect + .poll(() => wire.historyRequests) + .toContainEqual({ conversationId: 'c1', cursor: 'c1-tool-1-result', limit: 50 }); await expect(assistant.messages.getByText('How did revenue change by region?', { exact: true })).toHaveCount(1); await expect(assistant.messages).not.toHaveAttribute('aria-busy', 'true'); const joinedAnswer = assistant.messages.getByRole('article', { name: 'Assistant answer', exact: true }).filter({ diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.spec.ts b/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.spec.ts index e1e763c3a94..31ba8fadded 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.spec.ts +++ b/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.spec.ts @@ -54,7 +54,7 @@ const fakeSocket = () => { }; }; -const forNote = (noteId: string, onAuthError?: (status: number) => void) => +const forNote = (noteId: string, onAuthError?: (status: number, location: string | null) => void) => createAssistantConversationApi('https://example.test/zeppelin/api', noteId, onAuthError); const iterate = (iterable: AsyncIterable): AsyncIterator => iterable[Symbol.asyncIterator](); @@ -226,7 +226,7 @@ describe('assistant transport', () => { 'fetch', vi .fn() - .mockResolvedValueOnce(new Response(null, { status: 401 })) + .mockResolvedValueOnce(new Response(null, { status: 401, headers: { Location: '/login' } })) .mockResolvedValueOnce(new Response(null, { status: 405 })) .mockResolvedValueOnce(new Response(null, { status: 500 })) ); @@ -236,7 +236,28 @@ describe('assistant transport', () => { await expect(transport.createConversation({})).rejects.toMatchObject({ status: 405 }); await expect(transport.getMessages('c')).rejects.toBeInstanceOf(AssistantHttpError); - expect(onAuthError.mock.calls).toEqual([[401], [405]]); + expect(onAuthError.mock.calls).toEqual([ + [401, '/login'], + [405, null] + ]); + }); + + it('preserves the rejected HTTP request when host authentication handling throws', async () => { + const handlerError = new Error('Host navigation failed'); + const logged = vi.spyOn(console, 'error').mockImplementation(() => undefined); + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(null, { status: 405 }))); + try { + const transport = forNote('n', () => { + throw handlerError; + }); + await expect(transport.listConversations()).rejects.toMatchObject({ + name: 'AssistantHttpError', + status: 405 + }); + expect(logged).toHaveBeenCalledWith('Assistant authentication handler failed', handlerError); + } finally { + logged.mockRestore(); + } }); it('sends over the socket and yields only this conversation until a terminal event', async () => { diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.ts b/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.ts index ef6f0951a6b..5ffa25a3487 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.ts +++ b/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/model/assistantTransport.ts @@ -159,12 +159,17 @@ const unwrapBody = (value: unknown): unknown => typeof value === 'object' && value !== null && 'body' in value ? value.body : value; /** Hands 401/405 to the host, which owns login redirects and logout. */ -export type AssistantAuthErrorHandler = (status: number) => void; +export type AssistantAuthErrorHandler = (status: number, location: string | null) => void; const httpError = (response: Response, onAuthError?: AssistantAuthErrorHandler): AssistantHttpError => { const error = new AssistantHttpError(response.status); if (error.status === 401 || error.status === 405) { - onAuthError?.(error.status); + try { + onAuthError?.(error.status, response.headers.get('Location')); + } catch (handlerError) { + // Authentication side effects must not replace the original request failure. + console.error('Assistant authentication handler failed', handlerError); + } } return error; }; diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/ui/AssistantMessageItem.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/ui/AssistantMessageItem.tsx index fa6203db258..110815bd673 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/ui/AssistantMessageItem.tsx +++ b/zeppelin-web-angular/projects/zeppelin-react/src/entities/assistant/ui/AssistantMessageItem.tsx @@ -57,7 +57,7 @@ export const AssistantMessageItem = ({ Assistant {content && ( -
+
{renderContent(content)}
)} diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.spec.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.spec.tsx index acb25258601..470e83ef51a 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.spec.tsx +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.spec.tsx @@ -10,14 +10,102 @@ * limitations under the License. */ -import { act } from 'react'; -import { fireEvent, screen } from '@testing-library/react'; -import { afterEach, describe, expect, it, vi } from 'vitest'; -import { NOTEBOOK_SIDEBAR_WIDTH, type AssistantShellProps } from '@zeppelin/sdk'; -import { mount } from './AssistantWorkspace'; +import { act, StrictMode } from 'react'; +import { fireEvent, render, screen, type RenderResult } from '@testing-library/react'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + NOTEBOOK_SIDEBAR_WIDTH, + type AssistantConnection, + type AssistantShellProps, + type AssistantSocketEvent +} from '@zeppelin/sdk'; +import { AssistantWorkspace, mount } from './AssistantWorkspace'; + +const socketHarness = () => { + const events = new Set<(event: AssistantSocketEvent) => void>(); + const closes = new Set<() => void>(); + return { + send: vi.fn(), + subscribe: (listener: (event: AssistantSocketEvent) => void) => { + events.add(listener); + return () => events.delete(listener); + }, + subscribeClose: (listener: () => void) => { + closes.add(listener); + return () => closes.delete(listener); + }, + emit: (event: AssistantSocketEvent) => events.forEach(listener => listener(event)), + close: () => closes.forEach(listener => listener()), + consumers: () => [events.size, closes.size] + }; +}; + +const originalFetch = globalThis.fetch; +let cleanup: (() => Promise) | undefined; + +beforeEach(() => { + vi.stubGlobal( + 'fetch', + vi.fn(() => Promise.reject(new Error('Unexpected Assistant REST request'))) + ); +}); + +afterEach(async () => { + try { + await cleanup?.(); + } finally { + cleanup = undefined; + globalThis.fetch = originalFetch; + } +}); + +const connection = (socket = socketHarness()): AssistantConnection => ({ + apiBase: '/zeppelin/api', + socket, + sessionKey: 1, + onAuthError: vi.fn() +}); + +const json = (body: unknown, status = 200) => + new Response(JSON.stringify({ status: 'OK', body }), { + status, + headers: { 'Content-Type': 'application/json' } + }); + +const mockEmptyConversations = () => + vi.stubGlobal( + 'fetch', + vi.fn((url: string, options?: RequestInit) => { + if (/^\/zeppelin\/api\/notes\/[^/]+\/conversations$/.test(url) && (options?.method ?? 'GET') === 'GET') { + return Promise.resolve(json([])); + } + return Promise.reject(new Error(`Unexpected Assistant REST request: ${options?.method ?? 'GET'} ${url}`)); + }) + ); + +const conversation = { + id: 'saved', + title: 'Saved conversation', + ownerId: 'alice', + canSendMessage: true, + updatedAt: '2026-11-01T00:00:00Z' +}; + +const question = () => screen.getByRole('textbox', { name: 'Question' }) as HTMLTextAreaElement; +const submit = async (value: string) => { + await act(async () => { + fireEvent.change(question(), { target: { value } }); + fireEvent.click(screen.getByRole('button', { name: 'Send' })); + }); +}; +const selectSaved = async () => { + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Open conversations' }))); + await act(async () => fireEvent.click(screen.getByRole('button', { name: /^Saved conversation/ }))); +}; describe('AssistantWorkspace remote', () => { it('portals the navigation and panel, coordinates close, and shares the sidebar width', async () => { + mockEmptyConversations(); const root = document.createElement('div'); const navigation = document.createElement('div'); const panel = document.createElement('div'); @@ -26,6 +114,9 @@ describe('AssistantWorkspace remote', () => { let requestClose = () => {}; const props: AssistantShellProps = { noteId: 'note-a', + connection: connection(), + paragraphs: [], + onOpenParagraph: vi.fn(), slots: [ { element: navigation, kind: 'navigation' }, { element: panel, kind: 'panel' } @@ -40,10 +131,19 @@ describe('AssistantWorkspace remote', () => { }; } }; - let handle: ReturnType; + let unmountRoot: (() => void) | undefined; + let updateRoot: ((props: AssistantShellProps) => void) | undefined; + const unmount = async () => { + const dispose = unmountRoot; + unmountRoot = undefined; + await act(async () => dispose?.()); + }; + cleanup = unmount; await act(async () => { - handle = mount(root, props); + const handle = mount(root, props); + unmountRoot = handle.unmount; + updateRoot = handle.update; }); expect(navigation.querySelector('button')).not.toBeNull(); expect(panel.textContent).toBe(''); @@ -81,56 +181,438 @@ describe('AssistantWorkspace remote', () => { expect(onPanelVisibilityChange).toHaveBeenLastCalledWith(false); await act(async () => navigation.querySelector('button')?.click()); - await act(async () => handle.update({ ...props, noteId: 'note-b' })); + await act(async () => updateRoot!({ ...props, noteId: 'note-b' })); expect(panel.textContent).toBe(''); - await act(async () => { - handle.unmount(); - }); + await unmount(); expect(navigation.innerHTML).toBe(''); }); }); describe('AssistantWorkspace', () => { - let cleanup: (() => Promise) | undefined; - - afterEach(async () => { - await cleanup?.(); - cleanup = undefined; - }); - - const setup = async () => { + const setup = async (overrides: Partial = {}) => { const navigation = document.createElement('div'); const panel = document.createElement('div'); document.body.append(navigation, panel); const onPanelVisibilityChange = vi.fn(); const onPanelWidthChange = vi.fn(); - let handle: ReturnType; - await act(async () => { - handle = mount(document.createElement('div'), { - noteId: 'note-a', - slots: [ - { element: navigation, kind: 'navigation' }, - { element: panel, kind: 'panel' } - ], - panelWidth: 370, - onPanelVisibilityChange, - onPanelWidthChange, - subscribePanelClose: () => () => {} - }); - }); + let unmountRoot: (() => void) | undefined; + let updateRoot: ((props: AssistantShellProps) => void) | undefined; + let props: AssistantShellProps = { + noteId: 'note-a', + connection: connection(), + paragraphs: [], + onOpenParagraph: vi.fn(), + slots: [ + { element: navigation, kind: 'navigation' }, + { element: panel, kind: 'panel' } + ], + panelWidth: 370, + onPanelVisibilityChange, + onPanelWidthChange, + subscribePanelClose: () => () => {}, + ...overrides + }; const unmount = async () => { - await act(async () => handle.unmount()); - navigation.remove(); - panel.remove(); + const dispose = unmountRoot; + unmountRoot = undefined; + try { + await act(async () => dispose?.()); + } finally { + navigation.remove(); + panel.remove(); + } }; cleanup = unmount; + await act(async () => { + const handle = mount(document.createElement('div'), props); + unmountRoot = handle.unmount; + updateRoot = handle.update; + }); + expect(screen.getByRole('button', { name: 'Toggle AI Assistant' })).toBeTruthy(); const toggle = () => screen.getByRole('button', { name: 'Toggle AI Assistant' }); const open = async () => act(async () => fireEvent.click(toggle())); - return { toggle, open, unmount, onPanelVisibilityChange, onPanelWidthChange }; + const update = async (changes: Partial) => { + props = { ...props, ...changes }; + await act(async () => updateRoot!(props)); + }; + return { toggle, open, unmount, update, onPanelVisibilityChange, onPanelWidthChange }; }; + it('keeps the latest conversation list during StrictMode replay and ignores auth errors after close', async () => { + const obsolete = Promise.withResolvers(); + const current = Promise.withResolvers(); + const history = Promise.withResolvers(); + const host = connection(); + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockReturnValueOnce(obsolete.promise) + .mockReturnValueOnce(current.promise) + .mockReturnValueOnce(history.promise) + ); + const navigation = document.createElement('div'); + const panel = document.createElement('div'); + document.body.append(navigation, panel); + let rendered: RenderResult | undefined; + cleanup = async () => { + try { + rendered?.unmount(); + } finally { + navigation.remove(); + panel.remove(); + } + }; + await act(async () => { + rendered = render( + + () => {}} + /> + + ); + }); + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Toggle AI Assistant' }))); + await act(async () => { + current.resolve(json([conversation])); + obsolete.resolve(json([{ ...conversation, id: 'old', title: 'Obsolete conversation' }])); + }); + await selectSaved(); + expect(screen.queryByText('Obsolete conversation')).toBeNull(); + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Close AI Assistant' }))); + await act(async () => history.resolve(new Response(null, { status: 401 }))); + expect(host.onAuthError).not.toHaveBeenCalled(); + expect(screen.queryByRole('complementary', { name: 'AI Assistant workspace' })).toBeNull(); + }); + + it('preserves the draft when the host visibility callback changes and notifies the latest host on close', async () => { + mockEmptyConversations(); + const { open, update, onPanelVisibilityChange } = await setup(); + await open(); + await act(async () => fireEvent.change(question(), { target: { value: 'Keep this draft' } })); + const latestHost = vi.fn(); + await update({ onPanelVisibilityChange: latestHost }); + expect(question().value).toBe('Keep this draft'); + expect(onPanelVisibilityChange).toHaveBeenLastCalledWith(true); + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Close AI Assistant' }))); + expect(latestHost).toHaveBeenLastCalledWith(false); + expect(onPanelVisibilityChange).toHaveBeenLastCalledWith(true); + }); + + it('uses the host REST and socket by default and resolves saved and streaming paragraph references', async () => { + const paragraphId = 'paragraph_1423500782552_-1439281894'; + const unknownId = 'paragraph_1423500782553_1439281894'; + const socket = socketHarness(); + const host = connection(socket); + const onOpenParagraph = vi.fn(); + const fetch = vi.fn((url: string) => + Promise.resolve( + url.includes('/messages') + ? json({ + messages: [{ id: 'answer', role: 'assistant', content: `Saved: ${paragraphId} and ${unknownId}` }], + cursor: null + }) + : json([conversation]) + ) + ); + vi.stubGlobal('fetch', fetch); + const { open, update } = await setup({ + connection: host, + noteId: 'note/a', + paragraphs: [{ id: paragraphId, title: 'Sales query' }], + onOpenParagraph + }); + expect(fetch).not.toHaveBeenCalled(); + await open(); + await selectSaved(); + + expect(fetch.mock.calls.map(([url]) => url)).toEqual( + expect.arrayContaining([ + '/zeppelin/api/notes/note%2Fa/conversations', + '/zeppelin/api/notes/note%2Fa/conversations/saved/messages?limit=50' + ]) + ); + expect(screen.getByText(new RegExp(unknownId))).toBeTruthy(); + fireEvent.click(screen.getByRole('button', { name: 'Paragraph 1: show Sales query in the notebook' })); + expect(onOpenParagraph).toHaveBeenCalledExactlyOnceWith(paragraphId); + + await submit('Explain this query'); + expect(socket.send).toHaveBeenCalledExactlyOnceWith({ + noteId: 'note/a', + conversationId: 'saved', + content: 'Explain this query' + }); + await act(async () => { + socket.emit({ conversationId: 'saved', type: 'run.started', payload: { runId: 'run-1' } }); + socket.emit({ + conversationId: 'saved', + type: 'message.delta', + payload: { messageId: 'stream', delta: `Streaming: ${paragraphId}` } + }); + }); + expect(screen.getAllByRole('button', { name: 'Paragraph 1: show Sales query in the notebook' })).toHaveLength(2); + await act(async () => fireEvent.change(question(), { target: { value: 'Keep the next draft' } })); + const updatedCallback = vi.fn(); + await update({ + panelWidth: 400, + paragraphs: [{ id: paragraphId, title: 'Updated query' }], + onOpenParagraph: updatedCallback + }); + expect(question().value).toBe('Keep the next draft'); + expect(socket.send).toHaveBeenCalledTimes(1); + const references = screen.getAllByRole('button', { name: 'Paragraph 1: show Updated query in the notebook' }); + fireEvent.click(references[1]); + expect(updatedCallback).toHaveBeenCalledExactlyOnceWith(paragraphId); + expect(onOpenParagraph).toHaveBeenCalledTimes(1); + }); + + it('retains a confirmed conversation after its first socket submission fails and retries without creating again', async () => { + const socket = socketHarness(); + const fetch = vi.fn().mockResolvedValueOnce(json([])).mockResolvedValueOnce(json(conversation, 201)); + vi.stubGlobal('fetch', fetch); + const { open } = await setup({ connection: connection(socket) }); + await open(); + await submit('Keep this question'); + await act(async () => + socket.emit({ + conversationId: 'saved', + type: 'run.failed', + payload: { error: { status: 429 } } + }) + ); + expect(question().value).toBe('Keep this question'); + expect(screen.getByText('The assistant is busy. Try again shortly.')).toBeTruthy(); + expect(socket.consumers()).toEqual([0, 0]); + + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Retry' }))); + expect(socket.send.mock.calls).toEqual([ + [{ noteId: 'note-a', conversationId: 'saved', content: 'Keep this question' }], + [{ noteId: 'note-a', conversationId: 'saved', content: 'Keep this question' }] + ]); + expect(fetch.mock.calls.filter(([, options]) => options?.method === 'POST')).toEqual([ + ['/zeppelin/api/notes/note-a/conversations', expect.objectContaining({ method: 'POST', body: '{}' })] + ]); + }); + + it.each(['close', 'note', 'session', 'unmount'] as const)( + 'disposes the real socket consumer on %s and never shows its late answer in a new panel', + async change => { + const socket = socketHarness(); + const host = connection(socket); + vi.stubGlobal( + 'fetch', + vi.fn((_url: string, options?: RequestInit) => + Promise.resolve(options?.method === 'POST' ? json(conversation, 201) : json([])) + ) + ); + const { open, update, unmount } = await setup({ connection: host }); + await open(); + await submit('Old question'); + await act(async () => { + socket.emit({ conversationId: 'saved', type: 'run.started', payload: {} }); + socket.emit({ + conversationId: 'saved', + type: 'message.delta', + payload: { messageId: 'old', delta: 'Previous session answer' } + }); + }); + expect(screen.getByText('Previous session answer')).toBeTruthy(); + + switch (change) { + case 'close': + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Close AI Assistant' }))); + break; + case 'note': + await update({ noteId: 'note-b' }); + break; + case 'session': + await update({ connection: { ...host, sessionKey: 2 } }); + break; + case 'unmount': + await unmount(); + break; + } + expect(socket.consumers()).toEqual([0, 0]); + expect(screen.queryByText('Previous session answer')).toBeNull(); + await act(async () => + socket.emit({ + conversationId: 'saved', + type: 'message.delta', + payload: { messageId: 'old', delta: 'Late old answer' } + }) + ); + if (change !== 'unmount') await open(); + expect(screen.queryByText(/Previous session answer|Late old answer/)).toBeNull(); + expect(socket.send).toHaveBeenCalledTimes(1); + } + ); + + it.each(['close', 'note', 'session', 'unmount'] as const)( + 'ignores late authentication failures after %s while forwarding current failures to the host', + async change => { + const pending = Promise.withResolvers(); + const host = connection(); + const fetch = vi + .fn() + .mockReturnValueOnce(pending.promise) + .mockResolvedValue(new Response(null, { status: 405, headers: { Location: '/current-login' } })); + vi.stubGlobal('fetch', fetch); + const { open, update, unmount } = await setup({ connection: host }); + await open(); + switch (change) { + case 'close': + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Close AI Assistant' }))); + break; + case 'note': + await update({ noteId: 'note-b' }); + break; + case 'session': + await update({ connection: { ...host, sessionKey: 2 } }); + break; + case 'unmount': + await unmount(); + break; + } + await act(async () => + pending.resolve(new Response(null, { status: 401, headers: { Location: '/obsolete-login' } })) + ); + expect(host.onAuthError).not.toHaveBeenCalled(); + if (change !== 'unmount') await open(); + expect(vi.mocked(host.onAuthError).mock.calls).toEqual(change === 'unmount' ? [] : [[405, '/current-login']]); + expect(screen.queryByText(/obsolete-login/)).toBeNull(); + } + ); + + it.each(['close', 'note', 'session', 'unmount'] as const)('ignores late history data after %s', async change => { + const pending = Promise.withResolvers(); + const host = connection(); + const fetch = vi.fn((url: string) => + url.includes('/messages') ? pending.promise : Promise.resolve(json([conversation])) + ); + vi.stubGlobal('fetch', fetch); + const { open, update, unmount } = await setup({ connection: host }); + await open(); + await selectSaved(); + switch (change) { + case 'close': + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Close AI Assistant' }))); + break; + case 'note': + await update({ noteId: 'note-b' }); + break; + case 'session': + await update({ connection: { ...host, sessionKey: 2 } }); + break; + case 'unmount': + await unmount(); + break; + } + expect(screen.queryByRole('complementary')).toBeNull(); + if (change !== 'unmount') await open(); + await act(async () => + pending.resolve( + json({ messages: [{ id: 'old', role: 'assistant', content: 'Obsolete saved answer' }], cursor: null }) + ) + ); + expect(screen.queryByText('Obsolete saved answer')).toBeNull(); + const field = screen.queryByRole('textbox', { name: 'Question' }) as HTMLTextAreaElement | null; + expect(field?.value).toBe(change === 'unmount' ? undefined : ''); + }); + + it.each(['close', 'note', 'session', 'unmount'] as const)( + 'ignores a late successful creation after %s instead of submitting it', + async change => { + const pending = Promise.withResolvers(); + const socket = socketHarness(); + const host = connection(socket); + vi.stubGlobal( + 'fetch', + vi.fn((_url: string, options?: RequestInit) => + options?.method === 'POST' ? pending.promise : Promise.resolve(json([])) + ) + ); + const { open, update, unmount } = await setup({ connection: host }); + await open(); + await submit('Obsolete creation'); + switch (change) { + case 'close': + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Close AI Assistant' }))); + break; + case 'note': + await update({ noteId: 'note-b' }); + break; + case 'session': + await update({ connection: { ...host, sessionKey: 2 } }); + break; + case 'unmount': + await unmount(); + break; + } + if (change !== 'unmount') await open(); + await act(async () => pending.resolve(json(conversation, 201))); + expect(socket.send).not.toHaveBeenCalled(); + const field = screen.queryByRole('textbox', { name: 'Question' }) as HTMLTextAreaElement | null; + expect(field?.value).toBe(change === 'unmount' ? undefined : ''); + } + ); + + it('shows a missing conversation from real history without sending', async () => { + const socket = socketHarness(); + const fetch = vi.fn((url: string) => + Promise.resolve(url.includes('/messages') ? new Response(null, { status: 404 }) : json([conversation])) + ); + vi.stubGlobal('fetch', fetch); + const { open } = await setup({ connection: connection(socket) }); + await open(); + await selectSaved(); + expect(screen.getByText('This conversation or note no longer exists.')).toBeTruthy(); + expect((screen.getByRole('button', { name: 'Send' }) as HTMLButtonElement).disabled).toBe(true); + expect(socket.send).not.toHaveBeenCalled(); + }); + + it('keeps unknown outcome after a real history reload and a subsequent 409 without automatically resending', async () => { + const socket = socketHarness(); + vi.stubGlobal( + 'fetch', + vi.fn((url: string) => + Promise.resolve(url.includes('/messages') ? json({ messages: [], cursor: null }) : json([conversation])) + ) + ); + const { open } = await setup({ connection: connection(socket) }); + await open(); + await selectSaved(); + await submit('Uncertain question'); + await act(async () => socket.close()); + expect(screen.getByText(/The outcome of an earlier question is unknown/)).toBeTruthy(); + await act(async () => fireEvent.click(screen.getByRole('button', { name: 'Reload conversation' }))); + expect(screen.getByText(/The outcome of an earlier question is unknown/)).toBeTruthy(); + expect(socket.send).toHaveBeenCalledTimes(1); + + await submit('Explicit new submission'); + await act(async () => + socket.emit({ conversationId: 'saved', type: 'run.failed', payload: { error: { status: 409 } } }) + ); + expect(screen.getByText('This conversation is still answering. Try again when it finishes.')).toBeTruthy(); + expect(screen.getByText(/The outcome of an earlier question is unknown/)).toBeTruthy(); + expect(question().value).toBe('Explicit new submission'); + expect(socket.send).toHaveBeenCalledTimes(2); + expect(socket.consumers()).toEqual([0, 0]); + }); + it('marks the toggle as pressed while the panel is open', async () => { + mockEmptyConversations(); const { toggle, open } = await setup(); expect(toggle().getAttribute('aria-pressed')).toBe('false'); @@ -140,6 +622,7 @@ describe('AssistantWorkspace', () => { }); it('closes on Escape and returns focus to the toggle', async () => { + mockEmptyConversations(); const { toggle, open, onPanelVisibilityChange } = await setup(); await open(); @@ -150,6 +633,7 @@ describe('AssistantWorkspace', () => { }); it('returns focus to the toggle when the close button is used', async () => { + mockEmptyConversations(); const { toggle, open } = await setup(); await open(); @@ -159,6 +643,7 @@ describe('AssistantWorkspace', () => { }); it('keeps keyboard resizing within the sidebar limits', async () => { + mockEmptyConversations(); const { open, onPanelWidthChange } = await setup(); await open(); const handle = screen.getByRole('separator', { name: 'Resize AI Assistant panel' }); @@ -173,9 +658,9 @@ describe('AssistantWorkspace', () => { }); it('reports the panel as closed when the remote unmounts while open', async () => { + mockEmptyConversations(); const { open, unmount, onPanelVisibilityChange } = await setup(); await open(); - cleanup = undefined; await unmount(); expect(onPanelVisibilityChange).toHaveBeenLastCalledWith(false); diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.tsx index 964032cbc0c..ce74f4b8fe9 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.tsx +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/AssistantWorkspace.tsx @@ -10,114 +10,18 @@ * limitations under the License. */ -import { useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react'; -import { createPortal } from 'react-dom'; import { createRoot, Root } from 'react-dom/client'; -import { CloseOutlined } from '@ant-design/icons'; -import { theme } from 'antd'; import type { AssistantShellProps } from '@zeppelin/sdk'; -import { ReactErrorBoundary } from '@/components'; -import { ZeppelinThemeProvider } from '@/theme'; -import { assistantThemeStyles, ASSISTANT_PREFIX_CLS, ASSISTANT_SEED_TOKENS } from '@/shared/ui/assistant-theme'; -import { scope } from '@/shared/ui/assistant-theme/AssistantScope.css'; -import { usePanelResize } from './assistant-workspace/lib/usePanelResize'; -import { createAssistantFixture } from './assistant-workspace/fixtures/assistantFixture'; -import { ASSISTANT_HISTORY_CONVERSATIONS, FIXTURE_NOW } from './assistant-workspace/fixtures/assistantFixtureData'; -import { AssistantConversationPanel } from './assistant-workspace/ui/AssistantConversationPanel'; -import * as styles from './AssistantWorkspace.css'; +import { ReactErrorBoundary } from '@/components/paragraph/ReactErrorBoundary'; +import { ZeppelinThemeProvider } from '@/theme/ZeppelinThemeProvider'; +import { ASSISTANT_PREFIX_CLS, ASSISTANT_SEED_TOKENS } from '@/shared/ui/assistant-theme/AssistantScope'; +import { AssistantWorkspaceShell } from './assistant-workspace/ui/AssistantWorkspaceShell'; export type AssistantWorkspaceProps = AssistantShellProps; -export const AssistantWorkspace = ({ - noteId, - slots, - panelWidth, - onPanelWidthChange, - onPanelVisibilityChange, - subscribePanelClose -}: AssistantWorkspaceProps) => { - const { token } = theme.useToken(); - const themeStyles = assistantThemeStyles(token); - const [visible, setVisible] = useState(false); - const fixture = useMemo( - () => ({ - noteId, - source: createAssistantFixture({ conversations: ASSISTANT_HISTORY_CONVERSATIONS, now: FIXTURE_NOW }) - }), - [noteId] - ); - const button = useRef(null); - const navigation = slots.find(slot => slot.kind === 'navigation'); - const panel = slots.find(slot => slot.kind === 'panel'); - const { handleProps } = usePanelResize(panel?.element, panelWidth, onPanelWidthChange); - - useEffect(() => onPanelVisibilityChange(visible), [onPanelVisibilityChange, visible]); - useEffect(() => () => onPanelVisibilityChange(false), [onPanelVisibilityChange]); - useEffect(() => subscribePanelClose(() => setVisible(false)), [subscribePanelClose]); - useEffect(() => setVisible(false), [noteId]); - useLayoutEffect(() => { - const element = panel?.element; - if (!element || !visible) return; - const measure = () => - element.style.setProperty('--assistant-panel-top', `${Math.max(0, element.getBoundingClientRect().top)}px`); - measure(); - window.addEventListener('scroll', measure, true); - window.addEventListener('resize', measure); - return () => { - window.removeEventListener('scroll', measure, true); - window.removeEventListener('resize', measure); - element.style.removeProperty('--assistant-panel-top'); - }; - }, [panel?.element, visible]); - - const close = () => { - setVisible(false); - button.current?.focus(); - }; - - return ( - <> - {navigation && - createPortal( - , - navigation.element - )} - {visible && - panel && - createPortal( - , - panel.element - )} - - ); +export const AssistantWorkspace = (props: AssistantWorkspaceProps) => { + const session = JSON.stringify([props.noteId, props.connection.sessionKey]); + return ; }; export const mount = (element: HTMLElement, initialProps: AssistantWorkspaceProps) => { diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/lib/useConversationScroll.ts b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/lib/useConversationScroll.ts index c1838273a53..24f78e928c5 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/lib/useConversationScroll.ts +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/lib/useConversationScroll.ts @@ -39,11 +39,11 @@ interface ScrollState { contentHeight: number; } -const markdownBlocks = - '[data-conversation-markdown] p, [data-conversation-markdown] pre, ' + - '[data-conversation-markdown] table, [data-conversation-markdown] li, ' + - '[data-conversation-markdown] h1, [data-conversation-markdown] h2, ' + - '[data-conversation-markdown] h3, [data-conversation-markdown] blockquote'; +const contentBlocks = + '[data-assistant-content] p, [data-assistant-content] pre, ' + + '[data-assistant-content] table, [data-assistant-content] li, ' + + '[data-assistant-content] h1, [data-assistant-content] h2, ' + + '[data-assistant-content] h3, [data-assistant-content] blockquote'; const captureAnchor = ( viewport: HTMLElement, @@ -58,7 +58,7 @@ const captureAnchor = ( if (rect.bottom <= top || rect.top >= bottom) continue; // Anchor within an answer because older pages can extend that same row. - const blocks = Array.from(row.querySelectorAll(markdownBlocks)); + const blocks = Array.from(row.querySelectorAll(contentBlocks)); const block = blocks.find(element => { const bounds = element.getBoundingClientRect(); @@ -154,7 +154,7 @@ export const useConversationScroll = ({ } else if (current.anchor) { const { row, block, text, blockIndex, blockCount, firstSegment, rowOffset, blockOffset, neighbors } = current.anchor; - const blocks = stack.contains(row) ? Array.from(row.querySelectorAll(markdownBlocks)) : []; + const blocks = stack.contains(row) ? Array.from(row.querySelectorAll(contentBlocks)) : []; // Repeated paragraphs need segment identity, not text matching alone. const prepended = diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.spec.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.spec.tsx index 31b83daf965..a7762616470 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.spec.tsx +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.spec.tsx @@ -80,6 +80,60 @@ const streamingFixture = () => { }; describe('AssistantConversationPanel', () => { + it('uses the supplied renderer for saved, streaming and failed partial answers', async () => { + const harness = streamingFixture(); + render( +
{content}
} + /> + ); + await flush(); + await select('Revenue analysis'); + expect(within(messages()).getByText('Revenue increased in all three regions, led by Europe.').tagName).toBe('PRE'); + + await submit('Render this streamed answer'); + await act(async () => harness.firstDelta.resolve()); + expect(within(messages()).getByText('Only revenue A').tagName).toBe('PRE'); + + await act(async () => { + harness.secondDelta.resolve(); + harness.finish.resolve(); + }); + expect(within(messages()).getByText('Only revenue A continues').tagName).toBe('PRE'); + expect(within(messages()).getByText('Controlled run stopped.')).toBeTruthy(); + }); + + it('updates answer rendering without losing the draft or selected conversation and uses it after completion', async () => { + const source = fixture(); + const createConversation = vi.spyOn(source.api, 'createConversation'); + const { rerender } = render( +
{content}
} /> + ); + await flush(); + await select('Notebook summary'); + fireEvent.change(input(), { target: { value: 'Keep this question' } }); + + rerender( +
{content}
} + /> + ); + await flush(); + expect(input().value).toBe('Keep this question'); + expect(within(messages()).getByText('This notebook compares monthly revenue across regions.').tagName).toBe( + 'BLOCKQUOTE' + ); + await submit('Keep this question'); + + expect(createConversation).not.toHaveBeenCalled(); + expect(within(messages()).getByText(/Revenue grew \*\*12%\*\*/).tagName).toBe('BLOCKQUOTE'); + expect(within(messages()).queryByRole('table')).toBeNull(); + }); + it('opens saved conversations and restores the trigger after select, Back and Escape without closing the shell', async () => { const source = fixture(); const getMessages = vi.spyOn(source.api, 'getMessages'); @@ -637,71 +691,79 @@ describe('AssistantConversationPanel', () => { expect(run).not.toHaveBeenCalled(); }); - it('keeps a draft blocked after an initial history failure until explicit reload succeeds without resending', async () => { - const source = fixture(); - const initial = deferred(); - const retry = deferred(); - const originalGetMessages = source.api.getMessages; - const getMessages = vi - .spyOn(source.api, 'getMessages') - .mockImplementationOnce(async () => { - await initial.promise; - throw new Error('Saved history is unavailable.'); - }) - .mockImplementationOnce(async id => { - await retry.promise; - return originalGetMessages(id); - }); - const run = vi.spyOn(source, 'run'); - await mountPanel(source); - await select('Revenue analysis'); + it.each(['button', 'Enter'] as const)( + 'preserves the draft through initial history failure and retry, then submits with %s', + async method => { + const source = fixture(); + const initial = deferred(); + const retry = deferred(); + const originalGetMessages = source.api.getMessages; + const getMessages = vi + .spyOn(source.api, 'getMessages') + .mockImplementationOnce(async () => { + await initial.promise; + throw new Error('Saved history is unavailable.'); + }) + .mockImplementationOnce(async id => { + await retry.promise; + return originalGetMessages(id); + }); + const run = vi.spyOn(source, 'run'); + await mountPanel(source); + await select('Revenue analysis'); - fireEvent.change(input(), { target: { value: 'Keep this draft after the failed lookup' } }); - await act(async () => initial.resolve()); + fireEvent.change(input(), { target: { value: 'Keep this draft after the failed lookup' } }); + await act(async () => initial.resolve()); - expect(screen.getByRole('alert').textContent).toContain('Saved history is unavailable.'); - expect(input().value).toBe('Keep this draft after the failed lookup'); - expect(input().disabled).toBe(false); - expect(input().readOnly).toBe(false); - expect(send().matches(':disabled')).toBe(true); + expect(screen.getByRole('alert').textContent).toContain('Saved history is unavailable.'); + expect(input().value).toBe('Keep this draft after the failed lookup'); + expect(input().disabled).toBe(false); + expect(input().readOnly).toBe(false); + expect(send().matches(':disabled')).toBe(true); - fireEvent.change(input(), { target: { value: 'Edited question after lookup failure' } }); - fireEvent.click(send()); - fireEvent.keyDown(input(), { key: 'Enter' }); - fireEvent.submit(input().form!); - await flush(); + fireEvent.change(input(), { target: { value: 'Edited question after lookup failure' } }); + fireEvent.click(send()); + fireEvent.keyDown(input(), { key: 'Enter' }); + fireEvent.submit(input().form!); + await flush(); - expect(input().value).toBe('Edited question after lookup failure'); - expect(getMessages.mock.calls).toEqual([['c1']]); - expect(run).not.toHaveBeenCalled(); + expect(input().value).toBe('Edited question after lookup failure'); + expect(getMessages.mock.calls).toEqual([['c1']]); + expect(run).not.toHaveBeenCalled(); - fireEvent.click(screen.getByRole('button', { name: 'Reload conversation' })); - await flush(); + fireEvent.click(screen.getByRole('button', { name: 'Reload conversation' })); + await flush(); - expect(getMessages.mock.calls).toEqual([['c1'], ['c1']]); - expect(input().value).toBe('Edited question after lookup failure'); - expect(send().matches(':disabled')).toBe(true); - expect(run).not.toHaveBeenCalled(); + expect(getMessages.mock.calls).toEqual([['c1'], ['c1']]); + expect(input().value).toBe('Edited question after lookup failure'); + expect(send().matches(':disabled')).toBe(true); + expect(run).not.toHaveBeenCalled(); - fireEvent.keyDown(input(), { key: 'Enter' }); - fireEvent.submit(input().form!); - await flush(); + fireEvent.keyDown(input(), { key: 'Enter' }); + fireEvent.submit(input().form!); + await flush(); - expect(run).not.toHaveBeenCalled(); + expect(run).not.toHaveBeenCalled(); - await act(async () => retry.resolve()); + await act(async () => retry.resolve()); - expect(screen.getByText('Revenue increased in all three regions, led by Europe.')).toBeTruthy(); - expect(screen.queryByRole('alert')).toBeNull(); - expect(input().value).toBe('Edited question after lookup failure'); - expect(send().matches(':disabled')).toBe(false); - expect(run).not.toHaveBeenCalled(); + expect(screen.getByText('Revenue increased in all three regions, led by Europe.')).toBeTruthy(); + expect(screen.queryByRole('alert')).toBeNull(); + expect(input().value).toBe('Edited question after lookup failure'); + expect(send().matches(':disabled')).toBe(false); + expect(run).not.toHaveBeenCalled(); - fireEvent.click(send()); - await flush(); + if (method === 'button') fireEvent.click(send()); + else fireEvent.keyDown(input(), { key: 'Enter' }); + await flush(); - expect(run).toHaveBeenCalledExactlyOnceWith('c1', 'Edited question after lookup failure', expect.any(AbortSignal)); - }); + expect(run).toHaveBeenCalledExactlyOnceWith( + 'c1', + 'Edited question after lookup failure', + expect.any(AbortSignal) + ); + } + ); it('confirms deletion of an active conversation and lets the API decide the result', async () => { const harness = streamingFixture(); diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.tsx index a0b68e71613..617ccff15b2 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.tsx +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantConversationPanel.tsx @@ -15,6 +15,7 @@ import { AssistantAnswerError } from '@/entities/assistant/ui/AssistantAnswerErr import { AssistantComposer } from '@/entities/assistant/ui/AssistantComposer'; import { ConversationHeader } from '@/entities/assistant/ui/ConversationHeader'; import { ConversationList } from '@/entities/assistant/ui/ConversationList'; +import type { AssistantContentRenderer } from '@/entities/assistant/ui/answerView'; import { useDeleteConfirmation } from '../lib/useDeleteConfirmation'; import type { AssistantDataSource } from '../model/assistantDataSource'; import { useAssistantPanel } from '../model/useAssistantPanel'; @@ -25,11 +26,12 @@ import * as styles from './AssistantConversationPanel.css'; export interface AssistantConversationPanelProps { source: AssistantDataSource; now?: number; + renderAnswer?: AssistantContentRenderer; } const ignoreSearchChange = () => undefined; -export const AssistantConversationPanel = ({ source, now }: AssistantConversationPanelProps) => { +export const AssistantConversationPanel = ({ source, now, renderAnswer }: AssistantConversationPanelProps) => { const panel = useAssistantPanel(source); const { entry, send: sendQuestion, reload: reloadHistory } = panel; @@ -134,6 +136,7 @@ export const AssistantConversationPanel = ({ source, now }: AssistantConversatio visible={!listOpen} loadEarlier={panel.loadEarlier} onRetry={answerRetry} + renderAnswer={renderAnswer} />
diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantWorkspaceShell.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantWorkspaceShell.tsx new file mode 100644 index 00000000000..0767119dcea --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/AssistantWorkspaceShell.tsx @@ -0,0 +1,156 @@ +/* + * Licensed 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. + */ + +import { useEffect, useEffectEvent, useLayoutEffect, useMemo, useRef, useState } from 'react'; +import { createPortal } from 'react-dom'; +import CloseOutlined from '@ant-design/icons/CloseOutlined'; +import { theme } from 'antd'; +import type { AssistantShellProps } from '@zeppelin/sdk'; +import { assistantThemeStyles } from '@/shared/ui/assistant-theme/AssistantScope'; +import { createAssistantConversationApi, socketRun } from '@/entities/assistant/model/assistantTransport'; +import { scope } from '@/shared/ui/assistant-theme/AssistantScope.css'; +import { usePanelResize } from '../lib/usePanelResize'; +import type { AssistantDataSource } from '../model/assistantDataSource'; +import { AssistantConversationPanel } from './AssistantConversationPanel'; +import { createNotebookAnswerRenderer } from './createNotebookAnswerRenderer'; +import * as styles from '../../AssistantWorkspace.css'; + +const OpenConversationPanel = ({ + noteId, + connection, + paragraphs, + onOpenParagraph +}: Pick) => { + const active = useRef(false); + const [dataSource] = useState(() => ({ + api: createAssistantConversationApi(connection.apiBase, noteId, (status, location) => { + if (active.current) { + connection.onAuthError(status, location); + } + }), + run: (conversationId, content, signal) => socketRun(connection.socket, { noteId, conversationId, content }, signal) + })); + + // Invalidate authentication callbacks at unmount, before passive request cleanup runs. + useLayoutEffect(() => { + active.current = true; + return () => { + active.current = false; + }; + }, []); + + const renderAnswer = useMemo( + () => createNotebookAnswerRenderer(paragraphs, onOpenParagraph), + [paragraphs, onOpenParagraph] + ); + + return ; +}; + +export const AssistantWorkspaceShell = ({ + noteId, + connection, + paragraphs, + onOpenParagraph, + slots, + panelWidth, + onPanelWidthChange, + onPanelVisibilityChange, + subscribePanelClose +}: AssistantShellProps) => { + const { token } = theme.useToken(); + const themeStyles = assistantThemeStyles(token); + const [visible, setVisible] = useState(false); + const button = useRef(null); + const navigation = slots.find(slot => slot.kind === 'navigation'); + const panel = slots.find(slot => slot.kind === 'panel'); + const { handleProps } = usePanelResize(panel?.element, panelWidth, onPanelWidthChange); + const notifyVisibility = useEffectEvent(onPanelVisibilityChange); + + useEffect(() => notifyVisibility(visible), [visible]); + useEffect(() => () => notifyVisibility(false), []); + useEffect(() => subscribePanelClose(() => setVisible(false)), [subscribePanelClose]); + + useLayoutEffect(() => { + const element = panel?.element; + if (!element || !visible) return; + + const measure = () => { + element.style.setProperty('--assistant-panel-top', `${Math.max(0, element.getBoundingClientRect().top)}px`); + }; + measure(); + + window.addEventListener('scroll', measure, { capture: true, passive: true }); + window.addEventListener('resize', measure); + + return () => { + window.removeEventListener('scroll', measure, true); + window.removeEventListener('resize', measure); + element.style.removeProperty('--assistant-panel-top'); + }; + }, [panel?.element, visible]); + + const close = () => { + setVisible(false); + button.current?.focus(); + }; + + return ( + <> + {navigation + ? createPortal( + , + navigation.element + ) + : null} + {visible && panel + ? createPortal( + , + panel.element + ) + : null} + + ); +}; diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/ConversationMessages.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/ConversationMessages.tsx index a777901bb94..36ebb817cdf 100644 --- a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/ConversationMessages.tsx +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/ConversationMessages.tsx @@ -13,30 +13,28 @@ import { useMemo } from 'react'; import { AssistantAnswer } from '@/entities/assistant/ui/AssistantAnswer'; import { AssistantMessageItem } from '@/entities/assistant/ui/AssistantMessageItem'; -import { MarkdownAnswer } from '@/entities/assistant/ui/MarkdownAnswer'; +import type { AssistantContentRenderer } from '@/entities/assistant/ui/answerView'; import { useConversationScroll } from '../lib/useConversationScroll'; import type { AssistantPanelEntry } from '../model/assistantPanelState'; import * as styles from './AssistantConversationPanel.css'; -const renderMarkdown = (content: string) => ( -
- -
-); +interface ConversationMessagesProps { + scope: object; + entry: AssistantPanelEntry; + visible: boolean; + loadEarlier: () => Promise; + onRetry?: () => void; + renderAnswer?: AssistantContentRenderer; +} export const ConversationMessages = ({ scope, entry, visible, loadEarlier, - onRetry -}: { - scope: object; - entry: AssistantPanelEntry; - visible: boolean; - loadEarlier: () => Promise; - onRetry?: () => void; -}) => { + onRetry, + renderAnswer +}: ConversationMessagesProps) => { const { messages, messageKeys, localAnswerViews, answer, answerKey } = entry; const messageItems = useMemo(() => { @@ -49,7 +47,7 @@ export const ConversationMessages = ({ ) : ( )}
@@ -59,13 +57,13 @@ export const ConversationMessages = ({ if (answer) { items.push(
- +
); } return items; - }, [messages, messageKeys, localAnswerViews, answer, answerKey, onRetry]); + }, [messages, messageKeys, localAnswerViews, answer, answerKey, onRetry, renderAnswer]); const completedMessage = answer?.status === 'done' ? answer.message : undefined; const rowSegments = useMemo(() => { diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/createNotebookAnswerRenderer.spec.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/createNotebookAnswerRenderer.spec.tsx new file mode 100644 index 00000000000..c17574e30a5 --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/createNotebookAnswerRenderer.spec.tsx @@ -0,0 +1,61 @@ +/* + * Licensed 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. + */ + +import { fireEvent, render, screen } from '@testing-library/react'; +import { describe, expect, it, vi } from 'vitest'; +import { createNotebookAnswerRenderer } from './createNotebookAnswerRenderer'; + +const firstId = 'paragraph_1423500782552_-1439281894'; +const secondId = 'paragraph_1423500782553_1439281894'; + +describe('createNotebookAnswerRenderer', () => { + it.each([ + { title: ' Sales query ', name: 'Sales query' }, + { title: ' ', name: 'Paragraph 1' } + ])('renders known references with $name and leaves unknown references as text', ({ title, name }) => { + const onOpenParagraph = vi.fn(); + const renderAnswer = createNotebookAnswerRenderer([{ id: firstId, title }], onOpenParagraph); + render(renderAnswer(`See ${firstId} and ${secondId}.`)); + + const reference = screen.getByRole('button', { name: `Paragraph 1: show ${name} in the notebook` }); + expect(reference.textContent).toBe('Paragraph 1'); + expect(screen.getByText(new RegExp(secondId))).toBeTruthy(); + expect(screen.getAllByRole('button')).toHaveLength(1); + fireEvent.click(reference); + expect(onOpenParagraph).toHaveBeenCalledExactlyOnceWith(firstId); + }); + + it('uses current paragraph order, titles and navigation when the notebook changes', () => { + const previousNavigation = vi.fn(); + const content = `${firstId} and ${secondId}`; + const initialRenderer = createNotebookAnswerRenderer( + [ + { id: firstId, title: 'Old query' }, + { id: secondId, title: 'Second query' } + ], + previousNavigation + ); + const { rerender } = render(initialRenderer(content)); + fireEvent.click(screen.getByRole('button', { name: 'Paragraph 2: show Second query in the notebook' })); + expect(previousNavigation).toHaveBeenCalledExactlyOnceWith(secondId); + + const currentNavigation = vi.fn(); + const currentRenderer = createNotebookAnswerRenderer([{ id: secondId, title: 'Updated query' }], currentNavigation); + rerender(currentRenderer(content)); + + expect(screen.getByText(new RegExp(firstId))).toBeTruthy(); + expect(screen.queryByRole('button', { name: /Old query|Second query/ })).toBeNull(); + fireEvent.click(screen.getByRole('button', { name: 'Paragraph 1: show Updated query in the notebook' })); + expect(currentNavigation).toHaveBeenCalledExactlyOnceWith(secondId); + expect(previousNavigation).toHaveBeenCalledTimes(1); + }); +}); diff --git a/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/createNotebookAnswerRenderer.tsx b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/createNotebookAnswerRenderer.tsx new file mode 100644 index 00000000000..c4ca420f486 --- /dev/null +++ b/zeppelin-web-angular/projects/zeppelin-react/src/pages/assistant-workspace/ui/createNotebookAnswerRenderer.tsx @@ -0,0 +1,34 @@ +/* + * Licensed 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. + */ + +import type { AssistantParagraphRef } from '@zeppelin/sdk'; +import type { AssistantContentRenderer } from '@/entities/assistant/ui/answerView'; +import { MarkdownAnswer } from '@/entities/assistant/ui/MarkdownAnswer'; + +export const createNotebookAnswerRenderer = ( + paragraphs: readonly AssistantParagraphRef[], + onOpenParagraph: (paragraphId: string) => void +): AssistantContentRenderer => { + const references = new Map( + paragraphs.map((paragraph, index) => { + const text = `Paragraph ${index + 1}`; + return [paragraph.id, { text, name: paragraph.title?.trim() || text }] as const; + }) + ); + const getParagraphReference = (paragraphId: string) => references.get(paragraphId); + const openParagraph = (paragraphId: string) => onOpenParagraph(paragraphId); + + const renderAnswer: AssistantContentRenderer = content => ( + + ); + return renderAnswer; +}; diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/assistant-ui.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/assistant-ui.ts index 2139ce57a12..dbd60b19ed9 100644 --- a/zeppelin-web-angular/projects/zeppelin-sdk/src/assistant-ui.ts +++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/assistant-ui.ts @@ -10,6 +10,8 @@ * limitations under the License. */ +import type { AssistantSocket } from './assistant'; + /** The notebook sidebar's width: its default and its resize range. The assistant panel shares it. */ export const NOTEBOOK_SIDEBAR_WIDTH = { initial: 370, min: 280, max: 800 } as const; @@ -25,9 +27,21 @@ export interface AssistantSlot { kind: 'navigation' | 'panel'; } -/** The layout contract; transport and conversation props are added with notebook integration. */ +/** Authentication session and notebook connection supplied by the Angular host. */ +export interface AssistantConnection { + apiBase: string; + socket: AssistantSocket; + /** Changes after logout or a change of authenticated user, not ordinary ticket refreshes. */ + sessionKey: number; + onAuthError: (status: number, location: string | null) => void; +} + +/** Props passed by the notebook host to the React Assistant. */ export type AssistantShellProps = { noteId: string; + connection: AssistantConnection; + paragraphs: AssistantParagraphRef[]; + onOpenParagraph: (paragraphId: string) => void; slots: AssistantSlot[]; panelWidth: number; onPanelWidthChange: (width: number) => void; diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.spec.ts index 8538ce94d78..7e6e7572021 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.spec.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.spec.ts @@ -10,26 +10,49 @@ * limitations under the License. */ +import { HttpClient, HttpErrorResponse, type HttpEvent, type HttpHandler } from '@angular/common/http'; import { SimpleChange, type ChangeDetectorRef } from '@angular/core'; import type { NzMessageService } from 'ng-zorro-antd/message'; +import { BehaviorSubject, firstValueFrom, Subject, throwError } from 'rxjs'; import { expect, it, vi } from 'vitest'; +import { OP, type AssistantSocketEvent } from '@zeppelin/sdk'; +import { TicketService, type BaseUrlService, type MessageService } from '@zeppelin/services'; +import type { Router } from '@angular/router'; +import { AppHttpInterceptor } from '../../../../app-http.interceptor'; import { AssistantHostComponent } from './assistant-host.component'; import { AssistantSlots } from './assistant-slots'; -const createHost = () => { +const createHost = (initialized = true) => { const slots = new AssistantSlots(); const message = { error: vi.fn() }; + const events = new Subject(); + const closed = new Subject(); + const socket = { + connectedStatus: true, + send: vi.fn(), + receive: vi.fn(() => events), + closed: () => closed + }; + const ticket = { + ticket: { init: initialized, principal: initialized ? 'alice' : '' }, + ticket$: new Subject<{ init: boolean; principal: string }>(), + logout$: new BehaviorSubject(false), + handleAuthFailure: vi.fn() + }; const host = new AssistantHostComponent( slots, { markForCheck: vi.fn() } as unknown as ChangeDetectorRef, - message as unknown as NzMessageService + message as unknown as NzMessageService, + { getRestApiBase: () => '/zeppelin/api' } as BaseUrlService, + socket as unknown as MessageService, + ticket as unknown as TicketService ); host.enabled = true; host.noteId = 'note-a'; host.panelWidth = 370; host.ngOnChanges({ enabled: new SimpleChange(undefined, true, true) }); host.ngOnInit(); - return { host, slots, message }; + return { host, slots, message, socket, events, closed, ticket }; }; it('updates remote slots and width, then retires the panel when disabled', async () => { @@ -82,3 +105,199 @@ it('logs a remote failure, tells the user, and remounts on the next note', () => consoleError.mockRestore(); host.ngOnDestroy(); }); + +it('passes notebook connection and navigation ports without exposing credentials', () => { + const { host, ticket } = createHost(); + const connection = host.assistantProps!.connection; + expect(connection.apiBase).toBe('/zeppelin/api'); + expect(connection).not.toHaveProperty('ticket'); + expect(connection).not.toHaveProperty('principal'); + connection.onAuthError(401, '/login/redirect'); + expect(ticket.handleAuthFailure).toHaveBeenCalledWith(401, '/login/redirect'); + host.paragraphs = [{ id: 'paragraph-a', title: 'Current paragraph' }]; + host.ngOnChanges({ paragraphs: new SimpleChange([], host.paragraphs, false) }); + expect(host.assistantProps!.paragraphs).toBe(host.paragraphs); + const open = vi.fn(); + host.openParagraph.subscribe(open); + host.assistantProps!.onOpenParagraph('paragraph-a'); + expect(open).toHaveBeenCalledWith('paragraph-a'); + host.ngOnDestroy(); +}); + +it('adapts SDK assistant operations, refuses disconnected sends and releases listeners', () => { + const { host, socket, events, closed } = createHost(); + const port = host.assistantProps!.connection.socket; + const eventListener = vi.fn(); + const closeListener = vi.fn(); + const unsubscribe = port.subscribe(eventListener); + const unsubscribeClose = port.subscribeClose(closeListener); + const message = { noteId: 'note-a', conversationId: 'conversation-a', content: 'hello' }; + port.send(message); + expect(socket.send).toHaveBeenCalledWith(OP.ASSISTANT_SEND_MESSAGE, message); + expect(socket.receive).toHaveBeenCalledWith(OP.ASSISTANT_EVENT); + const event: AssistantSocketEvent = { conversationId: 'conversation-a', type: 'run.started', payload: {} }; + events.next(event); + expect(eventListener).toHaveBeenCalledWith(event); + closed.next(new CloseEvent('close')); + expect(closeListener).toHaveBeenCalledOnce(); + socket.connectedStatus = false; + expect(() => port.send(message)).toThrow('not connected'); + expect(socket.send).toHaveBeenCalledOnce(); + unsubscribe(); + unsubscribeClose(); + expect(events.observed).toBe(false); + expect(closed.observed).toBe(false); + host.ngOnDestroy(); +}); + +it('keeps connection identity across layout updates and ordinary ticket refreshes', async () => { + const { host, slots, ticket } = createHost(); + const connection = host.assistantProps!.connection; + host.panelWidth = 400; + host.ngOnChanges({ panelWidth: new SimpleChange(370, 400, false) }); + ticket.ticket$.next({ init: true, principal: 'alice' }); + slots.set({ element: document.createElement('div'), kind: 'panel' }); + await Promise.resolve(); + expect(host.assistantProps!.connection).toBe(connection); + host.ngOnDestroy(); +}); + +it('retires auth callbacks and socket events immediately on note transitions', async () => { + const { host, ticket, events, closed } = createHost(); + const connection = host.assistantProps!.connection; + const listener = vi.fn(); + const closeListener = vi.fn(); + connection.socket.subscribe(listener); + connection.socket.subscribeClose(closeListener); + host.retire(); + connection.onAuthError(405, null); + events.next({ conversationId: 'old', type: 'run.completed', payload: {} }); + closed.next(new CloseEvent('close')); + await Promise.resolve(); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); + expect(listener).not.toHaveBeenCalled(); + expect(closeListener).not.toHaveBeenCalled(); + expect(host.assistantProps).toBeNull(); + expect(() => connection.socket.send({ noteId: 'note-a', conversationId: 'old', content: 'stale' })).toThrow(); + host.noteId = 'note-b'; + host.ngOnChanges({ noteId: new SimpleChange('note-a', 'note-b', false) }); + expect(host.assistantProps!.connection).not.toBe(connection); + expect(host.assistantProps!.connection.sessionKey).toBe(connection.sessionKey); + connection.onAuthError(405, null); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); + host.assistantProps!.connection.onAuthError(405, null); + expect(ticket.handleAuthFailure).toHaveBeenCalledWith(405, null); + host.ngOnDestroy(); + expect(events.observed).toBe(false); + expect(closed.observed).toBe(false); +}); + +it('creates new session identities for logout and same-user relogin, not initial logout state', () => { + const { host, ticket } = createHost(); + const initial = host.assistantProps!.connection; + ticket.logout$.next(true); + expect(host.assistantProps).toBeNull(); + initial.onAuthError(405, null); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); + ticket.ticket = { init: false, principal: '' }; + ticket.logout$.next(false); + expect(host.assistantProps).toBeNull(); + ticket.ticket = { init: true, principal: 'alice' }; + ticket.ticket$.next(ticket.ticket); + const loggedIn = host.assistantProps!.connection; + expect(loggedIn.sessionKey).not.toBe(initial.sessionKey); + expect(loggedIn).not.toBe(initial); + initial.onAuthError(401, '/stale-login'); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); + host.ngOnDestroy(); + loggedIn.onAuthError(405, null); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); +}); + +it('resets the session on principal initialization or user change', () => { + const { host, ticket } = createHost(); + const initial = host.assistantProps!.connection; + ticket.ticket$.next({ init: false, principal: '' }); + expect(host.assistantProps).toBeNull(); + ticket.ticket$.next({ init: true, principal: 'bob' }); + expect(host.assistantProps!.connection.sessionKey).not.toBe(initial.sessionKey); + initial.onAuthError(405, null); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); + host.ngOnDestroy(); +}); + +it('waits for the initial ticket and does not reset on a duplicate initialized ticket', () => { + const { host, ticket } = createHost(false); + expect(host.assistantProps).toBeNull(); + ticket.ticket = { init: true, principal: 'alice' }; + ticket.ticket$.next(ticket.ticket); + const connection = host.assistantProps!.connection; + ticket.ticket$.next({ ...ticket.ticket }); + expect(host.assistantProps!.connection).toBe(connection); + host.ngOnDestroy(); +}); + +it('ignores authentication failures after the feature is disabled', () => { + const { host, ticket } = createHost(); + const connection = host.assistantProps!.connection; + host.enabled = false; + host.ngOnChanges({ enabled: new SimpleChange(true, false, false) }); + connection.onAuthError(405, null); + expect(ticket.handleAuthFailure).not.toHaveBeenCalled(); + host.ngOnDestroy(); +}); + +it('shares one real logout with an overlapping Angular 405 and completes shared ticket cleanup', async () => { + const requests: string[] = []; + const logoutResponse = new Subject>(); + const failure = new HttpErrorResponse({ status: 405, url: '/zeppelin/api/login' }); + let interceptor: AppHttpInterceptor; + const backend: HttpHandler = { + handle: request => { + requests.push(request.url); + return request.url.endsWith('/login/logout') ? logoutResponse : throwError(() => failure); + } + }; + const client = new HttpClient({ handle: request => interceptor.intercept(request, backend) }); + const baseUrl = { getRestApiBase: () => '/zeppelin/api' } as BaseUrlService; + const navigate = vi.fn(() => Promise.resolve(true)); + const ticket = new TicketService( + client, + baseUrl, + { navigate } as unknown as Router, + { success: vi.fn() } as unknown as NzMessageService + ); + ticket.ticket.init = true; + ticket.ticket.principal = 'alice'; + interceptor = new AppHttpInterceptor(ticket); + const host = new AssistantHostComponent( + new AssistantSlots(), + { markForCheck: vi.fn() } as unknown as ChangeDetectorRef, + { error: vi.fn() } as unknown as NzMessageService, + baseUrl, + {} as MessageService, + ticket + ); + host.enabled = true; + host.noteId = 'note-a'; + host.panelWidth = 370; + host.ngOnChanges({ enabled: new SimpleChange(undefined, true, true) }); + host.ngOnInit(); + const connection = host.assistantProps!.connection; + + connection.onAuthError(405, null); + expect(host.assistantProps).toBeNull(); + expect(ticket.logout$.value).toBe(true); + await expect(firstValueFrom(client.get('/zeppelin/api/notebook'))).rejects.toBe(failure); + expect(requests).toEqual(['/zeppelin/api/login/logout', '/zeppelin/api/notebook']); + connection.onAuthError(405, null); + expect(requests).toHaveLength(2); + + logoutResponse.error(failure); + expect(requests).toHaveLength(2); + expect(ticket.ticket.init).toBe(false); + expect(ticket.ticket.principal).toBe(''); + expect(ticket.logout$.value).toBe(false); + expect(navigate).toHaveBeenCalledExactlyOnceWith(['/login']); + host.ngOnDestroy(); +}); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.ts index 2ffbdfe0c2c..529bac16a4f 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/assistant/assistant-host.component.ts @@ -24,7 +24,8 @@ import { import { NzMessageService } from 'ng-zorro-antd/message'; import { Subject, takeUntil } from 'rxjs'; -import type { AssistantShellProps } from '@zeppelin/sdk'; +import { OP, type AssistantConnection, type AssistantParagraphRef, type AssistantShellProps } from '@zeppelin/sdk'; +import { BaseUrlService, MessageService, TicketService } from '@zeppelin/services'; import { AssistantSlots } from './assistant-slots'; @Component({ @@ -40,20 +41,51 @@ export class AssistantHostComponent implements OnInit, OnChanges, OnDestroy { @Input() enabled = false; @Input() noteId!: string; @Input() panelWidth!: number; + @Input() paragraphs: AssistantParagraphRef[] = []; + @Output() readonly openParagraph = new EventEmitter(); @Output() readonly panelWidthChange = new EventEmitter(); assistantProps: AssistantShellProps | null = null; failed = false; private readonly destroy$ = new Subject(); private destroyed = false; + private connection: AssistantConnection | null = null; + private sessionKey = 0; + private principal: string | null; + private loggingOut: boolean; + private retired = false; + private awaitingLogin = false; constructor( private readonly slots: AssistantSlots, private readonly cdr: ChangeDetectorRef, - private readonly nzMessageService: NzMessageService - ) {} + private readonly nzMessageService: NzMessageService, + private readonly baseUrlService: BaseUrlService, + private readonly messageService: MessageService, + private readonly ticketService: TicketService + ) { + this.loggingOut = ticketService.logout$.value; + this.awaitingLogin = this.loggingOut || !ticketService.ticket.init; + this.principal = this.loggingOut || !ticketService.ticket.init ? null : ticketService.ticket.principal; + } ngOnInit(): void { + this.ticketService.logout$.pipe(takeUntil(this.destroy$)).subscribe(loggingOut => { + if (loggingOut && !this.loggingOut) { + this.loggingOut = true; + this.awaitingLogin = true; + this.principal = null; + this.changeSession(); + } else if (!loggingOut) { + this.loggingOut = false; + this.updatePrincipal(this.ticketService.ticket); + } + }); + this.ticketService.ticket$.pipe(takeUntil(this.destroy$)).subscribe(ticket => { + this.loggingOut = false; + this.awaitingLogin = !ticket.init; + this.updatePrincipal(ticket); + }); this.slots.slots.pipe(takeUntil(this.destroy$)).subscribe(() => { queueMicrotask(() => { if (this.destroyed) return; @@ -64,25 +96,91 @@ export class AssistantHostComponent implements OnInit, OnChanges, OnDestroy { ngOnChanges(changes: SimpleChanges): void { if (changes.noteId && !changes.noteId.firstChange && changes.noteId.previousValue !== this.noteId) { - this.slots.requestPanelClose(); + this.retire(); this.failed = false; } if (!this.enabled) { - this.slots.setPanelOpen(false); - this.assistantProps = null; + this.retire(); return; } + if (changes.noteId || changes.enabled) this.retired = false; if (changes.enabled) this.failed = false; this.refreshProps(); } ngOnDestroy(): void { this.destroyed = true; - this.slots.setPanelOpen(false); + this.retire(); this.destroy$.next(); this.destroy$.complete(); } + /** Invalidates this note's connection before the next note has loaded. */ + retire(): void { + this.retired = true; + this.connection = null; + this.assistantProps = null; + this.slots.requestPanelClose(); + this.slots.setPanelOpen(false); + } + + private updatePrincipal(ticket: { init: boolean; principal: string }): void { + const principal = ticket.init ? ticket.principal : null; + if (this.principal === principal) return; + this.principal = principal; + this.changeSession(); + } + + private changeSession(): void { + const retired = this.retired; + this.sessionKey++; + this.retire(); + this.retired = retired; + this.refreshProps(); + } + + private createConnection(): AssistantConnection { + const current = (): boolean => !this.destroyed && this.enabled && this.connection === connection; + const connection: AssistantConnection = { + apiBase: this.baseUrlService.getRestApiBase(), + sessionKey: this.sessionKey, + onAuthError: (status, location) => { + if (current()) this.ticketService.handleAuthFailure(status, location); + }, + socket: { + send: message => { + if (!current() || !this.messageService.connectedStatus) { + throw new Error('Assistant WebSocket is not connected.'); + } + this.messageService.send(OP.ASSISTANT_SEND_MESSAGE, message); + }, + subscribe: listener => { + const subscription = this.messageService + .receive(OP.ASSISTANT_EVENT) + .pipe(takeUntil(this.destroy$)) + .subscribe(event => { + if (current()) listener(event); + }); + return () => subscription.unsubscribe(); + }, + subscribeClose: listener => { + const subscription = this.messageService + .closed() + .pipe(takeUntil(this.destroy$)) + .subscribe(() => { + if (current()) listener(); + }); + return () => subscription.unsubscribe(); + } + } + }; + return connection; + } + + private readonly onOpenParagraph = (paragraphId: string): void => { + this.openParagraph.emit(paragraphId); + }; + private readonly onPanelVisibilityChange = (visible: boolean): void => { this.slots.setPanelOpen(visible); }; @@ -105,9 +203,14 @@ export class AssistantHostComponent implements OnInit, OnChanges, OnDestroy { }; private refreshProps(): void { - if (!this.enabled || !this.noteId) return; + if (!this.enabled || !this.noteId || this.destroyed || this.loggingOut || this.retired) return; + if (this.awaitingLogin) return; + this.connection ??= this.createConnection(); this.assistantProps = { noteId: this.noteId, + connection: this.connection, + paragraphs: this.paragraphs, + onOpenParagraph: this.onOpenParagraph, slots: this.slots.slots.value, panelWidth: this.panelWidth, onPanelWidthChange: this.onPanelWidthChange, diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.html b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.html index f966cee5526..bcd4b2b5757 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.html +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.html @@ -118,6 +118,8 @@ @if (!viewOnly && !revisionView) { ({ NotebookParagraphComponent: class NotebookParagraphComponent {} @@ -40,7 +42,11 @@ const createComponent = () => { receive: vi.fn(() => NEVER), receiveEnvelope: vi.fn(() => NEVER), consumeLocalAddFocusMsgId: vi.fn(() => false), - getInterpreterBindings: vi.fn() + getInterpreterBindings: vi.fn(), + getNote: vi.fn(), + listRevisionHistory: vi.fn(), + connectedStatus: true, + connectedStatus$: new Subject() }; const router = { @@ -48,26 +54,33 @@ const createComponent = () => { } as unknown as Router; const ngZService = new NgZService(); const routeParams: Record = { noteId: 'note-b' }; + const params = new BehaviorSubject({ ...routeParams }); + const queryParamMap = new BehaviorSubject(convertToParamMap({})); const component = new NotebookComponent( messageService as unknown as MessageService, ngZService, { + params, + queryParamMap, snapshot: { - params: routeParams + params: routeParams, + queryParamMap: queryParamMap.value } } as unknown as ActivatedRoute, { markForCheck: vi.fn() } as unknown as ChangeDetectorRef, {} as NoteStatusService, - {} as NoteVarShareService, + { clear: vi.fn() } as unknown as NoteVarShareService, {} as TicketService, { getPermissions: () => NEVER } as unknown as SecurityService, router, { setTitle: vi.fn() } as unknown as Title, { updateMonacoTheme: vi.fn() } as unknown as ThemeService, - {} as ReactFeatureService + { + isEnabled: (feature, map) => map?.get(feature === 'assistantPanel' ? 'reactAssistant' : 'reactFooter') === 'true' + } as ReactFeatureService ); return { @@ -75,7 +88,9 @@ const createComponent = () => { router, messageService, ngZService, - routeParams + routeParams, + params, + queryParamMap }; }; @@ -320,4 +335,48 @@ describe('NotebookComponent', () => { ngZService.setContextValue('controlValue', 'live-model', 'A', false, binding); expect(recreatedOutput.controlValue).toBe('live-model'); }); + + it('selects referenced paragraphs and reissues scrolling when the same paragraph is activated again', () => { + const { component, router } = createComponent(); + component.note = { + id: 'note-b', + config: {}, + paragraphs: [{ id: 'A', config: {} }] + } as NonNullable; + const scrollIfNeeded = vi.fn(); + component.listOfNotebookParagraphComponent = new QueryList(); + component.listOfNotebookParagraphComponent.reset([ + { paragraph: { id: 'A' }, scrollIfNeeded } as unknown as NotebookParagraphComponent + ]); + component.openAssistantParagraph('A'); + expect(component.selectId).toBe('A'); + expect(component.scrolledId).toBe('A'); + expect(scrollIfNeeded).not.toHaveBeenCalled(); + component.openAssistantParagraph('A'); + expect(scrollIfNeeded).toHaveBeenCalledOnce(); + component.openAssistantParagraph('missing'); + expect(component.selectId).toBe('A'); + expect(router.navigate).not.toHaveBeenCalled(); + }); + + it('retires the assistant before the next route note loads and ignores old NOTE replies', () => { + const { component, params, routeParams, messageService } = createComponent(); + const oldNote = { + id: 'note-b', + config: {}, + paragraphs: [] + } as NonNullable; + component.note = oldNote; + const retire = vi.fn(); + component.assistantHost = { retire } as unknown as AssistantHostComponent; + component.ngOnInit(); + routeParams.noteId = 'note-c'; + params.next({ ...routeParams }); + expect(retire).toHaveBeenCalledOnce(); + expect(component.note).toBeUndefined(); + expect(messageService.getNote).toHaveBeenLastCalledWith('note-c'); + component.getNote({ op: OP.NOTE, data: { note: oldNote } }); + expect(component.note).toBeUndefined(); + component.ngOnDestroy(); + }); }); diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts index 0e3cce4244a..f2daa2ebd70 100644 --- a/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts +++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/notebook.component.ts @@ -18,6 +18,7 @@ import { OnDestroy, OnInit, QueryList, + ViewChild, ViewChildren } from '@angular/core'; import { Title } from '@angular/platform-browser'; @@ -54,6 +55,7 @@ import { import { scrollIntoViewIfNeeded } from '@zeppelin/utility'; import { NotebookParagraphComponent } from './paragraph/paragraph.component'; import { AssistantSlots } from './assistant/assistant-slots'; +import { AssistantHostComponent } from './assistant/assistant-host.component'; @Component({ selector: 'zeppelin-notebook', @@ -65,6 +67,7 @@ import { AssistantSlots } from './assistant/assistant-slots'; }) export class NotebookComponent extends MessageListenersManager implements OnInit, AfterViewInit, OnDestroy { @ViewChildren(NotebookParagraphComponent) listOfNotebookParagraphComponent!: QueryList; + @ViewChild(AssistantHostComponent) assistantHost?: AssistantHostComponent; private destroy$ = new Subject(); private searchTerm = ''; note?: Exclude; @@ -95,6 +98,7 @@ export class NotebookComponent extends MessageListenersManager implements OnInit if (isNil(note)) { this.router.navigate(['/']).then(); } else { + if (note.id !== this.activatedRoute.snapshot.params.noteId) return; const { paragraphId } = this.activatedRoute.snapshot.params; const nextNote = paragraphId ? this.cleanParagraphExcept(note, paragraphId) : note; this.removeObsoleteParagraphBindings(nextNote); @@ -343,6 +347,18 @@ export class NotebookComponent extends MessageListenersManager implements OnInit this.scrolledId = id; } + openAssistantParagraph(id: string): void { + if (!this.note?.paragraphs.some(paragraph => paragraph.id === id)) return; + const alreadyScrolled = this.scrolledId === id; + this.onParagraphSelect(id); + this.onParagraphScrolled(id); + // Angular does not rerun an unchanged boolean input when the same link is clicked again. + if (alreadyScrolled) { + this.listOfNotebookParagraphComponent?.find(paragraph => paragraph.paragraph.id === id)?.scrollIfNeeded(); + } + this.cdr.markForCheck(); + } + onSelectAtIndex(index: number) { if (!this.note) { throw new Error(`"note" is not defined. Please check if note data is loaded before calling this method.`); @@ -487,7 +503,17 @@ export class NotebookComponent extends MessageListenersManager implements OnInit this.useReactAssistant = this.reactFeature.isEnabled('assistantPanel', data); this.cdr.markForCheck(); }); - this.activatedRoute.params.pipe(takeUntil(this.destroy$), distinctUntilKeyChanged('noteId')).subscribe(() => { + this.activatedRoute.params.pipe(takeUntil(this.destroy$), distinctUntilKeyChanged('noteId')).subscribe(params => { + if (this.note && this.note.id !== params.noteId) { + this.assistantHost?.retire(); + this.saveNote(); + this.killSaveTimer(); + this.removeObsoleteParagraphBindings(); + this.note = undefined; + this.selectId = null; + this.scrolledId = null; + this.cdr.markForCheck(); + } this.noteVarShareService.clear(); }); this.activatedRoute.params.pipe(takeUntil(this.destroy$)).subscribe(param => {