wehub-resource-sync
247 lines
8.2 KiB
Python
247 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from opensquilla.memory.dream import Dream
|
|
from opensquilla.memory.dream.evidence import load_evidence_store
|
|
|
|
|
|
class _Response:
|
|
def __init__(self, content: str) -> None:
|
|
self.content = content
|
|
|
|
|
|
class _PatchProvider:
|
|
def __init__(self, text: str | None = None, *, delete_path=None) -> None:
|
|
self.calls = 0
|
|
self.delete_path = delete_path
|
|
self.text = text or json.dumps(
|
|
{
|
|
"operations": [
|
|
{
|
|
"op": "upsert",
|
|
"candidate_ids": ["auto"],
|
|
"section": "User Preferences",
|
|
"memory_id": "mem_provider_benchmarks",
|
|
"text": "- User prefers provider-backed benchmarks over toy simulations.",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
async def complete(self, *, messages, max_tokens):
|
|
self.calls += 1
|
|
if self.delete_path is not None and self.delete_path.exists():
|
|
self.delete_path.unlink()
|
|
return _Response(self.text)
|
|
|
|
|
|
class _FailIfCalledProvider:
|
|
def __init__(self) -> None:
|
|
self.calls = 0
|
|
|
|
async def complete(self, *, messages, max_tokens): # noqa: ARG002
|
|
self.calls += 1
|
|
raise AssertionError("provider should not be called")
|
|
|
|
|
|
def _dream(workspace, *, provider=None, preview=False):
|
|
config = SimpleNamespace(
|
|
max_batch_size=10,
|
|
min_batch_size=1,
|
|
input_slimming="off",
|
|
preview_mode=preview,
|
|
dry_run=preview,
|
|
evidence_min_score=0.0,
|
|
evidence_negative_recurrence_threshold=2,
|
|
evidence_quarantine_enabled=True,
|
|
evidence_curated_writes_enabled=True,
|
|
)
|
|
return Dream(
|
|
workspace=workspace,
|
|
provider=provider or _PatchProvider(),
|
|
session_lock=None,
|
|
config=config,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_records_evidence_and_writes_curated_memory(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
(memory_dir / "note.md").write_text(
|
|
"User prefers provider-backed benchmarks over toy simulations.",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = await _dream(tmp_path).run()
|
|
|
|
assert result.evidence_status == "ok"
|
|
assert result.apply_status == "ok"
|
|
assert "User Preferences" in (tmp_path / "MEMORY.md").read_text(encoding="utf-8")
|
|
assert (memory_dir / ".dream_state" / "promotion_evidence.json").exists()
|
|
assert (memory_dir / "note.md").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_result_and_log_use_single_path_names(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
(memory_dir / "note.md").write_text(
|
|
"User prefers provider-backed benchmarks over toy simulations.",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = await _dream(tmp_path).run()
|
|
log_path = next((tmp_path / "logs").glob("dream-main-*.jsonl"))
|
|
log_row = json.loads(log_path.read_text(encoding="utf-8").splitlines()[-1])
|
|
|
|
assert result.evidence_status == "ok"
|
|
assert result.apply_status == "ok"
|
|
assert result.provider_calls == 1
|
|
assert not hasattr(result, "phase" + "1_status")
|
|
assert not hasattr(result, "phase" + "2_status")
|
|
assert not hasattr(result, "files_deleted")
|
|
assert "evidence_status" in log_row
|
|
assert "apply_status" in log_row
|
|
assert "provider_calls" in log_row
|
|
assert "files_deleted" not in log_row
|
|
assert all("phase" not in key for key in log_row)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_preview_mutates_nothing(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
note = memory_dir / "note.md"
|
|
note.write_text("User prefers concise implementation notes.", encoding="utf-8")
|
|
|
|
result = await _dream(tmp_path, preview=True).run()
|
|
|
|
assert result.evidence_status == "ok"
|
|
assert result.apply_status == "ok"
|
|
assert note.exists()
|
|
assert not (tmp_path / "MEMORY.md").exists()
|
|
assert not (memory_dir / ".dream_state" / "promotion_evidence.json").exists()
|
|
assert result.cursor_after == result.cursor_before
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_externally_missing_candidate_after_evidence_does_not_raise(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
note = memory_dir / "note.md"
|
|
note.write_text("User prefers concise implementation notes.", encoding="utf-8")
|
|
provider = _PatchProvider(delete_path=note)
|
|
|
|
result = await _dream(tmp_path, provider=provider).run()
|
|
|
|
assert result.apply_status == "ok"
|
|
assert "FileNotFoundError" not in (result.error or "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_writes_evidence_receipt(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
(memory_dir / "note.md").write_text(
|
|
"User prefers provider-backed benchmarks over toy simulations.",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = await _dream(tmp_path).run()
|
|
|
|
assert result.edit_receipt_path is not None
|
|
receipt = json.loads((tmp_path / result.edit_receipt_path).read_text(encoding="utf-8"))
|
|
assert receipt["schema_version"] == 1
|
|
assert "version" not in receipt
|
|
assert "ranked_candidates" in receipt
|
|
assert "applied_promotions" in receipt
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_skips_provider_when_no_candidates_rank(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
note = memory_dir / "note.md"
|
|
note.write_text("Do not use this one-off label.", encoding="utf-8")
|
|
provider = _FailIfCalledProvider()
|
|
|
|
result = await _dream(tmp_path, provider=provider).run()
|
|
|
|
assert provider.calls == 0
|
|
assert result.provider_calls == 0
|
|
assert result.evidence_status == "ok"
|
|
assert result.apply_status == "skipped"
|
|
assert note.exists()
|
|
assert not (tmp_path / "MEMORY.md").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_does_not_promote_stale_candidate_from_mixed_operation(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
stale = memory_dir / "stale.md"
|
|
live = memory_dir / "live.md"
|
|
stale.write_text("User prefers stale source.", encoding="utf-8")
|
|
live.write_text("User prefers live source.", encoding="utf-8")
|
|
provider = _PatchProvider(
|
|
text=json.dumps(
|
|
{
|
|
"operations": [
|
|
{
|
|
"op": "upsert",
|
|
"candidate_ids": ["auto"],
|
|
"section": "User Preferences",
|
|
"memory_id": "mem_mixed",
|
|
"text": "- User prefers live source.",
|
|
}
|
|
]
|
|
}
|
|
),
|
|
delete_path=stale,
|
|
)
|
|
|
|
result = await _dream(tmp_path, provider=provider).run()
|
|
store = load_evidence_store(tmp_path)
|
|
stale_entries = [
|
|
entry for entry in store.entries.values() if entry.source_path == "memory/stale.md"
|
|
]
|
|
live_entries = [
|
|
entry for entry in store.entries.values() if entry.source_path == "memory/live.md"
|
|
]
|
|
|
|
assert result.apply_status == "ok"
|
|
assert stale_entries and stale_entries[0].status == "candidate"
|
|
assert stale_entries[0].last_skip_reason == "source_missing"
|
|
assert live_entries and live_entries[0].status == "promoted"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dream_marks_noop_curated_write_as_represented_not_promoted(tmp_path):
|
|
memory_dir = tmp_path / "memory"
|
|
memory_dir.mkdir()
|
|
existing = "- User prefers provider-backed benchmarks over toy simulations."
|
|
(tmp_path / "MEMORY.md").write_text(
|
|
f"# Long-Term Memory\n\n## User Preferences\n\n{existing}\n",
|
|
encoding="utf-8",
|
|
)
|
|
(memory_dir / "note.md").write_text(
|
|
"User prefers provider-backed benchmarks over toy simulations.",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = await _dream(tmp_path).run()
|
|
store = load_evidence_store(tmp_path)
|
|
entry = next(iter(store.entries.values()))
|
|
receipt = json.loads((tmp_path / result.edit_receipt_path).read_text(encoding="utf-8"))
|
|
|
|
assert result.apply_status == "ok"
|
|
assert entry.status == "represented"
|
|
assert entry.promoted_at is None
|
|
assert entry.last_skip_reason == "no_curated_change"
|
|
assert receipt["applied_promotions"][0]["changed"] is False
|