chore: applied feedback and merged dev into branch

ref: N25B-355

.gitlab/merge_request_templates/default.md

@@ -0,0 +1,9 @@
+%{first_multiline_commit_description}
+
+To verify:
+
+- [ ] Style checks pass
+- [ ] Pipeline (tests) pass
+- [ ] Documentation is up to date
+- [ ] Tests are up to date (new code is covered)
+- [ ] ...

pyproject.toml

@@ -15,7 +15,6 @@ dependencies = [
   "pydantic>=2.12.0",
   "pydantic-settings>=2.11.0",
   "python-json-logger>=4.0.0",
-  "python-slugify>=8.0.4",
   "pyyaml>=6.0.3",
   "pyzmq>=27.1.0",
   "silero-vad>=6.0.0",

src/control_backend/agents/actuation/robot_gesture_agent.py

@@ -28,6 +28,7 @@ class RobotGestureAgent(BaseAgent):
     address = ""
     bind = False
     gesture_data = []
+    single_gesture_data = []
 
     def __init__(
         self,
@@ -35,8 +36,10 @@ class RobotGestureAgent(BaseAgent):
         address=settings.zmq_settings.ri_command_address,
         bind=False,
         gesture_data=None,
+        single_gesture_data=None,
     ):
         self.gesture_data = gesture_data or []
+        self.single_gesture_data = single_gesture_data or []
         super().__init__(name)
         self.address = address
         self.bind = bind
@@ -99,7 +102,13 @@ class RobotGestureAgent(BaseAgent):
                         gesture_command.data,
                     )
                     return
-
+            elif gesture_command.endpoint == RIEndpoint.GESTURE_SINGLE:
+                if gesture_command.data not in self.single_gesture_data:
+                    self.logger.warning(
+                        "Received gesture '%s' which is not in available gestures. Early returning",
+                        gesture_command.data,
+                    )
+                    return
             await self.pubsocket.send_json(gesture_command.model_dump())
         except Exception:
             self.logger.exception("Error processing internal message.")

src/control_backend/agents/bdi/agentspeak_ast.py

@@ -1,273 +0,0 @@
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from enum import StrEnum
-
-
-class AstNode(ABC):
-    """
-    Abstract base class for all elements of an AgentSpeak program.
-    """
-
-    @abstractmethod
-    def _to_agentspeak(self) -> str:
-        """
-        Generates the AgentSpeak code string.
-        """
-        pass
-
-    def __str__(self) -> str:
-        return self._to_agentspeak()
-
-
-class AstExpression(AstNode, ABC):
-    """
-    Intermediate class for anything that can be used in a logical expression.
-    """
-
-    def __and__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.AND, _coalesce_expr(other))
-
-    def __or__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.OR, _coalesce_expr(other))
-
-    def __invert__(self) -> AstLogicalExpression:
-        if isinstance(self, AstLogicalExpression):
-            self.negated = not self.negated
-            return self
-        return AstLogicalExpression(self, negated=True)
-
-
-type ExprCoalescible = AstExpression | str | int | float
-
-
-def _coalesce_expr(value: ExprCoalescible) -> AstExpression:
-    if isinstance(value, AstExpression):
-        return value
-    if isinstance(value, str):
-        return AstString(value)
-    if isinstance(value, (int, float)):
-        return AstNumber(value)
-    raise TypeError(f"Cannot coalesce type {type(value)} into an AstTerm.")
-
-
-@dataclass
-class AstTerm(AstExpression, ABC):
-    """
-    Base class for terms appearing inside literals.
-    """
-
-    def __ge__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.GREATER_EQUALS, _coalesce_expr(other))
-
-    def __gt__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.GREATER_THAN, _coalesce_expr(other))
-
-    def __le__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.LESS_EQUALS, _coalesce_expr(other))
-
-    def __lt__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.LESS_THAN, _coalesce_expr(other))
-
-    def __eq__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.EQUALS, _coalesce_expr(other))
-
-    def __ne__(self, other: ExprCoalescible) -> AstBinaryOp:
-        return AstBinaryOp(self, BinaryOperatorType.NOT_EQUALS, _coalesce_expr(other))
-
-
-@dataclass
-class AstAtom(AstTerm):
-    """
-    Grounded expression in all lowercase.
-    """
-
-    value: str
-
-    def _to_agentspeak(self) -> str:
-        return self.value.lower()
-
-
-@dataclass
-class AstVar(AstTerm):
-    """
-    Ungrounded variable expression. First letter capitalized.
-    """
-
-    name: str
-
-    def _to_agentspeak(self) -> str:
-        return self.name.capitalize()
-
-
-@dataclass
-class AstNumber(AstTerm):
-    value: int | float
-
-    def _to_agentspeak(self) -> str:
-        return str(self.value)
-
-
-@dataclass
-class AstString(AstTerm):
-    value: str
-
-    def _to_agentspeak(self) -> str:
-        return f'"{self.value}"'
-
-
-@dataclass
-class AstLiteral(AstTerm):
-    functor: str
-    terms: list[AstTerm] = field(default_factory=list)
-
-    def _to_agentspeak(self) -> str:
-        if not self.terms:
-            return self.functor
-        args = ", ".join(map(str, self.terms))
-        return f"{self.functor}({args})"
-
-
-class BinaryOperatorType(StrEnum):
-    AND = "&"
-    OR = "|"
-    GREATER_THAN = ">"
-    LESS_THAN = "<"
-    EQUALS = "=="
-    NOT_EQUALS = "\\=="
-    GREATER_EQUALS = ">="
-    LESS_EQUALS = "<="
-
-
-@dataclass
-class AstBinaryOp(AstExpression):
-    left: AstExpression
-    operator: BinaryOperatorType
-    right: AstExpression
-
-    def __post_init__(self):
-        self.left = _as_logical(self.left)
-        self.right = _as_logical(self.right)
-
-    def _to_agentspeak(self) -> str:
-        l_str = str(self.left)
-        r_str = str(self.right)
-
-        assert isinstance(self.left, AstLogicalExpression)
-        assert isinstance(self.right, AstLogicalExpression)
-
-        if isinstance(self.left.expression, AstBinaryOp) or self.left.negated:
-            l_str = f"({l_str})"
-        if isinstance(self.right.expression, AstBinaryOp) or self.right.negated:
-            r_str = f"({r_str})"
-
-        return f"{l_str} {self.operator.value} {r_str}"
-
-
-@dataclass
-class AstLogicalExpression(AstExpression):
-    expression: AstExpression
-    negated: bool = False
-
-    def _to_agentspeak(self) -> str:
-        expr_str = str(self.expression)
-        if isinstance(self.expression, AstBinaryOp) and self.negated:
-            expr_str = f"({expr_str})"
-        return f"{'not ' if self.negated else ''}{expr_str}"
-
-
-def _as_logical(expr: AstExpression) -> AstLogicalExpression:
-    if isinstance(expr, AstLogicalExpression):
-        return expr
-    return AstLogicalExpression(expr)
-
-
-class StatementType(StrEnum):
-    EMPTY = ""
-    DO_ACTION = "."
-    ACHIEVE_GOAL = "!"
-    TEST_GOAL = "?"
-    ADD_BELIEF = "+"
-    REMOVE_BELIEF = "-"
-    REPLACE_BELIEF = "-+"
-
-
-@dataclass
-class AstStatement(AstNode):
-    """
-    A statement that can appear inside a plan.
-    """
-
-    type: StatementType
-    expression: AstExpression
-
-    def _to_agentspeak(self) -> str:
-        return f"{self.type.value}{self.expression}"
-
-
-@dataclass
-class AstRule(AstNode):
-    result: AstExpression
-    condition: AstExpression | None = None
-
-    def __post_init__(self):
-        if self.condition is not None:
-            self.condition = _as_logical(self.condition)
-
-    def _to_agentspeak(self) -> str:
-        if not self.condition:
-            return f"{self.result}."
-        return f"{self.result} :- {self.condition}."
-
-
-class TriggerType(StrEnum):
-    ADDED_BELIEF = "+"
-    # REMOVED_BELIEF = "-"  # TODO
-    # MODIFIED_BELIEF = "^"  # TODO
-    ADDED_GOAL = "+!"
-    # REMOVED_GOAL = "-!"  # TODO
-
-
-@dataclass
-class AstPlan(AstNode):
-    type: TriggerType
-    trigger_literal: AstExpression
-    context: list[AstExpression]
-    body: list[AstStatement]
-
-    def _to_agentspeak(self) -> str:
-        assert isinstance(self.trigger_literal, AstLiteral)
-
-        indent = " " * 6
-        colon = "  :   "
-        arrow = "  <-  "
-
-        lines = []
-
-        lines.append(f"{self.type.value}{self.trigger_literal}")
-
-        if self.context:
-            lines.append(colon + f" &\n{indent}".join(str(c) for c in self.context))
-
-        if self.body:
-            lines.append(arrow + f";\n{indent}".join(str(s) for s in self.body) + ".")
-
-        lines.append("")
-
-        return "\n".join(lines)
-
-
-@dataclass
-class AstProgram(AstNode):
-    rules: list[AstRule] = field(default_factory=list)
-    plans: list[AstPlan] = field(default_factory=list)
-
-    def _to_agentspeak(self) -> str:
-        lines = []
-        lines.extend(map(str, self.rules))
-
-        lines.extend(["", ""])
-        lines.extend(map(str, self.plans))
-
-        return "\n".join(lines)

src/control_backend/agents/bdi/agentspeak_generator.py

