Dependency Injection

The AmritaSense workflow engine integrates the dependency injection (DI) system, providing powerful dependency resolution and injection capabilities for workflow nodes. This integration allows node functions to declare their dependencies, and the engine will automatically resolve and inject those dependencies at execution time.

Overview: node and event DI mechanism

In AmritaSense, every workflow node is essentially a callable function. Through dependency injection, these functions can declare the dependencies they require, including:

• Workflow interpreter instance: obtained via POINTER_DEPENDS
• Address computation tools: obtained via ADDR, NEAR_OFFSET, FAR_OFFSET for dynamic node address calculation
• Custom dependency providers: any function that returns the required type can act as a dependency provider

The dependency injection system resolves dependencies before node execution and ensures that all declared dependencies are provided. If dependency resolution fails, the workflow throws an exception and terminates.

Basic usage: Depends() declaration

Dependency injection is implemented through Depends(). Depends() accepts a dependency provider function and returns a dependency factory that is called at node execution time to obtain the actual dependency value.

Syntax

from amrita_sense.hook.matcher import Depends
from amrita_sense.runtime.deps import POINTER_DEPENDS, ADDR, NEAR_OFFSET


@Node()
def my_node(dependency_value: ReturnType = Depends(dependency_provider_function)):
    # use dependency_value
    pass

Depends may also be written inside the annotation with Annotated — see Annotated declarations.

Built-in dependency tools

AmritaSense provides several built-in dependency helpers in the amrita_sense.runtime.deps module:

• POINTER_DEPENDS: injects the current WorkflowInterpreter instance
• ADDR(alias): injects the absolute address (PointerVector) of the specified alias node
• NEAR_OFFSET(alias): injects the near offset (int) of the specified alias node
• FAR_OFFSET(alias): injects the far offset (PointerVector) of the specified alias node

Example usage

from amrita_sense.runtime.deps import POINTER_DEPENDS, ADDR, NEAR_OFFSET
from amrita_sense.runtime.workflow import WorkflowInterpreter
from amrita_sense.types import PointerVector


@Node()
def navigation_node(
    pc: WorkflowInterpreter = Depends(POINTER_DEPENDS),
    target_addr: PointerVector = Depends(ADDR("my_target")),
    offset: int = Depends(NEAR_OFFSET("my_target")),
):
    # Use the interpreter to jump
    pc.jump_to(target_addr)
    # or use a relative offset for near jumps
    pc.jump_offset(offset)

Concurrent resolution and runtime injection

AmritaSense’s dependency injection system supports concurrent resolution and runtime injection, which means:

1. Concurrency-safe: dependency resolution is thread-safe and can be used safely in concurrent environments.
2. Runtime dynamism: dependency values are computed at node execution time, not at workflow compile time.
3. Context awareness: dependency provider functions can access the current workflow context.

The dependency injection system automatically handles both synchronous and asynchronous dependency providers. If a provider is asynchronous, the system awaits it; if it is synchronous, it calls it directly.

Asynchronous dependency example

async def async_dependency():
    await asyncio.sleep(0.1)
    return "async_result"


@Node()
def async_node(result: str = Depends(async_dependency)):
    print(f"Received: {result}")

Event and hook integration

AmritaSense uses the same dependency matcher for workflow nodes and hook/event handlers. That means event listener callbacks can also declare Depends(...) dependencies, and the runtime will resolve them before invoking the callback. This makes it possible to share the same dependency provider functions across normal nodes and external hooks.

from amrita_sense.hook.matcher import Depends


async def on_event(event: Any, pc: WorkflowInterpreter = Depends(POINTER_DEPENDS)):
    # Event handlers can also receive runtime context via Depends
    pass

The event/hook system resolves dependencies through the same MatcherFactory machinery used by node execution, so the behavior is consistent across the engine.

Important behavior: returning None terminates the workflow

The dependency injection system has an important behavior: if a dependency provider function returns None, the workflow terminates immediately.

This design decision is based on:

1. Clear failure semantics: None is treated as a clear signal that dependency resolution failed.
2. Avoid null propagation: preventing None values from spreading through the workflow reduces debugging complexity.
3. Fail-fast principle: if a dependency cannot be satisfied, fail immediately rather than continuing with potentially invalid logic.

Handling optional dependencies

If a dependency can legitimately be absent, use a pattern like:

def optional_dependency():
    if some_condition:
        return "value"
    else:
        return OptionalValue(None)


class OptionalValue:
    def __init__(self, value):
        self.value = value

Or handle the conditional inside the node function instead of at the injection layer:

def get_maybe_value():
    if some_condition:
        return "value"
    return "default_value"


@Node()
def safe_node(value: str = Depends(get_maybe_value)):
    pass

Error handling

If a dependency provider returns None, resolving that node fails and the workflow raises a DependsInjectFailed (a DependsException). That is an ordinary exception, so a Try/CATCH block can handle it:

def failing_dependency() -> str | None:
    return None


@Node()
def needs_dependency(value: str = Depends(failing_dependency)) -> None:
    print("This won't execute")


@Node()
def on_failure(exc_val: DependsInjectFailed) -> None:
    print(f"Caught dependency failure: {exc_val}")


Try(needs_dependency).CATCH(DependsInjectFailed, on_failure)

CATCH matches on the exception type, so an exception that no handler claims keeps propagating — a CATCH(ValueError, ...) will not swallow a KeyError. The handler can also ask for the exception itself: exc_type, exc_val and exc_tb are injected like any other dependency.

This design ensures that dependency injection remains robust and predictable while giving developers a clear error handling mechanism.

DI Result Cache

The WorkflowInterpreter maintains an internal DI result cache (_di_cache) to avoid redundant dependency resolution when the same node is executed multiple times with the same argument types.

How it works

The cache key is built from the node function identity plus the argument fingerprint (the pointer position is not part of the key — the same node function shares cache entries across call sites):

• Function identity: id(node.func) — the node's underlying function object
• Args fingerprint: computed from the types of _ava_args and _ava_kwargs

The utility function _fingerprint_args() generates the args fingerprint by:

1. Extracting type(arg).__name__ for each positional argument
2. Extracting (key, type(v).__name__) for each keyword argument (sorted for stability)
3. Hashing the combined tuple

# Simplified illustration of the cache key
code = _fingerprint_args(
    ava_args, ava_kwargs
)  # or the cached args_hash when no extra args
cache_key = hash((id(node.func), code))

The cache payload is an LRUCache (from cachetools) with a maximum of 2048 entries. When the cache is full, the least recently used entry is evicted. Each entry stores (static_kwargs, non_cacheable_factories) — see cacheable below.

cacheable / use_cache factories

DependsFactory(cacheable=True) — or, with the preferred spelling, Depends(f, use_cache=True) — providers are resolved once, at cache-write time, and their results are stored in the cache. cacheable=False (the default) providers are stored as-is and re-resolved on every call — use this for providers with side effects or time-varying values. This separation is orthogonal to cache validity (hash_trustable).

For generator providers, sharing is governed by scope instead of use_cache; see Generator dependencies and lifecycle.

Cache lifecycle

• Initialization: The cache is created during WorkflowInterpreter.__init__() with an initial args fingerprint.
• Lookup: Before resolving dependencies for a node, the interpreter checks _di_cache.payload for a matching key. On a cache hit, the cached static kwargs are used directly, skipping static dependency resolution.
• Per-call factories: cacheable=False factories stored in the cache entry are re-resolved on every invocation.
• Invalidation: Modifying _ava_args or _ava_kwargs sets hash_trustable = False (a cache-validity gate), indicating the args fingerprint may be stale. Call rehash_args() to recompute the fingerprint and restore trust. If the new fingerprint differs from the old one, the entire cache is cleared.
• Disable: Set __flags__.WORKFLOW_DI_NO_CACHE = True to disable caching entirely.

Code example

from amrita_sense._unsafe import __flags__
from amrita_sense.runtime.workflow import WorkflowInterpreter

# Default: DI cache enabled
pc = WorkflowInterpreter(rendered, extra_args=(my_service,))
await pc.run()  # Second pass of a loop body will reuse cached DI results

# Disable cache for providers with side effects
__flags__.WORKFLOW_DI_NO_CACHE = True
pc2 = WorkflowInterpreter(rendered)
await pc2.run()  # Every node re-resolves dependencies from scratch

DI Preload Cache

When __flags__.WORKFLOW_DI_PRELOAD_CACHE is enabled, the interpreter pre-resolves dependency injection for every node during the run() initialization phase — before the first node executes.

How it works

1. run() calls _refresh_di_cache_full() after resolving runtime arguments
2. The method walks the entire workflow graph using advance_pointer() with a temporary PointerVector
3. For each node, it spawns an async worker that resolves DI and stores the result in _di_cache.payload
4. Workers run in concurrent batches controlled by WORKFLOW_DI_PRELOAD_BATCH (default: 10)
5. The preloader respects the cache capacity — if _di_cache.payload reaches its maximum size (2048), remaining nodes are skipped to avoid cache thrashing
6. After preloading completes, the main loop starts — every _call() is a cache hit

Performance characteristics

