ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

feat: /btw rides the background-review cache-parity fork for full-context answers

Browse Source 
The initial /btw implementation (#97937) answered from a rendered
plain-text transcript digest — truncated context, cold-written tokens on
every question. Teknium's call: reuse the self-improvement review fork
instead, which keeps the entire prompt cache stable for the fork and
gives it the complete conversation for very cheap.

- agent/background_review.py: extract the review-fork construction into
  build_cache_parity_fork() — same runtime/credentials as the parent,
  byte-identical system prompt / tools[] / reasoning config on the
  same-model path, shared session_id for prefix warmth, full persistence
  detachment (no state.db writes, no rotation, no external memory,
  in-place-only compaction). The review thread now calls the helper;
  behavior unchanged (full review test suite green).
- agent/side_question.py: /btw prefers the fork when a live parent
  AIAgent exists — replays the untruncated snapshot as warm cache reads,
  denies every tool at dispatch via an empty thread whitelist (tools[]
  stays byte-identical for cache parity), attributes usage to the parent,
  and trims a mid-turn snapshot tail so role alternation holds. The
  one-shot digest remains as fallback (no live agent = cold cache anyway,
  and any fork failure degrades gracefully).
- CLI passes self.agent, TUI passes the session agent, gateway looks up
  the chat's cached agent (parity with how turns reuse it).

Live-verified: /btw on the worktree runs the fork path (agent.log shows
the side question as a forked conversation turn on the parent session_id
with the full history replayed), answers correctly from context.
This commit is contained in:
Teknium
2026-08-29 08:14:34 -07:00
parent 360761c8cf
commit 578f85cfb0
6 changed files with 628 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/background_review.py
+296 -261
View File
@@ -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
agent/side_question.py
+201 -31
View File
@@ -1,19 +1,31 @@
"""Context-aware side questions (``/btw``).
``/btw <question>`` 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,
)
gateway/slash_commands.py
+20
View File
@@ -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:
hermes_cli/cli_commands_mixin.py
+3
View File
@@ -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:
tests/agent/test_side_question.py
+107
View File
@@ -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")
tui_gateway/methods_prompt.py
+1
View File
@@ -1427,6 +1427,7 @@ def _(rid, params: dict) -> dict:
answer = answer_side_question(
text,
snapshot,
parent_agent=agent,
main_runtime=main_runtime,
)
finally: