fix(supermemory): honor auto capture across lifecycle flushes

plugins/memory/supermemory/__init__.py

@@ -750,7 +750,7 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
         self._session_turns.append({"user": clean_user, "assistant": clean_assistant})
 
     def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
-        if not self._active or not self._write_enabled or not self._client or not self._session_id:
+        if not self._active or not self._auto_capture or not self._write_enabled or not self._client or not self._session_id:
             return
         cleaned = []
         for message in messages or []:
@@ -791,9 +791,10 @@ def on_session_switch(
         **kwargs,
     ) -> None:
         """Flush any buffered turns from the old session as one document, then reset for the new session."""
-        if not self._active or not self._write_enabled or not self._client:
+        if not self._active or not self._auto_capture or not self._write_enabled or not self._client:
             self._session_id = str(new_session_id or "").strip() or self._session_id
             self._session_turns = []
+            self._turn_count = 0
             return
 
         old_session_id = self._session_id
@@ -851,7 +852,7 @@ def _run():
 
     def shutdown(self) -> None:
         # Emergency fallback (crashes only). Buffer is cleared on normal on_session_end().
-        if self._active and self._write_enabled and self._client and self._session_turns and self._session_id:
+        if self._active and self._auto_capture and self._write_enabled and self._client and self._session_turns and self._session_id:
             logger.warning("Supermemory: Saving session via shutdown (session=%s, turns=%d)", self._session_id, len(self._session_turns))
 
             messages: list[dict] = []

tests/plugins/memory/test_supermemory_autocapture_guard.py

@@ -0,0 +1,112 @@
+"""Regression coverage for the ``auto_capture: false`` lifecycle contract.
+
+Automatic session-end, session-switch, and shutdown ingestion must remain off
+when capture is disabled. Explicit user-intent saves remain available.
+"""
+
+import json
+from typing import Any
+
+import pytest
+
+from plugins.memory.supermemory import SupermemoryMemoryProvider
+
+TRANSCRIPT = [
+    {"role": "user", "content": "Explain the compaction boundary behavior in detail."},
+    {"role": "assistant", "content": "Compaction rewrites the transcript in place."},
+]
+BUFFERED_TURNS = [
+    {"user": "A buffered user request worth capturing.", "assistant": "A buffered assistant response worth capturing."},
+]
+
+
+class FakeClient:
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.ingest_calls: list[dict] = []
+        self.add_calls: list[dict] = []
+
+    def ingest_conversation(self, session_id: str, messages: list[dict], metadata=None):
+        self.ingest_calls.append({"session_id": session_id, "messages": messages, "metadata": metadata})
+
+    def add_memory(self, content: str, *args: Any, **kwargs: Any) -> dict:
+        self.add_calls.append({"content": content, "args": args, "kwargs": kwargs})
+        return {"id": "mem_123"}
+
+
+def _make_provider(monkeypatch, tmp_path, *, auto_capture: bool):
+    monkeypatch.setenv("SUPERMEMORY_API_KEY", "test-key")
+    monkeypatch.setattr("plugins.memory.supermemory._SupermemoryClient", FakeClient)
+    (tmp_path / "supermemory.json").write_text(
+        json.dumps({"container_tag": "hermes_solar", "auto_capture": auto_capture}), encoding="utf-8"
+    )
+    provider = SupermemoryMemoryProvider()
+    provider.initialize("session-solar", hermes_home=str(tmp_path), platform="cli")
+    return provider
+
+
+@pytest.fixture
+def disabled(monkeypatch, tmp_path):
+    return _make_provider(monkeypatch, tmp_path, auto_capture=False)
+
+
+@pytest.fixture
+def enabled(monkeypatch, tmp_path):
+    return _make_provider(monkeypatch, tmp_path, auto_capture=True)
+
+
+def test_config_flag_is_loaded(disabled, enabled):
+    assert disabled._auto_capture is False
+    assert enabled._auto_capture is True
+
+
+def test_session_end_does_not_ingest_when_auto_capture_disabled(disabled):
+    disabled.on_session_end(TRANSCRIPT)
+    assert disabled._client.ingest_calls == []
+
+
+def test_compression_style_repeated_session_end_does_not_ingest_when_disabled(disabled):
+    for messages in (TRANSCRIPT, TRANSCRIPT[:1], TRANSCRIPT):
+        disabled.on_session_end(messages)
+    assert disabled._client.ingest_calls == []
+
+
+def test_session_switch_does_not_ingest_when_auto_capture_disabled(disabled):
+    disabled.on_turn_start(9, "outgoing session turn")
+    disabled._session_turns[:] = BUFFERED_TURNS
+    disabled.on_session_switch("session-solar-2")
+    assert disabled._client.ingest_calls == []
+    assert disabled._session_id == "session-solar-2"
+    assert disabled._session_turns == []
+    assert disabled._turn_count == 0
+
+
+def test_shutdown_does_not_ingest_when_auto_capture_disabled(disabled):
+    disabled._session_turns[:] = BUFFERED_TURNS
+    disabled.shutdown()
+    assert disabled._client.ingest_calls == []
+
+
+def test_enabled_preserves_all_automatic_lifecycle_ingests(enabled):
+    enabled.on_session_end(TRANSCRIPT)
+    enabled._session_turns[:] = BUFFERED_TURNS
+    enabled.on_session_switch("session-solar-2")
+    enabled._session_turns[:] = BUFFERED_TURNS
+    enabled.shutdown()
+    calls = enabled._client.ingest_calls
+    assert [call["session_id"] for call in calls] == ["session-solar", "session-solar", "session-solar-2"]
+    assert calls[0]["metadata"]["type"] == "full_session"
+    assert calls[1]["metadata"]["partial"] is True
+    assert calls[2]["metadata"]["partial"] is True
+
+
+def test_explicit_store_tool_writes_when_auto_capture_disabled(disabled):
+    result = json.loads(disabled._tool_store({"content": "Kylan prefers Canada-first market framing."}))
+    assert result["saved"] is True
+    assert len(disabled._client.add_calls) == 1
+
+
+def test_on_memory_write_writes_when_auto_capture_disabled(disabled):
+    disabled.on_memory_write("add", "MEMORY.md", "Kylan prefers Canada-first market framing.")
+    if disabled._write_thread is not None:
+        disabled._write_thread.join(timeout=5)
+    assert len(disabled._client.add_calls) == 1