Method run_async

tests/unittests/workflow/testing_utils.py:328–346

(
      self,
      new_message: Optional[types.ContentUnion] = None,
      invocation_id: Optional[str] = None,
  )

    return False

  async def run_async(
      self,
      new_message: Optional[types.ContentUnion] = None,
      invocation_id: Optional[str] = None,
  ) -> list[Event]:
    # For non-resumable apps, don't reuse invocation_id on resume.
    # State reconstruction relies on scanning events from *previous*
    # invocations, so the resume call must get a fresh invocation_id.
    if invocation_id and not self.is_resumable:
      invocation_id = None
    events = []
    async for event in self.runner.run_async(
        user_id=self.session.user_id,
        session_id=self.session.id,
        invocation_id=invocation_id,
        new_message=get_user_content(new_message) if new_message else None,
    ):
      events.append(event)
    return events

  def run_live(
      self, live_request_queue: LiveRequestQueue, run_config: RunConfig = None