feat: add spec.ttl for automatic job cleanup after completion

fournos/core/duration.py

@@ -0,0 +1,32 @@
+"""Parse Go-style duration strings (e.g. "12h", "30m", "7d", "1h30m") into timedelta."""
+
+from __future__ import annotations
+
+import re
+from datetime import timedelta
+
+_DURATION_RE = re.compile(r"(?:(\d+)d)?(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$")
+
+
+def parse_duration(value: str) -> timedelta | None:
+    """Parse a Go-style duration string into a timedelta.
+
+    Supports days (d), hours (h), minutes (m), and seconds (s).
+    Returns None if the string is empty or does not match.
+    """
+    value = value.strip()
+    if not value:
+        return None
+
+    m = _DURATION_RE.match(value)
+    if not m or not any(m.groups()):
+        return None
+
+    try:
+        days = int(m.group(1) or 0)
+        hours = int(m.group(2) or 0)
+        minutes = int(m.group(3) or 0)
+        seconds = int(m.group(4) or 0)
+        return timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
+    except (ValueError, OverflowError):
+        return None

fournos/handlers/execution.py

@@ -22,6 +22,7 @@
     COND_WORKLOAD_ADMITTED,
     owner_ref,
     set_condition,
+    set_terminal_phase,
 )
 
 logger = logging.getLogger(__name__)
@@ -75,8 +76,7 @@ def handle_shutdown(name, status, patch, shutdown):
         )
     else:
         ctx.kueue.delete_workload(name)
