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
+
+    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)

fournos/operator.py

@@ -9,13 +9,15 @@
 import logging
 import threading
 import time
+from datetime import UTC, datetime
 
 import kopf
 from kubernetes import client, config
 
 from fournos import __version__, handlers
 from fournos.core.clusters import ClusterRegistry
-from fournos.core.constants import LABEL_JOB_NAME, Phase
+from fournos.core.constants import LABEL_JOB_NAME, TERMINAL_PHASES, Phase
+from fournos.core.duration import parse_duration
 from fournos.core.kueue import KueueClient
 from fournos.core.resolve import ResolveClient
 from fournos.core.tekton import TektonClient
@@ -142,6 +144,10 @@ def _gc_loop():
             _gc_stale_resources()
         except Exception:
             logger.exception("Resource GC failed")
+        try:
+            _gc_expired_jobs()
+        except Exception:
+            logger.exception("TTL job GC failed")
 
 
 def _gc_stale_resources():
@@ -165,3 +171,75 @@ def _gc_stale_resources():
         if job_name and job_name not in job_names:
             logger.info("GC: deleting stale PipelineRun for job %s", job_name)
             ctx.tekton.delete_pipeline_run(job_name)
+
+
+# ---------------------------------------------------------------------------
+# TTL GC — delete terminal FournosJobs whose TTL has expired
+# ---------------------------------------------------------------------------
+
+
+def _get_completion_time(job: dict) -> datetime | None:
+    """Return the time the job entered its terminal phase, or None."""
+    conditions = job.get("status", {}).get("conditions") or []
+    phase = job.get("status", {}).get("phase", "")
+    for cond in reversed(conditions):
+        if cond.get("reason") == phase and cond.get("lastTransitionTime"):
+            try:
+                ts = datetime.fromisoformat(cond["lastTransitionTime"])
+                if ts.tzinfo is None:
+                    ts = ts.replace(tzinfo=UTC)
+                return ts
+            except (ValueError, TypeError):
+                pass
+    return None
+
+
+def _gc_expired_jobs():
+    custom = client.CustomObjectsApi()
+    jobs = custom.list_namespaced_custom_object(
+        "fournos.dev",
+        "v1",
+        settings.workload_namespace,
+        "fournosjobs",
+    )
+
+    now = datetime.now(UTC)
+    for job in jobs.get("items", []):
+        name = job["metadata"]["name"]
+        phase = job.get("status", {}).get("phase", "")
+        if phase not in TERMINAL_PHASES:
+            continue
+
+        ttl_raw = job.get("spec", {}).get("ttl")
+        if not ttl_raw:
+            continue
+
+        ttl = parse_duration(ttl_raw)
+        if ttl is None:
+            logger.warning(
+                "TTL GC: job %s has invalid ttl %r, ignoring", name, ttl_raw
+            )
+            continue
+
+        completion_time = _get_completion_time(job)
+        if completion_time is None:
+            logger.warning(
+                "TTL GC: job %s has ttl but no completion timestamp in conditions",
+                name,
+            )
+            continue
+
+        if now >= completion_time + ttl:
+            logger.info("TTL GC: deleting expired job %s (ttl=%s)", name, ttl_raw)
+            try:
+                custom.delete_namespaced_custom_object(
+                    "fournos.dev",
+                    "v1",
+                    settings.workload_namespace,
+                    "fournosjobs",
+                    name,
+                )
+            except client.exceptions.ApiException as exc:
+                logger.error(
+                    "TTL GC: failed to delete job %s: %s", name, exc.reason
+                )

manifests/crd.yaml

@@ -172,6 +172,13 @@ spec:
                     tasks). "Terminate" cancels immediately (Cancelled — skips
                     finally tasks). Both wait for the PipelineRun to finish
                     before releasing Kueue quota.
+                ttl:
+                  type: string
+                  description: >-
+                    Duration after job completion (Succeeded/Failed/Stopped)
+                    before the FournosJob CR is automatically deleted. Uses
+                    Go duration format (e.g. "12h", "30m", "7d"). When not
+                    set, the job is never auto-pruned.
             status:
               type: object
               properties:

tests/test_ttl.py

