feat(reflect): step-by-step context caching for the Gemini tool loop

hindsight-api-slim/hindsight_api/config.py

@@ -617,6 +617,7 @@ def _resolve_operation_temperature(operation_env: str, default: float) -> float
 
 # Reflect agent settings
 ENV_REFLECT_MAX_ITERATIONS = "HINDSIGHT_API_REFLECT_MAX_ITERATIONS"
+ENV_REFLECT_PROMPT_CACHE_ENABLED = "HINDSIGHT_API_REFLECT_PROMPT_CACHE_ENABLED"
 ENV_REFLECT_MAX_CONTEXT_TOKENS = "HINDSIGHT_API_REFLECT_MAX_CONTEXT_TOKENS"
 ENV_REFLECT_WALL_TIMEOUT = "HINDSIGHT_API_REFLECT_WALL_TIMEOUT"
 ENV_REFLECT_MISSION = "HINDSIGHT_API_REFLECT_MISSION"
@@ -1054,6 +1055,10 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 
 # Reflect agent settings
 DEFAULT_REFLECT_MAX_ITERATIONS = 10  # Max tool call iterations before forcing response
+# Step-by-step context caching for the reflect tool loop (Gemini). On by default;
+# requires the global prompt cache (HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED) to also
+# be on. Set false to force reflect to run uncached even when prompt caching is on.
+DEFAULT_REFLECT_PROMPT_CACHE_ENABLED = True
 DEFAULT_REFLECT_MAX_CONTEXT_TOKENS = 100_000  # Max accumulated context tokens before forcing final prompt
 DEFAULT_REFLECT_WALL_TIMEOUT = 300  # Wall-clock timeout in seconds for the entire reflect operation (5 minutes)
 DEFAULT_REFLECT_SOURCE_FACTS_MAX_TOKENS = -1  # Token budget for source facts in search_observations (-1 = disabled)
@@ -1914,6 +1919,7 @@ class HindsightConfig:
     reflect_max_iterations: int
     reflect_max_context_tokens: int
     reflect_wall_timeout: int
+    reflect_prompt_cache_enabled: bool
 
     # OpenTelemetry tracing configuration
     otel_traces_enabled: bool
@@ -2927,6 +2933,10 @@ def from_env(cls) -> "HindsightConfig":
             retain_max_concurrent=int(os.getenv(ENV_RETAIN_MAX_CONCURRENT, str(DEFAULT_RETAIN_MAX_CONCURRENT))),
             # Reflect agent settings
             reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))),
+            reflect_prompt_cache_enabled=os.getenv(
+                ENV_REFLECT_PROMPT_CACHE_ENABLED, str(DEFAULT_REFLECT_PROMPT_CACHE_ENABLED)
+            ).lower()
+            in ("1", "true", "yes", "on"),
             reflect_max_context_tokens=int(
                 os.getenv(ENV_REFLECT_MAX_CONTEXT_TOKENS, str(DEFAULT_REFLECT_MAX_CONTEXT_TOKENS))
             ),

hindsight-api-slim/hindsight_api/engine/llm_interface.py

@@ -116,6 +116,7 @@ async def call_with_tools(
         max_backoff: float = 30.0,
         tool_choice: str | dict[str, Any] = "auto",
         cached_prefix: str | None = None,
+        cached_prefix_message_count: int = 0,
     ) -> LLMToolCallResult:
         """
         Make an LLM API call with tool/function calling support.
@@ -185,6 +186,45 @@ async def get_or_create_cached_prefix(
         """
         return None
 
