From 1f37f90b4b8eabc4d8eddd70de23f080bf842502 Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Wed, 23 Sep 2026 09:55:59 +0800 Subject: [PATCH] feat(chat): show MinerU parse progress while waiting in-turn When summarizing the latest upload, poll parse state up to MinerU timeout, emit document_parsing phase events so the bubble stays live and input stays locked, then summarize or return a clear slow/failed reply. --- .doc/ops/CURRENT.md | 1 + .../2026-09-23-latest-upload-parse-wait.md | 55 +++++++++ internal/coordinator/context.go | 2 + internal/materialindex/pipeline.go | 8 ++ internal/server/chat_document_reader.go | 108 +++++++++++++++++- internal/server/chat_document_reader_test.go | 78 +++++++++++++ web/src/engine/phaseEventMapper.test.ts | 16 ++- web/src/engine/phaseEventMapper.ts | 2 + 8 files changed, 268 insertions(+), 2 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-23-latest-upload-parse-wait.md diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index dd9129d7..bea79117 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -18,6 +18,7 @@ | owner | task | branch | worktree | mutex | status | |---|---|---|---|---|---| +| Cursor | fix-latest-upload-parse-wait-2026-09-23 | feature/fix-latest-upload-summary-2026-09-23 | .worktrees/fix-latest-upload-summary-2026-09-23 | intent+chat-doc-summary | done | | Cursor | fix-latest-upload-summary-2026-09-23 | feature/fix-latest-upload-summary-2026-09-23 | .worktrees/fix-latest-upload-summary-2026-09-23 | intent+chat-doc-summary | done | | Cursor | modular-refactor-review-2026-09-23 | feature/modular-refactor-review-2026-09-23 | .worktrees/modular-refactor-review-2026-09-23 | ops-audit-modular-refactor-review | done | | Cursor | 2026-09-22-status-reconcile | feature/status-reconcile-2026-09-22 | .worktrees/status-reconcile-2026-09-22 | ops-status-reconcile | done | diff --git a/.doc/ops/short-term/2026-09-23-latest-upload-parse-wait.md b/.doc/ops/short-term/2026-09-23-latest-upload-parse-wait.md new file mode 100644 index 00000000..9c6c6bc4 --- /dev/null +++ b/.doc/ops/short-term/2026-09-23-latest-upload-parse-wait.md @@ -0,0 +1,55 @@ +--- +title: "上传文档解析中可见进度与同轮等待" +task_id: "fix-latest-upload-parse-wait-2026-09-23" +parent_task_id: "fix-latest-upload-summary-2026-09-23" +status: done +created: 2026-09-23 +updated: 2026-09-23 +owner: Cursor +task_type: implementation +task_tier: small +policy_revision: "agent-governance-2026-09-21-r1 @ 2ce8688db4f49fdd116799b9d36ef7ea802afc33" +branch: feature/fix-latest-upload-summary-2026-09-23 +worktree: .worktrees/fix-latest-upload-summary-2026-09-23 +base_dev_sha: "2ce8688db4f49fdd116799b9d36ef7ea802afc33" +write_scope: + - internal/server + - internal/coordinator + - internal/materialindex + - web/src + - .doc/ops +allowed_environments: [] +head_sha: "" +mutex: "intent+chat-doc-summary" +review_state: pending +integration_state: pending +deployment: not_required +cleanup: pending +cleanup_owner: 独立维护者 +cleanup_trigger: tip 为 local_dev 祖先且无进程后 remove +remote_sync: not_requested +--- + +# 上传文档解析中可见进度与同轮等待 + +父任务：[2026-09-23-latest-upload-doc-summary.md](./2026-09-23-latest-upload-doc-summary.md)。本追问：解析中发 `document_parsing` phase，同轮等待并锁定发送。 + +## 改动摘要 + +- `materialindex.ParseWaitLimit()`：暴露等待上限。 +- `waitForUploadParse`：同轮每 2s 轮询并 `EmitPhase("document_parsing", …)`；超时 `slow`；失败/完成分派。 +- `PHASE_META`：`document_parsing` / `document_summarizing`。 +- 前端发送锁：复用既有 `isStreamingActive`（phase → streaming_status → 禁发）。 + +## 验证与证据 + +| 范围 | 命令 | 判定 | +|---|---|---| +| server | `go test ./internal/server/ -run 'ResolveLatest\|WaitForUploadParse\|PickLatest\|SelectReadable\|CrossWorkflow'` | PASS | +| vet | `go vet ./internal/materialindex/... ./internal/server/... ./internal/coordinator/...` | PASS | +| arch | `./scripts/arch-check.sh --mode block` | PASS | +| web | `npx tsx --test src/engine/phaseEventMapper.test.ts` | PASS（16） | + +## 收口 + +- 本地提交见本分支 tip；dev 集成 / 部署另授；不 push。 diff --git a/internal/coordinator/context.go b/internal/coordinator/context.go index c6a55638..f51c7fd6 100644 --- a/internal/coordinator/context.go +++ b/internal/coordinator/context.go @@ -167,6 +167,8 @@ func (e *ErrUploadParseState) Error() string { return fmt.Sprintf("《%s》仍在解析中，请稍后再试总结。", name) case "failed": return fmt.Sprintf("《%s》解析失败。可在附件卡片点击重新解析后再试总结。", name) + case "slow": + return fmt.Sprintf("《%s》解析耗时较长（仍在后台进行），请稍后再让我总结。", name) default: return fmt.Sprintf("《%s》暂时无法读取正文，请稍后再试。", name) } diff --git a/internal/materialindex/pipeline.go b/internal/materialindex/pipeline.go index 1fbcdad8..aa7d34bb 100644 --- a/internal/materialindex/pipeline.go +++ b/internal/materialindex/pipeline.go @@ -147,6 +147,14 @@ func (p *Pipeline) RAGTopK() int { return p.ragTopK } +// ParseWaitLimit returns the bounded wait for pending/running MinerU parses. +func (p *Pipeline) ParseWaitLimit() time.Duration { + if p == nil || p.waitLimit <= 0 { + return defaultWait + } + return p.waitLimit +} + // GetParseState returns parse state for a blob (empty string if unknown). func (p *Pipeline) GetParseState(ctx context.Context, blobSHA string) string { rec, err := p.getRecord(ctx, blobSHA) diff --git a/internal/server/chat_document_reader.go b/internal/server/chat_document_reader.go index 0c9dabdd..b1a7c1a1 100644 --- a/internal/server/chat_document_reader.go +++ b/internal/server/chat_document_reader.go @@ -5,6 +5,7 @@ import ( "fmt" "path/filepath" "strings" + "time" "unicode" "chat-agent-system/internal/auth" @@ -16,6 +17,7 @@ import ( "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/squad" "chat-agent-system/internal/storage" + "chat-agent-system/internal/stream" "chat-agent-system/internal/workflowscope" ) @@ -119,11 +121,50 @@ func (s *Server) resolveLatestUploadDocument(ctx context.Context, projectID int6 state = strings.TrimSpace(matIdx.GetParseState(ctx, latest.BlobSHA256)) } switch state { - case "pending", "running", "failed": + case "failed": return nil, &coordinator.ErrUploadParseState{ State: state, Filename: latest.FileName, } + case "pending", "running": + limit := 90 * time.Second + if s.cfg != nil && s.cfg.MineruTimeout > 0 { + limit = s.cfg.MineruTimeout + } + if matIdx != nil { + if w := matIdx.ParseWaitLimit(); w > 0 { + limit = w + } + } + finalState := waitForUploadParse(ctx, func() string { + if matIdx == nil { + return "" + } + return strings.TrimSpace(matIdx.GetParseState(ctx, latest.BlobSHA256)) + }, latest.FileName, limit) + switch finalState { + case "done": + // continue to load content + case "failed": + return nil, &coordinator.ErrUploadParseState{ + State: "failed", + Filename: latest.FileName, + } + case "slow", "pending", "running": + return nil, &coordinator.ErrUploadParseState{ + State: "slow", + Filename: latest.FileName, + Message: fmt.Sprintf( + "《%s》解析耗时较长（仍在后台进行），请稍后再让我总结。", + defaultUploadTitle(latest.FileName), + ), + } + default: + if ctx.Err() != nil { + return nil, ctx.Err() + } + // unknown / empty after wait: fall through and try EnsureParsed + } } data, ref, err := s.cas.ReadRefBytes(ctx, latest.ID) @@ -164,6 +205,71 @@ func (s *Server) resolveLatestUploadDocument(ctx context.Context, projectID int6 }, nil } +var uploadParsePollInterval = 2 * time.Second + +// waitForUploadParse 在同轮对话内有界等待 MinerU 解析，并周期性发 document_parsing phase。 +// stateFn 返回当前解析状态；终态 "done"/"failed" 立即返回；超时返回 "slow"。 +func waitForUploadParse(ctx context.Context, stateFn func() string, fileName string, limit time.Duration) string { + if limit <= 0 { + limit = 90 * time.Second + } + title := defaultUploadTitle(fileName) + started := time.Now() + emitUploadParsePhase(ctx, title, 0, 5) + + ticker := time.NewTicker(uploadParsePollInterval) + defer ticker.Stop() + deadline := time.NewTimer(limit) + defer deadline.Stop() + + for { + state := "" + if stateFn != nil { + state = strings.TrimSpace(stateFn()) + } + switch state { + case "done", "failed": + return state + } + + select { + case <-ctx.Done(): + return state + case <-deadline.C: + final := "" + if stateFn != nil { + final = strings.TrimSpace(stateFn()) + } + if final == "done" || final == "failed" { + return final + } + return "slow" + case <-ticker.C: + elapsed := time.Since(started) + progress := int(elapsed.Seconds() / limit.Seconds() * 100) + if progress > 95 { + progress = 95 + } + if progress < 5 { + progress = 5 + } + emitUploadParsePhase(ctx, title, int(elapsed.Seconds()), progress) + } + } +} + +func emitUploadParsePhase(ctx context.Context, title string, elapsedSec, progress int) { + st := stream.FromContext(ctx) + if st == nil { + return + } + msg := fmt.Sprintf("《%s》已排队解析，正在等待 MinerU…", title) + if elapsedSec > 0 { + msg = fmt.Sprintf("《%s》正在解析中…已等待 %ds", title, elapsedSec) + } + st.EmitPhase("document_parsing", msg, "assistant", 0, 0, progress) +} + func pickLatestFileReference(refs []*storage.FileReference) *storage.FileReference { var latest *storage.FileReference for _, ref := range refs { diff --git a/internal/server/chat_document_reader_test.go b/internal/server/chat_document_reader_test.go index ab5e7e81..85368d41 100644 --- a/internal/server/chat_document_reader_test.go +++ b/internal/server/chat_document_reader_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "strings" + "sync/atomic" "testing" "time" @@ -13,6 +14,7 @@ import ( "chat-agent-system/internal/docgov" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/storage" + "chat-agent-system/internal/stream" "chat-agent-system/internal/workflowscope" ) @@ -182,3 +184,79 @@ func TestResolveLatestUploadDocument_NoAsset(t *testing.T) { t.Fatalf("want ErrNoUploadAsset, got %T %v", err, err) } } + +func TestWaitForUploadParse_PendingThenDone(t *testing.T) { + prev := uploadParsePollInterval + uploadParsePollInterval = 15 * time.Millisecond + t.Cleanup(func() { uploadParsePollInterval = prev }) + + var calls atomic.Int32 + stateFn := func() string { + n := calls.Add(1) + if n >= 3 { + return "done" + } + return "pending" + } + + var phases []string + ctx, st := stream.WithListener(context.Background(), false, func(ev stream.StreamEvent) { + if ev.Type == stream.EventPhase && ev.Phase != nil { + phases = append(phases, ev.Phase.PhaseID+":"+ev.Phase.Message) + } + }) + _ = st + + got := waitForUploadParse(ctx, stateFn, "冲击波说明书.pdf", 2*time.Second) + if got != "done" { + t.Fatalf("got %q want done", got) + } + if calls.Load() < 3 { + t.Fatalf("expected multiple polls, calls=%d", calls.Load()) + } + if len(phases) == 0 { + t.Fatal("expected document_parsing phase events") + } + if !strings.Contains(phases[0], "document_parsing:") { + t.Fatalf("first phase=%q", phases[0]) + } + if !strings.Contains(phases[0], "MinerU") && !strings.Contains(phases[0], "解析") { + t.Fatalf("initial phase should mention parse wait: %q", phases[0]) + } +} + +func TestWaitForUploadParse_TimeoutSlow(t *testing.T) { + prev := uploadParsePollInterval + uploadParsePollInterval = 10 * time.Millisecond + t.Cleanup(func() { uploadParsePollInterval = prev }) + + got := waitForUploadParse(context.Background(), func() string { return "pending" }, "spec.pdf", 40*time.Millisecond) + if got != "slow" { + t.Fatalf("got %q want slow", got) + } +} + +func TestWaitForUploadParse_ContextCancel(t *testing.T) { + prev := uploadParsePollInterval + uploadParsePollInterval = 50 * time.Millisecond + t.Cleanup(func() { uploadParsePollInterval = prev }) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + got := waitForUploadParse(ctx, func() string { return "running" }, "spec.pdf", time.Second) + if got != "running" && got != "" { + // cancelled before/at select: returns last observed state + t.Fatalf("got %q on cancel", got) + } +} + +func TestWaitForUploadParse_Failed(t *testing.T) { + prev := uploadParsePollInterval + uploadParsePollInterval = 10 * time.Millisecond + t.Cleanup(func() { uploadParsePollInterval = prev }) + + got := waitForUploadParse(context.Background(), func() string { return "failed" }, "spec.pdf", time.Second) + if got != "failed" { + t.Fatalf("got %q want failed", got) + } +} diff --git a/web/src/engine/phaseEventMapper.test.ts b/web/src/engine/phaseEventMapper.test.ts index 639461d8..491f50c4 100644 --- a/web/src/engine/phaseEventMapper.test.ts +++ b/web/src/engine/phaseEventMapper.test.ts @@ -1,6 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { applyPhaseEvent } from './phaseEventMapper'; +import { applyPhaseEvent, PHASE_META } from './phaseEventMapper'; test('applyPhaseEvent: appends first phase as running node', () => { const strip = applyPhaseEvent([], { phase_id: 'squad_assembly', message: '组建小队' }); @@ -126,3 +126,17 @@ test('applyPhaseEvent: stores detail from phase.message', () => { const strip = applyPhaseEvent([], { phase_id: 'squad_assembly', message: '正在组建真实小队...' }); assert.equal(strip[0].detail, '正在组建真实小队...'); }); + +test('PHASE_META: document_parsing and document_summarizing are registered', () => { + assert.equal(PHASE_META.document_parsing.icon, '📄'); + assert.equal(PHASE_META.document_parsing.label, '文档解析'); + assert.equal(PHASE_META.document_summarizing.icon, '📝'); + assert.equal(PHASE_META.document_summarizing.label, '正文总结'); + const strip = applyPhaseEvent([], { + phase_id: 'document_parsing', + message: '《说明书》正在解析中…已等待 4s', + progress: 20, + }); + assert.equal(strip[0].icon, '📄'); + assert.match(strip[0].label, /解析中/); +}); diff --git a/web/src/engine/phaseEventMapper.ts b/web/src/engine/phaseEventMapper.ts index 341c7645..2bf5c611 100644 --- a/web/src/engine/phaseEventMapper.ts +++ b/web/src/engine/phaseEventMapper.ts @@ -24,6 +24,8 @@ export const PHASE_META: Record = { stage_completed: { icon: '🎉', label: '阶段完成' }, step_failed: { icon: '⚠️', label: '步骤异常' }, task_failed: { icon: '⚠️', label: '任务异常' }, + document_parsing: { icon: '📄', label: '文档解析' }, + document_summarizing: { icon: '📝', label: '正文总结' }, }; /** 收口型 phase：不追加新节点，仅把当前 running 条目置为 completed */