@@ -0,0 +1,165 @@
+"""TTL GC tests — verify that terminal FournosJobs with an expired TTL
+are deleted by the background GC loop, and that the duration parser works.
+"""
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fournos.core.duration import parse_duration
+from fournos.operator import _gc_expired_jobs, _get_completion_time
+
+# ---------------------------------------------------------------------------
+# Duration parser unit tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        ("12h", timedelta(hours=12)),
+        ("30m", timedelta(minutes=30)),
+        ("7d", timedelta(days=7)),
+        ("1h30m", timedelta(hours=1, minutes=30)),
+        ("2d6h30m", timedelta(days=2, hours=6, minutes=30)),
+        ("90s", timedelta(seconds=90)),
+        ("1d0h0m0s", timedelta(days=1)),
+        ("", None),
+        ("invalid", None),
+        ("abc123", None),
+    ],
+)
+def test_parse_duration(value, expected):
+    assert parse_duration(value) == expected
+
+
+# ---------------------------------------------------------------------------
+# _get_completion_time unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_get_completion_time_from_conditions():
+    job = {
+        "status": {
+            "phase": "Succeeded",
+            "conditions": [
+                {
+                    "type": "PipelineRunReady",
+                    "status": "True",
+                    "reason": "Succeeded",
+                    "lastTransitionTime": "2026-08-20T10:00:00Z",
+                },
+            ],
+        },
+    }
+    result = _get_completion_time(job)
+    assert result == datetime(2026, 8, 20, 10, 0, 0, tzinfo=UTC)
+
+
+def test_get_completion_time_no_matching_condition():
+    job = {
+        "status": {
+            "phase": "Succeeded",
+            "conditions": [
+                {
+                    "type": "WorkloadAdmitted",
+                    "status": "True",
+                    "reason": "Admitted",
+                    "lastTransitionTime": "2026-08-20T09:00:00Z",
+                },
+            ],
+        },
+    }
+    assert _get_completion_time(job) is None
+
+
+def test_get_completion_time_no_conditions():
+    job = {"status": {"phase": "Failed"}}
+    assert _get_completion_time(job) is None
+
+
+# ---------------------------------------------------------------------------
+# _gc_expired_jobs integration-style unit tests (mocked K8s client)
+# ---------------------------------------------------------------------------
+
+
+def _make_terminal_job(name, phase, ttl, completed_at):
+    """Build a minimal FournosJob dict in a terminal phase."""
+    return {
+        "metadata": {"name": name},
+        "spec": {"ttl": ttl},
+        "status": {
+            "phase": phase,
+            "conditions": [
+                {
+                    "type": "PipelineRunReady",
+                    "status": "True" if phase == "Succeeded" else "False",
+                    "reason": phase,
+                    "lastTransitionTime": completed_at,
+                },
+            ],
+        },
+    }
+
+
+@patch("fournos.operator.client")
+@patch("fournos.operator.settings")
+def test_gc_expired_jobs_deletes_expired(mock_settings, mock_client):
+    mock_settings.workload_namespace = "test-ns"
+
+    expired_job = _make_terminal_job(
+        "old-job", "Succeeded", "1h", "2026-08-20T08:00:00Z"
+    )
+    fresh_job = _make_terminal_job(
+        "fresh-job", "Succeeded", "1h", "2099-12-31T23:00:00Z"
+    )
+    no_ttl_job = {
+        "metadata": {"name": "no-ttl"},
+        "spec": {},
+        "status": {
+            "phase": "Failed",
+            "conditions": [
+                {
+                    "type": "PipelineRunReady",
+                    "status": "False",
+                    "reason": "Failed",
+                    "lastTransitionTime": "2020-01-01T00:00:00Z",
+                }
+            ],
+        },
+    }
+    running_job = {
+        "metadata": {"name": "running"},
+        "spec": {"ttl": "1h"},
+        "status": {"phase": "Running", "conditions": []},
+    }
+
+    mock_custom = MagicMock()
+    mock_client.CustomObjectsApi.return_value = mock_custom
+    mock_custom.list_namespaced_custom_object.return_value = {
+        "items": [expired_job, fresh_job, no_ttl_job, running_job]
+    }
+
+    _gc_expired_jobs()
+
+    mock_custom.delete_namespaced_custom_object.assert_called_once_with(
+        "fournos.dev", "v1", "test-ns", "fournosjobs", "old-job"
+    )
+
+
+@patch("fournos.operator.client")
+@patch("fournos.operator.settings")
+def test_gc_expired_jobs_no_deletions_when_none_expired(mock_settings, mock_client):
+    mock_settings.workload_namespace = "test-ns"
+
+    fresh_job = _make_terminal_job(
+        "fresh-job", "Failed", "12h", "2099-12-31T23:00:00Z"
+    )
+
+    mock_custom = MagicMock()
+    mock_client.CustomObjectsApi.return_value = mock_custom
+    mock_custom.list_namespaced_custom_object.return_value = {"items": [fresh_job]}
+
+    _gc_expired_jobs()
+
+    mock_custom.delete_namespaced_custom_object.assert_not_called()