feat(knowledge): track per-stage parsing progress with /stages API

Adds a five-segment progress model for the document parsing pipeline so the UI (PR③) can render a timeline showing where each document is (DocReader → Chunking → Embedding → Multimodal → PostProcess) and which stage failed with what error code. - New table `knowledge_processing_stages` (migration 000052) with one row per (knowledge_id, stage). UPSERT on Begin/Done/Fail bumps an attempt counter so re-parses don't lose history. - StageTracker service exposes Begin/Done/Fail/Skip; all calls are best-effort and never break the pipeline if persistence fails. - Stable error codes (DOCREADER_TIMEOUT / EMBEDDING_RATE_LIMIT / VECTORSTORE_WRITE_FAILED / ...) the UI can map to localized remediation hints. - Tracker call sites added at the four meaningful failure points: convert (DocReader), CreateChunks (Chunking), BatchIndex (Embedding), enqueueImageMultimodalTasks (Multimodal start), KnowledgePostProcess.Handle (Multimodal close + PostProcess). - New endpoint `GET /api/v1/knowledge/:id/stages` returns the five canonical stages — missing rows are synthesized as "pending" so the timeline always renders five segments. Includes current_stage and last_error block.
2026-05-27 13:31:06 +08:00
parent 3ae3ea97c5
commit 04f56f9cda
12 changed files with 666 additions and 6 deletions
@@ -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
+}
@@ -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 {
@@ -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
 }

@@ -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 {
@@ -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)                {}
@@ -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))
@@ -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"
+)
@@ -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  获取知识库下的知识列表，支持分页和筛选
@@ -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)
@@ -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"
+}
@@ -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 $$;
@@ -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 $$;