feat(openai): 支持 Agent Identity 认证

backend/internal/handler/admin/account_codex_agent_identity_import_test.go

@@ -0,0 +1,47 @@
+package admin
+
+import (
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/x509"
+	"encoding/base64"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNormalizeCodexImportEntryAcceptsAgentIdentityAuthJSON(t *testing.T) {
+	_, privateKey, err := ed25519.GenerateKey(rand.Reader)
+	require.NoError(t, err)
+	der, err := x509.MarshalPKCS8PrivateKey(privateKey)
+	require.NoError(t, err)
+	privateKeyBase64 := base64.StdEncoding.EncodeToString(der)
+
+	item, err := normalizeCodexImportEntry(codexImportEntry{
+		Index: 1,
+		Value: map[string]any{
+			"auth_mode": "agentIdentity",
+			"agent_identity": map[string]any{
+				"agent_runtime_id":           "runtime-import",
+				"agent_private_key":          privateKeyBase64,
+				"account_id":                 "account-import",
+				"chatgpt_user_id":            "user-import",
+				"email":                      "agent@example.invalid",
+				"plan_type":                  "pro",
+				"chatgpt_account_is_fedramp": false,
+			},
+		},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, item)
+	require.True(t, item.IsAgentIdentity)
+	require.Equal(t, service.OpenAIAuthModeAgentIdentity, item.Credentials["auth_mode"])
+	require.Equal(t, "runtime-import", item.Credentials["agent_runtime_id"])
+	require.Equal(t, privateKeyBase64, item.Credentials["agent_private_key"])
+	require.Equal(t, "account-import", item.Credentials["chatgpt_account_id"])
+	require.Equal(t, "user-import", item.Credentials["chatgpt_user_id"])
+	require.NotContains(t, item.Credentials, "access_token")
+	require.NotContains(t, item.Credentials, "refresh_token")
+	require.NotEmpty(t, item.WarningTexts)
+}

backend/internal/handler/admin/account_codex_import.go

@@ -72,20 +72,25 @@ type codexImportEntry struct {
 }
 
 type codexImportAccount struct {
-	Name           string
-	AccessToken    string
-	RefreshToken   string
-	IDToken        string
-	Email          string
-	AccountID      string
-	UserID         string
-	PlanType       string
-	Organization   string
-	Credentials    map[string]any
-	Extra          map[string]any
-	TokenExpiresAt *time.Time
-	IdentityKeys   []string
-	WarningTexts   []string
+	Name            string
+	AccessToken     string
+	RefreshToken    string
+	IDToken         string
+	Email           string
+	AccountID       string
+	UserID          string
+	PlanType        string
+	Organization    string
+	AgentRuntimeID  string
+	AgentPrivateKey string
+	AgentTaskID     string
+	AgentFedRAMP    bool
+	IsAgentIdentity bool
+	Credentials     map[string]any
+	Extra           map[string]any
+	TokenExpiresAt  *time.Time
+	IdentityKeys    []string
+	WarningTexts    []string
 }
 
 type codexJWTClaims struct {
@@ -492,6 +497,41 @@ func normalizeCodexImportEntry(entry codexImportEntry) (*codexImportAccount, err
 	case string:
 		item.AccessToken = strings.TrimSpace(raw)
 	case map[string]any:
+		if agentIdentity, ok := firstCodexMap(raw, []string{"agent_identity"}, []string{"agentIdentity"}); ok || strings.EqualFold(firstCodexString(raw, []string{"auth_mode"}, []string{"authMode"}), service.OpenAIAuthModeAgentIdentity) {
+			if !ok {
+				agentIdentity = raw
+			}
+			item.IsAgentIdentity = true
+			item.AgentRuntimeID = firstCodexString(agentIdentity, []string{"agent_runtime_id"}, []string{"agentRuntimeId"})
+			item.AgentPrivateKey = firstCodexString(agentIdentity, []string{"agent_private_key"}, []string{"agentPrivateKey"})
+			item.AgentTaskID = firstCodexString(agentIdentity, []string{"task_id"}, []string{"taskId"})
+			item.AccountID = firstCodexString(agentIdentity, []string{"account_id"}, []string{"accountId"})
+			item.UserID = firstCodexString(agentIdentity, []string{"chatgpt_user_id"}, []string{"chatgptUserId"})
+			item.Email = firstCodexString(agentIdentity, []string{"email"})
+			item.PlanType = firstCodexString(agentIdentity, []string{"plan_type"}, []string{"planType"})
+			item.AgentFedRAMP = firstCodexBool(agentIdentity, []string{"chatgpt_account_is_fedramp"}, []string{"chatgptAccountIsFedramp"})
+			if item.AgentRuntimeID == "" || item.AgentPrivateKey == "" || item.AccountID == "" || item.UserID == "" {
+				return nil, errors.New("Agent Identity 缺少必要字段")
+			}
+			if err := service.ValidateOpenAIAgentIdentityPrivateKey(item.AgentPrivateKey); err != nil {
+				return nil, errors.New("Agent Identity private key 格式无效")
+			}
+			item.Credentials["auth_mode"] = service.OpenAIAuthModeAgentIdentity
+			item.Credentials["agent_runtime_id"] = item.AgentRuntimeID
+			item.Credentials["agent_private_key"] = item.AgentPrivateKey
+			item.Credentials["chatgpt_account_id"] = item.AccountID
+			item.Credentials["chatgpt_user_id"] = item.UserID
+			item.Credentials["chatgpt_account_is_fedramp"] = item.AgentFedRAMP
+			setCodexCredentialIfNotEmpty(item.Credentials, "task_id", item.AgentTaskID)
+			setCodexCredentialIfNotEmpty(item.Credentials, "email", item.Email)
+			setCodexCredentialIfNotEmpty(item.Credentials, "plan_type", item.PlanType)
+			if item.AgentTaskID == "" {
+				item.WarningTexts = append(item.WarningTexts, "未包含 task_id，首次请求会使用现有 runtime 注册新 task")
+			}
+			item.IdentityKeys = buildCodexAgentIdentityKeys(item.AccountID, item.UserID, item.Email, item.AgentRuntimeID)
+			item.Name = buildCodexImportAccountName(item, entry.Index)
+			return item, nil
+		}
 		item.AccessToken = firstCodexString(raw,
 			[]string{"tokens", "access_token"},
 			[]string{"tokens", "accessToken"},
@@ -573,6 +613,9 @@ func normalizeCodexImportEntry(entry codexImportEntry) (*codexImportAccount, err
 		return nil, fmt.Errorf("第 %d 条格式不支持", entry.Index)
 	}
 
+	if item.IsAgentIdentity {
+		return item, nil
+	}
 	if item.AccessToken == "" {
 		return nil, errors.New("缺少 accessToken/access_token")
 	}
@@ -808,6 +851,9 @@ func sanitizeCodexImportCredentialExtras(input map[string]any) map[string]any {
 		"openai_auth_mode":           {},
 		"token_type":                 {},
 		"chatgpt_account_is_fedramp": {},
+		"agent_runtime_id":           {},
+		"agent_private_key":          {},
+		"task_id":                    {},
 	}
 	out := make(map[string]any, len(input))
 	for key, value := range input {
@@ -838,6 +884,14 @@ func buildCodexImportIdentityKeys(accountID, userID, email, accessToken, refresh
 	return buildCodexStoredIdentityKeys(accountID, userID, email, accessToken)
 }
 
+func buildCodexAgentIdentityKeys(accountID, userID, email, runtimeID string) []string {
+	keys := buildCodexStoredIdentityKeys(accountID, userID, email, "")
+	if runtimeID = strings.TrimSpace(runtimeID); runtimeID != "" {
+		keys = append([]string{"agent:" + runtimeID}, keys...)
+	}
+	return keys
+}
+
 // buildCodexStoredIdentityKeys 生成存量账号索引键，保留 user/account 维度，
 // 让 accessToken-only 账号后续升级为完整 OAuth 时仍能命中并更新原账号。
 func buildCodexStoredIdentityKeys(accountID, userID, email, accessToken string) []string {
@@ -887,6 +941,10 @@ func (i *codexAccountIndex) Add(account service.Account) {
 	for _, key := range keys {
 		i.accountsByKey[key] = upsertCodexAccount(i.accountsByKey[key], account)
 	}
+	if runtimeID := codexCredentialString(account.Credentials, "agent_runtime_id"); runtimeID != "" {
+		key := "agent:" + runtimeID
+		i.accountsByKey[key] = upsertCodexAccount(i.accountsByKey[key], account)
+	}
 }
 
 func (i *codexAccountIndex) remove(accountID int64) {
@@ -1039,6 +1097,38 @@ func firstCodexString(obj map[string]any, paths ...[]string) string {
 	return ""
 }
 
+func firstCodexMap(obj map[string]any, paths ...[]string) (map[string]any, bool) {
+	for _, path := range paths {
+		value, ok := codexPathValue(obj, path)
+		if !ok || value == nil {
+			continue
+		}
+		if mapped, ok := value.(map[string]any); ok {
+			return mapped, true
+		}
+	}
+	return nil, false
+}
+
+func firstCodexBool(obj map[string]any, paths ...[]string) bool {
+	for _, path := range paths {
+		value, ok := codexPathValue(obj, path)
+		if !ok {
+			continue
+		}
+		switch value := value.(type) {
+		case bool:
+			return value
+		case string:
+			parsed, err := strconv.ParseBool(strings.TrimSpace(value))
+			if err == nil {
+				return parsed
+			}
+		}
+	}
+	return false
+}
+
 func copyCodexExtraString(obj map[string]any, extra map[string]any, key string, path []string) {
 	value := firstCodexString(obj, path)
 	if value != "" {

backend/internal/handler/dto/credentials_redact_test.go

@@ -20,6 +20,7 @@ func TestRedactCredentials_StripsSensitiveKeysAndReportsStatus(t *testing.T) {
 		"aws_secret_access_key": "aws-secret",
 		"service_account_json":  map[string]any{"private_key": "..."},
 		"private_key":           "raw-key",
+		"agent_private_key":     "agent-key-secret",
 		// 非敏感
 		"base_url":      "https://api.example.com",
 		"model_mapping": map[string]any{"foo": "bar"},
@@ -35,6 +36,7 @@ func TestRedactCredentials_StripsSensitiveKeysAndReportsStatus(t *testing.T) {
 	require.NotContains(t, out, "aws_secret_access_key")
 	require.NotContains(t, out, "service_account_json")
 	require.NotContains(t, out, "private_key")
+	require.NotContains(t, out, "agent_private_key")
 
 	require.Equal(t, "https://api.example.com", out["base_url"])
 	require.Equal(t, map[string]any{"foo": "bar"}, out["model_mapping"])
@@ -47,6 +49,7 @@ func TestRedactCredentials_StripsSensitiveKeysAndReportsStatus(t *testing.T) {
 	require.True(t, status["has_aws_secret_access_key"])
 	require.True(t, status["has_service_account_json"])
 	require.True(t, status["has_private_key"])
+	require.True(t, status["has_agent_private_key"])
 
 	// 状态 map 不应携带非敏感键的 has_*
 	require.NotContains(t, status, "has_base_url")
@@ -84,6 +87,7 @@ func TestRedactCredentials_AllKnownSensitiveKeys(t *testing.T) {
 		"api_key", "session_key", "cookie",
 		"aws_secret_access_key", "aws_session_token",
 		"service_account_json", "service_account", "private_key",
+		"agent_private_key",
 	}
 	in := make(map[string]any, len(keys))
 	for _, k := range keys {

backend/internal/service/account_credentials_redact.go

@@ -4,7 +4,7 @@ package service
 // dto 层做响应脱敏、service 层做更新合并都引用此清单——新增凭证类型时务必同步。
 var SensitiveCredentialKeys = []string{
 	// OAuth
-	"access_token", "refresh_token", "id_token",
+	"access_token", "refresh_token", "id_token", "agent_private_key",
 	// API Key 类
 	"api_key", "session_key", "cookie",
 	// 云服务凭据

backend/internal/service/account_test_service.go

@@ -15,6 +15,7 @@ import (
 	"net/http/httptest"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/Wei-Shaw/sub2api/internal/config"
@@ -72,6 +73,7 @@ type AccountTestService struct {
 	httpUpstream              HTTPUpstream
 	cfg                       *config.Config
 	tlsFPProfileService       *TLSFingerprintProfileService
+	agentIdentityTaskMu       sync.Mutex
 }
 
 // NewAccountTestService creates a new AccountTestService
@@ -544,9 +546,11 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
 
 	if credentialAccount.IsOAuth() {
 		isOAuth = true
-		// OAuth - use Bearer token with ChatGPT internal API
-		authToken = credentialAccount.GetOpenAIAccessToken()
-		if authToken == "" {
+		// Agent Identity signs each request and does not retain the OAuth token.
+		if !credentialAccount.IsOpenAIAgentIdentity() {
+			authToken = credentialAccount.GetOpenAIAccessToken()
+		}
+		if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() {
 			return s.sendErrorAndEnd(c, "No access token available")
 		}
 
@@ -597,7 +601,19 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
 
 	// Set common headers
 	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Authorization", "Bearer "+authToken)
+	if credentialAccount.IsOpenAIAgentIdentity() {
+		authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, credentialAccount)
+		if authErr != nil {
+			return s.sendErrorAndEnd(c, "Failed to build Agent Identity authentication")
+		}
+		for key, values := range authHeaders {
+			for _, value := range values {
+				req.Header.Add(key, value)
+			}
+		}
+	} else {
+		req.Header.Set("Authorization", "Bearer "+authToken)
+	}
 
 	// Set OAuth-specific headers for ChatGPT internal API
 	if isOAuth {
@@ -804,16 +820,26 @@ func (s *AccountTestService) testOpenAIChatCompletionsConnection(
 // resulting capability state on the account.
 func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account *Account, testModelID string) error {
 	ctx := c.Request.Context()
+	credentialAccount := account
+	if account.IsShadow() {
+		resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account)
+		if err != nil {
+			return s.sendErrorAndEnd(c, "Failed to resolve account credentials")
+		}
+		credentialAccount = resolved
+	}
 
 	authToken := ""
 	apiURL := ""
 	isOAuth := false
 
 	switch {
-	case account.IsOAuth():
+	case credentialAccount.IsOAuth():
 		isOAuth = true
-		authToken = account.GetOpenAIAccessToken()
-		if authToken == "" {
+		if !credentialAccount.IsOpenAIAgentIdentity() {
+			authToken = credentialAccount.GetOpenAIAccessToken()
+		}
+		if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() {
 			return s.sendErrorAndEnd(c, "No access token available")
 		}
 		apiURL = chatgptCodexAPIURL + "/compact"
@@ -852,7 +878,19 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account
 
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json")
-	req.Header.Set("Authorization", "Bearer "+authToken)
+	if credentialAccount.IsOpenAIAgentIdentity() {
+		authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, credentialAccount)
+		if authErr != nil {
+			return s.sendErrorAndEnd(c, "Failed to build Agent Identity authentication")
+		}
+		for key, values := range authHeaders {
+			for _, value := range values {
+				req.Header.Add(key, value)
+			}
+		}
+	} else {
+		req.Header.Set("Authorization", "Bearer "+authToken)
+	}
 	req.Header.Set("OpenAI-Beta", "responses=experimental")
 	req.Header.Set("Originator", "codex_cli_rs")
 	req.Header.Set("User-Agent", codexCLIUserAgent)
@@ -863,7 +901,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account
 
 	if isOAuth {
 		req.Host = "chatgpt.com"
-		setOpenAIChatGPTAccountHeaders(req.Header, account)
+		setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount)
 	}
 
 	// 账号级请求头覆写：测试请求与真实转发保持一致的最终头
@@ -1677,8 +1715,19 @@ func (s *AccountTestService) testOpenAIImageAPIKey(c *gin.Context, ctx context.C
 
 // testOpenAIImageOAuth tests OpenAI image generation using an OAuth account via Codex /responses API.
 func (s *AccountTestService) testOpenAIImageOAuth(c *gin.Context, ctx context.Context, account *Account, modelID, prompt string) error {
-	authToken := account.GetOpenAIAccessToken()
-	if authToken == "" {
+	credentialAccount := account
+	if account.IsShadow() {
+		resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account)
+		if err != nil {
+			return s.sendErrorAndEnd(c, "Failed to resolve account credentials")
+		}
+		credentialAccount = resolved
+	}
+	authToken := ""
+	if !credentialAccount.IsOpenAIAgentIdentity() {
+		authToken = credentialAccount.GetOpenAIAccessToken()
+	}
+	if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() {
 		return s.sendErrorAndEnd(c, "No access token available")
 	}
 
@@ -1710,17 +1759,29 @@ func (s *AccountTestService) testOpenAIImageOAuth(c *gin.Context, ctx context.Co
 	}
 	req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
 	req.Host = "chatgpt.com"
-	req.Header.Set("Authorization", "Bearer "+authToken)
+	if credentialAccount.IsOpenAIAgentIdentity() {
+		authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, credentialAccount)
+		if authErr != nil {
+			return s.sendErrorAndEnd(c, "Failed to build Agent Identity authentication")
+		}
+		for key, values := range authHeaders {
+			for _, value := range values {
+				req.Header.Add(key, value)
+			}
+		}
+	} else {
+		req.Header.Set("Authorization", "Bearer "+authToken)
+	}
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "text/event-stream")
 	req.Header.Set("OpenAI-Beta", "responses=experimental")
 	req.Header.Set("originator", "codex_cli_rs")
-	if customUA := strings.TrimSpace(account.GetOpenAIUserAgent()); customUA != "" {
+	if customUA := strings.TrimSpace(credentialAccount.GetOpenAIUserAgent()); customUA != "" {
 		req.Header.Set("User-Agent", customUA)
 	} else {
 		req.Header.Set("User-Agent", codexCLIUserAgent)
 	}
-	setOpenAIChatGPTAccountHeaders(req.Header, account)
+	setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount)
 	// 与真实转发一致：originator 与最终 User-Agent 首段配套（原 opencode 与 Codex UA 错配会 404，issue #3901）。
 	enforceCodexIdentityHeaders(req.Header)
 

backend/internal/service/account_usage_service.go

@@ -291,6 +291,7 @@ type AccountUsageService struct {
 	cache                   *UsageCache
 	identityCache           IdentityCache
 	tlsFPProfileService     *TLSFingerprintProfileService
+	agentIdentityTaskMu     sync.Mutex
 }
 
 // NewAccountUsageService 创建AccountUsageService实例
@@ -682,8 +683,11 @@ func (s *AccountUsageService) probeOpenAICodexSnapshot(ctx context.Context, acco
 	if account == nil || !account.IsOAuth() {
 		return nil, nil
 	}
-	accessToken := account.GetOpenAIAccessToken()
-	if accessToken == "" {
+	accessToken := ""
+	if !account.IsOpenAIAgentIdentity() {
+		accessToken = account.GetOpenAIAccessToken()
+	}
+	if accessToken == "" && !account.IsOpenAIAgentIdentity() {
 		return nil, fmt.Errorf("no access token available")
 	}
 	modelID := openaipkg.DefaultTestModel
@@ -701,7 +705,19 @@ func (s *AccountUsageService) probeOpenAICodexSnapshot(ctx context.Context, acco
 	}
 	req.Host = "chatgpt.com"
 	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Authorization", "Bearer "+accessToken)
+	if account.IsOpenAIAgentIdentity() {
+		authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, account)
+		if authErr != nil {
+			return nil, fmt.Errorf("build Agent Identity authentication: %w", authErr)
+		}
+		for key, values := range authHeaders {
+			for _, value := range values {
+				req.Header.Add(key, value)
+			}
+		}
+	} else {
+		req.Header.Set("Authorization", "Bearer "+accessToken)
+	}
 	req.Header.Set("Accept", "text/event-stream")
 	req.Header.Set("OpenAI-Beta", "responses=experimental")
 	req.Header.Set("Originator", "codex_cli_rs")

backend/internal/service/openai_agent_identity.go

@@ -0,0 +1,431 @@
+package service
+
+import (
+	"context"
+	"crypto"
+	"crypto/ed25519"
+	"crypto/sha512"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/httpclient"
+	"golang.org/x/crypto/curve25519"
+	"golang.org/x/crypto/nacl/box"
+)
+
+const (
+	OpenAIAuthModeAgentIdentity          = "agentIdentity"
+	agentIdentityAuthAPIBaseURL          = "https://auth.openai.com/api/accounts"
+	agentIdentityTaskRegistrationTimeout = 30 * time.Second
+)
+
+var openAIAgentIdentityAuthAPIBaseURL = agentIdentityAuthAPIBaseURL
+
+type agentIdentityKey struct {
+	runtimeID  string
+	privateKey ed25519.PrivateKey
+	taskID     string
+}
+
+type agentIdentityTaskRegistrationResponse struct {
+	TaskID               string `json:"task_id"`
+	TaskIDCamel          string `json:"taskId"`
+	EncryptedTaskID      string `json:"encrypted_task_id"`
+	EncryptedTaskIDCamel string `json:"encryptedTaskId"`
+}
+
+type agentIdentityTaskRecoveredError struct{}
+
+func (e *agentIdentityTaskRecoveredError) Error() string {
+	return "agent identity task recovered"
+}
+
+func (a *Account) IsOpenAIAgentIdentity() bool {
+	if a == nil || !a.IsOpenAIOAuth() {
+		return false
+	}
+	return strings.EqualFold(strings.TrimSpace(a.GetCredential(openAIAuthModeCredentialKey)), OpenAIAuthModeAgentIdentity)
+}
+
+func agentIdentityPrivateKey(account *Account) (ed25519.PrivateKey, error) {
+	if account == nil {
+		return nil, errors.New("agent identity account is nil")
+	}
+	raw := strings.TrimSpace(account.GetCredential("agent_private_key"))
+	if raw == "" {
+		return nil, errors.New("agent identity private key is missing")
+	}
+	der, err := base64.StdEncoding.DecodeString(raw)
+	if err != nil {
+		return nil, errors.New("agent identity private key is not valid base64")
+	}
+	key, err := x509.ParsePKCS8PrivateKey(der)
+	if err != nil {
+		return nil, errors.New("agent identity private key is not valid PKCS#8")
+	}
+	privateKey, ok := key.(ed25519.PrivateKey)
+	if !ok || len(privateKey) != ed25519.PrivateKeySize {
+		return nil, errors.New("agent identity private key is not Ed25519")
+	}
+	return privateKey, nil
+}
+
+// ValidateOpenAIAgentIdentityPrivateKey validates the stored PKCS#8 Ed25519
+// form without returning or logging the key material.
+func ValidateOpenAIAgentIdentityPrivateKey(encoded string) error {
+	account := &Account{Credentials: map[string]any{"agent_private_key": encoded}}
+	_, err := agentIdentityPrivateKey(account)
+	return err
+}
+
+func agentIdentityKeyFromAccount(account *Account) (agentIdentityKey, error) {
+	privateKey, err := agentIdentityPrivateKey(account)
+	if err != nil {
+		return agentIdentityKey{}, err
+	}
+	runtimeID := strings.TrimSpace(account.GetCredential("agent_runtime_id"))
+	if runtimeID == "" {
+		return agentIdentityKey{}, errors.New("agent identity runtime id is missing")
+	}
+	return agentIdentityKey{
+		runtimeID:  runtimeID,
+		privateKey: privateKey,
+		taskID:     strings.TrimSpace(account.GetCredential("task_id")),
+	}, nil
+}
+
+func buildAgentAssertion(key agentIdentityKey, now time.Time) (string, error) {
+	if key.runtimeID == "" || key.taskID == "" {
+		return "", errors.New("agent identity runtime or task id is missing")
+	}
+	timestamp := now.UTC().Format(time.RFC3339)
+	payload := []byte(key.runtimeID + ":" + key.taskID + ":" + timestamp)
+	signature, err := key.privateKey.Sign(nil, payload, crypto.Hash(0))
+	if err != nil {
+		return "", errors.New("failed to sign agent assertion")
+	}
+	envelope := map[string]string{
+		"agent_runtime_id": key.runtimeID,
+		"task_id":          key.taskID,
+		"timestamp":        timestamp,
+		"signature":        base64.StdEncoding.EncodeToString(signature),
+	}
+	encoded, err := json.Marshal(envelope)
+	if err != nil {
+		return "", errors.New("failed to serialize agent assertion")
+	}
+	return "AgentAssertion " + base64.RawURLEncoding.EncodeToString(encoded), nil
+}
+
+func signAgentTaskRegistration(key agentIdentityKey, timestamp time.Time) (string, string, error) {
+	if key.runtimeID == "" {
+		return "", "", errors.New("agent identity runtime id is missing")
+	}
+	formatted := timestamp.UTC().Format(time.RFC3339)
+	signature, err := key.privateKey.Sign(nil, []byte(key.runtimeID+":"+formatted), crypto.Hash(0))
+	if err != nil {
+		return "", "", errors.New("failed to sign agent task registration")
+	}
+	return formatted, base64.StdEncoding.EncodeToString(signature), nil
+}
+
+func decryptAgentTaskID(key agentIdentityKey, encoded string) (string, error) {
+	ciphertext, err := base64.StdEncoding.DecodeString(strings.TrimSpace(encoded))
+	if err != nil {
+		return "", errors.New("encrypted agent task id is not valid base64")
+	}
+	seed := key.privateKey.Seed()
+	digest := sha512.Sum512(seed)
+	var curvePrivate [32]byte
+	copy(curvePrivate[:], digest[:32])
+	curvePrivate[0] &= 248
+	curvePrivate[31] &= 127
+	curvePrivate[31] |= 64
+	curvePublicBytes, err := curve25519.X25519(curvePrivate[:], curve25519.Basepoint)
+	if err != nil {
+		return "", errors.New("failed to derive agent identity decryption key")
+	}
+	var curvePublic [32]byte
+	copy(curvePublic[:], curvePublicBytes)
+	plaintext, ok := box.OpenAnonymous(nil, ciphertext, &curvePublic, &curvePrivate)
+	if !ok {
+		return "", errors.New("failed to decrypt encrypted agent task id")
+	}
+	taskID := strings.TrimSpace(string(plaintext))
+	if taskID == "" {
+		return "", errors.New("decrypted agent task id is empty")
+	}
+	return taskID, nil
+}
+
+func registerAgentIdentityTask(ctx context.Context, account *Account) (string, error) {
+	key, err := agentIdentityKeyFromAccount(account)
+	if err != nil {
+		return "", err
+	}
+	timestamp, signature, err := signAgentTaskRegistration(key, time.Now())
+	if err != nil {
+		return "", err
+	}
+	proxyURL := ""
+	if account.ProxyID != nil && account.Proxy != nil {
+		proxyURL = account.Proxy.URL()
+	}
+	client, err := httpclient.GetClient(httpclient.Options{
+		ProxyURL:              proxyURL,
+		Timeout:               agentIdentityTaskRegistrationTimeout,
+		ResponseHeaderTimeout: 15 * time.Second,
+	})
+	if err != nil {
+		return "", errors.New("invalid proxy configuration for agent task registration")
+	}
+	body, err := json.Marshal(map[string]string{
+		"timestamp": timestamp,
+		"signature": signature,
+	})
+	if err != nil {
+		return "", errors.New("failed to serialize agent task registration")
+	}
+	url := strings.TrimRight(strings.TrimSpace(openAIAgentIdentityAuthAPIBaseURL), "/") + "/v1/agent/" + key.runtimeID + "/task/register"
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, strings.NewReader(string(body)))
+	if err != nil {
+		return "", errors.New("failed to build agent task registration request")
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", errors.New("agent task registration request failed")
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
+		return "", fmt.Errorf("agent task registration returned status %d", resp.StatusCode)
+	}
+	var result agentIdentityTaskRegistrationResponse
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 64*1024)).Decode(&result); err != nil {
+		return "", errors.New("agent task registration response is invalid")
+	}
+	if taskID := strings.TrimSpace(result.TaskID); taskID != "" {
+		return taskID, nil
+	}
+	if taskID := strings.TrimSpace(result.TaskIDCamel); taskID != "" {
+		return taskID, nil
+	}
+	encrypted := strings.TrimSpace(result.EncryptedTaskID)
+	if encrypted == "" {
+		encrypted = strings.TrimSpace(result.EncryptedTaskIDCamel)
+	}
+	if encrypted == "" {
+		return "", errors.New("agent task registration response omitted task id")
+	}
+	return decryptAgentTaskID(key, encrypted)
+}
+
+func ensureAgentIdentityTaskForAccount(ctx context.Context, repo AccountRepository, pool *openAIWSConnPool, taskMu *sync.Mutex, account *Account, expectedTaskID string) error {
+	if account == nil || !account.IsOpenAIAgentIdentity() {
+		return nil
+	}
+	credAccount := account
+	if account.IsShadow() {
+		resolved, err := resolveCredentialAccount(ctx, repo, account)
+		if err != nil {
+			return err
+		}
+		credAccount = resolved
+	}
+	if credAccount == nil || !credAccount.IsOpenAIAgentIdentity() {
+		return errors.New("agent identity credentials are unavailable")
+	}
+	currentTaskID := strings.TrimSpace(credAccount.GetCredential("task_id"))
+	if currentTaskID != "" && (expectedTaskID == "" || currentTaskID != expectedTaskID) {
+		return nil
+	}
+	if taskMu == nil {
+		return errors.New("agent identity task lock is unavailable")
+	}
+	taskMu.Lock()
+	defer taskMu.Unlock()
+	currentTaskID = strings.TrimSpace(credAccount.GetCredential("task_id"))
+	if currentTaskID != "" && (expectedTaskID == "" || currentTaskID != expectedTaskID) {
+		return nil
+	}
+	newTaskID, err := registerAgentIdentityTask(ctx, credAccount)
+	if err != nil {
+		return err
+	}
+	credentials := make(map[string]any, len(credAccount.Credentials)+1)
+	for key, value := range credAccount.Credentials {
+		credentials[key] = value
+	}
+	credentials["task_id"] = newTaskID
+	if err := persistAccountCredentials(ctx, repo, credAccount, credentials); err != nil {
+		return err
+	}
+	if pool != nil {
+		pool.ClearAccount(credAccount.ID)
+	}
+	return nil
+}
+
+func (s *OpenAIGatewayService) ensureAgentIdentityTask(ctx context.Context, account *Account, expectedTaskID string) error {
+	if s == nil {
+		return errors.New("openai gateway service is nil")
+	}
+	return ensureAgentIdentityTaskForAccount(ctx, s.accountRepo, s.openaiWSPool, &s.agentIdentityTaskMu, account, expectedTaskID)
+}
+
+func isAgentIdentityTaskInvalidHTTPResponse(statusCode int, body []byte) bool {
+	if statusCode != http.StatusUnauthorized {
+		return false
+	}
+	lower := strings.ToLower(string(body))
+	for _, marker := range []string{
+		"invalid task",
+		"task_id",
+		"task id",
+		"task_not_found",
+		"task_expired",
+		"unknown task",
+	} {
+		if strings.Contains(lower, marker) {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *OpenAIGatewayService) buildOpenAIAuthenticationHeaders(ctx context.Context, account *Account, token string) (http.Header, error) {
+	if account == nil {
+		return nil, errors.New("account is nil")
+	}
+	credAccount := account
+	if account.IsShadow() {
+		resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account)
+		if err != nil {
+			return nil, err
+		}
+		credAccount = resolved
+	}
+	headers := make(http.Header)
+	if credAccount != nil && credAccount.IsOpenAIAgentIdentity() {
+		agentHeaders, err := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, s.openaiWSPool, &s.agentIdentityTaskMu, credAccount)
+		if err != nil {
+			return nil, err
+		}
+		return agentHeaders, nil
+	}
+	headers.Set("Authorization", "Bearer "+token)
+	return headers, nil
+}
+
+func buildAgentIdentityAuthenticationHeaders(ctx context.Context, repo AccountRepository, pool *openAIWSConnPool, taskMu *sync.Mutex, account *Account) (http.Header, error) {
+	if account == nil || !account.IsOpenAIAgentIdentity() {
+		return nil, errors.New("agent identity account is required")
+	}
+	if err := ensureAgentIdentityTaskForAccount(ctx, repo, pool, taskMu, account, ""); err != nil {
+		return nil, err
+	}
+	key, err := agentIdentityKeyFromAccount(account)
+	if err != nil {
+		return nil, err
+	}
+	assertion, err := buildAgentAssertion(key, time.Now())
+	if err != nil {
+		return nil, err
+	}
+	headers := make(http.Header)
+	headers.Set("Authorization", assertion)
+	return headers, nil
+}
+
+func (s *OpenAIGatewayService) refreshOpenAIAgentIdentityHeaders(ctx context.Context, account *Account, headers http.Header) (http.Header, error) {
+	if account == nil {
+		return cloneHeader(headers), nil
+	}
+	credAccount := account
+	if account.IsShadow() {
+		resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account)
+		if err != nil {
+			return nil, err
+		}
+		credAccount = resolved
+	}
+	if !credAccount.IsOpenAIAgentIdentity() {
+		return cloneHeader(headers), nil
+	}
+	refreshed := cloneHeader(headers)
+	if refreshed == nil {
+		refreshed = make(http.Header)
+	}
+	authHeaders, err := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, s.openaiWSPool, &s.agentIdentityTaskMu, credAccount)
+	if err != nil {
+		return nil, err
+	}
+	refreshed.Set("Authorization", authHeaders.Get("Authorization"))
+	return refreshed, nil
+}
+
+func (s *OpenAIGatewayService) recoverAgentIdentityTask(ctx context.Context, account *Account, expectedTaskID string) error {
+	if account != nil && account.IsShadow() {
+		if resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account); err == nil && resolved != nil && strings.TrimSpace(expectedTaskID) == "" {
+			expectedTaskID = strings.TrimSpace(resolved.GetCredential("task_id"))
+		}
+	}
+	return s.ensureAgentIdentityTask(ctx, account, expectedTaskID)
+}
+
+func (s *OpenAIGatewayService) isAgentIdentityAccount(ctx context.Context, account *Account) bool {
+	if account == nil {
+		return false
+	}
+	credAccount := account
+	if account.IsShadow() {
+		resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account)
+		if err != nil {
+			return false
+		}
+		credAccount = resolved
+	}
+	return credAccount != nil && credAccount.IsOpenAIAgentIdentity()
+}
+
+// redactAgentIdentitySensitiveBody removes credential values before an
+// upstream error can reach logs, ops events, or returned error text. Agent
+// Identity responses should not echo these values, but keeping this boundary
+// defensive prevents accidental disclosure if an upstream error does.
+func (s *OpenAIGatewayService) redactAgentIdentitySensitiveBody(ctx context.Context, account *Account, body []byte) []byte {
+	if !s.isAgentIdentityAccount(ctx, account) || len(body) == 0 {
+		return body
+	}
+	credAccount := account
+	if account != nil && account.IsShadow() {
+		if resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account); err == nil && resolved != nil {
+			credAccount = resolved
+		}
+	}
+	redacted := string(body)
+	for _, key := range []string{
+		"agent_private_key",
+		"agent_runtime_id",
+		"task_id",
+		"access_token",
+		"refresh_token",
+		"id_token",
+		"api_key",
+		"session_key",
+		"cookie",
+	} {
+		if value := strings.TrimSpace(credAccount.GetCredential(key)); value != "" {
+			redacted = strings.ReplaceAll(redacted, value, "[redacted]")
+		}
+	}
+	return []byte(redacted)
+}

backend/internal/service/openai_agent_identity_compat_test.go

@@ -0,0 +1,247 @@
+package service
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAccountTestServiceOpenAICompactAgentIdentityUsesFreshAssertion(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	key, privateKey := newTestAgentIdentityKey(t)
+	account := Account{
+		ID:          21,
+		Name:        "agent-identity",
+		Platform:    PlatformOpenAI,
+		Type:        AccountTypeOAuth,
+		Status:      StatusActive,
+		Schedulable: true,
+		Concurrency: 1,
+		Credentials: map[string]any{
+			"auth_mode":                  OpenAIAuthModeAgentIdentity,
+			"agent_runtime_id":           key.runtimeID,
+			"agent_private_key":          privateKey,
+			"task_id":                    key.taskID,
+			"chatgpt_account_id":         "account-agent-test",
+			"chatgpt_account_is_fedramp": true,
+		},
+	}
+	repo := &snapshotUpdateAccountRepo{stubOpenAIAccountRepo: stubOpenAIAccountRepo{accounts: []Account{account}}}
+	upstream := &httpUpstreamRecorder{resp: &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"application/json"}},
+		Body:       io.NopCloser(strings.NewReader(`{"id":"compact-agent","status":"completed"}`)),
+	}}
+	svc := &AccountTestService{accountRepo: repo, httpUpstream: upstream}
+
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/21/test", bytes.NewReader(nil))
+
+	require.NoError(t, svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact))
+	require.Equal(t, "AgentAssertion", strings.SplitN(upstream.lastReq.Header.Get("Authorization"), " ", 2)[0])
+	require.Equal(t, "account-agent-test", upstream.lastReq.Header.Get("chatgpt-account-id"))
+	require.Equal(t, "true", upstream.lastReq.Header.Get("x-openai-fedramp"))
+	require.NotContains(t, upstream.lastReq.Header.Get("Authorization"), privateKey)
+}
+
+func TestOpenAIAgentIdentityPassthroughKeepsSessionAndPromptCacheHeaders(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	key, privateKey := newTestAgentIdentityKey(t)
+	account := &Account{
+		ID:       24,
+		Platform: PlatformOpenAI,
+		Type:     AccountTypeOAuth,
+		Credentials: map[string]any{
+			"auth_mode":          OpenAIAuthModeAgentIdentity,
+			"agent_runtime_id":   key.runtimeID,
+			"agent_private_key":  privateKey,
+			"task_id":            key.taskID,
+			"chatgpt_account_id": "account-agent-passthrough",
+		},
+	}
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	body := []byte(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":true,"prompt_cache_key":"cache-agent"}`)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body))
+	c.Request.Header.Set("session_id", "client-session")
+	c.Request.Header.Set("conversation_id", "client-conversation")
+	c.Request.Header.Set("Authorization", "Bearer inbound-must-not-forward")
+
+	svc := &OpenAIGatewayService{}
+	req, err := svc.buildUpstreamRequestOpenAIPassthrough(context.Background(), c, account, body, "")
+	require.NoError(t, err)
+	require.Equal(t, "AgentAssertion", strings.SplitN(req.Header.Get("Authorization"), " ", 2)[0])
+	require.Equal(t, "account-agent-passthrough", req.Header.Get("chatgpt-account-id"))
+	require.NotEqual(t, "client-session", req.Header.Get("session_id"))
+	require.NotEqual(t, "client-conversation", req.Header.Get("conversation_id"))
+	require.Equal(t, isolateOpenAISessionID(0, "cache-agent"), req.Header.Get("session_id"))
+}
+
+func TestOpenAIAgentIdentityErrorRedactionDoesNotLeakCredentialValues(t *testing.T) {
+	key, privateKey := newTestAgentIdentityKey(t)
+	account := &Account{
+		ID:       25,
+		Platform: PlatformOpenAI,
+		Type:     AccountTypeOAuth,
+		Credentials: map[string]any{
+			"auth_mode":         OpenAIAuthModeAgentIdentity,
+			"agent_runtime_id":  key.runtimeID,
+			"agent_private_key": privateKey,
+			"task_id":           key.taskID,
+			"access_token":      key.runtimeID + "-oauth-value",
+		},
+	}
+	svc := &OpenAIGatewayService{}
+	oauthValue := account.GetCredential("access_token")
+	redacted := svc.redactAgentIdentitySensitiveBody(context.Background(), account, []byte(`{"message":"runtime-test task-test `+oauthValue+`"}`))
+	require.NotContains(t, string(redacted), key.runtimeID)
+	require.NotContains(t, string(redacted), key.taskID)
+	require.NotContains(t, string(redacted), oauthValue)
+	require.Contains(t, string(redacted), "[redacted]")
+}
+
+func TestOpenAIWSConnPoolHeadersFactoryRunsAtDialAndStalePrewarmIsDiscarded(t *testing.T) {
+	cfg := &config.Config{}
+	cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1
+	pool := newOpenAIWSConnPool(cfg)
+	defer pool.Close()
+	pool.setClientDialerForTest(&openAIWSFakeDialer{})
+
+	accountID := int64(22)
+	ap := pool.getOrCreateAccountPool(accountID)
+	factoryCalls := 0
+	latestHeader := ""
+	req := openAIWSAcquireRequest{
+		Account: &Account{ID: accountID, Platform: PlatformOpenAI, Type: AccountTypeOAuth},
+		WSURL:   "wss://example.com/v1/responses",
+		HeadersFactory: func(_ context.Context, headers http.Header) (http.Header, error) {
+			factoryCalls++
+			latestHeader = "AgentAssertion dial-" + string(rune('0'+factoryCalls))
+			if headers == nil {
+				headers = make(http.Header)
+			}
+			headers.Set("Authorization", latestHeader)
+			return headers, nil
+		},
+	}
+	ap.mu.Lock()
+	ap.lastAcquire = &req
+	generation := ap.generation
+	ap.mu.Unlock()
+
+	pool.prewarmConns(accountID, req, 1, generation)
+	require.Equal(t, 1, factoryCalls, "prewarm must generate authorization inside the actual dial")
+	require.Equal(t, "AgentAssertion dial-1", latestHeader)
+
+	pool.ClearAccount(accountID)
+	ap.mu.Lock()
+	require.Empty(t, ap.conns, "credential recovery must remove pooled connections")
+	require.Nil(t, ap.lastAcquire, "credential recovery must discard delayed acquire state")
+	require.Equal(t, generation+1, ap.generation)
+	ap.mu.Unlock()
+
+	// A prewarm captured before ClearAccount must not be admitted after recovery.
+	pool.prewarmConns(accountID, req, 1, generation)
+	ap.mu.Lock()
+	require.Empty(t, ap.conns)
+	ap.mu.Unlock()
+}
+
+func TestOpenAIAgentIdentityTaskInvalidRetriesExactlyOnce(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	key, privateKey := newTestAgentIdentityKey(t)
+	account := &Account{
+		ID:          23,
+		Name:        "agent-identity",
+		Platform:    PlatformOpenAI,
+		Type:        AccountTypeOAuth,
+		Status:      StatusActive,
+		Schedulable: true,
+		Concurrency: 1,
+		Credentials: map[string]any{
+			"auth_mode":          OpenAIAuthModeAgentIdentity,
+			"agent_runtime_id":   key.runtimeID,
+			"agent_private_key":  privateKey,
+			"task_id":            "task-old",
+			"chatgpt_account_id": "account-agent-retry",
+		},
+	}
+	repo := &agentIdentityForwardRepo{account: account}
+	registerCalls := 0
+	registerServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		registerCalls++
+		_, _ = io.WriteString(w, `{"task_id":"task-new"}`)
+	}))
+	defer registerServer.Close()
+	oldBase := openAIAgentIdentityAuthAPIBaseURL
+	openAIAgentIdentityAuthAPIBaseURL = registerServer.URL
+	t.Cleanup(func() { openAIAgentIdentityAuthAPIBaseURL = oldBase })
+
+	successBody := `{"id":"resp-agent-retry","object":"response","model":"gpt-5.4","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`
+	upstream := &httpUpstreamRecorder{responses: []*http.Response{
+		{StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))},
+		{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(successBody))},
+	}}
+	svc := &OpenAIGatewayService{cfg: &config.Config{}, accountRepo: repo, httpUpstream: upstream}
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`))
+
+	_, err := svc.Forward(context.Background(), c, account, []byte(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`))
+	require.NoError(t, err)
+	require.Equal(t, 1, registerCalls)
+	require.Len(t, upstream.requests, 2)
+	require.NotEqual(t, upstream.requests[0].Header.Get("Authorization"), upstream.requests[1].Header.Get("Authorization"))
+	require.Equal(t, "task-new", decodeAgentAssertionTask(t, upstream.requests[1].Header.Get("Authorization")))
+
+	// Two consecutive invalid responses still produce only one retry for this
+	// request; the recovery path must not loop indefinitely.
+	upstream.responses = []*http.Response{
+		{StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))},
+		{StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))},
+	}
+	rec2 := httptest.NewRecorder()
+	c2, _ := gin.CreateTestContext(rec2)
+	c2.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`))
+	_, err = svc.Forward(context.Background(), c2, account, []byte(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`))
+	require.Error(t, err)
+	require.Equal(t, 2, registerCalls)
+	require.Len(t, upstream.requests, 4)
+}
+
+func decodeAgentAssertionTask(t *testing.T, header string) string {
+	t.Helper()
+	encoded := strings.TrimPrefix(header, "AgentAssertion ")
+	decoded, err := base64.RawURLEncoding.DecodeString(encoded)
+	require.NoError(t, err)
+	var envelope struct {
+		TaskID string `json:"task_id"`
+	}
+	require.NoError(t, json.Unmarshal(decoded, &envelope))
+	return envelope.TaskID
+}
+
+type agentIdentityForwardRepo struct {
+	AccountRepository
+	account *Account
+}
+
+func (r *agentIdentityForwardRepo) GetByID(_ context.Context, _ int64) (*Account, error) {
+	return r.account, nil
+}
+
+func (r *agentIdentityForwardRepo) UpdateCredentials(_ context.Context, _ int64, credentials map[string]any) error {
+	r.account.Credentials = credentials
+	return nil
+}

backend/internal/service/openai_agent_identity_test.go

@@ -0,0 +1,177 @@
+package service
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/sha512"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/crypto/curve25519"
+	"golang.org/x/crypto/nacl/box"
+)
+
+func newTestAgentIdentityKey(t *testing.T) (agentIdentityKey, string) {
+	t.Helper()
+	_, privateKey, err := ed25519.GenerateKey(rand.Reader)
+	require.NoError(t, err)
+	der, err := x509.MarshalPKCS8PrivateKey(privateKey)
+	require.NoError(t, err)
+	return agentIdentityKey{
+		runtimeID:  "runtime-test",
+		privateKey: privateKey,
+		taskID:     "task-test",
+	}, base64.StdEncoding.EncodeToString(der)
+}
+
+func TestBuildAgentAssertionMatchesCodexEnvelopeAndSignature(t *testing.T) {
+	key, _ := newTestAgentIdentityKey(t)
+	now := time.Date(2026, 7, 14, 8, 9, 10, 0, time.FixedZone("UTC+8", 8*60*60))
+	assertion, err := buildAgentAssertion(key, now)
+	require.NoError(t, err)
+	require.True(t, strings.HasPrefix(assertion, "AgentAssertion "))
+
+	encoded := strings.TrimPrefix(assertion, "AgentAssertion ")
+	decoded, err := base64.RawURLEncoding.DecodeString(encoded)
+	require.NoError(t, err)
+	var envelope struct {
+		AgentRuntimeID string `json:"agent_runtime_id"`
+		TaskID         string `json:"task_id"`
+		Timestamp      string `json:"timestamp"`
+		Signature      string `json:"signature"`
+	}
+	require.NoError(t, json.Unmarshal(decoded, &envelope))
+	require.Equal(t, "runtime-test", envelope.AgentRuntimeID)
+	require.Equal(t, "task-test", envelope.TaskID)
+	require.Equal(t, "2026-07-14T00:09:10Z", envelope.Timestamp)
+	signature, err := base64.StdEncoding.DecodeString(envelope.Signature)
+	require.NoError(t, err)
+	require.True(t, ed25519.Verify(key.privateKey.Public().(ed25519.PublicKey), []byte("runtime-test:task-test:2026-07-14T00:09:10Z"), signature))
+}
+
+func TestDecryptAgentTaskIDSupportsCodexSealedBoxResponse(t *testing.T) {
+	key, _ := newTestAgentIdentityKey(t)
+	digest := sha512.Sum512(key.privateKey.Seed())
+	var curvePrivate [32]byte
+	copy(curvePrivate[:], digest[:32])
+	curvePrivate[0] &= 248
+	curvePrivate[31] &= 127
+	curvePrivate[31] |= 64
+	curvePublicBytes, err := curve25519.X25519(curvePrivate[:], curve25519.Basepoint)
+	require.NoError(t, err)
+	var curvePublic [32]byte
+	copy(curvePublic[:], curvePublicBytes)
+	ciphertext, err := box.SealAnonymous(nil, []byte("task-sealed"), &curvePublic, rand.Reader)
+	require.NoError(t, err)
+	got, err := decryptAgentTaskID(key, base64.StdEncoding.EncodeToString(ciphertext))
+	require.NoError(t, err)
+	require.Equal(t, "task-sealed", got)
+}
+
+func TestRegisterAgentIdentityTaskAcceptsPlaintextAndEncryptedResponses(t *testing.T) {
+	key, privateKey := newTestAgentIdentityKey(t)
+	requestCount := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, http.MethodPost, r.Method)
+		require.Equal(t, "/v1/agent/runtime-test/task/register", r.URL.Path)
+		var request map[string]string
+		require.NoError(t, json.NewDecoder(r.Body).Decode(&request))
+		require.NotEmpty(t, request["timestamp"])
+		require.NotEmpty(t, request["signature"])
+		requestCount++
+		if requestCount == 2 {
+			digest := sha512.Sum512(key.privateKey.Seed())
+			var curvePrivate [32]byte
+			copy(curvePrivate[:], digest[:32])
+			curvePrivate[0] &= 248
+			curvePrivate[31] &= 127
+			curvePrivate[31] |= 64
+			curvePublicBytes, curveErr := curve25519.X25519(curvePrivate[:], curve25519.Basepoint)
+			require.NoError(t, curveErr)
+			var curvePublic [32]byte
+			copy(curvePublic[:], curvePublicBytes)
+			ciphertext, sealErr := box.SealAnonymous(nil, []byte("task-encrypted"), &curvePublic, rand.Reader)
+			require.NoError(t, sealErr)
+			_, _ = fmt.Fprintf(w, `{"encrypted_task_id":%q}`, base64.StdEncoding.EncodeToString(ciphertext))
+			return
+		}
+		_, _ = w.Write([]byte(`{"task_id":"task-plain"}`))
+	}))
+	defer server.Close()
+	oldBase := openAIAgentIdentityAuthAPIBaseURL
+	openAIAgentIdentityAuthAPIBaseURL = server.URL
+	t.Cleanup(func() { openAIAgentIdentityAuthAPIBaseURL = oldBase })
+
+	account := &Account{ID: 1, Type: AccountTypeOAuth, Platform: PlatformOpenAI, Credentials: map[string]any{
+		"auth_mode":         OpenAIAuthModeAgentIdentity,
+		"agent_runtime_id":  key.runtimeID,
+		"agent_private_key": privateKey,
+	}}
+	taskID, err := registerAgentIdentityTask(context.Background(), account)
+	require.NoError(t, err)
+	require.Equal(t, "task-plain", taskID)
+	taskID, err = registerAgentIdentityTask(context.Background(), account)
+	require.NoError(t, err)
+	require.Equal(t, "task-encrypted", taskID)
+}
+
+func TestEnsureAgentIdentityTaskPersistsAndRedactsCredentials(t *testing.T) {
+	key, privateKey := newTestAgentIdentityKey(t)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(`{"task_id":"task-persisted"}`))
+	}))
+	defer server.Close()
+	oldBase := openAIAgentIdentityAuthAPIBaseURL
+	openAIAgentIdentityAuthAPIBaseURL = server.URL
+	t.Cleanup(func() { openAIAgentIdentityAuthAPIBaseURL = oldBase })
+
+	repo := &agentIdentityCredentialsRepo{}
+	account := &Account{ID: 7, Type: AccountTypeOAuth, Platform: PlatformOpenAI, Credentials: map[string]any{
+		"auth_mode":          OpenAIAuthModeAgentIdentity,
+		"agent_runtime_id":   key.runtimeID,
+		"agent_private_key":  privateKey,
+		"chatgpt_account_id": "account-test",
+	}}
+	service := &OpenAIGatewayService{accountRepo: repo}
+	require.NoError(t, service.ensureAgentIdentityTask(context.Background(), account, ""))
+	require.Equal(t, "task-persisted", account.GetCredential("task_id"))
+	require.Equal(t, "task-persisted", repo.credentials["task_id"])
+	require.True(t, IsSensitiveCredentialKey("agent_private_key"))
+	redacted := make(map[string]any)
+	for key, value := range account.Credentials {
+		if !IsSensitiveCredentialKey(key) {
+			redacted[key] = value
+		}
+	}
+	require.NotContains(t, string(mustJSON(t, redacted)), privateKey)
+}
+
+type agentIdentityCredentialsRepo struct {
+	AccountRepository
+	credentials map[string]any
+	mu          sync.Mutex
+}
+
+func (r *agentIdentityCredentialsRepo) UpdateCredentials(_ context.Context, _ int64, credentials map[string]any) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.credentials = credentials
+	return nil
+}
+
+func mustJSON(t *testing.T, value any) []byte {
+	t.Helper()
+	encoded, err := json.Marshal(value)
+	require.NoError(t, err)
+	return encoded
+}

