mirror of
https://github.com/Graphify-Labs/graphify.git
synced 2026-09-14 19:34:09 +08:00
daichiyasunami-vottia
2f743aea89
save_query_result named files query_<second>_<slug50>.md and wrote them with
write_text, so two saves in the same second whose questions share the first
50 characters resolved to one path and the later one silently replaced the
earlier. Both calls returned normally. This is the common case when several
agents sweep one subsystem in parallel.
Add a short uuid to the name. The query_ prefix and .md suffix are unchanged
so reflect.load_memory_docs and existing tests keep working.
Fixes #3301
(cherry picked from commit 0bdf89f64b)
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""Tests for graphify.ingest.save_query_result"""
|
|
from __future__ import annotations
|
|
import re
|
|
from pathlib import Path
|
|
import pytest
|
|
from graphify.ingest import save_query_result
|
|
|
|
|
|
def test_file_created(tmp_path):
|
|
out = save_query_result("what is attention?", "Attention is...", tmp_path / "memory")
|
|
assert out.exists()
|
|
|
|
|
|
def test_filename_format(tmp_path):
|
|
mem = tmp_path / "memory"
|
|
out = save_query_result("what connects A to B?", "They share...", mem)
|
|
assert out.name.startswith("query_")
|
|
assert out.suffix == ".md"
|
|
|
|
|
|
def test_frontmatter_question(tmp_path):
|
|
mem = tmp_path / "memory"
|
|
question = "what is attention?"
|
|
out = save_query_result(question, "Attention is softmax.", mem)
|
|
content = out.read_text()
|
|
assert "question:" in content
|
|
assert "attention" in content.lower()
|
|
|
|
|
|
def test_frontmatter_type(tmp_path):
|
|
mem = tmp_path / "memory"
|
|
out = save_query_result("q", "a", mem, query_type="path_query")
|
|
content = out.read_text()
|
|
assert 'type: "path_query"' in content
|
|
|
|
|
|
def test_source_nodes_included(tmp_path):
|
|
mem = tmp_path / "memory"
|
|
nodes = ["AttentionLayer", "SoftmaxFunc"]
|
|
out = save_query_result("q", "a", mem, source_nodes=nodes)
|
|
content = out.read_text()
|
|
assert "AttentionLayer" in content
|
|
assert "SoftmaxFunc" in content
|
|
|
|
|
|
def test_source_nodes_capped_at_10(tmp_path):
|
|
mem = tmp_path / "memory"
|
|
nodes = [f"Node{i}" for i in range(20)]
|
|
out = save_query_result("q", "a", mem, source_nodes=nodes)
|
|
content = out.read_text()
|
|
# Only first 10 should appear in frontmatter source_nodes line
|
|
fm_line = [l for l in content.splitlines() if l.startswith("source_nodes:")][0]
|
|
assert fm_line.count('"Node') == 10
|
|
|
|
|
|
def test_memory_dir_created(tmp_path):
|
|
mem = tmp_path / "deep" / "memory"
|
|
assert not mem.exists()
|
|
save_query_result("q", "a", mem)
|
|
assert mem.exists()
|
|
|
|
|
|
def test_answer_in_body(tmp_path):
|
|
mem = tmp_path / "memory"
|
|
answer = "The answer is forty-two."
|
|
out = save_query_result("what is the answer?", answer, mem)
|
|
content = out.read_text()
|
|
assert answer in content
|
|
|
|
def test_outcome_in_frontmatter_and_body(tmp_path):
|
|
"""An outcome signal is written to both frontmatter (for `reflect`) and an
|
|
## Outcome body section (so it round-trips into the graph on re-extraction)."""
|
|
out = save_query_result("q", "a", tmp_path / "memory", outcome="useful")
|
|
content = out.read_text()
|
|
assert 'outcome: "useful"' in content
|
|
assert "## Outcome" in content
|
|
assert "- Signal: useful" in content
|
|
|
|
|
|
def test_correction_in_frontmatter_and_body(tmp_path):
|
|
out = save_query_result(
|
|
"what hashes passwords?", "MD5", tmp_path / "memory",
|
|
outcome="corrected", correction="It's bcrypt, see PasswordHasher",
|
|
)
|
|
content = out.read_text()
|
|
assert 'correction: "It\'s bcrypt, see PasswordHasher"' in content
|
|
assert "- Correction: It's bcrypt, see PasswordHasher" in content
|
|
|
|
|
|
def test_no_outcome_means_no_outcome_section(tmp_path):
|
|
"""Backward compatible: a result without an outcome looks exactly as before."""
|
|
out = save_query_result("q", "a", tmp_path / "memory")
|
|
content = out.read_text()
|
|
assert "outcome:" not in content
|
|
assert "## Outcome" not in content
|
|
|
|
|
|
def test_invalid_outcome_rejected(tmp_path):
|
|
with pytest.raises(ValueError):
|
|
save_query_result("q", "a", tmp_path / "memory", outcome="great")
|
|
|
|
|
|
def test_concurrent_saves_of_the_same_question_do_not_overwrite(tmp_path):
|
|
"""Regression for #3301: a second-granularity stamp plus a 50-char slug is
|
|
not unique, so saves in the same second sharing a prefix collapsed into one
|
|
file and the earlier ones were silently lost."""
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
mem = tmp_path / "memory"
|
|
with ThreadPoolExecutor(max_workers=20) as ex:
|
|
paths = list(ex.map(lambda _: save_query_result("how does auth work", "a", mem), range(20)))
|
|
assert len({p.name for p in paths}) == 20
|
|
assert len(list(mem.glob("*.md"))) == 20
|