Rhizomorph Behavior Trees API Reference¶

Core Module¶

mycorrhizal.rhizomorph.core ¶

Rhizomorph - Asyncio Behavior Tree Framework

A dual-API framework for defining and executing behavior trees with support for asyncio, multi-file composition, and type-safe blackboard interfaces.

Two Usage Patterns:

Decorator DSL (Declarative): from mycorrhizal.rhizomorph.core import bt, Runner, Status

@bt.tree def MyBehaviorTree(): @bt.action async def do_work(bb) -> Status: return Status.SUCCESS
```
@bt.condition
def should_work(bb) -> bool:
    return bb.work_available

@bt.root
@bt.sequence
def root():
    yield should_work
    yield do_work
```
runner = Runner(MyBehaviorTree, bb=blackboard) result = await runner.tick_until_complete()
TreeBuilder API (Programmatic): from mycorrhizal.rhizomorph.core import TreeBuilder, Runner, Status

builder = TreeBuilder("MyTree")

has_battery = builder.condition("has_battery", lambda bb: bb.battery > 20) move_forward = builder.action("move_forward", move_func)

Fluent wrapper chaining¶

protected_move = move_forward.gate(has_battery).timeout(5.0).retry(3)

patrol = builder.sequence(has_battery, protected_move, memory=True) tree = builder.build(patrol)

runner = Runner(tree, bb=blackboard) result = await runner.tick_until_complete()

Key Classes

Status - Result status for behavior tree nodes (SUCCESS, FAILURE, RUNNING, etc.) Runner - Runtime for executing behavior trees Node - Base class for all behavior tree nodes Action - Leaf node that executes a function Condition - Leaf node that evaluates a predicate Sequence - Composite that runs children in order Selector - Composite that runs children until one succeeds Parallel - Composite that runs children concurrently TreeBuilder - Programmatic API for building trees without decorators

Multi-file Composition

Use bt.subtree() or builder.subtree() to reference trees defined in other modules: from other_module import OtherTree

@bt.tree
def MainTree():
    @bt.root
    @bt.sequence
    def root():
        yield bt.subtree(OtherTree)

# Or with TreeBuilder:
builder = TreeBuilder("MainTree")
builder.root(builder.sequence(
    builder.subtree(OtherTree),
    other_action
))

Status ¶

Bases: Enum

Result status for behavior tree node execution.

Note

Composite nodes use these statuses to determine control flow: - Sequence stops on first FAILURE - Selector stops on first SUCCESS - Parallel waits for all children, fails if any fail

ExceptionPolicy ¶

Bases: Enum

Policy for handling exceptions in action/condition nodes.

RecursionError ¶

Bases: Exception

Raised when recursive behavior tree structure is detected

Node ¶

