diff --git a/internal/application/repository/knowledge_stage_repo.go b/internal/application/repository/knowledge_stage_repo.go new file mode 100644 index 00000000..92bad002 --- /dev/null +++ b/internal/application/repository/knowledge_stage_repo.go @@ -0,0 +1,110 @@ +package repository + +import ( + "context" + "time" + + "github.com/Tencent/WeKnora/internal/types" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// KnowledgeStageRepository persists per-stage progress for a Knowledge. +// +// The repo is intentionally narrow: only the operations the tracker +// service needs (upsert, list, mark all skipped). We don't expose a +// generic Update — every state transition goes through one of the +// purpose-built methods so the audit trail is consistent. +type KnowledgeStageRepository interface { + // Upsert inserts or updates a stage row by (knowledge_id, stage). + // On conflict, ALL provided columns are overwritten and the + // attempt counter is incremented. The caller fills in the row + // fields it wants to set and leaves the rest at their zero values. + Upsert(ctx context.Context, row *types.KnowledgeProcessingStage) error + // ListByKnowledge returns every stage row for the given knowledge, + // ordered by the canonical stage list. Missing stages are NOT + // synthesized here — the API layer decides whether to fill them + // with "pending" placeholders. + ListByKnowledge(ctx context.Context, knowledgeID string) ([]types.KnowledgeProcessingStage, error) + // MarkPendingAll resets every stage row for a knowledge to + // "pending" with a fresh attempt number. Called at the start of + // re-parsing so a previous run's "failed" badge doesn't linger + // while the retry is in flight. + MarkPendingAll(ctx context.Context, knowledgeID string) error +} + +type knowledgeStageRepository struct { + db *gorm.DB +} + +// NewKnowledgeStageRepository wires the GORM-backed implementation. +func NewKnowledgeStageRepository(db *gorm.DB) KnowledgeStageRepository { + return &knowledgeStageRepository{db: db} +} + +func (r *knowledgeStageRepository) Upsert(ctx context.Context, row *types.KnowledgeProcessingStage) error { + if row == nil || row.KnowledgeID == "" || row.Stage == "" { + return nil + } + now := time.Now() + row.UpdatedAt = now + if row.CreatedAt.IsZero() { + row.CreatedAt = now + } + if row.Attempt == 0 { + row.Attempt = 1 + } + // ON CONFLICT (knowledge_id, stage) bumps attempt and overwrites + // state. We must also explicitly null-out finished_at / duration_ms + // when the new state is "running" — otherwise stale completion + // timestamps from a previous attempt leak into the timeline. + return r.db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "knowledge_id"}, {Name: "stage"}}, + DoUpdates: clause.Assignments(map[string]interface{}{ + "status": row.Status, + "started_at": row.StartedAt, + "finished_at": row.FinishedAt, + "duration_ms": row.DurationMs, + "error_code": row.ErrorCode, + "error_message": row.ErrorMessage, + "error_detail": row.ErrorDetail, + "attempt": gorm.Expr("knowledge_processing_stages.attempt + ?", 1), + "updated_at": now, + }), + }).Create(row).Error +} + +func (r *knowledgeStageRepository) ListByKnowledge(ctx context.Context, knowledgeID string) ([]types.KnowledgeProcessingStage, error) { + if knowledgeID == "" { + return nil, nil + } + var rows []types.KnowledgeProcessingStage + err := r.db.WithContext(ctx). + Where("knowledge_id = ?", knowledgeID). + Find(&rows).Error + return rows, err +} + +func (r *knowledgeStageRepository) MarkPendingAll(ctx context.Context, knowledgeID string) error { + if knowledgeID == "" { + return nil + } + now := time.Now() + // Reset every existing stage. We don't pre-create rows for stages + // that haven't been touched yet — those will be added lazily by + // the tracker on first Begin call. This keeps the table small for + // knowledge that gets created but never parsed. + return r.db.WithContext(ctx).Model(&types.KnowledgeProcessingStage{}). + Where("knowledge_id = ?", knowledgeID). + Updates(map[string]interface{}{ + "status": types.ProcessingStagePending, + "started_at": nil, + "finished_at": nil, + "duration_ms": 0, + "error_code": "", + "error_message": "", + "error_detail": "", + "attempt": gorm.Expr("attempt + ?", 1), + "updated_at": now, + }).Error +} diff --git a/internal/application/service/knowledge.go b/internal/application/service/knowledge.go index 8d372482..2e56a7e6 100644 --- a/internal/application/service/knowledge.go +++ b/internal/application/service/knowledge.go @@ -64,6 +64,12 @@ type knowledgeService struct { memFAQRunningImport sync.Map // kbID -> *runningFAQImportInfo wikiRepo interfaces.WikiPageRepository wikiService interfaces.WikiPageService + + // stageTracker records per-stage progress for the parsing pipeline. + // Best-effort: a nil tracker (e.g. test harness) is safely handled + // because every call site goes through the StageTracker interface + // which has a no-op fallback. See knowledge_stage_tracker.go. + stageTracker StageTracker } const ( @@ -96,6 +102,7 @@ func NewKnowledgeService( wikiRepo interfaces.WikiPageRepository, wikiService interfaces.WikiPageService, taskPendingRepo interfaces.TaskPendingOpsRepository, + stageTracker StageTracker, ) (interfaces.KnowledgeService, error) { return &knowledgeService{ config: config, @@ -120,9 +127,20 @@ func NewKnowledgeService( wikiRepo: wikiRepo, wikiService: wikiService, taskPendingRepo: taskPendingRepo, + stageTracker: stageTracker, }, nil } +// stage returns a usable StageTracker — falls back to a no-op when the +// service was constructed without one (test harness, lite mode w/o repo). +// All pipeline call sites go through this so they never need a nil check. +func (s *knowledgeService) stage() StageTracker { + if s.stageTracker == nil { + return noopStageTracker{} + } + return s.stageTracker +} + // getParserEngineOverridesFromContext returns parser engine overrides from tenant in context (e.g. MinerU endpoint, API key). // Used when building document ReadRequest so UI-configured values take precedence over env. func (s *knowledgeService) getParserEngineOverridesFromContext(ctx context.Context) map[string]string { diff --git a/internal/application/service/knowledge_post_process.go b/internal/application/service/knowledge_post_process.go index 8ce5106f..9cc5d832 100644 --- a/internal/application/service/knowledge_post_process.go +++ b/internal/application/service/knowledge_post_process.go @@ -17,12 +17,13 @@ import ( // KnowledgePostProcessService acts as an orchestrator for all post-processing tasks // after a document has been parsed and split into chunks (including multimodal OCR/Caption). type KnowledgePostProcessService struct { - knowledgeRepo interfaces.KnowledgeRepository - kbService interfaces.KnowledgeBaseService - chunkService interfaces.ChunkService - taskEnqueuer interfaces.TaskEnqueuer - pendingRepo interfaces.TaskPendingOpsRepository - redisClient *redis.Client + knowledgeRepo interfaces.KnowledgeRepository + kbService interfaces.KnowledgeBaseService + chunkService interfaces.ChunkService + taskEnqueuer interfaces.TaskEnqueuer + pendingRepo interfaces.TaskPendingOpsRepository + redisClient *redis.Client + stageTracker StageTracker } func NewKnowledgePostProcessService( @@ -32,6 +33,7 @@ func NewKnowledgePostProcessService( taskEnqueuer interfaces.TaskEnqueuer, pendingRepo interfaces.TaskPendingOpsRepository, redisClient *redis.Client, + stageTracker StageTracker, ) interfaces.TaskHandler { return &KnowledgePostProcessService{ knowledgeRepo: knowledgeRepo, @@ -40,9 +42,17 @@ func NewKnowledgePostProcessService( taskEnqueuer: taskEnqueuer, pendingRepo: pendingRepo, redisClient: redisClient, + stageTracker: stageTracker, } } +func (s *KnowledgePostProcessService) tracker() StageTracker { + if s.stageTracker == nil { + return noopStageTracker{} + } + return s.stageTracker +} + // Handle implements asynq handler for TypeKnowledgePostProcess. func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Task) error { var payload types.KnowledgePostProcessPayload @@ -57,6 +67,15 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta ctx = context.WithValue(ctx, types.LanguageContextKey, payload.Language) } + s.tracker().Begin(ctx, payload.KnowledgeID, types.ProcessingStagePostProcess) + + // Track Multimodal completion: by the time we reach post-process, + // any per-image work has either finished or been counted by + // finalize-on-last-attempt (see image_multimodal.go). If the parent + // has reached this stage at all, multimodal effectively succeeded + // from the parent's perspective even if individual images failed. + s.tracker().Done(ctx, payload.KnowledgeID, types.ProcessingStageMultimodal) + // 1. Fetch Knowledge and KB knowledge, err := s.knowledgeRepo.GetKnowledgeByIDOnly(ctx, payload.KnowledgeID) if err != nil { @@ -132,6 +151,7 @@ func (s *KnowledgePostProcessService) Handle(ctx context.Context, task *asynq.Ta EnqueueWikiIngest(ctx, s.taskEnqueuer, s.pendingRepo, payload.TenantID, payload.KnowledgeBaseID, payload.KnowledgeID) logger.Infof(ctx, "[KnowledgePostProcess] Enqueued wiki ingest task for %s", payload.KnowledgeID) } + s.tracker().Done(ctx, payload.KnowledgeID, types.ProcessingStagePostProcess) return nil } diff --git a/internal/application/service/knowledge_process.go b/internal/application/service/knowledge_process.go index f78464ce..67962338 100644 --- a/internal/application/service/knowledge_process.go +++ b/internal/application/service/knowledge_process.go @@ -466,19 +466,24 @@ func (s *knowledgeService) processChunks(ctx context.Context, // Chunks are needed for wiki generation, graph extraction, and summary generation // even when vector/keyword indexing is disabled. span.AddEvent("create chunks") + s.stage().Begin(ctx, knowledge.ID, types.ProcessingStageChunking) if err := s.chunkService.CreateChunks(ctx, insertChunks); err != nil { knowledge.ParseStatus = types.ParseStatusFailed knowledge.ErrorMessage = err.Error() knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) span.RecordError(err) + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageChunking, + werrors.ErrCodeChunkingFailed, "create chunks failed", err) return } + s.stage().Done(ctx, knowledge.ID, types.ProcessingStageChunking) // Create index information and perform vector indexing — only when vector/keyword is enabled. // Chunks are ALWAYS saved to DB (above) because wiki and graph need them even without vector indexing. var totalStorageSize int64 if kb.NeedsEmbeddingModel() && embeddingModel != nil { + s.stage().Begin(ctx, knowledge.ID, types.ProcessingStageEmbedding) // Create index information — only for child/flat chunks, NOT parent chunks. // Parent chunks are stored for context retrieval but do not need vector embeddings. // Prepend the document title to improve semantic alignment between @@ -560,9 +565,18 @@ func (s *knowledgeService) processChunks(ctx context.Context, logger.Errorf(ctx, "Delete index failed: %v", err) } span.RecordError(err) + // Map vector store / embedding rate-limit errors to a + // stable code so the UI can offer "retry later" hints. + code := werrors.ErrCodeVectorStoreWriteFailed + if isLikelyRateLimitError(err) { + code = werrors.ErrCodeEmbeddingRateLimit + } + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageEmbedding, + code, "batch index failed", err) return } logger.GetLogger(ctx).Infof("processChunks batch index successfully, with %d index", len(indexInfoList)) + s.stage().Done(ctx, knowledge.ID, types.ProcessingStageEmbedding) // Final check before marking as completed - if deleted during processing, don't update status if s.isKnowledgeDeleting(ctx, knowledge.TenantID, knowledge.ID) { @@ -579,6 +593,7 @@ func (s *knowledgeService) processChunks(ctx context.Context, } } else { logger.Infof(ctx, "Vector/keyword indexing disabled for KB %s, skipping BatchIndex", kb.ID) + s.stage().Skip(ctx, knowledge.ID, types.ProcessingStageEmbedding) } // Check if this document has extracted images that will be processed asynchronously @@ -602,8 +617,10 @@ func (s *knowledgeService) processChunks(ctx context.Context, // Enqueue multimodal tasks for images (async, non-blocking) if options.EnableMultimodel && len(options.StoredImages) > 0 { + s.stage().Begin(ctx, knowledge.ID, types.ProcessingStageMultimodal) s.enqueueImageMultimodalTasks(ctx, knowledge, kb, options.StoredImages, chunks, options.Metadata) } else { + s.stage().Skip(ctx, knowledge.ID, types.ProcessingStageMultimodal) // If there are no multimodal tasks, enqueue the post process task immediately lang, _ := types.LanguageFromContext(ctx) postProcessPayload := types.KnowledgePostProcessPayload{ @@ -2283,6 +2300,12 @@ func (s *knowledgeService) convert( knowledge *types.Knowledge, isLastRetry bool, ) (*types.ReadResult, error) { + // Stage tracking: docreader. Mark the stage as running here so the + // timeline reflects "DocReader" the moment a worker picks the task + // up — before that, the stage stays "pending" from the initial + // upload. Failure/skip transitions are emitted at the specific + // failure points below; success is emitted at the bottom. + s.stage().Begin(ctx, knowledge.ID, types.ProcessingStageDocReader) isURL := payload.URL != "" fileType := payload.FileType overrides := s.getParserEngineOverridesFromContext(ctx) @@ -2294,6 +2317,8 @@ func (s *knowledgeService) convert( knowledge.ErrorMessage = "URL is not allowed for security reasons" knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageDocReader, + werrors.ErrCodeDocReaderParseFailed, "URL rejected for security reasons", err) return nil, nil } } @@ -2314,6 +2339,8 @@ func (s *knowledgeService) convert( knowledge.ErrorMessage = "Document parsing service is not configured. Please use text/paragraph import or set DOCREADER_ADDR." knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageDocReader, + werrors.ErrCodeDocReaderUnavailable, knowledge.ErrorMessage, nil) return nil, nil } @@ -2328,11 +2355,15 @@ func (s *knowledgeService) convert( if !isURL { fileReader, err := s.resolveFileServiceForPath(ctx, kb, payload.FilePath).GetFile(ctx, payload.FilePath) if err != nil { + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageDocReader, + werrors.ErrCodeDocReaderParseFailed, "failed to get file", err) return s.failKnowledge(ctx, knowledge, isLastRetry, "failed to get file: %v", err) } defer fileReader.Close() contentBytes, err := io.ReadAll(fileReader) if err != nil { + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageDocReader, + werrors.ErrCodeDocReaderParseFailed, "failed to read file", err) return s.failKnowledge(ctx, knowledge, isLastRetry, "failed to read file: %v", err) } req.FileContent = contentBytes @@ -2342,6 +2373,15 @@ func (s *knowledgeService) convert( result, err := s.callDocReaderWithTimeout(ctx, reader, req) if err != nil { + // Distinguish DocReader timeout (a knowable user-facing + // failure) from generic read errors so the UI can suggest + // "split this large file" specifically when relevant. + code := werrors.ErrCodeDocReaderParseFailed + if errors.Is(err, context.DeadlineExceeded) || strings.Contains(err.Error(), "docreader call timeout") { + code = werrors.ErrCodeDocReaderTimeout + } + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageDocReader, + code, "document read failed", err) return s.failKnowledge(ctx, knowledge, isLastRetry, "document read failed: %v", err) } if result.Error != "" { @@ -2351,8 +2391,11 @@ func (s *knowledgeService) convert( knowledge.ErrorMessage = result.Error knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) + s.stage().Fail(ctx, knowledge.ID, types.ProcessingStageDocReader, + werrors.ErrCodeDocReaderParseFailed, result.Error, nil) return nil, nil } + s.stage().Done(ctx, knowledge.ID, types.ProcessingStageDocReader) return result, nil } @@ -2392,6 +2435,24 @@ func (s *knowledgeService) callDocReaderWithTimeout( return result, nil } +// isLikelyRateLimitError performs a fuzzy classification of an error as a +// rate-limit / quota / backpressure failure. We only need a hint — the +// caller maps to one of two error_codes so the UI can offer "retry later" +// vs. "fix configuration" advice. False positives are harmless (the +// detail is preserved in error_detail anyway). +func isLikelyRateLimitError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + for _, needle := range []string{"rate limit", "ratelimit", "429", "too many requests", "quota"} { + if strings.Contains(msg, needle) { + return true + } + } + return false +} + // Returns nil when the required service is unavailable. func (s *knowledgeService) resolveDocReader(ctx context.Context, engine, fileType string, isURL bool, overrides map[string]string) interfaces.DocReader { switch engine { diff --git a/internal/application/service/knowledge_stage_tracker.go b/internal/application/service/knowledge_stage_tracker.go new file mode 100644 index 00000000..1d3e14f9 --- /dev/null +++ b/internal/application/service/knowledge_stage_tracker.go @@ -0,0 +1,156 @@ +// Package service: stage tracker. +// +// StageTracker is the thin facade every pipeline call site uses to record +// progress. Its three operations (Begin, Done, Fail) cover the lifecycle +// of every stage; everything else (ordering, finalize semantics) is the +// caller's responsibility — the tracker just persists what it's told. +// +// All operations are best-effort: a DB error is logged and swallowed so +// the parsing pipeline itself never breaks because tracker bookkeeping +// failed. The Knowledge.parse_status column remains the authoritative +// source of truth for whether a document is done. +package service + +import ( + "context" + "strings" + "time" + + "github.com/Tencent/WeKnora/internal/application/repository" + "github.com/Tencent/WeKnora/internal/logger" + "github.com/Tencent/WeKnora/internal/types" +) + +// StageTracker exposes the trio of write operations the pipeline needs. +// Kept as an interface so unit tests can swap in a no-op implementation +// without spinning up a database. +type StageTracker interface { + Begin(ctx context.Context, knowledgeID, stage string) + Done(ctx context.Context, knowledgeID, stage string) + Fail(ctx context.Context, knowledgeID, stage, errorCode, errorMessage string, errorDetail error) + Skip(ctx context.Context, knowledgeID, stage string) +} + +type stageTracker struct { + repo repository.KnowledgeStageRepository + // in-memory map of (knowledge_id|stage) → start time so we can + // compute duration_ms when Done/Fail fires without an extra DB + // read. Best-effort: a missed Begin (e.g. process restart between + // Begin and Done) just yields duration_ms=0, which the UI renders + // as "—" instead of a wrong number. + starts map[string]time.Time +} + +// NewStageTracker constructs a tracker backed by the given repo. +func NewStageTracker(repo repository.KnowledgeStageRepository) StageTracker { + if repo == nil { + return noopStageTracker{} + } + return &stageTracker{ + repo: repo, + starts: make(map[string]time.Time), + } +} + +func startKey(kid, stage string) string { + return kid + "|" + stage +} + +func (t *stageTracker) Begin(ctx context.Context, knowledgeID, stage string) { + if knowledgeID == "" || stage == "" { + return + } + now := time.Now() + t.starts[startKey(knowledgeID, stage)] = now + row := &types.KnowledgeProcessingStage{ + KnowledgeID: knowledgeID, + Stage: stage, + Status: types.ProcessingStageRunning, + StartedAt: &now, + } + if err := t.repo.Upsert(ctx, row); err != nil { + logger.Warnf(ctx, "[StageTracker] Begin failed kid=%s stage=%s: %v", knowledgeID, stage, err) + } +} + +func (t *stageTracker) Done(ctx context.Context, knowledgeID, stage string) { + if knowledgeID == "" || stage == "" { + return + } + now := time.Now() + var dur int64 + if start, ok := t.starts[startKey(knowledgeID, stage)]; ok { + dur = now.Sub(start).Milliseconds() + delete(t.starts, startKey(knowledgeID, stage)) + } + row := &types.KnowledgeProcessingStage{ + KnowledgeID: knowledgeID, + Stage: stage, + Status: types.ProcessingStageDone, + FinishedAt: &now, + DurationMs: dur, + } + if err := t.repo.Upsert(ctx, row); err != nil { + logger.Warnf(ctx, "[StageTracker] Done failed kid=%s stage=%s: %v", knowledgeID, stage, err) + } +} + +func (t *stageTracker) Fail(ctx context.Context, knowledgeID, stage, errorCode, errorMessage string, errorDetail error) { + if knowledgeID == "" || stage == "" { + return + } + now := time.Now() + var dur int64 + if start, ok := t.starts[startKey(knowledgeID, stage)]; ok { + dur = now.Sub(start).Milliseconds() + delete(t.starts, startKey(knowledgeID, stage)) + } + detail := "" + if errorDetail != nil { + detail = errorDetail.Error() + if len(detail) > 8192 { + detail = detail[:8192] + } + } + if len(errorMessage) > 1024 { + errorMessage = errorMessage[:1024] + } + row := &types.KnowledgeProcessingStage{ + KnowledgeID: knowledgeID, + Stage: stage, + Status: types.ProcessingStageFailed, + FinishedAt: &now, + DurationMs: dur, + ErrorCode: strings.TrimSpace(errorCode), + ErrorMessage: errorMessage, + ErrorDetail: detail, + } + if err := t.repo.Upsert(ctx, row); err != nil { + logger.Warnf(ctx, "[StageTracker] Fail failed kid=%s stage=%s: %v", knowledgeID, stage, err) + } +} + +func (t *stageTracker) Skip(ctx context.Context, knowledgeID, stage string) { + if knowledgeID == "" || stage == "" { + return + } + now := time.Now() + row := &types.KnowledgeProcessingStage{ + KnowledgeID: knowledgeID, + Stage: stage, + Status: types.ProcessingStageSkipped, + FinishedAt: &now, + } + if err := t.repo.Upsert(ctx, row); err != nil { + logger.Warnf(ctx, "[StageTracker] Skip failed kid=%s stage=%s: %v", knowledgeID, stage, err) + } +} + +// noopStageTracker is returned when the repo is nil — keeps call sites +// trivial (no `if t != nil` guards) at the cost of one struct allocation. +type noopStageTracker struct{} + +func (noopStageTracker) Begin(_ context.Context, _, _ string) {} +func (noopStageTracker) Done(_ context.Context, _, _ string) {} +func (noopStageTracker) Fail(_ context.Context, _, _, _, _ string, _ error) {} +func (noopStageTracker) Skip(_ context.Context, _, _ string) {} diff --git a/internal/container/container.go b/internal/container/container.go index 8559262c..3deef841 100644 --- a/internal/container/container.go +++ b/internal/container/container.go @@ -140,6 +140,7 @@ func BuildContainer(container *dig.Container) *dig.Container { must(container.Provide(repository.NewAuditLogRepository)) must(container.Provide(repository.NewKnowledgeBaseRepository)) must(container.Provide(repository.NewKnowledgeRepository)) + must(container.Provide(repository.NewKnowledgeStageRepository)) must(container.Provide(repository.NewChunkRepository)) must(container.Provide(repository.NewKnowledgeTagRepository)) must(container.Provide(repository.NewSessionRepository)) @@ -182,6 +183,7 @@ func BuildContainer(container *dig.Container) *dig.Container { must(container.Provide(service.NewKBShareService)) // KBShareService must be registered before KnowledgeService and KnowledgeTagService must(container.Provide(service.NewAgentShareService)) must(container.Provide(service.NewKnowledgeService)) + must(container.Provide(service.NewStageTracker)) must(container.Provide(service.NewChunkService)) must(container.Provide(service.NewKnowledgeTagService)) must(container.Provide(embedding.NewBatchEmbedder)) diff --git a/internal/errors/parse_error_codes.go b/internal/errors/parse_error_codes.go new file mode 100644 index 00000000..e698470f --- /dev/null +++ b/internal/errors/parse_error_codes.go @@ -0,0 +1,63 @@ +// Package errors: parse stage error codes. +// +// These constants are the stable wire format the frontend uses to map a +// failure to a localized "what to do about it" message. Adding a new code +// is non-breaking; renaming an existing one requires a coordinated +// frontend release because the i18n keys are looked up by code. +// +// Codes are SCREAMING_SNAKE so they survive JSON case transforms and look +// distinct from Go identifiers in logs/dashboards. +package errors + +const ( + // ErrCodeDocReaderTimeout — DocReader RPC exceeded + // WEKNORA_DOCREADER_CALL_TIMEOUT (default 30m). Suggest splitting + // large files or checking docreader load. + ErrCodeDocReaderTimeout = "DOCREADER_TIMEOUT" + + // ErrCodeDocReaderUnavailable — no DocReader configured for the + // requested file type / engine, or the service refused connection. + ErrCodeDocReaderUnavailable = "DOCREADER_UNAVAILABLE" + + // ErrCodeDocReaderParseFailed — DocReader returned an explicit parse + // error (encoding, corrupted file, OCR engine crash, ...). + ErrCodeDocReaderParseFailed = "DOCREADER_PARSE_FAILED" + + // ErrCodeChunkingFailed — text chunking step itself failed (rare; + // usually only on extreme-size inputs). + ErrCodeChunkingFailed = "CHUNKING_FAILED" + + // ErrCodeEmbeddingRateLimit — embedding provider returned 429 or + // equivalent. Suggest retrying later or scaling out. + ErrCodeEmbeddingRateLimit = "EMBEDDING_RATE_LIMIT" + + // ErrCodeEmbeddingProviderFail — non-rate-limit embedding error + // (auth, model not found, bad input). Usually permanent without + // config changes. + ErrCodeEmbeddingProviderFail = "EMBEDDING_PROVIDER_FAIL" + + // ErrCodeVectorStoreWriteFailed — chunks embedded fine but the + // vector DB write failed (quota, connectivity, schema mismatch). + ErrCodeVectorStoreWriteFailed = "VECTORSTORE_WRITE_FAILED" + + // ErrCodeMultimodalVLMFailed — a single image's OCR or caption call + // failed. Note: per-image failures don't fail the whole parent; + // see image_multimodal.go finalize-on-last-attempt. + ErrCodeMultimodalVLMFailed = "MULTIMODAL_VLM_FAILED" + + // ErrCodeMultimodalAllFailed — every image task hit dead-letter. + // The parent still completes (caption / OCR is optional content), + // but stage status is marked failed so the UI can warn. + ErrCodeMultimodalAllFailed = "MULTIMODAL_ALL_FAILED" + + // ErrCodeTaskTimeout — asynq retry budget exhausted. Used by the + // dead-letter callback when promoting a task failure into a stage + // failure. Distinct from DocReaderTimeout: this is the asynq-level + // timeout (whole task), not the docreader-call-level timeout. + ErrCodeTaskTimeout = "TASK_TIMEOUT" + + // ErrCodeUnknown — fallback when a wrapped error doesn't classify. + // The full message is still recorded in error_detail so operators + // can debug; the UI shows a generic "see admin" hint. + ErrCodeUnknown = "UNKNOWN" +) diff --git a/internal/handler/knowledge.go b/internal/handler/knowledge.go index eefc49b5..8cb55558 100644 --- a/internal/handler/knowledge.go +++ b/internal/handler/knowledge.go @@ -34,6 +34,7 @@ type KnowledgeHandler struct { kbShareService interfaces.KBShareService agentShareService interfaces.AgentShareService asynqClient interfaces.TaskEnqueuer + stageRepo repository.KnowledgeStageRepository } // NewKnowledgeHandler creates a new knowledge handler instance @@ -43,6 +44,7 @@ func NewKnowledgeHandler( kbShareService interfaces.KBShareService, agentShareService interfaces.AgentShareService, asynqClient interfaces.TaskEnqueuer, + stageRepo repository.KnowledgeStageRepository, ) *KnowledgeHandler { return &KnowledgeHandler{ kgService: kgService, @@ -50,6 +52,7 @@ func NewKnowledgeHandler( kbShareService: kbShareService, agentShareService: agentShareService, asynqClient: asynqClient, + stageRepo: stageRepo, } } @@ -546,6 +549,99 @@ func (h *KnowledgeHandler) GetKnowledge(c *gin.Context) { }) } +// GetKnowledgeStages godoc +// @Summary 获取知识文档解析阶段进度 +// @Description 返回该知识在解析流水线(docreader/chunking/embedding/multimodal/postprocess)的每段状态、耗时与错误码。前端用于渲染时间线。 +// @Tags 知识管理 +// @Accept json +// @Produce json +// @Param id path string true "知识ID" +// @Success 200 {object} map[string]interface{} +// @Router /api/v1/knowledge/{id}/stages [get] +// +// Always returns AllProcessingStages segments — missing rows are +// synthesized as "pending" so the frontend timeline always renders five +// segments and doesn't have to know about backend persistence races. +func (h *KnowledgeHandler) GetKnowledgeStages(c *gin.Context) { + ctx := c.Request.Context() + + id := secutils.SanitizeForLog(c.Param("id")) + if id == "" { + c.Error(errors.NewBadRequestError("Knowledge ID cannot be empty")) + return + } + + knowledge, _, err := h.resolveKnowledgeAndValidateKBAccess(c, id, types.OrgRoleViewer) + if err != nil { + c.Error(err) + return + } + + rows := []types.KnowledgeProcessingStage{} + if h.stageRepo != nil { + rows, err = h.stageRepo.ListByKnowledge(ctx, knowledge.ID) + if err != nil { + // A repo error doesn't break the response — we still + // return the placeholder timeline so the UI renders. + logger.Warnf(ctx, "stages list failed for %s: %v", knowledge.ID, err) + rows = nil + } + } + + // Index by stage so synthesizing missing rows is O(N). + byStage := make(map[string]types.KnowledgeProcessingStage, len(rows)) + for _, r := range rows { + byStage[r.Stage] = r + } + + now := time.Now() + out := make([]types.KnowledgeProcessingStage, 0, len(types.AllProcessingStages)) + currentStage := "" + var lastFailure *types.KnowledgeProcessingStage + for _, name := range types.AllProcessingStages { + if r, ok := byStage[name]; ok { + out = append(out, r) + if r.Status == types.ProcessingStageRunning && currentStage == "" { + currentStage = r.Stage + } + if r.Status == types.ProcessingStageFailed { + cp := r + lastFailure = &cp + } + continue + } + // Synthesize a pending placeholder. CreatedAt set to now so + // the JSON response is well-formed; clients should treat + // timestamps on placeholders as "not meaningful". + out = append(out, types.KnowledgeProcessingStage{ + KnowledgeID: knowledge.ID, + Stage: name, + Status: types.ProcessingStagePending, + CreatedAt: now, + UpdatedAt: now, + }) + } + + resp := gin.H{ + "knowledge_id": knowledge.ID, + "parse_status": knowledge.ParseStatus, + "current_stage": currentStage, + "stages": out, + } + if lastFailure != nil { + resp["last_error"] = gin.H{ + "stage": lastFailure.Stage, + "code": lastFailure.ErrorCode, + "message": lastFailure.ErrorMessage, + "finished_at": lastFailure.FinishedAt, + } + } + c.JSON(http.StatusOK, gin.H{ + "success": true, + "data": resp, + }) +} + // ListKnowledge godoc // @Summary 获取知识列表 // @Description 获取知识库下的知识列表,支持分页和筛选 diff --git a/internal/router/router.go b/internal/router/router.go index b0a2044c..e31461c1 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -283,6 +283,7 @@ func RegisterKnowledgeRoutes(r *gin.RouterGroup, handler *handler.KnowledgeHandl // the role-only floor; /move and /batch-delete stay Contributor. k.GET("/batch", g.Viewer(), handler.GetKnowledgeBatch) k.GET("/:id", g.Viewer(), g.KBAccessReadFromKnowledgeIDParam("id"), handler.GetKnowledge) + k.GET("/:id/stages", g.Viewer(), g.KBAccessReadFromKnowledgeIDParam("id"), handler.GetKnowledgeStages) k.DELETE("/:id", g.OwnedKnowledgeKBOrAdmin(), g.KBAccessWriteFromKnowledgeIDParam("id"), handler.DeleteKnowledge) k.PUT("/:id", g.OwnedKnowledgeKBOrAdmin(), g.KBAccessWriteFromKnowledgeIDParam("id"), handler.UpdateKnowledge) k.PUT("/manual/:id", g.OwnedKnowledgeKBOrAdmin(), g.KBAccessWriteFromKnowledgeIDParam("id"), handler.UpdateManualKnowledge) diff --git a/internal/types/knowledge_stage.go b/internal/types/knowledge_stage.go new file mode 100644 index 00000000..af6812fa --- /dev/null +++ b/internal/types/knowledge_stage.go @@ -0,0 +1,64 @@ +package types + +import "time" + +// ProcessingStage names — kept as a closed set on the Go side so the +// frontend timeline can rely on exactly five segments. Adding a new stage +// requires a coordinated frontend release; rename in this file is breaking. +const ( + ProcessingStageDocReader = "docreader" + ProcessingStageChunking = "chunking" + ProcessingStageEmbedding = "embedding" + ProcessingStageMultimodal = "multimodal" + ProcessingStagePostProcess = "postprocess" +) + +// AllProcessingStages is the canonical, ordered list. Used by the API +// to fill in pending entries when no row exists yet for a stage so the +// timeline always renders five segments even before parsing starts. +var AllProcessingStages = []string{ + ProcessingStageDocReader, + ProcessingStageChunking, + ProcessingStageEmbedding, + ProcessingStageMultimodal, + ProcessingStagePostProcess, +} + +// ProcessingStageStatus values. Distinct from Knowledge.parse_status — +// each stage row tracks its own state independently of the parent. +const ( + ProcessingStagePending = "pending" // not started yet + ProcessingStageRunning = "running" // currently executing + ProcessingStageDone = "done" + ProcessingStageFailed = "failed" + ProcessingStageSkipped = "skipped" // e.g. multimodal skipped for image-free docs +) + +// KnowledgeProcessingStage is one row in knowledge_processing_stages. +// +// We intentionally keep it as a flat struct with both Go and JSON tags so +// the same type round-trips through GORM (DB layer) and the API response. +// Sensitive ErrorDetail (full stack) is excluded from the default JSON +// output via "-" — handlers must opt in explicitly for admin views. +type KnowledgeProcessingStage struct { + ID int64 `gorm:"primaryKey;column:id" json:"-"` + KnowledgeID string `gorm:"column:knowledge_id;index" json:"knowledge_id"` + Stage string `gorm:"column:stage;size:32" json:"stage"` + Status string `gorm:"column:status;size:16" json:"status"` + StartedAt *time.Time `gorm:"column:started_at" json:"started_at,omitempty"` + FinishedAt *time.Time `gorm:"column:finished_at" json:"finished_at,omitempty"` + DurationMs int64 `gorm:"column:duration_ms" json:"duration_ms,omitempty"` + ErrorCode string `gorm:"column:error_code;size:64" json:"error_code,omitempty"` + ErrorMessage string `gorm:"column:error_message;type:text" json:"error_message,omitempty"` + ErrorDetail string `gorm:"column:error_detail;type:text" json:"-"` + Attempt int `gorm:"column:attempt" json:"attempt"` + CreatedAt time.Time `gorm:"column:created_at;autoCreateTime" json:"created_at"` + UpdatedAt time.Time `gorm:"column:updated_at;autoUpdateTime" json:"updated_at"` +} + +// TableName pins the GORM table since the natural pluralization +// ("knowledge_processing_stages") happens to match what the migration +// creates — but explicit beats implicit here. +func (KnowledgeProcessingStage) TableName() string { + return "knowledge_processing_stages" +} diff --git a/migrations/versioned/000052_knowledge_processing_stages.down.sql b/migrations/versioned/000052_knowledge_processing_stages.down.sql new file mode 100644 index 00000000..9d1cf0e2 --- /dev/null +++ b/migrations/versioned/000052_knowledge_processing_stages.down.sql @@ -0,0 +1,8 @@ +-- Migration: 000052_knowledge_processing_stages (rollback) +DO $$ BEGIN RAISE NOTICE '[Migration 000052 rollback] Dropping table: knowledge_processing_stages'; END $$; + +DROP INDEX IF EXISTS idx_kps_status_started; +DROP INDEX IF EXISTS idx_kps_knowledge; +DROP TABLE IF EXISTS knowledge_processing_stages; + +DO $$ BEGIN RAISE NOTICE '[Migration 000052 rollback] complete'; END $$; diff --git a/migrations/versioned/000052_knowledge_processing_stages.up.sql b/migrations/versioned/000052_knowledge_processing_stages.up.sql new file mode 100644 index 00000000..4e2ce4d8 --- /dev/null +++ b/migrations/versioned/000052_knowledge_processing_stages.up.sql @@ -0,0 +1,61 @@ +-- Migration: 000052_knowledge_processing_stages +-- Per-knowledge stage progress for the document parsing pipeline. +-- +-- Background: knowledge.parse_status is a 4-state field (pending / +-- processing / completed / failed). When users hit "stuck in processing", +-- they have no way to tell which stage is actually running — DocReader, +-- chunking, embedding, multimodal OCR/VLM, or the final post-process +-- handoff. Operators face the same problem: a 500MB PDF stuck in +-- "processing" for 90 minutes could be legitimate OCR work or a frozen +-- docreader call, and the only signal is the absence of progress. +-- +-- This table persists per-stage progress so: +-- 1. The frontend can render a five-segment timeline showing where +-- each document is in the pipeline (PR③ work). +-- 2. Failures carry a stable error_code (DOCREADER_TIMEOUT, +-- EMBEDDING_RATE_LIMIT, ...) the UI can map to localized +-- remediation text. +-- 3. Operators can SQL-query "all rows stuck on stage=embedding for +-- >30 min" without log-scraping. +-- +-- Schema decisions: +-- - One row per (knowledge_id, stage). Retry attempts BUMP attempt +-- counter and reset status; we don't keep historical attempt rows +-- because the timeline UI only needs the latest state per stage. +-- - error_message holds the user-facing summary (≤1KB); +-- error_detail is the full stack/payload (≤8KB) gated behind +-- admin views in the UI. +-- - Sweeps that recover stuck knowledge will leave these rows +-- intact so operators can post-mortem the failure. +DO $$ BEGIN RAISE NOTICE '[Migration 000052] Creating table: knowledge_processing_stages'; END $$; + +CREATE TABLE IF NOT EXISTS knowledge_processing_stages ( + id BIGSERIAL PRIMARY KEY, + knowledge_id VARCHAR(64) NOT NULL, + stage VARCHAR(32) NOT NULL, + status VARCHAR(16) NOT NULL, + started_at TIMESTAMP WITH TIME ZONE, + finished_at TIMESTAMP WITH TIME ZONE, + duration_ms BIGINT, + error_code VARCHAR(64), + error_message TEXT, + error_detail TEXT, + attempt INT NOT NULL DEFAULT 1, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT uq_kps_knowledge_stage UNIQUE (knowledge_id, stage) +); + +-- Primary read path: "give me every stage row for this knowledge_id". +-- The unique constraint already covers point lookups by (knowledge_id, +-- stage); this index lets us fetch all 5 rows for a knowledge with one +-- index range scan, which is the API surface the frontend timeline calls. +CREATE INDEX IF NOT EXISTS idx_kps_knowledge + ON knowledge_processing_stages (knowledge_id); + +-- Operator query: "find stages stuck in running too long". Used by the +-- housekeeping cron in a future iteration and by ad-hoc SQL diagnostics. +CREATE INDEX IF NOT EXISTS idx_kps_status_started + ON knowledge_processing_stages (status, started_at); + +DO $$ BEGIN RAISE NOTICE '[Migration 000052] knowledge_processing_stages table ready'; END $$;