Interactions¶

When an agent performs PickupDrop or Toggle, the step pipeline runs a list of interaction functions — called branches — to determine what happens. Each branch independently decides whether it should fire and what state changes to apply.

Step Pipeline Context¶

Interactions run after tick functions and movement:

tick → movement → interactions → observations → rewards

PickupDrop ("pick up / put down") and Toggle ("activate / use") are semantically different actions. Built-in branches for pickup, drop, and place only fire on PickupDrop. Custom branches can check ctx.action against ctx.action_id to distinguish which action the agent chose.

Branch Function Signature¶

Every interaction function has the same interface:

def my_branch(ctx: InteractionContext) -> tuple[bool, dict]:
    should_apply = ...  # bool: should this interaction fire?
    changes = {...}     # dict: array name -> updated array
    return should_apply, changes

• should_apply — boolean. If false, changes is ignored.
• changes — dict mapping state array names (e.g. "object_type_map", "agent_inv") to their new values.

The pipeline applies changes via xp.where: for each key in changes, the new value replaces the old one where should_apply is true. When multiple branches fire for the same agent, later branches overwrite earlier ones for overlapping keys.

InteractionContext¶

The pipeline builds a frozen InteractionContext dataclass before calling branches. Each agent gets its own context per step.

InteractionContext(can_interact: object, action: object, action_id: object, facing_row: object, facing_col: object, facing_type: object, agent_index: object, held_item: object, agent_inv: object, object_type_map: object, object_state_map: object, type_ids: dict = dict(), extra: dict = dict(), _tables: dict = dict())

can_interact      bool -- agent performed PickupDrop or Toggle and cell ahead is clear
action            int  -- raw action index this agent chose this step
action_id         ActionID -- indices for all actions in this env
facing_row        int  -- row of the cell this agent faces
facing_col        int  -- col of the cell this agent faces
facing_type       int  -- type ID of object in faced cell (0 = empty)
agent_index       int  -- which agent (0, 1, ...)
held_item         int  -- type ID of item the agent holds (-1 = empty)
agent_inv         (n_agents, 1) int32 -- all inventories
object_type_map   (H, W) int32 -- the grid
object_state_map  (H, W) int32 -- per-cell state
type_ids          dict -- {"goal": 5, "wall": 1, ...}

Any extra_state key is accessible by name with the scope prefix
stripped.  ``state.extra_state["global.goals_collected"]`` becomes
``ctx.goals_collected``.

Standard fields¶

Extra-state fields¶

Arrays declared by components via extra_state_schema are accessible directly as attributes on ctx. The scope prefix is stripped automatically — if a component declares goals_collected, access it as ctx.goals_collected.

Lookup tables¶

Built-in branches use guard tables to decide whether an action is allowed for a given object type and held item. These are available on the context:

• ctx.CAN_PICKUP — which object types can be picked up
• ctx.PICKUP_FROM_GUARD — which (object, held-item) pairs allow pickup-from
• ctx.PLACE_ON_GUARD — which (object, held-item) pairs allow place-on

Built-in Branches¶

The autowire system assembles the branch list automatically from registered components. These are the built-in branches, in the order they run:

Container and consume branches are only included when the environment has container or consume objects. Autowire handles this automatically.

Helper Functions¶

Import from cogrid.core.pipeline.context:

clear_facing_cell(ctx: InteractionContext) -> ArrayLike

set_facing_cell(ctx: InteractionContext, type_id: ArrayLike) -> ArrayLike

pickup_from_facing_cell(ctx: InteractionContext) -> tuple[ArrayLike, ArrayLike]

place_in_facing_cell(ctx: InteractionContext) -> tuple[ArrayLike, ArrayLike]

give_item(ctx: InteractionContext, type_id: ArrayLike) -> ArrayLike

empty_hands(ctx: InteractionContext) -> ArrayLike

increment(array: ArrayLike, index: int) -> ArrayLike

find_facing_instance(positions: ArrayLike, facing_row: int, facing_col: int) -> tuple[ArrayLike, ArrayLike]

Writing a Custom Interaction¶

Define a function with the branch signature and pass it in the config. Here is an example that removes a goal when an agent picks it up:

from cogrid.core.pipeline.context import clear_facing_cell, increment


def collect_goal(ctx):
    """Remove a goal when the agent picks it up."""
    goal_id = ctx.type_ids["goal"]
    is_pickup = ctx.action == ctx.action_id.pickup_drop
    should_apply = ctx.can_interact & is_pickup & (ctx.facing_type == goal_id)
    changes = {
        "object_type_map": clear_facing_cell(ctx),
        "goals_collected": increment(ctx.goals_collected, ctx.agent_index),
    }
    return should_apply, changes

Wire it into the environment config:

goal_config = {
    ...
    "interactions": [collect_goal],
}

User-provided interactions run before auto-discovered ones, so they take priority for overlapping keys.

Composition¶

The autowire system reads container and consume metadata from registered objects and calls compose_interactions() internally. This returns:

• An extras function that pre-computes container context (contents, timers, recipe matching) and attaches it to the interaction context.
• A branch list assembled conditionally — container branches are included only when containers exist, consume branches only when consume surfaces exist.

User-provided "interactions" from the config are prepended to the auto-discovered list, so they run first.