Custom Nodes

In AmritaSense, nodes are the basic units of execution flow. Built-in instructions are ultimately expanded into node compositions, while custom nodes are the direct way developers encapsulate their business logic. This chapter explains node essence, lifecycle, and how to use POINTER_DEPENDS to gain full interpreter control when necessary.

The @Node decorator and node essence

The @Node() decorator converts an ordinary Python function or coroutine into a workflow node. It does not do anything complex — it simply packages the function object, signature metadata, and a few data fields into a Node instance.

Decorator parameters

@Node(
    tag="custom_tag",  # optional: breakpoint label, autogenerated by default
    wrap_to_async=True,  # optional: whether to wrap sync functions as async
    address_able=True,  # optional: whether it can be referenced by ALIAS
)
def my_function():
    pass

tag is both a debugging identifier and the suspension point name used by flow interrupts. If omitted, it defaults to NodeSuspend::{function_name}.

Node essence: a thin wrapper around a callable

Each node is essentially a thin wrapper around the original function. It preserves the original function’s signature information (fun_sign) and execution capability (func), while adding AmritaSense metadata:

• func: the original function object, which the interpreter calls during execution
• fun_sign: the function signature extracted by inspect.signature, used by dependency injection to match parameters
• tag: the node’s unique identifier string
• wrap_to_async: whether synchronous functions should be wrapped to async
• address_able: whether the node can be referenced by ALIAS; only True nodes can become JMP or INVOKE targets

Everything is a node — this is AmritaSense’s core philosophy. Conditionals, loop bodies, exception handlers, and JMP targets are all Node or BaseNode instances. Custom nodes are no exception.

Handling sync and async nodes

AmritaSense unifies synchronous and asynchronous nodes. In _call(), it decides how to execute based on iscoroutinefunction and wrap_to_async.

Asynchronous nodes

Nodes defined with async def are awaited directly:

@Node()
async def fetch_data():
    return await http_get("/api")

The interpreter detects the coroutine function and executes await fun(*args, **kwargs).

Synchronous nodes + wrap_to_async=True (default)

Synchronous functions default to wrap_to_async=True, and the interpreter runs them with asyncio.to_thread:

@Node()
def heavy_compute():
    return sum(range(10**7))

This avoids blocking the event loop. It is suitable for CPU-bound tasks or legacy synchronous code.

Synchronous nodes + wrap_to_async=False

When wrap_to_async=False, the interpreter calls fun(*args, **kwargs) directly:

@Node(wrap_to_async=False)
def quick_check():
    return len(queue) > 0

This is appropriate for extremely lightweight sync operations, such as simple condition checks with zero scheduling overhead.

Performance considerations

• CPU-bound tasks -> wrap_to_async=True to avoid blocking the event loop.
• I/O-bound tasks -> prefer native async def.
• Minimal sync operations (like NOP or simple boolean checks) -> wrap_to_async=False.

The Node class is also memory-optimized using __slots__, avoiding a default __dict__ and keeping each node as lightweight as possible.

Node lifecycle and atomicity

Lifecycle

1. Creation: when the module loads, the @Node() decorator runs and creates a Node instance.
2. Compilation: during render(), the node is placed into NodeComposeRendered's _graph array and given a PointerVector address. After the graph is fully built, _post_compile is called, where jump nodes resolve aliases at compile time.
3. Pre-check: before each execution, _pre_check is called. Used for runtime checks that depend on interpreter state (e.g., BatchRun forks child interpreters here).
4. Execution: the interpreter acquires the lock, checks suspension points, resolves dependencies, and calls func.
5. Completion: after execution, the lock is released and the interpreter advances the pointer.

Atomicity guarantees

Node atomicity is guaranteed by the interpreter lock and cooperative interruption:

• Only one node executes at a time: the interpreter lock prevents concurrent node execution or external injection.
• Node boundaries are safe boundaries: interrupt signals are checked only after a node finishes, so the node body is not preempted.
• Exception safety: if a node raises an exception, the interpreter’s run_step_by() loop catches and handles it at the top level. The return stack (_ret_addr_stack) and pointer state remain consistent because nodes do not manipulate those structures directly; the interpreter manages them.

Compile-time and pre-check hooks

If a custom node class inherits from BaseNode, it can override two lifecycle hooks:

• _post_compile(compose: NodeComposeRendered) — called after the workflow graph is fully compiled. InvokeNode and JumpNode use this hook to resolve aliases to addresses at compile time, avoiding runtime overhead.
• _pre_check(pointer: WorkflowInterpreter) — called before each node execution. Used for runtime checks that depend on interpreter state (e.g., BatchRun forks child interpreters here).

POINTER_DEPENDS: access to the interpreter

POINTER_DEPENDS is a special dependency injection factory that allows a node to obtain the current WorkflowInterpreter instance.

from amrita_sense.runtime.deps import POINTER_DEPENDS
from amrita_sense.runtime.workflow import WorkflowInterpreter


@Node()
def my_node(pc: WorkflowInterpreter = Depends(POINTER_DEPENDS)):
    current_addr = pc._pointer  # read current position
    target = pc.get_graph().calc.resolve_alias("foo")  # resolve alias
    await pc.call_sub(target, arg=42)  # call a subroutine

What does the node gain?

By injecting pc, the node gains full access to the interpreter — reading the current pointer, resolving aliases, calling subroutines, and even performing dynamic jumps. This is one of AmritaSense’s core ideas: nodes are not passive units called by the framework; they can actively control execution flow.

When to use and when not to use it

• Use POINTER_DEPENDS when a node needs to call subroutines (call_sub), resolve aliases (get_graph().calc.resolve_alias()), or perform dynamic jumps.
• Do not use it when the node only implements pure business logic, such as data processing or API calls. In those cases, declare only business dependencies like a database connection or HTTP client.

Power comes with responsibility

With interpreter access, nodes can directly manipulate pointers and the call stack. This power comes with responsibility — internal jumps set the JMP bit, affecting interpreter behavior, and manual stack management can break call stack integrity.

Therefore, inject POINTER_DEPENDS only when necessary. Most nodes should use normal Python logic and composition-level instructions (IF, WHILE, INVOKE) to express control flow, and only directly access the interpreter when instructions cannot express the desired behavior.

Safe runtime integration

Custom nodes that need runtime context should use POINTER_DEPENDS to access WorkflowInterpreter rather than manipulating internal interpreter state directly. The interpreter exposes safe APIs such as:

• jump_to(addr)
• jump_near(addr)
• jump_offset(offset)
• call_sub(addr, *args, interrupt=False)
• get_graph().calc.resolve_alias(alias)
• object_io

Because object_io is a generic SuspendObjectStream subclass, custom nodes can also participate in external suspend/resume or streaming I/O patterns without changing the core interpreter loop.

TIP

If you only need data dependencies, prefer Depends(...) with provider functions. Use POINTER_DEPENDS only for workflow control or low-level runtime inspection.

Summary

Custom nodes are the “cells” of AmritaSense. They remain simple in essence — a thin wrapper around a Python function — while gaining full access to the Amrita ecosystem via Depends and POINTER_DEPENDS. In the next section, we will explore how to package recurring composition patterns into new self-compiled instructions to extend workflow expressiveness.