Persist Space sessions in MongoDB (#170)

* Persist Space sessions across backend restarts

The hosted Space lost active conversations because sessions, approval state, event streams, and Claude quotas only lived in process memory. This adds a shared optional Mongo store for frontend sessions while keeping local and CLI runs on the no-op/local path when Mongo is absent.

Constraint: Space deployment needs durable state without making PyPI/local CLI users configure Mongo.

Constraint: Frontend deletes must hide sessions with a visibility tag, not erase traces needed for recovery or SFT export.

Rejected: Restore every active session at startup | large user counts would create a thundering herd after deploys.

Rejected: Store one giant session document | Mongo document limits make per-message snapshot docs safer.

Confidence: medium

Scope-risk: broad

Directive: Keep session_messages as compactable runtime snapshots and session_trace_messages/session_events as append-only raw traces; do not collapse them into one collection without revisiting SFT/replay needs.

Tested: UV_CACHE_DIR=/tmp/uv-cache uv run --extra dev pytest tests/unit/test_session_persistence.py tests/unit/test_user_quotas.py -q

Tested: UV_CACHE_DIR=/tmp/uv-cache uv run --extra dev python -m py_compile backend/session_manager.py backend/routes/agent.py backend/user_quotas.py backend/models.py agent/core/session_persistence.py agent/core/session.py agent/context_manager/manager.py

Tested: local Mongo smoke for snapshot load, event replay, quota cap, soft delete

Tested: npm run build

Not-tested: Full background worker recovery after Space restart; intentionally Phase 3/backlog.

Not-tested: Full unit suite currently has pre-existing doom-loop wording expectation failures unrelated to this change.

* Protect persisted session restore boundaries

Lazy restore now authorizes an existing session before mutating its HF token and starts at most one runtime task when concurrent reconnects race to restore the same Mongo-backed session. Regression tests cover cross-user token overwrite, duplicate task prevention, pending approval restore, dev listing, and Mongo quota cap delegation.

Constraint: Session restore is lazy because the Space may have many durable sessions and cannot eagerly hydrate all of them on startup.

Rejected: Per-session restore locks | idempotent task startup plus a post-load in-memory check prevents leaked loops with less bookkeeping.

Confidence: high

Scope-risk: narrow

Directive: Do not update runtime credentials before access checks in ensure_session_loaded.

Tested: UV_CACHE_DIR=/tmp/uv-cache uv run --extra dev pytest tests/unit/test_session_persistence.py tests/unit/test_session_manager_persistence.py tests/unit/test_user_quotas.py -q

Tested: UV_CACHE_DIR=/tmp/uv-cache uv run python -m py_compile backend/session_manager.py backend/user_quotas.py agent/core/session_persistence.py

Tested: git diff --check

agent/context_manager/manager.py

@@ -160,6 +160,7 @@ class ContextManager:
         self.running_context_usage = 0
         self.untouched_messages = untouched_messages
         self.items: list[Message] = [Message(role="system", content=self.system_prompt)]
+        self.on_message_added = None
 
     def _load_system_prompt(
         self,
@@ -219,6 +220,8 @@ class ContextManager:
         if token_count:
             self.running_context_usage = token_count
         self.items.append(message)
+        if self.on_message_added:
+            self.on_message_added(message)
 
     def get_messages(self) -> list[Message]:
         """Get all messages for sending to LLM.

agent/core/session.py

@@ -65,6 +65,7 @@ class OpType(Enum):
 class Event:
     event_type: str
     data: Optional[dict[str, Any]] = None
+    seq: Optional[int] = None
 
 
 class Session:
@@ -87,9 +88,11 @@ class Session:
         defer_turn_complete_notification: bool = False,
         session_id: str | None = None,
         user_id: str | None = None,
+        persistence_store: Any | None = None,
     ):
         self.hf_token: Optional[str] = hf_token
         self.user_id: Optional[str] = user_id
+        self.persistence_store = persistence_store
         self.tool_router = tool_router
         self.stream = stream
         if config is None:
@@ -135,11 +138,10 @@ class Session:
         #          thinking params at all
         # Key absent → not probed yet; fall back to the raw preference.
         self.model_effective_effort: dict[str, str | None] = {}
+        self.context_manager.on_message_added = self._schedule_trace_message
 
     async def send_event(self, event: Event) -> None:
         """Send event back to client and log to trajectory"""
-        await self.event_queue.put(event)
-
         # Log event to trajectory
         self.logged_events.append(
             {
@@ -148,6 +150,15 @@ class Session:
                 "data": event.data,
             }
         )
+        if self.persistence_store is not None:
+            try:
+                event.seq = await self.persistence_store.append_event(
+                    self.session_id, event.event_type, event.data
+                )
+            except Exception as e:
+                logger.debug("Event persistence failed for %s: %s", self.session_id, e)
+
+        await self.event_queue.put(event)
         await self._enqueue_auto_notification_requests(event)
 
         # Mid-turn heartbeat flush (owned by telemetry module).
@@ -155,6 +166,25 @@ class Session:
 
         HeartbeatSaver.maybe_fire(self)
 
+    def _schedule_trace_message(self, message: Any) -> None:
+        """Best-effort append-only trace save for SFT/KPI export."""
+        if self.persistence_store is None:
+            return
+        try:
+            payload = message.model_dump(mode="json")
+        except Exception:
+            return
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        source = str(payload.get("role") or "message")
+        loop.create_task(
+            self.persistence_store.append_trace_message(
+                self.session_id, payload, source=source
+            )
+        )
+
     def set_notification_destinations(self, destinations: list[str]) -> None:
         """Replace the session's opted-in auto-notification destinations."""
         deduped: list[str] = []

agent/core/session_persistence.py

@@ -0,0 +1,428 @@
+"""Optional durable session persistence for the hosted backend.
+
+The public CLI must keep working without MongoDB.  This module therefore
+exposes one small async store interface and returns a no-op implementation
+unless ``MONGODB_URI`` is configured and reachable.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from datetime import UTC, datetime
+from typing import Any
+
+from bson import BSON
+from pymongo import AsyncMongoClient, DeleteMany, ReturnDocument, UpdateOne
+from pymongo.errors import DuplicateKeyError, InvalidDocument, PyMongoError
+
+logger = logging.getLogger(__name__)
+
+SCHEMA_VERSION = 1
+MAX_BSON_BYTES = 15 * 1024 * 1024
+
+
+def _now() -> datetime:
+    return datetime.now(UTC)
+
+
+def _doc_id(session_id: str, idx: int) -> str:
+    return f"{session_id}:{idx}"
+
+
+def _safe_message_doc(message: dict[str, Any]) -> dict[str, Any]:
+    """Return a Mongo-safe message document payload.
+
+    Mongo's hard document limit is 16 MB.  We stay below that and store an
+    explicit marker rather than failing the whole snapshot for one huge tool log.
+    """
+    try:
+        if len(BSON.encode({"message": message})) <= MAX_BSON_BYTES:
+            return message
+    except (InvalidDocument, OverflowError):
+        pass
+    return {
+        "role": "tool",
+        "content": (
+            "[SYSTEM: A single persisted message exceeded MongoDB's document "
+            "size/encoding limit and was replaced by this marker.]"
+        ),
+        "ml_intern_persistence_error": "message_too_large_or_invalid",
+    }
+
+
+class NoopSessionStore:
+    """Async no-op store used when Mongo is not configured."""
+
+    enabled = False
+
+    async def init(self) -> None:
+        return None
+
+    async def close(self) -> None:
+        return None
+
+    async def upsert_session(self, **_: Any) -> None:
+        return None
+
+    async def save_snapshot(self, **_: Any) -> None:
+        return None
+
+    async def load_session(self, *_: Any, **__: Any) -> dict[str, Any] | None:
+        return None
+
+    async def list_sessions(self, *_: Any, **__: Any) -> list[dict[str, Any]]:
+        return []
+
+    async def soft_delete_session(self, *_: Any, **__: Any) -> None:
+        return None
+
+    async def update_session_fields(self, *_: Any, **__: Any) -> None:
+        return None
+
+    async def append_event(self, *_: Any, **__: Any) -> int | None:
+        return None
+
+    async def load_events_after(self, *_: Any, **__: Any) -> list[dict[str, Any]]:
+        return []
+
+    async def append_trace_message(self, *_: Any, **__: Any) -> int | None:
+        return None
+
+    async def get_quota(self, *_: Any, **__: Any) -> int | None:
+        return None
+
+    async def try_increment_quota(self, *_: Any, **__: Any) -> int | None:
+        return None
+
+    async def refund_quota(self, *_: Any, **__: Any) -> None:
+        return None
+
+
+class MongoSessionStore(NoopSessionStore):
+    """MongoDB-backed session store."""
+
+    enabled = True
+
+    def __init__(self, uri: str, db_name: str) -> None:
+        self.uri = uri
+        self.db_name = db_name
+        self.enabled = False
+        self.client: AsyncMongoClient | None = None
+        self.db = None
+
+    async def init(self) -> None:
+        try:
+            self.client = AsyncMongoClient(self.uri, serverSelectionTimeoutMS=3000)
+            self.db = self.client[self.db_name]
+            await self.client.admin.command("ping")
+            await self._create_indexes()
+            self.enabled = True
+            logger.info("Mongo session persistence enabled (db=%s)", self.db_name)
+        except Exception as e:
+            logger.warning("Mongo session persistence disabled: %s", e)
+            self.enabled = False
+            if self.client is not None:
+                await self.client.close()
+            self.client = None
+            self.db = None
+
+    async def close(self) -> None:
+        if self.client is not None:
+            await self.client.close()
+        self.client = None
+        self.db = None
+
+    async def _create_indexes(self) -> None:
+        if self.db is None:
+            return
+        await self.db.sessions.create_index(
+            [("user_id", 1), ("visibility", 1), ("updated_at", -1)]
+        )
+        await self.db.sessions.create_index(
+            [("visibility", 1), ("status", 1), ("last_active_at", -1)]
+        )
+        await self.db.session_messages.create_index(
+            [("session_id", 1), ("idx", 1)], unique=True
+        )
+        await self.db.session_events.create_index(
+            [("session_id", 1), ("seq", 1)], unique=True
+        )
+        await self.db.session_trace_messages.create_index(
+            [("session_id", 1), ("seq", 1)], unique=True
+        )
+        await self.db.session_trace_messages.create_index([("created_at", -1)])
+
+    def _ready(self) -> bool:
+        return bool(self.enabled and self.db is not None)
+
+    async def upsert_session(
+        self,
+        *,
+        session_id: str,
+        user_id: str,
+        model: str,
+        title: str | None = None,
+        surface: str = "frontend",
+        created_at: datetime | None = None,
+        runtime_state: str = "idle",
+        status: str = "active",
+        message_count: int = 0,
+        turn_count: int = 0,
+        pending_approval: list[dict[str, Any]] | None = None,
+        claude_counted: bool = False,
+        notification_destinations: list[str] | None = None,
+    ) -> None:
+        if not self._ready():
+            return
+        now = _now()
+        await self.db.sessions.update_one(
+            {"_id": session_id},
+            {
+                "$setOnInsert": {
+                    "_id": session_id,
+                    "session_id": session_id,
+                    "user_id": user_id,
+                    "surface": surface,
+                    "created_at": created_at or now,
+                    "schema_version": SCHEMA_VERSION,
+                    "visibility": "live",
+                },
+                "$set": {
+                    "title": title,
+                    "model": model,
+                    "status": status,
+                    "runtime_state": runtime_state,
+                    "updated_at": now,
+                    "last_active_at": now,
+                    "message_count": message_count,
+                    "turn_count": turn_count,
+                    "pending_approval": pending_approval or [],
+                    "claude_counted": claude_counted,
+                    "notification_destinations": notification_destinations or [],
+                },
+            },
+            upsert=True,
+        )
+
+    async def save_snapshot(
+        self,
+        *,
+        session_id: str,
+        user_id: str,
+        model: str,
+        messages: list[dict[str, Any]],
+        title: str | None = None,
+        runtime_state: str = "idle",
+        status: str = "active",
+        turn_count: int = 0,
+        pending_approval: list[dict[str, Any]] | None = None,
+        claude_counted: bool = False,
+        created_at: datetime | None = None,
+        notification_destinations: list[str] | None = None,
+    ) -> None:
+        if not self._ready():
+            return
+        now = _now()
+        await self.upsert_session(
+            session_id=session_id,
+            user_id=user_id,
+            model=model,
+            title=title,
+            created_at=created_at,
+            runtime_state=runtime_state,
+            status=status,
+            message_count=len(messages),
+            turn_count=turn_count,
+            pending_approval=pending_approval,
+            claude_counted=claude_counted,
+            notification_destinations=notification_destinations,
+        )
+        ops: list[Any] = []
+        for idx, raw in enumerate(messages):
+            ops.append(
+                UpdateOne(
+                    {"_id": _doc_id(session_id, idx)},
+                    {
+                        "$set": {
+                            "session_id": session_id,
+                            "idx": idx,
+                            "message": _safe_message_doc(raw),
+                            "updated_at": now,
+                        },
+                        "$setOnInsert": {"created_at": now},
+                    },
+                    upsert=True,
+                )
+            )
+        ops.append(DeleteMany({"session_id": session_id, "idx": {"$gte": len(messages)}}))
+        try:
+            if ops:
+                await self.db.session_messages.bulk_write(ops, ordered=False)
+        except PyMongoError as e:
+            logger.warning("Failed to persist session %s snapshot: %s", session_id, e)
+
+    async def load_session(
+        self, session_id: str, *, include_deleted: bool = False
+    ) -> dict[str, Any] | None:
+        if not self._ready():
+            return None
+        meta = await self.db.sessions.find_one({"_id": session_id})
+        if not meta:
+            return None
+        if meta.get("visibility") == "deleted" and not include_deleted:
+            return None
+        cursor = self.db.session_messages.find({"session_id": session_id}).sort("idx", 1)
+        messages = [row.get("message") async for row in cursor]
+        return {"metadata": meta, "messages": messages}
+
+    async def list_sessions(
+        self, user_id: str, *, include_deleted: bool = False
+    ) -> list[dict[str, Any]]:
+        if not self._ready():
+            return []
+        query: dict[str, Any] = {"user_id": user_id}
+        if user_id == "dev":
+            query = {}
+        if not include_deleted:
+            query["visibility"] = {"$ne": "deleted"}
+        cursor = self.db.sessions.find(query).sort("updated_at", -1)
+        return [row async for row in cursor]
+
+    async def soft_delete_session(self, session_id: str) -> None:
+        if not self._ready():
+            return
+        await self.db.sessions.update_one(
+            {"_id": session_id},
+            {
+                "$set": {
+                    "visibility": "deleted",
+                    "runtime_state": "idle",
+                    "updated_at": _now(),
+                }
+            },
+        )
+
+    async def update_session_fields(self, session_id: str, **fields: Any) -> None:
+        if not self._ready() or not fields:
+            return
+        fields["updated_at"] = _now()
+        await self.db.sessions.update_one({"_id": session_id}, {"$set": fields})
+
+    async def _next_seq(self, counter_id: str) -> int:
+        doc = await self.db.counters.find_one_and_update(
+            {"_id": counter_id},
+            {"$inc": {"seq": 1}},
+            upsert=True,
+            return_document=ReturnDocument.AFTER,
+        )
+        return int(doc["seq"])
+
+    async def append_event(
+        self, session_id: str, event_type: str, data: dict[str, Any] | None
+    ) -> int | None:
+        if not self._ready():
+            return None
+        try:
+            seq = await self._next_seq(f"event:{session_id}")
+            await self.db.session_events.insert_one(
+                {
+                    "_id": _doc_id(session_id, seq),
+                    "session_id": session_id,
+                    "seq": seq,
+                    "event_type": event_type,
+                    "data": data or {},
+                    "created_at": _now(),
+                }
+            )
+            return seq
+        except PyMongoError as e:
+            logger.debug("Failed to append event for %s: %s", session_id, e)
+            return None
+
+    async def load_events_after(self, session_id: str, after_seq: int = 0) -> list[dict[str, Any]]:
+        if not self._ready():
+            return []
+        cursor = self.db.session_events.find(
+            {"session_id": session_id, "seq": {"$gt": int(after_seq or 0)}}
+        ).sort("seq", 1)
+        return [row async for row in cursor]
+
+    async def append_trace_message(
+        self, session_id: str, message: dict[str, Any], source: str = "message"
+    ) -> int | None:
+        if not self._ready():
+            return None
+        try:
+            seq = await self._next_seq(f"trace:{session_id}")
+            await self.db.session_trace_messages.insert_one(
+                {
+                    "_id": _doc_id(session_id, seq),
+                    "session_id": session_id,
+                    "seq": seq,
+                    "role": message.get("role"),
+                    "message": _safe_message_doc(message),
+                    "source": source,
+                    "created_at": _now(),
+                }
+            )
+            return seq
+        except PyMongoError as e:
+            logger.debug("Failed to append trace message for %s: %s", session_id, e)
+            return None
+
+    async def get_quota(self, user_id: str, day: str) -> int | None:
+        if not self._ready():
+            return None
+        doc = await self.db.claude_quotas.find_one({"_id": f"{user_id}:{day}"})
+        return int(doc.get("count", 0)) if doc else 0
+
+    async def try_increment_quota(self, user_id: str, day: str, cap: int) -> int | None:
+        if not self._ready():
+            return None
+        key = f"{user_id}:{day}"
+        now = _now()
+        try:
+            await self.db.claude_quotas.insert_one(
+                {
+                    "_id": key,
+                    "user_id": user_id,
+                    "day": day,
+                    "count": 1,
+                    "updated_at": now,
+                }
+            )
+            return 1
+        except DuplicateKeyError:
+            pass
+        doc = await self.db.claude_quotas.find_one_and_update(
+            {"_id": key, "count": {"$lt": cap}},
+            {"$inc": {"count": 1}, "$set": {"updated_at": now}},
+            return_document=ReturnDocument.AFTER,
+        )
+        return int(doc["count"]) if doc else None
+
+    async def refund_quota(self, user_id: str, day: str) -> None:
+        if not self._ready():
+            return
+        await self.db.claude_quotas.update_one(
+            {"_id": f"{user_id}:{day}", "count": {"$gt": 0}},
+            {"$inc": {"count": -1}, "$set": {"updated_at": _now()}},
+        )
+
+
+_store: NoopSessionStore | MongoSessionStore | None = None
+
+
+def get_session_store() -> NoopSessionStore | MongoSessionStore:
+    global _store
+    if _store is None:
+        uri = os.environ.get("MONGODB_URI")
+        db_name = os.environ.get("MONGODB_DB", "ml-intern")
+        _store = MongoSessionStore(uri, db_name) if uri else NoopSessionStore()
+    return _store
+
+
+def _reset_store_for_tests(store: NoopSessionStore | MongoSessionStore | None = None) -> None:
+    global _store
+    _store = store

backend/main.py

@@ -6,6 +6,11 @@ from contextlib import asynccontextmanager
 from pathlib import Path
 
 from dotenv import load_dotenv
+
+# Load .env before importing routes/session_manager so persistence and quota
+# modules see local Mongo settings during startup.
+load_dotenv(Path(__file__).parent.parent / ".env")
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
@@ -13,9 +18,6 @@ from routes.agent import router as agent_router
 from routes.auth import router as auth_router
 from session_manager import session_manager
 
-# Load .env from project root (parent directory)
-load_dotenv(Path(__file__).parent.parent / ".env")
-
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,

backend/models.py

@@ -87,6 +87,7 @@ class SessionInfo(BaseModel):
     user_id: str = "dev"
     pending_approval: list[PendingApprovalTool] | None = None
     model: str | None = None
+    title: str | None = None
     notification_destinations: list[str] = Field(default_factory=list)
 
 

backend/routes/agent.py

@@ -120,9 +120,9 @@ async def _enforce_claude_quota(
     if not _is_anthropic_model(model_name):
         return
     user_id = user["user_id"]
-    used = await user_quotas.get_claude_used_today(user_id)
     cap = user_quotas.daily_cap_for(user.get("plan"))
-    if used >= cap:
+    new_count = await user_quotas.try_increment_claude(user_id, cap)
+    if new_count is None:
         raise HTTPException(
             status_code=429,
             detail={
@@ -135,8 +135,8 @@ async def _enforce_claude_quota(
                 ),
             },
         )
-    await user_quotas.increment_claude(user_id)
     agent_session.claude_counted = True
+    await session_manager.persist_session_snapshot(agent_session)
 
 
 async def _enforce_jobs_access_for_approvals(
@@ -241,13 +241,23 @@ async def _enforce_jobs_access_for_approvals(
     )
 
 
-def _check_session_access(session_id: str, user: dict[str, Any]) -> None:
-    """Verify the user has access to the given session. Raises 403 or 404."""
-    info = session_manager.get_session_info(session_id)
-    if not info:
+async def _check_session_access(
+    session_id: str,
+    user: dict[str, Any],
+    request: Request | None = None,
+) -> AgentSession:
+    """Verify and lazily load the user's session. Raises 403 or 404."""
+    hf_token = resolve_hf_request_token(request) if request is not None else user.get("hf_token")
+    agent_session = await session_manager.ensure_session_loaded(
+        session_id,
+        user["user_id"],
+        hf_token=hf_token,
+    )
+    if not agent_session:
         raise HTTPException(status_code=404, detail="Session not found")
-    if not session_manager.verify_session_access(session_id, user["user_id"]):
+    if user["user_id"] != "dev" and agent_session.user_id not in {user["user_id"], "dev"}:
         raise HTTPException(status_code=403, detail="Access denied to this session")
+    return agent_session
 
 
 @router.get("/health", response_model=HealthResponse)
@@ -369,11 +379,21 @@ async def generate_title(
         title = title.translate(_TITLE_STRIP_CHARS).strip()
         if len(title) > 50:
             title = title[:50].rstrip() + "…"
+        try:
+            await _check_session_access(request.session_id, user)
+            await session_manager.update_session_title(request.session_id, title)
+        except Exception:
+            logger.debug("Skipping title persistence for missing session %s", request.session_id)
         return {"title": title}
     except Exception as e:
         logger.warning(f"Title generation failed: {e}")
         fallback = request.text.strip()
         title = fallback[:40].rstrip() + "…" if len(fallback) > 40 else fallback
+        try:
+            await _check_session_access(request.session_id, user)
+            await session_manager.update_session_title(request.session_id, title)
+        except Exception:
+            logger.debug("Skipping fallback title persistence for missing session %s", request.session_id)
         return {"title": title}
 
 
@@ -477,7 +497,7 @@ async def get_session(
     session_id: str, user: dict = Depends(get_current_user)
 ) -> SessionInfo:
     """Get session information. Only accessible by the session owner."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     info = session_manager.get_session_info(session_id)
     return SessionInfo(**info)
 
@@ -498,7 +518,7 @@ async def set_session_model(
     Switching TO an Anthropic model requires HF org membership (PR #63);
     free-model switches are unrestricted.
     """
-    _check_session_access(session_id, user)
+    agent_session = await _check_session_access(session_id, user, request)
     model_id = body.get("model")
     if not model_id:
         raise HTTPException(status_code=400, detail="Missing 'model' field")
@@ -506,10 +526,9 @@ async def set_session_model(
     if model_id not in valid_ids:
         raise HTTPException(status_code=400, detail=f"Unknown model: {model_id}")
     await _require_hf_for_anthropic(request, model_id)
-    agent_session = session_manager.sessions.get(session_id)
     if not agent_session:
         raise HTTPException(status_code=404, detail="Session not found")
-    agent_session.session.update_model(model_id)
+    await session_manager.update_session_model(session_id, model_id)
     logger.info(
         f"Session {session_id} model → {model_id} "
         f"(by {user.get('username', 'unknown')})"
@@ -524,13 +543,14 @@ async def set_session_notifications(
     user: dict = Depends(get_current_user),
 ) -> dict:
     """Replace the session's auto-notification destinations."""
-    _check_session_access(session_id, user)
+    agent_session = await _check_session_access(session_id, user)
     try:
         destinations = session_manager.set_notification_destinations(
             session_id, body.destinations
         )
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
+    await session_manager.persist_session_snapshot(agent_session)
     return {
         "session_id": session_id,
         "notification_destinations": destinations,
@@ -568,7 +588,7 @@ async def get_jobs_access_info(request: Request, user: dict = Depends(get_curren
 @router.get("/sessions", response_model=list[SessionInfo])
 async def list_sessions(user: dict = Depends(get_current_user)) -> list[SessionInfo]:
     """List sessions belonging to the authenticated user."""
-    sessions = session_manager.list_sessions(user_id=user["user_id"])
+    sessions = await session_manager.list_sessions(user_id=user["user_id"])
     return [SessionInfo(**s) for s in sessions]
 
 
@@ -577,7 +597,7 @@ async def delete_session(
     session_id: str, user: dict = Depends(get_current_user)
 ) -> dict:
     """Delete a session. Only accessible by the session owner."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     success = await session_manager.delete_session(session_id)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found")
@@ -589,10 +609,8 @@ async def submit_input(
     request: SubmitRequest, user: dict = Depends(get_current_user)
 ) -> dict:
     """Submit user input to a session. Only accessible by the session owner."""
-    _check_session_access(request.session_id, user)
-    agent_session = session_manager.sessions.get(request.session_id)
-    if agent_session is not None:
-        await _enforce_claude_quota(user, agent_session)
+    agent_session = await _check_session_access(request.session_id, user)
+    await _enforce_claude_quota(user, agent_session)
     success = await session_manager.submit_user_input(request.session_id, request.text)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
@@ -604,10 +622,7 @@ async def submit_approval(
     request: ApprovalRequest, user: dict = Depends(get_current_user)
 ) -> dict:
     """Submit tool approvals to a session. Only accessible by the session owner."""
-    _check_session_access(request.session_id, user)
-    agent_session = session_manager.sessions.get(request.session_id)
-    if agent_session is None:
-        raise HTTPException(status_code=404, detail="Session not found or inactive")
+    agent_session = await _check_session_access(request.session_id, user)
     approvals = [
         {
             "tool_call_id": a.tool_call_id,
@@ -632,9 +647,7 @@ async def chat_sse(
     user: dict = Depends(get_current_user),
 ) -> StreamingResponse:
     """SSE endpoint: submit input or approval, then stream events until turn ends."""
-    _check_session_access(session_id, user)
-
-    agent_session = session_manager.sessions.get(session_id)
+    agent_session = await _check_session_access(session_id, user, request)
     if not agent_session or not agent_session.is_active:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
 
@@ -700,10 +713,7 @@ async def record_pro_click(
     user: dict = Depends(get_current_user),
 ) -> dict:
     """Record a click on a Pro upgrade CTA shown from inside a session."""
-    _check_session_access(session_id, user)
-    agent_session = session_manager.sessions.get(session_id)
-    if not agent_session:
-        raise HTTPException(status_code=404, detail="Session not found")
+    agent_session = await _check_session_access(session_id, user)
 
     from agent.core import telemetry
     await telemetry.record_pro_cta_click(
@@ -725,12 +735,53 @@ _TERMINAL_EVENTS = {"turn_complete", "approval_required", "error", "interrupted"
 _SSE_KEEPALIVE_SECONDS = 15
 
 
-def _sse_response(broadcaster, event_queue, sub_id) -> StreamingResponse:
+def _last_event_seq(request: Request) -> int:
+    raw = request.headers.get("last-event-id") or request.query_params.get("after") or "0"
+    try:
+        return max(0, int(raw))
+    except (TypeError, ValueError):
+        return 0
+
+
+def _format_sse(msg: dict[str, Any]) -> str:
+    seq = msg.get("seq")
+    body = {"event_type": msg.get("event_type"), "data": msg.get("data") or {}}
+    if seq is not None:
+        body["seq"] = seq
+        return f"id: {seq}\ndata: {json.dumps(body)}\n\n"
+    return f"data: {json.dumps(body)}\n\n"
+
+
+def _event_doc_to_msg(doc: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "event_type": doc.get("event_type"),
+        "data": doc.get("data") or {},
+        "seq": doc.get("seq"),
+    }
+
+
+def _sse_response(
+    broadcaster,
+    event_queue,
+    sub_id,
+    *,
+    replay_events: list[dict[str, Any]] | None = None,
+    after_seq: int = 0,
+) -> StreamingResponse:
     """Build a StreamingResponse that drains *event_queue* as SSE,
     sending keepalive comments every 15 s to prevent proxy timeouts."""
 
     async def event_generator():
         try:
+            for doc in replay_events or []:
+                msg = _event_doc_to_msg(doc)
+                seq = msg.get("seq")
+                if isinstance(seq, int) and seq <= after_seq:
+                    continue
+                yield _format_sse(msg)
+                if msg.get("event_type", "") in _TERMINAL_EVENTS:
+                    return
+
             while True:
                 try:
                     msg = await asyncio.wait_for(
@@ -741,7 +792,7 @@ def _sse_response(broadcaster, event_queue, sub_id) -> StreamingResponse:
                     yield ": keepalive\n\n"
                     continue
                 event_type = msg.get("event_type", "")
-                yield f"data: {json.dumps(msg)}\n\n"
+                yield _format_sse(msg)
                 if event_type in _TERMINAL_EVENTS:
                     break
         finally:
@@ -761,6 +812,7 @@ def _sse_response(broadcaster, event_queue, sub_id) -> StreamingResponse:
 @router.get("/events/{session_id}")
 async def subscribe_events(
     session_id: str,
+    request: Request,
     user: dict = Depends(get_current_user),
 ) -> StreamingResponse:
     """Subscribe to events for a running session without submitting new input.
@@ -768,15 +820,21 @@ async def subscribe_events(
     Used by the frontend to re-attach after a connection drop (e.g. screen
     sleep).  Returns 404 if the session isn't active or isn't processing.
     """
-    _check_session_access(session_id, user)
-
-    agent_session = session_manager.sessions.get(session_id)
+    agent_session = await _check_session_access(session_id, user, request)
     if not agent_session or not agent_session.is_active:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
 
+    after_seq = _last_event_seq(request)
+    replay_events = await session_manager._store().load_events_after(session_id, after_seq)
     broadcaster = agent_session.broadcaster
     sub_id, event_queue = broadcaster.subscribe()
-    return _sse_response(broadcaster, event_queue, sub_id)
+    return _sse_response(
+        broadcaster,
+        event_queue,
+        sub_id,
+        replay_events=replay_events,
+        after_seq=after_seq,
+    )
 
 
 @router.post("/interrupt/{session_id}")
@@ -784,7 +842,7 @@ async def interrupt_session(
     session_id: str, user: dict = Depends(get_current_user)
 ) -> dict:
     """Interrupt the current operation in a session."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     success = await session_manager.interrupt(session_id)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
@@ -796,17 +854,16 @@ async def get_session_messages(
     session_id: str, user: dict = Depends(get_current_user)
 ) -> list[dict]:
     """Return the session's message history from memory."""
-    _check_session_access(session_id, user)
-    agent_session = session_manager.sessions.get(session_id)
+    agent_session = await _check_session_access(session_id, user)
     if not agent_session or not agent_session.is_active:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
-    return [msg.model_dump() for msg in agent_session.session.context_manager.items]
+    return [msg.model_dump(mode="json") for msg in agent_session.session.context_manager.items]
 
 
 @router.post("/undo/{session_id}")
 async def undo_session(session_id: str, user: dict = Depends(get_current_user)) -> dict:
     """Undo the last turn in a session."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     success = await session_manager.undo(session_id)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
@@ -818,7 +875,7 @@ async def truncate_session(
     session_id: str, body: TruncateRequest, user: dict = Depends(get_current_user)
 ) -> dict:
     """Truncate conversation to before a specific user message."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     success = await session_manager.truncate(session_id, body.user_message_index)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found, inactive, or message index out of range")
@@ -830,7 +887,7 @@ async def compact_session(
     session_id: str, user: dict = Depends(get_current_user)
 ) -> dict:
     """Compact the context in a session."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     success = await session_manager.compact(session_id)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
@@ -842,7 +899,7 @@ async def shutdown_session(
     session_id: str, user: dict = Depends(get_current_user)
 ) -> dict:
     """Shutdown a session."""
-    _check_session_access(session_id, user)
+    await _check_session_access(session_id, user)
     success = await session_manager.shutdown_session(session_id)
     if not success:
         raise HTTPException(status_code=404, detail="Session not found or inactive")
@@ -860,10 +917,7 @@ async def submit_feedback(
            turn_index?: int, comment?: str, message_id?: str}
     Appended as a `feedback` event and saved with the session trajectory.
     """
-    _check_session_access(session_id, user)
-    agent_session = session_manager.sessions.get(session_id)
-    if not agent_session:
-        raise HTTPException(status_code=404, detail="Session not found")
+    agent_session = await _check_session_access(session_id, user)
 
     rating = body.get("rating")
     if rating not in {"up", "down", "outcome_success", "outcome_fail"}:

backend/session_manager.py

@@ -1,6 +1,7 @@
 """Session manager for handling multiple concurrent agent sessions."""
 
 import asyncio
+import json
 import logging
 import uuid
 from dataclasses import dataclass, field
@@ -12,6 +13,7 @@ from agent.config import load_config
 from agent.core.agent_loop import process_submission
 from agent.messaging.gateway import NotificationGateway
 from agent.core.session import Event, OpType, Session
+from agent.core.session_persistence import get_session_store
 from agent.core.tools import ToolRouter
 
 # Get project root (parent of backend directory)
@@ -42,9 +44,8 @@ logger = logging.getLogger(__name__)
 class EventBroadcaster:
     """Reads from the agent's event queue and fans out to SSE subscribers.
 
-    Events that arrive when no subscribers are listening are discarded.
-    With SSE each turn is a separate request, so there is no reconnect
-    scenario that would need buffered replay.
+    Events that arrive when no subscribers are listening are discarded by
+    this in-memory fanout. Durable replay is handled by session_persistence.
     """
 
     def __init__(self, event_queue: asyncio.Queue):
@@ -68,7 +69,7 @@ class EventBroadcaster:
         while True:
             try:
                 event: Event = await self._source.get()
-                msg = {"event_type": event.event_type, "data": event.data}
+                msg = {"event_type": event.event_type, "data": event.data, "seq": event.seq}
                 for q in self._subscribers.values():
                     await q.put(msg)
             except asyncio.CancelledError:
@@ -92,6 +93,7 @@ class AgentSession:
     is_active: bool = True
     is_processing: bool = False  # True while a submission is being executed
     broadcaster: Any = None
+    title: str | None = None
     # True once this session has been counted against the user's daily
     # Claude quota. Guards double-counting when the user re-selects an
     # Anthropic model mid-session.
@@ -123,14 +125,24 @@ class SessionManager:
         self.messaging_gateway = NotificationGateway(self.config.messaging)
         self.sessions: dict[str, AgentSession] = {}
         self._lock = asyncio.Lock()
+        self.persistence_store = None
 
     async def start(self) -> None:
         """Start shared background resources."""
+        self.persistence_store = get_session_store()
+        await self.persistence_store.init()
         await self.messaging_gateway.start()
 
     async def close(self) -> None:
         """Flush and close shared background resources."""
         await self.messaging_gateway.close()
+        if self.persistence_store is not None:
+            await self.persistence_store.close()
+
+    def _store(self):
+        if self.persistence_store is None:
+            self.persistence_store = get_session_store()
+        return self.persistence_store
 
     def _count_user_sessions(self, user_id: str) -> int:
         """Count active sessions owned by a specific user."""
@@ -140,6 +152,314 @@ class SessionManager:
             if s.user_id == user_id and s.is_active
         )
 
+    def _create_session_sync(
+        self,
+        *,
+        session_id: str,
+        user_id: str,
+        hf_token: str | None,
+        model: str | None,
+        event_queue: asyncio.Queue,
+        notification_destinations: list[str] | None = None,
+    ) -> tuple[ToolRouter, Session]:
+        """Build blocking per-session resources in a worker thread."""
+        import time as _time
+
+        t0 = _time.monotonic()
+        tool_router = ToolRouter(self.config.mcpServers, hf_token=hf_token)
+        # Deep-copy config so each session's model switches independently —
+        # tab A picking GLM doesn't flip tab B off Claude.
+        session_config = self.config.model_copy(deep=True)
+        if model:
+            session_config.model_name = model
+        session = Session(
+            event_queue=event_queue,
+            config=session_config,
+            tool_router=tool_router,
+            hf_token=hf_token,
+            user_id=user_id,
+            notification_gateway=self.messaging_gateway,
+            notification_destinations=notification_destinations or [],
+            session_id=session_id,
+            persistence_store=self._store(),
+        )
+        t1 = _time.monotonic()
+        logger.info("Session initialized in %.2fs", t1 - t0)
+        return tool_router, session
+
+    def _serialize_messages(self, session: Session) -> list[dict[str, Any]]:
+        return [
+            msg.model_dump(mode="json")
+            for msg in session.context_manager.items
+        ]
+
+    def _serialize_pending_approval(self, session: Session) -> list[dict[str, Any]]:
+        pending = session.pending_approval or {}
+        tool_calls = pending.get("tool_calls") or []
+        serialized: list[dict[str, Any]] = []
+        for tc in tool_calls:
+            if hasattr(tc, "model_dump"):
+                serialized.append(tc.model_dump(mode="json"))
+            elif isinstance(tc, dict):
+                serialized.append(tc)
+        return serialized
+
+    @staticmethod
+    def _pending_tools_for_api(session: Session) -> list[dict[str, Any]] | None:
+        pending = session.pending_approval or {}
+        tool_calls = pending.get("tool_calls") or []
+        if not tool_calls:
+            return None
+        result: list[dict[str, Any]] = []
+        for tc in tool_calls:
+            try:
+                args = json.loads(tc.function.arguments)
+            except (json.JSONDecodeError, AttributeError, TypeError):
+                args = {}
+            result.append(
+                {
+                    "tool": getattr(tc.function, "name", None),
+                    "tool_call_id": getattr(tc, "id", None),
+                    "arguments": args,
+                }
+            )
+        return result
+
+    def _restore_pending_approval(
+        self, session: Session, pending_approval: list[dict[str, Any]] | None
+    ) -> None:
+        if not pending_approval:
+            session.pending_approval = None
+            return
+        from litellm import ChatCompletionMessageToolCall as ToolCall
+
+        restored = []
+        for raw in pending_approval:
+            try:
+                if "function" in raw:
+                    restored.append(ToolCall(**raw))
+                else:
+                    restored.append(
+                        ToolCall(
+                            id=raw["tool_call_id"],
+                            type="function",
+                            function={
+                                "name": raw["tool"],
+                                "arguments": json.dumps(raw.get("arguments") or {}),
+                            },
+                        )
+                    )
+            except Exception as e:
+                logger.warning("Dropping malformed pending approval: %s", e)
+        session.pending_approval = {"tool_calls": restored} if restored else None
+
+    @staticmethod
+    def _pending_docs_for_api(
+        pending_approval: list[dict[str, Any]] | None,
+    ) -> list[dict[str, Any]] | None:
+        if not pending_approval:
+            return None
+        result: list[dict[str, Any]] = []
+        for raw in pending_approval:
+            if "function" in raw:
+                function = raw.get("function") or {}
+                try:
+                    args = json.loads(function.get("arguments") or "{}")
+                except (json.JSONDecodeError, TypeError):
+                    args = {}
+                result.append(
+                    {
+                        "tool": function.get("name"),
+                        "tool_call_id": raw.get("id"),
+                        "arguments": args,
+                    }
+                )
+            elif {"tool", "tool_call_id"}.issubset(raw):
+                result.append(
+                    {
+                        "tool": raw.get("tool"),
+                        "tool_call_id": raw.get("tool_call_id"),
+                        "arguments": raw.get("arguments") or {},
+                    }
+                )
+        return result or None
+
+    @staticmethod
+    def _runtime_state(agent_session: AgentSession) -> str:
+        if agent_session.session.pending_approval:
+            return "waiting_approval"
+        if agent_session.is_processing:
+            return "processing"
+        if not agent_session.is_active:
+            return "ended"
+        return "idle"
+
+    async def _start_agent_session(
+        self,
+        *,
+        agent_session: AgentSession,
+        event_queue: asyncio.Queue,
+        tool_router: ToolRouter,
+    ) -> AgentSession:
+        async with self._lock:
+            existing = self.sessions.get(agent_session.session_id)
+            if existing:
+                return existing
+            self.sessions[agent_session.session_id] = agent_session
+
+        task = asyncio.create_task(
+            self._run_session(
+                agent_session.session_id,
+                agent_session.submission_queue,
+                event_queue,
+                tool_router,
+            )
+        )
+        agent_session.task = task
+        return agent_session
+
+    @staticmethod
+    def _can_access_session(agent_session: AgentSession, user_id: str) -> bool:
+        return (
+            user_id == "dev"
+            or agent_session.user_id == "dev"
+            or agent_session.user_id == user_id
+        )
+
+    @staticmethod
+    def _update_hf_token(agent_session: AgentSession, hf_token: str | None) -> None:
+        if not hf_token:
+            return
+        agent_session.hf_token = hf_token
+        agent_session.session.hf_token = hf_token
+
+    async def persist_session_snapshot(
+        self,
+        agent_session: AgentSession,
+        *,
+        runtime_state: str | None = None,
+        status: str = "active",
+    ) -> None:
+        """Persist the current runtime context snapshot."""
+        store = self._store()
+        if not getattr(store, "enabled", False):
+            return
+        try:
+            await store.save_snapshot(
+                session_id=agent_session.session_id,
+                user_id=agent_session.user_id,
+                model=agent_session.session.config.model_name,
+                title=agent_session.title,
+                messages=self._serialize_messages(agent_session.session),
+                runtime_state=runtime_state or self._runtime_state(agent_session),
+                status=status,
+                turn_count=agent_session.session.turn_count,
+                pending_approval=self._serialize_pending_approval(agent_session.session),
+                claude_counted=agent_session.claude_counted,
+                created_at=agent_session.created_at,
+                notification_destinations=list(
+                    agent_session.session.notification_destinations
+                ),
+            )
+        except Exception as e:
+            logger.warning(
+                "Failed to persist snapshot for %s: %s",
+                agent_session.session_id,
+                e,
+            )
+
+    async def ensure_session_loaded(
+        self,
+        session_id: str,
+        user_id: str,
+        hf_token: str | None = None,
+    ) -> AgentSession | None:
+        """Return a live runtime session, lazily restoring it from Mongo."""
+        async with self._lock:
+            existing = self.sessions.get(session_id)
+        if existing:
+            if self._can_access_session(existing, user_id):
+                self._update_hf_token(existing, hf_token)
+                return existing
+            return None
+
+        store = self._store()
+        loaded = await store.load_session(session_id)
+        if not loaded:
+            return None
+
+        async with self._lock:
+            existing = self.sessions.get(session_id)
+        if existing:
+            if self._can_access_session(existing, user_id):
+                self._update_hf_token(existing, hf_token)
+                return existing
+            return None
+
+        meta = loaded.get("metadata") or {}
+        owner = str(meta.get("user_id") or "")
+        if user_id != "dev" and owner != "dev" and owner != user_id:
+            return None
+
+        from litellm import Message
+
+        model = meta.get("model") or self.config.model_name
+        event_queue: asyncio.Queue = asyncio.Queue()
+        submission_queue: asyncio.Queue = asyncio.Queue()
+        tool_router, session = await asyncio.to_thread(
+            self._create_session_sync,
+            session_id=session_id,
+            user_id=owner or user_id,
+            hf_token=hf_token,
+            model=model,
+            event_queue=event_queue,
+            notification_destinations=meta.get("notification_destinations") or [],
+        )
+
+        restored_messages: list[Message] = []
+        for raw in loaded.get("messages") or []:
+            if not isinstance(raw, dict) or raw.get("role") == "system":
+                continue
+            try:
+                restored_messages.append(Message.model_validate(raw))
+            except Exception as e:
+                logger.warning("Dropping malformed restored message: %s", e)
+        if restored_messages:
+            # Keep the freshly-rendered system prompt, then attach the durable
+            # non-system context so tools/date/user context stay current.
+            session.context_manager.items = [session.context_manager.items[0], *restored_messages]
+
+        self._restore_pending_approval(session, meta.get("pending_approval") or [])
+        session.turn_count = int(meta.get("turn_count") or 0)
+
+        created_at = meta.get("created_at")
+        if not isinstance(created_at, datetime):
+            created_at = datetime.utcnow()
+
+        agent_session = AgentSession(
+            session_id=session_id,
+            session=session,
+            tool_router=tool_router,
+            submission_queue=submission_queue,
+            user_id=owner or user_id,
+            hf_token=hf_token,
+            created_at=created_at,
+            is_active=True,
+            is_processing=False,
+            claude_counted=bool(meta.get("claude_counted")),
+            title=meta.get("title"),
+        )
+        started = await self._start_agent_session(
+            agent_session=agent_session,
+            event_queue=event_queue,
+            tool_router=tool_router,
+        )
+        if started is not agent_session:
+            self._update_hf_token(started, hf_token)
+            return started
+        logger.info("Restored session %s for user %s", session_id, owner or user_id)
+        return agent_session
+
     async def create_session(
         self,
         user_id: str = "dev",
@@ -188,31 +508,14 @@ class SessionManager:
         event_queue: asyncio.Queue = asyncio.Queue()
 
         # Run blocking constructors in a thread to keep the event loop responsive.
-        # Without this, Session.__init__ → ContextManager → litellm.get_max_tokens()
-        # blocks all HTTP/SSE handling.
-        import time as _time
-
-        def _create_session_sync():
-            t0 = _time.monotonic()
-            tool_router = ToolRouter(self.config.mcpServers, hf_token=hf_token)
-            # Deep-copy config so each session's model switches independently —
-            # tab A picking GLM doesn't flip tab B off Claude.
-            session_config = self.config.model_copy(deep=True)
-            if model:
-                session_config.model_name = model
-            session = Session(
-                event_queue, config=session_config, tool_router=tool_router,
-                hf_token=hf_token,
-                user_id=user_id,
-                notification_gateway=self.messaging_gateway,
-                notification_destinations=[],
-                session_id=session_id,
-            )
-            t1 = _time.monotonic()
-            logger.info(f"Session initialized in {t1 - t0:.2f}s")
-            return tool_router, session
-
-        tool_router, session = await asyncio.to_thread(_create_session_sync)
+        tool_router, session = await asyncio.to_thread(
+            self._create_session_sync,
+            session_id=session_id,
+            user_id=user_id,
+            hf_token=hf_token,
+            model=model,
+            event_queue=event_queue,
+        )
 
         # Create wrapper
         agent_session = AgentSession(
@@ -224,14 +527,12 @@ class SessionManager:
             hf_token=hf_token,
         )
 
-        async with self._lock:
-            self.sessions[session_id] = agent_session
-
-        # Start the agent loop task
-        task = asyncio.create_task(
-            self._run_session(session_id, submission_queue, event_queue, tool_router)
+        await self._start_agent_session(
+            agent_session=agent_session,
+            event_queue=event_queue,
+            tool_router=tool_router,
         )
-        agent_session.task = task
+        await self.persist_session_snapshot(agent_session, runtime_state="idle")
 
         logger.info(f"Created session {session_id} for user {user_id}")
         return session_id
@@ -297,6 +598,7 @@ class SessionManager:
             ),
         )
         session.context_manager.items.append(seed)
+        await self.persist_session_snapshot(agent_session, runtime_state="idle")
         return len(parsed)
 
     @staticmethod
@@ -367,6 +669,7 @@ class SessionManager:
                             should_continue = await process_submission(session, submission)
                         finally:
                             agent_session.is_processing = False
+                            await self.persist_session_snapshot(agent_session)
                         if not should_continue:
                             break
                     except asyncio.TimeoutError:
@@ -401,6 +704,11 @@ class SessionManager:
             async with self._lock:
                 if session_id in self.sessions:
                     self.sessions[session_id].is_active = False
+                    await self.persist_session_snapshot(
+                        self.sessions[session_id],
+                        runtime_state="ended",
+                        status="ended",
+                    )
 
             logger.info(f"Session {session_id} ended")
 
@@ -450,7 +758,10 @@ class SessionManager:
             agent_session = self.sessions.get(session_id)
         if not agent_session or not agent_session.is_active:
             return False
-        return agent_session.session.context_manager.truncate_to_user_message(user_message_index)
+        success = agent_session.session.context_manager.truncate_to_user_message(user_message_index)
+        if success:
+            await self.persist_session_snapshot(agent_session, runtime_state="idle")
+        return success
 
     async def compact(self, session_id: str) -> bool:
         """Compact context in a session."""
@@ -475,12 +786,15 @@ class SessionManager:
         return success
 
     async def delete_session(self, session_id: str) -> bool:
-        """Delete a session entirely."""
+        """Soft-delete a session and stop its runtime resources."""
         async with self._lock:
             agent_session = self.sessions.pop(session_id, None)
 
         if not agent_session:
-            return False
+            await self._store().soft_delete_session(session_id)
+            return True
+
+        await self._store().soft_delete_session(session_id)
 
         # Clean up sandbox Space before cancelling the task
         await self._cleanup_sandbox(agent_session.session)
@@ -495,6 +809,21 @@ class SessionManager:
 
         return True
 
+    async def update_session_title(self, session_id: str, title: str | None) -> None:
+        """Persist a user-visible title for sidebar rehydration."""
+        agent_session = self.sessions.get(session_id)
+        if agent_session:
+            agent_session.title = title
+        await self._store().update_session_fields(session_id, title=title)
+
+    async def update_session_model(self, session_id: str, model_id: str) -> bool:
+        agent_session = self.sessions.get(session_id)
+        if not agent_session or not agent_session.is_active:
+            return False
+        agent_session.session.update_model(model_id)
+        await self.persist_session_snapshot(agent_session, runtime_state="idle")
+        return True
+
     def get_session_owner(self, session_id: str) -> str | None:
         """Get the user_id that owns a session, or None if session doesn't exist."""
         agent_session = self.sessions.get(session_id)
@@ -522,22 +851,7 @@ class SessionManager:
         if not agent_session:
             return None
 
-        # Extract pending approval tools if any
-        pending_approval = None
-        pa = agent_session.session.pending_approval
-        if pa and pa.get("tool_calls"):
-            pending_approval = []
-            for tc in pa["tool_calls"]:
-                import json
-                try:
-                    args = json.loads(tc.function.arguments)
-                except (json.JSONDecodeError, AttributeError):
-                    args = {}
-                pending_approval.append({
-                    "tool": tc.function.name,
-                    "tool_call_id": tc.id,
-                    "arguments": args,
-                })
+        pending_approval = self._pending_tools_for_api(agent_session.session)
 
         return {
             "session_id": session_id,
@@ -548,6 +862,7 @@ class SessionManager:
             "user_id": agent_session.user_id,
             "pending_approval": pending_approval,
             "model": agent_session.session.config.model_name,
+            "title": agent_session.title,
             "notification_destinations": list(
                 agent_session.session.notification_destinations
             ),
@@ -581,14 +896,46 @@ class SessionManager:
         agent_session.session.set_notification_destinations(normalized)
         return normalized
 
-    def list_sessions(self, user_id: str | None = None) -> list[dict[str, Any]]:
+    async def list_sessions(self, user_id: str | None = None) -> list[dict[str, Any]]:
         """List sessions, optionally filtered by user.
 
         Args:
             user_id: If provided, only return sessions owned by this user.
                      If "dev", return all sessions (dev mode).
         """
-        results = []
+        results: list[dict[str, Any]] = []
+        store = self._store()
+        if getattr(store, "enabled", False):
+            for row in await store.list_sessions(user_id or "dev"):
+                sid = row.get("session_id") or row.get("_id")
+                if not sid:
+                    continue
+                runtime_info = self.get_session_info(str(sid))
+                if runtime_info:
+                    results.append(runtime_info)
+                    continue
+                created_at = row.get("created_at")
+                if isinstance(created_at, datetime):
+                    created_at_str = created_at.isoformat()
+                else:
+                    created_at_str = str(created_at or datetime.utcnow().isoformat())
+                pending = self._pending_docs_for_api(row.get("pending_approval") or [])
+                results.append(
+                    {
+                        "session_id": str(sid),
+                        "created_at": created_at_str,
+                        "is_active": row.get("status") != "ended",
+                        "is_processing": row.get("runtime_state") == "processing",
+                        "message_count": int(row.get("message_count") or 0),
+                        "user_id": row.get("user_id") or "dev",
+                        "pending_approval": pending or None,
+                        "model": row.get("model"),
+                        "title": row.get("title"),
+                        "notification_destinations": row.get("notification_destinations") or [],
+                    }
+                )
+            return results
+
         for sid in self.sessions:
             info = self.get_session_info(sid)
             if not info:

backend/user_quotas.py

@@ -1,9 +1,8 @@
-"""In-memory daily quota for Claude session creations.
+"""Daily quota for Claude session creations.
 
 Tracks per-user Claude session starts against a daily cap derived from the
-user's HF plan. Caps reset at UTC midnight; the store itself is in-process
-and wipes on restart (deliberate — the cost of occasional over-subsidy at
-restart is much lower than running a DB).
+user's HF plan. MongoDB is the source of truth when configured; the
+in-process dict remains the fallback for local/dev/test runs.
 
 Unit: session *creations*, not messages. A user who selects Claude in a new
 session consumes one quota point; switching an existing Claude session to
@@ -18,6 +17,8 @@ import asyncio
 import os
 from datetime import UTC, datetime
 
+from agent.core.session_persistence import NoopSessionStore, get_session_store, _reset_store_for_tests
+
 CLAUDE_FREE_DAILY: int = int(os.environ.get("CLAUDE_FREE_DAILY", "1"))
 CLAUDE_PRO_DAILY: int = int(os.environ.get("CLAUDE_PRO_DAILY", "20"))
 
@@ -37,6 +38,11 @@ def daily_cap_for(plan: str | None) -> int:
 
 async def get_claude_used_today(user_id: str) -> int:
     """Return today's Claude session count for the user (0 if none / stale day)."""
+    store = get_session_store()
+    if getattr(store, "enabled", False):
+        db_count = await store.get_quota(user_id, _today())
+        return db_count or 0
+
     async with _lock:
         entry = _claude_counts.get(user_id)
         if entry is None:
@@ -51,11 +57,37 @@ async def get_claude_used_today(user_id: str) -> int:
 
 async def increment_claude(user_id: str) -> int:
     """Bump today's Claude session count for the user. Returns the new value."""
+    store = get_session_store()
+    if getattr(store, "enabled", False):
+        db_count = await store.try_increment_quota(user_id, _today(), cap=10**9)
+        return db_count or 0
+
+    async with _lock:
+        today = _today()
+        day, count = _claude_counts.get(user_id, (today, 0))
+        if day != today:
+            count = 0
+        count += 1
+        _claude_counts[user_id] = (today, count)
+        return count
+
+
+async def try_increment_claude(user_id: str, cap: int) -> int | None:
+    """Atomically bump today's count if below *cap*.
+
+    Returns the new count, or None when the user is already at the cap.
+    """
+    store = get_session_store()
+    if getattr(store, "enabled", False):
+        return await store.try_increment_quota(user_id, _today(), cap)
+
     async with _lock:
         today = _today()
         day, count = _claude_counts.get(user_id, (today, 0))
         if day != today:
             count = 0
+        if count >= cap:
+            return None
         count += 1
         _claude_counts[user_id] = (today, count)
         return count
@@ -63,6 +95,11 @@ async def increment_claude(user_id: str) -> int:
 
 async def refund_claude(user_id: str) -> None:
     """Decrement today's count — used when session creation fails after a successful gate."""
+    store = get_session_store()
+    if getattr(store, "enabled", False):
+        await store.refund_quota(user_id, _today())
+        return
+
     async with _lock:
         entry = _claude_counts.get(user_id)
         if entry is None:
@@ -81,3 +118,4 @@ async def refund_claude(user_id: str) -> None:
 def _reset_for_tests() -> None:
     """Test-only: clear the in-memory store."""
     _claude_counts.clear()
+    _reset_store_for_tests(NoopSessionStore())

frontend/src/components/SessionSidebar/SessionSidebar.tsx

@@ -1,4 +1,4 @@
-import { useCallback, useState } from 'react';
+import { useCallback, useEffect, useState } from 'react';
 import {
   Alert,
   Box,
@@ -25,13 +25,30 @@ interface SessionSidebarProps {
 }
 
 export default function SessionSidebar({ onClose }: SessionSidebarProps) {
-  const { sessions, activeSessionId, createSession, deleteSession, switchSession } =
+  const { sessions, activeSessionId, createSession, deleteSession, switchSession, mergeServerSessions } =
     useSessionStore();
   const { setPlan, clearPanel } =
     useAgentStore();
   const [isCreatingSession, setIsCreatingSession] = useState(false);
   const [capacityError, setCapacityError] = useState<string | null>(null);
 
+  useEffect(() => {
+    let cancelled = false;
+    (async () => {
+      try {
+        const response = await apiFetch('/api/sessions');
+        if (!response.ok) return;
+        const data = await response.json();
+        if (!cancelled && Array.isArray(data)) {
+          mergeServerSessions(data);
+        }
+      } catch {
+        /* local sidebar metadata is still usable */
+      }
+    })();
+    return () => { cancelled = true; };
+  }, [mergeServerSessions]);
+
   // -- Handlers -----------------------------------------------------------
 
   const handleNewSession = useCallback(async () => {

frontend/src/hooks/useAgentChat.ts

@@ -371,7 +371,7 @@ export function useAgentChat({ sessionId, isActive, onReady, onError, onSessionD
     } catch {
       return null;
     }
-  }, [sessionId, setNeedsAttention]);
+  }, [sessionId, setNeedsAttention, updateSession]);
 
   // -- useChat from Vercel AI SDK -----------------------------------------
   const chat = useChat({
@@ -621,7 +621,10 @@ export function useAgentChat({ sessionId, isActive, onReady, onError, onSessionD
     /** Read the event stream from GET /api/events and forward to side-channel. */
     const consumeEventStream = async (signal: AbortSignal) => {
       try {
-        const res = await apiFetch(`/api/events/${sessionId}`, {
+        const lastEventKey = `hf-agent-last-event:${sessionId}`;
+        const lastSeq = localStorage.getItem(lastEventKey);
+        const qs = lastSeq ? `?after=${encodeURIComponent(lastSeq)}` : '';
+        const res = await apiFetch(`/api/events/${sessionId}${qs}`, {
           headers: { 'Accept': 'text/event-stream' },
           signal,
         });
@@ -629,6 +632,71 @@ export function useAgentChat({ sessionId, isActive, onReady, onError, onSessionD
 
         const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
         let buf = '';
+        let eventId: string | null = null;
+        let eventData = '';
+        const dispatch = async () => {
+          if (!eventData.trim()) {
+            eventId = null;
+            eventData = '';
+            return false;
+          }
+          const event = JSON.parse(eventData.trim());
+          const seq = event.seq ?? (eventId ? Number(eventId) : undefined);
+          if (Number.isFinite(seq)) {
+            localStorage.setItem(lastEventKey, String(seq));
+          }
+          eventId = null;
+          eventData = '';
+          // Forward to side-channel for real-time UI updates
+          const et = event.event_type as string;
+          if (et === 'processing') sideChannel.onProcessing();
+          else if (et === 'assistant_chunk') sideChannel.onStreaming();
+          else if (et === 'tool_call') {
+            const t = event.data?.tool as string;
+            const d = event.data?.arguments?.description as string | undefined;
+            sideChannel.onToolRunning(t, d);
+            sideChannel.onToolCallPanel(t, (event.data?.arguments || {}) as Record<string, unknown>);
+          } else if (et === 'tool_output') {
+            sideChannel.onToolOutputPanel(
+              event.data?.tool as string,
+              event.data?.tool_call_id as string,
+              event.data?.output as string,
+              event.data?.success as boolean,
+            );
+          } else if (et === 'tool_state_change') {
+            const state = event.data?.state as string;
+            const toolName = event.data?.tool as string;
+            if (state === 'running' && toolName) sideChannel.onToolRunning(toolName);
+          } else if (et === 'turn_complete' || et === 'error' || et === 'interrupted') {
+            sideChannel.onProcessingDone();
+            stopReconnect();
+            // Final hydration to get the complete message state
+            const result = await hydrateMessages();
+            if (result) {
+              const uiMsgs = llmMessagesToUIMessages(result.data, result.pendingIds, chatActionsRef.current.messages);
+              if (uiMsgs.length > 0) {
+                chat.setMessages(uiMsgs);
+                saveMessages(sessionId, uiMsgs);
+              }
+            }
+            return true;
+          } else if (et === 'approval_required') {
+            sideChannel.onApprovalRequired(
+              (event.data?.tools || []) as Array<{ tool: string; arguments: Record<string, unknown>; tool_call_id: string }>,
+            );
+            stopReconnect();
+            const result = await hydrateMessages();
+            if (result) {
+              const uiMsgs = llmMessagesToUIMessages(result.data, result.pendingIds, chatActionsRef.current.messages);
+              if (uiMsgs.length > 0) {
+                chat.setMessages(uiMsgs);
+                saveMessages(sessionId, uiMsgs);
+              }
+            }
+            return true;
+          }
+          return false;
+        };
         while (true) {
           const { value, done } = await reader.read();
           if (done || signal.aborted) break;
@@ -636,59 +704,21 @@ export function useAgentChat({ sessionId, isActive, onReady, onError, onSessionD
           const lines = buf.split('\n');
           buf = lines.pop() || '';
           for (const line of lines) {
-            const trimmed = line.trim();
-            if (!trimmed.startsWith('data: ')) continue;
-            try {
-              const event = JSON.parse(trimmed.slice(6));
-              // Forward to side-channel for real-time UI updates
-              const et = event.event_type as string;
-              if (et === 'processing') sideChannel.onProcessing();
-              else if (et === 'assistant_chunk') sideChannel.onStreaming();
-              else if (et === 'tool_call') {
-                const t = event.data?.tool as string;
-                const d = event.data?.arguments?.description as string | undefined;
-                sideChannel.onToolRunning(t, d);
-                sideChannel.onToolCallPanel(t, (event.data?.arguments || {}) as Record<string, unknown>);
-              } else if (et === 'tool_output') {
-                sideChannel.onToolOutputPanel(
-                  event.data?.tool as string,
-                  event.data?.tool_call_id as string,
-                  event.data?.output as string,
-                  event.data?.success as boolean,
-                );
-              } else if (et === 'tool_state_change') {
-                const state = event.data?.state as string;
-                const toolName = event.data?.tool as string;
-                if (state === 'running' && toolName) sideChannel.onToolRunning(toolName);
-              } else if (et === 'turn_complete' || et === 'error' || et === 'interrupted') {
-                sideChannel.onProcessingDone();
-                stopReconnect();
-                // Final hydration to get the complete message state
-                const result = await hydrateMessages();
-                if (result) {
-                  const uiMsgs = llmMessagesToUIMessages(result.data, result.pendingIds, chatActionsRef.current.messages);
-                  if (uiMsgs.length > 0) {
-                    chat.setMessages(uiMsgs);
-                    saveMessages(sessionId, uiMsgs);
-                  }
-                }
-                return;
-              } else if (et === 'approval_required') {
-                sideChannel.onApprovalRequired(
-                  (event.data?.tools || []) as Array<{ tool: string; arguments: Record<string, unknown>; tool_call_id: string }>,
-                );
-                stopReconnect();
-                const result = await hydrateMessages();
-                if (result) {
-                  const uiMsgs = llmMessagesToUIMessages(result.data, result.pendingIds, chatActionsRef.current.messages);
-                  if (uiMsgs.length > 0) {
-                    chat.setMessages(uiMsgs);
-                    saveMessages(sessionId, uiMsgs);
-                  }
-                }
-                return;
-              }
-            } catch { /* ignore parse errors */ }
+            const trimmed = line.replace(/\r$/, '');
+            if (trimmed === '') {
+              try {
+                if (await dispatch()) return;
+              } catch { /* ignore parse errors */ }
+              continue;
+            }
+            if (trimmed.startsWith(':')) continue;
+            if (trimmed.startsWith('id:')) {
+              eventId = trimmed.slice(3).trim();
+              continue;
+            }
+            if (trimmed.startsWith('data:')) {
+              eventData += trimmed.slice(5).trimStart() + '\n';
+            }
           }
         }
       } catch {

frontend/src/lib/sse-chat-transport.ts

@@ -42,35 +42,66 @@ function nextPartId(prefix: string): string {
   return `${prefix}-${Date.now()}-${++partIdCounter}`;
 }
 
+function lastEventKey(sessionId: string): string {
+  return `hf-agent-last-event:${sessionId}`;
+}
+
 /** Parse an SSE text stream into AgentEvent objects. */
-function createSSEParserStream(): TransformStream<string, AgentEvent> {
+function createSSEParserStream(sessionId: string): TransformStream<string, AgentEvent> {
   let buffer = '';
+  let eventId: string | null = null;
+  let data = '';
+
+  const dispatch = (controller: TransformStreamDefaultController<AgentEvent>) => {
+    if (!data.trim()) {
+      eventId = null;
+      data = '';
+      return;
+    }
+    try {
+      const json = JSON.parse(data.trim()) as AgentEvent;
+      const seq = json.seq ?? (eventId ? Number(eventId) : undefined);
+      if (Number.isFinite(seq)) {
+        json.seq = seq;
+        localStorage.setItem(lastEventKey(sessionId), String(seq));
+      }
+      controller.enqueue(json);
+    } catch {
+      logger.warn('SSE parse error:', data.trim());
+    } finally {
+      eventId = null;
+      data = '';
+    }
+  };
+
   return new TransformStream<string, AgentEvent>({
     transform(chunk, controller) {
       buffer += chunk;
       const lines = buffer.split('\n');
       // Keep the last (possibly incomplete) line in the buffer
       buffer = lines.pop() || '';
-      for (const line of lines) {
-        const trimmed = line.trim();
-        if (trimmed.startsWith('data: ')) {
-          try {
-            const json = JSON.parse(trimmed.slice(6));
-            controller.enqueue(json as AgentEvent);
-          } catch {
-            logger.warn('SSE parse error:', trimmed);
-          }
+      for (const rawLine of lines) {
+        const line = rawLine.replace(/\r$/, '');
+        if (line === '') {
+          dispatch(controller);
+          continue;
+        }
+        if (line.startsWith(':')) continue;
+        if (line.startsWith('id:')) {
+          eventId = line.slice(3).trim();
+        } else if (line.startsWith('data:')) {
+          data += line.slice(5).trimStart() + '\n';
         }
       }
     },
     flush(controller) {
-      // Process any remaining data in buffer
-      if (buffer.trim().startsWith('data: ')) {
-        try {
-          const json = JSON.parse(buffer.trim().slice(6));
-          controller.enqueue(json as AgentEvent);
-        } catch { /* ignore incomplete */ }
+      const line = buffer.replace(/\r$/, '');
+      if (line.startsWith('id:')) {
+        eventId = line.slice(3).trim();
+      } else if (line.startsWith('data:')) {
+        data += line.slice(5).trimStart() + '\n';
       }
+      dispatch(controller);
     },
   });
 }
@@ -426,7 +457,7 @@ export class SSEChatTransport implements ChatTransport<UIMessage> {
     // Pipe: response bytes → text → SSE events → UIMessageChunks
     return response.body
       .pipeThrough(new TextDecoderStream())
-      .pipeThrough(createSSEParserStream())
+      .pipeThrough(createSSEParserStream(sessionId))
       .pipeThrough(createEventToChunkStream(this.sideChannel));
   }
 
@@ -441,7 +472,9 @@ export class SSEChatTransport implements ChatTransport<UIMessage> {
       if (!info.is_processing) return null;
 
       // Session is mid-turn — subscribe to its event broadcast.
-      const response = await apiFetch(`/api/events/${this.sessionId}`, {
+      const lastSeq = localStorage.getItem(lastEventKey(this.sessionId));
+      const qs = lastSeq ? `?after=${encodeURIComponent(lastSeq)}` : '';
+      const response = await apiFetch(`/api/events/${this.sessionId}${qs}`, {
         headers: { 'Accept': 'text/event-stream' },
       });
       if (!response.ok || !response.body) return null;
@@ -450,7 +483,7 @@ export class SSEChatTransport implements ChatTransport<UIMessage> {
 
       return response.body
         .pipeThrough(new TextDecoderStream())
-        .pipeThrough(createSSEParserStream())
+        .pipeThrough(createSSEParserStream(this.sessionId))
         .pipeThrough(createEventToChunkStream(this.sideChannel));
     } catch {
       return null;

frontend/src/store/sessionStore.ts

@@ -20,6 +20,14 @@ interface SessionStore {
   markExpired: (id: string) => void;
   /** Clear the expired flag (used after restore-with-summary succeeds). */
   clearExpired: (id: string) => void;
+  /** Merge durable server-side sessions into local sidebar metadata. */
+  mergeServerSessions: (sessions: Array<{
+    session_id: string;
+    title?: string | null;
+    created_at: string;
+    is_active?: boolean;
+    pending_approval?: unknown[] | null;
+  }>) => void;
   /** Atomically swap a session's id in the list + both localStorage caches.
    *  Used when we rehydrate an expired session into a freshly-created backend
    *  session — preserves title, timestamps, and messages. */
@@ -76,6 +84,45 @@ export const useSessionStore = create<SessionStore>()(
         }));
       },
 
+      mergeServerSessions: (serverSessions) => {
+        set((state) => {
+          const byId = new Map(state.sessions.map((s) => [s.id, s]));
+          const merged = [...state.sessions];
+          for (const server of serverSessions) {
+            const id = server.session_id;
+            if (!id) continue;
+            const existing = byId.get(id);
+            if (existing) {
+              const updated = {
+                ...existing,
+                title: server.title || existing.title,
+                isActive: server.is_active ?? existing.isActive,
+                needsAttention: Boolean(server.pending_approval?.length) || existing.needsAttention,
+                expired: false,
+              };
+              const idx = merged.findIndex((s) => s.id === id);
+              if (idx >= 0) merged[idx] = updated;
+              byId.set(id, updated);
+              continue;
+            }
+            const newSession: SessionMeta = {
+              id,
+              title: server.title || `Chat ${merged.length + 1}`,
+              createdAt: server.created_at || new Date().toISOString(),
+              isActive: server.is_active ?? true,
+              needsAttention: Boolean(server.pending_approval?.length),
+              expired: false,
+            };
+            merged.push(newSession);
+            byId.set(id, newSession);
+          }
+          return {
+            sessions: merged,
+            activeSessionId: state.activeSessionId || merged[merged.length - 1]?.id || null,
+          };
+        });
+      },
+
       renameSession: (oldId: string, newId: string) => {
         if (oldId === newId) return;
         moveMessages(oldId, newId);

frontend/src/types/events.ts

@@ -24,6 +24,7 @@ export type EventType =
 export interface AgentEvent {
   event_type: EventType;
   data?: Record<string, unknown>;
+  seq?: number;
 }
 
 export interface ReadyEventData {

pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
     "httpx>=0.27.0",
     "websockets>=13.0",
     "apscheduler>=3.10,<4",
+    "pymongo>=4.17.0",
 ]
 
 [project.optional-dependencies]

tests/unit/test_session_manager_persistence.py

@@ -0,0 +1,240 @@
+"""Regression tests for server-side session persistence restore/access."""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from datetime import datetime, UTC
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+_BACKEND_DIR = Path(__file__).resolve().parent.parent.parent / "backend"
+if str(_BACKEND_DIR) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_DIR))
+
+from agent.core.session_persistence import NoopSessionStore  # noqa: E402
+from session_manager import AgentSession, SessionManager  # noqa: E402
+
+
+class FakeRuntimeSession:
+    def __init__(self, *, hf_token: str | None = None, model: str = "test-model"):
+        self.hf_token = hf_token
+        self.context_manager = SimpleNamespace(items=[])
+        self.pending_approval = None
+        self.turn_count = 0
+        self.config = SimpleNamespace(model_name=model)
+        self.notification_destinations = []
+
+
+class RestoreStore(NoopSessionStore):
+    enabled = True
+
+    def __init__(
+        self,
+        *,
+        metadata: dict[str, Any] | None = None,
+        messages: list[dict[str, Any]] | None = None,
+        delay: float = 0,
+    ) -> None:
+        self.metadata = metadata or {
+            "session_id": "persisted-session",
+            "user_id": "owner",
+            "model": "test-model",
+            "created_at": datetime.now(UTC),
+        }
+        self.messages = messages or []
+        self.delay = delay
+        self.load_calls = 0
+
+    async def load_session(self, session_id: str, **_: Any) -> dict[str, Any] | None:
+        self.load_calls += 1
+        if self.delay:
+            await asyncio.sleep(self.delay)
+        metadata = dict(self.metadata)
+        metadata.setdefault("session_id", session_id)
+        metadata.setdefault("_id", session_id)
+        return {"metadata": metadata, "messages": self.messages}
+
+
+def _manager_with_store(store: NoopSessionStore) -> SessionManager:
+    manager = object.__new__(SessionManager)
+    manager.config = SimpleNamespace(model_name="test-model")
+    manager.sessions = {}
+    manager._lock = asyncio.Lock()
+    manager.persistence_store = store
+    return manager
+
+
+def _runtime_agent_session(
+    session_id: str,
+    *,
+    user_id: str = "owner",
+    hf_token: str | None = "owner-token",
+) -> AgentSession:
+    runtime_session = FakeRuntimeSession(hf_token=hf_token)
+    return AgentSession(
+        session_id=session_id,
+        session=runtime_session,  # type: ignore[arg-type]
+        tool_router=object(),  # type: ignore[arg-type]
+        submission_queue=asyncio.Queue(),
+        user_id=user_id,
+        hf_token=hf_token,
+    )
+
+
+def _install_fake_runtime(manager: SessionManager) -> asyncio.Event:
+    stop = asyncio.Event()
+    manager.run_calls = 0  # type: ignore[attr-defined]
+
+    def fake_create_session_sync(**kwargs: Any):
+        return object(), FakeRuntimeSession(
+            hf_token=kwargs.get("hf_token"),
+            model=kwargs.get("model") or "test-model",
+        )
+
+    async def fake_run_session(*_: Any) -> None:
+        manager.run_calls += 1  # type: ignore[attr-defined]
+        await stop.wait()
+
+    manager._create_session_sync = fake_create_session_sync  # type: ignore[method-assign]
+    manager._run_session = fake_run_session  # type: ignore[method-assign]
+    return stop
+
+
+async def _cancel_runtime_tasks(manager: SessionManager) -> None:
+    tasks = [
+        agent_session.task
+        for agent_session in manager.sessions.values()
+        if agent_session.task and not agent_session.task.done()
+    ]
+    for task in tasks:
+        task.cancel()
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test_existing_session_rejects_cross_user_token_overwrite():
+    manager = _manager_with_store(NoopSessionStore())
+    existing = _runtime_agent_session("s1", user_id="victim", hf_token="victim-token")
+    manager.sessions["s1"] = existing
+
+    result = await manager.ensure_session_loaded(
+        "s1", user_id="attacker", hf_token="attacker-token"
+    )
+
+    assert result is None
+    assert existing.hf_token == "victim-token"
+    assert existing.session.hf_token == "victim-token"
+
+
+@pytest.mark.asyncio
+async def test_existing_session_updates_token_after_access_check():
+    manager = _manager_with_store(NoopSessionStore())
+    existing = _runtime_agent_session("s1", user_id="owner", hf_token="old-token")
+    manager.sessions["s1"] = existing
+
+    result = await manager.ensure_session_loaded(
+        "s1", user_id="owner", hf_token="new-token"
+    )
+
+    assert result is existing
+    assert existing.hf_token == "new-token"
+    assert existing.session.hf_token == "new-token"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_lazy_restore_starts_only_one_agent_task():
+    store = RestoreStore(delay=0.01)
+    manager = _manager_with_store(store)
+    stop = _install_fake_runtime(manager)
+
+    try:
+        first, second = await asyncio.gather(
+            manager.ensure_session_loaded("persisted-session", user_id="owner"),
+            manager.ensure_session_loaded("persisted-session", user_id="owner"),
+        )
+        await asyncio.sleep(0)
+
+        assert first is second
+        assert list(manager.sessions) == ["persisted-session"]
+        assert manager.run_calls == 1  # type: ignore[attr-defined]
+        assert not stop.is_set()
+    finally:
+        stop.set()
+        await _cancel_runtime_tasks(manager)
+
+
+@pytest.mark.asyncio
+async def test_lazy_restore_preserves_pending_approval_tool_calls():
+    store = RestoreStore(
+        metadata={
+            "session_id": "approval-session",
+            "user_id": "owner",
+            "model": "test-model",
+            "pending_approval": [
+                {
+                    "id": "call_123",
+                    "type": "function",
+                    "function": {
+                        "name": "create_file",
+                        "arguments": '{"path":"app.py"}',
+                    },
+                }
+            ],
+        }
+    )
+    manager = _manager_with_store(store)
+    stop = _install_fake_runtime(manager)
+
+    try:
+        restored = await manager.ensure_session_loaded("approval-session", user_id="owner")
+
+        assert restored is not None
+        tool_calls = restored.session.pending_approval["tool_calls"]
+        assert len(tool_calls) == 1
+        assert tool_calls[0].id == "call_123"
+        assert tool_calls[0].function.name == "create_file"
+        assert tool_calls[0].function.arguments == '{"path":"app.py"}'
+    finally:
+        stop.set()
+        await _cancel_runtime_tasks(manager)
+
+
+@pytest.mark.asyncio
+async def test_list_sessions_dev_uses_store_dev_visibility():
+    class ListStore(NoopSessionStore):
+        enabled = True
+
+        def __init__(self) -> None:
+            self.seen_user_id: str | None = None
+
+        async def list_sessions(self, user_id: str, **_: Any) -> list[dict[str, Any]]:
+            self.seen_user_id = user_id
+            if user_id == "dev":
+                return [
+                    {
+                        "session_id": "s1",
+                        "user_id": "alice",
+                        "model": "m",
+                        "created_at": datetime.now(UTC),
+                    },
+                    {
+                        "session_id": "s2",
+                        "user_id": "bob",
+                        "model": "m",
+                        "created_at": datetime.now(UTC),
+                    },
+                ]
+            return []
+
+    store = ListStore()
+    manager = _manager_with_store(store)
+
+    sessions = await manager.list_sessions(user_id="dev")
+
+    assert store.seen_user_id == "dev"
+    assert {session["session_id"] for session in sessions} == {"s1", "s2"}

tests/unit/test_session_persistence.py

@@ -0,0 +1,31 @@
+"""Unit tests for the optional durable session store abstraction."""
+
+import pytest
+
+from agent.core.session_persistence import NoopSessionStore, _safe_message_doc
+
+
+@pytest.mark.asyncio
+async def test_noop_store_keeps_local_cli_and_tests_db_free():
+    store = NoopSessionStore()
+
+    await store.init()
+    await store.upsert_session(session_id="s1", user_id="u1", model="m")
+    await store.save_snapshot(
+        session_id="s1",
+        user_id="u1",
+        model="m",
+        messages=[{"role": "user", "content": "hello"}],
+    )
+
+    assert await store.load_session("s1") is None
+    assert await store.list_sessions("u1") == []
+    assert await store.append_event("s1", "processing", {}) is None
+    assert await store.try_increment_quota("u1", "2099-01-01", 1) is None
+
+
+def test_unsafe_message_payload_is_replaced_with_marker():
+    marker = _safe_message_doc({"role": "assistant", "content": object()})
+
+    assert marker["role"] == "tool"
+    assert marker["ml_intern_persistence_error"] == "message_too_large_or_invalid"

tests/unit/test_user_quotas.py

@@ -15,6 +15,7 @@ if str(_BACKEND_DIR) not in sys.path:
     sys.path.insert(0, str(_BACKEND_DIR))
 
 import user_quotas  # noqa: E402
+from agent.core.session_persistence import NoopSessionStore, _reset_store_for_tests  # noqa: E402
 
 
 @pytest.fixture(autouse=True)
@@ -74,6 +75,33 @@ async def test_concurrent_increments_under_lock_do_not_lose_writes():
     assert await user_quotas.get_claude_used_today("race") == 50
 
 
+@pytest.mark.asyncio
+async def test_try_increment_returns_none_at_cap():
+    assert await user_quotas.try_increment_claude("freebie", 1) == 1
+    assert await user_quotas.try_increment_claude("freebie", 1) is None
+    assert await user_quotas.get_claude_used_today("freebie") == 1
+
+
+@pytest.mark.asyncio
+async def test_try_increment_delegates_cap_to_enabled_store():
+    class StoreAtCap(NoopSessionStore):
+        enabled = True
+
+        async def try_increment_quota(self, user_id: str, day: str, cap: int):
+            assert user_id == "mongo-user"
+            assert cap == 1
+            return None
+
+        async def get_quota(self, user_id: str, day: str):
+            return 1
+
+    _reset_store_for_tests(StoreAtCap())
+
+    assert await user_quotas.try_increment_claude("mongo-user", 1) is None
+    assert await user_quotas.get_claude_used_today("mongo-user") == 1
+    assert "mongo-user" not in user_quotas._claude_counts
+
+
 @pytest.mark.asyncio
 async def test_refund_decrements_and_drops_entry_at_zero():
     await user_quotas.increment_claude("u1")

uv.lock

@@ -1786,6 +1786,7 @@ dependencies = [
     { name = "nbformat" },
     { name = "prompt-toolkit" },
     { name = "pydantic" },
+    { name = "pymongo" },
     { name = "python-dotenv" },
     { name = "requests" },
     { name = "rich" },
@@ -1833,6 +1834,7 @@ requires-dist = [
     { name = "pandas", marker = "extra == 'eval'", specifier = ">=2.3.3" },
     { name = "prompt-toolkit", specifier = ">=3.0.0" },
     { name = "pydantic", specifier = ">=2.12.3" },
+    { name = "pymongo", specifier = ">=4.17.0" },
     { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.2" },
     { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=1.2.0" },
     { name = "python-dotenv", specifier = ">=1.2.1" },
@@ -2769,6 +2771,67 @@ crypto = [
     { name = "cryptography" },
 ]
 
+[[package]]
+name = "pymongo"
+version = "4.17.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "dnspython" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ca/64/50be6fbac9c79fe2e4c17401a467da2d8764d82833d83cec325afe5cab32/pymongo-4.17.0.tar.gz", hash = "sha256:70ffa08ba641468cc068cf46c06b34f01a8ce3489f6411309fcb5ceabe6b2fc0", size = 2523370, upload-time = "2026-04-20T16:39:53.524Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c4/e2/336d86f221cf1b56b2ed9330d4a3b98f9f38f0b37829ae9a9184617d5419/pymongo-4.17.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:4141e6c6a339789b2974efa00ecd9409101672d77a0e3ee2cc3839eedf8ec4df", size = 874668, upload-time = "2026-04-20T16:37:41.39Z" },
+    { url = "https://files.pythonhosted.org/packages/34/8e/75d3c6c935d187ab59c61e9c15d9aab3f274b563eaf1706e8cae5f508dec/pymongo-4.17.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:e68c76b84e0c132d9dbf9307f12ff8185702328187a87b9aca8c941303873433", size = 875294, upload-time = "2026-04-20T16:37:43.432Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/ec/62e855744489dbcd54fd778aae4d80fa4c4819e8fb228ca0cf6f21a03997/pymongo-4.17.0-cp311-cp311-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:ba2195d4f386f839a52a23ea1cfd60ffaaba78a3d7841db51b7e433001139918", size = 1496233, upload-time = "2026-04-20T16:37:45.518Z" },
+    { url = "https://files.pythonhosted.org/packages/82/e8/93e4e5e5ce8fdf8929dabeefe24aafa5ce046028eed0dfa8eeb936e72c49/pymongo-4.17.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8446ff4bfcb6ec2a2e50998c860986a1e992136f998b7f53e7a717fb8aa5a0b9", size = 1522927, upload-time = "2026-04-20T16:37:47.492Z" },
+    { url = "https://files.pythonhosted.org/packages/f7/ca/425dc1d21e0f17bdea0072fc463f662f7fa06d2852af52975c9eced3c07c/pymongo-4.17.0-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:2a0d5ac205728c86e0a02192f1aa5f865b0d7d51f8df6101c01a69a7fc620d72", size = 1583468, upload-time = "2026-04-20T16:37:49.221Z" },
+    { url = "https://files.pythonhosted.org/packages/b3/9d/f08b07eeffda1a43c1759f0fa625e88ae12360996eb56d42aad832fa7dff/pymongo-4.17.0-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:485c8a8eaa4c739f00a331fc73757898ee7c092c214a79e63866ff76aaf282ff", size = 1572787, upload-time = "2026-04-20T16:37:51.061Z" },
+    { url = "https://files.pythonhosted.org/packages/e9/c2/6855a07aafa7b894929af23675b6fb9634800ce43122b76a62f6eeb8da2a/pymongo-4.17.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b2dfcc795f5b9fedbe179a11fdf6051581479d196582a3fe819a92a00e9b9969", size = 1526184, upload-time = "2026-04-20T16:37:53.358Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/05/c952bac7db71c1942ea3559fcd308b49754cc5004b455935fb4000d1f37b/pymongo-4.17.0-cp311-cp311-win32.whl", hash = "sha256:c2292144505fb12156b981bd440f3dc994a883da06ac726c0c8692ccdbc1c510", size = 852621, upload-time = "2026-04-20T16:37:55.28Z" },
+    { url = "https://files.pythonhosted.org/packages/11/c0/c04da9f4c0c6252404598f4e394b862a58a9e866822a70ae261c8a018fdf/pymongo-4.17.0-cp311-cp311-win_amd64.whl", hash = "sha256:2e190827834fce70ecdf9d46796c6dbc0ce08ea87dc2ff5bc6f3f5579b605cb9", size = 867852, upload-time = "2026-04-20T16:37:57.233Z" },
+    { url = "https://files.pythonhosted.org/packages/1d/b2/c7b4870fbeef471e947d3e014676f5910d02e0197074d692ebcf24ec049a/pymongo-4.17.0-cp311-cp311-win_arm64.whl", hash = "sha256:a8f9c40a09bb7d4b9fc8b1da65ecf6efa79bda5cb2756f39d9b6940fac1d19ae", size = 855019, upload-time = "2026-04-20T16:37:58.983Z" },
+    { url = "https://files.pythonhosted.org/packages/98/90/60bcb508840135d5ee46b51b1a950f548338aa8145a8366dbe6639ae51ac/pymongo-4.17.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:d53ffa94b2340dbf6b055e09a0090618c60482c158ecfc9565642fc996bf0944", size = 930529, upload-time = "2026-04-20T16:38:00.936Z" },
+    { url = "https://files.pythonhosted.org/packages/a6/e9/313840f1e52c6dfac47f704428cbfbce59956ebe7633bffc92b03f74f0ad/pymongo-4.17.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:6fe0de9d0f6791abce3471230b32b4817bf89d27b1182b6a550e1ec0fa72aa9a", size = 930665, upload-time = "2026-04-20T16:38:02.915Z" },
+    { url = "https://files.pythonhosted.org/packages/78/35/9d3565ea45b1606f635c1e2cd2563c28d66caafdc50f7ad7d979fcd1b363/pymongo-4.17.0-cp312-cp312-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:e537e95514dae1aaa718f481ec03151a0f0394bcd05f1322896d8fc1330cb729", size = 1762369, upload-time = "2026-04-20T16:38:05.375Z" },
+    { url = "https://files.pythonhosted.org/packages/95/ee/149b0d4b1a11c38bff6f14c23d5814c9b0843fd6dc38ad40596bdb1a62d2/pymongo-4.17.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:37a8385c29881b43eab31f584100fa0eaddedd5607adf010147ba1810118be90", size = 1798044, upload-time = "2026-04-20T16:38:07.195Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/d4/4cee4a7b8d8f6f0550ef6cd2fea42455c5ed619a220cb6ba4fb40d6a5bc8/pymongo-4.17.0-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:f3ee3d241ed77a4fc99ce3cff3b289c3ebce37f61fdd7349d3592c23b82c8784", size = 1878567, upload-time = "2026-04-20T16:38:09.121Z" },
+    { url = "https://files.pythonhosted.org/packages/45/ef/7fe366c84952619ee2f69973566c214775e083dd4df465751912153e4b72/pymongo-4.17.0-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:9eb5d63a3c518cb0804ed678f5e2b875af032d89a7cf57a57360322cf6a4d222", size = 1864881, upload-time = "2026-04-20T16:38:10.896Z" },
+    { url = "https://files.pythonhosted.org/packages/2f/35/b577d82c6d1be7aee7ac7e249bc86f7847998345042e5f8360de238e177b/pymongo-4.17.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8e97e03fa13327c87e3fdc5656acd01e71817f0c1dc3221cd8f30de136bf4ec3", size = 1800349, upload-time = "2026-04-20T16:38:13.589Z" },
+    { url = "https://files.pythonhosted.org/packages/b8/69/dafcf04f66e130ddd91aeb92e7a692480eda46dcd04ec1dbe82c06619e10/pymongo-4.17.0-cp312-cp312-win32.whl", hash = "sha256:6877214bff5f06f6884a9fc8d9016a4a7a5f51f537f5c51ac3a576f93e7dfb32", size = 900518, upload-time = "2026-04-20T16:38:15.541Z" },
+    { url = "https://files.pythonhosted.org/packages/11/35/5c9262a459f988b4eb2605f70815240b77a0d4131136c4326d18f1822b89/pymongo-4.17.0-cp312-cp312-win_amd64.whl", hash = "sha256:9828485f72f63c7d802e0ec41f71906f633c2692621ab3af55ca990186b091b1", size = 920335, upload-time = "2026-04-20T16:38:17.665Z" },
+    { url = "https://files.pythonhosted.org/packages/8d/da/e9c7265ee176faccf4e52c4797837e794d93569a1046f6b19a4acc36e5ad/pymongo-4.17.0-cp312-cp312-win_arm64.whl", hash = "sha256:1195370a77baf003b59b10e91ecc4706297197f0dd9d29c840cc556dc08f7cee", size = 903289, upload-time = "2026-04-20T16:38:19.33Z" },
+    { url = "https://files.pythonhosted.org/packages/2a/6b/c1206879708b94e82fcd8b9653440ec271f79a3674d122192df383047f5a/pymongo-4.17.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:809ec74de3b9148ae43fa8df9faf53470f511c8d384f13b99d6f671f2a379f15", size = 985829, upload-time = "2026-04-20T16:38:21.031Z" },
+    { url = "https://files.pythonhosted.org/packages/cb/cf/bb044ed85160e5c40f568c7c4f4e8ea16f40764ff5d302e5befbe8f6f814/pymongo-4.17.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:a431b737816bf4cddd4fa0fcef04e424ad36b7692734a64150f872fb8f3208be", size = 985899, upload-time = "2026-04-20T16:38:23.409Z" },
+    { url = "https://files.pythonhosted.org/packages/74/0a/f6dfd5ea3901e5d6888da8de8ba728971a1d447debab681cfc56f90d1208/pymongo-4.17.0-cp313-cp313-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:e4fab10f8403169ce92f3cea921609d9ee81107306caae06c08f592d4b8ad2b5", size = 2028569, upload-time = "2026-04-20T16:38:25.343Z" },
+    { url = "https://files.pythonhosted.org/packages/4a/c5/081f59a1c02ae8c0dc73ae58e563838c44eec81aeafa7d0b93a637841c9b/pymongo-4.17.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:20323b0b1c1d33770ad1fc68d429c757734ce9ad3594421c3d6618f10572b1b9", size = 2072916, upload-time = "2026-04-20T16:38:27.291Z" },
+    { url = "https://files.pythonhosted.org/packages/31/42/6e41d434297ffe8b30d9c3717916591a4a7be9075a0dcc2fafdfaaaa62ed/pymongo-4.17.0-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:5a5de048e6da5c18e27cc2437e8c15b3b0cdc8385c15b41178b0caa3322a09c2", size = 2173234, upload-time = "2026-04-20T16:38:29.474Z" },
+    { url = "https://files.pythonhosted.org/packages/3d/cf/1e4a7db352ef9485831c7268dfe8402f0117b32a9ad54b16e810699e3617/pymongo-4.17.0-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:dff3de1294fbbc1db0ba6b511f77b8e540601d092538a31312e99c8a91a78b1e", size = 2156784, upload-time = "2026-04-20T16:38:32.134Z" },
+    { url = "https://files.pythonhosted.org/packages/12/10/6195be29962a61ebb5f4bd9e4c7519890b172f7968a0a0d880398c6ddb02/pymongo-4.17.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:faf03e4c2aafd6de626dbd30ba246d369ae33f47f10629d1bbe40f72115027a6", size = 2074446, upload-time = "2026-04-20T16:38:34.004Z" },
+    { url = "https://files.pythonhosted.org/packages/37/48/33410b8819837ed370c738587306bdf060b59cef11823be212f4a07703c5/pymongo-4.17.0-cp313-cp313-win32.whl", hash = "sha256:c9786665926a09630c5d420c79762cfadbff35a9438bcbc4c81a9fb5ab9228b7", size = 948435, upload-time = "2026-04-20T16:38:35.922Z" },
+    { url = "https://files.pythonhosted.org/packages/6f/77/c0ed522f798a286b99acaa7914ed8d9c80ab091f97f57c59ffed72906e5e/pymongo-4.17.0-cp313-cp313-win_amd64.whl", hash = "sha256:5960519b4d7168f1ecdd3ea10c81b2aedeb9423651aca953cfbc8e76705d3b38", size = 972847, upload-time = "2026-04-20T16:38:37.888Z" },
+    { url = "https://files.pythonhosted.org/packages/97/f0/c39480a2db385fde23861d0c8acda41cdaf1d43e46579db72c5c013a2e81/pymongo-4.17.0-cp313-cp313-win_arm64.whl", hash = "sha256:0ff6bd2f735ab5356541e3e57d5b7dbfbc3f2ee1ccb10b6b0f82d58af69d1d8e", size = 951575, upload-time = "2026-04-20T16:38:40.544Z" },
+    { url = "https://files.pythonhosted.org/packages/da/49/2b0250762a89737ed6f9cea238331baca061b89a8ddd10dd17fee52c3970/pymongo-4.17.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:ff5aa3f1c7e3f08eb0e7a016c91ba468b1850ccfd63d9b1f12f56350f4974cef", size = 1040945, upload-time = "2026-04-20T16:38:42.783Z" },
+    { url = "https://files.pythonhosted.org/packages/89/1c/7a9b5447a08be20e84b6e5b17330917e8d6d9507daa3cd099a9309f11ad7/pymongo-4.17.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:e816db649ba5d7de0568cf3a9f287a9dc9aad21cf0ca667ab156a7ef47fca0b0", size = 1041187, upload-time = "2026-04-20T16:38:45.358Z" },
+    { url = "https://files.pythonhosted.org/packages/78/a1/71704f61632dfc90407a5834fe5f6132854937c4a3648f6c05c351d85a45/pymongo-4.17.0-cp314-cp314-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:12c4fded3a9f1d6a687e36ebd384ac6d00b9b00de1969aa74048e7051ec2a713", size = 2294806, upload-time = "2026-04-20T16:38:47.734Z" },
+    { url = "https://files.pythonhosted.org/packages/ad/b9/aff42be75108b96c2469b1d9329b912c15108f3e7ef32fdc86da8423c330/pymongo-4.17.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2db66aa8dd253a0fc1fad3b0d23d5b3993f7ebde02fbbd7727128debf2853675", size = 2348231, upload-time = "2026-04-20T16:38:50.371Z" },
+    { url = "https://files.pythonhosted.org/packages/f2/30/44c115b8ba1479942c15fd9480eb29a7da0ba68acd56983423ba0deb4a94/pymongo-4.17.0-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:3987e96e7c7be4083d42e8ac2cc6c0d5b78db9973c90fce42ae800b616ca6b20", size = 2467614, upload-time = "2026-04-20T16:38:52.665Z" },
+    { url = "https://files.pythonhosted.org/packages/d2/84/21ee95c8bf0ca7acae7ec7eb365d740bf8fc0156c194baf2c3bdfcb85ec0/pymongo-4.17.0-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:cee36b3c0d0354f880fa7a7fdcdaf2bb5e542c2281e25c1bfadf8cfe21eba7d2", size = 2445970, upload-time = "2026-04-20T16:38:55.175Z" },
+    { url = "https://files.pythonhosted.org/packages/06/89/081d7f1809d5ca09d1e47e49f2111b245f5694de3a7af32cd3a353a6f43f/pymongo-4.17.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:320b34457b20bbcc79997801f95d25ce00472915ca5241167242b42c4359e027", size = 2348605, upload-time = "2026-04-20T16:38:57.557Z" },
+    { url = "https://files.pythonhosted.org/packages/ea/c3/0d949f9d3f2a341c1f635c398c16615e96f89f51ff424ed81e914cf1a4de/pymongo-4.17.0-cp314-cp314-win32.whl", hash = "sha256:df4a644af9ae132d4bfdb2e9516ea51a615fd881caddfbfbd071cf1354844479", size = 1004119, upload-time = "2026-04-20T16:39:00.309Z" },
+    { url = "https://files.pythonhosted.org/packages/f7/55/5c3a3db1048054c695c75c5964cc8bedc2247fdb5a75ef6fab4ec8bb013e/pymongo-4.17.0-cp314-cp314-win_amd64.whl", hash = "sha256:c797f8a80957134f6dd9690367a0f8f5906d672119af2c6aa55f0c527b656bed", size = 1032314, upload-time = "2026-04-20T16:39:02.665Z" },
+    { url = "https://files.pythonhosted.org/packages/e0/19/e235f39906134cb0ffd5574c5a59c355ef5380f0499644ab94994afbb109/pymongo-4.17.0-cp314-cp314-win_arm64.whl", hash = "sha256:68fca71e05ee5da23a8d73cee8379dfb3d26e609a377cae731d742771ed96946", size = 1007627, upload-time = "2026-04-20T16:39:04.678Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/e0/c4c1a86791415b14c684fa0908f9da96de91594a3fd1fa1b8dc689fbb800/pymongo-4.17.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:b4384700cffc3f1dd98e088bc0072dedf6d7d68a230bb4b972665cf69c071c1e", size = 1099151, upload-time = "2026-04-20T16:39:06.969Z" },
+    { url = "https://files.pythonhosted.org/packages/81/4b/69c67f3e23fd9b23b9bedc7ebd23754881cc9d5c5d5b2a9811e96b07f475/pymongo-4.17.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:93641192644fa1ee0f34030e774fd31022a27ad11ba22cb1716142231524f8bd", size = 1099346, upload-time = "2026-04-20T16:39:08.996Z" },
+    { url = "https://files.pythonhosted.org/packages/a2/19/a5208f62f9508a26d73acc69bd3821b8c8adae253679a3c26d2f9652f0d5/pymongo-4.17.0-cp314-cp314t-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:75bc3aa5b94fdb7138d357ec6ca61cd97e0c79f4f7f0bd3efe9639b15cc50942", size = 2619034, upload-time = "2026-04-20T16:39:11.049Z" },
+    { url = "https://files.pythonhosted.org/packages/77/27/426cba1ec5973082a56d4150798529bfdf4151c31391ed1fbbecb23ef2ac/pymongo-4.17.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:50e8f8e23c6df7c6d6929f5e734980b227706e73ee847517c9ba5af90f7fc466", size = 2689939, upload-time = "2026-04-20T16:39:13.617Z" },
+    { url = "https://files.pythonhosted.org/packages/ef/2e/f70993d1255e33f6ee59a4ec4371cc65bff7a7e3fda7d55c3386f25287e8/pymongo-4.17.0-cp314-cp314t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:15d3f3d732aecac1f8d481bde4029755615639bd3076f258a2147210aec8515a", size = 2824994, upload-time = "2026-04-20T16:39:16.057Z" },
+    { url = "https://files.pythonhosted.org/packages/b3/eb/87b0e988ba889e1fcc3430c2cfc166b251872c813e92b43174298bee17ff/pymongo-4.17.0-cp314-cp314t-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:6c5f62862d0f87be481fa1fe8cb811994486773c94a2b61e509285e3f2890763", size = 2801745, upload-time = "2026-04-20T16:39:18.476Z" },
+    { url = "https://files.pythonhosted.org/packages/67/4c/3f83412d086f682d4d468761d66ddc49cf161e786ea74073045eb4491c60/pymongo-4.17.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:64837adbbd72073301af51bb0fc80e3d7707fe5527cea1033ba0320f0b2f881b", size = 2684636, upload-time = "2026-04-20T16:39:20.878Z" },
+    { url = "https://files.pythonhosted.org/packages/9e/d8/b75f6f4ab6c8beb50b0270a4f1e2530b5774f5e116563440e1677ca1820f/pymongo-4.17.0-cp314-cp314t-win32.whl", hash = "sha256:b93b22eedc62598cf5ee9d8c8007a8e9121c50fd88137012d8985500e9dc3151", size = 1056356, upload-time = "2026-04-20T16:39:22.996Z" },
+    { url = "https://files.pythonhosted.org/packages/e4/5e/648c8a238eef18a25ed8a169ea6542d4a860bbec3e95b3d9badac2935c71/pymongo-4.17.0-cp314-cp314t-win_amd64.whl", hash = "sha256:3689ea34f6b647c7d1e7bdc60fcfb214b2789ed1359a7fb96569c69f50e5f18f", size = 1090964, upload-time = "2026-04-20T16:39:24.989Z" },
+    { url = "https://files.pythonhosted.org/packages/dc/cb/d9780b66939c4fc1f024bcc7be23a2abcfe06a9745ca8fa76dc73395482e/pymongo-4.17.0-cp314-cp314t-win_arm64.whl", hash = "sha256:9543d8f84c2e5608565c08ac679774811e6730770d8a645439b073422a4276fb", size = 1058526, upload-time = "2026-04-20T16:39:27.924Z" },
+]
+
 [[package]]
 name = "pyperclip"
 version = "1.11.0"