@@ -1,373 +0,0 @@
-from functools import singledispatchmethod
-
-from slugify import slugify
-
-from control_backend.agents.bdi.agentspeak_ast import (
-    AstBinaryOp,
-    AstExpression,
-    AstLiteral,
-    AstPlan,
-    AstProgram,
-    AstRule,
-    AstStatement,
-    AstString,
-    AstVar,
-    BinaryOperatorType,
-    StatementType,
-    TriggerType,
-)
-from control_backend.schemas.program import (
-    BasicNorm,
-    ConditionalNorm,
-    GestureAction,
-    Goal,
-    InferredBelief,
-    KeywordBelief,
-    LLMAction,
-    LogicalOperator,
-    Norm,
-    Phase,
-    PlanElement,
-    Program,
-    ProgramElement,
-    SemanticBelief,
-    SpeechAction,
-    Trigger,
-)
-
-
-class AgentSpeakGenerator:
-    _asp: AstProgram
-
-    def generate(self, program: Program) -> str:
-        self._asp = AstProgram()
-
-        self._asp.rules.append(AstRule(self._astify(program.phases[0])))
-        self._add_keyword_inference()
-        self._add_default_plans()
-
-        self._process_phases(program.phases)
-
-        self._add_fallbacks()
-
-        return str(self._asp)
-
-    def _add_keyword_inference(self) -> None:
-        keyword = AstVar("Keyword")
-        message = AstVar("Message")
-        position = AstVar("Pos")
-
-        self._asp.rules.append(
-            AstRule(
-                AstLiteral("keyword_said", [keyword]),
-                AstLiteral("user_said", [message])
-                & AstLiteral(".substring", [keyword, message, position])
-                & (position >= 0),
-            )
-        )
-
-    def _add_default_plans(self):
-        self._add_reply_with_goal_plan()
-        self._add_say_plan()
-        self._add_reply_plan()
-
-    def _add_reply_with_goal_plan(self):
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                AstLiteral("reply_with_goal", [AstVar("Goal")]),
-                [AstLiteral("user_said", [AstVar("Message")])],
-                [
-                    AstStatement(StatementType.ADD_BELIEF, AstLiteral("responded_this_turn")),
-                    AstStatement(
-                        StatementType.DO_ACTION,
-                        AstLiteral(
-                            "findall",
-                            [AstVar("Norm"), AstLiteral("norm", [AstVar("Norm")]), AstVar("Norms")],
-                        ),
-                    ),
-                    AstStatement(
-                        StatementType.DO_ACTION,
-                        AstLiteral(
-                            "reply_with_goal", [AstVar("Message"), AstVar("Norms"), AstVar("Goal")]
-                        ),
-                    ),
-                ],
-            )
-        )
-
-    def _add_say_plan(self):
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                AstLiteral("say", [AstVar("Text")]),
-                [],
-                [
-                    AstStatement(StatementType.ADD_BELIEF, AstLiteral("responded_this_turn")),
-                    AstStatement(StatementType.DO_ACTION, AstLiteral("say", [AstVar("Text")])),
-                ],
-            )
-        )
-
-    def _add_reply_plan(self):
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                AstLiteral("reply"),
-                [AstLiteral("user_said", [AstVar("Message")])],
-                [
-                    AstStatement(StatementType.ADD_BELIEF, AstLiteral("responded_this_turn")),
-                    AstStatement(
-                        StatementType.DO_ACTION,
-                        AstLiteral(
-                            "findall",
-                            [AstVar("Norm"), AstLiteral("norm", [AstVar("Norm")]), AstVar("Norms")],
-                        ),
-                    ),
-                    AstStatement(
-                        StatementType.DO_ACTION,
-                        AstLiteral("reply", [AstVar("Message"), AstVar("Norms")]),
-                    ),
-                ],
-            )
-        )
-
-    def _process_phases(self, phases: list[Phase]) -> None:
-        for curr_phase, next_phase in zip([None] + phases, phases + [None], strict=True):
-            if curr_phase:
-                self._process_phase(curr_phase)
-            self._add_phase_transition(curr_phase, next_phase)
-
-        # End phase behavior
-        # When deleting this, the entire `reply` plan and action can be deleted
-        self._asp.plans.append(
-            AstPlan(
-                type=TriggerType.ADDED_BELIEF,
-                trigger_literal=AstLiteral("user_said", [AstVar("Message")]),
-                context=[AstLiteral("phase", [AstString("end")])],
-                body=[AstStatement(StatementType.ACHIEVE_GOAL, AstLiteral("reply"))],
-            )
-        )
-
-    def _process_phase(self, phase: Phase) -> None:
-        for norm in phase.norms:
-            self._process_norm(norm, phase)
-
-        self._add_default_loop(phase)
-
-        previous_goal = None
-        for goal in phase.goals:
-            self._process_goal(goal, phase, previous_goal)
-            previous_goal = goal
-
-        for trigger in phase.triggers:
-            self._process_trigger(trigger, phase)
-
-    def _add_phase_transition(self, from_phase: Phase | None, to_phase: Phase | None) -> None:
-        if from_phase is None:
-            return
-        from_phase_ast = self._astify(from_phase)
-        to_phase_ast = (
-            self._astify(to_phase) if to_phase else AstLiteral("phase", [AstString("end")])
-        )
-
-        context = [from_phase_ast, ~AstLiteral("responded_this_turn")]
-        if from_phase and from_phase.goals:
-            context.append(self._astify(from_phase.goals[-1], achieved=True))
-
-        body = [
-            AstStatement(StatementType.REMOVE_BELIEF, from_phase_ast),
-            AstStatement(StatementType.ADD_BELIEF, to_phase_ast),
-        ]
-
-        if from_phase:
-            body.extend(
-                [
-                    AstStatement(
-                        StatementType.TEST_GOAL, AstLiteral("user_said", [AstVar("Message")])
-                    ),
-                    AstStatement(
-                        StatementType.REPLACE_BELIEF, AstLiteral("user_said", [AstVar("Message")])
-                    ),
-                ]
-            )
-
-        self._asp.plans.append(
-            AstPlan(TriggerType.ADDED_GOAL, AstLiteral("transition_phase"), context, body)
-        )
-
-    def _process_norm(self, norm: Norm, phase: Phase) -> None:
-        rule: AstRule | None = None
-
-        match norm:
-            case ConditionalNorm(condition=cond):
-                rule = AstRule(self._astify(norm), self._astify(phase) & self._astify(cond))
-            case BasicNorm():
-                rule = AstRule(self._astify(norm), self._astify(phase))
-
-        if not rule:
-            return
-
-        self._asp.rules.append(rule)
-
-    def _add_default_loop(self, phase: Phase) -> None:
-        actions = []
-
-        actions.append(AstStatement(StatementType.REMOVE_BELIEF, AstLiteral("responded_this_turn")))
-        actions.append(AstStatement(StatementType.ACHIEVE_GOAL, AstLiteral("check_triggers")))
-
-        for goal in phase.goals:
-            actions.append(AstStatement(StatementType.ACHIEVE_GOAL, self._astify(goal)))
-
-        actions.append(AstStatement(StatementType.ACHIEVE_GOAL, AstLiteral("transition_phase")))
-
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_BELIEF,
-                AstLiteral("user_said", [AstVar("Message")]),
-                [self._astify(phase)],
-                actions,
-            )
-        )
-
-    def _process_goal(
-        self,
-        goal: Goal,
-        phase: Phase,
-        previous_goal: Goal | None = None,
-        continues_response: bool = False,
-    ) -> None:
-        context: list[AstExpression] = [self._astify(phase)]
-        context.append(~self._astify(goal, achieved=True))
-        if previous_goal and previous_goal.can_fail:
-            context.append(self._astify(previous_goal, achieved=True))
-        if not continues_response:
-            context.append(~AstLiteral("responded_this_turn"))
-
-        body = []
-
-        subgoals = []
-        for step in goal.plan.steps:
-            body.append(self._step_to_statement(step))
-            if isinstance(step, Goal):
-                subgoals.append(step)
-
-        if not goal.can_fail and not continues_response:
-            body.append(AstStatement(StatementType.ADD_BELIEF, self._astify(goal, achieved=True)))
-
-        self._asp.plans.append(AstPlan(TriggerType.ADDED_GOAL, self._astify(goal), context, body))
-
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                self._astify(goal),
-                context=[],
-                body=[AstStatement(StatementType.EMPTY, AstLiteral("true"))],
-            )
-        )
-
-        prev_goal = None
-        for subgoal in subgoals:
-            self._process_goal(subgoal, phase, prev_goal)
-            prev_goal = subgoal
-
-    def _step_to_statement(self, step: PlanElement) -> AstStatement:
-        match step:
-            case Goal() | SpeechAction() | LLMAction() as a:
-                return AstStatement(StatementType.ACHIEVE_GOAL, self._astify(a))
-            case GestureAction() as a:
-                return AstStatement(StatementType.DO_ACTION, self._astify(a))
-
-    # TODO: separate handling of keyword and others
-    def _process_trigger(self, trigger: Trigger, phase: Phase) -> None:
-        body = []
-        subgoals = []
-
-        for step in trigger.plan.steps:
-            body.append(self._step_to_statement(step))
-            if isinstance(step, Goal):
-                step.can_fail = False  # triggers are continuous sequence
-                subgoals.append(step)
-
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                AstLiteral("check_triggers"),
-                [self._astify(phase), self._astify(trigger.condition)],
-                body,
-            )
-        )
-
-        for subgoal in subgoals:
-            self._process_goal(subgoal, phase, continues_response=True)
-
-    def _add_fallbacks(self):
-        # Trigger fallback
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                AstLiteral("check_triggers"),
-                [],
-                [AstStatement(StatementType.EMPTY, AstLiteral("true"))],
-            )
-        )
-
-        # Phase transition fallback
-        self._asp.plans.append(
-            AstPlan(
-                TriggerType.ADDED_GOAL,
-                AstLiteral("transition_phase"),
-                [],
-                [AstStatement(StatementType.EMPTY, AstLiteral("true"))],
-            )
-        )
-
-    @singledispatchmethod
-    def _astify(self, element: ProgramElement) -> AstExpression:
-        raise NotImplementedError(f"Cannot convert element {element} to an AgentSpeak expression.")
-
-    @_astify.register
-    def _(self, kwb: KeywordBelief) -> AstExpression:
-        return AstLiteral("keyword_said", [AstString(kwb.keyword)])
-
-    @_astify.register
-    def _(self, sb: SemanticBelief) -> AstExpression:
-        return AstLiteral(f"semantic_{self._slugify_str(sb.description)}")
-
-    @_astify.register
-    def _(self, ib: InferredBelief) -> AstExpression:
-        return AstBinaryOp(
-            self._astify(ib.left),
-            BinaryOperatorType.AND if ib.operator == LogicalOperator.AND else BinaryOperatorType.OR,
-            self._astify(ib.right),
-        )
-
-    @_astify.register
-    def _(self, norm: Norm) -> AstExpression:
-        functor = "critical_norm" if norm.critical else "norm"
-        return AstLiteral(functor, [AstString(norm.norm)])
-
-    @_astify.register
-    def _(self, phase: Phase) -> AstExpression:
-        return AstLiteral("phase", [AstString(str(phase.id))])
-
-    @_astify.register
-    def _(self, goal: Goal, achieved: bool = False) -> AstExpression:
-        return AstLiteral(f"{'achieved_' if achieved else ''}{self._slugify_str(goal.name)}")
-
-    @_astify.register
-    def _(self, sa: SpeechAction) -> AstExpression:
-        return AstLiteral("say", [AstString(sa.text)])
-
-    @_astify.register
-    def _(self, ga: GestureAction) -> AstExpression:
-        gesture = ga.gesture
-        return AstLiteral("gesture", [AstString(gesture.type), AstString(gesture.name)])
-
-    @_astify.register
-    def _(self, la: LLMAction) -> AstExpression:
-        return AstLiteral("reply_with_goal", [AstString(la.goal)])
-
-    @staticmethod
-    def _slugify_str(text: str) -> str:
-        return slugify(text, separator="_", stopwords=["a", "an", "the", "we", "you", "I"])

src/control_backend/agents/bdi/asl_ast.py

@@ -1,203 +0,0 @@
-import typing
-from dataclasses import dataclass, field
-
-# --- Types ---
-
-
-@dataclass
-class BeliefLiteral:
-    """
-    Represents a literal or atom.
-    Example: phase(1), user_said("hello"), ~started
-    """
-
-    functor: str
-    args: list[str] = field(default_factory=list)
-    negated: bool = False
-
-    def __str__(self):
-        # In ASL, 'not' is usually for closed-world assumption (prolog style),
-        # '~' is for explicit negation in beliefs.
-        # For simplicity in behavior trees, we often use 'not' for conditions.
-        prefix = "not " if self.negated else ""
-        if not self.args:
-            return f"{prefix}{self.functor}"
-
-        # Clean args to ensure strings are quoted if they look like strings,
-        # but usually the converter handles the quoting of string literals.
-        args_str = ", ".join(self.args)
-        return f"{prefix}{self.functor}({args_str})"
-
-
-@dataclass
-class GoalLiteral:
-    name: str
-
-    def __str__(self):
-        return f"!{self.name}"
-
-
-@dataclass
-class ActionLiteral:
-    """
-    Represents a step in a plan body.
-    Example: .say("Hello") or !achieve_goal
-    """
-
-    code: str
-
-    def __str__(self):
-        return self.code
-
-
-@dataclass
-class BinaryOp:
-    """
-    Represents logical operations.
-    Example: (A & B) | C
-    """
-
-    left: "Expression | str"
-    operator: typing.Literal["&", "|"]
-    right: "Expression | str"
-
-    def __str__(self):
-        l_str = str(self.left)
-        r_str = str(self.right)
-
-        if isinstance(self.left, BinaryOp):
-            l_str = f"({l_str})"
-        if isinstance(self.right, BinaryOp):
-            r_str = f"({r_str})"
-
-        return f"{l_str} {self.operator} {r_str}"
-
-
-Literal = BeliefLiteral | GoalLiteral | ActionLiteral
-Expression = Literal | BinaryOp | str
-
-
-@dataclass
-class Rule:
-    """
-    Represents an inference rule.
-    Example: head :- body.
-    """
-
-    head: Expression
-    body: Expression | None = None
-
-    def __str__(self):
-        if not self.body:
-            return f"{self.head}."
-        return f"{self.head} :- {self.body}."
-
-
-@dataclass
-class PersistentRule:
-    """
-    Represents an inference rule, where the inferred belief is persistent when formed.
-    """
-
-    head: Expression
-    body: Expression
-
-    def __str__(self):
-        if not self.body:
-            raise Exception("Rule without body should not be persistent.")
-
-        lines = []
-
-        if isinstance(self.body, BinaryOp):
-            lines.append(f"+{self.body.left}")
-            if self.body.operator == "&":
-                lines.append(f"  :   {self.body.right}")
-            lines.append(f"  <-  +{self.head}.")
-            if self.body.operator == "|":
-                lines.append(f"+{self.body.right}")
-                lines.append(f"  <-  +{self.head}.")
-
-        return "\n".join(lines)
-
-
-@dataclass
-class Plan:
-    """
-    Represents a plan.
-    Syntax: +trigger : context <- body.
-    """
-
-    trigger: BeliefLiteral | GoalLiteral
-    context: list[Expression] = field(default_factory=list)
-    body: list[ActionLiteral] = field(default_factory=list)
-
-    def __str__(self):
-        # Indentation settings
-        INDENT = "      "
-        ARROW = "\n  <-  "
-        COLON = "\n  :   "
-
-        # Build Header
-        header = f"+{self.trigger}"
-        if self.context:
-            ctx_str = f" &\n{INDENT}".join(str(c) for c in self.context)
-            header += f"{COLON}{ctx_str}"
-
-        # Case 1: Empty body
-        if not self.body:
-            return f"{header}."
-
-        # Case 2: Short body (optional optimization, keeping it uniform usually better)
-        header += ARROW
-
-        lines = []
-        # We start the first action on the same line or next line.
-        # Let's put it on the next line for readability if there are multiple.
-
-        if len(self.body) == 1:
-            return f"{header}{self.body[0]}."
-
-        # First item
-        lines.append(f"{header}{self.body[0]};")
-        # Middle items
-        for item in self.body[1:-1]:
-            lines.append(f"{INDENT}{item};")
-        # Last item
-        lines.append(f"{INDENT}{self.body[-1]}.")
-
-        return "\n".join(lines)
-
-
-@dataclass
-class AgentSpeakFile:
-    """
-    Root element representing the entire generated file.
-    """
-
-    initial_beliefs: list[Rule] = field(default_factory=list)
-    inference_rules: list[Rule | PersistentRule] = field(default_factory=list)
-    plans: list[Plan] = field(default_factory=list)
-
-    def __str__(self):
-        sections = []
-
-        if self.initial_beliefs:
-            sections.append("// --- Initial Beliefs & Facts ---")
-            sections.extend(str(rule) for rule in self.initial_beliefs)
-            sections.append("")
-
-        if self.inference_rules:
-            sections.append("// --- Inference Rules ---")
-            sections.extend(str(rule) for rule in self.inference_rules if isinstance(rule, Rule))
-            sections.append("")
-            sections.extend(
-                str(rule) for rule in self.inference_rules if isinstance(rule, PersistentRule)
-            )
-            sections.append("")
-
-        if self.plans:
-            sections.append("// --- Plans ---")
-            # Separate plans by a newline for readability
-            sections.extend(str(plan) + "\n" for plan in self.plans)
-
-        return "\n".join(sections)

src/control_backend/agents/bdi/asl_gen.py

@@ -1,425 +0,0 @@
-import asyncio
-import time
-from functools import singledispatchmethod
-
-from slugify import slugify
-
-from control_backend.agents.bdi import BDICoreAgent
-from control_backend.agents.bdi.asl_ast import (
-    ActionLiteral,
-    AgentSpeakFile,
-    BeliefLiteral,
-    BinaryOp,
-    Expression,
-    GoalLiteral,
-    PersistentRule,
-    Plan,
-    Rule,
-)
-from control_backend.agents.bdi.bdi_program_manager import test_program
-from control_backend.schemas.program import (
-    BasicBelief,
-    Belief,
-    ConditionalNorm,
-    GestureAction,
-    Goal,
-    InferredBelief,
-    KeywordBelief,
-    LLMAction,
-    LogicalOperator,
-    Phase,
-    Program,
-    ProgramElement,
-    SemanticBelief,
-    SpeechAction,
-)
-
-
-async def do_things():
-    res = input("Wanna generate")
-    if res == "y":
-        program = AgentSpeakGenerator().generate(test_program)
-        filename = f"{int(time.time())}.asl"
-        with open(filename, "w") as f:
-            f.write(program)
-    else:
-        # filename = "0test.asl"
-        filename = "1766062491.asl"
-    bdi_agent = BDICoreAgent("BDICoreAgent", filename)
-    flag = asyncio.Event()
-    await bdi_agent.start()
-    await flag.wait()
-
-
-def do_other_things():
-    print(AgentSpeakGenerator().generate(test_program))
-
-
-class AgentSpeakGenerator:
-    """
-    Converts a Pydantic Program behavior model into an AgentSpeak(L) AST,
-    then renders it to a string.
-    """
-
-    def generate(self, program: Program) -> str:
-        asl = AgentSpeakFile()
-
-        self._generate_startup(program, asl)
-
-        for i, phase in enumerate(program.phases):
-            next_phase = program.phases[i + 1] if i < len(program.phases) - 1 else None
-
-            self._generate_phase_flow(phase, next_phase, asl)
-
-            self._generate_norms(phase, asl)
-
-            self._generate_goals(phase, asl)
-
-            self._generate_triggers(phase, asl)
-
-        self._generate_fallbacks(program, asl)
-
-        return str(asl)
-
-    # --- Section: Startup & Phase Management ---
-
-    def _generate_startup(self, program: Program, asl: AgentSpeakFile):
-        if not program.phases:
-            return
-
-        # Initial belief: phase(start).
-        asl.initial_beliefs.append(Rule(head=BeliefLiteral("phase", ['"start"'])))
-
-        # Startup plan: +started : phase(start) <- -phase(start); +phase(first_id).
-        asl.plans.append(
-            Plan(
-                trigger=BeliefLiteral("started"),
-                context=[BeliefLiteral("phase", ['"start"'])],
-                body=[
-                    ActionLiteral('-phase("start")'),
-                    ActionLiteral(f'+phase("{program.phases[0].id}")'),
-                ],
-            )
-        )
-
-        # Initial plans:
-        asl.plans.append(
-            Plan(
-                trigger=GoalLiteral("generate_response_with_goal(Goal)"),
-                context=[BeliefLiteral("user_said", ["Message"])],
-                body=[
-                    ActionLiteral("+responded_this_turn"),
-                    ActionLiteral(".findall(Norm, norm(Norm), Norms)"),
-                    ActionLiteral(".reply_with_goal(Message, Norms, Goal)"),
-                ],
-            )
-        )
-
-    def _generate_phase_flow(self, phase: Phase, next_phase: Phase | None, asl: AgentSpeakFile):
-        """Generates the main loop listener and the transition logic for this phase."""
-
-        # +user_said(Message) : phase(ID) <- !goal1; !goal2; !transition_phase.
-        goal_actions = [ActionLiteral("-responded_this_turn")]
-        goal_actions += [
-            ActionLiteral(f"!check_{self._slugify_str(keyword)}")
-            for keyword in self._get_keyword_conditionals(phase)
-        ]
-        goal_actions += [ActionLiteral(f"!{self._slugify(g)}") for g in phase.goals]
-        goal_actions.append(ActionLiteral("!transition_phase"))
-
-        asl.plans.append(
-            Plan(
-                trigger=BeliefLiteral("user_said", ["Message"]),
-                context=[BeliefLiteral("phase", [f'"{phase.id}"'])],
-                body=goal_actions,
-            )
-        )
-
-        # +!transition_phase : phase(ID) <- -phase(ID); +(NEXT_ID).
-        next_id = str(next_phase.id) if next_phase else "end"
-
-        transition_context = [BeliefLiteral("phase", [f'"{phase.id}"'])]
-        if phase.goals:
-            transition_context.append(BeliefLiteral(f"achieved_{self._slugify(phase.goals[-1])}"))
-
-        asl.plans.append(
-            Plan(
-                trigger=GoalLiteral("transition_phase"),
-                context=transition_context,
-                body=[
-                    ActionLiteral(f'-phase("{phase.id}")'),
-                    ActionLiteral(f'+phase("{next_id}")'),
-                    ActionLiteral("user_said(Anything)"),
-                    ActionLiteral("-+user_said(Anything)"),
-                ],
-            )
-        )
-
-    def _get_keyword_conditionals(self, phase: Phase) -> list[str]:
-        res = []
-        for belief in self._extract_basic_beliefs_from_phase(phase):
-            if isinstance(belief, KeywordBelief):
-                res.append(belief.keyword)
-
-        return res
-
-    # --- Section: Norms & Beliefs ---
-
-    def _generate_norms(self, phase: Phase, asl: AgentSpeakFile):
-        for norm in phase.norms:
-            norm_slug = f'"{norm.norm}"'
-            head = BeliefLiteral("norm", [norm_slug])
-
-            # Base context is the phase
-            phase_lit = BeliefLiteral("phase", [f'"{phase.id}"'])
-
-            if isinstance(norm, ConditionalNorm):
-                self._ensure_belief_inference(norm.condition, asl)
-
-                condition_expr = self._belief_to_expr(norm.condition)
-                body = BinaryOp(phase_lit, "&", condition_expr)
-            else:
-                body = phase_lit
-
-            asl.inference_rules.append(Rule(head=head, body=body))
-
-    def _ensure_belief_inference(self, belief: Belief, asl: AgentSpeakFile):
-        """
-        Recursively adds rules to infer beliefs.
-        Checks strictly to avoid duplicates if necessary,
-        though ASL engines often handle redefinition or we can use a set to track processed IDs.
-        """
-        if isinstance(belief, KeywordBelief):
-            pass
-            # # Rule: keyword_said("word") :- user_said(M) & .substring("word", M, P) & P >= 0.
-            # kwd_slug = f'"{belief.keyword}"'
-            # head = BeliefLiteral("keyword_said", [kwd_slug])
-            #
-            # # Avoid duplicates
-            # if any(str(r.head) == str(head) for r in asl.inference_rules):
-            #     return
-            #
-            # body = BinaryOp(
-            #     BeliefLiteral("user_said", ["Message"]),
-            #     "&",
-            #     BinaryOp(f".substring({kwd_slug}, Message, Pos)", "&", "Pos >= 0"),
-            # )
-            #
-            # asl.inference_rules.append(Rule(head=head, body=body))
-
-        elif isinstance(belief, InferredBelief):
-            self._ensure_belief_inference(belief.left, asl)
-            self._ensure_belief_inference(belief.right, asl)
-
-            slug = self._slugify(belief)
-            head = BeliefLiteral(slug)
-
-            if any(str(r.head) == str(head) for r in asl.inference_rules):
-                return
-
-            op_char = "&" if belief.operator == LogicalOperator.AND else "|"
-            body = BinaryOp(
-                self._belief_to_expr(belief.left), op_char, self._belief_to_expr(belief.right)
-            )
-            asl.inference_rules.append(PersistentRule(head=head, body=body))
-
-    def _belief_to_expr(self, belief: Belief) -> Expression:
-        if isinstance(belief, KeywordBelief):
-            return BeliefLiteral("keyword_said", [f'"{belief.keyword}"'])
-        else:
-            return BeliefLiteral(self._slugify(belief))
-
-    # --- Section: Goals ---
-
-    def _generate_goals(self, phase: Phase, asl: AgentSpeakFile):
-        previous_goal: Goal | None = None
-        for goal in phase.goals:
-            self._generate_goal_plan_recursive(goal, str(phase.id), previous_goal, asl)
-            previous_goal = goal
-
-    def _generate_goal_plan_recursive(
-        self,
-        goal: Goal,
-        phase_id: str,
-        previous_goal: Goal | None,
-        asl: AgentSpeakFile,
-        responded_needed: bool = True,
-        can_fail: bool = True,
-    ):
-        goal_slug = self._slugify(goal)
-
-        # phase(ID) & not responded_this_turn & not achieved_goal
-        context = [
-            BeliefLiteral("phase", [f'"{phase_id}"']),
-        ]
-
-        if responded_needed:
-            context.append(BeliefLiteral("responded_this_turn", negated=True))
-        if can_fail:
-            context.append(BeliefLiteral(f"achieved_{goal_slug}", negated=True))
-
-        if previous_goal:
-            prev_slug = self._slugify(previous_goal)
-            context.append(BeliefLiteral(f"achieved_{prev_slug}"))
-
-        body_actions = []
-        sub_goals_to_process = []
-
-        for step in goal.plan.steps:
-            if isinstance(step, Goal):
-                sub_slug = self._slugify(step)
-                body_actions.append(ActionLiteral(f"!{sub_slug}"))
-                sub_goals_to_process.append(step)
-            elif isinstance(step, SpeechAction):
-                body_actions.append(ActionLiteral(f'.say("{step.text}")'))
-            elif isinstance(step, GestureAction):
-                body_actions.append(ActionLiteral(f'.gesture("{step.gesture}")'))
-            elif isinstance(step, LLMAction):
-                body_actions.append(ActionLiteral(f'!generate_response_with_goal("{step.goal}")'))
-
-        # Mark achievement
-        if not goal.can_fail:
-            body_actions.append(ActionLiteral(f"+achieved_{goal_slug}"))
-
-        asl.plans.append(Plan(trigger=GoalLiteral(goal_slug), context=context, body=body_actions))
-        asl.plans.append(
-            Plan(trigger=GoalLiteral(goal_slug), context=[], body=[ActionLiteral("true")])
-        )
-
-        prev_sub = None
-        for sub_goal in sub_goals_to_process:
-            self._generate_goal_plan_recursive(sub_goal, phase_id, prev_sub, asl)
-            prev_sub = sub_goal
-
-    # --- Section: Triggers ---
-
-    def _generate_triggers(self, phase: Phase, asl: AgentSpeakFile):
-        for keyword in self._get_keyword_conditionals(phase):
-            asl.plans.append(
-                Plan(
-                    trigger=GoalLiteral(f"check_{self._slugify_str(keyword)}"),
-                    context=[
-                        ActionLiteral(
-                            f'user_said(Message) & .substring("{keyword}", Message, Pos) & Pos >= 0'
-                        )
-                    ],
-                    body=[
-                        ActionLiteral(f'+keyword_said("{keyword}")'),
-                        ActionLiteral(f'-keyword_said("{keyword}")'),
-                    ],
-                )
-            )
-            asl.plans.append(
-                Plan(
-                    trigger=GoalLiteral(f"check_{self._slugify_str(keyword)}"),
-                    body=[ActionLiteral("true")],
-                )
-            )
-
-        for trigger in phase.triggers:
-            self._ensure_belief_inference(trigger.condition, asl)
-
-            trigger_belief_slug = self._belief_to_expr(trigger.condition)
-
-            body_actions = []
-            sub_goals = []
-
-            for step in trigger.plan.steps:
-                if isinstance(step, Goal):
-                    sub_slug = self._slugify(step)
-                    body_actions.append(ActionLiteral(f"!{sub_slug}"))
-                    sub_goals.append(step)
-                elif isinstance(step, SpeechAction):
-                    body_actions.append(ActionLiteral(f'.say("{step.text}")'))
-                elif isinstance(step, GestureAction):
-                    body_actions.append(
-                        ActionLiteral(f'.gesture("{step.gesture.type}", "{step.gesture.name}")')
-                    )
-                elif isinstance(step, LLMAction):
-                    body_actions.append(
-                        ActionLiteral(f'!generate_response_with_goal("{step.goal}")')
-                    )
-
-            asl.plans.append(
-                Plan(
-                    trigger=BeliefLiteral(trigger_belief_slug),
-                    context=[BeliefLiteral("phase", [f'"{phase.id}"'])],
-                    body=body_actions,
-                )
-            )
-
-            # Recurse for triggered goals
-            prev_sub = None
-            for sub_goal in sub_goals:
-                self._generate_goal_plan_recursive(
-                    sub_goal, str(phase.id), prev_sub, asl, False, False
-                )
-                prev_sub = sub_goal
-
-    # --- Section: Fallbacks ---
-
-    def _generate_fallbacks(self, program: Program, asl: AgentSpeakFile):
-        asl.plans.append(
-            Plan(trigger=GoalLiteral("transition_phase"), context=[], body=[ActionLiteral("true")])
-        )
-
-    # --- Helpers ---
-
-    @singledispatchmethod
-    def _slugify(self, element: ProgramElement) -> str:
-        if element.name:
-            raise NotImplementedError("Cannot slugify this element.")
-        return self._slugify_str(element.name)
-
-    @_slugify.register
-    def _(self, goal: Goal) -> str:
-        if goal.name:
-            return self._slugify_str(goal.name)
-        return f"goal_{goal.id.hex}"
-
-    @_slugify.register
-    def _(self, kwb: KeywordBelief) -> str:
-        return f"keyword_said({kwb.keyword})"
-
-    @_slugify.register
-    def _(self, sb: SemanticBelief) -> str:
-        return self._slugify_str(sb.description)
-
-    @_slugify.register
-    def _(self, ib: InferredBelief) -> str:
-        return self._slugify_str(ib.name)
-
-    def _slugify_str(self, text: str) -> str:
-        return slugify(text, separator="_", stopwords=["a", "an", "the", "we", "you", "I"])
-
-    def _extract_basic_beliefs_from_program(self, program: Program) -> list[BasicBelief]:
-        beliefs = []
-
-        for phase in program.phases:
-            beliefs.extend(self._extract_basic_beliefs_from_phase(phase))
-
-        return beliefs
-
-    def _extract_basic_beliefs_from_phase(self, phase: Phase) -> list[BasicBelief]:
-        beliefs = []
-
-        for norm in phase.norms:
-            if isinstance(norm, ConditionalNorm):
-                beliefs += self._extract_basic_beliefs_from_belief(norm.condition)
-
-        for trigger in phase.triggers:
-            beliefs += self._extract_basic_beliefs_from_belief(trigger.condition)
-
-        return beliefs
-
-    def _extract_basic_beliefs_from_belief(self, belief: Belief) -> list[BasicBelief]:
-        if isinstance(belief, InferredBelief):
-            return self._extract_basic_beliefs_from_belief(
-                belief.left
-            ) + self._extract_basic_beliefs_from_belief(belief.right)
-        return [belief]
-
-
-if __name__ == "__main__":
-    asyncio.run(do_things())
-    # do_other_things()y

src/control_backend/agents/bdi/bdi_core_agent.py

@@ -42,13 +42,13 @@ class BDICoreAgent(BaseAgent):
 
     bdi_agent: agentspeak.runtime.Agent
 
-    def __init__(self, name: str):
+    def __init__(self, name: str, asl: str):
         super().__init__(name)
+        self.asl_file = asl
         self.env = agentspeak.runtime.Environment()
         # Deep copy because we don't actually want to modify the standard actions globally
         self.actions = copy.deepcopy(agentspeak.stdlib.actions)
         self._wake_bdi_loop = asyncio.Event()
-        self._bdi_loop_task = None
 
     async def setup(self) -> None:
         """
@@ -65,22 +65,19 @@ class BDICoreAgent(BaseAgent):
         await self._load_asl()
 
         # Start the BDI cycle loop
-        self._bdi_loop_task = self.add_behavior(self._bdi_loop())
+        self.add_behavior(self._bdi_loop())
         self._wake_bdi_loop.set()
         self.logger.debug("Setup complete.")
 
-    async def _load_asl(self, file_name: str | None = None) -> None:
+    async def _load_asl(self):
         """
         Load and parse the AgentSpeak source file.
         """
-        file_name = file_name or "src/control_backend/agents/bdi/default_behavior.asl"
-
         try:
-            with open(file_name) as source:
+            with open(self.asl_file) as source:
                 self.bdi_agent = self.env.build_agent(source, self.actions)
-                self.logger.info(f"Loaded new ASL from {file_name}.")
         except FileNotFoundError:
-            self.logger.warning(f"Could not find the specified ASL file at {file_name}.")
+            self.logger.warning(f"Could not find the specified ASL file at {self.asl_file}.")
             self.bdi_agent = agentspeak.runtime.Agent(self.env, self.name)
 
     async def _bdi_loop(self):
@@ -119,7 +116,6 @@ class BDICoreAgent(BaseAgent):
         Handle incoming messages.
 
         - **Beliefs**: Updates the internal belief base.
-        - **Program**: Updates the internal agentspeak file to match the current program.
         - **LLM Responses**: Forwards the generated text to the Robot Speech Agent (actuation).
 
         :param msg: The received internal message.
@@ -134,13 +130,6 @@ class BDICoreAgent(BaseAgent):
                 self.logger.exception("Error processing belief.")
             return
 
-        # New agentspeak file
-        if msg.thread == "new_program":
-            if self._bdi_loop_task:
-                self._bdi_loop_task.cancel()
-            await self._load_asl(msg.body)
-            self.add_behavior(self._bdi_loop())
-
         # The message was not a belief, handle special cases based on sender
         match msg.sender:
             case settings.agent_settings.llm_name:
@@ -171,7 +160,7 @@ class BDICoreAgent(BaseAgent):
                 self._remove_all_with_name(belief.name)
             self._add_belief(belief.name, belief.arguments)
 
-    def _add_belief(self, name: str, args: list[str] = None):
+    def _add_belief(self, name: str, args: Iterable[str] = []):
         """
         Add a single belief to the BDI agent.
 
@@ -179,13 +168,9 @@ class BDICoreAgent(BaseAgent):
         :param args: Arguments for the belief.
         """
         # new_args = (agentspeak.Literal(arg) for arg in args)  # TODO: Eventually support multiple
-        args = args or []
-        if args:
-            merged_args = DELIMITER.join(arg for arg in args)
-            new_args = (agentspeak.Literal(merged_args),)
-            term = agentspeak.Literal(name, new_args)
-        else:
-            term = agentspeak.Literal(name)
+        merged_args = DELIMITER.join(arg for arg in args)
+        new_args = (agentspeak.Literal(merged_args),)
+        term = agentspeak.Literal(name, new_args)
 
         self.bdi_agent.call(
             agentspeak.Trigger.addition,
@@ -250,86 +235,32 @@ class BDICoreAgent(BaseAgent):
         the function expects (which will be located in `term.args`).
         """
 
-        @self.actions.add(".reply", 2)
+        @self.actions.add(".reply", 3)
         def _reply(agent: "BDICoreAgent", term, intention):
             """
-            Let the LLM generate a response to a user's utterance with the current norms and goals.
+            Sends text to the LLM (AgentSpeak action).
+            Example: .reply("Hello LLM!", "Some norm", "Some goal")
             """
             message_text = agentspeak.grounded(term.args[0], intention.scope)
             norms = agentspeak.grounded(term.args[1], intention.scope)
+            goals = agentspeak.grounded(term.args[2], intention.scope)
 
             self.logger.debug("Norms: %s", norms)
+            self.logger.debug("Goals: %s", goals)
             self.logger.debug("User text: %s", message_text)
 
-            self.add_behavior(self._send_to_llm(str(message_text), str(norms), ""))
+            asyncio.create_task(self._send_to_llm(str(message_text), str(norms), str(goals)))
             yield
 
-        @self.actions.add(".reply_with_goal", 3)
-        def _reply_with_goal(agent: "BDICoreAgent", term, intention):
-            """
-            Let the LLM generate a response to a user's utterance with the current norms and a
-            specific goal.
-            """
-            message_text = agentspeak.grounded(term.args[0], intention.scope)
-            norms = agentspeak.grounded(term.args[1], intention.scope)
-            goal = agentspeak.grounded(term.args[2], intention.scope)
-
-            self.logger.debug(
-                '"reply_with_goal" action called with message=%s, norms=%s, goal=%s',
-                message_text,
-                norms,
-                goal,
-            )
-            self.add_behavior(self._send_to_llm(str(message_text), str(norms), str(goal)))
-            yield
-
-        @self.actions.add(".say", 1)
-        def _say(agent: "BDICoreAgent", term, intention):
-            """
-            Make the robot say the given text instantly.
-            """
-            message_text = agentspeak.grounded(term.args[0], intention.scope)
-
-            self.logger.debug('"say" action called with text=%s', message_text)
-
-            speech_command = SpeechCommand(data=message_text)
-            speech_message = InternalMessage(
-                to=settings.agent_settings.robot_speech_name,
-                sender=settings.agent_settings.bdi_core_name,
-                body=speech_command.model_dump_json(),
-            )
-            # TODO: add to conversation history
-            self.add_behavior(self.send(speech_message))
-            yield
-
-        @self.actions.add(".gesture", 2)
-        def _gesture(agent: "BDICoreAgent", term, intention):
-            """
-            Make the robot perform the given gesture instantly.
-            """
-            gesture_type = agentspeak.grounded(term.args[0], intention.scope)
-            gesture_name = agentspeak.grounded(term.args[1], intention.scope)
-
-            self.logger.debug(
-                '"gesture" action called with type=%s, name=%s',
-                gesture_type,
-                gesture_name,
-            )
-
-            # gesture = Gesture(type=gesture_type, name=gesture_name)
-            # gesture_message = InternalMessage(
-            #     to=settings.agent_settings.robot_gesture_name,
-            #     sender=settings.agent_settings.bdi_core_name,
-            #     body=gesture.model_dump_json(),
-            # )
-            # asyncio.create_task(agent.send(gesture_message))
-            yield
-
-    async def _send_to_llm(self, text: str, norms: str, goals: str):
+    async def _send_to_llm(self, text: str, norms: str = None, goals: str = None):
         """
         Sends a text query to the LLM agent asynchronously.
         """
-        prompt = LLMPromptMessage(text=text, norms=norms.split("\n"), goals=goals.split("\n"))
+        prompt = LLMPromptMessage(
+            text=text,
+            norms=norms.split("\n") if norms else [],
+            goals=goals.split("\n") if norms else [],
+        )
         msg = InternalMessage(
             to=settings.agent_settings.llm_name,
             sender=self.name,

src/control_backend/agents/bdi/bdi_program_manager.py

@@ -3,9 +3,9 @@ from pydantic import ValidationError
 from zmq.asyncio import Context
 
 from control_backend.agents import BaseAgent
-from control_backend.agents.bdi.agentspeak_generator import AgentSpeakGenerator
+from control_backend.core.agent_system import InternalMessage
 from control_backend.core.config import settings
-from control_backend.schemas.internal_message import InternalMessage
+from control_backend.schemas.belief_message import Belief, BeliefMessage
 from control_backend.schemas.program import Program
 
 
@@ -25,7 +25,7 @@ class BDIProgramManager(BaseAgent):
         super().__init__(**kwargs)
         self.sub_socket = None
 
-    async def _create_agentspeak_and_send_to_bdi(self, program: Program):
+    async def _send_to_bdi(self, program: Program):
         """
         Convert a received program into BDI beliefs and send them to the BDI Core Agent.
 
@@ -38,23 +38,42 @@ class BDIProgramManager(BaseAgent):
 
         :param program: The program object received from the API.
         """
-        asg = AgentSpeakGenerator()
-
-        asl_str = asg.generate(program)
-
-        file_name = "src/control_backend/agents/bdi/agentspeak.asl"
-
-        with open(file_name, "w") as f:
-            f.write(asl_str)
-
-        msg = InternalMessage(
-            sender=self.name,
-            to=settings.agent_settings.bdi_core_name,
-            body=file_name,
-            thread="new_program",
+        first_phase = program.phases[0]
+        norms_belief = Belief(
+            name="norms",
+            arguments=[norm.norm for norm in first_phase.norms],
+            replace=True,
         )
+        goals_belief = Belief(
+            name="goals",
+            arguments=[goal.description for goal in first_phase.goals],
+            replace=True,
+        )
+        program_beliefs = BeliefMessage(beliefs=[norms_belief, goals_belief])
 
-        await self.send(msg)
+        message = InternalMessage(
+            to=settings.agent_settings.bdi_core_name,
+            sender=self.name,
+            body=program_beliefs.model_dump_json(),
+            thread="beliefs",
+        )
+        await self.send(message)
+        self.logger.debug("Sent new norms and goals to the BDI agent.")
+
+    async def _send_clear_llm_history(self):
+        """
+        Clear the LLM Agent's conversation history.
+
+        Sends a message to the LLM Agent instructing it to clear its history.
+        """
+        message = InternalMessage(
+            to=settings.agent_settings.llm_name,
+            sender=self.name,
+            body="clear_history",
+            threads="clear history message",
+        )
+        await self.send(message)
+        self.logger.debug("Sent message to LLM agent to clear history.")
 
     async def _receive_programs(self):
         """
@@ -62,18 +81,20 @@ class BDIProgramManager(BaseAgent):
 
         It listens to the ``program`` topic on the internal ZMQ SUB socket.
         When a program is received, it is validated and forwarded to BDI via :meth:`_send_to_bdi`.
+        Additionally, the LLM history is cleared via :meth:`_send_clear_llm_history`.
         """
         while True:
             topic, body = await self.sub_socket.recv_multipart()
 
             try:
                 program = Program.model_validate_json(body)
+                await self._send_to_bdi(program)
+                await self._send_clear_llm_history()
+
             except ValidationError:
                 self.logger.exception("Received an invalid program.")
                 continue
 
-            await self._create_agentspeak_and_send_to_bdi(program)
-
     async def setup(self):
         """
         Initialize the agent.

src/control_backend/agents/bdi/belief_collector_agent.py

@@ -101,7 +101,7 @@ class BDIBeliefCollectorAgent(BaseAgent):
             :return: A Belief object if the input is valid or None.
             """
             try:
-                return Belief(name=name, arguments=arguments, replace=name == "user_said")
+                return Belief(name=name, arguments=arguments)
             except ValidationError:
                 return None
 

src/control_backend/agents/bdi/default_behavior.asl

@@ -1,5 +0,0 @@
-norms("").
-
-+user_said(Message) : norms(Norms) <-
-  -user_said(Message);
-  .reply(Message, Norms).

src/control_backend/agents/bdi/rules.asl

@@ -0,0 +1,6 @@
+norms("").
+goals("").
+
++user_said(Message) : norms(Norms) & goals(Goals) <-
+  -user_said(Message);
+  .reply(Message, Norms, Goals).

src/control_backend/agents/communication/ri_communication_agent.py

@@ -182,6 +182,7 @@ class RICommunicationAgent(BaseAgent):
                             self._req_socket.bind(addr)
                 case "actuation":
                     gesture_data = port_data.get("gestures", [])
+                    single_gesture_data = port_data.get("single_gestures", [])
                     robot_speech_agent = RobotSpeechAgent(
                         settings.agent_settings.robot_speech_name,
                         address=addr,
@@ -192,6 +193,7 @@ class RICommunicationAgent(BaseAgent):
                         address=addr,
                         bind=bind,
                         gesture_data=gesture_data,
+                        single_gesture_data=single_gesture_data,
                     )
                     await robot_speech_agent.start()
                     await asyncio.sleep(0.1)  # Small delay

src/control_backend/agents/llm/llm_agent.py

@@ -52,6 +52,10 @@ class LLMAgent(BaseAgent):
                 await self._process_bdi_message(prompt_message)
             except ValidationError:
                 self.logger.debug("Prompt message from BDI core is invalid.")
+        elif msg.sender == settings.agent_settings.bdi_program_manager_name:
+            if msg.body == "clear_history":
+                self.logger.debug("Clearing conversation history.")
+                self.history.clear()
         else:
             self.logger.debug("Message ignored (not from BDI core.")
 

src/control_backend/agents/user_interrupt/user_interrupt_agent.py

@@ -0,0 +1,146 @@
+import json
+
+import zmq
+from zmq.asyncio import Context
+
+from control_backend.agents import BaseAgent
+from control_backend.core.agent_system import InternalMessage
+from control_backend.core.config import settings
+from control_backend.schemas.ri_message import GestureCommand, RIEndpoint, SpeechCommand
+
+
+class UserInterruptAgent(BaseAgent):
+    """
+    User Interrupt Agent.
+
+    This agent receives button_pressed events from the external HTTP API
+    (via ZMQ) and uses the associated context to trigger one of the following actions:
+
+    - Send a prioritized message to the `RobotSpeechAgent`
+    - Send a prioritized gesture to the `RobotGestureAgent`
+    - Send a belief override to the `BDIProgramManager`in order to activate a
+        trigger/conditional norm or complete a goal.
+
+    Prioritized actions clear the current RI queue before inserting the new item,
+    ensuring they are executed immediately after Pepper's current action has been fulfilled.
+
+    :ivar sub_socket: The ZMQ SUB socket used to receive user intterupts.
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.sub_socket = None
+
+    async def _receive_button_event(self):
+        """
+        The behaviour of the UserInterruptAgent.
+        Continuous loop that receives button_pressed events from the button_pressed HTTP endpoint.
+        These events contain a type and a context.
+
+        These are the different types and contexts:
+        - type: "speech", context: string that the robot has to say.
+        - type: "gesture", context: single gesture name that the robot has to perform.
+        - type: "override", context: belief_id that overrides the goal/trigger/conditional norm.
+        """
+        while True:
+            topic, body = await self.sub_socket.recv_multipart()
+
+            try:
+                event_data = json.loads(body)
+                event_type = event_data.get("type")  # e.g., "speech", "gesture"
+                event_context = event_data.get("context")  # e.g., "Hello, I am Pepper!"
+            except json.JSONDecodeError:
+                self.logger.error("Received invalid JSON payload on topic %s", topic)
+                continue
+
+            if event_type == "speech":
+                await self._send_to_speech_agent(event_context)
+                self.logger.info(
+                    "Forwarded button press (speech) with context '%s' to RobotSpeechAgent.",
+                    event_context,
+                )
+            elif event_type == "gesture":
+                await self._send_to_gesture_agent(event_context)
+                self.logger.info(
+                    "Forwarded button press (gesture) with context '%s' to RobotGestureAgent.",
+                    event_context,
+                )
+            elif event_type == "override":
+                await self._send_to_program_manager(event_context)
+                self.logger.info(
+                    "Forwarded button press (override) with context '%s' to BDIProgramManager.",
+                    event_context,
+                )
+            else:
+                self.logger.warning(
+                    "Received button press with unknown type '%s' (context: '%s').",
+                    event_type,
+                    event_context,
+                )
+
+    async def _send_to_speech_agent(self, text_to_say: str):
+        """
+        method to send prioritized speech command to RobotSpeechAgent.
+
+        :param text_to_say: The string that the robot has to say.
+        """
+        cmd = SpeechCommand(data=text_to_say, is_priority=True)
+        out_msg = InternalMessage(
+            to=settings.agent_settings.robot_speech_name,
+            sender=self.name,
+            body=cmd.model_dump_json(),
+        )
+        await self.send(out_msg)
+
+    async def _send_to_gesture_agent(self, single_gesture_name: str):
+        """
+        method to send prioritized gesture command to RobotGestureAgent.
+
+        :param single_gesture_name: The gesture tag that the robot has to perform.
+        """
+        # the endpoint is set to always be GESTURE_SINGLE for user interrupts
+        cmd = GestureCommand(
+            endpoint=RIEndpoint.GESTURE_SINGLE, data=single_gesture_name, is_priority=True
+        )
+        out_msg = InternalMessage(
+            to=settings.agent_settings.robot_gesture_name,
+            sender=self.name,
+            body=cmd.model_dump_json(),
+        )
+        await self.send(out_msg)
+
+    async def _send_to_program_manager(self, belief_id: str):
+        """
+        Send a button_override belief to the BDIProgramManager.
+
+        :param belief_id: The belief_id that overrides the goal/trigger/conditional norm.
+        this id can belong to a basic belief or an inferred belief.
+        See also: https://utrechtuniversity.youtrack.cloud/articles/N25B-A-27/UI-components
+        """
+        data = {"belief": belief_id}
+        message = InternalMessage(
+            to=settings.agent_settings.bdi_program_manager_name,
+            sender=self.name,
+            body=json.dumps(data),
+            thread="belief_override_id",
+        )
+        await self.send(message)
+        self.logger.info(
+            "Sent button_override belief with id '%s' to Program manager.",
+            belief_id,
+        )
+
+    async def setup(self):
+        """
+        Initialize the agent.
+
+        Connects the internal ZMQ SUB socket and subscribes to the 'button_pressed' topic.
+        Starts the background behavior to receive the user interrupts.
+        """
+        context = Context.instance()
+
+        self.sub_socket = context.socket(zmq.SUB)
+        self.sub_socket.connect(settings.zmq_settings.internal_sub_address)
+        self.sub_socket.subscribe("button_pressed")
+
+        self.add_behavior(self._receive_button_event())

src/control_backend/api/v1/endpoints/button_pressed.py

@@ -0,0 +1,31 @@
+import logging
+
+from fastapi import APIRouter, Request
+
+from control_backend.schemas.events import ButtonPressedEvent
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.post("/button_pressed", status_code=202)
+async def receive_button_event(event: ButtonPressedEvent, request: Request):
+    """
+    Endpoint to handle external button press events.
+
+    Validates the event payload and publishes it to the internal 'button_pressed' topic.
+    Subscribers (in this case user_interrupt_agent) will pick this up to trigger
+    specific behaviors or state changes.
+
+    :param event: The parsed ButtonPressedEvent object.
+    :param request: The FastAPI request object.
+    """
+    logger.debug("Received button event: %s | %s", event.type, event.context)
+
+    topic = b"button_pressed"
+    body = event.model_dump_json().encode()
+
+    pub_socket = request.app.state.endpoints_pub_socket
+    await pub_socket.send_multipart([topic, body])
+
+    return {"status": "Event received"}

src/control_backend/api/v1/router.py

@@ -1,6 +1,6 @@
 from fastapi.routing import APIRouter
 
-from control_backend.api.v1.endpoints import logs, message, program, robot, sse
+from control_backend.api.v1.endpoints import button_pressed, logs, message, program, robot, sse
 
 api_router = APIRouter()
 
@@ -13,3 +13,5 @@ api_router.include_router(robot.router, prefix="/robot", tags=["Pings", "Command
 api_router.include_router(logs.router, tags=["Logs"])
 
 api_router.include_router(program.router, tags=["Program"])
+
+api_router.include_router(button_pressed.router, tags=["Button Pressed Events"])

src/control_backend/core/config.py

@@ -48,6 +48,7 @@ class AgentSettings(BaseModel):
     ri_communication_name: str = "ri_communication_agent"
     robot_speech_name: str = "robot_speech_agent"
     robot_gesture_name: str = "robot_gesture_agent"
+    user_interrupt_name: str = "user_interrupt_agent"
 
 
 class BehaviourSettings(BaseModel):

src/control_backend/main.py

@@ -39,6 +39,9 @@ from control_backend.agents.communication import RICommunicationAgent
 # LLM Agents
 from control_backend.agents.llm import LLMAgent
 
+# User Interrupt Agent
+from control_backend.agents.user_interrupt.user_interrupt_agent import UserInterruptAgent
+
 # Other backend imports
 from control_backend.api.v1.router import api_router
 from control_backend.core.config import settings
@@ -117,6 +120,7 @@ async def lifespan(app: FastAPI):
             BDICoreAgent,
             {
                 "name": settings.agent_settings.bdi_core_name,
+                "asl": "src/control_backend/agents/bdi/rules.asl",
             },
         ),
         "BeliefCollectorAgent": (
@@ -137,6 +141,12 @@ async def lifespan(app: FastAPI):
                 "name": settings.agent_settings.bdi_program_manager_name,
             },
         ),
+        "UserInterruptAgent": (
+            UserInterruptAgent,
+            {
+                "name": settings.agent_settings.user_interrupt_name,
+            },
+        ),
     }
 
     agents = []

src/control_backend/schemas/events.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class ButtonPressedEvent(BaseModel):
+    type: str
+    context: str

src/control_backend/schemas/program.py

@@ -1,202 +1,64 @@
-from enum import Enum
-from typing import Literal
-
-from pydantic import UUID4, BaseModel
+from pydantic import BaseModel
 
 
-class ProgramElement(BaseModel):
-    """
-    Represents a basic element of our behavior program.
-
-    :ivar name: The researcher-assigned name of the element.
-    :ivar id: Unique identifier.
-    """
-
-    name: str
-    id: UUID4
-
-
-class LogicalOperator(Enum):
-    AND = "AND"
-    OR = "OR"
-
-
-type Belief = KeywordBelief | SemanticBelief | InferredBelief
-type BasicBelief = KeywordBelief | SemanticBelief
-
-
-class KeywordBelief(ProgramElement):
-    """
-    Represents a belief that is set when the user spoken text contains a certain keyword.
-
-    :ivar keyword: The keyword on which this belief gets set.
-    """
-
-    name: str = ""
-    keyword: str
-
-
-class SemanticBelief(ProgramElement):
-    """
-    Represents a belief that is set by semantic LLM validation.
-
-    :ivar description: Description of how to form the belief, used by the LLM.
-    """
-
-    name: str = ""
-    description: str
-
-
-class InferredBelief(ProgramElement):
-    """
-    Represents a belief that gets formed by combining two beliefs with a logical AND or OR.
-
-    These beliefs can also be :class:`InferredBelief`, leading to arbitrarily deep nesting.
-
-    :ivar operator: The logical operator to apply.
-    :ivar left: The left part of the logical expression.
-    :ivar right: The right part of the logical expression.
-    """
-
-    name: str = ""
-    operator: LogicalOperator
-    left: Belief
-    right: Belief
-
-
-class Norm(ProgramElement):
-    name: str = ""
-    norm: str
-    critical: bool = False
-
-
-class BasicNorm(Norm):
+class Norm(BaseModel):
     """
     Represents a behavioral norm.
 
+    :ivar id: Unique identifier.
+    :ivar label: Human-readable label.
     :ivar norm: The actual norm text describing the behavior.
-    :ivar critical: When true, this norm should absolutely not be violated (checked separately).
     """
 
-    pass
+    id: str
+    label: str
+    norm: str
 
 
-class ConditionalNorm(Norm):
+class Goal(BaseModel):
     """
-    Represents a norm that is only active when a condition is met (i.e., a certain belief holds).
+    Represents an objective to be achieved.
 
-    :ivar condition: When to activate this norm.
+    :ivar id: Unique identifier.
+    :ivar label: Human-readable label.
+    :ivar description: Detailed description of the goal.
+    :ivar achieved: Status flag indicating if the goal has been met.
     """
 
-    condition: Belief
+    id: str
+    label: str
+    description: str
+    achieved: bool
 
 
-type PlanElement = Goal | Action
+class TriggerKeyword(BaseModel):
+    id: str
+    keyword: str
 
 
-class Plan(ProgramElement):
-    """
-    Represents a list of steps to execute. Each of these steps can be a goal (with its own plan)
-    or a simple action.
-
-    :ivar steps: The actions or subgoals to execute, in order.
-    """
-
-    name: str = ""
-    steps: list[PlanElement]
+class KeywordTrigger(BaseModel):
+    id: str
+    label: str
+    type: str
+    keywords: list[TriggerKeyword]
 
 
-class Goal(ProgramElement):
-    """
-    Represents an objective to be achieved. To reach the goal, we should execute
-    the corresponding plan. If we can fail to achieve a goal after executing the plan,
-    for example when the achieving of the goal is dependent on the user's reply, this means
-    that the achieved status will be set from somewhere else in the program.
-
-    :ivar plan: The plan to execute.
-    :ivar can_fail: Whether we can fail to achieve the goal after executing the plan.
-    """
-
-    plan: Plan
-    can_fail: bool = True
-
-
-type Action = SpeechAction | GestureAction | LLMAction
-
-
-class SpeechAction(ProgramElement):
-    """
-    Represents the action of the robot speaking a literal text.
-
-    :ivar text: The text to speak.
-    """
-
-    name: str = ""
-    text: str
-
-
-class Gesture(BaseModel):
-    """
-    Represents a gesture to be performed. Can be either a single gesture,
-    or a random gesture from a category (tag).
-
-    :ivar type: The type of the gesture, "tag" or "single".
-    :ivar name: The name of the single gesture or tag.
-    """
-
-    type: Literal["tag", "single"]
-    name: str
-
-
-class GestureAction(ProgramElement):
-    """
-    Represents the action of the robot performing a gesture.
-
-    :ivar gesture: The gesture to perform.
-    """
-
-    name: str = ""
-    gesture: Gesture
-
-
-class LLMAction(ProgramElement):
-    """
-    Represents the action of letting an LLM generate a reply based on its chat history
-    and an additional goal added in the prompt.
-
-    :ivar goal: The extra (temporary) goal to add to the LLM.
-    """
-
-    name: str = ""
-    goal: str
-
-
-class Trigger(ProgramElement):
-    """
-    Represents a belief-based trigger. When a belief is set, the corresponding plan is executed.
-
-    :ivar condition: When to activate the trigger.
-    :ivar plan: The plan to execute.
-    """
-
-    name: str = ""
-    condition: Belief
-    plan: Plan
-
-
-class Phase(ProgramElement):
+class Phase(BaseModel):
     """
     A distinct phase within a program, containing norms, goals, and triggers.
 
+    :ivar id: Unique identifier.
+    :ivar label: Human-readable label.
     :ivar norms: List of norms active in this phase.
     :ivar goals: List of goals to pursue in this phase.
     :ivar triggers: List of triggers that define transitions out of this phase.
     """
 
-    name: str = ""
-    norms: list[BasicNorm | ConditionalNorm]
+    id: str
+    label: str
+    norms: list[Norm]
     goals: list[Goal]
-    triggers: list[Trigger]
+    triggers: list[KeywordTrigger]
 
 
 class Program(BaseModel):

src/control_backend/schemas/ri_message.py

@@ -38,6 +38,7 @@ class SpeechCommand(RIMessage):
 
     endpoint: RIEndpoint = RIEndpoint(RIEndpoint.SPEECH)
     data: str
+    is_priority: bool = False
 
 
 class GestureCommand(RIMessage):
@@ -52,6 +53,7 @@ class GestureCommand(RIMessage):
         RIEndpoint.GESTURE_SINGLE, RIEndpoint.GESTURE_TAG
     ]
     data: str
+    is_priority: bool = False
 
     @model_validator(mode="after")
     def check_endpoint(self):

test/unit/agents/actuation/test_robot_speech_agent.py

@@ -64,7 +64,7 @@ async def test_handle_message_sends_command():
     agent = mock_speech_agent()
     agent.pubsocket = pubsocket
 
-    payload = {"endpoint": "actuate/speech", "data": "hello"}
+    payload = {"endpoint": "actuate/speech", "data": "hello", "is_priority": False}
     msg = InternalMessage(to="robot", sender="tester", body=json.dumps(payload))
 
     await agent.handle_message(msg)
@@ -75,7 +75,7 @@ async def test_handle_message_sends_command():
 @pytest.mark.asyncio
 async def test_zmq_command_loop_valid_payload(zmq_context):
     """UI command is read from SUB and published."""
-    command = {"endpoint": "actuate/speech", "data": "hello"}
+    command = {"endpoint": "actuate/speech", "data": "hello", "is_priority": False}
     fake_socket = AsyncMock()
 
     async def recv_once():

test/unit/agents/bdi/test_bdi_program_manager.py

@@ -39,7 +39,7 @@ async def test_send_to_bdi():
     manager.send = AsyncMock()
 
     program = Program.model_validate_json(make_valid_program_json())
-    await manager._create_agentspeak_and_send_to_bdi(program)
+    await manager._send_to_bdi(program)
 
     assert manager.send.await_count == 1
     msg: InternalMessage = manager.send.await_args[0][0]
@@ -62,7 +62,8 @@ async def test_receive_programs_valid_and_invalid():
 
     manager = BDIProgramManager(name="program_manager_test")
     manager.sub_socket = sub
-    manager._create_agentspeak_and_send_to_bdi = AsyncMock()
+    manager._send_to_bdi = AsyncMock()
+    manager._send_clear_llm_history = AsyncMock()
 
     try:
         # Will give StopAsyncIteration when the predefined `sub.recv_multipart` side-effects run out
@@ -71,7 +72,28 @@ async def test_receive_programs_valid_and_invalid():
         pass
 
     # Only valid Program should have triggered _send_to_bdi
-    assert manager._create_agentspeak_and_send_to_bdi.await_count == 1
-    forwarded: Program = manager._create_agentspeak_and_send_to_bdi.await_args[0][0]
+    assert manager._send_to_bdi.await_count == 1
+    forwarded: Program = manager._send_to_bdi.await_args[0][0]
     assert forwarded.phases[0].norms[0].norm == "N1"
     assert forwarded.phases[0].goals[0].description == "G1"
+
+    # Verify history clear was triggered
+    assert manager._send_clear_llm_history.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_send_clear_llm_history(mock_settings):
+    # Ensure the mock returns a string for the agent name (just like in your LLM tests)
+    mock_settings.agent_settings.llm_agent_name = "llm_agent"
+
+    manager = BDIProgramManager(name="program_manager_test")
+    manager.send = AsyncMock()
+
+    await manager._send_clear_llm_history()
+
+    assert manager.send.await_count == 1
+    msg: InternalMessage = manager.send.await_args[0][0]
+
+    # Verify the content and recipient
+    assert msg.body == "clear_history"
+    assert msg.to == "llm_agent"

test/unit/agents/communication/test_ri_communication_agent.py

@@ -67,6 +67,7 @@ async def test_setup_success_connects_and_starts_robot(zmq_context):
         address="tcp://localhost:5556",
         bind=False,
         gesture_data=[],
+        single_gesture_data=[],
     )
     agent.add_behavior.assert_called_once()
 

test/unit/agents/llm/test_llm_agent.py

@@ -197,6 +197,9 @@ async def test_query_llm_yields_final_tail_chunk(mock_settings):
     agent = LLMAgent("llm_agent")
     agent.send = AsyncMock()
 
+    agent.logger = MagicMock()
+    agent.logger.llm = MagicMock()
+
     # Patch _stream_query_llm to yield tokens that do NOT end with punctuation
     async def fake_stream(messages):
         yield "Hello"
@@ -262,3 +265,23 @@ async def test_stream_query_llm_skips_non_data_lines(mock_httpx_client, mock_set
 
     # Only the valid 'data:' line should yield content
     assert tokens == ["Hi"]
+
+
+@pytest.mark.asyncio
+async def test_clear_history_command(mock_settings):
+    """Test that the 'clear_history' message clears the agent's memory."""
+    # setup LLM to have some history
+    mock_settings.agent_settings.bdi_program_manager_name = "bdi_program_manager_agent"
+    agent = LLMAgent("llm_agent")
+    agent.history = [
+        {"role": "user", "content": "Old conversation context"},
+        {"role": "assistant", "content": "Old response"},
+    ]
+    assert len(agent.history) == 2
+    msg = InternalMessage(
+        to="llm_agent",
+        sender=mock_settings.agent_settings.bdi_program_manager_name,
+        body="clear_history",
+    )
+    await agent.handle_message(msg)
+    assert len(agent.history) == 0

test/unit/agents/user_interrupt/test_user_interrupt.py

@@ -0,0 +1,146 @@
+import asyncio
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from control_backend.agents.user_interrupt.user_interrupt_agent import UserInterruptAgent
+from control_backend.core.agent_system import InternalMessage
+from control_backend.core.config import settings
+from control_backend.schemas.ri_message import RIEndpoint
+
+
+@pytest.fixture
+def agent():
+    agent = UserInterruptAgent(name="user_interrupt_agent")
+    agent.send = AsyncMock()
+    agent.logger = MagicMock()
+    agent.sub_socket = AsyncMock()
+    return agent
+
+
+@pytest.mark.asyncio
+async def test_send_to_speech_agent(agent):
+    """Verify speech command format."""
+    await agent._send_to_speech_agent("Hello World")
+
+    agent.send.assert_awaited_once()
+    sent_msg: InternalMessage = agent.send.call_args.args[0]
+
+    assert sent_msg.to == settings.agent_settings.robot_speech_name
+    body = json.loads(sent_msg.body)
+    assert body["data"] == "Hello World"
+    assert body["is_priority"] is True
+
+
+@pytest.mark.asyncio
+async def test_send_to_gesture_agent(agent):
+    """Verify gesture command format."""
+    await agent._send_to_gesture_agent("wave_hand")
+
+    agent.send.assert_awaited_once()
+    sent_msg: InternalMessage = agent.send.call_args.args[0]
+
+    assert sent_msg.to == settings.agent_settings.robot_gesture_name
+    body = json.loads(sent_msg.body)
+    assert body["data"] == "wave_hand"
+    assert body["is_priority"] is True
+    assert body["endpoint"] == RIEndpoint.GESTURE_SINGLE.value
+
+
+@pytest.mark.asyncio
+async def test_send_to_program_manager(agent):
+    """Verify belief update format."""
+    context_str = "2"
+
+    await agent._send_to_program_manager(context_str)
+
+    agent.send.assert_awaited_once()
+    sent_msg: InternalMessage = agent.send.call_args.args[0]
+
+    assert sent_msg.to == settings.agent_settings.bdi_program_manager_name
+    assert sent_msg.thread == "belief_override_id"
+
+    body = json.loads(sent_msg.body)
+
+    assert body["belief"] == context_str
+
+
+@pytest.mark.asyncio
+async def test_receive_loop_routing_success(agent):
+    """
+    Test that the loop correctly:
+    1. Receives 'button_pressed' topic from ZMQ
+    2. Parses the JSON payload to find 'type' and 'context'
+    3. Calls the correct handler method based on 'type'
+    """
+    # Prepare JSON payloads as bytes
+    payload_speech = json.dumps({"type": "speech", "context": "Hello Speech"}).encode()
+    payload_gesture = json.dumps({"type": "gesture", "context": "Hello Gesture"}).encode()
+    payload_override = json.dumps({"type": "override", "context": "Hello Override"}).encode()
+
+    agent.sub_socket.recv_multipart.side_effect = [
+        (b"button_pressed", payload_speech),
+        (b"button_pressed", payload_gesture),
+        (b"button_pressed", payload_override),
+        asyncio.CancelledError,  # Stop the infinite loop
+    ]
+
+    agent._send_to_speech_agent = AsyncMock()
+    agent._send_to_gesture_agent = AsyncMock()
+    agent._send_to_program_manager = AsyncMock()
+
+    try:
+        await agent._receive_button_event()
+    except asyncio.CancelledError:
+        pass
+
+    await asyncio.sleep(0)
+
+    # Speech
+    agent._send_to_speech_agent.assert_awaited_once_with("Hello Speech")
+
+    # Gesture
+    agent._send_to_gesture_agent.assert_awaited_once_with("Hello Gesture")
+
+    # Override
+    agent._send_to_program_manager.assert_awaited_once_with("Hello Override")
+
+    assert agent._send_to_speech_agent.await_count == 1
+    assert agent._send_to_gesture_agent.await_count == 1
+    assert agent._send_to_program_manager.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_receive_loop_unknown_type(agent):
+    """Test that unknown 'type' values in the JSON log a warning and do not crash."""
+
+    # Prepare a payload with an unknown type
+    payload_unknown = json.dumps({"type": "unknown_thing", "context": "some_data"}).encode()
+
+    agent.sub_socket.recv_multipart.side_effect = [
+        (b"button_pressed", payload_unknown),
+        asyncio.CancelledError,
+    ]
+
+    agent._send_to_speech_agent = AsyncMock()
+    agent._send_to_gesture_agent = AsyncMock()
+    agent._send_to_belief_collector = AsyncMock()
+
+    try:
+        await agent._receive_button_event()
+    except asyncio.CancelledError:
+        pass
+
+    await asyncio.sleep(0)
+
+    # Ensure no handlers were called
+    agent._send_to_speech_agent.assert_not_called()
+    agent._send_to_gesture_agent.assert_not_called()
+    agent._send_to_belief_collector.assert_not_called()
+
+    agent.logger.warning.assert_called_with(
+        "Received button press with unknown type '%s' (context: '%s').",
+        "unknown_thing",
+        "some_data",
+    )

uv.lock

@@ -997,7 +997,6 @@ dependencies = [
     { name = "pydantic" },
     { name = "pydantic-settings" },
     { name = "python-json-logger" },
-    { name = "python-slugify" },
     { name = "pyyaml" },
     { name = "pyzmq" },
     { name = "silero-vad" },
@@ -1047,7 +1046,6 @@ requires-dist = [
     { name = "pydantic", specifier = ">=2.12.0" },
     { name = "pydantic-settings", specifier = ">=2.11.0" },
     { name = "python-json-logger", specifier = ">=4.0.0" },
-    { name = "python-slugify", specifier = ">=8.0.4" },
     { name = "pyyaml", specifier = ">=6.0.3" },
     { name = "pyzmq", specifier = ">=27.1.0" },
     { name = "silero-vad", specifier = ">=6.0.0" },
@@ -1343,18 +1341,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/45/58/38b5afbc1a800eeea951b9285d3912613f2603bdf897a4ab0f4bd7f405fc/python_multipart-0.0.20-py3-none-any.whl", hash = "sha256:8a62d3a8335e06589fe01f2a3e178cdcc632f3fbe0d492ad9ee0ec35aab1f104", size = 24546, upload-time = "2024-12-16T19:45:44.423Z" },
 ]
 
-[[package]]
-name = "python-slugify"
-version = "8.0.4"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "text-unidecode" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/87/c7/5e1547c44e31da50a460df93af11a535ace568ef89d7a811069ead340c4a/python-slugify-8.0.4.tar.gz", hash = "sha256:59202371d1d05b54a9e7720c5e038f928f45daaffe41dd10822f3907b937c856", size = 10921, upload-time = "2024-02-08T18:32:45.488Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/a4/62/02da182e544a51a5c3ccf4b03ab79df279f9c60c5e82d5e8bec7ca26ac11/python_slugify-8.0.4-py2.py3-none-any.whl", hash = "sha256:276540b79961052b66b7d116620b36518847f52d5fd9e3a70164fc8c50faa6b8", size = 10051, upload-time = "2024-02-08T18:32:43.911Z" },
-]
-
 [[package]]
 name = "pyyaml"
 version = "6.0.3"
@@ -1878,15 +1864,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a2/09/77d55d46fd61b4a135c444fc97158ef34a095e5681d0a6c10b75bf356191/sympy-1.14.0-py3-none-any.whl", hash = "sha256:e091cc3e99d2141a0ba2847328f5479b05d94a6635cb96148ccb3f34671bd8f5", size = 6299353, upload-time = "2025-04-27T18:04:59.103Z" },
 ]
 
-[[package]]
-name = "text-unidecode"
-version = "1.3"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/ab/e2/e9a00f0ccb71718418230718b3d900e71a5d16e701a3dae079a21e9cd8f8/text-unidecode-1.3.tar.gz", hash = "sha256:bad6603bb14d279193107714b288be206cac565dfa49aa5b105294dd5c4aab93", size = 76885, upload-time = "2019-08-30T21:36:45.405Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/a6/a5/c0b6468d3824fe3fde30dbb5e1f687b291608f9473681bbf7dabbf5a87d7/text_unidecode-1.3-py2.py3-none-any.whl", hash = "sha256:1311f10e8b895935241623731c2ba64f4c455287888b18189350b67134a822e8", size = 78154, upload-time = "2019-08-30T21:37:03.543Z" },
-]
-
 [[package]]
 name = "tiktoken"
 version = "0.12.0"