test(agentcompat): model FD diagnostic lifecycles

Co-authored-by: naiba/CloudCode <hi+cloudcode@nai.ba>

integration/agentcompat/internal/scenario/stress_fd_diagnostic_fixture_test.go

@@ -0,0 +1,96 @@
+//go:build linux
+
+package scenario
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/agent"
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
+	processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
+)
+
+type fdDiagnosticWindowPair struct {
+	baseline fdDiagnosticAgentWindow
+	end      fdDiagnosticAgentWindow
+}
+
+type stressDiagnosticAgentWindowSpec struct {
+	Ordinal           int
+	PID               int
+	BaselineCount     int
+	EndCount          int
+	Target            string
+	BaselineSampledAt time.Time
+	EndSampledAt      time.Time
+}
+
+type stressDiagnosticProcessWindowSpec struct {
+	PID       int
+	Count     int
+	Target    string
+	SampledAt time.Time
+}
+
+type stressDiagnosticDashboardWindowSpec struct {
+	PID           int
+	BaselineCount int
+	EndCount      int
+}
+
+func stressDiagnosticAgentWindow(t *testing.T, spec stressDiagnosticAgentWindowSpec) fdDiagnosticWindowPair {
+	t.Helper()
+	if spec.BaselineSampledAt.IsZero() {
+		spec.BaselineSampledAt = time.Unix(int64(spec.Ordinal), 0).UTC()
+	}
+	if spec.EndSampledAt.IsZero() {
+		spec.EndSampledAt = spec.BaselineSampledAt.Add(time.Minute)
+	}
+	agentOrdinal, err := NewStressAgentOrdinal(spec.Ordinal)
+	require.NoError(t, err)
+	process, err := NewStressAgentProcess(agentOrdinal, spec.PID)
+	require.NoError(t, err)
+	identity := agent.ProcessIdentity{Generation: 1, PID: spec.PID}
+	return fdDiagnosticWindowPair{
+		baseline: fdDiagnosticAgentWindow{Process: process, Identity: identity, Window: stressDiagnosticProcessWindow(stressDiagnosticProcessWindowSpec{PID: spec.PID, Count: spec.BaselineCount, Target: spec.Target, SampledAt: spec.BaselineSampledAt})},
+		end:      fdDiagnosticAgentWindow{Process: process, Identity: identity, Window: stressDiagnosticProcessWindow(stressDiagnosticProcessWindowSpec{PID: spec.PID, Count: spec.EndCount, Target: spec.Target, SampledAt: spec.EndSampledAt})},
+	}
+}
+
+func stressDiagnosticDashboardWindow(t *testing.T, spec stressDiagnosticDashboardWindowSpec) fdDiagnosticWindowPair {
+	t.Helper()
+	process, err := NewStressDashboardProcess(spec.PID)
+	require.NoError(t, err)
+	return fdDiagnosticWindowPair{
+		baseline: fdDiagnosticAgentWindow{Process: process, Window: stressDiagnosticProcessWindow(stressDiagnosticProcessWindowSpec{PID: spec.PID, Count: spec.BaselineCount, Target: "dashboard"})},
+		end:      fdDiagnosticAgentWindow{Process: process, Window: stressDiagnosticProcessWindow(stressDiagnosticProcessWindowSpec{PID: spec.PID, Count: spec.EndCount, Target: "dashboard"})},
+	}
+}
+
+func stressDiagnosticProcessWindow(spec stressDiagnosticProcessWindowSpec) processharness.Window {
+	samples := make([]processharness.Sample, contract.ResourceSampleCount)
+	for index := range samples {
+		samples[index] = processharness.Sample{PID: spec.PID, NonStdioFDCount: spec.Count, FDObservations: []processharness.FDObservation{{Number: 3, Target: spec.Target}}, SampledAt: spec.SampledAt}
+	}
+	return processharness.Window{PID: spec.PID, Samples: samples}
+}
+
+func stressDiagnosticSample(ordinal int, observations []processharness.FDObservation) fdDiagnosticSample {
+	return fdDiagnosticSample{Ordinal: ordinal, FDObservations: observations}
+}
+
+func fdDiagnosticTestSampler(_ context.Context, pid int) (processharness.Sample, error) {
+	return processharness.Sample{PID: pid, FDObservations: []processharness.FDObservation{{Number: 4, Target: fmt.Sprintf("added-%d", pid)}}, SampledAt: time.Unix(int64(pid), 0).UTC()}, nil
+}
+
+func mustStressPRFullProfile(t *testing.T) contract.Profile {
+	t.Helper()
+	profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
+	require.NoError(t, err)
+	return profile
+}

