package workbench import ( "context" "time" "github.com/Tencent/WeKnora/internal/types" "github.com/google/uuid" "gorm.io/gorm" "gorm.io/gorm/clause" ) type consumptionContextKey struct{} type consumptionAttribution struct{ AgentID, ResourceID, SegmentID, JobID string } // Only a segment generation reserves evidence. Planning and previews are reads. func (s *Service) reserveEvidence(ctx context.Context, source *types.SearchResult) (bool, error) { a, ok := ctx.Value(consumptionContextKey{}).(consumptionAttribution) if !ok { return true, nil } t, u, _, e := identity(ctx) if e != nil { return false, e } var knowledge struct { KnowledgeBaseID string FolderPath string } if e = s.db.WithContext(ctx).Table("knowledges").Select("knowledge_base_id,folder_path").Where("id=?", source.KnowledgeID).Take(&knowledge).Error; e != nil { return false, e } allowed := false e = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { now := time.Now() until := now.Add(10 * time.Minute) excerpt := []rune(source.Content) if len(excerpt) > 800 { excerpt = excerpt[:800] } row := Consumption{ID: uuid.NewString(), TenantID: t, OwnerID: u, AgentID: a.AgentID, KBID: knowledge.KnowledgeBaseID, ChunkID: source.ID, KnowledgeID: source.KnowledgeID, FileName: source.KnowledgeFilename, Category: knowledge.FolderPath, Excerpt: string(excerpt), Status: "open", CreatedAt: now, UpdatedAt: now} if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&row).Error; err != nil { return err } // A uniqueness conflict keeps the newly generated ID in row. Clear it // so GORM locks the existing business key, not that unused primary key. row = Consumption{} if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("tenant_id=? AND owner_id=? AND agent_id=? AND kb_id=? AND chunk_id=?", t, u, a.AgentID, knowledge.KnowledgeBaseID, source.ID).Take(&row).Error; err != nil { return err } if row.Banned { return nil } if row.Status == "used" { allowed = row.ResourceID == a.ResourceID && row.SegmentID == a.SegmentID return nil } if row.Status == "reserved" && row.JobID != a.JobID && row.LeaseUntil != nil && row.LeaseUntil.After(now) { return nil } if err := tx.Model(&row).Updates(map[string]any{"status": "reserved", "resource_id": a.ResourceID, "segment_id": a.SegmentID, "job_id": a.JobID, "lease_until": until, "updated_at": now}).Error; err != nil { return err } allowed = true return nil }) return allowed, e } func finishConsumption(tx *gorm.DB, job *Job, success bool) error { state := "open" if success { state = "used" } return tx.Model(&Consumption{}).Where("tenant_id=? AND owner_id=? AND job_id=? AND status=?", job.TenantID, job.OwnerID, job.ID, "reserved").Updates(map[string]any{"status": state, "lease_until": nil, "updated_at": time.Now()}).Error } func (s *Service) releaseJobEvidence(ctx context.Context, job *Job) error { if !s.db.Migrator().HasTable(&Consumption{}) { return nil } return finishConsumption(s.db.WithContext(ctx), job, false) } func (s *Service) releaseExpiredEvidence(ctx context.Context) error { return s.db.WithContext(ctx).Model(&Consumption{}).Where("status=? AND lease_until<=?", "reserved", time.Now()).Updates(map[string]any{"status": "open", "lease_until": nil, "updated_at": time.Now()}).Error }