Behaviors as Data

# labs/behavior_engine.py
from dataclasses import dataclass


@dataclass
class Behavior:
    name: str
    trigger: str
    action: str
    priority: int = 1
    enabled: bool = True
    description: str = ""


if __name__ == "__main__":
    halt = Behavior("Emergency halt", "emergency", "halt.all", priority=5,
                    description="Stop every moving part, then explain.")
    print(f"{halt.name}: {halt.trigger!r} -> {halt.action!r} @ priority {halt.priority}")
    print("enabled:", halt.enabled)

$ uv run python -m labs.behavior_engine
Emergency halt: 'emergency' -> 'halt.all' @ priority 5
enabled: True

No methods, on purpose. A Behavior holds only what should happen: a trigger to look for, the name of an action, a number saying how badly it wants to run, a flag, and a description for whoever reads the table later. Zero logic is the property that matters, because it lets a row be typed by hand, loaded from JSON, or produced by the model without the engine noticing. The defaults mean the common case takes three arguments and the urgent case four. !r asks for repr() instead of str(), so the values arrive with their quotes.

from typing import Callable


class BehaviorEngine:
    def __init__(self, behaviors: list[Behavior] | None = None) -> None:
        self.behaviors: list[Behavior] = list(behaviors or [])
        self.handlers: dict[str, Callable[[dict], None]] = {}

    def evaluate(self, context: dict) -> list[Behavior]:
        haystack = str(context).lower()
        matched = [b for b in self.behaviors if b.enabled and b.trigger in haystack]
        return sorted(matched, key=lambda b: -b.priority)


BEHAVIORS = [
    Behavior("Greet on wake", "wake_word", "greet.user", priority=2),
    Behavior("Emergency halt", "emergency", "halt.all", priority=5),
    Behavior("Status on request", "status", "status.report"),
    Behavior("Night snark", "wake_word", "snark.reply", priority=3, enabled=False),
]

if __name__ == "__main__":
    engine = BehaviorEngine(BEHAVIORS)
    context = {"event": "wake_word", "text": "GLaDOS emergency stop", "hour": 23}
    for b in engine.evaluate(context):
        print(f"{b.priority}  {b.name}")

$ uv run python -m labs.behavior_engine
5  Emergency halt
2  Greet on wake

Read the output against the table and the whole design is visible in two lines. Two behaviors matched, because both "wake_word" and "emergency" appear in the stringified context, and both are allowed to run. "status" is nowhere in this context, so that row never enters the list. Night snark does match, and at priority 3 would have landed in the middle, but its enabled flag is false and the filter drops it before priority is consulted. Then the sort: negating the priority turns Python's ascending sorted into a descending one, so the largest number leads. Filtering first and sorting the survivors costs less than sorting everything, and puts the two decisions on two lines you can read one at a time.

    def register(self, action: str, handler: Callable[[dict], None]) -> None:
        self.handlers[action] = handler

    def execute(self, behaviors: list[Behavior], context: dict) -> list[str]:
        ran = []
        for b in behaviors:
            handler = self.handlers.get(b.action)
            if handler is None:
                continue
            handler(context)
            ran.append(b.name)
        return ran


if __name__ == "__main__":
    engine = BehaviorEngine(BEHAVIORS)
    engine.register("greet.user", lambda ctx: print("[ACTION] Oh. It's you."))
    engine.register("status.report", lambda ctx: print("[ACTION] All systems nominal."))

    context = {"event": "wake_word", "text": "GLaDOS emergency stop"}
    matched = engine.evaluate(context)
    ran = engine.execute(matched, context)
    print("matched:", [b.name for b in matched])
    print("ran:    ", ran)
    print("no handler:", [b.action for b in matched if b.action not in engine.handlers])

$ uv run python -m labs.behavior_engine
[ACTION] Oh. It's you.
matched: ['Emergency halt', 'Greet on wake']
ran:     ['Greet on wake']
no handler: ['halt.all']