Aspect	Without preload	With preload
Startup latency	Minimal	Proportional to graph size × batch count
Per-node latency	First visit: full DI resolution	Always: cache hit (O(1) lookup)
Memory	Grows lazily as nodes are visited	Pre-allocated for all nodes at startup
Best for	Short workflows, one-shot executions	Long-running loops, repeated node visits

Code example

from amrita_sense._unsafe import __flags__

__flags__.WORKFLOW_DI_PRELOAD_CACHE = True
__flags__.WORKFLOW_DI_PRELOAD_BATCH = 20  # Increase parallelism

pc = WorkflowInterpreter(rendered)
await pc.run()  # DI is pre-resolved for all nodes before the first node runs

Cache Constraints and Flag Conflicts

NO_DEPENDENCY_META_CACHE conflict

Setting WORKFLOW_DI_PRELOAD_CACHE = True together with NO_DEPENDENCY_META_CACHE = True raises a RuntimeError. The preload mechanism depends on cached DependencyMeta (from sign_func) for efficient batch resolution — disabling the meta cache makes preloading unreliable.

WORKFLOW_DI_NO_CACHE conflict

Setting WORKFLOW_DI_NO_CACHE = True together with WORKFLOW_DI_PRELOAD_CACHE = True also raises a RuntimeError. These flags have contradictory intent: one disables caching, the other pre-populates it.

hash_trustable guard

_refresh_di_cache_full() will raise DependsResolveFailed if hash_trustable is False when called. Always call rehash_args() after modifying DI arguments to ensure cache integrity.

Annotated Declarations

Depends may be written inside the annotation instead of as the default value. Both forms are equivalent, and a parameter must not use both at once:

from typing import Annotated

from amrita_sense.hook.matcher import Depends


@Node()
def my_node(
    by_default: ReturnType = Depends(provider),
    by_annotation: Annotated[ReturnType, Depends(provider)] = ...,
): ...

Only the type inside Annotated participates in type-based matching; any other metadata is ignored. When several Depends markers appear, the last one wins. This follows FastAPI's analyze_param.

Declaration	Result
x: T	required — keyword match, then type traversal
x: T = value	optional, keeps value unless a keyword match wins
x: T = Depends(f)	injected by f
x: Annotated[T, Depends(f)]	injected by f
x: Annotated[T, Depends(f)] = value	injected by f; value is ignored
x: Annotated[T, Depends(f)] = Depends(g)	DependsDeclarationError
x: Annotated[T, "other metadata"]	treated as x: T

A malformed declaration raises DependsDeclarationError while the node or handler is being constructed, not when it runs.

Generator Dependencies and Lifecycle

A provider may be a generator. The value before yield is injected, and everything after it runs as a teardown:

from collections.abc import AsyncIterator


async def get_session() -> AsyncIterator[Session]:
    session = Session()
    try:
        yield session
    finally:
        await session.close()


@Node()
async def handler(session: Session = Depends(get_session, scope="workflow")): ...

Both bare generator functions (def ... yield, async def ... yield) and providers already wrapped with @contextmanager / @asynccontextmanager are accepted.

Scopes

Scope	Closed when
"call"	the node call that requested the dependency returns
"dispatch"	every handler of the current event dispatch has finished
"workflow"	the interpreter stops

"call" is available inside nodes, "dispatch" inside event handlers, and "workflow" in both. Leaving scope unset resolves to the widest scope available at the resolution site — "workflow" for nodes, "dispatch" for event handlers.

Declaring a scope for a non-generator provider raises DependsDeclarationError. Declaring one that is not active at the resolution site raises DependsInjectFailed.

Teardown runs in the reverse of declaration order, and it runs even when the node raises. A generator that yields None counts as a failed resolution — as with any other provider — but its teardown still runs.

A "workflow" scope is deliberately not closed when a run ends in a panic or a suspend: both are resumable, so resources stay alive until the interpreter is terminated. await interpreter.terminate() releases them and is safe to call more than once.

Sharing and de-duplication

Resources are keyed by the provider, the fingerprint of the arguments it was given, and the scope. Two nodes depending on the same "workflow"-scoped provider receive the same instance, and two resolvers racing on the same key await a single computation rather than opening the resource twice.

Generator providers never enter the DI result cache — an LRU eviction would orphan their teardown — so they are also skipped by the cache preload.

Nested dependencies

A provider may itself declare Depends parameters. Sub-dependencies resolve through the same lifecycle-aware path, so a nested generator registers on the caller's scope and is torn down together with it.

Database sessions

For ORM sessions, prefer scope="call" — the session is then opened and closed around a single node, which is the safest default. Only widen to "workflow" when the session is genuinely meant to be shared across the run.