package impl

import (
	"context"
	"errors"
	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/child"
	"strconv"
	"testing"
	"time"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	app_log "github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

func TestObservationModelPagesAndPersistsCompletePositions(t *testing.T) {
	now := time.Date(2026, time.August, 2, 14, 30, 0, 0, time.UTC)
	checkpointTSO := uint64(now.UnixMilli()) << 18
	lakeResolvedTSO := uint64(now.Add(-2*time.Second).UnixMilli()) << 18
	incremental := workerObservedPipeline(101, true)
	full := workerObservedPipeline(102, false)
	failed := workerObservedPipeline(103, false)
	failed.Status = entity.DataPipelineStatusFailed
	repo := &observationRepo{pages: [][]entity.DataPipeline{{incremental, full, failed}, {}}}
	exports := &observationExportClient{items: map[string]*globalngv1.Export{
		"export-101": {ExportId: "export-101", State: globalngv1.Export_SUCCEEDED},
		"export-102": {ExportId: "export-102", State: globalngv1.Export_SUCCEEDED},
		"export-103": {ExportId: "export-103", State: globalngv1.Export_SUCCEEDED},
	}}
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On(
		"GetBizChangefeed",
		mock.Anything,
		&domain.GetBizChangefeedReq{BizChangefeedID: 101},
	).Return(&domain.GetBizChangefeedResp{BizChangefeed: &domain.BizChangefeed{
		BizChangefeedID:  101,
		OperationStatus:  entity.BizChangefeedOperationStatusNormal,
		ChangefeedStatus: entity.ChangefeedKernelStatusNormal,
		CheckpointTSO:    checkpointTSO,
	}}, nil).Once()
	lakeClient := lake.NewFakeClient()
	lakeClient.SetClock(func() time.Time { return now })
	seedLakeChildrenForObservation(t, lakeClient, 11, 101, lakeResolvedTSO)
	seedLakeChildrenForObservation(t, lakeClient, 12, 102, checkpointTSO)
	seedLakeChildrenForObservation(t, lakeClient, 13, 103, checkpointTSO)

	observer := NewObservationModel(
		repo, exports, changefeeds, lakeClient,
		observationClock{now: now},
		ObservationConfig{PageSize: 3, StaleThreshold: 2 * time.Minute},
	)
	require.NoError(t, observer.SyncObservations(context.Background()))
	require.Equal(t, 2, repo.listCalls)
	require.Len(t, repo.saved, 3)

	savedIncremental := repo.saved[101]
	require.Equal(t, checkpointTSO, savedIncremental.Observation.ChangefeedCheckpointTSO)
	require.Equal(t, lakeResolvedTSO, savedIncremental.Observation.LakeResolvedTSO)
	require.Equal(t, int64(2), savedIncremental.Observation.LatencySeconds)
	require.Equal(t, globalngv1.Export_SUCCEEDED.String(), savedIncremental.Observation.ExportStatus)
	require.Equal(t, string(lake.TaskStatusRunning), savedIncremental.Observation.LakeIntegrationStatus)
	require.Equal(t, "READY", savedIncremental.Observation.LakeDataSourceStatus)
	require.Equal(t, now, *savedIncremental.Observation.LastSucceededAt)
	require.False(t, savedIncremental.Observation.Stale)
	require.Nil(t, savedIncremental.Observation.PollError)
	require.Equal(t, entity.DataPipelineStatusRunning, savedIncremental.Status,
		"Lake full-sync completion must not complete a continuous Pipeline")

	savedFull := repo.saved[102]
	require.Zero(t, savedFull.Observation.ChangefeedCheckpointTSO)
	require.Equal(t, now, *savedFull.Observation.LastSucceededAt)
	require.False(t, savedFull.Observation.Stale)
	require.Equal(t, entity.DataPipelineStatusCompleted, savedFull.Status)

	savedFailed := repo.saved[103]
	require.Equal(t, entity.DataPipelineStatusFailed, savedFailed.Status,
		"a successful Lake observation must not overwrite a terminal Pipeline status")
	require.Equal(t, now, *savedFailed.Observation.LastSucceededAt)
	require.False(t, savedFailed.Observation.Stale)
}

func TestObservationModelDoesNotCompleteFullOnlyPipelineFromFailedOrStaleLakeRead(t *testing.T) {
	now := time.Date(2026, time.August, 2, 14, 50, 0, 0, time.UTC)
	lastSucceededAt := now.Add(-5 * time.Minute)
	lastResolvedTSO := uint64(now.Add(-3*time.Minute).UnixMilli()) << 18

	tests := map[string]struct {
		configureLake func(*lake.FakeClient)
	}{
		"Lake observation fails": {
			configureLake: func(client *lake.FakeClient) {
				client.FailNext(lake.OperationObserveTask, errors.New("test Lake observation failure"))
			},
		},
		"Lake resolved TSO regresses": {
			configureLake: func(client *lake.FakeClient) {
				client.PutTask(lake.Task{
					ID: 221, Status: lake.TaskStatusRunning,
					LastProgress: lake.TiDBProgress{
						LastResolvedTS: lastResolvedTSO - 1, FullSyncCompleted: true,
					},
				})
			},
		},
	}

	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			pipeline := workerObservedPipeline(301, false)
			pipeline.Observation = entity.DataPipelineObservation{
				LastSucceededAt: &lastSucceededAt,
				LakeResolvedTSO: lastResolvedTSO,
				Stale:           true,
			}
			repo := &observationRepo{pages: [][]entity.DataPipeline{{pipeline}}}
			exports := &observationExportClient{items: map[string]*globalngv1.Export{
				"export-301": {ExportId: "export-301", State: globalngv1.Export_SUCCEEDED},
			}}
			lakeClient := lake.NewFakeClient()
			lakeClient.SetClock(func() time.Time { return now })
			lakeClient.PutDataSource(lake.DataSource{ID: 211, Name: "test-ds-301"})
			lakeClient.PutTask(lake.Task{
				ID: 221, Status: lake.TaskStatusRunning,
				LastProgress: lake.TiDBProgress{
					LastResolvedTS: lastResolvedTSO, FullSyncCompleted: true,
				},
			})
			test.configureLake(lakeClient)

			observer := NewObservationModel(
				repo, exports, nil, lakeClient,
				observationClock{now: now},
				ObservationConfig{PageSize: 10, StaleThreshold: 2 * time.Minute},
			)
			require.NoError(t, observer.SyncObservations(context.Background()))

			saved := repo.saved[301]
			require.Equal(t, entity.DataPipelineStatusRunning, saved.Status)
			require.Equal(t, lastSucceededAt, *saved.Observation.LastSucceededAt)
			require.Equal(t, lastResolvedTSO, saved.Observation.LakeResolvedTSO)
			require.True(t, saved.Observation.Stale)
			require.NotNil(t, saved.Observation.PollError)
			require.Equal(t, "failed to observe Lake integration", saved.Observation.LakeErrorMessage)
		})
	}
}

