From 78154e1e28b92e100f015384ef5d29f9f107d405 Mon Sep 17 00:00:00 2001 From: jay77721 <164177721+jay77721@users.noreply.github.com> Date: Mon, 28 Sep 2026 13:06:54 +0800 Subject: [PATCH] fix(ingestion): read Python logs in Go backend (#20214) --- internal/dao/pipeline_operation_log.go | 34 ++++--- internal/entity/pipeline.go | 4 +- internal/service/dataset/ingestion.go | 19 +++- .../dataset/ingestion_messages_test.go | 90 +++++++++++++++++-- web/src/pages/dataset/dataset/hooks.test.ts | 84 +++++++++++++++-- web/src/pages/dataset/dataset/hooks.ts | 35 ++++---- 6 files changed, 221 insertions(+), 45 deletions(-) diff --git a/internal/dao/pipeline_operation_log.go b/internal/dao/pipeline_operation_log.go index 95e4d127e5..c7275c8b4b 100644 --- a/internal/dao/pipeline_operation_log.go +++ b/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) diff --git a/internal/entity/pipeline.go b/internal/entity/pipeline.go index 52b1998963..923a8835ed 100644 --- a/internal/entity/pipeline.go +++ b/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"` diff --git a/internal/service/dataset/ingestion.go b/internal/service/dataset/ingestion.go index 39b63952a6..323012da99 100644 --- a/internal/service/dataset/ingestion.go +++ b/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 { diff --git a/internal/service/dataset/ingestion_messages_test.go b/internal/service/dataset/ingestion_messages_test.go index 5411ef8bfc..ec54fc2e70 100644 --- a/internal/service/dataset/ingestion_messages_test.go +++ b/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) } } diff --git a/web/src/pages/dataset/dataset/hooks.test.ts b/web/src/pages/dataset/dataset/hooks.test.ts index 897b6481af..dad2bf5e68 100644 --- a/web/src/pages/dataset/dataset/hooks.test.ts +++ b/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 diff --git a/web/src/pages/dataset/dataset/hooks.ts b/web/src/pages/dataset/dataset/hooks.ts index aecd78ca3e..e5f28dd0c4 100644 --- a/web/src/pages/dataset/dataset/hooks.ts +++ b/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,