From faef3a09dcb674b288e53c75dee9522c5ead7ee5 Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Tue, 27 Oct 2026 20:42:07 -0400
Subject: [PATCH 1/7] fix(memory): store completed BidiAgent transcripts

BidiAgent adds each spoken turn to history as an empty placeholder
(MessageAddedEvent, bidi status "pending") and later replaces it with
the finished text through MessageUpdatedEvent (status "complete").
The session manager only persisted on MessageAddedEvent, so the
placeholder was dropped as empty and the finished text was never
written. Voice agents therefore stored no user or assistant turns and
long term memory strategies had nothing to extract.

Persist MessageUpdatedEvent messages whose bidi status is "complete"
through append_message, the same path text messages use, in both sync
and async mode. Pending placeholders are skipped explicitly so a custom
converter cannot store them, and "incomplete" updates are ignored.
Each transcript completes once, so it is written exactly once. The hook
is only registered when the installed strands-agents provides
MessageUpdatedEvent.

Closes #716
---
 .../integrations/strands/session_manager.py   | 22 ++++-
 .../test_agentcore_memory_session_manager.py  | 83 +++++++++++++++++++
 2 files changed, 104 insertions(+), 1 deletion(-)

diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
index be5632dc..ee1127a7 100644
--- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
+++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
@@ -7,9 +7,10 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Dict, NamedTuple, Optional
+from typing import TYPE_CHECKING, Any, Dict, Mapping, NamedTuple, Optional
 
 import boto3
+import strands.hooks
 from botocore.config import Config as BotocoreConfig
 from strands.experimental.bidi import BidiAgent
 from strands.experimental.bidi.hooks import BidiAgentStopEvent
@@ -46,6 +47,8 @@
 
 logger = logging.getLogger(__name__)
 
+MessageUpdatedEvent = getattr(strands.hooks, "MessageUpdatedEvent", None)
+
 MAX_FETCH_ALL_RESULTS = 10000
 
 # Legacy prefixes for backwards compatibility with old events
@@ -63,6 +66,13 @@
 RESERVED_METADATA_KEYS = frozenset({STATE_TYPE_KEY, AGENT_ID_KEY})
 
 
+def _bidi_status(message: Mapping[str, Any]) -> Optional[str]:
+    metadata = message.get("metadata") or {}
+    bidi = (metadata.get("custom") or {}).get("bidi") or {}
+    status: Optional[str] = bidi.get("status")
+    return status
+
+
 class BufferedMessage(NamedTuple):
     """A pre-processed message waiting to be flushed to AgentCore Memory."""
 
@@ -841,12 +851,18 @@ def append_message(self, message: Message, agent: "LocalAgent", **kwargs: Any) -
             agent: Agent to append the message to
             **kwargs: Additional keyword arguments for future extensibility.
         """
+        if _bidi_status(message) == "pending":
+            return
         created_message = self.create_message(self.session_id, agent.agent_id, SessionMessage.from_message(message, 0))
         if created_message is None:
             return
         session_message = SessionMessage.from_message(message, created_message.get("eventId"))
         self._latest_agent_message[agent.agent_id] = session_message
 
+    def _append_completed_message(self, event: Any) -> None:
+        if _bidi_status(event.message) == "complete":
+            self.append_message(event.message, event.agent)
+
     def retrieve_customer_context(self, event: MessageAddedEvent) -> None:
         """Retrieve customer LTM context for regular Agent invocations.
 
@@ -955,6 +971,8 @@ def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
 
             RepositorySessionManager.register_hooks(self, registry, **kwargs)
             registry.add_callback(MessageAddedEvent, lambda event: self.retrieve_customer_context(event))
+            if MessageUpdatedEvent is not None:
+                registry.add_callback(MessageUpdatedEvent, self._append_completed_message)
             return
 
         # Async mode: register async callbacks that offload the existing sync
@@ -990,6 +1008,8 @@ async def _on_message_added_persist(event: MessageAddedEvent) -> None:
             await asyncio.to_thread(self.sync_agent, event.agent)
 
         registry.add_callback(MessageAddedEvent, _on_message_added_persist)
+        if MessageUpdatedEvent is not None:
+            registry.add_callback(MessageUpdatedEvent, _offload(self._append_completed_message, lambda e: e))
         registry.add_callback(AfterInvocationEvent, _offload(self.sync_agent, lambda e: e.agent))
         registry.add_callback(MessageAddedEvent, _offload(self.retrieve_customer_context, lambda e: e))
 
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 11f505a0..1d4348e1 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
@@ -11,6 +11,7 @@
 import pytest
 from botocore.config import Config as BotocoreConfig
 from botocore.exceptions import ClientError
+from strands import hooks as strands_hooks
 from strands.agent.agent import Agent
 from strands.experimental.bidi import BidiAgent
 from strands.experimental.bidi.hooks import BidiAgentStopEvent
@@ -34,6 +35,8 @@
     BufferedMessage,
 )
 
+MessageUpdatedEvent = getattr(strands_hooks, "MessageUpdatedEvent", None)
+
 
 @pytest.fixture
 def agentcore_config():
