fix(openai): 闭合直通 WS 恢复与错误脱敏

backend/internal/service/openai_agent_identity.go

@@ -261,6 +261,25 @@ func ensureAgentIdentityTaskForAccount(ctx context.Context, repo AccountReposito
 	}
 	sharedTaskMu.Lock()
 	defer sharedTaskMu.Unlock()
+	// Re-read inside the shared lock. Different request paths often receive
+	// independent repository snapshots; checking only the caller's snapshot
+	// would allow sequential duplicate registrations after the first writer
+	// has already persisted a new task.
+	if repo != nil && credAccount.ID > 0 {
+		if refreshed, refreshErr := repo.GetByID(ctx, credAccount.ID); refreshErr == nil && refreshed != nil {
+			if refreshed.IsShadow() {
+				if resolved, resolveErr := resolveCredentialAccount(ctx, repo, refreshed); resolveErr == nil && resolved != nil {
+					refreshed = resolved
+				}
+			}
+			if refreshed.IsOpenAIAgentIdentity() {
+				credAccount = refreshed
+				if !account.IsShadow() {
+					account.Credentials = shallowCopyMap(credAccount.Credentials)
+				}
+			}
+		}
+	}
 	currentTaskID = strings.TrimSpace(credAccount.GetCredential("task_id"))
 	if currentTaskID != "" && (expectedTaskID == "" || currentTaskID != expectedTaskID) {
 		return nil
@@ -277,6 +296,9 @@ func ensureAgentIdentityTaskForAccount(ctx context.Context, repo AccountReposito
 	if err := persistAccountCredentials(ctx, repo, credAccount, credentials); err != nil {
 		return err
 	}
+	if !account.IsShadow() && account != credAccount {
+		account.Credentials = shallowCopyMap(credAccount.Credentials)
+	}
 	if pool != nil {
 		pool.ClearAccount(credAccount.ID)
 	}

backend/internal/service/openai_agent_identity_test.go

@@ -163,7 +163,7 @@ func TestEnsureAgentIdentityTaskSharesLockAcrossServicesForSameAccount(t *testin
 		"agent_runtime_id":  key.runtimeID,
 		"agent_private_key": privateKey,
 	}}
-	repo := &agentIdentityCredentialsRepo{}
+	repo := &agentIdentityCredentialsRepo{account: account}
 	registerCalls := 0
 	var registerMu sync.Mutex
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -179,10 +179,11 @@ func TestEnsureAgentIdentityTaskSharesLockAcrossServicesForSameAccount(t *testin
 
 	start := make(chan struct{})
 	errors := make(chan error, 2)
-	for range 2 {
+	requests := []*Account{cloneAgentIdentityTestAccount(account), cloneAgentIdentityTestAccount(account)}
+	for _, request := range requests {
 		go func() {
 			<-start
-			errors <- ensureAgentIdentityTaskForAccount(context.Background(), repo, nil, &sync.Mutex{}, account, "")
+			errors <- ensureAgentIdentityTaskForAccount(context.Background(), repo, nil, &sync.Mutex{}, request, "")
 		}()
 	}
 	close(start)
@@ -191,15 +192,26 @@ func TestEnsureAgentIdentityTaskSharesLockAcrossServicesForSameAccount(t *testin
 	registerMu.Lock()
 	defer registerMu.Unlock()
 	require.Equal(t, 1, registerCalls)
-	require.Equal(t, "task-shared", account.GetCredential("task_id"))
+	require.Equal(t, "task-shared", repo.account.GetCredential("task_id"))
+}
+
+func cloneAgentIdentityTestAccount(account *Account) *Account {
+	copy := *account
+	copy.Credentials = shallowCopyMap(account.Credentials)
+	return &copy
 }
 
 type agentIdentityCredentialsRepo struct {
 	AccountRepository
 	credentials map[string]any
+	account     *Account
 	mu          sync.Mutex
 }
 
+func (r *agentIdentityCredentialsRepo) GetByID(_ context.Context, _ int64) (*Account, error) {
+	return r.account, nil
+}
+
 func (r *agentIdentityCredentialsRepo) UpdateCredentials(_ context.Context, _ int64, credentials map[string]any) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()

backend/internal/service/openai_gateway_upstream_errors.go

@@ -471,6 +471,7 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse(
 	requestedModel ...string,
 ) (*OpenAIForwardResult, error) {
 	body := s.readUpstreamErrorBody(resp)
+	body = s.redactAgentIdentitySensitiveBody(context.Background(), account, body)
 
 	// cyber_policy：兼容路径（Chat Completions / Anthropic）以各自格式回写错误，
 	// 不原样透传 responses 格式的 cyber body（否则对下游格式不合法）。cyber 是上游网络

backend/internal/service/openai_images.go

@@ -632,6 +632,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
 	if resp.StatusCode >= 400 {
 		respBody := s.readUpstreamErrorBody(resp)
 		_ = resp.Body.Close()
+		respBody = s.redactAgentIdentitySensitiveBody(upstreamCtx, account, respBody)
 		resp.Body = io.NopCloser(bytes.NewReader(respBody))
 		upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
 		upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)

backend/internal/service/openai_ws_v2_passthrough_adapter.go

@@ -374,8 +374,13 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough(
 		if err == nil {
 			break
 		}
-		var dialErr *openAIWSDialError
-		if s.isAgentIdentityAccount(ctx, account) && errors.As(err, &dialErr) && isAgentIdentityTaskInvalidWSDialError(dialErr) && !agentTaskRecoveryTried {
+		var handshakeErr *openAIWSHandshakeError
+		responseBody := []byte(nil)
+		if errors.As(err, &handshakeErr) && handshakeErr != nil {
+			responseBody = handshakeErr.Body
+		}
+		dialErr := &openAIWSDialError{StatusCode: statusCode, ResponseBody: responseBody, Err: err}
+		if s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidWSDialError(dialErr) && !agentTaskRecoveryTried {
 			agentTaskRecoveryTried = true
 			if recoveryErr := s.recoverAgentIdentityTask(ctx, account, account.GetCredential("task_id")); recoveryErr != nil {
 				return fmt.Errorf("agent identity task recovery failed: %w", recoveryErr)