cortex/tests/test_reconcile.py

175 lines · python

"""Darwin Cortex - reconciliation regression test.
 
THE CRITICAL TEST: proves a 40-day-stale agent-memory hit asserting "Hermes active"
LOSES to a live-infra hit asserting "Hermes stopped".
 
This is the exact failure mode that created Darwin. If this test breaks, the valve is broken.
"""
import sys
import os
 
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
import pytest
from datetime import datetime, timezone, timedelta
 
from models import Hit
from ranking import compute_score, rank_hits, detect_conflicts, recency_decay, authority_weight
 
 
 
class TestRecencyDecay:
    def test_fresh_hit_high_decay(self):
        """A hit from today decays almost to 1.0."""
        now = datetime.now(timezone.utc)
        d = recency_decay(now)
        assert d > 0.95, f"Expected decay > 0.95 for fresh hit, got {d}"
 
    def test_stale_hit_low_decay(self):
        """A 40-day-old hit decays significantly below 0.5 (HALF_LIFE=30d)."""
        stale = datetime.now(timezone.utc) - timedelta(days=40)
        d = recency_decay(stale)
        assert d < 0.3, f"Expected decay < 0.3 for 40d-old hit, got {d}"
 
    def test_unknown_last_seen_midpoint(self):
        """Unknown last_seen returns 0.5 (conservative)."""
        d = recency_decay(None)
        assert d == 0.5
 
 
class TestAuthorityWeight:
    def test_live_infra_is_highest(self):
        assert authority_weight("live-infra") == 1.0
 
    def test_agent_memory_below_recent_file(self):
        assert authority_weight("agent-memory") < authority_weight("recent-file")
 
    def test_older_note_is_lowest(self):
        assert authority_weight("older-note") == 0.4
 
 
 
class TestHermesLieRegression:
    """
    Proves: a 40-day-stale agent-memory hit saying "Hermes is active" MUST score
    lower than a live-infra hit saying "Hermes stopped".
 
    This is the exact Darwin failure mode: the agent read stale memory and reported
    Hermes as running when it was actually stopped.
    """
 
    def _make_stale_memory_hit(self) -> Hit:
        """40-day-old agent-memory claim: Hermes active."""
        return Hit(
            engine="memgraph",
            title="Hermes trading bot",
            path="memory://hermes-status",
            snippet="Hermes is active and running auto-execute trading",
            score=0.9,
            last_seen=datetime.now(timezone.utc) - timedelta(days=40),
            confidence=0.85,
            authority_tier="agent-memory",
        )
 
    def _make_live_infra_hit(self) -> Hit:
        """Fresh live-infra hit: Hermes stopped."""
        return Hit(
            engine="rag",
            title="Hermes systemd status",
            path="/var/log/hermes/status",
            snippet="Hermes stopped - service inactive (dead)",
            score=0.75,
            last_seen=datetime.now(timezone.utc) - timedelta(hours=1),
            confidence=0.92,
            authority_tier="live-infra",
        )
 
    def test_live_infra_beats_stale_memory_score(self):
        """Live-infra score must exceed stale agent-memory score."""
        stale = self._make_stale_memory_hit()
        live = self._make_live_infra_hit()
 
        stale_score = compute_score(stale, relevance=stale.score)
        live_score = compute_score(live, relevance=live.score)
 
        print(f"\nStale agent-memory score: {stale_score:.4f}")
        print(f"Live-infra score:         {live_score:.4f}")
        print(f"Decay (stale, 40d):       {recency_decay(stale.last_seen):.4f}")
        print(f"Decay (live, 1h):         {recency_decay(live.last_seen):.4f}")
 
        assert live_score > stale_score, (
            f"REGRESSION: stale agent-memory ({stale_score:.4f}) beat "
            f"live-infra ({live_score:.4f}). The Darwin valve is broken."
        )
 
    def test_rank_hits_puts_live_infra_first(self):
        """After rank_hits(), the live-infra hit must be ranked #1."""
        stale = self._make_stale_memory_hit()
        live = self._make_live_infra_hit()
 
        ranked = rank_hits([stale, live], k=10)
 
        assert ranked[0].engine == "rag", (
            f"REGRESSION: expected 'rag' (live-infra) ranked first, "
            f"got '{ranked[0].engine}'. Stale agent-memory is beating live truth."
        )
        assert ranked[1].engine == "memgraph", (
            f"Expected 'memgraph' (stale) ranked second, got '{ranked[1].engine}'"
        )
 
    def test_conflict_detected_and_stale_is_loser(self):
        """Conflict detection must find the disagreement and record stale as loser."""
        stale = self._make_stale_memory_hit()
        live = self._make_live_infra_hit()
 
        ranked = rank_hits([stale, live], k=10)
        conflicts = detect_conflicts(ranked)
 
        assert len(conflicts) >= 1, (
            "REGRESSION: no conflict detected between 'Hermes active' (stale memory) "
            "and 'Hermes stopped' (live infra). Conflict detection is broken."
        )
 
        c = conflicts[0]
        print(f"\nConflict detected: {c}")
        assert "memgraph" in c.losers[0], (
            f"Expected memgraph (stale) to be the loser, got: {c.losers}"
        )
        assert "rag" in c.winner, (
            f"Expected rag (live-infra) to be the winner, got: {c.winner}"
        )
 
    def test_score_margin_is_significant(self):
        """The score margin must be large enough to matter (> 0.05)."""
        stale = self._make_stale_memory_hit()
        live = self._make_live_infra_hit()
 
        stale_score = compute_score(stale, relevance=stale.score)
        live_score = compute_score(live, relevance=live.score)
        margin = live_score - stale_score
 
        assert margin > 0.05, (
            f"Score margin too small ({margin:.4f}). "
            f"The decay valve is not working strongly enough."
        )
 
 
 
class TestGracefulDegradation:
    def test_rank_hits_empty_list(self):
        """rank_hits with no hits returns empty list (not an error)."""
        result = rank_hits([], k=10)
        assert result == []
 
    def test_detect_conflicts_empty(self):
        """detect_conflicts with no hits returns empty list."""
        result = detect_conflicts([])
        assert result == []
 
    def test_single_hit_no_conflict(self):
        """Single hit never creates a conflict."""
        hit = Hit(engine="rag", title="test", snippet="something", score=0.5,
                  authority_tier="recent-file")
        rank_hits([hit], k=1)
        conflicts = detect_conflicts([hit])
        assert conflicts == []

1	"""Darwin Cortex - reconciliation regression test.
2
3	THE CRITICAL TEST: proves a 40-day-stale agent-memory hit asserting "Hermes active"
4	LOSES to a live-infra hit asserting "Hermes stopped".
5
6	This is the exact failure mode that created Darwin. If this test breaks, the valve is broken.
7	"""
8	import sys
9	import os
10
11	sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
12
13	import pytest
14	from datetime import datetime, timezone, timedelta
15
16	from models import Hit
17	from ranking import compute_score, rank_hits, detect_conflicts, recency_decay, authority_weight
18
19
20
21	class TestRecencyDecay:
22	def test_fresh_hit_high_decay(self):
23	"""A hit from today decays almost to 1.0."""
24	now = datetime.now(timezone.utc)
25	d = recency_decay(now)
26	assert d > 0.95, f"Expected decay > 0.95 for fresh hit, got {d}"
27
28	def test_stale_hit_low_decay(self):
29	"""A 40-day-old hit decays significantly below 0.5 (HALF_LIFE=30d)."""
30	stale = datetime.now(timezone.utc) - timedelta(days=40)
31	d = recency_decay(stale)
32	assert d < 0.3, f"Expected decay < 0.3 for 40d-old hit, got {d}"
33
34	def test_unknown_last_seen_midpoint(self):
35	"""Unknown last_seen returns 0.5 (conservative)."""
36	d = recency_decay(None)
37	assert d == 0.5
38
39
40	class TestAuthorityWeight:
41	def test_live_infra_is_highest(self):
42	assert authority_weight("live-infra") == 1.0
43
44	def test_agent_memory_below_recent_file(self):
45	assert authority_weight("agent-memory") < authority_weight("recent-file")
46
47	def test_older_note_is_lowest(self):
48	assert authority_weight("older-note") == 0.4
49
50
51
52	class TestHermesLieRegression:
53	"""
54	Proves: a 40-day-stale agent-memory hit saying "Hermes is active" MUST score
55	lower than a live-infra hit saying "Hermes stopped".
56
57	This is the exact Darwin failure mode: the agent read stale memory and reported
58	Hermes as running when it was actually stopped.
59	"""
60
61	def _make_stale_memory_hit(self) -> Hit:
62	"""40-day-old agent-memory claim: Hermes active."""
63	return Hit(
64	engine="memgraph",
65	title="Hermes trading bot",
66	path="memory://hermes-status",
67	snippet="Hermes is active and running auto-execute trading",
68	score=0.9,
69	last_seen=datetime.now(timezone.utc) - timedelta(days=40),
70	confidence=0.85,
71	authority_tier="agent-memory",
72	)
73
74	def _make_live_infra_hit(self) -> Hit:
75	"""Fresh live-infra hit: Hermes stopped."""
76	return Hit(
77	engine="rag",
78	title="Hermes systemd status",
79	path="/var/log/hermes/status",
80	snippet="Hermes stopped - service inactive (dead)",
81	score=0.75,
82	last_seen=datetime.now(timezone.utc) - timedelta(hours=1),
83	confidence=0.92,
84	authority_tier="live-infra",
85	)
86
87	def test_live_infra_beats_stale_memory_score(self):
88	"""Live-infra score must exceed stale agent-memory score."""
89	stale = self._make_stale_memory_hit()
90	live = self._make_live_infra_hit()
91
92	stale_score = compute_score(stale, relevance=stale.score)
93	live_score = compute_score(live, relevance=live.score)
94
95	print(f"\nStale agent-memory score: {stale_score:.4f}")
96	print(f"Live-infra score: {live_score:.4f}")
97	print(f"Decay (stale, 40d): {recency_decay(stale.last_seen):.4f}")
98	print(f"Decay (live, 1h): {recency_decay(live.last_seen):.4f}")
99
100	assert live_score > stale_score, (
101	f"REGRESSION: stale agent-memory ({stale_score:.4f}) beat "
102	f"live-infra ({live_score:.4f}). The Darwin valve is broken."
103	)
104
105	def test_rank_hits_puts_live_infra_first(self):
106	"""After rank_hits(), the live-infra hit must be ranked #1."""
107	stale = self._make_stale_memory_hit()
108	live = self._make_live_infra_hit()
109
110	ranked = rank_hits([stale, live], k=10)
111
112	assert ranked[0].engine == "rag", (
113	f"REGRESSION: expected 'rag' (live-infra) ranked first, "
114	f"got '{ranked[0].engine}'. Stale agent-memory is beating live truth."
115	)
116	assert ranked[1].engine == "memgraph", (
117	f"Expected 'memgraph' (stale) ranked second, got '{ranked[1].engine}'"
118	)
119
120	def test_conflict_detected_and_stale_is_loser(self):
121	"""Conflict detection must find the disagreement and record stale as loser."""
122	stale = self._make_stale_memory_hit()
123	live = self._make_live_infra_hit()
124
125	ranked = rank_hits([stale, live], k=10)
126	conflicts = detect_conflicts(ranked)
127
128	assert len(conflicts) >= 1, (
129	"REGRESSION: no conflict detected between 'Hermes active' (stale memory) "
130	"and 'Hermes stopped' (live infra). Conflict detection is broken."
131	)
132
133	c = conflicts[0]
134	print(f"\nConflict detected: {c}")
135	assert "memgraph" in c.losers[0], (
136	f"Expected memgraph (stale) to be the loser, got: {c.losers}"
137	)
138	assert "rag" in c.winner, (
139	f"Expected rag (live-infra) to be the winner, got: {c.winner}"
140	)
141
142	def test_score_margin_is_significant(self):
143	"""The score margin must be large enough to matter (> 0.05)."""
144	stale = self._make_stale_memory_hit()
145	live = self._make_live_infra_hit()
146
147	stale_score = compute_score(stale, relevance=stale.score)
148	live_score = compute_score(live, relevance=live.score)
149	margin = live_score - stale_score
150
151	assert margin > 0.05, (
152	f"Score margin too small ({margin:.4f}). "
153	f"The decay valve is not working strongly enough."
154	)
155
156
157
158	class TestGracefulDegradation:
159	def test_rank_hits_empty_list(self):
160	"""rank_hits with no hits returns empty list (not an error)."""
161	result = rank_hits([], k=10)
162	assert result == []
163
164	def test_detect_conflicts_empty(self):
165	"""detect_conflicts with no hits returns empty list."""
166	result = detect_conflicts([])
167	assert result == []
168
169	def test_single_hit_no_conflict(self):
170	"""Single hit never creates a conflict."""
171	hit = Hit(engine="rag", title="test", snippet="something", score=0.5,
172	authority_tier="recent-file")
173	rank_hits([hit], k=1)
174	conflicts = detect_conflicts([hit])
175	assert conflicts == []