feat: record bounded heartbeat and device transition diagnostics (#378)

server/app/goauto/device/diagnostics.go

@@ -0,0 +1,165 @@
+package device
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"go-admin/app/goauto/models"
+
+	log "github.com/go-admin-team/go-admin-core/logger"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+)
+
+const heartbeatHistoryBudget = 200 * time.Millisecond
+
+type heartbeatClientReport struct {
+	SnapshotSeq             int64      `json:"snapshotSeq"`
+	FailedRounds            int64      `json:"failedRounds"`
+	LatestFailureAt         *time.Time `json:"latestFailureAt"`
+	LatestFailureStep       *string    `json:"latestFailureStep"`
+	LatestFailureCategory   *string    `json:"latestFailureCategory"`
+	LatestFailureDurationMs *int64     `json:"latestFailureDurationMs"`
+	PreviousSyncDurationMs  *int64     `json:"previousSyncDurationMs"`
+	HasInterruptedRound     bool       `json:"hasInterruptedRound"`
+}
+
+func validateHeartbeatClientReport(raw json.RawMessage) (*string, string) {
+	if len(raw) == 0 {
+		return nil, "none"
+	}
+	invalid := func() (*string, string) { return nil, "client_report_invalid" }
+	if len(raw) > 2048 {
+		return invalid()
+	}
+	var fields map[string]json.RawMessage
+	if json.Unmarshal(raw, &fields) != nil || len(fields) != 8 {
+		return invalid()
+	}
+	// encoding/json matches struct keys case-insensitively. Require every
+	// exact contract key before decoding so aliases cannot hide a missing key.
+	for _, name := range []string{"snapshotSeq", "failedRounds", "latestFailureAt", "latestFailureStep", "latestFailureCategory", "latestFailureDurationMs", "previousSyncDurationMs", "hasInterruptedRound"} {
+		if _, ok := fields[name]; !ok {
+			return invalid()
+		}
+	}
+	for _, name := range []string{"snapshotSeq", "failedRounds", "hasInterruptedRound"} {
+		if v, ok := fields[name]; !ok || bytes.Equal(bytes.TrimSpace(v), []byte("null")) {
+			return invalid()
+		}
+	}
+	var report heartbeatClientReport
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.DisallowUnknownFields()
+	if decoder.Decode(&report) != nil || report.SnapshotSeq < 1 || report.SnapshotSeq > 9007199254740991 || report.FailedRounds < 0 || report.FailedRounds > 1000000 {
+		return invalid()
+	}
+	for _, v := range []*int64{report.LatestFailureDurationMs, report.PreviousSyncDurationMs} {
+		if v != nil && (*v < 0 || *v > 86400000) {
+			return invalid()
+		}
+	}
+	if report.FailedRounds == 0 {
+		if report.LatestFailureAt != nil || report.LatestFailureStep != nil || report.LatestFailureCategory != nil || report.LatestFailureDurationMs != nil {
+			return invalid()
+		}
+	} else {
+		if report.LatestFailureAt == nil || report.LatestFailureAt.IsZero() || report.LatestFailureStep == nil || report.LatestFailureCategory == nil || report.LatestFailureDurationMs == nil {
+			return invalid()
+		}
+		if !oneOf(*report.LatestFailureStep, "sync", "register", "heartbeat", "recover", "flush", "claim") || !oneOf(*report.LatestFailureCategory, "api_auth", "api_conflict", "api_server", "api_other", "timeout_connect", "timeout_response", "timeout_unknown", "network", "exception") {
+			return invalid()
+		}
+	}
+	canonical, _ := json.Marshal(report)
+	value := string(canonical)
+	return &value, "accepted"
+}
+func oneOf(value string, allowed ...string) bool {
+	for _, v := range allowed {
+		if value == v {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *Service) saveHeartbeatHistory(ctx context.Context, row models.AgentHeartbeatLog) {
+	// Client cancellation after the core commit must not suppress history. There
+	// is deliberately no goroutine: the caller waits at most this DB deadline.
+	writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), heartbeatHistoryBudget)
+	defer cancel()
+	err := s.DB.WithContext(writeCtx).Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent), SkipDefaultTransaction: true}).Create(&row).Error
+	if err != nil {
+		classification := "database_error"
+		if errors.Is(err, context.DeadlineExceeded) || writeCtx.Err() != nil {
+			classification = "deadline_exceeded"
+		}
+		log.Warnf("agent heartbeat history skipped: device_id=%d classification=%s", row.DeviceID, classification)
+	}
+}
+
+func insertDeviceStatusEvent(tx *gorm.DB, event *models.AgentDeviceStatusEvent) error {
+	// Never issue a diagnostic SELECT or log the SQL error/body.
+	if err := tx.Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)}).Create(event).Error; err != nil {
+		return internalError(errors.New("device status event write failed"))
+	}
+	return nil
+}
+func logDeviceStatusEvent(event models.AgentDeviceStatusEvent) {
+	log.Infof("agent device status changed: device_id=%d reason=%s failed_task_count=%d order_result_unknown_count=%d", event.DeviceID, event.Reason, event.FailedTaskCount, event.OrderResultUnknownCount)
+}
+
+const diagnosticsCleanupBatch = 5000
+
+// RunDiagnosticsCleanup shares the API server lifecycle but never occupies the
+// offline monitor's 15-second scan loop.
+func RunDiagnosticsCleanup(ctx context.Context, s *Service, onError func(error)) {
+	heartbeats := time.NewTicker(time.Hour)
+	defer heartbeats.Stop()
+	events := time.NewTicker(24 * time.Hour)
+	defer events.Stop()
+	runDiagnosticsCleanup(ctx, s, heartbeats.C, events.C, onError)
+}
+func runDiagnosticsCleanup(ctx context.Context, s *Service, heartbeats, events <-chan time.Time, onError func(error)) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-heartbeats:
+			if _, err := s.CleanupHeartbeatHistory(ctx); err != nil && ctx.Err() == nil && onError != nil {
+				onError(err)
+			}
+		case <-events:
+			if _, err := s.CleanupDeviceStatusEvents(ctx); err != nil && ctx.Err() == nil && onError != nil {
+				onError(err)
+			}
+		}
+	}
+}
+
+func (s *Service) CleanupHeartbeatHistory(ctx context.Context) (int64, error) {
+	return cleanupDiagnostics(s.DB.WithContext(ctx), &models.AgentHeartbeatLog{}, "received_at", s.Now().Add(-7*24*time.Hour))
+}
+func (s *Service) CleanupDeviceStatusEvents(ctx context.Context) (int64, error) {
+	return cleanupDiagnostics(s.DB.WithContext(ctx), &models.AgentDeviceStatusEvent{}, "occurred_at", s.Now().Add(-30*24*time.Hour))
+}
+func cleanupDiagnostics(db *gorm.DB, model any, timestamp string, cutoff time.Time) (int64, error) {
+	// A fixed-size ID batch works on both MySQL and SQLite. Recheck expiration
+	// in DELETE, and never loop through an unbounded backlog in one run.
+	var ids []uint64
+	if err := db.Model(model).Where(timestamp+" < ?", cutoff).Order(timestamp+", id").Limit(diagnosticsCleanupBatch).Pluck("id", &ids).Error; err != nil {
+		return 0, errors.New("diagnostics cleanup read failed")
+	}
+	if len(ids) == 0 {
+		return 0, nil
+	}
+	result := db.Where("id IN ? AND "+timestamp+" < ?", ids, cutoff).Delete(model)
+	if result.Error != nil {
+		return 0, errors.New("diagnostics cleanup delete failed")
+	}
+	return result.RowsAffected, nil
+}

server/app/goauto/device/diagnostics_mysql_test.go

