feat: Make resource dependencies more flexible (#295)

.changeset/tough-ghosts-try.md

@@ -0,0 +1,9 @@
+---
+"llama-index-workflows": minor
+---
+
+Add support for injecting resources more flexibly
+
+- Add support for injecting Resources recursively, so a Resource can depend on another Resource or ResourceConfig
+- Add support for injecting ResourceConfig directly into steps
+- Fix issues with resolving from String quoted types

packages/llama-index-workflows/src/workflows/decorators.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import inspect
 import sys
 from typing import (
     TYPE_CHECKING,
@@ -130,33 +131,36 @@ def step(
     """
 
     def decorator(func: Callable[P, R]) -> StepFunction[P, R]:
-        if not isinstance(num_workers, int) or num_workers <= 0:
-            raise WorkflowValidationError(
-                "num_workers must be an integer greater than 0"
-            )
-
-        func = make_step_function(func, num_workers, retry_policy)
-
-        # If this is a free function, call add_step() explicitly.
-        if is_free_function(func.__qualname__):
-            if workflow is None:
-                msg = f"To decorate {func.__name__} please pass a workflow class to the @step decorator."
-                raise WorkflowValidationError(msg)
-            workflow.add_step(func)
-
-        return func
+        localns = _capture_decorator_localns()
+        return _apply_step_decorator(
+            func,
+            num_workers=num_workers,
+            retry_policy=retry_policy,
+            workflow=workflow,
+            localns=localns,
+        )
 
     if func is not None:
         # The decorator was used without parentheses, like `@step`
-        return decorator(func)
+        localns = _capture_callsite_localns()
+        return _apply_step_decorator(
+            func,
+            num_workers=num_workers,
+            retry_policy=retry_policy,
+            workflow=workflow,
+            localns=localns,
+        )
     return decorator
 
 
 def make_step_function(
-    func: Callable[P, R], num_workers: int = 4, retry_policy: RetryPolicy | None = None
+    func: Callable[P, R],
+    num_workers: int = 4,
+    retry_policy: RetryPolicy | None = None,
+    localns: dict[str, Any] | None = None,
 ) -> StepFunction[P, R]:
     # This will raise providing a message with the specific validation failure
-    spec = inspect_signature(func)
+    spec = inspect_signature(func, localns=localns)
     validate_step_signature(spec)
 
     event_name, accepted_events = next(iter(spec.accepted_events.items()))
@@ -174,3 +178,60 @@ def make_step_function(
     )
 
     return casted
+
+
+def _apply_step_decorator(
+    func: Callable[P, R],
+    *,
+    num_workers: int,
+    retry_policy: RetryPolicy | None,
+    workflow: Type["Workflow"] | None,
+    localns: dict[str, Any] | None,
+) -> StepFunction[P, R]:
+    if not isinstance(num_workers, int) or num_workers <= 0:
+        raise WorkflowValidationError("num_workers must be an integer greater than 0")
+
+    func = make_step_function(
+        func, num_workers=num_workers, retry_policy=retry_policy, localns=localns
+    )
+
+    # If this is a free function, call add_step() explicitly.
+    if is_free_function(func.__qualname__):
+        if workflow is None:
+            msg = f"To decorate {func.__name__} please pass a workflow class to the @step decorator."
+            raise WorkflowValidationError(msg)
+        workflow.add_step(func)
+
+    return func
+
+
+def _capture_decorator_localns() -> dict[str, Any]:
+    frame = inspect.currentframe()
+    if frame is None or frame.f_back is None:
+        return {}
+
+    try:
+        decorator_frame = frame.f_back
+        localns: dict[str, Any] = {}
+        localns.update(decorator_frame.f_locals)
+        if decorator_frame.f_back is not None:
+            localns.update(decorator_frame.f_back.f_locals)
+        return localns
+    finally:
+        del frame
+
+
+def _capture_callsite_localns() -> dict[str, Any]:
+    frame = inspect.currentframe()
+    if frame is None or frame.f_back is None or frame.f_back.f_back is None:
+        return {}
+
+    try:
+        callsite_frame = frame.f_back.f_back
+        localns: dict[str, Any] = {}
+        localns.update(callsite_frame.f_locals)
+        if callsite_frame.f_back is not None:
+            localns.update(callsite_frame.f_back.f_locals)
+        return localns
+    finally:
+        del frame

packages/llama-index-workflows/src/workflows/representation/build.py

@@ -20,13 +20,33 @@ from workflows.representation.types import (
     WorkflowResourceNode,
     WorkflowStepNode,
 )
-from workflows.resource import ResourceDefinition
+from workflows.resource import (
+    ResourceDefinition,
+    ResourceDescriptor,
+    _Resource,
+    _ResourceConfig,
+)
 from workflows.utils import (
     get_steps_from_class,
     get_steps_from_instance,
 )
 
 
+def _get_resource_identity(resource: ResourceDescriptor) -> int:
+    """Get a unique identifier for resource deduplication.
+
+    For _Resource, uses the factory function identity.
+    For _ResourceConfig, uses (config_file, path_selector) hash.
+    """
+    if isinstance(resource, _Resource):
+        return id(resource._factory)
+    if isinstance(resource, _ResourceConfig):
+        # Use hash of config_file + path_selector for deduplication
+        hash_input = f"{resource.config_file}:{resource.path_selector or ''}"
+        return hash(hash_input)
+    return id(resource)
+
+
 def _get_event_type_chain(cls: type) -> list[str]:
     """Get the event type inheritance chain including the class itself.
 
@@ -49,7 +69,6 @@ def _create_resource_node(resource_def: ResourceDefinition) -> WorkflowResourceN
     rather than at Resource creation time for performance.
     """
     resource = resource_def.resource
-    factory = resource._factory
 
     # Get type name from annotation
     type_name: str | None = None
@@ -60,18 +79,25 @@ def _create_resource_node(resource_def: ResourceDefinition) -> WorkflowResourceN
         else:
             type_name = str(type_annotation)
 
-    # Extract source metadata lazily
+    # Extract source metadata lazily - only available for _Resource with factory
     source_file: str | None = None
     source_line: int | None = None
-    try:
-        source_file = inspect.getfile(factory)
-    except (TypeError, OSError):
-        pass
-    try:
-        _, source_line = inspect.getsourcelines(factory)
-    except (TypeError, OSError):
-        pass
-    resource_description = inspect.getdoc(factory)
+    resource_description: str | None = None
+
+    if isinstance(resource, _Resource):
+        factory = resource._factory
+        try:
+            source_file = inspect.getfile(factory)  # type: ignore[arg-type]
+        except (TypeError, OSError):
+            pass
+        try:
+            _, source_line = inspect.getsourcelines(factory)  # type: ignore[arg-type]
+        except (TypeError, OSError):
+            pass
+        resource_description = inspect.getdoc(factory)
+    elif isinstance(resource, _ResourceConfig):
+        # For ResourceConfig, the source is the config file
+        source_file = resource.config_file
 
     # Compute unique hash for deduplication
     hash_input = f"{resource.name}:{source_file or 'unknown'}"
@@ -189,13 +215,13 @@ def get_workflow_representation(workflow: Workflow) -> WorkflowGraph:
                 )
                 added_nodes.add("external_step")
 
-        # Add resource nodes (deduplicated by factory identity)
+        # Add resource nodes (deduplicated by resource identity)
         for resource_def in step_config.resources:
-            factory_id = id(resource_def.resource._factory)
-            if factory_id not in added_resource_nodes:
+            resource_id = _get_resource_identity(resource_def.resource)
+            if resource_id not in added_resource_nodes:
                 resource_node = _create_resource_node(resource_def)
                 nodes.append(resource_node)
-                added_resource_nodes[factory_id] = resource_node
+                added_resource_nodes[resource_id] = resource_node
 
     # Second pass: Add edges
     for step_name, step_func in steps.items():
@@ -238,8 +264,8 @@ def get_workflow_representation(workflow: Workflow) -> WorkflowGraph:
 
         # Edges from steps to resources (with variable name as label)
         for resource_def in step_config.resources:
-            factory_id = id(resource_def.resource._factory)
-            resource_node = added_resource_nodes[factory_id]
+            resource_id = _get_resource_identity(resource_def.resource)
+            resource_node = added_resource_nodes[resource_id]
             edges.append(
                 WorkflowGraphEdge(
                     source=step_name,

packages/llama-index-workflows/src/workflows/resource.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import functools
 import inspect
 import json
+from contextlib import contextmanager
 from pathlib import Path
 from typing import (
     Annotated,
@@ -13,12 +14,16 @@ from typing import (
     Awaitable,
     Callable,
     Generic,
+    Iterator,
     Optional,
+    Protocol,
     Type,
     TypeVar,
     cast,
     get_args,
     get_origin,
+    get_type_hints,
+    runtime_checkable,
 )
 
 from pydantic import (
@@ -30,6 +35,63 @@ T = TypeVar("T")
 B = TypeVar("B", bound=BaseModel)
 
 
+def _get_factory_type_hints(
+    factory: Callable[..., Any],
+    localns: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Resolve type hints for a factory function, avoiding shadowing.
+
+    Filters localns to exclude names that exist in factory's __globals__,
+    so types resolve from the factory's module while allowing closure variables.
+    """
+    filtered_localns = localns
+    if filtered_localns:
+        globalns = getattr(factory, "__globals__", {})
+        filtered_localns = {
+            k: v for k, v in filtered_localns.items() if k not in globalns
+        }
+
+    try:
+        return get_type_hints(factory, include_extras=True, localns=filtered_localns)
+    except NameError:
+        return {}
+
+
+@runtime_checkable
+class ResourceDescriptor(Protocol):
+    """Common interface for resource descriptors.
+
+    Both _Resource and _ResourceConfig implement this protocol, allowing
+    unified resolution through ResourceManager without isinstance checks.
+    """
+
+    @property
+    def name(self) -> str:
+        """Unique identifier for caching and cycle detection."""
+        ...
+
+    @property
+    def cache(self) -> bool:
+        """Whether to cache the resolved value."""
+        ...
+
+    async def resolve(self, manager: ResourceManager) -> Any:
+        """Resolve the resource, returning the concrete value."""
+        ...
+
+    def set_type_annotation(self, type_annotation: Any) -> None:
+        """Provide the annotated type for config-backed resources."""
+        ...
+
+    def set_localns(self, localns: dict[str, Any] | None) -> None:
+        """Store local namespace for resolving deferred type annotations."""
+        ...
+
+    def get_dependencies(self) -> list[tuple[str, ResourceDescriptor, type | None]]:
+        """Return factory dependencies. Empty for non-factory resources."""
+        ...
+
+
 class _Resource(Generic[T]):
     """Internal wrapper for resource factories.
 
@@ -42,34 +104,62 @@ class _Resource(Generic[T]):
         self._is_async = inspect.iscoroutinefunction(factory)
         self.name = getattr(factory, "__qualname__", type(factory).__name__)
         self.cache = cache
-        self.resource_configs: Optional[dict[str, BaseModel]] = None  # noqa: UP045
+        self._localns: dict[str, Any] | None = None
 
-    def prepare_resource_configs(self) -> None:
-        if self.resource_configs is None:
-            params = inspect.signature(self._factory).parameters
-            resource_configs: dict[str, BaseModel] = {}
-            if len(params) > 0:
-                for param in params.values():
-                    if get_origin(param.annotation) is Annotated:
-                        args = get_args(param.annotation)
-                        if len(args) == 2 and isinstance(args[1], _ResourceConfig):
-                            resource_config = args[1]
-                            resource_config.cls_factory = args[0]
-                            value = resource_config.call()
-                            resource_configs.update({param.name: value})
-            self.resource_configs = resource_configs
-        return None
+    async def _resolve_dependencies(
+        self, resource_manager: ResourceManager
+    ) -> dict[str, Any]:
+        """Resolve annotated ResourceDescriptor dependencies."""
+        resolved: dict[str, Any] = {}
 
-    async def call(self) -> T:
+        for param_name, descriptor, type_annotation in self.get_dependencies():
+            descriptor.set_type_annotation(type_annotation)
+            descriptor.set_localns(self._localns)
+            resolved[param_name] = await resource_manager.get(descriptor)
+
+        return resolved
+
+    async def call(self, resource_manager: ResourceManager) -> T:
         """Invoke the underlying factory, awaiting if necessary."""
-        self.prepare_resource_configs()
-        args = cast(dict[str, BaseModel], self.resource_configs)
+        args = await self._resolve_dependencies(resource_manager)
         if self._is_async:
             result = await cast(Callable[..., Awaitable[T]], self._factory)(**args)
         else:
             result = cast(Callable[..., T], self._factory)(**args)
         return result
 
+    async def resolve(self, manager: ResourceManager) -> T:
+        """Resolve the resource via the manager.
+
+        Implements ResourceDescriptor protocol.
+        """
+        return await self.call(manager)
+
+    def set_type_annotation(self, type_annotation: Any) -> None:
+        """No-op for factory-backed resources."""
+        return None
+
+    def set_localns(self, localns: dict[str, Any] | None) -> None:
+        """Store local namespace for resolving deferred type annotations."""
+        self._localns = localns
+
+    def get_dependencies(self) -> list[tuple[str, ResourceDescriptor, type | None]]:
+        """Extract ResourceDescriptor dependencies from factory signature."""
+        deps: list[tuple[str, ResourceDescriptor, type | None]] = []
+        params = inspect.signature(self._factory).parameters
+        type_hints = _get_factory_type_hints(self._factory, self._localns)
+
+        for param in params.values():
+            annotation = type_hints.get(param.name, param.annotation)
+            if get_origin(annotation) is Annotated:
+                args = get_args(annotation)
+                if len(args) >= 2:
+                    descriptor = args[1]
+                    if isinstance(descriptor, ResourceDescriptor):
+                        type_annotation = args[0]
+                        deps.append((param.name, descriptor, type_annotation))
+        return deps
+
 
 @functools.lru_cache(maxsize=1)
 def _get_resource_config_data(
@@ -99,19 +189,25 @@ class _ResourceConfig(Generic[B]):
     Internal wrapper for a pydantic-based resource whose configuration can be read from a JSON file.
     """
 
+    config_file: str
+    path_selector: str | None
+    cls_factory: Type[B] | None
+
     def __init__(
         self,
         config_file: str,
         path_selector: str | None,
         cls_factory: Type[B] | None = None,
     ) -> None:
-        if not Path(config_file).is_file():
+        config_path = Path(config_file)
+        if not config_path.is_file():
             raise FileNotFoundError(f"No such file: {config_file}")
-        if Path(config_file).suffix != ".json":
+        if config_path.suffix != ".json":
             raise ValueError(
                 "Only JSON files can be used to load Pydantic-based resources."
             )
-        self.config_file = config_file
+        # Store absolute path to ensure cache keys are unique across working directories
+        self.config_file = str(config_path.resolve())
         self.path_selector = path_selector
         self.cls_factory = cls_factory
 
@@ -121,7 +217,11 @@ class _ResourceConfig(Generic[B]):
             return self.config_file + "." + self.path_selector
         return self.config_file
 
-    # make async for compatibility with _Resource
+    @property
+    def cache(self) -> bool:
+        """ResourceConfig instances are always cached."""
+        return True
+
     def call(self) -> B:
         sel_data = _get_resource_config_data(
             config_file=self.config_file, path_selector=self.path_selector
@@ -134,6 +234,27 @@ class _ResourceConfig(Generic[B]):
                 "Class factory should be set to a BaseModel subclass before calling"
             )
 
+    async def resolve(self, manager: ResourceManager) -> B:
+        """Resolve the config resource.
+
+        Implements ResourceDescriptor protocol.
+        Note: cls_factory must be set before calling this method.
+        """
+        return self.call()
+
+    def set_type_annotation(self, type_annotation: Any) -> None:
+        """Assign the annotated class for config-backed resources when missing."""
+        if self.cls_factory is None:
+            self.cls_factory = cast(Type[B], type_annotation)
+
+    def set_localns(self, localns: dict[str, Any] | None) -> None:
+        """No-op for config-backed resources."""
+        pass
+
+    def get_dependencies(self) -> list[tuple[str, ResourceDescriptor, type | None]]:
+        """No dependencies for config-backed resources."""
+        return []
+
 
 def ResourceConfig(
     config_file: str,
@@ -159,13 +280,13 @@ class ResourceDefinition(BaseModel):
 
     Attributes:
         name (str): Parameter name in the step function.
-        resource (_Resource): Factory wrapper used by the manager to produce the dependency.
-        type_annotation (type | None): The type annotation from Annotated[T, Resource(...)].
+        resource (ResourceDescriptor): Descriptor used to produce the dependency.
+        type_annotation (type | None): The type annotation from Annotated[T, ...].
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
     name: str
-    resource: _Resource
+    resource: ResourceDescriptor
     type_annotation: Any = None
 
 
@@ -215,21 +336,58 @@ class ResourceManager:
 
     def __init__(self) -> None:
         self.resources: dict[str, Any] = {}
+        self._resolving: list[str] = []  # Track resources being resolved in order
+        self._resolution_cache: dict[str, Any] = {}
+        self._resolution_depth = 0
+
+    @contextmanager
+    def resolution_scope(self) -> Iterator[None]:
+        """Scope non-cached resolution values to a single dependency graph."""
+        self._resolution_depth += 1
+        try:
+            yield
+        finally:
+            self._resolution_depth -= 1
+            if self._resolution_depth == 0:
+                self._resolution_cache.clear()
 
     async def set(self, name: str, val: Any) -> None:
         """Register a resource instance under a name."""
         self.resources.update({name: val})
 
-    async def get(self, resource: _Resource) -> Any:
-        """Return a resource instance, honoring cache settings."""
-        if not resource.cache:
-            val = await resource.call()
-        elif resource.cache and not self.resources.get(resource.name, None):
-            val = await resource.call()
-            await self.set(resource.name, val)
-        else:
-            val = self.resources.get(resource.name)
-        return val
+    async def get(self, resource: ResourceDescriptor) -> Any:
+        if self._resolution_depth == 0:
+            with self.resolution_scope():
+                return await self._get(resource)
+        return await self._get(resource)
+
+    async def _get(self, resource: ResourceDescriptor) -> Any:
+        """Return a resource instance, honoring cache settings.
+
+        Works with any ResourceDescriptor implementation (_Resource or _ResourceConfig).
+        """
+        # Cycle detection
+        if resource.name in self._resolving:
+            chain = " -> ".join(self._resolving) + f" -> {resource.name}"
+            raise ValueError(f"Circular resource dependency detected: {chain}")
+
+        # Check cache first (before marking as resolving)
+        if resource.cache and resource.name in self.resources:
+            return self.resources[resource.name]
+        if resource.name in self._resolution_cache:
+            return self._resolution_cache[resource.name]
+
+        # Mark as resolving for cycle detection
+        self._resolving.append(resource.name)
+        try:
+            val = await resource.resolve(self)
+            if resource.cache:
+                await self.set(resource.name, val)
+            self._resolution_cache[resource.name] = val
+            return val
+        finally:
+            if resource.name in self._resolving:
+                self._resolving.remove(resource.name)
 
     def get_all(self) -> dict[str, Any]:
         """Return all materialized resources."""

packages/llama-index-workflows/src/workflows/runtime/control_loop.py

@@ -158,7 +158,7 @@ class _ControlLoopRunner:
                     )
                 )
             except Exception as e:
-                logger.error("error running step worker function: ", e, exc_info=True)
+                logger.error("error running step worker function: %s", e, exc_info=True)
                 self.queue_tick(
                     TickStepResult(
                         step_name=command.step_name,

packages/llama-index-workflows/src/workflows/runtime/types/step_function.py

@@ -52,11 +52,13 @@ async def partial(
     kwargs[step_config.event_name] = event
     if step_config.context_parameter:
         kwargs[step_config.context_parameter] = context
-    for resource in step_config.resources:
-        resource_value = await workflow._resource_manager.get(
-            resource=resource.resource
-        )
-        kwargs[resource.name] = resource_value
+    with workflow._resource_manager.resolution_scope():
+        for resource_def in step_config.resources:
+            descriptor = resource_def.resource
+            descriptor.set_type_annotation(resource_def.type_annotation)
+            # Unified resolution through ResourceManager
+            resource_value = await workflow._resource_manager.get(resource=descriptor)
+            kwargs[resource_def.name] = resource_value
     return functools.partial(func, **kwargs)
 
 

packages/llama-index-workflows/src/workflows/utils.py

@@ -36,7 +36,7 @@ from pydantic import BaseModel
 
 from .errors import WorkflowValidationError
 from .events import Event, EventType
-from .resource import ResourceDefinition
+from .resource import ResourceDefinition, ResourceDescriptor
 
 BUSY_WAIT_DELAY = 0.01
 
@@ -51,7 +51,9 @@ class StepSignatureSpec(BaseModel):
     resources: list[Any]
 
 
-def inspect_signature(fn: Callable) -> StepSignatureSpec:
+def inspect_signature(
+    fn: Callable, localns: dict[str, Any] | None = None
+) -> StepSignatureSpec:
     """
     Given a function, ensure the signature is compatible with a workflow step.
 
@@ -72,7 +74,7 @@ def inspect_signature(fn: Callable) -> StepSignatureSpec:
         raise TypeError(f"Expected a callable object, got {type(fn).__name__}")
 
     sig = inspect.signature(fn)
-    type_hints = get_type_hints(fn, include_extras=True)
+    type_hints = _resolve_type_hints(fn, include_extras=True, localns=localns)
 
     accepted_events: dict[str, list[EventType]] = {}
     context_parameter = None
@@ -104,11 +106,13 @@ def inspect_signature(fn: Callable) -> StepSignatureSpec:
         if get_origin(annotation) is Annotated:
             args = get_args(annotation)
             type_annotation = args[0] if args else None
-            resource = args[1] if len(args) > 1 else None
-            if resource is not None:
+            descriptor = args[1] if len(args) > 1 else None
+            if descriptor is not None and isinstance(descriptor, ResourceDescriptor):
+                # Pass localns to resource for nested annotation resolution
+                descriptor.set_localns(localns)
                 resources.append(
                     ResourceDefinition(
-                        name=name, resource=resource, type_annotation=type_annotation
+                        name=name, resource=descriptor, type_annotation=type_annotation
                     )
                 )
             continue
@@ -130,7 +134,7 @@ def inspect_signature(fn: Callable) -> StepSignatureSpec:
 
     return StepSignatureSpec(
         accepted_events=accepted_events,
-        return_types=_get_return_types(fn),
+        return_types=_get_return_types(fn, localns=localns),
         context_parameter=context_parameter,
         context_state_type=context_state_type,
         resources=resources,
@@ -230,13 +234,15 @@ def _get_param_types(param: inspect.Parameter, type_hints: dict) -> list[Any]:
     return [typ]
 
 
-def _get_return_types(func: Callable) -> list[Any]:
+def _get_return_types(
+    func: Callable, localns: dict[str, Any] | None = None
+) -> list[Any]:
     """
     Extract the return type hints from a function.
 
     Handles Union, Optional, and List types.
     """
-    type_hints = get_type_hints(func)
+    type_hints = _resolve_type_hints(func, localns=localns)
     return_hint = type_hints.get("return")
     if return_hint is None:
         return []
@@ -249,6 +255,28 @@ def _get_return_types(func: Callable) -> list[Any]:
         return [return_hint]
 
 
+def _resolve_type_hints(
+    func: Callable,
+    *,
+    include_extras: bool = False,
+    localns: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    try:
+        return get_type_hints(func, include_extras=include_extras, localns=localns)
+    except NameError as exc:
+        missing_name = getattr(exc, "name", None)
+        missing_msg = f" Missing name: {missing_name}." if missing_name else ""
+        func_name = getattr(func, "__qualname__", type(func).__name__)
+        msg = (
+            "Failed to resolve type annotations for "
+            f"{func_name}.{missing_msg} "
+            "If you are using 'from __future__ import annotations' or string "
+            "annotations, ensure referenced names are available in module scope "
+            "or in the scope where the @step decorator is applied."
+        )
+        raise WorkflowValidationError(msg) from exc
+
+
 def is_free_function(qualname: str) -> bool:
     """
     Determines whether a certain qualified name points to a free function.

packages/llama-index-workflows/tests/test_annotation_resolution.py

@@ -0,0 +1,183 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 LlamaIndex Inc.
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING, Annotated, cast
+
+import pytest
+from pydantic import BaseModel
+from workflows.decorators import step
+from workflows.errors import WorkflowValidationError
+from workflows.events import StartEvent, StopEvent
+from workflows.resource import Resource, ResourceConfig
+from workflows.workflow import Workflow
+
+if TYPE_CHECKING:
+
+    class MissingReturn:  # pragma: no cover
+        pass
+
+
+def test_step_decorator_resolves_local_resource_factory_with_future_annotations() -> (
+    None
+):
+    class Repo:
+        pass
+
+    def get_repo() -> Repo:
+        return Repo()
+
+    class LocalWorkflow(Workflow):
+        @step
+        async def start(
+            self,
+            ev: StartEvent,
+            repo: Annotated[Repo, Resource(get_repo)],
+        ) -> StopEvent:
+            return StopEvent(result="ok")
+
+    resources = LocalWorkflow.start._step_config.resources
+    assert len(resources) == 1
+    assert resources[0].name == "repo"
+    assert resources[0].type_annotation is Repo
+
+
+def test_step_decorator_resolves_local_return_type_with_future_annotations() -> None:
+    class ResultEvent(StopEvent):
+        pass
+
+    class LocalWorkflow(Workflow):
+        @step
+        async def start(self, ev: StartEvent) -> ResultEvent:
+            return ResultEvent()
+
+    return_types = LocalWorkflow.start._step_config.return_types
+    assert return_types == [ResultEvent]
+
+
+def test_step_decorator_error_message_for_unresolved_string_annotations() -> None:
+    with pytest.raises(
+        WorkflowValidationError,
+        match="Failed to resolve type annotations",
+    ):
+
+        class BadWorkflow(Workflow):
+            @step
+            async def start(self, ev: StartEvent) -> "MissingReturn":
+                return cast("MissingReturn", StopEvent(result="ok"))
+
+
+@pytest.mark.asyncio
+async def test_resource_config_in_factory_with_future_annotations(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """ResourceConfig in resource factories should resolve under future annotations."""
+    monkeypatch.chdir(tmp_path)
+
+    class SimpleConfig(BaseModel):
+        name: str
+
+    class SimpleClient:
+        def __init__(self, config: SimpleConfig) -> None:
+            self.config = config
+
+    config_path = tmp_path / "config.json"
+    config_path.write_text(json.dumps({"name": "demo"}))
+
+    def get_client(
+        config: Annotated[SimpleConfig, ResourceConfig(config_file=str(config_path))],
+    ) -> SimpleClient:
+        return SimpleClient(config=config)
+
+    class WorkflowWithConfig(Workflow):
+        @step
+        async def start_step(
+            self,
+            ev: StartEvent,
+            client: Annotated[SimpleClient, Resource(get_client)],
+        ) -> StopEvent:
+            assert client.config.name == "demo"
+            return StopEvent(result="done")
+
+    wf = WorkflowWithConfig(disable_validation=True)
+    await wf.run()
+
+
+@pytest.mark.asyncio
+async def test_nested_resource_in_resource_with_future_annotations() -> None:
+    """Nested Resource dependencies should resolve under future annotations."""
+
+    class DBConnection:
+        def __init__(self) -> None:
+            self.connected = True
+
+    class Repository:
+        def __init__(self, db: DBConnection) -> None:
+            self.db = db
+
+    def get_db() -> DBConnection:
+        return DBConnection()
+
+    def get_repo(
+        db: Annotated[DBConnection, Resource(get_db)],
+    ) -> Repository:
+        return Repository(db=db)
+
+    class WorkflowWithNestedResources(Workflow):
+        @step
+        async def start_step(
+            self,
+            ev: StartEvent,
+            repo: Annotated[Repository, Resource(get_repo)],
+        ) -> StopEvent:
+            assert repo.db.connected
+            return StopEvent(result="done")
+
+    wf = WorkflowWithNestedResources(disable_validation=True)
+    result = await wf.run()
+    assert result == "done"
+
+
+@pytest.mark.asyncio
+async def test_localns_does_not_shadow_factory_module_types(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Factory annotations should resolve from factory's module, not step's scope."""
+    monkeypatch.chdir(tmp_path)
+
+    # Import the factory helper from test_resources which has its own _FactoryConfig class
+    from tests.test_resources import _get_factory_with_config_path
+
+    # Define a LOCAL class with the same name as test_resources._FactoryConfig
+    # This should NOT shadow the factory's type when resolving annotations
+    class _FactoryConfig:
+        """Local shadow - factory should NOT use this."""
+
+        wrong_type = True
+
+    config_path = tmp_path / "config.json"
+    config_path.write_text('{"name": "test-value"}')
+
+    # Get a resource that uses the module-scoped _FactoryConfig from test_resources
+    # The factory is defined in test_resources module, so its annotations should
+    # resolve using test_resources' namespace, NOT this local namespace
+    resource = _get_factory_with_config_path(str(config_path))
+
+    class WorkflowTestingShadow(Workflow):
+        @step
+        async def start_step(
+            self,
+            ev: StartEvent,
+            result: Annotated[dict, resource],
+        ) -> StopEvent:
+            return StopEvent(result=result)
+
+    wf = WorkflowTestingShadow(disable_validation=True)
+    result = await wf.run()
+    # If the factory used the local _FactoryConfig (wrong), this would fail
+    assert result == {"name": "test-value"}

packages/llama-index-workflows/tests/test_resources.py

@@ -1,7 +1,10 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2026 LlamaIndex Inc.
 
+from __future__ import annotations
+
 import json
+import re
 from pathlib import Path
 from typing import Annotated, Optional
 from unittest import mock
@@ -25,6 +28,24 @@ cc1: int
 cc2: int
 
 
+# Test fixtures for annotation shadowing tests
+class _FactoryConfig(BaseModel):
+    """Config class defined in test_resources module scope."""
+
+    name: str
+
+
+def _get_factory_with_config_path(config_path: str) -> _Resource:
+    """Returns a Resource that creates a factory using module-scoped _FactoryConfig."""
+
+    def factory(
+        config: Annotated[_FactoryConfig, ResourceConfig(config_file=config_path)],
+    ) -> dict:
+        return {"name": config.name}
+
+    return Resource(factory)
+
+
 class SecondEvent(Event):
     msg: str = Field(description="A message")
 
@@ -91,7 +112,8 @@ async def test_function_resource_init() -> None:
     assert retval.cache
     assert not retval._is_async
 
-    result = await retval.call()
+    resource_manager = ResourceManager()
+    result = await retval.call(resource_manager)
     assert result == "string"
 
 
@@ -109,13 +131,15 @@ def test_resource_config_init(
     retval = ResourceConfig(config_file="config.json")
     assert isinstance(retval, _ResourceConfig)
     assert retval.path_selector is None
-    assert retval.config_file == "config.json"
+    # config_file is resolved to absolute path
+    expected_path = str(tmp_path / "config.json")
+    assert retval.config_file == expected_path
     assert retval.cls_factory is None
-    assert retval.name == "config.json"
+    assert retval.name == expected_path
 
     # modify path selector, modify name
     retval.path_selector = "hello.world"
-    assert retval.name == "config.json.hello.world"
+    assert retval.name == f"{expected_path}.hello.world"
 
     retval.path_selector = None
 
@@ -146,14 +170,15 @@ def test_resource_config_path_selector(
     with open("config.json", "w") as f:
         json.dump(data, f)
 
+    expected_path = str(tmp_path / "config.json")
     resource = ResourceConfig(config_file="config.json", path_selector="memory")
-    assert resource.name == "config.json.memory"
+    assert resource.name == f"{expected_path}.memory"
     resource.cls_factory = ChatMessages
     value = resource.call()
     assert isinstance(value, ChatMessages)
     assert value.messages == ["hello"]
     resource.path_selector = "core.fs"
-    assert resource.name == "config.json.core.fs"
+    assert resource.name == f"{expected_path}.core.fs"
     resource.cls_factory = Fs
     value = resource.call()
     assert isinstance(value, Fs)
@@ -178,7 +203,7 @@ def test_resource_config_path_selector_error(
     resource.cls_factory = Fs
     with pytest.raises(
         ValueError,
-        match=r"Expected dictionary for configuration from config.json at path core.fs.files, got: .*",
+        match=r"Expected dictionary for configuration from .+config\.json at path core\.fs\.files, got: .*",
     ):
         resource.call()
 
@@ -186,7 +211,7 @@ def test_resource_config_path_selector_error(
     resource.path_selector = "core.filesystem"
     with pytest.raises(
         ValueError,
-        match=r"Expected dictionary for configuration from config.json at path core.filesystem, got: .*",
+        match=r"Expected dictionary for configuration from .+config\.json at path core\.filesystem, got: .*",
     ):
         resource.call()
 
@@ -195,7 +220,7 @@ def test_resource_config_path_selector_error(
     resource.path_selector = "core.filesystem.fs"
     with pytest.raises(
         ValueError,
-        match=r"Expected dictionary for configuration from config.json at path core.filesystem, got: .*",
+        match=r"Expected dictionary for configuration from .+config\.json at path core\.filesystem, got: .*",
     ):
         resource.call()
 
@@ -442,3 +467,237 @@ async def test_resource_manager() -> None:
     m = ResourceManager()
     await m.set("test_resource", 42)
     assert m.get_all() == {"test_resource": 42}
+
+
+@pytest.mark.asyncio
+async def test_recursive_resource_injection() -> None:
+    """Test that a Resource can depend on another Resource."""
+
+    class DBConnection:
+        def __init__(self, host: str):
+            self.host = host
+
+    class Repository:
+        def __init__(self, db: DBConnection):
+            self.db = db
+
+    def get_db_connection() -> DBConnection:
+        return DBConnection(host="localhost")
+
+    def get_repository(
+        db: Annotated[DBConnection, Resource(get_db_connection)],
+    ) -> Repository:
+        return Repository(db)
+
+    class TestWorkflow(Workflow):
+        @step
+        def start_step(self, ev: StartEvent) -> SecondEvent:
+            return SecondEvent(msg="Hello")
+
+        @step
+        def use_repo(
+            self,
+            ev: SecondEvent,
+            repo: Annotated[Repository, Resource(get_repository)],
+        ) -> StopEvent:
+            assert repo.db.host == "localhost"
+            return StopEvent()
+
+    wf = TestWorkflow(disable_validation=True)
+    await wf.run()
+
+
+@pytest.mark.asyncio
+async def test_recursive_resource_caching() -> None:
+    """Test that nested resources respect individual cache settings."""
+    call_counts = {"db": 0, "repo": 0}
+
+    class DBConnection:
+        pass
+
+    class Repository:
+        def __init__(self, db: DBConnection):
+            self.db = db
+
+    def get_db_connection() -> DBConnection:
+        call_counts["db"] += 1
+        return DBConnection()
+
+    def get_repository(
+        db: Annotated[DBConnection, Resource(get_db_connection)],
+    ) -> Repository:
+        call_counts["repo"] += 1
+        return Repository(db)
+
+    class StepEvent(Event):
+        pass
+
+    class TestWorkflow(Workflow):
+        @step
+        def step1(
+            self,
+            ev: StartEvent,
+            repo: Annotated[Repository, Resource(get_repository)],
+        ) -> StepEvent:
+            return StepEvent()
+
+        @step
+        def step2(
+            self,
+            ev: StepEvent,
+            repo: Annotated[Repository, Resource(get_repository)],
+        ) -> StopEvent:
+            return StopEvent()
+
+    wf = TestWorkflow(disable_validation=True)
+    await wf.run()
+
+    # Both should be cached (called once each)
+    assert call_counts["db"] == 1
+    assert call_counts["repo"] == 1
+
+
+@pytest.mark.asyncio
+async def test_circular_resource_dependency_detection() -> None:
+    """Test that circular dependencies are detected at runtime."""
+
+    class A:
+        pass
+
+    class B:
+        pass
+
+    # Create the cycle by modifying __annotations__ after creating the resources
+    # This allows us to create mutual dependencies
+
+    def cyclic_factory_a(b: Annotated[B, "placeholder"]) -> A:  # type: ignore
+        return A()
+
+    def cyclic_factory_b(a: Annotated[A, "placeholder"]) -> B:  # type: ignore
+        return B()
+
+    # Create resources
+    cyclic_res_a = Resource(cyclic_factory_a)
+    cyclic_res_b = Resource(cyclic_factory_b)
+
+    # Modify annotations to create the cycle:
+    # cyclic_res_a depends on cyclic_res_b, and cyclic_res_b depends on cyclic_res_a
+    cyclic_factory_a.__annotations__["b"] = Annotated[B, cyclic_res_b]
+    cyclic_factory_b.__annotations__["a"] = Annotated[A, cyclic_res_a]
+
+    class TestWorkflow(Workflow):
+        @step
+        def start_step(
+            self,
+            ev: StartEvent,
+            a: Annotated[A, cyclic_res_a],
+        ) -> StopEvent:
+            return StopEvent()
+
+    wf = TestWorkflow(disable_validation=True)
+    expected_chain = (
+        f"{cyclic_factory_a.__qualname__} -> "
+        f"{cyclic_factory_b.__qualname__} -> "
+        f"{cyclic_factory_a.__qualname__}"
+    )
+    with pytest.raises(ValueError, match=re.escape(expected_chain)):
+        await wf.run()
+
+
+@pytest.mark.asyncio
+async def test_non_cached_resource_single_resolution_cycle() -> None:
+    """Non-cached resources should resolve once per dependency graph."""
+    call_counts = {"d": 0}
+
+    class D:
+        pass
+
+    def get_d() -> D:
+        call_counts["d"] += 1
+        return D()
+
+    def get_b(d: Annotated[D, Resource(get_d, cache=False)]) -> str:
+        return "b"
+
+    def get_c(d: Annotated[D, Resource(get_d, cache=False)]) -> str:
+        return "c"
+
+    class TestWorkflow(Workflow):
+        @step
+        def start_step(
+            self,
+            ev: StartEvent,
+            b: Annotated[str, Resource(get_b)],
+            c: Annotated[str, Resource(get_c)],
+        ) -> StopEvent:
+            assert b == "b"
+            assert c == "c"
+            return StopEvent()
+
+    wf = TestWorkflow(disable_validation=True)
+    await wf.run()
+    assert call_counts["d"] == 1
+
+
+@pytest.mark.asyncio
+async def test_resource_config_in_step_signature(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test that ResourceConfig can be used directly in step signatures."""
+    monkeypatch.chdir(tmp_path)
+
+    data = {"file": "test.txt", "permission_mode": "r"}
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    class TestWorkflow(Workflow):
+        @step
+        def start_step(self, ev: StartEvent) -> SecondEvent:
+            return SecondEvent(msg="Hello")
+
+        @step
+        def use_config(
+            self,
+            ev: SecondEvent,
+            config: Annotated[FileData, ResourceConfig(config_file="config.json")],
+        ) -> StopEvent:
+            assert config.file == "test.txt"
+            assert config.permission_mode == "r"
+            return StopEvent()
+
+    wf = TestWorkflow(disable_validation=True)
+    await wf.run()
+
+
+@pytest.mark.asyncio
+async def test_resource_config_in_step_with_path_selector(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test ResourceConfig with path_selector in step signatures."""
+    monkeypatch.chdir(tmp_path)
+
+    data = {
+        "database": {"file": "db.sqlite", "permission_mode": "rw"},
+        "cache": {"file": "cache.json", "permission_mode": "r"},
+    }
+    with open("config.json", "w") as f:
+        json.dump(data, f)
+
+    class TestWorkflow(Workflow):
+        @step
+        def use_db_config(
+            self,
+            ev: StartEvent,
+            db_config: Annotated[
+                FileData,
+                ResourceConfig(config_file="config.json", path_selector="database"),
+            ],
+        ) -> StopEvent:
+            assert db_config.file == "db.sqlite"
+            assert db_config.permission_mode == "rw"
+            return StopEvent()
+
+    wf = TestWorkflow(disable_validation=True)
+    await wf.run()