@@ -2846,6 +2849,86 @@ async def test_completion_flushes_messages_and_final_state(
         assert [json.loads(payload["blob"])["state"] for payload in state_payloads] == [{}, {"status": "stopped"}]
 
 
+def _bidi_message(role, text, status):
+    content = [] if text is None else [{"text": text}]
+    return {
+        "role": role,
+        "content": content,
+        "tracking_id": f"{role}-1",
+        "metadata": {"custom": {"bidi": {"kind": "transcript", "status": status}}},
+    }
+
+
+class TestBidiTranscripts:
+    """BidiAgent transcripts are stored once they are complete."""
+
+    @pytest.fixture(params=[False, True], ids=["sync", "async"])
+    def bidi_manager(self, request, agentcore_config, mock_memory_client):
+        agentcore_config.async_mode = request.param
+        manager = _create_session_manager(agentcore_config, mock_memory_client)
+        manager._latest_agent_message = {}
+        manager.sync_agent = Mock()
+        registry = HookRegistry()
+        manager.register_hooks(registry)
+        return registry, Mock(spec=BidiAgent, agent_id="test-agent")
+
+    async def test_pending_placeholder_is_not_stored(self, bidi_manager, mock_memory_client):
+        registry, agent = bidi_manager
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+
+        mock_memory_client.create_event.assert_not_called()
+
+    @pytest.mark.parametrize("metadata", [None, {}, {"custom": None}, {"custom": {"bidi": {"kind": "tool_result"}}}])
+    async def test_message_without_bidi_status_is_stored(self, bidi_manager, mock_memory_client, metadata):
+        registry, agent = bidi_manager
+        message = {"role": "user", "content": [{"text": "Hello"}], "metadata": metadata}
+
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=message))
+
+        mock_memory_client.create_event.assert_called_once()
+
+    @pytest.mark.skipif(MessageUpdatedEvent is None, reason="strands-agents has no MessageUpdatedEvent")
+    @pytest.mark.parametrize(("role", "memory_role"), [("user", "USER"), ("assistant", "ASSISTANT")])
+    async def test_complete_transcript_is_stored_once(self, bidi_manager, mock_memory_client, role, memory_role):
+        registry, agent = bidi_manager
+        final = _bidi_message(role, "I need a hotel in London.", "complete")
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message(role, None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, final["tracking_id"], final))
+
+        mock_memory_client.create_event.assert_called_once()
+        [(payload, stored_role)] = mock_memory_client.create_event.call_args.kwargs["messages"]
+        assert stored_role.upper() == memory_role
+        assert json.loads(payload)["message"]["content"] == [{"text": "I need a hotel in London."}]
+
+    @pytest.mark.skipif(MessageUpdatedEvent is None, reason="strands-agents has no MessageUpdatedEvent")
+    @pytest.mark.parametrize(
+        ("text", "status"),
+        [("[Transcript unavailable.]", "incomplete"), ("", "complete"), ("   ", "complete"), ("partial", "pending")],
+    )
+    async def test_unfinished_or_empty_transcript_is_not_stored(self, bidi_manager, mock_memory_client, text, status):
+        registry, agent = bidi_manager
+        update = _bidi_message("assistant", text, status)
+
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, update["tracking_id"], update))
+
+        mock_memory_client.create_event.assert_not_called()
+
+    @pytest.mark.skipif(MessageUpdatedEvent is None, reason="strands-agents has no MessageUpdatedEvent")
+    async def test_update_without_bidi_status_is_not_stored(self, bidi_manager, mock_memory_client):
+        registry, agent = bidi_manager
+        message = {"role": "assistant", "content": [{"text": "redacted"}], "tracking_id": "x"}
+
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, "x", message))
+
+        mock_memory_client.create_event.assert_not_called()
+
+
 class TestIntervalFlush:
     """Test interval-based flush mechanism for long-running agents."""
 

From 3005c435c270b224d858299f827b10921900fc24 Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Thu, 29 Oct 2026 01:15:05 -0400
Subject: [PATCH 2/7] build(deps): require strands-agents 1.57.1 for
 MessageUpdatedEvent

MessageUpdatedEvent, which BidiAgent uses to report a finished
transcript, first shipped in strands-agents 1.57.1. The lock pinned
1.56.0, so CI skipped every test that exercises the transcript fix.
Raise the minimum to 1.57.1 and relock to exactly that version.

strands-agents 1.57.1 also adds RepositorySessionManager._held_records
and walks agent.messages in sync_agent. The test helper patches the
parent __init__ away and some tests use bare Mock agents, so set
_held_records in the helper and give those agents an empty message
list. Production code calls the parent __init__ and is unaffected.
---
 pyproject.toml                                         |  4 ++--
 .../strands/test_agentcore_memory_session_manager.py   |  5 +++--
 uv.lock                                                | 10 +++++-----
 3 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index e7a85960..12b7c627 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -149,7 +149,7 @@ dev = [
     "ruff>=0.12.0",
     "websockets>=14.1",
     "wheel>=0.45.1",
-    "strands-agents>=1.56.0",
+    "strands-agents>=1.57.1",
     "strands-agents-evals>=1.0.3,<2.0.0",
     "deepeval>=3.5.0,<5.0.0",
     "autoevals>=0.3.0,<1.0.0",
@@ -169,7 +169,7 @@ a2a = ["a2a-sdk[http-server]>=0.3,<0.4"]
 a2a-v1 = ["a2a-sdk[http-server]>=1.0.1,<2.0"]
 ag-ui = ["ag-ui-protocol>=0.1.10"]
 strands-agents = [
-    "strands-agents>=1.56.0",
+    "strands-agents>=1.57.1",
     "mcp>=1.23.0,<2.0.0",
 ]
 langgraph = [
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 1d4348e1..7c0967ac 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
@@ -89,6 +89,7 @@ def _create_session_manager(config, mock_memory_client):
         manager = AgentCoreMemorySessionManager(config)
         manager.session_id = config.session_id
         manager.session = Session(session_id=config.session_id, session_type=SessionType.AGENT)
+        manager._held_records = {}
         return manager
 
 
@@ -2788,7 +2789,7 @@ def test_after_invocation_hook_flushes_buffer(self, batching_session_manager, mo
         batching_session_manager.register_hooks(registry)
 
         # Create mock event with mock agent
-        mock_agent = Mock()
+        mock_agent = Mock(messages=[])
         mock_event = AfterInvocationEvent(agent=mock_agent)
         registry.invoke_callbacks(mock_event)
 
@@ -2828,7 +2829,7 @@ async def test_completion_flushes_messages_and_final_state(
         batching_config.async_mode = async_mode
         manager = _create_session_manager(batching_config, mock_memory_client)
         manager.session_repository = manager
-        agent = Mock(agent_id="test-agent")
+        agent = Mock(agent_id="test-agent", messages=[])
         agent.state.get.return_value = {}
         manager.create_agent(manager.session_id, SessionAgent.from_agent(agent))
         manager.create_message(
diff --git a/uv.lock b/uv.lock
index a7b5cc45..78b525e3 100644
--- a/uv.lock
+++ b/uv.lock
@@ -606,7 +606,7 @@ requires-dist = [
     { name = "ragas", marker = "extra == 'ragas'", specifier = ">=0.4.3,<1.0.0" },
     { name = "requests", marker = "extra == 'datasets'", specifier = ">=2.31.0" },
     { name = "starlette", specifier = ">=0.46.2" },
-    { name = "strands-agents", marker = "extra == 'strands-agents'", specifier = ">=1.56.0" },
+    { name = "strands-agents", marker = "extra == 'strands-agents'", specifier = ">=1.57.1" },
     { name = "strands-agents-evals", marker = "extra == 'autoevals'", specifier = ">=1.0.3,<2.0.0" },
     { name = "strands-agents-evals", marker = "extra == 'deepeval'", specifier = ">=1.0.3,<2.0.0" },
     { name = "strands-agents-evals", marker = "extra == 'ragas'", specifier = ">=1.0.3,<2.0.0" },
@@ -641,7 +641,7 @@ dev = [
     { name = "pytest-rerunfailures", specifier = ">=15.0" },
     { name = "ragas", specifier = ">=0.4.3,<1.0.0" },
     { name = "ruff", specifier = ">=0.12.0" },
-    { name = "strands-agents", specifier = ">=1.56.0" },
+    { name = "strands-agents", specifier = ">=1.57.1" },
     { name = "strands-agents-evals", specifier = ">=1.0.3,<2.0.0" },
     { name = "websockets", specifier = ">=14.1" },
     { name = "wheel", specifier = ">=0.45.1" },
@@ -5043,7 +5043,7 @@ wheels = [
 
 [[package]]
 name = "strands-agents"
-version = "1.56.0"
+version = "1.57.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "boto3" },
@@ -5060,9 +5060,9 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "watchdog" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/4c/2f/c980715e56a2017280ba597aea83e77ec37a94eeaaabc47114e354a6846e/strands_agents-1.56.0.tar.gz", hash = "sha256:d0a4587065f9b9f4890455ce947474067e5deaadf140d1b2ef5153a67ebf80a3", size = 1662954, upload-time = "2026-09-15T19:54:39.56Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/6d/2c/061ba1354b8586e8bd6edb42b2f8de3c6a4db83488b6b28d82cbc88a9185/strands_agents-1.57.1.tar.gz", hash = "sha256:109ea0a0f7da82a3b32e4c02cef12f53d46637d3e1e0fb8a285c5fa7116eb1d8", size = 1709312, upload-time = "2026-09-25T20:12:35.457Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/89/07/0d8955760ed673c3112af65d2a026e79bbcb3c869b3eadc896820b64edd7/strands_agents-1.56.0-py3-none-any.whl", hash = "sha256:8026a2fde7ca3d2f2760ea57003dd7163761bfba74972762468b745b4f7bfd08", size = 838144, upload-time = "2026-09-15T19:54:37.448Z" },
+    { url = "https://files.pythonhosted.org/packages/8c/1b/2b7e12fafee0e6ab6ce9fa16a0ad850bb570d8e2bd6326f5cca7c2cd75a7/strands_agents-1.57.1-py3-none-any.whl", hash = "sha256:797ea3de2ba4a66aaf7e804cba8308e3d582e2412c59978433d3216973d02a9d", size = 858212, upload-time = "2026-09-25T20:12:33.361Z" },
 ]
 
 [[package]]

From cf76dc6e81f2ceae0c7022613545adcdcbc79b03 Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Thu, 29 Oct 2026 01:28:57 -0400
Subject: [PATCH 3/7] fix(memory): keep BidiAgent history order when
 transcripts finish late

BidiAgent reserves a message's place in history when a transcript
starts and fills it in when the transcript finishes. With Gemini and
OpenAI realtime the caller's transcript can still be open while the
assistant reply starts and finishes, so writing each transcript when it
finished stored the reply before the question, and a restored session
replayed them in that order.

Hold every message that comes after an unfinished BidiAgent message and
write them in history order once it finishes. An incomplete transcript
releases the messages behind it without being stored, and
BidiAgentStopEvent and close() write anything still held so nothing is
lost when a session ends with a transcript open. A failed write is
dropped and raised as before, so it cannot block later messages.
Regular Agent messages never wait because nothing is pending.

Import MessageUpdatedEvent directly now that strands-agents 1.57.1 is
the minimum, give the update callback its real event type, and add
tests that fail without the fix, including one with a converter that
keeps empty content to show why pending placeholders are never written.
---
 .../integrations/strands/session_manager.py   |  73 +++++++--
 .../test_agentcore_memory_session_manager.py  | 145 +++++++++++++++---
 2 files changed, 185 insertions(+), 33 deletions(-)

diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
index ee1127a7..26b5aa74 100644
--- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
+++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
@@ -10,7 +10,6 @@
 from typing import TYPE_CHECKING, Any, Dict, Mapping, NamedTuple, Optional
 
 import boto3
-import strands.hooks
 from botocore.config import Config as BotocoreConfig
 from strands.experimental.bidi import BidiAgent
 from strands.experimental.bidi.hooks import BidiAgentStopEvent
@@ -19,7 +18,7 @@
     AfterNodeCallEvent,
     MultiAgentInitializedEvent,
 )
-from strands.hooks import AfterInvocationEvent, MessageAddedEvent
+from strands.hooks import AfterInvocationEvent, MessageAddedEvent, MessageUpdatedEvent
 from strands.hooks.events import AgentInitializedEvent
 from strands.hooks.registry import HookRegistry
 from strands.session.repository_session_manager import RepositorySessionManager
@@ -47,8 +46,6 @@
 
 logger = logging.getLogger(__name__)
 
-MessageUpdatedEvent = getattr(strands.hooks, "MessageUpdatedEvent", None)
-
 MAX_FETCH_ALL_RESULTS = 10000
 
 # Legacy prefixes for backwards compatibility with old events
@@ -73,6 +70,12 @@ def _bidi_status(message: Mapping[str, Any]) -> Optional[str]:
     return status
 
 
+class _HeldMessage(NamedTuple):
+    message: Message
+    agent: "LocalAgent"
+    ready: bool
+
+
 class BufferedMessage(NamedTuple):
     """A pre-processed message waiting to be flushed to AgentCore Memory."""
 
@@ -176,6 +179,9 @@ def __init__(
         self._agent_state_buffer: list[tuple[str, SessionAgent]] = []
         self._agent_state_lock = threading.Lock()
 
+        self._held_messages: list[_HeldMessage] = []
+        self._held_lock = threading.Lock()
+
         # Cache for agent created_at timestamps to avoid fetching on every update
         self._agent_created_at_cache: dict[str, datetime] = {}
 
@@ -846,22 +852,64 @@ def _filter_restored_tool_context(self, messages: list[SessionMessage]) -> list[
     def append_message(self, message: Message, agent: "LocalAgent", **kwargs: Any) -> None:
         """Append a message to the agent's session using AgentCore's eventId as message_id.
 
+        A BidiAgent message that is still being streamed is written when it completes, and
+        messages added after it wait for it, so memory keeps the order of the agent's history.
+
         Args:
             message: Message to add to the agent in the session
             agent: Agent to append the message to
             **kwargs: Additional keyword arguments for future extensibility.
         """
-        if _bidi_status(message) == "pending":
+        with self._held_lock:
+            self._hold_or_write(message, agent)
+
+    def _hold_or_write(self, message: Message, agent: "LocalAgent") -> None:
+        ready = _bidi_status(message) != "pending"
+        if ready and not self._held_messages:
+            self._write_message(message, agent)
             return
+        self._held_messages.append(_HeldMessage(message, agent, ready))
+        self._write_ready_messages()
+
+    def _write_message(self, message: Message, agent: "LocalAgent") -> None:
         created_message = self.create_message(self.session_id, agent.agent_id, SessionMessage.from_message(message, 0))
         if created_message is None:
             return
         session_message = SessionMessage.from_message(message, created_message.get("eventId"))
         self._latest_agent_message[agent.agent_id] = session_message
 
-    def _append_completed_message(self, event: Any) -> None:
-        if _bidi_status(event.message) == "complete":
-            self.append_message(event.message, event.agent)
+    def _complete_message(self, event: MessageUpdatedEvent) -> None:
+        status = _bidi_status(event.message)
+        if status == "pending":
+            return
+        with self._held_lock:
+            index = next(
+                (
+                    i
+                    for i, held in enumerate(self._held_messages)
+                    if not held.ready and held.message.get("tracking_id") == event.tracking_id
+                ),
+                None,
+            )
+            if index is None:
+                if status == "complete":
+                    self._hold_or_write(event.message, event.agent)
+                return
+            if status == "complete":
+                self._held_messages[index] = _HeldMessage(event.message, event.agent, True)
+            else:
+                del self._held_messages[index]
+            self._write_ready_messages()
+
+    def _write_ready_messages(self) -> None:
+        while self._held_messages and self._held_messages[0].ready:
+            held = self._held_messages.pop(0)
+            self._write_message(held.message, held.agent)
+
+    def _release_held_messages(self) -> None:
+        with self._held_lock:
+            self._held_messages = [held for held in self._held_messages if held.ready]
+            self._write_ready_messages()
 
     def retrieve_customer_context(self, event: MessageAddedEvent) -> None:
         """Retrieve customer LTM context for regular Agent invocations.
@@ -971,8 +1019,8 @@ def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
 
             RepositorySessionManager.register_hooks(self, registry, **kwargs)
             registry.add_callback(MessageAddedEvent, lambda event: self.retrieve_customer_context(event))
-            if MessageUpdatedEvent is not None:
-                registry.add_callback(MessageUpdatedEvent, self._append_completed_message)
+            registry.add_callback(MessageUpdatedEvent, self._complete_message)
+            registry.add_callback(BidiAgentStopEvent, lambda event: self._release_held_messages())
             return
 
         # Async mode: register async callbacks that offload the existing sync
@@ -1008,8 +1056,7 @@ async def _on_message_added_persist(event: MessageAddedEvent) -> None:
             await asyncio.to_thread(self.sync_agent, event.agent)
 
         registry.add_callback(MessageAddedEvent, _on_message_added_persist)
-        if MessageUpdatedEvent is not None:
-            registry.add_callback(MessageUpdatedEvent, _offload(self._append_completed_message, lambda e: e))
+        registry.add_callback(MessageUpdatedEvent, _offload(self._complete_message, lambda e: e))
         registry.add_callback(AfterInvocationEvent, _offload(self.sync_agent, lambda e: e.agent))
         registry.add_callback(MessageAddedEvent, _offload(self.retrieve_customer_context, lambda e: e))
 
@@ -1019,6 +1066,7 @@ async def _on_message_added_persist(event: MessageAddedEvent) -> None:
         registry.add_callback(AfterMultiAgentInvocationEvent, _offload(self.sync_multi_agent, lambda e: e.source))
 
         registry.add_callback(BidiAgentStopEvent, _offload(self.sync_agent, lambda e: e.agent))
+        registry.add_callback(BidiAgentStopEvent, _offload(self._release_held_messages))
 
     @override
     def initialize(self, agent: "LocalAgent", **kwargs: Any) -> None:
@@ -1258,6 +1306,7 @@ def close(self) -> None:
         manager protocol (with statement) for automatic cleanup.
         """
         self._stop_flush_timer()
+        self._release_held_messages()
         self._flush_messages()
 
     def __enter__(self) -> "AgentCoreMemorySessionManager":
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 7c0967ac..86f90662 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
@@ -11,7 +11,6 @@
 import pytest
 from botocore.config import Config as BotocoreConfig
 from botocore.exceptions import ClientError
-from strands import hooks as strands_hooks
 from strands.agent.agent import Agent
 from strands.experimental.bidi import BidiAgent
 from strands.experimental.bidi.hooks import BidiAgentStopEvent
@@ -20,7 +19,7 @@
     AfterNodeCallEvent,
     MultiAgentInitializedEvent,
 )
-from strands.hooks import AfterInvocationEvent, AgentInitializedEvent, MessageAddedEvent
+from strands.hooks import AfterInvocationEvent, AgentInitializedEvent, MessageAddedEvent, MessageUpdatedEvent
 from strands.hooks.registry import HookRegistry
 from strands.types.exceptions import SessionException
 from strands.types.session import Session, SessionAgent, SessionMessage, SessionType
@@ -35,8 +34,6 @@
     BufferedMessage,
 )
 
-MessageUpdatedEvent = getattr(strands_hooks, "MessageUpdatedEvent", None)
-
 
 @pytest.fixture
 def agentcore_config():
@@ -2850,18 +2847,33 @@ async def test_completion_flushes_messages_and_final_state(
         assert [json.loads(payload["blob"])["state"] for payload in state_payloads] == [{}, {"status": "stopped"}]
 
 
-def _bidi_message(role, text, status):
+def _bidi_message(role, text, status, tracking_id=None):
     content = [] if text is None else [{"text": text}]
     return {
         "role": role,
         "content": content,
-        "tracking_id": f"{role}-1",
+        "tracking_id": tracking_id or f"{role}-1",
         "metadata": {"custom": {"bidi": {"kind": "transcript", "status": status}}},
     }
 
 
+def _stored_texts(mock_memory_client):
+    return [
+        (role.upper(), json.loads(payload)["message"]["content"][0])
+        for call in mock_memory_client.create_event.call_args_list
+        for payload, role in call.kwargs["messages"]
+    ]
+
+
+class _KeepEmptyConverter(AgentCoreMemoryConverter):
+    @staticmethod
+    def message_to_payload(session_message):
+        session_dict = session_message.to_dict()
+        return [(json.dumps(session_dict), session_dict["message"]["role"])]
+
+
 class TestBidiTranscripts:
-    """BidiAgent transcripts are stored once they are complete."""
+    """BidiAgent transcripts are stored once they are complete, in history order."""
 
     @pytest.fixture(params=[False, True], ids=["sync", "async"])
     def bidi_manager(self, request, agentcore_config, mock_memory_client):
@@ -2871,30 +2883,33 @@ def bidi_manager(self, request, agentcore_config, mock_memory_client):
         manager.sync_agent = Mock()
         registry = HookRegistry()
         manager.register_hooks(registry)
-        return registry, Mock(spec=BidiAgent, agent_id="test-agent")
+        return manager, registry, Mock(spec=BidiAgent, agent_id="test-agent")
 
-    async def test_pending_placeholder_is_not_stored(self, bidi_manager, mock_memory_client):
-        registry, agent = bidi_manager
+    async def test_pending_placeholder_is_not_stored_by_converter_keeping_empty_content(
+        self, bidi_manager, mock_memory_client
+    ):
+        manager, registry, agent = bidi_manager
+        manager.converter = _KeepEmptyConverter
 
         await registry.invoke_callbacks_async(
             MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
         )
+        await registry.invoke_callbacks_async(BidiAgentStopEvent(agent=agent))
 
         mock_memory_client.create_event.assert_not_called()
 
     @pytest.mark.parametrize("metadata", [None, {}, {"custom": None}, {"custom": {"bidi": {"kind": "tool_result"}}}])
     async def test_message_without_bidi_status_is_stored(self, bidi_manager, mock_memory_client, metadata):
-        registry, agent = bidi_manager
+        manager, registry, agent = bidi_manager
         message = {"role": "user", "content": [{"text": "Hello"}], "metadata": metadata}
 
         await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=message))
 
         mock_memory_client.create_event.assert_called_once()
 
-    @pytest.mark.skipif(MessageUpdatedEvent is None, reason="strands-agents has no MessageUpdatedEvent")
     @pytest.mark.parametrize(("role", "memory_role"), [("user", "USER"), ("assistant", "ASSISTANT")])
     async def test_complete_transcript_is_stored_once(self, bidi_manager, mock_memory_client, role, memory_role):
-        registry, agent = bidi_manager
+        manager, registry, agent = bidi_manager
         final = _bidi_message(role, "I need a hotel in London.", "complete")
 
         await registry.invoke_callbacks_async(
@@ -2902,33 +2917,121 @@ async def test_complete_transcript_is_stored_once(self, bidi_manager, mock_memor
         )
         await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, final["tracking_id"], final))
 
-        mock_memory_client.create_event.assert_called_once()
-        [(payload, stored_role)] = mock_memory_client.create_event.call_args.kwargs["messages"]
-        assert stored_role.upper() == memory_role
-        assert json.loads(payload)["message"]["content"] == [{"text": "I need a hotel in London."}]
+        assert _stored_texts(mock_memory_client) == [(memory_role, {"text": "I need a hotel in London."})]
 
-    @pytest.mark.skipif(MessageUpdatedEvent is None, reason="strands-agents has no MessageUpdatedEvent")
     @pytest.mark.parametrize(
         ("text", "status"),
         [("[Transcript unavailable.]", "incomplete"), ("", "complete"), ("   ", "complete"), ("partial", "pending")],
     )
     async def test_unfinished_or_empty_transcript_is_not_stored(self, bidi_manager, mock_memory_client, text, status):
-        registry, agent = bidi_manager
+        manager, registry, agent = bidi_manager
         update = _bidi_message("assistant", text, status)
 
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
         await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, update["tracking_id"], update))
 
         mock_memory_client.create_event.assert_not_called()
 
-    @pytest.mark.skipif(MessageUpdatedEvent is None, reason="strands-agents has no MessageUpdatedEvent")
     async def test_update_without_bidi_status_is_not_stored(self, bidi_manager, mock_memory_client):
-        registry, agent = bidi_manager
+        manager, registry, agent = bidi_manager
         message = {"role": "assistant", "content": [{"text": "redacted"}], "tracking_id": "x"}
 
         await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, "x", message))
 
         mock_memory_client.create_event.assert_not_called()
 
+    async def test_reply_completed_before_question_is_stored_after_it(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        question = _bidi_message("user", "Any rooms in London?", "complete")
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+        follow_up = {"role": "user", "content": [{"text": "Book one."}]}
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+        mock_memory_client.create_event.assert_not_called()
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, question["tracking_id"], question))
+
+        assert _stored_texts(mock_memory_client) == [
+            ("USER", {"text": "Any rooms in London?"}),
+            ("ASSISTANT", {"text": "Yes, two rooms."}),
+            ("USER", {"text": "Book one."}),
+        ]
+
+    async def test_incomplete_transcript_releases_messages_after_it(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        lost = _bidi_message("user", "[Transcript unavailable.]", "incomplete")
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, lost["tracking_id"], lost))
+
+        assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Yes, two rooms."})]
+
+    @pytest.mark.parametrize("finish", ["stop", "close"])
+    async def test_finishing_writes_messages_held_behind_open_transcript(
+        self, bidi_manager, mock_memory_client, finish
+    ):
+        manager, registry, agent = bidi_manager
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        if finish == "stop":
+            await registry.invoke_callbacks_async(BidiAgentStopEvent(agent=agent))
+        else:
+            manager.close()
+
+        assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Yes, two rooms."})]
+
+    async def test_failed_write_does_not_block_messages_after_it(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        question = _bidi_message("user", "Any rooms in London?", "complete")
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+        follow_up = {"role": "user", "content": [{"text": "Book one."}]}
+        mock_memory_client.create_event.side_effect = [ClientError({"Error": {"Code": "500"}}, "CreateEvent"), {}, {}]
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        with pytest.raises(SessionException):
+            await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, question["tracking_id"], question))
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+
+        failed, *stored = _stored_texts(mock_memory_client)
+        assert failed == ("USER", {"text": "Any rooms in London?"})
+        assert stored == [("ASSISTANT", {"text": "Yes, two rooms."}), ("USER", {"text": "Book one."})]
+
+    async def test_completed_message_without_placeholder_is_stored(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        final = _bidi_message("assistant", "Restored reply.", "complete", tracking_id="restored")
+
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, "restored", final))
+
+        assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Restored reply."})]
+
 
 class TestIntervalFlush:
     """Test interval-based flush mechanism for long-running agents."""

From c386c2ffe5f912c6fc4a95973b4ae7e4dc14ba7e Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Thu, 29 Oct 2026 01:41:54 -0400
Subject: [PATCH 4/7] fix(memory): drop held transcript only when it ends
 incomplete

A MessageUpdatedEvent that matched a waiting BidiAgent transcript but
carried no bidi status fell into the delete branch, so the transcript was
dropped and the messages behind it were written ahead of it. Only an
"incomplete" status means the transcript will never finish; any other
update now leaves it waiting until it completes or the agent stops.

Also cover history order when batch_size is above one, in sync and
async mode, including messages released when the agent stops.
---
 .../integrations/strands/session_manager.py   |  2 +-
 .../test_agentcore_memory_session_manager.py  | 58 +++++++++++++++++++
 2 files changed, 59 insertions(+), 1 deletion(-)

diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
index 26b5aa74..105d2178 100644
--- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
+++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
@@ -897,7 +897,7 @@ def _complete_message(self, event: MessageUpdatedEvent) -> None:
                 return
             if status == "complete":
                 self._held_messages[index] = _HeldMessage(event.message, event.agent, True)
-            else:
+            elif status == "incomplete":
                 del self._held_messages[index]
             self._write_ready_messages()
 
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 86f90662..91aa2538 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
@@ -2981,6 +2981,64 @@ async def test_incomplete_transcript_releases_messages_after_it(self, bidi_manag
 
         assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Yes, two rooms."})]
 
+    async def test_update_without_bidi_status_keeps_transcript_waiting(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        redacted = {"role": "user", "content": [{"text": "redacted"}], "tracking_id": "user-1"}
+        question = _bidi_message("user", "Any rooms in London?", "complete")
+        follow_up = {"role": "assistant", "content": [{"text": "Yes, two rooms."}]}
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, "user-1", redacted))
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+        mock_memory_client.create_event.assert_not_called()
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, question["tracking_id"], question))
+
+        assert _stored_texts(mock_memory_client) == [
+            ("USER", {"text": "Any rooms in London?"}),
+            ("ASSISTANT", {"text": "Yes, two rooms."}),
+        ]
+
+    @pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
+    async def test_batched_messages_keep_history_order(self, agentcore_config, mock_memory_client, async_mode):
+        agentcore_config.async_mode = async_mode
+        agentcore_config.batch_size = 10
+        manager = _create_session_manager(agentcore_config, mock_memory_client)
+        manager._latest_agent_message = {}
+        manager.sync_agent = Mock()
+        registry = HookRegistry()
+        manager.register_hooks(registry)
+        agent = Mock(spec=BidiAgent, agent_id="test-agent")
+        question = _bidi_message("user", "Any rooms in London?", "complete")
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+        follow_up = {"role": "user", "content": [{"text": "Book one."}]}
+        confirmation = {"role": "assistant", "content": [{"text": "Booked."}]}
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, question["tracking_id"], question))
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending", tracking_id="user-2"))
+        )
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=confirmation))
+        assert manager.pending_message_count() == 3
+        await registry.invoke_callbacks_async(BidiAgentStopEvent(agent=agent))
+
+        payload = mock_memory_client.gmdp_client.create_event.call_args.kwargs["payload"]
+        assert [json.loads(item["conversational"]["content"]["text"])["message"]["content"][0] for item in payload] == [
+            {"text": "Any rooms in London?"},
+            {"text": "Yes, two rooms."},
+            {"text": "Book one."},
+            {"text": "Booked."},
+        ]
+
     @pytest.mark.parametrize("finish", ["stop", "close"])
     async def test_finishing_writes_messages_held_behind_open_transcript(
         self, bidi_manager, mock_memory_client, finish

From 2779c0777b70c915e6743823dc8d78c6ad714f60 Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Fri, 6 Nov 2026 21:29:40 -0500
Subject: [PATCH 5/7] fix(memory): write a repeated transcript completion once
 and always flush on close

---
 .../integrations/strands/session_manager.py   | 24 ++++++++++---
 .../test_agentcore_memory_session_manager.py  | 36 +++++++++++++++++++
 2 files changed, 56 insertions(+), 4 deletions(-)

diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
index 5d3f9fc3..b205b56c 100644
--- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
+++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
@@ -185,6 +185,7 @@ def __init__(
         self._agent_state_lock = threading.Lock()
 
         self._held_messages: list[_HeldMessage] = []
+        self._completed_tracking_ids: set[str] = set()
         self._held_lock = threading.Lock()
 
         # Cache for agent created_at timestamps to avoid fetching on every update
@@ -879,6 +880,9 @@ def append_message(self, message: Message, agent: "LocalAgent", **kwargs: Any) -
 
     def _hold_or_write(self, message: Message, agent: "LocalAgent") -> None:
         ready = _bidi_status(message) != "pending"
+        tracking_id = message.get("tracking_id")
+        if ready and tracking_id:
+            self._completed_tracking_ids.add(tracking_id)
         if ready and not self._held_messages:
             self._write_message(message, agent)
             return
@@ -908,19 +912,29 @@ def _complete_message(self, event: MessageUpdatedEvent) -> None:
                 None,
             )
             if index is None:
-                if status == "complete":
+                if status == "complete" and event.tracking_id not in self._completed_tracking_ids:
                     self._hold_or_write(event.message, event.agent)
                 return
             if status == "complete":
+                self._completed_tracking_ids.add(event.tracking_id)
                 self._held_messages[index] = _HeldMessage(event.message, event.agent, True)
             elif status == "incomplete":
                 del self._held_messages[index]
             self._write_ready_messages()
 
     def _write_ready_messages(self) -> None:
+        first_error: Optional[Exception] = None
         while self._held_messages and self._held_messages[0].ready:
             held = self._held_messages.pop(0)
-            self._write_message(held.message, held.agent)
+            try:
+                self._write_message(held.message, held.agent)
+            except Exception as error:
+                if first_error is not None:
+                    logger.error("Failed to write held message: %s", error)
+                else:
+                    first_error = error
+        if first_error is not None:
+            raise first_error
 
     def _release_held_messages(self) -> None:
         with self._held_lock:
@@ -1345,8 +1359,10 @@ def close(self) -> None:
         manager protocol (with statement) for automatic cleanup.
         """
         self._stop_flush_timer()
-        self._release_held_messages()
-        self._flush_messages()
+        try:
+            self._release_held_messages()
+        finally:
+            self._flush_messages()
 
     def __enter__(self) -> "AgentCoreMemorySessionManager":
         """Enter the context manager.
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 63019edd..16668cd0 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
@@ -3303,6 +3303,42 @@ async def test_completed_message_without_placeholder_is_stored(self, bidi_manage
 
         assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Restored reply."})]
 
+    @pytest.mark.parametrize("placeholder", [False, True], ids=["orphan", "placeholder"])
+    async def test_repeated_complete_is_stored_once(self, bidi_manager, mock_memory_client, placeholder):
+        manager, registry, agent = bidi_manager
+        final = _bidi_message("assistant", "Yes, two rooms.", "complete")
+        if placeholder:
+            await registry.invoke_callbacks_async(
+                MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+            )
+
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, final["tracking_id"], final))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, final["tracking_id"], final))
+
+        assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Yes, two rooms."})]
+
+    async def test_close_writes_and_flushes_after_failed_held_write(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+        follow_up = {"role": "user", "content": [{"text": "Book one."}]}
+        mock_memory_client.create_event.side_effect = [ClientError({"Error": {"Code": "500"}}, "CreateEvent"), {}]
+        manager._flush_messages = Mock(wraps=manager._flush_messages)
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=reply))
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+        with pytest.raises(SessionException):
+            manager.close()
+
+        assert _stored_texts(mock_memory_client) == [
+            ("ASSISTANT", {"text": "Yes, two rooms."}),
+            ("USER", {"text": "Book one."}),
+        ]
+        assert manager._held_messages == []
+        manager._flush_messages.assert_called_once()
+
 
 class TestIntervalFlush:
     """Test interval-based flush mechanism for long-running agents."""

