conversation

Per-room conversation history manager.

class conversation.ConversationManager(prompt_renderer, max_history=100)[source]

Bases: object

Maintains per-room conversation histories.

Each room has its own message list. A PromptRenderer produces the system prompt dynamically from a Jinja2 template so that it can include room-specific and tool-specific context.

A sliding-window strategy keeps at most max_history user/assistant messages per room to avoid unbounded memory growth.

Parameters:

prompt_renderer (PromptRenderer)
max_history (int)

__init__(prompt_renderer, max_history=100)[source]

Initialize the instance.

Parameters:

prompt_renderer (PromptRenderer) – The prompt renderer value.
max_history (int) – The max history value.

Return type:

None

append(room_id, role, content)[source]

Add a message to room_id’s history and trim if needed.

content may be a plain string or a list of OpenRouter multimodal content parts.

Return type:

None

Parameters:

room_id (str)
role (str)
content (str | list[dict[str, Any]])

get_messages(room_id, room_context=None)[source]

Return the full message list for room_id, including the system prompt.

room_context is forwarded to the PromptRenderer so the Jinja2 template can reference room-level variables such as room_name, room_id, sender, etc.

Return type:

list[dict[str, Any]]

Parameters:

room_id (str)
room_context (dict[str, Any] | None)

async get_messages_async(room_id, room_context=None)[source]

Return the full message list for room_id, including the system prompt.

Same as get_messages() but runs Jinja2 rendering in a thread pool so the event loop is not blocked by CPU-bound template work.

Return type:

list[dict[str, Any]]

Parameters:

room_id (str)
room_context (dict[str, Any] | None)

update_message(room_id, message_id, new_content)[source]

Replace the content of an existing history entry by message ID.

Scans backward through room_id’s history for an entry whose text contains [Message ID: {message_id}] and replaces its content. Returns True if a match was found.

Return type:

bool

Parameters:

room_id (str)
message_id (str)
new_content (str | list[dict[str, Any]])

mark_deleted(room_id, message_id, deleted_at_iso)[source]

Inject a [deleted at TIMESTAMP] tag into an existing entry.

The original content is preserved so the bot retains full context. Returns True if the message was found.

Return type:

bool

Parameters:

room_id (str)
message_id (str)
deleted_at_iso (str)

clear(room_id)[source]

Wipe conversation history for a room.

Return type:: None
Parameters:: room_id (str)

MIN_CHANNEL_LIMIT = 50: Absolute bounds for per-channel overrides.

MAX_CHANNEL_LIMIT = 1000

set_channel_limit(room_id, limit)[source]

Set a per-channel override for max_history.

The value is clamped to [MIN_CHANNEL_LIMIT, MAX_CHANNEL_LIMIT]. Returns the clamped value actually stored.

Return type:

int

Parameters:

room_id (str)
limit (int)

get_channel_limit(room_id)[source]

Return the effective message limit for room_id.

Uses the per-channel override if set, otherwise the global default.

Return type:: int
Parameters:: room_id (str)