weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): expose dedup hint, parse error and reparse endpoint

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-21 09:18:20 +00:00
co-authored by Cursor
parent 31dde45b72
commit 63f94ff23b
3 changed files with 167 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/materialindex/pipeline.go
+47 -5
View File
@@ -150,14 +150,30 @@ func (p *Pipeline) GetParseState(ctx context.Context, blobSHA string) string {
return rec.State
}
// ParseInfo is the user-facing parse status for one blob.
type ParseInfo struct {
State string `json:"state"`
ErrorMessage string `json:"error_message,omitempty"`
}
// GetParseStates batch-loads parse states keyed by blob sha.
func (p *Pipeline) GetParseStates(ctx context.Context, blobSHAs []string) map[string]string {
out := make(map[string]string, len(blobSHAs))
infos := p.GetParseInfos(ctx, blobSHAs)
out := make(map[string]string, len(infos))
for sha, info := range infos {
out[sha] = info.State
}
return out
}
// GetParseInfos batch-loads parse state + error message keyed by blob sha.
func (p *Pipeline) GetParseInfos(ctx context.Context, blobSHAs []string) map[string]ParseInfo {
out := make(map[string]ParseInfo, len(blobSHAs))
if p == nil || p.pool == nil || len(blobSHAs) == 0 {
return out
}
rows, err := p.pool.Query(ctx, `
SELECT blob_sha256, state FROM asset_parses
SELECT blob_sha256, state, COALESCE(error_message, '') FROM asset_parses
WHERE engine = $1 AND blob_sha256 = ANY($2)
`, engineMinerU, blobSHAs)
if err != nil {
@@ -165,14 +181,40 @@ func (p *Pipeline) GetParseStates(ctx context.Context, blobSHAs []string) map[st
}
defer rows.Close()
for rows.Next() {
var sha, state string
if rows.Scan(&sha, &state) == nil {
out[sha] = state
var sha, state, errMsg string
if rows.Scan(&sha, &state, &errMsg) == nil {
out[sha] = ParseInfo{State: state, ErrorMessage: errMsg}
}
}
return out
}
// ResetForReparse rolls a failed/exhausted parse back to pending so the next
// EnqueueAsync/EnsureParsed cycle retries MinerU. Returns false when the blob
// is unknown or still in progress (pending/running/done).
func (p *Pipeline) ResetForReparse(ctx context.Context, blobSHA string) bool {
if p == nil || p.pool == nil || blobSHA == "" {
return false
}
tag, err := p.pool.Exec(ctx, `
UPDATE asset_parses SET state = $2, attempts = 0, error_message = '', updated_at = now()
WHERE blob_sha256 = $1 AND engine = $3 AND state NOT IN ($4, $5)
`, blobSHA, statePending, engineMinerU, stateRunning, stateDone)
if err != nil {
return false
}
if tag.RowsAffected() > 0 {
return true
}
// No row updated: either the record is done/running (must not reset) or it
// never existed. When it never existed, allow a fresh parse to start.
rec, err := p.getRecord(ctx, blobSHA)
if err != nil || rec != nil {
return false
}
return true
}
// EnsureParsed waits for or triggers MinerU parse. On miss/disable/fail returns
// fallback text from squad.ExtractTextWithAnchors (may be empty).
func (p *Pipeline) EnsureParsed(ctx context.Context, fileName string, blobSHA string, data []byte) (markdown string, fromMinerU bool, err error) {
internal/server/asset_handlers.go
+119 -24
View File
@@ -8,17 +8,21 @@ import (
"strings"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/materialindex"
"chat-agent-system/internal/mineru"
"chat-agent-system/internal/upload"
)
type projectAssetResponse struct {
AssetID int64 `json:"asset_id"`
Filename string `json:"filename"`
SHA256 string `json:"sha256"`
MimeType string `json:"mime_type"`
SizeBytes int64 `json:"size_bytes"`
ProjectID int64 `json:"project_id"`
ParseState string `json:"parse_state,omitempty"`
AssetID int64 `json:"asset_id"`
Filename string `json:"filename"`
SHA256 string `json:"sha256"`
MimeType string `json:"mime_type"`
SizeBytes int64 `json:"size_bytes"`
ProjectID int64 `json:"project_id"`
ParseState string `json:"parse_state,omitempty"`
ParseError string `json:"parse_error,omitempty"`
Deduplicated bool `json:"deduplicated,omitempty"`
}
func (s *Server) handleUploadProjectAsset(w http.ResponseWriter, r *http.Request) {
@@ -80,21 +84,79 @@ func (s *Server) handleUploadProjectAsset(w http.ResponseWriter, r *http.Request
}
parseState := ""
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil {
matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID)
if matIdx.Enabled() {
parseState = "pending"
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() {
if mineru.SupportsExt(ref.FileName) {
matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID)
// Report the real state: same-content uploads hit an existing parse
// (possibly already done) instead of faking "pending".
if state := matIdx.GetParseState(r.Context(), ref.BlobSHA256); state != "" {
parseState = state
} else {
parseState = "pending"
}
} else {
parseState = "skipped"
}
}
writeJSONResponse(w, http.StatusOK, projectAssetResponse{
AssetID: ref.ID,
Filename: ref.FileName,
SHA256: ref.BlobSHA256,
MimeType: mimeType,
SizeBytes: int64(len(data)),
ProjectID: projectID,
ParseState: parseState,
AssetID: ref.ID,
Filename: ref.FileName,
SHA256: ref.BlobSHA256,
MimeType: mimeType,
SizeBytes: int64(len(data)),
ProjectID: projectID,
ParseState: parseState,
Deduplicated: ref.IsDeduplicated,
})
}
// handleReparseProjectAsset resets a failed material parse and re-enqueues MinerU.
func (s *Server) handleReparseProjectAsset(w http.ResponseWriter, r *http.Request) {
projectID, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
if err != nil || projectID <= 0 {
writeSafeError(w, r, http.StatusBadRequest, "invalid project id")
return
}
assetID, err := strconv.ParseInt(r.PathValue("asset_id"), 10, 64)
if err != nil || assetID <= 0 {
writeSafeError(w, r, http.StatusBadRequest, "invalid asset id")
return
}
if _, ok := s.requireProjectAccess(w, r, projectID); !ok {
return
}
cas := s.cas
if cas == nil {
writeSafeError(w, r, http.StatusServiceUnavailable, "storage unavailable")
return
}
data, ref, err := cas.ReadRefBytes(r.Context(), assetID)
if err != nil || ref == nil {
writeSafeError(w, r, http.StatusNotFound, "asset not found")
return
}
if ref.ProjectID != projectID {
writeSafeError(w, r, http.StatusNotFound, "asset not found")
return
}
matIdx := bizdev.Snapshot(s.svc).MaterialIndex
if matIdx == nil || !matIdx.Enabled() {
writeSafeError(w, r, http.StatusServiceUnavailable, "parse engine unavailable")
return
}
if !mineru.SupportsExt(ref.FileName) {
writeSafeError(w, r, http.StatusBadRequest, "file type not supported by parser")
return
}
if !matIdx.ResetForReparse(r.Context(), ref.BlobSHA256) {
writeSafeError(w, r, http.StatusConflict, "parse is in progress or already complete")
return
}
matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID)
writeJSONResponse(w, http.StatusOK, map[string]any{
"asset_id": ref.ID,
"parse_state": "pending",
})
}
@@ -124,20 +186,53 @@ func (s *Server) handleListProjectAssets(w http.ResponseWriter, r *http.Request)
shas = append(shas, ref.BlobSHA256)
}
}
states := map[string]string{}
infos := map[string]materialindex.ParseInfo{}
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil {
states = matIdx.GetParseStates(r.Context(), shas)
infos = matIdx.GetParseInfos(r.Context(), shas)
}
out := make([]projectAssetResponse, 0, len(refs))
seenSHA := make(map[string]bool, len(refs))
shaCount := make(map[string]int, len(refs))
for _, ref := range refs {
out = append(out, projectAssetResponse{
if ref != nil && ref.BlobSHA256 != "" {
shaCount[ref.BlobSHA256]++
}
}
out := make([]projectAssetResponse, 0, len(refs))
firstRefOfSHA := make(map[string]int64, len(refs))
for _, ref := range refs {
if ref != nil && ref.BlobSHA256 != "" {
if _, ok := firstRefOfSHA[ref.BlobSHA256]; !ok {
firstRefOfSHA[ref.BlobSHA256] = ref.ID
}
}
}
for _, ref := range refs {
info := infos[ref.BlobSHA256]
item := projectAssetResponse{
AssetID: ref.ID,
Filename: ref.FileName,
SHA256: ref.BlobSHA256,
ProjectID: ref.ProjectID,
ParseState: states[ref.BlobSHA256],
})
ParseState: info.State,
ParseError: info.ErrorMessage,
}
// 同项目内相同内容多次引用：除首条外全部标注重复（物理只存一份）。
if shaCount[ref.BlobSHA256] > 1 && firstRefOfSHA[ref.BlobSHA256] != ref.ID {
item.Deduplicated = true
}
if item.ParseState == "" && ref.BlobSHA256 != "" && !seenSHA[ref.BlobSHA256] {
// 内容已入库但从未触发解析（如早于 MinerU 接入的历史资产）：
// 列表读取时自动补触发，前端轮询即可看到 pending→running→done。
seenSHA[ref.BlobSHA256] = true
if matIdx := bizdev.Snapshot(s.svc).MaterialIndex; matIdx != nil && matIdx.Enabled() && mineru.SupportsExt(ref.FileName) {
if data, _, err := cas.ReadRefBytes(r.Context(), ref.ID); err == nil {
matIdx.EnqueueAsync(ref.FileName, ref.BlobSHA256, data, projectID)
item.ParseState = "pending"
}
}
}
out = append(out, item)
}
writeJSONResponse(w, http.StatusOK, out)
}
internal/server/server.go
+1
View File
@@ -246,6 +246,7 @@ func (s *Server) Routes() http.Handler {
mux.HandleFunc("POST /api/v1/projects/{id}/assets", s.handleUploadProjectAsset)
mux.HandleFunc("GET /api/v1/projects/{id}/assets", s.handleListProjectAssets)
mux.HandleFunc("GET /api/v1/projects/{id}/assets/{asset_id}/raw", s.handleGetProjectAssetRaw)
mux.HandleFunc("POST /api/v1/projects/{id}/assets/{asset_id}/reparse", s.handleReparseProjectAsset)
// 6. 多级上下文与偏好记忆管理
mux.HandleFunc("GET /api/v1/context/user", s.handleGetUserContext)