func TestObservationModelIsolatesFailuresAndPreservesLastGoodPositions(t *testing.T) {
	now := time.Date(2026, time.August, 2, 14, 40, 0, 0, time.UTC)
	oldCheckpoint := uint64(now.Add(-10*time.Second).UnixMilli()) << 18
	oldResolved := uint64(now.Add(-12*time.Second).UnixMilli()) << 18
	failed := workerObservedPipeline(201, true)
	failed.Observation = entity.DataPipelineObservation{
		LastSucceededAt:         &[]time.Time{now.Add(-5 * time.Minute)}[0],
		ChangefeedCheckpointTSO: oldCheckpoint,
		LakeResolvedTSO:         oldResolved,
	}
	healthy := workerObservedPipeline(202, false)
	repo := &observationRepo{pages: [][]entity.DataPipeline{{failed, healthy}}}
	exports := &observationExportClient{items: map[string]*globalngv1.Export{
		"export-201": {ExportId: "export-201", State: globalngv1.Export_SUCCEEDED},
		"export-202": {ExportId: "export-202", State: globalngv1.Export_SUCCEEDED},
	}}
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On(
		"GetBizChangefeed",
		mock.Anything,
		&domain.GetBizChangefeedReq{BizChangefeedID: 201},
	).Return(&domain.GetBizChangefeedResp{BizChangefeed: &domain.BizChangefeed{
		BizChangefeedID:  201,
		OperationStatus:  entity.BizChangefeedOperationStatusNormal,
		ChangefeedStatus: entity.ChangefeedKernelStatusNormal,
		CheckpointTSO:    oldCheckpoint - 1,
	}}, nil).Once()
	lakeClient := lake.NewFakeClient()
	lakeClient.SetClock(func() time.Time { return now })
	lakeClient.PutDataSource(lake.DataSource{ID: 111, Name: "test-ds-201"})
	lakeClient.PutDataSource(lake.DataSource{ID: 112, Name: "test-ds-202"})
	seedLakeChildrenForObservation(t, lakeClient, 111, 201, oldResolved)
	seedLakeChildrenForObservation(t, lakeClient, 112, 202, uint64(now.UnixMilli())<<18)
	lakeClient.FailNext(lake.OperationGetDataSource, errors.New("test-secret-key"))

	observer := NewObservationModel(
		repo, exports, changefeeds, lakeClient,
		observationClock{now: now},
		ObservationConfig{PageSize: 10, StaleThreshold: 2 * time.Minute},
	)
	require.NoError(t, observer.SyncObservations(context.Background()))
	require.Len(t, repo.saved, 2, "one Pipeline failure must not stop later Pipelines")

	savedFailed := repo.saved[201]
	require.Equal(t, oldCheckpoint, savedFailed.Observation.ChangefeedCheckpointTSO)
	require.Equal(t, oldResolved, savedFailed.Observation.LakeResolvedTSO)
	require.True(t, savedFailed.Observation.Stale)
	require.NotNil(t, savedFailed.Observation.PollError)
	require.NotContains(t, savedFailed.Observation.PollError.Message, "test-secret-key")
	require.Equal(t, "failed to observe Lake integration", savedFailed.Observation.LakeErrorMessage)

	savedHealthy := repo.saved[202]
	require.False(t, savedHealthy.Observation.Stale)
	require.Equal(t, now, *savedHealthy.Observation.LastSucceededAt)
}