integration/agentcompat/internal/scenario/stress_fd_diagnostic_model_test.go

@@ -0,0 +1,201 @@
+//go:build linux
+
+package scenario
+
+import (
+	"sort"
+	"time"
+
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/agent"
+	processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
+)
+
+type fdDiagnosticAgentWindow struct {
+	Process  StressProcessIdentity
+	Identity agent.ProcessIdentity
+	Window   processharness.Window
+}
+
+type fdDiagnosticCandidate struct {
+	Baseline fdDiagnosticAgentWindow
+	End      fdDiagnosticAgentWindow
+}
+
+type fdDiagnosticSample struct {
+	Ordinal           int                            `json:"sample_ordinal"`
+	PID               int                            `json:"pid"`
+	RSSBytes          uint64                         `json:"rss_bytes"`
+	DescendantPIDs    []int                          `json:"descendant_pids"`
+	DescendantCount   int                            `json:"descendant_count"`
+	NonStdioFDCount   int                            `json:"non_stdio_fd_count"`
+	TCPListenerCount  int                            `json:"tcp_listener_count"`
+	TCP6ListenerCount int                            `json:"tcp6_listener_count"`
+	FDObservations    []processharness.FDObservation `json:"fd_observations"`
+	ObservedAt        time.Time                      `json:"observed_at"`
+}
+
+type fdDiagnosticWindow struct {
+	PID     int                  `json:"pid"`
+	Samples []fdDiagnosticSample `json:"samples"`
+}
+
+type fdDiagnosticLifecycle struct {
+	Observation  processharness.FDObservation `json:"observation"`
+	Status       string                       `json:"status"`
+	NumberReused bool                         `json:"number_reused,omitempty"`
+}
+
+type fdDiagnosticRecord struct {
+	SamplerPID         int                            `json:"sampler_pid"`
+	AgentOrdinal       int                            `json:"agent_ordinal"`
+	AgentPID           int                            `json:"agent_pid"`
+	BaselineGeneration uint64                         `json:"baseline_generation"`
+	BaselinePID        int                            `json:"baseline_pid"`
+	EndPID             int                            `json:"end_pid"`
+	EndGeneration      uint64                         `json:"end_generation"`
+	Baseline           *fdDiagnosticWindow            `json:"baseline"`
+	End                *fdDiagnosticWindow            `json:"end"`
+	Tail               []fdDiagnosticSample           `json:"tail"`
+	AddedFinal         []processharness.FDObservation `json:"added_final"`
+	RemovedFinal       []processharness.FDObservation `json:"removed_final"`
+	Lifecycle          []fdDiagnosticLifecycle        `json:"lifecycle"`
+	LifecycleStatus    string                         `json:"lifecycle_status"`
+	DiagnosticError    string                         `json:"diagnostic_error,omitempty"`
+}
+
+func fdDiagnosticEnabled(value string) bool { return value == "1" }
+
+func newFDDiagnosticRecord(candidate fdDiagnosticCandidate, samplerPID int) (fdDiagnosticRecord, bool) {
+	record := fdDiagnosticRecord{
+		SamplerPID:         samplerPID,
+		AgentOrdinal:       candidate.Baseline.Process.Agent.Int(),
+		AgentPID:           candidate.Baseline.Identity.PID,
+		BaselineGeneration: candidate.Baseline.Identity.Generation,
+		BaselinePID:        candidate.Baseline.Identity.PID,
+		EndPID:             candidate.End.Identity.PID,
+		EndGeneration:      candidate.End.Identity.Generation,
+		Baseline:           fdDiagnosticWindowFromProcess(candidate.Baseline.Window),
+		End:                fdDiagnosticWindowFromProcess(candidate.End.Window),
+	}
+	if !fdDiagnosticIdentityMatches(candidate) {
+		record.LifecycleStatus = "process_identity_changed"
+		return record, false
+	}
+	baselineFinal := fdDiagnosticFinalObservations(candidate.Baseline.Window)
+	endFinal := fdDiagnosticFinalObservations(candidate.End.Window)
+	record.AddedFinal = fdDiagnosticDifference(endFinal, baselineFinal)
+	record.RemovedFinal = fdDiagnosticDifference(baselineFinal, endFinal)
+	record.LifecycleStatus = "tail_pending"
+	return record, true
+}
+
+func classifyFDDiagnosticLifecycle(added []processharness.FDObservation, tail []fdDiagnosticSample) []fdDiagnosticLifecycle {
+	result := make([]fdDiagnosticLifecycle, 0, len(added))
+	for _, observation := range added {
+		present := make([]bool, len(tail))
+		reused := false
+		for index, sample := range tail {
+			for _, tailObservation := range sample.FDObservations {
+				if tailObservation.Number == observation.Number && tailObservation.Target != observation.Target {
+					reused = true
+				}
+				if tailObservation == observation {
+					present[index] = true
+				}
+			}
+		}
+		status := "intermittent_or_reused"
+		switch {
+		case reused:
+			status = "intermittent_or_reused"
+		case !fdDiagnosticAnyPresent(present):
+			status = "cleared_before_tail"
+		case fdDiagnosticAllPresent(present):
+			status = "observed_through_tail"
+		case fdDiagnosticPrefixPresent(present):
+			status = "cleared_during_tail"
+		}
+		result = append(result, fdDiagnosticLifecycle{Observation: observation, Status: status, NumberReused: reused})
+	}
+	return result
+}
+
+func fdDiagnosticFinalCount(window processharness.Window) int {
+	if len(window.Samples) == 0 {
+		return 0
+	}
+	return window.Samples[len(window.Samples)-1].NonStdioFDCount
+}
+
+func fdDiagnosticIdentityMatches(candidate fdDiagnosticCandidate) bool {
+	baseline := candidate.Baseline
+	end := candidate.End
+	return baseline.Process.PID > 0 && baseline.Process.PID == baseline.Identity.PID && baseline.Identity.PID == baseline.Window.PID && end.Process.PID > 0 && end.Process.PID == end.Identity.PID && end.Identity.PID == end.Window.PID && baseline.Process.PID == end.Process.PID && baseline.Identity.Generation != 0 && baseline.Identity.Generation == end.Identity.Generation
+}
+
+func fdDiagnosticWindowFromProcess(window processharness.Window) *fdDiagnosticWindow {
+	result := &fdDiagnosticWindow{PID: window.PID, Samples: make([]fdDiagnosticSample, 0, len(window.Samples))}
+	for index, sample := range window.Samples {
+		result.Samples = append(result.Samples, fdDiagnosticSample{Ordinal: index + 1, PID: sample.PID, RSSBytes: sample.RSSBytes, DescendantPIDs: append([]int(nil), sample.DescendantPIDs...), DescendantCount: sample.DescendantCount, NonStdioFDCount: sample.NonStdioFDCount, TCPListenerCount: sample.TCPListenerCount, TCP6ListenerCount: sample.TCP6ListenerCount, FDObservations: fdDiagnosticSortedObservations(sample.FDObservations), ObservedAt: sample.SampledAt})
+	}
+	return result
+}
+
+func fdDiagnosticFinalObservations(window processharness.Window) []processharness.FDObservation {
+	if len(window.Samples) == 0 {
+		return nil
+	}
+	return fdDiagnosticSortedObservations(window.Samples[len(window.Samples)-1].FDObservations)
+}
+
+func fdDiagnosticDifference(left, right []processharness.FDObservation) []processharness.FDObservation {
+	rightSet := make(map[processharness.FDObservation]struct{}, len(right))
+	for _, observation := range right {
+		rightSet[observation] = struct{}{}
+	}
+	result := make([]processharness.FDObservation, 0)
+	for _, observation := range left {
+		if _, exists := rightSet[observation]; !exists {
+			result = append(result, observation)
+		}
+	}
+	return fdDiagnosticSortedObservations(result)
+}
+
+func fdDiagnosticSortedObservations(input []processharness.FDObservation) []processharness.FDObservation {
+	result := append([]processharness.FDObservation(nil), input...)
+	sort.Slice(result, func(left, right int) bool {
+		return result[left].Number < result[right].Number || (result[left].Number == result[right].Number && result[left].Target < result[right].Target)
+	})
+	return result
+}
+
+func fdDiagnosticAllPresent(present []bool) bool {
+	for _, value := range present {
+		if !value {
+			return false
+		}
+	}
+	return true
+}
+
+func fdDiagnosticAnyPresent(present []bool) bool {
+	for _, value := range present {
+		if value {
+			return true
+		}
+	}
+	return false
+}
+
+func fdDiagnosticPrefixPresent(present []bool) bool {
+	cleared := false
+	for _, value := range present {
+		if !value {
+			cleared = true
+		} else if cleared {
+			return false
+		}
+	}
+	return cleared
+}

