"""Тесты канонического AgentHippocampus. Запуск: python3 test_agent_hippocampus.py """ from __future__ import annotations import json import shutil import tempfile import traceback from dataclasses import replace from pathlib import Path from agent_hippocampus import ( AgentHippocampus, NoveltyLevel, RecordKind, SignedRecord, load_embedder, ) PASSED: list[str] = [] FAILED: list[tuple[str, str]] = [] def case(name): def deco(fn): def wrapper(): tmp = Path(tempfile.mkdtemp(prefix="hip_test_")) try: fn(tmp) PASSED.append(name) print(f" ✓ {name}") except Exception as exc: FAILED.append((name, traceback.format_exc())) print(f" ✗ {name}: {exc}") finally: shutil.rmtree(tmp, ignore_errors=True) wrapper.__name__ = fn.__name__ return wrapper return deco def fresh(tmp: Path) -> AgentHippocampus: return AgentHippocampus( agent_id="test_agent", signing_key=AgentHippocampus.generate_signing_key(), data_dir=tmp, ) @case("init: rejects key not 32 bytes") def t_init_key_size(tmp: Path): try: AgentHippocampus(agent_id="a", signing_key=b"short", data_dir=tmp) except ValueError: return raise AssertionError("expected ValueError for short key") @case("init: creates data_dir") def t_init_dir(tmp: Path): sub = tmp / "deep" / "nested" AgentHippocampus(agent_id="a", signing_key=AgentHippocampus.generate_signing_key(), data_dir=sub) assert sub.exists() @case("record: returns SignedRecord with non-empty fields") def t_record_basic(tmp: Path): hip = fresh(tmp) rec = hip.record("first decision", kind=RecordKind.DECISION) assert isinstance(rec, SignedRecord) assert rec.agent_id == "test_agent" assert rec.kind == "agent.decision" assert rec.signature assert rec.record_id assert rec.prev_id is None @case("record: rejects empty content") def t_record_empty(tmp: Path): hip = fresh(tmp) try: hip.record(" ") except ValueError: return raise AssertionError("expected ValueError for empty content") @case("record: writes append-only line to stream.jsonl") def t_record_persists(tmp: Path): hip = fresh(tmp) hip.record("alpha") hip.record("beta") lines = hip.stream_file.read_text().splitlines() assert len(lines) == 2 for line in lines: json.loads(line) @case("chain: prev_id forms a chain") def t_chain_links(tmp: Path): hip = fresh(tmp) a = hip.record("alpha") b = hip.record("beta") c = hip.record("gamma") assert a.prev_id is None assert b.prev_id == a.record_id assert c.prev_id == b.record_id @case("verify: original record verifies True") def t_verify_ok(tmp: Path): hip = fresh(tmp) rec = hip.record("identity drift in SOUL.md") assert hip.verify(rec) is True @case("verify: tampered content fails") def t_verify_tampered_content(tmp: Path): hip = fresh(tmp) rec = hip.record("genuine") forged = replace(rec, content="forged") assert hip.verify(forged) is False @case("verify: tampered metadata fails") def t_verify_tampered_meta(tmp: Path): hip = fresh(tmp) rec = hip.record("with meta", metadata={"k": "v"}) forged = replace(rec, metadata={"k": "v2"}) assert hip.verify(forged) is False @case("verify: different signing key fails") def t_verify_wrong_key(tmp: Path): hip1 = AgentHippocampus(agent_id="a", signing_key=AgentHippocampus.generate_signing_key(), data_dir=tmp / "1") rec = hip1.record("hello") hip2 = AgentHippocampus(agent_id="a", signing_key=AgentHippocampus.generate_signing_key(), data_dir=tmp / "2") assert hip2.verify(rec) is False @case("verify_chain: clean chain passes") def t_verify_chain_ok(tmp: Path): hip = fresh(tmp) for i in range(5): hip.record(f"step {i}") ok, err = hip.verify_chain() assert ok, err @case("verify_chain: tampered file detected") def t_verify_chain_tampered(tmp: Path): hip = fresh(tmp) hip.record("real one") hip.record("real two") text = hip.stream_file.read_text() hip.stream_file.write_text(text.replace("real one", "fake one")) ok, err = hip.verify_chain() assert not ok assert err is not None @case("persistence: tail_id recovered after reopen") def t_persistence(tmp: Path): key = AgentHippocampus.generate_signing_key() hip1 = AgentHippocampus(agent_id="a", signing_key=key, data_dir=tmp) a = hip1.record("first") b = hip1.record("second") hip2 = AgentHippocampus(agent_id="a", signing_key=key, data_dir=tmp) c = hip2.record("third") assert c.prev_id == b.record_id ok, err = hip2.verify_chain() assert ok, err @case("novelty (word-freq): repeating routine flagged ROUTINE eventually") def t_novelty_word_freq(tmp: Path): hip = fresh(tmp) hip.record("alpha beta gamma delta epsilon zeta") hip.record("totally different vocabulary set here always") rec = hip.record("alpha beta gamma delta epsilon zeta") assert rec.novelty == NoveltyLevel.ROUTINE.value, f"got {rec.novelty}" @case("novelty (word-freq): brand new content flagged PREDICTION_ERROR") def t_novelty_prediction_error(tmp: Path): hip = fresh(tmp) rec = hip.record("ξυλοφαγος αρμενιος μνημοσυνη απομνημονευσις ουτοπια") assert rec.novelty == NoveltyLevel.PREDICTION_ERROR.value, f"got {rec.novelty}" @case("selective_load: respects token budget and skips ROUTINE") def t_selective_load(tmp: Path): hip = fresh(tmp) hip.record("alpha beta gamma delta epsilon zeta") hip.record("brand new vocabulary surprises pattern detector clearly") hip.record("alpha beta gamma delta epsilon zeta") big_budget = hip.selective_load(token_budget=1000) assert all(r.novelty != NoveltyLevel.ROUTINE.value for r in big_budget) tiny_budget = hip.selective_load(token_budget=2) total_chars = sum(len(r.content) for r in tiny_budget) assert total_chars <= 2 * 4 @case("pattern_completion (substring fallback): finds matching record") def t_pattern_substring(tmp: Path): hip = fresh(tmp) hip.record("My SOUL.md changed without owner approval") hip.record("Cron job optimized to 3 dollars per day") hits = hip.pattern_completion("SOUL.md", top_k=3) assert any("SOUL.md" in r.content for r in hits) @case("daily_anchor: empty day returns count 0") def t_anchor_empty(tmp: Path): hip = fresh(tmp) payload = hip.daily_anchor(date="2026-01-01") assert payload["count"] == 0 assert payload["dna_hash"] is None @case("daily_anchor: deterministic dna_hash for same set") def t_anchor_deterministic(tmp: Path): hip = fresh(tmp) hip.record("a") hip.record("b") today = hip.iter_records().__iter__().__next__().timestamp[:10] p1 = hip.daily_anchor(date=today) p2 = hip.daily_anchor(date=today) assert p1["dna_hash"] == p2["dna_hash"] assert p1["anchor_payload_hash"] == p2["anchor_payload_hash"] assert p1["count"] == 2 @case("daily_anchor: different content => different dna_hash") def t_anchor_changes(tmp: Path): hip1 = AgentHippocampus(agent_id="a1", signing_key=AgentHippocampus.generate_signing_key(), data_dir=tmp / "1") hip2 = AgentHippocampus(agent_id="a2", signing_key=AgentHippocampus.generate_signing_key(), data_dir=tmp / "2") hip1.record("alpha") hip2.record("beta") today = next(hip1.iter_records()).timestamp[:10] p1 = hip1.daily_anchor(date=today) p2 = hip2.daily_anchor(date=today) assert p1["dna_hash"] != p2["dna_hash"] @case("stats: reports count and novelty distribution") def t_stats(tmp: Path): hip = fresh(tmp) hip.record("alpha") hip.record("alpha") s = hip.stats() assert s["count"] == 2 assert sum(s["novelty_distribution"].values()) == 2 @case("domain separation: signature differs across kinds") def t_domain_separation(tmp: Path): hip = fresh(tmp) a = hip.record("same content", kind=RecordKind.STATE) b = hip.record("same content", kind=RecordKind.DECISION) assert a.signature != b.signature def main(): print("AgentHippocampus tests") print("=" * 60) tests = [v for k, v in globals().items() if k.startswith("t_") and callable(v)] for fn in tests: fn() print("=" * 60) print(f"PASSED: {len(PASSED)} FAILED: {len(FAILED)}") if FAILED: for name, tb in FAILED: print(f"\n--- {name} ---\n{tb}") raise SystemExit(1) if __name__ == "__main__": main()