feat(memory): event metadata state identification, message batching, and redundant sync elimination (aws#244)

* feat(memory): use event metadata for session/agent state identification

Replace actorId prefix-based approach with event metadata for distinguishing
session and agent state events. Add auto-migration for legacy events.

Changes:
- Add StateType enum (SESSION, AGENT) and metadata keys
- Update create_session/create_agent to include stateType metadata
- Update read_session/read_agent to filter by metadata
- Add backwards-compatible auto-migration: legacy events are converted
  to new format on read (create new with metadata, delete old)
- Add tests for legacy migration behavior

* fix(memory): handle eventual consistency and pagination with metadata filters

- Fix pagination bug in list_events where API returns nextToken even with
  0 results, causing "metadata filter mismatch" error on subsequent page
- Add _retry_with_backoff method for handling eventual consistency when
  reading newly created agents via metadata filter
- Track created agent IDs to handle updates during consistency window
- Update test to account for retry behavior in legacy migration

* fix: Fixed pagination bug with nextToken

* feat(memory): implement true message batching for flush_messages() (aws#256)

* feat(memory): add batch_size configuration for message buffering

Add optional batch_size parameter to AgentCoreMemoryConfig that allows
customers to buffer messages before sending to AgentCore Memory.

Changes:
- Add batch_size parameter (default=1, max=100) to AgentCoreMemoryConfig
- Add message buffering with thread-safe _message_buffer and _buffer_lock
- Modify create_message() to buffer when batch_size > 1
- Add flush_messages() to send all buffered messages
- Add pending_message_count() to check buffer size
- Add close() and context manager for cleanup
- Add 24 unit tests covering batching functionality

Default batch_size=1 preserves backward compatibility (immediate send).

* feat(memory): implement true message batching in flush_messages()

Previously, batch_size buffered messages but still made N separate API
calls for N messages. Now flush_messages() groups conversational messages
by session_id and combines them into a single create_event() call per
session, significantly reducing API calls.

Key changes:
- Group conversational messages by session_id into combined payloads
- Preserve message order within each session's payload (earliest first)
- Use the latest timestamp from grouped messages for the combined event
- Send blob messages (>9KB) individually (different API path)
- Clear buffer only after ALL API calls succeed to prevent data loss
- Improve error messages to include session context

Tests added for critical scenarios:
- Multiple sessions grouped into separate API calls
- Latest timestamp used for combined events
- Partial failure with multiple sessions preserves entire buffer
- Multiple blob messages sent individually (not batched)
- Mixed sessions with blobs and conversational messages

Also fixes pre-existing test issues:
- Fix test_read_agent_legacy_migration mock setup to match actual impl
- Fix test_load_long_term_memories_with_validation_failure for strands API

* refactor: rename flush_messages to _flush_messages (private method)

* refactor(tests): deduplicate batching test fixtures

Extract _create_session_manager helper to eliminate triple-nested
context managers and move batching_config/batching_session_manager
to module-level fixtures shared across all test classes.

* fix(memory): prevent flush exception from masking original exception in __exit__

Wrap _flush_messages() in try/except so that if the with-block raises
and flush also fails, the flush error is logged instead of replacing
the original exception.

* fix(memory): fix events_to_messages ordering for batched events

Iterate reversed(events) instead of reversing the flat message list
after iteration, which scrambled intra-event payload order for batched
events. Add 6 unit tests covering multi-event ordering, batched payload
ordering, mixed blob/conversational ordering, and malformed payloads.

* test(memory): rewrite integration tests to use real Agent objects

Replace 5 tests that called session manager methods directly (duplicating
unit test coverage) with 2 Agent-based integration tests that validate
the full lifecycle: context manager flush + session resume, and multi-turn
conversation with batching. Remove unnecessary time.sleep from blocking
close() calls.

* fix(tests): resolve ruff E501 line-too-long and F401 unused import

Break long test method signatures and constructor args to stay within
120-char limit. Remove unused json import.

* fix: remove unused imports in session_manager.py

Remove unused time, Callable, and TypeVar imports flagged by ruff.

Author: jariy17

src/bedrock_agentcore/memory/client.py

@@ -788,6 +788,7 @@ def list_events(
                 all_events.extend(events)
 
                 next_token = response.get("nextToken")
+                # Break if: no more pages or reached max
                 if not next_token or len(all_events) >= max_results:
                     break
 

src/bedrock_agentcore/memory/integrations/strands/bedrock_converter.py

@@ -72,7 +72,7 @@ def events_to_messages(events: list[dict[str, Any]]) -> list[SessionMessage]:
             list[SessionMessage]: list of SessionMessage objects.
         """
         messages = []
-        for event in events:
+        for event in reversed(events):
             for payload_item in event.get("payload", []):
                 if "conversational" in payload_item:
                     conv = payload_item["conversational"]
@@ -93,7 +93,7 @@ def events_to_messages(events: list[dict[str, Any]]) -> list[SessionMessage]:
                                 logger.error("This is not a SessionMessage but just a blob message. Ignoring")
                     except (json.JSONDecodeError, ValueError):
                         logger.error("Failed to parse blob content: %s", payload_item)
-        return list(reversed(messages))
+        return messages
 
     @staticmethod
     def total_length(message: tuple[str, str]) -> int:

src/bedrock_agentcore/memory/integrations/strands/config.py

@@ -29,9 +29,12 @@ class AgentCoreMemoryConfig(BaseModel):
         session_id: Required unique ID for the session
         actor_id: Required unique ID for the agent instance/user
         retrieval_config: Optional dictionary mapping namespaces to retrieval configurations
+        batch_size: Number of messages to batch before sending to AgentCore Memory.
+            Default of 1 means immediate sending (no batching). Max 100.
     """
 
     memory_id: str = Field(min_length=1)
     session_id: str = Field(min_length=1)
     actor_id: str = Field(min_length=1)
     retrieval_config: Optional[Dict[str, RetrievalConfig]] = None
+    batch_size: int = Field(default=1, ge=1, le=100)