Feat(aws): Add list_jobs.py

Author: klieret

.cursor/rules/style.mdc

@@ -23,6 +23,7 @@ alwaysApply: true
 15. Command line arguments should be parsed with `argparse`.
 16. Generally prefer to use keyword arguments over positional arguments. If there are somewhat specific optional arguments, mark them as keyword-only.
 17. Use logger instead of print statements.
+18. Avoid long indented blocks in functions if they are only for else: return None. Instead you can just test for this early.
 
 ## Test style
 

aws/cancel_job.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""
+AWS Batch job cancellation script for CodeClash.
+
+This script can cancel AWS Batch jobs by job name or job ID.
+It can handle multiple jobs at once for cancellation.
+
+Usage:
+    python cancel_job.py job-id-1 job-id-2 job-name-3
+    python cancel_job.py --reason "Manual cancellation" job-id-1
+"""
+
+import argparse
+from typing import Any
+
+import boto3
+
+from codeclash.utils.log import get_logger
+
+logger = get_logger("cancel")
+
+
+class AWSBatchJobCanceller:
+    def __init__(self):
+        self.batch_client = boto3.client("batch")
+
+    def find_job_by_name(self, job_name: str) -> list[dict[str, Any]]:
+        """Find jobs by name. Returns list of job info dictionaries."""
+        jobs = []
+
+        # Search in different job queues and statuses
+        for status in ["SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING"]:
+            response = self.batch_client.list_jobs(
+                jobQueue="codeclash-queue",  # Default queue name from run_job.py
+                jobStatus=status,
+            )
+
+            for job in response["jobs"]:
+                if job["jobName"] == job_name:
+                    jobs.append(job)
+
+        return jobs
+
+    def get_job_info(self, job_identifier: str) -> dict[str, Any] | None:
+        """Get job info by job ID or job name."""
+        # First try as job ID
+        try:
+            response = self.batch_client.describe_jobs(jobs=[job_identifier])
+            if response["jobs"]:
+                return response["jobs"][0]
+        except Exception:
+            pass
+
+        # Try as job name
+        jobs = self.find_job_by_name(job_identifier)
+        if len(jobs) == 1:
+            return jobs[0]
+        elif len(jobs) > 1:
+            logger.warning(f"Multiple jobs found with name '{job_identifier}', using most recent")
+            return max(jobs, key=lambda x: x["createdAt"])
+
+        return None
+
+    def cancel_job(self, job_identifier: str, *, reason: str = "Cancelled by user") -> bool:
+        """Cancel a job by ID or name. Returns True if successful."""
+        job_info = self.get_job_info(job_identifier)
+
+        if not job_info:
+            logger.error(f"Job not found: {job_identifier}")
+            return False
+
+        job_id = job_info["jobId"]
+        job_name = job_info["jobName"]
+        status = job_info["status"]
+
+        if status in ["SUCCEEDED", "FAILED"]:
+            logger.warning(f"Job {job_name} ({job_id}) is already {status.lower()}")
+            return True
+
+        try:
+            self.batch_client.cancel_job(jobId=job_id, reason=reason)
+            logger.info(f"Successfully cancelled job: {job_name} ({job_id})")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to cancel job {job_name} ({job_id}): {e}", exc_info=True)
+            return False
+
+    def cancel_jobs(self, job_identifiers: list[str], *, reason: str = "Cancelled by user") -> int:
+        """Cancel multiple jobs. Returns number of successfully cancelled jobs."""
+        successful_cancellations = 0
+
+        for job_identifier in job_identifiers:
+            if self.cancel_job(job_identifier, reason=reason):
+                successful_cancellations += 1
+
+        return successful_cancellations
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Cancel AWS Batch jobs by name or ID",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("jobs", nargs="+", help="Job IDs or job names to cancel")
+    parser.add_argument(
+        "--reason", default="Cancelled by user", help="Reason for cancellation (default: 'Cancelled by user')"
+    )
+
+    args = parser.parse_args()
+
+    canceller = AWSBatchJobCanceller()
+
+    # Cancel jobs
+    logger.info(f"Cancelling {len(args.jobs)} job(s)...")
+    canceller.cancel_jobs(args.jobs, reason=args.reason)
+
+
+if __name__ == "__main__":
+    main()

aws/list_jobs.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""
+AWS Batch job listing script for CodeClash.
+
+This is a convenience script that lists all AWS Batch jobs sorted by creation time.
+
+Usage:
+    python list_jobs.py
+    python list_jobs.py --status RUNNING
+    python list_jobs.py --queue my-queue --limit 20
+"""
+
+import argparse
+from datetime import datetime
+from typing import Any
+
+import boto3
+
+from codeclash.utils.log import get_logger
+
+logger = get_logger("list_jobs", emoji="📋")
+
+
+class AWSBatchJobLister:
+    def __init__(self, job_queue: str = "codeclash-queue"):
+        self.batch_client = boto3.client("batch")
+        self.job_queue = job_queue
+
+    def list_jobs(self, *, status: str | None = None, limit: int | None = None) -> list[dict[str, Any]]:
+        """List jobs from AWS Batch, optionally filtered by status."""
+        all_jobs = []
+
+        # If specific status is requested, only query that status
+        if status:
+            statuses = [status.upper()]
+        else:
+            # Query all active statuses
+            statuses = ["SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING", "SUCCEEDED", "FAILED"]
+
+        for job_status in statuses:
+            try:
+                # Use paginator to handle large number of jobs
+                paginator = self.batch_client.get_paginator("list_jobs")
+                page_iterator = paginator.paginate(jobQueue=self.job_queue, jobStatus=job_status)
+
+                for page in page_iterator:
+                    jobs_in_page = page.get("jobSummaryList", [])
+                    all_jobs.extend(jobs_in_page)
+
+            except Exception as e:
+                logger.warning(f"Failed to list jobs with status {job_status}: {e}")
+                continue
+
+        # Sort by creation time (newest first)
+        all_jobs.sort(key=lambda x: x["createdAt"], reverse=True)
+
+        # Apply limit if specified
+        if limit:
+            all_jobs = all_jobs[:limit]
+
+        return all_jobs
+
+    def format_job_info(self, job: dict[str, Any]) -> str:
+        """Format job information for display."""
+        job_id = job["jobId"]
+        job_name = job["jobName"]
+        status = job["status"]
+        created_at = job["createdAt"]
+
+        # Format creation time
+        if isinstance(created_at, datetime):
+            created_str = created_at.strftime("%Y-%m-%d %H:%M:%S")
+        else:
+            # Handle timestamp (milliseconds since epoch)
+            created_str = datetime.fromtimestamp(created_at / 1000).strftime("%Y-%m-%d %H:%M:%S")
+
+        # Get additional info if available
+        started_at = job.get("startedAt")
+        stopped_at = job.get("stoppedAt")
+
+        duration = ""
+        if started_at and stopped_at:
+            if isinstance(started_at, datetime):
+                start_time = started_at
+                stop_time = stopped_at
+            else:
+                start_time = datetime.fromtimestamp(started_at / 1000)
+                stop_time = datetime.fromtimestamp(stopped_at / 1000)
+            duration_seconds = (stop_time - start_time).total_seconds()
+            duration = f" ({duration_seconds:.0f}s)"
+        elif started_at:
+            if isinstance(started_at, datetime):
+                start_time = started_at
+            else:
+                start_time = datetime.fromtimestamp(started_at / 1000)
+            running_seconds = (datetime.now() - start_time).total_seconds()
+            duration = f" (running {running_seconds:.0f}s)"
+
+        return f"{job_id:<36} {status:<10} {created_str} {job_name}{duration}"
+
+    def print_jobs(self, jobs: list[dict[str, Any]]) -> None:
+        """Print jobs in a formatted table."""
+        if not jobs:
+            logger.info("No jobs found")
+            return
+
+        # Print header
+        header = f"{'Job ID':<36} {'Status':<10} {'Created':<19} {'Job Name'}"
+        print(header)
+        print("-" * len(header))
+
+        # Print jobs
+        for job in jobs:
+            print(self.format_job_info(job))
+
+        print(f"\nTotal: {len(jobs)} job(s)")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="List AWS Batch jobs",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument(
+        "--status",
+        choices=["SUBMITTED", "PENDING", "RUNNABLE", "STARTING", "RUNNING", "SUCCEEDED", "FAILED"],
+        help="Filter jobs by status (default: show all statuses)",
+    )
+    parser.add_argument("--queue", default="codeclash-queue", help="Job queue name (default: codeclash-queue)")
+    parser.add_argument("--limit", type=int, help="Maximum number of jobs to display")
+
+    args = parser.parse_args()
+
+    lister = AWSBatchJobLister(job_queue=args.queue)
+
+    # List jobs
+    logger.info(f"Listing jobs from queue: {args.queue}")
+    if args.status:
+        logger.info(f"Filtering by status: {args.status}")
+
+    jobs = lister.list_jobs(status=args.status, limit=args.limit)
+    lister.print_jobs(jobs)
+
+
+if __name__ == "__main__":
+    main()

aws/setup/batch/job_definition.json

@@ -21,7 +21,7 @@
         "type": "VCPU"
       },
       {
-        "value": "8192",
+        "value": "7500",
         "type": "MEMORY"
       }
     ],

aws/setup/batch/launch_template.json

@@ -5,7 +5,7 @@
       {
         "DeviceName": "/dev/xvda",
         "Ebs": {
-          "VolumeSize": 30,
+          "VolumeSize": 50,
           "VolumeType": "gp3"
         }
       }