auto: update all repos

Author: Homer Quan

mn_cli/libs/job_cleanup.py

@@ -20,22 +20,31 @@
 _RESOURCE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
 
 
+class JobResourceCleanupError(RuntimeError):
+    """Raised when a cleared job still has adapter-owned local resources."""
+
+
 def cleanup_cancelled_job_resources(
     job_id: str, *, runtime_client: Any, log: Any
-) -> None:
+) -> dict[str, list[str]]:
+    summary: dict[str, list[str]] = {
+        "process_removed": [],
+        "process_skipped": [],
+        "errors": [],
+    }
     original_job_id = job_id
     job_id = _validated_resource_id(job_id)
     if job_id is None:
         log.warning(
             "Refusing local cleanup for invalid job ID: %r", original_job_id
         )
-        return
+        summary["errors"].append(f"invalid job ID: {original_job_id!r}")
+        return summary
 
-    summary = {"process_removed": [], "process_skipped": [], "errors": []}
     run_id = blueprint_run_id_for_job(job_id, runtime_client=runtime_client)
     if run_id:
         run_dir = default_runs_root() / run_id
-        if run_dir.is_dir():
+        if run_dir.is_dir() and not run_dir.is_symlink():
             cleanup_blueprint_host_hooks(
                 run_dir, dry_run=False, summary=summary, reason="job_cancelled"
             )
@@ -46,6 +55,7 @@ def cleanup_cancelled_job_resources(
     cleanup_local_openshell_sandboxes(job_id, summary)
     for error in summary["errors"]:
         log.warning("Failed to cleanup local resources for job %s: %s", job_id, error)
+    return summary
 
 
 def cleanup_cleared_job_resources(
@@ -58,13 +68,22 @@ def cleanup_cleared_job_resources(
             log.warning(
                 "Refusing local cleanup for invalid job ID: %r", original_job_id
             )
-        return
+        raise JobResourceCleanupError(f"invalid job ID: {original_job_id!r}")
 
     run_id = blueprint_run_id_for_job(job_id, runtime_client=runtime_client)
-    cleanup_cancelled_job_resources(job_id, runtime_client=runtime_client, log=log)
+    cancelled_summary = (
+        cleanup_cancelled_job_resources(
+            job_id, runtime_client=runtime_client, log=log
+        )
+        or {}
+    )
+    errors = list(cancelled_summary.get("errors") or [])
     try:
-        cleanup_docker_worker_services(job_id=job_id)
-    except Exception:
+        docker_result = cleanup_docker_worker_services(job_id=job_id)
+        if isinstance(docker_result, dict):
+            errors.extend(str(error) for error in docker_result.get("errors") or [])
+    except Exception as error:
+        errors.append(f"DockerWorker cleanup failed: {error}")
         log.warning(
             "Failed to cleanup DockerWorker resources for cleared job %s",
             job_id,
@@ -82,12 +101,21 @@ def cleanup_cleared_job_resources(
 
     for path in paths:
         try:
-            shutil.rmtree(path)
+            if path.is_symlink():
+                path.unlink()
+            else:
+                shutil.rmtree(path)
         except FileNotFoundError:
             pass
-        except OSError:
+        except OSError as error:
+            errors.append(f"failed to remove {path}: {error}")
             log.warning("Failed to remove cleared job path %s", path, exc_info=True)
 
+    if errors:
+        raise JobResourceCleanupError(
+            f"local cleanup incomplete for {job_id}: {'; '.join(errors)}"
+        )
+
 
 def blueprint_run_id_for_job(job_id: str, *, runtime_client: Any) -> str | None:
     job_id = _validated_resource_id(job_id)

mn_cli/libs/job_cmds.py

@@ -7,6 +7,7 @@
 from mn_cli.shared import console, client, config, logger
 from mn_cli.error_handler import handle_cli_error
 from mn_cli.libs.job_cleanup import (
+    JobResourceCleanupError,
     blueprint_run_id_for_job,
     blueprint_run_id_from_run_store,
     cleanup_cancelled_job_resources,
@@ -115,23 +116,36 @@ def clear(
     yes: bool = typer.Option(False, "--yes", "-y", help="Clear terminal jobs without prompting."),
 ):
     """Remove terminal jobs and all runtime resources they own."""
+    cleanup_errors: list[str] = []
+
+    def cleanup_item(event: dict) -> None:
+        job_id = str(event.get("item_id") or "")
+        try:
+            _cleanup_cleared_job_resources(job_id)
+        except JobResourceCleanupError as error:
+            cleanup_errors.append(str(error))
+            event["status"] = "failed"
+            event["error"] = f"local runtime cleanup incomplete: {error}"
+            logger.error("Local cleanup failed for cleared job %s: %s", job_id, error)
+
     try:
         if not yes and not typer.confirm(
             "Clear all terminal jobs and their runtime resources?", default=False
         ):
             print_confirmed(console, "Job clear", status="aborted")
             return
 
+        _cleanup_job_ids_or_raise(_terminal_job_ids())
         result = start_and_watch(
             "clear_jobs",
             {},
             action="Job clear",
-            on_accepted_item=lambda event: _cleanup_cleared_job_resources(
-                str(event.get("item_id") or "")
-            ),
+            on_accepted_item=cleanup_item,
             runtime_client=client,
         )
         logger.info("Finished clear operation %s", result.get("operation_id"))
+        if cleanup_errors:
+            raise JobResourceCleanupError("; ".join(cleanup_errors))
     except grpc.RpcError as e:
         if e.code() == grpc.StatusCode.PERMISSION_DENIED and "MN_GRPC_ADMIN_TOKEN" in str(e.details()):
             print_error(console, "ClearJobs admin authorization failed.")
@@ -242,6 +256,34 @@ def _cleanup_cleared_job_resources(job_id: str) -> None:
     cleanup_cleared_job_resources(job_id, runtime_client=client, log=logger)
 
 
+def _cleanup_job_ids_or_raise(job_ids: list[str]) -> None:
+    errors = []
+    for job_id in job_ids:
+        try:
+            _cleanup_cleared_job_resources(job_id)
+        except JobResourceCleanupError as error:
+            errors.append(str(error))
+    if errors:
+        raise JobResourceCleanupError("; ".join(errors))
+
+
+def _terminal_job_ids() -> list[str]:
+    result = json.loads(
+        client.list_jobs(limit=_ALL_JOBS_LIMIT, include_terminal=True)
+    )
+    jobs = result.get("data") if isinstance(result, dict) else None
+    if not isinstance(jobs, list):
+        raise ValueError("runtime returned an invalid job list")
+    return [
+        job_id
+        for job in jobs
+        if isinstance(job, dict)
+        and job.get("status") in {"completed", "failed", "cancelled"}
+        and isinstance((job_id := job.get("job_id")), str)
+        and job_id
+    ]
+
+
 def _blueprint_run_id_for_job(job_id: str) -> str | None:
     return blueprint_run_id_for_job(job_id, runtime_client=client)
 

mn_cli/libs/operation_cmds.py

@@ -172,6 +172,9 @@ def _print_plain_event(
     if event_type == "stream_heartbeat":
         return
 
+    if on_accepted_item and _accepted_item_status(event.get("status")):
+        on_accepted_item(event)
+
     item_id = str(event.get("item_id") or "operation")
     status = str(event.get("status") or "")
 
@@ -180,13 +183,9 @@ def _print_plain_event(
     elif status == "failed":
         print_warning(console, f"{item_id}: {event.get('error') or 'operation item failed'}")
     elif status == "cancellation_pending":
-        if on_accepted_item:
-            on_accepted_item(event)
         print_info(console, f"{item_id}: cancellation accepted; cleanup queued on owner node")
     elif event_type in {"item_completed", "item_deferred"}:
         prefix = "✓" if status in _SUCCESS_ITEM_STATUSES else "→"
-        if on_accepted_item and _accepted_item_status(status):
-            on_accepted_item(event)
         console.print(f"{prefix} {item_id}: {status or 'completed'}")
 
 

mn_cli/libs/stable_job_cmds.py

@@ -8,7 +8,7 @@
 
 from mn_cli.error_handler import handle_cli_error
 from mn_cli.libs.bundles import read_bundle
-from mn_cli.libs.job_cleanup import cleanup_cleared_job_resources
+from mn_cli.libs.job_cleanup import JobResourceCleanupError, cleanup_cleared_job_resources
 from mn_cli.libs.ui import print_success_confirmation
 from mn_cli.shared import client, console, logger
 
@@ -102,10 +102,23 @@ def delete(
         return
     try:
         run_ids = _stable_job_run_ids(job_id)
+        cleanup_errors = []
+        for resource_id in [*run_ids, job_id]:
+            try:
+                cleanup_cleared_job_resources(
+                    resource_id, runtime_client=client, log=logger
+                )
+            except JobResourceCleanupError as error:
+                cleanup_errors.append(str(error))
+        if cleanup_errors:
+            raise JobResourceCleanupError("; ".join(cleanup_errors))
+
         result = json.loads(client.delete_stable_job(job_id, confirmed=True))
-        for run_id in run_ids:
-            cleanup_cleared_job_resources(run_id, runtime_client=client, log=logger)
-        cleanup_cleared_job_resources(job_id, runtime_client=client, log=logger)
+        cleanup_errors = [
+            str(error) for error in result.get("resource_cleanup_errors") or []
+        ]
+        if cleanup_errors:
+            raise JobResourceCleanupError("; ".join(cleanup_errors))
         console.print_json(data=result)
     except Exception as exc:
         handle_cli_error(exc, console, "job delete")