feat: 动态 block 索引分配、OAuth 自动启停联动与 Antigravity 签名注入增强

- Gemini/OpenAI normalizer 改为按实际出现顺序延迟分配 thinking/text block 索引
- OAuth 失效标记时自动停用 Key,清除/刷新成功时自动启用
- Antigravity thought signature 注入支持 tool-specific 签名优先,再回退 session 级签名
- Claude normalizer 对非字符串 thinking block 降级为 UnknownBlock
- Gemini normalizer 反序列化时为 Antigravity 目标补充 signature
- signature_cache _prune 增加超限驱逐
- model/provider 包 __init__ 改为惰性导入避免循环依赖
- build_antigravity_url 复制 query_params 防止修改调用方原始字典
- upstream_fetcher build_all_format_configs 兼容非 EndpointFetchConfig 对象
- Antigravity HTTP 状态判定扩展与 Gemini endpoint check 支持 v1internal
- 新增 thought signature 注入与 signature cache 驱逐测试

Closes #165

Co-authored-by: AAEE86 <ppk0227@hotmail.com>
This commit is contained in:
fawney19
2026-02-10 17:53:04 +08:00
parent 0bb17a1502
commit 8949ad6d9e
19 changed files with 467 additions and 87 deletions

View File

@@ -1518,12 +1518,13 @@ async function handleClearOAuthInvalid(key: EndpointAPIKey) {
clearingOAuthInvalidKeyId.value = key.id clearingOAuthInvalidKeyId.value = key.id
try { try {
await clearOAuthInvalid(key.id) await clearOAuthInvalid(key.id)
showSuccess('已清除 OAuth 异常标记') showSuccess('已清除 OAuth 异常标记Key 已自动启用')
// 更新本地数据 // 更新本地数据
const keyInList = providerKeys.value.find(k => k.id === key.id) const keyInList = providerKeys.value.find(k => k.id === key.id)
if (keyInList) { if (keyInList) {
keyInList.oauth_invalid_at = null keyInList.oauth_invalid_at = null
keyInList.oauth_invalid_reason = null keyInList.oauth_invalid_reason = null
keyInList.is_active = true
} }
await loadEndpoints() await loadEndpoints()
} catch (err: any) { } catch (err: any) {

View File

@@ -199,11 +199,12 @@ async def clear_oauth_invalid(
old_reason = key.oauth_invalid_reason old_reason = key.oauth_invalid_reason
key.oauth_invalid_at = None key.oauth_invalid_at = None
key.oauth_invalid_reason = None key.oauth_invalid_reason = None
key.is_active = True
db.commit() db.commit()
logger.info("[OK] 手动清除 Key {}... 的 OAuth 失效标记 (原因: {})", key_id[:8], old_reason) logger.info("[OK] 手动清除 Key {}... 的 OAuth 失效标记并自动启用 (原因: {})", key_id[:8], old_reason)
return {"message": "已清除 OAuth 失效标记"} return {"message": "已清除 OAuth 失效标记Key 已自动启用"}
# ========== Provider Keys API ========== # ========== Provider Keys API ==========
@@ -1575,8 +1576,9 @@ class AdminRefreshProviderQuotaAdapter(AdminApiAdapter):
if "401" in error_msg or "认证失败" in error_msg: if "401" in error_msg or "认证失败" in error_msg:
key.oauth_invalid_at = datetime.now(timezone.utc) key.oauth_invalid_at = datetime.now(timezone.utc)
key.oauth_invalid_reason = "Kiro Token 无效或已过期" key.oauth_invalid_reason = "Kiro Token 无效或已过期"
key.is_active = False
db.commit() db.commit()
logger.warning("[KIRO_QUOTA] Key {} Token 无效,已标记为异常", key.id) logger.warning("[KIRO_QUOTA] Key {} Token 无效,已标记为异常并自动停用", key.id)
return { return {
"key_id": key.id, "key_id": key.id,
"key_name": key.name, "key_name": key.name,

View File

@@ -690,8 +690,9 @@ async def refresh_oauth(
# 标记为失效 # 标记为失效
key.oauth_invalid_at = datetime.now(timezone.utc) key.oauth_invalid_at = datetime.now(timezone.utc)
key.oauth_invalid_reason = str(e) key.oauth_invalid_reason = str(e)
key.is_active = False
db.commit() db.commit()
logger.warning("Kiro Key {} token 刷新失败,已标记为失效: {}", key_id, e) logger.warning("Kiro Key {} token 刷新失败,已标记为失效并自动停用: {}", key_id, e)
raise InvalidRequestException("Kiro token refresh 失败,请检查凭据是否有效") raise InvalidRequestException("Kiro token refresh 失败,请检查凭据是否有效")
# 更新 key # 更新 key
@@ -699,6 +700,7 @@ async def refresh_oauth(
key.auth_config = crypto_service.encrypt(json.dumps(new_cfg.to_dict())) key.auth_config = crypto_service.encrypt(json.dumps(new_cfg.to_dict()))
key.oauth_invalid_at = None key.oauth_invalid_at = None
key.oauth_invalid_reason = None key.oauth_invalid_reason = None
key.is_active = True
db.commit() db.commit()
return CompleteOAuthResponse( return CompleteOAuthResponse(
@@ -787,8 +789,9 @@ async def refresh_oauth(
key.oauth_invalid_at = datetime.now(timezone.utc) key.oauth_invalid_at = datetime.now(timezone.utc)
key.oauth_invalid_reason = error_reason key.oauth_invalid_reason = error_reason
key.is_active = False
db.commit() db.commit()
logger.warning("Key {} OAuth token 刷新失败,已标记为失效: {}", key_id, error_reason) logger.warning("Key {} OAuth token 刷新失败,已标记为失效并自动停用: {}", key_id, error_reason)
raise InvalidRequestException(f"token refresh 失败: {error_reason}") raise InvalidRequestException(f"token refresh 失败: {error_reason}")
@@ -841,6 +844,7 @@ async def refresh_oauth(
if not is_account_level_block(getattr(key, "oauth_invalid_reason", None)): if not is_account_level_block(getattr(key, "oauth_invalid_reason", None)):
key.oauth_invalid_at = None key.oauth_invalid_at = None
key.oauth_invalid_reason = None key.oauth_invalid_reason = None
key.is_active = True
db.commit() db.commit()
return CompleteOAuthResponse( return CompleteOAuthResponse(

View File

@@ -886,6 +886,7 @@ async def test_model(
api_key.oauth_invalid_reason = ( api_key.oauth_invalid_reason = (
f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号" f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号"
) )
api_key.is_active = False
db.commit() db.commit()
oauth_email = None oauth_email = None
if getattr(api_key, "auth_config", None): if getattr(api_key, "auth_config", None):

View File

@@ -265,10 +265,10 @@ class GeminiChatAdapter(ChatAdapterBase):
provider_id: str | None = None, provider_id: str | None = None,
api_key_id: str | None = None, api_key_id: str | None = None,
model_name: str | None = None, model_name: str | None = None,
# Provider 上下文Gemini Chat 适配器忽略,仅保持签名兼容) # Provider 上下文
auth_type: str | None = None, # noqa: ARG003 auth_type: str | None = None,
provider_type: str | None = None, # noqa: ARG003 provider_type: str | None = None,
decrypted_auth_config: dict[str, Any] | None = None, # noqa: ARG003 decrypted_auth_config: dict[str, Any] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""测试 Gemini API 模型连接性(非流式)""" """测试 Gemini API 模型连接性(非流式)"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check from src.api.handlers.base.endpoint_checker import run_endpoint_check
@@ -283,18 +283,60 @@ class GeminiChatAdapter(ChatAdapterBase):
"status_code": 400, "status_code": 400,
} }
is_antigravity = provider_type and provider_type.lower() == "antigravity"
is_oauth = auth_type == "oauth"
# Antigravity provider 使用 v1internal 路径,而非标准 Gemini API 路径
if is_antigravity:
from src.services.provider.adapters.antigravity.constants import (
V1INTERNAL_PATH_TEMPLATE,
get_http_user_agent as _get_antigravity_ua,
)
from src.services.provider.adapters.antigravity.url_availability import url_availability
ordered_urls = url_availability.get_ordered_urls(prefer_daily=True)
ag_base = ordered_urls[0] if ordered_urls else base_url
path = V1INTERNAL_PATH_TEMPLATE.format(action="generateContent")
url = f"{str(ag_base).rstrip('/')}{path}"
else:
# 使用基类配置方法但重写URL构建逻辑 # 使用基类配置方法但重写URL构建逻辑
base_url_resolved = cls.build_endpoint_url(base_url) base_url_resolved = cls.build_endpoint_url(base_url)
url = f"{base_url_resolved}/models/{effective_model_name}:generateContent" url = f"{base_url_resolved}/models/{effective_model_name}:generateContent"
# 构建请求组件 # 构建请求组件
headers = cls.build_headers_with_extra(api_key, extra_headers) # Antigravity 需要特定的 User-Agent
merged_extra = dict(extra_headers) if extra_headers else {}
if is_antigravity:
merged_extra["User-Agent"] = _get_antigravity_ua()
headers = cls.build_headers_with_extra(api_key, merged_extra if merged_extra else None)
# OAuth 统一处理替换端点默认认证头x-goog-api-key为 Authorization: Bearer
if is_oauth:
from src.core.api_format import get_auth_config_for_endpoint
default_auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
if default_auth_header.lower() != "authorization":
headers.pop(default_auth_header, None)
headers["Authorization"] = f"Bearer {api_key}"
body = cls.build_request_body(request_data) body = cls.build_request_body(request_data)
# 应用请求体规则(在格式转换后应用,确保规则效果不被覆盖) # 应用请求体规则(在格式转换后应用,确保规则效果不被覆盖)
if body_rules: if body_rules:
body = apply_body_rules(body, body_rules) body = apply_body_rules(body, body_rules)
# Antigravity 需要将请求体包装为 v1internal 信封格式
if is_antigravity:
from src.services.provider.adapters.antigravity.envelope import wrap_v1internal_request
project_id = (decrypted_auth_config or {}).get("project_id", "")
body = wrap_v1internal_request(
body,
project_id=project_id,
model=effective_model_name,
request_type="endpoint_test",
)
# 应用请求头规则(在请求头构建后应用) # 应用请求头规则(在请求头构建后应用)
if header_rules: if header_rules:
# 获取认证头名称,防止被规则覆盖 # 获取认证头名称,防止被规则覆盖

View File

@@ -720,15 +720,21 @@ class ClaudeNormalizer(FormatNormalizer):
btype = str(block.get("type") or "unknown") btype = str(block.get("type") or "unknown")
if btype == "thinking": if btype == "thinking":
thinking = str(block.get("thinking") or "") thinking_raw = block.get("thinking")
signature = block.get("signature") signature = block.get("signature")
if thinking:
if isinstance(thinking_raw, str) and thinking_raw:
blocks.append( blocks.append(
ThinkingBlock( ThinkingBlock(
thinking=thinking, thinking=thinking_raw,
signature=str(signature) if signature else None, signature=str(signature) if signature else None,
) )
) )
else:
dropped_key = f"claude_block:{btype}"
dropped[dropped_key] = dropped.get(dropped_key, 0) + 1
blocks.append(UnknownBlock(raw_type=btype, payload=block))
continue continue
if btype == "text": if btype == "text":

View File

@@ -565,12 +565,27 @@ class GeminiNormalizer(FormatNormalizer):
state.message_id = state.message_id or "gemini" state.message_id = state.message_id or "gemini"
ss["message_started"] = True ss["message_started"] = True
ss.setdefault("text_block_started", False) ss.setdefault("text_block_started", False)
ss.setdefault("text_block_stopped", False)
ss.setdefault("thinking_block_started", False) ss.setdefault("thinking_block_started", False)
ss.setdefault("thinking_block_stopped", False)
# 根据首先出现的内容类型延迟分配块索引。
# 当不存在思考内容时golden tests 期望文本位于索引0。
ss.setdefault("text_block_index", None)
ss.setdefault("thinking_block_index", None)
ss.setdefault("accumulated_text", "") ss.setdefault("accumulated_text", "")
ss.setdefault("accumulated_thinking", "") ss.setdefault("accumulated_thinking", "")
ss.setdefault("next_block_index", 2) # 0 预留给 thinking, 1 预留给 text ss.setdefault("next_block_index", 0)
events.append(MessageStartEvent(message_id=state.message_id, model=model)) events.append(MessageStartEvent(message_id=state.message_id, model=model))
def _reserve_block_index(slot_key: str) -> int:
idx = ss.get(slot_key)
if isinstance(idx, int) and idx >= 0:
return idx
next_idx = int(ss.get("next_block_index") or 0)
ss[slot_key] = next_idx
ss["next_block_index"] = next_idx + 1
return next_idx
candidates = chunk.get("candidates") or [] candidates = chunk.get("candidates") or []
if not isinstance(candidates, list) or not candidates: if not isinstance(candidates, list) or not candidates:
return events return events
@@ -595,7 +610,7 @@ class GeminiNormalizer(FormatNormalizer):
is_thought = part.get("thought") is True is_thought = part.get("thought") is True
if is_thought: if is_thought:
# Thinking content → block_index=0 # Thinking content
prev = str(ss.get("accumulated_thinking") or "") prev = str(ss.get("accumulated_thinking") or "")
if text.startswith(prev): if text.startswith(prev):
delta = text[len(prev) :] delta = text[len(prev) :]
@@ -609,12 +624,18 @@ class GeminiNormalizer(FormatNormalizer):
ss["thinking_block_started"] = True ss["thinking_block_started"] = True
events.append( events.append(
ContentBlockStartEvent( ContentBlockStartEvent(
block_index=0, block_type=ContentType.THINKING block_index=_reserve_block_index("thinking_block_index"),
block_type=ContentType.THINKING,
)
)
events.append(
ContentDeltaEvent(
block_index=_reserve_block_index("thinking_block_index"),
text_delta=delta,
) )
) )
events.append(ContentDeltaEvent(block_index=0, text_delta=delta))
else: else:
# Regular text → block_index=1 # Regular text
prev = str(ss.get("accumulated_text") or "") prev = str(ss.get("accumulated_text") or "")
if text.startswith(prev): if text.startswith(prev):
delta = text[len(prev) :] delta = text[len(prev) :]
@@ -624,29 +645,38 @@ class GeminiNormalizer(FormatNormalizer):
ss["accumulated_text"] = prev + delta ss["accumulated_text"] = prev + delta
if delta: if delta:
# 切换:先关闭 thinking blockClaude 协议要求顺序 stop/start # Transition: stop thinking block before text (Claude requires stop/start ordering).
if ss.get("thinking_block_started") and not ss.get( if ss.get("thinking_block_started") and not ss.get(
"thinking_block_stopped" "thinking_block_stopped"
): ):
ss["thinking_block_stopped"] = True ss["thinking_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=0)) events.append(
ContentBlockStopEvent(
block_index=_reserve_block_index("thinking_block_index")
)
)
if not ss.get("text_block_started"): if not ss.get("text_block_started"):
ss["text_block_started"] = True ss["text_block_started"] = True
events.append( events.append(
ContentBlockStartEvent( ContentBlockStartEvent(
block_index=1, block_type=ContentType.TEXT block_index=_reserve_block_index("text_block_index"),
block_type=ContentType.TEXT,
)
)
events.append(
ContentDeltaEvent(
block_index=_reserve_block_index("text_block_index"),
text_delta=delta,
) )
) )
events.append(ContentDeltaEvent(block_index=1, text_delta=delta))
# 提取 thoughtSignature对齐 AM缓存到 session
sig = part.get("thoughtSignature") or part.get("thought_signature") sig = part.get("thoughtSignature") or part.get("thought_signature")
if isinstance(sig, str) and sig and ss.get("thinking_block_started"): if isinstance(sig, str) and sig and ss.get("thinking_block_started"):
# 仅在 thinking block 已开启时发射 signature delta # 仅在 thinking block 已开启时发射 signature delta
events.append( events.append(
ContentDeltaEvent( ContentDeltaEvent(
block_index=0, block_index=_reserve_block_index("thinking_block_index"),
text_delta="", text_delta="",
extra={"thought_signature": sig}, extra={"thought_signature": sig},
) )
@@ -662,10 +692,10 @@ class GeminiNormalizer(FormatNormalizer):
# 关闭前面的 thinking/text block如果还开着 # 关闭前面的 thinking/text block如果还开着
if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"): if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"):
ss["thinking_block_stopped"] = True ss["thinking_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=0)) events.append(ContentBlockStopEvent(block_index=_reserve_block_index("thinking_block_index")))
if ss.get("text_block_started") and not ss.get("text_block_stopped"): if ss.get("text_block_started") and not ss.get("text_block_stopped"):
ss["text_block_stopped"] = True ss["text_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=1)) events.append(ContentBlockStopEvent(block_index=_reserve_block_index("text_block_index")))
name = str(func_call.get("name") or "") name = str(func_call.get("name") or "")
args = func_call.get("args") args = func_call.get("args")
@@ -676,7 +706,7 @@ class GeminiNormalizer(FormatNormalizer):
fc_id = func_call.get("id") fc_id = func_call.get("id")
tool_id = fc_id if isinstance(fc_id, str) and fc_id else None tool_id = fc_id if isinstance(fc_id, str) and fc_id else None
block_index = int(ss.get("next_block_index") or 2) block_index = int(ss.get("next_block_index") or 0)
ss["next_block_index"] = block_index + 1 ss["next_block_index"] = block_index + 1
events.append( events.append(
@@ -711,7 +741,7 @@ class GeminiNormalizer(FormatNormalizer):
# 确保 mime_type 和 data 都非空 # 确保 mime_type 和 data 都非空
if mime_type and data and len(data) > 10: # base64 图片数据至少几十个字符 if mime_type and data and len(data) > 10: # base64 图片数据至少几十个字符
block_index = int(ss.get("next_block_index") or 1) block_index = int(ss.get("next_block_index") or 0)
ss["next_block_index"] = block_index + 1 ss["next_block_index"] = block_index + 1
# 使用 ContentBlockStartEvent 传递图片数据 # 使用 ContentBlockStartEvent 传递图片数据
@@ -736,10 +766,10 @@ class GeminiNormalizer(FormatNormalizer):
# 先补齐 content_block_stop所有已开启的 block再发送 MessageStop # 先补齐 content_block_stop所有已开启的 block再发送 MessageStop
if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"): if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"):
ss["thinking_block_stopped"] = True ss["thinking_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=0)) events.append(ContentBlockStopEvent(block_index=_reserve_block_index("thinking_block_index")))
if ss.get("text_block_started") and not ss.get("text_block_stopped"): if ss.get("text_block_started") and not ss.get("text_block_stopped"):
ss["text_block_stopped"] = True ss["text_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=1)) events.append(ContentBlockStopEvent(block_index=_reserve_block_index("text_block_index")))
events.append(MessageStopEvent(stop_reason=stop_reason, usage=usage_info)) events.append(MessageStopEvent(stop_reason=stop_reason, usage=usage_info))
if "error" in chunk: if "error" in chunk:
@@ -1377,9 +1407,41 @@ class GeminiNormalizer(FormatNormalizer):
if isinstance(b, ThinkingBlock): if isinstance(b, ThinkingBlock):
if b.thinking: if b.thinking:
signature: str | None = b.signature or None
if signature is None and target_variant == "antigravity":
model_str = str(model or "")
try:
from src.services.provider.adapters.antigravity.constants import (
DUMMY_THOUGHT_SIGNATURE,
)
from src.services.provider.adapters.antigravity.signature_cache import (
signature_cache,
)
cached_or_dummy = signature_cache.get_or_dummy(model_str, b.thinking)
# Prefer cached real signature > dummy signature.
if (
isinstance(cached_or_dummy, str)
and cached_or_dummy
and cached_or_dummy != DUMMY_THOUGHT_SIGNATURE
):
signature = cached_or_dummy
elif isinstance(cached_or_dummy, str) and cached_or_dummy:
signature = cached_or_dummy
except Exception:
# Best-effort fallback: Gemini models can accept a dummy signature.
if model_str.startswith("gemini-"):
signature = "skip_thought_signature_validator"
# For Antigravity, missing signature is likely to fail upstream validation.
if target_variant == "antigravity" and not signature:
continue
thought_part: dict[str, Any] = {"text": b.thinking, "thought": True} thought_part: dict[str, Any] = {"text": b.thinking, "thought": True}
if b.signature: if signature:
thought_part["thoughtSignature"] = b.signature thought_part["thoughtSignature"] = signature
parts.append(thought_part) parts.append(thought_part)
continue continue

View File

@@ -402,11 +402,26 @@ class OpenAINormalizer(FormatNormalizer):
state.model = model state.model = model
ss["message_started"] = True ss["message_started"] = True
ss.setdefault("thinking_block_started", False) ss.setdefault("thinking_block_started", False)
ss.setdefault("thinking_block_stopped", False)
ss.setdefault("text_block_started", False) ss.setdefault("text_block_started", False)
ss.setdefault("text_block_stopped", False)
# 根据首先出现的内容类型延迟分配块索引。
# 这样可以避免预留空白当不存在思考内容时测试期望文本位于索引0。
ss.setdefault("thinking_block_index", None)
ss.setdefault("text_block_index", None)
ss.setdefault("tool_id_to_block_index", {}) ss.setdefault("tool_id_to_block_index", {})
ss.setdefault("next_block_index", 2) # 0=thinking, 1=text, 2+=tools ss.setdefault("next_block_index", 0)
events.append(MessageStartEvent(message_id=msg_id, model=model)) events.append(MessageStartEvent(message_id=msg_id, model=model))
def _reserve_block_index(slot_key: str) -> int:
idx = ss.get(slot_key)
if isinstance(idx, int) and idx >= 0:
return idx
next_idx = int(ss.get("next_block_index") or 0)
ss[slot_key] = next_idx
ss["next_block_index"] = next_idx + 1
return next_idx
choices = chunk.get("choices") or [] choices = chunk.get("choices") or []
if not choices or not isinstance(choices, list): if not choices or not isinstance(choices, list):
# OpenAI streaming may send a final "usage-only" chunk when # OpenAI streaming may send a final "usage-only" chunk when
@@ -432,24 +447,44 @@ class OpenAINormalizer(FormatNormalizer):
# reasoning_content delta (thinking) # reasoning_content delta (thinking)
reasoning_delta = delta.get("reasoning_content") reasoning_delta = delta.get("reasoning_content")
if isinstance(reasoning_delta, str) and reasoning_delta: if isinstance(reasoning_delta, str) and reasoning_delta:
thinking_index = _reserve_block_index("thinking_block_index")
if not ss.get("thinking_block_started"): if not ss.get("thinking_block_started"):
ss["thinking_block_started"] = True ss["thinking_block_started"] = True
events.append( events.append(
ContentBlockStartEvent(block_index=0, block_type=ContentType.THINKING) ContentBlockStartEvent(
block_index=thinking_index,
block_type=ContentType.THINKING,
) )
events.append(ContentDeltaEvent(block_index=0, text_delta=reasoning_delta)) )
events.append(ContentDeltaEvent(block_index=thinking_index, text_delta=reasoning_delta))
# content delta # content delta
content_delta = delta.get("content") content_delta = delta.get("content")
if isinstance(content_delta, str) and content_delta: if isinstance(content_delta, str) and content_delta:
# thinking 过渡到 text 时,先关闭 thinking block # From thinking -> text, close thinking block first (only when text hasn't started yet).
if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"): if (
ss.get("thinking_block_started")
and not ss.get("thinking_block_stopped")
and not ss.get("text_block_started")
):
ss["thinking_block_stopped"] = True ss["thinking_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=0)) events.append(
ContentBlockStopEvent(block_index=_reserve_block_index("thinking_block_index"))
)
if not ss.get("text_block_started"): if not ss.get("text_block_started"):
ss["text_block_started"] = True ss["text_block_started"] = True
events.append(ContentBlockStartEvent(block_index=1, block_type=ContentType.TEXT)) events.append(
events.append(ContentDeltaEvent(block_index=1, text_delta=content_delta)) ContentBlockStartEvent(
block_index=_reserve_block_index("text_block_index"),
block_type=ContentType.TEXT,
)
)
events.append(
ContentDeltaEvent(
block_index=_reserve_block_index("text_block_index"),
text_delta=content_delta,
)
)
# tool_calls delta # tool_calls delta
tool_calls = delta.get("tool_calls") tool_calls = delta.get("tool_calls")
@@ -497,10 +532,14 @@ class OpenAINormalizer(FormatNormalizer):
# 先补齐 content_block_stopthinking + text再发送 MessageStop # 先补齐 content_block_stopthinking + text再发送 MessageStop
if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"): if ss.get("thinking_block_started") and not ss.get("thinking_block_stopped"):
ss["thinking_block_stopped"] = True ss["thinking_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=0)) events.append(
ContentBlockStopEvent(block_index=_reserve_block_index("thinking_block_index"))
)
if ss.get("text_block_started") and not ss.get("text_block_stopped"): if ss.get("text_block_started") and not ss.get("text_block_stopped"):
ss["text_block_stopped"] = True ss["text_block_stopped"] = True
events.append(ContentBlockStopEvent(block_index=1)) events.append(
ContentBlockStopEvent(block_index=_reserve_block_index("text_block_index"))
)
# 解析 usage需要请求时设置 stream_options.include_usage: true # 解析 usage需要请求时设置 stream_options.include_usage: true
usage_info = self._openai_usage_to_internal(chunk.get("usage")) usage_info = self._openai_usage_to_internal(chunk.get("usage"))
events.append(MessageStopEvent(stop_reason=stop_reason, usage=usage_info)) events.append(MessageStopEvent(stop_reason=stop_reason, usage=usage_info))
@@ -1504,7 +1543,7 @@ class OpenAINormalizer(FormatNormalizer):
if tool_key in mapping: if tool_key in mapping:
return int(mapping[tool_key]) return int(mapping[tool_key])
next_idx = int(ss.get("next_block_index") or 1) next_idx = int(ss.get("next_block_index") or 0)
mapping[tool_key] = next_idx mapping[tool_key] = next_idx
ss["next_block_index"] = next_idx + 1 ss["next_block_index"] = next_idx + 1
return next_idx return next_idx

View File

@@ -4,6 +4,11 @@
包含模型管理、成本计算等功能。 包含模型管理、成本计算等功能。
""" """
from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from src.services.model.availability import ModelAvailabilityQuery from src.services.model.availability import ModelAvailabilityQuery
from src.services.model.cost import ModelCostService from src.services.model.cost import ModelCostService
from src.services.model.fetch_scheduler import ModelFetchScheduler, get_model_fetch_scheduler from src.services.model.fetch_scheduler import ModelFetchScheduler, get_model_fetch_scheduler
@@ -18,3 +23,46 @@ __all__ = [
"ModelFetchScheduler", "ModelFetchScheduler",
"get_model_fetch_scheduler", "get_model_fetch_scheduler",
] ]
def __getattr__(name: str) -> Any:
"""Lazy attribute access to avoid import-time side effects.
Importing `src.services.model` should not eagerly import the whole model
service stack (scheduler/services), which can create circular imports during
test collection.
"""
if name == "ModelAvailabilityQuery":
from src.services.model.availability import ModelAvailabilityQuery as _ModelAvailabilityQuery
return _ModelAvailabilityQuery
if name == "ModelCostService":
from src.services.model.cost import ModelCostService as _ModelCostService
return _ModelCostService
if name == "ModelFetchScheduler":
from src.services.model.fetch_scheduler import ModelFetchScheduler as _ModelFetchScheduler
return _ModelFetchScheduler
if name == "get_model_fetch_scheduler":
from src.services.model.fetch_scheduler import (
get_model_fetch_scheduler as _get_model_fetch_scheduler,
)
return _get_model_fetch_scheduler
if name == "GlobalModelService":
from src.services.model.global_model import GlobalModelService as _GlobalModelService
return _GlobalModelService
if name == "ModelService":
from src.services.model.service import ModelService as _ModelService
return _ModelService
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

View File

@@ -202,12 +202,26 @@ def build_all_format_configs(
fmt = next((f for f in candidates if f in format_to_endpoint), None) fmt = next((f for f in candidates if f in format_to_endpoint), None)
if fmt is not None: if fmt is not None:
cfg = format_to_endpoint[fmt] cfg = format_to_endpoint[fmt]
base_url = str(getattr(cfg, "base_url", "") or "")
extra_headers: dict[str, str] | None
if isinstance(cfg, EndpointFetchConfig):
extra_headers = cfg.extra_headers
else:
# 允许直接传递类似 ProviderEndpoint 的对象(测试/独立使用场景)。
candidate_extra = getattr(cfg, "extra_headers", None)
if isinstance(candidate_extra, dict):
extra_headers = {str(k): str(v) for k, v in candidate_extra.items() if k}
else:
extra_headers = get_extra_headers_from_endpoint(cfg)
configs.append( configs.append(
{ {
"api_key": api_key_value, "api_key": api_key_value,
"base_url": cfg.base_url, "base_url": base_url,
"api_format": fmt, "api_format": fmt,
"extra_headers": cfg.extra_headers, "extra_headers": extra_headers,
} }
) )
return configs return configs

View File

@@ -747,9 +747,10 @@ class ErrorClassifier:
key.oauth_invalid_at = datetime.now(timezone.utc) key.oauth_invalid_at = datetime.now(timezone.utc)
key.oauth_invalid_reason = f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号" key.oauth_invalid_reason = f"{OAUTH_ACCOUNT_BLOCK_PREFIX}Google 要求验证账号"
key.is_active = False
self.db.commit() self.db.commit()
logger.warning( logger.warning(
" [{}] {} 因 403 VALIDATION_REQUIRED 已标记为账号异常", " [{}] {} 因 403 VALIDATION_REQUIRED 已标记为账号异常并自动停用",
request_id, request_id,
self._format_key_display(key), self._format_key_display(key),
) )

View File

@@ -4,6 +4,11 @@ Provider 服务模块
包含 Provider 管理、格式处理、传输层等功能。 包含 Provider 管理、格式处理、传输层等功能。
""" """
from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from src.services.provider.format import normalize_endpoint_signature from src.services.provider.format import normalize_endpoint_signature
from src.services.provider.service import ProviderService from src.services.provider.service import ProviderService
from src.services.provider.transport import build_provider_url from src.services.provider.transport import build_provider_url
@@ -13,3 +18,28 @@ __all__ = [
"normalize_endpoint_signature", "normalize_endpoint_signature",
"build_provider_url", "build_provider_url",
] ]
def __getattr__(name: str) -> Any:
"""Lazy attribute access to avoid import-time side effects.
Importing `src.services.provider` should not eagerly import the whole provider
service stack (which can create circular imports during test collection).
"""
if name == "ProviderService":
from src.services.provider.service import ProviderService as _ProviderService
return _ProviderService
if name == "normalize_endpoint_signature":
from src.services.provider.format import (
normalize_endpoint_signature as _normalize_endpoint_signature,
)
return _normalize_endpoint_signature
if name == "build_provider_url":
from src.services.provider.transport import build_provider_url as _build_provider_url
return _build_provider_url
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

View File

@@ -53,10 +53,14 @@ def get_http_user_agent() -> str:
def update_user_agent_version(version: str) -> None: def update_user_agent_version(version: str) -> None:
"""更新 User-Agent 中的版本号(由 refresh_user_agent 调用)。""" """更新 User-Agent 中的版本号(由 refresh_user_agent 调用)。"""
global _ua_version # noqa: PLW0603 global HTTP_USER_AGENT, _ua_version # noqa: PLW0603
version = str(version or "").strip()
if not version:
return
with _ua_lock: with _ua_lock:
_ua_version = version _ua_version = version
# Backward compat: keep module-level constant in sync.
HTTP_USER_AGENT = f"antigravity/{_ua_version} {_PLATFORM_TAG}"
def parse_version_string(text: str) -> str | None: def parse_version_string(text: str) -> str | None:
"""从任意文本中提取 X.Y.Z 格式的版本号。""" """从任意文本中提取 X.Y.Z 格式的版本号。"""

View File

@@ -197,7 +197,7 @@ def _inject_claude_tool_ids_request(inner_request: dict[str, Any], model: str) -
name = "unknown" name = "unknown"
fc_id = fc.get("id") fc_id = fc.get("id")
if fc_id is None: if not (isinstance(fc_id, str) and fc_id):
# 生成新 ID # 生成新 ID
count = name_counters.get(name, 0) count = name_counters.get(name, 0)
fc_id = f"call_{name}_{count}" fc_id = f"call_{name}_{count}"
@@ -231,7 +231,7 @@ def _inject_claude_tool_ids_request(inner_request: dict[str, Any], model: str) -
continue continue
fr = part.get("functionResponse") or part.get("function_response") fr = part.get("functionResponse") or part.get("function_response")
if isinstance(fr, dict) and fr.get("id") is None: if isinstance(fr, dict) and not (isinstance(fr.get("id"), str) and fr.get("id")):
name = fr.get("name", "") name = fr.get("name", "")
if not isinstance(name, str) or not name: if not isinstance(name, str) or not name:
name = "unknown" name = "unknown"
@@ -514,22 +514,20 @@ def _inject_google_search_tool(inner_request: dict[str, Any]) -> None:
def _inject_thought_signatures(inner_request: dict[str, Any], session_id: str | None) -> None: def _inject_thought_signatures(inner_request: dict[str, Any], session_id: str | None) -> None:
"""为 functionCall parts 注入 thoughtSignature(从 session signature cache """Inject thought signatures into functionCall parts.
对齐 AM wrapper.rs当 functionCall part 缺少 thoughtSignature 时, Prefer tool-specific signatures (tool_use_id -> thoughtSignature), then fall back
session cache 中恢复签名,确保 thinking 模型的多轮 tool call 连续性。 to a session-level signature when available.
""" """
if not session_id:
return
try: try:
from src.services.provider.adapters.antigravity.signature_cache import signature_cache from src.services.provider.adapters.antigravity.signature_cache import signature_cache
except Exception: except Exception:
return return
cached_sig = signature_cache.get_session_signature(session_id) session_sig: str | None = None
if not cached_sig: if session_id:
return session_sig = signature_cache.get_session_signature(session_id)
contents = inner_request.get("contents") contents = inner_request.get("contents")
if not isinstance(contents, list): if not isinstance(contents, list):
@@ -545,9 +543,48 @@ def _inject_thought_signatures(inner_request: dict[str, Any], session_id: str |
for part in parts: for part in parts:
if not isinstance(part, dict): if not isinstance(part, dict):
continue continue
# 只处理有 functionCall 且缺少 thoughtSignature 的 part
if "functionCall" in part and part.get("thoughtSignature") is None: fc = part.get("functionCall") or part.get("function_call")
part["thoughtSignature"] = cached_sig if not isinstance(fc, dict):
continue
# Normalize existing signature aliases to thoughtSignature (if present).
sig_val: str | None = None
ts = part.get("thoughtSignature")
if isinstance(ts, str) and ts.strip():
sig_val = ts.strip()
if ts != sig_val:
part["thoughtSignature"] = sig_val
else:
tss = part.get("thought_signature")
if isinstance(tss, str) and tss.strip():
sig_val = tss.strip()
part["thoughtSignature"] = sig_val
part.pop("thought_signature", None)
else:
legacy = part.get("signature")
if isinstance(legacy, str) and legacy.strip():
sig_val = legacy.strip()
part["thoughtSignature"] = sig_val
part.pop("signature", None)
if sig_val:
continue
tool_id = fc.get("id")
tool_sig: str | None = None
if isinstance(tool_id, str) and tool_id:
tool_sig = signature_cache.get_tool_signature(tool_id)
chosen = tool_sig or session_sig
if not chosen:
continue
part["thoughtSignature"] = chosen
# Avoid sending duplicate aliases once we inject.
part.pop("thought_signature", None)
if part.get("signature") in (None, ""):
part.pop("signature", None)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -782,8 +819,8 @@ def _inject_claude_tool_ids_response(response: dict[str, Any], model: str) -> No
for part in parts: for part in parts:
if not isinstance(part, dict): if not isinstance(part, dict):
continue continue
fc = part.get("functionCall") fc = part.get("functionCall") or part.get("function_call")
if isinstance(fc, dict) and fc.get("id") is None: if isinstance(fc, dict) and not (isinstance(fc.get("id"), str) and fc.get("id")):
name = fc.get("name", "unknown") name = fc.get("name", "unknown")
if not isinstance(name, str): if not isinstance(name, str):
name = "unknown" name = "unknown"
@@ -976,7 +1013,7 @@ def cache_thought_signatures(model: str, response: dict[str, Any]) -> None:
signature_cache.cache(model, text, sig) signature_cache.cache(model, text, sig)
# 缓存 tool call signatureLayer 1 # 缓存 tool call signatureLayer 1
fc = part.get("functionCall") fc = part.get("functionCall") or part.get("function_call")
if isinstance(fc, dict) and isinstance(sig, str) and sig: if isinstance(fc, dict) and isinstance(sig, str) and sig:
tool_id = fc.get("id") tool_id = fc.get("id")
if isinstance(tool_id, str) and tool_id: if isinstance(tool_id, str) and tool_id:
@@ -1070,9 +1107,9 @@ class AntigravityV1InternalEnvelope:
def on_http_status(self, *, base_url: str | None, status_code: int) -> None: def on_http_status(self, *, base_url: str | None, status_code: int) -> None:
if not base_url: if not base_url:
return return
if status_code == 200: if 200 <= status_code < 300:
url_availability.mark_success(base_url) url_availability.mark_success(base_url)
elif status_code in (429, 500, 502, 503, 504): elif status_code in (404, 408, 429) or 500 <= status_code < 600:
url_availability.mark_unavailable(base_url) url_availability.mark_unavailable(base_url)
def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None: # noqa: ARG002 def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None: # noqa: ARG002

View File

@@ -46,16 +46,18 @@ def build_antigravity_url(
action = "streamGenerateContent" if is_stream else "generateContent" action = "streamGenerateContent" if is_stream else "generateContent"
path = V1INTERNAL_PATH_TEMPLATE.format(action=action) path = V1INTERNAL_PATH_TEMPLATE.format(action=action)
query_params = dict(effective_query_params)
# v1internal 流式请求同样支持 ?alt=sse # v1internal 流式请求同样支持 ?alt=sse
if is_stream: if is_stream:
effective_query_params.setdefault("alt", "sse") query_params.setdefault("alt", "sse")
# 移除 v1internal 不支持的查询参数 # 移除 v1internal 不支持的查询参数
effective_query_params.pop("beta", None) query_params.pop("beta", None)
url = f"{str(base_url).rstrip('/')}{path}" url = f"{str(base_url).rstrip('/')}{path}"
if effective_query_params: if query_params:
query_string = urlencode(effective_query_params, doseq=True) query_string = urlencode(query_params, doseq=True)
if query_string: if query_string:
url = f"{url}?{query_string}" url = f"{url}?{query_string}"

View File

@@ -85,7 +85,7 @@ class ThinkingSignatureCache:
with self._lock: with self._lock:
self._tool_sigs[tool_use_id] = _CacheEntry(signature) self._tool_sigs[tool_use_id] = _CacheEntry(signature)
if len(self._tool_sigs) > _TOOL_CACHE_LIMIT: if len(self._tool_sigs) > _TOOL_CACHE_LIMIT:
self._prune(self._tool_sigs) self._prune(self._tool_sigs, limit=_TOOL_CACHE_LIMIT)
def get_tool_signature(self, tool_use_id: str) -> str | None: def get_tool_signature(self, tool_use_id: str) -> str | None:
"""查找工具调用对应的 signature。""" """查找工具调用对应的 signature。"""
@@ -107,7 +107,7 @@ class ThinkingSignatureCache:
with self._lock: with self._lock:
self._families[signature] = _CacheEntry(family) self._families[signature] = _CacheEntry(family)
if len(self._families) > _FAMILY_CACHE_LIMIT: if len(self._families) > _FAMILY_CACHE_LIMIT:
self._prune(self._families) self._prune(self._families, limit=_FAMILY_CACHE_LIMIT)
def get_signature_family(self, signature: str) -> str | None: def get_signature_family(self, signature: str) -> str | None:
"""查找 signature 所属的模型家族。""" """查找 signature 所属的模型家族。"""
@@ -150,7 +150,7 @@ class ThinkingSignatureCache:
if should_store: if should_store:
self._sessions[session_id] = _CacheEntry(_SessionEntry(signature, message_count)) self._sessions[session_id] = _CacheEntry(_SessionEntry(signature, message_count))
if len(self._sessions) > _SESSION_CACHE_LIMIT: if len(self._sessions) > _SESSION_CACHE_LIMIT:
self._prune(self._sessions) self._prune(self._sessions, limit=_SESSION_CACHE_LIMIT)
def get_session_signature(self, session_id: str) -> str | None: def get_session_signature(self, session_id: str) -> str | None:
"""获取会话的最新 thinking signature。""" """获取会话的最新 thinking signature。"""
@@ -210,13 +210,21 @@ class ThinkingSignatureCache:
return hashlib.sha256(content.encode("utf-8")).hexdigest()[:32] return hashlib.sha256(content.encode("utf-8")).hexdigest()[:32]
@staticmethod @staticmethod
def _prune(d: dict[str, _CacheEntry]) -> None: def _prune(d: dict[str, _CacheEntry], *, limit: int | None = None) -> None:
"""清理已过期的缓存条目。""" """Remove expired entries and optionally enforce a size limit."""
now = time.monotonic() now = time.monotonic()
expired = [k for k, v in d.items() if v.is_expired(now)] expired = [k for k, v in d.items() if v.is_expired(now)]
for k in expired: for k in expired:
d.pop(k, None) d.pop(k, None)
if limit is None or len(d) <= limit:
return
# Evict oldest entries by created_at.
excess = len(d) - limit
for k, _entry in sorted(d.items(), key=lambda kv: kv[1].created_at)[:excess]:
d.pop(k, None)
def clear(self) -> None: def clear(self) -> None:
"""清空所有缓存层(用于测试或手动重置)。""" """清空所有缓存层(用于测试或手动重置)。"""
with self._lock: with self._lock:

View File

@@ -103,6 +103,7 @@ async def resolve_oauth_access_token(
if not is_account_level_block(getattr(row, "oauth_invalid_reason", None)): if not is_account_level_block(getattr(row, "oauth_invalid_reason", None)):
row.oauth_invalid_at = None row.oauth_invalid_at = None
row.oauth_invalid_reason = None row.oauth_invalid_reason = None
row.is_active = True
db.commit() db.commit()
except Exception as e: except Exception as e:
# Don't fail caller path; token is still usable for this request. # Don't fail caller path; token is still usable for this request.

View File

@@ -204,3 +204,63 @@ async def test_antigravity_forces_conversion_path_in_stream_with_prefetch() -> N
assert mock_convert.call_count >= 1 assert mock_convert.call_count >= 1
assert any(b"data: {}" in c for c in out) assert any(b"data: {}" in c for c in out)
def test_wrap_v1internal_request_injects_thought_signature_from_tool_cache() -> None:
from src.services.provider.adapters.antigravity.signature_cache import signature_cache
signature_cache.clear()
tool_id = "toolu_123"
sig = "a" * 60
signature_cache.cache_tool_signature(tool_id, sig)
gemini_request = {
"model": "gemini-2.0-flash",
"contents": [
{"role": "user", "parts": [{"text": "hi"}]},
{"role": "model", "parts": [{"function_call": {"name": "do", "id": tool_id}}]},
],
}
wrapped = wrap_v1internal_request(
gemini_request,
project_id="project-123",
model="gemini-2.0-flash",
)
model_turn = wrapped["request"]["contents"][1]
part = model_turn["parts"][0]
assert part["thoughtSignature"] == sig
def test_wrap_v1internal_request_injects_session_signature_when_tool_cache_missing() -> None:
from src.services.provider.adapters.antigravity.signature_cache import signature_cache
signature_cache.clear()
session_id = "sid-123"
sig = "b" * 60
signature_cache.cache_session_signature(session_id, sig, message_count=2)
gemini_request = {
"model": "gemini-2.0-flash",
"sessionId": session_id,
"contents": [
{"role": "user", "parts": [{"text": "hi"}]},
{
"role": "model",
"parts": [{"functionCall": {"name": "do", "id": "toolu_missing"}}],
},
],
}
wrapped = wrap_v1internal_request(
gemini_request,
project_id="project-123",
model="gemini-2.0-flash",
)
model_turn = wrapped["request"]["contents"][1]
part = model_turn["parts"][0]
assert part["thoughtSignature"] == sig

View File

@@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
from typing import Any
from src.services.provider.adapters.antigravity.constants import DUMMY_THOUGHT_SIGNATURE from src.services.provider.adapters.antigravity.constants import DUMMY_THOUGHT_SIGNATURE
from src.services.provider.adapters.antigravity.signature_cache import ThinkingSignatureCache from src.services.provider.adapters.antigravity.signature_cache import ThinkingSignatureCache
@@ -51,6 +53,22 @@ def test_tool_signature_short_ignored() -> None:
assert cache.get_tool_signature("toolu_123") is None assert cache.get_tool_signature("toolu_123") is None
def test_tool_signature_cache_enforces_limit(monkeypatch: Any) -> None:
import src.services.provider.adapters.antigravity.signature_cache as sc_mod
# Use a small limit to make eviction deterministic in tests.
monkeypatch.setattr(sc_mod, "_TOOL_CACHE_LIMIT", 3)
cache = sc_mod.ThinkingSignatureCache()
cache.cache_tool_signature("toolu_1", _SIG_A)
cache.cache_tool_signature("toolu_2", _SIG_B)
cache.cache_tool_signature("toolu_3", _SIG_C)
cache.cache_tool_signature("toolu_4", _SIG_D)
# Oldest entry should be evicted once the limit is exceeded.
assert cache.get_tool_signature("toolu_1") is None
assert cache.get_tool_signature("toolu_4") == _SIG_D
# ===== Layer 2: Thinking Families ===== # ===== Layer 2: Thinking Families =====