diff --git a/agent/background_review.py b/agent/background_review.py index 32a0505501..6ff57f6f16 100644 --- a/agent/background_review.py +++ b/agent/background_review.py @@ -24,7 +24,7 @@ import logging import os from pathlib import Path import threading -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from agent.thread_scoped_output import thread_scoped_silence @@ -1099,6 +1099,299 @@ def _log_review_completion(usage: Dict[str, Any], result: str) -> None: ) +def build_cache_parity_fork( + agent: Any, + task_cfg: Optional[Dict[str, Any]] = None, + *, + max_iterations: int, + write_origin: str = "background_review", +) -> Tuple[Any, Dict[str, Any], bool]: + """Construct a detached AIAgent fork with warm prompt-cache parity. + + This is the fork recipe the self-improvement background review uses, + extracted so other conversation-snapshot consumers (``/btw`` side + questions) get the identical cache-parity guarantees: same runtime and + credentials as the parent, byte-identical system prompt / tools[] / + reasoning config on the same-model path, shared session_id for prefix + warmth, and full persistence detachment (no state.db writes, no session + rotation, no external memory providers, in-place-only compaction). + + Returns ``(fork_agent, runtime_dict, routed)`` where ``routed`` is True + when auxiliary config redirected the fork to a different model (cache + cold; callers should replay a digest instead of the full snapshot). + + The caller keeps ownership of: registering the fork on the parent's + ``_active_children`` / ``_background_review_agent`` slots, thread tool + whitelisting, running the conversation, usage attribution, and teardown + (``shutdown_memory_provider()`` + ``close()``). + """ + # Local import to avoid a hard circular dep at module load. + from run_agent import AIAgent + + # Inherit the parent agent's live runtime (provider, model, + # base_url, api_key, api_mode) so the fork uses the exact + # same credentials the main turn is using. Without this, + # AIAgent.__init__ re-runs auto-resolution from env vars, + # which fails for OAuth-only providers, session-scoped + # creds, or credential-pool setups where the resolver can't + # reconstruct auth from scratch -- producing the spurious + # "No LLM provider configured" warning at end of turn. + # _resolve_review_runtime() returns the parent's live runtime by + # default (routed=False; main model, warm cache), or — when the user + # set auxiliary.background_review.{provider,model} to a different + # model — that model's runtime (routed=True). The codex_app_server + # -> codex_responses downgrade is applied inside the resolver. + _rt = _resolve_review_runtime(agent, task_cfg) + _routed = bool(_rt.get("routed")) + # skip_memory=True keeps the review fork from + # touching external memory plugins (honcho, mem0, + # supermemory, etc.). Without it, the fork's + # __init__ rebuilds its own _memory_manager from + # config, scoped to the parent's session_id, and + # run_conversation() then leaks the harness prompt + # into the user's real memory namespace via three + # ingestion sites: on_turn_start (cadence + turn + # message), prefetch_all (recall query), and + # sync_all (harness prompt + review output recorded + # as a (user, assistant) turn pair). Built-in + # MEMORY.md / USER.md state is re-bound from the + # parent below so memory(action="add") writes from + # the review still land on disk; the review just + # has zero side effects on external providers. + # Match parent's toolset config so ``tools[]`` is byte-identical + # in the request body — Anthropic's cache key includes it. + # (The runtime whitelist below still restricts dispatch.) + _fork_kwargs: Dict[str, Any] = {} + if isinstance(_rt.get("max_tokens"), int): + _fork_kwargs["max_tokens"] = _rt["max_tokens"] + if isinstance(_rt.get("command"), str) and _rt["command"]: + _fork_kwargs["acp_command"] = _rt["command"] + _fork_kwargs["acp_args"] = _rt.get("args") or [] + # Match parent's reasoning config so the fork's ``thinking`` / + # ``output_config`` are byte-identical in the request body — + # Anthropic's cache key is namespaced by ``thinking`` presence. + # Same-model path only: when routed to a different aux model the + # cache is cold regardless (parity buys nothing) and the parent's + # effort vocabulary may not be valid for the routed model/provider + # (e.g. OpenRouter ``extra_body.reasoning.effort`` is forwarded + # unclamped; codex_responses passes ``max``/``ultra`` through + # unmapped except on gpt-5.6/xAI). Let the routed fork use + # provider defaults — matching the ``not _routed`` gate on + # _cached_system_prompt below. + if not _routed: + _fork_kwargs["reasoning_config"] = getattr(agent, "reasoning_config", None) + # Gateway session context is appended to the parent's cached + # system prompt at API-call time through this field. Preserve + # it on same-model forks so the complete effective system + # prompt remains byte-identical and can reuse the warm prefix. + _fork_kwargs["ephemeral_system_prompt"] = getattr( + agent, "ephemeral_system_prompt", None + ) + # Prefill messages are inserted immediately after the system + # message at API-call time (chat_completion_helpers.py / + # conversation_loop.py), so a parent with prefill configured + # (gateway prefill_messages_file) would otherwise diverge + # from the warm prefix at message index 1 — same bug class + # as the ephemeral prompt above, one position later. + # Deep copy: the unicode-error recovery path mutates + # prefill entries IN PLACE (_sanitize_messages_surrogates + # via conversation_loop), so sharing dicts would let a + # fork-side sanitize rewrite the parent's prefill bytes. + _parent_prefill = copy.deepcopy( + getattr(agent, "prefill_messages", None) or [] + ) + if _parent_prefill: + _fork_kwargs["prefill_messages"] = _parent_prefill + # OpenRouter provider-routing pins: prompt caches live per + # UPSTREAM provider, so a fork without the parent's pins can + # be routed to a different upstream and miss the warm cache + # even with byte-identical prompt/tools bytes. + for _pref_attr in ( + "providers_allowed", + "providers_ignored", + "providers_order", + "provider_sort", + "provider_require_parameters", + "provider_data_collection", + ): + _pref_val = getattr(agent, _pref_attr, None) + if _pref_val: + _fork_kwargs[_pref_attr] = _pref_val + review_agent = AIAgent( + model=_rt.get("model") or agent.model, + max_iterations=max_iterations, + quiet_mode=True, + platform=agent.platform, + provider=_rt.get("provider") or agent.provider, + api_mode=_rt.get("api_mode"), + base_url=_rt.get("base_url") or None, + api_key=_rt.get("api_key") or None, + credential_pool=_rt.get("credential_pool"), + request_overrides=_rt.get("request_overrides") or {}, + parent_session_id=agent.session_id, + enabled_toolsets=getattr(agent, "enabled_toolsets", None), + disabled_toolsets=getattr(agent, "disabled_toolsets", None), + skip_memory=True, + **_fork_kwargs, + ) + review_agent._memory_write_origin = write_origin + review_agent._memory_write_context = write_origin + # The review fork pins the parent's cached system prompt and keeps + # ``tools[]`` byte-identical to the parent so its outbound request + # hits the same provider cache prefix (see the toolset-parity note + # above). The between-turns MCP refresh in build_turn_context would + # add late-connecting MCP tools to this fork and break that parity, + # so opt the review fork out of it. + review_agent._skip_mcp_refresh = True + review_agent._memory_store = agent._memory_store + review_agent._memory_enabled = agent._memory_enabled + review_agent._user_profile_enabled = agent._user_profile_enabled + review_agent._memory_nudge_interval = 0 + review_agent._skill_nudge_interval = 0 + # PERSISTENCE ISOLATION (the curator-takeover root cause): the fork + # shares the parent's session_id (set below, for prompt-cache + # warmth), so without this it would write its harness turn ("Review + # the conversation above and update the skill library…") + its own + # response straight into the user's REAL session in state.db. On the + # user's next live turn the agent re-reads that injected user message + # as a standing instruction and "becomes" the curator, refusing the + # actual task. _persist_disabled hard-stops every DB write/lazy-open + # path (_flush_messages_to_session_db, _ensure_db_session, + # _get_session_db_for_recall); the review writes only to the skill + # and memory stores via its tools, which is all it needs. + review_agent._persist_disabled = True + review_agent._session_db = None + review_agent._session_json_enabled = False + # Suppress all status/warning emits from the fork so the + # user only sees the final successful-action summary. + # Without this, mid-review "Iteration budget exhausted", + # rate-limit retries, compression warnings, and other + # lifecycle messages bubble up through _emit_status -> + # _vprint and leak past the stdout redirect (they go via + # _print_fn/status_callback, which bypass sys.stdout). + review_agent.suppress_status_output = True + # Inherit the parent's cached system prompt verbatim so + # the review fork's outbound HTTP request hits the same + # Anthropic/OpenRouter prefix cache the parent warmed. + # Without this, the fork rebuilds the system prompt from + # scratch (fresh _hermes_now() timestamp, fresh + # session_id, narrower toolset → different skills_prompt) + # and the byte-exact prefix-cache key misses. See + # issue #25322 and PR #17276 for the full analysis + + # measured impact (~26% end-to-end cost reduction on + # Sonnet 4.5). + # Share the parent's warm cached system prompt ONLY when the review + # runs on the SAME model (not routed). When routed to a different + # model the parent's cached prompt is for the wrong model/cache key + # and would miss anyway, so let the routed fork build its own. + if not _routed: + review_agent._cached_system_prompt = agent._cached_system_prompt + # Defensive: pin session_start + session_id to the + # parent's so any code path that re-renders parts of + # the system prompt (compression, plugin hooks) still + # produces byte-identical output. The cached-prompt + # assignment above already short-circuits the normal + # rebuild path, but these pins guarantee parity even + # if a future code path bypasses the cache. + review_agent.session_start = agent.session_start + review_agent.session_id = agent.session_id + # The fork shares the parent's live session_id (pinned above for + # prefix-cache parity). It is single-lifecycle and calls close() + # right after this run_conversation(); without opting out, close() + # would finalize the parent's still-active session row mid + # conversation (the review fires every ~10 turns). Leave session + # finalization to the real owner (CLI close / gateway reset / cron). + review_agent._end_session_on_close = False + # DETACHED IN-MEMORY COMPACTION (issue #93057). The fork shares + # the parent's session_id (pinned above for prefix-cache parity), + # so the historical guard here was ``compression_enabled = False``: + # if the fork ran the ordinary compression path it could rotate / + # archive the parent's live session — the sibling-session race + # behind #38727. But disabling compaction was a proxy for + # detachment, and it removed the ONLY bound on the review's + # private snapshot: as the review performs tool calls, every + # follow-up provider request replayed the snapshot plus the + # growing review tool loop (350k-384k input tokens per request in + # production, 1.49M total across one 8-request review). + # + # The fix is detachment, not disablement: + # • Persistence is already off above (_persist_disabled / + # _session_db=None), so the commit site in compress_context + # (``if agent._session_db:``) skips every durable write and + # compaction can only ever rewrite the fork's private + # in-memory transcript. + # • The compressor's OWN session binding still needs severing: + # AIAgent.__init__ bound it to the parent's SessionDB and + # session_id before this function nulled the agent-level + # binding, so durable cooldown/streak/ineffective-count + # writes would otherwise land on the parent's row. Rebinding + # with session_db=None / session_id="" makes every + # compressor persist guard a no-op. + # • Force in-place mode (never rotation) even if the parent's + # config selected rotation, and re-enable compression ONLY + # after the rebind succeeds (fail-closed — see below). While + # enabled, both compression gates stay deferred until the + # fork's first provider response so request #1 replays the + # full snapshot as a warm cache read. + _review_compressor = getattr(review_agent, "context_compressor", None) + _bind_review_compressor = getattr( + _review_compressor, "bind_session_state", None + ) + _review_compression_detached = False + if callable(_bind_review_compressor): + try: + # Plugin/third-party context engines may not accept these + # kwargs; they own their own persistence policy, so a + # failed rebind leaves the pre-existing flags in place + # and must never abort the review (same tolerance as the + # init-time binding in agent_init.py). + _bind_review_compressor(session_db=None, session_id="") + _review_compression_detached = True + except Exception: + # FAIL-CLOSED (adversarial review, #93057): if the rebind + # could not sever the engine's session binding, the + # compressor may still point at the parent's + # SessionDB/session_id. Enabling compression in that + # state would let durable cooldown/streak/ineffective- + # count writes land on the parent's row and re-open the + # #38727 sibling race. Keep the historical + # compression_enabled=False behavior instead and warn; + # the review still runs, bounded by the iteration cap + # and the aggregate input budget below. + logger.warning( + "background-review compressor detachment failed; " + "keeping compression DISABLED on this review fork " + "(fail-closed, issue #93057 / #38727)", + exc_info=True, + ) + # Force in-place mode (never rotation) even if the parent's + # config selected rotation. Re-enable compression ONLY after the + # compressor's session binding was successfully severed; an + # engine without a bind hook keeps the historical disabled + # behavior as well. + review_agent.compression_in_place = True + review_agent.compression_enabled = _review_compression_detached + if _review_compression_detached: + # Warm-cache parity: the fork's FIRST provider request + # replays the parent's full snapshot as a warm prompt-cache + # read, so compaction must not rewrite the snapshot before + # that first request goes out. Defer both compression gates + # until the first provider response arrives (see + # _review_fork_first_request_pending in agent/turn_context.py + # and the pre-API gate in agent/conversation_loop.py); from + # the second request on, the fork's transcript is its own and + # compaction bounds it. + review_agent._review_defer_compaction_before_first_response = True + # Aggregate input budget: compaction bounds any single request; + # this bounds the WHOLE review. Iterations are already capped by + # _REVIEW_MAX_ITERATIONS. Checked in agent/conversation_loop.py + # via _review_input_budget_exhausted (issue #93057). + review_agent._review_input_token_budget = _review_input_token_budget( + task_cfg + ) + return review_agent, _rt, _routed + + def _run_review_in_thread( agent: Any, messages_snapshot: List[Dict], @@ -1207,266 +1500,8 @@ def _run_review_in_thread( # thread's writes to devnull and leaves all other threads on the real # streams. with thread_scoped_silence(): - # Inherit the parent agent's live runtime (provider, model, - # base_url, api_key, api_mode) so the fork uses the exact - # same credentials the main turn is using. Without this, - # AIAgent.__init__ re-runs auto-resolution from env vars, - # which fails for OAuth-only providers, session-scoped - # creds, or credential-pool setups where the resolver can't - # reconstruct auth from scratch -- producing the spurious - # "No LLM provider configured" warning at end of turn. - # _resolve_review_runtime() returns the parent's live runtime by - # default (routed=False; main model, warm cache), or — when the user - # set auxiliary.background_review.{provider,model} to a different - # model — that model's runtime (routed=True). The codex_app_server - # -> codex_responses downgrade is applied inside the resolver. - _rt = _resolve_review_runtime(agent, task_cfg) - _routed = bool(_rt.get("routed")) - # skip_memory=True keeps the review fork from - # touching external memory plugins (honcho, mem0, - # supermemory, etc.). Without it, the fork's - # __init__ rebuilds its own _memory_manager from - # config, scoped to the parent's session_id, and - # run_conversation() then leaks the harness prompt - # into the user's real memory namespace via three - # ingestion sites: on_turn_start (cadence + turn - # message), prefetch_all (recall query), and - # sync_all (harness prompt + review output recorded - # as a (user, assistant) turn pair). Built-in - # MEMORY.md / USER.md state is re-bound from the - # parent below so memory(action="add") writes from - # the review still land on disk; the review just - # has zero side effects on external providers. - # Match parent's toolset config so ``tools[]`` is byte-identical - # in the request body — Anthropic's cache key includes it. - # (The runtime whitelist below still restricts dispatch.) - _fork_kwargs: Dict[str, Any] = {} - if isinstance(_rt.get("max_tokens"), int): - _fork_kwargs["max_tokens"] = _rt["max_tokens"] - if isinstance(_rt.get("command"), str) and _rt["command"]: - _fork_kwargs["acp_command"] = _rt["command"] - _fork_kwargs["acp_args"] = _rt.get("args") or [] - # Match parent's reasoning config so the fork's ``thinking`` / - # ``output_config`` are byte-identical in the request body — - # Anthropic's cache key is namespaced by ``thinking`` presence. - # Same-model path only: when routed to a different aux model the - # cache is cold regardless (parity buys nothing) and the parent's - # effort vocabulary may not be valid for the routed model/provider - # (e.g. OpenRouter ``extra_body.reasoning.effort`` is forwarded - # unclamped; codex_responses passes ``max``/``ultra`` through - # unmapped except on gpt-5.6/xAI). Let the routed fork use - # provider defaults — matching the ``not _routed`` gate on - # _cached_system_prompt below. - if not _routed: - _fork_kwargs["reasoning_config"] = getattr(agent, "reasoning_config", None) - # Gateway session context is appended to the parent's cached - # system prompt at API-call time through this field. Preserve - # it on same-model forks so the complete effective system - # prompt remains byte-identical and can reuse the warm prefix. - _fork_kwargs["ephemeral_system_prompt"] = getattr( - agent, "ephemeral_system_prompt", None - ) - # Prefill messages are inserted immediately after the system - # message at API-call time (chat_completion_helpers.py / - # conversation_loop.py), so a parent with prefill configured - # (gateway prefill_messages_file) would otherwise diverge - # from the warm prefix at message index 1 — same bug class - # as the ephemeral prompt above, one position later. - # Deep copy: the unicode-error recovery path mutates - # prefill entries IN PLACE (_sanitize_messages_surrogates - # via conversation_loop), so sharing dicts would let a - # fork-side sanitize rewrite the parent's prefill bytes. - _parent_prefill = copy.deepcopy( - getattr(agent, "prefill_messages", None) or [] - ) - if _parent_prefill: - _fork_kwargs["prefill_messages"] = _parent_prefill - # OpenRouter provider-routing pins: prompt caches live per - # UPSTREAM provider, so a fork without the parent's pins can - # be routed to a different upstream and miss the warm cache - # even with byte-identical prompt/tools bytes. - for _pref_attr in ( - "providers_allowed", - "providers_ignored", - "providers_order", - "provider_sort", - "provider_require_parameters", - "provider_data_collection", - ): - _pref_val = getattr(agent, _pref_attr, None) - if _pref_val: - _fork_kwargs[_pref_attr] = _pref_val - review_agent = AIAgent( - model=_rt.get("model") or agent.model, - max_iterations=_REVIEW_MAX_ITERATIONS, - quiet_mode=True, - platform=agent.platform, - provider=_rt.get("provider") or agent.provider, - api_mode=_rt.get("api_mode"), - base_url=_rt.get("base_url") or None, - api_key=_rt.get("api_key") or None, - credential_pool=_rt.get("credential_pool"), - request_overrides=_rt.get("request_overrides") or {}, - parent_session_id=agent.session_id, - enabled_toolsets=getattr(agent, "enabled_toolsets", None), - disabled_toolsets=getattr(agent, "disabled_toolsets", None), - skip_memory=True, - **_fork_kwargs, - ) - review_agent._memory_write_origin = "background_review" - review_agent._memory_write_context = "background_review" - # The review fork pins the parent's cached system prompt and keeps - # ``tools[]`` byte-identical to the parent so its outbound request - # hits the same provider cache prefix (see the toolset-parity note - # above). The between-turns MCP refresh in build_turn_context would - # add late-connecting MCP tools to this fork and break that parity, - # so opt the review fork out of it. - review_agent._skip_mcp_refresh = True - review_agent._memory_store = agent._memory_store - review_agent._memory_enabled = agent._memory_enabled - review_agent._user_profile_enabled = agent._user_profile_enabled - review_agent._memory_nudge_interval = 0 - review_agent._skill_nudge_interval = 0 - # PERSISTENCE ISOLATION (the curator-takeover root cause): the fork - # shares the parent's session_id (set below, for prompt-cache - # warmth), so without this it would write its harness turn ("Review - # the conversation above and update the skill library…") + its own - # response straight into the user's REAL session in state.db. On the - # user's next live turn the agent re-reads that injected user message - # as a standing instruction and "becomes" the curator, refusing the - # actual task. _persist_disabled hard-stops every DB write/lazy-open - # path (_flush_messages_to_session_db, _ensure_db_session, - # _get_session_db_for_recall); the review writes only to the skill - # and memory stores via its tools, which is all it needs. - review_agent._persist_disabled = True - review_agent._session_db = None - review_agent._session_json_enabled = False - # Suppress all status/warning emits from the fork so the - # user only sees the final successful-action summary. - # Without this, mid-review "Iteration budget exhausted", - # rate-limit retries, compression warnings, and other - # lifecycle messages bubble up through _emit_status -> - # _vprint and leak past the stdout redirect (they go via - # _print_fn/status_callback, which bypass sys.stdout). - review_agent.suppress_status_output = True - # Inherit the parent's cached system prompt verbatim so - # the review fork's outbound HTTP request hits the same - # Anthropic/OpenRouter prefix cache the parent warmed. - # Without this, the fork rebuilds the system prompt from - # scratch (fresh _hermes_now() timestamp, fresh - # session_id, narrower toolset → different skills_prompt) - # and the byte-exact prefix-cache key misses. See - # issue #25322 and PR #17276 for the full analysis + - # measured impact (~26% end-to-end cost reduction on - # Sonnet 4.5). - # Share the parent's warm cached system prompt ONLY when the review - # runs on the SAME model (not routed). When routed to a different - # model the parent's cached prompt is for the wrong model/cache key - # and would miss anyway, so let the routed fork build its own. - if not _routed: - review_agent._cached_system_prompt = agent._cached_system_prompt - # Defensive: pin session_start + session_id to the - # parent's so any code path that re-renders parts of - # the system prompt (compression, plugin hooks) still - # produces byte-identical output. The cached-prompt - # assignment above already short-circuits the normal - # rebuild path, but these pins guarantee parity even - # if a future code path bypasses the cache. - review_agent.session_start = agent.session_start - review_agent.session_id = agent.session_id - # The fork shares the parent's live session_id (pinned above for - # prefix-cache parity). It is single-lifecycle and calls close() - # right after this run_conversation(); without opting out, close() - # would finalize the parent's still-active session row mid - # conversation (the review fires every ~10 turns). Leave session - # finalization to the real owner (CLI close / gateway reset / cron). - review_agent._end_session_on_close = False - # DETACHED IN-MEMORY COMPACTION (issue #93057). The fork shares - # the parent's session_id (pinned above for prefix-cache parity), - # so the historical guard here was ``compression_enabled = False``: - # if the fork ran the ordinary compression path it could rotate / - # archive the parent's live session — the sibling-session race - # behind #38727. But disabling compaction was a proxy for - # detachment, and it removed the ONLY bound on the review's - # private snapshot: as the review performs tool calls, every - # follow-up provider request replayed the snapshot plus the - # growing review tool loop (350k-384k input tokens per request in - # production, 1.49M total across one 8-request review). - # - # The fix is detachment, not disablement: - # • Persistence is already off above (_persist_disabled / - # _session_db=None), so the commit site in compress_context - # (``if agent._session_db:``) skips every durable write and - # compaction can only ever rewrite the fork's private - # in-memory transcript. - # • The compressor's OWN session binding still needs severing: - # AIAgent.__init__ bound it to the parent's SessionDB and - # session_id before this function nulled the agent-level - # binding, so durable cooldown/streak/ineffective-count - # writes would otherwise land on the parent's row. Rebinding - # with session_db=None / session_id="" makes every - # compressor persist guard a no-op. - # • Force in-place mode (never rotation) even if the parent's - # config selected rotation, and re-enable compression ONLY - # after the rebind succeeds (fail-closed — see below). While - # enabled, both compression gates stay deferred until the - # fork's first provider response so request #1 replays the - # full snapshot as a warm cache read. - _review_compressor = getattr(review_agent, "context_compressor", None) - _bind_review_compressor = getattr( - _review_compressor, "bind_session_state", None - ) - _review_compression_detached = False - if callable(_bind_review_compressor): - try: - # Plugin/third-party context engines may not accept these - # kwargs; they own their own persistence policy, so a - # failed rebind leaves the pre-existing flags in place - # and must never abort the review (same tolerance as the - # init-time binding in agent_init.py). - _bind_review_compressor(session_db=None, session_id="") - _review_compression_detached = True - except Exception: - # FAIL-CLOSED (adversarial review, #93057): if the rebind - # could not sever the engine's session binding, the - # compressor may still point at the parent's - # SessionDB/session_id. Enabling compression in that - # state would let durable cooldown/streak/ineffective- - # count writes land on the parent's row and re-open the - # #38727 sibling race. Keep the historical - # compression_enabled=False behavior instead and warn; - # the review still runs, bounded by the iteration cap - # and the aggregate input budget below. - logger.warning( - "background-review compressor detachment failed; " - "keeping compression DISABLED on this review fork " - "(fail-closed, issue #93057 / #38727)", - exc_info=True, - ) - # Force in-place mode (never rotation) even if the parent's - # config selected rotation. Re-enable compression ONLY after the - # compressor's session binding was successfully severed; an - # engine without a bind hook keeps the historical disabled - # behavior as well. - review_agent.compression_in_place = True - review_agent.compression_enabled = _review_compression_detached - if _review_compression_detached: - # Warm-cache parity: the fork's FIRST provider request - # replays the parent's full snapshot as a warm prompt-cache - # read, so compaction must not rewrite the snapshot before - # that first request goes out. Defer both compression gates - # until the first provider response arrives (see - # _review_fork_first_request_pending in agent/turn_context.py - # and the pre-API gate in agent/conversation_loop.py); from - # the second request on, the fork's transcript is its own and - # compaction bounds it. - review_agent._review_defer_compaction_before_first_response = True - # Aggregate input budget: compaction bounds any single request; - # this bounds the WHOLE review. Iterations are already capped by - # _REVIEW_MAX_ITERATIONS. Checked in agent/conversation_loop.py - # via _review_input_budget_exhausted (issue #93057). - review_agent._review_input_token_budget = _review_input_token_budget( - task_cfg + review_agent, _rt, _routed = build_cache_parity_fork( + agent, task_cfg, max_iterations=_REVIEW_MAX_ITERATIONS ) # Register this fork on the PARENT's _active_children (the same diff --git a/agent/side_question.py b/agent/side_question.py index 79bc7fce3f..b2ff082191 100644 --- a/agent/side_question.py +++ b/agent/side_question.py @@ -1,19 +1,31 @@ """Context-aware side questions (``/btw``). ``/btw `` answers a quick question ABOUT the current conversation -without interrupting it: a one-shot auxiliary LLM call receives a read-only -transcript snapshot plus the question, and the answer is delivered alongside -the live session. The live conversation history is never touched — no +without interrupting it. The live conversation history is never touched — no synthetic turns, no role-alternation risk, no prompt-cache invalidation. -This is deliberately different from ``/bg`` (``/background``'s successor), -which spawns a fresh, contextless agent session for independent work. +Two execution paths, picked automatically: -Model selection rides the standard auxiliary plumbing -(:func:`agent.auxiliary_client.call_llm` via :func:`agent.oneshot.run_oneshot`): -pass ``main_runtime`` to inherit the live session's provider/model; users can -override per-task via ``auxiliary.side_question.provider`` / ``.model`` in -config.yaml. +* **Cache-parity fork (preferred).** When a live parent ``AIAgent`` is + available, the answer comes from a detached fork built by + :func:`agent.background_review.build_cache_parity_fork` — the exact + mechanism the self-improvement background review uses. The fork inherits + the parent's runtime, byte-identical system prompt / ``tools[]`` / + reasoning config, and shared ``session_id``, then replays the parent's + message snapshot verbatim. The provider prefix cache is already warm for + that entire replay, so the fork sees the FULL untruncated conversation at + cache-read prices. Tool calls are denied at dispatch (thread whitelist), + persistence is fully detached, and usage is attributed to the parent. + +* **One-shot digest (fallback).** When no live parent exists (e.g. the + gateway evicted the session's cached agent — the provider cache is cold + there anyway), a rendered plain-text transcript snapshot is sent through + one auxiliary :func:`agent.oneshot.run_oneshot` call. + +Model selection rides the standard auxiliary plumbing: main model by +default; users can override per-task via ``auxiliary.side_question.provider`` +/ ``.model`` in config.yaml (an override routes the fork to that model and +replays a compact digest, since the cache is cold on a different model). """ import logging @@ -25,14 +37,29 @@ logger = logging.getLogger(__name__) # config.yaml, falls back main-model-first like every other aux task. SIDE_QUESTION_TASK = "side_question" -# Per-message and total character budgets for the transcript snapshot. The -# snapshot is rendered to plain text (never replayed as raw provider messages) -# so assistant tool_calls entries can't trip provider-side validation on a -# tools-less one-shot request. +# Fork path: the model may waste an iteration attempting a (denied) tool +# call before answering in text; give it a little headroom. +_FORK_MAX_ITERATIONS = 3 + +# Fallback one-shot path: per-message and total character budgets for the +# rendered transcript snapshot. _PER_MESSAGE_CHAR_CAP = 2000 _TRANSCRIPT_CHAR_BUDGET = 24000 -_INSTRUCTIONS = ( +_FORK_PROMPT = ( + "The user asked a quick SIDE question with /btw while the main work " + "continues in the original session.\n" + "Rules:\n" + "- Answer ONLY the side question, using the conversation above as " + "context. Do not continue, redo, or critique the main task.\n" + "- Do NOT call any tools — they are disabled for this side question. " + "Answer directly in text.\n" + "- If the conversation does not contain enough information to answer, " + "say so plainly instead of guessing.\n" + "- Be concise and direct." +) + +_ONESHOT_INSTRUCTIONS = ( "You are the same AI assistant that is currently working inside the " "conversation transcribed below. The user has asked a quick SIDE question " "with /btw while the main work continues.\n" @@ -63,16 +90,40 @@ def _msg_text(msg: Dict[str, Any]) -> str: return "" +def trim_snapshot_for_fork(history: Optional[List[Dict[str, Any]]]) -> List[Dict[str, Any]]: + """Trim a possibly mid-turn snapshot so appending a user message is valid. + + A /btw issued while a turn is running can snapshot the transcript in the + middle of a tool loop — ending on an assistant message with unresolved + ``tool_calls``, a tool result, or the in-flight user message. Appending + the side question after any of those would violate role alternation on + strict providers. Drop trailing messages until the snapshot ends with a + completed assistant text message. Trimming only the TAIL preserves the + warm prefix-cache property of everything kept. + """ + msgs = list(history or []) + while msgs: + last = msgs[-1] + if not isinstance(last, dict): + msgs.pop() + continue + role = last.get("role") + if role == "assistant" and not last.get("tool_calls"): + break + msgs.pop() + return msgs + + def render_history_for_side_question( history: Optional[List[Dict[str, Any]]], char_budget: int = _TRANSCRIPT_CHAR_BUDGET, ) -> str: """Render a conversation snapshot as a plain-text transcript. - Keeps the most recent messages that fit ``char_budget``, newest-biased - (older context is what gets dropped). Tool calls are summarized by name; - tool results are included truncated so "what did that command output" - style questions remain answerable. + Fallback path only. Keeps the most recent messages that fit + ``char_budget``, newest-biased (older context is what gets dropped). + Tool calls are summarized by name; tool results are included truncated + so "what did that command output" style questions remain answerable. """ lines: List[str] = [] for msg in history or []: @@ -119,7 +170,90 @@ def render_history_for_side_question( return prefix + "\n".join(kept) -def answer_side_question( +def _side_question_task_config() -> Dict[str, Any]: + """Return ``auxiliary.side_question`` from config (or ``{}``).""" + try: + from hermes_cli.config import load_config_readonly + + cfg = load_config_readonly() + except Exception: + return {} + aux = cfg.get("auxiliary", {}) if isinstance(cfg.get("auxiliary"), dict) else {} + task = aux.get(SIDE_QUESTION_TASK, {}) + return task if isinstance(task, dict) else {} + + +def _answer_via_fork( + parent_agent: Any, + question: str, + history: Optional[List[Dict[str, Any]]], +) -> str: + """Answer via a cache-parity fork of ``parent_agent``. + + Runs synchronously on the CALLING thread (all /btw surfaces invoke this + from a worker thread). The thread-scoped tool whitelist is emptied so + any tool call the fork attempts is denied at dispatch — the request's + ``tools[]`` stays byte-identical to the parent's for cache parity, but + the side question can never mutate anything. + """ + from agent.background_review import ( + _digest_history, + _record_review_usage_to_parent, + _snapshot_review_usage, + build_cache_parity_fork, + ) + from hermes_cli.plugins import ( + clear_thread_tool_whitelist, + set_thread_tool_whitelist, + ) + + task_cfg = _side_question_task_config() + fork, _rt, routed = build_cache_parity_fork( + parent_agent, + task_cfg, + max_iterations=_FORK_MAX_ITERATIONS, + write_origin="side_question", + ) + try: + set_thread_tool_whitelist( + set(), + deny_msg_fmt=( + "Side question (/btw) denied tool call: {tool_name}. " + "Tools are disabled here — answer directly from the " + "conversation context." + ), + ) + snapshot = trim_snapshot_for_fork(history) + replay = _digest_history(snapshot) if routed else snapshot + result = fork.run_conversation( + user_message=f"{_FORK_PROMPT}\n\nSide question: {question}", + conversation_history=replay, + ) + answer = (result or {}).get("final_response", "") or "" + if not answer and result and result.get("error"): + raise RuntimeError(str(result["error"])) + return answer.strip() + finally: + clear_thread_tool_whitelist() + # Attribute the fork's token usage to the parent session (same + # pattern as the background review, issue #87250). Best-effort. + try: + _record_review_usage_to_parent( + parent_agent, _snapshot_review_usage(fork) + ) + except Exception: + pass + try: + fork.shutdown_memory_provider() + except Exception: + pass + try: + fork.close() + except Exception: + pass + + +def _answer_via_oneshot( question: str, history: Optional[List[Dict[str, Any]]], *, @@ -128,18 +262,9 @@ def answer_side_question( temperature: Optional[float] = 0.3, timeout: float = 180.0, ) -> str: - """Answer ``question`` against a snapshot of ``history``. - - Returns the model's text answer. Raises whatever the auxiliary client - raises (RuntimeError on no provider, etc.) — callers surface the error - on their own UI. - """ + """Fallback: answer from a rendered transcript digest in one aux call.""" from agent.oneshot import run_oneshot - question = (question or "").strip() - if not question: - raise ValueError("answer_side_question requires a non-empty question") - transcript = render_history_for_side_question(history) user_input = ( "Conversation transcript (snapshot):\n" @@ -149,7 +274,7 @@ def answer_side_question( f"Side question: {question}" ) return run_oneshot( - instructions=_INSTRUCTIONS, + instructions=_ONESHOT_INSTRUCTIONS, user_input=user_input, task=SIDE_QUESTION_TASK, max_tokens=max_tokens, @@ -157,3 +282,48 @@ def answer_side_question( timeout=timeout, main_runtime=main_runtime, ) + + +def answer_side_question( + question: str, + history: Optional[List[Dict[str, Any]]], + *, + parent_agent: Any = None, + main_runtime: Optional[Dict[str, Any]] = None, + max_tokens: int = 2048, + temperature: Optional[float] = 0.3, + timeout: float = 180.0, +) -> str: + """Answer ``question`` against a snapshot of ``history``. + + When ``parent_agent`` is a live ``AIAgent``, the answer comes from a + cache-parity fork replaying the full snapshot against the warm provider + prefix cache (see module docstring). Otherwise a one-shot digest call is + used. Raises on failure — callers surface the error on their own UI. + """ + question = (question or "").strip() + if not question: + raise ValueError("answer_side_question requires a non-empty question") + + if parent_agent is not None: + try: + answer = _answer_via_fork(parent_agent, question, history) + if answer: + return answer + logger.warning( + "/btw fork returned an empty answer; falling back to one-shot" + ) + except Exception: + logger.warning( + "/btw cache-parity fork failed; falling back to one-shot", + exc_info=True, + ) + + return _answer_via_oneshot( + question, + history, + main_runtime=main_runtime, + max_tokens=max_tokens, + temperature=temperature, + timeout=timeout, + ) diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 12548e5ac0..6aa7de68a8 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -3684,6 +3684,25 @@ class GatewaySlashCommandsMixin: "api_mode": runtime_kwargs.get("api_mode"), } history_snapshot = list(history) + # Prefer the cache-parity fork when this chat has a live cached + # AIAgent: the fork replays the snapshot against the warm provider + # prefix cache (same mechanism as the background self-improvement + # review), giving the side answer FULL conversation context at + # cache-read prices. If no cached agent exists (evicted / first + # message), the provider cache is cold anyway — the one-shot digest + # fallback inside answer_side_question handles it. + parent_agent = None + try: + session_key = self._session_key_for_source(source) + _cache_lock = getattr(self, "_agent_cache_lock", None) + if _cache_lock is not None: + with _cache_lock: + _cached = self._agent_cache.get(session_key) + parent_agent = ( + _cached[0] if isinstance(_cached, tuple) else _cached + ) or None + except Exception: + parent_agent = None event_message_id = self._reply_anchor_for_event(event) _thread_metadata = self._thread_metadata_for_source(source, event_message_id) adapter = self._adapter_for_source(source) @@ -3696,6 +3715,7 @@ class GatewaySlashCommandsMixin: answer_side_question, question, history_snapshot, + parent_agent=parent_agent, main_runtime=main_runtime, ) except Exception as e: diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 9d41538fd6..1fe2223090 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -2386,6 +2386,8 @@ class CLICommandsMixin: # Snapshot NOW, on the UI thread — the foreground turn keeps appending # to conversation_history while the worker runs. history_snapshot = list(self.conversation_history or []) + # Live agent → cache-parity fork (full context, warm cache reads). + parent_agent = self.agent turn_route = self._resolve_turn_agent_config(question) main_runtime = { "model": turn_route["model"], @@ -2405,6 +2407,7 @@ class CLICommandsMixin: answer = answer_side_question( question, history_snapshot, + parent_agent=parent_agent, main_runtime=main_runtime, ) if self._app: diff --git a/tests/agent/test_side_question.py b/tests/agent/test_side_question.py index 4bcc4c3c1f..cc82cfcc06 100644 --- a/tests/agent/test_side_question.py +++ b/tests/agent/test_side_question.py @@ -6,6 +6,7 @@ from agent.side_question import ( SIDE_QUESTION_TASK, answer_side_question, render_history_for_side_question, + trim_snapshot_for_fork, ) @@ -90,3 +91,109 @@ class TestAnswerSideQuestion: assert "Side question: which file had the error?" in captured["user_input"] # The instructions steer the model to answer only the side question. assert "side" in captured["instructions"].lower() + + +class TestTrimSnapshotForFork: + def test_trims_unresolved_tool_loop_tail(self): + history = [ + {"role": "user", "content": "u1"}, + {"role": "assistant", "content": "done first task"}, + {"role": "user", "content": "u2 (in-flight)"}, + {"role": "assistant", "content": "", "tool_calls": [{"id": "1"}]}, + {"role": "tool", "content": "result"}, + ] + trimmed = trim_snapshot_for_fork(history) + assert trimmed[-1] == {"role": "assistant", "content": "done first task"} + assert len(trimmed) == 2 + + def test_keeps_completed_history(self): + history = [ + {"role": "user", "content": "u1"}, + {"role": "assistant", "content": "a1"}, + ] + assert trim_snapshot_for_fork(history) == history + + def test_empty_when_no_completed_assistant(self): + history = [{"role": "user", "content": "first message, turn running"}] + assert trim_snapshot_for_fork(history) == [] + + +class TestForkPath: + def test_prefers_fork_when_parent_agent_given(self): + seen = {} + + def fake_fork(parent, question, history): + seen["parent"] = parent + seen["question"] = question + return "fork answer" + + parent = object() + with patch("agent.side_question._answer_via_fork", side_effect=fake_fork), \ + patch("agent.side_question._answer_via_oneshot") as oneshot: + out = answer_side_question("q?", [], parent_agent=parent) + assert out == "fork answer" + assert seen["parent"] is parent + oneshot.assert_not_called() + + def test_falls_back_to_oneshot_when_fork_fails(self): + with patch( + "agent.side_question._answer_via_fork", + side_effect=RuntimeError("boom"), + ), patch( + "agent.side_question._answer_via_oneshot", return_value="digest answer" + ) as oneshot: + out = answer_side_question("q?", [], parent_agent=object()) + assert out == "digest answer" + oneshot.assert_called_once() + + def test_no_parent_agent_uses_oneshot(self): + with patch("agent.side_question._answer_via_fork") as fork, patch( + "agent.side_question._answer_via_oneshot", return_value="digest" + ): + out = answer_side_question("q?", []) + assert out == "digest" + fork.assert_not_called() + + def test_fork_denies_tools_and_replays_snapshot(self): + """_answer_via_fork wires the empty whitelist, replays the trimmed + snapshot, runs the fork, attributes usage, and tears down.""" + from agent.side_question import _answer_via_fork + + calls = {} + + class FakeFork: + def run_conversation(self, user_message, conversation_history): + calls["user_message"] = user_message + calls["history"] = conversation_history + return {"final_response": "it was foo.py"} + + def shutdown_memory_provider(self): + calls["shutdown"] = True + + def close(self): + calls["closed"] = True + + def fake_build(parent, task_cfg, *, max_iterations, write_origin): + calls["write_origin"] = write_origin + return FakeFork(), {"model": "m"}, False + + whitelists = [] + + history = [ + {"role": "user", "content": "fix foo.py"}, + {"role": "assistant", "content": "fixed"}, + ] + with patch("agent.background_review.build_cache_parity_fork", fake_build), \ + patch("hermes_cli.plugins.set_thread_tool_whitelist", + side_effect=lambda allowed, **kw: whitelists.append(allowed)), \ + patch("hermes_cli.plugins.clear_thread_tool_whitelist"), \ + patch("agent.background_review._snapshot_review_usage", return_value={}), \ + patch("agent.background_review._record_review_usage_to_parent"): + answer = _answer_via_fork(object(), "which file?", history) + + assert answer == "it was foo.py" + assert whitelists == [set()] # every tool denied at dispatch + assert calls["history"] == history # full snapshot replayed verbatim + assert "which file?" in calls["user_message"] + assert calls["write_origin"] == "side_question" + assert calls.get("shutdown") and calls.get("closed") diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index 43a80ca715..34e38aacf9 100644 --- a/tui_gateway/methods_prompt.py +++ b/tui_gateway/methods_prompt.py @@ -1427,6 +1427,7 @@ def _(rid, params: dict) -> dict: answer = answer_side_question( text, snapshot, + parent_agent=agent, main_runtime=main_runtime, ) finally: