[anomalydetection] guard negative flush_interval; preserve log event time

Two fixes from Codex review:

1. NewComponent treated only flush_interval == 0 as needing the default, so
   a negative value (e.g. user types -1s) flowed through to time.NewTicker,
   which panics on non-positive durations. Change the guard to <= 0 to match
   the existing retention guard three lines below.

2. recordingHandle.ObserveLog stamped recorded logs with time.Now().UnixMilli()
   instead of the LogView's own event time. Replayed, buffered, or delayed
   logs were recorded with ingestion time rather than the actual event time,
   breaking chronological reconstruction. Use msg.GetTimestampUnixMilli(),
   falling back to wall clock only when the message reports zero.

Tests (both verified to fail against the unpatched code):
- TestRecorderNegativeFlushIntervalFallsBackToDefault — without the fix,
  the writer goroutine panics in time.NewTicker.
- TestRecordingHandle_PreservesLogTimestamp — without the fix, the
  recorded TimestampMs is wall-clock-now, not the message's event time.

Codex flagged a third issue (parquet_log_reader.go reading only chunks[0]).
Investigated and confirmed it is not reachable in practice: pqarrow's
ReadTable consolidates all row groups into a single Arrow chunk on read in
arrow-go v18.4.0 (verified empirically against a parquet file with 8 row
groups — produces a 1-chunk table). The chunks[0] pattern is therefore
correct for current arrow-go behavior. Not applied.

Author: ellataira

comp/anomalydetection/recorder/impl/recorder.go

@@ -51,7 +51,7 @@ func NewComponent(req Requires) Provides {
 	}
 
 	flushInterval := req.Config.GetDuration("anomaly_detection.recording.flush_interval")
-	if flushInterval == 0 {
+	if flushInterval <= 0 {
 		flushInterval = 60 * time.Second
 	}
 
@@ -224,19 +224,27 @@ func (h *recordingHandle) ObserveMetric(sample observer.MetricView) {
 }
 
 // ObserveLog forwards the log to the inner handle and records it.
+// Uses the LogView's own timestamp (event time) rather than wall-clock now
+// so that delayed, buffered, or replayed logs are recorded with their
+// original time. Wall-clock falls back only when the message reports no
+// timestamp (event time == 0), which would otherwise sort to the epoch.
 func (h *recordingHandle) ObserveLog(msg observer.LogView) {
 	h.inner.ObserveLog(msg)
 
 	content := msg.GetContent()
 	contentCopy := make([]byte, len(content))
 	copy(contentCopy, content)
 
+	timestampMs := msg.GetTimestampUnixMilli()
+	if timestampMs == 0 {
+		timestampMs = time.Now().UnixMilli()
+	}
 	h.recorder.logParquetWriter.WriteLog(
 		h.name,
 		contentCopy,
 		msg.GetStatus(),
 		msg.GetHostname(),
 		msg.GetTags(),
-		time.Now().UnixMilli(),
+		timestampMs,
 	)
 }

comp/anomalydetection/recorder/impl/recorder_test.go

@@ -32,6 +32,22 @@ type fakeHandle struct {
 func (h *fakeHandle) ObserveMetric(_ observerdef.MetricView) { h.metricCalls++ }
 func (h *fakeHandle) ObserveLog(_ observerdef.LogView)       { h.logCalls++ }
 
+// fakeLogView is a minimal observer.LogView with caller-controlled fields,
+// used to assert that ObserveLog reads the message's own timestamp.
+type fakeLogView struct {
+	content     []byte
+	status      string
+	tags        []string
+	hostname    string
+	timestampMs int64
+}
+
+func (f *fakeLogView) GetContent() []byte           { return f.content }
+func (f *fakeLogView) GetStatus() string            { return f.status }
+func (f *fakeLogView) GetTags() []string            { return f.tags }
+func (f *fakeLogView) GetHostname() string          { return f.hostname }
+func (f *fakeLogView) GetTimestampUnixMilli() int64 { return f.timestampMs }
+
 // TestRecorderDisabledByDefault locks in the most important contract: when
 // anomaly_detection.recording.enabled is false (the default), GetHandle returns
 // the inner handle unwrapped and no writer goroutines are started. This is the
@@ -168,3 +184,65 @@ func TestMetricTimestampUnitConsistency(t *testing.T) {
 	require.Equal(t, tsSec*1000, metrics[0].TimestampMs,
 		"MetricData.TimestampMs must be in milliseconds (same unit as LogData.TimestampMs); writer received %d seconds, expected %d ms on read", tsSec, tsSec*1000)
 }
+
+// TestRecorderNegativeFlushIntervalFallsBackToDefault locks in the Codex #1
+// fix: a negative anomaly_detection.recording.flush_interval must not be
+// passed through to time.NewTicker (which panics on non-positive durations).
+// The recorder should fall back to the 60s default instead.
+func TestRecorderNegativeFlushIntervalFallsBackToDefault(t *testing.T) {
+	cfg := config.NewMockWithOverrides(t, map[string]interface{}{
+		"anomaly_detection.recording.enabled":        true,
+		"anomaly_detection.recording.output_dir":     t.TempDir(),
+		"anomaly_detection.recording.flush_interval": -1 * time.Second,
+		"anomaly_detection.recording.retention":      time.Hour,
+	})
+	lc := compdef.NewTestLifecycle(t)
+
+	out := NewComponent(Requires{Lifecycle: lc, Config: cfg})
+	require.NotNil(t, out.Comp)
+
+	impl := out.Comp.(*recorderImpl)
+	require.NotNil(t, impl.metricParquetWriter, "writer should be constructed (recorder not disabled)")
+	require.Equal(t, 60*time.Second, impl.metricParquetWriter.flushInterval,
+		"negative flush_interval must fall back to 60s default")
+	require.NoError(t, lc.Stop(context.Background()))
+}
+
+// TestRecordingHandle_PreservesLogTimestamp locks in the Codex #3 fix:
+// ObserveLog must use the message's own GetTimestampUnixMilli() instead of
+// wall-clock time, so replayed/delayed/buffered logs are recorded with their
+// original event time.
+func TestRecordingHandle_PreservesLogTimestamp(t *testing.T) {
+	tmpDir := t.TempDir()
+	cfg := config.NewMockWithOverrides(t, map[string]interface{}{
+		"anomaly_detection.recording.enabled":        true,
+		"anomaly_detection.recording.output_dir":     tmpDir,
+		"anomaly_detection.recording.flush_interval": time.Hour, // only flush on Stop
+		"anomaly_detection.recording.retention":      time.Hour,
+	})
+	lc := compdef.NewTestLifecycle(t)
+
+	out := NewComponent(Requires{Lifecycle: lc, Config: cfg})
+	require.NotNil(t, out.Comp)
+
+	inner := &fakeHandle{}
+	handle := out.Comp.GetHandle(func(_ string) observerdef.Handle { return inner })("test-source")
+
+	// Pick an event time that is unambiguously not "now".
+	const eventTimeMs = int64(1_700_000_000_000)
+	handle.ObserveLog(&fakeLogView{
+		content:     []byte("hello"),
+		status:      "info",
+		hostname:    "host-x",
+		tags:        []string{"env:test"},
+		timestampMs: eventTimeMs,
+	})
+
+	require.NoError(t, lc.Stop(context.Background()))
+
+	logs, err := out.Comp.ReadAllLogs(tmpDir)
+	require.NoError(t, err)
+	require.Len(t, logs, 1)
+	require.Equal(t, eventTimeMs, logs[0].TimestampMs,
+		"recorded log timestamp must come from LogView.GetTimestampUnixMilli, not wall clock")
+}