Method stream_message

ohmo/gateway/runtime.py:212–313

Submit an inbound channel message and yield progress + final reply updates.

(self, message: InboundMessage, session_key: str)

        return bundle

    async def stream_message(self, message: InboundMessage, session_key: str):
        """Submit an inbound channel message and yield progress + final reply updates."""
        user_message = _build_inbound_user_message(message)
        user_prompt = user_message.text
        command_prompt = (message.content or "").strip()
        session_cwd = self._cwd_for_message(message)
        bundle = await self.get_bundle(session_key, latest_user_prompt=user_prompt, cwd=session_cwd)
        logger.info(
            "ohmo runtime processing start channel=%s chat_id=%s session_key=%s session_id=%s content=%r",
            message.channel,
            message.chat_id,
            session_key,
            bundle.session_id,
            _content_snippet(user_prompt),
        )

        command_context: CommandContext | None = None

        def get_command_context() -> CommandContext:
            nonlocal command_context
            if command_context is None:
                command_context = CommandContext(
                    engine=bundle.engine,
                    hooks_summary=getattr(bundle, "hook_summary", lambda: "")(),
                    mcp_summary=getattr(bundle, "mcp_summary", lambda: "")(),
                    plugin_summary=getattr(bundle, "plugin_summary", lambda: "")(),
                    cwd=getattr(bundle, "cwd", str(self._cwd)),
                    tool_registry=getattr(bundle, "tool_registry", None),
                    app_state=getattr(bundle, "app_state", None),
                    session_backend=getattr(bundle, "session_backend", self._session_backend),
                    session_id=getattr(bundle, "session_id", None),
                    extra_skill_dirs=getattr(bundle, "extra_skill_dirs", ()),
                    extra_plugin_roots=getattr(bundle, "extra_plugin_roots", ()),
                    memory_backend=create_memory_command_backend(self._workspace),
                    include_project_memory=False,
                )
            return command_context

        parsed = bundle.commands.lookup(command_prompt)
        if parsed is None and not message.media:
            parsed = lookup_skill_slash_command(command_prompt, get_command_context())
        if parsed is not None and not message.media:
            command, args = parsed
            command_name = str(getattr(command, "name", "") or "")
            gateway_result = self._handle_gateway_scoped_command(command_name, args)
            if gateway_result is not None:
                message_text, refresh_runtime = gateway_result
                result = CommandResult(message=message_text, refresh_runtime=refresh_runtime)
                async for update in self._stream_command_result(
                    bundle=bundle,
                    message=message,
                    session_key=session_key,
                    user_prompt=user_prompt,
                    result=result,
                ):
                    yield update
                return
            remote_allowed = getattr(command, "remote_invocable", True)