@@ -0,0 +1,393 @@
+package device_test
+
+import (
+	"context"
+	"errors"
+	"net"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	. "go-admin/app/goauto/device"
+	"go-admin/app/goauto/models"
+	versionlocal "go-admin/cmd/migrate/migration/version-local"
+	common "go-admin/common/models"
+
+	driver "github.com/go-sql-driver/mysql"
+	"github.com/google/uuid"
+	"gorm.io/driver/mysql"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"gorm.io/gorm/logger"
+)
+
+// Explicit opt-in only; refuses production/remote DSNs and any nonempty schema.
+// Credentials are injected by the caller and never included in test errors.
+func TestMySQLAgentDiagnostics(t *testing.T) {
+	raw := os.Getenv("GOAUTO_378_TEST_MYSQL_DSN")
+	if raw == "" {
+		t.Skip("isolated local MySQL DSN not provided")
+	}
+	cfg, err := driver.ParseDSN(raw)
+	if err != nil {
+		t.Fatal("invalid test DSN")
+	}
+	host, _, err := net.SplitHostPort(cfg.Addr)
+	if err != nil || cfg.Net != "tcp" || (host != "127.0.0.1" && host != "localhost" && host != "::1") || cfg.DBName != "goauto_378_test" {
+		t.Fatal("requires local tcp database goauto_378_test")
+	}
+	cfg.ParseTime = true
+	db, err := gorm.Open(mysql.Open(cfg.FormatDSN()), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	if err != nil {
+		t.Fatal("test MySQL connection failed")
+	}
+	sqlDB, err := db.DB()
+	if err != nil {
+		t.Fatal("test database unavailable")
+	}
+	t.Cleanup(func() { _ = sqlDB.Close() })
+	sqlDB.SetMaxOpenConns(8)
+	var version string
+	if db.Raw("SELECT VERSION()").Scan(&version).Error != nil || !strings.HasPrefix(version, "8.") {
+		t.Fatal("requires MySQL 8")
+	}
+	tables, err := db.Migrator().GetTables()
+	if err != nil || len(tables) != 0 {
+		t.Fatal("test schema must be empty")
+	}
+	owned := []any{&common.Migration{}, &models.AgentDevice{}, &models.PDDProduct{}, &models.CollectionRule{}, &models.CollectionTask{}, &models.PurchaseTask{}, &models.PurchaseTaskAttempt{}, &models.PDDAccount{}, &models.ShopeeProduct{}, &models.SYBProduct{}}
+	t.Cleanup(func() {
+		for _, model := range append([]any{&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}}, owned...) {
+			if err := db.Migrator().DropTable(model); err != nil {
+				t.Errorf("test table cleanup failed for %T", model)
+			}
+		}
+	})
+	if err := db.AutoMigrate(owned...); err != nil {
+		t.Fatal("test base migration failed")
+	}
+	const migrationVersion = "1791600000000_agent_diagnostics"
+	for i := 0; i < 2; i++ {
+		if versionlocal.MigrateAgentDiagnostics(db, migrationVersion) != nil {
+			t.Fatal("diagnostic migration failed")
+		}
+	}
+	for _, idx := range []struct {
+		model any
+		name  string
+	}{{&models.AgentHeartbeatLog{}, "ix_heartbeat_device_received"}, {&models.AgentHeartbeatLog{}, "ix_heartbeat_received"}, {&models.AgentDeviceStatusEvent{}, "ix_device_event_occurred"}} {
+		if !db.Migrator().HasIndex(idx.model, idx.name) {
+			t.Fatalf("missing MySQL index %s", idx.name)
+		}
+	}
+	// Exercise reviewed rollback only on fresh diagnostic tables in this schema.
+	if db.Migrator().DropTable(&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}) != nil {
+		t.Fatal("diagnostic test rollback failed")
+	}
+	if db.Where("version = ?", migrationVersion).Delete(&common.Migration{}).Error != nil {
+		t.Fatal("test version rollback failed")
+	}
+	if versionlocal.MigrateAgentDiagnostics(db, migrationVersion) != nil {
+		t.Fatal("diagnostic reapply failed")
+	}
+	seed := func(t *testing.T) (*Service, RegisterResponse, string) {
+		t.Helper()
+		s := NewService(db)
+		s.Now = func() time.Time { return time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC) }
+		token := "isolated-mysql-" + uuid.NewString()
+		s.GenerateToken = func() (string, error) { return token, nil }
+		d, err := s.Register(context.Background(), RegisterRequest{RequestID: uuid.NewString(), InstallID: uuid.NewString(), Name: "test", Manufacturer: "test", Model: "test", AndroidVersion: "14", AgentVersion: "test", PDDVersion: "test"}, "")
+		if err != nil {
+			t.Fatal("synthetic device registration failed")
+		}
+		if db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("last_heartbeat_at", s.Now().Add(-time.Minute)).Error != nil {
+			t.Fatal("test seed failed")
+		}
+		t.Cleanup(func() {
+			if db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("status", models.DeviceStatusDisabled).Error != nil {
+				t.Error("synthetic device cleanup failed")
+			}
+		})
+		return s, d, token
+	}
+
+	t.Run("scan_revalidates_heartbeat_after_lock_wait", func(t *testing.T) {
+		s, d, _ := seed(t)
+		lock := db.Begin()
+		if lock.Error != nil {
+			t.Fatal("begin failed")
+		}
+		defer lock.Rollback()
+		var device models.AgentDevice
+		if lock.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, d.DeviceID).Error != nil {
+			t.Fatal("lock failed")
+		}
+		done := make(chan struct {
+			n   int64
+			err error
+		}, 1)
+		go func() {
+			n, e := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+			done <- struct {
+				n   int64
+				err error
+			}{n, e}
+		}()
+		select {
+		case <-done:
+			t.Fatal("scan did not wait for device lock")
+		case <-time.After(100 * time.Millisecond):
+		}
+		if lock.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("last_heartbeat_at", s.Now()).Error != nil {
+			t.Fatal("refresh failed")
+		}
+		if lock.Commit().Error != nil {
+			t.Fatal("commit failed")
+		}
+		select {
+		case result := <-done:
+			if result.err != nil || result.n != 0 {
+				t.Fatalf("fresh device falsely marked offline: count=%d error=%v", result.n, result.err != nil)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatal("scan stuck")
+		}
+	})
+	t.Run("concurrent_scans_and_resume_are_unique", func(t *testing.T) {
+		s, d, testDeviceToken := seed(t)
+		var wg sync.WaitGroup
+		errs := make(chan error, 2)
+		for i := 0; i < 2; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				_, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+				errs <- err
+			}()
+		}
+		wg.Wait()
+		for i := 0; i < 2; i++ {
+			if err := <-errs; err != nil {
+				var dbErr *driver.MySQLError
+				if errors.As(err, &dbErr) {
+					t.Fatalf("concurrent scan failed mysql_errno=%d", dbErr.Number)
+				}
+				t.Fatalf("concurrent scan failed classification=%T", err)
+			}
+		}
+		req := HeartbeatRequest{RequestID: uuid.NewString()}
+		for i := 0; i < 2; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				_, err := s.Heartbeat(context.Background(), req, testDeviceToken)
+				errs <- err
+			}()
+		}
+		wg.Wait()
+		for i := 0; i < 2; i++ {
+			if <-errs != nil {
+				t.Fatal("concurrent heartbeat failed")
+			}
+		}
+		var events []models.AgentDeviceStatusEvent
+		db.Where("device_id = ?", d.DeviceID).Order("id").Find(&events)
+		if len(events) != 2 || events[0].Reason != "heartbeat_timeout" || events[1].Reason != "heartbeat_resumed" {
+			t.Fatalf("transition count=%d", len(events))
+		}
+		var n int64
+		db.Model(&models.AgentHeartbeatLog{}).Where("device_id = ?", d.DeviceID).Count(&n)
+		if n != 1 {
+			t.Fatalf("replay history count=%d", n)
+		}
+	})
+	t.Run("event_failure_rolls_back_real_mysql", func(t *testing.T) {
+		s, d, _ := seed(t)
+		product := models.PDDProduct{GoodsID: "37801", URL: "https://example.invalid/test"}
+		rule := models.CollectionRule{Name: "test", ContentJSON: `{}`}
+		if db.Create(&product).Error != nil || db.Create(&rule).Error != nil {
+			t.Fatal("synthetic task input failed")
+		}
+		task := models.CollectionTask{PDDProductID: &product.ID, RuleID: rule.ID, DeviceID: &d.DeviceID, Status: models.TaskStatusRunning, URLSnapshot: product.URL, GoodsIDSnapshot: product.GoodsID, RuleSnapshot: rule.ContentJSON}
+		if db.Create(&task).Error != nil {
+			t.Fatal("synthetic task creation failed")
+		}
+		if db.Exec("CREATE TRIGGER diagnostic_test_reject BEFORE INSERT ON agent_device_status_event FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'synthetic_failure'").Error != nil {
+			t.Fatal("test trigger creation failed")
+		}
+		defer db.Exec("DROP TRIGGER IF EXISTS diagnostic_test_reject")
+		n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+		if err == nil || n != 0 {
+			t.Fatal("failed event did not abort")
+		}
+		var device models.AgentDevice
+		db.First(&device, d.DeviceID)
+		if device.Status != models.DeviceStatusOnline {
+			t.Fatal("state escaped rollback")
+		}
+		var storedTask models.CollectionTask
+		if db.First(&storedTask, task.ID).Error != nil || storedTask.Status != models.TaskStatusRunning || storedTask.DeviceRunSlot == nil {
+			t.Fatal("task escaped rollback")
+		}
+		if db.Exec("DROP TRIGGER diagnostic_test_reject").Error != nil {
+			t.Fatal("test trigger removal failed")
+		}
+		n, err = s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+		if err != nil || n != 1 {
+			t.Fatal("retry failed")
+		}
+		var event models.AgentDeviceStatusEvent
+		if db.Where("device_id = ?", d.DeviceID).First(&event).Error != nil || event.FailedTaskCount != 1 || event.OrderResultUnknownCount != 0 {
+			t.Fatal("per-device counts incorrect")
+		}
+	})
+	for _, domain := range []string{"collection", "purchase"} {
+		t.Run("scan_does_not_deadlock_task_then_device_reset_"+domain, func(t *testing.T) {
+			s, d, _ := seed(t)
+			goodsID := "37802"
+			if domain == "purchase" {
+				goodsID = "37803"
+			}
+			product := models.PDDProduct{GoodsID: goodsID, URL: "https://example.invalid/test"}
+			rule := models.CollectionRule{Name: "test", ContentJSON: `{}`}
+			if db.Create(&product).Error != nil || db.Create(&rule).Error != nil {
+				t.Fatal("synthetic inputs failed")
+			}
+			var taskID uint64
+			var taskModel any
+			if domain == "collection" {
+				task := models.CollectionTask{PDDProductID: &product.ID, RuleID: rule.ID, DeviceID: &d.DeviceID, Status: models.TaskStatusRunning, URLSnapshot: product.URL, GoodsIDSnapshot: product.GoodsID, RuleSnapshot: rule.ContentJSON}
+				if db.Create(&task).Error != nil {
+					t.Fatal("synthetic task failed")
+				}
+				taskID = task.ID
+				taskModel = &models.CollectionTask{}
+			} else {
+				task := models.PurchaseTask{PDDProductID: product.ID, DeviceID: &d.DeviceID, ExecutionMode: models.PurchaseExecutionModeRehearsal, Status: models.PurchaseTaskStatusRunning, PDDURLSnapshot: product.URL, PDDGoodsIDSnapshot: product.GoodsID, Quantity: 1, ReferenceUnitPriceCent: 100, MinUnitPriceCent: 20, MaxUnitPriceCent: 150, Currency: "CNY", RuleType: "pddPurchase", RuleSchemaVersion: 1, RequiredCapabilitiesJSON: `[]`, RuleSnapshot: `{}`, CreateRequestID: uuid.NewString()}
+				if db.Create(&task).Error != nil {
+					t.Fatal("synthetic purchase task failed")
+				}
+				taskID = task.ID
+				taskModel = &models.PurchaseTask{}
+			}
+			resetTx := db.Begin()
+			defer resetTx.Rollback()
+			if resetTx.Clauses(clause.Locking{Strength: "UPDATE"}).First(taskModel, taskID).Error != nil {
+				t.Fatal("reset task lock failed")
+			}
+			type scanMarker struct{}
+			deviceLocked, releaseScan := make(chan struct{}), make(chan struct{})
+			callback := "diagnostic_test_scan_lock_barrier"
+			if db.Callback().Query().After("gorm:query").Register(callback, func(tx *gorm.DB) {
+				if tx.Statement.Context.Value(scanMarker{}) == true && tx.Statement.Table == "agent_device" {
+					if _, ok := tx.Statement.Clauses["FOR"]; ok {
+						close(deviceLocked)
+						<-releaseScan
+					}
+				}
+			}) != nil {
+				t.Fatal("register scan barrier failed")
+			}
+			defer db.Callback().Query().Remove(callback)
+			scanDone, resetDone := make(chan error, 1), make(chan error, 1)
+			go func() {
+				_, err := s.MarkStaleDevicesOffline(context.WithValue(context.Background(), scanMarker{}, true), DefaultOfflineThreshold)
+				scanDone <- err
+			}()
+			select {
+			case <-deviceLocked:
+			case <-time.After(5 * time.Second):
+				close(releaseScan)
+				t.Fatal("scan device lock not reached")
+			}
+			go func() {
+				var device models.AgentDevice
+				err := resetTx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, d.DeviceID).Error
+				resetTx.Rollback()
+				resetDone <- err
+			}()
+			close(releaseScan)
+			scanErr, resetErr := <-scanDone, <-resetDone
+			if resetErr != nil {
+				var mysqlErr *driver.MySQLError
+				if errors.As(resetErr, &mysqlErr) {
+					t.Fatalf("reset lock failed mysql_errno=%d", mysqlErr.Number)
+				}
+				t.Fatal("reset lock failed")
+			}
+			var mysqlErr *driver.MySQLError
+			if scanErr == nil || !errors.As(scanErr, &mysqlErr) || mysqlErr.Number != 3572 {
+				if mysqlErr != nil {
+					t.Fatalf("scan must decline busy task without deadlock; mysql_errno=%d", mysqlErr.Number)
+				}
+				t.Fatal("scan must decline busy task")
+			}
+			var eventCount int64
+			db.Model(&models.AgentDeviceStatusEvent{}).Where("device_id = ?", d.DeviceID).Count(&eventCount)
+			if eventCount != 0 {
+				t.Fatal("busy task created event")
+			}
+			var device models.AgentDevice
+			db.First(&device, d.DeviceID)
+			if device.Status != models.DeviceStatusOnline {
+				t.Fatal("busy task changed status")
+			}
+			db.Callback().Query().Remove(callback)
+			n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+			if err != nil || n != 1 {
+				t.Fatal("scan did not retry after reset released locks")
+			}
+		})
+	}
+	t.Run("history_timeout_is_bounded_on_real_lock", func(t *testing.T) {
+		s, _, testDeviceToken := seed(t)
+		// Hold the diagnostic table's insert gap; liveness updates remain free.
+		lock := db.Begin()
+		defer lock.Rollback()
+		if lock.Exec("SELECT id FROM agent_heartbeat_log WHERE id > 0 FOR UPDATE").Error != nil {
+			t.Fatal("history lock failed")
+		}
+		started := time.Now()
+		res, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken)
+		t.Logf("heartbeat with locked history table elapsed=%s", time.Since(started))
+		if err != nil || !res.Online || time.Since(started) > time.Second {
+			t.Fatal("history lock changed heartbeat outcome or exceeded budget")
+		}
+	})
+	t.Run("strict_cleanup", func(t *testing.T) {
+		s, d, _ := seed(t)
+		boundary := models.AgentHeartbeatLog{DeviceID: d.DeviceID, ReceivedAt: s.Now().Add(-7 * 24 * time.Hour), RequestID: uuid.NewString(), AgentVersion: "test", ClientReportStatus: "none"}
+		expired := boundary
+		expired.ReceivedAt = expired.ReceivedAt.Add(-time.Second)
+		db.Create(&boundary)
+		db.Create(&expired)
+		n, err := s.CleanupHeartbeatHistory(context.Background())
+		if err != nil || n != 1 {
+			t.Fatalf("cleanup=%d failed=%v", n, err != nil)
+		}
+		history := make([]models.AgentHeartbeatLog, 5001)
+		events := make([]models.AgentDeviceStatusEvent, 5001)
+		for i := range history {
+			history[i] = models.AgentHeartbeatLog{DeviceID: d.DeviceID, ReceivedAt: s.Now().Add(-31 * 24 * time.Hour), ClientReportStatus: "none"}
+			events[i] = models.AgentDeviceStatusEvent{DeviceID: d.DeviceID, OccurredAt: s.Now().Add(-31 * 24 * time.Hour)}
+		}
+		if db.CreateInBatches(&history, 500).Error != nil || db.CreateInBatches(&events, 500).Error != nil {
+			t.Fatal("synthetic cleanup batch seed failed")
+		}
+		n, err = s.CleanupHeartbeatHistory(context.Background())
+		if err != nil || n != 5000 {
+			t.Fatalf("history cleanup batch=%d failed=%v", n, err != nil)
+		}
+		n, err = s.CleanupDeviceStatusEvents(context.Background())
+		if err != nil || n != 5000 {
+			t.Fatalf("event cleanup batch=%d failed=%v", n, err != nil)
+		}
+		var remaining int64
+		db.Model(&models.AgentHeartbeatLog{}).Where("device_id = ?", d.DeviceID).Count(&remaining)
+		if remaining != 2 {
+			t.Fatalf("history boundary/remaining=%d", remaining)
+		}
+	})
+}

server/app/goauto/device/diagnostics_test.go

@@ -0,0 +1,434 @@
+package device
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"go-admin/app/goauto/models"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+)
+
+func TestHeartbeatDiagnosticsHistoryAndReportIsolation(t *testing.T) {
+	db := openTestDatabase(t)
+	service := newTestService(t, db)
+	_, registered := registerHeartbeatDevice(t, service)
+	gin.SetMode(gin.TestMode)
+	router := gin.New()
+	router.POST("/heartbeat", Handler{DB: db}.Heartbeat)
+	for _, tc := range []struct{ name, report, status string }{
+		{"missing", "", "none"},
+		{"accepted", `,"clientReport":{"snapshotSeq":1,"failedRounds":0,"latestFailureAt":null,"latestFailureStep":null,"latestFailureCategory":null,"latestFailureDurationMs":null,"previousSyncDurationMs":25,"hasInterruptedRound":false}`, "accepted"},
+		{"wrong_type", `,"clientReport":"secret-not-stored"`, "client_report_invalid"},
+		{"unknown_field", `,"clientReport":{"url":"secret-not-stored"}`, "client_report_invalid"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			id := uuid.NewString()
+			body := fmt.Sprintf(`{"requestId":%q,"currentTaskId":null%s}`, id, tc.report)
+			for i := 0; i < 2; i++ {
+				req := httptest.NewRequest(http.MethodPost, "/heartbeat", strings.NewReader(body))
+				req.Header.Set("Authorization", "Bearer "+testDeviceToken)
+				res := httptest.NewRecorder()
+				router.ServeHTTP(res, req)
+				if res.Code != 200 {
+					t.Fatalf("heartbeat status=%d body=%s", res.Code, res.Body.String())
+				}
+				var envelope struct {
+					Data map[string]any `json:"data"`
+				}
+				_ = json.Unmarshal(res.Body.Bytes(), &envelope)
+				if envelope.Data["acceptsClientReport"] != true {
+					t.Fatal("capability missing")
+				}
+			}
+			var rows []struct {
+				DeviceID           uint64
+				ClientReportStatus string
+				ClientReportJSON   *string
+			}
+			if err := db.Table("agent_heartbeat_log").Where("request_id = ?", id).Find(&rows).Error; err != nil {
+				t.Fatal(err)
+			}
+			if len(rows) != 1 || rows[0].DeviceID != registered.DeviceID || rows[0].ClientReportStatus != tc.status || (rows[0].ClientReportJSON != nil) != (tc.status == "accepted") {
+				t.Fatalf("history=%+v", rows)
+			}
+		})
+	}
+}
+
+func TestDisableWritesOneAtomicStatusEvent(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	_, d := registerHeartbeatDevice(t, s)
+	for i := 0; i < 2; i++ {
+		if err := s.Disable(context.Background(), d.DeviceID); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var events []models.AgentDeviceStatusEvent
+	if err := db.Find(&events).Error; err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 1 || events[0].Reason != "disabled" || events[0].FromStatus != "online" || events[0].ToStatus != "disabled" {
+		t.Fatalf("events=%+v", events)
+	}
+}
+
+func TestStatusEventFailureRollsBackAndCanRetry(t *testing.T) {
+	for _, action := range []string{"timeout", "resumed", "disabled"} {
+		t.Run(action, func(t *testing.T) {
+			db := openTestDatabase(t)
+			s := newTestService(t, db)
+			_, d := registerHeartbeatDevice(t, s)
+			status := models.DeviceStatusOnline
+			if action == "resumed" {
+				status = models.DeviceStatusOffline
+			}
+			old := s.Now().Add(-time.Minute)
+			if err := db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Updates(map[string]any{"status": status, "last_heartbeat_at": old}).Error; err != nil {
+				t.Fatal(err)
+			}
+			var task models.CollectionTask
+			if action == "timeout" || action == "disabled" {
+				product := models.PDDProduct{GoodsID: "1001", URL: "https://example.invalid/test"}
+				rule := models.CollectionRule{Name: "test", ContentJSON: `{}`}
+				if err := db.Create(&product).Error; err != nil {
+					t.Fatal(err)
+				}
+				if err := db.Create(&rule).Error; err != nil {
+					t.Fatal(err)
+				}
+				task = newHeartbeatTask(product, rule, d.DeviceID)
+				if err := db.Create(&task).Error; err != nil {
+					t.Fatal(err)
+				}
+			}
+			operation := func() error {
+				switch action {
+				case "timeout":
+					_, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+					return err
+				case "disabled":
+					return s.Disable(context.Background(), d.DeviceID)
+				default:
+					_, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken)
+					return err
+				}
+			}
+			callback := "diagnostic_event_failure"
+			if err := db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) {
+				if tx.Statement.Table == "agent_device_status_event" {
+					tx.AddError(errors.New("injected private error"))
+				}
+			}); err != nil {
+				t.Fatal(err)
+			}
+			if err := operation(); err == nil {
+				t.Fatal("event failure must abort transaction")
+			}
+			var stored models.AgentDevice
+			if err := db.First(&stored, d.DeviceID).Error; err != nil {
+				t.Fatal(err)
+			}
+			if stored.Status != status || !stored.LastHeartbeatAt.Equal(old) {
+				t.Fatalf("state escaped rollback=%+v", stored.Status)
+			}
+			if task.ID != 0 {
+				var storedTask models.CollectionTask
+				if err := db.First(&storedTask, task.ID).Error; err != nil {
+					t.Fatal(err)
+				}
+				if storedTask.Status != models.TaskStatusRunning || storedTask.DeviceRunSlot == nil {
+					t.Fatal("task changes escaped rollback")
+				}
+			}
+			_ = db.Callback().Create().Remove(callback)
+			if err := operation(); err != nil {
+				t.Fatal(err)
+			}
+			var n int64
+			db.Model(&models.AgentDeviceStatusEvent{}).Count(&n)
+			if n != 1 {
+				t.Fatalf("events=%d", n)
+			}
+		})
+	}
+}
+
+func TestOfflineEventCountsPerDeviceIncludingUnknownBoundary(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	ids := make([]uint64, 3)
+	for i := range ids {
+		s.GenerateToken = func() (string, error) { return fmt.Sprintf("test-token-%d", i), nil }
+		_, registered := registerHeartbeatDevice(t, s)
+		ids[i] = registered.DeviceID
+		if err := db.Model(&models.AgentDevice{}).Where("id = ?", ids[i]).Update("last_heartbeat_at", s.Now().Add(-time.Minute)).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	product := models.PDDProduct{GoodsID: "1002", URL: "https://example.invalid/test"}
+	rule := models.CollectionRule{Name: "test", ContentJSON: `{}`}
+	if err := db.Create(&product).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&rule).Error; err != nil {
+		t.Fatal(err)
+	}
+	collection := newHeartbeatTask(product, rule, ids[0])
+	if err := db.Create(&collection).Error; err != nil {
+		t.Fatal(err)
+	}
+	for i, status := range []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted} {
+		task := newHeartbeatPurchaseTask(product, ids[i], status)
+		if err := db.Create(&task).Error; err != nil {
+			t.Fatal(err)
+		}
+		attempt := newRunningPurchaseAttempt(task.ID, ids[i])
+		if err := db.Create(&attempt).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	if n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold); err != nil || n != 3 {
+		t.Fatalf("scan=%d %v", n, err)
+	}
+	var events []models.AgentDeviceStatusEvent
+	if err := db.Order("device_id").Find(&events).Error; err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 3 || events[0].FailedTaskCount != 2 || events[0].OrderResultUnknownCount != 0 || events[1].FailedTaskCount != 0 || events[1].OrderResultUnknownCount != 1 || events[2].FailedTaskCount != 0 || events[2].OrderResultUnknownCount != 0 {
+		t.Fatalf("events=%+v", events)
+	}
+}
+
+func TestHeartbeatHistoryOnlyForSuccessfulMainRequest(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	registerHeartbeatDevice(t, s)
+	if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, "wrong-token"); err == nil {
+		t.Fatal("bad token accepted")
+	}
+	if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: "invalid"}, testDeviceToken); err == nil {
+		t.Fatal("bad request accepted")
+	}
+	id := uint64(999)
+	if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString(), CurrentTaskID: &id}, testDeviceToken); err == nil {
+		t.Fatal("task mismatch accepted")
+	}
+	var count int64
+	if err := db.Model(&models.AgentHeartbeatLog{}).Count(&count).Error; err != nil {
+		t.Fatal(err)
+	}
+	if count != 0 {
+		t.Fatal("failed request stored history")
+	}
+}
+
+func TestHeartbeatHistoryBoundedIndependentAndBestEffort(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	_, d := registerHeartbeatDevice(t, s)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	callback := "history_wait_for_deadline"
+	if err := db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) {
+		if tx.Statement.Table != "agent_heartbeat_log" {
+			return
+		}
+		// We are already outside the core transaction; cancellation must not
+		// cancel the independent history context.
+		cancel()
+		if tx.Statement.Context.Err() != nil {
+			t.Error("history inherited request cancellation")
+		}
+		deadline, ok := tx.Statement.Context.Deadline()
+		if !ok || time.Until(deadline) > 201*time.Millisecond {
+			t.Error("missing bounded deadline")
+		}
+		var current models.AgentDevice
+		if err := db.First(&current, d.DeviceID).Error; err != nil || current.LastHeartbeatAt == nil {
+			t.Error("core was not committed before history")
+		}
+		<-tx.Statement.Context.Done()
+		tx.AddError(tx.Statement.Context.Err())
+	}); err != nil {
+		t.Fatal(err)
+	}
+	started := time.Now()
+	res, err := s.Heartbeat(ctx, HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken)
+	if err != nil || !res.Online || time.Since(started) > time.Second {
+		t.Fatalf("best effort changed result: %v", err)
+	}
+	_ = db.Callback().Create().Remove(callback)
+	if err := db.Migrator().DropTable(&models.AgentHeartbeatLog{}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestClientReportWhitelistBounds(t *testing.T) {
+	valid := `{"snapshotSeq":1,"failedRounds":1,"latestFailureAt":"2026-10-10T00:00:00Z","latestFailureStep":"heartbeat","latestFailureCategory":"network","latestFailureDurationMs":12,"previousSyncDurationMs":null,"hasInterruptedRound":false}`
+	for _, tc := range []struct{ name, raw, status string }{
+		{"accepted", valid, "accepted"}, {"none", "", "none"}, {"null", "null", "client_report_invalid"}, {"array", "[]", "client_report_invalid"},
+		{"sequence", strings.Replace(valid, `"snapshotSeq":1`, `"snapshotSeq":0`, 1), "client_report_invalid"},
+		{"duration", strings.Replace(valid, `"latestFailureDurationMs":12`, `"latestFailureDurationMs":86400001`, 1), "client_report_invalid"},
+		{"unknown_step", strings.Replace(valid, `"heartbeat"`, `"http://private"`, 1), "client_report_invalid"},
+		{"inconsistent", strings.Replace(valid, `"failedRounds":1`, `"failedRounds":0`, 1), "client_report_invalid"},
+		{"unknown", strings.Replace(valid, `"snapshotSeq"`, `"privateText"`, 1), "client_report_invalid"},
+		{"null_bool", strings.Replace(valid, `"hasInterruptedRound":false`, `"hasInterruptedRound":null`, 1), "client_report_invalid"},
+		{"case_alias_cannot_replace_missing_required", `{"snapshotSeq":1,"failedRounds":0,"latestFailureAt":null,"latestFailureStep":null,"latestFailureCategory":null,"SNAPSHOTSEQ":2,"previousSyncDurationMs":null,"hasInterruptedRound":false}`, "client_report_invalid"},
+		{"oversize", strings.Repeat(" ", 2049) + valid, "client_report_invalid"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			value, status := validateHeartbeatClientReport(json.RawMessage(tc.raw))
+			if status != tc.status || (status != "accepted" && value != nil) {
+				t.Fatalf("status=%s valuePresent=%v", status, value != nil)
+			}
+		})
+	}
+}
+
+func TestDiagnosticsCleanupIsBoundedAndStrictlyExpired(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	now := s.Now()
+	old := now.Add(-31 * 24 * time.Hour)
+	heartbeats := make([]models.AgentHeartbeatLog, 5001)
+	events := make([]models.AgentDeviceStatusEvent, 5001)
+	for i := range heartbeats {
+		heartbeats[i] = models.AgentHeartbeatLog{DeviceID: 1, ReceivedAt: old, RequestID: uuid.NewString(), AgentVersion: "test", ClientReportStatus: "none"}
+		events[i] = models.AgentDeviceStatusEvent{DeviceID: 1, OccurredAt: old, FromStatus: "online", ToStatus: "offline", Reason: "heartbeat_timeout"}
+	}
+	if err := db.CreateInBatches(&heartbeats, 100).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.CreateInBatches(&events, 100).Error; err != nil {
+		t.Fatal(err)
+	}
+	boundaryHeartbeat := models.AgentHeartbeatLog{DeviceID: 1, ReceivedAt: now.Add(-7 * 24 * time.Hour), RequestID: uuid.NewString(), AgentVersion: "test", ClientReportStatus: "none"}
+	boundaryEvent := models.AgentDeviceStatusEvent{DeviceID: 1, OccurredAt: now.Add(-30 * 24 * time.Hour), FromStatus: "online", ToStatus: "offline", Reason: "heartbeat_timeout"}
+	db.Create(&boundaryHeartbeat)
+	db.Create(&boundaryEvent)
+	n, err := s.CleanupHeartbeatHistory(context.Background())
+	if err != nil || n != 5000 {
+		t.Fatalf("heartbeat cleanup=%d %v", n, err)
+	}
+	n, err = s.CleanupDeviceStatusEvents(context.Background())
+	if err != nil || n != 5000 {
+		t.Fatalf("event cleanup=%d %v", n, err)
+	}
+	var count int64
+	db.Model(&models.AgentHeartbeatLog{}).Count(&count)
+	if count != 2 {
+		t.Fatalf("remaining=%d", count)
+	}
+	db.Model(&models.AgentDeviceStatusEvent{}).Count(&count)
+	if count != 2 {
+		t.Fatalf("remaining=%d", count)
+	}
+	n, err = s.CleanupHeartbeatHistory(context.Background())
+	if err != nil || n != 1 {
+		t.Fatalf("strict heartbeat cleanup=%d %v", n, err)
+	}
+	n, err = s.CleanupDeviceStatusEvents(context.Background())
+	if err != nil || n != 1 {
+		t.Fatalf("strict event cleanup=%d %v", n, err)
+	}
+}
+
+func TestDiagnosticsCleanupRunsOnSeparateSchedulesAndStops(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	old := s.Now().Add(-31 * 24 * time.Hour)
+	history := models.AgentHeartbeatLog{ReceivedAt: old, ClientReportStatus: "none"}
+	event := models.AgentDeviceStatusEvent{OccurredAt: old}
+	if err := db.Create(&history).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&event).Error; err != nil {
+		t.Fatal(err)
+	}
+	heartbeats, events := make(chan time.Time), make(chan time.Time)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		runDiagnosticsCleanup(ctx, s, heartbeats, events, func(err error) { t.Error(err) })
+	}()
+	heartbeats <- s.Now()
+	// A second tick synchronizes completion of the first cleanup.
+	heartbeats <- s.Now()
+	var n int64
+	db.Model(&models.AgentHeartbeatLog{}).Count(&n)
+	if n != 0 {
+		t.Fatal("hourly history cleanup did not run")
+	}
+	db.Model(&models.AgentDeviceStatusEvent{}).Count(&n)
+	if n != 1 {
+		t.Fatal("hourly tick cleaned daily events")
+	}
+	events <- s.Now()
+	events <- s.Now()
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("cleanup did not stop")
+	}
+	db.Model(&models.AgentDeviceStatusEvent{}).Count(&n)
+	if n != 0 {
+		t.Fatal("daily event cleanup did not run")
+	}
+}
+
+func TestOfflineDiagnosticsAtomicAndResumed(t *testing.T) {
+	db := openTestDatabase(t)
+	s := newTestService(t, db)
+	_, d := registerHeartbeatDevice(t, s)
+	old := s.Now().Add(-time.Minute)
+	if err := db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("last_heartbeat_at", old).Error; err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 2; i++ {
+		n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold)
+		if err != nil || n != int64(1-i) {
+			t.Fatalf("scan=%d %v", n, err)
+		}
+	}
+	var events []struct {
+		Reason, FromStatus, ToStatus             string
+		LastHeartbeatAt                          *time.Time
+		FailedTaskCount, OrderResultUnknownCount int64
+	}
+	if err := db.Table("agent_device_status_event").Find(&events).Error; err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 1 || events[0].Reason != "heartbeat_timeout" || events[0].FromStatus != "online" || events[0].ToStatus != "offline" || events[0].LastHeartbeatAt == nil || !events[0].LastHeartbeatAt.Equal(old) {
+		t.Fatalf("events=%+v", events)
+	}
+	req := HeartbeatRequest{RequestID: uuid.NewString()}
+	for i := 0; i < 2; i++ {
+		if _, err := s.Heartbeat(context.Background(), req, testDeviceToken); err != nil {
+			t.Fatal(err)
+		}
+	}
+	events = nil
+	if err := db.Table("agent_device_status_event").Order("id").Find(&events).Error; err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 2 || events[1].Reason != "heartbeat_resumed" {
+		t.Fatalf("events=%+v", events)
+	}
+}

server/app/goauto/device/heartbeat.go

@@ -2,6 +2,7 @@ package device
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"time"
@@ -10,6 +11,7 @@ import (
 
 	"github.com/google/uuid"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 const (
@@ -18,9 +20,10 @@ const (
 )
 
 type HeartbeatRequest struct {
-	RequestID     string   `json:"requestId"`
-	CurrentTaskID *uint64  `json:"currentTaskId"`
-	Capabilities  []string `json:"capabilities,omitempty"`
+	RequestID     string          `json:"requestId"`
+	CurrentTaskID *uint64         `json:"currentTaskId"`
+	Capabilities  []string        `json:"capabilities,omitempty"`
+	ClientReport  json.RawMessage `json:"clientReport,omitempty"`
 }
 
 type HeartbeatResponse struct {
@@ -32,6 +35,7 @@ type HeartbeatResponse struct {
 	ServerTime               string  `json:"serverTime"`
 	HeartbeatIntervalSeconds int     `json:"heartbeatIntervalSeconds"`
 	Replayed                 bool    `json:"replayed,omitempty"`
+	AcceptsClientReport      bool    `json:"acceptsClientReport"`
 }
 
 func (service *Service) Heartbeat(ctx context.Context, request HeartbeatRequest, presentedToken string) (HeartbeatResponse, error) {
@@ -50,10 +54,12 @@ func (service *Service) Heartbeat(ctx context.Context, request HeartbeatRequest,
 		interval = DefaultHeartbeatIntervalSeconds
 	}
 	var response HeartbeatResponse
+	var history models.AgentHeartbeatLog
+	var resumed *models.AgentDeviceStatusEvent
 	err = service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		var device models.AgentDevice
 		digest := tokenDigest(presentedToken)
-		if err := tx.Where("token_digest = ? AND token_revoked_at IS NULL", digest).First(&device).Error; err != nil {
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("token_digest = ? AND token_revoked_at IS NULL", digest).First(&device).Error; err != nil {
 			if errors.Is(err, gorm.ErrRecordNotFound) {
 				return tokenInvalidError()
 			}
@@ -93,14 +99,30 @@ func (service *Service) Heartbeat(ctx context.Context, request HeartbeatRequest,
 			if result.RowsAffected == 0 {
 				return tokenInvalidError()
 			}
+			if device.Status == models.DeviceStatusOffline {
+				resumed = &models.AgentDeviceStatusEvent{DeviceID: device.ID, OccurredAt: now, FromStatus: device.Status, ToStatus: models.DeviceStatusOnline, Reason: "heartbeat_resumed", LastHeartbeatAt: device.LastHeartbeatAt}
+				if err := insertDeviceStatusEvent(tx, resumed); err != nil {
+					return err
+				}
+			}
 		}
 		response = HeartbeatResponse{
 			RequestID: request.RequestID, DeviceID: device.ID, CurrentTaskID: runningTaskID,
 			Online: true, Busy: runningTaskID != nil, ServerTime: now.UTC().Format(time.RFC3339Nano),
-			HeartbeatIntervalSeconds: interval, Replayed: replayed,
+			HeartbeatIntervalSeconds: interval, Replayed: replayed, AcceptsClientReport: true,
 		}
+		history = models.AgentHeartbeatLog{DeviceID: device.ID, ReceivedAt: now, RequestID: request.RequestID, CurrentTaskID: request.CurrentTaskID, RunningTaskID: runningTaskID, Busy: runningTaskID != nil, AgentVersion: device.AgentVersion}
 		return nil
 	})
+	if err == nil {
+		if resumed != nil {
+			logDeviceStatusEvent(*resumed)
+		}
+		if !response.Replayed {
+			history.ClientReportJSON, history.ClientReportStatus = validateHeartbeatClientReport(request.ClientReport)
+			service.saveHeartbeatHistory(ctx, history)
+		}
+	}
 	return response, err
 }
 
@@ -124,50 +146,107 @@ func (service *Service) MarkStaleDevicesOffline(ctx context.Context, threshold t
 	now := service.Now()
 	cutoff := now.Add(-threshold)
 	var changed int64
+	var events []models.AgentDeviceStatusEvent
+	var deviceIDs []uint64
+	if err := service.DB.WithContext(ctx).Model(&models.AgentDevice{}).
+		Where("status = ? AND COALESCE(last_heartbeat_at, created_at) < ?", models.DeviceStatusOnline, cutoff).
+		Order("id").Pluck("id", &deviceIDs).Error; err != nil {
+		return 0, internalError(err)
+	}
+	if len(deviceIDs) == 0 {
+		return 0, nil
+	}
 	err := service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-		var deviceIDs []uint64
-		if err := tx.Model(&models.AgentDevice{}).
-			Where("status = ? AND COALESCE(last_heartbeat_at, created_at) < ?", models.DeviceStatusOnline, cutoff).
-			Pluck("id", &deviceIDs).Error; err != nil {
-			return internalError(err)
+		// Lock primary keys in deterministic order, not the mutable status index:
+		// a status-range FOR UPDATE can deadlock concurrent scans on gap locks.
+		// Candidate selection is only a hint; revalidate the current locked row.
+		for _, deviceID := range deviceIDs {
+			var device models.AgentDevice
+			if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, deviceID).Error; err != nil {
+				if errors.Is(err, gorm.ErrRecordNotFound) {
+					continue
+				}
+				return internalError(err)
+			}
+			last := device.CreatedAt
+			if device.LastHeartbeatAt != nil {
+				last = *device.LastHeartbeatAt
+			}
+			if device.Status != models.DeviceStatusOnline || !last.Before(cutoff) {
+				continue
+			}
+			// Existing Admin reset paths lock task -> device, while Agent paths
+			// lock device -> task. Never wait for a task while holding this device:
+			// contention aborts this atomic scan and the next scan retries it.
+			if err := lockOfflineTasksNowait(tx, device.ID); err != nil {
+				return err
+			}
+			result := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).
+				Where("device_id = ? AND status = ?", device.ID, models.TaskStatusRunning).
+				Updates(map[string]any{
+					"status": models.TaskStatusFailed, "active_slot": gorm.Expr("NULL"), "device_run_slot": gorm.Expr("NULL"),
+					"lease_expires_at": gorm.Expr("NULL"), "error_code": "DEVICE_OFFLINE",
+					"error_message": "设备心跳超时，任务失败且不会自动重试", "finished_at": now,
+				})
+			if result.Error != nil {
+				return internalError(result.Error)
+			}
+			failed, unknown, err := markOfflinePurchaseTasks(tx, []uint64{device.ID}, now)
+			if err != nil {
+				return err
+			}
+			event := models.AgentDeviceStatusEvent{DeviceID: device.ID, OccurredAt: now, FromStatus: device.Status, ToStatus: models.DeviceStatusOffline, Reason: "heartbeat_timeout", LastHeartbeatAt: device.LastHeartbeatAt, FailedTaskCount: result.RowsAffected + failed, OrderResultUnknownCount: unknown}
+			result = tx.Model(&models.AgentDevice{}).Where("id = ? AND status = ?", device.ID, models.DeviceStatusOnline).
+				Update("status", models.DeviceStatusOffline)
+			if result.Error != nil {
+				return internalError(result.Error)
+			}
+			if err := insertDeviceStatusEvent(tx, &event); err != nil {
+				return err
+			}
+			events = append(events, event)
+			changed += result.RowsAffected
 		}
-		if len(deviceIDs) == 0 {
-			return nil
-		}
-		if err := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).
-			Where("device_id IN ? AND status = ?", deviceIDs, models.TaskStatusRunning).
-			Updates(map[string]any{
-				"status": models.TaskStatusFailed, "active_slot": gorm.Expr("NULL"), "device_run_slot": gorm.Expr("NULL"),
-				"lease_expires_at": gorm.Expr("NULL"), "error_code": "DEVICE_OFFLINE",
-				"error_message": "设备心跳超时，任务失败且不会自动重试", "finished_at": now,
-			}).Error; err != nil {
-			return internalError(err)
-		}
-		if err := markOfflinePurchaseTasks(tx, deviceIDs, now); err != nil {
-			return err
-		}
-		result := tx.Model(&models.AgentDevice{}).Where("id IN ? AND status = ?", deviceIDs, models.DeviceStatusOnline).
-			Update("status", models.DeviceStatusOffline)
-		if result.Error != nil {
-			return internalError(result.Error)
-		}
-		changed = result.RowsAffected
 		return nil
 	})
+	if err != nil {
+		return 0, err
+	}
+	for _, event := range events {
+		logDeviceStatusEvent(event)
+	}
 	return changed, err
 }
 
-func markOfflinePurchaseTasks(tx *gorm.DB, deviceIDs []uint64, now time.Time) error {
-	if err := markOfflinePurchaseStatus(
+func lockOfflineTasksNowait(tx *gorm.DB, deviceID uint64) error {
+	var collectionIDs, purchaseIDs []uint64
+	locking := clause.Locking{Strength: "UPDATE", Options: "NOWAIT"}
+	if err := tx.Model(&models.CollectionTask{}).Clauses(locking).
+		Where("device_id = ? AND status = ?", deviceID, models.TaskStatusRunning).
+		Order("id").Pluck("id", &collectionIDs).Error; err != nil {
+		return internalError(err)
+	}
+	if err := tx.Model(&models.PurchaseTask{}).Clauses(locking).
+		Where("device_id = ? AND status IN ?", deviceID, []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted}).
+		Order("id").Pluck("id", &purchaseIDs).Error; err != nil {
+		return internalError(err)
+	}
+	return nil
+}
+
+func markOfflinePurchaseTasks(tx *gorm.DB, deviceIDs []uint64, now time.Time) (int64, int64, error) {
+	failed, err := markOfflinePurchaseStatus(
 		tx, deviceIDs, models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusFailed,
 		"failed", "设备心跳超时，采购任务失败且不会自动重试", now,
-	); err != nil {
-		return err
+	)
+	if err != nil {
+		return 0, 0, err
 	}
-	return markOfflinePurchaseStatus(
+	unknown, err := markOfflinePurchaseStatus(
 		tx, deviceIDs, models.PurchaseTaskStatusOrderSubmitStarted, models.PurchaseTaskStatusOrderResultUnknown,
 		"order_result_unknown", "设备心跳超时，订单结果未知，请人工核对，禁止自动重试", now,
 	)
+	return failed, unknown, err
 }
 
 func markOfflinePurchaseStatus(
@@ -178,15 +257,15 @@ func markOfflinePurchaseStatus(
 	resultType string,
 	errorMessage string,
 	now time.Time,
-) error {
+) (int64, error) {
 	var taskIDs []uint64
 	if err := tx.Model(&models.PurchaseTask{}).
 		Where("device_id IN ? AND status = ?", deviceIDs, fromStatus).
 		Pluck("id", &taskIDs).Error; err != nil {
-		return internalError(err)
+		return 0, internalError(err)
 	}
 	if len(taskIDs) == 0 {
-		return nil
+		return 0, nil
 	}
 	errorCode := "DEVICE_OFFLINE"
 	if err := tx.Model(&models.PurchaseTaskAttempt{}).
@@ -195,7 +274,7 @@ func markOfflinePurchaseStatus(
 			"status": models.PurchaseAttemptStatusFailed, "result_type": resultType,
 			"error_code": errorCode, "error_message": errorMessage, "finished_at": now,
 		}).Error; err != nil {
-		return internalError(err)
+		return 0, internalError(err)
 	}
 	updates := map[string]any{
 		"status": toStatus, "device_run_slot": gorm.Expr("NULL"), "account_run_slot": gorm.Expr("NULL"),
@@ -205,12 +284,13 @@ func markOfflinePurchaseStatus(
 	if toStatus == models.PurchaseTaskStatusFailed {
 		updates["active_slot"] = gorm.Expr("NULL")
 	}
-	if err := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).
+	result := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}).
 		Where("id IN ? AND status = ?", taskIDs, fromStatus).
-		Updates(updates).Error; err != nil {
-		return internalError(err)
+		Updates(updates)
+	if result.Error != nil {
+		return 0, internalError(result.Error)
 	}
-	return nil
+	return result.RowsAffected, nil
 }
 
 func RunOfflineMonitor(ctx context.Context, service *Service, scanInterval, threshold time.Duration, onError func(error)) {

server/app/goauto/device/service.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/google/uuid"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 const DefaultHeartbeatIntervalSeconds = 15
@@ -305,7 +306,15 @@ func (service *Service) RevokeToken(ctx context.Context, deviceID uint64) error
 
 func (service *Service) deactivate(ctx context.Context, deviceID uint64, revoke bool) error {
 	now := service.Now()
-	return service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+	var event *models.AgentDeviceStatusEvent
+	err := service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var device models.AgentDevice
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, deviceID).Error; err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				return &ServiceError{Code: CodeDeviceNotFound, Message: "设备不存在", Retryable: false}
+			}
+			return internalError(err)
+		}
 		updates := map[string]any{"status": models.DeviceStatusDisabled}
 		if revoke {
 			updates["token_revoked_at"] = now
@@ -324,17 +333,26 @@ func (service *Service) deactivate(ctx context.Context, deviceID uint64, revoke
 			}
 		}
 		errorCode, errorMessage := CodeDeviceDisabled, "设备已由管理员停用"
-		if err := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).
+		result = tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}).
 			Where("device_id = ? AND status = ?", deviceID, models.TaskStatusRunning).
 			Updates(map[string]any{
 				"status": models.TaskStatusFailed, "active_slot": gorm.Expr("NULL"),
 				"device_run_slot": gorm.Expr("NULL"), "lease_expires_at": gorm.Expr("NULL"),
 				"error_code": errorCode, "error_message": errorMessage, "finished_at": now,
-			}).Error; err != nil {
-			return internalError(err)
+			})
+		if result.Error != nil {
+			return internalError(result.Error)
+		}
+		if device.Status != models.DeviceStatusDisabled {
+			event = &models.AgentDeviceStatusEvent{DeviceID: device.ID, OccurredAt: now, FromStatus: device.Status, ToStatus: models.DeviceStatusDisabled, Reason: "disabled", LastHeartbeatAt: device.LastHeartbeatAt, FailedTaskCount: result.RowsAffected}
+			return insertDeviceStatusEvent(tx, event)
 		}
 		return nil
 	})
