atlas-iac/services/hermes/scripts/patch_ttyd_index.py

#!/usr/bin/env python3
"""Add reliable clipboard and reconnect handling to ttyd's pinned client."""

from __future__ import annotations

import argparse
import socket
import subprocess
import time
import urllib.request
from pathlib import Path


MARKER = "atlas-ttyd-clipboard"
UPSTREAM_COPY = 'document.execCommand("copy")'
CLIPBOARD_ADAPTER = r"""
<style>
  #atlas-clipboard-status {
    background: #181825;
    border: 1px solid #f9e2af;
    border-radius: 6px;
    color: #f9e2af;
    font: 13px/1.4 monospace;
    padding: 6px 10px;
    position: fixed;
    right: 12px;
    top: 12px;
    z-index: 10000;
  }
  #atlas-terminal-recovery {
    background: #181825;
    border: 1px solid #89b4fa;
    border-radius: 8px;
    color: #cdd6f4;
    font: 16px/1.4 monospace;
    left: 50%;
    padding: 12px 18px;
    position: fixed;
    top: 50%;
    transform: translate(-50%, -50%);
    z-index: 10001;
  }
</style>
<script id="atlas-ttyd-clipboard">
(() => {
  let unloading = false;
  let recoveryStarted = false;

  const showStatus = (message, failed = false) => {
    let status = document.getElementById('atlas-clipboard-status');
    if (!status) {
      status = document.createElement('div');
      status.id = 'atlas-clipboard-status';
      document.body.appendChild(status);
    }
    status.textContent = message;
    status.style.borderColor = failed ? '#f38ba8' : '#a6e3a1';
    status.style.color = failed ? '#f38ba8' : '#a6e3a1';
    clearTimeout(status.removeTimer);
    status.removeTimer = setTimeout(() => status.remove(), 1200);
  };

  const sleep = (milliseconds) => new Promise(
    (resolve) => window.setTimeout(resolve, milliseconds),
  );

  const showRecoveryStatus = (message) => {
    let status = document.getElementById('atlas-terminal-recovery');
    if (!status) {
      status = document.createElement('div');
      status.id = 'atlas-terminal-recovery';
      document.body.appendChild(status);
    }
    status.textContent = message;
  };

  const tokenUrl = new URL('token', window.location.href).toString();
  const waitForFreshPage = async () => {
    if (recoveryStarted || unloading) return;
    recoveryStarted = true;
    showRecoveryStatus('Reconnecting terminal automatically...');

    let failures = 0;
    while (!unloading) {
      try {
        const response = await fetch(tokenUrl, {
          cache: 'no-store',
          credentials: 'same-origin',
          redirect: 'manual',
        });
        if (response.type === 'opaqueredirect'
            || (response.status >= 300 && response.status < 400)) {
          window.location.reload();
          return;
        }
        if (response.ok) {
          const payload = await response.json();
          if (typeof payload.token === 'string') {
            window.location.reload();
            return;
          }
        }
      } catch (error) {
        console.debug('[atlas] terminal endpoint is not ready', error);
      }

      failures += 1;
      if (failures === 5) {
        showRecoveryStatus('Waiting for terminal service...');
      }
      await sleep(Math.min(500 + failures * 250, 3000));
    }
  };

  const NativeWebSocket = window.WebSocket;
  window.WebSocket = class AtlasRecoveringWebSocket extends NativeWebSocket {
    constructor(...args) {
      super(...args);
      this.addEventListener('close', () => {
        if (!unloading) void waitForFreshPage();
      }, { once: true });
    }
  };

  window.addEventListener('beforeunload', () => {
    unloading = true;
  });

  window.addEventListener('keydown', (event) => {
    if (!recoveryStarted || event.key !== 'Enter') return;
    event.preventDefault();
    event.stopImmediatePropagation();
  }, true);

  const fallbackCopy = (text) => {
    const input = document.createElement('textarea');
    input.value = text;
    input.setAttribute('readonly', '');
    input.style.position = 'fixed';
    input.style.opacity = '0';
    document.body.appendChild(input);
    input.select();
    const copied = document.execCommand('copy');
    input.remove();
    if (!copied) throw new Error('browser rejected clipboard write');
  };

  const install = () => {
    const term = window.term;
    if (!term) {
      setTimeout(install, 50);
      return;
    }
    if (term.__atlasClipboardInstalled) return;
    term.__atlasClipboardInstalled = true;

    const copySelection = async () => {
      const text = term.getSelection();
      if (!text) return false;
      try {
        if (navigator.clipboard && window.isSecureContext) {
          await navigator.clipboard.writeText(text);
        } else {
          fallbackCopy(text);
        }
        showStatus('Copied selection');
        return true;
      } catch (error) {
        try {
          fallbackCopy(text);
          showStatus('Copied selection');
          return true;
        } catch (fallbackError) {
          console.error('[atlas] clipboard copy failed', error, fallbackError);
          showStatus('Clipboard copy failed', true);
          return false;
        }
      }
    };

    window.addEventListener('mouseup', () => {
      if (term.hasSelection()) void copySelection();
    });
    window.addEventListener('keydown', (event) => {
      const copyKey = event.key.toLowerCase() === 'c'
        && (event.ctrlKey || event.metaKey)
        && term.hasSelection();
      if (!copyKey) return;
      event.preventDefault();
      event.stopImmediatePropagation();
      void copySelection();
    }, true);
  };

  install();
})();
</script>
"""


def patch_html(content: str) -> str:
    """Insert the adapter before ttyd and disable its false-success copy call."""
    if content.count(UPSTREAM_COPY) != 1 or content.count("<body>") != 1:
        raise RuntimeError("ttyd index patch context changed")
    if MARKER in content:
        raise RuntimeError("ttyd index is already patched")
    content = content.replace(UPSTREAM_COPY, "void 0", 1)
    return content.replace("<body>", f"<body>{CLIPBOARD_ADAPTER}", 1)


def fetch_embedded_index(ttyd: Path) -> str:
    """Serve and retrieve ttyd's embedded page without vendoring its bundle."""
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        port = listener.getsockname()[1]
    process = subprocess.Popen(
        [
            str(ttyd),
            "--interface",
            "127.0.0.1",
            "--port",
            str(port),
            "/bin/sh",
            "-c",
            "sleep 30",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        for _ in range(100):
            try:
                with urllib.request.urlopen(
                    f"http://127.0.0.1:{port}/", timeout=1
                ) as response:
                    return response.read().decode("utf-8")
            except OSError:
                if process.poll() is not None:
                    raise RuntimeError("temporary ttyd exited before serving its index")
                time.sleep(0.05)
        raise RuntimeError("timed out retrieving ttyd's embedded index")
    finally:
        process.terminate()
        try:
            process.wait(timeout=3)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=3)


def main() -> int:
    """Generate one patched index for the ttyd sidecar."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("ttyd", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    content = patch_html(fetch_embedded_index(args.ttyd))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(content, encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())