diff --git a/src/ai_client.py b/src/ai_client.py index 4e89ded5..3de555cd 100644 --- a/src/ai_client.py +++ b/src/ai_client.py @@ -1664,6 +1664,46 @@ def _should_cache_gemini_result(sys_instr: str) -> Result[bool]: errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to count gemini tokens: {e}", source="ai_client._should_cache_gemini_result", original=e)], ) +def _create_gemini_cache_result(sys_instr: str, tools_decl: Any, file_items: list[dict[str, Any]] | None) -> Result[Any]: + """Create a Gemini cache and the corresponding GenerateContentConfig. + + Returns Result(data=chat_config_with_cached_content) on success and + Result(data=None, errors=[ErrorInfo]) on SDK failure. Side effects on + globals _gemini_cache, _gemini_cache_created_at, _gemini_cached_file_paths + are managed inside the helper (set on success, reset on failure to match + original semantics). + """ + global _gemini_cache, _gemini_cache_created_at, _gemini_cached_file_paths + types = _require_warmed("google.genai").types + try: + _gemini_cache = _gemini_client.caches.create( + model=_model, + config=types.CreateCachedContentConfig( + system_instruction=sys_instr, + tools=cast(Any, tools_decl), + ttl=f"{_GEMINI_CACHE_TTL}s", + ) + ) + _gemini_cache_created_at = time.time() + _gemini_cached_file_paths = [str(item.get("path", "")) for item in (file_items or []) if item.get("path")] + chat_config = types.GenerateContentConfig( + cached_content=_gemini_cache.name, + temperature=_temperature, + max_output_tokens=_max_tokens, + safety_settings=[types.SafetySetting(category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH)] + ) + _append_comms("OUT", "request", {"message": f"[CACHE CREATED] {_gemini_cache.name}"}) + return Result(data=chat_config) + except Exception as e: + _gemini_cache = None + _gemini_cache_created_at = None + _gemini_cached_file_paths = [] + _append_comms("OUT", "request", {"message": f"[CACHE FAILED] {type(e).__name__}: {e} \u2014 falling back to inline system_instruction"}) + return Result( + data=None, + errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to create gemini cache: {type(e).__name__}: {e}", source="ai_client._create_gemini_cache_result", original=e)], + ) + def _extract_gemini_thoughts(resp: Any) -> str: """ Extracts concatenated thinking text from a Gemini response object's parts. @@ -1752,29 +1792,9 @@ def _send_gemini(md_content: str, user_message: str, base_dir: str, should_cache = _should_cache_gemini_result(sys_instr).data if should_cache and _gemini_client: - try: - _gemini_cache = _gemini_client.caches.create( - model=_model, - config=types.CreateCachedContentConfig( - system_instruction=sys_instr, - tools=cast(Any, tools_decl), - ttl=f"{_GEMINI_CACHE_TTL}s", - ) - ) - _gemini_cache_created_at = time.time() - _gemini_cached_file_paths = [str(item.get("path", "")) for item in (file_items or []) if item.get("path")] - chat_config = types.GenerateContentConfig( - cached_content=_gemini_cache.name, - temperature=_temperature, - max_output_tokens=_max_tokens, - safety_settings=[types.SafetySetting(category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH)] - ) - _append_comms("OUT", "request", {"message": f"[CACHE CREATED] {_gemini_cache.name}"}) - except Exception as e: - _gemini_cache = None - _gemini_cache_created_at = None - _gemini_cached_file_paths = [] - _append_comms("OUT", "request", {"message": f"[CACHE FAILED] {type(e).__name__}: {e} \u2014 falling back to inline system_instruction"}) + cached_config_result = _create_gemini_cache_result(sys_instr, tools_decl, file_items) + if cached_config_result.ok: + chat_config = cached_config_result.data kwargs: dict[str, Any] = {"model": _model, "config": chat_config} if old_history: kwargs["history"] = old_history diff --git a/tests/tier2/phase10_site5_test.py b/tests/tier2/phase10_site5_test.py new file mode 100644 index 00000000..bb9bba33 --- /dev/null +++ b/tests/tier2/phase10_site5_test.py @@ -0,0 +1,18 @@ +"""Phase 10 site 5: _create_gemini_cache_result helper.""" +import sys +sys.path.insert(0, ".") + + +def test_phase10_site5_create_gemini_cache_result_exists(): + import src.ai_client + assert hasattr(src.ai_client, "_create_gemini_cache_result"), \ + "_create_gemini_cache_result helper missing" + + +def test_phase10_site5_create_gemini_cache_result_returns_result(): + import src.ai_client + import inspect + fn = src.ai_client._create_gemini_cache_result + sig = inspect.signature(fn) + assert "Result" in str(sig.return_annotation), \ + f"_create_gemini_cache_result return must be Result, got {sig.return_annotation}" \ No newline at end of file