fix(circleci): collect pipelines when the sync policy has no timeAfter

backend/plugins/circleci/tasks/pipeline_collector.go

@@ -20,6 +20,7 @@ package tasks
 import (
 	"encoding/json"
 	"net/http"
+	"time"
 
 	"github.com/apache/devlake/core/errors"
 	"github.com/apache/devlake/core/plugin"
@@ -41,7 +42,10 @@ var CollectPipelinesMeta = plugin.SubTaskMeta{
 func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error {
 	rawDataSubTaskArgs, data := CreateRawDataSubTaskArgs(taskCtx, RAW_PIPELINE_TABLE)
 	logger := taskCtx.GetLogger()
-	timeAfter := rawDataSubTaskArgs.Ctx.TaskContext().SyncPolicy().TimeAfter
+	var timeAfter *time.Time
+	if syncPolicy := rawDataSubTaskArgs.Ctx.TaskContext().SyncPolicy(); syncPolicy != nil {
+		timeAfter = syncPolicy.TimeAfter
+	}
 	logger.Info("collect pipelines")
 	collector, err := api.NewStatefulApiCollectorForFinalizableEntity(api.FinalizableApiCollectorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
@@ -53,25 +57,7 @@ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error {
 				UrlTemplate: "/v2/project/{{ .Params.ProjectSlug }}/pipeline",
 				Query:       BuildQueryParamsWithPageToken,
 				ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
-					data := CircleciPageTokenResp[[]json.RawMessage]{}
-					err := api.UnmarshalResponse(res, &data)
-
-					if err != nil {
-						return nil, err
-					}
-					filteredItems := []json.RawMessage{}
-					for _, item := range data.Items {
-						pipelineCreatedAt, err := extractCreatedAt(item)
-
-						if err != nil {
-							return nil, err
-						}
-						if pipelineCreatedAt.Before(*timeAfter) {
-							return filteredItems, api.ErrFinishCollect
-						}
-						filteredItems = append(filteredItems, item)
-					}
-					return filteredItems, nil
+					return parsePipelines(res, timeAfter)
 				},
 			},
 			GetCreated: extractCreatedAt,
@@ -83,3 +69,30 @@ func CollectPipelines(taskCtx plugin.SubTaskContext) errors.Error {
 	}
 	return collector.Execute()
 }
+
+// parsePipelines returns the pipelines of a page. CircleCI lists pipelines
+// newest first, so the collection ends at the first pipeline created before
+// timeAfter. timeAfter is nil when the sync policy has no time range (e.g. an
+// advanced mode blueprint), and then every pipeline is kept.
+func parsePipelines(res *http.Response, timeAfter *time.Time) ([]json.RawMessage, errors.Error) {
+	data := CircleciPageTokenResp[[]json.RawMessage]{}
+	err := api.UnmarshalResponse(res, &data)
+	if err != nil {
+		return nil, err
+	}
+	if timeAfter == nil {
+		return data.Items, nil
+	}
+	filteredItems := []json.RawMessage{}
+	for _, item := range data.Items {
+		pipelineCreatedAt, err := extractCreatedAt(item)
+		if err != nil {
+			return nil, err
+		}
+		if pipelineCreatedAt.Before(*timeAfter) {
+			return filteredItems, api.ErrFinishCollect
+		}
+		filteredItems = append(filteredItems, item)
+	}
+	return filteredItems, nil
+}

backend/plugins/circleci/tasks/pipeline_collector_test.go

@@ -0,0 +1,63 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/apache/devlake/helpers/pluginhelper/api"
+	"github.com/stretchr/testify/assert"
+)
+
+const pipelinesPage = `{
+	"items": [
+		{"id": "p3", "created_at": "2026-03-01T00:00:00Z"},
+		{"id": "p2", "created_at": "2026-02-01T00:00:00Z"},
+		{"id": "p1", "created_at": "2026-01-01T00:00:00Z"}
+	],
+	"next_page_token": "token"
+}`
+
+func makePipelinesResponse() *http.Response {
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       io.NopCloser(bytes.NewBufferString(pipelinesPage)),
+		Request:    &http.Request{},
+	}
+}
+
+func TestParsePipelinesWithoutTimeAfter(t *testing.T) {
+	// An advanced mode blueprint has no timeAfter: keep every pipeline
+	// instead of dereferencing a nil time.
+	items, err := parsePipelines(makePipelinesResponse(), nil)
+	assert.Nil(t, err)
+	assert.Len(t, items, 3)
+}
+
+func TestParsePipelinesStopsBeforeTimeAfter(t *testing.T) {
+	timeAfter := time.Date(2026, 1, 15, 0, 0, 0, 0, time.UTC)
+	items, err := parsePipelines(makePipelinesResponse(), &timeAfter)
+	assert.ErrorIs(t, err, api.ErrFinishCollect)
+	assert.Len(t, items, 2)
+	assert.Contains(t, string(items[0]), `"p3"`)
+	assert.Contains(t, string(items[1]), `"p2"`)
+}