+    # ── Step-by-step incremental prompt caching (optional) ─────────────────────
+    #
+    # For agentic loops (reflect) the dominant cost is the conversation prefix
+    # re-sent every turn, not the static system prefix. Providers that can cache
+    # a *growing* prefix implement these: the caller rolls one cache per step
+    # (each covering the previous step's full input), passes its handle plus the
+    # message count it covers to ``call_with_tools`` so only the new turns are
+    # sent fresh, and tears the caches down when the loop ends. Default no-ops so
+    # non-supporting providers transparently run uncached.
+
+    def supports_incremental_prompt_cache(self) -> bool:
+        """Whether this provider can cache a growing multi-turn conversation prefix."""
+        return False
+
+    async def create_incremental_cache(
+        self,
+        *,
+        session_id: str,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None = None,
+    ) -> str | None:
+        """Cache ``system + tools + messages`` and return an opaque handle, or None.
+
+        The handle is passed back to ``call_with_tools(cached_prefix=...,
+        cached_prefix_message_count=len(messages))``. Caches are grouped under
+        ``session_id`` for teardown via ``delete_cache_session``. Returns None
+        when caching is unavailable or the prefix is too small — caller falls
+        back to an uncached call.
+        """
+        return None
+
+    async def delete_cached_prefix(self, name: str) -> None:
+        """Best-effort delete of a single cache handle (a superseded step)."""
+        return None
+
+    async def delete_cache_session(self, session_id: str) -> None:
+        """Best-effort teardown of every cache created under ``session_id``."""
+        return None
+
     async def submit_batch(
         self,
         requests: list[dict[str, Any]],

hindsight-api-slim/hindsight_api/engine/llm_wrapper.py

@@ -967,6 +967,7 @@ async def call_with_tools(
         max_backoff: float | None = None,
         tool_choice: str | dict[str, Any] = "auto",
         cached_prefix: str | None = None,
+        cached_prefix_message_count: int = 0,
     ) -> "LLMToolCallResult":
         """
         Make an LLM API call with tool/function calling support.
@@ -1034,9 +1035,14 @@ async def call_with_tools(
                     await stack.enter_async_context(sem)
 
                 # cached_prefix is only set for providers that returned a handle
-                # from get_or_create_cached_prefix(); forward it only when present
-                # so non-caching providers keep their signature (same as call()).
-                cache_kwarg = {"cached_prefix": cached_prefix} if cached_prefix is not None else {}
+                # from get_or_create_cached_prefix() / create_incremental_cache();
+                # forward it (plus how many leading messages it covers) only when
+                # present so non-caching providers keep their signature.
+                cache_kwarg = (
+                    {"cached_prefix": cached_prefix, "cached_prefix_message_count": cached_prefix_message_count}
+                    if cached_prefix is not None
+                    else {}
+                )
                 try:
                     # Delegate to provider implementation
                     result = await self._provider_impl.call_with_tools(

hindsight-api-slim/hindsight_api/engine/providers/gemini_cache.py

@@ -56,6 +56,14 @@
 # to None and callers proceed uncached, rather than stalling the whole batch.
 _DEFAULT_CREATE_TIMEOUT_SECONDS = 30.0
 
+# TTL for the per-step reflect caches created by ``create_incremental``. These
+# live only for the duration of one reflect (seconds), so the TTL is just a
+# storage backstop in case the explicit ``delete_session`` at reflect end is
+# missed (crash / event-loop teardown). Short so orphaned caches age out fast —
+# storage is billed per token-hour, so a 5-minute cap keeps the cost of a leaked
+# cache negligible.
+_DEFAULT_INCREMENTAL_TTL_SECONDS = 5 * 60
+
 
 @dataclass
 class _CacheEntry:
@@ -92,6 +100,10 @@ def __init__(
         self._create_timeout_seconds = create_timeout_seconds
         self._entries: dict[str, _CacheEntry] = {}
         self._lock = asyncio.Lock()
+        # session_id -> CachedContent names created via ``create_incremental``.
+        # A reflect creates a fresh rolling cache per step under one session id;
+        # ``delete_session`` tears them all down when the reflect finishes.
+        self._sessions: dict[str, list[str]] = {}
 
     @staticmethod
     def fingerprint(
@@ -229,18 +241,99 @@ def invalidate(self, name: str) -> None:
             if entry.name == name:
                 self._entries.pop(key, None)
 
+    async def create_incremental(
+        self,
+        *,
+        session_id: str,
+        model: str,
+        system_instruction: str,
+        contents: list[Any],
+        tools: list[dict[str, Any]] | None = None,
+    ) -> str | None:
+        """Create a fresh CachedContent holding ``system + tools + contents`` and
+        track it under ``session_id`` for later teardown.
+
+        Unlike ``get_or_create``, this does NOT deduplicate by fingerprint: each
+        step of a reflect grows the conversation prefix, so every call is a
+        distinct, single-use cache. The reflect loop creates one per step (each
+        covering the previous step's full input) and reuses it for exactly the
+        next model turn, then supersedes it. All caches for the session are
+        deleted by ``delete_session`` when the reflect ends; the short TTL is
+        only a backstop.
+
+        Returns the cache resource name, or ``None`` when caching is disabled,
+        the prefix is below the model minimum, or the create otherwise fails —
+        callers MUST fall back to an uncached call in that case.
+        """
+        try:
+            name = await self._create_cache(
+                model=model,
+                system_instruction=system_instruction,
+                tools=tools,
+                contents=contents,
+                ttl_seconds=_DEFAULT_INCREMENTAL_TTL_SECONDS,
+            )
+        except _CacheNotEligible as e:
+            logger.debug(
+                "GeminiCacheManager: incremental prefix not eligible (model=%s, reason=%s) — caller falls back",
+                model,
+                e,
+            )
+            return None
+        except Exception:
+            logger.exception(
+                "GeminiCacheManager: failed to create incremental cache (model=%s); caller falls back",
+                model,
+            )
+            return None
+        if name is not None:
+            self._sessions.setdefault(session_id, []).append(name)
+        return name
+
+    async def delete(self, name: str) -> None:
+        """Best-effort server-side delete of a single CachedContent.
+
+        Swallows all errors: a failed delete just means the cache ages out on
+        its TTL. Also drops any matching in-process entry.
+        """
+        self.invalidate(name)
+        try:
+            await self._client.aio.caches.delete(name=name)
+        except Exception:
+            logger.debug("GeminiCacheManager: delete of cache %s failed (will age out on TTL)", name, exc_info=True)
+
+    async def delete_session(self, session_id: str) -> None:
+        """Delete every CachedContent created for ``session_id`` (reflect teardown).
+
+        Deletes concurrently and best-effort — a reflect must never fail because
+        a cache couldn't be torn down; the short TTL is the backstop.
+        """
+        names = self._sessions.pop(session_id, [])
+        if not names:
+            return
+        await asyncio.gather(*(self.delete(n) for n in names), return_exceptions=True)
+
     async def _create_cache(
         self,
         *,
         model: str,
         system_instruction: str,
         tools: list[dict[str, Any]] | None = None,
+        contents: list[Any] | None = None,
+        ttl_seconds: int | None = None,
     ) -> str | None:
         """Wrap ``client.aio.caches.create`` with the config we want.
 
         The SDK surface differs slightly across google-genai versions;
         this implementation targets the >=1.0.0 line where caches live
         under ``client.aio.caches``.
+
+        ``contents`` (already-converted ``genai_types.Content`` turns) is
+        appended after the system_instruction/tools so the cache can hold a
+        growing multi-turn conversation prefix, not just the static prefix —
+        this is what the step-by-step reflect cache relies on. ``ttl_seconds``
+        overrides the manager default (used to give per-step reflect caches a
+        short backstop TTL).
         """
         # Lazy import so this module doesn't require the SDK at import time.
         from google.genai import types as genai_types
@@ -254,8 +347,10 @@ async def _create_cache(
         # still part of the fingerprint so a schema change keys a fresh cache.
         config_kwargs: dict[str, Any] = {
             "system_instruction": system_instruction,
-            "ttl": f"{self._ttl_seconds}s",
+            "ttl": f"{ttl_seconds if ttl_seconds is not None else self._ttl_seconds}s",
         }
+        if contents:
+            config_kwargs["contents"] = contents
         if tools:
             # OpenAI-style {"function": {...}} entries must be converted to
             # Gemini's Tool/FunctionDeclaration shape before caching.