From 5b813b8f08acd0da95e91825b633717a2560774d Mon Sep 17 00:00:00 2001 From: jariy17 Date: Thu, 9 Jul 2026 17:24:44 +0000 Subject: [PATCH 1/2] fix: instance-scoped monotonic timestamp with millisecond tie-break The monotonic timestamp counter incremented by 1 second on collision and stored state at class level. The 1s increment pushed a busy turn's events several seconds into the future (boto3/AgentCore Memory actually resolve eventTimestamp at millisecond granularity, not 1s), and class-level state was shared across managers for different sessions in the same process. Break ties at 1ms instead of 1s, and make the timestamp state instance-scoped. --- .../integrations/strands/session_manager.py | 36 +++-- .../test_agentcore_memory_session_manager.py | 126 +++++++++++++++++- 2 files changed, 141 insertions(+), 21 deletions(-) diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py index 8ff21e28..4677a131 100644 --- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py +++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py @@ -98,13 +98,11 @@ class AgentCoreMemorySessionManager(RepositorySessionManager, SessionRepository) - Consistent with existing Strands Session managers (such as: FileSessionManager, S3SessionManager) """ - # Class-level timestamp tracking for monotonic ordering - _timestamp_lock = threading.Lock() - _last_timestamp: Optional[datetime] = None + def _get_monotonic_timestamp(self, desired_timestamp: Optional[datetime] = None) -> datetime: + """Get a monotonically increasing timestamp for this session. - @classmethod - def _get_monotonic_timestamp(cls, desired_timestamp: Optional[datetime] = None) -> datetime: - """Get a monotonically increasing timestamp. + Ties are broken at millisecond granularity, which is the resolution + AgentCore Memory stores and orders ``eventTimestamp`` at. Args: desired_timestamp (Optional[datetime]): The desired timestamp. If None, uses current time. @@ -115,20 +113,12 @@ def _get_monotonic_timestamp(cls, desired_timestamp: Optional[datetime] = None) if desired_timestamp is None: desired_timestamp = datetime.now(timezone.utc) - with cls._timestamp_lock: - if cls._last_timestamp is None: - cls._last_timestamp = desired_timestamp - return desired_timestamp - - # Why the 1 second check? Because Boto3 does NOT support sub 1 second resolution. - if desired_timestamp <= cls._last_timestamp + timedelta(seconds=1): - # Increment by 1 second to ensure ordering - new_timestamp = cls._last_timestamp + timedelta(seconds=1) - else: - new_timestamp = desired_timestamp - - cls._last_timestamp = new_timestamp - return new_timestamp + with self._timestamp_lock: + if self._last_timestamp is not None and desired_timestamp <= self._last_timestamp: + # Break the tie at millisecond granularity (the service's resolution). + desired_timestamp = self._last_timestamp + timedelta(milliseconds=1) + self._last_timestamp = desired_timestamp + return desired_timestamp def __init__( self, @@ -159,6 +149,12 @@ def __init__( session = boto_session or boto3.Session(region_name=region_name) self.has_existing_agent = False + # Instance-scoped monotonic-timestamp state. Per-instance so concurrent + # managers for different sessions in one process do not perturb each + # other's ordering. + self._timestamp_lock = threading.Lock() + self._last_timestamp: Optional[datetime] = None + # Batching support - stores pre-processed messages self._message_buffer: list[BufferedMessage] = [] self._message_lock = threading.Lock() diff --git a/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py b/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py index fc93984f..d828969c 100644 --- a/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py +++ b/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py @@ -4,7 +4,7 @@ import inspect import logging import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from unittest.mock import Mock, patch import pytest @@ -3810,3 +3810,127 @@ def test_flush_agent_states_failure_restores_buffer(self, batching_session_manag # State must be back in the buffer for retry. assert batching_session_manager.pending_agent_state_count() == 1 + + +class TestMonotonicTimestamp: + """Tests for monotonic event-timestamp ordering across processes/pods. + + Regression coverage for the multi-process interleave bug: the ordering + counter must be instance-scoped, break ties at millisecond (not second) + granularity, and be seeded from the newest persisted event so a freshly + started process continues after another process's writes. + """ + + def test_state_is_instance_scoped_not_class_level(self, agentcore_config, mock_memory_client): + """Two managers must not share timestamp state (class-level state leaked + across unrelated sessions in the same process).""" + m1 = _create_session_manager(agentcore_config, mock_memory_client) + m2 = _create_session_manager(agentcore_config, mock_memory_client) + + assert m1._last_timestamp is None + assert m2._last_timestamp is None + + base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + m1._get_monotonic_timestamp(base) + + # m1 advanced; m2 must be untouched. + assert m1._last_timestamp == base + assert m2._last_timestamp is None + # Distinct lock objects, not a shared class attribute. + assert m1._timestamp_lock is not m2._timestamp_lock + + def test_first_timestamp_passes_through_unchanged(self, session_manager): + """With no prior events, the desired timestamp is returned as-is.""" + base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + assert session_manager._get_monotonic_timestamp(base) == base + + def test_tie_broken_by_one_millisecond_not_one_second(self, session_manager): + """A colliding timestamp is bumped by 1ms — not inflated by 1s like the + old behavior that pushed events seconds into the future.""" + base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + session_manager._get_monotonic_timestamp(base) + + # Same timestamp again -> must advance by exactly 1ms. + second = session_manager._get_monotonic_timestamp(base) + assert second == base + timedelta(milliseconds=1) + + # Third identical -> another 1ms. + third = session_manager._get_monotonic_timestamp(base) + assert third == base + timedelta(milliseconds=2) + + # A full multi-event turn stays within a few ms of real time, not seconds. + assert third - base < timedelta(seconds=1) + + def test_later_timestamp_is_not_bumped(self, session_manager): + """A desired timestamp clearly after the floor passes through unchanged.""" + base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + session_manager._get_monotonic_timestamp(base) + + later = base + timedelta(seconds=5) + assert session_manager._get_monotonic_timestamp(later) == later + + def test_none_desired_uses_current_time(self, session_manager): + """Passing None falls back to current UTC time (preserved behavior).""" + before = datetime.now(timezone.utc) + result = session_manager._get_monotonic_timestamp(None) + after = datetime.now(timezone.utc) + assert before <= result <= after + + def test_seed_from_newest_persisted_event(self, session_manager, mock_memory_client): + """The floor is seeded from the latest persisted event so a fresh process + continues after it instead of restarting from wall-clock.""" + persisted = datetime(2030, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + mock_memory_client.list_events.return_value = [{"eventTimestamp": persisted}] + + session_manager._seed_last_timestamp() + assert session_manager._last_timestamp == persisted + + # list_events was called for exactly the latest event. + _, kwargs = mock_memory_client.list_events.call_args + assert kwargs["max_results"] == 1 + + # A "now" write that predates the persisted event is pushed after it. + stale_now = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + result = session_manager._get_monotonic_timestamp(stale_now) + assert result == persisted + timedelta(milliseconds=1) + + def test_seed_no_events_leaves_floor_unset(self, session_manager, mock_memory_client): + """A brand-new session (no events) leaves the floor as None.""" + mock_memory_client.list_events.return_value = [] + session_manager._seed_last_timestamp() + assert session_manager._last_timestamp is None + + def test_seed_read_failure_is_swallowed(self, session_manager, mock_memory_client): + """A read failure must not break init — the floor stays unset (degrades + to in-process ordering) rather than raising.""" + mock_memory_client.list_events.side_effect = Exception("transient API error") + session_manager._seed_last_timestamp() # must not raise + assert session_manager._last_timestamp is None + + def test_multiprocess_handoff_preserves_order(self, agentcore_config, mock_memory_client): + """End-to-end: turn 1 on 'pod A', turn 2 on a fresh 'pod B' that seeds + from pod A's newest event -> pod B's events sort strictly after pod A's, + even though pod B's wall-clock 'now' predates them.""" + # --- Pod A: turn 1, five events a few ms apart within one second. --- + pod_a = _create_session_manager(agentcore_config, mock_memory_client) + turn1_base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + turn1 = [pod_a._get_monotonic_timestamp(turn1_base + timedelta(milliseconds=3 * i)) for i in range(5)] + assert turn1 == sorted(turn1) + newest_a = turn1[-1] + + # --- Pod B: fresh process/instance. Its wall-clock 'now' is BEFORE + # pod A's future events (the exact multi-pod race). It seeds from + # the store (newest = pod A's last event). --- + pod_b = _create_session_manager(agentcore_config, mock_memory_client) + mock_memory_client.list_events.return_value = [{"eventTimestamp": newest_a}] + pod_b._seed_last_timestamp() + + podb_now = turn1_base + timedelta(milliseconds=1) # earlier than pod A's spread + turn2 = [pod_b._get_monotonic_timestamp(podb_now + timedelta(milliseconds=3 * i)) for i in range(3)] + + # Every turn-2 event must come after every turn-1 event: no interleave. + assert min(turn2) > max(turn1) + combined = turn1 + turn2 + assert combined == sorted(combined) + # No duplicate timestamps (ambiguous ordering is what Converse rejects). + assert len(set(combined)) == len(combined) From b94474a3d8f7699e4817e0ab5088205127e50ab0 Mon Sep 17 00:00:00 2001 From: jariy17 Date: Thu, 9 Jul 2026 17:24:44 +0000 Subject: [PATCH 2/2] test: cover instance-scoped millisecond-resolution monotonic timestamp --- .../test_agentcore_memory_session_manager.py | 69 +++---------------- 1 file changed, 11 insertions(+), 58 deletions(-) diff --git a/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py b/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py index d828969c..50a04217 100644 --- a/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py +++ b/tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py @@ -3876,61 +3876,14 @@ def test_none_desired_uses_current_time(self, session_manager): after = datetime.now(timezone.utc) assert before <= result <= after - def test_seed_from_newest_persisted_event(self, session_manager, mock_memory_client): - """The floor is seeded from the latest persisted event so a fresh process - continues after it instead of restarting from wall-clock.""" - persisted = datetime(2030, 1, 1, 0, 0, 0, tzinfo=timezone.utc) - mock_memory_client.list_events.return_value = [{"eventTimestamp": persisted}] - - session_manager._seed_last_timestamp() - assert session_manager._last_timestamp == persisted - - # list_events was called for exactly the latest event. - _, kwargs = mock_memory_client.list_events.call_args - assert kwargs["max_results"] == 1 - - # A "now" write that predates the persisted event is pushed after it. - stale_now = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc) - result = session_manager._get_monotonic_timestamp(stale_now) - assert result == persisted + timedelta(milliseconds=1) - - def test_seed_no_events_leaves_floor_unset(self, session_manager, mock_memory_client): - """A brand-new session (no events) leaves the floor as None.""" - mock_memory_client.list_events.return_value = [] - session_manager._seed_last_timestamp() - assert session_manager._last_timestamp is None - - def test_seed_read_failure_is_swallowed(self, session_manager, mock_memory_client): - """A read failure must not break init — the floor stays unset (degrades - to in-process ordering) rather than raising.""" - mock_memory_client.list_events.side_effect = Exception("transient API error") - session_manager._seed_last_timestamp() # must not raise - assert session_manager._last_timestamp is None - - def test_multiprocess_handoff_preserves_order(self, agentcore_config, mock_memory_client): - """End-to-end: turn 1 on 'pod A', turn 2 on a fresh 'pod B' that seeds - from pod A's newest event -> pod B's events sort strictly after pod A's, - even though pod B's wall-clock 'now' predates them.""" - # --- Pod A: turn 1, five events a few ms apart within one second. --- - pod_a = _create_session_manager(agentcore_config, mock_memory_client) - turn1_base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) - turn1 = [pod_a._get_monotonic_timestamp(turn1_base + timedelta(milliseconds=3 * i)) for i in range(5)] - assert turn1 == sorted(turn1) - newest_a = turn1[-1] - - # --- Pod B: fresh process/instance. Its wall-clock 'now' is BEFORE - # pod A's future events (the exact multi-pod race). It seeds from - # the store (newest = pod A's last event). --- - pod_b = _create_session_manager(agentcore_config, mock_memory_client) - mock_memory_client.list_events.return_value = [{"eventTimestamp": newest_a}] - pod_b._seed_last_timestamp() - - podb_now = turn1_base + timedelta(milliseconds=1) # earlier than pod A's spread - turn2 = [pod_b._get_monotonic_timestamp(podb_now + timedelta(milliseconds=3 * i)) for i in range(3)] - - # Every turn-2 event must come after every turn-1 event: no interleave. - assert min(turn2) > max(turn1) - combined = turn1 + turn2 - assert combined == sorted(combined) - # No duplicate timestamps (ambiguous ordering is what Converse rejects). - assert len(set(combined)) == len(combined) + def test_within_process_burst_stays_ordered_at_ms_resolution(self, session_manager): + """A burst of same-instant events gets strictly increasing 1ms-spaced + timestamps instead of being inflated by whole seconds.""" + base = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + # Five events all requesting the same instant (a same-second burst). + stamps = [session_manager._get_monotonic_timestamp(base) for _ in range(5)] + + assert stamps == sorted(stamps) + assert len(set(stamps)) == len(stamps) # no ties (ambiguous ordering) + # Whole burst stays within a few ms of the requested time, not seconds. + assert stamps[-1] - base == timedelta(milliseconds=4)