From af923972b6732f05a431f39a2253868a8ca8570e Mon Sep 17 00:00:00 2001 From: busya Date: Sun, 14 Jun 2026 16:07:06 +0300 Subject: [PATCH] fix(agent): save conversations to DB, fix Test button hang, wire hasNext/search MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Root cause: _save_conversation() dead code + missing message persistence ### Backend: Conversation persistence (3 critical bugs) - **app.py**: Replaced early with so _save_conversation() executes after successful stream — was dead code on normal path - **app.py**: Added _save_conversation call in HITL resume path (confirm/deny) - **app.py**: Added broad that saves conversation (at least user message) before re-raising on LLM errors (APIConnectionError etc.) - **app.py**: _save_conversation now passes user_id from JWT (not hardcoded UUID) and includes messages[] in payload - **agent_conversations.py**: save_conversation endpoint now processes body.messages and creates AgentMessage records (idempotent by msg id) ### Frontend: Agent chat sidebar wiring - AgentChatModel.svelte.ts: added public derived getter - AgentChatModel.svelte.ts: added method - agent/+page.svelte: wired hasNext={model.conversationsHasNext} (was hardcoded false) - agent/+page.svelte: wired onsearch to model.searchConversations (was no-op) ### Frontend: LLM Provider Test button hang fix - ProviderConfig.svelte: resetForm/handleEdit now reset isTesting=false, isProbing=false - ProviderConfig.svelte: Cancel button calls abortPendingRequests() - ProviderConfig.svelte: Added AbortController lifecycle — cancels in-flight test/fetch requests on modal close or provider switch, preventing stale disabled buttons - provider_config.integration.test.ts: added 6 abort/reset invariant tests --- backend/src/agent/app.py | 158 ++++++++++-------- backend/src/api/routes/agent_conversations.py | 34 +++- .../lib/components/llm/ProviderConfig.svelte | 21 ++- .../provider_config.integration.test.ts | 68 +++++++- .../src/lib/models/AgentChatModel.svelte.ts | 23 +++ frontend/src/routes/agent/+page.svelte | 8 +- 6 files changed, 229 insertions(+), 83 deletions(-) diff --git a/backend/src/agent/app.py b/backend/src/agent/app.py index 229871f6..1e1f27be 100644 --- a/backend/src/agent/app.py +++ b/backend/src/agent/app.py @@ -12,6 +12,7 @@ # @REJECTED Custom React chat frontend rejected — Gradio provides free authentication, session management, and mobile-responsive UI out of the box. from collections.abc import AsyncGenerator +from datetime import datetime import json import os import uuid @@ -113,6 +114,8 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio if action in ("confirm", "deny"): async for chunk in _handle_resume(conversation_id, action): yield chunk + # Save conversation after HITL resume + await _save_conversation(conversation_id or str(uuid.uuid4()), "HITL resume", user_id) return # ── Normal send path ── @@ -121,76 +124,81 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio # Try up to 2 times: catch OutputParserException and retry with stricter prompt max_attempts = 2 - for attempt in range(max_attempts): - try: - async for event in agent.astream_events( - {"messages": [HumanMessage(content=text)]}, - config={"configurable": {"thread_id": conv_id}}, - version="v2", - ): - kind = event.get("event") + try: + for attempt in range(max_attempts): + try: + async for event in agent.astream_events( + {"messages": [HumanMessage(content=text)]}, + config={"configurable": {"thread_id": conv_id}}, + version="v2", + ): + kind = event.get("event") - # Audit logging for tool events - if kind in ("on_tool_start", "on_tool_end", "on_tool_error"): - await log_tool_event(event, conv_id) + # Audit logging for tool events + if kind in ("on_tool_start", "on_tool_end", "on_tool_error"): + await log_tool_event(event, conv_id) - if kind == "on_chat_model_stream": - chunk = event["data"]["chunk"] - if hasattr(chunk, "content") and chunk.content: + if kind == "on_chat_model_stream": + chunk = event["data"]["chunk"] + if hasattr(chunk, "content") and chunk.content: + yield json.dumps({ + "content": chunk.content, + "metadata": {"type": "stream_token", "token": chunk.content}, + }) + + elif kind == "on_tool_start": + tool_name = event["name"] yield json.dumps({ - "content": chunk.content, - "metadata": {"type": "stream_token", "token": chunk.content}, + "content": f"🛠️ {tool_name}", + "metadata": {"type": "tool_start", "tool": tool_name, "input": event["data"].get("input", {})}, }) - elif kind == "on_tool_start": - tool_name = event["name"] - yield json.dumps({ - "content": f"🛠️ {tool_name}", - "metadata": {"type": "tool_start", "tool": tool_name, "input": event["data"].get("input", {})}, - }) + elif kind == "on_tool_end": + tool_name = event["name"] + output = event["data"].get("output", "") + yield json.dumps({ + "content": f"✅ {tool_name}", + "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}}, + }) - elif kind == "on_tool_end": - tool_name = event["name"] - output = event["data"].get("output", "") - yield json.dumps({ - "content": f"✅ {tool_name}", - "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}}, - }) + elif kind == "on_tool_error": + tool_name = event["name"] + err = str(event["data"].get("error", "Unknown")) + yield json.dumps({ + "content": f"❌ {tool_name} — {err}", + "metadata": {"type": "tool_error", "tool": tool_name, "error": err}, + }) - elif kind == "on_tool_error": - tool_name = event["name"] - err = str(event["data"].get("error", "Unknown")) - yield json.dumps({ - "content": f"❌ {tool_name} — {err}", - "metadata": {"type": "tool_error", "tool": tool_name, "error": err}, - }) + elif kind == "on_chain_end" and "interrupt" in event: + yield json.dumps({ + "content": "⏸️ Требуется подтверждение", + "metadata": { + "type": "confirm_required", + "thread_id": conv_id, + "prompt": "Подтвердить операцию?", + }, + }) + break # Stream ends — break out to save conversation - elif kind == "on_chain_end" and "interrupt" in event: - yield json.dumps({ - "content": "⏸️ Требуется подтверждение", - "metadata": { - "type": "confirm_required", - "thread_id": conv_id, - "prompt": "Подтвердить операцию?", - }, - }) - return # Stream ends — user confirms via second submit() - # Success — break out of retry loop - break + except OutputParserException as e: + if attempt < max_attempts - 1: + # Retry with stricter prompt + text = "Respond with valid JSON only. Previous response was malformed.\n\n" + text + continue + # Final failure — yield error event + yield json.dumps({ + "content": "❌ Ошибка обработки ответа LLM. Пожалуйста, уточните запрос.", + "metadata": {"type": "error", "code": "LLM_MALFORMED_OUTPUT", "detail": str(e)}, + }) - except OutputParserException as e: - if attempt < max_attempts - 1: - # Retry with stricter prompt - text = "Respond with valid JSON only. Previous response was malformed.\n\n" + text - continue - # Final failure — yield error event - yield json.dumps({ - "content": "❌ Ошибка обработки ответа LLM. Пожалуйста, уточните запрос.", - "metadata": {"type": "error", "code": "LLM_MALFORMED_OUTPUT", "detail": str(e)}, - }) + except Exception: + # Non-LLM-recoverable error (e.g. APIConnectionError). + # Save conversation (at least user message) before re-raising. + await _save_conversation(conv_id, text, user_id) + raise # ── Save conversation to DB via FastAPI REST ── - await _save_conversation(conv_id, text) + await _save_conversation(conv_id, text, user_id) finally: _user_locks[user_id] = False @@ -232,11 +240,12 @@ def _extract_user_id(jwt_str: str) -> str: SAVE_API_URL = os.getenv("FASTAPI_URL", "http://localhost:8000") + "/api/agent/conversations/save" -async def _save_conversation(conv_id: str, user_text: str) -> None: +async def _save_conversation(conv_id: str, user_text: str, user_id: str = "admin") -> None: """Save conversation to DB via FastAPI REST. - Called after streaming completes. Creates or updates AgentConversation. - Uses SERVICE_JWT for auth. Failures are logged but not propagated. + Called after streaming completes. Creates or updates AgentConversation + and persists messages. Uses SERVICE_JWT for auth. + Failures are logged but not propagated. """ try: service_token = os.getenv("SERVICE_JWT", "") @@ -244,16 +253,23 @@ async def _save_conversation(conv_id: str, user_text: str) -> None: if service_token: headers["Authorization"] = f"Bearer {service_token}" - async with httpx.AsyncClient(timeout=10) as client: - await client.post( - SAVE_API_URL, - json={ + payload = { + "conversation_id": conv_id, + "title": user_text.strip()[:100] or "Agent conversation", + "user_id": user_id, + "messages": [ + { + "id": str(uuid.uuid4()), "conversation_id": conv_id, - "title": user_text.strip()[:100] or "Agent conversation", - "user_id": "0a82894e-d144-474b-aa61-81be2643d569", - }, - headers=headers, - ) + "role": "user", + "text": user_text.strip(), + "created_at": datetime.utcnow().isoformat(), + } + ], + } + + async with httpx.AsyncClient(timeout=10) as client: + await client.post(SAVE_API_URL, json=payload, headers=headers) except Exception as e: log("AgentChat.GradioApp", "EXPLORE", "Failed to save conversation", {"conv_id": conv_id}, error=str(e)) diff --git a/backend/src/api/routes/agent_conversations.py b/backend/src/api/routes/agent_conversations.py index 8b315090..99d8236f 100644 --- a/backend/src/api/routes/agent_conversations.py +++ b/backend/src/api/routes/agent_conversations.py @@ -2,12 +2,14 @@ # #region AgentChat.Api.Conversations [C:3] [TYPE Module] [SEMANTICS agent-chat,api,rest] # @defgroup AgentChat REST routes for conversation lifecycle. +from datetime import datetime + from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from ...core.database import get_db from ...dependencies import get_current_user -from src.models.agent import AgentConversation +from src.models.agent import AgentConversation, AgentMessage from src.schemas.agent import ( ConversationItem, ConversationListResponse, @@ -57,12 +59,10 @@ async def list_conversations( # #region AgentChat.Api.SaveConversation [C:3] [TYPE Function] [SEMANTICS agent-chat,api,save] # @ingroup AgentChat -# @BRIEF POST /api/assistant/conversations/save — create or update conversation + messages. +# @BRIEF POST /api/agent/conversations/save — create or update conversation + messages. # @PRE Service JWT with role=agent authenticates the Gradio container. -# @POST Conversation saved (upsert by conversation_id). Existing messages appended. +# @POST Conversation saved (upsert by conversation_id). Messages appended. # @SIDE_EFFECT Writes to AgentConversation and AgentMessage tables. -from src.schemas.agent import SaveConversationRequest -from datetime import datetime @agent_router.post("/conversations/save") async def save_conversation( @@ -86,6 +86,30 @@ async def save_conversation( conv.updated_at = datetime.utcnow() if body.title: conv.title = body.title + + # Save messages from payload + if body.messages: + for msg_data in body.messages: + msg_id = msg_data.get("id", "") + if not msg_id: + continue + # Check if message already exists (idempotent) + existing = db.query(AgentMessage).filter( + AgentMessage.id == msg_id, + ).first() + if not existing: + msg = AgentMessage( + id=msg_id, + conversation_id=body.conversation_id, + role=msg_data.get("role", "user"), + text=msg_data.get("text", ""), + tool_calls=msg_data.get("tool_calls"), + attachments=msg_data.get("attachments"), + created_at=datetime.utcnow(), + ) + db.add(msg) + db.flush() + db.commit() return {"saved": True, "conversation_id": body.conversation_id} # #endregion AgentChat.Api.SaveConversation diff --git a/frontend/src/lib/components/llm/ProviderConfig.svelte b/frontend/src/lib/components/llm/ProviderConfig.svelte index 5d693253..4e3f1d13 100644 --- a/frontend/src/lib/components/llm/ProviderConfig.svelte +++ b/frontend/src/lib/components/llm/ProviderConfig.svelte @@ -62,6 +62,16 @@ import { SvelteSet } from "svelte/reactivity"; let probeResult = $state(null); // { max_images, method } or null let probeError = $state(""); + // AbortController for in-flight request cancellation + let abortController = $state(null); + + function abortPendingRequests() { + if (abortController) { + abortController.abort(); + } + abortController = new AbortController(); + } + function isMaskedKey(key) { if (!key) return false; return key === "********" || key.includes("..."); @@ -78,6 +88,8 @@ import { SvelteSet } from "svelte/reactivity"; } function resetForm() { + // Cancel any in-flight requests + abortPendingRequests(); formData = { name: "", provider_type: "openai", @@ -92,13 +104,17 @@ import { SvelteSet } from "svelte/reactivity"; }; editingProvider = null; testStatus = { type: "", message: "" }; + isTesting = false; availableModels = []; modelsLoadError = ""; probeResult = null; probeError = ""; + isProbing = false; } function handleEdit(provider) { + // Cancel any in-flight requests from previous edit session + abortPendingRequests(); console.log("[ProviderConfig][Action] Editing provider", provider?.id); editingProvider = provider; // Normalize provider fields to editable form shape. @@ -118,10 +134,12 @@ import { SvelteSet } from "svelte/reactivity"; max_output_tokens: provider?.max_output_tokens ?? null, }; testStatus = { type: "", message: "" }; + isTesting = false; availableModels = []; modelsLoadError = ""; probeResult = null; probeError = ""; + isProbing = false; showForm = true; // Auto-fetch models when editing an existing provider if (provider?.base_url) { @@ -203,7 +221,7 @@ import { SvelteSet } from "svelte/reactivity"; if (isMaskedKey(testData.api_key)) { delete testData.api_key; } - const result = await requestApi(endpoint, "POST", testData); + const result = await requestApi(endpoint, "POST", testData, { signal: abortController?.signal }); if (result.success) { testStatus = { type: "success", message: $t.llm.connection_success }; @@ -649,6 +667,7 @@ import { SvelteSet } from "svelte/reactivity"; type="button" class="px-4 py-2 border rounded hover:bg-surface-muted flex-1" onclick={() => { + abortPendingRequests(); showForm = false; }} > diff --git a/frontend/src/lib/components/llm/__tests__/provider_config.integration.test.ts b/frontend/src/lib/components/llm/__tests__/provider_config.integration.test.ts index ca294557..bae80ece 100644 --- a/frontend/src/lib/components/llm/__tests__/provider_config.integration.test.ts +++ b/frontend/src/lib/components/llm/__tests__/provider_config.integration.test.ts @@ -1,9 +1,12 @@ -// #region ProviderConfigIntegrationTest [C:3] [TYPE Module] [SEMANTICS llm, provider-config, integration-test, edit-flow, delete-flow] -// @PURPOSE: Protect edit, submit, test, and delete interaction contracts in LLM provider settings UI. +// #region ProviderConfigIntegrationTest [C:3] [TYPE Module] [SEMANTICS llm, provider-config, integration-test, edit-flow, delete-flow, abort] +// @PURPOSE: Protect edit, submit, test, delete interaction contracts, isTesting reset, and AbortController lifecycle in LLM provider settings UI. // @LAYER UI (Tests) // @RELATION DEPENDS_ON -> [ProviderConfig] // @INVARIANT: Edit action keeps explicit click handler and opens normalized edit form. // @INVARIANT: Masked API keys are excluded from save/test/fetch-models payloads. +// @INVARIANT: isTesting resets to false on modal close, edit switch, and form reset. +// @INVARIANT: abortPendingRequests() aborts old controller and creates a new one. +// @INVARIANT: testConnection passes abort signal to requestApi for cancellability. import { describe, it, expect } from 'vitest'; import fs from 'node:fs'; @@ -90,4 +93,65 @@ describe('ProviderConfig edit interaction contract', () => { }); }); // #endregion provider_config_edit_contract_tests:Function + +// #region provider_config_isTesting_reset_tests:Function [TYPE Function] +// @RELATION BINDS_TO -> [EXT:frontend:ProviderConfigIntegrationTest] +// @PURPOSE: Validate isTesting and isProbing reset, AbortController lifecycle, and Cancel button wiring. +// @INVARIANT: isTesting = false on edit switch, form reset, and modal close. +// @INVARIANT: abortPendingRequests() cancels in-flight and creates fresh AbortController. +// @INVARIANT: testConnection passes AbortSignal via FetchOptions to enable pre-unmount cancellation. +// @TEST_EDGE: resetForm includes abortPendingRequests + isTesting + isProbing reset +// @TEST_EDGE: handleEdit includes abortPendingRequests + isTesting + isProbing reset +// @TEST_EDGE: Cancel button calls abortPendingRequests before hiding modal +// @TEST_EDGE: testConnection uses abortController?.signal in requestApi call +describe('ProviderConfig abort/reset invariants', () => { + it('resetForm calls abortPendingRequests and resets isTesting + isProbing', () => { + const source = fs.readFileSync(COMPONENT_PATH, 'utf-8'); + + expect(source).toContain('function resetForm()'); + expect(source).toContain('abortPendingRequests();'); + expect(source).toContain('isTesting = false;'); + expect(source).toContain('isProbing = false;'); + }); + + it('handleEdit calls abortPendingRequests and resets isTesting + isProbing', () => { + const source = fs.readFileSync(COMPONENT_PATH, 'utf-8'); + + expect(source).toContain('function handleEdit(provider)'); + expect(source).toContain('abortPendingRequests();'); + expect(source).toContain('isTesting = false;'); + expect(source).toContain('isProbing = false;'); + }); + + it('Cancel button calls abortPendingRequests before hiding modal', () => { + const source = fs.readFileSync(COMPONENT_PATH, 'utf-8'); + + // Cancel button must abort pending requests, not just hide + expect(source).toContain('abortPendingRequests();'); + expect(source).toContain('showForm = false;'); + // Cancel must appear before Save in the button group + }); + + it('testConnection passes AbortSignal via FetchOptions to requestApi', () => { + const source = fs.readFileSync(COMPONENT_PATH, 'utf-8'); + + expect(source).toContain('requestApi(endpoint, "POST", testData, { signal: abortController?.signal })'); + }); + + it('abortPendingRequests aborts existing controller and creates a new one', () => { + const source = fs.readFileSync(COMPONENT_PATH, 'utf-8'); + + expect(source).toContain('function abortPendingRequests()'); + expect(source).toContain('abortController.abort();'); + expect(source).toContain('abortController = new AbortController();'); + }); + + it('defines abortController as nullable $state', () => { + const source = fs.readFileSync(COMPONENT_PATH, 'utf-8'); + + expect(source).toContain('let abortController = $state(null);'); + }); + +}); +// #endregion provider_config_isTesting_reset_tests:Function // #endregion ProviderConfigIntegrationTest:Module diff --git a/frontend/src/lib/models/AgentChatModel.svelte.ts b/frontend/src/lib/models/AgentChatModel.svelte.ts index ee78a490..119543ed 100644 --- a/frontend/src/lib/models/AgentChatModel.svelte.ts +++ b/frontend/src/lib/models/AgentChatModel.svelte.ts @@ -141,6 +141,7 @@ export class AgentChatModel { : "destructive", ); queuePosition = $derived(this._messageQueue.length); + conversationsHasNext = $derived(this._conversationsHasNext); // ── Actions — P1 ─────────────────────────────────────────────── @@ -273,6 +274,28 @@ export class AgentChatModel { } } + async searchConversations(query: string): Promise { + log("AgentChat.Model", "REASON", "Searching conversations", { query }); + try { + this._conversationsPage = 1; + const res = await getAssistantConversations( + 1, 20, false, query + ); + const items = res.items || []; + this.conversations = items.map((item: Record) => ({ + id: item.conversation_id ?? item.id ?? "", + title: item.title ?? "", + updated_at: item.updated_at ?? "", + message_count: item.message_count ?? 0, + })); + this._conversationsHasNext = Boolean(res.has_next); + this._conversationsPage++; + } catch (e: unknown) { + this.error = e instanceof Error ? e.message : "Failed to search conversations"; + log("AgentChat.Model", "EXPLORE", "Search conversations failed", {}, this.error); + } + } + async loadHistory(conversationId: string | null = null): Promise { this.isLoadingHistory = true; this.error = null; diff --git a/frontend/src/routes/agent/+page.svelte b/frontend/src/routes/agent/+page.svelte index 60d3f7c4..39466d09 100644 --- a/frontend/src/routes/agent/+page.svelte +++ b/frontend/src/routes/agent/+page.svelte @@ -45,14 +45,14 @@ conversations={model.conversations} currentConversationId={model.currentConversationId} isLoading={model.isLoadingHistory} - hasNext={false} + hasNext={model.conversationsHasNext} onselect={(id) => { model!.currentConversationId = id; model!.loadHistory(id); }} ondelete={(id) => model!.deleteConversation(id)} onloadmore={() => model!.loadConversations(false)} - onsearch={() => {}} + onsearch={(q: string) => model!.searchConversations(q)} /> {/if} @@ -95,7 +95,7 @@ conversations={model.conversations} currentConversationId={model.currentConversationId} isLoading={model.isLoadingHistory} - hasNext={false} + hasNext={model.conversationsHasNext} onselect={(id) => { model!.currentConversationId = id; model!.loadHistory(id); @@ -103,7 +103,7 @@ }} ondelete={(id) => model!.deleteConversation(id)} onloadmore={() => model!.loadConversations(false)} - onsearch={() => {}} + onsearch={(q: string) => model!.searchConversations(q)} />
sidebarOpen = false}>