Repository: skypilot-org/skypilot
Pull request #6041 (2025-06-19 05:51:14)

[UI] Add cluster history
<!-- Describe the changes in this PR -->

<img width="1714" alt="Screenshot 2025-06-18 at 10 37 47 PM" src="https://github.com/user-attachments/assets/e4aee045-d716-42e9-982b-76e8e487986c" />

<img width="1717" alt="image" src="https://github.com/user-attachments/assets/2e66f53c-ecca-4729-b56a-b448335b42d5" />


<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [ ] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [ ] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [ ] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)

<!-- CI commands (/-prefixed) can only be triggered by repo members -->



Patch:
diff --git a/docs/source/_static/custom.js b/docs/source/_static/custom.js
index aa03a59e2..d134e0471 100644
--- a/docs/source/_static/custom.js
+++ b/docs/source/_static/custom.js
@@ -30,6 +30,7 @@ document.addEventListener('DOMContentLoaded', () => {
         { selector: '.toctree-l2 > a', text: 'HTTPS Encryption' },
         { selector: '.toctree-l1 > a', text: 'Examples' },
         { selector: '.toctree-l1 > a', text: 'Workspaces: Isolating Teams' },
+        { selector: '.toctree-l1 > a', text: 'External Logging Storage' },
     ];
     newItems.forEach(({ selector, text }) => {
         document.querySelectorAll(selector).forEach((el) => {
diff --git a/sky/client/cli/command.py b/sky/client/cli/command.py
index 5ea5adef8..6e12419a6 100644
--- a/sky/client/cli/command.py
+++ b/sky/client/cli/command.py
@@ -1796,8 +1796,13 @@ def status(verbose: bool, refresh: bool, ip: bool, endpoints: bool,
 @cli.command()
 @flags.config_option(expose_value=False)
 @flags.all_option('Show all cluster information.')
+@click.option('--days',
+              default=30,
+              type=int,
+              help='Show clusters from the last N days. Default is 30 days. '
+              'If set to 0, show all clusters.')
 @usage_lib.entrypoint
-def cost_report(all: bool):  # pylint: disable=redefined-builtin
+def cost_report(all: bool, days: int):  # pylint: disable=redefined-builtin
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Show estimated costs for launched clusters.
 
@@ -1816,13 +1821,22 @@ def cost_report(all: bool):  # pylint: disable=redefined-builtin
 
     - Clusters that were terminated/stopped on the cloud console.
     """
-    cluster_records = sdk.get(sdk.cost_report())
+    days_to_query: Optional[int] = days
+    if days == 0:
+        days_to_query = None
+    cluster_records = sdk.get(sdk.cost_report(days=days_to_query))
 
     normal_cluster_records = []
     controllers = dict()
     for cluster_record in cluster_records:
         cluster_name = cluster_record['name']
-        controller = controller_utils.Controllers.from_name(cluster_name)
+        try:
+            controller = controller_utils.Controllers.from_name(cluster_name)
+        except AssertionError:
+            # There could be some old controller clusters from previous
+            # versions that we should not show in the cost report.
+            logger.debug(f'Cluster {cluster_name} is not a controller cluster.')
+            continue
         if controller is not None:
             controller_name = controller.value.name
             # to display most recent entry for each controller cluster
@@ -1835,10 +1849,14 @@ def cost_report(all: bool):  # pylint: disable=redefined-builtin
     total_cost = status_utils.get_total_cost_of_displayed_records(
         normal_cluster_records, all)
 
-    status_utils.show_cost_report_table(normal_cluster_records, all)
+    status_utils.show_cost_report_table(normal_cluster_records,
+                                        all,
+                                        days=days_to_query)
     for controller_name, cluster_record in controllers.items():
-        status_utils.show_cost_report_table(
-            [cluster_record], all, controller_name=controller_name.capitalize())
+        status_utils.show_cost_report_table([cluster_record],
+                                            all,
+                                            controller_name=controller_name,
+                                            days=days_to_query)
         total_cost += cluster_record['total_cost']
 
     click.echo(f'\n{colorama.Style.BRIGHT}'
diff --git a/sky/client/sdk.py b/sky/client/sdk.py
index 1b8011b8b..59434c309 100644
--- a/sky/client/sdk.py
+++ b/sky/client/sdk.py
@@ -1338,7 +1338,7 @@ def endpoints(
 @usage_lib.entrypoint
 @server_common.check_server_healthy_or_start
 @annotations.client_api
-def cost_report() -> server_common.RequestId:  # pylint: disable=redefined-builtin
+def cost_report(days: Optional[int] = None) -> server_common.RequestId:  # pylint: disable=redefined-builtin
     """Gets all cluster cost reports, including those that have been downed.
 
     The estimated cost column indicates price for the cluster based on the type
@@ -1348,6 +1348,10 @@ def cost_report() -> server_common.RequestId:  # pylint: disable=redefined-built
     cache of the cluster status, and may not be accurate for the cluster with
     autostop/use_spot set or terminated/stopped on the cloud console.
 
+    Args:
+        days: The number of days to get the cost report for. If not provided,
+            the default is 30 days.
+
     Returns:
         The request ID of the cost report request.
 
@@ -1369,8 +1373,10 @@ def cost_report() -> server_common.RequestId:  # pylint: disable=redefined-built
               'total_cost': (float) cost given resources and usage intervals,
             }
     """
-    response = requests.get(f'{server_common.get_server_url()}/cost_report',
-                            cookies=server_common.get_api_cookie_jar())
+    body = payloads.CostReportBody(days=days)
+    response = requests.post(f'{server_common.get_server_url()}/cost_report',
+                             json=json.loads(body.model_dump_json()),
+                             cookies=server_common.get_api_cookie_jar())
     return server_common.get_request_id(response)
 
 
diff --git a/sky/core.py b/sky/core.py
index 115f41f82..50cab3295 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -33,6 +33,7 @@ from sky.utils import admin_policy_utils
 from sky.utils import common
 from sky.utils import common_utils
 from sky.utils import controller_utils
+from sky.utils import resources_utils
 from sky.utils import rich_utils
 from sky.utils import status_lib
 from sky.utils import subprocess_utils
@@ -265,7 +266,7 @@ def endpoints(cluster: str,
         the dictionary will contain all ports:endpoints exposed on the cluster.
 
     Raises:
-        ValueError: if the cluster is not UP or the endpoint is not exposed.
+    ValueError: if the cluster is not UP or the endpoint is not exposed.
         RuntimeError: if the cluster has no ports to be exposed or no endpoints
             are exposed yet.
     """
@@ -276,7 +277,7 @@ def endpoints(cluster: str,
 
 
 @usage_lib.entrypoint
-def cost_report() -> List[Dict[str, Any]]:
+def cost_report(days: Optional[int] = None) -> List[Dict[str, Any]]:
     # NOTE(dev): Keep the docstring consistent between the Python API and CLI.
     """Get all cluster cost reports, including those that have been downed.
 
@@ -294,6 +295,13 @@ def cost_report() -> List[Dict[str, Any]]:
             'cluster_hash': (str) unique hash identifying cluster,
             'usage_intervals': (List[Tuple[int, int]]) cluster usage times,
             'total_cost': (float) cost given resources and usage intervals,
+            'cloud': (str) cloud of the cluster,
+            'region': (str) region of the cluster,
+            'cpus': (str) number of vCPUs of the cluster,
+            'memory': (str) memory of the cluster,
+            'accelerators': (str) accelerators of the cluster,
+            'resources_str': (str) resources string of the cluster,
+            'resources_str_full': (str) full resources string of the cluster,
         }
 
     The estimated cost column indicates price for the cluster based on the type
@@ -303,27 +311,92 @@ def cost_report() -> List[Dict[str, Any]]:
     cache of the cluster status, and may not be accurate for the cluster with
     autostop/use_spot set or terminated/stopped on the cloud console.
 
+    Args:
+        days: Number of days to look back from now. Active clusters are always
+            included. Historical clusters are only included if they were last
+            used within the past 'days' days. Defaults to 30 days.
+
     Returns:
         A list of dicts, with each dict containing the cost information of a
         cluster.
     """
-    cluster_reports = global_user_state.get_clusters_from_history()
+    if days is None:
+        days = constants.COST_REPORT_DEFAULT_DAYS
+
+    cluster_reports = global_user_state.get_clusters_from_history(days=days)
+    logger.debug(
+        f'{len(cluster_reports)} clusters found from history with {days} days.')
+
+    def _process_cluster_report(
+            cluster_report: Dict[str, Any]) -> Dict[str, Any]:
+        """Process cluster report by calculating cost and adding fields."""
+        # Make a copy to avoid modifying the original
+        report = cluster_report.copy()
+
+        def get_total_cost(cluster_report: dict) -> float:
+            duration = cluster_report['duration']
+            launched_nodes = cluster_report['num_nodes']
+            launched_resources = cluster_report['resources']
+
+            cost = (launched_resources.get_cost(duration) * launched_nodes)
+            return cost
+
+        def _update_record_with_resources(record: Dict[str, Any]) -> None:
+            """Add resource fields for dashboard compatibility."""
+            if record is None:
+                return
+            resources = record.get('resources')
+            if resources is None:
+                return
+            fields = ['cloud', 'region', 'cpus', 'memory', 'accelerators']
+            for field in fields:
+                try:
+                    record[field] = str(getattr(resources, field))
+                except Exception as e:  # pylint: disable=broad-except
+                    # Ok to skip the fields as this is just for display
+                    # purposes.
+                    logger.debug(f'Failed to get resources.{field} for cluster '
+                                 f'{record["name"]}: {str(e)}')
+                    record[field] = None
+
+            # Add resources_str and resources_str_full for dashboard
+            # compatibility
+            num_nodes = record.get('num_nodes', 1)
+            try:
+                resource_str_simple = resources_utils.format_resource(
+                    resources, simplify=True)
+                resource_str_full = resources_utils.format_resource(
+                    resources, simplify=False)
+                record['resources_str'] = f'{num_nodes}x{resource_str_simple}'
+                record[
+                    'resources_str_full'] = f'{num_nodes}x{resource_str_full}'
+            except Exception as e:  # pylint: disable=broad-except
+                logger.debug(f'Failed to get resources_str for cluster '
+                             f'{record["name"]}: {str(e)}')
+                for field in fields:
+                    record[field] = None
+                record['resources_str'] = '-'
+                record['resources_str_full'] = '-'
+
+        try:
+            report['total_cost'] = get_total_cost(report)
+        except Exception as e:  # pylint: disable=broad-except
+            # Ok to skip the total cost as this is just for display purposes.
+            logger.warning(f'Failed to get total cost for cluster '
+                           f'{report["name"]}: {str(e)}')
+            report['total_cost'] = 0.0
 
-    def get_total_cost(cluster_report: dict) -> float:
-        duration = cluster_report['duration']
-        launched_nodes = cluster_report['num_nodes']
-        launched_resources = cluster_report['resources']
+        _update_record_with_resources(report)
+        return report
 
-        cost = (launched_resources.get_cost(duration) * launched_nodes)
-        return cost
+    # Process clusters in parallel
+    if not cluster_reports:
+        return []
 
-    for cluster_report in cluster_reports:
-        cluster_report['total_cost'] = get_total_cost(cluster_report)
-        cluster_report['cloud'] = str(cluster_report['resources'].cloud)
-        cluster_report['accelerators'] = cluster_report[
-            'resources'].accelerators
+    processed_reports = subprocess_utils.run_in_parallel(
+        _process_cluster_report, cluster_reports)
 
-    return cluster_reports
+    return processed_reports
 
 
 def _start(
diff --git a/sky/dashboard/src/components/clusters.jsx b/sky/dashboard/src/components/clusters.jsx
index ea1c1032f..e58fac172 100755
--- a/sky/dashboard/src/components/clusters.jsx
+++ b/sky/dashboard/src/components/clusters.jsx
@@ -31,7 +31,7 @@ import {
   TableBody,
   TableCell,
 } from '@/components/ui/table';
-import { getClusters } from '@/data/connectors/clusters';
+import { getClusters, getClusterHistory } from '@/data/connectors/clusters';
 import { getWorkspaces } from '@/data/connectors/workspaces';
 import { getUsers } from '@/data/connectors/users';
 import { sortData } from '@/data/utils';
@@ -127,6 +127,40 @@ const formatUserDisplay = (username, userId) => {
   return usernameBase;
 };
 
+// Helper function to format duration in a human-readable way
+const formatDuration = (durationSeconds) => {
+  if (!durationSeconds || durationSeconds === 0) {
+    return '-';
+  }
+
+  // Convert to a whole number if it's a float
+  durationSeconds = Math.floor(durationSeconds);
+
+  const units = [
+    { value: 31536000, label: 'y' }, // years (365 days)
+    { value: 2592000, label: 'mo' }, // months (30 days)
+    { value: 86400, label: 'd' }, // days
+    { value: 3600, label: 'h' }, // hours
+    { value: 60, label: 'm' }, // minutes
+    { value: 1, label: 's' }, // seconds
+  ];
+
+  let remaining = durationSeconds;
+  let result = '';
+  let count = 0;
+
+  for (const unit of units) {
+    if (remaining >= unit.value && count < 2) {
+      const value = Math.floor(remaining / unit.value);
+      result += `${value}${unit.label} `;
+      remaining %= unit.value;
+      count++;
+    }
+  }
+
+  return result.trim() || '0s';
+};
+
 export function Clusters() {
   const router = useRouter();
   const [loading, setLoading] = useState(false);
@@ -139,6 +173,7 @@ export function Clusters() {
   const [nameFilter, setNameFilter] = useState('');
   const [workspaces, setWorkspaces] = useState([]);
   const [users, setUsers] = useState([]);
+  const [showHistory, setShowHistory] = useState(false); // 'active' or 'history'
   const isMobile = useMobile();
 
   // Handle URL query parameters for workspace and user filtering
@@ -313,6 +348,7 @@ export function Clusters() {
   const handleRefresh = () => {
     // Invalidate cache to ensure fresh data is fetched
     dashboardCache.invalidate(getClusters);
+    dashboardCache.invalidate(getClusterHistory);
     dashboardCache.invalidate(getWorkspaces);
     dashboardCache.invalidate(getUsers);
 
@@ -331,13 +367,37 @@ export function Clusters() {
           >
             Sky Clusters
           </Link>
+          <div className="flex items-center ml-6 space-x-3">
+            <label className="flex items-center cursor-pointer">
+              <input
+                type="checkbox"
+                checked={showHistory}
+                onChange={(e) => setShowHistory(e.target.checked)}
+                className="sr-only"
+              />
+              <div
+                className={`relative inline-flex h-5 w-9 items-center rounded-full transition-colors ${
+                  showHistory ? 'bg-sky-600' : 'bg-gray-300'
+                }`}
+              >
+                <span
+                  className={`inline-block h-3 w-3 transform rounded-full bg-white transition-transform ${
+                    showHistory ? 'translate-x-5' : 'translate-x-1'
+                  }`}
+                />
+              </div>
+              <span className="ml-2 text-sm text-gray-700">
+                Show History (last 30 days)
+              </span>
+            </label>
+          </div>
           <div className="relative ml-4 mr-2">
             <input
               type="text"
               placeholder="Filter by cluster name"
               value={nameFilter}
               onChange={(e) => handleNameFilterChange(e.target.value)}
-              className="h-8 w-48 px-3 pr-8 text-sm border border-gray-300 rounded-md focus:ring-1 focus:ring-sky-500 focus:border-sky-500 outline-none"
+              className="h-8 w-32 sm:w-48 px-3 pr-8 text-sm border border-gray-300 rounded-md focus:ring-1 focus:ring-sky-500 focus:border-sky-500 outline-none"
             />
             {nameFilter && (
               <button
@@ -426,6 +486,7 @@ export function Clusters() {
         workspaceFilter={workspaceFilter}
         userFilter={userFilter}
         nameFilter={nameFilter}
+        showHistory={showHistory}
         onOpenSSHModal={(cluster) => {
           setSelectedCluster(cluster);
           setIsSSHModalOpen(true);
@@ -459,6 +520,7 @@ export function ClusterTable({
   workspaceFilter,
   userFilter,
   nameFilter,
+  showHistory,
   onOpenSSHModal,
   onOpenVSCodeModal,
 }) {
@@ -475,12 +537,51 @@ export function ClusterTable({
   const fetchData = React.useCallback(async () => {
     setLoading(true);
     setLocalLoading(true);
-    const initialData = await dashboardCache.get(getClusters);
-    setData(initialData);
+
+    try {
+      const activeClusters = await dashboardCache.get(getClusters);
+
+      if (showHistory) {
+        const historyClusters = await dashboardCache.get(getClusterHistory);
+        // Mark clusters as active or historical for UI distinction
+        const markedActiveClusters = activeClusters.map((cluster) => ({
+          ...cluster,
+          isHistorical: false,
+        }));
+        const markedHistoryClusters = historyClusters.map((cluster) => ({
+          ...cluster,
+          isHistorical: true,
+        }));
+        // Combine and remove duplicates (prefer active over historical)
+        const combinedData = [...markedActiveClusters];
+        markedHistoryClusters.forEach((histCluster) => {
+          const existsInActive = activeClusters.some(
+            (activeCluster) =>
+              (activeCluster.cluster || activeCluster.name) ===
+              (histCluster.cluster || histCluster.name)
+          );
+          if (!existsInActive) {
+            combinedData.push(histCluster);
+          }
+        });
+        setData(combinedData);
+      } else {
+        // Mark active clusters for consistency
+        const markedActiveClusters = activeClusters.map((cluster) => ({
+          ...cluster,
+          isHistorical: false,
+        }));
+        setData(markedActiveClusters);
+      }
+    } catch (error) {
+      console.error('Error fetching cluster data:', error);
+      setData([]);
+    }
+
     setLoading(false);
     setLocalLoading(false);
     setIsInitialLoad(false);
-  }, [setLoading]);
+  }, [setLoading, showHistory]);
 
   // Use useMemo to compute sorted data
   const sortedData = React.useMemo(() => {
@@ -575,156 +676,181 @@ export function ClusterTable({
   return (
     <div>
       <Card>
-        <Table>
-          <TableHeader>
-            <TableRow>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('status')}
-              >
-                Status{getSortDirection('status')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('cluster')}
-              >
-                Cluster{getSortDirection('cluster')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('user')}
-              >
-                User{getSortDirection('user')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('workspace')}
-              >
-                Workspace{getSortDirection('workspace')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('infra')}
-              >
-                Infra{getSortDirection('infra')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('resources_str')}
-              >
-                Resources{getSortDirection('resources_str')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('time')}
-              >
-                Started{getSortDirection('time')}
-              </TableHead>
-              <TableHead
-                className="sortable whitespace-nowrap"
-                onClick={() => requestSort('autostop')}
-              >
-                Autostop{getSortDirection('autostop')}
-              </TableHead>
-              <TableHead>Actions</TableHead>
-            </TableRow>
-          </TableHeader>
-
-          <TableBody>
-            {loading && isInitialLoad ? (
+        <div className="overflow-x-auto">
+          <Table className="min-w-full">
+            <TableHeader>
               <TableRow>
-                <TableCell
-                  colSpan={9}
-                  className="text-center py-6 text-gray-500"
+                <TableHead
+                  className="sortable whitespace-nowrap"
+                  onClick={() => requestSort('status')}
                 >
-                  <div className="flex justify-center items-center">
-                    <CircularProgress size={20} className="mr-2" />
-                    <span>Loading...</span>
-                  </div>
-                </TableCell>
-              </TableRow>
-            ) : paginatedData.length > 0 ? (
-              paginatedData.map((item, index) => {
-                return (
-                  <TableRow key={index}>
-                    <TableCell>
-                      <StatusBadge status={item.status} />
-                    </TableCell>
-                    <TableCell>
-                      <Link
-                        href={`/clusters/${item.cluster}`}
-                        className="text-blue-600"
-                      >
-                        {item.cluster}
-                      </Link>
-                    </TableCell>
-                    <TableCell>{item.user}</TableCell>
-                    <TableCell>
-                      <Link
-                        href="/workspaces"
-                        className="text-blue-600 hover:underline"
-                      >
-                        {item.workspace || 'default'}
-                      </Link>
-                    </TableCell>
-                    <TableCell>
-                      <NonCapitalizedTooltip
-                        content={item.full_infra || item.infra}
-                        className="text-sm text-muted-foreground"
-                      >
-                        <span>
-                          <Link
-                            href="/infra"
-                            className="text-blue-600 hover:underline"
-                          >
-                            {item.cloud}
-                          </Link>
-                          {item.infra.includes('(') && (
-                            <span>
-                              {' ' +
-                                item.infra.substring(item.infra.indexOf('('))}
-                            </span>
-                          )}
-                        </span>
-                      </NonCapitalizedTooltip>
-                    </TableCell>
-                    <TableCell>
-                      <NonCapitalizedTooltip
-                        content={item.resources_str_full || item.resources_str}
-                        className="text-sm text-muted-foreground"
-                      >
-                        <span>{item.resources_str}</span>
-                      </NonCapitalizedTooltip>
-                    </TableCell>
-                    <TableCell>
-                      <TimestampWithTooltip date={item.time} />
-                    </TableCell>
-                    <TableCell>
-                      {formatAutostop(item.autostop, item.to_down)}
-                    </TableCell>
-                    <TableCell className="text-left">
-                      <Status2Actions
-                        cluster={item.cluster}
-                        status={item.status}
-                        onOpenSSHModal={onOpenSSHModal}
-                        onOpenVSCodeModal={onOpenVSCodeModal}
-                      />
-                    </TableCell>
-                  </TableRow>
-                );
-              })
-            ) : (
-              <TableRow>
-                <TableCell
-                  colSpan={9}
-                  className="text-center py-6 text-gray-500"
+                  Status{getSortDirection('status')}
+                </TableHead>
+                <TableHead
+                  className="sortable whitespace-nowrap"
+                  onClick={() => requestSort('cluster')}
+                >
+                  Cluster{getSortDirection('cluster')}
+                </TableHead>
+                <TableHead
+                  className="sortable whitespace-nowrap hidden sm:table-cell"
+                  onClick={() => requestSort('user')}
+                >
+                  User{getSortDirection('user')}
+                </TableHead>
+                <TableHead
+                  className="sortable whitespace-nowrap hidden md:table-cell"
+                  onClick={() => requestSort('workspace')}
+                >
+                  Workspace{getSortDirection('workspace')}
+                </TableHead>
+                <TableHead
+                  className="sortable whitespace-nowrap hidden lg:table-cell"
+                  onClick={() => requestSort('infra')}
+                >
+                  Infra{getSortDirection('infra')}
+                </TableHead>
+                <TableHead
+                  className="sortable whitespace-nowrap hidden xl:table-cell"
+                  onClick={() => requestSort('resources_str')}
                 >
-                  No active clusters
-                </TableCell>
+                  Resources{getSortDirection('resources_str')}
+                </TableHead>
+                <TableHead
+                  className="sortable whitespace-nowrap"
+                  onClick={() => requestSort('time')}
+                >
+                  Started{getSortDirection('time')}
+                </TableHead>
+                {showHistory && (
+                  <TableHead
+                    className="sortable whitespace-nowrap hidden lg:table-cell"
+                    onClick={() => requestSort('duration')}
+                  >
+                    Duration{getSortDirection('duration')}
+                  </TableHead>
+                )}
+                <TableHead
+                  className="sortable whitespace-nowrap hidden md:table-cell"
+                  onClick={() => requestSort('autostop')}
+                >
+                  Autostop{getSortDirection('autostop')}
+                </TableHead>
+                <TableHead className="sticky right-0 bg-white">
+                  Actions
+                </TableHead>
               </TableRow>
-            )}
-          </TableBody>
-        </Table>
+            </TableHeader>
+
+            <TableBody>
+              {loading && isInitialLoad ? (
+                <TableRow>
+                  <TableCell
+                    colSpan={9}
+                    className="text-center py-6 text-gray-500"
+                  >
+                    <div className="flex justify-center items-center">
+                      <CircularProgress size={20} className="mr-2" />
+                      <span>Loading...</span>
+                    </div>
+                  </TableCell>
+                </TableRow>
+              ) : paginatedData.length > 0 ? (
+                paginatedData.map((item, index) => {
+                  return (
+                    <TableRow key={index}>
+                      <TableCell>
+                        <StatusBadge status={item.status} />
+                      </TableCell>
+                      <TableCell>
+                        <Link
+                          href={`/clusters/${item.isHistorical ? item.cluster_hash : item.cluster || item.name}`}
+                          className="text-blue-600"
+                        >
+                          {item.cluster || item.name}
+                        </Link>
+                      </TableCell>
+                      <TableCell className="hidden sm:table-cell">
+                        {item.user}
+                      </TableCell>
+                      <TableCell className="hidden md:table-cell">
+                        <Link
+                          href="/workspaces"
+                          className="text-blue-600 hover:underline"
+                        >
+                          {item.workspace || 'default'}
+                        </Link>
+                      </TableCell>
+                      <TableCell className="hidden lg:table-cell">
+                        <NonCapitalizedTooltip
+                          content={item.full_infra || item.infra}
+                          className="text-sm text-muted-foreground"
+                        >
+                          <span>
+                            <Link
+                              href="/infra"
+                              className="text-blue-600 hover:underline"
+                            >
+                              {item.cloud}
+                            </Link>
+                            {item.infra.includes('(') && (
+                              <span>
+                                {' ' +
+                                  item.infra.substring(item.infra.indexOf('('))}
+                              </span>
+                            )}
+                          </span>
+                        </NonCapitalizedTooltip>
+                      </TableCell>
+                      <TableCell className="hidden xl:table-cell">
+                        <NonCapitalizedTooltip
+                          content={
+                            item.resources_str_full || item.resources_str
+                          }
+                          className="text-sm text-muted-foreground"
+                        >
+                          <span>{item.resources_str}</span>
+                        </NonCapitalizedTooltip>
+                      </TableCell>
+                      <TableCell>
+                        <TimestampWithTooltip date={item.time} />
+                      </TableCell>
+                      {showHistory && (
+                        <TableCell className="hidden lg:table-cell">
+                          {formatDuration(item.duration)}
+                        </TableCell>
+                      )}
+                      <TableCell className="hidden md:table-cell">
+                        {item.isHistorical
+                          ? '-'
+                          : formatAutostop(item.autostop, item.to_down)}
+                      </TableCell>
+                      <TableCell className="text-left sticky right-0 bg-white">
+                        {!item.isHistorical && (
+                          <Status2Actions
+                            cluster={item.cluster}
+                            status={item.status}
+                            onOpenSSHModal={onOpenSSHModal}
+                            onOpenVSCodeModal={onOpenVSCodeModal}
+                          />
+                        )}
+                      </TableCell>
+                    </TableRow>
+                  );
+                })
+              ) : (
+                <TableRow>
+                  <TableCell
+                    colSpan={9}
+                    className="text-center py-6 text-gray-500"
+                  >
+                    {showHistory ? 'No clusters found' : 'No active clusters'}
+                  </TableCell>
+                </TableRow>
+              )}
+            </TableBody>
+          </Table>
+        </div>
       </Card>
 
       {/* Pagination controls */}
diff --git a/sky/dashboard/src/components/utils.jsx b/sky/dashboard/src/components/utils.jsx
index 44588824e..3d9051f14 100644
--- a/sky/dashboard/src/components/utils.jsx
+++ b/sky/dashboard/src/components/utils.jsx
@@ -205,6 +205,8 @@ export function formatDuration(durationInSeconds) {
   durationInSeconds = Math.floor(durationInSeconds);
 
   const units = [
+    { value: 31536000, label: 'y' }, // years (365 days)
+    { value: 2592000, label: 'mo' }, // months (30 days)
     { value: 86400, label: 'd' }, // days
     { value: 3600, label: 'h' }, // hours
     { value: 60, label: 'm' }, // minutes
diff --git a/sky/dashboard/src/data/connectors/clusters.jsx b/sky/dashboard/src/data/connectors/clusters.jsx
index 261de19b8..9db366ac3 100644
--- a/sky/dashboard/src/data/connectors/clusters.jsx
+++ b/sky/dashboard/src/data/connectors/clusters.jsx
@@ -106,6 +106,71 @@ export async function getClusters({ clusterNames = null } = {}) {
   }
 }
 
+export async function getClusterHistory() {
+  try {
+    const history = await apiClient.fetch('/cost_report', {
+      days: 30,
+    });
+
+    console.log('Raw cluster history data:', history); // Debug log
+
+    const historyData = history.map((cluster) => {
+      // Get cloud name from resources if available
+      let cloud = 'Unknown';
+      if (cluster.cloud) {
+        cloud = cluster.cloud;
+      } else if (cluster.resources && cluster.resources.cloud) {
+        cloud = cluster.resources.cloud;
+      }
+
+      // Get user name - need to look up from user_hash if needed
+      let user_name = cluster.user_name || '-';
+
+      // Extract resource info
+
+      return {
+        status: cluster.status
+          ? clusterStatusMap[cluster.status]
+          : 'TERMINATED',
+        cluster: cluster.name,
+        user: user_name,
+        user_hash: cluster.user_hash,
+        cloud: cloud,
+        region: '',
+        infra: cloud,
+        full_infra: cloud,
+        resources_str: cluster.resources_str,
+        resources_str_full: cluster.resources_str_full,
+        time: cluster.launched_at ? new Date(cluster.launched_at * 1000) : null,
+        num_nodes: cluster.num_nodes || 1,
+        duration: cluster.duration,
+        total_cost: cluster.total_cost,
+        workspace: cluster.workspace || 'default',
+        autostop: -1,
+        to_down: false,
+        cluster_hash: cluster.cluster_hash,
+        usage_intervals: cluster.usage_intervals,
+        command: cluster.last_creation_command || '',
+        task_yaml: cluster.last_creation_yaml || '{}',
+        events: [
+          {
+            time: cluster.launched_at
+              ? new Date(cluster.launched_at * 1000)
+              : new Date(),
+            event: 'Cluster created.',
+          },
+        ],
+      };
+    });
+
+    console.log('Processed cluster history data:', historyData); // Debug log
+    return historyData;
+  } catch (error) {
+    console.error('Error fetching cluster history:', error);
+    return [];
+  }
+}
+
 export async function streamClusterJobLogs({
   clusterName,
   jobId,
diff --git a/sky/dashboard/src/lib/cache-preloader.js b/sky/dashboard/src/lib/cache-preloader.js
index 10d2ad133..2b4327214 100644
--- a/sky/dashboard/src/lib/cache-preloader.js
+++ b/sky/dashboard/src/lib/cache-preloader.js
@@ -2,7 +2,7 @@
 // This utility manages background preloading of cache data to improve page switching performance
 
 import dashboardCache from './cache';
-import { getClusters } from '@/data/connectors/clusters';
+import { getClusters, getClusterHistory } from '@/data/connectors/clusters';
 import { getManagedJobs } from '@/data/connectors/jobs';
 import { getWorkspaces, getEnabledClouds } from '@/data/connectors/workspaces';
 import { getUsers } from '@/data/connectors/users';
@@ -15,6 +15,7 @@ export const DASHBOARD_CACHE_FUNCTIONS = {
   // Base functions used across multiple pages (no arguments)
   base: {
     getClusters: { fn: getClusters, args: [] },
+    getClusterHistory: { fn: getClusterHistory, args: [] },
     getManagedJobs: { fn: getManagedJobs, args: [{ allUsers: true }] },
     getWorkspaces: { fn: getWorkspaces, args: [] },
     getUsers: { fn: getUsers, args: [] },
@@ -28,7 +29,7 @@ export const DASHBOARD_CACHE_FUNCTIONS = {
 
   // Page-specific function requirements
   pages: {
-    clusters: ['getClusters', 'getWorkspaces', 'getUsers'],
+    clusters: ['getClusters', 'getClusterHistory', 'getWorkspaces', 'getUsers'],
     jobs: ['getManagedJobs', 'getClusters', 'getWorkspaces', 'getUsers'],
     infra: ['getInfraData', 'getClusters', 'getManagedJobs'],
     workspaces: [
diff --git a/sky/dashboard/src/pages/clusters/[cluster].js b/sky/dashboard/src/pages/clusters/[cluster].js
index bce871679..2f00077c4 100644
--- a/sky/dashboard/src/pages/clusters/[cluster].js
+++ b/sky/dashboard/src/pages/clusters/[cluster].js
@@ -7,7 +7,11 @@ import Link from 'next/link';
 import { Status2Actions } from '@/components/clusters';
 import { StatusBadge } from '@/components/elements/StatusBadge';
 import { Card } from '@/components/ui/card';
-import { useClusterDetails } from '@/data/connectors/clusters';
+import {
+  useClusterDetails,
+  getClusterHistory,
+} from '@/data/connectors/clusters';
+import dashboardCache from '@/lib/cache';
 import {
   RotateCwIcon,
   ChevronDownIcon,
@@ -58,6 +62,9 @@ function ClusterDetails() {
   const [isInitialLoad, setIsInitialLoad] = useState(true);
   const [isSSHModalOpen, setIsSSHModalOpen] = useState(false);
   const [isVSCodeModalOpen, setIsVSCodeModalOpen] = useState(false);
+  const [historyData, setHistoryData] = useState(null);
+  const [isHistoricalCluster, setIsHistoricalCluster] = useState(false);
+  const [historyLoading, setHistoryLoading] = useState(false);
   const isMobile = useMobile();
   const {
     clusterData,
@@ -76,6 +83,34 @@ function ClusterDetails() {
     }
   }, [clusterDetailsLoading, isInitialLoad]);
 
+  // Check for historical cluster if active cluster is not found
+  React.useEffect(() => {
+    const checkHistoryCluster = async () => {
+      if (!cluster || clusterDetailsLoading || clusterData) return;
+
+      setHistoryLoading(true);
+      try {
+        const historyData = await dashboardCache.get(getClusterHistory);
+        const foundHistoryCluster = historyData.find(
+          (c) => c.cluster_hash === cluster || c.cluster === cluster
+        );
+        if (foundHistoryCluster) {
+          setHistoryData(foundHistoryCluster);
+          setIsHistoricalCluster(true);
+        }
+      } catch (error) {
+        console.error('Error fetching cluster history:', error);
+      } finally {
+        setHistoryLoading(false);
+      }
+    };
+
+    // Only check history if we've finished loading and no active cluster found
+    if (!clusterDetailsLoading && !clusterData) {
+      checkHistoryCluster();
+    }
+  }, [cluster, clusterDetailsLoading, clusterData]);
+
   const handleManualRefresh = async () => {
     setIsRefreshing(true);
     await refreshData();
@@ -155,7 +190,7 @@ function ClusterDetails() {
           </div>
         </div>
 
-        {clusterDetailsLoading && isInitialLoad ? (
+        {(clusterDetailsLoading && isInitialLoad) || historyLoading ? (
           <div className="flex justify-center items-center py-12">
             <CircularProgress size={24} className="mr-2" />
             <span className="text-gray-500">Loading cluster details...</span>
@@ -166,8 +201,23 @@ function ClusterDetails() {
             clusterJobData={clusterJobData}
             clusterJobsLoading={clusterJobsLoading}
             refreshClusterJobsOnly={refreshClusterJobsOnly}
+            isHistoricalCluster={false}
+          />
+        ) : isHistoricalCluster && historyData ? (
+          <ActiveTab
+            clusterData={historyData}
+            clusterJobData={[]}
+            clusterJobsLoading={false}
+            refreshClusterJobsOnly={() => {}}
+            isHistoricalCluster={true}
           />
-        ) : null}
+        ) : (
+          <div className="flex justify-center items-center py-12">
+            <span className="text-gray-500">
+              Cluster not found in active clusters or history.
+            </span>
+          </div>
+        )}
 
         {/* SSH Instructions Modal */}
         <SSHInstructionsModal
@@ -192,6 +242,7 @@ function ActiveTab({
   clusterJobData,
   clusterJobsLoading,
   refreshClusterJobsOnly,
+  isHistoricalCluster = false,
 }) {
   const [isYamlExpanded, setIsYamlExpanded] = useState(false);
   const [isCopied, setIsCopied] = useState(false);
@@ -203,7 +254,9 @@ function ActiveTab({
 
   const copyYamlToClipboard = async () => {
     try {
-      const formattedYaml = formatYaml(clusterData.task_yaml);
+      const yamlContent =
+        clusterData.task_yaml || clusterData.last_creation_yaml;
+      const formattedYaml = formatYaml(yamlContent);
       await navigator.clipboard.writeText(formattedYaml);
       setIsCopied(true);
       setTimeout(() => setIsCopied(false), 2000); // Reset after 2 seconds
@@ -214,7 +267,8 @@ function ActiveTab({
 
   const copyCommandToClipboard = async () => {
     try {
-      await navigator.clipboard.writeText(clusterData.command);
+      const command = clusterData.command || clusterData.last_creation_command;
+      await navigator.clipboard.writeText(command);
       setIsCommandCopied(true);
       setTimeout(() => setIsCommandCopied(false), 2000); // Reset after 2 seconds
     } catch (err) {
@@ -222,7 +276,57 @@ function ActiveTab({
     }
   };
 
-  const hasCreationArtifacts = clusterData?.command || clusterData?.task_yaml;
+  const hasCreationArtifacts =
+    clusterData?.last_creation_command ||
+    clusterData?.last_creation_yaml ||
+    clusterData?.command ||
+    clusterData?.task_yaml;
+
+  // Helper functions for historical clusters
+  const formatDuration = (durationSeconds) => {
+    if (!durationSeconds || durationSeconds === 0) {
+      return '-';
+    }
+
+    // Convert to a whole number if it's a float
+    durationSeconds = Math.floor(durationSeconds);
+
+    const units = [
+      { value: 31536000, label: 'y' }, // years (365 days)
+      { value: 2592000, label: 'mo' }, // months (30 days)
+      { value: 86400, label: 'd' }, // days
+      { value: 3600, label: 'h' }, // hours
+      { value: 60, label: 'm' }, // minutes
+      { value: 1, label: 's' }, // seconds
+    ];
+
+    let remaining = durationSeconds;
+    let result = '';
+    let count = 0;
+
+    for (const unit of units) {
+      if (remaining >= unit.value && count < 2) {
+        const value = Math.floor(remaining / unit.value);
+        result += `${value}${unit.label} `;
+        remaining %= unit.value;
+        count++;
+      }
+    }
+
+    return result.trim() || '0s';
+  };
+
+  const formatCost = (cost) => {
+    if (cost === null || cost === undefined || cost === 0) {
+      return '-';
+    }
+    // Convert to number and check if it's valid
+    const numericCost = Number(cost);
+    if (isNaN(numericCost)) {
+      return '-';
+    }
+    return `$${numericCost.toFixed(2)}`;
+  };
 
   return (
     <div>
@@ -230,7 +334,9 @@ function ActiveTab({
       <div className="mb-6">
         <Card>
           <div className="flex items-center justify-between px-4 pt-4">
-            <h3 className="text-lg font-semibold">Details</h3>
+            <h3 className="text-lg font-semibold">
+              {isHistoricalCluster ? 'Historical Cluster Details' : 'Details'}
+            </h3>
           </div>
           <div className="p-4">
             <div className="grid grid-cols-2 gap-6">
@@ -246,16 +352,22 @@ function ActiveTab({
                 <div className="text-gray-600 font-medium text-base">
                   Cluster
                 </div>
-                <div className="text-base mt-1">{clusterData.cluster}</div>
+                <div className="text-base mt-1">
+                  {clusterData.cluster || clusterData.name}
+                </div>
               </div>
               <div>
                 <div className="text-gray-600 font-medium text-base">User</div>
                 <div className="text-base mt-1">{clusterData.user}</div>
               </div>
               <div>
-                <div className="text-gray-600 font-medium text-base">Infra</div>
+                <div className="text-gray-600 font-medium text-base">
+                  {isHistoricalCluster ? 'Cloud' : 'Infra'}
+                </div>
                 <div className="text-base mt-1">
-                  {clusterData.infra ? (
+                  {isHistoricalCluster ? (
+                    clusterData.cloud || 'N/A'
+                  ) : clusterData.infra ? (
                     <NonCapitalizedTooltip
                       content={clusterData.full_infra || clusterData.infra}
                       className="text-sm text-muted-foreground"
@@ -303,58 +415,90 @@ function ActiveTab({
                     : 'N/A'}
                 </div>
               </div>
-              <div>
-                <div className="text-gray-600 font-medium text-base">
-                  Autostop
-                </div>
-                <div className="text-base mt-1">
-                  {formatAutostop(clusterData.autostop, clusterData.to_down)}
+
+              {/* Show duration and cost for historical clusters */}
+              {isHistoricalCluster ? (
+                <>
+                  <div>
+                    <div className="text-gray-600 font-medium text-base">
+                      Duration
+                    </div>
+                    <div className="text-base mt-1">
+                      {formatDuration(clusterData.duration)}
+                    </div>
+                  </div>
+                  <div>
+                    <div className="text-gray-600 font-medium text-base">
+                      Cost
+                    </div>
+                    <div className="text-base mt-1">
+                      {formatCost(clusterData.total_cost)}
+                    </div>
+                  </div>
+                </>
+              ) : (
+                <div>
+                  <div className="text-gray-600 font-medium text-base">
+                    Autostop
+                  </div>
+                  <div className="text-base mt-1">
+                    {formatAutostop(clusterData.autostop, clusterData.to_down)}
+                  </div>
                 </div>
-              </div>
+              )}
 
               {/* Created by section - spans both columns */}
               {hasCreationArtifacts && (
                 <div className="col-span-2">
-                  <div className="flex items-center">
-                    <div className="text-gray-600 font-medium text-base">
-                      Entrypoint
-                    </div>
-                    {clusterData.command && (
-                      <Tooltip
-                        content={isCommandCopied ? 'Copied!' : 'Copy command'}
-                        className="text-muted-foreground"
-                      >
-                        <button
-                          onClick={copyCommandToClipboard}
-                          className="flex items-center text-gray-500 hover:text-gray-700 transition-colors duration-200 p-1 ml-2"
+                  {(clusterData.command ||
+                    clusterData.last_creation_command) && (
+                    <div className="flex items-center">
+                      <div className="text-gray-600 font-medium text-base">
+                        Entrypoint
+                      </div>
+                      {clusterData.command && (
+                        <Tooltip
+                          content={isCommandCopied ? 'Copied!' : 'Copy command'}
+                          className="text-muted-foreground"
                         >
-                          {isCommandCopied ? (
-                            <CheckIcon className="w-4 h-4 text-green-600" />
-                          ) : (
-                            <CopyIcon className="w-4 h-4" />
-                          )}
-                        </button>
-                      </Tooltip>
-                    )}
-                  </div>
+                          <button
+                            onClick={copyCommandToClipboard}
+                            className="flex items-center text-gray-500 hover:text-gray-700 transition-colors duration-200 p-1 ml-2"
+                          >
+                            {isCommandCopied ? (
+                              <CheckIcon className="w-4 h-4 text-green-600" />
+                            ) : (
+                              <CopyIcon className="w-4 h-4" />
+                            )}
+                          </button>
+                        </Tooltip>
+                      )}
+                    </div>
+                  )}
 
                   <div className="space-y-4 mt-3">
                     {/* Creation Command */}
-                    {clusterData.command && (
+                    {(clusterData.command ||
+                      clusterData.last_creation_command) && (
                       <div>
                         <div className="bg-gray-50 border border-gray-200 rounded-md p-3">
                           <code className="text-sm text-gray-800 font-mono break-all">
-                            {clusterData.command}
+                            {clusterData.command ||
+                              clusterData.last_creation_command}
                           </code>
                         </div>
                       </div>
                     )}
 
                     {/* Task YAML - Collapsible */}
-                    {clusterData.task_yaml &&
+                    {(clusterData.task_yaml ||
+                      clusterData.last_creation_yaml) &&
                       clusterData.task_yaml !== '{}' &&
-                      !clusterData.cluster.startsWith('sky-jobs-controller-') &&
-                      !clusterData.cluster.startsWith(
+                      clusterData.last_creation_yaml !== '{}' &&
+                      !(clusterData.cluster || clusterData.name)?.startsWith(
+                        'sky-jobs-controller-'
+                      ) &&
+                      !(clusterData.cluster || clusterData.name)?.startsWith(
                         'sky-serve-controller-'
                       ) && (
                         <div>
@@ -393,7 +537,10 @@ function ActiveTab({
                           {isYamlExpanded && (
                             <div className="bg-gray-50 border border-gray-200 rounded-md p-3 max-h-96 overflow-y-auto">
                               <pre className="text-sm text-gray-800 font-mono whitespace-pre-wrap">
-                                {formatYaml(clusterData.task_yaml)}
+                                {formatYaml(
+                                  clusterData.task_yaml ||
+                                    clusterData.last_creation_yaml
+                                )}
                               </pre>
                             </div>
                           )}
@@ -407,15 +554,17 @@ function ActiveTab({
         </Card>
       </div>
 
-      {/* Jobs Table */}
-      <div className="mb-8">
-        <ClusterJobs
-          clusterName={clusterData.cluster}
-          clusterJobData={clusterJobData}
-          loading={clusterJobsLoading}
-          refreshClusterJobsOnly={refreshClusterJobsOnly}
-        />
-      </div>
+      {/* Jobs Table - Only show for active clusters */}
+      {!isHistoricalCluster && (
+        <div className="mb-8">
+          <ClusterJobs
+            clusterName={clusterData.cluster}
+            clusterJobData={clusterJobData}
+            loading={clusterJobsLoading}
+            refreshClusterJobsOnly={refreshClusterJobsOnly}
+          />
+        </div>
+      )}
     </div>
   );
 }
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index c691a8739..8690b0203 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -134,6 +134,12 @@ cluster_history_table = sqlalchemy.Table(
     sqlalchemy.Column('launched_resources', sqlalchemy.LargeBinary),
     sqlalchemy.Column('usage_intervals', sqlalchemy.LargeBinary),
     sqlalchemy.Column('user_hash', sqlalchemy.Text),
+    sqlalchemy.Column('last_creation_yaml',
+                      sqlalchemy.Text,
+                      server_default=None),
+    sqlalchemy.Column('last_creation_command',
+                      sqlalchemy.Text,
+                      server_default=None),
 )
 
 ssh_key_table = sqlalchemy.Table(
@@ -308,6 +314,21 @@ def create_table():
             'password',
             sqlalchemy.Text(),
             default_statement='DEFAULT NULL')
+
+        db_utils.add_column_to_table_sqlalchemy(
+            session,
+            'cluster_history',
+            'last_creation_yaml',
+            sqlalchemy.Text(),
+            default_statement='DEFAULT NULL')
+
+        db_utils.add_column_to_table_sqlalchemy(
+            session,
+            'cluster_history',
+            'last_creation_command',
+            sqlalchemy.Text(),
+            default_statement='DEFAULT NULL')
+
         session.commit()
 
 
@@ -597,6 +618,14 @@ def add_or_update_cluster(cluster_name: str,
         # Modify cluster history table
         launched_nodes = getattr(cluster_handle, 'launched_nodes', None)
         launched_resources = getattr(cluster_handle, 'launched_resources', None)
+        creation_info = {}
+        if conditional_values.get('last_creation_yaml') is not None:
+            creation_info = {
+                'last_creation_yaml':
+                    conditional_values.get('last_creation_yaml'),
+                'last_creation_command':
+                    conditional_values.get('last_creation_command'),
+            }
 
         insert_stmnt = insert_func(cluster_history_table).values(
             cluster_hash=cluster_hash,
@@ -605,7 +634,9 @@ def add_or_update_cluster(cluster_name: str,
             requested_resources=pickle.dumps(requested_resources),
             launched_resources=pickle.dumps(launched_resources),
             usage_intervals=pickle.dumps(usage_intervals),
-            user_hash=user_hash)
+            user_hash=user_hash,
+            **creation_info,
+        )
         do_update_stmt = insert_stmnt.on_conflict_do_update(
             index_elements=[cluster_history_table.c.cluster_hash],
             set_={
@@ -617,7 +648,8 @@ def add_or_update_cluster(cluster_name: str,
                     pickle.dumps(launched_resources),
                 cluster_history_table.c.usage_intervals:
                     pickle.dumps(usage_intervals),
-                cluster_history_table.c.user_hash: user_hash
+                cluster_history_table.c.user_hash: user_hash,
+                **creation_info,
             })
         session.execute(do_update_stmt)
 
@@ -1027,40 +1059,122 @@ def get_clusters() -> List[Dict[str, Any]]:
 
 
 @_init_db
-def get_clusters_from_history() -> List[Dict[str, Any]]:
+def get_clusters_from_history(
+        days: Optional[int] = None) -> List[Dict[str, Any]]:
+    """Get cluster reports from history.
+
+    Args:
+        days: If specified, only include historical clusters (those not
+              currently active) that were last used within the past 'days'
+              days. Active clusters are always included regardless of this
+              parameter.
+
+    Returns:
+        List of cluster records with history information.
+    """
     assert _SQLALCHEMY_ENGINE is not None
     with orm.Session(_SQLALCHEMY_ENGINE) as session:
-        rows = session.query(
-            cluster_history_table.join(cluster_table,
-                                       cluster_history_table.c.cluster_hash ==
-                                       cluster_table.c.cluster_hash,
-                                       isouter=True)).all()
-
-    # '(cluster_hash, name, num_nodes, requested_resources, '
-    #         'launched_resources, usage_intervals) '
+        # Explicitly select columns from both tables to avoid ambiguity
+        query = session.query(
+            cluster_history_table.c.cluster_hash, cluster_history_table.c.name,
+            cluster_history_table.c.num_nodes,
+            cluster_history_table.c.requested_resources,
+            cluster_history_table.c.launched_resources,
+            cluster_history_table.c.usage_intervals,
+            cluster_history_table.c.user_hash,
+            cluster_history_table.c.last_creation_yaml,
+            cluster_history_table.c.last_creation_command,
+            cluster_table.c.status, cluster_table.c.workspace,
+            cluster_table.c.status_updated_at).select_from(
+                cluster_history_table.join(cluster_table,
+                                           cluster_history_table.c.cluster_hash
+                                           == cluster_table.c.cluster_hash,
+                                           isouter=True))
+
+        rows = query.all()
+
+    # Prepare filtering parameters
+    cutoff_time = None
+    if days is not None:
+        cutoff_time = int(time.time()) - (days * 24 * 60 * 60)
+
     records = []
     for row in rows:
-        # TODO: use namedtuple instead of dict
         user_hash = _get_user_hash_or_current_user(row.user_hash)
-        status = row.status
-        if status is not None:
-            status = status_lib.ClusterStatus[status]
+        launched_at = _get_cluster_launch_time(row.cluster_hash)
+        duration = _get_cluster_duration(row.cluster_hash)
+
+        # Parse status
+        status = None
+        if row.status:
+            status = status_lib.ClusterStatus[row.status]
+
+        # Apply filtering: always include active clusters, filter historical
+        # ones by time
+        if cutoff_time is not None and status is None:  # Historical cluster
+            # For historical clusters, check if they were used recently
+            # Use the most recent activity from usage_intervals to determine
+            # last use
+            usage_intervals = []
+            if row.usage_intervals:
+                try:
+                    usage_intervals = pickle.loads(row.usage_intervals)
+                except (pickle.PickleError, AttributeError):
+                    usage_intervals = []
+
+            # Find the most recent activity time from usage_intervals
+            last_activity_time = None
+            if usage_intervals:
+                # Get the end time of the last interval (or start time if
+                # still running)
+                last_interval = usage_intervals[-1]
+                last_activity_time = (last_interval[1] if last_interval[1]
+                                      is not None else last_interval[0])
+
+            # Skip historical clusters that haven't been used recently
+            if last_activity_time is None or last_activity_time < cutoff_time:
+                continue
+
+        # Parse launched resources safely
+        launched_resources = None
+        if row.launched_resources:
+            try:
+                launched_resources = pickle.loads(row.launched_resources)
+            except (pickle.PickleError, AttributeError):
+                launched_resources = None
+
+        # Parse usage intervals safely
+        usage_intervals = []
+        if row.usage_intervals:
+            try:
+                usage_intervals = pickle.loads(row.usage_intervals)
+            except (pickle.PickleError, AttributeError):
+                usage_intervals = []
+
+        # Get user name from user hash
+        user = get_user(user_hash)
+        user_name = user.name if user is not None else None
+
         record = {
             'name': row.name,
-            'launched_at': _get_cluster_launch_time(row.cluster_hash),
-            'duration': _get_cluster_duration(row.cluster_hash),
+            'launched_at': launched_at,
+            'duration': duration,
             'num_nodes': row.num_nodes,
-            'resources': pickle.loads(row.launched_resources),
+            'resources': launched_resources,
             'cluster_hash': row.cluster_hash,
-            'usage_intervals': pickle.loads(row.usage_intervals),
+            'usage_intervals': usage_intervals,
             'status': status,
             'user_hash': user_hash,
+            'user_name': user_name,
+            'workspace': row.workspace,
+            'last_creation_yaml': row.last_creation_yaml,
+            'last_creation_command': row.last_creation_command,
         }
 
         records.append(record)
 
     # sort by launch time, descending in recency
-    records = sorted(records, key=lambda record: -record['launched_at'])
+    records = sorted(records, key=lambda record: -(record['launched_at'] or 0))
     return records
 
 
diff --git a/sky/server/requests/payloads.py b/sky/server/requests/payloads.py
index 0d07b8eb2..d347c9408 100644
--- a/sky/server/requests/payloads.py
+++ b/sky/server/requests/payloads.py
@@ -611,3 +611,8 @@ class UpdateConfigBody(RequestBody):
 class GetConfigBody(RequestBody):
     """The request body for getting the entire SkyPilot configuration."""
     pass
+
+
+class CostReportBody(RequestBody):
+    """The request body for the cost report endpoint."""
+    days: Optional[int] = 30
diff --git a/sky/server/server.py b/sky/server/server.py
index f8845961a..1ec38dec8 100644
--- a/sky/server/server.py
+++ b/sky/server/server.py
@@ -1044,13 +1044,14 @@ async def download(download_body: payloads.DownloadBody) -> None:
                                     detail=f'Error creating zip file: {str(e)}')
 
 
-@app.get('/cost_report')
-async def cost_report(request: fastapi.Request) -> None:
+@app.post('/cost_report')
+async def cost_report(request: fastapi.Request,
+                      cost_report_body: payloads.CostReportBody) -> None:
     """Gets the cost report of a cluster."""
     executor.schedule_request(
         request_id=request.state.request_id,
         request_name='cost_report',
-        request_body=payloads.RequestBody(),
+        request_body=cost_report_body,
         func=core.cost_report,
         schedule_type=requests_lib.ScheduleType.SHORT,
     )
diff --git a/sky/skylet/constants.py b/sky/skylet/constants.py
index 78ed4a207..b3dbd086e 100644
--- a/sky/skylet/constants.py
+++ b/sky/skylet/constants.py
@@ -464,3 +464,5 @@ MEMORY_SIZE_PLUS_PATTERN = f'{MEMORY_SIZE_PATTERN[:-3]}+?$/i'
 MIN_PRIORITY = -1000
 MAX_PRIORITY = 1000
 DEFAULT_PRIORITY = 0
+
+COST_REPORT_DEFAULT_DAYS = 30
diff --git a/sky/utils/cli_utils/status_utils.py b/sky/utils/cli_utils/status_utils.py
index bd1ad4ee3..0a46dbea3 100644
--- a/sky/utils/cli_utils/status_utils.py
+++ b/sky/utils/cli_utils/status_utils.py
@@ -7,7 +7,6 @@ import colorama
 
 from sky import backends
 from sky.utils import common_utils
-from sky.utils import controller_utils
 from sky.utils import log_utils
 from sky.utils import resources_utils
 from sky.utils import status_lib
@@ -137,7 +136,8 @@ def get_total_cost_of_displayed_records(
 
 def show_cost_report_table(cluster_records: List[_ClusterCostReportRecord],
                            show_all: bool,
-                           controller_name: Optional[str] = None):
+                           controller_name: Optional[str] = None,
+                           days: Optional[int] = None):
     """Compute cluster table values and display for cost report.
 
     For each cluster, this shows: cluster name, resources, launched time,
@@ -200,23 +200,21 @@ def show_cost_report_table(cluster_records: List[_ClusterCostReportRecord],
         cluster_table.add_row(row)
 
     if cluster_records:
+        controller_record = cluster_records[0]
         if controller_name is not None:
-            controller = controller_utils.Controllers.from_name(controller_name)
-            if controller is None:
-                raise ValueError(f'Controller {controller_name} not found.')
-            controller_handle: backends.CloudVmRayResourceHandle = (
-                cluster_records[0]['handle'])
-            autostop_config = (
-                controller_handle.launched_resources.autostop_config)
-            if autostop_config is not None:
+            autostop = controller_record.get('autostop', None)
+            autostop_str = ''
+            if autostop is not None:
                 autostop_str = (f'{colorama.Style.DIM} (will be autostopped if '
-                                f'idle for {autostop_config.idle_minutes}min)'
+                                f'idle for {autostop}min)'
                                 f'{colorama.Style.RESET_ALL}')
             click.echo(f'\n{colorama.Fore.CYAN}{colorama.Style.BRIGHT}'
                        f'{controller_name}{colorama.Style.RESET_ALL}'
                        f'{autostop_str}')
         else:
-            click.echo(f'{colorama.Fore.CYAN}{colorama.Style.BRIGHT}Clusters'
+            days_str = '' if days is None else f' (last {days} days)'
+            click.echo(f'{colorama.Fore.CYAN}{colorama.Style.BRIGHT}'
+                       f'Clusters{days_str}'
                        f'{colorama.Style.RESET_ALL}')
         click.echo(cluster_table)
 
@@ -345,7 +343,9 @@ def _get_infra(cluster_record: _ClusterRecord, truncate: bool = True) -> str:
 
 
 def _get_status_value_for_cost_report(
-        cluster_cost_report_record: _ClusterCostReportRecord) -> int:
+        cluster_cost_report_record: _ClusterCostReportRecord,
+        truncate: bool = True) -> int:
+    del truncate
     status = cluster_cost_report_record['status']
     if status is None:
         return -1
@@ -353,7 +353,9 @@ def _get_status_value_for_cost_report(
 
 
 def _get_status_for_cost_report(
-        cluster_cost_report_record: _ClusterCostReportRecord) -> str:
+        cluster_cost_report_record: _ClusterCostReportRecord,
+        truncate: bool = True) -> str:
+    del truncate
     status = cluster_cost_report_record['status']
     if status is None:
         return f'{colorama.Style.DIM}TERMINATED{colorama.Style.RESET_ALL}'
@@ -361,7 +363,9 @@ def _get_status_for_cost_report(
 
 
 def _get_resources_for_cost_report(
-        cluster_cost_report_record: _ClusterCostReportRecord) -> str:
+        cluster_cost_report_record: _ClusterCostReportRecord,
+        truncate: bool = True) -> str:
+    del truncate
     launched_nodes = cluster_cost_report_record['num_nodes']
     launched_resources = cluster_cost_report_record['resources']
 
@@ -373,7 +377,9 @@ def _get_resources_for_cost_report(
 
 
 def _get_price_for_cost_report(
-        cluster_cost_report_record: _ClusterCostReportRecord) -> str:
+        cluster_cost_report_record: _ClusterCostReportRecord,
+        truncate: bool = True) -> str:
+    del truncate
     launched_nodes = cluster_cost_report_record['num_nodes']
     launched_resources = cluster_cost_report_record['resources']
 
@@ -383,7 +389,9 @@ def _get_price_for_cost_report(
 
 
 def _get_estimated_cost_for_cost_report(
-        cluster_cost_report_record: _ClusterCostReportRecord) -> str:
+        cluster_cost_report_record: _ClusterCostReportRecord,
+        truncate: bool = True) -> str:
+    del truncate
     cost = cluster_cost_report_record['total_cost']
 
     if not cost:


Test patch:
diff --git a/tests/unit_tests/test_sky/test_cost_report.py b/tests/unit_tests/test_sky/test_cost_report.py
new file mode 100644
index 000000000..b41f4c04f
--- /dev/null
+++ b/tests/unit_tests/test_sky/test_cost_report.py
@@ -0,0 +1,475 @@
+"""Unit tests for sky cost-report functionality."""
+import unittest
+from unittest import mock
+
+from sky import core
+from sky.server.requests import payloads
+from sky.skylet import constants
+from sky.utils import status_lib
+from sky.utils.cli_utils import status_utils
+
+
+class TestCostReportCore(unittest.TestCase):
+    """Test core cost-report functionality."""
+
+    def test_cost_report_default_days(self):
+        """Test cost_report with default days parameter."""
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            mock_get_history.return_value = []
+
+            result = core.cost_report()
+
+            # Should call with default 30 days
+            mock_get_history.assert_called_once_with(days=30)
+            self.assertEqual(result, [])
+
+    def test_cost_report_custom_days(self):
+        """Test cost_report with custom days parameter."""
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            mock_get_history.return_value = []
+
+            result = core.cost_report(days=7)
+
+            # Should call with custom 7 days
+            mock_get_history.assert_called_once_with(days=7)
+            self.assertEqual(result, [])
+
+    def test_cost_report_none_days(self):
+        """Test cost_report with None days parameter."""
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            mock_get_history.return_value = []
+
+            result = core.cost_report(days=None)
+
+            # Should call with default 30 days when None is passed
+            mock_get_history.assert_called_once_with(days=30)
+            self.assertEqual(result, [])
+
+    def test_cost_report_with_pickle_errors(self):
+        """Test cost_report handles pickle errors gracefully when loading historical data."""
+        import pickle
+
+        # Mock get_clusters_from_history to simulate pickle errors being handled internally
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            # Simulate the function handling pickle errors gracefully and returning empty list
+            mock_get_history.return_value = []
+
+            # Even if there are pickle errors internally, the function should not crash
+            result = core.cost_report(days=30)
+
+            self.assertEqual(result, [])
+            mock_get_history.assert_called_once_with(days=30)
+
+
+class TestCostReportStatusUtils(unittest.TestCase):
+    """Test cost-report status utilities."""
+
+    def test_show_cost_report_table_with_days(self):
+        """Test show_cost_report_table displays days information."""
+        # Need non-empty records for show_cost_report_table to call echo
+        mock_resources = mock.Mock()
+        mock_resources.get_cost.return_value = 0.05  # Return numeric value for cost calculation
+        mock_records = [{
+            'name': 'test-cluster',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock_resources,
+            'total_cost': 5.50,
+            'launched_at': 1640995200,
+            'duration': 3600,
+        }]
+
+        with mock.patch('click.echo') as mock_echo:
+            with mock.patch(
+                    'sky.utils.log_utils.create_table') as mock_create_table:
+                mock_table = mock.Mock()
+                mock_create_table.return_value = mock_table
+
+                status_utils.show_cost_report_table(mock_records,
+                                                    show_all=False,
+                                                    days=7)
+
+                # Should display days information in header
+                mock_echo.assert_called()
+                echo_calls = [
+                    str(call[0][0]) for call in mock_echo.call_args_list
+                ]
+                header_with_days = any(
+                    '(last 7 days)' in call for call in echo_calls)
+                self.assertTrue(header_with_days,
+                                "Should display days in header")
+
+    def test_show_cost_report_table_without_days(self):
+        """Test show_cost_report_table without days information."""
+        # Need non-empty records for show_cost_report_table to call echo
+        mock_resources = mock.Mock()
+        mock_resources.get_cost.return_value = 0.05  # Return numeric value for cost calculation
+        mock_records = [{
+            'name': 'test-cluster',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock_resources,
+            'total_cost': 5.50,
+            'launched_at': 1640995200,
+            'duration': 3600,
+        }]
+
+        with mock.patch('click.echo') as mock_echo:
+            with mock.patch(
+                    'sky.utils.log_utils.create_table') as mock_create_table:
+                mock_table = mock.Mock()
+                mock_create_table.return_value = mock_table
+
+                status_utils.show_cost_report_table(mock_records,
+                                                    show_all=False,
+                                                    days=None)
+
+                # Should not display days information in header
+                mock_echo.assert_called()
+                echo_calls = [
+                    str(call[0][0]) for call in mock_echo.call_args_list
+                ]
+                header_with_days = any('(last' in call for call in echo_calls)
+                self.assertFalse(header_with_days,
+                                 "Should not display days in header when None")
+
+    def test_cost_report_helper_functions_signature(self):
+        """Test that cost report helper functions have correct signatures."""
+        # Test that helper functions accept truncate parameter (regression test)
+        mock_resources = mock.Mock()
+        mock_resources.get_cost.return_value = 0.05  # Return numeric value for cost calculation
+        mock_record = {
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock_resources,
+            'total_cost': 10.50
+        }
+
+        # These should not raise TypeError
+        status_utils._get_status_value_for_cost_report(mock_record,
+                                                       truncate=True)
+        status_utils._get_status_for_cost_report(mock_record, truncate=False)
+        status_utils._get_resources_for_cost_report(mock_record, truncate=True)
+        status_utils._get_price_for_cost_report(mock_record, truncate=False)
+        status_utils._get_estimated_cost_for_cost_report(mock_record,
+                                                         truncate=True)
+
+
+class TestCostReportServer(unittest.TestCase):
+    """Test cost-report server functionality."""
+
+    def test_cost_report_body_payload(self):
+        """Test CostReportBody payload structure."""
+        # Test default days
+        body = payloads.CostReportBody()
+        self.assertEqual(body.days, 30)
+
+        # Test custom days
+        body = payloads.CostReportBody(days=7)
+        self.assertEqual(body.days, 7)
+
+        # Test None days
+        body = payloads.CostReportBody(days=None)
+        self.assertIsNone(body.days)
+
+    @mock.patch('sky.server.server.core.cost_report')
+    def test_cost_report_endpoint_calls_core(self, mock_core_cost_report):
+        """Test that cost_report server endpoint calls core function with correct args."""
+        mock_core_cost_report.return_value = []
+
+        # Create mock request and body
+        mock_request = mock.Mock()
+        mock_request.state.request_id = 'test_request_id'
+
+        cost_report_body = payloads.CostReportBody(days=15)
+
+        # Import and test the server function
+        from sky.server import server
+
+        with mock.patch(
+                'sky.server.server.executor.schedule_request') as mock_schedule:
+            # Call the server endpoint
+            import asyncio
+            asyncio.run(server.cost_report(mock_request, cost_report_body))
+
+            # Verify executor.schedule_request was called with correct parameters
+            mock_schedule.assert_called_once()
+            call_args = mock_schedule.call_args
+
+            self.assertEqual(call_args[1]['request_id'], 'test_request_id')
+            self.assertEqual(call_args[1]['request_name'], 'cost_report')
+            self.assertEqual(call_args[1]['request_body'], cost_report_body)
+            self.assertEqual(call_args[1]['func'], server.core.cost_report)
+
+
+class TestHistoricalClusterRobustness(unittest.TestCase):
+    """Test cost report handles historical clusters with missing/invalid resources gracefully."""
+
+    def test_cost_report_with_missing_instance_type(self):
+        """Test cost report doesn't crash when historical cluster has unknown instance type."""
+        # Mock a cluster record with an instance type that doesn't exist in catalogs
+        mock_cluster_record = {
+            'name': 'old-cluster',
+            'status': None,  # Historical cluster
+            'num_nodes': 2,
+            'resources': mock.Mock(),
+            'total_cost': 0.0,
+            'launched_at': 1640995200,  # Some timestamp
+            'duration': 3600,
+            'cluster_hash': 'abc123',
+            'usage_intervals': [(1640995200, 1640998800)],
+            'user_hash': 'user123',
+            'user_name': 'testuser',
+            'workspace': 'default',
+        }
+
+        # Mock the resources object to have an unknown instance type
+        mock_cluster_record[
+            'resources'].instance_type = 'unknown-instance-type-v1'
+        mock_cluster_record['resources'].cloud = mock.Mock()
+        mock_cluster_record['resources'].cloud.__str__ = lambda: 'aws'
+
+        # Mock catalog functions to return None for unknown instance type
+        with mock.patch('sky.catalog.get_hourly_cost', return_value=None):
+            with mock.patch('sky.global_user_state.get_clusters_from_history',
+                            return_value=[mock_cluster_record]):
+
+                # This should not raise an exception
+                result = core.cost_report(days=30)
+
+                # Should return the cluster even if cost calculation fails
+                self.assertEqual(len(result), 1)
+                self.assertEqual(result[0]['name'], 'old-cluster')
+
+    def test_status_utils_with_invalid_resources(self):
+        """Test status utility functions handle invalid resources gracefully."""
+        # Create a mock cluster record with problematic resources
+        mock_record = {
+            'status': None,
+            'num_nodes': 1,
+            'resources': None,  # Problematic: None resources
+            'total_cost': 0.0
+        }
+
+        # These should not crash even with None resources
+        try:
+            status_utils._get_resources_for_cost_report(mock_record,
+                                                        truncate=True)
+            status_utils._get_price_for_cost_report(mock_record, truncate=True)
+            status_utils._get_estimated_cost_for_cost_report(mock_record,
+                                                             truncate=True)
+        except (AttributeError, TypeError):
+            # Expected - the functions might fail gracefully, but shouldn't crash the whole system
+            pass
+
+    def test_cost_report_with_corrupted_resources_data(self):
+        """Test cost report handles corrupted/unpicklable resources data."""
+        mock_cluster_record = {
+            'name': 'corrupted-cluster',
+            'status': None,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 0.0,
+            'launched_at': 1640995200,
+            'duration': 1800,
+            'cluster_hash': 'def456',
+            'usage_intervals': [(1640995200, 1640997000)],
+            'user_hash': 'user456',
+            'user_name': 'testuser2',
+            'workspace': 'default',
+        }
+
+        # Mock resources with missing/invalid attributes
+        mock_cluster_record['resources'].instance_type = None
+        mock_cluster_record['resources'].cloud = None
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history',
+                        return_value=[mock_cluster_record]):
+
+            # Should handle gracefully and not crash
+            result = core.cost_report(days=30)
+            self.assertIsInstance(result, list)
+
+    def test_cost_report_with_empty_usage_intervals(self):
+        """Test cost report handles clusters with empty or malformed usage intervals."""
+        mock_cluster_record = {
+            'name': 'empty-intervals-cluster',
+            'status': None,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 0.0,
+            'launched_at': None,  # Missing launch time
+            'duration': 0,
+            'cluster_hash': 'ghi789',
+            'usage_intervals': [],  # Empty intervals
+            'user_hash': 'user789',
+            'user_name': 'testuser3',
+            'workspace': 'default',
+        }
+
+        mock_cluster_record['resources'].instance_type = 'valid-type'
+        mock_cluster_record['resources'].cloud = mock.Mock()
+        mock_cluster_record['resources'].cloud.__str__ = lambda: 'gcp'
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history',
+                        return_value=[mock_cluster_record]):
+
+            # Should handle gracefully
+            result = core.cost_report(days=30)
+            self.assertEqual(len(result), 1)
+            self.assertEqual(result[0]['name'], 'empty-intervals-cluster')
+
+    def test_cost_report_mixed_valid_invalid_clusters(self):
+        """Test cost report works when some clusters are valid and others have issues."""
+        valid_cluster = {
+            'name': 'valid-cluster',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 5.50,
+            'launched_at': 1640995200,
+            'duration': 3600,
+            'cluster_hash': 'valid123',
+            'usage_intervals': [(1640995200, 1640998800)],
+            'user_hash': 'user_valid',
+            'user_name': 'validuser',
+            'workspace': 'default',
+        }
+        valid_cluster['resources'].instance_type = 'standard-instance'
+        valid_cluster['resources'].cloud = mock.Mock()
+        valid_cluster['resources'].cloud.__str__ = lambda: 'aws'
+
+        invalid_cluster = {
+            'name': 'invalid-cluster',
+            'status': None,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 0.0,
+            'launched_at': 1640995200,
+            'duration': 1800,
+            'cluster_hash': 'invalid456',
+            'usage_intervals': [(1640995200, 1640997000)],
+            'user_hash': 'user_invalid',
+            'user_name': 'invaliduser',
+            'workspace': 'default',
+        }
+        invalid_cluster[
+            'resources'].instance_type = 'discontinued-instance-type'
+        invalid_cluster['resources'].cloud = mock.Mock()
+        invalid_cluster['resources'].cloud.__str__ = lambda: 'nonexistent-cloud'
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history',
+                        return_value=[valid_cluster, invalid_cluster]):
+
+            # Should return both clusters, even if one has issues
+            result = core.cost_report(days=30)
+            self.assertEqual(len(result), 2)
+
+            cluster_names = [r['name'] for r in result]
+            self.assertIn('valid-cluster', cluster_names)
+            self.assertIn('invalid-cluster', cluster_names)
+
+    def test_cost_report_with_controller_clusters(self):
+        """Test cost report handles controller clusters without errors."""
+        controller_cluster = {
+            'name': 'sky-jobs-controller-abc123',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 2.50,
+            'launched_at': 1640995200,
+            'duration': 7200,
+            'cluster_hash': 'controller123',
+            'usage_intervals': [(1640995200, 1641002400)],
+            'user_hash': 'user_controller',
+            'user_name': 'controlleruser',
+            'workspace': 'default',
+        }
+        controller_cluster['resources'].instance_type = 'controller-instance'
+        controller_cluster['resources'].cloud = mock.Mock()
+        controller_cluster['resources'].cloud.__str__ = lambda: 'aws'
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history',
+                        return_value=[controller_cluster]):
+
+            # Should handle controller clusters without issues
+            result = core.cost_report(days=30)
+            self.assertEqual(len(result), 1)
+            self.assertEqual(result[0]['name'], 'sky-jobs-controller-abc123')
+
+    def test_status_utils_with_none_resources_string(self):
+        """Test status utils generate safe strings when resources are problematic."""
+        mock_record_with_none = {
+            'status': None,
+            'num_nodes': 1,
+            'resources': None,
+            'total_cost': 0.0
+        }
+
+        mock_record_with_missing_attrs = {
+            'status': None,
+            'num_nodes': 2,
+            'resources': mock.Mock(),
+            'total_cost': 10.0
+        }
+        # Mock resources object missing expected attributes
+        mock_record_with_missing_attrs['resources'].instance_type = None
+        del mock_record_with_missing_attrs[
+            'resources'].cloud  # Simulate missing attribute
+
+        # Test that these don't crash the status utility functions
+        for record in [mock_record_with_none, mock_record_with_missing_attrs]:
+            try:
+                status_utils._get_resources_for_cost_report(record,
+                                                            truncate=True)
+            except:
+                pass  # May fail, but shouldn't crash the whole system
+
+            try:
+                status_utils._get_price_for_cost_report(record, truncate=True)
+            except:
+                pass
+
+            try:
+                status_utils._get_estimated_cost_for_cost_report(record,
+                                                                 truncate=True)
+            except:
+                pass
+
+
+class TestCostReportCLI(unittest.TestCase):
+    """Test cost-report CLI functionality."""
+
+    @mock.patch('sky.client.sdk.get')
+    @mock.patch('sky.client.sdk.cost_report')
+    @mock.patch('sky.utils.cli_utils.status_utils.show_cost_report_table')
+    @mock.patch(
+        'sky.utils.cli_utils.status_utils.get_total_cost_of_displayed_records')
+    def test_cost_report_cli_function_calls(self, mock_get_total,
+                                            mock_show_table,
+                                            mock_sdk_cost_report, mock_sdk_get):
+        """Test cost-report CLI function calls with mocking."""
+        from sky.client.cli import command
+
+        mock_sdk_cost_report.return_value = 'request_id'
+        mock_sdk_get.return_value = []
+        mock_get_total.return_value = 0.0
+
+        # Test the CLI command logic by calling with a test runner
+        from click.testing import CliRunner
+        runner = CliRunner()
+        result = runner.invoke(command.cost_report, ['--days', '7'])
+
+        # Verify the command completed successfully
+        self.assertEqual(result.exit_code, 0)
+
+        # Verify cost_report was called
+        mock_sdk_cost_report.assert_called_once_with(days=7)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit_tests/test_sky/test_sky_logging.py b/tests/unit_tests/test_sky/test_sky_logging.py
index e18556f6f..88e010fe9 100644
--- a/tests/unit_tests/test_sky/test_sky_logging.py
+++ b/tests/unit_tests/test_sky/test_sky_logging.py
@@ -93,8 +93,9 @@ def test_handler_invalid_level(mock_get, handler, monkeypatch):
     assert handler._level == logging.WARNING
 
 
+@pytest.mark.asyncio
 @mock.patch('sky.utils.context.get', return_value=None)
-def test_handler_with_context_override(mock_get, handler, monkeypatch):
+async def test_handler_with_context_override(mock_get, handler, monkeypatch):
     """Test setting invalid log levels."""
     assert handler.level == logging.INFO
     ctx = context.Context()
diff --git a/tests/unit_tests/test_sky/utils/test_cost_report_controllers.py b/tests/unit_tests/test_sky/utils/test_cost_report_controllers.py
new file mode 100644
index 000000000..f810e2ac4
--- /dev/null
+++ b/tests/unit_tests/test_sky/utils/test_cost_report_controllers.py
@@ -0,0 +1,339 @@
+"""Unit tests for sky cost-report functionality with controllers."""
+import unittest
+from unittest import mock
+
+import colorama
+
+from sky import core
+from sky.client.cli import command
+from sky.utils import controller_utils
+from sky.utils import status_lib
+from sky.utils.cli_utils import status_utils
+
+
+class TestCostReportControllers(unittest.TestCase):
+    """Test cost-report functionality specifically with controllers."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.controller_cluster_record = {
+            'name': 'sky-jobs-controller-fdeaebfa',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 2.50,
+            'launched_at': 1640995200,
+            'duration': 7200,
+            'cluster_hash': 'controller123',
+            'usage_intervals': [(1640995200, 1641002400)],
+            'user_hash': 'fdeaebfa',
+            'user_name': 'zhwu',
+            'workspace': 'default',
+            'autostop': 10,  # 10 minutes autostop
+        }
+        self.controller_cluster_record[
+            'resources'].instance_type = 'n1-standard-1'
+        self.controller_cluster_record['resources'].cloud = mock.Mock()
+        self.controller_cluster_record[
+            'resources'].cloud.__str__ = lambda: 'gcp'
+        self.controller_cluster_record['resources'].get_cost = mock.Mock(
+            return_value=0.05)
+
+    def test_cost_report_controller_name_fix(self):
+        """Test that cost report correctly gets controller name from controller.value.name (not cluster_name)."""
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            with mock.patch('sky.utils.controller_utils.Controllers.from_name'
+                           ) as mock_from_name:
+                mock_get_history.return_value = [self.controller_cluster_record]
+
+                # Mock controller object with proper name attribute
+                mock_controller = mock.Mock()
+                mock_controller.value.name = 'sky-jobs-controller-fdeaebfa'
+                mock_from_name.return_value = mock_controller
+
+                # This should not raise AttributeError on cluster_name
+                result = core.cost_report(days=30)
+
+                self.assertEqual(len(result), 1)
+                self.assertEqual(result[0]['name'],
+                                 'sky-jobs-controller-fdeaebfa')
+
+    def test_cost_report_with_zero_total_cost_fix(self):
+        """Test that cost report sets total_cost to 0.0 (not '-') when there's an error."""
+        error_cluster_record = self.controller_cluster_record.copy()
+        error_cluster_record['resources'].get_cost = mock.Mock(
+            side_effect=Exception("Cost calculation error"))
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            mock_get_history.return_value = [error_cluster_record]
+
+            result = core.cost_report(days=30)
+
+            self.assertEqual(len(result), 1)
+            # Should be 0.0, not string '-'
+            self.assertEqual(result[0]['total_cost'], 0.0)
+            self.assertIsInstance(result[0]['total_cost'], float)
+
+    def test_show_cost_report_table_controller_autostop_fix(self):
+        """Test that show_cost_report_table correctly displays autostop for controllers."""
+        mock_records = [self.controller_cluster_record]
+        controller_name = 'sky-jobs-controller-fdeaebfa'
+
+        with mock.patch('click.echo') as mock_echo:
+            with mock.patch(
+                    'sky.utils.log_utils.create_table') as mock_create_table:
+                mock_table = mock.Mock()
+                mock_create_table.return_value = mock_table
+
+                status_utils.show_cost_report_table(
+                    mock_records,
+                    show_all=False,
+                    controller_name=controller_name)
+
+                # Verify that autostop information is displayed correctly
+                echo_calls = [
+                    str(call[0][0]) for call in mock_echo.call_args_list
+                ]
+                autostop_displayed = any(
+                    '(will be autostopped if idle for 10min)' in call
+                    for call in echo_calls)
+                self.assertTrue(autostop_displayed,
+                                f"Autostop not found in: {echo_calls}")
+
+                # Verify controller name is displayed
+                controller_displayed = any(
+                    controller_name in call for call in echo_calls)
+                self.assertTrue(controller_displayed,
+                                f"Controller name not found in: {echo_calls}")
+
+    def test_show_cost_report_table_controller_no_autostop(self):
+        """Test cost report display when controller has no autostop configured."""
+        no_autostop_record = self.controller_cluster_record.copy()
+        no_autostop_record['autostop'] = None
+
+        with mock.patch('click.echo') as mock_echo:
+            with mock.patch(
+                    'sky.utils.log_utils.create_table') as mock_create_table:
+                mock_table = mock.Mock()
+                mock_create_table.return_value = mock_table
+
+                status_utils.show_cost_report_table(
+                    [no_autostop_record],
+                    show_all=False,
+                    controller_name='sky-jobs-controller-fdeaebfa')
+
+                # Should not display autostop information
+                echo_calls = [
+                    str(call[0][0]) for call in mock_echo.call_args_list
+                ]
+                autostop_displayed = any(
+                    'autostopped' in call for call in echo_calls)
+                self.assertFalse(autostop_displayed)
+
+    def test_cost_report_cli_with_controller_filtering(self):
+        """Test that cost-report CLI correctly handles controller filtering."""
+        controller_record = self.controller_cluster_record.copy()
+        regular_cluster_record = {
+            'name': 'regular-cluster',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 2,
+            'resources': mock.Mock(),
+            'total_cost': 5.00,
+            'launched_at': 1640995200,
+            'duration': 3600,
+            'cluster_hash': 'regular123',
+            'usage_intervals': [(1640995200, 1640998800)],
+            'user_hash': 'user123',
+            'user_name': 'testuser',
+            'workspace': 'default',
+        }
+
+        with mock.patch('sky.client.sdk.cost_report') as mock_sdk_cost_report:
+            with mock.patch('sky.client.sdk.get') as mock_sdk_get:
+                with mock.patch(
+                        'sky.utils.controller_utils.Controllers.from_name'
+                ) as mock_from_name:
+                    with mock.patch(
+                            'sky.utils.cli_utils.status_utils.show_cost_report_table'
+                    ) as mock_show_table:
+                        with mock.patch(
+                                'sky.utils.cli_utils.status_utils.get_total_cost_of_displayed_records'
+                        ) as mock_get_total:
+
+                            # Mock SDK returns
+                            mock_sdk_cost_report.return_value = 'request_id'
+                            mock_sdk_get.return_value = [
+                                controller_record, regular_cluster_record
+                            ]
+                            mock_get_total.return_value = 7.50
+
+                            # Mock controller lookup
+                            mock_controller = mock.Mock()
+                            mock_controller.value.name = 'sky-jobs-controller-fdeaebfa'
+                            mock_from_name.return_value = mock_controller
+
+                            # Test the CLI command logic by calling with a test runner
+                            from click.testing import CliRunner
+                            runner = CliRunner()
+                            result = runner.invoke(command.cost_report,
+                                                   ['--days', '7'])
+
+                            # Verify the command completed successfully
+                            self.assertEqual(result.exit_code, 0)
+
+                            # Verify cost_report was called
+                            mock_sdk_cost_report.assert_called_once_with(days=7)
+
+                            # Verify show_cost_report_table was called
+                            self.assertTrue(mock_show_table.called)
+
+    def test_cost_report_multiple_controllers(self):
+        """Test cost report with multiple different controllers."""
+        jobs_controller_record = self.controller_cluster_record.copy()
+
+        serve_controller_record = {
+            'name': 'sky-serve-controller-abc123',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 3.75,
+            'launched_at': 1640995200,
+            'duration': 5400,
+            'cluster_hash': 'serve123',
+            'usage_intervals': [(1640995200, 1641000600)],
+            'user_hash': 'abc123',
+            'user_name': 'testuser',
+            'workspace': 'default',
+            'autostop': 15,
+        }
+        serve_controller_record['resources'].instance_type = 'n1-standard-2'
+        serve_controller_record['resources'].cloud = mock.Mock()
+        serve_controller_record['resources'].cloud.__str__ = lambda: 'gcp'
+        serve_controller_record['resources'].get_cost = mock.Mock(
+            return_value=0.10)
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            mock_get_history.return_value = [
+                jobs_controller_record, serve_controller_record
+            ]
+
+            result = core.cost_report(days=30)
+
+            self.assertEqual(len(result), 2)
+
+            cluster_names = [r['name'] for r in result]
+            self.assertIn('sky-jobs-controller-fdeaebfa', cluster_names)
+            self.assertIn('sky-serve-controller-abc123', cluster_names)
+
+            # Verify total costs are correctly calculated
+            for record in result:
+                self.assertIsInstance(record['total_cost'], (int, float))
+                self.assertGreater(record['total_cost'], 0)
+
+    def test_cost_report_controller_with_missing_attributes(self):
+        """Test cost report handles controllers with missing or None attributes gracefully."""
+        incomplete_controller_record = {
+            'name': 'sky-jobs-controller-incomplete',
+            'status': status_lib.ClusterStatus.UP,
+            'num_nodes': 1,
+            'resources': mock.Mock(),
+            'total_cost': 1.25,
+            'launched_at': 1640995200,
+            'duration': 1800,
+            'cluster_hash': 'incomplete123',
+            'usage_intervals': [(1640995200, 1640997000)],
+            'user_hash': 'incomplete',
+            'user_name': 'testuser',
+            'workspace': 'default',
+            'autostop': None,  # No autostop configured
+        }
+        incomplete_controller_record['resources'].instance_type = 'unknown-type'
+        incomplete_controller_record['resources'].cloud = mock.Mock()
+        incomplete_controller_record['resources'].cloud.__str__ = lambda: 'aws'
+        incomplete_controller_record['resources'].get_cost = mock.Mock(
+            return_value=0.03)
+
+        with mock.patch('sky.global_user_state.get_clusters_from_history'
+                       ) as mock_get_history:
+            mock_get_history.return_value = [incomplete_controller_record]
+
+            # Should not raise exception
+            result = core.cost_report(days=30)
+
+            self.assertEqual(len(result), 1)
+            self.assertEqual(result[0]['name'],
+                             'sky-jobs-controller-incomplete')
+
+    def test_cost_report_helper_functions_with_controllers(self):
+        """Test that cost report helper functions work correctly with controller records."""
+        # Test helper functions from status_utils that are used in cost reporting
+
+        # Test _get_resources_for_cost_report
+        resources_str = status_utils._get_resources_for_cost_report(
+            self.controller_cluster_record, truncate=True)
+        self.assertIsInstance(resources_str, str)
+        self.assertIn('1x', resources_str)  # Should show node count
+
+        # Test _get_price_for_cost_report
+        price_str = status_utils._get_price_for_cost_report(
+            self.controller_cluster_record, truncate=True)
+        self.assertIsInstance(price_str, str)
+        self.assertIn('$', price_str)  # Should have dollar sign
+
+        # Test _get_estimated_cost_for_cost_report
+        cost_str = status_utils._get_estimated_cost_for_cost_report(
+            self.controller_cluster_record, truncate=True)
+        self.assertIsInstance(cost_str, str)
+        self.assertEqual(cost_str, '$ 2.50')  # Should match our test data
+
+        # Test _get_status_for_cost_report
+        status_str = status_utils._get_status_for_cost_report(
+            self.controller_cluster_record, truncate=True)
+        self.assertIsInstance(status_str, str)
+        # Should contain ANSI color codes for UP status
+        self.assertIn('\x1b[', status_str)  # ANSI escape sequence
+
+    def test_cost_report_table_formatting_with_controller(self):
+        """Test that cost report table is formatted correctly for controllers."""
+        with mock.patch('click.echo') as mock_echo:
+            with mock.patch(
+                    'sky.utils.log_utils.create_table') as mock_create_table:
+                mock_table = mock.Mock()
+                mock_create_table.return_value = mock_table
+
+                status_utils.show_cost_report_table(
+                    [self.controller_cluster_record],
+                    show_all=True,  # Test with show_all=True
+                    controller_name='sky-jobs-controller-fdeaebfa')
+
+                # Verify table creation and display
+                mock_create_table.assert_called_once()
+                mock_table.add_row.assert_called_once()
+
+                # Verify proper formatting in echo calls
+                echo_calls = [
+                    str(call[0][0]) for call in mock_echo.call_args_list
+                ]
+
+                # Should have controller name with proper formatting
+                controller_name_found = False
+                for call in echo_calls:
+                    if 'sky-jobs-controller-fdeaebfa' in call:
+                        # Should have ANSI color codes
+                        self.assertIn(colorama.Fore.CYAN, call)
+                        self.assertIn(colorama.Style.BRIGHT, call)
+                        self.assertIn(colorama.Style.RESET_ALL, call)
+                        controller_name_found = True
+                        break
+
+                self.assertTrue(
+                    controller_name_found,
+                    f"Properly formatted controller name not found in: {echo_calls}"
+                )
+
+
+if __name__ == '__main__':
+    unittest.main()