niuniu/sub2api-cn-relay-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix fresh-host acceptance and document real-host debugging learnings

Browse Source
This commit is contained in:
phamnazage-jpg
2026-05-21 21:19:19 +08:00
parent 7c6e18f94d
commit 3ba3244ea6
85 changed files with 1721 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
internal/provision/batch_detail_and_reconcile_service.go
View File

@@ -73,6 +73,7 @@ type ReconcileResult struct {
Status string
MissingCount int
ExtraCount int
StaleNoiseCount int
ProbeFailureCount int
AccessStatus string
Summary map[string]any
@@ -128,7 +129,11 @@ func (s *ReconcileService) Reconcile(ctx context.Context, req ReconcileRequest)
default:
return ReconcileResult{}, fmt.Errorf("latest import batch is %s; run import again before reconcile", batchRow.BatchStatus)
}
storedResources, err := s.store.ManagedResources().GetByBatchID(ctx, batchRow.ID)
storedResources, err := s.storedResourcesForReconcile(ctx, providerRow.ID, hostRow.ID, batchRow.ID)
if err != nil {
return ReconcileResult{}, err
}
currentBatchResources, err := s.store.ManagedResources().GetByBatchID(ctx, batchRow.ID)
if err != nil {
return ReconcileResult{}, err
}
@@ -145,6 +150,14 @@ func (s *ReconcileService) Reconcile(ctx context.Context, req ReconcileRequest)
return ReconcileResult{}, fmt.Errorf("list managed resources: %w", err)
}
missing, extra := diffManagedResources(storedResources, snapshot)
rawExtra := extra
staleNoiseAccounts := classifyHistoricalAccountNoise(currentBatchResources, snapshot.Accounts, SuggestAccountNamePrefix(req.Provider))
if len(staleNoiseAccounts) > 0 {
extra -= len(staleNoiseAccounts)
if extra < 0 {
extra = 0
}
}
probeFailures, err := s.rerunAccountProbes(ctx, batchItems, req.Provider.SmokeTestModel)
if err != nil {
return ReconcileResult{}, err
@@ -160,12 +173,15 @@ func (s *ReconcileService) Reconcile(ctx context.Context, req ReconcileRequest)
status = "degraded"
}
summary := map[string]any{
"missing_count": missing,
"extra_count": extra,
"host_version": hostVersion,
"probe_failures": probeFailures,
"access_status": accessStatus,
"access_rechecked": accessChecked,
"missing_count": missing,
"extra_count": extra,
"raw_extra_count": rawExtra,
"stale_noise_count": len(staleNoiseAccounts),
"stale_noise_accounts": staleNoiseAccounts,
"host_version": hostVersion,
"probe_failures": probeFailures,
"access_status": accessStatus,
"access_rechecked": accessChecked,
}
summaryJSON, err := json.Marshal(summary)
if err != nil {
@@ -174,7 +190,7 @@ func (s *ReconcileService) Reconcile(ctx context.Context, req ReconcileRequest)
if _, err := s.store.ReconcileRuns().Create(ctx, sqlite.ReconcileRun{BatchID: batchRow.ID, HostID: hostRow.ID, ProviderID: providerRow.ID, Status: status, SummaryJSON: string(summaryJSON)}); err != nil {
return ReconcileResult{}, err
}
return ReconcileResult{BatchID: batchRow.ID, Status: status, MissingCount: missing, ExtraCount: extra, ProbeFailureCount: probeFailures, AccessStatus: accessStatus, Summary: summary}, nil
return ReconcileResult{BatchID: batchRow.ID, Status: status, MissingCount: missing, ExtraCount: extra, StaleNoiseCount: len(staleNoiseAccounts), ProbeFailureCount: probeFailures, AccessStatus: accessStatus, Summary: summary}, nil
}
func (s *ReconcileService) rerunAccountProbes(ctx context.Context, items []sqlite.ImportBatchItem, expectedModel string) (int, error) {
@@ -190,7 +206,7 @@ func (s *ReconcileService) rerunAccountProbes(ctx context.Context, items []sqlit
if strings.TrimSpace(accountID) == "" {
return 0, fmt.Errorf("import batch item %d missing account_id in probe summary", item.ID)
}
probe, err := s.host.TestAccount(ctx, accountID)
probe, err := s.host.TestAccount(ctx, accountID, expectedModel)
if err != nil {
return 0, fmt.Errorf("re-test account %s: %w", accountID, err)
}
@@ -199,15 +215,22 @@ func (s *ReconcileService) rerunAccountProbes(ctx context.Context, items []sqlit
return 0, fmt.Errorf("reload account models %s: %w", accountID, err)
}
smokeModelSeen := hasModel(models, expectedModel)
status := firstNonEmpty(probe.Status, "unknown")
result := AccountImportResult{
Probe: probe,
Models: models,
SmokeModelSeen: smokeModelSeen,
}
status := result.ValidationStatus()
payload, err := json.Marshal(map[string]any{
"account_id": accountID,
"probe_ok": probe.OK,
"probe_status": probe.Status,
"probe_message": probe.Message,
"models": models,
"smoke_model_seen": smokeModelSeen,
"reconcile_rerun": true,
"account_id": accountID,
"probe_ok": probe.OK,
"probe_status": probe.Status,
"probe_message": probe.Message,
"models": models,
"smoke_model_seen": smokeModelSeen,
"probe_advisory": result.HasAdvisoryWarning(),
"validation_status": status,
"reconcile_rerun": true,
})
if err != nil {
return 0, fmt.Errorf("marshal probe rerun summary for %s: %w", accountID, err)
@@ -218,7 +241,7 @@ func (s *ReconcileService) rerunAccountProbes(ctx context.Context, items []sqlit
if _, err := s.store.ProbeResults().Create(ctx, sqlite.ProbeResult{BatchItemID: item.ID, ProbeType: "account_smoke_rerun", Status: status, SummaryJSON: string(payload)}); err != nil {
return 0, err
}
if !probe.OK || !smokeModelSeen {
if result.HasBlockingFailure() {
failures++
}
}
@@ -239,6 +262,21 @@ func (s *ReconcileService) rerunAccessClosure(ctx context.Context, batchID int64
return "", false, fmt.Errorf("re-check gateway access: %w", err)
}
if result.OK && result.HasExpectedModel {
completion, err := s.host.CheckGatewayCompletion(ctx, sub2api.GatewayCompletionCheckRequest{
APIKey: probeAPIKey,
Model: expectedModel,
Prompt: "ping",
MaxTokens: 8,
})
if err != nil {
return "", false, fmt.Errorf("re-check gateway completion: %w", err)
}
result.CompletionOK = completion.OK
result.CompletionStatus = completion.StatusCode
result.CompletionType = completion.ContentType
result.CompletionBody = completion.BodyPreview
}
if GatewayAccessReady(result) {
status = deriveHealthyAccessStatus(latest.ClosureType)
} else {
status = AccessStatusBroken
@@ -248,6 +286,10 @@ func (s *ReconcileService) rerunAccessClosure(ctx context.Context, batchID int64
"ok": result.OK,
"has_expected_model": result.HasExpectedModel,
"models": result.Models,
"completion_ok": result.CompletionOK,
"completion_status": result.CompletionStatus,
"completion_type": result.CompletionType,
"completion_preview": result.CompletionBody,
"reconcile_rerun": true,
})
if err != nil {
@@ -302,6 +344,42 @@ func accountIDFromProbeSummary(summaryJSON string) (string, error) {
return strings.TrimSpace(accountID), nil
}
func (s *ReconcileService) storedResourcesForReconcile(ctx context.Context, providerID, hostID, batchID int64) ([]sqlite.ManagedResource, error) {
storedResources, err := s.store.ManagedResources().GetByBatchID(ctx, batchID)
if err != nil {
return nil, err
}
sharedResources, err := s.store.ManagedResources().ListByProviderIDAndHostID(ctx, providerID, hostID)
if err != nil {
return nil, err
}
merged := make([]sqlite.ManagedResource, 0, len(storedResources)+len(sharedResources))
seen := make(map[string]struct{}, len(storedResources)+len(sharedResources))
appendUnique := func(resource sqlite.ManagedResource) {
resourceType := strings.TrimSpace(resource.ResourceType)
resourceID := strings.TrimSpace(resource.HostResourceID)
if resourceType == "" || resourceID == "" {
return
}
key := resourceType + ":" + resourceID
if _, ok := seen[key]; ok {
return
}
seen[key] = struct{}{}
merged = append(merged, resource)
}
for _, resource := range storedResources {
appendUnique(resource)
}
for _, resource := range sharedResources {
switch strings.TrimSpace(resource.ResourceType) {
case "group", "channel", "plan":
appendUnique(resource)
}
}
return merged, nil
}
func diffManagedResources(stored []sqlite.ManagedResource, snapshot sub2api.ManagedResourceSnapshot) (int, int) {
live := map[string]map[string]struct{}{
"group": make(map[string]struct{}),
@@ -348,3 +426,32 @@ func diffManagedResources(stored []sqlite.ManagedResource, snapshot sub2api.Mana
}
return missing, extra
}
func classifyHistoricalAccountNoise(currentBatchResources []sqlite.ManagedResource, snapshotAccounts []sub2api.NamedResource, accountNamePrefix string) []sub2api.NamedResource {
currentAccountIDs := make(map[string]struct{})
for _, resource := range currentBatchResources {
if strings.TrimSpace(resource.ResourceType) != "account" {
continue
}
if id := strings.TrimSpace(resource.HostResourceID); id != "" {
currentAccountIDs[id] = struct{}{}
}
}
prefix := strings.TrimSpace(accountNamePrefix)
staleNoise := make([]sub2api.NamedResource, 0)
for _, account := range snapshotAccounts {
id := strings.TrimSpace(account.ID)
if id == "" {
continue
}
if _, ok := currentAccountIDs[id]; ok {
continue
}
if prefix != "" && !strings.HasPrefix(strings.TrimSpace(account.Name), prefix) {
continue
}
staleNoise = append(staleNoise, sub2api.NamedResource{ID: id, Name: strings.TrimSpace(account.Name)})
}
return staleNoise
}

81
internal/provision/import_service.go
View File

@@ -31,6 +31,10 @@ const (
AccessStatusSelfServiceReady = "self_service_ready"
AccessStatusFullyReady = "fully_ready"
AccessStatusBroken = "broken"
AccountStatusPassed = "passed"
AccountStatusWarning = "warning"
AccountStatusFailed = "failed"
)
type AccessRequest struct {
@@ -70,11 +74,80 @@ type AccountImportResult struct {
SmokeModelSeen bool
}
func (r AccountImportResult) ValidationStatus() string {
if !r.SmokeModelSeen {
return AccountStatusFailed
}
if r.Probe.OK {
return AccountStatusPassed
}
if isAdvisoryAccountProbeFailure(r.Probe) {
return AccountStatusWarning
}
return AccountStatusFailed
}
func (r AccountImportResult) HasBlockingFailure() bool {
return r.ValidationStatus() == AccountStatusFailed
}
func (r AccountImportResult) HasAdvisoryWarning() bool {
return r.ValidationStatus() == AccountStatusWarning
}
type hostAdapter interface {
sub2api.HostAdapter
CheckGatewayAccess(ctx context.Context, req sub2api.GatewayAccessCheckRequest) (sub2api.GatewayAccessResult, error)
}
func GatewayAccessReady(result sub2api.GatewayAccessResult) bool {
return result.OK && result.HasExpectedModel && result.CompletionOK
}
func isAdvisoryAccountProbeFailure(probe sub2api.ProbeResult) bool {
if probe.OK {
return false
}
message := strings.ToLower(strings.TrimSpace(probe.Message))
if message == "" {
return false
}
if isTransientAccountProbeFailure(message) {
return true
}
if !strings.Contains(message, "responses api") {
return false
}
return strings.Contains(message, "当前测试接口仅支持") ||
strings.Contains(message, "账号本身可正常使用") ||
strings.Contains(message, "please directly") ||
strings.Contains(message, "actual api")
}
func isTransientAccountProbeFailure(message string) bool {
if !(strings.Contains(message, "429") ||
strings.Contains(message, "rate limit") ||
strings.Contains(message, "too many requests") ||
strings.Contains(message, "502") ||
strings.Contains(message, "503") ||
strings.Contains(message, "504") ||
strings.Contains(message, "bad gateway") ||
strings.Contains(message, "service unavailable") ||
strings.Contains(message, "timeout")) {
return false
}
return strings.Contains(message, "api returned") ||
strings.Contains(message, "rate_limit") ||
strings.Contains(message, "upstream") ||
strings.Contains(message, "temporar") ||
strings.Contains(message, "retry")
}
type resolvedManagedResources struct {
Group sub2api.GroupRef
Channel sub2api.ChannelRef
@@ -143,7 +216,7 @@ func (s *ImportService) Import(ctx context.Context, req ImportRequest) (report I
}
rollback.AddAccounts(accounts)
for _, account := range accounts {
probe, err := s.host.TestAccount(ctx, account.ID)
probe, err := s.host.TestAccount(ctx, account.ID, req.Provider.SmokeTestModel)
if err != nil {
return failOrDegrade(report, req.Mode, fmt.Errorf("test account %s: %w", account.ID, err))
}
@@ -157,7 +230,7 @@ func (s *ImportService) Import(ctx context.Context, req ImportRequest) (report I
failedAccounts := 0
for _, account := range report.Accounts {
if !account.Probe.OK || !account.SmokeModelSeen {
if account.HasBlockingFailure() {
failedAccounts++
}
}
@@ -188,7 +261,7 @@ func (s *ImportService) Import(ctx context.Context, req ImportRequest) (report I
report.BatchStatus = BatchStatusSucceeded
report.ProviderStatus = ProviderStatusActive
if failedAccounts > 0 || !gateway.OK || !gateway.HasExpectedModel {
if failedAccounts > 0 || !GatewayAccessReady(gateway) {
report.BatchStatus = BatchStatusPartial
report.ProviderStatus = ProviderStatusDegraded
}
@@ -198,7 +271,7 @@ func (s *ImportService) Import(ctx context.Context, req ImportRequest) (report I
case AccessModeSelfService:
report.AccessStatus = AccessStatusSelfServiceReady
}
if !gateway.OK || !gateway.HasExpectedModel {
if !GatewayAccessReady(gateway) {
report.AccessStatus = AccessStatusBroken
}
return report, nil

156
internal/provision/import_service_test.go
View File

@@ -65,6 +65,9 @@ func TestImportServiceImportSubscriptionFlow(t *testing.T) {
if host.gatewayProbe.ExpectedModel != "deepseek-chat" {
t.Fatalf("gateway probe model = %q, want %q", host.gatewayProbe.ExpectedModel, "deepseek-chat")
}
if host.testedModels["account_1"] != "deepseek-chat" || host.testedModels["account_2"] != "deepseek-chat" {
t.Fatalf("testedModels = %#v, want deepseek-chat for all created accounts", host.testedModels)
}
}
func TestImportServiceStrictModeFailsWhenAnyAccountProbeFails(t *testing.T) {
@@ -111,6 +114,136 @@ func TestImportServiceRejectsUnknownMode(t *testing.T) {
}
}
func TestImportServiceMarksBrokenWhenCompletionSmokeFails(t *testing.T) {
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1"}},
testResults: map[string]sub2api.ProbeResult{
"account_1": {OK: true, Status: "passed"},
},
models: map[string][]sub2api.AccountModel{
"account_1": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{OK: true, StatusCode: 200, HasExpectedModel: true, Models: []string{"deepseek-chat"}},
completionResult: sub2api.GatewayCompletionResult{OK: false, StatusCode: 502, ContentType: "application/json", BodyPreview: `{"error":"upstream_error"}`},
}
report, err := NewImportService(host).Import(context.Background(), ImportRequest{
Provider: sampleProviderManifest(),
Mode: ImportModePartial,
Access: AccessRequest{Mode: AccessModeSelfService, ProbeAPIKey: "user-key"},
Keys: []string{"key-1"},
})
if err != nil {
t.Fatalf("Import() error = %v", err)
}
if report.BatchStatus != BatchStatusPartial {
t.Fatalf("BatchStatus = %q, want %q", report.BatchStatus, BatchStatusPartial)
}
if report.ProviderStatus != ProviderStatusDegraded {
t.Fatalf("ProviderStatus = %q, want %q", report.ProviderStatus, ProviderStatusDegraded)
}
if report.AccessStatus != AccessStatusBroken {
t.Fatalf("AccessStatus = %q, want %q", report.AccessStatus, AccessStatusBroken)
}
}
func TestImportServiceTreatsResponsesOnlyProbeFailureAsAdvisoryWhenGatewaySucceeds(t *testing.T) {
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1", Name: "minimax-01"}},
testResults: map[string]sub2api.ProbeResult{
"account_1": {
OK: false,
Status: "failed",
Message: "账号本身可正常使用，但当前测试接口仅支持 Responses API 路径。请直接通过实际 API 调用验证。",
},
},
models: map[string][]sub2api.AccountModel{
"account_1": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{
OK: true,
StatusCode: 200,
HasExpectedModel: true,
Models: []string{"deepseek-chat"},
CompletionOK: true,
CompletionStatus: 200,
CompletionType: "text/event-stream",
},
}
report, err := NewImportService(host).Import(context.Background(), ImportRequest{
Provider: sampleProviderManifest(),
Mode: ImportModePartial,
Access: AccessRequest{Mode: AccessModeSelfService, ProbeAPIKey: "user-key"},
Keys: []string{"key-1"},
})
if err != nil {
t.Fatalf("Import() error = %v", err)
}
if report.BatchStatus != BatchStatusSucceeded {
t.Fatalf("BatchStatus = %q, want %q", report.BatchStatus, BatchStatusSucceeded)
}
if report.ProviderStatus != ProviderStatusActive {
t.Fatalf("ProviderStatus = %q, want %q", report.ProviderStatus, ProviderStatusActive)
}
if report.AccessStatus != AccessStatusSelfServiceReady {
t.Fatalf("AccessStatus = %q, want %q", report.AccessStatus, AccessStatusSelfServiceReady)
}
if len(report.Accounts) != 1 {
t.Fatalf("Accounts len = %d, want 1", len(report.Accounts))
}
if got := report.Accounts[0].ValidationStatus(); got != AccountStatusWarning {
t.Fatalf("ValidationStatus = %q, want %q", got, AccountStatusWarning)
}
}
func TestImportServiceTreatsTransientProbeFailureAsAdvisoryWhenGatewaySucceeds(t *testing.T) {
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1", Name: "deepseek-01"}},
testResults: map[string]sub2api.ProbeResult{
"account_1": {
OK: false,
Status: "failed",
Message: "API returned 429: {\"error\":{\"message\":\"Rate limited (429); user=1997/2000 model=49/50; daily_exhausted=False\",\"type\":\"rate_limit_error\",\"code\":\"rate_limit_exceeded\"}}",
},
},
models: map[string][]sub2api.AccountModel{
"account_1": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{
OK: true,
StatusCode: 200,
HasExpectedModel: true,
Models: []string{"deepseek-chat"},
CompletionOK: true,
CompletionStatus: 200,
CompletionType: "application/json",
},
}
report, err := NewImportService(host).Import(context.Background(), ImportRequest{
Provider: sampleProviderManifest(),
Mode: ImportModePartial,
Access: AccessRequest{Mode: AccessModeSelfService, ProbeAPIKey: "user-key"},
Keys: []string{"key-1"},
})
if err != nil {
t.Fatalf("Import() error = %v", err)
}
if report.BatchStatus != BatchStatusSucceeded {
t.Fatalf("BatchStatus = %q, want %q", report.BatchStatus, BatchStatusSucceeded)
}
if report.ProviderStatus != ProviderStatusActive {
t.Fatalf("ProviderStatus = %q, want %q", report.ProviderStatus, ProviderStatusActive)
}
if report.AccessStatus != AccessStatusSelfServiceReady {
t.Fatalf("AccessStatus = %q, want %q", report.AccessStatus, AccessStatusSelfServiceReady)
}
if got := report.Accounts[0].ValidationStatus(); got != AccountStatusWarning {
t.Fatalf("ValidationStatus = %q, want %q", got, AccountStatusWarning)
}
}
func TestImportServiceStrictModeRollsBackCreatedResources(t *testing.T) {
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1"}, {ID: "account_2"}},
@@ -325,7 +458,7 @@ func TestImportDeletesExistingProviderAccountsBeforeGatewayClosure(t *testing.T)
if !reflect.DeepEqual(host.deletedResources, wantDeleted) {
t.Fatalf("deleted resources = %#v, want %#v", host.deletedResources, wantDeleted)
}
if !reflect.DeepEqual(host.callSequence, []string{"deleteAccount:account_old_2", "deleteAccount:account_old_1", "gateway"}) {
if !reflect.DeepEqual(host.callSequence, []string{"deleteAccount:account_old_2", "deleteAccount:account_old_1", "gateway", "completion"}) {
t.Fatalf("call sequence = %#v, want stale-account cleanup before gateway probe", host.callSequence)
}
}
@@ -374,6 +507,7 @@ type fakeHostAdapter struct {
hostVersion string
assignedSubscriptions []sub2api.AssignSubscriptionRequest
gatewayProbe sub2api.GatewayAccessCheckRequest
completionProbe sub2api.GatewayCompletionCheckRequest
deletedResources []string
managedSnapshot sub2api.ManagedResourceSnapshot
listManagedReq sub2api.ListManagedResourcesRequest
@@ -386,6 +520,9 @@ type fakeHostAdapter struct {
updateChannelID string
updateChannelReq sub2api.CreateChannelRequest
callSequence []string
completionResult sub2api.GatewayCompletionResult
completionErr error
testedModels map[string]string
}
func (f *fakeHostAdapter) GetHostVersion(context.Context) (string, error) {
@@ -444,7 +581,11 @@ func (f *fakeHostAdapter) DeleteAccount(_ context.Context, accountID string) err
f.deletedResources = append(f.deletedResources, "account:"+accountID)
return nil
}
func (f *fakeHostAdapter) TestAccount(_ context.Context, accountID string) (sub2api.ProbeResult, error) {
func (f *fakeHostAdapter) TestAccount(_ context.Context, accountID, modelID string) (sub2api.ProbeResult, error) {
if f.testedModels == nil {
f.testedModels = map[string]string{}
}
f.testedModels[accountID] = modelID
result, ok := f.testResults[accountID]
if !ok {
return sub2api.ProbeResult{}, fmt.Errorf("missing test result for %s", accountID)
@@ -476,6 +617,17 @@ func (f *fakeHostAdapter) CheckGatewayAccess(_ context.Context, req sub2api.Gate
}
return f.gatewayResult, nil
}
func (f *fakeHostAdapter) CheckGatewayCompletion(_ context.Context, req sub2api.GatewayCompletionCheckRequest) (sub2api.GatewayCompletionResult, error) {
f.callSequence = append(f.callSequence, "completion")
f.completionProbe = req
if f.completionErr != nil {
return sub2api.GatewayCompletionResult{}, f.completionErr
}
if f.completionResult.StatusCode == 0 && !f.completionResult.OK {
return sub2api.GatewayCompletionResult{OK: true, StatusCode: 200, ContentType: "application/json"}, nil
}
return f.completionResult, nil
}
func (f *fakeHostAdapter) ListManagedResources(_ context.Context, req sub2api.ListManagedResourcesRequest) (sub2api.ManagedResourceSnapshot, error) {
f.listManagedReq = req
return sub2api.ManagedResourceSnapshot{

232
internal/provision/reconcile_service_test.go
View File

@@ -105,6 +105,63 @@ func TestReconcileServiceReturnsDegradedWhenProbeRerunFails(t *testing.T) {
}
}
func TestReconcileServiceIgnoresAdvisoryProbeFailureWhenModelsAndGatewayAreHealthy(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1", Name: "deepseek-01"}, {ID: "account_2", Name: "deepseek-02"}},
testResults: map[string]sub2api.ProbeResult{
"account_1": {
OK: false,
Status: "failed",
Message: "账号本身可正常使用，但当前测试接口仅支持 Responses API 路径。请直接通过实际 API 调用验证。",
},
"account_2": {
OK: false,
Status: "failed",
Message: "账号本身可正常使用，但当前测试接口仅支持 Responses API 路径。请直接通过实际 API 调用验证。",
},
},
models: map[string][]sub2api.AccountModel{
"account_1": {{ID: "deepseek-chat"}},
"account_2": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{
OK: true,
StatusCode: 200,
HasExpectedModel: true,
Models: []string{"deepseek-chat"},
CompletionOK: true,
CompletionStatus: 200,
},
}
seedRuntimeImportForReconcile(t, store, host)
host.managedSnapshot = sub2api.ManagedResourceSnapshot{
Groups: []sub2api.NamedResource{{ID: "group_1", Name: "DeepSeek 默认分组-self-service"}},
Channels: []sub2api.NamedResource{{ID: "channel_1", Name: "DeepSeek 默认渠道-self-service"}},
Accounts: []sub2api.NamedResource{{ID: "account_1", Name: "deepseek-01"}, {ID: "account_2", Name: "deepseek-02"}},
}
result, err := NewReconcileService(store, host).Reconcile(context.Background(), ReconcileRequest{
HostID: "host-1",
HostBaseURL: "https://sub2api.example.com",
AccessProbeAPIKey: "user-key",
Pack: pack.LoadedPack{Manifest: pack.Manifest{PackID: "openai-cn-pack", Version: "1.0.0", TargetHost: "sub2api", MinHostVersion: "0.1.126", MaxHostVersion: "0.2.x"}},
Provider: sampleProviderManifest(),
})
if err != nil {
t.Fatalf("Reconcile() error = %v", err)
}
if result.Status != "active" {
t.Fatalf("Status = %q, want active", result.Status)
}
if result.ProbeFailureCount != 0 {
t.Fatalf("ProbeFailureCount = %d, want 0", result.ProbeFailureCount)
}
}
func TestReconcileServiceReturnsDriftedWhenManagedResourceMissing(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)
@@ -232,6 +289,181 @@ func TestReconcileServicePassesAccountNamePrefixToManagedResourceSnapshot(t *tes
}
}
func TestReconcileServiceIgnoresSharedResourceReuseAcrossBatches(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1", Name: "deepseek-01"}, {ID: "account_2", Name: "deepseek-02"}},
testResults: map[string]sub2api.ProbeResult{
"account_1": {OK: true, Status: "passed"},
"account_2": {OK: true, Status: "passed"},
"account_3": {OK: true, Status: "passed"},
"account_4": {OK: true, Status: "passed"},
},
models: map[string][]sub2api.AccountModel{
"account_1": {{ID: "deepseek-chat"}},
"account_2": {{ID: "deepseek-chat"}},
"account_3": {{ID: "deepseek-chat"}},
"account_4": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{OK: true, StatusCode: 200, HasExpectedModel: true, Models: []string{"deepseek-chat"}},
}
seedRuntimeImportForReconcile(t, store, host)
ctx := context.Background()
packRow, err := store.Packs().GetByPackID(ctx, "openai-cn-pack")
if err != nil {
t.Fatalf("Packs().GetByPackID() error = %v", err)
}
providerRow, err := store.Providers().GetByPackIDAndProviderID(ctx, packRow.ID, sampleProviderManifest().ProviderID)
if err != nil {
t.Fatalf("Providers().GetByPackIDAndProviderID() error = %v", err)
}
hostRow, err := store.Hosts().GetByHostID(ctx, "host-1")
if err != nil {
t.Fatalf("Hosts().GetByHostID() error = %v", err)
}
latestBatchID, err := store.ImportBatches().Create(ctx, sqlite.ImportBatch{HostID: hostRow.ID, PackID: packRow.ID, ProviderID: providerRow.ID, Mode: ImportModePartial, BatchStatus: BatchStatusSucceeded, AccessStatus: AccessStatusSelfServiceReady})
if err != nil {
t.Fatalf("ImportBatches().Create() error = %v", err)
}
if _, err := store.ImportBatchItems().Create(ctx, sqlite.ImportBatchItem{BatchID: latestBatchID, KeyFingerprint: "key-3", AccountStatus: "passed", ProbeSummaryJSON: `{"account_id":"account_3"}`}); err != nil {
t.Fatalf("ImportBatchItems().Create() error = %v", err)
}
if _, err := store.ImportBatchItems().Create(ctx, sqlite.ImportBatchItem{BatchID: latestBatchID, KeyFingerprint: "key-4", AccountStatus: "passed", ProbeSummaryJSON: `{"account_id":"account_4"}`}); err != nil {
t.Fatalf("ImportBatchItems().Create() error = %v", err)
}
if _, err := store.ManagedResources().Create(ctx, sqlite.ManagedResource{BatchID: latestBatchID, HostID: hostRow.ID, ResourceType: "account", HostResourceID: "account_3", ResourceName: "deepseek-03"}); err != nil {
t.Fatalf("ManagedResources().Create(account_3) error = %v", err)
}
if _, err := store.ManagedResources().Create(ctx, sqlite.ManagedResource{BatchID: latestBatchID, HostID: hostRow.ID, ResourceType: "account", HostResourceID: "account_4", ResourceName: "deepseek-04"}); err != nil {
t.Fatalf("ManagedResources().Create(account_4) error = %v", err)
}
if _, err := store.AccessClosures().Create(ctx, sqlite.AccessClosureRecord{BatchID: latestBatchID, ClosureType: AccessModeSelfService, Status: AccessStatusSelfServiceReady, DetailsJSON: "{}"}); err != nil {
t.Fatalf("AccessClosures().Create() error = %v", err)
}
host.managedSnapshot = sub2api.ManagedResourceSnapshot{
Groups: []sub2api.NamedResource{{ID: "group_1", Name: "DeepSeek 默认分组-self-service"}},
Channels: []sub2api.NamedResource{{ID: "channel_1", Name: "DeepSeek 默认渠道-self-service"}},
Accounts: []sub2api.NamedResource{{ID: "account_3", Name: "deepseek-03"}, {ID: "account_4", Name: "deepseek-04"}},
}
result, err := NewReconcileService(store, host).Reconcile(ctx, ReconcileRequest{
HostID: "host-1",
HostBaseURL: "https://sub2api.example.com",
AccessProbeAPIKey: "user-key",
Pack: pack.LoadedPack{Manifest: pack.Manifest{PackID: "openai-cn-pack", Version: "1.0.0", TargetHost: "sub2api", MinHostVersion: "0.1.126", MaxHostVersion: "0.2.x"}},
Provider: sampleProviderManifest(),
})
if err != nil {
t.Fatalf("Reconcile() error = %v", err)
}
if result.BatchID != latestBatchID {
t.Fatalf("BatchID = %d, want latest batch %d", result.BatchID, latestBatchID)
}
if result.Status != "active" {
t.Fatalf("Status = %q, want active when only shared resources are reused", result.Status)
}
if result.MissingCount != 0 || result.ExtraCount != 0 {
t.Fatalf("Managed resource diff = (%d, %d), want (0, 0)", result.MissingCount, result.ExtraCount)
}
}
func TestReconcileServiceClassifiesHistoricalAccountsAsStaleNoise(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)
host := &fakeHostAdapter{
testResults: map[string]sub2api.ProbeResult{
"account_3": {OK: true, Status: "passed"},
"account_4": {OK: true, Status: "passed"},
},
models: map[string][]sub2api.AccountModel{
"account_3": {{ID: "deepseek-chat"}},
"account_4": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{OK: true, StatusCode: 200, HasExpectedModel: true, Models: []string{"deepseek-chat"}},
}
seedRuntimeImportForReconcile(t, store, host)
ctx := context.Background()
packRow, err := store.Packs().GetByPackID(ctx, "openai-cn-pack")
if err != nil {
t.Fatalf("Packs().GetByPackID() error = %v", err)
}
providerRow, err := store.Providers().GetByPackIDAndProviderID(ctx, packRow.ID, sampleProviderManifest().ProviderID)
if err != nil {
t.Fatalf("Providers().GetByPackIDAndProviderID() error = %v", err)
}
hostRow, err := store.Hosts().GetByHostID(ctx, "host-1")
if err != nil {
t.Fatalf("Hosts().GetByHostID() error = %v", err)
}
latestBatchID, err := store.ImportBatches().Create(ctx, sqlite.ImportBatch{HostID: hostRow.ID, PackID: packRow.ID, ProviderID: providerRow.ID, Mode: ImportModePartial, BatchStatus: BatchStatusSucceeded, AccessStatus: AccessStatusSelfServiceReady})
if err != nil {
t.Fatalf("ImportBatches().Create() error = %v", err)
}
for _, item := range []struct {
keyFingerprint string
accountID string
accountName string
}{
{keyFingerprint: "key-3", accountID: "account_3", accountName: "deepseek-03"},
{keyFingerprint: "key-4", accountID: "account_4", accountName: "deepseek-04"},
} {
if _, err := store.ImportBatchItems().Create(ctx, sqlite.ImportBatchItem{BatchID: latestBatchID, KeyFingerprint: item.keyFingerprint, AccountStatus: "passed", ProbeSummaryJSON: `{"account_id":"` + item.accountID + `"}`}); err != nil {
t.Fatalf("ImportBatchItems().Create(%s) error = %v", item.accountID, err)
}
if _, err := store.ManagedResources().Create(ctx, sqlite.ManagedResource{BatchID: latestBatchID, HostID: hostRow.ID, ResourceType: "account", HostResourceID: item.accountID, ResourceName: item.accountName}); err != nil {
t.Fatalf("ManagedResources().Create(%s) error = %v", item.accountID, err)
}
}
if _, err := store.AccessClosures().Create(ctx, sqlite.AccessClosureRecord{BatchID: latestBatchID, ClosureType: AccessModeSelfService, Status: AccessStatusSelfServiceReady, DetailsJSON: "{}"}); err != nil {
t.Fatalf("AccessClosures().Create() error = %v", err)
}
host.managedSnapshot = sub2api.ManagedResourceSnapshot{
Groups: []sub2api.NamedResource{{ID: "group_1", Name: "DeepSeek 默认分组-self-service"}},
Channels: []sub2api.NamedResource{{ID: "channel_1", Name: "DeepSeek 默认渠道-self-service"}},
Accounts: []sub2api.NamedResource{{ID: "account_1", Name: "deepseek-01"}, {ID: "account_2", Name: "deepseek-02"}, {ID: "account_3", Name: "deepseek-03"}, {ID: "account_4", Name: "deepseek-04"}},
}
result, err := NewReconcileService(store, host).Reconcile(ctx, ReconcileRequest{
HostID: "host-1",
HostBaseURL: "https://sub2api.example.com",
AccessProbeAPIKey: "user-key",
Pack: pack.LoadedPack{Manifest: pack.Manifest{PackID: "openai-cn-pack", Version: "1.0.0", TargetHost: "sub2api", MinHostVersion: "0.1.126", MaxHostVersion: "0.2.x"}},
Provider: sampleProviderManifest(),
})
if err != nil {
t.Fatalf("Reconcile() error = %v", err)
}
if result.BatchID != latestBatchID {
t.Fatalf("BatchID = %d, want latest batch %d", result.BatchID, latestBatchID)
}
if result.Status != "active" {
t.Fatalf("Status = %q, want active when only historical same-prefix accounts remain (missing=%d extra=%d stale=%d summary=%#v)", result.Status, result.MissingCount, result.ExtraCount, result.StaleNoiseCount, result.Summary)
}
if result.ExtraCount != 0 {
t.Fatalf("ExtraCount = %d, want 0 after stale-noise classification", result.ExtraCount)
}
if result.StaleNoiseCount != 2 {
t.Fatalf("StaleNoiseCount = %d, want 2", result.StaleNoiseCount)
}
if got, ok := result.Summary["stale_noise_count"].(int); !ok || got != 2 {
t.Fatalf("Summary[stale_noise_count] = %#v, want int(2)", result.Summary["stale_noise_count"])
}
accounts, ok := result.Summary["stale_noise_accounts"].([]sub2api.NamedResource)
if !ok {
t.Fatalf("Summary[stale_noise_accounts] type = %T, want []sub2api.NamedResource", result.Summary["stale_noise_accounts"])
}
if len(accounts) != 2 || accounts[0].ID != "account_1" || accounts[1].ID != "account_2" {
t.Fatalf("Summary[stale_noise_accounts] = %#v, want historical accounts 1 and 2", accounts)
}
}
func TestReconcileServiceRejectsRolledBackLatestBatch(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)

23
internal/provision/runtime_import_service.go
View File

@@ -169,13 +169,16 @@ func (s *RuntimeImportService) ensureProvider(ctx context.Context, packID int64,
func (s *RuntimeImportService) persistRuntimeArtifacts(ctx context.Context, batchID, hostID int64, accessMode string, report ImportReport, includeManagedResources bool) error {
for i, account := range report.Accounts {
validationStatus := account.ValidationStatus()
payload, err := json.Marshal(map[string]any{
"account_id": account.Ref.ID,
"probe_ok": account.Probe.OK,
"probe_status": account.Probe.Status,
"probe_message": account.Probe.Message,
"models": account.Models,
"smoke_model_seen": account.SmokeModelSeen,
"account_id": account.Ref.ID,
"probe_ok": account.Probe.OK,
"probe_status": account.Probe.Status,
"probe_message": account.Probe.Message,
"models": account.Models,
"smoke_model_seen": account.SmokeModelSeen,
"probe_advisory": account.HasAdvisoryWarning(),
"validation_status": validationStatus,
})
if err != nil {
return fmt.Errorf("marshal account probe summary: %w", err)
@@ -183,7 +186,7 @@ func (s *RuntimeImportService) persistRuntimeArtifacts(ctx context.Context, batc
itemID, err := s.store.ImportBatchItems().Create(ctx, sqlite.ImportBatchItem{
BatchID: batchID,
KeyFingerprint: fingerprintKey(report.AcceptedKeys, i),
AccountStatus: firstNonEmpty(account.Probe.Status, "unknown"),
AccountStatus: validationStatus,
ProbeSummaryJSON: string(payload),
})
if err != nil {
@@ -192,7 +195,7 @@ func (s *RuntimeImportService) persistRuntimeArtifacts(ctx context.Context, batc
if _, err := s.store.ProbeResults().Create(ctx, sqlite.ProbeResult{
BatchItemID: itemID,
ProbeType: "account_smoke",
Status: firstNonEmpty(account.Probe.Status, "unknown"),
Status: validationStatus,
SummaryJSON: string(payload),
}); err != nil {
return err
@@ -223,6 +226,10 @@ func (s *RuntimeImportService) persistRuntimeArtifacts(ctx context.Context, batc
"ok": report.Gateway.OK,
"has_expected_model": report.Gateway.HasExpectedModel,
"models": report.Gateway.Models,
"completion_ok": report.Gateway.CompletionOK,
"completion_status": report.Gateway.CompletionStatus,
"completion_type": report.Gateway.CompletionType,
"completion_preview": report.Gateway.CompletionBody,
})
if err != nil {
return fmt.Errorf("marshal gateway access summary: %w", err)

65
internal/provision/runtime_import_service_test.go
View File

@@ -5,6 +5,7 @@ import (
"database/sql"
"fmt"
"path/filepath"
"strings"
"testing"
_ "modernc.org/sqlite"
@@ -171,6 +172,70 @@ func TestRuntimeImportServicePersistsFailedBatchAfterStrictRollback(t *testing.T
}
}
func TestRuntimeImportServicePersistsWarningAccountStatusForAdvisoryProbeFailure(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)
seedProvisionHost(t, store, "host-1", "https://sub2api.example.com")
host := &fakeHostAdapter{
batchAccounts: []sub2api.AccountRef{{ID: "account_1"}},
testResults: map[string]sub2api.ProbeResult{
"account_1": {
OK: false,
Status: "failed",
Message: "账号本身可正常使用，但当前测试接口仅支持 Responses API 路径。请直接通过实际 API 调用验证。",
},
},
models: map[string][]sub2api.AccountModel{
"account_1": {{ID: "deepseek-chat"}},
},
gatewayResult: sub2api.GatewayAccessResult{
OK: true,
StatusCode: 200,
HasExpectedModel: true,
Models: []string{"deepseek-chat"},
CompletionOK: true,
CompletionStatus: 200,
},
}
svc := NewRuntimeImportService(store, host)
result, err := svc.Import(context.Background(), RuntimeImportRequest{
HostID: "host-1",
HostBaseURL: "https://sub2api.example.com",
Pack: pack.LoadedPack{
Manifest: pack.Manifest{PackID: "openai-cn-pack", Version: "1.0.0", TargetHost: "sub2api", MinHostVersion: "0.1.126", MaxHostVersion: "0.2.x"},
Checksum: "checksum-1",
},
Provider: sampleProviderManifest(),
Mode: ImportModePartial,
Keys: []string{"key-1"},
Access: AccessRequest{
Mode: AccessModeSelfService,
ProbeAPIKey: "user-key",
},
})
if err != nil {
t.Fatalf("RuntimeImportService.Import() error = %v", err)
}
if result.Report.BatchStatus != BatchStatusSucceeded {
t.Fatalf("BatchStatus = %q, want %q", result.Report.BatchStatus, BatchStatusSucceeded)
}
var accountStatus string
var summary string
if err := store.SQLDB().QueryRowContext(context.Background(), "SELECT account_status, probe_summary_json FROM import_batch_items WHERE batch_id = ? ORDER BY id LIMIT 1", result.BatchID).Scan(&accountStatus, &summary); err != nil {
t.Fatalf("query import batch item: %v", err)
}
if accountStatus != AccountStatusWarning {
t.Fatalf("account_status = %q, want %q", accountStatus, AccountStatusWarning)
}
if !strings.Contains(summary, "\"probe_advisory\":true") {
t.Fatalf("probe_summary_json = %s, want probe_advisory=true", summary)
}
}
func TestRuntimeImportServicePersistsPartialManagedResourcesOnAccessFailure(t *testing.T) {
store := openProvisionTestStore(t)
defer closeProvisionTestStore(t, store)