diff --git a/backend/src/agent/app.py b/backend/src/agent/app.py index 669fb1ab..15a48f81 100644 --- a/backend/src/agent/app.py +++ b/backend/src/agent/app.py @@ -55,6 +55,33 @@ from src.core.logger import logger JWT_SECRET = os.environ["JWT_SECRET"] # @INVARIANT JWT_SECRET must be set in environment — crash-early, no default fallback MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024 # 10 MB + +def _now_iso() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +async def _build_agent_context(env_id: str | None) -> str: + """Build hidden runtime context for the LLM without storing it as user text.""" + parts = [ + "[RUNTIME CONTEXT]", + f"Current datetime: {_now_iso()}", + "If the user asks to start/run something without an explicit start time, use Current datetime as start_time.", + "If the user gives a duration such as '15 minutes' or '2 hours', compute end_time from Current datetime.", + ] + if env_id: + parts.append(f"Current environment_id: {env_id}") + dashboards = await prefetch_dashboards(env_id) + if dashboards: + parts.extend([ + "", + "[PRE-FETCHED DASHBOARDS]", + dashboards, + "[/PRE-FETCHED DASHBOARDS]", + "Use the pre-fetched dashboards directly for dashboard name/id resolution.", + ]) + parts.append("[/RUNTIME CONTEXT]") + return "\n".join(parts) + # ── Session state ─────────────────────────────────────────────── # In-memory per-user lock (keyed by user_id) _user_locks: dict[str, bool] = {} @@ -262,6 +289,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio text = text[:last_sentence_end + 1] + "\n[...truncated]" else: text = truncated + "\n[...truncated]" + visible_user_text = text # ── File upload ── file_storage_path: str | None = None @@ -285,6 +313,11 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio parsed = parse_upload(file_obj) text = f"{text}\n\n--- Uploaded file content ---\n{parsed}" + runtime_context = await _build_agent_context(env_id) + agent_text = f"{visible_user_text}\n\n{runtime_context}" + if text != visible_user_text: + agent_text = f"{text}\n\n{runtime_context}" + # ── Yield file metadata for frontend download link ── if file_storage_path and file_original_name: yield json.dumps({ @@ -338,7 +371,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio try: emitted_any = False async for event in agent.astream_events( - {"messages": [HumanMessage(content=text)]}, + {"messages": [HumanMessage(content=agent_text)]}, config=config, version="v2", ): @@ -382,7 +415,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio state = await agent.aget_state(config) if getattr(state, "next", None): emitted_any = True - yield confirmation_payload(conv_id, state, text) + yield confirmation_payload(conv_id, state, visible_user_text) return elif not emitted_any: yield json.dumps({ @@ -408,7 +441,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio "retryable": True, }, }) - await save_conversation(conv_id, text, user_id, assistant_text="") + await save_conversation(conv_id, visible_user_text, user_id, assistant_text="") return except (APITimeoutError, httpx.ReadTimeout) as exc: @@ -426,7 +459,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio "retryable": True, }, }) - await save_conversation(conv_id, text, user_id, assistant_text="") + await save_conversation(conv_id, visible_user_text, user_id, assistant_text="") return except AuthenticationError as exc: @@ -444,7 +477,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio "retryable": False, }, }) - await save_conversation(conv_id, text, user_id, assistant_text="") + await save_conversation(conv_id, visible_user_text, user_id, assistant_text="") return except OutputParserException as e: @@ -472,7 +505,7 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio try: state = await agent.aget_state(config) if getattr(state, "next", None): - yield confirmation_payload(conv_id, state, text) + yield confirmation_payload(conv_id, state, visible_user_text) return except Exception: pass @@ -480,12 +513,12 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio "content": f"❌ Ошибка: {exc}", "metadata": {"type": "error", "code": "PROCESSING_ERROR", "detail": str(exc)}, }) - await save_conversation(conv_id, text, user_id, assistant_text="".join(assistant_parts)) + await save_conversation(conv_id, visible_user_text, user_id, assistant_text="".join(assistant_parts)) return - await save_conversation(conv_id, text, user_id, assistant_text="".join(assistant_parts)) + await save_conversation(conv_id, visible_user_text, user_id, assistant_text="".join(assistant_parts)) # Fire-and-forget: generate LLM title in background (best-effort) - asyncio.create_task(generate_llm_title(conv_id, text)) + asyncio.create_task(generate_llm_title(conv_id, visible_user_text)) logger.reflect( "Agent handler completed", payload={"conv_id": conv_id, "assistant_len": len("".join(assistant_parts))}, diff --git a/backend/src/agent/langgraph_setup.py b/backend/src/agent/langgraph_setup.py index 1936c91d..f14a7bc4 100644 --- a/backend/src/agent/langgraph_setup.py +++ b/backend/src/agent/langgraph_setup.py @@ -192,7 +192,12 @@ async def create_agent( "You handle all intent detection — multi-intent queries, negations (\"don't run\"), " "synonyms (\"панели\" = \"дашборды\"), and typos are your responsibility. " "Call the right tool(s) for the job. If data is already provided in context, " - "use it directly rather than calling redundant tools." + "use it directly rather than calling redundant tools. " + "For maintenance requests, use the RUNTIME CONTEXT current datetime when the user says " + "\"start\", \"run\", \"now\", \"запусти\", or \"сейчас\" without an explicit start time. " + "Convert user durations into end_time. Do not ask for ISO datetime in that case. " + "If a user asks for dashboard maintenance, resolve the dashboard from provided context or tools, " + "then infer affected tables when possible; ask for table names only after resolution fails." ) if env_id: prompt += f"\n\nCurrent environment: '{env_id}'. When calling tools that accept env_id, use this value." diff --git a/backend/tests/test_agent/test_app.py b/backend/tests/test_agent/test_app.py index 09480e3d..689a23c1 100644 --- a/backend/tests/test_agent/test_app.py +++ b/backend/tests/test_agent/test_app.py @@ -265,6 +265,41 @@ class TestAgentHandler: token_data = json.loads(results[0]) assert token_data["metadata"]["type"] == "stream_token" + @pytest.mark.asyncio + async def test_normal_send_hides_runtime_context_from_saved_history(self, mock_request): + from src.agent.app import agent_handler + mock_chunk = MagicMock() + mock_chunk.content = "ok" + agent = _make_agent_mock([ + {"event": "on_chat_model_stream", "data": {"chunk": mock_chunk}}, + ]) + save_mock = AsyncMock() + + with patch("src.agent.app.create_agent", return_value=agent), \ + patch("src.agent.app.get_all_tools", return_value=[]), \ + patch("src.agent.app.prefetch_dashboards", AsyncMock(return_value="Available dashboards\n- USA Births Names (id: 15)")), \ + patch("src.agent.app.generate_llm_title", AsyncMock()), \ + patch("src.agent.app.save_conversation", save_mock): + results = [r async for r in agent_handler( + "Запусти обслуживание дашборда USA на 15 минут", + [], + mock_request, + "conv-runtime", + None, + None, + None, + "ss-dev", + )] + + assert len(results) == 1 + messages_arg = agent.astream_events.call_args.args[0]["messages"] + llm_text = messages_arg[0].content + assert "[RUNTIME CONTEXT]" in llm_text + assert "Current datetime:" in llm_text + assert "Available dashboards" in llm_text + save_mock.assert_called_once() + assert save_mock.call_args.args[1] == "Запусти обслуживание дашборда USA на 15 минут" + @pytest.mark.asyncio async def test_tool_events_yielded(self, mock_request): from src.agent.app import agent_handler diff --git a/frontend/src/lib/components/agent/AgentChat.svelte b/frontend/src/lib/components/agent/AgentChat.svelte index 3180d9ba..a61f9931 100644 --- a/frontend/src/lib/components/agent/AgentChat.svelte +++ b/frontend/src/lib/components/agent/AgentChat.svelte @@ -36,6 +36,7 @@ let messagesContainer: HTMLDivElement | undefined = $state(); let pendingFile: File | null = $state(null); let isDragOver = $state(false); + let isDiagnosticsMenuOpen = $state(false); let dragCounter = 0; const ALLOWED_FILE_TYPES = [".pdf", ".xlsx", ".json", ".csv", ".txt", ".png", ".jpeg", ".jpg"]; @@ -268,6 +269,17 @@ pendingFile = file; } + function closeDiagnosticsMenu() { + isDiagnosticsMenuOpen = false; + } + + function handleDiagnosticsKeydown(e: KeyboardEvent) { + if (e.key === "Escape") { + e.preventDefault(); + closeDiagnosticsMenu(); + } + } + function formatFileSize(bytes: number): string { if (bytes < 1024) return bytes + " B"; if (bytes < 1024 * 1024) return (bytes / 1024).toFixed(1) + " KB"; @@ -286,6 +298,13 @@ return "border-border bg-surface-muted text-text-muted"; } + function processStepClasses(state: string): string { + if (state === "done") return "border-success bg-success-light text-success"; + if (state === "active") return "border-warning bg-warning-light text-warning"; + if (state === "blocked") return "border-destructive-ring bg-destructive-light text-destructive"; + return "border-border bg-surface-page text-text-subtle"; + } + async function copyDebugInfo() { const lastMsg = model.messages.at(-1); const payload = { @@ -380,29 +399,57 @@
- - +
+ + {#if isDiagnosticsMenuOpen} + + + {/if} +
{#if model.streamingState === "streaming"} + + + +
+ -

{model.error}

- {/if} diff --git a/frontend/src/lib/models/AgentChatModel.svelte.ts b/frontend/src/lib/models/AgentChatModel.svelte.ts index db56aff1..64560d08 100644 --- a/frontend/src/lib/models/AgentChatModel.svelte.ts +++ b/frontend/src/lib/models/AgentChatModel.svelte.ts @@ -57,6 +57,7 @@ export interface AgentChatModelOptions { export type AgentPhase = "ready" | "thinking" | "tooling" | "confirming" | "failed" | "offline"; export type AgentPhaseTone = "success" | "warning" | "destructive" | "muted"; export type ConfirmationRisk = "read" | "write" | "unknown"; +export type AgentProcessStepState = "done" | "active" | "blocked" | "idle"; export interface AgentQuickAction { id: "dashboards" | "migration" | "tasks" | "health"; @@ -66,6 +67,12 @@ export interface AgentQuickAction { prompt: string; } +export interface AgentProcessStep { + id: "context" | "reasoning" | "tools" | "confirmation" | "result"; + label: string; + state: AgentProcessStepState; +} + export class AgentChatModel { // ── Atoms ────────────────────────────────────────────────────── messages: AgentMessage[] = $state([]); @@ -185,6 +192,63 @@ export class AgentChatModel { ? (this.conversationListItems.find((c) => c.id === this.currentConversationId)?.title || "Агент-чат") : "Агент-чат", ); + hasSilentStream = $derived( + this.streamingState === "streaming" && + this.partialText.length === 0 && + this.partialTokens.length === 0 && + this.activeToolCalls.length === 0, + ); + userFacingStatusTitle = $derived.by(() => { + if (this.connectionState !== "connected") return "Нет соединения с агентом"; + if (this.streamingState === "awaiting_confirmation") return "Требуется подтверждение"; + if (this.streamingState === "error") return "Агент остановился"; + if (this.activeToolCalls.some((tool) => tool.status === "executing")) return "Выполняется инструмент"; + if (this.hasSilentStream) return "Ожидаю первый ответ"; + if (this.streamingState === "streaming") return "Агент отвечает"; + return "Готов к задаче"; + }); + userFacingStatusDetail = $derived.by(() => { + if (this.connectionState !== "connected") return "Переподключитесь перед отправкой новой команды."; + if (this.streamingState === "awaiting_confirmation") return "Проверьте действие и параметры перед продолжением."; + if (this.streamingState === "error") return this.error || "Можно повторить последний запрос или начать новый диалог."; + if (this.activeToolCalls.some((tool) => tool.status === "executing")) return "Действие уже передано инструменту, дождитесь результата."; + if (this.hasSilentStream) return "Если upstream LLM не начнет отвечать, чат покажет восстановление вместо бесконечного ожидания."; + if (this.streamingState === "streaming") return "Ответ поступает по стриму."; + if (this.envId) return `Контекст готов: окружение ${this.envId}.`; + return "Выберите окружение перед задачами, которые работают с Superset."; + }); + processSteps = $derived.by((): AgentProcessStep[] => { + const hasTools = this.activeToolCalls.length > 0 || this.messages.some((msg) => (msg.toolCalls || []).length > 0); + const hasAssistantResult = this.messages.some((msg) => msg.role === "assistant" && msg.text.trim().length > 0); + const failed = this.streamingState === "error"; + return [ + { + id: "context", + label: this.envId ? `Контекст: ${this.envId}` : "Контекст", + state: this.envId ? "done" : failed ? "blocked" : "idle", + }, + { + id: "reasoning", + label: this.hasSilentStream ? "Ожидание LLM" : "Понимание задачи", + state: failed ? "blocked" : this.streamingState === "streaming" && !hasTools ? "active" : hasAssistantResult || hasTools ? "done" : "idle", + }, + { + id: "tools", + label: "Инструменты", + state: this.activeToolCalls.some((tool) => tool.status === "executing") ? "active" : hasTools ? "done" : "idle", + }, + { + id: "confirmation", + label: "Подтверждение", + state: this.streamingState === "awaiting_confirmation" ? "active" : this.pendingThreadId ? "done" : "idle", + }, + { + id: "result", + label: "Результат", + state: failed ? "blocked" : this.streamingState === "idle" && hasAssistantResult ? "done" : "idle", + }, + ]; + }); agentPhase = $derived.by((): AgentPhase => { if (this.connectionState !== "connected") return "offline"; if (this.streamingState === "awaiting_confirmation") return "confirming"; @@ -312,9 +376,11 @@ export class AgentChatModel { normalizeConversationTitle(rawTitle: unknown): string { const title = String(rawTitle ?? "") // Strip file upload markers (fallback for old conversations) - .replace(/\n--- Uploaded file content ---[\s\S]*$/i, "") + .replace(/(?:^|\n)--- Uploaded file content ---[\s\S]*$/i, "") // Strip pre-fetched data blocks .replace(/\s*\[PRE-FETCHED DATA[\s\S]*$/i, "") + .replace(/^✅\s*/, "") + .replace(/^(list|get|show|check|search)_\w+$/i, "Системное действие") // Strip trailing "Available dashboards..." suffix .replace(/\s+Available(?:\s+\S+)*$/i, "") // Collapse whitespace @@ -323,6 +389,25 @@ export class AgentChatModel { return title || "Новый диалог"; } + cleanVisibleMessageText(rawText: unknown): string { + return String(rawText ?? "") + .replace(/(?:^|\n)--- Uploaded file content ---[\s\S]*$/i, "") + .replace(/\n?\[PRE-FETCHED DATA[\s\S]*?```/i, "") + .replace(/\n?\[PRE-FETCHED DATA[\s\S]*?<\/pre>/i, "") + .replace(/\n?\[PRE-FETCHED DATA[\s\S]*$/i, "") + .replace(/\n?\[\/PRE-FETCHED DATA\][\s\S]*$/i, "") + .trim(); + } + + async retryLastUserMessage(): Promise { + const lastUserMessage = [...this.messages].reverse().find((message) => message.role === "user"); + const text = this.cleanVisibleMessageText(lastUserMessage?.text || ""); + if (!text) return; + this.streamingState = "idle"; + this.error = null; + await this.sendMessage(text); + } + async sendMessage(text: string, files?: File[]): Promise { // Sync reactive values from the page component this.onBeforeSend?.(); @@ -374,6 +459,7 @@ export class AgentChatModel { // processStream iterates all data events from the Gradio SSE stream. const streamDone = await Promise.race([ this.streamProcessor.processStream(this._submission, convId), + this._firstActivityTimeout(60_000), this.streamProcessor.streamCloseWatcher(this._client!, 180_000), new Promise((resolve) => setTimeout(() => resolve(false), 180_000)), ]); @@ -444,6 +530,21 @@ export class AgentChatModel { this.pendingRiskLevel = null; } + private async _firstActivityTimeout(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); + const hasActivity = + this.partialText.length > 0 || + this.partialTokens.length > 0 || + this.activeToolCalls.length > 0 || + this.streamingState === "awaiting_confirmation" || + Boolean(this.pendingThreadId); + if (!hasActivity && this.streamingState === "streaming" && !this._userCancelled) { + this.error = "Агент не начал отвечать за 60 секунд. Проверьте LLM/Gradio и попробуйте ещё раз."; + return false; + } + return new Promise(() => {}); + } + /** Commit in-progress streaming text + tool calls as an assistant message. * Called before switching conversations or creating a new one while streaming. */ private _commitStreamingPartial(): void { @@ -528,6 +629,7 @@ export class AgentChatModel { // if Gradio stream closes without a clean completion event. const streamDone = await Promise.race([ this.streamProcessor.processStream(this._submission, this.currentConversationId), + this._firstActivityTimeout(60_000), this.streamProcessor.streamCloseWatcher(this._client!, 180_000), new Promise((resolve) => setTimeout(() => resolve(false), 180_000)), ]); @@ -660,7 +762,7 @@ export class AgentChatModel { id: (msg.message_id as string) ?? (msg.id as string) ?? "", conversation_id: msg.conversation_id as string ?? "", role: msg.role as string ?? "assistant", - text: msg.text as string ?? "", + text: this.cleanVisibleMessageText(msg.text), metadata: msg.metadata as StreamMetadata, toolCalls: (msg.tool_calls as ToolCall[]) ?? [], created_at: msg.created_at as string ?? "", @@ -800,7 +902,10 @@ export class AgentChatModel { private _loadFromStorage(conversationId: string): boolean { const data = this.storage.load(conversationId); if (data) { - this.messages = data.messages; + this.messages = data.messages.map((message) => ({ + ...message, + text: this.cleanVisibleMessageText(message.text), + })); this.currentConversationId = data.conversationId; return true; } diff --git a/frontend/src/lib/models/__tests__/AgentChatModel.test.ts b/frontend/src/lib/models/__tests__/AgentChatModel.test.ts index 5f6695ef..023be046 100644 --- a/frontend/src/lib/models/__tests__/AgentChatModel.test.ts +++ b/frontend/src/lib/models/__tests__/AgentChatModel.test.ts @@ -258,6 +258,45 @@ describe("AgentChatModel — Derived State", () => { expect(model.isConversationSidebarOpen).toBe(false); }); + it("user-facing status distinguishes silent stream from normal ready state", () => { + model.envId = "ss-dev"; + expect(model.userFacingStatusTitle).toBe("Готов к задаче"); + expect(model.userFacingStatusDetail).toContain("ss-dev"); + + model.streamingState = "streaming"; + expect(model.hasSilentStream).toBe(true); + expect(model.userFacingStatusTitle).toBe("Ожидаю первый ответ"); + expect(model.processSteps.find((step) => step.id === "reasoning")?.state).toBe("active"); + + model.error = "timeout"; + model.streamingState = "error"; + expect(model.userFacingStatusTitle).toBe("Агент остановился"); + expect(model.processSteps.find((step) => step.id === "result")?.state).toBe("blocked"); + }); + + it("normalizeConversationTitle removes service-only history noise", () => { + expect(model.normalizeConversationTitle("--- Uploaded file content ---\nid,name")).toBe("Новый диалог"); + expect(model.normalizeConversationTitle("✅ list_environments")).toBe("Системное действие"); + expect(model.normalizeConversationTitle("Покажи доступные дашборды\n[PRE-FETCHED DATA]\nsecret")).toBe("Покажи доступные дашборды"); + }); + + it("retryLastUserMessage resends the last clean user request", async () => { + const sendSpy = vi.spyOn(model, "sendMessage").mockResolvedValue(undefined); + model.streamingState = "error"; + model.error = "timeout"; + model.messages = [ + { id: "1", conversation_id: "c1", role: "user", text: "first", toolCalls: [], created_at: "" }, + { id: "2", conversation_id: "c1", role: "assistant", text: "failed", toolCalls: [], created_at: "" }, + { id: "3", conversation_id: "c1", role: "user", text: "run\n[PRE-FETCHED DATA]\nsecret", toolCalls: [], created_at: "" }, + ]; + + await model.retryLastUserMessage(); + + expect(model.streamingState).toBe("idle"); + expect(model.error).toBeNull(); + expect(sendSpy).toHaveBeenCalledWith("run"); + }); + it("conversationListItems sanitizes pre-fetched prompt noise for views", () => { model.conversations = [{ id: "c1", @@ -269,6 +308,20 @@ describe("AgentChatModel — Derived State", () => { expect(model.conversationListItems[0].title).toBe("Покажи доступные дашборды"); }); + it("cleanVisibleMessageText strips hidden context from displayed messages", () => { + const raw = [ + "Запусти обслуживание дашборда USA на 15 минут", + "", + "[PRE-FETCHED DATA — use this directly, do NOT call tools]", + "Available dashboards in environment 'ss-dev'", + "- USA Births Names (id: 15)", + "[/PRE-FETCHED DATA]", + ].join("\n"); + + expect(model.cleanVisibleMessageText(raw)).toBe("Запусти обслуживание дашборда USA на 15 минут"); + expect(model.cleanVisibleMessageText("Проанализируй\n\n--- Uploaded file content ---\na,b")).toBe("Проанализируй"); + }); + it("currentConversationTitle uses sanitized conversation title", () => { model.currentConversationId = "c1"; model.conversations = [{ diff --git a/frontend/src/routes/agent/+page.svelte b/frontend/src/routes/agent/+page.svelte index 6d254d7e..17b039f3 100644 --- a/frontend/src/routes/agent/+page.svelte +++ b/frontend/src/routes/agent/+page.svelte @@ -8,13 +8,14 @@