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

simplify(agent): surface switch — reuse flatten_message_text / agent_tool_names / one runtime-boundary split

Browse Source 
- _transcript_row_texts re-implemented agent.message_content.flatten_message_text
  and the api_content sidecar rule; the note can only land on a user row,
  so the transcript scan now skips assistant/tool rows (the bulk of the bytes).
- Three sites computed "names of agent.tools"; tools.mcp_tool_agent gains
  agent_tool_names() used by the switch note and conversation_loop, which
  also stops importing the private _def_name across modules. The name list
  is only captured when a switch was announced.
- split_runtime_boundary() is the single owner of the runtime-block
  rpartition/END check for both identity_line_value and
  _stored_prompt_matches_runtime.
- platform_surface_hint was a public alias of _platform_hint; the function is
  now platform_hint (its docstring pointed at the pre-move module).
- consume_gateway_turn_context_notes and consume_surface_switch_note share
  _pop_turn_note so the two one-shot channels have identical semantics.
- platform check hoisted above the transcript scan.
This commit is contained in:
kshitijk4poor
2026-09-08 20:46:06 +05:30
committed by kshitij
parent 6cc177a76c
commit defdf64790
5 changed files with 60 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
agent/conversation_loop.py
View File

@@ -29,7 +29,9 @@ from agent.prompt_caching import (
strip_anthropic_tool_cache_control,
)
from agent.runtime_cwd import resolve_agent_cwd
from agent.surface_switch import identity_line_value, note_inert_pinned_tools, stage_surface_switch_note
from agent.surface_switch import (
identity_line_value, note_inert_pinned_tools, split_runtime_boundary, stage_surface_switch_note,
)
from agent.turn_context import PreflightCompressionTimedOut, build_turn_context
from agent.turn_retry_state import TurnRetryState
# Phase helpers of the turn loop, bound at import so a source-tree swap cannot load a
@@ -705,8 +707,9 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history)
try:
saved_tools = session_row.get("tool_names") if session_row else None
if saved_tools:
from tools.mcp_tool_agent import _def_name, restore_agent_tool_prefix
built_for_this_surface = [_def_name(t) for t in agent.tools or []]
from tools.mcp_tool_agent import agent_tool_names, restore_agent_tool_prefix
# Captured BEFORE the pin merges the previous surface's tools back in.
built_for_this_surface = agent_tool_names(agent) if announced_switch else []
restore_agent_tool_prefix(agent, json.loads(saved_tools))
if announced_switch:
note_inert_pinned_tools(agent, built_for_this_surface)
@@ -777,9 +780,7 @@ def _restore_or_build_system_prompt(agent, system_message, conversation_history)
def _stored_prompt_matches_runtime(agent, prompt: str) -> bool:
"""Return False when the persisted runtime-identity lines are stale."""
identity, runtime_marker, runtime = prompt.rpartition(f"\n\n{RUNTIME_ENVIRONMENT_HEADING}\n\n")
# Legacy prose may quote the heading, but only the new renderer ends in this boundary.
runtime_marker = runtime_marker if prompt.endswith(RUNTIME_ENVIRONMENT_END) else ""
_identity, runtime_marker, runtime = split_runtime_boundary(prompt)
def host_info_value(label: str) -> str:
"""New prompts delimit runtime hints; legacy prompts put them before context."""

74
agent/surface_switch.py
View File