integration/agentcompat/internal/scenario/stress_fd_diagnostic_tail_support_test.go

@@ -0,0 +1,208 @@
+//go:build linux && agentcompat
+
+package scenario
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
+	processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
+)
+
+const fdDiagnosticTailSampleCount = 20
+
+type fdDiagnosticSampleFunc func(context.Context, int) (processharness.Sample, error)
+
+type fdDiagnosticLogger interface {
+	Logf(string, ...any)
+}
+
+type fdDiagnosticCollectorSpec struct {
+	Enabled            bool
+	SamplerPID         int
+	TailResultCapacity int
+	TailInterval       time.Duration
+	Sample             fdDiagnosticSampleFunc
+}
+
+type fdDiagnosticTailResult struct {
+	AgentOrdinal int
+	Samples      []fdDiagnosticSample
+	Err          error
+}
+
+type fdDiagnosticTailSpec struct {
+	Context             context.Context
+	PID                 int
+	Interval            time.Duration
+	Sample              fdDiagnosticSampleFunc
+	FirstSampleComplete chan<- struct{}
+}
+
+type fdDiagnosticCollector struct {
+	enabled   bool
+	samplerID int
+	interval  time.Duration
+	sample    fdDiagnosticSampleFunc
+	baseline  map[int]fdDiagnosticAgentWindow
+	records   map[int]fdDiagnosticRecord
+	results   chan fdDiagnosticTailResult
+	started   int
+	waitOnce  sync.Once
+	logOnce   sync.Once
+	completed []fdDiagnosticRecord
+}
+
+func newFDDiagnosticCollector(spec fdDiagnosticCollectorSpec) *fdDiagnosticCollector {
+	capacity := spec.TailResultCapacity
+	if capacity < contract.PRFullAgentCount {
+		capacity = contract.PRFullAgentCount
+	}
+	return &fdDiagnosticCollector{
+		enabled:   spec.Enabled,
+		samplerID: spec.SamplerPID,
+		interval:  spec.TailInterval,
+		sample:    spec.Sample,
+		baseline:  make(map[int]fdDiagnosticAgentWindow),
+		records:   make(map[int]fdDiagnosticRecord),
+		results:   make(chan fdDiagnosticTailResult, capacity),
+	}
+}
+
+func newRealFDDiagnosticCollector(enabled bool) *fdDiagnosticCollector {
+	return newFDDiagnosticCollector(fdDiagnosticCollectorSpec{
+		Enabled:            enabled,
+		SamplerPID:         os.Getpid(),
+		TailResultCapacity: contract.PRFullAgentCount,
+		TailInterval:       contract.ResourceSampleInterval,
+		Sample: func(_ context.Context, pid int) (processharness.Sample, error) {
+			return processharness.SampleProcessWithFDObservations(pid)
+		},
+	})
+}
+
+func (collector *fdDiagnosticCollector) Enabled() bool { return collector != nil && collector.enabled }
+
+func (collector *fdDiagnosticCollector) RecordBaseline(window fdDiagnosticAgentWindow) {
+	if collector.Enabled() && window.Process.Kind == StressProcessAgent {
+		collector.baseline[window.Process.Agent.Int()] = window
+	}
+}
+
+func (collector *fdDiagnosticCollector) RecordEnd(ctx context.Context, window fdDiagnosticAgentWindow) {
+	if !collector.Enabled() || window.Process.Kind != StressProcessAgent {
+		return
+	}
+	baseline, exists := collector.baseline[window.Process.Agent.Int()]
+	if !exists || fdDiagnosticFinalCount(baseline.Window) == fdDiagnosticFinalCount(window.Window) {
+		return
+	}
+	record, startTail := newFDDiagnosticRecord(fdDiagnosticCandidate{Baseline: baseline, End: window}, collector.samplerID)
+	collector.records[record.AgentOrdinal] = record
+	if !startTail {
+		return
+	}
+	collector.started++
+	firstSampleComplete := make(chan struct{})
+	tailSpec := fdDiagnosticTailSpec{Context: ctx, PID: record.AgentPID, Interval: collector.interval, Sample: collector.sample, FirstSampleComplete: firstSampleComplete}
+	go func(ordinal int) {
+		result := collectFDDiagnosticTail(tailSpec)
+		result.AgentOrdinal = ordinal
+		collector.results <- result
+	}(record.AgentOrdinal)
+	select {
+	case <-firstSampleComplete:
+	case <-ctx.Done():
+	}
+}
+
+func (collector *fdDiagnosticCollector) WaitRecords() []fdDiagnosticRecord {
+	if !collector.Enabled() {
+		return nil
+	}
+	collector.waitOnce.Do(func() {
+		for range collector.started {
+			result := <-collector.results
+			record := collector.records[result.AgentOrdinal]
+			record.Tail = result.Samples
+			if result.Err != nil {
+				record.DiagnosticError = result.Err.Error()
+				record.LifecycleStatus = "tail_error"
+			} else {
+				record.Lifecycle = classifyFDDiagnosticLifecycle(record.AddedFinal, record.Tail)
+				record.LifecycleStatus = "tail_complete"
+			}
+			collector.records[result.AgentOrdinal] = record
+		}
+		collector.completed = make([]fdDiagnosticRecord, 0, len(collector.records))
+		for _, record := range collector.records {
+			collector.completed = append(collector.completed, record)
+		}
+		sort.Slice(collector.completed, func(left, right int) bool {
+			return collector.completed[left].AgentOrdinal < collector.completed[right].AgentOrdinal
+		})
+	})
+	return append([]fdDiagnosticRecord(nil), collector.completed...)
+}
+
+// The real collector is directly logger-injectable so tests exercise candidate selection and logging without shadow copies.
+func (collector *fdDiagnosticCollector) WaitAndLog(logger fdDiagnosticLogger) {
+	if !collector.Enabled() {
+		return
+	}
+	collector.logOnce.Do(func() {
+		for _, record := range collector.WaitRecords() {
+			encoded, err := json.Marshal(record)
+			if err != nil {
+				logger.Logf("agentcompat_fd_diagnostic={\"diagnostic_error\":%q}", err.Error())
+				continue
+			}
+			logger.Logf("agentcompat_fd_diagnostic=%s", encoded)
+		}
+	})
+}
+
+func collectFDDiagnosticTail(spec fdDiagnosticTailSpec) fdDiagnosticTailResult {
+	result := fdDiagnosticTailResult{Samples: make([]fdDiagnosticSample, 0, fdDiagnosticTailSampleCount)}
+	signalFirstSampleComplete := func() {
+		if spec.FirstSampleComplete != nil {
+			close(spec.FirstSampleComplete)
+			spec.FirstSampleComplete = nil
+		}
+	}
+	for index := 0; index < fdDiagnosticTailSampleCount; index++ {
+		if index > 0 {
+			timer := time.NewTimer(spec.Interval)
+			select {
+			case <-timer.C:
+			case <-spec.Context.Done():
+				timer.Stop()
+				result.Err = spec.Context.Err()
+				return result
+			}
+		}
+		if err := spec.Context.Err(); err != nil {
+			signalFirstSampleComplete()
+			result.Err = err
+			return result
+		}
+		processSample, err := spec.Sample(spec.Context, spec.PID)
+		if err != nil {
+			signalFirstSampleComplete()
+			result.Err = err
+			return result
+		}
+		result.Samples = append(result.Samples, fdDiagnosticSampleFromProcess(index+1, processSample))
+		signalFirstSampleComplete()
+	}
+	return result
+}
+
+func fdDiagnosticSampleFromProcess(ordinal int, sample processharness.Sample) fdDiagnosticSample {
+	return fdDiagnosticSample{Ordinal: ordinal, PID: sample.PID, RSSBytes: sample.RSSBytes, DescendantPIDs: append([]int(nil), sample.DescendantPIDs...), DescendantCount: sample.DescendantCount, NonStdioFDCount: sample.NonStdioFDCount, TCPListenerCount: sample.TCPListenerCount, TCP6ListenerCount: sample.TCP6ListenerCount, FDObservations: fdDiagnosticSortedObservations(sample.FDObservations), ObservedAt: sample.SampledAt}
+}