func TestCalculateLatencySeconds(t *testing.T) {
	baseMillis := uint64(1_800_000_000_000)
	require.Equal(t, int64(3), calculateLatencySeconds(
		(baseMillis+3500)<<18,
		baseMillis<<18,
	))
	require.Zero(t, calculateLatencySeconds(baseMillis<<18, (baseMillis+1000)<<18))
	require.Zero(t, calculateLatencySeconds(0, baseMillis<<18))
}

func workerObservedPipeline(id uint64, incremental bool) entity.DataPipeline {
	pipeline := entity.DataPipeline{
		ID: entity.ID(id), TenantID: 10, ClusterID: 20, CreatorID: 30,
		Status: entity.DataPipelineStatusRunning,
		Params: entity.DataPipelineParams{
			SyncMode: entity.DataPipelineSyncModeFull,
			Staging: entity.DataPipelineStagingConfig{S3: entity.DataPipelineS3Config{
				AccessKeyID: "test-access-key",
			}},
			Destination: entity.DataPipelineDestinationConfig{LakeOrgID: "7"},
		},
		ExecutionPlan: entity.DataPipelineExecutionPlan{
			Snapshot: entity.DataPipelineSnapshotPlan{
				Engine: entity.DataPipelineSnapshotEngineDumpling,
				Format: entity.DataPipelineSnapshotFormatCSV,
			},
		},
		Children: entity.DataPipelineChildResources{
			ExportID: "export-" + strconv.FormatUint(id, 10),
		},
	}
	settleChildNames(&pipeline)
	if incremental {
		pipeline.Params.SyncMode = entity.DataPipelineSyncModeFullAndIncremental
		pipeline.ExecutionPlan.Incremental = &entity.DataPipelineIncrementalPlan{
			Engine: entity.DataPipelineIncrementalEngineTiCDC,
			Format: entity.DataPipelineIncrementalFormatCanalJSON,
		}
		pipeline.Children.ChangefeedID = id
	}
	return pipeline
}

type observationRepo struct {
	pages     [][]entity.DataPipeline
	listCalls int
	saved     map[uint64]*entity.DataPipeline
}

func (r *observationRepo) AllocateID() entity.ID                                        { return 0 }
func (r *observationRepo) Create(context.Context, *entity.DataPipeline) bizerr.BizError { return nil }
func (r *observationRepo) Load(context.Context, uint64) (*entity.DataPipeline, bizerr.BizError) {
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
}
func (r *observationRepo) Get(
	context.Context,
	...data_pipeline_repo.FilterOp,
) (*entity.DataPipeline, bizerr.BizError) {
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
}
func (r *observationRepo) List(
	_ context.Context,
	_ ...data_pipeline_repo.FilterOp,
) ([]entity.DataPipeline, bizerr.BizError) {
	call := r.listCalls
	r.listCalls++
	if call >= len(r.pages) {
		return nil, nil
	}
	result := make([]entity.DataPipeline, len(r.pages[call]))
	copy(result, r.pages[call])
	return result, nil
}
func (r *observationRepo) Count(context.Context, ...data_pipeline_repo.FilterOp) (int64, bizerr.BizError) {
	return 0, nil
}
func (r *observationRepo) Save(_ context.Context, pipeline *entity.DataPipeline) bizerr.BizError {
	if r.saved == nil {
		r.saved = make(map[uint64]*entity.DataPipeline)
	}
	copy := *pipeline
	r.saved[pipeline.ID.Uint64()] = &copy
	return nil
}
func (r *observationRepo) Delete(context.Context, uint64) bizerr.BizError { return nil }

type observationExportClient struct {
	items map[string]*globalngv1.Export
}

func (c *observationExportClient) ListExports(
	context.Context,
	*globalngv1.ListExportsRequest,
) (*globalngv1.ListExportsResponse, bizerr.BizError) {
	return nil, nil
}
func (c *observationExportClient) GetExport(
	_ context.Context,
	caller grpcli.ExportCaller,
	req *globalngv1.GetExportRequest,
) (*globalngv1.Export, bizerr.BizError) {
	if err := requireCaller(caller); err != nil {
		return nil, err
	}
	export := c.items[req.GetExportId()]
	if export == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
	}
	return export, nil
}
func (c *observationExportClient) CreateExport(
	context.Context,
	grpcli.ExportCaller,
	*globalngv1.CreateExportRequest,
	uint64,
) (*globalngv1.Export, bizerr.BizError) {
	return nil, nil
}
func (c *observationExportClient) DeleteExport(
	context.Context,
	grpcli.ExportCaller,
	*globalngv1.DeleteExportRequest,
) bizerr.BizError {
	return nil
}
func (c *observationExportClient) CancelExport(
	context.Context,
	grpcli.ExportCaller,
	*globalngv1.CancelExportRequest,
) bizerr.BizError {
	return nil
}

type observationClock struct{ now time.Time }

func (c observationClock) Now() time.Time { return c.now }

// seedLakeChildrenForObservation puts the destination under the deterministic
// names the Pipeline row carries, which is how the observer addresses it.
func seedLakeChildrenForObservation(
	t *testing.T, client *lake.FakeClient, lakeID int32, pipelineID, resolvedTS uint64,
) {
	t.Helper()
	names, err := child.DeterministicResourceNames(pipelineID)
	require.NoError(t, err)
	client.PutDataSource(lake.DataSource{ID: lakeID, Name: names.LakeDataSource})
	client.PutTask(lake.Task{
		ID: lakeID + 10, Name: names.LakeTask, Status: lake.TaskStatusRunning,
		LastProgress: lake.TiDBProgress{
			LastResolvedTS: resolvedTS, FullSyncCompleted: true,
		},
	})
}

