Improve Fray job name in remote & StepSpec (#3076)

* follow up to #3051
* if `name` is specified in `remote` don't change it
* if name is not specified, best effort to get base name + short uuid
* I don't think arg hashing is worth the complexity + on restart would
collide, and if the user wants name lock, they can handle it
* `StepSpec` uses the name with hash + short uuid

This way:
 * afaiu you can get name "lock" via explicit name 
 * restart of steps should just work

Author: ravwojdyla

lib/marin/src/marin/execution/remote.py

@@ -17,8 +17,9 @@ def train(...): ...             # explicit resources
 
 from __future__ import annotations
 
-import hashlib
+import dataclasses
 import re
+import uuid
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Generic, ParamSpec, TypeVar
@@ -29,7 +30,7 @@ def train(...): ...             # explicit resources
 P = ParamSpec("P")
 R = TypeVar("R")
 
-DEFAULT_JOB_NAME = "fray_exec_job"
+DEFAULT_JOB_NAME = "remote_job"
 
 
 def _sanitize_job_name(name: str) -> str:
@@ -39,12 +40,6 @@ def _sanitize_job_name(name: str) -> str:
     return sanitized or DEFAULT_JOB_NAME
 
 
-def _args_hash(*args: object, **kwargs: object) -> str:
-    """Short hash of call arguments to disambiguate jobs with the same function name."""
-    content = repr(args) + repr(sorted(kwargs.items()))
-    return hashlib.sha256(content.encode()).hexdigest()[:8]
-
-
 @dataclass(frozen=True)
 class RemoteCallable(Generic[P, R]):
     """A callable wrapper that submits its function to Fray when called.
@@ -60,12 +55,21 @@ class RemoteCallable(Generic[P, R]):
     pip_dependency_groups: list[str] = field(default_factory=list)
     name: str | None = None
 
+    def named(self, name: str) -> RemoteCallable:
+        """Noop if already has a name. Otherwise use provided name."""
+        if self.name:
+            return self
+        return dataclasses.replace(self, name=name)
+
     # TODO: JobHandle doesn't have this option now, but we could make this return the R
     def __call__(self, *args: P.args, **kwargs: P.kwargs) -> None:
         """Submit fn to Fray and block until completion."""
 
-        base_name = self.name or getattr(self.fn, "__name__", None) or DEFAULT_JOB_NAME
-        name = f"{base_name}-{_args_hash(*args, **kwargs)}"
+        if self.name:
+            name = self.name
+        else:
+            fn_name = getattr(self.fn, "__name__", None) or DEFAULT_JOB_NAME
+            name = f"{fn_name}-{uuid.uuid4().hex[:8]}"
         c = fray_client.current_client()
         handle = c.submit(
             JobRequest(

lib/marin/src/marin/execution/step_spec.py

@@ -6,6 +6,7 @@
 import dataclasses
 import hashlib
 import json
+import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
 from functools import cached_property
@@ -92,6 +93,8 @@ def executable_fn(self) -> Callable[[str], Any]:
         )
 
         if isinstance(self.fn, RemoteCallable):
-            wrapped = dataclasses.replace(self.fn, fn=wrapped)
+            job_name = f"{self.name_with_hash}-{uuid.uuid4().hex[:8]}"
+            remote_callable = self.fn.named(job_name)
+            wrapped = dataclasses.replace(remote_callable, fn=wrapped)
 
         return wrapped