generated from coulomb/repo-seed
tegwick
abb888f3ef
- session_memory/core/digest.py: build_digest (cost totals, kind/tool histograms, markers, snippets) + cross-flavor infer_outcome heuristic; analyze() promotes Tier1->Tier2 and sets analyzed_at (-> evictable) - tests/test_digest.py Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
"""Digest tests (T04): outcome heuristic + Tier 2 promotion."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from session_memory.adapters.claude import Normalized # noqa: E402
|
|
from session_memory.core.digest import analyze, build_digest, infer_outcome # noqa: E402
|
|
from session_memory.core.schema import Cost, Session, SessionEvent # noqa: E402
|
|
from session_memory.core.store import Store # noqa: E402
|
|
|
|
|
|
def _ev(uid, seq, kind, **kw):
|
|
return SessionEvent(session_uid=uid, seq=seq, kind=kind, **kw)
|
|
|
|
|
|
def test_infer_outcome_abandoned():
|
|
uid = "claude:s"
|
|
assert infer_outcome([_ev(uid, 0, "user_msg")]) == "abandoned"
|
|
|
|
|
|
def test_infer_outcome_success_on_passing_test():
|
|
uid = "claude:s"
|
|
events = [
|
|
_ev(uid, 0, "user_msg"),
|
|
_ev(uid, 1, "assistant_msg"),
|
|
_ev(uid, 2, "test_run", tool="Bash"),
|
|
_ev(uid, 3, "tool_result", payload_ref="b3"),
|
|
]
|
|
assert infer_outcome(events, {"b3": "6 passed in 0.4s"}) == "success"
|
|
|
|
|
|
def test_infer_outcome_fail_on_failing_test():
|
|
uid = "claude:s"
|
|
events = [
|
|
_ev(uid, 0, "user_msg"),
|
|
_ev(uid, 1, "assistant_msg"),
|
|
_ev(uid, 2, "test_run", tool="Bash"),
|
|
_ev(uid, 3, "tool_result", payload_ref="b3"),
|
|
]
|
|
assert infer_outcome(events, {"b3": "1 failed, traceback ..."}) == "fail"
|
|
|
|
|
|
def test_build_digest_histograms_and_markers():
|
|
uid = "claude:s"
|
|
s = Session(session_uid=uid, flavor="claude", native_session_id="s",
|
|
repo="agentic-resources", cost=Cost(input_tokens=100, output_tokens=40, turns=2))
|
|
events = [
|
|
_ev(uid, 0, "user_msg"),
|
|
_ev(uid, 1, "edit", tool="Edit"),
|
|
_ev(uid, 2, "edit", tool="Write"),
|
|
_ev(uid, 3, "test_run", tool="Bash"),
|
|
_ev(uid, 4, "error"),
|
|
_ev(uid, 5, "assistant_msg"),
|
|
]
|
|
d = build_digest(s, events)
|
|
assert d["tool_histogram"] == {"Edit": 1, "Write": 1, "Bash": 1}
|
|
assert d["markers"]["edits"] == 2
|
|
assert d["markers"]["errors"] == 1
|
|
assert d["markers"]["test_runs"] == 1
|
|
assert d["event_count"] == 6
|
|
assert d["cost"]["input_tokens"] == 100
|
|
|
|
|
|
def test_analyze_writes_digest_and_sets_analyzed(tmp_path):
|
|
st = Store(str(tmp_path / "m.db"), str(tmp_path / "blobs"))
|
|
uid = Session.make_uid("claude", "s1")
|
|
s = Session(session_uid=uid, flavor="claude", native_session_id="s1")
|
|
events = [
|
|
SessionEvent(session_uid=uid, seq=0, kind="user_msg", payload_ref="b0"),
|
|
SessionEvent(session_uid=uid, seq=1, kind="assistant_msg", payload_ref="b1"),
|
|
]
|
|
blobs = {"b0": "please help", "b1": "done"}
|
|
st.ingest(Normalized(session=s, events=events, blobs=blobs))
|
|
|
|
assert st.get_session(uid).is_evictable is False
|
|
d = analyze(st, uid)
|
|
assert d["outcome"] == "success"
|
|
assert d["first_prompt"] == "please help"
|
|
assert st.get_session(uid).analyzed_at is not None
|
|
assert st.get_session(uid).is_evictable is True # now promoted -> evictable
|