574 lines
26 KiB
Python
574 lines
26 KiB
Python
# -*- 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")
|