diff --git a/backend/docs/docs.go b/backend/docs/docs.go index c463867e2..1e68b4ab8 100644 --- a/backend/docs/docs.go +++ b/backend/docs/docs.go @@ -4847,7 +4847,7 @@ const docTemplate = `{ "BearerAuth": [] } ], - "description": "调用上游 models 接口，仅返回可导入预览，不直接落库", + "description": "调用上游 models 接口，返回可导入模型与目录变更预览，不直接落库", "consumes": [ "application/json" ], @@ -4908,7 +4908,7 @@ const docTemplate = `{ "BearerAuth": [] } ], - "description": "调用上游 models 接口写入上游真实模型清单，不自动绑定平台模型", + "description": "调用上游 models 接口获取完整目录，原子更新远端管理模型可用状态，不删除平台模型或路由配置", "consumes": [ "application/json" ], @@ -4926,6 +4926,18 @@ const docTemplate = `{ "name": "id", "in": "path", "required": true + }, + { + "type": "boolean", + "description": "确认允许空模型目录对账", + "name": "allow_empty", + "in": "query" + }, + { + "type": "string", + "description": "用户确认的远端目录快照标识", + "name": "expected_snapshot", + "in": "query" } ], "responses": { @@ -4947,6 +4959,12 @@ const docTemplate = `{ "$ref": "#/definitions/ChannelErrorDoc" } }, + "409": { + "description": "Conflict", + "schema": { + "$ref": "#/definitions/ChannelErrorDoc" + } + }, "500": { "description": "Internal Server Error", "schema": { @@ -25162,9 +25180,14 @@ const docTemplate = `{ "createdUpstreamModels", "existingUpstreamModels", "inactivatedModels", + "protectedUpstreamModels", + "reactivatedModels", "skippedUpstreamModels", + "snapshotID", "syncedModels", - "totalUpstream" + "totalUpstream", + "unchangedUpstreamModels", + "updatedUpstreamModels" ], "properties": { "createdUpstreamModels": { @@ -25176,9 +25199,18 @@ const docTemplate = `{ "inactivatedModels": { "type": "integer" }, + "protectedUpstreamModels": { + "type": "integer" + }, + "reactivatedModels": { + "type": "integer" + }, "skippedUpstreamModels": { "type": "integer" }, + "snapshotID": { + "type": "string" + }, "syncedModels": { "type": "array", "items": { @@ -25187,6 +25219,12 @@ const docTemplate = `{ }, "totalUpstream": { "type": "integer" + }, + "unchangedUpstreamModels": { + "type": "integer" + }, + "updatedUpstreamModels": { + "type": "integer" } } }, @@ -26623,6 +26661,55 @@ const docTemplate = `{ } } }, + "UpstreamModelSyncPlanResponse": { + "type": "object", + "required": [ + "addedModels", + "inactivatedModels", + "protectedModels", + "reactivatedModels", + "unchangedModels", + "updatedModels" + ], + "properties": { + "addedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "inactivatedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "protectedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "reactivatedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "unchangedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "updatedModels": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, "UpstreamRemoteModelResponse": { "type": "object", "required": [ @@ -26680,6 +26767,8 @@ const docTemplate = `{ "type": "object", "required": [ "items", + "snapshotID", + "syncPlan", "total" ], "properties": { @@ -26689,6 +26778,12 @@ const docTemplate = `{ "$ref": "#/definitions/UpstreamRemoteModelResponse" } }, + "snapshotID": { + "type": "string" + }, + "syncPlan": { + "$ref": "#/definitions/UpstreamModelSyncPlanResponse" + }, "total": { "type": "integer" } @@ -26817,8 +26912,11 @@ const docTemplate = `{ "bindingCode", "created", "kindsJSON", + "protected", + "reactivated", "status", "suggestedProtocol", + "updated", "upstreamModelName" ], "properties": { @@ -26831,12 +26929,21 @@ const docTemplate = `{ "kindsJSON": { "type": "string" }, + "protected": { + "type": "boolean" + }, + "reactivated": { + "type": "boolean" + }, "status": { "type": "string" }, "suggestedProtocol": { "type": "string" }, + "updated": { + "type": "boolean" + }, "upstreamModelName": { "type": "string" } diff --git a/backend/docs/swagger.json b/backend/docs/swagger.json index 7ec135953..01c6044a5 100644 --- a/backend/docs/swagger.json +++ b/backend/docs/swagger.json @@ -4840,7 +4840,7 @@ "BearerAuth": [] } ], - "description": "调用上游 models 接口，仅返回可导入预览，不直接落库", + "description": "调用上游 models 接口，返回可导入模型与目录变更预览，不直接落库", "consumes": [ "application/json" ], @@ -4901,7 +4901,7 @@ "BearerAuth": [] } ], - "description": "调用上游 models 接口写入上游真实模型清单，不自动绑定平台模型", + "description": "调用上游 models 接口获取完整目录，原子更新远端管理模型可用状态，不删除平台模型或路由配置", "consumes": [ "application/json" ], @@ -4919,6 +4919,18 @@ "name": "id", "in": "path", "required": true + }, + { + "type": "boolean", + "description": "确认允许空模型目录对账", + "name": "allow_empty", + "in": "query" + }, + { + "type": "string", + "description": "用户确认的远端目录快照标识", + "name": "expected_snapshot", + "in": "query" } ], "responses": { @@ -4940,6 +4952,12 @@ "$ref": "#/definitions/ChannelErrorDoc" } }, + "409": { + "description": "Conflict", + "schema": { + "$ref": "#/definitions/ChannelErrorDoc" + } + }, "500": { "description": "Internal Server Error", "schema": { @@ -25155,9 +25173,14 @@ "createdUpstreamModels", "existingUpstreamModels", "inactivatedModels", + "protectedUpstreamModels", + "reactivatedModels", "skippedUpstreamModels", + "snapshotID", "syncedModels", - "totalUpstream" + "totalUpstream", + "unchangedUpstreamModels", + "updatedUpstreamModels" ], "properties": { "createdUpstreamModels": { @@ -25169,9 +25192,18 @@ "inactivatedModels": { "type": "integer" }, + "protectedUpstreamModels": { + "type": "integer" + }, + "reactivatedModels": { + "type": "integer" + }, "skippedUpstreamModels": { "type": "integer" }, + "snapshotID": { + "type": "string" + }, "syncedModels": { "type": "array", "items": { @@ -25180,6 +25212,12 @@ }, "totalUpstream": { "type": "integer" + }, + "unchangedUpstreamModels": { + "type": "integer" + }, + "updatedUpstreamModels": { + "type": "integer" } } }, @@ -26616,6 +26654,55 @@ } } }, + "UpstreamModelSyncPlanResponse": { + "type": "object", + "required": [ + "addedModels", + "inactivatedModels", + "protectedModels", + "reactivatedModels", + "unchangedModels", + "updatedModels" + ], + "properties": { + "addedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "inactivatedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "protectedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "reactivatedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "unchangedModels": { + "type": "array", + "items": { + "type": "string" + } + }, + "updatedModels": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, "UpstreamRemoteModelResponse": { "type": "object", "required": [ @@ -26673,6 +26760,8 @@ "type": "object", "required": [ "items", + "snapshotID", + "syncPlan", "total" ], "properties": { @@ -26682,6 +26771,12 @@ "$ref": "#/definitions/UpstreamRemoteModelResponse" } }, + "snapshotID": { + "type": "string" + }, + "syncPlan": { + "$ref": "#/definitions/UpstreamModelSyncPlanResponse" + }, "total": { "type": "integer" } @@ -26810,8 +26905,11 @@ "bindingCode", "created", "kindsJSON", + "protected", + "reactivated", "status", "suggestedProtocol", + "updated", "upstreamModelName" ], "properties": { @@ -26824,12 +26922,21 @@ "kindsJSON": { "type": "string" }, + "protected": { + "type": "boolean" + }, + "reactivated": { + "type": "boolean" + }, "status": { "type": "string" }, "suggestedProtocol": { "type": "string" }, + "updated": { + "type": "boolean" + }, "upstreamModelName": { "type": "string" } diff --git a/backend/docs/swagger.yaml b/backend/docs/swagger.yaml index a65a42e50..a9afc1bdf 100644 --- a/backend/docs/swagger.yaml +++ b/backend/docs/swagger.yaml @@ -7591,21 +7591,36 @@ definitions: type: integer inactivatedModels: type: integer + protectedUpstreamModels: + type: integer + reactivatedModels: + type: integer skippedUpstreamModels: type: integer + snapshotID: + type: string syncedModels: items: $ref: '#/definitions/UpstreamSyncModelResponse' type: array totalUpstream: type: integer + unchangedUpstreamModels: + type: integer + updatedUpstreamModels: + type: integer required: - createdUpstreamModels - existingUpstreamModels - inactivatedModels + - protectedUpstreamModels + - reactivatedModels - skippedUpstreamModels + - snapshotID - syncedModels - totalUpstream + - unchangedUpstreamModels + - updatedUpstreamModels type: object SyncUpstreamModelsResponseDoc: properties: @@ -8628,6 +8643,40 @@ definitions: - upstreamModelVendor - weight type: object + UpstreamModelSyncPlanResponse: + properties: + addedModels: + items: + type: string + type: array + inactivatedModels: + items: + type: string + type: array + protectedModels: + items: + type: string + type: array + reactivatedModels: + items: + type: string + type: array + unchangedModels: + items: + type: string + type: array + updatedModels: + items: + type: string + type: array + required: + - addedModels + - inactivatedModels + - protectedModels + - reactivatedModels + - unchangedModels + - updatedModels + type: object UpstreamRemoteModelResponse: properties: alreadyBound: @@ -8672,10 +8721,16 @@ definitions: items: $ref: '#/definitions/UpstreamRemoteModelResponse' type: array + snapshotID: + type: string + syncPlan: + $ref: '#/definitions/UpstreamModelSyncPlanResponse' total: type: integer required: - items + - snapshotID + - syncPlan - total type: object UpstreamRemoteModelsResponseDoc: @@ -8771,18 +8826,27 @@ definitions: type: boolean kindsJSON: type: string + protected: + type: boolean + reactivated: + type: boolean status: type: string suggestedProtocol: type: string + updated: + type: boolean upstreamModelName: type: string required: - bindingCode - created - kindsJSON + - protected + - reactivated - status - suggestedProtocol + - updated - upstreamModelName type: object UsageDailyListResponseDoc: @@ -12931,7 +12995,7 @@ paths: get: consumes: - application/json - description: 调用上游 models 接口，仅返回可导入预览，不直接落库 + description: 调用上游 models 接口，返回可导入模型与目录变更预览，不直接落库 parameters: - description: 上游ID in: path @@ -12970,13 +13034,21 @@ paths: post: consumes: - application/json - description: 调用上游 models 接口写入上游真实模型清单，不自动绑定平台模型 + description: 调用上游 models 接口获取完整目录，原子更新远端管理模型可用状态，不删除平台模型或路由配置 parameters: - description: 上游ID in: path name: id required: true type: integer + - description: 确认允许空模型目录对账 + in: query + name: allow_empty + type: boolean + - description: 用户确认的远端目录快照标识 + in: query + name: expected_snapshot + type: string produces: - application/json responses: @@ -12992,6 +13064,10 @@ paths: description: Not Found schema: $ref: '#/definitions/ChannelErrorDoc' + "409": + description: Conflict + schema: + $ref: '#/definitions/ChannelErrorDoc' "500": description: Internal Server Error schema: diff --git a/backend/internal/application/channel/dto.go b/backend/internal/application/channel/dto.go index df63f3fe7..4be57dd95 100644 --- a/backend/internal/application/channel/dto.go +++ b/backend/internal/application/channel/dto.go @@ -33,8 +33,20 @@ type BatchDeleteData struct { // UpstreamRemoteModelsData 上游远程模型预览响应数据（内部传输，不携带序列化标记）。 type UpstreamRemoteModelsData struct { - Total int - Items []UpstreamRemoteModelView + Total int + Items []UpstreamRemoteModelView + SnapshotID string + SyncPlan UpstreamModelSyncPlanView +} + +// UpstreamModelSyncPlanView 描述应用指定远端快照时将发生的目录变化。 +type UpstreamModelSyncPlanView struct { + AddedModels []string + UpdatedModels []string + ReactivatedModels []string + InactivatedModels []string + UnchangedModels []string + ProtectedModels []string } // UpstreamRemoteModelView 上游远程模型预览项（内部传输，不携带序列化标记）。 @@ -53,12 +65,19 @@ type UpstreamRemoteModelView struct { // SyncUpstreamModelsData 同步上游模型响应数据（内部传输，不携带序列化标记）。 type SyncUpstreamModelsData struct { - TotalUpstream int - CreatedUpstreamModels int + SnapshotID string + TotalUpstream int + CreatedUpstreamModels int + UpdatedUpstreamModels int + UnchangedUpstreamModels int + ProtectedUpstreamModels int + // ExistingUpstreamModels 保留旧版响应语义，表示远端目录中未新增的模型总数。 ExistingUpstreamModels int - SkippedUpstreamModels int - InactivatedModels int64 - SyncedModels []UpstreamSyncModelView + // SkippedUpstreamModels 保留旧版响应字段；原子对账失败时整体回滚，因此始终为 0。 + SkippedUpstreamModels int + InactivatedModels int64 + ReactivatedModels int + SyncedModels []UpstreamSyncModelView } // UpstreamSyncModelView 单个同步结果（内部传输，不携带序列化标记）。 @@ -69,6 +88,9 @@ type UpstreamSyncModelView struct { KindsJSON string Status string Created bool + Updated bool + Reactivated bool + Protected bool } // ImportUpstreamModelsData 批量导入上游模型响应数据（内部传输，不携带序列化标记）。 diff --git a/backend/internal/application/channel/errs.go b/backend/internal/application/channel/errs.go index 60f80b997..662e65a09 100644 --- a/backend/internal/application/channel/errs.go +++ b/backend/internal/application/channel/errs.go @@ -96,6 +96,10 @@ var ( ErrUpstreamSourceUnavailable = errors.New("upstream source unavailable") // ErrRemoteModelsUnavailable 上游远程模型目录不可用。 ErrRemoteModelsUnavailable = errors.New("remote models unavailable") + // ErrEmptyRemoteModels 上游返回空模型目录，必须显式确认后才允许对账。 + ErrEmptyRemoteModels = errors.New("remote models snapshot is empty") + // ErrRemoteModelsSnapshotChanged 表示确认后远端目录已变化，必须重新预览。 + ErrRemoteModelsSnapshotChanged = errors.New("remote models snapshot changed") // ErrNoActiveKey 无可用密钥。 ErrNoActiveKey = errors.New("no active api key") // ErrLLMSettingNotFound LLM 全局设置不存在。 diff --git a/backend/internal/application/channel/input.go b/backend/internal/application/channel/input.go index 4ebead8e0..31c87751b 100644 --- a/backend/internal/application/channel/input.go +++ b/backend/internal/application/channel/input.go @@ -119,13 +119,20 @@ type UpsertUpstreamModelInput struct { Status *string Priority *int Weight *int - Source *string + Source *string // 路由配置来源。 + CatalogSource *string // 上游目录生命周期来源；为空时沿用 Source。 CbFailureThreshold *int CbDurationMin *int CbWindowMin *int HeadersJSON *string } +// SyncUpstreamModelsInput 定义远端模型目录对账约束。 +type SyncUpstreamModelsInput struct { + ExpectedSnapshot string + AllowEmpty bool +} + // UpdateModelUpstreamSourceInput 定义更新模型来源入参。 type UpdateModelUpstreamSourceInput struct { Protocol *string diff --git a/backend/internal/application/channel/service_model_update_test.go b/backend/internal/application/channel/service_model_update_test.go index 770d9dbf2..daf6df81d 100644 --- a/backend/internal/application/channel/service_model_update_test.go +++ b/backend/internal/application/channel/service_model_update_test.go @@ -9,6 +9,7 @@ import ( domainchannel "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/channel" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/cache/memory" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/config" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/ports/llm" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" ) @@ -632,6 +633,129 @@ func modelProtocolSource(routeID uint, upstreamID uint, upstreamModelID uint, pr } } +func TestReconcileRemoteModelSnapshotSoftlyReconcilesManagedCatalog(t *testing.T) { + repo := &modelUpdateRepo{upstreamModels: map[string]domainchannel.UpstreamModel{ + "returning": { + ID: 1, UpstreamID: 9, BindingCode: "returning-code", UpstreamModelName: "returning", Status: "inactive", Source: "import", + }, + "manual-model": { + ID: 2, UpstreamID: 9, BindingCode: "manual-code", UpstreamModelName: "manual-model", Status: "inactive", Source: "manual", + }, + "removed": { + ID: 3, UpstreamID: 9, BindingCode: "removed-code", UpstreamModelName: "removed", Status: "active", Source: "sync", + }, + }} + service := NewService(config.Config{}, repo, repo, nil, nil) + result, err := service.reconcileRemoteModelSnapshot(t.Context(), &domainchannel.Upstream{ + ID: 9, Name: "test", Compatible: "openai", BaseURL: "https://example.com", + }, []llm.ModelItem{ + {ID: " new-model ", OwnedBy: "openai"}, + {ID: "returning", OwnedBy: "openai"}, + {ID: "manual-model", OwnedBy: "custom"}, + {ID: "new-model", OwnedBy: "duplicate"}, + {ID: " "}, + }, false) + if err != nil { + t.Fatalf("reconcile snapshot: %v", err) + } + if result.TotalUpstream != 3 || result.CreatedUpstreamModels != 1 || result.ExistingUpstreamModels != 2 { + t.Fatalf("unexpected sync counts: %+v", result) + } + if result.ReactivatedModels != 1 || result.ProtectedUpstreamModels != 1 || result.InactivatedModels != 1 { + t.Fatalf("unexpected availability counts: %+v", result) + } + if result.UpdatedUpstreamModels != 0 || result.UnchangedUpstreamModels != 0 { + t.Fatalf("expected exclusive catalog categories, got %+v", result) + } + if categorized := result.CreatedUpstreamModels + result.UpdatedUpstreamModels + result.ReactivatedModels + result.UnchangedUpstreamModels + result.ProtectedUpstreamModels; categorized != result.TotalUpstream { + t.Fatalf("categorized remote models = %d, want %d", categorized, result.TotalUpstream) + } + if got := repo.upstreamModels["returning"]; got.Status != "active" || got.Source != "sync" { + t.Fatalf("expected legacy imported model to be restored and migrated, got %+v", got) + } + if got := repo.upstreamModels["manual-model"]; got.Status != "inactive" || got.Source != "manual" { + t.Fatalf("expected manual model to remain untouched, got %+v", got) + } + if got := repo.upstreamModels["removed"]; got.Status != "inactive" { + t.Fatalf("expected missing managed model to be inactive, got %+v", got) + } +} + +func TestReconcileRemoteModelSnapshotRequiresConfirmationForEmptyCatalog(t *testing.T) { + repo := &modelUpdateRepo{upstreamModels: map[string]domainchannel.UpstreamModel{ + "existing": {ID: 1, UpstreamID: 9, UpstreamModelName: "existing", Status: "active", Source: "sync"}, + }} + service := NewService(config.Config{}, repo, repo, nil, nil) + upstream := &domainchannel.Upstream{ID: 9} + + if _, err := service.reconcileRemoteModelSnapshot(t.Context(), upstream, nil, false); !errors.Is(err, ErrEmptyRemoteModels) { + t.Fatalf("expected empty snapshot error, got %v", err) + } + if repo.catalogApplyCalls != 0 { + t.Fatalf("empty snapshot changed data without confirmation") + } + + result, err := service.reconcileRemoteModelSnapshot(t.Context(), upstream, nil, true) + if err != nil { + t.Fatalf("confirmed empty snapshot: %v", err) + } + if result.InactivatedModels != 1 || repo.upstreamModels["existing"].Status != "inactive" { + t.Fatalf("expected confirmed empty snapshot to deactivate managed catalog, got %+v", result) + } +} + +func TestBuildUpstreamModelSyncPlanSeparatesCatalogActions(t *testing.T) { + upstream := &domainchannel.Upstream{ID: 9, Name: "test", Compatible: "openai", BaseURL: "https://example.com"} + unchangedItem := llm.ModelItem{ID: "unchanged", OwnedBy: "openai"} + unchangedKinds := inferKindsJSON(unchangedItem.ID) + unchangedProtocol, err := resolveRouteProtocol("", upstream.Compatible, upstream.ProtocolDefaultsJSON, unchangedKinds) + if err != nil { + t.Fatalf("resolve unchanged protocol: %v", err) + } + unchanged := *syncedUpstreamModel(upstream, unchangedItem, "unchanged-code", nil, unchangedProtocol, unchangedKinds) + unchanged.ID = 1 + updated := unchanged + updated.ID = 2 + updated.BindingCode = "updated-code" + updated.UpstreamModelName = "updated" + updated.Vendor = "stale-vendor" + updated.RawJSON = `{}` + + plan, err := buildUpstreamModelSyncPlan( + upstream, + []llm.ModelItem{ + {ID: "added", OwnedBy: "openai"}, + {ID: "manual", OwnedBy: "custom"}, + {ID: "reactivated", OwnedBy: "openai"}, + unchangedItem, + {ID: "updated", OwnedBy: "openai"}, + }, + []domainchannel.UpstreamModel{ + unchanged, + updated, + {ID: 3, UpstreamID: 9, BindingCode: "reactivated-code", UpstreamModelName: "reactivated", Status: "inactive", Source: "sync"}, + {ID: 4, UpstreamID: 9, BindingCode: "removed-code", UpstreamModelName: "removed", Status: "active", Source: "sync"}, + }, + map[string]repositoryUpstreamModelSnapshot{ + "manual": {BindingCode: "manual-code", Status: "active"}, + }, + ) + if err != nil { + t.Fatalf("build sync plan: %v", err) + } + if !reflect.DeepEqual(plan.AddedModels, []string{"added"}) || + !reflect.DeepEqual(plan.UpdatedModels, []string{"updated"}) || + !reflect.DeepEqual(plan.ReactivatedModels, []string{"reactivated"}) || + !reflect.DeepEqual(plan.InactivatedModels, []string{"removed"}) || + !reflect.DeepEqual(plan.UnchangedModels, []string{"unchanged"}) || + !reflect.DeepEqual(plan.ProtectedModels, []string{"manual"}) { + t.Fatalf("unexpected sync plan: %+v", plan) + } + if remoteModelsSnapshotID([]llm.ModelItem{{ID: "a"}}) == remoteModelsSnapshotID([]llm.ModelItem{{ID: "b"}}) { + t.Fatal("different remote snapshots produced the same identifier") + } +} + type modelUpdateRepo struct { model domainchannel.PlatformModel upstream domainchannel.Upstream @@ -654,6 +778,8 @@ type modelUpdateRepo struct { breakerDefaults domainchannel.BreakerDefaults llmSetting domainchannel.LLMSetting upsertLLMSettingErr error + upstreamModels map[string]domainchannel.UpstreamModel + catalogApplyCalls int } func (r *modelUpdateRepo) WithinTransaction(ctx context.Context, fn func(repository.ChannelRepository) error) error { @@ -772,7 +898,16 @@ func (r *modelUpdateRepo) ListModels(context.Context, repository.ListChannelMode return r.modelRows, int64(len(r.modelRows)), nil } -func (r *modelUpdateRepo) UpsertUpstreamModel(context.Context, *domainchannel.UpstreamModel) error { +func (r *modelUpdateRepo) UpsertUpstreamModel(_ context.Context, item *domainchannel.UpstreamModel) error { + if r.upstreamModels != nil { + stored := *item + if existing, ok := r.upstreamModels[item.UpstreamModelName]; ok { + stored.ID = existing.ID + } else if stored.ID == 0 { + stored.ID = uint(len(r.upstreamModels) + 1) + } + r.upstreamModels[item.UpstreamModelName] = stored + } return nil } @@ -784,24 +919,71 @@ func (r *modelUpdateRepo) GetUpstreamModelByID(context.Context, uint, uint) (*do return nil, repository.ErrNotFound } -func (r *modelUpdateRepo) GetUpstreamModelByUpstreamName(context.Context, uint, string) (*domainchannel.UpstreamModel, error) { - return nil, repository.ErrNotFound +func (r *modelUpdateRepo) GetUpstreamModelByUpstreamName(_ context.Context, upstreamID uint, name string) (*domainchannel.UpstreamModel, error) { + if item, ok := r.upstreamModels[name]; ok && item.UpstreamID == upstreamID { + result := item + return &result, nil + } + return nil, ErrUpstreamModelNotFound } func (r *modelUpdateRepo) DeleteUpstreamModel(context.Context, uint, uint) error { return nil } -func (r *modelUpdateRepo) MarkMissingSyncedUpstreamModelsInactive(context.Context, uint, []string) (int64, error) { - return 0, nil +func (r *modelUpdateRepo) ListManagedUpstreamModels(_ context.Context, upstreamID uint) ([]domainchannel.UpstreamModel, error) { + items := make([]domainchannel.UpstreamModel, 0) + for _, item := range r.upstreamModels { + if item.UpstreamID == upstreamID && (item.Source == "sync" || item.Source == "import") { + items = append(items, item) + } + } + return items, nil +} + +func (r *modelUpdateRepo) ApplyUpstreamModelCatalogChanges(_ context.Context, upstreamID uint, input repository.ApplyUpstreamModelCatalogChangesInput) (int64, error) { + r.catalogApplyCalls++ + for _, item := range input.Create { + stored := item + if stored.ID == 0 { + stored.ID = uint(len(r.upstreamModels) + 1) + } + r.upstreamModels[stored.UpstreamModelName] = stored + } + for _, item := range input.Update { + r.upstreamModels[item.UpstreamModelName] = item + } + inactiveIDs := make(map[uint]struct{}, len(input.InactivateIDs)) + for _, id := range input.InactivateIDs { + inactiveIDs[id] = struct{}{} + } + var count int64 + for name, item := range r.upstreamModels { + if item.UpstreamID != upstreamID || item.Status != "active" || (item.Source != "sync" && item.Source != "import") { + continue + } + if _, exists := inactiveIDs[item.ID]; !exists { + continue + } + item.Status = "inactive" + r.upstreamModels[name] = item + count++ + } + return count, nil } func (r *modelUpdateRepo) ListUpstreamModels(context.Context, uint, repository.ListChannelUpstreamModelsInput) ([]repository.ChannelUpstreamModelListRow, int64, error) { return nil, 0, nil } -func (r *modelUpdateRepo) ListUpstreamModelsByNames(context.Context, uint, []string) ([]repository.ChannelUpstreamModelListRow, error) { - return nil, nil +func (r *modelUpdateRepo) ListUpstreamModelsByNames(_ context.Context, upstreamID uint, names []string) ([]repository.ChannelUpstreamModelListRow, error) { + items := make([]repository.ChannelUpstreamModelListRow, 0) + for _, name := range names { + if item, exists := r.upstreamModels[name]; exists && item.UpstreamID == upstreamID { + items = append(items, repository.ChannelUpstreamModelListRow{UpstreamModel: item}) + } + } + return items, nil } func (r *modelUpdateRepo) GetUpstreamModelRouteByID(context.Context, uint, uint) (*repository.ChannelUpstreamModelListRow, error) { diff --git a/backend/internal/application/channel/service_sync.go b/backend/internal/application/channel/service_sync.go index a5ba82c89..d75385c91 100644 --- a/backend/internal/application/channel/service_sync.go +++ b/backend/internal/application/channel/service_sync.go @@ -2,6 +2,7 @@ package channel import ( "context" + "crypto/sha256" "encoding/json" "errors" "fmt" @@ -12,6 +13,7 @@ import ( domainchannel "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/channel" "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/ports/llm" + "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/repository" "go.uber.org/zap" ) @@ -33,6 +35,7 @@ func (s *Service) ListRemoteModels(ctx context.Context, upstreamID uint) (*Upstr if err != nil { return nil, err } + items = normalizeRemoteModelItems(items) remoteNames := make([]string, 0, len(items)) for _, item := range items { @@ -44,6 +47,10 @@ func (s *Service) ListRemoteModels(ctx context.Context, upstreamID uint) (*Upstr if err != nil { return nil, err } + managedModels, err := s.repo.ListManagedUpstreamModels(ctx, upstreamID) + if err != nil { + return nil, err + } existingByName := make(map[string]repositoryUpstreamModelSnapshot, len(rows)) for _, row := range rows { name := strings.TrimSpace(row.UpstreamModelName) @@ -86,7 +93,16 @@ func (s *Service) ListRemoteModels(ctx context.Context, upstreamID uint) (*Upstr }) } - return &UpstreamRemoteModelsData{Total: len(views), Items: views}, nil + syncPlan, err := buildUpstreamModelSyncPlan(upstreamItem, items, managedModels, existingByName) + if err != nil { + return nil, err + } + return &UpstreamRemoteModelsData{ + Total: len(views), + Items: views, + SnapshotID: remoteModelsSnapshotID(items), + SyncPlan: syncPlan, + }, nil } type repositoryUpstreamModelSnapshot struct { @@ -108,8 +124,8 @@ func appendUniqueString(items []string, value string) []string { return append(items, value) } -// SyncUpstreamModels 拉取上游 models 并写入上游真实模型清单。 -func (s *Service) SyncUpstreamModels(ctx context.Context, upstreamID uint) (*SyncUpstreamModelsData, error) { +// SyncUpstreamModels 拉取上游完整模型快照，并与本地远端管理目录进行原子软对账。 +func (s *Service) SyncUpstreamModels(ctx context.Context, upstreamID uint, input SyncUpstreamModelsInput) (*SyncUpstreamModelsData, error) { upstreamItem, err := s.repo.GetUpstreamByID(ctx, upstreamID) if err != nil { if errors.Is(err, ErrUpstreamNotFound) { @@ -122,51 +138,260 @@ func (s *Service) SyncUpstreamModels(ctx context.Context, upstreamID uint) (*Syn if err != nil { return nil, err } + items = normalizeRemoteModelItems(items) + if expected := strings.TrimSpace(input.ExpectedSnapshot); expected != "" && expected != remoteModelsSnapshotID(items) { + return nil, ErrRemoteModelsSnapshotChanged + } + return s.reconcileRemoteModelSnapshot(ctx, upstreamItem, items, input.AllowEmpty) +} - slices.SortFunc(items, func(a, b llm.ModelItem) int { - return strings.Compare(a.ID, b.ID) - }) +// reconcileRemoteModelSnapshot 在一个事务内完成目录读取、分类与批量写入； +// 上游网络请求已在事务外完成，避免长事务占用数据库连接。 +func (s *Service) reconcileRemoteModelSnapshot( + ctx context.Context, + upstreamItem *domainchannel.Upstream, + items []llm.ModelItem, + allowEmpty bool, +) (*SyncUpstreamModelsData, error) { + items = normalizeRemoteModelItems(items) + if len(items) == 0 && !allowEmpty { + return nil, ErrEmptyRemoteModels + } result := &SyncUpstreamModelsData{ + SnapshotID: remoteModelsSnapshotID(items), TotalUpstream: len(items), SyncedModels: make([]UpstreamSyncModelView, 0, len(items)), } - seen := make(map[string]struct{}, len(items)) + remoteNames := make([]string, 0, len(items)) + for _, item := range items { + remoteNames = append(remoteNames, item.ID) + } + + err := s.repo.WithinTransaction(ctx, func(txRepo repository.ChannelRepository) error { + existingRows, syncErr := txRepo.ListUpstreamModelsByNames(ctx, upstreamItem.ID, remoteNames) + if syncErr != nil { + return syncErr + } + managedModels, syncErr := txRepo.ListManagedUpstreamModels(ctx, upstreamItem.ID) + if syncErr != nil { + return syncErr + } + + existingByName := make(map[string]repositoryUpstreamModelSnapshot, len(existingRows)) + for _, row := range existingRows { + name := strings.TrimSpace(row.UpstreamModelName) + if name == "" { + continue + } + existingByName[name] = repositoryUpstreamModelSnapshot{ + BindingCode: row.BindingCode, + Status: row.Status, + } + } + managedByName := make(map[string]domainchannel.UpstreamModel, len(managedModels)) + for _, model := range managedModels { + name := strings.TrimSpace(model.UpstreamModelName) + if name != "" { + managedByName[name] = model + } + } + + remoteNameSet := make(map[string]struct{}, len(items)) + changes := repository.ApplyUpstreamModelCatalogChangesInput{ + Create: make([]domainchannel.UpstreamModel, 0), + Update: make([]domainchannel.UpstreamModel, 0), + } + now := time.Now() + for _, item := range items { + name := item.ID + remoteNameSet[name] = struct{}{} + kindsJSON := inferKindsJSON(name) + protocol, resolveErr := resolveRouteProtocol("", upstreamItem.Compatible, upstreamItem.ProtocolDefaultsJSON, kindsJSON) + if resolveErr != nil { + return resolveErr + } + + if existing, managed := managedByName[name]; managed { + desired := syncedUpstreamModel(upstreamItem, item, existing.BindingCode, &now, protocol, kindsJSON) + desired.ID = existing.ID + desired.CreatedAt = existing.CreatedAt + desired.UpdatedAt = existing.UpdatedAt + reactivated := !strings.EqualFold(strings.TrimSpace(existing.Status), "active") + updated := !reactivated && upstreamModelMetadataChanged(existing, desired) + switch { + case reactivated: + result.ReactivatedModels++ + case updated: + result.UpdatedUpstreamModels++ + default: + result.UnchangedUpstreamModels++ + } + changes.Update = append(changes.Update, *desired) + result.SyncedModels = append(result.SyncedModels, UpstreamSyncModelView{ + UpstreamModelName: desired.UpstreamModelName, + BindingCode: desired.BindingCode, + SuggestedProtocol: desired.SuggestedProtocol, + KindsJSON: desired.KindsJSON, + Status: desired.Status, + Updated: updated, + Reactivated: reactivated, + }) + continue + } + + if existing, protected := existingByName[name]; protected { + result.ProtectedUpstreamModels++ + result.SyncedModels = append(result.SyncedModels, UpstreamSyncModelView{ + UpstreamModelName: name, + BindingCode: existing.BindingCode, + SuggestedProtocol: protocol, + KindsJSON: kindsJSON, + Status: existing.Status, + Protected: true, + }) + continue + } + + created := syncedUpstreamModel(upstreamItem, item, generateBindingCode(), &now, protocol, kindsJSON) + changes.Create = append(changes.Create, *created) + result.CreatedUpstreamModels++ + result.SyncedModels = append(result.SyncedModels, UpstreamSyncModelView{ + UpstreamModelName: created.UpstreamModelName, + BindingCode: created.BindingCode, + SuggestedProtocol: created.SuggestedProtocol, + KindsJSON: created.KindsJSON, + Status: created.Status, + Created: true, + }) + } + + for _, model := range managedModels { + if !strings.EqualFold(strings.TrimSpace(model.Status), "active") { + continue + } + if _, present := remoteNameSet[strings.TrimSpace(model.UpstreamModelName)]; !present { + changes.InactivateIDs = append(changes.InactivateIDs, model.ID) + } + } + + inactivated, syncErr := txRepo.ApplyUpstreamModelCatalogChanges(ctx, upstreamItem.ID, changes) + if syncErr != nil { + if errors.Is(syncErr, repository.ErrDuplicate) { + return ErrRemoteModelsSnapshotChanged + } + return syncErr + } + result.InactivatedModels = inactivated + result.ExistingUpstreamModels = result.TotalUpstream - result.CreatedUpstreamModels + return nil + }) + if err != nil { + return nil, err + } + + s.InvalidateModelCatalog() + return result, nil +} + +func normalizeRemoteModelItems(items []llm.ModelItem) []llm.ModelItem { + unique := make(map[string]llm.ModelItem, len(items)) for _, item := range items { name := strings.TrimSpace(item.ID) if name == "" { continue } - if _, exists := seen[name]; exists { + if _, exists := unique[name]; exists { continue } - seen[name] = struct{}{} + item.ID = name + item.OwnedBy = strings.TrimSpace(item.OwnedBy) + unique[name] = item + } + result := make([]llm.ModelItem, 0, len(unique)) + for _, item := range unique { + result = append(result, item) + } + slices.SortFunc(result, func(a, b llm.ModelItem) int { + return strings.Compare(a.ID, b.ID) + }) + return result +} - view, syncErr := s.syncSingleUpstreamModel(ctx, upstreamItem, item) - if syncErr != nil { - result.SkippedUpstreamModels++ +func remoteModelsSnapshotID(items []llm.ModelItem) string { + payload, _ := json.Marshal(items) + sum := sha256.Sum256(payload) + return fmt.Sprintf("%x", sum) +} + +func buildUpstreamModelSyncPlan( + upstream *domainchannel.Upstream, + remoteItems []llm.ModelItem, + managedModels []domainchannel.UpstreamModel, + existingByName map[string]repositoryUpstreamModelSnapshot, +) (UpstreamModelSyncPlanView, error) { + plan := UpstreamModelSyncPlanView{ + AddedModels: []string{}, + UpdatedModels: []string{}, + ReactivatedModels: []string{}, + InactivatedModels: []string{}, + UnchangedModels: []string{}, + ProtectedModels: []string{}, + } + managedByName := make(map[string]domainchannel.UpstreamModel, len(managedModels)) + for _, item := range managedModels { + name := strings.TrimSpace(item.UpstreamModelName) + if name != "" { + managedByName[name] = item + } + } + remoteNames := make(map[string]struct{}, len(remoteItems)) + for _, item := range remoteItems { + name := strings.TrimSpace(item.ID) + remoteNames[name] = struct{}{} + existing, managed := managedByName[name] + if !managed { + if _, protected := existingByName[name]; protected { + plan.ProtectedModels = append(plan.ProtectedModels, name) + } else { + plan.AddedModels = append(plan.AddedModels, name) + } continue } - if view.Created { - result.CreatedUpstreamModels++ + if !strings.EqualFold(strings.TrimSpace(existing.Status), "active") { + plan.ReactivatedModels = append(plan.ReactivatedModels, name) + continue + } + kindsJSON := inferKindsJSON(name) + protocol, err := resolveRouteProtocol("", upstream.Compatible, upstream.ProtocolDefaultsJSON, kindsJSON) + if err != nil { + return UpstreamModelSyncPlanView{}, err + } + desired := syncedUpstreamModel(upstream, item, existing.BindingCode, nil, protocol, kindsJSON) + if upstreamModelMetadataChanged(existing, desired) { + plan.UpdatedModels = append(plan.UpdatedModels, name) } else { - result.ExistingUpstreamModels++ + plan.UnchangedModels = append(plan.UnchangedModels, name) } - result.SyncedModels = append(result.SyncedModels, view) - } - - activeNames := make([]string, 0, len(seen)) - for name := range seen { - activeNames = append(activeNames, name) } - inactivated, err := s.repo.MarkMissingSyncedUpstreamModelsInactive(ctx, upstreamID, activeNames) - if err != nil { - return nil, err + for _, item := range managedModels { + name := strings.TrimSpace(item.UpstreamModelName) + if !strings.EqualFold(strings.TrimSpace(item.Status), "active") { + continue + } + if _, present := remoteNames[name]; !present { + plan.InactivatedModels = append(plan.InactivatedModels, name) + } } - result.InactivatedModels = inactivated + return plan, nil +} - s.InvalidateModelCatalog() - return result, nil +func upstreamModelMetadataChanged(existing domainchannel.UpstreamModel, desired *domainchannel.UpstreamModel) bool { + return strings.TrimSpace(existing.Vendor) != strings.TrimSpace(desired.Vendor) || + strings.TrimSpace(existing.Icon) != strings.TrimSpace(desired.Icon) || + strings.TrimSpace(existing.SuggestedProtocol) != strings.TrimSpace(desired.SuggestedProtocol) || + strings.TrimSpace(existing.KindsJSON) != strings.TrimSpace(desired.KindsJSON) || + strings.TrimSpace(existing.RawJSON) != strings.TrimSpace(desired.RawJSON) } // ImportUpstreamModels 批量把上游真实模型绑定到平台模型。 @@ -332,52 +557,33 @@ func (s *Service) fetchRemoteModels(ctx context.Context, up *domainchannel.Upstr return items, nil } -func (s *Service) syncSingleUpstreamModel(ctx context.Context, up *domainchannel.Upstream, item llm.ModelItem) (UpstreamSyncModelView, error) { - upstreamModelName := strings.TrimSpace(item.ID) - kindsJSON := inferKindsJSON(upstreamModelName) - protocol, err := resolveRouteProtocol("", up.Compatible, up.ProtocolDefaultsJSON, kindsJSON) - if err != nil { - return UpstreamSyncModelView{}, err - } - created := false - bindingCode := generateBindingCode() - if existing, err := s.repo.GetUpstreamModelByUpstreamName(ctx, up.ID, upstreamModelName); err == nil { - bindingCode = existing.BindingCode - } else if errors.Is(err, ErrUpstreamModelNotFound) { - created = true - } else { - return UpstreamSyncModelView{}, err - } - now := time.Now() +func syncedUpstreamModel( + upstream *domainchannel.Upstream, + item llm.ModelItem, + bindingCode string, + lastSyncedAt *time.Time, + protocol string, + kindsJSON string, +) *domainchannel.UpstreamModel { + name := strings.TrimSpace(item.ID) rawJSON, _ := json.Marshal(map[string]string{ - "id": item.ID, - "owned_by": item.OwnedBy, + "id": name, + "owned_by": strings.TrimSpace(item.OwnedBy), }) - vendor := normalizeUpstreamModelVendor(item.OwnedBy, upstreamModelName, up.Name, up.BaseURL) - upstreamModel := &domainchannel.UpstreamModel{ - UpstreamID: up.ID, + vendor := normalizeUpstreamModelVendor(item.OwnedBy, name, upstream.Name, upstream.BaseURL) + return &domainchannel.UpstreamModel{ + UpstreamID: upstream.ID, BindingCode: bindingCode, - UpstreamModelName: upstreamModelName, + UpstreamModelName: name, Vendor: vendor, - Icon: normalizeModelIcon("", vendor, upstreamModelName), + Icon: normalizeModelIcon("", vendor, name), SuggestedProtocol: protocol, KindsJSON: kindsJSON, Status: "active", Source: "sync", - LastSyncedAt: &now, + LastSyncedAt: lastSyncedAt, RawJSON: string(rawJSON), } - if err := s.repo.UpsertUpstreamModel(ctx, upstreamModel); err != nil { - return UpstreamSyncModelView{}, err - } - return UpstreamSyncModelView{ - UpstreamModelName: upstreamModel.UpstreamModelName, - BindingCode: upstreamModel.BindingCode, - SuggestedProtocol: upstreamModel.SuggestedProtocol, - KindsJSON: upstreamModel.KindsJSON, - Status: upstreamModel.Status, - Created: created, - }, nil } func (s *Service) importSingleUpstreamModel(ctx context.Context, upstreamItem *domainchannel.Upstream, input ImportUpstreamModelItemInput) (ImportUpstreamModelResultView, error) { @@ -423,7 +629,8 @@ func (s *Service) importSingleUpstreamModel(ctx context.Context, upstreamItem *d status := input.Status priority := input.Priority weight := 1 - source := "import" + routeSource := "import" + catalogSource := "sync" view, err := s.UpsertUpstreamModel(ctx, upstreamItem.ID, UpsertUpstreamModelInput{ PlatformModelName: platformModelName, UpstreamModelName: upstreamModelName, @@ -432,7 +639,8 @@ func (s *Service) importSingleUpstreamModel(ctx context.Context, upstreamItem *d Status: &status, Priority: &priority, Weight: &weight, - Source: &source, + Source: &routeSource, + CatalogSource: &catalogSource, }) if err != nil { return ImportUpstreamModelResultView{}, err diff --git a/backend/internal/application/channel/service_upstream_model.go b/backend/internal/application/channel/service_upstream_model.go index d6b6fabb8..81995ffb1 100644 --- a/backend/internal/application/channel/service_upstream_model.go +++ b/backend/internal/application/channel/service_upstream_model.go @@ -104,7 +104,9 @@ func (s *Service) UpsertUpstreamModel(ctx context.Context, upstreamID uint, inpu upstreamModelVendor := normalizeUpstreamModelVendor("", upstreamModelName, upstream.Name, upstream.BaseURL) upstreamModelIcon := normalizeModelIcon("", upstreamModelVendor, upstreamModelName) upstreamModelSource := "manual" - if input.Source != nil { + if input.CatalogSource != nil { + upstreamModelSource = normalizeSource(*input.CatalogSource) + } else if input.Source != nil { upstreamModelSource = normalizeSource(*input.Source) } upstreamModel, txErr := ensureUpstreamCatalogModel( diff --git a/backend/internal/infra/llm/anthropic.go b/backend/internal/infra/llm/anthropic.go index 84991193d..29c516d05 100644 --- a/backend/internal/infra/llm/anthropic.go +++ b/backend/internal/infra/llm/anthropic.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "net/http" + "net/url" "strings" "time" ) @@ -1588,53 +1589,75 @@ func compactAnthropicStreamServerToolCalls(result *GenerateOutput) { // // {"data":[{"id":"claude-opus-4-5","display_name":"...","type":"model"},...]} func (c *Client) listModelsAnthropic(ctx context.Context, route RouteConfig) ([]ModelItem, error) { - requestURL := buildAnthropicModelsURL(route.BaseURL) - if requestURL == "" { + baseRequestURL := buildAnthropicModelsURL(route.BaseURL) + if baseRequestURL == "" { return nil, fmt.Errorf("invalid base url") } requestCtx, cancel := context.WithTimeout(ctx, resolveReadTimeout(route.ReadTimeoutMS)) defer cancel() - req, err := c.newAnthropicRequest(requestCtx, http.MethodGet, requestURL, nil, route, nil) - if err != nil { - return nil, err - } - - resp, err := c.doRouteRequest(route, req) - if err != nil { - return nil, err - } - defer resp.Body.Close() //nolint:errcheck - - body, err := readUpstreamBody(resp.Body) - if err != nil { - return nil, err - } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, parseAnthropicError(resp.StatusCode, body, upstreamDebugSnapshot(req, nil, resp, body)) - } - - parsed := struct { - Data []struct { - ID string `json:"id"` - DisplayName string `json:"display_name"` - } `json:"data"` - }{} - if err = json.Unmarshal(body, &parsed); err != nil { - return nil, err - } - - results := make([]ModelItem, 0, len(parsed.Data)) - for _, item := range parsed.Data { - modelID := strings.TrimSpace(item.ID) - if modelID == "" { - continue + results := make([]ModelItem, 0) + afterID := "" + seenAfterIDs := make(map[string]struct{}) + for { + pageURL, err := url.Parse(baseRequestURL) + if err != nil { + return nil, err + } + query := pageURL.Query() + query.Set("limit", "1000") + if afterID != "" { + query.Set("after_id", afterID) + } + pageURL.RawQuery = query.Encode() + + req, err := c.newAnthropicRequest(requestCtx, http.MethodGet, pageURL.String(), nil, route, nil) + if err != nil { + return nil, err + } + resp, err := c.doRouteRequest(route, req) + if err != nil { + return nil, err + } + body, readErr := readUpstreamBody(resp.Body) + _ = resp.Body.Close() + if readErr != nil { + return nil, readErr + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, parseAnthropicError(resp.StatusCode, body, upstreamDebugSnapshot(req, nil, resp, body)) + } + + parsed := struct { + Data []struct { + ID string `json:"id"` + DisplayName string `json:"display_name"` + } `json:"data"` + HasMore bool `json:"has_more"` + LastID string `json:"last_id"` + }{} + if err = json.Unmarshal(body, &parsed); err != nil { + return nil, err + } + for _, item := range parsed.Data { + modelID := strings.TrimSpace(item.ID) + if modelID == "" { + continue + } + results = append(results, ModelItem{ID: modelID, OwnedBy: "anthropic"}) } - results = append(results, ModelItem{ - ID: modelID, - OwnedBy: "anthropic", - }) + if !parsed.HasMore { + return results, nil + } + nextAfterID := strings.TrimSpace(parsed.LastID) + if nextAfterID == "" { + return nil, fmt.Errorf("invalid anthropic models pagination cursor") + } + if _, exists := seenAfterIDs[nextAfterID]; exists { + return nil, fmt.Errorf("repeated anthropic models pagination cursor") + } + seenAfterIDs[nextAfterID] = struct{}{} + afterID = nextAfterID } - return results, nil } diff --git a/backend/internal/infra/llm/client_http_test.go b/backend/internal/infra/llm/client_http_test.go index 74e0b7852..e27041414 100644 --- a/backend/internal/infra/llm/client_http_test.go +++ b/backend/internal/infra/llm/client_http_test.go @@ -130,6 +130,90 @@ func TestListModelsFallsBackToOpenAICompatibleModels(t *testing.T) { } } +func TestListModelsAnthropicFetchesEveryPage(t *testing.T) { + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if r.URL.Path != "/v1/models" || r.URL.Query().Get("limit") != "1000" { + t.Fatalf("unexpected models request: %s", r.URL.String()) + } + switch calls { + case 1: + if got := r.URL.Query().Get("after_id"); got != "" { + t.Fatalf("unexpected first-page cursor %q", got) + } + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": []map[string]string{{"id": "claude-first"}}, + "has_more": true, + "last_id": "cursor-1", + }) + case 2: + if got := r.URL.Query().Get("after_id"); got != "cursor-1" { + t.Fatalf("unexpected second-page cursor %q", got) + } + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "data": []map[string]string{{"id": "claude-second"}}, + "has_more": false, + }) + default: + t.Fatalf("unexpected request %d", calls) + } + })) + defer server.Close() + + items, err := NewClient(security.NewStrictOutboundPolicy(true)).listModelsAnthropic(t.Context(), RouteConfig{ + BaseURL: server.URL, + APIKey: "test-key", + }) + if err != nil { + t.Fatalf("list anthropic models: %v", err) + } + if calls != 2 || len(items) != 2 || items[0].ID != "claude-first" || items[1].ID != "claude-second" { + t.Fatalf("unexpected paginated models: calls=%d items=%#v", calls, items) + } +} + +func TestListModelsGeminiFetchesEveryPage(t *testing.T) { + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if r.URL.Path != "/v1beta/models" || r.URL.Query().Get("pageSize") != "1000" { + t.Fatalf("unexpected models request: %s", r.URL.String()) + } + switch calls { + case 1: + if got := r.URL.Query().Get("pageToken"); got != "" { + t.Fatalf("unexpected first-page token %q", got) + } + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "models": []map[string]string{{"name": "models/gemini-first"}}, + "nextPageToken": "token-1", + }) + case 2: + if got := r.URL.Query().Get("pageToken"); got != "token-1" { + t.Fatalf("unexpected second-page token %q", got) + } + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "models": []map[string]string{{"name": "models/gemini-second"}}, + }) + default: + t.Fatalf("unexpected request %d", calls) + } + })) + defer server.Close() + + items, err := NewClient(security.NewStrictOutboundPolicy(true)).listModelsGemini(t.Context(), RouteConfig{ + BaseURL: server.URL, + APIKey: "test-key", + }) + if err != nil { + t.Fatalf("list gemini models: %v", err) + } + if calls != 2 || len(items) != 2 || items[0].ID != "gemini-first" || items[1].ID != "gemini-second" { + t.Fatalf("unexpected paginated models: calls=%d items=%#v", calls, items) + } +} + type repeatingReader struct { remaining int } diff --git a/backend/internal/infra/llm/gemini.go b/backend/internal/infra/llm/gemini.go index 49643f810..6f3ee3056 100644 --- a/backend/internal/infra/llm/gemini.go +++ b/backend/internal/infra/llm/gemini.go @@ -24,6 +24,7 @@ import ( "fmt" "io" "net/http" + "net/url" "strings" "time" ) @@ -1535,51 +1536,68 @@ func mergeGeminiServerSideToolUsage(current map[string]int64, next map[string]in // 响应：{"models":[{"name":"models/gemini-2.0-flash","displayName":"..."},...]} func (c *Client) listModelsGemini(ctx context.Context, route RouteConfig) ([]ModelItem, error) { base := geminiBaseURL(route) - requestURL := buildGeminiModelsURL(base) + baseRequestURL := buildGeminiModelsURL(base) requestCtx, cancel := context.WithTimeout(ctx, resolveReadTimeout(route.ReadTimeoutMS)) defer cancel() - req, err := c.newGeminiRequest(requestCtx, http.MethodGet, requestURL, nil, route, nil) - if err != nil { - return nil, err - } - - resp, err := c.doRouteRequest(route, req) - if err != nil { - return nil, err - } - defer resp.Body.Close() //nolint:errcheck - - body, err := readUpstreamBody(resp.Body) - if err != nil { - return nil, err - } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, parseGeminiError(resp.StatusCode, body, upstreamDebugSnapshot(req, nil, resp, body)) - } - - parsed := struct { - Models []struct { - Name string `json:"name"` - DisplayName string `json:"displayName"` - } `json:"models"` - }{} - if err = json.Unmarshal(body, &parsed); err != nil { - return nil, err - } - - results := make([]ModelItem, 0, len(parsed.Models)) - for _, item := range parsed.Models { - // name 格式为 "models/gemini-xxx"，提取末段作为 ID - id := strings.TrimPrefix(strings.TrimSpace(item.Name), "models/") - if id == "" { - continue + results := make([]ModelItem, 0) + pageToken := "" + seenPageTokens := make(map[string]struct{}) + for { + pageURL, err := url.Parse(baseRequestURL) + if err != nil { + return nil, err + } + query := pageURL.Query() + query.Set("pageSize", "1000") + if pageToken != "" { + query.Set("pageToken", pageToken) + } + pageURL.RawQuery = query.Encode() + + req, err := c.newGeminiRequest(requestCtx, http.MethodGet, pageURL.String(), nil, route, nil) + if err != nil { + return nil, err + } + resp, err := c.doRouteRequest(route, req) + if err != nil { + return nil, err + } + body, readErr := readUpstreamBody(resp.Body) + _ = resp.Body.Close() + if readErr != nil { + return nil, readErr + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, parseGeminiError(resp.StatusCode, body, upstreamDebugSnapshot(req, nil, resp, body)) + } + + parsed := struct { + Models []struct { + Name string `json:"name"` + DisplayName string `json:"displayName"` + } `json:"models"` + NextPageToken string `json:"nextPageToken"` + }{} + if err = json.Unmarshal(body, &parsed); err != nil { + return nil, err + } + for _, item := range parsed.Models { + id := strings.TrimPrefix(strings.TrimSpace(item.Name), "models/") + if id == "" { + continue + } + results = append(results, ModelItem{ID: id, OwnedBy: "google"}) } - results = append(results, ModelItem{ - ID: id, - OwnedBy: "google", - }) + nextPageToken := strings.TrimSpace(parsed.NextPageToken) + if nextPageToken == "" { + return results, nil + } + if _, exists := seenPageTokens[nextPageToken]; exists { + return nil, fmt.Errorf("repeated gemini models pagination token") + } + seenPageTokens[nextPageToken] = struct{}{} + pageToken = nextPageToken } - return results, nil } diff --git a/backend/internal/infra/persistence/postgres/channel/repository.go b/backend/internal/infra/persistence/postgres/channel/repository.go index a6ddaea9f..3cf5c09b9 100644 --- a/backend/internal/infra/persistence/postgres/channel/repository.go +++ b/backend/internal/infra/persistence/postgres/channel/repository.go @@ -860,19 +860,108 @@ func (r *Repo) DeleteUpstreamModel(ctx context.Context, sourceID uint, upstreamI })) } -// MarkMissingSyncedUpstreamModelsInactive 将本次同步未返回的同步来源模型置为停用。 -func (r *Repo) MarkMissingSyncedUpstreamModelsInactive(ctx context.Context, upstreamID uint, activeNames []string) (int64, error) { - query := r.db.WithContext(ctx). - Model(&model.LLMUpstreamModel{}). - Where("upstream_id = ? AND source = ? AND status = ?", upstreamID, "sync", "active") - if len(activeNames) > 0 { - query = query.Where("upstream_model_name NOT IN ?", activeNames) +// ListManagedUpstreamModels 返回由远端同步管理的目录项，用于生成同步变更预览。 +func (r *Repo) ListManagedUpstreamModels(ctx context.Context, upstreamID uint) ([]domainchannel.UpstreamModel, error) { + items := make([]model.LLMUpstreamModel, 0) + if err := r.db.WithContext(ctx). + Where("upstream_id = ? AND source IN ?", upstreamID, []string{"sync", "import"}). + Order("upstream_model_name ASC, id ASC"). + Find(&items).Error; err != nil { + return nil, translateError(err) } - result := query.Update("status", "inactive") - if result.Error != nil { - return 0, translateError(result.Error) + result := make([]domainchannel.UpstreamModel, 0, len(items)) + for _, item := range items { + result = append(result, toUpstreamModelDomain(item)) + } + return result, nil +} + +// ApplyUpstreamModelCatalogChanges 批量应用应用层已经分类完成的目录变更。 +// import 是旧版本写入目录项时使用的来源值，停用条件中保留该值用于平滑迁移。 +func (r *Repo) ApplyUpstreamModelCatalogChanges( + ctx context.Context, + upstreamID uint, + input repository.ApplyUpstreamModelCatalogChangesInput, +) (int64, error) { + if upstreamID == 0 { + return 0, repository.ErrInvalidInput + } + + createdRows := make([]model.LLMUpstreamModel, 0, len(input.Create)) + for i := range input.Create { + item := input.Create[i] + if item.ID != 0 || item.UpstreamID != upstreamID || strings.TrimSpace(item.UpstreamModelName) == "" || strings.TrimSpace(item.BindingCode) == "" { + return 0, repository.ErrInvalidInput + } + createdRows = append(createdRows, toUpstreamModelModel(&item)) + } + + now := time.Now() + updatedRows := make([]model.LLMUpstreamModel, 0, len(input.Update)) + for i := range input.Update { + item := input.Update[i] + if item.ID == 0 || item.UpstreamID != upstreamID || strings.TrimSpace(item.UpstreamModelName) == "" || strings.TrimSpace(item.BindingCode) == "" { + return 0, repository.ErrInvalidInput + } + entity := toUpstreamModelModel(&item) + entity.ID = item.ID + entity.CreatedAt = item.CreatedAt + entity.UpdatedAt = now + updatedRows = append(updatedRows, entity) + } + + db := r.db.WithContext(ctx) + if len(createdRows) > 0 { + if err := db.CreateInBatches(&createdRows, 200).Error; err != nil { + return 0, translateError(err) + } + } + if len(updatedRows) > 0 { + if err := db.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "id"}}, + DoUpdates: clause.AssignmentColumns([]string{ + "binding_code", + "upstream_model_name", + "vendor", + "icon", + "suggested_protocol", + "kinds_json", + "status", + "source", + "last_synced_at", + "raw_json", + "updated_at", + }), + }).CreateInBatches(&updatedRows, 200).Error; err != nil { + return 0, translateError(err) + } + } + + uniqueInactiveIDs := make([]uint, 0, len(input.InactivateIDs)) + seenInactiveIDs := make(map[uint]struct{}, len(input.InactivateIDs)) + for _, id := range input.InactivateIDs { + if id == 0 { + return 0, repository.ErrInvalidInput + } + if _, exists := seenInactiveIDs[id]; exists { + continue + } + seenInactiveIDs[id] = struct{}{} + uniqueInactiveIDs = append(uniqueInactiveIDs, id) } - return result.RowsAffected, nil + + var inactivated int64 + for start := 0; start < len(uniqueInactiveIDs); start += 200 { + end := min(start+200, len(uniqueInactiveIDs)) + result := db.Model(&model.LLMUpstreamModel{}). + Where("upstream_id = ? AND id IN ? AND source IN ? AND status = ?", upstreamID, uniqueInactiveIDs[start:end], []string{"sync", "import"}, "active"). + Update("status", "inactive") + if result.Error != nil { + return 0, translateError(result.Error) + } + inactivated += result.RowsAffected + } + return inactivated, nil } // --------------------------------------------------------------------------- @@ -1000,19 +1089,24 @@ func (r *Repo) ListUpstreamModelsByNames(ctx context.Context, upstreamID uint, u return []UpstreamModelListRow{}, nil } items := make([]UpstreamModelListRow, 0) - if err := r.db.WithContext(ctx). - Table("llm_upstream_models AS um"). - Select( - "um.*, r.id AS route_id, r.platform_model_id, pm.name AS platform_model_name, pm.vendor AS model_vendor, pm.kinds_json AS model_kinds_json, pm.icon AS model_icon, "+ - "r.protocol, r.status AS route_status, r.priority, r.weight, r.source AS route_source, "+ - "r.cb_failure_threshold, r.cb_duration_min, r.cb_window_min, r.headers_json", - ). - Joins("LEFT JOIN llm_model_routes r ON r.upstream_model_id = um.id"). - Joins("LEFT JOIN llm_platform_models pm ON pm.id = r.platform_model_id"). - Where("um.upstream_id = ? AND um.upstream_model_name IN ?", upstreamID, names). - Order("um.upstream_model_name ASC, r.id ASC NULLS LAST"). - Scan(&items).Error; err != nil { - return nil, translateError(err) + for start := 0; start < len(names); start += 500 { + end := min(start+500, len(names)) + chunk := make([]UpstreamModelListRow, 0) + if err := r.db.WithContext(ctx). + Table("llm_upstream_models AS um"). + Select( + "um.*, r.id AS route_id, r.platform_model_id, pm.name AS platform_model_name, pm.vendor AS model_vendor, pm.kinds_json AS model_kinds_json, pm.icon AS model_icon, "+ + "r.protocol, r.status AS route_status, r.priority, r.weight, r.source AS route_source, "+ + "r.cb_failure_threshold, r.cb_duration_min, r.cb_window_min, r.headers_json", + ). + Joins("LEFT JOIN llm_model_routes r ON r.upstream_model_id = um.id"). + Joins("LEFT JOIN llm_platform_models pm ON pm.id = r.platform_model_id"). + Where("um.upstream_id = ? AND um.upstream_model_name IN ?", upstreamID, names[start:end]). + Order("um.upstream_model_name ASC, r.id ASC NULLS LAST"). + Scan(&chunk).Error; err != nil { + return nil, translateError(err) + } + items = append(items, chunk...) } return items, nil } diff --git a/backend/internal/infra/persistence/postgres/channel/repository_sqlite_test.go b/backend/internal/infra/persistence/postgres/channel/repository_sqlite_test.go index 9ddd8f30b..1ae2fa0c2 100644 --- a/backend/internal/infra/persistence/postgres/channel/repository_sqlite_test.go +++ b/backend/internal/infra/persistence/postgres/channel/repository_sqlite_test.go @@ -150,6 +150,108 @@ func TestWithinTransactionSQLiteRollsBackAllChannelWrites(t *testing.T) { } } +func TestApplyUpstreamModelCatalogChangesPreservesManualCatalogAndRoutes(t *testing.T) { + db := openChannelSQLiteTestDB(t) + ctx := context.Background() + upstream := model.LLMUpstream{Name: "catalog-reconcile", Status: "active"} + if err := db.Create(&upstream).Error; err != nil { + t.Fatalf("create upstream: %v", err) + } + items := []model.LLMUpstreamModel{ + {UpstreamID: upstream.ID, BindingCode: "present", UpstreamModelName: "present", Status: "active", Source: "sync"}, + {UpstreamID: upstream.ID, BindingCode: "missing-sync", UpstreamModelName: "missing-sync", Status: "active", Source: "sync"}, + {UpstreamID: upstream.ID, BindingCode: "missing-import", UpstreamModelName: "missing-import", Status: "active", Source: "import"}, + {UpstreamID: upstream.ID, BindingCode: "missing-manual", UpstreamModelName: "missing-manual", Status: "active", Source: "manual"}, + } + if err := db.Create(&items).Error; err != nil { + t.Fatalf("create upstream models: %v", err) + } + platformModel := model.LLMPlatformModel{Name: "catalog-route", Vendor: "openai", Status: "active"} + if err := db.Create(&platformModel).Error; err != nil { + t.Fatalf("create platform model: %v", err) + } + route := model.LLMPlatformModelRoute{ + PlatformModelID: platformModel.ID, + UpstreamModelID: items[1].ID, + Protocol: "openai_responses", + Status: "active", + } + if err := db.Create(&route).Error; err != nil { + t.Fatalf("create platform route: %v", err) + } + + count, err := NewRepo(db).ApplyUpstreamModelCatalogChanges(ctx, upstream.ID, repository.ApplyUpstreamModelCatalogChangesInput{ + Create: []domainchannel.UpstreamModel{{ + UpstreamID: upstream.ID, BindingCode: "created", UpstreamModelName: "created", Status: "active", Source: "sync", RawJSON: "{}", + }}, + Update: []domainchannel.UpstreamModel{{ + ID: items[0].ID, UpstreamID: upstream.ID, BindingCode: items[0].BindingCode, UpstreamModelName: items[0].UpstreamModelName, + Vendor: "updated-vendor", Status: "active", Source: "sync", RawJSON: "{}", CreatedAt: items[0].CreatedAt, + }}, + InactivateIDs: []uint{items[1].ID, items[2].ID}, + }) + if err != nil { + t.Fatalf("apply catalog changes: %v", err) + } + if count != 2 { + t.Fatalf("expected two managed models to be deactivated, got %d", count) + } + var stored []model.LLMUpstreamModel + if err := db.Where("upstream_id = ?", upstream.ID).Order("upstream_model_name ASC").Find(&stored).Error; err != nil { + t.Fatalf("load upstream models: %v", err) + } + statuses := make(map[string]string, len(stored)) + for _, item := range stored { + statuses[item.UpstreamModelName] = item.Status + } + if statuses["missing-import"] != "inactive" || statuses["missing-sync"] != "inactive" { + t.Fatalf("expected missing managed models inactive, got %v", statuses) + } + if statuses["missing-manual"] != "active" || statuses["present"] != "active" { + t.Fatalf("expected manual and present models active, got %v", statuses) + } + if statuses["created"] != "active" { + t.Fatalf("expected batch-created model active, got %v", statuses) + } + var updated model.LLMUpstreamModel + if err := db.First(&updated, items[0].ID).Error; err != nil || updated.Vendor != "updated-vendor" { + t.Fatalf("expected managed model metadata update, got %+v, err=%v", updated, err) + } + var storedRoute model.LLMPlatformModelRoute + if err := db.First(&storedRoute, route.ID).Error; err != nil { + t.Fatalf("load preserved route: %v", err) + } + if storedRoute.Status != "active" { + t.Fatalf("expected route configuration to remain active, got %q", storedRoute.Status) + } +} + +func TestListManagedUpstreamModelsExcludesManualCatalog(t *testing.T) { + db := openChannelSQLiteTestDB(t) + ctx := t.Context() + upstream := model.LLMUpstream{Name: "managed-list", Status: "active"} + if err := db.Create(&upstream).Error; err != nil { + t.Fatalf("create upstream: %v", err) + } + for _, item := range []model.LLMUpstreamModel{ + {UpstreamID: upstream.ID, BindingCode: "sync", UpstreamModelName: "sync-model", Status: "active", Source: "sync"}, + {UpstreamID: upstream.ID, BindingCode: "legacy", UpstreamModelName: "legacy-model", Status: "inactive", Source: "import"}, + {UpstreamID: upstream.ID, BindingCode: "manual", UpstreamModelName: "manual-model", Status: "active", Source: "manual"}, + } { + if err := db.Create(&item).Error; err != nil { + t.Fatalf("create upstream model: %v", err) + } + } + + items, err := NewRepo(db).ListManagedUpstreamModels(ctx, upstream.ID) + if err != nil { + t.Fatalf("list managed upstream models: %v", err) + } + if len(items) != 2 || items[0].UpstreamModelName != "legacy-model" || items[1].UpstreamModelName != "sync-model" { + t.Fatalf("unexpected managed models: %+v", items) + } +} + func TestCreateUpstreamModelSQLiteDoesNotOverwriteExistingCatalogEntry(t *testing.T) { db := openChannelSQLiteTestDB(t) ctx := context.Background() diff --git a/backend/internal/repository/channel.go b/backend/internal/repository/channel.go index 0fe87ec43..c501798f7 100644 --- a/backend/internal/repository/channel.go +++ b/backend/internal/repository/channel.go @@ -463,7 +463,8 @@ type ChannelRepository interface { GetUpstreamModelByID(ctx context.Context, sourceID uint, upstreamID uint) (*domainchannel.UpstreamModel, error) GetUpstreamModelByUpstreamName(ctx context.Context, upstreamID uint, upstreamModelName string) (*domainchannel.UpstreamModel, error) DeleteUpstreamModel(ctx context.Context, sourceID uint, upstreamID uint) error - MarkMissingSyncedUpstreamModelsInactive(ctx context.Context, upstreamID uint, activeNames []string) (int64, error) + ListManagedUpstreamModels(ctx context.Context, upstreamID uint) ([]domainchannel.UpstreamModel, error) + ApplyUpstreamModelCatalogChanges(ctx context.Context, upstreamID uint, input ApplyUpstreamModelCatalogChangesInput) (int64, error) ListUpstreamModels(ctx context.Context, upstreamID uint, input ListChannelUpstreamModelsInput) ([]ChannelUpstreamModelListRow, int64, error) ListUpstreamModelsByNames(ctx context.Context, upstreamID uint, upstreamModelNames []string) ([]ChannelUpstreamModelListRow, error) GetUpstreamModelRouteByID(ctx context.Context, upstreamID uint, routeID uint) (*ChannelUpstreamModelListRow, error) @@ -491,3 +492,10 @@ type ChannelRepository interface { DeleteUpstreamCascade(ctx context.Context, upstreamID uint) error DeleteModelCascade(ctx context.Context, modelID uint) error } + +// ApplyUpstreamModelCatalogChangesInput 描述一次远端目录对账需要持久化的批量变更。 +type ApplyUpstreamModelCatalogChangesInput struct { + Create []domainchannel.UpstreamModel + Update []domainchannel.UpstreamModel + InactivateIDs []uint +} diff --git a/backend/internal/transport/http/channel/dto_response.go b/backend/internal/transport/http/channel/dto_response.go index cf53258b6..cb403de57 100644 --- a/backend/internal/transport/http/channel/dto_response.go +++ b/backend/internal/transport/http/channel/dto_response.go @@ -425,8 +425,20 @@ type UpstreamRemoteModelResponse struct { // UpstreamRemoteModelsResponse 上游远程模型预览列表响应 DTO。 type UpstreamRemoteModelsResponse struct { - Total int `json:"total"` - Items []UpstreamRemoteModelResponse `json:"items"` + Total int `json:"total"` + Items []UpstreamRemoteModelResponse `json:"items"` + SnapshotID string `json:"snapshotID"` + SyncPlan UpstreamModelSyncPlanResponse `json:"syncPlan"` +} + +// UpstreamModelSyncPlanResponse 描述确认同步后将应用的目录变化。 +type UpstreamModelSyncPlanResponse struct { + AddedModels []string `json:"addedModels"` + UpdatedModels []string `json:"updatedModels"` + ReactivatedModels []string `json:"reactivatedModels"` + InactivatedModels []string `json:"inactivatedModels"` + UnchangedModels []string `json:"unchangedModels"` + ProtectedModels []string `json:"protectedModels"` } func toUpstreamRemoteModelsResponse(d appchannel.UpstreamRemoteModelsData) UpstreamRemoteModelsResponse { @@ -445,7 +457,19 @@ func toUpstreamRemoteModelsResponse(d appchannel.UpstreamRemoteModelsData) Upstr AlreadyBound: item.AlreadyBound, }) } - return UpstreamRemoteModelsResponse{Total: d.Total, Items: items} + return UpstreamRemoteModelsResponse{ + Total: d.Total, + Items: items, + SnapshotID: d.SnapshotID, + SyncPlan: UpstreamModelSyncPlanResponse{ + AddedModels: stringList(d.SyncPlan.AddedModels), + UpdatedModels: stringList(d.SyncPlan.UpdatedModels), + ReactivatedModels: stringList(d.SyncPlan.ReactivatedModels), + InactivatedModels: stringList(d.SyncPlan.InactivatedModels), + UnchangedModels: stringList(d.SyncPlan.UnchangedModels), + ProtectedModels: stringList(d.SyncPlan.ProtectedModels), + }, + } } func stringList(items []string) []string { @@ -463,16 +487,24 @@ type UpstreamSyncModelResponse struct { KindsJSON string `json:"kindsJSON"` Status string `json:"status"` Created bool `json:"created"` + Updated bool `json:"updated"` + Reactivated bool `json:"reactivated"` + Protected bool `json:"protected"` } // SyncUpstreamModelsResponse 同步上游模型响应 DTO。 type SyncUpstreamModelsResponse struct { - TotalUpstream int `json:"totalUpstream"` - CreatedUpstreamModels int `json:"createdUpstreamModels"` - ExistingUpstreamModels int `json:"existingUpstreamModels"` - SkippedUpstreamModels int `json:"skippedUpstreamModels"` - InactivatedModels int64 `json:"inactivatedModels"` - SyncedModels []UpstreamSyncModelResponse `json:"syncedModels"` + SnapshotID string `json:"snapshotID"` + TotalUpstream int `json:"totalUpstream"` + CreatedUpstreamModels int `json:"createdUpstreamModels"` + UpdatedUpstreamModels int `json:"updatedUpstreamModels"` + UnchangedUpstreamModels int `json:"unchangedUpstreamModels"` + ProtectedUpstreamModels int `json:"protectedUpstreamModels"` + ExistingUpstreamModels int `json:"existingUpstreamModels"` + SkippedUpstreamModels int `json:"skippedUpstreamModels"` + InactivatedModels int64 `json:"inactivatedModels"` + ReactivatedModels int `json:"reactivatedModels"` + SyncedModels []UpstreamSyncModelResponse `json:"syncedModels"` } func toSyncUpstreamModelsResponse(d appchannel.SyncUpstreamModelsData) SyncUpstreamModelsResponse { @@ -485,15 +517,23 @@ func toSyncUpstreamModelsResponse(d appchannel.SyncUpstreamModelsData) SyncUpstr KindsJSON: m.KindsJSON, Status: m.Status, Created: m.Created, + Updated: m.Updated, + Reactivated: m.Reactivated, + Protected: m.Protected, }) } return SyncUpstreamModelsResponse{ - TotalUpstream: d.TotalUpstream, - CreatedUpstreamModels: d.CreatedUpstreamModels, - ExistingUpstreamModels: d.ExistingUpstreamModels, - SkippedUpstreamModels: d.SkippedUpstreamModels, - InactivatedModels: d.InactivatedModels, - SyncedModels: models, + SnapshotID: d.SnapshotID, + TotalUpstream: d.TotalUpstream, + CreatedUpstreamModels: d.CreatedUpstreamModels, + UpdatedUpstreamModels: d.UpdatedUpstreamModels, + UnchangedUpstreamModels: d.UnchangedUpstreamModels, + ProtectedUpstreamModels: d.ProtectedUpstreamModels, + ExistingUpstreamModels: d.ExistingUpstreamModels, + SkippedUpstreamModels: d.SkippedUpstreamModels, + InactivatedModels: d.InactivatedModels, + ReactivatedModels: d.ReactivatedModels, + SyncedModels: models, } } diff --git a/backend/internal/transport/http/channel/handler.go b/backend/internal/transport/http/channel/handler.go index 0b5d3b62f..d1d472632 100644 --- a/backend/internal/transport/http/channel/handler.go +++ b/backend/internal/transport/http/channel/handler.go @@ -771,7 +771,7 @@ func (h *Handler) TestUpstreamModelRoute(c *gin.Context) { // ListRemoteModels godoc // @Summary 管理员预览上游远程模型 -// @Description 调用上游 models 接口，仅返回可导入预览，不直接落库 +// @Description 调用上游 models 接口，返回可导入模型与目录变更预览，不直接落库 // @Tags llm // @Accept json // @Produce json @@ -809,15 +809,18 @@ func (h *Handler) ListRemoteModels(c *gin.Context) { // SyncUpstreamModels godoc // @Summary 管理员同步上游模型目录 -// @Description 调用上游 models 接口写入上游真实模型清单，不自动绑定平台模型 +// @Description 调用上游 models 接口获取完整目录，原子更新远端管理模型可用状态，不删除平台模型或路由配置 // @Tags llm // @Accept json // @Produce json // @Security BearerAuth // @Param id path int true "上游ID" +// @Param allow_empty query bool false "确认允许空模型目录对账" +// @Param expected_snapshot query string false "用户确认的远端目录快照标识" // @Success 200 {object} SyncUpstreamModelsResponseDoc // @Failure 400 {object} ErrorDoc // @Failure 404 {object} ErrorDoc +// @Failure 409 {object} ErrorDoc // @Failure 502 {object} ErrorDoc // @Failure 500 {object} ErrorDoc // @Router /admin/llm/upstreams/{id}/models/sync [post] @@ -828,7 +831,10 @@ func (h *Handler) SyncUpstreamModels(c *gin.Context) { return } - data, err := h.service.SyncUpstreamModels(c.Request.Context(), upstreamID) + data, err := h.service.SyncUpstreamModels(c.Request.Context(), upstreamID, appchannel.SyncUpstreamModelsInput{ + AllowEmpty: c.Query("allow_empty") == "true", + ExpectedSnapshot: c.Query("expected_snapshot"), + }) if err != nil { switch { case errors.Is(err, appchannel.ErrUpstreamNotFound): @@ -837,6 +843,10 @@ func (h *Handler) SyncUpstreamModels(c *gin.Context) { response.Error(c, http.StatusBadRequest, "no active api key") case errors.Is(err, appchannel.ErrRemoteModelsUnavailable): response.Error(c, http.StatusBadGateway, "remote models unavailable") + case errors.Is(err, appchannel.ErrEmptyRemoteModels): + response.ErrorWithCode(c, http.StatusConflict, "llm.remote_models_empty_confirmation_required", "remote models snapshot is empty") + case errors.Is(err, appchannel.ErrRemoteModelsSnapshotChanged): + response.ErrorWithCode(c, http.StatusConflict, "llm.remote_models_snapshot_changed", "remote models snapshot changed") default: response.Error(c, http.StatusInternalServerError, "sync upstream models failed") } diff --git a/frontend/components/ui/tooltip.tsx b/frontend/components/ui/tooltip.tsx index fbc685421..e32f82d9b 100644 --- a/frontend/components/ui/tooltip.tsx +++ b/frontend/components/ui/tooltip.tsx @@ -36,14 +36,19 @@ function TooltipTrigger({ return } +type TooltipContentProps = React.ComponentProps & { + portalContainer?: React.ComponentProps["container"] +} + function TooltipContent({ className, + portalContainer, sideOffset = 0, children, ...props -}: React.ComponentProps) { +}: TooltipContentProps) { return ( - + { return authedRequest( `/api/v1/admin/llm/upstreams/${upstreamID}/models/remote`, - { accessToken }, + { accessToken, signal }, + true, + ); +} + +export async function syncAdminLLMUpstreamModels( + accessToken: string, + upstreamID: number, + options: { allowEmpty?: boolean; expectedSnapshot?: string; signal?: AbortSignal } = {}, +): Promise { + const params = new URLSearchParams(); + if (options.allowEmpty) { + params.set("allow_empty", "true"); + } + if (options.expectedSnapshot) { + params.set("expected_snapshot", options.expectedSnapshot); + } + const query = params.size > 0 ? `?${params.toString()}` : ""; + return authedRequest( + `/api/v1/admin/llm/upstreams/${upstreamID}/models/sync${query}`, + { method: "POST", accessToken, signal: options.signal }, true, ); } @@ -303,10 +325,11 @@ export async function importAdminLLMUpstreamModels( accessToken: string, upstreamID: number, payload: ImportAdminLLMUpstreamModelsRequest, + signal?: AbortSignal, ): Promise { return authedRequest( `/api/v1/admin/llm/upstreams/${upstreamID}/models/import`, - { method: "POST", accessToken, body: payload }, + { method: "POST", accessToken, body: payload, signal }, true, ); } diff --git a/frontend/features/admin/api/llm.types.ts b/frontend/features/admin/api/llm.types.ts index f53798b12..200ce9dfc 100644 --- a/frontend/features/admin/api/llm.types.ts +++ b/frontend/features/admin/api/llm.types.ts @@ -29,6 +29,7 @@ import type { ReorderModelsRequest, SetModelProtocolsRequest, SetModelsDisplayGroupRequest, + SyncUpstreamModelsResponse, UpdateModelDisplayGroupRequest, UpdateModelRequest, UpdateModelUpstreamSourceRequest, @@ -306,6 +307,8 @@ export type ListAdminLLMRemoteModelsData = Omit & { results: Array< Omit & { diff --git a/frontend/features/admin/api/permission-groups.ts b/frontend/features/admin/api/permission-groups.ts index fa6211049..2cd20063f 100644 --- a/frontend/features/admin/api/permission-groups.ts +++ b/frontend/features/admin/api/permission-groups.ts @@ -44,10 +44,10 @@ export type DeletePermissionGroupResult = Omit { +export async function listPermissionGroups(accessToken: string, signal?: AbortSignal): Promise { const data = await authedRequest( "/api/v1/admin/permission-groups", - { accessToken }, + { accessToken, signal }, true, ); return data.results ?? []; diff --git a/frontend/features/admin/components/sections/groups/permission-group-selector.tsx b/frontend/features/admin/components/sections/groups/permission-group-selector.tsx index 6a62df705..0cb38dab7 100644 --- a/frontend/features/admin/components/sections/groups/permission-group-selector.tsx +++ b/frontend/features/admin/components/sections/groups/permission-group-selector.tsx @@ -18,6 +18,7 @@ type PermissionGroupSelectorProps = { matchedIDs?: number[]; disabled?: boolean; loading?: boolean; + triggerPrefix?: string; placeholder: string; emptyLabel: string; autoBadgeLabel: string; @@ -30,6 +31,7 @@ export function PermissionGroupSelector({ matchedIDs = [], disabled, loading, + triggerPrefix, placeholder, emptyLabel, autoBadgeLabel, @@ -68,6 +70,9 @@ export function PermissionGroupSelector({ disabled={disabled || loading} className="h-8 w-full justify-between gap-2 border-input/40 bg-transparent px-3 py-1 text-xs font-normal hover:bg-transparent focus-visible:border-ring/60 focus-visible:ring-[1px] focus-visible:ring-ring/40 has-[>svg]:px-3" > + {triggerPrefix ? ( + {triggerPrefix} + ) : null} {triggerLabel} diff --git a/frontend/features/admin/components/sections/upstreams/upstreams-models-dialog.tsx b/frontend/features/admin/components/sections/upstreams/upstreams-models-dialog.tsx index 1dca992ca..fa4c4d0e5 100644 --- a/frontend/features/admin/components/sections/upstreams/upstreams-models-dialog.tsx +++ b/frontend/features/admin/components/sections/upstreams/upstreams-models-dialog.tsx @@ -1,6 +1,6 @@ import * as React from "react"; import { toast } from "sonner"; -import { Activity, Cable, Check, ChevronDownIcon, CloudDownload, Plus, Tags, ToggleLeft, Trash2 } from "lucide-react"; +import { Activity, Cable, Check, ChevronDownIcon, CloudDownload, Plus, RefreshCw, Search, Tags, ToggleLeft, Trash2 } from "lucide-react"; import { useTranslations } from "next-intl"; import { AlertDialog, @@ -16,6 +16,7 @@ import { Button } from "@/components/ui/button"; import { Checkbox } from "@/components/ui/checkbox"; import { Dialog, + DialogCollapsible, DialogContent, DialogDescription, DialogFooter, @@ -59,6 +60,7 @@ import { AdminBulkConfirmDialog } from "@/features/admin/components/bulk-confirm import { Badge } from "@/components/ui/badge"; import { useLocalizedErrorMessage } from "@/i18n/use-localized-error"; import { resolveAccessToken } from "@/shared/auth/resolve-access-token"; +import { ApiError } from "@/shared/api/http-client"; import { useDialogSnapshot } from "@/shared/hooks/use-dialog-snapshot"; import { mergeBatchResultData, @@ -67,16 +69,10 @@ import { import { batchDeleteAdminLLMUpstreamModels, deleteAdminLLMUpstreamModel, - importAdminLLMUpstreamModels, - listAdminLLMRemoteModels, listAdminLLMUpstreamModels, testAdminLLMUpstreamModelRoute, upsertAdminLLMUpstreamModel, } from "@/features/admin/api"; -import { - listPermissionGroups, - type PermissionGroup, -} from "@/features/admin/api/permission-groups"; import { cn } from "@/lib/utils"; import type { AdminLLMAdapter, @@ -105,6 +101,11 @@ import { type RowDraft, } from "@/features/admin/model/upstreams-models"; import { PermissionGroupSelector } from "@/features/admin/components/sections/groups/permission-group-selector"; +import { + isUpstreamModelSyncAbort, + UpstreamModelBindingsApplyError, + useUpstreamModelSync, +} from "@/features/admin/hooks/use-upstream-model-sync"; function KindsDropdown({ value, @@ -505,74 +506,60 @@ function RemoteModelsDialog({ const t = useTranslations("adminUpstreams"); const commonT = useTranslations("common"); const resolveErrorMessage = useLocalizedErrorMessage(); - const [loading, setLoading] = React.useState(false); const [importing, setImporting] = React.useState(false); const [remoteItems, setRemoteItems] = React.useState([]); const [selected, setSelected] = React.useState>(new Set()); const [draftPlatformModelNames, setDraftPlatformModelNames] = React.useState>(new Map()); const [query, setQuery] = React.useState(""); - const [permissionGroups, setPermissionGroups] = React.useState([]); const [permissionGroupIDs, setPermissionGroupIDs] = React.useState([]); - const [permissionGroupsLoading, setPermissionGroupsLoading] = React.useState(false); + const [syncConfirmationOpen, setSyncConfirmationOpen] = React.useState(false); + const [tooltipPortalContainer, setTooltipPortalContainer] = React.useState(null); + const { + catalog, + catalogError, + catalogLoading: loading, + permissionGroups, + permissionGroupsError, + permissionGroupsLoading, + reloadCatalog: loadRemoteModels, + applySync, + } = useUpstreamModelSync(open, upstream?.id ?? null); + const remoteTotal = catalog?.total ?? null; + const remoteSnapshotID = catalog?.snapshotID ?? ""; + const syncPlan = catalog?.syncPlan ?? null; - const loadRemoteModels = React.useCallback(async () => { - if (!upstream) return; + React.useEffect(() => { setRemoteItems([]); setSelected(new Set()); setDraftPlatformModelNames(new Map()); setQuery(""); - setLoading(true); - try { - const token = await resolveAccessToken(); - const data = await listAdminLLMRemoteModels(token, upstream.id); - const syncableItems = dedupeRemoteModels(data.items.filter((i) => !i.alreadyBound)); - setRemoteItems(syncableItems); - setSelected(new Set(syncableItems.map((i) => i.upstreamModelName))); - setDraftPlatformModelNames(createDraftPlatformModelNameMap(syncableItems)); - } catch (err) { - toast.error(t("modelsDialog.remoteLoadFailed"), { description: resolveErrorMessage(err) }); - onOpenChange(false); - } finally { - setLoading(false); - } - }, [onOpenChange, resolveErrorMessage, t, upstream]); + if (!catalog) return; + const syncableItems = dedupeRemoteModels(catalog.items.filter((item) => !item.alreadyBound)); + setRemoteItems(syncableItems); + setSelected(new Set(syncableItems.map((item) => item.upstreamModelName))); + setDraftPlatformModelNames(createDraftPlatformModelNameMap(syncableItems)); + }, [catalog]); React.useEffect(() => { - if (!open || !upstream) return; - void loadRemoteModels(); - }, [loadRemoteModels, open, upstream]); + if (!catalogError) return; + toast.error(t("modelsDialog.remoteLoadFailed"), { description: resolveErrorMessage(catalogError) }); + onOpenChange(false); + }, [catalogError, onOpenChange, resolveErrorMessage, t]); + + React.useEffect(() => { + if (!permissionGroupsError) return; + toast.error(t("modelsDialog.permissionGroupsLoadFailed"), { description: resolveErrorMessage(permissionGroupsError) }); + }, [permissionGroupsError, resolveErrorMessage, t]); React.useEffect(() => { if (!open) { - setPermissionGroups([]); setPermissionGroupIDs([]); - setPermissionGroupsLoading(false); + setSyncConfirmationOpen(false); return; } - let cancelled = false; - setPermissionGroupsLoading(true); - void (async () => { - try { - const token = await resolveAccessToken(); - const groups = await listPermissionGroups(token); - if (!cancelled) { - setPermissionGroups(groups); - } - } catch (err) { - if (!cancelled) { - setPermissionGroups([]); - toast.error(t("modelsDialog.permissionGroupsLoadFailed"), { description: resolveErrorMessage(err) }); - } - } finally { - if (!cancelled) { - setPermissionGroupsLoading(false); - } - } - })(); - return () => { - cancelled = true; - }; - }, [open, resolveErrorMessage, t]); + const defaultGroup = permissionGroups.find((group) => group.isDefault); + setPermissionGroupIDs(defaultGroup ? [defaultGroup.id] : []); + }, [open, permissionGroups]); function setDraftPlatformModelName(name: string, platformModelName: string) { setDraftPlatformModelNames((prev) => new Map(prev).set(name, platformModelName)); @@ -619,172 +606,314 @@ function RemoteModelsDialog({ }); }, [normalizedQuery, remoteItems, t]); const selectedRemoteItems = React.useMemo( - () => filteredRemoteItems.filter((item) => selected.has(item.upstreamModelName)), - [filteredRemoteItems, selected], + () => remoteItems.filter((item) => selected.has(item.upstreamModelName)), + [remoteItems, selected], ); const allSelected = filteredRemoteItems.length > 0 && filteredRemoteItems.every((i) => selected.has(i.upstreamModelName)); const someSelected = filteredRemoteItems.some((i) => selected.has(i.upstreamModelName)); const hasQuery = normalizedQuery.length > 0; + const catalogChangeCount = syncPlan + ? syncPlan.addedModels.length + + syncPlan.updatedModels.length + + syncPlan.reactivatedModels.length + + syncPlan.inactivatedModels.length + : 0; + const hasCatalogChanges = catalogChangeCount > 0; + const hasSyncWork = hasCatalogChanges || selectedRemoteItems.length > 0; + const syncPlanStatuses = syncPlan + ? [ + { key: "added", label: t("modelsDialog.syncPlanAddedLabel"), models: syncPlan.addedModels }, + { key: "updated", label: t("modelsDialog.syncPlanUpdatedLabel"), models: syncPlan.updatedModels }, + { key: "reactivated", label: t("modelsDialog.syncPlanReactivatedLabel"), models: syncPlan.reactivatedModels }, + { key: "inactivated", label: t("modelsDialog.syncPlanInactivatedLabel"), models: syncPlan.inactivatedModels }, + { key: "unchanged", label: t("modelsDialog.syncPlanUnchangedLabel"), models: syncPlan.unchangedModels }, + { key: "protected", label: t("modelsDialog.syncPlanProtectedLabel"), models: syncPlan.protectedModels }, + ] + : []; + + function formatCatalogSummary(result: Awaited>["catalog"]) { + return t("modelsDialog.catalogSyncSummary", { + createdUpstreamModels: result.createdUpstreamModels, + updatedUpstreamModels: result.updatedUpstreamModels, + reactivatedModels: result.reactivatedModels, + inactivatedModels: result.inactivatedModels, + unchangedUpstreamModels: result.unchangedUpstreamModels, + protectedUpstreamModels: result.protectedUpstreamModels, + }); + } - async function handleSyncBindings() { - if (!upstream || selectedRemoteItems.length === 0) return; + async function executeSyncBindings(allowEmpty: boolean) { + if (!upstream) return; setImporting(true); try { - const token = await resolveAccessToken(); - const items = selectedRemoteItems.map((i) => ({ - upstreamModelName: i.upstreamModelName, - platformModelName: (draftPlatformModelNames.get(i.upstreamModelName) || i.upstreamModelName).trim(), - protocols: i.suggestedProtocols?.length - ? sortProtocolsForDisplay(i.suggestedProtocols) - : i.suggestedProtocol - ? [i.suggestedProtocol] + const items = selectedRemoteItems.map((item) => ({ + upstreamModelName: item.upstreamModelName, + platformModelName: (draftPlatformModelNames.get(item.upstreamModelName) || item.upstreamModelName).trim(), + protocols: item.suggestedProtocols?.length + ? sortProtocolsForDisplay(item.suggestedProtocols) + : item.suggestedProtocol + ? [item.suggestedProtocol] : undefined, - kindsJSON: i.suggestedKindsJSON || undefined, + kindsJSON: item.suggestedKindsJSON || undefined, })); - const result = await importAdminLLMUpstreamModels(token, upstream.id, { + const result = await applySync({ + allowEmpty, + expectedSnapshot: remoteSnapshotID, items, permissionGroupIDs: permissionGroupIDs.length > 0 ? permissionGroupIDs : undefined, }); - const description = summarizeImportResult(result, { - importSummary: (summary) => t("modelsDialog.importSummary", summary), - }); - if (result.failedCount > 0) { - toast.error(t("modelsDialog.importPartialFailed"), { - description, - }); + const catalogSummary = formatCatalogSummary(result.catalog); + const summaries = [catalogSummary]; + + if (result.bindings) { + summaries.push(summarizeImportResult(result.bindings, { + importSummary: (summary) => t("modelsDialog.importSummary", summary), + })); + if (result.bindings.failedCount > 0) { + toast.error(t("modelsDialog.importPartialFailed"), { + description: summaries.join(" · "), + }); + } else { + toast.success(t("modelsDialog.importDone"), { + description: summaries.join(" · "), + }); + } } else { toast.success(t("modelsDialog.importDone"), { - description, + description: summaries.join(" · "), }); } onImported(); onOpenChange(false); } catch (err) { - toast.error(t("modelsDialog.importFailed"), { description: resolveErrorMessage(err) }); + if (isUpstreamModelSyncAbort(err)) return; + const catalogSummary = err instanceof UpstreamModelBindingsApplyError + ? formatCatalogSummary(err.catalog) + : ""; + const reportedError = err instanceof UpstreamModelBindingsApplyError ? err.originalError : err; + toast.error(t(catalogSummary ? "modelsDialog.importAfterSyncFailed" : "modelsDialog.importFailed"), { + description: [catalogSummary, resolveErrorMessage(reportedError)].filter(Boolean).join(" · "), + }); + if (catalogSummary) { + onImported(); + } else if (err instanceof ApiError && err.errorCode === "llm.remote_models_snapshot_changed") { + await loadRemoteModels(); + } } finally { setImporting(false); } } + function handleSyncBindings() { + if ((syncPlan?.inactivatedModels.length ?? 0) > 0) { + setSyncConfirmationOpen(true); + return; + } + void executeSyncBindings(remoteTotal === 0); + } + return ( - - + + {t("modelsDialog.syncTitle", { name: upstream?.name ?? "" })} {t("modelsDialog.syncDescription")} -
-
- void loadRemoteModels()} - /> -
- - +
+
+
+ {t("modelsDialog.syncPlanTitle")} + {loading && !syncPlan ? ( + + {t("modelsDialog.syncPlanLoading")} + + ) : ( + syncPlanStatuses.map((status) => { + const destructive = status.key === "inactivated" && status.models.length > 0; + return ( + + + {status.label} + {status.models.length} + + +

+ {status.label} · {status.models.length} +

+ {status.models.length > 0 ? ( +
+ {status.models.map((modelName) => ( +
+ {modelName} +
+ ))} +
+ ) : ( +

{t("modelsDialog.syncPlanNoModels")}

+ )} + + + ); + }) + )} +
+
-
- - - - -
- toggleAll(v === true)} - aria-label={t("table.selectAll")} - /> -
- - {t("modelsDialog.upstreamModelName")} - {t("modelsDialog.platformModelName")} - {t("fields.status")} - - - - {loading && filteredRemoteItems.length === 0 ? ( - - ) : null} - {!loading && filteredRemoteItems.length === 0 ? ( - - {hasQuery ? t("modelsDialog.noMatchedModels") : t("modelsDialog.noSyncableModels")} - - ) : null} - {filteredRemoteItems.map((item) => ( - - -
- toggleOne(item.upstreamModelName, v === true)} - aria-label={item.upstreamModelName} - /> -
- - - - {item.upstreamModelName} - - - -
- setDraftPlatformModelName(item.upstreamModelName, e.target.value)} - /> -
- - -
- - {t(`modelsDialog.remoteStatus.${remoteModelStatusKey(item)}`)} - -
- - - ))} - -
-
+ 0} className="shrink-0"> +
+
+
+ + setQuery(event.target.value)} + disabled={loading || importing} + className="bg-background pl-8" + /> +
+
+ +
+
+
+ + + + +
+ toggleAll(v === true)} + aria-label={t("table.selectAll")} + /> +
+ + {t("modelsDialog.upstreamModelName")} + {t("modelsDialog.platformModelName")} + {t("fields.status")} + + + + {!loading && filteredRemoteItems.length === 0 ? ( + + {hasQuery ? t("modelsDialog.noMatchedModels") : t("modelsDialog.noSyncableModels")} + + ) : null} + {filteredRemoteItems.map((item) => ( + + +
+ toggleOne(item.upstreamModelName, v === true)} + aria-label={item.upstreamModelName} + /> +
+ + + + {item.upstreamModelName} + + + +
+ setDraftPlatformModelName(item.upstreamModelName, e.target.value)} + /> +
+ + +
+ + {t(`modelsDialog.remoteStatus.${remoteModelStatusKey(item)}`)} + +
+ + + ))} + +
+
+
+ + + +
+
+ {loading || remoteItems.length > 0 ? ( + {t("modelsDialog.loadingRemote")} + ) : ( + t("modelsDialog.noSyncableModels") + )} +
+
+ - + - {t("modelsDialog.syncSummary", { - total: remoteItems.length, - shown: filteredRemoteItems.length, - selected: selectedRemoteItems.length, - hasQuery: hasQuery ? "true" : "false", - hasSelected: selectedRemoteItems.length > 0 ? "true" : "false", - })} + {remoteItems.length > 0 + ? t("modelsDialog.syncSummary", { + total: remoteItems.length, + shown: filteredRemoteItems.length, + selected: selectedRemoteItems.length, + hasQuery: hasQuery ? "true" : "false", + hasSelected: selectedRemoteItems.length > 0 ? "true" : "false", + }) + : t("modelsDialog.remoteCatalogSummary", { total: remoteTotal ?? 0 })}
+ + + + {t("modelsDialog.inactivateSyncTitle")} + +
+

