Compare commits

...
Author SHA1 Message Date
Nicolò Boschi 87dc621631 fix(claude-code): fix config tests after USER_CONFIG_PATH removal
Update tests to use HOME env var override instead of monkeypatching
the removed USER_CONFIG_PATH constant. Add autouse fixture to
TestLoadConfig to isolate all config tests from real user config
and HINDSIGHT_* env vars.
2026-03-26 11:36:33 +01:00
Nicolò Boschi d1b96c5e73 fix(claude-code): respect retainEveryNTurns in full-session mode
The turn-count gating was only applied in chunked mode, meaning
full-session mode would re-ingest the entire transcript on every
single Stop event. Now retainEveryNTurns gates both modes.

Also fix test isolation: resolve ~/.hindsight/claude-code.json at
call time (not module load) so HOME override in tests works correctly.
2026-03-26 11:29:47 +01:00
Nicolò Boschi 1e7e50ccd3 feat(claude-code): full-session retain mode with document upsert and configurable tags
Switch default retain behavior from per-turn chunks to full-session upsert.
Each session is now retained as a single document (document_id = session_id)
that gets updated on every Stop event, instead of creating fragmented
documents with timestamp-suffixed IDs.

New config options:
- retainMode: "full-session" (default) or "chunked" (legacy)
- retainTags: list with template variable support ({session_id}, {bank_id}, {timestamp})
- retainMetadata: extra metadata dict merged with built-in fields, supports templates
2026-03-25 18:45:34 +01:00
6 changed files with 200 additions and 25 deletions
@@ -105,6 +105,7 @@ class HindsightClient:
document_id: str = "conversation",
context: Optional[str] = None,
metadata: Optional[dict] = None,
tags: Optional[list] = None,
timeout: int = 15,
) -> dict:
"""Retain content into a bank's memory.
@@ -121,6 +122,8 @@ class HindsightClient:
}
if context:
item["context"] = context
if tags:
item["tags"] = tags
body = {
"items": [item],
"async": True,
@@ -25,10 +25,13 @@ DEFAULTS = {
"recallTopK": None,
# Retain
"autoRetain": True,
"retainMode": "full-session",
"retainRoles": ["user", "assistant"],
"retainEveryNTurns": 10,
"retainOverlapTurns": 2,
"retainContext": "claude-code",
"retainTags": [],
"retainMetadata": {},
# Connection
"hindsightApiUrl": None,
"hindsightApiToken": None,
@@ -60,6 +63,7 @@ ENV_OVERRIDES = {
"HINDSIGHT_AGENT_NAME": ("agentName", str),
"HINDSIGHT_AUTO_RECALL": ("autoRecall", bool),
"HINDSIGHT_AUTO_RETAIN": ("autoRetain", bool),
"HINDSIGHT_RETAIN_MODE": ("retainMode", str),
"HINDSIGHT_RECALL_BUDGET": ("recallBudget", str),
"HINDSIGHT_RECALL_MAX_TOKENS": ("recallMaxTokens", int),
"HINDSIGHT_RECALL_MAX_QUERY_CHARS": ("recallMaxQueryChars", int),
@@ -100,9 +104,6 @@ def _load_settings_file(path: str, config: dict) -> None:
debug_log(config, f"Failed to load {path}: {e}")
USER_CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".hindsight", "claude-code.json")
def load_config() -> dict:
"""Load plugin configuration from settings.json + env overrides.
@@ -125,7 +126,8 @@ def load_config() -> dict:
_load_settings_file(os.path.join(plugin_root, "settings.json"), config)
# 2. User config — stable, version-independent, matches openclaw convention
_load_settings_file(USER_CONFIG_PATH, config)
user_config_path = os.path.join(os.path.expanduser("~"), ".hindsight", "claude-code.json")
_load_settings_file(user_config_path, config)
# Apply environment variable overrides
for env_name, (key, typ) in ENV_OVERRIDES.items():
@@ -96,11 +96,13 @@ def main():
debug_log(config, f"Read {len(all_messages)} messages from transcript")
# Chunked retention logic — port of Openclaw's retainEveryNTurns + sliding window
# Retention mode: full session (default) or chunked (legacy)
retain_mode = config.get("retainMode", "full-session")
retain_every_n = max(1, config.get("retainEveryNTurns", 1))
retain_full_window = False
messages_to_retain = all_messages
# Respect retainEveryNTurns in both modes
if retain_every_n > 1:
turn_count = increment_turn_count(session_id)
if turn_count % retain_every_n != 0:
@@ -108,6 +110,7 @@ def main():
debug_log(config, f"Turn {turn_count}/{retain_every_n}, skipping retain (next at turn {next_at})")
return
if retain_mode == "chunked" and retain_every_n > 1:
# Sliding window: N turns + configured overlap
overlap_turns = config.get("retainOverlapTurns", 0)
window_turns = retain_every_n + overlap_turns
@@ -115,9 +118,12 @@ def main():
retain_full_window = True
debug_log(
config,
f"Turn {turn_count}: chunked retain firing "
f"(window: {window_turns} turns, {len(messages_to_retain)} messages)",
f"Chunked retain firing (window: {window_turns} turns, {len(messages_to_retain)} messages)",
)
else:
# Full session mode: retain all messages, always as full window
retain_full_window = True
debug_log(config, f"Full session retain: {len(all_messages)} messages")
# Format transcript
retain_roles = config.get("retainRoles", ["user", "assistant"])
@@ -148,12 +154,44 @@ def main():
bank_id = derive_bank_id(hook_input, config)
ensure_bank_mission(client, bank_id, config, debug_fn=_dbg)
# Unique document ID — mirrors Openclaw: {sessionKey}-{timestamp}
document_id = f"{session_id}-{int(time.time() * 1000)}"
# Document ID: use session_id so the same session always upserts the same document.
# In chunked mode, append timestamp to create distinct documents per chunk.
if retain_mode == "chunked" and retain_every_n > 1:
document_id = f"{session_id}-{int(time.time() * 1000)}"
else:
document_id = session_id
# Resolve template variables in tags and metadata.
# Supported variables: {session_id}, {bank_id}, {timestamp}
template_vars = {
"session_id": session_id,
"bank_id": bank_id,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
def _resolve_template(value: str) -> str:
for k, v in template_vars.items():
value = value.replace(f"{{{k}}}", v)
return value
# Tags from config with template resolution
raw_tags = config.get("retainTags", [])
tags = [_resolve_template(t) for t in raw_tags] if raw_tags else None
# Metadata: merge built-in defaults with user-configured extras
metadata = {
"retained_at": template_vars["timestamp"],
"message_count": str(message_count),
"session_id": session_id,
}
for k, v in config.get("retainMetadata", {}).items():
metadata[k] = _resolve_template(str(v))
debug_log(
config, f"Retaining to bank '{bank_id}', doc '{document_id}', {message_count} messages, {len(transcript)} chars"
)
if tags:
debug_log(config, f"Tags: {tags}")
# POST to Hindsight retain API
try:
@@ -162,11 +200,8 @@ def main():
content=transcript,
document_id=document_id,
context=config.get("retainContext", "claude-code"),
metadata={
"retained_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"message_count": str(message_count),
"session_id": session_id,
},
metadata=metadata,
tags=tags,
timeout=15,
)
debug_log(config, f"Retain response: {json.dumps(response)[:200]}")
@@ -5,6 +5,7 @@
"retainMission": "Extract technical decisions, architectural choices, user preferences, project context, and people/tool relationships. Ignore routine greetings and transient operational details.",
"autoRecall": true,
"autoRetain": true,
"retainMode": "full-session",
"recallBudget": "mid",
"recallMaxTokens": 1024,
"recallTypes": ["world", "experience"],
@@ -16,6 +17,8 @@
"retainRoles": ["user", "assistant"],
"retainEveryNTurns": 10,
"retainOverlapTurns": 2,
"retainTags": ["{session_id}"],
"retainMetadata": {},
"retainContext": "claude-code",
"hindsightApiToken": null,
"apiPort": 9077,
@@ -28,6 +28,14 @@ class TestCastEnv:
class TestLoadConfig:
@pytest.fixture(autouse=True)
def _isolate_config(self, tmp_path, monkeypatch):
"""Isolate from real user config and env vars."""
monkeypatch.setenv("HOME", str(tmp_path))
for k in list(os.environ):
if k.startswith("HINDSIGHT_"):
monkeypatch.delenv(k, raising=False)
def test_defaults_applied_when_no_settings_file(self, tmp_path, monkeypatch):
monkeypatch.setenv("CLAUDE_PLUGIN_ROOT", str(tmp_path))
# No settings.json in tmp_path
@@ -95,27 +103,26 @@ class TestLoadConfig:
user_cfg.write_text(json.dumps({"recallBudget": "high"}))
monkeypatch.setenv("CLAUDE_PLUGIN_ROOT", str(plugin_root))
import lib.config as cfg_mod
monkeypatch.setattr(cfg_mod, "USER_CONFIG_PATH", str(user_cfg))
monkeypatch.setenv("HOME", str(tmp_path))
cfg = load_config()
assert cfg["recallBudget"] == "high"
def test_user_config_missing_falls_back_gracefully(self, tmp_path, monkeypatch):
monkeypatch.setenv("CLAUDE_PLUGIN_ROOT", str(tmp_path))
import lib.config as cfg_mod
monkeypatch.setattr(cfg_mod, "USER_CONFIG_PATH", str(tmp_path / "nonexistent.json"))
# HOME points to tmp_path where no .hindsight/claude-code.json exists
monkeypatch.setenv("HOME", str(tmp_path))
cfg = load_config()
assert cfg["recallBudget"] == "mid" # default
def test_env_var_wins_over_user_config(self, tmp_path, monkeypatch):
plugin_root = tmp_path / "plugin"
plugin_root.mkdir()
user_cfg = tmp_path / "claude-code.json"
user_cfg.write_text(json.dumps({"recallBudget": "low"}))
user_cfg_dir = tmp_path / ".hindsight"
user_cfg_dir.mkdir()
(user_cfg_dir / "claude-code.json").write_text(json.dumps({"recallBudget": "low"}))
monkeypatch.setenv("CLAUDE_PLUGIN_ROOT", str(plugin_root))
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("HINDSIGHT_RECALL_BUDGET", "high")
import lib.config as cfg_mod
monkeypatch.setattr(cfg_mod, "USER_CONFIG_PATH", str(user_cfg))
cfg = load_config()
assert cfg["recallBudget"] == "high"
@@ -25,7 +25,7 @@ from conftest import FakeHTTPResponse, make_hook_input, make_memory, make_transc
# ---------------------------------------------------------------------------
def _run_hook(module_name, hook_input, monkeypatch, tmp_path, urlopen_side_effect=None, extra_env=None):
def _run_hook(module_name, hook_input, monkeypatch, tmp_path, urlopen_side_effect=None, extra_env=None, extra_settings=None):
"""Import and run a hook script's main() with mocked stdin/stdout/HTTP."""
# Isolated plugin dirs
monkeypatch.setenv("CLAUDE_PLUGIN_ROOT", str(tmp_path / "plugin_root"))
@@ -33,16 +33,19 @@ def _run_hook(module_name, hook_input, monkeypatch, tmp_path, urlopen_side_effec
(tmp_path / "plugin_root").mkdir(exist_ok=True)
(tmp_path / "plugin_data").mkdir(exist_ok=True)
# Strip real HINDSIGHT_* env vars
# Strip real HINDSIGHT_* env vars and neutralize user config (~/.hindsight/claude-code.json)
for k in list(os.environ):
if k.startswith("HINDSIGHT_"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("HOME", str(tmp_path))
for k, v in (extra_env or {}).items():
monkeypatch.setenv(k, v)
# Write a minimal settings.json enabling fast retains
settings = {"autoRecall": True, "autoRetain": True, "retainEveryNTurns": 1, "hindsightApiUrl": "http://fake:9077"}
if extra_settings:
settings.update(extra_settings)
(tmp_path / "plugin_root" / "settings.json").write_text(json.dumps(settings))
stdin_data = io.StringIO(json.dumps(hook_input))
@@ -251,13 +254,133 @@ class TestRetainHook:
assert "old memories" not in content
assert "actual question" in content
def test_retain_tags_with_template_variables(self, monkeypatch, tmp_path):
"""retainTags config should resolve template variables like {session_id}."""
messages = [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "world"}]
transcript = make_transcript_file(tmp_path, messages)
hook_input = make_hook_input(transcript_path=transcript, session_id="sess-tag-test")
captured = {}
def capture(req, timeout=None):
if "/memories" in req.full_url and "/recall" not in req.full_url:
captured["body"] = json.loads(req.data.decode())
return FakeHTTPResponse({})
_run_hook(
"retain", hook_input, monkeypatch, tmp_path,
urlopen_side_effect=capture,
extra_settings={"retainTags": ["{session_id}", "claude-code", "custom-tag"]},
)
assert "body" in captured, "retain API was not called"
item = captured["body"]["items"][0]
assert item["tags"] == ["sess-tag-test", "claude-code", "custom-tag"]
def test_retain_custom_metadata(self, monkeypatch, tmp_path):
"""retainMetadata config should be merged with built-in metadata."""
messages = [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "world"}]
transcript = make_transcript_file(tmp_path, messages)
hook_input = make_hook_input(transcript_path=transcript, session_id="sess-meta-test")
captured = {}
def capture(req, timeout=None):
if "/memories" in req.full_url and "/recall" not in req.full_url:
captured["body"] = json.loads(req.data.decode())
return FakeHTTPResponse({})
_run_hook(
"retain", hook_input, monkeypatch, tmp_path,
urlopen_side_effect=capture,
extra_settings={"retainMetadata": {"project": "my-project", "session": "{session_id}"}},
)
assert "body" in captured, "retain API was not called"
meta = captured["body"]["items"][0]["metadata"]
# Built-in metadata
assert meta["session_id"] == "sess-meta-test"
assert "retained_at" in meta
# Custom metadata with template resolution
assert meta["project"] == "my-project"
assert meta["session"] == "sess-meta-test"
def test_full_session_uses_session_id_as_document_id(self, monkeypatch, tmp_path):
"""In full-session mode, document_id should be the session_id (for upsert)."""
messages = [
{"role": "user", "content": "first question"},
{"role": "assistant", "content": "first answer"},
{"role": "user", "content": "second question"},
{"role": "assistant", "content": "second answer"},
]
transcript = make_transcript_file(tmp_path, messages)
hook_input = make_hook_input(transcript_path=transcript, session_id="sess-full-123")
captured = {}
def capture(req, timeout=None):
if "/memories" in req.full_url and "/recall" not in req.full_url:
captured["body"] = json.loads(req.data.decode())
return FakeHTTPResponse({})
_run_hook("retain", hook_input, monkeypatch, tmp_path, urlopen_side_effect=capture)
assert "body" in captured, "retain API was not called"
item = captured["body"]["items"][0]
# document_id should be just the session_id, no timestamp suffix
assert item["document_id"] == "sess-full-123"
# Should contain ALL messages, not just the last turn
assert "first question" in item["content"]
assert "second question" in item["content"]
def test_full_session_respects_retain_every_n_turns(self, monkeypatch, tmp_path):
"""In full-session mode, retainEveryNTurns should still gate when retain fires."""
messages = [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "world"}]
transcript = make_transcript_file(tmp_path, messages)
hook_input = make_hook_input(transcript_path=transcript, session_id="sess-throttle")
captured = {}
def capture(req, timeout=None):
if "/memories" in req.full_url and "/recall" not in req.full_url:
captured["called"] = True
captured["body"] = json.loads(req.data.decode())
return FakeHTTPResponse({})
# retainEveryNTurns=3 in full-session mode — first 2 calls should be skipped
_run_hook(
"retain", hook_input, monkeypatch, tmp_path,
urlopen_side_effect=capture,
extra_settings={"retainEveryNTurns": 3},
)
# Turn 1 of 3 — should NOT retain
assert "called" not in captured
# Turn 2 — still skip
captured.clear()
_run_hook(
"retain", hook_input, monkeypatch, tmp_path,
urlopen_side_effect=capture,
extra_settings={"retainEveryNTurns": 3},
)
assert "called" not in captured
# Turn 3 — should fire, with full session content and session_id as doc ID
captured.clear()
_run_hook(
"retain", hook_input, monkeypatch, tmp_path,
urlopen_side_effect=capture,
extra_settings={"retainEveryNTurns": 3},
)
assert "called" in captured, "retain should fire on turn 3"
item = captured["body"]["items"][0]
assert item["document_id"] == "sess-throttle" # full-session uses session_id
assert "hello" in item["content"]
def test_chunked_retain_skips_below_threshold(self, monkeypatch, tmp_path):
"""With retainEveryNTurns=5, first call should be skipped."""
"""With retainEveryNTurns=5 and retainMode=chunked, first call should be skipped."""
(tmp_path / "plugin_root").mkdir(exist_ok=True)
(tmp_path / "plugin_data").mkdir(exist_ok=True)
settings = {
"autoRetain": True,
"autoRecall": True,
"retainMode": "chunked",
"retainEveryNTurns": 5,
"hindsightApiUrl": "http://fake:9077",
}
@@ -279,6 +402,8 @@ class TestRetainHook:
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("CLAUDE_PLUGIN_ROOT", str(tmp_path / "plugin_root"))
monkeypatch.setenv("CLAUDE_PLUGIN_DATA", str(tmp_path / "plugin_data"))
monkeypatch.setenv("HINDSIGHT_RETAIN_MODE", "chunked")
monkeypatch.setenv("HOME", str(tmp_path))
stdin_data = io.StringIO(json.dumps(hook_input))
scripts_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "scripts"))