Skip to content

closed: opened in error #31634

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HeVeMi wants to merge 1 commit into from
+73 −17
Closed

closed: opened in error #31634

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 42 additions & 14 deletions plugins/memory/hindsight/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -574,7 +574,8 @@ def __init__(self):
self._retain_async = True
self._retain_context = "conversation between Hermes Agent and the User"
self._turn_counter = 0
self._session_turns: list[str] = [] # accumulates ALL turns for the session
self._session_turns: list[str] = [] # accumulates turns for this session lifecycle
self._last_retained_turn_count = 0

# Recall controls
self._auto_recall = True
Expand Down Expand Up @@ -1109,6 +1110,7 @@ def initialize(self, session_id: str, **kwargs) -> None:
self._agent_workspace = str(kwargs.get("agent_workspace") or "").strip()
self._turn_index = 0
self._session_turns = []
self._last_retained_turn_count = 0
self._mode = self._config.get("mode", "cloud")
# Read timeout from config or env var, fall back to default
self._timeout = _parse_int_setting(
Expand Down Expand Up @@ -1441,9 +1443,22 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
self._turn_counter, self._turn_counter + (self._retain_every_n_turns - self._turn_counter % self._retain_every_n_turns))
return

logger.debug("sync_turn: retaining %d turns, total session content %d chars",
len(self._session_turns), sum(len(t) for t in self._session_turns))
content = "[" + ",".join(self._session_turns) + "]"
document_id, update_mode = self._resolve_retain_target(self._document_id)
if update_mode == "append":
turns_to_retain = list(self._session_turns[self._last_retained_turn_count:])
else:
# Legacy APIs overwrite a stable document_id, so preserve the old
# behavior and send the full lifecycle buffer each time.
turns_to_retain = list(self._session_turns)
if not turns_to_retain:
logger.debug("sync_turn: skipped retain; no new turns to send")
return
retained_until = len(self._session_turns)
content = "[" + ",".join(turns_to_retain) + "]"
logger.debug(
"sync_turn: retaining %d turn(s), total buffered turns=%d, content %d chars, mode=%s",
len(turns_to_retain), len(self._session_turns), len(content), update_mode,
)

lineage_tags: list[str] = []
if self._session_id:
Expand All @@ -1454,11 +1469,10 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
# Snapshot the state needed for the retain. The writer may run after
# _session_turns / _turn_index are mutated by a later sync_turn().
metadata_snapshot = self._build_metadata(
message_count=len(self._session_turns) * 2,
message_count=len(turns_to_retain) * 2,
turn_index=self._turn_index,
)
num_turns = len(self._session_turns)
document_id, update_mode = self._resolve_retain_target(self._document_id)
num_turns = len(turns_to_retain)
bank_id = self._bank_id
retain_async_flag = self._retain_async
retain_context = self._retain_context
Expand All @@ -1484,6 +1498,11 @@ def _do_retain() -> None:
retain_async=retain_async_flag,
)
)
if update_mode == "append":
self._last_retained_turn_count = max(
self._last_retained_turn_count,
retained_until,
)
logger.debug("Hindsight retain succeeded")

