fix(agent): namespace prompt_cache_key with logical session scope

agent/chat_completion_helpers.py

@@ -1222,6 +1222,10 @@ def build_api_kwargs(agent, api_messages: list, tools_for_api: list | None = Non
             tools=tools_for_api,
             reasoning_config=agent.reasoning_config,
             session_id=getattr(agent, "session_id", None),
+            gateway_session_key=getattr(agent, "_gateway_session_key", None),
+            parent_session_id=getattr(agent, "_parent_session_id", None),
+            is_subagent=getattr(agent, "is_subagent", False),
+            session_db=getattr(agent, "_session_db", None),
             base_url=agent.base_url,
             max_tokens=agent.max_tokens,
             timeout=agent._resolved_api_call_timeout(),
@@ -1329,6 +1333,10 @@ def build_api_kwargs(agent, api_messages: list, tools_for_api: list | None = Non
             reasoning_config=agent.reasoning_config,
             request_overrides=agent.request_overrides,
             session_id=getattr(agent, "session_id", None),
+            gateway_session_key=getattr(agent, "_gateway_session_key", None),
+            parent_session_id=getattr(agent, "_parent_session_id", None),
+            is_subagent=getattr(agent, "is_subagent", False),
+            session_db=getattr(agent, "_session_db", None),
             provider_profile=_profile,
             ollama_num_ctx=agent._ollama_num_ctx,
             # Context forwarded to profile hooks:
@@ -1361,6 +1369,10 @@ def build_api_kwargs(agent, api_messages: list, tools_for_api: list | None = Non
         reasoning_config=agent.reasoning_config,
         request_overrides=agent.request_overrides,
         session_id=getattr(agent, "session_id", None),
+        gateway_session_key=getattr(agent, "_gateway_session_key", None),
+        parent_session_id=getattr(agent, "_parent_session_id", None),
+        is_subagent=getattr(agent, "is_subagent", False),
+        session_db=getattr(agent, "_session_db", None),
         model_lower=(agent.model or "").lower(),
         is_openrouter=_is_or,
         is_nous=_is_nous,

agent/transports/chat_completions.py

@@ -47,6 +47,11 @@ def _add_prompt_cache_key(
     messages: list[dict[str, Any]],
     tools: list[dict[str, Any]] | None,
     supports_prompt_cache_key: bool,
+    session_id: str | None = None,
+    gateway_session_key: str | None = None,
+    parent_session_id: str | None = None,
+    is_subagent: bool = False,
+    session_db: Any = None,
 ) -> None:
     """Add a content-addressed key only for an explicitly capable endpoint."""
     if not supports_prompt_cache_key:
@@ -62,9 +67,16 @@ def _add_prompt_cache_key(
 
     # Reuse the Responses transport's single authoritative hash algorithm so
     # equivalent static prefixes route to the same cache bucket across modes.
-    from agent.transports.codex import _content_cache_key
-
-    cache_key = _content_cache_key(_static_prompt_instructions(messages), tools)
+    from agent.transports.codex import _logical_cache_scope, _scoped_cache_key
+
+    scope = _logical_cache_scope(
+        session_id=session_id,
+        gateway_session_key=gateway_session_key,
+        parent_session_id=parent_session_id,
+        is_subagent=is_subagent,
+        session_db=session_db,
+    )
+    cache_key = _scoped_cache_key(scope, _static_prompt_instructions(messages), tools)
     if cache_key:
         api_kwargs["prompt_cache_key"] = cache_key
 
@@ -584,6 +596,11 @@ def build_kwargs(
             tools=api_kwargs.get("tools"),
             supports_prompt_cache_key=bool(params.get("supports_prompt_cache_key"))
             or _is_openai_api_base_url(params.get("base_url")),
+            session_id=params.get("session_id"),
+            gateway_session_key=params.get("gateway_session_key") or params.get("conversation_id"),
+            parent_session_id=params.get("parent_session_id"),
+            is_subagent=bool(params.get("is_subagent")),
+            session_db=params.get("session_db"),
         )
 
         return api_kwargs
@@ -733,6 +750,11 @@ def _build_kwargs_from_profile(self, profile, model, sanitized, tools, params):
             messages=sanitized,
             tools=api_kwargs.get("tools"),
             supports_prompt_cache_key=bool(profile.supports_prompt_cache_key),
+            session_id=params.get("session_id"),
+            gateway_session_key=params.get("gateway_session_key") or params.get("conversation_id"),
+            parent_session_id=params.get("parent_session_id"),
+            is_subagent=bool(params.get("is_subagent")),
+            session_db=params.get("session_db"),
         )
 
         return api_kwargs

agent/transports/codex.py

@@ -117,6 +117,98 @@ def _default_prompt_cache_retention_for_request(
     return None
 
 
+def _logical_cache_scope(
+    session_id: Optional[str] = None,
+    gateway_session_key: Optional[str] = None,
+    parent_session_id: Optional[str] = None,
+    is_subagent: bool = False,
+    conversation_id: Optional[str] = None,
+    session_db: Any = None,
+) -> Optional[str]:
+    """Derive the logical isolation scope for prompt cache routing.
+
+    Scope precedence order:
+    1. Cron job id (e.g. `cron:<job_id>`, parsed from `cron_<job>_<timestamp>`, ignoring per-fire timestamps)
+    2. Child session id (e.g. `session:<child_session_id>`, for subagent turns)
+    3. Gateway conversation id (e.g. `session:<gateway_session_key>` / `conversation_id`)
+    4. Session id / lineage root (e.g. `session:<parent_session_id>` or `session:<session_id>`)
+    5. Content-only fallback (returns None when no session identity is provided)
+    """
+    sid = str(session_id or "").strip()
+    effective_gateway_key = (gateway_session_key or conversation_id or "").strip()
+
+    # 1. Cron job id precedence
+    if sid and (sid.startswith("cron_") or sid.startswith("cron:")):
+        raw = sid[5:]  # strip 'cron_' or 'cron:'
+        # Strip trailing timestamp (ISO-8601, compact YYYYMMDD_HHMMSS, or epoch digits)
+        job_id = re.sub(
+            r'(?:[_\:](?:\d{4}-\d{2}-\d{2}[T_\-]?\d{2}[\:\-]?\d{2}[\:\-]?\d{2}.*|\d{8}[_\-]\d{6}.*|\d{9,13}))$',
+            '',
+            raw,
+        )
+        job_id = job_id or raw
+        return f"cron:{job_id}"
+
+    # 2. Child session id precedence (subagents)
+    if is_subagent:
+        return f"session:{sid}" if sid else None
+
+    # 3. Gateway conversation id precedence
+    if effective_gateway_key:
+        return f"session:{effective_gateway_key}"
+
+    # 4. Session id / lineage root precedence (interactive CLI/TUI, compression lineage)
+    root_id = None
+    if session_db and hasattr(session_db, "get_conversation_root"):
+        try:
+            target = sid or str(parent_session_id or "").strip()
+            if target:
+                root_id = session_db.get_conversation_root(target)
+        except Exception:
+            root_id = None
+    if not root_id:
+        if parent_session_id and str(parent_session_id).strip():
+            root_id = str(parent_session_id).strip()
+        elif sid:
+            root_id = sid
+
+    if root_id:
+        return f"session:{root_id}"
+
+    # 5. Content-only fallback (no session identity available)
+    return None
+
+
+def _scoped_cache_key(
+    scope: Optional[str],
+    instructions: str,
+    tools: Optional[List[Dict[str, Any]]],
+) -> Optional[str]:
+    """Content-address the prompt cache key from (logical_scope + static_prefix).
+
+    Returns ``pck_<sha256[:24]>`` of (scope + "\x00" + instructions + sorted tools),
+    or None when there is no static prefix (instructions and tools both empty/None).
+    If scope is None, falls back to unscoped _content_cache_key.
+    """
+    if not instructions and not tools:
+        return None
+    if not scope:
+        return _content_cache_key(instructions, tools)
+
+    tools_part = ""
+    if tools:
+        sorted_tools = sorted(
+            (t for t in tools if isinstance(t, dict)),
+            key=lambda t: str(t.get("name") or t.get("type") or ""),
+        )
+        tools_part = json.dumps(
+            sorted_tools, sort_keys=True, ensure_ascii=False, separators=(",", ":")
+        )
+    content = f"{scope}\x00{instructions or ''}\x00{tools_part}"
+    digest = hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest()[:24]
+    return f"pck_{digest}"
+
+
 def _content_cache_key(instructions: str, tools: Optional[List[Dict[str, Any]]]) -> Optional[str]:
     """Content-address the prompt cache key from the static request prefix.
 
@@ -340,13 +432,30 @@ def build_kwargs(
             kwargs["parallel_tool_calls"] = True
 
         session_id = params.get("session_id")
-        # prompt_cache_key is content-addressed from the static prefix
-        # (instructions + tools), NOT session_id — recurring cron jobs carry a
-        # per-fire timestamp in session_id (cron_<id>_<ts>) that made every run
-        # cache-cold. session_id is left untouched for transcript isolation and
-        # the cache-scope routing headers below. Falls back to session_id when
-        # there is no static content to hash.
-        cache_key = _content_cache_key(instructions, response_tools) or session_id
+        gateway_session_key = params.get("gateway_session_key") or params.get("conversation_id")
+        parent_session_id = params.get("parent_session_id")
+        is_subagent = bool(params.get("is_subagent"))
+
+        # Explicit prompt_cache_key override (top-level param, request_overrides, or extra_body)
+        explicit_cache_key = params.get("prompt_cache_key")
+        request_overrides = params.get("request_overrides")
+        if not explicit_cache_key and isinstance(request_overrides, dict):
+            explicit_cache_key = request_overrides.get("prompt_cache_key")
+            if not explicit_cache_key and isinstance(request_overrides.get("extra_body"), dict):
+                explicit_cache_key = request_overrides["extra_body"].get("prompt_cache_key")
+
+        if explicit_cache_key:
+            cache_key = explicit_cache_key
+        else:
+            scope = _logical_cache_scope(
+                session_id=session_id,
+                gateway_session_key=gateway_session_key,
+                parent_session_id=parent_session_id,
+                is_subagent=is_subagent,
+                session_db=params.get("session_db"),
+            )
+            cache_key = _scoped_cache_key(scope, instructions, response_tools) or session_id
+
         # xAI Responses takes prompt_cache_key in extra_body (set further
         # down); GitHub Models opts out of cache-key routing entirely.
         if not is_github_responses and not is_xai_responses and cache_key:

contributors/emails/StanleyStetson@users.noreply.github.com

@@ -0,0 +1 @@
+StanleyStetson

tests/agent/transports/test_codex_transport.py

@@ -46,10 +46,8 @@ class TestCodexBuildKwargs:
 
 
     def test_cache_key_is_content_addressed_not_session_id(self, transport):
-        """prompt_cache_key is content-addressed from the static prefix
-        (instructions + tools), not the session_id. This keeps recurring cron
-        jobs — whose session_id carries a per-fire timestamp — on a stable warm
-        cache key. The key is a 'pck_' hash and must NOT equal session_id."""
+        """prompt_cache_key is content-addressed from (logical_scope + static_prefix),
+        not raw session_id. The key is a 'pck_' hash and must NOT equal session_id."""
         messages = [{"role": "user", "content": "Hi"}]
         kw = transport.build_kwargs(
             model="gpt-5.4", messages=messages, tools=[],
@@ -59,10 +57,42 @@ def test_cache_key_is_content_addressed_not_session_id(self, transport):
         assert pck.startswith("pck_")
         assert pck != "cron_job42_20260624_143000"
 
-    def test_cache_key_stable_across_session_ids(self, transport):
-        """Same static prefix + different session_id (e.g. two cron fires of the
-        same job) must yield the same prompt_cache_key — the whole point of the
-        fix: repeated fires reuse the warm prefix instead of going cold."""
+    def test_cache_key_logical_scope_precedence(self):
+        """Verify _logical_cache_scope precedence:
+        cron job id > child session id > gateway key > lineage root/session id > content-only fallback.
+        """
+        from agent.transports.codex import _logical_cache_scope
+
+        # 1. Cron job id (ignores timestamps)
+        assert _logical_cache_scope(session_id="cron_job42_20260624_143000") == "cron:job42"
+        assert _logical_cache_scope(session_id="cron_job42_1722800000") == "cron:job42"
+        assert _logical_cache_scope(session_id="cron_job42") == "cron:job42"
+
+        # 2. Subagent child session id
+        assert _logical_cache_scope(
+            session_id="subagent_child",
+            parent_session_id="parent_root",
+            is_subagent=True,
+        ) == "session:subagent_child"
+
+        # 3. Gateway conversation key
+        assert _logical_cache_scope(
+            session_id="sess_123",
+            gateway_session_key="telegram:chat_999",
+        ) == "session:telegram:chat_999"
+
+        # 4. Lineage root / session id
+        assert _logical_cache_scope(
+            session_id="sess_segment_2",
+            parent_session_id="sess_root_1",
+        ) == "session:sess_root_1"
+        assert _logical_cache_scope(session_id="sess_plain") == "session:sess_plain"
+
+        # 5. Content-only fallback
+        assert _logical_cache_scope() is None
+
+    def test_cache_key_cron_same_job_same_key(self, transport):
+        """Two fires of the same cron job (timestamped session_ids) must yield the same key."""
         messages = [{"role": "user", "content": "Hi"}]
         kw1 = transport.build_kwargs(
             model="gpt-5.4", messages=messages, tools=[],
@@ -74,6 +104,94 @@ def test_cache_key_stable_across_session_ids(self, transport):
         )
         assert kw1["prompt_cache_key"] == kw2["prompt_cache_key"]
 
+    def test_cache_key_cron_different_jobs_different_keys(self, transport):
+        """Two different cron jobs with the same static prefix must yield different keys."""
+        messages = [{"role": "user", "content": "Hi"}]
+        kw1 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="cron_job42_20260624_143000",
+        )
+        kw2 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="cron_job99_20260624_143000",
+        )
+        assert kw1["prompt_cache_key"] != kw2["prompt_cache_key"]
+
+    def test_cache_key_independent_interactive_sessions_different_keys(self, transport):
+        """Two independent interactive sessions with identical tools/instructions get different keys."""
+        messages = [{"role": "user", "content": "Hi"}]
+        kw1 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="session_A",
+        )
+        kw2 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="session_B",
+        )
+        assert kw1["prompt_cache_key"] != kw2["prompt_cache_key"]
+
+    def test_cache_key_gateway_session_key_same_scope(self, transport):
+        """Same gateway conversation key across session rotations yields the same key."""
+        messages = [{"role": "user", "content": "Hi"}]
+        kw1 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="sess_1",
+            gateway_session_key="telegram:chat_100",
+        )
+        kw2 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="sess_2",
+            gateway_session_key="telegram:chat_100",
+        )
+        assert kw1["prompt_cache_key"] == kw2["prompt_cache_key"]
+
+    def test_cache_key_parent_child_subagent_different_keys(self, transport):
+        """Parent session and subagent child session yield different cache keys."""
+        messages = [{"role": "user", "content": "Hi"}]
+        kw_parent = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="parent_sess",
+        )
+        kw_child = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="child_subagent",
+            parent_session_id="parent_sess",
+            is_subagent=True,
+        )
+        assert kw_parent["prompt_cache_key"] != kw_child["prompt_cache_key"]
+
+    def test_cache_key_sibling_subagents_different_keys(self, transport):
+        """Two sibling subagent children of the same parent yield different cache keys."""
+        messages = [{"role": "user", "content": "Hi"}]
+        kw_child1 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="child_subagent_1",
+            parent_session_id="parent_sess",
+            is_subagent=True,
+        )
+        kw_child2 = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="child_subagent_2",
+            parent_session_id="parent_sess",
+            is_subagent=True,
+        )
+        assert kw_child1["prompt_cache_key"] != kw_child2["prompt_cache_key"]
+
+    def test_cache_key_compression_lineage_same_root_key(self, transport):
+        """Context compression lineage (parent_session_id set without is_subagent) uses root scope."""
+        messages = [{"role": "user", "content": "Hi"}]
+        kw_root = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="root_sess",
+        )
+        kw_rotated = transport.build_kwargs(
+            model="gpt-5.4", messages=messages, tools=[],
+            session_id="rotated_sess",
+            parent_session_id="root_sess",
+            is_subagent=False,
+        )
+        assert kw_root["prompt_cache_key"] == kw_rotated["prompt_cache_key"]
+
 
     def test_github_responses_drops_message_item_id_end_to_end(self, transport):
         # #32716: Copilot binds codex_message_items ids to a backend