Skip to content

Instantly share code, notes, and snippets.

@amxv
Created June 26, 2025 03:32
Show Gist options
  • Select an option

  • Save amxv/286d176187b47a2ac95f3c0e96ad18bf to your computer and use it in GitHub Desktop.

Select an option

Save amxv/286d176187b47a2ac95f3c0e96ad18bf to your computer and use it in GitHub Desktop.
clw-v3.py
#!/usr/bin/env python3
"""
CLW – Chunk‑Line‑Wrapper
Breaks every stdout/stderr line into ≤1550‑byte slices so Codex Cloud never
hits the 1 600‑byte PTY limit.
Changes vs upstream v1
----------------------
* Handle SIGPIPE quietly → no BrokenPipeError stack‑trace.
"""
import binascii
import os
import re
import sys
import signal # <<< NEW
signal.signal(signal.SIGPIPE, signal.SIG_DFL) # <<< NEW
DEFAULT_MAX_LINE_LENGTH = 1550
DEFAULT_WRAP_MARK = "⏎"
def split_into_chunks(s, chunk_size):
rx_word_wrap = re.compile(rb"^.*(\b).+", re.DOTALL)
start = 0
while True:
end = start + chunk_size
chunk = s[start:end]
if len(chunk) < 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 main():
max_line_length = 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("utf‑8")) + b"\n"
chunk_size = max_line_length - len(wrap_mark)
assert chunk_size > 0
for line in sys.stdin.buffer:
for wrapped, chunk in split_into_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