Class: LittleGhost::SessionStores::AgentCoreMemory

Inherits:
LittleGhost::SessionStore show all
Defined in:
lib/little_ghost/session_stores/agent_core_memory.rb

Overview

AgentCoreMemory keeps LittleGhost conversations in Amazon Bedrock AgentCore Memory so they can resume across Ruby processes and deployments.

store = LittleGhost::SessionStores::AgentCoreMemory.new(
memory_id: ENV.fetch("AGENTCORE_MEMORY_ID"),
region: "us-east-1"
)

Configure the resulting store through Configuration#session_store; a Runtime then owns its construction and lifetime. The optional aws-sdk-bedrockagentcore dependency is loaded only when a client is not supplied.

Privacy and concurrency

This store sends session data to Amazon Bedrock AgentCore Memory. For stored transcripts and checkpoints, Session removes system messages, transient messages, and private reasoning first. The remaining complete message records may still contain personal data, visible text, attachments, tool calls and results, and message metadata. Checkpoints also send application state and session metadata.

Conversation projection is a separate path. It removes private reasoning, but sends visible text from every message the caller supplies, including system or transient messages. Callers must filter projection input when those messages should stay local. Projection also sends selected metadata. None of this filtering anonymizes the remaining content.

Use a memory, region, IAM policy, retention policy, and logging policy approved for that data. Do not enable this store for content that is not approved to leave the Ruby process.

Session and actor identifiers become deterministic SHA-256 pseudonyms before leaving the process. These values remain linkable, and low-entropy identifiers may be recovered by dictionary matching. Treat them as sensitive identifiers, not anonymous data.

AgentCore's immutable event API requires one active writer for each actor/session pair. This store serializes writers inside one Ruby process, but horizontally scaled applications need an external lock or unique active-run record. Commits use generation and checkpoint records so an incomplete write is never exposed as a successful snapshot.

AgentCore checkpoints are versioned. This store reads supported historical versions without writing during #load, then writes the current checkpoint format when the session next appends or replaces its snapshot.

Constant Summary collapse

MESSAGE_PREFIX =

:nodoc:

"little_ghost:message:v4:"
MESSAGE_CHUNK_PREFIX =

:nodoc:

"little_ghost:message_chunk:v4:"
CHECKPOINT_PREFIX =

:nodoc:

"little_ghost:checkpoint:v5:"
LEGACY_CHECKPOINT_PREFIX =

:nodoc:

"little_ghost:checkpoint:v4:"
CONVERSATIONAL_TEXT_LIMIT =

:nodoc:

100_000
MESSAGE_CHUNK_CONTENT_LIMIT =

:nodoc:

90_000
EVENT_PAYLOAD_LIMIT =

:nodoc:

100
MESSAGE_CHUNK_COUNT_LIMIT =

:nodoc:

10_000
EVENT_TYPE_METADATA_KEY =

:nodoc:

"little_ghost_type"
GENERATION_METADATA_KEY =

:nodoc:

"little_ghost_generation"
COMMIT_METADATA_KEY =

:nodoc:

"little_ghost_commit"
SYMBOL_KEY_PREFIX =

:nodoc:

"little_ghost:symbol:"
STRING_KEY_PREFIX =

:nodoc:

"little_ghost:string:"
MESSAGE_EVENT_TYPE =

:nodoc:

"message_v4"
CHECKPOINT_EVENT_TYPE =

:nodoc:

"checkpoint_v5"
LEGACY_CHECKPOINT_EVENT_TYPE =

:nodoc:

"checkpoint_v4"
CONVERSATION_PROJECTION_EVENT_TYPE =

:nodoc:

"conversation_projection_v1"
PROJECTION_METADATA_KEYS = 
%w[
  little_ghost_parent_link
  little_ghost_conversation_id
  little_ghost_subagent_id
  little_ghost_kind
  little_ghost_turn
].freeze
EVENT_TIMESTAMP_INCREMENT =

AgentCore's SDK timestamp transport cannot preserve sub-second ordering.

1
LIST_PAGE_SIZE =

:nodoc:

100
MAX_LIST_PAGES =

:nodoc:

1_000
MAX_CHECKPOINT_EVENTS =

:nodoc:

10_000
MAX_GENERATION_EVENTS =

:nodoc:

10_000
MAX_GENERATION_PAYLOADS =

:nodoc:

25_000
MAX_EVENT_SERIALIZED_BYTES =

:nodoc:

10 * 1024 * 1024
MAX_CHECKPOINT_READ_BYTES =

:nodoc:

64 * 1024 * 1024
MAX_SESSION_SERIALIZED_BYTES =

:nodoc:

128 * 1024 * 1024
MAX_MESSAGE_SERIALIZED_BYTES =

:nodoc:

16 * 1024 * 1024
MAX_CHECKPOINT_SERIALIZED_BYTES =

:nodoc:

1 * 1024 * 1024
MAX_SESSION_MESSAGES =

:nodoc:

10_000
MAX_REVISION =

:nodoc:

(2**63) - 1

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from LittleGhost::SessionStore

#synchronize

Constructor Details

#initialize(memory_id:, client: nil, client_factory: nil, region: nil, clock: -> { Time.now }) ⇒ AgentCoreMemory

Supply client for explicit dependency injection, or region and an optional client_factory for lazy refresh.

Raises:

  • (ArgumentError) 


104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 104