From 2f9a27930d4a04fd0f14a3b0fb98bb12bc8088ae Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Fri, 6 Nov 2026 21:35:54 -0500
Subject: [PATCH 6/7] fix(memory): ignore a transcript completion with no
 waiting message

---
 .../integrations/strands/session_manager.py   | 17 ++------
 .../test_agentcore_memory_session_manager.py  | 41 ++++++++++++++-----
 2 files changed, 34 insertions(+), 24 deletions(-)

diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
index b205b56c..2b002bc8 100644
--- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
+++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
@@ -4,6 +4,7 @@
 import json
 import logging
 import threading
+from collections import deque
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta, timezone
 from enum import Enum
@@ -184,8 +185,7 @@ def __init__(
         self._agent_state_buffer: list[tuple[str, SessionAgent]] = []
         self._agent_state_lock = threading.Lock()
 
-        self._held_messages: list[_HeldMessage] = []
-        self._completed_tracking_ids: set[str] = set()
+        self._held_messages: deque[_HeldMessage] = deque()
         self._held_lock = threading.Lock()
 
         # Cache for agent created_at timestamps to avoid fetching on every update
@@ -864,9 +864,6 @@ def _filter_restored_tool_context(self, messages: list[SessionMessage]) -> list[
     def append_message(self, message: Message, agent: "LocalAgent", **kwargs: Any) -> None:
         """Append a message to the agent's session using AgentCore's eventId as message_id.
 
-        A BidiAgent message that is still being streamed is written when it completes, and
-        messages added after it wait for it, so memory keeps the order of the agent's history.
-
         With batching, hold this message through early flushes because Strands may
         still request redaction of its latest message.
 
@@ -880,9 +877,6 @@ def append_message(self, message: Message, agent: "LocalAgent", **kwargs: Any) -
 
     def _hold_or_write(self, message: Message, agent: "LocalAgent") -> None:
         ready = _bidi_status(message) != "pending"
-        tracking_id = message.get("tracking_id")
-        if ready and tracking_id:
-            self._completed_tracking_ids.add(tracking_id)
         if ready and not self._held_messages:
             self._write_message(message, agent)
             return
@@ -912,11 +906,8 @@ def _complete_message(self, event: MessageUpdatedEvent) -> None:
                 None,
             )
             if index is None:
-                if status == "complete" and event.tracking_id not in self._completed_tracking_ids:
-                    self._hold_or_write(event.message, event.agent)
                 return
             if status == "complete":
-                self._completed_tracking_ids.add(event.tracking_id)
                 self._held_messages[index] = _HeldMessage(event.message, event.agent, True)
             elif status == "incomplete":
                 del self._held_messages[index]
@@ -925,7 +916,7 @@ def _complete_message(self, event: MessageUpdatedEvent) -> None:
     def _write_ready_messages(self) -> None:
         first_error: Optional[Exception] = None
         while self._held_messages and self._held_messages[0].ready:
-            held = self._held_messages.pop(0)
+            held = self._held_messages.popleft()
             try:
                 self._write_message(held.message, held.agent)
             except Exception as error:
@@ -938,7 +929,7 @@ def _write_ready_messages(self) -> None:
 
     def _release_held_messages(self) -> None:
         with self._held_lock:
-            self._held_messages = [held for held in self._held_messages if held.ready]
+            self._held_messages = deque(held for held in self._held_messages if held.ready)
             self._write_ready_messages()
 
     def retrieve_customer_context(self, event: MessageAddedEvent) -> None:
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 16668cd0..1b676298 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
@@ -3295,28 +3295,47 @@ async def test_failed_write_does_not_block_messages_after_it(self, bidi_manager,
         assert failed == ("USER", {"text": "Any rooms in London?"})
         assert stored == [("ASSISTANT", {"text": "Yes, two rooms."}), ("USER", {"text": "Book one."})]
 
-    async def test_completed_message_without_placeholder_is_stored(self, bidi_manager, mock_memory_client):
+    async def test_complete_without_waiting_message_is_ignored(self, bidi_manager, mock_memory_client):
         manager, registry, agent = bidi_manager
-        final = _bidi_message("assistant", "Restored reply.", "complete", tracking_id="restored")
+        final = _bidi_message("assistant", "Late reply.", "complete", tracking_id="late")
 
-        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, "restored", final))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, "late", final))
+        manager.close()
 
-        assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Restored reply."})]
+        mock_memory_client.create_event.assert_not_called()
 
-    @pytest.mark.parametrize("placeholder", [False, True], ids=["orphan", "placeholder"])
-    async def test_repeated_complete_is_stored_once(self, bidi_manager, mock_memory_client, placeholder):
+    async def test_repeated_complete_is_stored_once(self, bidi_manager, mock_memory_client):
         manager, registry, agent = bidi_manager
         final = _bidi_message("assistant", "Yes, two rooms.", "complete")
-        if placeholder:
-            await registry.invoke_callbacks_async(
-                MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
-            )
 
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
         await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, final["tracking_id"], final))
         await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, final["tracking_id"], final))
 
         assert _stored_texts(mock_memory_client) == [("ASSISTANT", {"text": "Yes, two rooms."})]
 
+    async def test_repeated_complete_while_waiting_is_stored_once(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        question = _bidi_message("user", "Any rooms in London?", "complete")
+        reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, reply["tracking_id"], reply))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, question["tracking_id"], question))
+
+        assert _stored_texts(mock_memory_client) == [
+            ("USER", {"text": "Any rooms in London?"}),
+            ("ASSISTANT", {"text": "Yes, two rooms."}),
+        ]
+
     async def test_close_writes_and_flushes_after_failed_held_write(self, bidi_manager, mock_memory_client):
         manager, registry, agent = bidi_manager
         reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
@@ -3336,7 +3355,7 @@ async def test_close_writes_and_flushes_after_failed_held_write(self, bidi_manag
             ("ASSISTANT", {"text": "Yes, two rooms."}),
             ("USER", {"text": "Book one."}),
         ]
-        assert manager._held_messages == []
+        assert not manager._held_messages
         manager._flush_messages.assert_called_once()
 
 

From e3d63ca77d8d6074f8fb14ec37cf307e89855876 Mon Sep 17 00:00:00 2001
From: Abhiram Shaji <write4abhiram@gmail.com>
Date: Fri, 6 Nov 2026 21:43:02 -0500
Subject: [PATCH 7/7] fix(memory): sync and flush on BidiAgent stop even when a
 held write fails

---
 .../integrations/strands/session_manager.py   | 28 +++++-----
 .../test_agentcore_memory_session_manager.py  | 53 +++++++++++++++++++
 2 files changed, 68 insertions(+), 13 deletions(-)

diff --git a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
index 2b002bc8..290e5f69 100644
--- a/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
+++ b/src/bedrock_agentcore/memory/integrations/strands/session_manager.py
@@ -872,16 +872,12 @@ def append_message(self, message: Message, agent: "LocalAgent", **kwargs: Any) -
             agent: Agent to append the message to
             **kwargs: Additional keyword arguments for future extensibility.
         """
-        with self._held_lock:
-            self._hold_or_write(message, agent)
-
-    def _hold_or_write(self, message: Message, agent: "LocalAgent") -> None:
         ready = _bidi_status(message) != "pending"
-        if ready and not self._held_messages:
-            self._write_message(message, agent)
-            return
-        self._held_messages.append(_HeldMessage(message, agent, ready))
-        self._write_ready_messages()
+        with self._held_lock:
+            if ready and not self._held_messages:
+                self._write_message(message, agent)
+            else:
+                self._held_messages.append(_HeldMessage(message, agent, ready))
 
     def _write_message(self, message: Message, agent: "LocalAgent") -> None:
         created_message = self.create_message(
@@ -906,6 +902,7 @@ def _complete_message(self, event: MessageUpdatedEvent) -> None:
                 None,
             )
             if index is None:
+                logger.debug("tracking_id=<%s> | no waiting message for transcript update, ignoring", event.tracking_id)
                 return
             if status == "complete":
                 self._held_messages[index] = _HeldMessage(event.message, event.agent, True)
@@ -932,6 +929,13 @@ def _release_held_messages(self) -> None:
             self._held_messages = deque(held for held in self._held_messages if held.ready)
             self._write_ready_messages()
 
+    def _release_and_flush(self) -> None:
+        try:
+            self._release_held_messages()
+        finally:
+            if self.config.batch_size > 1:
+                self._flush_messages()
+
     def retrieve_customer_context(self, event: MessageAddedEvent) -> None:
         """Retrieve customer LTM context for regular Agent invocations.
 
@@ -1036,12 +1040,11 @@ def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
             if self.config.batch_size > 1:
                 # Completion callbacks run in reverse order, so register flushes before state syncs.
                 registry.add_callback(AfterInvocationEvent, lambda event: self._flush_messages())
-                registry.add_callback(BidiAgentStopEvent, lambda event: self._flush_messages())
+            registry.add_callback(BidiAgentStopEvent, lambda event: self._release_and_flush())
 
             RepositorySessionManager.register_hooks(self, registry, **kwargs)
             registry.add_callback(MessageAddedEvent, lambda event: self.retrieve_customer_context(event))
             registry.add_callback(MessageUpdatedEvent, self._complete_message)
-            registry.add_callback(BidiAgentStopEvent, lambda event: self._release_held_messages())
             return
 
         # Async mode: register async callbacks that offload the existing sync
@@ -1068,7 +1071,7 @@ async def _callback(event):
         if self.config.batch_size > 1:
             # Completion callbacks run in reverse order, so register flushes before state syncs.
             registry.add_callback(AfterInvocationEvent, _offload(self._flush_messages))
-            registry.add_callback(BidiAgentStopEvent, _offload(self._flush_messages))
+        registry.add_callback(BidiAgentStopEvent, _offload(self._release_and_flush))
 
         registry.add_callback(AgentInitializedEvent, lambda event: self.initialize(event.agent))
 
@@ -1087,7 +1090,6 @@ async def _on_message_added_persist(event: MessageAddedEvent) -> None:
         registry.add_callback(AfterMultiAgentInvocationEvent, _offload(self.sync_multi_agent, lambda e: e.source))
 
         registry.add_callback(BidiAgentStopEvent, _offload(self.sync_agent, lambda e: e.agent))
-        registry.add_callback(BidiAgentStopEvent, _offload(self._release_held_messages))
 
     @override
     def initialize(self, agent: "LocalAgent", **kwargs: Any) -> None:
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 1b676298..943e8200 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
@@ -3336,6 +3336,29 @@ async def test_repeated_complete_while_waiting_is_stored_once(self, bidi_manager
             ("ASSISTANT", {"text": "Yes, two rooms."}),
         ]
 
+    async def test_incomplete_transcript_in_middle_is_skipped(self, bidi_manager, mock_memory_client):
+        manager, registry, agent = bidi_manager
+        question = _bidi_message("user", "Any rooms in London?", "complete")
+        cut_off = _bidi_message("assistant", "[Transcript unavailable.]", "incomplete")
+        follow_up = {"role": "user", "content": [{"text": "Book one."}]}
+
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(
+            MessageAddedEvent(agent=agent, message=_bidi_message("assistant", None, "pending"))
+        )
+        await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, cut_off["tracking_id"], cut_off))
+        mock_memory_client.create_event.assert_not_called()
+        await registry.invoke_callbacks_async(MessageUpdatedEvent(agent, question["tracking_id"], question))
+
+        assert _stored_texts(mock_memory_client) == [
+            ("USER", {"text": "Any rooms in London?"}),
+            ("USER", {"text": "Book one."}),
+        ]
+        assert not manager._held_messages
+
     async def test_close_writes_and_flushes_after_failed_held_write(self, bidi_manager, mock_memory_client):
         manager, registry, agent = bidi_manager
         reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
@@ -3359,6 +3382,36 @@ async def test_close_writes_and_flushes_after_failed_held_write(self, bidi_manag
         manager._flush_messages.assert_called_once()
 
 
+@pytest.mark.parametrize("async_mode", [False, True])
+async def test_stop_syncs_and_flushes_after_failed_held_write(agentcore_config, mock_memory_client, async_mode):
+    agentcore_config.async_mode = async_mode
+    agentcore_config.batch_size = 10
+    manager = _create_session_manager(agentcore_config, mock_memory_client)
+    manager._latest_agent_message = {}
+    manager.sync_agent = Mock()
+    registry = HookRegistry()
+    manager.register_hooks(registry)
+    agent = Mock(spec=BidiAgent, agent_id="test-agent")
+    reply = _bidi_message("assistant", "Yes, two rooms.", "complete")
+    follow_up = {"role": "user", "content": [{"text": "Book one."}]}
+
+    await registry.invoke_callbacks_async(
+        MessageAddedEvent(agent=agent, message=_bidi_message("user", None, "pending"))
+    )
+    await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=reply))
+    await registry.invoke_callbacks_async(MessageAddedEvent(agent=agent, message=follow_up))
+    manager.sync_agent.reset_mock()
+    manager.create_message = Mock(side_effect=[SessionException("write failed"), {"eventId": "e1"}])
+    manager._flush_messages = Mock()
+    with pytest.raises(SessionException):
+        await registry.invoke_callbacks_async(BidiAgentStopEvent(agent=agent))
+
+    manager.sync_agent.assert_called_once_with(agent)
+    assert manager.create_message.call_count == 2
+    manager._flush_messages.assert_called_once()
+    assert not manager._held_messages
+
+
 class TestIntervalFlush:
     """Test interval-based flush mechanism for long-running agents."""