fix(remote): multi-machine spawn + dated target names

contrib/lewm_hillclimb_guided/flow.yaml

@@ -35,7 +35,7 @@
 # Run (long-lived — use tmux/nohup):
 #   OPENROUTER_API_KEY=... saage run contrib/lewm_hillclimb_guided/flow.yaml
 #   ... --set train_epochs=8 --set target_success=74.0   # tunable
-provider: { type: openrouter, model: "deepseek/deepseek-v4-flash" }
+provider: { type: openrouter, model: "deepseek/deepseek-v4-flash-0731" }
 workspace: /home/cpadwick/code/le-wm
 venv: .venv
 # what the remote sidecar collects from the workspace (local runs ignore this)

saage/remote/cli.py

@@ -72,7 +72,8 @@ def add_parser(sub: argparse._SubParsersAction) -> None:
     sp.add_argument("--gpu", default="auto",
                     help="GPU class (a10/a100/h100/gh200), exact instance type, "
                          "or 'auto' = cheapest with capacity (default)")
-    sp.add_argument("--name", default=None, help="target name (default: lambda-<hhmm>)")
+    sp.add_argument("--name", default=None,
+                    help="target name (default: lambda-<yyyymmdd-hhmmss>)")
     sp.add_argument("--extra-key", action="append", default=[],
                     help="also authorize this Lambda-registered ssh key name on "
                          "the node (repeatable)")
@@ -256,16 +257,21 @@ def _lambda_api() -> LambdaAPI:
     return LambdaAPI(key)
 
 
-def _spawn(args: argparse.Namespace) -> int:
+def _default_spawn_name() -> str:
     from datetime import datetime, timezone
+    return f"lambda-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"
+
+
+def _spawn(args: argparse.Namespace) -> int:
     api = _lambda_api()
     key_path = ensure_ssh_key()
-    api.ensure_ssh_key(SAAGE_KEY_NAME, key_path.with_suffix(".pub").read_text().strip())
+    key_name = api.ensure_ssh_key(SAAGE_KEY_NAME,
+                                  key_path.with_suffix(".pub").read_text().strip())
 
     itype, region, price = pick_instance_type(api.instance_types(), args.gpu)
-    name = args.name or f"lambda-{datetime.now(timezone.utc).strftime('%H%M')}"
+    name = args.name or _default_spawn_name()
     print(f"launching {itype} in {region} (${price:.2f}/hr) as {name!r} …")
-    iid = api.launch(itype, region, SAAGE_KEY_NAME, f"saage-{name}")
+    iid = api.launch(itype, region, key_name, f"saage-{name}")
     # billing starts NOW — print the id before anything that can fail, and
     # terminate on ANY failure (incl. Ctrl-C) so an error never leaks a node
     print(f"instance {iid} launching (billing started)")

saage/remote/lambda_api.py

@@ -16,13 +16,19 @@
 """
 from __future__ import annotations
 
+import hashlib
 import json
 import logging
 import subprocess
 import time
 import urllib.error
 import urllib.request
 
+
+def _fp8(pubkey: str) -> str:
+    """Short stable fingerprint used to derive a per-machine key-name variant."""
+    return hashlib.sha256(pubkey.encode()).hexdigest()[:8]
+
 log = logging.getLogger("saage.remote")
 
 BASE = "https://cloud.lambda.ai/api/v1"
@@ -79,10 +85,26 @@ def ssh_keys(self) -> list[dict]:
 
     # -- write -----------------------------------------------------------------
 
-    def ensure_ssh_key(self, name: str, public_key: str) -> None:
-        if any(k["name"] == name for k in self.ssh_keys()):
-            return
-        self._request("/ssh-keys", {"name": name, "public_key": public_key})
+    def ensure_ssh_key(self, name: str, public_key: str) -> str:
+        """Return the name of a registered key whose CONTENT matches public_key.
+
+        Matching by name alone is a trap: a second machine's saage key would be
+        shadowed by the name registered from the first machine, the instance
+        would boot authorized for the wrong key, and ssh would never come up.
+        On a content mismatch the key is registered under a stable
+        '<name>-<fp8>' variant (never touching the other machine's entry)."""
+        want = " ".join(public_key.split()[:2])     # 'type blob' — comment varies
+        keys = {k["name"]: " ".join(k["public_key"].split()[:2])
+                for k in self.ssh_keys()}
+        alt = f"{name}-{_fp8(want)}"
+        for cand in (name, alt):
+            if keys.get(cand) == want:
+                return cand
+            if cand not in keys:
+                self._request("/ssh-keys", {"name": cand, "public_key": public_key})
+                return cand
+        raise LambdaError(f"ssh key names {name!r} and {alt!r} are both taken by "
+                          f"different keys — remove one in the Lambda dashboard")
 
     def launch(self, instance_type: str, region: str, ssh_key_name: str,
                name: str) -> str:
@@ -175,5 +197,11 @@ def wait_ssh(host: str, user: str, key_path: str, timeout_s: int = 300) -> None:
             capture_output=True)
         if proc.returncode == 0:
             return
