feat: add snapshot session manager to python

strands-py/src/strands/session/__init__.py

@@ -1,18 +1,28 @@
 """Session module.
 
 This module provides session management functionality.
+
+``SnapshotSessionManager`` is the recommended path for new agents: it persists the whole
+agent as a versioned snapshot to a unified :class:`~strands.storage.storage.Storage` backend
+and supports checkpointing (time-travel restore). The repository-based managers
+(``FileSessionManager``, ``S3SessionManager``, ``RepositorySessionManager``) persist each
+message individually and are deprecated.
 """
 
 from .file_session_manager import FileSessionManager
 from .repository_session_manager import RepositorySessionManager
 from .s3_session_manager import S3SessionManager
 from .session_manager import SessionManager
 from .session_repository import SessionRepository
+from .snapshot_session_manager import SaveLatestStrategy, SnapshotSessionManager, SnapshotTrigger
 
 __all__ = [
     "FileSessionManager",
     "RepositorySessionManager",
     "S3SessionManager",
+    "SaveLatestStrategy",
     "SessionManager",
     "SessionRepository",
+    "SnapshotSessionManager",
+    "SnapshotTrigger",
 ]

strands-py/src/strands/session/_snapshot_id.py

@@ -0,0 +1,73 @@
+"""Monotonic UUIDv7 generation and validation for immutable snapshot ids.
+
+Immutable snapshot ids are UUID version 7 (RFC 9562), matching the TypeScript SDK. The 48-bit
+millisecond timestamp in the high bits makes lexicographic order equal creation order, so
+listing returns oldest-first without a separate index. Ids are opaque handles that callers get
+from ``list_snapshot_ids`` and never construct.
+"""
+
+import os
+import re
+import threading
+import time
+import uuid
+
+# ``\A...\Z`` (not ``^...$``) so a trailing newline cannot slip past validation into a storage key.
+_SNAPSHOT_ID_PATTERN = re.compile(r"\A[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\Z")
+
+# UUIDv7 monotonic counter state (RFC 9562 method 2): within a single millisecond, ids that
+# would otherwise collide are ordered by a counter seeded into the random bits, so sorting by
+# the id preserves creation order even for sub-millisecond bursts. Guarded by a lock because
+# snapshots can be written from the sync bridge and async paths on different threads.
+_lock = threading.Lock()
+_last_ms = -1
+_counter = 0
+
+
+def new_snapshot_id() -> str:
+    """Return a fresh, monotonic UUIDv7 immutable snapshot id.
+
+    Monotonic across calls so lexicographic order equals creation order, even for bursts that
+    mint more than 4096 ids inside one millisecond.
+    """
+    global _last_ms, _counter
+
+    with _lock:
+        now_ms = int(time.time() * 1000)
+        if now_ms > _last_ms:
+            _last_ms = now_ms
+            # Seed in the low half of the 12-bit field (fresh randomness for uniqueness) so at
+            # least 2048 increments fit before overflow.
+            _counter = int.from_bytes(os.urandom(2), "big") & 0x07FF
+        else:
+            _counter += 1
+            if _counter > 0x0FFF:
+                # Counter exhausted within this millisecond: borrow from the clock by advancing
+                # the timestamp 1ms (RFC 9562 method 1), keeping ids strictly increasing rather
+                # than wrapping the counter field backwards.
+                _last_ms += 1
+                _counter = int.from_bytes(os.urandom(2), "big") & 0x07FF
+        counter = _counter
+        timestamp_ms = _last_ms
+
+    # Layout: 48-bit ms timestamp | version(7) | 12-bit counter | variant(0b10) | 62 random bits.
+    tail = int.from_bytes(os.urandom(8), "big") & ((1 << 62) - 1)
+    value = (timestamp_ms & ((1 << 48) - 1)) << 80
+    value |= 0x7 << 76
+    value |= (counter & 0x0FFF) << 64
+    value |= 0b10 << 62
+    value |= tail
+    return str(uuid.UUID(int=value))
+
+
+def validate_snapshot_id(snapshot_id: str) -> None:
+    """Validate that a string is an SDK-vended snapshot id (a UUIDv7).
+
+    Args:
+        snapshot_id: The string to validate.
+
+    Raises:
+        ValueError: If the string is not a valid snapshot id.
+    """
+    if not _SNAPSHOT_ID_PATTERN.match(snapshot_id):
+        raise ValueError(f"'{snapshot_id}' is not a valid snapshot id")

strands-py/src/strands/session/file_session_manager.py

@@ -8,6 +8,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, cast
 
+from typing_extensions import deprecated
+
 from .. import _identifier
 from ..types.exceptions import SessionException
 from ..types.session import Session, SessionAgent, SessionMessage
@@ -25,6 +27,12 @@
 MULTI_AGENT_PREFIX = "multi_agent_"
 
 
+@deprecated(
+    "FileSessionManager is deprecated and will be removed in Strands SDK 2.0. Use "
+    "SnapshotSessionManager with a LocalFileStorage instead (single- and multi-agent). "
+    "See migration guide: "
+    "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/session-management/"
+)
 class FileSessionManager(RepositorySessionManager, SessionRepository):
     """File-based session manager for local filesystem storage.
 

strands-py/src/strands/session/repository_session_manager.py

@@ -4,6 +4,8 @@
 import logging
 from typing import TYPE_CHECKING, Any
 
+from typing_extensions import deprecated
+
 from ..agent.state import AgentState
 from ..tools._tool_helpers import generate_missing_tool_result_content
 from ..types.content import ContentBlock, Message, _generate_tracking_id
@@ -25,6 +27,12 @@
 logger = logging.getLogger(__name__)
 
 
+@deprecated(
+    "RepositorySessionManager is deprecated and will be removed in Strands SDK 2.0. Use "
+    "SnapshotSessionManager with a Storage backend instead (single- and multi-agent). "
+    "See migration guide: "
+    "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/session-management/"
+)
 class RepositorySessionManager(SessionManager):
     """Session manager for persisting agents in a SessionRepository."""
 
@@ -279,9 +287,7 @@ def _fix_broken_tool_use(self, messages: list[Message]) -> list[Message]:
 
         for index in reversed(tool_use_indices):
             message = messages[index]
-            tool_use_ids = [
-                content["toolUse"]["toolUseId"] for content in message["content"] if "toolUse" in content
-            ]
+            tool_use_ids = [content["toolUse"]["toolUseId"] for content in message["content"] if "toolUse" in content]
 
             next_message = messages[index + 1]
             next_content = next_message["content"]

strands-py/src/strands/session/s3_session_manager.py

@@ -8,6 +8,7 @@
 import boto3
 from botocore.config import Config as BotocoreConfig
 from botocore.exceptions import ClientError
+from typing_extensions import deprecated
 
 from .. import _identifier
 from ..types.exceptions import SessionException
@@ -28,6 +29,12 @@
 MULTI_AGENT_PREFIX = "multi_agent_"
 
 
+@deprecated(
+    "S3SessionManager is deprecated and will be removed in Strands SDK 2.0. Use "
+    "SnapshotSessionManager with an S3Storage instead (single- and multi-agent). "
+    "See migration guide: "
+    "https://strandsagents.com/latest/documentation/docs/user-guide/concepts/agents/session-management/"
+)
 class S3SessionManager(RepositorySessionManager, SessionRepository):
     """S3-based session manager for cloud storage.