Session Management

Conversation threads, multi-user isolation, persistence, and lifecycle.

Session Concepts

Thread = Session

Thread (Conversation)
├── Metadata
│   ├── thread_id (UUID)
│   ├── user_id / tenant_id
│   ├── agent_id / graph_id
│   ├── created_at, updated_at
│   ├── status: active | paused | archived | deleted
│   └── tags: dict
├── State (Checkpointed)
│   ├── Messages
│   ├── Memory refs
│   ├── Tool history
│   └── ...
└── Runs (Executions)
    ├── run_id
    ├── input
    ├── output
    ├── status
    ├── started_at, completed_at
    └── error

Multi-User Isolation

class SessionManager:
    def __init__(self, checkpointer):
        self.checkpointer = checkpointer

    async def create_thread(self, user_id: str, agent_id: str, 
                           metadata: dict = None) -> Thread:
        thread_id = str(uuid4())
        thread = Thread(
            thread_id=thread_id,
            user_id=user_id,
            agent_id=agent_id,
            metadata=metadata or {},
            created_at=datetime.utcnow()
        )
        await self.persist_thread(thread)
        return thread

    async def get_user_threads(self, user_id: str, 
                               status: str = "active") -> list[Thread]:
        # Query with user_id + status
        return await self.query_threads(user_id=user_id, status=status)

    async def get_thread(self, thread_id: str, user_id: str) -> Thread | None:
        thread = await self.load_thread(thread_id)
        if thread and thread.user_id == user_id:
            return thread
        return None  # Not found or unauthorized

Session Lifecycle

States

CREATED → ACTIVE → PAUSED → ARCHIVED
              ↓
           COMPLETED
              ↓
           ARCHIVED
              ↓
           DELETED (soft)

Transitions

class SessionLifecycle:
    VALID_TRANSITIONS = {
        "created": ["active", "deleted"],
        "active": ["paused", "completed", "deleted"],
        "paused": ["active", "completed", "deleted"],
        "completed": ["archived", "deleted"],
        "paused": ["active", "completed", "deleted"],
        "archived": ["deleted"],
        "deleted": []  # Terminal
    }

    def transition(self, session: Session, new_status: str) -> bool:
        if new_status in self.VALID_TRANSITIONS.get(session.status, []):
            session.status = new_status
            session.updated_at = datetime.utcnow()
            return True
        raise InvalidTransition(f"{session.status} → {new_status}")

Conversation Management

Message Structure

@dataclass
class Message:
    id: str
    thread_id: str
    role: Literal["system", "user", "assistant", "tool", "human"]
    content: str | list[ContentBlock]  # Text, image, tool_call, tool_result
    tool_calls: list[ToolCall] | None
    tool_call_id: str | None  # For tool results
    metadata: dict
    tokens: int
    created_at: datetime

Message Operations

class ConversationManager:
    def __init__(self, checkpointer):
        self.checkpointer = checkpointer

    async def add_message(self, thread_id: str, message: Message):
        state = await self.checkpointer.load(thread_id)
        state.messages.append(message)
        await self.checkpointer.save(thread_id, state)

    async def get_messages(self, thread_id: str, 
                          limit: int = None, 
                          before_id: str = None) -> list[Message]:
        state = await self.checkpointer.load(thread_id)
        messages = state.messages
        if before_id:
            idx = next((i for i, m in enumerate(messages) if m.id == before_id), len(messages))
            messages = messages[:idx]
        if limit:
            messages = messages[-limit:]
        return messages

    async def delete_message(self, thread_id: str, message_id: str):
        state = await self.checkpointer.load(thread_id)
        state.messages = [m for m in state.messages if m.id != message_id]
        await self.checkpointer.save(thread_id, state)

Multi-Session Patterns

1. Single Thread per User-Task

# One thread per (user, task)
thread_key = f"{user_id}:{task_type}:{task_id}"
thread = await get_or_create_thread(thread_key, user_id, agent_id)

2. Single Thread per User (Continuous)

# One thread per user, continuous conversation
thread_key = f"user:{user_id}"
thread = await get_or_create_thread(thread_key, user_id, agent_id)

3. Thread per Conversation (Slack/Teams)

# One thread per channel/conversation
thread_key = f"channel:{channel_id}"
thread = await get_or_create_thread(thread_key, bot_user_id, agent_id)

4. Ephemeral Threads (Anonymous)

# No user_id, session-based
thread_key = f"anon:{session_id}"
thread = await get_or_create_thread(thread_key, None, agent_id)
# TTL-based cleanup

Context Management per Session

Context Window per Session

class SessionContext:
    def __init__(self, thread_id: str, budget: ContextBudget):
        self.thread_id = thread_id
        self.budget = budget

    def build(self, state: AgentState) -> list[Message]:
        # Per-session budget allocation
        allocation = self.budget.allocate_for_session(state)

        return ContextBuilder(self.budget).build(state)

Session-Specific Memory

class SessionMemory:
    def __init__(self, thread_id: str, memory_store: MemoryStore):
        self.thread_id = thread_id
        self.store = memory_store

    def get_namespace(self) -> str:
        return f"session:{self.thread_id}"

    async def store(self, content: str, metadata: dict):
        await self.store.save(
            content=content,
            namespace=self.get_namespace(),
            metadata={**metadata, "thread_id": self.thread_id}
        )

    async def search(self, query: str, k: int = 10) -> list[MemoryItem]:
        return await self.store.search(
            query=query,
            namespace=self.get_namespace(),
            k=k
        )

