fix(ingestion): read Python logs in Go backend (#20214)

internal/dao/pipeline_operation_log.go

@@ -37,19 +37,33 @@ var legacyPipelineOperationStatuses = map[string]string{
 	"5": "SCHEDULE",
 }
 
-func normalizePipelineOperationStatuses(statuses []string) []string {
+func expandPipelineOperationStatuses(statuses []string) []string {
 	if len(statuses) == 0 {
 		return statuses
 	}
-	normalized := make([]string, len(statuses))
-	for i, status := range statuses {
+	values := make([]string, 0, len(statuses)*2)
+	seen := make(map[string]struct{}, len(statuses)*2)
+	add := func(status string) {
+		if _, ok := seen[status]; ok {
+			return
+		}
+		seen[status] = struct{}{}
+		values = append(values, status)
+	}
+	for _, status := range statuses {
+		add(status)
 		if canonical, ok := legacyPipelineOperationStatuses[status]; ok {
-			normalized[i] = canonical
-		} else {
-			normalized[i] = status
+			add(canonical)
+			continue
+		}
+		for legacy, canonical := range legacyPipelineOperationStatuses {
+			if canonical == status {
+				add(legacy)
+				break
+			}
 		}
 	}
-	return normalized
+	return values
 }
 
 // PipelineOperationLogDAO data access object for pipeline_operation_log.
@@ -80,7 +94,7 @@ func (dao *PipelineOperationLogDAO) GetDatasetLogsByKBID(ctx context.Context, db
 		query = query.Where("document_id = ?", documentID)
 	}
 	if len(operationStatus) > 0 {
-		query = query.Where("operation_status IN ?", normalizePipelineOperationStatuses(operationStatus))
+		query = query.Where("operation_status IN ?", expandPipelineOperationStatuses(operationStatus))
 	}
 	if createDateFrom != "" {
 		query = query.Where("create_date >= ?", createDateFrom)
@@ -119,7 +133,7 @@ func (dao *PipelineOperationLogDAO) GetDatasetLogsByKBID(ctx context.Context, db
 // documents share a name.
 func (dao *PipelineOperationLogDAO) GetFileLogsByKBID(ctx context.Context, db *gorm.DB, kbID string, page, pageSize int, terms []OrderTerm, keywords, documentID string, operationStatus []string, createDateFrom, createDateTo string) ([]*entity.PipelineOperationLog, int64, error) {
 	query := db.WithContext(ctx).Model(&entity.PipelineOperationLog{}).
-		Where("kb_id = ? AND run_count > 0", kbID)
+		Where("kb_id = ? AND (run_count > 0 OR run_count IS NULL)", kbID)
 
 	if keywords != "" {
 		query = query.Where("LOWER(document_name) LIKE ?", "%"+strings.ToLower(keywords)+"%")
@@ -130,7 +144,7 @@ func (dao *PipelineOperationLogDAO) GetFileLogsByKBID(ctx context.Context, db *g
 	query = query.Where("document_id <> ?", entity.DatasetLogDocumentID)
 
 	if len(operationStatus) > 0 {
-		query = query.Where("operation_status IN ?", operationStatus)
+		query = query.Where("operation_status IN ?", expandPipelineOperationStatuses(operationStatus))
 	}
 	if createDateFrom != "" {
 		query = query.Where("create_date >= ?", createDateFrom)

internal/entity/pipeline.go

@@ -26,8 +26,8 @@ const DatasetLogDocumentID = "graph_raptor_x"
 type PipelineOperationLog struct {
 	ID         string `gorm:"column:id;primaryKey;size:32" json:"id"`
 	DocumentID string `gorm:"column:document_id;size:32;index;uniqueIndex:idx_pipeline_operation_log_document_run,priority:1" json:"document_id"`
-	// RunCount is the monotonically allocated display number for a Go ingestion
-	// run. The Go read path only exposes positive run numbers.
+	// RunCount is the monotonically allocated display number for a numbered
+	// ingestion run.
 	RunCount        *int       `gorm:"column:run_count;uniqueIndex:idx_pipeline_operation_log_document_run,priority:2" json:"run_count,omitempty"`
 	TenantID        string     `gorm:"column:tenant_id;size:32;not null;index" json:"tenant_id"`
 	KbID            string     `gorm:"column:kb_id;size:32;not null;index" json:"kb_id"`

internal/service/dataset/ingestion.go

@@ -135,7 +135,7 @@ func isReadableIngestionLog(log *entity.PipelineOperationLog) bool {
 	if log.DocumentID == entity.DatasetLogDocumentID {
 		return log.RunCount == nil
 	}
-	return log.RunCount != nil && *log.RunCount > 0
+	return log.RunCount == nil || *log.RunCount > 0
 }
 
 func (d *DatasetService) ListIngestionLogs(ctx context.Context, datasetID, userID string, page, pageSize int, terms []dao.OrderTerm, operationStatus []string, createDateFrom, createDateTo, logType, keywords, documentID string) (map[string]interface{}, common.ErrorCode, error) {
@@ -222,7 +222,11 @@ func (d *DatasetService) GetIngestionLog(ctx context.Context, datasetID, userID,
 	if err != nil {
 		return nil, common.CodeServerError, fmt.Errorf("get latest ingestion event: %w", err)
 	}
-	return datasetIngestionLogToMap(log, ingestionEventItem(latestEvents[log.ID])), common.CodeSuccess, nil
+	latestEvent := ingestionEventItem(latestEvents[log.ID])
+	if log.DocumentID == entity.DatasetLogDocumentID {
+		return datasetIngestionLogToMap(log, latestEvent), common.CodeSuccess, nil
+	}
+	return fileIngestionLogToMap(log, latestEvent), common.CodeSuccess, nil
 }
 
 func datasetIngestionLogToMap(log *entity.PipelineOperationLog, latestEvent *service.IngestionEventItem) map[string]interface{} {
@@ -247,6 +251,7 @@ func datasetIngestionLogToMap(log *entity.PipelineOperationLog, latestEvent *ser
 		"update_date":            log.UpdateDate,
 		"latest_ingestion_event": latestEvent,
 	}
+	addIngestionProgressFallback(m, log, latestEvent)
 	if log.PipelineID != nil {
 		m["pipeline_id"] = *log.PipelineID
 	}
@@ -257,7 +262,7 @@ func datasetIngestionLogToMap(log *entity.PipelineOperationLog, latestEvent *ser
 }
 
 func fileIngestionLogToMap(log *entity.PipelineOperationLog, latestEvent *service.IngestionEventItem) map[string]interface{} {
-	return map[string]interface{}{
+	m := map[string]interface{}{
 		"id":                     log.ID,
 		"document_id":            log.DocumentID,
 		"tenant_id":              log.TenantID,
@@ -283,6 +288,14 @@ func fileIngestionLogToMap(log *entity.PipelineOperationLog, latestEvent *servic
 		"update_date":            timePointerValue(log.UpdateDate),
 		"latest_ingestion_event": latestEvent,
 	}
+	addIngestionProgressFallback(m, log, latestEvent)
+	return m
+}
+
+func addIngestionProgressFallback(values map[string]interface{}, log *entity.PipelineOperationLog, latestEvent *service.IngestionEventItem) {
+	if latestEvent == nil && log.ProgressMsg != nil && *log.ProgressMsg != "" {
+		values["progress_msg"] = *log.ProgressMsg
+	}
 }
 
 func ingestionEventItem(event *entity.IngestionTaskLog) *service.IngestionEventItem {

internal/service/dataset/ingestion_messages_test.go

@@ -113,7 +113,7 @@ func TestDatasetIngestionLogUsesEventStream(t *testing.T) {
 	assertLatestEventMap(t, log, 1, "Knowledge compilation completed")
 }
 
-func TestListIngestionLogsExcludesUnnumberedRuns(t *testing.T) {
+func TestListIngestionLogsIncludesPythonFileLogsAndStatusFilters(t *testing.T) {
 	db := setupServiceTestDB(t)
 	pushServiceDB(t, db)
 	if err := db.AutoMigrate(&entity.PipelineOperationLog{}); err != nil {
@@ -121,15 +121,95 @@ func TestListIngestionLogsExcludesUnnumberedRuns(t *testing.T) {
 	}
 	insertCompilationOwnerKB(t, "kb-1", "user-1")
 	insertMessageRun(t, "run-1", "kb-1", "doc-1", entity.TaskStatusDone, 1)
-	insertMessageRun(t, "old-run", "kb-1", "doc-2", entity.TaskStatusDone, 0)
+	insertMessageRun(t, "zero-run", "kb-1", "doc-3", entity.TaskStatusDone, 0)
+	progressMessage := "Python parse completed"
+	if err := db.Create(&entity.PipelineOperationLog{
+		ID: "python-run", DocumentID: "doc-2", TenantID: "user-1", KbID: "kb-1",
+		ParserID: "naive", DocumentName: "doc.txt", DocumentSuffix: ".txt", DocumentType: "text",
+		SourceFrom: "local", TaskType: string(entity.PipelineTaskTypeParse), OperationStatus: "3",
+		ProgressMsg: &progressMessage,
+	}).Error; err != nil {
+		t.Fatalf("insert Python run: %v", err)
+	}
 
-	result, code, err := NewDatasetService().ListIngestionLogs(t.Context(), "kb-1", "user-1", 1, 30, nil, nil, "", "", "file", "", "")
+	result, code, err := NewDatasetService().ListIngestionLogs(t.Context(), "kb-1", "user-1", 1, 30, nil, []string{"DONE"}, "", "", "file", "", "")
 	if err != nil || code != common.CodeSuccess {
 		t.Fatalf("ListIngestionLogs = (%+v, %v, %v), want success", result, code, err)
 	}
 	logs, ok := result["logs"].([]map[string]interface{})
-	if !ok || len(logs) != 1 || logs[0]["id"] != "run-1" {
-		t.Fatalf("logs = %#v, want only numbered run", result["logs"])
+	if !ok || len(logs) != 2 || result["total"] != int64(2) {
+		t.Fatalf("logs = %#v total=%#v, want numbered and Python runs only", result["logs"], result["total"])
+	}
+	byID := make(map[string]map[string]interface{}, len(logs))
+	for _, log := range logs {
+		byID[log["id"].(string)] = log
+	}
+	if byID["python-run"]["operation_status"] != "3" {
+		t.Fatalf("Python operation_status = %#v, want stored status 3", byID["python-run"]["operation_status"])
+	}
+	if byID["python-run"]["progress_msg"] != progressMessage {
+		t.Fatalf("Python progress_msg = %#v, want %q", byID["python-run"]["progress_msg"], progressMessage)
+	}
+	if _, ok := byID["python-run"]["parser_id"]; !ok {
+		t.Fatalf("Python file log = %#v, want file-log fields", byID["python-run"])
+	}
+
+	detail, code, err := NewDatasetService().GetIngestionLog(t.Context(), "kb-1", "user-1", "python-run")
+	if err != nil || code != common.CodeSuccess {
+		t.Fatalf("GetIngestionLog = (%+v, %v, %v), want success", detail, code, err)
+	}
+	if detail["progress_msg"] != progressMessage {
+		t.Fatalf("Python detail progress_msg = %#v, want %q", detail["progress_msg"], progressMessage)
+	}
+	if _, ok := detail["parser_id"]; !ok {
+		t.Fatalf("Python detail = %#v, want file-log fields", detail)
+	}
+
+	messages, code, err := NewDatasetService().ListIngestionMessages(t.Context(), "kb-1", "user-1", "python-run", 200, nil, nil)
+	if err != nil || code != common.CodeSuccess || !messages.Terminal || len(messages.Items) != 0 {
+		t.Fatalf("Python ListIngestionMessages = (%+v, %v, %v), want empty terminal success", messages, code, err)
+	}
+}
+
+func TestPythonDatasetLogIsReadableAndKeepsProgressMessage(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	if err := db.AutoMigrate(&entity.PipelineOperationLog{}); err != nil {
+		t.Fatalf("migrate pipeline operation log: %v", err)
+	}
+	insertCompilationOwnerKB(t, "kb-1", "user-1")
+	progressMessage := "Knowledge compilation completed"
+	if err := db.Create(&entity.PipelineOperationLog{
+		ID: "python-dataset-run", DocumentID: entity.DatasetLogDocumentID, TenantID: "user-1", KbID: "kb-1",
+		ParserID: "knowledge_compile", DocumentName: "Wiki", DocumentType: "dataset", SourceFrom: "knowledgebase",
+		TaskType: "Wiki", OperationStatus: "3", ProgressMsg: &progressMessage,
+	}).Error; err != nil {
+		t.Fatalf("insert Python dataset run: %v", err)
+	}
+
+	result, code, err := NewDatasetService().ListIngestionLogs(t.Context(), "kb-1", "user-1", 1, 30, nil, []string{"DONE"}, "", "", "dataset", "", "")
+	if err != nil || code != common.CodeSuccess {
+		t.Fatalf("ListIngestionLogs = (%+v, %v, %v), want success", result, code, err)
+	}
+	logs, ok := result["logs"].([]map[string]interface{})
+	if !ok || len(logs) != 1 || logs[0]["id"] != "python-dataset-run" {
+		t.Fatalf("dataset logs = %#v, want the Python log", result["logs"])
+	}
+	if logs[0]["operation_status"] != "3" || logs[0]["progress_msg"] != progressMessage {
+		t.Fatalf("Python dataset log fields = %#v, want stored status and progress_msg", logs[0])
+	}
+
+	log, code, err := NewDatasetService().GetIngestionLog(t.Context(), "kb-1", "user-1", "python-dataset-run")
+	if err != nil || code != common.CodeSuccess {
+		t.Fatalf("GetIngestionLog = (%+v, %v, %v), want success", log, code, err)
+	}
+	if log["progress_msg"] != progressMessage {
+		t.Fatalf("detail progress_msg = %#v, want %q", log["progress_msg"], progressMessage)
+	}
+
+	messages, code, err := NewDatasetService().ListIngestionMessages(t.Context(), "kb-1", "user-1", "python-dataset-run", 200, nil, nil)
+	if err != nil || code != common.CodeSuccess || !messages.Terminal || len(messages.Items) != 0 {
+		t.Fatalf("legacy ListIngestionMessages = (%+v, %v, %v), want empty terminal success", messages, code, err)
 	}
 }
 

web/src/pages/dataset/dataset/hooks.test.ts

@@ -254,7 +254,7 @@ describe('useShowLog — Go backend early-log fallback', () => {
       expect.objectContaining({
         document_id: 'doc-1',
         log_type: 'file',
-        orderby: 'run_count',
+        orderby: 'create_time',
         desc: true,
         page_size: 1,
       }),
@@ -264,31 +264,101 @@ describe('useShowLog — Go backend early-log fallback', () => {
     });
   });
 
-  it('does not read the old pipeline log progress field on Go', async () => {
+  it('shows the Python pipeline log progress field on Go', async () => {
     mockIsGo = true;
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     mockList.mockResolvedValue({
       data: {
         data: {
-          logs: [{ document_id: 'doc-1', progress_msg: 'Task is queued...' }],
+          logs: [
+            {
+              id: 'python-run',
+              document_id: 'doc-1',
+              progress_msg: 'Python parse completed',
+            },
+          ],
           total: 1,
         },
       },
     } as any);
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    mockMessages.mockResolvedValue({
+      data: {
+        data: {
+          run_count: 0,
+          items: [],
+          has_more_before: false,
+          has_more_after: false,
+          terminal: true,
+        },
+      },
+    } as any);
 
-    const doc = makeDoc({ ingestion_status: IngestionTaskStatus.CREATED });
+    const doc = makeDoc({ ingestion_status: IngestionTaskStatus.COMPLETED });
     const { result } = renderLogs([doc]);
     act(() => result.current.showLog(doc));
 
-    await waitFor(() => expect(mockList).toHaveBeenCalled());
-    expect(result.current.logInfo.details).toBe('-');
-    expect(result.current.logInfo.status).toBe(RunningStatus.QUEUED);
+    await waitFor(() =>
+      expect(mockMessages).toHaveBeenCalledWith('kb-1', 'python-run', {
+        limit: 200,
+      }),
+    );
+    expect(result.current.logInfo.details).toBe('Python parse completed');
+    expect(result.current.logInfo.events).toBeUndefined();
+    expect(result.current.logInfo.status).toBe(RunningStatus.DONE);
     expect(mockList).toHaveBeenCalledWith(
       'kb-1',
       expect.objectContaining({ log_type: 'file' }),
     );
   });
 
+  it('prefers the selected log event over its stored progress message', async () => {
+    mockIsGo = true;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    mockList.mockResolvedValue({
+      data: {
+        data: {
+          logs: [
+            {
+              id: 'run-1',
+              document_id: 'doc-1',
+              progress_msg: 'old snapshot',
+              latest_ingestion_event: {
+                id: 7,
+                ts: '2026-01-01T00:00:00Z',
+                event_type: 3,
+                component: '',
+                phase: 0,
+                message: 'latest run event',
+              },
+            },
+          ],
+          total: 1,
+        },
+      },
+    } as any);
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    mockMessages.mockResolvedValue({
+      data: {
+        data: {
+          run_count: 0,
+          items: [],
+          has_more_before: false,
+          has_more_after: false,
+          terminal: true,
+        },
+      },
+    } as any);
+
+    const doc = makeDoc({ ingestion_status: IngestionTaskStatus.COMPLETED });
+    const { result } = renderLogs([doc]);
+    act(() => result.current.showLog(doc));
+
+    await waitFor(() =>
+      expect(result.current.logInfo.details).toBe('latest run event'),
+    );
+  });
+
   it('scopes the query to the document instead of a fuzzy name search', async () => {
     mockIsGo = true;
     // eslint-disable-next-line @typescript-eslint/no-explicit-any

web/src/pages/dataset/dataset/hooks.ts

@@ -79,7 +79,7 @@ export const useShowLog = (documents: IDocumentInfo[]) => {
           // share a name.
           document_id: sourceDoc?.id,
           log_type: 'file',
-          orderby: 'run_count',
+          orderby: 'create_time',
           desc: true,
           page_size: 1,
         },
@@ -88,6 +88,7 @@ export const useShowLog = (documents: IDocumentInfo[]) => {
     },
   });
   const logID = documentLog?.logs[0]?.id;
+  const selectedLog = documentLog?.logs[0];
   const {
     data: messages,
     fetchPreviousPage,
@@ -101,18 +102,21 @@ export const useShowLog = (documents: IDocumentInfo[]) => {
 
   const logInfo = useMemo(() => {
     const source = sourceDoc;
+    const details = source
+      ? messages?.items.length
+        ? ''
+        : selectedLog?.latest_ingestion_event?.message ||
+          selectedLog?.progress_msg ||
+          getDocumentProgressMessage({
+            ...source,
+            latest_ingestion_event:
+              latestEvent ?? source.latest_ingestion_event,
+          })
+      : '-';
     let log: ILogInfo = {
       taskId: source?.id,
       fileName: source?.name || '-',
-      details: source
-        ? messages?.items.length
-          ? ''
-          : getDocumentProgressMessage({
-              ...source,
-              latest_ingestion_event:
-                latestEvent ?? source.latest_ingestion_event,
-            })
-        : '-',
+      details,
     };
     if (source) {
       log = {
@@ -127,14 +131,8 @@ export const useShowLog = (documents: IDocumentInfo[]) => {
         // Go derives status from ingestion_status (queued included);
         // Python reads the legacy run field.
         status: getDocumentRunningStatus(source),
-        details: messages?.items.length
-          ? ''
-          : getDocumentProgressMessage({
-              ...source,
-              latest_ingestion_event:
-                latestEvent ?? source.latest_ingestion_event,
-            }),
-        events: messages?.items,
+        details,
+        events: messages?.items.length ? messages.items : undefined,
         loadPreviousEvents: hasPreviousPage
           ? () => fetchPreviousPage()
           : undefined,
@@ -145,6 +143,7 @@ export const useShowLog = (documents: IDocumentInfo[]) => {
     return log;
   }, [
     sourceDoc,
+    selectedLog,
     latestEvent,
     messages,
     fetchPreviousPage,