@@ -12,8 +12,9 @@ converges at the next rebuild boundary (compaction).
from __future__ import annotations
import logging
from typing import Any, Iterator, List
from typing import Any, List
from agent.message_content import flatten_message_text
from agent.prompt_builder import RUNTIME_ENVIRONMENT_END, RUNTIME_ENVIRONMENT_HEADING
logger = logging.getLogger("run_agent")
@@ -27,41 +28,23 @@ _SURFACE_NAME_END = " — any earlier interface guidance"
_NOTE_SCAN_TAIL = 200
def split_runtime_boundary(prompt: str) -> tuple:
"""``(identity, runtime_marker, runtime)`` of a persisted prompt. Legacy prose may quote
the runtime heading, but only the new renderer ENDS in the boundary; when the marker is
empty the whole prompt is identity."""
identity, runtime_marker, runtime = prompt.rpartition(f"\n\n{RUNTIME_ENVIRONMENT_HEADING}\n\n")
return (identity, runtime_marker, runtime) if prompt.endswith(RUNTIME_ENVIRONMENT_END) else (prompt, "", "")
def identity_line_value(prompt: str, label: str) -> str:
"""Last ``Label: value`` line in the authoritative identity portion of a persisted prompt.
Legacy prose may quote the runtime heading, but only the new renderer ENDS in the boundary;
the final runtime block is embedder prose, never identity, so it is excluded when present.
Last match wins — safe only for volatile-tier trailer fields at the end of the prompt."""
identity, runtime_marker, _ = prompt.rpartition(f"\n\n{RUNTIME_ENVIRONMENT_HEADING}\n\n")
runtime_marker = runtime_marker if prompt.endswith(RUNTIME_ENVIRONMENT_END) else ""
"""Last ``Label: value`` line in the identity portion (the final runtime block is embedder
prose, never identity). Last match wins — safe only for the volatile-tier trailer fields."""
prefix = f"{label}:"
matches = [
line[len(prefix):].strip()
for line in (identity if runtime_marker else prompt).splitlines()
if line.startswith(prefix)
]
matches = [line[len(prefix):].strip() for line in split_runtime_boundary(prompt)[0].splitlines()
if line.startswith(prefix)]
return matches[-1] if matches else ""
def _transcript_row_texts(msg: Any) -> Iterator[str]:
"""Every string the model actually saw for one transcript row: the wire sidecar first, then
the stored content (plain string, or the text parts of a multimodal list — where the note
lands as a durable text part because a list cannot take the string sidecar)."""
if not isinstance(msg, dict):
return
sidecar = msg.get("api_content")
if isinstance(sidecar, str):
yield sidecar
content = msg.get("content")
if isinstance(content, str):
yield content
elif isinstance(content, list):
for part in content:
if isinstance(part, dict) and isinstance(part.get("text"), str):
yield part["text"]
def _last_announced_surface(conversation_history: Any) -> str:
"""The surface named by the NEWEST switch note in the transcript ("" when none).
@@ -69,18 +52,18 @@ def _last_announced_surface(conversation_history: Any) -> str:
is the last thing the model was told it runs on. Reading it back is also what keeps a fresh
AIAgent per turn (the gateway shape) from stacking one copy of the note per turn."""
for msg in reversed((conversation_history or [])[-_NOTE_SCAN_TAIL:]):
for text in _transcript_row_texts(msg):
if _SURFACE_SWITCH_NOTE_PREFIX in text:
tail = text.rsplit(_SURFACE_SWITCH_NOTE_PREFIX, 1)[1]
return tail.split(_SURFACE_NAME_END, 1)[0].strip()
# The note only ever lands on a user row: in its api_content sidecar, or as a text part
# when the content is a multimodal list (which cannot take the string sidecar).
if not isinstance(msg, dict) or msg.get("role") != "user":
continue
sidecar = msg.get("api_content")
text = (sidecar if isinstance(sidecar, str) else "") + "\n" + flatten_message_text(msg.get("content"))
if _SURFACE_SWITCH_NOTE_PREFIX in text:
tail = text.rsplit(_SURFACE_SWITCH_NOTE_PREFIX, 1)[1]
return tail.split(_SURFACE_NAME_END, 1)[0].strip()
return ""
def _agent_tool_names(agent: Any) -> List[str]:
from tools.mcp_tool_agent import _def_name
return [name for name in map(_def_name, getattr(agent, "tools", None) or []) if name]
def note_inert_pinned_tools(agent: Any, built_for_this_surface: List[str]) -> None:
"""Name, at the end of the staged note, the pinned tools THIS surface did not build.
@@ -88,8 +71,9 @@ def note_inert_pinned_tools(agent: Any, built_for_this_surface: List[str]) -> No
thing that would still re-prefill the request behind the preserved prompt — so the model has
to be TOLD they are inert here, or it plans around a ``focus_pane`` a terminal turn can only
answer with ``tool_error("desktop only")``."""
from tools.mcp_tool_agent import agent_tool_names
surface_names = set(built_for_this_surface)
inert = [name for name in _agent_tool_names(agent) if name not in surface_names]
inert = [name for name in agent_tool_names(agent) if name not in surface_names]
note = getattr(agent, "_surface_switch_note", "") or ""
if not inert or not note:
return
@@ -117,12 +101,14 @@ def stage_surface_switch_note(agent: Any, prompt: str, conversation_history: Any
if getattr(agent, "provider", None) == "moa" or getattr(agent, "api_mode", None) == "codex_app_server":
return False
current = str(getattr(agent, "platform", "") or "").strip()
if not current:
return False
described = identity_line_value(prompt, "Platform")
told = _last_announced_surface(conversation_history) or described
if not current or not told or told == current:
if not told or told == current:
return False
from agent.system_prompt import platform_surface_hint
hint = platform_surface_hint(agent) if described != current else ""
from agent.system_prompt import platform_hint
hint = platform_hint(agent) if described != current else ""
where = "the guidance below" if hint else "the interface section in the system prompt above"
note = (
f"{_SURFACE_SWITCH_NOTE_PREFIX}{current}{_SURFACE_NAME_END} in this conversation is "

16
agent/system_prompt.py
View File

@@ -381,7 +381,7 @@ def _active_profile_line(agent: Any) -> str:
)
def _platform_hint(agent: Any) -> str:
def platform_hint(agent: Any) -> str:
"""Built-in/plugin platform hint + Telegram rich-messages opt-in + config
override + desktop TUI clarifier."""
platform_key = (agent.platform or "").lower().strip()
@@ -457,16 +457,6 @@ def _timestamp_line(agent: Any) -> str:
return timestamp_line + "".join(f"\n{label}: {value}" for label, value in trailer if value)
def platform_surface_hint(agent: Any) -> str:
"""The rendering-surface guidance for ``agent.platform`` ("" when the surface has none).
Public because a session that changed surface mid-conversation keeps its stored prompt
(rebuilding it re-prefills the whole request) and delivers the CURRENT surface's guidance
through the per-turn user-message channel instead — see ``_SURFACE_SWITCH_NOTE_PREFIX``
in ``agent.conversation_loop`` (#104414)."""
return _platform_hint(agent) or ""
def _memory_parts(agent: Any) -> List[str]:
"""Built-in memory/USER.md blocks plus the external provider block (gated on
the same check ``inject_memory_provider_tools`` uses, so we never advertise
@@ -589,7 +579,7 @@ def _post_workspace_parts(agent: Any) -> List[str]:
pass # Probe failure must never block prompt build.
if getattr(agent, "_bot_mode_protocol", True):
parts.extend(_bot_mode_parts(agent))
parts += [_active_profile_line(agent), _platform_hint(agent)]
parts += [_active_profile_line(agent), platform_hint(agent)]
return parts
@@ -746,7 +736,7 @@ def format_tools_for_system_message(agent: Any) -> str:
__all__ = ["build_system_prompt_parts", "build_system_prompt", "invalidate_system_prompt",
"platform_surface_hint", "restore_plugin_prompt_sections", "format_tools_for_system_message"]
"platform_hint", "restore_plugin_prompt_sections", "format_tools_for_system_message"]
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----

27
agent/turn_context.py
View File

@@ -114,22 +114,25 @@ def extract_api_content_sidecar(msg: Mapping[str, Any]) -> Optional[str]:
return v if isinstance(v, str) else None
def consume_gateway_turn_context_notes(agent: Any) -> str:
"""Pop the gateway's per-turn must-deliver notes off the agent (one-shot, so the
system prompt stays byte-stable and a cached agent never replays a stale note)."""
notes = getattr(agent, "_gateway_turn_context_notes", "") or ""
if hasattr(agent, "_gateway_turn_context_notes"):
def _pop_turn_note(agent: Any, attr: str) -> str:
"""One-shot per-turn note: read and clear, so the system prompt stays byte-stable and a
cached agent never replays a stale note."""
note = getattr(agent, attr, "") or ""
if hasattr(agent, attr):
with suppress(Exception):
agent._gateway_turn_context_notes = ""
return notes if isinstance(notes, str) else ""
setattr(agent, attr, "")
return note if isinstance(note, str) else ""
def consume_gateway_turn_context_notes(agent: Any) -> str:
"""Pop the gateway's per-turn must-deliver notes."""
return _pop_turn_note(agent, "_gateway_turn_context_notes")
def consume_surface_switch_note(agent: Any) -> str:
"""Pop the one-shot surface-switch note staged by the system-prompt restore (#104414); it rides
the same user-message channel as the gateway's must-deliver notes, behind the cached prefix."""
note = getattr(agent, "_surface_switch_note", "") or ""
agent._surface_switch_note = ""
return note if isinstance(note, str) else ""
"""Pop the surface-switch note staged by the system-prompt restore (#104414); rides the same
user-message channel as the gateway notes, behind the cached prefix."""
return _pop_turn_note(agent, "_surface_switch_note")
def append_notes_to_multimodal_content(content: Any, notes: str) -> bool:

5
tools/mcp_tool_agent.py
View File

@@ -24,6 +24,11 @@ def _agent_tool_defs(agent) -> list:
return list(getattr(agent, "tools", None) or [])
def agent_tool_names(agent) -> list:
"""Names of ``agent.tools`` in wire order (unnamed entries skipped)."""
return [name for name in map(_def_name, _agent_tool_defs(agent)) if name]
def _resolve_refresh_toolsets(agent, enabled_override, disabled_override):
"""Explicit reloads pass freshly-resolved toolsets (so a server just ENABLED in config is
picked up) and the agent's selection is updated to match; automatic paths pass nothing