Repository: ErikBjare/gptme
File: gptme/tools/shell.py (Python)

Code before:
        except BrokenPipeError:
            # process has died
            if tries == 0:
                # log warning and restart, once
                logger.warning("Warning: shell process died, restarting")
                self.restart()
                return self._run(command, output=output, tries=tries + 1)
            else:
                raise

        self.process.stdin.flush()

        stdout = []
        stderr = []
        return_code = None

        while True:
            rlist, _, _ = select.select([self.stdout_fd, self.stderr_fd], [], [])
            if not rlist: continue
            for fd in rlist:
                assert fd in [self.stdout_fd, self.stderr_fd]
                # We use a higher value, because there is a bug which leads to spaces at the boundary
                # 2**12 = 4096
                # 2**16 = 65536
                data = os.read(fd, 2**16).decode("utf-8")
                lines = data.splitlines(True)
                for line in lines:
                    if f"ReturnCode:" in line and self.delimiter in line:
                        return_code = int(line.split("ReturnCode:")[1].split()[0])
                        # if command is cd and successful, we need to change the directory
                        if command.startswith("cd ") and return_code == 0:
                            ex, pwd, _ = self._run("pwd", output=False)
                            assert ex == 0
                            os.chdir(pwd.strip())
                        return (
                            return_code,
                            "".join(stdout).strip(),
                            "".join(stderr).strip(),
                        )
                    if fd == self.stdout_fd:
                        stdout.append(line)
                        if output:
                            print(line, end="", file=sys.stdout)
                    elif fd == self.stderr_fd:
                        stderr.append(line)
                        if output:
                            print(line, end="", file=sys.stderr)

    def close(self):
        assert self.process.stdin
        self.process.stdin.close()

Review comment (bug):
Why no longer use this regex? I recently added it in https://github.com/ErikBjare/gptme/commit/df5c9d6eca2d3d084a1060f97290ad3f629930c0 and I think it fixed a bug. Not sure if it currently supports negative returncodes though.

Code after:
            self.process.stdin.write(full_command)
        except BrokenPipeError:
            # process has died
            if tries == 0:
                # log warning and restart, once
                logger.warning("Warning: shell process died, restarting")
                self.restart()
                return self._run(command, output=output, tries=tries + 1)
            else:
                raise

        self.process.stdin.flush()

        stdout: list[str] = []
        stderr: list[str] = []
        return_code: int | None = None

        while True:
            rlist, _, _ = select.select([self.stdout_fd, self.stderr_fd], [], [])
            for fd in rlist:
                assert fd in [self.stdout_fd, self.stderr_fd]
                # We use a higher value, because there is a bug which leads to spaces at the boundary
                # 2**12 = 4096
                # 2**16 = 65536
                data = os.read(fd, 2**16).decode("utf-8")
                lines = data.splitlines(keepends=True)
                re_returncode = re.compile(r"ReturnCode:(\d+)")
                for line in lines:
                    if "ReturnCode:" in line and self.delimiter in line:
                        if match := re_returncode.search(line):
                            return_code = int(match.group(1))
                        # if command is cd and successful, we need to change the directory
                        if command.startswith("cd ") and return_code == 0:
                            ex, pwd, _ = self._run("pwd", output=False)
                            assert ex == 0
                            os.chdir(pwd.strip())
                        return (
                            return_code,
                            "".join(stdout).strip(),
                            "".join(stderr).strip(),
                        )
                    if fd == self.stdout_fd:
                        stdout.append(line)
                        if output:
                            print(line, end="", file=sys.stdout)
                    elif fd == self.stderr_fd:
                        stderr.append(line)
                        if output:
                            print(line, end="", file=sys.stderr)

    def close(self):