def initialize(
  memory_id:,
  client: nil,
  client_factory: nil,
  region: nil,
  clock: -> { Time.now }
)
  super()
  @memory_id = String(memory_id)
  raise ArgumentError, "memory_id must not be empty" if @memory_id.empty?

  @region = region
  @client_factory = client_factory || -> { build_client(@region) }
  @client = client || @client_factory.call
  @clock = clock
  @operation_context_key = :"little_ghost_session_store_operation_#{object_id}"
  @client_mutex = Mutex.new
  @persistence_locks = {}
  @persistence_locks_mutex = Mutex.new
end

Class Method Details

.safe_id(value) ⇒ Object

Produces a stable AgentCore-safe pseudonym. This is not anonymization.



98
99
100
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 98

def self.safe_id(value)
  "lg_#{Digest::SHA256.hexdigest(String(value))}"
end

Instance Method Details

#append(id, messages:, state:, metadata:, expected_count:, actor_id: nil) ⇒ Object

Appends sanitized messages as a new committed checkpoint when expected_count matches the latest remote generation.



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 143

def append(id, messages:, state:, metadata:, expected_count:, actor_id: nil)
  messages = persistable_messages(messages)
  state = DataMap.new(state).to_h
  metadata = DataMap.new(metadata).to_h
  actor = self.class.safe_id(required_actor_id(actor_id))
  session = self.class.safe_id(id)
  key = [actor, session]
  synchronize_persistence(key) do
    head, lineage = latest_checkpoint(actor, session)
    persistence = head&.fetch(:checkpoint)
    persisted_count = persistence&.fetch(:message_count, 0) || 0
    unless persisted_count == expected_count
      raise ProtocolError, "Session changed while it was being updated"
    end

    if head && head.fetch(:format) != CHECKPOINT_EVENT_TYPE
      existing = messages_from(message_records_for(actor, session, lineage:), lineage:)
      messages = [*existing, *messages].freeze
      expected_count = 0
      persistence = head.fetch(:checkpoint)
      root = true
      generation = SecureRandom.uuid
    else
      root = persistence.nil?
      generation = persistence&.fetch(:generation) || SecureRandom.uuid
    end
    commit_id = SecureRandom.uuid
    plan = plan_messages(messages, generation:, commit_id:, offset: expected_count)
    checkpoint = build_checkpoint(
      persistence:,
      generation:,
      commit_id:,
      root:,
      plan:,
      message_count: root ? messages.length : expected_count + messages.length,
      state:,
      metadata:
    )
    persist_commit(actor, session, plan:, checkpoint:, previous_timestamp: head&.fetch(:event_timestamp))
  end
  {messages:, state:, metadata:}
end

#load(id, actor_id: nil) ⇒ Object

Loads the latest committed generation for the required actor and session.



126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 126

def load(id, actor_id: nil)
  actor = self.class.safe_id(required_actor_id(actor_id))
  session = self.class.safe_id(id)
  head, lineage = latest_checkpoint(actor, session)
  return unless head

  checkpoint = head.fetch(:checkpoint)
  records = message_records_for(actor, session, lineage:)
  {
    messages: messages_from(records, lineage:),
    state: checkpoint.fetch(:state),
    metadata: checkpoint.fetch(:metadata)
  }
end

#project_conversation(id, messages:, metadata:, actor_id: nil) ⇒ Object

Writes visible conversational text for AgentCore Memory extraction without changing LittleGhost's stored session transcript. This removes private reasoning, but does not remove system or transient messages; callers must omit any message whose visible text should stay local.



219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 219

def project_conversation(id, messages:, metadata:, actor_id: nil)
  payload = persistable_messages(messages).filter_map do |message|
    text = message.text
    next if text.empty?

    conversational_payload(text, message.role)
  end
  return if payload.empty?

  event_metadata = {
    EVENT_TYPE_METADATA_KEY => {string_value: CONVERSATION_PROJECTION_EVENT_TYPE}
  }
  PROJECTION_METADATA_KEYS.each do |key|
    value = metadata[key] || metadata[key.to_sym]
    event_metadata[key] = {string_value: value.to_s} unless value.nil?
  end
  agent_core_call(
    :create_event,
    memory_id: @memory_id,
    actor_id: self.class.safe_id(required_actor_id(actor_id)),
    session_id: self.class.safe_id(id),
    event_timestamp: next_event_timestamp(nil),
    payload:,
    metadata: event_metadata,
    extraction_mode: "SKIP"
  )
end

#replace(id, messages:, state:, metadata:, actor_id: nil) ⇒ Object

Replaces the visible snapshot by committing a new remote generation.



187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 187

def replace(id, messages:, state:, metadata:, actor_id: nil)
  messages = persistable_messages(messages)
  state = DataMap.new(state).to_h
  metadata = DataMap.new(metadata).to_h
  actor = self.class.safe_id(required_actor_id(actor_id))
  session = self.class.safe_id(id)
  key = [actor, session]
  synchronize_persistence(key) do
    head, = latest_checkpoint(actor, session)
    persistence = head&.fetch(:checkpoint)
    generation = SecureRandom.uuid
    commit_id = SecureRandom.uuid
    plan = plan_messages(messages, generation:, commit_id:, offset: 0)
    checkpoint = build_checkpoint(
      persistence:,
      generation:,
      commit_id:,
      root: true,
      plan:,
      message_count: messages.length,
      state:,
      metadata:
    )
    persist_commit(actor, session, plan:, checkpoint:, previous_timestamp: head&.fetch(:event_timestamp))
  end
  {messages:, state:, metadata:}
end

#with_operation_context(operation_id) ⇒ Object

Parents AgentCore telemetry emitted in the block to operation_id.



248
249
250
 
# File 'lib/little_ghost/session_stores/agent_core_memory.rb', line 248

def with_operation_context(operation_id)
  ExecutionState.with(@operation_context_key => operation_id) { yield }
end