def __init__(
        self,
        session_id: str,
        workspace: Workspace,
        app_injector: Injector,
        logger: TelemetryLogger,
        tracing: Tracing,
        config: AppSessionConfig,  # TODO: change to SessionConfig
        role_registry: RoleRegistry,
    ) -> None:
        """
        Initialize the session.
        :param session_id: The session ID.
        :param workspace: The workspace.
        :param app_injector: The app injector.
        :param logger: The logger.
        :param tracing: The tracing.
        :param config: The configuration.
        :param role_registry: The role registry.
        """
        assert session_id is not None, "session_id must be provided"
        self.logger = logger
        self.tracing = tracing
        self.session_injector = app_injector.create_child_injector()
        self.config = config

        self.session_id: str = session_id

        self.workspace = workspace.get_session_dir(self.session_id)
        self.execution_cwd = os.path.join(self.workspace, "cwd")

        self.metadata = SessionMetadata(
            session_id=self.session_id,
            workspace=self.workspace,
            execution_cwd=self.execution_cwd,
        )
        self.session_injector.binder.bind(SessionMetadata, self.metadata)

        self._init()

        self.round_index = 0
        self.memory = Memory(session_id=self.session_id)

        self.session_var: Dict[str, str] = {}

        self.event_emitter = self.session_injector.get(SessionEventEmitter)
        self.session_injector.binder.bind(SessionEventEmitter, self.event_emitter)

        self.role_registry = role_registry
        self.worker_instances = {}
        for role_name in self.config.roles:
            if role_name == "planner":
                continue
            if role_name not in role_registry.get_role_name_list():
                raise ValueError(f"Unknown role {role_name}")
            role_entry = self.role_registry.get(role_name)
            role_instance = self.session_injector.create_object(
                role_entry.module,
                {
                    "role_entry": role_entry,
                },
            )
            self.session_injector.binder.bind(role_entry.module, role_instance)
            self.worker_instances[role_instance.get_alias()] = role_instance

        if "planner" in self.config.roles:
            self.planner = self.session_injector.create_object(
                Planner,
                {
                    "workers": self.worker_instances,
                },
            )
            self.session_injector.binder.bind(Planner, self.planner)

        self.max_internal_chat_round_num = self.config.max_internal_chat_round_num
        self.internal_chat_num = 0

        self.logger.dump_log_file(
            self,
            file_path=os.path.join(self.workspace, f"{self.session_id}.json"),
        )

        Initialize the session.
        :param session_id: The session ID.
        :param workspace: The workspace.
        :param app_injector: The app injector.
        :param logger: The logger.
        :param tracing: The tracing.
        :param config: The configuration.
        :param role_registry: The role registry.
        
taskweaver/session/session.py