Simplify JobAlreadyExists and fix stale job ID under auth

JobAlreadyExists no longer carries a Job object — the client-side
job_id could differ from the server's canonical ID when auth rewrites
the user segment, making the Job handle point at a nonexistent job.

Instead, Fray now passes existing_job_policy=KEEP when adopt_existing
is True, so the server returns the canonical ID via LaunchJobResponse
on the success path. No error-path workaround needed.

Author: rjpower

lib/fray/src/fray/v2/iris_backend.py

@@ -508,6 +508,7 @@ def submit(self, request: JobRequest, adopt_existing: bool = True) -> IrisJobHan
         replicas = request.replicas or 1
         coscheduling = resolve_coscheduling(request.resources.device, replicas)
 
+        policy = cluster_pb2.EXISTING_JOB_POLICY_KEEP if adopt_existing else cluster_pb2.EXISTING_JOB_POLICY_UNSPECIFIED
         try:
             job = self._iris.submit(
                 entrypoint=iris_entrypoint,
@@ -519,12 +520,10 @@ def submit(self, request: JobRequest, adopt_existing: bool = True) -> IrisJobHan
                 replicas=replicas,
                 max_retries_failure=request.max_retries_failure,
                 max_retries_preemption=request.max_retries_preemption,
+                existing_job_policy=policy,
             )
         except IrisJobAlreadyExists as e:
-            if adopt_existing:
-                logger.info("Job %s already exists, adopting existing job", request.name)
-                return IrisJobHandle(e.job)
-            raise FrayJobAlreadyExists(request.name, handle=IrisJobHandle(e.job)) from e
+            raise FrayJobAlreadyExists(request.name) from e
         return IrisJobHandle(job)
 
     def host_actor(

lib/iris/src/iris/client/client.py

@@ -132,8 +132,7 @@ def __init__(self, job_id: JobName, status: cluster_pb2.JobStatus):
 class JobAlreadyExists(Exception):
     """Raised when a job with the same name is already running."""
 
-    def __init__(self, job: "Job", message: str):
-        self.job = job
+    def __init__(self, message: str):
         super().__init__(message)
 
 
@@ -723,7 +722,7 @@ def submit(
             )
         except ConnectError as e:
             if e.code == Code.ALREADY_EXISTS:
-                raise JobAlreadyExists(Job(self, job_id), str(e)) from e
+                raise JobAlreadyExists(str(e)) from e
             raise
 
         return Job(self, canonical_id)

scripts/iris/dev_tpu.py

@@ -521,7 +521,7 @@ def allocate(
                 constraints=constraints or None,
             )
         except JobAlreadyExists as exc:
-            raise click.ClickException(f"Job already exists for session '{session_name}': {exc.job.job_id}") from exc
+            raise click.ClickException(f"Job already exists for session '{session_name}': {exc}") from exc
 
         workers = wait_for_workers(job, timeout=timeout, project=project)
         for worker in workers: