fix(logs): refresh symlinkPolicy in Tailer.ReplaceSource

When the launcher reuses an existing tailer for a new source (same scan
key, different source object), it calls ReplaceSource().  Previously the
tailer's symlinkPolicy was frozen at construction time, so swapping a
NoFollow=true source for a NoFollow=false source (or vice-versa) left the
open policy stale — subsequent reopens and rotation checks used the
policy from the original source.

Fix by updating symlinkPolicy from the new source's NoFollow flag inside
ReplaceSource(), and add a unit test that covers all three transitions
(false→true→false).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: vitkyrka

pkg/logs/tailers/file/tailer.go

@@ -479,9 +479,15 @@ func (t *Tailer) recordBytes(n int64) {
 	t.bytesRead.Add(n)
 }
 
-// ReplaceSource replaces the current source
+// ReplaceSource replaces the current source and refreshes open-policy state
+// derived from the source config (e.g. NoFollow).
 func (t *Tailer) ReplaceSource(newSource *sources.LogSource) {
 	t.file.Source.Replace(newSource)
+	if cfg := newSource.Config; cfg != nil && cfg.NoFollow {
+		t.symlinkPolicy = opener.RejectSymlinks
+	} else {
+		t.symlinkPolicy = opener.FollowSymlinks
+	}
 }
 
 // Source gets the source (currently only used for testing)

pkg/logs/tailers/file/tailer_test.go

@@ -680,3 +680,54 @@ func TestNoGoLeakWithNonBlockingStop(t *testing.T) {
 
 	// The deferred goleak.VerifyNone() will detect if goroutine leaked
 }
+
+// TestReplaceSourceRefreshesSymlinkPolicy verifies that ReplaceSource updates
+// the tailer's symlinkPolicy to match the new source's NoFollow flag.
+func TestReplaceSourceRefreshesSymlinkPolicy(t *testing.T) {
+	testDir := t.TempDir()
+	testPath := filepath.Join(testDir, "tailer.log")
+	f, err := os.Create(testPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+
+	info := status.NewInfoRegistry()
+	makeSource := func(noFollow bool) *sources.LogSource {
+		return sources.NewLogSource("", &config.LogsConfig{
+			Type:     config.FileType,
+			Path:     testPath,
+			NoFollow: noFollow,
+		})
+	}
+
+	tailerOptions := &TailerOptions{
+		OutputChan:      make(chan *message.Message, chanSize),
+		File:            NewFile(testPath, makeSource(false), false),
+		SleepDuration:   10 * time.Millisecond,
+		Decoder:         decoder.NewDecoderFromSource(sources.NewReplaceableSource(makeSource(false)), info),
+		Info:            info,
+		CapacityMonitor: metrics.NewNoopPipelineMonitor("").GetCapacityMonitor("", ""),
+		Registry:        auditor.NewMockRegistry(),
+		FileOpener:      opener.NewFileOpener(),
+	}
+
+	tailer := NewTailer(tailerOptions)
+
+	// Initial state: NoFollow=false → FollowSymlinks.
+	if tailer.symlinkPolicy != opener.FollowSymlinks {
+		t.Fatalf("expected FollowSymlinks initially, got %v", tailer.symlinkPolicy)
+	}
+
+	// Replace with NoFollow=true → policy must become RejectSymlinks.
+	tailer.ReplaceSource(makeSource(true))
+	if tailer.symlinkPolicy != opener.RejectSymlinks {
+		t.Fatalf("expected RejectSymlinks after NoFollow=true source, got %v", tailer.symlinkPolicy)
+	}
+
+	// Replace again with NoFollow=false → policy must revert to FollowSymlinks.
+	tailer.ReplaceSource(makeSource(false))
+	if tailer.symlinkPolicy != opener.FollowSymlinks {
+		t.Fatalf("expected FollowSymlinks after NoFollow=false source, got %v", tailer.symlinkPolicy)
+	}
+}