Conversation Manager

The ConversationManager v0.4.0 is a high-level coordinator that manages conversation state, handles message exchanges, and provides advanced checkpoint and snapshot capabilities for conversation management.

Overview

The ConversationManager coordinates between:

1. ConversationSystem - Core message handling
2. ContextWindowManager - Token budgeting and trimming
3. SessionManager - Session persistence and boundaries
4. ContextLoader - Context file management
5. CheckpointManager - Conversation state snapshots and branching

Architecture

Message Flow

Session Management

Session, backup, and index writes use unique sibling temporary files, flush them to disk, and atomically replace the destination. Managers for the same directory share a reentrant lock within one process. Index updates merge changed entries with the saved index so another manager's session entries are retained.

On Windows, replacement retries are limited to sharing/lock violations. An access-denied error is retried only when a delete-access probe confirms a sharing conflict or that access has recovered; read-only and ACL denials fail immediately. External Windows readers should allow delete sharing to avoid blocking a rename; public manager reads use the same store lock. There are at most six attempts with 310 ms total backoff. Failed replacements retain the previous destination, clean up only their own temporary file, and keep cached sessions marked dirty. Callers must check save_session()'s boolean result.

The session file and index are separate atomic writes, not a cross-file transaction. If the index write fails after the session succeeds, the save reports failure and the existing stale-index recovery can rebuild it. Cross-process revision merging and arbitrary concurrent mutation of a Session object are not provided by the in-process lock.

Initialization

def __init__(
    self,
    model_config=None,
    api_client=None,
    workspace_path: Optional[Path] = None,
    system_prompt: str = "",
    max_messages_per_session: int = 5000,
    max_sessions_in_memory: int = 20,
    auto_save_interval: int = 60,
    checkpoint_config: Optional[CheckpointConfig] = None
)

Configuration Resolution

The ConversationManager automatically resolves configuration from multiple sources:

1. Model Configuration: model_config for API settings
2. Live Config: config_obj from PenguinCore for runtime updates
3. Default Values: Sensible defaults for all parameters
4. Checkpoint Config: Automatic checkpoint configuration if not provided

Parameters

• model_config: Configuration for the AI model with provider-specific settings
• api_client: Client for API interactions with enhanced streaming support
• workspace_path: Path to workspace directory for conversation storage
• system_prompt: Initial system prompt with advanced template support
• max_messages_per_session: Maximum messages before creating a new session (default: 5000)
• max_sessions_in_memory: Maximum sessions to keep in memory cache (default: 20)
• auto_save_interval: Seconds between auto-saves (default: 60, 0 to disable)
• checkpoint_config: Configuration for checkpoint system with retention policies

Automatic Component Initialization

• Session Manager: Created with optimized caching and persistence
• Context Window: Initialized with provider-aware token counting
• Checkpoint Manager: Lazy initialization with automatic worker startup
• Context Loader: Pre-loads core context files on startup

Key Methods

Message Processing

async def process_message(
    self,
    message: str,
    conversation_id: Optional[str] = None,
    image_path: Optional[str] = None,
    streaming: bool = False,
    stream_callback: Optional[Callable[[str], None]] = None,
    context_files: Optional[List[str]] = None
) -> Union[str, AsyncGenerator[str, None]]

Processes a user message:

• Conversation Loading: Automatic conversation ID resolution
• Context File Integration: Loads specified context files before processing
• Image Support: Handles multimodal content with automatic token counting
• Streaming Support: Real-time response streaming with callback handling
• Error Recovery: Comprehensive error handling with fallback mechanisms

Checkpoint and Snapshot System

The ConversationManager includes checkpoint and snapshot capabilities for conversation state management.

Checkpoint Management

async def create_manual_checkpoint(
    self,
    name: Optional[str] = None,
    description: Optional[str] = None
) -> Optional[str]

Creates a manual checkpoint of the current conversation state with optional metadata.

async def rollback_to_checkpoint(self, checkpoint_id: str) -> bool

Rollbacks conversation to a specific checkpoint, restoring the exact state.

async def branch_from_checkpoint(
    self,
    checkpoint_id: str,
    name: Optional[str] = None,
    description: Optional[str] = None
) -> Optional[str]

Creates a new conversation branch from a checkpoint for experimental conversations.

def list_checkpoints(
    self,
    session_id: Optional[str] = None,
    checkpoint_type: Optional[CheckpointType] = None,
    limit: int = 50
) -> List[Dict[str, Any]]

Lists available checkpoints with filtering and pagination support.

Snapshot Management

def create_snapshot(self, *, meta: Optional[Dict[str, Any]] = None) -> Optional[str]

Creates a lightweight snapshot of the current conversation for quick restoration.

def restore_snapshot(self, snapshot_id: str) -> bool

Restores conversation from a snapshot with minimal overhead.

def branch_from_snapshot(self, snapshot_id: str, *, meta: Optional[Dict[str, Any]] = None) -> Optional[str]

Creates a new conversation branch from a snapshot for parallel experimentation.

Automatic Checkpointing

The system automatically creates checkpoints based on the configured policy:

• Auto Checkpoints: Created every N messages (configurable frequency)
• Manual Checkpoints: Created on-demand with custom metadata
• Branch Checkpoints: Created when branching from existing checkpoints
• Retention Management: Automatic cleanup of old checkpoints based on policies

Checkpoint Configuration

@dataclass
class CheckpointConfig:
    enabled: bool = True
    frequency: int = 1  # Checkpoint every message
    planes: Dict[str, bool] = field(default_factory=lambda: {"conversation": True, "tasks": False, "code": False})
    retention: Dict[str, Any] = field(default_factory=lambda: {
        "keep_all_hours": 24,
        "keep_every_nth": 10,
        "max_age_days": 30
    })
    max_auto_checkpoints: int = 1000

Context Management

def add_context(self, content: str, source: Optional[str] = None) -> Message

Adds context information to the conversation.

def load_context_file(self, file_path: str) -> bool

Loads a context file into the conversation.

def list_context_files(self) -> List[Dict[str, Any]]

Lists available context files.

Conversation State

def save(self) -> bool

Saves the current conversation state.

def load(self, conversation_id: str) -> bool

Loads a specific conversation by ID.

def list_conversations(
    self, 
    limit: int = 100,
    offset: int = 0,
    search_term: Optional[str] = None
) -> List[Dict]

Lists available conversations with search and pagination.

def reset(self) -> None

Resets the conversation state.

Token Management

def get_token_usage(self) -> Dict[str, Any]

Gets token usage statistics for the current conversation across all sessions.

Advanced Usage Examples

Basic Setup with Enhanced Features

from penguin.system.conversation_manager import ConversationManager
from penguin.system.checkpoint_manager import CheckpointConfig

# Initialize with checkpoint configuration
checkpoint_config = CheckpointConfig(
    enabled=True,
    frequency=1,  # Checkpoint every message
    retention={
        "keep_all_hours": 24,
        "keep_every_nth": 10,
        "max_age_days": 30
    }
)

conversation_manager = ConversationManager(
    model_config=model_config,
    api_client=api_client,
    workspace_path=Path("./workspace"),
    system_prompt="You are a helpful assistant.",
    checkpoint_config=checkpoint_config
)

Checkpoint and Snapshot Management

# Create a checkpoint before major changes
checkpoint_id = await conversation_manager.create_manual_checkpoint(
    name="Before refactoring",
    description="Saving state before major code changes"
)

# Continue conversation
response = await conversation_manager.process_message("Please refactor this code...")
print(response)

# If something goes wrong, rollback to the checkpoint
success = await conversation_manager.rollback_to_checkpoint(checkpoint_id)
if success:
    print("Successfully rolled back to previous state")

# Create a branch for experimental conversation
branch_id = await conversation_manager.branch_from_checkpoint(
    checkpoint_id,
    name="Experimental approach",
    description="Testing alternative refactoring strategy"
)

Snapshot-Based Quick Save/Restore

# Create a lightweight snapshot
snapshot_id = conversation_manager.create_snapshot(
    meta={"purpose": "quick_backup", "timestamp": datetime.now().isoformat()}
)

# Continue working
await conversation_manager.process_message("Make this code more efficient...")

# Quick restore if needed
success = conversation_manager.restore_snapshot(snapshot_id)
if success:
    print("Restored from snapshot")

Key Features

• Session Management: Automatically handles transitions between sessions when the context window fills up
• Token Budgeting: Ensures conversations don't exceed model limits through intelligent trimming
• Streaming Support: Works with both streaming and non-streaming response modes
• Multimodal Content: Handles text, images, and structured content
• Context Loading: Integrates external files into conversation context
• Conversation Persistence: Automatically saves state and allows resuming conversations
• Checkpoint & Snapshot System: Full conversation state management with branching
• Context Loading: Smart integration of external files and documentation
• Conversation Persistence: Automatic state saving with metadata preservation
• Error Recovery: Robust error handling with emergency checkpoint creation
• Performance Monitoring: Real-time token usage and performance tracking