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

fix(gateway): preserve exact API content through timestamp replay

Browse Source
This commit is contained in:
3ASiC
2026-09-06 12:07:50 +08:00
committed by kshitij
parent cb9546c055
commit 08f269fa13
2 changed files with 172 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
gateway/run.py
View File

@@ -1059,7 +1059,8 @@ def _build_replay_entry(
providers.
"""
entry: Dict[str, Any] = {"role": role, "content": content}
# api_content sidecar keeps the request prefix byte-stable — ONLY if this pipeline did not rewrite content.
# Preserve exact sent content unless cleanup rewrote the body.
# Timestamp-only rendering is checked separately by the history builder.
_sidecar = msg.get("api_content")
if (
role in ("user", "assistant")
@@ -1150,7 +1151,10 @@ def _build_gateway_agent_history(
Observed context stays out of ``conversation_history`` so consecutive-user repair can't merge it in."""
from hermes_time import get_timezone as _get_msg_tz
from gateway.message_timestamps import render_user_content_with_timestamp as _render_msg_ts
from gateway.message_timestamps import (
render_user_content_with_timestamp as _render_msg_ts,
strip_leading_message_timestamps as _strip_msg_ts,
)
_msg_tz = _get_msg_tz()
agent_history: List[Dict[str, Any]] = []
@@ -1164,9 +1168,9 @@ def _build_gateway_agent_history(
continue
content = msg.get("content")
if inject_timestamps and role == "user" and isinstance(content, str):
content = _render_msg_ts(content, msg.get("timestamp"), tz=_msg_tz)
if separate_observed_context and msg.get("observed") and role == "user" and content:
if inject_timestamps and isinstance(content, str):
content = _render_msg_ts(content, msg.get("timestamp"), tz=_msg_tz)
observed_group_context.append(str(content).strip())
continue
@@ -1175,16 +1179,36 @@ def _build_gateway_agent_history(
clean_msg = {k: v for k, v in msg.items() if k not in {"timestamp", "observed"}}
agent_history.append(clean_msg)
elif content:
# Strip persisted auto-continue notes: keep the real user text, never replay the recovery note.
replay_timestamp = msg.get("timestamp")
# Clean before rendering: a timestamp prefix hides recovery notes
# from the startswith-based stripper. Retain an embedded original time.
if role == "user":
content = _strip_auto_continue_noise(content)
if isinstance(content, str):
body, embedded_timestamp = _strip_msg_ts(content, tz=_msg_tz)
clean_body = _strip_auto_continue_noise(body)
if clean_body != body:
content = clean_body
if embedded_timestamp is not None:
replay_timestamp = embedded_timestamp
if not content:
continue
if msg.get("mirror"):
mirror_src = msg.get("mirror_source", "another session")
content = f"[Delivered from {mirror_src}] {content}"
# Keep user timestamps for the stale-dangerous-confirmation stripper in agent/replay_cleanup.py.
entry = _build_replay_entry(role, content, msg, preserve_timestamp=(role == "user"))
if inject_timestamps and role == "user" and isinstance(content, str):
rendered = _render_msg_ts(content, replay_timestamp, tz=_msg_tz)
# Preserve only a sidecar matching the complete rendered message,
# optionally followed by the normal context separator. Cleanup
# above already invalidated sidecars containing stripped content.
sidecar = entry.get("api_content")
if rendered != content and sidecar and not (
sidecar == rendered or sidecar.startswith(rendered + "\n\n")
):
entry.pop("api_content", None)
entry["content"] = rendered
if msg.get("mirror"):
mirror_src = msg.get("mirror_source", "another session")
entry["content"] = f"[Delivered from {mirror_src}] {entry['content']}"
entry.pop("api_content", None)
agent_history.append(entry)
# Strip interrupted tool-call tails so the LLM doesn't re-execute tools killed mid-flight.

139
tests/gateway/test_timestamp_sidecar_replay.py Normal file
View File

@@ -0,0 +1,139 @@
"""Timestamp rendering must not discard the exact sent user-message prefix."""
from copy import deepcopy
from datetime import datetime
from zoneinfo import ZoneInfo
import json
import pytest
from gateway.message_timestamps import render_user_content_with_timestamp
from gateway.run import _build_gateway_agent_history, _select_cached_agent_history
STAMP = datetime(2026, 8, 20, 12, 0, tzinfo=ZoneInfo("UTC")).timestamp()
POLICY = "## Recall policy\nUse the retrieval tool when earlier details are needed."
NOTE = "[System note: Your previous turn was interrupted. Continue the old task.]"
def _render(text, timestamp=STAMP):
from hermes_time import get_timezone
return render_user_content_with_timestamp(text, timestamp, tz=get_timezone())
@pytest.mark.parametrize("timestamps", [False, True])
@pytest.mark.parametrize("embedded", [False, True])
@pytest.mark.parametrize("real_text", ["", "please check the result"])
def test_recovery_cleanup_never_restores_a_sidecar(timestamps, embedded, real_text):
original = NOTE + (" " + real_text if real_text else "")
if embedded:
original = "[2026-08-20T12:00:00+00:00] " + original
replay, _ = _build_gateway_agent_history(
[{"role": "user", "content": original, "api_content": original + "\n\n" + POLICY, "timestamp": STAMP + 100}],
inject_timestamps=timestamps,
)
if not real_text:
assert replay == []
else:
assert "api_content" not in replay[0]
assert "System note:" not in replay[0]["content"]
expected_timestamp = STAMP if embedded else STAMP + 100
assert replay[0]["content"] == (_render(real_text, expected_timestamp) if timestamps else real_text)
@pytest.fixture
def responses_agent(tmp_path, monkeypatch):
"""Use the real agent/Responses converter; replace only the network call."""
from hermes_state import SessionDB
from run_agent import AIAgent
captured = []
responses = []
sid = "sanitized-timestamp-replay"
db = SessionDB(db_path=tmp_path / "state.db")
monkeypatch.setattr(
"hermes_cli.plugins.invoke_hook",
lambda hook, **kw: [{"context": POLICY}] if hook == "pre_llm_call" else [],
)
def respond(kwargs, **unused):
captured.append(deepcopy(kwargs))
output = responses.pop(0) if responses else [
{"type": "message", "id": "msg_done", "role": "assistant", "status": "completed", "phase": "final_answer", "content": [{"type": "output_text", "text": "done", "annotations": []}]}
]
from openai.types.responses import Response
return Response.model_validate({
"id": "resp_test", "object": "response", "created_at": STAMP,
"model": "test-model", "status": "completed", "output": output,
"usage": None, "error": None, "incomplete_details": None,
"instructions": None, "metadata": {}, "parallel_tool_calls": True,
"temperature": None, "tool_choice": "auto", "tools": [], "top_p": None,
})
def make_agent():
agent = AIAgent(
api_key="test-key", base_url="http://127.0.0.1:1/v1", provider="openai-compat",
model="test-model", api_mode="codex_responses", max_iterations=4,
enabled_toolsets=[], quiet_mode=True, skip_context_files=True,
skip_memory=True, save_trajectories=False, session_db=db, session_id=sid,
)
agent._cached_system_prompt = "Stable synthetic system prompt."
agent.valid_tool_names = {"read_file"}
monkeypatch.setattr(agent, "_interruptible_streaming_api_call", respond)
monkeypatch.setattr(agent, "_interruptible_api_call", respond)
return agent
yield make_agent, captured, responses, db, sid
db.close()
@pytest.mark.parametrize("timestamps", [False, True])
@pytest.mark.parametrize("resume", ["cached", "db"])
def test_full_builder_to_responses_keeps_cross_turn_prefix(responses_agent, tmp_path, timestamps, resume):
make_agent, captured, responses, db, sid = responses_agent
agent = make_agent()
tool_file = tmp_path / "sanitized-tool.txt"
tool_file.write_text("fixture result", encoding="utf-8")
responses.extend([
[
{"type": "reasoning", "id": "rs_test", "summary": [], "encrypted_content": "synthetic-reasoning"},
{"type": "function_call", "id": "fc_test", "call_id": "call_test", "name": "read_file", "arguments": json.dumps({"path": str(tool_file)})},
],
[{"type": "message", "id": "msg_done", "role": "assistant", "status": "completed", "phase": "final_answer", "content": [{"type": "output_text", "text": "done", "annotations": []}]}],
])
current = _render("first question") if timestamps else "first question"
result = agent.run_conversation(current, conversation_history=[], task_id="first", persist_user_message="first question", persist_user_timestamp=STAMP)
assert result["completed"]
assert len(captured) == 2
first_input = captured[0]["input"]
continuation = captured[1]["input"]
assert continuation[:len(first_input)] == first_input
assert any(item.get("call_id") == "call_test" and item.get("type") == "function_call_output" for item in continuation)
history = db.get_messages_as_conversation(sid)
if resume == "db":
from hermes_state import SessionDB
reopened = SessionDB(db_path=tmp_path / "state.db")
try:
history = reopened.get_messages_as_conversation(sid)
finally:
reopened.close()
replay, observed = _build_gateway_agent_history(history, inject_timestamps=timestamps)
assert observed is None
if resume == "cached":
replay = _select_cached_agent_history(replay, agent._session_messages)
else:
agent = make_agent()
next_user = _render("second question", STAMP + 30) if timestamps else "second question"
result = agent.run_conversation(next_user, conversation_history=replay, task_id="second", persist_user_message="second question", persist_user_timestamp=STAMP + 30)
assert result["completed"]
assert len(captured) == 3
next_input = captured[2]["input"]
assert next_input[:len(continuation)] == continuation
assert not any("api_content" in item or "timestamp" in item for item in next_input)
assert any(item.get("encrypted_content") == "synthetic-reasoning" for item in next_input)
assert any(item.get("phase") == "final_answer" for item in next_input)