Node(name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Generic[BB]

Base class for behavior tree nodes.

All behavior tree nodes inherit from this class. Nodes are executed by calling the tick() method, which returns a Status indicating the result of execution.

Parameters:

Name	Type	Description	Default
`name`	`Optional[str]`	Optional name for this node (used in debugging and logging)	`None`
`exception_policy`	`ExceptionPolicy`	How to handle exceptions during execution	`LOG_AND_CONTINUE`

Attributes:

Name	Type	Description
`name`	`str`	Node name
`parent`	`Optional[Node[BB]]`	Parent node in the tree
`exception_policy`		Exception handling policy
`_entered`	`bool`	Whether on_enter has been called
`_last_status`	`Optional[Status]`	Last status returned by tick

Methods:

Name	Description
`tick`	Execute the node, return Status
`on_enter`	Called when node is first entered
`on_exit`	Called when node exits (not RUNNING)
`reset`	Reset node state for reuse

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    self.name: str = name or _name_of(self)
    self.parent: Optional[Node[BB]] = None
    self.exception_policy = exception_policy
    self._entered: bool = False
    self._last_status: Optional[Status] = None

Action ¶

Action(func: Callable[..., Any], exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Leaf node that executes a function.

Action nodes wrap sync or async functions that perform work or check conditions. The function can return: - Status enum (SUCCESS, FAILURE, RUNNING, etc.) - bool (True -> SUCCESS, False -> FAILURE) - None (treated as SUCCESS)

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any]`	Function to execute (sync or async)	required
`name`		Optional name for this action	required
`exception_policy`	`ExceptionPolicy`	How to handle exceptions	`LOG_AND_CONTINUE`

The function signature can be: - func(bb) -> Status | bool | None - func(bb, tb) -> Status | bool | None

where bb is the blackboard and tb is an optional timebase.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    func: Callable[..., Any],
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(name=_name_of(func), exception_policy=exception_policy)
    self._func = func
    # Cache whether function supports timebase parameter (checked during construction)
    self._supports_timebase = _supports_timebase(func)
    # Cache fully qualified name for tracing
    self._fq_name = _fully_qualified_name(func)

Condition ¶

Condition(func: Callable[..., Any], exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Action[BB]

Boolean leaf: True→SUCCESS, False→FAILURE (status also accepted).

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    func: Callable[..., Any],
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(name=_name_of(func), exception_policy=exception_policy)
    self._func = func
    # Cache whether function supports timebase parameter (checked during construction)
    self._supports_timebase = _supports_timebase(func)
    # Cache fully qualified name for tracing
    self._fq_name = _fully_qualified_name(func)

Sequence ¶

Sequence(children: Sequence[Node[BB]], *, memory: bool = True, name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Sequence (AND): fail/err fast; RUNNING bubbles; all SUCCESS → SUCCESS.

Memory Behavior

When memory=True (default), the sequence remembers its position across ticks. This allows the sequence to progress through its children incrementally.

When memory=False, the sequence restarts from the beginning on every tick. This is useful for reactive sequences that should always start from the first child.

Important Note on do_while Loops

If a sequence is used as the child of a do_while loop, it typically needs memory=True to make progress. Without memory, the sequence will restart from its first child on each tick, preventing it from completing all children.

Example: @bt.sequence(memory=True) # Required for progress def image_samples(): yield bt.subtree(MoveToSample) yield send_image_request yield bt.subtree(IncrementSampleCounter)

@bt.sequence(memory=True)
def happy_path():
    yield bt.do_while(samples_remain)(image_samples)
    yield set_pod_to_grow

Without memory=True on image_samples, the do_while loop would never progress past MoveToSample because the sequence would restart each tick.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    children: SequenceT[Node[BB]],
    *,
    memory: bool = True,
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(name or "Sequence", exception_policy=exception_policy)
    self.children = list(children)
    for ch in self.children:
        ch.parent = self
    self.memory = memory
    self._idx = 0

Selector ¶

Selector(children: Sequence[Node[BB]], *, memory: bool = True, name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Selector (Fallback): first SUCCESS wins; RUNNING bubbles; else FAILURE.

Memory Behavior

When memory=True (default), the selector remembers its position across ticks. This allows the selector to continue trying children from where it left off.

When memory=False, the selector restarts from the beginning on every tick. This is useful for reactive selectors that should always re-evaluate from the first child.

Important Note on do_while Loops

Similar to Sequence, if a selector is used as the child of a do_while loop, it typically needs memory=True to make progress. Without memory, the selector will restart from its first child on each tick.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    children: SequenceT[Node[BB]],
    *,
    memory: bool = True,
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(name or "Selector", exception_policy=exception_policy)
    self.children = list(children)
    for ch in self.children:
        ch.parent = self
    self.memory = memory
    self._idx = 0

Parallel ¶

Parallel(children: Sequence[Node[BB]], *, success_threshold: int, failure_threshold: Optional[int] = None, name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Parallel: tick all children per tick concurrently. - success_threshold (k) SUCCESS to report SUCCESS - failure_threshold defaults to n-k+1 (cannot reach success anymore) to report FAILURE Otherwise RUNNING.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    children: SequenceT[Node[BB]],
    *,
    success_threshold: int,
    failure_threshold: Optional[int] = None,
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(name or "Parallel", exception_policy=exception_policy)
    self.children = list(children)
    for ch in self.children:
        ch.parent = self
    self.n = len(self.children)
    self.k = max(1, min(success_threshold, self.n))
    self.f = (
        failure_threshold
        if failure_threshold is not None
        else (self.n - self.k + 1)
    )

RateLimit ¶

RateLimit(child: Node[BB], *, hz: Optional[float] = None, period: Optional[float] = None, name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Throttle starting the child to at most 1 per period (or hz). If child is RUNNING, do not throttle (avoid starvation).

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    child: Node[BB],
    *,
    hz: Optional[float] = None,
    period: Optional[float] = None,
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    if (hz is not None) and (period is not None):
        raise ValueError("RateLimit requires exactly one of (hz, period)")

    if period is not None:
        per = period
    elif hz is not None:
        per = 1.0 / float(hz)
    else:
        raise ValueError("RateLimit requires exactly one of (hz, period)")

    super().__init__(
        name or f"RateLimit({_name_of(child)},{per:.6f}s)",
        exception_policy=exception_policy,
    )
    self.child = child
    self.child.parent = self
    self._period = max(0.0, per)
    self._next_allowed: Optional[float] = None
    self._last: Optional[Status] = None

Gate ¶

Gate(condition: Node[BB], child: Node[BB], name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Guard a child with a condition

SUCCESS → tick child RUNNING → RUNNING else → FAILURE

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    condition: Node[BB],
    child: Node[BB],
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(
        name or f"Gate(cond={_name_of(condition)}, child={_name_of(child)})",
        exception_policy=exception_policy,
    )
    self.condition = condition
    self.child = child
    self.condition.parent = self
    self.child.parent = self

When ¶

When(condition: Node[BB], child: Node[BB], name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Conditionally execute a child, returning SUCCESS if the condition fails.

Unlike gate(), when() does NOT fail the parent when the condition is false. This is useful for optional steps or feature flags in sequences.

Behavior

If condition returns SUCCESS → execute child, return child's status
If condition returns RUNNING → return RUNNING
Otherwise → return SUCCESS (skip but don't fail parent)

Example

yield bt.when(feature_enabled)(optional_action)

If feature_enabled is false, returns SUCCESS and sequence continues¶

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    condition: Node[BB],
    child: Node[BB],
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(
        name or f"When(cond={_name_of(condition)}, child={_name_of(child)})",
        exception_policy=exception_policy,
    )
    self.condition = condition
    self.child = child
    self.condition.parent = self
    self.child.parent = self

Match ¶

Match(key_fn: Callable[[Any], Any], cases: List[Tuple[Any, Node[BB]]], name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Pattern-matching dispatch node.

Evaluates a key function against the blackboard, then checks each case in order. The first matching case's child is executed. If the child completes (SUCCESS or FAILURE), that status is returned immediately.

Cases can match by

Type: isinstance(value, case_type)
Predicate: case_predicate(value) returns True
Value: value == case_value
Default: always matches (should be last)

If no case matches and there's no default, returns FAILURE.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    key_fn: Callable[[Any], Any],
    cases: List[Tuple[Any, Node[BB]]],
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(
        name or f"Match({_name_of(key_fn)})",
        exception_policy=exception_policy,
    )
    self._key_fn = key_fn
    self._cases = cases
    for _, child in self._cases:
        child.parent = self
    self._matched_idx: Optional[int] = None
    # Cache whether key_fn supports timebase parameter
    self._key_fn_supports_timebase = _supports_timebase(key_fn)

_matches ¶

_matches(matcher: Any, value: Any) -> bool

Check if a matcher matches the given value.

Source code in src/mycorrhizal/rhizomorph/core.py

def _matches(self, matcher: Any, value: Any) -> bool:
    """Check if a matcher matches the given value."""
    if matcher is _DefaultCase:
        return True
    if isinstance(matcher, type):
        return isinstance(value, matcher)
    if callable(matcher):
        return bool(matcher(value))
    return value == matcher

DoWhile ¶

DoWhile(condition: Node[BB], child: Node[BB], name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Loop decorator that repeats its child while a condition is true.

Behavior

Evaluate condition
If condition is FALSE → return SUCCESS (loop complete)
If condition is TRUE → tick child
If child returns RUNNING → return RUNNING (resume child next tick)
If child returns SUCCESS → reset child, return RUNNING (re-check condition next tick)
If child returns FAILURE → return FAILURE (loop aborted)

The "return RUNNING after child SUCCESS" prevents infinite loops within a single tick.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    condition: Node[BB],
    child: Node[BB],
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(
        name or f"DoWhile(cond={_name_of(condition)}, child={_name_of(child)})",
        exception_policy=exception_policy,
    )
    self.condition = condition
    self.child = child
    self.condition.parent = self
    self.child.parent = self
    self._child_running = False

TryCatch ¶

TryCatch(try_block: Node[BB], catch_block: Node[BB], name: Optional[str] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE)

Bases: Node[BB]

Try-catch error handling node.

Executes the try block first. If it returns SUCCESS, that status is returned. If the try block returns FAILURE, the catch block is executed instead. Returns SUCCESS if either block succeeds, FAILURE if both fail.

This is semantically equivalent to a selector with two children, but provides clearer intent and better visualization with labeled edges.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    try_block: Node[BB],
    catch_block: Node[BB],
    name: Optional[str] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
) -> None:
    super().__init__(
        name or f"TryCatch(try={_name_of(try_block)}, catch={_name_of(catch_block)})",
        exception_policy=exception_policy,
    )
    self.try_block = try_block
    self.catch_block = catch_block
    self.try_block.parent = self
    self.catch_block.parent = self
    self._use_catch = False

_DefaultCase ¶

Sentinel for default case in match expressions.

CaseSpec `dataclass` ¶

CaseSpec(matcher: Any, child: 'NodeSpec', label: str = '')

Specification for a case in a match expression.

_WrapperChain ¶

_WrapperChain(builders: Optional[List[Callable[..., Any]]] = None, labels: Optional[List[str]] = None, metadata: Optional[List[Dict[str, Any]]] = None)

Fluent factory for decorator stacks that read left→right.

chain = bt.failer().gate(battery_ok).timeout(0.12)
yield chain(engage)

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    builders: Optional[List[Callable[..., Any]]] = None,
    labels: Optional[List[str]] = None,
    metadata: Optional[List[Dict[str, Any]]] = None,
) -> None:
    self._builders: List[Callable[..., Any]] = list(builders or [])
    self._labels: List[str] = list(labels or [])
    self._metadata: List[Dict[str, Any]] = list(metadata or [])

when ¶

when(condition_spec_or_fn: Union['NodeSpec', Callable[[Any], Any]]) -> '_WrapperChain'

Add a conditional wrapper that executes the child only when condition is true.

Unlike gate(), when() returns SUCCESS (not FAILURE) when the condition fails. This allows sequences to continue when optional steps are skipped.

Parameters:

Name	Type	Description	Default
`condition_spec_or_fn`	`Union['NodeSpec', Callable[[Any], Any]]`	A node spec or callable that returns True/False	required

Returns:

Type	Description
`'_WrapperChain'`	The chain for further wrapping

Example

yield bt.when(is_enabled)(optional_action)

If is_enabled is False, returns SUCCESS and sequence continues¶

Source code in src/mycorrhizal/rhizomorph/core.py

def when(
    self, condition_spec_or_fn: Union["NodeSpec", Callable[[Any], Any]]
) -> "_WrapperChain":
    """
    Add a conditional wrapper that executes the child only when condition is true.

    Unlike gate(), when() returns SUCCESS (not FAILURE) when the condition fails.
    This allows sequences to continue when optional steps are skipped.

    Args:
        condition_spec_or_fn: A node spec or callable that returns True/False

    Returns:
        The chain for further wrapping

    Example:
        yield bt.when(is_enabled)(optional_action)
        # If is_enabled is False, returns SUCCESS and sequence continues
    """
    cond_spec = bt.as_spec(condition_spec_or_fn)
    # Store condition spec in metadata for compiler access
    return self._append(
        f"When(cond={_name_of(cond_spec)})",
        lambda ch: When(cond_spec.to_node(), ch),
        metadata={"condition": cond_spec},
    )

call ¶

__call__(inner: Union['NodeSpec', Callable[[Any], Any]]) -> 'NodeSpec'

Apply the chain to a child spec → nested decorator NodeSpecs. Left→right call order becomes outermost→…→innermost when built.

Source code in src/mycorrhizal/rhizomorph/core.py

def __call__(self, inner: Union["NodeSpec", Callable[[Any], Any]]) -> "NodeSpec":
    """
    Apply the chain to a child spec → nested decorator NodeSpecs.
    Left→right call order becomes outermost→…→innermost when built.
    """
    spec = bt.as_spec(inner)
    result = spec
    for label, builder, metadata in reversed(list(zip(self._labels, self._builders, self._metadata))):
        payload = builder
        # Include metadata (like condition spec) in payload for compiler access
        # Only wrap in dict if there's metadata to store
        if metadata:
            payload = {"builder": builder, **metadata}
        result = NodeSpec(
            kind=NodeSpecKind.DECORATOR,
            name=f"{label}({_name_of(result)})",
            payload=payload,
            children=[result],
        )
    return result

_CaseBuilder ¶

_CaseBuilder(matcher: Any)

Builder for individual match cases.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(self, matcher: Any) -> None:
    self._matcher = matcher

_MatchBuilder ¶

_MatchBuilder(key_fn: Callable[[Any], Any], name: Optional[str] = None)

Builder for match expressions.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(self, key_fn: Callable[[Any], Any], name: Optional[str] = None) -> None:
    self._key_fn = key_fn
    self._name = name

_DoWhileBuilder ¶

_DoWhileBuilder(condition_spec: NodeSpec)

Builder for do_while loops.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(self, condition_spec: NodeSpec) -> None:
    self._condition_spec = condition_spec

_BT ¶

_BT()

User-facing decorator/constructor namespace.

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(self):
    self._tracking_stack: List[List[Tuple[str, Any]]] = []

action ¶

action(fn: Callable[..., Any]) -> Callable[..., Any]

Decorator to mark a function as an action node.

Source code in src/mycorrhizal/rhizomorph/core.py

def action(self, fn: Callable[..., Any]) -> Callable[..., Any]:
    """Decorator to mark a function as an action node."""
    spec = NodeSpec(kind=NodeSpecKind.ACTION, name=_name_of(fn), payload=fn)
    fn.node_spec = spec  # type: ignore
    if self._tracking_stack:
        self._tracking_stack[-1].append((fn.__name__, fn))
    return fn

condition ¶

condition(fn: Callable[..., Any]) -> Callable[..., Any]

Decorator to mark a function as a condition node.

Source code in src/mycorrhizal/rhizomorph/core.py

def condition(self, fn: Callable[..., Any]) -> Callable[..., Any]:
    """Decorator to mark a function as a condition node."""
    spec = NodeSpec(kind=NodeSpecKind.CONDITION, name=_name_of(fn), payload=fn)
    fn.node_spec = spec  # type: ignore
    if self._tracking_stack:
        self._tracking_stack[-1].append((fn.__name__, fn))
    return fn

sequence ¶

sequence(*args: Union[F, NodeSpec, Callable[[Any], Any]], memory: bool = True) -> Union[F, Callable[[F], F], NodeSpec]

Decorator to mark a generator function as a sequence composite.

Can be used in three ways

Decorator without parentheses: @bt.sequence def root(): ...
Decorator with parameters: @bt.sequence(memory=False) def root(): ...
Direct call with child nodes: bt.sequence(action1, action2, action3)

Memory Parameter

The memory parameter controls whether the sequence remembers its position across ticks:

memory=None (default): Use the Runner's memory setting
memory=True: Remember position, allowing incremental progress
memory=False: Restart from beginning each tick (reactive behavior)

IMPORTANT: If a sequence is inside a do_while loop and needs to execute all its children incrementally, use memory=True. Otherwise, the sequence will restart from the first child on every tick and never complete.

Example: # CORRECT - sequence progresses through children @bt.sequence(memory=True) def process_samples(): yield move_to_sample yield capture_image yield bt.do_while(samples_remain)(process_samples)

# WRONG - sequence restarts at move_to_sample every tick
@bt.sequence(memory=False)
def process_samples():
    yield move_to_sample
    yield capture_image  # Never reached!
yield bt.do_while(samples_remain)(process_samples)

Source code in src/mycorrhizal/rhizomorph/core.py

def sequence(
    self, *args: Union[F, NodeSpec, Callable[[Any], Any]], memory: bool = True
) -> Union[F, Callable[[F], F], NodeSpec]:
    """Decorator to mark a generator function as a sequence composite.

    Can be used in three ways:
        1. Decorator without parentheses:
            @bt.sequence
            def root():
                ...

        2. Decorator with parameters:
            @bt.sequence(memory=False)
            def root():
                ...

        3. Direct call with child nodes:
            bt.sequence(action1, action2, action3)

    Memory Parameter:
        The memory parameter controls whether the sequence remembers its position
        across ticks:

        - memory=None (default): Use the Runner's memory setting
        - memory=True: Remember position, allowing incremental progress
        - memory=False: Restart from beginning each tick (reactive behavior)

        IMPORTANT: If a sequence is inside a do_while loop and needs to execute
        all its children incrementally, use memory=True. Otherwise, the sequence
        will restart from the first child on every tick and never complete.

        Example:
            # CORRECT - sequence progresses through children
            @bt.sequence(memory=True)
            def process_samples():
                yield move_to_sample
                yield capture_image
            yield bt.do_while(samples_remain)(process_samples)

            # WRONG - sequence restarts at move_to_sample every tick
            @bt.sequence(memory=False)
            def process_samples():
                yield move_to_sample
                yield capture_image  # Never reached!
            yield bt.do_while(samples_remain)(process_samples)
    """
    # Case 3: Direct call with children - bt.sequence(node1, node2, ...)
    # This is detected when we have multiple args, or a single arg that's not a generator function
    if len(args) == 0:
        # Case 2a: bt.sequence() with no arguments - return decorator
        return self._sequence_impl(memory=memory)

    if len(args) > 1:
        # Multiple children - create sequence directly
        return self._sequence_from_children(args, memory)

    # Single argument - check if it's a generator function (decorator case) or a node (direct call)
    single_arg = args[0]

    # Check if it's a generator function (decorator form)
    if inspect.isfunction(single_arg) and inspect.isgeneratorfunction(single_arg):
        # Case 1: @bt.sequence def root(): ...
        spec = NodeSpec(
            kind=NodeSpecKind.SEQUENCE,
            name=_name_of(single_arg),
            payload={"factory": single_arg, "memory": memory},
        )
        single_arg.node_spec = spec  # type: ignore
        if self._tracking_stack:
            self._tracking_stack[-1].append((single_arg.__name__, single_arg))
        return single_arg

    # Case 3b: Single child node - bt.sequence(action1)
    return self._sequence_from_children(args, memory)

_sequence_from_children ¶

_sequence_from_children(children: Tuple[Any, ...], memory: bool) -> NodeSpec

Create a sequence NodeSpec from child nodes.

Source code in src/mycorrhizal/rhizomorph/core.py

def _sequence_from_children(self, children: Tuple[Any, ...], memory: bool) -> NodeSpec:
    """Create a sequence NodeSpec from child nodes."""
    # Create a uniquely named factory to avoid false recursion detection
    child_names = ', '.join(_name_of(c) for c in children)

    def _sequence_factory_direct() -> Generator[Any, None, None]:
        for child in children:
            yield child

    # Set a unique name for the factory function
    _sequence_factory_direct.__name__ = f"_sequence_factory_direct_{id(children)}"
    _sequence_factory_direct.__qualname__ = f"_sequence_factory_direct_{id(children)}"

    name = f"Sequence({child_names})" if children else "Sequence"

    return NodeSpec(
        kind=NodeSpecKind.SEQUENCE,
        name=name,
        payload={"factory": _sequence_factory_direct, "memory": memory},
    )

_sequence_impl ¶

_sequence_impl(memory: bool) -> Callable[[F], F]

Implementation of sequence decorator.

Source code in src/mycorrhizal/rhizomorph/core.py

def _sequence_impl(self, memory: bool) -> Callable[[F], F]:
    """Implementation of sequence decorator."""
    def deco(factory: F) -> F:
        spec = NodeSpec(
            kind=NodeSpecKind.SEQUENCE,
            name=_name_of(factory),
            payload={"factory": factory, "memory": memory},
        )
        factory.node_spec = spec  # type: ignore
        if self._tracking_stack:
            self._tracking_stack[-1].append((factory.__name__, factory))
        return factory

    return deco

selector ¶

selector(*args: Union[F, NodeSpec, Callable[[Any], Any]], memory: bool = True) -> Union[F, Callable[[F], F], NodeSpec]

Decorator to mark a generator function as a selector composite.

Can be used in three ways

Decorator without parentheses: @bt.selector def root(): ...
Decorator with parameters: @bt.selector(memory=False) def root(): ...
Direct call with child nodes: bt.selector(option1, option2, option3)

The memory parameter defaults to None, which means use the Runner's memory setting. Explicitly set to True or False to override the Runner's setting.

Source code in src/mycorrhizal/rhizomorph/core.py

def selector(
    self, *args: Union[F, NodeSpec, Callable[[Any], Any]], memory: bool = True
) -> Union[F, Callable[[F], F], NodeSpec]:
    """Decorator to mark a generator function as a selector composite.

    Can be used in three ways:
        1. Decorator without parentheses:
            @bt.selector
            def root():
                ...

        2. Decorator with parameters:
            @bt.selector(memory=False)
            def root():
                ...

        3. Direct call with child nodes:
            bt.selector(option1, option2, option3)

    The memory parameter defaults to None, which means use the Runner's memory setting.
    Explicitly set to True or False to override the Runner's setting.
    """
    # Case 3: Direct call with children - bt.selector(node1, node2, ...)
    # This is detected when we have multiple args, or a single arg that's not a generator function
    if len(args) == 0:
        # Case 2a: bt.selector() with no arguments - return decorator
        return self._selector_impl(memory=memory)

    if len(args) > 1:
        # Multiple children - create selector directly
        return self._selector_from_children(args, memory)

    # Single argument - check if it's a generator function (decorator case) or a node (direct call)
    single_arg = args[0]

    # Check if it's a generator function (decorator form)
    if inspect.isfunction(single_arg) and inspect.isgeneratorfunction(single_arg):
        # Case 1: @bt.selector def root(): ...
        spec = NodeSpec(
            kind=NodeSpecKind.SELECTOR,
            name=_name_of(single_arg),
            payload={"factory": single_arg, "memory": memory},
        )
        single_arg.node_spec = spec  # type: ignore
        if self._tracking_stack:
            self._tracking_stack[-1].append((single_arg.__name__, single_arg))
        return single_arg

    # Case 3b: Single child node - bt.selector(action1)
    return self._selector_from_children(args, memory)

_selector_from_children ¶

_selector_from_children(children: Tuple[Any, ...], memory: bool) -> NodeSpec

Create a selector NodeSpec from child nodes.

Source code in src/mycorrhizal/rhizomorph/core.py

def _selector_from_children(self, children: Tuple[Any, ...], memory: bool) -> NodeSpec:
    """Create a selector NodeSpec from child nodes."""
    # Create a uniquely named factory to avoid false recursion detection
    child_names = ', '.join(_name_of(c) for c in children)

    def _selector_factory_direct() -> Generator[Any, None, None]:
        for child in children:
            yield child

    # Set a unique name for the factory function
    _selector_factory_direct.__name__ = f"_selector_factory_direct_{id(children)}"
    _selector_factory_direct.__qualname__ = f"_selector_factory_direct_{id(children)}"

    name = f"Selector({child_names})" if children else "Selector"

    return NodeSpec(
        kind=NodeSpecKind.SELECTOR,
        name=name,
        payload={"factory": _selector_factory_direct, "memory": memory},
    )

_selector_impl ¶

_selector_impl(memory: bool) -> Callable[[F], F]

Implementation of selector decorator.

Source code in src/mycorrhizal/rhizomorph/core.py

def _selector_impl(self, memory: bool) -> Callable[[F], F]:
    """Implementation of selector decorator."""
    def deco(factory: F) -> F:
        spec = NodeSpec(
            kind=NodeSpecKind.SELECTOR,
            name=_name_of(factory),
            payload={"factory": factory, "memory": memory},
        )
        factory.node_spec = spec  # type: ignore
        if self._tracking_stack:
            self._tracking_stack[-1].append((factory.__name__, factory))
        return factory

    return deco

parallel ¶

parallel(*, success_threshold: int, failure_threshold: Optional[int] = None) -> Callable[[F], F]

Decorator to mark a generator function as a parallel composite.

Source code in src/mycorrhizal/rhizomorph/core.py

def parallel(
    self, *, success_threshold: int, failure_threshold: Optional[int] = None
) -> Callable[[F], F]:
    """Decorator to mark a generator function as a parallel composite."""

    def deco(factory: F) -> F:
        spec = NodeSpec(
            kind=NodeSpecKind.PARALLEL,
            name=_name_of(factory),
            payload={
                "factory": factory,
                "success_threshold": success_threshold,
                "failure_threshold": failure_threshold,
            },
        )
        factory.node_spec = spec  # type: ignore
        if self._tracking_stack:
            self._tracking_stack[-1].append((factory.__name__, factory))
        return factory

    return deco

when ¶

when(condition: Union[NodeSpec, Callable[[Any], Any]]) -> _WrapperChain

Create a conditional wrapper that executes the child only when condition is true.

Unlike gate(), when() returns SUCCESS (not FAILURE) when the condition fails. This is useful for optional steps or feature flags in sequences.

Parameters:

Name	Type	Description	Default
`condition`	`Union[NodeSpec, Callable[[Any], Any]]`	A node spec or callable that returns True/False	required

Returns:

Type	Description
`_WrapperChain`	A wrapper chain that can be applied to a child node

Example

@bt.sequence def my_sequence(): yield bt.when(feature_enabled)(optional_feature) yield next_step # Always reached, even if flag is disabled

Source code in src/mycorrhizal/rhizomorph/core.py

def when(self, condition: Union[NodeSpec, Callable[[Any], Any]]) -> _WrapperChain:
    """
    Create a conditional wrapper that executes the child only when condition is true.

    Unlike gate(), when() returns SUCCESS (not FAILURE) when the condition fails.
    This is useful for optional steps or feature flags in sequences.

    Args:
        condition: A node spec or callable that returns True/False

    Returns:
        A wrapper chain that can be applied to a child node

    Example:
        @bt.sequence
        def my_sequence():
            yield bt.when(feature_enabled)(optional_feature)
            yield next_step  # Always reached, even if flag is disabled
    """
    cond_spec = self.as_spec(condition)
    return _WrapperChain().when(cond_spec)

match ¶

match(key_fn: Callable[[Any], Any], name: Optional[str] = None) -> '_MatchBuilder'

Create a pattern-matching dispatch node.

Usage

bt.match(lambda bb: bb.current_action, name="action_type")( bt.case(ImageAction)(bt.subtree(handle_image)), bt.case(MoveAction)(bt.subtree(handle_move)), bt.case(lambda a: a.priority > 5)(bt.subtree(handle_urgent)), bt.defaultcase(log_unknown), )

Parameters:

Name	Type	Description	Default
`key_fn`	`Callable[[Any], Any]`	Function that extracts the value to match against from the blackboard	required
`name`	`Optional[str]`	Optional display name for the match node (useful when key_fn is a lambda)	`None`

Returns:

Type	Description
`'_MatchBuilder'`	A builder that accepts case specs and returns a NodeSpec

Source code in src/mycorrhizal/rhizomorph/core.py

def match(
    self, key_fn: Callable[[Any], Any], name: Optional[str] = None
) -> "_MatchBuilder":
    """
    Create a pattern-matching dispatch node.

    Usage:
        bt.match(lambda bb: bb.current_action, name="action_type")(
            bt.case(ImageAction)(bt.subtree(handle_image)),
            bt.case(MoveAction)(bt.subtree(handle_move)),
            bt.case(lambda a: a.priority > 5)(bt.subtree(handle_urgent)),
            bt.defaultcase(log_unknown),
        )

    Args:
        key_fn: Function that extracts the value to match against from the blackboard
        name: Optional display name for the match node (useful when key_fn is a lambda)

    Returns:
        A builder that accepts case specs and returns a NodeSpec
    """
    return _MatchBuilder(key_fn, name=name)

case ¶

case(matcher: Any) -> '_CaseBuilder'

Define a case for a match expression.

Parameters:

Name	Type	Description	Default
`matcher`	`Any`	Can be: - A type: matches via isinstance(value, matcher) - A callable: matches if matcher(value) returns True - Any other value: matches via value == matcher	required

Returns:

Type	Description
`'_CaseBuilder'`	A builder that accepts a child node spec

Source code in src/mycorrhizal/rhizomorph/core.py

def case(self, matcher: Any) -> "_CaseBuilder":
    """
    Define a case for a match expression.

    Args:
        matcher: Can be:
            - A type: matches via isinstance(value, matcher)
            - A callable: matches if matcher(value) returns True
            - Any other value: matches via value == matcher

    Returns:
        A builder that accepts a child node spec
    """
    return _CaseBuilder(matcher)

defaultcase ¶

defaultcase(child: Union[NodeSpec, Callable[[Any], Any]]) -> CaseSpec

Define a default case for a match expression (matches anything).

Parameters:

Name	Type	Description	Default
`child`	`Union[NodeSpec, Callable[[Any], Any]]`	The node to execute if this case matches	required

Returns:

Type	Description
`CaseSpec`	A CaseSpec that always matches

Source code in src/mycorrhizal/rhizomorph/core.py

def defaultcase(self, child: Union[NodeSpec, Callable[[Any], Any]]) -> CaseSpec:
    """
    Define a default case for a match expression (matches anything).

    Args:
        child: The node to execute if this case matches

    Returns:
        A CaseSpec that always matches
    """
    child_spec = self.as_spec(child)
    return CaseSpec(matcher=_DefaultCase, child=child_spec, label="default")

do_while ¶

do_while(condition: Union[NodeSpec, Callable[[Any], Any]]) -> '_DoWhileBuilder'

Create a loop that repeats its child while a condition is true.

Usage

@bt.condition def samples_remain(bb): return bb.sample_index < bb.total_samples

yield bt.do_while(samples_remain)(process_sample)

Behavior

Evaluate condition
If condition is FALSE → return SUCCESS (loop complete)
If condition is TRUE → tick child
If child returns RUNNING → return RUNNING (resume child next tick)
If child returns SUCCESS → reset child, return RUNNING (re-check next tick)
If child returns FAILURE → return FAILURE (loop aborted)

Parameters:

Name	Type	Description	Default
`condition`	`Union[NodeSpec, Callable[[Any], Any]]`	A condition node or function to evaluate each iteration	required

Returns:

Type	Description
`'_DoWhileBuilder'`	A builder that accepts a child node spec

Source code in src/mycorrhizal/rhizomorph/core.py

def do_while(
    self, condition: Union[NodeSpec, Callable[[Any], Any]]
) -> "_DoWhileBuilder":
    """
    Create a loop that repeats its child while a condition is true.

    Usage:
        @bt.condition
        def samples_remain(bb):
            return bb.sample_index < bb.total_samples

        yield bt.do_while(samples_remain)(process_sample)

    Behavior:
        1. Evaluate condition
        2. If condition is FALSE → return SUCCESS (loop complete)
        3. If condition is TRUE → tick child
           - If child returns RUNNING → return RUNNING (resume child next tick)
           - If child returns SUCCESS → reset child, return RUNNING (re-check next tick)
           - If child returns FAILURE → return FAILURE (loop aborted)

    Args:
        condition: A condition node or function to evaluate each iteration

    Returns:
        A builder that accepts a child node spec
    """
    cond_spec = self.as_spec(condition)
    return _DoWhileBuilder(cond_spec)

try_catch ¶

try_catch(try_block: Union[NodeSpec, Callable[[Any], Any]]) -> Callable[[Union[NodeSpec, Callable[[Any], Any]]], NodeSpec]

Create a try-catch error handling pattern.

Usage

Define try and catch blocks with explicit memory settings¶

@bt.sequence(memory=True) def try_block(): yield action1 yield action2

@bt.sequence(memory=True) def catch_block(): yield cleanup

Use in the tree¶

@bt.root @bt.sequence def root(): yield bt.try_catch(try_block)(catch_block)

Behavior

Execute try block
If try block returns SUCCESS → return SUCCESS
If try block returns FAILURE → execute catch block
Return SUCCESS if either block succeeds, FAILURE if both fail

This is semantically equivalent to a selector with two children

First child (try) runs first
Second child (catch) runs only if try fails
Returns SUCCESS if either succeeds

Parameters:

Name	Type	Description	Default
`try_block`	`Union[NodeSpec, Callable[[Any], Any]]`	The node to try first (pre-defined sequence/selector/etc with explicit memory settings)	required

Returns:

Type	Description
`Callable[[Union[NodeSpec, Callable[[Any], Any]]], NodeSpec]`	A callable that accepts a catch block (pre-defined sequence/selector/etc with explicit memory settings)

Source code in src/mycorrhizal/rhizomorph/core.py

def try_catch(
    self, try_block: Union[NodeSpec, Callable[[Any], Any]]
) -> Callable[[Union[NodeSpec, Callable[[Any], Any]]], NodeSpec]:
    """
    Create a try-catch error handling pattern.

    Usage:
        # Define try and catch blocks with explicit memory settings
        @bt.sequence(memory=True)
        def try_block():
            yield action1
            yield action2

        @bt.sequence(memory=True)
        def catch_block():
            yield cleanup

        # Use in the tree
        @bt.root
        @bt.sequence
        def root():
            yield bt.try_catch(try_block)(catch_block)

    Behavior:
        1. Execute try block
        2. If try block returns SUCCESS → return SUCCESS
        3. If try block returns FAILURE → execute catch block
        4. Return SUCCESS if either block succeeds, FAILURE if both fail

    This is semantically equivalent to a selector with two children:
        - First child (try) runs first
        - Second child (catch) runs only if try fails
        - Returns SUCCESS if either succeeds

    Args:
        try_block: The node to try first (pre-defined sequence/selector/etc with explicit memory settings)

    Returns:
        A callable that accepts a catch block (pre-defined sequence/selector/etc with explicit memory settings)
    """
    try_spec = self.as_spec(try_block)

    def catcher(catch_block: Union[NodeSpec, Callable[[Any], Any]]) -> NodeSpec:
        catch_spec = self.as_spec(catch_block)
        return NodeSpec(
            kind=NodeSpecKind.TRY_CATCH,
            name=f"TryCatch(try={_name_of(try_spec)}, catch={_name_of(catch_spec)})",
            payload={
                "try": try_spec,
                "catch": catch_spec,
            },
            children=[try_spec, catch_spec],
        )

    return catcher

subtree ¶

subtree(tree: SimpleNamespace) -> NodeSpec

Mount another tree's root spec as a subtree.

Parameters:

Name	Type	Description	Default
`tree`	`SimpleNamespace`	A tree namespace created with @bt.tree	required

Source code in src/mycorrhizal/rhizomorph/core.py

def subtree(self, tree: SimpleNamespace) -> NodeSpec:
    """
    Mount another tree's root spec as a subtree.

    Args:
        tree: A tree namespace created with @bt.tree
    """
    if not hasattr(tree, "root"):
        raise ValueError(
            f"Tree namespace must have a 'root' attribute. Did you forget @bt.root on a composite?"
        )

    root_spec = tree.root
    tree_name = getattr(tree, "_tree_name", root_spec.name)
    return NodeSpec(
        kind=NodeSpecKind.SUBTREE,
        name=tree_name,
        payload={"root": root_spec},
        children=[root_spec],
    )

tree ¶

tree(fn: Callable[[], Any]) -> SimpleNamespace

Decorator to create a behavior tree namespace.

Usage

@bt.tree def MyTree(): @bt.action def my_action(bb: BB) -> Status: ...

@bt.sequence()
def root():
    yield my_action

Source code in src/mycorrhizal/rhizomorph/core.py

def tree(self, fn: Callable[[], Any]) -> SimpleNamespace:
    """
    Decorator to create a behavior tree namespace.

    Usage:
        @bt.tree
        def MyTree():
            @bt.action
            def my_action(bb: BB) -> Status:
                ...

            @bt.sequence()
            def root():
                yield my_action
    """
    created_nodes = []
    self._tracking_stack.append(created_nodes)

    try:
        fn()
    finally:
        self._tracking_stack.pop()

    nodes = {
        name: node for name, node in created_nodes if hasattr(node, "node_spec")
    }
    namespace = SimpleNamespace(**nodes)

    # Store the tree's name for use in subtree references
    namespace._tree_name = fn.__name__

    for name, node in nodes.items():
        if hasattr(node, "node_spec"):
            node.node_spec.owner = namespace

    root_nodes = [v for v in nodes.values() if hasattr(v.node_spec, "is_root")]
    if root_nodes:
        namespace.root = root_nodes[0].node_spec

    namespace.to_mermaid = lambda: _generate_mermaid(namespace)

    return namespace

root ¶

root(fn: F) -> F

Mark a composite as the root of the tree.

Source code in src/mycorrhizal/rhizomorph/core.py

def root(self, fn: F) -> F:
    """Mark a composite as the root of the tree."""
    if not hasattr(fn, "node_spec"):
        raise TypeError(
            f"@bt.root can only be used on composites (sequences, selectors, etc.), "
            f"got {fn!r}"
        )
    fn.node_spec.is_root = True  # type: ignore
    return fn

as_spec ¶

as_spec(maybe: Union[NodeSpec, Callable[[Any], Any]]) -> NodeSpec

Convert a function or NodeSpec to a NodeSpec.

Source code in src/mycorrhizal/rhizomorph/core.py

def as_spec(self, maybe: Union[NodeSpec, Callable[[Any], Any]]) -> NodeSpec:
    """Convert a function or NodeSpec to a NodeSpec."""
    if isinstance(maybe, NodeSpec):
        return maybe

    spec = getattr(maybe, "node_spec", None)
    if spec is None:
        raise TypeError(
            f"{maybe!r} is not a BT node (missing node_spec attribute)."
        )
    return spec

_NodeWrapper ¶

_NodeWrapper(spec: NodeSpec, builder: 'TreeBuilder')

Fluent wrapper for programmatic nodes that enables decorator-like chaining.

Similar to _WrapperChain but for NodeSpecs created via TreeBuilder.

Example

wrapped = builder.action("scan", scan_func).timeout(1.0).retry(3)

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(self, spec: NodeSpec, builder: "TreeBuilder"):
    self._spec = spec
    self._builder = builder

spec `property` ¶

spec: NodeSpec

Get the underlying NodeSpec

timeout ¶

timeout(seconds: float) -> '_NodeWrapper'

Wrap with a Timeout decorator

Source code in src/mycorrhizal/rhizomorph/core.py

def timeout(self, seconds: float) -> "_NodeWrapper":
    """Wrap with a Timeout decorator"""
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"Timeout({seconds}s, {self._spec.name})",
        payload=lambda ch: Timeout(ch, seconds=seconds),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

retry ¶

retry(max_attempts: int, retry_on: Tuple[Status, ...] = (FAILURE, ERROR)) -> '_NodeWrapper'

Wrap with a Retry decorator

Source code in src/mycorrhizal/rhizomorph/core.py

def retry(
    self,
    max_attempts: int,
    retry_on: Tuple[Status, ...] = (Status.FAILURE, Status.ERROR),
) -> "_NodeWrapper":
    """Wrap with a Retry decorator"""
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"Retry({max_attempts}, {self._spec.name})",
        payload=lambda ch: Retry(ch, max_attempts=max_attempts, retry_on=retry_on),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

ratelimit ¶

ratelimit(*, hz: Optional[float] = None, period: Optional[float] = None) -> '_NodeWrapper'

Wrap with a RateLimit decorator

Source code in src/mycorrhizal/rhizomorph/core.py

def ratelimit(
    self, *, hz: Optional[float] = None, period: Optional[float] = None
) -> "_NodeWrapper":
    """Wrap with a RateLimit decorator"""
    label = "RateLimit(?)"
    if hz is not None:
        label = f"RateLimit({1.0 / float(hz):.6f}s)"
    elif period is not None:
        label = f"RateLimit({float(period):.6f}s)"

    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"{label}, {self._spec.name}",
        payload=lambda ch: RateLimit(ch, hz=hz, period=period),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

gate ¶

gate(condition: Union[NodeSpec, Callable[[Any], Any]]) -> '_NodeWrapper'

Wrap with a Gate decorator - fails when condition is false

Source code in src/mycorrhizal/rhizomorph/core.py

def gate(self, condition: Union[NodeSpec, Callable[[Any], Any]]) -> "_NodeWrapper":
    """Wrap with a Gate decorator - fails when condition is false"""
    cond_spec = self._builder.as_spec(condition)
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"Gate(cond={_name_of(cond_spec)}, {self._spec.name})",
        payload=lambda ch: Gate(cond_spec.to_node(), ch),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

when ¶

when(condition: Union[NodeSpec, Callable[[Any], Any]]) -> '_NodeWrapper'

Wrap with a When decorator - succeeds when condition is false (no-op)

Source code in src/mycorrhizal/rhizomorph/core.py

def when(self, condition: Union[NodeSpec, Callable[[Any], Any]]) -> "_NodeWrapper":
    """Wrap with a When decorator - succeeds when condition is false (no-op)"""
    cond_spec = self._builder.as_spec(condition)
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"When(cond={_name_of(cond_spec)}, {self._spec.name})",
        payload=lambda ch: When(cond_spec.to_node(), ch),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

succeeder ¶

succeeder() -> '_NodeWrapper'

Wrap with a Succeeder decorator - always returns SUCCESS

Source code in src/mycorrhizal/rhizomorph/core.py

def succeeder(self) -> "_NodeWrapper":
    """Wrap with a Succeeder decorator - always returns SUCCESS"""
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"Succeeder({self._spec.name})",
        payload=lambda ch: Succeeder(ch),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

failer ¶

failer() -> '_NodeWrapper'

Wrap with a Failer decorator - always returns FAILURE

Source code in src/mycorrhizal/rhizomorph/core.py

def failer(self) -> "_NodeWrapper":
    """Wrap with a Failer decorator - always returns FAILURE"""
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"Failer({self._spec.name})",
        payload=lambda ch: Failer(ch),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

inverter ¶

inverter() -> '_NodeWrapper'

Wrap with an Inverter decorator - inverts SUCCESS/FAILURE

Source code in src/mycorrhizal/rhizomorph/core.py

def inverter(self) -> "_NodeWrapper":
    """Wrap with an Inverter decorator - inverts SUCCESS/FAILURE"""
    wrapper_spec = NodeSpec(
        kind=NodeSpecKind.DECORATOR,
        name=f"Inverter({self._spec.name})",
        payload=lambda ch: Inverter(ch),
        children=[self._spec],
    )
    return _NodeWrapper(wrapper_spec, self._builder)

TreeBuilder ¶

TreeBuilder(name: str)

Programmatic builder for behavior trees.

Provides an imperative API for constructing behavior trees without decorators. Similar to NetBuilder in Hypha.

Example

builder = TreeBuilder("MyTree")

has_battery = builder.condition("has_battery", lambda bb: bb.battery > 20) move_forward = builder.action("move_forward", move_func)

patrol = builder.sequence(has_battery, move_forward, memory=True) tree = builder.root(patrol)

runner = Runner(tree, bb=blackboard) await runner.tick()

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(self, name: str):
    self._name = name
    self._root_spec: Optional[NodeSpec] = None

action ¶

action(name: str, func: Callable[..., Any]) -> _NodeWrapper

Create an action node.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name for the action	required
`func`	`Callable[..., Any]`	Action function (can be async) that takes (bb) and returns Status	required

Returns:

Type	Description
`_NodeWrapper`	_NodeWrapper for fluent chaining

Source code in src/mycorrhizal/rhizomorph/core.py

def action(self, name: str, func: Callable[..., Any]) -> _NodeWrapper:
    """
    Create an action node.

    Args:
        name: Name for the action
        func: Action function (can be async) that takes (bb) and returns Status

    Returns:
        _NodeWrapper for fluent chaining
    """
    spec = NodeSpec(kind=NodeSpecKind.ACTION, name=name, payload=func)
    return _NodeWrapper(spec, self)

condition ¶

condition(name: str, func: Callable[[Any], bool]) -> _NodeWrapper

Create a condition node.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name for the condition	required
`func`	`Callable[[Any], bool]`	Condition function that takes (bb) and returns bool	required

Returns:

Type	Description
`_NodeWrapper`	_NodeWrapper for fluent chaining

Source code in src/mycorrhizal/rhizomorph/core.py

def condition(self, name: str, func: Callable[[Any], bool]) -> _NodeWrapper:
    """
    Create a condition node.

    Args:
        name: Name for the condition
        func: Condition function that takes (bb) and returns bool

    Returns:
        _NodeWrapper for fluent chaining
    """
    spec = NodeSpec(kind=NodeSpecKind.CONDITION, name=name, payload=func)
    return _NodeWrapper(spec, self)

sequence ¶

sequence(*children: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]], memory: bool = True) -> NodeSpec

Create a sequence composite.

Parameters:

Name	Type	Description	Default
`*children`	`Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]`	Child nodes (NodeSpec, _NodeWrapper, or callables with node_spec)	`()`
`memory`	`bool`	Whether to remember position across ticks	`True`

Returns:

Type	Description
`NodeSpec`	NodeSpec for the sequence

Source code in src/mycorrhizal/rhizomorph/core.py

def sequence(
    self, *children: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]], memory: bool = True
) -> NodeSpec:
    """
    Create a sequence composite.

    Args:
        *children: Child nodes (NodeSpec, _NodeWrapper, or callables with node_spec)
        memory: Whether to remember position across ticks

    Returns:
        NodeSpec for the sequence
    """
    child_specs = [self.as_spec(c) for c in children]

    # Use a counter to generate unique factory names
    if not hasattr(self, "_factory_counter"):
        self._factory_counter = 0
    self._factory_counter += 1
    factory_id = self._factory_counter

    def _sequence_factory() -> Generator[Any, None, None]:
        for child in child_specs:
            yield child

    child_names = ", ".join(c.name for c in child_specs)
    name = f"Sequence({child_names})" if child_specs else "Sequence"

    # Set unique name including builder name to avoid false recursion detection
    _sequence_factory.__name__ = f"{self._name}._sequence_factory_{factory_id}"
    _sequence_factory.__qualname__ = f"{self._name}._sequence_factory_{factory_id}"

    return NodeSpec(
        kind=NodeSpecKind.SEQUENCE,
        name=name,
        payload={"factory": _sequence_factory, "memory": memory},
    )

selector ¶

selector(*children: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]], memory: bool = True) -> NodeSpec

Create a selector composite.

Parameters:

Name	Type	Description	Default
`*children`	`Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]`	Child nodes (NodeSpec, _NodeWrapper, or callables with node_spec)	`()`
`memory`	`bool`	Whether to remember position across ticks	`True`

Returns:

Type	Description
`NodeSpec`	NodeSpec for the selector

Source code in src/mycorrhizal/rhizomorph/core.py

def selector(
    self, *children: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]], memory: bool = True
) -> NodeSpec:
    """
    Create a selector composite.

    Args:
        *children: Child nodes (NodeSpec, _NodeWrapper, or callables with node_spec)
        memory: Whether to remember position across ticks

    Returns:
        NodeSpec for the selector
    """
    child_specs = [self.as_spec(c) for c in children]

    # Use a counter to generate unique factory names
    if not hasattr(self, "_factory_counter"):
        self._factory_counter = 0
    self._factory_counter += 1
    factory_id = self._factory_counter

    def _selector_factory() -> Generator[Any, None, None]:
        for child in child_specs:
            yield child

    child_names = ", ".join(c.name for c in child_specs)
    name = f"Selector({child_names})" if child_specs else "Selector"

    # Set unique name including builder name to avoid false recursion detection
    _selector_factory.__name__ = f"{self._name}._selector_factory_{factory_id}"
    _selector_factory.__qualname__ = f"{self._name}._selector_factory_{factory_id}"

    return NodeSpec(
        kind=NodeSpecKind.SELECTOR,
        name=name,
        payload={"factory": _selector_factory, "memory": memory},
    )

parallel ¶

parallel(*children: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]], success_threshold: int, failure_threshold: Optional[int] = None) -> NodeSpec

Create a parallel composite.

Parameters:

Name	Type	Description	Default
`*children`	`Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]`	Child nodes (NodeSpec, _NodeWrapper, or callables with node_spec)	`()`
`success_threshold`	`int`	Minimum number of children that must succeed	required
`failure_threshold`	`Optional[int]`	Minimum number of children that must fail (None = len(children))	`None`

Returns:

Type	Description
`NodeSpec`	NodeSpec for the parallel node

Source code in src/mycorrhizal/rhizomorph/core.py

def parallel(
    self,
    *children: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]],
    success_threshold: int,
    failure_threshold: Optional[int] = None,
) -> NodeSpec:
    """
    Create a parallel composite.

    Args:
        *children: Child nodes (NodeSpec, _NodeWrapper, or callables with node_spec)
        success_threshold: Minimum number of children that must succeed
        failure_threshold: Minimum number of children that must fail (None = len(children))

    Returns:
        NodeSpec for the parallel node
    """
    child_specs = [self.as_spec(c) for c in children]

    # Use a counter to generate unique factory names
    if not hasattr(self, "_factory_counter"):
        self._factory_counter = 0
    self._factory_counter += 1
    factory_id = self._factory_counter

    def _parallel_factory() -> Generator[Any, None, None]:
        for child in child_specs:
            yield child

    child_names = ", ".join(c.name for c in child_specs)
    name = f"Parallel({child_names})" if child_specs else "Parallel"

    # Set unique name including builder name to avoid false recursion detection
    _parallel_factory.__name__ = f"{self._name}._parallel_factory_{factory_id}"
    _parallel_factory.__qualname__ = f"{self._name}._parallel_factory_{factory_id}"

    return NodeSpec(
        kind=NodeSpecKind.PARALLEL,
        name=name,
        payload={
            "factory": _parallel_factory,
            "success_threshold": success_threshold,
            "failure_threshold": failure_threshold,
        },
    )

subtree ¶

subtree(tree: Union[SimpleNamespace, Callable[..., Any]]) -> NodeSpec

Mount another tree's root spec as a subtree.

Parameters:

Name	Type	Description	Default
`tree`	`Union[SimpleNamespace, Callable[..., Any]]`	A tree namespace created with @bt.tree, or a tree built by TreeBuilder	required

Returns:

Type	Description
`NodeSpec`	NodeSpec for the subtree

Example

DSL tree¶

engage_tree = Engage # @bt.tree decorated function

Programmatic tree¶

builder = TreeBuilder("Demo") builder.root(builder.sequence( builder.subtree(engage_tree), other_action ))

Two programmatic trees¶

engage_builder = TreeBuilder("Engage")

... build engage ...¶

engage_tree_func = engage_builder.build()

demo_builder = TreeBuilder("Demo") demo_builder.root(demo_builder.sequence( demo_builder.subtree(engage_tree_func), other_action ))

Source code in src/mycorrhizal/rhizomorph/core.py

def subtree(self, tree: Union[SimpleNamespace, Callable[..., Any]]) -> NodeSpec:
    """
    Mount another tree's root spec as a subtree.

    Args:
        tree: A tree namespace created with @bt.tree, or a tree built by TreeBuilder

    Returns:
        NodeSpec for the subtree

    Example:
        # DSL tree
        engage_tree = Engage  # @bt.tree decorated function

        # Programmatic tree
        builder = TreeBuilder("Demo")
        builder.root(builder.sequence(
            builder.subtree(engage_tree),
            other_action
        ))

        # Two programmatic trees
        engage_builder = TreeBuilder("Engage")
        # ... build engage ...
        engage_tree_func = engage_builder.build()

        demo_builder = TreeBuilder("Demo")
        demo_builder.root(demo_builder.sequence(
            demo_builder.subtree(engage_tree_func),
            other_action
        ))
    """
    # Check if it's a tree function from build()
    if hasattr(tree, "_spec"):
        root_spec = tree._spec
        tree_name = tree._name
        return NodeSpec(
            kind=NodeSpecKind.SUBTREE,
            name=tree_name,
            payload={"root": root_spec},
            children=[root_spec],
        )

    # Otherwise assume it's a SimpleNamespace from @bt.tree
    if not hasattr(tree, "root"):
        raise ValueError(
            f"Tree must have a 'root' attribute or be built via TreeBuilder.build(). "
            f"Did you forget @bt.root on a composite, or TreeBuilder.build()?"
        )

    root_spec = tree.root
    tree_name = getattr(tree, "_tree_name", root_spec.name)
    return NodeSpec(
        kind=NodeSpecKind.SUBTREE,
        name=tree_name,
        payload={"root": root_spec},
        children=[root_spec],
    )

as_spec ¶

as_spec(maybe: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]) -> NodeSpec

Convert various node types to a NodeSpec.

Handles: - NodeSpec: returns as-is - _NodeWrapper: extracts the spec - Callable with node_spec attribute: extracts the spec

Parameters:

Name	Type	Description	Default
`maybe`	`Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]`	NodeSpec, _NodeWrapper, or callable with node_spec	required

Returns:

Type	Description
`NodeSpec`	NodeSpec

Raises:

Type	Description
`TypeError`	If the input cannot be converted to a NodeSpec

Source code in src/mycorrhizal/rhizomorph/core.py

def as_spec(self, maybe: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]) -> NodeSpec:
    """
    Convert various node types to a NodeSpec.

    Handles:
    - NodeSpec: returns as-is
    - _NodeWrapper: extracts the spec
    - Callable with node_spec attribute: extracts the spec

    Args:
        maybe: NodeSpec, _NodeWrapper, or callable with node_spec

    Returns:
        NodeSpec

    Raises:
        TypeError: If the input cannot be converted to a NodeSpec
    """
    if isinstance(maybe, NodeSpec):
        return maybe
    if isinstance(maybe, _NodeWrapper):
        return maybe.spec

    spec = getattr(maybe, "node_spec", None)
    if spec is None:
        raise TypeError(
            f"{maybe!r} is not a BT node (missing node_spec attribute)."
        )
    return spec

root ¶

root(child: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]) -> Callable[..., Any]

Set the root of the tree and return a tree function.

The returned function is compatible with the Runner API and has a _spec attribute that holds the root NodeSpec.

Parameters:

Name	Type	Description	Default
`child`	`Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]`	Root node (NodeSpec, _NodeWrapper, or callable with node_spec)	required

Returns:

Type	Description
`Callable[..., Any]`	A callable function with _spec attribute for use with Runner

Example

builder = TreeBuilder("MyTree") action1 = builder.action("act1", func1) action2 = builder.action("act2", func2) root_seq = builder.sequence(action1, action2) tree = builder.root(root_seq)

runner = Runner(tree, bb=blackboard) await runner.tick()

Source code in src/mycorrhizal/rhizomorph/core.py

def root(self, child: Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]) -> Callable[..., Any]:
    """
    Set the root of the tree and return a tree function.

    The returned function is compatible with the Runner API and has a _spec attribute
    that holds the root NodeSpec.

    Args:
        child: Root node (NodeSpec, _NodeWrapper, or callable with node_spec)

    Returns:
        A callable function with _spec attribute for use with Runner

    Example:
        builder = TreeBuilder("MyTree")
        action1 = builder.action("act1", func1)
        action2 = builder.action("act2", func2)
        root_seq = builder.sequence(action1, action2)
        tree = builder.root(root_seq)

        runner = Runner(tree, bb=blackboard)
        await runner.tick()
    """
    root_spec = self.as_spec(child)
    self._root_spec = root_spec

    # Create a wrapper function that holds the spec (for Runner compatibility)
    def tree_func():
        pass

    tree_func._spec = root_spec
    tree_func._name = self._name
    tree_func.__name__ = self._name
    tree_func.root = root_spec  # Runner expects .root attribute

    # Add convenience methods for serialization and visualization
    def to_mermaid() -> str:
        """Generate Mermaid diagram for this tree."""
        from mycorrhizal.rhizomorph.core import _generate_mermaid

        class TreeNamespace:
            def __init__(self, root_spec: NodeSpec, name: str):
                self.root = root_spec
                self._tree_name = name

        tree_ns = TreeNamespace(root_spec, self._name)
        return _generate_mermaid(tree_ns)

    def to_dict() -> Dict[str, Any]:
        """Serialize tree to dictionary."""
        from mycorrhizal.rhizomorph.core import NodeSpecKind

        def spec_to_dict(spec: NodeSpec) -> Dict[str, Any]:
            result = {"kind": spec.kind.value, "name": spec.name}
            if spec.payload is not None:
                if callable(spec.payload):
                    result["payload"] = f"<function:{spec.payload.__name__}>"
                    if spec.kind == NodeSpecKind.DECORATOR:
                        result["decorator"] = spec.name.split("(")[0]
                elif isinstance(spec.payload, dict):
                    payload_dict = {}
                    for k, v in spec.payload.items():
                        if callable(v):
                            payload_dict[k] = f"<function:{v.__name__}>"
                        else:
                            payload_dict[k] = v
                    result["payload"] = payload_dict
                else:
                    result["payload"] = str(spec.payload)
            if spec.children:
                result["children"] = [spec_to_dict(child) for child in spec.children]
            return result

        return {"name": self._name, "root": spec_to_dict(root_spec)}

    tree_func.to_mermaid = to_mermaid
    tree_func.to_dict = to_dict

    return tree_func

build ¶

build(child: Optional[Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]] = None) -> Callable[..., Any]

Build and return the tree function.

This is an alias for root() that makes the intent clearer when you're at the end of building a tree.

Parameters:

Name	Type	Description	Default
`child`	`Optional[Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]]`	Root node (optional if already set via root())	`None`

Returns:

Type	Description
`Callable[..., Any]`	A callable function with _spec attribute for use with Runner

Example

builder = TreeBuilder("MyTree") tree = builder.build( builder.sequence( builder.action("act", func) ) )

Source code in src/mycorrhizal/rhizomorph/core.py

def build(self, child: Optional[Union[NodeSpec, _NodeWrapper, Callable[[Any], Any]]] = None) -> Callable[..., Any]:
    """
    Build and return the tree function.

    This is an alias for root() that makes the intent clearer when
    you're at the end of building a tree.

    Args:
        child: Root node (optional if already set via root())

    Returns:
        A callable function with _spec attribute for use with Runner

    Example:
        builder = TreeBuilder("MyTree")
        tree = builder.build(
            builder.sequence(
                builder.action("act", func)
            )
        )
    """
    if child is not None:
        return self.root(child)
    if self._root_spec is None:
        raise ValueError(f"TreeBuilder '{self._name}' has no root. Call root() or build(child) first.")
    return self.root(self._root_spec)

to_mermaid ¶

to_mermaid() -> str

Generate Mermaid flowchart diagram for this tree.

Returns:

Type	Description
`str`	Mermaid flowchart syntax as a string

Example

builder = TreeBuilder("MyTree") tree = builder.build(...) print(tree.to_mermaid())

Source code in src/mycorrhizal/rhizomorph/core.py

def to_mermaid(self) -> str:
    """
    Generate Mermaid flowchart diagram for this tree.

    Returns:
        Mermaid flowchart syntax as a string

    Example:
        builder = TreeBuilder("MyTree")
        tree = builder.build(...)
        print(tree.to_mermaid())
    """
    if self._root_spec is None:
        raise ValueError(f"TreeBuilder '{self._name}' has no root. Cannot generate Mermaid diagram.")

    # Create a mock tree namespace for the existing _generate_mermaid function
    class TreeNamespace:
        def __init__(self, root_spec: NodeSpec, name: str):
            self.root = root_spec
            self._tree_name = name

    tree_ns = TreeNamespace(self._root_spec, self._name)
    return _generate_mermaid(tree_ns)

to_dict ¶

to_dict() -> Dict[str, Any]

Serialize the tree to a dictionary.

Creates a JSON-serializable representation of the tree structure that can be saved to a file or transmitted over network.

Returns:

Type	Description
`Dict[str, Any]`	Dictionary representation of the tree

Example

builder = TreeBuilder("MyTree") tree = builder.build(...) tree_dict = tree.to_dict()

import json with open("tree.json", "w") as f: json.dump(tree_dict, f, indent=2)

Source code in src/mycorrhizal/rhizomorph/core.py

def to_dict(self) -> Dict[str, Any]:
    """
    Serialize the tree to a dictionary.

    Creates a JSON-serializable representation of the tree structure
    that can be saved to a file or transmitted over network.

    Returns:
        Dictionary representation of the tree

    Example:
        builder = TreeBuilder("MyTree")
        tree = builder.build(...)
        tree_dict = tree.to_dict()

        import json
        with open("tree.json", "w") as f:
            json.dump(tree_dict, f, indent=2)
    """
    if self._root_spec is None:
        raise ValueError(f"TreeBuilder '{self._name}' has no root. Cannot serialize.")

    def spec_to_dict(spec: NodeSpec) -> Dict[str, Any]:
        """Convert a NodeSpec to dictionary representation."""
        result = {
            "kind": spec.kind.value,
            "name": spec.name,
        }

        # Add payload if present (serializable only)
        if spec.payload is not None:
            # Check if payload is a lambda/function (not serializable)
            if callable(spec.payload):
                result["payload"] = f"<function:{spec.payload.__name__}>"

                # For decorators, include decorator name
                if spec.kind == NodeSpecKind.DECORATOR:
                    result["decorator"] = spec.name.split("(")[0]
            elif isinstance(spec.payload, dict):
                # For composites (sequence, selector, parallel)
                payload_dict = {}
                for k, v in spec.payload.items():
                    if callable(v):
                        payload_dict[k] = f"<function:{v.__name__}>"

                    else:
                        payload_dict[k] = v
                result["payload"] = payload_dict
            else:
                result["payload"] = str(spec.payload)

        # Add children recursively
        if spec.children:
            result["children"] = [spec_to_dict(child) for child in spec.children]

        return result

    return {
        "name": self._name,
        "root": spec_to_dict(self._root_spec),
    }

from_dict `classmethod` ¶

from_dict(data: Dict[str, Any], function_registry: Dict[str, Callable[..., Any]]) -> 'TreeBuilder'

Deserialize a tree from a dictionary with function registry.

Reconstructs a TreeBuilder from a serialized dictionary. You must provide a registry of functions to restore node behaviors.

Parameters:

Name	Type	Description	Default
`data`	`Dict[str, Any]`	Dictionary output from to_dict()	required
`function_registry`	`Dict[str, Callable[..., Any]]`	Dict mapping function names to callable functions	required

Returns:

Type	Description
`'TreeBuilder'`	A new TreeBuilder instance

Example

Define your functions¶

async def my_action(bb): return Status.SUCCESS

def my_condition(bb): return True

Build and save tree¶

builder = TreeBuilder("MyTree") action = builder.action("my_action", my_action) cond = builder.condition("my_condition", my_condition) tree = builder.build(builder.sequence(cond, action)) tree_dict = tree.to_dict()

Load tree later¶

registry = { "my_action": my_action, "my_condition": my_condition, } loaded_builder = TreeBuilder.from_dict(tree_dict, function_registry=registry) loaded_tree = loaded_builder.build()

Source code in src/mycorrhizal/rhizomorph/core.py

@classmethod
def from_dict(
    cls,
    data: Dict[str, Any],
    function_registry: Dict[str, Callable[..., Any]],
) -> "TreeBuilder":
    """
    Deserialize a tree from a dictionary with function registry.

    Reconstructs a TreeBuilder from a serialized dictionary.
    You must provide a registry of functions to restore node behaviors.

    Args:
        data: Dictionary output from to_dict()
        function_registry: Dict mapping function names to callable functions

    Returns:
        A new TreeBuilder instance

    Example:
        # Define your functions
        async def my_action(bb):
            return Status.SUCCESS

        def my_condition(bb):
            return True

        # Build and save tree
        builder = TreeBuilder("MyTree")
        action = builder.action("my_action", my_action)
        cond = builder.condition("my_condition", my_condition)
        tree = builder.build(builder.sequence(cond, action))
        tree_dict = tree.to_dict()

        # Load tree later
        registry = {
            "my_action": my_action,
            "my_condition": my_condition,
        }
        loaded_builder = TreeBuilder.from_dict(tree_dict, function_registry=registry)
        loaded_tree = loaded_builder.build()
    """
    name = data["name"]

    def dict_to_spec(spec_dict: Dict[str, Any]) -> NodeSpec:
        """Convert dictionary back to NodeSpec."""
        kind = NodeSpecKind(spec_dict["kind"])
        spec_name = spec_dict["name"]

        # Restore payload/function based on kind
        if kind in (NodeSpecKind.ACTION, NodeSpecKind.CONDITION):
            # Look up function in registry
            func_ref = spec_dict.get("payload", "")
            if isinstance(func_ref, str) and func_ref.startswith("<function:"):
                func_name = func_ref[10:-1]  # Extract name from <function:name>
                if func_name not in function_registry:
                    raise ValueError(f"Function '{func_name}' not found in registry. "
                                   f"Available: {list(function_registry.keys())}")
                payload = function_registry[func_name]
            else:
                payload = function_registry.get(func_ref)

            return NodeSpec(kind=kind, name=spec_name, payload=payload)

        elif kind == NodeSpecKind.SUBTREE:
            # Subtrees have special payload structure
            return NodeSpec(
                kind=kind,
                name=spec_name,
                payload=spec_dict.get("payload"),
            )

        elif kind in (NodeSpecKind.SEQUENCE, NodeSpecKind.SELECTOR, NodeSpecKind.PARALLEL):
            # Composites have factory in payload
            # We'll reconstruct from children
            payload = spec_dict.get("payload", {})
            children = [dict_to_spec(child) for child in spec_dict.get("children", [])]

            # Create factory function from children
            def make_factory(child_specs: List[NodeSpec]):
                def factory():
                    for child in child_specs:
                        yield child
                return factory

            # Reconstruct payload with factory
            payload["factory"] = make_factory(children)
            return NodeSpec(kind=kind, name=spec_name, payload=payload, children=children)

        elif kind == NodeSpecKind.DECORATOR:
            # Decorators - extract wrapper name and child
            decorator_name = spec_dict.get("decorator", spec_name.split("(")[0])
            children = [dict_to_spec(child) for child in spec_dict.get("children", [])]

            # Map decorator names to wrapper builders
            # This is limited - full reconstruction would require more metadata
            # For now, we'll create a placeholder
            payload = lambda ch: ch  # Identity function as placeholder
            return NodeSpec(kind=kind, name=spec_name, payload=payload, children=children)

        else:
            # Generic fallback
            children = [dict_to_spec(child) for child in spec_dict.get("children", [])]
            return NodeSpec(kind=kind, name=spec_name, payload=spec_dict.get("payload"), children=children)

    root_spec = dict_to_spec(data["root"])

    # Create new TreeBuilder
    builder = cls(name)
    builder._root_spec = root_spec

    return builder

Runner ¶

Runner(tree: SimpleNamespace, bb: BB, tb: Optional[Timebase] = None, exception_policy: ExceptionPolicy = LOG_AND_CONTINUE, trace: Optional[Logger] = None)

Bases: Generic[BB]

Runtime for executing behavior trees.

The Runner manages the execution of a behavior tree, handling tick calls, timebase management, and result processing.

Parameters:

Name	Type	Description	Default
`tree`	`SimpleNamespace`	A behavior tree namespace with a 'root' attribute (from @bt.tree)	required
`bb`	`BB`	Blackboard containing shared state	required
`tb`	`Optional[Timebase]`	Optional timebase for time management (defaults to MonotonicClock)	`None`
`exception_policy`	`ExceptionPolicy`	How to handle exceptions during tree execution	`LOG_AND_CONTINUE`
`trace`	`Optional[Logger]`	Optional logger instance for tracing action/condition execution	`None`

Methods:

Name	Description
`tick`	Execute one tick of the behavior tree
`tick_until_complete`	Run the tree until it returns a terminal status

Example

@bt.tree def MyTree(): @bt.root @bt.sequence def root(): yield check_condition yield do_work

runner = Runner(MyTree, bb=blackboard) result = await runner.tick_until_complete()

Tracing

import logging trace_logger = logging.getLogger("bt.trace") runner = Runner(MyTree, bb=blackboard, trace=trace_logger) result = await runner.tick_until_complete()

Logs: "action: module.do_work | SUCCESS"¶

"condition: module.check_condition | SUCCESS"¶

Source code in src/mycorrhizal/rhizomorph/core.py

def __init__(
    self,
    tree: SimpleNamespace,
    bb: BB,
    tb: Optional[Timebase] = None,
    exception_policy: ExceptionPolicy = ExceptionPolicy.LOG_AND_CONTINUE,
    trace: Optional[logging.Logger] = None,
) -> None:
    self.tree = tree
    self.bb: BB = bb
    self.tb = tb or MonotonicClock()
    self.exception_policy = exception_policy
    self.trace = trace

    if not hasattr(tree, "root"):
        raise ValueError("Tree namespace must have a 'root' attribute")

    self.root: Node[BB] = tree.root.to_node(
        exception_policy=exception_policy
    )

_clear_interface_view_cache ¶

_clear_interface_view_cache() -> None

Clear the interface view cache. Useful for testing.

Source code in src/mycorrhizal/rhizomorph/core.py

def _clear_interface_view_cache() -> None:
    """Clear the interface view cache. Useful for testing."""
    global _interface_view_cache
    _interface_view_cache.clear()
    # Also clear the compilation cache from common module
    _clear_compilation_cache()

_create_interface_view_if_needed ¶

_create_interface_view_if_needed(bb: Any, func: Callable) -> Any

Create a constrained view if the function has an interface type hint on its blackboard parameter.

This enables type-safe, constrained access to blackboard state based on interface definitions created with @blackboard_interface.

The function signature can use an interface type

async def my_action(bb: MyInterface) -> Status: # bb is automatically a constrained view return Status.SUCCESS

Parameters:

Name	Type	Description	Default
`bb`	`Any`	The blackboard instance	required
`func`	`Callable`	The function to check for interface type hints	required

Returns:

Type	Description
`Any`	Either the original blackboard or a constrained view based on interface metadata

Raises:

Type	Description
`TypeError`	If func is not callable or type hints are malformed
`AttributeError`	If type hints reference undefined types

Source code in src/mycorrhizal/rhizomorph/core.py

def _create_interface_view_if_needed(bb: Any, func: Callable) -> Any:
    """
    Create a constrained view if the function has an interface type hint on its
    blackboard parameter.

    This enables type-safe, constrained access to blackboard state based on
    interface definitions created with @blackboard_interface.

    The function signature can use an interface type:
        async def my_action(bb: MyInterface) -> Status:
            # bb is automatically a constrained view
            return Status.SUCCESS

    Args:
        bb: The blackboard instance
        func: The function to check for interface type hints

    Returns:
        Either the original blackboard or a constrained view based on interface metadata

    Raises:
        TypeError: If func is not callable or type hints are malformed
        AttributeError: If type hints reference undefined types
    """
    # Get compiled metadata (uses EAFP pattern internally)
    # Raises specific exceptions if compilation fails
    metadata = _get_compiled_metadata(func)

    # If handler has interface type hint, create constrained view
    if metadata.has_interface and metadata.interface_type:
        # EAFP: Try to get view from cache, create if not present
        cache_key = (id(bb), metadata.interface_type)
        try:
            return _interface_view_cache[cache_key]
        except KeyError:
            # Create view with pre-extracted interface metadata
            view = create_view_from_protocol(
                bb,
                metadata.interface_type,
                readonly_fields=metadata.readonly_fields
            )

            # Cache for reuse
            _interface_view_cache[cache_key] = view
            return view

    return bb

_supports_timebase ¶

_supports_timebase(func: Callable) -> bool

Check if a function accepts a 'tb' parameter.

Source code in src/mycorrhizal/rhizomorph/core.py

def _supports_timebase(func: Callable) -> bool:
    """Check if a function accepts a 'tb' parameter."""
    try:
        sig = inspect.signature(func)
        return 'tb' in sig.parameters
    except (ValueError, TypeError):
        return False

_call_node_function `async` ¶

_call_node_function(func: Callable, bb: Any, tb: Timebase, supports_timebase: bool) -> Any

Call a node function with appropriate parameters based on its signature.

If the function has an interface type hint on its 'bb' parameter, a constrained view will be created automatically to enforce access control.

Parameters:

Name	Type	Description	Default
`func`	`Callable`	The function to call	required
`bb`	`Any`	The blackboard	required
`tb`	`Timebase`	The timebase	required
`supports_timebase`	`bool`	Cached flag indicating if func accepts 'tb' parameter	required

Source code in src/mycorrhizal/rhizomorph/core.py

async def _call_node_function(func: Callable, bb: Any, tb: Timebase, supports_timebase: bool) -> Any:
    """
    Call a node function with appropriate parameters based on its signature.

    If the function has an interface type hint on its 'bb' parameter, a
    constrained view will be created automatically to enforce access control.

    Args:
        func: The function to call
        bb: The blackboard
        tb: The timebase
        supports_timebase: Cached flag indicating if func accepts 'tb' parameter
    """
    # Create interface view if function has interface type hint
    bb_to_pass = _create_interface_view_if_needed(bb, func)

    if supports_timebase:
        if inspect.iscoroutinefunction(func):
            return await func(bb=bb_to_pass, tb=tb)
        else:
            return func(bb=bb_to_pass, tb=tb)
    else:
        if inspect.iscoroutinefunction(func):
            return await func(bb=bb_to_pass)
        else:
            return func(bb=bb_to_pass)

_fully_qualified_name ¶

_fully_qualified_name(func: Callable[..., Any]) -> str

Get the fully qualified name of a function.

Returns module.function_name if the function has a module, otherwise returns just the function name.

Source code in src/mycorrhizal/rhizomorph/core.py

def _fully_qualified_name(func: Callable[..., Any]) -> str:
    """
    Get the fully qualified name of a function.

    Returns module.function_name if the function has a module,
    otherwise returns just the function name.
    """
    name = func.__name__
    module = getattr(func, "__module__", None)
    if module:
        # Handle nested functions by trying to get qualname
        qualname = getattr(func, "__qualname__", None)
        if qualname and qualname != name:
            return f"{module}.{qualname}"
        return f"{module}.{name}"
    return name

_bt_expand_children ¶

_bt_expand_children(factory: Callable[..., Generator[Any, None, None]], expansion_stack: Optional[Set[str]] = None) -> List[NodeSpec]

Execute a composite factory to get child specs.

Parameters:

Name	Type	Description	Default
`factory`	`Callable[..., Generator[Any, None, None]]`	The generator function that yields child specs	required
`expansion_stack`	`Optional[Set[str]]`	Stack of factory names to detect recursion	`None`

Source code in src/mycorrhizal/rhizomorph/core.py

def _bt_expand_children(
    factory: Callable[..., Generator[Any, None, None]],
    expansion_stack: Optional[Set[str]] = None,
) -> List[NodeSpec]:
    """
    Execute a composite factory to get child specs.

    Args:
        factory: The generator function that yields child specs
        expansion_stack: Stack of factory names to detect recursion
    """
    if expansion_stack is None:
        expansion_stack = set()

    factory_name = _name_of(factory)
    if factory_name in expansion_stack:
        chain = " -> ".join(expansion_stack) + f" -> {factory_name}"
        raise RecursionError(
            f"Recursive behavior tree structure detected: {chain}\n"
            f"Behavior trees must be acyclic. Consider using:\n"
            f"  - A Selector with memory to iterate through options\n"
            f"  - A Sequence with conditions to control flow\n"
            f"  - A Retry decorator for repeated attempts\n"
            f"  - State in the blackboard to track progress"
        )

    expansion_stack = expansion_stack.copy()
    expansion_stack.add(factory_name)

    gen = factory()

    if not inspect.isgenerator(gen):
        raise TypeError(
            f"Composite factory {factory.__name__} must be a generator (use 'yield')."
        )

    out: List[NodeSpec] = []
    for yielded in gen:
        if isinstance(yielded, (list, tuple)):
            for y in yielded:
                spec = bt.as_spec(y)
                if (
                    hasattr(spec, "payload")
                    and isinstance(spec.payload, dict)
                    and "factory" in spec.payload
                ):
                    spec._expansion_stack = expansion_stack  # type: ignore
                out.append(spec)
            continue
        spec = bt.as_spec(yielded)
        if (
            hasattr(spec, "payload")
            and isinstance(spec.payload, dict)
            and "factory" in spec.payload
        ):
            spec._expansion_stack = expansion_stack  # type: ignore
        out.append(spec)

    for spec in out:
        if spec.kind in (
            NodeSpecKind.SEQUENCE,
            NodeSpecKind.SELECTOR,
            NodeSpecKind.PARALLEL,
        ) and hasattr(spec, "_expansion_stack"):
            child_factory = spec.payload["factory"]
            _bt_expand_children(child_factory, spec._expansion_stack)  # type: ignore

    return out

_generate_mermaid ¶

_generate_mermaid(tree: SimpleNamespace) -> str

Render a static structure graph for a tree namespace.

Source code in src/mycorrhizal/rhizomorph/core.py

def _generate_mermaid(tree: SimpleNamespace) -> str:
    """
    Render a static structure graph for a tree namespace.
    """
    if not hasattr(tree, "root"):
        raise ValueError("Tree namespace must have a 'root' attribute")

    lines: List[str] = ["flowchart TD"]
    node_ids: Dict[NodeSpec, str] = {}
    counter = 0

    def nid(spec: NodeSpec) -> str:
        nonlocal counter
        if spec in node_ids:
            return node_ids[spec]
        counter += 1
        node_ids[spec] = f"N{counter}"
        return node_ids[spec]

    def label(spec: NodeSpec) -> str:
        match spec.kind:
            case NodeSpecKind.ACTION | NodeSpecKind.CONDITION:
                return f"{spec.kind.value.upper()}<br/>{spec.name}"
            case NodeSpecKind.SEQUENCE | NodeSpecKind.SELECTOR | NodeSpecKind.PARALLEL:
                return f"{spec.kind.value.capitalize()}<br/>{spec.name}"
            case NodeSpecKind.DECORATOR:
                return f"Decor<br/>{spec.name}"
            case NodeSpecKind.SUBTREE:
                return f"Subtree<br/>{spec.name}"
            case NodeSpecKind.MATCH:
                return f"Match<br/>{spec.name}"
            case NodeSpecKind.DO_WHILE:
                return f"DoWhile<br/>{spec.name}"
            case NodeSpecKind.TRY_CATCH:
                return f"TryCatch<br/>{spec.name}"
            case _:
                return spec.name

    def ensure_children(spec: NodeSpec) -> List[NodeSpec]:
        match spec.kind:
            case NodeSpecKind.SEQUENCE | NodeSpecKind.SELECTOR | NodeSpecKind.PARALLEL:
                factory = spec.payload["factory"]
                spec.children = _bt_expand_children(factory)
            case NodeSpecKind.SUBTREE:
                subtree_root = spec.payload["root"]
                spec.children = [subtree_root]
            case NodeSpecKind.MATCH:
                case_specs: List[CaseSpec] = spec.payload["cases"]
                spec.children = [cs.child for cs in case_specs]
            case NodeSpecKind.DO_WHILE:
                # Children already set (just the body), but we also want to show condition
                cond_spec = spec.payload["condition"]
                spec.children = [cond_spec] + spec.children
            case NodeSpecKind.TRY_CATCH:
                # Children already set in _TryCatchBuilder
                pass
        return spec.children

    def walk(spec: NodeSpec) -> None:
        this_id = nid(spec)
        shape = (
            "((%s))"
            if spec.kind in (NodeSpecKind.ACTION, NodeSpecKind.CONDITION)
            else '["%s"]'
        ) % label(spec)
        lines.append(f"  {this_id}{shape}")

        children = ensure_children(spec)

        if spec.kind == NodeSpecKind.MATCH:
            case_specs: List[CaseSpec] = spec.payload["cases"]
            for case_spec, child in zip(case_specs, children):
                child_id = nid(child)
                edge_label = case_spec.label.replace('"', "'")
                lines.append(f'  {this_id} -->|"{edge_label}"| {child_id}')
                walk(child)
        elif spec.kind == NodeSpecKind.DO_WHILE:
            # First child is condition, second is body
            cond_id = nid(children[0])
            lines.append(f'  {this_id} -->|"condition"| {cond_id}')
            walk(children[0])
            if len(children) > 1:
                body_id = nid(children[1])
                lines.append(f'  {this_id} -->|"body"| {body_id}')
                walk(children[1])
        elif spec.kind == NodeSpecKind.TRY_CATCH:
            # First child is try, second is catch
            try_id = nid(children[0])
            lines.append(f'  {this_id} -->|"try"| {try_id}')
            walk(children[0])
            if len(children) > 1:
                catch_id = nid(children[1])
                lines.append(f'  {this_id} -->|"catch"| {catch_id}')
                walk(children[1])
        else:
            for i, child in enumerate(children, start=1):
                child_id = nid(child)
                if spec.kind == NodeSpecKind.PARALLEL:
                    lines.append(f'  {this_id} -->|"P"| {child_id}')
                elif spec.kind in (NodeSpecKind.SEQUENCE, NodeSpecKind.SELECTOR):
                    lines.append(f'  {this_id} -->|"{i}"| {child_id}')
                else:
                    lines.append(f"  {this_id} --> {child_id}")
                walk(child)

    walk(tree.root)
    return "\n".join(lines)

Utilities¶

mycorrhizal.rhizomorph.util ¶

Rhizomorph Behavior Tree Utilities

Utility functions for behavior tree operations like Mermaid diagram generation.

to_mermaid ¶

to_mermaid(tree: SimpleNamespace) -> str

Generate Mermaid diagram from a behavior tree namespace.

This is a utility wrapper that accesses the tree's internal to_mermaid method. Use this when you have a tree namespace and want to generate a diagram.

Parameters:

Name	Type	Description	Default
`tree`	`SimpleNamespace`	The behavior tree namespace (created with @bt.tree decorator)	required

Returns:

Type	Description
`str`	Mermaid diagram as a string

Example

from mycorrhizal.rhizomorph.core import bt
from mycorrhizal.rhizomorph.util import to_mermaid

@bt.tree
def MyTree():
    # ... define tree ...
    pass

# Generate diagram
mermaid_diagram = to_mermaid(MyTree)
print(mermaid_diagram)

Source code in src/mycorrhizal/rhizomorph/util.py

def to_mermaid(tree: SimpleNamespace) -> str:
    """
    Generate Mermaid diagram from a behavior tree namespace.

    This is a utility wrapper that accesses the tree's internal to_mermaid method.
    Use this when you have a tree namespace and want to generate a diagram.

    Args:
        tree: The behavior tree namespace (created with @bt.tree decorator)

    Returns:
        Mermaid diagram as a string

    Example:
        ```python
        from mycorrhizal.rhizomorph.core import bt
        from mycorrhizal.rhizomorph.util import to_mermaid

        @bt.tree
        def MyTree():
            # ... define tree ...
            pass

        # Generate diagram
        mermaid_diagram = to_mermaid(MyTree)
        print(mermaid_diagram)
        ```
    """
    # The tree namespace has a to_mermaid lambda attached by @bt.tree
    if not hasattr(tree, "to_mermaid"):
        raise ValueError(
            "Tree namespace must have a 'to_mermaid' attribute. "
            "Did you create this tree with @bt.tree decorator?"
        )
    return tree.to_mermaid()

Rhizomorph Behavior Trees API Reference¶

Core Module¶

mycorrhizal.rhizomorph.core ¶

Fluent wrapper chaining¶

Status ¶

ExceptionPolicy ¶

RecursionError ¶

Node ¶

Action ¶

Condition ¶

Sequence ¶

Selector ¶

Parallel ¶

RateLimit ¶

Gate ¶

When ¶

If feature_enabled is false, returns SUCCESS and sequence continues¶

Match ¶

_matches ¶

DoWhile ¶

TryCatch ¶

_DefaultCase ¶

CaseSpec dataclass ¶

_WrapperChain ¶

when ¶

If is_enabled is False, returns SUCCESS and sequence continues¶

__call__ ¶

_CaseBuilder ¶

_MatchBuilder ¶

_DoWhileBuilder ¶

_BT ¶

action ¶

condition ¶

sequence ¶

_sequence_from_children ¶

_sequence_impl ¶

selector ¶

_selector_from_children ¶

_selector_impl ¶

parallel ¶

when ¶

match ¶

case ¶

defaultcase ¶

do_while ¶

try_catch ¶

Define try and catch blocks with explicit memory settings¶

Use in the tree¶

subtree ¶

tree ¶

root ¶

as_spec ¶

_NodeWrapper ¶

spec property ¶

timeout ¶

retry ¶

ratelimit ¶

gate ¶

when ¶

succeeder ¶

failer ¶

inverter ¶

TreeBuilder ¶

action ¶

condition ¶

sequence ¶

selector ¶

parallel ¶

subtree ¶

DSL tree¶

Programmatic tree¶

Two programmatic trees¶

... build engage ...¶

as_spec ¶

root ¶

build ¶

to_mermaid ¶

to_dict ¶

from_dict classmethod ¶

Define your functions¶

CaseSpec `dataclass` ¶

call ¶

spec `property` ¶

from_dict `classmethod` ¶

_call_node_function `async` ¶