[优化] 改名

2026-10-07 08:06:37 +08:00 · 2026-03-15 16:04:54 +08:00
parent d68773c554
commit 32d90ba641
304 changed files with 629 additions and 969 deletions
@@ -0,0 +1,404 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"time"
+
+	"openflare-agent/internal/config"
+	"openflare-agent/internal/observability"
+	"openflare-agent/internal/protocol"
+	"openflare-agent/internal/state"
+)
+
+type HeartbeatService interface {
+	Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error)
+	Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error)
+	SetToken(token string)
+}
+
+type SyncService interface {
+	SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error
+	SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
+}
+
+type Updater interface {
+	CheckAndUpdate(ctx context.Context, repo string, options UpdateOptions) error
+}
+
+type RuntimeManager interface {
+	CheckHealth(ctx context.Context) error
+	Restart(ctx context.Context) error
+}
+
+type UpdateOptions struct {
+	Channel string
+	TagName string
+	Force   bool
+}
+
+type Runner struct {
+	Config              *config.Config
+	StateStore          *state.Store
+	ObservabilityBuffer *state.ObservabilityBufferStore
+	HeartbeatService    HeartbeatService
+	SyncService         SyncService
+	Updater             Updater
+	RuntimeManager      RuntimeManager
+
+	autoUpdate          bool
+	updateNow           bool
+	updateRepo          string
+	updateChan          string
+	updateTag           string
+	restartOpenrestyNow bool
+}
+
+func (r *Runner) Run(ctx context.Context) error {
+	nodeID, err := r.StateStore.EnsureNodeID()
+	if err != nil {
+		return err
+	}
+	slog.Info("agent runner started", "node_id", nodeID, "node", r.Config.NodeName, "ip", r.Config.NodeIP)
+	if r.hasAgentToken() {
+		r.refreshOpenrestyHealth(ctx)
+		payload, ackWindows := r.prepareHeartbeatPayload(nodeID)
+		heartbeatResult, hbErr := r.HeartbeatService.Heartbeat(ctx, payload)
+		if hbErr != nil {
+			slog.Error("agent startup heartbeat failed", "error", hbErr)
+		} else {
+			r.ackObservabilityWindows(ackWindows)
+			if heartbeatResult == nil {
+				heartbeatResult = &protocol.HeartbeatResult{}
+			}
+			slog.Debug("agent startup heartbeat succeeded", "node_id", nodeID)
+			r.applySettings(heartbeatResult.AgentSettings)
+			if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
+				r.recordSyncError(err)
+				slog.Error("agent startup sync failed", "error", err)
+			} else {
+				slog.Debug("agent startup sync completed")
+			}
+			r.tryRestartOpenresty(ctx)
+			r.tryAutoUpdate(ctx)
+		}
+	} else if err = r.tryRegister(ctx, &nodeID); err != nil {
+		slog.Error("agent initial discovery register failed", "error", err)
+	}
+
+	heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration())
+	defer heartbeatTicker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			slog.Info("agent runner shutting down", "error", ctx.Err())
+			return ctx.Err()
+		case <-heartbeatTicker.C:
+			if !r.hasAgentToken() {
+				if err = r.tryRegister(ctx, &nodeID); err != nil {
+					slog.Error("agent discovery register failed", "error", err)
+				}
+				continue
+			}
+			r.refreshOpenrestyHealth(ctx)
+			payload, ackWindows := r.prepareHeartbeatPayload(nodeID)
+			heartbeatResult, hbErr := r.HeartbeatService.Heartbeat(ctx, payload)
+			if hbErr != nil {
+				slog.Error("agent heartbeat failed", "error", hbErr)
+			} else {
+				r.ackObservabilityWindows(ackWindows)
+				if heartbeatResult == nil {
+					heartbeatResult = &protocol.HeartbeatResult{}
+				}
+				if changed := r.applySettings(heartbeatResult.AgentSettings); changed {
+					heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration())
+				}
+				if err = r.SyncService.SyncOnce(ctx, heartbeatResult.ActiveConfig); err != nil {
+					r.recordSyncError(err)
+					slog.Error("agent sync failed", "error", err)
+				}
+				r.tryRestartOpenresty(ctx)
+				r.tryAutoUpdate(ctx)
+			}
+		}
+	}
+}
+
+func (r *Runner) hasAgentToken() bool {
+	return strings.TrimSpace(r.Config.AgentToken) != ""
+}
+
+func (r *Runner) applySettings(settings *protocol.AgentSettings) bool {
+	if settings == nil {
+		return false
+	}
+	changed := false
+	if settings.HeartbeatInterval > 0 {
+		newInterval := config.MillisecondDuration(time.Duration(settings.HeartbeatInterval) * time.Millisecond)
+		if newInterval != r.Config.HeartbeatInterval {
+			slog.Info("agent heartbeat interval updated", "from", r.Config.HeartbeatInterval, "to", newInterval)
+			r.Config.HeartbeatInterval = newInterval
+			changed = true
+		}
+	}
+	r.autoUpdate = settings.AutoUpdate
+	r.updateNow = settings.UpdateNow
+	r.updateRepo = strings.TrimSpace(settings.UpdateRepo)
+	r.updateChan = strings.TrimSpace(settings.UpdateChannel)
+	r.updateTag = strings.TrimSpace(settings.UpdateTag)
+	r.restartOpenrestyNow = settings.RestartOpenrestyNow
+	return changed
+}
+
+func (r *Runner) tryRestartOpenresty(ctx context.Context) {
+	if !r.restartOpenrestyNow {
+		return
+	}
+	r.restartOpenrestyNow = false
+	if r.RuntimeManager == nil {
+		return
+	}
+	slog.Info("agent openresty restart requested by server")
+	if err := r.RuntimeManager.Restart(ctx); err != nil {
+		slog.Error("agent openresty restart failed", "error", err)
+		r.recordOpenrestyUnhealthy(err, false)
+		return
+	}
+	slog.Info("agent openresty restart succeeded")
+	r.recordOpenrestyHealthy()
+}
+
+func (r *Runner) tryAutoUpdate(ctx context.Context) {
+	force := r.updateNow
+	shouldCheck := r.autoUpdate || force
+	r.updateNow = false
+	r.updateTag = strings.TrimSpace(r.updateTag)
+	if !shouldCheck || r.Updater == nil || r.updateRepo == "" {
+		return
+	}
+	channel := "stable"
+	if force && r.updateChan != "" {
+		channel = r.updateChan
+	}
+	if err := r.Updater.CheckAndUpdate(ctx, r.updateRepo, UpdateOptions{
+		Channel: channel,
+		TagName: r.updateTag,
+		Force:   force,
+	}); err != nil {
+		slog.Error("agent update check failed", "error", err)
+	}
+	if force {
+		r.updateTag = ""
+		r.updateChan = ""
+	}
+}
+
+func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error {
+	if strings.TrimSpace(r.Config.DiscoveryToken) == "" {
+		return errors.New("agent_token 为空且未配置 discovery_token")
+	}
+	slog.Info("agent discovery registration started")
+	response, err := r.HeartbeatService.Register(ctx, r.nodePayload(*nodeID))
+	if err != nil {
+		return err
+	}
+	if response == nil || strings.TrimSpace(response.AgentToken) == "" || strings.TrimSpace(response.NodeID) == "" {
+		return errors.New("discovery register response 缺少 node_id 或 agent_token")
+	}
+	snapshot, err := r.StateStore.Load()
+	if err != nil {
+		return err
+	}
+	snapshot.NodeID = response.NodeID
+	if err = r.StateStore.Save(snapshot); err != nil {
+		return err
+	}
+	r.Config.AgentToken = response.AgentToken
+	r.Config.DiscoveryToken = ""
+	if err = r.Config.Save(); err != nil {
+		return err
+	}
+	r.HeartbeatService.SetToken(response.AgentToken)
+	*nodeID = response.NodeID
+	slog.Info("agent discovery registration succeeded", "node_id", response.NodeID)
+	r.refreshOpenrestyHealth(ctx)
+	payload, ackWindows := r.prepareHeartbeatPayload(*nodeID)
+	heartbeatResult, heartbeatErr := r.HeartbeatService.Heartbeat(ctx, payload)
+	if heartbeatErr != nil {
+		slog.Error("agent post-register heartbeat failed", "error", heartbeatErr)
+		return nil
+	}
+	r.ackObservabilityWindows(ackWindows)
+	if heartbeatResult == nil {
+		heartbeatResult = &protocol.HeartbeatResult{}
+	}
+	r.applySettings(heartbeatResult.AgentSettings)
+	if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
+		r.recordSyncError(err)
+		slog.Error("agent post-register startup sync failed", "error", err)
+	} else {
+		slog.Debug("agent post-register startup sync completed")
+	}
+	r.tryRestartOpenresty(ctx)
+	r.tryAutoUpdate(ctx)
+	return nil
+}
+
+func (r *Runner) recordSyncError(err error) {
+	if err == nil || r.StateStore == nil {
+		return
+	}
+	snapshot, loadErr := r.StateStore.Load()
+	if loadErr != nil {
+		slog.Error("load state before recording sync error failed", "error", loadErr)
+		return
+	}
+	snapshot.LastError = err.Error()
+	slog.Warn("recording sync error into state", "error", snapshot.LastError)
+	if saveErr := r.StateStore.Save(snapshot); saveErr != nil {
+		slog.Error("save state after sync error failed", "error", saveErr)
+	}
+}
+
+func (r *Runner) refreshOpenrestyHealth(ctx context.Context) {
+	if r.RuntimeManager == nil || r.StateStore == nil {
+		return
+	}
+	if err := r.RuntimeManager.CheckHealth(ctx); err != nil {
+		r.recordOpenrestyUnhealthy(err, true)
+		return
+	}
+	r.recordOpenrestyHealthy()
+}
+
+func (r *Runner) recordOpenrestyHealthy() {
+	if r.StateStore == nil {
+		return
+	}
+	snapshot, err := r.StateStore.Load()
+	if err != nil {
+		slog.Error("load state before recording openresty health failed", "error", err)
+		return
+	}
+	if snapshot.OpenrestyStatus == protocol.OpenrestyStatusHealthy && strings.TrimSpace(snapshot.OpenrestyMessage) == "" {
+		return
+	}
+	snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
+	snapshot.OpenrestyMessage = ""
+	if err = r.StateStore.Save(snapshot); err != nil {
+		slog.Error("save state after recording openresty health failed", "error", err)
+	}
+}
+
+func (r *Runner) recordOpenrestyUnhealthy(err error, fallbackOnly bool) {
+	if err == nil || r.StateStore == nil {
+		return
+	}
+	snapshot, loadErr := r.StateStore.Load()
+	if loadErr != nil {
+		slog.Error("load state before recording openresty error failed", "error", loadErr)
+		return
+	}
+	message := strings.TrimSpace(err.Error())
+	if !fallbackOnly || strings.TrimSpace(snapshot.OpenrestyMessage) == "" {
+		snapshot.OpenrestyMessage = message
+	}
+	snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
+	if saveErr := r.StateStore.Save(snapshot); saveErr != nil {
+		slog.Error("save state after recording openresty error failed", "error", saveErr)
+	}
+}
+
+func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
+	snapshot, _ := r.StateStore.Load()
+	openrestyStatus := strings.TrimSpace(snapshot.OpenrestyStatus)
+	if openrestyStatus == "" {
+		openrestyStatus = protocol.OpenrestyStatusUnknown
+	}
+	profile := observability.BuildProfile(r.Config, r.StateStore)
+	managedOpenRestyMetrics := observability.CollectManagedOpenRestyMetrics(r.Config)
+	trafficReport, accessLogs, fallbackMetrics := observability.BuildTrafficObservability(r.Config, r.StateStore, managedOpenRestyMetrics)
+	if managedOpenRestyMetrics == nil {
+		managedOpenRestyMetrics = fallbackMetrics
+	}
+	metricSnapshot := observability.BuildSnapshot(r.Config, r.StateStore, managedOpenRestyMetrics)
+	healthEvents := observability.BuildHealthEvents(snapshot)
+	return protocol.NodePayload{
+		NodeID:           nodeID,
+		Name:             r.Config.NodeName,
+		IP:               r.Config.NodeIP,
+		AgentVersion:     r.Config.AgentVersion,
+		NginxVersion:     r.Config.NginxVersion,
+		CurrentVersion:   snapshot.CurrentVersion,
+		LastError:        snapshot.LastError,
+		OpenrestyStatus:  openrestyStatus,
+		OpenrestyMessage: snapshot.OpenrestyMessage,
+		Profile:          profile,
+		Snapshot:         metricSnapshot,
+		TrafficReport:    trafficReport,
+		AccessLogs:       accessLogs,
+		HealthEvents:     healthEvents,
+	}
+}
+
+func (r *Runner) prepareHeartbeatPayload(nodeID string) (protocol.NodePayload, []int64) {
+	payload := r.nodePayload(nodeID)
+	if r.ObservabilityBuffer == nil || (payload.Snapshot == nil && payload.TrafficReport == nil && len(payload.AccessLogs) == 0) {
+		return payload, nil
+	}
+	now := time.Now().UTC()
+	retainAfterUnix := now.Add(-time.Duration(r.Config.ObservabilityReplayMinutes) * time.Minute).Unix()
+	windowStartedAtUnix := state.ObservabilityWindowStartedAt(payload.Snapshot, payload.TrafficReport)
+	if windowStartedAtUnix <= 0 {
+		return payload, nil
+	}
+
+	record := state.ObservabilityBufferRecord{
+		WindowStartedAtUnix: windowStartedAtUnix,
+		Snapshot:            payload.Snapshot,
+		TrafficReport:       payload.TrafficReport,
+		AccessLogs:          payload.AccessLogs,
+		QueuedAtUnix:        now.Unix(),
+	}
+	if err := r.ObservabilityBuffer.Upsert(record, retainAfterUnix); err != nil {
+		slog.Error("upsert observability buffer failed", "error", err)
+		return payload, nil
+	}
+
+	records, err := r.ObservabilityBuffer.Replayable(windowStartedAtUnix, retainAfterUnix)
+	if err != nil {
+		slog.Error("load replayable observability buffer failed", "error", err)
+		return payload, []int64{windowStartedAtUnix}
+	}
+
+	ackWindows := make([]int64, 0, len(records)+1)
+	buffered := make([]protocol.BufferedObservabilityRecord, 0, len(records))
+	for _, item := range records {
+		if item.WindowStartedAtUnix <= 0 {
+			continue
+		}
+		buffered = append(buffered, protocol.BufferedObservabilityRecord{
+			WindowStartedAtUnix: item.WindowStartedAtUnix,
+			Snapshot:            item.Snapshot,
+			TrafficReport:       item.TrafficReport,
+			AccessLogs:          item.AccessLogs,
+		})
+		ackWindows = append(ackWindows, item.WindowStartedAtUnix)
+	}
+	payload.BufferedObservability = buffered
+	ackWindows = append(ackWindows, windowStartedAtUnix)
+	return payload, ackWindows
+}
+
+func (r *Runner) ackObservabilityWindows(windowStartedAtUnix []int64) {
+	if r.ObservabilityBuffer == nil || len(windowStartedAtUnix) == 0 {
+		return
+	}
+	retainAfterUnix := time.Now().UTC().Add(-time.Duration(r.Config.ObservabilityReplayMinutes) * time.Minute).Unix()
+	if err := r.ObservabilityBuffer.Ack(windowStartedAtUnix, retainAfterUnix); err != nil {
+		slog.Error("ack observability buffer failed", "error", err)
+	}
+}
@@ -0,0 +1,496 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"openflare-agent/internal/config"
+	"openflare-agent/internal/protocol"
+	"openflare-agent/internal/state"
+)
+
+type fakeHeartbeatService struct {
+	mu                sync.Mutex
+	registerCalls     int
+	heartbeatCalls    int
+	registerErr       error
+	registerResp      *protocol.RegisterNodeResponse
+	heartbeatErrs     []error
+	heartbeatResults  []*protocol.HeartbeatResult
+	heartbeatPayloads []protocol.NodePayload
+	onHeartbeat       func(int)
+	lastToken         string
+}
+
+func (f *fakeHeartbeatService) Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.registerCalls++
+	return f.registerResp, f.registerErr
+}
+
+func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error) {
+	f.mu.Lock()
+	f.heartbeatCalls++
+	callIndex := f.heartbeatCalls
+	f.heartbeatPayloads = append(f.heartbeatPayloads, payload)
+	var err error
+	if len(f.heartbeatErrs) >= callIndex {
+		err = f.heartbeatErrs[callIndex-1]
+	}
+	var result *protocol.HeartbeatResult
+	if len(f.heartbeatResults) >= callIndex {
+		result = f.heartbeatResults[callIndex-1]
+	}
+	onHeartbeat := f.onHeartbeat
+	f.mu.Unlock()
+	if onHeartbeat != nil {
+		onHeartbeat(callIndex)
+	}
+	return result, err
+}
+
+func (f *fakeHeartbeatService) SetToken(token string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.lastToken = token
+}
+
+type fakeSyncService struct {
+	mu             sync.Mutex
+	startupErr     error
+	syncOnceErr    error
+	startupCalls   int
+	syncOnceCalls  int
+	onSyncOnceCall func(int)
+}
+
+type fakeRuntimeManager struct {
+	mu                   sync.Mutex
+	healthErr            error
+	restartErr           error
+	restartCalls         int
+	clearHealthOnRestart bool
+}
+
+func (f *fakeRuntimeManager) CheckHealth(ctx context.Context) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.healthErr
+}
+
+func (f *fakeRuntimeManager) Restart(ctx context.Context) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.restartCalls++
+	if f.clearHealthOnRestart && f.restartErr == nil {
+		f.healthErr = nil
+	}
+	return f.restartErr
+}
+
+func (f *fakeSyncService) SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.startupCalls++
+	return f.startupErr
+}
+
+func (f *fakeSyncService) SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
+	f.mu.Lock()
+	f.syncOnceCalls++
+	callIndex := f.syncOnceCalls
+	callback := f.onSyncOnceCall
+	f.mu.Unlock()
+	if callback != nil {
+		callback(callIndex)
+	}
+	return f.syncOnceErr
+}
+
+func TestRunnerKeepsHeartbeatWhenStartupSyncFails(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
+	heartbeatService := &fakeHeartbeatService{
+		heartbeatResults: []*protocol.HeartbeatResult{{}},
+		onHeartbeat: func(callCount int) {
+			if callCount >= 2 {
+				cancel()
+			}
+		},
+	}
+	syncService := &fakeSyncService{
+		startupErr: errors.New("当前没有激活版本，保持当前 OpenResty 配置"),
+	}
+	runner := &Runner{
+		Config: &config.Config{
+			AgentToken:        "agent-token",
+			NodeName:          "edge-01",
+			NodeIP:            "10.0.0.8",
+			AgentVersion:      config.AgentVersion,
+			NginxVersion:      "1.27.1.2",
+			HeartbeatInterval: config.MillisecondDuration(10 * time.Millisecond),
+		},
+		StateStore:       stateStore,
+		HeartbeatService: heartbeatService,
+		SyncService:      syncService,
+	}
+
+	err := runner.Run(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context cancellation, got %v", err)
+	}
+	if heartbeatService.registerCalls != 0 {
+		t.Fatalf("expected no discovery register call, got %d", heartbeatService.registerCalls)
+	}
+	if heartbeatService.heartbeatCalls < 2 {
+		t.Fatalf("expected heartbeat loop to continue, got %d heartbeat calls", heartbeatService.heartbeatCalls)
+	}
+	snapshot, loadErr := stateStore.Load()
+	if loadErr != nil {
+		t.Fatalf("failed to load state: %v", loadErr)
+	}
+	if snapshot.LastError != "当前没有激活版本，保持当前 OpenResty 配置" {
+		t.Fatalf("expected startup sync error to be recorded, got %q", snapshot.LastError)
+	}
+}
+
+func TestRunnerDoesNotExitOnHeartbeatOrSyncError(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
+	heartbeatService := &fakeHeartbeatService{
+		registerErr:   errors.New("register timeout"),
+		heartbeatErrs: []error{errors.New("heartbeat timeout")},
+		heartbeatResults: []*protocol.HeartbeatResult{
+			{},
+		},
+	}
+	syncService := &fakeSyncService{
+		syncOnceErr: errors.New("openresty reload failed"),
+		onSyncOnceCall: func(callCount int) {
+			if callCount >= 1 {
+				cancel()
+			}
+		},
+	}
+	runner := &Runner{
+		Config: &config.Config{
+			AgentToken:        "agent-token",
+			NodeName:          "edge-01",
+			NodeIP:            "10.0.0.8",
+			AgentVersion:      config.AgentVersion,
+			NginxVersion:      "1.27.1.2",
+			HeartbeatInterval: config.MillisecondDuration(10 * time.Millisecond),
+		},
+		StateStore:       stateStore,
+		HeartbeatService: heartbeatService,
+		SyncService:      syncService,
+	}
+
+	err := runner.Run(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context cancellation, got %v", err)
+	}
+	if heartbeatService.registerCalls != 0 {
+		t.Fatalf("expected no register attempt, got %d", heartbeatService.registerCalls)
+	}
+	if syncService.syncOnceCalls == 0 {
+		t.Fatal("expected sync loop to continue after heartbeat/register errors")
+	}
+	snapshot, loadErr := stateStore.Load()
+	if loadErr != nil {
+		t.Fatalf("failed to load state: %v", loadErr)
+	}
+	if snapshot.LastError != "openresty reload failed" {
+		t.Fatalf("expected sync error to be recorded, got %q", snapshot.LastError)
+	}
+}
+
+func TestRunnerReportsOpenrestyHealthAndExecutesRestart(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
+	if err := stateStore.Save(&state.Snapshot{
+		OpenrestyStatus:  protocol.OpenrestyStatusUnhealthy,
+		OpenrestyMessage: "docker run openresty failed: bind 80 already allocated",
+	}); err != nil {
+		t.Fatalf("failed to seed state: %v", err)
+	}
+	heartbeatService := &fakeHeartbeatService{
+		heartbeatResults: []*protocol.HeartbeatResult{{
+			AgentSettings: &protocol.AgentSettings{RestartOpenrestyNow: true},
+		}},
+		onHeartbeat: func(callCount int) {
+			if callCount >= 1 {
+				cancel()
+			}
+		},
+	}
+	runtimeManager := &fakeRuntimeManager{
+		healthErr:            errors.New("docker openresty container is not running"),
+		clearHealthOnRestart: true,
+	}
+	runner := &Runner{
+		Config: &config.Config{
+			AgentToken:        "agent-token",
+			NodeName:          "edge-01",
+			NodeIP:            "10.0.0.8",
+			AgentVersion:      config.AgentVersion,
+			NginxVersion:      "1.27.1.2",
+			HeartbeatInterval: config.MillisecondDuration(10 * time.Millisecond),
+		},
+		StateStore:       stateStore,
+		HeartbeatService: heartbeatService,
+		SyncService:      &fakeSyncService{},
+		RuntimeManager:   runtimeManager,
+	}
+
+	err := runner.Run(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context cancellation, got %v", err)
+	}
+	if len(heartbeatService.heartbeatPayloads) == 0 {
+		t.Fatal("expected at least one heartbeat payload")
+	}
+	payload := heartbeatService.heartbeatPayloads[0]
+	if payload.OpenrestyStatus != protocol.OpenrestyStatusUnhealthy {
+		t.Fatalf("expected unhealthy openresty status in heartbeat payload, got %q", payload.OpenrestyStatus)
+	}
+	if payload.OpenrestyMessage != "docker run openresty failed: bind 80 already allocated" {
+		t.Fatalf("unexpected openresty message: %q", payload.OpenrestyMessage)
+	}
+	if runtimeManager.restartCalls != 1 {
+		t.Fatalf("expected one openresty restart attempt, got %d", runtimeManager.restartCalls)
+	}
+	snapshot, loadErr := stateStore.Load()
+	if loadErr != nil {
+		t.Fatalf("failed to load state: %v", loadErr)
+	}
+	if snapshot.OpenrestyStatus != protocol.OpenrestyStatusHealthy || snapshot.OpenrestyMessage != "" {
+		t.Fatal("expected restart success to mark openresty healthy")
+	}
+}
+
+func TestRunnerHeartbeatPayloadIncludesObservabilityExtensions(t *testing.T) {
+	tempDir := t.TempDir()
+	stateStore := state.NewStore(filepath.Join(tempDir, "state.json"))
+	if err := stateStore.Save(&state.Snapshot{
+		NodeID:           "node-observe",
+		CurrentVersion:   "20260314-001",
+		LastError:        "sync failed",
+		OpenrestyStatus:  protocol.OpenrestyStatusUnhealthy,
+		OpenrestyMessage: "reload failed",
+	}); err != nil {
+		t.Fatalf("failed to seed state: %v", err)
+	}
+
+	runner := &Runner{
+		Config: &config.Config{
+			NodeName:          "edge-observe-1",
+			NodeIP:            "10.0.0.51",
+			AgentVersion:      config.AgentVersion,
+			NginxVersion:      "1.27.1.2",
+			DataDir:           tempDir,
+			RouteConfigPath:   filepath.Join(tempDir, "conf.d", "openflare_routes.conf"),
+			HeartbeatInterval: config.MillisecondDuration(10 * time.Millisecond),
+		},
+		StateStore: stateStore,
+	}
+	if err := os.MkdirAll(filepath.Dir(runner.Config.RouteConfigPath), 0o755); err != nil {
+		t.Fatalf("failed to prepare route config dir: %v", err)
+	}
+	if err := os.WriteFile(
+		filepath.Join(filepath.Dir(runner.Config.RouteConfigPath), "openflare_access.log"),
+		[]byte("{\"ts\":\""+time.Now().UTC().Format(time.RFC3339)+"\",\"host\":\"edge.example.com\",\"path\":\"/\",\"remote_addr\":\"10.0.0.8\",\"status\":200}\n"),
+		0o644,
+	); err != nil {
+		t.Fatalf("failed to prepare access log: %v", err)
+	}
+
+	firstPayload := runner.nodePayload("node-observe")
+	if firstPayload.Profile == nil {
+		t.Fatal("expected first heartbeat payload to include system profile")
+	}
+	if firstPayload.Snapshot == nil {
+		t.Fatal("expected first heartbeat payload to include metric snapshot")
+	}
+	if firstPayload.TrafficReport == nil || firstPayload.TrafficReport.RequestCount != 1 {
+		t.Fatalf("expected first heartbeat payload to include traffic report, got %+v", firstPayload.TrafficReport)
+	}
+	if len(firstPayload.AccessLogs) != 1 || firstPayload.AccessLogs[0].Path != "/" {
+		t.Fatalf("expected first heartbeat payload to include access logs, got %+v", firstPayload.AccessLogs)
+	}
+	if len(firstPayload.HealthEvents) != 2 {
+		t.Fatalf("expected health events for openresty and sync error, got %+v", firstPayload.HealthEvents)
+	}
+
+	secondPayload := runner.nodePayload("node-observe")
+	if secondPayload.Profile != nil {
+		t.Fatal("expected unchanged profile to be omitted on subsequent heartbeat")
+	}
+	if secondPayload.Snapshot == nil {
+		t.Fatal("expected metric snapshot to continue reporting on subsequent heartbeat")
+	}
+	if secondPayload.TrafficReport != nil {
+		t.Fatalf("expected unchanged traffic window to be omitted on subsequent heartbeat, got %+v", secondPayload.TrafficReport)
+	}
+	if len(secondPayload.AccessLogs) != 0 {
+		t.Fatalf("expected unchanged access log delta to be omitted on subsequent heartbeat, got %+v", secondPayload.AccessLogs)
+	}
+}
+
+func TestRunnerReplaysBufferedObservabilityAfterHeartbeatRecovery(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	tempDir := t.TempDir()
+	stateStore := state.NewStore(filepath.Join(tempDir, "state.json"))
+	bufferStore := state.NewObservabilityBufferStore(filepath.Join(tempDir, "observability-buffer.json"))
+	nowUnix := time.Now().UTC().Unix()
+	bufferWindow := nowUnix - (nowUnix % 60) - 60
+	if err := bufferStore.Upsert(state.ObservabilityBufferRecord{
+		WindowStartedAtUnix: bufferWindow,
+		Snapshot:            &protocol.NodeMetricSnapshot{CapturedAtUnix: bufferWindow + 5, CPUUsagePercent: 30},
+		TrafficReport:       &protocol.NodeTrafficReport{WindowStartedAtUnix: bufferWindow, WindowEndedAtUnix: bufferWindow + 60, RequestCount: 8},
+		QueuedAtUnix:        bufferWindow + 60,
+	}, 0); err != nil {
+		t.Fatalf("failed to seed observability buffer: %v", err)
+	}
+	heartbeatService := &fakeHeartbeatService{
+		heartbeatErrs:    []error{errors.New("server offline"), nil},
+		heartbeatResults: []*protocol.HeartbeatResult{{}, {}},
+		onHeartbeat: func(callCount int) {
+			if callCount >= 2 {
+				cancel()
+			}
+		},
+	}
+	runner := &Runner{
+		Config: &config.Config{
+			AgentToken:                 "agent-token",
+			NodeName:                   "edge-buffer-01",
+			NodeIP:                     "10.0.0.52",
+			AgentVersion:               config.AgentVersion,
+			NginxVersion:               "1.27.1.2",
+			DataDir:                    tempDir,
+			RouteConfigPath:            filepath.Join(tempDir, "conf.d", "openflare_routes.conf"),
+			HeartbeatInterval:          config.MillisecondDuration(10 * time.Millisecond),
+			ObservabilityReplayMinutes: 15,
+		},
+		StateStore:          stateStore,
+		ObservabilityBuffer: bufferStore,
+		HeartbeatService:    heartbeatService,
+		SyncService:         &fakeSyncService{},
+	}
+	if err := os.MkdirAll(filepath.Dir(runner.Config.RouteConfigPath), 0o755); err != nil {
+		t.Fatalf("failed to prepare route config dir: %v", err)
+	}
+	if err := os.WriteFile(
+		filepath.Join(filepath.Dir(runner.Config.RouteConfigPath), "openflare_access.log"),
+		[]byte("{\"ts\":\""+time.Now().UTC().Format(time.RFC3339)+"\",\"host\":\"edge.example.com\",\"path\":\"/\",\"remote_addr\":\"10.0.0.8\",\"status\":200}\n"),
+		0o644,
+	); err != nil {
+		t.Fatalf("failed to prepare access log: %v", err)
+	}
+
+	runErr := runner.Run(ctx)
+	if runErr != context.Canceled {
+		t.Fatalf("expected run to stop by context cancellation, got %v", runErr)
+	}
+	if len(heartbeatService.heartbeatPayloads) != 2 {
+		t.Fatalf("expected two heartbeat payloads, got %d", len(heartbeatService.heartbeatPayloads))
+	}
+	secondPayload := heartbeatService.heartbeatPayloads[1]
+	if len(secondPayload.BufferedObservability) != 1 {
+		t.Fatalf("expected second heartbeat to replay one buffered observation, got %+v", secondPayload.BufferedObservability)
+	}
+	if len(secondPayload.BufferedObservability[0].AccessLogs) != 0 {
+		t.Fatalf("expected seeded buffered observation to keep empty access logs, got %+v", secondPayload.BufferedObservability[0].AccessLogs)
+	}
+
+	replayable, err := bufferStore.Replayable(0, 0)
+	if err != nil {
+		t.Fatalf("Replayable after recovery failed: %v", err)
+	}
+	if len(replayable) != 0 {
+		t.Fatalf("expected buffer to be acked after successful heartbeat, got %+v", replayable)
+	}
+}
+
+func TestRunnerDiscoveryRegisterUpdatesTokenAndNodeID(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
+	heartbeatService := &fakeHeartbeatService{
+		registerResp: &protocol.RegisterNodeResponse{
+			NodeID:     "node-server-assigned",
+			AgentToken: "agent-token-issued",
+			Name:       "edge-01",
+		},
+		heartbeatResults: []*protocol.HeartbeatResult{{}},
+		onHeartbeat: func(callCount int) {
+			if callCount >= 1 {
+				cancel()
+			}
+		},
+	}
+	syncService := &fakeSyncService{}
+	configPath := filepath.Join(t.TempDir(), "agent.json")
+	if err := os.WriteFile(configPath, []byte(`{"server_url":"http://127.0.0.1:3000","discovery_token":"discovery-token","node_name":"edge-01","node_ip":"10.0.0.8"}`), 0o644); err != nil {
+		t.Fatalf("failed to seed config file: %v", err)
+	}
+	cfg, err := config.Load(configPath)
+	if err != nil {
+		t.Fatalf("failed to load config: %v", err)
+	}
+	runner := &Runner{
+		Config: &config.Config{
+			ServerURL:         cfg.ServerURL,
+			DiscoveryToken:    cfg.DiscoveryToken,
+			NodeName:          cfg.NodeName,
+			NodeIP:            cfg.NodeIP,
+			AgentVersion:      config.AgentVersion,
+			NginxVersion:      "1.27.1.2",
+			HeartbeatInterval: config.MillisecondDuration(10 * time.Millisecond),
+		},
+		StateStore:       stateStore,
+		HeartbeatService: heartbeatService,
+		SyncService:      syncService,
+	}
+	runner.Config = cfg
+	runner.Config.AgentVersion = config.AgentVersion
+	runner.Config.NginxVersion = "1.27.1.2"
+	runner.Config.HeartbeatInterval = config.MillisecondDuration(10 * time.Millisecond)
+
+	err = runner.Run(ctx)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context cancellation, got %v", err)
+	}
+	if heartbeatService.registerCalls == 0 {
+		t.Fatal("expected discovery register to be attempted")
+	}
+	if heartbeatService.lastToken != "agent-token-issued" {
+		t.Fatalf("expected client token to be updated, got %q", heartbeatService.lastToken)
+	}
+	snapshot, loadErr := stateStore.Load()
+	if loadErr != nil {
+		t.Fatalf("failed to load state: %v", loadErr)
+	}
+	if snapshot.NodeID != "node-server-assigned" {
+		t.Fatalf("expected node id to be replaced, got %q", snapshot.NodeID)
+	}
+	if runner.Config.AgentToken != "agent-token-issued" || runner.Config.DiscoveryToken != "" {
+		t.Fatal("expected config token rotation to complete")
+	}
+}