API Reference

class CreateSandboxBashSessionRequest(BaseModel):
    startup_source: list[str] = []
    startup_cmd: str = ""
    """The Startup command to run inside the sandbox such as mount, unshare, etc."""
    session: str = "default"
    session_type: Literal["bash"] = "bash"
    startup_timeout: float = 1.0
    """The timeout for the startup commands."""

class BashSession(Session):
    _UNIQUE_STRING = "UNIQUESTRING29234"

    def __init__(self, request: CreateSandboxBashSessionRequest, *, logger: logging.Logger | None = None):

        self.request = request
        self._ps1 = "SHELLPS1PREFIX"
        self._shell: pexpect.spawn | None = None
        self.logger = logger or get_logger("rex-session")
        self.create_cmd=None
    @property
    def shell(self) -> pexpect.spawn:
        if self._shell is None:
            msg = "shell not initialized"
            raise RuntimeError(msg)
        return self._shell

    def _get_reset_commands(self) -> list[str]:

        return [
            f"export PS1='{self._ps1}'",
            "export PS2=''",
            "export PS0=''",
        ]

    async def start(self) -> CreateBashSessionResponse:
        """Starts a bash session based on the request.startup_cmd."""
        self._shell = pexpect.spawn(
            self.request.startup_cmd,
            encoding="utf-8",
            codec_errors="backslashreplace",
            echo=False,
            env=dict(os.environ.copy(), **{"PS1": self._ps1, "PS2": "", "PS0": ""}),  # type: ignore
        )

        time.sleep(0.3)
        cmds = []
        if self.request.startup_source:
            cmds += [f"source {path}" for path in self.request.startup_source] + ["sleep 0.3"]
        cmds += self._get_reset_commands()
        cmd = " ; ".join(cmds)
        self.shell.sendline(cmd)
        self.shell.expect(self._ps1, timeout=self.request.startup_timeout)

        output = _strip_control_chars(self.shell.before)  # type: ignore
        return CreateBashSessionResponse(output=output)

    def _eat_following_output(self, timeout: float = 0.5) -> str:

        time.sleep(timeout)
        try:
            output = self.shell.read_nonblocking(timeout=0.1)
        except pexpect.TIMEOUT:
            return ""
        return _strip_control_chars(output)

    async def interrupt(self, action: BashInterruptAction) -> BashObservation:

        output = ""
        for _ in range(action.n_retry):
            self.shell.sendintr()
            expect_strings = action.expect + [self._ps1]
            try:
                expect_index = self.shell.expect(expect_strings, timeout=action.timeout)  # type: ignore
                matched_expect_string = expect_strings[expect_index]
            except Exception:
                time.sleep(0.2)
                continue
            output += _strip_control_chars(self.shell.before)  # type: ignore
            output += self._eat_following_output()
            output = output.strip()
            return BashObservation(output=output, exit_code=0, expect_string=matched_expect_string)
        # Fall back to putting job to background and killing it there:
        try:
            self.shell.sendcontrol("z")
            self.shell.expect(expect_strings, timeout=action.timeout)
            output += self.shell.before
            self.shell.sendline("kill -9 %1")
            expect_index = self.shell.expect(expect_strings, timeout=action.timeout)  # type: ignore
            matched_expect_string = expect_strings[expect_index]
            output += self.shell.before
            output += self._eat_following_output()
            output = output.strip()
            return BashObservation(output=output, exit_code=0, expect_string=matched_expect_string)
        except pexpect.TIMEOUT:
            msg = "Failed to interrupt session"
            raise pexpect.TIMEOUT(msg)

    async def run(self, action: BashAction | BashInterruptAction) -> BashObservation:

        if self.shell is None:
            msg = "shell not initialized"
            raise SessionNotInitializedError(msg)
        if isinstance(action, BashInterruptAction):
            return await self.interrupt(action)
        if action.is_interactive_command or action.is_interactive_quit:
            return await self._run_interactive(action)
        r = await self._run_normal(action)

        if action.check == "raise" and r.exit_code != 0:
            msg = f"Command {action.command!r} failed with exit code {r.exit_code}. Here is the output:\n{r.output!r}"
            if action.error_msg:
                msg = f"{action.error_msg}: {msg}"
            raise NonZeroExitCodeError(msg)
        return r

    async def _run_interactive(self, action: BashAction) -> BashObservation:

        assert self.shell is not None
        self.shell.sendline(action.command)
        expect_strings = action.expect + [self._ps1]
        try:
            expect_index = self.shell.expect(expect_strings, timeout=action.timeout)  # type: ignore
            matched_expect_string = expect_strings[expect_index]
        except pexpect.TIMEOUT as e:
            msg = f"timeout after {action.timeout} seconds while running command {action.command!r}"
            raise CommandTimeoutError(msg) from e
        output: str = _strip_control_chars(self.shell.before)  # type: ignore
        if action.is_interactive_quit:
            assert not action.is_interactive_command
            self.shell.setecho(False)
            self.shell.waitnoecho()
            self.shell.sendline(f"stty -echo; echo '{self._UNIQUE_STRING}'")
            # Might need two expects for some reason
            self.shell.expect(self._UNIQUE_STRING, timeout=1)
            self.shell.expect(self._ps1, timeout=1)
        else:
            # Interactive command.
            # For some reason, this often times enables echo mode within the shell.
            output = output.lstrip().removeprefix(action.command).strip()

        return BashObservation(output=output, exit_code=0, expect_string=matched_expect_string)

    async def _run_normal(self, action: BashAction) -> BashObservation:
        """Run a normal action. This is the default mode.

        There are three steps to this:

        1. Check if the command is valid
        2. Execute the command
        3. Get the exit code
        """
        action = deepcopy(action)

        assert self.shell is not None
        _check_bash_command(action.command)

        # Part 2: Execute the command

        fallback_terminator = False
        # Running multiple interactive commands by sending them with linebreaks would break things
        # because we get multiple PS1s back to back. Instead we just join them with ;
        # However, sometimes bashlex errors and we can't do this. In this case
        # we add a unique string to the end of the command and then seek to that
        # (which is also somewhat brittle, so we don't do this by default).
        try:
            individual_commands = _split_bash_command(action.command)
        except Exception as e:
            # Bashlex is very buggy and can throw a variety of errors, including
            # ParsingErrors, NotImplementedErrors, TypeErrors, possibly more. So we catch them all
            self.logger.error("Bashlex fail: %s", e)
            action.command += f"\n TMPEXITCODE=$? ; sleep 0.1; echo -n '{self._UNIQUE_STRING}' ; (exit $TMPEXITCODE)"
            fallback_terminator = True
        else:
            action.command = " ; ".join(individual_commands)
        self.shell.sendline(action.command)
        if not fallback_terminator:
            expect_strings = action.expect + [self._ps1]
        else:
            expect_strings = [self._UNIQUE_STRING]
        try:
            expect_index = self.shell.expect(expect_strings, timeout=action.timeout)  # type: ignore
            matched_expect_string = expect_strings[expect_index]
        except pexpect.TIMEOUT as e:
            msg = f"timeout after {action.timeout} seconds while running command {action.command!r}"
            raise CommandTimeoutError(msg) from e
        output: str = _strip_control_chars(self.shell.before)  # type: ignore

        # Part 3: Get the exit code
        if action.check == "ignore":
            return BashObservation(output=output, exit_code=None, expect_string=matched_expect_string)

        try:
            _exit_code_prefix = "EXITCODESTART"
            _exit_code_suffix = "EXITCODEEND"
            self.shell.sendline(f"\necho {_exit_code_prefix}$?{_exit_code_suffix}")
            try:
                self.shell.expect(_exit_code_suffix, timeout=1)
            except pexpect.TIMEOUT:
                msg = "timeout while getting exit code"
                raise NoExitCodeError(msg)
            exit_code_raw: str = _strip_control_chars(self.shell.before)  # type: ignore
            exit_code = re.findall(f"{_exit_code_prefix}([0-9]+)", exit_code_raw)
            if len(exit_code) != 1:
                msg = f"failed to parse exit code from output {exit_code_raw!r} (command: {action.command!r}, matches: {exit_code})"
                raise NoExitCodeError(msg)
            output += exit_code_raw.split(_exit_code_prefix)[0]
            exit_code = int(exit_code[0])
            # We get at least one more PS1 here.
            try:
                self.shell.expect(self._ps1, timeout=0.1)
            except pexpect.TIMEOUT:
                msg = "Timeout while getting PS1 after exit code extraction"
                raise CommandTimeoutError(msg)
            output = output.replace(self._UNIQUE_STRING, "").replace(self._ps1, "")
        except Exception:
            # Ignore all exceptions if check == 'silent'
            if action.check == "raise":
                raise
            exit_code = None
        return BashObservation(output=output, exit_code=exit_code, expect_string=matched_expect_string)

    async def close(self) -> CloseSessionResponse:
        if self._shell is None:
            return CloseBashSessionResponse()
        self.shell.close()
        self._shell = None
        return CloseBashSessionResponse()

    def interact(self) -> None:

        self.shell.interact()

