# 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
```python
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
```python
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
```python
@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
```python
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
```python
# 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)
```python
# 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)
```python
# 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)
```python
# 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
```python
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
```python
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
```python
@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
```python
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
```python
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
```python
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
```python
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)
```python
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
| Metric | Description |
|---|---|
| **Sessions per user** | Engagement |
| **Messages per session** | Depth |
| **Session duration** | Time spent |
| **Turns per session** | Interaction count |
| **Completion rate** | % sessions reaching goal |
| **Escalation rate** | % needing human |
| **Cost per session** | $ per conversation |
| **Tokens per session** | Context usage |

### Analytics Pipeline
```python
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

- [[StateManagement/Checkpointing.md]] — Persistence backend
- [[StateManagement/Event-Sourcing.md]] — Event-based sessions
- [[StateManagement/Distributed-State.md]] — Distributed sessions
- [[StateManagement/Versioning-Migration.md]] — Session schema migration
- [[ContextEngineering/Memory-Systems.md]] — Session memory
- [[Frameworks/LangGraph.md]] — Thread management
- [[Frameworks/AgentKit.md]] — AgentKit threads
- [[Production/Observability.md]] — Session metrics
- [[Production/Security.md]] — Session security
- [[Domains/Customer-Facing-Agents.md]] — Customer sessions

---

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