-        patch.status["phase"] = Phase.STOPPED
-        patch.status["message"] = "Job stopped by user"
+        set_terminal_phase(patch, Phase.STOPPED, "Job stopped by user")
         set_condition(
             patch,
             conditions,
@@ -121,8 +121,7 @@ def _finish_stop(name, conditions, patch, pr_message):
     """Transition from Stopping to Stopped: delete Workload and set terminal status."""
     ctx.kueue.delete_workload(name)
 
-    patch.status["phase"] = Phase.STOPPED
-    patch.status["message"] = "Job stopped by user"
+    set_terminal_phase(patch, Phase.STOPPED, "Job stopped by user")
 
     set_condition(
         patch,
@@ -181,8 +180,9 @@ def reconcile_admitted(spec, name, namespace, status, patch, body):
                     cluster, name, owner_ref(body)
                 )
             except client.exceptions.ApiException as exc:
-                patch.status["phase"] = Phase.FAILED
-                patch.status["message"] = f"Failed to copy kubeconfig: {exc.reason}"
+                set_terminal_phase(
+                    patch, Phase.FAILED, f"Failed to copy kubeconfig: {exc.reason}"
+                )
                 set_condition(
                     patch,
                     conditions,
@@ -202,8 +202,7 @@ def reconcile_admitted(spec, name, namespace, status, patch, body):
             )
         except (KeyError, client.exceptions.ApiException) as exc:
             msg = str(exc).strip("'\"") if isinstance(exc, KeyError) else exc.reason
-            patch.status["phase"] = Phase.FAILED
-            patch.status["message"] = msg
+            set_terminal_phase(patch, Phase.FAILED, msg)
             set_condition(
                 patch,
                 conditions,
@@ -233,8 +232,9 @@ def reconcile_admitted(spec, name, namespace, status, patch, body):
                 )
         except client.exceptions.ApiException as exc:
             if exc.status != 409:
-                patch.status["phase"] = Phase.FAILED
-                patch.status["message"] = f"Failed to create PipelineRun: {exc.reason}"
+                set_terminal_phase(
+                    patch, Phase.FAILED, f"Failed to create PipelineRun: {exc.reason}"
+                )
                 set_condition(
                     patch,
                     conditions,
@@ -276,8 +276,7 @@ def reconcile_running(name, status, patch):
     conditions = list(status.get("conditions") or [])
 
     if pr is None:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = "PipelineRun not found"
+        set_terminal_phase(patch, Phase.FAILED, "PipelineRun not found")
         set_condition(
             patch,
             conditions,
@@ -295,8 +294,7 @@ def reconcile_running(name, status, patch):
         "Job %s: PipelineRun status=%s, message=%s", name, pr_status, pr_message
     )
     if pr_status == "succeeded":
-        patch.status["phase"] = Phase.SUCCEEDED
-        patch.status["message"] = "Pipeline completed successfully"
+        set_terminal_phase(patch, Phase.SUCCEEDED, "Pipeline completed successfully")
         set_condition(
             patch,
             conditions,
@@ -308,8 +306,7 @@ def reconcile_running(name, status, patch):
         ctx.kueue.delete_workload(name)
         logger.info("Job %s: succeeded", name)
     elif pr_status == "failed":
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = pr_message or "PipelineRun failed"
+        set_terminal_phase(patch, Phase.FAILED, pr_message or "PipelineRun failed")
         set_condition(
             patch,
             conditions,

fournos/handlers/lifecycle.py

@@ -24,6 +24,7 @@
     LOCK_HOLDING_PHASES,
     Phase,
 )
+from fournos.core.duration import parse_duration
 from fournos.core.kueue import KueueClient
 from fournos.settings import settings
 from fournos.state import ctx
@@ -34,6 +35,7 @@
     CRD_VERSION,
     owner_ref,
     set_condition,
+    set_terminal_phase,
 )
 
 logger = logging.getLogger(__name__)
@@ -50,30 +52,36 @@ def on_create(spec, name, namespace, status, patch, body):
 
     shutdown = spec.get("shutdown")
     if shutdown is not None:
-        patch.status["phase"] = Phase.STOPPED
-        patch.status["message"] = "Job stopped by user"
+        set_terminal_phase(patch, Phase.STOPPED, "Job stopped by user")
         logger.info("Job %s: created with shutdown=%s, skipping", name, shutdown)
         return
 
+    ttl_raw = spec.get("ttl")
+    if ttl_raw and parse_duration(ttl_raw) is None:
+        set_terminal_phase(patch, Phase.FAILED, f"Invalid ttl value: {ttl_raw!r}")
+        logger.error("Job %s: invalid ttl %r", name, ttl_raw)
+        return
+
     cron_expr = spec.get("schedule")
     try:
         scheduled_time = _parse_scheduled_time(spec)
     except ValueError as exc:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = str(exc)
+        set_terminal_phase(patch, Phase.FAILED, str(exc))
         return
 
     if cron_expr and scheduled_time is not None:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = (
-            "'schedule' and 'scheduledStartTime' are mutually exclusive"
+        set_terminal_phase(
+            patch,
+            Phase.FAILED,
+            "'schedule' and 'scheduledStartTime' are mutually exclusive",
         )
         return
 
     if cron_expr:
         if not croniter.is_valid(cron_expr):
-            patch.status["phase"] = Phase.FAILED
-            patch.status["message"] = f"Invalid cron expression: {cron_expr}"
+            set_terminal_phase(
+                patch, Phase.FAILED, f"Invalid cron expression: {cron_expr}"
+            )
             return
         patch.status["phase"] = Phase.RECURRING
         patch.status["message"] = f"Recurring schedule: {cron_expr}"
@@ -92,28 +100,30 @@ def on_create(spec, name, namespace, status, patch, body):
     clusterless = spec.get("clusterless", False)
 
     if lock_only and not cluster:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = "lockOnly: true requires 'cluster' to be set"
+        set_terminal_phase(
+            patch, Phase.FAILED, "lockOnly: true requires 'cluster' to be set"
+        )
         return
 
     if spec.get("lockUntil") and not lock_only:
         # lock_only is only False here if the user explicitly wrote
         # lockOnly: false — a bare lockUntil already implies lockOnly: true.
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = "lockUntil cannot be combined with lockOnly: false"
+        set_terminal_phase(
+            patch, Phase.FAILED, "lockUntil cannot be combined with lockOnly: false"
+        )
         return
 
     try:
         parse_iso_timestamp(spec.get("lockUntil"), "lockUntil")
     except ValueError as exc:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = str(exc)
+        set_terminal_phase(patch, Phase.FAILED, str(exc))
         return
 
     if not lock_only and not spec.get("executionEngine"):
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = (
-            "spec.executionEngine is required for non-lockOnly jobs"
+        set_terminal_phase(
+            patch,
+            Phase.FAILED,
+            "spec.executionEngine is required for non-lockOnly jobs",
         )
         return
 
@@ -127,28 +137,28 @@ def on_create(spec, name, namespace, status, patch, body):
             ),
         ]:
             if cond:
-                patch.status["phase"] = Phase.FAILED
-                patch.status["message"] = msg
+                set_terminal_phase(patch, Phase.FAILED, msg)
                 return
         patch.status["phase"] = Phase.RESOLVING
         patch.status["message"] = "Resolving job requirements"
         return
     if exclusive and not cluster:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = "exclusive: true requires 'cluster' to be set"
+        set_terminal_phase(
+            patch, Phase.FAILED, "exclusive: true requires 'cluster' to be set"
+        )
         return
 
     if cluster:
         try:
             known_flavors = ctx.kueue.list_flavors()
         except k8s_client.exceptions.ApiException as exc:
-            patch.status["phase"] = Phase.FAILED
-            patch.status["message"] = f"Failed to list clusters: {exc.reason}"
+            set_terminal_phase(
+                patch, Phase.FAILED, f"Failed to list clusters: {exc.reason}"
+            )
             logger.error("Job %s: list_flavors failed: %s", name, exc.reason)
             return
         if cluster not in known_flavors:
-            patch.status["phase"] = Phase.FAILED
-            patch.status["message"] = f"Cluster '{cluster}' not found"
+            set_terminal_phase(patch, Phase.FAILED, f"Cluster '{cluster}' not found")
             return
 
     if exclusive:
@@ -214,8 +224,7 @@ def reconcile_scheduled(spec, name, namespace, status, patch, body):
     try:
         scheduled_time = _parse_scheduled_time(spec)
     except ValueError as exc:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = str(exc)
+        set_terminal_phase(patch, Phase.FAILED, str(exc))
         return
     if scheduled_time is not None and datetime.now(UTC) < scheduled_time:
         return
@@ -459,8 +468,9 @@ def reconcile_pending(spec, name, status, patch, body):
     # --- Workload admitted ---
     assigned_cluster = KueueClient.get_assigned_flavor(wl)
     if not assigned_cluster:
-        patch.status["phase"] = Phase.FAILED
-        patch.status["message"] = "Workload admitted but no flavor assigned"
+        set_terminal_phase(
+            patch, Phase.FAILED, "Workload admitted but no flavor assigned"
+        )
         set_condition(
             patch,
             conditions,

fournos/handlers/resolving.py

@@ -21,15 +21,15 @@
     COND_WORKLOAD_ADMITTED,
     owner_ref,
     set_condition,
+    set_terminal_phase,
 )
 
 logger = logging.getLogger(__name__)
 
 
 def _resolve_failed(patch, conditions, name, message, *, reason, cond_message=None):
     """Set phase=Failed with a Resolved=False condition."""
-    patch.status["phase"] = Phase.FAILED
-    patch.status["message"] = message
+    set_terminal_phase(patch, Phase.FAILED, message)
     set_condition(
         patch,
         conditions,

fournos/handlers/status.py

@@ -3,6 +3,9 @@
 from __future__ import annotations
 
 import datetime
+import logging
+
+from fournos.core.constants import TERMINAL_PHASES, Phase
 
 CRD_GROUP = "fournos.dev"
 CRD_VERSION = "v1"
@@ -11,6 +14,8 @@
 COND_WORKLOAD_ADMITTED = "WorkloadAdmitted"
 COND_PIPELINE_RUN_READY = "PipelineRunReady"
 
+logger = logging.getLogger(__name__)
+
 
 def owner_ref(body: dict) -> dict:
     """Build a Kubernetes ownerReference pointing at the given FournosJob."""
@@ -60,6 +65,17 @@ def set_condition(
     patch.status["conditions"] = result
 
 
+def set_terminal_phase(patch, phase: str, message: str) -> None:
+    """Set phase, message, and completionTime for a terminal transition."""
+    if phase not in TERMINAL_PHASES:
+        logger.error("set_terminal_phase called with non-terminal phase %r", phase)
+        message = f"{message} (internal error: {phase!r} is not a terminal phase)"
+        phase = Phase.FAILED
+    patch.status["phase"] = phase
+    patch.status["message"] = message
+    patch.status["completionTime"] = utcnow()
+
+
 def create_workload_for_job(spec, name, body):
     """Create a Kueue Workload with cluster-slot reservation."""
     from fournos.state import ctx