+ {t("modelsDialog.inactivateSyncSummary", { + count: syncPlan?.inactivatedModels.length ?? 0, + })} +

+

{t("modelsDialog.inactivateSyncImpact")}

+
+ + + + {commonT("actions.cancel")} + { + setSyncConfirmationOpen(false); + void executeSyncBindings(remoteTotal === 0); + }} + > + {t("modelsDialog.confirmApplySync")} + + + +
); } diff --git a/frontend/features/admin/hooks/use-upstream-model-sync.ts b/frontend/features/admin/hooks/use-upstream-model-sync.ts new file mode 100644 index 000000000..174d11628 --- /dev/null +++ b/frontend/features/admin/hooks/use-upstream-model-sync.ts @@ -0,0 +1,201 @@ +import * as React from "react"; +import { + importAdminLLMUpstreamModels, + listAdminLLMRemoteModels, + syncAdminLLMUpstreamModels, +} from "@/features/admin/api"; +import { + listPermissionGroups, + type PermissionGroup, +} from "@/features/admin/api/permission-groups"; +import type { + ImportAdminLLMUpstreamModelsData, + ImportAdminLLMUpstreamModelsRequest, + ListAdminLLMRemoteModelsData, + SyncAdminLLMUpstreamModelsData, +} from "@/features/admin/api/llm.types"; +import { resolveAccessToken } from "@/shared/auth/resolve-access-token"; + +type ApplyUpstreamModelSyncInput = { + allowEmpty: boolean; + expectedSnapshot: string; + items: ImportAdminLLMUpstreamModelsRequest["items"]; + permissionGroupIDs?: number[]; +}; + +export type ApplyUpstreamModelSyncResult = { + catalog: SyncAdminLLMUpstreamModelsData; + bindings: ImportAdminLLMUpstreamModelsData | null; +}; + +export class UpstreamModelBindingsApplyError extends Error { + readonly catalog: SyncAdminLLMUpstreamModelsData; + readonly originalError: unknown; + + constructor(catalog: SyncAdminLLMUpstreamModelsData, originalError: unknown) { + super("upstream model catalog applied before route import failed"); + this.name = "UpstreamModelBindingsApplyError"; + this.catalog = catalog; + this.originalError = originalError; + } +} + +export function isUpstreamModelSyncAbort(error: unknown): boolean { + return error instanceof Error && error.name === "AbortError"; +} + +function throwIfAborted(signal: AbortSignal) { + if (!signal.aborted) return; + if (signal.reason instanceof Error && signal.reason.name === "AbortError") { + throw signal.reason; + } + const error = new Error("The operation was aborted"); + error.name = "AbortError"; + throw error; +} + +export function useUpstreamModelSync(open: boolean, upstreamID: number | null) { + const [catalog, setCatalog] = React.useState(null); + const [catalogLoading, setCatalogLoading] = React.useState(false); + const [catalogError, setCatalogError] = React.useState(null); + const [permissionGroups, setPermissionGroups] = React.useState([]); + const [permissionGroupsLoading, setPermissionGroupsLoading] = React.useState(false); + const [permissionGroupsError, setPermissionGroupsError] = React.useState(null); + const catalogRequestRef = React.useRef<{ controller: AbortController; version: number } | null>(null); + const catalogRequestVersionRef = React.useRef(0); + const applyControllerRef = React.useRef(null); + + const cancelCatalogRequest = React.useCallback(() => { + catalogRequestVersionRef.current += 1; + catalogRequestRef.current?.controller.abort(); + catalogRequestRef.current = null; + }, []); + + const reloadCatalog = React.useCallback(async () => { + if (!upstreamID) return; + cancelCatalogRequest(); + const controller = new AbortController(); + const version = catalogRequestVersionRef.current; + catalogRequestRef.current = { controller, version }; + setCatalogError(null); + setCatalogLoading(true); + try { + const token = await resolveAccessToken(); + throwIfAborted(controller.signal); + const data = await listAdminLLMRemoteModels(token, upstreamID, controller.signal); + if (catalogRequestRef.current?.version === version) { + setCatalog(data); + } + } catch (error) { + if (catalogRequestRef.current?.version === version && !isUpstreamModelSyncAbort(error)) { + setCatalogError(error); + } + } finally { + if (catalogRequestRef.current?.version === version) { + catalogRequestRef.current = null; + setCatalogLoading(false); + } + } + }, [cancelCatalogRequest, upstreamID]); + + React.useEffect(() => { + if (!open || !upstreamID) { + cancelCatalogRequest(); + applyControllerRef.current?.abort(); + applyControllerRef.current = null; + setCatalog(null); + setCatalogError(null); + setCatalogLoading(false); + return; + } + setCatalog(null); + void reloadCatalog(); + return () => { + cancelCatalogRequest(); + applyControllerRef.current?.abort(); + applyControllerRef.current = null; + }; + }, [cancelCatalogRequest, open, reloadCatalog, upstreamID]); + + React.useEffect(() => { + if (!open) { + setPermissionGroups([]); + setPermissionGroupsError(null); + setPermissionGroupsLoading(false); + return; + } + const controller = new AbortController(); + setPermissionGroups([]); + setPermissionGroupsError(null); + setPermissionGroupsLoading(true); + void (async () => { + try { + const token = await resolveAccessToken(); + throwIfAborted(controller.signal); + const groups = await listPermissionGroups(token, controller.signal); + if (!controller.signal.aborted) { + setPermissionGroups(groups); + } + } catch (error) { + if (!controller.signal.aborted && !isUpstreamModelSyncAbort(error)) { + setPermissionGroupsError(error); + } + } finally { + if (!controller.signal.aborted) { + setPermissionGroupsLoading(false); + } + } + })(); + return () => controller.abort(); + }, [open]); + + const applySync = React.useCallback(async (input: ApplyUpstreamModelSyncInput): Promise => { + if (!upstreamID) { + throw new Error("upstream is required"); + } + applyControllerRef.current?.abort(); + const controller = new AbortController(); + applyControllerRef.current = controller; + try { + const token = await resolveAccessToken(); + throwIfAborted(controller.signal); + const catalogResult = await syncAdminLLMUpstreamModels(token, upstreamID, { + allowEmpty: input.allowEmpty, + expectedSnapshot: input.expectedSnapshot, + signal: controller.signal, + }); + throwIfAborted(controller.signal); + const currentRemoteNames = new Set(catalogResult.syncedModels.map((item) => item.upstreamModelName)); + const items = input.items.filter((item) => currentRemoteNames.has(item.upstreamModelName)); + if (items.length === 0) { + return { catalog: catalogResult, bindings: null }; + } + try { + const bindings = await importAdminLLMUpstreamModels(token, upstreamID, { + items, + permissionGroupIDs: input.permissionGroupIDs, + }, controller.signal); + throwIfAborted(controller.signal); + return { catalog: catalogResult, bindings }; + } catch (error) { + if (isUpstreamModelSyncAbort(error)) throw error; + throw new UpstreamModelBindingsApplyError(catalogResult, error); + } + } finally { + if (applyControllerRef.current === controller) { + applyControllerRef.current = null; + } + } + }, [upstreamID]); + + return { + catalog, + catalogError, + catalogLoading, + permissionGroups, + permissionGroupsError, + permissionGroupsLoading, + reloadCatalog, + applySync, + }; +} diff --git a/frontend/i18n/messages/en-US/admin-upstreams.json b/frontend/i18n/messages/en-US/admin-upstreams.json index ab8864697..d2041505c 100644 --- a/frontend/i18n/messages/en-US/admin-upstreams.json +++ b/frontend/i18n/messages/en-US/admin-upstreams.json @@ -185,11 +185,31 @@ "importPartialFailed": "Some models failed to sync", "importDone": "Model sync completed", "importFailed": "Model sync failed", + "importAfterSyncFailed": "The model catalog was synced, but platform model or route creation failed", "importSummary": "Imported {importedCount}, failed {failedCount}, created {createdPlatform} platform models, created {createdRoutes} routes, reused {existingRoutes} routes", + "catalogSyncSummary": "Catalog: {createdUpstreamModels} added, {updatedUpstreamModels} updated, {reactivatedModels} restored, {inactivatedModels} deactivated, {unchangedUpstreamModels} unchanged, {protectedUpstreamModels} preserved", "syncTitle": "Sync upstream models - {name}", - "syncDescription": "Fetch remote models, confirm platform model names, then create platform models and route bindings.", + "syncDescription": "Preview remote catalog changes, then update the catalog and bind selected models after confirmation.", + "syncPlanTitle": "Remote catalog", + "syncPlanLoading": "Checking the remote catalog…", + "syncPlanAddedLabel": "Add", + "syncPlanUpdatedLabel": "Update", + "syncPlanReactivatedLabel": "Restore", + "syncPlanInactivatedLabel": "Deactivate", + "syncPlanUnchangedLabel": "Unchanged", + "syncPlanProtectedLabel": "Preserve", + "syncPlanNoModels": "No matching models", + "inactivateSyncTitle": "Sync the remote catalog?", + "inactivateSyncSummary": "This sync will deactivate {count} models that are no longer returned by the upstream.", + "inactivateSyncImpact": "This action only updates the current upstream catalog status. Platform models and route configuration remain intact; models will be restored automatically if the upstream returns them in a later sync.", + "confirmApplySync": "Apply changes", + "applySync": "Apply sync", + "syncPlanCurrent": "Up to date", + "loadingRemote": "Loading remote models", + "remoteCatalogSummary": "{total} models in the remote catalog", "syncSearchPlaceholder": "Search upstream model, platform model, protocol, or status", "permissionGroups": "Permission groups", + "importPermissionGroups": "Add to groups", "permissionGroupsPlaceholder": "Select groups to add imported models to", "permissionGroupsEmpty": "No permission groups", "permissionGroupsAutoBadge": "Auto", @@ -197,8 +217,8 @@ "reloadRemote": "Reload", "upstreamModelName": "Upstream model name", "noMatchedModels": "No matching models", - "noSyncableModels": "No models available to sync", - "syncSummary": "{total} available{hasQuery, select, true {, {shown} shown} other {}}{hasSelected, select, true {, {selected} selected} other {}}", + "noSyncableModels": "No models need new bindings", + "syncSummary": "{total} awaiting binding{hasQuery, select, true {, {shown} shown} other {}}{hasSelected, select, true {, {selected} selected} other {}}", "syncing": "Syncing", "bindingNamesRequired": "Upstream model name and platform model name are required", "bindingCreated": "Route binding created", diff --git a/frontend/i18n/messages/en-US/errors.json b/frontend/i18n/messages/en-US/errors.json index a42e83e3a..aa784dfe0 100644 --- a/frontend/i18n/messages/en-US/errors.json +++ b/frontend/i18n/messages/en-US/errors.json @@ -214,6 +214,8 @@ "platformModelNameExists": "Platform model name already exists.", "platformModelNameRequired": "Platform model name is required.", "protocolRequired": "Protocol is required.", + "remoteModelsEmptyConfirmationRequired": "The upstream returned an empty model catalog. Refresh and confirm before syncing it.", + "remoteModelsSnapshotChanged": "The upstream model catalog changed after the preview. The latest changes have been reloaded for confirmation.", "remoteModelsUnavailable": "Remote models are unavailable.", "routeConflict": "This upstream model already has a binding for the same platform model and protocol. Remove the duplicate binding before saving.", "routesUnavailable": "All model routes are unavailable." diff --git a/frontend/i18n/messages/zh-CN/admin-upstreams.json b/frontend/i18n/messages/zh-CN/admin-upstreams.json index 11fd33c9b..1ab348877 100644 --- a/frontend/i18n/messages/zh-CN/admin-upstreams.json +++ b/frontend/i18n/messages/zh-CN/admin-upstreams.json @@ -185,11 +185,31 @@ "importPartialFailed": "部分模型同步失败", "importDone": "模型同步完成", "importFailed": "模型同步失败", + "importAfterSyncFailed": "模型目录已同步，但创建平台模型或路由绑定失败", "importSummary": "成功 {importedCount} 条，失败 {failedCount} 条，新建平台模型 {createdPlatform} 个，新建路由 {createdRoutes} 条，复用路由 {existingRoutes} 条", + "catalogSyncSummary": "目录新增 {createdUpstreamModels} 个，更新 {updatedUpstreamModels} 个，恢复 {reactivatedModels} 个，停用 {inactivatedModels} 个，未变 {unchangedUpstreamModels} 个，保留 {protectedUpstreamModels} 个", "syncTitle": "同步上游模型 - {name}", - "syncDescription": "拉取远端模型，确认平台模型名后创建平台模型和路由绑定。", + "syncDescription": "预览远端目录变化，确认后更新目录并绑定选中的模型。", + "syncPlanTitle": "远端目录", + "syncPlanLoading": "正在检查远端目录…", + "syncPlanAddedLabel": "新增", + "syncPlanUpdatedLabel": "更新", + "syncPlanReactivatedLabel": "恢复", + "syncPlanInactivatedLabel": "停用", + "syncPlanUnchangedLabel": "未变", + "syncPlanProtectedLabel": "保留", + "syncPlanNoModels": "暂无对应模型", + "inactivateSyncTitle": "确认同步远端目录？", + "inactivateSyncSummary": "本次同步将停用 {count} 个上游已不再返回的模型。", + "inactivateSyncImpact": "此次操作仅调整当前上游的目录状态。平台模型和路由配置将完整保留；模型重新出现在上游后，同步时将自动恢复。", + "confirmApplySync": "确认应用", + "applySync": "应用同步", + "syncPlanCurrent": "已是最新", + "loadingRemote": "正在获取远端模型", + "remoteCatalogSummary": "远端目录 {total} 个模型", "syncSearchPlaceholder": "搜索上游模型名、平台模型名、协议或状态", "permissionGroups": "权限组", + "importPermissionGroups": "加入权限组", "permissionGroupsPlaceholder": "选择本次导入模型要加入的权限组", "permissionGroupsEmpty": "暂无权限组", "permissionGroupsAutoBadge": "自动", @@ -197,8 +217,8 @@ "reloadRemote": "重新拉取", "upstreamModelName": "上游模型名", "noMatchedModels": "暂无匹配的模型", - "noSyncableModels": "暂无可同步的模型", - "syncSummary": "共 {total} 个可同步{hasQuery, select, true {，显示 {shown} 个} other {}}{hasSelected, select, true {，已选 {selected} 个} other {}}", + "noSyncableModels": "暂无需要新建绑定的模型", + "syncSummary": "待绑定 {total} 个{hasQuery, select, true {，显示 {shown} 个} other {}}{hasSelected, select, true {，已选 {selected} 个} other {}}", "syncing": "同步中", "bindingNamesRequired": "上游模型名和平台模型名不能为空", "bindingCreated": "路由绑定已创建", diff --git a/frontend/i18n/messages/zh-CN/errors.json b/frontend/i18n/messages/zh-CN/errors.json index 98c68bbad..bc2c067f0 100644 --- a/frontend/i18n/messages/zh-CN/errors.json +++ b/frontend/i18n/messages/zh-CN/errors.json @@ -214,6 +214,8 @@ "platformModelNameExists": "平台模型名称已存在。", "platformModelNameRequired": "平台模型名称不能为空。", "protocolRequired": "协议不能为空。", + "remoteModelsEmptyConfirmationRequired": "上游返回了空模型目录，请刷新后确认是否继续同步。", + "remoteModelsSnapshotChanged": "上游模型目录在预览后发生了变化，已重新加载，请确认最新变更。", "remoteModelsUnavailable": "远程模型列表暂不可用。", "routeConflict": "该上游模型已存在相同平台模型和协议的绑定，请删除重复绑定后再保存。", "routesUnavailable": "所有模型路由都不可用。" diff --git a/packages/api-contract/src/types.generated.ts b/packages/api-contract/src/types.generated.ts index ef3f8e2fe..ef7ea600b 100644 --- a/packages/api-contract/src/types.generated.ts +++ b/packages/api-contract/src/types.generated.ts @@ -3365,9 +3365,14 @@ export interface SyncUpstreamModelsResponse { createdUpstreamModels: number; existingUpstreamModels: number; inactivatedModels: number; + protectedUpstreamModels: number; + reactivatedModels: number; skippedUpstreamModels: number; + snapshotID: string; syncedModels: UpstreamSyncModelResponse[]; totalUpstream: number; + unchangedUpstreamModels: number; + updatedUpstreamModels: number; } export interface SyncUpstreamModelsResponseDoc { @@ -3895,6 +3900,15 @@ export interface UpstreamModelResponse { weight: number; } +export interface UpstreamModelSyncPlanResponse { + addedModels: string[]; + inactivatedModels: string[]; + protectedModels: string[]; + reactivatedModels: string[]; + unchangedModels: string[]; + updatedModels: string[]; +} + export interface UpstreamRemoteModelResponse { alreadyBound: boolean; alreadySynced: boolean; @@ -3910,6 +3924,8 @@ export interface UpstreamRemoteModelResponse { export interface UpstreamRemoteModelsResponse { items: UpstreamRemoteModelResponse[]; + snapshotID: string; + syncPlan: UpstreamModelSyncPlanResponse; total: number; } @@ -3948,8 +3964,11 @@ export interface UpstreamSyncModelResponse { bindingCode: string; created: boolean; kindsJSON: string; + protected: boolean; + reactivated: boolean; status: string; suggestedProtocol: string; + updated: boolean; upstreamModelName: string; } @@ -6100,7 +6119,7 @@ export namespace Admin { } /** - * @description 调用上游 models 接口，仅返回可导入预览，不直接落库 + * @description 调用上游 models 接口，返回可导入模型与目录变更预览，不直接落库 * @tags llm * @name LlmUpstreamsModelsRemoteList * @summary 管理员预览上游远程模型 @@ -6119,7 +6138,7 @@ export namespace Admin { } /** - * @description 调用上游 models 接口写入上游真实模型清单，不自动绑定平台模型 + * @description 调用上游 models 接口获取完整目录，原子更新远端管理模型可用状态，不删除平台模型或路由配置 * @tags llm * @name LlmUpstreamsModelsSyncCreate * @summary 管理员同步上游模型目录 @@ -6131,7 +6150,12 @@ export namespace Admin { /** 上游ID */ id: number; }; - export type RequestQuery = {}; + export type RequestQuery = { + /** 确认允许空模型目录对账 */ + allow_empty?: boolean; + /** 用户确认的远端目录快照标识 */ + expected_snapshot?: string; + }; export type RequestBody = never; export type RequestHeaders = {}; export type ResponseBody = SyncUpstreamModelsResponseDoc;