Temporary macOS arm64 VS Code Insiders terminal workaround for microsoft/vscode#313694

vscode-insiders-terminal-313694-fix.py

from pathlib import Path
import platform
import shutil
import sys

INDEX_REPLACEMENT = """var terminalCtor;
if (process.platform === 'darwin' && process.arch === 'arm64') {
    terminalCtor = require('./vscodeDarwinPtyTerminal').DarwinPtyTerminal;
}
else if (process.platform === 'win32') {
    terminalCtor = require('./windowsTerminal').WindowsTerminal;
}
else {
    terminalCtor = require('./unixTerminal').UnixTerminal;
}
"""

INDEX_PATTERNS = [
    """var terminalCtor;
if (process.platform === 'win32') {
    terminalCtor = require('./windowsTerminal').WindowsTerminal;
}
else {
    terminalCtor = require('./unixTerminal').UnixTerminal;
}
""",
    """var terminalCtor = process.platform === 'win32' ? require('./windowsTerminal').WindowsTerminal : require('./unixTerminal').UnixTerminal;
"""
]

DARWIN_TERMINAL = r'''"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.DarwinPtyTerminal = void 0;
var child_process = require("child_process");
var path = require("path");
var DEFAULT_FILE = "sh";
var DEFAULT_NAME = "xterm";
var DEFAULT_COLS = 80;
var DEFAULT_ROWS = 24;
var DarwinPtyTerminal = function () {
    function DarwinPtyTerminal(file, args, opt) {
        this._dataListeners = new Set();
        this._exitListeners = new Set();
        this._pid = 0;
        this._cols = DEFAULT_COLS;
        this._rows = DEFAULT_ROWS;
        this._exited = false;
        this._processName = DEFAULT_FILE;

        if (typeof args === "string")
            throw new Error("args as a string is not supported on unix.");

        if (Array.isArray(args))
            this._args = args;
        else
            this._args = [];

        if (file)
            this._file = file;
        else
            this._file = DEFAULT_FILE;

        if (opt === undefined)
            opt = {};

        if (opt.cols !== undefined && opt.cols > 0)
            this._cols = opt.cols;

        if (opt.rows !== undefined && opt.rows > 0)
            this._rows = opt.rows;

        if (opt.cwd !== undefined)
            this._cwd = opt.cwd;
        else
            this._cwd = process.cwd();

        this._env = this._copyEnv(opt.env);
        this._env.PWD = this._cwd;

        if (opt.name !== undefined)
            this._env.TERM = opt.name;
        else if (this._env.TERM === undefined)
            this._env.TERM = DEFAULT_NAME;

        this._processName = path.basename(this._file);
        this._startPythonHelper();
    }
    Object.defineProperty(DarwinPtyTerminal.prototype, "pid", {
        get: function () { return this._pid; },
        enumerable: false,
        configurable: true
    });
    Object.defineProperty(DarwinPtyTerminal.prototype, "cols", {
        get: function () { return this._cols; },
        enumerable: false,
        configurable: true
    });
    Object.defineProperty(DarwinPtyTerminal.prototype, "rows", {
        get: function () { return this._rows; },
        enumerable: false,
        configurable: true
    });
    Object.defineProperty(DarwinPtyTerminal.prototype, "process", {
        get: function () { return this._processName; },
        enumerable: false,
        configurable: true
    });
    Object.defineProperty(DarwinPtyTerminal.prototype, "master", {
        get: function () { return undefined; },
        enumerable: false,
        configurable: true
    });
    Object.defineProperty(DarwinPtyTerminal.prototype, "slave", {
        get: function () { return undefined; },
        enumerable: false,
        configurable: true
    });
    DarwinPtyTerminal.prototype.onData = function (listener) {
        return this._addListener(this._dataListeners, listener);
    };
    DarwinPtyTerminal.prototype.onExit = function (listener) {
        return this._addListener(this._exitListeners, listener);
    };
    DarwinPtyTerminal.prototype.write = function (data) {
        if (this._exited)
            return;

        if (!this._child)
            return;

        if (!this._child.stdin)
            return;

        if (this._child.stdin.destroyed)
            return;

        this._child.stdin.write(data);
    };
    DarwinPtyTerminal.prototype.end = function (data) {
        if (data !== undefined)
            this.write(data);

        if (this._child && this._child.stdin)
            this._child.stdin.end();
    };
    DarwinPtyTerminal.prototype.pause = function () {
        if (this._child && this._child.stdout)
            return this._child.stdout.pause();
    };
    DarwinPtyTerminal.prototype.resume = function () {
        if (this._child && this._child.stdout)
            return this._child.stdout.resume();
    };
    DarwinPtyTerminal.prototype.resize = function (cols, rows) {
        if (cols <= 0 || rows <= 0 || isNaN(cols) || isNaN(rows) || cols === Infinity || rows === Infinity)
            throw new Error("resizing must be done using positive cols and rows");

        this._cols = cols;
        this._rows = rows;

        if (!this._control)
            return;

        if (this._control.destroyed)
            return;

        this._control.write(JSON.stringify({ cols: cols, rows: rows }) + "\n");
    };
    DarwinPtyTerminal.prototype.clear = function () {
    };
    DarwinPtyTerminal.prototype.destroy = function () {
        this.kill("SIGHUP");
    };
    DarwinPtyTerminal.prototype.kill = function (signal) {
        var actualSignal = signal;

        if (actualSignal === undefined)
            actualSignal = "SIGHUP";

        if (this._child)
            this._child.kill(actualSignal);
    };
    DarwinPtyTerminal.prototype.pipe = function (dest, options) {
        if (!this._child)
            throw new Error("Cannot pipe before terminal helper starts");

        if (!this._child.stdout)
            throw new Error("Cannot pipe terminal without stdout");

        return this._child.stdout.pipe(dest, options);
    };
    DarwinPtyTerminal.prototype.setEncoding = function () {
    };
    DarwinPtyTerminal.prototype._copyEnv = function (env) {
        var source = env;
        var copy = {};

        if (source === undefined)
            source = process.env;

        Object.keys(source).forEach(function (key) {
            var value = source[key];

            if (value !== undefined)
                copy[key] = String(value);
        });

        delete copy.TMUX;
        delete copy.TMUX_PANE;
        delete copy.STY;
        delete copy.WINDOW;
        delete copy.WINDOWID;
        delete copy.TERMCAP;
        delete copy.COLUMNS;
        delete copy.LINES;

        return copy;
    };
    DarwinPtyTerminal.prototype._startPythonHelper = function () {
        var python = process.env.VSCODE_DARWIN_PTY_PYTHON;

        if (!python)
            python = "/usr/bin/python3";

        var helper = path.join(__dirname, "vscodeDarwinPtyHelper.py");
        var child = child_process.spawn(python, [helper], {
            cwd: this._cwd,
            env: process.env,
            stdio: ["pipe", "pipe", "pipe", "pipe"]
        });
        var didSpawn = false;
        var self = this;

        child.once("spawn", function () {
            didSpawn = true;
            self._control = child.stdio[3];
            self._writeConfig();
        });
        child.once("error", function (error) {
            if (didSpawn)
                return;

            self._emitData("\r\nVS Code terminal patch failed: " + error.message + "\r\n");
            self._emitExit(1, undefined);
        });
        this._bindChild(child);
    };
    DarwinPtyTerminal.prototype._writeConfig = function () {
        if (!this._control)
            throw new Error("Terminal helper control pipe missing");

        this._control.write(JSON.stringify({
            file: this._file,
            args: this._args,
            cwd: this._cwd,
            env: this._env,
            cols: this._cols,
            rows: this._rows
        }) + "\n");
    };
    DarwinPtyTerminal.prototype._bindChild = function (child) {
        var self = this;
        this._child = child;

        if (child.pid)
            this._pid = child.pid;

        if (child.stdout)
            child.stdout.on("data", function (data) { self._emitData(data.toString("utf8")); });

        if (child.stderr)
            child.stderr.on("data", function (data) { self._emitData(data.toString("utf8")); });

        child.on("exit", function (code, signal) {
            if (self._child !== child)
                return;

            self._emitExit(code, signal);
        });
    };
    DarwinPtyTerminal.prototype._emitData = function (data) {
        this._dataListeners.forEach(function (listener) { return listener(data); });
    };
    DarwinPtyTerminal.prototype._emitExit = function (code, signal) {
        var exitCode = code;

        if (exitCode === null || exitCode === undefined)
            exitCode = 0;

        if (this._exited)
            return;

        this._exited = true;
        this._exitListeners.forEach(function (listener) { return listener({ exitCode: exitCode, signal: signal }); });
    };
    DarwinPtyTerminal.prototype._addListener = function (listeners, listener) {
        var disposed = false;
        listeners.add(listener);

        return {
            dispose: function () {
                if (disposed)
                    return;

                disposed = true;
                listeners.delete(listener);
            }
        };
    };
    return DarwinPtyTerminal;
}();
exports.DarwinPtyTerminal = DarwinPtyTerminal;
'''