// TestObservationFailsPipelineOnTerminalDestinationRun covers the destination
// failure that has no other way out. Lake reports it on the run, not on the
// task: a task whose configuration it cannot satisfy - an unknown warehouse,
// say - stops with a failed run and stays that way. Reading only the task
// status leaves the Pipeline reporting a healthy observation of a destination
// that will never deliver another row.
func TestObservationFailsPipelineOnTerminalDestinationRun(t *testing.T) {
	now := time.Date(2026, 8, 11, 3, 0, 0, 0, time.UTC)
	lakeClient := lake.NewFakeClient()
	lakeClient.SetClock(func() time.Time { return now })
	exports := &observationExportClient{items: map[string]*globalngv1.Export{}}
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)

	const pipelineID = uint64(901)
	pipeline := workerObservedPipeline(pipelineID, false)
	repo := &observationRepo{pages: [][]entity.DataPipeline{{pipeline}, {}}}
	exports.items[pipeline.Children.ExportID] = &globalngv1.Export{
		ExportId: pipeline.Children.ExportID, State: globalngv1.Export_SUCCEEDED,
	}

	names, err := child.DeterministicResourceNames(pipelineID)
	require.NoError(t, err)
	lakeClient.PutDataSource(lake.DataSource{ID: 71, Name: names.LakeDataSource})
	lakeClient.PutTask(lake.Task{
		ID: 81, Name: names.LakeTask, Status: lake.TaskStatusStopped,
	})
	// The real message Lake returned on staging when the configured warehouse
	// did not exist. It is the whole point: without it a user sees only that
	// something failed.
	lakeClient.PutTaskRun(lake.TaskRun{
		ExecuteID: "run-1", TaskID: 81, Status: lake.TaskRunStatusFail,
		Message: "create target database e2e_crm: query error: code: 0, " +
			"message: warehouse tn8upxuaq dp_e2e_wh not found: Bad warehouse",
	})

	observer := NewObservationModel(
		repo, exports, changefeeds, lakeClient,
		observationClock{now: now},
		ObservationConfig{PageSize: 10, StaleThreshold: 2 * time.Minute},
	)
	require.NoError(t, observer.SyncObservations(context.Background()))

	saved := repo.saved[pipelineID]
	require.Equal(t, entity.DataPipelineStatusFailed, saved.Status)
	require.NotNil(t, saved.LastError)
	require.Equal(t, lifecycleErrorChildFailed, saved.LastError.Code)
	require.False(t, saved.LastError.Retryable)
	require.Equal(t,
		"Lake integration run failed: create target database e2e_crm: query error: "+
			"code: 0, message: warehouse tn8upxuaq dp_e2e_wh not found: Bad warehouse",
		saved.Observation.LakeErrorMessage)
	require.Equal(t, saved.Observation.LakeErrorMessage, saved.LastError.Message,
		"the reason the user is shown must be the reason recorded on the row")
	// The answer arrived, so this is not a polling failure and the row is not
	// stale: the reason shown must be the run failure, not a transport error.
	require.Nil(t, saved.Observation.PollError)
	require.False(t, saved.Observation.Stale)
}

// TestObservationKeepsFullSyncCompletionOnceSeen covers the fact that outlives
// the run reporting it. The destination writes full-sync completion on the
// snapshot run, which is the oldest run of a continuous task and eventually
// falls outside the window one observation reads. Taking the answer at face
// value each time would then walk a Pipeline that finished loading months ago
// back to "still loading".
func TestObservationKeepsFullSyncCompletionOnceSeen(t *testing.T) {
	now := time.Date(2026, 8, 11, 5, 0, 0, 0, time.UTC)
	lakeClient := lake.NewFakeClient()
	lakeClient.SetClock(func() time.Time { return now })
	exports := &observationExportClient{items: map[string]*globalngv1.Export{}}
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)

	const pipelineID = uint64(902)
	pipeline := workerObservedPipeline(pipelineID, false)
	repo := &observationRepo{pages: [][]entity.DataPipeline{{pipeline}, {}}}
	exports.items[pipeline.Children.ExportID] = &globalngv1.Export{
		ExportId: pipeline.Children.ExportID, State: globalngv1.Export_SUCCEEDED,
	}
	names, err := child.DeterministicResourceNames(pipelineID)
	require.NoError(t, err)
	lakeClient.PutDataSource(lake.DataSource{ID: 91, Name: names.LakeDataSource})
	lakeClient.PutTask(lake.Task{
		ID: 92, Name: names.LakeTask, Status: lake.TaskStatusRunning,
		LastProgress: lake.TiDBProgress{FullSyncCompleted: true},
	})

	observer := NewObservationModel(
		repo, exports, changefeeds, lakeClient,
		observationClock{now: now},
		ObservationConfig{PageSize: 10, StaleThreshold: 2 * time.Minute},
	)
	require.NoError(t, observer.SyncObservations(context.Background()))
	require.True(t, repo.saved[pipelineID].Observation.LakeFullSyncCompleted)

	// The snapshot run has aged out: the destination now answers without it.
	lakeClient.PutTask(lake.Task{
		ID: 92, Name: names.LakeTask, Status: lake.TaskStatusRunning,
		LastProgress: lake.TiDBProgress{RowsSynced: 40},
	})
	repo.pages = [][]entity.DataPipeline{{*repo.saved[pipelineID]}, {}}
	require.NoError(t, observer.SyncObservations(context.Background()))
	require.True(t, repo.saved[pipelineID].Observation.LakeFullSyncCompleted,
		"a completed full sync must not become incomplete")
}

