diff --git a/plugins/spire/skills/gremlins/gremlins.py b/plugins/spire/skills/gremlins/gremlins.py index 633cbc7..12a32ea 100644 --- a/plugins/spire/skills/gremlins/gremlins.py +++ b/plugins/spire/skills/gremlins/gremlins.py @@ -119,6 +119,14 @@ USER_SETTINGS_PATH: str | None = None PHASE2_COMMANDS = () # commands that tolerate unknown arguments (none: all are implemented) # N3: a gremlin may only run where, list, usage and wait (`tick` launches pings). CHILD_REFUSED = ("train", "spawn", "continue", "rebase", "kill", "expire", "init", "tick") +# The platform gate (digest §4, amendment A1): on an unsupported platform these launch nothing. +# `init` checks the gate itself, as its first step (B1); `where` reports it. +GATED_COMMANDS = ("train", "spawn", "continue", "rebase", "tick") +# Porting aid only (C12, documented in the platform handoff, not in SKILL.md): `=1` opens the +# gate for that call on an unsupported platform. +UNSUPPORTED_OK_ENV = "SPIRE_GREMLIN_UNSUPPORTED_OK" +GATE_OVERRIDE_WARNING = ("platform gate overridden (%s=1); this platform is untested, use it only " + "to port and test spire:gremlins" % UNSUPPORTED_OK_ENV) # Warmth and keep-warm (design 5, block 2.1). Module-level so tests can shorten them. TICK_SUP_S = 60.0 # a supervisor runs one tick pass this often @@ -726,6 +734,38 @@ PLATFORM = select_platform() # --- platform seam: end --------------------------------------------------------------- +# -------------------------------------------------------------------------------------- +# Platform gate (digest §4): fail closed where the helper is untested +# -------------------------------------------------------------------------------------- + + +def gate_overridden() -> bool: + """A12: the override counts only where the gate is closed (on a supported platform the env + var changes nothing).""" + return os.environ.get(UNSUPPORTED_OK_ENV) == "1" and not PLATFORM.supported + + +def platform_report() -> dict: + """`where`'s `platform` block: `PLATFORM.describe()` plus `override`.""" + return {**PLATFORM.describe(), "override": gate_overridden()} + + +def platform_gate() -> str | None: + """None when the platform is supported or the gate is overridden; else the reason.""" + if PLATFORM.supported or gate_overridden(): + return None + d = PLATFORM.describe() + return ("spire:gremlins is not tested on %s%s yet; use fresh agents for every block " + "(see SKILL.md, Platforms)." % (d.get("sys_platform"), ", WSL" if d.get("wsl") else "")) + + +def require_platform() -> None: + """Refuse (exit 2) on a closed gate: `init` and the launching commands (A1).""" + reason = platform_gate() + if reason is not None: + raise GremlinsError("platform not supported", reason=reason, platform=platform_report()) + + # -------------------------------------------------------------------------------------- # Paths # -------------------------------------------------------------------------------------- @@ -2216,29 +2256,45 @@ def build_parser() -> _Parser: def cmd_where(args) -> tuple[dict, int]: + """The preflight report. On a closed platform gate: `ok: false` with the gate reason first, + and every other part still filled (the porting hosts need it).""" reasons = [] + gate = platform_gate() + if gate is not None: + reasons.append(gate) py_ok = sys.version_info >= (3, 10) if not py_ok: reasons.append("python < 3.10") py = {"version": platform.python_version(), "executable": sys.executable, "base_executable": getattr(sys, "_base_executable", sys.executable), "supervisor_interpreter": supervisor_interpreter(), "ok": py_ok} - path, how = resolve_binary(args.binary) - binary = {"path": path, "how": how} - if path is None: - reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") + try: + path, how = resolve_binary(args.binary) + except GremlinsError as e: # A11: a path form this platform refuses + path = None + how = ("--binary" if args.binary else + "CLAUDE_CODE_EXECPATH" if os.environ.get("CLAUDE_CODE_EXECPATH") else "PATH") + binary = {"path": path, "how": how, "error": e.payload["error"], + "given": e.payload.get("path", args.binary)} + reasons.append("%s: %s" % (e.payload["error"], binary["given"])) else: - try: - binary.update(binary_info(path)) - except GremlinsError as e: - binary["error"] = e.payload["error"] - reasons.append("%s: %s" % (e.payload["error"], path)) + binary = {"path": path, "how": how} + if path is None: + reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") + else: + try: + binary.update(binary_info(path)) + except GremlinsError as e: + binary["error"] = e.payload["error"] + reasons.append("%s: %s" % (e.payload["error"], path)) warnings = [] base_url = os.environ.get("ANTHROPIC_BASE_URL") if base_url is not None and base_url.strip().rstrip("/").lower() != DEFAULT_BASE_URL: warnings.append("ANTHROPIC_BASE_URL is %r, not the default %s; it is passed to every " "base and gremlin, so they authenticate and bill through that endpoint" % (base_url, DEFAULT_BASE_URL)) + if gate_overridden(): + warnings.append(GATE_OVERRIDE_WARNING) out = { "script": script_path(), "python": py, @@ -2246,7 +2302,7 @@ def cmd_where(args) -> tuple[dict, int]: "binary": binary, "git": _git_report(), "scrub": scrub_report(), - "platform": PLATFORM.describe(), + "platform": platform_report(), "warnings": warnings, "ok": not reasons, } @@ -2417,6 +2473,7 @@ def _diff_fields(a, b, prefix="") -> list[str]: def cmd_init(args) -> tuple[dict, int]: + require_platform() # B1: the first step, before any path is read or written state = state_dir_for(args.run) snap = read_registry(state) if os.path.isdir(state) else None old_cfg = (snap or {}).get("config") @@ -3250,7 +3307,10 @@ def tick_pass(state: str) -> dict: """One pass of the ping rule over every base of the run. The decision and the `ping_in_flight` mark are made under the registry lock (C3d), so concurrent tickers launch one ping per base. The binary check runs before the lock for the bases that look eligible. Pings go through the - same hidden spawn as any job. Returns {"pinged": [job...], "bases": [decision...]}.""" + same hidden spawn as any job. Returns {"pinged": [job...], "bases": [decision...]}. + On a closed platform gate (A1) it returns at once: no ping, nothing read or written.""" + if platform_gate() is not None: + return {"pinged": [], "bases": []} reg = read_registry(state) if not reg or not reg.get("config"): return {"pinged": [], "bases": []} @@ -3977,6 +4037,8 @@ def run(argv) -> tuple[dict, int]: if args.command in CHILD_REFUSED and in_child(): raise GremlinsError("`%s` is refused inside a gremlin (%s is set); a gremlin may only run where, " "list, usage and wait" % (args.command, CHILD_MARKER), command=args.command) + if args.command in GATED_COMMANDS: + require_platform() # A1: an existing registry launches nothing on a closed gate return HANDLERS[args.command](args) diff --git a/plugins/spire/skills/gremlins/tests/test_gremlins.py b/plugins/spire/skills/gremlins/tests/test_gremlins.py index 9962ef2..b6607c0 100644 --- a/plugins/spire/skills/gremlins/tests/test_gremlins.py +++ b/plugins/spire/skills/gremlins/tests/test_gremlins.py @@ -71,6 +71,12 @@ def child_env(): return env +def gate_warning(out: dict) -> list[str]: + """The warning `where` adds when the platform gate is overridden (C13 on a closed platform; + never on Windows), so that the exact `warnings` assertions hold on every host.""" + return [gremlins.GATE_OVERRIDE_WARNING] if out["platform"]["override"] else [] + + class Base(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="gremlins-test-") @@ -79,6 +85,9 @@ class Base(unittest.TestCase): env_patch.start() self.addCleanup(env_patch.stop) os.environ.pop(gremlins.CHILD_MARKER, None) + # C13: the suite runs with the platform gate overridden, so it exercises the POSIX code on + # the porting hosts; on Windows (supported) the variable changes nothing. GateTests drop it. + os.environ[gremlins.UNSUPPORTED_OK_ENV] = "1" def assert_one_json_line(self, raw: str): self.assertTrue(raw.endswith("\n"), repr(raw[-40:])) @@ -1086,15 +1095,51 @@ class CliTests(Base): code, out, raw = call(["where", "--binary", stub]) self.assertEqual(code, 0, out) self.assert_one_json_line(raw) - self.assertEqual(out["warnings"], []) + self.assertEqual(out["warnings"], gate_warning(out)) os.environ["ANTHROPIC_BASE_URL"] = "https://llm-gateway.example.internal/anthropic" code, out, raw = call(["where", "--binary", stub]) self.assertEqual(code, 0, out) self.assert_one_json_line(raw) self.assertTrue(out["ok"]) - self.assertEqual(len(out["warnings"]), 1, out["warnings"]) + self.assertEqual(len(out["warnings"]), 1 + len(gate_warning(out)), out["warnings"]) self.assertIn("ANTHROPIC_BASE_URL", out["warnings"][0]) self.assertIn("llm-gateway.example.internal", out["warnings"][0]) + self.assertEqual(out["warnings"][1:], gate_warning(out)) + + def test_where_platform(self): + """Digest §3.6/§4, A12: `where.platform` is structured; `override` is false on a supported + platform even though the suite sets the override env (C13).""" + stub = make_stub_binary(self.tmp) + self.assertEqual(os.environ.get(gremlins.UNSUPPORTED_OK_ENV), "1") + code, out, raw = call(["where", "--binary", stub]) + self.assertEqual(code, 0, out) + self.assert_one_json_line(raw) + self.assertNotIn("spawn", out) + pf = out["platform"] + self.assertEqual(set(pf), {"name", "sys_platform", "wsl", "supported", "detach", "override"}) + self.assertEqual(pf["sys_platform"], sys.platform) + self.assertEqual(pf["override"], not gremlins.PLATFORM.supported) + if WIN_PLATFORM: + self.assertEqual(pf, {"name": "windows", "sys_platform": sys.platform, "wsl": False, + "supported": True, "detach": "cim-hidden", "override": False}) + self.assertEqual(out["warnings"], []) + + def test_where_refused_binary_path(self): + """A11: a `--binary` path form this platform refuses lands in `binary.error` and the + reasons (exit 2); the rest of the report is still filled.""" + bad = "/mnt/c/x/claude.exe" if WIN_PLATFORM else "C:\\x\\claude.exe" + code, out, raw = call(["where", "--binary", bad]) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertFalse(out["ok"]) + self.assertIsNone(out["binary"]["path"]) + self.assertEqual(out["binary"]["how"], "--binary") + self.assertEqual(out["binary"]["given"], bad) + self.assertIn("path form", out["binary"]["error"]) + self.assertIn(out["binary"]["error"], out["reason"]) + self.assertEqual(out["error"], out["reason"]) + for k in ("script", "python", "uv", "git", "scrub", "platform", "warnings"): + self.assertIn(k, out) def test_child_refuses_kill_expire_init(self): """M4: a gremlin may not run kill, expire or init.""" @@ -1142,6 +1187,175 @@ class CliTests(Base): self.assertIn("error", out) +# -------------------------------------------------------------------------------------- +def closed_platform(): + """B1: a fake closed platform. It subclasses the host's own platform class (on Windows a + `WindowsPlatform`), so path handling stays the host's and `--binary`, `--run` and + `script_path()` work; it reports itself as WSL Linux and unsupported.""" + host = type(gremlins.PLATFORM) + + class ClosedPlatform(host): + supported = False + + def is_wsl(self): + return True + + def describe(self): + return {"name": "posix", "sys_platform": "linux", "wsl": True, "supported": False, + "detach": "setsid"} + return ClosedPlatform() + + +class GateTests(Base): + """Digest §4 with amendments B1, A1, A7, A12: the Windows-only gate.""" + + def setUp(self): + super().setUp() + os.environ.pop(gremlins.UNSUPPORTED_OK_ENV, None) # C13: the gate tests run without it + self.cwd = os.path.join(self.tmp, "proj") + os.makedirs(self.cwd) + os.makedirs(os.path.join(self.tmp, "reports")) + self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-20-g-run.md") + self.state = gremlins.state_dir_for(self.run_log) + p = mock.patch.object(gremlins, "USER_SETTINGS_PATH", os.path.join(self.tmp, "home", "settings.json")) + p.start() + self.addCleanup(p.stop) + self.stub = make_stub_binary(self.tmp) + + def close_gate(self): + p = mock.patch.object(gremlins, "PLATFORM", closed_platform()) + p.start() + self.addCleanup(p.stop) + + def init(self): + return call(["init", "--run", self.run_log, "--model", "claude-opus-5-5[1m]", "--cwd", self.cwd, + "--binary", self.stub]) + + def test_supported_flags(self): + self.assertIs(gremlins.WindowsPlatform().supported, True) + for sp in ("linux", "darwin"): + with self.subTest(sys_platform=sp), mock.patch.object(sys, "platform", sp): + pp = gremlins.PosixPlatform() + self.assertIs(pp.supported, False) + self.assertIs(pp.describe()["supported"], False) # A12: read from `supported` + self.assertEqual(pp.describe()["sys_platform"], sp) + + def test_is_wsl(self): + pp = gremlins.PosixPlatform() + os.environ["WSL_DISTRO_NAME"] = "Ubuntu-20.04" + self.assertTrue(pp.is_wsl()) + os.environ.pop("WSL_DISTRO_NAME") + with mock.patch("builtins.open", mock.mock_open(read_data=b"5.15.90.1-microsoft-standard-WSL2")): + self.assertTrue(pp.is_wsl()) + with mock.patch("builtins.open", mock.mock_open(read_data=b"6.8.0-45-generic")): + self.assertFalse(pp.is_wsl()) + with mock.patch("builtins.open", side_effect=OSError("no /proc")): + self.assertFalse(pp.is_wsl()) + + def test_gate_reason(self): + """A7: no version number; names the platform and WSL; A12: read from describe().""" + if gremlins.PLATFORM.supported: + self.assertIsNone(gremlins.platform_gate()) + self.close_gate() + self.assertEqual(gremlins.platform_gate(), + "spire:gremlins is not tested on linux, WSL yet; use fresh agents for every " + "block (see SKILL.md, Platforms).") + os.environ[gremlins.UNSUPPORTED_OK_ENV] = "1" + self.assertIsNone(gremlins.platform_gate()) + self.assertTrue(gremlins.gate_overridden()) + + def test_host_where(self): + """On the real host without the override env: Windows is supported, no override.""" + code, out, raw = call(["where", "--binary", self.stub]) + self.assert_one_json_line(raw) + if WIN_PLATFORM: + self.assertEqual(code, 0, out) + self.assertTrue(out["platform"]["supported"]) + self.assertFalse(out["platform"]["override"]) + self.assertEqual(out["warnings"], []) + + def test_closed_where(self): + self.close_gate() + code, out, raw = call(["where", "--binary", self.stub]) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertFalse(out["ok"]) + self.assertIn("not tested on linux, WSL", out["reason"]) + self.assertIn("fresh agents", out["reason"]) + self.assertEqual(out["error"], out["reason"]) + self.assertEqual(out["platform"], {"name": "posix", "sys_platform": "linux", "wsl": True, + "supported": False, "detach": "setsid", "override": False}) + # everything else is still filled (the porting hosts need it) + self.assertEqual(out["binary"]["version"], STUB_VERSION) + self.assertNotIn("error", out["binary"]) + for k in ("script", "python", "uv", "git", "scrub", "warnings"): + self.assertIn(k, out) + self.assertEqual(out["warnings"], []) + + def test_closed_init(self): + self.close_gate() + code, out, raw = self.init() + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertEqual(out["error"], "platform not supported") + self.assertIn("not tested on linux, WSL", out["reason"]) + self.assertEqual(out["platform"]["supported"], False) + self.assertEqual(out["platform"]["override"], False) + self.assertFalse(os.path.exists(self.state)) + # B1: the gate comes before any path handling (a bad --run is not even looked at) + code, out, _ = call(["init", "--run", os.path.join(self.tmp, "not-a-run.txt"), "--model", + "claude-opus-5-5[1m]", "--cwd", self.cwd, "--binary", self.stub]) + self.assertEqual((code, out["error"]), (2, "platform not supported")) + + def test_override(self): + self.close_gate() + os.environ[gremlins.UNSUPPORTED_OK_ENV] = "1" + code, out, raw = call(["where", "--binary", self.stub]) + self.assertEqual(code, 0, out) + self.assert_one_json_line(raw) + self.assertTrue(out["ok"]) + self.assertTrue(out["platform"]["override"]) + self.assertFalse(out["platform"]["supported"]) + self.assertEqual(out["warnings"], [gremlins.GATE_OVERRIDE_WARNING]) + self.assertIn(gremlins.UNSUPPORTED_OK_ENV, out["warnings"][0]) + code, out, _ = self.init() + self.assertEqual(code, 0, out) + self.assertTrue(out["created"]) + self.assertTrue(os.path.isfile(gremlins.registry_path(self.state))) + + def test_closed_gate_on_existing_registry(self): + """A1: a registry made with the gate open; with it closed, nothing launches and no job + entry is written; the tick launches no ping (it reads nothing).""" + code, out, _ = self.init() + self.assertEqual(code, 0, out) + before = gremlins.read_registry(self.state) + brief = os.path.join(self.tmp, "b.md") + with open(brief, "w") as f: + f.write("x\n") + self.close_gate() + for argv in (["spawn", "--base", "B1", "--block", "1.1", "--brief", brief], + ["train", "--base", "B1", "--brief", brief], + ["continue", "--job", "j0001", "--brief", brief], + ["rebase", "--base", "B1", "--as", "B2", "--brief", brief], + ["tick"]): + with self.subTest(cmd=argv[0]): + code, out, raw = call([argv[0], "--run", self.run_log, *argv[1:]]) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertEqual(out["error"], "platform not supported") + self.assertIn("not tested on linux, WSL", out["reason"]) + self.assertEqual(out["platform"]["supported"], False) + after = gremlins.read_registry(self.state) + self.assertEqual(after["jobs"], {}) + self.assertEqual(after, before) + with mock.patch.object(gremlins, "read_registry", side_effect=AssertionError("tick read the registry")): + self.assertEqual(gremlins.tick_pass(self.state), {"pinged": [], "bases": []}) + # list and wait still work on a closed gate + self.assertEqual(call(["list", "--run", self.run_log])[0], 0) + code, out, _ = call(["wait", "--run", self.run_log, "--all", "--timeout", "1"]) + self.assertEqual((code, out["pings"]), (0, [])) + + # -------------------------------------------------------------------------------------- @unittest.skipUnless(shutil.which("uv"), "uv not on PATH") class SubprocessTests(Base):