Class: ClaudeAgentSDK::TranscriptMirrorBatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/claude_agent_sdk/transcript_mirror_batcher.rb

Overview

Batching layer between transcript_mirror stdout frames and a SessionStore.

The CLI subprocess emits {"type":"transcript_mirror","filePath":...,"entries":[...]} frames interleaved with normal SDK messages. The Query read loop peels these off and hands them to #enqueue, which accumulates them and flushes to SessionStore#append either when a result message arrives (explicit #flush) or when the pending buffer exceeds size thresholds (eager background flush). This keeps adapter latency off the hot path during model streaming.

Adapter failures are retried (MIRROR_APPEND_MAX_ATTEMPTS total) with short backoff; timeouts are not retried since the in-flight call may still land. Failures never raise — the local-disk transcript is already durable, so the session continues unaffected — and are reported via the on_error callback (which surfaces them as a MirrorErrorMessage). Adapters should dedupe by entry when present, since a retried batch may overlap a prior write.

The semaphore serializes appends, but a #send that exceeds send_timeout is abandoned (its worker thread keeps running) and the next drain proceeds, so two #append calls for the SAME key can briefly overlap. SessionStore#append must be thread-safe per key (see that method's contract). For adapters declaring callback_scheduling -> :inline the timed-out call is instead cancelled cooperatively (interrupted at its next suspension point, ensure runs) — but only on the adapter's own fiber; work the adapter offloaded may still land, so timeouts are not retried in either mode and the cancelled append may remain permanently HALF-applied in the store. The drop is surfaced (MirrorErrorMessage, batches_dropped?); the local transcript remains the source of truth.

Constant Summary collapse

MAX_PENDING_ENTRIES =

Eager-flush thresholds (exposed for tests).

500
MAX_PENDING_BYTES =

1 MiB

1 << 20
SEND_TIMEOUT_SECONDS = 
60.0
MIRROR_APPEND_MAX_ATTEMPTS =

Bounded retry for transient adapter failures. The backoff list length is MIRROR_APPEND_MAX_ATTEMPTS - 1 (one delay between each pair of attempts).

3
MIRROR_APPEND_BACKOFF_S = 
[0.2, 0.8].freeze

Instance Method Summary collapse

Constructor Details

#initialize(store:, projects_dir:, on_error:, send_timeout: SEND_TIMEOUT_SECONDS, max_pending_entries: MAX_PENDING_ENTRIES, max_pending_bytes: MAX_PENDING_BYTES, callback_wrapper: nil) ⇒ TranscriptMirrorBatcher

Returns a new instance of TranscriptMirrorBatcher.

Parameters:

  • store (SessionStore) —

    the adapter to mirror into

  • projects_dir (String) —

    base dir for file_path -> SessionKey mapping

  • on_error (#call) —

    called as on_error.call(key, message) after a batch exhausts retries; must not raise

  • callback_wrapper (#call, nil) (defaults to: nil) —

    the session's ClaudeAgentOptions#callback_wrapper, composed around each #append inside the timeout bound (worker thread in :thread mode, cooperative timeout in :inline mode) — so e.g. Rails' executor.wrap covers an AR-backed adapter and checks connections back in per call



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/claude_agent_sdk/transcript_mirror_batcher.rb', line 59

def initialize(store:, projects_dir:, on_error:, send_timeout: SEND_TIMEOUT_SECONDS,
               max_pending_entries: MAX_PENDING_ENTRIES, max_pending_bytes: MAX_PENDING_BYTES,
               callback_wrapper: nil)
  @store = store
  @projects_dir = projects_dir
  @on_error = on_error
  @send_timeout = send_timeout
  @callback_wrapper = callback_wrapper
  # Probed ONCE here so an invalid callback_scheduling declaration fails
  # at construction, not mid-session inside a flush.
  @store_scheduling = SessionStores.store_callback_scheduling(store)
  @max_pending_entries = max_pending_entries
  @max_pending_bytes = max_pending_bytes
  @pending = []
  @pending_entries = 0
  @pending_bytes = 0
  # Batches that exhausted retries (or could not be keyed) and never
  # reached the store. Written only under @lock; read cross-thread by
  # #batches_dropped? (a plain Integer read is safe under the GVL).
  @dropped_batches = 0
  # Fiber-aware lock: the critical section blocks on SessionStore#append
  # (a thread hop), so a Thread::Mutex would deadlock the reactor. The
  # semaphore serializes drains so append ordering matches enqueue order.
  @lock = Async::Semaphore.new(1)
end

Instance Method Details

#batches_dropped? ⇒ Boolean

True when at least one batch of entries never reached the store — the mirror copy is incomplete. Consulted at teardown by the resume-from-store cleanup so the materialized temp dir (which then holds the only copy of the dropped turns) is preserved instead of deleted.

Returns:

  • (Boolean) 


89
90
91
 
# File 'lib/claude_agent_sdk/transcript_mirror_batcher.rb', line 89

def batches_dropped?
  @dropped_batches.positive?
end

#close ⇒ Object

Final flush before teardown. Never raises.



126
127
128
129
130
 
# File 'lib/claude_agent_sdk/transcript_mirror_batcher.rb', line 126

def close
  flush
rescue StandardError => e
  warn "Claude SDK: TranscriptMirrorBatcher close flush failed: #{e.message}"
end

#enqueue(file_path, entries) ⇒ Object

Buffer a frame; schedule an eager background flush if thresholds are exceeded. Synchronous and fire-and-forget.

entries are deep-stringified because the transport parses CLI output with symbolized keys, but SessionStore entries are opaque JSON blobs that must round-trip through string keys (Postgres JSONB / Redis) and feed fold_session_summary, which reads string keys.



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/claude_agent_sdk/transcript_mirror_batcher.rb', line 100

def enqueue(file_path, entries)
  entries = deep_stringify(Array(entries))
  # An empty frame mirrors nothing (do_flush skips empty keys anyway), so
  # drop it here: otherwise its 2-byte "[]" inflates @pending_bytes and, in
  # eager mode (thresholds 0), schedules a no-op background drain per frame.
  return if entries.empty?

  # Approximate wire size — one stringify per frame (not per entry).
  size = JSON.generate(entries).bytesize
  @pending << { file_path: file_path, entries: entries }
  @pending_entries += entries.length
  @pending_bytes += size
  return unless @pending_entries > @max_pending_entries || @pending_bytes > @max_pending_bytes

  task = Async::Task.current?
  # Fire-and-forget on the reactor; @lock in #drain serializes against any
  # in-flight flush so append ordering holds. #drain never raises.
  task ? task.async { drain } : drain
end

#flush ⇒ Object

Flush all pending entries, serialized after any in-flight eager flush.



121
122
123
 
# File 'lib/claude_agent_sdk/transcript_mirror_batcher.rb', line 121

def flush
  drain
end