fix: large Kitty direct image transfers over SSH.

[Xyhlon]

First of all this fixes #95 #294 . When running over SSH, the Kitty backend uses direct transfer mode. Large images are split into many base64 chunks and written to the terminal stream. Without passing editor_tty, those chunks go through Neovim's stdout path, where Neovim redraw/control bytes can interleave with the Kitty graphics payload. This corrupts the base64 stream.

I confirmed this with kitty --dump-bytes: failed large transfers contained Neovim escape sequences inside the image payload. Passing editor_tty makes the existing tty write path handle direct transfers and prevents that corruption.

This also increases the direct transfer chunk size from 4096 to 65536 bytes. In my test case, this reduced a large image transfer from 1223 Kitty graphics commands to 77 while preserving a valid decoded payload.

To create the dump:

kitty --dump-bytes /tmp/kitty.bytes --dump-commands sh -lc 'ssh faepmac1'

then create a large image:

magick -size 1600x1000 plasma:fractal /tmp/image-large.png

then open a large image with Neovim

To analyze the dump:

#!/usr/bin/env python3
import base64
import hashlib
import re
import sys
from pathlib import Path

dump_path = Path(sys.argv[1])
candidate_root = Path(sys.argv[2]) if len(sys.argv) > 2 else None

b = dump_path.read_bytes()

seqs = re.findall(rb"\x1b_G(.*?)\x1b\\", b, flags=re.S)

print(f"dump bytes: {len(b)}")
print(f"kitty graphics sequences: {len(seqs)}")

unescaped_g = 0
for m in re.finditer(rb"_G", b):
    i = m.start()
    if i == 0 or b[i - 1] != 0x1B:
        unescaped_g += 1

print(f"unescaped literal _G occurrences: {unescaped_g}")

transfers = []
current = None

for seq in seqs:
    if b";" not in seq:
        continue

    header, payload = seq.split(b";", 1)
    if not payload:
        continue

    # Parse simple k=v header fields.
    params = {}
    for part in header.split(b","):
        if b"=" in part:
            k, v = part.split(b"=", 1)
            params[k] = v

    # First chunk has a full header; continuation chunks are usually just m=1/m=0.
    is_continuation = set(params.keys()).issubset({b"m"})

    if current is None or not is_continuation:
        current = {
            "first_header": header,
            "chunks": [],
        }

    current["chunks"].append(payload)

    if params.get(b"m") == b"0":
        transfers.append(current)
        current = None

if current is not None:
    transfers.append(current)

print(f"data transfers found: {len(transfers)}")

candidate_hashes = {}
if candidate_root and candidate_root.exists():
    for path in candidate_root.rglob("*.png"):
        try:
            data = path.read_bytes()
        except Exception:
            continue
        candidate_hashes[hashlib.sha256(data).hexdigest()] = path

for idx, tr in enumerate(transfers, 1):
    joined = b"".join(tr["chunks"])
    padded = joined + b"=" * ((4 - len(joined) % 4) % 4)

    print()
    print(f"transfer #{idx}")
    print(f"  first header: {tr['first_header'][:200].decode('ascii', 'replace')}")
    print(f"  chunks: {len(tr['chunks'])}")
    print(f"  b64 bytes: {len(joined)}")

    try:
        raw = base64.b64decode(padded, validate=True)
    except Exception as e:
        print(f"  base64 decode: FAILED: {e}")
        continue

    sha = hashlib.sha256(raw).hexdigest()
    print(f"  decoded bytes: {len(raw)}")
    print(f"  sha256: {sha}")

    out = Path(f"/tmp/reconstructed-kitty-transfer-{idx}.png")
    out.write_bytes(raw)
    print(f"  wrote: {out}")

    if sha in candidate_hashes:
        print(f"  matches candidate: {candidate_hashes[sha]}")
    else:
        print("  matches candidate: no")

Test with ssh and ssh+tmux and just tmux.

[3rd]

Hi @Xyhlon, thank you for the PR!
Surprised the 65536 bump worked, Kitty docs say we shouldn't send more than 4096 🤷
https://sw.kovidgoyal.net/kitty/graphics-protocol/#:~:text=chunks%20no%20larger%20than%204096%20bytes

[Xyhlon]

Yeah works fine on my machine, but i understand that if they only guarantee this from the official channel you might want to keep it as a default. My suggestion would be an over-writable option so that the users downstream change it if they have performance issues and they get it to work. If that's fine by you i will add this option to the main plugin?

[3rd]

Looks great, thank you!