integration/agentcompat/internal/scenario/stress_fd_diagnostic_test.go

@@ -0,0 +1,171 @@
+//go:build linux && agentcompat
+
+package scenario
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/agent"
+	processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
+)
+
+func TestFDDiagnosticEnabled_OnlyAcceptsExactOne(t *testing.T) {
+	for _, value := range []string{"", "0", "true", "01", " 1", "1 ", "\t1"} {
+		require.False(t, fdDiagnosticEnabled(value), "value=%q", value)
+	}
+	require.True(t, fdDiagnosticEnabled("1"))
+}
+
+func TestFDDiagnosticCandidates_ExcludeDashboardAndMatchOrdinals(t *testing.T) {
+	collector := newFDDiagnosticCollector(fdDiagnosticCollectorSpec{Enabled: true, TailResultCapacity: 8, Sample: fdDiagnosticTestSampler})
+	agentOne := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 9, Target: "one"})
+	agentTwo := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 2, PID: 102, BaselineCount: 8, EndCount: 8, Target: "two"})
+	dashboard := stressDiagnosticDashboardWindow(t, stressDiagnosticDashboardWindowSpec{PID: 100, BaselineCount: 8, EndCount: 9})
+	collector.RecordBaseline(dashboard.baseline)
+	collector.RecordBaseline(agentTwo.baseline)
+	collector.RecordBaseline(agentOne.baseline)
+
+	collector.RecordEnd(t.Context(), agentTwo.end)
+	collector.RecordEnd(t.Context(), dashboard.end)
+	collector.RecordEnd(t.Context(), agentOne.end)
+	records := collector.WaitRecords()
+
+	require.Len(t, records, 1)
+	require.Equal(t, 1, records[0].AgentOrdinal)
+	require.Equal(t, 101, records[0].AgentPID)
+}
+
+func TestFDDiagnosticCandidates_UseOnlyFinalSampleCounts(t *testing.T) {
+	collector := newFDDiagnosticCollector(fdDiagnosticCollectorSpec{Enabled: true, TailResultCapacity: 8, Sample: fdDiagnosticTestSampler})
+	pair := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 8, Target: "stable"})
+	pair.end.Window.Samples[0].NonStdioFDCount = 9
+	collector.RecordBaseline(pair.baseline)
+
+	collector.RecordEnd(t.Context(), pair.end)
+	require.Empty(t, collector.WaitRecords())
+
+	pair.end.Window.Samples[len(pair.end.Window.Samples)-1].NonStdioFDCount = 9
+	collector = newFDDiagnosticCollector(fdDiagnosticCollectorSpec{Enabled: true, TailResultCapacity: 8, Sample: fdDiagnosticTestSampler})
+	collector.RecordBaseline(pair.baseline)
+	collector.RecordEnd(t.Context(), pair.end)
+	require.Len(t, collector.WaitRecords(), 1)
+}
+
+func TestFDDiagnosticRecord_ReportsIdentityChangeWithoutTail(t *testing.T) {
+	pair := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 9, Target: "changed"})
+	pair.end.Identity = agent.ProcessIdentity{Generation: 2, PID: 202}
+	candidate := fdDiagnosticCandidate{Baseline: pair.baseline, End: pair.end}
+
+	record, tail := newFDDiagnosticRecord(candidate, 999)
+
+	require.Equal(t, "process_identity_changed", record.LifecycleStatus)
+	require.False(t, tail)
+	require.Empty(t, record.Lifecycle)
+}
+
+func TestFDDiagnosticRecord_RejectsStressProcessPIDMismatch(t *testing.T) {
+	pair := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 9, Target: "process-pid-mismatch"})
+	pair.end.Process.PID = 202
+	candidate := fdDiagnosticCandidate{Baseline: pair.baseline, End: pair.end}
+
+	record, tail := newFDDiagnosticRecord(candidate, 999)
+
+	require.Equal(t, "process_identity_changed", record.LifecycleStatus)
+	require.False(t, tail)
+}
+
+func TestFDDiagnosticRecord_RejectsZeroRuntimeGeneration(t *testing.T) {
+	pair := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 9, Target: "zero-generation"})
+	pair.end.Identity.Generation = 0
+	candidate := fdDiagnosticCandidate{Baseline: pair.baseline, End: pair.end}
+
+	record, tail := newFDDiagnosticRecord(candidate, 999)
+
+	require.Equal(t, "process_identity_changed", record.LifecycleStatus)
+	require.False(t, tail)
+}
+
+func TestFDDiagnosticRecord_EncodesStableObservedAtForEveryWindowSample(t *testing.T) {
+	pair := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 9, Target: "timestamps"})
+	baselineTime := time.Date(2025, time.January, 2, 3, 4, 5, 0, time.UTC)
+	endTime := baselineTime.Add(time.Minute)
+	for index := range pair.baseline.Window.Samples {
+		pair.baseline.Window.Samples[index].SampledAt = baselineTime.Add(time.Duration(index) * time.Second)
+		pair.end.Window.Samples[index].SampledAt = endTime.Add(time.Duration(index) * time.Second)
+	}
+	candidate := fdDiagnosticCandidate{Baseline: pair.baseline, End: pair.end}
+
+	record, tail := newFDDiagnosticRecord(candidate, 999)
+	encoded, err := json.Marshal(record)
+
+	require.True(t, tail)
+	require.NoError(t, err)
+	require.Equal(t, baselineTime, record.Baseline.Samples[0].ObservedAt)
+	require.Equal(t, endTime, record.End.Samples[0].ObservedAt)
+	require.Contains(t, string(encoded), `"observed_at":"2025-01-02T03:04:05Z"`)
+}
+
+func TestFDDiagnosticRecord_DiffsFinalObservationsByExactIdentity(t *testing.T) {
+	pair := stressDiagnosticAgentWindow(t, stressDiagnosticAgentWindowSpec{Ordinal: 1, PID: 101, BaselineCount: 8, EndCount: 9, Target: "diff"})
+	pair.baseline.Window.Samples[4].FDObservations = []processharness.FDObservation{{Number: 5, Target: "beta"}, {Number: 4, Target: "alpha"}}
+	pair.end.Window.Samples[4].FDObservations = []processharness.FDObservation{{Number: 6, Target: "gamma"}, {Number: 5, Target: "beta"}}
+	candidate := fdDiagnosticCandidate{Baseline: pair.baseline, End: pair.end}
+
+	record, tail := newFDDiagnosticRecord(candidate, 999)
+
+	require.True(t, tail)
+	require.Equal(t, []processharness.FDObservation{{Number: 6, Target: "gamma"}}, record.AddedFinal)
+	require.Equal(t, []processharness.FDObservation{{Number: 4, Target: "alpha"}}, record.RemovedFinal)
+}
+
+func TestFDDiagnosticLifecycle_ClassifiesEachFinalAddition(t *testing.T) {
+	added := []processharness.FDObservation{{Number: 3, Target: "before"}, {Number: 4, Target: "during"}, {Number: 5, Target: "through"}, {Number: 6, Target: "intermittent"}, {Number: 7, Target: "reused"}}
+	tail := []fdDiagnosticSample{
+		stressDiagnosticSample(1, []processharness.FDObservation{{Number: 4, Target: "during"}, {Number: 5, Target: "through"}, {Number: 6, Target: "intermittent"}, {Number: 7, Target: "reused"}}),
+		stressDiagnosticSample(2, []processharness.FDObservation{{Number: 5, Target: "through"}, {Number: 7, Target: "other"}}),
+		stressDiagnosticSample(3, []processharness.FDObservation{{Number: 5, Target: "through"}, {Number: 6, Target: "intermittent"}, {Number: 7, Target: "reused"}}),
+	}
+
+	lifecycle := classifyFDDiagnosticLifecycle(added, tail)
+
+	require.Equal(t, []fdDiagnosticLifecycle{
+		{Observation: processharness.FDObservation{Number: 3, Target: "before"}, Status: "cleared_before_tail"},
+		{Observation: processharness.FDObservation{Number: 4, Target: "during"}, Status: "cleared_during_tail"},
+		{Observation: processharness.FDObservation{Number: 5, Target: "through"}, Status: "observed_through_tail"},
+		{Observation: processharness.FDObservation{Number: 6, Target: "intermittent"}, Status: "intermittent_or_reused"},
+		{Observation: processharness.FDObservation{Number: 7, Target: "reused"}, Status: "intermittent_or_reused", NumberReused: true},
+	}, lifecycle)
+}
+
+func TestFDDiagnosticFields_DoNotChangeStressEvaluationOrEvidenceJSON(t *testing.T) {
+	resource := stressDashboardResourceFixture(100)
+	beforeEvaluation, err := EvaluateStressResource(resource)
+	require.NoError(t, err)
+	for sampleIndex := range resource.Baseline.Samples {
+		resource.Baseline.Samples[sampleIndex].FDObservations = []processharness.FDObservation{{Number: 9, Target: "diagnostic"}}
+		resource.End.Samples[sampleIndex].FDObservations = []processharness.FDObservation{{Number: 9, Target: "diagnostic"}}
+	}
+	afterEvaluation, err := EvaluateStressResource(resource)
+	require.NoError(t, err)
+	profile := mustStressPRFullProfile(t)
+	evidence := validStressEvidence(t, profile)
+	beforeEvidence, err := json.Marshal(evidence)
+	require.NoError(t, err)
+	for iteration := range evidence.Iterations {
+		for resourceIndex := range evidence.Iterations[iteration].Resources {
+			for sampleIndex := range evidence.Iterations[iteration].Resources[resourceIndex].Baseline.Samples {
+				evidence.Iterations[iteration].Resources[resourceIndex].Baseline.Samples[sampleIndex].FDObservations = []processharness.FDObservation{{Number: 9, Target: "diagnostic"}}
+				evidence.Iterations[iteration].Resources[resourceIndex].End.Samples[sampleIndex].FDObservations = []processharness.FDObservation{{Number: 9, Target: "diagnostic"}}
+			}
+		}
+	}
+	afterEvidence, err := json.Marshal(evidence)
+	require.NoError(t, err)
+
+	require.Equal(t, beforeEvaluation, afterEvaluation)
+	require.Equal(t, string(beforeEvidence), string(afterEvidence))
+}