From 332d34c495c9b2149f8deb442101ae30b0ede9f1 Mon Sep 17 00:00:00 2001 From: Hz_ Date: Tue, 7 Jul 2026 16:32:11 +0800 Subject: [PATCH] fix(agent): save draft version on agent update (#16691) --- internal/dao/user.go | 13 +++ internal/dao/user_canvas_version.go | 130 ++++++++++++++++++++-- internal/dao/user_canvas_version_test.go | 2 + internal/entity/canvas.go | 1 + internal/service/agent.go | 66 +++++++++-- internal/service/agent_test.go | 134 +++++++++++++++++++++++ 6 files changed, 332 insertions(+), 14 deletions(-) diff --git a/internal/dao/user.go b/internal/dao/user.go index d55ea3f482..09e28ddc95 100644 --- a/internal/dao/user.go +++ b/internal/dao/user.go @@ -17,6 +17,8 @@ package dao import ( + "context" + "ragflow/internal/entity" ) @@ -52,6 +54,17 @@ func (dao *UserDAO) GetByTenantID(tenantID string) (*entity.User, error) { return &user, nil } +// GetNicknameByID returns a user's nickname by string id. +func (dao *UserDAO) GetNicknameByID(ctx context.Context, id string) (string, error) { + var nickname string + err := DB.WithContext(ctx). + Model(&entity.User{}). + Where("id = ?", id). + Select("nickname"). + Scan(&nickname).Error + return nickname, err +} + // GetByEmail get user by email func (dao *UserDAO) GetByEmail(email string) (*entity.User, error) { var user entity.User diff --git a/internal/dao/user_canvas_version.go b/internal/dao/user_canvas_version.go index f22ced4e00..b049290183 100644 --- a/internal/dao/user_canvas_version.go +++ b/internal/dao/user_canvas_version.go @@ -18,8 +18,10 @@ package dao import ( "errors" + "reflect" "gorm.io/gorm" + "gorm.io/gorm/clause" "ragflow/internal/entity" ) @@ -29,12 +31,20 @@ import ( var ErrUserCanvasVersionNotFound = errors.New("user_canvas_version: not found") // UserCanvasVersionDAO persists and queries UserCanvasVersion rows. -// -// One UserCanvasVersion row is created on every agent publish (§2.9); rows -// are append-only and never updated. Cascade delete of a parent canvas -// removes all child versions via DeleteByCanvasID. type UserCanvasVersionDAO struct{} +// SaveOrReplaceLatestVersionOptions controls a version-history save. +type SaveOrReplaceLatestVersionOptions struct { + NewID string + UserCanvasID string + Title *string + Description *string + DSL entity.JSONMap + Release bool + KeepUnpublished int + SameDSL func(entity.JSONMap) bool +} + // NewUserCanvasVersionDAO returns a zero-value DAO. The struct is stateless // so callers can share a single instance or create their own. func NewUserCanvasVersionDAO() *UserCanvasVersionDAO { @@ -117,9 +127,115 @@ func (dao *UserCanvasVersionDAO) DeleteByCanvasIDTx(tx *gorm.DB, canvasID string return res.RowsAffected, res.Error } -// CreateTx is the transactional variant of Create. Used by -// service.AgentService.PublishAgent so the new version row and the -// parent canvas update land in one atomic write. +// CreateTx is the transactional variant of Create. func (dao *UserCanvasVersionDAO) CreateTx(tx *gorm.DB, v *entity.UserCanvasVersion) error { return tx.Create(v).Error } + +// SaveOrReplaceLatest inserts a new version or refreshes the latest matching +// draft in place. If the latest matching version is released and the current +// save is a draft, it creates a new draft to preserve the released snapshot. +func (dao *UserCanvasVersionDAO) SaveOrReplaceLatest(opts SaveOrReplaceLatestVersionOptions) (*entity.UserCanvasVersion, error) { + if opts.KeepUnpublished <= 0 { + opts.KeepUnpublished = 20 + } + var saved *entity.UserCanvasVersion + if err := DB.Transaction(func(tx *gorm.DB) error { + var parent struct { + ID string + } + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Table((&entity.UserCanvas{}).TableName()). + Select("id"). + Where("id = ?", opts.UserCanvasID). + Take(&parent).Error; err != nil { + return err + } + + var latest entity.UserCanvasVersion + err := tx.Where("user_canvas_id = ?", opts.UserCanvasID). + Order("create_time DESC, id DESC"). + First(&latest).Error + if err != nil { + if !errors.Is(err, gorm.ErrRecordNotFound) { + return err + } + } else if opts.sameDSL(latest.DSL) { + if !latest.Release || opts.Release { + updates := map[string]interface{}{ + "dsl": opts.DSL, + "release": opts.Release, + } + if opts.Title != nil { + updates["title"] = opts.Title + } + if opts.Description != nil { + updates["description"] = opts.Description + } + if err := tx.Model(&entity.UserCanvasVersion{}). + Where("id = ?", latest.ID). + Updates(updates).Error; err != nil { + return err + } + latest.DSL = opts.DSL + latest.Release = opts.Release + if opts.Title != nil { + latest.Title = opts.Title + } + if opts.Description != nil { + latest.Description = opts.Description + } + saved = &latest + return dao.deleteAllUnpublishedExcessTx(tx, opts.UserCanvasID, opts.KeepUnpublished) + } + } + row := &entity.UserCanvasVersion{ + ID: opts.NewID, + UserCanvasID: opts.UserCanvasID, + Title: opts.Title, + Description: opts.Description, + Release: opts.Release, + DSL: opts.DSL, + } + if err := tx.Create(row).Error; err != nil { + return err + } + saved = row + return dao.deleteAllUnpublishedExcessTx(tx, opts.UserCanvasID, opts.KeepUnpublished) + }); err != nil { + return nil, err + } + return saved, nil +} + +func (opts SaveOrReplaceLatestVersionOptions) sameDSL(dsl entity.JSONMap) bool { + if opts.SameDSL != nil { + return opts.SameDSL(dsl) + } + return reflect.DeepEqual(dsl, opts.DSL) +} + +// DeleteAllUnpublishedExcess keeps the newest keep unpublished versions for a +// canvas and deletes older unpublished rows. Released versions are never +// removed by this cleanup. +func (dao *UserCanvasVersionDAO) DeleteAllUnpublishedExcess(canvasID string, keep int) error { + return dao.deleteAllUnpublishedExcessTx(DB, canvasID, keep) +} + +func (dao *UserCanvasVersionDAO) deleteAllUnpublishedExcessTx(tx *gorm.DB, canvasID string, keep int) error { + if keep < 0 { + keep = 0 + } + var ids []string + if err := tx.Model(&entity.UserCanvasVersion{}). + Where(map[string]interface{}{"user_canvas_id": canvasID, "release": false}). + Order("create_time DESC"). + Pluck("id", &ids).Error; err != nil { + return err + } + if len(ids) <= keep { + return nil + } + ids = ids[keep:] + return tx.Where("id IN ?", ids).Delete(&entity.UserCanvasVersion{}).Error +} diff --git a/internal/dao/user_canvas_version_test.go b/internal/dao/user_canvas_version_test.go index fb5f498f89..454e0bf6fe 100644 --- a/internal/dao/user_canvas_version_test.go +++ b/internal/dao/user_canvas_version_test.go @@ -65,8 +65,10 @@ type daoInterface interface { GetByID(id string) (*entity.UserCanvasVersion, error) ListByCanvasID(canvasID string) ([]*entity.UserCanvasVersion, error) GetLatest(canvasID string) (*entity.UserCanvasVersion, error) + SaveOrReplaceLatest(opts SaveOrReplaceLatestVersionOptions) (*entity.UserCanvasVersion, error) Delete(id string) error DeleteByCanvasID(canvasID string) (int64, error) + DeleteAllUnpublishedExcess(canvasID string, keep int) error } var _ daoInterface = (*UserCanvasVersionDAO)(nil) diff --git a/internal/entity/canvas.go b/internal/entity/canvas.go index af390fd6a6..f98f2d271f 100644 --- a/internal/entity/canvas.go +++ b/internal/entity/canvas.go @@ -61,6 +61,7 @@ type UserCanvasVersion struct { UserCanvasID string `gorm:"column:user_canvas_id;size:255;not null;index" json:"user_canvas_id"` Title *string `gorm:"column:title;size:255" json:"title,omitempty"` Description *string `gorm:"column:description;type:longtext" json:"description,omitempty"` + Release bool `gorm:"column:release;not null;default:false;index" json:"release"` DSL JSONMap `gorm:"column:dsl;type:longtext" json:"dsl,omitempty"` BaseModel } diff --git a/internal/service/agent.go b/internal/service/agent.go index 146708a94e..d83071642a 100644 --- a/internal/service/agent.go +++ b/internal/service/agent.go @@ -21,6 +21,7 @@ import ( "encoding/json" "errors" "fmt" + "reflect" "strings" "sync" "time" @@ -125,6 +126,7 @@ var ErrAgentStorageError = errors.New("agent storage error") type AgentService struct { canvasDAO *dao.UserCanvasDAO canvasTemplateDAO *dao.CanvasTemplateDAO + userDAO *dao.UserDAO userTenantDAO *dao.UserTenantDAO versionDAO *dao.UserCanvasVersionDAO api4ConversationDAO *dao.API4ConversationDAO @@ -181,6 +183,7 @@ func NewAgentServiceWithOptions( return &AgentService{ canvasDAO: dao.NewUserCanvasDAO(), canvasTemplateDAO: dao.NewCanvasTemplateDAO(), + userDAO: dao.NewUserDAO(), userTenantDAO: dao.NewUserTenantDAO(), versionDAO: dao.NewUserCanvasVersionDAO(), api4ConversationDAO: dao.NewAPI4ConversationDAO(), @@ -422,7 +425,8 @@ func (s *AgentService) GetAgent(ctx context.Context, userID, canvasID string) (* // UpdateAgent applies a draft patch to user_canvas. Settings updates may omit // dsl; in that case the existing draft DSL must be preserved. func (s *AgentService) UpdateAgent(ctx context.Context, userID, canvasID string, patch map[string]interface{}) error { - if _, err := s.loadCanvasForUser(ctx, userID, canvasID); err != nil { + canvasInstance, err := s.loadCanvasForUser(ctx, userID, canvasID) + if err != nil { return err } @@ -451,10 +455,25 @@ func (s *AgentService) UpdateAgent(ctx context.Context, userID, canvasID string, updates["dsl"] = entity.JSONMap(dslpkg.NormalizeForCanvas(entity.JSONMap(dslMap))) } - _, err := s.canvasDAO.UpdateFields(canvasID, updates) + _, err = s.canvasDAO.UpdateFields(canvasID, updates) if err != nil { return fmt.Errorf("update agent %s: %w", canvasID, err) } + if dslValue, ok := updates["dsl"]; ok { + dsl, ok := dslValue.(entity.JSONMap) + if !ok { + return fmt.Errorf("update agent %s: normalized dsl must be an object", canvasID) + } + title := "" + if value, ok := updates["title"]; ok { + title, _ = value.(string) + } else if canvasInstance.Title != nil { + title = *canvasInstance.Title + } + if _, err := s.saveOrReplaceVersion(ctx, userID, canvasID, dsl, title, nil, false); err != nil { + return fmt.Errorf("update agent %s: save version: %w", canvasID, err) + } + } return nil } @@ -466,11 +485,9 @@ func (s *AgentService) UpdateAgent(ctx context.Context, userID, canvasID string, // returned so the caller can render it back to the client without an // extra GET. // -// Reset does NOT create a new user_canvas_version row — that mirrors -// the Python behavior and UpdateAgent: versions are owned by -// PublishAgent. It also does NOT touch the in-flight run state of any -// currently executing canvas session; that is owned by the Python task -// executor and is out of scope for the Go port. +// Reset does NOT create a new user_canvas_version row. It also does NOT touch +// the in-flight run state of any currently executing canvas session; that is +// owned by the Python task executor and is out of scope for the Go port. // // Errors propagate the same way as GetAgent: a missing canvas, or a // canvas that the user has no access to, surfaces as @@ -582,6 +599,41 @@ func (s *AgentService) PublishAgent(ctx context.Context, userID, canvasID string return row, nil } +func (s *AgentService) saveOrReplaceVersion(ctx context.Context, userID, canvasID string, dsl entity.JSONMap, title string, description *string, release bool) (*entity.UserCanvasVersion, error) { + nickname, err := s.userDAO.GetNicknameByID(ctx, userID) + if err != nil || strings.TrimSpace(nickname) == "" { + nickname = userID + } + versionTitle := buildVersionTitle(nickname, title, time.Now()) + return s.versionDAO.SaveOrReplaceLatest(dao.SaveOrReplaceLatestVersionOptions{ + NewID: genID32(), + UserCanvasID: canvasID, + Title: &versionTitle, + Description: description, + DSL: dsl, + Release: release, + KeepUnpublished: 20, + SameDSL: func(latestDSL entity.JSONMap) bool { + return reflect.DeepEqual( + entity.JSONMap(dslpkg.NormalizeForCanvas(latestDSL)), + dsl, + ) + }, + }) +} + +func buildVersionTitle(userNickname, agentTitle string, ts time.Time) string { + tenant := strings.TrimSpace(userNickname) + if tenant == "" { + tenant = "tenant" + } + title := strings.TrimSpace(agentTitle) + if title == "" { + title = "agent" + } + return fmt.Sprintf("%s_%s_%s", tenant, title, ts.Format("2006-01-02 15:04:05")) +} + // ListVersions returns every version for a canvas the user can see, // newest first. The parent-canvas access check is enforced before the // version list is loaded so unauthorized users cannot enumerate version diff --git a/internal/service/agent_test.go b/internal/service/agent_test.go index 37c1a8142b..be180e6f24 100644 --- a/internal/service/agent_test.go +++ b/internal/service/agent_test.go @@ -584,9 +584,15 @@ func setupAgentSessionServiceTest(t *testing.T) { t.Helper() testDB := setupServiceTestDB(t) + sqlDB, err := testDB.DB() + if err != nil { + t.Fatalf("failed to access sqlite handle: %v", err) + } + sqlDB.SetMaxOpenConns(1) if err := testDB.AutoMigrate( &entity.User{}, &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, &entity.UserTenant{}, &entity.API4Conversation{}, ); err != nil { @@ -1272,6 +1278,134 @@ func TestUpdateAgentPersistsDSLAsJSONMap(t *testing.T) { } } +func TestUpdateAgentDSLCreatesAndReplacesDraftVersion(t *testing.T) { + setupAgentSessionServiceTest(t) + + if err := dao.DB.Create(&entity.User{ID: "user-1", Nickname: "owner", Email: "owner@test.com"}).Error; err != nil { + t.Fatalf("failed to seed user: %v", err) + } + if err := dao.DB.Create(&entity.UserCanvas{ + ID: "canvas-version-draft", + UserID: "user-1", + Title: sptr("Draft Agent"), + CanvasCategory: "agent_canvas", + DSL: entity.JSONMap{}, + }).Error; err != nil { + t.Fatalf("failed to seed canvas: %v", err) + } + + patch := map[string]interface{}{ + "title": "Draft Agent", + "dsl": map[string]interface{}{ + "graph": map[string]interface{}{ + "nodes": []interface{}{map[string]interface{}{"id": "begin"}}, + "edges": []interface{}{}, + }, + "components": map[string]interface{}{ + "begin": map[string]interface{}{ + "obj": map[string]interface{}{"component_name": "Begin"}, + }, + }, + }, + } + if err := NewAgentService().UpdateAgent(context.Background(), "user-1", "canvas-version-draft", patch); err != nil { + t.Fatalf("first UpdateAgent failed: %v", err) + } + secondPatch := map[string]interface{}{ + "title": "Renamed Agent", + "dsl": patch["dsl"], + } + if err := NewAgentService().UpdateAgent(context.Background(), "user-1", "canvas-version-draft", secondPatch); err != nil { + t.Fatalf("second UpdateAgent failed: %v", err) + } + + versions, err := dao.NewUserCanvasVersionDAO().ListByCanvasID("canvas-version-draft") + if err != nil { + t.Fatalf("failed to list versions: %v", err) + } + if len(versions) != 1 { + t.Fatalf("expected same DSL to replace latest draft, got %d versions", len(versions)) + } + if versions[0].Title == nil || !strings.HasPrefix(*versions[0].Title, "owner_Renamed Agent_") { + t.Fatalf("unexpected version title: %v", versions[0].Title) + } + var release bool + if err := dao.DB.Table("user_canvas_version").Select("release").Where("id = ?", versions[0].ID).Scan(&release).Error; err != nil { + t.Fatalf("failed to read release flag: %v", err) + } + if release { + t.Fatal("draft update saved a released version") + } +} + +func TestUpdateAgentDSLDoesNotOverwriteLatestReleasedVersion(t *testing.T) { + setupAgentSessionServiceTest(t) + + if err := dao.DB.Create(&entity.User{ID: "user-1", Nickname: "owner", Email: "owner@test.com"}).Error; err != nil { + t.Fatalf("failed to seed user: %v", err) + } + dsl := entity.JSONMap{ + "graph": map[string]any{ + "nodes": []any{map[string]any{"id": "begin"}}, + "edges": []any{}, + }, + "components": map[string]any{ + "begin": map[string]any{ + "obj": map[string]any{"component_name": "Begin"}, + }, + }, + } + if err := dao.DB.Create(&entity.UserCanvas{ + ID: "canvas-released-latest", + UserID: "user-1", + Title: sptr("Released Agent"), + CanvasCategory: "agent_canvas", + DSL: dsl, + }).Error; err != nil { + t.Fatalf("failed to seed canvas: %v", err) + } + releasedAt := time.Now().Add(-time.Minute) + if err := dao.DB.Create(&entity.UserCanvasVersion{ + ID: "released-version", + UserCanvasID: "canvas-released-latest", + Title: sptr("released"), + Release: true, + DSL: dsl, + BaseModel: entity.BaseModel{ + CreateTime: ptr(releasedAt.UnixMilli()), + UpdateTime: ptr(releasedAt.UnixMilli()), + }, + }).Error; err != nil { + t.Fatalf("failed to seed released version: %v", err) + } + + if err := NewAgentService().UpdateAgent(context.Background(), "user-1", "canvas-released-latest", map[string]interface{}{"dsl": map[string]interface{}(dsl)}); err != nil { + t.Fatalf("UpdateAgent failed: %v", err) + } + + versions, err := dao.NewUserCanvasVersionDAO().ListByCanvasID("canvas-released-latest") + if err != nil { + t.Fatalf("failed to list versions: %v", err) + } + if len(versions) != 2 { + t.Fatalf("expected draft save to create a new version beside the released one, got %d", len(versions)) + } + var releasedCount int64 + if err := dao.DB.Table("user_canvas_version").Where("user_canvas_id = ? AND release = ?", "canvas-released-latest", true).Count(&releasedCount).Error; err != nil { + t.Fatalf("failed to count released versions: %v", err) + } + if releasedCount != 1 { + t.Fatalf("released version count = %d, want 1", releasedCount) + } + var draftCount int64 + if err := dao.DB.Table("user_canvas_version").Where("user_canvas_id = ? AND release = ?", "canvas-released-latest", false).Count(&draftCount).Error; err != nil { + t.Fatalf("failed to count draft versions: %v", err) + } + if draftCount != 1 { + t.Fatalf("draft version count = %d, want 1", draftCount) + } +} + // TestResetAgentServiceOtherTenant asserts the access-denied path: // a canvas owned by user-2 is not visible to user-1, so the same // not-found error type is returned. The service layer does not