#!/usr/bin/env python3
"""etch — drive a persistent EternalTerminal session programmatically.
Wraps `et` in a PTY (via a small built-in engine), neutralizes the prompt to get a
clean byte stream, and runs commands framed by unique sentinels so output and
exit codes extract reliably regardless of the remote prompt's rendering.
Library: EtSession (one-shot) / open_session(key=...) (warm daemon).
CLI: `etch <run|open|attach|send|sendline|expect|read|peep|ls|close|gc> KEY ...` (see --help).
"""
from __future__ import annotations
import base64
import codecs
import fcntl
import getpass
import glob
import json
import os
import pty
import re
import select
import shlex
import signal
import socket
import struct
import subprocess
import sys
import termios
import time
import tty
import uuid
from typing import Any, Callable, Union, cast
# The remote PTY runs in canonical mode, which truncates a single input line at
# the kernel's cap (~1KB). A framed run() line longer than this is silently cut,
# the end sentinel never matches, and the leftover poisons the session. run()
# reroutes anything longer than this through run_script()'s chunked path.
_LINE_CAP = 900
# ─────────────────────────────────────────────────────────────────────────────
# PTY/expect engine
# -----------------
# A small, self-contained pseudo-terminal driver: spawn a command under a PTY and
# drive it with an expect-style read buffer. This is all etch needs to make `et`
# believe a human is on the other end, with no third-party dependencies.
#
# spawn(cmd, encoding=, timeout=, dimensions=, codec_errors=) -> child
# child.expect(pattern | list, timeout=) child.expect_exact(s, timeout=)
# child.read_nonblocking(size=, timeout=) child.send(s) child.sendline(s)
# child.before child.after child.match child.close(force=)
# TIMEOUT / EOF (raised, and usable as expect() patterns to match instead of raise)
# ─────────────────────────────────────────────────────────────────────────────
class TIMEOUT(Exception):
"""No match or data arrived before the deadline. Usable as an expect() pattern
to match-on-timeout instead of raising."""
class EOF(Exception):
"""The child's PTY hit end-of-file / the process exited. Usable as an expect()
pattern to match-on-EOF instead of raising."""
_Pattern = Union[str, "re.Pattern[str]", type[TIMEOUT], type[EOF]]
_ExpectArg = Union[_Pattern, list[_Pattern]]
class _PtyProcess:
"""Spawn a command under a pseudo-terminal and drive it with an expect-style
read buffer. A minimal driver: only the behavior etch needs."""
# A brief pause before each write avoids losing input to a remote PTY that is
# not yet ready to read. Tunable if latency ever matters.
delaybeforesend: float = 0.05
def __init__(
self,
command: str,
encoding: str = "utf-8",
timeout: float = 30,
dimensions: tuple[int, int] = (24, 80),
codec_errors: str = "strict",
) -> None:
self.timeout = timeout
self.before: str = ""
self.after: str | type[TIMEOUT] | type[EOF] = ""
self.match: re.Match[str] | None = None
self._buf: str = ""
self._decoder = codecs.getincrementaldecoder(encoding)(codec_errors)
argv = shlex.split(command)
pid, fd = pty.fork()
if pid == 0: # child: replace ourselves with the command on the PTY slave
try:
os.execvp(argv[0], argv)
except Exception:
os._exit(127)
self.pid: int = pid
self.fd: int = fd
rows, cols = dimensions
try:
fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
except OSError:
pass
# ── low-level I/O ─────────────────────────────────────────────────────────
def _read_fd(self, timeout: float) -> str | None:
"""Wait up to `timeout` for output; decode and return a chunk (possibly ''
on a partial multibyte char). Return None on timeout; raise EOF when gone."""
try:
empty: list[int] = []
ready, _, _ = select.select([self.fd], empty, empty, max(0.0, timeout))
except (OSError, ValueError):
raise EOF()
if not ready:
return None
try:
data = os.read(self.fd, 4096)
except OSError: # Linux reports the PTY master EOF as EIO
raise EOF()
if not data: # macOS reports EOF as an empty read
raise EOF()
return self._decoder.decode(data)
def send(self, s: str) -> int:
if self.delaybeforesend:
time.sleep(self.delaybeforesend)
return os.write(self.fd, s.encode("utf-8", "replace"))
def sendline(self, s: str = "") -> int:
return self.send(s + "\n")
def read_nonblocking(self, size: int = 1, timeout: float | None = None) -> str:
"""Up to `size` chars: drain the expect buffer first, else read the fd.
Raise TIMEOUT if nothing arrives in time, EOF at end of file."""
if self._buf:
out, self._buf = self._buf[:size], self._buf[size:]
return out
chunk = self._read_fd(self.timeout if timeout is None else timeout)
if chunk is None:
raise TIMEOUT()
if size and len(chunk) > size:
self._buf = chunk[size:] + self._buf
chunk = chunk[:size]
return chunk
# ── expect machinery ──────────────────────────────────────────────────────
@staticmethod
def _compile(pattern: _ExpectArg) -> list[object]:
items = pattern if isinstance(pattern, list) else [pattern]
out: list[object] = []
for p in items:
out.append(re.compile(p) if isinstance(p, str) else p)
return out
def _set(
self,
before: str,
after: str | type[TIMEOUT] | type[EOF],
match: re.Match[str] | None,
rest: str,
) -> None:
self.before, self.after, self.match, self._buf = before, after, match, rest
def expect(self, pattern: _ExpectArg, timeout: float | None = None) -> int:
"""Read until one of `pattern` (a regex, or a list mixing regexes with the
EOF/TIMEOUT classes) matches; return its index and set before/after/match.
EOF/TIMEOUT in the list match instead of raising."""
compiled = self._compile(pattern)
eof_idx: int | None = None
tmo_idx: int | None = None
regexes: list[tuple[int, re.Pattern[str]]] = []
for i, p in enumerate(compiled):
if p is EOF:
eof_idx = i
elif p is TIMEOUT:
tmo_idx = i
elif isinstance(p, re.Pattern):
regexes.append((i, cast("re.Pattern[str]", p)))
deadline = time.time() + (self.timeout if timeout is None else timeout)
while True:
best: tuple[int, int, re.Match[str]] | None = None
for i, rx in regexes:
m = rx.search(self._buf)
if m is not None and (best is None or m.start() < best[0]):
best = (m.start(), i, m)
if best is not None:
_, i, m = best
self._set(self._buf[: m.start()], m.group(0), m, self._buf[m.end() :])
return i
remaining = deadline - time.time()
if remaining <= 0:
if tmo_idx is not None:
self._set(self._buf, TIMEOUT, None, "")
return tmo_idx
raise TIMEOUT()
try:
chunk = self._read_fd(remaining)
except EOF:
if eof_idx is not None:
self._set(self._buf, EOF, None, "")
return eof_idx
raise
if chunk:
self._buf += chunk
def expect_exact(self, s: str, timeout: float | None = None) -> int:
"""Like expect() but matches a literal substring (no regex)."""
deadline = time.time() + (self.timeout if timeout is None else timeout)
while True:
idx = self._buf.find(s)
if idx >= 0:
self._set(self._buf[:idx], s, None, self._buf[idx + len(s) :])
return 0
remaining = deadline - time.time()
if remaining <= 0:
raise TIMEOUT()
chunk = self._read_fd(remaining) # raises EOF when the child is gone
if chunk:
self._buf += chunk
def close(self, force: bool = True) -> None:
try:
os.close(self.fd)
except OSError:
pass
if self.pid:
try:
os.kill(self.pid, signal.SIGKILL if force else signal.SIGHUP)
except ProcessLookupError:
pass
try:
os.waitpid(self.pid, 0)
except OSError:
pass
self.pid = 0
# Alias so call sites read `spawn(...)`.
spawn = _PtyProcess
_ANSI = re.compile(
r"\x1b\[[0-9;?]*[ -/]*[@-~]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|\x1b[=>]"
)
def _clean(s: str) -> str:
return _ANSI.sub("", s).replace("\r", "")
_TRANSPORTS = {"et": "et {host}", "ssh": "ssh -tt {host}"}
def _transport_cmd(via: str, host: str) -> str:
"""The command that opens an interactive PTY shell on HOST. `via` is a known
transport name (et/ssh) or a literal 'CMD {host}' template (host is appended if
it has no {host}). The default, et, gives the resilient eternal session."""
tmpl = _TRANSPORTS.get(via, via)
return tmpl.format(host=host) if "{host}" in tmpl else "%s %s" % (tmpl, host)
class EtSession:
def __init__(
self,
host: str,
via: str = "et",
connect_timeout: float = 60,
dims: tuple[int, int] = (50, 200),
remote_histfile: str | None = "/dev/null",
cmdlog: str | None = None,
) -> None:
self.host = host
self.timeout = connect_timeout
# remote_histfile: where the remote shell writes history.
# '/dev/null' (default) = discard; or a path = dedicated audit file;
# set None to leave the shell's default untouched.
self.remote_histfile = remote_histfile
# cmdlog: optional LOCAL file; the wrapper appends every command it runs.
self.cmdlog = cmdlog
self.prompt = "" # verbatim remote prompt line from the last expect()
# before/after/match are populated dynamically; keep the child typed Any and
# lean on the sentinel framing. The engine itself is fully typed above.
self.child: Any = spawn(
_transport_cmd(via, host),
encoding="utf-8",
timeout=connect_timeout,
dimensions=dims,
codec_errors="replace",
)
def connect(self) -> EtSession:
c = self.child
c.expect("❯", timeout=self.timeout)
self._drain(0.6) # flush the full p10k prompt render
hf = self.remote_histfile
# zsh side: stop it saving the exec line / anything to the real history
if hf is not None:
c.sendline("HISTFILE=%s; fc -p /dev/null 2>/dev/null" % hf)
self._drain(0.3)
# replace noisy interactive zsh+p10k with a bare, quiet bash
c.sendline("exec bash --norc --noprofile")
self._drain(0.6)
# clean-room env: empty prompts, dumb term, no \r, no pagers/interactive prompts
setup = (
"export PS1='' PS2='' PROMPT_COMMAND='' TERM=dumb "
"PAGER=cat GIT_PAGER=cat GIT_TERMINAL_PROMPT=0 "
"DEBIAN_FRONTEND=noninteractive; stty -echo -onlcr 2>/dev/null"
)
if hf is None:
pass
elif hf == "/dev/null":
setup = "set +o history; " + setup # don't record at all
else:
setup += "; export HISTFILE=%s" % hf # dedicated audit file
c.sendline(setup)
self._drain(0.6)
nonce = "RDY_" + uuid.uuid4().hex[:8]
for _ in range(8):
c.sendline("echo %s" % nonce)
try:
c.expect_exact(nonce, timeout=4) # stty -echo => appears once (output)
except TIMEOUT:
continue
except EOF:
raise RuntimeError("et session closed during connect")
else:
self._drain(0.4)
return self
raise RuntimeError("sync failed; tail=%r" % (c.before[-300:],))
def _drain(self, idle: float = 0.3) -> None:
"""Discard whatever output is buffered right now (settle the stream)."""
self.read_available(idle)
def _resync(self, attempts: int = 3) -> bool:
"""Recover a desynced session after a framing failure (a command that
hung reading stdin, or a truncated over-long line). Interrupt whatever is
running with Ctrl-C, drain, and confirm a clean prompt with a nonce.
Returns True if the session is usable again. This keeps one bad command
from poisoning the whole warm key."""
c = self.child
try:
for _ in range(attempts):
c.send("\x03") # Ctrl-C: kill a hung/partial command
self._drain(0.4)
nonce = "RSY_" + uuid.uuid4().hex[:8]
c.sendline("echo %s" % nonce)
try:
c.expect_exact(nonce, timeout=4)
self._drain(0.3)
return True
except TIMEOUT:
continue
except EOF:
return False
except Exception:
return False
return False
def run(self, cmd: str, timeout: float = 60) -> tuple[str, int]:
c = self.child
if self.cmdlog:
with open(self.cmdlog, "a") as fh:
fh.write(cmd + chr(10))
s = "S_" + uuid.uuid4().hex[:8]
e = "E_" + uuid.uuid4().hex[:8]
framed = "printf '%%s\\n' '%s'; %s; printf '%s%%d\\n' \"$?\"" % (s, cmd, e)
# Over-long lines would be truncated by the PTY's canonical cap and wedge
# the session; route them through the chunk-safe run_script path instead.
if len(framed) > _LINE_CAP:
return self.run_script(cmd, timeout=timeout)
c.sendline(framed)
try:
c.expect(re.escape(e) + r"(\d+)", timeout=timeout)
except TIMEOUT:
ok = self._resync()
raise RuntimeError(
"run() timed out waiting for the end sentinel (the command may "
"read stdin or hang; run() does not forward stdin, and is for "
"short commands, use run_script/`script` for input or long "
"payloads). session %s. cmd=%r"
% ("resynced" if ok else "may be wedged", cmd[:120])
)
code = int(c.match.group(1))
raw = _clean(c.before)
if s in raw:
raw = raw.split(s, 1)[1]
return raw.strip("\n"), code
def run_script(
self, script: str, timeout: float = 180, chunk: int = 400
) -> tuple[str, int]:
"""Inject an arbitrarily long/multi-line script safely. PTY canonical mode
caps a single input line (~1KB), so we base64-encode and append in small
chunks to a remote file, then decode and run it (preserving its exit code)."""
blob = base64.b64encode(script.encode()).decode()
remote = "/tmp/etch_%s.b64" % uuid.uuid4().hex[:8]
_, code = self.run(": > %s" % remote)
if code != 0:
raise RuntimeError("could not create remote temp file")
for i in range(0, len(blob), chunk):
_, code = self.run(
"printf '%%s' '%s' >> %s" % (blob[i : i + chunk], remote)
)
if code != 0:
raise RuntimeError("chunk append failed at offset %d" % i)
return self.run(
"base64 -d %s | bash; rc=$?; rm -f %s; (exit $rc)" % (remote, remote),
timeout=timeout,
)
# ── Interactive driving (write_bash-style) ────────────────────────────────
# run()/run_script() are for commands that finish and return to the prompt.
# For programs that PROMPT mid-run (sudo, [y/n], REPLs, mysql>), use these
# raw primitives: send input, expect a prompt pattern, read the live stream.
# Output here is raw (the program's own prompts/echo) — you match prompts
# yourself. Don't interleave run() while an interactive program is mid-flight;
# expect it back to the shell first. (Never hardcode a password — for an auth
# prompt, route the secret through secure input, not the script.)
def send(self, data: str) -> None:
"""Write raw bytes/text to the remote stdin (no trailing newline)."""
self.child.send(data)
def sendline(self, data: str = "") -> None:
"""Write a line (adds newline) to the remote stdin."""
self.child.sendline(data)
def _prompt_line(self) -> str:
"""Reconstruct the verbatim prompt the program is waiting at: the last
non-empty line of what preceded the match, plus the matched text itself
(e.g. '[sudo] password for gmendezb:'). Show this to a human who has to
answer it, so they know exactly what's being asked."""
before = _clean(self.child.before or "")
after = self.child.after if isinstance(self.child.after, str) else ""
last = ""
for ln in before.splitlines():
if ln.strip():
last = ln
return (last + _clean(after)).strip()
def expect(self, pattern: str | list[str], timeout: float = 30) -> tuple[str, int]:
"""Wait for a regex pattern (or list); returns (cleaned_text_before, index).
Also sets self.prompt to the verbatim prompt line that matched."""
idx = self.child.expect(pattern, timeout=timeout)
self.prompt = self._prompt_line()
return _clean(self.child.before), idx
def expect_exact(self, s: str, timeout: float = 30) -> str:
"""Like expect() but matches a literal string (no regex). Returns text before."""
self.child.expect_exact(s, timeout=timeout)
self.prompt = self._prompt_line()
return _clean(self.child.before)
def read_available(self, idle: float = 0.3) -> str:
"""Drain and return whatever output is currently buffered."""
buf = ""
while True:
try:
buf += self.child.read_nonblocking(4096, timeout=idle)
except (TIMEOUT, EOF):
break
return _clean(buf)
def read(self, timeout: float = 10, quiet: float = 0.4) -> str:
"""Show whatever the other end is saying when you DON'T have a prompt to
expect: wait up to `timeout` for output to start, then drain until `quiet`
seconds of silence. Returns cleaned text. Pattern-free 'peek at the screen'.
Line/stream output works; full-screen TUIs (vim/top) that redraw via cursor
escapes won't linearize cleanly — those need a terminal emulator (out of scope)."""
deadline = time.time() + timeout
buf = ""
while True:
remaining = deadline - time.time()
if remaining <= 0:
break
try:
buf += self.child.read_nonblocking(4096, timeout=min(quiet, remaining))
except TIMEOUT:
if buf:
break # had output, now silent => done
except EOF:
break
return _clean(buf)
def close(self) -> None:
try:
self.child.sendline("exit")
self.child.expect(EOF, timeout=5)
except Exception:
pass
finally:
self.child.close(force=True)
# ─────────────────────────────────────────────────────────────────────────────
# Daemon mode — keep ONE warm EtSession alive across separate processes/one-shot
# calls, so any later invocation reuses it (no ~13s reconnect) and shell state
# persists for the whole experiment/session.
#
# CONCURRENCY MODEL (important): a single EtSession is ONE serial conversation —
# interleaving run()s would tangle their sentinels. So the daemon is keyed: each
# `key` is its own daemon → its own `et` client → its own remote shell.
# * Same key = same warm shell, requests SERIALIZED (safe, but not parallel).
# * Diff key = isolated shell, runs in PARALLEL.
# => give each agent/subagent its own key (e.g. ETCH_KEY). Never share one key
# across actors that run concurrently.
# ─────────────────────────────────────────────────────────────────────────────
# ── Daemon: paths, process liveness & filesystem introspection ──
def _safe(key: str) -> str:
"""Filesystem-safe form of a daemon key."""
return re.sub(r"[^A-Za-z0-9_.-]", "_", key)
def _user_prefix() -> str:
"""Shared /tmp path prefix for this user's daemon files."""
return "/tmp/etch-%s-" % getpass.getuser()
def _sock_path(key: str) -> str:
return _user_prefix() + _safe(key) + ".sock"
def _pid_path(key: str) -> str:
return _sock_path(key)[:-5] + ".pid" # /tmp/etch-<user>-<key>.pid
def _trace_path(key: str) -> str:
return _sock_path(key)[:-5] + ".trace" # /tmp/etch-<user>-<key>.trace (peep)
def _alive(pid: int) -> bool:
# zombie-aware: a process that exists only as a defunct/zombie (state 'Z')
# is treated as NOT alive, so reap() isn't fooled by a just-killed daemon.
try:
st = subprocess.run(
["ps", "-p", str(pid), "-o", "state="],
capture_output=True,
text=True,
timeout=5,
).stdout.strip()
except Exception:
# ps unavailable: fall back to signal-0 probe
try:
os.kill(pid, 0)
return True
except OSError:
return False
return bool(st) and not st.startswith("Z")
def reap() -> None:
"""Sweep leftovers from crashed daemons: if a daemon PID is dead, kill its
orphaned `et` child (by recorded PID, verified) and remove its socket+pidfile.
Cheap; call opportunistically (open_session does)."""
for pf in glob.glob(_user_prefix() + "*.pid"):
try:
parts = open(pf).read().split()
dpid, epid = int(parts[0]), int(parts[1]) # pidfile: "<dpid> <epid> [host]"
except Exception:
try:
os.unlink(pf)
except OSError:
pass
continue
if _alive(dpid):
continue # daemon still running; leave it
if _alive(epid):
# verify the PID is really an `et` client before killing (PID reuse guard)
try:
cmd = subprocess.run(
["ps", "-p", str(epid), "-o", "command="],
capture_output=True,
text=True,
timeout=5,
).stdout
except Exception:
cmd = ""
if "/et " in cmd or cmd.strip().startswith("et ") or " et " in cmd:
try:
os.kill(epid, signal.SIGTERM)
except OSError:
pass
for p in (pf, pf[:-4] + ".sock", pf[:-4] + ".trace"):
try:
os.unlink(p)
except OSError:
pass
# Also sweep orphaned sockets with no pidfile (a daemon that crashed after
# binding but before writing its pidfile, e.g. connect() failed). The pidfile
# loop above can't see those. A socket with a live listener is a daemon still
# starting up, so probe it and only remove the ones nothing answers on.
for sk in glob.glob(_user_prefix() + "*.sock"):
if os.path.exists(sk[:-5] + ".pid"):
continue # has a pidfile (handled above, or a running daemon)
try:
probe = socket.socket(socket.AF_UNIX)
probe.settimeout(1)
probe.connect(sk)
probe.close()
continue # someone is listening (a daemon starting up): leave it
except OSError:
for p in (sk, sk[:-5] + ".trace"):
try:
os.unlink(p)
except OSError:
pass
def _live_keys() -> list[str]:
"""Keys of currently-bound daemons (from their socket files)."""
prefix = _user_prefix()
return [s[len(prefix) : -5] for s in sorted(glob.glob(prefix + "*.sock"))]
def _pidfile_fields(key: str) -> tuple[str, str, str | None]:
"""(daemon_pid_str, et_pid_str, host) from a daemon's pidfile, or ('?','?',None)."""
try:
parts = open(_pid_path(key)).read().split()
return parts[0], parts[1], (parts[2] if len(parts) > 2 else None)
except Exception:
return "?", "?", None
def _daemon_host(key: str) -> str | None:
"""The host a live daemon for KEY serves (from its pidfile), or None."""
return _pidfile_fields(key)[2]
# ── Daemon: length-prefixed JSON wire protocol (4-byte big-endian length + body) ──
def _recvn(conn: socket.socket, n: int) -> bytes | None:
buf = b""
while len(buf) < n:
chunk = conn.recv(n - len(buf))
if not chunk:
return None
buf += chunk
return buf
def _send_msg(conn: socket.socket, obj: Any) -> None:
data = json.dumps(obj).encode()
conn.sendall(struct.pack(">I", len(data)) + data)
def _recv_msg(conn: socket.socket) -> dict[str, Any] | None:
hdr = _recvn(conn, 4)
if not hdr:
return None
(n,) = struct.unpack(">I", hdr)
body = _recvn(conn, n)
return None if body is None else json.loads(body.decode())
def _xlate_pat(pat: str | list[str]) -> Any:
# the wire can't carry TIMEOUT/EOF objects, so callers pass the
# sentinels "<<TIMEOUT>>"/"<<EOF>>" and we translate them back here.
def one(p: str) -> Any:
if p == "<<TIMEOUT>>":
return TIMEOUT
if p == "<<EOF>>":
return EOF
return p
return [one(p) for p in pat] if isinstance(pat, list) else one(pat)
# ── Daemon: client (warm session reuse across processes) ──
class DaemonClient:
"""Thin client to a per-key etch daemon. Auto-starts the daemon on first
use. Same .run()/.run_script() API as EtSession, but warm across processes."""
def __init__(
self,
key: str,
host: str | None = None,
via: str = "et",
idle: int = 900,
autostart: bool = True,
connect_timeout: float = 90,
) -> None:
self.key, self.host, self.via, self.idle = key, host, via, idle
self.path = _sock_path(key)
self.conn: socket.socket | None = None # set once connected
self.prompt = "" # verbatim remote prompt line from the last expect()
self.started = (
False # True if this client spawned the daemon (vs reused a warm one)
)
if not self._connect():
if not autostart:
raise RuntimeError("no daemon for key %r" % key)
self._start(connect_timeout)
self.started = True
def _connect(self, timeout: float = 10) -> bool:
try:
c = socket.socket(socket.AF_UNIX)
c.settimeout(timeout)
c.connect(self.path)
self.conn = c
return True
except OSError:
return False
def _start(self, connect_timeout: float) -> None:
if not self.host:
raise RuntimeError("no host to start a daemon for key %r" % self.key)
try:
os.unlink(self.path) # clear a stale socket
except OSError:
pass
log = open("/tmp/etch-%s.log" % _safe(self.key), "a")
subprocess.Popen(
[
sys.executable,
os.path.realpath(__file__),
"--daemon",
self.key,
self.host,
self.via,
str(self.idle),
],
start_new_session=True,
stdin=subprocess.DEVNULL,
stdout=log,
stderr=log,
)
# The daemon binds the socket quickly but only accepts after its ~13s
# EtSession.connect(). So: wait for the socket to appear, then send one
# ping with a long timeout — the daemon answers once the shell is ready.
deadline = time.time() + connect_timeout
while time.time() < deadline:
if self._connect(timeout=connect_timeout):
break
time.sleep(0.3)
else:
raise RuntimeError("daemon socket for key %r never appeared" % self.key)
try:
_send_msg(cast(socket.socket, self.conn), {"op": "ping"})
r = _recv_msg(cast(socket.socket, self.conn))
except OSError as e:
raise RuntimeError("daemon ping failed for key %r: %r" % (self.key, e))
if not (r and r.get("ok")):
raise RuntimeError("daemon for key %r not ready" % self.key)
def _rpc(self, msg: dict[str, Any], timeout: float = 60) -> dict[str, Any]:
assert self.conn is not None, "not connected"
self.conn.settimeout(
timeout + 15
) # generous margin over the command's own timeout
_send_msg(self.conn, msg)
r = _recv_msg(self.conn)
if r is None:
raise RuntimeError("daemon closed the connection")
if "error" in r:
raise RuntimeError(r["error"])
return r
def run(self, cmd: str, timeout: float = 60) -> tuple[str, int]:
r = self._rpc({"op": "run", "cmd": cmd, "timeout": timeout}, timeout=timeout)
return r["out"], r["code"]
def run_script(self, script: str, timeout: float = 180) -> tuple[str, int]:
r = self._rpc(
{"op": "run_script", "script": script, "timeout": timeout}, timeout=timeout
)
return r["out"], r["code"]
# ── Interactive driving over the daemon (mirrors EtSession's primitives) ──
def send(self, data: str, secret: bool = False) -> None:
self._rpc({"op": "send", "data": data, "secret": secret}, timeout=10)
def sendline(self, data: str = "", secret: bool = False) -> None:
self._rpc({"op": "sendline", "data": data, "secret": secret}, timeout=10)
def expect(self, pattern: str | list[str], timeout: float = 30) -> tuple[str, int]:
# pass TIMEOUT/EOF as the sentinels "<<TIMEOUT>>"/"<<EOF>>"
r = self._rpc(
{"op": "expect", "pattern": pattern, "timeout": timeout}, timeout=timeout
)
self.prompt = r.get("prompt", "")
return r["out"], r["idx"]
def expect_exact(self, s: str, timeout: float = 30) -> str:
r = self._rpc(
{"op": "expect_exact", "s": s, "timeout": timeout}, timeout=timeout
)
self.prompt = r.get("prompt", "")
return r["out"]
def read(self, timeout: float = 10, quiet: float = 0.4) -> str:
r = self._rpc(
{"op": "read", "timeout": timeout, "quiet": quiet}, timeout=timeout
)
return r["out"]
def read_available(self, idle: float = 0.3) -> str:
r = self._rpc({"op": "read_available", "idle": idle}, timeout=max(5, idle * 3))
return r["out"]
def peep_start(self) -> str:
"""Enable the daemon's live trace; returns the trace file path to tail."""
return cast(str, self._rpc({"op": "peep_start"}, timeout=10).get("path"))
def peep_stop(self) -> None:
try:
self._rpc({"op": "peep_stop"}, timeout=10)
except Exception:
pass
def shutdown(self) -> None:
"""Stop the daemon (closes its et session + remote shell)."""
try:
_send_msg(cast(socket.socket, self.conn), {"op": "shutdown"})
_recv_msg(cast(socket.socket, self.conn))
except OSError:
pass
self.close()
def close(self) -> None:
"""Disconnect the client; the daemon stays warm for the next caller."""
if self.conn:
try:
self.conn.close()
except OSError:
pass
self.conn = None
def _random_key() -> str:
"""A short, opaque session handle (used when `open` gets no --key)."""
return uuid.uuid4().hex[:8]
def open_session(
host: str, key: str | None = None, via: str = "et", idle: int = 900
) -> DaemonClient:
"""Open (spawn-or-reuse) a warm per-key daemon to HOST and return a client.
KEY is a logical handle for the session (random if omitted). The host lives
with the daemon, not the key. `via` is the transport (et/ssh or a 'CMD {host}'
template). Operate on the result with .run()/.send()/etc.; a later caller
reattaches with DaemonClient(key, autostart=False).
CONCURRENCY: a daemon is ONE serial conversation. One logical actor per key;
give concurrent actors (e.g. subagents) distinct keys, or use EtSession()
directly for an isolated one-shot shell.
"""
reap() # sweep any crashed-daemon orphans first
return DaemonClient(key=key or _random_key(), host=host, via=via, idle=idle)
# ── Daemon: server (the long-lived worker that owns one warm EtSession) ──
def _daemon_main(key: str, host: str, via: str, idle: int) -> None:
idle = int(idle)
path = _sock_path(key)
pidpath = _pid_path(key)
srv = socket.socket(socket.AF_UNIX)
try:
srv.bind(path) # fails if another daemon already won
except OSError:
return
os.chmod(path, 0o600)
srv.listen(8)
try:
sess = EtSession(
host=host,
via=via,
cmdlog="/tmp/etch-%s.cmdlog" % _safe(key),
).connect()
except Exception:
# connect() failed (e.g. VM unreachable): don't leave the bound socket
# behind with no pidfile, that would show in `ls` as a dead orphan.
try:
srv.close()
finally:
try:
os.unlink(path)
except OSError:
pass
os._exit(1)
# record daemon PID + et-client PID (for reap) + the host this daemon serves
try:
open(pidpath, "w").write("%d %d %s" % (os.getpid(), sess.child.pid, host))
except OSError:
pass
tracepath = _trace_path(key)
trace: dict[str, Any] = {"fh": None, "n": 0} # peep: refcounted op-level live trace
def _trace(arrow: str, text: str) -> None:
if not trace["fh"]:
return # zero cost when nobody is peeping
t = text.replace("\n", "\\n")
if len(t) > 4000:
t = t[:4000] + "...(+%d)" % (len(t) - 4000)
try:
trace["fh"].write("%s %s %s\n" % (time.strftime("%H:%M:%S"), arrow, t))
trace["fh"].flush()
except Exception:
pass
def _graceful(*_: Any) -> None:
# Unlink the fs entries FIRST so a stopped daemon vanishes from ls/peers
# immediately, before the (slow) session teardown waits on et to exit.
for p in (path, pidpath, tracepath):
try:
os.unlink(p)
except OSError:
pass
try:
srv.close()
except Exception:
pass
try:
if trace["fh"]:
trace["fh"].close()
except Exception:
pass
try:
sess.close()
except Exception:
pass
os._exit(0)
# Session ops: each turns a request into a response dict; the loop wraps any
# exception into {"error": ...}. (Control ops ping/peep/shutdown are inline.)
def _op_run(msg: dict[str, Any]) -> dict[str, Any]:
_trace("»", "run: " + msg["cmd"])
out, code = sess.run(msg["cmd"], timeout=msg.get("timeout", 60))
_trace("«", "(rc=%d) %s" % (code, out))
return {"out": out, "code": code}
def _op_run_script(msg: dict[str, Any]) -> dict[str, Any]:
_trace("»", "run_script: <%d bytes>" % len(msg.get("script", "")))
out, code = sess.run_script(msg["script"], timeout=msg.get("timeout", 180))
_trace("«", "(rc=%d) %s" % (code, out))
return {"out": out, "code": code}
def _op_send(msg: dict[str, Any]) -> dict[str, Any]:
sess.send(msg.get("data", ""))
_trace("»", "<secret>" if msg.get("secret") else repr(msg.get("data", "")))
return {"ok": True}
def _op_sendline(msg: dict[str, Any]) -> dict[str, Any]:
sess.sendline(msg.get("data", ""))
_trace("»", "<secret>" if msg.get("secret") else msg.get("data", ""))
return {"ok": True}
def _op_expect(msg: dict[str, Any]) -> dict[str, Any]:
out, idx = sess.expect(
_xlate_pat(msg["pattern"]), timeout=msg.get("timeout", 30)
)
_trace("«", "[matched %d] %s" % (idx, sess.prompt or out[-160:]))
return {"out": out, "idx": idx, "prompt": sess.prompt}
def _op_expect_exact(msg: dict[str, Any]) -> dict[str, Any]:
out = sess.expect_exact(msg["s"], timeout=msg.get("timeout", 30))
_trace("«", sess.prompt or out[-160:])
return {"out": out, "prompt": sess.prompt}
def _op_read(msg: dict[str, Any]) -> dict[str, Any]:
out = sess.read(timeout=msg.get("timeout", 10), quiet=msg.get("quiet", 0.4))
_trace("«", out)
return {"out": out}
def _op_read_available(msg: dict[str, Any]) -> dict[str, Any]:
out = sess.read_available(idle=msg.get("idle", 0.3))
_trace("«", out)
return {"out": out}
session_ops: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
"run": _op_run,
"run_script": _op_run_script,
"send": _op_send,
"sendline": _op_sendline,
"expect": _op_expect,
"expect_exact": _op_expect_exact,
"read": _op_read,
"read_available": _op_read_available,
}
def _op_peep_start() -> dict[str, Any]:
if trace["n"] == 0:
try:
trace["fh"] = open(tracepath, "a")
except Exception:
trace["fh"] = None
trace["n"] += 1
return {"ok": True, "path": tracepath}
def _op_peep_stop() -> dict[str, Any]:
if trace["n"] > 0:
trace["n"] -= 1
if trace["n"] == 0 and trace["fh"]:
try:
trace["fh"].close()
except Exception:
pass
trace["fh"] = None
try:
os.unlink(tracepath)
except OSError:
pass
return {"ok": True}
# SIGALRM fires the idle timeout in ANY blocking state (accept OR recv), so a
# client that holds a connection open-but-idle can't pin the daemon forever.
# SIGTERM (e.g. from reap() on a sibling, or a manual stop) cleans up too.
signal.signal(signal.SIGALRM, _graceful)
signal.signal(signal.SIGTERM, _graceful)
try:
while True:
signal.alarm(idle) # idle countdown while waiting to accept
conn, _ = srv.accept()
signal.alarm(0)
with conn:
try:
while True:
signal.alarm(idle) # idle countdown while waiting for a request
msg = _recv_msg(conn)
signal.alarm(0) # request in hand; never interrupt a command
if msg is None:
break
op = msg.get("op")
if op in session_ops:
try:
_send_msg(conn, session_ops[op](msg))
except Exception as e:
_send_msg(conn, {"error": repr(e)})
elif op == "ping":
_send_msg(conn, {"ok": True})
elif op == "peep_start":
_send_msg(conn, _op_peep_start())
elif op == "peep_stop":
_send_msg(conn, _op_peep_stop())
elif op == "shutdown":
_send_msg(conn, {"ok": True})
_graceful()
else:
_send_msg(conn, {"error": "unknown op %r" % op})
except OSError:
pass # client vanished mid-request; back to accept()
finally:
_graceful()
# ═════════════════════════════════════════════════════════════════════════════
# Command-line interface
# helpers (host guard, open/attach, payload) -> command funcs -> argparse
# ═════════════════════════════════════════════════════════════════════════════
def _use(key: str, host: str | None = None, via: str = "et") -> DaemonClient:
"""Open KEY's session for an operation: reuse the warm daemon if it exists,
else autospawn it on HOST (the optional --host) via the chosen transport. With
no warm daemon and no host, exit with a hint. Shared by every session verb."""
try:
s = DaemonClient(key=key, host=host, via=via, autostart=host is not None)
except RuntimeError:
sys.exit(
"etch: no session %r; pass --host <user@host> to start it, "
"or run `etch open <user@host>` first" % key
)
if host and not s.started:
# reused a warm daemon; refuse a --host that contradicts its real host
live = _daemon_host(key)
if live and live != host:
s.close()
sys.exit(
"etch: session %r is live on %s, not %s; close it (`etch close %s`) "
"or use a different key" % (key, live, host, key)
)
if s.started:
sys.stderr.write(
"(started a warm daemon for %r on %s; `close %s` to end it)\n"
% (key, s.host, key)
)
return s
def _payload(key: str, secret: bool, message: str | None) -> str:
"""Source the bytes to send: -m literal, --secret (silent getpass), a stdin
pipe, or an interactive prompt."""
if message is not None:
return message
if secret:
return getpass.getpass("secret for %s: " % key)
if not sys.stdin.isatty():
return sys.stdin.read()
return input("send to %s: " % key)
# ── CLI: command implementations (one per subcommand) ──
def _do_send(
key: str,
newline: bool,
secret: bool = False,
message: str | None = None,
host: str | None = None,
via: str = "et",
) -> None:
"""`send`/`sendline`: feed input into a running daemon's warm session from
YOUR own shell. The value goes terminal -> daemon -> remote stdin and is
never seen by the agent (not in its transcript, checkpoints, logs, cmdlog,
or the remote shell history). With --secret it's also redacted from peep."""
payload = _payload(key, secret, message)
if newline and payload.endswith("\n"):
payload = payload[:-1] # sendline re-adds exactly one newline
cli = _use(key, host, via)
(cli.sendline if newline else cli.send)(payload, secret=secret)
cli.close()
sys.stderr.write(
"sent %d chars to %r%s%s\n"
% (
len(payload),
key,
" + newline" if newline else "",
" (silent, redacted in peep)" if secret else "",
)
)
def _cmd_expect(
key: str,
pattern: str,
timeout: float = 30,
exact: bool = False,
host: str | None = None,
via: str = "et",
) -> None:
"""Wait for PATTERN (a regex, or a literal with --exact). Prints the matched
prompt line and exits 0 on a match, or exits 1 on timeout."""
cli = _use(key, host, via)
pat = re.escape(pattern) if exact else pattern
out, idx = cli.expect([pat, "<<TIMEOUT>>"], timeout=timeout)
prompt = cli.prompt
cli.close()
if idx == 0:
if prompt:
print(prompt)
sys.exit(0)
sys.stderr.write("(timeout after %gs waiting for %r)\n" % (timeout, pattern))
sys.exit(1)
def _cmd_open(host: str, key: str | None = None, via: str = "et") -> None:
key = key or _random_key()
existing = _daemon_host(key)
if existing and existing != host:
sys.exit(
"etch: key %r is already open on %s; close it or pick a different --key"
% (key, existing)
)
open_session(host, key=key, via=via).close() # spawn-or-reuse, then leave warm
print(key) # the handle to reuse; a printed key means the session is open
def _cmd_run(
key: str, cmd: str, timeout: float, host: str | None = None, via: str = "et"
) -> None:
s = _use(key, host, via)
try:
out, code = s.run(cmd, timeout=timeout)
finally:
s.close()
if out:
print(out)
sys.exit(code)
def _cmd_script(
key: str,
path: str | None,
timeout: float,
host: str | None = None,
via: str = "et",
) -> None:
"""Run a multi-line script on the warm session via run_script (line-cap-safe,
chunked base64). Reads from FILE, or stdin if FILE is omitted or '-'."""
text = sys.stdin.read() if path in (None, "-") else open(path).read()
s = _use(key, host, via)
try:
out, code = s.run_script(text, timeout=timeout)
finally:
s.close()
if out:
print(out)
sys.exit(code)
def _cmd_attach(key: str, host: str | None = None, via: str = "et") -> None:
"""Raw interactive terminal bridged into the warm session: drive prompts,
REPLs, sudo, etc. live. Ctrl-] detaches (the daemon stays warm). Exclusive:
holds the serial daemon for the whole session, so other callers (e.g. an agent
run) block until you detach. Don't drive one session from two places at once."""
if not sys.stdin.isatty():
sys.exit("attach: needs an interactive terminal (stdin is not a tty)")
s = _use(key, host, via)
# turn the clean-room shell into a normal interactive terminal for the attach
s.sendline(
"stty sane 2>/dev/null; export TERM=xterm-256color PS1='[et:%s] \\w\\$ ' PS2='> '"
% key
)
s.read(timeout=1.0)
sys.stderr.write(
"== attached to %r on %s — press Ctrl-] to detach ==\r\n" % (key, s.host or "?")
)
sys.stderr.flush()
s.send("\n") # draw a fresh prompt
fd = sys.stdin.fileno()
old = termios.tcgetattr(fd)
_no_fds: list[int] = []
try:
tty.setraw(fd)
while True:
r, _, _ = select.select([fd], _no_fds, _no_fds, 0.03)
if r:
data = os.read(fd, 4096)
if not data or b"\x1d" in data: # EOF or Ctrl-] detaches
head = data.split(b"\x1d", 1)[0] if data else b""
if head:
s.send(head.decode("utf-8", "replace"))
break
s.send(data.decode("utf-8", "replace"))
out = s.read_available(idle=0.02)
if out:
sys.stdout.write(out.replace("\n", "\r\n")) # raw tty needs CRLF
sys.stdout.flush()
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old)
try: # restore clean-room so scripted run()s keep working
s.sendline("stty -echo -onlcr 2>/dev/null; export PS1='' PS2='' TERM=dumb")
s.read(timeout=1.0)
except Exception:
pass
s.close()
sys.stderr.write("\r\n== detached from %r (still warm) ==\r\n" % key)
sys.stderr.flush()
def _cmd_peep(
key: str,
duration: float | None = None,
host: str | None = None,
via: str = "et",
) -> None:
cli = _use(key, host, via)
path = cli.peep_start()
cli.close() # release the daemon: it stays serial, we just tail the file
signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt()))
suffix = (" for %gs" % duration) if duration else ""
print(
"== peep %r%s (» sent « received) — Ctrl-C to stop ==" % (key, suffix),
file=sys.stderr,
)
for _ in range(20):
if path and os.path.exists(path):
break
time.sleep(0.1)
deadline = (time.time() + duration) if duration else None
try:
with open(path) as fh:
fh.seek(0, 2) # only traffic from now on
while deadline is None or time.time() < deadline:
line = fh.readline()
if line:
sys.stdout.write(line)
sys.stdout.flush()
else:
time.sleep(0.2)
except (KeyboardInterrupt, FileNotFoundError):
pass
finally:
try:
DaemonClient(key=key, autostart=False).peep_stop()
except Exception:
pass
print("\n== peep stopped ==", file=sys.stderr)
def _cmd_read(
key: str,
timeout: float | None = None,
host: str | None = None,
via: str = "et",
) -> None:
"""Show what the session has emitted, without attaching: a pattern-free peek.
Default grabs whatever is buffered right now; --timeout waits for output to
start then drains until quiet. Pairs with `send`/`sendline` for step-by-step
interactive driving from the shell."""
cli = _use(key, host, via)
out = cli.read(timeout=timeout) if timeout else cli.read_available()
cli.close()
sys.stdout.write(out if (not out or out.endswith("\n")) else out + "\n")
def _cmd_ls() -> None:
prefix = _user_prefix()
socks = sorted(glob.glob(prefix + "*.sock"))
if not socks:
print("no etch daemons running")
return
now = time.time()
print(
"%-22s %-10s %-7s %-6s %-34s %s"
% ("KEY", "DAEMON", "ET", "AGE", "HOST", "SOCKET")
)
for sk in socks:
key = sk[len(prefix) : -5]
dpid, epid, host = _pidfile_fields(key)
alive = dpid.isdigit() and _alive(int(dpid))
try:
age = "%ds" % int(now - os.path.getmtime(sk))
except OSError:
age = "?"
print(
"%-22s %-10s %-7s %-6s %-34s %s"
% (key, dpid + ("" if alive else "!dead"), epid, age, host or "?", sk)
)
def _cmd_stop(key: str | None, all_: bool) -> None:
if all_:
keys = _live_keys()
elif key:
keys = [key]
else:
sys.exit("close: pass a KEY or --all")
if not keys:
print("no sessions to close")
return
for k in keys:
try:
DaemonClient(key=k, autostart=False).shutdown()
print("closed %r" % k)
except Exception as e:
print("could not close %r: %s" % (k, e))
def _cmd_selftest(host: str, via: str = "et") -> None:
t0 = time.time()
sess = EtSession(host, via=via).connect()
print(f"== connected in {time.time() - t0:.1f}s ==", flush=True)
for cmd in [
"hostname",
"whoami",
"pwd",
"echo $((6*7))",
"uname -sm",
"echo multi; echo line; echo test",
"false",
"ls /nonexistent 2>&1",
]:
out, code = sess.run(cmd)
print(f" rc={code:<3} {cmd!r:38} -> {out!r}", flush=True)
sess.close()
print("== done ==")
# ── CLI: argument parsing & dispatch ──
def _main_cli(argv: list[str]) -> None:
import argparse
p = argparse.ArgumentParser(
prog="etch",
description="Drive a persistent EternalTerminal session to the VM: warm daemon, "
"clean output + exit codes, interactive send/read, live peep.",
epilog=(
"command groups:\n"
" run a command: run, script clean output + exit codes\n"
" drive a session: attach, send, sendline, expect, read interactive (live or step-by-step)\n"
" observe: peep live tap of what's sent/received\n"
" daemons: open, close/stop, ls/status, gc open / list / close warm shells\n"
"\nexamples:\n"
" etch open user@vm prints a key; or: etch open user@vm --key main\n"
" etch run main uname -a reuse the open session\n"
" etch run --host user@vm main uname -a autospawn 'main' on first use, reuse after\n"
" etch run main -- some-tool --timeout 5 -- makes the rest a literal command\n"
" etch sendline main --secret feed a password into a parked prompt\n"
"\nKEY is a logical handle for a session; the host lives with the daemon, not\n"
"the key. Set it once at `open`, or pass --host on any verb to autospawn a\n"
"session that isn't open yet (optional once it is). `ls` shows each host.\n"
"\nlibrary: from Python, `from etch import open_session, EtSession`.\n"
" open_session(host, key=...) opens/reuses a warm daemon; EtSession(host) is\n"
" a one-shot with no daemon.\n"
"\nadvanced: `--daemon KEY HOST IDLE` is the internal daemon worker,\n"
"spawned automatically by open_session(); do not run it by hand.\n"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
sub = p.add_subparsers(dest="cmd", metavar="<command>")
def via_arg(sp: argparse.ArgumentParser) -> None:
sp.add_argument(
"--via",
default="et",
metavar="TRANSPORT",
help="transport to open with: et (default), ssh, or a 'CMD {host}' template (only when starting)",
)
def host_arg(sp: argparse.ArgumentParser) -> None:
sp.add_argument(
"--host",
metavar="USER@HOST",
help="autospawn the session here if it isn't open yet (optional once open)",
)
via_arg(sp)
def key_arg(sp: argparse.ArgumentParser) -> None:
sp.add_argument("key", help="session key (a handle from `etch open`)")
# ── run a command (clean output + exit codes) ──
pr = sub.add_parser(
"run",
help="run one command on the session (open it first, or --host to autospawn); exit code = remote rc",
)
pr.add_argument("--timeout", type=float, default=60)
host_arg(pr)
key_arg(pr)
pr.add_argument(
"command",
nargs=argparse.REMAINDER,
help="the command (rest of the line, flags and all); etch's own options go "
"before the key, or put `--` first for a fully literal command",
)
psc = sub.add_parser(
"script",
help="run a multi-line script (from FILE or stdin) on the warm session; exit code = remote rc",
)
psc.add_argument("--timeout", type=float, default=180)
host_arg(psc)
key_arg(psc)
psc.add_argument("file", nargs="?", help="script file (default: stdin)")
# ── drive a session interactively (live, or step-by-step send/expect/read) ──
pat = sub.add_parser(
"attach",
help="interactive terminal bridged into the warm session: prompts/REPLs/sudo, Ctrl-] detaches (exclusive until you detach)",
)
host_arg(pat)
key_arg(pat)
for name, nl in (("send", False), ("sendline", True)):
sp = sub.add_parser(
name,
help=(
"%s to the session (%s); -m TEXT, --secret (silent), or stdin"
% (name, "appends a newline" if nl else "raw, no newline")
),
)
sp.add_argument("key", help="session key to send to")
host_arg(sp)
sp.add_argument(
"-s",
"--secret",
action="store_true",
help="read the payload via a silent no-echo prompt",
)
sp.add_argument(
"-m",
"--message",
help="payload on the command line (visible; not for secrets)",
)
pex = sub.add_parser(
"expect",
help="wait for a regex (or literal with --exact) to appear; exit 0 on match, 1 on timeout",
)
pex.add_argument("key", help="daemon key to watch")
pex.add_argument("pattern", help="regex to wait for (literal if --exact)")
pex.add_argument(
"--timeout",
type=float,
default=30,
metavar="SECS",
help="give up after SECS (default 30)",
)
pex.add_argument(
"--exact", action="store_true", help="match PATTERN literally, not as a regex"
)
host_arg(pex)
prd = sub.add_parser(
"read",
help="drain session output without attaching: bare = peek what's buffered now (read_available), --timeout = wait for output then drain (read)",
)
prd.add_argument("key", help="daemon key to read from")
prd.add_argument(
"--timeout",
type=float,
default=None,
metavar="SECS",
help="wait up to SECS for output to start, then drain until quiet "
"(default: just grab what's buffered now)",
)
host_arg(prd)
# ── observe ──
ppe = sub.add_parser(
"peep", help="live tap: show what's sent/received on the session"
)
ppe.add_argument("key", help="daemon key to tap")
ppe.add_argument(
"--for",
dest="duration",
type=float,
default=None,
metavar="SECS",
help="auto-stop after SECS (for non-interactive/scripted taps)",
)
host_arg(ppe)
# ── sessions: open / close ──
po = sub.add_parser(
"open",
help="open a session on HOST (pre-warm); prints the key to reuse with run/send/...",
)
po.add_argument("host", metavar="USER@HOST", help="user@host to connect")
po.add_argument("--key", help="name for the session (default: a random handle)")
via_arg(po)
for nm, h in (
("close", "close a session (shut down its daemon)"),
("stop", "alias for close"),
):
psx = sub.add_parser(nm, help=h)
psx.add_argument("key", nargs="?", help="session key to close (or use --all)")
psx.add_argument(
"--all", action="store_true", help="close every running session"
)
# ── inspect / maintain ──
sub.add_parser("ls", help="list running sessions")
sub.add_parser("status", help="alias for ls")
sub.add_parser("gc", help="reap dead daemons / orphaned et processes")
pse = sub.add_parser(
"selftest", help="connect to HOST and run a few sample commands"
)
pse.add_argument("host", metavar="USER@HOST", help="user@host to connect")
via_arg(pse)
args = p.parse_args(argv)
if not args.cmd:
p.print_help()
return
if args.cmd == "run":
cmd = " ".join(args.command).strip()
if not cmd:
sys.exit("run: no command given")
if "--" not in argv:
_flags = ("--timeout", "--host", "--via")
stray = [
t
for t in args.command
if t in _flags or any(t.startswith(f + "=") for f in _flags)
]
if stray:
sys.stderr.write(
"etch: note: %s is inside the command, not applied as an option. "
"etch's flags go before the key (e.g. `run --timeout 5 KEY ...`); "
"use `--` for a command that legitimately contains them.\n"
% ", ".join(stray)
)
_cmd_run(args.key, cmd, args.timeout, args.host, args.via)
elif args.cmd == "script":
_cmd_script(args.key, args.file, args.timeout, args.host, args.via)
elif args.cmd == "open":
_cmd_open(args.host, args.key, args.via)
elif args.cmd == "attach":
_cmd_attach(args.key, args.host, args.via)
elif args.cmd in ("send", "sendline"):
_do_send(
args.key,
args.cmd == "sendline",
args.secret,
args.message,
args.host,
args.via,
)
elif args.cmd == "expect":
_cmd_expect(
args.key, args.pattern, args.timeout, args.exact, args.host, args.via
)
elif args.cmd == "peep":
_cmd_peep(args.key, args.duration, args.host, args.via)
elif args.cmd == "read":
_cmd_read(args.key, args.timeout, args.host, args.via)
elif args.cmd in ("ls", "status"):
_cmd_ls()
elif args.cmd in ("close", "stop"):
_cmd_stop(args.key, args.all)
elif args.cmd == "gc":
reap()
print("reaped stale etch daemons/orphans")
elif args.cmd == "selftest":
_cmd_selftest(args.host, args.via)
if __name__ == "__main__":
a = sys.argv[1:]
if a and a[0] == "--daemon": # internal worker, spawned by DaemonClient._start
_daemon_main(a[1], a[2], a[3], int(a[4]))
sys.exit(0)
_main_cli(a)