class DockerDeployment(AbstractDeployment):
    def __init__(
        self,
        *,
        logger: logging.Logger | None = None,
        **kwargs: Any,
    ):

        self._config = DockerDeploymentConfig(**kwargs)
        self._runtime: RemoteRuntime | None = None
        self._container_process = None
        self._container_name = None
        self.logger = logger or get_logger("rex-deploy")
        self._runtime_timeout = 0.15
        self._hooks = CombinedDeploymentHook()

    def add_hook(self, hook: DeploymentHook):
        self._hooks.add_hook(hook)

    @classmethod
    def from_config(cls, config: DockerDeploymentConfig) -> Self:
        return cls(**config.model_dump())

    def apply_test_patch(self,patch):
        """Applies a patch to the testbed repository inside the container."""
        GIT_APPLY_CMDS = [
        "git apply --verbose",
        "git apply --verbose --reject",
        "patch --batch --fuzz=5 -p1 -i",
        ]
        #asyncio.run(self.runtime.run_in_session(BashAction(command=reset_command, timeout=120, check='raise')))

    def reset_swesmith_tests(self):
        """
        resets the tests in the testbed repository inside the container to their original state.
        """
        f2p_files = list(set([x.split("::", 1)[0] for x in self.ds[FAIL_TO_PASS]]))
        p2p_files = list(set([x.split("::", 1)[0] for x in self.ds[PASS_TO_PASS]]))
        all_files = list(set(f2p_files + p2p_files))
        all_files = [f for f in all_files if 
             os.path.basename(f).startswith('test_') and os.path.basename(f).endswith('.py') or
             os.path.basename(f).endswith('_test.py')]
        commit_id ='origin/main'
        reset_command = (
            f'cd "/testbed" && '
            f'printf "%s\\n" {" ".join(all_files)} | '
            f'xargs -n1 -I{{}} git checkout {commit_id} -- "{{}}" 2>/dev/null'
        )
        asyncio.run(self.runtime.run_in_session(BashAction(command=reset_command, timeout=60, check='raise')))
    # def setup_env_swesmith(self):
    #     try:
    #         commit_id = self.ds['base_commit']
    #         self.run("git fetch")
    #         self.run(f"git checkout {commit_id}")
    #         # Setup the run_test.sh script for subsequent testing.  
    #         test_command, _ = get_test_command(self.ds)
    #         eval_script_content = "\n".join(
    #             [
    #                 "#!/bin/bash",
    #                 "set -uxo pipefail",
    #                 "source /opt/miniconda3/bin/activate",
    #                 f"conda activate testbed",
    #                 f"cd testbed/",
    #                 f": '>>>>> Start Test Output'",
    #                 test_command,
    #                 f": '>>>>> End Test Output'",
    #             ]
    #         ) + "\n"

    #         with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.sh') as temp_file:
    #             temp_file.write(eval_script_content)
    #             temp_file.flush()  # Ensure content is written to disk
    #             temp_file_path = temp_file.name

    #         # Copy the file to container and clean up
    #         self.copy_to_container(temp_file_path, "/run_tests.sh")
    #         os.unlink(temp_file_path)  # Clean up the temporary file

    #         self.run("chmod +x /run_tests.sh")

    #         # Ensure can call and execute the tools in /usr/local/bin.
    #         self.run(f"ln -s /opt/miniconda3/envs/testbed /root/.venv")
    #         self.run('echo \'export PATH="/usr/local/bin:$PATH"\' >> ~/.bashrc')
    #         self.run("python -m pip install chardet")
    #     except Exception as e:
    #         self.logger.error(f"Error setting up environment: {repr(e)}")
    def setup_env_swesmith(self):
        """
        sets up the environment for running tests in the testbed repository inside the container.
        """

        test_command, _ = get_test_command(self.ds)

        eval_script_content = "\n".join(
            [
                "#!/bin/bash",
                "set -uxo pipefail",
                f"source /opt/miniconda3/bin/activate",
                f"conda activate testbed",
                f"cd /testbed",
                f": '>>>>> Start Test Output'",
                test_command,
                f": '>>>>> End Test Output'",
            ]
        ) + "\n"
        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.sh') as temp_file:
            temp_file.write(eval_script_content)
            temp_file.flush()  # Ensure content is written to disk
            temp_file_path = temp_file.name

        # Copy the file to container and clean up

        asyncio.run(self.runtime.upload(UploadRequest(source_path=temp_file_path,target_path="/run_tests.sh")))
        os.unlink(temp_file_path)  # Clean up the temporary file
        # self.run_command("chmod +x /run_tests.sh && python -m pip install chardet")

        #asyncio.run(self.runtime.run_in_session(BashAction(command='"ln -s /opt/miniconda3/envs/testbed /root/.venv"', timeout=120, check='raise')))
        asyncio.run(self.runtime.run_in_session(BashAction(command='echo \'export PATH="/usr/local/bin:$PATH"\' >> ~/.bashrc', timeout=10, check='raise')))
        asyncio.run(self.runtime.run_in_session(BashAction(command=f"chmod +x /run_tests.sh && python -m pip install chardet", timeout=60, check='raise')))
    def parse_logs(self, log_output: str) -> dict:
        """Parses the log output from the testbed repository inside the container."""
        return parse_log_fn("testbed")(log_output)
    def _calculate_reward(self):
        """Calculates the reward based on the test results from the testbed repository inside the container."""
        timeout= self._config.eval_timeout
        self.reset_swesmith_tests()
        self.setup_env_swesmith()

        output= asyncio.run(self.runtime.run_in_session(BashAction(command="/run_tests.sh", timeout=timeout, check='raise'))).output

        # print(output)
        parse = self.parse_logs(output)

        fail2pass = [ ".".join(line.split("::")[1:]) for line in self.ds['FAIL_TO_PASS']]
        pass2pass = [ ".".join(line.split("::")[1:]) for line in self.ds['PASS_TO_PASS']]
        # @(Naman, Jas): Parse the output and return the reward. This implementation is a hack rn.
        if not parse:
            return 0.0,{},{},output

        fail2pass_dic={}
        for test_name in fail2pass:
            if test_name not in parse:
                # Check if test_name is substring of any key
                matching_key = next((k for k in parse.keys() if test_name in k), None)

                fail2pass_dic[test_name]=matching_key


            else:

                fail2pass_dic[test_name]=parse[test_name]


        # Check pass2pass
        pass2pass_dic={}
        for test_name in pass2pass:
            if test_name not in parse:
                # Check if test_name is substring of any key
                matching_key = next((k for k in parse.keys() if test_name in k), None)
                pass2pass_dic[test_name]=matching_key

                test_name = matching_key
            else:
                pass2pass_dic[test_name]=parse[test_name]

        # Check fail2pass
        for test_name in fail2pass:
            if test_name not in parse:
                # Check if test_name is substring of any key
                matching_key = next((k for k in parse.keys() if test_name in k), None)
                if matching_key is None:
                    return 0.0,fail2pass_dic,pass2pass_dic,output
                if parse[matching_key] != 'PASSED':
                    return 0.0,fail2pass_dic,pass2pass_dic,output
                test_name = matching_key
            if parse[test_name] != 'PASSED':
                return 0.0,fail2pass_dic,pass2pass_dic,output

        # Check pass2pass
        for test_name in pass2pass:
            if test_name not in parse:
                # Check if test_name is substring of any key
                matching_key = next((k for k in parse.keys() if test_name in k), None)
                if matching_key is None:
                    return 0.0,fail2pass_dic,pass2pass_dic,output
                test_name = matching_key
            if parse[test_name] != 'PASSED':
                return 0.0,fail2pass_dic,pass2pass_dic,output
        return 1.0,fail2pass_dic,pass2pass_dic,output

    def _get_container_name(self) -> str:
        """Returns a unique container name based on the image name."""
        image_name_sanitized = "".join(c for c in self._config.image if c.isalnum() or c in "-_.")
        return f"{image_name_sanitized}-{uuid.uuid4()}"

    @property
    def container_name(self) -> str | None:
        return self._container_name

    async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:

        if self._runtime is None:
            msg = "Runtime not started"
            raise RuntimeError(msg)
        if self._container_process is None:
            msg = "Container process not started"
            raise RuntimeError(msg)
        if self._container_process.poll() is not None:
            msg = "Container process terminated."
            output = "stdout:\n" + self._container_process.stdout.read().decode()  # type: ignore
            output += "\nstderr:\n" + self._container_process.stderr.read().decode()  # type: ignore
            msg += "\n" + output
            raise RuntimeError(msg)
        return await self._runtime.is_alive(timeout=timeout)

    async def _wait_until_alive(self, timeout: float = 10.0):
        try:
            return await _wait_until_alive(self.is_alive, timeout=timeout, function_timeout=self._runtime_timeout)
        except TimeoutError as e:
            self.logger.error("Runtime did not start within timeout. Here's the output from the container process.")
            self.logger.error(self._container_process.stdout.read().decode())  # type: ignore
            self.logger.error(self._container_process.stderr.read().decode())  # type: ignore
            assert self._container_process is not None
            await self.stop()
            raise e

    def _get_token(self) -> str:
        return str(uuid.uuid4())

    def _get_swerex_start_cmd(self, token: str) -> list[str]:
        rex_args = f"--auth-token {token}"
        pipx_install = "python3 -m pip install pipx && python3 -m pipx ensurepath"
        if self._config.python_standalone_dir:
            cmd = f"{self._config.python_standalone_dir}/python3.11/bin/{REMOTE_EXECUTABLE_NAME} {rex_args}"
        else:
            cmd = f"{REMOTE_EXECUTABLE_NAME} {rex_args} || ({pipx_install} && pipx run {PACKAGE_NAME} {rex_args})"
        # Need to wrap with /bin/sh -c to avoid having '&&' interpreted by the parent shell
        return [
            "/bin/sh",
            # "-l",
            "-c",
            cmd,
        ]

    def _pull_image(self) -> None:
        if self._config.pull == "never":
            return
        if self._config.pull == "missing" and _is_image_available(self._config.image, self._config.container_runtime):
            return
        self.logger.info(f"Pulling image {self._config.image!r}")
        self._hooks.on_custom_step("Pulling container image")
        try:
            _pull_image(self._config.image, self._config.container_runtime)
        except subprocess.CalledProcessError as e:
            msg = f"Failed to pull image {self._config.image}. "
            msg += f"Error: {e.stderr.decode()}"
            msg += f"Output: {e.output.decode()}"
            raise DockerPullError(msg) from e

    @property
    def glibc_dockerfile(self) -> str:
        # will only work with glibc-based systems
        if self._config.platform:
            platform_arg = f"--platform={self._config.platform}"
        else:
            platform_arg = ""
        return (
            "ARG BASE_IMAGE\n\n"
            # Build stage for standalone Python
            f"FROM {platform_arg} python:3.11.9-slim-bookworm AS builder\n"
            # Install build dependencies
            "RUN apt-get update && apt-get install -y \\\n"
            "    wget \\\n"
            "    gcc \\\n"
            "    make \\\n"
            "    zlib1g-dev \\\n"
            "    libssl-dev \\\n"
            "    && rm -rf /var/lib/apt/lists/*\n\n"
            # Download and compile Python as standalone
            "WORKDIR /build\n"
            "RUN wget https://www.python.org/ftp/python/3.11.8/Python-3.11.8.tgz \\\n"
            "    && tar xzf Python-3.11.8.tgz\n"
            "WORKDIR /build/Python-3.11.8\n"
            "RUN ./configure \\\n"
            "    --prefix=/root/python3.11 \\\n"
            "    --enable-shared \\\n"
            "    LDFLAGS='-Wl,-rpath=/root/python3.11/lib' && \\\n"
            "    make -j$(nproc) && \\\n"
            "    make install && \\\n"
            "    ldconfig\n\n"
            # Production stage
            f"FROM {platform_arg} $BASE_IMAGE\n"
            # Ensure we have the required runtime libraries
            "RUN apt-get update && apt-get install -y \\\n"
            "    libc6 \\\n"
            "    && rm -rf /var/lib/apt/lists/*\n"
            # Copy the standalone Python installation
            f"COPY --from=builder /root/python3.11 {self._config.python_standalone_dir}/python3.11\n"
            f"ENV LD_LIBRARY_PATH={self._config.python_standalone_dir}/python3.11/lib:${{LD_LIBRARY_PATH:-}}\n"
            # Verify installation
            f"RUN {self._config.python_standalone_dir}/python3.11/bin/python3 --version\n"
            # Install swe-rex using the standalone Python
            f"RUN /root/python3.11/bin/pip3 install --no-cache-dir {PACKAGE_NAME}\n\n"
            f"RUN ln -s /root/python3.11/bin/{REMOTE_EXECUTABLE_NAME} /usr/local/bin/{REMOTE_EXECUTABLE_NAME}\n\n"
            f"RUN {REMOTE_EXECUTABLE_NAME} --version\n"
        )

    def _build_image(self) -> str:
        """Builds image, returns image ID."""
        self.logger.info(
            f"Building image {self._config.image} to install a standalone python to {self._config.python_standalone_dir}. "
            "This might take a while (but you only have to do it once). To skip this step, set `python_standalone_dir` to None."
        )
        dockerfile = self.glibc_dockerfile
        platform_arg = []
        if self._config.platform:
            platform_arg = ["--platform", self._config.platform]
        build_cmd = [
            self._config.container_runtime,
            "build",
            "-q",
            *platform_arg,
            "--build-arg",
            f"BASE_IMAGE={self._config.image}",
            "-",
        ]
        image_id = (
            subprocess.check_output(
                build_cmd,
                input=dockerfile.encode(),
            )
            .decode()
            .strip()
        )
        if not image_id.startswith("sha256:"):
            msg = f"Failed to build image. Image ID is not a SHA256: {image_id}"
            raise RuntimeError(msg)
        return image_id

    async def start(self):

        self._pull_image()
        if self._config.python_standalone_dir:
            image_id = self._build_image()
        else:
            image_id = self._config.image
        if self._config.port is None:
            self._config.port = find_free_port()
        assert self._container_name is None
        self._container_name = self._get_container_name()
        token = self._get_token()
        platform_arg = []
        if self._config.platform is not None:
            platform_arg = ["--platform", self._config.platform]
        rm_arg = []
        if self._config.remove_container:
            rm_arg = ["--rm"]
        cmds = [
            self._config.container_runtime,
            "run",
            *rm_arg,
            "-p",
            f"{self._config.port}:8000",
            *platform_arg,
            *self._config.docker_args,
            "--name",
            self._container_name,
            image_id,
            *self._get_swerex_start_cmd(token),
        ]
        cmd_str = shlex.join(cmds)
        self.logger.info(
            f"Starting container {self._container_name} with image {self._config.image} serving on port {self._config.port}"
        )
        self.logger.debug(f"Command: {cmd_str!r}")
        # shell=True required for && etc.
        self._container_process = subprocess.Popen(cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self._hooks.on_custom_step("Starting runtime")
        self.logger.info(f"Starting runtime at {self._config.port}")
        self._runtime = RemoteRuntime.from_config(
            RemoteRuntimeConfig(port=self._config.port, timeout=self._runtime_timeout, auth_token=token,host=self._config.host)
        )
        t0 = time.time()
        await self._wait_until_alive(timeout=self._config.startup_timeout)
        self.logger.info(f"Runtime started in {time.time() - t0:.2f}s")

    async def stop(self):

        if self._runtime is not None:
            await self._runtime.close()
            self._runtime = None

        if self._container_process is not None:
            try:
                subprocess.check_call(
                    [self._config.container_runtime, "kill", self._container_name],  # type: ignore
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    timeout=10,
                )
            except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
                self.logger.warning(
                    f"Failed to kill container {self._container_name}: {e}. Will try harder.",
                    exc_info=False,
                )
            for _ in range(3):
                self._container_process.kill()
                try:
                    self._container_process.wait(timeout=5)
                    break
                except subprocess.TimeoutExpired:
                    continue
            else:
                self.logger.warning(f"Failed to kill container {self._container_name} with SIGKILL")

            self._container_process = None
            self._container_name = None

        if self._config.remove_images:
            if _is_image_available(self._config.image, self._config.container_runtime):
                self.logger.info(f"Removing image {self._config.image}")
                try:
                    _remove_image(self._config.image, self._config.container_runtime)
                except subprocess.CalledProcessError:
                    self.logger.error(f"Failed to remove image {self._config.image}", exc_info=True)

    @property
    def runtime(self) -> RemoteRuntime:

        if self._runtime is None:
            raise DeploymentNotStartedError()
        return self._runtime