// The observation worker saves every polled row on every pass. If that write
// moved StatusEnteredAt, a pipeline stuck in creating would report a fresh
// status age forever and the stuck-lifecycle alerts would never fire.
func TestObservingDoesNotMoveTheStatusTimestamp(t *testing.T) {
	entered := time.Date(2026, time.August, 1, 0, 0, 0, 0, time.UTC)
	pipeline := lifecycleTestPipeline(true)
	pipeline.Status = entity.DataPipelineStatusRunning
	pipeline.StatusEnteredAt = &entered

	before := *pipeline.StatusEnteredAt
	setPipelineStatus(context.Background(), pipeline, entity.DataPipelineStatusRunning, entered.Add(72*time.Hour))
	require.Equal(t, before, *pipeline.StatusEnteredAt,
		"re-asserting the same status must not restart the clock")

	setPipelineStatus(context.Background(), pipeline, entity.DataPipelineStatusFailed, entered.Add(72*time.Hour))
	require.Equal(t, entered.Add(72*time.Hour), *pipeline.StatusEnteredAt,
		"a real transition must move it")
	require.Equal(t, entity.DataPipelineStatusFailed, pipeline.Status)
}

// The observation loop runs every interval for every live Pipeline, so the
// value of this log is as much in what it does not write as in what it does.
func TestObservationChangeIsLoggedOnlyWhenSomethingChanged(t *testing.T) {
	t.Parallel()

	pipeline := &entity.DataPipeline{Status: entity.DataPipelineStatusRunning}
	pipeline.ID = 4242
	base := entity.DataPipelineObservation{
		ExportStatus: "SUCCEEDED", ChangefeedStatus: "RUNNING",
		LakeIntegrationStatus: "TaskRunning",
	}

	capture := func(before, after entity.DataPipelineObservation) []observer.LoggedEntry {
		core, logs := observer.New(zap.InfoLevel)
		ctx := app_log.WithLogger(context.Background(), zap.New(core))
		logObservationChange(ctx, pipeline, before, after)
		return logs.All()
	}

	t.Run("an unchanged pass writes nothing", func(t *testing.T) {
		t.Parallel()
		// Only the attempt timestamp moved, which every pass does.
		after := base
		now := time.Now()
		after.LastAttemptAt = &now
		require.Empty(t, capture(base, after))
	})

	t.Run("a child that started erroring is reported with both sides", func(t *testing.T) {
		t.Parallel()
		after := base
		after.LakeIntegrationStatus = "TaskStopped"
		after.LakeErrorMessage = "run failed"
		entries := capture(base, after)
		require.Len(t, entries, 1)
		fields := entries[0].ContextMap()
		require.Equal(t, "TaskRunning -> TaskStopped", fields["lakeIntegrationStatus"])
		require.Equal(t, " -> run failed", fields["lakeError"])
		// pipelineID comes from the per-Pipeline logger the sweep binds, which
		// this test calls past.
		// Untouched children must not appear.
		require.NotContains(t, fields, "exportStatus")
	})

	t.Run("recovery is reported too", func(t *testing.T) {
		t.Parallel()
		before := base
		before.ChangefeedErrorMessage = "unreachable"
		after := base
		entries := capture(before, after)
		require.Len(t, entries, 1)
		require.Equal(t, "unreachable -> ", entries[0].ContextMap()["changefeedError"])
	})
}