diff --git a/job/health.go b/job/health.go
new file mode 100644
index 00000000..96d4d6b1
--- /dev/null
+++ b/job/health.go
@@ -0,0 +1,445 @@
+package job
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/kun/log"
+)
+
+// HealthChecker manages job health monitoring
+type HealthChecker struct {
+	interval time.Duration
+	ctx      context.Context
+	cancel   context.CancelFunc
+}
+
+var globalHealthChecker *HealthChecker
+
+// NewHealthChecker creates a new health checker
+func NewHealthChecker(interval time.Duration) *HealthChecker {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &HealthChecker{
+		interval: interval,
+		ctx:      ctx,
+		cancel:   cancel,
+	}
+}
+
+// Start starts the health check goroutine
+func (hc *HealthChecker) Start() {
+	ticker := time.NewTicker(hc.interval)
+	defer ticker.Stop()
+
+	log.Info("Job health checker started with interval: %v", hc.interval)
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := hc.performHealthCheck(); err != nil {
+				log.Error("Health check failed: %v", err)
+			}
+		case <-hc.ctx.Done():
+			log.Info("Job health checker stopped")
+			return
+		}
+	}
+}
+
+// Stop stops the health checker
+func (hc *HealthChecker) Stop() {
+	if hc.cancel != nil {
+		hc.cancel()
+	}
+}
+
+// performHealthCheck performs health check
+func (hc *HealthChecker) performHealthCheck() error {
+	log.Debug("Starting health check...")
+
+	// 1. Query all running jobs
+	runningJobs, err := hc.getRunningJobs()
+	if err != nil {
+		return fmt.Errorf("failed to get running jobs: %w", err)
+	}
+
+	if len(runningJobs) == 0 {
+		log.Debug("No running jobs found")
+		return nil
+	}
+
+	log.Debug("Found %d running jobs to check", len(runningJobs))
+
+	// 2. Check worker status for each job
+	wm := GetWorkerManager()
+	for _, job := range runningJobs {
+		if err := hc.checkJobHealth(job, wm); err != nil {
+			log.Error("Failed to check job %s health: %v", job.JobID, err)
+		}
+	}
+
+	return nil
+}
+
+// getRunningJobs gets all jobs with running status
+func (hc *HealthChecker) getRunningJobs() ([]*Job, error) {
+	mod := model.Select("__yao.job")
+	if mod == nil {
+		return nil, fmt.Errorf("job model not found")
+	}
+
+	param := model.QueryParam{
+		Select: JobFields,
+		Wheres: []model.QueryWhere{
+			{Column: "status", Value: "running"},
+			{Column: "enabled", Value: true},
+		},
+	}
+
+	results, err := mod.Get(param)
+	if err != nil {
+		// If database is closed during testing, return empty slice instead of error
+		if err.Error() == "sql: database is closed" {
+			log.Debug("Database is closed, returning empty job list")
+			return []*Job{}, nil
+		}
+		return nil, err
+	}
+
+	jobs := make([]*Job, 0, len(results))
+	for _, result := range results {
+		job := &Job{}
+		if err := mapToStruct(result, job); err != nil {
+			log.Warn("Failed to parse job data: %v", err)
+			continue
+		}
+		jobs = append(jobs, job)
+	}
+
+	return jobs, nil
+}
+
+// checkJobHealth checks the health status of a single job
+func (hc *HealthChecker) checkJobHealth(job *Job, wm *WorkerManager) error {
+	// Get current execution record for the job
+	if job.CurrentExecutionID == nil || *job.CurrentExecutionID == "" {
+		// No current execution ID but status is running, this is abnormal
+		return hc.markJobAsFailed(job, "Job status is running but no current execution ID found")
+	}
+
+	execution, err := GetExecution(*job.CurrentExecutionID, model.QueryParam{})
+	if err != nil {
+		return hc.markJobAsFailed(job, fmt.Sprintf("Failed to get execution: %v", err))
+	}
+
+	// Check execution status
+	if execution.Status != "running" {
+		// Execution status is not running but job status is running, this is inconsistent
+		return hc.markJobAsFailed(job, fmt.Sprintf("Job status is running but execution status is %s", execution.Status))
+	}
+
+	// Check if worker exists and is working
+	if execution.WorkerID == nil || *execution.WorkerID == "" {
+		return hc.markJobAsFailed(job, "Execution is running but no worker ID assigned")
+	}
+
+	// Check if worker still exists in active workers
+	if !hc.isWorkerActive(*execution.WorkerID, wm) {
+		return hc.markJobAsFailed(job, fmt.Sprintf("Worker %s is no longer active", *execution.WorkerID))
+	}
+
+	// Check if execution has timed out
+	if hc.isExecutionTimeout(execution) {
+		return hc.markJobAsFailed(job, "Execution has timed out")
+	}
+
+	log.Debug("Job %s health check passed", job.JobID)
+	return nil
+}
+
+// isWorkerActive checks if worker is still active
+func (hc *HealthChecker) isWorkerActive(workerID string, wm *WorkerManager) bool {
+	wm.mu.RLock()
+	defer wm.mu.RUnlock()
+
+	_, exists := wm.activeWorkers[workerID]
+	return exists
+}
+
+// isExecutionTimeout checks if execution has timed out
+func (hc *HealthChecker) isExecutionTimeout(execution *Execution) bool {
+	if execution.StartedAt == nil {
+		return false // No start time, cannot determine timeout
+	}
+
+	// Only check timeout if timeout is explicitly set
+	if execution.TimeoutSeconds == nil || *execution.TimeoutSeconds <= 0 {
+		return false // No timeout configured, execution can run indefinitely
+	}
+
+	timeoutDuration := time.Duration(*execution.TimeoutSeconds) * time.Second
+	return time.Since(*execution.StartedAt) > timeoutDuration
+}
+
+// markJobAsFailed marks a job as failed
+func (hc *HealthChecker) markJobAsFailed(job *Job, reason string) error {
+	log.Warn("Marking job %s as failed: %s", job.JobID, reason)
+
+	// Update job status
+	job.Status = "failed"
+	if err := SaveJob(job); err != nil {
+		return fmt.Errorf("failed to update job status: %w", err)
+	}
+
+	// Update execution status (if exists)
+	if job.CurrentExecutionID != nil && *job.CurrentExecutionID != "" {
+		execution, err := GetExecution(*job.CurrentExecutionID, model.QueryParam{})
+		if err == nil {
+			execution.Status = "failed"
+			now := time.Now()
+			execution.EndedAt = &now
+
+			// Set error information
+			errorInfo := map[string]interface{}{
+				"error":  reason,
+				"time":   now,
+				"source": "health_checker",
+			}
+			errorData, _ := jsoniter.Marshal(errorInfo)
+			execution.ErrorInfo = (*json.RawMessage)(&errorData)
+
+			if err := SaveExecution(execution); err != nil {
+				log.Error("Failed to update execution status: %v", err)
+			}
+		}
+	}
+
+	// Record log entry
+	logEntry := &Log{
+		JobID:       job.JobID,
+		Level:       "error",
+		Message:     fmt.Sprintf("Job marked as failed by health checker: %s", reason),
+		ExecutionID: job.CurrentExecutionID,
+		Source:      stringPtr("health_checker"),
+		Timestamp:   time.Now(),
+		Sequence:    0,
+	}
+
+	if err := SaveLog(logEntry); err != nil {
+		log.Error("Failed to save health check log: %v", err)
+	}
+
+	// Clear current execution ID
+	job.CurrentExecutionID = nil
+	if err := SaveJob(job); err != nil {
+		log.Error("Failed to clear current execution ID: %v", err)
+	}
+
+	return nil
+}
+
+// stringPtr returns a string pointer
+func stringPtr(s string) *string {
+	return &s
+}
+
+// GetHealthChecker gets the global health checker (if needed)
+// Note: Health checker is now started in job.go init() function
+func GetHealthChecker() *HealthChecker {
+	return globalHealthChecker
+}
+
+// ========================
+// Data Cleaner - Independent cleanup functionality
+// ========================
+
+// DataCleaner manages cleanup of old job data
+type DataCleaner struct {
+	ctx             context.Context
+	cancel          context.CancelFunc
+	retentionDays   int
+	lastCleanupTime time.Time
+}
+
+var globalDataCleaner *DataCleaner
+
+// NewDataCleaner creates a new data cleaner
+func NewDataCleaner(retentionDays int) *DataCleaner {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &DataCleaner{
+		ctx:             ctx,
+		cancel:          cancel,
+		retentionDays:   retentionDays,
+		lastCleanupTime: time.Now(), // Initialize to avoid immediate cleanup on startup
+	}
+}
+
+// Start starts the daily data cleanup routine
+func (dc *DataCleaner) Start() {
+	// Check every hour if daily cleanup is needed
+	ticker := time.NewTicker(1 * time.Hour)
+	defer ticker.Stop()
+
+	log.Info("Data cleaner started with %d days retention", dc.retentionDays)
+
+	for {
+		select {
+		case <-ticker.C:
+			if dc.shouldRunCleanup() {
+				if err := dc.performCleanup(); err != nil {
+					log.Error("Data cleanup failed: %v", err)
+				} else {
+					dc.lastCleanupTime = time.Now()
+				}
+			}
+		case <-dc.ctx.Done():
+			log.Info("Data cleaner stopped")
+			return
+		}
+	}
+}
+
+// Stop stops the data cleaner
+func (dc *DataCleaner) Stop() {
+	if dc.cancel != nil {
+		dc.cancel()
+	}
+}
+
+// shouldRunCleanup checks if cleanup should run (once per day)
+func (dc *DataCleaner) shouldRunCleanup() bool {
+	return time.Since(dc.lastCleanupTime) >= 24*time.Hour
+}
+
+// performCleanup performs the actual data cleanup
+func (dc *DataCleaner) performCleanup() error {
+	log.Info("Starting daily data cleanup...")
+
+	cutoffTime := time.Now().AddDate(0, 0, -dc.retentionDays)
+
+	// Clean up jobs (excluding running jobs)
+	deletedJobs, err := dc.cleanupJobs(cutoffTime)
+	if err != nil {
+		return fmt.Errorf("failed to cleanup jobs: %w", err)
+	}
+
+	// Clean up executions
+	deletedExecutions, err := dc.cleanupExecutions(cutoffTime)
+	if err != nil {
+		return fmt.Errorf("failed to cleanup executions: %w", err)
+	}
+
+	// Clean up logs
+	deletedLogs, err := dc.cleanupLogs(cutoffTime)
+	if err != nil {
+		return fmt.Errorf("failed to cleanup logs: %w", err)
+	}
+
+	log.Info("Data cleanup completed: %d jobs, %d executions, %d logs deleted",
+		deletedJobs, deletedExecutions, deletedLogs)
+
+	return nil
+}
+
+// cleanupJobs removes old jobs that are not running
+func (dc *DataCleaner) cleanupJobs(cutoffTime time.Time) (int, error) {
+	mod := model.Select("__yao.job")
+	if mod == nil {
+		return 0, fmt.Errorf("job model not found")
+	}
+
+	// Get jobs to delete (older than cutoff and not running)
+	param := model.QueryParam{
+		Select: []interface{}{"job_id"},
+		Wheres: []model.QueryWhere{
+			{Column: "created_at", OP: "<", Value: cutoffTime},
+			{Column: "status", OP: "!=", Value: "running"},
+		},
+	}
+
+	results, err := mod.Get(param)
+	if err != nil {
+		return 0, err
+	}
+
+	if len(results) == 0 {
+		return 0, nil
+	}
+
+	// Extract job IDs
+	jobIDs := make([]string, 0, len(results))
+	for _, result := range results {
+		if jobID, ok := result["job_id"].(string); ok {
+			jobIDs = append(jobIDs, jobID)
+		}
+	}
+
+	if len(jobIDs) == 0 {
+		return 0, nil
+	}
+
+	// Delete jobs
+	deleteParam := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "job_id", OP: "in", Value: jobIDs},
+		},
+	}
+
+	deleted, err := mod.DeleteWhere(deleteParam)
+	if err != nil {
+		return 0, err
+	}
+
+	return deleted, nil
+}
+
+// cleanupExecutions removes old executions
+func (dc *DataCleaner) cleanupExecutions(cutoffTime time.Time) (int, error) {
+	mod := model.Select("__yao.job.execution")
+	if mod == nil {
+		return 0, fmt.Errorf("job execution model not found")
+	}
+
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "created_at", OP: "<", Value: cutoffTime},
+		},
+	}
+
+	deleted, err := mod.DeleteWhere(param)
+	if err != nil {
+		return 0, err
+	}
+
+	return deleted, nil
+}
+
+// cleanupLogs removes old logs
+func (dc *DataCleaner) cleanupLogs(cutoffTime time.Time) (int, error) {
+	mod := model.Select("__yao.job.log")
+	if mod == nil {
+		return 0, fmt.Errorf("job log model not found")
+	}
+
+	param := model.QueryParam{
+		Wheres: []model.QueryWhere{
+			{Column: "created_at", OP: "<", Value: cutoffTime},
+		},
+	}
+
+	deleted, err := mod.DeleteWhere(param)
+	if err != nil {
+		return 0, err
+	}
+
+	return deleted, nil
+}
+
+// GetDataCleaner gets the global data cleaner
+func GetDataCleaner() *DataCleaner {
+	return globalDataCleaner
+}
diff --git a/job/health_test.go b/job/health_test.go
new file mode 100644
index 00000000..dad03155
--- /dev/null
+++ b/job/health_test.go
@@ -0,0 +1,462 @@
+package job_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/yao/config"
+	"github.com/yaoapp/yao/job"
+	"github.com/yaoapp/yao/test"
+)
+
+// registerHealthTestProcesses registers test processes for health testing
+func registerHealthTestProcesses() {
+	// Register a test process that simulates long-running execution
+	process.Register("test.health.longrunning", func(process *process.Process) interface{} {
+		args := process.Args
+		message := "Long running process"
+		if len(args) > 0 {
+			message = args[0].(string)
+		}
+
+		// Simulate long-running process by sleeping
+		time.Sleep(5 * time.Second)
+
+		return map[string]interface{}{
+			"message": message,
+			"status":  "success",
+		}
+	})
+
+	// Register a test process that simulates quick execution
+	process.Register("test.health.quick", func(process *process.Process) interface{} {
+		args := process.Args
+		message := "Quick process"
+		if len(args) > 0 {
+			message = args[0].(string)
+		}
+
+		return map[string]interface{}{
+			"message": message,
+			"status":  "success",
+		}
+	})
+}
+
+// TestHealthCheckerCreation tests health checker creation
+func TestHealthCheckerCreation(t *testing.T) {
+	// Test creating health checker with different intervals
+	hc := job.NewHealthChecker(10 * time.Second)
+	if hc == nil {
+		t.Fatal("Expected health checker to be created")
+	}
+
+	// Test stopping health checker
+	hc.Stop()
+}
+
+// TestHealthCheckerBasicFunction tests basic health checker functionality
+func TestHealthCheckerBasicFunction(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerHealthTestProcesses()
+
+	// Create a short-interval health checker for testing (2 seconds for fast testing)
+	hc := job.NewHealthChecker(2 * time.Second)
+	defer hc.Stop()
+
+	// Start health checker in background
+	go hc.Start()
+
+	// Wait a bit to let health checker run
+	time.Sleep(3 * time.Second)
+
+	t.Log("Health checker basic function test completed")
+}
+
+// TestHealthCheckerWithRunningJob tests health checker with actual running jobs
+func TestHealthCheckerWithRunningJob(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerHealthTestProcesses()
+
+	// Create a job that will run quickly
+	testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":        "Health Test Quick Job",
+		"description": "Job for testing health checker with quick execution",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create test job: %v", err)
+	}
+
+	// Add execution to the job
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"test_context": "health check test",
+		},
+	}, "test.health.quick", "Quick execution for health test")
+	if err != nil {
+		t.Fatalf("Failed to add execution: %v", err)
+	}
+
+	// Start the job
+	err = testJob.Push()
+	if err != nil {
+		t.Fatalf("Failed to start job: %v", err)
+	}
+
+	// Wait for job to complete
+	time.Sleep(2 * time.Second)
+
+	// Get updated job status
+	updatedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to get updated job: %v", err)
+	}
+
+	t.Logf("Job status after execution: %s", updatedJob.Status)
+}
+
+// TestHealthCheckerWithTimeoutJob tests health checker with timeout configuration
+func TestHealthCheckerWithTimeoutJob(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerHealthTestProcesses()
+
+	// Create a job with timeout
+	testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":            "Health Test Timeout Job",
+		"description":     "Job for testing health checker timeout handling",
+		"default_timeout": 2, // 2 seconds timeout
+	})
+	if err != nil {
+		t.Fatalf("Failed to create test job: %v", err)
+	}
+
+	// Add execution that will run longer than timeout
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"test_context": "timeout test",
+		},
+	}, "test.health.longrunning", "Long running execution for timeout test")
+	if err != nil {
+		t.Fatalf("Failed to add execution: %v", err)
+	}
+
+	// Start the job (this will run in background)
+	err = testJob.Push()
+	if err != nil {
+		t.Fatalf("Failed to start job: %v", err)
+	}
+
+	// Create health checker with short interval for testing
+	hc := job.NewHealthChecker(1 * time.Second)
+	defer hc.Stop()
+
+	// Start health checker
+	go hc.Start()
+
+	// Wait for health checker to detect and handle timeout
+	time.Sleep(8 * time.Second)
+
+	// Check if job was marked as failed due to timeout
+	updatedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to get updated job: %v", err)
+	}
+
+	t.Logf("Job status after timeout check: %s", updatedJob.Status)
+
+	// Note: We don't assert failed status here because the test process might complete
+	// before timeout is detected, depending on system performance
+}
+
+// TestHealthCheckerWithNoTimeoutJob tests health checker with jobs that have no timeout
+func TestHealthCheckerWithNoTimeoutJob(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerHealthTestProcesses()
+
+	// Create a job without timeout (should run indefinitely)
+	testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":        "Health Test No Timeout Job",
+		"description": "Job for testing health checker with no timeout",
+		// No default_timeout specified
+	})
+	if err != nil {
+		t.Fatalf("Failed to create test job: %v", err)
+	}
+
+	// Add execution
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"test_context": "no timeout test",
+		},
+	}, "test.health.quick", "Quick execution with no timeout")
+	if err != nil {
+		t.Fatalf("Failed to add execution: %v", err)
+	}
+
+	// Start the job
+	err = testJob.Push()
+	if err != nil {
+		t.Fatalf("Failed to start job: %v", err)
+	}
+
+	// Wait for execution to complete
+	time.Sleep(2 * time.Second)
+
+	// Check job status
+	updatedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to get updated job: %v", err)
+	}
+
+	t.Logf("Job status (no timeout): %s", updatedJob.Status)
+}
+
+// TestHealthCheckerStopAndStart tests stopping and starting health checker
+func TestHealthCheckerStopAndStart(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Create health checker
+	hc := job.NewHealthChecker(1 * time.Second)
+
+	// Start health checker
+	go hc.Start()
+
+	// Let it run for a bit
+	time.Sleep(2 * time.Second)
+
+	// Stop health checker
+	hc.Stop()
+
+	// Wait a bit to ensure it stops
+	time.Sleep(1 * time.Second)
+
+	t.Log("Health checker stop and start test completed")
+}
+
+// TestGlobalHealthChecker tests the global health checker functions
+func TestGlobalHealthChecker(t *testing.T) {
+	// Test getting global health checker
+	globalHC := job.GetHealthChecker()
+	if globalHC == nil {
+		t.Log("Global health checker is nil, this is expected if not initialized")
+	} else {
+		t.Log("Global health checker exists")
+	}
+
+	// Test stopping global health checker
+	job.StopHealthChecker()
+
+	t.Log("Global health checker test completed")
+}
+
+// TestHealthCheckerRestart tests restarting health checker with different intervals
+func TestHealthCheckerRestart(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Test restarting with different intervals
+	job.RestartHealthChecker(1 * time.Second)
+	time.Sleep(2 * time.Second)
+
+	job.RestartHealthChecker(3 * time.Second)
+	time.Sleep(1 * time.Second)
+
+	// Stop the health checker
+	job.StopHealthChecker()
+
+	t.Log("Health checker restart test completed")
+}
+
+// TestDataCleanerCreation tests data cleaner creation
+func TestDataCleanerCreation(t *testing.T) {
+	// Test creating data cleaner with different retention periods
+	dc := job.NewDataCleaner(30)
+	if dc == nil {
+		t.Fatal("Expected data cleaner to be created")
+	}
+
+	// Test stopping data cleaner
+	dc.Stop()
+}
+
+// TestDataCleanerBasicFunction tests basic data cleaner functionality
+func TestDataCleanerBasicFunction(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Create data cleaner with 1 day retention for testing
+	dc := job.NewDataCleaner(1)
+	defer dc.Stop()
+
+	// Start data cleaner (won't actually clean on first run due to initialization)
+	go dc.Start()
+
+	// Wait a bit
+	time.Sleep(1 * time.Second)
+
+	t.Log("Data cleaner basic function test completed")
+}
+
+// TestDataCleanupWithOldJobs tests data cleanup with old completed jobs
+func TestDataCleanupWithOldJobs(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerHealthTestProcesses()
+
+	// Create an old completed job (simulate by creating and completing it)
+	testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":        "Old Test Job",
+		"description": "Job for testing data cleanup",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create test job: %v", err)
+	}
+
+	// Add execution and complete it
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"test_context": "cleanup test",
+		},
+	}, "test.health.quick", "Quick execution for cleanup test")
+	if err != nil {
+		t.Fatalf("Failed to add execution: %v", err)
+	}
+
+	err = testJob.Push()
+	if err != nil {
+		t.Fatalf("Failed to start job: %v", err)
+	}
+
+	// Wait for job to complete
+	time.Sleep(2 * time.Second)
+
+	// Verify job is completed
+	updatedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to get updated job: %v", err)
+	}
+
+	t.Logf("Job status before cleanup: %s", updatedJob.Status)
+
+	// Test force cleanup (this won't delete recent jobs due to retention period)
+	err = job.ForceCleanup()
+	if err != nil {
+		t.Fatalf("Failed to force cleanup: %v", err)
+	}
+
+	// Verify job still exists (should not be deleted due to retention period)
+	_, err = job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Log("Job was cleaned up (expected if older than retention period)")
+	} else {
+		t.Log("Job still exists (expected for recent jobs)")
+	}
+}
+
+// TestGlobalDataCleaner tests global data cleaner functions
+func TestGlobalDataCleaner(t *testing.T) {
+	// Test getting global data cleaner
+	globalDC := job.GetDataCleaner()
+	if globalDC == nil {
+		t.Log("Global data cleaner is nil, this is expected if not initialized")
+	} else {
+		t.Log("Global data cleaner exists")
+	}
+
+	// Test stopping global data cleaner
+	job.StopDataCleaner()
+
+	t.Log("Global data cleaner test completed")
+}
+
+// TestHealthCheckerWithMultipleJobs tests health checker with multiple jobs
+func TestHealthCheckerWithMultipleJobs(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerHealthTestProcesses()
+
+	// Create multiple jobs
+	jobs := make([]*job.Job, 3)
+	for i := 0; i < 3; i++ {
+		testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
+			"name":        fmt.Sprintf("Health Test Job %d", i+1),
+			"description": fmt.Sprintf("Job %d for testing health checker with multiple jobs", i+1),
+		})
+		if err != nil {
+			t.Fatalf("Failed to create test job %d: %v", i+1, err)
+		}
+
+		// Add execution to each job
+		err = testJob.Add(&job.ExecutionOptions{
+			Priority: 1,
+			SharedData: map[string]interface{}{
+				"test_context": fmt.Sprintf("multi job test %d", i+1),
+			},
+		}, "test.health.quick", fmt.Sprintf("Execution for job %d", i+1))
+		if err != nil {
+			t.Fatalf("Failed to add execution to job %d: %v", i+1, err)
+		}
+
+		jobs[i] = testJob
+	}
+
+	// Start all jobs
+	for i, testJob := range jobs {
+		err := testJob.Push()
+		if err != nil {
+			t.Fatalf("Failed to start job %d: %v", i+1, err)
+		}
+	}
+
+	// Create health checker for monitoring
+	hc := job.NewHealthChecker(1 * time.Second)
+	defer hc.Stop()
+
+	// Start health checker
+	go hc.Start()
+
+	// Wait for jobs to complete and health checker to run
+	time.Sleep(5 * time.Second)
+
+	// Check status of all jobs
+	for i, testJob := range jobs {
+		updatedJob, err := job.GetJob(testJob.JobID)
+		if err != nil {
+			t.Errorf("Failed to get updated job %d: %v", i+1, err)
+			continue
+		}
+		t.Logf("Job %d status: %s", i+1, updatedJob.Status)
+	}
+}
diff --git a/job/job.go b/job/job.go
index e848cffe..a2d22532 100644
--- a/job/job.go
+++ b/job/job.go
@@ -12,6 +12,98 @@ import (
 	"github.com/yaoapp/kun/log"
 )
 
