From 31d662963c546814d39b6720242fd630b67f6d9b Mon Sep 17 00:00:00 2001 From: Sora-bluesky Date: Sun, 20 Sep 2026 09:40:05 +0900 Subject: [PATCH] refactor(agent): consolidate SOUL.md identity resolution and pin drift propagation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Editing SOUL.md never reached a continuing session: the restore path reuses the persisted system prompt verbatim, and _stored_prompt_matches_runtime only rejects Model/Provider/cwd/Platform drift, not identity content drift (#68563). Consolidate identity resolution into resolve_identity_block() (used by the prompt builder) and pin SOUL.md drift propagation with regression tests. The compaction-gate staleness check this PR originally added was superseded by #98426's unconditional rebuild at the compaction boundary: a drifted identity now simply makes the fresh build differ from the cached prompt, which already routes to the rebuild branch on its own. stored_identity_is_stale() remains as a tested helper — anchored on the identity block plus HERMES_AGENT_HELP_GUIDANCE, and reporting provenance and checkability the same way resolve_identity_block does — but it has no production caller today. Restore path I/O is unchanged: with prompt caching off it never reads SOUL.md and never rewrites stored prompt bytes; with caching on, reconstruct_static_prefix rebuilds the static parts (and so reads SOUL.md) exactly as on main. Both are pinned by regression tests. Co-Authored-By: Claude Fable 5.1 --- agent/system_prompt.py | 107 ++- tests/agent/test_system_prompt_restore.py | 805 ++++++++++++++++++++++ 2 files changed, 906 insertions(+), 6 deletions(-) diff --git a/agent/system_prompt.py b/agent/system_prompt.py index 9d5c5e1ecccd7..873c3c9079577 100644 --- a/agent/system_prompt.py +++ b/agent/system_prompt.py @@ -173,6 +173,69 @@ def _plugin_section_blocks(sections: tuple, position: str) -> List[str]: return [block] if block else [] +def _resolve_context_length(agent: Any) -> Optional[int]: + """Resolve the model context window used for context-file caps. + + Stable for the life of the conversation, so it does not threaten prompt + caching. ``None`` falls back to the historical flat loader default.""" + compressor = getattr(agent, "context_compressor", None) + if compressor is not None: + context_length = getattr(compressor, "context_length", None) + if isinstance(context_length, int) and context_length > 0: + return context_length + return None + + +def resolve_identity_block( + agent: Any, ctx_len: Optional[int] = None +) -> Dict[str, Any]: + """Resolve the identity block (slot #1) exactly as the prompt builder does. + + Returns ``{"text": str, "from_soul": bool, "checkable": bool}``. + + An explicit ``ctx_len`` preserves the prompt builder's context-file cap. + When omitted, resolve the stable context length from the agent. + + ``from_soul`` preserves the builder's ``soul_loaded`` semantics (it controls + whether SOUL.md is injected again as project context), and it is provenance, + not text comparison: a SOUL.md equal to ``DEFAULT_AGENT_IDENTITY`` still + counts as loaded. + + ``checkable`` distinguishes the legitimate default-identity states (an + absent or readable-empty SOUL.md) from cases where identity cannot be + judged safely (an unreadable SOUL.md or an absent/unmounted Hermes home). + Callers must fail open to reuse when it is false. + """ + text = None + from_soul = False + checkable = True + if agent.load_soul_identity or not agent.skip_context_files: + home = _agent_home(agent) + context_length = ( + ctx_len if ctx_len is not None else _resolve_context_length(agent) + ) + soul_content = _pb.load_soul_md(context_length, home_override=home) + if soul_content: + text = soul_content + from_soul = True + else: + try: + probe_home = home or get_hermes_home() + if not probe_home.exists(): + checkable = False + else: + soul_path = probe_home / "SOUL.md" + if soul_path.exists(): + # A successful read distinguishes the documented + # readable-empty reset state from an unreadable file. + soul_path.read_text(encoding="utf-8") + except Exception: + checkable = False + if text is None: + text = DEFAULT_AGENT_IDENTITY + return {"text": text, "from_soul": from_soul, "checkable": checkable} + + def _session_start_like(agent: Any, now: Any) -> Any: """Best-known conversation start time, or ``now`` as a fallback. ``Conversation started:`` must be byte-stable across rebuilds (compression, @@ -542,10 +605,10 @@ def _memory_parts(agent: Any) -> List[str]: def _identity_parts(agent: Any, ctx_len: Optional[int]) -> Tuple[List[str], bool]: """SOUL.md (primary identity; cron keeps the persona while skipping cwd instructions, scoped to the agent's OWN home) or the default identity. - Returns ``(parts, soul_loaded)``.""" - wants_soul = agent.load_soul_identity or not agent.skip_context_files - _soul_content = _pb.load_soul_md(ctx_len, home_override=_agent_home(agent)) if wants_soul else None - return ([_soul_content], True) if _soul_content else ([DEFAULT_AGENT_IDENTITY], False) + Returns ``(parts, soul_loaded)``. The explicit ``ctx_len`` must reach the + loader unchanged to preserve the prompt builder's historical output.""" + identity = resolve_identity_block(agent, ctx_len) + return [identity["text"]], identity["from_soul"] def _guidance_parts(agent: Any) -> List[str]: @@ -665,8 +728,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) shared context file can remain in the longest common prefix across worktrees. Never re-rendered mid-session.""" # Model context window scales the context-file caps; stable per conversation. - _cc_len = getattr(getattr(agent, "context_compressor", None), "context_length", None) - _ctx_len = _cc_len if isinstance(_cc_len, int) and _cc_len > 0 else None + _ctx_len = _resolve_context_length(agent) # ── Stable tier ──────────────────────────────────────────────── stable_parts, _soul_loaded = _identity_parts(agent, _ctx_len) # The skill_view() pointer dangles without skill tools OR without the @@ -752,6 +814,39 @@ def invalidate_system_prompt(agent: Any) -> None: agent._memory_store.load_from_disk() +def stored_identity_is_stale(agent: Any, stored_prompt: str) -> bool: + """Return whether a stored prompt's opening identity differs from the fresh one. + + Identity is slot #1. One of the two renderer-owned Hermes help-guidance + variants follows it, so the pair supplies the boundary that a substring + comparison lacks (for example, deleting only the tail of SOUL.md must still + be detected). + + Fail open to reuse when identity is not checkable (including a failed read + or an absent/unmounted Hermes home), the resolver raises, or the stored + prompt has neither normal help-guidance anchor. Those states do not + establish a confident identity mismatch. + """ + try: + identity = resolve_identity_block(agent) + if not (identity["checkable"] and identity["text"]): + return False + help_anchors = ( + HERMES_AGENT_HELP_GUIDANCE.strip(), + HERMES_AGENT_HELP_GUIDANCE_NO_SKILLS.strip(), + ) + if not any(anchor in stored_prompt for anchor in help_anchors): + return False + identity_prefix = identity["text"].strip() + "\n\n" + return not any( + stored_prompt.startswith(identity_prefix + anchor) + for anchor in help_anchors + ) + except Exception: + logger.debug("identity staleness check failed", exc_info=True) + return False + + def reconstruct_static_prefix(agent: Any, system_message: Optional[str] = None, *, log_label: str = "restore") -> None: """Reconstruct ``_cached_system_prompt_static`` for a stored prompt. Only the full prompt is persisted, so restore / keep-prompt compression / diff --git a/tests/agent/test_system_prompt_restore.py b/tests/agent/test_system_prompt_restore.py index 55fdfe6a1bba5..44e9878175cb2 100644 --- a/tests/agent/test_system_prompt_restore.py +++ b/tests/agent/test_system_prompt_restore.py @@ -22,6 +22,7 @@ from agent.conversation_loop import _restore_or_build_system_prompt from agent.surface_switch import _SURFACE_NAME_END, _SURFACE_SWITCH_NOTE_PREFIX, identity_line_value +from agent.system_prompt import stored_identity_is_stale def _make_agent(session_db=None, prebuilt_prompt: str = "BUILT_PROMPT"): @@ -391,6 +392,106 @@ def test_restored_prompt_is_byte_identical_to_stored(self): assert agent._cached_system_prompt.encode("utf-8") == stored.encode("utf-8") +# --------------------------------------------------------------------------- +# PR #72253 redesign (v6) — restore-path contract pins (§6-C). Restore never +# rewrites the stored prompt on identity drift: AGENTS.md:19-23 forbids +# mid-conversation system-prompt rewrites outside the explicit compression +# exception, and restore is not that exception. agent/conversation_loop.py +# was reverted to main's behavior, so _restore_or_build_system_prompt no +# longer calls stored_identity_is_stale at all; identity drift is instead +# handled at the compaction keep-prompt gate (TestCompactionIdentityDriftGate). +# See .claude/archive/72253-redesign-v6.md §6-C for the design rationale. +# --------------------------------------------------------------------------- + + +class TestV6RedesignFailBeforePins: + def test_soul_drift_does_not_rewrite_stored_prompt(self, monkeypatch, caplog): + """v6 §1/§6-C-1: SOUL.md drift on restore must NOT rebuild or persist — + AGENTS.md:19-23 forbids mid-conversation system-prompt rewrites outside + the explicit compression exception. Restore is not that exception. + + _restore_or_build_system_prompt no longer imports or calls + stored_identity_is_stale at all (agent/conversation_loop.py was fully + reverted to main's behavior), so this reuse holds unconditionally. + """ + stored = _stored_with_identity("OLD SOUL IDENTITY") + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + _patch_identity(monkeypatch, "NEW SOUL IDENTITY") + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + db.update_system_prompt.assert_not_called() + + def test_restore_without_prompt_caching_does_not_read_soul_md( + self, monkeypatch + ): + """With prompt caching disabled, restore must not re-read SOUL.md. + + ``_restore_or_build_system_prompt`` no longer calls + ``stored_identity_is_stale``, and ``reconstruct_static_prefix`` returns + at its prompt-caching guard before its failed-rebuild memoization guard + matters. The real resolver below proves the read path is genuinely + unreached rather than mocked away. + """ + identity = "CURRENT SOUL IDENTITY" + stored = _stored_with_identity(identity) + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + agent.load_soul_identity = True + agent.skip_context_files = False + + # Undo the file's autouse neutral-resolver patch for this test only, + # so the real resolver (and therefore the real load_soul_md call) is + # actually exercised — a fully-mocked resolver would make this pin + # vacuous (it would pass regardless of what the restore path does). + monkeypatch.setattr( + "agent.system_prompt.resolve_identity_block", _REAL_RESOLVE_IDENTITY + ) + soul_reader = MagicMock(return_value=identity) + monkeypatch.setattr("agent.prompt_builder.load_soul_md", soul_reader) + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + soul_reader.assert_not_called() + + def test_restore_with_prompt_caching_reads_soul_md_via_static_prefix( + self, monkeypatch + ): + """With prompt caching enabled, upstream static-prefix reconstruction + rebuilds prompt parts and therefore reads SOUL.md.""" + identity = "CURRENT SOUL IDENTITY" + stored = _stored_with_identity(identity) + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + agent.load_soul_identity = True + agent.skip_context_files = False + agent._use_prompt_caching = True + agent._cached_system_prompt_static = None + agent._static_rebuild_failed_for = None + + monkeypatch.setattr( + "agent.system_prompt.resolve_identity_block", _REAL_RESOLVE_IDENTITY + ) + soul_reader = MagicMock(return_value=identity) + monkeypatch.setattr("agent.prompt_builder.load_soul_md", soul_reader) + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + soul_reader.assert_called_once() + + # --------------------------------------------------------------------------- # Cross-session static prefix reconstruction (issue #68191 follow-up) # --------------------------------------------------------------------------- @@ -546,6 +647,710 @@ def test_success_clears_failure_memo_and_early_returns(self): assert build.call_count == 1 assert agent._cached_system_prompt_static == stable assert getattr(agent, "_static_rebuild_failed_for", None) is None +# --------------------------------------------------------------------------- +# Identity (SOUL.md) staleness on restore — issue #68563 +# --------------------------------------------------------------------------- + +from agent.prompt_builder import HERMES_AGENT_HELP_GUIDANCE as _HELP +# Captured at import time, BEFORE the neutral autouse fixture patches the +# module attribute — the classification tests exercise the real function. +from agent.system_prompt import resolve_identity_block as _REAL_RESOLVE_IDENTITY + + +@pytest.fixture(autouse=True) +def _neutral_identity_resolver(monkeypatch): + """Neutralize ``resolve_identity_block`` (#68563) for every test below. + + Restore itself never calls this resolver — it reuses the stored prompt + verbatim (see TestV6RedesignFailBeforePins). But the prompt builder and + ``stored_identity_is_stale`` (a tested standalone helper with no + production caller since #98426's unconditional rebuild absorbed drift + propagation) both call it, and several tests below build a real + ``AIAgent`` or invoke that helper directly. Running the real resolver + against these MagicMock agents would read the test HERMES_HOME and + randomly flip the decision, so default it to "no basis to judge" (empty + text -> check skipped); the staleness tests below override it with + explicit values.""" + monkeypatch.setattr( + "agent.system_prompt.resolve_identity_block", + lambda agent, _ctx_len=None: { + "text": "", + "from_soul": False, + "checkable": True, + }, + ) + + +def _stored_with_identity(identity: str, tail: str = "per-session context") -> str: + """Assemble a stored prompt the way the builder joins the stable tier.""" + return identity.strip() + "\n\n" + _HELP.strip() + "\n\n" + tail + + +def _patch_identity(monkeypatch, text, checkable=True, from_soul=True): + monkeypatch.setattr( + "agent.system_prompt.resolve_identity_block", + lambda agent, _ctx_len=None: { + "text": text, + "from_soul": from_soul, + "checkable": checkable, + }, + ) + + +class TestIdentityStalenessRebuild: + """Restore-path pins asserting that identity drift does not rebuild. + + These tests cover only verbatim reuse of the restored prompt; they do not + assert that restore detects staleness. ``AGENTS.md:19-23`` exempts context + compression, not restore, from the "never rebuild mid-conversation" rule. + Direct stale-identity classification is covered separately by + ``TestStoredIdentityIsStale``, while propagation occurs at the Hermes-native + compaction boundary in ``TestCompactionIdentityDriftGate``. + """ + + def test_edited_soul_reuses_verbatim(self, monkeypatch, caplog): + stored = _stored_with_identity("OLD SOUL IDENTITY") + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + _patch_identity(monkeypatch, "NEW SOUL IDENTITY") + + with caplog.at_level(logging.INFO, logger="agent.conversation_loop"): + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + db.update_system_prompt.assert_not_called() + + def test_trailing_deletion_from_soul_reuses_verbatim(self, monkeypatch): + """Deleting the TAIL of SOUL.md leaves the new block a prefix of the + old one — restore doesn't compare them at all anymore, so the + stored prompt is kept regardless.""" + stored = _stored_with_identity("You are concise.\nNever disclose secrets.") + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + _patch_identity(monkeypatch, "You are concise.") + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + + def test_deleted_soul_reuses_verbatim(self, monkeypatch): + """SOUL.md removed → the resolver would return the hardcoded + default, but restore never consults it, so the stored (SOUL-built) + prompt is kept as-is.""" + stored = _stored_with_identity("CUSTOM SOUL IDENTITY") + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + _patch_identity( + monkeypatch, "DEFAULT HARDCODED IDENTITY", from_soul=False + ) + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + + def test_matching_identity_reuses_verbatim(self, monkeypatch): + identity = "CURRENT SOUL IDENTITY" + stored = _stored_with_identity(identity) + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + _patch_identity(monkeypatch, identity) + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + db.update_system_prompt.assert_not_called() + + def test_unreadable_soul_fails_open_to_reuse(self, monkeypatch): + """checkable=False = SOUL.md exists but could not be read. Declaring + staleness would persist a default-identity downgrade over a healthy + custom identity, so the check must fail open to reuse.""" + stored = _stored_with_identity("CUSTOM SOUL IDENTITY") + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + _patch_identity( + monkeypatch, "DEFAULT HARDCODED IDENTITY", + checkable=False, from_soul=False, + ) + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + + def test_resolver_exception_fails_open_to_reuse(self, monkeypatch): + stored = _stored_with_identity("CUSTOM SOUL IDENTITY") + db = MagicMock() + db.get_session.return_value = {"system_prompt": stored} + agent = _make_agent(session_db=db) + + def _boom(agent): + raise RuntimeError("resolver crashed") + + monkeypatch.setattr("agent.system_prompt.resolve_identity_block", _boom) + + _restore_or_build_system_prompt( + agent, None, [{"role": "user", "content": "hi"}] + ) + + assert agent._cached_system_prompt == stored + agent._build_system_prompt.assert_not_called() + + +class TestStoredIdentityIsStale: + """Direct unit tests of ``stored_identity_is_stale()`` itself (v6 §6-F). + + It has no production caller. It was originally wired into the + compaction keep-prompt gate, but #98426's unconditional rebuild at that + boundary replaced the containment-based gate this function attached to + — a drifted identity now simply makes the fresh build differ from the + cached prompt, which already routes to the rebuild branch on its own, + so the gate stopped calling this helper (see + ``TestCompactionIdentityDriftGate``). Restore never called it either — + it reuses the stored prompt verbatim (see + ``TestV6RedesignFailBeforePins``). These pins keep the anchored-detection + and fail-open properties covered as a standalone, tested library + function in case a future caller needs one.""" + + def test_trailing_deletion_is_detected_as_stale(self, monkeypatch): + """Deleting the TAIL of SOUL.md leaves the new block a prefix of the + old one — a bare containment check would still "match". The anchor + (help guidance immediately after the identity) catches it.""" + stored = _stored_with_identity("You are concise.\nNever disclose secrets.") + _patch_identity(monkeypatch, "You are concise.") + + assert stored_identity_is_stale(MagicMock(), stored) is True + + def test_undetermined_identity_fails_open(self, monkeypatch): + """checkable=True but text="" (no basis to judge, e.g. the file's + own autouse neutral-resolver default) must not be called stale.""" + stored = _stored_with_identity("CUSTOM SOUL IDENTITY") + _patch_identity(monkeypatch, "", checkable=True, from_soul=False) + + assert stored_identity_is_stale(MagicMock(), stored) is False + + def test_unreadable_soul_fails_open(self, monkeypatch): + """checkable=False = SOUL.md exists but could not be read. Declaring + staleness on unreadable input is not safe, so this must fail open.""" + stored = _stored_with_identity("CUSTOM SOUL IDENTITY") + _patch_identity( + monkeypatch, "DEFAULT HARDCODED IDENTITY", + checkable=False, from_soul=False, + ) + + assert stored_identity_is_stale(MagicMock(), stored) is False + + def test_resolver_exception_fails_open(self, monkeypatch): + stored = _stored_with_identity("CUSTOM SOUL IDENTITY") + + def _boom(agent): + raise RuntimeError("resolver crashed") + + monkeypatch.setattr("agent.system_prompt.resolve_identity_block", _boom) + + assert stored_identity_is_stale(MagicMock(), stored) is False + + +class TestResolveIdentityBlockClassification: + """Real-resolver pins for the absent/empty/unreadable provenance split. + + A readable-but-empty SOUL.md is the documented way to reset to the default + personality, so it remains checkable. A failed read or an absent/unmounted + Hermes home makes staleness unjudgeable. + """ + + @staticmethod + def _resolver_agent(): + agent = MagicMock() + agent.load_soul_identity = True + agent.skip_context_files = False + agent.context_compressor = None + # No profile-scoped home: _agent_home must resolve to None so the + # resolver falls back to the ambient home these tests configure + # (a bare MagicMock would fabricate a nonexistent db-derived home + # and flip `checkable` — see #50233 home scoping). + agent._session_db = None + return agent + + @pytest.fixture(autouse=True) + def _no_seeding(self, monkeypatch): + # ensure_hermes_home may seed a default SOUL.md on first run; these + # tests pin the classification of a state the USER created, so the + # first-run seeding is out of scope and disabled. + monkeypatch.setattr( + "hermes_cli.config.ensure_hermes_home", lambda: None + ) + + def test_context_length_override_and_agent_fallback(self, monkeypatch): + agent = self._resolver_agent() + agent.context_compressor = MagicMock(context_length=8192) + soul_reader = MagicMock(return_value="CUSTOM IDENTITY") + monkeypatch.setattr("agent.prompt_builder.load_soul_md", soul_reader) + + explicit = _REAL_RESOLVE_IDENTITY(agent, 4096) + + assert explicit["text"] == "CUSTOM IDENTITY" + soul_reader.assert_called_once_with(4096, home_override=None) + + soul_reader.reset_mock() + fallback = _REAL_RESOLVE_IDENTITY(agent) + + assert fallback["text"] == "CUSTOM IDENTITY" + soul_reader.assert_called_once_with(8192, home_override=None) + + def test_readable_empty_soul_is_checkable_default(self): + from hermes_constants import get_hermes_home + + from agent.prompt_builder import DEFAULT_AGENT_IDENTITY + + soul = get_hermes_home() / "SOUL.md" + soul.parent.mkdir(parents=True, exist_ok=True) + soul.write_text(" \n\n ", encoding="utf-8") + + ident = _REAL_RESOLVE_IDENTITY(self._resolver_agent()) + + assert ident["checkable"] is True + assert ident["from_soul"] is False + assert ident["text"] == DEFAULT_AGENT_IDENTITY + + def test_undecodable_soul_is_not_checkable(self): + from hermes_constants import get_hermes_home + + + soul = get_hermes_home() / "SOUL.md" + soul.parent.mkdir(parents=True, exist_ok=True) + soul.write_bytes(b"\xff\xfe\x9c invalid utf-8 \x80") + + ident = _REAL_RESOLVE_IDENTITY(self._resolver_agent()) + + assert ident["checkable"] is False + assert ident["from_soul"] is False + + def test_absent_soul_is_checkable_default(self): + from hermes_constants import get_hermes_home + + from agent.prompt_builder import DEFAULT_AGENT_IDENTITY + + soul = get_hermes_home() / "SOUL.md" + assert not soul.exists() + + ident = _REAL_RESOLVE_IDENTITY(self._resolver_agent()) + + assert ident["checkable"] is True + assert ident["from_soul"] is False + assert ident["text"] == DEFAULT_AGENT_IDENTITY + + def test_absent_hermes_home_is_not_checkable(self, monkeypatch, tmp_path): + """v6 class 12: HERMES_HOME itself missing/unmounted is a different + failure mode than an existing HERMES_HOME with no SOUL.md inside it + (the case above). Nothing here can confirm what identity applies, so + this must NOT be checkable — fail open to reuse on restore, rather + than confidently reporting "default identity".""" + missing_home = tmp_path / "does_not_exist" + assert not missing_home.exists() + monkeypatch.setenv("HERMES_HOME", str(missing_home)) + + ident = _REAL_RESOLVE_IDENTITY(self._resolver_agent()) + + assert ident["checkable"] is False + assert ident["from_soul"] is False + + +# --------------------------------------------------------------------------- +# PR #72253 redesign (v6) — pins for the compaction keep-prompt gate +# (agent/conversation_compression.py). §5 applies identity drift detection +# ONLY at this gate (the one place AGENTS.md:19-23 already carves out as the +# explicit exception), not on the restore hot path above. +# +# Absorbed onto #98426's always-rebuild redesign (rb-72253 rebase, 2026-08): +# the gate no longer calls `stored_identity_is_stale` at all — #98426 made +# the prompt rebuild unconditional at every compaction, so a drifted +# identity now always surfaces as a byte mismatch between the fresh build +# and the cached prompt, which already routes to the rebuild branch on its +# own. `stored_identity_is_stale` survives as a standalone resolver (used +# and unit-tested above), not as a compaction-gate condition. The tests +# below pin the resulting behavior. +# See 72253-redesign-v6.md §6-D. +# --------------------------------------------------------------------------- + + +class TestCompactionIdentityDriftGate: + """Real AIAgent + real SessionDB, mirroring + tests/run_agent/test_in_place_compaction.py's harness, so compress_context + runs its real locking/persistence path instead of a mocked stand-in.""" + + def _make_agent(self, session_db, session_id, cached_prompt): + import os + from unittest.mock import patch as _patch + + with _patch.dict(os.environ, {"OPENROUTER_API_KEY": "tk"}): + from run_agent import AIAgent + + agent = AIAgent( + api_key="tk", + base_url="https://openrouter.ai/api/v1", + model="test/model", + quiet_mode=True, + session_db=session_db, + session_id=session_id, + skip_context_files=True, + skip_memory=True, + ) + agent.compression_in_place = True + agent._cached_system_prompt = cached_prompt + + def _fake_compress(messages, current_tokens=None, focus_topic=None, force=False): + return [ + {"role": "user", "content": "[CONTEXT COMPACTION] summary of prior turns"}, + {"role": "assistant", "content": "recent reply"}, + ] + + agent.context_compressor.compress = _fake_compress + agent.context_compressor._last_compress_aborted = False + agent.context_compressor._last_summary_error = None + agent.context_compressor.compression_count = 1 + return agent + + def _seed(self, db, sid, n=8): + db.create_session(sid, "cli", model="test/model") + for i in range(n): + db.append_message( + session_id=sid, + role="user" if i % 2 == 0 else "assistant", + content=f"msg {i}", + ) + + def test_compaction_keeps_prompt_when_identity_unchanged(self, monkeypatch): + """Object-identity preservation on a byte-equal rebuild (the + surviving form of the old keep-prompt KV-cache optimization) must + hold — regression pin, expected to PASS today. + + #98426 replaced the containment-based keep-prompt branch with an + unconditional rebuild at every compaction (agent/conversation_ + compression.py, always-rebuild arc #95681): ``_build_system_prompt`` + is now called on every compaction regardless of drift, and only + object identity — not invocation — is what survives when the fresh + build is byte-equal to the cached prompt. Asserting non-invocation + (the pre-rebase form of this pin) is no longer meaningful; see + test_compaction_rebuilds_prompt_when_soul_changed for the drift case. + """ + import tempfile + from pathlib import Path + + from hermes_state import SessionDB + from agent.conversation_compression import compress_context + + identity = "CURRENT SOUL IDENTITY" + resolver = MagicMock( + return_value={ + "text": identity, + "from_soul": True, + "checkable": True, + } + ) + monkeypatch.setattr( + "agent.system_prompt.resolve_identity_block", resolver + ) + + with tempfile.TemporaryDirectory() as tmp: + db = SessionDB(db_path=Path(tmp) / "t.db") + sid = "20260804_120000_keep01" + self._seed(db, sid) + agent = self._make_agent(db, sid, None) + agent.context_compressor.context_length = 8192 + monkeypatch.setattr( + "agent.system_prompt._resolve_context_length", + lambda _agent: 4096, + ) + stored = agent._build_system_prompt("sys") + assert identity in stored + agent._cached_system_prompt = stored + db.update_system_prompt(sid, stored) + resolver.reset_mock() + + _compressed, new_sp = compress_context( + agent, [{"role": "user", "content": "x"}] * 8, + approx_tokens=100_000, system_message="sys", + ) + + assert new_sp == stored + assert new_sp is stored, ( + "byte-equal rebuild must preserve the cached object identity" + ) + resolver.assert_called_once_with(agent, 4096) + assert db.get_session(sid)["system_prompt"] == stored + db.close() + + def test_compaction_rebuilds_prompt_when_soul_changed(self, monkeypatch): + """v6 §5: identity drift reaches the persisted prompt at the + compaction boundary — this is the one place AGENTS.md:19-23 already + allows the system prompt to change mid-conversation. + + Post-#98426 the compaction gate rebuilds unconditionally and only + keeps the cached object when the fresh build is byte-equal to it + (see agent/conversation_compression.py); it no longer calls + `stored_identity_is_stale` at all. Drift falls through to the + rebuild branch simply because a drifted identity makes the fresh + build differ from the cached bytes — the byte comparison alone is + enough, with no separate staleness check required. + """ + import tempfile + from pathlib import Path + + from hermes_state import SessionDB + from agent.conversation_compression import compress_context + + old_identity = "OLD SOUL IDENTITY" + new_identity = "NEW SOUL IDENTITY" + resolver = MagicMock( + return_value={ + "text": old_identity, + "from_soul": True, + "checkable": True, + } + ) + monkeypatch.setattr( + "agent.system_prompt.resolve_identity_block", resolver + ) + + with tempfile.TemporaryDirectory() as tmp: + db = SessionDB(db_path=Path(tmp) / "t.db") + sid = "20260804_120100_drift01" + self._seed(db, sid) + agent = self._make_agent(db, sid, None) + stored = agent._build_system_prompt("sys") + assert old_identity in stored + agent._cached_system_prompt = stored + db.update_system_prompt(sid, stored) + resolver.return_value = { + "text": new_identity, + "from_soul": True, + "checkable": True, + } + resolver.reset_mock() + + _compressed, new_sp = compress_context( + agent, [{"role": "user", "content": "x"}] * 8, + approx_tokens=100_000, system_message="sys", + ) + + assert new_identity in new_sp + assert new_sp != stored, ( + "expected the real builder to replace the stale identity; " + f"got the stored prompt unchanged: {new_sp!r}" + ) + resolver.assert_called_once_with( + agent, agent.context_compressor.context_length + ) + assert db.get_session(sid)["system_prompt"] == new_sp + db.close() + + def test_adopted_child_path_passes_through_on_drift(self, monkeypatch): + """v6 §4.2 (F6): when another process has already rotated this + session via its own compression, the adopting call returns the + CHILD's own already-persisted prompt unchanged — never a rebuild, + even if that prompt would be judged stale by the current resolver. + ``_adopt_live_compression_child`` sets ``agent._cached_system_prompt`` + to the child's value before this call site reads it, so identity + drift on top of that must not trigger a rebuild here. This documents + the intentional pass-through rather than relying on it silently.""" + import tempfile + from pathlib import Path + + from hermes_state import SessionDB + import agent.conversation_compression as cc_module + + child_prompt = _stored_with_identity("CHILD PERSISTED IDENTITY") + _patch_identity(monkeypatch, "SOME OTHER NEW IDENTITY") + rebuild = MagicMock(return_value="SHOULD_NOT_BE_USED") + + def _fake_rotated(db, sid): + return True + + def _fake_adopt(agent, db, sid): + agent._cached_system_prompt = child_prompt + return [{"role": "user", "content": "recovered from child"}] + + monkeypatch.setattr( + cc_module, "_session_was_rotated_by_compression", _fake_rotated + ) + monkeypatch.setattr( + cc_module, "_adopt_live_compression_child", _fake_adopt + ) + + with tempfile.TemporaryDirectory() as tmp: + db = SessionDB(db_path=Path(tmp) / "t.db") + sid = "20260804_120200_adopt01" + self._seed(db, sid) + agent = self._make_agent(db, sid, "PARENT STALE PROMPT") + agent._build_system_prompt = rebuild + + _compressed, new_sp = cc_module.compress_context( + agent, [{"role": "user", "content": "x"}] * 8, + approx_tokens=100_000, system_message="sys", + ) + + assert new_sp == child_prompt, ( + "adoption path must pass through the child's persisted " + f"prompt unchanged despite drift; got {new_sp!r}" + ) + rebuild.assert_not_called() + db.close() + + +# --------------------------------------------------------------------------- +# PR #72253 redesign (v6) — pass-through pins for the CAS (codex_app_server) +# compression path, which this PR intentionally leaves untouched (v6 §3.2). +# That path has no system-prompt write-back on `main` or here, so adding +# drift detection to it would change the in-flight prompt for one turn and +# then lose it on the next restore. These pins document that limitation +# rather than relying on it silently. See 72253-redesign-v6.md §6-D item 5. +# --------------------------------------------------------------------------- + + +class TestCasPassThroughOnDrift: + """Mirrors tests/run_agent/test_codex_app_server_compaction.py's + DummyAgent/FakeCodexSession pattern — the CAS dispatch in + compress_context never acquires the compression lock or touches + session_db, so a lightweight fake agent (rather than the real + AIAgent+SessionDB harness above) is enough to reach every return form.""" + + class _FakeCodexSession: + def __init__(self, result): + self.result = result + self.calls = 0 + + def compact_thread(self): + self.calls += 1 + return self.result + + def close(self): + pass + + @staticmethod + def _cas_agent(cached_prompt, *, auto_compaction, codex_session): + from types import SimpleNamespace + + agent = MagicMock() + agent.api_mode = "codex_app_server" + agent.codex_app_server_auto_compaction = auto_compaction + agent.session_id = "cas-drift-session" + agent.platform = "cli" + agent._cached_system_prompt = cached_prompt + agent._codex_session = codex_session + agent._build_system_prompt = MagicMock(return_value="SHOULD_NOT_BE_BUILT") + agent.context_compressor = SimpleNamespace( + compression_count=0, + last_compression_rough_tokens=0, + last_prompt_tokens=0, + last_completion_tokens=0, + awaiting_real_usage_after_compression=False, + ) + return agent + + @staticmethod + def _run(agent, force): + from agent.conversation_compression import compress_context + + return compress_context( + agent, [{"role": "user", "content": "hi"}], "sys", + approx_tokens=100_000, force=force, + ) + + def test_skip_native_mode_passes_through_on_drift(self, monkeypatch): + stored = _stored_with_identity("OLD SOUL IDENTITY") + _patch_identity(monkeypatch, "NEW SOUL IDENTITY") + agent = self._cas_agent( + stored, auto_compaction="native", codex_session=None + ) + + _messages, prompt = self._run(agent, force=False) + + assert prompt == stored + agent._build_system_prompt.assert_not_called() + + def test_thread_absent_passes_through_on_drift(self, monkeypatch): + stored = _stored_with_identity("OLD SOUL IDENTITY") + _patch_identity(monkeypatch, "NEW SOUL IDENTITY") + agent = self._cas_agent( + stored, auto_compaction="hermes", codex_session=None + ) + + _messages, prompt = self._run(agent, force=True) + + assert prompt == stored + agent._build_system_prompt.assert_not_called() + + def test_failure_passes_through_on_drift(self, monkeypatch): + from agent.transports.codex_app_server_session import TurnResult + + stored = _stored_with_identity("OLD SOUL IDENTITY") + _patch_identity(monkeypatch, "NEW SOUL IDENTITY") + session = self._FakeCodexSession(TurnResult(interrupted=True)) + agent = self._cas_agent( + stored, auto_compaction="hermes", codex_session=session + ) + + _messages, prompt = self._run(agent, force=True) + + assert prompt == stored + assert session.calls == 1 + agent._build_system_prompt.assert_not_called() + + def test_success_passes_through_on_drift(self, monkeypatch): + from agent.transports.codex_app_server_session import TurnResult + + stored = _stored_with_identity("OLD SOUL IDENTITY") + _patch_identity(monkeypatch, "NEW SOUL IDENTITY") + session = self._FakeCodexSession(TurnResult(compacted=True)) + agent = self._cas_agent( + stored, auto_compaction="hermes", codex_session=session + ) + + _messages, prompt = self._run(agent, force=True) + + assert prompt == stored, ( + "CAS success has no system-prompt write-back (v6 F1 residual) " + "— identity drift must not reach the returned prompt even on a " + f"successful compaction; got {prompt!r}" + ) + assert session.calls == 1 + agent._build_system_prompt.assert_not_called() + + +# --------------------------------------------------------------------------- +# TestSessionStartHookGuard was removed from this file (v6 §6-B). It pinned +# on_session_start firing/not-firing across the identity-stale, fresh-build, +# and preview-restart restore states. This PR only touches the compaction +# keep-prompt gate, not restore, so the identity-stale restore state it was +# guarding no longer exists here — it belongs to a follow-up PR that fixes +# the pre-existing on_session_start double-fire bug for the stale_runtime +# and null/empty stored-prompt classes in +# agent/conversation_loop.py's restore path (unrelated to SOUL.md identity). +# That follow-up PR must carry forward the preview-restart boundary case +# this class used to pin (a brand-new session that legitimately receives +# nonempty PARENT history but has no session DB): a guard keyed on history +# truthiness instead of the specific stale states would wrongly suppress +# that session's on_session_start hook. +# --------------------------------------------------------------------------- class TestPerResponseSessionWritePath: