"""Topic reorganization tools."""
from pydantic import BaseModel, Field, field_validator
from ..bot import BotContext
from ..session_manager import TopicReassignment, validate_topic_name
from .base import ToolResult
from .registry import folder_bot
[docs]
class TopicAssignment(BaseModel, frozen=True):
"""A single message-to-topic assignment."""
message_index: int = Field(
description="Zero-based index of the message in the full history."
)
topic: str = Field(
description=(
"New topic name. Must be lowercase letters, numbers, and "
"underscores only (e.g. 'project_planning', 'weather')."
),
)
[docs]
@field_validator("topic")
@classmethod
def normalize_topic(cls, v: str) -> str:
return validate_topic_name(v)
[docs]
class GetFullHistoryRequest(BaseModel, frozen=True):
"""Request to get the full conversation history with message indices."""
[docs]
@folder_bot.tool(
name="get_full_history",
request_type=GetFullHistoryRequest,
response_type=ToolResult,
)
async def get_full_history(
request: GetFullHistoryRequest, _context: BotContext | None = None
) -> ToolResult:
"""Get the full conversation history with message indices and current topics.
Returns all messages numbered by index. Use this before calling
reorganize_topics to see what needs to be reassigned.
"""
if _context is None or "session" not in _context.services:
return ToolResult(content="Session manager not available.", is_error=True)
session_manager = _context.services["session"]
messages = session_manager.get_history(_context.user_id)
if not messages:
return ToolResult(content="No messages in history.")
lines = [f"Full history ({len(messages)} messages):"]
for i, msg in enumerate(messages):
content_preview = msg["content"][:120]
if len(msg["content"]) > 120:
content_preview += "..."
lines.append(f" [{i}] ({msg['role']}) [{msg['topic']}] {content_preview}")
return ToolResult(content="\n".join(lines))
[docs]
class ReorganizeTopicsRequest(BaseModel, frozen=True):
"""Request to reassign topic labels on conversation messages."""
assignments: list[TopicAssignment] = Field(
description=(
"List of message index + new topic pairs. Only include messages "
"whose topic should CHANGE."
),
)
[docs]
@folder_bot.tool(
name="reorganize_topics",
request_type=ReorganizeTopicsRequest,
response_type=ToolResult,
)
async def reorganize_topics(
request: ReorganizeTopicsRequest, _context: BotContext | None = None
) -> ToolResult:
"""Reassign topic labels on conversation messages.
Use this when the user asks to re-split, reorganize, or rename their
conversation topics. First use get_full_history to see all messages with
their indices, then call this tool with the new assignments.
"""
if _context is None or "session" not in _context.services:
return ToolResult(content="Session manager not available.", is_error=True)
session_manager = _context.services["session"]
if not request.assignments:
return ToolResult(content="No assignments provided.", is_error=True)
reassignments = [
TopicReassignment(message_index=a.message_index, new_topic=a.topic)
for a in request.assignments
]
try:
updated = session_manager.update_message_topics(_context.user_id, reassignments)
except ValueError as e:
return ToolResult(content=f"Error: {e}", is_error=True)
topics = session_manager.get_topics(_context.user_id)
topic_summary = ", ".join(
f"{t['topic']} ({t['message_count']} msgs)" for t in topics
)
return ToolResult(content=f"Updated {updated} messages. Topics: {topic_summary}")