+// init initializes the job package
+func init() {
+	// Initialize and start health checker
+	initHealthChecker()
+
+	// Initialize and start data cleaner
+	initDataCleaner()
+
+	log.Info("Job package initialized with health checker and data cleaner")
+}
+
+// initHealthChecker initializes the health checker
+func initHealthChecker() {
+	// Get health check interval from configuration or use default
+	interval := getHealthCheckInterval()
+	globalHealthChecker = NewHealthChecker(interval)
+
+	// Start health check goroutine
+	go globalHealthChecker.Start()
+
+	log.Info("Job health checker started with %v interval", interval)
+}
+
+// getHealthCheckInterval returns the configured health check interval or default
+func getHealthCheckInterval() time.Duration {
+	// Default interval: 5 minutes (balanced between detection speed and resource usage)
+	// This is suitable for most job monitoring scenarios:
+	// - Short jobs (< 5min): Health check won't interfere much
+	// - Medium jobs (5min - 1h): Good detection without excessive overhead
+	// - Long jobs (> 1h): Timely detection of issues
+	defaultInterval := 5 * time.Minute
+
+	// TODO: Add configuration support from environment variables or config file
+	// For example:
+	// if envInterval := os.Getenv("YAO_JOB_HEALTH_CHECK_INTERVAL"); envInterval != "" {
+	//     if duration, err := time.ParseDuration(envInterval); err == nil {
+	//         return duration
+	//     }
+	// }
+
+	return defaultInterval
+}
+
+// StopHealthChecker stops the health checker
+func StopHealthChecker() {
+	if globalHealthChecker != nil {
+		globalHealthChecker.Stop()
+		log.Info("Job health checker stopped")
+	}
+}
+
+// RestartHealthChecker restarts the health checker with a new interval
+// This is useful for testing or dynamic configuration changes
+func RestartHealthChecker(interval time.Duration) {
+	// Stop existing health checker
+	StopHealthChecker()
+
+	// Create and start new health checker with specified interval
+	globalHealthChecker = NewHealthChecker(interval)
+	go globalHealthChecker.Start()
+
+	log.Info("Job health checker restarted with %v interval", interval)
+}
+
+// initDataCleaner initializes the data cleaner
+func initDataCleaner() {
+	// Create data cleaner with 90 days retention
+	retentionDays := 90
+	globalDataCleaner = NewDataCleaner(retentionDays)
+
+	// Start data cleaner goroutine
+	go globalDataCleaner.Start()
+
+	log.Info("Job data cleaner started with %d days retention", retentionDays)
+}
+
+// StopDataCleaner stops the data cleaner
+func StopDataCleaner() {
+	if globalDataCleaner != nil {
+		globalDataCleaner.Stop()
+		log.Info("Job data cleaner stopped")
+	}
+}
+
+// ForceCleanup forces an immediate data cleanup (useful for testing)
+func ForceCleanup() error {
+	if globalDataCleaner != nil {
+		return globalDataCleaner.performCleanup()
+	}
+	return fmt.Errorf("data cleaner not initialized")
+}
+
 // Once create a new job
 func Once(mode ModeType, data map[string]interface{}) (*Job, error) {
 	data["mode"] = mode