Skip to content

Instantly share code, notes, and snippets.

@amxv
Created July 5, 2025 22:10
Show Gist options
  • Select an option

  • Save amxv/54f822e92c2cd8501d99f08bb7ae594c to your computer and use it in GitHub Desktop.

Select an option

Save amxv/54f822e92c2cd8501d99f08bb7ae594c to your computer and use it in GitHub Desktop.
clw-v4.py
#!/usr/bin/env python3
"""
CLW – Chunk-Line-Wrapper (v4-pnpm-fix)
Break every stdout/stderr line – whether delimited by LF or CR – into
≤1550-byte slices so Codex Cloud never exceeds the 1 600-byte PTY limit.
"""
import binascii, os, re, sys, signal
signal.signal(signal.SIGPIPE, signal.SIG_DFL)
DEFAULT_MAX_LINE_LENGTH = 1550
DEFAULT_WRAP_MARK = "⏎"
_rx_word_wrap = re.compile(rb"^.*(\b).+", re.DOTALL)
def _chunks(buf: bytes, size: int):
start = 0
while True:
end = start + size
chunk = buf[start:end]
if len(chunk) < size:
yield False, chunk
return
m = _rx_word_wrap.match(chunk)
if m and m.start(1):
chunk = chunk[:m.start(1)]
yield True, chunk
start += len(chunk)
def _iter_lines(stream):
"""Yield lines split on LF **or CR**, keeping delimiters."""
for raw in stream:
for part in raw.split(b"\r"):
if part:
yield part
# preserve carriage-return semantics
if raw.endswith(b"\r"):
yield b"\r"
def main():
max_len = int(os.getenv("CLW_MAX_LINE_LENGTH", DEFAULT_MAX_LINE_LENGTH))
wrap_mark = (
binascii.unhexlify(os.getenv("CLW_WRAP_MARK"))
if os.getenv("CLW_WRAP_MARK")
else DEFAULT_WRAP_MARK.encode()
) + b"\n"
chunk_size = max_len - len(wrap_mark)
for line in _iter_lines(sys.stdin.buffer):
for wrapped, chunk in _chunks(line, chunk_size):
sys.stdout.buffer.write(chunk)
if wrapped:
sys.stdout.buffer.write(wrap_mark)
sys.stdout.buffer.flush()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment