Hindsight plugin transcript duplication and session bugs

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #15079 fix(hindsight): scope document_id per process + optional bank_id_template

  merged to main

  files: plugins/memory/hindsight/README.md, plugins/memory/hindsight/__init__.py, tests/plugins/memory/test_hindsight_provider.py

  Show the diff

  • plugins/memory/hindsight/README.md

    @@ -59,7 +59,8 @@ Config file: `~/.hermes/hindsight/config.json`
     
     | Key | Default | Description |
     |-----|---------|-------------|
    -| `bank_id` | `hermes` | Memory bank name |
    +| `bank_id` | `hermes` | Memory bank name (static fallback used when `bank_id_template` is unset or resolves empty) |
    +| `bank_id_template` | — | Optional template to derive the bank name dynamically. Placeholders: `{profile}`, `{workspace}`, `{platform}`, `{user}`, `{session}`. Example: `hermes-{profile}` isolates memory per active Hermes profile. Empty placeholders collapse cleanly (e.g. `hermes-{user}` with no user becomes `hermes`). |
     | `bank_mission` | — | Reflect mission (identity/framing for reflect reasoning). Applied via Banks API. |
     | `bank_retain_mission` | — | Retain mission (steers what gets extracted). Applied via Banks API. |
     

  • plugins/memory/hindsight/__init__.py

    @@ -324,6 +324,60 @@ def _materialize_embedded_profile_env(config: dict[str, Any], *, llm_api_key: st
         )
         return profile_env
     
    +def _sanitize_bank_segment(value: str) -> str:
    +    """Sanitize a bank_id_template placeholder value.
    +
    +    Bank IDs should be safe for URL paths and filesystem use. Replaces any
    +    character that isn't alphanumeric, dash, or underscore with a dash, and
    +    collapses runs of dashes.
    +    """
    +    if not value:
    +        return ""
    +    out = []
    +    prev_dash = False
    +    for ch in str(value):
    +        if ch.isalnum() or ch == "-" or ch == "_":
    +            out.append(ch)
    +            prev_dash = False
    +        else:
    +            if not prev_dash:
    +                out.append("-")
    +                prev_dash = True
    +    return "".join(out).strip("-_")
    +
    +
    +def _resolve_bank_id_template(template: str, fallback: str, **placeholders: str) -> str:
    +    """Resolve a bank_id template string with the given placeholders.
    +
    +    Supported placeholders (each is sanitized before substitution):
    +      {profile}   — active Hermes profile name (from agent_identity)
    +      {workspace} — Hermes workspace name (from agent_workspace)
    +      {platform}  — "cli", "telegram", "discord", etc.
    +      {user}      — platform user id (gateway sessions)
    +      {session}   — current session id
    +
    +    Missing/empty placeholders are rendered as the empty string and then
    +    collapsed — e.g. ``hermes-{user}`` with no user becomes ``hermes``.
    +
    +    If the template is empty, resolution falls back to *fallback*.
    +    Returns the sanitized bank id.
    +    """
    +    if not template:
    +        return fallback
    +    sanitized = {k: _sanitize_bank_segment(v) for k, v in placeholders.items()}
    +    try:
    +        rendered = template.format(**sanitized)
    +    except (KeyError, IndexError) as exc:
    +        logger.warning("Invalid bank_id_template %r: %s — using fallback %r",
    +                       template, exc, fallback)
    +        return fallback
    +    while "--" in rendered:
    +        rendered = rendered.replace("--", "-")
    +    while "__" in rendered:
    +        rendered = rendered.replace("__", "_")
    +    rendered = rendered.strip("-_")
    +    return rendered or fallback
    +
     
     # ---------------------------------------------------------------------------
     # MemoryProvider implementation
    @@ -354,6 +408,7 @@ def __init__(self):
             self._chat_type = ""
             self._thread_id = ""
             self._agent_identity = ""
    +        self._agent_workspace = ""
             self._turn_index = 0
             self._client = None
             self._timeout = _DEFAULT_TIMEOUT
    @@ -362,6 +417,8 @@ def __init__(self):
             self._prefetch_thread = None
             self._sync_thread = None
             self._session_id = ""
    +        self._parent_session_id = ""
    +        self._document_id = ""
     
             # Tags
             self._tags: list[str] | None = None
    @@ -386,6 +443,7 @@ def __init__(self):
             # Bank
             self._bank_mission = ""
             self._bank_retain_mission: str | None = None
    +        self._bank_id_template = ""
     
         @property
         def name(self) -> str:
    @@ -613,7 +671,8 @@ def get_config_schema(self):
                 {"key": "llm_base_url", "description": "Endpoint URL (e.g. http://192.168.1.10:8080/v1)", "default": "", "when": {"mode": "local_embedded", "llm_provider": "openai_compatible"}},
                 {"key": "llm_api_key", "description": "LLM API key (optional for openai_compatible)", "secret": True, "env_var": "HINDSIGHT_LLM_API_KEY", "when": {"mode": "local_embedded"}},
                 {"key": "llm_model", "description": "LLM model", "default": "gpt-4o-mini", "default_from": {"field": "llm_provider", "map": _PROVIDER_DEFAULT_MODELS}, "when": {"mode": "local_embedded"}},
    -            {"key": "bank_id", "description": "Memory bank name", "default": "hermes"},
    +            {"key": "bank_id", "description": "Memory bank name (static fallback when bank_id_template is unset)", "default": "hermes"},
    +            {"key": "bank_id_template", "description": "Optional template to derive bank_id dynamically. Placeholders: {profile}, {workspace}, {platform}, {user}, {session}. Example: hermes-{profile}", "default": ""},
                 {"key": "bank_mission", "description": "Mission/purpose description for the memory bank"},
                 {"key": "bank_retain_mission", "description": "Custom extraction prompt for memory retention"},
                 {"key": "recall_budget", "description": "Recall thoroughness", "default": "mid", "choices": ["low", "mid", "high"]},
    @@ -679,6 +738,15 @@ def _run_sync(self, coro):
     
         def initialize(self, session_id: str, **kwargs) -> None:
             self._session_id = str(session_id or "").strip()
    +        self._parent_session_id = str(kwargs.get("parent_session_id", "") or "").strip()
    +
    +        # Each process lifecycle gets its own document_id. Reusing session_id
    +        # alone caused overwrites on /resume — the reloaded session starts
    +        # with an empty _session_turns, so the next retain would replace the
    +        # previously stored content. session_id stays in tags so processes
    +        # for the same session remain filterable together.
    +        start_ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    +        self._document_id = f"{self._session_id}-{start_ts}"
     
             # Check client version and auto-upgrade if needed
             try:
    @@ -717,6 +785,7 @@ def initialize(self, session_id: str, **kwargs) -> None:
             self._chat_type = str(kwargs.get("chat_type") or "").strip()
             self._thread_id = str(kwargs.get("thread_id") or "").strip()
             self._agent_identity = str(kwargs.get("agent_identity") or "").strip()
    +        self._agent_workspace = str(kwargs.get("agent_workspace") or "").strip()
             self._turn_index = 0
             self._session_turns = []
             self._mode = self._config.get("mode", "cloud")
    @@ -740,7 +809,17 @@ def initialize(self, session_id: str, **kwargs) -> None:
             self._llm_base_url = self._config.get("llm_base_url", "")
     
             banks = self._config.get("banks", {}).get("hermes", {})
    -        self._bank_id = self._config.get("bank_id") or banks.get("bankId", "hermes")
    +        static_bank_id = self._config.get("bank_id") or banks.get("bankId", "hermes")
    +        self._bank_id_template = self._config.get("bank_id_template", "") or ""
    +        self._bank_id = _resolve_bank_id_template(
    +            self._bank_id_template,
    +            fallback=static_bank_id,
    +            profile=self._agent_identity,
    +            workspace=self._agent_workspace,
    +            platform=self._platform,
    +            user=self._user_id,
    +            session=self._session_id,
    +        )
             budget = self._config.get("recall_budget") or self._config.get("budget") or banks.get("budget", "mid")
             self._budget = budget if budget in _VALID_BUDGETS else "mid"
     
    @@ -793,6 +872,10 @@ def initialize(self, session_id: str, **kwargs) -> None:
                 pass
             logger.info("Hindsight initialized: mode=%s, api_url=%s, bank=%s, budget=%s, memory_mode=%s, prefetch_method=%s, client=%s",
                          self._mode, self._api_url, self._bank_id, self._budget, self._memory_mode, self._prefetch_method, _client_version)
    +        if self._bank_id_template:
    +            logger.debug("Hindsight bank resolved from template %r: profile=%s workspace=%s platform=%s user=%s -> bank=%s",
    +                         self._bank_id_template, self._agent_identity, self._agent_workspace,
    +                         self._platform, self._user_id, self._bank_id)
             logger.debug("Hindsight config: auto_retain=%s, auto_recall=%s, retain_every_n=%d, "
                          "retain_async=%s, retain_context=%s, recall_max_tokens=%d, recall_max_input_chars=%d, tags=%s, recall_tags=%s",
                          self._auto_retain, self._auto_recall, self._retain_every_n_turns,
    @@ -1026,6 +1109,12 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
                          len(self._session_turns), sum(len(t) for t in self._session_turns))
             content = "[" + ",".join(self._session_turns) + "]"
     
    +        lineage_tags: list[str] = []
    +        if self._session_id:
    +            lineage_tags.append(f"session:{self._session_id}")
    +        if self._parent_session_id:
    +            lineage_tags.append(f"parent:{self._parent_session_id}")
    +
             def _sync():
                 try:
                     client = self._get_client()
    @@ -1036,15 +1125,16 @@ def _sync():
                             message_count=len(self._session_turns) * 2,
                             turn_index=self._turn_index,
                         ),
    +                    tags=lineage_tags or None,
                     )
                     item.pop("bank_id", None)
                     item.pop("retain_async", None)
                     logger.debug("Hindsight retain: bank=%s, doc=%s, async=%s, content_len=%d, num_turns=%d",
    -                             self._bank_id, self._session_id, self._retain_async, len(content), len(self._session_turns))
    +                             self._bank_id, self._document_id, self._retain_async, len(content), len(self._session_turns))
                     self._run_sync(client.aretain_batch(
                         bank_id=self._bank_id,
                         items=[item],
    -                    document_id=self._session_id,
    +                    document_id=self._document_id,
                         retain_async=self._retain_async,
                     ))
                     logger.debug("Hindsight retain succeeded")

  • tests/plugins/memory/test_hindsight_provider.py

    @@ -19,6 +19,8 @@
         RETAIN_SCHEMA,
         _load_config,
         _normalize_retain_tags,
    +    _resolve_bank_id_template,
    +    _sanitize_bank_segment,
     )
     
     
    @@ -549,12 +551,12 @@ def test_sync_turn_retains_metadata_rich_turn(self, provider_with_config):
             p._client.aretain_batch.assert_called_once()
             call_kwargs = p._client.aretain_batch.call_args.kwargs
             assert call_kwargs["bank_id"] == "test-bank"
    -        assert call_kwargs["document_id"] == "session-1"
    +        assert call_kwargs["document_id"].startswith("session-1-")
             assert call_kwargs["retain_async"] is True
             assert len(call_kwargs["items"]) == 1
             item = call_kwargs["items"][0]
             assert item["context"] == "conversation between Hermes Agent and the User"
    -        assert item["tags"] == ["conv", "session1"]
    +        assert item["tags"] == ["conv", "session1", "session:session-1"]
             content = json.loads(item["content"])
             assert len(content) == 1
             assert content[0][0]["role"] == "user"
    @@ -582,6 +584,36 @@ def test_sync_turn_skipped_when_auto_retain_off(self, provider_with_config):
             assert p._sync_thread is None
             p._client.aretain_batch.assert_not_called()
     
    +    def test_sync_turn_with_tags(self, provider_with_config):
    +        p = provider_with_config(retain_tags=["conv", "session1"])
    +        p.sync_turn("hello", "hi")
    +        if p._sync_thread:
    +            p._sync_thread.join(timeout=5.0)
    +        item = p._client.aretain_batch.call_args.kwargs["items"][0]
    +        assert "conv" in item["tags"]
    +        assert "session1" in item["tags"]
    +        assert "session:test-session" in item["tags"]
    +
    +    def test_sync_turn_uses_aretain_batch(self, provider):
    +        """sync_turn should use aretain_batch with retain_async."""
    +        provider.sync_turn("hello", "hi")
    +        if provider._sync_thread:
    +            provider._sync_thread.join(timeout=5.0)
    +        provider._client.aretain_batch.assert_called_once()
    +        call_kwargs = provider._client.aretain_batch.call_args.kwargs
    +        assert call_kwargs["document_id"].startswith("test-session-")
    +        assert call_kwargs["retain_async"] is True
    +        assert len(call_kwargs["items"]) == 1
    +        assert call_kwargs["items"][0]["context"] == "conversation between Hermes Agent and the User"
    +
    +    def test_sync_turn_custom_context(self, provider_with_config):
    +        p = provider_with_config(retain_context="my-agent")
    +        p.sync_turn("hello", "hi")
    +        if p._sync_thread:
    +            p._sync_thread.join(timeout=5.0)
    +        item = p._client.aretain_batch.call_args.kwargs["items"][0]
    +        assert item["context"] == "my-agent"
    +
         def test_sync_turn_every_n_turns(self, provider_with_config):
             p = provider_with_config(retain_every_n_turns=3, retain_async=False)
             p.sync_turn("turn1-user", "turn1-asst")
    @@ -592,7 +624,7 @@ def test_sync_turn_every_n_turns(self, provider_with_config):
             p._sync_thread.join(timeout=5.0)
             p._client.aretain_batch.assert_called_once()
             call_kwargs = p._client.aretain_batch.call_args.kwargs
    -        assert call_kwargs["document_id"] == "test-session"
    +        assert call_kwargs["document_id"].startswith("test-session-")
             assert call_kwargs["retain_async"] is False
             item = call_kwargs["items"][0]
             content = json.loads(item["content"])
    @@ -604,6 +636,95 @@ 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."""
    +        p = provider_with_config(retain_every_n_turns=2)
    +
    +        p.sync_turn("turn1-user", "turn1-asst")
    +        p.sync_turn("turn2-user", "turn2-asst")
    +        if p._sync_thread:
    +            p._sync_thread.join(timeout=5.0)
    +
    +        p._client.aretain_batch.reset_mock()
    +
    +        p.sync_turn("turn3-user", "turn3-asst")
    +        p.sync_turn("turn4-user", "turn4-asst")
    +        if p._sync_thread:
    +            p._sync_thread.join(timeout=5.0)
    +
    +        content = p._client.aretain_batch.call_args.kwargs["items"][0]["content"]
    +        # Should contain ALL turns from the session
    +        assert "turn1-user" in content
    +        assert "turn2-user" 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")
    +        if provider._sync_thread:
    +            provider._sync_thread.join(timeout=5.0)
    +        call_kwargs = provider._client.aretain_batch.call_args.kwargs
    +        # Format: {session_id}-{YYYYMMDD_HHMMSS_microseconds}
    +        assert call_kwargs["document_id"].startswith("test-session-")
    +        assert call_kwargs["document_id"] == provider._document_id
    +
    +    def test_resume_creates_new_document(self, tmp_path, monkeypatch):
    +        """Resuming a session (re-initializing) gets a new document_id
    +        so previously stored content is not overwritten."""
    +        config = {"mode": "cloud", "apiKey": "k", "api_url": "http://x", "bank_id": "b"}
    +        config_path = tmp_path / "hindsight" / "config.json"
    +        config_path.parent.mkdir(parents=True, exist_ok=True)
    +        config_path.write_text(json.dumps(config))
    +        monkeypatch.setattr("plugins.memory.hindsight.get_hermes_home", lambda: tmp_path)
    +
    +        p1 = HindsightMemoryProvider()
    +        p1.initialize(session_id="resumed-session", hermes_home=str(tmp_path), platform="cli")
    +
    +        # Sleep just enough that the microsecond timestamp differs
    +        import time
    +        time.sleep(0.001)
    +
    +        p2 = HindsightMemoryProvider()
    +        p2.initialize(session_id="resumed-session", hermes_home=str(tmp_path), platform="cli")
    +
    +        # Same session, but each process gets its own document_id
    +        assert p1._document_id != p2._document_id
    +        assert p1._document_id.startswith("resumed-session-")
    +        assert p2._document_id.startswith("resumed-session-")
    +
    +    def test_sync_turn_session_tag(self, provider):
    +        """Each retain should be tagged with session:<id> for filtering."""
    +        provider.sync_turn("hello", "hi")
    +        if provider._sync_thread:
    +            provider._sync_thread.join(timeout=5.0)
    +        item = provider._client.aretain_batch.call_args.kwargs["items"][0]
    +        assert "session:test-session" in item["tags"]
    +
    +    def test_sync_turn_parent_session_tag(self, tmp_path, monkeypatch):
    +        """When initialized with parent_session_id, parent tag is added."""
    +        config = {"mode": "cloud", "apiKey": "k", "api_url": "http://x", "bank_id": "b"}
    +        config_path = tmp_path / "hindsight" / "config.json"
    +        config_path.parent.mkdir(parents=True, exist_ok=True)
    +        config_path.write_text(json.dumps(config))
    +        monkeypatch.setattr("plugins.memory.hindsight.get_hermes_home", lambda: tmp_path)
    +
    +        p = HindsightMemoryProvider()
    +        p.initialize(
    +            session_id="child-session",
    +            hermes_home=str(tmp_path),
    +            platform="cli",
    +            parent_session_id="parent-session",
    +        )
    +        p._client = _make_mock_client()
    +        p.sync_turn("hello", "hi")
    +        if p._sync_thread:
    +            p._sync_thread.join(timeout=5.0)
    +
    +        item = p._client.aretain_batch.call_args.kwargs["items"][0]
    +        assert "session:child-session" in item["tags"]
    +        assert "parent:parent-session" in item["tags"]
    +
         def test_sync_turn_error_does_not_raise(self, provider):
             provider._client.aretain_batch.side_effect = RuntimeError("network error")
             provider.sync_turn("hello", "hi")
    @@ -663,7 +784,7 @@ def test_schema_has_all_new_fields(self, provider):
             keys = {f["key"] for f in schema}
             expected_keys = {
                 "mode", "api_url", "api_key", "llm_provider", "llm_api_key",
    -            "llm_model", "bank_id", "bank_mission", "bank_retain_mission",
    +            "llm_model", "bank_id", "bank_id_template", "bank_mission", "bank_retain_mission",
                 "recall_budget", "memory_mode", "recall_prefetch_method",
                 "retain_tags", "retain_source",
                 "retain_user_prefix", "retain_assistant_prefix",
    @@ -676,6 +797,150 @@ def test_schema_has_all_new_fields(self, provider):
             assert expected_keys.issubset(keys), f"Missing: {expected_keys - keys}"
     
     
    +# ---------------------------------------------------------------------------
    +# bank_id_template tests
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestBankIdTemplate:
    +    def test_sanitize_bank_segment_passthrough(self):
    +        assert _sanitize_bank_segment("hermes") == "hermes"
    +        assert _sanitize_bank_segment("my-agent_1") == "my-agent_1"
    +
    +    def test_sanitize_bank_segment_strips_unsafe(self):
    +        assert _sanitize_bank_segment("[redacted:email]") == "josh-example-com"
    +        assert _sanitize_bank_segment("chat:#general") == "chat-general"
    +        assert _sanitize_bank_segment("  spaces  ") == "spaces"
    +
    +    def test_sanitize_bank_segment_empty(self):
    +        assert _sanitize_bank_segment("") == ""
    +        assert _sanitize_bank_segment(None) == ""
    +
    +    def test_resolve_empty_template_uses_fallback(self):
    +        result = _resolve_bank_id_template(
    +            "", fallback="hermes", profile="coder"
    +        )
    +        assert result == "hermes"
    +
    +    def test_resolve_with_profile(self):
    +        result = _resolve_bank_id_template(
    +            "hermes-{profile}", fallback="hermes",
    +            profile="coder", workspace="", platform="", user="", session="",
    +        )
    +        assert result == "hermes-coder"
    +
    +    def test_resolve_with_multiple_placeholders(self):
    +        result = _resolve_bank_id_template(
    +            "{workspace}-{profile}-{platform}",
    +            fallback="hermes",
    +            profile="coder", workspace="myorg", platform="cli",
    +            user="", session="",
    +        )
    +        assert result == "myorg-coder-cli"
    +
    +    def test_resolve_collapses_empty_placeholders(self):
    +        # When user is empty, "hermes-{user}" becomes "hermes-" -> trimmed to "hermes"
    +        result = _resolve_bank_id_template(
    +            "hermes-{user}", fallback="default",
    +            profile="", workspace="", platform="", user="", session="",
    +        )
    +        assert result == "hermes"
    +
    +    def test_resolve_collapses_double_dashes(self):
    +        # Two empty placeholders with a dash between them should collapse
    +        result = _resolve_bank_id_template(
    +            "{workspace}-{profile}-{user}", fallback="fallback",
    +            profile="coder", workspace="", platform="", user="", session="",
    +        )
    +        assert result == "coder"
    +
    +    def test_resolve_empty_rendered_falls_back(self):
    +        result = _resolve_bank_id_template(
    +            "{user}-{profile}", fallback="fallback",
    +            profile="", workspace="", platform="", user="", session="",
    +        )
    +        assert result == "fallback"
    +
    +    def test_resolve_sanitizes_placeholder_values(self):
    +        result = _resolve_bank_id_template(
    +            "user-{user}", fallback="hermes",
    +            profile="", workspace="", platform="",
    +            user="[redacted:email]", session="",
    +        )
    +        assert result == "user-josh-example-com"
    +
    +    def test_resolve_invalid_template_returns_fallback(self):
    +        # Unknown placeholder should fall back without raising
    +        result = _resolve_bank_id_template(
    +            "hermes-{unknown}", fallback="hermes",
    +            profile="", workspace="", platform="", user="", session="",
    +        )
    +        assert result == "hermes"
    +
    +    def test_provider_uses_bank_id_template_from_config(self, tmp_path, monkeypatch):
    +        config = {
    +            "mode": "cloud",
    +            "apiKey": "k",
    +            "api_url": "http://x",
    +            "bank_id": "fallback-bank",
    +            "bank_id_template": "hermes-{profile}",
    +        }
    +        config_path = tmp_path / "hindsight" / "config.json"
    +        config_path.parent.mkdir(parents=True, exist_ok=True)
    +        config_path.write_text(json.dumps(config))
    +        monkeypatch.setattr("plugins.memory.hindsight.get_hermes_home", lambda: tmp_path)
    +
    +        p = HindsightMemoryProvider()
    +        p.initialize(
    +            session_id="s1",
    +            hermes_home=str(tmp_path),
    +            platform="cli",
    +            agent_identity="coder",
    +            agent_workspace="hermes",
    +        )
    +        assert p._bank_id == "hermes-coder"
    +        assert p._bank_id_template == "hermes-{profile}"
    +
    +    def test_provider_without_template_uses_static_bank_id(self, tmp_path, monkeypatch):
    +        config = {
    +            "mode": "cloud",
    +            "apiKey": "k",
    +            "api_url": "http://x",
    +            "bank_id": "my-static-bank",
    +        }
    +        config_path = tmp_path / "hindsight" / "config.json"
    +        config_path.parent.mkdir(parents=True, exist_ok=True)
    +        config_path.write_text(json.dumps(config))
    +        monkeypatch.setattr("plugins.memory.hindsight.get_hermes_home", lambda: tmp_path)
    +
    +        p = HindsightMemoryProvider()
    +        p.initialize(
    +            session_id="s1",
    +            hermes_home=str(tmp_path),
    +            platform="cli",
    +            agent_identity="coder",
    +        )
    +        assert p._bank_id == "my-static-bank"
    +
    +    def test_provider_template_with_missing_profile_falls_back(self, tmp_path, monkeypatch):
    +        config = {
    +            "mode": "cloud",
    +            "apiKey": "k",
    +            "api_url": "http://x",
    +            "bank_id": "hermes-fallback",
    +            "bank_id_template": "hermes-{profile}",
    +        }
    +        config_path = tmp_path / "hindsight" / "config.json"
    +        config_path.parent.mkdir(parents=True, exist_ok=True)
    +        config_path.write_text(json.dumps(config))
    +        monkeypatch.setattr("plugins.memory.hindsight.get_hermes_home", lambda: tmp_path)
    +
    +        p = HindsightMemoryProvider()
    +        # No agent_identity passed — template renders to "hermes-" which collapses to "hermes"
    +        p.initialize(session_id="s1", hermes_home=str(tmp_path), platform="cli")
    +        assert p._bank_id == "hermes"
    +
    +
     # ---------------------------------------------------------------------------
     # Availability tests
     # ---------------------------------------------------------------------------

• #40605 fix(hindsight): send only new-turn delta on append retains instead of whole session

  merged to main

  files: plugins/memory/hindsight/__init__.py, tests/plugins/memory/test_hindsight_provider.py

  Show the diff

  • plugins/memory/hindsight/__init__.py

    @@ -575,6 +575,10 @@ def __init__(self):
             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
    +        # How many turns the last append-mode retain already shipped. Used to
    +        # send only the new delta on subsequent retains when the API supports
    +        # update_mode='append' (legacy/overwrite path still sends everything).
    +        self._last_retained_turn_count = 0
     
             # Recall controls
             self._auto_recall = True
    @@ -1119,6 +1123,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(
    @@ -1461,9 +1466,24 @@ 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)
    +
    +        # On append-capable APIs each retain only needs to ship the turns
    +        # accumulated since the last retain — the server appends them to the
    +        # existing document. On legacy/overwrite APIs we must resend the whole
    +        # session because each retain replaces the document.
    +        if update_mode == "append":
    +            turns_to_retain = self._session_turns[self._last_retained_turn_count:]
    +            if not turns_to_retain:
    +                logger.debug("sync_turn: skipped append retain; no new turns since last retain")
    +                return
    +        else:
    +            turns_to_retain = list(self._session_turns)
    +
    +        logger.debug("sync_turn: retaining %d/%d turns, payload %d chars",
    +                     len(turns_to_retain), len(self._session_turns),
    +                     sum(len(t) for t in turns_to_retain))
    +        content = "[" + ",".join(turns_to_retain) + "]"
     
             lineage_tags: list[str] = []
             if self._session_id:
    @@ -1474,11 +1494,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
    @@ -1509,6 +1528,10 @@ def _do_retain() -> None:
             self._ensure_writer()
             self._register_atexit()
             self._retain_queue.put(_do_retain)
    +        # Advance the append watermark only after the delta is queued, so a
    +        # later retain doesn't re-ship turns we've already handed to the writer.
    +        if update_mode == "append":
    +            self._last_retained_turn_count = len(self._session_turns)
     
         def get_tool_schemas(self) -> List[Dict[str, Any]]:
             if self._memory_mode == "context":
    @@ -1706,6 +1729,7 @@ def _flush():
             self._session_turns = []
             self._turn_counter = 0
             self._turn_index = 0
    +        self._last_retained_turn_count = 0
             logger.debug(
                 "Hindsight on_session_switch: new_session=%s parent=%s reset=%s doc=%s",
                 self._session_id, self._parent_session_id, reset, self._document_id,

  • tests/plugins/memory/test_hindsight_provider.py

    @@ -780,8 +780,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/overwrite APIs (no update_mode=append) resend the ENTIRE session each retain."""
             p = provider_with_config(retain_every_n_turns=2)
     
             p.sync_turn("turn1-user", "turn1-asst")
    @@ -795,12 +795,59 @@ 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
    +        # Without append support the document is overwritten, so it must
    +        # contain ALL turns from the session.
             assert "turn1-user" in content
             assert "turn2-user" in content
             assert "turn3-user" in content
             assert "turn4-user" in content
     
    +    def test_sync_turn_appends_only_delta_when_append_supported(self, provider_with_config, monkeypatch):
    +        """On append-capable APIs each retain ships only the new turns, not the whole session."""
    +        monkeypatch.setattr(
    +            "plugins.memory.hindsight._fetch_hindsight_api_version",
    +            lambda *a, **kw: "0.5.6",
    +        )
    +        from plugins.memory.hindsight import _append_capability_cache, _append_capability_lock
    +        # Clear before AND after: the capability cache is module-global and keyed
    +        # per api_url, so a stale entry would leak into other tests.
    +        with _append_capability_lock:
    +            _append_capability_cache.clear()
    +        try:
    +            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 = p._client.aretain_batch.call_args.kwargs
    +            first_item = first["items"][0]
    +            assert first["document_id"] == "test-session"
    +            assert first_item["update_mode"] == "append"
    +            assert "turn1-user" in first_item["content"]
    +            assert "turn2-user" in first_item["content"]
    +
    +            p._client.aretain_batch.reset_mock()
    +
    +            p.sync_turn("turn3-user", "turn3-asst")
    +            p.sync_turn("turn4-user", "turn4-asst")
    +            p._retain_queue.join()
    +
    +            second = p._client.aretain_batch.call_args.kwargs
    +            second_item = second["items"][0]
    +            assert second["document_id"] == "test-session"
    +            assert second_item["update_mode"] == "append"
    +            # Only the delta — the already-retained turns must NOT be resent.
    +            assert "turn1-user" not in second_item["content"]
    +            assert "turn2-user" not in second_item["content"]
    +            assert "turn3-user" in second_item["content"]
    +            assert "turn4-user" in second_item["content"]
    +            # message_count reflects only the delta (2 turns -> 4 messages).
    +            assert second_item["metadata"]["message_count"] == "4"
    +        finally:
    +            with _append_capability_lock:
    +                _append_capability_cache.clear()
    +
         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")