+        if b"Permission denied" in proc.stderr:
+            # auth rejection is deterministic — waiting out the timeout would
+            # only bury the real problem (wrong key authorized on the node)
+            raise LambdaError(
+                f"ssh to {host} rejected the saage key (Permission denied) — "
+                f"the node is up but authorized for a different key")
         time.sleep(10)
     raise LambdaError(f"instance at {host} is active but ssh never came up")

tests/remote/test_lambda_api.py

@@ -75,6 +75,79 @@ def terminate(self, ids):
     assert NeverActive.terminated == ["i-2"]
 
 
+# --------------------------------------------------------------------------- #
+# ensure_ssh_key: content-aware — a second machine's saage key must not be
+# shadowed by the same NAME registered from another machine (the instance
+# would boot authorized for the wrong key and ssh would never come up)
+# --------------------------------------------------------------------------- #
+class _KeyAPI(LambdaAPI):
+    def __init__(self, registered):
+        super().__init__("key")
+        self.registered = registered            # list of {"name", "public_key"}
+        self.uploads = []
+
+    def ssh_keys(self):
+        return self.registered
+
+    def _request(self, path, payload=None):
+        assert path == "/ssh-keys"
+        self.uploads.append(payload)
+        self.registered.append(payload)
+        return {"data": payload}
+
+
+_KEY_A = "ssh-ed25519 AAAA_machine_A saage-remote"
+_KEY_B = "ssh-ed25519 AAAA_machine_B saage-remote"
+
+
+def test_ensure_ssh_key_registers_when_absent():
+    api = _KeyAPI([])
+    assert api.ensure_ssh_key("saage-remote", _KEY_A) == "saage-remote"
+    assert len(api.uploads) == 1
+
+
+def test_ensure_ssh_key_reuses_matching_key_ignoring_comment():
+    api = _KeyAPI([{"name": "saage-remote",
+                    "public_key": "ssh-ed25519 AAAA_machine_A other-comment"}])
+    assert api.ensure_ssh_key("saage-remote", _KEY_A) == "saage-remote"
+    assert api.uploads == []
+
+
+def test_ensure_ssh_key_name_taken_by_other_machine_registers_variant():
+    api = _KeyAPI([{"name": "saage-remote", "public_key": _KEY_A}])
+    name = api.ensure_ssh_key("saage-remote", _KEY_B)
+    assert name != "saage-remote"                 # never launch with the wrong key
+    assert name.startswith("saage-remote-")
+    assert len(api.uploads) == 1
+
+
+def test_ensure_ssh_key_variant_is_stable_and_reused():
+    api = _KeyAPI([{"name": "saage-remote", "public_key": _KEY_A}])
+    first = api.ensure_ssh_key("saage-remote", _KEY_B)
+    again = api.ensure_ssh_key("saage-remote", _KEY_B)
+    assert again == first
+    assert len(api.uploads) == 1                  # second call reuses, no re-upload
+
+
+def test_wait_ssh_fails_fast_on_permission_denied(monkeypatch):
+    # auth rejection is deterministic — looping the full timeout just hides the
+    # real problem (wrong key registered) behind "ssh never came up"
+    import subprocess as sp
+    from saage.remote.lambda_api import wait_ssh
+    calls = []
+
+    def fake_run(cmd, capture_output=True, **kw):
+        calls.append(cmd)
+        if cmd[0] == "ssh-keygen":
+            return sp.CompletedProcess(cmd, 0, b"", b"")
+        return sp.CompletedProcess(cmd, 255, b"", b"Permission denied (publickey).")
+
+    monkeypatch.setattr(sp, "run", fake_run)
+    with pytest.raises(LambdaError, match="[Pp]ermission denied"):
+        wait_ssh("1.2.3.4", "ubuntu", "/k", timeout_s=300)
+    assert len([c for c in calls if c[0] == "ssh"]) == 1   # no 300s retry loop
+
+
 def test_network_errors_become_lambda_errors(monkeypatch):
     # URLError (DNS/conn refused) must be wrapped like HTTPError, or it escapes
     # every `except LambdaError` net (incl. wait_active's transient tolerance)

tests/remote/test_target_cleanup.py

@@ -147,6 +147,16 @@ def test_list_no_targets(saage_home, capsys):
     assert "none registered" in capsys.readouterr().out
 
 
+# -- spawn default name ----------------------------------------------------------
+
+def test_default_spawn_name_is_full_timestamp():
+    # lambda-<hhmm> collided across days and told you nothing about age;
+    # the full stamp sorts and dates the box at a glance
+    import re
+    from saage.remote.cli import _default_spawn_name
+    assert re.fullmatch(r"lambda-\d{8}-\d{6}", _default_spawn_name())
+
+
 # -- terminate unregisters -----------------------------------------------------
 
 class _FakeLambda: