mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-07-08 11:56:59 +00:00
refactor(routes): move history domain into routes/history/ subpackage (#5090)
Slice 2d of the route-domain reorganization (#4082/#4071, per specs/architecture-runtime-inventory.md §6.3). Moves history_routes.py into routes/history/, leaving a backward-compat sys.modules shim at the old path. Pure file reorganization, no behavior change. The shim uses sys.modules replacement (same pattern as the merged gallery #4903, research #4975, and memory #5007 slices) so that `import routes.history_routes`, `from routes.history_routes import X`, `importlib.import_module(...)`, and the `import ... as history_routes` + `monkeypatch.setattr(history_routes, ...)` pattern used by test_history_compact_tool_calls.py / test_fork_session_metadata.py all operate on the same module object the application uses. The canonical module does NOT depend on the shim — routes/history/ history_routes.py imports only from core/, src/, and routes.session_routes (a sibling route module whose old import path stays valid via its own shim when session is migrated later). Three source-introspection test sites repointed to the new canonical path: - test_history_db_fallback_hidden.py - test_history_order_by_timestamp_regression.py - test_model_helper_owner_scope.py Adds tests/test_history_routes_shim.py to pin the sys.modules shim contract (legacy and canonical paths resolve to the same module object; monkeypatch via legacy alias reaches the canonical module). Verified: compileall clean; full suite 4351 passed, 3 skipped.
This commit is contained in:
Tal.Yuan
@@ -645,7 +645,7 @@ from routes.research.research_routes import setup_research_routes
|
||||
app.include_router(setup_research_routes(research_handler, session_manager=session_manager))
|
||||
|
||||
# History
|
||||
from routes.history_routes import setup_history_routes
|
||||
from routes.history.history_routes import setup_history_routes
|
||||
app.include_router(setup_history_routes(session_manager))
|
||||
|
||||
# Search
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
"""History route domain package (slice 2d, #4082/#4071).
|
||||
|
||||
Contains history_routes.py, migrated from the flat routes/ directory.
|
||||
Backward-compat shim at routes/history_routes.py re-exports from here.
|
||||
"""
|
||||
@@ -0,0 +1,768 @@
|
||||
"""History routes — session history, truncation, fork, conversation topics."""
|
||||
|
||||
import json
|
||||
import uuid
|
||||
import logging
|
||||
import re
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
from fastapi import APIRouter, Request, HTTPException
|
||||
|
||||
from core.models import ChatMessage
|
||||
from core.database import SessionLocal, ChatMessage as DbChatMessage, Session as DbSession
|
||||
from src.topic_analyzer import analyze_topics
|
||||
from routes.session_routes import (
|
||||
_message_role,
|
||||
_message_text,
|
||||
_reject_compact_during_active_run,
|
||||
_verify_session_owner,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_HISTORY_INLINE_MEDIA_THRESHOLD = 200_000
|
||||
_DATA_IMAGE_RE = re.compile(r"data:image/[^;,\"]+;base64,[A-Za-z0-9+/=\s]+")
|
||||
|
||||
|
||||
def _history_display_content(content: Any) -> Any:
|
||||
"""Return a lightweight browser-display copy of stored message content.
|
||||
|
||||
Older multimodal user messages may be persisted as a JSON *string*
|
||||
containing image_url blocks with inline base64 image bytes. Those bytes are
|
||||
needed for model calls when the turn is first sent, but they should not be
|
||||
sent back through /api/history every time the user opens the chat. The
|
||||
attachment metadata already carries file ids/names for the UI cards.
|
||||
"""
|
||||
if isinstance(content, list):
|
||||
text_parts = []
|
||||
omitted_media = 0
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
if block.get("type") == "text":
|
||||
text = block.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
text_parts.append(text)
|
||||
elif block.get("type") in {"image_url", "input_image", "audio", "input_audio"}:
|
||||
omitted_media += 1
|
||||
text = "\n".join(text_parts).strip()
|
||||
if omitted_media and not text:
|
||||
return f"[{omitted_media} media attachment{'s' if omitted_media != 1 else ''} omitted from history view]"
|
||||
return text
|
||||
|
||||
if not isinstance(content, str):
|
||||
return content
|
||||
if len(content) < _HISTORY_INLINE_MEDIA_THRESHOLD and "data:image/" not in content:
|
||||
return content
|
||||
|
||||
stripped = content.lstrip()
|
||||
if stripped.startswith("["):
|
||||
try:
|
||||
blocks = json.loads(content)
|
||||
except (json.JSONDecodeError, TypeError, ValueError):
|
||||
blocks = None
|
||||
if isinstance(blocks, list):
|
||||
text_parts = []
|
||||
for block in blocks:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
if block.get("type") == "text":
|
||||
text = block.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
text_parts.append(text)
|
||||
if text_parts:
|
||||
return "\n".join(text_parts).strip()
|
||||
|
||||
if "data:image/" in content:
|
||||
return _DATA_IMAGE_RE.sub("[inline image omitted from history view]", content)
|
||||
return content
|
||||
|
||||
|
||||
def _merge_continue_rows_to_delete(db_messages, db1, db2):
|
||||
"""DB rows to delete when merging the last two assistant messages.
|
||||
|
||||
Always the second assistant message (db2), plus ONLY the single
|
||||
intervening "continue" user message (the one carrying "previous response
|
||||
was interrupted") — matching the in-memory merge. The previous code
|
||||
deleted the whole index range between the two assistant rows, destroying
|
||||
any tool/system/user messages in between and desyncing the DB from the
|
||||
in-memory history.
|
||||
"""
|
||||
to_delete = [db2]
|
||||
i1 = next((i for i, m in enumerate(db_messages) if m is db1), None)
|
||||
i2 = next((i for i, m in enumerate(db_messages) if m is db2), None)
|
||||
if i1 is not None and i2 is not None and i2 - 1 > i1:
|
||||
between = db_messages[i2 - 1]
|
||||
if getattr(between, "role", "") == "user" and "previous response was interrupted" in (getattr(between, "content", "") or ""):
|
||||
to_delete.append(between)
|
||||
return to_delete
|
||||
|
||||
|
||||
def setup_history_routes(session_manager) -> APIRouter:
|
||||
router = APIRouter(tags=["history"])
|
||||
|
||||
def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]:
|
||||
entry = {"role": m.role, "content": _history_display_content(m.content)}
|
||||
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"
|
||||
if meta:
|
||||
entry["metadata"] = meta
|
||||
return entry
|
||||
|
||||
@router.get("/api/history/{session_id}")
|
||||
async def get_session_history(
|
||||
request: Request,
|
||||
session_id: str,
|
||||
limit: Optional[int] = None,
|
||||
offset: Optional[int] = None,
|
||||
) -> Dict[str, Any]:
|
||||
_verify_session_owner(request, session_id)
|
||||
if limit is not None:
|
||||
page_limit = max(1, min(int(limit), 100))
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
|
||||
if db_session is None:
|
||||
raise HTTPException(404, f"Session '{session_id}' not found")
|
||||
|
||||
total = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.count()
|
||||
)
|
||||
page_offset = int(offset) if offset is not None else max(total - page_limit, 0)
|
||||
page_offset = max(0, min(page_offset, total))
|
||||
rows = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.order_by(DbChatMessage.timestamp)
|
||||
.offset(page_offset)
|
||||
.limit(page_limit)
|
||||
.all()
|
||||
)
|
||||
history_dict = [
|
||||
entry for entry in (_db_history_entry(m) for m in rows)
|
||||
if not (entry.get("metadata") or {}).get("hidden")
|
||||
]
|
||||
return {
|
||||
"history": history_dict,
|
||||
"model": db_session.model,
|
||||
"endpoint_url": db_session.endpoint_url,
|
||||
"name": db_session.name,
|
||||
"offset": page_offset,
|
||||
"limit": page_limit,
|
||||
"total": total,
|
||||
"has_more_before": page_offset > 0,
|
||||
"has_more_after": page_offset + len(rows) < total,
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
raise HTTPException(404, f"Session '{session_id}' not found")
|
||||
|
||||
history_dict = []
|
||||
for msg in session.history:
|
||||
if isinstance(msg, ChatMessage):
|
||||
# Skip hidden messages (e.g. compaction summaries for AI context)
|
||||
if msg.metadata and msg.metadata.get("hidden"):
|
||||
continue
|
||||
entry = {"role": msg.role, "content": _history_display_content(msg.content)}
|
||||
if msg.metadata:
|
||||
entry["metadata"] = msg.metadata
|
||||
history_dict.append(entry)
|
||||
elif isinstance(msg, dict):
|
||||
if msg.get("metadata", {}).get("hidden"):
|
||||
continue
|
||||
entry = {
|
||||
"role": msg.get("role", ""),
|
||||
"content": _history_display_content(msg.get("content", "")),
|
||||
}
|
||||
if msg.get("metadata"):
|
||||
entry["metadata"] = msg["metadata"]
|
||||
history_dict.append(entry)
|
||||
|
||||
# Fallback: load from DB if in-memory is empty
|
||||
if not history_dict:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_messages = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.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.
|
||||
session.history = [
|
||||
ChatMessage(role=m["role"], content=m["content"], metadata=m.get("metadata"))
|
||||
for m in db_history
|
||||
]
|
||||
# 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")
|
||||
]
|
||||
except Exception as e:
|
||||
logger.error(f"DB fallback failed for {session_id}: {e}")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
return {
|
||||
"history": history_dict,
|
||||
"model": session.model,
|
||||
"endpoint_url": session.endpoint_url,
|
||||
"name": session.name,
|
||||
}
|
||||
|
||||
@router.post("/api/session/{session_id}/truncate")
|
||||
async def truncate_session(request: Request, session_id: str):
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
keep_count = body.get("keep_count", 0)
|
||||
result = session_manager.truncate_messages(session_id, keep_count)
|
||||
return {"status": "ok", "kept": keep_count, "truncated": result}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Truncate error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/message")
|
||||
async def add_message(request: Request, session_id: str):
|
||||
"""Add a message to a session (for slash command persistence)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
role = body.get("role", "assistant")
|
||||
content = body.get("content", "")
|
||||
if not content:
|
||||
raise HTTPException(400, "content is required")
|
||||
msg = ChatMessage(role=role, content=content, metadata=body.get("metadata"))
|
||||
session_manager.add_message(session_id, msg)
|
||||
return {"status": "ok"}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
|
||||
@router.post("/api/session/{session_id}/delete-messages")
|
||||
async def delete_messages(request: Request, session_id: str):
|
||||
"""Delete specific messages by DB ID (or legacy index)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
msg_ids = body.get("msg_ids", [])
|
||||
indices = body.get("indices") # legacy fallback
|
||||
|
||||
session = session_manager.get_session(session_id)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
if msg_ids:
|
||||
# New ID-based delete
|
||||
deleted = 0
|
||||
for mid in msg_ids:
|
||||
db_msg = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.id == mid,
|
||||
DbChatMessage.session_id == session_id,
|
||||
).first()
|
||||
if db_msg:
|
||||
db.delete(db_msg)
|
||||
deleted += 1
|
||||
|
||||
# Remove from in-memory history by matching _db_id
|
||||
def _get_db_id(m):
|
||||
meta = m.metadata if isinstance(m, ChatMessage) else (m.get('metadata') if isinstance(m, dict) else None)
|
||||
return meta.get('_db_id') if isinstance(meta, dict) else None
|
||||
session.history = [m for m in session.history if _get_db_id(m) not in msg_ids]
|
||||
elif indices:
|
||||
# Legacy index-based delete
|
||||
indices = sorted(indices, reverse=True)
|
||||
db_messages = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.session_id == session_id
|
||||
).order_by(DbChatMessage.timestamp).all()
|
||||
|
||||
deleted = 0
|
||||
for idx in indices:
|
||||
if 0 <= idx < len(db_messages):
|
||||
db.delete(db_messages[idx])
|
||||
deleted += 1
|
||||
if 0 <= idx < len(session.history):
|
||||
session.history.pop(idx)
|
||||
else:
|
||||
return {"status": "ok", "deleted": 0}
|
||||
|
||||
session.message_count = len(session.history)
|
||||
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
|
||||
if db_session:
|
||||
db_session.message_count = len(session.history)
|
||||
from datetime import datetime, timezone
|
||||
db_session.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
db.commit()
|
||||
return {"status": "ok", "deleted": deleted}
|
||||
finally:
|
||||
db.close()
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Delete messages error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/edit-message")
|
||||
async def edit_message(request: Request, session_id: str):
|
||||
"""Edit the content of a message by its database ID."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
msg_id = body.get("msg_id")
|
||||
content = body.get("content")
|
||||
if not msg_id or content is None:
|
||||
raise HTTPException(400, "msg_id and content are required")
|
||||
|
||||
session = session_manager.get_session(session_id)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_msg = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.id == msg_id,
|
||||
DbChatMessage.session_id == session_id,
|
||||
).first()
|
||||
if not db_msg:
|
||||
raise HTTPException(404, "Message not found")
|
||||
|
||||
db_msg.content = content
|
||||
meta = {}
|
||||
if db_msg.meta_data:
|
||||
try: meta = json.loads(db_msg.meta_data)
|
||||
except (json.JSONDecodeError, ValueError): pass
|
||||
meta['edited'] = True
|
||||
db_msg.meta_data = json.dumps(meta)
|
||||
|
||||
# Update in-memory history by matching _db_id
|
||||
for hmsg in session.history:
|
||||
hmeta = hmsg.metadata if isinstance(hmsg, ChatMessage) else hmsg.get('metadata')
|
||||
if isinstance(hmeta, dict) and hmeta.get('_db_id') == msg_id:
|
||||
if isinstance(hmsg, ChatMessage):
|
||||
hmsg.content = content
|
||||
hmsg.metadata['edited'] = True
|
||||
elif isinstance(hmsg, dict):
|
||||
hmsg['content'] = content
|
||||
hmsg['metadata']['edited'] = True
|
||||
break
|
||||
|
||||
db.commit()
|
||||
return {"status": "ok"}
|
||||
finally:
|
||||
db.close()
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Edit message error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/mark-stopped")
|
||||
async def mark_stopped(request: Request, session_id: str):
|
||||
"""Mark the last assistant message as stopped by user."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
# Find last assistant message and add stopped metadata
|
||||
for msg in reversed(session.history):
|
||||
if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
|
||||
(isinstance(msg, dict) and msg.get('role') == 'assistant'):
|
||||
if isinstance(msg, ChatMessage):
|
||||
if not msg.metadata:
|
||||
msg.metadata = {}
|
||||
msg.metadata['stopped'] = True
|
||||
if not msg.metadata.get('model'):
|
||||
msg.metadata['model'] = session.model
|
||||
else:
|
||||
if 'metadata' not in msg:
|
||||
msg['metadata'] = {}
|
||||
msg['metadata']['stopped'] = True
|
||||
if not msg['metadata'].get('model'):
|
||||
msg['metadata']['model'] = session.model
|
||||
break
|
||||
# Also update in DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
import json as _json
|
||||
db_messages = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
|
||||
.order_by(DbChatMessage.timestamp.desc())
|
||||
.first()
|
||||
)
|
||||
if db_messages:
|
||||
meta = {}
|
||||
if db_messages.meta_data:
|
||||
try:
|
||||
meta = _json.loads(db_messages.meta_data)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
pass
|
||||
meta['stopped'] = True
|
||||
if not meta.get('model'):
|
||||
meta['model'] = session.model
|
||||
db_messages.meta_data = _json.dumps(meta)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
session_manager.save_sessions()
|
||||
return {"status": "ok"}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Mark stopped error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/update-last-meta")
|
||||
async def update_last_meta(request: Request, session_id: str):
|
||||
"""Merge metadata into the last assistant message (e.g. save variants)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
meta_update = body.get("metadata", {})
|
||||
session = session_manager.get_session(session_id)
|
||||
|
||||
# Update in-memory
|
||||
for msg in reversed(session.history):
|
||||
if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
|
||||
(isinstance(msg, dict) and msg.get('role') == 'assistant'):
|
||||
if isinstance(msg, ChatMessage):
|
||||
if not msg.metadata:
|
||||
msg.metadata = {}
|
||||
msg.metadata.update(meta_update)
|
||||
else:
|
||||
if 'metadata' not in msg:
|
||||
msg['metadata'] = {}
|
||||
msg['metadata'].update(meta_update)
|
||||
break
|
||||
|
||||
# Update in DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
import json as _json
|
||||
db_msg = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
|
||||
.order_by(DbChatMessage.timestamp.desc())
|
||||
.first()
|
||||
)
|
||||
if db_msg:
|
||||
meta = {}
|
||||
if db_msg.meta_data:
|
||||
try: meta = _json.loads(db_msg.meta_data)
|
||||
except (json.JSONDecodeError, ValueError): pass
|
||||
meta.update(meta_update)
|
||||
db_msg.meta_data = _json.dumps(meta)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
session_manager.save_sessions()
|
||||
return {"status": "ok"}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Update last meta error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/merge-last-assistant")
|
||||
async def merge_last_assistant(request: Request, session_id: str):
|
||||
"""Merge the last two assistant messages into one (for continue)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
separator = body.get("separator", "\n\n")
|
||||
session = session_manager.get_session(session_id)
|
||||
|
||||
# Find last two assistant messages in-memory
|
||||
ai_indices = []
|
||||
for i, msg in enumerate(session.history):
|
||||
role = msg.role if isinstance(msg, ChatMessage) else msg.get('role', '')
|
||||
if role == 'assistant':
|
||||
ai_indices.append(i)
|
||||
|
||||
if len(ai_indices) < 2:
|
||||
return {"status": "ok", "merged": False}
|
||||
|
||||
idx1, idx2 = ai_indices[-2], ai_indices[-1]
|
||||
msg1, msg2 = session.history[idx1], session.history[idx2]
|
||||
|
||||
content1 = msg1.content if isinstance(msg1, ChatMessage) else msg1.get('content', '')
|
||||
content2 = msg2.content if isinstance(msg2, ChatMessage) else msg2.get('content', '')
|
||||
merged_content = content1 + separator + content2
|
||||
|
||||
# Merge metadata
|
||||
meta1 = (msg1.metadata if isinstance(msg1, ChatMessage) else msg1.get('metadata')) or {}
|
||||
meta2 = (msg2.metadata if isinstance(msg2, ChatMessage) else msg2.get('metadata')) or {}
|
||||
merged_meta = {**meta1, **meta2}
|
||||
merged_meta.pop('stopped', None) # no longer stopped after continue
|
||||
|
||||
# Update first message, remove second
|
||||
if isinstance(msg1, ChatMessage):
|
||||
msg1.content = merged_content
|
||||
msg1.metadata = merged_meta
|
||||
else:
|
||||
msg1['content'] = merged_content
|
||||
msg1['metadata'] = merged_meta
|
||||
|
||||
# Also remove the hidden "continue" user message between them if present
|
||||
# It's the message at idx2-1 if it's a user message with continue text
|
||||
remove_indices = [idx2]
|
||||
if idx2 - 1 > idx1:
|
||||
between = session.history[idx2 - 1]
|
||||
between_role = between.role if isinstance(between, ChatMessage) else between.get('role', '')
|
||||
between_content = between.content if isinstance(between, ChatMessage) else between.get('content', '')
|
||||
if between_role == 'user' and 'previous response was interrupted' in between_content:
|
||||
remove_indices.insert(0, idx2 - 1)
|
||||
|
||||
for ri in sorted(remove_indices, reverse=True):
|
||||
session.history.pop(ri)
|
||||
|
||||
# Update DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
import json as _json
|
||||
db_messages = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.order_by(DbChatMessage.timestamp)
|
||||
.all()
|
||||
)
|
||||
# Find last two assistant messages in DB
|
||||
ai_db = [(i, m) for i, m in enumerate(db_messages) if m.role == 'assistant']
|
||||
if len(ai_db) >= 2:
|
||||
(_, db1), (_, db2) = ai_db[-2], ai_db[-1]
|
||||
db1.content = merged_content
|
||||
db1.meta_data = _json.dumps(merged_meta)
|
||||
|
||||
# Mirror the in-memory deletion: remove the second assistant
|
||||
# message and ONLY the "continue" user message between them
|
||||
# (not arbitrary tool/system/user rows). The old
|
||||
# range-delete destroyed every row between the two assistant
|
||||
# messages, desyncing the DB from the in-memory history.
|
||||
for _row in _merge_continue_rows_to_delete(db_messages, db1, db2):
|
||||
db.delete(_row)
|
||||
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
session_manager.save_sessions()
|
||||
return {"status": "ok", "merged": True}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Merge assistant error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/fork")
|
||||
async def fork_session(request: Request, session_id: str):
|
||||
"""Create a new session with messages copied up to keep_count."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
keep_count = body.get("keep_count", 0)
|
||||
|
||||
# Get the source session
|
||||
source = session_manager.sessions.get(session_id)
|
||||
if not source:
|
||||
raise HTTPException(404, "Session not found")
|
||||
|
||||
# Create new session
|
||||
new_id = str(uuid.uuid4())
|
||||
fork_name = f"\u2ADD {source.name}"
|
||||
new_session = session_manager.create_session(
|
||||
session_id=new_id,
|
||||
name=fork_name,
|
||||
endpoint_url=source.endpoint_url,
|
||||
model=source.model,
|
||||
rag=False,
|
||||
owner=getattr(source, 'owner', None),
|
||||
)
|
||||
|
||||
# Copy messages up to keep_count
|
||||
msgs_to_copy = source.history[:keep_count]
|
||||
for msg in msgs_to_copy:
|
||||
# Copy the metadata dict. Sharing it would let the fork's
|
||||
# persistence (add_message -> _persist_message stamps
|
||||
# _db_id/timestamp onto the dict) mutate the SOURCE session's
|
||||
# in-memory messages, corrupting their _db_id and breaking
|
||||
# edit/delete-by-id on the original conversation.
|
||||
meta = dict(msg.metadata) if isinstance(msg.metadata, dict) else None
|
||||
new_session.add_message(ChatMessage(msg.role, msg.content, meta))
|
||||
try:
|
||||
from src.event_bus import fire_event
|
||||
fire_event("session_created", getattr(source, 'owner', None))
|
||||
except Exception:
|
||||
logger.debug("session_created event dispatch failed", exc_info=True)
|
||||
|
||||
return {
|
||||
"status": "ok",
|
||||
"id": new_id,
|
||||
"name": fork_name,
|
||||
"kept": len(msgs_to_copy),
|
||||
}
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Fork error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.get("/api/conversations/topics")
|
||||
async def get_conversation_topics(request: Request) -> Dict[str, Any]:
|
||||
from src.auth_helpers import require_user
|
||||
user = require_user(request)
|
||||
try:
|
||||
return analyze_topics(session_manager, owner=user or None)
|
||||
except Exception as e:
|
||||
raise HTTPException(500, f"Topic analysis failed: {e}")
|
||||
|
||||
@router.post("/api/session/{session_id}/compact")
|
||||
async def compact_session(request: Request, session_id: str):
|
||||
"""Manually trigger context compaction for a session."""
|
||||
_verify_session_owner(request, session_id)
|
||||
from src.auth_helpers import effective_user
|
||||
owner = effective_user(request)
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
_reject_compact_during_active_run(session_id)
|
||||
|
||||
try:
|
||||
from src.model_context import estimate_tokens, get_context_length
|
||||
from src.llm_core import llm_call_async
|
||||
from src.endpoint_resolver import resolve_endpoint
|
||||
|
||||
if len(session.history) < 6:
|
||||
return {"status": "ok", "message": "Not enough messages to compact"}
|
||||
|
||||
ctx_len = get_context_length(session.endpoint_url, session.model)
|
||||
messages_before = session.get_context_messages()
|
||||
used_before = estimate_tokens(messages_before)
|
||||
pct_before = round((used_before / ctx_len) * 100, 1) if ctx_len else 0
|
||||
msg_count_before = len(session.history)
|
||||
|
||||
# Keep only last 4 messages, summarize the rest
|
||||
keep_count = 4
|
||||
older = session.history[:-keep_count]
|
||||
recent = session.history[-keep_count:]
|
||||
|
||||
# Build text to summarize
|
||||
convo_text = "\n".join(
|
||||
f"{_message_role(m).upper()}: "
|
||||
f"{_message_text(m)[:2000]}"
|
||||
for m in older
|
||||
)
|
||||
|
||||
# Use utility model if available
|
||||
util_url, util_model, util_headers = resolve_endpoint("utility", owner=owner or None)
|
||||
compact_url = util_url or session.endpoint_url
|
||||
compact_model = util_model or session.model
|
||||
compact_headers = util_headers if util_url else session.headers
|
||||
|
||||
from src.context_compactor import SELF_SUMMARY_SYSTEM_PROMPT
|
||||
compaction_count = sum(1 for m in session.history if isinstance(m, ChatMessage) and "[Conversation summary" in (m.content or ""))
|
||||
sys_prompt = SELF_SUMMARY_SYSTEM_PROMPT.replace("{count}", str(len(older))).replace("{n}", str(compaction_count + 1))
|
||||
summary = await llm_call_async(
|
||||
compact_url, compact_model,
|
||||
[
|
||||
{"role": "system", "content": sys_prompt},
|
||||
{"role": "user", "content": convo_text},
|
||||
],
|
||||
temperature=0.2, max_tokens=1024,
|
||||
headers=compact_headers, timeout=30,
|
||||
)
|
||||
|
||||
# Replace session history: summary as system message + recent messages
|
||||
# System message holds the full summary for AI context
|
||||
system_summary = ChatMessage(
|
||||
role="system",
|
||||
content=f"[Conversation summary — {len(older)} earlier messages were compacted]\n\n{summary}",
|
||||
metadata={"compacted": True, "hidden": True},
|
||||
)
|
||||
# Visible assistant message just shows stats
|
||||
summary_msg = ChatMessage(
|
||||
role="assistant",
|
||||
content=f"**Conversation compacted** — {len(older)} messages summarized, {len(recent)} kept.",
|
||||
metadata={"compacted": True, "messages_removed": len(older)},
|
||||
)
|
||||
new_history = [system_summary, summary_msg] + list(recent)
|
||||
session.history = new_history
|
||||
session.message_count = len(session.history)
|
||||
logger.info(f"Compact: session {session_id} history now has {len(session.history)} messages (was {msg_count_before})")
|
||||
|
||||
# Update DB: delete old messages, insert summary
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_msgs = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.session_id == session_id
|
||||
).order_by(DbChatMessage.timestamp).all()
|
||||
|
||||
# Delete all but the last keep_count
|
||||
for m in db_msgs[:-keep_count]:
|
||||
db.delete(m)
|
||||
|
||||
# Insert system summary (hidden, for AI context) and visible summary
|
||||
import json as _json
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
now = datetime.now(timezone.utc)
|
||||
db_sys_summary = DbChatMessage(
|
||||
id=str(uuid.uuid4()),
|
||||
session_id=session_id,
|
||||
role="system",
|
||||
content=system_summary.content,
|
||||
meta_data=_json.dumps(system_summary.metadata),
|
||||
timestamp=now,
|
||||
)
|
||||
db.add(db_sys_summary)
|
||||
db_summary = DbChatMessage(
|
||||
id=str(uuid.uuid4()),
|
||||
session_id=session_id,
|
||||
role="assistant",
|
||||
content=summary_msg.content,
|
||||
meta_data=_json.dumps(summary_msg.metadata),
|
||||
timestamp=now,
|
||||
)
|
||||
db.add(db_summary)
|
||||
|
||||
# Update session record
|
||||
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
|
||||
if db_session:
|
||||
db_session.message_count = len(session.history)
|
||||
db_session.updated_at = datetime.now(timezone.utc)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
session_manager.save_sessions()
|
||||
|
||||
used_after = estimate_tokens(session.get_context_messages())
|
||||
pct_after = round((used_after / ctx_len) * 100, 1) if ctx_len else 0
|
||||
|
||||
return {
|
||||
"status": "ok",
|
||||
"message": f"Compacted: {msg_count_before} msgs → {len(session.history)} msgs ({pct_before}% → {pct_after}%)",
|
||||
"before": pct_before,
|
||||
"after": pct_after,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Manual compact error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
return router
|
||||
+13
-764
@@ -1,768 +1,17 @@
|
||||
"""History routes — session history, truncation, fork, conversation topics."""
|
||||
"""Backward-compat shim — canonical location is routes/history/history_routes.py.
|
||||
|
||||
import json
|
||||
import uuid
|
||||
import logging
|
||||
import re
|
||||
from typing import Dict, Any, Optional
|
||||
This module is replaced in ``sys.modules`` by the canonical module object so
|
||||
that ``import routes.history_routes``, ``from routes.history_routes import X``,
|
||||
``importlib.import_module("routes.history_routes")``, and the
|
||||
``import ... as history_routes`` + ``monkeypatch.setattr(history_routes, ...)``
|
||||
pattern used by test_history_compact_tool_calls.py / test_fork_session_metadata.py
|
||||
all operate on the *same* object the application actually uses. Keeps existing
|
||||
import paths working after slice 2d (#4082/#4071). Source-introspection tests
|
||||
read the canonical file by path.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, Request, HTTPException
|
||||
import sys as _sys
|
||||
|
||||
from core.models import ChatMessage
|
||||
from core.database import SessionLocal, ChatMessage as DbChatMessage, Session as DbSession
|
||||
from src.topic_analyzer import analyze_topics
|
||||
from routes.session_routes import (
|
||||
_message_role,
|
||||
_message_text,
|
||||
_reject_compact_during_active_run,
|
||||
_verify_session_owner,
|
||||
)
|
||||
from routes.history import history_routes as _canonical # noqa: F401
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_HISTORY_INLINE_MEDIA_THRESHOLD = 200_000
|
||||
_DATA_IMAGE_RE = re.compile(r"data:image/[^;,\"]+;base64,[A-Za-z0-9+/=\s]+")
|
||||
|
||||
|
||||
def _history_display_content(content: Any) -> Any:
|
||||
"""Return a lightweight browser-display copy of stored message content.
|
||||
|
||||
Older multimodal user messages may be persisted as a JSON *string*
|
||||
containing image_url blocks with inline base64 image bytes. Those bytes are
|
||||
needed for model calls when the turn is first sent, but they should not be
|
||||
sent back through /api/history every time the user opens the chat. The
|
||||
attachment metadata already carries file ids/names for the UI cards.
|
||||
"""
|
||||
if isinstance(content, list):
|
||||
text_parts = []
|
||||
omitted_media = 0
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
if block.get("type") == "text":
|
||||
text = block.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
text_parts.append(text)
|
||||
elif block.get("type") in {"image_url", "input_image", "audio", "input_audio"}:
|
||||
omitted_media += 1
|
||||
text = "\n".join(text_parts).strip()
|
||||
if omitted_media and not text:
|
||||
return f"[{omitted_media} media attachment{'s' if omitted_media != 1 else ''} omitted from history view]"
|
||||
return text
|
||||
|
||||
if not isinstance(content, str):
|
||||
return content
|
||||
if len(content) < _HISTORY_INLINE_MEDIA_THRESHOLD and "data:image/" not in content:
|
||||
return content
|
||||
|
||||
stripped = content.lstrip()
|
||||
if stripped.startswith("["):
|
||||
try:
|
||||
blocks = json.loads(content)
|
||||
except (json.JSONDecodeError, TypeError, ValueError):
|
||||
blocks = None
|
||||
if isinstance(blocks, list):
|
||||
text_parts = []
|
||||
for block in blocks:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
if block.get("type") == "text":
|
||||
text = block.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
text_parts.append(text)
|
||||
if text_parts:
|
||||
return "\n".join(text_parts).strip()
|
||||
|
||||
if "data:image/" in content:
|
||||
return _DATA_IMAGE_RE.sub("[inline image omitted from history view]", content)
|
||||
return content
|
||||
|
||||
|
||||
def _merge_continue_rows_to_delete(db_messages, db1, db2):
|
||||
"""DB rows to delete when merging the last two assistant messages.
|
||||
|
||||
Always the second assistant message (db2), plus ONLY the single
|
||||
intervening "continue" user message (the one carrying "previous response
|
||||
was interrupted") — matching the in-memory merge. The previous code
|
||||
deleted the whole index range between the two assistant rows, destroying
|
||||
any tool/system/user messages in between and desyncing the DB from the
|
||||
in-memory history.
|
||||
"""
|
||||
to_delete = [db2]
|
||||
i1 = next((i for i, m in enumerate(db_messages) if m is db1), None)
|
||||
i2 = next((i for i, m in enumerate(db_messages) if m is db2), None)
|
||||
if i1 is not None and i2 is not None and i2 - 1 > i1:
|
||||
between = db_messages[i2 - 1]
|
||||
if getattr(between, "role", "") == "user" and "previous response was interrupted" in (getattr(between, "content", "") or ""):
|
||||
to_delete.append(between)
|
||||
return to_delete
|
||||
|
||||
|
||||
def setup_history_routes(session_manager) -> APIRouter:
|
||||
router = APIRouter(tags=["history"])
|
||||
|
||||
def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]:
|
||||
entry = {"role": m.role, "content": _history_display_content(m.content)}
|
||||
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"
|
||||
if meta:
|
||||
entry["metadata"] = meta
|
||||
return entry
|
||||
|
||||
@router.get("/api/history/{session_id}")
|
||||
async def get_session_history(
|
||||
request: Request,
|
||||
session_id: str,
|
||||
limit: Optional[int] = None,
|
||||
offset: Optional[int] = None,
|
||||
) -> Dict[str, Any]:
|
||||
_verify_session_owner(request, session_id)
|
||||
if limit is not None:
|
||||
page_limit = max(1, min(int(limit), 100))
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
|
||||
if db_session is None:
|
||||
raise HTTPException(404, f"Session '{session_id}' not found")
|
||||
|
||||
total = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.count()
|
||||
)
|
||||
page_offset = int(offset) if offset is not None else max(total - page_limit, 0)
|
||||
page_offset = max(0, min(page_offset, total))
|
||||
rows = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.order_by(DbChatMessage.timestamp)
|
||||
.offset(page_offset)
|
||||
.limit(page_limit)
|
||||
.all()
|
||||
)
|
||||
history_dict = [
|
||||
entry for entry in (_db_history_entry(m) for m in rows)
|
||||
if not (entry.get("metadata") or {}).get("hidden")
|
||||
]
|
||||
return {
|
||||
"history": history_dict,
|
||||
"model": db_session.model,
|
||||
"endpoint_url": db_session.endpoint_url,
|
||||
"name": db_session.name,
|
||||
"offset": page_offset,
|
||||
"limit": page_limit,
|
||||
"total": total,
|
||||
"has_more_before": page_offset > 0,
|
||||
"has_more_after": page_offset + len(rows) < total,
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
raise HTTPException(404, f"Session '{session_id}' not found")
|
||||
|
||||
history_dict = []
|
||||
for msg in session.history:
|
||||
if isinstance(msg, ChatMessage):
|
||||
# Skip hidden messages (e.g. compaction summaries for AI context)
|
||||
if msg.metadata and msg.metadata.get("hidden"):
|
||||
continue
|
||||
entry = {"role": msg.role, "content": _history_display_content(msg.content)}
|
||||
if msg.metadata:
|
||||
entry["metadata"] = msg.metadata
|
||||
history_dict.append(entry)
|
||||
elif isinstance(msg, dict):
|
||||
if msg.get("metadata", {}).get("hidden"):
|
||||
continue
|
||||
entry = {
|
||||
"role": msg.get("role", ""),
|
||||
"content": _history_display_content(msg.get("content", "")),
|
||||
}
|
||||
if msg.get("metadata"):
|
||||
entry["metadata"] = msg["metadata"]
|
||||
history_dict.append(entry)
|
||||
|
||||
# Fallback: load from DB if in-memory is empty
|
||||
if not history_dict:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_messages = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.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.
|
||||
session.history = [
|
||||
ChatMessage(role=m["role"], content=m["content"], metadata=m.get("metadata"))
|
||||
for m in db_history
|
||||
]
|
||||
# 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")
|
||||
]
|
||||
except Exception as e:
|
||||
logger.error(f"DB fallback failed for {session_id}: {e}")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
return {
|
||||
"history": history_dict,
|
||||
"model": session.model,
|
||||
"endpoint_url": session.endpoint_url,
|
||||
"name": session.name,
|
||||
}
|
||||
|
||||
@router.post("/api/session/{session_id}/truncate")
|
||||
async def truncate_session(request: Request, session_id: str):
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
keep_count = body.get("keep_count", 0)
|
||||
result = session_manager.truncate_messages(session_id, keep_count)
|
||||
return {"status": "ok", "kept": keep_count, "truncated": result}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Truncate error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/message")
|
||||
async def add_message(request: Request, session_id: str):
|
||||
"""Add a message to a session (for slash command persistence)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
role = body.get("role", "assistant")
|
||||
content = body.get("content", "")
|
||||
if not content:
|
||||
raise HTTPException(400, "content is required")
|
||||
msg = ChatMessage(role=role, content=content, metadata=body.get("metadata"))
|
||||
session_manager.add_message(session_id, msg)
|
||||
return {"status": "ok"}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
|
||||
@router.post("/api/session/{session_id}/delete-messages")
|
||||
async def delete_messages(request: Request, session_id: str):
|
||||
"""Delete specific messages by DB ID (or legacy index)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
msg_ids = body.get("msg_ids", [])
|
||||
indices = body.get("indices") # legacy fallback
|
||||
|
||||
session = session_manager.get_session(session_id)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
if msg_ids:
|
||||
# New ID-based delete
|
||||
deleted = 0
|
||||
for mid in msg_ids:
|
||||
db_msg = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.id == mid,
|
||||
DbChatMessage.session_id == session_id,
|
||||
).first()
|
||||
if db_msg:
|
||||
db.delete(db_msg)
|
||||
deleted += 1
|
||||
|
||||
# Remove from in-memory history by matching _db_id
|
||||
def _get_db_id(m):
|
||||
meta = m.metadata if isinstance(m, ChatMessage) else (m.get('metadata') if isinstance(m, dict) else None)
|
||||
return meta.get('_db_id') if isinstance(meta, dict) else None
|
||||
session.history = [m for m in session.history if _get_db_id(m) not in msg_ids]
|
||||
elif indices:
|
||||
# Legacy index-based delete
|
||||
indices = sorted(indices, reverse=True)
|
||||
db_messages = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.session_id == session_id
|
||||
).order_by(DbChatMessage.timestamp).all()
|
||||
|
||||
deleted = 0
|
||||
for idx in indices:
|
||||
if 0 <= idx < len(db_messages):
|
||||
db.delete(db_messages[idx])
|
||||
deleted += 1
|
||||
if 0 <= idx < len(session.history):
|
||||
session.history.pop(idx)
|
||||
else:
|
||||
return {"status": "ok", "deleted": 0}
|
||||
|
||||
session.message_count = len(session.history)
|
||||
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
|
||||
if db_session:
|
||||
db_session.message_count = len(session.history)
|
||||
from datetime import datetime, timezone
|
||||
db_session.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
db.commit()
|
||||
return {"status": "ok", "deleted": deleted}
|
||||
finally:
|
||||
db.close()
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Delete messages error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/edit-message")
|
||||
async def edit_message(request: Request, session_id: str):
|
||||
"""Edit the content of a message by its database ID."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
msg_id = body.get("msg_id")
|
||||
content = body.get("content")
|
||||
if not msg_id or content is None:
|
||||
raise HTTPException(400, "msg_id and content are required")
|
||||
|
||||
session = session_manager.get_session(session_id)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_msg = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.id == msg_id,
|
||||
DbChatMessage.session_id == session_id,
|
||||
).first()
|
||||
if not db_msg:
|
||||
raise HTTPException(404, "Message not found")
|
||||
|
||||
db_msg.content = content
|
||||
meta = {}
|
||||
if db_msg.meta_data:
|
||||
try: meta = json.loads(db_msg.meta_data)
|
||||
except (json.JSONDecodeError, ValueError): pass
|
||||
meta['edited'] = True
|
||||
db_msg.meta_data = json.dumps(meta)
|
||||
|
||||
# Update in-memory history by matching _db_id
|
||||
for hmsg in session.history:
|
||||
hmeta = hmsg.metadata if isinstance(hmsg, ChatMessage) else hmsg.get('metadata')
|
||||
if isinstance(hmeta, dict) and hmeta.get('_db_id') == msg_id:
|
||||
if isinstance(hmsg, ChatMessage):
|
||||
hmsg.content = content
|
||||
hmsg.metadata['edited'] = True
|
||||
elif isinstance(hmsg, dict):
|
||||
hmsg['content'] = content
|
||||
hmsg['metadata']['edited'] = True
|
||||
break
|
||||
|
||||
db.commit()
|
||||
return {"status": "ok"}
|
||||
finally:
|
||||
db.close()
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Edit message error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/mark-stopped")
|
||||
async def mark_stopped(request: Request, session_id: str):
|
||||
"""Mark the last assistant message as stopped by user."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
# Find last assistant message and add stopped metadata
|
||||
for msg in reversed(session.history):
|
||||
if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
|
||||
(isinstance(msg, dict) and msg.get('role') == 'assistant'):
|
||||
if isinstance(msg, ChatMessage):
|
||||
if not msg.metadata:
|
||||
msg.metadata = {}
|
||||
msg.metadata['stopped'] = True
|
||||
if not msg.metadata.get('model'):
|
||||
msg.metadata['model'] = session.model
|
||||
else:
|
||||
if 'metadata' not in msg:
|
||||
msg['metadata'] = {}
|
||||
msg['metadata']['stopped'] = True
|
||||
if not msg['metadata'].get('model'):
|
||||
msg['metadata']['model'] = session.model
|
||||
break
|
||||
# Also update in DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
import json as _json
|
||||
db_messages = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
|
||||
.order_by(DbChatMessage.timestamp.desc())
|
||||
.first()
|
||||
)
|
||||
if db_messages:
|
||||
meta = {}
|
||||
if db_messages.meta_data:
|
||||
try:
|
||||
meta = _json.loads(db_messages.meta_data)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
pass
|
||||
meta['stopped'] = True
|
||||
if not meta.get('model'):
|
||||
meta['model'] = session.model
|
||||
db_messages.meta_data = _json.dumps(meta)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
session_manager.save_sessions()
|
||||
return {"status": "ok"}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Mark stopped error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/update-last-meta")
|
||||
async def update_last_meta(request: Request, session_id: str):
|
||||
"""Merge metadata into the last assistant message (e.g. save variants)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
meta_update = body.get("metadata", {})
|
||||
session = session_manager.get_session(session_id)
|
||||
|
||||
# Update in-memory
|
||||
for msg in reversed(session.history):
|
||||
if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
|
||||
(isinstance(msg, dict) and msg.get('role') == 'assistant'):
|
||||
if isinstance(msg, ChatMessage):
|
||||
if not msg.metadata:
|
||||
msg.metadata = {}
|
||||
msg.metadata.update(meta_update)
|
||||
else:
|
||||
if 'metadata' not in msg:
|
||||
msg['metadata'] = {}
|
||||
msg['metadata'].update(meta_update)
|
||||
break
|
||||
|
||||
# Update in DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
import json as _json
|
||||
db_msg = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
|
||||
.order_by(DbChatMessage.timestamp.desc())
|
||||
.first()
|
||||
)
|
||||
if db_msg:
|
||||
meta = {}
|
||||
if db_msg.meta_data:
|
||||
try: meta = _json.loads(db_msg.meta_data)
|
||||
except (json.JSONDecodeError, ValueError): pass
|
||||
meta.update(meta_update)
|
||||
db_msg.meta_data = _json.dumps(meta)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
session_manager.save_sessions()
|
||||
return {"status": "ok"}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Update last meta error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/merge-last-assistant")
|
||||
async def merge_last_assistant(request: Request, session_id: str):
|
||||
"""Merge the last two assistant messages into one (for continue)."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
separator = body.get("separator", "\n\n")
|
||||
session = session_manager.get_session(session_id)
|
||||
|
||||
# Find last two assistant messages in-memory
|
||||
ai_indices = []
|
||||
for i, msg in enumerate(session.history):
|
||||
role = msg.role if isinstance(msg, ChatMessage) else msg.get('role', '')
|
||||
if role == 'assistant':
|
||||
ai_indices.append(i)
|
||||
|
||||
if len(ai_indices) < 2:
|
||||
return {"status": "ok", "merged": False}
|
||||
|
||||
idx1, idx2 = ai_indices[-2], ai_indices[-1]
|
||||
msg1, msg2 = session.history[idx1], session.history[idx2]
|
||||
|
||||
content1 = msg1.content if isinstance(msg1, ChatMessage) else msg1.get('content', '')
|
||||
content2 = msg2.content if isinstance(msg2, ChatMessage) else msg2.get('content', '')
|
||||
merged_content = content1 + separator + content2
|
||||
|
||||
# Merge metadata
|
||||
meta1 = (msg1.metadata if isinstance(msg1, ChatMessage) else msg1.get('metadata')) or {}
|
||||
meta2 = (msg2.metadata if isinstance(msg2, ChatMessage) else msg2.get('metadata')) or {}
|
||||
merged_meta = {**meta1, **meta2}
|
||||
merged_meta.pop('stopped', None) # no longer stopped after continue
|
||||
|
||||
# Update first message, remove second
|
||||
if isinstance(msg1, ChatMessage):
|
||||
msg1.content = merged_content
|
||||
msg1.metadata = merged_meta
|
||||
else:
|
||||
msg1['content'] = merged_content
|
||||
msg1['metadata'] = merged_meta
|
||||
|
||||
# Also remove the hidden "continue" user message between them if present
|
||||
# It's the message at idx2-1 if it's a user message with continue text
|
||||
remove_indices = [idx2]
|
||||
if idx2 - 1 > idx1:
|
||||
between = session.history[idx2 - 1]
|
||||
between_role = between.role if isinstance(between, ChatMessage) else between.get('role', '')
|
||||
between_content = between.content if isinstance(between, ChatMessage) else between.get('content', '')
|
||||
if between_role == 'user' and 'previous response was interrupted' in between_content:
|
||||
remove_indices.insert(0, idx2 - 1)
|
||||
|
||||
for ri in sorted(remove_indices, reverse=True):
|
||||
session.history.pop(ri)
|
||||
|
||||
# Update DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
import json as _json
|
||||
db_messages = (
|
||||
db.query(DbChatMessage)
|
||||
.filter(DbChatMessage.session_id == session_id)
|
||||
.order_by(DbChatMessage.timestamp)
|
||||
.all()
|
||||
)
|
||||
# Find last two assistant messages in DB
|
||||
ai_db = [(i, m) for i, m in enumerate(db_messages) if m.role == 'assistant']
|
||||
if len(ai_db) >= 2:
|
||||
(_, db1), (_, db2) = ai_db[-2], ai_db[-1]
|
||||
db1.content = merged_content
|
||||
db1.meta_data = _json.dumps(merged_meta)
|
||||
|
||||
# Mirror the in-memory deletion: remove the second assistant
|
||||
# message and ONLY the "continue" user message between them
|
||||
# (not arbitrary tool/system/user rows). The old
|
||||
# range-delete destroyed every row between the two assistant
|
||||
# messages, desyncing the DB from the in-memory history.
|
||||
for _row in _merge_continue_rows_to_delete(db_messages, db1, db2):
|
||||
db.delete(_row)
|
||||
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
session_manager.save_sessions()
|
||||
return {"status": "ok", "merged": True}
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
except Exception as e:
|
||||
logger.error(f"Merge assistant error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.post("/api/session/{session_id}/fork")
|
||||
async def fork_session(request: Request, session_id: str):
|
||||
"""Create a new session with messages copied up to keep_count."""
|
||||
_verify_session_owner(request, session_id)
|
||||
try:
|
||||
body = await request.json()
|
||||
keep_count = body.get("keep_count", 0)
|
||||
|
||||
# Get the source session
|
||||
source = session_manager.sessions.get(session_id)
|
||||
if not source:
|
||||
raise HTTPException(404, "Session not found")
|
||||
|
||||
# Create new session
|
||||
new_id = str(uuid.uuid4())
|
||||
fork_name = f"\u2ADD {source.name}"
|
||||
new_session = session_manager.create_session(
|
||||
session_id=new_id,
|
||||
name=fork_name,
|
||||
endpoint_url=source.endpoint_url,
|
||||
model=source.model,
|
||||
rag=False,
|
||||
owner=getattr(source, 'owner', None),
|
||||
)
|
||||
|
||||
# Copy messages up to keep_count
|
||||
msgs_to_copy = source.history[:keep_count]
|
||||
for msg in msgs_to_copy:
|
||||
# Copy the metadata dict. Sharing it would let the fork's
|
||||
# persistence (add_message -> _persist_message stamps
|
||||
# _db_id/timestamp onto the dict) mutate the SOURCE session's
|
||||
# in-memory messages, corrupting their _db_id and breaking
|
||||
# edit/delete-by-id on the original conversation.
|
||||
meta = dict(msg.metadata) if isinstance(msg.metadata, dict) else None
|
||||
new_session.add_message(ChatMessage(msg.role, msg.content, meta))
|
||||
try:
|
||||
from src.event_bus import fire_event
|
||||
fire_event("session_created", getattr(source, 'owner', None))
|
||||
except Exception:
|
||||
logger.debug("session_created event dispatch failed", exc_info=True)
|
||||
|
||||
return {
|
||||
"status": "ok",
|
||||
"id": new_id,
|
||||
"name": fork_name,
|
||||
"kept": len(msgs_to_copy),
|
||||
}
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Fork error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
@router.get("/api/conversations/topics")
|
||||
async def get_conversation_topics(request: Request) -> Dict[str, Any]:
|
||||
from src.auth_helpers import require_user
|
||||
user = require_user(request)
|
||||
try:
|
||||
return analyze_topics(session_manager, owner=user or None)
|
||||
except Exception as e:
|
||||
raise HTTPException(500, f"Topic analysis failed: {e}")
|
||||
|
||||
@router.post("/api/session/{session_id}/compact")
|
||||
async def compact_session(request: Request, session_id: str):
|
||||
"""Manually trigger context compaction for a session."""
|
||||
_verify_session_owner(request, session_id)
|
||||
from src.auth_helpers import effective_user
|
||||
owner = effective_user(request)
|
||||
try:
|
||||
session = session_manager.get_session(session_id)
|
||||
except KeyError:
|
||||
raise HTTPException(404, "Session not found")
|
||||
_reject_compact_during_active_run(session_id)
|
||||
|
||||
try:
|
||||
from src.model_context import estimate_tokens, get_context_length
|
||||
from src.llm_core import llm_call_async
|
||||
from src.endpoint_resolver import resolve_endpoint
|
||||
|
||||
if len(session.history) < 6:
|
||||
return {"status": "ok", "message": "Not enough messages to compact"}
|
||||
|
||||
ctx_len = get_context_length(session.endpoint_url, session.model)
|
||||
messages_before = session.get_context_messages()
|
||||
used_before = estimate_tokens(messages_before)
|
||||
pct_before = round((used_before / ctx_len) * 100, 1) if ctx_len else 0
|
||||
msg_count_before = len(session.history)
|
||||
|
||||
# Keep only last 4 messages, summarize the rest
|
||||
keep_count = 4
|
||||
older = session.history[:-keep_count]
|
||||
recent = session.history[-keep_count:]
|
||||
|
||||
# Build text to summarize
|
||||
convo_text = "\n".join(
|
||||
f"{_message_role(m).upper()}: "
|
||||
f"{_message_text(m)[:2000]}"
|
||||
for m in older
|
||||
)
|
||||
|
||||
# Use utility model if available
|
||||
util_url, util_model, util_headers = resolve_endpoint("utility", owner=owner or None)
|
||||
compact_url = util_url or session.endpoint_url
|
||||
compact_model = util_model or session.model
|
||||
compact_headers = util_headers if util_url else session.headers
|
||||
|
||||
from src.context_compactor import SELF_SUMMARY_SYSTEM_PROMPT
|
||||
compaction_count = sum(1 for m in session.history if isinstance(m, ChatMessage) and "[Conversation summary" in (m.content or ""))
|
||||
sys_prompt = SELF_SUMMARY_SYSTEM_PROMPT.replace("{count}", str(len(older))).replace("{n}", str(compaction_count + 1))
|
||||
summary = await llm_call_async(
|
||||
compact_url, compact_model,
|
||||
[
|
||||
{"role": "system", "content": sys_prompt},
|
||||
{"role": "user", "content": convo_text},
|
||||
],
|
||||
temperature=0.2, max_tokens=1024,
|
||||
headers=compact_headers, timeout=30,
|
||||
)
|
||||
|
||||
# Replace session history: summary as system message + recent messages
|
||||
# System message holds the full summary for AI context
|
||||
system_summary = ChatMessage(
|
||||
role="system",
|
||||
content=f"[Conversation summary — {len(older)} earlier messages were compacted]\n\n{summary}",
|
||||
metadata={"compacted": True, "hidden": True},
|
||||
)
|
||||
# Visible assistant message just shows stats
|
||||
summary_msg = ChatMessage(
|
||||
role="assistant",
|
||||
content=f"**Conversation compacted** — {len(older)} messages summarized, {len(recent)} kept.",
|
||||
metadata={"compacted": True, "messages_removed": len(older)},
|
||||
)
|
||||
new_history = [system_summary, summary_msg] + list(recent)
|
||||
session.history = new_history
|
||||
session.message_count = len(session.history)
|
||||
logger.info(f"Compact: session {session_id} history now has {len(session.history)} messages (was {msg_count_before})")
|
||||
|
||||
# Update DB: delete old messages, insert summary
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db_msgs = db.query(DbChatMessage).filter(
|
||||
DbChatMessage.session_id == session_id
|
||||
).order_by(DbChatMessage.timestamp).all()
|
||||
|
||||
# Delete all but the last keep_count
|
||||
for m in db_msgs[:-keep_count]:
|
||||
db.delete(m)
|
||||
|
||||
# Insert system summary (hidden, for AI context) and visible summary
|
||||
import json as _json
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
now = datetime.now(timezone.utc)
|
||||
db_sys_summary = DbChatMessage(
|
||||
id=str(uuid.uuid4()),
|
||||
session_id=session_id,
|
||||
role="system",
|
||||
content=system_summary.content,
|
||||
meta_data=_json.dumps(system_summary.metadata),
|
||||
timestamp=now,
|
||||
)
|
||||
db.add(db_sys_summary)
|
||||
db_summary = DbChatMessage(
|
||||
id=str(uuid.uuid4()),
|
||||
session_id=session_id,
|
||||
role="assistant",
|
||||
content=summary_msg.content,
|
||||
meta_data=_json.dumps(summary_msg.metadata),
|
||||
timestamp=now,
|
||||
)
|
||||
db.add(db_summary)
|
||||
|
||||
# Update session record
|
||||
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
|
||||
if db_session:
|
||||
db_session.message_count = len(session.history)
|
||||
db_session.updated_at = datetime.now(timezone.utc)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
session_manager.save_sessions()
|
||||
|
||||
used_after = estimate_tokens(session.get_context_messages())
|
||||
pct_after = round((used_after / ctx_len) * 100, 1) if ctx_len else 0
|
||||
|
||||
return {
|
||||
"status": "ok",
|
||||
"message": f"Compacted: {msg_count_before} msgs → {len(session.history)} msgs ({pct_before}% → {pct_after}%)",
|
||||
"before": pct_before,
|
||||
"after": pct_after,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Manual compact error {session_id}: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
return router
|
||||
_sys.modules[__name__] = _canonical
|
||||
|
||||
@@ -16,7 +16,7 @@ in this repo do).
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
SRC = Path(__file__).resolve().parent.parent / "routes" / "history_routes.py"
|
||||
SRC = Path(__file__).resolve().parent.parent / "routes" / "history" / "history_routes.py"
|
||||
|
||||
|
||||
def _function_source(src_text, name):
|
||||
|
||||
@@ -25,7 +25,7 @@ from sqlalchemy.orm import sessionmaker
|
||||
from core.database import Base, ChatMessage as DbChatMessage, Session as DbSession
|
||||
|
||||
|
||||
HISTORY_ROUTES = Path(__file__).resolve().parent.parent / "routes" / "history_routes.py"
|
||||
HISTORY_ROUTES = Path(__file__).resolve().parent.parent / "routes" / "history" / "history_routes.py"
|
||||
|
||||
|
||||
def test_chatmessage_model_has_timestamp_not_created_at():
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Regression test for the history route shim (slice 2d, #4082/#4071).
|
||||
|
||||
The backward-compat shim at ``routes/history_routes.py`` uses ``sys.modules``
|
||||
replacement so the legacy import path and the canonical ``routes.history.*``
|
||||
path resolve to the *same* module object. This is required because
|
||||
``test_history_compact_tool_calls.py`` and ``test_fork_session_metadata.py``
|
||||
do ``import routes.history_routes as history_routes`` followed by
|
||||
``monkeypatch.setattr(history_routes, "_verify_session_owner", ...)`` — for
|
||||
those patches to take effect at runtime, the legacy module object and the
|
||||
canonical one must be identical. This test pins that contract.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
|
||||
import routes.history_routes as _shim_history # noqa: F401
|
||||
|
||||
|
||||
def test_legacy_and_canonical_history_module_are_same_object():
|
||||
"""``import routes.history_routes`` must alias the canonical module."""
|
||||
legacy = importlib.import_module("routes.history_routes")
|
||||
canonical = importlib.import_module("routes.history.history_routes")
|
||||
assert legacy is canonical, (
|
||||
"routes.history_routes shim must resolve to the canonical "
|
||||
"routes.history.history_routes module object"
|
||||
)
|
||||
|
||||
|
||||
def test_monkeypatch_via_legacy_alias_reaches_canonical(monkeypatch):
|
||||
"""Patching through the legacy alias must reach the canonical module.
|
||||
|
||||
Several history tests do ``import routes.history_routes as history_routes``
|
||||
followed by ``monkeypatch.setattr(history_routes, "_verify_session_owner",
|
||||
...)``. For that to take effect at runtime, the legacy module object and the
|
||||
canonical one must be identical.
|
||||
"""
|
||||
legacy = importlib.import_module("routes.history_routes")
|
||||
canonical = importlib.import_module("routes.history.history_routes")
|
||||
|
||||
sentinel = object()
|
||||
monkeypatch.setattr(legacy, "setup_history_routes", sentinel)
|
||||
assert canonical.setup_history_routes is sentinel, (
|
||||
"monkeypatch via legacy alias did not reach the canonical module"
|
||||
)
|
||||
@@ -34,7 +34,7 @@ def test_task_parse_resolves_with_owner_scope():
|
||||
|
||||
|
||||
def test_history_compact_resolves_with_owner_scope():
|
||||
body = _function_source("routes/history_routes.py", "compact_session")
|
||||
body = _function_source("routes/history/history_routes.py", "compact_session")
|
||||
assert "owner = effective_user(request)" in body
|
||||
assert 'resolve_endpoint("utility", owner=owner or None)' in body
|
||||
|
||||
|
||||
Reference in New Issue
Block a user