mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-09-10 10:12:20 +02:00
fix(history): defer full transcript hydration to model sends (#5929)
* fix(history): defer full hydration to model sends * fix(session): key hydration on real rows, fork through get_session Two regressions from the display/model-context split, both reproducible against dev. The hydration gate compared the cached transcript against the denormalized sessions.message_count column. That column drifts in normal operation — _persist_message swallows a failed insert while add_message has already appended in memory, so the next successful persist writes rows+1 — and _db_to_session re-read the same column after each reload, so the shortfall never closed. Every send, edit, delete and truncate on a warm session re-selected the whole message table: the cost this change set out to remove, relocated onto the hot path. The other direction was just as bad — a persist for an uncached session writes message_count = 0, and a stale-low counter with a partly filled cache meant no hydration at all and a silently truncated transcript for the model. sync_session_metadata now reconciles message_count against COUNT(*) on chat_messages (one indexed count inside the connection it already opens), and _db_to_session trusts the rows it just loaded. A hydrate always closes the gap, so the next read is a cache hit. fork_session read session_manager.sessions directly and never hydrated. keep_count indexes into source.history, and display pagination no longer fills that cache, so forking after a restart returned HTTP 200 with an empty conversation and no error surfaced. It goes through get_session now. _hydrate_session_history_from_db is gone with its helper: get_session is the hydration seam, and rebuilding session.history from raw rows in the display fallback overwrote the parsed multimodal content and the _db_id edit/delete keys that had just been set. Tests drive a real SessionManager over a temp DB instead of a stub that only proved the stub hydrates — both drift directions, the send path warm and cold, and a fork taken after a restart. All five fail without this change. The brittle SQL-text assertions are dropped; the page bounds are already proven by the response body. * fix(history): route pagination through canonical handler --------- Co-authored-by: Léo <leograndcontact@gmail.com>
This commit is contained in:
+41
-12
@@ -194,7 +194,12 @@ class SessionManager:
|
||||
is_important=getattr(db_session, 'is_important', False) or False,
|
||||
)
|
||||
|
||||
session.message_count = getattr(db_session, 'message_count', len(history))
|
||||
# The rows just loaded are the whole transcript, so they — not the
|
||||
# denormalized sessions.message_count column — are the truth for this
|
||||
# cached object. get_session's hydration gate compares against this
|
||||
# number; seeding it from a drifted column would ask for a reload that
|
||||
# can never close the gap.
|
||||
session.message_count = len(history)
|
||||
return session
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
@@ -398,30 +403,50 @@ class SessionManager:
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_session(self, session_id: str) -> Session:
|
||||
"""Get a session by ID, loading from DB if needed.
|
||||
"""Get a session by ID, loading complete DB history when needed.
|
||||
|
||||
Sessions seeded by `load_sessions` start with empty history. The
|
||||
first read here hydrates them with the message rows.
|
||||
Sessions seeded by ``load_sessions`` start with empty history, and a
|
||||
cached session can also become partially stale. Refresh metadata first,
|
||||
then hydrate whenever the cached transcript is short of the stored rows.
|
||||
Model-send routes enter through this method before building context,
|
||||
while paginated display history reads SQLite directly.
|
||||
|
||||
The gate compares against ``sync_session_metadata``'s reconciled count
|
||||
(the real ``chat_messages`` total), never the denormalized column, so a
|
||||
hydrate always closes the gap and the next read is a cache hit.
|
||||
"""
|
||||
if session_id not in self.sessions:
|
||||
self._load_session_from_db(session_id)
|
||||
else:
|
||||
cached = self.sessions[session_id]
|
||||
# Lazy hydrate: metadata-only entries get their messages on first read.
|
||||
if not cached.history and getattr(cached, "message_count", 0) > 0:
|
||||
self._load_session_from_db(session_id)
|
||||
|
||||
# Keep model/endpoint metadata fresh. Endpoint deletion can clear the
|
||||
# DB row while a session object is still cached in RAM.
|
||||
# DB row while a session object is still cached in RAM. Refreshing first
|
||||
# also exposes the authoritative message count before completeness is
|
||||
# checked.
|
||||
self.sync_session_metadata(session_id)
|
||||
|
||||
cached = self.sessions[session_id]
|
||||
cached_count = len(cached.history or [])
|
||||
stored_count = int(getattr(cached, "message_count", 0) or 0)
|
||||
if cached_count < stored_count:
|
||||
self._load_session_from_db(session_id)
|
||||
|
||||
# Update last_accessed
|
||||
self._touch_session(session_id)
|
||||
|
||||
return self.sessions[session_id]
|
||||
|
||||
def sync_session_metadata(self, session_id: str) -> bool:
|
||||
"""Refresh non-message session fields from the DB into the cached object."""
|
||||
"""Refresh non-message session fields from the DB into the cached object.
|
||||
|
||||
``message_count`` is reconciled against the real ``chat_messages`` rows
|
||||
rather than copied from the denormalized ``sessions.message_count``
|
||||
column. That column drifts in normal operation — ``_persist_message``
|
||||
swallows a failed insert but ``add_message`` has already appended in
|
||||
memory, so the next successful persist writes rows+1, and a persist for
|
||||
an uncached session writes 0. Hydration keys off this number: a
|
||||
drifted-high column would reload the whole transcript on every warm
|
||||
read, and a drifted-low one would leave the model a truncated one.
|
||||
"""
|
||||
session = self.sessions.get(session_id)
|
||||
if session is None:
|
||||
return False
|
||||
@@ -444,7 +469,11 @@ class SessionManager:
|
||||
session.archived = db_session.archived
|
||||
session.owner = getattr(db_session, "owner", None)
|
||||
session.is_important = getattr(db_session, "is_important", False) or False
|
||||
session.message_count = getattr(db_session, "message_count", session.message_count) or 0
|
||||
session.message_count = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.count()
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Error syncing session metadata {session_id}: {e}")
|
||||
|
||||
@@ -137,44 +137,6 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
|
||||
entry["metadata"] = meta
|
||||
return entry
|
||||
|
||||
def _db_message_metadata(m: DbChatMessage) -> Dict[str, Any]:
|
||||
meta = {}
|
||||
if m.meta_data:
|
||||
try:
|
||||
meta = json.loads(m.meta_data) or {}
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
meta = {}
|
||||
if m.timestamp and "timestamp" not in meta:
|
||||
meta["timestamp"] = m.timestamp.isoformat() + "Z"
|
||||
return meta
|
||||
|
||||
def _hydrate_session_history_from_db(session_id: str, rows: list[DbChatMessage]) -> None:
|
||||
"""Rebuild in-memory context from raw DB rows after a history load.
|
||||
|
||||
The browser history endpoint can return paged/display-trimmed messages,
|
||||
but the next model call reads ``session.history``. After a restart or a
|
||||
stale in-memory session, selecting an old chat through the paged endpoint
|
||||
used to show the transcript while the model only saw fresh context.
|
||||
"""
|
||||
if not rows:
|
||||
return
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
return
|
||||
session.history = [
|
||||
ChatMessage(role=m.role, content=m.content, metadata=_db_message_metadata(m) or None)
|
||||
for m in rows
|
||||
]
|
||||
session.message_count = len(session.history)
|
||||
|
||||
def _session_needs_db_history_hydration(session_id: str, total: int) -> bool:
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
return False
|
||||
return len(session.history or []) < int(total or 0)
|
||||
|
||||
@router.get("/api/history/{session_id}")
|
||||
async def get_session_history(
|
||||
request: Request,
|
||||
@@ -198,6 +160,8 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
|
||||
)
|
||||
page_offset = int(offset) if offset is not None else max(total - page_limit, 0)
|
||||
page_offset = max(0, min(page_offset, total))
|
||||
# Keep display pagination page-scoped. ``get_session`` is the
|
||||
# full model-context hydration seam and must not be entered here.
|
||||
rows = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
@@ -206,14 +170,6 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
|
||||
.limit(page_limit)
|
||||
.all()
|
||||
)
|
||||
if _session_needs_db_history_hydration(session_id, total):
|
||||
full_rows = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.order_by(DbChatMessage.timestamp)
|
||||
.all()
|
||||
)
|
||||
_hydrate_session_history_from_db(session_id, full_rows)
|
||||
history_dict = [
|
||||
entry for entry in (_db_history_entry(m) for m in rows)
|
||||
if not (entry.get("metadata") or {}).get("hidden")
|
||||
@@ -258,7 +214,10 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
|
||||
entry["metadata"] = msg["metadata"]
|
||||
history_dict.append(entry)
|
||||
|
||||
# Fallback: load from DB if in-memory is empty
|
||||
# Fallback: load from DB if in-memory renders empty. Display only —
|
||||
# get_session above is the hydration seam, so nothing here writes back
|
||||
# into session.history — rebuilding it from raw rows would overwrite
|
||||
# parsed multimodal content and the _db_id edit/delete keys it just set.
|
||||
if not history_dict:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
@@ -268,17 +227,10 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
|
||||
.order_by(DbChatMessage.timestamp)
|
||||
.all()
|
||||
)
|
||||
db_history = []
|
||||
for m in db_messages:
|
||||
db_history.append(_db_history_entry(m))
|
||||
if db_history:
|
||||
# Rebuild in-memory history from the full set so hidden
|
||||
# messages (e.g. compaction summaries) are kept for AI context.
|
||||
_hydrate_session_history_from_db(session_id, db_messages)
|
||||
# Response excludes hidden messages, matching the in-memory path.
|
||||
history_dict = [
|
||||
m for m in db_history
|
||||
if not (m.get("metadata") or {}).get("hidden")
|
||||
entry for entry in (_db_history_entry(m) for m in db_messages)
|
||||
if not (entry.get("metadata") or {}).get("hidden")
|
||||
]
|
||||
except Exception as e:
|
||||
logger.error(f"DB fallback failed for {session_id}: {e}")
|
||||
@@ -645,8 +597,14 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
|
||||
body = await request.json()
|
||||
keep_count = body.get("keep_count", 0)
|
||||
|
||||
# Get the source session
|
||||
source = session_manager.sessions.get(session_id)
|
||||
# Get the source session. keep_count indexes into source.history,
|
||||
# so this must go through get_session — reading the cache directly
|
||||
# forks an empty transcript out of a metadata-only session after a
|
||||
# restart (display pagination no longer hydrates it).
|
||||
try:
|
||||
source = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
if not source:
|
||||
raise HTTPException(404, "Session not found")
|
||||
|
||||
|
||||
@@ -801,15 +801,6 @@ def setup_session_routes(
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@router.get("/history/{sid}")
|
||||
def get_history(request: Request, sid: str):
|
||||
_verify_session_owner(request, sid)
|
||||
try:
|
||||
session = session_manager.get_session(sid)
|
||||
except KeyError:
|
||||
raise HTTPException(404, f"Session {sid} not found")
|
||||
return {"history": [msg.to_dict() for msg in session.history]}
|
||||
|
||||
@router.get("/session/{sid}/export")
|
||||
def export_session(request: Request, sid: str, fmt: str = "md", filename: str = ""):
|
||||
"""Export conversation history as a downloadable file.
|
||||
|
||||
@@ -34,6 +34,11 @@ class _FakeSessionManager:
|
||||
self.sessions = {"src-id": source}
|
||||
self.created = None
|
||||
|
||||
def get_session(self, session_id):
|
||||
# Fork looks the source up through get_session — the hydration seam —
|
||||
# so a session only present in the DB still forks a real transcript.
|
||||
return self.sessions[session_id]
|
||||
|
||||
def create_session(self, session_id=None, name=None, endpoint_url=None,
|
||||
model=None, rag=False, owner=None):
|
||||
self.created = _FakeSession(name=name, owner=owner)
|
||||
|
||||
@@ -5,9 +5,9 @@ The in-memory branch skips messages whose metadata has ``hidden`` (e.g.
|
||||
compaction summaries that are kept for AI context but not shown to the user).
|
||||
The DB fallback (taken when the in-memory history is empty, e.g. after a
|
||||
restart) built the client response from every DB row with no such filter, so
|
||||
hidden messages leaked to the client on DB-served sessions. The rebuilt
|
||||
in-memory ``session.history`` must still keep them, though, so only the response
|
||||
is filtered.
|
||||
hidden messages leaked to the client on DB-served sessions. Hydration of
|
||||
``session.history`` belongs to ``get_session``; this fallback only shapes the
|
||||
response, so only the response is filtered.
|
||||
|
||||
get_session_history depends on the DB, the session manager and a FastAPI
|
||||
request, so this pins the regression at the source level (as other route tests
|
||||
|
||||
@@ -0,0 +1,549 @@
|
||||
"""Display pagination must stay separate from full model-context hydration."""
|
||||
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from fastapi import APIRouter, FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from starlette.requests import Request
|
||||
from sqlalchemy import create_engine, event
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from core.database import Base, ChatMessage as DbChatMessage, Session as DbSession
|
||||
from core.models import ChatMessage, Session
|
||||
from core.session_manager import SessionManager
|
||||
from routes import chat_routes
|
||||
from routes.history import history_routes
|
||||
from routes import session_routes
|
||||
from src.request_models import ChatRequest
|
||||
|
||||
|
||||
def _database():
|
||||
engine = create_engine(
|
||||
"sqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(
|
||||
engine,
|
||||
tables=[DbSession.__table__, DbChatMessage.__table__],
|
||||
)
|
||||
return engine, sessionmaker(bind=engine, autocommit=False, autoflush=False)
|
||||
|
||||
|
||||
def _seed_session(db_factory, *, session_id="session-1", message_count=6, stored_count=None):
|
||||
"""Seed `message_count` real rows; `stored_count` overrides the denormalized
|
||||
sessions.message_count column so drift can be reproduced."""
|
||||
db = db_factory()
|
||||
try:
|
||||
db.add(
|
||||
DbSession(
|
||||
id=session_id,
|
||||
name="Long chat",
|
||||
endpoint_url="http://model.test/v1",
|
||||
model="test-model",
|
||||
owner="alice",
|
||||
message_count=message_count if stored_count is None else stored_count,
|
||||
)
|
||||
)
|
||||
start = datetime(2026, 1, 1, 12, 0, 0)
|
||||
for index in range(message_count):
|
||||
db.add(
|
||||
DbChatMessage(
|
||||
id=f"message-{index}",
|
||||
session_id=session_id,
|
||||
role="user" if index % 2 == 0 else "assistant",
|
||||
content=f"content-{index}",
|
||||
timestamp=start + timedelta(seconds=index),
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _chat_message_selects(statements):
|
||||
return [
|
||||
" ".join(statement.lower().split())
|
||||
for statement in statements
|
||||
if statement.lstrip().lower().startswith("select")
|
||||
and "chat_messages" in statement.lower()
|
||||
]
|
||||
|
||||
|
||||
def _manager(db_factory, monkeypatch, sessions=None):
|
||||
"""A real SessionManager bound to the temp DB, with load counting."""
|
||||
monkeypatch.setattr("core.session_manager.SessionLocal", db_factory)
|
||||
manager = object.__new__(SessionManager)
|
||||
manager.upload_handler = None
|
||||
manager.sessions = sessions if sessions is not None else {}
|
||||
manager.full_loads = 0
|
||||
|
||||
original_load = manager._load_session_from_db
|
||||
|
||||
def counting_load(session_id):
|
||||
manager.full_loads += 1
|
||||
return original_load(session_id)
|
||||
|
||||
manager._load_session_from_db = counting_load
|
||||
return manager
|
||||
|
||||
|
||||
def test_paginated_history_reads_only_count_and_requested_page(monkeypatch):
|
||||
engine, db_factory = _database()
|
||||
_seed_session(db_factory)
|
||||
|
||||
class DisplayOnlyManager:
|
||||
def get_session(self, _session_id):
|
||||
raise AssertionError("paginated display history must not hydrate model context")
|
||||
|
||||
monkeypatch.setattr(history_routes, "SessionLocal", db_factory)
|
||||
monkeypatch.setattr(history_routes, "_verify_session_owner", lambda *_args: None)
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(history_routes.setup_history_routes(DisplayOnlyManager()))
|
||||
|
||||
statements = []
|
||||
|
||||
def capture_sql(_conn, _cursor, statement, _parameters, _context, _executemany):
|
||||
statements.append(statement)
|
||||
|
||||
event.listen(engine, "before_cursor_execute", capture_sql)
|
||||
try:
|
||||
response = TestClient(app).get("/api/history/session-1?limit=2")
|
||||
finally:
|
||||
event.remove(engine, "before_cursor_execute", capture_sql)
|
||||
engine.dispose()
|
||||
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert [message["content"] for message in payload["history"]] == [
|
||||
"content-4",
|
||||
"content-5",
|
||||
]
|
||||
assert payload["total"] == 6
|
||||
assert payload["offset"] == 4
|
||||
assert payload["has_more_before"] is True
|
||||
assert payload["has_more_after"] is False
|
||||
|
||||
# One COUNT for the total plus one page read — never a full-transcript
|
||||
# select. The page bounds are asserted through the response above rather
|
||||
# than by matching SQL text.
|
||||
chat_selects = _chat_message_selects(statements)
|
||||
assert len(chat_selects) == 2, chat_selects
|
||||
assert sum("count(" in statement for statement in chat_selects) == 1
|
||||
|
||||
|
||||
def test_production_router_order_reaches_bounded_canonical_history(monkeypatch):
|
||||
"""The assembled app must not shadow canonical history with session routes."""
|
||||
engine, db_factory = _database()
|
||||
_seed_session(db_factory, message_count=1200)
|
||||
|
||||
class DisplayOnlyManager:
|
||||
def get_session(self, _session_id):
|
||||
raise AssertionError("bounded initial history must not hydrate all messages")
|
||||
|
||||
manager = DisplayOnlyManager()
|
||||
monkeypatch.setattr(
|
||||
session_routes,
|
||||
"router",
|
||||
APIRouter(prefix="/api", tags=["sessions"]),
|
||||
)
|
||||
monkeypatch.setattr(history_routes, "SessionLocal", db_factory)
|
||||
monkeypatch.setattr(history_routes, "_verify_session_owner", lambda *_args: None)
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(session_routes.setup_session_routes(manager, {}))
|
||||
app.include_router(history_routes.setup_history_routes(manager))
|
||||
|
||||
try:
|
||||
response = TestClient(app).get("/api/history/session-1?limit=24")
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.request.url.params["limit"] == "24"
|
||||
payload = response.json()
|
||||
displayed = len(payload["history"])
|
||||
assert 0 < displayed <= payload["limit"] <= 100
|
||||
assert payload["total"] >= 1200
|
||||
assert payload["has_more_before"] is True
|
||||
assert displayed < payload["total"]
|
||||
|
||||
|
||||
def test_incomplete_cached_history_hydrates_once_for_model_context(monkeypatch):
|
||||
engine, db_factory = _database()
|
||||
raw_multimodal = json.dumps(
|
||||
[
|
||||
{"type": "text", "text": "look at the source image"},
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {"url": "data:image/png;base64,AAAA"},
|
||||
},
|
||||
]
|
||||
)
|
||||
db = db_factory()
|
||||
try:
|
||||
db.add(
|
||||
DbSession(
|
||||
id="session-1",
|
||||
name="Long chat",
|
||||
endpoint_url="http://model.test/v1",
|
||||
model="test-model",
|
||||
owner="alice",
|
||||
message_count=3,
|
||||
)
|
||||
)
|
||||
start = datetime(2026, 1, 1, 12, 0, 0)
|
||||
db.add_all(
|
||||
[
|
||||
DbChatMessage(
|
||||
id="message-0",
|
||||
session_id="session-1",
|
||||
role="user",
|
||||
content=raw_multimodal,
|
||||
meta_data=json.dumps(
|
||||
{
|
||||
"attachments": [
|
||||
{
|
||||
"id": "upload-1",
|
||||
"filename": "source.png",
|
||||
"content_type": "image/png",
|
||||
}
|
||||
]
|
||||
}
|
||||
),
|
||||
timestamp=start,
|
||||
),
|
||||
DbChatMessage(
|
||||
id="message-1",
|
||||
session_id="session-1",
|
||||
role="assistant",
|
||||
content="answer",
|
||||
timestamp=start + timedelta(seconds=1),
|
||||
),
|
||||
DbChatMessage(
|
||||
id="message-2",
|
||||
session_id="session-1",
|
||||
role="system",
|
||||
content="compaction summary",
|
||||
meta_data=json.dumps({"hidden": True}),
|
||||
timestamp=start + timedelta(seconds=2),
|
||||
),
|
||||
]
|
||||
)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
manager = _manager(
|
||||
db_factory,
|
||||
monkeypatch,
|
||||
sessions={
|
||||
"session-1": Session(
|
||||
id="session-1",
|
||||
name="Long chat",
|
||||
endpoint_url="http://model.test/v1",
|
||||
model="test-model",
|
||||
owner="alice",
|
||||
history=[ChatMessage("user", "stale partial cache")],
|
||||
# Deliberately stale too: get_session must refresh metadata before
|
||||
# checking whether the cached transcript is complete.
|
||||
message_count=1,
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
hydrated = manager.get_session("session-1")
|
||||
first_full_loads = manager.full_loads
|
||||
warm = manager.get_session("session-1")
|
||||
second_full_loads = manager.full_loads
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert hydrated is warm
|
||||
assert len(hydrated.history) == 3
|
||||
assert first_full_loads == 1
|
||||
assert second_full_loads == first_full_loads
|
||||
|
||||
context = hydrated.get_context_messages()
|
||||
assert len(context) == 3
|
||||
assert context[0]["content"][1]["image_url"]["url"] == "data:image/png;base64,AAAA"
|
||||
assert context[0]["metadata"]["attachments"] == [
|
||||
{
|
||||
"id": "upload-1",
|
||||
"filename": "source.png",
|
||||
"content_type": "image/png",
|
||||
}
|
||||
]
|
||||
hidden_summary = next(message for message in context if message["role"] == "system")
|
||||
assert hidden_summary["content"] == "compaction summary"
|
||||
assert hidden_summary["metadata"]["hidden"] is True
|
||||
|
||||
|
||||
def test_inflated_message_count_column_does_not_reload_warm_sessions(monkeypatch):
|
||||
"""A drifted-high sessions.message_count must not reload on every read.
|
||||
|
||||
`_persist_message` swallows a failed insert while `add_message` has already
|
||||
appended in memory, so the next successful persist writes rows+1. Keyed on
|
||||
that column, the hydration gate would stay true forever and re-select the
|
||||
whole transcript on every send, edit, delete and truncate.
|
||||
"""
|
||||
engine, db_factory = _database()
|
||||
_seed_session(db_factory, message_count=6, stored_count=8)
|
||||
|
||||
manager = _manager(db_factory, monkeypatch)
|
||||
try:
|
||||
session = manager.get_session("session-1")
|
||||
cold_loads = manager.full_loads
|
||||
for _ in range(3):
|
||||
manager.get_session("session-1")
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert len(session.history) == 6
|
||||
assert cold_loads == 1
|
||||
assert manager.full_loads == 1
|
||||
|
||||
|
||||
def test_stale_low_message_count_column_still_hydrates_for_the_model(monkeypatch):
|
||||
"""The other drift direction must not hand the model a truncated transcript.
|
||||
|
||||
`_persist_message` writes message_count = 0 when the session is not cached.
|
||||
A partly-filled cache plus that stale-low column previously left the send
|
||||
path with whatever RAM happened to hold.
|
||||
"""
|
||||
engine, db_factory = _database()
|
||||
_seed_session(db_factory, message_count=6, stored_count=0)
|
||||
|
||||
manager = _manager(
|
||||
db_factory,
|
||||
monkeypatch,
|
||||
sessions={
|
||||
"session-1": Session(
|
||||
id="session-1",
|
||||
name="Long chat",
|
||||
endpoint_url="http://model.test/v1",
|
||||
model="test-model",
|
||||
owner="alice",
|
||||
history=[ChatMessage("user", "content-0")],
|
||||
message_count=0,
|
||||
)
|
||||
},
|
||||
)
|
||||
try:
|
||||
session = manager.get_session("session-1")
|
||||
manager.get_session("session-1")
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert [message.content for message in session.history] == [
|
||||
f"content-{index}" for index in range(6)
|
||||
]
|
||||
assert manager.full_loads == 1
|
||||
|
||||
|
||||
def test_fork_after_restart_copies_the_real_transcript(monkeypatch):
|
||||
"""Forking reads source.history, so it must hydrate through get_session.
|
||||
|
||||
Display pagination no longer fills the cache, so a fork taken after a
|
||||
restart used to return HTTP 200 with an empty conversation.
|
||||
"""
|
||||
engine, db_factory = _database()
|
||||
_seed_session(db_factory, message_count=6)
|
||||
|
||||
# Restart state: metadata-only cache entry, exactly what load_sessions seeds.
|
||||
manager = _manager(db_factory, monkeypatch)
|
||||
manager.load_sessions()
|
||||
|
||||
monkeypatch.setattr(history_routes, "SessionLocal", db_factory)
|
||||
monkeypatch.setattr(history_routes, "_verify_session_owner", lambda *_args: None)
|
||||
monkeypatch.setattr("core.models._SESSION_MANAGER_INSTANCE", manager)
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(history_routes.setup_history_routes(manager))
|
||||
client = TestClient(app)
|
||||
|
||||
try:
|
||||
page = client.get("/api/history/session-1?limit=2")
|
||||
assert page.status_code == 200
|
||||
assert len(manager.sessions["session-1"].history) == 0
|
||||
|
||||
response = client.post("/api/session/session-1/fork", json={"keep_count": 4})
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert payload["kept"] == 4
|
||||
|
||||
forked = manager.get_session(payload["id"])
|
||||
assert [message.content for message in forked.history] == [
|
||||
f"content-{index}" for index in range(4)
|
||||
]
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
class _ContextBuildReached(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class _ToolPolicy:
|
||||
block_all_tool_calls = False
|
||||
|
||||
def blocks(self, _tool_name):
|
||||
return False
|
||||
|
||||
|
||||
class _ChatHandler:
|
||||
async def handle_memory_command(self, _session, _message):
|
||||
return None
|
||||
|
||||
|
||||
def _json_request(path, payload):
|
||||
raw = json.dumps(payload).encode()
|
||||
sent = False
|
||||
|
||||
async def receive():
|
||||
nonlocal sent
|
||||
if sent:
|
||||
return {"type": "http.request", "body": b"", "more_body": False}
|
||||
sent = True
|
||||
return {"type": "http.request", "body": raw, "more_body": False}
|
||||
|
||||
scope = {
|
||||
"type": "http",
|
||||
"asgi": {"version": "3.0"},
|
||||
"http_version": "1.1",
|
||||
"method": "POST",
|
||||
"scheme": "http",
|
||||
"path": path,
|
||||
"raw_path": path.encode(),
|
||||
"root_path": "",
|
||||
"query_string": b"",
|
||||
"headers": [(b"content-type", b"application/json")],
|
||||
"client": ("127.0.0.1", 1234),
|
||||
"server": ("testserver", 80),
|
||||
}
|
||||
return Request(scope, receive)
|
||||
|
||||
|
||||
def _route_endpoint(router, path):
|
||||
return next(route.endpoint for route in router.routes if route.path == path)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("path", ["/api/chat", "/api/chat_stream"])
|
||||
async def test_model_send_routes_hydrate_before_context_build(monkeypatch, path):
|
||||
# A real SessionManager over a real (temp) DB — a stub here would only
|
||||
# assert that the stub hydrates, not that SessionManager does.
|
||||
engine, db_factory = _database()
|
||||
_seed_session(db_factory, message_count=6, stored_count=8)
|
||||
manager = _manager(db_factory, monkeypatch)
|
||||
manager.load_sessions() # restart state: metadata only, no messages cached
|
||||
contexts_built = []
|
||||
|
||||
async def assert_complete_context(session, *_args, **_kwargs):
|
||||
contexts_built.append(session)
|
||||
assert [message.content for message in session.history] == [
|
||||
f"content-{index}" for index in range(6)
|
||||
]
|
||||
raise _ContextBuildReached
|
||||
|
||||
monkeypatch.setattr(chat_routes, "_set_user_time_from_request", lambda *_args: None)
|
||||
monkeypatch.setattr(chat_routes, "_verify_session_owner", lambda *_args: None)
|
||||
monkeypatch.setattr(chat_routes, "effective_user", lambda *_args: "alice")
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_clear_orphaned_session_endpoint",
|
||||
lambda *_args, **_kwargs: False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_recover_empty_session_model",
|
||||
lambda *_args, **_kwargs: False,
|
||||
)
|
||||
monkeypatch.setattr(chat_routes, "_enforce_chat_privileges", lambda *_args: None)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"build_effective_tool_policy",
|
||||
lambda **_kwargs: _ToolPolicy(),
|
||||
)
|
||||
monkeypatch.setattr(chat_routes, "build_chat_context", assert_complete_context)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_resolve_request_workspace",
|
||||
lambda *_args: (None, False),
|
||||
)
|
||||
monkeypatch.setattr(chat_routes, "_classify_tool_intent", lambda *_args: None)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_is_contextual_web_followup",
|
||||
lambda *_args: False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_is_contextual_browser_followup",
|
||||
lambda *_args: False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_resolve_workspace_from_message_path",
|
||||
lambda *_args: (None, None),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_reconcile_selected_route_from_request",
|
||||
lambda *_args, **_kwargs: False,
|
||||
)
|
||||
monkeypatch.setattr(chat_routes, "resolve_session_auth", lambda *_args, **_kwargs: None)
|
||||
monkeypatch.setattr(chat_routes, "get_session_mode", lambda *_args: "chat")
|
||||
monkeypatch.setattr(
|
||||
chat_routes,
|
||||
"_is_image_generation_session",
|
||||
lambda *_args, **_kwargs: False,
|
||||
)
|
||||
monkeypatch.setattr(chat_routes, "web_search_enabled_for_turn", lambda *_args: False)
|
||||
|
||||
router = chat_routes.setup_chat_routes(
|
||||
manager,
|
||||
_ChatHandler(),
|
||||
object(),
|
||||
object(),
|
||||
object(),
|
||||
object(),
|
||||
)
|
||||
endpoint = _route_endpoint(router, path)
|
||||
|
||||
async def send():
|
||||
if path == "/api/chat":
|
||||
await endpoint(
|
||||
_json_request(path, {}),
|
||||
ChatRequest(message="hello", session="session-1"),
|
||||
)
|
||||
else:
|
||||
await endpoint(
|
||||
_json_request(
|
||||
path,
|
||||
{"message": "hello", "session": "session-1"},
|
||||
)
|
||||
)
|
||||
|
||||
try:
|
||||
with pytest.raises(_ContextBuildReached):
|
||||
await send()
|
||||
first_loads = manager.full_loads
|
||||
|
||||
# Second send on the now-warm session: the transcript is complete, so
|
||||
# it must be served from RAM even though sessions.message_count is
|
||||
# still drifted high in the DB.
|
||||
with pytest.raises(_ContextBuildReached):
|
||||
await send()
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert len(contexts_built) == 2
|
||||
assert contexts_built[0] is contexts_built[1]
|
||||
assert first_loads == 1
|
||||
assert manager.full_loads == 1
|
||||
Reference in New Issue
Block a user