refactor(ai_client): extract _create_gemini_cache_result helper (Phase 10 site 5)

Site L1773: cache.create block in _send_gemini had multiple global side
effects (sets _gemini_cache, _gemini_cache_created_at, _gemini_cached_file_paths,
returns chat_config with cached_content). Except body reset globals on failure.

Per TIER1_REVIEW: logging is NOT a drain. MIGRATE to Result[Any].

New helper _create_gemini_cache_result(sys_instr, tools_decl, file_items) -> Result[Any]:
- Returns Result(data=chat_config) on SDK success (sets globals, logs [CACHE CREATED])
- Returns Result(data=None, errors=[ErrorInfo]) on SDK failure (resets globals,
  logs [CACHE FAILED])
- Preserves original semantics: globals set on success, reset on failure

Caller:
  cached_config_result = _create_gemini_cache_result(sys_instr, tools_decl, file_items)
  if cached_config_result.ok:
      chat_config = cached_config_result.data

Audit: ai_client BC 5 -> 4. _send_gemini cache-related BC sites all migrated.
This commit is contained in:
ed
2026-06-20 13:05:48 -04:00
parent ef99b0e3f5
commit 1b03c280a9
2 changed files with 61 additions and 23 deletions
+43 -23
View File
@@ -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
+18
View File
@@ -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}"