feat(cron): Show health status in job list

Part of #349. Shows current vs. cleared errors, success history.

cron/jobs.py

@@ -547,20 +547,30 @@ def resume_job(job_id: str) -> Optional[Dict[str, Any]]:
 
 
 def trigger_job(job_id: str) -> Optional[Dict[str, Any]]:
-    """Schedule a job to run on the next scheduler tick."""
+    """Schedule a job to run on the next scheduler tick.
+    
+    Clears stale error state when re-triggering a previously-failed job
+    so the stale failure doesn't persist until the next tick completes.
+    """
     job = get_job(job_id)
     if not job:
         return None
-    return update_job(
-        job_id,
-        {
-            "enabled": True,
-            "state": "scheduled",
-            "paused_at": None,
-            "paused_reason": None,
-            "next_run_at": _hermes_now().isoformat(),
-        },
-    )
+    
+    updates = {
+        "enabled": True,
+        "state": "scheduled",
+        "paused_at": None,
+        "paused_reason": None,
+        "next_run_at": _hermes_now().isoformat(),
+    }
+    
+    # Clear stale error state when re-triggering
+    if job.get("last_status") == "error":
+        updates["last_status"] = "retrying"
+        updates["last_error"] = None
+        updates["error_cleared_at"] = _hermes_now().isoformat()
+    
+    return update_job(job_id, updates)
 
 
 def run_job_now(job_id: str) -> Optional[Dict[str, Any]]:
@@ -618,6 +628,7 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None):
     
     Updates last_run_at, last_status, increments completed count,
     computes next_run_at, and auto-deletes if repeat limit reached.
+    Tracks health timestamps for error/success history.
     """
     jobs = load_jobs()
     for i, job in enumerate(jobs):
@@ -627,6 +638,18 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None):
             job["last_status"] = "ok" if success else "error"
             job["last_error"] = error if not success else None
             
+            # Track health timestamps
+            if success:
+                job["last_success_at"] = now
+                # Clear stale error tracking on success
+                if job.get("last_error_at"):
+                    job["error_resolved_at"] = now
+            else:
+                job["last_error_at"] = now
+                # Clear resolved tracking on new error
+                if job.get("error_resolved_at"):
+                    del job["error_resolved_at"]
+            
             # Increment completed count
             if job.get("repeat"):
                 job["repeat"]["completed"] = job["repeat"].get("completed", 0) + 1
@@ -656,6 +679,32 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None):
     save_jobs(jobs)
 
 
+
+def clear_job_error(job_id: str) -> Optional[Dict[str, Any]]:
+    """
+    Clear stale error state for a job.
+    
+    Resets last_status to 'ok', last_error to None, and 
+    records when the error was cleared. Useful after auth
+    recovery when the job itself is healthy but stale error
+    state persists.
+    
+    Returns:
+        Updated job dict, or None if not found.
+    """
+    jobs = load_jobs()
+    for job in jobs:
+        if job["id"] == job_id:
+            job["last_status"] = "ok"
+            job["last_error"] = None
+            job["error_cleared_at"] = _hermes_now().isoformat()
+            save_jobs(jobs)
+            return job
+    save_jobs(jobs)
+    return None
+
+
+
 def advance_next_run(job_id: str) -> bool:
     """Preemptively advance next_run_at for a recurring job before execution.
 

hermes_cli/cron.py

@@ -93,6 +93,39 @@ def cron_list(show_all: bool = False):
         script = job.get("script")
         if script:
             print(f"    Script:    {script}")