self._ensure_writer()
Expand Down Expand Up @@ -1614,23 +1633,31 @@ def on_session_switch(
old_session_id = self._session_id
old_parent_session_id = self._parent_session_id
old_turn_index = self._turn_index
old_metadata = self._build_metadata(
message_count=len(old_turns) * 2,
turn_index=old_turn_index,
)
old_lineage_tags: list[str] = []
if old_session_id:
old_lineage_tags.append(f"session:{old_session_id}")
if old_parent_session_id:
old_lineage_tags.append(f"parent:{old_parent_session_id}")
old_content = "[" + ",".join(old_turns) + "]"
# Resolve doc_id + update_mode against the OLD session BEFORE
# we rotate _session_id, so the flush lands in the old
# session's document either way (legacy: per-process unique;
# ≥0.5.0: stable session-scoped + append).
old_document_id, old_update_mode = self._resolve_retain_target(
self._document_id
)
if old_update_mode == "append":
old_turns_to_flush = old_turns[self._last_retained_turn_count:]
else:
old_turns_to_flush = old_turns
if not old_turns_to_flush:
old_content = ""
else:
old_content = "[" + ",".join(old_turns_to_flush) + "]"
old_num_turns = len(old_turns_to_flush)
old_metadata = self._build_metadata(
message_count=old_num_turns * 2,
turn_index=old_turn_index,
)

def _flush():
try:
Expand All @@ -1646,7 +1673,7 @@ def _flush():
item["update_mode"] = old_update_mode
logger.debug(
"Hindsight flush-on-switch: bank=%s, doc=%s, mode=%s, num_turns=%d",
self._bank_id, old_document_id, old_update_mode, len(old_turns),
self._bank_id, old_document_id, old_update_mode, old_num_turns,
)
self._run_hindsight_operation(
lambda client: client.aretain_batch(
Expand All @@ -1665,7 +1692,7 @@ def _flush():
# two threads on aretain_batch against the same document, and
# keeps shutdown's drain semantics intact. Skip enqueue if
# shutdown has already fired — the writer is draining/gone.
if not self._shutting_down.is_set():
if old_num_turns and not self._shutting_down.is_set():
self._ensure_writer()
self._register_atexit()
self._retain_queue.put(_flush)
Expand All @@ -1684,6 +1711,7 @@ def _flush():
start_ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
self._document_id = f"{self._session_id}-{start_ts}"
self._session_turns = []
self._last_retained_turn_count = 0
self._turn_counter = 0
self._turn_index = 0
logger.debug(
Expand Down
34 changes: 31 additions & 3 deletions tests/plugins/memory/test_hindsight_provider.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -756,8 +756,8 @@ def test_sync_turn_every_n_turns(self, provider_with_config):
assert item["metadata"]["turn_index"] == "3"
assert item["metadata"]["message_count"] == "6"

def test_sync_turn_accumulates_full_session(self, provider_with_config):
"""Each retain sends the ENTIRE session, not just the latest batch."""
def test_sync_turn_accumulates_full_session_without_append_support(self, provider_with_config):
"""Legacy APIs still receive the full lifecycle buffer on each retain."""
p = provider_with_config(retain_every_n_turns=2)

p.sync_turn("turn1-user", "turn1-asst")
Expand All @@ -771,12 +771,40 @@ def test_sync_turn_accumulates_full_session(self, provider_with_config):
p._retain_queue.join()

content = p._client.aretain_batch.call_args.kwargs["items"][0]["content"]
# Should contain ALL turns from the session
# Should contain ALL turns from the session for overwrite-only APIs.
assert "turn1-user" in content
assert "turn2-user" in content
assert "turn3-user" in content
assert "turn4-user" in content

def test_sync_turn_sends_only_new_batch_with_append_support(self, provider_with_config, monkeypatch):
"""Append-capable APIs should not resend the whole session forever."""
monkeypatch.setattr(
"plugins.memory.hindsight._check_api_supports_update_mode_append",
lambda api_url, api_key: True,
)
p = provider_with_config(retain_every_n_turns=2)

p.sync_turn("turn1-user", "turn1-asst")
p.sync_turn("turn2-user", "turn2-asst")
p._retain_queue.join()

first_call = p._client.aretain_batch.call_args.kwargs
assert first_call["document_id"] == "test-session"
assert first_call["items"][0]["update_mode"] == "append"

p._client.aretain_batch.reset_mock()

p.sync_turn("turn3-user", "turn3-asst")
p.sync_turn("turn4-user", "turn4-asst")
p._retain_queue.join()

content = p._client.aretain_batch.call_args.kwargs["items"][0]["content"]
assert "turn1-user" not in content
assert "turn2-user" not in content
assert "turn3-user" in content
assert "turn4-user" in content

def test_sync_turn_passes_document_id(self, provider):
"""sync_turn should pass document_id (session_id + per-startup ts)."""
provider.sync_turn("hello", "hi")
Expand Down