Class: Pikuri::Memory::Recorder

Inherits:
Object
  • Object
show all
Defined in:
lib/pikuri/memory/recorder.rb

Overview

The off-the-interaction-path capture queue. A single background worker thread drains enqueued user turns into Mem0Client#add, so a turn never blocks on the ~3s extraction round-trip.

rec = Recorder.new(client:, user_id: 'martin').start
rec.enqueue("I prefer terse test output")  # non-blocking
rec.close                                   # bounded flush, then stop

A thread (not a fork or external queue) keeps capture per-turn: a kill -9 / closed tab loses at most the in-flight turn, not a session. No persistence — durability is "within seconds, in mem0," not "survives a crash mid-extraction."

#close signals the worker to drain the backlog, then joins with a timeout so "quit" never hangs for minutes; anything still queued is dropped and reaped at process exit. One enqueue per turn means the common case flushes in one add. An add that raises (mem0 down, timeout, 5xx) is logged at WARN and the item dropped — a transient failure must never crash the worker (which would silently end all future capture). See DESIGN.md §"Capture: off-path and bounded".

Constant Summary collapse

LOGGER = 
Pikuri.logger_for('Memory::Recorder')
STOP =

Returns sentinel pushed by #close to stop the worker after it drains everything enqueued ahead of it.

Returns:

  • (Symbol) —

    sentinel pushed by #close to stop the worker after it drains everything enqueued ahead of it.

:__pikuri_memory_stop__
DEFAULT_FLUSH_TIMEOUT =

Returns default seconds #close waits for the worker to drain before giving up and letting process exit reap it.

Returns:

  • (Integer) —

    default seconds #close waits for the worker to drain before giving up and letting process exit reap it.

10

Instance Method Summary collapse

Constructor Details

#initialize(client:, user_id:, infer: true, prompt: nil, flush_timeout: DEFAULT_FLUSH_TIMEOUT) ⇒ Recorder

Parameters:

  • client (Mem0Client) —

    the mem0 client add is dispatched to.

  • user_id (String) —

    the mem0 namespace captures land in.

  • infer (Boolean) (defaults to: true) —

    forwarded to Mem0Client#add; true stores extracted facts.

  • prompt (String, nil) (defaults to: nil) —

    optional per-request extraction prompt forwarded to Mem0Client#add.

  • flush_timeout (Integer) (defaults to: DEFAULT_FLUSH_TIMEOUT) —

    seconds #close blocks waiting for the backlog to drain.



45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/pikuri/memory/recorder.rb', line 45

def initialize(client:, user_id:, infer: true, prompt: nil,
               flush_timeout: DEFAULT_FLUSH_TIMEOUT)
  @client = client
  @user_id = user_id
  @infer = infer
  @prompt = prompt
  @flush_timeout = flush_timeout
  @queue = Thread::Queue.new
  @thread = nil
  @closed = false
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Stop the worker and flush the backlog, bounded by flush_timeout. Pushes STOP (so queued items drain first), then joins for at most the timeout; anything still pending is abandoned to process-exit reaping. Idempotent.



86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/pikuri/memory/recorder.rb', line 86

def close
  return if @closed

  @closed = true
  @queue << STOP
  return unless @thread

  return if @thread.join(@flush_timeout)

  LOGGER.warn("flush did not finish within #{@flush_timeout}s; " \
              "#{@queue.size} memory write(s) abandoned at exit")
  nil
end

#enqueue(content) ⇒ void

This method returns an undefined value.

Enqueue one user turn for asynchronous extraction. Non-blocking (a bounded push to an in-memory queue). Blank content and post-close enqueues are silently ignored.

Parameters:

  • content (String) —

    the user's message to capture.



72
73
74
75
76
77
78
 
# File 'lib/pikuri/memory/recorder.rb', line 72

def enqueue(content)
  return if @closed
  return if content.nil? || content.to_s.strip.empty?

  @queue << content
  nil
end

#start ⇒ self

Start the background worker. Idempotent — a second call is a no-op, so Extension#bind can call it without guarding.

Returns:

  • (self) 


61
62
63
64
 
# File 'lib/pikuri/memory/recorder.rb', line 61

def start
  @thread ||= Thread.new { run_loop }
  self
end