xinmi/internal/workbench/content_memory.go

package workbench

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/Tencent/WeKnora/internal/types"
	"github.com/google/uuid"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const contentMemorySourcePrefix = "内容任务/"

func contentMemorySource(resourceID, segmentID string) string {
	return contentMemorySourcePrefix + resourceID + "/内容段/" + segmentID
}

func contentMemoryAdvice(advice string) string {
	content := strings.Join(strings.Fields(advice), " ")
	content, _ = types.RedactSensitive(content)
	if types.IsMostlyRedacted(content) {
		return ""
	}
	return content
}

func currentContentFeedback(r *Resource, segments []Segment, candidate *PendingMemory) bool {
	for _, segment := range segments {
		if segment.Status == "accepted" && strings.TrimSpace(segment.Content) != "" &&
			segment.AgentID == candidate.AgentID && candidate.SourceRef == contentMemorySource(r.ID, segment.ID) &&
			candidate.Content != "" && candidate.Content == contentMemoryAdvice(segment.MemoryAdvice) {
			return true
		}
	}
	return false
}

// Candidate feedback is human-authored and remains untrusted until reviewed.
// Generated prose is kept in task revisions, never promoted into a user fact.
func (s *Service) contentFeedbackCandidate(ctx context.Context, r *Resource, segment Segment, snapshots map[string]AgentSnapshot) *PendingMemory {
	if s.memories == nil || s.agents == nil || r.ID == "" || segment.Status != "accepted" || strings.TrimSpace(segment.Content) == "" {
		return nil
	}
	tenant, owner, _, err := identity(ctx)
	if err != nil || tenant != r.TenantID || owner != r.OwnerID {
		return nil
	}
	content := contentMemoryAdvice(segment.MemoryAdvice)
	if content == "" {
		return nil
	}
	agent, err := s.agents.GetAgentByID(ctx, segment.AgentID)
	if err != nil || agent == nil || agent.TenantID != tenant || agent.Config.MemoryEnabled != nil && !*agent.Config.MemoryEnabled {
		return nil
	}
	if saved, ok := snapshots[segment.AgentID]; ok && saved.MemoryEnabled != nil && !*saved.MemoryEnabled {
		return nil
	}
	if !s.memories.MemoryAvailable(types.ApplyAgentMemoryPreference(ctx, agent.Config.MemoryEnabled)) {
		return nil
	}
	source := contentMemorySource(r.ID, segment.ID)
	key := compact(JSON{"kind": "content-feedback-v1", "tenant": tenant, "owner": owner, "agent": segment.AgentID, "source": source, "content": content})
	now := time.Now()
	return &PendingMemory{ID: uuid.NewSHA1(uuid.NameSpaceOID, []byte(key)).String(), TenantID: tenant,
		OwnerID: owner, AgentID: segment.AgentID, Content: content, SourceRef: source,
		Status: "pending", CreatedAt: now, UpdatedAt: now}
}

func syncContentFeedbackMemories(tx *gorm.DB, before, after *Resource) error {
	if before.Kind != "content-tasks" {
		return nil
	}
	oldData, err := decode[TaskData](before.Data)
	if err != nil {
		return ErrInvalid
	}
	newData, err := decode[TaskData](after.Data)
	if err != nil {
		return ErrInvalid
	}
	current := make(map[string]Segment, len(newData.Segments))
	for _, segment := range newData.Segments {
		current[segment.ID] = segment
	}
	var retired []string
	for _, old := range oldData.Segments {
		next, exists := current[old.ID]
		if !exists || next.Status != "accepted" || next.AgentID != old.AgentID || next.Content != old.Content || next.MemoryAdvice != old.MemoryAdvice {
			retired = append(retired, contentMemorySource(before.ID, old.ID))
		}
	}
	if len(retired) > 0 {
		if err := tx.Model(&PendingMemory{}).Where("tenant_id = ? AND owner_id = ? AND source_ref IN ? AND status = ?", before.TenantID, before.OwnerID, retired, "pending").Updates(map[string]any{"status": "invalidated", "updated_at": time.Now()}).Error; err != nil {
			return fmt.Errorf("保存内容记忆状态失败，请重试: %w", err)
		}
	}
	candidate := after.pendingMemory
	if candidate == nil {
		return nil
	}
	if candidate.TenantID != after.TenantID || candidate.OwnerID != after.OwnerID {
		return ErrForbidden
	}
	if !currentContentFeedback(after, newData.Segments, candidate) {
		return nil
	}
	// Reacceptance may restore an invalidated source, but never resurrect a
	// candidate the user explicitly accepted or rejected.
	if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "id"}},
		DoUpdates: clause.Assignments(map[string]any{"status": "pending", "updated_at": candidate.UpdatedAt}),
		Where:     clause.Where{Exprs: []clause.Expression{clause.Eq{Column: clause.Column{Table: clause.CurrentTable, Name: "status"}, Value: "invalidated"}}},
	}).Create(candidate).Error; err != nil {
		return fmt.Errorf("保存待审核记忆失败，请重试: %w", err)
	}
	return nil
}

func invalidateDeletedContentFeedback(tx *gorm.DB, r *Resource) error {
	if r.Kind != "content-tasks" {
		return nil
	}
	prefix := contentMemorySourcePrefix + r.ID + "/内容段/"
	return tx.Model(&PendingMemory{}).
		Where("tenant_id = ? AND owner_id = ? AND substr(source_ref, 1, ?) = ? AND status = ?", r.TenantID, r.OwnerID, len([]rune(prefix)), prefix, "pending").
		Updates(map[string]any{"status": "invalidated", "updated_at": time.Now()}).Error
}

// Match the resource-save lock order so review cannot adopt deleted or rewritten feedback.
func lockContentFeedbackSource(tx *gorm.DB, row *PendingMemory) error {
	if !strings.HasPrefix(row.SourceRef, contentMemorySourcePrefix) {
		return nil
	}
	parts := strings.SplitN(strings.TrimPrefix(row.SourceRef, contentMemorySourcePrefix), "/内容段/", 2)
	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
		return ErrInvalid
	}
	var source Resource
	err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
		Where("id = ? AND kind = ? AND tenant_id = ? AND owner_id = ?", parts[0], "content-tasks", row.TenantID, row.OwnerID).Take(&source).Error
	if err != nil {
		if err == gorm.ErrRecordNotFound {
			return fmt.Errorf("%w: 内容来源已删除，请刷新记忆列表", ErrConflict)
		}
		return err
	}
	data, err := decode[TaskData](source.Data)
	if err != nil {
		return ErrInvalid
	}
	if !currentContentFeedback(&source, data.Segments, row) {
		return fmt.Errorf("%w: 内容来源已更改，请刷新记忆列表", ErrConflict)
	}
	return nil
}