Run Management

Run = Single Execution

@dataclass
class Run:
    run_id: str
    thread_id: str
    input: str | dict
    config: RunConfig
    status: Literal["pending", "running", "completed", "failed", "cancelled"]
    output: str | dict | None
    error: str | None
    started_at: datetime
    completed_at: datetime | None
    tokens_used: int
    cost_usd: float
    checkpoints: list[str]  # Checkpoint IDs during run

Run Queue

class RunQueue:
    def __init__(self, max_concurrent=10):
        self.queue = asyncio.Queue()
        self.running = {}
        self.max_concurrent = max_concurrent

    async def enqueue(self, run: Run):
        await self.queue.put(run)

    async def process(self, executor: Callable):
        while True:
            run = await self.queue.get()
            if len(self.running) >= self.max_concurrent:
                await self.queue.put(run)  # Re-queue
                await asyncio.sleep(1)
                continue

            self.running[run.run_id] = run
            run.status = "running"
            run.started_at = datetime.utcnow()

            try:
                result = await executor(run)
                run.output = result
                run.status = "completed"
            except Exception as e:
                run.error = str(e)
                run.status = "failed"
            finally:
                run.completed_at = datetime.utcnow()
                del self.running[run.run_id]

Session Cleanup & Retention

Retention Policies

class RetentionPolicy:
    def __init__(self):
        self.policies = {
            "active": timedelta(days=365),
            "paused": timedelta(days=90),
            "completed": timedelta(days=30),
            "archived": timedelta(days=3650),  # 10 years
            "anonymous": timedelta(days=7)     # Ephemeral
        }

    def get_ttl(self, status: str) -> timedelta:
        return self.policies.get(status, timedelta(days=30))

Cleanup Job

class SessionCleanupJob:
    def __init__(self, session_store, policy: RetentionPolicy):
        self.store = session_store
        self.policy = policy

    async def run(self):
        expired = await self.store.find_expired(self.policy)
        for session in expired:
            await self.archive_or_delete(session)

    async def archive_or_delete(self, session: Session):
        if session.status == "archived":
            await self.store.hard_delete(session.thread_id)
        else:
            session.status = "archived"
            await self.store.save(session)

Multi-Tenant Sessions

Tenant Isolation

class TenantSessionManager:
    def __init__(self, checkpointer):
        self.checkpointer = checkpointer

    def get_thread_key(self, tenant_id: str, thread_id: str) -> str:
        return f"tenant:{tenant_id}:thread:{thread_id}"

    async def create_thread(self, tenant_id: str, user_id: str, 
                           agent_id: str) -> Thread:
        thread_id = str(uuid4())
        thread = Thread(
            thread_id=thread_id,
            tenant_id=tenant_id,
            user_id=user_id,
            agent_id=agent_id
        )
        key = self.get_thread_key(tenant_id, thread_id)
        await self.checkpointer.save(key, thread)
        return thread

    async def list_tenant_threads(self, tenant_id: str) -> list[Thread]:
        prefix = f"tenant:{tenant_id}:thread:"
        keys = await self.checkpointer.list(prefix)
        threads = []
        for key in keys:
            thread = await self.checkpointer.load(key)
            if thread:
                threads.append(thread)
        return threads

Cross-Tenant Sharing (Controlled)

class CrossTenantSharing:
    def __init__(self, session_manager: TenantSessionManager):
        self.manager = session_manager

    async def share_thread(self, source_tenant: str, thread_id: str,
                          target_tenant: str, permissions: list[str]) -> Thread:
        # Copy thread with new tenant_id
        source_key = f"tenant:{source_tenant}:thread:{thread_id}"
        thread = await self.manager.checkpointer.load(source_key)

        if not thread:
            raise NotFound()

        # Check permissions
        if "read" not in permissions:
            raise Forbidden()

        new_thread = Thread(
            **asdict(thread),
            thread_id=str(uuid4()),
            tenant_id=target_tenant,
            metadata={**thread.metadata, "shared_from": source_tenant}
        )

        target_key = f"tenant:{target_tenant}:thread:{new_thread.thread_id}"
        await self.manager.checkpointer.save(target_key, new_thread)
        return new_thread

Session Analytics

Metrics to Track

Analytics Pipeline

class SessionAnalytics:
    def __init__(self, event_store):
        self.events = event_store

    async def record_event(self, event: SessionEvent):
        await self.events.append(event)

    async def get_session_metrics(self, thread_id: str) -> SessionMetrics:
        events = await self.events.query(thread_id=thread_id)
        return compute_metrics(events)

    async def get_aggregate_metrics(self, tenant_id: str, 
                                    window: timedelta) -> AggregateMetrics:
        events = await self.events.query(
            tenant_id=tenant_id,
            since=datetime.utcnow() - window
        )
        return compute_aggregates(events)

Related Notes

• Checkpointing — Persistence backend
• Event Sourcing — Event-based sessions
• Distributed State — Distributed sessions
• Versioning Migration — Session schema migration
• Memory Systems — Session memory
• LangGraph — Thread management
• AgentKit — AgentKit threads
• Observability — Session metrics
• Security — Session security
• Customer Facing Agents — Customer sessions

Sources: LangGraph threads, AgentKit threads, temporal.io sessions, session management patterns, conversation AI platforms (Intercom, Zendesk, custom), distributed session patterns.