temporalio.contrib.langsmith._interceptor._ReplaySafeRunTree

class documentation

class _ReplaySafeRunTree(RunTree): (source)

Known subclasses: temporalio.contrib.langsmith._interceptor._RootReplaySafeRunTreeFactory

Constructor: _ReplaySafeRunTree(run_tree, executor)

View In Hierarchy

Wrapper around a RunTree with replay-safe post, end, and patch.

Inherits from RunTree so isinstance checks pass, but does not call super().__init__()—the wrapped _run is the real RunTree. Attribute access is delegated via __getattr__/__setattr__.

During replay, post(), end(), and patch() become no-ops (I/O suppression), but create_child() still runs to maintain parent-child linkage so @traceable's _setup_run can build the run tree across the replay boundary. In workflow context, post() and patch() submit to a single-worker ThreadPoolExecutor for FIFO ordering, avoiding blocking on the workflow task thread.

Method __getattr__ Delegate attribute access to the wrapped RunTree.
Method __init__ Wrap an existing RunTree with replay-safe overrides.
Method __setattr__ Delegate attribute setting to the wrapped RunTree.
Method create_child Create a child run, returning another _ReplaySafeRunTree.
Method end End the run, skipping during replay.
Method patch Patch the run to LangSmith, skipping during replay.
Method post Post the run to LangSmith, skipping during replay.
Method to_headers Delegate to the wrapped RunTree's to_headers.
Method _inject_deterministic_ids Inject deterministic run_id and start_time in workflow context.
Method _submit Submit work to the background executor.
def __getattr__(self, name: str) -> Any: (source)

Delegate attribute access to the wrapped RunTree.

def __init__(self, run_tree: RunTree, *, executor: ThreadPoolExecutor): (source)
overridden in temporalio.contrib.langsmith._interceptor._RootReplaySafeRunTreeFactory

Wrap an existing RunTree with replay-safe overrides.

def __setattr__(self, name: str, value: Any): (source)

Delegate attribute setting to the wrapped RunTree.

def create_child(self, *args: Any, **kwargs: Any) -> _ReplaySafeRunTree: (source)
overridden in temporalio.contrib.langsmith._interceptor._RootReplaySafeRunTreeFactory

Create a child run, returning another _ReplaySafeRunTree.

In workflow context, injects deterministic run_id and start_time unless they are passed in manually via kwargs.

def end(self, **kwargs: Any): (source)
overridden in temporalio.contrib.langsmith._interceptor._RootReplaySafeRunTreeFactory

End the run, skipping during replay.

Pre-computes end_time via workflow.now() in workflow context so RunTree.end() doesn't call datetime.now() (non-deterministic and sandbox-restricted).

def patch(self, *, exclude_inputs: bool = False): (source)
overridden in temporalio.contrib.langsmith._interceptor._RootReplaySafeRunTreeFactory

Patch the run to LangSmith, skipping during replay.

def post(self, exclude_child_runs: bool = True): (source)
overridden in temporalio.contrib.langsmith._interceptor._RootReplaySafeRunTreeFactory

Post the run to LangSmith, skipping during replay.

def to_headers(self) -> dict[str, str]: (source)

Delegate to the wrapped RunTree's to_headers.

def _inject_deterministic_ids(self, kwargs: dict[str, Any]): (source)

Inject deterministic run_id and start_time in workflow context.

def _submit(self, fn: Callable[..., object], *args: Any, **kwargs: Any): (source)

Submit work to the background executor.