Module: ClaudeAgentSDK::FiberBoundary

Defined in:
lib/claude_agent_sdk/fiber_boundary.rb

Overview

Internal. Consumers of the SDK should never need this directly.

The SDK depends on async, which installs a Fiber scheduler whenever an Async { } block is active. That scheduler multiplexes fibers onto a single OS thread and intercepts IO so blocking calls yield to siblings.

Most mature Ruby libraries are thread-safe but not fiber-safe: they key state (checked-out DB connections, per-thread caches, request stores) on Thread.current. When the scheduler interleaves two fibers on one thread, those fibers share one state slot — and interleaved IO on a shared connection silently corrupts wire protocols. This bites every DB driver keyed by thread (pg, mysql2, sqlite3), ActiveRecord's connection pool, and any HTTP/cache client pooled per-thread.

The SDK invokes user-supplied callbacks (tool handlers, hooks, permission callbacks, message blocks, observer methods) from inside its reactor. FiberBoundary.invoke hops those calls to a plain Ruby thread so user code runs on a fiber-scheduler-free thread and inherits the same thread-keyed state assumptions the rest of the user's app makes.

No-op when no scheduler is active, so it's cheap to use unconditionally.

OPT-IN INLINE MODE: hosts that are already fiber-isolated (e.g. Rails apps with IsolatedExecutionState.isolation_level = :fiber running solid_queue fiber workers) can pass scheduling: :inline to run callbacks in place on the reactor fiber instead of hopping. This is the same code path as the no-scheduler case — semantically the already- shipped synchronous path, and Python SDK parity (async callbacks run natively on the event loop there). The mode is plumbed per-call from ClaudeAgentOptions#callback_scheduling — never a thread-local (the reactor thread is shared by many fibers, so a set/reset window across suspension points would leak across sessions). The one path a call argument cannot cross (SDK-MCP dispatch through the mcp gem) carries it via a scoped, invalidatable fiber-storage entry instead — see DISPATCH_KEY / CallbackDispatchScope below.

A timeout: forces the thread hop — Thread#join(timeout) is a hard bound that can abandon a wedged call, which cooperative with_timeout cancellation cannot guarantee — unless the caller passes scheduling: :inline inside a reactor, where the bound becomes a cooperative with_timeout cancellation instead (issue #47 phase 3: fiber-native SessionStore adapters that declare callback_scheduling -> :inline). The store-adapter call sites (TranscriptMirrorBatcher, SessionResume) default to the thread hop so an undeclared (possibly scheduler-opaque) adapter can never stall the reactor; a declared-inline adapter accepts that a scheduler-opaque blocking call would stall it AND escape the cooperative deadline. Outside a reactor the hard bound applies even to inline-declared adapters — the timeout guarantee is never lost.