PYTHON_HELPER = r'''import fcntl
import json
import os
import pty
import selectors
import signal
import struct
import sys
import termios
import traceback


def read_line(fd):
    chunks = []

    while True:
        data = os.read(fd, 1)

        if not data:
            return b"".join(chunks)

        if data == b"\n":
            return b"".join(chunks)

        chunks.append(data)


def set_size(fd, cols, rows):
    if cols <= 0 or rows <= 0:
        return

    size = struct.pack("HHHH", rows, cols, 0, 0)

    try:
        fcntl.ioctl(fd, termios.TIOCSWINSZ, size)
    except OSError:
        return


def exit_from_status(status):
    if os.WIFEXITED(status):
        return os.WEXITSTATUS(status)

    if os.WIFSIGNALED(status):
        return 128 + os.WTERMSIG(status)

    return 0


def forward_signal(pid, signum):
    try:
        os.kill(pid, signum)
    except ProcessLookupError:
        pass


def main():
    control_fd = 3
    raw_config = read_line(control_fd)

    if not raw_config:
        raise RuntimeError("Missing terminal helper config")

    config = json.loads(raw_config.decode("utf-8"))
    file = config["file"]
    args = config["args"]
    cwd = config["cwd"]
    env = config["env"]
    cols = int(config["cols"])
    rows = int(config["rows"])
    pid, master_fd = pty.fork()

    if pid == 0:
        try:
            os.close(control_fd)
        except OSError:
            pass

        os.chdir(cwd)
        os.execvpe(file, [file] + args, env)

    def handle_signal(signum, frame):
        forward_signal(pid, signum)
        raise SystemExit(0)

    signal.signal(signal.SIGHUP, handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)
    set_size(master_fd, cols, rows)
    selector = selectors.DefaultSelector()
    selector.register(sys.stdin.buffer, selectors.EVENT_READ, "stdin")
    selector.register(master_fd, selectors.EVENT_READ, "pty")
    selector.register(control_fd, selectors.EVENT_READ, "control")
    control_buffer = b""

    while True:
        waited_pid, status = os.waitpid(pid, os.WNOHANG)

        if waited_pid == pid:
            return exit_from_status(status)

        for key, event in selector.select(0.1):
            if key.data == "stdin":
                try:
                    data = os.read(sys.stdin.fileno(), 65536)
                except OSError:
                    selector.unregister(sys.stdin.buffer)
                    continue

                if data:
                    try:
                        os.write(master_fd, data)
                    except OSError:
                        continue
                else:
                    selector.unregister(sys.stdin.buffer)
            elif key.data == "pty":
                try:
                    data = os.read(master_fd, 65536)
                except OSError:
                    data = b""

                if not data:
                    waited_pid, status = os.waitpid(pid, 0)
                    return exit_from_status(status)

                try:
                    os.write(sys.stdout.fileno(), data)
                except OSError:
                    return 0
            elif key.data == "control":
                try:
                    chunk = os.read(control_fd, 4096)
                except OSError:
                    selector.unregister(control_fd)
                    continue

                if not chunk:
                    selector.unregister(control_fd)
                    continue

                control_buffer += chunk

                while b"\n" in control_buffer:
                    line, control_buffer = control_buffer.split(b"\n", 1)

                    if not line:
                        continue

                    message = json.loads(line.decode("utf-8"))

                    if "cols" in message and "rows" in message:
                        set_size(master_fd, int(message["cols"]), int(message["rows"]))


def run():
    try:
        code = main()
    except SystemExit as error:
        code = error.code
    except BaseException:
        traceback.print_exc()
        code = 1

    if isinstance(code, int):
        raise SystemExit(code)

    raise SystemExit(0)


run()
'''


