Merge pull request #283 from tbrandenburg/codex/improve-crontab-task-execution-reliability

Author: tbrandenburg

packages/pybackend/app.py

@@ -52,6 +52,7 @@
 from harness_service import is_process_running, list_harnesses, run_harness
 from workflow_service import list_workspace_workflows, read_workflows, write_workflows
 from cron_service import (
+    get_cron_job_diagnostics,
     get_cron_job_last_runs,
     refresh_cron_clock,
     start_cron_clock,
@@ -601,7 +602,10 @@ def save_global_workflows(payload: dict = Body(...)):
 def workspace_workflows():
     try:
         logger.info("Listing workspace workflows")
-        return list_workspace_workflows(get_cron_job_last_runs())
+        return list_workspace_workflows(
+            get_cron_job_last_runs(),
+            get_cron_job_diagnostics(),
+        )
     except Exception as exc:
         logger.exception("Failed to list workspace workflows")
         raise HTTPException(

packages/pybackend/cron_service.py

@@ -18,10 +18,15 @@
 _state_lock = Lock()
 _started_jobs = 0
 _successful_jobs = 0
+_failed_jobs = 0
 _configured_jobs = 0
 _invalid_jobs = 0
 _started_at: datetime | None = None
 _last_run_by_job: dict[str, datetime] = {}
+_last_finished_by_job: dict[str, datetime] = {}
+_last_duration_ms_by_job: dict[str, int] = {}
+_last_exit_code_by_job: dict[str, int] = {}
+_last_error_by_job: dict[str, str] = {}
 _running_process_by_job: dict[str, subprocess.Popen[str]] = {}
 
 
@@ -48,7 +53,9 @@ def _resolve_script_path(repo_path: Path, shell_script_path: str) -> Path:
 
 
 def _run_workflow_script(repo_path: Path, workflow_id: str, script_path: Path) -> None:
-    global _started_jobs, _successful_jobs
+    global _started_jobs, _successful_jobs, _failed_jobs
+
+    started_at = datetime.now(timezone.utc)
 
     with _state_lock:
         _terminate_running_job(workflow_id)
@@ -66,28 +73,42 @@ def _run_workflow_script(repo_path: Path, workflow_id: str, script_path: Path) -
     logger.info("Running cron workflow '%s' in '%s'", workflow_id, repo_path)
     stdout, stderr = process.communicate()
     returncode = process.returncode
+    finished_at = datetime.now(timezone.utc)
+    duration_ms = int((finished_at - started_at).total_seconds() * 1000)
 
     with _state_lock:
         if _running_process_by_job.get(workflow_id) is process:
             _running_process_by_job.pop(workflow_id, None)
+        _last_finished_by_job[workflow_id] = finished_at
+        _last_duration_ms_by_job[workflow_id] = duration_ms
+        _last_exit_code_by_job[workflow_id] = returncode
 
     if returncode == 0:
         with _state_lock:
             _successful_jobs += 1
+            _last_error_by_job.pop(workflow_id, None)
         logger.info("Cron workflow '%s' completed", workflow_id)
         if stdout.strip():
             logger.info("Cron workflow '%s' stdout: %s", workflow_id, stdout.strip())
         return
 
+    with _state_lock:
+        _failed_jobs += 1
+
     logger.warning("Cron workflow '%s' failed with exit code %s", workflow_id, returncode)
     if stdout.strip():
         logger.warning("Cron workflow '%s' stdout: %s", workflow_id, stdout.strip())
     if stderr.strip():
+        with _state_lock:
+            _last_error_by_job[workflow_id] = stderr.strip()
         logger.warning("Cron workflow '%s' stderr: %s", workflow_id, stderr.strip())
+    else:
+        with _state_lock:
+            _last_error_by_job[workflow_id] = f"Exit code {returncode} without stderr"
 
 
 def start_cron_clock() -> None:
-    global _scheduler, _started_jobs, _successful_jobs, _configured_jobs, _invalid_jobs, _started_at
+    global _scheduler, _started_jobs, _successful_jobs, _failed_jobs, _configured_jobs, _invalid_jobs, _started_at
 
     if _scheduler is not None:
         return
@@ -110,12 +131,20 @@ def start_cron_clock() -> None:
             shell_script_path = workflow.get("shellScriptPath")
             workflow_id = workflow.get("id") or "workflow"
             if not isinstance(schedule, str) or not schedule.strip():
+                logger.warning("Skipping workflow '%s' in '%s': missing schedule", workflow_id, repo_name)
                 continue
             if not isinstance(shell_script_path, str) or not shell_script_path.strip():
+                logger.warning("Skipping workflow '%s' in '%s': missing shellScriptPath", workflow_id, repo_name)
                 continue
 
             script_path = _resolve_script_path(repo_path, shell_script_path)
             if not script_path.exists() or not script_path.is_file():
+                logger.warning(
+                    "Skipping workflow '%s' in '%s': script not found at '%s'",
+                    workflow_id,
+                    repo_name,
+                    script_path,
+                )
                 continue
 
             job_id = f"{repo_name}:{workflow_id}"
@@ -125,8 +154,9 @@ def start_cron_clock() -> None:
                     CronTrigger.from_crontab(schedule),
                     id=job_id,
                     replace_existing=True,
-                    max_instances=2,
+                    max_instances=1,
                     coalesce=True,
+                    misfire_grace_time=300,
                     args=[repo_path, job_id, script_path],
                 )
                 configured_jobs += 1
@@ -145,10 +175,15 @@ def start_cron_clock() -> None:
     with _state_lock:
         _started_jobs = 0
         _successful_jobs = 0
+        _failed_jobs = 0
         _configured_jobs = configured_jobs
         _invalid_jobs = invalid_jobs
         _started_at = datetime.now(timezone.utc)
         _last_run_by_job.clear()
+        _last_finished_by_job.clear()
+        _last_duration_ms_by_job.clear()
+        _last_exit_code_by_job.clear()
+        _last_error_by_job.clear()
         _running_process_by_job.clear()
 
     logger.info(
@@ -186,6 +221,7 @@ def get_cron_clock_status() -> dict[str, object]:
     with _state_lock:
         started_jobs = _started_jobs
         successful_jobs = _successful_jobs
+        failed_jobs = _failed_jobs
         configured_jobs = _configured_jobs
         invalid_jobs = _invalid_jobs
         started_at = _started_at
@@ -213,6 +249,7 @@ def get_cron_clock_status() -> dict[str, object]:
         "invalidSchedules": invalid_jobs,
         "startedJobsSinceStartup": started_jobs,
         "successfulJobsSinceStartup": successful_jobs,
+        "failedJobsSinceStartup": failed_jobs,
     }
 
 
@@ -231,3 +268,44 @@ def get_cron_job_last_runs() -> dict[str, str | None]:
         job_last_runs[job.id] = last_runs.get(job.id)
 
     return job_last_runs
+
+
+def get_cron_job_diagnostics() -> dict[str, dict[str, object | None]]:
+    if _scheduler is None:
+        return {}
+
+    with _state_lock:
+        last_runs = {
+            workflow_id: timestamp.isoformat()
+            for workflow_id, timestamp in _last_run_by_job.items()
+        }
+        finished_runs = {
+            workflow_id: timestamp.isoformat()
+            for workflow_id, timestamp in _last_finished_by_job.items()
+        }
+        durations = dict(_last_duration_ms_by_job)
+        exit_codes = dict(_last_exit_code_by_job)
+        errors = dict(_last_error_by_job)
+        running = {
+            workflow_id
+            for workflow_id, process in _running_process_by_job.items()
+            if process.poll() is None
+        }
+
+    diagnostics: dict[str, dict[str, object | None]] = {}
+    for job in _scheduler.get_jobs():
+        next_run_time = None
+        if job.next_run_time is not None:
+            next_run_time = job.next_run_time.isoformat()
+
+        diagnostics[job.id] = {
+            "lastStartedAt": last_runs.get(job.id),
+            "lastFinishedAt": finished_runs.get(job.id),
+            "lastDurationMs": durations.get(job.id),
+            "lastExitCode": exit_codes.get(job.id),
+            "lastError": errors.get(job.id),
+            "nextRunAt": next_run_time,
+            "running": job.id in running,
+        }
+
+    return diagnostics

packages/pybackend/tests/unit/test_api.py

@@ -1095,18 +1095,24 @@ def test_save_global_workflows_success(self, mock_write, mock_refresh):
         mock_write.assert_called_once_with({"workflows": []}, None)
         mock_refresh.assert_called_once_with()
 
+    @patch("app.get_cron_job_diagnostics")
     @patch("app.get_cron_job_last_runs")
     @patch("app.list_workspace_workflows")
-    def test_workspace_workflows_success(self, mock_list, mock_last_runs):
+    def test_workspace_workflows_success(self, mock_list, mock_last_runs, mock_diagnostics):
         mock_last_runs.return_value = {"sample:wf_1": "2026-01-02T03:04:05+00:00"}
+        mock_diagnostics.return_value = {"sample:wf_1": {"lastExitCode": 0, "running": False}}
         mock_list.return_value = {"workflows": [{"repository": "sample", "id": "wf_1", "name": "Release", "enabled": True, "schedule": None}]}
 
         response = client.get("/api/workspace/workflows")
 
         assert response.status_code == 200
         assert response.json()["workflows"][0]["repository"] == "sample"
         mock_last_runs.assert_called_once_with()
-        mock_list.assert_called_once_with({"sample:wf_1": "2026-01-02T03:04:05+00:00"})
+        mock_diagnostics.assert_called_once_with()
+        mock_list.assert_called_once_with(
+            {"sample:wf_1": "2026-01-02T03:04:05+00:00"},
+            {"sample:wf_1": {"lastExitCode": 0, "running": False}},
+        )
 
     @patch("app.read_workflows")
     @patch("app._repository_path")

packages/pybackend/tests/unit/test_cron_service.py

@@ -6,6 +6,11 @@
 def teardown_function():
     cron_service.stop_cron_clock()
     cron_service._running_process_by_job = {}
+    cron_service._last_run_by_job = {}
+    cron_service._last_finished_by_job = {}
+    cron_service._last_duration_ms_by_job = {}
+    cron_service._last_exit_code_by_job = {}
+    cron_service._last_error_by_job = {}
 
 
 @patch("cron_service.CronTrigger.from_crontab")
@@ -227,3 +232,42 @@ def test_get_cron_job_last_runs_includes_only_registered_jobs():
     assert result["repo-a:wf-1"] == "2026-01-02T03:04:05+00:00"
     assert result["repo-a:wf-2"] is None
     assert "repo-a:other" not in result
+
+
+def test_get_cron_job_diagnostics_includes_runtime_metadata():
+    cron_service._scheduler = MagicMock()
+    cron_service._scheduler.get_jobs.return_value = [
+        MagicMock(id="repo-a:wf-1", next_run_time=cron_service.datetime(2026, 1, 2, 4, 5, 6, tzinfo=cron_service.timezone.utc)),
+        MagicMock(id="repo-a:wf-2", next_run_time=None),
+    ]
+
+    active_process = MagicMock()
+    active_process.poll.return_value = None
+    cron_service._running_process_by_job = {"repo-a:wf-1": active_process}
+    cron_service._last_run_by_job = {
+        "repo-a:wf-1": cron_service.datetime(2026, 1, 2, 3, 4, 5, tzinfo=cron_service.timezone.utc),
+    }
+    cron_service._last_finished_by_job = {
+        "repo-a:wf-1": cron_service.datetime(2026, 1, 2, 3, 4, 8, tzinfo=cron_service.timezone.utc),
+    }
+    cron_service._last_duration_ms_by_job = {"repo-a:wf-1": 3123}
+    cron_service._last_exit_code_by_job = {"repo-a:wf-1": 0}
+    cron_service._last_error_by_job = {"repo-a:wf-2": "boom"}
+
+    result = cron_service.get_cron_job_diagnostics()
+
+    assert result["repo-a:wf-1"]["lastStartedAt"] == "2026-01-02T03:04:05+00:00"
+    assert result["repo-a:wf-1"]["lastFinishedAt"] == "2026-01-02T03:04:08+00:00"
+    assert result["repo-a:wf-1"]["lastDurationMs"] == 3123
+    assert result["repo-a:wf-1"]["lastExitCode"] == 0
+    assert result["repo-a:wf-1"]["nextRunAt"] == "2026-01-02T04:05:06+00:00"
+    assert result["repo-a:wf-1"]["running"] is True
+    assert result["repo-a:wf-2"]["lastError"] == "boom"
+
+
+def test_get_cron_job_diagnostics_returns_empty_when_scheduler_not_running():
+    cron_service._scheduler = None
+
+    result = cron_service.get_cron_job_diagnostics()
+
+    assert result == {}

packages/pybackend/tests/unit/test_workflow_service.py

@@ -85,7 +85,10 @@ def test_list_workspace_workflows_collects_repository_workflows(
             {"workflows": []},
         ]
 
-        result = list_workspace_workflows({"repo-a:wf_a": "2026-01-02T03:04:05+00:00"})
+        result = list_workspace_workflows(
+            {"repo-a:wf_a": "2026-01-02T03:04:05+00:00"},
+            {"repo-a:wf_a": {"lastExitCode": 0, "running": False}},
+        )
 
     assert result == {
         "workflows": [
@@ -97,6 +100,7 @@ def test_list_workspace_workflows_collects_repository_workflows(
                 "schedule": "* * * * *",
                 "shellScriptPath": None,
                 "lastRun": "2026-01-02T03:04:05+00:00",
+                "diagnostics": {"lastExitCode": 0, "running": False},
             }
         ]
     }

packages/pybackend/workflow_service.py

@@ -116,6 +116,7 @@ def write_workflows(
 
 def list_workspace_workflows(
     last_runs_by_job: dict[str, str | None] | None = None,
+    diagnostics_by_job: dict[str, dict[str, Any]] | None = None,
 ) -> dict[str, list[dict[str, Any]]]:
     workspace_home = get_workspace_home()
     workflows: list[dict[str, Any]] = []
@@ -138,6 +139,7 @@ def list_workspace_workflows(
                     "schedule": workflow.get("schedule"),
                     "shellScriptPath": workflow.get("shellScriptPath"),
                     "lastRun": (last_runs_by_job or {}).get(job_id),
+                    "diagnostics": (diagnostics_by_job or {}).get(job_id),
                 }
             )