The thread hop severs break/return/next from the surrounding method, so SDK loops yielding user callbacks must keep loop control outside the invoked block (see Client#receive_response); user-initiated break is bridged back to the calling fiber via .invoke_iteration.

Deliberate carve-out: the STREAMING-INPUT enumerable is the one user-code path iterated ON the reactor (Query#stream_input), matching Python where async input generators run on the event loop. Enumerator#next is fiber-based and cannot be pulled across threads, and a whole-iteration thread bridge would break Async-native producers (Async::Queue#dequeue etc.). Thread::Queue#pop / sleep / socket IO inside the enumerator are scheduler-aware and park only the stream task; CPU-bound or scheduler-opaque work must be moved by the user (a producer Thread feeding a Thread::Queue, or FiberBoundary.invoke inside the enumerator).

Defined Under Namespace

Classes: Break, CallbackDispatchScope, InlineCancellation, JoinTimeout

Constant Summary collapse

DISPATCH_KEY =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Fiber-storage key carrying the dispatching session's callback dispatch pair (scheduling mode + wrapper) across the SDK-MCP dispatch path (Query -> MCP::Server -> dynamic tool class). Fiber storage is per-fiber, so concurrent sessions with different modes sharing one SdkMcpServer instance cannot cross-contaminate — unlike mutating the shared server (last-writer-wins, persists past close) or a thread-local (the reactor thread is shared by many fibers).

:claude_agent_sdk_callback_dispatch

Class Method Summary collapse

Class Method Details

.invoke(timeout: nil, scheduling: :thread, wrapper: nil, &block) ⇒ Object

Run the given block on a plain thread when a Fiber scheduler is active. Returns the block's value. Exceptions propagate to the caller.

With timeout (seconds) the thread hop happens regardless of a scheduler being active, so the bound is enforced in plain synchronous code too; JoinTimeout is raised when it expires. Exception: with scheduling: :inline INSIDE an Async task, the block instead runs in place under a cooperative with_timeout — the deadline is delivered as InlineCancellation at the block's next suspension point (its ensure blocks run; not swallowable by rescue StandardError) and surfaces as the same JoinTimeout, so callers need no changes. When inline is requested but no Async task is present, the hard thread-hop bound applies — the timeout guarantee is never lost.

With scheduling: :inline (and no timeout) the block runs in place on the current fiber, scheduler or not. The caller opts in via ClaudeAgentOptions#callback_scheduling.

With wrapper (a callable, from ClaudeAgentOptions#callback_wrapper) the executed body becomes wrapper.call(block) — composed BEFORE the thread hop, so the wrapper runs on the same execution context as the callback: on the worker thread in :thread mode (the whole point — Rails.application.executor.wrap must run on the thread that touches ActiveRecord), in place on the reactor fiber in :inline mode. The wrapper must call its argument and return its value; exceptions from the callback propagate through it unchanged, and exceptions raised by the wrapper itself are treated exactly like callback exceptions. The wrapper must not swallow exceptions and must return the invocation's value — a user's break in a message block reaches the wrapper as a clean return (invoke_iteration translates it BEFORE the wrapper sees it), never as an exception a rescue/report wrapper could falsely flag. In inline/no-scheduler mode break unwinds natively through the wrapper's stack, so ensure-based wrappers (executor.wrap) are safe.

On timeout-bounded calls (the store-adapter paths: TranscriptMirrorBatcher, SessionResume) the wrapper composes INSIDE the bound: on the worker thread within the Thread#join deadline in :thread mode (so executor.wrap covers an AR-backed store adapter — connections check back in when the call ends; on a timeout the abandoned worker still runs the wrapper's ensure whenever the wedged call eventually finishes), and inside the cooperative with_timeout in inline mode (the cancellation passes through the wrapper un-swallowed — InlineCancellation is not a StandardError — and the wrapper's ensure runs at cancellation).

Raises:



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/claude_agent_sdk/fiber_boundary.rb', line 188

def invoke(timeout: nil, scheduling: :thread, wrapper: nil, &block)
  body = wrapper ? -> { wrapper.call(block) } : block
  return body.call if timeout.nil? && (scheduling == :inline || !Fiber.scheduler)

  # Cooperative timeout for inline-declared store adapters: in place on
  # the reactor fiber, cancelled at the next suspension point (delivery
  # and translation mechanics live in .with_cooperative_timeout). The
  # wrapper is already composed into `body`, so it sits inside the
  # timeout scope.
  if timeout && scheduling == :inline && (task = Async::Task.current?)
    expired = -> { JoinTimeout.new("timed out after #{timeout}s") }
    return with_cooperative_timeout(task, timeout, on_timeout: expired) { body.call }
  end

  thread = Thread.new(&body)
  thread.report_on_exception = false
  return thread.value if timeout.nil?
  raise JoinTimeout, "timed out after #{timeout}s" unless thread.join(timeout)

  thread.value
end

.invoke_iteration(block, *args, scheduling: :thread, wrapper: nil) ⇒ Object

Invoke a user-supplied iteration block across the boundary. The thread hop severs break from the surrounding loop, surfacing as LocalJumpError(reason: :break) on the worker thread; translate it into a Break sentinel so the SDK loop can break on the calling fiber. Returns Break when the user broke, nil when the block completed. Without a scheduler (or with scheduling: :inline) the block runs in place and break unwinds natively, never reaching the translation. The LocalJumpError -> Break translation lives INSIDE the invocation handed to the wrapper: a user's break is normal loop control, not an error, so a conforming rescue/report/re-raise wrapper must observe a clean return (the Break sentinel as the invocation's value), never a LocalJumpError it could falsely report or swallow. In inline / no-scheduler mode break unwinds natively through the wrapper's stack instead (ensure-based wrappers still run their cleanup).



254
255
256
257
258
259
260
261
262
263
264
265
 
# File 'lib/claude_agent_sdk/fiber_boundary.rb', line 254

def invoke_iteration(block, *args, scheduling: :thread, wrapper: nil)
  invocation = lambda do
    block.call(*args)
    nil
  rescue LocalJumpError => e
    raise unless e.reason == :break

    Break.new(e.exit_value)
  end
  body = wrapper ? -> { wrapper.call(invocation) } : invocation
  invoke(scheduling: scheduling) { body.call }
end

.with_cooperative_timeout(task, timeout, on_timeout:, &block) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Run block on task bounded by a cooperative timeout: the deadline is delivered INSIDE user code as an InlineCancellation at the block's next suspension point (ensure blocks run; a callback's ordinary rescue StandardError cannot swallow it), then translated into the exception built by on_timeout (a zero-arg callable) once control has returned from user code — each call site keeps its own outward contract (JoinTimeout on the store-adapter path in .invoke, Async::TimeoutError on the inline hook path in Query).

The cancellation class is a fresh per-invocation subclass: rescuing the shared base would also catch an OUTER timeout's cancellation delivered while this call is suspended (nested with_timeout — e.g. an inline store call nested inside a timed inline hook), mis-attributing the outer deadline to this call and letting execution continue past it. An outer cancellation is a different subclass, so it propagates through untouched.

Wrapper composition is the caller's choice — block runs verbatim inside the timeout scope (.invoke composes the callback wrapper into its body beforehand; the hook path composes it inside the block).



231
232
233
234
235
236
237
238
 
# File 'lib/claude_agent_sdk/fiber_boundary.rb', line 231

def with_cooperative_timeout(task, timeout, on_timeout:, &block)
  cancellation = Class.new(InlineCancellation)
  begin
    task.with_timeout(timeout, cancellation, &block)
  rescue cancellation
    raise on_timeout.call
  end
end