feat(memory): Optimize API calls with batched storage for better latency/cost

[kevmyung]

Closes #222

Summary

• Optimize AgentCore Memory API calls by batching message and agent state into single requests
• Add state hash tracking to skip redundant sync calls
• Implement dual-read approach for backward compatibility with existing data

API Call Reduction

Scenario	Before	After	Reduction
Simple message (new session)	5+ calls	2 calls	60%+
With 2 tool calls	9+ calls	4 calls	55%+

Implementation Details

Batched Storage

• Messages and agent state are saved in a single API call via _save_message_with_state()
• Payload uses _type markers: "message" and "agent_state"

State Hash Tracking

• _compute_state_hash() computes hash excluding timestamps (created_at, updated_at)
• AfterInvocationEvent sync is skipped when state is unchanged

Dual-Read for Backward Compatibility

• read_agent() tries new format first (unified actorId with type markers)
• Falls back to legacy format (agent_{id} actorId) for existing data
• Events are processed in reverse order to get latest state

Files Changed

• session_manager.py: Batched storage, hash tracking, dual-read
• bedrock_converter.py: Parse new payload format

Test Plan

• Unit tests for batched message + state (13 TestOptimizedStorage tests)
• Unit tests for hash computation (excludes timestamps, detects state changes)
• Unit tests for dual-read with legacy fallback
• Unit tests for multi-turn and multi-agent scenarios
• Integration tests with API call counting
• Multi-turn context preservation test
• All 69 tests pass

Breaking Changes

None - existing data is read via legacy fallback, new data uses optimized format.

[aidandaly24]

issue: This v2 branch is missing the multi-agent hooks that the parent class registers:

• MultiAgentInitializedEvent → initialize_multi_agent
• AfterNodeCallEvent → sync_multi_agent
• AfterMultiAgentInvocationEvent → sync_multi_agent

Multi-agent scenarios will silently fail to persist state with v2. Consider either:

1. Adding these hooks explicitly here, or
2. Calling RepositorySessionManager.register_hooks() first, then overriding specific callbacks

Suggested fix:

# Add after line 735:
registry.add_callback(MultiAgentInitializedEvent, lambda event: self.initialize_multi_agent(event.source))
registry.add_callback(AfterNodeCallEvent, lambda event: self.sync_multi_agent(event.source))
registry.add_callback(AfterMultiAgentInvocationEvent, lambda event: self.sync_multi_agent(event.source))

[kevmyung]

Added multi-agent hooks from strands.experimental.hooks.multiagent

[aidandaly24]

small: This silently swallows the exception, but save_message_with_state (line 243) raises SessionException on failure. Should this also raise to maintain consistent error handling? Silent failures here could lead to undetected data loss.

Suggested fix:

except Exception as e:
   logger.error("Failed to sync agent state: %s", e)
   raise SessionException(f"Failed to sync agent state: {e}") from e

[kevmyung]

Fixed - raises SessionException on failure

[aidandaly24]

issue: V2 only queries config.actor_id, but V1 stored agent state under agent_{id}. If a user switches from v1 to v2 on an existing session, their agent state becomes inaccessible (appears as a new agent, losing previous state).

This silently loses access to existing data when upgrading.

Suggested fix - add fallback to v1 location:

# After the v2 loop finds nothing, before returning None:
# Fallback: check v1 location for migration support
events = self.memory_client.list_events(
   memory_id=self.config.memory_id,
   actor_id=self._get_full_agent_id(agent_id),
   session_id=session_id,
   max_results=1,
)
if events:
   agent_data = json.loads(events[0].get("payload", {})[0].get("blob"))
   return SessionAgent.from_dict(agent_data)

[kevmyung]

Fixed - dual-read with legacy fallback

[aidandaly24]

optimization: I know you stated one of the main purposes of the PR is to reduce redundant API calls, this isn't a big issue but I believe there is a chance for more optimization here. Agent state is saved twice at the end of each invocation:

1. With the final message via save_message_with_state (line 732)
2. Again here via _sync_agent_state

This adds an extra API call per invocation. The final sync is meant to capture conversation_manager_state updates that happen after the last message, but in most cases the state hasn't changed.

Suggested optimization: Track whether state changed since last save, and skip the final sync if unchanged:

# In save_message_with_state, store a hash/snapshot:
self._last_synced_state = hash(json.dumps(session_agent.to_dict(), sort_keys=True))

# In syncagent_state, check before saving:
def syncagent_state(self, agent: "Agent") -> None:
   current_state = hash(json.dumps(SessionAgent.from_agent(agent).to_dict(), sort_keys=True))
   if current_state == getattr(self, '_last_synced_state', None):
       return  # No change, skip sync
   # ... rest of method

[kevmyung]

Fixed - skips sync when state unchanged

[codecov-commenter]

Codecov Report

❌ Patch coverage is 77.65957% with 21 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@52bc194). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
...ore/memory/integrations/strands/session_manager.py	77.50%	16 Missing and 2 partials ⚠️
...e/memory/integrations/strands/bedrock_converter.py	78.57%	0 Missing and 3 partials ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main     #223   +/-   ##
=======================================
  Coverage        ?   90.25%           
=======================================
  Files           ?       35           
  Lines           ?     3488           
  Branches        ?      518           
=======================================
  Hits            ?     3148           
  Misses          ?      190           
  Partials        ?      150           

Flag	Coverage Δ
unittests	90.25% <77.65%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[aidandaly24]

looks good to me -- approved