backend/internal/service/openai_codex_models_service.go

@@ -42,7 +42,7 @@ func (s *OpenAIGatewayService) FetchCodexModelsManifest(ctx context.Context, acc
 		return nil, infraerrors.Newf(http.StatusInternalServerError, "OPENAI_CODEX_MODELS_CREDENTIALS_FAILED", "resolve credential account: %v", err)
 	}
 	accessToken := credAccount.GetOpenAIAccessToken()
-	if accessToken == "" {
+	if accessToken == "" && !credAccount.IsOpenAIAgentIdentity() {
 		return nil, infraerrors.New(http.StatusBadGateway, "OPENAI_CODEX_MODELS_TOKEN_MISSING", "account has no Codex backend access token")
 	}
 
@@ -58,7 +58,15 @@ func (s *OpenAIGatewayService) FetchCodexModelsManifest(ctx context.Context, acc
 	if err != nil {
 		return nil, infraerrors.Newf(http.StatusInternalServerError, "OPENAI_CODEX_MODELS_REQUEST_FAILED", "create codex models request: %v", err)
 	}
-	req.Header.Set("Authorization", "Bearer "+accessToken)
+	authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, credAccount, accessToken)
+	if err != nil {
+		return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_AUTH_FAILED", "build Codex models authentication: %v", err)
+	}
+	for key, values := range authHeaders {
+		for _, value := range values {
+			req.Header.Add(key, value)
+		}
+	}
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("Originator", "codex_cli_rs")
 	req.Header.Set("Version", clientVersion)

backend/internal/service/openai_gateway_count_tokens.go

@@ -215,7 +215,15 @@ func (s *OpenAIGatewayService) buildInputTokensUpstreamRequest(
 		return nil, err
 	}
 	req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
-	req.Header.Set("authorization", "Bearer "+token)
+	authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token)
+	if err != nil {
+		return nil, err
+	}
+	for key, values := range authHeaders {
+		for _, value := range values {
+			req.Header.Add(key, value)
+		}
+	}
 	req.Header.Set("content-type", "application/json")
 	req.Header.Set("accept", "application/json")
 

