Initial open source release

tests/coding/test_scene_coding_variant.py

@@ -0,0 +1,573 @@
+# -*- coding: utf-8 -*-
+"""
+scene="coding" —— coding 链路的软件工程变体。
+
+与 scene="working" 的关系: 同一条链路(同一个 store / 表 / reconciler), 差别只有两处,
+本文件按这两处分组:
+  1. 抽取规范 —— EXTRACT_PROMPT_CODING(TestPromptSelection / TestPromptContent)
+  2. 版本演进链层 —— chain_store(TestChainObserve / TestChainDecorate)
+
+外加路由本身(TestSceneRouting)和"不传 scene 行为不变"的回归闸门(TestWorkingUnchanged)。
+
+⚠️ 这里全部是零 LLM / 零 embedding 的测试。链层本身不调模型, 抽取规范只验"选了哪份
+prompt", 不验模型读了之后的行为 —— 那个要上 CodeMemoryBench 跑。
+"""
+import os
+import tempfile
+
+import pytest
+
+from hy_memory.client import HyMemoryClient
+from hy_memory.coding.chain_store import HyCodeMem
+from hy_memory.coding.extractor import (
+    EXTRACT_PROMPT,
+    EXTRACT_PROMPT_CODING,
+    CodingMemoryExtractor,
+)
+
+
+class _FakeMsg:
+    """最小 ChatMessage 替身 —— 只要 _chain_text 读的那几个字段。"""
+
+    def __init__(self, role, content, tool_name=None):
+        self.role = role
+        self.content = content
+        self.tool_name = tool_name
+        self.tool_calls = None
+
+    def is_tool_message(self):
+        return self.role == "tool"
+
+
+class _FakeProvider:
+    config = None
+    _llm_config = None
+
+
+def _edit_msgs(said, old, new, path="qutebrowser/config/configtypes.py"):
+    """一轮"用户说要改 + agent 写文件"。
+
+    工具消息写成 `path=/old=/new=` 三段是刻意的: 链层在 git 反查落空时靠这个形态
+    合成 diff(_patch_from_edits)。线上大多数场景给不了本地仓路径, 走的就是这条。
+    """
+    return [
+        _FakeMsg("user", said),
+        _FakeMsg(
+            "tool",
+            f"path='{path}'\nold='''{old}'''\nnew='''{new}'''",
+            tool_name="edit",
+        ),
+    ]
+
+
+# ════════════════════════════════════════════════════════════════════
+# 1. 路由
+# ════════════════════════════════════════════════════════════════════
+
+
+class TestSceneRouting:
+    def test_coding_routes_to_coding_chain(self):
+        # "coding" 和 "working" 归一成同一个路由值 —— add()/search() 里所有
+        # `scene_forced == "coding"` 的判断对两者同时生效, 不必逐处加分支。
+        assert HyMemoryClient._normalize_scene("coding") == "coding"
+        assert HyMemoryClient._normalize_scene("working") == "coding"
+
+    def test_chat_unaffected(self):
+        assert HyMemoryClient._normalize_scene("chat") == "chat"
+
+    def test_none_stays_auto(self):
+        assert HyMemoryClient._normalize_scene(None) is None
+
+    def test_case_and_whitespace_insensitive(self):
+        assert HyMemoryClient._normalize_scene("  CODING ") == "coding"
+        assert HyMemoryClient._scene_variant(" Coding ") == "coding"
+
+    def test_illegal_value_raises(self):
+        with pytest.raises(ValueError) as ei:
+            HyMemoryClient._normalize_scene("productivity")
+        # 报错要把三个合法值都列出来 —— 只说 'chat' or 'working' 会让人以为
+        # coding 不存在。
+        assert "coding" in str(ei.value)
+
+    def test_variant_is_orthogonal_to_route(self):
+        # 同一个路由值("coding")下两个变体; 变体只由 _scene_variant 决定。
+        assert HyMemoryClient._scene_variant("coding") == "coding"
+        assert HyMemoryClient._scene_variant("working") == "working"
+        assert HyMemoryClient._scene_variant("chat") == "working"
+        assert HyMemoryClient._scene_variant(None) == "working"
+
+
+# ════════════════════════════════════════════════════════════════════
+# 2. 抽取规范的选择
+# ════════════════════════════════════════════════════════════════════
+
+
+class TestPromptSelection:
+    def test_default_is_working(self):
+        # ★ 回归闸门: 不传 scene 必须还是 working 那份。已落盘的 working 评测
+        #   靠这条保证不被动。
+        e = CodingMemoryExtractor(_FakeProvider())
+        assert e.scene == "working"
+        assert e._prompt_tmpl is EXTRACT_PROMPT
+
+    def test_coding_picks_coding_prompt(self):
+        e = CodingMemoryExtractor(_FakeProvider(), scene="coding")
+        assert e.scene == "coding"
+        assert e._prompt_tmpl is EXTRACT_PROMPT_CODING
+
+    def test_unknown_scene_falls_back_to_working(self):
+        # 不抛错: 抽取是一次性的, 这段轨迹不会再回来, 为一个拼错的参数把整段丢掉
+        # 不值得(同 extract() 里 LLM 失败走原文兜底的取舍)。
+        e = CodingMemoryExtractor(_FakeProvider(), scene="bogus")
+        assert e.scene == "working"
+        assert e._prompt_tmpl is EXTRACT_PROMPT
+
+    def test_both_prompts_share_placeholders(self):
+        # 两份必须能被同一份 .format() 参数填 —— extract() 只有一处 format 调用。
+        kw = dict(workspace_present="true", branch_present="true",
+                  conversation_block="x")
+        assert EXTRACT_PROMPT.format(**kw)
+        assert EXTRACT_PROMPT_CODING.format(**kw)
+
+
+class TestPromptContent:
+    """两份规范各自独有的段落。钉住的是**内容边界**, 不是措辞质量。"""
+
+    def test_coding_has_hard_constraint_header(self):
+        # 前置覆盖后文任何"想要叙事"的措辞。
+        assert "HARD CONSTRAINT" in EXTRACT_PROMPT_CODING
+        assert "HARD CONSTRAINT" not in EXTRACT_PROMPT
+        # 三条核心要求
+        assert "FUTURE AGENT needs in order to act" in EXTRACT_PROMPT_CODING
+        assert "never \"the user\"" in EXTRACT_PROMPT_CODING
+        assert "the session was completed" in EXTRACT_PROMPT_CODING
+
+    def test_coding_has_code_work_section(self):
+        assert "CODE WORK" in EXTRACT_PROMPT_CODING
+        assert "CODE WORK" not in EXTRACT_PROMPT
+        # 四件未来 agent 无法从代码反推的东西
+        assert "BRANCH / WORKING LOCATION" in EXTRACT_PROMPT_CODING
+        assert "WHICH VERSION WON" in EXTRACT_PROMPT_CODING
+        assert "EXACT COORDINATES, NOT PASTED CODE" in EXTRACT_PROMPT_CODING
+        assert "VERIFIED vs MERELY WRITTEN" in EXTRACT_PROMPT_CODING
+
+    def test_coding_marks_superseded_version_phrases(self):
+        # 用户改主意的语言标记 —— 链层那边也认同一批词。
+        assert "scratch that" in EXTRACT_PROMPT_CODING
+
+    def test_working_keeps_the_ask_user_exclusion(self):
+        # working 独有: 卡在向用户要参数时强制 experiences: []。
+        # coding 侧刻意不带 —— 那边的轨迹是 agent 自主执行, 不存在这种停顿。
+        assert "ASKING THE USER" in EXTRACT_PROMPT
+        assert "ASKING THE USER" not in EXTRACT_PROMPT_CODING
+
+    def test_both_keep_evidence_fidelity(self):
+        # 这条是两份共有的底线: 只记真执行过的动作。任何一份丢了都是缺陷。
+        for p in (EXTRACT_PROMPT, EXTRACT_PROMPT_CODING):
+            assert "only actions actually executed" in p
+            assert "Never promote a partial search to" in p
+
+
+# ════════════════════════════════════════════════════════════════════
+# 3. 版本演进链层 —— 写
+# ════════════════════════════════════════════════════════════════════
+
+
+@pytest.fixture
+def chain():
+    with tempfile.TemporaryDirectory() as d:
+        yield HyCodeMem(os.path.join(d, "chain"))
+
+
+class TestChainText:
+    """client._chain_text: messages → 链层能读的纯文本。"""
+
+    def test_renders_lowercase_role_prefixes(self):
+        # 形态必须对齐链层的 _USER_LC / _DIFF_END 认的那套小写前缀。
+        t = HyMemoryClient._chain_text(
+            [_FakeMsg("user", "hi"), _FakeMsg("assistant", "ok"),
+             _FakeMsg("tool", "out", tool_name="bash")]
+        )
+        assert t.startswith("user: hi")
+        assert "assistant: ok" in t
+        assert "[tool:bash] out" in t
+
+    def test_no_index_prefix(self):
+        # ⚠️ 不能复用 extractor._format_conversation —— 那个给每行加 `[i] ` 下标
+        # (给 LLM 读的), 而链层的正则锚在行首, 前缀一加全部匹配不上。
+        t = HyMemoryClient._chain_text([_FakeMsg("user", "hi")])
+        assert not t.startswith("[0]")
+
+    def test_tolerates_empty_and_missing_content(self):
+        t = HyMemoryClient._chain_text(
+            [_FakeMsg("user", None), _FakeMsg("assistant", "")]
+        )
+        assert isinstance(t, str)
+
+    def test_empty_input(self):
+        assert HyMemoryClient._chain_text([]) == ""
+        assert HyMemoryClient._chain_text(None) == ""
+
+
+class TestChainObserve:
+    def test_extracts_patch_from_dialog_without_git(self):
+        # 没有本地仓也要能存下来 —— git 反查落空时从对话里的 old/new 合成。
+        # agent 修改文件后不一定立即提交，因此不能只依赖 commit diff。
+        with tempfile.TemporaryDirectory() as d:
+            H = HyCodeMem(os.path.join(d, "c"))   # repo_root=None
+            n = H.observe(
+                HyMemoryClient._chain_text(
+                    _edit_msgs("On branch feat/q4, set mult.",
+                               "mult = 255.0\n", "mult = 360.0\n")),
+                meta={"branch": "feat/q4"},
+            )
+            assert n >= 1
+            assert H.stats()["threads"] == 1
+
+    def test_multiple_versions_land_on_one_thread(self, chain):
+        # ★ 这层存在的理由: 同一处代码被改了几版, 每版的原话 + 真实 patch 都留着。
+        for said, old, new in [
+            ("On branch feat/q4, first pass: 255.0 everywhere.",
+             "mult = 255.0\n", "mult = 360.0\n"),
+            ("Hold on, scratch that — special-case only hue.",
+             "mult = 360.0\n", "mult = 360.0 if kind == 'h' else 255.0\n"),
+        ]:
+            chain.observe(HyMemoryClient._chain_text(_edit_msgs(said, old, new)),
+                          meta={"branch": "feat/q4"})
+        st = chain.stats()
+        assert st["threads"] == 1, "同一符号的多版必须并到一条链, 否则版本序号会从 1 重数"
+        assert st["rounds"] == 2
+        assert st["multi"] == 1
+
+    def test_branch_from_meta_beats_guessing(self, chain):
+        chain.observe(
+            HyMemoryClient._chain_text(
+                _edit_msgs("no branch token here", "a = 1\n", "a = 2\n")),
+            meta={"branch": "fix/explicit"},
+        )
+        assert chain.stats()["branches"] == 1
+
+    def test_no_code_change_stores_nothing(self, chain):
+        # 纯对话没有可识别的改动 → 0。这是正常情况, 不是错误。
+        assert chain.observe(
+            HyMemoryClient._chain_text([_FakeMsg("user", "早上好")])) == 0
+        assert chain.stats()["rounds"] == 0
+
+    def test_empty_text_is_noop(self, chain):
+        assert chain.observe("") == 0
+        assert chain.observe(None) == 0
+
+
+# ════════════════════════════════════════════════════════════════════
+# 4. 版本演进链层 —— 读
+# ════════════════════════════════════════════════════════════════════
+
+
+def _seed_two_versions(H, branch="feat/q4",
+                       path="qutebrowser/config/configtypes.py"):
+    for said, old, new in [
+        ("On branch %s, first pass: 255.0 everywhere." % branch,
+         "mult = 255.0\n", "mult = 360.0\n"),
+        ("Hold on, scratch that — special-case only hue.",
+         "mult = 360.0\n", "mult = 360.0 if kind == 'h' else 255.0\n"),
+    ]:
+        H.observe(HyMemoryClient._chain_text(_edit_msgs(said, old, new, path=path)),
+                  meta={"branch": branch})
+
+
+_FACT = {
+    "statement": ("On branch feat/q4, in qutebrowser/config/configtypes.py "
+                  "QtColor._parse_value: settled on the hue multiplier."),
+    "branch": "feat/q4",
+    "files": ["qutebrowser/config/configtypes.py"],
+}
+
+
+class TestChainDecorate:
+    def test_block_carries_branch_and_all_versions(self, chain):
+        _seed_two_versions(chain)
+        blk, used = chain.decorate([dict(_FACT)], instruction="which version won?")
+        assert used == [0]
+        # 分支名必须在块里 —— agent 在错分支上改的代码会被静默丢掉。
+        assert "feat/q4" in blk
+        # 两版都在, 不只留最后一版: 遇到被弃那版的代码要认得出来, 否则会"修"回去。
+        assert "第1版" in blk and "第2版" in blk
+        assert "255.0" in blk and "360.0" in blk
+
+    def test_fact_without_branch_falls_back_to_files(self, chain):
+        # 事实层的 fact 可能没有 branch 字段。branch 认不出就按 files 退回，
+        # 不能直接放弃，否则 diff 块会始终为空。
+        _seed_two_versions(chain)
+        fa = {"statement": "the hue multiplier decision",
+              "files": ["qutebrowser/config/configtypes.py"]}
+        blk, used = chain.decorate([fa], instruction="hue multiplier")
+        assert used == [0]
+        assert blk
+
+    def test_unrelated_fact_gets_no_block(self, chain):
+        _seed_two_versions(chain)
+        blk, used = chain.decorate(
+            [{"statement": "something else entirely", "files": ["other/file.py"]}],
+            instruction="x")
+        assert used == []
+        assert blk == ""
+
+    def test_empty_facts(self, chain):
+        assert chain.decorate([], instruction="x") == ("", [])
+
+    def test_block_off_switch_disables_layer(self, chain):
+        # 消融开关: 让 coding 退化成"纯向量双层"(= working 的召回形态), 用来做
+        # 单变量 A/B。CodeMemoryBench 消融靠它: 去掉这层 77/216 → 45/216。
+        _seed_two_versions(chain)
+        os.environ["HCM_BLOCK_OFF"] = "1"
+        try:
+            assert chain.decorate([dict(_FACT)], instruction="x") == ("", [])
+        finally:
+            os.environ.pop("HCM_BLOCK_OFF", None)
+
+
+class TestDecorateGlue:
+    """client._decorate_with_chain —— 胶水层的 fail-safe。"""
+
+    def test_returns_empty_when_chain_unavailable(self):
+        import types
+        ns = types.SimpleNamespace(
+            _chain_store=None, _chain_store_failed=True,
+            _get_chain_store=lambda owner="": None,
+        )
+        assert HyMemoryClient._decorate_with_chain(ns, [dict(_FACT)]) == ("", [])
+
+    def test_store_exception_degrades_not_raises(self):
+        # 这层是增量, 挂了要退回纯向量双层而不是把整次 search 拖失败。
+        import types
+
+        class _Boom:
+            def decorate(self, *a, **k):
+                raise RuntimeError("boom")
+
+        ns = types.SimpleNamespace(_get_chain_store=lambda owner="": _Boom())
+        assert HyMemoryClient._decorate_with_chain(ns, [dict(_FACT)]) == ("", [])
+
+
+class TestObserveGlue:
+    def test_returns_zero_when_chain_unavailable(self):
+        import types
+        ns = types.SimpleNamespace(_get_chain_store=lambda owner="": None)
+        assert HyMemoryClient._observe_chain(ns, [_FakeMsg("user", "hi")]) == 0
+
+    def test_store_exception_degrades_not_raises(self):
+        import types
+
+        class _Boom:
+            def observe(self, *a, **k):
+                raise RuntimeError("boom")
+
+        ns = types.SimpleNamespace(_get_chain_store=lambda owner="": _Boom())
+        n = HyMemoryClient._observe_chain(
+            ns, _edit_msgs("x", "a = 1\n", "a = 2\n"))
+        assert n == 0
+
+
+# ════════════════════════════════════════════════════════════════════
+# 5. working 不受影响
+# ════════════════════════════════════════════════════════════════════
+
+
+class TestWorkingUnchanged:
+    """★ 整个改动的前提: working 路一个字节都不动。"""
+
+    def test_working_prompt_byte_length_pinned(self):
+        # 钉住长度而不是内容: 内容有自己的断言(tests/coding/test_extractor.py 30+ 条),
+        # 这里防的是"改 coding 时手滑动了 working 那份"。
+        #
+        # ⚠️ 这个数字必须用 `len(EXTRACT_PROMPT)` 实测得到, 不要从 diff / 正则抽取的
+        #    片段长度推 —— 正则取到的 body 不含尾换行等边界字符, 按它填会差几个字节,
+        #    于是这条断言在完全没改 working 的情况下也会红(我第一版就栽在这)。
+        assert len(EXTRACT_PROMPT) == 18517, (
+            "working 的抽取规范被改动了。它带着自己的实测结论(statement 长度三版对照, "
+            "见其正文), 动它要先复现那批 A/B。"
+        )
+
+    def test_coding_prompt_is_separate_object(self):
+        assert EXTRACT_PROMPT is not EXTRACT_PROMPT_CODING
+
+    def test_working_variant_does_not_touch_chain(self):
+        # variant="working" 时写入侧不该调链层 —— 否则 working 的库里会多出
+        # hy_code_chain 目录, 且行为与改动前不一致。
+        assert HyMemoryClient._scene_variant("working") == "working"
+        assert HyMemoryClient._scene_variant(None) == "working"
+
+
+# ════════════════════════════════════════════════════════════════════
+# 6. 两个变体的库分开
+# ════════════════════════════════════════════════════════════════════
+
+
+class TestStoreSeparation:
+    """★ working 和 coding 读写**不同的库**。
+
+    为什么必须分库而不是加一个 variant 列做过滤: 检索是**先**在向量集合里取 top-K、
+    **再**用 sqlite 过滤。两个变体的 fact 混在同一个集合里时, 向量那一步就已经混了 ——
+    后面加任何 SQL 条件都救不回来, 因为 top-K 的名额已经被另一个变体占掉了。
+    """
+
+    def test_sqlite_path_diverges(self):
+        from hy_memory.coding.store import _variant_db_path
+        assert _variant_db_path("/d/coding_memory.db", "_cv") == "/d/coding_memory_cv.db"
+
+    def test_working_paths_byte_identical(self):
+        # ★ 回归闸门: working 的路径和集合名必须与加变体之前逐字相同, 否则已落盘的
+        #   库读不到(表现为"库空了")。
+        from hy_memory.coding.store import (
+            _variant_db_path, _derive_fact_collection_name,
+            _derive_coding_collection_name,
+        )
+        assert _variant_db_path("/d/coding_memory.db", "") == "/d/coding_memory.db"
+        assert _derive_fact_collection_name("am", "") == "am_working_facts"
+        assert _derive_coding_collection_name("am", "") == "am_coding_keys"
+
+    def test_collections_diverge(self):
+        from hy_memory.coding.store import (
+            _derive_fact_collection_name, _derive_coding_collection_name,
+        )
+        assert _derive_fact_collection_name("am", "_cv") == "am_working_facts_cv"
+        assert _derive_coding_collection_name("am", "_cv") == "am_coding_keys_cv"
+
+
+# ════════════════════════════════════════════════════════════════════
+# 7. 链层的数据域隔离
+# ════════════════════════════════════════════════════════════════════
+
+
+class TestChainOwnerIsolation:
+    """★ thread 的键必须含 owner。
+
+    没有它时 thread 键只有 (branch, file, pkey) —— 用户 A 的 feat/q4 和用户 B 的
+    feat/q4 会并成同一条链, A 能在注入块里读到 B 的 diff。这是数据泄漏。
+    """
+
+    _FACT = [{"statement": "On branch feat/q4, cfg.py 改了什么",
+              "branch": "feat/q4", "files": ["cfg.py"]}]
+
+    @staticmethod
+    def _edit(said, old, new, path="cfg.py"):
+        return [
+            _FakeMsg("user", said),
+            _FakeMsg("tool",
+                     f"path='{path}'\nold='''{old}'''\nnew='''{new}'''",
+                     tool_name="edit"),
+        ]
+
+    def test_owner_key_shape(self):
+        assert HyMemoryClient._chain_owner("u1", "repo1") == "u1|repo1"
+        # 两个都空 → ""，保持旧的单用户单仓数据兼容
+        assert HyMemoryClient._chain_owner(None, None) == ""
+        assert HyMemoryClient._chain_owner("", "") == ""
+
+    def test_two_users_same_branch_do_not_leak(self):
+        with tempfile.TemporaryDirectory() as d:
+            p = os.path.join(d, "ch")
+            HA = HyCodeMem(p, owner="userA|repo1")
+            HA.observe(HyMemoryClient._chain_text(
+                self._edit("On branch feat/q4, A 改的", "x = 1\n", "x = 2\n")),
+                meta={"branch": "feat/q4"})
+            # 同一个库文件、同一个分支、同一个文件名
+            HB = HyCodeMem(p, owner="userB|repo1")
+            HB.observe(HyMemoryClient._chain_text(
+                self._edit("On branch feat/q4, B 改的", "y = 1\n", "y = 2\n")),
+                meta={"branch": "feat/q4"})
+
+            blkA, _ = HA.decorate([dict(self._FACT[0])], instruction="q")
+            blkB, _ = HB.decorate([dict(self._FACT[0])], instruction="q")
+            assert "x = 2" in blkA and "y = 2" not in blkA, "A 读到了 B 的 diff"
+            assert "y = 2" in blkB and "x = 2" not in blkB, "B 读到了 A 的 diff"
+
+    def test_empty_owner_is_not_a_public_pool(self):
+        # 空 owner 只匹配空 owner，否则旧的空 owner 数据会错误读取所有新写入。
+        with tempfile.TemporaryDirectory() as d:
+            p = os.path.join(d, "ch")
+            H1 = HyCodeMem(p, owner="u1|r1")
+            H1.observe(HyMemoryClient._chain_text(
+                self._edit("On branch feat/q4, u1 改的", "x = 1\n", "x = 2\n")),
+                meta={"branch": "feat/q4"})
+            H0 = HyCodeMem(p)      # owner=''
+            blk, used = H0.decorate([dict(self._FACT[0])], instruction="q")
+            assert blk == "" and used == []
+
+    def test_stats_scoped_to_owner_by_default(self):
+        with tempfile.TemporaryDirectory() as d:
+            p = os.path.join(d, "ch")
+            for o, var in (("u1|r1", "a"), ("u2|r1", "b")):
+                H = HyCodeMem(p, owner=o)
+                H.observe(HyMemoryClient._chain_text(
+                    self._edit(f"On branch feat/q4, {o}", f"{var} = 1\n", f"{var} = 2\n")),
+                    meta={"branch": "feat/q4"})
+            H = HyCodeMem(p, owner="u1|r1")
+            # 默认只数自己的 —— 灌库自检靠这个数, 报全库会整体偏大
+            assert H.stats()["threads"] == 1
+            assert H.stats(all_owners=True)["threads"] == 2
+
+
+# ════════════════════════════════════════════════════════════════════
+# 8. 路径③ 的失败闸门
+# ════════════════════════════════════════════════════════════════════
+
+
+class TestEditFailureGate:
+    """从对话合成 diff 时, 失败的编辑不该进链。
+
+    链是"哪一版赢了"的唯一依据 —— 掺一个没生效的版本进去, agent 可能照着它写。
+    """
+
+    _EDIT = ('path = "cfg.py"\n'
+             'old = """mult = 255.0"""\n'
+             'new = """mult = 360.0"""')
+
+    def _observe(self, tmpdir, tail, tag):
+        H = HyCodeMem(os.path.join(tmpdir, tag))
+        t = HyMemoryClient._chain_text([
+            _FakeMsg("user", "On branch feat/q4, 改 mult"),
+            _FakeMsg("tool", self._EDIT + "\n" + tail, tool_name="bash"),
+        ])
+        return H.observe(t, meta={"branch": "feat/q4"})
+
+    @pytest.mark.parametrize("tail,tag", [
+        ("Traceback (most recent call last):\n  File \"<stdin>\"\nAssertionError", "as"),
+        ("old not in s", "ni"),
+        ("OBSERVATION: <returncode>2</returncode>", "rc2"),
+        ("SyntaxError: unterminated string literal", "se"),
+    ])
+    def test_failed_edits_blocked(self, tail, tag):
+        with tempfile.TemporaryDirectory() as d:
+            assert self._observe(d, tail, tag) == 0
+
+    @pytest.mark.parametrize("tail,tag", [
+        ("OBSERVATION: <returncode>0</returncode>", "rc0"),
+        ("REPLACED=1", "rp"),
+        ("", "none"),          # 没有任何结果标记 → 不拦(宁可收, 别漏真改动)
+    ])
+    def test_successful_and_unmarked_edits_kept(self, tail, tag):
+        with tempfile.TemporaryDirectory() as d:
+            assert self._observe(d, tail, tag) == 1
+
+    def test_retry_after_error_is_kept(self):
+        """★ 失败词之后又有成功迹象 → 重试成功了, 不能拦。
+
+        实测九仓 910 个路径③轮里有 9 轮命中失败词, 其中 **6 轮**后面紧跟
+        returncode>0 / REPLACED=1。只看失败词会把这 6 轮真改动误拦。
+        """
+        with tempfile.TemporaryDirectory() as d:
+            assert self._observe(d, "AssertionError\n再试一次\nREPLACED=1", "rt") == 1
+
+    def test_failure_patterns_are_case_sensitive(self):
+        """★ 判据不能用 re.I —— 会撞源码标识符。
+
+        我第一版用 re.I 扫 `FAILED|...`, 在九仓上报出 50 轮失败, 其实 50 个全是
+        ansible 源码里的函数名 `v2_runner_on_failed` 被匹配到, 而那些轮
+        <returncode>0</returncode> 明明是成功的。
+        """
+        from hy_memory.coding.chain_store import _EDIT_FAIL
+        assert not _EDIT_FAIL.search("def v2_runner_on_failed(self, result):")
+        assert not _EDIT_FAIL.search("    self._handle_failed_task()")
+        # 真的异常类名要抓到
+        assert _EDIT_FAIL.search("\nAssertionError\n")