def app_path():
    if len(sys.argv) > 1:
        return Path(sys.argv[1]).expanduser()

    return Path('/Applications/Visual Studio Code - Insiders.app')


def backup(path):
    backup_path = path.with_name(path.name + '.bak-vscode-313694')

    if backup_path.exists():
        return backup_path

    shutil.copy2(path, backup_path)
    return backup_path


def patch_index(index_path):
    text = index_path.read_text()

    if "vscodeDarwinPtyTerminal" in text:
        return 'index-already-patched'

    for pattern in INDEX_PATTERNS:
        if pattern in text:
            backup(index_path)
            index_path.write_text(text.replace(pattern, INDEX_REPLACEMENT, 1))
            return 'index-patched'

    raise RuntimeError('index-pattern-not-found')


def write_file(path, content):
    if path.exists():
        existing = path.read_text()

        if existing == content:
            return path.name + '-already-current'

        backup(path)

    path.write_text(content)
    return path.name + '-written'


def main():
    if sys.platform != 'darwin':
        raise RuntimeError('this-patch-is-for-macos-only')

    if platform.machine() != 'arm64':
        raise RuntimeError('this-patch-is-for-macos-arm64-only')

    app = app_path()
    root = app / 'Contents/Resources/app/node_modules/node-pty/lib'
    index_path = root / 'index.js'
    terminal_path = root / 'vscodeDarwinPtyTerminal.js'
    helper_path = root / 'vscodeDarwinPtyHelper.py'

    if not index_path.exists():
        raise RuntimeError('node-pty-index-missing: ' + str(index_path))

    root.mkdir(parents=True, exist_ok=True)
    results = [
        patch_index(index_path),
        write_file(terminal_path, DARWIN_TERMINAL),
        write_file(helper_path, PYTHON_HELPER)
    ]

    print('\n'.join(results))
    print('restart VS Code Insiders completely')


main()