The behavior knows the string "halt.all" and nothing else; execute is where that string becomes a function, or fails to. Nobody registered a handler, so .get returns None, the row is skipped, and the reactions that do have handlers run normally. That silence has a cost. The upside: a table can name actions the current build has not implemented, so reactions for a machine with an arm can sit in the repository months before there is an arm. The downside: a mistyped action looks exactly like a not-yet-implemented one. So the engine stays quiet and execute hands back the names it ran, and comparing that list against matched puts the gap one line away, on demand, instead of in a log nobody reads.

# labs/behavior_engine.py — full file
from dataclasses import dataclass
from typing import Callable

from labs.event_bus import bus


@dataclass
class Behavior:
    name: str
    trigger: str
    action: str
    priority: int = 1
    enabled: bool = True
    description: str = ""


class BehaviorEngine:
    def __init__(self, behaviors: list[Behavior] | None = None) -> None:
        self.behaviors: list[Behavior] = list(behaviors or [])
        self.handlers: dict[str, Callable[[dict], None]] = {}

    def register(self, action: str, handler: Callable[[dict], None]) -> None:
        self.handlers[action] = handler

    def evaluate(self, context: dict) -> list[Behavior]:
        haystack = str(context).lower()
        matched = [b for b in self.behaviors if b.enabled and b.trigger in haystack]
        return sorted(matched, key=lambda b: -b.priority)

    def execute(self, behaviors: list[Behavior], context: dict) -> list[str]:
        ran = []
        for b in behaviors:
            handler = self.handlers.get(b.action)
            if handler is None:
                continue
            handler(context)
            ran.append(b.name)
        return ran

    def react(self, context: dict) -> list[str]:
        return self.execute(self.evaluate(context), context)


BEHAVIORS = [
    Behavior("Emergency halt", "emergency", "halt.all", priority=5,
             description="Stop every moving part, then explain."),
    Behavior("Battery warning", "battery_low", "warn.power", priority=4,
             description="Announce the level before it becomes an outage."),
    Behavior("Greet on wake", "wake_word", "greet.user", priority=2),
    Behavior("Status on request", "status", "status.report",
             description="Report health when asked."),
    Behavior("Night snark", "wake_word", "snark.reply", priority=3, enabled=False,
             description="Extra hostility. Off by default."),
]


def main() -> None:
    engine = BehaviorEngine(BEHAVIORS)
    engine.register("halt.all", lambda ctx: print("[ACTION] Everything stops. You are welcome."))
    engine.register("warn.power", lambda ctx: print(f"[ACTION] Battery at {ctx.get('battery', '?')} percent."))
    engine.register("greet.user", lambda ctx: print("[ACTION] Oh. It's you."))
    engine.register("status.report", lambda ctx: print("[ACTION] All systems nominal."))

    bus.subscribe("speech_detected", engine.react)
    bus.subscribe("battery_low", engine.react)

    traffic = [
        ("speech_detected", {"event": "wake_word", "text": "GLaDOS, status", "battery": 82}),
        ("battery_low", {"event": "battery_low", "battery": 7}),
        ("speech_detected", {"event": "wake_word", "text": "GLaDOS emergency stop", "battery": 7}),
    ]
    for name, data in traffic:
        print(f"--- {name}")
        bus.publish(name, data)


if __name__ == "__main__":
    main()

$ uv run python -m labs.behavior_engine
--- speech_detected
[ACTION] Oh. It's you.
[ACTION] All systems nominal.
--- battery_low
[ACTION] Battery at 7 percent.
--- speech_detected
[ACTION] Everything stops. You are welcome.
[ACTION] Oh. It's you.

react is the whole public surface: hand it a context, it evaluates, executes, and reports what it ran. That signature is what bus.subscribe wants, a callable taking one argument, so the engine plugs into chapter 15's bus with no adapter. Notice what the bus does with the returned list: nothing. Publishing is fire-and-forget, so react is written to be useful when called directly too. Then read the second event. Nobody said a word and no microphone was involved, yet a reaction fired, because a context dictionary from a battery monitor is the same kind of input as one built from a transcript.