mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-19 00:15:05 +00:00
fix(vault): enrichment pipeline reliability + cross-agent classify
9 fixes for vault enrichment pipeline: 1. Queue key = tenant-only (was per-agent, caused multiple batches blocking EventBus workers and progress bar flashing) 2. Classify chunks 5 candidates per LLM call (prevents response truncation that caused parse_still_failed errors) 3. Classify prompt improved: explicit "EXACTLY one entry per candidate", 5-entry example, ctx capped at 30 words 4. max_tokens kept at 1024 (sufficient for 5 candidates) 5. Progress AddDone removes !running guard (safe before Start) 6. Rescan defers event publishing via PendingEvents — Start() called before workers receive events, eliminating race 7. Upload handler same deferred publish pattern 8. Frontend enrichment timer cancels stale "complete" timeout when new enrichment starts (prevents bar disappearing) 9. Sidebar tree reloads after rescan completes Classify now searches across entire tenant (empty agentID) to build cross-agent links for future vault sharing. Access control enforced at query time — agents only see their own docs.
This commit is contained in:
1 parent
bfdaeb2382
commit
28d29ba046
10 files changed
+152
-139
No files matched your search
@@ -129,6 +129,7 @@ func (h *VaultHandler) handleUpload(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
var results []uploadResult
|
||||
var created int
|
||||
var pendingEvents []eventbus.DomainEvent
|
||||
|
||||
for _, fh := range files {
|
||||
// Sanitize filename — basename only, no path traversal.
|
||||
@@ -208,13 +209,13 @@ func (h *VaultHandler) handleUpload(w http.ResponseWriter, r *http.Request) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Publish enrichment event (same pattern as rescan.go).
|
||||
// Collect enrichment events — published after Start() to avoid race.
|
||||
if h.eventBus != nil {
|
||||
agentForEvent := ""
|
||||
if agentIDStr != "" {
|
||||
agentForEvent = agentIDStr
|
||||
}
|
||||
h.eventBus.Publish(eventbus.DomainEvent{
|
||||
pendingEvents = append(pendingEvents, eventbus.DomainEvent{
|
||||
ID: uuid.Must(uuid.NewV7()).String(),
|
||||
Type: eventbus.EventVaultDocUpserted,
|
||||
SourceID: doc.ID + ":" + hash,
|
||||
@@ -236,10 +237,13 @@ func (h *VaultHandler) handleUpload(w http.ResponseWriter, r *http.Request) {
|
||||
created++
|
||||
}
|
||||
|
||||
// Signal enrichment progress for WS subscribers.
|
||||
// Start progress BEFORE publishing events to avoid race with workers.
|
||||
if h.enrichProgress != nil && created > 0 {
|
||||
h.enrichProgress.Start(created, tenantID)
|
||||
}
|
||||
for _, event := range pendingEvents {
|
||||
h.eventBus.Publish(event)
|
||||
}
|
||||
|
||||
slog.Info("vault.upload", "tenant", tenantIDStr, "uploaded", created, "errors", len(files)-created)
|
||||
|
||||
|
||||
@@ -195,9 +195,16 @@ func (h *VaultHandler) handleRescan(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Signal enrichment progress so WS subscribers see running=true immediately.
|
||||
if h.enrichProgress != nil && (result.New+result.Updated) > 0 {
|
||||
h.enrichProgress.Start(result.New+result.Updated, store.TenantIDFromContext(r.Context()))
|
||||
// Start progress BEFORE publishing events so workers see running=true
|
||||
// and AddDone calls are not dropped by the !running guard.
|
||||
total := result.New + result.Updated
|
||||
if h.enrichProgress != nil && total > 0 {
|
||||
h.enrichProgress.Start(total, store.TenantIDFromContext(r.Context()))
|
||||
}
|
||||
|
||||
// Now publish enrichment events — workers will call AddDone after Start.
|
||||
for _, event := range result.PendingEvents {
|
||||
h.eventBus.Publish(event)
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
|
||||
Reference in new issue
Block a user