+        
+        # Show health status
+        last_status = job.get("last_status")
+        last_error = job.get("last_error")
+        last_error_at = job.get("last_error_at")
+        last_success_at = job.get("last_success_at")
+        error_cleared_at = job.get("error_cleared_at")
+        error_resolved_at = job.get("error_resolved_at")
+        
+        if last_status == "error" and last_error:
+            if error_cleared_at or error_resolved_at:
+                # Error was cleared/resolved
+                cleared_time = error_cleared_at or error_resolved_at
+                print(color(f"    Status:      ok (error cleared)", Colors.GREEN))
+                print(color(f"    Last error:  {last_error[:80]}...", Colors.DIM))
+                print(color(f"    Resolved:    {cleared_time}", Colors.DIM))
+            else:
+                # Current error
+                print(color(f"    Status:      ERROR", Colors.RED))
+                print(color(f"    Error:       {last_error[:80]}...", Colors.RED))
+                if last_error_at:
+                    print(color(f"    Since:       {last_error_at}", Colors.RED))
+        elif last_status == "retrying":
+            print(color(f"    Status:      retrying (error cleared)", Colors.YELLOW))
+        elif last_status == "ok":
+            if last_success_at:
+                print(color(f"    Status:      ok (last success: {last_success_at})", Colors.GREEN))
+        elif last_status:
+            print(f"    Status:      {last_status}")
+        
+        # Show success history if available
+        if last_success_at and last_status != "error":
+            print(f"    Last ok:     {last_success_at}")
         print()
 
     from hermes_cli.gateway import find_gateway_pids
@@ -222,7 +255,18 @@ def cron_edit(args):
 
 
 def _job_action(action: str, job_id: str, success_verb: str, now: bool = False) -> int:
-    if action == "run" and now:
+    if action == "clear_error":
+        result = _cron_api(action="clear_error", job_id=job_id)
+        if not result.get("success"):
+            print(color(f"Failed to clear error: {result.get('error', 'unknown error')}", Colors.RED))
+            return 1
+        job = result.get("job", {})
+        name = job.get("name", job_id)
+        print(color(f"Cleared stale error state for job '{name}'", Colors.GREEN))
+        if job.get("error_cleared_at"):
+            print(f"  Cleared at: {job['error_cleared_at']}")
+        return 0
+if action == "run" and now:
         # Synchronous execution — run job immediately and show result
         result = _cron_api(action="run_now", job_id=job_id)
         if not result.get("success"):
@@ -292,9 +336,13 @@ def cron_command(args):
         now = getattr(args, 'now', False)
         return _job_action("run", args.job_id, "Triggered", now=now)
 
+    
+    if subcmd == "clear-error":
+        return _job_action("clear_error", args.job_id, "Cleared")
+    
     if subcmd in {"remove", "rm", "delete"}:
         return _job_action("remove", args.job_id, "Removed")
 
     print(f"Unknown cron command: {subcmd}")
-    print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|status|tick]")
+    print("Usage: hermes cron [list|create|edit|pause|resume|run|remove|clear-error|status|tick]")
     sys.exit(1)

hermes_cli/main.py

@@ -4576,6 +4576,9 @@ For more help on a command:
     cron_run.add_argument("job_id", help="Job ID to trigger")
     cron_run.add_argument("--now", action="store_true", help="Execute immediately and wait for result (clears stale errors)")
 
+    cron_clear_error = cron_subparsers.add_parser("clear-error", help="Clear stale error state for a job")
+    cron_clear_error.add_argument("job_id", help="Job ID to clear error for")
+
     cron_remove = cron_subparsers.add_parser("remove", aliases=["rm", "delete"], help="Remove a scheduled job")
     cron_remove.add_argument("job_id", help="Job ID to remove")
 

tools/cronjob_tools.py

@@ -201,6 +201,17 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
         "paused_at": job.get("paused_at"),
         "paused_reason": job.get("paused_reason"),
     }
+    # Health timestamps
+    if job.get("last_error_at"):
+        result["last_error_at"] = job["last_error_at"]
+    if job.get("last_success_at"):
+        result["last_success_at"] = job["last_success_at"]
+    if job.get("error_resolved_at"):
+        result["error_resolved_at"] = job["error_resolved_at"]
+    if job.get("error_cleared_at"):
+        result["error_cleared_at"] = job["error_cleared_at"]
+    
+
     if job.get("script"):
         result["script"] = job["script"]
     return result
@@ -326,6 +337,13 @@ def cronjob(
             if result is None:
                 return json.dumps({"success": False, "error": "Job not found"}, indent=2)
             return json.dumps(result, indent=2)
+        if normalized == "clear_error":
+            from cron.jobs import clear_job_error
+            job = clear_job_error(job_id)
+            if job is None:
+                return json.dumps({"success": False, "error": "Job not found"}, indent=2)
+            return json.dumps({"success": True, "job": _format_job(job)}, indent=2)
+
 
         if normalized == "update":
             updates: Dict[str, Any] = {}