def tracked_method_wrapper(
        self,
        query: serial_utils.Lens,
        func: Callable,
        method_name: str,
        cls: type,
        obj: object,
        span_type: Optional[SpanAttributes.SpanType] = None,
        attributes: Optional[Attributes] = None,
        must_be_first_wrapper: bool = False,
    ):
        """Wrap a method to capture its inputs/outputs/errors."""

        if self.app is None:
            raise ValueError("Instrumentation requires an app but is None.")

        if self.app.session.experimental_feature(
            core_experimental.Feature.OTEL_TRACING, freeze=True
        ):
            from trulens.core.otel.instrument import instrument

            if span_type is None:
                span_type = SpanAttributes.SpanType.UNKNOWN
            wrapper = instrument(
                span_type=span_type,
                attributes=attributes,
                must_be_first_wrapper=must_be_first_wrapper,
            )
            # return wrapper(func)?
            of_cls_method = getattr(cls, method_name)
            return wrapper(of_cls_method)

        if python_utils.safe_hasattr(func, "__func__"):
            raise ValueError("Function expected but method received.")

        if python_utils.safe_hasattr(func, Instrument.INSTRUMENT):
            logger.debug("\t\t\t%s: %s is already instrumented", query, func)

            # Notify the app instrumenting this method where it is located. Note
            # we store the method being instrumented in the attribute
            # Instrument.INSTRUMENT of the wrapped variant.
            original_func = getattr(func, Instrument.INSTRUMENT)

            self.app.on_method_instrumented(obj, original_func, path=query)

            # Add self.app, the app requesting this method to be
            # instrumented, to the list of apps expecting to be notified of
            # calls.
            existing_apps = getattr(func, Instrument.APPS)
            # The __repr__.__self__ undoes weakref.proxy .
            existing_apps.add(self.app.__repr__.__self__)  # weakref set

            return func

        # Notify the app instrumenting this method where it is located:
        self.app.on_method_instrumented(obj, func, path=query)

        logger.debug("\t\t\t%s: instrumenting %s=%s", query, method_name, func)

        sig = python_utils.safe_signature(func)

        @functools.wraps(func)
        def tru_wrapper(*args, **kwargs):
            # NOTE(piotrm): don't capture any apps in the closure of this
            # method. This method will override various instrumented functions
            # in various class definitions and will never get deallocated. If
            # there are apps in this closure, those apps will also never get
            # deallocated.

            logger.debug(
                "%s: calling instrumented sync method %s of type %s, "
                "iscoroutinefunction=%s, "
                "isasyncgeneratorfunction=%s",
                query,
                func,
                type(func),
                python_utils.is_really_coroutinefunction(func),
                inspect.isasyncgenfunction(func),
            )

            if do_not_track.get():
                return func(*args, **kwargs)

            apps = getattr(tru_wrapper, Instrument.APPS)  # weakref

            if len(apps) > 0 and not Instrument._have_context():
                return func(*args, **kwargs)

            # If not within a root method, call the wrapped function without
            # any recording.

            # Get any contexts already known from higher in the call stack.
            contexts = set(
                WithInstrumentCallbacks._context_contexts.get()
            )  # copy

            # And add any new contexts from all apps wishing to record this
            # function. This may produce some of the same contexts that were
            # already being tracked which is ok. Importantly, this might produce
            # contexts for apps that didn't instrument a method higher in the
            # call stack hence this might be the first time they are seeing an
            # instrumented method being called.
            for app in apps:
                for ctx in app.on_new_record(func):
                    contexts.add(ctx)

            if len(contexts) == 0:
                # If no app wants this call recorded, run and return without
                # instrumentation.
                logger.debug(
                    "%s: no record found or requested, not recording.", query
                )

                return func(*args, **kwargs)

            else:
                pass

            context_token = WithInstrumentCallbacks._context_contexts.set(
                contexts
            )

            # If a wrapped method was called in this call stack, get the prior
            # calls from this variable. Otherwise create a new chain stack. As
            # another wrinkle, the addresses of methods in the stack may vary
            # from app to app that are watching this method. Hence we index the
            # stacks by id of the call record list which is unique to each app.
            stacks = dict(WithInstrumentCallbacks._stack_contexts.get())  # copy

            # My own stacks to be looked up by further subcalls by the logic
            # right above. We make a copy here since we need subcalls to access
            # it but we don't want them to modify it.
            stacks_token = WithInstrumentCallbacks._stack_contexts.set(stacks)

            error = None
            rets = None
            start_time = None
            bindings = None

            # Prepare stacks with call information of this wrapped method so
            # subsequent (inner) calls will see it. For every root_method in the
            # call stack, we make a call record to add to the existing list
            # found in the stack. Path stored in `query` of this method may
            # differ between apps that use it so we have to create a separate
            # frame identifier for each, and therefore the stack. We also need
            # to use a different stack for the same reason. We index the stack
            # in `stacks` via id of the (unique) list `record`.

            # First prepare the stacks for each context.
            for ctx in contexts:
                # Get app that has instrumented this method.
                app = ctx.app

                # The path to this method according to the app.
                path = app.get_method_path(
                    args[0], func
                )  # hopefully args[0] is self, owner of func

                if path is None:
                    logger.warning(
                        "App of type %s no longer knows about object %s method %s. "
                        "Something might be going wrong.",
                        python_utils.class_name(type(app)),
                        python_utils.id_str(args[0]),
                        python_utils.callable_name(func),
                    )
                    continue

                if ctx not in stacks:
                    # If we are the first instrumented method in the chain
                    # stack, make a new stack tuple for subsequent deeper calls
                    # (if any) to look up.
                    stack = ()
                else:
                    stack = stacks[ctx]

                frame_ident = record_schema.RecordAppCallMethod(
                    path=path,
                    method=pyschema_utils.Method.of_method(
                        func, obj=args[0], cls=cls
                    ),  # important: don't use obj here as that would capture obj in closure of this wrapper
                )

                stack = stack + (frame_ident,)

                stacks[ctx] = stack  # for deeper calls to get

            # Now we will call the wrapped method. We only do so once.

            # Start of run wrapped block.
            start_time = datetime.now()

            # Create a unique call_id for this method call. This will be the
            # same across everyone Record or RecordAppCall that refers to this
            # method invocation.
            call_id = types_schema.new_call_id()

            error_str = None

            tally = None

            # Capture the current values of the relevant context vars before we
            # reset them.
            context_vars = {
                WithInstrumentCallbacks._context_contexts: WithInstrumentCallbacks._context_contexts.get(),
                WithInstrumentCallbacks._stack_contexts: WithInstrumentCallbacks._stack_contexts.get(),
            }

            try:
                # Using sig bind here so we can produce a list of key-value
                # pairs even if positional arguments were provided.
                bindings: BoundArguments = sig.bind(*args, **kwargs)

                logger.info(f"calling {func} with {args}")

                rets, tally = core_endpoint.Endpoint.track_all_costs_tally(
                    func, *args, **kwargs
                )

            except BaseException as e:
                error = e
                error_str = str(e)

                logger.error(
                    "Error calling wrapped function %s.",
                    python_utils.callable_name(func),
                )
                logger.error(traceback.format_exc())

            # Done running the wrapped function. Lets collect the results.
            # Create common information across all records.

            # Don't include self in the recorded arguments.
            nonself = {
                k: json_utils.jsonify(v)
                for k, v in (
                    bindings.arguments.items() if bindings is not None else {}
                )
                if k
                not in ["self", "_self"]  # llama_index uses "_self" sometimes
            }

            records = {}

            if app is None:
                raise ValueError("Instrumentation requires an app but is None.")

            # We implicitly assume that all apps that have instrumented this
            # method are of the same type. We use some classmethods of that
            # app type below.

            def update_call_info(rets, final=True):
                """Notify the app of the call and let it update the record if
                needed.

                `final` is passed to the new record handler in the app and tells
                it whether to the record is ready to run feedback.
                """

                # (re) generate end_time here because cases where the
                # initial end_time was just to produce an awaitable/lazy
                # before being awaited. If a stream is used, this will
                # update the end_time to the last chunk.
                end_time = datetime.now()

                record_app_args = dict(
                    call_id=call_id,
                    args=nonself,
                    perf=base_schema.Perf(
                        start_time=start_time, end_time=end_time
                    ),
                    pid=os.getpid(),
                    tid=th.get_native_id(),
                    rets=json_utils.jsonify(rets),
                    error=error_str if error is not None else None,
                )
                # End of run wrapped block.

                # Now record calls to each context.
                for ctx in contexts:
                    stack = stacks[ctx]

                    # Note that only the stack differs between each of the records in this loop.
                    record_app_args["stack"] = stack
                    call = record_schema.RecordAppCall(**record_app_args)
                    ctx.add_call(call)

                    # If stack has only 1 thing on it, we are looking at a "root
                    # call". Create a record of the result and notify the app:

                    existing_record = records.get(ctx, None)

                    if tally is None:
                        cost = base_schema.Cost()
                    else:
                        cost = tally()  # get updated cost

                    if len(stack) == 1 or existing_record is not None:
                        # If this is a root call, notify app to add the completed record
                        # into its containers:

                        records[ctx] = ctx.app.on_add_record(
                            ctx=ctx,
                            func=func,
                            sig=sig,
                            bindings=bindings,
                            ret=rets,
                            error=error,
                            perf=base_schema.Perf(
                                start_time=start_time, end_time=end_time
                            ),
                            cost=cost,
                            existing_record=existing_record,
                            final=final,
                        )

                if error is not None:
                    raise error

                return rets

            def rewrap(rets):
                """Wrap any lazy return values with handlers to update
                call/record when ready."""

                if python_utils.is_lazy(rets):
                    type_name = python_utils.class_name(type(rets))

                    logger.debug(
                        "This app produced a lazy response of type `%s`."
                        "This record will be updated once the response is available.",
                        type_name,
                    )

                    # Placeholder:
                    temp_rets = f"""
    The method {python_utils.callable_name(func)} produced a lazy response of type
    `{type_name}`. This record will be updated once the response is available. If
    this message persists, check that you are using the correct version of the app
    method and await and/or iterate over any results it produces.
    """

                    # Will add placeholder to the record:
                    update_call_info(temp_rets, final=False)

                    return python_utils.wrap_lazy(
                        rets,
                        wrap=None,
                        on_done=rewrap,
                        context_vars=context_vars,
                    )

                # Handle app-specific lazy values (like llama_index StreamResponse):
                # NOTE(piotrm): self.app is a weakref

                if python_utils.WRAP_LAZY:
                    for ctx in contexts:
                        rets = ctx.app.wrap_lazy_values(
                            rets,
                            wrap=None,
                            on_done=update_call_info,
                            context_vars=context_vars,
                        )
                else:
                    update_call_info(rets, final=True)

                return rets

            # HACK: disable these to debug context issues. See
            # App._set_context_vars .
            WithInstrumentCallbacks._stack_contexts.reset(stacks_token)
            WithInstrumentCallbacks._context_contexts.reset(context_token)

            return rewrap(rets)

        # Create a new set of apps expecting to be notified about calls to the
        # instrumented method. Making this a weakref set so that if the
        # recorder/app gets garbage collected, it will be evicted from this set.

        # NOTE(piotrm): __repr__.__self__ undoes weakref.proxy .
        apps = WeakSet([self.app.__repr__.__self__])

        # Indicate that the wrapper is an instrumented method so that we dont
        # further instrument it in another layer accidentally.
        setattr(tru_wrapper, Instrument.INSTRUMENT, func)
        setattr(tru_wrapper, Instrument.APPS, apps)

        return tru_wrapper
Wrap a method to capture its inputs/outputs/errors.
src/core/trulens/core/instruments.py