Repository: skypilot-org/skypilot
Pull request: [Core] Exit with non-zero code on launch/exec/logs/jobs launch/jobs logs
Closes https://github.com/skypilot-org/skypilot/issues/4599.

A better fix than #4823. tail_logs API is now updated to return an exit code representing the job's status (failed or succeeded). This eliminates the need for a separate API call to fetch the job status.

For now, we do not return the job's actual exit code because of complexities related to multi-node jobs. 

Need to test backward compatibility.

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [x] New smoke tests
- [x] All smoke tests: `pytest tests/test_smoke.py` 


Problem statement:
If a job fails it should exit with 1 instead of 0
```bash
$ sky launch ..
ERROR: Job 439 failed with return code list: [1]
✓ Job finished (status: FAILED).
$ echo $?
0
```

It would be nice if the sky command would exit with 1 instead of 0 to indicate that the job failed 👍

Hints from issue discussion:
Thanks for the report @kesitrifork. Is this for ease of scripting?

As a workaround, would `sky logs [cluster] [job id] --status` work?
@concretevitamin Yes exactly, ideally we would switch to using the python interface, but scripting is the quick start.

I had not seen that, I thought I had looked, I will try using that ✌️
> If specified, do not show logs but exit with a status code for the job’s status: 0 for succeeded, or 1 for all other statuses.
+1 to using a non zero exit code on `sky launch` and `sky exec` if the job fails. Keeping this issue open. 
User requested this for `sky jobs launch` too:
```
ERROR: Job 1 failed with return code list: [1] 
✓ Job finished (status: FAILED).
✓ Managed job finished: 7 (status: FAILED).

$ echo $?
0 # Should not be zero
```

Commit message: minor

Patch to review:
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index d5c04fa4612..acd92fb3f27 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -3823,6 +3823,10 @@ def tail_logs(self,
             follow: Whether to follow the logs.
             tail: The number of lines to display from the end of the
                 log file. If 0, print all lines.
+
+        Returns:
+            The exit code of the tail command. Returns code 100 if the job has
+            failed. See job_lib.JobExitCode for possible return codes.
         """
         code = job_lib.JobLibCodeGen.tail_logs(job_id,
                                                managed_job_id=managed_job_id,
@@ -3856,7 +3860,7 @@ def tail_managed_job_logs(self,
                               job_id: Optional[int] = None,
                               job_name: Optional[str] = None,
                               controller: bool = False,
-                              follow: bool = True) -> None:
+                              follow: bool = True) -> int:
         # if job_name is not None, job_id should be None
         assert job_name is None or job_id is None, (job_name, job_id)
         code = managed_jobs.ManagedJobCodeGen.stream_logs(
@@ -3869,13 +3873,17 @@ def tail_managed_job_logs(self,
             signal.signal(signal.SIGTSTP, backend_utils.stop_handler)
 
         # Refer to the notes in tail_logs.
-        self.run_on_head(
-            handle,
-            code,
-            stream_logs=True,
-            process_stream=False,
-            ssh_mode=command_runner.SshMode.INTERACTIVE,
-        )
+        try:
+            returncode = self.run_on_head(
+                handle,
+                code,
+                stream_logs=True,
+                process_stream=False,
+                ssh_mode=command_runner.SshMode.INTERACTIVE,
+            )
+        except SystemExit as e:
+            returncode = e.code
+        return returncode
 
     def sync_down_managed_job_logs(
             self,
diff --git a/sky/cli.py b/sky/cli.py
index ed7045dd9fd..5c06989d3bf 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -1227,11 +1227,15 @@ def launch(
             clusters=[handle.get_cluster_name()])
         # job_id will be None if no job was submitted (e.g. no entrypoint
         # provided)
+        ret_code = 0
         if not detach_run and job_id is not None:
-            sdk.tail_logs(handle.get_cluster_name(), job_id, follow=True)
+            ret_code = sdk.tail_logs(handle.get_cluster_name(),
+                                     job_id,
+                                     follow=True)
         click.secho(
             ux_utils.command_hint_messages(ux_utils.CommandHintType.CLUSTER_JOB,
                                            job_id, handle.get_cluster_name()))
+        sys.exit(ret_code)
 
 
 @cli.command(cls=_DocumentedCodeCommand)
@@ -1377,7 +1381,8 @@ def exec(cluster: Optional[str], cluster_option: Optional[str],
     job_id_handle = _async_call_or_wait(request_id, async_call, 'sky.exec')
     if not async_call and not detach_run:
         job_id, _ = job_id_handle
-        sdk.tail_logs(cluster, job_id, follow=True)
+        ret_code = sdk.tail_logs(cluster, job_id, follow=True)
+        sys.exit(ret_code)
 
 
 def _handle_jobs_queue_request(
@@ -2121,12 +2126,20 @@ def logs(
     one job_id can be provided.
 
     2. If ``--status`` is specified, print the status of the job and exit with
-    returncode 0 if the job succeeded, or 1 otherwise. At most one job_id can
-    be specified.
+    returncode 0 if the job succeeded. At most one job_id can
+    be specified. Other possible return codes:
+
+    - 100: job failed.
+    - 101: job not finished.
+    - 102: job not found.
+    - 103: job was cancelled by the user.
 
     3. If ``--sync-down`` is specified, the logs of the job will be downloaded
     from the cluster and saved to the local machine under
-    ``~/sky_logs``. Mulitple job_ids can be specified.
+    ``~/sky_logs``. Multiple job_ids can be specified.
+
+    4. If the job fails or fetching the logs fails, the command will exit with
+    a non-zero return code.
     """
     if sync_down and status:
         raise click.UsageError(
@@ -2174,17 +2187,18 @@ def logs(
         # it will return {None: None}.
         if job_id is None:
             click.secho(f'No job found on cluster {cluster!r}.', fg='red')
-            sys.exit(1)
+            sys.exit(job_lib.JobExitCode.NOT_FOUND)
         job_status = list(job_statuses.values())[0]
         job_status_str = job_status.value if job_status is not None else 'None'
         click.echo(f'Job {job_id}: {job_status_str}')
         if job_status == job_lib.JobStatus.SUCCEEDED:
             return
         else:
+            ret_code = job_lib.JobExitCode.from_job_status(job_status)
             if job_status is None:
                 id_str = '' if job_id is None else f'{job_id} '
                 click.secho(f'Job {id_str}not found', fg='red')
-            sys.exit(1)
+            sys.exit(ret_code)
 
     job_str = f'job {job_id}'
     if job_id is None:
@@ -2194,7 +2208,8 @@ def logs(
                 f'{colorama.Style.RESET_ALL}')
 
     # Stream logs from the server.
-    sdk.tail_logs(cluster, job_id, follow, tail=tail)
+    ret_code = sdk.tail_logs(cluster, job_id, follow, tail=tail)
+    sys.exit(ret_code)
 
 
 @cli.command()
@@ -3893,10 +3908,11 @@ def jobs_launch(
                                         'sky.jobs.launch')
     if not async_call and not detach_run:
         job_id = job_id_handle[0]
-        managed_jobs.tail_logs(name=None,
-                               job_id=job_id,
-                               follow=True,
-                               controller=False)
+        ret_code = managed_jobs.tail_logs(name=None,
+                                          job_id=job_id,
+                                          follow=True,
+                                          controller=False)
+        sys.exit(ret_code)
 
 
 @jobs.command('queue', cls=_DocumentedCodeCommand)
@@ -4127,11 +4143,12 @@ def jobs_logs(name: Optional[str], job_id: Optional[int], follow: bool,
                 logger.info(f'{fore.CYAN}Job {job} logs{controller_str}: '
                             f'{log_local_path}{style.RESET_ALL}')
         else:
-            managed_jobs.tail_logs(name=name,
-                                   job_id=job_id,
-                                   follow=follow,
-                                   controller=controller,
-                                   refresh=refresh)
+            ret_code = managed_jobs.tail_logs(name=name,
+                                              job_id=job_id,
+                                              follow=follow,
+                                              controller=controller,
+                                              refresh=refresh)
+            sys.exit(ret_code)
     except exceptions.ClusterNotUpError:
         with ux_utils.print_exception_no_traceback():
             raise
diff --git a/sky/client/sdk.py b/sky/client/sdk.py
index f4778a1b9c9..6606760e9d3 100644
--- a/sky/client/sdk.py
+++ b/sky/client/sdk.py
@@ -562,7 +562,7 @@ def tail_logs(cluster_name: str,
               job_id: Optional[int],
               follow: bool,
               tail: int = 0,
-              output_stream: Optional['io.TextIOBase'] = None) -> None:
+              output_stream: Optional['io.TextIOBase'] = None) -> int:
     """Tails the logs of a job.
 
     Args:
@@ -575,7 +575,8 @@ def tail_logs(cluster_name: str,
             console.
 
     Returns:
-        None
+        Exit code based on success or failure of the job. 0 if success,
+        100 if the job failed. See job_lib.JobExitCode for possible exit codes.
 
     Request Raises:
         ValueError: if arguments are invalid or the cluster is not supported.
@@ -601,7 +602,7 @@ def tail_logs(cluster_name: str,
         timeout=(client_common.API_SERVER_REQUEST_CONNECTION_TIMEOUT_SECONDS,
                  None))
     request_id = server_common.get_request_id(response)
-    stream_response(request_id, response, output_stream)
+    return stream_response(request_id, response, output_stream)
 
 
 @usage_lib.entrypoint
diff --git a/sky/core.py b/sky/core.py
index 5869dcdd4ca..e484e9934fe 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -783,7 +783,7 @@ def cancel(
 def tail_logs(cluster_name: str,
               job_id: Optional[int],
               follow: bool = True,
-              tail: int = 0) -> None:
+              tail: int = 0) -> int:
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Tails the logs of a job.
 
@@ -799,6 +799,12 @@ def tail_logs(cluster_name: str,
           not the same as the user who created the cluster.
         sky.exceptions.CloudUserIdentityError: if we fail to get the current
           user identity.
+
+    Returns:
+        Return code based on success or failure of the job. 0 if success,
+          100 if the job failed. Note: This is not the return code of the job
+          script.
+
     """
     # Check the status of the cluster.
     handle = backend_utils.check_cluster_available(
@@ -808,7 +814,7 @@ def tail_logs(cluster_name: str,
     backend = backend_utils.get_backend_from_handle(handle)
 
     usage_lib.record_cluster_name_for_current_operation(cluster_name)
-    backend.tail_logs(handle, job_id, follow=follow, tail=tail)
+    return backend.tail_logs(handle, job_id, follow=follow, tail=tail)
 
 
 @usage_lib.entrypoint
diff --git a/sky/exceptions.py b/sky/exceptions.py
index bc9ba1b33c2..c593cfcd4ab 100644
--- a/sky/exceptions.py
+++ b/sky/exceptions.py
@@ -234,7 +234,7 @@ def __init__(self, returncode: int, command: str, error_msg: str,
         else:
             if (len(command) > 100 and
                     not env_options.Options.SHOW_DEBUG_INFO.get()):
-                # Chunck the command to avoid overflow.
+                # Chunk the command to avoid overflow.
                 command = command[:100] + '...'
             message = (f'Command {command} failed with return code '
                        f'{returncode}.\n{error_msg}')
diff --git a/sky/jobs/client/sdk.py b/sky/jobs/client/sdk.py
index b910e38c18b..37a37a24643 100644
--- a/sky/jobs/client/sdk.py
+++ b/sky/jobs/client/sdk.py
@@ -184,7 +184,7 @@ def tail_logs(name: Optional[str] = None,
               follow: bool = True,
               controller: bool = False,
               refresh: bool = False,
-              output_stream: Optional['io.TextIOBase'] = None) -> None:
+              output_stream: Optional['io.TextIOBase'] = None) -> int:
     """Tails logs of managed jobs.
 
     You can provide either a job name or a job ID to tail logs. If both are not
@@ -199,6 +199,10 @@ def tail_logs(name: Optional[str] = None,
         output_stream: The stream to write the logs to. If None, print to the
             console.
 
+    Returns:
+        Exit code based on success or failure of the job. 0 if success,
+        100 if the job failed. See job_lib.JobExitCode for possible exit codes.
+
     Request Raises:
         ValueError: invalid arguments.
         sky.exceptions.ClusterNotUpError: the jobs controller is not up.
@@ -217,7 +221,7 @@ def tail_logs(name: Optional[str] = None,
         timeout=(5, None),
     )
     request_id = server_common.get_request_id(response)
-    sdk.stream_response(request_id, response, output_stream)
+    return sdk.stream_response(request_id, response, output_stream)
 
 
 @usage_lib.entrypoint
diff --git a/sky/jobs/constants.py b/sky/jobs/constants.py
index de158c9a826..75cbf05edb3 100644
--- a/sky/jobs/constants.py
+++ b/sky/jobs/constants.py
@@ -40,7 +40,7 @@
 # The version of the lib files that jobs/utils use. Whenever there is an API
 # change for the jobs/utils, we need to bump this version and update
 # job.utils.ManagedJobCodeGen to handle the version update.
-MANAGED_JOBS_VERSION = 2
+MANAGED_JOBS_VERSION = 3
 
 # The command for setting up the jobs dashboard on the controller. It firstly
 # checks if the systemd services are available, and if not (e.g., Kubernetes
diff --git a/sky/jobs/server/core.py b/sky/jobs/server/core.py
index 5d95fc6150e..8ebb2e99d16 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -460,12 +460,16 @@ def cancel(name: Optional[str] = None,
 
 @usage_lib.entrypoint
 def tail_logs(name: Optional[str], job_id: Optional[int], follow: bool,
-              controller: bool, refresh: bool) -> None:
+              controller: bool, refresh: bool) -> int:
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Tail logs of managed jobs.
 
     Please refer to sky.cli.job_logs for documentation.
 
+    Returns:
+        Exit code based on success or failure of the job. 0 if success,
+        100 if the job failed. See job_lib.JobExitCode for possible exit codes.
+
     Raises:
         ValueError: invalid arguments.
         sky.exceptions.ClusterNotUpError: the jobs controller is not up.
@@ -494,11 +498,11 @@ def tail_logs(name: Optional[str], job_id: Optional[int], follow: bool,
     backend = backend_utils.get_backend_from_handle(handle)
     assert isinstance(backend, backends.CloudVmRayBackend), backend
 
-    backend.tail_managed_job_logs(handle,
-                                  job_id=job_id,
-                                  job_name=name,
-                                  follow=follow,
-                                  controller=controller)
+    return backend.tail_managed_job_logs(handle,
+                                         job_id=job_id,
+                                         job_name=name,
+                                         follow=follow,
+                                         controller=controller)
 
 
 def start_dashboard_forwarding(refresh: bool = False) -> Tuple[int, int]:
diff --git a/sky/jobs/utils.py b/sky/jobs/utils.py
index 2b158234bcd..609aeb0c123 100644
--- a/sky/jobs/utils.py
+++ b/sky/jobs/utils.py
@@ -511,8 +511,14 @@ def cancel_job_by_name(job_name: str) -> str:
     return f'Job {job_name!r} is scheduled to be cancelled.'
 
 
-def stream_logs_by_id(job_id: int, follow: bool = True) -> str:
-    """Stream logs by job id."""
+def stream_logs_by_id(job_id: int, follow: bool = True) -> Tuple[str, int]:
+    """Stream logs by job id.
+
+    Returns:
+        A tuple containing the log message and an exit code based on success or
+        failure of the job. 0 if success, 100 if the job failed.
+        See job_lib.JobExitCode for possible exit codes.
+    """
 
     def should_keep_logging(status: managed_job_state.ManagedJobStatus) -> bool:
         # If we see CANCELLING, just exit - we could miss some job logs but the
@@ -547,13 +553,16 @@ def should_keep_logging(status: managed_job_state.ManagedJobStatus) -> bool:
                             start_streaming = True
                         if start_streaming:
                             print(line, end='', flush=True)
-                return ''
-            return (f'{colorama.Fore.YELLOW}'
-                    f'Job {job_id} is already in terminal state '
-                    f'{managed_job_status.value}. For more details, run: '
-                    f'sky jobs logs --controller {job_id}'
-                    f'{colorama.Style.RESET_ALL}'
-                    f'{job_msg}')
+                return '', job_lib.JobExitCode.from_managed_job_status(
+                    managed_job_status)
+            return (
+                f'{colorama.Fore.YELLOW}'
+                f'Job {job_id} is already in terminal state '
+                f'{managed_job_status.value}. For more details, run: '
+                f'sky jobs logs --controller {job_id}'
+                f'{colorama.Style.RESET_ALL}'
+                f'{job_msg}',
+                job_lib.JobExitCode.from_managed_job_status(managed_job_status))
         backend = backends.CloudVmRayBackend()
         task_id, managed_job_status = (
             managed_job_state.get_latest_task_id_status(job_id))
@@ -604,11 +613,12 @@ def should_keep_logging(status: managed_job_state.ManagedJobStatus) -> bool:
                                            job_id=None,
                                            managed_job_id=job_id,
                                            follow=follow)
-            if returncode == 0:
-                # If the log tailing exit successfully (the real job can be
-                # SUCCEEDED or FAILED), we can safely break the loop. We use the
-                # status in job queue to show the information, as the
-                # ManagedJobStatus is not updated yet.
+            if returncode in [rc.value for rc in job_lib.JobExitCode]:
+                # If the log tailing exits with a known exit code we can safely
+                # break the loop because it indicates the tailing process
+                # succeeded (even though the real job can be SUCCEEDED or
+                # FAILED). We use the status in job queue to show the
+                # information, as the ManagedJobStatus is not updated yet.
                 job_statuses = backend.get_job_status(handle, stream_logs=False)
                 job_status = list(job_statuses.values())[0]
                 assert job_status is not None, 'No job found.'
@@ -728,18 +738,24 @@ def is_managed_job_status_updated(
     logger.info(
         ux_utils.finishing_message(f'Managed job finished: {job_id} '
                                    f'(status: {managed_job_status.value}).'))
-    return ''
+    return '', job_lib.JobExitCode.from_managed_job_status(managed_job_status)
 
 
 def stream_logs(job_id: Optional[int],
                 job_name: Optional[str],
                 controller: bool = False,
-                follow: bool = True) -> str:
-    """Stream logs by job id or job name."""
+                follow: bool = True) -> Tuple[str, int]:
+    """Stream logs by job id or job name.
+
+    Returns:
+        A tuple containing the log message and the exit code based on success
+        or failure of the job. 0 if success, 100 if the job failed.
+        See job_lib.JobExitCode for possible exit codes.
+    """
     if job_id is None and job_name is None:
         job_id = managed_job_state.get_latest_job_id()
         if job_id is None:
-            return 'No managed job found.'
+            return 'No managed job found.', job_lib.JobExitCode.NOT_FOUND
 
     if controller:
         if job_id is None:
@@ -754,7 +770,8 @@ def stream_logs(job_id: Optional[int],
                 if job['job_name'] == job_name
             }
             if not managed_job_ids:
-                return f'No managed job found with name {job_name!r}.'
+                return (f'No managed job found with name {job_name!r}.',
+                        job_lib.JobExitCode.NOT_FOUND)
             if len(managed_job_ids) > 1:
                 job_ids_str = ', '.join(
                     str(job_id) for job_id in managed_job_ids)
@@ -776,7 +793,7 @@ def stream_logs(job_id: Optional[int],
             if not follow:
                 # Assume that the log file hasn't been written yet. Since we
                 # aren't following, just return.
-                return ''
+                return '', job_lib.JobExitCode.SUCCESS
 
             job_status = managed_job_state.get_status(job_id)
             if job_status is None:
@@ -787,7 +804,8 @@ def stream_logs(job_id: Optional[int],
                 # point, it never will be. This job may have been submitted
                 # using an old version that did not create the log file, so this
                 # is not considered an exceptional case.
-                return ''
+                return '', job_lib.JobExitCode.from_managed_job_status(
+                    job_status)
 
             time.sleep(log_lib.SKY_LOG_WAITING_GAP_SECONDS)
 
@@ -833,15 +851,17 @@ def stream_logs(job_id: Optional[int],
 
         if follow:
             return ux_utils.finishing_message(
-                f'Job finished (status: {job_status}).')
+                f'Job finished (status: {job_status}).'
+            ), job_lib.JobExitCode.from_managed_job_status(job_status)
 
-        return ''
+        return '', job_lib.JobExitCode.SUCCESS
 
     if job_id is None:
         assert job_name is not None
         job_ids = managed_job_state.get_nonterminal_job_ids_by_name(job_name)
         if not job_ids:
-            return f'No running managed job found with name {job_name!r}.'
+            return (f'No running managed job found with name {job_name!r}.',
+                    job_lib.JobExitCode.NOT_FOUND)
         if len(job_ids) > 1:
             raise ValueError(
                 f'Multiple running jobs found with name {job_name!r}.')
@@ -1167,6 +1187,7 @@ class ManagedJobCodeGen:
       >> codegen = ManagedJobCodeGen.show_jobs(...)
     """
     _PREFIX = textwrap.dedent("""\
+        import sys
         from sky.jobs import utils
         from sky.jobs import state as managed_job_state
         from sky.jobs import constants as managed_job_constants
@@ -1222,9 +1243,17 @@ def stream_logs(cls,
                     follow: bool = True,
                     controller: bool = False) -> str:
         code = textwrap.dedent(f"""\
-        msg = utils.stream_logs({job_id!r}, {job_name!r},
+        result = utils.stream_logs(job_id={job_id!r}, job_name={job_name!r},
                                 follow={follow}, controller={controller})
-        print(msg, flush=True)
+        if managed_job_version < 3:
+            # Versions 2 and older did not return a retcode, so we just print
+            # the result.
+            # TODO: Remove compatibility before 0.12.0
+            print(result, flush=True)
+        else:
+            msg, retcode = result
+            print(msg, flush=True)
+            sys.exit(retcode)
         """)
         return cls._build(code)
 
diff --git a/sky/skylet/constants.py b/sky/skylet/constants.py
index 1075925a238..86ae1ffbace 100644
--- a/sky/skylet/constants.py
+++ b/sky/skylet/constants.py
@@ -93,7 +93,7 @@
 # The version of the lib files that skylet/jobs use. Whenever there is an API
 # change for the job_lib or log_lib, we need to bump this version, so that the
 # user can be notified to update their SkyPilot version on the remote cluster.
-SKYLET_LIB_VERSION = 2
+SKYLET_LIB_VERSION = 3
 SKYLET_VERSION_FILE = '~/.sky/skylet_version'
 
 # `sky jobs dashboard`-related
diff --git a/sky/skylet/job_lib.py b/sky/skylet/job_lib.py
index c758f8f6870..54a2bc2dea4 100644
--- a/sky/skylet/job_lib.py
+++ b/sky/skylet/job_lib.py
@@ -11,6 +11,7 @@
 import signal
 import sqlite3
 import time
+import typing
 from typing import Any, Dict, List, Optional, Sequence
 
 import colorama
@@ -26,6 +27,9 @@
 from sky.utils import message_utils
 from sky.utils import subprocess_utils
 
+if typing.TYPE_CHECKING:
+    from sky.jobs.state import ManagedJobStatus
+
 logger = sky_logging.init_logger(__name__)
 
 _LINUX_NEW_LINE = '\n'
@@ -187,6 +191,77 @@ def colored_str(self) -> str:
         return f'{color}{self.value}{colorama.Style.RESET_ALL}'
 
 
+class JobExitCode(enum.IntEnum):
+    """Job exit code enum.
+
+    These codes are used as return codes for job-related operations and as
+    process exit codes to indicate job status.
+    """
+
+    SUCCESS = 0
+    """The job completed successfully"""
+
+    FAILED = 100
+    """The job failed (due to user code, setup, or driver failure)"""
+
+    NOT_FINISHED = 101
+    """The job has not finished yet"""
+
+    NOT_FOUND = 102
+    """The job was not found"""
+
+    CANCELLED = 103
+    """The job was cancelled by the user"""
+
+    @classmethod
+    def from_job_status(cls, status: Optional[JobStatus]) -> 'JobExitCode':
+        """Convert a job status to an exit code."""
+        if status is None:
+            return cls.NOT_FOUND
+
+        if not status.is_terminal():
+            return cls.NOT_FINISHED
+
+        if status == JobStatus.SUCCEEDED:
+            return cls.SUCCESS
+
+        if status == JobStatus.CANCELLED:
+            return cls.CANCELLED
+
+        if status in JobStatus.user_code_failure_states(
+        ) or status == JobStatus.FAILED_DRIVER:
+            return cls.FAILED
+
+        # Should not hit this case, but included to avoid errors
+        return cls.FAILED
+
+    @classmethod
+    def from_managed_job_status(
+            cls, status: Optional['ManagedJobStatus']) -> 'JobExitCode':
+        """Convert a managed job status to an exit code."""
+        # Import here to avoid circular imports
+        # pylint: disable=import-outside-toplevel
+        from sky.jobs.state import ManagedJobStatus
+
+        if status is None:
+            return cls.NOT_FOUND
+
+        if not status.is_terminal():
+            return cls.NOT_FINISHED
+
+        if status == ManagedJobStatus.SUCCEEDED:
+            return cls.SUCCESS
+
+        if status == ManagedJobStatus.CANCELLED:
+            return cls.CANCELLED
+
+        if status.is_failed():
+            return cls.FAILED
+
+        # Should not hit this case, but included to avoid errors
+        return cls.FAILED
+
+
 # We have two steps for job submissions:
 # 1. Client reserve a job id from the job table by adding a INIT state job.
 # 2. Client updates the job status to PENDING by actually submitting the job's
@@ -938,6 +1013,7 @@ class JobLibCodeGen:
     _PREFIX = [
         'import os',
         'import getpass',
+        'import sys',
         'from sky.skylet import job_lib, log_lib, constants',
     ]
 
@@ -1033,6 +1109,13 @@ def tail_logs(cls,
             f'tail_log_kwargs = {{"job_id": job_id, "log_dir": log_dir, "managed_job_id": {managed_job_id!r}, "follow": {follow}}}',
             f'{_LINUX_NEW_LINE}if getattr(constants, "SKYLET_LIB_VERSION", 1) > 1: tail_log_kwargs["tail"] = {tail}',
             f'{_LINUX_NEW_LINE}log_lib.tail_logs(**tail_log_kwargs)',
+            # After tailing, check the job status and exit with appropriate code
+            'job_status = job_lib.get_status(job_id)',
+            # Backward compatibility for returning exit code: Skylet versions 2
+            # and older did not have JobExitCode, so we use 0 for those versions
+            # TODO: Remove this special handling after 0.10.0.
+            'exit_code = job_lib.JobExitCode.from_job_status(job_status) if getattr(constants, "SKYLET_LIB_VERSION", 1) > 2 else 0',
+            'sys.exit(exit_code)',
         ]
         return cls._build(code)
 
diff --git a/tests/smoke_tests/test_api_server.py b/tests/smoke_tests/test_api_server.py
index 6899cff414e..286f3c8724c 100644
--- a/tests/smoke_tests/test_api_server.py
+++ b/tests/smoke_tests/test_api_server.py
@@ -56,7 +56,7 @@ def test_multi_tenant(generic_cloud: str):
                 f's=$(sky queue -u {name}-1) && echo "$s" && echo "$s" | grep {user_1_name} | grep job-1 | grep SUCCEEDED',
             ]),
             *set_user(user_2, user_2_name, [
-                f'sky exec {name}-1 -n job-2 \'echo "hello" && exit 1\'',
+                f'sky exec {name}-1 -n job-2 \'echo "hello" && exit 1\' || [ $? -eq 100 ]',
                 f's=$(sky queue {name}-1) && echo "$s" && echo "$s" | grep job-2 | grep FAILED | awk \'{{print $1}}\' | grep 2',
                 f's=$(sky queue {name}-1) && echo "$s" && echo "$s" | grep job-1 && exit 1 || true',
                 f's=$(sky queue {name}-1 -u) && echo "$s" && echo "$s" | grep {user_2_name} | grep job-2 | grep FAILED',
diff --git a/tests/smoke_tests/test_basic.py b/tests/smoke_tests/test_basic.py
index 9dccd6a3523..4c828f2e42b 100644
--- a/tests/smoke_tests/test_basic.py
+++ b/tests/smoke_tests/test_basic.py
@@ -657,3 +657,28 @@ def test_kubernetes_context_failover():
             },
         )
         smoke_tests_utils.run_one_test(test)
+
+
+# ---------- Testing Exit Codes for CLI commands ----------
+def test_cli_exit_codes(generic_cloud: str):
+    """Test that CLI commands properly return exit codes based on job success/failure."""
+    name = smoke_tests_utils.get_cluster_name()
+    test = smoke_tests_utils.Test(
+        'cli_exit_codes',
+        [
+            # Test successful job exit code (0)
+            f'sky launch -y -c {name} --cloud {generic_cloud} "echo success" && echo "Exit code: $?"',
+            f'sky logs {name} 1 --status | grep SUCCEEDED',
+
+            # Test that sky logs with successful job returns 0
+            f'sky logs {name} 1 && echo "Exit code: $?"',
+
+            # Test failed job exit code (100)
+            f'sky exec {name} "exit 1" || echo "Command failed with code: $?" | grep "Command failed with code: 100"',
+            f'sky logs {name} 2 --status | grep FAILED',
+            f'sky logs {name} 2 || echo "Job logs exit code: $?" | grep "Job logs exit code: 100"',
+        ],
+        f'sky down -y {name}',
+        timeout=smoke_tests_utils.get_timeout(generic_cloud),
+    )
+    smoke_tests_utils.run_one_test(test)
diff --git a/tests/smoke_tests/test_cluster_job.py b/tests/smoke_tests/test_cluster_job.py
index 163fa70ca05..9d39b062a79 100644
--- a/tests/smoke_tests/test_cluster_job.py
+++ b/tests/smoke_tests/test_cluster_job.py
@@ -636,9 +636,9 @@ def test_multi_node_failure(generic_cloud: str):
     test = smoke_tests_utils.Test(
         'multi_node_failure',
         [
-            f'sky launch -y -c {name} --cloud {generic_cloud} tests/test_yamls/failed_worker_setup.yaml',
+            f'sky launch -y -c {name} --cloud {generic_cloud} tests/test_yamls/failed_worker_setup.yaml || [ $? -eq 100 ]',
             f'sky logs {name} 1 --status | grep FAILED_SETUP',  # Ensure the job setup failed.
-            f'sky exec {name} tests/test_yamls/failed_worker_run.yaml',
+            f'sky exec {name} tests/test_yamls/failed_worker_run.yaml || [ $? -eq 100 ]',
             f'sky logs {name} 2 --status | grep FAILED',  # Ensure the job failed.
             f'sky logs {name} 2 | grep "My hostname:" | wc -l | grep 2',  # Ensure there 2 of the hosts printed their hostname.
         ],
diff --git a/tests/smoke_tests/test_managed_job.py b/tests/smoke_tests/test_managed_job.py
index fd2d1db3504..2c13e864f79 100644
--- a/tests/smoke_tests/test_managed_job.py
+++ b/tests/smoke_tests/test_managed_job.py
@@ -88,6 +88,50 @@ def test_managed_jobs_basic(generic_cloud: str):
     smoke_tests_utils.run_one_test(test)
 
 
+@pytest.mark.managed_jobs
+@pytest.mark.no_nebius  # Autodown and Autostop not supported.
+def test_managed_jobs_cli_exit_codes(generic_cloud: str):
+    """Test that managed jobs CLI commands properly return exit codes based on job success/failure."""
+    name = smoke_tests_utils.get_cluster_name()
+    test = smoke_tests_utils.Test(
+        'managed_jobs_exit_codes',
+        [
+            # Test jobs launch with successful job
+            f'sky jobs launch -y -n jobs-{name} --cloud {generic_cloud} "echo jobs success" && echo "Jobs launch exit code: $?"',
+            smoke_tests_utils.
+            get_cmd_wait_until_managed_job_status_contains_matching_job_name(
+                job_name=f'jobs-{name}',
+                job_status=[sky.ManagedJobStatus.SUCCEEDED],
+                timeout=60),
+
+            # Get job ID from the queue and test logs with successful job
+            f'JOB_ROW=$(sky jobs queue | grep jobs-{name} | head -n1) && '
+            f'echo "$JOB_ROW" && '
+            f'JOB_ID=$(echo "$JOB_ROW" | awk \'{{print $1}}\') && '
+            f'echo "JOB_ID=$JOB_ID" && '
+            f'sky jobs logs $JOB_ID && echo "Jobs logs exit code: $?"',
+
+            # Test jobs launch with failing job
+            f'sky jobs launch -y -n jobs-fail-{name} --cloud {generic_cloud} "exit 1" || echo "Jobs launch failed exit code: $?" | grep "Jobs launch failed exit code: 100"',
+            smoke_tests_utils.
+            get_cmd_wait_until_managed_job_status_contains_matching_job_name(
+                job_name=f'jobs-fail-{name}',
+                job_status=[sky.ManagedJobStatus.FAILED],
+                timeout=60),
+
+            # Get job ID from the queue and test logs with failed job
+            f'JOB_ROW=$(sky jobs queue | grep jobs-fail-{name} | head -n1) && '
+            f'echo "$JOB_ROW" && '
+            f'JOB_ID=$(echo "$JOB_ROW" | awk \'{{print $1}}\') && '
+            f'echo "JOB_ID=$JOB_ID" && '
+            f'sky jobs logs $JOB_ID || echo "Failed jobs logs exit code: $?" | grep "Failed jobs logs exit code: 100"',
+        ],
+        f'sky jobs cancel -y -n jobs-{name}; sky jobs cancel -y -n jobs-fail-{name}',
+        timeout=20 * 60,  # Consistent with other managed jobs tests
+    )
+    smoke_tests_utils.run_one_test(test)
+
+
 @pytest.mark.no_fluidstack  #fluidstack does not support spot instances
 @pytest.mark.no_lambda_cloud  # Lambda Cloud does not support spot instances
 @pytest.mark.no_ibm  # IBM Cloud does not support spot instances


Code review comments:
diff_hunk: @@ -187,6 +191,77 @@ def colored_str(self) -> str:
         return f'{color}{self.value}{colorama.Style.RESET_ALL}'
 
 
+class JobExitCode(enum.IntEnum):
+    """Job exit code enum.
+
+    These codes are used as return codes for job-related operations and as
+    process exit codes to indicate job status.
+    """
+
+    SUCCESS = 0
path: sky/skylet/job_lib.py
text: @user1:
Just to be aligned with our other enum keys?
```suggestion
    SUCCEEDED = 0
```

diff_hunk: @@ -1227,11 +1227,15 @@ def launch(
             clusters=[handle.get_cluster_name()])
         # job_id will be None if no job was submitted (e.g. no entrypoint
         # provided)
+        ret_code = 0
path: sky/cli.py
text: @user1:
minor: How about keep the variable naming the same as other places, i.e. `returncode`?

diff_hunk: @@ -2174,17 +2187,18 @@ def logs(
         # it will return {None: None}.
         if job_id is None:
             click.secho(f'No job found on cluster {cluster!r}.', fg='red')
-            sys.exit(1)
+            sys.exit(job_lib.JobExitCode.NOT_FOUND)
path: sky/cli.py
text: @user1:
minor: how about all the exceptions and exit code in the `exceptions` module? This also helps reduce the chance of circular imports.

@author:
Good idea, refactored