Source code for council.runners.while_runner

from council.contexts import ChainContext, ChatMessage
from council.runners import RunnerBase, RunnerPredicate, RunnerExecutor, RunnerPredicateError




class While(RunnerBase):
    """
    Runner that executes while the given predicate returns `True`
    """



    def __init__(self, predicate: RunnerPredicate, runner: RunnerBase):
        """
        Args:
            predicate: a predicate function evaluated in the while loop
            runner: a runner to be executed while the predicate returns `True`
        """
        super().__init__("whileRunner")
        self._predicate = predicate
        self._body = self.new_monitor("whileBody", runner)


    def _run(self, context: ChainContext, executor: RunnerExecutor):
        while self.check_predicate(context):
            self._body.inner.run(context, executor)

    def check_predicate(self, context: ChainContext) -> bool:
        try:
            return self._predicate(context)
        except Exception as e:
            context.append(ChatMessage.skill("WhileRunner", f"Predicate raised exception: {e}", is_error=True))
            raise RunnerPredicateError from e