dirigence/reliquary
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

♻️ gremlins: platform seam (WindowsPlatform, PosixPlatform) and path matrix

Browse Source
This commit is contained in:
g4borg
2026-10-16 20:36:18 +02:00
parent 065afcba4c
commit 25d03587b4
2 changed files with 629 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/spire/skills/gremlins/gremlins.py
+470 -258
View File
@@ -45,13 +45,6 @@ EXIT_USAGE = 2
EXIT_WAIT_TIMEOUT = 3
EXIT_JOB_FAILED = 4
IS_WINDOWS = os.name == "nt"
# How the detached supervisor is created (C6, P0.1 rung 3 hidden; POSIX: new session).
DETACH_METHOD = "cim-hidden" if IS_WINDOWS else "setsid"
CREATE_NO_WINDOW = 0x08000000
_NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0
# Registry file semantics (C3a). A lock is stale when its holder (pid + creation time) is dead,
# or when it is older than LOCK_ABANDONED_S. Every critical section is a read-modify-write of a few
# small JSON files: milliseconds, at worst a few seconds of PermissionError retries (40 x 50 ms
@@ -176,14 +169,6 @@ NOTE_MTIME_SLACK_S = 2.0
CLAUDE_PROJECTS_DIR: str | None = None
TRANSCRIPT_TAIL_BYTES = 256 * 1024
# CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars.
POWERSHELL_CIM = (
"$si = New-CimInstance -ClassName Win32_ProcessStartup -ClientOnly -Property @{ShowWindow=[uint16]0}; "
"$r = Invoke-CimMethod -ClassName Win32_Process -MethodName Create -Arguments "
"@{CommandLine=$env:SPIRE_GREMLIN_DETACH_CMD; CurrentDirectory=$env:SPIRE_GREMLIN_DETACH_CWD; "
"ProcessStartupInformation=$si}; "
"\"{0} {1}\" -f $r.ReturnValue, $r.ProcessId")
# --------------------------------------------------------------------------------------
# Errors, output, clock
# --------------------------------------------------------------------------------------
@@ -271,23 +256,154 @@ def fingerprint_hash(launch: dict) -> str:
return _sha256_bytes(canonical_json(fingerprint_view(launch)).encode("utf-8"))
# --------------------------------------------------------------------------------------
# Paths
# --- platform seam: begin -------------------------------------------------------------
# Everything that differs between Windows and POSIX lives in this one section (digest §3):
# the `Platform` interface, its two implementations, `select_platform()` and `PLATFORM`.
# Neutral code calls `PLATFORM.<method>` or the module-level delegates below the seam. Both
# classes can be built on any host: ctypes is loaded lazily (on the first process or window
# call), and `norm_path` is pure string logic (except Windows' `cygpath -w` for Git Bash
# rooted paths), so the path matrix and the gate tests run everywhere.
# --------------------------------------------------------------------------------------
class Platform:
"""The platform interface. `supported` is the release gate (digest §4): only a tested
platform is supported; lifting the gate for another one is a code change."""
name = "?" # "windows" | "posix"
detach_method = "?" # "cim-hidden" | "setsid"
supported = False
def describe(self) -> dict:
return {"name": self.name, "sys_platform": sys.platform, "wsl": self.is_wsl(),
"supported": bool(self.supported), "detach": self.detach_method}
def is_wsl(self) -> bool:
return False
@staticmethod
def norm_path(p: str, cwd: str | None = None) -> str:
raise NotImplementedError
def popen_kwargs(self, new_group: bool = False) -> dict:
raise NotImplementedError
def detach(self, argv: list[str], cwd: str, log_path: str) -> dict:
"""Start `argv` detached from the caller. {"ok": True, "method", "pid", "ctime"} or
{"ok": False, "method", "detail"}."""
raise NotImplementedError
def proc_ctime(self, pid: int) -> int | None:
raise NotImplementedError
def epoch_to_ctime(self, t: float | None) -> int | None:
raise NotImplementedError
def _running(self, pid: int) -> bool:
raise NotImplementedError
def pid_alive(self, pid: int, ctime: int | None = None) -> bool:
"""True if `pid` runs and (when both are known) its creation time equals `ctime`."""
if pid is None or int(pid) <= 0:
return False
pid = int(pid)
if not self._running(pid):
return False
if ctime is not None:
now_ct = self.proc_ctime(pid)
if now_ct is not None and now_ct != ctime:
return False
return True
def process_table(self) -> list[tuple[int, int, str]]:
raise NotImplementedError
def terminate(self, pid: int) -> bool:
raise NotImplementedError
def tree_kill(self, pid: int) -> int | None:
raise NotImplementedError
def restore_no_replace(self, src: str, dst: str) -> bool:
raise NotImplementedError
def visible_windows(self) -> list[tuple[int, int, str, str]] | None:
return None
class WindowsPlatform(Platform):
name = "windows"
detach_method = "cim-hidden" # C6, P0.1 rung 3 with ShowWindow=0
supported = True
CREATE_NO_WINDOW = 0x08000000
_GITBASH_DRIVE = re.compile(r"^/([A-Za-z])(?=/|$)")
_WSL_MOUNT = re.compile(r"^/mnt/[A-Za-z](?=/|$)")
_FILETIME_EPOCH = 116444736000000000 # 1970-01-01 in FILETIME ticks (100 ns since 1601)
_PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
_PROCESS_TERMINATE = 0x0001
_STILL_ACTIVE = 259
_ERROR_ACCESS_DENIED = 5
_TH32CS_SNAPPROCESS = 0x00000002
# CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars.
POWERSHELL_CIM = (
"$si = New-CimInstance -ClassName Win32_ProcessStartup -ClientOnly -Property @{ShowWindow=[uint16]0}; "
"$r = Invoke-CimMethod -ClassName Win32_Process -MethodName Create -Arguments "
"@{CommandLine=$env:SPIRE_GREMLIN_DETACH_CMD; CurrentDirectory=$env:SPIRE_GREMLIN_DETACH_CWD; "
"ProcessStartupInformation=$si}; "
"\"{0} {1}\" -f $r.ReturnValue, $r.ProcessId")
def __init__(self):
self._api = None
def _win(self):
"""(ctypes, ctypes.wintypes, kernel32, PROCESSENTRY32W, INVALID_HANDLE_VALUE), loaded on
first use (never at import or construction)."""
if self._api is None:
import ctypes
import ctypes.wintypes as wt
k32 = ctypes.WinDLL("kernel32", use_last_error=True)
k32.OpenProcess.argtypes = [wt.DWORD, wt.BOOL, wt.DWORD]
k32.OpenProcess.restype = wt.HANDLE
k32.GetProcessTimes.argtypes = [wt.HANDLE] + [ctypes.POINTER(wt.FILETIME)] * 4
k32.GetProcessTimes.restype = wt.BOOL
k32.GetExitCodeProcess.argtypes = [wt.HANDLE, ctypes.POINTER(wt.DWORD)]
k32.GetExitCodeProcess.restype = wt.BOOL
k32.CloseHandle.argtypes = [wt.HANDLE]
k32.CloseHandle.restype = wt.BOOL
class PE32(ctypes.Structure):
_fields_ = [("dwSize", wt.DWORD), ("cntUsage", wt.DWORD), ("th32ProcessID", wt.DWORD),
("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", wt.DWORD),
("cntThreads", wt.DWORD), ("th32ParentProcessID", wt.DWORD),
("pcPriClassBase", wt.LONG), ("dwFlags", wt.DWORD),
("szExeFile", wt.WCHAR * 260)]
k32.CreateToolhelp32Snapshot.argtypes = [wt.DWORD, wt.DWORD]
k32.CreateToolhelp32Snapshot.restype = wt.HANDLE
k32.Process32FirstW.argtypes = [wt.HANDLE, ctypes.POINTER(PE32)]
k32.Process32FirstW.restype = wt.BOOL
k32.Process32NextW.argtypes = [wt.HANDLE, ctypes.POINTER(PE32)]
k32.Process32NextW.restype = wt.BOOL
k32.TerminateProcess.argtypes = [wt.HANDLE, wt.UINT]
k32.TerminateProcess.restype = wt.BOOL
self._api = (ctypes, wt, k32, PE32, wt.HANDLE(-1).value)
return self._api
# --- paths ---
@staticmethod
def find_cygpath() -> str | None:
"""`cygpath` on PATH (Git Bash ships it), or None. Tests monkeypatch this lookup."""
"""`cygpath` on PATH (Git Bash ships it), or None. Tests patch this lookup."""
return shutil.which("cygpath")
@staticmethod
def cygpath_w(p: str) -> str:
"""M13: a Git Bash rooted path that is not a drive form (`/tmp/x`, `/usr/...`) converted with
`cygpath -w`. Refused (exit 2) when cygpath is not on PATH or fails; never mapped onto the
cwd's drive."""
exe = find_cygpath()
"""M13: a Git Bash rooted path that is not a drive form (`/tmp/x`, `/usr/...`) converted
with `cygpath -w`. Refused (exit 2) when cygpath is not on PATH or fails; never mapped
onto the cwd's drive."""
exe = WindowsPlatform.find_cygpath()
if not exe:
raise GremlinsError("Git Bash rooted path is not a drive form and cygpath is not on PATH to "
"convert it; pass a drive path (G:\\... or /g/...)", path=p)
@@ -301,32 +417,27 @@ def cygpath_w(p: str) -> str:
detail=r.stderr.decode("utf-8", "replace").strip()[:200] or "no output")
return out
def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str:
"""Normalise relative, `G:\\…`, `G:/…` and Git Bash `/g/…` forms.
Windows: absolute path with backslashes and an upper-case drive letter. A Git Bash rooted
path that is not a drive form (`/tmp/x`) goes through `cygpath -w` (M13); a native rooted
path (`\\x`) keeps its Windows meaning (the cwd's drive).
POSIX: `os.path.abspath` semantics (relative to `cwd`).
"""
if windows is None:
windows = IS_WINDOWS
@staticmethod
def norm_path(p: str, cwd: str | None = None) -> str:
"""Normalise relative, `G:\\…`, `G:/…`, UNC and Git Bash `/g/…` forms to an absolute path
with backslashes and an upper-case drive letter. A Git Bash rooted path that is not a
drive form (`/tmp/x`) goes through `cygpath -w` (M13); a native rooted path (`\\x`) keeps
its Windows meaning (the cwd's drive). Refused: empty, the WSL form `/mnt/<d>/…`, and
anything that does not end up absolute with a drive (`C:`, `C:x`); digest §3.4."""
if not isinstance(p, str) or not p.strip():
raise GremlinsError("empty path")
raise GremlinsError("empty path", path=p)
p = p.strip()
if not windows:
base = cwd if cwd is not None else os.getcwd()
return posixpath.normpath(posixpath.join(base, p))
m = _GITBASH_DRIVE.match(p)
if WindowsPlatform._WSL_MOUNT.match(p):
raise GremlinsError("WSL path form on Windows; pass C:\\... or /c/...", path=p)
m = WindowsPlatform._GITBASH_DRIVE.match(p)
if m:
p = m.group(1).upper() + ":" + (p[2:] or "/")
elif p.startswith("/") and not p.startswith("//"):
p = cygpath_w(p)
p = WindowsPlatform.cygpath_w(p)
p = p.replace("/", "\\")
drive, rest = ntpath.splitdrive(p)
if not (drive and rest.startswith("\\")):
base = norm_path(cwd if cwd is not None else os.getcwd(), windows=True)
base = WindowsPlatform.norm_path(cwd if cwd is not None else os.getcwd())
if not drive and rest.startswith("\\"):
p = ntpath.splitdrive(base)[0] + rest # rooted, no drive: cwd's drive
else:
@@ -339,6 +450,294 @@ def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> st
p = drive.upper() + rest
return p
# --- launching ---
def popen_kwargs(self, new_group: bool = False) -> dict:
"""No console window for any child. `new_group` has no Windows meaning here (the tree
kill walks the whole tree)."""
return {"creationflags": self.CREATE_NO_WINDOW}
@staticmethod
def _powershell() -> str:
w = shutil.which("powershell")
if w:
return w
return os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "WindowsPowerShell",
"v1.0", "powershell.exe")
def detach(self, argv: list[str], cwd: str, log_path: str) -> dict:
"""CIM `Win32_Process.Create` with a hidden window, through a `powershell` that itself has
CREATE_NO_WINDOW. `log_path` is unused: the supervisor redirects itself."""
env = dict(os.environ)
env["SPIRE_GREMLIN_DETACH_CMD"] = subprocess.list2cmdline(argv)
env["SPIRE_GREMLIN_DETACH_CWD"] = cwd
try:
r = _run([self._powershell(), "-NoProfile", "-NonInteractive", "-Command", self.POWERSHELL_CIM],
env=env, timeout=120)
except (OSError, subprocess.SubprocessError) as e:
return {"ok": False, "method": self.detach_method, "detail": "powershell: %s" % e}
toks = r.stdout.decode("utf-8", "replace").split()
try:
rv, pid = int(toks[-2]), int(toks[-1])
except (ValueError, IndexError):
return {"ok": False, "method": self.detach_method,
"detail": "unparsable CIM output: %r / %r" % (r.stdout[-200:], r.stderr[-400:])}
if rv != 0:
return {"ok": False, "method": self.detach_method, "detail": "Win32_Process.Create returned %d" % rv}
return {"ok": True, "method": self.detach_method, "pid": pid, "ctime": self.proc_ctime(pid)}
# --- processes ---
def proc_ctime(self, pid: int) -> int | None:
"""Process creation time in FILETIME ticks, or None."""
ctypes, wt, k32, _, _ = self._win()
h = k32.OpenProcess(self._PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid))
if not h:
return None
try:
c, e, k, u = wt.FILETIME(), wt.FILETIME(), wt.FILETIME(), wt.FILETIME()
if not k32.GetProcessTimes(h, ctypes.byref(c), ctypes.byref(e), ctypes.byref(k),
ctypes.byref(u)):
return None
return (c.dwHighDateTime << 32) | c.dwLowDateTime
finally:
k32.CloseHandle(h)
def epoch_to_ctime(self, t: float | None) -> int | None:
if t is None:
return None
return int(float(t) * 10_000_000) + self._FILETIME_EPOCH
def _running(self, pid: int) -> bool:
ctypes, wt, k32, _, _ = self._win()
h = k32.OpenProcess(self._PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
if not h:
# Exists but not openable: alive, creation time unverifiable.
return ctypes.get_last_error() == self._ERROR_ACCESS_DENIED
try:
code = wt.DWORD()
return bool(k32.GetExitCodeProcess(h, ctypes.byref(code))) and code.value == self._STILL_ACTIVE
finally:
k32.CloseHandle(h)
def process_table(self) -> list[tuple[int, int, str]]:
"""One Toolhelp32 snapshot (fast, no PowerShell)."""
ctypes, _, k32, PE32, invalid = self._win()
out = []
snap = k32.CreateToolhelp32Snapshot(self._TH32CS_SNAPPROCESS, 0)
if not snap or snap == invalid:
return out
try:
e = PE32()
e.dwSize = ctypes.sizeof(PE32)
ok = k32.Process32FirstW(snap, ctypes.byref(e))
while ok:
out.append((int(e.th32ProcessID), int(e.th32ParentProcessID), e.szExeFile))
ok = k32.Process32NextW(snap, ctypes.byref(e))
finally:
k32.CloseHandle(snap)
return out
def terminate(self, pid: int) -> bool:
_, _, k32, _, _ = self._win()
h = k32.OpenProcess(self._PROCESS_TERMINATE | self._PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid))
if not h:
return False
try:
return bool(k32.TerminateProcess(h, 1))
finally:
k32.CloseHandle(h)
@staticmethod
def _taskkill() -> str:
p = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "taskkill.exe")
return p if os.path.isfile(p) else "taskkill"
def tree_kill(self, pid: int) -> int | None:
"""`taskkill /T /F` on the tree; its exit code, or None when it could not run."""
try:
return _run([self._taskkill(), "/T", "/F", "/PID", str(int(pid))], timeout=60).returncode
except (OSError, subprocess.SubprocessError) as e:
_diag("taskkill failed: %s" % e)
return None
# --- files ---
def restore_no_replace(self, src: str, dst: str) -> bool:
try:
_retry(lambda: os.rename(src, dst)) # Windows rename never overwrites
return True
except OSError:
return False
# --- desktop ---
def visible_windows(self) -> list[tuple[int, int, str, str]] | None:
"""Visible top-level windows: [(hwnd, pid, class, title)] (the M8 window check)."""
ctypes, wt, _, _, _ = self._win()
u32 = ctypes.WinDLL("user32", use_last_error=True)
out = []
proto = ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPARAM)
def cb(hwnd, _):
if u32.IsWindowVisible(hwnd):
pid = wt.DWORD()
u32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
cls = ctypes.create_unicode_buffer(256)
u32.GetClassNameW(hwnd, cls, 256)
ttl = ctypes.create_unicode_buffer(256)
u32.GetWindowTextW(hwnd, ttl, 256)
out.append((int(hwnd or 0), pid.value, cls.value, ttl.value))
return True
u32.EnumWindows(proto(cb), 0)
return out
class PosixPlatform(Platform):
"""Linux, WSL (it is Linux) and macOS. Untested in this release: the gate is closed."""
name = "posix"
detach_method = "setsid"
# Lifting the gate (digest §4): add the tested `sys.platform` values here, with a test.
TESTED_SYS_PLATFORMS: tuple[str, ...] = ()
_DRIVE_FORM = re.compile(r"^[A-Za-z]:")
@property
def supported(self) -> bool:
return sys.platform in self.TESTED_SYS_PLATFORMS
def is_wsl(self) -> bool:
if os.environ.get("WSL_DISTRO_NAME"):
return True
try:
with open("/proc/sys/kernel/osrelease", "rb") as f:
return b"microsoft" in f.read().lower()
except OSError:
return False
@staticmethod
def norm_path(p: str, cwd: str | None = None) -> str:
"""`os.path.abspath` semantics relative to `cwd`. Refused: empty, a drive form (`C:…`)
and any backslash (on POSIX almost always a Windows path); digest §3.4."""
if not isinstance(p, str) or not p.strip():
raise GremlinsError("empty path", path=p)
orig, p = p, p.strip()
if PosixPlatform._DRIVE_FORM.match(p):
raise GremlinsError("Windows drive path form on POSIX; pass a POSIX path", path=orig)
if "\\" in p:
raise GremlinsError("backslash in a path on POSIX (a Windows path form); pass a POSIX path",
path=orig)
base = cwd if cwd is not None else os.getcwd()
return posixpath.normpath(posixpath.join(base, p))
def popen_kwargs(self, new_group: bool = False) -> dict:
return {"start_new_session": bool(new_group)}
def detach(self, argv: list[str], cwd: str, log_path: str) -> dict:
"""A new session, stdout/stderr to `log_path`."""
log = open(log_path, "ab")
try:
p = subprocess.Popen(argv, cwd=cwd, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
start_new_session=True, close_fds=True)
except OSError as e:
return {"ok": False, "method": self.detach_method, "detail": str(e)}
finally:
log.close()
return {"ok": True, "method": self.detach_method, "pid": p.pid, "ctime": self.proc_ctime(p.pid)}
def proc_ctime(self, pid: int) -> int | None:
"""/proc start time (clock ticks since boot), or None without /proc."""
try:
with open("/proc/%d/stat" % int(pid), "rb") as f:
stat = f.read().decode("ascii", "replace")
return int(stat[stat.rindex(")") + 2:].split()[19])
except (OSError, ValueError, IndexError):
return None
def epoch_to_ctime(self, t: float | None) -> int | None:
"""Epoch seconds in clock ticks since boot (/proc/stat btime), or None without /proc."""
if t is None:
return None
try:
with open("/proc/stat", "rb") as f:
btime = next(int(ln.split()[1]) for ln in f.read().splitlines() if ln.startswith(b"btime "))
return int((float(t) - btime) * os.sysconf("SC_CLK_TCK"))
except (OSError, ValueError, StopIteration, IndexError, AttributeError):
return None
def _running(self, pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
pass
return True
def process_table(self) -> list[tuple[int, int, str]]:
"""A /proc walk; [] without /proc (macOS: the known gap)."""
out = []
try:
names = os.listdir("/proc")
except OSError:
return out
for n in names:
if not n.isdigit():
continue
try:
with open("/proc/%s/stat" % n, "rb") as f:
st = f.read().decode("ascii", "replace")
comm = st[st.index("(") + 1: st.rindex(")")]
ppid = int(st[st.rindex(")") + 2:].split()[1])
out.append((int(n), ppid, comm))
except (OSError, ValueError, IndexError):
continue
return out
def terminate(self, pid: int) -> bool:
try:
os.kill(int(pid), signal.SIGKILL)
return True
except OSError:
return False
def tree_kill(self, pid: int) -> int | None:
"""SIGKILL to the process group (M14: `claude` runs in its own session and is missed;
the per-pid sweep catches it)."""
with contextlib.suppress(OSError):
os.killpg(int(pid), signal.SIGKILL)
return None
def restore_no_replace(self, src: str, dst: str) -> bool:
try:
os.link(src, dst) # fails if the path exists
os.remove(src)
return True
except OSError:
return False
def select_platform() -> Platform:
return WindowsPlatform() if os.name == "nt" else PosixPlatform()
PLATFORM = select_platform()
# --- platform seam: end ---------------------------------------------------------------
# --------------------------------------------------------------------------------------
# Paths
# --------------------------------------------------------------------------------------
def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str:
"""Normalise a path for this host (`PLATFORM.norm_path`). `windows` selects a platform's
rules explicitly (True: Windows, False: POSIX); the tests use it to check both anywhere."""
if windows is None:
return PLATFORM.norm_path(p, cwd)
return (WindowsPlatform if windows else PosixPlatform).norm_path(p, cwd)
def state_dir_for(run: str) -> str:
"""`reports/<x>-run.md` -> `reports/<x>-run-gremlins` (a sibling of the run log)."""
@@ -501,15 +900,7 @@ class RegistryLock:
def _restore(self, aside: str) -> bool:
"""Put a lock renamed aside back, but never over a lock that appeared meanwhile."""
try:
if IS_WINDOWS:
_retry(lambda: os.rename(aside, self.path)) # Windows rename never overwrites
else:
os.link(aside, self.path) # fails if the path exists
os.remove(aside)
return True
except OSError:
return False
return PLATFORM.restore_no_replace(aside, self.path)
def _break(self, raw: bytes) -> bool:
"""Remove the lock atomically if it still is `raw` (the content judged stale)."""
@@ -632,81 +1023,18 @@ def ledger_append(state: str, line: dict) -> bool:
# Processes: pid + creation time (A19), subprocess helpers, git (C7)
# --------------------------------------------------------------------------------------
if IS_WINDOWS:
import ctypes
import ctypes.wintypes as _wt
_k32 = ctypes.WinDLL("kernel32", use_last_error=True)
_k32.OpenProcess.argtypes = [_wt.DWORD, _wt.BOOL, _wt.DWORD]
_k32.OpenProcess.restype = _wt.HANDLE
_k32.GetProcessTimes.argtypes = [_wt.HANDLE] + [ctypes.POINTER(_wt.FILETIME)] * 4
_k32.GetProcessTimes.restype = _wt.BOOL
_k32.GetExitCodeProcess.argtypes = [_wt.HANDLE, ctypes.POINTER(_wt.DWORD)]
_k32.GetExitCodeProcess.restype = _wt.BOOL
_k32.CloseHandle.argtypes = [_wt.HANDLE]
_k32.CloseHandle.restype = _wt.BOOL
_PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
_PROCESS_TERMINATE = 0x0001
_STILL_ACTIVE = 259
_ERROR_ACCESS_DENIED = 5
_TH32CS_SNAPPROCESS = 0x00000002
_INVALID_HANDLE = _wt.HANDLE(-1).value
class _PE32(ctypes.Structure):
_fields_ = [("dwSize", _wt.DWORD), ("cntUsage", _wt.DWORD), ("th32ProcessID", _wt.DWORD),
("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", _wt.DWORD),
("cntThreads", _wt.DWORD), ("th32ParentProcessID", _wt.DWORD),
("pcPriClassBase", _wt.LONG), ("dwFlags", _wt.DWORD),
("szExeFile", _wt.WCHAR * 260)]
_k32.CreateToolhelp32Snapshot.argtypes = [_wt.DWORD, _wt.DWORD]
_k32.CreateToolhelp32Snapshot.restype = _wt.HANDLE
_k32.Process32FirstW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)]
_k32.Process32FirstW.restype = _wt.BOOL
_k32.Process32NextW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)]
_k32.Process32NextW.restype = _wt.BOOL
_k32.TerminateProcess.argtypes = [_wt.HANDLE, _wt.UINT]
_k32.TerminateProcess.restype = _wt.BOOL
# The process primitives are the platform's (`PLATFORM`); these module-level names stay so that
# neutral code and the tests' patches (`mock.patch.object(gremlins, "proc_ctime", …)`) keep working.
def proc_ctime(pid: int) -> int | None:
"""Process creation time: Windows FILETIME ticks; POSIX /proc start time; else None."""
if IS_WINDOWS:
h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid))
if not h:
return None
try:
c, e, k, u = _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME()
if not _k32.GetProcessTimes(h, ctypes.byref(c), ctypes.byref(e), ctypes.byref(k),
ctypes.byref(u)):
return None
return (c.dwHighDateTime << 32) | c.dwLowDateTime
finally:
_k32.CloseHandle(h)
try:
with open("/proc/%d/stat" % int(pid), "rb") as f:
stat = f.read().decode("ascii", "replace")
return int(stat[stat.rindex(")") + 2:].split()[19])
except (OSError, ValueError, IndexError):
return None
_FILETIME_EPOCH = 116444736000000000 # 1970-01-01 in FILETIME ticks (100 ns since 1601)
"""Process creation time in the platform's units, or None when unknown."""
return PLATFORM.proc_ctime(pid)
def epoch_to_ctime(t: float | None) -> int | None:
"""Epoch seconds in the units of `proc_ctime` (Windows FILETIME ticks; POSIX clock ticks since
boot), or None when the conversion is unknown here."""
if t is None:
return None
if IS_WINDOWS:
return int(float(t) * 10_000_000) + _FILETIME_EPOCH
try:
with open("/proc/stat", "rb") as f:
btime = next(int(ln.split()[1]) for ln in f.read().splitlines() if ln.startswith(b"btime "))
return int((float(t) - btime) * os.sysconf("SC_CLK_TCK"))
except (OSError, ValueError, StopIteration, IndexError, AttributeError):
return None
"""Epoch seconds in the units of `proc_ctime`, or None when the conversion is unknown here."""
return PLATFORM.epoch_to_ctime(t)
def pid_record(pid: int) -> dict:
@@ -716,32 +1044,7 @@ def pid_record(pid: int) -> dict:
def pid_alive(pid: int, ctime: int | None = None) -> bool:
"""True if `pid` runs and (when both are known) its creation time equals `ctime`."""
if pid is None or int(pid) <= 0:
return False
pid = int(pid)
if IS_WINDOWS:
h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
if not h:
# Exists but not openable: alive, creation time unverifiable.
return ctypes.get_last_error() == _ERROR_ACCESS_DENIED
try:
code = _wt.DWORD()
if not _k32.GetExitCodeProcess(h, ctypes.byref(code)) or code.value != _STILL_ACTIVE:
return False
finally:
_k32.CloseHandle(h)
else:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
pass
if ctime is not None:
now_ct = proc_ctime(pid)
if now_ct is not None and now_ct != ctime:
return False
return True
return PLATFORM.pid_alive(pid, ctime)
def record_alive(rec: dict | None) -> bool:
@@ -749,38 +1052,8 @@ def record_alive(rec: dict | None) -> bool:
def process_table() -> list[tuple[int, int, str]]:
"""[(pid, ppid, name)] of every process. Windows: one Toolhelp32 snapshot (fast, no PowerShell)."""
out = []
if IS_WINDOWS:
snap = _k32.CreateToolhelp32Snapshot(_TH32CS_SNAPPROCESS, 0)
if not snap or snap == _INVALID_HANDLE:
return out
try:
e = _PE32()
e.dwSize = ctypes.sizeof(_PE32)
ok = _k32.Process32FirstW(snap, ctypes.byref(e))
while ok:
out.append((int(e.th32ProcessID), int(e.th32ParentProcessID), e.szExeFile))
ok = _k32.Process32NextW(snap, ctypes.byref(e))
finally:
_k32.CloseHandle(snap)
return out
try:
names = os.listdir("/proc")
except OSError:
return out
for n in names:
if not n.isdigit():
continue
try:
with open("/proc/%s/stat" % n, "rb") as f:
st = f.read().decode("ascii", "replace")
comm = st[st.index("(") + 1: st.rindex(")")]
ppid = int(st[st.rindex(")") + 2:].split()[1])
out.append((int(n), ppid, comm))
except (OSError, ValueError, IndexError):
continue
return out
"""[(pid, ppid, name)] of every process ([] where the platform cannot list them)."""
return PLATFORM.process_table()
ALIVE_UNTIL_SLACK_S = 1.0 # job-file times are ISO with whole seconds: allow the truncated part
@@ -842,32 +1115,14 @@ def terminate(rec: dict) -> bool:
"""Kill one process identified by pid + creation time. True if it is gone or was signalled."""
if not record_alive(rec):
return True
pid = int(rec["pid"])
if IS_WINDOWS:
h = _k32.OpenProcess(_PROCESS_TERMINATE | _PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
if not h:
return False
try:
return bool(_k32.TerminateProcess(h, 1))
finally:
_k32.CloseHandle(h)
try:
os.kill(pid, signal.SIGKILL)
return True
except OSError:
return False
def _taskkill() -> str:
p = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "taskkill.exe")
return p if os.path.isfile(p) else "taskkill"
return PLATFORM.terminate(int(rec["pid"]))
def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> dict:
"""Snapshot the tree of `rec` (plus the trees of `extra` records), kill it (`taskkill /T /F` on
Windows, the process group on POSIX), sweep every snapshot pid still alive, and verify.
"""Snapshot the tree of `rec` (plus the trees of `extra` records), kill it (`PLATFORM.tree_kill`
of the first live root), sweep every snapshot pid still alive, and verify.
`alive_until` bounds the descendants of a dead root (see `process_tree`).
Returns {"snapshot", "taskkill_rc", "survivors"}; pids are always checked with their ctime."""
Returns {"snapshot", "tree_kill_rc", "survivors"}; pids are always checked with their ctime."""
roots = [r for r in [rec, *extra] if r and r.get("pid")]
snap, seen = [], set()
for r in roots:
@@ -880,14 +1135,7 @@ def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> d
for r in roots:
if not record_alive(r):
continue
if IS_WINDOWS:
try:
rc = _run([_taskkill(), "/T", "/F", "/PID", str(int(r["pid"]))], timeout=60).returncode
except (OSError, subprocess.SubprocessError) as e:
_diag("taskkill failed: %s" % e)
else:
with contextlib.suppress(OSError):
os.killpg(int(r["pid"]), signal.SIGKILL)
rc = PLATFORM.tree_kill(int(r["pid"]))
break # one tree kill of the first live root; the sweep covers the rest
deadline = time.monotonic() + KILL_VERIFY_S
while True:
@@ -897,13 +1145,13 @@ def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> d
for p in survivors:
terminate(p)
time.sleep(0.1)
return {"snapshot": snap, "taskkill_rc": rc, "survivors": survivors}
return {"snapshot": snap, "tree_kill_rc": rc, "survivors": survivors}
def _run(args, cwd=None, timeout=60, env=None) -> subprocess.CompletedProcess:
"""Run a short helper process: stdout captured (never leaks to ours), no window."""
return subprocess.run(args, cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True,
timeout=timeout, creationflags=_NOWIN)
timeout=timeout, **PLATFORM.popen_kwargs())
def git(args, cwd=None, timeout=60) -> subprocess.CompletedProcess:
@@ -1423,46 +1671,11 @@ def render_template(name: str, values: dict) -> str:
return string.Template(text).safe_substitute(values)
def _powershell() -> str:
w = shutil.which("powershell")
if w:
return w
return os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "WindowsPowerShell",
"v1.0", "powershell.exe")
def detach_supervisor(state: str, run: str, job: str) -> dict:
"""Start `_supervise` detached (C6). Windows: CIM `Win32_Process.Create` with a hidden window,
through a `powershell` that itself has CREATE_NO_WINDOW. POSIX: a new session.
Returns {"ok", "pid", "ctime", "method"} or {"ok": False, "detail"}."""
"""Start `_supervise` detached (C6) through `PLATFORM.detach`, in the state dir.
Returns {"ok", "pid", "ctime", "method"} or {"ok": False, "method", "detail"}."""
argv = [supervisor_interpreter(), "-I", script_path(), "_supervise", "--run", run, "--job", job]
if IS_WINDOWS:
env = dict(os.environ)
env["SPIRE_GREMLIN_DETACH_CMD"] = subprocess.list2cmdline(argv)
env["SPIRE_GREMLIN_DETACH_CWD"] = state
try:
r = _run([_powershell(), "-NoProfile", "-NonInteractive", "-Command", POWERSHELL_CIM],
env=env, timeout=120)
except (OSError, subprocess.SubprocessError) as e:
return {"ok": False, "method": DETACH_METHOD, "detail": "powershell: %s" % e}
toks = r.stdout.decode("utf-8", "replace").split()
try:
rv, pid = int(toks[-2]), int(toks[-1])
except (ValueError, IndexError):
return {"ok": False, "method": DETACH_METHOD,
"detail": "unparsable CIM output: %r / %r" % (r.stdout[-200:], r.stderr[-400:])}
if rv != 0:
return {"ok": False, "method": DETACH_METHOD, "detail": "Win32_Process.Create returned %d" % rv}
return {"ok": True, "method": DETACH_METHOD, "pid": pid, "ctime": proc_ctime(pid)}
log = open(raw_file(state, job, ".sup.log"), "ab")
try:
p = subprocess.Popen(argv, cwd=state, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
start_new_session=True, close_fds=True)
except OSError as e:
return {"ok": False, "method": DETACH_METHOD, "detail": str(e)}
finally:
log.close()
return {"ok": True, "method": DETACH_METHOD, "pid": p.pid, "ctime": proc_ctime(p.pid)}
return PLATFORM.detach(argv, cwd=state, log_path=raw_file(state, job, ".sup.log"))
def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: str, flags, reason,
@@ -1759,7 +1972,7 @@ def reap_lost(state: str) -> list[str]:
def kill_job(state: str, job: str, by: str) -> dict:
"""`kill` (by="kill") and `expire` (by="expire"): snapshot the supervisor's tree, taskkill it,
"""`kill` (by="kill") and `expire` (by="expire"): snapshot the supervisor's tree, tree-kill it,
sweep, verify, finalise as `killed`."""
with RegistryLock(state):
reg = read_registry(state)
@@ -1788,7 +2001,7 @@ def kill_job(state: str, job: str, by: str) -> dict:
write_json_atomic(job_file(state, job), rt)
final = read_registry(state)["jobs"][job].get("status")
return {"job": job, "killed": True, "status": final, "ok": not k["survivors"],
"snapshot": k["snapshot"], "survivors": k["survivors"], "taskkill_rc": k["taskkill_rc"]}
"snapshot": k["snapshot"], "survivors": k["survivors"], "tree_kill_rc": k["tree_kill_rc"]}
# --------------------------------------------------------------------------------------
@@ -1871,7 +2084,7 @@ def build_parser() -> _Parser:
sp.add_argument("--run", required=True, help="path of the run log .md")
return sp
w = cmd("where", "Preflight report: script, python, uv, binary, git, spawn method, env scrub.")
w = cmd("where", "Preflight report: script, python, uv, binary, git, env scrub, platform.")
w.add_argument("--binary", help="claude binary to report instead of the default resolution")
# Minor 8: every init flag defaults to None. On the first init an omitted flag takes its
@@ -2032,9 +2245,8 @@ def cmd_where(args) -> tuple[dict, int]:
"uv": _uv_report(),
"binary": binary,
"git": _git_report(),
"spawn": DETACH_METHOD,
"scrub": scrub_report(),
"platform": sys.platform,
"platform": PLATFORM.describe(),
"warnings": warnings,
"ok": not reasons,
}
@@ -2639,13 +2851,13 @@ def _new_entry(job: str, kind: str, **kw) -> dict:
def _start_job(state: str, run: str, job: str, prompt: str, launch: dict) -> dict:
"""Write the brief and the child env, spawn the supervisor, record it. A failed spawn ends the
job as `failed`, reason `spawn_failed`."""
"""Write the brief and the child env, detach the supervisor, record it. A failed detach ends
the job as `failed`, reason `detach_failed`."""
atomic_write_bytes(os.path.join(state, "briefs", job + ".md"), prompt.encode("utf-8"))
write_json_atomic(raw_file(state, job, ".env.json"), scrubbed_env(launch["binary"]["path"]))
sp = detach_supervisor(state, run, job)
if not sp.get("ok"):
finalize_job(state, job, ss=None, hint="failed", reason="spawn_failed", detail=sp.get("detail"))
finalize_job(state, job, ss=None, hint="failed", reason="detach_failed", detail=sp.get("detail"))
return sp
with RegistryLock(state):
reg = read_registry(state)
@@ -3219,7 +3431,7 @@ class Supervisor:
finalize_job(self.state, self.job, ss=None, hint="failed", reason="supervisor_error",
detail="missing env file")
return {"error": "missing env file"}, EXIT_USAGE
os.environ.clear() # git and taskkill lookups use the child's PATH
os.environ.clear() # git and tree-kill lookups use the child's PATH
os.environ.update(self.env)
self.write_rt()
with open(os.path.join(self.state, "briefs", self.job + ".md"), "rb") as f:
@@ -3229,8 +3441,8 @@ class Supervisor:
err = open(raw_file(self.state, self.job, ".err"), "wb")
try:
p = subprocess.Popen(e["argv"], cwd=e["cwd"], env=self.env, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=err, creationflags=_NOWIN,
start_new_session=not IS_WINDOWS)
stdout=subprocess.PIPE, stderr=err,
**PLATFORM.popen_kwargs(new_group=True))
except OSError as ex:
raw.close()
err.close()
plugins/spire/skills/gremlins/tests/test_gremlins.py
+125 -36
View File
@@ -37,7 +37,8 @@ def _load_gremlins():
gremlins = _load_gremlins()
IS_WINDOWS = os.name == "nt"
IS_WINDOWS = os.name == "nt" # test-only: generated wrappers (.cmd vs sh) and the module-end sweep
WIN_PLATFORM = gremlins.PLATFORM.name == "windows" # platform behaviour under test (digest §3.5)
STUB_VERSION = "2.1.999"
@@ -134,7 +135,7 @@ class PathTests(Base):
"""M13: a Git Bash rooted path that is not a drive form is refused (exit 2) when cygpath is
not on PATH; it is never mapped onto the cwd's drive. Drive forms need no cygpath."""
cwd = "G:\\dev\\proj"
with mock.patch.object(gremlins, "find_cygpath", lambda: None):
with mock.patch.object(gremlins.WindowsPlatform, "find_cygpath", lambda: None):
for src in ("/tmp/x", "/usr/share/doc", "/tmp/x-run.md"):
with self.subTest(src=src):
with self.assertRaises(gremlins.GremlinsError) as cm:
@@ -145,7 +146,7 @@ class PathTests(Base):
self.assertEqual(gremlins.norm_path("/g/x/y.md", cwd=cwd, windows=True), "G:\\x\\y.md")
self.assertEqual(gremlins.norm_path("/c", cwd=cwd, windows=True), "C:\\")
self.assertEqual(gremlins.norm_path("\\rooted\\a", cwd=cwd, windows=True), "G:\\rooted\\a")
if IS_WINDOWS: # end to end: --run /tmp/... is refused, nothing written on the cwd's drive
if WIN_PLATFORM: # end to end: --run /tmp/... is refused, nothing written on the cwd's drive
drive = os.path.splitdrive(os.getcwd())[0]
code, out, raw = call(["init", "--run", "/tmp/gremlins-m13-x-run.md", "--model",
"claude-opus-5-5", "--cwd", self.tmp])
@@ -158,19 +159,19 @@ class PathTests(Base):
"""M13: with cygpath on PATH the path is converted by `cygpath -w`; a failing cygpath
refuses."""
exe, log = self.fake_cygpath()
with mock.patch.object(gremlins, "find_cygpath", lambda: exe):
with mock.patch.object(gremlins.WindowsPlatform, "find_cygpath", lambda: exe):
self.assertEqual(gremlins.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True),
"C:\\Fake\\tmp\\x")
self.assertEqual(gremlins.norm_path("/g/x", cwd="G:\\dev\\proj", windows=True), "G:\\x")
with open(log) as f:
self.assertEqual(f.read().splitlines(), ["-w /tmp/x"]) # drive forms skip cygpath
bad, _ = self.fake_cygpath(code=1)
with mock.patch.object(gremlins, "find_cygpath", lambda: bad):
with mock.patch.object(gremlins.WindowsPlatform, "find_cygpath", lambda: bad):
with self.assertRaises(gremlins.GremlinsError) as cm:
gremlins.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True)
self.assertEqual(cm.exception.payload["error"], "cygpath failed")
real = shutil.which("cygpath")
if IS_WINDOWS and real: # the real one (Git Bash): never the cwd's drive + \tmp
if WIN_PLATFORM and real: # the real one (Git Bash): never the cwd's drive + \tmp
got = gremlins.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True)
self.assertNotEqual(got, "G:\\tmp\\x")
self.assertRegex(got, r"^[A-Z]:\\.+\\x$")
@@ -192,7 +193,7 @@ class PathTests(Base):
with self.assertRaises(gremlins.GremlinsError):
gremlins.norm_path(" ")
@unittest.skipUnless(IS_WINDOWS, "Windows path forms")
@unittest.skipUnless(WIN_PLATFORM, "Windows path forms")
def test_state_dir_windows_forms(self):
want = "G:\\w\\reports\\2026-10-16-08-x-run-gremlins"
for src in ("G:\\w\\reports\\2026-10-16-08-x-run.md", "G:/w/reports/2026-10-16-08-x-run.md",
@@ -211,6 +212,108 @@ class PathTests(Base):
self.assertEqual(cm.exception.code, 2)
# --------------------------------------------------------------------------------------
class PathMatrixTests(Base):
"""Digest §3.4 (amendment A5): every path form x both platform classes, on any host. Pure: the
cygpath lookup and its one call are faked, nothing touches the filesystem."""
WIN_CWD = "G:\\dev\\proj"
POSIX_CWD = "/home/u/proj"
# (row, input, Windows result, POSIX result). A result is the normalised path, or
# ("refused", <word the error names>).
ROWS = (
(1, "C:\\x\\y.md", "C:\\x\\y.md", ("refused", "drive")),
(2, "c:/x/y.md", "C:\\x\\y.md", ("refused", "drive")),
(3, "C:", ("refused", "cannot normalise"), ("refused", "drive")),
(3, "C:x", ("refused", "cannot normalise"), ("refused", "drive")),
(4, "/c/x/y.md", "C:\\x\\y.md", "/c/x/y.md"),
(5, "/c", "C:\\", "/c"),
(6, "/tmp/x", "C:\\Fake\\tmp\\x", "/tmp/x"),
(7, "\\\\srv\\share\\x\\y.md", "\\\\srv\\share\\x\\y.md", ("refused", "backslash")),
(8, "//srv/share/x/y.md", "\\\\srv\\share\\x\\y.md", "//srv/share/x/y.md"),
(9, "/mnt/c/x/y.md", ("refused", "WSL"), "/mnt/c/x/y.md"),
(10, "docs/a.md", "G:\\dev\\proj\\docs\\a.md", "/home/u/proj/docs/a.md"),
(11, "docs\\a.md", "G:\\dev\\proj\\docs\\a.md", ("refused", "backslash")),
(12, "\\rooted\\a", "G:\\rooted\\a", ("refused", "backslash")),
(13, "../z", "G:\\dev\\z", "/home/u/z"),
(14, " G:/x/y.md ", "G:\\x\\y.md", ("refused", "drive")),
(15, "", ("refused", "empty"), ("refused", "empty")),
(15, " ", ("refused", "empty"), ("refused", "empty")),
)
FAKE_CYGPATH = "fake-cygpath"
def setUp(self):
super().setUp()
self.cyg_calls = []
def fake_run(args, cwd=None, timeout=60, env=None):
self.assertEqual(list(args[:2]), [self.FAKE_CYGPATH, "-w"]) # the only process call
self.cyg_calls.append(args[2])
out = "C:\\Fake" + args[2].replace("/", "\\") + "\r\n"
return subprocess.CompletedProcess(args, 0, stdout=out.encode(), stderr=b"")
for target, name, fn in ((gremlins.WindowsPlatform, "find_cygpath", lambda: self.FAKE_CYGPATH),
(gremlins, "_run", fake_run)):
p = mock.patch.object(target, name, fn)
p.start()
self.addCleanup(p.stop)
def check(self, fn, cwd, src, want):
if isinstance(want, tuple):
with self.assertRaises(gremlins.GremlinsError) as cm:
fn(src, cwd)
self.assertEqual(cm.exception.code, 2)
self.assertIn(want[1], cm.exception.payload["error"])
self.assertEqual(cm.exception.payload["path"], src)
else:
self.assertEqual(fn(src, cwd), want)
def test_windows_rows(self):
for row, src, want, _ in self.ROWS:
with self.subTest(row=row, src=src):
self.check(gremlins.WindowsPlatform.norm_path, self.WIN_CWD, src, want)
self.assertEqual(self.cyg_calls, ["/tmp/x"]) # row 6 only; drive forms skip cygpath
def test_posix_rows(self):
for row, src, _, want in self.ROWS:
with self.subTest(row=row, src=src):
self.check(gremlins.PosixPlatform.norm_path, self.POSIX_CWD, src, want)
self.assertEqual(self.cyg_calls, [])
def test_row6_without_cygpath(self):
"""M13: without cygpath a Git Bash rooted non-drive path is refused on Windows."""
with mock.patch.object(gremlins.WindowsPlatform, "find_cygpath", lambda: None):
self.check(gremlins.WindowsPlatform.norm_path, self.WIN_CWD, "/tmp/x", ("refused", "cygpath"))
self.assertEqual(self.cyg_calls, [])
def test_delegate_selects_the_class(self):
"""The module-level `norm_path(windows=…)` is a selector over the two classes."""
self.assertEqual(gremlins.norm_path("../z", cwd=self.WIN_CWD, windows=True), "G:\\dev\\z")
self.assertEqual(gremlins.norm_path("../z", cwd=self.POSIX_CWD, windows=False), "/home/u/z")
with self.assertRaises(gremlins.GremlinsError):
gremlins.norm_path("C:\\x", cwd=self.POSIX_CWD, windows=False)
def test_both_classes_build_anywhere(self):
"""No ctypes at construction: both classes and their pure parts work on any host."""
w, p = gremlins.WindowsPlatform(), gremlins.PosixPlatform()
self.assertIsNone(w._api)
self.assertEqual((w.name, w.detach_method), ("windows", "cim-hidden"))
self.assertEqual((p.name, p.detach_method), ("posix", "setsid"))
self.assertEqual(w.popen_kwargs(), {"creationflags": 0x08000000})
self.assertEqual(w.popen_kwargs(new_group=True), {"creationflags": 0x08000000})
self.assertEqual(p.popen_kwargs(), {"start_new_session": False})
self.assertEqual(p.popen_kwargs(new_group=True), {"start_new_session": True})
self.assertEqual(w.epoch_to_ctime(0), 116444736000000000)
self.assertIsNone(w.epoch_to_ctime(None))
self.assertIsNone(p.visible_windows())
for d, name in ((w.describe(), "windows"), (p.describe(), "posix")):
self.assertEqual(set(d), {"name", "sys_platform", "wsl", "supported", "detach"})
self.assertEqual((d["name"], d["sys_platform"]), (name, sys.platform))
self.assertIsNone(w._api)
# select_platform: the test's own os.name check is the reference here (IS_WINDOWS)
self.assertIs(type(gremlins.PLATFORM), gremlins.WindowsPlatform if IS_WINDOWS
else gremlins.PosixPlatform)
# --------------------------------------------------------------------------------------
class ScrubTests(Base):
def test_scrubbed_env(self):
@@ -571,7 +674,7 @@ class ProcessTreeTests(Base):
seen = []
with mock.patch.object(gremlins, "kill_tree", lambda rec, extra=(), alive_until=None:
seen.append((rec, alive_until)) or {"snapshot": [], "survivors": [],
"taskkill_rc": None}):
"tree_kill_rc": None}):
self.assertEqual(gremlins.reap_lost(st), ["j0001"])
self.assertEqual(seen, [(self.rec(), gremlins._ts(gremlins._iso(old)))])
@@ -1052,7 +1155,8 @@ class SubprocessTests(Base):
self.assertEqual(r.returncode, 0, r.stderr)
out = self.assert_one_json_line(r.stdout.decode("utf-8"))
self.assertTrue(out["ok"])
self.assertEqual(out["spawn"], "cim-hidden" if IS_WINDOWS else "setsid")
self.assertEqual(out["platform"]["detach"], "cim-hidden" if WIN_PLATFORM else "setsid")
self.assertNotIn("spawn", out)
self.assertEqual(out["binary"]["version"], STUB_VERSION)
self.assertEqual(out["binary"]["how"], "--binary")
self.assertEqual(os.path.normcase(out["script"]), os.path.normcase(os.path.abspath(GREMLINS_PY)))
@@ -1294,25 +1398,9 @@ def price(tokens: dict) -> float:
def visible_windows() -> list[tuple]:
"""Visible top-level windows: [(hwnd, pid, class, title)] (Windows only)."""
import ctypes
import ctypes.wintypes as wt
u32 = ctypes.WinDLL("user32", use_last_error=True)
out = []
proto = ctypes.WINFUNCTYPE(wt.BOOL, wt.HWND, wt.LPARAM)
def cb(hwnd, _):
if u32.IsWindowVisible(hwnd):
pid = wt.DWORD()
u32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
cls = ctypes.create_unicode_buffer(256)
u32.GetClassNameW(hwnd, cls, 256)
ttl = ctypes.create_unicode_buffer(256)
u32.GetWindowTextW(hwnd, ttl, 256)
out.append((int(hwnd or 0), pid.value, cls.value, ttl.value))
return True
u32.EnumWindows(proto(cb), 0)
return out
"""Visible top-level windows: [(hwnd, pid, class, title)] (`PLATFORM.visible_windows()`; []
where the platform has no such check)."""
return gremlins.PLATFORM.visible_windows() or []
# P0.1's detector: a visible console window (conhost) or Windows Terminal window. The latter belongs
@@ -2689,7 +2777,7 @@ class LaunchTests(LaunchBase):
self.assertIn(rt["claude"]["pid"], pids)
self.assertIn(stub["pid"], pids)
self.assertIn(stub["child_pid"], pids)
if IS_WINDOWS:
if WIN_PLATFORM:
# spawned through CIM: the supervisor's parent is the WMI provider host, not us
table = {pid: (ppid, name) for pid, ppid, name in gremlins.process_table()}
parent = table.get(table[sup["pid"]][0], (None, None))[1]
@@ -2818,19 +2906,20 @@ class LaunchTests(LaunchBase):
["wait", "--run", self.run_log, "--all", "j0001"]):
self.assertEqual(call(argv)[0], 2, argv)
def test_spawn_failed(self):
def test_detach_failed(self):
self.setup_run()
failed = {"ok": False, "method": gremlins.DETACH_METHOD, "detail": "Win32_Process.Create returned 9"}
failed = {"ok": False, "method": gremlins.PLATFORM.detach_method,
"detail": "Win32_Process.Create returned 9"}
with mock.patch.object(gremlins, "detach_supervisor", lambda *a: failed):
t = self.train("B1")
code, out = self.wait(t["job"], timeout=5)
self.assertEqual(code, 4, out)
r = out["done"][0]
self.assertEqual((r["status"], r["reason"]), ("failed", "spawn_failed"))
self.assertEqual((r["status"], r["reason"]), ("failed", "detach_failed"))
self.assertEqual(r["tokens"], {k: None for k in gremlins.TOKEN_KEYS})
self.assertEqual(self.reg()["bases"]["B1"]["status"], "failed")
self.assertEqual(self.reg()["jobs"][t["job"]]["reason_detail"], failed["detail"])
self.assertEqual([x["reason"] for x in self.ledger()], ["spawn_failed"])
self.assertEqual([x["reason"] for x in self.ledger()], ["detach_failed"])
self.assertEqual(stub_calls(self.stub), [])
def test_refusals(self):
@@ -2896,14 +2985,14 @@ class LaunchTests(LaunchBase):
# --- block 2.1 -----------------------------------------------------------------------
@unittest.skipUnless(IS_WINDOWS, "visible windows are a Windows check")
@unittest.skipUnless(WIN_PLATFORM, "visible windows are a Windows check")
def test_no_visible_window(self):
"""M8: while a job runs, no visible window belongs to its process tree, and no visible
console or Windows Terminal window appears anywhere on the desktop (a Terminal window
belongs to WindowsTerminal.exe, not to the tree). A positive control, a deliberately visible
console (minimised, not activated) closed at once, shows the check sees one. Skipped, not
failed, when there is no desktop."""
if not visible_windows():
failed, when there is no desktop (or the platform has no window check: None)."""
if not gremlins.PLATFORM.visible_windows():
self.skipTest("no desktop: window enumeration sees no visible window")
# positive control
before = {w[0] for w in console_windows()}