Class OSInteraction

src/server/tasks/os_interaction/task.py:146–736

class OSInteraction(Task):

    def __init__(self,
                 data_config,
                 docker_config,
                 round_limit=8,
                 tools=None,
                 env_driver: str = 'docker',
                 env_options: Optional[dict] = None,
                 **kwargs):
        super().__init__(**kwargs)
        self.round_limit: int = round_limit
        self.data_config = data_config
        self.docker_config = docker_config
        self.tools = tools
        self.full_async = True
        self.problem_configs: Dict[str, Dict[str, Any]] = {}  # {index: CONFIG}

        matches = []
        for item in self.data_config["files"]:
            path = item["problem_file"]
            for file in glob.glob(path):
                if file.endswith(".json") or file.endswith(".jsonl"):
                    matches.append(
                        {
                            "problem_file": file,
                            "script_dir": item["script_dir"],
                            "index_prefix": item["index_prefix"]
                                            + os.path.basename(file)
                                            .removesuffix(".json")
                                            .removesuffix(".jsonl")
                                            + "-",
                        }
                    )
        self.data_config["files"] = matches

        next_idx = 0
        for item in self.data_config["files"]:
            problem_file = item["problem_file"]
            single_file_configs = self._load_configs(problem_file, item["script_dir"])
            dict_configs = {}
            for config in single_file_configs:
                dict_configs[next_idx] = {
                    "file": problem_file,
                    "config": config,
                    "index": next_idx,
                }
                next_idx += 1
            self.problem_configs.update(dict_configs)

        logging.info(f"Initialized OSInteraction with {len(self.problem_configs)} problem configs")

        self.env_delegation = OSEnvironmentDelegation(self.docker_config['localhost'])
        self.env_controller = create_controller(env_driver, self.env_delegation, **env_options)
        self.env_controller_background_task = None

    def _load_configs(self, config_path, script_root_dir=".") -> List[JudgeConfig]:
        def load_script(script_obj):