+	if err == nil && event != nil {
+		logDeviceStatusEvent(*event)
+	}
+	return err
 }
 
 func normalizeRegisterRequest(request RegisterRequest) RegisterRequest {

server/app/goauto/migrations/migrate.go

@@ -30,6 +30,8 @@ LIMIT 1`
 func MigratedModels() []any {
 	return []any{
 		&models.AgentDevice{},
+		&models.AgentHeartbeatLog{},
+		&models.AgentDeviceStatusEvent{},
 		&models.AgentAppRelease{},
 		&models.AgentAppReleaseSetting{},
 		&models.PDDProduct{},

server/app/goauto/models/agent_diagnostics.go

@@ -0,0 +1,33 @@
+package models
+
+import "time"
+
+// AgentHeartbeatLog is diagnostic history, never a device liveness authority.
+type AgentHeartbeatLog struct {
+	ID                 uint64    `gorm:"primaryKey;autoIncrement"`
+	DeviceID           uint64    `gorm:"not null;index:ix_heartbeat_device_received,priority:1"`
+	ReceivedAt         time.Time `gorm:"not null;index:ix_heartbeat_device_received,priority:2;index:ix_heartbeat_received"`
+	RequestID          string    `gorm:"size:36;not null"`
+	CurrentTaskID      *uint64
+	RunningTaskID      *uint64
+	Busy               bool    `gorm:"not null"`
+	AgentVersion       string  `gorm:"size:32;not null"`
+	ClientReportJSON   *string `gorm:"type:text"`
+	ClientReportStatus string  `gorm:"size:32;not null;check:ck_heartbeat_report_status,client_report_status IN ('none','accepted','client_report_invalid')"`
+}
+
+func (AgentHeartbeatLog) TableName() string { return "agent_heartbeat_log" }
+
+type AgentDeviceStatusEvent struct {
+	ID                      uint64    `gorm:"primaryKey;autoIncrement"`
+	DeviceID                uint64    `gorm:"not null"`
+	OccurredAt              time.Time `gorm:"not null;index:ix_device_event_occurred"`
+	FromStatus              string    `gorm:"size:16;not null"`
+	ToStatus                string    `gorm:"size:16;not null"`
+	Reason                  string    `gorm:"size:32;not null"`
+	LastHeartbeatAt         *time.Time
+	FailedTaskCount         int64 `gorm:"not null"`
+	OrderResultUnknownCount int64 `gorm:"not null"`
+}
+
+func (AgentDeviceStatusEvent) TableName() string { return "agent_device_status_event" }

server/cmd/api/server.go

@@ -129,6 +129,9 @@ func run() error {
 	defer stopOfflineMonitors()
 	for _, db := range sdk.Runtime.GetDb() {
 		service := goautodevice.NewService(db)
+		go goautodevice.RunDiagnosticsCleanup(offlineMonitorContext, service,
+			func(err error) { log.Error("device diagnostics cleanup failed") },
+		)
 		go goautopurchase.RunFailureSnapshotCleanup(
 			offlineMonitorContext, goautopurchase.NewService(db), time.Hour,
 			func(err error) { log.Error("purchase failure snapshot cleanup failed") },

server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics.go

@@ -0,0 +1,29 @@
+package version_local
+
+import (
+	"runtime"
+
+	"go-admin/app/goauto/models"
+	"go-admin/cmd/migrate/migration"
+	common "go-admin/common/models"
+
+	"gorm.io/gorm"
+)
+
+func init() {
+	_, file, _, _ := runtime.Caller(0)
+	migration.Migrate.SetVersion(migration.GetFilename(file), MigrateAgentDiagnostics)
+}
+
+// MigrateAgentDiagnostics only adds #378 diagnostic tables. MySQL DDL commits
+// implicitly: idempotent table creation allows recovery after a partial up.
+func MigrateAgentDiagnostics(db *gorm.DB, version string) error {
+	for _, model := range []any{&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}} {
+		if !db.Migrator().HasTable(model) {
+			if err := db.Migrator().CreateTable(model); err != nil {
+				return err
+			}
+		}
+	}
+	return db.Where("version = ?", version).FirstOrCreate(&common.Migration{Version: version}).Error
+}

server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics_test.go

@@ -0,0 +1,48 @@
+package version_local
+
+import (
+	"go-admin/app/goauto/models"
+	common "go-admin/common/models"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+	"testing"
+)
+
+func TestAgentDiagnosticsMigrationAppendAndRollback(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&common.Migration{}); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 2; i++ {
+		if err := MigrateAgentDiagnostics(db, "test378"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, model := range []any{&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}} {
+		if !db.Migrator().HasTable(model) {
+			t.Fatalf("missing table %T", model)
+		}
+	}
+	for _, idx := range []struct {
+		model any
+		name  string
+	}{{&models.AgentHeartbeatLog{}, "ix_heartbeat_device_received"}, {&models.AgentHeartbeatLog{}, "ix_heartbeat_received"}, {&models.AgentDeviceStatusEvent{}, "ix_device_event_occurred"}} {
+		if !db.Migrator().HasIndex(idx.model, idx.name) {
+			t.Fatalf("missing index %s", idx.name)
+		}
+	}
+	// Existing migration runner only supports up. Rollback is an explicit,
+	// operator-authorized removal of these two new diagnostic tables/version.
+	if err := db.Migrator().DropTable(&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Where("version = ?", "test378").Delete(&common.Migration{}).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := MigrateAgentDiagnostics(db, "test378"); err != nil {
+		t.Fatal(err)
+	}
+}