fix(agent): save conversations to DB, fix Test button hang, wire hasNext/search

## 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
2026-06-14 16:07:06 +03:00
parent 997329e2a5
commit af923972b6
6 changed files with 229 additions and 83 deletions
--- 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))
--- 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
--- 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<AbortController | null>(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;
            }}
          >
--- 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<AbortController | null>(null);');
+  });
+
+});
+// #endregion provider_config_isTesting_reset_tests:Function
 // #endregion ProviderConfigIntegrationTest:Module
--- 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<void> {
+    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<string, unknown>) => ({
+        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<void> {
    this.isLoadingHistory = true;
    this.error = null;
--- 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)}
          />
        </div>
        <div class="flex-1 bg-black/30" onclick={() => sidebarOpen = false}></div>