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 SCHEDULING_KEY / SchedulingScope 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, InlineCancellation, JoinTimeout, SchedulingScope

Constant Summary

SCHEDULING_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 scheduling mode 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_scheduling

Class Method Summary

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

  Run the given block on a plain thread when a Fiber scheduler is active.

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

  Invoke a user-supplied iteration block across the boundary.

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.

The timeout path deliberately ignores the wrapper: it belongs to the store-adapter carve-out (TranscriptMirrorBatcher, SessionResume), which never carries user callbacks.

Raises:

• (JoinTimeout)

# File 'lib/claude_agent_sdk/fiber_boundary.rb', line 181

def invoke(timeout: nil, scheduling: :thread, wrapper: nil, &block)
  body = wrapper && timeout.nil? ? -> { 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. The
  # wrapper stays ignored on timeout paths (store adapters are never
  # wrapped), so `body` is the bare block here. 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. a store call 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.
  if timeout && scheduling == :inline && (task = Async::Task.current?)
    cancellation = Class.new(InlineCancellation)
    begin
      return task.with_timeout(timeout, cancellation, &block)
    rescue cancellation
      raise JoinTimeout, "timed out after #{timeout}s"
    end
  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).

# File 'lib/claude_agent_sdk/fiber_boundary.rb', line 226

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