From 970d2ae672ca8157df2665b1940962fe03f99a5b Mon Sep 17 00:00:00 2001 From: naiba Date: Mon, 20 Jul 2026 22:08:01 +0000 Subject: [PATCH] test(agentcompat): model FD diagnostic lifecycles Co-authored-by: naiba/CloudCode --- .../stress_fd_diagnostic_fixture_test.go | 96 ++++++++ .../stress_fd_diagnostic_model_test.go | 201 +++++++++++++++++ .../stress_fd_diagnostic_tail_support_test.go | 208 ++++++++++++++++++ .../scenario/stress_fd_diagnostic_test.go | 171 ++++++++++++++ 4 files changed, 676 insertions(+) create mode 100644 integration/agentcompat/internal/scenario/stress_fd_diagnostic_fixture_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_fd_diagnostic_model_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_fd_diagnostic_tail_support_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_fd_diagnostic_test.go diff --git a/integration/agentcompat/internal/scenario/stress_fd_diagnostic_fixture_test.go b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_fixture_test.go new file mode 100644 index 00000000..c10a2054 --- /dev/null +++ b/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 +} diff --git a/integration/agentcompat/internal/scenario/stress_fd_diagnostic_model_test.go b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_model_test.go new file mode 100644 index 00000000..4a490a80 --- /dev/null +++ b/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 +} diff --git a/integration/agentcompat/internal/scenario/stress_fd_diagnostic_tail_support_test.go b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_tail_support_test.go new file mode 100644 index 00000000..0b270b96 --- /dev/null +++ b/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} +} diff --git a/integration/agentcompat/internal/scenario/stress_fd_diagnostic_test.go b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_test.go new file mode 100644 index 00000000..c8261dfa --- /dev/null +++ b/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)) +}