mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 09:08:46 +02:00
Assistant: stop stale clients from resurrecting cleared threads
Root cause of the "chunks" clear bug: conversation history is server-owned, but MAF seeds a thread from whatever the client sends when no snapshot exists. Any stale tab (open across a clear or from an older session) still holds the old conversation in memory, and its next question re-persisted the whole thing — clears appeared to peel off one layer at a time. The agent middleware now drops client-sent history for threads with no stored snapshot, keeping only the last user turn (fresh conversations and empty-messages hydration requests pass through untouched). Verified live: simulated a cleared-thread stale tab, asked from it, and only the new turn was persisted. Also: clearThread surfaces DELETE failures as a toast instead of silently reloading, and conftest sets a dummy OPENAI_API_KEY so tests can import app.main. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -65,6 +65,40 @@ def _pair_orphan_tool_calls(messages: list) -> list:
|
|||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def _drop_stale_client_history(session, thread_id, messages: list) -> list:
|
||||||
|
"""Conversation history is server-owned (the AG-UI snapshot store).
|
||||||
|
|
||||||
|
A request that carries assistant/tool history for a thread with NO
|
||||||
|
stored snapshot comes from a stale client — typically a second tab that
|
||||||
|
was open when 'Nueva conversación' cleared the thread. Trusting it
|
||||||
|
would re-persist the cleared conversation ('chunks' resurrecting after
|
||||||
|
every clear). Keep only the last user turn; MAF starts the thread
|
||||||
|
fresh from there.
|
||||||
|
"""
|
||||||
|
if not thread_id or not messages:
|
||||||
|
return messages
|
||||||
|
has_history = any(m.get("role") in ("assistant", "tool") for m in messages)
|
||||||
|
if not has_history:
|
||||||
|
return messages
|
||||||
|
from sqlmodel import select
|
||||||
|
|
||||||
|
from app.models.models import ChatThreadSnapshot
|
||||||
|
|
||||||
|
row = session.exec(
|
||||||
|
select(ChatThreadSnapshot.id).where(
|
||||||
|
ChatThreadSnapshot.thread_id == str(thread_id)
|
||||||
|
)
|
||||||
|
).first()
|
||||||
|
if row is not None:
|
||||||
|
return messages # thread exists; history is legitimate
|
||||||
|
last_user = None
|
||||||
|
for i in range(len(messages) - 1, -1, -1):
|
||||||
|
if messages[i].get("role") == "user":
|
||||||
|
last_user = i
|
||||||
|
break
|
||||||
|
return messages[last_user:] if last_user is not None else []
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(app: FastAPI):
|
async def lifespan(app: FastAPI):
|
||||||
run_alembic_upgrade()
|
run_alembic_upgrade()
|
||||||
@@ -121,22 +155,27 @@ async def agent_auth_and_session(request: Request, call_next):
|
|||||||
except JWTError:
|
except JWTError:
|
||||||
return Response(status_code=401, content="Invalid token")
|
return Response(status_code=401, content="Invalid token")
|
||||||
|
|
||||||
|
session_gen = get_session()
|
||||||
|
session = next(session_gen)
|
||||||
|
token_var = set_session(session)
|
||||||
|
|
||||||
# Repair orphan tool_calls before the MAF agent sees the message history.
|
# Repair orphan tool_calls before the MAF agent sees the message history.
|
||||||
if request.method == "POST" and "application/json" in request.headers.get("content-type", ""):
|
if request.method == "POST" and "application/json" in request.headers.get("content-type", ""):
|
||||||
raw = await request.body()
|
raw = await request.body()
|
||||||
try:
|
try:
|
||||||
body = json.loads(raw)
|
body = json.loads(raw)
|
||||||
if isinstance(body.get("messages"), list):
|
if isinstance(body.get("messages"), list):
|
||||||
|
body["messages"] = _drop_stale_client_history(
|
||||||
|
session,
|
||||||
|
body.get("threadId") or body.get("thread_id"),
|
||||||
|
body["messages"],
|
||||||
|
)
|
||||||
body["messages"] = _pair_orphan_tool_calls(body["messages"])
|
body["messages"] = _pair_orphan_tool_calls(body["messages"])
|
||||||
raw = json.dumps(body).encode()
|
raw = json.dumps(body).encode()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
# Starlette caches the body; replace it so call_next sees the fixed bytes.
|
# Starlette caches the body; replace it so call_next sees the fixed bytes.
|
||||||
request._body = raw # type: ignore[attr-defined]
|
request._body = raw # type: ignore[attr-defined]
|
||||||
|
|
||||||
session_gen = get_session()
|
|
||||||
session = next(session_gen)
|
|
||||||
token_var = set_session(session)
|
|
||||||
# Browser's IANA timezone, forwarded by the BFF — lets tools resolve
|
# Browser's IANA timezone, forwarded by the BFF — lets tools resolve
|
||||||
# "today" wherever the user is (falls back to Costa Rica).
|
# "today" wherever the user is (falls back to Costa Rica).
|
||||||
tz_token = set_client_timezone(request.headers.get("x-client-timezone"))
|
tz_token = set_client_timezone(request.headers.get("x-client-timezone"))
|
||||||
|
|||||||
@@ -10,6 +10,9 @@ os.environ.setdefault(
|
|||||||
os.environ.setdefault("ADMIN_USERNAME", "testadmin")
|
os.environ.setdefault("ADMIN_USERNAME", "testadmin")
|
||||||
os.environ.setdefault("ADMIN_PASSWORD", "test-password")
|
os.environ.setdefault("ADMIN_PASSWORD", "test-password")
|
||||||
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
||||||
|
# app.main builds the MAF agent at import; the OpenAI client only needs a
|
||||||
|
# syntactically-present key.
|
||||||
|
os.environ.setdefault("OPENAI_API_KEY", "test-dummy-key")
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from sqlalchemy.pool import StaticPool
|
from sqlalchemy.pool import StaticPool
|
||||||
|
|||||||
@@ -129,3 +129,39 @@ def test_clear_thread_endpoint(monkeypatch, engine, session):
|
|||||||
|
|
||||||
assert clear_thread(session=session, _user="testadmin") == {"cleared": True}
|
assert clear_thread(session=session, _user="testadmin") == {"cleared": True}
|
||||||
assert asyncio.run(store.get(scope="default", thread_id="main")) is None
|
assert asyncio.run(store.get(scope="default", thread_id="main")) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_drop_stale_client_history(session, monkeypatch, engine):
|
||||||
|
from app.main import _drop_stale_client_history
|
||||||
|
|
||||||
|
history = [
|
||||||
|
{"role": "user", "content": "vieja pregunta"},
|
||||||
|
{"role": "assistant", "content": "vieja respuesta"},
|
||||||
|
{"role": "tool", "content": "x", "toolCallId": "t1"},
|
||||||
|
{"role": "user", "content": "nueva pregunta"},
|
||||||
|
]
|
||||||
|
|
||||||
|
# No stored snapshot for the thread → stale client: keep last user turn.
|
||||||
|
assert _drop_stale_client_history(session, "main", list(history)) == [
|
||||||
|
{"role": "user", "content": "nueva pregunta"}
|
||||||
|
]
|
||||||
|
|
||||||
|
# Fresh conversation (no assistant/tool yet) passes through.
|
||||||
|
fresh = [{"role": "user", "content": "hola"}]
|
||||||
|
assert _drop_stale_client_history(session, "main", list(fresh)) == fresh
|
||||||
|
|
||||||
|
# Empty messages (hydration trigger) pass through.
|
||||||
|
assert _drop_stale_client_history(session, "main", []) == []
|
||||||
|
|
||||||
|
# With a stored snapshot the full history is legitimate.
|
||||||
|
store = _store(monkeypatch, engine)
|
||||||
|
asyncio.run(
|
||||||
|
store.save(
|
||||||
|
scope="default",
|
||||||
|
thread_id="main",
|
||||||
|
snapshot=AGUIThreadSnapshot(
|
||||||
|
messages=[{"id": "u1", "role": "user", "content": "hola"}]
|
||||||
|
),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert _drop_stale_client_history(session, "main", list(history)) == history
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import {
|
|||||||
} from "@copilotkit/react-core/v2";
|
} from "@copilotkit/react-core/v2";
|
||||||
import { useCopilotAction } from "@copilotkit/react-core";
|
import { useCopilotAction } from "@copilotkit/react-core";
|
||||||
import { MessageSquarePlus, Sparkles } from "lucide-react";
|
import { MessageSquarePlus, Sparkles } from "lucide-react";
|
||||||
|
import { toast } from "sonner";
|
||||||
import { Button } from "@/components/ui/button";
|
import { Button } from "@/components/ui/button";
|
||||||
import ConfirmDialog from "@/components/ConfirmDialog";
|
import ConfirmDialog from "@/components/ConfirmDialog";
|
||||||
import { PageHeader } from '@/components/page-header';
|
import { PageHeader } from '@/components/page-header';
|
||||||
@@ -78,10 +79,13 @@ export default function Asistente() {
|
|||||||
const clearThread = async () => {
|
const clearThread = async () => {
|
||||||
setClearing(true);
|
setClearing(true);
|
||||||
try {
|
try {
|
||||||
await fetch("/api/v1/chat/thread", {
|
const res = await fetch("/api/v1/chat/thread", {
|
||||||
method: "DELETE",
|
method: "DELETE",
|
||||||
credentials: "include",
|
credentials: "include",
|
||||||
});
|
});
|
||||||
|
if (!res.ok) {
|
||||||
|
throw new Error(`DELETE /chat/thread → ${res.status}`);
|
||||||
|
}
|
||||||
// Full reload is the reliable reset: CopilotKit keeps chat state in
|
// Full reload is the reliable reset: CopilotKit keeps chat state in
|
||||||
// more places than agent.messages (clearing it in place left the
|
// more places than agent.messages (clearing it in place left the
|
||||||
// conversation rendered), while a fresh boot hydrates from the
|
// conversation rendered), while a fresh boot hydrates from the
|
||||||
@@ -89,7 +93,9 @@ export default function Asistente() {
|
|||||||
window.location.reload();
|
window.location.reload();
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
console.error("[asistente] clear thread failed:", err);
|
console.error("[asistente] clear thread failed:", err);
|
||||||
|
toast.error("No se pudo borrar la conversación. Intenta de nuevo.");
|
||||||
setClearing(false);
|
setClearing(false);
|
||||||
|
setConfirmClear(false);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user