backend/internal/service/openai_gateway_forward.go

@@ -487,6 +487,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 		var wsResult *OpenAIForwardResult
 		var wsErr error
 		wsLastFailureReason := ""
+		agentTaskRecoveryTried := false
 		wsPrevResponseRecoveryTried := false
 		wsInvalidEncryptedContentRecoveryTried := false
 		recoverPrevResponseNotFound := func(attempt int) bool {
@@ -571,6 +572,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 				startTime,
 				attempt,
 				wsLastFailureReason,
+				&agentTaskRecoveryTried,
 			)
 			if wsErr == nil {
 				break
@@ -578,6 +580,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 			if c != nil && c.Writer != nil && c.Writer.Written() {
 				break
 			}
+			var taskRecoveredErr *agentIdentityTaskRecoveredError
+			if errors.As(wsErr, &taskRecoveredErr) {
+				continue
+			}
 
 			reason, retryable := classifyOpenAIWSReconnectReason(wsErr)
 			if reason != "" {
@@ -684,6 +690,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 	}
 
 	httpInvalidEncryptedContentRetryTried := false
+	agentTaskRecoveryTried := false
 	for {
 		// Build upstream request
 		upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
@@ -719,6 +726,16 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 			upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
 			upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
 			upstreamCode := extractUpstreamErrorCode(respBody)
+			if !agentTaskRecoveryTried && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, respBody) {
+				agentTaskRecoveryTried = true
+				expectedTaskID := account.GetCredential("task_id")
+				if err := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); err != nil {
+					return nil, fmt.Errorf("agent identity task recovery failed: %w", err)
+				}
+				continue
+			}
+			respBody = s.redactAgentIdentitySensitiveBody(ctx, account, respBody)
+			resp.Body = io.NopCloser(bytes.NewReader(respBody))
 			if !httpInvalidEncryptedContentRetryTried && resp.StatusCode == http.StatusBadRequest && upstreamCode == "invalid_encrypted_content" {
 				decoded, decodeErr := ensureReqBody()
 				if decodeErr != nil {
@@ -869,8 +886,17 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin.
 	}
 	req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
 
-	// Set authentication header
-	req.Header.Set("authorization", "Bearer "+token)
+	// Build authentication for this request. Agent Identity signs a fresh
+	// assertion here; OAuth/PAT/API-key keep their existing Bearer behavior.
+	authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token)
+	if err != nil {
+		return nil, fmt.Errorf("build openai authentication headers: %w", err)
+	}
+	for key, values := range authHeaders {
+		for _, value := range values {
+			req.Header.Add(key, value)
+		}
+	}
 
 	// Set headers specific to OAuth accounts (ChatGPT internal API)
 	if account.Type == AccountTypeOAuth {

backend/internal/service/openai_gateway_passthrough.go

@@ -338,7 +338,15 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough(
 	req.Header.Del("authorization")
 	req.Header.Del("x-api-key")
 	req.Header.Del("x-goog-api-key")
-	req.Header.Set("authorization", "Bearer "+token)
+	authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token)
+	if err != nil {
+		return nil, fmt.Errorf("build openai authentication headers: %w", err)
+	}
+	for key, values := range authHeaders {
+		for _, value := range values {
+			req.Header.Add(key, value)
+		}
+	}
 
 	// OAuth 透传到 ChatGPT internal API 时补齐必要头。
 	if account.Type == AccountTypeOAuth {
@@ -434,6 +442,7 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough(
 	requestBody []byte,
 ) error {
 	body := s.readUpstreamErrorBody(resp)
+	body = s.redactAgentIdentitySensitiveBody(ctx, account, body)
 
 	upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
 	upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
@@ -477,6 +486,7 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough(
 ) error {
 	MarkResponseCommitted(c)
 	body := s.readUpstreamErrorBody(resp)
+	body = s.redactAgentIdentitySensitiveBody(ctx, account, body)
 
 	// cyber_policy：透传账号本就把原始 body 回给客户端（下方 c.Data），此处仅打标记，
 	// 供 handler 事后写风控/邮件。cyber 是上游网络安全策略拦截，不冷却账号，

backend/internal/service/openai_gateway_service.go

@@ -356,6 +356,7 @@ type OpenAIGatewayService struct {
 	openaiWSStateStoreOnce        sync.Once
 	openaiSchedulerOnce           sync.Once
 	openaiWSPassthroughDialerOnce sync.Once
+	agentIdentityTaskMu           sync.Mutex
 	openaiWSPool                  *openAIWSConnPool
 	openaiWSStateStore            OpenAIWSStateStore
 	openaiScheduler               OpenAIAccountScheduler
@@ -1052,6 +1053,9 @@ func (s *OpenAIGatewayService) GetAccessToken(ctx context.Context, account *Acco
 	}
 	switch account.Type {
 	case AccountTypeOAuth:
+		if account.IsOpenAIAgentIdentity() {
+			return "", OpenAIAuthModeAgentIdentity, nil
+		}
 		if account.Platform == PlatformGrok {
 			if s.grokTokenProvider != nil {
 				accessToken, err := s.grokTokenProvider.GetAccessToken(ctx, account)

backend/internal/service/openai_gateway_upstream_errors.go

@@ -279,6 +279,7 @@ func (s *OpenAIGatewayService) handleErrorResponse(
 	requestedModel ...string,
 ) (*OpenAIForwardResult, error) {
 	body := s.readUpstreamErrorBody(resp)
+	body = s.redactAgentIdentitySensitiveBody(ctx, account, body)
 
 	// cyber_policy 硬阻断：透传上游原始错误体给客户端（不重包成通用 502），不冷却账号。
 	// 当前请求恒透传（需求1）；标记供 handler 事后写风控/邮件。400 cyber 不可 failover

backend/internal/service/openai_images.go

@@ -752,7 +752,15 @@ func (s *OpenAIGatewayService) buildOpenAIImagesRequest(
 		return nil, err
 	}
 	req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI))
-	req.Header.Set("Authorization", "Bearer "+token)
+	authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token)
+	if err != nil {
+		return nil, fmt.Errorf("build openai authentication headers: %w", err)
+	}
+	for key, values := range authHeaders {
+		for _, value := range values {
+			req.Header.Add(key, value)
+		}
+	}
 	for key, values := range c.Request.Header {
 		if !openaiPassthroughAllowedHeaders[strings.ToLower(key)] {
 			continue

backend/internal/service/openai_quota_service.go

@@ -10,6 +10,7 @@ import (
 	"log/slog"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
@@ -118,6 +119,7 @@ type OpenAIQuotaService struct {
 	proxyRepo            ProxyRepository
 	tokenProvider        *OpenAITokenProvider
 	privacyClientFactory PrivacyClientFactory
+	agentIdentityTaskMu  sync.Mutex
 }
 
 // NewOpenAIQuotaService constructs a quota service. token provider is required —
@@ -154,10 +156,14 @@ func (s *OpenAIQuotaService) QueryUsage(ctx context.Context, accountID int64) (*
 	callCtx, cancel := context.WithTimeout(ctx, openaiQuotaUpstreamTimeout)
 	defer cancel()
 
+	quotaHeaders, headerErr := s.buildCodexQuotaHeaders(callCtx, accountID, accessToken, chatGPTAccountID, fedRAMP)
+	if headerErr != nil {
+		return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_AUTH_FAILED", "failed to build upstream authentication: %v", headerErr)
+	}
 	var payload OpenAIQuotaUsage
 	resp, err := client.R().
 		SetContext(callCtx).
-		SetHeaders(buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP)).
+		SetHeaders(quotaHeaders).
 		SetSuccessResult(&payload).
 		Get(chatGPTUsageURL)
 	if err != nil {
@@ -178,9 +184,14 @@ func (s *OpenAIQuotaService) QueryUsage(ctx context.Context, accountID int64) (*
 }
 
 func (s *OpenAIQuotaService) queryResetCreditDetails(ctx context.Context, client *req.Client, accessToken, chatGPTAccountID string, fedRAMP bool, accountID int64) []OpenAIRateLimitResetCreditDetail {
+	quotaHeaders, headerErr := s.buildCodexQuotaHeaders(ctx, accountID, accessToken, chatGPTAccountID, fedRAMP)
+	if headerErr != nil {
+		slog.Warn("openai_quota_reset_credit_details_auth_failed", "account_id", accountID, "error", headerErr)
+		return nil
+	}
 	resp, err := client.R().
 		SetContext(ctx).
-		SetHeaders(buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP)).
+		SetHeaders(quotaHeaders).
 		Get(chatGPTRateLimitCreditsURL)
 	if err != nil {
 		slog.Warn("openai_quota_reset_credit_details_failed", "account_id", accountID, "error", err)
@@ -239,7 +250,10 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) (
 	callCtx, cancel := context.WithTimeout(ctx, openaiQuotaUpstreamTimeout)
 	defer cancel()
 
-	headers := buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP)
+	headers, headerErr := s.buildCodexQuotaHeaders(callCtx, accountID, accessToken, chatGPTAccountID, fedRAMP)
+	if headerErr != nil {
+		return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_AUTH_FAILED", "failed to build upstream authentication: %v", headerErr)
+	}
 	headers["content-type"] = "application/json"
 
 	var payload OpenAIQuotaResetResult
@@ -271,7 +285,7 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) (
 // token via the shared TokenProvider, and resolves the chatgpt-account-id and
 // proxy URL. Centralized so QueryUsage / ResetCredit share validation.
 func (s *OpenAIQuotaService) prepareUpstreamCall(ctx context.Context, accountID int64) (accessToken, chatGPTAccountID, proxyURL string, fedRAMP bool, err error) {
-	if s == nil || s.accountRepo == nil || s.tokenProvider == nil || s.privacyClientFactory == nil {
+	if s == nil || s.accountRepo == nil || s.privacyClientFactory == nil {
 		return "", "", "", false, infraerrors.New(http.StatusInternalServerError, "OPENAI_QUOTA_NOT_CONFIGURED", "openai quota service is not configured")
 	}
 
@@ -309,12 +323,17 @@ func (s *OpenAIQuotaService) prepareUpstreamCall(ctx context.Context, accountID
 		return "", "", "", false, infraerrors.New(http.StatusBadRequest, "OPENAI_QUOTA_MISSING_ACCOUNT_ID", "chatgpt_account_id is missing; please re-authorize this account")
 	}
 
-	accessToken, err = s.tokenProvider.GetAccessToken(ctx, account)
-	if err != nil {
-		return "", "", "", false, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "failed to acquire access token: %v", err)
-	}
-	if strings.TrimSpace(accessToken) == "" {
-		return "", "", "", false, infraerrors.New(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "access token is empty")
+	if !account.IsOpenAIAgentIdentity() {
+		if s.tokenProvider == nil {
+			return "", "", "", false, infraerrors.New(http.StatusInternalServerError, "OPENAI_QUOTA_NOT_CONFIGURED", "openai quota token provider is not configured")
+		}
+		accessToken, err = s.tokenProvider.GetAccessToken(ctx, account)
+		if err != nil {
+			return "", "", "", false, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "failed to acquire access token: %v", err)
+		}
+		if strings.TrimSpace(accessToken) == "" {
+			return "", "", "", false, infraerrors.New(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "access token is empty")
+		}
 	}
 	fedRAMP = account.IsChatGPTAccountFedRAMP()
 
@@ -337,6 +356,43 @@ func (s *OpenAIQuotaService) prepareUpstreamCall(ctx context.Context, accountID
 	return accessToken, chatGPTAccountID, proxyURL, fedRAMP, nil
 }
 
+func (s *OpenAIQuotaService) buildCodexQuotaHeaders(ctx context.Context, accountID int64, accessToken, chatGPTAccountID string, fedRAMP bool) (map[string]string, error) {
+	headers := buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP)
+	if s == nil || s.accountRepo == nil {
+		return headers, nil
+	}
+	account, err := s.accountRepo.GetByID(ctx, accountID)
+	if err != nil || account == nil {
+		if strings.TrimSpace(accessToken) == "" {
+			return nil, fmt.Errorf("agent identity account credentials are unavailable")
+		}
+		return headers, nil
+	}
+	if account.IsShadow() {
+		if resolved, resolveErr := resolveCredentialAccount(ctx, s.accountRepo, account); resolveErr == nil && resolved != nil {
+			account = resolved
+		} else if strings.TrimSpace(accessToken) == "" {
+			return nil, fmt.Errorf("agent identity shadow credentials are unavailable")
+		}
+	}
+	if !account.IsOpenAIAgentIdentity() {
+		return headers, nil
+	}
+	if err := ensureAgentIdentityTaskForAccount(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, account, ""); err != nil {
+		return nil, err
+	}
+	key, err := agentIdentityKeyFromAccount(account)
+	if err != nil {
+		return nil, err
+	}
+	assertion, err := buildAgentAssertion(key, time.Now())
+	if err != nil {
+		return nil, err
+	}
+	headers["authorization"] = assertion
+	return headers, nil
+}
+
 // buildCodexCommonHeaders sets the request headers expected by the chatgpt.com
 // backend so calls succeed past Cloudflare/WASM checks.
 func buildCodexCommonHeaders(accessToken, chatGPTAccountID string, fedRAMP bool) map[string]string {

backend/internal/service/openai_quota_spark_window_test.go

@@ -2,6 +2,10 @@ package service
 
 import (
 	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/x509"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -214,6 +218,45 @@ func TestPrepareUpstreamCallShadowResolve(t *testing.T) {
 		"prepareUpstreamCall should use parent's chatgpt_account_id after shadow resolve")
 }
 
+func TestQueryUsageAgentIdentityUsesAssertionWithoutOAuthToken(t *testing.T) {
+	_, privateKey, err := ed25519.GenerateKey(rand.Reader)
+	require.NoError(t, err)
+	der, err := x509.MarshalPKCS8PrivateKey(privateKey)
+	require.NoError(t, err)
+	account := &Account{
+		ID:       300,
+		Platform: PlatformOpenAI,
+		Type:     AccountTypeOAuth,
+		Credentials: map[string]any{
+			"auth_mode":                  OpenAIAuthModeAgentIdentity,
+			"agent_runtime_id":           "runtime-quota",
+			"agent_private_key":          base64.StdEncoding.EncodeToString(der),
+			"task_id":                    "task-quota",
+			"chatgpt_account_id":         "account-quota",
+			"chatgpt_account_is_fedramp": true,
+		},
+	}
+	repo := &stubQuotaAccountRepo{accounts: map[int64]*Account{account.ID: account}}
+	var authorization string
+	var accountHeader string
+	var fedrampHeader string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		authorization = r.Header.Get("authorization")
+		accountHeader = r.Header.Get("chatgpt-account-id")
+		fedrampHeader = r.Header.Get("x-openai-fedramp")
+		w.Header().Set("content-type", "application/json")
+		_, _ = w.Write([]byte(`{"plan_type":"pro","rate_limit":{"allowed":true}}`))
+	}))
+	defer srv.Close()
+	svc := NewOpenAIQuotaService(repo, nil, nil, newQuotaRedirectingFactory(srv))
+	usage, err := svc.QueryUsage(context.Background(), account.ID)
+	require.NoError(t, err)
+	require.NotNil(t, usage)
+	require.True(t, strings.HasPrefix(authorization, "AgentAssertion "))
+	require.Equal(t, "account-quota", accountHeader)
+	require.Equal(t, "true", fedrampHeader)
+}
+
 func TestParseOpenAIRateLimitResetCreditDetails_CompatibleContainers(t *testing.T) {
 	tests := []struct {
 		name string

backend/internal/service/openai_ws_forwarder_ingress.go

@@ -539,6 +539,9 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 		Account: account,
 		WSURL:   wsURL,
 		Headers: wsHeaders,
+		HeadersFactory: func(factoryCtx context.Context, headers http.Header) (http.Header, error) {
+			return s.refreshOpenAIAgentIdentityHeaders(factoryCtx, account, headers)
+		},
 		ProxyURL: func() string {
 			if account.ProxyID != nil && account.Proxy != nil {
 				return account.Proxy.URL()
@@ -602,7 +605,9 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 		acquireTimeout = 30 * time.Second
 	}
 
-	acquireTurnLease := func(turn int, preferred string, forcePreferredConn bool) (*openAIWSConnLease, error) {
+	agentTaskRecoveryTried := false
+	var acquireTurnLease func(int, string, bool) (*openAIWSConnLease, error)
+	acquireTurnLease = func(turn int, preferred string, forcePreferredConn bool) (*openAIWSConnLease, error) {
 		req := cloneOpenAIWSAcquireRequest(baseAcquireReq)
 		req.PreferredConnID = strings.TrimSpace(preferred)
 		req.ForcePreferredConn = forcePreferredConn
@@ -611,6 +616,14 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 		acquireCtx, acquireCancel := context.WithTimeout(ctx, acquireTimeout)
 		lease, acquireErr := pool.Acquire(acquireCtx, req)
 		acquireCancel()
+		var dialErr *openAIWSDialError
+		if acquireErr != nil && s.isAgentIdentityAccount(ctx, account) && errors.As(acquireErr, &dialErr) && dialErr != nil && dialErr.StatusCode == http.StatusUnauthorized && !agentTaskRecoveryTried {
+			agentTaskRecoveryTried = true
+			if recoveryErr := s.recoverAgentIdentityTask(ctx, account, account.GetCredential("task_id")); recoveryErr != nil {
+				return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr)
+			}
+			return acquireTurnLease(turn, preferred, forcePreferredConn)
+		}
 		if acquireErr != nil {
 			dialStatus, dialClass, dialCloseStatus, dialCloseReason, dialRespServer, dialRespVia, dialRespCFRay, dialRespReqID := summarizeOpenAIWSDialError(acquireErr)
 			logOpenAIWSModeInfo(

backend/internal/service/openai_ws_forwarder_payload.go

@@ -67,7 +67,9 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders(
 	promptCacheKey string,
 ) (http.Header, openAIWSSessionHeaderResolution, error) {
 	headers := make(http.Header)
-	headers.Set("authorization", "Bearer "+token)
+	if account == nil || !account.IsOpenAIAgentIdentity() {
+		headers.Set("authorization", "Bearer "+token)
+	}
 
 	sessionResolution := resolveOpenAIWSSessionHeaders(c, promptCacheKey)
 	if c != nil && c.Request != nil {

backend/internal/service/openai_ws_forwarder_v2.go

@@ -30,6 +30,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
 	startTime time.Time,
 	attempt int,
 	lastFailureReason string,
+	agentTaskRecoveryTried *bool,
 ) (*OpenAIForwardResult, error) {
 	if s == nil || account == nil {
 		return nil, wrapOpenAIWSFallback("invalid_state", errors.New("service or account is nil"))
@@ -174,9 +175,12 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
 	defer acquireCancel()
 
 	lease, err := s.getOpenAIWSConnPool().Acquire(acquireCtx, openAIWSAcquireRequest{
-		Account:         account,
-		WSURL:           wsURL,
-		Headers:         wsHeaders,
+		Account: account,
+		WSURL:   wsURL,
+		Headers: wsHeaders,
+		HeadersFactory: func(factoryCtx context.Context, headers http.Header) (http.Header, error) {
+			return s.refreshOpenAIAgentIdentityHeaders(factoryCtx, account, headers)
+		},
 		PreferredConnID: preferredConnID,
 		ForceNewConn:    forceNewConn,
 		ProxyURL: func() string {
@@ -187,6 +191,14 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
 		}(),
 	})
 	if err != nil {
+		var agentDialErr *openAIWSDialError
+		if s.isAgentIdentityAccount(ctx, account) && errors.As(err, &agentDialErr) && agentDialErr != nil && agentDialErr.StatusCode == http.StatusUnauthorized && agentTaskRecoveryTried != nil && !*agentTaskRecoveryTried {
+			*agentTaskRecoveryTried = true
+			if recoveryErr := s.recoverAgentIdentityTask(ctx, account, account.GetCredential("task_id")); recoveryErr != nil {
+				return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr)
+			}
+			return nil, &agentIdentityTaskRecoveredError{}
+		}
 		dialStatus, dialClass, dialCloseStatus, dialCloseReason, dialRespServer, dialRespVia, dialRespCFRay, dialRespReqID := summarizeOpenAIWSDialError(err)
 		logOpenAIWSModeInfo(
 			"acquire_fail account_id=%d account_type=%s transport=%s reason=%s dial_status=%d dial_class=%s dial_close_status=%s dial_close_reason=%s dial_resp_server=%s dial_resp_via=%s dial_resp_cf_ray=%s dial_resp_x_request_id=%s cause=%s preferred_conn_id=%s force_new_conn=%v ws_host=%s ws_path=%s proxy_enabled=%v",

backend/internal/service/openai_ws_pool.go

@@ -60,9 +60,13 @@ func (e *openAIWSDialError) Unwrap() error {
 }
 
 type openAIWSAcquireRequest struct {
-	Account         *Account
-	WSURL           string
-	Headers         http.Header
+	Account *Account
+	WSURL   string
+	Headers http.Header
+	// HeadersFactory is evaluated inside dialConn. It exists so credentials
+	// whose authorization is per-dial (Agent Identity) are never cached in
+	// lastAcquire or delayed prewarm state.
+	HeadersFactory  func(context.Context, http.Header) (http.Header, error)
 	ProxyURL        string
 	PreferredConnID string
 	// ForceNewConn: 强制本次获取新连接（避免复用导致连接内续链状态互相污染）。
@@ -517,6 +521,7 @@ type openAIWSAccountPool struct {
 	conns         map[string]*openAIWSConn
 	pinnedConns   map[string]int
 	creating      int
+	generation    uint64
 	lastCleanupAt time.Time
 	lastAcquire   *openAIWSAcquireRequest
 	prewarmActive bool
@@ -1283,6 +1288,7 @@ func (p *openAIWSConnPool) ensureTargetIdleAsync(accountID int64) {
 	}
 
 	var req openAIWSAcquireRequest
+	generation := uint64(0)
 	need := 0
 	ap, ok := p.getAccountPool(accountID)
 	if !ok || ap == nil {
@@ -1317,6 +1323,7 @@ func (p *openAIWSConnPool) ensureTargetIdleAsync(accountID int64) {
 		return
 	}
 	req = cloneOpenAIWSAcquireRequest(*ap.lastAcquire)
+	generation = ap.generation
 	ap.prewarmActive = true
 	if cooldown := p.prewarmCooldown(); cooldown > 0 {
 		ap.prewarmUntil = now.Add(cooldown)
@@ -1324,7 +1331,7 @@ func (p *openAIWSConnPool) ensureTargetIdleAsync(accountID int64) {
 	ap.creating += need
 	p.metrics.scaleUpTotal.Add(int64(need))
 
-	go p.prewarmConns(accountID, req, need)
+	go p.prewarmConns(accountID, req, need, generation)
 }
 
 func (p *openAIWSConnPool) targetConnCountLocked(ap *openAIWSAccountPool, maxConns int) int {
@@ -1367,7 +1374,11 @@ func (p *openAIWSConnPool) targetConnCountLocked(ap *openAIWSAccountPool, maxCon
 	return target
 }
 
-func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequest, total int) {
+func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequest, total int, generations ...uint64) {
+	generation := uint64(0)
+	if len(generations) > 0 {
+		generation = generations[0]
+	}
 	defer func() {
 		if ap, ok := p.getAccountPool(accountID); ok && ap != nil {
 			ap.mu.Lock()
@@ -1398,6 +1409,11 @@ func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequ
 			ap.mu.Unlock()
 			continue
 		}
+		if ap.generation != generation || ap.lastAcquire == nil {
+			ap.mu.Unlock()
+			conn.close()
+			continue
+		}
 		if len(ap.conns) >= p.effectiveMaxConnsByAccount(req.Account) {
 			ap.mu.Unlock()
 			conn.close()
@@ -1410,6 +1426,35 @@ func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequ
 	}
 }
 
+// ClearAccount closes all pooled connections and discards delayed prewarm
+// state for one account. The generation guard prevents an in-flight prewarm
+// started before credential recovery from re-entering the pool afterwards.
+func (p *openAIWSConnPool) ClearAccount(accountID int64) {
+	if p == nil || accountID <= 0 {
+		return
+	}
+	ap, ok := p.getAccountPool(accountID)
+	if !ok || ap == nil {
+		return
+	}
+	ap.mu.Lock()
+	ap.generation++
+	conns := make([]*openAIWSConn, 0, len(ap.conns))
+	for id, conn := range ap.conns {
+		delete(ap.conns, id)
+		delete(ap.pinnedConns, id)
+		if conn != nil {
+			conns = append(conns, conn)
+		}
+	}
+	ap.lastAcquire = nil
+	ap.prewarmUntil = time.Time{}
+	ap.prewarmFails = 0
+	ap.prewarmFailAt = time.Time{}
+	ap.mu.Unlock()
+	closeOpenAIWSConns(conns)
+}
+
 func (p *openAIWSConnPool) evictConn(accountID int64, connID string) {
 	if p == nil || accountID <= 0 || stringsTrim(connID) == "" {
 		return
@@ -1485,7 +1530,15 @@ func (p *openAIWSConnPool) dialConn(ctx context.Context, req openAIWSAcquireRequ
 	if p == nil || p.clientDialer == nil {
 		return nil, errors.New("openai ws client dialer is nil")
 	}
-	conn, status, handshakeHeaders, err := p.clientDialer.Dial(ctx, req.WSURL, req.Headers, req.ProxyURL)
+	headers := cloneHeader(req.Headers)
+	var err error
+	if req.HeadersFactory != nil {
+		headers, err = req.HeadersFactory(ctx, headers)
+		if err != nil {
+			return nil, err
+		}
+	}
+	conn, status, handshakeHeaders, err := p.clientDialer.Dial(ctx, req.WSURL, headers, req.ProxyURL)
 	if err != nil {
 		return nil, &openAIWSDialError{
 			StatusCode:      status,

backend/internal/service/openai_ws_v2_passthrough_adapter.go

@@ -359,10 +359,28 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough(
 		return errors.New("openai ws passthrough dialer is nil")
 	}
 
-	dialCtx, cancelDial := context.WithTimeout(ctx, s.openAIWSDialTimeout())
-	defer cancelDial()
-	upstreamConn, statusCode, handshakeHeaders, err := dialer.Dial(dialCtx, wsURL, headers, proxyURL)
-	if err != nil {
+	agentTaskRecoveryTried := false
+	var upstreamConn openAIWSClientConn
+	statusCode := 0
+	var handshakeHeaders http.Header
+	for {
+		headers, err = s.refreshOpenAIAgentIdentityHeaders(ctx, account, headers)
+		if err != nil {
+			return fmt.Errorf("refresh ws authentication headers: %w", err)
+		}
+		dialCtx, cancelDial := context.WithTimeout(ctx, s.openAIWSDialTimeout())
+		upstreamConn, statusCode, handshakeHeaders, err = dialer.Dial(dialCtx, wsURL, headers, proxyURL)
+		cancelDial()
+		if err == nil {
+			break
+		}
+		if s.isAgentIdentityAccount(ctx, account) && statusCode == http.StatusUnauthorized && !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)
+			}
+			continue
+		}
 		logOpenAIWSV2Passthrough(
 			"relay_dial_failed account_id=%d status_code=%d err=%s",
 			account.ID,